vh
a249073a08
Both from jackdaw-dev feedback after a mis-namespaced item (missing host prefix) hid under a prefix nobody searches: - 'secret rm <name>' — delete an item by exact name (bw soft-delete to trash, recoverable); closes the 'no delete path, append-only in practice' gap. - 'secret put' now warns (stderr, non-blocking) when a name opens a brand-new top-level namespace, listing existing ones + suggesting the host prefix — catches a typo'd/missing prefix at store time. Installed copy at ~/.local/bin/secret synced.
331 lines
13 KiB
Python
Executable File
331 lines
13 KiB
Python
Executable File
#!/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 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
|
|
|
|
|
|
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:
|
|
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):
|
|
for it in json.loads(bw(["list", "items", "--search", name], env, s).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 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)
|
|
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()
|
|
try:
|
|
a.fn(a)
|
|
except BwError as e:
|
|
die(str(e))
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|