Files
devx/src/devx/utils/vault.py
T

136 lines
4.0 KiB
Python

"""Ansible Vault helpers for encrypting and decrypting YAML files.
Wraps ``ansible-vault`` to provide a convenient API for loading and
saving vault-encrypted YAML files. Falls back to plain YAML when no
vault-password file is available, making it safe to use in both
local (with vault) and CI (without vault) environments.
Usage::
from devx.utils.vault import load_vault_yaml, save_vault_yaml
data = load_vault_yaml(Path("secrets.yml"), vault_pass=Path("vault-password"))
data["new_key"] = "value"
save_vault_yaml(Path("secrets.yml"), data, vault_pass=Path("vault-password"))
"""
from __future__ import annotations
import subprocess # nosec B404
from pathlib import Path
import yaml
def encrypt_file(path: Path, vault_pass: Path) -> None:
"""Encrypt a file in-place using ansible-vault.
Args:
path: File to encrypt.
vault_pass: Path to the vault-password file.
"""
subprocess.run( # nosec B603, B607
[
"ansible-vault",
"encrypt",
str(path),
"--vault-password-file",
str(vault_pass),
"--encrypt-vault-id",
"default",
],
check=True,
)
def decrypt_file(path: Path, vault_pass: Path) -> None:
"""Decrypt a file in-place using ansible-vault.
Args:
path: File to decrypt.
vault_pass: Path to the vault-password file.
"""
subprocess.run( # nosec B603, B607
[
"ansible-vault",
"decrypt",
str(path),
"--vault-password-file",
str(vault_pass),
],
check=True,
)
def load_vault_yaml(path: Path, vault_pass: Path | None = None) -> dict:
"""Load a YAML file, decrypting with ansible-vault if vault-password exists.
If *vault_pass* is None or doesn't exist, the file is read as plain
YAML. If decryption fails (file not vault-encrypted), it falls back
to plain YAML.
Args:
path: YAML file path.
vault_pass: Path to the vault-password file (optional).
Returns:
Parsed YAML content as a dict (empty dict if file is empty).
"""
if vault_pass is None or not vault_pass.exists():
with open(path, encoding="utf-8") as f:
return yaml.safe_load(f) or {}
result = subprocess.run( # nosec B603, B607
["ansible-vault", "view", str(path), "--vault-password-file", str(vault_pass)],
capture_output=True,
text=True,
check=False,
)
if result.returncode == 0:
return yaml.safe_load(result.stdout) or {}
if "is not vault encrypted" in result.stderr:
with open(path, encoding="utf-8") as f:
return yaml.safe_load(f) or {}
result.check_returncode() # pragma: no cover
return {} # pragma: no cover
def save_vault_yaml(path: Path, data: dict, vault_pass: Path | None = None) -> None:
"""Write YAML data, encrypting with ansible-vault if vault-password exists.
Args:
path: Destination YAML file path.
data: Data to serialize.
vault_pass: Path to the vault-password file (optional).
"""
plain = yaml.dump(data, default_flow_style=False, sort_keys=False)
with open(path, "w", encoding="utf-8") as f:
f.write(plain)
if vault_pass is not None and vault_pass.exists():
subprocess.run( # nosec B603, B607
[
"ansible-vault",
"encrypt",
str(path),
"--vault-password-file",
str(vault_pass),
"--encrypt-vault-id",
"default",
],
capture_output=True,
check=True,
)
def is_encrypted(path: Path) -> bool:
"""Check if a file is ansible-vault encrypted.
Args:
path: File to check.
Returns:
True if the file starts with the ``$ANSIBLE_VAULT`` marker.
"""
with open(path, encoding="utf-8") as f:
first_line = f.readline()
return "$ANSIBLE_VAULT" in first_line