scripts.secrets_cli

Reference › scripts › secrets_cli

scripts/secrets_cli.py is the command-line tool that owns the encrypt and decrypt lifecycle for the repository's secrets, using SOPS with age keys. A human runs it when a machine is set up or the secrets change. Other scripts only consume its output; for example provision_worktree_env materialises a worktree's .env. It has no internal imports.

The program name is secrets and it has five subcommands, dispatched by main: doctor (a preflight of everything needed to decrypt), init (create an age identity and .sops.yaml if absent), encrypt (encrypt the plaintext .env, or --source, into the tracked ciphertext), decrypt and status. They share --repo-root and --identity (default SOPS_AGE_KEY_FILE or the standard age keys path); the others add --sops-config and --ciphertext. Helpers include resolve_identity_path, derive_public_key (the age recipient for an identity), load_sops_config, find_creation_rule and write_sops_config, and diff_keys, which compares plaintext and ciphertext by key names only, never by value. _run wraps the sops and age subprocesses. An unhandled command returns the failure code instead of falling through to success.

At a glance

Class diagram

Dependency diagram

No internal imports and no importers.

API

Classes

SecretsError

Bases: RuntimeError · Line 73

A sops/age invocation failed in a way the caller must see.

CommandResult (dataclass)

Decorators: dataclass · Line 83

DoctorCheck (dataclass)

Decorators: dataclass · Line 279

DoctorReport (dataclass)

Decorators: dataclass · Line 291

InitResult (dataclass)

Decorators: dataclass · Line 419

EncryptResult (dataclass)

Decorators: dataclass · Line 492

DecryptResult (dataclass)

Decorators: dataclass · Line 600

StatusReport (dataclass)

Decorators: dataclass · Line 732

Functions

Source

Permalink: https://gitlab.com/MSamy/orchestrata-ultimate-architecture/-/blob/a4cbc2bdf/scripts/secrets_cli.py

#!/usr/bin/env python3
"""CLI that owns the encrypt/decrypt lifecycle for the repo's secrets.

Threat model: ``.env`` holds live API keys and passwords. It must be
recoverable from the repo on any machine that has THE KEY, and unreadable to
anyone who does not. Plaintext must be structurally incapable of reaching a
commit.

Crypto: SOPS (system binary) + age (system binary) — no PyPI crypto
dependency. The age identity (PRIVATE key) lives outside the repo, at
``~/.config/sops/age/keys.txt`` by default (overridable via
``SOPS_AGE_KEY_FILE``, or the ``--identity`` flag on every subcommand). The
age recipient (PUBLIC key) lives in the tracked ``.sops.yaml``.

Tracked ciphertext: ``secrets/env.sops.env`` — SOPS dotenv format, so a
``git diff`` shows WHICH setting changed without revealing what it changed
to. Plaintext ``.env`` stays gitignored forever; nothing here installs a
clean/smudge filter, because a filter absent on a fresh clone would silently
commit plaintext.

Idempotence: SOPS mints a fresh data key on every ``encrypt`` invocation, so
a naive "always write" implementation churns the ciphertext (and git) on
every run even when nothing changed. ``encrypt`` decrypts the EXISTING
ciphertext (if any), canonicalizes it and the source into key/value maps via
SOPS's own JSON codec (never a hand-rolled dotenv parser — comments, blank
lines, and literal quoting inside values are SOPS's call, not ours), and
skips the write when the maps are equal.

Usage::

    python scripts/secrets_cli.py doctor
    python scripts/secrets_cli.py init
    python scripts/secrets_cli.py encrypt [--source .env]
    python scripts/secrets_cli.py decrypt [--dest .env] [--force]
    python scripts/secrets_cli.py status [--json]
"""

from __future__ import annotations

import argparse
import json
import os
import shutil
import subprocess
import sys
import tempfile
from collections.abc import Mapping
from dataclasses import dataclass, field
from pathlib import Path

import yaml

REPO_ROOT = Path(__file__).resolve().parent.parent

