""" AutoPOE2 - Offset Registry Builder (TASK-RE-PIPELINE-2026 / Phase C4) Chuyển kết quả trích xuất PyGhidra (data/re_out//extraction.json) thành Offset Registry ở HAI định dạng: 1. bin/Release/offsets_.toml - build-keyed, C++ OffsetRegistry nạp trực tiếp (không cần SQLite dependency trong C++ Core). 2. data/poe2_offsets.db (SQLite WAL) - SSoT tầng Python với provenance đầy đủ. INVARIANT được thực thi: INV-BUILD-01 : tệp TOML gắn khóa build_time_date_stamp + build_section_layout_hash; C++ sẽ từ chối dùng nếu client đang chạy không khớp fingerprint. INV-SIG-UNIQUE : CHỈ signature có match_count == 1 được phép vào registry. INV-OFFSET-01 : không có bất kỳ địa chỉ heap tuyệt đối nào (registry chỉ chứa RVA/offset tĩnh suy từ binary trên đĩa). Rule 14 : mọi record kèm provenance (build_id, sha256, tool, thời điểm). Ngữ nghĩa cuối cùng (ô nào là InGameState / LifeComponent ...) là bước HUMAN-IN-THE-LOOP: tool đánh dấu target='unreviewed' cho tới khi được xác nhận bằng dữ liệu live (captures/*_RAM.json, memprobe_report.txt). Cách dùng: python tools/re_pipeline/build_registry.py --extraction data/re_out/6A9E477A/extraction.json """ from __future__ import annotations import argparse import json import sqlite3 import sys from pathlib import Path from typing import List if hasattr(sys.stdout, "reconfigure"): sys.stdout.reconfigure(encoding="utf-8") REPO_ROOT = Path(__file__).resolve().parent.parent.parent DEFAULT_EXTRACTION = REPO_ROOT / "data" / "re_out" / "6A9E477A" / "extraction.json" DEFAULT_DB = REPO_ROOT / "data" / "poe2_offsets.db" SQL_SCHEMA = """ CREATE TABLE IF NOT EXISTS builds ( build_id TEXT PRIMARY KEY, time_date_stamp TEXT NOT NULL, size_of_image TEXT NOT NULL, file_size INTEGER NOT NULL, file_sha256 TEXT NOT NULL, section_layout_hash TEXT NOT NULL, binary_path TEXT NOT NULL, first_seen TEXT NOT NULL, last_verified TEXT ); CREATE TABLE IF NOT EXISTS signatures ( id INTEGER PRIMARY KEY AUTOINCREMENT, build_id TEXT NOT NULL REFERENCES builds(build_id), name TEXT NOT NULL, pattern TEXT NOT NULL, rip_offset INTEGER NOT NULL, match_count INTEGER NOT NULL, unique_ok INTEGER NOT NULL, pattern_bytes INTEGER, target_address TEXT NOT NULL, target_block TEXT NOT NULL, fan_in INTEGER, instruction_addr TEXT, mnemonic TEXT, cross_build_ok INTEGER DEFAULT 0, provenance TEXT NOT NULL ); CREATE TABLE IF NOT EXISTS offsets ( id INTEGER PRIMARY KEY AUTOINCREMENT, build_id TEXT NOT NULL REFERENCES builds(build_id), kind TEXT NOT NULL, target TEXT NOT NULL, payload TEXT NOT NULL, validator_id TEXT, source TEXT NOT NULL, discovered_at TEXT NOT NULL, last_verified TEXT, hit_count INTEGER DEFAULT 0, miss_count INTEGER DEFAULT 0 ); CREATE INDEX IF NOT EXISTS idx_offsets_build ON offsets(build_id, kind); CREATE INDEX IF NOT EXISTS idx_signatures_build ON signatures(build_id, unique_ok); """ # Chỉ các block nằm trong image PE mới được đưa vào registry C++. # Ghidra còn tạo block ngoài image (tdb / TEB / EXTERNAL) — fan-in cao nhưng # không phải static root của PathOfExile.exe. Bằng chứng extract 10/09/2026: # slot tdb ff00000058 (GS:[...] TEB) unique=True nhưng RVA âm/ngoài SizeOfImage. IN_IMAGE_DATA_BLOCKS = (".data", ".rdata", "_RDATA") def rva_from_address(address_str: str, image_base: int) -> int: """Chuyển địa chỉ ảnh (VA) Ghidra trả về thành RVA.""" return int(address_str, 16) - image_base def is_in_image_data_slot( record: dict, image_base: int, size_of_image: int ) -> bool: """INV-BUILD-01: registry chỉ chứa slot .data/.rdata/_RDATA nằm trong image.""" block = str(record.get("block") or "") if block not in IN_IMAGE_DATA_BLOCKS: return False try: rva = rva_from_address(str(record["address"]), image_base) except (KeyError, TypeError, ValueError): return False return 0 <= rva < size_of_image def build_registry(extraction_path: Path, db_path: Path, toml_path: Path, min_fan_in: int = 0) -> int: payload = json.loads(extraction_path.read_text(encoding="utf-8")) if payload.get("schema") != "autopoe2.re_extraction/v1": print(f"[ERROR] Schema không hỗ trợ: {payload.get('schema')}", file=sys.stderr) return 1 provenance = payload["provenance"] result = payload["result"] build_id = provenance["build_id"] image_base = int(str(result["image_base"]), 16) generated_at = payload["generated_at"] tool = payload["tool"] size_of_image = int(str(provenance["size_of_image"]), 16) # INV-SIG-UNIQUE: chỉ nhận signature khớp đúng 1 lần trong .text # INV-BUILD-01: loại slot ngoài image (tdb/TEB) — không phải RVA client. candidates: List[dict] = [ c for c in result["candidates"] if c.get("unique") and c.get("match_count") == 1 and c.get("signature") and int(c.get("fan_in", 0)) >= min_fan_in and is_in_image_data_slot(c, image_base, size_of_image) ] if not candidates: print("[WARN] Không có signature UNIQUE nào - registry rỗng phần tri thức.") print(" Hãy tăng --max-candidates/--pattern-window khi chạy re_extract.py.") # ---------------- SQLite (SSoT tầng Python) ---------------- db_path.parent.mkdir(parents=True, exist_ok=True) conn = sqlite3.connect(str(db_path)) try: conn.executescript(SQL_SCHEMA) conn.execute("PRAGMA journal_mode=WAL;") conn.execute( "INSERT OR REPLACE INTO builds VALUES (?,?,?,?,?,?,?,?,?)", ( build_id, provenance["time_date_stamp"], provenance["size_of_image"], provenance["file_size"], provenance["file_sha256"], provenance["section_layout_hash"], provenance["binary_path"], generated_at, generated_at, ), ) conn.execute("DELETE FROM signatures WHERE build_id = ?", (build_id,)) conn.execute("DELETE FROM offsets WHERE build_id = ?", (build_id,)) for index, record in enumerate(candidates, start=1): conn.execute( "INSERT INTO signatures (build_id,name,pattern,rip_offset,match_count," "unique_ok,pattern_bytes,target_address,target_block,fan_in," "instruction_addr,mnemonic,cross_build_ok,provenance) " "VALUES (?,?,?,?,?,?,?,?,?,?,?,?,?,?)", ( build_id, f"sig_{build_id}_{index:03d}", record["signature"], record.get("rip_offset", 0), record["match_count"], 1, record.get("pattern_bytes"), record["address"], record["block"], record.get("fan_in"), record.get("instruction_address"), record.get("mnemonic"), 0, # cross_build_ok: cần xác chứng trên >=2 build trước khi bật json.dumps( {"tool": tool, "generated_at": generated_at, "extraction": str(extraction_path)}, ensure_ascii=False, ), ), ) conn.execute( "INSERT INTO offsets (build_id,kind,target,payload,validator_id,source," "discovered_at,last_verified,hit_count,miss_count) " "VALUES (?,?,?,?,?,?,?,?,?,?)", ( build_id, "aob_pattern", "unreviewed", # HUMAN-IN-THE-LOOP: chưa xác định ngữ nghĩa json.dumps( {"pattern": record["signature"], "rip_offset": record.get("rip_offset", 0), "target_rva": f"0x{rva_from_address(record['address'], image_base):X}"}, ensure_ascii=False, ), "ValidateInGameState", "static_re", generated_at, None, 0, 0, ), ) conn.commit() finally: conn.close() # ---------------- TOML build-keyed (cho C++ OffsetRegistry) ---------------- lines = [ "# AutoPOE2 Offset Registry - INV-BUILD-01 build-keyed knowledge", f"# build_id = {build_id}", f"# binary_sha256 = {provenance['file_sha256']}", f"# section_layout_hash = {provenance['section_layout_hash']}", f"# generated_at = {generated_at}", f"# tool = {tool}", "# INV-SIG-UNIQUE: moi pattern duoi day chi khop DUNG 1 lan trong .text", "# INV-OFFSET-01: khong co dia chi heap tuyet doi nao o day", "", f"build_time_date_stamp = {provenance['time_date_stamp']}", f"build_section_layout_hash = {provenance['section_layout_hash']}", "", ] static_roots: List[str] = [] for record in candidates: if record["block"] == ".data": static_roots.append(record["address"]) for index, address_str in enumerate(static_roots[:32], start=1): lines.append(f"static_root_{index} = 0x{rva_from_address(address_str, image_base):X}") if static_roots: lines.append("") for index, record in enumerate(candidates[:32], start=1): lines.append(f"aob_pattern_{index} = {record['signature']}") lines.append(f"aob_rip_offset_{index} = {record.get('rip_offset', 0)}") if candidates: lines.append("") toml_path.parent.mkdir(parents=True, exist_ok=True) toml_path.write_text("\n".join(lines) + "\n", encoding="utf-8") print(f"[OK] SQLite registry : {db_path}") print(f"[OK] TOML registry : {toml_path}") print(f" build_id={build_id} unique_signatures={len(candidates)} " f"static_roots={len(static_roots)}") print(" [HUMAN-IN-THE-LOOP] Ngu nghia cac slot dang la 'unreviewed' -") print(" can doi chuan live (memprobe_report.txt, captures/*_RAM.json)") print(" truoc khi gan ten InGameState/LifeComponent/... vao cot target.") return 0 def main() -> int: parser = argparse.ArgumentParser( description="AutoPOE2 Offset Registry Builder (Phase C4)" ) parser.add_argument("--extraction", default=str(DEFAULT_EXTRACTION)) parser.add_argument("--db", default=str(DEFAULT_DB)) parser.add_argument( "--toml", default=None, help="Tệp TOML xuất ra (mặc định: bin/Release/offsets_.toml)", ) parser.add_argument( "--min-fan-in", type=int, default=0, help="Lọc signature theo fan-in tối thiểu (0 = giữ tất cả unique)", ) args = parser.parse_args() extraction_path = Path(args.extraction) if not extraction_path.is_file(): print(f"[ERROR] Không thấy tệp extraction: {extraction_path}", file=sys.stderr) print(" Hãy chạy: python tools/re_pipeline/re_extract.py trước.", file=sys.stderr) return 2 payload = json.loads(extraction_path.read_text(encoding="utf-8")) build_id = payload["provenance"]["build_id"] toml_path = Path(args.toml) if args.toml else ( REPO_ROOT / "bin" / "Release" / f"offsets_{build_id}.toml" ) return build_registry(extraction_path, Path(args.db), toml_path, args.min_fan_in) if __name__ == "__main__": raise SystemExit(main())