From bb51f81e568d472f7f10371aacf591c68231cacb Mon Sep 17 00:00:00 2001 From: Sylvester Kaczmarek <16242628+sylvesterkaczmarek@users.noreply.github.com> Date: Sun, 16 Aug 2026 10:51:50 +0100 Subject: [PATCH] fix(infonet): fail closed on malformed authoritative snapshots --- .../infonet/markets/event_selection.py | 81 +++++++++ backend/services/infonet/markets/lifecycle.py | 91 ++++++----- backend/services/infonet/markets/snapshot.py | 83 +++++----- .../tests/test_lifecycle_invalid_ordering.py | 154 ++++++++++++++++++ 4 files changed, 329 insertions(+), 80 deletions(-) create mode 100644 backend/services/infonet/markets/event_selection.py create mode 100644 backend/services/infonet/tests/test_lifecycle_invalid_ordering.py diff --git a/backend/services/infonet/markets/event_selection.py b/backend/services/infonet/markets/event_selection.py new file mode 100644 index 0000000..524327c --- /dev/null +++ b/backend/services/infonet/markets/event_selection.py @@ -0,0 +1,81 @@ +"""Canonical selection helpers for market events. + +The Infonet hashchain is already an append-only ordered sequence. Market +application code must preserve that iteration order instead of deriving a +second history from partially trusted ``timestamp`` or ``sequence`` fields. + +Authoritative event selection therefore always chooses the first matching +event in hashchain order. Callers validate that event separately and fail +closed; they must not skip a malformed first authoritative event in favour of +a later replacement. +""" + +from __future__ import annotations + +import math +from typing import Any, Iterable + + +def payload(event: dict[str, Any]) -> dict[str, Any]: + value = event.get("payload") + return value if isinstance(value, dict) else {} + + +def market_id(event: dict[str, Any]) -> str: + return str(payload(event).get("market_id") or "") + + +def finite_float(value: Any) -> float | None: + try: + parsed = float(value) + except (TypeError, ValueError, OverflowError): + return None + return parsed if math.isfinite(parsed) else None + + +def safe_int(value: Any) -> int | None: + try: + return int(value) + except (TypeError, ValueError, OverflowError): + return None + + +def has_valid_ordering(event: dict[str, Any]) -> bool: + """Whether an event carries usable ordering metadata.""" + return finite_float(event.get("timestamp")) is not None and safe_int( + event.get("sequence") + ) is not None + + +def events_for_market( + market_id_value: str, + chain: Iterable[dict[str, Any]], +) -> list[dict[str, Any]]: + """Filter one market while preserving canonical hashchain iteration order.""" + return [ + event + for event in chain + if isinstance(event, dict) and market_id(event) == market_id_value + ] + + +def first_authoritative_event( + events: Iterable[dict[str, Any]], + event_type: str, +) -> dict[str, Any] | None: + """Return the first matching event without skipping malformed metadata.""" + return next( + (event for event in events if event.get("event_type") == event_type), + None, + ) + + +__all__ = [ + "events_for_market", + "finite_float", + "first_authoritative_event", + "has_valid_ordering", + "market_id", + "payload", + "safe_int", +] diff --git a/backend/services/infonet/markets/lifecycle.py b/backend/services/infonet/markets/lifecycle.py index a91a92f..ca5dc14 100644 --- a/backend/services/infonet/markets/lifecycle.py +++ b/backend/services/infonet/markets/lifecycle.py @@ -24,6 +24,13 @@ from enum import Enum from typing import Any, Iterable from services.infonet.config import CONFIG +from services.infonet.markets.event_selection import ( + events_for_market, + finite_float, + first_authoritative_event, + has_valid_ordering, + payload, +) class MarketStatus(str, Enum): @@ -37,24 +44,17 @@ class MarketStatus(str, Enum): _SECONDS_PER_HOUR = 3600.0 -def _payload(event: dict[str, Any]) -> dict[str, Any]: - p = event.get("payload") - return p if isinstance(p, dict) else {} - - -def _market_id(event: dict[str, Any]) -> str: - return str(_payload(event).get("market_id") or "") - - -def _events_for_market(market_id: str, chain: Iterable[dict[str, Any]]) -> list[dict[str, Any]]: - out: list[dict[str, Any]] = [] - for ev in chain: - if not isinstance(ev, dict): - continue - if _market_id(ev) == market_id: - out.append(ev) - out.sort(key=lambda e: (float(e.get("timestamp") or 0.0), int(e.get("sequence") or 0))) - return out +def _snapshot_time(snapshot: dict[str, Any]) -> float | None: + """Return a valid snapshot time without masking malformed metadata.""" + if not has_valid_ordering(snapshot): + return None + raw_timestamp = snapshot.get("timestamp") + # Preserve the pre-existing fallback only for an actually absent/zero + # timestamp. A present-but-malformed timestamp must fail closed instead + # of being disguised by a valid payload frozen_at. + if raw_timestamp in (None, "", 0, 0.0): + return finite_float(payload(snapshot).get("frozen_at")) + return finite_float(raw_timestamp) def compute_market_status( @@ -71,24 +71,31 @@ def compute_market_status( ``chain_majority_time(chain)``) so every node converges on the same status. """ - events = _events_for_market(market_id, chain) + events = events_for_market(market_id, chain) if not events: return MarketStatus.PREDICTING # treated as not-yet-existing - create_event = next((e for e in events if e.get("event_type") == "prediction_create"), None) - if create_event is None: + create_event = first_authoritative_event(events, "prediction_create") + if create_event is None or not has_valid_ordering(create_event): return MarketStatus.PREDICTING - finalize = next((e for e in events if e.get("event_type") == "resolution_finalize"), None) - if finalize is not None: - outcome = _payload(finalize).get("outcome") + # The first finalize remains authoritative by chain order. A malformed + # first finalize cannot exert terminal authority, and a later finalize + # must not replace it; projection therefore falls through to the prior + # snapshot-backed phase. + finalize = first_authoritative_event(events, "resolution_finalize") + if finalize is not None and has_valid_ordering(finalize): + outcome = payload(finalize).get("outcome") return MarketStatus.INVALID if outcome == "invalid" else MarketStatus.FINAL - snapshot = next((e for e in events if e.get("event_type") == "market_snapshot"), None) + # Likewise, never let a later snapshot replace the first commitment. + snapshot = first_authoritative_event(events, "market_snapshot") if snapshot is None: return MarketStatus.PREDICTING - snapshot_ts = float(snapshot.get("timestamp") or _payload(snapshot).get("frozen_at") or 0.0) + snapshot_ts = _snapshot_time(snapshot) + if snapshot_ts is None: + return MarketStatus.PREDICTING evidence_close = snapshot_ts + float(CONFIG["evidence_window_hours"]) * _SECONDS_PER_HOUR if now < evidence_close: return MarketStatus.EVIDENCE @@ -110,29 +117,37 @@ def should_advance_phase( - EVIDENCE → RESOLVING: just a status change (no chain event). - RESOLVING → FINAL/INVALID: emit ``resolution_finalize``. """ - events = _events_for_market(market_id, chain) + events = events_for_market(market_id, chain) if not events: return None - create_event = next((e for e in events if e.get("event_type") == "prediction_create"), None) - if create_event is None: + create_event = first_authoritative_event(events, "prediction_create") + if create_event is None or not has_valid_ordering(create_event): return None - finalize = next((e for e in events if e.get("event_type") == "resolution_finalize"), None) - if finalize is not None: - return None # already terminal - create_payload = _payload(create_event) - trigger_date = float(create_payload.get("trigger_date") or 0.0) - snapshot = next((e for e in events if e.get("event_type") == "market_snapshot"), None) + finalize = first_authoritative_event(events, "resolution_finalize") + if finalize is not None: + # Whether valid (terminal) or malformed (fail closed), do not let a + # later finalize replace the first authoritative chain event. + return None + + create_payload = payload(create_event) + trigger_date = finite_float(create_payload.get("trigger_date")) + snapshot = first_authoritative_event(events, "market_snapshot") if snapshot is None: # PREDICTING — advance to EVIDENCE iff trigger_date has passed in - # majority chain time. - if now >= trigger_date: + # majority chain time. Invalid trigger metadata cannot authorize + # a phase transition. + if trigger_date is not None and now >= trigger_date: return (MarketStatus.PREDICTING, MarketStatus.EVIDENCE) return None - snapshot_ts = float(snapshot.get("timestamp") or _payload(snapshot).get("frozen_at") or 0.0) + snapshot_ts = _snapshot_time(snapshot) + if snapshot_ts is None: + # An invalid first snapshot cannot authorize a transition, and a + # later snapshot must not replace the commitment boundary. + return None evidence_close = snapshot_ts + float(CONFIG["evidence_window_hours"]) * _SECONDS_PER_HOUR resolution_close = evidence_close + float(CONFIG["resolution_window_hours"]) * _SECONDS_PER_HOUR diff --git a/backend/services/infonet/markets/snapshot.py b/backend/services/infonet/markets/snapshot.py index 5d8fa0d..469391e 100644 --- a/backend/services/infonet/markets/snapshot.py +++ b/backend/services/infonet/markets/snapshot.py @@ -15,41 +15,34 @@ evaluation. Once frozen: can't pre-mine before the boundary. The snapshot itself is **immutable** by spec — the producer emits it -once and never updates it. Sprint 4 enforces immutability by ignoring -any subsequent ``market_snapshot`` events with the same market_id -(``find_snapshot`` returns the FIRST one). Tests assert this invariant. +once and never updates it. The first snapshot in canonical hashchain +append order is authoritative; a malformed first snapshot fails closed +and cannot be replaced by a later one. """ from __future__ import annotations import hashlib import json -import math from typing import Any, Iterable - -def _payload(event: dict[str, Any]) -> dict[str, Any]: - p = event.get("payload") - return p if isinstance(p, dict) else {} +from services.infonet.markets.event_selection import ( + events_for_market, + finite_float, + first_authoritative_event, + has_valid_ordering, + payload, +) -def _finite_float(value: Any) -> float | None: - try: - parsed = float(value) - except (TypeError, ValueError, OverflowError): - return None - return parsed if math.isfinite(parsed) else None +class InvalidAuthoritativeSnapshot(ValueError): + """Raised when the first on-chain snapshot exists but is malformed. - -def _events_for_market(market_id: str, chain: Iterable[dict[str, Any]]) -> list[dict[str, Any]]: - out: list[dict[str, Any]] = [] - for ev in chain: - if not isinstance(ev, dict): - continue - if _payload(ev).get("market_id") == market_id: - out.append(ev) - out.sort(key=lambda e: (float(e.get("timestamp") or 0.0), int(e.get("sequence") or 0))) - return out + ``None`` from :func:`find_snapshot` is reserved for the distinct state in + which no snapshot exists. Raising here prevents downstream resolution and + eligibility code from interpreting a corrupted commitment as an absent + snapshot with relaxed restrictions. + """ def build_snapshot( @@ -60,21 +53,21 @@ def build_snapshot( ) -> dict[str, Any]: """Compute the snapshot payload deterministically from chain history. - Walks ``prediction_place`` events for ``market_id``, in chain order, - and produces the frozen counts / stake totals / predictor list / - yes-no probability state. The resulting dict is ready to be written - as the payload of a ``market_snapshot`` event. + Walks ``prediction_place`` events for ``market_id`` in canonical hashchain + append order and produces the frozen counts / stake totals / predictor list / + yes-no probability state. The resulting dict is ready to be written as the + payload of a ``market_snapshot`` event. ``frozen_at`` is the canonical commitment timestamp — typically ``chain_majority_time(chain)`` at the moment the producer decides to advance to EVIDENCE. Pass it explicitly so the function stays pure and deterministic. """ - frozen_at_value = _finite_float(frozen_at) + frozen_at_value = finite_float(frozen_at) if frozen_at_value is None: raise ValueError("frozen_at must be finite") - events = _events_for_market(market_id, chain) + events = events_for_market(market_id, chain) predictor_ids: list[str] = [] seen_predictors: set[str] = set() @@ -88,7 +81,7 @@ def build_snapshot( node = ev.get("node_id") if not isinstance(node, str) or not node: continue - p = _payload(ev) + p = payload(ev) side = p.get("side") if side not in ("yes", "no"): continue @@ -98,7 +91,7 @@ def build_snapshot( weight = 1.0 # Free pick = 1.0 virtual stake (RULES §5.2). staked_amount = 0.0 else: - parsed_stake = _finite_float(stake) + parsed_stake = finite_float(stake) if parsed_stake is None or parsed_stake <= 0: # Invalid paid predictions must not inflate participant # counts or poison the frozen probability state. @@ -165,21 +158,27 @@ def find_snapshot( market_id: str, chain: Iterable[dict[str, Any]], ) -> dict[str, Any] | None: - """Return the FIRST ``market_snapshot`` payload for ``market_id``. + """Return the first authoritative ``market_snapshot`` payload. - Subsequent ``market_snapshot`` events with the same market_id are - ignored — snapshots are immutable per RULES §2.2. This is a - structural enforcement, not just a convention; an attacker who - forges a second snapshot cannot influence resolution. + Selection follows canonical hashchain append order. ``None`` means no + snapshot exists. If the first snapshot exists but has malformed ordering + metadata, :class:`InvalidAuthoritativeSnapshot` is raised instead of + treating the corruption as absence or allowing a later snapshot to replace + the commitment. """ - events = _events_for_market(market_id, chain) - for ev in events: - if ev.get("event_type") == "market_snapshot": - return _payload(ev) - return None + events = events_for_market(market_id, chain) + snapshot = first_authoritative_event(events, "market_snapshot") + if snapshot is None: + return None + if not has_valid_ordering(snapshot): + raise InvalidAuthoritativeSnapshot( + f"authoritative snapshot for market '{market_id}' has invalid ordering metadata" + ) + return payload(snapshot) __all__ = [ + "InvalidAuthoritativeSnapshot", "build_snapshot", "compute_snapshot_event_hash", "find_snapshot", diff --git a/backend/services/infonet/tests/test_lifecycle_invalid_ordering.py b/backend/services/infonet/tests/test_lifecycle_invalid_ordering.py new file mode 100644 index 0000000..4b2e245 --- /dev/null +++ b/backend/services/infonet/tests/test_lifecycle_invalid_ordering.py @@ -0,0 +1,154 @@ +"""Regression coverage for malformed market lifecycle ordering metadata.""" + +from typing import Any + +import pytest + +from services.infonet.config import CONFIG +from services.infonet.markets import MarketStatus, compute_market_status, should_advance_phase +from services.infonet.markets.event_selection import events_for_market +from services.infonet.markets.resolution import resolve_market +from services.infonet.markets.snapshot import InvalidAuthoritativeSnapshot, find_snapshot + + +def _create(trigger_date: Any = 200.0) -> dict[str, Any]: + return { + "event_type": "prediction_create", + "node_id": "creator", + "timestamp": 100.0, + "sequence": 1, + "payload": {"market_id": "m1", "trigger_date": trigger_date}, + } + + +def _snapshot( + timestamp: Any = 200.0, frozen_at: Any = 200.0, sequence: Any = 2 +) -> dict[str, Any]: + return { + "event_type": "market_snapshot", + "node_id": "creator", + "timestamp": timestamp, + "sequence": sequence, + "payload": { + "market_id": "m1", + "frozen_at": frozen_at, + "frozen_predictor_ids": ["predictor"], + }, + } + + +def _finalize(timestamp: Any = 300.0, sequence: Any = 3) -> dict[str, Any]: + return { + "event_type": "resolution_finalize", + "node_id": "creator", + "timestamp": timestamp, + "sequence": sequence, + "payload": {"market_id": "m1", "outcome": "yes"}, + } + + +def test_malformed_non_authoritative_metadata_does_not_break_projection() -> None: + chain = [ + _create(), + _snapshot(), + { + "event_type": "prediction_place", + "node_id": "peer", + "timestamp": "not-a-timestamp", + "sequence": "not-a-sequence", + "payload": {"market_id": "m1"}, + }, + ] + + assert compute_market_status("m1", chain, now=201.0) == MarketStatus.EVIDENCE + + +def test_market_events_preserve_canonical_hashchain_order() -> None: + first = _snapshot(timestamp=300.0, sequence=5) + second = _snapshot(timestamp=200.0, sequence=2) + + events = events_for_market("m1", [_create(), first, second]) + + assert events[1:] == [first, second] + + +def test_lifecycle_and_find_snapshot_share_authoritative_append_order() -> None: + evidence_window = float(CONFIG["evidence_window_hours"]) * 3600.0 + first_timestamp = 10.0 * evidence_window + first = _snapshot(timestamp=first_timestamp, frozen_at=first_timestamp, sequence=5) + second = _snapshot(timestamp=1.0, frozen_at=1.0, sequence=2) + chain = [_create(), first, second] + + # Lifecycle must use the first appended snapshot. If snapshot.py re-sorted + # by timestamp/sequence, the much older second event would be selected and + # this same `now` would project RESOLVING instead of EVIDENCE. + now = first_timestamp + evidence_window / 2.0 + assert compute_market_status("m1", chain, now=now) == MarketStatus.EVIDENCE + assert find_snapshot("m1", chain) == first["payload"] + + +def test_invalid_first_snapshot_fails_closed_without_replacement() -> None: + chain = [ + _create(), + _snapshot(timestamp="bad", frozen_at=200.0, sequence="bad"), + _snapshot(timestamp=200.0, frozen_at=200.0, sequence=3), + ] + + assert compute_market_status("m1", chain, now=10_000.0) == MarketStatus.PREDICTING + assert should_advance_phase("m1", chain, now=10_000.0) is None + with pytest.raises(InvalidAuthoritativeSnapshot): + find_snapshot("m1", chain) + + +def test_invalid_first_snapshot_halts_direct_resolution() -> None: + chain = [ + _create(), + _snapshot(timestamp="bad", frozen_at=200.0, sequence="bad"), + _snapshot(timestamp=200.0, frozen_at=200.0, sequence=3), + { + "event_type": "resolution_stake", + "node_id": "resolver", + "timestamp": 400.0, + "sequence": 4, + "payload": { + "market_id": "m1", + "side": "yes", + "amount": 10.0, + "rep_type": "oracle", + }, + }, + ] + + # A malformed authoritative commitment is distinct from no snapshot. + # Resolution must halt instead of treating it as an empty exclusion set. + with pytest.raises(InvalidAuthoritativeSnapshot): + resolve_market("m1", chain) + + +def test_nonfinite_snapshot_timestamp_does_not_fall_back_to_frozen_at() -> None: + chain = [_create(), _snapshot(timestamp=float("nan"), frozen_at=200.0)] + + assert compute_market_status("m1", chain, now=201.0) == MarketStatus.PREDICTING + assert should_advance_phase("m1", chain, now=10_000.0) is None + with pytest.raises(InvalidAuthoritativeSnapshot): + find_snapshot("m1", chain) + + +def test_invalid_first_finalize_cannot_be_replaced_by_later_finalize() -> None: + chain = [ + _create(), + _snapshot(), + _finalize(timestamp="bad", sequence="bad"), + _finalize(timestamp=300.0, sequence=4), + ] + + # The malformed finalize has no terminal authority, so projection stays + # in the prior snapshot-backed phase. The later finalize cannot replace it. + assert compute_market_status("m1", chain, now=201.0) == MarketStatus.EVIDENCE + assert should_advance_phase("m1", chain, now=10_000.0) is None + + +def test_invalid_trigger_date_cannot_authorize_phase_advance() -> None: + chain = [_create(trigger_date="not-a-date")] + + assert should_advance_phase("m1", chain, now=10_000.0) is None