diff --git a/README.md b/README.md index 9f8e153..8bd502a 100644 --- a/README.md +++ b/README.md @@ -97,6 +97,58 @@ age-edit -e "${HOME}/.local/bin/codew" -t /tmp/ -M -a ~/.age/phg-age-dotfiles

` options before the +subcommand when rotating keys. + +```shell +age-docker init +age-docker check +age-docker list +age-docker edit secrets/prod.env.age +age-docker encrypt prod.env secrets/prod.env.age +age-docker decrypt secrets/prod.env.age prod.env +age-docker rekey secrets/prod.env.age +age-docker rekey --all +``` + +Manage public keys without changing groups or file policies: + +```shell +age-docker key add operator 'age1...' +age-docker key scan server server.example.com +age-docker key remove unused-server +``` + +`key scan` displays the retrieved SSH key and its SHA256 fingerprint before +asking for confirmation. `ssh-keyscan` does not authenticate the result; +compare the fingerprint through a trusted channel. Replacing an existing +alias requires `--replace`, and non-interactive confirmation requires +`--yes`. + ## Backup/Restore settings for macOS native user preferences See here for a defaults documentation: diff --git a/config.yaml b/config.yaml index 9934ef4..d7f8c8a 100644 --- a/config.yaml +++ b/config.yaml @@ -5,6 +5,7 @@ config: variables: ageidentity: "{{@@ env['HOME'] @@}}/.age/phg-age-dotfiles" ageidentity_pub: '{{@@ ageidentity@@}}.pub' + agedockersecretsidentity: "{{@@ env['HOME'] @@}}/.age/phg-age-docker-secrets" SHELL_ERR_MESSAGE: \033[41;30m SHELL_RESET_COLOR: \033[0m trans_install: @@ -149,6 +150,10 @@ dotfiles: f_agenix_helper: dst: ~/.local/bin/agenix-helper src: local/bin/agenix-helper + f_age_docker: + dst: ~/.local/bin/age-docker + src: local/bin/age-docker + chmod: '700' f_config: src: ssh/config dst: ~/.ssh/config @@ -175,6 +180,11 @@ dotfiles: dst: ~/.vwfunc trans_install: _decrypt trans_update: _encrypt + f_vws2-artifacts-config: + src: vws2-artifacts-config + dst: ~/.vws2-artifacts-config + trans_install: _decrypt + trans_update: _encrypt profiles: meta_base: variables: @@ -183,6 +193,7 @@ profiles: wsl: false dotfiles: - d_colors + - f_age_docker - f_agenix_helper - f_codew - f_commonfunc @@ -207,6 +218,7 @@ profiles: - f_rpki - f_vwrc - f_vwfunc + - f_vws2-artifacts-config meta_wsl: variables: wsl: true diff --git a/config/brew/Brewfile b/config/brew/Brewfile index dbd39d7..77497f8 100644 --- a/config/brew/Brewfile +++ b/config/brew/Brewfile @@ -167,6 +167,8 @@ cask "mist" cask "monitorcontrol" # Utility to move and zoom windows—on one display cask "moom" +# Read-only disk space visualiser +cask "neodisk" # Desktop sync client for Nextcloud software products cask "nextcloud" # Knowledge base that works on top of a local folder of plain text Markdown files diff --git a/dotfiles/commonfunc b/dotfiles/commonfunc index e8ee8d3..27bd5a5 100644 --- a/dotfiles/commonfunc +++ b/dotfiles/commonfunc @@ -101,6 +101,67 @@ fabric() { fabric-ai ${@:-} } +####################################### +# Run Claude Code through the local Codex gateway with an interactively selected model. +# Globals: +# CLAUDE_CODEX_BASE_URL +# Arguments: +# Arguments to pass to claude +# Outputs: +# Model picker and output from claude +# Returns: +# Exit code from fzf, the model picker, or claude +####################################### +claude-codex() { + local gateway_url="${CLAUDE_CODEX_BASE_URL:-http://localhost:18765}" + local model + local yolo_reply + local -a claude_args=("$@") + local -a models=() + local -a fallback_models=( + gpt-5.6-sol + gpt-5.6-terra + gpt-5.6-luna + ) + + if (( ${+commands[curl]} && ${+commands[jq]} )); then + models=(${(f)"$( + curl -fsS --max-time 2 "${gateway_url}/v1/models" 2>/dev/null | + jq -r '.data[] | select((.display_name // "") | endswith(" (codex)")) | .id' | + sort -u + )"}) + fi + + (( ${#models[@]} )) || models=("${fallback_models[@]}") + + if (( ${+commands[fzf]} )); then + model=$( + printf '%s\n' "${models[@]}" | + fzf --height=60% --reverse --prompt='Claude Codex model> ' + ) || return $? + else + local PS3='Claude Codex model: ' + select model in "${models[@]}"; do + [[ -n "${model}" ]] && break + print -u2 -- "Invalid selection." + done + fi + + if read -q "yolo_reply?Run Claude in YOLO mode? [y/N] "; then + claude_args+=(--dangerously-skip-permissions) + fi + print + + ANTHROPIC_BASE_URL="${gateway_url}" \ + ANTHROPIC_AUTH_TOKEN=unused \ + ANTHROPIC_MODEL="${model}[1m]" \ + ANTHROPIC_SMALL_FAST_MODEL=gpt-5.6-luna[1m] \ + CLAUDE_CODE_AUTO_COMPACT_WINDOW=272000 \ + CLAUDE_CODE_DISABLE_NONESSENTIAL_TRAFFIC=1 \ + CLAUDE_CODE_DISABLE_NONSTREAMING_FALLBACK=1 \ + command claude "${claude_args[@]}" +} + ####################################### # wttr - show the weather forecast in Terminal. # Globals: @@ -203,6 +264,38 @@ addec() { age -d -i {{@@ ageidentity @@}} -o "${2}" "${1}" } +####################################### +# Manage age-encrypted deployment secrets using a project-local policy. +# Globals: +# AGE_DOCKER_IDENTITY +# Arguments: +# age-docker command and arguments +# Outputs: +# Writes command output to stdout and errors to stderr +# Returns: +# age-docker exit status +####################################### +export AGE_DOCKER_IDENTITY="${AGE_DOCKER_IDENTITY:-{{@@ agedockersecretsidentity @@}}}" + +age-docker() { + {{@@ env['HOME'] @@}}/.local/bin/age-docker "$@" +} + +# Register completion only when commonfunc is sourced from an interactive Zsh +# after its completion system has been initialized. +if [[ -n "${ZSH_VERSION:-}" ]] && [[ -o interactive ]] && \ + command -v compdef >/dev/null 2>&1 && \ + [[ -z "${AGE_DOCKER_ZSH_COMPLETION_LOADED:-}" ]]; then + _age_docker_completion="$({{@@ env['HOME'] @@}}/.local/bin/age-docker completion zsh 2>/dev/null)" || \ + _age_docker_completion="" + if [[ -n "${_age_docker_completion}" ]]; then + if eval "${_age_docker_completion}" 2>/dev/null; then + AGE_DOCKER_ZSH_COMPLETION_LOADED=1 + fi + fi + unset _age_docker_completion +fi + {%@@ if distro == 'macos' @@%} ####################################### diff --git a/dotfiles/local/bin/age-docker b/dotfiles/local/bin/age-docker new file mode 100755 index 0000000..b9f8df1 --- /dev/null +++ b/dotfiles/local/bin/age-docker @@ -0,0 +1,991 @@ +#!/usr/bin/env -S uv run --script +# /// script +# requires-python = ">=3.14" +# /// + +# {{@@ header() @@}} +# +# Manage age-encrypted deployment secrets from a project-local policy file. + +from __future__ import annotations + +import argparse +import fcntl +import filecmp +import hashlib +import json +import os +import re +import shlex +import signal +import subprocess +import sys +import tempfile +from dataclasses import dataclass +from pathlib import Path + +import tomllib + +CONFIG_NAME = ".age-docker.toml" +EMPTY_CONFIG = "version = 1\n\n[keys]\n\n[groups]\n\n[secrets]\n" +NAME_PATTERN = re.compile(r"^[A-Za-z0-9_-]+$") +SUPPORTED_SHELLS = ("bash", "zsh") + + +class AgeDockerError(Exception): + pass + + +def install_signal_handlers() -> None: + def interrupted(signum: int, _frame: object) -> None: + raise AgeDockerError(f"Interrupted by {signal.Signals(signum).name}") + + for signum in (signal.SIGHUP, signal.SIGINT, signal.SIGTERM): + signal.signal(signum, interrupted) + + +@dataclass(frozen=True) +class SecretRule: + relative_path: str + path: Path + aliases: tuple[str, ...] + recipients: tuple[str, ...] + + +@dataclass(frozen=True) +class Policy: + path: Path + keys: dict[str, str] + groups: dict[str, list[str]] + secrets: dict[str, SecretRule] + + +def discover_config(explicit: str | None, start: Path | None = None) -> Path: + configured = explicit or os.environ.get("AGE_DOCKER_CONFIG") + if configured: + path = Path(configured).expanduser().resolve() + if not path.is_file(): + raise AgeDockerError(f"Config does not exist: {path}") + return path + + current = (start or Path.cwd()).resolve() + if not current.is_dir(): + current = current.parent + while True: + candidate = current / CONFIG_NAME + if candidate.is_file(): + return candidate + if (current / ".git").exists() or current.parent == current: + break + current = current.parent + raise AgeDockerError(f"Could not find {CONFIG_NAME}") + + +def _string_list(value: object, location: str) -> list[str]: + if not isinstance(value, list) or not all(isinstance(item, str) for item in value): + raise AgeDockerError(f"{location} must be an array of names") + return value + + +def validate_recipient(recipient: str) -> None: + result = subprocess.run( + ["age", "--encrypt", "--recipient", recipient, "--output", os.devnull], + input=b"", + capture_output=True, + check=False, + ) + if result.returncode != 0: + message = result.stderr.decode(errors="replace").strip() + raise AgeDockerError(f"Invalid age recipient {recipient!r}: {message}") + + +def load_policy(path: Path) -> Policy: + try: + data = tomllib.loads(path.read_text(encoding="utf-8")) + except (OSError, tomllib.TOMLDecodeError) as error: + raise AgeDockerError(f"Could not read {path}: {error}") from error + if data.get("version") != 1: + raise AgeDockerError("Config version must be 1") + + keys_raw = data.get("keys", {}) + groups_raw = data.get("groups", {}) + secrets_raw = data.get("secrets", {}) + if ( + not isinstance(keys_raw, dict) + or not isinstance(groups_raw, dict) + or not isinstance(secrets_raw, dict) + ): + raise AgeDockerError("[keys], [groups], and [secrets] must be TOML tables") + + keys: dict[str, str] = {} + for name, recipient in keys_raw.items(): + if not NAME_PATTERN.fullmatch(name) or not isinstance(recipient, str): + raise AgeDockerError(f"Invalid key entry: {name}") + validate_recipient(recipient) + keys[name] = recipient + + groups = { + name: _string_list(value, f"groups.{name}") + for name, value in groups_raw.items() + } + for name in groups: + if not NAME_PATTERN.fullmatch(name): + raise AgeDockerError(f"Invalid group name: {name}") + if name in keys: + raise AgeDockerError(f"Name is both a key and a group: {name}") + + resolved_groups: dict[str, tuple[str, ...]] = {} + + def resolve_name(name: str, stack: tuple[str, ...]) -> tuple[str, ...]: + if name in keys: + return (name,) + if name not in groups: + raise AgeDockerError(f"Unknown key or group: {name}") + if name in stack: + chain = " -> ".join((*stack, name)) + raise AgeDockerError(f"Recipient group cycle: {chain}") + if name not in resolved_groups: + aliases: list[str] = [] + for member in groups[name]: + aliases.extend(resolve_name(member, (*stack, name))) + resolved_groups[name] = tuple(dict.fromkeys(aliases)) + return resolved_groups[name] + + for group_name in groups: + resolve_name(group_name, ()) + + root = path.parent.resolve() + secrets: dict[str, SecretRule] = {} + for relative_path, names_value in secrets_raw.items(): + if not isinstance(relative_path, str): + raise AgeDockerError("Secret paths must be strings") + configured_path = Path(relative_path) + if configured_path.is_absolute(): + raise AgeDockerError(f"Secret path must be relative: {relative_path}") + resolved_path = (root / configured_path).resolve(strict=False) + try: + resolved_path.relative_to(root) + except ValueError as error: + raise AgeDockerError( + f"Secret path escapes the config directory: {relative_path}" + ) from error + if resolved_path.is_symlink() or (root / configured_path).is_symlink(): + raise AgeDockerError(f"Secret path must not be a symlink: {relative_path}") + aliases: list[str] = [] + for name in _string_list(names_value, f"secrets.{relative_path}"): + aliases.extend(resolve_name(name, ())) + aliases = list(dict.fromkeys(aliases)) + if not aliases: + raise AgeDockerError(f"Secret has no recipients: {relative_path}") + normalized = resolved_path.relative_to(root).as_posix() + if normalized in secrets: + raise AgeDockerError(f"Duplicate secret path: {normalized}") + secrets[normalized] = SecretRule( + relative_path=normalized, + path=resolved_path, + aliases=tuple(aliases), + recipients=tuple(dict.fromkeys(keys[alias] for alias in aliases)), + ) + return Policy(path=path, keys=keys, groups=groups, secrets=secrets) + + +def find_rule(policy: Policy, value: str) -> SecretRule: + requested = Path(value).expanduser() + if not requested.is_absolute(): + requested = Path.cwd() / requested + requested = requested.resolve(strict=False) + for rule in policy.secrets.values(): + if requested == rule.path: + return rule + raise AgeDockerError(f"Secret is not configured: {value}") + + +def identity_arguments(identities: list[str] | None) -> list[str]: + configured = os.environ.get("AGE_DOCKER_IDENTITY") + values = identities or ([configured] if configured else []) + if not values: + raise AgeDockerError( + "No age identity configured; set AGE_DOCKER_IDENTITY or use --identity" + ) + paths = [Path(value).expanduser() for value in values] + for path in paths: + if not path.is_file(): + raise AgeDockerError(f"Identity is not readable: {path}") + return [argument for path in paths for argument in ("--identity", str(path))] + + +def run_age(arguments: list[str], *, input_path: Path | None = None) -> None: + command = ["age", *arguments] + if input_path is not None: + command.append(str(input_path)) + result = subprocess.run(command, capture_output=True, check=False) + if result.returncode != 0: + raise AgeDockerError( + result.stderr.decode(errors="replace").strip() or "age failed" + ) + + +def temporary_output(parent: Path, prefix: str) -> Path: + descriptor, name = tempfile.mkstemp(prefix=prefix, dir=parent) + os.close(descriptor) + path = Path(name) + path.unlink() + return path + + +def decrypt_to(source: Path, destination: Path, identities: list[str] | None) -> None: + run_age( + [ + "--decrypt", + *identity_arguments(identities), + "--output", + str(destination), + "--", + ], + input_path=source, + ) + destination.chmod(0o600) + + +def stage_encryption( + rule: SecretRule, + plaintext: Path, + identities: list[str] | None, + *, + allow_empty: bool, +) -> Path: + if not plaintext.is_file() or plaintext.is_symlink(): + raise AgeDockerError(f"Plaintext source is not a regular file: {plaintext}") + if plaintext.stat().st_size == 0 and not allow_empty: + raise AgeDockerError("Refusing to encrypt an empty file; use --allow-empty") + rule.path.parent.mkdir(parents=True, exist_ok=True, mode=0o700) + if rule.path.is_symlink(): + raise AgeDockerError(f"Secret path must not be a symlink: {rule.relative_path}") + staged = temporary_output(rule.path.parent, f".{rule.path.name}.tmp.") + try: + recipients = [ + argument + for recipient in rule.recipients + for argument in ("--recipient", recipient) + ] + run_age( + ["--encrypt", "--armor", *recipients, "--output", str(staged), "--"], + input_path=plaintext, + ) + staged.chmod(0o600) + with tempfile.TemporaryDirectory(prefix="age-docker-verify-") as directory: + verification = Path(directory) / "plaintext" + decrypt_to(staged, verification, identities) + if not filecmp.cmp(plaintext, verification, shallow=False): + raise AgeDockerError( + "Encrypted-file verification did not reproduce the plaintext" + ) + return staged + except BaseException: + staged.unlink(missing_ok=True) + raise + + +def encrypt_file( + rule: SecretRule, + plaintext: Path, + identities: list[str] | None, + *, + allow_empty: bool, +) -> None: + staged = stage_encryption(rule, plaintext, identities, allow_empty=allow_empty) + try: + os.replace(staged, rule.path) + finally: + staged.unlink(missing_ok=True) + + +def decrypt_file( + rule: SecretRule, + output: Path, + identities: list[str] | None, + *, + force: bool, +) -> None: + if not rule.path.is_file(): + raise AgeDockerError(f"Encrypted secret does not exist: {rule.relative_path}") + if output.is_symlink(): + raise AgeDockerError(f"Plaintext destination must not be a symlink: {output}") + if output.exists() and not force: + raise AgeDockerError( + f"Plaintext destination already exists: {output}; use --force" + ) + output.parent.mkdir(parents=True, exist_ok=True) + staged = temporary_output(output.parent, f".{output.name}.tmp.") + try: + decrypt_to(rule.path, staged, identities) + os.replace(staged, output) + output.chmod(0o600) + finally: + staged.unlink(missing_ok=True) + + +def edit_file( + rule: SecretRule, + identities: list[str] | None, + *, + allow_empty: bool, +) -> bool: + with tempfile.TemporaryDirectory(prefix="age-docker-edit-") as directory: + temporary_directory = Path(directory) + temporary_directory.chmod(0o700) + plaintext = temporary_directory / rule.path.name.removesuffix(".age") + existed = rule.path.is_file() + if existed: + decrypt_to(rule.path, plaintext, identities) + before = plaintext.read_bytes() + else: + plaintext.touch(mode=0o600) + before = None + + editor_value = os.environ.get("VISUAL") or os.environ.get("EDITOR") or "vi" + editor = shlex.split(editor_value) + if not editor: + raise AgeDockerError("VISUAL or EDITOR is empty") + result = subprocess.run([*editor, str(plaintext)], check=False) + if result.returncode != 0: + raise AgeDockerError(f"Editor exited with status {result.returncode}") + if not plaintext.is_file() or plaintext.is_symlink(): + raise AgeDockerError("Editor did not leave a regular plaintext file") + if before is not None and plaintext.read_bytes() == before: + return False + encrypt_file(rule, plaintext, identities, allow_empty=allow_empty) + return True + + +def rekey_files( + rules: list[SecretRule], + identities: list[str] | None, + *, + skip_missing: bool, +) -> int: + staged_files: list[tuple[SecretRule, Path]] = [] + rekeyed = 0 + try: + for rule in rules: + if not rule.path.is_file(): + if skip_missing: + print( + f"Warning: skipping missing secret {rule.relative_path}", + file=sys.stderr, + ) + continue + raise AgeDockerError( + f"Encrypted secret does not exist: {rule.relative_path}" + ) + with tempfile.TemporaryDirectory(prefix="age-docker-rekey-") as directory: + plaintext = Path(directory) / "plaintext" + decrypt_to(rule.path, plaintext, identities) + staged = stage_encryption(rule, plaintext, identities, allow_empty=True) + staged_files.append((rule, staged)) + for rule, staged in staged_files: + os.replace(staged, rule.path) + rekeyed += 1 + return rekeyed + finally: + for _, staged in staged_files: + staged.unlink(missing_ok=True) + + +def normalize_recipient(value: str) -> str: + fields = value.strip().split() + if fields and fields[0].startswith("ssh-"): + if len(fields) < 2: + raise AgeDockerError( + "SSH recipients must contain a key type and key material" + ) + recipient = f"{fields[0]} {fields[1]}" + elif len(fields) == 1: + recipient = fields[0] + else: + raise AgeDockerError( + "Recipient must be an SSH public key or native age recipient" + ) + validate_recipient(recipient) + return recipient + + +def recipient_description(recipient: str) -> str: + if not recipient.startswith("ssh-"): + return recipient + with tempfile.TemporaryDirectory(prefix="age-docker-fingerprint-") as directory: + public_key = Path(directory) / "key.pub" + public_key.write_text(f"{recipient}\n", encoding="utf-8") + result = subprocess.run( + ["ssh-keygen", "-lf", str(public_key), "-E", "sha256"], + text=True, + capture_output=True, + check=False, + ) + if result.returncode != 0: + raise AgeDockerError( + result.stderr.strip() or "ssh-keygen could not fingerprint the key" + ) + return result.stdout.strip() + + +def confirm(message: str, *, assume_yes: bool) -> None: + if assume_yes: + return + if not sys.stdin.isatty(): + raise AgeDockerError(f"{message}; pass --yes after verifying the key") + answer = input(f"{message} [y/N] ") + if answer.casefold() not in {"y", "yes"}: + raise AgeDockerError("Cancelled") + + +def mutate_key(policy: Policy, name: str, recipient: str | None) -> None: + if not NAME_PATTERN.fullmatch(name): + raise AgeDockerError(f"Invalid key name: {name}") + lines = policy.path.read_text(encoding="utf-8").splitlines(keepends=True) + try: + section_start = next( + index for index, line in enumerate(lines) if line.strip() == "[keys]" + ) + except StopIteration as error: + raise AgeDockerError("Config has no [keys] section") from error + section_end = next( + ( + index + for index in range(section_start + 1, len(lines)) + if lines[index].lstrip().startswith("[") + ), + len(lines), + ) + entry_pattern = re.compile( + rf'^(?P\s*){re.escape(name)}\s*=\s*"(?:\\.|[^"\\])*"(?P\s*(?:#.*)?)(?P\r?\n)?$' + ) + entry_index = next( + ( + index + for index in range(section_start + 1, section_end) + if entry_pattern.match(lines[index]) + ), + None, + ) + if recipient is None: + if entry_index is None: + raise AgeDockerError(f"Unknown key: {name}") + del lines[entry_index] + elif entry_index is not None: + match = entry_pattern.match(lines[entry_index]) + assert match is not None + lines[entry_index] = ( + f"{match.group('indent')}{name} = {json.dumps(recipient)}" + f"{match.group('suffix')}{match.group('newline') or ''}" + ) + else: + insertion = section_end + while insertion > section_start + 1 and not lines[insertion - 1].strip(): + insertion -= 1 + lines.insert(insertion, f"{name} = {json.dumps(recipient)}\n") + + staged = temporary_output(policy.path.parent, f".{policy.path.name}.tmp.") + try: + staged.write_text("".join(lines), encoding="utf-8") + staged.chmod(policy.path.stat().st_mode & 0o777) + load_policy(staged) + os.replace(staged, policy.path) + finally: + staged.unlink(missing_ok=True) + + +def key_is_referenced(policy: Policy, name: str) -> bool: + if any(name in members for members in policy.groups.values()): + return True + data = tomllib.loads(policy.path.read_text(encoding="utf-8")) + secrets = data.get("secrets", {}) + return isinstance(secrets, dict) and any( + isinstance(members, list) and name in members for members in secrets.values() + ) + + +def add_or_replace_key( + policy: Policy, + name: str, + recipient: str, + *, + replace: bool, + assume_yes: bool, +) -> bool: + recipient = normalize_recipient(recipient) + existing = policy.keys.get(name) + if existing == recipient: + print(f"Key {name} is already configured") + return False + if existing is not None: + if not replace: + raise AgeDockerError( + f"Key {name} already exists with a different value; use --replace" + ) + print(f"Old: {recipient_description(existing)}") + print(f"New: {recipient_description(recipient)}") + confirm(f"Replace key {name}?", assume_yes=assume_yes) + mutate_key(policy, name, recipient) + print(f"{'Replaced' if existing is not None else 'Added'} key {name}") + if existing is not None: + print("Recipient policy changed; run age-docker rekey --all") + return True + + +def scan_host_key(host: str, port: int, key_type: str) -> str: + scan_type = {"ssh-ed25519": "ed25519", "ssh-rsa": "rsa"}.get(key_type, key_type) + recipient_type = {"ed25519": "ssh-ed25519", "rsa": "ssh-rsa"}.get( + scan_type, key_type + ) + result = subprocess.run( + ["ssh-keyscan", "-p", str(port), "-t", scan_type, host], + text=True, + capture_output=True, + check=False, + ) + if result.returncode != 0: + raise AgeDockerError(result.stderr.strip() or f"ssh-keyscan failed for {host}") + for line in result.stdout.splitlines(): + fields = line.split() + if len(fields) >= 3 and fields[1] == recipient_type: + return normalize_recipient(f"{fields[1]} {fields[2]}") + raise AgeDockerError(f"ssh-keyscan returned no {recipient_type} key for {host}") + + +def acquire_policy_lock(policy: Policy): + digest = hashlib.sha256(str(policy.path.resolve()).encode()).hexdigest()[:24] + lock_path = Path(tempfile.gettempdir()) / f"age-docker-{digest}.lock" + lock_file = lock_path.open("a+") + lock_path.chmod(0o600) + try: + fcntl.flock(lock_file.fileno(), fcntl.LOCK_EX | fcntl.LOCK_NB) + except BlockingIOError as error: + lock_file.close() + raise AgeDockerError( + f"Another age-docker command is active for {policy.path}" + ) from error + return lock_file + + +def bash_completion() -> str: + return r'''# bash completion for age-docker +_age_docker() { + local cur prev command key_command word index command_index + COMPREPLY=() + cur="${COMP_WORDS[COMP_CWORD]}" + prev="${COMP_WORDS[COMP_CWORD-1]}" + + case "$prev" in + --config|--identity) + COMPREPLY=( $(compgen -f -- "$cur") ) + return + ;; + esac + + for (( index = 1; index < COMP_CWORD; index++ )); do + word="${COMP_WORDS[index]}" + case "$word" in + --config|--identity) + (( index++ )) + ;; + --config=*|--identity=*) + ;; + init|check|list|encrypt|decrypt|edit|rekey|key|completion) + command="$word" + command_index="$index" + break + ;; + esac + done + + if [[ -z "$command" ]]; then + COMPREPLY=( $(compgen -W '--config --identity --help init check list encrypt decrypt edit rekey key completion' -- "$cur") ) + return + fi + + case "$command" in + init|check|list) + COMPREPLY=( $(compgen -W '--help' -- "$cur") ) + ;; + encrypt) + if [[ "$cur" == -* ]]; then + COMPREPLY=( $(compgen -W '--allow-empty --help' -- "$cur") ) + else + COMPREPLY=( $(compgen -f -- "$cur") ) + fi + ;; + decrypt) + if [[ "$cur" == -* ]]; then + COMPREPLY=( $(compgen -W '--force --help' -- "$cur") ) + else + COMPREPLY=( $(compgen -f -- "$cur") ) + fi + ;; + edit) + if [[ "$cur" == -* ]]; then + COMPREPLY=( $(compgen -W '--allow-empty --help' -- "$cur") ) + else + COMPREPLY=( $(compgen -f -- "$cur") ) + fi + ;; + rekey) + if [[ "$cur" == -* ]]; then + COMPREPLY=( $(compgen -W '--all --help' -- "$cur") ) + else + COMPREPLY=( $(compgen -f -- "$cur") ) + fi + ;; + completion) + COMPREPLY=( $(compgen -W 'bash zsh --help' -- "$cur") ) + ;; + key) + for (( index = command_index + 1; index < COMP_CWORD; index++ )); do + word="${COMP_WORDS[index]}" + case "$word" in + add|scan|remove) + key_command="$word" + break + ;; + esac + done + case "$key_command" in + add) + COMPREPLY=( $(compgen -W '--replace --yes --help' -- "$cur") ) + ;; + scan) + COMPREPLY=( $(compgen -W '--port --type --replace --yes --help' -- "$cur") ) + ;; + remove) + COMPREPLY=( $(compgen -W '--help' -- "$cur") ) + ;; + *) + COMPREPLY=( $(compgen -W 'add scan remove --help' -- "$cur") ) + ;; + esac + ;; + esac +} +complete -F _age_docker age-docker +''' + + +def zsh_completion() -> str: + return r'''#compdef age-docker +# zsh completion for age-docker +_age_docker() { + local command key_command word + local -i command_index key_command_index + + for (( command_index = 2; command_index < CURRENT; command_index++ )); do + word="${words[command_index]}" + case "$word" in + --config|--identity) + (( command_index++ )) + ;; + --config=*|--identity=*) + ;; + init|check|list|encrypt|decrypt|edit|rekey|key|completion) + command="$word" + break + ;; + esac + done + + if [[ -z "$command" ]]; then + _arguments \ + '(-h --help)'{-h,--help}'[show help and exit]' \ + '--config[path to .age-docker.toml]:config file:_files' \ + '*--identity[age identity used to decrypt and verify]:identity file:_files' \ + '1:command:((init\:"create an empty deployment secrets policy" check\:"validate the deployment secrets policy" list\:"list secrets and resolved key aliases" encrypt\:"encrypt a plaintext file" decrypt\:"decrypt a configured secret" edit\:"edit a configured secret" rekey\:"re-encrypt configured secrets" key\:"manage public keys" completion\:"generate shell completion"))' + return + fi + + words=("${words[1]}" "${words[@]:$command_index}") + (( CURRENT = CURRENT - command_index + 1 )) + + case "$command" in + init|check|list) + _arguments '(-h --help)'{-h,--help}'[show help and exit]' + ;; + encrypt) + _arguments \ + '(-h --help)'{-h,--help}'[show help and exit]' \ + '--allow-empty[allow an empty plaintext file]' \ + '1:plaintext file:_files' \ + '2:configured secret:_files' + ;; + decrypt) + _arguments \ + '(-h --help)'{-h,--help}'[show help and exit]' \ + '--force[overwrite an existing plaintext destination]' \ + '1:configured secret:_files' \ + '2:plaintext destination:_files' + ;; + edit) + _arguments \ + '(-h --help)'{-h,--help}'[show help and exit]' \ + '--allow-empty[allow an empty plaintext file]' \ + '1:configured secret:_files' + ;; + rekey) + _arguments \ + '(-h --help)'{-h,--help}'[show help and exit]' \ + '--all[rekey all configured secrets]' \ + '*:configured secret:_files' + ;; + completion) + _arguments \ + '(-h --help)'{-h,--help}'[show help and exit]' \ + '1:shell:(bash zsh)' + ;; + key) + for (( key_command_index = 2; key_command_index < CURRENT; key_command_index++ )); do + word="${words[key_command_index]}" + case "$word" in + add|scan|remove) + key_command="$word" + break + ;; + esac + done + if [[ -n "$key_command" ]]; then + words=("${words[1]}" "${words[@]:$key_command_index}") + (( CURRENT = CURRENT - key_command_index + 1 )) + fi + case "$key_command" in + add) + _arguments \ + '(-h --help)'{-h,--help}'[show help and exit]' \ + '--replace[replace an existing key]' \ + '--yes[skip confirmation]' \ + '1:key name:' \ + '2:age recipient:' + ;; + scan) + _arguments \ + '(-h --help)'{-h,--help}'[show help and exit]' \ + '--port[SSH port]:port:' \ + '--type[SSH key type]:key type:(ssh-ed25519 ssh-rsa)' \ + '--replace[replace an existing key]' \ + '--yes[skip confirmation]' \ + '1:key name:' \ + '2:SSH host:_hosts' + ;; + remove) + _arguments \ + '(-h --help)'{-h,--help}'[show help and exit]' \ + '1:key name:' + ;; + *) + _arguments \ + '(-h --help)'{-h,--help}'[show help and exit]' \ + '1:key command:((add\:"add an existing age recipient" scan\:"retrieve an SSH host public key" remove\:"remove an unreferenced public key"))' + ;; + esac + ;; + esac +} +compdef _age_docker age-docker +''' + + +def generate_completion(shell: str) -> str: + if shell == "bash": + return bash_completion() + if shell == "zsh": + return zsh_completion() + raise AgeDockerError(f"Unsupported shell: {shell}") + + +def build_parser() -> argparse.ArgumentParser: + parser = argparse.ArgumentParser(prog="age-docker") + parser.add_argument("--config", help="path to .age-docker.toml") + parser.add_argument( + "--identity", action="append", help="age identity used to decrypt and verify" + ) + subparsers = parser.add_subparsers(dest="command", required=True) + subparsers.add_parser("init", help="create an empty deployment secrets policy") + subparsers.add_parser("check", help="validate the deployment secrets policy") + subparsers.add_parser("list", help="list secrets and their resolved key aliases") + encrypt = subparsers.add_parser( + "encrypt", help="encrypt a plaintext file for a configured secret" + ) + encrypt.add_argument("plaintext") + encrypt.add_argument("secret") + encrypt.add_argument("--allow-empty", action="store_true") + decrypt = subparsers.add_parser( + "decrypt", help="decrypt a configured secret to a persistent file" + ) + decrypt.add_argument("secret") + decrypt.add_argument("plaintext") + decrypt.add_argument("--force", action="store_true") + edit = subparsers.add_parser("edit", help="edit a configured secret") + edit.add_argument("secret") + edit.add_argument("--allow-empty", action="store_true") + rekey = subparsers.add_parser( + "rekey", help="re-encrypt secrets using the configured recipients" + ) + rekey.add_argument("secrets", nargs="*") + rekey.add_argument("--all", action="store_true", dest="all_secrets") + key = subparsers.add_parser("key", help="manage public keys in the policy") + key_commands = key.add_subparsers(dest="key_command", required=True) + key_add = key_commands.add_parser("add", help="add an existing age recipient") + key_add.add_argument("name") + key_add.add_argument("recipient") + key_add.add_argument("--replace", action="store_true") + key_add.add_argument("--yes", action="store_true") + key_scan = key_commands.add_parser("scan", help="retrieve an SSH host public key") + key_scan.add_argument("name") + key_scan.add_argument("host") + key_scan.add_argument("--port", type=int, default=22) + key_scan.add_argument("--type", default="ssh-ed25519", dest="key_type") + key_scan.add_argument("--replace", action="store_true") + key_scan.add_argument("--yes", action="store_true") + key_remove = key_commands.add_parser( + "remove", help="remove an unreferenced public key" + ) + key_remove.add_argument("name") + completion = subparsers.add_parser( + "completion", help="generate shell completion code" + ) + completion.add_argument("shell", choices=SUPPORTED_SHELLS) + return parser + + +def command_start(args: argparse.Namespace) -> Path: + value: str | None = None + if args.command in {"encrypt", "decrypt", "edit"}: + value = args.secret + elif args.command == "rekey" and args.secrets: + value = args.secrets[0] + if value is None: + return Path.cwd() + path = Path(value).expanduser() + if not path.is_absolute(): + path = Path.cwd() / path + return path.parent + + +def main(argv: list[str] | None = None) -> int: + install_signal_handlers() + args = build_parser().parse_args(argv) + if args.command == "completion": + print(generate_completion(args.shell), end="") + return 0 + if args.command == "init": + configured = args.config or os.environ.get("AGE_DOCKER_CONFIG") + config = ( + Path(configured).expanduser() if configured else Path.cwd() / CONFIG_NAME + ) + if not config.is_absolute(): + config = Path.cwd() / config + try: + with config.open("x", encoding="utf-8") as file: + file.write(EMPTY_CONFIG) + except FileExistsError: + print(f"Config already exists: {config}", file=sys.stderr) + return 1 + print(f"Created {config}") + return 0 + policy = load_policy(discover_config(args.config, command_start(args))) + _lock_file = ( + acquire_policy_lock(policy) + if args.command in {"encrypt", "decrypt", "edit", "rekey", "key"} + else None + ) + if args.command == "check": + count = len(policy.secrets) + print(f"Config is valid: {count} secret{'s' if count != 1 else ''}") + return 0 + if args.command == "list": + for rule in policy.secrets.values(): + print(f"{rule.relative_path}: {', '.join(rule.aliases)}") + return 0 + if args.command == "encrypt": + rule = find_rule(policy, args.secret) + encrypt_file( + rule, + Path(args.plaintext).expanduser().resolve(), + args.identity, + allow_empty=args.allow_empty, + ) + print(f"Encrypted {rule.relative_path}") + return 0 + if args.command == "decrypt": + rule = find_rule(policy, args.secret) + decrypt_file( + rule, + Path(args.plaintext).expanduser().resolve(strict=False), + args.identity, + force=args.force, + ) + print(f"Decrypted {rule.relative_path} to {args.plaintext}") + return 0 + if args.command == "edit": + rule = find_rule(policy, args.secret) + changed = edit_file(rule, args.identity, allow_empty=args.allow_empty) + print( + f"Encrypted {rule.relative_path}" + if changed + else f"{rule.relative_path} is unchanged" + ) + return 0 + if args.command == "rekey": + if args.all_secrets == bool(args.secrets): + raise AgeDockerError("Specify configured secret paths or --all") + rules = ( + list(policy.secrets.values()) + if args.all_secrets + else [find_rule(policy, value) for value in args.secrets] + ) + rekeyed = rekey_files(rules, args.identity, skip_missing=args.all_secrets) + print(f"Rekeyed {rekeyed} secret{'s' if rekeyed != 1 else ''}") + return 0 + if args.command == "key": + if args.key_command == "add": + add_or_replace_key( + policy, + args.name, + args.recipient, + replace=args.replace, + assume_yes=args.yes, + ) + return 0 + if args.key_command == "scan": + recipient = scan_host_key(args.host, args.port, args.key_type) + print(f"Retrieved: {recipient}") + print(f"Fingerprint: {recipient_description(recipient)}") + print( + "Verify this fingerprint through a trusted channel; ssh-keyscan does not authenticate it." + ) + replacing = args.name in policy.keys + if not replacing: + confirm(f"Store key as {args.name}?", assume_yes=args.yes) + add_or_replace_key( + policy, + args.name, + recipient, + replace=args.replace, + assume_yes=args.yes if replacing else True, + ) + return 0 + if args.key_command == "remove": + if args.name not in policy.keys: + raise AgeDockerError(f"Unknown key: {args.name}") + if key_is_referenced(policy, args.name): + raise AgeDockerError( + f"Key {args.name} is still referenced by a group or secret" + ) + mutate_key(policy, args.name, None) + print(f"Removed key {args.name}") + return 0 + return 2 + + +if __name__ == "__main__": + try: + raise SystemExit(main()) + except (AgeDockerError, OSError, subprocess.SubprocessError) as error: + print(f"age-docker: {error}", file=sys.stderr) + raise SystemExit(1) diff --git a/dotfiles/vws2-artifacts-config b/dotfiles/vws2-artifacts-config new file mode 100644 index 0000000..5d74584 --- /dev/null +++ b/dotfiles/vws2-artifacts-config @@ -0,0 +1,17 @@ +-----BEGIN AGE ENCRYPTED FILE----- +YWdlLWVuY3J5cHRpb24ub3JnL3YxCi0+IFgyNTUxOSA0VnFlYkRyUG41RStrd21a +Z0thZEE0bXh3WHJPekttbmxaUVRyWXMvdTNVCmpwUG4rSGYzRzFLVmdJd2tMSEtE +ZEZqazdSL1hPaUYyMy9hR29iSk5Cc1EKLS0tIGt5UGh3dXNFYWg4bm9VS2I4alF4 +dkJuQllIRS9TZFA2c1FRSGZOQUZuV2cKJjjpGuc7qRULyD4MsvFFGDaROSDm1ljz +cX7b2zu4D4AA2yuncDql08eqG5EvxAhl76E4um0sUlqaC9+cN+7VZBRTLXWDMnFt +3XFS9Gl4XvRcSSW+FGD9ciLRy+UtgXRDgkvWfQFEV9Q3JsedDjBk8kdouHILTEKB +xW377ErtABEK1gf7qfucg6wACzBdwp0aYGsoGYKMjz5OZbfFZ2pY+hePcLD0kGXR +qMIFZzlaA0zSGvWt+quEjnV25UbkZHRwitTL+DJkis0tmprMLQhUgNc+UiP8VcVx +3syB6gsdYDhc4Ert+A61rFm4Fs1/50zlFwq3IaOQwrwQVwu33W+mLF0CEL/lT/AI +en9KqFJATRjzPjDEpq026vz/6DQ/td5CtZ2eY8vUj2B2wY1pHRKseqFVEqFElQGq +O1B593h1khgtygelEJ1KLcb5ogifZ6XAWuj93AlQciGlcmidGtafy+Wk9jaR5+u+ +/Cz4lJFtHQDOOqa7rX6l/wE7BFCKLuCgybT9qgRwR8uYfxul8kQ1mYLMKaZVgCXw +O+i61LW3NHJr3LrsL+YskLw2GUNHGsqCaWAfCrKIQTeumOm7M28khIM9gkDiUB2r +sIb+mUwZw88EvQVKeZq6NJ+46ZlCCfesp05j1j13P9pG9NFRC/mb7Jp9zYVDlUh5 +KOclQNKaCYzOkgkPyEE= +-----END AGE ENCRYPTED FILE----- diff --git a/tests/test_age_docker.py b/tests/test_age_docker.py new file mode 100644 index 0000000..fcba435 --- /dev/null +++ b/tests/test_age_docker.py @@ -0,0 +1,834 @@ +from __future__ import annotations + +import json +import os +import signal +import subprocess +import tempfile +import time +import unittest +from pathlib import Path + +SCRIPT = Path(__file__).parents[1] / "dotfiles/local/bin/age-docker" + + +class AgeDockerCliTests(unittest.TestCase): + def run_cli( + self, + *arguments: str, + cwd: Path, + env: dict[str, str] | None = None, + ) -> subprocess.CompletedProcess[str]: + return subprocess.run( + [str(SCRIPT), *arguments], + cwd=cwd, + env=env, + text=True, + capture_output=True, + check=False, + ) + + def make_identity(self, root: Path, name: str = "identity") -> tuple[Path, str]: + identity = root / name + generated = subprocess.run( + ["age-keygen", "-o", str(identity)], + text=True, + capture_output=True, + check=True, + ) + public_key = next( + line.removeprefix("Public key: ") + for line in generated.stderr.splitlines() + if line.startswith("Public key: ") + ) + return identity, public_key + + def make_ssh_public_key(self, root: Path, name: str) -> str: + private_key = root / name + subprocess.run( + ["ssh-keygen", "-q", "-t", "ed25519", "-N", "", "-f", str(private_key)], + check=True, + ) + fields = private_key.with_suffix(".pub").read_text().split() + return f"{fields[0]} {fields[1]}" + + def write_policy( + self, + root: Path, + personal: str, + server: str, + secret: str = "secrets/prod.env.age", + ) -> None: + (root / ".age-docker.toml").write_text( + f'''version = 1 + +[keys] +phg = "{personal}" +server = "{server}" + +[groups] +users = ["phg"] +production = ["server"] + +[secrets] +"{secret}" = ["users", "production"] +''' + ) + + def test_init_creates_a_versioned_empty_config_without_overwriting_it(self) -> None: + with tempfile.TemporaryDirectory() as directory: + root = Path(directory) + + created = self.run_cli("init", cwd=root) + + self.assertEqual(created.returncode, 0, created.stderr) + config = root / ".age-docker.toml" + self.assertEqual( + config.read_text(), + "version = 1\n\n[keys]\n\n[groups]\n\n[secrets]\n", + ) + + refused = self.run_cli("init", cwd=root) + self.assertNotEqual(refused.returncode, 0) + self.assertIn("already exists", refused.stderr) + + def test_init_honors_the_config_environment_override(self) -> None: + with tempfile.TemporaryDirectory() as directory: + root = Path(directory) + destination = root / "policy.toml" + + result = self.run_cli( + "init", + cwd=root, + env={**os.environ, "AGE_DOCKER_CONFIG": str(destination)}, + ) + + self.assertEqual(result.returncode, 0, result.stderr) + self.assertTrue(destination.is_file()) + self.assertFalse((root / ".age-docker.toml").exists()) + + def test_identity_environment_default_and_explicit_override(self) -> None: + with tempfile.TemporaryDirectory() as directory: + root = Path(directory) + identity, personal = self.make_identity(root, "personal") + _, server = self.make_identity(root, "server") + self.write_policy(root, personal, server) + plaintext = root / "secret.txt" + plaintext.write_text("secret\n") + encrypted = root / "secrets/prod.env.age" + + from_environment = self.run_cli( + "encrypt", + str(plaintext), + str(encrypted), + cwd=root, + env={**os.environ, "AGE_DOCKER_IDENTITY": str(identity)}, + ) + explicit_override = self.run_cli( + "--identity", + str(identity), + "decrypt", + str(encrypted), + str(root / "decrypted.txt"), + cwd=root, + env={**os.environ, "AGE_DOCKER_IDENTITY": str(root / "missing")}, + ) + + self.assertEqual(from_environment.returncode, 0, from_environment.stderr) + self.assertEqual(explicit_override.returncode, 0, explicit_override.stderr) + + def test_identity_is_required_only_for_identity_dependent_commands(self) -> None: + with tempfile.TemporaryDirectory() as directory: + root = Path(directory) + _, personal = self.make_identity(root, "personal") + _, server = self.make_identity(root, "server") + self.write_policy(root, personal, server) + env = {key: value for key, value in os.environ.items() if key != "AGE_DOCKER_IDENTITY"} + + checked = self.run_cli("check", cwd=root, env=env) + plaintext = root / "secret.txt" + plaintext.write_text("secret\n") + encrypted = self.run_cli( + "encrypt", + str(plaintext), + str(root / "secrets/prod.env.age"), + cwd=root, + env=env, + ) + + self.assertEqual(checked.returncode, 0, checked.stderr) + self.assertNotEqual(encrypted.returncode, 0) + self.assertIn("AGE_DOCKER_IDENTITY", encrypted.stderr) + + def test_completion_generators_emit_valid_self_contained_shell_code(self) -> None: + with tempfile.TemporaryDirectory() as directory: + root = Path(directory) + for shell in ("bash", "zsh"): + with self.subTest(shell=shell): + generated = self.run_cli("completion", shell, cwd=root) + + self.assertEqual(generated.returncode, 0, generated.stderr) + self.assertIn("_age_docker", generated.stdout) + self.assertIn("completion", generated.stdout) + syntax = subprocess.run( + [shell, "-n"], + input=generated.stdout, + text=True, + capture_output=True, + check=False, + ) + self.assertEqual(syntax.returncode, 0, syntax.stderr) + + def test_age_docker_contains_no_runtime_dotdrop_templates(self) -> None: + templates = [ + line + for line in SCRIPT.read_text().splitlines() + if "{{@@" in line or "{%@@" in line + ] + + self.assertEqual(templates, ["# {{@@ header() @@}}"]) + + def test_edit_changes_plaintext_but_a_noop_editor_preserves_ciphertext( + self, + ) -> None: + with tempfile.TemporaryDirectory() as directory: + root = Path(directory) + personal_identity, personal = self.make_identity(root, "personal") + _, server = self.make_identity(root, "server") + self.write_policy(root, personal, server) + plaintext = root / "initial.env" + plaintext.write_text("TOKEN=before\n") + encrypted = root / "secrets/prod.env.age" + created = self.run_cli( + "--identity", + str(personal_identity), + "encrypt", + str(plaintext), + str(encrypted), + cwd=root, + ) + self.assertEqual(created.returncode, 0, created.stderr) + + editor = root / "editor" + editor.write_text('#!/bin/sh\nprintf "TOKEN=after\\n" > "$1"\n') + editor.chmod(0o700) + editor_env = { + **os.environ, + "VISUAL": str(editor), + "EDITOR": "/usr/bin/false", + } + changed = self.run_cli( + "--identity", + str(personal_identity), + "edit", + str(encrypted), + cwd=root, + env=editor_env, + ) + changed_ciphertext = encrypted.read_bytes() + + self.assertEqual(changed.returncode, 0, changed.stderr) + decrypted = subprocess.run( + [ + "age", + "--decrypt", + "--identity", + str(personal_identity), + str(encrypted), + ], + capture_output=True, + check=True, + ).stdout + self.assertEqual(decrypted, b"TOKEN=after\n") + + noop_env = {**os.environ, "VISUAL": "/usr/bin/true"} + noop = self.run_cli( + "--identity", + str(personal_identity), + "edit", + str(encrypted), + cwd=root, + env=noop_env, + ) + + self.assertEqual(noop.returncode, 0, noop.stderr) + self.assertIn("unchanged", noop.stdout) + self.assertEqual(encrypted.read_bytes(), changed_ciphertext) + + def test_check_and_list_resolve_nested_recipient_groups(self) -> None: + with tempfile.TemporaryDirectory() as directory: + root = Path(directory) + _, personal = self.make_identity(root, "personal") + _, server = self.make_identity(root, "server") + (root / ".age-docker.toml").write_text( + f'''version = 1 + +[keys] +phg = "{personal}" +sbx0docker01 = "{server}" + +[groups] +users = ["phg"] +production = ["sbx0docker01"] +prod_access = ["users", "production"] + +[secrets] +"secrets/prod.env.age" = ["prod_access"] +''' + ) + + checked = self.run_cli("check", cwd=root) + listed = self.run_cli("list", cwd=root) + + self.assertEqual(checked.returncode, 0, checked.stderr) + self.assertIn("1 secret", checked.stdout) + self.assertEqual(listed.returncode, 0, listed.stderr) + self.assertEqual( + listed.stdout, + "secrets/prod.env.age: phg, sbx0docker01\n", + ) + + def test_encrypt_and_decrypt_use_configured_recipients_and_safe_output( + self, + ) -> None: + with tempfile.TemporaryDirectory() as directory: + root = Path(directory) + personal_identity, personal = self.make_identity(root, "personal") + server_identity, server = self.make_identity(root, "server") + self.write_policy(root, personal, server) + plaintext = root / "prod.env" + plaintext.write_text("TOKEN=correct-horse\n") + encrypted = root / "secrets/prod.env.age" + + encrypted_result = self.run_cli( + "--identity", + str(personal_identity), + "encrypt", + str(plaintext), + str(encrypted), + cwd=root, + ) + + self.assertEqual(encrypted_result.returncode, 0, encrypted_result.stderr) + self.assertTrue( + encrypted.read_text().startswith("-----BEGIN AGE ENCRYPTED FILE-----") + ) + server_plaintext = subprocess.run( + [ + "age", + "--decrypt", + "--identity", + str(server_identity), + str(encrypted), + ], + capture_output=True, + check=True, + ).stdout + self.assertEqual(server_plaintext, b"TOKEN=correct-horse\n") + + output = root / "decrypted.env" + decrypted_result = self.run_cli( + "--identity", + str(personal_identity), + "decrypt", + str(encrypted), + str(output), + cwd=root, + ) + refused = self.run_cli( + "--identity", + str(personal_identity), + "decrypt", + str(encrypted), + str(output), + cwd=root, + ) + + self.assertEqual(decrypted_result.returncode, 0, decrypted_result.stderr) + self.assertEqual(output.read_text(), "TOKEN=correct-horse\n") + self.assertEqual(output.stat().st_mode & 0o777, 0o600) + self.assertNotEqual(refused.returncode, 0) + self.assertIn("already exists", refused.stderr) + + def test_rekey_all_changes_nothing_on_failure_then_uses_updated_recipients( + self, + ) -> None: + with tempfile.TemporaryDirectory() as directory: + root = Path(directory) + personal_identity, personal = self.make_identity(root, "personal") + _, old_server = self.make_identity(root, "old-server") + new_server_identity, new_server = self.make_identity(root, "new-server") + config = root / ".age-docker.toml" + + def write_config(server: str) -> None: + config.write_text( + f'''version = 1 + +[keys] +phg = "{personal}" +server = "{server}" + +[groups] +deployment = ["phg", "server"] + +[secrets] +"secrets/one.age" = ["deployment"] +"secrets/two.age" = ["deployment"] +''' + ) + + write_config(old_server) + for name in ("one", "two"): + plaintext = root / f"{name}.txt" + plaintext.write_text(f"secret-{name}\n") + result = self.run_cli( + "--identity", + str(personal_identity), + "encrypt", + str(plaintext), + str(root / f"secrets/{name}.age"), + cwd=root, + ) + self.assertEqual(result.returncode, 0, result.stderr) + + first = root / "secrets/one.age" + second = root / "secrets/two.age" + original_first = first.read_bytes() + original_second = second.read_bytes() + write_config(new_server) + second.write_text("not age ciphertext\n") + + failed = self.run_cli( + "--identity", + str(personal_identity), + "rekey", + "--all", + cwd=root, + ) + + self.assertNotEqual(failed.returncode, 0) + self.assertEqual(first.read_bytes(), original_first) + + second.write_bytes(original_second) + succeeded = self.run_cli( + "--identity", + str(personal_identity), + "rekey", + "--all", + cwd=root, + ) + + self.assertEqual(succeeded.returncode, 0, succeeded.stderr) + for name in ("one", "two"): + decrypted = subprocess.run( + [ + "age", + "--decrypt", + "--identity", + str(new_server_identity), + str(root / f"secrets/{name}.age"), + ], + capture_output=True, + check=True, + ).stdout + self.assertEqual(decrypted, f"secret-{name}\n".encode()) + + def test_key_scan_preserves_config_format_and_remove_refuses_referenced_keys( + self, + ) -> None: + with tempfile.TemporaryDirectory() as directory: + root = Path(directory) + _, personal = self.make_identity(root, "personal") + scanned_key = self.make_ssh_public_key(root, "host-key") + config = root / ".age-docker.toml" + config.write_text( + f'''version = 1 + +[keys] +# This comment and surrounding policy must survive key edits. +phg = "{personal}" + +[groups] +users = ["phg"] + +[secrets] +"secrets/prod.age" = ["users"] +''' + ) + fake_bin = root / "bin" + fake_bin.mkdir() + fake_keyscan = fake_bin / "ssh-keyscan" + fake_keyscan.write_text( + f'#!/bin/sh\n[ "$4" = "ed25519" ] || exit 9\nprintf "example.test {scanned_key}\\n"\n' + ) + fake_keyscan.chmod(0o700) + env = {**os.environ, "PATH": f"{fake_bin}:{os.environ['PATH']}"} + + scanned = self.run_cli( + "key", + "scan", + "server", + "example.test", + "--yes", + cwd=root, + env=env, + ) + + self.assertEqual(scanned.returncode, 0, scanned.stderr) + self.assertIn("SHA256:", scanned.stdout) + self.assertIn( + "# This comment and surrounding policy must survive key edits.", + config.read_text(), + ) + self.assertIn(f'server = "{scanned_key}"', config.read_text()) + + replacement_key = self.make_ssh_public_key(root, "replacement-host-key") + fake_keyscan.write_text( + f'#!/bin/sh\n[ "$4" = "ed25519" ] || exit 9\nprintf "example.test {replacement_key}\\n"\n' + ) + replacement = self.run_cli( + "key", + "scan", + "server", + "example.test", + "--replace", + cwd=root, + env=env, + ) + + self.assertNotEqual(replacement.returncode, 0) + self.assertIn("Old:", replacement.stdout) + self.assertIn("New:", replacement.stdout) + self.assertIn("--yes", replacement.stderr) + + removed = self.run_cli("key", "remove", "server", cwd=root) + refused = self.run_cli("key", "remove", "phg", cwd=root) + + self.assertEqual(removed.returncode, 0, removed.stderr) + self.assertNotIn("server =", config.read_text()) + self.assertNotEqual(refused.returncode, 0) + self.assertIn("still referenced", refused.stderr) + + def test_mutating_commands_refuse_concurrent_use_of_the_same_policy(self) -> None: + with tempfile.TemporaryDirectory() as directory: + root = Path(directory) + personal_identity, personal = self.make_identity(root, "personal") + _, server = self.make_identity(root, "server") + self.write_policy(root, personal, server) + plaintext = root / "initial.env" + plaintext.write_text("TOKEN=value\n") + encrypted = root / "secrets/prod.env.age" + created = self.run_cli( + "--identity", + str(personal_identity), + "encrypt", + str(plaintext), + str(encrypted), + cwd=root, + ) + self.assertEqual(created.returncode, 0, created.stderr) + + ready = root / "editor-ready" + editor = root / "slow-editor" + editor.write_text(f'#!/bin/sh\ntouch "{ready}"\nsleep 10\n') + editor.chmod(0o700) + env = {**os.environ, "VISUAL": str(editor)} + first = subprocess.Popen( + [ + str(SCRIPT), + "--identity", + str(personal_identity), + "edit", + str(encrypted), + ], + cwd=root, + env=env, + text=True, + stdout=subprocess.PIPE, + stderr=subprocess.PIPE, + start_new_session=True, + ) + try: + deadline = time.monotonic() + 5 + while not ready.exists() and time.monotonic() < deadline: + time.sleep(0.05) + self.assertTrue(ready.exists(), "first editor did not start") + + second = self.run_cli( + "--identity", + str(personal_identity), + "edit", + str(encrypted), + cwd=root, + env={**os.environ, "VISUAL": "/usr/bin/true"}, + ) + + self.assertNotEqual(second.returncode, 0) + self.assertIn("Another age-docker command is active", second.stderr) + finally: + os.killpg(first.pid, signal.SIGTERM) + first.communicate(timeout=5) + + def test_encrypt_preserves_existing_ciphertext_when_personal_access_is_omitted( + self, + ) -> None: + with tempfile.TemporaryDirectory() as directory: + root = Path(directory) + personal_identity, personal = self.make_identity(root, "personal") + _, server = self.make_identity(root, "server") + self.write_policy(root, personal, server) + plaintext = root / "secret.txt" + plaintext.write_text("before\n") + encrypted = root / "secrets/prod.env.age" + created = self.run_cli( + "--identity", + str(personal_identity), + "encrypt", + str(plaintext), + str(encrypted), + cwd=root, + ) + self.assertEqual(created.returncode, 0, created.stderr) + original = encrypted.read_bytes() + (root / ".age-docker.toml").write_text( + f'''version = 1 + +[keys] +phg = "{personal}" +server = "{server}" + +[groups] +production = ["server"] + +[secrets] +"secrets/prod.env.age" = ["production"] +''' + ) + plaintext.write_text("after\n") + + refused = self.run_cli( + "--identity", + str(personal_identity), + "encrypt", + str(plaintext), + str(encrypted), + cwd=root, + ) + + self.assertNotEqual(refused.returncode, 0) + self.assertEqual(encrypted.read_bytes(), original) + + def test_check_rejects_cycles_unknown_names_escaping_paths_and_symlinks( + self, + ) -> None: + with tempfile.TemporaryDirectory() as directory: + root = Path(directory) + _, personal = self.make_identity(root, "personal") + cases = { + "cycle": ( + f'''version = 1 +[keys] +phg = "{personal}" +[groups] +one = ["two"] +two = ["one"] +[secrets] +"secret.age" = ["phg"] +''', + "cycle", + ), + "unknown": ( + f'''version = 1 +[keys] +phg = "{personal}" +[groups] +users = ["missing"] +[secrets] +"secret.age" = ["users"] +''', + "Unknown", + ), + "escape": ( + f'''version = 1 +[keys] +phg = "{personal}" +[groups] +users = ["phg"] +[secrets] +"../secret.age" = ["users"] +''', + "escapes", + ), + } + for name, (config, message) in cases.items(): + with self.subTest(name=name): + (root / ".age-docker.toml").write_text(config) + result = self.run_cli("check", cwd=root) + self.assertNotEqual(result.returncode, 0) + self.assertIn(message, result.stderr) + + target = root / "actual.age" + target.touch() + symlink = root / "secret.age" + symlink.symlink_to(target) + (root / ".age-docker.toml").write_text( + f'''version = 1 +[keys] +phg = "{personal}" +[groups] +users = ["phg"] +[secrets] +"secret.age" = ["users"] +''' + ) + result = self.run_cli("check", cwd=root) + self.assertNotEqual(result.returncode, 0) + self.assertIn("symlink", result.stderr) + + def test_rekey_all_skips_missing_but_explicit_rekey_rejects_it(self) -> None: + with tempfile.TemporaryDirectory() as directory: + root = Path(directory) + personal_identity, personal = self.make_identity(root, "personal") + _, server = self.make_identity(root, "server") + self.write_policy(root, personal, server, secret="secrets/missing.age") + + all_result = self.run_cli( + "--identity", + str(personal_identity), + "rekey", + "--all", + cwd=root, + ) + explicit_result = self.run_cli( + "--identity", + str(personal_identity), + "rekey", + "secrets/missing.age", + cwd=root, + ) + + self.assertEqual(all_result.returncode, 0, all_result.stderr) + self.assertIn("skipping missing", all_result.stderr) + self.assertNotEqual(explicit_result.returncode, 0) + self.assertIn("does not exist", explicit_result.stderr) + + def test_key_replacement_requires_explicit_authorization(self) -> None: + with tempfile.TemporaryDirectory() as directory: + root = Path(directory) + _, first = self.make_identity(root, "first") + _, second = self.make_identity(root, "second") + initialized = self.run_cli("init", cwd=root) + self.assertEqual(initialized.returncode, 0, initialized.stderr) + added = self.run_cli("key", "add", "operator", first, cwd=root) + refused = self.run_cli("key", "add", "operator", second, cwd=root) + replaced = self.run_cli( + "key", + "add", + "operator", + second, + "--replace", + "--yes", + cwd=root, + ) + + self.assertEqual(added.returncode, 0, added.stderr) + self.assertNotEqual(refused.returncode, 0) + self.assertIn("--replace", refused.stderr) + self.assertEqual(replaced.returncode, 0, replaced.stderr) + self.assertIn( + f'operator = "{second}"', (root / ".age-docker.toml").read_text() + ) + + def test_edit_removes_plaintext_temporary_files_after_sigterm(self) -> None: + with tempfile.TemporaryDirectory() as directory: + root = Path(directory) + temp_root = root / "tmp" + temp_root.mkdir() + personal_identity, personal = self.make_identity(root, "personal") + _, server = self.make_identity(root, "server") + self.write_policy(root, personal, server) + plaintext = root / "initial.env" + plaintext.write_text("TOKEN=value\n") + encrypted = root / "secrets/prod.env.age" + created = self.run_cli( + "--identity", + str(personal_identity), + "encrypt", + str(plaintext), + str(encrypted), + cwd=root, + ) + self.assertEqual(created.returncode, 0, created.stderr) + + ready = root / "signal-editor-ready" + editor = root / "signal-editor" + editor.write_text(f'#!/bin/sh\ntouch "{ready}"\nsleep 10\n') + editor.chmod(0o700) + process = subprocess.Popen( + [ + str(SCRIPT), + "--identity", + str(personal_identity), + "edit", + str(encrypted), + ], + cwd=root, + env={**os.environ, "VISUAL": str(editor), "TMPDIR": str(temp_root)}, + stdout=subprocess.DEVNULL, + stderr=subprocess.DEVNULL, + start_new_session=True, + ) + try: + deadline = time.monotonic() + 5 + while not ready.exists() and time.monotonic() < deadline: + time.sleep(0.05) + self.assertTrue(ready.exists(), "editor did not start") + process.terminate() + process.wait(timeout=5) + self.assertEqual(list(temp_root.glob("age-docker-edit-*")), []) + finally: + try: + os.killpg(process.pid, signal.SIGTERM) + except ProcessLookupError: + pass + + def test_duplicate_key_aliases_produce_one_recipient_stanza(self) -> None: + with tempfile.TemporaryDirectory() as directory: + root = Path(directory) + identity, recipient = self.make_identity(root, "personal") + (root / ".age-docker.toml").write_text( + f'''version = 1 +[keys] +primary = "{recipient}" +duplicate = "{recipient}" +[groups] +users = ["primary", "duplicate"] +[secrets] +"secret.age" = ["users"] +''' + ) + plaintext = root / "secret.txt" + plaintext.write_text("secret\n") + encrypted = root / "secret.age" + + result = self.run_cli( + "--identity", + str(identity), + "encrypt", + str(plaintext), + str(encrypted), + cwd=root, + ) + inspection = subprocess.run( + ["age-inspect", "--json", str(encrypted)], + text=True, + capture_output=True, + check=True, + ) + + self.assertEqual(result.returncode, 0, result.stderr) + self.assertEqual(json.loads(inspection.stdout)["stanza_types"], ["X25519"]) + + +if __name__ == "__main__": + unittest.main()