From bdaf84c1fe333eb88f0a025d290a2f0bbd0a52d2 Mon Sep 17 00:00:00 2001 From: Ari Sulistiono Date: Sat, 17 Oct 2026 15:25:28 +0700 Subject: [PATCH 1/2] fix(ptp-m3e): reject stale SOURCE readiness and incomplete boot evidence --- docs/PTP_M3E_BENCH_EVIDENCE_GATE.md | 20 ++++++++++- scripts/ptp_m3e_bench_gate.py | 53 ++++++++++++++++++++++------- tests/test_ptp_m3e_bench_gate.py | 37 ++++++++++++++++++++ 3 files changed, 96 insertions(+), 14 deletions(-) diff --git a/docs/PTP_M3E_BENCH_EVIDENCE_GATE.md b/docs/PTP_M3E_BENCH_EVIDENCE_GATE.md index a99ce5baa..1234c6d0a 100644 --- a/docs/PTP_M3E_BENCH_EVIDENCE_GATE.md +++ b/docs/PTP_M3E_BENCH_EVIDENCE_GATE.md @@ -111,4 +111,22 @@ and canonical PCAP continuity. All hardware qualification fields remain All M3E.2 changes are Python host-tool regression/doc additions. The proven M3D.2 SOURCE/RECEIVER firmware binaries and native ARStack Studio are not -rebuilt or modified by this checkpoint. \ No newline at end of file +rebuilt or modified by this checkpoint. +## M3E.3 — SOURCE readiness freshness / boot provenance regression (2026-10-17) + +A previous host gate accepted **any** historical `PTP status=RUNNING` +sample with positive Announce/Sync/FollowUp as SOURCE readiness. This could +return `operationalLifecycle.status=OBSERVED` even when a later STOP, rejected +START, unavailable status, or TX fault was already present in the same raw log. + +The M3E.3 host-only gate now requires the **latest** SOURCE status to be +RUNNING with complete native firmware counters and `TXfail=0`. It revokes +readiness on `PTP stop requested`, `PTP start rejected`, STOPPED, truncated +SHOW, or unavailable status. Any nonzero reported PTP TX failure in the same +boot is conservatively latched as a veto. Every IDENTIFY line must also carry +the same 16-hex-digit boot ID; an identity without a boot ID cannot inherit +a preceding measurement. Receiver logic and wire decoding are unchanged. + +This is regression protection for **firmware-reported** software evidence, +not proof of wire transmission, EMAC phase alignment, IED lock, or public +release readiness. No firmware images or Studio workflow have changed. diff --git a/scripts/ptp_m3e_bench_gate.py b/scripts/ptp_m3e_bench_gate.py index d79222741..4e9b36582 100644 --- a/scripts/ptp_m3e_bench_gate.py +++ b/scripts/ptp_m3e_bench_gate.py @@ -27,9 +27,11 @@ SCENARIOS = ("auto", "receiver-acquisition", "receiver-recovery", "receiver-gm-handoff", "source-ready") BOOT_ID = re.compile(r"\bboot_id=([a-fA-F0-9]{16})\b") -SOURCE_READY = re.compile( - r"\bPTP status=RUNNING Announce=([0-9]+) Sync=([0-9]+) " - r"FollowUp=([0-9]+)\b") +SOURCE_STATUS = re.compile( + r"\bPTP status=(RUNNING|STOPPED) Announce=([0-9]+) Sync=([0-9]+) " + r"FollowUp=([0-9]+) PdelayFrames=([0-9]+) TXfail=([0-9]+)\b") +SOURCE_INVALIDATION = re.compile( + r"\bPTP (?:stop requested|start rejected|status unavailable)\b") STATUS_FIELD = re.compile(r"\b([A-Za-z][A-Za-z0-9]*)=([^\s]+)") class EvidenceError(ValueError): @@ -203,24 +205,49 @@ def operational_evidence(data: bytes, role: str, scenario: str) -> dict: require((role == "source") == (scenario == "source-ready"), "scenario incompatible with firmware role") lines = data.decode("utf-8", errors="replace").splitlines() - boot_ids = [m.group(1).lower() for line in lines - if IDENTITY.search(line) and (m := BOOT_ID.search(line))] - if len(set(boot_ids)) != 1 or not boot_ids: + identity_lines = [line for line in lines if IDENTITY.search(line)] + boot_ids = [m.group(1).lower() for line in identity_lines + if (m := BOOT_ID.search(line))] + # Every board identity must belong to the same observed boot. A later + # identity missing boot_id must not silently inherit the previous lease. + if (not boot_ids or len(boot_ids) != len(identity_lines) or + len(set(boot_ids)) != 1): return {"status": "INCOMPLETE", "scenario": scenario, "reason": "one unambiguous 16-digit boot_id required", "reportedLockSamples": 0, "reportedRecovery": False, "reportedGmHandoff": False} if role == "source": - ready = [tuple(int(x) for x in m.groups()) - for line in lines if (m := SOURCE_READY.search(line))] - usable = any(all(x > 0 for x in values) for values in ready) + # A historical RUNNING line is not current readiness. The actual + # firmware SHOW includes TXfail and PdelayFrames; partial output is + # never acceptable. STOP, rejected START and unavailable status + # invalidate all prior ready observations. + last_ready = False + tx_fault_observed = False + ready_samples = 0 + for line in lines: + if SOURCE_INVALIDATION.search(line): + last_ready = False + if "PTP status=" not in line: + continue + match = SOURCE_STATUS.search(line) + if match is None: + last_ready = False + continue + state, *reported = match.groups() + announce, sync, followup, _pdelay, txfail = map(int, reported) + tx_fault_observed = tx_fault_observed or txfail != 0 + last_ready = (state == "RUNNING" and + announce > 0 and sync > 0 and followup > 0 and + txfail == 0) + ready_samples += int(last_ready) + usable = last_ready and not tx_fault_observed return {"status": "OBSERVED" if usable else "INCOMPLETE", "scenario": scenario, - "reason": "source Announce+Sync+FollowUp TX readiness reported" - if usable else "verified source TX readiness evidence missing", - "reportedSourceReadySamples": sum(all(x > 0 for x in v) - for v in ready), + "reason": "latest SOURCE status RUNNING with successful PTP TX" + if usable else "latest SOURCE readiness absent, stopped or TX faulted", + "reportedSourceReadySamples": ready_samples, + "sourceTxFaultObserved": tx_fault_observed, "reportedLockSamples": 0, "reportedRecovery": False, "reportedGmHandoff": False} diff --git a/tests/test_ptp_m3e_bench_gate.py b/tests/test_ptp_m3e_bench_gate.py index ea093c56b..642fc737c 100644 --- a/tests/test_ptp_m3e_bench_gate.py +++ b/tests/test_ptp_m3e_bench_gate.py @@ -319,6 +319,43 @@ def test_source_requires_actual_sent_ptp_triplet(self): "source-ready", "source") self.assertEqual(complete["status"], "OBSERVED") + def test_source_readiness_is_latest_not_any_historical_running(self): + prefix = (f"ARSTACK identity build={HEAD[:16]} boot_id={BOOT}\\n" + "PTP2 role=SOURCE discipline=UNLOCKED source=NONE offset=NA\\n") + running = "PTP status=RUNNING Announce=4 Sync=4 FollowUp=4 PdelayFrames=0 TXfail=0\\n" + for suffix in ("PTP stop requested\\n", + "PTP status=STOPPED Announce=4 Sync=4 FollowUp=4 PdelayFrames=0 TXfail=0\\n", + "PTP status unavailable\\n", + "PTP start rejected: link down\\n", + "PTP status=RUNNING Announce=4 Sync=4 FollowUp=4 PdelayFrames=0\\n"): + with self.subTest(suffix=suffix): + actual = self.run_gate((prefix + running + suffix).encode(), + "source-ready", "source") + self.assertEqual(actual["status"], "INCOMPLETE") + # A later clean RUNNING status is evidence of a newly reported state. + clean_restart = prefix + running + "PTP stop requested\\n" + running + self.assertEqual(self.run_gate(clean_restart.encode(), "source-ready", + "source")["status"], "OBSERVED") + + def test_source_any_reported_tx_failure_vetoes_same_boot(self): + prefix = (f"ARSTACK identity build={HEAD[:16]} boot_id={BOOT}\\n" + "PTP2 role=SOURCE discipline=UNLOCKED source=NONE offset=NA\\n") + bad = "PTP status=RUNNING Announce=4 Sync=4 FollowUp=4 PdelayFrames=0 TXfail=1\\n" + good = "PTP status=RUNNING Announce=8 Sync=8 FollowUp=8 PdelayFrames=0 TXfail=0\\n" + actual = self.run_gate((prefix + bad + good).encode(), "source-ready", "source") + self.assertEqual(actual["status"], "INCOMPLETE") + self.assertTrue(actual["sourceTxFaultObserved"]) + self.assertEqual(self.run_gate((prefix + bad).encode(), "source-ready", + "source")["status"], "INCOMPLETE") + + def test_missing_boot_id_on_later_identity_fails_closed(self): + data = (f"ARSTACK identity build={HEAD[:16]} boot_id={BOOT}\\n" + "PTP2 role=SOURCE discipline=UNLOCKED source=NONE offset=NA\\n" + f"ARSTACK identity build={HEAD[:16]}\\n" + "PTP status=RUNNING Announce=4 Sync=4 FollowUp=4 PdelayFrames=0 TXfail=0\\n") + self.assertEqual(self.run_gate(data.encode(), "source-ready", + "source")["status"], "INCOMPLETE") + def test_incompatible_role_scenario_is_rejected(self): with self.assertRaisesRegex(module.EvidenceError, "incompatible"): self.run_gate(bench_log("LOCKED"), "source-ready", "receiver") From 7472f6918c364d8e4453a65fe7e6a811b0f2f0e3 Mon Sep 17 00:00:00 2001 From: Ari Sulistiono Date: Sat, 17 Oct 2026 15:27:41 +0700 Subject: [PATCH 2/2] test(ptp-m3e): use actual serial line breaks in SOURCE fixtures --- tests/test_ptp_m3e_bench_gate.py | 34 ++++++++++++++++---------------- 1 file changed, 17 insertions(+), 17 deletions(-) diff --git a/tests/test_ptp_m3e_bench_gate.py b/tests/test_ptp_m3e_bench_gate.py index 642fc737c..89bb952b0 100644 --- a/tests/test_ptp_m3e_bench_gate.py +++ b/tests/test_ptp_m3e_bench_gate.py @@ -320,28 +320,28 @@ def test_source_requires_actual_sent_ptp_triplet(self): self.assertEqual(complete["status"], "OBSERVED") def test_source_readiness_is_latest_not_any_historical_running(self): - prefix = (f"ARSTACK identity build={HEAD[:16]} boot_id={BOOT}\\n" - "PTP2 role=SOURCE discipline=UNLOCKED source=NONE offset=NA\\n") - running = "PTP status=RUNNING Announce=4 Sync=4 FollowUp=4 PdelayFrames=0 TXfail=0\\n" - for suffix in ("PTP stop requested\\n", - "PTP status=STOPPED Announce=4 Sync=4 FollowUp=4 PdelayFrames=0 TXfail=0\\n", - "PTP status unavailable\\n", - "PTP start rejected: link down\\n", - "PTP status=RUNNING Announce=4 Sync=4 FollowUp=4 PdelayFrames=0\\n"): + prefix = (f"ARSTACK identity build={HEAD[:16]} boot_id={BOOT}\n" + "PTP2 role=SOURCE discipline=UNLOCKED source=NONE offset=NA\n") + running = "PTP status=RUNNING Announce=4 Sync=4 FollowUp=4 PdelayFrames=0 TXfail=0\n" + for suffix in ("PTP stop requested\n", + "PTP status=STOPPED Announce=4 Sync=4 FollowUp=4 PdelayFrames=0 TXfail=0\n", + "PTP status unavailable\n", + "PTP start rejected: link down\n", + "PTP status=RUNNING Announce=4 Sync=4 FollowUp=4 PdelayFrames=0\n"): with self.subTest(suffix=suffix): actual = self.run_gate((prefix + running + suffix).encode(), "source-ready", "source") self.assertEqual(actual["status"], "INCOMPLETE") # A later clean RUNNING status is evidence of a newly reported state. - clean_restart = prefix + running + "PTP stop requested\\n" + running + clean_restart = prefix + running + "PTP stop requested\n" + running self.assertEqual(self.run_gate(clean_restart.encode(), "source-ready", "source")["status"], "OBSERVED") def test_source_any_reported_tx_failure_vetoes_same_boot(self): - prefix = (f"ARSTACK identity build={HEAD[:16]} boot_id={BOOT}\\n" - "PTP2 role=SOURCE discipline=UNLOCKED source=NONE offset=NA\\n") - bad = "PTP status=RUNNING Announce=4 Sync=4 FollowUp=4 PdelayFrames=0 TXfail=1\\n" - good = "PTP status=RUNNING Announce=8 Sync=8 FollowUp=8 PdelayFrames=0 TXfail=0\\n" + prefix = (f"ARSTACK identity build={HEAD[:16]} boot_id={BOOT}\n" + "PTP2 role=SOURCE discipline=UNLOCKED source=NONE offset=NA\n") + bad = "PTP status=RUNNING Announce=4 Sync=4 FollowUp=4 PdelayFrames=0 TXfail=1\n" + good = "PTP status=RUNNING Announce=8 Sync=8 FollowUp=8 PdelayFrames=0 TXfail=0\n" actual = self.run_gate((prefix + bad + good).encode(), "source-ready", "source") self.assertEqual(actual["status"], "INCOMPLETE") self.assertTrue(actual["sourceTxFaultObserved"]) @@ -349,10 +349,10 @@ def test_source_any_reported_tx_failure_vetoes_same_boot(self): "source")["status"], "INCOMPLETE") def test_missing_boot_id_on_later_identity_fails_closed(self): - data = (f"ARSTACK identity build={HEAD[:16]} boot_id={BOOT}\\n" - "PTP2 role=SOURCE discipline=UNLOCKED source=NONE offset=NA\\n" - f"ARSTACK identity build={HEAD[:16]}\\n" - "PTP status=RUNNING Announce=4 Sync=4 FollowUp=4 PdelayFrames=0 TXfail=0\\n") + data = (f"ARSTACK identity build={HEAD[:16]} boot_id={BOOT}\n" + "PTP2 role=SOURCE discipline=UNLOCKED source=NONE offset=NA\n" + f"ARSTACK identity build={HEAD[:16]}\n" + "PTP status=RUNNING Announce=4 Sync=4 FollowUp=4 PdelayFrames=0 TXfail=0\n") self.assertEqual(self.run_gate(data.encode(), "source-ready", "source")["status"], "INCOMPLETE")