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.

    store:    /secrets/store.yaml          sops+age ciphertext, mode 0600
    age key:  ~/.config/sops/age/keys.txt      the ONLY plaintext secret on disk, 0600
    outside git by construction — /secrets is not a repo and has no remote.

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)
    python3 engine/secrets.py materialize <name>      # write a runtime file a consumer needs

NO SECOND COPIES. A secret must not be written to a second file "so something can read it" —
copies drift from the store, get committed, and multiply what an attacker (or a careless
`grep`) can find. Consumers read the store: our own code imports this module; anything else
gets the value injected at RUN TIME and nothing is left at rest.

    secrets.py exec-env cc_ci_testenv -- some-command    # group as env vars, no file
    secrets.py with-file ssh_keys.tangled-ed25519 -- ssh -i {} host   # 0600 file in a private
                                                        # tmpdir, deleted when the command exits

For systemd, wrap ExecStart in `exec-env` instead of using an EnvironmentFile — same effect,
no plaintext on disk. The few OS-level paths that genuinely cannot be taught this (nix's
`authKeyFile`, sshd host keys) are the exception, and are noted in engine/README.md.

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 _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", "-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()