DEFAULT_IDENTITY_PATH = Path.home() / ".config" / "sops" / "age" / "keys.txt"
DEFAULT_SOURCE = REPO_ROOT / ".env"
DEFAULT_CIPHERTEXT = REPO_ROOT / "secrets" / "env.sops.env"
DEFAULT_SOPS_CONFIG = REPO_ROOT / ".sops.yaml"

# The ciphertext's path RELATIVE TO THE REPO ROOT. This is both the tracked
# artifact's real location and the string SOPS matches .sops.yaml
# creation_rules against (via --filename-override, since the real source
# file we hand SOPS is .env, not secrets/env.sops.env — the plaintext is
# never the thing SOPS's own path-based rule matching sees).
CIPHERTEXT_RELPATH = "secrets/env.sops.env"
PATH_REGEX = r"secrets/env\.sops\.env$"

EXIT_OK = 0
EXIT_FAIL = 1
EXIT_PRECONDITION = 2


class SecretsError(RuntimeError):
    """A sops/age invocation failed in a way the caller must see."""


# ---------------------------------------------------------------------------
# Subprocess plumbing — the ONE place sops/age get invoked.
# ---------------------------------------------------------------------------


@dataclass
class CommandResult:
    returncode: int
    stdout: str
    stderr: str

    @property
    def ok(self) -> bool:
        return self.returncode == 0


def _run(
    cmd: list[str],
    *,
    cwd: Path,
    env: Mapping[str, str],
    input_text: str | None = None,
) -> CommandResult:
    """Run one subprocess, real stdin/stdout/stderr, never raising on a
    nonzero exit — the caller decides what a failure means.

    ``subprocess.run`` raises the same ``FileNotFoundError`` whether the
    executable is missing or ``cwd`` doesn't exist; checking ``cwd`` first
    keeps the reported cause honest instead of blaming the binary for a
    caller bug in the working directory.
    """
    if not Path(cwd).is_dir():
        return CommandResult(
            returncode=127, stdout="", stderr=f"cwd does not exist: {cwd}"
        )
    try:
        proc = subprocess.run(
            cmd,
            cwd=str(cwd),
            env=dict(env),
            input=input_text,
            capture_output=True,
            text=True,
            encoding="utf-8",
        )
    except FileNotFoundError:
        return CommandResult(
            returncode=127, stdout="", stderr=f"{cmd[0]}: command not found on PATH"
        )
    return CommandResult(proc.returncode, proc.stdout, proc.stderr)


def resolve_identity_path(env: Mapping[str, str] | None = None) -> Path:
    """SOPS_AGE_KEY_FILE if set (read from ``env``, not ``os.environ``
    directly, so a caller can inject a throwaway identity without mutating
    process state), else the XDG default."""
    source = env if env is not None else os.environ
    override = (source.get("SOPS_AGE_KEY_FILE") or "").strip()
    if override:
        return Path(override)
    return DEFAULT_IDENTITY_PATH


def _sops_env(
    identity_path: Path, base_env: Mapping[str, str] | None = None
) -> dict[str, str]:
    merged = dict(base_env if base_env is not None else os.environ)
    merged["SOPS_AGE_KEY_FILE"] = str(identity_path)
    return merged


def derive_public_key(identity_path: Path, *, cwd: Path, env: Mapping[str, str]) -> str:
    """The age recipient (public key) for an identity file, via
    ``age-keygen -y`` — the same derivation used to populate .sops.yaml and
    to check it hasn't drifted."""
    result = _run(["age-keygen", "-y", str(identity_path)], cwd=cwd, env=env)
    if not result.ok:
        raise SecretsError(f"age-keygen -y failed: {result.stderr.strip()}")
    pub = result.stdout.strip()
    if not pub:
        raise SecretsError("age-keygen -y produced no public key")
    return pub


# ---------------------------------------------------------------------------
# .sops.yaml
# ---------------------------------------------------------------------------


def load_sops_config(config_path: Path) -> dict:
    if not config_path.exists():
        return {}
    loaded = yaml.safe_load(config_path.read_text(encoding="utf-8"))
    return loaded or {}


