""" 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 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.""" # 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() 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) computed_mac = self.hmac_sha256(mac_key, iv + ct) 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 (only for non-Note types, Notes already handled above) 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)