XFESystem

#!/usr/bin/env python3
"""Negative provenance guards against actual pinned reference files, never GPU mocks."""
import copy
import json
import pathlib
import time
import unittest

import CaptureReferences as capture
import Preflight as preflight


class ReferenceGuards(unittest.TestCase):
    @classmethod
    def setUpClass(cls):
        cls.lock = json.loads(pathlib.Path(__file__).with_name("RmSources.lock.json").read_text(encoding="utf-8"))

    def test_actual_pinned_cache_validates(self):
        self.assertGreater(len(preflight.verify_references(self.lock)), 10)

    def test_unknown_release_is_rejected(self):
        lock = copy.deepcopy(self.lock)
        lock["Commit"] = "0" * 40
        with self.assertRaisesRegex(ValueError, "identity"):
            preflight.verify_references(lock)

    def test_expected_sha256_tamper_cannot_accept_real_source(self):
        lock = copy.deepcopy(self.lock)
        lock["Files"][0]["Sha256"] = "0" * 64
        with self.assertRaisesRegex(ValueError, "Source/hash differs"):
            preflight.verify_references(lock)

    def test_git_object_hash_is_checked_in_addition_to_sha256(self):
        lock = copy.deepcopy(self.lock)
        lock["Files"][0]["GitBlob"] = "0" * 40
        with self.assertRaisesRegex(ValueError, "Source/hash differs"):
            preflight.verify_references(lock)

    def test_duplicate_paths_are_rejected(self):
        lock = copy.deepcopy(self.lock)
        lock["Files"].append(copy.deepcopy(lock["Files"][0]))
        with self.assertRaisesRegex(ValueError, "Duplicate"):
            preflight.verify_references(lock)

    def test_source_receipt_cannot_assert_hardware_success(self):
        for field in ("GuestCudaPassed", "GpuExecutionPassed"):
            lock = copy.deepcopy(self.lock)
            lock[field] = True
            with self.assertRaisesRegex(ValueError, "hardware"):
                preflight.verify_references(lock)

    def test_components_and_include_search_order_are_bound(self):
        for field in ("Components", "IncludeDirectories"):
            lock = copy.deepcopy(self.lock)
            lock[field] = list(reversed(lock[field]))
            with self.assertRaisesRegex(ValueError, "Unreviewed"):
                preflight.verify_references(lock)

    def test_unsupported_git_file_mode_is_rejected(self):
        lock = copy.deepcopy(self.lock)
        lock["Files"][0]["Mode"] = "120000"
        with self.assertRaisesRegex(ValueError, "Unsupported Git file mode"):
            preflight.verify_references(lock)

    def test_missing_per_file_license_provenance_is_rejected(self):
        lock = copy.deepcopy(self.lock)
        lock["Files"][0]["License"] = ""
        with self.assertRaisesRegex(ValueError, "license"):
            preflight.verify_references(lock)

    def test_cache_budget_cannot_be_relaxed_by_lock(self):
        lock = copy.deepcopy(self.lock)
        lock["MaximumCacheBytes"] += 1
        with self.assertRaisesRegex(ValueError, "bound differs"):
            preflight.verify_references(lock)

    def test_windows_absolute_traversal_and_ads_paths_are_rejected(self):
        for relative in ("../Host", "/Host", "C:/Host", "A:Stream", "A\\B"):
            with self.assertRaisesRegex(ValueError, "Unsafe source path"):
                capture.source_path(relative)

    def test_actual_gb202_headers_match_production_read_only_probe(self):
        result = preflight.verify_probe_definitions(self.lock["Files"])
        self.assertTrue(result["ProductionRegisterMatches"])
        self.assertFalse(result["DeviceReadPerformed"])
        self.assertFalse(result["GpuExecutionPassed"])
        self.assertEqual(result["RegisterOffset"] & 4095, 0xbc)

    def test_different_gb202_header_cannot_bind_production_probe(self):
        rows = copy.deepcopy(self.lock["Files"])
        next(row for row in rows if row["Path"] in preflight.PROBE_HEADERS)["Sha256"] = "0" * 64
        with self.assertRaisesRegex(ValueError, "header identity differs"):
            preflight.verify_probe_definitions(rows)

    def test_source_transport_rejects_non_https_and_unofficial_origins(self):
        for url in ("http://raw.githubusercontent.com/", "https://example.org/", "https://raw.githubusercontent.com:444/", "https://user@raw.githubusercontent.com/"):
            with self.assertRaisesRegex(ValueError, "official HTTPS origins"):
                capture.download(url, 1)


if __name__ == "__main__":
    before = preflight.tool_sources()
    program = unittest.main(exit=False)
    result = program.result
    output = capture.TASK_ROOT / "build/Gpu/RmPort" / ("Guards-" + str(time.time_ns()))
    output.mkdir(parents=True)
    summary = {"Kind": "XfeNvidiaRmPortReferenceGuards", "Tests": result.testsRun,
               "Failures": len(result.failures), "Errors": len(result.errors), "Skipped": len(result.skipped),
               "Passed": result.wasSuccessful() and before == preflight.tool_sources(), "SourceSha256": before,
               "GuestCudaPassed": False, "GpuExecutionPassed": False}
    (output / "Result.json").write_text(json.dumps(summary, indent=2) + "\n", encoding="utf-8")
    print(output / "Result.json")
    raise SystemExit(0 if summary["Passed"] else 1)