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

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

Serves the CC sessions on this dev box. 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 (named for the hostname) + a
`<host>/<path>` name convention. Small text -> item note; small binary -> base64
hidden field; anything over the note cap -> a bw attachment. Metadata fields carry
sha256 + source for drift detection.

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 rm   <name>                       # delete an item (soft-delete to trash)
  secret backfill [--dry-run]              # scan THIS box's local secrets, upsert each
"""
import argparse
import base64
import fcntl
import glob
import hashlib
import json
import os
import shutil
import socket
import subprocess
import sys
import tempfile
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"
NOTE_MAX = 6000  # plaintext bytes; larger -> attachment (Vaultwarden note cap ~10000 encrypted chars)


class BwError(Exception):
    pass


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


BW_LOCK = Path(os.environ.get("SECRET_LOCK", "/tmp/.secrets-broker-bw.lock"))


def bw(args, env, session=None, stdin=None, check=True):
    """Run one `bw` command. Serialisation is at the COMMAND level — see `main`.

    ⚠ `bw` is NOT concurrency-safe against a shared session. Measured 2026-09-15:
    four parallel `secret get` calls for distinct items returned **empty strings
    with exit code 0** — 0 of 4 succeeded here, 1 of 4 in svos-dev's run that
    reported it. No error, no timeout, no diagnostic. Sequentially all four are
    correct. That is the worst possible failure shape for a credential tool: a
    caller treating "empty optional secret" as "not configured" degrades silently
    and never knows (it cost SVOS its ability to page the operator, and nothing
    said so).

    The lock makes concurrent callers queue instead of race, so callers no longer
    have to know. It is held for the whole `bw` call — vault reads are ~17s, so
    parallelism was never buying much anyway. Paired with the empty-value guard in
    `cmd_get`: the lock prevents the race, the guard makes any future silent-empty
    loud rather than invisible.
    """
    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:
        raise BwError(f"bw {args[0]}: {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):
    raw = bw(["list", "items", "--search", name], env, s).stdout.decode().strip()
    # ⚠ Empty stdout is a FAILED listing, not an empty result set. The old code said
    # `or "[]"`, which turned a broken read into a confident "no such secret" — the
    # same silent-wrong-answer shape as the empty-value bug below. bw returns nothing
    # here when its session has been invalidated underneath it.
    if not raw:
        raise BwError(
            "`bw list items` returned NOTHING (not an empty list). The vault session "
            "was probably invalidated mid-call — most often by a concurrent `secret` "
            "invocation. Retry sequentially."
        )
    for it in json.loads(raw):
        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 store_secret(env, s, name, data, folder_name, extra_fields):
    """Upsert a secret item; route by size: note / base64-field / attachment.
    Returns (item_id, sha256, mode)."""
    sha = hashlib.sha256(data).hexdigest()
    try:
        text, is_text = data.decode("utf-8"), True
    except UnicodeDecodeError:
        text, is_text = None, False
    large = len(data) > NOTE_MAX
    fields = [{"name": "sha256", "value": sha, "type": 0},
              {"name": "synced_at", "value": time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime()), "type": 0}]
    fields += extra_fields
    attach_name = None
    if large:
        attach_name = name.split("/")[-1] or "secret"
        notes = f"<{len(data)} bytes — stored as attachment '{attach_name}'>"
        fields.append({"name": "storage", "value": "attachment", "type": 0})
    elif is_text:
        notes = text
    else:
        notes = "<binary — value in content_b64 field>"
        fields.append({"name": "content_b64", "value": base64.b64encode(data).decode(), "type": 1})
    item = {"type": 2, "name": name, "notes": notes, "organizationId": ORG_ID,
            "collectionIds": [COLLECTION_ID], "folderId": folder_id(env, s, folder_name),
            "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))
        item_id = existing["id"]
        full = json.loads(bw(["get", "item", item_id], env, s).stdout.decode())
        for att in full.get("attachments") or []:   # clear stale attachments (no dupes on re-store)
            bw(["delete", "attachment", att["id"], "--itemid", item_id], env, s)
    else:
        item_id = json.loads(bw(["create", "item"], env, s, stdin=encode(item, env)).stdout.decode())["id"]
    if large:
        d = tempfile.mkdtemp()
        try:
            fp = os.path.join(d, attach_name)
            Path(fp).write_bytes(data)
            bw(["create", "attachment", "--itemid", item_id, "--file", fp], env, s)
        finally:
            shutil.rmtree(d, ignore_errors=True)
    return item_id, sha, ("attachment" if large else ("field" if not is_text else "note"))


def read_secret(env, s, name):
    """Return the stored secret bytes for `name` (attachment / base64-field / note)."""
    it = find(env, s, name)
    if not it:
        return None
    full = json.loads(bw(["get", "item", it["id"]], env, s).stdout.decode())
    atts = full.get("attachments") or []
    if atts:
        return bw(["get", "attachment", atts[0]["fileName"], "--itemid", full["id"], "--raw"], env, s).stdout
    for f in full.get("fields") or []:
        if f["name"] == "content_b64":
            return base64.b64decode(f["value"])
    return (full.get("notes") or "").encode()


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")
    extra = []
    for kv in (a.field or []):
        k, v = kv.split("=", 1)
        extra.append({"name": k, "value": v, "type": 0})
    # namespace heads-up: warn (don't block) when this opens a brand-new top-level
    # namespace — catches a typo'd/missing host prefix at store time instead of
    # letting the item hide under a prefix nobody searches.
    if not find(env, s, a.name):
        top = a.name.split("/")[0]
        tops = {i["name"].split("/")[0] for i in json.loads(bw(["list", "items"], env, s).stdout.decode() or "[]")}
        if top not in tops:
            print(f"secret: note — '{a.name}' opens a NEW top-level namespace '{top}/' "
                  f"(existing: {', '.join(sorted(tops)) or 'none'}). If you meant to host-prefix it, "
                  f"'secret rm {a.name}' and re-put as {socket.gethostname()}/...", file=sys.stderr)
    _, sha, mode = store_secret(env, s, a.name, data, a.folder, extra)
    print(f"stored: {a.name}  (sha256 {sha[:12]}, {len(data)} bytes, {mode})")


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:
                return print(f["value"])
        die(f"no field '{a.field}' on {a.name}")
    data = read_secret(env, s, a.name)
    # ⚠ An EMPTY value with exit 0 is the silent-failure surface — see bw() above.
    # A stored secret is never legitimately zero-length, so refuse rather than hand
    # a caller "" that it will read as "not configured".
    if not data:
        die(f"{a.name} resolved to an EMPTY value. A stored secret is never empty, so "
            "this is a failed read, not an unset secret. Most likely cause: a "
            "concurrent `secret` call — bw is not safe under parallel invocation and "
            "returns empty with success. Retry sequentially; if it persists the item "
            "has no note/field/attachment body.")
    if a.file:
        Path(a.file).write_bytes(data)
        os.chmod(a.file, 0o600)
        print(f"wrote {a.file} (0600)")
    else:
        sys.stdout.buffer.write(data if data.endswith(b"\n") else data + b"\n")


def cmd_rm(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())
    sha = next((f["value"][:12] for f in (full.get("fields") or []) if f["name"] == "sha256"), "?")
    bw(["delete", "item", it["id"]], env, s)   # soft-delete to trash (recoverable in the vault)
    print(f"deleted: {a.name}  (sha256 {sha}, id {it['id']}) -> trash")


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)


LOCAL_GLOBS = [
    "development/*/env.sh", "development/*/*/env.sh",
    "development/*/.env", "development/*/*/.env",
    ".config/secrets/env.sh",
    ".config/*/*.env", ".config/*/*token*", ".config/*/*-key*",
    ".config/*/*.pem", ".config/*/*.key", ".config/*/secrets.env", ".config/*/cred*",
]
LOCAL_EXCLUDE = ("secrets-broker/bootstrap.env", ".example", "/node_modules/",
                 "/.cargo/", "/.cache/", "/AIPA-Data/")


def _local_secret_files():
    home = str(Path.home())
    hits = set()
    for g in LOCAL_GLOBS:
        hits.update(glob.glob(f"{home}/{g}"))
    return [p for p in sorted(hits)
            if not any(e in p for e in LOCAL_EXCLUDE) and Path(p).is_file()]


def _local_name(path):
    home = str(Path.home())
    rel = path[len(home) + 1:] if path.startswith(home + "/") else path.lstrip("/")
    return f"{socket.gethostname()}/{rel}"


def cmd_backfill(a):
    files = _local_secret_files()
    host = socket.gethostname()
    if not files:
        return print(f"{host}: no local secret files matched")
    print(f"{host}: {len(files)} secret file(s)")
    if a.dry_run:
        for path in files:
            data = Path(path).read_bytes()
            tag = "attachment" if len(data) > NOTE_MAX else "note"
            print(f"  WOULD store  {_local_name(path)}\t({len(data)} bytes, "
                  f"sha {hashlib.sha256(data).hexdigest()[:12]}, {tag})")
        print("  (dry-run — nothing written; bootstrap.env / *.example / AIPA-Data excluded)")
        return
    env = load_env()
    s = session(env)
    ok, failed = 0, []
    for path in files:
        name = _local_name(path)
        data = Path(path).read_bytes()
        try:
            _, sha, mode = store_secret(env, s, name, data, host,
                                        [{"name": "source_host", "value": host, "type": 0},
                                         {"name": "source_path", "value": path, "type": 0}])
            back = read_secret(env, s, name)
            v = back is not None and hashlib.sha256(back).hexdigest() == sha
            print(f"  stored {name}\t({len(data)}b, {mode}) verify={'OK' if v else 'MISMATCH'}")
            ok += 1
        except BwError as e:
            print(f"  FAILED {name}: {e}")
            failed.append(name)
    print(f"{host}: {ok}/{len(files)} stored" + (f"; {len(failed)} FAILED: {', '.join(failed)}" if failed else ""))


def main():
    p = argparse.ArgumentParser(prog="secret", description="per-box 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)
    pr = sub.add_parser("rm", help="delete an item by exact name (soft-delete to trash)")
    pr.add_argument("name"); pr.set_defaults(fn=cmd_rm)
    pb = sub.add_parser("backfill", help="scan THIS box's local secret files and upsert each")
    pb.add_argument("--dry-run", action="store_true"); pb.set_defaults(fn=cmd_backfill)
    a = p.parse_args()
    # ⚠ SERIALISE THE WHOLE COMMAND, not individual `bw` calls.
    #
    # Every invocation runs `bw unlock` to mint a session (see `session`), and
    # concurrent unlocks against the shared BITWARDENCLI_APPDATA_DIR invalidate each
    # other. The damage then surfaces DOWNSTREAM — `bw list items` returns empty
    # stdout, or an item read returns an empty body — so a per-call lock cannot help:
    # by the time the read runs, the session it holds is already dead.
    #
    # Measured 2026-09-15: four parallel `secret get` calls for distinct items
    # returned EMPTY STRINGS WITH EXIT 0. Zero of four succeeded here; one of four in
    # svos-dev's run that reported it. No error, no timeout, no diagnostic — and a
    # caller that treats an empty optional secret as "not configured" then degrades
    # silently and never knows. It cost SVOS its ability to page the operator.
    #
    # Vault reads are ~17s each, so parallelism was never buying much. Correctness
    # first: callers queue, and no caller has to know any of this.
    BW_LOCK.touch(exist_ok=True)
    lk = open(BW_LOCK, "a")
    fcntl.flock(lk, fcntl.LOCK_EX)
    try:
        a.fn(a)
    except BwError as e:
        die(str(e))
    finally:
        fcntl.flock(lk, fcntl.LOCK_UN)
        lk.close()


if __name__ == "__main__":
    main()