From 8623aab2819e390de508572ec3b644b38ea35507 Mon Sep 17 00:00:00 2001 From: Fred E <7602667+wallscaler@users.noreply.github.com> Date: Sun, 2 Aug 2026 03:15:32 -0400 Subject: [PATCH 1/4] fix(sn39): isolate the public validator status stream Keep the complete validator journal private and project only a fixed operational allowlist into the group-readable status file. Bind the config, release launcher, systemd units, publisher, CLI, and regression tests to the same split so arbitrary event fields cannot reach the public stream. --- config/validator-mainnet-sn39.toml | 3 + .../sn39/cathedral-sn39-public-status.service | 6 +- .../sn39/cathedral-sn39-release-launcher.py | 4 +- ...cathedral-validator-sn39-reconcile.service | 2 +- deploy/sn39/cathedral-validator-sn39.service | 6 +- scaffold/cli.py | 10 ++ scaffold/events.py | 121 +++++++++++------- .../tests/test_validator_two_mode.py | 13 +- scaffold/validator_thin.py | 2 + scripts/publish_sn39_validator_status.py | 4 +- tests/thin/test_status_sanitization.py | 114 +++++++++++++++++ tests/thin/test_status_stream_contract.py | 69 ++++++++++ 12 files changed, 297 insertions(+), 57 deletions(-) create mode 100644 tests/thin/test_status_sanitization.py create mode 100644 tests/thin/test_status_stream_contract.py diff --git a/config/validator-mainnet-sn39.toml b/config/validator-mainnet-sn39.toml index 7577c40b..3f40e1dd 100644 --- a/config/validator-mainnet-sn39.toml +++ b/config/validator-mainnet-sn39.toml @@ -63,3 +63,6 @@ require_completed_launch_for_broadcast = true [logs] jsonl = "/var/log/cathedral-validator/validator-events.jsonl" +# Public status reads this fixed allowlist projection. The raw journal above +# carries hotkeys and caller-supplied fields and remains private. +status_jsonl = "/var/log/cathedral-validator/validator-status.jsonl" diff --git a/deploy/sn39/cathedral-sn39-public-status.service b/deploy/sn39/cathedral-sn39-public-status.service index 92dd48a1..16a0ab61 100644 --- a/deploy/sn39/cathedral-sn39-public-status.service +++ b/deploy/sn39/cathedral-sn39-public-status.service @@ -9,7 +9,7 @@ After=cathedral-validator-sn39.service # pre-cutover publisher: the live host runs its own status publisher against # the current validator's event stream, and that unit must stay in place until # step 8 of the staged cutover replaces it. -ConditionPathExists=/var/log/cathedral-validator/validator-events.jsonl +ConditionPathExists=/var/log/cathedral-validator/validator-status.jsonl [Service] Type=oneshot @@ -47,7 +47,9 @@ SystemCallArchitectures=native RestrictAddressFamilies=AF_UNIX CapabilityBoundingSet= AmbientCapabilities= -ReadOnlyPaths=/var/log/cathedral-validator/validator-events.jsonl +# The raw event journal remains private. This unit sees only the strict +# fixed-field projection produced by the validator. +ReadOnlyPaths=/var/log/cathedral-validator/validator-status.jsonl ReadOnlyPaths=/var/lib/cathedral-public-evidence/index.json # Optional. release.json exists only once a root-signed release has been # sealed, and it is absent on the host today. An unprefixed ReadOnlyPaths= on a diff --git a/deploy/sn39/cathedral-sn39-release-launcher.py b/deploy/sn39/cathedral-sn39-release-launcher.py index 96cbb83f..b9e2d68d 100644 --- a/deploy/sn39/cathedral-sn39-release-launcher.py +++ b/deploy/sn39/cathedral-sn39-release-launcher.py @@ -299,7 +299,9 @@ def _child_environment( "PYTHONNOUSERSITE": "1", } if mode in {"preflight", "launch", "continuous", "reconcile"}: - environment["CATHEDRAL_VALIDATOR_JSONL_GROUP"] = "cathedral-validator-log" + # The separate public publisher reads only the fixed-field status + # projection. Never grant its group access to the raw event journal. + environment["CATHEDRAL_VALIDATOR_STATUS_GROUP"] = "cathedral-validator-log" if release_sha is not None: environment["CATHEDRAL_SN39_RELEASE_SHA"] = release_sha if launch_config_sha256 is not None: diff --git a/deploy/sn39/cathedral-validator-sn39-reconcile.service b/deploy/sn39/cathedral-validator-sn39-reconcile.service index d98f08b6..3edbee89 100644 --- a/deploy/sn39/cathedral-validator-sn39-reconcile.service +++ b/deploy/sn39/cathedral-validator-sn39-reconcile.service @@ -14,7 +14,7 @@ User=cathedral-validator # /var/log/cathedral-validator from group cathedral-validator-log to group # cathedral-validator the first time it ran, silently revoking the group read # that cathedral-sn39-public-status.service depends on. The status publisher -# would then stop being able to read validator-events.jsonl, and because it is +# would then stop being able to read validator-status.jsonl, and because it is # a oneshot timer the failure surfaces only as a status page that stopped # updating. Group=cathedral-validator-log diff --git a/deploy/sn39/cathedral-validator-sn39.service b/deploy/sn39/cathedral-validator-sn39.service index a50345db..8df9259d 100644 --- a/deploy/sn39/cathedral-validator-sn39.service +++ b/deploy/sn39/cathedral-validator-sn39.service @@ -13,8 +13,9 @@ User=cathedral-validator # Primary group is cathedral-validator-log, NOT cathedral-validator, and the # difference from the launch and reconcile units is deliberate. LogsDirectory= # below creates /var/log/cathedral-validator 0750 owned by this process's -# primary group, and cathedral-sn39-public-status.service reads -# validator-events.jsonl out of it using SupplementaryGroups=cathedral-validator-log. +# primary group, and cathedral-sn39-public-status.service reads the sanitized +# validator-status.jsonl projection out of it using the reader group. The raw +# validator-events.jsonl journal remains 0600 and is never group-readable. # Setting this to cathedral-validator to match the one-shots would leave the # status publisher unable to read the stream it exists to publish. The one-shots # write their own LogsDirectory that nothing else reads, so they have no reason @@ -22,6 +23,7 @@ User=cathedral-validator Group=cathedral-validator-log SupplementaryGroups=cathedral-validator-evidence Environment=HOME=/var/lib/cathedral-validator +Environment=CATHEDRAL_VALIDATOR_STATUS_GROUP=cathedral-validator-log # Belt and braces. Conflicts= only acts on units systemd is asked to start; # it cannot stop a writer someone launched by hand. This refuses to start # at all while any known SN39 writer is active, because the legacy writer diff --git a/scaffold/cli.py b/scaffold/cli.py index 52e01b79..564af79b 100644 --- a/scaffold/cli.py +++ b/scaffold/cli.py @@ -76,6 +76,7 @@ "provenance_burn_hotkey": None, "provenance_index_max_age_secs": 3600.0, "jsonl": None, # JSONL event stream file + "status_jsonl": None, # sanitized status projection } # config-file keys -> our flat config keys (a [section].key map, flattened) @@ -116,6 +117,7 @@ ("provenance", "source_revision"): "provenance_source_revision", ("provenance", "burn_hotkey"): "provenance_burn_hotkey", ("logs", "jsonl"): "jsonl", + ("logs", "status_jsonl"): "status_jsonl", } # env var -> our flat config key @@ -145,6 +147,7 @@ "CATHEDRAL_PROVENANCE_SOURCE_REVISION": "provenance_source_revision", "CATHEDRAL_PROVENANCE_BURN_HOTKEY": "provenance_burn_hotkey", "CATHEDRAL_VALIDATOR_JSONL": "jsonl", + "CATHEDRAL_VALIDATOR_STATUS_JSONL": "status_jsonl", } @@ -211,6 +214,7 @@ def _resolve_serve_config(ns: argparse.Namespace) -> SimpleNamespace: "provenance_source_revision", "provenance_burn_hotkey", "jsonl", + "status_jsonl", ): v = getattr(ns, flat, None) if v is not None: @@ -499,6 +503,12 @@ def main(argv: list[str] | None = None) -> int: default=None, help="append the stable JSONL event stream to this file", ) + sp.add_argument( + "--status-jsonl", + dest="status_jsonl", + default=None, + help="append the sanitized operational status projection to this file", + ) sp.add_argument( "--dry-run", action="store_true", diff --git a/scaffold/events.py b/scaffold/events.py index 23c9bf49..f7d63fef 100644 --- a/scaffold/events.py +++ b/scaffold/events.py @@ -23,6 +23,7 @@ import os import re import sys +from collections.abc import Mapping from datetime import UTC, datetime from typing import IO, Any @@ -149,6 +150,64 @@ def _redact(value: str) -> str: return _neutralize(value) +STATUS_FIELDS = ( + "ts", + "event", + "stage", + "mode", + "status", + "duration_ms", + "artifact", + "detail", + "remediation", +) + + +def _open_secure_append(path: str, group: str | None, label: str) -> IO[str]: + """Open an owner-controlled append-only file without following symlinks.""" + import stat as _stat + + group_gid = grp.getgrnam(group).gr_gid if group is not None else None + flags = ( + os.O_WRONLY + | os.O_APPEND + | os.O_CREAT + | getattr(os, "O_NOFOLLOW", 0) + | getattr(os, "O_CLOEXEC", 0) + ) + descriptor = os.open(path, flags, 0o600) + try: + opened = os.fstat(descriptor) + opened_mode = _stat.S_IMODE(opened.st_mode) + if ( + not _stat.S_ISREG(opened.st_mode) + or opened.st_uid != os.geteuid() + or opened_mode & 0o007 + or opened_mode not in (0o600, 0o640) + ): + raise ValueError(f"{label} must be an owner-controlled regular file") + if group_gid is None: + if opened_mode & 0o070: + raise ValueError( + f"{label} must be private (0600) without a reader group" + ) + else: + os.fchown(descriptor, -1, group_gid) + os.fchmod(descriptor, 0o640) + secured = os.fstat(descriptor) + if secured.st_gid != group_gid or _stat.S_IMODE(secured.st_mode) != 0o640: + raise ValueError(f"{label} reader-group setup failed") + except BaseException: + os.close(descriptor) + raise + return os.fdopen(descriptor, "a", encoding="utf-8") + + +def sanitized_status_record(record: Mapping[str, Any]) -> dict[str, Any]: + """Project a raw event onto the fixed public operational allowlist.""" + return {key: record[key] for key in STATUS_FIELDS if key in record} + + class EventLogger: def __init__( self, @@ -157,60 +216,21 @@ def __init__( jsonl: IO[str] | None = None, jsonl_path: str | None = None, jsonl_group: str | None = None, + status_path: str | None = None, + status_group: str | None = None, tty: IO[str] | None = None, color: bool | None = None, ) -> None: self.mode = _neutralize(mode)[:32] self._jsonl = jsonl self._jsonl_file: IO[str] | None = None + self._status_file: IO[str] | None = None if jsonl_path: - group_gid = ( - grp.getgrnam(jsonl_group).gr_gid if jsonl_group is not None else None + self._jsonl_file = _open_secure_append(jsonl_path, jsonl_group, "event log") + if status_path: + self._status_file = _open_secure_append( + status_path, status_group, "status log" ) - # Secure append: refuse symlinks/non-regular files, create 0600, - # and permit 0640 only when an explicit reader group is pinned by - # the service. This lets a separate sanitizer read the private - # source without giving the validator access to the public tree. - flags = ( - os.O_WRONLY - | os.O_APPEND - | os.O_CREAT - | getattr(os, "O_NOFOLLOW", 0) - | getattr(os, "O_CLOEXEC", 0) - ) - descriptor = os.open(jsonl_path, flags, 0o600) - import stat as _stat - - try: - opened = os.fstat(descriptor) - opened_mode = _stat.S_IMODE(opened.st_mode) - if ( - not _stat.S_ISREG(opened.st_mode) - or opened.st_uid != os.geteuid() - or opened_mode & 0o007 - or opened_mode not in (0o600, 0o640) - ): - raise ValueError( - "event log must be an owner-controlled regular file" - ) - if group_gid is None: - if opened_mode & 0o070: - raise ValueError( - "event log must be private (0600) without a reader group" - ) - else: - os.fchown(descriptor, -1, group_gid) - os.fchmod(descriptor, 0o640) - secured = os.fstat(descriptor) - if ( - secured.st_gid != group_gid - or _stat.S_IMODE(secured.st_mode) != 0o640 - ): - raise ValueError("event log reader-group setup failed") - except BaseException: - os.close(descriptor) - raise - self._jsonl_file = os.fdopen(descriptor, "a", encoding="utf-8") self._tty = tty if tty is not None else sys.stdout if color is None: color = ( @@ -225,6 +245,9 @@ def close(self) -> None: if self._jsonl_file is not None: self._jsonl_file.close() self._jsonl_file = None + if self._status_file is not None: + self._status_file.close() + self._status_file = None def event( self, @@ -270,6 +293,12 @@ def event( if target is not None: target.write(line + "\n") target.flush() + if self._status_file is not None: + status_line = json.dumps( + sanitized_status_record(record), separators=(",", ":"), allow_nan=False + ) + self._status_file.write(status_line + "\n") + self._status_file.flush() self._write_tty(record) return record diff --git a/scaffold/publisher/tests/test_validator_two_mode.py b/scaffold/publisher/tests/test_validator_two_mode.py index 8552254d..c72de986 100644 --- a/scaffold/publisher/tests/test_validator_two_mode.py +++ b/scaffold/publisher/tests/test_validator_two_mode.py @@ -3755,6 +3755,7 @@ def test_cli_to_tick_to_current_assertion_and_immutable_reproducer( config = root / "config/validator-mainnet-sn39.toml" state = tmp_path / "state.json" events = tmp_path / "events.jsonl" + status_events = tmp_path / "status-events.jsonl" vector = validated_supply_payload() vector.update( { @@ -3822,6 +3823,8 @@ def test_cli_to_tick_to_current_assertion_and_immutable_reproducer( str(tmp_path / "runtime"), "--jsonl", str(events), + "--status-jsonl", + str(status_events), "--dry-run", "--once", ] @@ -7072,15 +7075,17 @@ def load(path: Path, name: str): "LC_ALL", "PYTHONDONTWRITEBYTECODE", "PYTHONNOUSERSITE", - "CATHEDRAL_VALIDATOR_JSONL_GROUP", + "CATHEDRAL_VALIDATOR_STATUS_GROUP", } assert ( - child_environment["CATHEDRAL_VALIDATOR_JSONL_GROUP"] + child_environment["CATHEDRAL_VALIDATOR_STATUS_GROUP"] == "cathedral-validator-log" ) + assert "CATHEDRAL_VALIDATOR_JSONL_GROUP" not in child_environment status_environment = launcher._child_environment("status") assert status_environment["HOME"] == "/var/lib/cathedral-public-evidence" assert "CATHEDRAL_VALIDATOR_JSONL_GROUP" not in status_environment + assert "CATHEDRAL_VALIDATOR_STATUS_GROUP" not in status_environment launch_environment = launcher._child_environment( "launch", release_sha="a" * 40, @@ -7138,7 +7143,7 @@ def _directives(text: str) -> list[str]: ) in release_guide assert '"$release/scripts/finalize_sn39_public_release.py"' not in release_guide assert ( - "ReadOnlyPaths=/var/log/cathedral-validator/validator-events.jsonl" + "ReadOnlyPaths=/var/log/cathedral-validator/validator-status.jsonl" in status_unit ) assert "ReadWritePaths=/var/lib/cathedral-public-evidence/logs" in status_unit @@ -7152,7 +7157,7 @@ def _directives(text: str) -> list[str]: # Units that declare the same LogsDirectory= must declare the same Group=. # systemd applies the unit's User:Group to a logs directory it manages, so # a mismatch lets whichever unit ran last silently re-group the directory - # and revoke the status publisher's group read on validator-events.jsonl. + # and revoke the status publisher's group read on validator-status.jsonl. _logs_dir_group: dict[str, set[str]] = {} for _unit_name in ( "cathedral-validator-sn39.service", diff --git a/scaffold/validator_thin.py b/scaffold/validator_thin.py index 95faf2a0..b6cbceee 100644 --- a/scaffold/validator_thin.py +++ b/scaffold/validator_thin.py @@ -1261,6 +1261,8 @@ def _get_events(args) -> EventLogger: mode=authority, jsonl_path=getattr(args, "jsonl", None) or None, jsonl_group=os.environ.get("CATHEDRAL_VALIDATOR_JSONL_GROUP") or None, + status_path=getattr(args, "status_jsonl", None) or None, + status_group=os.environ.get("CATHEDRAL_VALIDATOR_STATUS_GROUP") or None, tty=sys.stdout, ) try: diff --git a/scripts/publish_sn39_validator_status.py b/scripts/publish_sn39_validator_status.py index ee8f0f04..b11fce5b 100644 --- a/scripts/publish_sn39_validator_status.py +++ b/scripts/publish_sn39_validator_status.py @@ -16,7 +16,9 @@ from pathlib import Path from typing import Any -SOURCE = Path("/var/log/cathedral-validator/validator-events.jsonl") +# The raw event journal carries hotkeys and arbitrary caller fields. The public +# publisher reads only the validator's strict fixed-field projection. +SOURCE = Path("/var/log/cathedral-validator/validator-status.jsonl") PUBLIC_ROOT = Path("/var/lib/cathedral-public-evidence") LOG_ROOT = PUBLIC_ROOT / "logs" INDEX = PUBLIC_ROOT / "index.json" diff --git a/tests/thin/test_status_sanitization.py b/tests/thin/test_status_sanitization.py new file mode 100644 index 00000000..4c14651c --- /dev/null +++ b/tests/thin/test_status_sanitization.py @@ -0,0 +1,114 @@ +"""Raw validator events must never widen the public status stream.""" + +from __future__ import annotations + +import grp +import json +import os +import stat + +import pytest + +from scaffold.events import STATUS_FIELDS, EventLogger, sanitized_status_record + + +def _mode(path) -> int: + return stat.S_IMODE(os.stat(path).st_mode) + + +def test_arbitrary_event_fields_never_enter_the_status_stream(tmp_path): + raw = tmp_path / "validator-events.jsonl" + status = tmp_path / "validator-status.jsonl" + logger = EventLogger( + mode="thin", + jsonl_path=str(raw), + status_path=str(status), + tty=None, + ) + logger.event( + "CHAIN_SUBMITTED", + stage="submit", + status="PASS", + hotkey="5FF6FtDUhn7XdPYmEdH5XjLAmLfmwLTCNVBgcrj3A4sstwaw", + artifact="sha256:" + "a" * 64, + detail="uids=2", + receipt_body="SECRET-RECEIPT-PAYLOAD", + evidence_blob="SECRET-EVIDENCE", + future_unreviewed_field={"nested": "SECRET-FUTURE-VALUE"}, + ) + logger.close() + + assert _mode(raw) == 0o600 + assert _mode(status) == 0o600 + + raw_record = json.loads(raw.read_text(encoding="utf-8")) + status_record = json.loads(status.read_text(encoding="utf-8")) + assert raw_record["receipt_body"] == "SECRET-RECEIPT-PAYLOAD" + assert raw_record["future_unreviewed_field"]["nested"] == "SECRET-FUTURE-VALUE" + + assert set(status_record) <= set(STATUS_FIELDS) + for field in ( + "hotkey", + "receipt_body", + "evidence_blob", + "future_unreviewed_field", + ): + assert field not in status_record + serialized = json.dumps(status_record) + for value in ( + "SECRET-RECEIPT-PAYLOAD", + "SECRET-EVIDENCE", + "SECRET-FUTURE-VALUE", + "5FF6FtDUhn7XdPYmEdH5XjLAmLfmwLTCNVBgcrj3A4sstwaw", + ): + assert value not in serialized + + +def test_projection_shape_does_not_expand_with_the_raw_schema(): + projected = sanitized_status_record( + { + "ts": "2026-07-27T00:00:00.000Z", + "event": "STARTUP", + "stage": "startup", + "mode": "thin", + "status": "INFO", + "hotkey": "5xxxx", + "unknown_later_field": "leak", + } + ) + assert projected == { + "ts": "2026-07-27T00:00:00.000Z", + "event": "STARTUP", + "stage": "startup", + "mode": "thin", + "status": "INFO", + } + + +def test_group_readable_status_does_not_make_raw_journal_group_readable(tmp_path): + raw = tmp_path / "validator-events.jsonl" + status = tmp_path / "validator-status.jsonl" + group = grp.getgrgid(os.getegid()).gr_name + logger = EventLogger( + mode="thin", + jsonl_path=str(raw), + status_path=str(status), + status_group=group, + tty=None, + ) + logger.event("STARTUP", stage="startup", status="INFO", arbitrary="private") + logger.close() + + assert _mode(raw) == 0o600 + assert _mode(status) == 0o640 + assert "arbitrary" not in json.loads(status.read_text(encoding="utf-8")) + + +def test_existing_group_readable_raw_journal_is_refused_without_explicit_group( + tmp_path, +): + raw = tmp_path / "validator-events.jsonl" + raw.touch() + raw.chmod(0o640) + with pytest.raises(ValueError, match="private \\(0600\\) without a reader group"): + EventLogger(mode="thin", jsonl_path=str(raw), tty=None) diff --git a/tests/thin/test_status_stream_contract.py b/tests/thin/test_status_stream_contract.py new file mode 100644 index 00000000..8b0a15e7 --- /dev/null +++ b/tests/thin/test_status_stream_contract.py @@ -0,0 +1,69 @@ +"""The writer, launcher, and public reader must agree on one safe stream.""" + +from __future__ import annotations + +import importlib.util +import pathlib +import re + +try: + import tomllib +except ModuleNotFoundError: # pragma: no cover + import tomli as tomllib + +_ROOT = pathlib.Path(__file__).resolve().parents[2] +_CONFIG = _ROOT / "config" / "validator-mainnet-sn39.toml" +_STATUS_UNIT = _ROOT / "deploy" / "sn39" / "cathedral-sn39-public-status.service" +_VALIDATOR_UNIT = _ROOT / "deploy" / "sn39" / "cathedral-validator-sn39.service" +_LAUNCHER = _ROOT / "deploy" / "sn39" / "cathedral-sn39-release-launcher.py" +_PUBLISHER = _ROOT / "scripts" / "publish_sn39_validator_status.py" + +_spec = importlib.util.spec_from_file_location("_sn39_launcher_contract", _LAUNCHER) +_launcher = importlib.util.module_from_spec(_spec) +_spec.loader.exec_module(_launcher) + + +def _logs() -> dict: + return tomllib.loads(_CONFIG.read_text(encoding="utf-8"))["logs"] + + +def test_config_writes_a_distinct_status_projection(): + logs = _logs() + assert logs["jsonl"] != logs["status_jsonl"] + assert logs["status_jsonl"].endswith("/validator-status.jsonl") + + +def test_public_unit_reads_exactly_the_configured_projection(): + unit = _STATUS_UNIT.read_text(encoding="utf-8") + assert f"ConditionPathExists={_logs()['status_jsonl']}" in unit + assert f"ReadOnlyPaths={_logs()['status_jsonl']}" in unit + assert f"ReadOnlyPaths={_logs()['jsonl']}" not in unit + assert "SupplementaryGroups=cathedral-validator-log" in unit + + +def test_publisher_source_is_the_projection_not_the_raw_journal(): + source = _PUBLISHER.read_text(encoding="utf-8") + assert f'SOURCE = Path("{_logs()["status_jsonl"]}")' in source + assert f'SOURCE = Path("{_logs()["jsonl"]}")' not in source + + +def test_launcher_grants_reader_group_only_to_projection(): + for mode in ("preflight", "launch", "continuous", "reconcile"): + environment = _launcher._child_environment(mode) + assert "CATHEDRAL_VALIDATOR_JSONL_GROUP" not in environment + assert ( + environment["CATHEDRAL_VALIDATOR_STATUS_GROUP"] == "cathedral-validator-log" + ) + status_environment = _launcher._child_environment("status") + assert "CATHEDRAL_VALIDATOR_JSONL_GROUP" not in status_environment + assert "CATHEDRAL_VALIDATOR_STATUS_GROUP" not in status_environment + + +def test_validator_unit_documents_the_same_access_split(): + unit = _VALIDATOR_UNIT.read_text(encoding="utf-8") + assert re.search( + r"^Environment=CATHEDRAL_VALIDATOR_STATUS_GROUP=cathedral-validator-log$", + unit, + re.M, + ) + assert not re.search(r"^Environment=CATHEDRAL_VALIDATOR_JSONL_GROUP=", unit, re.M) From dc55646e06dfa860a62c9d913b8e68290af4c8c0 Mon Sep 17 00:00:00 2001 From: Fred E <7602667+wallscaler@users.noreply.github.com> Date: Sun, 2 Aug 2026 03:51:00 -0400 Subject: [PATCH 2/4] fix: fail closed across SN39 status publication Keep raw validator journals private during upgrade and publish only typed, event-specific operational fields. Fence raw/status transitions durably so interrupted or malformed publication cannot preserve a stale PASS, and document the stopped-writer migration and rollback boundary. --- .github/workflows/two-mode-provenance.yml | 2 + docs/SN39_LAUNCH_CUTOVER_20260726.md | 55 +++- docs/SN39_MAINNET_RELEASE_20260724.md | 34 ++ scaffold/events.py | 214 ++++++++++++- .../tests/test_sn39_public_status.py | 213 ++++++++++++- .../tests/test_validator_two_mode.py | 3 + scaffold/validator_thin.py | 2 + scripts/build_sn39_release_manifest.py | 1 + scripts/migrate_sn39_status_stream.py | 153 +++++++++ scripts/publish_sn39_validator_status.py | 293 +++++++++++++++--- tests/thin/test_status_sanitization.py | 64 +++- tests/thin/test_status_stream_migration.py | 83 +++++ 12 files changed, 1037 insertions(+), 80 deletions(-) create mode 100644 scripts/migrate_sn39_status_stream.py create mode 100644 tests/thin/test_status_stream_migration.py diff --git a/.github/workflows/two-mode-provenance.yml b/.github/workflows/two-mode-provenance.yml index b2847268..e40bd013 100644 --- a/.github/workflows/two-mode-provenance.yml +++ b/.github/workflows/two-mode-provenance.yml @@ -57,6 +57,7 @@ jobs: scripts/build_sn39_rotation_manifest.py \ scripts/build_sn39_release_manifest.py \ scripts/finalize_sn39_public_release.py \ + scripts/migrate_sn39_status_stream.py \ scripts/publish_sn39_validator_status.py \ scripts/run_sn39_public_reproduction.py \ scripts/sn39_hotkey_rotation_operator.py \ @@ -92,6 +93,7 @@ jobs: scripts/build_sn39_rotation_manifest.py \ scripts/build_sn39_release_manifest.py \ scripts/finalize_sn39_public_release.py \ + scripts/migrate_sn39_status_stream.py \ scripts/publish_sn39_validator_status.py \ scripts/run_sn39_public_reproduction.py \ scripts/sn39_hotkey_rotation_operator.py \ diff --git a/docs/SN39_LAUNCH_CUTOVER_20260726.md b/docs/SN39_LAUNCH_CUTOVER_20260726.md index d8b767be..23c8ea4e 100644 --- a/docs/SN39_LAUNCH_CUTOVER_20260726.md +++ b/docs/SN39_LAUNCH_CUTOVER_20260726.md @@ -526,19 +526,42 @@ Apply in this order. Each step is independently reversible. 6. **Verify the publisher emits v2** with the exact required field set, using the assertion command in section 3 against port 8012. Do not continue until it prints `OK`. This is the gate for step 7. -7. **Install the validator release** with the bumped pins. Provenance can only +7. **Stop every validator writer and privatize both legacy raw journals.** + Prove every unit named by `WRITER_UNITS` in + `scripts/migrate_sn39_status_stream.py` is inactive, then run the reviewed + script from the immutable release as root: + + ```sh + /usr/bin/python3.12 -I -E -s \ + "$release/scripts/migrate_sn39_status_stream.py" + stat -c '%a %U:%G %n' \ + /var/log/cathedral-validator/validator-events.jsonl \ + /var/log/cathedral-validator-launch/validator-events.jsonl 2>/dev/null || true + ``` + + Every existing raw journal must report `600 cathedral-validator:*`. The + script refuses active writers, symlinks, hard links, unexpected ownership, + and modes other than 0600 or the one reviewed legacy mode, 0640. Do not + change modes with a pathname-based `chmod` while a writer is running. +8. **Install the validator release** with the bumped pins. Provenance can only reach FULL after step 4 has produced at least one manifest stamped `9540de44...`; before that the pin and the evidence still disagree. -8. **Replace the pre-cutover status publisher.** The live host runs its own - `cathedral-sn39-public-status.service` against the current validator's event - stream. The shipped unit is paired with `cathedral-validator-sn39.service` - and reads `/var/log/cathedral-validator/validator-events.jsonl`, which does - not exist until step 7 has run. Install it only after step 7, and confirm it - actually ran: its `ConditionPathExists=` makes systemd SKIP it silently if - the log is absent, so a skipped unit and a working one look identical in - `systemctl status`. Check `journalctl -u cathedral-sn39-public-status` for a - real execution, not just an absence of errors. -9. **Leave the three validator units disabled and inactive** until the launch +9. **Start the selected validator once and require a committed STARTUP.** The + raw journal remains 0600. The validator writes the separate + `/var/log/cathedral-validator/validator-status.jsonl` projection as 0640. + Its STARTUP row must preserve the reviewed `authority` and + `provenance_mode`. An unmatched `STATUS_PUBLICATION_PENDING` means the + raw/status transition was interrupted and every public gate is + `NOT_PROVEN`. Stop the selected validator after this observation. The + launch window remains closed. +10. **Replace the pre-cutover status publisher.** The shipped unit reads only + `/var/log/cathedral-validator/validator-status.jsonl`. It never receives + access to `validator-events.jsonl`. Install it only after step 9 created the + projection, then confirm a real execution in + `journalctl -u cathedral-sn39-public-status`. `ConditionPathExists=` skips a + missing projection without making the unit fail, so `systemctl status` + alone is insufficient. +11. **Leave the three validator units disabled and inactive** until the launch window. Their single-writer guards are unchanged: each names the other SN39 writers in `Conflicts=` and refuses to start via `ExecStartPre` while any of them is active. @@ -550,15 +573,17 @@ Apply in this order. Each step is independently reversible. | 2 | `rm /etc/sysusers.d/cathedral-sn39-validator.conf`. Accounts already created remain, which is harmless and matches the current host. | | 3 | `rm /etc/tmpfiles.d/cathedral-sn39-validator.conf`. No ownership was changed, so there is nothing to restore. This is the property proven above. | | 4 | Reinstall the previous exporter. Evidence returns to stamping `b77c7cf...`, which is wrong but is the current production behavior. | -| 5 | Reinstall the previous publisher revision. It returns to emitting v1, which the pre-cutover validator accepts. Roll back step 7 with it or the validator has nothing it will accept. | -| 7 | Reinstall the previous validator release. The pins revert together because they moved together. | -| 8 | Reinstall the pre-cutover status publisher unit. Nothing else depends on the shipped one. | +| 5 | Reinstall the previous publisher revision. It returns to emitting v1, which the pre-cutover validator accepts. Roll back step 8 with it or the validator has nothing it will accept. | +| 7 | Do not restore group access to a raw journal. Mode 0600 is compatible with the validator and is a permanent privacy hardening. | +| 8 | Reinstall the previous validator release. The pins revert together because they moved together. | +| 9 | Stop the selected validator. A partial public transition remains `NOT_PROVEN`; do not delete the fence to manufacture a PASS. | +| 10 | Either keep the sanitized publisher or stop public publication. Do not reinstall a publisher that requires group access to the raw journal. | Steps 1 through 3 do not touch the producer and can be done outside a maintenance window. Step 4 restarts the exporter and should be done between export cycles. -Steps 5 through 7 are one transaction. Rolling back the validator without +Steps 5, 6, and 8 are one transaction. Rolling back the validator without rolling back the publisher, or the reverse, reproduces the version mismatch in section 3 from the opposite direction: a v2 publisher feeding the `98b862b` validator fails the same exact-field-set comparison. Roll them back together. diff --git a/docs/SN39_MAINNET_RELEASE_20260724.md b/docs/SN39_MAINNET_RELEASE_20260724.md index 09dc07a7..ab523643 100644 --- a/docs/SN39_MAINNET_RELEASE_20260724.md +++ b/docs/SN39_MAINNET_RELEASE_20260724.md @@ -606,6 +606,22 @@ install -D -o root -g root -m 0644 \ systemd-sysusers /etc/sysusers.d/cathedral-sn39-validator.conf systemd-tmpfiles --create /etc/tmpfiles.d/cathedral-sn39-validator.conf +# Maintenance boundary for the status-stream privacy upgrade. Stop every +# possible writer before changing either existing raw journal from the legacy +# 0640 mode to 0600. The migration verifies stopped services, opens with +# O_NOFOLLOW, validates the descriptor and inode, and covers both the +# continuous and one-shot launch log paths. +systemctl stop \ + cathedral-validator-sn39.service \ + cathedral-validator-sn39-launch.service \ + cathedral-validator-sn39-reconcile.service \ + cathedral-thin-validator.service \ + cathedral-confidential-validator-sn39.service \ + cathedral-confidential-validator.service \ + cathedral-validator.service +/usr/bin/python3.12 -I -E -s \ + "$release/scripts/migrate_sn39_status_stream.py" + # Provision only the already-registered validator HOTKEY into the service # account. Run this on a secure interactive console. The source key may prompt # for its password; neither its mnemonic nor private bytes are printed. @@ -916,6 +932,24 @@ sudo /usr/bin/python3 -I -E -s \ `cathedral-validator-sn39.service` and `cathedral-sn39-public-status.timer`. + Before enabling the timer, require a committed STARTUP in + `/var/log/cathedral-validator/validator-status.jsonl` with the reviewed + `authority` and `provenance_mode`. The raw + `/var/log/cathedral-validator/validator-events.jsonl` must remain 0600. The + public unit reads only the 0640 status projection. If the newest unmatched + row is `STATUS_PUBLICATION_PENDING`, the transition was interrupted and the + published authority and provenance gates remain `NOT_PROVEN` until a clean + process restart emits a new committed STARTUP. + +### Status-stream rollback + +Stop every validator unit before rolling back. Reverting application bytes +does not justify restoring raw-journal group access. Leave both raw journals +0600. Either retain the sanitized status publisher or stop the public timer. +Never reinstall a legacy publisher that reads `validator-events.jsonl`. A +failed or partial status transition stays `NOT_PROVEN`; deleting its fence is +not a rollback. + The public status card is operational telemetry, not launch authorization. It reports authority `PASS` only for a fresh observed exact 90/10 submission whose UIDs were resolved from the signed rewarded and owner/burn hotkeys. A diff --git a/scaffold/events.py b/scaffold/events.py index f7d63fef..77f1454f 100644 --- a/scaffold/events.py +++ b/scaffold/events.py @@ -23,6 +23,7 @@ import os import re import sys +import uuid from collections.abc import Mapping from datetime import UTC, datetime from typing import IO, Any @@ -150,7 +151,7 @@ def _redact(value: str) -> str: return _neutralize(value) -STATUS_FIELDS = ( +STATUS_COMMON_FIELDS = ( "ts", "event", "stage", @@ -158,10 +159,99 @@ def _redact(value: str) -> str: "status", "duration_ms", "artifact", - "detail", - "remediation", ) +# Public status is a closed, event-specific schema. In particular, free-form +# ``detail``/``remediation`` and identifiers such as hotkeys never cross this +# boundary. The publisher derives fixed human text from the event code and the +# small structured fields below. +STATUS_EVENT_FIELDS: dict[str, tuple[str, ...]] = { + "STARTUP": ("authority", "provenance_mode"), + "PROVENANCE_AUDIT_NOT_PROVEN": ("positive_raw_replay",), + "WEIGHTS_DRY_RUN": ( + "authority", + "uid_count", + "burn_uid", + "burn_share", + "uid_weights", + ), + "WEIGHTS_SUBMITTED": ( + "authority", + "uid_count", + "burn_uid", + "burn_share", + "uid_weights", + ), + "PENDING_RECEIPT_RECOVERED": ( + "authority", + "uid_count", + "burn_uid", + "burn_share", + "uid_weights", + ), +} +STATUS_FENCE_FIELDS = ( + "publication_id", + "publication_phase", + "target_event", +) +STATUS_FIELDS = tuple( + dict.fromkeys( + (*STATUS_COMMON_FIELDS, *STATUS_FENCE_FIELDS) + + tuple(field for fields in STATUS_EVENT_FIELDS.values() for field in fields) + ) +) + +_SAFE_AUTHORITIES = frozenset({"thin", "full_provenance"}) +_SAFE_PROVENANCE_MODES = frozenset({"off", "shadow", "authority"}) +_STARTUP_MODE_PAIRS = frozenset( + { + ("thin", "off"), + ("thin", "shadow"), + ("full_provenance", "authority"), + } +) + + +def _structured_status_value(field: str, value: Any) -> Any | None: + """Return a safe public value, or ``None`` when it is not admissible.""" + if field == "authority": + return value if value in _SAFE_AUTHORITIES else None + if field == "provenance_mode": + return value if value in _SAFE_PROVENANCE_MODES else None + if field == "positive_raw_replay": + return value if isinstance(value, bool) else None + if field in {"uid_count", "burn_uid"}: + if isinstance(value, bool) or not isinstance(value, int) or value < 0: + return None + return value + if field == "burn_share": + if isinstance(value, bool) or not isinstance(value, (int, float)): + return None + parsed = float(value) + return round(parsed, 12) if math.isfinite(parsed) and 0 <= parsed <= 1 else None + if field == "uid_weights": + if not isinstance(value, Mapping) or len(value) > 512: + return None + clean: dict[str, float] = {} + for raw_uid, raw_weight in value.items(): + try: + uid = str(int(raw_uid)) + weight = float(raw_weight) + except (TypeError, ValueError): + return None + if ( + uid in clean + or int(uid) < 0 + or isinstance(raw_weight, bool) + or not math.isfinite(weight) + or not 0 <= weight <= 1 + ): + return None + clean[uid] = round(weight, 12) + return clean + return None + def _open_secure_append(path: str, group: str | None, label: str) -> IO[str]: """Open an owner-controlled append-only file without following symlinks.""" @@ -182,6 +272,7 @@ def _open_secure_append(path: str, group: str | None, label: str) -> IO[str]: if ( not _stat.S_ISREG(opened.st_mode) or opened.st_uid != os.geteuid() + or opened.st_nlink != 1 or opened_mode & 0o007 or opened_mode not in (0o600, 0o640) ): @@ -203,9 +294,80 @@ def _open_secure_append(path: str, group: str | None, label: str) -> IO[str]: return os.fdopen(descriptor, "a", encoding="utf-8") -def sanitized_status_record(record: Mapping[str, Any]) -> dict[str, Any]: - """Project a raw event onto the fixed public operational allowlist.""" - return {key: record[key] for key in STATUS_FIELDS if key in record} +def sanitized_status_record( + record: Mapping[str, Any], + *, + publication_id: str | None = None, +) -> dict[str, Any]: + """Project a raw event onto the closed public operational schema.""" + event = str(record.get("event", "")) + clean: dict[str, Any] = { + "ts": record["ts"], + "event": event, + "stage": ( + record["stage"] + if isinstance(record.get("stage"), str) + and re.fullmatch(r"[a-z][a-z0-9_-]{0,31}", record["stage"]) + else "unknown" + ), + "mode": record["mode"] + if record.get("mode") in _SAFE_AUTHORITIES + else "unknown", + "status": record["status"], + } + duration = record.get("duration_ms") + if ( + not isinstance(duration, bool) + and isinstance(duration, (int, float)) + and math.isfinite(float(duration)) + ): + clean["duration_ms"] = round(max(0.0, float(duration)), 3) + artifact = record.get("artifact") + if isinstance(artifact, str) and re.fullmatch(r"sha256:[0-9a-f]{64}", artifact): + clean["artifact"] = artifact + for field in STATUS_EVENT_FIELDS.get(event, ()): + value = _structured_status_value(field, record.get(field)) + if value is not None: + clean[field] = value + if ( + event == "STARTUP" + and (clean.get("authority"), clean.get("provenance_mode")) + not in _STARTUP_MODE_PAIRS + ): + # A self-contradictory STARTUP cannot establish public runtime mode. + clean.pop("authority", None) + clean.pop("provenance_mode", None) + if publication_id is not None: + clean["publication_id"] = publication_id + clean["publication_phase"] = "COMMITTED" + return clean + + +def pending_status_record( + record: Mapping[str, Any], publication_id: str +) -> dict[str, Any]: + """A durable fail-closed fence written before the corresponding raw event.""" + return { + "ts": record["ts"], + "event": "STATUS_PUBLICATION_PENDING", + "stage": "status", + "mode": record["mode"], + "status": NOT_PROVEN, + "publication_id": publication_id, + "publication_phase": "PENDING", + "target_event": record["event"], + } + + +def _durable_jsonl_write(target: IO[str], record: Mapping[str, Any]) -> None: + line = json.dumps(record, separators=(",", ":"), allow_nan=False) + target.write(line + "\n") + target.flush() + try: + descriptor = target.fileno() + except (AttributeError, OSError): + return + os.fsync(descriptor) class EventLogger: @@ -228,9 +390,22 @@ def __init__( if jsonl_path: self._jsonl_file = _open_secure_append(jsonl_path, jsonl_group, "event log") if status_path: - self._status_file = _open_secure_append( - status_path, status_group, "status log" - ) + try: + self._status_file = _open_secure_append( + status_path, status_group, "status log" + ) + except BaseException: + self.close() + raise + if self._jsonl_file is not None and self._status_file is not None: + raw_info = os.fstat(self._jsonl_file.fileno()) + status_info = os.fstat(self._status_file.fileno()) + if (raw_info.st_dev, raw_info.st_ino) == ( + status_info.st_dev, + status_info.st_ino, + ): + self.close() + raise ValueError("raw and public status logs must be distinct files") self._tty = tty if tty is not None else sys.stdout if color is None: color = ( @@ -288,17 +463,24 @@ def event( for key, value in fields.items(): if key not in record: record[key] = _scrub(value) - line = json.dumps(record, separators=(",", ":"), allow_nan=False) + if self._status_file is not None: + publication_id = uuid.uuid4().hex + # Fence first. If the process dies before the COMMITTED row, the + # public reader sees NOT_PROVEN rather than retaining an older PASS. + _durable_jsonl_write( + self._status_file, + pending_status_record(record, publication_id), + ) + else: + publication_id = None for target in (self._jsonl, self._jsonl_file): if target is not None: - target.write(line + "\n") - target.flush() + _durable_jsonl_write(target, record) if self._status_file is not None: - status_line = json.dumps( - sanitized_status_record(record), separators=(",", ":"), allow_nan=False + _durable_jsonl_write( + self._status_file, + sanitized_status_record(record, publication_id=publication_id), ) - self._status_file.write(status_line + "\n") - self._status_file.flush() self._write_tty(record) return record diff --git a/scaffold/publisher/tests/test_sn39_public_status.py b/scaffold/publisher/tests/test_sn39_public_status.py index 17b2f0f9..07dfcbe0 100644 --- a/scaffold/publisher/tests/test_sn39_public_status.py +++ b/scaffold/publisher/tests/test_sn39_public_status.py @@ -9,6 +9,8 @@ import pytest +from scaffold import events as event_stream +from scaffold.events import EventLogger from scripts import publish_sn39_validator_status as status @@ -215,6 +217,187 @@ def test_invalid_or_self_contradictory_startup_is_dropped() -> None: assert status.clean_event(invalid_pair) is None +def test_writer_to_publisher_preserves_startup_runtime_modes( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, +) -> None: + raw = tmp_path / "raw.jsonl" + public = tmp_path / "status.jsonl" + logger = EventLogger( + mode="thin", + jsonl_path=str(raw), + status_path=str(public), + tty=None, + ) + logger.event( + "STARTUP", + stage="startup", + status="INFO", + detail=( + "submission_authority=thin provenance=shadow " + "policy_pin=validated_supply_v1 network=finney netuid=39" + ), + authority="thin", + provenance_mode="shadow", + private_hotkey="5G3qVaXzKMPDm5AJ3dpzbpUC27kpccBvDwzSWXrq8M6qMmbC", + ) + logger.close() + + monkeypatch.setattr(status, "SOURCE", public) + rows = status.tail_events() + assert len(rows) == 1 + assert rows[0]["event"] == "STARTUP" + assert rows[0]["authority"] == "thin" + assert rows[0]["provenance_mode"] == "shadow" + assert rows[0]["detail"] == ( + "thin authority and concurrent provenance shadow started" + ) + assert "private_hotkey" not in rows[0] + + +def test_interrupted_raw_to_status_transition_overrides_stale_pass( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, +) -> None: + raw = tmp_path / "raw.jsonl" + public = tmp_path / "status.jsonl" + logger = EventLogger( + mode="thin", + jsonl_path=str(raw), + status_path=str(public), + tty=None, + ) + logger.event( + "STARTUP", + stage="startup", + status="INFO", + authority="thin", + provenance_mode="shadow", + ) + weights = {"163": 0.9, "204": 0.1} + logger.event( + "WEIGHTS_SUBMITTED", + stage="submit", + status="PASS", + authority="thin", + uid_count=2, + burn_uid=204, + burn_share=0.1, + uid_weights=weights, + ) + monkeypatch.setattr(status, "SOURCE", public) + assert status.build_status(status.tail_events())["authority"]["status"] == "PASS" + + original_write = event_stream._durable_jsonl_write + + def fail_before_status_commit(target, record): + if ( + target is logger._status_file + and record.get("publication_phase") == "COMMITTED" + ): + raise OSError("injected status commit failure") + original_write(target, record) + + monkeypatch.setattr( + event_stream, + "_durable_jsonl_write", + fail_before_status_commit, + ) + with pytest.raises(OSError, match="injected"): + logger.event( + "TICK_FAILED", + stage="result", + status="FAIL", + detail="private failure", + ) + logger.close() + + rows = status.tail_events() + assert rows[-1]["event"] == "STATUS_PUBLICATION_PENDING" + assert rows[-1]["status"] == "NOT_PROVEN" + document = status.build_status(rows) + assert document["authority"]["status"] == "NOT_PROVEN" + assert document["authority"]["latest_event"] == ("STATUS_PUBLICATION_PENDING") + assert document["provenance"]["current_whole_epoch_full"] == "NOT_PROVEN" + + +def test_unknown_private_event_commit_closes_its_publication_fence( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, +) -> None: + raw = tmp_path / "raw.jsonl" + public = tmp_path / "status.jsonl" + logger = EventLogger( + mode="thin", + jsonl_path=str(raw), + status_path=str(public), + tty=None, + ) + logger.event( + "CHAIN_SUBMITTED", + stage="submit", + status="PASS", + detail="private event outside the public allowlist", + ) + logger.close() + + monkeypatch.setattr(status, "SOURCE", public) + assert status.tail_events() == [] + + +def test_malformed_or_mismatched_commit_does_not_clear_pending_fence( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, +) -> None: + public = tmp_path / "status.jsonl" + publication_id = "a" * 32 + rows = [ + { + "ts": _timestamp(), + "event": "STATUS_PUBLICATION_PENDING", + "stage": "status", + "mode": "thin", + "status": "NOT_PROVEN", + "publication_id": publication_id, + "publication_phase": "PENDING", + "target_event": "STARTUP", + }, + { + "ts": _timestamp(), + "event": "VECTOR_ACCEPTED", + "stage": "policy", + "mode": "thin", + "status": "PASS", + "publication_id": publication_id, + "publication_phase": "COMMITTED", + }, + { + "ts": _timestamp(), + "event": "STARTUP", + "stage": "startup", + "mode": "thin", + "status": "INFO", + "publication_id": publication_id, + "publication_phase": "COMMITTED", + # The required structured mode pair is intentionally absent. + }, + ] + public.write_text( + "".join(json.dumps(row) + "\n" for row in rows), + encoding="utf-8", + ) + monkeypatch.setattr(status, "SOURCE", public) + + events = status.tail_events() + assert [event["event"] for event in events] == [ + "STATUS_PUBLICATION_PENDING", + "VECTOR_ACCEPTED", + ] + document = status.build_status(events) + assert document["authority"]["status"] == "NOT_PROVEN" + assert document["authority"]["latest_event"] == "STATUS_PUBLICATION_PENDING" + + def test_pending_receipt_contradiction_overrides_prior_thin_pass() -> None: startup = status.clean_event(_startup("thin", "shadow")) launch = status.clean_event( @@ -252,6 +435,12 @@ def test_event_status_mismatch_is_dropped() -> None: assert status.clean_event(_event("WEIGHTS_DRY_RUN", "FAIL"))["status"] == "FAIL" +def test_unexpected_fence_field_cannot_be_used_as_free_form_public_text() -> None: + document = _event("VECTOR_ACCEPTED", "PASS") + document["target_event"] = "5G3qVaXzKMPDm5AJ3dpzbpUC27kpccBvDwzSWXrq8M6qMmbC" + assert status.clean_event(document) is None + + def test_public_status_is_time_bounded() -> None: stale = status.clean_event( _event( @@ -274,18 +463,30 @@ def test_public_status_is_time_bounded() -> None: def test_rewarded_set_pass_does_not_claim_whole_epoch_full() -> None: rewarded = status.clean_event(_event("LAUNCH_REWARDED_SET_GATE_PASS", "PASS")) - provenance = status.clean_event( - _event( - "PROVENANCE_AUDIT_NOT_PROVEN", - "NOT_PROVEN", - detail="positive raw evidence replayed for 1 miners", - ) + raw_provenance = _event( + "PROVENANCE_AUDIT_NOT_PROVEN", + "NOT_PROVEN", + detail="private validator-local diagnostics", ) + raw_provenance["positive_raw_replay"] = True + provenance = status.clean_event(raw_provenance) document = status.build_status([rewarded, provenance]) assert document["provenance"]["rewarded_set_full"] == "PASS" assert document["provenance"]["positive_tdx_raw_replay"] == "PASS" assert document["provenance"]["whole_epoch_full"] == "NOT_PROVEN" + prose_only = status.clean_event( + _event( + "PROVENANCE_AUDIT_NOT_PROVEN", + "NOT_PROVEN", + detail="positive raw evidence replayed for a private identifier", + ) + ) + assert ( + status.build_status([prose_only])["provenance"]["positive_tdx_raw_replay"] + == "NOT_PROVEN" + ) + def test_current_full_audit_does_not_upgrade_receipts_only_launch( monkeypatch: pytest.MonkeyPatch, diff --git a/scaffold/publisher/tests/test_validator_two_mode.py b/scaffold/publisher/tests/test_validator_two_mode.py index c72de986..2fdac013 100644 --- a/scaffold/publisher/tests/test_validator_two_mode.py +++ b/scaffold/publisher/tests/test_validator_two_mode.py @@ -2223,6 +2223,7 @@ def event(self, name, **kw): fields for name, fields in events_seen if name == "PROVENANCE_AUDIT_NOT_PROVEN" ) assert "positive raw evidence replayed for 1 miner(s)" in not_proven["detail"] + assert not_proven["positive_raw_replay"] is True assert "replayable negative evidence" in not_proven["detail"] assert "raw evidence was not replayed" not in not_proven["detail"] state = json.loads(state_file.read_text()) if state_file.exists() else {} @@ -2253,6 +2254,7 @@ def event(self, name, **kw): fields for name, fields in events_seen if name == "PROVENANCE_AUDIT_NOT_PROVEN" ) assert "no positive raw evidence replayed" in fields["detail"] + assert fields["positive_raw_replay"] is False assert "positive raw evidence replayed for" not in fields["detail"] @@ -7231,6 +7233,7 @@ def _directives(text: str) -> list[str]: "deploy/sn39/cathedral-sn39-validator.sysusers", "deploy/sn39/cathedral-sn39-validator.tmpfiles", "scripts/publish_sn39_validator_status.py", + "scripts/migrate_sn39_status_stream.py", "scripts/finalize_sn39_public_release.py", "scripts/build_sn39_rotation_manifest.py", "scripts/sn39_hotkey_rotation_operator.py", diff --git a/scaffold/validator_thin.py b/scaffold/validator_thin.py index b6cbceee..bb7e9ab6 100644 --- a/scaffold/validator_thin.py +++ b/scaffold/validator_thin.py @@ -1498,6 +1498,7 @@ def _log_audit_events(args, audit, state_file: Path, *, persist: bool = True) -> duration_ms=audit.duration_ms, artifact=audit.manifest_digest, detail=detail[:512], + positive_raw_replay=bool(raw_replayed), vector_agrees=audit.agrees_with_vector, remediation=( "keep thin authority; FULL requires independently replayable " @@ -1615,6 +1616,7 @@ def _run_provenance_stage( status=NOT_PROVEN, artifact=audit.manifest_digest, detail="receipts-only recomputation cannot back authority", + positive_raw_replay=False, remediation="provide the controlled package and verifier pins", ) raise wire.VectorError( diff --git a/scripts/build_sn39_release_manifest.py b/scripts/build_sn39_release_manifest.py index 9b38a1b6..56ba5faa 100644 --- a/scripts/build_sn39_release_manifest.py +++ b/scripts/build_sn39_release_manifest.py @@ -59,6 +59,7 @@ "scripts/finalize_sn39_public_release.py", "scripts/build_sn39_rotation_manifest.py", "scripts/publish_sn39_validator_status.py", + "scripts/migrate_sn39_status_stream.py", "scripts/sn39_hotkey_rotation_operator.py", "deploy/sn39/cathedral-sn39-public-status.service", "deploy/sn39/cathedral-sn39-public-status.timer", diff --git a/scripts/migrate_sn39_status_stream.py b/scripts/migrate_sn39_status_stream.py new file mode 100644 index 00000000..2c35191f --- /dev/null +++ b/scripts/migrate_sn39_status_stream.py @@ -0,0 +1,153 @@ +#!/usr/bin/env python3 +"""Privatize legacy SN39 raw journals before status-stream cutover. + +Run only inside the documented maintenance window, after every validator +writer is stopped. The migration opens each known raw journal without +following symlinks, verifies the opened descriptor, changes 0640 to 0600, and +verifies the same inode again before returning success. +""" + +from __future__ import annotations + +import os +import pwd +import stat +import subprocess +import sys +from pathlib import Path + +ROOT_UID = 0 +SYSTEMCTL = Path("/usr/bin/systemctl") +VALIDATOR_USER = "cathedral-validator" +RAW_JOURNALS = ( + Path("/var/log/cathedral-validator/validator-events.jsonl"), + Path("/var/log/cathedral-validator-launch/validator-events.jsonl"), +) +WRITER_UNITS = ( + "cathedral-validator-sn39.service", + "cathedral-validator-sn39-launch.service", + "cathedral-validator-sn39-reconcile.service", + "cathedral-thin-validator.service", + "cathedral-confidential-validator-sn39.service", + "cathedral-confidential-validator.service", + "cathedral-validator.service", +) + + +class MigrationError(RuntimeError): + """The journal migration cannot be completed safely.""" + + +def require_writers_stopped() -> None: + """Refuse mode changes while any known validator writer is running.""" + for unit in WRITER_UNITS: + try: + result = subprocess.run( + [str(SYSTEMCTL), "is-active", unit], + cwd="/", + text=True, + capture_output=True, + timeout=5, + check=False, + env={"PATH": "/usr/bin:/bin", "LANG": "C", "LC_ALL": "C"}, + ) + except (OSError, subprocess.SubprocessError) as exc: + raise MigrationError( + "cannot prove every validator writer is stopped" + ) from exc + if result.stdout.strip() not in {"inactive", "failed"}: + raise MigrationError(f"validator writer is not stopped: {unit}") + + +def privatize_journal(path: Path, *, expected_uid: int) -> str: + """Safely convert one existing 0640 raw journal to 0600.""" + if not path.is_absolute() or path.name in {"", ".", ".."}: + raise MigrationError("raw journal path is not canonical") + nofollow = getattr(os, "O_NOFOLLOW", 0) + close_on_exec = getattr(os, "O_CLOEXEC", 0) + directory_flags = ( + os.O_RDONLY | nofollow | close_on_exec | getattr(os, "O_DIRECTORY", 0) + ) + try: + parent_descriptor = os.open(path.parent, directory_flags) + except FileNotFoundError: + return "absent" + except OSError as exc: + raise MigrationError( + f"raw journal directory cannot be opened safely: {path}" + ) from exc + try: + parent = os.fstat(parent_descriptor) + if ( + not stat.S_ISDIR(parent.st_mode) + or parent.st_uid not in {ROOT_UID, expected_uid} + or stat.S_IMODE(parent.st_mode) & 0o022 + ): + raise MigrationError( + f"raw journal directory is not controlled: {path.parent}" + ) + try: + descriptor = os.open( + path.name, + os.O_RDONLY | nofollow | close_on_exec, + dir_fd=parent_descriptor, + ) + except FileNotFoundError: + return "absent" + except OSError as exc: + raise MigrationError( + f"raw journal cannot be opened safely: {path}" + ) from exc + try: + before = os.fstat(descriptor) + before_mode = stat.S_IMODE(before.st_mode) + if ( + not stat.S_ISREG(before.st_mode) + or before.st_uid != expected_uid + or before.st_nlink != 1 + or before_mode not in {0o600, 0o640} + ): + raise MigrationError( + "raw journal is not a single-linked validator-owned " + f"0600/0640 file: {path}" + ) + if before_mode == 0o640: + os.fchmod(descriptor, 0o600) + os.fsync(descriptor) + after = os.fstat(descriptor) + if ( + (after.st_dev, after.st_ino) != (before.st_dev, before.st_ino) + or not stat.S_ISREG(after.st_mode) + or after.st_uid != expected_uid + or after.st_nlink != 1 + or stat.S_IMODE(after.st_mode) != 0o600 + ): + raise MigrationError(f"raw journal privacy verification failed: {path}") + return "changed" if before_mode == 0o640 else "already-private" + finally: + os.close(descriptor) + finally: + os.close(parent_descriptor) + + +def main() -> int: + if os.geteuid() != ROOT_UID: + print("status-stream migration must run as root", file=sys.stderr) + return 1 + try: + expected_uid = pwd.getpwnam(VALIDATOR_USER).pw_uid + require_writers_stopped() + results = [ + (path, privatize_journal(path, expected_uid=expected_uid)) + for path in RAW_JOURNALS + ] + except (KeyError, MigrationError) as exc: + print(f"status-stream migration failed closed: {exc}", file=sys.stderr) + return 1 + for path, result in results: + print(f"{path}: {result}") + return 0 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/scripts/publish_sn39_validator_status.py b/scripts/publish_sn39_validator_status.py index b11fce5b..49953c4a 100644 --- a/scripts/publish_sn39_validator_status.py +++ b/scripts/publish_sn39_validator_status.py @@ -60,6 +60,7 @@ "PENDING_RECEIPT_CONTRADICTION", "PENDING_RECEIPT_NOT_PROVEN", "PENDING_RECEIPT_RECOVERED", + "STATUS_PUBLICATION_PENDING", "TICK_FAILED", "VECTOR_ACCEPTED", "VECTOR_REJECTED", @@ -83,6 +84,7 @@ "PENDING_RECEIPT_CONTRADICTION": "FAIL", "PENDING_RECEIPT_NOT_PROVEN": "NOT_PROVEN", "PENDING_RECEIPT_RECOVERED": "PASS", + "STATUS_PUBLICATION_PENDING": "NOT_PROVEN", "TICK_FAILED": "FAIL", "VECTOR_ACCEPTED": "PASS", "VECTOR_REJECTED": "FAIL", @@ -124,6 +126,16 @@ "artifact", "detail", "remediation", + "authority", + "provenance_mode", + "uid_count", + "burn_uid", + "burn_share", + "uid_weights", + "positive_raw_replay", + "publication_id", + "publication_phase", + "target_event", ) TEXT_LIMITS = { "ts": 64, @@ -184,6 +196,7 @@ ) SAFE_STAGE = re.compile(r"^[a-z][a-z0-9_-]{0,31}$") SAFE_MODES = frozenset({"thin", "full_provenance"}) +PUBLICATION_ID = re.compile(r"^[0-9a-f]{32}$") ALLOWED_RAW_STATUSES = { event: frozenset({status}) for event, status in EVENT_STATUS.items() } @@ -212,14 +225,23 @@ def scrub(value: str, limit: int) -> str: return text[:limit] -def public_detail(event: str, raw: Any) -> str | None: +def public_detail( + event: str, + raw: Any, + document: dict[str, Any] | None = None, +) -> str | None: """Convert private diagnostics into fixed public templates.""" + document = document or {} detail = raw if isinstance(raw, str) else "" if event == "STARTUP": - match = STARTUP_DETAIL.fullmatch(detail) - if match is None: - return None - modes = match.groups() + authority = document.get("authority") + provenance_mode = document.get("provenance_mode") + modes = (authority, provenance_mode) + if modes not in STARTUP_MODE_PAIRS: + match = STARTUP_DETAIL.fullmatch(detail) + if match is None: + return None + modes = match.groups() return { ("thin", "off"): "thin authority started; provenance audit is off", ( @@ -236,20 +258,31 @@ def public_detail(event: str, raw: Any) -> str | None: if event == "VECTOR_REJECTED": return "signed vector was rejected; nothing was submitted" if event in ("WEIGHTS_DRY_RUN", "WEIGHTS_SUBMITTED"): - match = WEIGHT_DETAIL.fullmatch(detail) - if match is None: + boundary = parse_weight_boundary_record(document) + if boundary is None: + boundary = parse_weight_boundary(detail) + if boundary is None: return ( "weight result recorded; detailed values remain in validator-local logs" ) - authority, count, burn_uid, burn_share, vector = match.groups() + authority = boundary["authority"] + count = boundary["uid_count"] + burn_uid = boundary["burn_uid"] + burn_share = boundary["burn_share"] + vector = ",".join( + f"{uid}:{weight:.6f}" + for uid, weight in sorted( + boundary["uid_weights"].items(), key=lambda item: int(item[0]) + ) + ) action = ( "submitted" if event == "WEIGHTS_SUBMITTED" else "verified without a chain write" ) burn = ( - f" burn_uid={burn_uid} burn_share={burn_share}" - if burn_uid and burn_share + f" burn_uid={burn_uid} burn_share={burn_share:.6f}" + if burn_uid is not None and burn_share is not None else "" ) return ( @@ -260,7 +293,7 @@ def public_detail(event: str, raw: Any) -> str | None: if event == "PROVENANCE_AUDIT_NOT_PROVEN": replay = ( "positive raw Intel TDX evidence replayed; " - if "positive raw evidence replayed for " in detail + if document.get("positive_raw_replay") is True else "" ) return replay + "whole-epoch FULL assurance is not established" @@ -299,9 +332,66 @@ def public_detail(event: str, raw: Any) -> str | None: "the validator tick failed; a write may have finalized, so inspect " "the named extrinsic and durable attempt journal before recovery" ) + if event == "STATUS_PUBLICATION_PENDING": + return "validator status transition did not commit; current state is not proven" return None +def parse_weight_boundary_record(document: Any) -> dict[str, Any] | None: + """Validate the structured weight boundary written by EventLogger.""" + if not isinstance(document, dict): + return None + authority = document.get("authority") + count = document.get("uid_count") + burn_uid = document.get("burn_uid") + burn_share = document.get("burn_share") + weights = document.get("uid_weights") + if ( + authority not in SAFE_MODES + or isinstance(count, bool) + or not isinstance(count, int) + or count < 0 + or not isinstance(weights, dict) + or len(weights) != count + or isinstance(burn_uid, bool) + or (burn_uid is not None and (not isinstance(burn_uid, int) or burn_uid < 0)) + or isinstance(burn_share, bool) + or ( + burn_share is not None + and ( + not isinstance(burn_share, (int, float)) + or not math.isfinite(float(burn_share)) + or not 0 <= float(burn_share) <= 1 + ) + ) + ): + return None + clean_weights: dict[str, float] = {} + for raw_uid, raw_weight in weights.items(): + if ( + not isinstance(raw_uid, str) + or re.fullmatch(r"0|[1-9][0-9]*", raw_uid) is None + ): + return None + if ( + isinstance(raw_weight, bool) + or not isinstance(raw_weight, (int, float)) + or not math.isfinite(float(raw_weight)) + or not 0 <= float(raw_weight) <= 1 + ): + return None + clean_weights[str(int(raw_uid))] = float(raw_weight) + if len(clean_weights) != len(weights): + return None + return { + "authority": authority, + "uid_count": count, + "burn_uid": burn_uid, + "burn_share": float(burn_share) if burn_share is not None else None, + "uid_weights": clean_weights, + } + + def parse_weight_boundary(raw: Any) -> dict[str, Any] | None: if not isinstance(raw, str): return None @@ -396,18 +486,53 @@ def clean_event(document: Any) -> dict[str, Any] | None: return None if event not in ALLOWED_EVENTS: return None + publication_id = document.get("publication_id") + publication_phase = document.get("publication_phase") + if publication_id is not None or publication_phase is not None: + if ( + not isinstance(publication_id, str) + or PUBLICATION_ID.fullmatch(publication_id) is None + or publication_phase not in {"PENDING", "COMMITTED"} + ): + return None + target_event = document.get("target_event") + if event == "STATUS_PUBLICATION_PENDING": + if ( + publication_phase != "PENDING" + or not isinstance(target_event, str) + or re.fullmatch(r"[A-Z][A-Z0-9_]{2,63}", target_event) is None + ): + return None + elif publication_phase == "PENDING" or target_event is not None: + return None startup_modes: tuple[str, str] | None = None if event == "STARTUP": - detail = document.get("detail") - match = STARTUP_DETAIL.fullmatch(detail) if isinstance(detail, str) else None - if match is None: - return None - startup_modes = match.groups() + authority = document.get("authority") + provenance_mode = document.get("provenance_mode") + startup_modes = (authority, provenance_mode) + if startup_modes not in STARTUP_MODE_PAIRS: + # Only legacy, unfenced status rows may derive these fixed values + # from the old strict template. A new COMMITTED row without the + # structured pair is invalid and leaves its PENDING fence active. + if publication_phase is not None: + return None + detail = document.get("detail") + match = ( + STARTUP_DETAIL.fullmatch(detail) if isinstance(detail, str) else None + ) + if match is None: + return None + startup_modes = match.groups() if ( startup_modes not in STARTUP_MODE_PAIRS or document.get("mode") != startup_modes[0] - or document.get("authority") != startup_modes[0] - or document.get("provenance_mode") != startup_modes[1] + or ( + publication_phase is not None + and ( + document.get("authority") != startup_modes[0] + or document.get("provenance_mode") != startup_modes[1] + ) + ) ): return None raw_status = document.get("status") @@ -431,6 +556,16 @@ def clean_event(document: Any) -> dict[str, Any] | None: continue clean[key] = min(86_400_000.0, max(0, round(float(value), 3))) continue + if key in { + "authority", + "provenance_mode", + "uid_count", + "burn_uid", + "burn_share", + "uid_weights", + "positive_raw_replay", + }: + continue if not isinstance(value, str): continue if key == "event": @@ -452,12 +587,18 @@ def clean_event(document: Any) -> dict[str, Any] | None: continue clean[key] = value continue + if key in {"publication_id", "publication_phase", "target_event"}: + clean[key] = value + continue clean[key] = scrub(value, TEXT_LIMITS[key]) clean["status"] = raw_status if startup_modes is not None: clean["authority"] = startup_modes[0] clean["provenance_mode"] = startup_modes[1] - detail = public_detail(event, document.get("detail")) + positive_raw_replay = document.get("positive_raw_replay") + if event == "PROVENANCE_AUDIT_NOT_PROVEN" and isinstance(positive_raw_replay, bool): + clean["positive_raw_replay"] = positive_raw_replay + detail = public_detail(event, document.get("detail"), document) if detail: clean["detail"] = detail[: TEXT_LIMITS["detail"]] if event in ( @@ -465,7 +606,9 @@ def clean_event(document: Any) -> dict[str, Any] | None: "WEIGHTS_SUBMITTED", "PENDING_RECEIPT_RECOVERED", ): - boundary = parse_weight_boundary(document.get("detail")) + boundary = parse_weight_boundary_record(document) + if boundary is None: + boundary = parse_weight_boundary(document.get("detail")) if boundary is not None: clean.update(boundary) remediation = EVENT_REMEDIATION.get(event) @@ -500,7 +643,8 @@ def tail_events() -> list[dict[str, Any]]: if offset: _discarded, _separator, payload = payload.partition(b"\n") lines = payload.splitlines() - events: deque[dict[str, Any]] = deque(maxlen=MAX_EVENTS) + parsed: list[dict[str, Any]] = [] + committed_pairs: set[tuple[str, str]] = set() for raw in lines: try: document = json.loads(raw) @@ -508,7 +652,40 @@ def tail_events() -> list[dict[str, Any]]: continue event = clean_event(document) if event is not None: - events.append(event) + parsed.append(event) + # Unknown event codes are intentionally absent from the public + # allowlist, but their valid COMMITTED row must still close the writer's + # preceding fence. For allowlisted events, only a row that passed the + # complete public validation closes it. A malformed STARTUP or weight + # transition therefore remains fail-closed. + if isinstance(document, dict): + raw_event = document.get("event") + publication_id = document.get("publication_id") + if ( + document.get("publication_phase") == "COMMITTED" + and isinstance(raw_event, str) + and re.fullmatch(r"[A-Z][A-Z0-9_]{2,63}", raw_event) is not None + and isinstance(publication_id, str) + and PUBLICATION_ID.fullmatch(publication_id) is not None + and (raw_event not in ALLOWED_EVENTS or event is not None) + ): + committed_pairs.add((publication_id, raw_event)) + + # Remove a PENDING fence only when its exact COMMITTED partner is present. + # An unmatched fence is itself public evidence that the raw/status + # transition was interrupted and must remain visible as NOT_PROVEN. + events: deque[dict[str, Any]] = deque(maxlen=MAX_EVENTS) + for event in parsed: + if ( + event.get("publication_phase") == "PENDING" + and ( + str(event.get("publication_id")), + str(event.get("target_event")), + ) + in committed_pairs + ): + continue + events.append(event) return list(events) @@ -625,14 +802,27 @@ def event_age_seconds(event: dict[str, Any] | None, now: datetime) -> float | No def build_status(events: list[dict[str, Any]]) -> dict[str, Any]: index = read_public_json(INDEX) release = read_signed_release() - provenance = latest_matching(events, ("PROVENANCE_",)) - rewarded_set = latest_matching(events, ("LAUNCH_REWARDED_SET_GATE_",)) startup = latest_matching(events, ("STARTUP",)) + # A fresh STARTUP is a session boundary. Observations from a previous + # process must not make a restarted validator look healthy before its new + # session has produced the corresponding evidence again. + if startup is not None: + startup_index = max( + index for index, event in enumerate(events) if event is startup + ) + current_events = events[startup_index:] + else: + current_events = events + publication_pending = latest_matching( + current_events, ("STATUS_PUBLICATION_PENDING",) + ) + provenance = latest_matching(current_events, ("PROVENANCE_",)) + rewarded_set = latest_matching(current_events, ("LAUNCH_REWARDED_SET_GATE_",)) # Authority status is about an observed live submission, not merely a # signed vector passing preflight or a no-write canary. Keep the last # successful live submission until a later tick records a real failure. authority = latest_matching( - events, + current_events, ( "WEIGHTS_SUBMITTED", "PENDING_RECEIPT_CONTRADICTION", @@ -658,7 +848,7 @@ def build_status(events: list[dict[str, Any]]) -> dict[str, Any]: authority_event = str((authority or {}).get("event", "")) provenance_event = str((provenance or {}).get("event", "")) mode_event = latest_matching( - events, + current_events, ("WEIGHTS_SUBMITTED", "WEIGHTS_DRY_RUN", "PENDING_RECEIPT_RECOVERED"), ) if startup is not None: @@ -672,7 +862,9 @@ def build_status(events: list[dict[str, Any]]) -> dict[str, Any]: else: current_authority_mode = "NOT_PROVEN" current_provenance_mode = "NOT_PROVEN" - if authority_event in ( + if publication_pending is not None: + authority_status = "NOT_PROVEN" + elif authority_event in ( "PENDING_RECEIPT_CONTRADICTION", "TICK_FAILED", "VECTOR_REJECTED", @@ -685,17 +877,19 @@ def build_status(events: list[dict[str, Any]]) -> dict[str, Any]: else: authority_status = "NOT_PROVEN" provenance_status = EVENT_STATUS.get(provenance_event, "NOT_PROVEN") + if publication_pending is not None: + provenance_status = "NOT_PROVEN" if provenance_status not in ("PASS", "FAIL"): provenance_status = "NOT_PROVEN" - detail = str((provenance or {}).get("detail", "")) positive_replay = ( "PASS" - if provenance_fresh + if publication_pending is None + and provenance_fresh and ( provenance_event == "PROVENANCE_AUDIT_PASS" or ( provenance_event == "PROVENANCE_AUDIT_NOT_PROVEN" - and "positive raw Intel TDX evidence replayed" in detail + and (provenance or {}).get("positive_raw_replay") is True ) ) else "NOT_PROVEN" @@ -727,7 +921,9 @@ def build_status(events: list[dict[str, Any]]) -> dict[str, Any]: "netuid": 39, "authority": { "mode": current_authority_mode, - "status": authority_status if authority_fresh else "NOT_PROVEN", + "status": authority_status + if authority_fresh and publication_pending is None + else "NOT_PROVEN", "burn_share": "0.10" if authority_fresh and current_authority_mode == "thin" @@ -737,11 +933,17 @@ def build_status(events: list[dict[str, Any]]) -> dict[str, Any]: "age_seconds": round(authority_age, 3) if authority_age is not None else None, - "latest_event": (authority or {}).get("event") - if authority_fresh - else "STALE", + "latest_event": ( + "STATUS_PUBLICATION_PENDING" + if publication_pending is not None + else (authority or {}).get("event") + if authority_fresh + else "STALE" + ), "detail": ( - (authority or {}).get("detail") + (publication_pending or {}).get("detail") + if publication_pending is not None + else (authority or {}).get("detail") if authority_fresh else f"no live submission observed within {MAX_EVENT_AGE_SECONDS} seconds" ), @@ -750,7 +952,8 @@ def build_status(events: list[dict[str, Any]]) -> dict[str, Any]: "mode": current_provenance_mode, "rewarded_set_full": ( "PASS" - if rewarded_set_fresh + if publication_pending is None + and rewarded_set_fresh and rewarded_set and rewarded_set.get("event") == "LAUNCH_REWARDED_SET_GATE_PASS" else "NOT_PROVEN" @@ -763,17 +966,25 @@ def build_status(events: list[dict[str, Any]]) -> dict[str, Any]: else "NOT_PROVEN" ), "current_whole_epoch_full": ( - provenance_status if provenance_fresh else "NOT_PROVEN" + provenance_status + if provenance_fresh and publication_pending is None + else "NOT_PROVEN" ), "fresh": provenance_fresh, "age_seconds": round(provenance_age, 3) if provenance_age is not None else None, - "latest_event": (provenance or {}).get("event") - if provenance_fresh - else "STALE", + "latest_event": ( + "STATUS_PUBLICATION_PENDING" + if publication_pending is not None + else (provenance or {}).get("event") + if provenance_fresh + else "STALE" + ), "detail": ( - (provenance or {}).get("detail") + (publication_pending or {}).get("detail") + if publication_pending is not None + else (provenance or {}).get("detail") if provenance_fresh else f"no provenance result observed within {MAX_EVENT_AGE_SECONDS} seconds" ), diff --git a/tests/thin/test_status_sanitization.py b/tests/thin/test_status_sanitization.py index 4c14651c..e43a51d9 100644 --- a/tests/thin/test_status_sanitization.py +++ b/tests/thin/test_status_sanitization.py @@ -42,7 +42,13 @@ def test_arbitrary_event_fields_never_enter_the_status_stream(tmp_path): assert _mode(status) == 0o600 raw_record = json.loads(raw.read_text(encoding="utf-8")) - status_record = json.loads(status.read_text(encoding="utf-8")) + status_records = [ + json.loads(line) for line in status.read_text(encoding="utf-8").splitlines() + ] + assert status_records[0]["event"] == "STATUS_PUBLICATION_PENDING" + assert status_records[0]["status"] == "NOT_PROVEN" + status_record = status_records[-1] + assert status_record["publication_phase"] == "COMMITTED" assert raw_record["receipt_body"] == "SECRET-RECEIPT-PAYLOAD" assert raw_record["future_unreviewed_field"]["nested"] == "SECRET-FUTURE-VALUE" @@ -101,7 +107,10 @@ def test_group_readable_status_does_not_make_raw_journal_group_readable(tmp_path assert _mode(raw) == 0o600 assert _mode(status) == 0o640 - assert "arbitrary" not in json.loads(status.read_text(encoding="utf-8")) + status_records = [ + json.loads(line) for line in status.read_text(encoding="utf-8").splitlines() + ] + assert all("arbitrary" not in row for row in status_records) def test_existing_group_readable_raw_journal_is_refused_without_explicit_group( @@ -112,3 +121,54 @@ def test_existing_group_readable_raw_journal_is_refused_without_explicit_group( raw.chmod(0o640) with pytest.raises(ValueError, match="private \\(0600\\) without a reader group"): EventLogger(mode="thin", jsonl_path=str(raw), tty=None) + + +def test_raw_and_status_streams_cannot_alias_the_same_inode(tmp_path): + raw = tmp_path / "validator-events.jsonl" + raw.touch(mode=0o600) + with pytest.raises(ValueError, match="must be distinct"): + EventLogger( + mode="thin", + jsonl_path=str(raw), + status_path=str(raw), + tty=None, + ) + + +def test_event_stream_refuses_hardlinked_journal(tmp_path): + raw = tmp_path / "validator-events.jsonl" + raw.touch(mode=0o600) + alias = tmp_path / "alias.jsonl" + os.link(raw, alias) + with pytest.raises(ValueError, match="owner-controlled"): + EventLogger(mode="thin", jsonl_path=str(raw), tty=None) + + +def test_free_form_fields_and_embedded_hotkeys_never_enter_status(tmp_path): + raw = tmp_path / "validator-events.jsonl" + status = tmp_path / "validator-status.jsonl" + hotkey = "5G3qVaXzKMPDm5AJ3dpzbpUC27kpccBvDwzSWXrq8M6qMmbC" + logger = EventLogger( + mode="thin", + jsonl_path=str(raw), + status_path=str(status), + tty=None, + ) + logger.event( + "TICK_FAILED", + stage="result", + status="FAIL", + hotkey=hotkey, + artifact=f"private-{hotkey}", + detail=f"failure for {hotkey}", + remediation=f"inspect {hotkey}", + nested={"identifier": hotkey}, + ) + logger.close() + + public = status.read_text(encoding="utf-8") + assert hotkey not in public + assert "failure for" not in public + assert "inspect" not in public + assert "nested" not in public + assert "artifact" not in json.loads(public.splitlines()[-1]) diff --git a/tests/thin/test_status_stream_migration.py b/tests/thin/test_status_stream_migration.py new file mode 100644 index 00000000..65fd2018 --- /dev/null +++ b/tests/thin/test_status_stream_migration.py @@ -0,0 +1,83 @@ +"""Upgrade migration keeps both legacy raw journals private.""" + +from __future__ import annotations + +import os +import stat +from pathlib import Path +from types import SimpleNamespace + +import pytest + +from scripts import migrate_sn39_status_stream as migration + + +def _mode(path: Path) -> int: + return stat.S_IMODE(path.stat().st_mode) + + +def test_both_known_legacy_journals_migrate_from_0640_to_0600( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, +) -> None: + continuous = tmp_path / "continuous.jsonl" + launch = tmp_path / "launch.jsonl" + for path in (continuous, launch): + path.write_text("private\n", encoding="utf-8") + path.chmod(0o640) + + monkeypatch.setattr(migration, "ROOT_UID", os.geteuid()) + monkeypatch.setattr(migration, "RAW_JOURNALS", (continuous, launch)) + monkeypatch.setattr( + migration.pwd, + "getpwnam", + lambda _name: SimpleNamespace(pw_uid=os.geteuid()), + ) + monkeypatch.setattr(migration, "require_writers_stopped", lambda: None) + + assert migration.main() == 0 + assert _mode(continuous) == 0o600 + assert _mode(launch) == 0o600 + assert continuous.read_text(encoding="utf-8") == "private\n" + assert launch.read_text(encoding="utf-8") == "private\n" + + +def test_migration_refuses_symlink_hardlink_and_unreviewed_mode(tmp_path: Path) -> None: + target = tmp_path / "target.jsonl" + target.write_text("private", encoding="utf-8") + target.chmod(0o640) + link = tmp_path / "link.jsonl" + link.symlink_to(target) + with pytest.raises(migration.MigrationError, match="opened safely"): + migration.privatize_journal(link, expected_uid=os.geteuid()) + + alias_directory = tmp_path / "alias-directory" + alias_directory.symlink_to(tmp_path, target_is_directory=True) + with pytest.raises(migration.MigrationError, match="directory cannot be opened"): + migration.privatize_journal( + alias_directory / target.name, + expected_uid=os.geteuid(), + ) + + hardlink = tmp_path / "hardlink.jsonl" + os.link(target, hardlink) + with pytest.raises(migration.MigrationError, match="single-linked"): + migration.privatize_journal(target, expected_uid=os.geteuid()) + hardlink.unlink() + + target.chmod(0o644) + with pytest.raises(migration.MigrationError, match="0600/0640"): + migration.privatize_journal(target, expected_uid=os.geteuid()) + + +def test_migration_refuses_while_any_writer_is_not_stopped( + monkeypatch: pytest.MonkeyPatch, +) -> None: + states = iter(("inactive\n", "active\n")) + monkeypatch.setattr( + migration.subprocess, + "run", + lambda *_args, **_kwargs: SimpleNamespace(stdout=next(states), returncode=0), + ) + with pytest.raises(migration.MigrationError, match="not stopped"): + migration.require_writers_stopped() From 29e854f57068d1761e86d2c02ff0382fed85f916 Mon Sep 17 00:00:00 2001 From: Fred E <7602667+wallscaler@users.noreply.github.com> Date: Sun, 2 Aug 2026 04:12:17 -0400 Subject: [PATCH 3/4] fix: commit SN39 public status generations Publish a fail-closed PENDING marker before both event views and a digest-bound COMMITTED marker after them so interrupted refreshes cannot retain a stale PASS. Close fence metadata to reviewed categories, validate private-event commits, and keep the status regressions in required launch CI. --- .github/workflows/two-mode-provenance.yml | 5 +- docs/SN39_LAUNCH_CUTOVER_20260726.md | 5 +- docs/SN39_MAINNET_RELEASE_20260724.md | 11 + scaffold/events.py | 33 ++- .../tests/test_sn39_public_status.py | 156 +++++++++++++ scripts/publish_sn39_validator_status.py | 211 ++++++++++++++++-- tests/thin/test_status_sanitization.py | 58 +++++ tests/thin/test_status_stream_contract.py | 11 + 8 files changed, 462 insertions(+), 28 deletions(-) diff --git a/.github/workflows/two-mode-provenance.yml b/.github/workflows/two-mode-provenance.yml index e40bd013..67ba24bc 100644 --- a/.github/workflows/two-mode-provenance.yml +++ b/.github/workflows/two-mode-provenance.yml @@ -136,7 +136,10 @@ jobs: scaffold/publisher/tests/test_legacy_chain_guard.py \ scaffold/publisher/tests/test_mechanism_weightset.py \ scaffold/publisher/tests/test_weight_vector_scope.py \ - tests/thin/test_validator.py + tests/thin/test_validator.py \ + tests/thin/test_status_sanitization.py \ + tests/thin/test_status_stream_contract.py \ + tests/thin/test_status_stream_migration.py - name: Reproduction imports run: | "$SN39_VENV/bin/python" - <<'PY' diff --git a/docs/SN39_LAUNCH_CUTOVER_20260726.md b/docs/SN39_LAUNCH_CUTOVER_20260726.md index 23c8ea4e..7dce9b6f 100644 --- a/docs/SN39_LAUNCH_CUTOVER_20260726.md +++ b/docs/SN39_LAUNCH_CUTOVER_20260726.md @@ -560,7 +560,10 @@ Apply in this order. Each step is independently reversible. projection, then confirm a real execution in `journalctl -u cathedral-sn39-public-status`. `ConditionPathExists=` skips a missing projection without making the unit fail, so `systemctl status` - alone is insufficient. + alone is insufficient. Treat `logs/status.json` as the commit marker: + require `publication.phase=COMMITTED`, verify both published event-view + digests, then re-read the same committed generation. `PENDING`, a digest + mismatch, or a changed generation is `NOT_PROVEN`. 11. **Leave the three validator units disabled and inactive** until the launch window. Their single-writer guards are unchanged: each names the other SN39 writers in `Conflicts=` and refuses to start via `ExecStartPre` while any of diff --git a/docs/SN39_MAINNET_RELEASE_20260724.md b/docs/SN39_MAINNET_RELEASE_20260724.md index ab523643..98320caf 100644 --- a/docs/SN39_MAINNET_RELEASE_20260724.md +++ b/docs/SN39_MAINNET_RELEASE_20260724.md @@ -29,6 +29,17 @@ stream are published at: - `https://api.cathedral.computer/v1/evidence/logs/validator-events.jsonl` - `https://api.cathedral.computer/v1/evidence/index.json` +`logs/status.json` is the commit marker for both event views. A reader must: + +1. require `publication.phase == "COMMITTED"`; +2. fetch both event views and verify their published SHA-256 digests; +3. fetch `status.json` again and require the same committed generation. + +`PENDING`, a digest mismatch, or a changed generation is `NOT_PROVEN`. The +publisher replaces a fail-closed PENDING marker first, then both views, then +the COMMITTED marker last. A crash at any intermediate write therefore cannot +leave an older PASS beside a newer unmatched event fence. + ## Current operational blockers The software candidate is not the final launch release, and this document does diff --git a/scaffold/events.py b/scaffold/events.py index 77f1454f..1d6712b6 100644 --- a/scaffold/events.py +++ b/scaffold/events.py @@ -212,6 +212,32 @@ def _redact(value: str) -> str: } ) +_STATUS_FENCE_TARGETS = frozenset( + { + "STARTUP", + "WEIGHT_RESULT", + "PROVENANCE_RESULT", + "RECEIPT_RECOVERY", + "VALIDATOR_RESULT", + "PRIVATE_EVENT", + } +) + + +def status_fence_target(event: str) -> str: + """Map a caller event code to a closed, non-sensitive fence category.""" + if event == "STARTUP": + return "STARTUP" + if event in {"WEIGHTS_DRY_RUN", "WEIGHTS_SUBMITTED"}: + return "WEIGHT_RESULT" + if event.startswith("PROVENANCE_") or event.startswith("LAUNCH_REWARDED_SET_GATE_"): + return "PROVENANCE_RESULT" + if event.startswith("PENDING_RECEIPT_"): + return "RECEIPT_RECOVERY" + if event in {"TICK_FAILED", "VECTOR_ACCEPTED", "VECTOR_REJECTED"}: + return "VALIDATOR_RESULT" + return "PRIVATE_EVENT" + def _structured_status_value(field: str, value: Any) -> Any | None: """Return a safe public value, or ``None`` when it is not admissible.""" @@ -340,6 +366,7 @@ def sanitized_status_record( if publication_id is not None: clean["publication_id"] = publication_id clean["publication_phase"] = "COMMITTED" + clean["target_event"] = status_fence_target(event) return clean @@ -355,7 +382,7 @@ def pending_status_record( "status": NOT_PROVEN, "publication_id": publication_id, "publication_phase": "PENDING", - "target_event": record["event"], + "target_event": status_fence_target(str(record["event"])), } @@ -383,7 +410,9 @@ def __init__( tty: IO[str] | None = None, color: bool | None = None, ) -> None: - self.mode = _neutralize(mode)[:32] + if mode not in _SAFE_AUTHORITIES: + raise ValueError("event logger mode must be a reviewed authority mode") + self.mode = mode self._jsonl = jsonl self._jsonl_file: IO[str] | None = None self._status_file: IO[str] | None = None diff --git a/scaffold/publisher/tests/test_sn39_public_status.py b/scaffold/publisher/tests/test_sn39_public_status.py index 07dfcbe0..d7426e67 100644 --- a/scaffold/publisher/tests/test_sn39_public_status.py +++ b/scaffold/publisher/tests/test_sn39_public_status.py @@ -345,6 +345,44 @@ def test_unknown_private_event_commit_closes_its_publication_fence( assert status.tail_events() == [] +def test_unknown_private_event_commit_requires_safe_common_fields( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, +) -> None: + public = tmp_path / "status.jsonl" + publication_id = "c" * 32 + rows = [ + { + "ts": _timestamp(), + "event": "STATUS_PUBLICATION_PENDING", + "stage": "status", + "mode": "thin", + "status": "NOT_PROVEN", + "publication_id": publication_id, + "publication_phase": "PENDING", + "target_event": "PRIVATE_EVENT", + }, + { + "ts": _timestamp(), + "event": "CHAIN_SUBMITTED", + "stage": "submit", + "mode": "caller-controlled-mode", + "status": "PASS", + "publication_id": publication_id, + "publication_phase": "COMMITTED", + "target_event": "PRIVATE_EVENT", + }, + ] + public.write_text( + "\n".join(json.dumps(row) for row in rows) + "\n", + encoding="utf-8", + ) + monkeypatch.setattr(status, "SOURCE", public) + + published = status.tail_events() + assert [row["event"] for row in published] == ["STATUS_PUBLICATION_PENDING"] + + def test_malformed_or_mismatched_commit_does_not_clear_pending_fence( tmp_path: Path, monkeypatch: pytest.MonkeyPatch, @@ -370,6 +408,7 @@ def test_malformed_or_mismatched_commit_does_not_clear_pending_fence( "status": "PASS", "publication_id": publication_id, "publication_phase": "COMMITTED", + "target_event": "VALIDATOR_RESULT", }, { "ts": _timestamp(), @@ -379,6 +418,7 @@ def test_malformed_or_mismatched_commit_does_not_clear_pending_fence( "status": "INFO", "publication_id": publication_id, "publication_phase": "COMMITTED", + "target_event": "STARTUP", # The required structured mode pair is intentionally absent. }, ] @@ -440,6 +480,18 @@ def test_unexpected_fence_field_cannot_be_used_as_free_form_public_text() -> Non document["target_event"] = "5G3qVaXzKMPDm5AJ3dpzbpUC27kpccBvDwzSWXrq8M6qMmbC" assert status.clean_event(document) is None + pending = { + "ts": _timestamp(), + "event": "STATUS_PUBLICATION_PENDING", + "stage": "status", + "mode": "caller-controlled-mode", + "status": "NOT_PROVEN", + "publication_id": "b" * 32, + "publication_phase": "PENDING", + "target_event": "PRIVATE_EVENT", + } + assert status.clean_event(pending) is None + def test_public_status_is_time_bounded() -> None: stale = status.clean_event( @@ -670,6 +722,110 @@ def test_publisher_emits_only_sanitized_bounded_outputs( "status.json", } assert all((path.stat().st_mode & 0o777) == 0o644 for path in logs.iterdir()) + published_status = json.loads((logs / "status.json").read_text(encoding="utf-8")) + publication = published_status["publication"] + assert publication["phase"] == "COMMITTED" + assert publication["events_jsonl_sha256"] == ( + "sha256:" + + hashlib.sha256((logs / "validator-events.jsonl").read_bytes()).hexdigest() + ) + assert publication["events_log_sha256"] == ( + "sha256:" + + hashlib.sha256((logs / "validator-events.log").read_bytes()).hexdigest() + ) + + +def test_interrupted_public_generation_replaces_stale_pass_with_pending( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, +) -> None: + raw = tmp_path / "raw.jsonl" + source = tmp_path / "status-source.jsonl" + logger = EventLogger( + mode="thin", + jsonl_path=str(raw), + status_path=str(source), + tty=None, + ) + logger.event( + "STARTUP", + stage="startup", + status="INFO", + authority="thin", + provenance_mode="shadow", + ) + logger.event( + "WEIGHTS_SUBMITTED", + stage="submit", + status="PASS", + authority="thin", + uid_count=2, + burn_uid=204, + burn_share=0.1, + uid_weights={"163": 0.9, "204": 0.1}, + ) + root = tmp_path / "public" + logs = root / "logs" + logs.mkdir(parents=True, mode=0o755) + (root / "index.json").write_text('{"recent":[]}', encoding="utf-8") + (root / "index.json").chmod(0o600) + monkeypatch.setattr(status, "SOURCE", source) + monkeypatch.setattr(status, "INDEX", root / "index.json") + monkeypatch.setattr(status, "RELEASE", root / "release.json") + monkeypatch.setattr(status, "LOG_ROOT", logs) + + assert status.main() == 0 + prior = json.loads((logs / "status.json").read_text(encoding="utf-8")) + assert prior["publication"]["phase"] == "COMMITTED" + assert prior["authority"]["status"] == "PASS" + + durable_write = event_stream._durable_jsonl_write + + def fail_status_commit(target, record): + if ( + target is logger._status_file + and record.get("publication_phase") == "COMMITTED" + ): + raise OSError("injected source status commit failure") + durable_write(target, record) + + monkeypatch.setattr(event_stream, "_durable_jsonl_write", fail_status_commit) + with pytest.raises(OSError, match="injected source"): + logger.event( + "TICK_FAILED", + stage="result", + status="FAIL", + detail="validator-local failure", + ) + logger.close() + + atomic_write = status.atomic_write + + def fail_between_public_views(path: Path, data: bytes) -> None: + if path.name == "validator-events.log": + raise OSError("injected public view failure") + atomic_write(path, data) + + monkeypatch.setattr(status, "atomic_write", fail_between_public_views) + with pytest.raises(OSError, match="injected public view"): + status.main() + + interrupted = json.loads((logs / "status.json").read_text(encoding="utf-8")) + assert interrupted["publication"]["phase"] == "PENDING" + assert ( + interrupted["publication"]["generation"] != (prior["publication"]["generation"]) + ) + assert interrupted["authority"]["status"] == "NOT_PROVEN" + assert interrupted["authority"]["burn_share"] is None + assert interrupted["authority"]["latest_event"] == ("STATUS_PUBLICATION_PENDING") + assert interrupted["provenance"]["current_whole_epoch_full"] == "NOT_PROVEN" + assert ( + "trust event views only when phase is COMMITTED" + in (interrupted["publication"]["reader_rule"]) + ) + assert ( + b"STATUS_PUBLICATION_PENDING" in (logs / "validator-events.jsonl").read_bytes() + ) def test_output_directory_must_be_owner_controlled( diff --git a/scripts/publish_sn39_validator_status.py b/scripts/publish_sn39_validator_status.py index 49953c4a..430b2770 100644 --- a/scripts/publish_sn39_validator_status.py +++ b/scripts/publish_sn39_validator_status.py @@ -4,6 +4,7 @@ from __future__ import annotations import base64 +import copy import hashlib import json import math @@ -197,10 +198,21 @@ SAFE_STAGE = re.compile(r"^[a-z][a-z0-9_-]{0,31}$") SAFE_MODES = frozenset({"thin", "full_provenance"}) PUBLICATION_ID = re.compile(r"^[0-9a-f]{32}$") +FENCE_TARGETS = frozenset( + { + "STARTUP", + "WEIGHT_RESULT", + "PROVENANCE_RESULT", + "RECEIPT_RECOVERY", + "VALIDATOR_RESULT", + "PRIVATE_EVENT", + } +) ALLOWED_RAW_STATUSES = { event: frozenset({status}) for event, status in EVENT_STATUS.items() } ALLOWED_RAW_STATUSES["WEIGHTS_DRY_RUN"] = frozenset({"PASS", "FAIL"}) +SAFE_RAW_STATUSES = frozenset(EVENT_STATUS.values()) def scrub(value: str, limit: int) -> str: @@ -225,6 +237,21 @@ def scrub(value: str, limit: int) -> str: return text[:limit] +def fence_target_for_event(event: str) -> str: + """Mirror the writer's closed fence categories without caller text.""" + if event == "STARTUP": + return "STARTUP" + if event in {"WEIGHTS_DRY_RUN", "WEIGHTS_SUBMITTED"}: + return "WEIGHT_RESULT" + if event.startswith("PROVENANCE_") or event.startswith("LAUNCH_REWARDED_SET_GATE_"): + return "PROVENANCE_RESULT" + if event.startswith("PENDING_RECEIPT_"): + return "RECEIPT_RECOVERY" + if event in {"TICK_FAILED", "VECTOR_ACCEPTED", "VECTOR_REJECTED"}: + return "VALIDATOR_RESULT" + return "PRIVATE_EVENT" + + def public_detail( event: str, raw: Any, @@ -493,6 +520,7 @@ def clean_event(document: Any) -> dict[str, Any] | None: not isinstance(publication_id, str) or PUBLICATION_ID.fullmatch(publication_id) is None or publication_phase not in {"PENDING", "COMMITTED"} + or document.get("mode") not in SAFE_MODES ): return None target_event = document.get("target_event") @@ -500,10 +528,15 @@ def clean_event(document: Any) -> dict[str, Any] | None: if ( publication_phase != "PENDING" or not isinstance(target_event, str) - or re.fullmatch(r"[A-Z][A-Z0-9_]{2,63}", target_event) is None + or target_event not in FENCE_TARGETS ): return None - elif publication_phase == "PENDING" or target_event is not None: + elif publication_phase == "PENDING": + return None + elif publication_phase == "COMMITTED": + if target_event != fence_target_for_event(event): + return None + elif target_event is not None: return None startup_modes: tuple[str, str] | None = None if event == "STARTUP": @@ -660,16 +693,22 @@ def tail_events() -> list[dict[str, Any]]: # transition therefore remains fail-closed. if isinstance(document, dict): raw_event = document.get("event") + target_event = document.get("target_event") publication_id = document.get("publication_id") if ( document.get("publication_phase") == "COMMITTED" and isinstance(raw_event, str) and re.fullmatch(r"[A-Z][A-Z0-9_]{2,63}", raw_event) is not None + and target_event == fence_target_for_event(raw_event) and isinstance(publication_id, str) and PUBLICATION_ID.fullmatch(publication_id) is not None + and document.get("mode") in SAFE_MODES + and document.get("status") in SAFE_RAW_STATUSES + and isinstance(document.get("ts"), str) + and parse_event_time(document["ts"]) is not None and (raw_event not in ALLOWED_EVENTS or event is not None) ): - committed_pairs.add((publication_id, raw_event)) + committed_pairs.add((publication_id, target_event)) # Remove a PENDING fence only when its exact COMMITTED partner is present. # An unmatched fence is itself public evidence that the raw/status @@ -1014,30 +1053,130 @@ def build_status(events: list[dict[str, Any]]) -> dict[str, Any]: def atomic_write(path: Path, data: bytes) -> None: + directory_flags = ( + os.O_RDONLY + | getattr(os, "O_DIRECTORY", 0) + | getattr(os, "O_NOFOLLOW", 0) + | getattr(os, "O_CLOEXEC", 0) + ) try: - info = path.parent.lstat() + directory_descriptor = os.open(path.parent, directory_flags) except OSError as exc: raise RuntimeError("public log directory is unavailable") from exc - if ( - path.parent.is_symlink() - or not stat.S_ISDIR(info.st_mode) - or info.st_uid != os.geteuid() - or stat.S_IMODE(info.st_mode) & 0o022 - ): - raise RuntimeError("public log directory is not owner-controlled") - descriptor, temporary = tempfile.mkstemp(prefix=f".{path.name}.", dir=path.parent) try: - os.fchmod(descriptor, 0o644) - with os.fdopen(descriptor, "wb") as handle: - handle.write(data) - handle.flush() - os.fsync(handle.fileno()) - os.replace(temporary, path) - finally: + info = os.fstat(directory_descriptor) + if ( + not stat.S_ISDIR(info.st_mode) + or info.st_uid != os.geteuid() + or stat.S_IMODE(info.st_mode) & 0o022 + ): + raise RuntimeError("public log directory is not owner-controlled") + descriptor, temporary = tempfile.mkstemp( + prefix=f".{path.name}.", dir=path.parent + ) try: - os.unlink(temporary) - except FileNotFoundError: - pass + os.fchmod(descriptor, 0o644) + with os.fdopen(descriptor, "wb") as handle: + handle.write(data) + handle.flush() + os.fsync(handle.fileno()) + os.replace(temporary, path) + # Persist the rename before the next phase of the publication + # transaction. This makes the PENDING -> views -> COMMITTED order + # durable across a host crash, not merely atomic per pathname. + os.fsync(directory_descriptor) + finally: + try: + os.unlink(temporary) + except FileNotFoundError: + pass + finally: + os.close(directory_descriptor) + + +def _sha256_label(payload: bytes) -> str: + return "sha256:" + hashlib.sha256(payload).hexdigest() + + +def _publication_metadata( + *, + phase: str, + generation: str, + status_core: bytes, + events_jsonl: bytes, + events_log: bytes, +) -> dict[str, str]: + return { + "schema": "cathedral.sn39.status-publication.v1", + "phase": phase, + "generation": generation, + "status_core_sha256": _sha256_label(status_core), + "events_jsonl_sha256": _sha256_label(events_jsonl), + "events_log_sha256": _sha256_label(events_log), + "reader_rule": ( + "trust event views only when phase is COMMITTED, both view digests " + "match, and a second status read returns the same generation" + ), + } + + +def _publication_generation( + status_core: bytes, events_jsonl: bytes, events_log: bytes +) -> str: + digest = hashlib.sha256() + for payload in (status_core, events_jsonl, events_log): + digest.update(len(payload).to_bytes(8, "big")) + digest.update(payload) + return "sha256:" + digest.hexdigest() + + +def _status_publication_view( + status_document: dict[str, Any], + *, + phase: str, + generation: str, + status_core: bytes, + events_jsonl: bytes, + events_log: bytes, +) -> dict[str, Any]: + if phase not in {"PENDING", "COMMITTED"}: + raise ValueError("unknown public status publication phase") + result = copy.deepcopy(status_document) + result["publication"] = _publication_metadata( + phase=phase, + generation=generation, + status_core=status_core, + events_jsonl=events_jsonl, + events_log=events_log, + ) + if phase == "PENDING": + # The commit marker is replaced first. Until every view is durable and + # the COMMITTED marker is replaced last, no stale PASS is public. + authority = result.get("authority") + if isinstance(authority, dict): + authority["status"] = "NOT_PROVEN" + authority["burn_share"] = None + authority["latest_event"] = "STATUS_PUBLICATION_PENDING" + authority["detail"] = "public status generation is not committed" + provenance = result.get("provenance") + if isinstance(provenance, dict): + for field in ( + "rewarded_set_full", + "positive_tdx_raw_replay", + "current_whole_epoch_full", + ): + provenance[field] = "NOT_PROVEN" + provenance["latest_event"] = "STATUS_PUBLICATION_PENDING" + provenance["detail"] = "public status generation is not committed" + result["validity"] = ( + "NOT_PROVEN while publication.phase is PENDING. Event views must " + "not be trusted until the matching COMMITTED generation is visible." + ) + return result + + +def _status_bytes(document: dict[str, Any]) -> bytes: + return json.dumps(document, indent=2, sort_keys=True).encode("utf-8") + b"\n" def main() -> int: @@ -1060,15 +1199,39 @@ def main() -> int: if event.get("remediation"): fields.append("next: " + str(event["remediation"])) human_lines.append(" | ".join(fields)) + human_log = ("\n".join(human_lines) + ("\n" if human_lines else "")).encode("utf-8") status = build_status(events) + status_core = json.dumps(status, sort_keys=True, separators=(",", ":")).encode( + "utf-8" + ) + generation = _publication_generation(status_core, jsonl, human_log) + pending_status = _status_publication_view( + status, + phase="PENDING", + generation=generation, + status_core=status_core, + events_jsonl=jsonl, + events_log=human_log, + ) + committed_status = _status_publication_view( + status, + phase="COMMITTED", + generation=generation, + status_core=status_core, + events_jsonl=jsonl, + events_log=human_log, + ) + # status.json is the commit marker. Publish a fail-closed PENDING marker + # first, replace both views durably, then replace COMMITTED last. + atomic_write(LOG_ROOT / "status.json", _status_bytes(pending_status)) atomic_write(LOG_ROOT / "validator-events.jsonl", jsonl) atomic_write( LOG_ROOT / "validator-events.log", - ("\n".join(human_lines) + ("\n" if human_lines else "")).encode("utf-8"), + human_log, ) atomic_write( LOG_ROOT / "status.json", - json.dumps(status, indent=2, sort_keys=True).encode("utf-8") + b"\n", + _status_bytes(committed_status), ) print( json.dumps( diff --git a/tests/thin/test_status_sanitization.py b/tests/thin/test_status_sanitization.py index e43a51d9..de7f376d 100644 --- a/tests/thin/test_status_sanitization.py +++ b/tests/thin/test_status_sanitization.py @@ -9,6 +9,7 @@ import pytest +from scaffold import events as event_stream from scaffold.events import STATUS_FIELDS, EventLogger, sanitized_status_record @@ -47,6 +48,7 @@ def test_arbitrary_event_fields_never_enter_the_status_stream(tmp_path): ] assert status_records[0]["event"] == "STATUS_PUBLICATION_PENDING" assert status_records[0]["status"] == "NOT_PROVEN" + assert status_records[0]["target_event"] == "PRIVATE_EVENT" status_record = status_records[-1] assert status_record["publication_phase"] == "COMMITTED" assert raw_record["receipt_body"] == "SECRET-RECEIPT-PAYLOAD" @@ -172,3 +174,59 @@ def test_free_form_fields_and_embedded_hotkeys_never_enter_status(tmp_path): assert "inspect" not in public assert "nested" not in public assert "artifact" not in json.loads(public.splitlines()[-1]) + + +def test_event_logger_rejects_unreviewed_mode_before_opening_outputs(tmp_path): + raw = tmp_path / "validator-events.jsonl" + with pytest.raises(ValueError, match="reviewed authority mode"): + EventLogger( + mode="5G3qVaXzKMPDm5AJ3dpzbpUC27kpccBvDwzSWXrq8M6qMmbC", + jsonl_path=str(raw), + tty=None, + ) + assert not raw.exists() + + +def test_crash_fence_maps_caller_event_and_never_preserves_private_text( + tmp_path, + monkeypatch, +): + raw = tmp_path / "validator-events.jsonl" + status = tmp_path / "validator-status.jsonl" + logger = EventLogger( + mode="thin", + jsonl_path=str(raw), + status_path=str(status), + tty=None, + ) + caller_event = "CALLER_CONTROLLED_PRIVATE_EVENT" + hotkey = "5G3qVaXzKMPDm5AJ3dpzbpUC27kpccBvDwzSWXrq8M6qMmbC" + durable_write = event_stream._durable_jsonl_write + + def fail_commit(target, record): + if ( + target is logger._status_file + and record.get("publication_phase") == "COMMITTED" + ): + raise OSError("injected commit crash") + durable_write(target, record) + + monkeypatch.setattr(event_stream, "_durable_jsonl_write", fail_commit) + with pytest.raises(OSError, match="injected commit"): + logger.event( + caller_event, + stage="result", + status="FAIL", + detail=f"private result for {hotkey}", + hotkey=hotkey, + ) + logger.close() + + fence = json.loads(status.read_text(encoding="utf-8")) + assert fence["event"] == "STATUS_PUBLICATION_PENDING" + assert fence["mode"] == "thin" + assert fence["target_event"] == "PRIVATE_EVENT" + serialized = json.dumps(fence) + assert caller_event not in serialized + assert hotkey not in serialized + assert "private result" not in serialized diff --git a/tests/thin/test_status_stream_contract.py b/tests/thin/test_status_stream_contract.py index 8b0a15e7..278918da 100644 --- a/tests/thin/test_status_stream_contract.py +++ b/tests/thin/test_status_stream_contract.py @@ -17,6 +17,7 @@ _VALIDATOR_UNIT = _ROOT / "deploy" / "sn39" / "cathedral-validator-sn39.service" _LAUNCHER = _ROOT / "deploy" / "sn39" / "cathedral-sn39-release-launcher.py" _PUBLISHER = _ROOT / "scripts" / "publish_sn39_validator_status.py" +_REQUIRED_WORKFLOW = _ROOT / ".github" / "workflows" / "two-mode-provenance.yml" _spec = importlib.util.spec_from_file_location("_sn39_launcher_contract", _LAUNCHER) _launcher = importlib.util.module_from_spec(_spec) @@ -67,3 +68,13 @@ def test_validator_unit_documents_the_same_access_split(): re.M, ) assert not re.search(r"^Environment=CATHEDRAL_VALIDATOR_JSONL_GROUP=", unit, re.M) + + +def test_required_workflow_runs_every_status_stream_regression(): + workflow = _REQUIRED_WORKFLOW.read_text(encoding="utf-8") + for test_path in ( + "tests/thin/test_status_sanitization.py", + "tests/thin/test_status_stream_contract.py", + "tests/thin/test_status_stream_migration.py", + ): + assert test_path in workflow From 9cfb137dbbd2241cab8e5b292bf6649b35c2113f Mon Sep 17 00:00:00 2001 From: Fred E <7602667+wallscaler@users.noreply.github.com> Date: Sun, 2 Aug 2026 04:29:26 -0400 Subject: [PATCH 4/4] fix: bind SN39 status publications fail closed Bind each durable PENDING fence to the exact sanitized committed event, restrict public stage values to the reviewed enum, and leave a public generation pending when the source changes during publication. Add adversarial regressions for category substitution, reordered commits, stage leakage, and concurrent source updates. --- scaffold/events.py | 36 ++- .../tests/test_sn39_public_status.py | 257 +++++++++++++++--- scripts/publish_sn39_validator_status.py | 133 +++++++-- tests/thin/test_status_sanitization.py | 22 ++ 4 files changed, 389 insertions(+), 59 deletions(-) diff --git a/scaffold/events.py b/scaffold/events.py index 1d6712b6..7a8bab2d 100644 --- a/scaffold/events.py +++ b/scaffold/events.py @@ -18,6 +18,7 @@ from __future__ import annotations import grp +import hashlib import json import math import os @@ -194,6 +195,7 @@ def _redact(value: str) -> str: "publication_id", "publication_phase", "target_event", + "publication_commitment", ) STATUS_FIELDS = tuple( dict.fromkeys( @@ -203,6 +205,9 @@ def _redact(value: str) -> str: ) _SAFE_AUTHORITIES = frozenset({"thin", "full_provenance"}) +_SAFE_STAGES = frozenset( + {"launch", "map", "provenance", "result", "startup", "status", "submit", "verify"} +) _SAFE_PROVENANCE_MODES = frozenset({"off", "shadow", "authority"}) _STARTUP_MODE_PAIRS = frozenset( { @@ -239,6 +244,20 @@ def status_fence_target(event: str) -> str: return "PRIVATE_EVENT" +def status_record_commitment(record: Mapping[str, Any]) -> str: + """Bind a PENDING fence to the exact sanitized row that must commit it.""" + payload = dict(record) + payload.pop("publication_commitment", None) + canonical = json.dumps( + payload, + sort_keys=True, + separators=(",", ":"), + ensure_ascii=False, + allow_nan=False, + ).encode("utf-8") + return "sha256:" + hashlib.sha256(canonical).hexdigest() + + def _structured_status_value(field: str, value: Any) -> Any | None: """Return a safe public value, or ``None`` when it is not admissible.""" if field == "authority": @@ -332,8 +351,7 @@ def sanitized_status_record( "event": event, "stage": ( record["stage"] - if isinstance(record.get("stage"), str) - and re.fullmatch(r"[a-z][a-z0-9_-]{0,31}", record["stage"]) + if isinstance(record.get("stage"), str) and record["stage"] in _SAFE_STAGES else "unknown" ), "mode": record["mode"] @@ -371,7 +389,7 @@ def sanitized_status_record( def pending_status_record( - record: Mapping[str, Any], publication_id: str + record: Mapping[str, Any], publication_id: str, publication_commitment: str ) -> dict[str, Any]: """A durable fail-closed fence written before the corresponding raw event.""" return { @@ -383,6 +401,7 @@ def pending_status_record( "publication_id": publication_id, "publication_phase": "PENDING", "target_event": status_fence_target(str(record["event"])), + "publication_commitment": publication_commitment, } @@ -494,21 +513,28 @@ def event( record[key] = _scrub(value) if self._status_file is not None: publication_id = uuid.uuid4().hex + committed_status = sanitized_status_record( + record, publication_id=publication_id + ) + publication_commitment = status_record_commitment(committed_status) + committed_status["publication_commitment"] = publication_commitment # Fence first. If the process dies before the COMMITTED row, the # public reader sees NOT_PROVEN rather than retaining an older PASS. _durable_jsonl_write( self._status_file, - pending_status_record(record, publication_id), + pending_status_record(record, publication_id, publication_commitment), ) else: publication_id = None + committed_status = None for target in (self._jsonl, self._jsonl_file): if target is not None: _durable_jsonl_write(target, record) if self._status_file is not None: + assert committed_status is not None _durable_jsonl_write( self._status_file, - sanitized_status_record(record, publication_id=publication_id), + committed_status, ) self._write_tty(record) return record diff --git a/scaffold/publisher/tests/test_sn39_public_status.py b/scaffold/publisher/tests/test_sn39_public_status.py index d7426e67..4cff2811 100644 --- a/scaffold/publisher/tests/test_sn39_public_status.py +++ b/scaffold/publisher/tests/test_sn39_public_status.py @@ -52,6 +52,15 @@ def _startup(authority: str, provenance: str) -> dict[str, object]: } +def _bind_status_rows( + pending: dict[str, object], committed: dict[str, object] +) -> list[dict[str, object]]: + commitment = event_stream.status_record_commitment(committed) + pending["publication_commitment"] = commitment + committed["publication_commitment"] = commitment + return [pending, committed] + + def test_exact_launch_boundary_is_pass_but_all_burn_is_not_proven() -> None: launch = status.clean_event( _event( @@ -351,7 +360,7 @@ def test_unknown_private_event_commit_requires_safe_common_fields( ) -> None: public = tmp_path / "status.jsonl" publication_id = "c" * 32 - rows = [ + rows = _bind_status_rows( { "ts": _timestamp(), "event": "STATUS_PUBLICATION_PENDING", @@ -372,7 +381,7 @@ def test_unknown_private_event_commit_requires_safe_common_fields( "publication_phase": "COMMITTED", "target_event": "PRIVATE_EVENT", }, - ] + ) public.write_text( "\n".join(json.dumps(row) for row in rows) + "\n", encoding="utf-8", @@ -389,39 +398,44 @@ def test_malformed_or_mismatched_commit_does_not_clear_pending_fence( ) -> None: public = tmp_path / "status.jsonl" publication_id = "a" * 32 - rows = [ - { - "ts": _timestamp(), - "event": "STATUS_PUBLICATION_PENDING", - "stage": "status", - "mode": "thin", - "status": "NOT_PROVEN", - "publication_id": publication_id, - "publication_phase": "PENDING", - "target_event": "STARTUP", - }, - { - "ts": _timestamp(), - "event": "VECTOR_ACCEPTED", - "stage": "policy", - "mode": "thin", - "status": "PASS", - "publication_id": publication_id, - "publication_phase": "COMMITTED", - "target_event": "VALIDATOR_RESULT", - }, - { - "ts": _timestamp(), - "event": "STARTUP", - "stage": "startup", - "mode": "thin", - "status": "INFO", - "publication_id": publication_id, - "publication_phase": "COMMITTED", - "target_event": "STARTUP", - # The required structured mode pair is intentionally absent. - }, - ] + pending = { + "ts": _timestamp(), + "event": "STATUS_PUBLICATION_PENDING", + "stage": "status", + "mode": "thin", + "status": "NOT_PROVEN", + "publication_id": publication_id, + "publication_phase": "PENDING", + "target_event": "STARTUP", + } + vector_commit = { + "ts": _timestamp(), + "event": "VECTOR_ACCEPTED", + "stage": "policy", + "mode": "thin", + "status": "PASS", + "publication_id": publication_id, + "publication_phase": "COMMITTED", + "target_event": "VALIDATOR_RESULT", + } + malformed_startup = { + "ts": _timestamp(), + "event": "STARTUP", + "stage": "startup", + "mode": "thin", + "status": "INFO", + "publication_id": publication_id, + "publication_phase": "COMMITTED", + "target_event": "STARTUP", + # The required structured mode pair is intentionally absent. + } + expected_commitment = event_stream.status_record_commitment(malformed_startup) + pending["publication_commitment"] = expected_commitment + malformed_startup["publication_commitment"] = expected_commitment + vector_commit["publication_commitment"] = event_stream.status_record_commitment( + vector_commit + ) + rows = [pending, vector_commit, malformed_startup] public.write_text( "".join(json.dumps(row) + "\n" for row in rows), encoding="utf-8", @@ -438,6 +452,104 @@ def test_malformed_or_mismatched_commit_does_not_clear_pending_fence( assert document["authority"]["latest_event"] == "STATUS_PUBLICATION_PENDING" +def test_same_category_commit_cannot_clear_a_different_pending_event( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, +) -> None: + public = tmp_path / "status.jsonl" + publication_id = "d" * 32 + expected = { + "ts": _timestamp(), + "event": "PROVENANCE_AUDIT_FAIL", + "stage": "provenance", + "mode": "thin", + "status": "FAIL", + "publication_id": publication_id, + "publication_phase": "COMMITTED", + "target_event": "PROVENANCE_RESULT", + } + pending = { + "ts": _timestamp(), + "event": "STATUS_PUBLICATION_PENDING", + "stage": "status", + "mode": "thin", + "status": "NOT_PROVEN", + "publication_id": publication_id, + "publication_phase": "PENDING", + "target_event": "PROVENANCE_RESULT", + "publication_commitment": event_stream.status_record_commitment(expected), + } + different = { + "ts": _timestamp(), + "event": "PROVENANCE_AUDIT_PASS", + "stage": "provenance", + "mode": "thin", + "status": "PASS", + "publication_id": publication_id, + "publication_phase": "COMMITTED", + "target_event": "PROVENANCE_RESULT", + } + different["publication_commitment"] = event_stream.status_record_commitment( + different + ) + public.write_text( + "".join(json.dumps(row) + "\n" for row in (pending, different)), + encoding="utf-8", + ) + monkeypatch.setattr(status, "SOURCE", public) + + events = status.tail_events() + assert [event["event"] for event in events] == [ + "STATUS_PUBLICATION_PENDING", + "PROVENANCE_AUDIT_PASS", + ] + document = status.build_status(events) + assert document["authority"]["status"] == "NOT_PROVEN" + assert document["provenance"]["current_whole_epoch_full"] == "NOT_PROVEN" + + +def test_commit_before_pending_does_not_clear_later_fence( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, +) -> None: + public = tmp_path / "status.jsonl" + publication_id = "e" * 32 + committed = { + "ts": _timestamp(), + "event": "VECTOR_ACCEPTED", + "stage": "verify", + "mode": "thin", + "status": "PASS", + "publication_id": publication_id, + "publication_phase": "COMMITTED", + "target_event": "VALIDATOR_RESULT", + } + pending = { + "ts": _timestamp(), + "event": "STATUS_PUBLICATION_PENDING", + "stage": "status", + "mode": "thin", + "status": "NOT_PROVEN", + "publication_id": publication_id, + "publication_phase": "PENDING", + "target_event": "VALIDATOR_RESULT", + } + commitment = event_stream.status_record_commitment(committed) + committed["publication_commitment"] = commitment + pending["publication_commitment"] = commitment + public.write_text( + "".join(json.dumps(row) + "\n" for row in (committed, pending)), + encoding="utf-8", + ) + monkeypatch.setattr(status, "SOURCE", public) + + events = status.tail_events() + assert [event["event"] for event in events] == [ + "VECTOR_ACCEPTED", + "STATUS_PUBLICATION_PENDING", + ] + + def test_pending_receipt_contradiction_overrides_prior_thin_pass() -> None: startup = status.clean_event(_startup("thin", "shadow")) launch = status.clean_event( @@ -489,6 +601,7 @@ def test_unexpected_fence_field_cannot_be_used_as_free_form_public_text() -> Non "publication_id": "b" * 32, "publication_phase": "PENDING", "target_event": "PRIVATE_EVENT", + "publication_commitment": "sha256:" + "0" * 64, } assert status.clean_event(pending) is None @@ -828,6 +941,80 @@ def fail_between_public_views(path: Path, data: bytes) -> None: ) +def test_source_change_before_public_commit_leaves_generation_pending( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, +) -> None: + raw = tmp_path / "raw.jsonl" + source = tmp_path / "status-source.jsonl" + logger = EventLogger( + mode="thin", + jsonl_path=str(raw), + status_path=str(source), + tty=None, + ) + logger.event( + "STARTUP", + stage="startup", + status="INFO", + authority="thin", + provenance_mode="shadow", + ) + logger.event( + "WEIGHTS_SUBMITTED", + stage="submit", + status="PASS", + authority="thin", + uid_count=2, + burn_uid=204, + burn_share=0.1, + uid_weights={"163": 0.9, "204": 0.1}, + ) + logger.close() + root = tmp_path / "public" + logs = root / "logs" + logs.mkdir(parents=True, mode=0o755) + (root / "index.json").write_text('{"recent":[]}', encoding="utf-8") + monkeypatch.setattr(status, "SOURCE", source) + monkeypatch.setattr(status, "INDEX", root / "index.json") + monkeypatch.setattr(status, "RELEASE", root / "release.json") + monkeypatch.setattr(status, "LOG_ROOT", logs) + + atomic_write = status.atomic_write + inserted = False + + def append_source_fence_after_pending(path: Path, data: bytes) -> None: + nonlocal inserted + atomic_write(path, data) + if path.name != "status.json" or inserted: + return + document = json.loads(data) + if document["publication"]["phase"] != "PENDING": + return + inserted = True + pending = { + "ts": _timestamp(), + "event": "STATUS_PUBLICATION_PENDING", + "stage": "status", + "mode": "thin", + "status": "NOT_PROVEN", + "publication_id": "f" * 32, + "publication_phase": "PENDING", + "target_event": "VALIDATOR_RESULT", + "publication_commitment": "sha256:" + "1" * 64, + } + with source.open("a", encoding="utf-8") as handle: + handle.write(json.dumps(pending) + "\n") + + monkeypatch.setattr(status, "atomic_write", append_source_fence_after_pending) + assert status.main() == 1 + + published = json.loads((logs / "status.json").read_text(encoding="utf-8")) + assert published["publication"]["phase"] == "PENDING" + assert published["authority"]["status"] == "NOT_PROVEN" + assert published["authority"]["latest_event"] == "STATUS_PUBLICATION_PENDING" + + def test_output_directory_must_be_owner_controlled( tmp_path: Path, ) -> None: diff --git a/scripts/publish_sn39_validator_status.py b/scripts/publish_sn39_validator_status.py index 430b2770..c65d29d5 100644 --- a/scripts/publish_sn39_validator_status.py +++ b/scripts/publish_sn39_validator_status.py @@ -137,6 +137,7 @@ "publication_id", "publication_phase", "target_event", + "publication_commitment", ) TEXT_LIMITS = { "ts": 64, @@ -195,9 +196,12 @@ ("full_provenance", "authority"), } ) -SAFE_STAGE = re.compile(r"^[a-z][a-z0-9_-]{0,31}$") +SAFE_STAGES = frozenset( + {"launch", "map", "provenance", "result", "startup", "status", "submit", "verify"} +) SAFE_MODES = frozenset({"thin", "full_provenance"}) PUBLICATION_ID = re.compile(r"^[0-9a-f]{32}$") +PUBLICATION_COMMITMENT = re.compile(r"^sha256:[0-9a-f]{64}$") FENCE_TARGETS = frozenset( { "STARTUP", @@ -252,6 +256,30 @@ def fence_target_for_event(event: str) -> str: return "PRIVATE_EVENT" +def publication_commitment_matches(document: dict[str, Any]) -> bool: + """Verify the exact sanitized COMMITTED row bound by its PENDING fence.""" + claimed = document.get("publication_commitment") + if ( + not isinstance(claimed, str) + or PUBLICATION_COMMITMENT.fullmatch(claimed) is None + ): + return False + payload = dict(document) + payload.pop("publication_commitment", None) + try: + canonical = json.dumps( + payload, + sort_keys=True, + separators=(",", ":"), + ensure_ascii=False, + allow_nan=False, + ).encode("utf-8") + except (TypeError, ValueError): + return False + expected = "sha256:" + hashlib.sha256(canonical).hexdigest() + return claimed == expected + + def public_detail( event: str, raw: Any, @@ -515,14 +543,23 @@ def clean_event(document: Any) -> dict[str, Any] | None: return None publication_id = document.get("publication_id") publication_phase = document.get("publication_phase") + publication_commitment = document.get("publication_commitment") if publication_id is not None or publication_phase is not None: if ( not isinstance(publication_id, str) or PUBLICATION_ID.fullmatch(publication_id) is None or publication_phase not in {"PENDING", "COMMITTED"} or document.get("mode") not in SAFE_MODES + or not isinstance(publication_commitment, str) + or PUBLICATION_COMMITMENT.fullmatch(publication_commitment) is None + ): + return None + if publication_phase == "COMMITTED" and not publication_commitment_matches( + document ): return None + elif publication_commitment is not None: + return None target_event = document.get("target_event") if event == "STATUS_PUBLICATION_PENDING": if ( @@ -608,7 +645,7 @@ def clean_event(document: Any) -> dict[str, Any] | None: clean[key] = timestamp continue if key == "stage": - clean[key] = value if SAFE_STAGE.fullmatch(value) else "unknown" + clean[key] = value if value in SAFE_STAGES else "unknown" continue if key == "mode": clean[key] = value if value in SAFE_MODES else "unknown" @@ -620,7 +657,12 @@ def clean_event(document: Any) -> dict[str, Any] | None: continue clean[key] = value continue - if key in {"publication_id", "publication_phase", "target_event"}: + if key in { + "publication_id", + "publication_phase", + "target_event", + "publication_commitment", + }: clean[key] = value continue clean[key] = scrub(value, TEXT_LIMITS[key]) @@ -652,12 +694,19 @@ def clean_event(document: Any) -> dict[str, Any] | None: return clean -def tail_events() -> list[dict[str, Any]]: +SourceSnapshot = tuple[int, int, int, int] + + +def _source_snapshot(info: os.stat_result) -> SourceSnapshot: + return (info.st_dev, info.st_ino, info.st_size, info.st_mtime_ns) + + +def current_source_snapshot() -> SourceSnapshot | None: flags = os.O_RDONLY | getattr(os, "O_NOFOLLOW", 0) | getattr(os, "O_CLOEXEC", 0) try: descriptor = os.open(SOURCE, flags) except OSError: - return [] + return None try: info = os.fstat(descriptor) if ( @@ -665,27 +714,49 @@ def tail_events() -> list[dict[str, Any]]: or stat.S_IMODE(info.st_mode) & 0o002 or info.st_size < 0 ): - return [] + return None + return _source_snapshot(info) + except OSError: + return None + finally: + os.close(descriptor) + + +def _tail_events_with_snapshot() -> tuple[list[dict[str, Any]], SourceSnapshot | None]: + flags = os.O_RDONLY | getattr(os, "O_NOFOLLOW", 0) | getattr(os, "O_CLOEXEC", 0) + try: + descriptor = os.open(SOURCE, flags) + except OSError: + return [], None + try: + info = os.fstat(descriptor) + if ( + not stat.S_ISREG(info.st_mode) + or stat.S_IMODE(info.st_mode) & 0o002 + or info.st_size < 0 + ): + return [], None + source_snapshot = _source_snapshot(info) offset = max(0, info.st_size - MAX_SOURCE_BYTES) os.lseek(descriptor, offset, os.SEEK_SET) payload = os.read(descriptor, MAX_SOURCE_BYTES) except OSError: - return [] + return [], None finally: os.close(descriptor) if offset: _discarded, _separator, payload = payload.partition(b"\n") lines = payload.splitlines() - parsed: list[dict[str, Any]] = [] - committed_pairs: set[tuple[str, str]] = set() - for raw in lines: + parsed: list[tuple[int, dict[str, Any]]] = [] + committed_lines: dict[tuple[str, str, str], int] = {} + for line_number, raw in enumerate(lines): try: document = json.loads(raw) except (UnicodeDecodeError, json.JSONDecodeError): continue event = clean_event(document) if event is not None: - parsed.append(event) + parsed.append((line_number, event)) # Unknown event codes are intentionally absent from the public # allowlist, but their valid COMMITTED row must still close the writer's # preceding fence. For allowlisted events, only a row that passed the @@ -695,6 +766,7 @@ def tail_events() -> list[dict[str, Any]]: raw_event = document.get("event") target_event = document.get("target_event") publication_id = document.get("publication_id") + publication_commitment = document.get("publication_commitment") if ( document.get("publication_phase") == "COMMITTED" and isinstance(raw_event, str) @@ -702,30 +774,42 @@ def tail_events() -> list[dict[str, Any]]: and target_event == fence_target_for_event(raw_event) and isinstance(publication_id, str) and PUBLICATION_ID.fullmatch(publication_id) is not None + and isinstance(publication_commitment, str) + and publication_commitment_matches(document) and document.get("mode") in SAFE_MODES and document.get("status") in SAFE_RAW_STATUSES and isinstance(document.get("ts"), str) and parse_event_time(document["ts"]) is not None and (raw_event not in ALLOWED_EVENTS or event is not None) ): - committed_pairs.add((publication_id, target_event)) + committed_lines[ + (publication_id, target_event, publication_commitment) + ] = line_number # Remove a PENDING fence only when its exact COMMITTED partner is present. # An unmatched fence is itself public evidence that the raw/status # transition was interrupted and must remain visible as NOT_PROVEN. events: deque[dict[str, Any]] = deque(maxlen=MAX_EVENTS) - for event in parsed: + for line_number, event in parsed: if ( event.get("publication_phase") == "PENDING" - and ( - str(event.get("publication_id")), - str(event.get("target_event")), + and committed_lines.get( + ( + str(event.get("publication_id")), + str(event.get("target_event")), + str(event.get("publication_commitment")), + ), + -1, ) - in committed_pairs + > line_number ): continue events.append(event) - return list(events) + return list(events), source_snapshot + + +def tail_events() -> list[dict[str, Any]]: + return _tail_events_with_snapshot()[0] def read_public_bytes(path: Path) -> bytes | None: @@ -1180,7 +1264,7 @@ def _status_bytes(document: dict[str, Any]) -> bytes: def main() -> int: - events = tail_events() + events, source_snapshot = _tail_events_with_snapshot() jsonl = b"".join( json.dumps(event, sort_keys=True, separators=(",", ":")).encode("utf-8") + b"\n" for event in events @@ -1229,6 +1313,17 @@ def main() -> int: LOG_ROOT / "validator-events.log", human_log, ) + if current_source_snapshot() != source_snapshot: + print( + json.dumps( + { + "events_published": len(events), + "status": "source_changed_publication_left_pending", + }, + sort_keys=True, + ) + ) + return 1 atomic_write( LOG_ROOT / "status.json", _status_bytes(committed_status), diff --git a/tests/thin/test_status_sanitization.py b/tests/thin/test_status_sanitization.py index de7f376d..cc10ac8c 100644 --- a/tests/thin/test_status_sanitization.py +++ b/tests/thin/test_status_sanitization.py @@ -176,6 +176,28 @@ def test_free_form_fields_and_embedded_hotkeys_never_enter_status(tmp_path): assert "artifact" not in json.loads(public.splitlines()[-1]) +def test_caller_controlled_stage_cannot_cross_the_public_boundary(tmp_path): + raw = tmp_path / "validator-events.jsonl" + status = tmp_path / "validator-status.jsonl" + logger = EventLogger( + mode="thin", + jsonl_path=str(raw), + status_path=str(status), + tty=None, + ) + caller_value = "secret_token_0123456789" + logger.event( + "VECTOR_ACCEPTED", + stage=caller_value, + status="PASS", + ) + logger.close() + + public = status.read_text(encoding="utf-8") + assert caller_value not in public + assert json.loads(public.splitlines()[-1])["stage"] == "unknown" + + def test_event_logger_rejects_unreviewed_mode_before_opening_outputs(tmp_path): raw = tmp_path / "validator-events.jsonl" with pytest.raises(ValueError, match="reviewed authority mode"):