From c6542c51ba92875da683b905565ac9979327dfcf Mon Sep 17 00:00:00 2001 From: Pigbibi <20649888+Pigbibi@users.noreply.github.com> Date: Thu, 16 Jul 2026 05:07:03 +0800 Subject: [PATCH 1/2] Use canonical status bytes as feed evidence Co-Authored-By: Codex --- .../feed_status_canonical.py | 316 ++++++++++++++++++ tests/test_feed_status_canonical.py | 98 ++++++ 2 files changed, 414 insertions(+) create mode 100644 src/political_event_tracking_research/feed_status_canonical.py create mode 100644 tests/test_feed_status_canonical.py diff --git a/src/political_event_tracking_research/feed_status_canonical.py b/src/political_event_tracking_research/feed_status_canonical.py new file mode 100644 index 0000000..c078208 --- /dev/null +++ b/src/political_event_tracking_research/feed_status_canonical.py @@ -0,0 +1,316 @@ +from __future__ import annotations + +import hashlib +import json +import re +from collections.abc import Iterable, Mapping +from dataclasses import dataclass +from enum import Enum + + +STATUS_VERSION = "pert.feed_status_canonical.v1" +MAX_SAFE_JSON_INTEGER = 2**53 - 1 +MAX_ROWS_PER_FEED = 10_000 +_ROW_KEYS = ("item_id", "published_at", "source_type", "source_url", "author", "text") +_OUTCOME_KEYS = frozenset({"feed_id", "feed_url", "kind", "state", "rows", "error_code"}) +_ERROR_RE = re.compile(r"^[a-z][a-z0-9_]{0,63}$") +_DIGEST_RE = re.compile(r"^[0-9a-f]{64}$") +_WIRE_KEYS = frozenset( + { + "status_version", + "configured_feed_count", + "feed_count", + "successful_feed_count", + "failed_feed_count", + "quarantined_feed_count", + "accepted_row_count", + "rejected_row_count", + "publication_complete", + "eligible_for_live_publication", + "aggregate_row_digest", + "feeds", + } +) +_FEED_KEYS = frozenset( + { + "feed_id", + "feed_url", + "kind", + "state", + "accepted_row_count", + "rejected_row_count", + "row_digest", + "error_code", + } +) + + +class DecisionContractError(ValueError): + """Sanitized canonical status contract error.""" + + def __init__(self, code: str): + super().__init__(code) + self.code = code + + +class DecisionKind(str, Enum): + SUCCESS = "success" + QUARANTINE = "quarantine" + HARD_FAIL = "hard_fail" + + +@dataclass(frozen=True) +class ProducerDecision: + kind: DecisionKind + + +@dataclass(frozen=True) +class CanonicalDecision: + status_bytes: bytes + decision: ProducerDecision + + +def _fail(code: str) -> None: + raise DecisionContractError(code) + + +def _mapping(value: object, keys: frozenset[str], code: str) -> dict[str, object]: + if not isinstance(value, Mapping): + _fail(code) + try: + data = dict(value) + except (AttributeError, KeyError, OverflowError, RuntimeError, TypeError, UnicodeError, ValueError): + _fail(code) + if set(data) != keys or any(type(key) is not str for key in data): + _fail(code) + return data + + +def _string(value: object, code: str, *, allow_empty: bool = False) -> str: + if type(value) is not str or (not allow_empty and not value) or any(ord(char) < 0x20 for char in value): + _fail(code) + return value + + +def _integer(value: object, code: str) -> int: + if type(value) is not int or value < 0 or value > MAX_SAFE_JSON_INTEGER: + _fail(code) + return value + + +def _row(value: object) -> dict[str, str]: + data = _mapping(value, frozenset(_ROW_KEYS), "row_invalid") + return {key: _string(data[key], "row_invalid", allow_empty=key == "author") for key in _ROW_KEYS} + + +def _digest(rows: list[dict[str, str]]) -> str: + ordered = sorted(rows, key=lambda row: tuple(row[key] for key in _ROW_KEYS)) + try: + payload = json.dumps( + ordered, + ensure_ascii=False, + sort_keys=True, + separators=(",", ":"), + allow_nan=False, + ).encode("utf-8") + except (RecursionError, TypeError, UnicodeError, ValueError): + _fail("row_digest_invalid") + return hashlib.sha256(payload).hexdigest() + + +def _parse_outcome(value: object) -> tuple[dict[str, object], list[dict[str, str]]]: + data = _mapping(value, _OUTCOME_KEYS, "outcome_invalid") + feed_id = _string(data["feed_id"], "feed_invalid") + feed_url = _string(data["feed_url"], "feed_invalid") + kind = data["kind"] + state = data["state"] + if type(kind) is not str or kind not in {"rss2", "atom", "unknown"}: + _fail("feed_kind_invalid") + if type(state) is not str or state not in {"accepted", "failed", "quarantined"}: + _fail("feed_state_invalid") + rows = data["rows"] + if not isinstance(rows, (list, tuple)) or len(rows) > MAX_ROWS_PER_FEED: + _fail("rows_invalid") + row_snapshot = [_row(item) for item in rows] + error = data["error_code"] + if error is not None and (type(error) is not str or not _ERROR_RE.fullmatch(error)): + _fail("error_code_invalid") + if state in {"accepted", "quarantined"} and kind not in {"rss2", "atom"}: + _fail("feed_kind_invalid") + if state == "accepted" and (not row_snapshot or error is not None): + _fail("outcome_invariant_invalid") + if state == "quarantined" and (row_snapshot or error is None): + _fail("outcome_invariant_invalid") + if state == "failed" and (row_snapshot or error is None): + _fail("outcome_invariant_invalid") + return {"feed_id": feed_id, "feed_url": feed_url, "kind": kind, "state": state, "error_code": error}, row_snapshot + + +def _canonical(value: object) -> bytes: + try: + return json.dumps(value, ensure_ascii=False, sort_keys=True, separators=(",", ":"), allow_nan=False).encode( + "utf-8" + ) + except (RecursionError, TypeError, UnicodeError, ValueError): + _fail("status_serialization_invalid") + + +def _build_wire(parsed: list[tuple[dict[str, object], list[dict[str, str]]]]) -> dict[str, object]: + parsed.sort(key=lambda pair: (pair[0]["feed_id"], pair[0]["feed_url"])) + accepted_rows = sorted( + [row for data, rows in parsed if data["state"] == "accepted" for row in rows], + key=lambda row: tuple(row[key] for key in _ROW_KEYS), + ) + feeds = [] + for data, rows in parsed: + accepted = data["state"] == "accepted" + feeds.append( + { + "feed_id": data["feed_id"], + "feed_url": data["feed_url"], + "kind": data["kind"], + "state": data["state"], + "accepted_row_count": len(rows) if accepted else 0, + "rejected_row_count": 0, + "row_digest": _digest(rows) if accepted else hashlib.sha256(b"[]").hexdigest(), + "error_code": data["error_code"], + } + ) + failed = sum(data["state"] == "failed" for data, _ in parsed) + quarantined = sum(data["state"] == "quarantined" for data, _ in parsed) + complete = failed == 0 and quarantined == 0 + return { + "status_version": STATUS_VERSION, + "configured_feed_count": len(parsed), + "feed_count": len(parsed), + "successful_feed_count": len(parsed) - failed - quarantined, + "failed_feed_count": failed, + "quarantined_feed_count": quarantined, + "accepted_row_count": len(accepted_rows), + "rejected_row_count": 0, + "publication_complete": complete, + "eligible_for_live_publication": complete, + "aggregate_row_digest": _digest(accepted_rows), + "feeds": feeds, + } + + +def build_decision(validated_outcomes: Iterable[Mapping[str, object]]) -> CanonicalDecision: + try: + values = list(validated_outcomes) + except (AttributeError, RuntimeError, TypeError, ValueError): + _fail("outcomes_invalid") + if not values: + _fail("feed_config_empty") + parsed = [_parse_outcome(value) for value in values] + if len({data["feed_id"] for data, _ in parsed}) != len(parsed): + _fail("feed_duplicate") + status = _build_wire(parsed) + if status["failed_feed_count"]: + decision = DecisionKind.HARD_FAIL + elif status["quarantined_feed_count"]: + decision = DecisionKind.QUARANTINE + else: + decision = DecisionKind.SUCCESS + return CanonicalDecision(_canonical(status), ProducerDecision(decision)) + + +def _reject_duplicate_keys(pairs: list[tuple[str, object]]) -> dict[str, object]: + result: dict[str, object] = {} + for key, value in pairs: + if key in result: + _fail("status_duplicate_key") + result[key] = value + return result + + +def _validate_wire(value: object) -> dict[str, object]: + data = _mapping(value, _WIRE_KEYS, "status_invalid") + if data["status_version"] != STATUS_VERSION: + _fail("status_version_invalid") + counter_keys = _WIRE_KEYS - { + "status_version", + "publication_complete", + "eligible_for_live_publication", + "aggregate_row_digest", + "feeds", + } + for key in counter_keys: + _integer(data[key], "status_counter_invalid") + for key in ("publication_complete", "eligible_for_live_publication"): + if type(data[key]) is not bool: + _fail("status_flag_invalid") + aggregate = _string(data["aggregate_row_digest"], "status_digest_invalid") + if not _DIGEST_RE.fullmatch(aggregate): + _fail("status_digest_invalid") + feeds = data["feeds"] + if not isinstance(feeds, list): + _fail("feed_invalid") + previous: tuple[str, str] | None = None + feed_ids: set[str] = set() + for value in feeds: + item = _mapping(value, _FEED_KEYS, "feed_invalid") + feed_id = _string(item["feed_id"], "feed_invalid") + feed_url = _string(item["feed_url"], "feed_invalid") + key = (feed_id, feed_url) + if feed_id in feed_ids: + _fail("feed_duplicate") + if previous is not None and key <= previous: + _fail("feed_order_invalid") + feed_ids.add(feed_id) + previous = key + kind = item["kind"] + state = item["state"] + if type(kind) is not str or kind not in {"rss2", "atom", "unknown"}: + _fail("feed_kind_invalid") + if type(state) is not str or state not in {"accepted", "failed", "quarantined"}: + _fail("feed_state_invalid") + if state in {"accepted", "quarantined"} and kind not in {"rss2", "atom"}: + _fail("feed_kind_invalid") + accepted_count = _integer(item["accepted_row_count"], "feed_counter_invalid") + _integer(item["rejected_row_count"], "feed_counter_invalid") + if state == "accepted" and accepted_count == 0: + _fail("feed_state_invalid") + if state != "accepted" and accepted_count != 0: + _fail("feed_state_invalid") + digest = _string(item["row_digest"], "status_digest_invalid") + if not _DIGEST_RE.fullmatch(digest): + _fail("status_digest_invalid") + error = item["error_code"] + if state == "accepted" and error is not None: + _fail("feed_state_invalid") + if state != "accepted" and (type(error) is not str or not _ERROR_RE.fullmatch(error)): + _fail("feed_error_invalid") + if data["configured_feed_count"] != len(feeds) or data["feed_count"] != len(feeds): + _fail("status_counter_invalid") + accepted = sum(item["state"] == "accepted" for item in feeds) + failed = sum(item["state"] == "failed" for item in feeds) + quarantined = sum(item["state"] == "quarantined" for item in feeds) + accepted_rows = sum(item["accepted_row_count"] for item in feeds) + if data["successful_feed_count"] != accepted or data["failed_feed_count"] != failed: + _fail("status_counter_invalid") + if data["quarantined_feed_count"] != quarantined or data["accepted_row_count"] != accepted_rows: + _fail("status_counter_invalid") + complete = failed == 0 and quarantined == 0 + if data["publication_complete"] != complete or data["eligible_for_live_publication"] != complete: + _fail("status_flag_invalid") + return data + + +def read_status(status_bytes: bytes) -> dict[str, object]: + if type(status_bytes) is not bytes: + _fail("status_bytes_invalid") + try: + value = json.loads(status_bytes.decode("utf-8"), object_pairs_hook=_reject_duplicate_keys) + except (UnicodeError, json.JSONDecodeError, RecursionError): + _fail("status_bytes_invalid") + data = _validate_wire(value) + if _canonical(data) != status_bytes: + _fail("status_noncanonical") + return json.loads(status_bytes.decode("utf-8")) + + +def status_digest(status_bytes: bytes) -> str: + if type(status_bytes) is not bytes: + _fail("status_bytes_invalid") + return hashlib.sha256(status_bytes).hexdigest() diff --git a/tests/test_feed_status_canonical.py b/tests/test_feed_status_canonical.py new file mode 100644 index 0000000..5fb93fa --- /dev/null +++ b/tests/test_feed_status_canonical.py @@ -0,0 +1,98 @@ +from __future__ import annotations + +import hashlib +import json + +import pytest + +from political_event_tracking_research.feed_status_canonical import ( + DecisionContractError, + DecisionKind, + build_decision, + read_status, + status_digest, +) + + +ROW = { + "item_id": "a-1", + "published_at": "2026-05-01T12:30:00Z", + "source_type": "official", + "source_url": "https://example.test/a", + "author": "", + "text": "event", +} + + +def outcome( + feed_id: str, + state: str, + *, + kind: str = "rss2", + rows: list[dict[str, str]] | None = None, + error_code: str | None = None, +) -> dict[str, object]: + return { + "feed_id": feed_id, + "feed_url": f"https://example.test/{feed_id}", + "kind": kind, + "state": state, + "rows": rows if rows is not None else ([ROW] if state == "accepted" else []), + "error_code": error_code, + } + + +def test_build_returns_only_canonical_bytes_and_explicit_decision() -> None: + result = build_decision([outcome("empty", "quarantined", error_code="zero_entries")]) + assert type(result.status_bytes) is bytes + assert result.decision.kind is DecisionKind.QUARANTINE + assert json.loads(result.status_bytes)["eligible_for_live_publication"] is False + + +def test_read_status_is_json_native_and_mutation_isolated() -> None: + result = build_decision([outcome("a", "accepted")]) + first = read_status(result.status_bytes) + assert isinstance(first, dict) + json.dumps(first) + first["feed_count"] = 99 + first["feeds"][0]["feed_id"] = "changed" + second = read_status(result.status_bytes) + assert second["feed_count"] == 1 + assert second["feeds"][0]["feed_id"] == "a" + + +def test_digest_only_depends_on_status_bytes() -> None: + result = build_decision([outcome("a", "accepted")]) + assert status_digest(result.status_bytes) == hashlib.sha256(result.status_bytes).hexdigest() + assert status_digest(result.status_bytes) == status_digest(bytes(result.status_bytes)) + + +@pytest.mark.parametrize( + "records,kind", + [ + ([outcome("a", "accepted")], DecisionKind.SUCCESS), + ([outcome("empty", "quarantined", error_code="zero_entries")], DecisionKind.QUARANTINE), + ([outcome("bad", "failed", kind="unknown", error_code="fetch_failed")], DecisionKind.HARD_FAIL), + ( + [ + outcome("bad", "failed", kind="unknown", error_code="fetch_failed"), + outcome("empty", "quarantined", error_code="zero_entries"), + ], + DecisionKind.HARD_FAIL, + ), + ], +) +def test_decision_status_combinations(records: list[dict[str, object]], kind: DecisionKind) -> None: + assert build_decision(records).decision.kind is kind + + +def test_tampered_duplicate_and_noncanonical_bytes_fail_closed() -> None: + result = build_decision([outcome("a", "accepted")]) + tampered = result.status_bytes.replace(b'"feed_count":1', b'"feed_count":2') + with pytest.raises(DecisionContractError, match="status_"): + read_status(tampered) + duplicate = result.status_bytes.replace(b'"feed_count":1', b'"feed_count":1,"feed_count":1') + with pytest.raises(DecisionContractError, match="duplicate"): + read_status(duplicate) + with pytest.raises(DecisionContractError, match="noncanonical"): + read_status(result.status_bytes + b"\n") From ef5b22da3eb4b8eb19c021c103009e062f4eee44 Mon Sep 17 00:00:00 2001 From: Pigbibi <20649888+Pigbibi@users.noreply.github.com> Date: Thu, 16 Jul 2026 05:15:37 +0800 Subject: [PATCH 2/2] Validate canonical status count invariants Co-Authored-By: Codex --- .../feed_status_canonical.py | 21 ++++++++++-- tests/test_feed_status_canonical.py | 34 +++++++++++++++++++ 2 files changed, 52 insertions(+), 3 deletions(-) diff --git a/src/political_event_tracking_research/feed_status_canonical.py b/src/political_event_tracking_research/feed_status_canonical.py index c078208..fb6259c 100644 --- a/src/political_event_tracking_research/feed_status_canonical.py +++ b/src/political_event_tracking_research/feed_status_canonical.py @@ -15,6 +15,7 @@ _OUTCOME_KEYS = frozenset({"feed_id", "feed_url", "kind", "state", "rows", "error_code"}) _ERROR_RE = re.compile(r"^[a-z][a-z0-9_]{0,63}$") _DIGEST_RE = re.compile(r"^[0-9a-f]{64}$") +_EMPTY_DIGEST = hashlib.sha256(b"[]").hexdigest() _WIRE_KEYS = frozenset( { "status_version", @@ -244,7 +245,7 @@ def _validate_wire(value: object) -> dict[str, object]: if not _DIGEST_RE.fullmatch(aggregate): _fail("status_digest_invalid") feeds = data["feeds"] - if not isinstance(feeds, list): + if not isinstance(feeds, list) or not feeds: _fail("feed_invalid") previous: tuple[str, str] | None = None feed_ids: set[str] = set() @@ -268,7 +269,11 @@ def _validate_wire(value: object) -> dict[str, object]: if state in {"accepted", "quarantined"} and kind not in {"rss2", "atom"}: _fail("feed_kind_invalid") accepted_count = _integer(item["accepted_row_count"], "feed_counter_invalid") - _integer(item["rejected_row_count"], "feed_counter_invalid") + rejected_count = _integer(item["rejected_row_count"], "feed_counter_invalid") + if accepted_count > MAX_ROWS_PER_FEED: + _fail("feed_counter_invalid") + if rejected_count != 0: + _fail("rejected_count_invalid") if state == "accepted" and accepted_count == 0: _fail("feed_state_invalid") if state != "accepted" and accepted_count != 0: @@ -276,6 +281,8 @@ def _validate_wire(value: object) -> dict[str, object]: digest = _string(item["row_digest"], "status_digest_invalid") if not _DIGEST_RE.fullmatch(digest): _fail("status_digest_invalid") + if state != "accepted" and digest != _EMPTY_DIGEST: + _fail("empty_digest_invalid") error = item["error_code"] if state == "accepted" and error is not None: _fail("feed_state_invalid") @@ -287,10 +294,18 @@ def _validate_wire(value: object) -> dict[str, object]: failed = sum(item["state"] == "failed" for item in feeds) quarantined = sum(item["state"] == "quarantined" for item in feeds) accepted_rows = sum(item["accepted_row_count"] for item in feeds) + rejected_rows = sum(item["rejected_row_count"] for item in feeds) if data["successful_feed_count"] != accepted or data["failed_feed_count"] != failed: _fail("status_counter_invalid") - if data["quarantined_feed_count"] != quarantined or data["accepted_row_count"] != accepted_rows: + if ( + data["quarantined_feed_count"] != quarantined + or data["accepted_row_count"] != accepted_rows + or rejected_rows != 0 + or data["rejected_row_count"] != 0 + ): _fail("status_counter_invalid") + if accepted_rows == 0 and data["aggregate_row_digest"] != _EMPTY_DIGEST: + _fail("empty_digest_invalid") complete = failed == 0 and quarantined == 0 if data["publication_complete"] != complete or data["eligible_for_live_publication"] != complete: _fail("status_flag_invalid") diff --git a/tests/test_feed_status_canonical.py b/tests/test_feed_status_canonical.py index 5fb93fa..ea2b54c 100644 --- a/tests/test_feed_status_canonical.py +++ b/tests/test_feed_status_canonical.py @@ -6,6 +6,7 @@ import pytest from political_event_tracking_research.feed_status_canonical import ( + MAX_ROWS_PER_FEED, DecisionContractError, DecisionKind, build_decision, @@ -96,3 +97,36 @@ def test_tampered_duplicate_and_noncanonical_bytes_fail_closed() -> None: read_status(duplicate) with pytest.raises(DecisionContractError, match="noncanonical"): read_status(result.status_bytes + b"\n") + + +def test_readback_rejects_empty_feed_list_and_wrong_empty_digests() -> None: + result = build_decision([outcome("empty", "quarantined", error_code="zero_entries")]) + payload = json.loads(result.status_bytes) + payload["feeds"] = [] + payload["feed_count"] = 0 + payload["configured_feed_count"] = 0 + with pytest.raises(DecisionContractError, match="feed_count|feed_invalid"): + read_status(json.dumps(payload, sort_keys=True, separators=(",", ":")).encode()) + + payload = json.loads(result.status_bytes) + payload["feeds"][0]["row_digest"] = "0" * 64 + with pytest.raises(DecisionContractError, match="empty_digest"): + read_status(json.dumps(payload, sort_keys=True, separators=(",", ":")).encode()) + + +def test_readback_rechecks_row_count_bounds_sums_and_rejected_zero() -> None: + result = build_decision([outcome("a", "accepted")]) + payload = json.loads(result.status_bytes) + payload["feeds"][0]["accepted_row_count"] = MAX_ROWS_PER_FEED + 1 + with pytest.raises(DecisionContractError, match="feed_counter"): + read_status(json.dumps(payload, sort_keys=True, separators=(",", ":")).encode()) + + payload = json.loads(result.status_bytes) + payload["accepted_row_count"] = 0 + with pytest.raises(DecisionContractError, match="counter"): + read_status(json.dumps(payload, sort_keys=True, separators=(",", ":")).encode()) + + payload = json.loads(result.status_bytes) + payload["feeds"][0]["rejected_row_count"] = 1 + with pytest.raises(DecisionContractError, match="rejected"): + read_status(json.dumps(payload, sort_keys=True, separators=(",", ":")).encode())