from __future__ import annotations import copy import json import tempfile import unittest from pathlib import Path from vcore_scripts.protocol_evidence import load_manifest from vcore_scripts.protocol_security_acceptance import ( gate_result, native_envelope, rust_command, vless_pass, ) from vcore_scripts.protocol_security_catalog import FIELDS, definitions, groups class SecurityAcceptanceTest(unittest.TestCase): def test_legacy_ech_gateway_keeps_response_inside_tls(self): from vcore_scripts.protocol_ech import legacy_gateway_command self.assertEqual( legacy_gateway_command(), [ "env", "XRAY_BUF_SPLICE=disable", "/data/fixture/peer", "run", "-c", "/data/fixture/config.json", ], ) def test_frozen_scope_and_compositions(self): cases = definitions() self.assertEqual( [c for c in load_manifest() if c["stage"] == "SECURITY"], cases ) self.assertEqual(len(cases), 43) self.assertEqual(set().union(*(set(c["row_ids"]) for c in cases)), FIELDS) self.assertEqual(len(FIELDS), 11) catalog = groups() self.assertEqual(len(catalog["SECURITY-ECH-STANDARD"]["selected"]), 50) for profile in ("chrome", "chrome120", "firefox", "safari"): self.assertEqual( catalog["SECURITY-ECH-" + profile.upper()]["client_fingerprint"], profile, ) for security in ("ECH", "JLS", "HYBRID"): self.assertEqual( sum(k.startswith(f"SECURITY-{security}-ENCRYPTION-") for k in catalog), 6, ) self.assertFalse( any("shadow" in str(c).lower() or "restls" in str(c).lower() for c in cases) ) def fixture(self, directory): name, test = "SECURITY-ECH-TCP-TLS-REJECT", "native_ech_fail_closed" group = dict(runner="vless", selected=[name], ech=True) report = dict( status="PASS", cleanup=True, source_unchanged=True, source=dict( parent_commit="7" * 40, source_tree_sha256="a" * 64, dirty_patch_sha256="b" * 64, lock_sha256="c" * 64, ), ech=True, jls=False, encryption_profile=None, client_fingerprint=None, cases=[ dict( case_id=name, command=rust_command("vless_native", test), exit_code=0, cleanup=True, command_cleanup=True, status="PASS", peer_kind="M", ) ], isolation=dict( network_mode="hostOnly", host_servers=False, guest_mtu=1500, image_digest="sha256:" + "d" * 64, peers=[dict(started=True, joined=True)], ), peers={ "M": dict( version="synthetic test metadata", binary_sha256="a" * 64, archive_sha256="b" * 64, source_url="https://github.com/MetaCubeX/mihomo/releases/latest/download/fixture", ) }, ) path = directory / (name + "-events.jsonl") events = [ dict(schema_version=1, suite="SECURITY-WIRE", assertion=test, status=s) for s in ("BEGIN", "PASS") ] path.write_text("".join(json.dumps(e) + "\n" for e in events)) return group, report, path, events def test_native_missing_duplicate_wrong_events_fail(self): with tempfile.TemporaryDirectory() as temp: directory = Path(temp) group, report, path, events = self.fixture(directory) self.assertTrue(vless_pass(group, report, directory)) for samples in ([], report["cases"] * 2): invalid = dict(report, cases=samples) self.assertFalse(vless_pass(group, invalid, directory)) for samples in ( events[:1], events * 2, [dict(e, suite="VLESS-WIRE") for e in events], [dict(e, status="PASS") for e in events], ): path.write_text("".join(json.dumps(e) + "\n" for e in samples)) self.assertFalse(vless_pass(group, report, directory)) def test_source_cleanup_and_identity_are_not_inferred(self): with tempfile.TemporaryDirectory() as temp: _, report, _, _ = self.fixture(Path(temp)) self.assertTrue(native_envelope(report, report["peers"], report["source"])) self.assertTrue(native_envelope(report, image_digest="sha256:" + "d" * 64)) self.assertFalse(native_envelope(report, image_digest="sha256:" + "e" * 64)) for field in ("cleanup", "source_unchanged"): self.assertFalse(native_envelope(dict(report, **{field: False}))) self.assertFalse(native_envelope(report, {}, report["source"])) self.assertFalse(native_envelope(report, report["peers"], {})) for mutate in ( lambda r: r["isolation"].update(host_servers=True), lambda r: r["isolation"].update(peers=[]), lambda r: r["isolation"]["peers"][0].update(joined=False), lambda r: r["peers"]["M"].update(binary_sha256=""), ): invalid = copy.deepcopy(report) mutate(invalid) self.assertFalse(native_envelope(invalid)) def test_close_reference_cannot_hide_a_different_official_profile(self): with tempfile.TemporaryDirectory() as temp: directory = Path(temp) group, report, _, _ = self.fixture(directory) name, test = "SECURITY-ECH-GRPC-TLS-CLOSE", "native_mihomo_close_alignment" group.update(selected=[name], client_fingerprint="safari") report.update(client_fingerprint="safari") report["cases"][0].update( case_id=name, command=rust_command("vless_native", test) ) (directory / (name + "-events.jsonl")).write_text( "".join( json.dumps( dict( schema_version=1, suite="SECURITY-WIRE", assertion=test, status=s, ) ) + "\n" for s in ("BEGIN", "PASS") ) ) path = directory / "grpc-tls-tls-close-reference.json" self.assertFalse(vless_pass(group, report, directory)) valid = dict( scope="ech-safari-chrome-baseline", client_fingerprint="chrome", dut_client_fingerprint="safari", terminated=True, tail_hex="", ) path.write_text(json.dumps(valid)) self.assertTrue(vless_pass(group, report, directory)) for field, value in ( ("scope", "same-mode"), ("client_fingerprint", "safari"), ("dut_client_fingerprint", "chrome"), ("terminated", False), ): path.write_text(json.dumps(dict(valid, **{field: value}))) self.assertFalse(vless_pass(group, report, directory)) def test_empty_gate_and_missing_script_proof_fail(self): cases = {c["case_id"]: c for c in definitions()} for name in ("SECURITY-CFG", "SECURITY-RELEASE", "SECURITY-SCRIPTS"): self.assertEqual(gate_result(cases[name], [], [])["status"], "FAIL") record = [dict(exit_code=0, cleanup=True)] self.assertEqual( gate_result(cases["SECURITY-CFG"], record, [])["status"], "FAIL" ) self.assertEqual( gate_result(cases["SECURITY-SCRIPTS"], record, [], {})["status"], "FAIL" )