def find_creation_rule(config: dict, path_regex: str) -> dict | None:
    for rule in config.get("creation_rules") or []:
        if rule.get("path_regex") == path_regex:
            return rule
    return None


def write_sops_config(config_path: Path, recipient: str, path_regex: str) -> None:
    config = load_sops_config(config_path)
    rules = list(config.get("creation_rules") or [])
    rule = find_creation_rule({"creation_rules": rules}, path_regex)
    if rule is not None:
        rule["age"] = recipient
    else:
        rules.append({"path_regex": path_regex, "age": recipient})
    config["creation_rules"] = rules
    config_path.write_text(
        yaml.safe_dump(config, sort_keys=False, default_flow_style=False),
        encoding="utf-8",
    )


# ---------------------------------------------------------------------------
# Canonical key/value view — always via SOPS's own dotenv<->json codec.
# ---------------------------------------------------------------------------


def _canonical_from_ciphertext(
    ciphertext_path: Path, *, cwd: Path, env: Mapping[str, str]
) -> dict[str, str]:
    result = _run(
        [
            "sops",
            "decrypt",
            "--input-type",
            "dotenv",
            "--output-type",
            "json",
            str(ciphertext_path),
        ],
        cwd=cwd,
        env=env,
    )
    if not result.ok:
        raise SecretsError(f"sops decrypt failed: {result.stderr.strip()}")
    return json.loads(result.stdout or "{}")


def _canonical_from_plaintext(
    source_path: Path,
    *,
    cwd: Path,
    env: Mapping[str, str],
    filename_override: str = CIPHERTEXT_RELPATH,
) -> dict[str, str]:
    """SOPS's own parse of a plaintext dotenv file, as key/value JSON —
    never a hand-rolled dotenv parser. Round-trips the plaintext through a
    throwaway in-memory encrypt+decrypt so comments/blank-lines/quoting are
    canonicalized exactly the way the real ciphertext canonicalizes them.
    Nothing is written to disk by this call.
    """
    encrypted = _run(
        [
            "sops",
            "encrypt",
            "--input-type",
            "dotenv",
            "--output-type",
            "json",
            "--filename-override",
            filename_override,
            str(source_path),
        ],
        cwd=cwd,
        env=env,
    )
    if not encrypted.ok:
        raise SecretsError(
            f"sops encrypt (canonicalization) failed: {encrypted.stderr.strip()}"
        )
    decrypted = _run(
        ["sops", "decrypt", "--input-type", "json", "--output-type", "json"],
        cwd=cwd,
        env=env,
        input_text=encrypted.stdout,
    )
    if not decrypted.ok:
        raise SecretsError(
            f"sops decrypt (canonicalization) failed: {decrypted.stderr.strip()}"
        )
    return json.loads(decrypted.stdout or "{}")


def diff_keys(a: Mapping[str, str], b: Mapping[str, str]) -> list[str]:
    """Key NAMES where the two maps disagree — added, removed, or changed.
    Never returns a value, only the name of the key that differs."""
    keys = set(a) | set(b)
    return sorted(k for k in keys if a.get(k) != b.get(k))


# ---------------------------------------------------------------------------
# doctor
# ---------------------------------------------------------------------------


@dataclass
class DoctorCheck:
    label: str
    passed: bool
    detail: str = ""

    def render(self) -> str:
        status = "PASS" if self.passed else "FAIL"
        suffix = f" — {self.detail}" if self.detail else ""
        return f"[{status}] {self.label}{suffix}"


@dataclass
class DoctorReport:
    checks: list[DoctorCheck] = field(default_factory=list)

    @property
    def all_passed(self) -> bool:
        return all(c.passed for c in self.checks)

    def render(self) -> str:
        return "\n".join(c.render() for c in self.checks)


