Add the import-layer repair and the Themida absorption

references/iat-virtualization-and-call-site-rewriting.md records the two edits one
protector family makes to a native image: the import slot stops holding an address and
the indirect call that read it becomes a relative call to a stub. It carries the pairing
rule -- a site's target equals a slot's value -- the three placements of the filler byte
in a six-byte rewrite and the refusal when there is none, the divergence between builds
that modify the code section and builds that only redirect the table, the small set of
imports redirected twice, the choice between repairing, extracting a payload, observing
and reporting, and the limits an automatic unpacker inherits from its own dump.

scripts/iat_callsite_restore.py implements the mechanical half offline: inventory the
slots from three sources, pair by value, decide the six-byte window per site, refuse an
ambiguous slot and a site whose sixth byte is the next instruction, rewrite, then
re-decode every written site from the file it wrote. It is report-only unless --out is
given, and it does not unpack: a virtualised entry has no site list to restore.

Registered in both symptom tables, the routing, reference and script indexes, the
capability registry as call_site_restore, and the coverage, route-inventory and evidence
tables. dynamic-debugging.md gains the instrument rule the second write-up demonstrates
(a target that runs at full speed and dies when stopped has a perturbation detector),
memory-and-dumps.md gains the call-site step of a rebuild, environment.md gains the
compatibility-shim trap, and protectors-and-packers.md gains one recognition row plus a
pointer rather than the detail, because that file makes no product-level claims.

The target-behaviour claims stay unverified: no protected sample was analysed here. What
is observed is the tool's behaviour on a generated image -- both filler positions, both
refusals, both displacement forms -- and what the source material states about itself.
FINDINGS F16 records the defect this pass produced, a slot's magnitude read as its
meaning, with the test that fails without the fix. .gitignore gains ref/, which was not
covered by the existing rule because that rule carries a leading dot.

tests/README.md

@@ -26,6 +26,8 @@ and "the script was measured doing X on this host".
 | `cli/test_cli_coverage.py` | Coverage derived from the scripts directory rather than a hand-written list: every shipped `.py` is either exercised or declared uncovered with a reason, and every one answers `--help` with its usage block. A new script fails here until it is registered. |
 | `cli/test_cli_json.py` | Each script's machine-readable output: keys, values, and the conventions that are not what a reader would assume -- including `pe_triage.py`'s identity block on a generated image (and that it writes nothing) and `pe_diff.py`'s self-comparison, which is the control for every other diff. |
 | `cli/test_cli_registry_surface.py` | The registry and doctor subprocess contracts: `--list` / `--json` / single-capability exit codes and `RESULT=` tokens, the doctor's three blocks and its 0-versus-3 verdict, and the view flags that print one block only. |
+| `unit/test_iat_callsite_restore.py` | The import-layer repair's four decisions on a generated image: the six-byte window on either side of a call, the refusal when the sixth byte is the next instruction, the refusal when two slots hold one address, the pairing rule, and the verdict token that follows from the evidence rather than from a count. |
+| `cli/test_cli_iat_callsite.py` | The repair's published contract as a subprocess: a report-only run writes nothing into the working directory, `--out` leaves the input byte-identical and the written sites re-decode to indirect calls, the PE32 displacement is an absolute slot address where the PE32+ one is relative, and a refusal is exit 1 rather than a quiet success. |
 | `cli/test_cli_dart_snapshot.py` | The snapshot reader's CLI: the refusals and their exit codes, `--find` case-insensitivity and the exit code a miss produces, `--at` by file offset and by virtual address, and the `--json` document. |
 | `integration/test_pipeline_end_to_end.py` | Build, locate, patch, re-read: the patched bytes, the untouched control site, the recomputed checksum, and a control run in which nothing changes. |
 | `integration/test_across_builds.py` | Pattern hit counts compared between two builds, including the case where a patch removes the very byte the pattern was located with. |

tests/cli/test_cli_iat_callsite.py

