381 lines
14 KiB
Python
381 lines
14 KiB
Python
"""
|
|
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)
|