def run_doctor(
    *,
    repo_root: Path = REPO_ROOT,
    identity_path: Path | None = None,
    sops_config_path: Path | None = None,
    ciphertext_path: Path | None = None,
    base_env: Mapping[str, str] | None = None,
) -> DoctorReport:
    base_env = dict(base_env if base_env is not None else os.environ)
    identity_path = identity_path or resolve_identity_path(base_env)
    sops_config_path = sops_config_path or (repo_root / ".sops.yaml")
    ciphertext_path = ciphertext_path or (repo_root / CIPHERTEXT_RELPATH)

    checks: list[DoctorCheck] = []

    sops_bin = shutil.which("sops")
    if sops_bin:
        version_result = _run(["sops", "--version"], cwd=repo_root, env=base_env)
        version = (
            version_result.stdout.strip().splitlines()[0]
            if version_result.ok
            else "unknown version"
        )
        checks.append(DoctorCheck("sops on PATH", True, version))
    else:
        checks.append(DoctorCheck("sops on PATH", False, "not found"))

    age_bin = shutil.which("age")
    checks.append(
        DoctorCheck("age on PATH", age_bin is not None, age_bin or "not found")
    )

    age_keygen_bin = shutil.which("age-keygen")
    checks.append(
        DoctorCheck(
            "age-keygen on PATH",
            age_keygen_bin is not None,
            age_keygen_bin or "not found",
        )
    )

    identity_present = identity_path.exists()
    identity_readable = identity_present and os.access(identity_path, os.R_OK)
    checks.append(
        DoctorCheck(
            "identity file present and readable",
            identity_present and identity_readable,
            (
                str(identity_path)
                if identity_present
                else f"{identity_path} does not exist"
            ),
        )
    )

    config_present = sops_config_path.exists()
    checks.append(
        DoctorCheck(
            ".sops.yaml present",
            config_present,
            (
                str(sops_config_path)
                if config_present
                else f"{sops_config_path} does not exist"
            ),
        )
    )

    recipient_matches = False
    recipient_detail = "cannot check — identity or .sops.yaml missing"
    if identity_readable and config_present and age_keygen_bin:
        try:
            derived = derive_public_key(identity_path, cwd=repo_root, env=base_env)
            config = load_sops_config(sops_config_path)
            rule = find_creation_rule(config, PATH_REGEX)
            configured = rule.get("age") if rule else None
            if configured is None:
                recipient_detail = f"no creation_rules entry for {PATH_REGEX!r}"
            elif configured == derived:
                recipient_matches = True
                recipient_detail = derived
            else:
                recipient_detail = (
                    f".sops.yaml has {configured}, identity derives {derived}"
                )
        except SecretsError as exc:
            recipient_detail = str(exc)
    checks.append(
        DoctorCheck(
            "recipient in .sops.yaml matches the identity",
            recipient_matches,
            recipient_detail,
        )
    )

    ciphertext_present = ciphertext_path.exists()
    checks.append(
        DoctorCheck(
            "ciphertext artifact present",
            ciphertext_present,
            (
                str(ciphertext_path)
                if ciphertext_present
                else f"{ciphertext_path} does not exist"
            ),
        )
    )

    return DoctorReport(checks=checks)


# ---------------------------------------------------------------------------
# init
# ---------------------------------------------------------------------------


@dataclass
class InitResult:
    ok: bool
    identity_created: bool
    identity_path: Path
    recipient: str | None
    message: str


