crypto.py (14699B)
1 """ 2 Obsidian Sync Encryption - Reversed from Obsidian 1.11.7 app.js 3 4 Three encryption versions are supported: 5 Version 0: AES-GCM with raw key (no salt). Key derived via scrypt(password, salt, N=32768, r=8, p=1, dklen=32). 6 Path/filename encryption uses AES-GCM with IV = first 12 bytes of SHA-256(plaintext). 7 KeyHash = hex(SHA-256(key)). 8 9 Version 2: AES-GCM for file content + AES-SIV for deterministic path encoding. 10 Key derived via scrypt as above. HKDF(key, salt, "ObsidianKeyHash") -> keyHash. 11 Path encoding: AES-SIV seal/open (using AES-CMAC-SIV). 12 File content: AES-GCM encrypt (random IV). 13 14 Version 3: Same as version 2 but different SIV implementation detail (handled identically in Python). 15 16 Scrypt parameters: N=32768, r=8, p=1, dklen=32. 17 """ 18 19 import os 20 import hashlib 21 import hmac 22 import struct 23 from typing import Optional 24 25 from cryptography.hazmat.primitives.ciphers.aead import AESGCM 26 from cryptography.hazmat.primitives.kdf.hkdf import HKDF 27 from cryptography.hazmat.primitives.kdf.scrypt import Scrypt 28 from cryptography.hazmat.primitives import hashes 29 from cryptography.hazmat.backends import default_backend 30 31 32 # --------------------------------------------------------------------------- 33 # Scrypt key derivation 34 # --------------------------------------------------------------------------- 35 36 def derive_key(password: str, salt: str) -> bytes: 37 """Derive 32-byte encryption key from password and salt using scrypt. 38 39 Matches wZ() in app.js: 40 scrypt(password.normalize("NFKC"), salt.normalize("NFKC"), N=32768, r=8, p=1, dklen=32) 41 """ 42 import unicodedata 43 password = unicodedata.normalize("NFKC", password) 44 salt_str = unicodedata.normalize("NFKC", salt) 45 kdf = Scrypt( 46 salt=salt_str.encode("utf-8"), 47 length=32, 48 n=32768, 49 r=8, 50 p=1, 51 backend=default_backend(), 52 ) 53 return kdf.derive(password.encode("utf-8")) 54 55 56 # --------------------------------------------------------------------------- 57 # Low-level AES-GCM (version 0 and content encryption for v2/v3) 58 # --------------------------------------------------------------------------- 59 60 def aes_gcm_encrypt(data: bytes, key: bytes, iv: Optional[bytes] = None) -> bytes: 61 """Encrypt data with AES-GCM. Returns iv (12 bytes) + ciphertext+tag. 62 63 Matches iZ() in app.js. 64 """ 65 if iv is None: 66 iv = os.urandom(12) 67 aesgcm = AESGCM(key) 68 ciphertext = aesgcm.encrypt(iv, data, None) 69 return iv + ciphertext 70 71 72 def aes_gcm_decrypt(data: bytes, key: bytes) -> bytes: 73 """Decrypt data with AES-GCM. Input format: iv (12 bytes) + ciphertext+tag. 74 75 Matches rZ() in app.js. 76 """ 77 if len(data) < 12: 78 raise ValueError("Encrypted data too short") 79 if len(data) == 12: 80 return b"" 81 iv = data[:12] 82 ciphertext = data[12:] 83 aesgcm = AESGCM(key) 84 return aesgcm.decrypt(iv, ciphertext, None) 85 86 87 # --------------------------------------------------------------------------- 88 # AES-SIV for deterministic path encoding (versions 2 & 3) 89 # --------------------------------------------------------------------------- 90 91 class CMAC: 92 """AES-CMAC implementation matching the browser's SubtleCrypto CMAC.""" 93 94 def __init__(self, key: bytes): 95 self.key = key 96 self._generate_subkeys() 97 self._buffer = bytearray(16) 98 self._buffer_pos = 0 99 self._finished = False 100 # Initialize cipher 101 from cryptography.hazmat.primitives.ciphers import Cipher, algorithms, modes 102 self._cipher_class = lambda k: Cipher(algorithms.AES(k), modes.ECB(), backend=default_backend()) 103 104 def _encrypt_block(self, block: bytes) -> bytes: 105 from cryptography.hazmat.primitives.ciphers import Cipher, algorithms, modes 106 cipher = Cipher(algorithms.AES(self.key), modes.ECB(), backend=default_backend()) 107 enc = cipher.encryptor() 108 return enc.update(bytes(block)) + enc.finalize() 109 110 def _generate_subkeys(self): 111 """Generate CMAC subkeys k1 and k2.""" 112 L = self._encrypt_block(bytes(16)) 113 114 def dbl(b: bytes) -> bytes: 115 b = bytearray(b) 116 overflow = 0 117 result = bytearray(16) 118 for i in range(15, -1, -1): 119 result[i] = ((b[i] << 1) | overflow) & 0xFF 120 overflow = b[i] >> 7 121 if overflow: 122 result[15] ^= 0x87 123 return bytes(result) 124 125 self.k1 = dbl(L) 126 self.k2 = dbl(self.k1) 127 128 def update(self, data: bytes) -> None: 129 for byte in data: 130 self._buffer[self._buffer_pos] ^= byte 131 self._buffer_pos += 1 132 if self._buffer_pos == 16: 133 # Encrypt the block (not the last one) 134 encrypted = self._encrypt_block(bytes(self._buffer)) 135 self._buffer = bytearray(encrypted) 136 self._buffer_pos = 0 137 138 def finish(self) -> bytes: 139 if not self._finished: 140 subkey = self.k2 if self._buffer_pos < 16 else self.k1 141 buf = bytearray(self._buffer) 142 for i, k in enumerate(subkey): 143 buf[i] ^= k 144 if self._buffer_pos < 16: 145 buf[self._buffer_pos] ^= 0x80 146 encrypted = self._encrypt_block(bytes(buf)) 147 self._finished = True 148 return encrypted 149 return bytes(self._buffer) 150 151 152 def _aes_ctr_encrypt(key: bytes, iv: bytes, data: bytes) -> bytes: 153 """AES-CTR encryption. Used by AES-SIV.""" 154 from cryptography.hazmat.primitives.ciphers import Cipher, algorithms, modes 155 cipher = Cipher(algorithms.AES(key), modes.CTR(iv + b'\x00\x00\x00\x00'), backend=default_backend()) 156 enc = cipher.encryptor() 157 return enc.update(data) + enc.finalize() 158 159 160 class AESSIV: 161 """AES-SIV (Synthetic IV) implementation for deterministic encryption. 162 163 Matches vZ (the SIV class) in app.js. 164 Used for versions 2 & 3 path encoding. 165 166 Key setup: Two keys from HKDF: 167 - k1: for CMAC (AES-CMAC key) 168 - k2: for CTR (AES-CTR key) 169 Both keys imported from the same base HKDF-derived material. 170 """ 171 172 def __init__(self, cmac_key: bytes, ctr_key: bytes): 173 self.cmac_key = cmac_key 174 self.ctr_key = ctr_key 175 176 @classmethod 177 def from_raw_key(cls, key: bytes) -> "AESSIV": 178 """Create from a raw 32-byte key (used for both CMAC and CTR).""" 179 return cls(key, key) 180 181 def _s2v(self, data: bytes) -> bytes: 182 """S2V computation.""" 183 # Initial CMAC of zeros 184 cmac1 = CMAC(self.cmac_key) 185 cmac1.update(bytes(16)) 186 d = bytearray(cmac1.finish()) 187 188 # DBL(d) 189 d = bytearray(self._dbl(bytes(d))) 190 191 if len(data) >= 16: 192 # XOR last 16 bytes of data with d 193 tail_start = len(data) - 16 194 msg_prefix = data[:tail_start] 195 msg_tail = data[tail_start:] 196 197 cmac2 = CMAC(self.cmac_key) 198 cmac2.update(msg_prefix) 199 # XOR last block with d 200 final_block = bytearray(msg_tail) 201 for i in range(16): 202 final_block[i] ^= d[i] 203 cmac2.update(bytes(final_block)) 204 return cmac2.finish() 205 else: 206 # Pad and XOR 207 padded = bytearray(data) + bytearray([0x80]) + bytearray(16 - len(data) - 1) 208 d = bytearray(self._dbl(bytes(d))) 209 for i in range(16): 210 padded[i] ^= d[i] 211 cmac2 = CMAC(self.cmac_key) 212 cmac2.update(bytes(padded)) 213 return cmac2.finish() 214 215 def _dbl(self, b: bytes) -> bytes: 216 """Double in GF(2^128).""" 217 b = bytearray(b) 218 overflow = 0 219 result = bytearray(16) 220 for i in range(15, -1, -1): 221 result[i] = ((b[i] << 1) | overflow) & 0xFF 222 overflow = b[i] >> 7 223 if overflow: 224 result[15] ^= 0x87 225 return bytes(result) 226 227 def _clear_bits(self, iv: bytes) -> bytes: 228 """Clear bits 31 and 63 counting from right (gZ function).""" 229 iv = bytearray(iv) 230 # iv[length-8] &= 127 (clear bit 7 of 8th-from-last byte) 231 # iv[length-4] &= 127 (clear bit 7 of 4th-from-last byte) 232 iv[-8] &= 0x7F 233 iv[-4] &= 0x7F 234 return bytes(iv) 235 236 def seal(self, plaintext: bytes) -> bytes: 237 """Encrypt with AES-SIV. Returns tag (16 bytes) + ciphertext.""" 238 siv = self._s2v(plaintext) 239 iv = self._clear_bits(siv) 240 ciphertext = _aes_ctr_encrypt(self.ctr_key, iv, plaintext) 241 return siv + ciphertext 242 243 def open(self, ciphertext: bytes) -> bytes: 244 """Decrypt with AES-SIV. Input: tag (16 bytes) + ciphertext.""" 245 if len(ciphertext) < 16: 246 raise ValueError("AES-SIV: ciphertext is truncated") 247 tag = ciphertext[:16] 248 ct = ciphertext[16:] 249 iv = self._clear_bits(tag) 250 plaintext = _aes_ctr_encrypt(self.ctr_key, iv, ct) 251 252 # Verify tag 253 expected = self._s2v(plaintext) 254 if not hmac.compare_digest(tag, expected): 255 raise ValueError("AES-SIV: ciphertext verification failure!") 256 return plaintext 257 258 259 # --------------------------------------------------------------------------- 260 # Encryption providers (matching EZ and MZ classes in app.js) 261 # --------------------------------------------------------------------------- 262 263 class EncryptionV0: 264 """Version 0: AES-GCM with raw 32-byte key. 265 266 - keyHash = hex(SHA-256(key)) 267 - deterministicEncodeStr: AES-GCM with IV = first 12 bytes of SHA-256(plaintext) 268 - encrypt/decrypt: AES-GCM (random IV for encrypt) 269 """ 270 271 encryption_version = 0 272 273 def __init__(self, key: bytes): 274 if len(key) != 32: 275 raise ValueError("Invalid encryption key, must be 32 bytes") 276 self.key = key 277 self.key_hash = hashlib.sha256(key).hexdigest() 278 279 @classmethod 280 def from_password(cls, password: str, salt: str) -> "EncryptionV0": 281 key = derive_key(password, salt) 282 return cls(key) 283 284 def deterministic_encode_str(self, text: str) -> str: 285 """Deterministically encrypt a string (used for file paths). 286 287 IV = first 12 bytes of SHA-256(utf8(text)). 288 Returns hex string. 289 """ 290 data = text.encode("utf-8") 291 iv = hashlib.sha256(data).digest()[:12] 292 return aes_gcm_encrypt(data, self.key, iv).hex() 293 294 def deterministic_decode_str(self, hex_str: str) -> str: 295 """Decrypt a deterministically encoded string.""" 296 data = bytes.fromhex(hex_str) 297 return aes_gcm_decrypt(data, self.key).decode("utf-8") 298 299 def encrypt(self, data: bytes) -> bytes: 300 """Encrypt data (random IV).""" 301 return aes_gcm_encrypt(data, self.key) 302 303 def decrypt(self, data: bytes) -> bytes: 304 """Decrypt data.""" 305 return aes_gcm_decrypt(data, self.key) 306 307 308 class EncryptionV2: 309 """Version 2/3: AES-SIV for paths + AES-GCM for content. 310 311 Matches MZ class in app.js. 312 313 Key derivation: 314 raw_key = scrypt(password, salt) 315 hkdf_key = HKDF(raw_key, "HKDF") 316 key_hash = hex(HKDF(raw_key, salt, "ObsidianKeyHash", AES-CBC-256)) 317 siv = AES-SIV from HKDF (same hkdf_key base) 318 aes_gcm_key = HKDF(raw_key, [], "ObsidianAesGcm", AES-GCM-256) 319 """ 320 321 def __init__(self, encryption_version: int, raw_key: bytes, salt: str): 322 if encryption_version not in (2, 3): 323 raise ValueError("Invalid encryption version, must be 2 or 3") 324 self.encryption_version = encryption_version 325 326 salt_bytes = salt.encode("utf-8") 327 328 # Import raw key for HKDF 329 backend = default_backend() 330 331 # Derive key hash: HKDF(raw_key, salt, "ObsidianKeyHash") -> AES-CBC-256 -> export -> hex 332 hkdf_keyhash = HKDF( 333 algorithm=hashes.SHA256(), 334 length=32, 335 salt=salt_bytes, 336 info=b"ObsidianKeyHash", 337 backend=backend, 338 ).derive(raw_key) 339 self.key_hash = hkdf_keyhash.hex() 340 341 # Derive AES-GCM key: HKDF(raw_key, [], "ObsidianAesGcm") 342 self.gcm_key = HKDF( 343 algorithm=hashes.SHA256(), 344 length=32, 345 salt=b"", 346 info=b"ObsidianAesGcm", 347 backend=backend, 348 ).derive(raw_key) 349 350 # AES-SIV: uses the HKDF base key material directly 351 # vZ.importKey() in app.js uses the same hkdf_key for both CMAC and CTR 352 # The importKey takes (r, key) where r is HKDF key handle and key is raw HKDF input 353 # Looking at vZ.importKey: it imports `r` (the HKDF key) and derives for CMAC and CTR 354 # Actually looking more carefully: vZ.importKey(r, o) where r is hkdf imported key and o is salt_bytes 355 # CMAC key: importKey("raw", key, ...) - the r param is already the hkdf_key handle 356 # This is the HKDF base key material (raw_key) used for CMAC and CTR separately 357 # The app uses the same imported HKDF key for both mZ (CMAC) and pZ (CTR) 358 # Both use the same raw key bytes 359 self.siv = AESSIV.from_raw_key(raw_key) 360 361 @classmethod 362 def from_password(cls, password: str, salt: str, encryption_version: int = 2) -> "EncryptionV2": 363 raw_key = derive_key(password, salt) 364 return cls(encryption_version, raw_key, salt) 365 366 def deterministic_encode_str(self, text: str) -> str: 367 """Encrypt path with AES-SIV (deterministic). Returns hex string.""" 368 data = text.encode("utf-8") 369 sealed = self.siv.seal(data) 370 return sealed.hex() 371 372 def deterministic_decode_str(self, hex_str: str) -> str: 373 """Decrypt SIV-encrypted path from hex string.""" 374 data = bytes.fromhex(hex_str) 375 return self.siv.open(data).decode("utf-8") 376 377 def encrypt(self, data: bytes) -> bytes: 378 """Encrypt file content with AES-GCM (random IV).""" 379 return aes_gcm_encrypt(data, self.gcm_key) 380 381 def decrypt(self, data: bytes) -> bytes: 382 """Decrypt file content.""" 383 return aes_gcm_decrypt(data, self.gcm_key) 384 385 386 def create_encryption_provider(password: str, salt: str, encryption_version: int): 387 """Factory function to create the appropriate encryption provider.""" 388 if encryption_version == 0: 389 return EncryptionV0.from_password(password, salt) 390 elif encryption_version in (2, 3): 391 return EncryptionV2.from_password(password, salt, encryption_version) 392 else: 393 raise ValueError(f"Unsupported encryption version: {encryption_version}")