mirror of
https://github.com/LifeArchiveProject/WeChatDataAnalysis.git
synced 2026-06-18 15:54:08 +08:00
fix(decrypt): 修复 SQLCipher 密钥兼容性回归
- 解密时同时兼容 raw enc_key 与 SQLCipher 派生密钥两种输入形态 - 通过首页 HMAC 自动识别可用密钥模式,避免真实账号密钥被误判为不匹配 - 后续页面解密统一使用识别出的有效密钥,恢复数据库解密流程 - 补充 SQLCipher passphrase 场景回归测试,覆盖此次回归问题
This commit is contained in:
@@ -19,11 +19,22 @@ from wechat_decrypt_tool.wechat_decrypt import (
|
||||
SQLITE_HEADER,
|
||||
WeChatDatabaseDecryptor,
|
||||
_derive_mac_key,
|
||||
_derive_sqlcipher_enc_key,
|
||||
decrypt_wechat_databases,
|
||||
)
|
||||
|
||||
|
||||
def _encrypt_page(raw_key: bytes, plain_page: bytes, page_num: int, salt: bytes, iv: bytes) -> bytes:
|
||||
def _encrypt_page(
|
||||
raw_key: bytes,
|
||||
plain_page: bytes,
|
||||
page_num: int,
|
||||
salt: bytes,
|
||||
iv: bytes,
|
||||
*,
|
||||
sqlcipher_passphrase: bool = False,
|
||||
) -> bytes:
|
||||
enc_key = _derive_sqlcipher_enc_key(raw_key, salt) if sqlcipher_passphrase else raw_key
|
||||
|
||||
if page_num == 1:
|
||||
encrypted_input = plain_page[SALT_SIZE : PAGE_SIZE - RESERVE_SIZE]
|
||||
prefix = salt
|
||||
@@ -32,7 +43,7 @@ def _encrypt_page(raw_key: bytes, plain_page: bytes, page_num: int, salt: bytes,
|
||||
prefix = b""
|
||||
|
||||
cipher = Cipher(
|
||||
algorithms.AES(raw_key),
|
||||
algorithms.AES(enc_key),
|
||||
modes.CBC(iv),
|
||||
backend=default_backend(),
|
||||
)
|
||||
@@ -40,7 +51,7 @@ def _encrypt_page(raw_key: bytes, plain_page: bytes, page_num: int, salt: bytes,
|
||||
encrypted = encryptor.update(encrypted_input) + encryptor.finalize()
|
||||
|
||||
page_without_hmac = prefix + encrypted + iv
|
||||
mac = hmac.new(_derive_mac_key(raw_key, salt), digestmod=hashlib.sha512)
|
||||
mac = hmac.new(_derive_mac_key(enc_key, salt), digestmod=hashlib.sha512)
|
||||
mac.update(page_without_hmac[SALT_SIZE if page_num == 1 else 0 :])
|
||||
mac.update(page_num.to_bytes(4, "little"))
|
||||
return page_without_hmac + mac.digest()
|
||||
@@ -74,6 +85,39 @@ class WeChatDecryptRawKeyTests(unittest.TestCase):
|
||||
self.assertTrue(decryptor.decrypt_database(str(src), str(dst)))
|
||||
self.assertEqual(dst.read_bytes(), page1 + page2)
|
||||
|
||||
def test_decrypt_database_falls_back_to_sqlcipher_passphrase_mode(self):
|
||||
passphrase_key = bytes.fromhex("9f5dd0d3b6d0477ea5045c9e380ee272e53927993eb548dd98a022e842d5f7bd")
|
||||
salt = bytes.fromhex("50f4090ef6897e146f94109f13743e34")
|
||||
iv1 = bytes.fromhex("0102030405060708090a0b0c0d0e0f10")
|
||||
iv2 = bytes.fromhex("1112131415161718191a1b1c1d1e1f20")
|
||||
|
||||
page1 = _build_plain_page(0x41, first_page=True)
|
||||
page2 = _build_plain_page(0x42, first_page=False)
|
||||
encrypted_db = _encrypt_page(
|
||||
passphrase_key,
|
||||
page1,
|
||||
1,
|
||||
salt,
|
||||
iv1,
|
||||
sqlcipher_passphrase=True,
|
||||
) + _encrypt_page(
|
||||
passphrase_key,
|
||||
page2,
|
||||
2,
|
||||
salt,
|
||||
iv2,
|
||||
sqlcipher_passphrase=True,
|
||||
)
|
||||
|
||||
with tempfile.TemporaryDirectory() as tmpdir:
|
||||
src = Path(tmpdir) / "source.db"
|
||||
dst = Path(tmpdir) / "out.db"
|
||||
src.write_bytes(encrypted_db)
|
||||
|
||||
decryptor = WeChatDatabaseDecryptor(passphrase_key.hex())
|
||||
self.assertTrue(decryptor.decrypt_database(str(src), str(dst)))
|
||||
self.assertEqual(dst.read_bytes(), page1 + page2)
|
||||
|
||||
def test_decrypt_database_keeps_existing_output_on_hmac_failure(self):
|
||||
good_key = bytes.fromhex("00112233445566778899aabbccddeefffedcba98765432100123456789abcdef")
|
||||
bad_key_hex = "ffeeddccbbaa998877665544332211000123456789abcdeffedcba9876543210"
|
||||
|
||||
Reference in New Issue
Block a user