XFESystem

#!/usr/bin/env python3
"""Verify pinned NVIDIA reference bytes and compile actual freestanding RM inputs.

No driver is installed, no GPU is accessed, and no unresolved symbol is filled
by a stub. A successful object compilation never claims GPU/CUDA execution.
"""
from __future__ import annotations

import argparse
import hashlib
import json
import pathlib
import re
import struct
import subprocess
import time
import traceback

import CaptureReferences as capture

TASK_ROOT = capture.TASK_ROOT
TOOL_ROOT = pathlib.Path(__file__).resolve().parent
LOCK_PATH = TOOL_ROOT / "RmSources.lock.json"
PROBE_HEADERS = {
    "src/common/inc/swref/published/blackwell/gb202/dev_therm.h": "39ab81dc68a69caebe9d220350fd3dad85daf174942717ba30979b73450d64b4",
    "src/common/inc/swref/published/blackwell/gb202/dev_therm_addendum.h": "6078e834e1b20470187fe542f18b7c8d6d4e5bc2d0e503ce10d1b51e87e6f1ae",
}


def file_hash(path: pathlib.Path) -> str:
    return capture.sha256(path.read_bytes())


def tool_sources() -> dict[str, str]:
    return {str(path.relative_to(TASK_ROOT)).replace("\\", "/"): file_hash(path)
            for path in sorted(TOOL_ROOT.iterdir()) if path.suffix in (".py", ".json", ".md")}


def verify_references(lock: dict) -> list[dict]:
    if lock.get("Schema") != "XfeNvidiaRmReferenceSources1" or lock.get("Commit") != capture.COMMIT or lock.get("Version") != capture.VERSION:
        raise ValueError("Unknown NVIDIA source identity")
    if lock.get("Repository") != capture.REPOSITORY or lock.get("MaximumCacheBytes") != capture.MAXIMUM_BYTES:
        raise ValueError("Source provenance or cache bound differs")
    if lock.get("Components") != capture.COMPONENTS or lock.get("IncludeDirectories") != capture.INCLUDE_DIRECTORIES:
        raise ValueError("Unreviewed component/include configuration")
    if lock.get("GuestCudaPassed") is not False or lock.get("GpuExecutionPassed") is not False:
        raise ValueError("Source references cannot certify hardware execution")
    rows = lock.get("Files")
    if not isinstance(rows, list) or not 1 <= len(rows) <= 4096:
        raise ValueError("Unbounded/empty source inventory")
    actual = sum(path.stat().st_size for path in capture.SOURCE_ROOT.rglob("*") if path.is_file())
    if actual > capture.MAXIMUM_BYTES:
        raise ValueError("Private reference cache exceeds 64 MiB")
    seen = set()
    for item in rows:
        relative = item["Path"]
        if relative in seen:
            raise ValueError("Duplicate source path")
        seen.add(relative)
        data = capture.source_path(relative).read_bytes()
        if len(data) != item["Bytes"] or len(data) > capture.MAXIMUM_FILE_BYTES or capture.sha256(data) != item["Sha256"] or capture.git_blob(data) != item["GitBlob"]:
            raise ValueError("Source/hash differs: " + relative)
        if not item.get("License"):
            raise ValueError("Missing per-file license provenance: " + relative)
        if item.get("Mode") not in ("100644", "100755"):
            raise ValueError("Unsupported Git file mode: " + relative)
    if "COPYING" not in seen or "version.mk" not in seen or "src/nvidia/Makefile" not in seen:
        raise ValueError("Build/version/license source is absent")
    version = capture.source_path("version.mk").read_text(encoding="utf-8")
    if not re.search(r"NVIDIA_VERSION\s*=\s*" + re.escape(capture.VERSION) + r"\s*$", version, re.MULTILINE):
        raise ValueError("Actual NVIDIA version.mk differs")
    copying = capture.source_path("COPYING").read_text(encoding="utf-8")
    if "Except where noted otherwise" not in copying or "licensed as MIT" not in copying:
        raise ValueError("Original COPYRIGHT/license is absent")
    return rows


