110 lines
5.5 KiB
Python
110 lines
5.5 KiB
Python
"""Export verified, decrypted game master databases; never exports keys or account data.
|
|
|
|
Usage: python scripts/import-game-db.py <recovery-directory>
|
|
Requires protobuf (use PYTHONPATH for an isolated installation).
|
|
The input contains decrypted-master/, descriptors.json and decrypt-manifest.json.
|
|
"""
|
|
import base64
|
|
import gzip
|
|
import hashlib
|
|
import json
|
|
from pathlib import Path
|
|
import sqlite3
|
|
import sys
|
|
from google.protobuf import descriptor_pool, message_factory, json_format
|
|
|
|
root = Path(__file__).resolve().parents[1]
|
|
recovery = Path(sys.argv[1]).resolve()
|
|
manifest = json.loads((recovery / 'decrypt-manifest.json').read_text(encoding='utf-8'))
|
|
version = manifest['masterVersion']
|
|
if not version.isalnum() or len(version) != 64:
|
|
raise ValueError('Invalid master version')
|
|
pool = descriptor_pool.DescriptorPool()
|
|
descriptors = (recovery / 'descriptors.json').read_bytes()
|
|
pending = {n: base64.b64decode(v) for n, v in json.loads(descriptors).items()}
|
|
while pending:
|
|
before = len(pending)
|
|
for name, raw in list(pending.items()):
|
|
try:
|
|
pool.AddSerializedFile(raw)
|
|
except Exception:
|
|
continue
|
|
del pending[name]
|
|
if len(pending) == before:
|
|
raise ValueError('Unresolved protobuf dependencies')
|
|
|
|
def digest(value):
|
|
return hashlib.sha256(value).hexdigest()
|
|
|
|
def collector_enums(value, descriptor):
|
|
"""Keep the existing collector's qualified enum spelling, without legacy input."""
|
|
fields = {f.json_name: f for f in descriptor.fields}
|
|
result = {}
|
|
for name, item in value.items():
|
|
field = fields[name]
|
|
def convert(v):
|
|
if field.enum_type and isinstance(v, str):
|
|
return field.enum_type.name + '_' + v
|
|
return collector_enums(v, field.message_type) if field.message_type else v
|
|
result[name] = [convert(v) for v in item] if field.is_repeated else convert(item)
|
|
return result
|
|
|
|
tables, outputs = {}, {}
|
|
for entry in sorted(manifest['files'], key=lambda e: e['name']):
|
|
name = entry['name']
|
|
if not name.replace('_', '').isalnum() or name in tables:
|
|
raise ValueError('Invalid or duplicate table name')
|
|
file = recovery / 'decrypted-master' / (name + '.sqlite')
|
|
if digest(file.read_bytes()) != entry['decryptedSha256']:
|
|
raise ValueError('SQLite hash mismatch: ' + name)
|
|
if entry['verifiedPages'] * manifest['pageSize'] != file.stat().st_size:
|
|
raise ValueError('Incomplete page verification: ' + name)
|
|
with sqlite3.connect(file.as_uri() + '?mode=ro', uri=True) as con:
|
|
con.row_factory = sqlite3.Row
|
|
if [r[0] for r in con.execute('PRAGMA integrity_check')] != ['ok']:
|
|
raise ValueError('SQLite integrity failure: ' + name)
|
|
sqltables = [r[0] for r in con.execute("SELECT name FROM sqlite_master WHERE type='table' AND name NOT LIKE 'sqlite_%'")]
|
|
if len(sqltables) != 1:
|
|
raise ValueError('Expected one master table: ' + name)
|
|
table = sqltables[0]
|
|
desc = pool.FindMessageTypeByName('entity.master.' + table)
|
|
cls = message_factory.GetMessageClass(desc)
|
|
rows = []
|
|
for row in con.execute('SELECT * FROM "' + table.replace('"', '""') + '"'):
|
|
message = cls.FromString(row['data'])
|
|
original = message.SerializeToString(deterministic=True)
|
|
message.DiscardUnknownFields()
|
|
if message.SerializeToString(deterministic=True) != original:
|
|
raise ValueError('Unknown protobuf fields: ' + name)
|
|
keys = {k: row[k] for k in row.keys() if k != 'data'}
|
|
# SQL index values can be int64; preserve exact integers for JavaScript.
|
|
keys = {k: str(v) if isinstance(v, int) and abs(v) > 2**53 - 1 else v for k, v in keys.items()}
|
|
rows.append({**keys, 'data': collector_enums(json_format.MessageToDict(message), desc)})
|
|
raw = (json.dumps(rows, ensure_ascii=False, separators=(',', ':')) + '\n').encode()
|
|
packed = gzip.compress(raw, mtime=0)
|
|
tables[name] = {'file': name + '.json.gz', 'rows': len(rows), 'sha256': digest(raw),
|
|
'gzipSha256': digest(packed), 'sqliteSha256': entry['decryptedSha256'],
|
|
'encryptedSha256': entry['encryptedSha256']}
|
|
outputs[name + '.json.gz'] = packed
|
|
|
|
snapshot = {'schemaVersion': 1, 'kind': 'android-master-db', 'masterDataVersion': version,
|
|
'package': 'game.qualiarts.hololive.dreams.com', 'locale': 'kor',
|
|
'capturedAt': manifest['generatedAt'], 'descriptorSha256': digest(descriptors),
|
|
'tables': tables}
|
|
raw_manifest = (json.dumps(snapshot, ensure_ascii=False, indent=2) + '\n').encode()
|
|
destination = root / 'data/game-db/snapshots' / version
|
|
# Validate the complete snapshot before writing, and keep versions immutable.
|
|
if destination.exists():
|
|
if (destination / 'manifest.json').read_bytes() != raw_manifest or any(
|
|
(destination / file).read_bytes() != data for file, data in outputs.items()):
|
|
raise ValueError('A different snapshot already uses this master version')
|
|
else:
|
|
destination.mkdir(parents=True)
|
|
for file, data in outputs.items():
|
|
(destination / file).write_bytes(data)
|
|
(destination / 'manifest.json').write_bytes(raw_manifest)
|
|
pointer = {'schemaVersion': 1, 'snapshot': 'snapshots/' + version,
|
|
'manifestSha256': digest(raw_manifest)}
|
|
(root / 'data/game-db/current.json').write_text(json.dumps(pointer, indent=2) + '\n', encoding='utf-8')
|
|
print(f'Imported {len(tables)} tables / {sum(t["rows"] for t in tables.values())} rows from {version}')
|