agent-orchestrator/secrets.py

#!/usr/bin/env python3
"""The one place secrets live on an orchestrator host: a sops+age encrypted store.

WHY: credentials used to sit in plaintext all over the box — passwords baked into git
remote URLs (`https://user:pass@host/...`, which `git remote -v` happily prints), API keys
in .env files, a private key at mode 0644. Anything in a repo is one `git add -A` away from
a push. So: ONE encrypted file, OUTSIDE every git tree, and a helper every project uses.

    /secrets/store.yaml            sops+age ciphertext (0600) — values our code reads
    /secrets/<project>/            PROJECT-SCOPED secrets (0600): everything belonging to one
                                   project lives together, e.g. /secrets/lichen/,
                                   /secrets/b1/, /secrets/notplants-orchestrator/
    /secrets/files/                CROSS-PROJECT files (0600) SYMLINKED from the fixed path a
                                   third party insists on: ~/.ssh keys, a systemd
                                   EnvironmentFile, nix authKeyFile, a TLS keypair
    ~/.config/sops/age/keys.txt    the age private key, 0600

PROJECT SECRETS GO IN /secrets/<project>/ (operator, 2026-08-20). If a secret belongs to one
project, it goes in that project's directory — not in files/, and not with the project name
baked into the filename. `/secrets/lichen/test-pds.env`, not `/secrets/files/lichen-test-pds.env`.
Reserve files/ for things genuinely shared across projects.

WHY: a flat directory forces every name to carry its own scope, which nobody does consistently,
and then nobody can answer "what does this project hold?" or "what do I revoke if this project is
compromised?" without grepping. A directory answers both by listing. Put a README.md in the
project directory saying what each file is, what consumes it, and what breaks if it is lost —
the next person to read it will be doing so under time pressure.

The symlink rule is unchanged and applies the same way: a consumer that insists on a fixed path
gets a SYMLINK into /secrets/<project>/, so the file still exists exactly once.

One home per secret: a value is in the store OR a file under /secrets, never both.
/secrets is outside every git tree — not a repo, no remote — and outside /srv, which agents
grep and walk constantly.

USAGE (library):
    from secrets import get, get_group
    cookie = get("tangled.cookie")
    env    = get_group("cc_ci_testenv")        # dict, e.g. to build an env

USAGE (CLI):
    python3 engine/secrets.py list                    # group/key names only, never values
    python3 engine/secrets.py get tangled.cookie      # value to stdout (careful in logs)

NO SECOND COPIES. A secret is never written to a second file "so something can read it" —
copies drift, get committed, and widen what a stray `grep` or an attacker finds. A consumer
that insists on a path gets a SYMLINK into /secrets/files (see above), so the file still
exists exactly once. For a one-off, inject at run time and leave nothing behind:

    secrets.py exec-env <group> -- some-command       # group as env vars, no file
    secrets.py with-file <group.key> -- cmd -i {}     # 0600 file in a private tmpdir,
                                                      # deleted when the command exits

Careful with symlinks: an app that rewrites its own credential file (an OAuth refresh writing
auth.json via write-temp+rename) REPLACES the symlink with a regular file and silently splits
the home again. Before symlinking, ask whether the owner ever writes it back.

ADDING A SECRET:  sops /secrets/store.yaml   (opens decrypted in $EDITOR, re-encrypts on save)
"""
import json, os, subprocess, sys, pathlib, tempfile, shutil

STORE = os.environ.get("AO_SECRETS_STORE", "/secrets/store.yaml")
AGE_KEY = os.environ.get("SOPS_AGE_KEY_FILE", os.path.expanduser("~/.config/sops/age/keys.txt"))


def _sops_bin():
    """Resolve sops. It is on PATH under the systemd unit, but not always in an
    interactive shell — fall back to the NixOS system profile before failing."""
    return (shutil.which("sops")
            or next((p for p in ("/run/current-system/sw/bin/sops",
                                 "/run/wrappers/bin/sops") if os.path.exists(p)), None)
            or "sops")


def _load():
    """Decrypt the store. Fails loudly: a silent empty dict would look like 'no secrets'."""
    if not pathlib.Path(STORE).exists():
        sys.exit(f"no secret store at {STORE} — see engine/README.md (Secrets)")
    env = {**os.environ, "SOPS_AGE_KEY_FILE": AGE_KEY}
    r = subprocess.run([_sops_bin(), "-d", "--output-type", "json", STORE],
                       capture_output=True, text=True, env=env)
    if r.returncode != 0:
        sys.exit(f"cannot decrypt {STORE} (age key at {AGE_KEY}?): {r.stderr.strip()[:300]}")
    return json.loads(r.stdout)


def get(dotted, default=None):
    """get('group.key') -> value. Missing key returns default (None) rather than raising."""
    cur = _load()
    for part in dotted.split("."):
        if not isinstance(cur, dict) or part not in cur:
            return default
        cur = cur[part]
    return cur


def get_group(name):
    """get_group('cc_ci_testenv') -> dict of that group (empty dict if absent)."""
    return _load().get(name, {})


def exec_env(group, argv):
    """Run argv with `group`'s keys added to the environment. Nothing touches the disk."""
    vals = get_group(group)
    if not vals:
        sys.exit(f"no group {group!r} in the store (secrets.py list)")
    env = {**os.environ, **{k: str(v) for k, v in vals.items()}}
    return subprocess.call(argv, env=env)


def with_file(dotted, argv):
    """Run argv with `{}` replaced by a 0600 temp file holding the value.

    The file lives in a private 0700 dir and is removed when the command exits — so a consumer
    that insists on a path (ssh -i, a TLS key) never leaves a lasting second copy of the secret.
    """
    val = get(dotted)
    if val is None:
        sys.exit(f"no such key: {dotted}")
    d = tempfile.mkdtemp(prefix="ao-secret-")   # mkdtemp is 0700
    try:
        p = pathlib.Path(d) / dotted.split(".")[-1]
        p.write_text(val if isinstance(val, str) else json.dumps(val))
        p.chmod(0o600)
        return subprocess.call([a.replace("{}", str(p)) for a in argv])
    finally:
        shutil.rmtree(d, ignore_errors=True)


def main():
    argv = sys.argv[1:]
    if not argv or argv[0] in ("-h", "--help"):
        print(__doc__)
        return
    cmd = argv[0]
    if cmd == "list":
        for g, v in _load().items():
            print(f"{g}: {', '.join(v) if isinstance(v, dict) else '<value>'}")
    elif cmd == "get":
        if len(argv) < 2:
            sys.exit("usage: secrets.py get <group.key>")
        v = get(argv[1])
        if v is None:
            sys.exit(f"no such key: {argv[1]}")
        print(v if isinstance(v, str) else json.dumps(v))
    elif cmd in ("exec-env", "with-file"):
        if "--" not in argv:
            sys.exit(f"usage: secrets.py {cmd} <name> -- <command...>")
        i = argv.index("--")
        if i != 2:
            sys.exit(f"usage: secrets.py {cmd} <name> -- <command...>")
        run = exec_env if cmd == "exec-env" else with_file
        sys.exit(run(argv[1], argv[i + 1:]))
    else:
        sys.exit(f"unknown command {cmd!r} — list | get | exec-env | with-file")


if __name__ == "__main__":
    main()