esh-pfi-infrastructure/services/secrets-broker/secret

#!/usr/bin/env python3
"""secret — fleet credential store & retrieve over Vaultwarden (bw-backed).

Stores each secret as an item in the `infra-ops` org's Default collection (so the
operator's primary account, an org member, sees it too), organised by folder +
a `<host>/<stack>/<file>` name convention. Text secrets (env files, tokens) live
in the item note; binary secrets (keys/certs) are base64'd into a hidden field.

Auth bootstraps from ~/.config/secrets-broker/bootstrap.env (0600): apikey login
+ master-password unlock -> per-invocation session. Secret VALUES are never printed
except by an explicit `get`.

Commands:
  secret put  <name> (--file P | --stdin) [--folder C] [--field k=v]...
  secret get  <name> [--field F] [--file OUT]
  secret list [--prefix P]
  secret backfill --host H [--dry-run]     # enumerate a host's .env/env.sh, upsert
"""
import argparse
import base64
import hashlib
import json
import os
import shlex
import subprocess
import sys
import time
from pathlib import Path

BW = os.environ.get("BW_BIN", str(Path.home() / ".local/bin/bw"))
CFG = Path(os.environ.get("SECRET_CFG_DIR", str(Path.home() / ".config/secrets-broker")))
BOOTSTRAP = CFG / "bootstrap.env"
SERVER = "https://vaultwarden.phasefinal.com"
ORG_ID = "d30c6b58-773c-4e39-916e-8e33ca7f8b81"
COLLECTION_ID = "b829376a-9db1-4091-a4ae-9a36132ad4c2"


def die(msg):
    print(f"secret: {msg}", file=sys.stderr)
    sys.exit(1)


def load_env():
    if not BOOTSTRAP.is_file():
        die(f"no bootstrap creds at {BOOTSTRAP}")
    if oct(BOOTSTRAP.stat().st_mode)[-3:] not in ("600", "400"):
        die(f"{BOOTSTRAP} must be 0600")
    creds = {}
    for line in BOOTSTRAP.read_text().splitlines():
        for k in ("BW_CLIENTID", "BW_CLIENTSECRET", "RBW_MASTER_PW"):
            if line.startswith(k + "="):
                creds[k] = line.split("=", 1)[1]
    env = dict(os.environ)
    env["BW_CLIENTID"] = creds.get("BW_CLIENTID", "")
    env["BW_CLIENTSECRET"] = creds.get("BW_CLIENTSECRET", "")
    env["BW_PASSWORD"] = creds.get("RBW_MASTER_PW", "")
    return env


def bw(args, env, session=None, stdin=None, check=True):
    cmd = [BW] + args + (["--session", session] if session else [])
    r = subprocess.run(cmd, env=env, input=stdin, capture_output=True)
    if check and r.returncode != 0:
        die(f"bw {args[0]} failed: {r.stderr.decode(errors='replace').strip()}")
    return r


def status(env):
    r = subprocess.run([BW, "status"], env=env, capture_output=True)
    try:
        return json.loads(r.stdout.decode()).get("status", "unauthenticated")
    except Exception:
        return "unauthenticated"


def session(env):
    if status(env) == "unauthenticated":
        subprocess.run([BW, "config", "server", SERVER], env=env, capture_output=True)
        bw(["login", "--apikey"], env)
    s = bw(["unlock", "--passwordenv", "BW_PASSWORD", "--raw"], env).stdout.decode().strip()
    return s or die("unlock produced no session")


def encode(obj, env):
    return bw(["encode"], env, stdin=json.dumps(obj).encode()).stdout


def find(env, s, name):
    r = bw(["list", "items", "--search", name], env, s)
    for it in json.loads(r.stdout.decode() or "[]"):
        if it.get("name") == name:
            return it
    return None


def folder_id(env, s, cls):
    if not cls:
        return None
    for f in json.loads(bw(["list", "folders"], env, s).stdout.decode() or "[]"):
        if f.get("name") == cls:
            return f["id"]
    r = bw(["create", "folder"], env, s, stdin=encode({"name": cls}, env))
    return json.loads(r.stdout.decode())["id"]


def cmd_put(a):
    env = load_env()
    s = session(env)
    if a.file:
        data = Path(a.file).read_bytes()
    elif a.stdin:
        data = sys.stdin.buffer.read()
    else:
        die("put needs --file or --stdin")
    sha = hashlib.sha256(data).hexdigest()
    try:
        notes, binary = data.decode("utf-8"), False
    except UnicodeDecodeError:
        notes, binary = "<binary — value in the content_b64 field>", True
    fields = [
        {"name": "sha256", "value": sha, "type": 0},
        {"name": "synced_at", "value": time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime()), "type": 0},
    ]
    for kv in (a.field or []):
        k, v = kv.split("=", 1)
        fields.append({"name": k, "value": v, "type": 0})
    if binary:
        fields.append({"name": "content_b64", "value": base64.b64encode(data).decode(), "type": 1})
    item = {
        "type": 2, "name": a.name, "notes": notes,
        "organizationId": ORG_ID, "collectionIds": [COLLECTION_ID],
        "folderId": folder_id(env, s, a.folder), "fields": fields,
        "secureNote": {"type": 0},
    }
    existing = find(env, s, a.name)
    if existing:
        item["id"] = existing["id"]
        bw(["edit", "item", existing["id"]], env, s, stdin=encode(item, env))
        action = "updated"
    else:
        bw(["create", "item"], env, s, stdin=encode(item, env))
        action = "created"
    print(f"{action}: {a.name}  (sha256 {sha[:12]}, {len(data)} bytes{', binary' if binary else ''})")