def run_init(
    *,
    repo_root: Path = REPO_ROOT,
    identity_path: Path | None = None,
    sops_config_path: Path | None = None,
    base_env: Mapping[str, str] | None = None,
) -> InitResult:
    base_env = dict(base_env if base_env is not None else os.environ)
    identity_path = identity_path or resolve_identity_path(base_env)
    sops_config_path = sops_config_path or (repo_root / ".sops.yaml")

    identity_created = False
    if identity_path.exists():
        message_lines = [
            f"age identity already present at {identity_path} — refusing to overwrite it."
        ]
    else:
        identity_path.parent.mkdir(parents=True, exist_ok=True)
        os.chmod(identity_path.parent, 0o700)
        env = _sops_env(identity_path, base_env)
        result = _run(["age-keygen", "-o", str(identity_path)], cwd=repo_root, env=env)
        if not result.ok:
            return InitResult(
                ok=False,
                identity_created=False,
                identity_path=identity_path,
                recipient=None,
                message=f"age-keygen failed: {result.stderr.strip()}",
            )
        os.chmod(identity_path, 0o600)
        identity_created = True
        message_lines = [f"created identity at {identity_path}"]

    env = _sops_env(identity_path, base_env)
    try:
        recipient = derive_public_key(identity_path, cwd=repo_root, env=env)
    except SecretsError as exc:
        return InitResult(
            ok=False,
            identity_created=identity_created,
            identity_path=identity_path,
            recipient=None,
            message=str(exc),
        )

    write_sops_config(sops_config_path, recipient, PATH_REGEX)
    message_lines.append(f"public recipient {recipient}")
    message_lines.append(f"identity path {identity_path}")
    message_lines.append(f"wrote {sops_config_path}")

    return InitResult(
        ok=True,
        identity_created=identity_created,
        identity_path=identity_path,
        recipient=recipient,
        message="\n".join(message_lines),
    )


# ---------------------------------------------------------------------------
# encrypt
# ---------------------------------------------------------------------------


@dataclass
class EncryptResult:
    ok: bool
    written: bool
    key_count: int
    message: str


def run_encrypt(
    *,
    repo_root: Path = REPO_ROOT,
    source_path: Path | None = None,
    ciphertext_path: Path | None = None,
    sops_config_path: Path | None = None,
    identity_path: Path | None = None,
    base_env: Mapping[str, str] | None = None,
) -> EncryptResult:
    base_env = dict(base_env if base_env is not None else os.environ)
    source_path = source_path or (repo_root / ".env")
    ciphertext_path = ciphertext_path or (repo_root / CIPHERTEXT_RELPATH)
    sops_config_path = sops_config_path or (repo_root / ".sops.yaml")
    identity_path = identity_path or resolve_identity_path(base_env)

    if not source_path.exists():
        return EncryptResult(
            ok=False,
            written=False,
            key_count=0,
            message=f"no source file at {source_path}",
        )
    if not identity_path.exists():
        return EncryptResult(
            ok=False,
            written=False,
            key_count=0,
            message=f"no age identity at {identity_path} — run 'secrets.py init' first",
        )
    if not sops_config_path.exists():
        return EncryptResult(
            ok=False,
            written=False,
            key_count=0,
            message=f"no {sops_config_path} — run 'secrets.py init' first",
        )

    env = _sops_env(identity_path, base_env)

    try:
        source_canonical = _canonical_from_plaintext(
            source_path, cwd=repo_root, env=env
        )
    except SecretsError as exc:
        return EncryptResult(ok=False, written=False, key_count=0, message=str(exc))

    if ciphertext_path.exists():
        try:
            existing_canonical = _canonical_from_ciphertext(
                ciphertext_path, cwd=repo_root, env=env
            )
        except SecretsError:
            existing_canonical = None
        if existing_canonical is not None and existing_canonical == source_canonical:
            return EncryptResult(
                ok=True,
                written=False,
                key_count=len(source_canonical),
                message=f"unchanged — {ciphertext_path} already matches {source_path} ({len(source_canonical)} keys)",
            )

    encrypted = _run(
        [
            "sops",
            "--input-type",
            "dotenv",
            "--output-type",
            "dotenv",
            "--filename-override",
            CIPHERTEXT_RELPATH,
            "--encrypt",
            str(source_path),
        ],
        cwd=repo_root,
        env=env,
    )
    if not encrypted.ok:
        return EncryptResult(
            ok=False,
            written=False,
            key_count=0,
            message=f"sops encrypt failed: {encrypted.stderr.strip()}",
        )

    ciphertext_path.parent.mkdir(parents=True, exist_ok=True)
    _atomic_write(ciphertext_path, encrypted.stdout, mode=0o644)

    return EncryptResult(
        ok=True,
        written=True,
        key_count=len(source_canonical),
        message=f"wrote {ciphertext_path} ({len(source_canonical)} keys)",
    )


