mirror of
https://github.com/kevinveenbirkenbach/computer-playbook.git
synced 2026-11-06 02:54:26 +00:00
kevinveenbirkenbachandClaude Opus 4.7
aa23cba583
The credentials baker at cli/administration/inventory/credentials/__main__.py
walked `schema_creds.items()` one level deep and called `str(value)` on
every leaf. For nested credential groups like
``credentials.recaptcha = {key, secret}`` (declared in
roles/web-app-espocrm/meta/schema.yml and
roles/web-app-listmonk/meta/schema.yml) this collapsed the inner dict
into its Python repr, vault-encrypted that blob, and wrote it as a
single scalar at ``credentials.recaptcha``. The runtime ``config``
lookup then choked on the descent into ``recaptcha.key``:
The lookup plugin 'config' failed: Expected dict for 'key', got
_AnsibleTaggedStr at 'key'
Full path so far: applications['web-app-espocrm'].credentials.recaptcha.key
Current object: "{'key': '{{ CAPTCHA.RECAPTCHA.KEY }}', 'secret':
'{{ CAPTCHA.RECAPTCHA.SECRET }}'}"
(jobs 77797371397 espocrm and 77797371442 listmonk on run 26428080957.)
Introduce ``emit_credentials`` (new module: ``credentials/emit.py``)
that walks the schema recursively. Scalar leaves still vault-encrypt
via ``to_vault_block``; dict nodes recurse into a nested CommentedMap
via ``ensure_map``. The walker also tracks dotted full-key paths so
the existing ``--force``-mode override-apply logic sees the same dotted
keys it always saw.
Auto-healing for previously-corrupted host_vars: ``ensure_map``
overwrites non-CommentedMap placeholders, so the next bake replaces a
stringified ``credentials.recaptcha`` blob with the correct nested
structure. Operators with corrupted entries simply re-run the
credentials generator; no manual cleanup needed.
Package refactor (KISS/DRY/SPOT):
The 450+ LoC ``__main__.py`` was carrying four orthogonal concerns
(vault, --set parsing, recursive emission, prompts) plus orchestration.
Split by concern:
- ``credentials/vault.py``: ``is_vault_encrypted``,
``is_ruamel_vault``, ``to_vault_block`` (+ private helpers
``_vault_body``, ``_make_vault_scalar_from_text``).
- ``credentials/overrides.py``: ``parse_overrides``, ``override_for``.
- ``credentials/emit.py``: ``ensure_map``, ``emit_credentials``.
- ``credentials/prompts.py``: ``ask_for_confirmation``.
- ``credentials/__main__.py``: argparse + ``main`` + two thin mode
handlers (``_run_snippet_mode``, ``_run_default_mode``) + the
force-override helper.
``credentials/__init__.py`` re-exports the public API
(``main``, ``ask_for_confirmation``, ``parse_overrides``, etc.) so
external callers (notably ``tests/unit/cli/administration/inventory/
test_credentials.py``) keep working unchanged.
Test layout follows the split:
- ``tests/unit/cli/administration/inventory/credentials/test_vault.py``
(20 tests: vault detection, body extraction, scalar construction,
to_vault_block paths, plus the VaultHandler subprocess coverage).
- ``tests/unit/cli/administration/inventory/credentials/test_overrides.py``
(13 tests: parse_overrides + override_for incl. dotted nested keys).
- ``tests/unit/cli/administration/inventory/credentials/test_emit.py``
(10 tests: ensure_map placeholder replacement, recursive emission,
none → empty, skip_existing semantics, override precedence,
dotted-path tracking, the on-disk corruption recovery case).
- ``tests/unit/cli/administration/inventory/credentials/test_prompts.py``
(5 tests: ask_for_confirmation yes/no/garbage/case/whitespace).
``tests/unit/cli/administration/inventory/test_credentials.py`` keeps
only the end-to-end ``main()`` integration coverage (including
``test_nested_credentials_dict_recurses_instead_of_stringifying`` —
the explicit regression guard for the original failure mode).
Coverage: 47 new per-module unit tests + the 5 existing main()
integration tests, all green. ``make test`` green across all five
targets.
Co-Authored-By: Claude Opus 4.7 <noreply@anthropic.com>
65 lines
2.2 KiB
Python
65 lines
2.2 KiB
Python
"""Vault block construction and detection helpers."""
|
|
|
|
from __future__ import annotations
|
|
|
|
from typing import Any
|
|
|
|
from ruamel.yaml import YAML
|
|
|
|
from utils.handler.vault import VaultHandler, VaultScalar
|
|
|
|
|
|
def is_ruamel_vault(val: Any) -> bool:
|
|
"""Detect if a ruamel scalar already carries the !vault tag."""
|
|
try:
|
|
return getattr(val, "tag", None) == "!vault"
|
|
except Exception:
|
|
return False
|
|
|
|
|
|
def is_vault_encrypted(val: Any) -> bool:
|
|
"""Detect if a value is already a vault string, a ruamel !vault scalar,
|
|
or an internal VaultScalar from InventoryManager.apply_schema()."""
|
|
if isinstance(val, VaultScalar):
|
|
return True
|
|
if is_ruamel_vault(val):
|
|
return True
|
|
return bool(isinstance(val, str) and ("$ANSIBLE_VAULT" in val or "!vault" in val))
|
|
|
|
|
|
def _vault_body(text: str) -> str:
|
|
"""Return only the vault body starting from the first line containing
|
|
``$ANSIBLE_VAULT``. Strips any leading ``!vault |`` header if present."""
|
|
lines = text.splitlines()
|
|
for i, ln in enumerate(lines):
|
|
if "$ANSIBLE_VAULT" in ln:
|
|
return "\n".join(lines[i:])
|
|
return text
|
|
|
|
|
|
def _make_vault_scalar_from_text(text: str) -> Any:
|
|
"""Build a ruamel object representing a literal block scalar tagged
|
|
with !vault by parsing a tiny YAML snippet. This avoids depending on
|
|
yaml_set_tag()."""
|
|
body = _vault_body(text)
|
|
indented = " " + body.replace("\n", "\n ")
|
|
snippet = f"v: !vault |\n{indented}\n"
|
|
y = YAML(typ="rt")
|
|
return y.load(snippet)["v"]
|
|
|
|
|
|
def to_vault_block(vault_handler: VaultHandler, value: str | Any, label: str) -> Any:
|
|
"""Return a ruamel scalar tagged as !vault. Reuses existing vault
|
|
payloads (VaultScalar, ruamel !vault, ``$ANSIBLE_VAULT`` strings)
|
|
and encrypts plaintext otherwise. Empty strings stay plain."""
|
|
if isinstance(value, str) and value == "":
|
|
return ""
|
|
if is_ruamel_vault(value):
|
|
return value
|
|
if isinstance(value, VaultScalar):
|
|
return _make_vault_scalar_from_text(str(value))
|
|
if isinstance(value, str) and ("$ANSIBLE_VAULT" in value or "!vault" in value):
|
|
return _make_vault_scalar_from_text(value)
|
|
snippet = vault_handler.encrypt_string(str(value), label)
|
|
return _make_vault_scalar_from_text(snippet)
|