feat(secrets-broker): nh3-dev backfill complete (25/25) + attachment + resilient run

Scope corrected to per-dev-box (CC sessions on this box), not a fleet service; each
box duplicates the stack and backs up its own local secrets, hostname-namespaced.

CLI:
- backfill is local-only (scan this box's ~/development/*/{env.sh,.env} + ~/.config
  credentials; exclude bootstrap.env/examples/AIPA-Data archives).
- large files (>6000 B) route to a bw ATTACHMENT instead of the note field
  (Vaultwarden caps notes at ~10000 encrypted chars); get/verify read it back.
- backfill catches per-item failures and continues (bw errors raise BwError,
  main converts to a clean exit); idempotent upsert makes re-runs safe.

Backfilled all 25 nh3-dev secret files into the infra-ops org's Default collection
(folder = hostname), every one round-trip verified (2 large via attachment, 23 via
note). README added for duplicating the stack to new dev boxes. Contract scope +
data-model sections updated (bw, org/collection, per-box).

services/secrets-broker/secret

@@ -1,29 +1,34 @@
 #!/usr/bin/env python3
-"""secret — fleet credential store & retrieve over Vaultwarden (bw-backed).
+"""secret — per-box 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.
+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
+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
+  secret backfill [--dry-run]              # scan THIS box's local secrets, upsert each
 """
 import argparse
 import base64
+import glob
 import hashlib
 import json
 import os
-import shlex
+import shutil
+import socket
 import subprocess
 import sys
+import tempfile
 import time
 from pathlib import Path
 
@@ -33,6 +38,11 @@ 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):
@@ -61,7 +71,7 @@ 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()}")
+        raise BwError(f"bw {args[0]}: {r.stderr.decode(errors='replace').strip()}")
     return r
 
 
@@ -86,8 +96,7 @@ def encode(obj, env):
 
 
 def find(env, s, name):
-    r = bw(["list", "items", "--search", name], env, s)
-    for it in json.loads(r.stdout.decode() or "[]"):
+    for it in json.loads(bw(["list", "items", "--search", name], env, s).stdout.decode() or "[]"):
         if it.get("name") == name:
             return it
     return None
@@ -103,6 +112,67 @@ def folder_id(env, s, cls):
     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)
@@ -112,35 +182,12 @@ def cmd_put(a):
         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},
-    ]
+    extra = []
     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 ''})")
+        extra.append({"name": k, "value": v, "type": 0})
+    _, 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):
@@ -151,19 +198,15 @@ def cmd_get(a):
     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 ""
+    data = read_secret(env, s, a.name)
     if a.file:
-        Path(a.file).write_text(content)
+        Path(a.file).write_bytes(data)
         os.chmod(a.file, 0o600)
         print(f"wrote {a.file} (0600)")
     else:
-        sys.stdout.write(content if content.endswith("\n") else content + "\n")
+        sys.stdout.buffer.write(data if data.endswith(b"\n") else data + b"\n")
 
 
 def cmd_list(a):
@@ -181,63 +224,68 @@ def cmd_list(a):
     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()]
+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 _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 _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):
-    host = a.host
-    files = _host_secret_files(host)
+    files = _local_secret_files()
+    host = socket.gethostname()
     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)
+        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 = _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
+        name = _local_name(path)
+        data = Path(path).read_bytes()
         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'}")
+            _, 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="fleet credential store over Vaultwarden")
+    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")
@@ -246,10 +294,13 @@ def main():
     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 = 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()
-    a.fn(a)
+    try:
+        a.fn(a)
+    except BwError as e:
+        die(str(e))
 
 
 if __name__ == "__main__":