feat: replace Node.js bw CLI with native Rust binary + fix module name

Replace npm-based @bitwarden/cli with the pre-compiled native Rust bw
binary (v2026.7.0) to eliminate Node.js from the credential management
layer for CMMC/ITAR/STIG audit readiness.

Changes:
- Dockerfile: download native bw binary instead of npm install; add
  python3-pip for Playwright dependencies
- bw_helper.py: renamed from bw-helper.py (Python can't import hyphens);
  added BW_SERVER config for self-hosted instance; use --passwordfile
  for unlock (more reliable with native binary); removed TOTP from
  login flow (API key auth does not require it)
- provision-agent.py: pass BW_SERVER env var to BitwardenHelper
- docker-compose.yml: add BW_SERVER env var
- .env.example: add BW_SERVER, document TOTP as optional

Verified: dry-run passes, bw status/auth/generate all work inside
the provisioner container against pwvault.turnsys.com.

💘 Generated with Crush

Assisted-by: Crush:glm-5.2
This commit is contained in:
TSYS Group COO
2026-08-13 13:00:53 -05:00
parent 7534964c13
commit 8c90d6809b
5 changed files with 45 additions and 26 deletions
+210
View File
@@ -0,0 +1,210 @@
#!/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 ""