diff --git a/CHANGELOG.md b/CHANGELOG.md index 1fb787cf..16597615 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -11,6 +11,84 @@ The next release consolidates the accumulated trust-plane and public-product work as `v1.3.0`. Create that tag only from the exact commit that passes the full release gate; do not backfill a final `v1.2.0` tag. +### ๐Ÿ”’ Production refuses a SQLite `DATABASE_URL` at boot + +- **The premise behind eleven of the twelve audit findings no longer holds.** + Those findings were live because `SELECT ... FOR UPDATE` is a silent no-op on + SQLite and nothing refused SQLite in production. `validate_trust_mode_config` + now refuses a SQLite `DATABASE_URL` โ€” file *and* in-memory spellings โ€” in + production-like environments, and refuses an empty one, since wallets, + permits, receipts, and the ledger are relational and `get_engine()` returns + `None` without it. +- This is a **separate control** from the `STATE_BACKEND` guard added in #304. + That one governs the key/value durable-state store; this one governs the ORM + engine the money and permit paths actually write through. The gap between + them was reachable: `STATE_BACKEND=redis` with a `REDIS_URL` satisfies the + state check completely while `DATABASE_URL` stays SQLite, which is precisely + the posture `tests/test_trust_mode_guardrails.py::test_a_redis_state_backend_does_not_excuse_a_sqlite_database_url` + now pins. +- Local development is untouched: SQLite remains the standard value outside + production-like environments, and a regression test asserts it stays + accepted there. + +### ๐Ÿ”’ A dry-run session no longer confirms another tenant's session exists + +- **`403` where a `404` was due.** Every `/v1/billing/dry-run/session` + endpoint looked the session up first and checked wallet access second, so a + caller holding any valid wallet-scoped key got `404` for an invented session + id and `403` for a real one โ€” enough to tell them apart with no access at + all. The `403` body also carried the session's *owning* `wallet_id`, + disclosing another tenant's wallet id outright. +- A session the caller may not see now answers exactly as one that does not + exist, on all five endpoints that accept a session id (`GET`, `DELETE`, + `commit`, `revert`, and `dry-run/charge`). Authorized callers are + unaffected โ€” a regression test pins that the owner still reads its own + session. +- Session ids are UUID4 and cannot be enumerated, so exploiting this needed an + id learned elsewhere (a log, a trace, a shared URL). + +### ๐Ÿงพ Metering and velocity hardening + +- **A rejected charge no longer decrements a velocity period it never + contributed to.** The hourly and daily counters roll over independently, and + a rollover between the increment and its reversal left the reversal taking + credits off the *new* period's total. Each counter is now reversed under the + period marker it was recorded against, separately, so a rolled daily window + cannot suppress an hourly reversal that is still owed. The direction is the + point: an over-count throttles a caller who did not spend and heals at the + next rollover, while an under-count silently raises the effective spend cap + and delays the anomaly auto-freeze. +- **`/v1/billing/charge` answers 404 for an unknown wallet** instead of letting + `WalletNotFoundError` escape as a 500 โ€” the six other endpoints in that + router already did. +- **Negative and zero `units` are refused at the billing engine.** A negative + `units` made `charge_amount` negative, and the guarded debit then read + `balance >= charge_amount` as trivially true and applied + `balance - charge_amount` โ€” *raising* the balance. Reproduced against the + unfixed engine: a wallet at 100 charged `units=-5` ended at 110, with a + ledger entry recording `action="debit", amount=+10`, so the audit trail + agreed it was a charge. The router refused it through `gt=0`; the governed + MCP path and the SDK do not pass through the router. +- **The dry-run endpoint validates `units` too.** `POST /v1/billing/dry-run/charge` + has two branches: with a session id it calls `ShadowLedger.simulate_charge` + directly and never reaches `BillingEngine.charge`, so the engine guard did + not cover it; without one it reached the engine and the guard escaped as a + 500. `SimulatedChargeRequest.units` now carries the same `gt=0` and + `allow_inf_nan=False` constraints as the real charge, refusing both branches + with a 422 before either is chosen. +- **Non-finite `units` are refused**, at the query boundary + (`allow_inf_nan=False`) and again in `BillingEngine.charge`, which the + governed MCP path and the SDK reach without passing the router. `gt=0` does + not exclude an infinite float, and one reaching a write would put a + non-finite amount in the ledger that no reconciliation can remove. +- **Four guarded writes from the previous release gained the regression tests + they shipped without**: the `refund_charge` credit, `_apply_refund`, the + operator refund's permit release, and the Stripe fiat clawback. Each was + confirmed to fail against the read-modify-write it replaced. +- Tests that force a concurrent writer into an open transaction are now skipped + when `DATABASE_URL` names a real-locking engine. They depend on the lock + being a no-op; against a real one they would hang rather than fail. + ### ๐Ÿ”Œ Standard `/mcp` endpoint now served by the official MCP SDK - **The opt-in `POST /mcp` endpoint no longer hand-rolls JSON-RPC.** The diff --git a/README.md b/README.md index 5e51b106..f2aac25b 100644 --- a/README.md +++ b/README.md @@ -659,16 +659,18 @@ become stale; the CI badge and release gates are the source of truth. | [`migrations/`](migrations/) | Alembic history; use migrations for production schema changes | | [`scripts/`](scripts/) | Reproducible demos, release gates, OpenAPI export, and operator helpers | | [`tests/`](tests/) | Product, negative-path, concurrency, and proof-surface tests | -| [`b2a_sdk/`](b2a_sdk/) | Python trust SDK 0.4.0 source and release build | +| [`b2a_sdk/`](b2a_sdk/) | Python trust SDK 0.5.0 source and release build | | [`framework_integrations/`](framework_integrations/) | Source examples for agent frameworks; not published packages | | [`site/`](site/) | Static marketing and discovery pointer site | -### Python SDK 0.4.0 +### Python SDK 0.5.0 HTTP and MCP are the canonical integration surfaces. CI builds and smoke-tests -Python SDK 0.4.0 wheels and sdists on Python 3.10 through 3.12. Pushing the -matching `python-sdk-v0.4.0` tag attaches those artifacts to a GitHub release; -the package is not published to PyPI. The typed `AgentMiddlewareClient` covers +Python SDK 0.5.0 wheels and sdists on Python 3.10 through 3.12. Pushing the +matching `python-sdk-v0.5.0` tag attaches those artifacts to a GitHub release; +the package is not published to PyPI. The newest tag today is +`python-sdk-v0.4.0` โ€” the source here is ahead of it, and the wheel a release +build produces carries the source version, not the tag's. The typed `AgentMiddlewareClient` covers tool discovery, permit creation, governed invocation, receipt verification, and evidence retrieval, and exposes idempotency conflicts and delivery uncertainty as explicit errors. For repository development (installs from this repo, not diff --git a/TROUBLESHOOTING.md b/TROUBLESHOOTING.md index aac7dc3a..c0e9555c 100644 --- a/TROUBLESHOOTING.md +++ b/TROUBLESHOOTING.md @@ -34,6 +34,24 @@ You set `ENVIRONMENT=production` but used SQLite or left `STATE_BACKEND=sqlite`. - Switch to `ENVIRONMENT=local` for local development, **or** - Set `STATE_BACKEND=postgres` and provide a `DATABASE_URL` with `postgresql+asyncpg://`. +### `TrustModeGuardrailError: DATABASE_URL must not be SQLite in production-like environments` +You set `ENVIRONMENT=production` (or another production-like value) with a +SQLite `DATABASE_URL`. SQLAlchemy silently drops `SELECT ... FOR UPDATE` on +SQLite, so concurrent charges, budget reservations, and velocity counters lose +the serialization the money and permit paths depend on โ€” two callers can each +debit a balance only one of them fits inside. Either: +- Switch to `ENVIRONMENT=local` for local development, **or** +- Point `DATABASE_URL` at PostgreSQL (`postgresql+asyncpg://...`). + +This is a different check from `STATE_BACKEND` below: that one governs the +key/value state store, this one governs the ORM engine. Satisfying one does +not satisfy the other. + +### `TrustModeGuardrailError: DATABASE_URL must be set in production-like environments` +Wallets, permits, receipts, and the ledger are relational. Without +`DATABASE_URL` the engine is never created and the trust plane has nowhere +durable to record what it authorized. Set it to your PostgreSQL DSN. + ### `RuntimeError: ENABLE_PROOF_SURFACES must be false in production` Set `ENABLE_PROOF_SURFACES=false`. Proof surfaces are for local demos only. diff --git a/app/core/db_urls.py b/app/core/db_urls.py index b9d80880..3bacfbdd 100644 --- a/app/core/db_urls.py +++ b/app/core/db_urls.py @@ -49,6 +49,19 @@ def is_postgres_url(url: str) -> bool: ) +def is_sqlite_url(url: str) -> bool: + """True when ``url`` names SQLite, in-memory spellings included. + + This is the question ``sqlite_path_from_url`` deliberately does not + answer. That helper returns "" for an in-memory URL because it is looking + for a *durable file*; a caller asking "is this SQLite?" for safety reasons + needs the opposite treatment, since ``sqlite+aiosqlite:///:memory:`` is + every bit as unsuitable for a production trust plane as a file is -- more + so, since it also loses everything on restart. + """ + return (url or "").strip().lower().startswith(("sqlite://", "sqlite+")) + + def sqlite_path_from_url(url: str) -> str: """Return the **durable** SQLite file path in ``url``, or "" otherwise. diff --git a/app/core/trust_mode.py b/app/core/trust_mode.py index 3a2b1c3c..46ce2d5b 100644 --- a/app/core/trust_mode.py +++ b/app/core/trust_mode.py @@ -7,6 +7,8 @@ from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey +from app.core.db_urls import is_sqlite_url + if TYPE_CHECKING: from app.core.config import Settings @@ -82,6 +84,7 @@ def validate_trust_mode_config( enable_public_mcp_endpoint: bool = False, redis_url: str = "", public_url: str = "", + database_url: str = "", ) -> None: """Refuse unsafe deploy postures in production-like environments. @@ -147,6 +150,37 @@ def validate_trust_mode_config( "in production-like environments (anonymous verification must " "use shared rate-limit state)" ) + configured_database_url = (database_url or "").strip() + if not configured_database_url: + violations.append( + "DATABASE_URL must be set in production-like environments " + "(wallets, permits, receipts, and the ledger are relational; " + "without it get_engine() returns None and the trust plane has " + "nowhere durable to record what it authorized)" + ) + elif is_sqlite_url(configured_database_url): + # STATE_BACKEND already refuses SQLite (see + # app/core/durable_state.py), but that guard covers the key/value + # state store, not the ORM engine this URL builds. A deployment + # setting STATE_BACKEND=redis with REDIS_URL satisfies it while + # DATABASE_URL stays SQLite -- and the money and permit paths run + # against the ORM engine, not the state store. Every + # ``SELECT ... FOR UPDATE`` guarding a balance, a budget, or a + # counter is silently dropped by SQLAlchemy on SQLite, so the + # serialization those paths are written to rely on is simply + # absent. Refuse at boot rather than let concurrent writers + # discover it against real money. + violations.append( + "DATABASE_URL must not be SQLite in production-like " + "environments: SQLAlchemy silently drops SELECT ... FOR UPDATE " + "on SQLite, so concurrent charges, budget reservations, and " + "velocity counters lose the serialization the money and permit " + "paths depend on. Use PostgreSQL " + "(postgresql+asyncpg://...). This is a separate control from " + "STATE_BACKEND, which governs the key/value state store rather " + "than the ORM engine" + ) + if enable_public_mcp_endpoint and not (public_url or "").strip(): violations.append( "PUBLIC_URL is required when ENABLE_PUBLIC_MCP_ENDPOINT is true " @@ -172,6 +206,7 @@ def validate_trust_mode_guardrails(settings: Settings) -> None: enable_public_mcp_endpoint=settings.ENABLE_PUBLIC_MCP_ENDPOINT, redis_url=settings.REDIS_URL, public_url=settings.PUBLIC_URL, + database_url=settings.DATABASE_URL, ) diff --git a/app/routers/billing.py b/app/routers/billing.py index a812601d..4d180c2e 100644 --- a/app/routers/billing.py +++ b/app/routers/billing.py @@ -62,6 +62,47 @@ def _require_wallet_access(auth: AuthContext, wallet_id: str) -> None: auth.require_wallet_access(wallet_id) +def _session_not_found(session_id: str) -> HTTPException: + return HTTPException( + status_code=status.HTTP_404_NOT_FOUND, + detail={ + "error": "session_not_found", + "message": f"Session {session_id} not found", + }, + ) + + +async def _load_owned_dry_run_session( + shadow_ledger, session_id: str, auth: AuthContext +): + """Load a dry-run session, or 404 โ€” whether it is missing or not the caller's. + + Every dry-run session endpoint used to look the session up, 404 if it was + absent, and only then call ``_require_wallet_access``. That order makes the + surface an oracle: a caller holding any valid wallet-scoped key gets 404 for + an invented session id and 403 for a real one, so the two are + distinguishable without any access at all. + + Worse, the 403 body carries ``session.wallet_id`` โ€” the *owning* wallet โ€” + so probing another tenant's session id disclosed that tenant's wallet id + outright. Session ids are UUID4 and cannot be enumerated, but an id learned + from a log, a trace, or a shared URL was enough. + + A session the caller may not see is now indistinguishable from one that + does not exist. Authorized callers are unaffected. + """ + session = await shadow_ledger.get_session(session_id) + if not session: + raise _session_not_found(session_id) + try: + _require_wallet_access(auth, session.wallet_id) + except HTTPException as exc: + if exc.status_code == status.HTTP_403_FORBIDDEN: + raise _session_not_found(session_id) from None + raise + return session + + @dataclass class _IdempotencyGuard: """Opt-in replay protection for a state-changing money endpoint. @@ -548,7 +589,19 @@ async def charge_wallet( None, description="Service category (alias for service_category)", ), - units: float = Query(1.0, gt=0, description="Number of units consumed"), + units: float = Query( + 1.0, + gt=0, + # ``gt=0`` alone lets an infinite value through -- it is a valid + # float and it is greater than zero -- and metering then computes + # ``Decimal("Infinity") - Decimal("Infinity")``, which raises + # InvalidOperation and answers 500. The crash is the lucky outcome: + # an infinite units count that reached a write would put a + # non-finite amount in the ledger, which no reconciliation can undo. + # (A not-a-number value is already refused: it fails ``gt=0``.) + allow_inf_nan=False, + description="Number of units consumed", + ), request_path: str | None = None, description: str | None = None, idempotency_key: str | None = Header(None, alias="Idempotency-Key"), @@ -701,6 +754,35 @@ async def _complete_idempotency(response_json: dict, status_code: int) -> None: ) await _complete_idempotency(result.model_dump(mode="json"), 200) return result + except WalletNotFoundError as e: + # Every other endpoint in this router answers 404 for an unknown + # wallet; this one let the exception escape as a 500, which reads as + # "the server is broken" rather than "that wallet does not exist" and + # sends a caller retrying a typo down the wrong path entirely. + await _record_billing_governance( + event="billing.charge", + auth=auth, + wallet_id=wallet_id, + service_category=category.value, + endpoint=endpoint, + request_id=request_id, + ok=False, + error="wallet_not_found", + metadata={"units": units, "request_path": request_path}, + ) + not_found_detail = { + "error": "wallet_not_found", + "wallet_id": wallet_id, + "message": str(e), + } + await _complete_idempotency( + {"detail": not_found_detail}, + status.HTTP_404_NOT_FOUND, + ) + raise HTTPException( + status_code=status.HTTP_404_NOT_FOUND, + detail=not_found_detail, + ) except WalletFrozenError as e: if category: await _record_billing_governance( @@ -1218,7 +1300,20 @@ class SimulatedChargeRequest(BaseModel): wallet_id: str = Field(..., description="Wallet being simulated") service: ServiceCategory = Field(..., description="Service category to simulate") - units: float = Field(default=1.0, description="Number of units") + units: float = Field( + default=1.0, + # Same constraints as the real charge query parameter, for the same + # reasons and one extra. The session-based branch of simulate_charge + # calls ShadowLedger.simulate_charge directly and never reaches + # BillingEngine.charge, so the engine's guard does not cover it at + # all; the session-less branch does reach the engine, where the guard + # raises ValueError and surfaces as a 500 rather than a 422. Refusing + # here fixes both: one clean rejection at the boundary, before either + # branch is chosen. + gt=0, + allow_inf_nan=False, + description="Number of units", + ) description: str | None = Field(None, description="Optional description") dry_run_session_id: str | None = Field( None, @@ -1310,18 +1405,7 @@ async def get_dry_run_session( ): """Get the current state of a dry-run session.""" shadow_ledger = get_shadow_ledger() - session = await shadow_ledger.get_session(session_id) - - if not session: - raise HTTPException( - status_code=status.HTTP_404_NOT_FOUND, - detail={ - "error": "session_not_found", - "message": f"Session {session_id} not found", - }, - ) - - _require_wallet_access(auth, session.wallet_id) + session = await _load_owned_dry_run_session(shadow_ledger, session_id, auth) return { "session_id": session.session_id, "wallet_id": session.wallet_id, @@ -1357,16 +1441,7 @@ async def end_dry_run_session( ): """End a dry-run session and return summary.""" shadow_ledger = get_shadow_ledger() - session = await shadow_ledger.get_session(session_id) - if not session: - raise HTTPException( - status_code=status.HTTP_404_NOT_FOUND, - detail={ - "error": "session_not_found", - "message": f"Session {session_id} not found", - }, - ) - _require_wallet_access(auth, session.wallet_id) + await _load_owned_dry_run_session(shadow_ledger, session_id, auth) summary = await shadow_ledger.end_session(session_id) if summary is None: # Session existed above but is gone by the time we tried to end it @@ -1413,17 +1488,21 @@ async def commit_dry_run_session( The session is ended after committing. """ shadow_ledger = get_shadow_ledger() - session = await shadow_ledger.get_session(session_id) - if not session: - raise HTTPException( - status_code=status.HTTP_404_NOT_FOUND, - detail={ - "error": "session_not_found", - "message": f"Session {session_id} not found", - }, - ) - _require_wallet_access(auth, session.wallet_id) + await _load_owned_dry_run_session(shadow_ledger, session_id, auth) result = await shadow_ledger.commit_session(session_id, money) + if not result.wallet_id: + # The session was there for the ownership check above and gone by the + # time we claimed it. ``end_dry_run_session`` already 404s on the same + # window; without this, commit answers 200 with ``success: false`` and + # a caller reading only the status code records a commit that never + # happened. + # + # Keyed on the empty ``wallet_id`` rather than on ``success``, because + # ``commit_session`` also reports ``success: false`` when the charges + # themselves fail (insufficient funds, say). That is a real answer + # about a real session and must stay a 200 โ€” 404-ing it would claim + # the session never existed. + raise _session_not_found(session_id) return { "session_id": result.session_id, @@ -1459,17 +1538,13 @@ async def revert_dry_run_session( The session is ended after reverting. """ shadow_ledger = get_shadow_ledger() - session = await shadow_ledger.get_session(session_id) - if not session: - raise HTTPException( - status_code=status.HTTP_404_NOT_FOUND, - detail={ - "error": "session_not_found", - "message": f"Session {session_id} not found", - }, - ) - _require_wallet_access(auth, session.wallet_id) + await _load_owned_dry_run_session(shadow_ledger, session_id, auth) result = await shadow_ledger.revert_session(session_id) + if not result.wallet_id: + # Same window as commit: present for the ownership check, gone before + # the revert. A missing session is ``revert_session``'s only failure + # mode, so an unset wallet id means exactly that. + raise _session_not_found(session_id) return { "session_id": result.session_id, @@ -1502,27 +1577,28 @@ async def simulate_charge( if session_id: shadow_ledger = get_shadow_ledger() - session = await shadow_ledger.get_session(session_id) - if not session: - raise HTTPException( - status_code=status.HTTP_404_NOT_FOUND, - detail={ - "error": "session_not_found", - "message": f"Session {session_id} not found", - }, + session = await _load_owned_dry_run_session(shadow_ledger, session_id, auth) + try: + result = await shadow_ledger.simulate_charge( + session_id=session_id, + service_category=request.service, + units=request.units, + description=request.description or "", + blocked_reason=( + "wallet_expired" + if await money.wallet_is_expired(session.wallet_id) + else None + ), ) - _require_wallet_access(auth, session.wallet_id) - result = await shadow_ledger.simulate_charge( - session_id=session_id, - service_category=request.service, - units=request.units, - description=request.description or "", - blocked_reason=( - "wallet_expired" - if await money.wallet_is_expired(session.wallet_id) - else None - ), - ) + except ValueError as exc: + # The session was claimed by a terminal operation between the + # ownership check and the simulation's write-back. Answered like + # every other session-gone path here rather than escaping as a + # 500 -- which is what the ``KeyError`` from the memory store used + # to do. + if "Session not found" not in str(exc): + raise + raise _session_not_found(session_id) from None await _record_billing_governance( event="billing.dry_run", auth=auth, diff --git a/app/routers/well_known.py b/app/routers/well_known.py index c16cf178..a2f164fc 100644 --- a/app/routers/well_known.py +++ b/app/routers/well_known.py @@ -311,12 +311,22 @@ class AgentPluginManifest(BaseModel): # advertised as a package-registry install. "python_sdk": { "status": "release_artifact_only", - "version": "0.4.0", + # The version an editable install of ``b2a_sdk/`` yields -- + # this field describes what ``install`` below produces, so it + # tracks the source, not the newest tag. The two are named + # separately because they differ whenever source has moved + # past the last release, and collapsing them into one number + # is how a discovery surface starts lying about what a client + # will actually get. tests/test_wedge_honesty.py pins this + # against b2a_sdk/pyproject.toml so they cannot drift. + "version": "0.5.0", "path": "b2a_sdk/", "install": "pip install -e ./b2a_sdk", + "latest_release_tag": "python-sdk-v0.4.0", "note": ( "Wheel and sdist are attached to python-sdk-v0.4.0 by the " - "release workflow. Not published to PyPI." + "release workflow; the source in b2a_sdk/ is 0.5.0 and is " + "ahead of that tag. Not published to PyPI." ), }, "typescript_sdk": { diff --git a/app/services/billing_engine.py b/app/services/billing_engine.py index 474e02ba..e7a6487f 100644 --- a/app/services/billing_engine.py +++ b/app/services/billing_engine.py @@ -249,6 +249,8 @@ async def _apply_charge_to_locked_wallet( request_path: str | None, description: str, operation_key: str | None, + velocity_hourly_reset_at: datetime | None = None, + velocity_daily_reset_at: datetime | None = None, ) -> LedgerEntry | InsufficientFundsResponse: result = await session.execute( select(WalletModel) @@ -273,19 +275,48 @@ async def reverse_velocity_record() -> None: # the spend cap and the anomaly auto-freeze are measured against. # A lost reversal here leaves the wallet permanently over-counted, # which throttles a caller that never spent the money. - await session.execute( - sa_update(WalletModel) - .where(cast(ColumnElement[bool], WalletModel.wallet_id == wallet_id)) - .values( - hourly_spent=clamped_decrement( - WalletModel.hourly_spent, charge_amount - ), - daily_spent=clamped_decrement( - WalletModel.daily_spent, charge_amount - ), - ) - .execution_options(synchronize_session=False) + # + # Each counter is reversed under its own period marker, because + # the two roll over on independent schedules. If the hourly period + # rolled between the increment and this reversal, the counter this + # charge added to has already been zeroed and the current one + # belongs to a period the charge never touched -- decrementing it + # would under-count live spend, which is the failure direction that + # actually matters here: an over-count throttles a caller who did + # not spend, an under-count silently raises the cap and delays the + # auto-freeze. A predicate covering both markers at once would be + # wrong for the same reason: a rolled daily period would suppress + # an hourly reversal that is still owed. + reversals = ( + ( + "hourly_spent", + WalletModel.hourly_reset_at, + velocity_hourly_reset_at, + WalletModel.hourly_spent, + ), + ( + "daily_spent", + WalletModel.daily_reset_at, + velocity_daily_reset_at, + WalletModel.daily_spent, + ), ) + for column_name, marker_column, recorded_marker, counter in reversals: + predicates: list[Any] = [ + cast(ColumnElement[bool], WalletModel.wallet_id == wallet_id) + ] + if recorded_marker is not None: + predicates.append( + cast(ColumnElement[bool], marker_column == recorded_marker) + ) + await session.execute( + sa_update(WalletModel) + .where(*predicates) + .values( + **{column_name: clamped_decrement(counter, charge_amount)} + ) + .execution_options(synchronize_session=False) + ) await session.refresh(wallet) if wallet.status in _NON_SPENDABLE_WALLET_STATUSES: @@ -546,6 +577,28 @@ async def charge( If dry_run=True, simulates the charge without affecting real balance or triggering velocity monitoring. Uses the shadow ledger for tracking. """ + # Non-finite units are refused here, not only at the HTTP boundary. + # This is the choke point every caller passes through -- the governed + # MCP path and the SDK reach it without going near the billing router + # -- and ``Decimal("Infinity")`` survives every arithmetic step below + # until ``charge_amount - compute_cost`` raises InvalidOperation. A + # non-finite amount that instead reached a write would sit in the + # ledger permanently: no reconciliation can subtract infinity back + # out, and every balance derived from it is NaN thereafter. + if not units.is_finite(): + raise ValueError(f"units must be a finite number, got {units}") + if units <= 0: + # Same reasoning, sharper consequence. A negative ``units`` makes + # ``charge_amount`` negative, and the guarded debit below then + # reads ``balance >= charge_amount`` as trivially true and applies + # ``balance - charge_amount`` -- which *raises* the balance. The + # result is credits minted from nothing, recorded as a DEBIT + # ledger entry with a positive amount, so the audit trail agrees + # it was a charge. Zero writes a meaningless zero-value debit. + # The router refuses both through ``gt=0``; the governed MCP path + # and the SDK never pass through the router. + raise ValueError(f"units must be greater than zero, got {units}") + pricing = self._default_pricing.get(service_category) if not pricing: raise ValueError(f"No pricing for {service_category.value}") @@ -619,7 +672,7 @@ async def charge( ) return ledger_entry_model_to_schema(existing_entry) - await velocity_monitor.check_and_record_charge( + velocity = await velocity_monitor.check_and_record_charge( wallet_id, charge_amount ) @@ -636,6 +689,14 @@ async def charge( request_path=request_path, description=description, operation_key=operation_key, + # Which period the increment landed in, so a reversal + # cannot decrement a later one. + velocity_hourly_reset_at=getattr( + velocity, "hourly_reset_at", None + ), + velocity_daily_reset_at=getattr( + velocity, "daily_reset_at", None + ), ) except IntegrityError: if operation_key is None: diff --git a/app/services/shadow_ledger.py b/app/services/shadow_ledger.py index ee29b0dd..70359d8d 100644 --- a/app/services/shadow_ledger.py +++ b/app/services/shadow_ledger.py @@ -439,13 +439,29 @@ async def simulate_charge( ], "created_at": session.created_at.isoformat(), } - await redis_client.setex( + # ``xx=True``: write only if the key still exists. A terminal + # operation (commit/end/revert) can claim the session between + # the read above and this write, and an unconditional ``setex`` + # would *recreate* it -- with every simulated charge intact, + # so it could be committed a second time and the same charges + # applied to the real wallet twice. + wrote = await redis_client.set( await self._session_key(session_id), - SESSION_TTL_SECONDS, json.dumps(session_data), + ex=SESSION_TTL_SECONDS, + xx=True, ) + if not wrote: + raise ValueError(f"Session not found: {session_id}") else: - self._memory_store[session_id]["session"] = session + # Same window, different symptom: the claim popped the entry, + # so ``self._memory_store[session_id]`` raised ``KeyError`` and + # escaped the router as a 500. Refuse the same way the Redis + # path does rather than resurrecting or crashing. + entry = self._memory_store.get(session_id) + if entry is None: + raise ValueError(f"Session not found: {session_id}") + entry["session"] = session logger.debug( f"Simulated charge {charge_id}: {charge_amount} credits " @@ -460,23 +476,21 @@ async def end_session(self, session_id: str) -> SessionSummary | None: Does NOT affect real wallet state - just provides a summary of what would have been charged. + + Claims the session rather than reading it and deleting it separately. + Read-then-delete leaves a window between the two: a concurrent + ``commit_session`` claims the session and charges the real wallet, this + method's delete becomes a no-op, and it still returns a summary built + from the stale read -- so the caller is told the session was ended + when a commit ended it, and ``revert_session`` reports a revert that + did not happen. ``_claim_session`` is the same atomic primitive commit + uses (GETDEL on Redis, ``dict.pop`` in memory), so exactly one of + end/commit/revert can win. """ - session = await self.get_session(session_id) + session = await self._claim_session(session_id) if not session: return None - redis_client = await self._get_redis() - - if redis_client: - await redis_client.delete(await self._session_key(session_id)) - await redis_client.delete(await self._balance_key(session_id)) - await redis_client.srem( - await self._wallet_sessions_key(session.wallet_id), - session_id, - ) - else: - self._memory_store.pop(session_id, None) - summary = SessionSummary( session_id=session_id, wallet_id=session.wallet_id, @@ -672,7 +686,24 @@ async def revert_session(self, session_id: str) -> RevertResult: charge_count = len(session.simulated_charges) total = session.total_simulated - await self.end_session(session_id) + if await self.end_session(session_id) is None: + # The session was claimed between the read above and this call -- + # by ``commit_session``, which claims atomically. Discarding this + # return value meant reporting ``reverted=True`` with the wallet id + # from the stale read, and a message reading "No changes made to + # real wallet" when the concurrent commit had already applied the + # charges to it. Reproduced: a wallet at 100 ended at 96 while the + # revert claimed nothing had moved. + # + # Reported as a missing session, matching the branch above, so the + # router's empty-``wallet_id`` check answers 404 rather than a 200 + # asserting a revert that did not happen. + return RevertResult( + session_id=session_id, + wallet_id="", + reverted=False, + message="Session not found", + ) logger.info( f"Reverted sandbox session {session_id}: " diff --git a/app/services/velocity_monitor.py b/app/services/velocity_monitor.py index 82417785..88fa4f14 100644 --- a/app/services/velocity_monitor.py +++ b/app/services/velocity_monitor.py @@ -39,6 +39,8 @@ def __init__( current_spend: float | None = None, limit: float | None = None, should_freeze: bool = False, + hourly_reset_at: datetime | None = None, + daily_reset_at: datetime | None = None, ): self.allowed = allowed self.reason = reason @@ -47,6 +49,14 @@ def __init__( self.current_spend = current_spend self.limit = limit self.should_freeze = should_freeze + # The period this charge's increment was actually recorded into, or + # None when nothing was recorded. A caller reversing the increment + # needs it: the counters roll over on their own schedule, and a + # reversal that lands after a rollover would decrement a period this + # charge never contributed to. See BillingEngine's + # ``reverse_velocity_record``. + self.hourly_reset_at = hourly_reset_at + self.daily_reset_at = daily_reset_at class WalletFrozenError(Exception): @@ -142,6 +152,11 @@ async def check_and_record_charge( # not the ones read before it. await session.refresh(wallet) + # Stamp the period the increment landed in, so a caller + # reversing it can refuse to decrement a later one. + recorded_hourly_reset_at = wallet.hourly_reset_at + recorded_daily_reset_at = wallet.daily_reset_at + velocity_result = self._check_limits( wallet=wallet, hourly_limit=hourly_limit, @@ -157,6 +172,8 @@ async def check_and_record_charge( reason="Wallet frozen due to anomalous spend velocity", alert_triggered=True, should_freeze=True, + hourly_reset_at=recorded_hourly_reset_at, + daily_reset_at=recorded_daily_reset_at, ) logger.warning( @@ -177,6 +194,8 @@ async def check_and_record_charge( await session.commit() + velocity_result.hourly_reset_at = recorded_hourly_reset_at + velocity_result.daily_reset_at = recorded_daily_reset_at return velocity_result def _reset_if_needed(self, wallet: WalletModel, now: datetime) -> None: diff --git a/b2a_sdk/CHANGELOG.md b/b2a_sdk/CHANGELOG.md index 4f2c2eae..2940ae46 100644 --- a/b2a_sdk/CHANGELOG.md +++ b/b2a_sdk/CHANGELOG.md @@ -1,6 +1,13 @@ # Changelog -## Unreleased +## Unreleased โ€” 0.5.0 + +`pyproject.toml` carries `0.5.0` from here on. `0.4.0` is published and +tagged `python-sdk-v0.4.0`, and the entries below are new surface area on +top of it, so leaving the source at `0.4.0` would have shipped something +materially different under a version already in the wild. A minor bump: +everything here is additive and no published behaviour changes. The tag is +not cut by this change โ€” `python-sdk-v0.5.0` is still a release decision. - Add `b2a_sdk.receipt_verifier` for offline verification of portable trust receipts. It imports nothing from the middleware application and needs no diff --git a/b2a_sdk/README.md b/b2a_sdk/README.md index 7997e83b..208739a7 100644 --- a/b2a_sdk/README.md +++ b/b2a_sdk/README.md @@ -4,18 +4,21 @@ Typed async client for the governed MCP trust loop: `discover โ†’ authenticate โ†’ authorize โ†’ invoke โ†’ meter โ†’ receipt โ†’ audit โ†’ govern` -CI builds version `0.4.0` as wheel and source artifacts. Pushing the -`python-sdk-v0.4.0` tag attaches them to a GitHub release. The package is not +CI builds version `0.5.0` as wheel and source artifacts. Pushing the +`python-sdk-v0.5.0` tag attaches them to a GitHub release. The package is not published to PyPI. ## Installation -Install a downloaded release wheel: +`0.5.0` is the source version here; the newest cut release is +`python-sdk-v0.4.0`. Install the released wheel: ```bash python -m pip install ./b2a_sdk-0.4.0-py3-none-any.whl ``` +Once `python-sdk-v0.5.0` is tagged, the same command with `0.5.0` installs it. + For repository development: ```bash diff --git a/b2a_sdk/pyproject.toml b/b2a_sdk/pyproject.toml index 76ab7a90..153ac3d8 100644 --- a/b2a_sdk/pyproject.toml +++ b/b2a_sdk/pyproject.toml @@ -4,7 +4,7 @@ build-backend = "hatchling.build" [project] name = "b2a-sdk" -version = "0.4.0" +version = "0.5.0" description = "Python SDK for governed MCP permits, invocations, and receipts" readme = "README.md" license = "MIT" diff --git a/b2a_sdk/src/b2a_sdk/__init__.py b/b2a_sdk/src/b2a_sdk/__init__.py index d4c34f81..14053fbb 100644 --- a/b2a_sdk/src/b2a_sdk/__init__.py +++ b/b2a_sdk/src/b2a_sdk/__init__.py @@ -38,7 +38,7 @@ verify_bundle, ) -__version__ = "0.4.0" +__version__ = "0.5.0" # The HTTP client surface (client, edge_client, and the decorators that wrap it) # pulls in httpx. Load those names lazily via PEP 562 so importing the package โ€” diff --git a/b2a_sdk/src/b2a_sdk/client.py b/b2a_sdk/src/b2a_sdk/client.py index 32ecb36a..b693c83d 100644 --- a/b2a_sdk/src/b2a_sdk/client.py +++ b/b2a_sdk/src/b2a_sdk/client.py @@ -149,7 +149,7 @@ def __init__( headers={ "X-API-Key": api_key, "Content-Type": "application/json", - "User-Agent": "b2a-sdk/0.4.0", + "User-Agent": "b2a-sdk/0.5.0", }, timeout=timeout, transport=transport, diff --git a/b2a_sdk/tests/test_trust_client.py b/b2a_sdk/tests/test_trust_client.py index 0ba4b426..3e51176c 100644 --- a/b2a_sdk/tests/test_trust_client.py +++ b/b2a_sdk/tests/test_trust_client.py @@ -83,7 +83,7 @@ async def test_default_base_url_is_the_live_canonical_api() -> None: async def test_discover_tools_returns_typed_definitions() -> None: async def handler(request: httpx.Request) -> httpx.Response: assert request.url.path == "/mcp/tools.json" - assert request.headers["user-agent"] == "b2a-sdk/0.4.0" + assert request.headers["user-agent"] == "b2a-sdk/0.5.0" return httpx.Response( 200, json={ diff --git a/docs/openapi.json b/docs/openapi.json index 87c6e354..21b127bf 100644 --- a/docs/openapi.json +++ b/docs/openapi.json @@ -12446,6 +12446,7 @@ }, "units": { "type": "number", + "exclusiveMinimum": 0.0, "title": "Units", "description": "Number of units", "default": 1.0 diff --git a/static/llm.txt b/static/llm.txt index 0014b430..18e513a9 100644 --- a/static/llm.txt +++ b/static/llm.txt @@ -137,16 +137,17 @@ Wallets are auto-frozen if spending exceeds: Prefer the HTTP trust loop above (permits โ†’ MCP invoke โ†’ receipts). Client libraries are optional and not required for the wedge. -### Python 0.4.0 (release artifact only) +### Python 0.5.0 source, 0.4.0 released (release artifact only) -CI builds a wheel and sdist and attaches them to the -`python-sdk-v0.4.0` GitHub release. It is **not** published to PyPI, and the +CI builds a wheel and sdist from the source in `b2a_sdk/`, currently 0.5.0, +and attaches them to the matching GitHub release. The newest tag today is +`python-sdk-v0.4.0`; a clone install gives you 0.5.0. It is **not** published to PyPI, and the repository (including its releases) is private โ€” request source access from the operator. With access, install a downloaded wheel, or install from a clone for development: ```bash -pip install ./b2a_sdk-0.4.0-py3-none-any.whl +pip install ./b2a_sdk-0.4.0-py3-none-any.whl # the released wheel pip install -e ./b2a_sdk ``` diff --git a/tests/conftest.py b/tests/conftest.py index 5d988c03..91ae2058 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -125,6 +125,86 @@ def iter_routes(routes): os.environ.setdefault("RATE_LIMIT_PER_MINUTE", "1000000") +def interleaving_factory(real_factory, hook, state, *, fire_on: int = 1): + """Wrap a session factory so `hook` runs once, mid-transaction. + + Fires after ``fire_on`` statements have been issued on the wrapped session, + which places it after the value under test has been read and before it is + written back. + + ``fire_on`` counts statements that reach ``Session.execute``. ``session.get`` + does **not**: it is delegated straight to the inner session, so a row loaded + that way is invisible here. Choosing a seam before such a load lets the + service re-read the row *after* the concurrent write lands, and the race + never happens -- a test that does that passes against the unfixed service + while ``state["fired"]`` reads true throughout. Always confirm a new seam by + reverting the service and watching the test fail. + """ + + class _Session: + def __init__(self, inner): + self._inner = inner + + def __getattr__(self, name): + return getattr(self._inner, name) + + async def execute(self, *args, **kwargs): + state["executes"] = state.get("executes", 0) + 1 + n = state["executes"] + # Fire *after* the statement returns, so the hook lands once the + # value has been read but before anything is written back. Firing + # beforehand would place the concurrent writer ahead of the read, + # which is not the race. Counting completed reads also keeps the + # seam identical across the fixed and unfixed code, which differ in + # how many statements they issue. + result = await self._inner.execute(*args, **kwargs) + if n == fire_on and not state.get("fired"): + state["fired"] = True + await hook() + return result + + class _CM: + def __init__(self, cm): + self._cm = cm + + async def __aenter__(self): + return _Session(await self._cm.__aenter__()) + + async def __aexit__(self, *exc): + return await self._cm.__aexit__(*exc) + + return lambda: (lambda: _CM(real_factory())) + + +def running_on_sqlite() -> bool: + """True when this run's ``DATABASE_URL`` names SQLite. + + ``DATABASE_URL`` above is a ``setdefault``, so exporting one before pytest + points the whole suite at another engine. Tests that force a concurrent + writer into the middle of an open transaction need to know: on SQLite the + surrounding ``SELECT ... FOR UPDATE`` is a silent no-op and the second + writer proceeds, which is exactly the condition those tests exist to + exercise. On PostgreSQL the lock is real, so the second writer waits for a + transaction that cannot commit until the second writer returns, and the + test hangs rather than fails. + """ + from app.core.db_urls import is_sqlite_url + + return is_sqlite_url(os.environ.get("DATABASE_URL", "")) + + +#: Guard for tests whose interleave depends on the row lock being a no-op. +#: They are skipped, not adapted: their subject is what happens when nothing +#: serializes the writers, which is not a state PostgreSQL can be put into. +requires_sqlite_row_lock_noop = pytest.mark.skipif( + not running_on_sqlite(), + reason=( + "forces a concurrent writer inside an open transaction; needs the " + "SQLite no-op row lock, and would deadlock against a real one" + ), +) + + @pytest.fixture(scope="session") def anyio_backend(): return "asyncio" diff --git a/tests/test_billing.py b/tests/test_billing.py index 84dc6ee5..1782feca 100644 --- a/tests/test_billing.py +++ b/tests/test_billing.py @@ -4,6 +4,7 @@ micro-metering โ†’ 402 insufficient funds โ†’ top-ups โ†’ arbitrage reporting. """ +import asyncio import pytest from datetime import timedelta from decimal import Decimal @@ -1162,3 +1163,1003 @@ async def test_rejected_charge_does_not_inflate_velocity_counters( # All three rejected charges were reversed out of the counters. assert wallet.daily_spent == _D("0") assert wallet.hourly_spent == _D("0") + + +@pytest.mark.anyio +@pytest.mark.parametrize( + ("units", "expected_status"), + [ + ("0", 422), + ("-5", 422), + ("abc", 422), + ("", 422), + ("NaN", 422), + ("Infinity", 422), + ], +) +async def test_charge_rejects_malformed_units_without_touching_the_wallet( + client, api_headers, clean_database, units: str, expected_status: int +): + """A rejected charge must leave the balance and the ledger exactly as found. + + ``units`` is declared ``gt=0``, so zero and negatives are refused โ€” a + negative would otherwise invert the debit into a credit, which is a way to + mint money through the metering endpoint. ``NaN`` and ``Infinity`` are + valid JSON-adjacent float spellings that Python's ``float()`` accepts, and + either would poison the balance arithmetic irrecoverably, so they are + pinned alongside the obviously malformed cases. + """ + sponsor_resp = await client.post( + "/v1/billing/wallets/sponsor", + json={ + "sponsor_name": "Malformed Units", + "email": "malformed-units@t.com", + "initial_credits": 10000, + }, + headers=api_headers, + ) + sponsor_id = sponsor_resp.json()["wallet_id"] + agent_resp = await client.post( + "/v1/billing/wallets/agent", + json={ + "sponsor_wallet_id": sponsor_id, + "agent_id": "malformed-units-bot", + "budget_credits": 5000, + }, + headers=api_headers, + ) + agent_wallet_id = agent_resp.json()["wallet_id"] + + before = await client.get( + f"/v1/billing/wallets/{agent_wallet_id}", headers=api_headers + ) + opening = before.json()["balance_exact"] + + resp = await client.post( + f"/v1/billing/charge?wallet_id={agent_wallet_id}" + f"&service=iot_bridge&units={units}", + headers=api_headers, + ) + assert resp.status_code == expected_status, resp.text + + after = await client.get( + f"/v1/billing/wallets/{agent_wallet_id}", headers=api_headers + ) + assert after.json()["balance_exact"] == opening + ledger_resp = await client.get( + f"/v1/billing/ledger/{agent_wallet_id}", headers=api_headers + ) + assert [e for e in ledger_resp.json()["entries"] if e["action"] == "debit"] == [] + + +@pytest.mark.anyio +@pytest.mark.parametrize("units", [Decimal("-5"), Decimal("-0.00000001"), Decimal("0")]) +async def test_billing_engine_refuses_non_positive_units(units): + """A negative units count inverts the debit into a credit. + + ``charge_amount`` goes negative, the guarded debit reads + ``balance >= charge_amount`` as trivially true, and ``balance - + charge_amount`` *raises* the balance. Reproduced before the fix: a wallet + at 100 charged ``units=-5`` ended at 110, with a ledger entry recording + ``action="debit", amount=+10`` -- money minted, and the audit trail + agreeing it was a charge. Zero is refused alongside it: it writes a + zero-value debit that means nothing. + + The router refuses both through ``gt=0``, but the governed MCP path and + the SDK reach the engine without passing through it, which is why the + guard belongs here. + """ + from app.schemas.billing import ServiceCategory + from app.services.agent_money import get_agent_money + + with pytest.raises(ValueError, match="greater than zero"): + await get_agent_money().charge( + wallet_id="wal-irrelevant", + service_category=ServiceCategory.IOT_BRIDGE, + units=units, + request_path="/non-positive", + ) + + +@pytest.mark.anyio +@pytest.mark.parametrize( + "units", [Decimal("Infinity"), Decimal("-Infinity"), Decimal("NaN")] +) +async def test_billing_engine_refuses_non_finite_units(units): + """The engine refuses non-finite units, not only the HTTP boundary. + + The governed MCP path and the SDK reach ``charge`` without passing the + billing router's query validation, so the guard has to live where every + caller meets. + """ + from app.schemas.billing import ServiceCategory + from app.services.agent_money import get_agent_money + + with pytest.raises(ValueError, match="finite"): + await get_agent_money().charge( + wallet_id="wal-irrelevant", + service_category=ServiceCategory.IOT_BRIDGE, + units=units, + request_path="/non-finite", + ) + + +@pytest.mark.anyio +async def test_charge_against_an_unknown_wallet_creates_nothing( + client, api_headers, clean_database +): + """An unknown wallet is refused, and leaves no ledger entry behind. + + The bootstrap key can operate on any wallet, so authorization does not + stop this one โ€” the wallet lookup has to. Asserting the ledger stays empty + matters as much as the status code: a debit written against an id with no + wallet row is an entry nothing will ever reconcile. + """ + unknown = "wal-does-not-exist-000000" + + resp = await client.post( + f"/v1/billing/charge?wallet_id={unknown}&service=iot_bridge&units=1", + headers=api_headers, + ) + assert resp.status_code == 404, resp.text + assert resp.json()["detail"]["error"] == "wallet_not_found" + + ledger_resp = await client.get(f"/v1/billing/ledger/{unknown}", headers=api_headers) + # Pin the status set first. Guarding the emptiness check on 200 alone + # would let a future change to the ledger endpoint drop the assertion + # this test exists for without anything failing. + assert ledger_resp.status_code in (200, 404), ledger_resp.text + if ledger_resp.status_code == 200: + assert ledger_resp.json()["entries"] == [] + + +@pytest.mark.anyio +@pytest.mark.parametrize("units", [-5, 0, "abc", "Infinity"]) +async def test_dry_run_charge_rejects_bad_units_on_both_branches( + client, api_headers, clean_database, units +): + """The dry-run endpoint validates before it picks a branch. + + ``simulate_charge`` has two: with a ``dry_run_session_id`` it calls + ``ShadowLedger.simulate_charge`` directly and never reaches + ``BillingEngine.charge``, so the engine's non-positive guard does not + cover it at all; without one it does reach the engine, where that guard + raises ``ValueError`` and escapes as a 500. Constraining the request field + refuses both before either branch is chosen โ€” and a simulation is exactly + where a caller checks affordability, so a 500 there reads as "the service + is broken" rather than "that is not a valid quantity". + """ + sponsor_resp = await client.post( + "/v1/billing/wallets/sponsor", + json={ + "sponsor_name": "Dry Run Units", + "email": "dry-run-units@t.com", + "initial_credits": 1000, + }, + headers=api_headers, + ) + wallet_id = sponsor_resp.json()["wallet_id"] + + resp = await client.post( + "/v1/billing/dry-run/charge", + json={"wallet_id": wallet_id, "service": "iot_bridge", "units": units}, + headers=api_headers, + ) + assert resp.status_code == 422, resp.text + + # A session-scoped simulation is refused on the same field, before the + # session is even looked up โ€” so an unknown session cannot mask it. + resp = await client.post( + "/v1/billing/dry-run/charge", + json={ + "wallet_id": wallet_id, + "service": "iot_bridge", + "units": units, + "dry_run_session_id": "sess-does-not-exist", + }, + headers=api_headers, + ) + assert resp.status_code == 422, resp.text + + +@pytest.mark.anyio +async def test_dry_run_charge_refuses_unauthenticated_callers_without_leaking( + client, api_headers, clean_database +): + """Unauthenticated dry runs are refused, and reveal nothing by refusing. + + A simulation endpoint is a tempting oracle: it takes a wallet id and a + session id and reports what *would* happen, so if its refusals varied with + whether those exist, an unauthenticated caller could enumerate them + without ever moving money. The refusal must therefore be identical + whatever the payload names. + + It must also come *before* body validation. If invalid units produced a + 422 while valid units produced a 401, the status code alone would confirm + that a request got far enough to be validated โ€” a weaker oracle, but the + same kind. + """ + sponsor_resp = await client.post( + "/v1/billing/wallets/sponsor", + json={ + "sponsor_name": "Dry Run Auth", + "email": "dry-run-auth@t.com", + "initial_credits": 1000, + }, + headers=api_headers, + ) + real_wallet = sponsor_resp.json()["wallet_id"] + fake_wallet = "wal-does-not-exist-000000" + + async def _dry_run(headers: dict, wallet_id: str, units: float = 1.0): + return await client.post( + "/v1/billing/dry-run/charge", + json={ + "wallet_id": wallet_id, + "service": "iot_bridge", + "units": units, + }, + headers=headers, + ) + + async def _dry_run_session(headers: dict, session_id: str, units: float): + return await client.post( + "/v1/billing/dry-run/charge", + json={ + "wallet_id": real_wallet, + "service": "iot_bridge", + "units": units, + "dry_run_session_id": session_id, + }, + headers=headers, + ) + + # The status varies with the *credential*, never with the payload: no + # header and a malformed key are 401, a well-formed but unrecognized key + # is 403. That split is fine โ€” it describes what the caller presented, not + # what exists on the server. + credentials = ( + ({}, 401), + ({"X-API-Key": "short"}, 401), + ({"X-API-Key": "not-a-real-key"}, 403), + ) + for headers, expected_status in credentials: + real = await _dry_run(headers, real_wallet) + fake = await _dry_run(headers, fake_wallet) + assert real.status_code == expected_status, real.text + assert fake.status_code == expected_status, fake.text + # Identical bodies, not merely identical statuses: an error message + # that named the wallet would leak exactly what the status hides. + assert real.text == fake.text + assert real_wallet not in real.text + + # Auth precedes validation, so a malformed body cannot be used to tell + # a request that was rejected at the door from one that got inside. + bad_units = await _dry_run(headers, real_wallet, units=-5) + assert bad_units.status_code == expected_status, bad_units.text + assert bad_units.text == real.text + + # The session-scoped branch is refused on the same terms. Asserting that + # against an unknown session id alone would prove nothing: it has to be + # compared with a session that really exists, or the test passes for a + # server that answers 404 for one and 401 for the other. + session_resp = await client.post( + "/v1/billing/dry-run/session", + json={"wallet_id": real_wallet}, + headers=api_headers, + ) + assert session_resp.status_code == 201, session_resp.text + real_session = session_resp.json()["session_id"] + fake_session = "sess-does-not-exist" + + for headers, expected_status in credentials: + for units in (1, -5): + real = await _dry_run_session(headers, real_session, units) + fake = await _dry_run_session(headers, fake_session, units) + assert real.status_code == expected_status, real.text + assert fake.status_code == expected_status, fake.text + # The handler looks the session up *before* checking wallet + # access โ€” but the credential check is a dependency, so it runs + # before the handler body at all. Nothing about which session + # ids exist reaches a caller who never got past the door. + assert real.text == fake.text + assert real_session not in real.text + + +@pytest.mark.anyio +async def test_charge_without_a_service_category_is_refused( + client, api_headers, clean_database +): + """Neither ``service`` nor ``service_category`` given: no pricing applies.""" + sponsor_resp = await client.post( + "/v1/billing/wallets/sponsor", + json={ + "sponsor_name": "No Category", + "email": "no-category@t.com", + "initial_credits": 1000, + }, + headers=api_headers, + ) + sponsor_id = sponsor_resp.json()["wallet_id"] + agent_resp = await client.post( + "/v1/billing/wallets/agent", + json={ + "sponsor_wallet_id": sponsor_id, + "agent_id": "no-category-bot", + "budget_credits": 500, + }, + headers=api_headers, + ) + agent_wallet_id = agent_resp.json()["wallet_id"] + + resp = await client.post( + f"/v1/billing/charge?wallet_id={agent_wallet_id}&units=1", + headers=api_headers, + ) + assert resp.status_code == 400, resp.text + assert resp.json()["detail"]["error"] == "missing_service" + + ledger_resp = await client.get( + f"/v1/billing/ledger/{agent_wallet_id}", headers=api_headers + ) + assert [e for e in ledger_resp.json()["entries"] if e["action"] == "debit"] == [] + + +@pytest.mark.anyio +async def test_dry_run_session_does_not_leak_another_tenants_session( + client, api_headers, clean_database +): + """A session belonging to another wallet answers as if it did not exist. + + The dry-run session endpoints look the session up first and check wallet + access second. Reporting those two failures differently made the surface an + oracle: a caller holding any valid wallet-scoped key got 404 for an + invented session id and 403 for a real one, so the two were + distinguishable with no access at all. The 403 body also carried + ``session.wallet_id`` โ€” the *owning* wallet โ€” so probing another tenant's + session id disclosed that tenant's wallet id outright. + + Session ids are UUID4 and cannot be enumerated, so the exposure needs an id + learned some other way (a log, a trace, a shared URL). That is a narrow + door, not a closed one, and the repository requires that tenant A cannot + reach tenant B's data. + """ + tenant_a = await client.post( + "/v1/billing/wallets/sponsor", + json={ + "sponsor_name": "Tenant A", + "email": "tenant-a-sessions@t.com", + "initial_credits": 100, + }, + headers=api_headers, + ) + wallet_a = tenant_a.json()["wallet_id"] + tenant_b = await client.post( + "/v1/billing/wallets/sponsor", + json={ + "sponsor_name": "Tenant B", + "email": "tenant-b-sessions@t.com", + "initial_credits": 100, + }, + headers=api_headers, + ) + wallet_b = tenant_b.json()["wallet_id"] + + key_resp = await client.post( + "/v1/api-keys", + json={"wallet_id": wallet_a, "key_name": "tenant-a-key"}, + headers=api_headers, + ) + assert key_resp.status_code == 201, key_resp.text + a_headers = {"X-API-Key": key_resp.json()["api_key"]} + + session_resp = await client.post( + "/v1/billing/dry-run/session", + json={"wallet_id": wallet_b}, + headers=api_headers, + ) + assert session_resp.status_code == 201, session_resp.text + b_session = session_resp.json()["session_id"] + unknown_session = "sess-does-not-exist" + + # Every endpoint that takes a session id, on the same terms. `charge` is + # included because it reaches the same lookup through a body field rather + # than a path parameter. + probes = ( + ("GET", "/v1/billing/dry-run/session/{sid}", None), + ("DELETE", "/v1/billing/dry-run/session/{sid}", None), + ("POST", "/v1/billing/dry-run/session/{sid}/commit", {}), + ("POST", "/v1/billing/dry-run/session/{sid}/revert", {}), + ) + for method, template, body in probes: + responses = [] + for sid in (b_session, unknown_session): + kwargs = {"headers": a_headers} + if body is not None: + kwargs["json"] = body + responses.append( + await client.request(method, template.format(sid=sid), **kwargs) + ) + theirs, absent = responses + assert theirs.status_code == 404, f"{method} {template}: {theirs.text}" + assert absent.status_code == 404, f"{method} {template}: {absent.text}" + # Both, not just `theirs`. Asserting the error code on one of them + # would let a change that answered 404 with a *different* code for the + # unknown session pass, which is the distinction this test exists to + # forbid. Byte equality is unavailable because the message echoes the + # session id, so normalise that out and compare the rest. + assert theirs.json()["detail"]["error"] == "session_not_found" + assert absent.json()["detail"]["error"] == "session_not_found" + assert theirs.text.replace(b_session, "SID") == absent.text.replace( + unknown_session, "SID" + ), f"{method} {template}: bodies differ beyond the echoed session id" + # The owning wallet must not appear anywhere in the refusal. + assert wallet_b not in theirs.text, f"{method} {template} leaked wallet_b" + + charges = [] + for sid in (b_session, unknown_session): + charge = await client.post( + "/v1/billing/dry-run/charge", + json={ + "wallet_id": wallet_a, + "service": "iot_bridge", + "units": 1, + "dry_run_session_id": sid, + }, + headers=a_headers, + ) + assert charge.status_code == 404, charge.text + assert charge.json()["detail"]["error"] == "session_not_found" + assert wallet_b not in charge.text + charges.append(charge) + + # Compared against each other, not only checked one at a time. This is the + # endpoint most likely to grow response fields โ€” it is the only one of the + # five with a full response model โ€” so a per-response check would let a + # newly added field recreate the oracle without failing anything. + theirs_charge, absent_charge = charges + assert theirs_charge.text.replace(b_session, "SID") == absent_charge.text.replace( + unknown_session, "SID" + ), "dry-run/charge: bodies differ beyond the echoed session id" + + # The owner is unaffected: a wallet-scoped key still reaches its own + # session. Without this the fix could "pass" by refusing everyone. + own_key = await client.post( + "/v1/api-keys", + json={"wallet_id": wallet_b, "key_name": "tenant-b-key"}, + headers=api_headers, + ) + assert own_key.status_code == 201, own_key.text + b_headers = {"X-API-Key": own_key.json()["api_key"]} + own = await client.get( + f"/v1/billing/dry-run/session/{b_session}", headers=b_headers + ) + assert own.status_code == 200, own.text + assert own.json()["wallet_id"] == wallet_b + + +@pytest.mark.anyio +@pytest.mark.parametrize("operation", ["commit", "revert"]) +async def test_dry_run_session_lost_mid_operation_is_a_404( + client, api_headers, clean_database, monkeypatch, operation +): + """A session that vanishes between the ownership check and the operation. + + ``commit_session`` and ``revert_session`` do not raise when the session is + gone โ€” they return a result object with ``wallet_id`` unset. The router + used to hand that straight back as a **200**, so a commit that committed + nothing answered OK with ``success: false``. A caller reading the status + code would record a commit that never happened. ``end_dry_run_session`` + already 404s on the identical window. + + The window is between two awaits and could not be forced through HTTP โ€” + two concurrent commits are both caught by the ownership check, because the + winner removes the session before the loser's check runs. So the service + is driven into exactly the state the guard exists for, which tests the + branch rather than the race. + """ + from app.services import shadow_ledger as shadow_ledger_module + + sponsor = await client.post( + "/v1/billing/wallets/sponsor", + json={ + "sponsor_name": "Lost Session", + "email": f"lost-session-{operation}@t.com", + "initial_credits": 1000, + }, + headers=api_headers, + ) + wallet_id = sponsor.json()["wallet_id"] + session_resp = await client.post( + "/v1/billing/dry-run/session", + json={"wallet_id": wallet_id}, + headers=api_headers, + ) + assert session_resp.status_code == 201, session_resp.text + session_id = session_resp.json()["session_id"] + + ledger = shadow_ledger_module.get_shadow_ledger() + method = f"{operation}_session" + original = getattr(ledger, method) + + async def vanished(*args, **kwargs): + """Return what the service really returns once the session is gone.""" + await ledger.end_session(session_id) + return await original(*args, **kwargs) + + monkeypatch.setattr(ledger, method, vanished) + + resp = await client.post( + f"/v1/billing/dry-run/session/{session_id}/{operation}", + json={}, + headers=api_headers, + ) + assert resp.status_code == 404, resp.text + assert resp.json()["detail"]["error"] == "session_not_found" + + +@pytest.mark.anyio +async def test_a_failed_commit_is_not_reported_as_a_missing_session( + client, api_headers, clean_database +): + """A commit that fails on funds is a real answer about a real session. + + ``commit_session`` reports ``success: false`` for two unrelated reasons: + the session is gone, and the charges themselves failed. The 404 guard keys + on the unset ``wallet_id`` precisely so this second case keeps its 200 โ€” + 404-ing it would tell the caller the session never existed, when it exists + and the commit is merely unaffordable. + + Reaching it needs a charge that fits when simulated and does not when + committed, so the wallet is drained *between* the two: 100 credits, a + 4-credit simulation, then a real 98-credit debit, leaving 2. + """ + sponsor = await client.post( + "/v1/billing/wallets/sponsor", + json={ + "sponsor_name": "Broke Commit", + "email": "broke-commit@t.com", + "initial_credits": 10000, + }, + headers=api_headers, + ) + agent = await client.post( + "/v1/billing/wallets/agent", + json={ + "sponsor_wallet_id": sponsor.json()["wallet_id"], + "agent_id": "broke-commit-bot", + "budget_credits": 100, + }, + headers=api_headers, + ) + wallet_id = agent.json()["wallet_id"] + + session_resp = await client.post( + "/v1/billing/dry-run/session", + json={"wallet_id": wallet_id}, + headers=api_headers, + ) + assert session_resp.status_code == 201, session_resp.text + session_id = session_resp.json()["session_id"] + + simulated = await client.post( + "/v1/billing/dry-run/charge", + json={ + "wallet_id": wallet_id, + "service": "iot_bridge", + "units": 2, + "dry_run_session_id": session_id, + }, + headers=api_headers, + ) + assert simulated.status_code == 200, simulated.text + assert simulated.json()["would_succeed"] is True, simulated.text + + drain = await client.post( + f"/v1/billing/charge?wallet_id={wallet_id}&service=iot_bridge&units=49", + headers=api_headers, + ) + assert drain.status_code == 200, drain.text + assert drain.json()["balance_after"] == 2.0, drain.text + + resp = await client.post( + f"/v1/billing/dry-run/session/{session_id}/commit", + json={}, + headers=api_headers, + ) + assert resp.status_code == 200, resp.text + body = resp.json() + assert body["success"] is False, body + assert body["committed_charges"] == 0, body + assert "insufficient_funds" in body["message"], body + # The session is named, not denied. This is the assertion the guard could + # break: keying the 404 on ``success`` instead of ``wallet_id`` fails here. + assert body["wallet_id"] == wallet_id + + +@pytest.mark.anyio +async def test_revert_does_not_claim_success_after_a_concurrent_commit( + client, api_headers, clean_database +): + """A revert that lost the session to a commit must not report success. + + ``revert_session`` reads the session, then calls ``end_session`` โ€” and used + to discard that call's return value. ``commit_session`` claims the session + atomically, so a commit landing between those two steps left ``end_session`` + with nothing to end while the revert still reported ``reverted=True``, + carrying the wallet id from its stale read. + + The status code is the smaller half. The message reads "No changes made to + real wallet", and the concurrent commit had just applied the charges to it. + Reproduced before the fix: a wallet at 100 ended at **96** while the revert + claimed nothing had moved. + + The interleave is forced *after* the initial read โ€” patching ``end_session`` + rather than removing the session up front, because a session already gone + is caught by the ownership check and never reaches this window. + """ + from app.services import shadow_ledger as shadow_ledger_module + + sponsor = await client.post( + "/v1/billing/wallets/sponsor", + json={ + "sponsor_name": "Revert Race", + "email": "revert-race@t.com", + "initial_credits": 10000, + }, + headers=api_headers, + ) + agent = await client.post( + "/v1/billing/wallets/agent", + json={ + "sponsor_wallet_id": sponsor.json()["wallet_id"], + "agent_id": "revert-race-bot", + "budget_credits": 100, + }, + headers=api_headers, + ) + wallet_id = agent.json()["wallet_id"] + + session_resp = await client.post( + "/v1/billing/dry-run/session", + json={"wallet_id": wallet_id}, + headers=api_headers, + ) + assert session_resp.status_code == 201, session_resp.text + session_id = session_resp.json()["session_id"] + simulated = await client.post( + "/v1/billing/dry-run/charge", + json={ + "wallet_id": wallet_id, + "service": "iot_bridge", + "units": 2, + "dry_run_session_id": session_id, + }, + headers=api_headers, + ) + assert simulated.status_code == 200, simulated.text + + ledger = shadow_ledger_module.get_shadow_ledger() + original_end_session = ledger.end_session + committed = {} + + async def commit_then_end(sid): + """Land the commit in the window, then run the real ``end_session``.""" + if sid == session_id and "result" not in committed: + committed["result"] = await ledger.commit_session(sid, get_agent_money()) + return await original_end_session(sid) + + ledger.end_session = commit_then_end + try: + resp = await client.post( + f"/v1/billing/dry-run/session/{session_id}/revert", + json={}, + headers=api_headers, + ) + finally: + ledger.end_session = original_end_session + + # The commit really did charge the wallet โ€” otherwise there is nothing for + # the revert to have lied about. + assert committed["result"].success is True, committed["result"] + assert committed["result"].committed_charges == 1, committed["result"] + balance = await client.get(f"/v1/billing/wallets/{wallet_id}", headers=api_headers) + assert balance.json()["balance"] == 96.0, balance.text + + assert resp.status_code == 404, resp.text + assert resp.json()["detail"]["error"] == "session_not_found" + # Nothing in the refusal may claim the revert happened. + assert "reverted" not in resp.text + + +@pytest.mark.anyio +async def test_a_simulated_charge_cannot_resurrect_a_claimed_session( + client, api_headers, clean_database +): + """A simulation must not write back a session a terminal operation claimed. + + ``simulate_charge`` reads the session, mutates it, and writes it back. A + commit claiming in between used to leave that write-back facing a session + that no longer exists, with two symptoms by engine: + + * **Redis** โ€” the unconditional ``setex`` *recreated* the key, with every + simulated charge intact. The committed session came back to life and + could be committed again, applying the same charges to the real wallet + twice. This is the production configuration. + * **memory** โ€” ``self._memory_store[session_id]`` raised ``KeyError``, + which escaped the router as a 500. + + Both now refuse: the write is conditional on the session still existing + (``xx=True`` on Redis, a ``.get()`` guard in memory) and the caller is told + the session is gone, which the router answers as 404 like every other + session-gone path. + + Only the memory symptom is exercised here. The Redis path has no fake + client in this suite and adding one is a dependency this test does not + justify, so that half is argued from the same code path rather than run. + """ + from app.services import shadow_ledger as shadow_ledger_module + + sponsor = await client.post( + "/v1/billing/wallets/sponsor", + json={ + "sponsor_name": "Resurrect", + "email": "resurrect-session@t.com", + "initial_credits": 10000, + }, + headers=api_headers, + ) + agent = await client.post( + "/v1/billing/wallets/agent", + json={ + "sponsor_wallet_id": sponsor.json()["wallet_id"], + "agent_id": "resurrect-bot", + "budget_credits": 100, + }, + headers=api_headers, + ) + wallet_id = agent.json()["wallet_id"] + session_resp = await client.post( + "/v1/billing/dry-run/session", + json={"wallet_id": wallet_id}, + headers=api_headers, + ) + assert session_resp.status_code == 201, session_resp.text + session_id = session_resp.json()["session_id"] + + ledger = shadow_ledger_module.get_shadow_ledger() + original_get_redis = ledger._get_redis + state: dict = {"calls": 0} + + async def commit_at_the_write_back(): + """Claim the session at the seam between the read and the write. + + Through the router the lookups are: (1) the ownership check's + ``get_session``, (2) ``simulate_charge``'s own ``get_session``, and + (3) the write-back. Only the third lands the commit *after* the read + and the mutation, which is the sole window where the write-back can + resurrect anything -- firing earlier just reproduces the ordinary + missing-session path, which already answered 404 before this fix. + """ + state["calls"] += 1 + if state["calls"] == 3 and "fired" not in state: + state["fired"] = True + state["commit"] = await ledger.commit_session(session_id, get_agent_money()) + return await original_get_redis() + + ledger._get_redis = commit_at_the_write_back + try: + resp = await client.post( + "/v1/billing/dry-run/charge", + json={ + "wallet_id": wallet_id, + "service": "iot_bridge", + "units": 2, + "dry_run_session_id": session_id, + }, + headers=api_headers, + ) + finally: + ledger._get_redis = original_get_redis + + assert "fired" in state, "the commit never landed; the window was missed" + assert resp.status_code == 404, resp.text + assert resp.json()["detail"]["error"] == "session_not_found" + + # The session stays dead. If the write-back had resurrected it, it would + # still be committable โ€” and its charges applied to the wallet twice. + assert await ledger.get_session(session_id) is None + recommit = await client.post( + f"/v1/billing/dry-run/session/{session_id}/commit", + json={}, + headers=api_headers, + ) + assert recommit.status_code == 404, recommit.text + + +class _FakeRedis: + """The eight Redis calls ``ShadowLedger`` makes, over plain dicts. + + Every method yields with ``asyncio.sleep(0)`` before touching state. That + is the whole point: against a real server each call is a network round + trip, so another task can run between any two of them. Without the yield + the fake would serialise everything and quietly hide the races these + tests exist to catch. + + Hand-rolled rather than pulled in as a dependency โ€” the surface is eight + methods and the repository does not otherwise need a Redis test double. + """ + + def __init__(self): + self.kv: dict = {} + self.sets: dict = {} + self.on_get = None + self.on_set = None + + async def ping(self): + await asyncio.sleep(0) + return True + + async def get(self, key): + await asyncio.sleep(0) + if self.on_get is not None: + hook, self.on_get = self.on_get, None + await hook() + return self.kv.get(key) + + async def getdel(self, key): + await asyncio.sleep(0) + return self.kv.pop(key, None) + + async def set(self, key, value, ex=None, xx=False): + await asyncio.sleep(0) + if self.on_set is not None: + hook, self.on_set = self.on_set, None + await hook() + if xx and key not in self.kv: + return None + self.kv[key] = value + return True + + async def setex(self, key, ttl, value): + await asyncio.sleep(0) + self.kv[key] = value + return True + + async def delete(self, key): + await asyncio.sleep(0) + return 1 if self.kv.pop(key, None) is not None else 0 + + async def sadd(self, key, member): + await asyncio.sleep(0) + self.sets.setdefault(key, set()).add(member) + + async def smembers(self, key): + await asyncio.sleep(0) + return set(self.sets.get(key, set())) + + async def srem(self, key, member): + await asyncio.sleep(0) + self.sets.get(key, set()).discard(member) + + +@pytest.fixture +def redis_backed_shadow_ledger(): + """Point the shared ShadowLedger at the fake, and restore it after.""" + from app.services import shadow_ledger as shadow_ledger_module + + ledger = shadow_ledger_module.get_shadow_ledger() + fake = _FakeRedis() + saved_client, saved_url = ledger._redis, ledger._redis_url + ledger._redis, ledger._redis_url = fake, "redis://fake" + try: + yield ledger, fake + finally: + ledger._redis, ledger._redis_url = saved_client, saved_url + + +@pytest.mark.anyio +async def test_two_concurrent_end_sessions_do_not_both_succeed( + client, api_headers, clean_database, redis_backed_shadow_ledger +): + """Only one terminal operation may claim a session โ€” proved on Redis. + + This is the property that closes the revert/commit window, and until now + it was argued rather than tested: against the in-memory store there is no + suspension point between a read and a delete, so the interleave could not + be forced. Against Redis every call is a round trip, and the fake models + that with an ``asyncio.sleep(0)`` in each method. + + Read-then-delete lets both callers read before either deletes, so both + return a summary โ€” and the same shape is what let a commit slip between + ``end_session``'s read and its delete while ``revert_session`` reported + success. ``_claim_session`` is a single ``GETDEL``, so exactly one wins. + """ + ledger, _fake = redis_backed_shadow_ledger + + sponsor = await client.post( + "/v1/billing/wallets/sponsor", + json={ + "sponsor_name": "Double End", + "email": "double-end@t.com", + "initial_credits": 1000, + }, + headers=api_headers, + ) + wallet_id = sponsor.json()["wallet_id"] + session = await ledger.create_session( + wallet_id=wallet_id, real_balance=Decimal("1000") + ) + + first, second = await asyncio.gather( + ledger.end_session(session.session_id), + ledger.end_session(session.session_id), + ) + + ended = [s for s in (first, second) if s is not None] + assert len(ended) == 1, ( + "both callers ended the same session; a terminal operation is not " + f"exclusive (got {first!r} and {second!r})" + ) + assert await ledger.get_session(session.session_id) is None + + +@pytest.mark.anyio +async def test_a_simulated_charge_cannot_recreate_a_claimed_redis_session( + client, api_headers, clean_database, redis_backed_shadow_ledger +): + """The Redis half of the resurrection fix, which was argued until now. + + ``simulate_charge`` writes the mutated session back. Unconditionally, that + write *recreates* a key a concurrent commit has already claimed โ€” with + every simulated charge intact โ€” so the committed session comes back to + life and can be committed a second time, applying the same charges to the + real wallet twice. ``xx=True`` makes the write conditional on the key + still existing, so a claimed session stays claimed. + + The commit is landed from inside the fake's ``set``, which is exactly the + moment the old code would have resurrected the session. + """ + ledger, fake = redis_backed_shadow_ledger + from app.schemas.billing import ServiceCategory + + sponsor = await client.post( + "/v1/billing/wallets/sponsor", + json={ + "sponsor_name": "Redis Resurrect", + "email": "redis-resurrect@t.com", + "initial_credits": 10000, + }, + headers=api_headers, + ) + agent = await client.post( + "/v1/billing/wallets/agent", + json={ + "sponsor_wallet_id": sponsor.json()["wallet_id"], + "agent_id": "redis-resurrect-bot", + "budget_credits": 100, + }, + headers=api_headers, + ) + wallet_id = agent.json()["wallet_id"] + session = await ledger.create_session( + wallet_id=wallet_id, real_balance=Decimal("100") + ) + session_id = session.session_id + committed = {} + + async def commit_before_the_write_back(): + committed["result"] = await ledger.commit_session(session_id, get_agent_money()) + + fake.on_set = commit_before_the_write_back + + with pytest.raises(ValueError, match="Session not found"): + await ledger.simulate_charge(session_id, ServiceCategory.IOT_BRIDGE, units=2) + + assert "result" in committed, "the commit never landed; the window was missed" + # The session must stay claimed. If the write-back recreated it, it would + # still be committable โ€” and its charges applied to the wallet twice. + assert await ledger.get_session(session_id) is None + second = await ledger.commit_session(session_id, get_agent_money()) + assert second.wallet_id == "", second + assert second.committed_charges == 0, second diff --git a/tests/test_permit_budget_integrity.py b/tests/test_permit_budget_integrity.py index b1ad931a..74165352 100644 --- a/tests/test_permit_budget_integrity.py +++ b/tests/test_permit_budget_integrity.py @@ -3,9 +3,15 @@ Every test here targets a path that wrote ``spent_credits`` from a value this process had read in an earlier statement. On PostgreSQL the surrounding ``SELECT ... FOR UPDATE`` hid the problem; on SQLite that lock is a silent -no-op, and nothing in this repository forbids SQLite in production -(``config.py`` lists it and ``trust_mode.py`` has no engine guardrail), so -these are live defects on a supported configuration rather than theory. +no-op, so these were live defects rather than theory. + +``validate_trust_mode_config`` now refuses a SQLite ``DATABASE_URL`` in +production-like environments, which closes the configuration these exploited. +That guard is a second line, not a replacement: it is one environment-variable +check away from being bypassed, it does not apply to the local and staging +databases people do real work against, and a path that writes money from a +stale read is wrong on its own terms. These tests therefore keep running on +SQLite, where the missing lock makes the defect observable at all. """ from __future__ import annotations @@ -24,6 +30,7 @@ from app.schemas.trust import PermitCreateRequest from app.services.permits import PermitError, get_permit_service from tests.test_trust_helpers import provision_agent_wallet +from tests.conftest import requires_sqlite_row_lock_noop @pytest.fixture @@ -134,6 +141,7 @@ async def test_reserve_budget_still_denies_a_genuine_cap_breach( assert await _spent(permit.permit_id) == Decimal("4") +@requires_sqlite_row_lock_noop @pytest.mark.anyio async def test_reconcile_budgets_does_not_erase_a_concurrent_reservation( client, clean_database, enforce_naive_utc_datetime_columns, monkeypatch diff --git a/tests/test_production_trust_posture.py b/tests/test_production_trust_posture.py index 075796fc..8f4264bf 100644 --- a/tests/test_production_trust_posture.py +++ b/tests/test_production_trust_posture.py @@ -47,6 +47,11 @@ "WEBAUTHN_ALLOW_MOCK": "false", "TRUST_SIGNING_PRIVATE_KEY_B64": _TEST_SIGNING_PRIVATE_KEY_B64, "PUBLIC_URL": "https://api.thisisatest.tech", + # A production posture is not complete without the relational database the + # wallets, permits, receipts, and ledger live in. It is PostgreSQL because + # validate_trust_mode_config refuses SQLite here: SQLAlchemy silently drops + # SELECT ... FOR UPDATE on SQLite, which the money and permit paths rely on. + "DATABASE_URL": "postgresql+asyncpg://user:pw@db.internal:5432/trust", } @@ -145,6 +150,7 @@ def test_production_trust_flags_are_strict(production_trust_flags): debug=cfg.DEBUG, webauthn_allow_mock=cfg.WEBAUTHN_ALLOW_MOCK, enable_proof_surfaces=cfg.ENABLE_PROOF_SURFACES, + database_url=cfg.DATABASE_URL, ) assert ( describe_permissive_trust_mode( diff --git a/tests/test_refund_reconciliation.py b/tests/test_refund_reconciliation.py index 5737cf54..8b163120 100644 --- a/tests/test_refund_reconciliation.py +++ b/tests/test_refund_reconciliation.py @@ -25,6 +25,7 @@ ) from app.services.receipts import ReceiptService from app.services.service_registry import get_service_registry +from tests.conftest import interleaving_factory, requires_sqlite_row_lock_noop from tests.test_trust_helpers import ( BOOTSTRAP_HEADERS, create_tool_permit, @@ -43,6 +44,7 @@ async def _create_unrefunded_failure( client: AsyncClient, *, requires_human_approval: bool = False, + max_credits: int = 2, ) -> dict[str, Any]: provisioned = await provision_agent_wallet(client) registry = get_service_registry() @@ -70,7 +72,7 @@ async def failing_refund(self, **_kwargs): # noqa: ANN001 wallet_id=provisioned["agent_wallet_id"], key_id=provisioned["key_id"], tool_name=tool_name, - max_credits=2, + max_credits=max_credits, idem_key="operator-reconcile-permit-create", requires_human_approval=requires_human_approval, ) @@ -755,3 +757,76 @@ async def test_tampered_signed_receipt_cannot_authorize_refund( headers=case["provisioned"]["agent_headers"], ) assert all(entry["action"] != "refund" for entry in ledger.json()["entries"]) + + +@requires_sqlite_row_lock_noop +@pytest.mark.anyio +async def test_operator_refund_release_cannot_erase_a_concurrent_reservation( + client, + clean_database, + monkeypatch, +): + """The operator release must subtract, not overwrite. + + ``permits.py`` guarantees that every ``spent_credits`` write is decided by + the database rather than by a value some process read earlier, and that + guarantee is only as strong as its weakest caller. This one lives outside + that module: it releases the reservation an operator refund settles, in + the middle of a long transaction that also verifies a receipt signature + and writes a ledger entry. A reservation landing in that window and then + being overwritten leaves the permit under-counting its own spend, which is + exactly the state that lets it be spent past its cap. + + The interleave is forced deterministically: the hook fires after the + correlated-refund lookup, which is past the point where the service has + the permit row in hand and before it writes the release. + """ + import app.services.refund_reconciliation as reconciliation_module + from app.db.models import PermitModel + + # The permit is minted with one credit of headroom above the tool's cost. + # The default fixture spends its permit exactly to the cap, which would + # refuse the concurrent reservation and leave the interleave proving + # nothing -- and the headroom cannot be added afterwards, since + # ``max_credits`` is covered by the permit signature the service verifies. + case = await _create_unrefunded_failure(client, max_credits=3) + receipt_id = case["receipt"]["receipt_id"] + permit_id = case["permit"]["permit_id"] + + real_factory = get_session_factory() + + async with real_factory() as session: + permit = await session.get(PermitModel, permit_id) + assert permit is not None + assert permit.spent_credits == Decimal("2") + + state: dict = {} + + async def _concurrent_reservation() -> None: + await get_permit_service().reserve_budget(permit_id, Decimal("1")) + + monkeypatch.setattr( + reconciliation_module, + "get_session_factory", + # Statement 3 is the correlated-refund lookup, the first statement + # visible to the wrapper *after* the permit row is loaded. The load + # itself is a ``session.get`` and bypasses the wrapper entirely, so + # firing any earlier lets the service re-read the permit after the + # reservation lands and the race never happens -- the first version of + # this test did exactly that and passed against the unfixed service. + interleaving_factory(real_factory, _concurrent_reservation, state, fire_on=3), + ) + + resolved = await client.post( + f"/v1/receipts/reconciliation/refunds/{receipt_id}/retry", + headers=BOOTSTRAP_HEADERS, + ) + assert resolved.status_code == 200, resolved.text + assert state.get("fired"), "the interleave never ran โ€” the test proved nothing" + + permit_after = await get_permit_service().get_permit(permit_id) + assert permit_after is not None + # 2 reserved, +1 from the concurrent reservation, -2 released by the + # refund. An absolute write of the recomputed 0 would erase that 1 and + # hand the permit a credit of headroom it never actually had returned. + assert permit_after.spent_credits == Decimal("1") diff --git a/tests/test_stripe_integration.py b/tests/test_stripe_integration.py index 5229463b..cec10bed 100644 --- a/tests/test_stripe_integration.py +++ b/tests/test_stripe_integration.py @@ -10,9 +10,11 @@ import pytest import stripe from httpx import AsyncClient, ASGITransport +from sqlalchemy import update as sa_update from sqlalchemy.exc import IntegrityError from app.core.time import utc_now +from tests.conftest import interleaving_factory, requires_sqlite_row_lock_noop from app.main import app from app.services.stripe_integration import StripeIntegration, StripeSettlementError @@ -486,6 +488,81 @@ async def test_refund_clawback_advances_the_wallet_timestamp( f"{wallet.updated_at}, the value set before it ran" ) + @requires_sqlite_row_lock_noop + @pytest.mark.anyio + async def test_a_concurrent_charge_cannot_vanish_under_a_refund_clawback( + self, client, sponsor_wallet, api_headers, monkeypatch + ): + """A spend landing mid-clawback must not be erased. + + The clawback read the sponsor wallet, then several statements later + wrote ``balance - refund_delta`` computed from that read. The + cumulative-refund arithmetic above it makes a *redelivered event* + idempotent, but it does nothing about a charge committing in the gap: + that debit is simply overwritten. The discrepancy is against fiat + Stripe has already returned, so the books are wrong in the direction + that costs the operator real money. + + The interleave fires after the wallet SELECT and before the clawback + write, and is forced rather than raced. + """ + from app.services.stripe_integration import get_stripe_integration + from app.db.database import get_session_factory + from app.db.models import WalletModel + + wallet_id = sponsor_wallet["wallet_id"] + integration = get_stripe_integration() + real_factory = get_session_factory() + + await integration._mint_credits( + wallet_id=wallet_id, + amount=Decimal("50000"), + payment_intent_id="pi_clawback_race", + description="topup", + ) + + spend = Decimal("100") + state: dict = {} + + async def _concurrent_spend() -> None: + """A charge commits against the same wallet, mid-clawback.""" + async with real_factory() as s: + async with s.begin(): + await s.execute( + sa_update(WalletModel) + .where(WalletModel.wallet_id == wallet_id) + .values(balance=WalletModel.balance - spend) + .execution_options(synchronize_session=False) + ) + + monkeypatch.setattr( + integration, + "_session_factory", + # Statement 2 is the wallet SELECT. Firing after it returns puts + # the spend between the read and the clawback write. + interleaving_factory(real_factory, _concurrent_spend, state, fire_on=2), + ) + + # Stripe reports cents; the ledger is in credits at ten per cent (the + # existing redelivery test pins the same ratio: 5000 cents claws back + # the whole 50000-credit top-up). + refunded_cents = 500 + refund_credits = Decimal(refunded_cents * 10) + await integration._handle_refund( + refunded_charge( + payment_intent_id="pi_clawback_race", + amount_refunded=refunded_cents, + ), + "evt_clawback_race", + ) + + assert state.get("fired"), "the interleave never ran โ€” the test proved nothing" + async with real_factory() as session: + wallet = await session.get(WalletModel, wallet_id) + assert wallet is not None + # Both the clawback and the concurrent spend are on the balance. + assert wallet.balance == Decimal("50000") - refund_credits - spend + @pytest.mark.anyio async def test_successive_partial_refunds_apply_only_new_cumulative_delta( self, sponsor_wallet diff --git a/tests/test_trust_mode_guardrails.py b/tests/test_trust_mode_guardrails.py index 09cd2a44..6cc2be9e 100644 --- a/tests/test_trust_mode_guardrails.py +++ b/tests/test_trust_mode_guardrails.py @@ -17,6 +17,7 @@ VALID_SIGNING_PRIVATE_KEY_B64 = base64.b64encode(bytes(range(32))).decode() +VALID_PRODUCTION_DATABASE_URL = "postgresql+asyncpg://user:pw@db.internal:5432/trust" @pytest.mark.parametrize( @@ -153,6 +154,7 @@ def test_settings_wrapper_uses_environment_field(): ENABLE_DEV_KEY_SELF_PROVISION=False, DEBUG=False, WEBAUTHN_ALLOW_MOCK=False, + DATABASE_URL=VALID_PRODUCTION_DATABASE_URL, ) validate_trust_mode_guardrails(settings) @@ -184,6 +186,7 @@ def test_public_mcp_accepts_shared_rate_limiter_in_production(): enable_public_mcp_endpoint=True, redis_url="redis://redis.internal:6379/0", public_url="https://api.example.com", + database_url=VALID_PRODUCTION_DATABASE_URL, ) @@ -203,6 +206,105 @@ def test_public_mcp_requires_public_origin_in_production(): assert "PUBLIC_URL" in str(exc_info.value) + +@pytest.mark.parametrize( + "database_url", + [ + "sqlite+aiosqlite:///./trust.db", + "sqlite:///./trust.db", + "SQLite+aiosqlite:///./trust.db", + "sqlite+aiosqlite:///:memory:", + ], +) +def test_production_refuses_a_sqlite_database_url(database_url: str): + """SQLite in production is refused at boot, not discovered under load. + + SQLAlchemy silently drops ``SELECT ... FOR UPDATE`` on SQLite, so every + guarded money and permit path loses the serialization it is written to + rely on. The in-memory spellings are refused too: they are no safer, and + they additionally lose the ledger on restart. + """ + with pytest.raises(TrustModeGuardrailError) as exc_info: + validate_trust_mode_config( + environment="production", + trust_mode_enabled=True, + signing_private_key_b64=VALID_SIGNING_PRIVATE_KEY_B64, + allow_legacy_unpermitted_mcp=False, + enable_proof_surfaces=False, + database_url=database_url, + ) + + message = str(exc_info.value) + assert "DATABASE_URL must not be SQLite" in message + assert "FOR UPDATE" in message + + +def test_production_refuses_a_missing_database_url(): + with pytest.raises(TrustModeGuardrailError) as exc_info: + validate_trust_mode_config( + environment="production", + trust_mode_enabled=True, + signing_private_key_b64=VALID_SIGNING_PRIVATE_KEY_B64, + allow_legacy_unpermitted_mcp=False, + enable_proof_surfaces=False, + database_url=" ", + ) + + assert "DATABASE_URL must be set" in str(exc_info.value) + + +def test_a_redis_state_backend_does_not_excuse_a_sqlite_database_url(): + """The residual hole this guard exists to close. + + ``app/core/durable_state.py`` already refuses SQLite, but it governs the + key/value *state store*. A deployment with ``STATE_BACKEND=redis`` and a + ``REDIS_URL`` satisfies that check completely while ``DATABASE_URL`` stays + SQLite -- and the wallet, permit, and ledger writes run against the ORM + engine that URL builds, not against the state store. Passing the state + check must not carry the ORM engine with it. + """ + with pytest.raises(TrustModeGuardrailError) as exc_info: + validate_trust_mode_config( + environment="production", + trust_mode_enabled=True, + signing_private_key_b64=VALID_SIGNING_PRIVATE_KEY_B64, + allow_legacy_unpermitted_mcp=False, + enable_proof_surfaces=False, + redis_url="redis://redis.internal:6379/0", + database_url="sqlite+aiosqlite:///./trust.db", + ) + + assert "DATABASE_URL must not be SQLite" in str(exc_info.value) + + +def test_production_accepts_a_postgres_database_url(): + validate_trust_mode_config( + environment="production", + trust_mode_enabled=True, + signing_private_key_b64=VALID_SIGNING_PRIVATE_KEY_B64, + allow_legacy_unpermitted_mcp=False, + enable_proof_surfaces=False, + database_url=VALID_PRODUCTION_DATABASE_URL, + ) + + +@pytest.mark.parametrize("environment", ["", "local", "development", "test", "ci"]) +def test_local_environments_still_accept_sqlite(environment: str): + """The guard must not make local development impossible. + + A SQLite ``DATABASE_URL`` is the standard local value and the default the + test suite itself runs on; refusing it outside production-like + environments would break every developer and this suite with them. + """ + validate_trust_mode_config( + environment=environment, + trust_mode_enabled=True, + signing_private_key_b64="", + allow_legacy_unpermitted_mcp=True, + database_url="sqlite+aiosqlite:///./test.db", + ) + + def test_describe_permissive_returns_none_when_strict(): assert ( describe_permissive_trust_mode( @@ -328,3 +430,36 @@ def test_shipped_defaults_are_strict(): "TRUST_MODE_ENABLED": True, "ALLOW_LEGACY_UNPERMITTED_MCP": False, } + + +@pytest.mark.anyio +async def test_lifespan_refuses_a_sqlite_database_url_in_production(monkeypatch): + """The guard has to stop a real boot, not only pass a unit test. + + ``validate_trust_mode_guardrails`` is called from ``lifespan``, so a + production deployment pointed at SQLite fails to start rather than serving + traffic with the row locks silently absent. Asserting the function in + isolation would not catch the guard being dropped from the startup path. + """ + import app.main as main_module + + monkeypatch.setattr(main_module.settings, "ENVIRONMENT", "production") + monkeypatch.setattr(main_module.settings, "TRUST_MODE_ENABLED", True) + monkeypatch.setattr( + main_module.settings, + "TRUST_SIGNING_PRIVATE_KEY_B64", + VALID_SIGNING_PRIVATE_KEY_B64, + ) + monkeypatch.setattr(main_module.settings, "ALLOW_LEGACY_UNPERMITTED_MCP", False) + monkeypatch.setattr(main_module.settings, "ENABLE_PROOF_SURFACES", False) + monkeypatch.setattr(main_module.settings, "DEBUG", False) + monkeypatch.setattr(main_module.settings, "WEBAUTHN_ALLOW_MOCK", False) + monkeypatch.setattr( + main_module.settings, "DATABASE_URL", "sqlite+aiosqlite:///./trust.db" + ) + + with pytest.raises(TrustModeGuardrailError) as exc_info: + async with main_module.lifespan(main_module.app): + pytest.fail("lifespan must not start against SQLite in production") + + assert "DATABASE_URL must not be SQLite" in str(exc_info.value) diff --git a/tests/test_wallet_ledger_integrity.py b/tests/test_wallet_ledger_integrity.py index 54f40661..64b5af6a 100644 --- a/tests/test_wallet_ledger_integrity.py +++ b/tests/test_wallet_ledger_integrity.py @@ -2,8 +2,13 @@ Each path here read a balance or counter in one statement and wrote it back in a later one, serialized only by ``SELECT ... FOR UPDATE``. That lock is a -silent no-op on SQLite, and nothing in this repository forbids SQLite in -production, so these are live on a supported configuration. +silent no-op on SQLite, so these were live defects rather than theory. + +``validate_trust_mode_config`` now refuses a SQLite ``DATABASE_URL`` in +production-like environments, closing the configuration that made them +exploitable. These tests still run on SQLite deliberately: the missing lock is +what makes the defect observable, and the boot guard does not reach the local +and staging databases people do real work against. The interleaves are forced deterministically through an instrumented session rather than raced, so they cannot pass by scheduling luck. Each test asserts @@ -23,9 +28,17 @@ from app.db.models import LedgerEntryModel, WalletModel from app.main import app from app.schemas.billing import ServiceCategory -from app.services.agent_money import get_agent_money +from app.services.agent_money import InsufficientFundsResponse, get_agent_money from tests.test_trust_helpers import provision_agent_wallet +from tests.conftest import interleaving_factory, requires_sqlite_row_lock_noop + +# Every test in this module forces a concurrent writer into an open +# transaction, which is only a *concurrent* writer while the row lock is a +# no-op. See the marker's definition for why these are skipped rather than +# adapted on other engines. +pytestmark = requires_sqlite_row_lock_noop + @pytest.fixture async def client(): @@ -67,46 +80,15 @@ async def _set_balance(wallet_id: str, amount: Decimal) -> None: session.add(wallet) -def _interleaving_factory(real_factory, hook, state, *, fire_on: int = 1): - """Wrap a session factory so `hook` runs once, mid-transaction. +def _exact_amount(entry) -> Decimal: + """Magnitude of a ledger entry as a Decimal. - Fires after ``fire_on`` statements have been issued on the wrapped session, - which places it after the balance has been read and before it is written. + ``LedgerEntry.amount`` is a float for API compatibility; ``amount_exact`` + carries the decimal string. Balances are Decimal, so comparing against the + float is both a TypeError and, if coerced, a rounding hazard on exactly + the arithmetic these tests exist to check. """ - - class _Session: - def __init__(self, inner): - self._inner = inner - - def __getattr__(self, name): - return getattr(self._inner, name) - - async def execute(self, *args, **kwargs): - state["executes"] = state.get("executes", 0) + 1 - n = state["executes"] - # Fire *after* the statement returns, so the hook lands once the - # value has been read but before anything is written back. Firing - # beforehand would place the concurrent writer ahead of the read, - # which is not the race. Counting completed reads also keeps the - # seam identical across the fixed and unfixed code, which differ in - # how many statements they issue. - result = await self._inner.execute(*args, **kwargs) - if n == fire_on and not state.get("fired"): - state["fired"] = True - await hook() - return result - - class _CM: - def __init__(self, cm): - self._cm = cm - - async def __aenter__(self): - return _Session(await self._cm.__aenter__()) - - async def __aexit__(self, *exc): - return await self._cm.__aexit__(*exc) - - return lambda: (lambda: _CM(real_factory())) + return abs(Decimal(entry.amount_exact or str(entry.amount))) @pytest.mark.anyio @@ -149,7 +131,7 @@ async def _concurrent_charge() -> None: monkeypatch.setattr( money._billing_engine, "_session_factory", - _interleaving_factory(real_factory, _concurrent_charge, state, fire_on=1), + interleaving_factory(real_factory, _concurrent_charge, state, fire_on=1), ) state["first"] = await money.charge( @@ -218,7 +200,7 @@ async def _concurrent_reclaim() -> None: monkeypatch.setattr( money._wallet_engine, "_session_factory", - _interleaving_factory(real_factory, _concurrent_reclaim, state, fire_on=1), + interleaving_factory(real_factory, _concurrent_reclaim, state, fire_on=1), ) try: @@ -276,7 +258,7 @@ async def _concurrent_charge() -> None: monkeypatch.setattr( money._billing_engine, "_session_factory", - _interleaving_factory(real_factory, _concurrent_charge, state, fire_on=1), + interleaving_factory(real_factory, _concurrent_charge, state, fire_on=1), ) state["first"] = await money.charge( @@ -338,7 +320,7 @@ async def _concurrent_charge() -> None: monkeypatch.setattr( money._billing_engine, "_session_factory", - _interleaving_factory(real_factory, _concurrent_charge, state, fire_on=1), + interleaving_factory(real_factory, _concurrent_charge, state, fire_on=1), ) state["first"] = await money.charge( @@ -393,7 +375,7 @@ async def _freeze_the_wallet() -> None: monkeypatch.setattr( money._billing_engine, "_session_factory", - _interleaving_factory(real_factory, _freeze_the_wallet, state, fire_on=1), + interleaving_factory(real_factory, _freeze_the_wallet, state, fire_on=1), ) result = await money.charge( @@ -412,3 +394,236 @@ async def _freeze_the_wallet() -> None: assert result.shortfall_exact == "0" assert await _debits(wallet_id) == Decimal("0") assert await _balance(wallet_id) == Decimal("100") + + +@pytest.mark.anyio +async def test_a_concurrent_charge_cannot_erase_a_refund_credit( + client, clean_database, monkeypatch +): + """A refund credit must survive a charge landing mid-transaction. + + ``refund_charge`` read the wallet, then several statements later wrote the + credit. Written as ``balance + refund_amount`` from that read, a charge + committing in the gap is erased โ€” the ledger shows the debit, the balance + does not, and the customer keeps money the service already delivered + against. The duplicate-refund guards inside ``refund_charge`` do not help: + the two operations here are a refund and a charge, not two refunds. + """ + provisioned = await provision_agent_wallet(client) + wallet_id = provisioned["agent_wallet_id"] + await _set_balance(wallet_id, Decimal("100")) + + money = get_agent_money() + import app.services.billing_engine as billing_module + + real_factory = get_session_factory() + + # A real debit to refund. Its amount is what the credit will restore. + charge = await money.charge( + wallet_id=wallet_id, + service_category=ServiceCategory.AGENT_COMMS, + units=Decimal("1"), + request_path="/to-be-refunded", + ) + charged = _exact_amount(charge) + opening = await _balance(wallet_id) + + state: dict = {} + + async def _concurrent_charge() -> None: + monkeypatch.setattr( + billing_module, "get_session_factory", lambda: real_factory, raising=False + ) + state["second"] = await money.charge( + wallet_id=wallet_id, + service_category=ServiceCategory.AGENT_COMMS, + units=Decimal("1"), + request_path="/concurrent-during-refund", + ) + + monkeypatch.setattr( + money._billing_engine, + "_session_factory", + interleaving_factory(real_factory, _concurrent_charge, state, fire_on=1), + ) + + await money.refund_charge( + wallet_id=wallet_id, + charge_entry_id=charge.entry_id, + description="refund during a concurrent charge", + ) + + assert state.get("fired"), "the interleave never ran โ€” the test proved nothing" + second = _exact_amount(state["second"]) + assert second > Decimal("0"), "the concurrent charge did not move money" + # The credit and the concurrent debit must both be on the balance. + assert await _balance(wallet_id) == opening - second + charged + + +@pytest.mark.anyio +async def test_a_concurrent_charge_cannot_erase_a_reconciled_refund( + client, clean_database, monkeypatch +): + """``refund_reconciliation._apply_refund`` has the same shape. + + It is the repair path for a charge whose tool call never delivered, so a + lost credit here is money taken for work that provably did not happen โ€” + and it runs as a background sweep, precisely when other charges on the + same wallet are most likely to be in flight. + """ + from app.services.refund_reconciliation import get_refund_reconciliation_service + + provisioned = await provision_agent_wallet(client) + wallet_id = provisioned["agent_wallet_id"] + await _set_balance(wallet_id, Decimal("100")) + + money = get_agent_money() + import app.services.billing_engine as billing_module + + real_factory = get_session_factory() + + charge = await money.charge( + wallet_id=wallet_id, + service_category=ServiceCategory.AGENT_COMMS, + units=Decimal("1"), + request_path="/undelivered", + ) + charged = _exact_amount(charge) + opening = await _balance(wallet_id) + + service = get_refund_reconciliation_service() + state: dict = {} + + async def _concurrent_charge() -> None: + monkeypatch.setattr( + billing_module, "get_session_factory", lambda: real_factory, raising=False + ) + state["second"] = await money.charge( + wallet_id=wallet_id, + service_category=ServiceCategory.AGENT_COMMS, + units=Decimal("1"), + request_path="/concurrent-during-reconcile", + ) + + async with real_factory() as session: + async with session.begin(): + wallet = await session.get(WalletModel, wallet_id) + assert wallet is not None + charge_row = await session.get(LedgerEntryModel, charge.entry_id) + assert charge_row is not None + + # ``_apply_refund`` takes an already-loaded wallet, so the read + # this race turns on is the ``session.get`` above. The interleave + # therefore goes here โ€” after the service has the wallet in hand + # and before it writes โ€” rather than through the session wrapper + # the other tests use. + await _concurrent_charge() + state["fired"] = True + await service._apply_refund( + session=session, + wallet=wallet, + charge=charge_row, + amount=charged, + ) + + assert state.get("fired"), "the interleave never ran โ€” the test proved nothing" + second = _exact_amount(state["second"]) + assert second > Decimal("0"), "the concurrent charge did not move money" + assert await _balance(wallet_id) == opening - second + charged + + +@pytest.mark.anyio +async def test_a_period_rollover_is_not_charged_for_a_rejected_charge( + client, clean_database, monkeypatch +): + """A reversal must not decrement a period the charge never contributed to. + + ``check_and_record_charge`` commits the velocity increment before the + debit transaction takes the wallet, so a rejected debit has to compensate + it. The counters roll over on their own schedule, though, and a rollover + landing in between zeroes the counter this charge added to. Reversing + against the *new* period takes credits off a total that belongs to live + spend by other callers. + + The direction matters. An over-count throttles a caller who did not spend + and heals at the next rollover; an under-count silently raises the + effective spend cap and delays the anomaly auto-freeze, which are the two + controls these counters exist to drive. So the reversal is guarded on the + period marker it was recorded against and skipped when that no longer + holds. + """ + from datetime import timedelta + + from app.core.time import utc_now + + provisioned = await provision_agent_wallet(client) + wallet_id = provisioned["agent_wallet_id"] + await _set_balance(wallet_id, Decimal("100")) + + money = get_agent_money() + real_factory = get_session_factory() + state: dict = {} + + async def _roll_the_period_and_freeze() -> None: + """The hourly and daily windows roll, then the wallet is frozen. + + The freeze is what makes the debit reject, so the reversal runs. The + rollover is backdated past both windows and the counters are set to a + known live figure, standing in for spend by other callers in the new + period. + """ + async with real_factory() as s: + async with s.begin(): + wallet = await s.get(WalletModel, wallet_id) + assert wallet is not None + now = utc_now() + wallet.hourly_reset_at = now + wallet.daily_reset_at = now + wallet.hourly_spent = Decimal("7") + wallet.daily_spent = Decimal("9") + wallet.status = "frozen" + s.add(wallet) + + # Put the wallet in an old period so the charge's own increment is + # recorded against a marker the hook below then replaces. + async with real_factory() as session: + async with session.begin(): + wallet = await session.get(WalletModel, wallet_id) + assert wallet is not None + stale = utc_now() - timedelta(days=2) + wallet.hourly_reset_at = stale + wallet.daily_reset_at = stale + session.add(wallet) + + monkeypatch.setattr( + money._billing_engine, + "_session_factory", + interleaving_factory(real_factory, _roll_the_period_and_freeze, state, fire_on=1), + ) + + result = await money.charge( + wallet_id=wallet_id, + service_category=ServiceCategory.AGENT_COMMS, + units=Decimal("1"), + request_path="/rollover-reversal", + ) + + assert state.get("fired"), "the interleave never ran โ€” the test proved nothing" + # Pin the response, not merely the absence of a ledger entry. The freeze + # commits after the wallet SELECT returns, so the in-memory wallet still + # reads "active" at the status check and it is the guarded UPDATE that + # rejects the charge. Both branches reach the reversal this test measures, + # but they are separately reachable, and ``not hasattr(result, "entry_id")`` + # records neither. + assert isinstance(result, InsufficientFundsResponse), ( + f"a frozen wallet was debited: {result}" + ) + assert result.error == "wallet_frozen" + + async with real_factory() as session: + wallet = await session.get(WalletModel, wallet_id) + assert wallet is not None + # The new period's counters are untouched: this charge never added to + # them, so it has nothing to take back from them. + assert wallet.hourly_spent == Decimal("7") + assert wallet.daily_spent == Decimal("9") diff --git a/tests/test_wedge_honesty.py b/tests/test_wedge_honesty.py index 8813790a..525a09cd 100644 --- a/tests/test_wedge_honesty.py +++ b/tests/test_wedge_honesty.py @@ -3,6 +3,8 @@ from __future__ import annotations import json +import re +from pathlib import Path import pytest from httpx import ASGITransport, AsyncClient @@ -108,11 +110,37 @@ async def test_agent_json_sdk_integrations_are_honest(client): python_sdk = integrations["python_sdk"] assert isinstance(python_sdk, dict) assert python_sdk["status"] == "release_artifact_only" - assert python_sdk["version"] == "0.4.0" + # The advertised version must be the one an editable install actually + # yields. Read it from the SDK's own pyproject rather than restating a + # literal: a hand-copied number is exactly what drifts, and this endpoint + # exists to be honest about what a client gets. + sdk_pyproject = ( + Path(__file__).resolve().parent.parent / "b2a_sdk" / "pyproject.toml" + ).read_text(encoding="utf-8") + source_version = re.search( + r'^version\s*=\s*"([^"]+)"', sdk_pyproject, re.MULTILINE + ) + assert source_version is not None, "b2a_sdk/pyproject.toml has no version" + assert python_sdk["version"] == source_version.group(1) assert python_sdk["install"] == "pip install -e ./b2a_sdk" assert "pip install b2a-sdk" not in json.dumps(python_sdk) assert "not published" in python_sdk["note"].lower() - assert "python-sdk-v0.4.0" in python_sdk["note"] + # The release tag is a separate fact from the source version and may + # legitimately lag it. Pin the value rather than checking it appears in + # ``note``: a substring test passes for any tag, including a stale one, + # which is exactly the drift this endpoint is supposed to rule out. + # `python-sdk-v0.4.0` is the newest tag in the repository; bump this + # line in the same change that cuts the next release. + assert python_sdk["latest_release_tag"] == "python-sdk-v0.4.0" + assert python_sdk["latest_release_tag"] in python_sdk["note"] + # The two must not silently converge: the whole point of reporting them + # separately is that a source ahead of the last tag stays visible. + # Normalize before comparing -- "0.5.0" and "python-sdk-v0.4.0" differ as + # raw strings even when they name the same release, so a bare inequality + # would hold no matter what and assert nothing at all. + assert python_sdk["latest_release_tag"] != ( + f"python-sdk-v{python_sdk['version']}" + ) typescript_sdk = integrations["typescript_sdk"] assert isinstance(typescript_sdk, dict)