# ---------------------------------------------------------------------------
# decrypt
# ---------------------------------------------------------------------------


@dataclass
class DecryptResult:
    ok: bool
    written: bool
    refused: bool
    differing_keys: list[str]
    message: str


def _atomic_write(path: Path, content: str, *, mode: int) -> None:
    """Write via a temp file + os.replace so an interrupted write can never
    leave a truncated file at ``path``."""
    fd, tmp_name = tempfile.mkstemp(
        dir=str(path.parent), prefix=f".{path.name}.", suffix=".tmp"
    )
    try:
        with os.fdopen(fd, "w", encoding="utf-8") as handle:
            handle.write(content)
        os.chmod(tmp_name, mode)
        os.replace(tmp_name, path)
    except BaseException:
        Path(tmp_name).unlink(missing_ok=True)
        raise


def run_decrypt(
    *,
    repo_root: Path = REPO_ROOT,
    dest_path: Path | None = None,
    ciphertext_path: Path | None = None,
    identity_path: Path | None = None,
    force: bool = False,
    base_env: Mapping[str, str] | None = None,
) -> DecryptResult:
    base_env = dict(base_env if base_env is not None else os.environ)
    dest_path = dest_path or (repo_root / ".env")
    ciphertext_path = ciphertext_path or (repo_root / CIPHERTEXT_RELPATH)
    identity_path = identity_path or resolve_identity_path(base_env)

    if not ciphertext_path.exists():
        return DecryptResult(
            ok=False,
            written=False,
            refused=False,
            differing_keys=[],
            message=f"no ciphertext at {ciphertext_path} — run 'secrets.py encrypt' first",
        )
    if not identity_path.exists():
        return DecryptResult(
            ok=False,
            written=False,
            refused=False,
            differing_keys=[],
            message=f"no age identity at {identity_path} — run 'secrets.py init' first",
        )

    env = _sops_env(identity_path, base_env)

    try:
        ciphertext_canonical = _canonical_from_ciphertext(
            ciphertext_path, cwd=repo_root, env=env
        )
    except SecretsError as exc:
        return DecryptResult(
            ok=False, written=False, refused=False, differing_keys=[], message=str(exc)
        )

    if dest_path.exists() and not force:
        try:
            dest_canonical = _canonical_from_plaintext(
                dest_path, cwd=repo_root, env=env
            )
        except SecretsError as exc:
            return DecryptResult(
                ok=False,
                written=False,
                refused=False,
                differing_keys=[],
                message=str(exc),
            )
        differing = diff_keys(dest_canonical, ciphertext_canonical)
        if differing:
            return DecryptResult(
                ok=False,
                written=False,
                refused=True,
                differing_keys=differing,
                message=(
                    f"{dest_path} differs from {ciphertext_path} in keys: "
                    f"{', '.join(differing)} — refusing without --force"
                ),
            )

    decrypted = _run(
        [
            "sops",
            "--input-type",
            "dotenv",
            "--output-type",
            "dotenv",
            "--decrypt",
            str(ciphertext_path),
        ],
        cwd=repo_root,
        env=env,
    )
    if not decrypted.ok:
        return DecryptResult(
            ok=False,
            written=False,
            refused=False,
            differing_keys=[],
            message=f"sops decrypt failed: {decrypted.stderr.strip()}",
        )

    dest_path.parent.mkdir(parents=True, exist_ok=True)
    _atomic_write(dest_path, decrypted.stdout, mode=0o600)

    return DecryptResult(
        ok=True,
        written=True,
        refused=False,
        differing_keys=[],
        message=f"wrote {dest_path} ({len(ciphertext_canonical)} keys)",
    )


# ---------------------------------------------------------------------------
# status
# ---------------------------------------------------------------------------


