diff --git a/gateway/app/services/approval_gate.py b/gateway/app/services/approval_gate.py index 9a122e00..a4bc6f6a 100644 --- a/gateway/app/services/approval_gate.py +++ b/gateway/app/services/approval_gate.py @@ -41,11 +41,11 @@ from datetime import datetime, timezone from typing import Any, Callable, Literal, TypeVar, cast -from sqlalchemy import select, update +from sqlalchemy import and_, or_, select, update from sqlalchemy.sql.elements import ColumnElement from sqlmodel import col -from app.core.time import to_naive_utc, utc_now +from app.core.time import to_naive_utc from app.db.approval_gate_models import ApprovalGateModel from app.db.database import get_session_factory @@ -162,12 +162,32 @@ class ApprovalRecord: class ApprovalStore: """Interface for pending approvals so a DB-backed store can replace memory.""" - def save(self, record: ApprovalRecord) -> None: + def save(self, record: ApprovalRecord, *, now: float | None = None) -> None: raise NotImplementedError def get(self, nonce: str) -> ApprovalRecord | None: raise NotImplementedError + def find_active_by_key( + self, idempotency_key: str, *, now: float + ) -> ApprovalRecord | None: + """Live record for a key: pending, or approved and unconsumed. + + Only unexpired records (``deadline`` after ``now``) count; None when + no such record exists. + """ + raise NotImplementedError + + def find_consumed_by_key( + self, idempotency_key: str, *, now: float + ) -> ApprovalRecord | None: + """Live spent record for a key: consumed with deadline after now. + + Used at creation to refuse a fresh mint while a prior spend's + deadline is still live; None when no such record exists. + """ + raise NotImplementedError + def list_pending(self, *, now: float | None = None) -> list[ApprovalRecord]: """Pending records, excluding ones past deadline when now is given.""" raise NotImplementedError @@ -184,7 +204,9 @@ def decide( def consume(self, nonce: str, *, now: float) -> ApprovalRecord | None: """Single-use: first call on approved marks consumed; later calls None. - Expired approvals (``deadline`` past ``now``) never consume. + Expired approvals (``now`` at or past ``deadline``) never consume, + and neither does a nonce whose idempotency key already spent on + another live record. """ raise NotImplementedError @@ -192,7 +214,7 @@ def mark_expired(self, nonce: str, *, now: float) -> ApprovalRecord | None: """Flip pending to expired; None when missing.""" raise NotImplementedError - async def asave(self, record: ApprovalRecord) -> None: + async def asave(self, record: ApprovalRecord, *, now: float | None = None) -> None: """Async ``save`` for handlers already on the app event loop.""" raise NotImplementedError @@ -200,6 +222,18 @@ async def aget(self, nonce: str) -> ApprovalRecord | None: """Async ``get`` for handlers already on the app event loop.""" raise NotImplementedError + async def afind_active_by_key( + self, idempotency_key: str, *, now: float + ) -> ApprovalRecord | None: + """Async ``find_active_by_key`` for handlers already on the app loop.""" + raise NotImplementedError + + async def afind_consumed_by_key( + self, idempotency_key: str, *, now: float + ) -> ApprovalRecord | None: + """Async ``find_consumed_by_key`` for handlers on the app loop.""" + raise NotImplementedError + async def alist_pending(self, *, now: float | None = None) -> list[ApprovalRecord]: """Async ``list_pending`` for handlers already on the app event loop.""" raise NotImplementedError @@ -226,7 +260,8 @@ def __init__(self) -> None: self._lock = threading.Lock() self._records: dict[str, ApprovalRecord] = {} - def save(self, record: ApprovalRecord) -> None: + def save(self, record: ApprovalRecord, *, now: float | None = None) -> None: + del now # in-memory records carry their own timestamps; nothing to stamp. with self._lock: self._records[record.nonce] = replace(record) @@ -235,6 +270,39 @@ def get(self, nonce: str) -> ApprovalRecord | None: record = self._records.get(nonce) return replace(record) if record else None + def find_active_by_key( + self, idempotency_key: str, *, now: float + ) -> ApprovalRecord | None: + with self._lock: + candidates = [ + record + for record in self._records.values() + if record.idempotency_key == idempotency_key + and record.deadline > now + and ( + record.status == "pending" + or (record.status == "approved" and not record.consumed) + ) + ] + if not candidates: + return None + return replace(min(candidates, key=lambda r: (r.created_at, r.nonce))) + + def find_consumed_by_key( + self, idempotency_key: str, *, now: float + ) -> ApprovalRecord | None: + with self._lock: + candidates = [ + record + for record in self._records.values() + if record.idempotency_key == idempotency_key + and record.consumed + and record.deadline > now + ] + if not candidates: + return None + return replace(min(candidates, key=lambda r: (r.created_at, r.nonce))) + def list_pending(self, *, now: float | None = None) -> list[ApprovalRecord]: with self._lock: return [ @@ -267,6 +335,20 @@ def consume(self, nonce: str, *, now: float) -> ApprovalRecord | None: return None if record.deadline <= now: return None + for other in self._records.values(): + if ( + other.nonce != nonce + and other.idempotency_key == record.idempotency_key + and other.consumed + and other.deadline > now + ): + # Spent sibling blocks this spend: terminalize to denied + # under the same lock so the approval can never revive + # past the sibling's deadline. + self._records[nonce] = replace( + record, status="denied", decided_at=now + ) + return None updated = replace(record, consumed=True) self._records[nonce] = updated return replace(updated) @@ -281,12 +363,22 @@ def mark_expired(self, nonce: str, *, now: float) -> ApprovalRecord | None: self._records[nonce] = record return replace(record) - async def asave(self, record: ApprovalRecord) -> None: - self.save(record) + async def asave(self, record: ApprovalRecord, *, now: float | None = None) -> None: + self.save(record, now=now) async def aget(self, nonce: str) -> ApprovalRecord | None: return self.get(nonce) + async def afind_active_by_key( + self, idempotency_key: str, *, now: float + ) -> ApprovalRecord | None: + return self.find_active_by_key(idempotency_key, now=now) + + async def afind_consumed_by_key( + self, idempotency_key: str, *, now: float + ) -> ApprovalRecord | None: + return self.find_consumed_by_key(idempotency_key, now=now) + async def alist_pending(self, *, now: float | None = None) -> list[ApprovalRecord]: return self.list_pending(now=now) @@ -336,7 +428,23 @@ def _from_db_time(value: datetime) -> float: return value.replace(tzinfo=timezone.utc).timestamp() -def _model_from_record(record: ApprovalRecord) -> ApprovalGateModel: +def _model_from_record( + record: ApprovalRecord, *, now: float | None = None +) -> ApprovalGateModel: + if record.consumed: + # The consume UPDATE stamps the caller's now; a consumed record + # reaching save() (merge) falls back to the closest record timestamp + # so no wall-clock read sneaks in beside the gate clock. + stamp = now + if stamp is None: + stamp = ( + record.decided_at + if record.decided_at is not None + else record.created_at + ) + consumed_at = _to_db_time(stamp) + else: + consumed_at = None return ApprovalGateModel( nonce=record.nonce, idempotency_key=record.idempotency_key, @@ -350,7 +458,7 @@ def _model_from_record(record: ApprovalRecord) -> ApprovalGateModel: decided_at=( _to_db_time(record.decided_at) if record.decided_at is not None else None ), - consumed_at=utc_now() if record.consumed else None, + consumed_at=consumed_at, ) @@ -384,13 +492,13 @@ class DbApprovalStore(ApprovalStore): retries can replay it, and an expired approval can never spend. """ - def save(self, record: ApprovalRecord) -> None: - _run_sync(self.asave(record)) + def save(self, record: ApprovalRecord, *, now: float | None = None) -> None: + _run_sync(self.asave(record, now=now)) - async def asave(self, record: ApprovalRecord) -> None: + async def asave(self, record: ApprovalRecord, *, now: float | None = None) -> None: factory = get_session_factory() async with factory() as session: - await session.merge(_model_from_record(record)) + await session.merge(_model_from_record(record, now=now)) await session.commit() def get(self, nonce: str) -> ApprovalRecord | None: @@ -407,6 +515,91 @@ async def aget(self, nonce: str) -> ApprovalRecord | None: model = result.scalar_one_or_none() return _record_from_model(model) if model is not None else None + def find_active_by_key( + self, idempotency_key: str, *, now: float + ) -> ApprovalRecord | None: + return _run_sync(self.afind_active_by_key(idempotency_key, now=now)) + + async def afind_active_by_key( + self, idempotency_key: str, *, now: float + ) -> ApprovalRecord | None: + factory = get_session_factory() + async with factory() as session: + result = await session.execute( + select(ApprovalGateModel) + .where( + cast( + ColumnElement[bool], + ApprovalGateModel.idempotency_key == idempotency_key, + ), + cast( + ColumnElement[bool], + ApprovalGateModel.deadline > _to_db_time(now), + ), + cast( + ColumnElement[bool], + or_( + cast( + ColumnElement[bool], + ApprovalGateModel.status == "pending", + ), + and_( + cast( + ColumnElement[bool], + ApprovalGateModel.status == "approved", + ), + cast( + ColumnElement[bool], + cast(Any, ApprovalGateModel.consumed_at).is_(None), + ), + ), + ), + ), + ) + .order_by( + col(ApprovalGateModel.created_at).asc(), + col(ApprovalGateModel.nonce).asc(), + ) + .limit(1) + ) + model = result.scalar_one_or_none() + return _record_from_model(model) if model is not None else None + + def find_consumed_by_key( + self, idempotency_key: str, *, now: float + ) -> ApprovalRecord | None: + return _run_sync(self.afind_consumed_by_key(idempotency_key, now=now)) + + async def afind_consumed_by_key( + self, idempotency_key: str, *, now: float + ) -> ApprovalRecord | None: + factory = get_session_factory() + async with factory() as session: + result = await session.execute( + select(ApprovalGateModel) + .where( + cast( + ColumnElement[bool], + ApprovalGateModel.idempotency_key == idempotency_key, + ), + cast( + ColumnElement[bool], + cast(Any, ApprovalGateModel.consumed_at).is_not(None), + ), + cast( + ColumnElement[bool], + ApprovalGateModel.deadline > _to_db_time(now), + ), + ) + .order_by( + col(ApprovalGateModel.created_at).asc(), + col(ApprovalGateModel.nonce).asc(), + ) + .limit(1) + ) + model = result.scalar_one_or_none() + return _record_from_model(model) if model is not None else None + def list_pending(self, *, now: float | None = None) -> list[ApprovalRecord]: return _run_sync(self.alist_pending(now=now)) @@ -493,9 +686,48 @@ async def aconsume(self, nonce: str, *, now: float) -> ApprovalRecord | None: one caller (one retry, one transport) wins, a restart cannot resurrect a spent approval, and an approval whose deadline passed between the gate's expiry check and this UPDATE still cannot spend. + The ``NOT EXISTS`` sibling check extends the single spend to the + whole idempotency key: a second nonce approved for the same key + cannot spend while the first spend's deadline has not passed. When + the sibling check is what blocks the spend, a second conditional + UPDATE terminalizes this nonce to denied so it can never revive + past the sibling's deadline. """ factory = get_session_factory() async with factory() as session: + # The key never changes after insert, so reading it first is safe; + # the NOT EXISTS check itself still runs inside the UPDATE below. + key_result = await session.execute( + select(ApprovalGateModel).where( + cast(ColumnElement[bool], ApprovalGateModel.nonce == nonce) + ) + ) + key_model = key_result.scalar_one_or_none() + if key_model is None: + return None + key = key_model.idempotency_key + + def _sibling_live() -> Any: + return ( + select(1) + .where( + cast( + ColumnElement[bool], + ApprovalGateModel.idempotency_key == key, + ), + cast(ColumnElement[bool], ApprovalGateModel.nonce != nonce), + cast( + ColumnElement[bool], + cast(Any, ApprovalGateModel.consumed_at).is_not(None), + ), + cast( + ColumnElement[bool], + ApprovalGateModel.deadline > _to_db_time(now), + ), + ) + .exists() + ) + updated = await session.execute( update(ApprovalGateModel) .where( @@ -512,19 +744,45 @@ async def aconsume(self, nonce: str, *, now: float) -> ApprovalRecord | None: ColumnElement[bool], ApprovalGateModel.deadline > _to_db_time(now), ), + cast(ColumnElement[bool], ~_sibling_live()), ) - .values(consumed_at=utc_now()) + .values(consumed_at=_to_db_time(now)) ) - await session.commit() - if cast(Any, updated).rowcount != 1: - return None - result = await session.execute( - select(ApprovalGateModel).where( - cast(ColumnElement[bool], ApprovalGateModel.nonce == nonce) + if cast(Any, updated).rowcount == 1: + await session.commit() + result = await session.execute( + select(ApprovalGateModel).where( + cast(ColumnElement[bool], ApprovalGateModel.nonce == nonce) + ) ) + model = result.scalar_one_or_none() + return _record_from_model(model) if model is not None else None + # Spend lost: when a live spent sibling is the blocker, move this + # nonce to denied in one conditional UPDATE so it stays refused + # even after the sibling's deadline passes. Other losses + # (missing, denied, consumed, expired) match no row here. + await session.execute( + update(ApprovalGateModel) + .where( + cast(ColumnElement[bool], ApprovalGateModel.nonce == nonce), + cast( + ColumnElement[bool], + ApprovalGateModel.status == "approved", + ), + cast( + ColumnElement[bool], + cast(Any, ApprovalGateModel.consumed_at).is_(None), + ), + cast( + ColumnElement[bool], + ApprovalGateModel.deadline > _to_db_time(now), + ), + cast(ColumnElement[bool], _sibling_live()), + ) + .values(status="denied", decided_at=_to_db_time(now)) ) - model = result.scalar_one_or_none() - return _record_from_model(model) if model is not None else None + await session.commit() + return None def mark_expired(self, nonce: str, *, now: float) -> ApprovalRecord | None: return _run_sync(self.amark_expired(nonce, now=now)) @@ -564,6 +822,16 @@ async def amark_expired(self, nonce: str, *, now: float) -> ApprovalRecord | Non def _log_receipt_sink(event: str, record: dict[str, Any]) -> None: + """Default gate receipt route: one structured log line per event. + + Deliberately not the durable ReceiptService pipeline: create_receipt is + an async database write that needs permit/wallet/ledger identity the gate + does not have, and the sync evaluate() path cannot safely touch the + loop-bound engine. Every gate event (denied, timeout, tampered, ..., + consumed) flows through this single sink, so none is silently more or + less durable than the others; callers that need durability inject a + receipt_sink. + """ logger.info( "approval_receipt event=%s record=%s", event, json.dumps(record, sort_keys=True) ) @@ -578,6 +846,33 @@ class ApprovalDecision: nonce: str | None = None +def _seal_pause_state( + secret: bytes, + *, + args_hash: str, + caller: str, + tool: str, + session_id: str, + idempotency_key: str, + deadline: float, + nonce: str, +) -> str: + """Seal the retry state for one pause, fresh or reused.""" + return _seal_state( + secret, + { + "v": STATE_VERSION, + "argsHash": args_hash, + "principal": caller, + "tool": tool, + "sessionId": session_id, + "idempotencyKey": idempotency_key, + "deadline": deadline, + "nonce": nonce, + }, + ) + + def _pause_payload(tool: str, nonce: str, request_state: str) -> dict[str, Any]: """Build the SEP-2322 ``input_required`` pause payload. @@ -814,10 +1109,13 @@ def _check_record( @staticmethod def _is_expired(now: float, sealed_deadline: Any, record: ApprovalRecord) -> bool: + # Expired when now >= deadline, matching the SQL conditions + # (deadline > now admits only the live). The in-memory store already + # uses deadline <= now / deadline > now, so all three agree. return ( not isinstance(sealed_deadline, (int, float)) - or now > sealed_deadline - or now > record.deadline + or now >= sealed_deadline + or now >= record.deadline or record.status == "expired" ) @@ -842,6 +1140,37 @@ def _pause_new_call( return ApprovalDecision(action="REJECT", reason="unsupported_client") secret = self._require_secret() now = self.clock() + spent = self.store.find_consumed_by_key(idempotency_key, now=now) + if spent is not None: + # Key already spent and the spend window is still live: refuse + # without minting so no second spendable nonce exists. + self._emit( + "replayed", + nonce=None, + tool=tool, + principal=caller, + session_id=session_id, + ) + return ApprovalDecision(action="REJECT", reason="replayed") + existing = self.store.find_active_by_key(idempotency_key, now=now) + if existing is not None: + # Same logical call already paused: reissue its pause instead of + # minting a second spendable approval for one idempotency key. + state = _seal_pause_state( + secret, + args_hash=args_hash, + caller=caller, + tool=tool, + session_id=session_id, + idempotency_key=idempotency_key, + deadline=existing.deadline, + nonce=existing.nonce, + ) + return ApprovalDecision( + action="PAUSE", + nonce=existing.nonce, + payload=_pause_payload(tool, existing.nonce, state), + ) nonce = secrets.token_urlsafe(16) deadline = now + self._ttl self.store.save( @@ -854,20 +1183,18 @@ def _pause_new_call( session_id=session_id, created_at=now, deadline=deadline, - ) + ), + now=now, ) - state = _seal_state( + state = _seal_pause_state( secret, - { - "v": STATE_VERSION, - "argsHash": args_hash, - "principal": caller, - "tool": tool, - "sessionId": session_id, - "idempotencyKey": idempotency_key, - "deadline": deadline, - "nonce": nonce, - }, + args_hash=args_hash, + caller=caller, + tool=tool, + session_id=session_id, + idempotency_key=idempotency_key, + deadline=deadline, + nonce=nonce, ) return ApprovalDecision( action="PAUSE", @@ -896,6 +1223,37 @@ async def _apause_new_call( return ApprovalDecision(action="REJECT", reason="unsupported_client") secret = self._require_secret() now = self.clock() + spent = await self.store.afind_consumed_by_key(idempotency_key, now=now) + if spent is not None: + # Key already spent and the spend window is still live: refuse + # without minting so no second spendable nonce exists. + self._emit( + "replayed", + nonce=None, + tool=tool, + principal=caller, + session_id=session_id, + ) + return ApprovalDecision(action="REJECT", reason="replayed") + existing = await self.store.afind_active_by_key(idempotency_key, now=now) + if existing is not None: + # Same logical call already paused: reissue its pause instead of + # minting a second spendable approval for one idempotency key. + state = _seal_pause_state( + secret, + args_hash=args_hash, + caller=caller, + tool=tool, + session_id=session_id, + idempotency_key=idempotency_key, + deadline=existing.deadline, + nonce=existing.nonce, + ) + return ApprovalDecision( + action="PAUSE", + nonce=existing.nonce, + payload=_pause_payload(tool, existing.nonce, state), + ) nonce = secrets.token_urlsafe(16) deadline = now + self._ttl await self.store.asave( @@ -908,20 +1266,18 @@ async def _apause_new_call( session_id=session_id, created_at=now, deadline=deadline, - ) + ), + now=now, ) - state = _seal_state( + state = _seal_pause_state( secret, - { - "v": STATE_VERSION, - "argsHash": args_hash, - "principal": caller, - "tool": tool, - "sessionId": session_id, - "idempotencyKey": idempotency_key, - "deadline": deadline, - "nonce": nonce, - }, + args_hash=args_hash, + caller=caller, + tool=tool, + session_id=session_id, + idempotency_key=idempotency_key, + deadline=deadline, + nonce=nonce, ) return ApprovalDecision( action="PAUSE", diff --git a/gateway/scripts/amw_approvals.py b/gateway/scripts/amw_approvals.py index 4ee597fc..09e93cd5 100644 --- a/gateway/scripts/amw_approvals.py +++ b/gateway/scripts/amw_approvals.py @@ -15,6 +15,7 @@ from __future__ import annotations import argparse +import ipaddress import os import sys from datetime import datetime, timezone @@ -29,6 +30,96 @@ API_KEY_ENV = "VALID_API_KEYS" API_KEY_HEADER = "X-API-Key" +_LOOPBACK_V4 = ipaddress.ip_network("127.0.0.0/8") +_LOOPBACK_V6 = ipaddress.ip_address("::1") + + +def _raw_authority(base_url: str) -> str | None: + """Authority after '://' up to / ? #; None when there is no '://'.""" + sep = base_url.find("://") + if sep == -1: + return None + rest = base_url[sep + 3 :] + for i, ch in enumerate(rest): + if ch in "/?#": + return rest[:i] + return rest + + +def _raw_port(authority: str) -> str | None: + """Port substring from a raw authority; None when absent or malformed.""" + hostport = authority.rsplit("@", 1)[-1] + if hostport.startswith("["): + end = hostport.find("]") + if end == -1: + return None + after = hostport[end + 1 :] + if not after: + return None + if after.startswith(":"): + return after[1:] + return None + if ":" not in hostport: + return None + return hostport.rsplit(":", 1)[-1] + + +def _refusal_reason(base_url: str) -> str | None: + """Why the API key must not be sent to base_url; None when allowed. + + Allowed: any https URL, or http only on loopback (localhost, 127.0.0.0/8, + ::1). Control characters, whitespace, and non-numeric ports refuse + before the allowed returns. Any userinfo (username or password) refuses + before the allowed returns, so http://guest@localhost still refuses. + Otherwise the host comes from httpx.URL.host, so userinfo tricks such as + http://127.0.0.1@evil resolve to the real host (evil), and look-alikes + such as localhost.example or 127.0.0.1.nip.io fail both the exact + localhost match and the IP parse. Unknown schemes and empty hosts refuse. + The check uses httpx.URL, the same parser as the request. IDNA decode + errors on host access (malformed punycode) refuse as unparseable. + """ + for ch in base_url: + if ord(ch) < 32 or ord(ch) == 127 or ch.isspace(): + return "invalid characters in URL" + try: + parsed = httpx.URL(base_url) + except (httpx.InvalidURL, UnicodeError, ValueError): + return "unparseable URL" + scheme = parsed.scheme.lower() + if scheme not in ("http", "https"): + if parsed.scheme: + return f"unsupported scheme {parsed.scheme!r}" + return "missing URL scheme" + try: + host = parsed.host or "" + has_userinfo = bool(parsed.username or parsed.password or parsed.userinfo) + except (UnicodeError, ValueError): + # httpx decodes IDNA on .host access, so malformed punycode + # (e.g. http://xn--) raises here, not at URL construction. + return "unparseable URL" + if not host: + return "missing host" + if has_userinfo: + return "userinfo not allowed" + authority = _raw_authority(base_url) + if authority is not None and "@" in authority: + return "userinfo not allowed" + if authority is not None: + port_str = _raw_port(authority) + if port_str and not (port_str.isascii() and port_str.isdigit()): + return f"invalid port {port_str!r}" + if scheme == "https": + return None + if host == "localhost": + return None + try: + addr = ipaddress.ip_address(host) + except ValueError: + return f"non-loopback http host {host!r}" + if addr in _LOOPBACK_V4 or addr == _LOOPBACK_V6: + return None + return f"non-loopback http host {host!r}" + def _read_api_key() -> str: """First key from API_KEY_ENV; empty when unset or blank.""" @@ -39,9 +130,22 @@ def _read_api_key() -> str: def _client(base_url: str, api_key: str) -> httpx.Client: - return httpx.Client( - base_url=base_url, timeout=10.0, headers={API_KEY_HEADER: api_key} - ) + reason = _refusal_reason(base_url) + if reason is not None: + raise ValueError(f"refusing to send the API key: {reason}") + # trust_env=False: httpx honors ambient HTTP_PROXY/HTTPS_PROXY/ALL_PROXY + # by default, which would hand the bootstrap admin key to a proxy. + try: + return httpx.Client( + base_url=base_url, + timeout=10.0, + headers={API_KEY_HEADER: api_key}, + trust_env=False, + ) + except (httpx.InvalidURL, UnicodeError, ValueError): + # All other arguments are constants, so any decode error here is + # URL-driven and refuses cleanly without ever printing the key. + raise ValueError("refusing to send the API key: unparseable URL") from None def _format_time(value: float) -> str: @@ -54,6 +158,12 @@ def cmd_list(args: argparse.Namespace) -> int: try: response = _client(args.base_url, args.api_key).get("/v1/approvals/pending") response.raise_for_status() + except ValueError as exc: + print(f"error: {exc}", file=sys.stderr) + return 1 + except httpx.InvalidURL as exc: + print(f"error: invalid gateway URL: {exc}", file=sys.stderr) + return 1 except httpx.HTTPError as exc: print( f"error: gateway not reachable at {args.base_url}: {exc}", file=sys.stderr @@ -76,6 +186,12 @@ def _decide(args: argparse.Namespace, action: str) -> int: response = _client(args.base_url, args.api_key).post( f"/v1/approvals/{args.nonce}/{action}" ) + except ValueError as exc: + print(f"error: {exc}", file=sys.stderr) + return 1 + except httpx.InvalidURL as exc: + print(f"error: invalid gateway URL: {exc}", file=sys.stderr) + return 1 except httpx.HTTPError as exc: print( f"error: gateway not reachable at {args.base_url}: {exc}", file=sys.stderr @@ -120,6 +236,17 @@ def build_parser() -> argparse.ArgumentParser: def main(argv: list[str] | None = None) -> int: args = build_parser().parse_args(argv) + # Validate the destination before the key is read, so a refusal never + # attaches the bootstrap admin key to a request. + reason = _refusal_reason(args.base_url) + if reason is not None: + print( + "error: refusing to send the API key: " + f"{reason}; use an https URL or http on loopback " + "(localhost, 127.0.0.0/8, ::1)", + file=sys.stderr, + ) + return 1 api_key = _read_api_key() if not api_key: print( diff --git a/gateway/tests/test_amw_approvals_cli.py b/gateway/tests/test_amw_approvals_cli.py index a8e908da..3a2f7227 100644 --- a/gateway/tests/test_amw_approvals_cli.py +++ b/gateway/tests/test_amw_approvals_cli.py @@ -84,7 +84,9 @@ def __init__( base_url: str = "", timeout: float | None = None, headers: dict[str, str] | None = None, + trust_env: bool = True, ) -> None: + assert trust_env is False self._headers = dict(headers or {}) def get(self, path: str) -> _StubResponse: @@ -116,3 +118,233 @@ def test_unreachable_gateway_error_hides_key( out = capsys.readouterr() assert PLACEHOLDER_KEY not in out.out assert PLACEHOLDER_KEY not in out.err + + +# Fixed documentation-only fixtures. Every host below is reserved and +# never routes to a real peer; no test in this file opens a real socket +# for refused URLs (see _ExplodingClient). +EXAMPLE_COM = "example.com" # RFC 2606, reserved example domain +EXAMPLE_INVALID = "example.invalid" # RFC 2606, invalid TLD never resolves +LOCALHOST_NAME = "localhost" # RFC 6761, special-use loopback name +TEST_NET_1 = "192.0.2.1" # RFC 5737, TEST-NET-1 documentation-only +TEST_NET_2 = "198.51.100.7" # RFC 5737, TEST-NET-2 documentation-only +DOC_IPV6 = "2001:db8::1" # RFC 3849, documentation-only IPv6 +LOOPBACK_V4 = "127.0.0.1" # RFC 1122, IPv4 loopback +LOOPBACK_V4_OTHER = "127.0.0.2" # RFC 1122, same 127/8 loopback range +LOOPBACK_V6 = "::1" # RFC 4291, IPv6 loopback +PRIVATE_10 = "10.0.0.1" # RFC 1918, private-use address +PRIVATE_192_168 = "192.168.0.1" # RFC 1918, private-use address +HTTPS_TEST_PORT = 8443 # fixed test port in the user range (RFC 6335) +HTTP_TEST_PORT = 8080 # fixed test port in the user range (RFC 6335) + + +def _allowed_url(case: str) -> str: + if case == "https-any-host": + return f"https://[{DOC_IPV6}]:{HTTPS_TEST_PORT}" + if case == "https-no-port": + return f"https://{EXAMPLE_COM}" + if case == "http-localhost": + return f"http://{LOCALHOST_NAME}:{HTTP_TEST_PORT}" + if case == "http-127-1": + return f"http://{LOOPBACK_V4}:{HTTP_TEST_PORT}" + if case == "http-127-range": + return f"http://{LOOPBACK_V4_OTHER}:{HTTP_TEST_PORT}" + if case == "http-ipv6-loopback": + return f"http://[{LOOPBACK_V6}]:{HTTP_TEST_PORT}" + if case == "http-uppercase-localhost": + return f"HTTP://{LOCALHOST_NAME}:{HTTP_TEST_PORT}" + raise AssertionError(f"unknown allowed case: {case}") + + +def _refused_url(case: str) -> str: + if case == "http-rfc1918-10": + return f"http://{PRIVATE_10}:{HTTP_TEST_PORT}" + if case == "http-rfc1918-192-168": + return f"http://{PRIVATE_192_168}:{HTTP_TEST_PORT}" + if case == "http-public-hostname": + return f"http://{EXAMPLE_COM}:{HTTP_TEST_PORT}" + if case == "http-userinfo-trick": + return f"http://{LOOPBACK_V4}@{EXAMPLE_INVALID}:{HTTP_TEST_PORT}" + if case == "http-localhost-userinfo-trick": + return f"http://{LOCALHOST_NAME}@{EXAMPLE_INVALID}:{HTTP_TEST_PORT}/" + if case == "http-userinfo-localhost": + # RFC 6761 localhost, RFC 6335 fixed port; userinfo refuses outright. + return f"http://guest@{LOCALHOST_NAME}:{HTTP_TEST_PORT}" + if case == "https-userinfo-example": + # RFC 2606 example.com; userinfo refuses even on allowed https. + return f"https://guest@{EXAMPLE_COM}" + if case == "http-malformed-ipv6-bracket": + # RFC 3849 documentation IPv6, missing closing bracket: ValueError. + return f"http://[{DOC_IPV6}" + if case == "http-localhost-lookalike": + return f"http://{LOCALHOST_NAME}.{EXAMPLE_INVALID}:{HTTP_TEST_PORT}" + if case == "http-ip-lookalike": + return f"http://{LOOPBACK_V4}.{EXAMPLE_INVALID}:{HTTP_TEST_PORT}" + if case == "ftp-loopback": + return f"ftp://{LOCALHOST_NAME}:{HTTP_TEST_PORT}/" + if case == "file-scheme": + return "file:///etc/hosts" + if case == "missing-host": + return "http:///no-host" + if case == "empty": + return "" + if case == "no-scheme": + return f"{LOCALHOST_NAME}:{HTTP_TEST_PORT}" + if case == "http-tab-in-host": + # RFC 6761 localhost variant with a tab; control char refuses. + return f"http://local\thost:{HTTP_TEST_PORT}" + if case == "http-non-numeric-port": + # RFC 6761 localhost; non-numeric port refuses. + return f"http://{LOCALHOST_NAME}:abc" + if case == "https-newline": + # RFC 2606 example.com; newline refuses even on https. + return f"https://{EXAMPLE_COM}\n/" + if case == "http-malformed-idna-label": + # RFC 3492/5891 punycode, malformed on purpose; .invalid is RFC 2606. + return f"http://xn--.{EXAMPLE_INVALID}:{HTTP_TEST_PORT}" + raise AssertionError(f"unknown refused case: {case}") + + +@pytest.mark.parametrize( + "case", + [ + "https-any-host", + "https-no-port", + "http-localhost", + "http-127-1", + "http-127-range", + "http-ipv6-loopback", + "http-uppercase-localhost", + ], +) +def test_allowed_base_url_sends_request( + monkeypatch: pytest.MonkeyPatch, capsys: pytest.CaptureFixture[str], case: str +) -> None: + sent: list[str] = [] + + class _RecordingClient: + def __init__(self, *args: object, **kwargs: object) -> None: + pass + + def get(self, path: str) -> _StubResponse: + sent.append(path) + return _StubResponse({"approvals": []}) + + monkeypatch.setattr(amw_approvals.httpx, "Client", _RecordingClient) + monkeypatch.setenv("VALID_API_KEYS", PLACEHOLDER_KEY) + assert amw_approvals.main(["--base-url", _allowed_url(case), "list"]) == 0 + assert sent == ["/v1/approvals/pending"] + out = capsys.readouterr() + assert PLACEHOLDER_KEY not in out.out + assert PLACEHOLDER_KEY not in out.err + + +class _ExplodingClient: + def __init__(self, *args: object, **kwargs: object) -> None: + raise AssertionError("refused URL must not open an HTTP client") + + +@pytest.mark.parametrize( + "case", + [ + "http-rfc1918-10", + "http-rfc1918-192-168", + "http-public-hostname", + "http-userinfo-trick", + "http-localhost-userinfo-trick", + "http-userinfo-localhost", + "https-userinfo-example", + "http-malformed-ipv6-bracket", + "http-localhost-lookalike", + "http-ip-lookalike", + "ftp-loopback", + "file-scheme", + "missing-host", + "empty", + "no-scheme", + "http-tab-in-host", + "http-non-numeric-port", + "https-newline", + "http-malformed-idna-label", + ], +) +def test_refused_base_url_makes_no_request( + monkeypatch: pytest.MonkeyPatch, capsys: pytest.CaptureFixture[str], case: str +) -> None: + monkeypatch.setattr(amw_approvals.httpx, "Client", _ExplodingClient) + monkeypatch.setenv("VALID_API_KEYS", PLACEHOLDER_KEY) + base_url = _refused_url(case) + nonce = "test-nonce" + for argv in (["list"], ["approve", nonce], ["deny", nonce]): + assert amw_approvals.main(["--base-url", base_url, *argv]) == 1 + out = capsys.readouterr() + assert "refusing to send the API key" in out.err + assert "Traceback" not in out.err + assert "Traceback" not in out.out + assert PLACEHOLDER_KEY not in out.err + assert PLACEHOLDER_KEY not in out.out + + +def test_refusal_happens_before_key_check( + monkeypatch: pytest.MonkeyPatch, capsys: pytest.CaptureFixture[str] +) -> None: + monkeypatch.setattr(amw_approvals.httpx, "Client", _ExplodingClient) + monkeypatch.delenv("VALID_API_KEYS", raising=False) + url = f"http://{TEST_NET_1}:{HTTP_TEST_PORT}" + assert amw_approvals.main(["--base-url", url, "list"]) == 1 + err = capsys.readouterr().err + assert "refusing to send the API key" in err + assert "VALID_API_KEYS" not in err + + +def test_refused_env_base_url_makes_no_request( + monkeypatch: pytest.MonkeyPatch, capsys: pytest.CaptureFixture[str] +) -> None: + monkeypatch.setattr(amw_approvals.httpx, "Client", _ExplodingClient) + monkeypatch.setenv("VALID_API_KEYS", PLACEHOLDER_KEY) + monkeypatch.setenv( + amw_approvals.BASE_URL_ENV, + f"http://{EXAMPLE_INVALID}:{HTTP_TEST_PORT}", + ) + assert amw_approvals.main(["list"]) == 1 + out = capsys.readouterr() + assert "refusing to send the API key" in out.err + assert PLACEHOLDER_KEY not in out.out + assert PLACEHOLDER_KEY not in out.err + + +def test_client_refuses_disallowed_url_directly() -> None: + urls = [ + f"http://{TEST_NET_2}:{HTTP_TEST_PORT}", + # RFC 3492/5891 punycode, malformed on purpose; .invalid is RFC 2606. + f"http://xn--.{EXAMPLE_INVALID}:{HTTP_TEST_PORT}", + ] + for url in urls: + with pytest.raises(ValueError, match="refusing to send the API key") as exc: + amw_approvals._client(url, PLACEHOLDER_KEY) + assert PLACEHOLDER_KEY not in str(exc.value) + + +def test_client_disables_trust_env_and_ignores_proxies( + monkeypatch: pytest.MonkeyPatch, +) -> None: + # RFC 5737 TEST-NET-1, RFC 6335 fixed port: proxy decoy, never dialed. + monkeypatch.setenv("HTTP_PROXY", f"http://{TEST_NET_1}:{HTTP_TEST_PORT}") + # RFC 5737 TEST-NET-2, RFC 6335 fixed port: proxy decoy, never dialed. + monkeypatch.setenv("ALL_PROXY", f"http://{TEST_NET_2}:{HTTP_TEST_PORT}") + # RFC 6761 localhost, RFC 6335 fixed port: allowed http loopback URL. + client = amw_approvals._client( + f"http://{LOCALHOST_NAME}:{HTTP_TEST_PORT}", PLACEHOLDER_KEY + ) + try: + assert client.trust_env is False + assert dict(client._mounts) == {} + # Same env with trust_env=True mounts proxies, proving the decoys + # above are live env httpx would otherwise honor. No request is sent. + proving = amw_approvals.httpx.Client(trust_env=True) + try: + assert len(proving._mounts) > 0 + finally: + proving.close() + finally: + client.close() diff --git a/gateway/tests/test_approval_gate.py b/gateway/tests/test_approval_gate.py index 7a45b990..ea414bf4 100644 --- a/gateway/tests/test_approval_gate.py +++ b/gateway/tests/test_approval_gate.py @@ -7,6 +7,7 @@ import asyncio import json +import logging import secrets import sys import uuid @@ -26,7 +27,7 @@ import pytest # noqa: E402 from goose_amw import canonical as goose_canonical # noqa: E402 -from sqlalchemy import delete # noqa: E402 +from sqlalchemy import delete, select # noqa: E402 import app.services.approval_gate as gate_module # noqa: E402 from app.db.approval_gate_models import ApprovalGateModel # noqa: E402 @@ -731,3 +732,521 @@ async def test_db_alist_pending_excludes_expired( assert second.nonce is not None pending = await DbApprovalStore().alist_pending(now=clock()) assert [record.nonce for record in pending] == [second.nonce] + + +def test_exact_deadline_pending_refuses_gate_and_store( + gate: ApprovalGate, clock: _Clock, receipts: list[tuple[str, dict[str, Any]]] +) -> None: + """At now == deadline the record is expired for both gate and store.""" + decision = _pause(gate) + assert decision.nonce is not None + assert decision.payload is not None + record = gate.store.get(decision.nonce) + assert record is not None + clock.now = record.deadline + assert gate.store.decide(decision.nonce, True, now=clock()) is None + rejected = _retry(gate, decision.payload["requestState"], decision.nonce) + assert rejected.action == "REJECT" + assert rejected.reason == "timeout" + assert gate.store.get(decision.nonce) is not None + assert gate.store.get(decision.nonce).status == "expired" # type: ignore[union-attr] + assert [event for event, _ in receipts] == ["timeout"] + + +def test_exact_deadline_approved_refuses_gate_and_store( + gate: ApprovalGate, clock: _Clock, receipts: list[tuple[str, dict[str, Any]]] +) -> None: + """An approval cannot spend at its exact deadline instant.""" + decision = _pause(gate) + assert decision.nonce is not None + assert decision.payload is not None + assert gate.store.decide(decision.nonce, True, now=clock()) is not None + record = gate.store.get(decision.nonce) + assert record is not None + clock.now = record.deadline + assert gate.store.consume(decision.nonce, now=clock()) is None + assert gate.store.get(decision.nonce) is not None + assert gate.store.get(decision.nonce).consumed is False # type: ignore[union-attr] + rejected = _retry(gate, decision.payload["requestState"], decision.nonce) + assert rejected.action == "REJECT" + assert rejected.reason == "timeout" + assert [event for event, _ in receipts] == ["timeout"] + + +async def test_db_exact_deadline_approved_refuses_gate_and_store( + db_gate: ApprovalGate, + clock: _Clock, + receipts: list[tuple[str, dict[str, Any]]], + clean_database: Any, + purge_approval_gate_table: Any, +) -> None: + """The atomic consume UPDATE and the gate agree at now == deadline.""" + decision = await _apause(db_gate) + assert decision.nonce is not None + assert decision.payload is not None + store = DbApprovalStore() + assert await store.adecide(decision.nonce, True, now=clock()) is not None + record = await store.aget(decision.nonce) + assert record is not None + clock.now = record.deadline + assert await store.aconsume(decision.nonce, now=clock()) is None + rejected = await _aretry(db_gate, decision.payload["requestState"], decision.nonce) + assert rejected.action == "REJECT" + assert rejected.reason == "timeout" + assert [event for event, _ in receipts] == ["timeout"] + + +async def test_db_consume_stamps_passed_now( + db_gate: ApprovalGate, + clock: _Clock, + clean_database: Any, + purge_approval_gate_table: Any, +) -> None: + """decided_at/consumed_at come from the gate clock, not wall clock.""" + decision = await _apause(db_gate) + assert decision.nonce is not None + store = DbApprovalStore() + assert await store.adecide(decision.nonce, True, now=clock()) is not None + assert await store.aconsume(decision.nonce, now=clock()) is not None + expected = gate_module._to_db_time(clock()) + factory = get_session_factory() + async with factory() as session: + result = await session.execute( + select(ApprovalGateModel).where(ApprovalGateModel.nonce == decision.nonce) + ) + model = result.scalar_one() + assert model.decided_at == expected + assert model.consumed_at == expected + + +def test_second_first_call_reuses_pending_record(gate: ApprovalGate) -> None: + """A repeat first call for one key reissues the same pause.""" + first = _pause(gate) + second = gate.evaluate(CALLER, TOOL, ARGS, SESSION, True) + assert second.action == "PAUSE" + assert second.nonce == first.nonce + assert len(gate.store.list_pending()) == 1 + + +def test_first_call_reuses_approved_unconsumed_record( + gate: ApprovalGate, +) -> None: + """Reuse covers approved-but-unspent records, and the pause still spends.""" + first = _pause(gate) + assert first.nonce is not None + assert gate.store.decide(first.nonce, True, now=gate.clock()) is not None + second = gate.evaluate(CALLER, TOOL, ARGS, SESSION, True) + assert second.action == "PAUSE" + assert second.nonce == first.nonce + assert second.payload is not None + executed = _retry(gate, second.payload["requestState"], second.nonce) + assert executed.action == "EXECUTE" + + +@pytest.mark.parametrize("terminal", ["denied", "expired"]) +def test_first_call_mints_after_terminal_record( + gate: ApprovalGate, clock: _Clock, terminal: str +) -> None: + """Denied and expired records are not reused for a fresh call.""" + first = _pause(gate) + assert first.nonce is not None + assert first.payload is not None + if terminal == "denied": + assert gate.store.decide(first.nonce, False, now=gate.clock()) is not None + else: + clock.now += 3600 + 1 + second = gate.evaluate(CALLER, TOOL, ARGS, SESSION, True) + assert second.action == "PAUSE" + assert second.nonce != first.nonce + + +async def test_async_second_first_call_reuses_active_record( + gate: ApprovalGate, +) -> None: + first = await gate.aevaluate(CALLER, TOOL, ARGS, SESSION, True) + second = await gate.aevaluate(CALLER, TOOL, ARGS, SESSION, True) + assert second.action == "PAUSE" + assert second.nonce == first.nonce + + +async def test_db_second_first_call_reuses_active_record( + db_gate: ApprovalGate, + clock: _Clock, + clean_database: Any, + purge_approval_gate_table: Any, +) -> None: + first = await _apause(db_gate) + second = await _apause(db_gate) + assert second.action == "PAUSE" + assert second.nonce == first.nonce + pending = await DbApprovalStore().alist_pending(now=clock()) + assert [record.nonce for record in pending] == [first.nonce] + + +def _seal_duplicate(nonce: str, deadline: float) -> str: + """Seal retry state for a duplicate nonce the way the pause path does.""" + return gate_module._seal_pause_state( + SECRET.encode("utf-8"), + args_hash=goose_canonical.args_hash(ARGS), + caller=CALLER, + tool=TOOL, + session_id=SESSION, + idempotency_key=goose_canonical.idempotency_key( + caller=CALLER, tool=TOOL, arguments=ARGS, session_id=SESSION + ), + deadline=deadline, + nonce=nonce, + ) + + +def _duplicate_nonce(first_nonce: str) -> str: + nonce = secrets.token_urlsafe(16) + while nonce == first_nonce: + nonce = secrets.token_urlsafe(16) + return nonce + + +def test_two_approved_pauses_same_key_execute_exactly_once( + gate: ApprovalGate, receipts: list[tuple[str, dict[str, Any]]] +) -> None: + """Two approved nonces for one key spend exactly once between them.""" + first = _pause(gate) + assert first.nonce is not None + assert first.payload is not None + record = gate.store.get(first.nonce) + assert record is not None + # A duplicate pause for the same key (minted before reuse, or raced in): + # same key fields, fresh nonce, saved directly. + duplicate = _duplicate_nonce(first.nonce) + gate.store.save(replace(record, nonce=duplicate), now=gate.clock()) + assert gate.store.decide(first.nonce, True, now=gate.clock()) is not None + assert gate.store.decide(duplicate, True, now=gate.clock()) is not None + executed = _retry(gate, first.payload["requestState"], first.nonce) + assert executed.action == "EXECUTE" + replayed = _retry(gate, _seal_duplicate(duplicate, record.deadline), duplicate) + assert replayed.action == "REJECT" + assert replayed.reason == "replayed" + assert [event for event, _ in receipts] == ["consumed", "replayed"] + + +async def test_db_two_approved_pauses_same_key_execute_exactly_once( + db_gate: ApprovalGate, + clock: _Clock, + receipts: list[tuple[str, dict[str, Any]]], + clean_database: Any, + purge_approval_gate_table: Any, +) -> None: + """The NOT EXISTS consume guard holds on the DB path, either order.""" + decision = await _apause(db_gate) + assert decision.nonce is not None + assert decision.payload is not None + store = DbApprovalStore() + record = await store.aget(decision.nonce) + assert record is not None + duplicate = _duplicate_nonce(decision.nonce) + await store.asave(replace(record, nonce=duplicate), now=clock()) + assert await store.adecide(decision.nonce, True, now=clock()) is not None + assert await store.adecide(duplicate, True, now=clock()) is not None + # Spend the duplicate first: the original must then refuse. + executed = await _aretry( + db_gate, _seal_duplicate(duplicate, record.deadline), duplicate + ) + assert executed.action == "EXECUTE" + replayed = await _aretry(db_gate, decision.payload["requestState"], decision.nonce) + assert replayed.action == "REJECT" + assert replayed.reason == "replayed" + assert [event for event, _ in receipts] == ["consumed", "replayed"] + + +def test_default_sink_routes_consumed_like_other_events( + clock: _Clock, caplog: pytest.LogCaptureFixture +) -> None: + """Without an injected sink, consumed logs exactly like other events.""" + gate = ApprovalGate(secret=SECRET, store=InMemoryApprovalStore(), clock=clock) + with caplog.at_level(logging.INFO, logger="app.services.approval_gate"): + first = gate.evaluate(CALLER, TOOL, ARGS, SESSION, True) + assert first.nonce is not None + assert first.payload is not None + assert gate.store.decide(first.nonce, True, now=gate.clock()) is not None + executed = _retry(gate, first.payload["requestState"], first.nonce) + assert executed.action == "EXECUTE" + # Past the spend window a fresh pause mints normally to deny. + spent = gate.store.get(first.nonce) + assert spent is not None + clock.now = spent.deadline + 1 + second = gate.evaluate(CALLER, TOOL, ARGS, SESSION, True) + assert second.nonce is not None + assert second.nonce != first.nonce + assert second.payload is not None + assert gate.store.decide(second.nonce, False, now=gate.clock()) is not None + denied = _retry(gate, second.payload["requestState"], second.nonce) + assert denied.action == "REJECT" + assert denied.reason == "denied" + lines = [ + record.getMessage() + for record in caplog.records + if record.name == "app.services.approval_gate" + ] + assert any("event=consumed" in line for line in lines) + assert any("event=denied" in line for line in lines) + assert all(line.startswith("approval_receipt event=") for line in lines) + + +def test_first_call_refuses_while_spent_live( + gate: ApprovalGate, clock: _Clock, receipts: list[tuple[str, dict[str, Any]]] +) -> None: + """Within the spend window a fresh call refuses without minting.""" + first = _pause(gate) + assert first.nonce is not None + assert first.payload is not None + assert gate.store.decide(first.nonce, True, now=gate.clock()) is not None + executed = _retry(gate, first.payload["requestState"], first.nonce) + assert executed.action == "EXECUTE" + second = gate.evaluate(CALLER, TOOL, ARGS, SESSION, True) + assert second.action == "REJECT" + assert second.reason == "replayed" + assert second.nonce is None + assert gate.store.list_pending(now=clock()) == [] + assert [event for event, _ in receipts] == ["consumed", "replayed"] + + +async def test_async_first_call_refuses_while_spent_live( + gate: ApprovalGate, clock: _Clock, receipts: list[tuple[str, dict[str, Any]]] +) -> None: + """Async pause path refuses a fresh mint while the spend is live.""" + first = await _apause(gate) + assert first.nonce is not None + assert first.payload is not None + assert await gate.store.adecide(first.nonce, True, now=gate.clock()) is not None + executed = await _aretry(gate, first.payload["requestState"], first.nonce) + assert executed.action == "EXECUTE" + second = await gate.aevaluate(CALLER, TOOL, ARGS, SESSION, True) + assert second.action == "REJECT" + assert second.reason == "replayed" + assert second.nonce is None + assert await gate.store.alist_pending(now=clock()) == [] + assert [event for event, _ in receipts] == ["consumed", "replayed"] + + +async def test_db_first_call_refuses_while_spent_live( + db_gate: ApprovalGate, + clock: _Clock, + receipts: list[tuple[str, dict[str, Any]]], + clean_database: Any, + purge_approval_gate_table: Any, +) -> None: + """DB pause path refuses a fresh mint while the spend is live.""" + first = await _apause(db_gate) + assert first.nonce is not None + assert first.payload is not None + store = DbApprovalStore() + assert await store.adecide(first.nonce, True, now=clock()) is not None + executed = await _aretry(db_gate, first.payload["requestState"], first.nonce) + assert executed.action == "EXECUTE" + second = await db_gate.aevaluate(CALLER, TOOL, ARGS, SESSION, True) + assert second.action == "REJECT" + assert second.reason == "replayed" + assert second.nonce is None + assert await store.alist_pending(now=clock()) == [] + assert [event for event, _ in receipts] == ["consumed", "replayed"] + + +def test_sibling_refused_stays_refused_past_spent_deadline( + gate: ApprovalGate, clock: _Clock +) -> None: + """A sibling blocked by a live spend stays refused past that deadline.""" + first = _pause(gate) + assert first.nonce is not None + assert first.payload is not None + record = gate.store.get(first.nonce) + assert record is not None + duplicate = _duplicate_nonce(first.nonce) + gate.store.save( + replace(record, nonce=duplicate, deadline=record.deadline + 500), + now=gate.clock(), + ) + assert gate.store.decide(first.nonce, True, now=gate.clock()) is not None + assert gate.store.decide(duplicate, True, now=gate.clock()) is not None + executed = _retry(gate, first.payload["requestState"], first.nonce) + assert executed.action == "EXECUTE" + assert gate.store.consume(duplicate, now=clock()) is None + blocked = gate.store.get(duplicate) + assert blocked is not None + assert blocked.status == "denied" + spent = gate.store.get(first.nonce) + assert spent is not None + clock.now = spent.deadline + 1 + assert clock() < blocked.deadline + assert gate.store.consume(duplicate, now=clock()) is None + assert gate.store.get(duplicate) is not None + assert gate.store.get(duplicate).status == "denied" # type: ignore[union-attr] + + +async def test_db_sibling_refused_stays_refused_past_spent_deadline( + db_gate: ApprovalGate, + clock: _Clock, + clean_database: Any, + purge_approval_gate_table: Any, +) -> None: + """DB sibling blocked by a live spend stays refused past that deadline.""" + first = await _apause(db_gate) + assert first.nonce is not None + assert first.payload is not None + store = DbApprovalStore() + record = await store.aget(first.nonce) + assert record is not None + duplicate = _duplicate_nonce(first.nonce) + await store.asave( + replace(record, nonce=duplicate, deadline=record.deadline + 500), + now=clock(), + ) + assert await store.adecide(first.nonce, True, now=clock()) is not None + assert await store.adecide(duplicate, True, now=clock()) is not None + executed = await _aretry(db_gate, first.payload["requestState"], first.nonce) + assert executed.action == "EXECUTE" + assert await store.aconsume(duplicate, now=clock()) is None + blocked = await store.aget(duplicate) + assert blocked is not None + assert blocked.status == "denied" + spent = await store.aget(first.nonce) + assert spent is not None + clock.now = spent.deadline + 1 + assert clock() < blocked.deadline + assert await store.aconsume(duplicate, now=clock()) is None + refused = await store.aget(duplicate) + assert refused is not None + assert refused.status == "denied" + + +def test_first_call_mints_after_spent_window(gate: ApprovalGate, clock: _Clock) -> None: + """Past the spend window a brand-new call mints and spends normally.""" + first = _pause(gate) + assert first.nonce is not None + assert first.payload is not None + assert gate.store.decide(first.nonce, True, now=gate.clock()) is not None + executed = _retry(gate, first.payload["requestState"], first.nonce) + assert executed.action == "EXECUTE" + spent = gate.store.get(first.nonce) + assert spent is not None + clock.now = spent.deadline + 1 + second = gate.evaluate(CALLER, TOOL, ARGS, SESSION, True) + assert second.action == "PAUSE" + assert second.nonce is not None + assert second.nonce != first.nonce + assert second.payload is not None + assert gate.store.decide(second.nonce, True, now=gate.clock()) is not None + replay = _retry(gate, second.payload["requestState"], second.nonce) + assert replay.action == "EXECUTE" + + +async def test_db_first_call_mints_after_spent_window( + db_gate: ApprovalGate, + clock: _Clock, + clean_database: Any, + purge_approval_gate_table: Any, +) -> None: + """DB path mints and spends normally past the spend window.""" + first = await _apause(db_gate) + assert first.nonce is not None + assert first.payload is not None + store = DbApprovalStore() + assert await store.adecide(first.nonce, True, now=clock()) is not None + executed = await _aretry(db_gate, first.payload["requestState"], first.nonce) + assert executed.action == "EXECUTE" + spent = await store.aget(first.nonce) + assert spent is not None + clock.now = spent.deadline + 1 + second = await db_gate.aevaluate(CALLER, TOOL, ARGS, SESSION, True) + assert second.action == "PAUSE" + assert second.nonce is not None + assert second.nonce != first.nonce + assert second.payload is not None + assert await store.adecide(second.nonce, True, now=clock()) is not None + replay = await _aretry(db_gate, second.payload["requestState"], second.nonce) + assert replay.action == "EXECUTE" + + +def test_first_call_prefers_spent_over_live_active_sibling( + gate: ApprovalGate, clock: _Clock, receipts: list[tuple[str, dict[str, Any]]] +) -> None: + """A live spend beats a live approved sibling on a fresh call.""" + first = _pause(gate) + assert first.nonce is not None + assert first.payload is not None + record = gate.store.get(first.nonce) + assert record is not None + duplicate = _duplicate_nonce(first.nonce) + gate.store.save(replace(record, nonce=duplicate), now=gate.clock()) + assert gate.store.decide(first.nonce, True, now=gate.clock()) is not None + assert gate.store.decide(duplicate, True, now=gate.clock()) is not None + executed = _retry(gate, first.payload["requestState"], first.nonce) + assert executed.action == "EXECUTE" + key = goose_canonical.idempotency_key( + caller=CALLER, tool=TOOL, arguments=ARGS, session_id=SESSION + ) + assert gate.store.find_consumed_by_key(key, now=clock()) is not None + assert gate.store.find_active_by_key(key, now=clock()) is not None + second = gate.evaluate(CALLER, TOOL, ARGS, SESSION, True) + assert second.action == "REJECT" + assert second.reason == "replayed" + assert second.nonce is None + assert [event for event, _ in receipts] == ["consumed", "replayed"] + + +async def test_async_first_call_prefers_spent_over_live_active_sibling( + gate: ApprovalGate, clock: _Clock, receipts: list[tuple[str, dict[str, Any]]] +) -> None: + """Async pause path also refuses when a live spend and sibling coexist.""" + first = await _apause(gate) + assert first.nonce is not None + assert first.payload is not None + record = await gate.store.aget(first.nonce) + assert record is not None + duplicate = _duplicate_nonce(first.nonce) + await gate.store.asave(replace(record, nonce=duplicate), now=gate.clock()) + assert await gate.store.adecide(first.nonce, True, now=gate.clock()) is not None + assert await gate.store.adecide(duplicate, True, now=gate.clock()) is not None + executed = await _aretry(gate, first.payload["requestState"], first.nonce) + assert executed.action == "EXECUTE" + key = goose_canonical.idempotency_key( + caller=CALLER, tool=TOOL, arguments=ARGS, session_id=SESSION + ) + assert await gate.store.afind_consumed_by_key(key, now=clock()) is not None + assert await gate.store.afind_active_by_key(key, now=clock()) is not None + second = await gate.aevaluate(CALLER, TOOL, ARGS, SESSION, True) + assert second.action == "REJECT" + assert second.reason == "replayed" + assert second.nonce is None + assert [event for event, _ in receipts] == ["consumed", "replayed"] + + +async def test_db_first_call_prefers_spent_over_live_active_sibling( + db_gate: ApprovalGate, + clock: _Clock, + receipts: list[tuple[str, dict[str, Any]]], + clean_database: Any, + purge_approval_gate_table: Any, +) -> None: + """DB pause path refuses when a live spend and sibling coexist.""" + first = await _apause(db_gate) + assert first.nonce is not None + assert first.payload is not None + store = DbApprovalStore() + record = await store.aget(first.nonce) + assert record is not None + duplicate = _duplicate_nonce(first.nonce) + await store.asave(replace(record, nonce=duplicate), now=clock()) + assert await store.adecide(first.nonce, True, now=clock()) is not None + assert await store.adecide(duplicate, True, now=clock()) is not None + executed = await _aretry(db_gate, first.payload["requestState"], first.nonce) + assert executed.action == "EXECUTE" + key = goose_canonical.idempotency_key( + caller=CALLER, tool=TOOL, arguments=ARGS, session_id=SESSION + ) + assert await store.afind_consumed_by_key(key, now=clock()) is not None + assert await store.afind_active_by_key(key, now=clock()) is not None + second = await db_gate.aevaluate(CALLER, TOOL, ARGS, SESSION, True) + assert second.action == "REJECT" + assert second.reason == "replayed" + assert second.nonce is None + assert [event for event, _ in receipts] == ["consumed", "replayed"]