(self, *,
add_policies: Optional[Iterable[admin_types.PedmPolicy]] = None,
update_policies: Optional[Iterable[admin_types.PedmUpdatePolicy]] = None,
remove_policies: Optional[Iterable[str]] = None)
| 539 | return admin_types.ModifyStatus.from_proto(status_rs) |
| 540 | |
| 541 | def modify_policies(self, *, |
| 542 | add_policies: Optional[Iterable[admin_types.PedmPolicy]] = None, |
| 543 | update_policies: Optional[Iterable[admin_types.PedmUpdatePolicy]] = None, |
| 544 | remove_policies: Optional[Iterable[str]] = None) -> admin_types.ModifyStatus: |
| 545 | rq = pedm_pb2.PolicyRequest() |
| 546 | if add_policies is not None: |
| 547 | for policy in add_policies: |
| 548 | pa = pedm_pb2.PolicyAdd() |
| 549 | policy_uid = policy.policy_uid or utils.generate_uid() |
| 550 | pa.policyUid = utils.base64_url_decode(policy_uid) |
| 551 | policy_key = policy.policy_key or utils.generate_aes_key() |
| 552 | pa.encryptedKey = crypto.encrypt_aes_v2(policy_key, self.agent_key) |
| 553 | |
| 554 | admin_data = json.dumps(policy.admin_data).encode('utf-8') |
| 555 | pa.plainData = admin_data |
| 556 | |
| 557 | policy_data = json.dumps(policy.data).encode('utf-8') |
| 558 | encrypted_data = crypto.encrypt_aes_v2(policy_data, policy_key) |
| 559 | pa.encryptedData = encrypted_data |
| 560 | |
| 561 | pa.disabled = policy.disabled |
| 562 | rq.addPolicy.append(pa) |
| 563 | if update_policies is not None: |
| 564 | for policy_update in update_policies: |
| 565 | policy_uid = policy_update.policy_uid |
| 566 | existing_policy = self.policies.get_entity(policy_uid) |
| 567 | if existing_policy is None: |
| 568 | raise Exception(f'Update: Policy {policy_uid} not found') |
| 569 | pu = pedm_pb2.PolicyUpdate() |
| 570 | pu.policyUid = utils.base64_url_decode(existing_policy.policy_uid) |
| 571 | if isinstance(policy_update.admin_data, dict): |
| 572 | pass |
| 573 | if isinstance(policy_update.data, dict): |
| 574 | json_policy = json.dumps(policy_update.data).encode('utf-8') |
| 575 | encrypted_data = crypto.encrypt_aes_v2(json_policy, existing_policy.policy_key) |
| 576 | pu.encryptedData = encrypted_data |
| 577 | if isinstance(policy_update.disabled, bool): |
| 578 | pu.disabled = folder_pb2.BOOLEAN_TRUE if policy_update.disabled else folder_pb2.BOOLEAN_FALSE |
| 579 | rq.updatePolicy.append(pu) |
| 580 | if remove_policies is not None: |
| 581 | rq.removePolicy.extend((utils.base64_url_decode(x) for x in remove_policies)) |
| 582 | |
| 583 | status_rs = api.execute_router(self.params, 'pedm/modify_policy', rq, rs_type=pedm_pb2.PedmStatusResponse) |
| 584 | self._need_sync = True |
| 585 | assert status_rs is not None |
| 586 | return admin_types.ModifyStatus.from_proto(status_rs) |
| 587 | |
| 588 | |
| 589 | @staticmethod |
no test coverage detected