diff --git a/packages/loopx-finance-execution/src/loopx_finance_execution/confirmation_gate.py b/packages/loopx-finance-execution/src/loopx_finance_execution/confirmation_gate.py new file mode 100644 index 0000000000..49c68c8af0 --- /dev/null +++ b/packages/loopx-finance-execution/src/loopx_finance_execution/confirmation_gate.py @@ -0,0 +1,228 @@ +"""Provider-neutral admission gate for one human-confirmed financial operation. + +The Core typed-action store owns the confirmation receipt, the single execution +claim and the canonical operation envelope. A venue executor must additionally +prove that the effect it is about to perform is the effect the operator +approved. + +This module performs that check against records that are already loaded from +the canonical store. It never mutates state, never reads credentials and never +talks to a venue, so it can run before any effect-bearing call. A refusal is a +typed result, not an exception: callers must report it honestly instead of +retrying or working around it. + +Every check fails closed. A record that does not carry the authoritative +original-card identity, a canonical claim id or a valid canonical deadline is +refused rather than admitted by default. +""" + +from __future__ import annotations + +from collections.abc import Mapping +from datetime import datetime, timezone +import hashlib +import json +from typing import Any + + +GATE_SCHEMA_VERSION = "finance_confirmation_gate_v0" + +#: Record statuses that still represent a live, human-confirmed operation. +#: ``applying`` is the ordinary state while the claimed effect is in flight. +LIVE_RECORD_STATUSES = frozenset({"applied", "applying"}) + +#: Canonical lifecycle states an executor may act on. +LIVE_LIFECYCLE_STATES = frozenset({"claimed", "consumed", "outcome_observed"}) + +CONFIRMATION_MISSING = "confirmation_missing" +CONFIRMATION_PROVIDER_MISMATCH = "confirmation_provider_mismatch" +CONFIRMATION_PRINCIPAL_MISMATCH = "confirmation_principal_mismatch" +CONFIRMATION_CARD_MISMATCH = "confirmation_card_mismatch" +CARD_BINDING_MISSING = "card_binding_missing" +CARD_BINDING_CONFLICT = "card_binding_conflict" +OPERATION_NOT_LIVE = "operation_not_live" +CLAIM_MISSING = "claim_missing" +PAYLOAD_TERMS_MISSING = "payload_terms_missing" +TERMS_MISMATCH = "terms_mismatch" +DEADLINE_MISSING = "deadline_missing" +DEADLINE_INVALID = "deadline_invalid" +EXPIRED = "expired" + + +def _mapping(value: object, *, field: str) -> dict[str, Any]: + if isinstance(value, str): + try: + value = json.loads(value) + except ValueError as exc: + raise ValueError(f"{field} must be a JSON object") from exc + if not isinstance(value, Mapping): + raise ValueError(f"{field} must be an object") + return dict(value) + + +def _canonical(value: object) -> str: + return json.dumps( + value, + allow_nan=False, + ensure_ascii=False, + sort_keys=True, + separators=(",", ":"), + ) + + +def _digest(value: object) -> str | None: + if value is None: + return None + return hashlib.sha256(_canonical(value).encode()).hexdigest() + + +def _timestamp(value: object) -> datetime | None: + if not isinstance(value, str) or not value.strip(): + return None + text = value.strip() + if text.endswith("Z"): + text = text[:-1] + "+00:00" + try: + parsed = datetime.fromisoformat(text) + except ValueError: + return None + return parsed if parsed.tzinfo else parsed.replace(tzinfo=timezone.utc) + + +def _origin_card_id(operation: Mapping[str, Any]) -> tuple[str | None, list[str]]: + """Return the authoritative original-card id and every observed conflict.""" + + delivery = operation.get("delivery") + delivery = dict(delivery) if isinstance(delivery, Mapping) else {} + observed: list[str] = [] + for candidate in (delivery.get("message_id"), operation.get("card_message_id")): + if isinstance(candidate, str) and candidate.strip(): + observed.append(candidate.strip()) + distinct = sorted(set(observed)) + if not distinct: + return None, [] + if len(distinct) > 1: + return None, distinct + return distinct[0], [] + + +def verify_bound_confirmation( + proposal: Mapping[str, Any], + plan: Mapping[str, Any], + *, + expected_provider: str = "lark", + now: datetime | None = None, +) -> dict[str, Any]: + """Check one canonical operation against a frozen execution plan. + + ``proposal`` is the record as stored by the canonical action store + (``normalized_parameters`` plus ``operation``). ``plan`` is the frozen plan + the executor would run: ``terms_key`` (default ``bound_order``), ``terms``, + optional ``authorized_principal`` and an optional ``expires_at``. + + The canonical envelope deadline is required; the payload and plan deadlines + are optional but, when present, must be valid and are enforced + independently of each other. + """ + + parameters = _mapping(proposal.get("normalized_parameters"), field="normalized_parameters") + operation = _mapping(proposal.get("operation"), field="operation") + payload = parameters.get("payload") + payload = dict(payload) if isinstance(payload, Mapping) else {} + + confirmation = operation.get("confirmation") + confirmation = dict(confirmation) if isinstance(confirmation, Mapping) else None + claim = operation.get("claim") + claim = dict(claim) if isinstance(claim, Mapping) else {} + + terms_key = str(plan.get("terms_key") or "bound_order") + expected_terms = plan.get("terms") + authorized_principal = plan.get("authorized_principal") + clock = (now or datetime.now(timezone.utc)).astimezone(timezone.utc) + + refusals: list[str] = [] + origin_card_id, card_conflicts = _origin_card_id(operation) + + if confirmation is None: + refusals.append(CONFIRMATION_MISSING) + else: + if str(confirmation.get("provider") or "") != expected_provider: + refusals.append(CONFIRMATION_PROVIDER_MISMATCH) + if authorized_principal and confirmation.get("principal") != authorized_principal: + refusals.append(CONFIRMATION_PRINCIPAL_MISMATCH) + + if origin_card_id is None: + refusals.append(CARD_BINDING_CONFLICT if card_conflicts else CARD_BINDING_MISSING) + if confirmation is not None: + confirmed_card = confirmation.get("message_id") + # Only a receipt from this operation's own card proves the operator + # confirmed this exact card; anything else proves nothing here. + if ( + not isinstance(confirmed_card, str) + or not confirmed_card.strip() + or origin_card_id is None + or confirmed_card.strip() != origin_card_id + ): + refusals.append(CONFIRMATION_CARD_MISMATCH) + + record_status = str(proposal.get("status") or "") + lifecycle_state = str(operation.get("lifecycle_state") or "") + if record_status not in LIVE_RECORD_STATUSES or lifecycle_state not in LIVE_LIFECYCLE_STATES: + refusals.append(OPERATION_NOT_LIVE) + + # Core remains the only first-consumption owner; the gate only requires + # that a canonical claim id exists. There is no claim bypass. + claim_id = claim.get("claim_id") + if not isinstance(claim_id, str) or not claim_id.strip(): + refusals.append(CLAIM_MISSING) + + payload_terms = payload.get(terms_key) + if not isinstance(payload_terms, Mapping) or expected_terms is None: + refusals.append(PAYLOAD_TERMS_MISSING) + elif _canonical(payload_terms) != _canonical(expected_terms): + refusals.append(TERMS_MISMATCH) + + deadlines: list[tuple[str, datetime]] = [] + envelope_deadline = _timestamp(operation.get("expires_at")) + if envelope_deadline is None: + refusals.append(DEADLINE_MISSING) + else: + deadlines.append(("canonical", envelope_deadline)) + for label, value in (("plan", plan.get("expires_at")), ("payload", payload.get("expires_at"))): + if value is None: + continue + parsed = _timestamp(value) + if parsed is None: + refusals.append(DEADLINE_INVALID) + continue + deadlines.append((label, parsed)) + for _label, deadline in deadlines: + if clock >= deadline: + refusals.append(EXPIRED) + break + + return { + "schema_version": GATE_SCHEMA_VERSION, + "allowed": not refusals, + "refusals": refusals, + "evidence": { + "operation_id": operation.get("operation_id") or proposal.get("proposal_id"), + "record_status": record_status, + "lifecycle_state": lifecycle_state, + "confirmation_digest": operation.get("confirmation_digest"), + "confirmation_provider": (confirmation or {}).get("provider"), + "confirmation_principal": (confirmation or {}).get("principal"), + "confirmation_card_message_id": (confirmation or {}).get("message_id"), + "origin_card_message_id": origin_card_id, + "card_conflict_ids": card_conflicts, + "claim_id": claim_id if isinstance(claim_id, str) else None, + "terms_key": terms_key, + "terms_digest": _digest(payload_terms) if isinstance(payload_terms, Mapping) else None, + "expected_terms_digest": _digest(expected_terms), + "canonical_expires_at": ( + envelope_deadline.isoformat() if envelope_deadline else None + ), + "enforced_deadlines": {label: moment.isoformat() for label, moment in deadlines}, + "observed_at": clock.isoformat(), + }, + } diff --git a/tests/extensions/test_finance_confirmation_gate.py b/tests/extensions/test_finance_confirmation_gate.py new file mode 100644 index 0000000000..a749883f9d --- /dev/null +++ b/tests/extensions/test_finance_confirmation_gate.py @@ -0,0 +1,241 @@ +"""Synthetic coverage for the provider-neutral confirmation gate.""" + +from __future__ import annotations + +from datetime import datetime, timezone +import copy +import json +from pathlib import Path +import sys + + +ROOT = Path(__file__).resolve().parents[2] +sys.path.insert(0, str(ROOT / "packages" / "loopx-finance-execution" / "src")) + +from loopx_finance_execution.confirmation_gate import ( # noqa: E402 + CARD_BINDING_CONFLICT, + CARD_BINDING_MISSING, + CLAIM_MISSING, + CONFIRMATION_CARD_MISMATCH, + CONFIRMATION_MISSING, + CONFIRMATION_PRINCIPAL_MISMATCH, + CONFIRMATION_PROVIDER_MISMATCH, + DEADLINE_INVALID, + DEADLINE_MISSING, + EXPIRED, + OPERATION_NOT_LIVE, + PAYLOAD_TERMS_MISSING, + TERMS_MISMATCH, + verify_bound_confirmation, +) + + +NOW = datetime(2026, 10, 17, 4, 0, tzinfo=timezone.utc) +CARD = "om_synthetic_card_1" +PRINCIPAL = "lark:ou_synthetic_operator" +CANONICAL_EXPIRY = "2026-10-17T05:00:00Z" +TERMS = { + "enabled": True, + "type": "market", + "side": "long", + "amount_usdc": "20", + "limit_price": None, + "reduce_only": False, + "take_profit_stop_loss": None, +} + + +def _proposal() -> dict: + return { + "proposal_id": "proposal-synthetic-1", + "status": "applying", + "normalized_parameters": json.dumps( + { + "payload": { + "schema_version": "finance_bound_order_v1", + "bound_order": copy.deepcopy(TERMS), + "expires_at": CANONICAL_EXPIRY, + } + } + ), + "operation": { + "operation_id": "proposal-synthetic-1", + "lifecycle_state": "claimed", + "confirmation_digest": "a" * 64, + "expires_at": CANONICAL_EXPIRY, + "confirmation": { + "provider": "lark", + "event_id": "synthetic-event-1", + "principal": PRINCIPAL, + "message_id": CARD, + }, + "claim": {"claim_id": "claim-synthetic-1"}, + "delivery": {"message_id": CARD}, + }, + } + + +def _plan() -> dict: + return { + "terms_key": "bound_order", + "terms": copy.deepcopy(TERMS), + "authorized_principal": PRINCIPAL, + } + + +def _payload_of(proposal: dict) -> dict: + return json.loads(proposal["normalized_parameters"])["payload"] + + +def _store_payload(proposal: dict, payload: dict) -> None: + parameters = json.loads(proposal["normalized_parameters"]) + parameters["payload"] = payload + proposal["normalized_parameters"] = json.dumps(parameters) + + +def test_confirmed_operation_is_admitted() -> None: + result = verify_bound_confirmation(_proposal(), _plan(), now=NOW) + assert result["allowed"] is True + assert result["refusals"] == [] + assert result["evidence"]["claim_id"] == "claim-synthetic-1" + assert result["evidence"]["origin_card_message_id"] == CARD + assert set(result["evidence"]["enforced_deadlines"]) == {"canonical", "payload"} + + +def test_missing_confirmation_refuses() -> None: + proposal = _proposal() + proposal["operation"].pop("confirmation") + result = verify_bound_confirmation(proposal, _plan(), now=NOW) + assert result["allowed"] is False + assert CONFIRMATION_MISSING in result["refusals"] + + +def test_confirmation_from_another_channel_refuses() -> None: + proposal = _proposal() + proposal["operation"]["confirmation"]["provider"] = "web" + result = verify_bound_confirmation(proposal, _plan(), now=NOW) + assert CONFIRMATION_PROVIDER_MISMATCH in result["refusals"] + + +def test_confirmation_from_another_principal_refuses() -> None: + proposal = _proposal() + proposal["operation"]["confirmation"]["principal"] = "lark:ou_someone_else" + result = verify_bound_confirmation(proposal, _plan(), now=NOW) + assert CONFIRMATION_PRINCIPAL_MISMATCH in result["refusals"] + + +def test_confirmation_from_another_card_refuses() -> None: + proposal = _proposal() + proposal["operation"]["confirmation"]["message_id"] = "om_other_card" + result = verify_bound_confirmation(proposal, _plan(), now=NOW) + assert CONFIRMATION_CARD_MISMATCH in result["refusals"] + + +def test_missing_origin_card_identity_fails_closed() -> None: + proposal = _proposal() + proposal["operation"].pop("delivery") + result = verify_bound_confirmation(proposal, _plan(), now=NOW) + assert result["allowed"] is False + assert CARD_BINDING_MISSING in result["refusals"] + assert CONFIRMATION_CARD_MISMATCH in result["refusals"] + + +def test_conflicting_origin_card_identity_fails_closed() -> None: + proposal = _proposal() + proposal["operation"]["card_message_id"] = "om_second_card" + result = verify_bound_confirmation(proposal, _plan(), now=NOW) + assert CARD_BINDING_CONFLICT in result["refusals"] + assert set(result["evidence"]["card_conflict_ids"]) == {CARD, "om_second_card"} + + +def test_cancelled_operation_is_not_live() -> None: + proposal = _proposal() + proposal["status"] = "cancelled" + proposal["operation"]["lifecycle_state"] = "prepared" + result = verify_bound_confirmation(proposal, _plan(), now=NOW) + assert OPERATION_NOT_LIVE in result["refusals"] + + +def test_omitted_claim_refuses() -> None: + proposal = _proposal() + proposal["operation"].pop("claim") + result = verify_bound_confirmation(proposal, _plan(), now=NOW) + assert CLAIM_MISSING in result["refusals"] + + +def test_empty_claim_refuses() -> None: + proposal = _proposal() + proposal["operation"]["claim"] = {} + result = verify_bound_confirmation(proposal, _plan(), now=NOW) + assert CLAIM_MISSING in result["refusals"] + + +def test_foreign_claim_shape_refuses() -> None: + proposal = _proposal() + proposal["operation"]["claim"] = {"claimed_at": "2026-10-17T03:59:00Z"} + result = verify_bound_confirmation(proposal, _plan(), now=NOW) + assert CLAIM_MISSING in result["refusals"] + + +def test_missing_payload_terms_refuses() -> None: + proposal = _proposal() + payload = _payload_of(proposal) + payload.pop("bound_order") + _store_payload(proposal, payload) + result = verify_bound_confirmation(proposal, _plan(), now=NOW) + assert PAYLOAD_TERMS_MISSING in result["refusals"] + + +def test_terms_drift_refuses() -> None: + plan = _plan() + plan["terms"]["amount_usdc"] = "500" + result = verify_bound_confirmation(_proposal(), plan, now=NOW) + assert TERMS_MISMATCH in result["refusals"] + assert result["evidence"]["terms_digest"] != result["evidence"]["expected_terms_digest"] + + +def test_missing_canonical_deadline_refuses() -> None: + proposal = _proposal() + proposal["operation"].pop("expires_at") + result = verify_bound_confirmation(proposal, _plan(), now=NOW) + assert result["allowed"] is False + assert DEADLINE_MISSING in result["refusals"] + + +def test_invalid_deadline_refuses() -> None: + proposal = _proposal() + proposal["operation"]["expires_at"] = "not-a-timestamp" + result = verify_bound_confirmation(proposal, _plan(), now=NOW) + assert DEADLINE_MISSING in result["refusals"] + plan = _plan() + plan["expires_at"] = "31/12/2026" + result = verify_bound_confirmation(_proposal(), plan, now=NOW) + assert DEADLINE_INVALID in result["refusals"] + + +def test_deadline_boundary_is_inclusive() -> None: + exact = datetime(2026, 10, 17, 5, 0, tzinfo=timezone.utc) + assert EXPIRED in verify_bound_confirmation(_proposal(), _plan(), now=exact)["refusals"] + before = datetime(2026, 10, 17, 4, 59, 59, tzinfo=timezone.utc) + assert verify_bound_confirmation(_proposal(), _plan(), now=before)["allowed"] is True + + +def test_conflicting_deadlines_all_enforced() -> None: + """A later payload deadline must not extend an expired canonical envelope.""" + + proposal = _proposal() + payload = _payload_of(proposal) + payload["expires_at"] = "2026-10-17T09:00:00Z" + _store_payload(proposal, payload) + later = datetime(2026, 10, 17, 6, 0, tzinfo=timezone.utc) + result = verify_bound_confirmation(proposal, _plan(), now=later) + assert EXPIRED in result["refusals"] + assert result["allowed"] is False + + +def test_plan_deadline_is_independent() -> None: + plan = _plan() + plan["expires_at"] = "2026-10-17T04:30:00Z" + later = datetime(2026, 10, 17, 4, 45, tzinfo=timezone.utc) + result = verify_bound_confirmation(_proposal(), plan, now=later) + assert EXPIRED in result["refusals"]