mirror of
https://github.com/kevinveenbirkenbach/computer-playbook.git
synced 2026-11-06 02:54:26 +00:00
kevinveenbirkenbachandClaude Opus 5
7e1d336869
e41be38cb replaced the ansible-vault subprocess with in-process VaultLib and imported ansible at module level. That pulled it into the chain cli.administration.deploy.development -> credentials_generator -> utils/manager/inventory.py -> utils/handler/vault.py, which the CI runner host resolves without ansible installed. Before that commit the module imported only subprocess and yaml, so the import cost nothing.
test_ci_host_dependencies names both remedies: declare the dependency in [project.dependencies], or make the import lazy. Lazy is right here because ansible serves one branch of this module and the declaration would be paid for on every CI job. _vault is already a cached_property, so the import runs once per handler instance and only when something actually encrypts.
Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
72 lines
2.5 KiB
Python
72 lines
2.5 KiB
Python
from __future__ import annotations
|
|
|
|
from functools import cached_property
|
|
from typing import TYPE_CHECKING, Any
|
|
|
|
import yaml
|
|
from yaml.dumper import SafeDumper
|
|
from yaml.loader import SafeLoader
|
|
|
|
if TYPE_CHECKING:
|
|
from ansible.parsing.vault import VaultLib
|
|
|
|
|
|
class VaultScalar(str):
|
|
"""A subclass of str to represent vault-encrypted strings."""
|
|
|
|
__slots__ = ()
|
|
|
|
|
|
def _vault_constructor(loader, node):
|
|
"""Load a !vault block as a VaultScalar so the tag survives a round-trip."""
|
|
return VaultScalar(node.value)
|
|
|
|
|
|
def _vault_representer(dumper, data):
|
|
"""Custom representer to dump VaultScalar as literal blocks."""
|
|
return dumper.represent_scalar("!vault", data, style="|")
|
|
|
|
|
|
SafeLoader.add_constructor("!vault", _vault_constructor)
|
|
getattr(yaml, "CSafeLoader", SafeLoader).add_constructor("!vault", _vault_constructor)
|
|
SafeDumper.add_representer(VaultScalar, _vault_representer)
|
|
|
|
|
|
class VaultHandler:
|
|
def __init__(self, vault_password_file: str):
|
|
self.vault_password_file = vault_password_file
|
|
|
|
@cached_property
|
|
def _vault(self) -> VaultLib:
|
|
from ansible.parsing.dataloader import DataLoader
|
|
from ansible.parsing.vault import FileVaultSecret, VaultLib
|
|
|
|
secret = FileVaultSecret(filename=self.vault_password_file, loader=DataLoader())
|
|
secret.load()
|
|
return VaultLib([("default", secret)])
|
|
|
|
def encrypt_string(self, value: str, name: str) -> str:
|
|
"""Return the ``name: !vault |`` snippet as ``ansible-vault`` lays it out.
|
|
|
|
Args:
|
|
value: plaintext to encrypt.
|
|
name: key the snippet is emitted under.
|
|
"""
|
|
body = self._vault.encrypt(value).decode()
|
|
indented = "\n".join(f" {line}" for line in body.splitlines())
|
|
return f"{name}: !vault |\n{indented}\n"
|
|
|
|
def encrypt_leaves(self, branch: dict[str, Any], vault_pw: str):
|
|
"""Recursively encrypt all leaves (plain text values) under the credentials section."""
|
|
for key, value in branch.items():
|
|
if isinstance(value, dict):
|
|
self.encrypt_leaves(value, vault_pw)
|
|
elif isinstance(value, str) and not value.lstrip().startswith(
|
|
"$ANSIBLE_VAULT"
|
|
):
|
|
snippet = self.encrypt_string(value, key)
|
|
lines = snippet.splitlines()
|
|
indent = len(lines[1]) - len(lines[1].lstrip())
|
|
body = "\n".join(line[indent:] for line in lines[1:])
|
|
branch[key] = VaultScalar(body)
|