diff --git a/keepercommander/commands/discoveryrotation.py b/keepercommander/commands/discoveryrotation.py index 031665ac2..17fb6fe4d 100644 --- a/keepercommander/commands/discoveryrotation.py +++ b/keepercommander/commands/discoveryrotation.py @@ -17,8 +17,8 @@ import re import time from datetime import datetime +from typing import Dict, Optional, Any, Set, List from urllib.parse import urlparse, urlunparse -from typing import Optional, List import requests from keeper_secrets_manager_core.utils import url_safe_str_to_bytes @@ -93,6 +93,7 @@ from .pam_saas.config import PAMActionSaasConfigCommand from .pam_saas.update import PAMActionSaasUpdateCommand from .tunnel_and_connections import PAMTunnelCommand, PAMConnectionCommand, PAMRbiCommand, PAMSplitCommand +from .pam.cnapp_commands import PAMCnappCommand from .universalsecretsync import ( PAMUniversalSyncConfigCommand, PAMUniversalSyncRunCommand @@ -287,6 +288,8 @@ def __init__(self): self.register_command('workflow', PAMWorkflowCommand(), 'Manage PAM Workflows', 'w') self.register_command('access', PAMPrivilegedAccessCommand(), 'Manage privileged cloud access operations', 'ac') + self.register_command('cnapp', PAMCnappCommand(), + 'Manage Cloud-Native Application Protection Platform integration', 'cn') self.register_command('universal-sync-config', PAMUniversalSyncConfigCommand(), 'Manage Universal Sync Configurations', 'usc') self.register_command('universal-sync-run', PAMUniversalSyncRunCommand(), 'Run Universal Sync', 'usr') diff --git a/keepercommander/commands/pam/cnapp_commands.py b/keepercommander/commands/pam/cnapp_commands.py new file mode 100644 index 000000000..bce1d5d16 --- /dev/null +++ b/keepercommander/commands/pam/cnapp_commands.py @@ -0,0 +1,650 @@ +# _ __ +# | |/ /___ ___ _ __ ___ _ _ ® +# | ' bytes|None + """Encrypter keys are typically `openssl rand -base64 32`. Try standard base64 first, + then base64url (legacy notes). 32 bytes only (AES-256) — anything else is rejected so we + don't pass garbage to AES-GCM.""" + if not raw or not isinstance(raw, str): + return None + candidate = raw.strip() + for decoder in (base64.b64decode, base64.urlsafe_b64decode): + try: + padding = '=' * (-len(candidate) % 4) + data = decoder(candidate + padding) + except (binascii.Error, ValueError): + continue + if len(data) == 32: + return data + return None + + +def _load_encrypter_key(params, config_record_uid): + """Resolve the AES key from the CNAPP encrypter vault record. Returns None when the + record can't be loaded or doesn't carry a recognizable key — callers should fall back + to showing the encrypted payload as-is.""" + if not config_record_uid: + return None + try: + record = vault.KeeperRecord.load(params, config_record_uid) + except Exception as e: + logger.debug('CNAPP: failed to load encrypter record %s: %s', config_record_uid, e) + return None + if not isinstance(record, vault.TypedRecord): + return None + # Match vault/cloudSecurityUtils.ts: prefer `secret` then `note` labeled "Encryption Key", + # then the first unlabeled `note` field only when no labeled key field exists. + labeled_raws = [] + secret_field = record.get_typed_field('secret', CNAPP_ENCRYPTION_KEY_LABEL) + if secret_field and secret_field.value: + labeled_raws.append(secret_field.value[0]) + note_labeled = record.get_typed_field('note', CNAPP_ENCRYPTION_KEY_LABEL) + if note_labeled and note_labeled.value: + labeled_raws.append(note_labeled.value[0]) + for raw in labeled_raws: + key = _decode_aes_key(raw) + if key: + return key + if labeled_raws: + logger.warning( + 'CNAPP: "%s" field is present on encrypter record %s but is not a valid AES-256 key; ' + 'not using other note fields.', + CNAPP_ENCRYPTION_KEY_LABEL, config_record_uid, + ) + return None + first_note = record.get_typed_field('note') + if first_note and first_note.value: + key = _decode_aes_key(first_note.value[0]) + if key: + return key + return None + + +def _decrypt_cnapp_payload(payload_bytes, key): + """Decrypt a CNAPP queue payload using the Encrypter's AES-256-GCM key. + + Wire format (matches vault's `decryptCnappQueueItem` in cloudSecurityUtils.ts): + payload_bytes (proto field, base64url-decoded by us) is UTF-8 base64url text + of a JSON envelope `{"encrypted_payload":"","alg":"AES-256-GCM","version":"1"}`. + encrypted_payload base64url-decodes to `nonce(12) || ciphertext || tag(16)` — + the standard layout AESGCM.decrypt expects. + + Returns a dict on success; raises Exception on bad envelope / wrong key / bad alg + so the caller can surface a meaningful warning.""" + from cryptography.hazmat.primitives.ciphers.aead import AESGCM + envelope_b64 = payload_bytes.decode('utf-8') + envelope_json = base64.urlsafe_b64decode(envelope_b64 + '=' * (-len(envelope_b64) % 4)) + envelope = json.loads(envelope_json) + alg = envelope.get('alg') + if alg != 'AES-256-GCM': + raise ValueError(f"Unsupported or missing CNAPP payload algorithm: {alg!r}") + ciphertext_b64 = envelope.get('encrypted_payload') or '' + ciphertext = base64.urlsafe_b64decode(ciphertext_b64 + '=' * (-len(ciphertext_b64) % 4)) + if len(ciphertext) < 12 + 16: + raise ValueError('CNAPP ciphertext shorter than nonce+tag — corrupt payload') + nonce, body = ciphertext[:12], ciphertext[12:] + plaintext = AESGCM(key).decrypt(nonce, body, None) + return json.loads(plaintext.decode('utf-8')) + + +def _resolve_status(value, allow_all=True): # type: (str|int|None, bool) -> int + """Accept either the numeric status id or its case-insensitive name.""" + if value is None or value == '': + status_id = 0 + elif isinstance(value, int): + status_id = value + else: + s = str(value).strip().lower() + if s.lstrip('-').isdigit(): + status_id = int(s) + elif s in QUEUE_STATUS_BY_NAME: + status_id = QUEUE_STATUS_BY_NAME[s] + else: + raise CommandError( + 'pam cnapp', + f"Unknown status '{value}'. Valid: {', '.join(QUEUE_STATUS_BY_NAME)} or 0 for ALL.", + ) + if status_id == 0: + if allow_all: + return 0 + raise CommandError('pam cnapp', 'A specific status is required (cannot be 0/ALL).') + if status_id not in QUEUE_STATUS_BY_ID: + raise CommandError( + 'pam cnapp', + f"Unknown status id {status_id}. Valid ids: {', '.join(str(i) for i in sorted(QUEUE_STATUS_BY_ID))}.", + ) + return status_id + + +def _format_timestamp(epoch_ms): + """krouter emits epoch-millis for received/resolved timestamps; render as UTC ISO.""" + if not epoch_ms: + return '' + try: + return datetime.fromtimestamp(int(epoch_ms) / 1000, tz=timezone.utc).isoformat() + except (ValueError, TypeError, OSError): + return f'' + + +class PAMCnappCommand(GroupCommand): + """Root for the `pam cnapp ...` command tree.""" + + def __init__(self): + super(PAMCnappCommand, self).__init__() + self.register_command('config', PAMCnappConfigCommand(), + 'Manage CNAPP provider configuration', 'c') + self.register_command('queue', PAMCnappQueueCommand(), + 'Manage CNAPP issue queue', 'q') + self.default_verb = 'queue' + + +# --------------------------------------------------------------------------- +# Configuration sub-tree +# --------------------------------------------------------------------------- + +class PAMCnappConfigCommand(GroupCommand): + + def __init__(self): + super(PAMCnappConfigCommand, self).__init__() + self.register_command('set', PAMCnappConfigSetCommand(), + 'Create or update CNAPP provider configuration') + self.register_command('test', PAMCnappConfigTestCommand(), + 'Validate CNAPP provider credentials without saving') + self.register_command('test-encrypter', PAMCnappConfigTestEncrypterCommand(), + 'Health-check the customer Encrypter at /health') + self.register_command('read', PAMCnappConfigReadCommand(), + 'Read the persisted CNAPP configuration for a network') + self.register_command('delete', PAMCnappConfigDeleteCommand(), + 'Delete the CNAPP configuration on a network') + self.default_verb = '' + + +def _add_configuration_args(parser, require_secret=True, optional_secret_on_set=False): + parser.add_argument('--network-uid', '-n', required=True, dest='network_uid', + help='Network record UID (base64url).') + parser.add_argument('--provider', '-p', required=True, dest='provider', + help='CNAPP provider keyword: wiz (case-insensitive).') + parser.add_argument('--client-id', required=True, dest='client_id', + help='Provider API client ID / app ID.') + if optional_secret_on_set: + parser.add_argument('--client-secret', required=False, default=None, dest='client_secret', + help='Provider API client secret. Omit on `config set` to keep the existing secret.') + else: + parser.add_argument('--client-secret', required=require_secret, dest='client_secret', + help='Provider API client secret.') + parser.add_argument('--api-endpoint', required=True, dest='api_endpoint_url', + help='Provider API endpoint URL (e.g. https://api.us1.app.wiz.io/graphql).') + parser.add_argument('--auth-endpoint', required=True, dest='auth_endpoint_url', + help='Provider OAuth2 token endpoint URL (e.g. https://auth.app.wiz.io/oauth/token).') + + +class PAMCnappConfigSetCommand(Command): + parser = argparse.ArgumentParser(prog='pam cnapp config set') + _add_configuration_args(parser, optional_secret_on_set=True) + parser.add_argument('--config-record', required=True, dest='cnapp_config_record_uid', + help='UID of the vault record holding the Encrypter URL + key.') + + def get_parser(self): + return PAMCnappConfigSetCommand.parser + + def execute(self, params, **kwargs): + provider = cnapp_helper.provider_from_name(kwargs.get('provider')) + response = cnapp_helper.set_cnapp_configuration( + params, + network_uid=kwargs.get('network_uid'), + provider=provider, + client_id=kwargs.get('client_id'), + client_secret='' if kwargs.get('client_secret') is None else kwargs.get('client_secret'), + api_endpoint_url=kwargs.get('api_endpoint_url'), + cnapp_config_record_uid=kwargs.get('cnapp_config_record_uid'), + auth_endpoint_url=kwargs.get('auth_endpoint_url'), + ) + print(f"{bcolors.OKGREEN}CNAPP configuration saved.{bcolors.ENDC}") + if response is not None: + _print_configuration(response) + return None + + +class PAMCnappConfigTestCommand(Command): + parser = argparse.ArgumentParser(prog='pam cnapp config test') + _add_configuration_args(parser, require_secret=True) + + def get_parser(self): + return PAMCnappConfigTestCommand.parser + + def execute(self, params, **kwargs): + provider = cnapp_helper.provider_from_name(kwargs.get('provider')) + cnapp_helper.test_cnapp_configuration( + params, + network_uid=kwargs.get('network_uid'), + provider=provider, + client_id=kwargs.get('client_id'), + client_secret=kwargs.get('client_secret'), + api_endpoint_url=kwargs.get('api_endpoint_url'), + auth_endpoint_url=kwargs.get('auth_endpoint_url'), + ) + print(f"{bcolors.OKGREEN}CNAPP credentials validated successfully.{bcolors.ENDC}") + + +class PAMCnappConfigTestEncrypterCommand(Command): + parser = argparse.ArgumentParser(prog='pam cnapp config test-encrypter') + parser.add_argument('--url', '-u', required=True, dest='url', + help='Base URL of the Encrypter. krouter probes /health.') + + def get_parser(self): + return PAMCnappConfigTestEncrypterCommand.parser + + def execute(self, params, **kwargs): + cnapp_helper.test_cnapp_encrypter(params, url_base_encrypter=kwargs.get('url')) + print(f"{bcolors.OKGREEN}Encrypter is reachable.{bcolors.ENDC}") + + +class PAMCnappConfigReadCommand(Command): + parser = argparse.ArgumentParser(prog='pam cnapp config read') + parser.add_argument('--network-uid', '-n', required=True, dest='network_uid', + help='Network record UID (base64url).') + parser.add_argument('--provider', '-p', required=True, dest='provider', + help='CNAPP provider keyword: wiz.') + parser.add_argument('--format', dest='format', choices=['table', 'json'], default='table', + help='Output format.') + + def get_parser(self): + return PAMCnappConfigReadCommand.parser + + def execute(self, params, **kwargs): + provider = cnapp_helper.provider_from_name(kwargs.get('provider')) + response = cnapp_helper.read_cnapp_configuration( + params, + network_uid=kwargs.get('network_uid'), + provider=provider, + ) + if response is None: + logger.warning('No CNAPP configuration returned.') + return None + if kwargs.get('format') == 'json': + print(json.dumps(_configuration_to_dict(response), indent=2)) + return None + _print_configuration(response) + return None + + +class PAMCnappConfigDeleteCommand(Command): + parser = argparse.ArgumentParser(prog='pam cnapp config delete') + parser.add_argument('--network-uid', '-n', required=True, dest='network_uid', + help='Network record UID (base64url).') + + def get_parser(self): + return PAMCnappConfigDeleteCommand.parser + + def execute(self, params, **kwargs): + cnapp_helper.delete_cnapp_configuration(params, network_uid=kwargs.get('network_uid')) + print(f"{bcolors.OKGREEN}CNAPP configuration deleted.{bcolors.ENDC}") + + +# --------------------------------------------------------------------------- +# Queue sub-tree +# --------------------------------------------------------------------------- + +class PAMCnappQueueCommand(GroupCommand): + + def __init__(self): + super(PAMCnappQueueCommand, self).__init__() + self.register_command('list', PAMCnappQueueListCommand(), 'List CNAPP queue items', 'l') + self.register_command('associate', PAMCnappQueueAssociateCommand(), + 'Attach a vault record to a queue item', 'a') + self.register_command('remediate', PAMCnappQueueRemediateCommand(), + 'Trigger a remediation action against the gateway', 'r') + self.register_command('set-status', PAMCnappQueueSetStatusCommand(), + 'Update local queue item status (notifies provider best-effort)', 's') + self.register_command('delete', PAMCnappQueueDeleteCommand(), 'Delete a queue item', 'd') + self.default_verb = 'list' + + +class PAMCnappQueueListCommand(Command): + parser = argparse.ArgumentParser(prog='pam cnapp queue list') + parser.add_argument('--network-uid', '-n', required=True, dest='network_uid', + help='Network record UID (base64url).') + parser.add_argument('--status', '-s', required=False, dest='status', default=0, + help='Filter by status name or id (pending/in_progress/resolved/failed/cancelled). Default: all.') + parser.add_argument('--provider', '-p', required=False, dest='provider', default='wiz', + help='CNAPP provider keyword for the config lookup (default: wiz).') + parser.add_argument('--config-record', required=False, dest='config_record_uid', + help='Explicit encrypter vault record UID. Overrides the lookup via `config read`.') + parser.add_argument('--no-decrypt', dest='no_decrypt', action='store_true', + help="Skip payload decryption — show the raw encrypted envelope's metadata only.") + parser.add_argument('--format', dest='format', choices=['table', 'json'], default='table', + help='Output format. Table and JSON are mutually exclusive.') + + def get_parser(self): + return PAMCnappQueueListCommand.parser + + def _resolve_encrypter_key(self, params, kwargs): + """Resolve the AES key: --config-record wins; otherwise fetch `config read` to get + the cnappConfigRecordUid and load the encrypter record from the local vault.""" + if kwargs.get('no_decrypt'): + return None, None + config_record_uid = kwargs.get('config_record_uid') + if not config_record_uid: + try: + provider = cnapp_helper.provider_from_name(kwargs.get('provider') or 'wiz') + config = cnapp_helper.read_cnapp_configuration( + params, network_uid=kwargs.get('network_uid'), provider=provider) + except Exception as e: + logger.debug('CNAPP: could not read configuration for decryption: %s', e) + return None, None + if config is None or not config.cnappConfigRecordUid: + return None, None + config_record_uid = bytes_to_base64(config.cnappConfigRecordUid) + key = _load_encrypter_key(params, config_record_uid) + return key, config_record_uid + + @staticmethod + def _decrypted_summary(decrypted): + """Compact human-readable summary for the table column. Mirrors the columns the + vault Cloud Security view shows: severity, title, resource.""" + if not isinstance(decrypted, dict): + return '' + issue = decrypted.get('issue') or {} + resource = decrypted.get('resource') or {} + control = decrypted.get('control') or {} + bits = [] + sev = issue.get('severity') + if sev: + bits.append(str(sev).upper()) + title = control.get('name') or issue.get('id') + if title: + bits.append(str(title)) + resource_name = resource.get('name') or resource.get('id') + if resource_name: + bits.append(f"on {resource_name}") + return ' · '.join(bits) + + def execute(self, params, **kwargs): + status_filter = _resolve_status(kwargs.get('status')) + response = cnapp_helper.list_cnapp_queue( + params, + network_uid=kwargs.get('network_uid'), + status_filter=status_filter, + ) + items = list(response.items) if response is not None else [] + has_more = bool(response.hasMore) if response is not None else False + + encrypter_key, encrypter_uid = self._resolve_encrypter_key(params, kwargs) + decrypted_by_id = {} + decrypt_errors = {} # type: dict[int, str] + if encrypter_key: + for item in items: + if not item.payload: + continue + try: + decrypted_by_id[item.cnappQueueId] = _decrypt_cnapp_payload(item.payload, encrypter_key) + except Exception as e: + decrypt_errors[item.cnappQueueId] = str(e) + + if kwargs.get('format') == 'json': + json_items = [] + for item in items: + d = _queue_item_to_dict(item) + d.pop('payload', None) + if item.cnappQueueId in decrypted_by_id: + d['decryptedPayload'] = decrypted_by_id[item.cnappQueueId] + elif item.cnappQueueId in decrypt_errors: + d['decryptError'] = decrypt_errors[item.cnappQueueId] + json_items.append(d) + payload = {'items': json_items, 'hasMore': has_more} + print(json.dumps(payload, indent=2, default=str)) + return None + + if not items: + print('No CNAPP queue items.') + return None + + if encrypter_key is None and not kwargs.get('no_decrypt'): + print(f"{bcolors.WARNING}No encrypter key resolved — payloads will be shown as 'encrypted'. " + f"Pass --config-record or run after `pam cnapp config read` succeeds.{bcolors.ENDC}") + + headers = ['Queue ID', 'Provider', 'Status', 'Received (UTC)', 'Resolved (UTC)', 'Record UID', 'Issue'] + rows = [] + for item in items: + if item.cnappQueueId in decrypted_by_id: + issue_cell = self._decrypted_summary(decrypted_by_id[item.cnappQueueId]) + elif not item.payload: + issue_cell = '' + elif kwargs.get('no_decrypt'): + issue_cell = '' + else: + issue_cell = f"{bcolors.WARNING}{bcolors.ENDC}" + rows.append([ + item.cnappQueueId, + cnapp_helper.CnappProvider.Name(item.cnappProviderId), + QUEUE_STATUS_BY_ID.get(item.cnappQueueStatusId, str(item.cnappQueueStatusId)), + _format_timestamp(item.receivedAt), + _format_timestamp(item.resolvedAt), + bytes_to_base64(item.recordUid) if item.recordUid else '', + issue_cell, + ]) + dump_report_data(rows, headers, fmt='table', filename='', row_number=False) + for queue_id, msg in decrypt_errors.items(): + print(f"{bcolors.WARNING}Queue item {queue_id}: failed to decrypt payload ({msg}).{bcolors.ENDC}") + if has_more: + print(f"{bcolors.WARNING}More queue items exist (hasMore=true). " + f"CLI paging is not available yet — resolve or delete returned items to see more.{bcolors.ENDC}") + return None + + +class PAMCnappQueueAssociateCommand(Command): + parser = argparse.ArgumentParser(prog='pam cnapp queue associate') + parser.add_argument('--queue-id', '-q', required=True, type=int, dest='cnapp_queue_id', + help='Queue item ID (from `pam cnapp queue list`).') + parser.add_argument('--record-uid', '-r', required=True, dest='record_uid', + help='Vault record UID to associate (base64url).') + + def get_parser(self): + return PAMCnappQueueAssociateCommand.parser + + def execute(self, params, **kwargs): + cnapp_helper.associate_cnapp_record( + params, + cnapp_queue_id=kwargs.get('cnapp_queue_id'), + record_uid=kwargs.get('record_uid'), + ) + print(f"{bcolors.OKGREEN}Record associated with queue item {kwargs.get('cnapp_queue_id')}.{bcolors.ENDC}") + + +class PAMCnappQueueRemediateCommand(Command): + parser = argparse.ArgumentParser(prog='pam cnapp queue remediate') + parser.add_argument('--queue-id', '-q', required=True, type=int, dest='cnapp_queue_id', + help='Queue item ID.') + parser.add_argument('--action', '-a', required=True, dest='action_type', + help='Remediation action: rotate_credentials, manage_access, jit_access, remove_standing_privilege.') + parser.add_argument('--provider', '-p', required=False, dest='provider', + help='Provider keyword (wiz). Optional — krouter resolves from queue item if omitted.') + parser.add_argument('--config-record', required=False, dest='cnapp_config_record_uid', + help='Configuration record UID (only required for some action types).') + parser.add_argument('--resource-ref', required=False, dest='resource_ref', + help='Resource reference UID for the action.') + parser.add_argument('--pwd-complexity', required=False, dest='pwd_complexity', + help='Password complexity JSON (rotate_credentials).') + parser.add_argument('--controller-uid', required=False, dest='controller_uid', + help='Override gateway UID.') + parser.add_argument('--message-uid', required=False, dest='message_uid', + help='Client-generated conversation UID for streaming responses.') + parser.add_argument('--group-name', required=False, dest='group_name', + help='Group name (remove_standing_privilege only).') + + def get_parser(self): + return PAMCnappQueueRemediateCommand.parser + + def execute(self, params, **kwargs): + action = cnapp_helper.action_from_name(kwargs.get('action_type')) + provider = None + if kwargs.get('provider'): + provider = cnapp_helper.provider_from_name(kwargs.get('provider')) + response = cnapp_helper.remediate_cnapp_queue_item( + params, + cnapp_queue_id=kwargs.get('cnapp_queue_id'), + action_type=action, + provider=provider, + cnapp_config_record_uid=kwargs.get('cnapp_config_record_uid'), + resource_ref=kwargs.get('resource_ref'), + pwd_complexity=kwargs.get('pwd_complexity'), + controller_uid=kwargs.get('controller_uid'), + message_uid=kwargs.get('message_uid'), + group_name=kwargs.get('group_name'), + ) + if response is None: + print(f"{bcolors.OKGREEN}Remediation dispatched.{bcolors.ENDC}") + return None + action_name = cnapp_helper.CnappRemediationAction.Name(response.actionType) + status_name = QUEUE_STATUS_BY_ID.get(response.cnappQueueStatusId, str(response.cnappQueueStatusId)) + print(f"{bcolors.OKGREEN}Remediation dispatched.{bcolors.ENDC}") + print(f" Action: {action_name}") + print(f" Status: {status_name}") + if response.result: + print(f" Result: {response.result}") + return None + + +class PAMCnappQueueSetStatusCommand(Command): + parser = argparse.ArgumentParser(prog='pam cnapp queue set-status') + parser.add_argument('--queue-id', '-q', required=True, type=int, dest='cnapp_queue_id', + help='Queue item ID.') + parser.add_argument('--status', '-s', required=True, dest='status', + help='New status: pending/in_progress/resolved/failed/cancelled, or its numeric id.') + parser.add_argument('--reason', required=False, dest='reason', + help='Free-form reason (forwarded to provider notification).') + + def get_parser(self): + return PAMCnappQueueSetStatusCommand.parser + + def execute(self, params, **kwargs): + status_id = _resolve_status(kwargs.get('status'), allow_all=False) + response = cnapp_helper.set_cnapp_queue_status( + params, + cnapp_queue_id=kwargs.get('cnapp_queue_id'), + cnapp_queue_status_id=status_id, + reason=kwargs.get('reason'), + ) + applied = response.cnappQueueStatusId if response is not None else status_id + print(f"{bcolors.OKGREEN}Status applied: {QUEUE_STATUS_BY_ID.get(applied, applied)}.{bcolors.ENDC}") + return None + + +class PAMCnappQueueDeleteCommand(Command): + parser = argparse.ArgumentParser(prog='pam cnapp queue delete') + parser.add_argument('--queue-id', '-q', required=True, type=int, dest='cnapp_queue_id', + help='Queue item ID to delete.') + + def get_parser(self): + return PAMCnappQueueDeleteCommand.parser + + def execute(self, params, **kwargs): + cnapp_helper.delete_cnapp_queue_item(params, cnapp_queue_id=kwargs.get('cnapp_queue_id')) + print(f"{bcolors.OKGREEN}Queue item {kwargs.get('cnapp_queue_id')} deleted.{bcolors.ENDC}") + + +# --------------------------------------------------------------------------- +# Formatting helpers +# --------------------------------------------------------------------------- + +def _configuration_to_dict(config): + return { + 'networkUid': bytes_to_base64(config.networkUid) if config.networkUid else '', + 'provider': cnapp_helper.CnappProvider.Name(config.provider), + 'clientId': config.clientId, + 'apiEndpointUrl': config.apiEndpointUrl, + 'authEndpointUrl': config.authEndpointUrl, + 'cnappConfigRecordUid': bytes_to_base64(config.cnappConfigRecordUid) if config.cnappConfigRecordUid else '', + } + + +def _queue_item_to_dict(item): + return { + 'cnappQueueId': item.cnappQueueId, + 'cnappProviderId': cnapp_helper.CnappProvider.Name(item.cnappProviderId), + 'cnappQueueStatusId': item.cnappQueueStatusId, + 'cnappQueueStatusName': QUEUE_STATUS_BY_ID.get(item.cnappQueueStatusId, str(item.cnappQueueStatusId)), + 'receivedAt': item.receivedAt, + 'resolvedAt': item.resolvedAt, + 'networkId': bytes_to_base64(item.networkId) if item.networkId else '', + 'recordUid': bytes_to_base64(item.recordUid) if item.recordUid else '', + } + + +def _uid_display(uid_bytes): + return bytes_to_base64(uid_bytes) if uid_bytes else '(none)' + + +def _print_configuration(config): + print(f"{bcolors.OKBLUE}CNAPP Configuration{bcolors.ENDC}") + print(f" Network UID : {_uid_display(config.networkUid)}") + print(f" Provider : {cnapp_helper.CnappProvider.Name(config.provider)}") + print(f" Client ID : {config.clientId or '(none)'}") + print(f" API Endpoint : {config.apiEndpointUrl or '(none)'}") + print(f" Auth Endpoint : {config.authEndpointUrl or '(none)'}") + print(f" Config Record : {_uid_display(config.cnappConfigRecordUid)}") diff --git a/keepercommander/commands/pam/cnapp_helper.py b/keepercommander/commands/pam/cnapp_helper.py new file mode 100644 index 000000000..c1f0302d6 --- /dev/null +++ b/keepercommander/commands/pam/cnapp_helper.py @@ -0,0 +1,265 @@ +# _ __ +# | |/ /___ ___ _ __ ___ _ _ ® +# | ' set_cnapp_configuration + configuration/test -> test_cnapp_configuration + configuration/test-encrypter -> test_cnapp_encrypter + configuration/read -> read_cnapp_configuration + configuration/delete -> delete_cnapp_configuration + + Queue: + queue -> list_cnapp_queue + queue/associate -> associate_cnapp_record + queue/remediate -> remediate_cnapp_queue_item + queue/set-status -> set_cnapp_queue_status + queue/delete -> delete_cnapp_queue_item + +Failures from the helper layer bubble up as Python exceptions raised by the underlying +HTTP/proto plumbing; callers convert them to user-readable output. +""" + +from typing import Optional + +from keeper_secrets_manager_core.utils import url_safe_str_to_bytes + +from ...params import KeeperParams +from ...proto import cnapp_pb2 + + +# NOTE: `router_helper` is imported lazily inside `_post_request_to_router` below. +# Importing it at module top creates this import-time chain: +# cnapp_helper -> router_helper -> gateway_helper +# -> keepercommander.commands.utils -> commands.ksm +# -> commands.record -> commands.ksm (ksm still partially loaded — crash) +# That `record <-> ksm` cycle is pre-existing and only works because production +# code paths load `record` first. Tests that import `cnapp_helper` cold hit the +# cycle directly. TODO(KC-1290): break the record↔ksm cycle so this wrapper can be removed. +def _post_request_to_router(params, endpoint, **kwargs): + """Lazy proxy to `router_helper._post_request_to_router`. + + Defined as a module-level function so callers (and `unittest.mock.patch.object`) + can keep referring to `cnapp_helper._post_request_to_router` as if it were the + original symbol.""" + from .router_helper import _post_request_to_router as _real_post + return _real_post(params, endpoint, **kwargs) + + +# Public re-exports — let commands/tests reach proto types via the helper module so they +# don't need to know the on-disk proto path. +CnappProvider = cnapp_pb2.CnappProvider +CnappRemediationAction = cnapp_pb2.CnappRemediationAction + + +# --------------------------------------------------------------------------- +# Conversion utilities +# --------------------------------------------------------------------------- + +def _to_uid_bytes(uid): # type: (Optional[str]) -> bytes + """Convert a base64url-encoded UID string to bytes; empty/None -> empty bytes.""" + if not uid: + return b'' + if isinstance(uid, bytes): + return uid + return url_safe_str_to_bytes(uid) + + +def provider_from_name(name): # type: (str) -> int + """Resolve a human-typed provider name (e.g. "wiz") to a CnappProvider enum value. + + Accepts the bare provider keyword ("wiz") or the full proto symbol + ("CNAPP_PROVIDER_WIZ"); case-insensitive. Raises ValueError on unknown input.""" + if not name: + return cnapp_pb2.CNAPP_PROVIDER_UNSPECIFIED + normalized = name.strip().upper() + if not normalized.startswith('CNAPP_PROVIDER_'): + normalized = 'CNAPP_PROVIDER_' + normalized + try: + return cnapp_pb2.CnappProvider.Value(normalized) + except ValueError as e: + valid = [n for n in cnapp_pb2.CnappProvider.keys() if n != 'CNAPP_PROVIDER_UNSPECIFIED'] + raise ValueError(f"Unknown CNAPP provider '{name}'. Valid options: {', '.join(valid)}") from e + + +def action_from_name(name): # type: (str) -> int + """Resolve a remediation action name to its enum int. Case-insensitive; accepts the + short keyword (e.g. "rotate_credentials") or the full proto symbol.""" + if not name: + return cnapp_pb2.UNSPECIFIED + normalized = name.strip().upper().replace('-', '_') + try: + return cnapp_pb2.CnappRemediationAction.Value(normalized) + except ValueError as e: + valid = [n for n in cnapp_pb2.CnappRemediationAction.keys() if n != 'UNSPECIFIED'] + raise ValueError(f"Unknown remediation action '{name}'. Valid options: {', '.join(valid)}") from e + + +# --------------------------------------------------------------------------- +# Configuration endpoints +# --------------------------------------------------------------------------- + +def _build_configuration(network_uid, provider, client_id=None, client_secret=None, + api_endpoint_url=None, cnapp_config_record_uid=None, + auth_endpoint_url=None): + # type: (str, int, Optional[str], Optional[str], Optional[str], Optional[str], Optional[str]) -> cnapp_pb2.CnappConfiguration + rq = cnapp_pb2.CnappConfiguration() + rq.networkUid = _to_uid_bytes(network_uid) + rq.provider = provider + if client_id: + rq.clientId = client_id + if client_secret: + rq.clientSecret = client_secret + if api_endpoint_url: + rq.apiEndpointUrl = api_endpoint_url + if cnapp_config_record_uid: + rq.cnappConfigRecordUid = _to_uid_bytes(cnapp_config_record_uid) + if auth_endpoint_url: + rq.authEndpointUrl = auth_endpoint_url + return rq + + +def set_cnapp_configuration(params, network_uid, provider, client_id, client_secret, + api_endpoint_url, cnapp_config_record_uid, auth_endpoint_url=None): + # type: (KeeperParams, str, int, str, str, str, str, Optional[str]) -> cnapp_pb2.CnappConfiguration + """Create or update the CNAPP provider configuration on a network. + + krouter validates the credentials against the provider before persisting; an empty + `client_secret` tells krouter to keep the previously stored value (useful for edits + that only change the endpoint or record UID). + + `auth_endpoint_url` is the provider's OAuth2 token endpoint, letting customers point + at their own tenant/region (e.g. EU vs US Wiz auth host) without a code change.""" + rq = _build_configuration(network_uid, provider, client_id, client_secret, + api_endpoint_url, cnapp_config_record_uid, auth_endpoint_url) + return _post_request_to_router(params, 'cnapp/configuration/set', rq_proto=rq, + rs_type=cnapp_pb2.CnappConfiguration) + + +def test_cnapp_configuration(params, network_uid, provider, client_id, client_secret, + api_endpoint_url, auth_endpoint_url=None): + # type: (KeeperParams, str, int, str, str, str, Optional[str]) -> None + """Probe the provider with the supplied credentials without persisting anything. + + Returns None on success; raises on validation failure (RRC_BAD_REQUEST with the + provider's reason in the message).""" + rq = _build_configuration(network_uid, provider, client_id, client_secret, + api_endpoint_url, cnapp_config_record_uid=None, + auth_endpoint_url=auth_endpoint_url) + return _post_request_to_router(params, 'cnapp/configuration/test', rq_proto=rq) + + +def test_cnapp_encrypter(params, url_base_encrypter): + # type: (KeeperParams, str) -> None + """Issue a `GET /health` against the customer-deployed Encrypter via krouter. + + Used by the UI/CLI to check that the Encrypter URL is reachable before saving a + configuration that references it. Raises on non-200 or transport error.""" + rq = cnapp_pb2.CnappTestEncrypterRequest() + rq.urlBaseEncrypter = url_base_encrypter + return _post_request_to_router(params, 'cnapp/configuration/test-encrypter', rq_proto=rq) + + +def read_cnapp_configuration(params, network_uid, provider): + # type: (KeeperParams, str, int) -> cnapp_pb2.CnappConfiguration + """Read the persisted CNAPP configuration for a network. Note: krouter never returns + the `clientSecret` field — only the endpoint, client id and config record UID.""" + rq = _build_configuration(network_uid, provider) + return _post_request_to_router(params, 'cnapp/configuration/read', rq_proto=rq, + rs_type=cnapp_pb2.CnappConfiguration) + + +def delete_cnapp_configuration(params, network_uid): + # type: (KeeperParams, str) -> None + """Remove the CNAPP configuration on a network. Raises RRC_BAD_STATE if none exists.""" + rq = cnapp_pb2.CnappDeleteConfigurationRequest() + rq.networkUid = _to_uid_bytes(network_uid) + return _post_request_to_router(params, 'cnapp/configuration/delete', rq_proto=rq) + + +# --------------------------------------------------------------------------- +# Queue endpoints +# --------------------------------------------------------------------------- + +def list_cnapp_queue(params, network_uid, status_filter=0): + # type: (KeeperParams, str, int) -> cnapp_pb2.CnappQueueListResponse + """List queued CNAPP issues for a network. `status_filter=0` returns all statuses.""" + rq = cnapp_pb2.CnappQueueListRequest() + rq.networkUid = _to_uid_bytes(network_uid) + rq.statusFilter = int(status_filter) if status_filter is not None else 0 + return _post_request_to_router(params, 'cnapp/queue', rq_proto=rq, + rs_type=cnapp_pb2.CnappQueueListResponse) + + +def associate_cnapp_record(params, cnapp_queue_id, record_uid): + # type: (KeeperParams, int, str) -> None + """Attach a vault record to a queue item — required before remediation.""" + rq = cnapp_pb2.CnappAssociateRequest() + rq.cnappQueueId = int(cnapp_queue_id) + rq.recordUid = _to_uid_bytes(record_uid) + return _post_request_to_router(params, 'cnapp/queue/associate', rq_proto=rq) + + +def remediate_cnapp_queue_item(params, cnapp_queue_id, action_type, provider=None, + cnapp_config_record_uid=None, resource_ref=None, + pwd_complexity=None, controller_uid=None, + message_uid=None, group_name=None): + # type: (KeeperParams, int, int, Optional[int], Optional[str], Optional[str], Optional[str], Optional[str], Optional[str], Optional[str]) -> cnapp_pb2.CnappRemediateResponse + """Trigger a remediation action against the gateway for a queued issue. + + Currently krouter only dispatches `ROTATE_CREDENTIALS`; other actions return + RRC_BAD_REQUEST. The optional fields are forwarded as-is so this helper stays + forward-compatible with new action types.""" + rq = cnapp_pb2.CnappRemediateRequest() + rq.cnappQueueId = int(cnapp_queue_id) + rq.actionType = int(action_type) + if provider is not None: + rq.provider = int(provider) + if cnapp_config_record_uid: + rq.cnappConfigurationRecordUid = _to_uid_bytes(cnapp_config_record_uid) + if resource_ref: + rq.resourceRef = _to_uid_bytes(resource_ref) + if pwd_complexity: + rq.pwdComplexity = pwd_complexity + if controller_uid: + rq.controllerUid = controller_uid + if message_uid: + rq.messageUid = _to_uid_bytes(message_uid) + if group_name: + rq.groupName = group_name + return _post_request_to_router(params, 'cnapp/queue/remediate', rq_proto=rq, + rs_type=cnapp_pb2.CnappRemediateResponse) + + +def set_cnapp_queue_status(params, cnapp_queue_id, cnapp_queue_status_id, reason=None): + # type: (KeeperParams, int, int, Optional[str]) -> cnapp_pb2.CnappSetStatusResponse + """Set the local status on a queue item; krouter best-effort notifies the provider.""" + rq = cnapp_pb2.CnappSetStatusRequest() + rq.cnappQueueId = int(cnapp_queue_id) + rq.cnappQueueStatusId = int(cnapp_queue_status_id) + if reason: + rq.reason = reason + return _post_request_to_router(params, 'cnapp/queue/set-status', rq_proto=rq, + rs_type=cnapp_pb2.CnappSetStatusResponse) + + +def delete_cnapp_queue_item(params, cnapp_queue_id): + # type: (KeeperParams, int) -> None + """Remove a queue item entirely. Raises RRC_BAD_REQUEST if the queue id is unknown.""" + rq = cnapp_pb2.CnappDeleteQueueItemRequest() + rq.cnappQueueId = int(cnapp_queue_id) + return _post_request_to_router(params, 'cnapp/queue/delete', rq_proto=rq) diff --git a/keepercommander/proto/cnapp_pb2.py b/keepercommander/proto/cnapp_pb2.py new file mode 100644 index 000000000..a146a3e0e --- /dev/null +++ b/keepercommander/proto/cnapp_pb2.py @@ -0,0 +1,181 @@ +# -*- coding: utf-8 -*- +# Generated by the protocol buffer compiler. DO NOT EDIT! +# source: cnapp.proto +"""Generated protocol buffer code.""" +from google.protobuf.internal import enum_type_wrapper +from google.protobuf import descriptor as _descriptor +from google.protobuf import descriptor_pool as _descriptor_pool +from google.protobuf import message as _message +from google.protobuf import reflection as _reflection +from google.protobuf import symbol_database as _symbol_database +# @@protoc_insertion_point(imports) + +_sym_db = _symbol_database.Default() + + + + +DESCRIPTOR = _descriptor_pool.Default().AddSerializedFile(b'\n\x0b\x63napp.proto\x12\x05\x43NAPP\"A\n\x15\x43nappQueueListRequest\x12\x12\n\nnetworkUid\x18\x01 \x01(\x0c\x12\x14\n\x0cstatusFilter\x18\x02 \x01(\x05\"O\n\x16\x43nappQueueListResponse\x12$\n\x05items\x18\x01 \x03(\x0b\x32\x15.CNAPP.CnappQueueItem\x12\x0f\n\x07hasMore\x18\x02 \x01(\x08\"\xd0\x01\n\x0e\x43nappQueueItem\x12\x14\n\x0c\x63nappQueueId\x18\x01 \x01(\x05\x12-\n\x0f\x63nappProviderId\x18\x02 \x01(\x0e\x32\x14.CNAPP.CnappProvider\x12\x1a\n\x12\x63nappQueueStatusId\x18\x03 \x01(\x05\x12\x12\n\nreceivedAt\x18\x04 \x01(\x03\x12\x12\n\nresolvedAt\x18\x05 \x01(\x03\x12\x11\n\tnetworkId\x18\x06 \x01(\x0c\x12\x0f\n\x07payload\x18\x07 \x01(\x0c\x12\x11\n\trecordUid\x18\x08 \x01(\x0c\"@\n\x15\x43nappAssociateRequest\x12\x11\n\trecordUid\x18\x01 \x01(\x0c\x12\x14\n\x0c\x63nappQueueId\x18\x02 \x01(\x05\"4\n\x16\x43nappAssociateResponse\x12\x1a\n\x12\x63nappQueueStatusId\x18\x01 \x01(\x05\"\x97\x02\n\x15\x43nappRemediateRequest\x12\x14\n\x0c\x63nappQueueId\x18\x01 \x01(\x05\x12\x31\n\nactionType\x18\x02 \x01(\x0e\x32\x1d.CNAPP.CnappRemediationAction\x12#\n\x1b\x63nappConfigurationRecordUid\x18\x03 \x01(\x0c\x12\x15\n\rpwdComplexity\x18\x04 \x01(\t\x12\x13\n\x0bresourceRef\x18\x05 \x01(\x0c\x12&\n\x08provider\x18\x06 \x01(\x0e\x32\x14.CNAPP.CnappProvider\x12\x15\n\rcontrollerUid\x18\x07 \x01(\t\x12\x12\n\nmessageUid\x18\x08 \x01(\x0c\x12\x11\n\tgroupName\x18\t \x01(\t\"w\n\x16\x43nappRemediateResponse\x12\x31\n\nactionType\x18\x01 \x01(\x0e\x32\x1d.CNAPP.CnappRemediationAction\x12\x0e\n\x06result\x18\x02 \x01(\t\x12\x1a\n\x12\x63nappQueueStatusId\x18\x03 \x01(\x05\"Y\n\x15\x43nappSetStatusRequest\x12\x14\n\x0c\x63nappQueueId\x18\x01 \x01(\x05\x12\x1a\n\x12\x63nappQueueStatusId\x18\x02 \x01(\x05\x12\x0e\n\x06reason\x18\x03 \x01(\t\"4\n\x16\x43nappSetStatusResponse\x12\x1a\n\x12\x63nappQueueStatusId\x18\x01 \x01(\x05\"3\n\x1b\x43nappDeleteQueueItemRequest\x12\x14\n\x0c\x63nappQueueId\x18\x01 \x01(\x05\"\x1e\n\x1c\x43nappDeleteQueueItemResponse\"\xc7\x01\n\x12\x43nappConfiguration\x12\x12\n\nnetworkUid\x18\x01 \x01(\x0c\x12&\n\x08provider\x18\x02 \x01(\x0e\x32\x14.CNAPP.CnappProvider\x12\x10\n\x08\x63lientId\x18\x03 \x01(\t\x12\x14\n\x0c\x63lientSecret\x18\x04 \x01(\t\x12\x16\n\x0e\x61piEndpointUrl\x18\x05 \x01(\t\x12\x1c\n\x14\x63nappConfigRecordUid\x18\x06 \x01(\x0c\x12\x17\n\x0f\x61uthEndpointUrl\x18\x07 \x01(\t\"5\n\x1f\x43nappDeleteConfigurationRequest\x12\x12\n\nnetworkUid\x18\x01 \x01(\x0c\"5\n\x19\x43nappTestEncrypterRequest\x12\x18\n\x10urlBaseEncrypter\x18\x01 \x01(\t*G\n\rCnappProvider\x12\x1e\n\x1a\x43NAPP_PROVIDER_UNSPECIFIED\x10\x00\x12\x16\n\x12\x43NAPP_PROVIDER_WIZ\x10\x01*\x83\x01\n\x16\x43nappRemediationAction\x12\x0f\n\x0bUNSPECIFIED\x10\x00\x12\x16\n\x12ROTATE_CREDENTIALS\x10\x01\x12\x11\n\rMANAGE_ACCESS\x10\x02\x12\x0e\n\nJIT_ACCESS\x10\x03\x12\x1d\n\x19REMOVE_STANDING_PRIVILEGE\x10\x04\x42!\n\x18\x63om.keepersecurity.protoB\x05\x43nappb\x06proto3') + +_CNAPPPROVIDER = DESCRIPTOR.enum_types_by_name['CnappProvider'] +CnappProvider = enum_type_wrapper.EnumTypeWrapper(_CNAPPPROVIDER) +_CNAPPREMEDIATIONACTION = DESCRIPTOR.enum_types_by_name['CnappRemediationAction'] +CnappRemediationAction = enum_type_wrapper.EnumTypeWrapper(_CNAPPREMEDIATIONACTION) +CNAPP_PROVIDER_UNSPECIFIED = 0 +CNAPP_PROVIDER_WIZ = 1 +UNSPECIFIED = 0 +ROTATE_CREDENTIALS = 1 +MANAGE_ACCESS = 2 +JIT_ACCESS = 3 +REMOVE_STANDING_PRIVILEGE = 4 + + +_CNAPPQUEUELISTREQUEST = DESCRIPTOR.message_types_by_name['CnappQueueListRequest'] +_CNAPPQUEUELISTRESPONSE = DESCRIPTOR.message_types_by_name['CnappQueueListResponse'] +_CNAPPQUEUEITEM = DESCRIPTOR.message_types_by_name['CnappQueueItem'] +_CNAPPASSOCIATEREQUEST = DESCRIPTOR.message_types_by_name['CnappAssociateRequest'] +_CNAPPASSOCIATERESPONSE = DESCRIPTOR.message_types_by_name['CnappAssociateResponse'] +_CNAPPREMEDIATEREQUEST = DESCRIPTOR.message_types_by_name['CnappRemediateRequest'] +_CNAPPREMEDIATERESPONSE = DESCRIPTOR.message_types_by_name['CnappRemediateResponse'] +_CNAPPSETSTATUSREQUEST = DESCRIPTOR.message_types_by_name['CnappSetStatusRequest'] +_CNAPPSETSTATUSRESPONSE = DESCRIPTOR.message_types_by_name['CnappSetStatusResponse'] +_CNAPPDELETEQUEUEITEMREQUEST = DESCRIPTOR.message_types_by_name['CnappDeleteQueueItemRequest'] +_CNAPPDELETEQUEUEITEMRESPONSE = DESCRIPTOR.message_types_by_name['CnappDeleteQueueItemResponse'] +_CNAPPCONFIGURATION = DESCRIPTOR.message_types_by_name['CnappConfiguration'] +_CNAPPDELETECONFIGURATIONREQUEST = DESCRIPTOR.message_types_by_name['CnappDeleteConfigurationRequest'] +_CNAPPTESTENCRYPTERREQUEST = DESCRIPTOR.message_types_by_name['CnappTestEncrypterRequest'] +CnappQueueListRequest = _reflection.GeneratedProtocolMessageType('CnappQueueListRequest', (_message.Message,), { + 'DESCRIPTOR' : _CNAPPQUEUELISTREQUEST, + '__module__' : 'cnapp_pb2' + # @@protoc_insertion_point(class_scope:CNAPP.CnappQueueListRequest) + }) +_sym_db.RegisterMessage(CnappQueueListRequest) + +CnappQueueListResponse = _reflection.GeneratedProtocolMessageType('CnappQueueListResponse', (_message.Message,), { + 'DESCRIPTOR' : _CNAPPQUEUELISTRESPONSE, + '__module__' : 'cnapp_pb2' + # @@protoc_insertion_point(class_scope:CNAPP.CnappQueueListResponse) + }) +_sym_db.RegisterMessage(CnappQueueListResponse) + +CnappQueueItem = _reflection.GeneratedProtocolMessageType('CnappQueueItem', (_message.Message,), { + 'DESCRIPTOR' : _CNAPPQUEUEITEM, + '__module__' : 'cnapp_pb2' + # @@protoc_insertion_point(class_scope:CNAPP.CnappQueueItem) + }) +_sym_db.RegisterMessage(CnappQueueItem) + +CnappAssociateRequest = _reflection.GeneratedProtocolMessageType('CnappAssociateRequest', (_message.Message,), { + 'DESCRIPTOR' : _CNAPPASSOCIATEREQUEST, + '__module__' : 'cnapp_pb2' + # @@protoc_insertion_point(class_scope:CNAPP.CnappAssociateRequest) + }) +_sym_db.RegisterMessage(CnappAssociateRequest) + +CnappAssociateResponse = _reflection.GeneratedProtocolMessageType('CnappAssociateResponse', (_message.Message,), { + 'DESCRIPTOR' : _CNAPPASSOCIATERESPONSE, + '__module__' : 'cnapp_pb2' + # @@protoc_insertion_point(class_scope:CNAPP.CnappAssociateResponse) + }) +_sym_db.RegisterMessage(CnappAssociateResponse) + +CnappRemediateRequest = _reflection.GeneratedProtocolMessageType('CnappRemediateRequest', (_message.Message,), { + 'DESCRIPTOR' : _CNAPPREMEDIATEREQUEST, + '__module__' : 'cnapp_pb2' + # @@protoc_insertion_point(class_scope:CNAPP.CnappRemediateRequest) + }) +_sym_db.RegisterMessage(CnappRemediateRequest) + +CnappRemediateResponse = _reflection.GeneratedProtocolMessageType('CnappRemediateResponse', (_message.Message,), { + 'DESCRIPTOR' : _CNAPPREMEDIATERESPONSE, + '__module__' : 'cnapp_pb2' + # @@protoc_insertion_point(class_scope:CNAPP.CnappRemediateResponse) + }) +_sym_db.RegisterMessage(CnappRemediateResponse) + +CnappSetStatusRequest = _reflection.GeneratedProtocolMessageType('CnappSetStatusRequest', (_message.Message,), { + 'DESCRIPTOR' : _CNAPPSETSTATUSREQUEST, + '__module__' : 'cnapp_pb2' + # @@protoc_insertion_point(class_scope:CNAPP.CnappSetStatusRequest) + }) +_sym_db.RegisterMessage(CnappSetStatusRequest) + +CnappSetStatusResponse = _reflection.GeneratedProtocolMessageType('CnappSetStatusResponse', (_message.Message,), { + 'DESCRIPTOR' : _CNAPPSETSTATUSRESPONSE, + '__module__' : 'cnapp_pb2' + # @@protoc_insertion_point(class_scope:CNAPP.CnappSetStatusResponse) + }) +_sym_db.RegisterMessage(CnappSetStatusResponse) + +CnappDeleteQueueItemRequest = _reflection.GeneratedProtocolMessageType('CnappDeleteQueueItemRequest', (_message.Message,), { + 'DESCRIPTOR' : _CNAPPDELETEQUEUEITEMREQUEST, + '__module__' : 'cnapp_pb2' + # @@protoc_insertion_point(class_scope:CNAPP.CnappDeleteQueueItemRequest) + }) +_sym_db.RegisterMessage(CnappDeleteQueueItemRequest) + +CnappDeleteQueueItemResponse = _reflection.GeneratedProtocolMessageType('CnappDeleteQueueItemResponse', (_message.Message,), { + 'DESCRIPTOR' : _CNAPPDELETEQUEUEITEMRESPONSE, + '__module__' : 'cnapp_pb2' + # @@protoc_insertion_point(class_scope:CNAPP.CnappDeleteQueueItemResponse) + }) +_sym_db.RegisterMessage(CnappDeleteQueueItemResponse) + +CnappConfiguration = _reflection.GeneratedProtocolMessageType('CnappConfiguration', (_message.Message,), { + 'DESCRIPTOR' : _CNAPPCONFIGURATION, + '__module__' : 'cnapp_pb2' + # @@protoc_insertion_point(class_scope:CNAPP.CnappConfiguration) + }) +_sym_db.RegisterMessage(CnappConfiguration) + +CnappDeleteConfigurationRequest = _reflection.GeneratedProtocolMessageType('CnappDeleteConfigurationRequest', (_message.Message,), { + 'DESCRIPTOR' : _CNAPPDELETECONFIGURATIONREQUEST, + '__module__' : 'cnapp_pb2' + # @@protoc_insertion_point(class_scope:CNAPP.CnappDeleteConfigurationRequest) + }) +_sym_db.RegisterMessage(CnappDeleteConfigurationRequest) + +CnappTestEncrypterRequest = _reflection.GeneratedProtocolMessageType('CnappTestEncrypterRequest', (_message.Message,), { + 'DESCRIPTOR' : _CNAPPTESTENCRYPTERREQUEST, + '__module__' : 'cnapp_pb2' + # @@protoc_insertion_point(class_scope:CNAPP.CnappTestEncrypterRequest) + }) +_sym_db.RegisterMessage(CnappTestEncrypterRequest) + +if _descriptor._USE_C_DESCRIPTORS == False: + + DESCRIPTOR._options = None + DESCRIPTOR._serialized_options = b'\n\030com.keepersecurity.protoB\005Cnapp' + _CNAPPPROVIDER._serialized_start=1446 + _CNAPPPROVIDER._serialized_end=1517 + _CNAPPREMEDIATIONACTION._serialized_start=1520 + _CNAPPREMEDIATIONACTION._serialized_end=1651 + _CNAPPQUEUELISTREQUEST._serialized_start=22 + _CNAPPQUEUELISTREQUEST._serialized_end=87 + _CNAPPQUEUELISTRESPONSE._serialized_start=89 + _CNAPPQUEUELISTRESPONSE._serialized_end=168 + _CNAPPQUEUEITEM._serialized_start=171 + _CNAPPQUEUEITEM._serialized_end=379 + _CNAPPASSOCIATEREQUEST._serialized_start=381 + _CNAPPASSOCIATEREQUEST._serialized_end=445 + _CNAPPASSOCIATERESPONSE._serialized_start=447 + _CNAPPASSOCIATERESPONSE._serialized_end=499 + _CNAPPREMEDIATEREQUEST._serialized_start=502 + _CNAPPREMEDIATEREQUEST._serialized_end=781 + _CNAPPREMEDIATERESPONSE._serialized_start=783 + _CNAPPREMEDIATERESPONSE._serialized_end=902 + _CNAPPSETSTATUSREQUEST._serialized_start=904 + _CNAPPSETSTATUSREQUEST._serialized_end=993 + _CNAPPSETSTATUSRESPONSE._serialized_start=995 + _CNAPPSETSTATUSRESPONSE._serialized_end=1047 + _CNAPPDELETEQUEUEITEMREQUEST._serialized_start=1049 + _CNAPPDELETEQUEUEITEMREQUEST._serialized_end=1100 + _CNAPPDELETEQUEUEITEMRESPONSE._serialized_start=1102 + _CNAPPDELETEQUEUEITEMRESPONSE._serialized_end=1132 + _CNAPPCONFIGURATION._serialized_start=1135 + _CNAPPCONFIGURATION._serialized_end=1334 + _CNAPPDELETECONFIGURATIONREQUEST._serialized_start=1336 + _CNAPPDELETECONFIGURATIONREQUEST._serialized_end=1389 + _CNAPPTESTENCRYPTERREQUEST._serialized_start=1391 + _CNAPPTESTENCRYPTERREQUEST._serialized_end=1444 +# @@protoc_insertion_point(module_scope) diff --git a/unit-tests/conftest.py b/unit-tests/conftest.py new file mode 100644 index 000000000..5650ef913 --- /dev/null +++ b/unit-tests/conftest.py @@ -0,0 +1,12 @@ +"""Pytest session hooks for unit-tests. + +CNAPP tests import `cnapp_helper` before `keepercommander.commands.record` is loaded, +which triggers a pre-existing record <-> ksm circular import. Loading `record` first +resolves the cycle (same as production startup order). +""" +import pytest + + +@pytest.fixture(scope='session', autouse=True) +def _preload_commands_record_module(): + import keepercommander.commands.record # noqa: F401 diff --git a/unit-tests/pam/test_cnapp.py b/unit-tests/pam/test_cnapp.py new file mode 100644 index 000000000..8d8421710 --- /dev/null +++ b/unit-tests/pam/test_cnapp.py @@ -0,0 +1,1001 @@ +"""Unit tests for the Commander CNAPP helper and command surface. + +Strategy: every test patches `_post_request_to_router` so we can assert on what the +helper sends to krouter and feed deterministic responses back into the commands. We +deliberately stay one layer below the network — no socket calls, no real protobuf +encryption — but we exercise the real proto serializers so wire-format breakage +surfaces here. +""" +import base64 +import io +import json +import os +import unittest +from contextlib import redirect_stdout +from unittest.mock import MagicMock, patch + +# isort: off +# Pre-load `record` before cnapp modules (record↔ksm cycle). Pytest also loads it via +# unit-tests/conftest.py; keep this guard for `python unit-tests/pam/test_cnapp.py`. +import keepercommander.commands.record # noqa: F401 +# isort: on + +from cryptography.hazmat.primitives.ciphers.aead import AESGCM # noqa: E402 +from keeper_secrets_manager_core.utils import bytes_to_base64 # noqa: E402 + +from keepercommander.commands.pam import cnapp_helper # noqa: E402 +from keepercommander.commands.pam import cnapp_commands # noqa: E402 +from keepercommander.error import CommandError # noqa: E402 +from keepercommander.proto import cnapp_pb2 # noqa: E402 + + +# Sample 16-byte UIDs as base64url (the format Commander callers pass in). +NETWORK_UID = 'AAAAAAAAAAAAAAAAAAAAAA' # 16 zero bytes +RECORD_UID = 'AQEBAQEBAQEBAQEBAQEBAQ' # 16 0x01 bytes +CONFIG_RECORD_UID = 'AgICAgICAgICAgICAgICAg' + + +def _mock_params(): + """Minimal KeeperParams stand-in — the helpers only use it to drive the router_helper + transport, which is mocked here, so a MagicMock is enough.""" + return MagicMock() + + +# --------------------------------------------------------------------------- +# cnapp_helper: enum parsing +# --------------------------------------------------------------------------- + +class TestEnumParsing(unittest.TestCase): + """provider_from_name and action_from_name must accept short or full names and + reject unknown values with a helpful error listing valid options.""" + + def test_provider_short_name(self): + self.assertEqual(cnapp_helper.provider_from_name('wiz'), cnapp_pb2.CNAPP_PROVIDER_WIZ) + + def test_provider_full_name_case_insensitive(self): + self.assertEqual( + cnapp_helper.provider_from_name('cnapp_provider_wiz'), + cnapp_pb2.CNAPP_PROVIDER_WIZ, + ) + + def test_provider_empty_returns_unspecified(self): + self.assertEqual(cnapp_helper.provider_from_name(''), cnapp_pb2.CNAPP_PROVIDER_UNSPECIFIED) + + def test_provider_unknown_raises_with_valid_options(self): + with self.assertRaises(ValueError) as ctx: + cnapp_helper.provider_from_name('aws') + self.assertIn('WIZ', str(ctx.exception).upper()) + + def test_action_short_name(self): + self.assertEqual( + cnapp_helper.action_from_name('rotate_credentials'), + cnapp_pb2.ROTATE_CREDENTIALS, + ) + + def test_action_hyphenated(self): + # The CLI accepts hyphens (`--action remove-standing-privilege`) for ergonomics; + # helper must normalize before resolving the enum. + self.assertEqual( + cnapp_helper.action_from_name('remove-standing-privilege'), + cnapp_pb2.REMOVE_STANDING_PRIVILEGE, + ) + + def test_action_unknown_raises(self): + with self.assertRaises(ValueError): + cnapp_helper.action_from_name('teleport') + + def test_action_empty_returns_unspecified(self): + self.assertEqual(cnapp_helper.action_from_name(''), cnapp_pb2.UNSPECIFIED) + + +# --------------------------------------------------------------------------- +# cnapp_helper: configuration endpoints +# --------------------------------------------------------------------------- + +class TestConfigurationHelpers(unittest.TestCase): + """Each helper must dispatch to the right krouter path with a correctly populated + protobuf request and return the typed response.""" + + def setUp(self): + self.params = _mock_params() + + def _patch_post(self, return_value=None): + return patch.object(cnapp_helper, '_post_request_to_router', return_value=return_value) + + def test_set_configuration_dispatches_with_full_payload(self): + expected_response = cnapp_pb2.CnappConfiguration( + clientId='abc', apiEndpointUrl='https://api.wiz.io') + with self._patch_post(return_value=expected_response) as post: + result = cnapp_helper.set_cnapp_configuration( + self.params, + network_uid=NETWORK_UID, + provider=cnapp_pb2.CNAPP_PROVIDER_WIZ, + client_id='abc', + client_secret='secret', + api_endpoint_url='https://api.wiz.io', + cnapp_config_record_uid=CONFIG_RECORD_UID, + auth_endpoint_url='https://auth.wiz.io/oauth/token', + ) + self.assertIs(result, expected_response) + args, kwargs = post.call_args + self.assertEqual(args[1], 'cnapp/configuration/set') + rq = kwargs['rq_proto'] + self.assertEqual(rq.provider, cnapp_pb2.CNAPP_PROVIDER_WIZ) + self.assertEqual(rq.clientId, 'abc') + self.assertEqual(rq.clientSecret, 'secret') + self.assertEqual(rq.apiEndpointUrl, 'https://api.wiz.io') + self.assertEqual(rq.authEndpointUrl, 'https://auth.wiz.io/oauth/token') + self.assertEqual(len(rq.networkUid), 16) + self.assertEqual(len(rq.cnappConfigRecordUid), 16) + self.assertIs(kwargs['rs_type'], cnapp_pb2.CnappConfiguration) + + def test_set_configuration_omits_empty_secret_to_keep_existing(self): + """Edge case: passing '' for client_secret on set must leave the field blank in + the request so krouter can splice in the previously stored secret.""" + with self._patch_post() as post: + cnapp_helper.set_cnapp_configuration( + self.params, + network_uid=NETWORK_UID, + provider=cnapp_pb2.CNAPP_PROVIDER_WIZ, + client_id='abc', + client_secret='', + api_endpoint_url='https://api.wiz.io', + cnapp_config_record_uid=CONFIG_RECORD_UID, + ) + rq = post.call_args.kwargs['rq_proto'] + self.assertEqual(rq.clientSecret, '') + + def test_test_configuration_dispatches_to_test_endpoint(self): + with self._patch_post() as post: + cnapp_helper.test_cnapp_configuration( + self.params, + network_uid=NETWORK_UID, + provider=cnapp_pb2.CNAPP_PROVIDER_WIZ, + client_id='abc', + client_secret='secret', + api_endpoint_url='https://api.wiz.io', + auth_endpoint_url='https://auth.wiz.io/oauth/token', + ) + self.assertEqual(post.call_args.args[1], 'cnapp/configuration/test') + self.assertEqual(post.call_args.kwargs['rq_proto'].authEndpointUrl, 'https://auth.wiz.io/oauth/token') + # test endpoint never persists, so it must not require / send config record UID + self.assertEqual(post.call_args.kwargs['rq_proto'].cnappConfigRecordUid, b'') + + def test_test_encrypter_sets_url(self): + with self._patch_post() as post: + cnapp_helper.test_cnapp_encrypter(self.params, url_base_encrypter='https://encr.local') + rq = post.call_args.kwargs['rq_proto'] + self.assertEqual(post.call_args.args[1], 'cnapp/configuration/test-encrypter') + self.assertEqual(rq.urlBaseEncrypter, 'https://encr.local') + + def test_read_configuration_uses_read_endpoint(self): + with self._patch_post(return_value=cnapp_pb2.CnappConfiguration()) as post: + cnapp_helper.read_cnapp_configuration( + self.params, network_uid=NETWORK_UID, provider=cnapp_pb2.CNAPP_PROVIDER_WIZ) + self.assertEqual(post.call_args.args[1], 'cnapp/configuration/read') + self.assertIs(post.call_args.kwargs['rs_type'], cnapp_pb2.CnappConfiguration) + + def test_delete_configuration_uses_delete_endpoint(self): + with self._patch_post() as post: + cnapp_helper.delete_cnapp_configuration(self.params, network_uid=NETWORK_UID) + self.assertEqual(post.call_args.args[1], 'cnapp/configuration/delete') + self.assertEqual(len(post.call_args.kwargs['rq_proto'].networkUid), 16) + + +# --------------------------------------------------------------------------- +# cnapp_helper: queue endpoints +# --------------------------------------------------------------------------- + +class TestQueueHelpers(unittest.TestCase): + def setUp(self): + self.params = _mock_params() + + def _patch_post(self, return_value=None): + return patch.object(cnapp_helper, '_post_request_to_router', return_value=return_value) + + def test_list_queue_with_status_filter(self): + items = cnapp_pb2.CnappQueueListResponse( + items=[cnapp_pb2.CnappQueueItem(cnappQueueId=42)]) + with self._patch_post(return_value=items) as post: + response = cnapp_helper.list_cnapp_queue( + self.params, network_uid=NETWORK_UID, status_filter=1) + self.assertEqual(post.call_args.args[1], 'cnapp/queue') + self.assertEqual(post.call_args.kwargs['rq_proto'].statusFilter, 1) + self.assertEqual(response.items[0].cnappQueueId, 42) + + def test_list_queue_defaults_to_all_status(self): + with self._patch_post(return_value=cnapp_pb2.CnappQueueListResponse()) as post: + cnapp_helper.list_cnapp_queue(self.params, network_uid=NETWORK_UID) + self.assertEqual(post.call_args.kwargs['rq_proto'].statusFilter, 0) + + def test_associate_record_dispatches(self): + with self._patch_post() as post: + cnapp_helper.associate_cnapp_record( + self.params, cnapp_queue_id=7, record_uid=RECORD_UID) + rq = post.call_args.kwargs['rq_proto'] + self.assertEqual(post.call_args.args[1], 'cnapp/queue/associate') + self.assertEqual(rq.cnappQueueId, 7) + self.assertEqual(len(rq.recordUid), 16) + + def test_remediate_forwards_optional_fields(self): + with self._patch_post(return_value=cnapp_pb2.CnappRemediateResponse()) as post: + cnapp_helper.remediate_cnapp_queue_item( + self.params, + cnapp_queue_id=3, + action_type=cnapp_pb2.ROTATE_CREDENTIALS, + provider=cnapp_pb2.CNAPP_PROVIDER_WIZ, + cnapp_config_record_uid=CONFIG_RECORD_UID, + resource_ref=RECORD_UID, + pwd_complexity='{"len":24}', + controller_uid='gateway-1', + message_uid=RECORD_UID, + group_name='Admins', + ) + rq = post.call_args.kwargs['rq_proto'] + self.assertEqual(post.call_args.args[1], 'cnapp/queue/remediate') + self.assertEqual(rq.cnappQueueId, 3) + self.assertEqual(rq.actionType, cnapp_pb2.ROTATE_CREDENTIALS) + self.assertEqual(rq.provider, cnapp_pb2.CNAPP_PROVIDER_WIZ) + self.assertEqual(rq.pwdComplexity, '{"len":24}') + self.assertEqual(rq.controllerUid, 'gateway-1') + self.assertEqual(rq.groupName, 'Admins') + + def test_remediate_minimal_fields(self): + """No optional fields — only queueId and actionType must be set on the wire.""" + with self._patch_post(return_value=cnapp_pb2.CnappRemediateResponse()) as post: + cnapp_helper.remediate_cnapp_queue_item( + self.params, cnapp_queue_id=9, action_type=cnapp_pb2.ROTATE_CREDENTIALS) + rq = post.call_args.kwargs['rq_proto'] + self.assertEqual(rq.cnappQueueId, 9) + self.assertEqual(rq.provider, 0) + self.assertEqual(rq.pwdComplexity, '') + self.assertEqual(rq.controllerUid, '') + self.assertEqual(rq.groupName, '') + + def test_set_status_with_reason(self): + with self._patch_post(return_value=cnapp_pb2.CnappSetStatusResponse(cnappQueueStatusId=3)) as post: + response = cnapp_helper.set_cnapp_queue_status( + self.params, cnapp_queue_id=11, cnapp_queue_status_id=3, reason='Manually resolved') + self.assertEqual(post.call_args.args[1], 'cnapp/queue/set-status') + self.assertEqual(post.call_args.kwargs['rq_proto'].reason, 'Manually resolved') + self.assertEqual(response.cnappQueueStatusId, 3) + + def test_delete_queue_item_dispatches(self): + with self._patch_post() as post: + cnapp_helper.delete_cnapp_queue_item(self.params, cnapp_queue_id=11) + self.assertEqual(post.call_args.args[1], 'cnapp/queue/delete') + self.assertEqual(post.call_args.kwargs['rq_proto'].cnappQueueId, 11) + + +# --------------------------------------------------------------------------- +# cnapp_helper: error propagation +# --------------------------------------------------------------------------- + +class TestHelperErrorPropagation(unittest.TestCase): + """The router layer raises on RRC_!=OK; helpers must NOT swallow those errors.""" + + def test_set_configuration_propagates_router_error(self): + params = _mock_params() + with patch.object(cnapp_helper, '_post_request_to_router', + side_effect=Exception('Credential validation failed: Unauthorized')): + with self.assertRaises(Exception) as ctx: + cnapp_helper.set_cnapp_configuration( + params, + network_uid=NETWORK_UID, + provider=cnapp_pb2.CNAPP_PROVIDER_WIZ, + client_id='abc', + client_secret='bad', + api_endpoint_url='https://api.wiz.io', + cnapp_config_record_uid=CONFIG_RECORD_UID, + ) + self.assertIn('Credential validation failed', str(ctx.exception)) + + +# --------------------------------------------------------------------------- +# cnapp_commands: status resolver +# --------------------------------------------------------------------------- + +class TestStatusResolver(unittest.TestCase): + + def test_numeric_passes_through(self): + self.assertEqual(cnapp_commands._resolve_status('1'), 1) + self.assertEqual(cnapp_commands._resolve_status(2), 2) + + def test_unknown_numeric_id_raises(self): + with self.assertRaises(CommandError): + cnapp_commands._resolve_status(99) + + def test_zero_is_all(self): + self.assertEqual(cnapp_commands._resolve_status('0'), 0) + self.assertEqual(cnapp_commands._resolve_status(None), 0) + self.assertEqual(cnapp_commands._resolve_status(''), 0) + + def test_named_status_case_insensitive(self): + self.assertEqual(cnapp_commands._resolve_status('PENDING'), 1) + self.assertEqual(cnapp_commands._resolve_status('in_progress'), 2) + self.assertEqual(cnapp_commands._resolve_status('Resolved'), 3) + + def test_unknown_status_raises_command_error(self): + with self.assertRaises(CommandError): + cnapp_commands._resolve_status('flapping') + + +# --------------------------------------------------------------------------- +# cnapp_commands: end-to-end (helpers patched) +# --------------------------------------------------------------------------- + +class TestConfigCommands(unittest.TestCase): + def setUp(self): + self.params = _mock_params() + + def _capture_stdout(self): + buf = io.StringIO() + return buf, redirect_stdout(buf) + + def test_config_set_calls_helper_with_resolved_provider(self): + with patch.object(cnapp_commands.cnapp_helper, 'set_cnapp_configuration', + return_value=cnapp_pb2.CnappConfiguration(clientId='abc', + apiEndpointUrl='https://api.wiz.io', + provider=cnapp_pb2.CNAPP_PROVIDER_WIZ)) as helper: + buf, ctx = self._capture_stdout() + with ctx: + cnapp_commands.PAMCnappConfigSetCommand().execute( + self.params, + network_uid=NETWORK_UID, + provider='wiz', + client_id='abc', + client_secret='secret', + api_endpoint_url='https://api.wiz.io', + cnapp_config_record_uid=CONFIG_RECORD_UID, + auth_endpoint_url='https://auth.wiz.io/oauth/token', + ) + helper.assert_called_once() + kwargs = helper.call_args.kwargs + self.assertEqual(kwargs['provider'], cnapp_pb2.CNAPP_PROVIDER_WIZ) + self.assertEqual(kwargs['client_secret'], 'secret') + self.assertEqual(kwargs['auth_endpoint_url'], 'https://auth.wiz.io/oauth/token') + self.assertIn('saved', buf.getvalue().lower()) + + def test_config_set_blank_secret_passes_through(self): + """Edge case: the CLI must forward an empty secret unchanged so krouter can + keep the existing value.""" + with patch.object(cnapp_commands.cnapp_helper, 'set_cnapp_configuration', + return_value=cnapp_pb2.CnappConfiguration()) as helper: + with redirect_stdout(io.StringIO()): + cnapp_commands.PAMCnappConfigSetCommand().execute( + self.params, + network_uid=NETWORK_UID, + provider='wiz', + client_id='abc', + client_secret='', # explicit + api_endpoint_url='https://api.wiz.io', + cnapp_config_record_uid=CONFIG_RECORD_UID, + ) + self.assertEqual(helper.call_args.kwargs['client_secret'], '') + + def test_config_set_omitted_secret_keeps_existing(self): + with patch.object(cnapp_commands.cnapp_helper, 'set_cnapp_configuration', + return_value=cnapp_pb2.CnappConfiguration()) as helper: + with redirect_stdout(io.StringIO()): + cnapp_commands.PAMCnappConfigSetCommand().execute( + self.params, + network_uid=NETWORK_UID, + provider='wiz', + client_id='abc', + client_secret=None, + api_endpoint_url='https://api.wiz.io', + cnapp_config_record_uid=CONFIG_RECORD_UID, + ) + self.assertEqual(helper.call_args.kwargs['client_secret'], '') + + def test_config_set_invalid_provider_raises(self): + with self.assertRaises(ValueError): + cnapp_commands.PAMCnappConfigSetCommand().execute( + self.params, + network_uid=NETWORK_UID, + provider='bogus', + client_id='abc', + client_secret='secret', + api_endpoint_url='https://api.wiz.io', + cnapp_config_record_uid=CONFIG_RECORD_UID, + ) + + def test_config_test_prints_success(self): + with patch.object(cnapp_commands.cnapp_helper, 'test_cnapp_configuration', return_value=None) as helper: + buf = io.StringIO() + with redirect_stdout(buf): + cnapp_commands.PAMCnappConfigTestCommand().execute( + self.params, + network_uid=NETWORK_UID, + provider='wiz', + client_id='abc', + client_secret='secret', + api_endpoint_url='https://api.wiz.io', + auth_endpoint_url='https://auth.wiz.io/oauth/token', + ) + self.assertEqual(helper.call_args.kwargs['auth_endpoint_url'], 'https://auth.wiz.io/oauth/token') + self.assertIn('validated', buf.getvalue().lower()) + + def test_config_test_propagates_helper_error(self): + with patch.object(cnapp_commands.cnapp_helper, 'test_cnapp_configuration', + side_effect=Exception('Credential validation failed: bad')): + with self.assertRaises(Exception): + cnapp_commands.PAMCnappConfigTestCommand().execute( + self.params, + network_uid=NETWORK_UID, + provider='wiz', + client_id='abc', + client_secret='bad', + api_endpoint_url='https://api.wiz.io', + ) + + def test_config_test_encrypter_success(self): + with patch.object(cnapp_commands.cnapp_helper, 'test_cnapp_encrypter', return_value=None) as helper: + buf = io.StringIO() + with redirect_stdout(buf): + cnapp_commands.PAMCnappConfigTestEncrypterCommand().execute( + self.params, url='https://encr.local') + helper.assert_called_once_with(self.params, url_base_encrypter='https://encr.local') + self.assertIn('reachable', buf.getvalue().lower()) + + def test_config_read_table_format(self): + config = cnapp_pb2.CnappConfiguration( + clientId='abc', + apiEndpointUrl='https://api.wiz.io', + authEndpointUrl='https://auth.wiz.io/oauth/token', + provider=cnapp_pb2.CNAPP_PROVIDER_WIZ, + ) + with patch.object(cnapp_commands.cnapp_helper, 'read_cnapp_configuration', return_value=config): + buf = io.StringIO() + with redirect_stdout(buf): + cnapp_commands.PAMCnappConfigReadCommand().execute( + self.params, network_uid=NETWORK_UID, provider='wiz', format='table') + output = buf.getvalue() + self.assertIn('CNAPP Configuration', output) + self.assertIn('https://api.wiz.io', output) + self.assertIn('https://auth.wiz.io/oauth/token', output) + + def test_config_read_json_format(self): + config = cnapp_pb2.CnappConfiguration( + clientId='abc', + apiEndpointUrl='https://api.wiz.io', + authEndpointUrl='https://auth.wiz.io/oauth/token', + provider=cnapp_pb2.CNAPP_PROVIDER_WIZ, + ) + with patch.object(cnapp_commands.cnapp_helper, 'read_cnapp_configuration', return_value=config): + buf = io.StringIO() + with redirect_stdout(buf): + result = cnapp_commands.PAMCnappConfigReadCommand().execute( + self.params, network_uid=NETWORK_UID, provider='wiz', format='json') + payload = json.loads(buf.getvalue()) + self.assertEqual(payload['clientId'], 'abc') + self.assertEqual(payload['provider'], 'CNAPP_PROVIDER_WIZ') + self.assertEqual(payload['apiEndpointUrl'], 'https://api.wiz.io') + self.assertEqual(payload['authEndpointUrl'], 'https://auth.wiz.io/oauth/token') + self.assertIsNone(result, 'JSON output is the channel — no value returned to the REPL') + + def test_config_read_handles_none_response(self): + with patch.object(cnapp_commands.cnapp_helper, 'read_cnapp_configuration', return_value=None): + self.assertIsNone(cnapp_commands.PAMCnappConfigReadCommand().execute( + self.params, network_uid=NETWORK_UID, provider='wiz', format='table')) + + def test_config_delete_success(self): + with patch.object(cnapp_commands.cnapp_helper, 'delete_cnapp_configuration', return_value=None) as helper: + buf = io.StringIO() + with redirect_stdout(buf): + cnapp_commands.PAMCnappConfigDeleteCommand().execute(self.params, network_uid=NETWORK_UID) + helper.assert_called_once_with(self.params, network_uid=NETWORK_UID) + self.assertIn('deleted', buf.getvalue().lower()) + + +class TestQueueCommands(unittest.TestCase): + def setUp(self): + self.params = _mock_params() + + def _queue_response(self, items=None, has_more=False): + return cnapp_pb2.CnappQueueListResponse(items=items or [], hasMore=has_more) + + def _queue_item(self, queue_id=1, status_id=1, record_uid=b''): + return cnapp_pb2.CnappQueueItem( + cnappQueueId=queue_id, + cnappProviderId=cnapp_pb2.CNAPP_PROVIDER_WIZ, + cnappQueueStatusId=status_id, + receivedAt=1700000000000, + networkId=b'\x00' * 16, + recordUid=record_uid, + ) + + def test_queue_list_empty(self): + with patch.object(cnapp_commands.cnapp_helper, 'list_cnapp_queue', + return_value=self._queue_response()): + buf = io.StringIO() + with redirect_stdout(buf): + result = cnapp_commands.PAMCnappQueueListCommand().execute( + self.params, network_uid=NETWORK_UID, status=0, format='table', + no_decrypt=True) + self.assertIn('No CNAPP queue items', buf.getvalue()) + self.assertIsNone(result, 'queue list must not return the proto so the REPL does not dump bytes') + + def test_queue_list_with_items_table(self): + item = self._queue_item(queue_id=99, status_id=2, record_uid=b'\x01' * 16) + with patch.object(cnapp_commands.cnapp_helper, 'list_cnapp_queue', + return_value=self._queue_response([item])): + buf = io.StringIO() + with redirect_stdout(buf): + result = cnapp_commands.PAMCnappQueueListCommand().execute( + self.params, network_uid=NETWORK_UID, status=0, format='table', + no_decrypt=True) + output = buf.getvalue() + self.assertIn('99', output) + self.assertIn('IN_PROGRESS', output) + self.assertIn('CNAPP_PROVIDER_WIZ', output) + self.assertIsNone(result) + + def test_queue_list_filter_resolves_named_status(self): + with patch.object(cnapp_commands.cnapp_helper, 'list_cnapp_queue', + return_value=self._queue_response()) as helper: + with redirect_stdout(io.StringIO()): + cnapp_commands.PAMCnappQueueListCommand().execute( + self.params, network_uid=NETWORK_UID, status='pending', format='table', + no_decrypt=True) + self.assertEqual(helper.call_args.kwargs['status_filter'], 1) + + def test_queue_list_json_format(self): + item = self._queue_item(queue_id=5, status_id=3, record_uid=b'\x02' * 16) + with patch.object(cnapp_commands.cnapp_helper, 'list_cnapp_queue', + return_value=self._queue_response([item], has_more=True)): + buf = io.StringIO() + with redirect_stdout(buf): + result = cnapp_commands.PAMCnappQueueListCommand().execute( + self.params, network_uid=NETWORK_UID, status=0, format='json', + no_decrypt=True) + payload = json.loads(buf.getvalue()) + self.assertEqual(payload['items'][0]['cnappQueueId'], 5) + self.assertEqual(payload['items'][0]['cnappQueueStatusName'], 'RESOLVED') + self.assertTrue(payload['hasMore']) + self.assertEqual(payload['items'][0]['recordUid'], + bytes_to_base64(b'\x02' * 16)) + self.assertNotIn('payload', payload['items'][0], + 'raw encrypted payload bytes must not leak into JSON output') + self.assertIsNone(result, 'JSON output stream must not also return a value') + + def test_queue_list_warns_when_has_more(self): + item = self._queue_item() + with patch.object(cnapp_commands.cnapp_helper, 'list_cnapp_queue', + return_value=self._queue_response([item], has_more=True)): + buf = io.StringIO() + with redirect_stdout(buf): + cnapp_commands.PAMCnappQueueListCommand().execute( + self.params, network_uid=NETWORK_UID, status=0, format='table', + no_decrypt=True) + self.assertIn('hasMore=true', buf.getvalue()) + + def test_queue_associate_success(self): + with patch.object(cnapp_commands.cnapp_helper, 'associate_cnapp_record', return_value=None) as helper: + buf = io.StringIO() + with redirect_stdout(buf): + cnapp_commands.PAMCnappQueueAssociateCommand().execute( + self.params, cnapp_queue_id=12, record_uid=RECORD_UID) + helper.assert_called_once_with(self.params, cnapp_queue_id=12, record_uid=RECORD_UID) + self.assertIn('12', buf.getvalue()) + + def test_queue_remediate_prints_response(self): + response = cnapp_pb2.CnappRemediateResponse( + actionType=cnapp_pb2.ROTATE_CREDENTIALS, + result='Scheduled', + cnappQueueStatusId=2, + ) + with patch.object(cnapp_commands.cnapp_helper, 'remediate_cnapp_queue_item', + return_value=response): + buf = io.StringIO() + with redirect_stdout(buf): + cnapp_commands.PAMCnappQueueRemediateCommand().execute( + self.params, + cnapp_queue_id=4, + action_type='rotate_credentials', + provider='wiz', + ) + output = buf.getvalue() + self.assertIn('ROTATE_CREDENTIALS', output) + self.assertIn('IN_PROGRESS', output) + self.assertIn('Scheduled', output) + + def test_queue_remediate_unsupported_action_propagates(self): + with patch.object(cnapp_commands.cnapp_helper, 'remediate_cnapp_queue_item', + side_effect=Exception('Unsupported action type response code: RRC_BAD_REQUEST')): + with self.assertRaises(Exception) as ctx: + cnapp_commands.PAMCnappQueueRemediateCommand().execute( + self.params, + cnapp_queue_id=4, + action_type='jit_access', + ) + self.assertIn('Unsupported', str(ctx.exception)) + + def test_queue_remediate_invalid_action_name(self): + with self.assertRaises(ValueError): + cnapp_commands.PAMCnappQueueRemediateCommand().execute( + self.params, cnapp_queue_id=1, action_type='nuke_everything') + + def test_queue_set_status_normalizes_named(self): + response = cnapp_pb2.CnappSetStatusResponse(cnappQueueStatusId=3) + with patch.object(cnapp_commands.cnapp_helper, 'set_cnapp_queue_status', + return_value=response) as helper: + with redirect_stdout(io.StringIO()): + cnapp_commands.PAMCnappQueueSetStatusCommand().execute( + self.params, cnapp_queue_id=8, status='resolved', reason='manual') + kwargs = helper.call_args.kwargs + self.assertEqual(kwargs['cnapp_queue_status_id'], 3) + self.assertEqual(kwargs['reason'], 'manual') + + def test_queue_set_status_rejects_zero(self): + with self.assertRaises(CommandError): + cnapp_commands.PAMCnappQueueSetStatusCommand().execute( + self.params, cnapp_queue_id=8, status=0) + + def test_queue_set_status_rejects_unknown_name(self): + with self.assertRaises(CommandError): + cnapp_commands.PAMCnappQueueSetStatusCommand().execute( + self.params, cnapp_queue_id=8, status='snoozed') + + def test_queue_delete_success(self): + with patch.object(cnapp_commands.cnapp_helper, 'delete_cnapp_queue_item', + return_value=None) as helper: + buf = io.StringIO() + with redirect_stdout(buf): + cnapp_commands.PAMCnappQueueDeleteCommand().execute(self.params, cnapp_queue_id=22) + helper.assert_called_once_with(self.params, cnapp_queue_id=22) + self.assertIn('22', buf.getvalue()) + + def test_queue_delete_unknown_id_propagates_error(self): + with patch.object(cnapp_commands.cnapp_helper, 'delete_cnapp_queue_item', + side_effect=Exception('Queue item not found: 99 Response code: RRC_BAD_REQUEST')): + with self.assertRaises(Exception): + cnapp_commands.PAMCnappQueueDeleteCommand().execute(self.params, cnapp_queue_id=99) + + +# --------------------------------------------------------------------------- +# Command tree wiring +# --------------------------------------------------------------------------- + +class TestCommandTree(unittest.TestCase): + """Sanity check that the cnapp commands are reachable via `pam cnapp ...`.""" + + def test_pam_cnapp_subcommands(self): + from keepercommander.commands.discoveryrotation import PAMControllerCommand + pam = PAMControllerCommand() + self.assertIn('cnapp', pam.subcommands) + config = pam.subcommands['cnapp'].subcommands['config'] + queue = pam.subcommands['cnapp'].subcommands['queue'] + self.assertEqual( + sorted(config.subcommands), + ['delete', 'read', 'set', 'test', 'test-encrypter'], + ) + self.assertEqual( + sorted(queue.subcommands), + ['associate', 'delete', 'list', 'remediate', 'set-status'], + ) + + +# --------------------------------------------------------------------------- +# Payload decryption — round-trip an AES-256-GCM envelope and decrypt it back +# --------------------------------------------------------------------------- + +def _encrypt_cnapp_payload_for_test(plaintext_json, key): + """Produce a CNAPP queue payload byte string the way the Encrypter would so we can + exercise `_decrypt_cnapp_payload` end-to-end without mocking AES-GCM.""" + nonce = os.urandom(12) + ciphertext = AESGCM(key).encrypt(nonce, plaintext_json.encode('utf-8'), None) + enc_b64url = base64.urlsafe_b64encode(nonce + ciphertext).rstrip(b'=').decode('ascii') + envelope = json.dumps({ + 'encrypted_payload': enc_b64url, + 'alg': 'AES-256-GCM', + 'version': '1', + }).encode('utf-8') + envelope_b64url = base64.urlsafe_b64encode(envelope).rstrip(b'=').decode('ascii') + return envelope_b64url.encode('utf-8') + + +class TestPayloadDecryption(unittest.TestCase): + """`_decrypt_cnapp_payload` must round-trip the envelope produced by the customer + Encrypter (UTF-8 base64url envelope wrapping nonce||ciphertext||tag).""" + + def setUp(self): + self.key = os.urandom(32) + self.plaintext = { + 'issue': {'id': 'wiz-001', 'severity': 'HIGH', 'created': '2026-05-01T00:00:00Z'}, + 'resource': {'name': 'i-abc', 'type': 'EC2', 'cloudPlatform': 'AWS'}, + 'control': {'name': 'Public S3', 'risks': ['data-exposure']}, + 'tags': ['team:platform'], + } + + def test_roundtrip(self): + payload = _encrypt_cnapp_payload_for_test(json.dumps(self.plaintext), self.key) + decrypted = cnapp_commands._decrypt_cnapp_payload(payload, self.key) + self.assertEqual(decrypted['issue']['id'], 'wiz-001') + self.assertEqual(decrypted['resource']['name'], 'i-abc') + + def test_wrong_key_raises(self): + payload = _encrypt_cnapp_payload_for_test(json.dumps(self.plaintext), self.key) + with self.assertRaises(Exception): + cnapp_commands._decrypt_cnapp_payload(payload, os.urandom(32)) + + def test_unsupported_alg_raises(self): + envelope = json.dumps({'encrypted_payload': '', 'alg': 'ChaCha20', 'version': '1'}).encode('utf-8') + payload = base64.urlsafe_b64encode(envelope).rstrip(b'=') + with self.assertRaises(ValueError): + cnapp_commands._decrypt_cnapp_payload(payload, self.key) + + def test_missing_alg_raises(self): + envelope = json.dumps({'encrypted_payload': '', 'version': '1'}).encode('utf-8') + payload = base64.urlsafe_b64encode(envelope).rstrip(b'=') + with self.assertRaises(ValueError) as ctx: + cnapp_commands._decrypt_cnapp_payload(payload, self.key) + self.assertIn('missing', str(ctx.exception).lower()) + + def test_short_ciphertext_raises(self): + envelope = json.dumps({ + 'encrypted_payload': base64.urlsafe_b64encode(b'abc').rstrip(b'=').decode('ascii'), + 'alg': 'AES-256-GCM', + }).encode('utf-8') + payload = base64.urlsafe_b64encode(envelope).rstrip(b'=') + with self.assertRaises(ValueError): + cnapp_commands._decrypt_cnapp_payload(payload, self.key) + + +class TestKeyDecode(unittest.TestCase): + """`_decode_aes_key` must accept both standard and url-safe base64, only when the + decoded length is exactly 32 bytes (AES-256). 16-byte keys are rejected.""" + + def test_standard_base64_32(self): + raw = base64.b64encode(b'\x11' * 32).decode('ascii') + self.assertEqual(cnapp_commands._decode_aes_key(raw), b'\x11' * 32) + + def test_urlsafe_base64_32(self): + raw = base64.urlsafe_b64encode(b'\x22' * 32).decode('ascii') + self.assertEqual(cnapp_commands._decode_aes_key(raw), b'\x22' * 32) + + def test_16_bytes_returns_none(self): + raw = base64.b64encode(b'\x44' * 16).decode('ascii') + self.assertIsNone(cnapp_commands._decode_aes_key(raw)) + + def test_wrong_length_returns_none(self): + raw = base64.b64encode(b'\x33' * 24).decode('ascii') + self.assertIsNone(cnapp_commands._decode_aes_key(raw)) + + def test_garbage_returns_none(self): + self.assertIsNone(cnapp_commands._decode_aes_key('not base64 at all!!!')) + self.assertIsNone(cnapp_commands._decode_aes_key('')) + self.assertIsNone(cnapp_commands._decode_aes_key(None)) + + +class TestLoadEncrypterKey(unittest.TestCase): + """_load_encrypter_key must try all labeled candidates before giving up.""" + + VALID_KEY = b'\xAB' * 32 + INVALID_RAW = 'not-a-valid-key!!' + + def _make_typed_field(self, type_ref, label=None, value=None): + field = MagicMock() + field.type = type_ref + field.label = label + field.value = value or [] + return field + + def _make_record(self, secret_labeled=None, note_labeled=None, note_unlabeled=None): + record = MagicMock(spec=cnapp_commands.vault.TypedRecord) + + def get_typed_field(type_ref, label=None): + if type_ref == 'secret' and label == cnapp_commands.CNAPP_ENCRYPTION_KEY_LABEL: + return secret_labeled + if type_ref == 'note' and label == cnapp_commands.CNAPP_ENCRYPTION_KEY_LABEL: + return note_labeled + if type_ref == 'note' and label is None: + return note_unlabeled + return None + + record.get_typed_field.side_effect = get_typed_field + return record + + def _field_with_value(self, raw): + f = MagicMock() + f.value = [raw] + return f + + def test_returns_valid_secret_labeled_key(self): + raw = base64.b64encode(self.VALID_KEY).decode('ascii') + record = self._make_record(secret_labeled=self._field_with_value(raw)) + params = MagicMock() + with patch.object(cnapp_commands.vault.KeeperRecord, 'load', return_value=record): + result = cnapp_commands._load_encrypter_key(params, 'uid123') + self.assertEqual(result, self.VALID_KEY) + + def test_falls_through_to_note_labeled_when_secret_invalid(self): + """When the secret-labeled field is invalid, note-labeled field must still be tried.""" + valid_raw = base64.b64encode(self.VALID_KEY).decode('ascii') + record = self._make_record( + secret_labeled=self._field_with_value(self.INVALID_RAW), + note_labeled=self._field_with_value(valid_raw), + ) + params = MagicMock() + with patch.object(cnapp_commands.vault.KeeperRecord, 'load', return_value=record): + result = cnapp_commands._load_encrypter_key(params, 'uid123') + self.assertEqual(result, self.VALID_KEY) + + def test_warns_and_returns_none_when_all_labeled_invalid(self): + """If labeled fields exist but all are invalid, warn and return None (no unlabeled fallback).""" + unlabeled = self._field_with_value(base64.b64encode(self.VALID_KEY).decode('ascii')) + record = self._make_record( + secret_labeled=self._field_with_value(self.INVALID_RAW), + note_labeled=self._field_with_value(self.INVALID_RAW), + note_unlabeled=unlabeled, + ) + params = MagicMock() + with patch.object(cnapp_commands.vault.KeeperRecord, 'load', return_value=record): + with self.assertLogs(cnapp_commands.__name__, level='WARNING') as cm: + result = cnapp_commands._load_encrypter_key(params, 'uid123') + self.assertIsNone(result) + self.assertTrue(any('not a valid AES-256 key' in msg for msg in cm.output)) + + def test_falls_back_to_unlabeled_note_when_no_labeled_fields(self): + """When no labeled key fields exist at all, the first unlabeled note field is used.""" + valid_raw = base64.b64encode(self.VALID_KEY).decode('ascii') + unlabeled = self._field_with_value(valid_raw) + record = self._make_record(note_unlabeled=unlabeled) + params = MagicMock() + with patch.object(cnapp_commands.vault.KeeperRecord, 'load', return_value=record): + result = cnapp_commands._load_encrypter_key(params, 'uid123') + self.assertEqual(result, self.VALID_KEY) + + def test_returns_none_for_missing_record_uid(self): + params = MagicMock() + self.assertIsNone(cnapp_commands._load_encrypter_key(params, None)) + self.assertIsNone(cnapp_commands._load_encrypter_key(params, '')) + + +class TestQueueListDecryptionIntegration(unittest.TestCase): + """End-to-end: `queue list` resolves the encrypter key via the vault record, decrypts + each payload, and writes the human summary into the table cell.""" + + def setUp(self): + self.params = _mock_params() + self.key = os.urandom(32) + + def _make_item(self, queue_id, plaintext): + return cnapp_pb2.CnappQueueItem( + cnappQueueId=queue_id, + cnappProviderId=cnapp_pb2.CNAPP_PROVIDER_WIZ, + cnappQueueStatusId=1, + receivedAt=1700000000000, + networkId=b'\x00' * 16, + payload=_encrypt_cnapp_payload_for_test(json.dumps(plaintext), self.key), + ) + + def test_table_shows_decrypted_summary_when_key_resolves(self): + items = [self._make_item(101, { + 'issue': {'id': 'wiz-999', 'severity': 'CRITICAL'}, + 'control': {'name': 'Open SSH'}, + 'resource': {'name': 'prod-db-1'}, + })] + response = cnapp_pb2.CnappQueueListResponse(items=items) + config = cnapp_pb2.CnappConfiguration( + cnappConfigRecordUid=b'\xab' * 16, + provider=cnapp_pb2.CNAPP_PROVIDER_WIZ, + ) + with patch.object(cnapp_commands.cnapp_helper, 'list_cnapp_queue', return_value=response), \ + patch.object(cnapp_commands.cnapp_helper, 'read_cnapp_configuration', return_value=config), \ + patch.object(cnapp_commands, '_load_encrypter_key', return_value=self.key): + buf = io.StringIO() + with redirect_stdout(buf): + result = cnapp_commands.PAMCnappQueueListCommand().execute( + self.params, network_uid=NETWORK_UID, status=0, format='table', + provider='wiz', config_record_uid=None, no_decrypt=False) + output = buf.getvalue() + self.assertIn('CRITICAL', output) + self.assertIn('Open SSH', output) + self.assertIn('prod-db-1', output) + self.assertNotIn('', output, 'payload should have been decrypted') + self.assertIsNone(result) + + def test_table_marks_encrypted_when_key_unavailable(self): + items = [self._make_item(7, {'issue': {'id': 'x'}})] + response = cnapp_pb2.CnappQueueListResponse(items=items) + with patch.object(cnapp_commands.cnapp_helper, 'list_cnapp_queue', return_value=response), \ + patch.object(cnapp_commands.cnapp_helper, 'read_cnapp_configuration', + return_value=cnapp_pb2.CnappConfiguration()): + buf = io.StringIO() + with redirect_stdout(buf): + cnapp_commands.PAMCnappQueueListCommand().execute( + self.params, network_uid=NETWORK_UID, status=0, format='table', + provider='wiz', no_decrypt=False) + output = buf.getvalue() + self.assertIn('', output) + self.assertIn('No encrypter key', output) + + def test_json_includes_decrypted_payload_and_no_raw_payload(self): + plaintext = {'issue': {'id': 'wiz-42'}, 'resource': {'name': 'i-xyz'}} + response = cnapp_pb2.CnappQueueListResponse(items=[self._make_item(42, plaintext)]) + config = cnapp_pb2.CnappConfiguration(cnappConfigRecordUid=b'\xcd' * 16, + provider=cnapp_pb2.CNAPP_PROVIDER_WIZ) + with patch.object(cnapp_commands.cnapp_helper, 'list_cnapp_queue', return_value=response), \ + patch.object(cnapp_commands.cnapp_helper, 'read_cnapp_configuration', return_value=config), \ + patch.object(cnapp_commands, '_load_encrypter_key', return_value=self.key): + buf = io.StringIO() + with redirect_stdout(buf): + cnapp_commands.PAMCnappQueueListCommand().execute( + self.params, network_uid=NETWORK_UID, status=0, format='json', + provider='wiz', no_decrypt=False) + payload = json.loads(buf.getvalue()) + self.assertEqual(payload['items'][0]['decryptedPayload']['issue']['id'], 'wiz-42') + self.assertNotIn('payload', payload['items'][0]) + + def test_decrypt_failure_keeps_other_rows_and_reports(self): + good = self._make_item(1, { + 'issue': {'id': 'wiz-good-should-not-show'}, + 'control': {'name': 'Open SSH'}, + 'resource': {'name': 'good-resource'}, + }) + bad = cnapp_pb2.CnappQueueItem( + cnappQueueId=2, + cnappProviderId=cnapp_pb2.CNAPP_PROVIDER_WIZ, + cnappQueueStatusId=1, + payload=b'this-is-not-a-valid-envelope', + ) + response = cnapp_pb2.CnappQueueListResponse(items=[good, bad]) + config = cnapp_pb2.CnappConfiguration(cnappConfigRecordUid=b'\xef' * 16, + provider=cnapp_pb2.CNAPP_PROVIDER_WIZ) + with patch.object(cnapp_commands.cnapp_helper, 'list_cnapp_queue', return_value=response), \ + patch.object(cnapp_commands.cnapp_helper, 'read_cnapp_configuration', return_value=config), \ + patch.object(cnapp_commands, '_load_encrypter_key', return_value=self.key): + buf = io.StringIO() + with redirect_stdout(buf): + cnapp_commands.PAMCnappQueueListCommand().execute( + self.params, network_uid=NETWORK_UID, status=0, format='table', + provider='wiz', no_decrypt=False) + output = buf.getvalue() + self.assertIn('Open SSH', output) + self.assertNotIn('wiz-good-should-not-show', output) + self.assertIn('good-resource', output) + self.assertIn('', output) + self.assertIn('failed to decrypt payload', output) + + def test_json_reports_decrypt_error(self): + good = self._make_item(1, {'issue': {'id': 'wiz-1'}}) + bad = cnapp_pb2.CnappQueueItem( + cnappQueueId=2, + cnappProviderId=cnapp_pb2.CNAPP_PROVIDER_WIZ, + cnappQueueStatusId=1, + payload=b'not-valid', + ) + response = cnapp_pb2.CnappQueueListResponse(items=[good, bad]) + config = cnapp_pb2.CnappConfiguration(cnappConfigRecordUid=b'\xef' * 16, + provider=cnapp_pb2.CNAPP_PROVIDER_WIZ) + with patch.object(cnapp_commands.cnapp_helper, 'list_cnapp_queue', return_value=response), \ + patch.object(cnapp_commands.cnapp_helper, 'read_cnapp_configuration', return_value=config), \ + patch.object(cnapp_commands, '_load_encrypter_key', return_value=self.key): + buf = io.StringIO() + with redirect_stdout(buf): + cnapp_commands.PAMCnappQueueListCommand().execute( + self.params, network_uid=NETWORK_UID, status=0, format='json', + provider='wiz', no_decrypt=False) + items = json.loads(buf.getvalue())['items'] + self.assertIn('decryptedPayload', items[0]) + self.assertIn('decryptError', items[1]) + self.assertNotIn('decryptedPayload', items[1]) + + def test_no_decrypt_flag_skips_key_lookup(self): + items = [self._make_item(11, {'issue': {'id': 'x'}})] + response = cnapp_pb2.CnappQueueListResponse(items=items) + with patch.object(cnapp_commands.cnapp_helper, 'list_cnapp_queue', return_value=response), \ + patch.object(cnapp_commands, '_load_encrypter_key') as key_loader: + buf = io.StringIO() + with redirect_stdout(buf): + cnapp_commands.PAMCnappQueueListCommand().execute( + self.params, network_uid=NETWORK_UID, status=0, format='table', + no_decrypt=True) + key_loader.assert_not_called() + self.assertNotIn('No encrypter key', buf.getvalue()) + self.assertIn('', buf.getvalue()) + + +if __name__ == '__main__': + unittest.main() diff --git a/unit-tests/pam/test_dag_layer_b_migration.py b/unit-tests/pam/test_dag_layer_b_migration.py index ba381cdcb..7f85e10db 100644 --- a/unit-tests/pam/test_dag_layer_b_migration.py +++ b/unit-tests/pam/test_dag_layer_b_migration.py @@ -155,6 +155,37 @@ def _capture(params, rq): }).encode() meta_mock.assert_called_once() + def test_happy_path_bundles_current_meta_so_krouter_persists_ai_edge(self): + """Regression: krouter's configure_resource only writes a settings edge + when it loads loopEdges, which it does only for requests carrying + meta/jit/connection (UserRest.kt:497). A keeperAiSettings-only request + leaves loopEdges null and the ai_settings write is silently dropped. The + Web Vault always sends meta alongside AI settings; Commander must mirror + that by bundling the resource's current meta in the same request.""" + captured = {} + + def _capture(params, rq): + captured['rq'] = rq + return None + + meta_dict = {'version': 1, 'allowedSettings': {'aiEnabled': True}, 'rotateOnTermination': False} + with _patch_inputs(), \ + patch.object(ai_mod, 'encrypt_aes', return_value=b'CIPHER_BYTES'), \ + patch.object(ai_mod, 'get_resource_settings', return_value=meta_dict) as meta_mock, \ + patch('keepercommander.commands.pam.router_helper.router_configure_resource', side_effect=_capture): + ok = ai_mod.set_resource_keeper_ai_settings( + _mock_params(), RESOURCE_UID_STR, {'level': 'critical'}, config_uid=CONFIG_UID_STR + ) + assert ok is True + rq = captured['rq'] + assert rq.keeperAiSettings == b'CIPHER_BYTES' + # The fix: meta must be present so krouter fetches loopEdges and persists + # the ai_settings edge. Without it the write is a silent no-op. + assert rq.meta == json.dumps(meta_dict).encode() + # meta is read from the resource's current 'meta' DATA edge. + meta_mock.assert_called_once() + assert meta_mock.call_args.args[2] == 'meta' + def test_permission_denied_with_fallback_enabled_calls_legacy(self): legacy_called = {'count': 0} diff --git a/unit-tests/test_command_enterprise_api_keys.py b/unit-tests/test_command_enterprise_api_keys.py index e90108927..92d4183f9 100644 --- a/unit-tests/test_command_enterprise_api_keys.py +++ b/unit-tests/test_command_enterprise_api_keys.py @@ -100,7 +100,7 @@ def test_api_key_list_json_format(self): "name": "SIEM Tool", "status": "Active", "issued_date": "2025-07-08 14:16:07", - "expiration_date": "2026-07-08 14:16:07", + "expiration_date": "2030-07-08 14:16:07", "integration": "SIEM:2" }, { @@ -667,7 +667,7 @@ def communicate_rest_success(params, request, path, rs_type=None): token4.name = "SIEM Tool" token4.enterprise_id = 8560 token4.issuedDate = int(datetime.datetime(2025, 7, 8, 14, 16, 7).timestamp() * 1000) - token4.expirationDate = int(datetime.datetime(2026, 7, 8, 14, 16, 7).timestamp() * 1000) + token4.expirationDate = int(datetime.datetime(2030, 7, 8, 14, 16, 7).timestamp() * 1000) integration7 = token4.integrations.add() integration7.roleName = "SIEM" integration7.apiIntegrationTypeName = "SIEM"