#!/usr/bin/env python3
"""Hostile and parity tests for the source-only attestation boundary."""

from __future__ import annotations

import copy
import hashlib
import importlib.util
import json
from pathlib import Path
import shutil
import socket
import tempfile
import unittest
from unittest import mock
import urllib.request

import jsonschema


HERE = Path(__file__).resolve().parent
SPEC = importlib.util.spec_from_file_location("attestation_boundary_check", HERE / "check.py")
assert SPEC and SPEC.loader
check = importlib.util.module_from_spec(SPEC)
SPEC.loader.exec_module(check)


def loaded(path: Path):
    return json.loads(path.read_text(encoding="utf-8"))


class AttestationBoundaryTests(unittest.TestCase):
    @classmethod
    def setUpClass(cls) -> None:
        cls.context = loaded(HERE / "context.json")
        cls.report = check.build_report()
        cls.context_schema = loaded(HERE / "context.schema.json")
        cls.report_schema = loaded(HERE / "report.schema.json")
        cls.vendor_lock = loaded(HERE / "vendor-lock.json")
        cls.vendor_lock_schema = loaded(HERE / "vendor-lock.schema.json")

    def test_generated_report_and_schema_close(self) -> None:
        self.assertEqual(loaded(HERE / "generated/report.json"), self.report)
        check.check_context(self.context)
        check.check_report(self.report)
        jsonschema.Draft202012Validator(self.context_schema).validate(self.context)
        jsonschema.Draft202012Validator(self.report_schema).validate(self.report)
        jsonschema.Draft202012Validator(self.vendor_lock_schema).validate(self.vendor_lock)
        check.check_vendor_lock()

    def test_canonical_boundaries(self) -> None:
        air = self.report["artifacts"]["air"]["subclaims"]
        nvidia = self.report["artifacts"]["nvidia"]["subclaims"]
        self.assertEqual(
            {name: value["state"] for name, value in air.items()},
            {
                "device_identity_and_state": "NOT_ESTABLISHED",
                "measured_workload": "NOT_ESTABLISHED",
                "model_commitment": "ESTABLISHED",
                "request_and_nonce_commitment": "ESTABLISHED",
                "response_commitment": "ESTABLISHED",
                "execution_relation": "NOT_ESTABLISHED",
            },
        )
        self.assertEqual(nvidia["device_identity_and_state"]["state"], "ESTABLISHED")
        self.assertTrue(all(value["state"] == "NOT_ESTABLISHED" for name, value in nvidia.items() if name != "device_identity_and_state"))
        self.assertEqual(self.report["combined_execution_relation"]["state"], "NOT_ESTABLISHED")
        self.assertEqual(self.report["live_h100_collection"]["state"], "UNCHECKABLE")
        self.assertNotIn("compromise", json.dumps(self.report).lower())

    def test_all_air_upstream_vectors_reproduce(self) -> None:
        vectors = self.report["artifacts"]["air"]["details"]["vectors"]
        self.assertEqual(len(vectors), 10)
        self.assertTrue(all(item["expected_outcome_reproduced"] is True for item in vectors))

    def test_unknown_context_and_bool_integer_fail_checker_and_schema(self) -> None:
        unknown = copy.deepcopy(self.context)
        unknown["air"]["worldly_truth"] = "ESTABLISHED"
        with self.assertRaises(check.AttestationBoundaryError):
            check.check_context(unknown)
        with self.assertRaises(jsonschema.ValidationError):
            jsonschema.Draft202012Validator(self.context_schema).validate(unknown)
        boolean = copy.deepcopy(self.context)
        boolean["air"]["frozen_now_epoch"] = True
        with self.assertRaises(check.AttestationBoundaryError):
            check.check_context(boolean)
        with self.assertRaises(jsonschema.ValidationError):
            jsonschema.Draft202012Validator(self.context_schema).validate(boolean)

    def test_unknown_report_field_and_state_fail_checker_and_schema(self) -> None:
        unknown = copy.deepcopy(self.report)
        unknown["artifacts"]["air"]["subclaims"]["model_commitment"]["worldly_truth"] = "ESTABLISHED"
        with self.assertRaises(check.AttestationBoundaryError):
            check.check_report(unknown)
        with self.assertRaises(jsonschema.ValidationError):
            jsonschema.Draft202012Validator(self.report_schema).validate(unknown)
        state = copy.deepcopy(self.report)
        state["live_h100_collection"]["state"] = "LIVE_H100_VERIFIED"
        with self.assertRaises(check.AttestationBoundaryError):
            check.check_report(state)
        with self.assertRaises(jsonschema.ValidationError):
            jsonschema.Draft202012Validator(self.report_schema).validate(state)

    def test_report_profile_and_acceptance_cuts_cannot_be_self_resealed(self) -> None:
        profile = copy.deepcopy(self.report)
        profile["profile"] = "attacker.profile/1"
        profile["report_sha256"] = check.digest(check.canonical({key: value for key, value in profile.items() if key != "report_sha256"}))
        with self.assertRaises(check.AttestationBoundaryError):
            check.check_report(profile)
        with self.assertRaises(jsonschema.ValidationError):
            jsonschema.Draft202012Validator(self.report_schema).validate(profile)
        cuts = copy.deepcopy(self.report)
        established = next(item for item in cuts["acceptance_cuts"] if item["cuts"])
        established["cuts"] = []
        cuts["report_sha256"] = check.digest(check.canonical({key: value for key, value in cuts.items() if key != "report_sha256"}))
        with self.assertRaises(check.AttestationBoundaryError):
            check.check_report(cuts)

    def test_vendor_source_substitution_fails_before_evaluation(self) -> None:
        with tempfile.TemporaryDirectory() as directory:
            root = Path(directory) / "attestation-boundary"
            shutil.copytree(HERE, root)
            vector = root / "vendor/air/vectors/valid/v1-tdx-with-nonce.json"
            vector.write_bytes(vector.read_bytes() + b" ")
            with mock.patch.object(check, "ROOT", root), mock.patch.object(check, "VENDOR_LOCK", root / "vendor-lock.json"):
                with self.assertRaises(check.AttestationBoundaryError):
                    check.check_vendor_lock()

    def test_vendor_lock_cannot_bootstrap_a_replacement_source_closure(self) -> None:
        replacement = copy.deepcopy(self.vendor_lock)
        replacement["members"][0]["sha256"] = "sha256:" + "0" * 64
        replacement["lock_root"] = check.digest(check.canonical({key: value for key, value in replacement.items() if key != "lock_root"}))
        with tempfile.TemporaryDirectory() as directory:
            path = Path(directory) / "vendor-lock.json"
            path.write_bytes(check.canonical(replacement))
            with mock.patch.object(check, "VENDOR_LOCK", path):
                with self.assertRaises(check.AttestationBoundaryError):
                    check.check_vendor_lock()
        with self.assertRaises(jsonschema.ValidationError):
            jsonschema.Draft202012Validator(self.vendor_lock_schema).validate(replacement)

    def test_verifier_dependency_substitution_fails(self) -> None:
        real_version = check.importlib.metadata.version

        def substituted(distribution: str) -> str:
            return "6.1.4" if distribution == "cbor2" else real_version(distribution)

        with mock.patch.object(check.importlib.metadata, "version", substituted):
            with self.assertRaises(check.AttestationBoundaryError):
                check.check_vendor_lock()

    def test_nvidia_report_byte_substitution_fails_signature(self) -> None:
        with tempfile.TemporaryDirectory() as directory:
            root = Path(directory) / "attestation-boundary"
            shutil.copytree(HERE, root)
            report = root / "vendor/nvtrust/guest_tools/gpu_verifiers/local_gpu_verifier/src/verifier/samples/hopperAttestationReport.txt"
            raw = report.read_text(encoding="ascii")
            index = 8000
            replacement = "0" if raw[index] != "0" else "1"
            report.write_text(raw[:index] + replacement + raw[index + 1 :], encoding="ascii")
            policy = "sha256:" + hashlib.sha256(check.canonical(self.context)).hexdigest()
            with mock.patch.object(check, "ROOT", root):
                with self.assertRaises(check.AttestationBoundaryError):
                    check.evaluate_nvidia(self.context["nvidia"], policy)

    def test_receiver_root_substitution_fails(self) -> None:
        context = copy.deepcopy(self.context)
        context["nvidia"]["accepted_device_root_der_sha256"] = "sha256:" + "0" * 64
        policy = "sha256:" + hashlib.sha256(check.canonical(context)).hexdigest()
        with self.assertRaises(check.AttestationBoundaryError):
            check.evaluate_nvidia(context["nvidia"], policy)

    def test_nvidia_policy_is_in_the_acceptance_cut(self) -> None:
        device = next(
            item for item in self.report["acceptance_cuts"]
            if item["conclusion"] == "nvidia.device_identity_and_state"
        )
        self.assertEqual(len(device["cuts"]), 1)
        self.assertIn("nvidia.accepted_policy", device["cuts"][0])

    def test_report_format_is_parser_policy_not_a_signed_architecture_claim(self) -> None:
        context = copy.deepcopy(self.context)
        context["nvidia"]["expected_report_format"] = "BLACKWELL_ATTESTATION_REPORT"
        with self.assertRaises(check.AttestationBoundaryError):
            check.check_context(context)
        with self.assertRaises(jsonschema.ValidationError):
            jsonschema.Draft202012Validator(self.context_schema).validate(context)
        details = self.report["artifacts"]["nvidia"]["details"]
        self.assertEqual(details["report_format_jurisdiction"], "PINNED_PARSER_PROFILE_NOT_SIGNED_REPORT_FIELD")
        self.assertNotIn("architecture", details)

    def test_context_scalar_constraints_match_the_schema(self) -> None:
        cases = [
            ("evaluation_time", "2026-08-24T00:00:00+00:00"),
            ("accepted_device_root_der_sha256", "SHA256:" + "0" * 64),
            ("expected_sample_nonce_hex", "A" * 64),
            ("expected_measurement_count", True),
        ]
        validator = jsonschema.Draft202012Validator(self.context_schema)
        for field, value in cases:
            with self.subTest(field=field):
                context = copy.deepcopy(self.context)
                context["nvidia"][field] = value
                with self.assertRaises(check.AttestationBoundaryError):
                    check.check_context(context)
                with self.assertRaises(jsonschema.ValidationError):
                    validator.validate(context)

    def test_evaluator_uses_no_network(self) -> None:
        def denied(*_args, **_kwargs):
            raise AssertionError("network access attempted")

        with mock.patch.object(socket, "socket", denied), mock.patch.object(urllib.request, "urlopen", denied):
            report = check.build_report()
        self.assertEqual(report["report_sha256"], self.report["report_sha256"])

    def test_acceptance_cuts_exist_only_for_established_conclusions(self) -> None:
        for item in self.report["acceptance_cuts"]:
            if item["state"] == "ESTABLISHED":
                self.assertEqual(len(item["cuts"]), 1)
                self.assertGreater(len(item["cuts"][0]), 0)
            else:
                self.assertEqual(item["cuts"], [])


if __name__ == "__main__":
    unittest.main()
