diff --git a/.github/workflows/build.yml b/.github/workflows/build.yml index bb7aa2d90..8f0769fef 100644 --- a/.github/workflows/build.yml +++ b/.github/workflows/build.yml @@ -106,6 +106,8 @@ jobs: run: | python .\ArIED61850Tester\scripts\test-maintenance-dependencies.py if ($LASTEXITCODE -ne 0) { throw "Maintenance inventory tests failed." } + python .\ArIED61850Tester\scripts\test-verify-p10-ge-field-scl.py + if ($LASTEXITCODE -ne 0) { throw "P10 GE trusted-IID SCL field gate fixture regressions failed." } python .\ArIED61850Tester\scripts\audit-maintenance-dependencies.py --root .\ArIED61850Tester --output "$env:RUNNER_TEMP\arsas-maintenance-dependencies.json" if ($LASTEXITCODE -ne 0) { throw "Maintenance inventory generation failed." } diff --git a/docs/P10_GE_ED1_FIELD_ACCEPTANCE.md b/docs/P10_GE_ED1_FIELD_ACCEPTANCE.md new file mode 100644 index 000000000..28001f6e5 --- /dev/null +++ b/docs/P10_GE_ED1_FIELD_ACCEPTANCE.md @@ -0,0 +1,46 @@ +# P10.3/P10.4 — GE F650 Edition 1 physical-evidence procedure + +**Code authority:** ARSAS staging P10.2 commit \`2369c286da6d14dda2470a2854daf2d46d588dfe\` and engine lock \`00ad2819b99ffe6b3f885e59aa24a6afb2b4d8e1\`. No rebasing to old main and no substitution of engine PR #160. The artifact built by the exact staging [Build ARSAS #38105708427](https://github.com/masarray/arsas/actions/runs/38105708427) is the P10.3 regression-qualified **candidate**, not field-qualified release. + +## Trust and safety gates + +- Use **an independently supplied original** \`BCU_GE.iid\`, not an SCL generated by ARSAS and later renamed as engineering authority. Record its **actual file SHA-256** before opening, and preserve the source unchanged. The source and generated SCL must be separate files. +- Observe real MMS session **192.16.1.33:102** on the authorized substation network, from a qualified operator computer. Do not use direct device control operations or write data to create test traffic. +- In ARSAS, **Open SCL** original IID and select the actual \`BCUGE\` / AccessPoint. This establishes SHA-bound trusted identity. Then Discovery on the same endpoint, without clearing the source-association context. If live workflow creates a new card, ensure the trusted SCL remains bound to that card before deciding identity is high-confidence. +- Discovery must show IED \`BCUGE\`, \`LDevice.inst=F650\`, and observed *raw* MMS domain \`BCUGEF650\`. Without trusted exact SCL evidence, provisional \`IED_192_16_1_33\` is safer than heuristically guessing a device name. +- Select **Static DataSet** (not dynamic/polling), confirm 8 static DataSets, 124 ordered members, 67 selected runtime points and six concrete RCB streams with actual InformationReport traffic. GI and report-backed causality must be observed. **No cyclic MMS process-value reads** and no auto fallback. +- **Save SCL** in full-model Edition 1, then **Open SCL** the saved file and reconnect using the reloaded model; collect separate before/after diagnostics, SCL report files and PCAP (where permitted). The existing \`CanonicalSclReloadValidator\` checks internal reload and association structure; our offline check below independently compares to the original SHA-pinned IID. +- Test BRCB continuity in its **own** session evidence: timestamp, RCB, BufOvfl, SqNum/SubSqNum, EntryID, GI reason and replay after reconnect. Treat missing/ambiguous fields as incomplete, not zero faults. +- Evaluate design-only LN/DO/DA structures, observed GetNameList cap 6400 and \`incompleteChains\` separately. 8/8 DataSets is not proof of full model closure. GOOSE bindings/addresses also require separate verification. + +## Read-only offline SCL verification + +On the authorized local computer, use Python 3 already used by ARSAS CI (no additional libraries). In PowerShell: + +\`\`\`powershell +$iid = 'C:\field\BCU_GE.iid' +$export = 'C:\field\BCUGE-discovery.icd' +$sha = '<64-character SHA256 copied from original independent intake record>' +python .\scripts\verify-p10-ge-field-scl.py \` + --trusted-iid $iid \` + --trusted-sha256 $sha \` + --discovery-scl $export \` + --observed-mms-domain BCUGEF650 \` + --endpoint-ip 192.16.1.33 \` + --access-point S1 \` + --out 'C:\field\p10-ge-structural-evidence.json' +\`\`\` + +Replace \`S1\` with the **actual trusted** AccessPoint if different. Do not derive the expected trusted hash from the candidate file during the same verification run; compare with the independently recorded original intake hash. The observed-domain argument is an operator-supplied claim; save independent device logs/PCAP proving that claim. The tool hashes and checks both input files, is strictly offline, does not connect to a relay, and has no write path other than an explicitly requested JSON report. It fails closed on missing source SHA, ambiguous AP/IED, wrong IP, changed MMS wire-domain, missing LD/LN, or drift in ordered DataSet/FCDA members. It compares engineering-vs-export logical RCB and GOOSE definitions and reports them separately for review. It cannot infer unobserved MMS values, packet continuity, or physical certification. + +Possible structural outcomes: \`STRUCTURE_PASS_ONLY\`, \`FAIL\`, or \`REJECTED_INPUT\`. Even the passing case **always** has \`physicalQualification=NOT_VERIFIED\` and \`r10PromotionAllowed=false\`. It is **never** a substitute for actual IEC 61850 field evidence. + +The JSON records input SHA, structural discrepancies and explicitly lists all unverified physical gates. Do **not** modify \`evidence/interoperability-reference-target.json\` or change CI approval flags from this output. + +## Field package to attach for P10.4 review + +Preserve original IID plus original SHA256, exported ICD/IID, generated export report, full ARSAS Diagnostics from Discovery and Open SCL, packet capture (if allowed), and RCB continuity evidence with timestamps for both sessions. If PCAP contains sensitive operational data, share a redacted engineering-safe version but retain original hash/provenance in the field record. + +### Exit criteria + +Only mark P10.4 physically accepted after: verified field provenance; correct BCUGE/F650/BCUGEF650 mapping; 8/124 member match; report-only 67 runtime / 6 active RCB with real traffic; no cyclic fallback; Save→Open SCL endpoint/association round-trip; independently reviewed LN/DO/DA/name-list completeness; and independently reviewed BRCB replay/GI continuity. No claim that successful CI or this offline tool satisfies those conditions. diff --git a/scripts/test-verify-p10-ge-field-scl.py b/scripts/test-verify-p10-ge-field-scl.py new file mode 100644 index 000000000..894dbbee3 --- /dev/null +++ b/scripts/test-verify-p10-ge-field-scl.py @@ -0,0 +1,175 @@ +#!/usr/bin/env python3 +"""Offline P10.4 GE field SCL gate regression: no hardware or network required.""" +import hashlib +import importlib.util +import tempfile +import unittest +from pathlib import Path +from xml.etree import ElementTree as ET + +SCRIPT = Path(__file__).with_name("verify-p10-ge-field-scl.py") +spec = importlib.util.spec_from_file_location("p10_ge_gate", SCRIPT) +gate = importlib.util.module_from_spec(spec) +spec.loader.exec_module(gate) +N = gate.NS +T = gate.TAG + + +def example(*, ip="192.16.1.33", ied="BCUGE", inst="F650", + ap="S1", wire_ld_name=None, members=124, remove_ln=False): + ET.register_namespace("", N) + root = ET.Element(T("SCL")) + comm = ET.SubElement(root, T("Communication")) + subnet = ET.SubElement(comm, T("SubNetwork")) + conn = ET.SubElement(subnet, T("ConnectedAP"), + {"iedName": ied, "apName": ap}) + address = ET.SubElement(conn, T("Address")) + ET.SubElement(address, T("P"), {"type": "IP"}).text = ip + node = ET.SubElement(root, T("IED"), {"name": ied}) + access = ET.SubElement(node, T("AccessPoint"), {"name": ap}) + server = ET.SubElement(access, T("Server")) + attrs = {"inst": inst} + if wire_ld_name: + attrs["ldName"] = wire_ld_name + ld = ET.SubElement(server, T("LDevice"), attrs) + ln = ET.SubElement(ld, T("LN0"), {"lnClass": "LLN0"}) + if not remove_ln: + ET.SubElement(ld, T("LN"), {"lnClass": "CSWI", "inst": "7"}) + for index in range(8): + ds = ET.SubElement(ln, T("DataSet"), {"name": f"DS{index + 1}"}) + # 4*16+4*15=124: non-overlapping ordered FCDA evidence. + count = (16 if index < 4 else 15) + if members != 124 and index == 0: + count += members - 124 + for subindex in range(max(count, 0)): + ET.SubElement(ds, T("FCDA"), + {"ldInst": inst, "lnClass": "CSWI", + "lnInst": "7", "doName": f"Pos{subindex}", + "daName": "stVal", "fc": "ST"}) + if index < 6: + ET.SubElement(ln, T("ReportControl"), + {"name": f"RCB{index + 1}", "datSet": f"DS{index + 1}", + "buffered": "true"}) + return ET.tostring(root, encoding="utf-8", xml_declaration=True) + + +class GeFieldEvidenceTests(unittest.TestCase): + def setUp(self): + self.temp = tempfile.TemporaryDirectory() + self.addCleanup(self.temp.cleanup) + self.src = Path(self.temp.name, "BCU_GE.iid") + self.out = Path(self.temp.name, "BCUGE-discovery.icd") + self.src.write_bytes(example()) + self.out.write_bytes(example()) + + def check(self, sha=None, domain="BCUGEF650"): + return gate.assess( + self.src, self.out, + sha or hashlib.sha256(self.src.read_bytes()).hexdigest(), + domain, "192.16.1.33", "S1") + + def test_valid_structure_is_explicitly_not_physical_qualification(self): + result = self.check() + self.assertEqual("STRUCTURE_PASS_ONLY", result["structuralVerdict"]) + self.assertEqual(8, result["dataSets"]) + self.assertEqual(124, result["staticMembers"]) + self.assertEqual("NOT_VERIFIED", result["physicalQualification"]) + self.assertFalse(result["r10PromotionAllowed"]) + self.assertTrue(any("BRCB" in s for s in result["unverifiedPhysicalGates"])) + + def test_source_sha_drift_fails_closed(self): + with self.assertRaisesRegex(gate.RejectedEvidence, "SHA-256 mismatch"): + self.check("0" * 64) + + def test_same_scl_file_fails_closed(self): + with self.assertRaisesRegex(gate.RejectedEvidence, "same file"): + gate.assess(self.src, self.src, + hashlib.sha256(self.src.read_bytes()).hexdigest(), + "BCUGEF650", "192.16.1.33", "S1") + + def test_wrong_observed_wire_domain_fails_even_if_scl_graph_equal(self): + self.assertEqual("FAIL", self.check(domain="BCUGECTRL")["structuralVerdict"]) + + def test_wrong_endpoint_rejected(self): + self.out.write_bytes(example(ip="192.16.1.34")) + self.assertEqual("FAIL", self.check()["structuralVerdict"]) + + def test_ipless_scl_is_explicit_review_not_invented_endpoint(self): + root = ET.fromstring(example()) + root.find(f".//{T('Address')}").clear() + self.out.write_bytes(ET.tostring(root)) + result = self.check() + self.assertEqual("STRUCTURE_PASS_ONLY", result["structuralVerdict"]) + self.assertTrue(any("no authoritative ConnectedAP IP" in s for s in result["review"])) + + def test_f650_must_not_become_mms_domain_inst(self): + self.out.write_bytes(example(inst="BCUGEF650")) + self.assertEqual("FAIL", self.check()["structuralVerdict"]) + + def test_extra_explicit_ldname_is_detected(self): + self.out.write_bytes(example(wire_ld_name="OTHER_WIRE_DOMAIN")) + self.assertEqual("FAIL", self.check()["structuralVerdict"]) + + def test_missing_static_member_fails(self): + self.out.write_bytes(example(members=123)) + self.assertEqual("FAIL", self.check()["structuralVerdict"]) + + def test_member_order_change_fails_not_just_count(self): + root = ET.fromstring(example()) + ds = root.find(f".//{T('DataSet')}") + ds[0], ds[1] = ds[1], ds[0] + self.out.write_bytes(ET.tostring(root)) + self.assertEqual("FAIL", self.check()["structuralVerdict"]) + + def test_fcda_unknown_ln_rejected(self): + self.out.write_bytes(example(remove_ln=True)) + with self.assertRaisesRegex(gate.RejectedEvidence, "unknown LN"): + self.check() + + def test_no_false_rcb_continuity_from_matching_logical_rcbs(self): + result = self.check() + self.assertEqual(6, result["logicalReportControls"]["discoveryExport"]) + self.assertEqual("NOT_VERIFIED", result["physicalQualification"]) + + def test_different_rcb_configuration_is_review_not_physical_pass(self): + root = ET.fromstring(example()) + root.find(f".//{T('ReportControl')}").set("buffered", "false") + self.out.write_bytes(ET.tostring(root)) + result = self.check() + self.assertEqual("STRUCTURE_PASS_ONLY", result["structuralVerdict"]) + self.assertTrue(any("Logical ReportControl" in s for s in result["review"])) + self.assertFalse(result["r10PromotionAllowed"]) + + def test_source_extra_design_ln_is_review_not_dataset_failure(self): + root = ET.fromstring(example()) + ld = root.find(f".//{T('LDevice')}") + ET.SubElement(ld, T("LN"), {"lnClass": "MMXU", "inst": "1"}) + self.src.write_bytes(ET.tostring(root)) + result = self.check() + self.assertEqual("STRUCTURE_PASS_ONLY", result["structuralVerdict"]) + self.assertTrue(any("LN design/live scope differs" in s for s in result["review"])) + + def test_external_ied_fcda_is_not_silently_treated_as_local(self): + root = ET.fromstring(example()) + root.find(f".//{T('FCDA')}").set("iedName", "DIFFERENT_IED") + self.out.write_bytes(ET.tostring(root)) + with self.assertRaisesRegex(gate.RejectedEvidence, "external IED"): + self.check() + + def test_dtd_and_entities_rejected(self): + evil = b']>' + self.out.write_bytes(evil) + with self.assertRaisesRegex(gate.RejectedEvidence, "DTD/ENTITY"): + self.check() + + def test_ambiguous_accesspoint_rejected(self): + root = ET.fromstring(example()) + access = root.find(f".//{T('AccessPoint')}") + root.find(f".//{T('IED')}").append(ET.fromstring(ET.tostring(access))) + self.out.write_bytes(ET.tostring(root)) + with self.assertRaisesRegex(gate.RejectedEvidence, "AccessPoint"): + self.check() + + +if __name__ == "__main__": + unittest.main() diff --git a/scripts/verify-p10-ge-field-scl.py b/scripts/verify-p10-ge-field-scl.py new file mode 100644 index 000000000..91ba07f3a --- /dev/null +++ b/scripts/verify-p10-ge-field-scl.py @@ -0,0 +1,297 @@ +#!/usr/bin/env python3 +"""P10.4 GE Edition 1 SCL structural evidence gate (offline, read-only). + +Not a physical IEC 61850 qualification: an operator must separately provide +association/PCAP, InformationReport, GI, continuity and semantic evidence. +""" +from __future__ import annotations + +import argparse +import hashlib +import json +import re +import sys +from collections import Counter +from pathlib import Path +from xml.etree import ElementTree as ET + +NS = "http://www.iec.ch/61850/2003/SCL" +TAG = lambda local: f"{{{NS}}}{local}" +MAX_BYTES = 16 * 1024 * 1024 +HEX_SHA = re.compile(r"^[0-9a-fA-F]{64}$") + + +class RejectedEvidence(ValueError): + """A malformed or unauthoritative field input.""" + + +def read_bounded(path: Path) -> bytes: + size = path.stat().st_size + if size <= 0 or size > MAX_BYTES: + raise RejectedEvidence(f"{path.name}: file size outside (0, {MAX_BYTES}] bytes") + data = path.read_bytes() + if len(data) != size: + raise RejectedEvidence(f"{path.name}: file changed during read") + return data + + +def scl_tree(data: bytes, label: str) -> ET.Element: + # Do not allow entities/DTD or silently accept namespace-free pseudo SCL. + if re.search(rb" str: + return (element.get(name) or "").strip() + + +def ied(root: ET.Element, name: str, label: str) -> ET.Element: + candidates = [x for x in root.findall(TAG("IED")) if attr(x, "name") == name] + return only(candidates, f"{label}: IED '{name}'") + + +def access_point(ied_node: ET.Element, name: str, label: str) -> ET.Element: + return only([x for x in ied_node.findall(TAG("AccessPoint")) + if attr(x, "name") == name], f"{label}: AccessPoint '{name}'") + + +def endpoint(root: ET.Element, name: str, ap: str, label: str) -> str | None: + comm = root.find(TAG("Communication")) + if comm is None: + return None + matches = [node for node in comm.iter(TAG("ConnectedAP")) + if attr(node, "iedName") == name and attr(node, "apName") == ap] + if len(matches) > 1: + raise RejectedEvidence(f"{label}: ambiguous ConnectedAP '{name}/{ap}'") + if not matches: + return None + addr = matches[0].find(TAG("Address")) + if addr is None: + return None + ips = [str(p.text or "").strip() for p in addr.findall(TAG("P")) + if attr(p, "type").upper() == "IP"] + if len(ips) > 1: + raise RejectedEvidence(f"{label}: multiple IP entries") + return ips[0] if ips else None + + +def snapshot(root: ET.Element, name: str, ap_name: str, label: str) -> dict: + own_ied = ied(root, name, label) + own_ap = access_point(own_ied, ap_name, label) + server = only(own_ap.findall(TAG("Server")), f"{label}: Server") + ld_by_inst = {} + domains = {} + logical_nodes = set() + datasets = {} + reports = [] + goose = [] + + for ld in server.findall(TAG("LDevice")): + inst = attr(ld, "inst") + if not inst or inst in ld_by_inst: + raise RejectedEvidence(f"{label}: missing/duplicate LDevice.inst '{inst}'") + ld_by_inst[inst] = ld + domain = attr(ld, "ldName") or name + inst + if domain in domains: + raise RejectedEvidence(f"{label}: duplicate resolved MMS domain '{domain}'") + domains[domain] = inst + for ln in list(ld): + if ln.tag not in (TAG("LN0"), TAG("LN")): + continue + ln_class = attr(ln, "lnClass") or ("LLN0" if ln.tag == TAG("LN0") else "") + prefix, ln_inst = attr(ln, "prefix"), attr(ln, "inst") + node_key = (inst, prefix, ln_class, ln_inst) + if not ln_class or node_key in logical_nodes: + raise RejectedEvidence(f"{label}: invalid/duplicate logical node {node_key}") + logical_nodes.add(node_key) + for ds in ln.findall(TAG("DataSet")): + ds_name = attr(ds, "name") + key = (inst, prefix, ln_class, ln_inst, ds_name) + if not ds_name or key in datasets: + raise RejectedEvidence(f"{label}: duplicate/empty DataSet {key}") + members = [] + for fcda in ds.findall(TAG("FCDA")): + target_ld = attr(fcda, "ldInst") or inst + fcda_ied = attr(fcda, "iedName") + if fcda_ied and fcda_ied != name: + raise RejectedEvidence(f"{label}: FCDA targets external IED '{fcda_ied}'") + fc = attr(fcda, "fc") + do = attr(fcda, "doName") + ln_cl = attr(fcda, "lnClass") + if not target_ld or not ln_cl or not do or not fc: + raise RejectedEvidence(f"{label}: incomplete FCDA in {key}") + members.append((target_ld, attr(fcda, "prefix"), ln_cl, + attr(fcda, "lnInst"), do, + attr(fcda, "daName"), fc, attr(fcda, "ix"))) + if not members: + raise RejectedEvidence(f"{label}: empty DataSet {key}") + datasets[key] = members + for rc in ln.findall(TAG("ReportControl")): + rc_name = attr(rc, "name") + ds = attr(rc, "datSet") + if not rc_name or not ds: + raise RejectedEvidence(f"{label}: ReportControl missing name/datSet") + key = (inst, prefix, ln_class, ln_inst, ds) + if key not in datasets: + raise RejectedEvidence(f"{label}: RCB '{rc_name}' binds missing DataSet {key}") + reports.append((inst, prefix, ln_class, ln_inst, rc_name, ds, + attr(rc, "buffered"), attr(rc, "indexed"))) + for cb in ln.findall(TAG("GSEControl")): + goose.append((inst, prefix, ln_class, ln_inst, + attr(cb, "name"), attr(cb, "datSet"))) + + if not ld_by_inst: + raise RejectedEvidence(f"{label}: missing LDevice") + for key, members in datasets.items(): + for member in members: + if member[0] not in ld_by_inst: + raise RejectedEvidence(f"{label}: FCDA targets unknown LD '{member[0]}'") + if member[:4] not in logical_nodes: + raise RejectedEvidence(f"{label}: FCDA targets unknown LN {member[:4]}") + return { + "domains": domains, + "logicalNodes": logical_nodes, + "datasets": datasets, + "reports": reports, + "goose": goose, + "endpoint": endpoint(root, name, ap_name, label), + } + + +def readable_key(key: tuple) -> str: + return "/".join(str(x) for x in key) + + +def compare(trusted: dict, saved: dict, domain: str, ip: str, + expected_datasets: int, expected_members: int) -> dict: + failures, review = [], [] + for label, snap in (("trusted", trusted), ("discovery export", saved)): + if snap["domains"] != {domain: "F650"}: + failures.append(f"{label}: MMS wire-domain→LD mapping differs from BCUGEF650→F650") + if snap["endpoint"] is not None and snap["endpoint"] != ip: + failures.append(f"{label}: SCL declared IP '{snap['endpoint']}' differs from '{ip}'") + if snap["endpoint"] is None: + review.append(f"{label}: no authoritative ConnectedAP IP; endpoint must be checked from field capture") + if len(snap["datasets"]) != expected_datasets: + failures.append(f"{label}: DataSets={len(snap['datasets'])}; expected {expected_datasets}") + count = sum(len(m) for m in snap["datasets"].values()) + if count != expected_members: + failures.append(f"{label}: FCDA members={count}; expected {expected_members}") + + if trusted["datasets"] != saved["datasets"]: + missing = trusted["datasets"].keys() - saved["datasets"].keys() + extra = saved["datasets"].keys() - trusted["datasets"].keys() + changed = [key for key in trusted["datasets"].keys() & saved["datasets"].keys() + if trusted["datasets"][key] != saved["datasets"][key]] + failures.append(f"DataSet FCDA ordered semantic mismatch: missing={len(missing)}, " + f"extra={len(extra)}, changed={len(changed)}") + review.extend("DataSet mismatch: " + readable_key(key) + for key in list(sorted(missing | extra | set(changed)))[:12]) + + # Different design/offline vs discoverable live scope is NOT made invisible; + # this is a separate engineering review, not a DataSet/reporting pass. + missing_ln = trusted["logicalNodes"] - saved["logicalNodes"] + added_ln = saved["logicalNodes"] - trusted["logicalNodes"] + if missing_ln or added_ln: + review.append(f"LN design/live scope differs: trusted-only={len(missing_ln)}, " + f"export-only={len(added_ln)}; DO/DA semantic parity NOT VERIFIED") + if Counter(trusted["reports"]) != Counter(saved["reports"]): + review.append("Logical ReportControl configuration differs from trusted IID; " + "RCB semantic parity and buffer continuity NOT VERIFIED") + if Counter(trusted["goose"]) != Counter(saved["goose"]): + review.append("GOOSE control set differs; GOOSE address/cb reference parity NOT VERIFIED") + + return {"structuralVerdict": "FAIL" if failures else "STRUCTURE_PASS_ONLY", + "failures": failures, "review": review, + "dataSets": len(saved["datasets"]), + "staticMembers": sum(len(m) for m in saved["datasets"].values()), + "logicalNodes": {"trusted": len(trusted["logicalNodes"]), + "discoveryExport": len(saved["logicalNodes"])}, + "logicalReportControls": {"trusted": len(trusted["reports"]), + "discoveryExport": len(saved["reports"])}} + + +def assess(trusted_file: Path, generated_file: Path, source_sha: str, + domain: str, ip: str, ap_name: str, datasets: int = 8, + members: int = 124) -> dict: + if not HEX_SHA.fullmatch(source_sha): + raise RejectedEvidence("trusted source SHA-256 must be exactly 64 hex characters") + if trusted_file.resolve() == generated_file.resolve(): + raise RejectedEvidence("source IID and Discovery-export SCL cannot be the same file") + origin = read_bounded(trusted_file) + actual = hashlib.sha256(origin).hexdigest() + if actual.lower() != source_sha.lower(): + raise RejectedEvidence("trusted engineering IID SHA-256 mismatch; fail closed") + output = read_bounded(generated_file) + t = snapshot(scl_tree(origin, "trusted IID"), "BCUGE", ap_name, "trusted IID") + g = snapshot(scl_tree(output, "Discovery SCL"), "BCUGE", ap_name, "Discovery SCL") + result = compare(t, g, domain, ip, datasets, members) + return { + "schema": "arsas-p10-ge-field-check-v1", + "source": {"sha256": actual, "bytes": len(origin)}, + "discoveryExport": {"sha256": hashlib.sha256(output).hexdigest(), "bytes": len(output)}, + "expectedEndpoint": f"{ip}:102", + "observedDomainClaim": domain, + "identity": "BCUGE", + "logicalDevice": "F650", + **result, + "unverifiedPhysicalGates": [ + "actual MMS association/device provenance and raw domain name-list", + "67/67 runtime points and six concrete static RCBs with InformationReport traffic", + "zero cyclic MMS process-value polling and valid GI causality", + "LN/DO/DA semantic closure, 6400 name-list cap and incomplete chains", + "BRCB BufOvfl/sequence/EntryID replay continuity across sessions", + "GOOSE address and wire subscriber semantics", + ], + "physicalQualification": "NOT_VERIFIED", + "r10PromotionAllowed": False, + } + + +def main(argv: list[str] | None = None) -> int: + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument("--trusted-iid", type=Path, required=True) + parser.add_argument("--trusted-sha256", required=True) + parser.add_argument("--discovery-scl", type=Path, required=True) + parser.add_argument("--observed-mms-domain", required=True) + parser.add_argument("--endpoint-ip", default="192.16.1.33") + parser.add_argument("--access-point", required=True) + parser.add_argument("--out", type=Path, help="Optional JSON report path") + args = parser.parse_args(argv) + try: + result = assess(args.trusted_iid, args.discovery_scl, + args.trusted_sha256, args.observed_mms_domain, + args.endpoint_ip, args.access_point) + code = 2 if result["structuralVerdict"] == "FAIL" else 0 + except (OSError, RejectedEvidence) as exc: + result = {"schema": "arsas-p10-ge-field-check-v1", + "structuralVerdict": "REJECTED_INPUT", + "reason": str(exc), "physicalQualification": "NOT_VERIFIED", + "r10PromotionAllowed": False} + code = 2 + report = json.dumps(result, ensure_ascii=False, indent=2, sort_keys=True) + print(report) + if args.out is not None: + try: + args.out.write_text(report + "\n", encoding="utf-8") + except OSError as exc: + print(f"Could not write report: {exc}", file=sys.stderr) + return 2 + return code + + +if __name__ == "__main__": + sys.exit(main())