diff --git a/README.md b/README.md index 8bd502a..9f8e153 100644 --- a/README.md +++ b/README.md @@ -97,58 +97,6 @@ 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 d7f8c8a..9934ef4 100644 --- a/config.yaml +++ b/config.yaml @@ -5,7 +5,6 @@ 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: @@ -150,10 +149,6 @@ 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 @@ -180,11 +175,6 @@ 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: @@ -193,7 +183,6 @@ profiles: wsl: false dotfiles: - d_colors - - f_age_docker - f_agenix_helper - f_codew - f_commonfunc @@ -218,7 +207,6 @@ 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 77497f8..dbd39d7 100644 --- a/config/brew/Brewfile +++ b/config/brew/Brewfile @@ -167,8 +167,6 @@ 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 27bd5a5..e8ee8d3 100644 --- a/dotfiles/commonfunc +++ b/dotfiles/commonfunc @@ -101,67 +101,6 @@ 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: @@ -264,38 +203,6 @@ 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 deleted file mode 100755 index b9f8df1..0000000 --- a/dotfiles/local/bin/age-docker +++ /dev/null @@ -1,991 +0,0 @@ -#!/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 deleted file mode 100644 index 5d74584..0000000 --- a/dotfiles/vws2-artifacts-config +++ /dev/null @@ -1,17 +0,0 @@ ------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 deleted file mode 100644 index fcba435..0000000 --- a/tests/test_age_docker.py +++ /dev/null @@ -1,834 +0,0 @@ -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()