chore: init decky vaultwarden plugin scaffold
This commit is contained in:
@@ -0,0 +1,344 @@
|
||||
"""
|
||||
Bitwarden/Vaultwarden REST API client.
|
||||
Handles authentication, vault sync, and API communication.
|
||||
"""
|
||||
import json
|
||||
from base64 import b64decode, b64encode
|
||||
from typing import Optional, Tuple
|
||||
|
||||
import aiohttp
|
||||
|
||||
from .crypto import BitwardenCrypto, PinCrypto
|
||||
|
||||
|
||||
# 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
|
||||
) 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": "connector",
|
||||
"deviceType": DEVICE_TYPE,
|
||||
"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
|
||||
) 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
|
||||
) 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,
|
||||
}
|
||||
@@ -0,0 +1,123 @@
|
||||
"""
|
||||
Clipboard management with 60-second auto-clear.
|
||||
Uses xclip/xdotool for SteamOS (KDE Plasma).
|
||||
"""
|
||||
import asyncio
|
||||
import subprocess
|
||||
import time
|
||||
from typing import Optional
|
||||
|
||||
# Clipboard state
|
||||
_clipboard_timers: dict = {}
|
||||
_clipboard_contents: dict = {}
|
||||
|
||||
|
||||
async def copy_to_clipboard(text: str, clear_after: int = 60, label: str = "default") -> dict:
|
||||
"""Copy text to clipboard with auto-clear after specified seconds."""
|
||||
global _clipboard_timers, _clipboard_contents
|
||||
|
||||
try:
|
||||
# Cancel any existing timer for this label
|
||||
if label in _clipboard_timers:
|
||||
_clipboard_timers[label].cancel()
|
||||
|
||||
# Store the text
|
||||
_clipboard_contents[label] = text
|
||||
|
||||
# Copy to clipboard using xclip
|
||||
process = await asyncio.create_subprocess_exec(
|
||||
"xclip", "-selection", "clipboard",
|
||||
stdin=asyncio.subprocess.PIPE,
|
||||
stdout=asyncio.subprocess.PIPE,
|
||||
stderr=asyncio.subprocess.PIPE,
|
||||
)
|
||||
await process.communicate(input=text.encode("utf-8"))
|
||||
|
||||
if process.returncode != 0:
|
||||
# Fallback to xsel if xclip not available
|
||||
process = await asyncio.create_subprocess_exec(
|
||||
"xsel", "--clipboard", "--input",
|
||||
stdin=asyncio.subprocess.PIPE,
|
||||
stdout=asyncio.subprocess.PIPE,
|
||||
stderr=asyncio.subprocess.PIPE,
|
||||
)
|
||||
await process.communicate(input=text.encode("utf-8"))
|
||||
|
||||
if process.returncode != 0:
|
||||
return {"success": False, "error": "No clipboard tool available (xclip or xsel)"}
|
||||
|
||||
# Set up auto-clear timer
|
||||
async def clear_clipboard():
|
||||
await asyncio.sleep(clear_after)
|
||||
await clear_clipboard_label(label)
|
||||
|
||||
timer = asyncio.create_task(clear_clipboard())
|
||||
_clipboard_timers[label] = timer
|
||||
|
||||
return {
|
||||
"success": True,
|
||||
"clear_after": clear_after,
|
||||
"expires_at": int(time.time()) + clear_after,
|
||||
}
|
||||
|
||||
except FileNotFoundError:
|
||||
return {"success": False, "error": "xclip/xsel not found. Install with: pacman -S xclip"}
|
||||
except Exception as e:
|
||||
return {"success": False, "error": str(e)}
|
||||
|
||||
|
||||
async def clear_clipboard_label(label: str = "default") -> dict:
|
||||
"""Clear the clipboard for a specific label."""
|
||||
global _clipboard_timers, _clipboard_contents
|
||||
|
||||
# Cancel timer if exists
|
||||
if label in _clipboard_timers:
|
||||
_clipboard_timers[label].cancel()
|
||||
del _clipboard_timers[label]
|
||||
|
||||
# Clear clipboard content
|
||||
if label in _clipboard_contents:
|
||||
del _clipboard_contents[label]
|
||||
|
||||
# Clear the actual clipboard
|
||||
try:
|
||||
process = await asyncio.create_subprocess_exec(
|
||||
"xclip", "-selection", "clipboard",
|
||||
stdin=asyncio.subprocess.PIPE,
|
||||
stdout=asyncio.subprocess.PIPE,
|
||||
stderr=asyncio.subprocess.PIPE,
|
||||
)
|
||||
await process.communicate(input=b"")
|
||||
except FileNotFoundError:
|
||||
pass
|
||||
|
||||
return {"success": True}
|
||||
|
||||
|
||||
async def clear_all_clipboards() -> dict:
|
||||
"""Clear all clipboard contents and cancel all timers."""
|
||||
global _clipboard_timers, _clipboard_contents
|
||||
|
||||
for label in list(_clipboard_timers.keys()):
|
||||
_clipboard_timers[label].cancel()
|
||||
|
||||
_clipboard_timers.clear()
|
||||
_clipboard_contents.clear()
|
||||
|
||||
await clear_clipboard_label("default")
|
||||
|
||||
return {"success": True}
|
||||
|
||||
|
||||
def get_clipboard_status(label: str = "default") -> dict:
|
||||
"""Get the status of clipboard content for a label."""
|
||||
global _clipboard_contents, _clipboard_timers
|
||||
|
||||
if label not in _clipboard_contents:
|
||||
return {"has_content": False}
|
||||
|
||||
return {
|
||||
"has_content": True,
|
||||
"label": label,
|
||||
"has_timer": label in _clipboard_timers,
|
||||
}
|
||||
@@ -0,0 +1,380 @@
|
||||
"""
|
||||
Bitwarden/Vaultwarden cryptography implementation.
|
||||
Handles KDF (PBKDF2, Argon2), AES-CBC-256, HMAC, HKDF, and vault decryption.
|
||||
"""
|
||||
import hashlib
|
||||
import hmac
|
||||
import os
|
||||
import struct
|
||||
from base64 import b64decode, b64encode
|
||||
from typing import Optional, Tuple
|
||||
|
||||
from cryptography.hazmat.primitives.ciphers import Cipher, algorithms, modes
|
||||
from cryptography.hazmat.primitives import hashes, padding
|
||||
from cryptography.hazmat.primitives.kdf.pbkdf2 import PBKDF2HMAC
|
||||
from cryptography.hazmat.backends import default_backend
|
||||
|
||||
# Bitwarden-specific constants
|
||||
ENCRYPTION_KEY_LENGTH = 32 # 256 bits
|
||||
MAC_KEY_LENGTH = 32
|
||||
HKDF_INFO_EMAIL = b"enc"
|
||||
HKDF_INFO_MASTER_PASSWORD = b"enc"
|
||||
|
||||
# Cipher type markers (ST arrays in Bitwarden)
|
||||
CIPHER_TYPE_AES_CBC_256_B64 = "2."
|
||||
CIPHER_TYPE_AES_CBC_256_HMAC_B64 = "3."
|
||||
CIPHER_TYPE_RSA_2048_OAEP_SHA256 = "4."
|
||||
|
||||
|
||||
class BitwardenCrypto:
|
||||
"""Core cryptographic operations for Bitwarden vault decryption."""
|
||||
|
||||
def __init__(self):
|
||||
self.backend = default_backend()
|
||||
|
||||
def derive_master_key_pbkdf2(
|
||||
self, password: str, email: str, iterations: int
|
||||
) -> bytes:
|
||||
"""Derive master key using PBKDF2-SHA256."""
|
||||
salt = email.lower().strip().encode("utf-8")
|
||||
kdf = PBKDF2HMAC(
|
||||
algorithm=hashes.SHA256(),
|
||||
length=ENCRYPTION_KEY_LENGTH,
|
||||
salt=salt,
|
||||
iterations=iterations,
|
||||
backend=self.backend,
|
||||
)
|
||||
return kdf.derive(password.encode("utf-8"))
|
||||
|
||||
def derive_master_key_argon2(
|
||||
self,
|
||||
password: str,
|
||||
email: str,
|
||||
iterations: int,
|
||||
memory: int,
|
||||
parallelism: int,
|
||||
) -> bytes:
|
||||
"""Derive master key using Argon2id."""
|
||||
try:
|
||||
import argon2
|
||||
salt = email.lower().strip().encode("utf-8")
|
||||
return argon2.low_level.hash_secret_raw(
|
||||
secret=password.encode("utf-8"),
|
||||
salt=salt,
|
||||
time_cost=iterations,
|
||||
memory_cost=memory * 1024, # Convert MB to KB
|
||||
parallelism=parallelism,
|
||||
hash_len=ENCRYPTION_KEY_LENGTH,
|
||||
type=argon2.low_level.Type.ID,
|
||||
)
|
||||
except ImportError:
|
||||
raise RuntimeError(
|
||||
"Argon2 support requires argon2-cffi: pip install argon2-cffi"
|
||||
)
|
||||
|
||||
def stretch_master_key(self, master_key: bytes) -> Tuple[bytes, bytes]:
|
||||
"""Stretch master key into encryption key + MAC key using HKDF."""
|
||||
# Bitwarden uses HKDF with empty salt and specific info strings
|
||||
prk = hmac.new(b"", master_key, hashlib.sha256).digest()
|
||||
|
||||
# Encryption key
|
||||
enc_key = hmac.new(
|
||||
prk, b"\x01" + HKDF_INFO_EMAIL + b"\x00", hashlib.sha256
|
||||
).digest()
|
||||
|
||||
# MAC key
|
||||
mac_key = hmac.new(
|
||||
prk, b"\x02" + HKDF_INFO_EMAIL + b"\x00", hashlib.sha256
|
||||
).digest()
|
||||
|
||||
return enc_key, mac_key
|
||||
|
||||
def decrypt_aes_cbc_256(
|
||||
self, key: bytes, iv: bytes, ciphertext: bytes
|
||||
) -> bytes:
|
||||
"""Decrypt data using AES-CBC-256."""
|
||||
cipher = Cipher(
|
||||
algorithms.AES(key), modes.CBC(iv), backend=self.backend
|
||||
)
|
||||
decryptor = cipher.decryptor()
|
||||
padded = decryptor.update(ciphertext) + decryptor.finalize()
|
||||
|
||||
# Remove PKCS7 padding
|
||||
unpadder = padding.PKCS7(128).unpadder()
|
||||
return unpadder.update(padded) + unpadder.finalize()
|
||||
|
||||
def hmac_sha256(self, key: bytes, data: bytes) -> bytes:
|
||||
"""Compute HMAC-SHA256."""
|
||||
return hmac.new(key, data, hashlib.sha256).digest()
|
||||
|
||||
def verify_mac(
|
||||
self, mac_key: bytes, data: bytes, expected_mac: bytes
|
||||
) -> bool:
|
||||
"""Verify HMAC-SHA256."""
|
||||
computed = self.hmac_sha256(mac_key, data)
|
||||
return hmac.compare_digest(computed, expected_mac)
|
||||
|
||||
def decrypt_cipher_string(self, enc_string: str, enc_key: bytes, mac_key: bytes) -> bytes:
|
||||
"""
|
||||
Decrypt a Bitwarden CipherString.
|
||||
|
||||
Format: TYPE.BASE64(IV) | BASE64(CT) | BASE64(MAC)
|
||||
or: TYPE.BASE64(IV).BASE64(CT).BASE64(MAC)
|
||||
"""
|
||||
# Determine cipher type
|
||||
if enc_string.startswith(CIPHER_TYPE_AES_CBC_256_B64):
|
||||
parts = enc_string[2:].split("|")
|
||||
if len(parts) == 2:
|
||||
# Format: TYPE.BASE64(IV)|BASE64(CT)
|
||||
iv_b64, ct_b64 = parts
|
||||
iv = b64decode(iv_b64)
|
||||
ct = b64decode(ct_b64)
|
||||
return self.decrypt_aes_cbc_256(enc_key, iv, ct)
|
||||
elif len(parts) == 3:
|
||||
# Format: TYPE.BASE64(IV)|BASE64(CT)|BASE64(MAC)
|
||||
iv_b64, ct_b64, mac_b64 = parts
|
||||
iv = b64decode(iv_b64)
|
||||
ct = b64decode(ct_b64)
|
||||
mac = b64decode(mac_b64)
|
||||
if not self.verify_mac(mac_key, iv + ct, mac):
|
||||
raise ValueError("MAC verification failed")
|
||||
return self.decrypt_aes_cbc_256(enc_key, iv, ct)
|
||||
elif enc_string.startswith(CIPHER_TYPE_AES_CBC_256_HMAC_B64):
|
||||
parts = enc_string[2:].split("|")
|
||||
if len(parts) == 3:
|
||||
iv_b64, ct_b64, mac_b64 = parts
|
||||
iv = b64decode(iv_b64)
|
||||
ct = b64decode(ct_b64)
|
||||
mac = b64decode(mac_b64)
|
||||
if not self.verify_mac(mac_key, iv + ct, mac):
|
||||
raise ValueError("MAC verification failed")
|
||||
return self.decrypt_aes_cbc_256(enc_key, iv, ct)
|
||||
elif len(parts) == 2:
|
||||
iv_b64, ct_b64 = parts
|
||||
iv = b64decode(iv_b64)
|
||||
ct = b64decode(ct_b64)
|
||||
return self.decrypt_aes_cbc_256(enc_key, iv, ct)
|
||||
|
||||
raise ValueError(f"Unsupported cipher type in: {enc_string[:10]}...")
|
||||
|
||||
def decrypt_user_key(
|
||||
self, encrypted_user_key: str, master_key: bytes
|
||||
) -> Tuple[bytes, bytes]:
|
||||
"""
|
||||
Decrypt the encrypted user key (Key from API response).
|
||||
Returns (encryption_key, mac_key).
|
||||
|
||||
The user key is encrypted with the stretched master key.
|
||||
"""
|
||||
stretched_enc, stretched_mac = self.stretch_master_key(master_key)
|
||||
user_key = self.decrypt_cipher_string(
|
||||
encrypted_user_key, stretched_enc, stretched_mac
|
||||
)
|
||||
|
||||
if len(user_key) == 64:
|
||||
# Has separate MAC key
|
||||
return user_key[:32], user_key[32:]
|
||||
elif len(user_key) == 32:
|
||||
# No separate MAC key, derive from the key itself
|
||||
return user_key, user_key
|
||||
else:
|
||||
raise ValueError(f"Unexpected user key length: {len(user_key)}")
|
||||
|
||||
def decrypt_cipher(
|
||||
self,
|
||||
cipher_data: dict,
|
||||
enc_key: bytes,
|
||||
mac_key: bytes,
|
||||
) -> Optional[dict]:
|
||||
"""Decrypt a single cipher (vault item)."""
|
||||
try:
|
||||
# Decrypt name
|
||||
name = ""
|
||||
if cipher_data.get("name"):
|
||||
try:
|
||||
name = self.decrypt_cipher_string(
|
||||
cipher_data["name"], enc_key, mac_key
|
||||
).decode("utf-8")
|
||||
except Exception:
|
||||
name = "[encrypted]"
|
||||
|
||||
# Decrypt fields based on type
|
||||
result = {
|
||||
"id": cipher_data.get("id"),
|
||||
"type": cipher_data.get("type"),
|
||||
"name": name,
|
||||
"folderId": cipher_data.get("folderId"),
|
||||
"organizationId": cipher_data.get("organizationId"),
|
||||
"favorite": cipher_data.get("favorite", False),
|
||||
"revisionDate": cipher_data.get("revisionDate"),
|
||||
}
|
||||
|
||||
cipher_type = cipher_data.get("type")
|
||||
|
||||
if cipher_type == 1: # Login
|
||||
login = cipher_data.get("login", {})
|
||||
result["login"] = {
|
||||
"username": self._decrypt_field(
|
||||
login.get("username"), enc_key, mac_key
|
||||
),
|
||||
"password": self._decrypt_field(
|
||||
login.get("password"), enc_key, mac_key
|
||||
),
|
||||
"totp": self._decrypt_field(
|
||||
login.get("totp"), enc_key, mac_key
|
||||
),
|
||||
"uris": [
|
||||
{
|
||||
"uri": self._decrypt_field(
|
||||
u.get("uri"), enc_key, mac_key
|
||||
),
|
||||
"match": u.get("match"),
|
||||
}
|
||||
for u in login.get("uris", [])
|
||||
],
|
||||
}
|
||||
elif cipher_type == 2: # Secure Note
|
||||
result["notes"] = self._decrypt_field(
|
||||
cipher_data.get("notes"), enc_key, mac_key
|
||||
)
|
||||
elif cipher_type == 3: # Card
|
||||
card = cipher_data.get("card", {})
|
||||
result["card"] = {
|
||||
"cardholderName": self._decrypt_field(
|
||||
card.get("cardholderName"), enc_key, mac_key
|
||||
),
|
||||
"brand": self._decrypt_field(
|
||||
card.get("brand"), enc_key, mac_key
|
||||
),
|
||||
"number": self._decrypt_field(
|
||||
card.get("number"), enc_key, mac_key
|
||||
),
|
||||
"expMonth": self._decrypt_field(
|
||||
card.get("expMonth"), enc_key, mac_key
|
||||
),
|
||||
"expYear": self._decrypt_field(
|
||||
card.get("expYear"), enc_key, mac_key
|
||||
),
|
||||
}
|
||||
elif cipher_type == 4: # Identity
|
||||
identity = cipher_data.get("identity", {})
|
||||
result["identity"] = {
|
||||
"firstName": self._decrypt_field(
|
||||
identity.get("firstName"), enc_key, mac_key
|
||||
),
|
||||
"lastName": self._decrypt_field(
|
||||
identity.get("lastName"), enc_key, mac_key
|
||||
),
|
||||
"email": self._decrypt_field(
|
||||
identity.get("email"), enc_key, mac_key
|
||||
),
|
||||
"phone": self._decrypt_field(
|
||||
identity.get("phone"), enc_key, mac_key
|
||||
),
|
||||
}
|
||||
|
||||
# Decrypt custom fields
|
||||
fields = []
|
||||
for field in cipher_data.get("fields", []):
|
||||
fields.append({
|
||||
"name": self._decrypt_field(
|
||||
field.get("name"), enc_key, mac_key
|
||||
),
|
||||
"value": self._decrypt_field(
|
||||
field.get("value"), enc_key, mac_key
|
||||
),
|
||||
"type": field.get("type"),
|
||||
"hidden": field.get("hidden", False),
|
||||
})
|
||||
result["fields"] = fields
|
||||
|
||||
# Decrypt notes
|
||||
if cipher_data.get("notes") and cipher_type != 2:
|
||||
result["notes"] = self._decrypt_field(
|
||||
cipher_data.get("notes"), enc_key, mac_key
|
||||
)
|
||||
|
||||
return result
|
||||
except Exception as e:
|
||||
# Return partial result with error info
|
||||
return {
|
||||
"id": cipher_data.get("id"),
|
||||
"type": cipher_data.get("type"),
|
||||
"name": f"[decryption error: {str(e)}]",
|
||||
"error": True,
|
||||
}
|
||||
|
||||
def _decrypt_field(
|
||||
self,
|
||||
value: Optional[str],
|
||||
enc_key: bytes,
|
||||
mac_key: bytes,
|
||||
) -> Optional[str]:
|
||||
"""Decrypt a single field value."""
|
||||
if not value:
|
||||
return None
|
||||
try:
|
||||
decrypted = self.decrypt_cipher_string(value, enc_key, mac_key)
|
||||
return decrypted.decode("utf-8")
|
||||
except Exception:
|
||||
return "[encrypted]"
|
||||
|
||||
|
||||
# PIN-related crypto
|
||||
class PinCrypto:
|
||||
"""Handles PIN-based vault unlock (PasswordProtectedKeyEnvelope)."""
|
||||
|
||||
def __init__(self):
|
||||
self.crypto = BitwardenCrypto()
|
||||
|
||||
def derive_pin_key(
|
||||
self, pin: str, salt: str, kdf_iterations: int = 200000
|
||||
) -> bytes:
|
||||
"""Derive a key from the PIN using PBKDF2."""
|
||||
kdf = PBKDF2HMAC(
|
||||
algorithm=hashes.SHA256(),
|
||||
length=ENCRYPTION_KEY_LENGTH,
|
||||
salt=salt.encode("utf-8") if isinstance(salt, str) else salt,
|
||||
iterations=kdf_iterations,
|
||||
backend=default_backend(),
|
||||
)
|
||||
return kdf.derive(pin.encode("utf-8"))
|
||||
|
||||
def encrypt_user_key_for_pin(
|
||||
self, user_key: bytes, pin_key: bytes
|
||||
) -> Tuple[bytes, bytes]:
|
||||
"""Encrypt the user key with the PIN-derived key. Returns (encrypted_key, iv)."""
|
||||
iv = os.urandom(16)
|
||||
cipher = Cipher(
|
||||
algorithms.AES(pin_key), modes.CBC(iv), backend=default_backend()
|
||||
)
|
||||
encryptor = cipher.encryptor()
|
||||
|
||||
# Pad user key with PKCS7
|
||||
padder = padding.PKCS7(128).padder()
|
||||
padded = padder.update(user_key) + padder.finalize()
|
||||
|
||||
encrypted = encryptor.update(padded) + encryptor.finalize()
|
||||
|
||||
# Compute HMAC
|
||||
mac = self.crypto.hmac_sha256(pin_key, iv + encrypted)
|
||||
|
||||
return iv + encrypted + mac, iv
|
||||
|
||||
def decrypt_user_key_with_pin(
|
||||
self, encrypted_envelope: bytes, pin_key: bytes
|
||||
) -> bytes:
|
||||
"""Decrypt the user key using the PIN-derived key."""
|
||||
# Envelope format: IV (16) + EncryptedData (32) + MAC (32)
|
||||
if len(encrypted_envelope) < 80:
|
||||
raise ValueError("Invalid PIN envelope length")
|
||||
|
||||
iv = encrypted_envelope[:16]
|
||||
mac = encrypted_envelope[-32:]
|
||||
ct = encrypted_envelope[16:-32]
|
||||
|
||||
# Verify HMAC
|
||||
if not self.crypto.verify_mac(pin_key, iv + ct, mac):
|
||||
raise ValueError("PIN MAC verification failed")
|
||||
|
||||
return self.crypto.decrypt_aes_cbc_256(pin_key, iv, ct)
|
||||
@@ -0,0 +1,107 @@
|
||||
"""
|
||||
TOTP (Time-based One-Time Password) generation for Bitwarden vault items.
|
||||
"""
|
||||
import hashlib
|
||||
import hmac
|
||||
import struct
|
||||
import time
|
||||
from typing import Optional
|
||||
|
||||
# Try to use pyotp if available, otherwise use manual implementation
|
||||
try:
|
||||
import pyotp
|
||||
HAS_PYOTP = True
|
||||
except ImportError:
|
||||
HAS_PYOTP = False
|
||||
|
||||
|
||||
def generate_totp(secret: str, period: int = 30, digits: int = 6) -> str:
|
||||
"""Generate a TOTP code from a secret."""
|
||||
if not secret:
|
||||
return ""
|
||||
|
||||
# Clean the secret (remove spaces, convert to uppercase)
|
||||
secret = secret.replace(" ", "").upper()
|
||||
|
||||
# Try to parse as otpauth:// URI
|
||||
if secret.startswith("otpauth://"):
|
||||
parsed = _parse_otpauth_uri(secret)
|
||||
if parsed:
|
||||
secret = parsed["secret"]
|
||||
period = parsed.get("period", period)
|
||||
digits = parsed.get("digits", digits)
|
||||
|
||||
if HAS_PYOTP:
|
||||
totp = pyotp.TOTP(secret, interval=period, digits=digits)
|
||||
return totp.now()
|
||||
|
||||
return _generate_totp_manual(secret, period, digits)
|
||||
|
||||
|
||||
def get_totp_remaining_seconds(period: int = 30) -> int:
|
||||
"""Get seconds remaining until current TOTP code expires."""
|
||||
return period - (int(time.time()) % period)
|
||||
|
||||
|
||||
def _generate_totp_manual(secret: str, period: int, digits: int) -> str:
|
||||
"""Manual TOTP implementation when pyotp is not available."""
|
||||
# Decode base32 secret
|
||||
_BASE32_CHARS = "ABCDEFGHIJKLMNOPQRSTUVWXYZ234567"
|
||||
secret = secret.upper()
|
||||
|
||||
# Remove padding
|
||||
padding_needed = (8 - len(secret) % 8) % 8
|
||||
secret += "=" * padding_needed
|
||||
|
||||
# Decode base32
|
||||
binary = b""
|
||||
for char in secret:
|
||||
if char == "=":
|
||||
continue
|
||||
try:
|
||||
val = _BASE32_CHARS.index(char)
|
||||
except ValueError:
|
||||
continue
|
||||
binary += struct.pack(">B", val)
|
||||
|
||||
# Time counter
|
||||
counter = int(time.time()) // period
|
||||
counter_bytes = struct.pack(">Q", counter)
|
||||
|
||||
# HMAC-SHA1
|
||||
hmac_result = hmac.new(binary, counter_bytes, hashlib.sha1).digest()
|
||||
|
||||
# Dynamic truncation
|
||||
offset = hmac_result[-1] & 0x0F
|
||||
truncated = struct.unpack(
|
||||
">I", hmac_result[offset : offset + 4]
|
||||
)[0]
|
||||
truncated &= 0x7FFFFFFF
|
||||
|
||||
# Generate code
|
||||
code = truncated % (10 ** digits)
|
||||
return str(code).zfill(digits)
|
||||
|
||||
|
||||
def _parse_otpauth_uri(uri: str) -> Optional[dict]:
|
||||
"""Parse an otpauth:// URI."""
|
||||
# otpauth://totp/Label?secret=XXX&issuer=XXX&period=30&digits=6
|
||||
if not uri.startswith("otpauth://"):
|
||||
return None
|
||||
|
||||
parts = uri.split("?", 1)
|
||||
if len(parts) < 2:
|
||||
return None
|
||||
|
||||
params = {}
|
||||
for param in parts[1].split("&"):
|
||||
key, _, value = param.partition("=")
|
||||
params[key] = value
|
||||
|
||||
secret = params.get("secret", "")
|
||||
return {
|
||||
"secret": secret,
|
||||
"issuer": params.get("issuer", ""),
|
||||
"period": int(params.get("period", "30")),
|
||||
"digits": int(params.get("digits", "6")),
|
||||
}
|
||||
Reference in New Issue
Block a user