#!/usr/bin/env python3 """ bw-helper.py — Bitwarden CLI wrapper for agent identity provisioning. Provides a clean Python interface to the `bw` CLI for: - Password generation - Item creation/retrieval in collections - TOTP code generation - Session management All BW commands run via subprocess. The BW session is established once and reused across calls. """ import json import os import subprocess import sys from typing import Optional class BitwardenHelper: """Wrapper around the Bitwarden CLI for credential management.""" def __init__(self, client_id: str, client_secret: str, password: str, totp_secret: str = "", server_url: str = ""): self.client_id = client_id self.client_secret = client_secret self.password = password self.totp_secret = totp_secret self.server_url = server_url self.session: Optional[str] = None def _run_bw(self, args: list[str], capture: bool = True) -> str: """Run a bw CLI command with the active session.""" env = os.environ.copy() if self.session: env["BW_SESSION"] = self.session result = subprocess.run( ["bw"] + args, capture_output=capture, text=True, env=env, ) if result.returncode != 0: raise RuntimeError( f"bw {' '.join(args)} failed: {result.stderr.strip()}" ) return result.stdout.strip() if capture else "" def login(self) -> None: """Authenticate via API key and unlock the vault. Configures the BW server URL (for self-hosted instances), logs in via API key, and unlocks the vault. API key auth does not require TOTP — the key itself is obtained from an authenticated session. """ env = os.environ.copy() env["BW_CLIENTID"] = self.client_id env["BW_CLIENTSECRET"] = self.client_secret # Configure server URL for self-hosted instances if self.server_url: subprocess.run( ["bw", "config", "server", self.server_url], capture_output=True, text=True, env=env, ) # Login via API key (tolerates already-logged-in state) result = subprocess.run( ["bw", "login", "--apikey"], capture_output=True, text=True, env=env, ) if result.returncode != 0 and "already" not in result.stderr.lower(): raise RuntimeError(f"BW login failed: {result.stderr.strip()}") # Unlock via password file (more reliable than stdin with native binary) import tempfile with tempfile.NamedTemporaryFile(mode="w", suffix=".pw", delete=False) as pw_file: pw_file.write(self.password) pw_file_path = pw_file.name try: self.session = subprocess.run( ["bw", "unlock", "--passwordfile", pw_file_path, "--raw"], capture_output=True, text=True, env=env, ).stdout.strip() finally: os.unlink(pw_file_path) if not self.session: raise RuntimeError("BW unlock failed — no session token returned") def generate_password(self, length: int = 32) -> str: """Generate a strong password.""" return self._run_bw(["generate", "-ulns", "--length", str(length)]) def get_totp(self, item_name: str) -> str: """Get the current TOTP code for a Bitwarden item.""" return self._run_bw(["get", "totp", item_name]) def create_item( self, name: str, username: str, password: str, uris: list[str], collection_name: str, totp_secret: Optional[str] = None, custom_fields: Optional[dict[str, str]] = None, ) -> str: """Create a login item in a Bitwarden collection. Returns the item ID. """ item = { "type": 1, # LOGIN "name": name, "login": { "username": username, "password": password, "uris": [{"uri": u, "match": None} for u in uris], }, "collectionIds": [], # resolved by collection_name below } if totp_secret: item["login"]["totp"] = totp_secret fields = [] if custom_fields: for key, value in custom_fields.items(): fields.append({"name": key, "value": value, "type": 0}) if fields: item["fields"] = fields # Resolve collection ID collection_id = self._get_collection_id(collection_name) if collection_id: item["collectionIds"] = [collection_id] # Create via BW CLI encoded = json.dumps(item) result = subprocess.run( ["bw", "encode"], input=encoded, capture_output=True, text=True, ) if result.returncode != 0: raise RuntimeError(f"bw encode failed: {result.stderr.strip()}") encoded_item = result.stdout.strip() output = self._run_bw(["create", "item", encoded_item]) created = json.loads(output) return created.get("id", "") def _get_collection_id(self, collection_name: str) -> Optional[str]: """Look up a collection ID by name. Returns None if not found.""" try: output = self._run_bw(["list", "collections"]) collections = json.loads(output) for col in collections: if col.get("name", "").lower() == collection_name.lower(): return col.get("id") except (RuntimeError, json.JSONDecodeError): pass return None def create_collection(self, collection_name: str, org_id: str) -> str: """Create a collection in an organization.""" item = {"name": collection_name, "organizationId": org_id} encoded = json.dumps(item) result = subprocess.run( ["bw", "encode"], input=encoded, capture_output=True, text=True, ) if result.returncode != 0: raise RuntimeError(f"bw encode failed: {result.stderr.strip()}") encoded_item = result.stdout.strip() output = self._run_bw(["create", "collection", encoded_item]) created = json.loads(output) return created.get("id", "") def item_exists(self, name: str) -> bool: """Check if a Bitwarden item with this name already exists.""" try: self._run_bw(["get", "item", name]) return True except RuntimeError as e: # Distinguish "not found" (expected) from real errors (network, session expired). err_msg = str(e).lower() if "not found" in err_msg or "no item" in err_msg: return False # Real error — re-raise so we don't silently create duplicates. raise def get_item_password(self, name: str) -> str: """Get the password field from a Bitwarden item.""" return self._run_bw(["get", "password", name]) def get_item_uri(self, name: str) -> str: """Get the URI from a Bitwarden item.""" output = self._run_bw(["get", "item", name]) item = json.loads(output) uris = item.get("login", {}).get("uris", []) return uris[0]["uri"] if uris else ""