@dataclass
class StatusReport:
    key_present: bool
    ciphertext_present: bool
    source_present: bool
    in_sync: bool | None
    key_count: int
    differing_keys: list[str]
    error: str | None = None

    def to_dict(self) -> dict:
        return {
            "key_present": self.key_present,
            "ciphertext_present": self.ciphertext_present,
            "source_present": self.source_present,
            "in_sync": self.in_sync,
            "key_count": self.key_count,
            "differing_keys": self.differing_keys,
            "error": self.error,
        }

    def render(self) -> str:
        lines = [
            f"age identity present: {self.key_present}",
            f"ciphertext present: {self.ciphertext_present}",
            f"plaintext source present: {self.source_present}",
        ]
        if self.error:
            lines.append(f"error: {self.error}")
        elif self.in_sync is None:
            lines.append("in sync: unknown (missing identity/ciphertext/source)")
        else:
            lines.append(f"in sync: {self.in_sync}")
            lines.append(f"key count: {self.key_count}")
            if not self.in_sync:
                lines.append(f"differing keys: {', '.join(self.differing_keys)}")
        return "\n".join(lines)


def run_status(
    *,
    repo_root: Path = REPO_ROOT,
    source_path: Path | None = None,
    ciphertext_path: Path | None = None,
    identity_path: Path | None = None,
    base_env: Mapping[str, str] | None = None,
) -> StatusReport:
    base_env = dict(base_env if base_env is not None else os.environ)
    source_path = source_path or (repo_root / ".env")
    ciphertext_path = ciphertext_path or (repo_root / CIPHERTEXT_RELPATH)
    identity_path = identity_path or resolve_identity_path(base_env)

    key_present = identity_path.exists()
    ciphertext_present = ciphertext_path.exists()
    source_present = source_path.exists()

    if not (key_present and ciphertext_present and source_present):
        return StatusReport(
            key_present=key_present,
            ciphertext_present=ciphertext_present,
            source_present=source_present,
            in_sync=None,
            key_count=0,
            differing_keys=[],
        )

    env = _sops_env(identity_path, base_env)
    try:
        ciphertext_canonical = _canonical_from_ciphertext(
            ciphertext_path, cwd=repo_root, env=env
        )
        source_canonical = _canonical_from_plaintext(
            source_path, cwd=repo_root, env=env
        )
    except SecretsError as exc:
        return StatusReport(
            key_present=key_present,
            ciphertext_present=ciphertext_present,
            source_present=source_present,
            in_sync=None,
            key_count=0,
            differing_keys=[],
            error=str(exc),
        )

    differing = diff_keys(source_canonical, ciphertext_canonical)
    return StatusReport(
        key_present=key_present,
        ciphertext_present=ciphertext_present,
        source_present=source_present,
        in_sync=not differing,
        key_count=len(ciphertext_canonical),
        differing_keys=differing,
    )


# ---------------------------------------------------------------------------
# argparse CLI
# ---------------------------------------------------------------------------


def build_parser() -> argparse.ArgumentParser:
    parser = argparse.ArgumentParser(
        prog="secrets",
        description="SOPS + age lifecycle for the repo's encrypted secrets.",
    )
    sub = parser.add_subparsers(dest="command", required=True)

    def _add_common(p: argparse.ArgumentParser) -> None:
        p.add_argument(
            "--repo-root",
            type=Path,
            default=REPO_ROOT,
            help="Root .env/.sops.yaml/secrets/ resolve against.",
        )
        p.add_argument(
            "--identity",
            type=Path,
            default=None,
            help="age identity path (default: $SOPS_AGE_KEY_FILE or ~/.config/sops/age/keys.txt).",
        )

    doctor = sub.add_parser("doctor", help="Preflight: everything needed to decrypt.")
    _add_common(doctor)
    doctor.add_argument("--sops-config", type=Path, default=None)
    doctor.add_argument("--ciphertext", type=Path, default=None)

    init = sub.add_parser(
        "init", help="Generate an age identity + .sops.yaml if absent."
    )
    _add_common(init)
    init.add_argument("--sops-config", type=Path, default=None)

    encrypt = sub.add_parser(
        "encrypt", help="Encrypt the plaintext .env into the tracked ciphertext."
    )
    _add_common(encrypt)
    encrypt.add_argument(
        "--source",
        type=Path,
        default=None,
        help="Plaintext source (default: <repo-root>/.env).",
    )
    encrypt.add_argument("--sops-config", type=Path, default=None)
    encrypt.add_argument("--ciphertext", type=Path, default=None)

    decrypt = sub.add_parser(
        "decrypt", help="Decrypt the tracked ciphertext to a plaintext .env."
    )
    _add_common(decrypt)
    decrypt.add_argument(
        "--dest",
        type=Path,
        default=None,
        help="Plaintext destination (default: <repo-root>/.env).",
    )
    decrypt.add_argument("--ciphertext", type=Path, default=None)
    decrypt.add_argument(
        "--force", action="store_true", help="Overwrite a differing destination."
    )

    status = sub.add_parser(
        "status", help="Report sync state between plaintext and ciphertext."
    )
    _add_common(status)
    status.add_argument("--source", type=Path, default=None)
    status.add_argument("--ciphertext", type=Path, default=None)
    status.add_argument(
        "--json", action="store_true", help="Emit machine-readable JSON."
    )

    return parser