def verify_probe_definitions(rows: list[dict]) -> dict:
    by_path = {row["Path"]: row for row in rows}
    for relative, expected in PROBE_HEADERS.items():
        if relative not in by_path or by_path[relative]["Sha256"] != expected or by_path[relative]["License"] != "MIT":
            raise ValueError("GB202 read-only probe header identity differs: " + relative)
    header, addendum = [capture.source_path(path).read_text(encoding="utf-8") for path in PROBE_HEADERS]
    def value(text: str, macro: str) -> int:
        match = re.search(r"^#define\s+" + re.escape(macro) + r"\s+(0x[0-9a-fA-F]+)\b", text, re.MULTILINE)
        if not match:
            raise ValueError("Pinned GB202 definition missing: " + macro)
        return int(match.group(1), 16)
    register = value(header, "NV_THERM_I2CS_SCRATCH")
    success = value(addendum, "NV_THERM_I2CS_SCRATCH_FSP_BOOT_COMPLETE_STATUS_SUCCESS")
    failed = value(addendum, "NV_THERM_I2CS_SCRATCH_FSP_BOOT_COMPLETE_STATUS_FAILED")
    if not re.search(r"^#define\s+NV_THERM_I2CS_SCRATCH_FSP_BOOT_COMPLETE\s+NV_THERM_I2CS_SCRATCH\s*$", addendum, re.MULTILINE):
        raise ValueError("Pinned GB202 FSP register alias differs")
    source = TASK_ROOT / "Source/XFESystem.NativeKernel/NvidiaBootProbe.cs"
    text = source.read_text(encoding="utf-8")
    match = re.search(r"\bconst\s+uint\s+FspBootComplete\s*=\s*(0x[0-9a-fA-F]+)\s*;", text)
    if not match or int(match.group(1), 16) != register:
        raise ValueError("Production read-only probe register differs from pinned NVIDIA header")
    return {"Headers": PROBE_HEADERS, "RegisterOffset": register, "StatusSuccess": success, "StatusFailed": failed,
            "ReadOnlyProbeSource": str(source.relative_to(TASK_ROOT)).replace("\\", "/"), "ReadOnlyProbeSha256": file_hash(source),
            "ProductionRegisterMatches": True, "DeviceReadPerformed": False, "GpuExecutionPassed": False}


def acquire_locked(lock: dict) -> None:
    """Fetch reviewed entries only; no include discovery or lock rewriting."""
    expected = sum(item["Bytes"] for item in lock["Files"] if not capture.source_path(item["Path"]).exists())
    actual = sum(path.stat().st_size for path in capture.SOURCE_ROOT.rglob("*") if path.is_file())
    if actual + expected > capture.MAXIMUM_BYTES:
        raise ValueError("Fetch would exceed private cache bound")
    for item in lock["Files"]:
        path = capture.source_path(item["Path"])
        if path.exists():
            continue
        data = capture.read_pinned_source(item["Path"], {"sha": item["GitBlob"], "size": item["Bytes"]})
        if len(data) != item["Bytes"] or capture.sha256(data) != item["Sha256"] or capture.git_blob(data) != item["GitBlob"]:
            raise ValueError("Downloaded reference differs: " + item["Path"])
        path.parent.mkdir(parents=True, exist_ok=True)
        path.write_bytes(data)


def invoke(command: list[str], log: pathlib.Path, timeout: int = 60) -> dict:
    started = time.monotonic()
    try:
        result = subprocess.run(command, cwd=TASK_ROOT, capture_output=True, timeout=timeout)
        content = result.stdout + result.stderr
        log.write_bytes(content)
        return {"Command": command, "ExitCode": result.returncode, "Milliseconds": round((time.monotonic() - started) * 1000, 3),
                "Log": str(log), "LogSha256": capture.sha256(content), "TimedOut": False}
    except subprocess.TimeoutExpired as error:
        content = (error.stdout or b"") + (error.stderr or b"")
        log.write_bytes(content)
        return {"Command": command, "ExitCode": None, "Milliseconds": round((time.monotonic() - started) * 1000, 3),
                "Log": str(log), "LogSha256": capture.sha256(content), "TimedOut": True}


def compiler_profile(clang: pathlib.Path) -> tuple[list[str], pathlib.Path]:
    resource = subprocess.run([str(clang), "--print-resource-dir"], capture_output=True, check=True, timeout=10).stdout.decode().strip()
    # Only Clang's architecture-independent built-ins; no Windows SDK/Linux
    # sysroot, system headers, libraries, synthetic RM definitions or OS stubs.
    flags = ["--target=x86_64-unknown-none-elf", "-std=gnu11", "-ffreestanding", "-fno-stack-protector", "-fno-common",
             "-fno-pic", "-mno-red-zone", "-mcmodel=kernel", "-msoft-float", "-mno-mmx", "-mno-sse", "-mno-sse2",
             "-ffunction-sections", "-fdata-sections", "-fno-strict-aliasing", "-fno-omit-frame-pointer", "-O2",
             "-DNV_X86_64", "-DNV_ARCH_BITS=64", "-Werror=implicit-function-declaration", "-nostdinc", "-isystem", str(pathlib.Path(resource) / "include"),
             "-include", str(capture.source_path("src/common/sdk/nvidia/inc/cpuopsys.h"))]
    makefile = capture.source_path("src/nvidia/Makefile").read_text(encoding="utf-8")
    for define in re.findall(r"^CFLAGS\s*\+=\s*(-D[A-Za-z0-9_=]+)\s*$", makefile, re.MULTILINE):
        flags.append(define)
    for directory in capture.INCLUDE_DIRECTORIES:
        flags += ["-I", str(capture.source_path(directory))]
    return flags, pathlib.Path(resource).resolve()