def cmd_get(a):
    env = load_env()
    s = session(env)
    it = find(env, s, a.name) or die(f"not found: {a.name}")
    full = json.loads(bw(["get", "item", it["id"]], env, s).stdout.decode())
    if a.field:
        for f in full.get("fields") or []:
            if f["name"] == a.field:
                if a.field == "content_b64" and a.file:
                    Path(a.file).write_bytes(base64.b64decode(f["value"]))
                    os.chmod(a.file, 0o600)
                    return print(f"wrote {a.file} (0600)")
                return print(f["value"])
        die(f"no field '{a.field}' on {a.name}")
    content = full.get("notes") or ""
    if a.file:
        Path(a.file).write_text(content)
        os.chmod(a.file, 0o600)
        print(f"wrote {a.file} (0600)")
    else:
        sys.stdout.write(content if content.endswith("\n") else content + "\n")


def cmd_list(a):
    env = load_env()
    s = session(env)
    items = json.loads(bw(["list", "items"], env, s).stdout.decode() or "[]")
    n = 0
    for it in sorted(items, key=lambda x: x.get("name", "")):
        name = it.get("name", "")
        if a.prefix and not name.startswith(a.prefix):
            continue
        meta = {f["name"]: f["value"] for f in (it.get("fields") or [])}
        print(f"{name}\t{meta.get('synced_at', '?')}\t{meta.get('sha256', '')[:12]}")
        n += 1
    print(f"# {n} item(s)", file=sys.stderr)


def _host_secret_files(host):
    find_cmd = (r"find /opt/docker -maxdepth 4 \( -name .env -o -name env.sh \) "
                r"! -name '*.example' -type f 2>/dev/null")
    r = subprocess.run(["ssh", host, find_cmd], capture_output=True, text=True)
    return [p for p in r.stdout.split("\n") if p.strip()]


def _name_for(host, path):
    parts = Path(path).parts  # e.g. /opt/docker/compose/gitea/.env
    stack = parts[-2] if len(parts) >= 2 else "root"
    return f"{host}/{stack}/{Path(path).name}"


def cmd_backfill(a):
    host = a.host
    files = _host_secret_files(host)
    if not files:
        return print(f"{host}: no .env/env.sh found under /opt/docker")
    env = load_env()
    s = None if a.dry_run else session(env)
    print(f"{host}: {len(files)} secret file(s)")
    for path in files:
        name = _name_for(host, path)
        data = subprocess.run(["ssh", host, f"cat {shlex.quote(path)}"], capture_output=True).stdout
        sha = hashlib.sha256(data).hexdigest()
        if a.dry_run:
            print(f"  WOULD store  {name}\t({len(data)} bytes, sha {sha[:12]})  <- {path}")
            continue
        try:
            notes = data.decode("utf-8")
        except UnicodeDecodeError:
            notes = "<binary>"
        fields = [
            {"name": "sha256", "value": sha, "type": 0},
            {"name": "source_host", "value": host, "type": 0},
            {"name": "source_path", "value": path, "type": 0},
            {"name": "synced_at", "value": time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime()), "type": 0},
        ]
        item = {"type": 2, "name": name, "notes": notes, "organizationId": ORG_ID,
                "collectionIds": [COLLECTION_ID], "folderId": folder_id(env, s, "hosts"),
                "fields": fields, "secureNote": {"type": 0}}
        existing = find(env, s, name)
        if existing:
            item["id"] = existing["id"]
            bw(["edit", "item", existing["id"]], env, s, stdin=encode(item, env))
            verb = "updated"
        else:
            bw(["create", "item"], env, s, stdin=encode(item, env))
            verb = "stored "
        # round-trip verify
        back = json.loads(bw(["get", "item", find(env, s, name)["id"]], env, s).stdout.decode())
        ok = hashlib.sha256((back.get("notes") or "").encode()).hexdigest() == sha
        print(f"  {verb} {name}\t({len(data)}b) verify={'OK' if ok else 'MISMATCH'}")


def main():
    p = argparse.ArgumentParser(prog="secret", description="fleet credential store over Vaultwarden")
    sub = p.add_subparsers(dest="cmd", required=True)
    pp = sub.add_parser("put"); pp.add_argument("name")
    pp.add_argument("--file"); pp.add_argument("--stdin", action="store_true")
    pp.add_argument("--folder"); pp.add_argument("--field", action="append")
    pp.set_defaults(fn=cmd_put)
    pg = sub.add_parser("get"); pg.add_argument("name")
    pg.add_argument("--field"); pg.add_argument("--file"); pg.set_defaults(fn=cmd_get)
    pl = sub.add_parser("list"); pl.add_argument("--prefix"); pl.set_defaults(fn=cmd_list)
    pb = sub.add_parser("backfill"); pb.add_argument("--host", required=True)
    pb.add_argument("--dry-run", action="store_true"); pb.set_defaults(fn=cmd_backfill)
    a = p.parse_args()
    a.fn(a)


if __name__ == "__main__":
    main()