def cmd_doctor(args: argparse.Namespace) -> int:
    report = run_doctor(
        repo_root=args.repo_root,
        identity_path=args.identity,
        sops_config_path=args.sops_config,
        ciphertext_path=args.ciphertext,
    )
    print(report.render())
    return EXIT_OK if report.all_passed else EXIT_PRECONDITION


def cmd_init(args: argparse.Namespace) -> int:
    result = run_init(
        repo_root=args.repo_root,
        identity_path=args.identity,
        sops_config_path=args.sops_config,
    )
    stream = sys.stdout if result.ok else sys.stderr
    print(result.message, file=stream)
    return EXIT_OK if result.ok else EXIT_FAIL


def cmd_encrypt(args: argparse.Namespace) -> int:
    result = run_encrypt(
        repo_root=args.repo_root,
        source_path=args.source,
        ciphertext_path=args.ciphertext,
        sops_config_path=args.sops_config,
        identity_path=args.identity,
    )
    if not result.ok:
        print(f"secrets encrypt: {result.message}", file=sys.stderr)
        return (
            EXIT_FAIL if "sops encrypt failed" in result.message else EXIT_PRECONDITION
        )
    print(result.message)
    return EXIT_OK


def cmd_decrypt(args: argparse.Namespace) -> int:
    result = run_decrypt(
        repo_root=args.repo_root,
        dest_path=args.dest,
        ciphertext_path=args.ciphertext,
        identity_path=args.identity,
        force=args.force,
    )
    if not result.ok:
        print(f"secrets decrypt: {result.message}", file=sys.stderr)
        if result.refused:
            return EXIT_FAIL
        return (
            EXIT_PRECONDITION
            if "no ciphertext" in result.message or "no age identity" in result.message
            else EXIT_FAIL
        )
    print(result.message)
    return EXIT_OK


def cmd_status(args: argparse.Namespace) -> int:
    report = run_status(
        repo_root=args.repo_root,
        source_path=args.source,
        ciphertext_path=args.ciphertext,
        identity_path=args.identity,
    )
    if args.json:
        print(json.dumps(report.to_dict(), indent=2, sort_keys=True))
    else:
        print(report.render())
    if report.error:
        return EXIT_FAIL
    if report.in_sync is None:
        return EXIT_PRECONDITION
    return EXIT_OK if report.in_sync else EXIT_FAIL


def main(argv: list[str] | None = None) -> int:
    args = build_parser().parse_args(argv)
    if args.command == "doctor":
        return cmd_doctor(args)
    if args.command == "init":
        return cmd_init(args)
    if args.command == "encrypt":
        return cmd_encrypt(args)
    if args.command == "decrypt":
        return cmd_decrypt(args)
    if args.command == "status":
        return cmd_status(args)
    # argparse's required=True makes this unreachable; returning an error
    # rather than falling through keeps an added subcommand from silently
    # exiting 0 before it is wired.
    print(f"secrets: unhandled command {args.command!r}", file=sys.stderr)
    return EXIT_FAIL


if __name__ == "__main__":
    sys.exit(main())