def compile_component(component: str, flags: list[str], clang: pathlib.Path, nm: pathlib.Path, output: pathlib.Path,
                      references: dict[str, dict], resource_root: pathlib.Path) -> dict:
    name = pathlib.PurePosixPath(component).stem
    object_path = output / (name + ".o")
    run = invoke([str(clang), *flags, "-H", "-c", str(capture.source_path(component)), "-o", str(object_path)], output / (name + ".Compile.log"))
    run.update({"Source": component, "Compiled": False, "Linked": False, "GpuExecutionPassed": False})
    diagnostics = pathlib.Path(run["Log"]).read_text(encoding="utf-8", errors="replace")
    run["MissingIncludes"] = re.findall(r"fatal error: '([^']+)' file not found", diagnostics)
    includes = {}
    for match in re.finditer(r"^\.+ (.+?)\s*$", diagnostics, re.MULTILINE):
        path = pathlib.Path(match.group(1)).resolve()
        if path.is_relative_to(capture.SOURCE_ROOT.resolve()):
            relative = str(path.relative_to(capture.SOURCE_ROOT.resolve())).replace("\\", "/")
            if relative not in references or file_hash(path) != references[relative]["Sha256"]:
                raise ValueError("Actual compiler include was not source-locked: " + relative)
            kind = "PinnedNvidiaSource"
        elif path.is_relative_to(resource_root):
            kind = "CompilerBuiltin"
        else:
            raise ValueError("Compiler read an unreviewed external header: " + str(path))
        includes[str(path)] = {"Sha256": file_hash(path), "Kind": kind}
    run["ActualIncludeInputs"] = includes
    if run["ExitCode"] != 0:
        return run
    data = object_path.read_bytes()
    if len(data) < 64 or data[:6] != b"\x7fELF\x02\x01" or struct.unpack_from("<H", data, 18)[0] != 62 or struct.unpack_from("<H", data, 16)[0] != 1:
        raise ValueError("Compiler output is not an actual relocatable ELF64 x86_64 object")
    symbol_log = output / (name + ".Undefined.log")
    symbols = invoke([str(nm), "--undefined-only", "--format=posix", str(object_path)], symbol_log, 10)
    if symbols["ExitCode"] != 0:
        raise ValueError("Cannot inventory actual unresolved symbols")
    defined_log = output / (name + ".Defined.log")
    defined = invoke([str(nm), "--defined-only", "--extern-only", "--format=posix", str(object_path)], defined_log, 10)
    if defined["ExitCode"] != 0:
        raise ValueError("Cannot inventory actual defined symbols")
    run.update({"Compiled": True, "Object": str(object_path), "ObjectBytes": len(data), "ObjectSha256": capture.sha256(data),
                "UndefinedSymbols": [line.split()[0] for line in symbol_log.read_text(encoding="utf-8").splitlines() if line.strip()],
                "DefinedSymbols": [line.split()[0] for line in defined_log.read_text(encoding="utf-8").splitlines() if line.strip()],
                "SymbolInventory": symbols, "DefinedSymbolInventory": defined})
    return run


