diff --git a/.github/workflows/test.ansible.yml b/.github/workflows/test.ansible.yml index e1706adba..c8b141c53 100644 --- a/.github/workflows/test.ansible.yml +++ b/.github/workflows/test.ansible.yml @@ -3,6 +3,10 @@ name: Test-Ansible on: pull_request: branches: [ master ] + paths: + - 'integration/keeper_secrets_manager_ansible/**' + - 'sdk/python/core/**' + - '.github/workflows/test.ansible.yml' jobs: test-ansible: diff --git a/integration/keeper_secrets_manager_ansible/README.md b/integration/keeper_secrets_manager_ansible/README.md index ee548fb05..b38a22655 100644 --- a/integration/keeper_secrets_manager_ansible/README.md +++ b/integration/keeper_secrets_manager_ansible/README.md @@ -19,6 +19,17 @@ For more information see our official documentation page https://docs.keeper.io/ # Changes +## 1.5.0 +* **Fix**: `keeper_create` crashed with "Could not create record: list index out of range" when a playbook supplied an unpopulated complex field (address, name, host, etc.) with `value: []`. Empty-value fields are now treated as unpopulated, matching the behavior of the underlying vault schema. Root cause in the Python helper library is tracked as KSM-1119. +* KSM-845: Added `folder_uid` parameter to `keeper_create` for subfolder targeting + - Records can now be created in a subfolder within a shared folder, rather than always at the shared folder root + - `shared_folder_uid` remains required; `folder_uid` is optional and additive +* **Security**: VM-1452 / CWE-502 — Replaced pickle with JSON for encrypted record cache serialization + - Cache encrypt/decrypt no longer uses `pickle.loads`, removing insecure deserialization risk + - Legacy or invalid registered caches are ignored; records are fetched from the vault until + `keeper_cache_records` rebuilds a JSON cache + - Existing playbook-registered caches are ephemeral; regenerate with `keeper_cache_records` after upgrade + ## 1.4.0 * KSM-827: Fixed Tower Execution Environment Docker image missing system packages required by AAP - Added `openssh-clients`, `sshpass`, `rsync`, and `git` to `additional_build_packages` in `execution-environment.yml` diff --git a/integration/keeper_secrets_manager_ansible/ansible_galaxy/keepersecurity/keeper_secrets_manager/README.md b/integration/keeper_secrets_manager_ansible/ansible_galaxy/keepersecurity/keeper_secrets_manager/README.md index 42e63e1bf..4ceb06cfb 100644 --- a/integration/keeper_secrets_manager_ansible/ansible_galaxy/keepersecurity/keeper_secrets_manager/README.md +++ b/integration/keeper_secrets_manager_ansible/ansible_galaxy/keepersecurity/keeper_secrets_manager/README.md @@ -119,6 +119,16 @@ configuration file or even a playbook. # Changes +## 1.5.0 +* KSM-845: Added `folder_uid` parameter to `keeper_create` for subfolder targeting + - Records can now be created in a subfolder within a shared folder, rather than always at the shared folder root + - `shared_folder_uid` remains required; `folder_uid` is optional and additive +* **Security**: VM-1452 / CWE-502 — Replaced pickle with JSON for encrypted record cache serialization + - Cache encrypt/decrypt no longer uses `pickle.loads`, removing insecure deserialization risk + - Legacy or invalid registered caches are ignored; records are fetched from the vault until + `keeper_cache_records` rebuilds a JSON cache + - Existing playbook-registered caches are ephemeral; regenerate with `keeper_cache_records` after upgrade + ## 1.4.0 * KSM-827: Fixed Tower Execution Environment Docker image missing system packages required by AAP - Added `openssh-clients`, `sshpass`, `rsync`, and `git` to the EE image diff --git a/integration/keeper_secrets_manager_ansible/keeper_secrets_manager_ansible/__init__.py b/integration/keeper_secrets_manager_ansible/keeper_secrets_manager_ansible/__init__.py index eb41bb7c3..5fa0f5206 100644 --- a/integration/keeper_secrets_manager_ansible/keeper_secrets_manager_ansible/__init__.py +++ b/integration/keeper_secrets_manager_ansible/keeper_secrets_manager_ansible/__init__.py @@ -21,8 +21,6 @@ import random from enum import Enum import traceback -import pickle -import io import base64 import socket @@ -37,6 +35,7 @@ from keeper_secrets_manager_core.core import KSMCache, CreateOptions from keeper_secrets_manager_core.storage import FileKeyValueStorage, InMemoryKeyValueStorage from keeper_secrets_manager_core.utils import generate_password as sdk_generate_password, strtobool + from keeper_secrets_manager_core.dto.dtos import Record as _Record, KeeperFile as _KeeperFile # If keeper_secrets_manager_core is installed, then these will be installed. They are deps. from cryptography.fernet import Fernet @@ -47,6 +46,10 @@ display = Display() +class CacheUnusableError(ValueError): + """Encrypted record cache cannot be decrypted or deserialized; treat as a cache miss.""" + + class KeeperFieldType(Enum): FIELD = "field" CUSTOM_FIELD = "custom_field" @@ -306,16 +309,125 @@ def get_encryption_key(self): return base64.urlsafe_b64encode(kdf.derive(cache_secret.encode())) - def encrypt(self, data): + @staticmethod + def _file_to_dict(keeper_file): + """Serialize a KeeperFile instance to a JSON-safe dictionary.""" + d = { + "name": keeper_file.name, + "title": keeper_file.title, + "type": keeper_file.type, + "last_modified": keeper_file.last_modified, + "size": keeper_file.size, + "f": keeper_file.f, + "file_key": keeper_file.file_key, + "meta_dict": keeper_file.meta_dict, + } + d["record_key_bytes"] = base64.b64encode(keeper_file.record_key_bytes).decode("ascii") \ + if keeper_file.record_key_bytes is not None else None + d["file_data"] = base64.b64encode(keeper_file.file_data).decode("ascii") \ + if keeper_file.file_data is not None else None + return d + + @staticmethod + def _file_from_dict(d): + """Reconstruct a KeeperFile instance from a JSON-deserialized dictionary.""" + f = object.__new__(_KeeperFile) + f.name = d["name"] + f.title = d["title"] + f.type = d["type"] + f.last_modified = d["last_modified"] + f.size = d["size"] + f.f = d["f"] + f.file_key = d["file_key"] + f.meta_dict = d["meta_dict"] + f.record_key_bytes = base64.b64decode(d["record_key_bytes"]) \ + if d["record_key_bytes"] is not None else None + f.file_data = base64.b64decode(d["file_data"]) \ + if d["file_data"] is not None else None + return f + + @staticmethod + def _record_to_dict(record): + """Serialize a Record instance to a JSON-safe dictionary.""" + d = { + "uid": record.uid, + "title": record.title, + "type": record.type, + "raw_json": record.raw_json, + "dict": record.dict, + "password": record.password, + "revision": record.revision, + "is_editable": record.is_editable, + "folder_uid": record.folder_uid, + "inner_folder_uid": record.inner_folder_uid, + "links": record.links, + "files": [KeeperAnsible._file_to_dict(f) for f in record.files], + } + d["record_key_bytes"] = base64.b64encode(record.record_key_bytes).decode("ascii") \ + if record.record_key_bytes is not None else None + return d + + @staticmethod + def _record_from_dict(d): + """Reconstruct a Record instance from a JSON-deserialized dictionary.""" + r = object.__new__(_Record) + r.uid = d["uid"] + r.title = d["title"] + r.type = d["type"] + r.raw_json = d["raw_json"] + r.dict = d["dict"] + r.password = d["password"] + r.revision = d["revision"] + r.is_editable = d["is_editable"] + r.folder_uid = d["folder_uid"] + r.inner_folder_uid = d["inner_folder_uid"] + r.links = d["links"] + r.record_key_bytes = base64.b64decode(d["record_key_bytes"]) \ + if d["record_key_bytes"] is not None else None + r.files = [KeeperAnsible._file_from_dict(fd) for fd in d["files"]] + return r + def encrypt(self, data): secret_key = self.get_encryption_key() - record_fh = io.BytesIO() - pickle.dump(data, record_fh) - return Fernet(secret_key).encrypt(record_fh.getvalue()) + serializable = [KeeperAnsible._record_to_dict(r) for r in data] + json_bytes = json.dumps(serializable).encode("utf-8") + return Fernet(secret_key).encrypt(json_bytes) def decrypt(self, ciphertext): secret_key = self.get_encryption_key() - return pickle.loads(Fernet(secret_key).decrypt(ciphertext)) + try: + plaintext = Fernet(secret_key).decrypt(ciphertext) + except Exception as err: + raise CacheUnusableError( + "Unable to decrypt the record cache. Check keeper_record_cache_secret " + "or regenerate the cache with keeper_cache_records." + ) from err + + # Pickle protocol markers (e.g. 0x80) -- never call pickle.loads (CWE-502 / VM-1452). + if plaintext.startswith(b"\x80"): + raise CacheUnusableError( + "Unable to deserialize the record cache. The cache may be from an older " + "plugin version or is invalid. Regenerate the cache with keeper_cache_records." + ) + + try: + payload = json.loads(plaintext.decode("utf-8")) + except (UnicodeDecodeError, json.JSONDecodeError, TypeError, ValueError) as err: + raise CacheUnusableError( + "Unable to deserialize the record cache. The cache may be from an older " + "plugin version or is invalid. Regenerate the cache with keeper_cache_records." + ) from err + + if not isinstance(payload, list): + raise CacheUnusableError( + "Unable to deserialize the record cache. Expected a list of records. " + "Regenerate the cache with keeper_cache_records." + ) + + try: + return [KeeperAnsible._record_from_dict(d) for d in payload] + except (KeyError, TypeError, ValueError) as err: + raise CacheUnusableError(str(err)) from err @staticmethod def convert_records_into_dict(records): @@ -460,7 +572,15 @@ def get_records_from_cache(self, cache, uids=None, titles=None): def get_records(self, uids=None, titles=None, cache=None, encrypt=False): if cache is not None: - records = self.get_records_from_cache(cache, uids=uids, titles=titles) + try: + records = self.get_records_from_cache(cache, uids=uids, titles=titles) + except CacheUnusableError: + # Invalidate legacy/invalid cache and start from scratch via the vault. + display.warning( + "Keeper record cache is unusable (legacy or invalid format) and was ignored. " + "Fetching records from the vault. Regenerate the cache with keeper_cache_records." + ) + records = self.get_records_from_vault(uids=uids, titles=titles, encrypt=encrypt) else: records = self.get_records_from_vault(uids=uids, titles=titles, encrypt=encrypt) @@ -477,14 +597,14 @@ def get_record(self, uids=None, titles=None, cache=None): return records[0] - def create_record(self, new_record, shared_folder_uid): + def create_record(self, new_record, shared_folder_uid, folder_uid=None): # KSM-816: use create_secret_with_options() instead of create_secret() so # that folder keys are fetched via the get_folders endpoint, which returns # all folders including empty ones. create_secret() uses get_secrets() which # only returns folder keys when the folder already contains records. try: record_uid = self.client.create_secret_with_options( - CreateOptions(shared_folder_uid, None), new_record + CreateOptions(shared_folder_uid, folder_uid), new_record ) except Exception as err: raise Exception("Cannot get create record: {}".format(err)) diff --git a/integration/keeper_secrets_manager_ansible/keeper_secrets_manager_ansible/plugins/action/keeper_create.py b/integration/keeper_secrets_manager_ansible/keeper_secrets_manager_ansible/plugins/action/keeper_create.py index 1444c89b5..3f809b13c 100644 --- a/integration/keeper_secrets_manager_ansible/keeper_secrets_manager_ansible/plugins/action/keeper_create.py +++ b/integration/keeper_secrets_manager_ansible/keeper_secrets_manager_ansible/plugins/action/keeper_create.py @@ -37,9 +37,16 @@ shared_folder_uid: description: - The UID of the top-level shared folder in your Keeper application. - - Must be a shared folder UID, not a subfolder UID. + - To create in a subfolder, also provide C(folder_uid). type: str required: yes + folder_uid: + description: + - The UID of a subfolder within the shared folder where the record should be created. + - When omitted, the record is created at the shared folder root. + - The subfolder must already exist and be accessible to the KSM application. + type: str + required: no record_type: description: - The type if record to create. @@ -204,9 +211,9 @@ ''' EXAMPLES = r''' -- name: Create a new record +- name: Create a record in a shared folder keeper_create: - share_folder_uid: XXX + shared_folder_uid: SHARED_FOLDER_UID record_type: login title: My Title notes: This record was created from Ansible @@ -221,6 +228,18 @@ label: Custom Field value: This is a value is a custom field. register: my_new_record + +- name: Create a record in a subfolder + keeper_create: + shared_folder_uid: SHARED_FOLDER_UID + folder_uid: SUBFOLDER_UID + record_type: login + title: My Subfolder Record + generate_password: True + fields: + - type: login + value: jane.doe@nowhere.com + register: my_subfolder_record ''' RETURN = r''' @@ -245,6 +264,7 @@ def run(self, tmp=None, task_vars=None): shared_folder_uid = self._task.args.get("shared_folder_uid") if shared_folder_uid is None: raise AnsibleError("The shared_folder_uid is blank. keeper_create requires this value to be set.") + folder_uid = self._task.args.get("folder_uid") record_type = self._task.args.get("record_type") if record_type is None: raise AnsibleError("The record_type is blank. keeper_create requires this value to be set.") @@ -272,20 +292,24 @@ def run(self, tmp=None, task_vars=None): try: for field in self._task.args.get("fields", []): + # Workaround: convert value: [] to None so helper FieldType.__init__ skips the + # dict-field index (value[0]) that crashes on empty lists. Remove once helper + # ships the "if self.value:" guard in FieldType.__init__. fields.append(Field( field_section=FieldSectionEnum.STANDARD, type=field.get("type"), label=field.get("label"), - value=field.get("value") + value=field.get("value") or None )) keeper.stash_secret_value(str(field.get("value"))) for field in self._task.args.get("custom_fields", []): + # Same workaround as above for custom fields. fields.append(Field( field_section=FieldSectionEnum.CUSTOM, type=field.get("type"), label=field.get("label"), - value=field.get("value", "text") + value=field.get("value", "text") or None )) keeper.stash_secret_value(str(field.get("value"))) @@ -313,7 +337,8 @@ def run(self, tmp=None, task_vars=None): password_complexity=password_complexity ) record_create = record[0].get_record_create_obj() - record_uid = keeper.create_record(record_create, shared_folder_uid=shared_folder_uid) + record_uid = keeper.create_record(record_create, shared_folder_uid=shared_folder_uid, + folder_uid=folder_uid) except Exception as err: raise AnsibleError("Could not create record: {}".format(err)) diff --git a/integration/keeper_secrets_manager_ansible/keeper_secrets_manager_ansible/plugins/modules/keeper_create.py b/integration/keeper_secrets_manager_ansible/keeper_secrets_manager_ansible/plugins/modules/keeper_create.py index 967209953..0373446a6 100644 --- a/integration/keeper_secrets_manager_ansible/keeper_secrets_manager_ansible/plugins/modules/keeper_create.py +++ b/integration/keeper_secrets_manager_ansible/keeper_secrets_manager_ansible/plugins/modules/keeper_create.py @@ -26,9 +26,16 @@ shared_folder_uid: description: - The UID of the top-level shared folder in your Keeper application. - - Must be a shared folder UID, not a subfolder UID. + - To create in a subfolder, also provide C(folder_uid). type: str required: yes + folder_uid: + description: + - The UID of a subfolder within the shared folder where the record should be created. + - When omitted, the record is created at the shared folder root. + - The subfolder must already exist and be accessible to the KSM application. + type: str + required: no record_type: description: - The type if record to create. @@ -182,9 +189,9 @@ ''' EXAMPLES = r''' -- name: Create a new record +- name: Create a record in a shared folder keeper_create: - share_folder_uid: XXX + shared_folder_uid: SHARED_FOLDER_UID record_type: login title: My Title notes: This record was created from Ansible @@ -199,6 +206,18 @@ label: Custom Field value: This is a value is a custom field. register: my_new_record + +- name: Create a record in a subfolder + keeper_create: + shared_folder_uid: SHARED_FOLDER_UID + folder_uid: SUBFOLDER_UID + record_type: login + title: My Subfolder Record + generate_password: True + fields: + - type: login + value: jane.doe@nowhere.com + register: my_subfolder_record ''' RETURN = r''' diff --git a/integration/keeper_secrets_manager_ansible/requirements.txt b/integration/keeper_secrets_manager_ansible/requirements.txt index fee550f07..db1fa4a38 100644 --- a/integration/keeper_secrets_manager_ansible/requirements.txt +++ b/integration/keeper_secrets_manager_ansible/requirements.txt @@ -1,3 +1,3 @@ ansible-core>=2.12.0 -keeper-secrets-manager-core>=17.2.0 -keeper-secrets-manager-helper>=1.1.0 +keeper-secrets-manager-core>=17.3.0 +keeper-secrets-manager-helper>=1.1.2 diff --git a/integration/keeper_secrets_manager_ansible/setup.py b/integration/keeper_secrets_manager_ansible/setup.py index 6c5f6da9b..927b29bb4 100644 --- a/integration/keeper_secrets_manager_ansible/setup.py +++ b/integration/keeper_secrets_manager_ansible/setup.py @@ -9,14 +9,14 @@ long_description = fp.read() install_requires = [ - 'keeper-secrets-manager-core>=17.2.0', - 'keeper-secrets-manager-helper>=1.1.0', + 'keeper-secrets-manager-core>=17.3.0', + 'keeper-secrets-manager-helper>=1.1.2', 'ansible-core>=2.12.0' # Use ansible-core instead of ansible to avoid community collections ] setup( name="keeper-secrets-manager-ansible", - version='1.4.0', + version='1.5.0', description="Keeper Secrets Manager plugins for Ansible.", long_description=long_description, long_description_content_type="text/markdown", diff --git a/integration/keeper_secrets_manager_ansible/tests/ansible_example/playbooks/keeper_create_subfolder.yml b/integration/keeper_secrets_manager_ansible/tests/ansible_example/playbooks/keeper_create_subfolder.yml new file mode 100644 index 000000000..2da2fa4ba --- /dev/null +++ b/integration/keeper_secrets_manager_ansible/tests/ansible_example/playbooks/keeper_create_subfolder.yml @@ -0,0 +1,23 @@ +# vim: set shiftwidth=2 tabstop=2 softtabstop=-1 expandtab: +--- +- name: Keeper Create Subfolder + hosts: "my_systems" + gather_facts: no + + tasks: + - name: "Create Record in Subfolder" + keeper_create: + shared_folder_uid: "{{ shared_folder_uid }}" + folder_uid: "{{ folder_uid | default(omit) }}" + record_type: login + generate_password: True + title: "My Subfolder Record" + fields: + - type: login + value: "johndoe@localhost" + register: "new_record" + + - name: "Print Record UID" + debug: + var: new_record.record_uid + verbosity: 0 diff --git a/integration/keeper_secrets_manager_ansible/tests/keeper_cache_encrypt_test.py b/integration/keeper_secrets_manager_ansible/tests/keeper_cache_encrypt_test.py new file mode 100644 index 000000000..b892d5432 --- /dev/null +++ b/integration/keeper_secrets_manager_ansible/tests/keeper_cache_encrypt_test.py @@ -0,0 +1,198 @@ +# -*- coding: utf-8 -*- +"""Unit tests for JSON-based record cache encrypt/decrypt (VM-1452 / CWE-502).""" + +import base64 +import io +import os +import pickle +import socket +import sys +import tempfile +import unittest +from unittest.mock import MagicMock, patch + +# ansible-core imports fcntl (unavailable on Windows). Stub ansible for unit tests +# that only exercise encrypt/decrypt crypto, not the full plugin runtime. +try: + import fcntl # noqa: F401 +except ImportError: + class _FakeAnsibleError(Exception): + pass + + _ansible_stub = MagicMock() + _ansible_stub.AnsibleError = _FakeAnsibleError + _ansible_stub.errors.AnsibleError = _FakeAnsibleError + sys.modules.setdefault("ansible", _ansible_stub) + sys.modules.setdefault("ansible.utils", _ansible_stub) + sys.modules.setdefault("ansible.utils.display", _ansible_stub) + sys.modules.setdefault("ansible.errors", _ansible_stub) + sys.modules.setdefault("ansible.module_utils", _ansible_stub) + sys.modules.setdefault("ansible.module_utils.basic", _ansible_stub) + sys.modules.setdefault("ansible.module_utils.common", _ansible_stub) + sys.modules.setdefault("ansible.module_utils.common.text", _ansible_stub) + sys.modules.setdefault("ansible.module_utils.common.text.converters", _ansible_stub) + _ansible_stub.module_utils.basic.missing_required_lib = lambda name: name + _ansible_stub.module_utils.common.text.converters.jsonify = lambda x: str(x) + _ansible_stub.utils.display.Display = MagicMock + +from cryptography.fernet import Fernet +from cryptography.hazmat.primitives import hashes +from cryptography.hazmat.primitives.kdf.pbkdf2 import PBKDF2HMAC +from keeper_secrets_manager_core.dto.dtos import Record + +from keeper_secrets_manager_ansible import CacheUnusableError, KeeperAnsible + + +def _make_record(uid="uid123", title="Test Record", password="secret-pass"): + """Build a minimal Record suitable for cache serialize/deserialize.""" + record = Record.__new__(Record) + record.uid = uid + record.title = title + record.type = "login" + record.dict = { + "title": title, + "type": "login", + "fields": [ + {"type": "login", "value": ["user1"]}, + {"type": "password", "value": [password]}, + ], + "custom": [], + } + record.raw_json = None + record.record_key_bytes = os.urandom(32) + record.folder_uid = "" + record.inner_folder_uid = "" + record.revision = 1 + record.is_editable = True + record.password = password + record.links = [] + record.files = [] + return record + + +def _stub_keeper(cache_secret="unit-test-cache-secret"): + """Return a KeeperAnsible instance with __init__ bypassed.""" + with patch.object(KeeperAnsible, "__init__", lambda self, *a, **k: None): + keeper = KeeperAnsible.__new__(KeeperAnsible) + keeper.task_vars = {"keeper_record_cache_secret": cache_secret} + keeper.client = MagicMock() + keeper.action_module = None + return keeper + + +class KeeperCacheEncryptTest(unittest.TestCase): + + def test_encrypt_decrypt_round_trip(self): + keeper = _stub_keeper() + original = _make_record() + + ciphertext = keeper.encrypt([original]) + restored = keeper.decrypt(ciphertext) + + self.assertEqual(len(restored), 1) + self.assertEqual(restored[0].uid, original.uid) + self.assertEqual(restored[0].title, original.title) + self.assertEqual(restored[0].type, original.type) + self.assertEqual(restored[0].dict, original.dict) + self.assertEqual(restored[0].record_key_bytes, original.record_key_bytes) + self.assertEqual(restored[0].field("password"), ["secret-pass"]) + self.assertEqual(restored[0].field("login"), ["user1"]) + + def test_decrypt_rejects_pickle_payload(self): + """Encrypted pickle must not execute; decrypt must fail safely (VM-1452).""" + cache_secret = "attacker-controlled-secret-12345" + proof_path = None + + with tempfile.NamedTemporaryFile(delete=False) as tmp: + proof_path = tmp.name + try: + if os.path.exists(proof_path): + os.remove(proof_path) + + class Exploit: + def __reduce__(self): + return (open, (proof_path, "w")) + + hostname = socket.gethostname() + salt = hostname.zfill(32)[0:32] + kdf = PBKDF2HMAC( + algorithm=hashes.SHA256(), + length=32, + salt=salt.encode(), + iterations=390000, + ) + key = base64.urlsafe_b64encode(kdf.derive(cache_secret.encode())) + buf = io.BytesIO() + pickle.dump(Exploit(), buf) + malicious = Fernet(key).encrypt(buf.getvalue()) + + keeper = _stub_keeper(cache_secret=cache_secret) + with self.assertRaises(CacheUnusableError): + keeper.decrypt(malicious) + + self.assertFalse( + os.path.exists(proof_path), + "pickle payload must not execute during decrypt", + ) + finally: + if proof_path and os.path.exists(proof_path): + os.remove(proof_path) + + def test_decrypt_rejects_invalid_json_shape(self): + keeper = _stub_keeper() + secret_key = keeper.get_encryption_key() + bad = Fernet(secret_key).encrypt(b'{"not": "a list"}') + with self.assertRaises(CacheUnusableError) as ctx: + keeper.decrypt(bad) + self.assertIn("keeper_cache_records", str(ctx.exception)) + + def test_get_records_falls_back_to_vault_on_legacy_cache(self): + """Unusable cache is ignored; records are fetched from the vault.""" + keeper = _stub_keeper() + vault_record = _make_record(uid="from-vault", title="Vault Record") + keeper.client.get_secrets.return_value = [vault_record] + + secret_key = keeper.get_encryption_key() + legacy_cache = Fernet(secret_key).encrypt(b"\x80\x04legacy-pickle-bytes") + + with patch("keeper_secrets_manager_ansible.display.warning") as mock_warn: + records = keeper.get_records(uids=["from-vault"], cache=legacy_cache) + + self.assertEqual(len(records), 1) + self.assertEqual(records[0].uid, "from-vault") + keeper.client.get_secrets.assert_called_once_with(["from-vault"]) + mock_warn.assert_called_once() + self.assertIn("ignored", mock_warn.call_args[0][0]) + + def test_get_records_falls_back_to_vault_on_invalid_json_cache(self): + keeper = _stub_keeper() + vault_record = _make_record(uid="from-vault") + keeper.client.get_secrets.return_value = [vault_record] + + secret_key = keeper.get_encryption_key() + bad_cache = Fernet(secret_key).encrypt(b'{"not": "a list"}') + + records = keeper.get_records(uids=["from-vault"], cache=bad_cache) + self.assertEqual(records[0].uid, "from-vault") + keeper.client.get_secrets.assert_called_once_with(["from-vault"]) + + def test_get_records_legacy_cache_vault_miss_still_fails(self): + """After cache invalidate, missing vault records still raise.""" + keeper = _stub_keeper() + keeper.client.get_secrets.return_value = [] + + secret_key = keeper.get_encryption_key() + legacy_cache = Fernet(secret_key).encrypt(b"\x80\x04legacy") + + with self.assertRaises(Exception) as ctx: + keeper.get_records(uids=["missing-uid"], cache=legacy_cache) + + # Prefer args over str()/ .message — newer ansible-core AnsibleError.__str__ + # can recurse when formatting the exception message. + detail = ctx.exception.args[0] if ctx.exception.args else "" + self.assertIn("missing-uid", detail) + keeper.client.get_secrets.assert_called_once_with(["missing-uid"]) + + +if __name__ == "__main__": + unittest.main() diff --git a/integration/keeper_secrets_manager_ansible/tests/keeper_create_subfolder_test.py b/integration/keeper_secrets_manager_ansible/tests/keeper_create_subfolder_test.py new file mode 100644 index 000000000..7a1193885 --- /dev/null +++ b/integration/keeper_secrets_manager_ansible/tests/keeper_create_subfolder_test.py @@ -0,0 +1,104 @@ +import json +import os +import tempfile +import unittest +from unittest.mock import MagicMock, patch +from keeper_secrets_manager_core.dto.payload import CreateOptions +from keeper_secrets_manager_core.mock import Record, Response +from keeper_secrets_manager_ansible import KeeperAnsible +from .ansible_test_framework import AnsibleTestFramework + + +class KeeperCreateSubfolderTest(unittest.TestCase): + """ + Unit tests for KeeperAnsible.create_record() subfolder support (KSM-845). + + Tests that folder_uid flows through to CreateOptions.subfolder_uid, + and that omitting folder_uid preserves backward-compatible None behavior. + """ + + def _make_keeper(self): + mock_client = MagicMock() + mock_client.create_secret_with_options.return_value = "NEW_UID" + keeper = object.__new__(KeeperAnsible) + keeper.client = mock_client + return keeper, mock_client + + def test_folder_uid_passed_as_subfolder_uid(self): + keeper, mock_client = self._make_keeper() + keeper.create_record(MagicMock(), "SHARED_UID", folder_uid="SUB_UID") + create_options = mock_client.create_secret_with_options.call_args[0][0] + self.assertIsInstance(create_options, CreateOptions) + self.assertEqual(create_options.folder_uid, "SHARED_UID") + self.assertEqual(create_options.subfolder_uid, "SUB_UID") + + def test_no_folder_uid_defaults_to_none(self): + keeper, mock_client = self._make_keeper() + keeper.create_record(MagicMock(), "SHARED_UID") + create_options = mock_client.create_secret_with_options.call_args[0][0] + self.assertIsInstance(create_options, CreateOptions) + self.assertEqual(create_options.folder_uid, "SHARED_UID") + self.assertIsNone(create_options.subfolder_uid) + + +class KeeperCreateSubfolderPlaybookTest(unittest.TestCase): + """ + Integration test for the keeper_create_subfolder.yml example playbook (KSM-845). + + Unlike KeeperCreateSubfolderTest above, this runs the actual playbook YAML + through ansible-playbook (via AnsibleTestFramework), the same way every other + example playbook in tests/ansible_example/playbooks/ is verified elsewhere in + this suite. It catches breakage in the YAML/Jinja/action-plugin wiring that a + pure Python unit test on create_record() cannot see. + + Ansible executes each task's action plugin in a forked worker process, so a + mocked side_effect cannot report back to the test via an in-memory closure + (the fork gets its own copy of that state). It writes what it saw to a temp + file instead, which is visible to the parent process after the fork exits. + """ + + def test_keeper_create_subfolder_playbook(self): + mock_response = Response() + mock_record = Record(title="Record 1", record_type="login") + mock_record.field("password", "MYPASSWORD") + mock_response.add_record(record=mock_record) + + with tempfile.NamedTemporaryFile(delete=False) as tmp: + capture_path = tmp.name + os.remove(capture_path) + + def mocked_create_secret(*args): + create_options = args[0] + with open(capture_path, "w") as fh: + json.dump({ + "folder_uid": create_options.folder_uid, + "subfolder_uid": create_options.subfolder_uid, + }, fh) + return "NEW_UID" + + try: + with patch( + "keeper_secrets_manager_core.core.SecretsManager.create_secret_with_options", + side_effect=mocked_create_secret, + ): + a = AnsibleTestFramework( + playbook="keeper_create_subfolder.yml", + vars={ + "shared_folder_uid": "SHARED_UID", + "folder_uid": "SUB_UID", + }, + mock_responses=[mock_response] + ) + result, out, err = a.run() + + self.assertEqual(result["ok"], 2, "2 things didn't happen") + self.assertEqual(result["failed"], 0, "failed was not 0") + + self.assertTrue(os.path.exists(capture_path), "create_secret_with_options was never called") + with open(capture_path) as fh: + captured = json.load(fh) + self.assertEqual(captured.get("folder_uid"), "SHARED_UID") + self.assertEqual(captured.get("subfolder_uid"), "SUB_UID") + finally: + if os.path.exists(capture_path): + os.remove(capture_path)