From fc8712c30b01480ef58cc59fe344bbb15dabb523 Mon Sep 17 00:00:00 2001 From: Ari Sulistiono Date: Tue, 27 Oct 2026 06:23:22 +0700 Subject: [PATCH 1/2] field(P7): restore bounded dual-ingress diagnostic intake with fail-closed evidence --- .../workflows/ci-p0-workflow-integrity.yml | 7 + .../audits/P7_6_FIELD_QUALIFICATION_INTAKE.md | 43 ++++ scripts/test-p7-dual-ingress-diagnostic.py | 186 +++++++++++++++ scripts/verify-p7-dual-ingress-diagnostic.py | 224 ++++++++++++++++++ 4 files changed, 460 insertions(+) create mode 100644 docs/audits/P7_6_FIELD_QUALIFICATION_INTAKE.md create mode 100644 scripts/test-p7-dual-ingress-diagnostic.py create mode 100644 scripts/verify-p7-dual-ingress-diagnostic.py diff --git a/.github/workflows/ci-p0-workflow-integrity.yml b/.github/workflows/ci-p0-workflow-integrity.yml index f12f055ae..14af8bf9c 100644 --- a/.github/workflows/ci-p0-workflow-integrity.yml +++ b/.github/workflows/ci-p0-workflow-integrity.yml @@ -7,6 +7,8 @@ on: - ".github/workflows/**" - "scripts/audit-ci-workflows.py" - "scripts/test-ci-workflows.py" + - "scripts/verify-p7-dual-ingress-diagnostic.py" + - "scripts/test-p7-dual-ingress-diagnostic.py" - "scripts/classify-ci-changes.py" - "scripts/test-classify-ci-changes.py" - "scripts/write-ci-canonical-proof.py" @@ -32,6 +34,8 @@ on: - ".github/workflows/**" - "scripts/audit-ci-workflows.py" - "scripts/test-ci-workflows.py" + - "scripts/verify-p7-dual-ingress-diagnostic.py" + - "scripts/test-p7-dual-ingress-diagnostic.py" - "scripts/classify-ci-changes.py" - "scripts/test-classify-ci-changes.py" - "scripts/write-ci-canonical-proof.py" @@ -78,6 +82,9 @@ jobs: - name: Run offline CI workflow guard tests run: python3 scripts/test-ci-workflows.py + - name: Validate sanitized P7.6 dual-ingress diagnostic parser (read-only) + run: python3 scripts/test-p7-dual-ingress-diagnostic.py + - name: Run conservative CI P1 classifier fixtures run: python3 scripts/test-classify-ci-changes.py 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..041d13162 --- /dev/null +++ b/docs/audits/P7_6_FIELD_QUALIFICATION_INTAKE.md @@ -0,0 +1,43 @@ +# P7.6 — Read-only dual-ingress field diagnostic intake (draft) + +Parent: issue #446. D22 restores the old #479 diagnostic intake as an isolated utility on integrated RC; historical #476 is already retired. 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. + +## D22 RC-aligned hardening + +The validator is matched against the current native `DiagnosticReportBuilder.AppendStaticIngressParity` and `StaticAcquisitionIngressEvidence` output. It requires one explicitly selected SCD AccessPoint, both ingress families, exact semantic SHA, exact per-target ROUTED evidence, and a complete bounded target inventory. An IED with more than eight planned targets cannot qualify from truncated Copy Diagnostic text. + +Read-only text is capped at 8 MiB. Results use atomic replacement; a failed recheck overwrites any earlier PASS with `qualification=NOT_ESTABLISHED` and false physical/PCAP flags. Do not upload private real IED reports to GitHub. No network, MMS write or physical authority is introduced. + +## 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..444f59a49 --- /dev/null +++ b/scripts/test-p7-dual-ingress-diagnostic.py @@ -0,0 +1,186 @@ +#!/usr/bin/env python3 +"""Sanitized P7.6 fail-closed qualification fixtures; no device/network access.""" +from __future__ import annotations + +import importlib.util +import json +from pathlib import Path +import subprocess +import sys +import tempfile +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") + + + def test_selected_ap_required_and_proven_even_when_optional_arg_omitted(self): + verdict = qualifier.qualify(diagnostic(), ied="GR_X_7SX85") + self.assertEqual("F", verdict["selectedAccessPoint"]) + self.assertNotQualified(diagnostic().replace("SCD selected AP : F\n", ""), + "selected AccessPoint") + self.assertNotQualified(diagnostic().replace("SCD selected AP : F", + "SCD selected AP : pending"), "not a proven selected AP") + self.assertNotQualified(diagnostic().replace("SCD selected AP : F", + "SCD selected AP : F\nSCD selected AP : J"), "exactly one") + + def test_large_or_nul_diagnostic_fail_closed(self): + self.assertNotQualified("ARSAS Diagnostic Report\n" + + "X" * (qualifier.MAX_DIAGNOSTIC_BYTES + 1), "exceeds safe size") + self.assertNotQualified(diagnostic() + "\x00", "contains NUL") + + def test_cli_failed_recheck_replaces_old_positive_verdict(self): + with tempfile.TemporaryDirectory() as tmp: + root = Path(tmp) + incoming, result = root / "private.txt", root / "result.json" + cmd = [sys.executable, str(SCRIPT), "--diagnostic", str(incoming), + "--ied", "GR_X_7SX85", "--selected-ap", "F", + "--min-brcb-signals", "6", "--min-urcb-signals", "6", + "--output", str(result)] + incoming.write_text(diagnostic(), encoding="utf-8") + self.assertEqual(0, subprocess.run(cmd, capture_output=True).returncode) + self.assertEqual("DIAGNOSTIC_CONSISTENCY_PASS_NOT_PHYSICAL_VERIFIED", + json.loads(result.read_text())["qualification"]) + incoming.write_text(diagnostic(traffic=False), encoding="utf-8") + self.assertNotEqual(0, subprocess.run(cmd, capture_output=True).returncode) + failed = json.loads(result.read_text()) + self.assertEqual("NOT_ESTABLISHED", failed["qualification"]) + self.assertFalse(failed["physicalVerification"]) + self.assertFalse(failed["wirePcapVerification"]) + self.assertFalse(list(root.glob(".result.json.*.tmp"))) + + def test_bounded_file_and_result_symlink_fail_closed(self): + with tempfile.TemporaryDirectory() as tmp: + root = Path(tmp) + f = root / "huge.txt" + f.write_bytes(b"Z" * (qualifier.MAX_DIAGNOSTIC_BYTES + 1)) + with self.assertRaisesRegex(qualifier.QualificationError, "exceeds safe size"): + qualifier.read_bounded_diagnostic(f) + destination, link = root / "other.json", root / "symlink.json" + destination.write_text('{"qualification":"KEEP"}') + try: + link.symlink_to(destination) + except (OSError, NotImplementedError): + self.skipTest("Symlinks unavailable") + with self.assertRaisesRegex(OSError, "Refusing symlink"): + qualifier.write_result(link, {"qualification": "PASS"}) + self.assertEqual('{"qualification":"KEEP"}', destination.read_text()) + +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..710a4bb5d --- /dev/null +++ b/scripts/verify-p7-dual-ingress-diagnostic.py @@ -0,0 +1,224 @@ +#!/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 +import os +from pathlib import Path +import re +import sys +import tempfile + +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 +MAX_DIAGNOSTIC_BYTES = 8 * 1024 * 1024 + + +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 len(text.encode("utf-8")) > MAX_DIAGNOSTIC_BYTES or "\x00" in text: + raise QualificationError("Diagnostic exceeds safe size or contains NUL data") + 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") + captured_ap = single(AP, section, "selected AccessPoint").group(1) + if captured_ap.lower() in {"none", "unknown", "pending", "n/a", "null", "-", "—"}: + raise QualificationError("SCD AccessPoint is not a proven selected AP") + if selected_ap is not None and (not selected_ap or selected_ap != captured_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": captured_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 write_result(output: Path, result: dict) -> None: + """Atomic result replacement: no partial JSON nor stale PASS on failure.""" + output.parent.mkdir(parents=True, exist_ok=True) + if output.is_symlink(): + raise OSError("Refusing symlink result path") + staging: Path | None = None + try: + with tempfile.NamedTemporaryFile( + mode="w", encoding="utf-8", prefix=f".{output.name}.", + suffix=".tmp", dir=output.parent, delete=False + ) as handle: + staging = Path(handle.name) + json.dump(result, handle, indent=2, sort_keys=True) + handle.write("\n") + handle.flush() + os.fsync(handle.fileno()) + staging.replace(output) + finally: + if staging is not None: + staging.unlink(missing_ok=True) + + +def read_bounded_diagnostic(path: Path) -> str: + if path.stat().st_size > MAX_DIAGNOSTIC_BYTES: + raise QualificationError("Diagnostic exceeds safe size or contains NUL data") + raw = path.read_bytes() + if len(raw) > MAX_DIAGNOSTIC_BYTES: + raise QualificationError("Diagnostic exceeds safe size or contains NUL data") + return raw.decode("utf-8-sig") + + +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(read_bounded_diagnostic(args.diagnostic), + ied=args.ied, selected_ap=args.selected_ap, + min_brcb=args.min_brcb_signals, min_urcb=args.min_urcb_signals) + except (OSError, UnicodeError, QualificationError) as exc: + print(f"P7.6 qualification not established: {exc}", file=sys.stderr) + try: + write_result(args.output, { + "qualification": "NOT_ESTABLISHED", + "physicalVerification": False, "wirePcapVerification": False, + "reason": str(exc), + }) + except OSError as output_error: + print(f"P7.6 failure result could not be saved: {output_error}", + file=sys.stderr) + return 1 + try: + write_result(args.output, report) + except OSError as exc: + print(f"P7.6 result could not be saved: {exc}", file=sys.stderr) + return 1 + print("P7.6 diagnostic consistency PASS; PHYSICAL VERIFICATION NOT ESTABLISHED") + return 0 + +if __name__ == "__main__": + raise SystemExit(main()) From 43b7eed442713194f76ea115ce3bf5cecdc6f5cf Mon Sep 17 00:00:00 2001 From: Ari Sulistiono Date: Tue, 27 Oct 2026 07:33:28 +0700 Subject: [PATCH 2/2] fix(p7.6): verify RCB binding totals and fail closed on malformed numeric evidence --- .../audits/P7_6_FIELD_QUALIFICATION_INTAKE.md | 8 ++- scripts/test-p7-dual-ingress-diagnostic.py | 65 ++++++++++++++++++- scripts/verify-p7-dual-ingress-diagnostic.py | 17 +++-- 3 files changed, 81 insertions(+), 9 deletions(-) diff --git a/docs/audits/P7_6_FIELD_QUALIFICATION_INTAKE.md b/docs/audits/P7_6_FIELD_QUALIFICATION_INTAKE.md index 041d13162..df7b441d6 100644 --- a/docs/audits/P7_6_FIELD_QUALIFICATION_INTAKE.md +++ b/docs/audits/P7_6_FIELD_QUALIFICATION_INTAKE.md @@ -16,6 +16,10 @@ The validator is matched against the current native `DiagnosticReportBuilder.App Read-only text is capped at 8 MiB. Results use atomic replacement; a failed recheck overwrites any earlier PASS with `qualification=NOT_ESTABLISHED` and false physical/PCAP flags. Do not upload private real IED reports to GitHub. No network, MMS write or physical authority is introduced. +## D24 — Source-consistency hardening + +Before passing a local copy, the validator now reconciles **each ingress** per-family BRCB/URCB `bindings=` totals with `staticBRCB`/`staticURCB` counters, matching `StaticAcquisitionParityTracker.BuildEvidence` rather than trusting either number in isolation. Binding counts are incidences; they must **not** be equated to distinct requested IEC points because DataSets/RCBs may overlap. Every numeric diagnostic field is bounded to ASCII digits of at most ten characters before `int()`, so malformed 5,000-digit values fail cleanly and cannot leave a stale PASS output. Negative tests cover both ingress families, long header/traffic/binding numbers, non-ASCII digits, and CLI PASS→NOT_ESTABLISHED replacement. These guards add no IED, network, protocol or physical authority. + ## 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. @@ -36,8 +40,8 @@ Read-only text is capped at 8 MiB. Results use atomic replacement; a failed rech ## 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. +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 now has **24 offline unittest methods** with sanitized positive/negative fixtures, including D24 review regressions. The existing Ubuntu `CI P0 Workflow Integrity` runs the tests; no new workflow or additional native/.NET build was introduced by this validator. Exact-head GitHub Actions evidence is still required after the D24 source change. -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. +Implementation is isolated in [PR #759](https://github.com/masarray/arsas/pull/759), based on the protected integrated RC. Owner review, required-check visibility and physical IEC 61850 acceptance remain separate. Do not merge into the RC/main or promote stable based only on text consistency or PR-green CI. 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 index 444f59a49..7333d6a5f 100644 --- a/scripts/test-p7-dual-ingress-diagnostic.py +++ b/scripts/test-p7-dual-ingress-diagnostic.py @@ -22,13 +22,14 @@ 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: + first: str = STAMP, brcb_bindings: int = 6, urcb_bindings: int = 6, + 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" + f" runtime : static-brcb|dataset=ied/lln0.events|rcb=ied/lln0.br.rpt_ind01|bindings={brcb_bindings} [ROUTED]\n" + f" runtime : static-urcb|dataset=ied/lln0.analog|rcb=ied/lln0.rp.rpt_meas01|bindings={urcb_bindings} [{last_status}]\n" ) @@ -140,6 +141,64 @@ def test_selected_ap_required_and_proven_even_when_optional_arg_omitted(self): self.assertNotQualified(diagnostic().replace("SCD selected AP : F", "SCD selected AP : F\nSCD selected AP : J"), "exactly one") + def test_each_ingress_requires_exact_per_family_binding_incidence(self): + # Summary counters cannot substitute for actual target totals. + for ingress_name, label, source in ( + ("discovery", "Discovery", "LiveDiscovery"), + ("scl", "Open SCL ", "OpenScl"), + ): + for parameter in ("brcb_bindings", "urcb_bindings"): + with self.subTest(ingress=ingress_name, family=parameter): + changed = ingress(label, source, **{parameter: 1}) + self.assertNotQualified(diagnostic(**{ingress_name: changed}), + "target bindings do not match") + + def test_binding_incidence_can_exceed_unique_requested_signal_count(self): + # Reused points across RCBs count as multiple binding incidences. + result = qualifier.qualify(diagnostic().replace("requested=12", "requested=9"), + ied="GR_X_7SX85", min_brcb=6, min_urcb=6) + self.assertEqual(6, result["openScl"]["brcbSignals"]) + self.assertEqual(6, result["discovery"]["urcbSignals"]) + + def test_overlong_numeric_fields_and_unicode_digits_fail_closed(self): + many = "9" * 5000 + cases = ( + ("ingress", diagnostic().replace("requested=12", "requested=" + many, 1), + "Malformed ingress header"), + ("traffic", diagnostic().replace("traffic : 2/2", + "traffic : " + many + "/2", 1), + "Unparseable target or traffic"), + ("binding", diagnostic().replace("bindings=6 [ROUTED]", + "bindings=" + many + " [ROUTED]", 1), + "Unparseable target or traffic"), + ("unicode", diagnostic().replace("requested=12", "requested=１２", 1), + "Malformed ingress header"), + ) + for name, bad, error in cases: + with self.subTest(field=name): + self.assertNotQualified(bad, error) + + def test_cli_long_numeric_recheck_cannot_leave_stale_pass(self): + with tempfile.TemporaryDirectory() as tmp: + root = Path(tmp) + incoming, result = root / "private.txt", root / "verdict.json" + cmd = [sys.executable, str(SCRIPT), "--diagnostic", str(incoming), + "--ied", "GR_X_7SX85", "--output", str(result)] + incoming.write_text(diagnostic(), encoding="utf-8") + self.assertEqual(0, subprocess.run(cmd, capture_output=True).returncode) + self.assertIn("PASS", json.loads(result.read_text())["qualification"]) + incoming.write_text( + diagnostic().replace("requested=12", "requested=" + "9" * 5000, 1), + encoding="utf-8") + rejected = subprocess.run(cmd, capture_output=True, text=True) + self.assertNotEqual(0, rejected.returncode) + self.assertNotIn("Traceback", rejected.stderr) + saved = json.loads(result.read_text()) + self.assertEqual("NOT_ESTABLISHED", saved["qualification"]) + self.assertFalse(saved["physicalVerification"]) + self.assertFalse(saved["wirePcapVerification"]) + self.assertFalse(list(root.glob(".verdict.json.*.tmp"))) + def test_large_or_nul_diagnostic_fail_closed(self): self.assertNotQualified("ARSAS Diagnostic Report\n" + "X" * (qualifier.MAX_DIAGNOSTIC_BYTES + 1), "exceeds safe size") diff --git a/scripts/verify-p7-dual-ingress-diagnostic.py b/scripts/verify-p7-dual-ingress-diagnostic.py index 710a4bb5d..793de3cc8 100644 --- a/scripts/verify-p7-dual-ingress-diagnostic.py +++ b/scripts/verify-p7-dual-ingress-diagnostic.py @@ -19,15 +19,17 @@ 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) +# Native C# counters are nonnegative 32-bit decimals; cap digit widths before int(). +# This prevents malformed integer input from bypassing NOT_ESTABLISHED output. 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*$" + r"fingerprint=([0-9a-f]{64}), requested=([0-9]{1,10}), plans=([0-9]{1,10}), " + r"staticBRCB=([0-9]{1,10}), staticURCB=([0-9]{1,10}), uncovered=([0-9]{1,10})\s*$" ) -SUMMARY = re.compile(r"^\s{4}traffic\s+:\s*(\d+)/(\d+) exact static RCB target\(s\) routed; first=(\S+)\s*$") +SUMMARY = re.compile(r"^\s{4}traffic\s+:\s*([0-9]{1,10})/([0-9]{1,10}) 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*$" + r"rcb=([^|]+)\|bindings=([0-9]{1,10}) \[(ROUTED|PENDING)\]\s*$" ) AP = re.compile(r"^SCD selected AP\s+:\s*(\S+)\s*$", re.M) MAX_DIAGNOSTIC_TARGETS = 8 @@ -135,6 +137,13 @@ def qualify(text: str, *, ied: str, min_brcb: int = 1, 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") + # Native BuildEvidence measures binding incidences for each RCB family, + # not the number of distinct selected IEC reference + FC points. + for family, reported in (("static-brcb", entry["brcb"]), + ("static-urcb", entry["urcb"])): + bound = sum(n for kind, _, _, n, _ in targets if kind == family) + if bound != reported: + raise QualificationError(f"{name}: {family} target bindings do not match signal coverage") details[name] = { "plannedTargets": count, "routedTargets": len(targets),