def main() -> int:
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument("--fetch", action="store_true")
    parser.add_argument("--compile", action="store_true")
    parser.add_argument("--output")
    parser.add_argument("--clang", default=str(TASK_ROOT / ".tools/llvm/bin/clang.exe"))
    parser.add_argument("--nm", default=str(TASK_ROOT / ".tools/llvm/bin/llvm-nm.exe"))
    args = parser.parse_args()
    parent = TASK_ROOT / "build/Gpu/RmPort"
    output = pathlib.Path(args.output).resolve() if args.output else parent / ("Proof-" + str(time.time_ns()))
    if not output.resolve().is_relative_to(parent.resolve()) or output.exists():
        raise ValueError("Use a new bounded build/Gpu/RmPort proof directory")
    output.mkdir(parents=True)
    sources_before = tool_sources()
    receipt = {"Kind": "XfeNvidiaRmPortPreflight", "Commit": capture.COMMIT, "Version": capture.VERSION,
               "ReferenceValidationPassed": False, "AnyRealComponentCompiled": False, "CompleteRmCompiled": False,
               "GpuExecutionPassed": False, "GuestCudaPassed": False, "GuestAcceptancePassed": False,
               "LinuxKernelReplacement": False, "HostDriverChanges": False, "FakeSymbolsOrStubs": False,
               "SourceSha256": sources_before, "Components": [], "Output": str(output)}
    try:
        lock = json.loads(LOCK_PATH.read_text(encoding="utf-8"))
        if args.fetch:
            acquire_locked(lock)
        rows = verify_references(lock)
        receipt.update({"ReferenceValidationPassed": True, "ReferenceFiles": len(rows), "ReferenceBytes": sum(row["Bytes"] for row in rows),
                        "CopyingSha256": file_hash(capture.source_path("COPYING")), "BuildInventory": lock["BuildInventory"],
                        "CaptureComplete": lock.get("CaptureComplete", False), "CaptureFailure": lock.get("CaptureFailure"),
                        "Gb202ReadOnlyProbeDefinitions": verify_probe_definitions(rows),
                        "UnresolvedLiteralIncludes": lock["UnresolvedLiteralIncludes"]})
        if args.compile:
            clang, nm = pathlib.Path(args.clang).resolve(), pathlib.Path(args.nm).resolve()
            receipt["ToolSha256"] = {str(path): file_hash(path) for path in [clang, nm]}
            version = invoke([str(clang), "--version"], output / "ClangVersion.log", 10)
            receipt["CompilerVersion"] = version
            flags, resource_root = compiler_profile(clang)
            receipt["CompilerProfile"] = flags
            for profile in ("BareMetalNoOsSelection", "UpstreamUnixInterfaceNeedsPort"):
                destination = output / profile
                destination.mkdir()
                selected = flags if profile == "BareMetalNoOsSelection" else flags + ["-DNV_UNIX"]
                for component in capture.COMPONENTS:
                    result = compile_component(component, selected, clang, nm, destination,
                                               {row["Path"]: row for row in rows}, resource_root)
                    result["Profile"] = profile
                    result["NvLinuxSelected"] = False
                    receipt["Components"].append(result)
                    print(profile, component, "Compiled" if result["Compiled"] else "Blocked", result["MissingIncludes"], flush=True)
            receipt["AnyRealComponentCompiled"] = any(row["Compiled"] for row in receipt["Components"])
        interface = capture.source_path("kernel-open/common/inc/os-interface.h").read_text(encoding="utf-8")
        declarations = sorted(set(re.findall(r"\b(os_[a-zA-Z0-9_]+)\s*\(", interface)))
        receipt["OsInterfaceInventory"] = {"Method": "Lexical declaration inventory, not complete linked import requirements",
                                           "Count": len(declarations), "Names": declarations}
        for component in receipt["Components"]:
            for header, identity in component["ActualIncludeInputs"].items():
                if file_hash(pathlib.Path(header)) != identity["Sha256"]:
                    raise ValueError("Compiler input changed while checking: " + header)
        verify_references(lock)
        receipt["ReferenceInputsUnchanged"] = True
        for tool, expected in receipt.get("ToolSha256", {}).items():
            if file_hash(pathlib.Path(tool)) != expected:
                raise ValueError("Compiler tool changed while checking: " + tool)
        receipt["CompilerToolsUnchanged"] = bool(args.compile)
        probe = receipt["Gb202ReadOnlyProbeDefinitions"]
        if file_hash(TASK_ROOT / probe["ReadOnlyProbeSource"]) != probe["ReadOnlyProbeSha256"]:
            raise ValueError("Read-only production probe changed during preflight")
        if tool_sources() != sources_before:
            raise ValueError("Port preflight inputs changed while running")
    except Exception as error:
        receipt["Error"] = type(error).__name__ + ": " + str(error)
        (output / "Failure.log").write_text(traceback.format_exc(), encoding="utf-8")
    receipt["PreflightPassed"] = receipt["ReferenceValidationPassed"] and (not args.compile or receipt["AnyRealComponentCompiled"]) and "Error" not in receipt
    (output / "Result.json").write_text(json.dumps(receipt, indent=2) + "\n", encoding="utf-8", newline="\n")
    print(output / "Result.json", flush=True)
    return 0 if receipt["PreflightPassed"] else 1


if __name__ == "__main__":
    raise SystemExit(main())