bw-secrets maintainers
fd50e70f33
Single-file pure-Python tool for retrieving secrets from a self-hosted Bitwarden/Vaultwarden vault. Org-first by default, agent exit codes (0/1/2), no session file on disk, no jq/pexpect dependencies. Includes: install.sh, README with org setup walkthrough and agent onboarding prompt, SKILL.md agent documentation.
531 lines
17 KiB
Python
Executable File
531 lines
17 KiB
Python
Executable File
#!/usr/bin/env python3
|
|
"""bw-secrets — Agent-friendly Bitwarden/Vaultwarden secrets helper.
|
|
|
|
Pure Python 3 (stdlib only). No jq, no pexpect, no session file on disk.
|
|
|
|
Design rules:
|
|
* Secrets ALWAYS live in Bitwarden. This tool only reads them at runtime.
|
|
* Master password file (chmod 600) — the ONLY secret on disk. Never write
|
|
the session token to disk; re-unlock per call.
|
|
* ALWAYS `bw sync` before reads (the CLI serves a local cache).
|
|
* Look items up by id, never by name (em-dash silent-failure bug).
|
|
* Agent-friendly: exit 0 = found, 1 = not found, 2 = ambiguous match.
|
|
* ORG-FIRST: get/list/env search the default org (BW_ORG_ID) by default.
|
|
Personal-vault items are excluded unless --personal; --all searches
|
|
everything. Collections matched case-insensitively.
|
|
|
|
Config resolution (highest wins): environment variable > config file
|
|
(~/.config/bw/config.env) > built-in default.
|
|
|
|
BW_SERVER Vaultwarden URL (required — no default)
|
|
BW_ORG_ID Default org id (required for org-first lookups)
|
|
BW_MASTER_PW_FILE Master password file (default ~/.config/bw/master_pw)
|
|
BW_BIN bw CLI path (default bw on PATH)
|
|
|
|
Usage:
|
|
bw-secrets get <name> [field] [--collection <name>] [--personal|--all] [--json]
|
|
bw-secrets list [search] [--collection <name>] [--personal|--all]
|
|
bw-secrets env <name> [field] [--collection <name>] [--personal|--all] # prints KEY=value for sourcing
|
|
bw-secrets sync | status | unlock
|
|
bw-secrets org list
|
|
bw-secrets org collections [org-id]
|
|
bw-secrets org items <collection> [search]
|
|
bw-secrets org get <collection> <name> [field] [--json]
|
|
|
|
Fields: username, password (default), notes, totp, url, custom:<field>, json,
|
|
or a dotted path into the item (e.g. login.uris.0.uri).
|
|
"""
|
|
import json
|
|
import os
|
|
import re
|
|
import subprocess
|
|
import sys
|
|
from pathlib import Path
|
|
|
|
# --------------------------------------------------------------------------
|
|
# Config
|
|
# --------------------------------------------------------------------------
|
|
|
|
DEFAULTS = {
|
|
"BW_SERVER": "",
|
|
"BW_ORG_ID": "",
|
|
"BW_MASTER_PW_FILE": str(Path.home() / ".config" / "bw" / "master_pw"),
|
|
"BW_BIN": "bw",
|
|
}
|
|
|
|
CONFIG_FILE = Path.home() / ".config" / "bw" / "config.env"
|
|
|
|
|
|
def load_config() -> dict:
|
|
cfg = dict(DEFAULTS)
|
|
if CONFIG_FILE.exists():
|
|
for line in CONFIG_FILE.read_text().splitlines():
|
|
line = line.strip()
|
|
if line and not line.startswith("#") and "=" in line:
|
|
k, v = line.split("=", 1)
|
|
cfg[k.strip()] = v.strip()
|
|
for k in DEFAULTS:
|
|
if os.environ.get(k):
|
|
cfg[k] = os.environ[k]
|
|
return cfg
|
|
|
|
|
|
CFG = load_config()
|
|
|
|
|
|
def require(key: str, hint: str) -> str:
|
|
val = CFG.get(key, "")
|
|
if not val:
|
|
sys.stderr.write(
|
|
f"ERROR: {key} is not set. {hint}\n"
|
|
f"Set it in {CONFIG_FILE} or export it as an environment variable.\n"
|
|
)
|
|
sys.exit(1)
|
|
return val
|
|
|
|
|
|
# --------------------------------------------------------------------------
|
|
# bw plumbing
|
|
# --------------------------------------------------------------------------
|
|
|
|
|
|
def _env() -> dict:
|
|
env = os.environ.copy()
|
|
if CFG.get("BW_SERVER"):
|
|
env["BW_SERVER"] = CFG["BW_SERVER"]
|
|
return env
|
|
|
|
|
|
def _bw(args, session=None, check=True):
|
|
"""Run bw with the given args. Returns (returncode, stdout)."""
|
|
cmd = [CFG["BW_BIN"]] + args
|
|
if session:
|
|
cmd += ["--session", session]
|
|
proc = subprocess.run(cmd, capture_output=True, text=True, env=_env(), timeout=60)
|
|
if check and proc.returncode != 0:
|
|
err = proc.stderr.strip() or proc.stdout.strip()
|
|
sys.stderr.write(f"bw error: {err}\n")
|
|
sys.exit(1)
|
|
return proc.returncode, proc.stdout
|
|
|
|
|
|
def unlock() -> str:
|
|
"""Unlock the vault, return session token. Never persists the token."""
|
|
pw_file = Path(require("BW_MASTER_PW_FILE",
|
|
"Point BW_MASTER_PW_FILE at your master password file."))
|
|
if not pw_file.exists():
|
|
sys.stderr.write(f"ERROR: master password file not found: {pw_file}\n")
|
|
sys.exit(1)
|
|
proc = subprocess.run(
|
|
[CFG["BW_BIN"], "unlock", "--passwordfile", str(pw_file), "--raw"],
|
|
capture_output=True, text=True, env=_env(), timeout=60,
|
|
)
|
|
session = proc.stdout.strip()
|
|
if not session or proc.returncode != 0:
|
|
sys.stderr.write("ERROR: bw unlock failed\n")
|
|
sys.exit(1)
|
|
return session
|
|
|
|
|
|
def sync(session: str) -> None:
|
|
_bw(["sync"], session=session, check=False)
|
|
|
|
|
|
def _list_json(session: str, what: str, retries: int = 3) -> list:
|
|
"""bw list <what> with retry. The CLI intermittently returns empty stdout
|
|
(verified 2026-09-10: `bw list items` returned 0 bytes on ~5% of runs).
|
|
Retry with a fresh session before giving up."""
|
|
for attempt in range(1, retries + 1):
|
|
_, out = _bw(["list", what], session=session, check=False)
|
|
try:
|
|
return json.loads(out)
|
|
except json.JSONDecodeError:
|
|
if attempt < retries:
|
|
session = unlock() # fresh session — stale token can cause empty output
|
|
continue
|
|
sys.stderr.write(f"ERROR: could not parse bw list {what} output "
|
|
f"(empty={len(out) == 0})\n")
|
|
sys.exit(1)
|
|
return [] # unreachable
|
|
|
|
|
|
def list_items(session: str) -> list:
|
|
"""Unfiltered item list (catches collection-only items the org filter misses)."""
|
|
return _list_json(session, "items")
|
|
|
|
|
|
def list_collections(session: str) -> list:
|
|
return _list_json(session, "collections")
|
|
|
|
|
|
def list_orgs(session: str) -> list:
|
|
return _list_json(session, "organizations")
|
|
|
|
|
|
# --------------------------------------------------------------------------
|
|
# Lookup helpers
|
|
# --------------------------------------------------------------------------
|
|
|
|
def find_collection_id(collections: list, name: str, org_id: str = None) -> str:
|
|
"""Case-insensitive collection lookup, preferring the target org."""
|
|
name_l = name.lower()
|
|
candidates = [c for c in collections if c.get("name", "").lower() == name_l]
|
|
if not candidates:
|
|
return None
|
|
if org_id:
|
|
for c in candidates:
|
|
if c.get("organizationId") == org_id:
|
|
return c["id"]
|
|
return candidates[0]["id"]
|
|
|
|
|
|
def match_items(items: list, name: str, collection_id: str = None) -> list:
|
|
"""Case-insensitive substring match on item name, optional collection filter."""
|
|
name_l = name.lower()
|
|
hits = [it for it in items if name_l in it.get("name", "").lower()]
|
|
if collection_id:
|
|
hits = [it for it in hits if collection_id in it.get("collectionIds", [])]
|
|
return hits
|
|
|
|
|
|
def resolve_scope(items: list, scope: str, org_id: str = None,
|
|
collections: list = None) -> list:
|
|
"""Filter items by search scope.
|
|
|
|
scope: 'org' (default) — items in the default org's collections
|
|
'personal' — items with no org/collection (personal vault)
|
|
'all' — everything
|
|
|
|
Org membership is determined by collectionIds intersecting the org's
|
|
collection IDs — NOT by item.organizationId, which is null on items
|
|
created via `bw create item` (verified 2026-06-28).
|
|
"""
|
|
if scope == "all":
|
|
return items
|
|
if scope == "personal":
|
|
return [it for it in items if not it.get("collectionIds")]
|
|
# org scope
|
|
if org_id and collections:
|
|
org_coll_ids = {c["id"] for c in collections
|
|
if c.get("organizationId") == org_id}
|
|
return [it for it in items
|
|
if org_coll_ids.intersection(it.get("collectionIds", []))]
|
|
return [it for it in items if it.get("collectionIds")]
|
|
|
|
|
|
def get_field(item: dict, field: str):
|
|
"""Extract a field from a bw item. Returns (value, is_json)."""
|
|
if field == "json":
|
|
return json.dumps(item, indent=2), True
|
|
if field == "password":
|
|
return item.get("login", {}).get("password", ""), False
|
|
if field == "username":
|
|
return item.get("login", {}).get("username", ""), False
|
|
if field == "notes":
|
|
return item.get("notes", ""), False
|
|
if field == "totp" or field == "totp_uri":
|
|
return item.get("login", {}).get("totp", ""), False
|
|
if field == "url":
|
|
uris = item.get("login", {}).get("uris") or []
|
|
return (uris[0].get("uri", "") if uris else ""), False
|
|
if field.startswith("custom:"):
|
|
fname = field[len("custom:"):]
|
|
for f in item.get("fields", []):
|
|
if f.get("name") == fname:
|
|
return f.get("value", ""), False
|
|
return "", False
|
|
# dotted path, e.g. login.uris.0.uri
|
|
val = item
|
|
for part in field.split("."):
|
|
if isinstance(val, list):
|
|
try:
|
|
val = val[int(part)]
|
|
except (ValueError, IndexError):
|
|
return "", False
|
|
elif isinstance(val, dict):
|
|
val = val.get(part)
|
|
else:
|
|
return "", False
|
|
return (val if val is not None else ""), False
|
|
|
|
|
|
# --------------------------------------------------------------------------
|
|
# Commands
|
|
# --------------------------------------------------------------------------
|
|
|
|
def cmd_get(args):
|
|
if not args:
|
|
sys.stderr.write("Usage: bw-secrets get <name> [field] [--collection <name>] [--personal|--all] [--json]\n")
|
|
sys.exit(1)
|
|
name = args[0]
|
|
field = "password"
|
|
collection = None
|
|
scope = "org"
|
|
as_json = False
|
|
i = 1
|
|
while i < len(args):
|
|
a = args[i]
|
|
if a == "--collection" and i + 1 < len(args):
|
|
collection = args[i + 1]
|
|
i += 2
|
|
elif a == "--personal":
|
|
scope = "personal"
|
|
i += 1
|
|
elif a == "--all":
|
|
scope = "all"
|
|
i += 1
|
|
elif a == "--json":
|
|
as_json = True
|
|
i += 1
|
|
else:
|
|
field = a
|
|
i += 1
|
|
if as_json:
|
|
field = "json"
|
|
|
|
session = unlock()
|
|
sync(session)
|
|
items = list_items(session)
|
|
collections = list_collections(session)
|
|
org_id = CFG.get("BW_ORG_ID", "")
|
|
coll_id = find_collection_id(collections, collection, org_id) if collection else None
|
|
if collection and not coll_id:
|
|
sys.stderr.write(f"ERROR: collection '{collection}' not found\n")
|
|
sys.exit(1)
|
|
|
|
if collection:
|
|
items = [it for it in items if coll_id in it.get("collectionIds", [])]
|
|
else:
|
|
items = resolve_scope(items, scope, org_id, collections)
|
|
|
|
hits = match_items(items, name)
|
|
if not hits:
|
|
sys.stderr.write(f"ERROR: no items found matching '{name}'\n")
|
|
sys.exit(1)
|
|
if len(hits) > 1:
|
|
names = ", ".join(h.get("name", "?") for h in hits)
|
|
sys.stderr.write(f"ERROR: ambiguous match for '{name}': {names}\n")
|
|
sys.exit(2)
|
|
|
|
value, is_json = get_field(hits[0], field)
|
|
if is_json:
|
|
print(value)
|
|
else:
|
|
print(value)
|
|
|
|
|
|
def cmd_list(args):
|
|
search = args[0] if args and not args[0].startswith("--") else None
|
|
collection = None
|
|
scope = "org"
|
|
if "--collection" in args:
|
|
idx = args.index("--collection")
|
|
if idx + 1 < len(args):
|
|
collection = args[idx + 1]
|
|
if "--personal" in args:
|
|
scope = "personal"
|
|
if "--all" in args:
|
|
scope = "all"
|
|
|
|
session = unlock()
|
|
sync(session)
|
|
items = list_items(session)
|
|
collections = list_collections(session)
|
|
org_id = CFG.get("BW_ORG_ID", "")
|
|
coll_id = find_collection_id(collections, collection, org_id) if collection else None
|
|
if collection and not coll_id:
|
|
sys.stderr.write(f"ERROR: collection '{collection}' not found\n")
|
|
sys.exit(1)
|
|
|
|
if collection:
|
|
items = [it for it in items if coll_id in it.get("collectionIds", [])]
|
|
else:
|
|
items = resolve_scope(items, scope, org_id, collections)
|
|
|
|
if search:
|
|
items = [it for it in items if search.lower() in it.get("name", "").lower()]
|
|
|
|
for it in sorted(items, key=lambda x: x.get("name", "").lower()):
|
|
kind = "login" if it.get("type") == 1 else "other"
|
|
print(f"{it.get('name', '?')} [{kind}] {it.get('id', '?')}")
|
|
|
|
|
|
def cmd_env(args):
|
|
if not args:
|
|
sys.stderr.write("Usage: bw-secrets env <name> [field] [--collection <name>] [--personal|--all]\n")
|
|
sys.exit(1)
|
|
name = args[0]
|
|
field = "password"
|
|
collection = None
|
|
scope = "org"
|
|
i = 1
|
|
while i < len(args):
|
|
a = args[i]
|
|
if a == "--collection" and i + 1 < len(args):
|
|
collection = args[i + 1]
|
|
i += 2
|
|
elif a == "--personal":
|
|
scope = "personal"
|
|
i += 1
|
|
elif a == "--all":
|
|
scope = "all"
|
|
i += 1
|
|
else:
|
|
field = a
|
|
i += 1
|
|
|
|
session = unlock()
|
|
sync(session)
|
|
items = list_items(session)
|
|
collections = list_collections(session)
|
|
org_id = CFG.get("BW_ORG_ID", "")
|
|
coll_id = find_collection_id(collections, collection, org_id) if collection else None
|
|
if collection:
|
|
items = [it for it in items if coll_id in it.get("collectionIds", [])]
|
|
else:
|
|
items = resolve_scope(items, scope, org_id, collections)
|
|
|
|
hits = match_items(items, name)
|
|
if not hits:
|
|
sys.stderr.write(f"ERROR: no items found matching '{name}'\n")
|
|
sys.exit(1)
|
|
if len(hits) > 1:
|
|
names = ", ".join(h.get("name", "?") for h in hits)
|
|
sys.stderr.write(f"ERROR: ambiguous match for '{name}': {names}\n")
|
|
sys.exit(2)
|
|
|
|
value, _ = get_field(hits[0], field)
|
|
key = re.sub(r"[^A-Z0-9_]", "_", hits[0].get("name", "SECRET").upper())
|
|
print(f"{key}={value}")
|
|
|
|
|
|
def cmd_sync(_args):
|
|
session = unlock()
|
|
sync(session)
|
|
print("Vault synced.")
|
|
|
|
|
|
def cmd_status(_args):
|
|
proc = subprocess.run([CFG["BW_BIN"], "status"], capture_output=True, text=True, env=_env(), timeout=30)
|
|
print(proc.stdout.strip() or proc.stderr.strip())
|
|
|
|
|
|
def cmd_unlock(_args):
|
|
session = unlock()
|
|
print(f"Vault unlocked. Session: {session[:8]}...")
|
|
|
|
|
|
def cmd_org(args):
|
|
if not args:
|
|
sys.stderr.write("Usage: bw-secrets org <list|collections|items|get> ...\n")
|
|
sys.exit(1)
|
|
sub = args[0]
|
|
rest = args[1:]
|
|
|
|
if sub == "list":
|
|
session = unlock()
|
|
sync(session)
|
|
for o in list_orgs(session):
|
|
print(f"{o.get('name', '?')} [{o.get('id', '?')}]")
|
|
return
|
|
|
|
if sub == "collections":
|
|
session = unlock()
|
|
sync(session)
|
|
org_id = rest[0] if rest else CFG.get("BW_ORG_ID", "")
|
|
for c in list_collections(session):
|
|
if c.get("organizationId") == org_id:
|
|
print(f"{c.get('name', '?')} [{c.get('id', '?')}]")
|
|
return
|
|
|
|
if sub == "items":
|
|
if not rest:
|
|
sys.stderr.write("Usage: bw-secrets org items <collection> [search]\n")
|
|
sys.exit(1)
|
|
collection = rest[0]
|
|
search = rest[1] if len(rest) > 1 else None
|
|
session = unlock()
|
|
sync(session)
|
|
collections = list_collections(session)
|
|
coll_id = find_collection_id(collections, collection, CFG.get("BW_ORG_ID", ""))
|
|
if not coll_id:
|
|
sys.stderr.write(f"ERROR: collection '{collection}' not found\n")
|
|
sys.exit(1)
|
|
items = [it for it in list_items(session) if coll_id in it.get("collectionIds", [])]
|
|
if search:
|
|
items = [it for it in items if search.lower() in it.get("name", "").lower()]
|
|
for it in sorted(items, key=lambda x: x.get("name", "").lower()):
|
|
kind = "login" if it.get("type") == 1 else "other"
|
|
print(f"{it.get('name', '?')} [{kind}] {it.get('id', '?')}")
|
|
return
|
|
|
|
if sub == "get":
|
|
if len(rest) < 2:
|
|
sys.stderr.write("Usage: bw-secrets org get <collection> <name> [field] [--json]\n")
|
|
sys.exit(1)
|
|
collection = rest[0]
|
|
name = rest[1]
|
|
field = "password"
|
|
as_json = False
|
|
for a in rest[2:]:
|
|
if a == "--json":
|
|
as_json = True
|
|
else:
|
|
field = a
|
|
if as_json:
|
|
field = "json"
|
|
session = unlock()
|
|
sync(session)
|
|
collections = list_collections(session)
|
|
coll_id = find_collection_id(collections, collection, CFG.get("BW_ORG_ID", ""))
|
|
if not coll_id:
|
|
sys.stderr.write(f"ERROR: collection '{collection}' not found\n")
|
|
sys.exit(1)
|
|
hits = match_items(list_items(session), name, coll_id)
|
|
if not hits:
|
|
sys.stderr.write(f"ERROR: no items found matching '{name}' in collection '{collection}'\n")
|
|
sys.exit(1)
|
|
if len(hits) > 1:
|
|
names = ", ".join(h.get("name", "?") for h in hits)
|
|
sys.stderr.write(f"ERROR: ambiguous match for '{name}': {names}\n")
|
|
sys.exit(2)
|
|
value, is_json = get_field(hits[0], field)
|
|
print(value if is_json else value)
|
|
return
|
|
|
|
sys.stderr.write(f"ERROR: unknown org subcommand '{sub}'\n")
|
|
sys.exit(1)
|
|
|
|
|
|
def cmd_help(_args=None):
|
|
print(__doc__)
|
|
|
|
|
|
def main():
|
|
if not args:
|
|
cmd_help()
|
|
return
|
|
cmd = args[0]
|
|
rest = args[1:]
|
|
handlers = {
|
|
"get": cmd_get,
|
|
"list": cmd_list,
|
|
"env": cmd_env,
|
|
"sync": cmd_sync,
|
|
"status": cmd_status,
|
|
"unlock": cmd_unlock,
|
|
"org": cmd_org,
|
|
"help": cmd_help,
|
|
"--help": cmd_help,
|
|
"-h": cmd_help,
|
|
}
|
|
if cmd not in handlers:
|
|
sys.stderr.write(f"ERROR: unknown command '{cmd}'. Run 'bw-secrets help'.\n")
|
|
sys.exit(1)
|
|
handlers[cmd](rest)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
args = sys.argv[1:]
|
|
main()
|