README.md

@@ -97,58 +97,6 @@ age-edit -e "${HOME}/.local/bin/codew" -t /tmp/ -M -a ~/.age/phg-age-dotfiles <p
 
 Both aliases are configured via my `.zshrc`.
 
-## Deployment secrets
-
-`age-docker` manages armored age secrets from a project-local
-`.age-docker.toml`. It searches upward for the nearest policy file, stopping
-at the Git root. A policy declares public keys, reusable groups, and the exact
-recipients for every encrypted file:
-
-```toml
-version = 1
-
-[keys]
-phg = "age1..."
-server = "ssh-ed25519 AAAA..."
-
-[groups]
-users = ["phg"]
-production = ["server"]
-
-[secrets]
-"secrets/prod.env.age" = ["users", "production"]
-```
-
-Recipients are never added implicitly. New ciphertext is verified by
-decrypting it with `~/.age/phg-age-dotfiles` before the configured file is
-atomically replaced. Use repeatable `--identity <path>` 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: <https://macos-defaults.com/>

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

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

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' @@%}
 
 #######################################

dotfiles/local/bin/age-docker

@@ -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<indent>\s*){re.escape(name)}\s*=\s*"(?:\\.|[^"\\])*"(?P<suffix>\s*(?:#.*)?)(?P<newline>\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)

dotfiles/vws2-artifacts-config

@@ -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-----

tests/test_age_docker.py

@@ -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()