@@ -0,0 +1,124 @@
+"""The import-layer repair's published contract, exercised as a subprocess.
+
+What a caller can rely on, and what this file pins:
+
+  * without --out, nothing is written and the run is an analysis (exit 0 or 1 depending on whether a
+    site was refused, never a silent file);
+  * with --out, the input file is byte-identical afterwards, the output is a different file, and the
+    six bytes written at each site decode to an indirect call through the slot the report names;
+  * a refusal is exit 1 and the word "incomplete" -- the failure mode this repository is organised
+    against is a partial repair reported as a finished one.
+"""
+import hashlib
+import json
+
+import pytest
+
+pytestmark = pytest.mark.cli
+
+
+def sha(path):
+    return hashlib.sha256(path.read_bytes()).hexdigest()
+
+
+def document(text):
+    return json.JSONDecoder().raw_decode(text)[0]
+
+
+def test_analysis_writes_nothing_and_reports_incomplete(run_script, iat_scenario, tmp_path):
+    scen = iat_scenario("pe64")
+    before = sha(scen.path)
+    result = run_script("iat_callsite_restore.py", scen.path, "--iat", "0x3020,0x18", cwd=tmp_path)
+    assert result.returncode == 1, result.stderr
+    assert "RESULT=incomplete" in result.stdout
+    assert sha(scen.path) == before
+    assert list(tmp_path.iterdir()) == []
+
+
+def test_the_rewrite_is_verified_and_leaves_the_input_alone(run_script, iat_scenario,
+                                                            load_script, tmp_path):
+    scen = iat_scenario("pe64")
+    before = sha(scen.path)
+    out = tmp_path / "fixed.exe"
+    result = run_script("iat_callsite_restore.py", scen.path, "--iat", "0x3020,0x18",
+                        "--out", out)
+    assert result.returncode == 1, result.stderr
+    assert "VERIFY FAILED" not in result.stdout
+    assert "RESULT=incomplete" in result.stdout
+    assert sha(scen.path) == before, "the input was modified"
+
+    pe = load_script("peutil.py").PE(out.read_bytes())
+    for window in (scen.sites["filler_after"], scen.sites["filler_before"] - 1):
+        raw = pe.rva_bytes(window, 6)
+        assert raw[:2] == b"\xFF\x15", raw.hex()
+    # The refused site is untouched: the byte after its call is still the original instruction.
+    assert pe.rva_bytes(scen.sites["no_filler"] + 5, 1) == b"\xC3"
+
+
+def test_the_pe32_displacement_is_an_absolute_slot_address(run_script, iat_scenario,
+                                                           load_script, tmp_path):
+    scen = iat_scenario("pe32")
+    out = tmp_path / "fixed32.exe"
+    result = run_script("iat_callsite_restore.py", scen.path, "--iat", "0x3020,0x18", "--out", out)
+    assert result.returncode == 1, result.stderr
+    pe = load_script("peutil.py").PE(out.read_bytes())
+    raw = pe.rva_bytes(scen.sites["filler_after"], 6)
+    assert raw[:2] == b"\xFF\x15"
+    assert int.from_bytes(raw[2:], "little") == scen.base + scen.slot_rva
+
+
+def test_a_clean_image_exits_zero(run_script, fixtures):
+    result = run_script("iat_callsite_restore.py", fixtures.pe64.path)
+    assert result.returncode == 0, result.stderr
+    assert "RESULT=no-pairs" in result.stdout
+
+
+def test_out_equal_to_the_input_is_a_usage_error(run_script, iat_scenario):
+    scen = iat_scenario("pe64")
+    result = run_script("iat_callsite_restore.py", scen.path, "--out", scen.path)
+    assert result.returncode == 2
+    assert "never written" in result.stderr
+
+
+def test_a_non_pe_is_exit_three(run_script, tmp_path):
+    junk = tmp_path / "notape.bin"
+    junk.write_bytes(b"MZ" + b"\x00" * 200)
+    result = run_script("iat_callsite_restore.py", junk)
+    assert result.returncode == 3
+    assert "error:" in result.stderr
+
+
+def test_json_carries_the_document_then_the_verdict_line(run_script, iat_scenario):
+    scen = iat_scenario("pe64")
+    result = run_script("iat_callsite_restore.py", scen.path, "--iat", "0x3020,0x18", "--json")
+    payload = document(result.stdout)
+    lines = [ln for ln in result.stdout.splitlines() if ln.strip()]
+    assert lines[-1].startswith("RESULT=")
+    assert payload["result"] == lines[-1].split("=", 1)[1]
+    assert payload["input"]["bitness"] == scen.bitness
+    assert payload["input"]["base_used"] == scen.base
+    assert sorted(payload["iat"]["classes"]) == ["above_image", "below_image", "in_image", "zero"]
+    assert [p["site_rva"] for p in payload["pairs"]] == [scen.sites["filler_after"],
+                                                         scen.sites["filler_before"]]
+    assert {r["reason"] for r in payload["refused"]} == {"no-filler", "ambiguous-slot"}
+    assert "output" not in payload, "a report-only run must not claim an output file"
+
+
+def test_full_out_writes_the_same_document_and_says_where(run_script, iat_scenario, tmp_path):
+    scen = iat_scenario("pe64")
+    record = tmp_path / "report.json"
+    result = run_script("iat_callsite_restore.py", scen.path, "--iat", "0x3020,0x18",
+                        "--full-out", record)
+    assert result.returncode == 1
+    assert "full record" in result.stdout
+    saved = json.loads(record.read_text(encoding="utf-8"))
+    assert saved["tool"] == "iat_callsite_restore.py"
+    assert saved["result"] == "incomplete"
+
+
+def test_max_sites_bounds_the_printed_lists(run_script, iat_scenario):
+    scen = iat_scenario("pe64")
+    result = run_script("iat_callsite_restore.py", scen.path, "--iat", "0x3020,0x18",
+                        "--max-sites", "0")
+    assert result.returncode == 1
+    assert "more" in result.stdout

