|
| 1 | +#!/usr/bin/env python3 |
| 2 | +"""Sanitized P7.6 fail-closed qualification fixtures; no device/network access.""" |
| 3 | +from __future__ import annotations |
| 4 | + |
| 5 | +import importlib.util |
| 6 | +import json |
| 7 | +from pathlib import Path |
| 8 | +import subprocess |
| 9 | +import sys |
| 10 | +import tempfile |
| 11 | +import unittest |
| 12 | + |
| 13 | +SCRIPT = Path(__file__).with_name("verify-p7-dual-ingress-diagnostic.py") |
| 14 | +spec = importlib.util.spec_from_file_location("p7_qualifier", SCRIPT) |
| 15 | +assert spec and spec.loader |
| 16 | +qualifier = importlib.util.module_from_spec(spec) |
| 17 | +spec.loader.exec_module(qualifier) |
| 18 | +SHA = "a" * 64 |
| 19 | +STAMP = "2026-10-08T10:01:02.0000000+00:00" |
| 20 | + |
| 21 | + |
| 22 | +def ingress(label: str, runtime: str, *, fingerprint: str = SHA, |
| 23 | + brcb: int = 6, urcb: int = 6, |
| 24 | + planned: int = 2, routed: int = 2, |
| 25 | + first: str = STAMP, brcb_bindings: int = 6, urcb_bindings: int = 6, |
| 26 | + last_status: str = "ROUTED") -> str: |
| 27 | + return ( |
| 28 | + f" {label} : {runtime}: fingerprint={fingerprint}, requested=12, " |
| 29 | + f"plans={planned}, staticBRCB={brcb}, staticURCB={urcb}, uncovered=0\n" |
| 30 | + f" traffic : {routed}/{planned} exact static RCB target(s) routed; first={first}\n" |
| 31 | + f" runtime : static-brcb|dataset=ied/lln0.events|rcb=ied/lln0.br.rpt_ind01|bindings={brcb_bindings} [ROUTED]\n" |
| 32 | + f" runtime : static-urcb|dataset=ied/lln0.analog|rcb=ied/lln0.rp.rpt_meas01|bindings={urcb_bindings} [{last_status}]\n" |
| 33 | + ) |
| 34 | + |
| 35 | + |
| 36 | +def diagnostic(*, discovery: str | None = None, |
| 37 | + scl: str | None = None, traffic: bool = True, ap: str = "F") -> str: |
| 38 | + routed = "DUAL INGRESS TRAFFIC PROVEN (routed static RCB data from both paths)" if traffic else "TRAFFIC PENDING" |
| 39 | + return ( |
| 40 | + "ARSAS Diagnostic Report\n" |
| 41 | + "IED : GR_X_7SX85\n" |
| 42 | + "Endpoint : 198.51.100.12:102\n" |
| 43 | + f"SCD selected AP : {ap}\n" |
| 44 | + f"Static parity : MATCH • {SHA}\n" |
| 45 | + f"Routed traffic : {routed}\n" |
| 46 | + + (discovery if discovery is not None else ingress("Discovery", "LiveDiscovery")) |
| 47 | + + (scl if scl is not None else ingress("Open SCL ", "OpenScl")) |
| 48 | + ) |
| 49 | + |
| 50 | + |
| 51 | +class QualificationTests(unittest.TestCase): |
| 52 | + def assertNotQualified(self, text: str, message: str, **kwargs) -> None: |
| 53 | + with self.assertRaisesRegex(qualifier.QualificationError, message): |
| 54 | + qualifier.qualify(text, ied="GR_X_7SX85", **kwargs) |
| 55 | + |
| 56 | + def test_exact_two_ingresses_both_brcb_urcb_with_source_identity(self): |
| 57 | + result = qualifier.qualify(diagnostic(), ied="GR_X_7SX85", |
| 58 | + min_brcb=6, min_urcb=6, selected_ap="F") |
| 59 | + self.assertFalse(result["physicalVerification"]) |
| 60 | + self.assertFalse(result["wirePcapVerification"]) |
| 61 | + self.assertIn("NOT_PHYSICAL_VERIFIED", result["qualification"]) |
| 62 | + self.assertEqual(result["discovery"]["plannedTargets"], 2) |
| 63 | + self.assertEqual(result["openScl"]["routedTargets"], 2) |
| 64 | + |
| 65 | + def test_semantic_match_alone_does_not_qualify(self): |
| 66 | + self.assertNotQualified(diagnostic(traffic=False), "dual-ingress routed traffic proof") |
| 67 | + |
| 68 | + def test_single_ingress_does_not_qualify(self): |
| 69 | + self.assertNotQualified(diagnostic(scl=""), "Both independent ingress") |
| 70 | + |
| 71 | + def test_pending_target_does_not_qualify(self): |
| 72 | + self.assertNotQualified( |
| 73 | + diagnostic(scl=ingress("Open SCL ", "OpenScl", last_status="PENDING")), |
| 74 | + "unrouted") |
| 75 | + |
| 76 | + def test_forged_count_or_truncated_inventory_does_not_qualify(self): |
| 77 | + self.assertNotQualified( |
| 78 | + diagnostic(scl=ingress("Open SCL ", "OpenScl", planned=9, routed=9)), |
| 79 | + "incomplete/truncated") |
| 80 | + |
| 81 | + def test_brcb_only_is_not_urcb_proof(self): |
| 82 | + self.assertNotQualified(diagnostic(scl=ingress("Open SCL ", "OpenScl") |
| 83 | + .replace("static-urcb|", "static-brcb|") |
| 84 | + .replace("rpt_meas01", "rpt_ind02")), "both BRCB and URCB") |
| 85 | + |
| 86 | + def test_mismatched_semantic_sha_is_rejected(self): |
| 87 | + self.assertNotQualified( |
| 88 | + diagnostic(scl=ingress("Open SCL ", "OpenScl", fingerprint="b" * 64)), |
| 89 | + "fingerprint does not match") |
| 90 | + |
| 91 | + def test_wrong_ied_is_rejected(self): |
| 92 | + with self.assertRaisesRegex(qualifier.QualificationError, "Exact IED identity"): |
| 93 | + qualifier.qualify(diagnostic(), ied="DIFFERENT") |
| 94 | + |
| 95 | + def test_two_sections_cannot_borrow_traffic_from_another_ied(self): |
| 96 | + text = (diagnostic(scl="") + |
| 97 | + "IED : OTHER_IED\n" + |
| 98 | + ingress("Open SCL ", "OpenScl")) |
| 99 | + self.assertNotQualified(text, "Both independent ingress") |
| 100 | + |
| 101 | + def test_duplicate_ied_block_is_rejected(self): |
| 102 | + self.assertNotQualified(diagnostic() + diagnostic(), "IED identity") |
| 103 | + |
| 104 | + def test_selected_ap_must_be_source_exact(self): |
| 105 | + self.assertNotQualified(diagnostic(), "AccessPoint", selected_ap="J") |
| 106 | + |
| 107 | + def test_missing_timestamp_is_not_traffic_proof(self): |
| 108 | + self.assertNotQualified( |
| 109 | + diagnostic(scl=ingress("Open SCL ", "OpenScl", first="pending")), |
| 110 | + "timestamp") |
| 111 | + |
| 112 | + def test_duplicate_target_is_rejected(self): |
| 113 | + target = "static-urcb|dataset=ied/lln0.analog|rcb=ied/lln0.rp.rpt_meas01" |
| 114 | + dup = ingress("Open SCL ", "OpenScl").replace(target, |
| 115 | + "static-brcb|dataset=ied/lln0.events|rcb=ied/lln0.br.rpt_ind01") |
| 116 | + self.assertNotQualified(diagnostic(scl=dup), "duplicate RCB target") |
| 117 | + |
| 118 | + def test_incomplete_or_emergency_report_not_accepted(self): |
| 119 | + self.assertNotQualified( |
| 120 | + diagnostic().replace("ARSAS Diagnostic Report", "ARSAS Emergency Diagnostic Report"), |
| 121 | + "Full Copy Diagnostic") |
| 122 | + |
| 123 | + def test_explicit_6_signal_threshold_is_not_downgraded(self): |
| 124 | + self.assertNotQualified(diagnostic(scl=ingress("Open SCL ", "OpenScl", brcb=5)), |
| 125 | + "insufficient static", min_brcb=6, min_urcb=6) |
| 126 | + |
| 127 | + def test_missing_full_count_even_if_summary_says_proven(self): |
| 128 | + self.assertNotQualified( |
| 129 | + diagnostic(discovery=ingress("Discovery", "LiveDiscovery").split( |
| 130 | + " runtime : static-urcb|")[0]), |
| 131 | + "incomplete/truncated") |
| 132 | + |
| 133 | + |
| 134 | + def test_selected_ap_required_and_proven_even_when_optional_arg_omitted(self): |
| 135 | + verdict = qualifier.qualify(diagnostic(), ied="GR_X_7SX85") |
| 136 | + self.assertEqual("F", verdict["selectedAccessPoint"]) |
| 137 | + self.assertNotQualified(diagnostic().replace("SCD selected AP : F\n", ""), |
| 138 | + "selected AccessPoint") |
| 139 | + self.assertNotQualified(diagnostic().replace("SCD selected AP : F", |
| 140 | + "SCD selected AP : pending"), "not a proven selected AP") |
| 141 | + self.assertNotQualified(diagnostic().replace("SCD selected AP : F", |
| 142 | + "SCD selected AP : F\nSCD selected AP : J"), "exactly one") |
| 143 | + |
| 144 | + def test_each_ingress_requires_exact_per_family_binding_incidence(self): |
| 145 | + # Summary counters cannot substitute for actual target totals. |
| 146 | + for ingress_name, label, source in ( |
| 147 | + ("discovery", "Discovery", "LiveDiscovery"), |
| 148 | + ("scl", "Open SCL ", "OpenScl"), |
| 149 | + ): |
| 150 | + for parameter in ("brcb_bindings", "urcb_bindings"): |
| 151 | + with self.subTest(ingress=ingress_name, family=parameter): |
| 152 | + changed = ingress(label, source, **{parameter: 1}) |
| 153 | + self.assertNotQualified(diagnostic(**{ingress_name: changed}), |
| 154 | + "target bindings do not match") |
| 155 | + |
| 156 | + def test_binding_incidence_can_exceed_unique_requested_signal_count(self): |
| 157 | + # Reused points across RCBs count as multiple binding incidences. |
| 158 | + result = qualifier.qualify(diagnostic().replace("requested=12", "requested=9"), |
| 159 | + ied="GR_X_7SX85", min_brcb=6, min_urcb=6) |
| 160 | + self.assertEqual(6, result["openScl"]["brcbSignals"]) |
| 161 | + self.assertEqual(6, result["discovery"]["urcbSignals"]) |
| 162 | + |
| 163 | + def test_overlong_numeric_fields_and_unicode_digits_fail_closed(self): |
| 164 | + many = "9" * 5000 |
| 165 | + cases = ( |
| 166 | + ("ingress", diagnostic().replace("requested=12", "requested=" + many, 1), |
| 167 | + "Malformed ingress header"), |
| 168 | + ("traffic", diagnostic().replace("traffic : 2/2", |
| 169 | + "traffic : " + many + "/2", 1), |
| 170 | + "Unparseable target or traffic"), |
| 171 | + ("binding", diagnostic().replace("bindings=6 [ROUTED]", |
| 172 | + "bindings=" + many + " [ROUTED]", 1), |
| 173 | + "Unparseable target or traffic"), |
| 174 | + ("unicode", diagnostic().replace("requested=12", "requested=１２", 1), |
| 175 | + "Malformed ingress header"), |
| 176 | + ) |
| 177 | + for name, bad, error in cases: |
| 178 | + with self.subTest(field=name): |
| 179 | + self.assertNotQualified(bad, error) |
| 180 | + |
| 181 | + def test_cli_long_numeric_recheck_cannot_leave_stale_pass(self): |
| 182 | + with tempfile.TemporaryDirectory() as tmp: |
| 183 | + root = Path(tmp) |
| 184 | + incoming, result = root / "private.txt", root / "verdict.json" |
| 185 | + cmd = [sys.executable, str(SCRIPT), "--diagnostic", str(incoming), |
| 186 | + "--ied", "GR_X_7SX85", "--output", str(result)] |
| 187 | + incoming.write_text(diagnostic(), encoding="utf-8") |
| 188 | + self.assertEqual(0, subprocess.run(cmd, capture_output=True).returncode) |
| 189 | + self.assertIn("PASS", json.loads(result.read_text())["qualification"]) |
| 190 | + incoming.write_text( |
| 191 | + diagnostic().replace("requested=12", "requested=" + "9" * 5000, 1), |
| 192 | + encoding="utf-8") |
| 193 | + rejected = subprocess.run(cmd, capture_output=True, text=True) |
| 194 | + self.assertNotEqual(0, rejected.returncode) |
| 195 | + self.assertNotIn("Traceback", rejected.stderr) |
| 196 | + saved = json.loads(result.read_text()) |
| 197 | + self.assertEqual("NOT_ESTABLISHED", saved["qualification"]) |
| 198 | + self.assertFalse(saved["physicalVerification"]) |
| 199 | + self.assertFalse(saved["wirePcapVerification"]) |
| 200 | + self.assertFalse(list(root.glob(".verdict.json.*.tmp"))) |
| 201 | + |
| 202 | + def test_large_or_nul_diagnostic_fail_closed(self): |
| 203 | + self.assertNotQualified("ARSAS Diagnostic Report\n" + |
| 204 | + "X" * (qualifier.MAX_DIAGNOSTIC_BYTES + 1), "exceeds safe size") |
| 205 | + self.assertNotQualified(diagnostic() + "\x00", "contains NUL") |
| 206 | + |
| 207 | + def test_cli_failed_recheck_replaces_old_positive_verdict(self): |
| 208 | + with tempfile.TemporaryDirectory() as tmp: |
| 209 | + root = Path(tmp) |
| 210 | + incoming, result = root / "private.txt", root / "result.json" |
| 211 | + cmd = [sys.executable, str(SCRIPT), "--diagnostic", str(incoming), |
| 212 | + "--ied", "GR_X_7SX85", "--selected-ap", "F", |
| 213 | + "--min-brcb-signals", "6", "--min-urcb-signals", "6", |
| 214 | + "--output", str(result)] |
| 215 | + incoming.write_text(diagnostic(), encoding="utf-8") |
| 216 | + self.assertEqual(0, subprocess.run(cmd, capture_output=True).returncode) |
| 217 | + self.assertEqual("DIAGNOSTIC_CONSISTENCY_PASS_NOT_PHYSICAL_VERIFIED", |
| 218 | + json.loads(result.read_text())["qualification"]) |
| 219 | + incoming.write_text(diagnostic(traffic=False), encoding="utf-8") |
| 220 | + self.assertNotEqual(0, subprocess.run(cmd, capture_output=True).returncode) |
| 221 | + failed = json.loads(result.read_text()) |
| 222 | + self.assertEqual("NOT_ESTABLISHED", failed["qualification"]) |
| 223 | + self.assertFalse(failed["physicalVerification"]) |
| 224 | + self.assertFalse(failed["wirePcapVerification"]) |
| 225 | + self.assertFalse(list(root.glob(".result.json.*.tmp"))) |
| 226 | + |
| 227 | + def test_bounded_file_and_result_symlink_fail_closed(self): |
| 228 | + with tempfile.TemporaryDirectory() as tmp: |
| 229 | + root = Path(tmp) |
| 230 | + f = root / "huge.txt" |
| 231 | + f.write_bytes(b"Z" * (qualifier.MAX_DIAGNOSTIC_BYTES + 1)) |
| 232 | + with self.assertRaisesRegex(qualifier.QualificationError, "exceeds safe size"): |
| 233 | + qualifier.read_bounded_diagnostic(f) |
| 234 | + destination, link = root / "other.json", root / "symlink.json" |
| 235 | + destination.write_text('{"qualification":"KEEP"}') |
| 236 | + try: |
| 237 | + link.symlink_to(destination) |
| 238 | + except (OSError, NotImplementedError): |
| 239 | + self.skipTest("Symlinks unavailable") |
| 240 | + with self.assertRaisesRegex(OSError, "Refusing symlink"): |
| 241 | + qualifier.write_result(link, {"qualification": "PASS"}) |
| 242 | + self.assertEqual('{"qualification":"KEEP"}', destination.read_text()) |
| 243 | + |
| 244 | +if __name__ == "__main__": |
| 245 | + unittest.main() |
0 commit comments