agent-identity-provisioning/bw-helper.py

#!/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):
        self.client_id = client_id
        self.client_secret = client_secret
        self.password = password
        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."""
        env = os.environ.copy()
        env["BW_CLIENTID"] = self.client_id
        env["BW_CLIENTSECRET"] = self.client_secret
        env["BW_PASSWORD"] = self.password

        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()}")

        self.session = subprocess.run(
            ["bw", "unlock", "--raw"],
            capture_output=True,
            text=True,
            env=env,
            input=self.password + "\n",
        ).stdout.strip()

        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 ""