tests/conftest.py

@@ -55,6 +55,10 @@ SCRIPT_FILES = (
     # script is neither in this tuple nor declared uncovered with a reason.
     "pe_triage.py",
     "pe_diff.py",
+    # The import-layer repair: it pairs a rewritten relative call with the slot that holds its
+    # target, and refuses the three cases a byte-swapping tool would corrupt. Its scenario is
+    # generated below, because the decisions it makes are about byte layout, not about any sample.
+    "iat_callsite_restore.py",
 )
 
 # A hang detector, not a performance assertion. Nothing in this suite is meant to be slow.
@@ -504,6 +508,89 @@ def fixtures(build_fixture):
     return FixtureSet({name: build_fixture(name) for name in FIXTURE_RECIPES})
 
 
+class IatScenario(object):
+    """A generated image carrying the six-byte decisions the repair script must make.
+
+    Five sites: one with the filler byte after the call, one with it before, one with no filler at
+    all, one whose target two slots both hold, and the slots themselves in a range the import
+    directory does not describe (which is the normal case for a captured image).
+    """
+
+    __slots__ = ("path", "pe_path", "base", "slot_rva", "dup_slot_rvas", "sites", "bitness")
+
+    def __init__(self, **kw):
+        for key in self.__slots__:
+            setattr(self, key, kw.get(key))
+
+    def __repr__(self):
+        return "IatScenario(%s)" % self.path
+
+
+@pytest.fixture(scope="session")
+def iat_scenario(tmp_path_factory, build_fixture, load_script):
+    """Build the scenario by editing a fixture the rest of the suite already uses.
+
+    Editing rather than synthesising is deliberate: the script's job is arithmetic over a real
+    header and a real section table, and a hand-built header would let a wrong RVA mapping agree
+    with itself. The fixture is copied first, so the shared build stays pristine for every other
+    test in the session.
+    """
+    peutil = load_script("peutil.py")
+    root = tmp_path_factory.mktemp("iat-scenario")
+    cache = {}
+
+    def _build(name="pe64"):
+        if name in cache:
+            return cache[name]
+        built = build_fixture(name)
+        pe = peutil.PE(built.data)
+        base = pe.image_base
+        psize = 8 if pe.is_pe32plus else 4
+        data = bytearray(built.data)
+
+        def put(rva, raw):
+            off = pe.rva_to_offset(rva)
+            assert off is not None, "scenario RVA %#x has no file offset" % rva
+            data[off:off + len(raw)] = raw
+
+        def call(rva, target_va):
+            return b"\xE8" + struct.pack("<i", target_va - (base + rva + 5))
+
+        target = base + 0x9000
+        target_dup = base + 0x9100
+        put(0x3020, target.to_bytes(psize, "little"))
+        put(0x3028, target_dup.to_bytes(psize, "little"))
+        put(0x3030, target_dup.to_bytes(psize, "little"))
+        # The generated .text is NOP-filled, so every context byte a case depends on is written
+        # explicitly. Without that, "NOP before" and "NOP after" hold everywhere and the three
+        # layout cases collapse into one.
+        put(0x11FF, b"\x00")
+        put(0x1200, call(0x1200, target) + b"\x90")
+        put(0x120F, b"\x90")
+        put(0x1211, call(0x1211, target) + b"\x00")
+        put(0x121F, b"\x00")
+        put(0x1220, call(0x1220, target) + b"\xC3")
+        put(0x122F, b"\x00")
+        put(0x1230, call(0x1230, target_dup) + b"\x90")
+
+        path = root / ("%s-iat.exe" % name)
+        path.write_bytes(bytes(data))
+        scenario = IatScenario(
+            path=path,
+            pe_path=built.path,
+            base=base,
+            slot_rva=0x3020,
+            dup_slot_rvas=(0x3028, 0x3030),
+            sites={"filler_after": 0x1200, "filler_before": 0x1211,
+                   "no_filler": 0x1220, "ambiguous": 0x1230},
+            bitness=64 if pe.is_pe32plus else 32,
+        )
+        cache[name] = scenario
+        return scenario
+
+    return _build
+
+
 @pytest.fixture(scope="session")
 def synthetic_elf(fixture_dir):
     """The generated ELF64, written once for the whole session."""

