357 lines
12 KiB
Python
357 lines
12 KiB
Python
"""
|
|
Bitwarden/Vaultwarden REST API client.
|
|
Handles authentication, vault sync, and API communication.
|
|
"""
|
|
import json
|
|
import os
|
|
import ssl
|
|
from base64 import b64decode, b64encode
|
|
from typing import Optional, Tuple
|
|
|
|
import aiohttp
|
|
|
|
from crypto import BitwardenCrypto, PinCrypto
|
|
|
|
|
|
def _get_ssl_context():
|
|
ssl_ctx = ssl.create_default_context()
|
|
for ca_path in ["/etc/ssl/certs/ca-certificates.crt", "/etc/ssl/certs/ca-bundle.crt"]:
|
|
if os.path.exists(ca_path):
|
|
ssl_ctx.load_verify_locations(ca_path)
|
|
break
|
|
return ssl_ctx
|
|
|
|
|
|
# Default URLs
|
|
BITWARDEN_API_BASE = "https://api.bitwarden.com"
|
|
BITWARDEN_IDENTITY_BASE = "https://identity.bitwarden.com"
|
|
|
|
# OAuth2 device type (2 = CLI)
|
|
DEVICE_TYPE = 15 # Use CLI device type for compatibility
|
|
|
|
|
|
class BitwardenClient:
|
|
"""REST API client for Bitwarden/Vaultwarden."""
|
|
|
|
def __init__(self, server_url: Optional[str] = None):
|
|
self.server_url = server_url or BITWARDEN_API_BASE
|
|
self.identity_url = self._get_identity_url()
|
|
self.crypto = BitwardenCrypto()
|
|
self.pin_crypto = PinCrypto()
|
|
self.access_token: Optional[str] = None
|
|
self.refresh_token: Optional[str] = None
|
|
self.enc_key: Optional[bytes] = None
|
|
self.mac_key: Optional[bytes] = None
|
|
self.user_id: Optional[str] = None
|
|
self.email: Optional[str] = None
|
|
self.kdf_type: int = 0
|
|
self.kdf_iterations: int = 600000
|
|
self.kdf_memory: Optional[int] = None
|
|
self.kdf_parallelism: Optional[int] = None
|
|
self._vault_data: Optional[dict] = None
|
|
|
|
def _get_identity_url(self) -> str:
|
|
"""Get the identity endpoint URL."""
|
|
if self.server_url:
|
|
base = self.server_url.rstrip("/")
|
|
# Handle Bitwarden cloud URLs
|
|
if base == "https://api.bitwarden.com":
|
|
return BITWARDEN_IDENTITY_BASE
|
|
if base == "https://api.bitwarden.eu":
|
|
return "https://identity.bitwarden.eu"
|
|
# Handle self-hosted URLs (Vaultwarden)
|
|
# Vaultwarden uses /identity directly
|
|
if "/identity" in base:
|
|
return base
|
|
if "/api" in base:
|
|
return base.replace("/api", "/identity")
|
|
return f"{base}/identity"
|
|
return BITWARDEN_IDENTITY_BASE
|
|
|
|
def _get_api_url(self) -> str:
|
|
"""Get the API endpoint URL."""
|
|
if self.server_url:
|
|
base = self.server_url.rstrip("/")
|
|
# Handle Bitwarden cloud URLs
|
|
if base == "https://identity.bitwarden.com":
|
|
return BITWARDEN_API_BASE
|
|
if base == "https://identity.bitwarden.eu":
|
|
return "https://api.bitwarden.eu"
|
|
# Handle self-hosted URLs (Vaultwarden)
|
|
if "/api" in base:
|
|
return base
|
|
if "/identity" in base:
|
|
return base.replace("/identity", "/api")
|
|
return f"{base}/api"
|
|
return BITWARDEN_API_BASE
|
|
|
|
async def _request(
|
|
self,
|
|
method: str,
|
|
url: str,
|
|
headers: dict = None,
|
|
data: dict = None,
|
|
params: dict = None,
|
|
) -> dict:
|
|
"""Make an HTTP request."""
|
|
if headers is None:
|
|
headers = {}
|
|
headers.setdefault("Content-Type", "application/json")
|
|
headers.setdefault("Accept", "application/json")
|
|
headers.setdefault("Device-Type", str(DEVICE_TYPE))
|
|
|
|
async with aiohttp.ClientSession() as session:
|
|
async with session.request(
|
|
method, url, headers=headers, json=data, params=params, ssl=_get_ssl_context()
|
|
) as resp:
|
|
text = await resp.text()
|
|
if resp.status >= 400:
|
|
try:
|
|
error_data = json.loads(text)
|
|
message = error_data.get("error_model", {}).get(
|
|
"message", text
|
|
)
|
|
except (json.JSONDecodeError, KeyError):
|
|
message = text
|
|
raise Exception(
|
|
f"API error {resp.status}: {message}"
|
|
)
|
|
return json.loads(text) if text else {}
|
|
|
|
async def prelogin(self, email: str) -> dict:
|
|
"""Get KDF settings for the user."""
|
|
url = f"{self._get_api_url()}/accounts/prelogin"
|
|
data = {"email": email}
|
|
result = await self._request("POST", url, data=data)
|
|
|
|
self.kdf_type = result.get("kdf", 0)
|
|
self.kdf_iterations = result.get("kdfIterations", 600000)
|
|
self.kdf_memory = result.get("kdfMemory")
|
|
self.kdf_parallelism = result.get("kdfParallelism")
|
|
self.email = email
|
|
|
|
return result
|
|
|
|
async def login_password(
|
|
self, email: str, password: str, two_factor_token: Optional[str] = None
|
|
) -> dict:
|
|
"""Login with email and master password."""
|
|
await self.prelogin(email)
|
|
|
|
# Derive master key
|
|
if self.kdf_type == 0: # PBKDF2
|
|
master_key = self.crypto.derive_master_key_pbkdf2(
|
|
password, email, self.kdf_iterations
|
|
)
|
|
elif self.kdf_type == 1: # Argon2id
|
|
master_key = self.crypto.derive_master_key_argon2(
|
|
password,
|
|
email,
|
|
self.kdf_iterations,
|
|
self.kdf_memory,
|
|
self.kdf_parallelism,
|
|
)
|
|
else:
|
|
raise ValueError(f"Unsupported KDF type: {self.kdf_type}")
|
|
|
|
# Hash master password for auth
|
|
master_password_hash = b64encode(
|
|
self.crypto.hmac_sha256(
|
|
master_key, password.encode("utf-8")
|
|
)
|
|
).decode("utf-8")
|
|
|
|
# Build auth request
|
|
url = f"{self.identity_url}/connect/token"
|
|
data = {
|
|
"grant_type": "password",
|
|
"username": email,
|
|
"password": master_password_hash,
|
|
"scope": "api offline_access",
|
|
"client_id": "web",
|
|
"deviceType": DEVICE_TYPE,
|
|
"deviceIdentifier": "decky-vaultwarden",
|
|
"deviceName": "decky-vaultwarden",
|
|
}
|
|
|
|
headers = {"Content-Type": "application/x-www-form-urlencoded"}
|
|
|
|
# Handle 2FA if needed
|
|
if two_factor_token:
|
|
data["twoFactorToken"] = two_factor_token
|
|
data["twoFactorProvider"] = "0" # Authenticator
|
|
data["twoFactorRemember"] = "1"
|
|
|
|
# Use form data instead of JSON
|
|
async with aiohttp.ClientSession() as session:
|
|
async with session.post(
|
|
url, data=data, headers=headers, ssl=_get_ssl_context()
|
|
) as resp:
|
|
text = await resp.text()
|
|
if resp.status >= 400:
|
|
try:
|
|
error_data = json.loads(text)
|
|
# Check if 2FA is required
|
|
if error_data.get("error") == "invalid_grant" and "twoFactor" in text:
|
|
return {"two_factor_required": True}
|
|
message = error_data.get("error_model", {}).get(
|
|
"message", text
|
|
)
|
|
except (json.JSONDecodeError, KeyError):
|
|
message = text
|
|
raise Exception(f"Login failed: {message}")
|
|
result = json.loads(text)
|
|
|
|
self.access_token = result.get("access_token")
|
|
self.refresh_token = result.get("refresh_token")
|
|
self.user_id = result.get("Profile", {}).get("id") or result.get("sub")
|
|
|
|
# Get encrypted user key
|
|
enc_user_key = result.get("Key")
|
|
if enc_user_key:
|
|
self.enc_key, self.mac_key = self.crypto.decrypt_user_key(
|
|
enc_user_key, master_key
|
|
)
|
|
return {
|
|
"success": True,
|
|
"master_key": master_key,
|
|
"enc_key": self.enc_key,
|
|
"mac_key": self.mac_key,
|
|
}
|
|
|
|
return {"success": True, "master_key": master_key}
|
|
|
|
async def login_api_key(
|
|
self, client_id: str, client_secret: str, email: str
|
|
) -> dict:
|
|
"""Login with API key (OAuth2 client_credentials)."""
|
|
url = f"{self.identity_url}/connect/token"
|
|
data = {
|
|
"grant_type": "client_credentials",
|
|
"scope": "api",
|
|
"client_id": client_id,
|
|
"client_secret": client_secret,
|
|
}
|
|
|
|
headers = {"Content-Type": "application/x-www-form-urlencoded"}
|
|
|
|
async with aiohttp.ClientSession() as session:
|
|
async with session.post(
|
|
url, data=data, headers=headers, ssl=_get_ssl_context()
|
|
) as resp:
|
|
text = await resp.text()
|
|
if resp.status >= 400:
|
|
try:
|
|
error_data = json.loads(text)
|
|
message = error_data.get("error_model", {}).get(
|
|
"message", text
|
|
)
|
|
except (json.JSONDecodeError, KeyError):
|
|
message = text
|
|
raise Exception(f"API key login failed: {message}")
|
|
result = json.loads(text)
|
|
|
|
self.access_token = result.get("access_token")
|
|
self.user_id = result.get("sub")
|
|
|
|
return {
|
|
"success": True,
|
|
"needs_master_password": True,
|
|
"message": "API key authenticated. Master password required for decryption.",
|
|
}
|
|
|
|
async def unlock_with_master_password(
|
|
self, master_password: str
|
|
) -> dict:
|
|
"""Derive keys from master password for decryption."""
|
|
if not self.email:
|
|
raise ValueError("Must login first")
|
|
|
|
await self.prelogin(self.email)
|
|
|
|
if self.kdf_type == 0:
|
|
master_key = self.crypto.derive_master_key_pbkdf2(
|
|
master_password, self.email, self.kdf_iterations
|
|
)
|
|
elif self.kdf_type == 1:
|
|
master_key = self.crypto.derive_master_key_argon2(
|
|
master_password,
|
|
self.email,
|
|
self.kdf_iterations,
|
|
self.kdf_memory,
|
|
self.kdf_parallelism,
|
|
)
|
|
else:
|
|
raise ValueError(f"Unsupported KDF type: {self.kdf_type}")
|
|
|
|
return {"master_key": master_key}
|
|
|
|
async def sync_vault(self) -> dict:
|
|
"""Sync and decrypt the full vault."""
|
|
if not self.access_token:
|
|
raise ValueError("Not authenticated")
|
|
|
|
url = f"{self._get_api_url()}/sync"
|
|
headers = {
|
|
"Authorization": f"Bearer {self.access_token}",
|
|
}
|
|
params = {"excludeDomains": "true"}
|
|
|
|
result = await self._request("GET", url, headers=headers, params=params)
|
|
self._vault_data = result
|
|
return result
|
|
|
|
def decrypt_vault(
|
|
self, sync_data: dict, enc_key: bytes, mac_key: bytes
|
|
) -> dict:
|
|
"""Decrypt all items in the vault."""
|
|
ciphers = sync_data.get("ciphers", [])
|
|
folders = sync_data.get("folders", [])
|
|
collections = sync_data.get("collections", [])
|
|
|
|
# Decrypt folders
|
|
decrypted_folders = []
|
|
for folder in folders:
|
|
name = ""
|
|
if folder.get("name"):
|
|
try:
|
|
name = self.crypto.decrypt_cipher_string(
|
|
folder["name"], enc_key, mac_key
|
|
).decode("utf-8")
|
|
except Exception:
|
|
name = "[encrypted]"
|
|
decrypted_folders.append({
|
|
"id": folder.get("id"),
|
|
"name": name,
|
|
"revisionDate": folder.get("revisionDate"),
|
|
})
|
|
|
|
# Decrypt ciphers
|
|
decrypted_ciphers = []
|
|
for cipher in ciphers:
|
|
if cipher.get("type") in [1, 2, 3, 4]: # Login, Note, Card, Identity
|
|
decrypted = self.crypto.decrypt_cipher(
|
|
cipher, enc_key, mac_key
|
|
)
|
|
if decrypted:
|
|
decrypted_ciphers.append(decrypted)
|
|
|
|
# Build folder map
|
|
folder_map = {f["id"]: f["name"] for f in decrypted_folders}
|
|
|
|
return {
|
|
"ciphers": decrypted_ciphers,
|
|
"folders": decrypted_folders,
|
|
"folder_map": folder_map,
|
|
"collections": collections,
|
|
}
|
|
|
|
def get_kdf_info(self) -> dict:
|
|
"""Get current KDF configuration."""
|
|
return {
|
|
"kdf_type": self.kdf_type,
|
|
"kdf_iterations": self.kdf_iterations,
|
|
"kdf_memory": self.kdf_memory,
|
|
"kdf_parallelism": self.kdf_parallelism,
|
|
}
|