164 lines
5.4 KiB
Python
164 lines
5.4 KiB
Python
"""
|
|
企业微信 (WXWork) 数据库解密模块
|
|
基于 wxSQLite3 AES-128-CBC 每页加密算法
|
|
|
|
加密方案:
|
|
- 页面大小: 4096 字节
|
|
- 算法: AES-128-CBC (每页独立)
|
|
- 密钥派生: AES key = MD5(raw_key + page_no(LE uint32) + b"sAlT")
|
|
- IV 生成: 基于页号的线性同余生成器 (sqlite3mcGenerateInitialVector)
|
|
- 第 1 页特殊处理: 字节 16-23 明文存储 (页面大小等头字段)
|
|
"""
|
|
|
|
import hashlib
|
|
import os
|
|
import sqlite3
|
|
import struct
|
|
|
|
from Crypto.Cipher import AES
|
|
|
|
PAGE_SZ = 4096
|
|
SQLITE_HDR = b"SQLite format 3\x00"
|
|
WXSQLITE3_SALT = b"sAlT"
|
|
|
|
|
|
def _modmult(a, b, c, m, s):
|
|
"""线性同余乘法, 模拟 sqlite3mc 的伪随机生成器"""
|
|
q = s // a
|
|
s = b * (s - a * q) - c * q
|
|
if s < 0:
|
|
s += m
|
|
return s
|
|
|
|
|
|
def generate_initial_vector(page_no):
|
|
"""生成每页 IV (匹配 SQLite3MultipleCiphers 的 sqlite3mcGenerateInitialVector)"""
|
|
z = page_no + 1
|
|
initkey = bytearray(16)
|
|
for idx in range(4):
|
|
z = _modmult(52774, 40692, 3791, 2147483399, z)
|
|
initkey[idx * 4: idx * 4 + 4] = struct.pack("<I", z & 0xFFFFFFFF)
|
|
return hashlib.md5(initkey).digest()
|
|
|
|
|
|
def derive_wxsqlite3_aes128_page_key(raw_key, page_no):
|
|
"""派生每页 AES-128 密钥"""
|
|
if len(raw_key) != 16:
|
|
raise ValueError("wxSQLite3 AES-128 raw key 必须是 16 字节")
|
|
material = raw_key + struct.pack("<I", page_no) + WXSQLITE3_SALT
|
|
return hashlib.md5(material).digest()
|
|
|
|
|
|
def is_plain_sqlite_page(page):
|
|
"""检查是否为未加密的 SQLite 页面"""
|
|
return page[:len(SQLITE_HDR)] == SQLITE_HDR
|
|
|
|
|
|
def has_wxsqlite3_plain_header_fragment(page):
|
|
"""检查是否为新版 wxSQLite3 AES 模式 (第 1 页字节 16-23 为明文)"""
|
|
if len(page) < 24:
|
|
return False
|
|
header = page[16:24]
|
|
page_size = (header[0] << 8) | header[1]
|
|
if page_size == 1:
|
|
page_size = 65536
|
|
return (
|
|
page_size >= 512
|
|
and page_size <= 65536
|
|
and (page_size & (page_size - 1)) == 0
|
|
and header[5] == 0x40
|
|
and header[6] == 0x20
|
|
and header[7] == 0x20
|
|
)
|
|
|
|
|
|
def is_wxsqlite3_aes128_page1(page):
|
|
"""判断是否为加密的 wxSQLite3 AES-128 数据库第 1 页"""
|
|
return not is_plain_sqlite_page(page) and has_wxsqlite3_plain_header_fragment(page)
|
|
|
|
|
|
def _decrypt_aes128_cbc(raw_key, page_no, data):
|
|
"""AES-128-CBC 解密一页"""
|
|
page_key = derive_wxsqlite3_aes128_page_key(raw_key, page_no)
|
|
iv = generate_initial_vector(page_no)
|
|
return AES.new(page_key, AES.MODE_CBC, iv).decrypt(data)
|
|
|
|
|
|
def decrypt_wxsqlite3_aes128_page(raw_key, page_data, page_no):
|
|
"""解密一页 wxSQLite3 AES-128-CBC 页面为普通 SQLite 页面"""
|
|
if len(page_data) != PAGE_SZ:
|
|
raise ValueError(f"页面必须正好 {PAGE_SZ} 字节")
|
|
|
|
data = bytearray(page_data)
|
|
|
|
# 第 1 页特殊处理: 字节 16-23 明文, 需要先搬移再解密
|
|
if page_no == 1 and has_wxsqlite3_plain_header_fragment(data):
|
|
db_header_fragment = bytes(data[16:24])
|
|
data[16:24] = data[8:16] # 把加密的 8-16 移到 16-24 位置
|
|
decrypted_tail = _decrypt_aes128_cbc(raw_key, page_no, bytes(data[16:]))
|
|
data[16:] = decrypted_tail
|
|
if bytes(data[16:24]) != db_header_fragment:
|
|
raise ValueError("wxSQLite3 AES-128 密钥验证失败 (header fragment 不匹配)")
|
|
data[:16] = SQLITE_HDR
|
|
return bytes(data)
|
|
|
|
# 其他页: 整页 AES-128-CBC 解密
|
|
return _decrypt_aes128_cbc(raw_key, page_no, bytes(data))
|
|
|
|
|
|
def looks_like_sqlite_page1(page):
|
|
"""验证解密后的页面是否像合法的 SQLite 第 1 页"""
|
|
if page[:len(SQLITE_HDR)] != SQLITE_HDR:
|
|
return False
|
|
if len(page) < 108:
|
|
return False
|
|
btree_page_type = page[100]
|
|
return btree_page_type in (0x02, 0x05, 0x0A, 0x0D)
|
|
|
|
|
|
def verify_wxsqlite3_aes128_key(raw_key, page1):
|
|
"""验证 raw key 是否正确 (通过解密第 1 页检查)"""
|
|
if len(raw_key) != 16 or len(page1) < PAGE_SZ:
|
|
return False
|
|
try:
|
|
decrypted = decrypt_wxsqlite3_aes128_page(raw_key, page1[:PAGE_SZ], 1)
|
|
except (ValueError, KeyError):
|
|
return False
|
|
return looks_like_sqlite_page1(decrypted)
|
|
|
|
|
|
def decrypt_wxwork_database(db_path, out_path, raw_key):
|
|
"""解密整个数据库文件"""
|
|
size = os.path.getsize(db_path)
|
|
total_pages = (size + PAGE_SZ - 1) // PAGE_SZ
|
|
os.makedirs(os.path.dirname(out_path), exist_ok=True)
|
|
|
|
with open(db_path, "rb") as fin, open(out_path, "wb") as fout:
|
|
for page_no in range(1, total_pages + 1):
|
|
page = fin.read(PAGE_SZ)
|
|
if not page:
|
|
break
|
|
if len(page) < PAGE_SZ:
|
|
page += b"\x00" * (PAGE_SZ - len(page))
|
|
fout.write(decrypt_wxsqlite3_aes128_page(raw_key, page, page_no))
|
|
|
|
|
|
def verify_sqlite_file(path):
|
|
"""验证解密后的文件是否为合法的 SQLite 数据库"""
|
|
conn = sqlite3.connect(path)
|
|
try:
|
|
return [row[0] for row in conn.execute(
|
|
"SELECT name FROM sqlite_master WHERE type='table' ORDER BY name"
|
|
).fetchall()]
|
|
finally:
|
|
conn.close()
|
|
|
|
|
|
if __name__ == "__main__":
|
|
# 简单自测: 验证算法实现与已知参考值一致
|
|
test_key = bytes.fromhex("00112233445566778899aabbccddeeff")
|
|
iv = generate_initial_vector(1)
|
|
print(f"IV for page 1: {iv.hex()}")
|
|
page_key = derive_wxsqlite3_aes128_page_key(test_key, 1)
|
|
print(f"AES key for page 1: {page_key.hex()}")
|