diff --git a/.github/workflows/scl-interoperability-r7.yml b/.github/workflows/scl-interoperability-r7.yml index 4c29dd4f1..40daeb248 100644 --- a/.github/workflows/scl-interoperability-r7.yml +++ b/.github/workflows/scl-interoperability-r7.yml @@ -18,6 +18,8 @@ on: - "docs/INTEROPERABILITY_REFERENCE_CONTRACT.md" - "tests/ARSAS.Tests/CanonicalLiveSclExportRegressionTests.cs" - "tests/ARSAS.Tests/CanonicalSclReloadValidatorTests.cs" + - "scripts/verify-p7-dual-ingress-diagnostic.py" + - "scripts/test-p7-dual-ingress-diagnostic.py" - ".github/workflows/scl-interoperability-r7.yml" workflow_dispatch: @@ -48,6 +50,17 @@ jobs: } "ARSAS_COMMIT=$arsasCommit" | Out-File $env:GITHUB_ENV -Encoding utf8 -Append + - name: Set up Python for read-only physical-intake guard + uses: actions/setup-python@v6 + with: + python-version: '3.12' + + - name: P7.6 verify field diagnostic parser safety fixtures + shell: pwsh + run: | + python .\ARSAS\scripts\test-p7-dual-ingress-diagnostic.py + if ($LASTEXITCODE -ne 0) { throw 'P7.6 diagnostic qualification safety fixtures failed.' } + - name: Resolve current engine while preserving R7/R10 authority shell: pwsh run: | diff --git a/docs/audits/P7_6_FIELD_QUALIFICATION_INTAKE.md b/docs/audits/P7_6_FIELD_QUALIFICATION_INTAKE.md new file mode 100644 index 000000000..90e18d35c --- /dev/null +++ b/docs/audits/P7_6_FIELD_QUALIFICATION_INTAKE.md @@ -0,0 +1,37 @@ +# P7.6 — Read-only dual-ingress field diagnostic intake (draft) + +Parent: issue #446, stacked after draft P7.5 PR #476. No Studio runtime, MMS engine, IED association, RCB planner, native installer, physical R10 or release authority changes. + +## Purpose + +P7.3 records semantic Discovery/Open-SCL fingerprints; P7.4 separately tracks actual routed values on each exact static BRCB and URCB target. The native Studio Copy Diagnostic truncates individual RCB target lines to eight per ingress. Semantic MATCH alone is not traffic proof. + +This milestone adds a fail-closed local validator of one **full native Studio Copy Diagnostic** for exactly one IED. It requires a matching SHA-256 fingerprint in both independent ingress summaries, explicit dual-ingress traffic qualification, nonzero selected static BRCB and URCB coverage, zero uncovered signals, an unambiguous complete per-ingress target inventory, accepted ROUTED status per distinct exact RCB, and a timezone-aware first routed timestamp. It rejects missing/duplicate IED sections, emergency reports, mismatched ingresses, empty or ambiguous target identities, any target inventory truncated beyond eight and wrong selected AccessPoint. + +Its JSON output is explicitly DIAGNOSTIC_CONSISTENCY_PASS_NOT_PHYSICAL_VERIFIED. Both physicalVerification and wirePcapVerification are **false** regardless of an otherwise valid diagnostic. Offline text can be copied or forged; this validator is not independent wire or physical verification. + +## Local 7SX85 field procedure + +1. Use the live IED Discovery IP path and choose a Static DataSet report-only selection covering BRCB and URCB. Wait for genuine routed process values on all configured targets, not only RptEna/GI. +2. Open the same SCD, choose the exact MMS AccessPoint J or F from the existing native WPF control, reconnect using a fresh MMS association, and repeat. Do not use cyclic MMS process polling as a reporting fallback. +3. Save the **full Copy Diagnostic** only after both ingresses have run in that device session. Keep customer names, endpoints, addresses and the original unredacted report outside the repository. +4. Run locally with the actual IED and case-specific AP and thresholds: + + python scripts/verify-p7-dual-ingress-diagnostic.py --diagnostic ./private-diagnostic.txt --ied GR_X_7SX85 --selected-ap F --min-brcb-signals 6 --min-urcb-signals 6 --output ./private-result.json + +5. Independently review **real MMS/PCAP** for both APs and both planned report families; inspect association/RCB/DataSet, accepted InformationReports, timestamps and absence of cyclic MMS process polling. Securely record source/engine SHA and PCAP provenance. The offline tool cannot assert physical identity, packet origin, or actual AP J/F connectivity. + +## Anti-naive acceptance limitations + +- The source-native format must include exactly one explicit MATCH and DUAL INGRESS TRAFFIC PROVEN for the exact IED, plus two different ingress identities. It does not rely on matching indexed RCB instance names across different associations. +- The tool enumerates all reported concrete targets and demands equality with the planning count. Because Copy Diagnostic intentionally prints at most eight, it refuses any larger target plan rather than accepting a hidden or partial list. +- Minimum BRCB and URCB signal thresholds are caller-supplied. For a 6+6 physical scenario use 6 and 6; never infer missing signals. +- It cannot verify all APs were physically tested, packet-level causality or real-time freshness. No output authorizes merge, release, engine promotion, or retirement of R10 field authority. + +## Tests, ownership and rollback + +The new scripts/verify-p7-dual-ingress-diagnostic.py is read-only and uses Python standard library. The companion scripts/test-p7-dual-ingress-diagnostic.py has **16 sanitized negative/positive fixtures** and makes no network call. An early check in the existing SCL Interoperability R7 Windows CI runs these fixtures, with no new workflow and no additional native/.NET build. + +Implementation lives in a new stacked **draft child PR** based on the exact P7.5 head. It does not commit onto P7.5 or other threads' branches. Software CI alone is CodeVerified only. The draft must not be merged into main until separately authorized physical test evidence is accepted. + +Rollback: remove only the two scripts, this document and the small R7 path/fixture step; never change runtime/engine code to accommodate a diagnostic-only test. diff --git a/scripts/test-p7-dual-ingress-diagnostic.py b/scripts/test-p7-dual-ingress-diagnostic.py new file mode 100644 index 000000000..2a35c1d2e --- /dev/null +++ b/scripts/test-p7-dual-ingress-diagnostic.py @@ -0,0 +1,130 @@ +#!/usr/bin/env python3 +"""Sanitized P7.6 fail-closed qualification fixtures; no device/network access.""" +from __future__ import annotations + +import importlib.util +from pathlib import Path +import unittest + +SCRIPT = Path(__file__).with_name("verify-p7-dual-ingress-diagnostic.py") +spec = importlib.util.spec_from_file_location("p7_qualifier", SCRIPT) +assert spec and spec.loader +qualifier = importlib.util.module_from_spec(spec) +spec.loader.exec_module(qualifier) +SHA = "a" * 64 +STAMP = "2026-10-08T10:01:02.0000000+00:00" + + +def ingress(label: str, runtime: str, *, fingerprint: str = SHA, + brcb: int = 6, urcb: int = 6, + planned: int = 2, routed: int = 2, + first: str = STAMP, last_status: str = "ROUTED") -> str: + return ( + f" {label} : {runtime}: fingerprint={fingerprint}, requested=12, " + f"plans={planned}, staticBRCB={brcb}, staticURCB={urcb}, uncovered=0\n" + f" traffic : {routed}/{planned} exact static RCB target(s) routed; first={first}\n" + f" runtime : static-brcb|dataset=ied/lln0.events|rcb=ied/lln0.br.rpt_ind01|bindings=6 [ROUTED]\n" + f" runtime : static-urcb|dataset=ied/lln0.analog|rcb=ied/lln0.rp.rpt_meas01|bindings=6 [{last_status}]\n" + ) + + +def diagnostic(*, discovery: str | None = None, + scl: str | None = None, traffic: bool = True, ap: str = "F") -> str: + routed = "DUAL INGRESS TRAFFIC PROVEN (routed static RCB data from both paths)" if traffic else "TRAFFIC PENDING" + return ( + "ARSAS Diagnostic Report\n" + "IED : GR_X_7SX85\n" + "Endpoint : 198.51.100.12:102\n" + f"SCD selected AP : {ap}\n" + f"Static parity : MATCH • {SHA}\n" + f"Routed traffic : {routed}\n" + + (discovery if discovery is not None else ingress("Discovery", "LiveDiscovery")) + + (scl if scl is not None else ingress("Open SCL ", "OpenScl")) + ) + + +class QualificationTests(unittest.TestCase): + def assertNotQualified(self, text: str, message: str, **kwargs) -> None: + with self.assertRaisesRegex(qualifier.QualificationError, message): + qualifier.qualify(text, ied="GR_X_7SX85", **kwargs) + + def test_exact_two_ingresses_both_brcb_urcb_with_source_identity(self): + result = qualifier.qualify(diagnostic(), ied="GR_X_7SX85", + min_brcb=6, min_urcb=6, selected_ap="F") + self.assertFalse(result["physicalVerification"]) + self.assertFalse(result["wirePcapVerification"]) + self.assertIn("NOT_PHYSICAL_VERIFIED", result["qualification"]) + self.assertEqual(result["discovery"]["plannedTargets"], 2) + self.assertEqual(result["openScl"]["routedTargets"], 2) + + def test_semantic_match_alone_does_not_qualify(self): + self.assertNotQualified(diagnostic(traffic=False), "dual-ingress routed traffic proof") + + def test_single_ingress_does_not_qualify(self): + self.assertNotQualified(diagnostic(scl=""), "Both independent ingress") + + def test_pending_target_does_not_qualify(self): + self.assertNotQualified( + diagnostic(scl=ingress("Open SCL ", "OpenScl", last_status="PENDING")), + "unrouted") + + def test_forged_count_or_truncated_inventory_does_not_qualify(self): + self.assertNotQualified( + diagnostic(scl=ingress("Open SCL ", "OpenScl", planned=9, routed=9)), + "incomplete/truncated") + + def test_brcb_only_is_not_urcb_proof(self): + self.assertNotQualified(diagnostic(scl=ingress("Open SCL ", "OpenScl") + .replace("static-urcb|", "static-brcb|") + .replace("rpt_meas01", "rpt_ind02")), "both BRCB and URCB") + + def test_mismatched_semantic_sha_is_rejected(self): + self.assertNotQualified( + diagnostic(scl=ingress("Open SCL ", "OpenScl", fingerprint="b" * 64)), + "fingerprint does not match") + + def test_wrong_ied_is_rejected(self): + with self.assertRaisesRegex(qualifier.QualificationError, "Exact IED identity"): + qualifier.qualify(diagnostic(), ied="DIFFERENT") + + def test_two_sections_cannot_borrow_traffic_from_another_ied(self): + text = (diagnostic(scl="") + + "IED : OTHER_IED\n" + + ingress("Open SCL ", "OpenScl")) + self.assertNotQualified(text, "Both independent ingress") + + def test_duplicate_ied_block_is_rejected(self): + self.assertNotQualified(diagnostic() + diagnostic(), "IED identity") + + def test_selected_ap_must_be_source_exact(self): + self.assertNotQualified(diagnostic(), "AccessPoint", selected_ap="J") + + def test_missing_timestamp_is_not_traffic_proof(self): + self.assertNotQualified( + diagnostic(scl=ingress("Open SCL ", "OpenScl", first="pending")), + "timestamp") + + def test_duplicate_target_is_rejected(self): + target = "static-urcb|dataset=ied/lln0.analog|rcb=ied/lln0.rp.rpt_meas01" + dup = ingress("Open SCL ", "OpenScl").replace(target, + "static-brcb|dataset=ied/lln0.events|rcb=ied/lln0.br.rpt_ind01") + self.assertNotQualified(diagnostic(scl=dup), "duplicate RCB target") + + def test_incomplete_or_emergency_report_not_accepted(self): + self.assertNotQualified( + diagnostic().replace("ARSAS Diagnostic Report", "ARSAS Emergency Diagnostic Report"), + "Full Copy Diagnostic") + + def test_explicit_6_signal_threshold_is_not_downgraded(self): + self.assertNotQualified(diagnostic(scl=ingress("Open SCL ", "OpenScl", brcb=5)), + "insufficient static", min_brcb=6, min_urcb=6) + + def test_missing_full_count_even_if_summary_says_proven(self): + self.assertNotQualified( + diagnostic(discovery=ingress("Discovery", "LiveDiscovery").split( + " runtime : static-urcb|")[0]), + "incomplete/truncated") + + +if __name__ == "__main__": + unittest.main() diff --git a/scripts/verify-p7-dual-ingress-diagnostic.py b/scripts/verify-p7-dual-ingress-diagnostic.py new file mode 100644 index 000000000..6061f21ec --- /dev/null +++ b/scripts/verify-p7-dual-ingress-diagnostic.py @@ -0,0 +1,176 @@ +#!/usr/bin/env python3 +"""P7.6 read-only qualification of one ARSAS full Copy Diagnostic snapshot. + +Diagnostic consistency is NOT physical, wire, PCAP, R10 or release authority. +Input is kept local. Never upload unredacted customer endpoints or reports. +""" +from __future__ import annotations + +import argparse +from datetime import datetime +import json +from pathlib import Path +import re +import sys + +HEX64 = re.compile(r"^[0-9a-f]{64}$") +IED = re.compile(r"^IED\s+:\s*(.+?)\s*$", re.M) +FINGERPRINT = re.compile(r"^Static parity\s+:\s*MATCH\s+•\s*([0-9a-f]{64})\s*$", re.M) +TRAFFIC = re.compile(r"^Routed traffic\s+:\s*DUAL INGRESS TRAFFIC PROVEN\b", re.M) +INGRESS = re.compile( + r"^\s{2}(Discovery|Open SCL)\s+:\s*(LiveDiscovery|OpenScl): " + r"fingerprint=([0-9a-f]{64}), requested=(\d+), plans=(\d+), " + r"staticBRCB=(\d+), staticURCB=(\d+), uncovered=(\d+)\s*$" +) +SUMMARY = re.compile(r"^\s{4}traffic\s+:\s*(\d+)/(\d+) exact static RCB target\(s\) routed; first=(\S+)\s*$") +TARGET = re.compile( + r"^\s{4}runtime\s+:\s*(static-brcb|static-urcb)\|dataset=([^|]+)\|" + r"rcb=([^|]+)\|bindings=(\d+) \[(ROUTED|PENDING)\]\s*$" +) +AP = re.compile(r"^SCD selected AP\s+:\s*(\S+)\s*$", re.M) +MAX_DIAGNOSTIC_TARGETS = 8 + + +class QualificationError(ValueError): + pass + + +def single(pattern: re.Pattern[str], data: str, name: str) -> re.Match[str]: + found = list(pattern.finditer(data)) + if len(found) != 1: + raise QualificationError(f"{name}: expected exactly one, found {len(found)}") + return found[0] + + +def qualify(text: str, *, ied: str, min_brcb: int = 1, + min_urcb: int = 1, selected_ap: str | None = None) -> dict: + if not ied.strip() or "\n" in ied or min_brcb < 1 or min_urcb < 1: + raise QualificationError("Exact IED and positive signal minimums required") + if not text.startswith("ARSAS Diagnostic Report"): + raise QualificationError("Full Copy Diagnostic required; emergency/partial summaries cannot qualify") + matches = list(IED.finditer(text)) + matching = [(m, i) for i, m in enumerate(matches) if m.group(1) == ied] + if len(matching) != 1: + raise QualificationError("Exact IED identity missing or ambiguous") + beginning, index = matching[0] + end = matches[index + 1].start() if index + 1 < len(matches) else len(text) + section = text[beginning.start():end] + parity = single(FINGERPRINT, section, "static semantic MATCH").group(1) + single(TRAFFIC, section, "explicit dual-ingress routed traffic proof") + if selected_ap is not None: + if not selected_ap or single(AP, section, "selected AccessPoint").group(1) != selected_ap: + raise QualificationError("SCD AccessPoint is not the explicitly requested AP") + + captures: dict[str, dict] = {} + current: str | None = None + for line in section.splitlines(): + found = INGRESS.fullmatch(line) + if found: + key, actual, fingerprint, requested, plans, brcb, urcb, uncovered = found.groups() + if key in captures or (key == "Discovery" and actual != "LiveDiscovery") or ( + key == "Open SCL" and actual != "OpenScl" + ): + raise QualificationError("Duplicated or mismatched ingress identity") + captures[key] = dict(fingerprint=fingerprint, requested=int(requested), + plans=int(plans), brcb=int(brcb), urcb=int(urcb), + uncovered=int(uncovered), targets=[], summary=None) + current = key + continue + if re.match(r"^\s{2}(?:Discovery|Open SCL)\s+:", line): + raise QualificationError("Malformed ingress header") + hit = SUMMARY.fullmatch(line) + if hit: + if current is None or captures[current]["summary"] is not None: + raise QualificationError("Missing or duplicate ingress traffic summary") + captures[current]["summary"] = (int(hit.group(1)), int(hit.group(2)), hit.group(3)) + continue + hit = TARGET.fullmatch(line) + if hit: + if current is None: + raise QualificationError("Unowned runtime RCB target") + kind, dataset, rcb, bindings, status = hit.groups() + captures[current]["targets"].append((kind, dataset, rcb, int(bindings), status)) + continue + if re.match(r"^\s{4}(?:runtime|traffic)\s+:", line): + raise QualificationError("Unparseable target or traffic line; fail closed") + + if set(captures) != {"Discovery", "Open SCL"}: + raise QualificationError("Both independent ingress summaries are required") + + details = {} + for name, entry in captures.items(): + if entry["fingerprint"] != parity or not HEX64.fullmatch(entry["fingerprint"]): + raise QualificationError(f"{name}: fingerprint does not match semantic MATCH") + if entry["uncovered"] != 0 or entry["requested"] <= 0: + raise QualificationError(f"{name}: uncovered or empty selection") + if entry["brcb"] < min_brcb or entry["urcb"] < min_urcb: + raise QualificationError(f"{name}: insufficient static BRCB/URCB signal coverage") + if entry["summary"] is None: + raise QualificationError(f"{name}: no routed-target count or first timestamp") + routed, expected, when = entry["summary"] + count = entry["plans"] + # Copy Diagnostic intentionally truncates to eight lines; never assert + # all target families are proven from a truncated or partial listing. + if not (0 < count <= MAX_DIAGNOSTIC_TARGETS and count == expected == routed == + len(entry["targets"])): + raise QualificationError(f"{name}: incomplete/truncated target inventory") + try: + time = datetime.fromisoformat(when.replace("Z", "+00:00")) + if time.tzinfo is None: + raise ValueError("timezone absent") + except ValueError as exc: + raise QualificationError(f"{name}: invalid routed-report timestamp") from exc + targets = entry["targets"] + if len({(k, ds, rc) for k, ds, rc, _, _ in targets}) != len(targets): + raise QualificationError(f"{name}: duplicate RCB target") + if any(not ds.strip() or not rc.strip() or n <= 0 or status != "ROUTED" + for _, ds, rc, n, status in targets): + raise QualificationError(f"{name}: target unbound, unrouted or empty") + if {kind for kind, _, _, _, _ in targets} != {"static-brcb", "static-urcb"}: + raise QualificationError(f"{name}: both BRCB and URCB must route independently") + details[name] = { + "plannedTargets": count, + "routedTargets": len(targets), + "brcbSignals": entry["brcb"], + "urcbSignals": entry["urcb"], + "firstRoutedUtc": time.isoformat(), + "rcbFamilies": sorted({kind for kind, _, _, _, _ in targets}), + } + return { + "ied": ied, + "semanticFingerprint": parity, + "selectedAccessPoint": selected_ap, + "qualification": "DIAGNOSTIC_CONSISTENCY_PASS_NOT_PHYSICAL_VERIFIED", + "physicalVerification": False, + "wirePcapVerification": False, + "discovery": details["Discovery"], + "openScl": details["Open SCL"], + "note": "A full current-session Copy Diagnostic is not independent PCAP/IED authority.", + } + + +def main() -> int: + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument("--diagnostic", type=Path, required=True) + parser.add_argument("--ied", required=True, help="Exact IED name from Copy Diagnostic") + parser.add_argument("--selected-ap", help="Optional exact SCD AccessPoint from the capture") + parser.add_argument("--min-brcb-signals", type=int, default=1) + parser.add_argument("--min-urcb-signals", type=int, default=1) + parser.add_argument("--output", type=Path, required=True) + args = parser.parse_args() + try: + report = qualify(args.diagnostic.read_text(encoding="utf-8-sig"), + ied=args.ied, selected_ap=args.selected_ap, + min_brcb=args.min_brcb_signals, min_urcb=args.min_urcb_signals) + args.output.parent.mkdir(parents=True, exist_ok=True) + args.output.write_text(json.dumps(report, indent=2, sort_keys=True) + "\n", + encoding="utf-8") + print("P7.6 diagnostic consistency PASS; PHYSICAL VERIFICATION NOT ESTABLISHED") + return 0 + except (OSError, UnicodeError, QualificationError) as exc: + print(f"P7.6 qualification not established: {exc}", file=sys.stderr) + return 1 + + +if __name__ == "__main__": + raise SystemExit(main())