""" Bitwarden/Vaultwarden cryptography implementation. Handles KDF (PBKDF2, Argon2), AES-CBC-256, HMAC, HKDF, and vault decryption. """ import hashlib import hmac import logging import os import struct from base64 import b64decode, b64encode from typing import Optional, Tuple _debug_log = "/tmp/decky-vaultwarden-debug.log" def _debug(msg: str): import datetime ts = datetime.datetime.now().isoformat() try: with open(_debug_log, "a") as f: f.write(f"[{ts}] {msg}\n") f.flush() except Exception: pass 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_ENC = b"enc" HKDF_INFO_MAC = b"mac" # 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-expand.""" _debug(f"[CRYPTO] stretch_master_key: master_key (first 8 b64)={b64encode(master_key[:8]).decode()}") _debug(f"[CRYPTO] stretch_master_key: master_key length={len(master_key)}") # Bitwarden SDK: HKDF-expand(key, info) = HMAC-SHA256(key, info || 0x01) enc_key = hmac.new(master_key, b"enc\x01", hashlib.sha256).digest() mac_key = hmac.new(master_key, b"mac\x01", hashlib.sha256).digest() _debug(f"[CRYPTO] stretch: enc_key (first 8 b64)={b64encode(enc_key[:8]).decode()}") _debug(f"[CRYPTO] stretch: mac_key (first 8 b64)={b64encode(mac_key[:8]).decode()}") 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) """ _debug(f"[CRYPTO] decrypt_cipher_string: type={enc_string[:2]}, full={enc_string[:50]}...") # Determine cipher type if enc_string.startswith(CIPHER_TYPE_AES_CBC_256_B64): parts = enc_string[2:].split("|") _debug(f"[CRYPTO] decrypt_cipher_string: type 2, parts count={len(parts)}") 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) _debug(f"[CRYPTO] decrypt_cipher_string: verifying MAC...") computed_mac = self.hmac_sha256(mac_key, iv + ct) _debug(f"[CRYPTO] decrypt_cipher_string: computed_mac (b64)={b64encode(computed_mac).decode()}") _debug(f"[CRYPTO] decrypt_cipher_string: expected_mac (b64)={mac_b64}") _debug(f"[CRYPTO] decrypt_cipher_string: mac_keys_equal={computed_mac == mac}") _debug(f"[CRYPTO] decrypt_cipher_string: enc_key (first 8 b64)={b64encode(enc_key[:8]).decode()}") _debug(f"[CRYPTO] decrypt_cipher_string: mac_key (first 8 b64)={b64encode(mac_key[:8]).decode()}") 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. """ _debug(f"[CRYPTO] decrypt_user_key: encrypted_user_key (first 30)={encrypted_user_key[:30]}...") stretched_enc, stretched_mac = self.stretch_master_key(master_key) _debug(f"[CRYPTO] decrypt_user_key: calling decrypt_cipher_string...") user_key = self.decrypt_cipher_string( encrypted_user_key, stretched_enc, stretched_mac ) _debug(f"[CRYPTO] decrypt_user_key: decrypted user_key length={len(user_key)}") 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)