tests/unit/test_iat_callsite_restore.py

@@ -0,0 +1,166 @@
+"""The import-layer repair's decisions, exercised on a generated scenario.
+
+The script makes three decisions per call site and one per slot, and every one of them can be wrong
+in a way that still produces a loadable file:
+
+  * which six-byte window belongs to a site (the filler byte can be on either side),
+  * whether a site can be rewritten at all (a site with no filler byte is a structural edit),
+  * which slot a target belongs to (two slots can hold the same address).
+
+So the tests are about those decisions rather than about the report's wording: each case asserts the
+window or the refusal reason, and the ones covering the rewrite re-decode the bytes the script wrote
+instead of trusting the counters it printed.
+"""
+import argparse
+
+import pytest
+
+pytestmark = pytest.mark.unit
+
+
+def _args(image, **kw):
+    base = {"image": str(image), "out": None, "base": None, "iat": None, "max_sites": 40}
+    base.update(kw)
+    return argparse.Namespace(**base)
+
+
+def test_parse_range_accepts_both_separators(load_script):
+    mod = load_script("iat_callsite_restore.py")
+    assert mod.parse_range("0x48D000,0x1000") == (0x48D000, 0x1000)
+    assert mod.parse_range("0x48D000:0x1000") == (0x48D000, 0x1000)
+    with pytest.raises(ValueError):
+        mod.parse_range("0x48D000")
+
+
+def test_encode_site_is_rip_relative_on_pe32plus(iat_scenario, load_script):
+    """PE32+ displacement is relative to the end of the instruction, not to the slot."""
+    mod = load_script("iat_callsite_restore.py")
+    scen = iat_scenario("pe64")
+    pe = load_script("peutil.py").PE(scen.path.read_bytes())
+    raw = mod.encode_site(pe, scen.base, 0x1200, "call", scen.slot_rva)
+    assert raw[:2] == b"\xFF\x15"
+    assert int.from_bytes(raw[2:], "little") == 0x3020 - 0x1200 - 6
+
+
+def test_encode_site_is_absolute_on_pe32(iat_scenario, load_script):
+    mod = load_script("iat_callsite_restore.py")
+    scen = iat_scenario("pe32")
+    pe = load_script("peutil.py").PE(scen.path.read_bytes())
+    raw = mod.encode_site(pe, scen.base, 0x1200, "jmp", scen.slot_rva)
+    assert raw[:2] == b"\xFF\x25"
+    assert int.from_bytes(raw[2:], "little") == scen.base + scen.slot_rva
+
+
+def test_plan_places_the_window_on_either_side_of_the_call(iat_scenario, load_script):
+    mod = load_script("iat_callsite_restore.py")
+    scen = iat_scenario("pe64")
+    blob = scen.path.read_bytes()
+    pe = load_script("peutil.py").PE(blob)
+    pairs = [
+        {"site_rva": scen.sites["filler_after"], "kind": "call", "target_va": 0,
+         "slot_rvas": [scen.slot_rva]},
+        {"site_rva": scen.sites["filler_before"], "kind": "call", "target_va": 0,
+         "slot_rvas": [scen.slot_rva]},
+    ]
+    planned, refused = mod.plan_windows(pe, pairs)
+    assert refused == []
+    windows = {item["site_rva"]: item["window_rva"] for item in planned}
+    assert windows[scen.sites["filler_after"]] == scen.sites["filler_after"]
+    assert windows[scen.sites["filler_before"]] == scen.sites["filler_before"] - 1
+
+
+def test_plan_refuses_a_site_whose_sixth_byte_is_an_instruction(iat_scenario, load_script):
+    """The byte after the call here is a return, so writing would consume the next instruction."""
+    mod = load_script("iat_callsite_restore.py")
+    scen = iat_scenario("pe64")
+    pe = load_script("peutil.py").PE(scen.path.read_bytes())
+    pairs = [{"site_rva": scen.sites["no_filler"], "kind": "call", "target_va": 0,
+              "slot_rvas": [scen.slot_rva]}]
+    planned, refused = mod.plan_windows(pe, pairs)
+    assert planned == []
+    assert [r["reason"] for r in refused] == ["no-filler"]
+    assert "next instruction" in refused[0]["detail"]
+
+
+def test_plan_refuses_a_target_two_slots_hold(iat_scenario, load_script):
+    mod = load_script("iat_callsite_restore.py")
+    scen = iat_scenario("pe64")
+    pe = load_script("peutil.py").PE(scen.path.read_bytes())
+    pairs = [{"site_rva": scen.sites["ambiguous"], "kind": "call",
+              "target_va": scen.base + 0x9100, "slot_rvas": list(scen.dup_slot_rvas)}]
+    planned, refused = mod.plan_windows(pe, pairs)
+    assert planned == []
+    assert [r["reason"] for r in refused] == ["ambiguous-slot"]
+    for rva in scen.dup_slot_rvas:
+        assert "%#x" % rva in refused[0]["detail"]
+
+
+def test_the_report_pairs_the_restorable_sites_and_names_the_others(iat_scenario, load_script):
+    mod = load_script("iat_callsite_restore.py")
+    scen = iat_scenario("pe64")
+    blob = scen.path.read_bytes()
+    pe = load_script("peutil.py").PE(blob)
+    args = _args(scen.path, iat=(0x3020, 0x18))
+    report = mod.build_report(args, blob, pe)
+
+    assert [p["site_rva"] for p in report["pairs"]] == [scen.sites["filler_after"],
+                                                        scen.sites["filler_before"]]
+    assert {r["reason"] for r in report["refused"]} == {"no-filler", "ambiguous-slot"}
+    assert "explicit-range" in report["iat"]["sources"]
+    assert report["input"]["base_used"] == scen.base
+
+
+def test_without_the_slot_range_the_report_says_needs_trace(iat_scenario, load_script):
+    """No slot describes these targets, and the calls leave the image: that is the handle case.
+
+    It is also the reason the verdict is not "clean". Nothing was found by pairing, and the correct
+    reading is that pairing cannot work on this image, not that the image is untouched.
+    """
+    mod = load_script("iat_callsite_restore.py")
+    scen = iat_scenario("pe64")
+    blob = scen.path.read_bytes()
+    pe = load_script("peutil.py").PE(blob)
+    report = mod.build_report(_args(scen.path), blob, pe)
+    assert report["pairs"] == []
+    assert len(report["unpaired"]) == 4
+    assert mod.decide_result(report, wrote=False) == "needs-trace"
+
+
+def test_a_wrong_base_produces_the_same_verdict_as_a_missing_table(iat_scenario, load_script):
+    """Two different causes, one report -- which is why the printed hint names the base first.
+
+    A site's target is computed in the address space the image ran at, so a base off by the
+    relocation delta turns every real pair into an unpaired call. The verdict cannot tell the two
+    apart, and a reader who assumes "the table holds handles" from this report alone would go and
+    write a trace they do not need.
+    """
+    mod = load_script("iat_callsite_restore.py")
+    scen = iat_scenario("pe64")
+    blob = scen.path.read_bytes()
+    pe = load_script("peutil.py").PE(blob)
+    report = mod.build_report(_args(scen.path, base=0x400000, iat=(0x3020, 0x18)), blob, pe)
+    assert report["pairs"] == []
+    assert len(report["unpaired"]) == 4
+    assert mod.decide_result(report, wrote=False) == "needs-trace"
+    assert report["input"]["base_source"] == "explicit --base"
+
+
+def test_a_clean_image_reports_no_pairs(load_script, fixtures):
+    mod = load_script("iat_callsite_restore.py")
+    blob = fixtures.pe64.data
+    pe = load_script("peutil.py").PE(blob)
+    report = mod.build_report(_args(fixtures.pe64.path), blob, pe)
+    assert report["pairs"] == []
+    assert report["unpaired"] == []
+    assert mod.decide_result(report, wrote=False) == "no-pairs"
+
+
+def test_result_tokens_follow_the_evidence(load_script):
+    mod = load_script("iat_callsite_restore.py")
+    base = {"pairs": [], "refused": [], "unpaired": [], "iat": {"classes": {}}}
+    assert mod.decide_result(dict(base), wrote=False) == "no-pairs"
+    assert mod.decide_result(dict(base, pairs=[1]), wrote=False) == "report-only"
+    assert mod.decide_result(dict(base, pairs=[1]), wrote=True) == "repaired"
+    assert mod.decide_result(dict(base, unpaired=[1]), wrote=False) == "needs-trace"
+    assert mod.decide_result(dict(base, refused=[1]), wrote=True) == "incomplete"
+    assert mod.decide_result(dict(base, verification_failures=[1]), wrote=True) == "incomplete"