From 52a4200a6fe73963928bce95fb6a3eaddc3b7ee6 Mon Sep 17 00:00:00 2001 From: ATC964 <82515918+atc964@users.noreply.github.com> Date: Thu, 17 Sep 2026 17:32:44 -0400 Subject: [PATCH 1/2] fix: honour an accepted negotiation's price when its quote is booked Booking resolved an accepted negotiation with storage.get_negotiation(request.quote_id). That only ever hits for a quote-led negotiation, which the router's key derivation happens to store under the quote id. Our buyer is proposal-led, so its negotiation is stored under a prop- id, the lookup missed every time, and the deal was booked at the seller's standard price while the log reported the negotiation as successful. NegotiationHistory now retains the quote_id that already arrived on NegotiationMessage and was being dropped. An accepted negotiation indexes itself as negotiation_by_quote:{quote_id}, holding the key its record is stored under, and booking resolves through that pointer. The direct read stays as a fallback so quote-led negotiations recorded before the index existed are still honoured. A pointer rather than a second copy of the record: rounds are appended to it, so two copies would drift on the first append. An accepted negotiation carrying no agreed price is now refused with negotiation_price_unresolved instead of quietly booking at the un-negotiated quoted price. A record in that state is corrupt, and list-pricing it is the worst available outcome. The negotiation storage key is also resolved against the store rather than read off whichever ids a given message happened to carry, so a continuation leading with a different id continues the same negotiation instead of silently restarting it at round one. Which id a negotiation is stored under is unchanged. Behaviour change: bookings that silently ignored an accepted negotiation now honour it, so a booked final_cpm can differ from the quoted one and its rationale says why. --- CHANGELOG.md | 20 + .../interfaces/api/routers/negotiation.py | 19 +- src/ad_seller/models/negotiation.py | 5 + src/ad_seller/services/deal_service.py | 88 ++- src/ad_seller/services/negotiation_service.py | 91 ++- src/ad_seller/storage/base.py | 15 + tests/unit/test_audience_plan_validation.py | 4 + .../test_negotiation_quote_correlation.py | 668 ++++++++++++++++++ 8 files changed, 884 insertions(+), 26 deletions(-) create mode 100644 tests/unit/test_negotiation_quote_correlation.py diff --git a/CHANGELOG.md b/CHANGELOG.md index 4a408c9a..e43397e8 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -38,6 +38,26 @@ All notable changes to the IAB Tech Lab Seller Agent are documented here. orphaned worker thread (CrewAI has no cancellation API). Budget <= 0 still means "no bound" and always runs the crew; wire answers on the timeout path are unchanged. +- **A negotiated price now actually books.** Booking looked for an accepted + negotiation with `get_negotiation(quote_id)`, which only ever hit for a + quote-led negotiation, since that one happens to be stored under the quote + id. A proposal-led negotiation is stored under its `prop-` id, so the + lookup missed and the deal was booked at the seller's standard price with + the negotiation reported as successful. `NegotiationHistory` now retains + the `quote_id` that already arrived on `NegotiationMessage` and was being + dropped, an accepted negotiation indexes itself as + `negotiation_by_quote:{quote_id}`, and booking resolves through that index + (falling back to the direct read, so quote-led negotiations recorded before + the index existed are still honored). **Behavior change:** bookings that + silently ignored an accepted negotiation now honor it, so a booked + `final_cpm` can differ from the quoted one — it carries the agreed price + and says so in its rationale. An accepted negotiation carrying no agreed + price is now refused (`negotiation_price_unresolved`) rather than quietly + booked at the un-negotiated quoted price. Relatedly, the negotiation + storage key is now resolved against the store instead of being read off + whichever ids a given message happened to carry, so a continuation leading + with a different id continues the same negotiation rather than silently + restarting it at round one. ### Fixed diff --git a/src/ad_seller/interfaces/api/routers/negotiation.py b/src/ad_seller/interfaces/api/routers/negotiation.py index a6ecceb8..144e9621 100644 --- a/src/ad_seller/interfaces/api/routers/negotiation.py +++ b/src/ad_seller/interfaces/api/routers/negotiation.py @@ -176,8 +176,17 @@ async def post_negotiation_message( # The seller keys negotiations by proposal_id; negotiation_id doubles as # that key for continuation, and quote-led opens key off the quote_id - # (the service resolves the stored quote to its product). - proposal_id = message.proposal_id or message.negotiation_id or message.quote_id + # (the service resolves the stored quote to its product). Which of those + # is the key is resolved against the STORE rather than read off this + # message: taking `proposal_id or negotiation_id or quote_id` made the key + # depend on the ids a given message happened to carry, so a continuation + # leading with a different id resolved to a different record and silently + # restarted the negotiation. + proposal_id = await negotiation_service.resolve_negotiation_key( + proposal_id=message.proposal_id, + negotiation_id=message.negotiation_id, + quote_id=message.quote_id, + ) if proposal_id is None: # unreachable: the shared model requires one key raise HTTPException( status_code=400, @@ -226,15 +235,17 @@ async def post_negotiation_message( proposal_id=proposal_id, buyer_price=buyer_price, buyer_context=buyer_context, + quote_id=message.quote_id, ) response = cm.negotiation_round_to_response(result) else: # accept / reject — terminal moves off the recorded history; the # price engine is not run (it stays untouched). The move is # PERSISTED onto the stored negotiation so downstream booking sees - # the agreed state. + # the agreed state, and an accept indexes the negotiation by its + # quote so booking that quote can find the agreed price. status_data = await negotiation_service.apply_terminal_action( - proposal_id, message.action.value, buyer_price + proposal_id, message.action.value, buyer_price, quote_id=message.quote_id ) response = cm.terminal_round_response(status_data, message.action, buyer_price) diff --git a/src/ad_seller/models/negotiation.py b/src/ad_seller/models/negotiation.py index d0579176..53ab2162 100644 --- a/src/ad_seller/models/negotiation.py +++ b/src/ad_seller/models/negotiation.py @@ -101,6 +101,11 @@ class NegotiationHistory(BaseModel): negotiation_id: str = Field(default_factory=lambda: f"neg-{uuid.uuid4().hex[:8]}") proposal_id: str + # The quote this negotiation concerns, when one was named. It arrives on + # the wire (``NegotiationMessage.quote_id``) and used to be dropped here, + # so nothing could correlate a negotiation to the quote a buyer books — + # booking then struck the stale quoted price and logged success. + quote_id: Optional[str] = None product_id: str buyer_tier: AccessTier strategy: NegotiationStrategy diff --git a/src/ad_seller/services/deal_service.py b/src/ad_seller/services/deal_service.py index 5609b066..4c234730 100644 --- a/src/ad_seller/services/deal_service.py +++ b/src/ad_seller/services/deal_service.py @@ -189,6 +189,34 @@ def match_agentic_audience(ref: dict[str, Any]) -> dict[str, Any]: # ============================================================================= +async def _resolve_accepted_negotiation(storage: Any, quote_id: str) -> Optional[dict[str, Any]]: + """Return the ACCEPTED negotiation about ``quote_id``, or None. + + ``DealBookingRequest`` carries only the quote id, so the quote is the only + handle the protocol offers for correlating a booking to a negotiation. + An accepted negotiation indexes itself under + ``negotiation_by_quote:{quote_id}`` (see + ``negotiation_service._persist_negotiation``); that pointer names the key + its record is stored under, whichever id the buyer led with. + + The direct read by quote id is retained as a fallback: a quote-led + negotiation is itself stored under the quote id, and records written + before the index existed have no pointer. + """ + from ..storage.base import negotiation_by_quote_key + + negotiation = None + pointer = await storage.get(negotiation_by_quote_key(quote_id)) + if isinstance(pointer, str) and pointer: + negotiation = await storage.get_negotiation(pointer) + if not negotiation: + negotiation = await storage.get_negotiation(quote_id) + + if isinstance(negotiation, dict) and negotiation.get("status") == "accepted": + return negotiation + return None + + async def book_deal(request: Any) -> dict[str, Any]: """Book a deal from a previously issued quote (``DealBookingRequestModel``). @@ -238,27 +266,51 @@ async def book_deal(request: Any) -> dict[str, Any]: }, ) - # Honor ACCEPTED negotiation state on this quote: a - # quote-led negotiation is keyed by the quote_id; when it concluded - # accepted, the booking strikes the AGREED price, not the stale quoted - # price. A re-quote at the agreed price (the buyer's historical - # workaround) carries no negotiation and books unchanged. - negotiation = await storage.get_negotiation(request.quote_id) - if negotiation and negotiation.get("status") == "accepted": + # Honor ACCEPTED negotiation state on this quote: when a negotiation about + # this quote concluded accepted, the booking strikes the AGREED price, not + # the stale quoted price. A quote with no accepted negotiation (including a + # re-quote at the agreed price, the buyer's historical workaround) books + # unchanged. + # + # Resolution goes through the quote index an accepted negotiation writes + # (`negotiation_by_quote:{quote_id}` -> the key its record is stored + # under). Passing the quote id straight to get_negotiation only ever + # worked for a QUOTE-LED negotiation, which happens to be stored under the + # quote id; a proposal-led one is stored under its `prop-` id, so the + # lookup missed every time and the agreed price was silently dropped. The + # direct read is kept as a fallback so quote-led negotiations recorded + # before this index existed are still honored. + negotiation = await _resolve_accepted_negotiation(storage, request.quote_id) + if negotiation: rounds = negotiation.get("rounds") or [] agreed_cpm = rounds[-1].get("seller_price") if rounds else None - if agreed_cpm is not None: - quote = { - **quote, - "pricing": { - **quote["pricing"], - "final_cpm": agreed_cpm, - "rationale": ( - f"{quote['pricing'].get('rationale', '')} | Negotiated to " - f"${agreed_cpm:.2f} CPM ({negotiation.get('negotiation_id')})" - ).strip(" |"), + if agreed_cpm is None: + # A negotiation that concluded accepted with no price is a corrupt + # record. Booking it at list price is the worst available outcome — + # it is exactly the silent overpayment this whole path exists to + # prevent — so refuse instead. + raise HTTPException( + status_code=500, + detail={ + "error": "negotiation_price_unresolved", + "message": ( + f"Negotiation '{negotiation.get('negotiation_id')}' for quote " + f"'{request.quote_id}' is accepted but carries no agreed price. " + "Refusing to book at the un-negotiated quoted price." + ), }, - } + ) + quote = { + **quote, + "pricing": { + **quote["pricing"], + "final_cpm": agreed_cpm, + "rationale": ( + f"{quote['pricing'].get('rationale', '')} | Negotiated to " + f"${agreed_cpm:.2f} CPM ({negotiation.get('negotiation_id')})" + ).strip(" |"), + }, + } # Pre-flight: if the buyer sent an audience_plan with this booking, validate # it against the seller's capability block. Per proposal §5.7 layer 3, any diff --git a/src/ad_seller/services/negotiation_service.py b/src/ad_seller/services/negotiation_service.py index 32f27b4f..6389836a 100644 --- a/src/ad_seller/services/negotiation_service.py +++ b/src/ad_seller/services/negotiation_service.py @@ -38,6 +38,66 @@ def build_negotiation_engine(): return NegotiationEngine(pricing_engine, yield_opt) +async def _persist_negotiation(storage: Any, proposal_id: str, data: dict[str, Any]) -> None: + """Store a negotiation, and index it by the quote it concerns. + + Single write path for every stored negotiation so the quote index cannot + be forgotten at one of the three call sites. The index entry is written + only once the negotiation has concluded ``accepted`` and carries a + ``quote_id``: only an accepted negotiation changes a booking price, so + presence in the index means exactly "this quote has an agreed price" + (resolved at booking by ``deal_service.book_deal``). + + A pointer, not a duplicated record — the record is mutable because rounds + are appended, so two copies of it would drift on the first append. + """ + from ..storage.base import negotiation_by_quote_key + + await storage.set_negotiation(proposal_id, data) + + quote_id = data.get("quote_id") + if data.get("status") == "accepted" and quote_id: + await storage.set(negotiation_by_quote_key(quote_id), proposal_id) + + +async def resolve_negotiation_key( + proposal_id: Optional[str] = None, + negotiation_id: Optional[str] = None, + quote_id: Optional[str] = None, +) -> Optional[str]: + """Resolve a negotiation message's ids to the key its history is stored under. + + The seller keys negotiations by ``proposal_id``, and a quote-led open keys + off the ``quote_id``. Deriving that key as ``proposal_id or negotiation_id + or quote_id`` made it depend on which ids a given MESSAGE happened to + carry rather than on the negotiation: a round-one open keyed on one id + followed by a continuation carrying a different id resolved to two + different records, and the continuation silently restarted the + negotiation from round one. + + So every supplied id is tried against the store first and an existing + negotiation wins, whichever id the continuation leads with. A new + negotiation is minted under the first supplied id — the previous + or-chain order, unchanged — only when none of them resolves. Which id a + negotiation is stored under is deliberately NOT changed here; re-keying + storage on ``negotiation_id`` is the structural fix. + + Returns ``None`` only when no id was supplied at all. + """ + from ..storage.factory import get_storage + + candidates = [candidate for candidate in (proposal_id, negotiation_id, quote_id) if candidate] + if not candidates: + return None + + storage = await get_storage() + for candidate in candidates: + if await storage.get_negotiation(candidate): + return candidate + + return candidates[0] + + async def submit_proposal( request: Any, buyer_context: Any, catalog: dict[str, Any] ) -> dict[str, Any]: @@ -119,7 +179,7 @@ async def submit_proposal( # round numbering) instead of silently starting over. flow_history = result.pop("_negotiation_history", None) if flow_history: - await storage.set_negotiation(proposal_id, flow_history) + await _persist_negotiation(storage, proposal_id, flow_history) quote_history = QuoteHistoryStore(storage) verification = await quote_history.verify_pricing( buyer_id=buyer_context.get_pricing_key(), @@ -190,11 +250,18 @@ async def counter_proposal( proposal_id: str, buyer_price: float, buyer_context: Any, + quote_id: Optional[str] = None, ) -> dict[str, Any]: """Submit a counter-offer in an ongoing negotiation. Loads or creates a NegotiationHistory, evaluates the buyer's offer, persists the updated history, and emits a NEGOTIATION_ROUND event. + + ``quote_id`` is the quote the buyer is negotiating, as supplied on the + inbound message. It is recorded on the history so an accepted negotiation + can be found again from the quote that gets booked. The first one seen + wins: a later round naming a different quote must not silently move which + quote the agreed price applies to. """ from ..events.helpers import emit_event from ..events.models import EventType @@ -224,6 +291,9 @@ async def counter_proposal( if not quote_data: raise HTTPException(status_code=404, detail="Proposal not found") product_id = (quote_data.get("product") or {}).get("product_id", "") + # Quote-led: the key IS the quote, so the quote id has arrived + # even when the message did not spell it out in its own field. + quote_id = quote_id or proposal_id product_data = await storage.get_product(product_id) if not product_data: @@ -262,12 +332,17 @@ async def counter_proposal( }, ) + # Retain the quote this negotiation concerns. Backfills a negotiation + # opened before the quote was named; first one seen wins. + if quote_id and not history.quote_id: + history.quote_id = quote_id + # Evaluate buyer's offer round_result = neg_engine.evaluate_buyer_offer(history, buyer_price, buyer_context) history = neg_engine.record_round(history, round_result) - # Persist - await storage.set_negotiation(proposal_id, history.model_dump(mode="json")) + # Persist (and index by quote, if this round concluded accepted) + await _persist_negotiation(storage, proposal_id, history.model_dump(mode="json")) # Emit round event await emit_event( @@ -313,6 +388,7 @@ async def apply_terminal_action( proposal_id: str, action: str, buyer_price: Optional[float] = None, + quote_id: Optional[str] = None, ) -> dict[str, Any]: """Record a buyer's terminal accept/reject on the stored negotiation. @@ -321,6 +397,10 @@ async def apply_terminal_action( the agreed state. The price engine is not run: accept strikes at the seller's last stated price; reject is a walk-away. + ``quote_id`` backfills the quote this negotiation concerns when the + accept is the first message to name it. On accept, this is the point the + quote index is written, which is what lets booking find the agreed price. + No-ops (returning current status) when no stored negotiation exists or it is already terminal, preserving the read-only legacy behavior. """ @@ -332,6 +412,8 @@ async def apply_terminal_action( if data: history = NegotiationHistory(**data) if history.status == "active" and action in ("accept", "reject"): + if quote_id and not history.quote_id: + history.quote_id = quote_id last = history.rounds[-1] if history.rounds else None seller_price = last.seller_price if last else history.base_price history.rounds.append( @@ -349,7 +431,7 @@ async def apply_terminal_action( ) history.status = "accepted" if action == "accept" else "rejected" history.completed_at = datetime.utcnow() - await storage.set_negotiation(proposal_id, history.model_dump(mode="json")) + await _persist_negotiation(storage, proposal_id, history.model_dump(mode="json")) return await get_negotiation_status(proposal_id) @@ -379,6 +461,7 @@ async def get_negotiation_status(proposal_id: str) -> dict[str, Any]: return { "negotiation_id": history.negotiation_id, "proposal_id": history.proposal_id, + "quote_id": history.quote_id, "product_id": history.product_id, "buyer_tier": history.buyer_tier.value, "status": history.status, diff --git a/src/ad_seller/storage/base.py b/src/ad_seller/storage/base.py index c09167e3..cb5d5c50 100644 --- a/src/ad_seller/storage/base.py +++ b/src/ad_seller/storage/base.py @@ -7,6 +7,21 @@ from typing import Any, Optional +def negotiation_by_quote_key(quote_id: str) -> str: + """Key of the pointer from a quote to the negotiation about it. + + The value stored under this key is the id the negotiation record is keyed + by (see :meth:`StorageBackend.get_negotiation`), so booking can resolve a + quote to the price agreed for it. A pointer rather than a second copy of + the record: the record is mutable because rounds are appended to it, so + two copies would drift on the first append. + + Written only when a negotiation concludes ``accepted``, so presence in + this index means exactly "this quote has an agreed price". + """ + return f"negotiation_by_quote:{quote_id}" + + class StorageBackend(ABC): """Abstract base class for storage backends.""" diff --git a/tests/unit/test_audience_plan_validation.py b/tests/unit/test_audience_plan_validation.py index 3e40d1b6..1c94cd28 100644 --- a/tests/unit/test_audience_plan_validation.py +++ b/tests/unit/test_audience_plan_validation.py @@ -455,6 +455,10 @@ def http_client(): ), "get_deal": staticmethod(lambda did: _async_return(store.get(f"deal:{did}"))), "set_deal": staticmethod(lambda did, data: _async_set(store, f"deal:{did}", data)), + # Generic KV path: booking resolves a quote's negotiation through + # the negotiation_by_quote pointer before reading the record. + "get": staticmethod(lambda key: _async_return(store.get(key))), + "set": staticmethod(lambda key, value, ttl=None: _async_set(store, key, value)), # Booking consults negotiation state; none stored here. "get_negotiation": staticmethod( lambda pid: _async_return(store.get(f"negotiation:{pid}")) diff --git a/tests/unit/test_negotiation_quote_correlation.py b/tests/unit/test_negotiation_quote_correlation.py new file mode 100644 index 00000000..a94f155c --- /dev/null +++ b/tests/unit/test_negotiation_quote_correlation.py @@ -0,0 +1,668 @@ +# Author: Green Mountain Systems AI Inc. +# Donated to IAB Tech Lab + +"""A negotiated price must actually book. + +The defect: ``deal_service.book_deal`` honored an accepted negotiation by +calling ``storage.get_negotiation(request.quote_id)``. Negotiations are keyed +by the id the opening message led with, so that lookup only ever hit for a +QUOTE-LED negotiation. A proposal-led buyer -- which is what ours is -- has +its negotiation stored under a ``prop-`` id, so the lookup missed every time: +the booked deal carried the seller's standard price while the log said +"Negotiation succeeded". + +Three things are pinned here: + +1. ``NegotiationHistory.quote_id`` -- the quote id arrives on + ``NegotiationMessage`` and used to be dropped, so nothing could correlate + a negotiation to the quote a buyer books. +2. The quote index -- an accepted negotiation writes + ``negotiation_by_quote:{quote_id}`` naming the key its record lives under, + and booking resolves through it whichever id the buyer led with. A + negotiation that concluded accepted but carries no price refuses to book + rather than silently falling back to the quoted price. +3. Key resolution (fault A2) -- the storage key is resolved against the store + rather than read off whichever ids a given message happened to carry, so a + continuation leading with a different id continues the SAME negotiation + instead of silently restarting it. +""" + +import os +import sys +from datetime import datetime, timedelta +from types import ModuleType +from unittest.mock import AsyncMock, MagicMock, patch + +import pytest + +# Settings require an Anthropic key even for key-less unit runs; same idiom +# as test_negotiation_cold_start.py (no LLM call is ever made here). +os.environ.setdefault("ANTHROPIC_API_KEY", "test-key-for-unit-tests") + +# Stub execution_activation_flow (cancel-scope leak on ad-server +# connection failure, unresolved -- issue #60 part 2). +_broken_flows = [ + "ad_seller.flows.execution_activation_flow", +] +for _mod_name in _broken_flows: + if _mod_name not in sys.modules: + _stub = ModuleType(_mod_name) + _cls_name = _mod_name.rsplit(".", 1)[-1].replace("_", " ").title().replace(" ", "") + setattr(_stub, _cls_name, type(_cls_name, (), {})) + sys.modules[_mod_name] = _stub + +import httpx # noqa: E402 +from fastapi import HTTPException # noqa: E402 +from httpx import ASGITransport # noqa: E402 + +from ad_seller.interfaces.api.main import _get_optional_api_key_record, app # noqa: E402 +from ad_seller.models.buyer_identity import BuyerIdentity # noqa: E402 +from ad_seller.services import deal_service, negotiation_service # noqa: E402 + +pytestmark = pytest.mark.asyncio + +QUOTE_ID = "qt-corr123456" +PROPOSAL_ID = "prop-corr01" +PRODUCT_ID = "ctv-premium-sports" +QUOTED_CPM = 28.26 + + +# ============================================================================= +# Helpers +# ============================================================================= + + +def _available_quote(quote_id=QUOTE_ID, product_id=PRODUCT_ID, final_cpm=QUOTED_CPM): + return { + "quote_id": quote_id, + "status": "available", + "deal_type": "PD", + "product": { + "product_id": product_id, + "name": "Premium CTV - Sports", + "inventory_type": "ctv", + }, + "pricing": { + "base_cpm": 35.0, + "tier_discount_pct": 15.0, + "volume_discount_pct": 5.0, + "final_cpm": final_cpm, + "currency": "USD", + "pricing_model": "cpm", + "rationale": "Base price: $35.00 CPM | Advertiser tier: -15%", + }, + "terms": { + "impressions": 5000000, + "flight_start": "2026-04-01", + "flight_end": "2026-04-30", + "guaranteed": False, + }, + "buyer_tier": "advertiser", + "expires_at": (datetime.utcnow() + timedelta(hours=23)).isoformat() + "Z", + "created_at": datetime.utcnow().isoformat() + "Z", + } + + +def _stored_product(product_id=PRODUCT_ID): + return {"product_id": product_id, "base_cpm": 35.0, "floor_cpm": 20.0} + + +def _stored_proposal(proposal_id=PROPOSAL_ID, product_id=PRODUCT_ID): + return { + "proposal_id": proposal_id, + "product_id": product_id, + "deal_type": "preferred_deal", + "price": 24.0, + "impressions": 5_000_000, + "start_date": "2026-04-01", + "end_date": "2026-04-30", + "buyer_id": "buyer-1", + } + + +def _accepted_negotiation( + proposal_id=PROPOSAL_ID, + quote_id=QUOTE_ID, + product_id=PRODUCT_ID, + agreed=24.0, + rounds=None, +): + """A negotiation record that concluded ``accepted``. + + ``rounds=[]`` models the corrupt state step 3 guards: accepted with no + agreed price anywhere on the record. + """ + if rounds is None: + rounds = [ + { + "round_number": 1, + "buyer_price": 22.0, + "seller_price": 26.0, + "action": "counter", + "rationale": "Counter at $26.00", + "timestamp": datetime.utcnow().isoformat(), + }, + { + "round_number": 2, + "buyer_price": agreed, + "seller_price": agreed, + "action": "accept", + "rationale": f"Buyer accepted at ${agreed:.2f} CPM.", + "timestamp": datetime.utcnow().isoformat(), + }, + ] + return { + "negotiation_id": "neg-agreed01", + "proposal_id": proposal_id, + "quote_id": quote_id, + "product_id": product_id, + "buyer_tier": "advertiser", + "strategy": "premium", + "limits": { + "max_rounds": 6, + "per_round_concession_cap": 0.06, + "total_concession_cap": 0.20, + "gap_split_buyer_share": 0.65, + }, + "base_price": 29.75, + "floor_price": 20.0, + "rounds": rounds, + "status": "accepted", + "started_at": datetime.utcnow().isoformat(), + "completed_at": datetime.utcnow().isoformat(), + "package_id": None, + } + + +def _buyer_context(buyer_tier="advertiser"): + from ad_seller.interfaces.api.deps import _build_buyer_context + + return _build_buyer_context( + buyer_tier=buyer_tier, agency_id="agency-test-1", advertiser_id="adv-test-1" + ) + + +def _booking_request(quote_id=QUOTE_ID): + request = MagicMock() + request.quote_id = quote_id + request.audience_plan = None + return request + + +@pytest.fixture +def mock_storage(): + store = {} + storage = AsyncMock() + storage.get = AsyncMock(side_effect=lambda k: store.get(k)) + storage.set = AsyncMock(side_effect=lambda k, v, ttl=None: store.__setitem__(k, v)) + storage.delete = AsyncMock(side_effect=lambda k: store.pop(k, None)) + storage.get_quote = AsyncMock(side_effect=lambda qid: store.get(f"quote:{qid}")) + storage.set_quote = AsyncMock( + side_effect=lambda qid, data, ttl=86400: store.__setitem__(f"quote:{qid}", data) + ) + storage.get_deal = AsyncMock(side_effect=lambda did: store.get(f"deal:{did}")) + storage.set_deal = AsyncMock( + side_effect=lambda did, data: store.__setitem__(f"deal:{did}", data) + ) + storage.get_proposal = AsyncMock(side_effect=lambda pid: store.get(f"proposal:{pid}")) + storage.set_proposal = AsyncMock( + side_effect=lambda pid, data: store.__setitem__(f"proposal:{pid}", data) + ) + storage.get_product = AsyncMock(side_effect=lambda pid: store.get(f"product:{pid}")) + storage.set_product = AsyncMock( + side_effect=lambda pid, data: store.__setitem__(f"product:{pid}", data) + ) + storage.get_negotiation = AsyncMock(side_effect=lambda pid: store.get(f"negotiation:{pid}")) + storage.set_negotiation = AsyncMock( + side_effect=lambda pid, data: store.__setitem__(f"negotiation:{pid}", data) + ) + storage._store = store + return storage + + +@pytest.fixture +def client(): + """An authenticated ADVERTISER-tier buyer. + + ``BuyerContext.effective_tier`` needs agency_id AND advertiser_id to + reach ADVERTISER (progressive revelation), which is what the fixture + quote's ``buyer_tier`` requires of a caller booking it. + """ + app.dependency_overrides[_get_optional_api_key_record] = lambda: MagicMock( + identity=BuyerIdentity( + seat_id="seat-test-1", agency_id="agency-test-1", advertiser_id="adv-test-1" + ) + ) + transport = ASGITransport(app=app) + c = httpx.AsyncClient(transport=transport, base_url="http://test") + yield c + app.dependency_overrides.clear() + + +def _seed_negotiable(mock_storage): + """Quote + product + proposal, the state a proposal-led buyer negotiates from.""" + mock_storage._store[f"quote:{QUOTE_ID}"] = _available_quote() + mock_storage._store[f"product:{PRODUCT_ID}"] = _stored_product() + mock_storage._store[f"proposal:{PROPOSAL_ID}"] = _stored_proposal() + + +# ============================================================================= +# Step 1: the quote id arrives on the wire and must be retained +# ============================================================================= + + +class TestQuoteIdRetained: + async def test_open_records_the_quote_id_from_the_message(self, mock_storage): + """A negotiation opened for a quote persists that quote id.""" + _seed_negotiable(mock_storage) + + with ( + patch("ad_seller.storage.factory.get_storage", return_value=mock_storage), + patch("ad_seller.events.helpers.emit_event", new_callable=AsyncMock), + ): + await negotiation_service.counter_proposal( + PROPOSAL_ID, + buyer_price=24.0, + buyer_context=_buyer_context(), + quote_id=QUOTE_ID, + ) + status = await negotiation_service.get_negotiation_status(PROPOSAL_ID) + + assert mock_storage._store[f"negotiation:{PROPOSAL_ID}"]["quote_id"] == QUOTE_ID + # Readable back out, not merely written. + assert status["quote_id"] == QUOTE_ID + + async def test_quote_id_survives_open_counter_accept(self, mock_storage): + """The id survives a full open -> counter -> accept cycle.""" + _seed_negotiable(mock_storage) + + with ( + patch("ad_seller.storage.factory.get_storage", return_value=mock_storage), + patch("ad_seller.events.helpers.emit_event", new_callable=AsyncMock), + ): + await negotiation_service.counter_proposal( + PROPOSAL_ID, buyer_price=22.0, buyer_context=_buyer_context(), quote_id=QUOTE_ID + ) + # A later round need not repeat the quote id. + await negotiation_service.counter_proposal( + PROPOSAL_ID, buyer_price=23.5, buyer_context=_buyer_context() + ) + status = await negotiation_service.apply_terminal_action(PROPOSAL_ID, "accept") + + assert status["quote_id"] == QUOTE_ID + assert status["status"] == "accepted" + + async def test_proposal_only_negotiation_stores_none(self, mock_storage): + """A pure proposal-led negotiation still works, with quote_id None.""" + _seed_negotiable(mock_storage) + + with ( + patch("ad_seller.storage.factory.get_storage", return_value=mock_storage), + patch("ad_seller.events.helpers.emit_event", new_callable=AsyncMock), + ): + result = await negotiation_service.counter_proposal( + PROPOSAL_ID, buyer_price=24.0, buyer_context=_buyer_context() + ) + + assert result["round_number"] == 1 + assert mock_storage._store[f"negotiation:{PROPOSAL_ID}"]["quote_id"] is None + + async def test_quote_id_is_not_overwritten_by_a_later_round(self, mock_storage): + """A later round naming a DIFFERENT quote must not silently move which + quote the agreed price applies to.""" + _seed_negotiable(mock_storage) + + with ( + patch("ad_seller.storage.factory.get_storage", return_value=mock_storage), + patch("ad_seller.events.helpers.emit_event", new_callable=AsyncMock), + ): + await negotiation_service.counter_proposal( + PROPOSAL_ID, buyer_price=22.0, buyer_context=_buyer_context(), quote_id=QUOTE_ID + ) + await negotiation_service.counter_proposal( + PROPOSAL_ID, + buyer_price=23.0, + buyer_context=_buyer_context(), + quote_id="qt-someotherquote", + ) + + assert mock_storage._store[f"negotiation:{PROPOSAL_ID}"]["quote_id"] == QUOTE_ID + + +# ============================================================================= +# Step 2: an accepted negotiation is findable from the quote it concerns +# ============================================================================= + + +class TestQuoteIndex: + async def test_accept_writes_the_quote_pointer(self, mock_storage): + """Accepting indexes the negotiation by its quote, pointing at the key + the record is actually stored under.""" + _seed_negotiable(mock_storage) + + with ( + patch("ad_seller.storage.factory.get_storage", return_value=mock_storage), + patch("ad_seller.events.helpers.emit_event", new_callable=AsyncMock), + ): + await negotiation_service.counter_proposal( + PROPOSAL_ID, buyer_price=22.0, buyer_context=_buyer_context(), quote_id=QUOTE_ID + ) + await negotiation_service.apply_terminal_action(PROPOSAL_ID, "accept") + + assert mock_storage._store[f"negotiation_by_quote:{QUOTE_ID}"] == PROPOSAL_ID + + async def test_active_negotiation_writes_no_pointer(self, mock_storage): + """Only an ACCEPTED negotiation changes a booking price, so only an + accepted one appears in the index.""" + _seed_negotiable(mock_storage) + + with ( + patch("ad_seller.storage.factory.get_storage", return_value=mock_storage), + patch("ad_seller.events.helpers.emit_event", new_callable=AsyncMock), + ): + await negotiation_service.counter_proposal( + PROPOSAL_ID, buyer_price=22.0, buyer_context=_buyer_context(), quote_id=QUOTE_ID + ) + + assert mock_storage._store[f"negotiation:{PROPOSAL_ID}"]["status"] == "active" + assert f"negotiation_by_quote:{QUOTE_ID}" not in mock_storage._store + + async def test_booking_resolves_a_proposal_led_negotiation(self, mock_storage): + """THE bug, at the service boundary: the negotiation is stored under a + ``prop-`` id, so booking can only find it through the index.""" + mock_storage._store[f"quote:{QUOTE_ID}"] = _available_quote() + mock_storage._store[f"negotiation:{PROPOSAL_ID}"] = _accepted_negotiation(agreed=24.0) + mock_storage._store[f"negotiation_by_quote:{QUOTE_ID}"] = PROPOSAL_ID + + with patch("ad_seller.storage.factory.get_storage", return_value=mock_storage): + deal = await deal_service.book_deal(_booking_request()) + + assert deal["pricing"]["final_cpm"] == 24.0 + assert deal["openrtb_params"]["bidfloor"] == 24.0 + assert "Negotiated to $24.00 CPM" in deal["pricing"]["rationale"] + + async def test_booking_ignores_a_pointer_to_an_active_negotiation(self, mock_storage): + """A resolvable but still-active negotiation must not change the price.""" + neg = _accepted_negotiation() + neg["status"] = "active" + neg["rounds"] = neg["rounds"][:1] + neg["completed_at"] = None + mock_storage._store[f"quote:{QUOTE_ID}"] = _available_quote() + mock_storage._store[f"negotiation:{PROPOSAL_ID}"] = neg + mock_storage._store[f"negotiation_by_quote:{QUOTE_ID}"] = PROPOSAL_ID + + with patch("ad_seller.storage.factory.get_storage", return_value=mock_storage): + deal = await deal_service.book_deal(_booking_request()) + + assert deal["pricing"]["final_cpm"] == QUOTED_CPM + assert "Negotiated to" not in deal["pricing"]["rationale"] + + async def test_quote_led_negotiation_still_honored_without_a_pointer(self, mock_storage): + """A quote-led negotiation is stored under the quote id itself. That + path worked before the index existed and must keep working, including + for records written before it.""" + mock_storage._store[f"quote:{QUOTE_ID}"] = _available_quote() + mock_storage._store[f"negotiation:{QUOTE_ID}"] = _accepted_negotiation( + proposal_id=QUOTE_ID, agreed=25.5 + ) + + with patch("ad_seller.storage.factory.get_storage", return_value=mock_storage): + deal = await deal_service.book_deal(_booking_request()) + + assert deal["pricing"]["final_cpm"] == 25.5 + + async def test_negotiated_price_books_end_to_end(self, client, mock_storage): + """The end-to-end case the defect broke: negotiate a price down over + the wire, accept, then book that quote -- the deal must carry the + agreed price and say so in its rationale.""" + _seed_negotiable(mock_storage) + + with ( + patch("ad_seller.storage.factory.get_storage", return_value=mock_storage), + patch("ad_seller.events.helpers.emit_event", new_callable=AsyncMock), + ): + async with client as c: + # Proposal-led, exactly what our buyer sends: proposal_id as + # the negotiation key, quote_id naming the quote at stake. + counter = await c.post( + "/api/v1/negotiations/messages", + json={ + "idempotency_key": "idem-corr-counter", + "action": "counter", + "proposal_id": PROPOSAL_ID, + "quote_id": QUOTE_ID, + "buyer_price": {"amount_micros": 22_000_000, "currency": "USD"}, + }, + ) + assert counter.status_code == 200, counter.text + + accept = await c.post( + "/api/v1/negotiations/messages", + json={ + "idempotency_key": "idem-corr-accept", + "action": "accept", + "proposal_id": PROPOSAL_ID, + "quote_id": QUOTE_ID, + }, + ) + assert accept.status_code == 200, accept.text + assert accept.json()["status"] == "accepted" + + booked = await c.post( + "/api/v1/deals", + json={ + "idempotency_key": "idem-corr-book", + "quote_id": QUOTE_ID, + "buyer_identity": { + "seat_id": "seat-test-1", + "agency_id": "agency-test-1", + "advertiser_id": "adv-test-1", + }, + }, + ) + + assert booked.status_code == 200, booked.text + stored_negotiation = mock_storage._store[f"negotiation:{PROPOSAL_ID}"] + agreed = stored_negotiation["rounds"][-1]["seller_price"] + # The agreed price must actually differ from the quoted one, or this + # test could pass without the fix doing anything. + assert agreed != QUOTED_CPM + + pricing = booked.json()["deal"]["pricing"] + assert pricing["final_cpm"]["amount_micros"] == round(agreed * 1_000_000) + assert f"Negotiated to ${agreed:.2f} CPM" in pricing["rationale"] + + async def test_quote_with_no_negotiation_books_at_the_quoted_price(self, client, mock_storage): + """The negative case: no negotiation, no change. A quote books at the + price it was quoted at.""" + mock_storage._store[f"quote:{QUOTE_ID}"] = _available_quote() + + with patch("ad_seller.storage.factory.get_storage", return_value=mock_storage): + async with client as c: + booked = await c.post( + "/api/v1/deals", + json={ + "idempotency_key": "idem-corr-plain", + "quote_id": QUOTE_ID, + "buyer_identity": { + "seat_id": "seat-test-1", + "agency_id": "agency-test-1", + "advertiser_id": "adv-test-1", + }, + }, + ) + + assert booked.status_code == 200, booked.text + pricing = booked.json()["deal"]["pricing"] + assert pricing["final_cpm"]["amount_micros"] == round(QUOTED_CPM * 1_000_000) + assert "Negotiated to" not in pricing["rationale"] + + +# ============================================================================= +# Step 2b (fault A2): the storage key is resolved, not read off the message +# ============================================================================= + + +class TestNegotiationKeyResolution: + async def test_continuation_with_a_different_id_continues_the_same_negotiation( + self, client, mock_storage + ): + """Round one opened quote-led, so the record is keyed by the quote. A + continuation leading with the proposal_id must resolve to the SAME + negotiation. + + Reading the key off the message (``proposal_id or negotiation_id or + quote_id``) picked the proposal id here, found no record under it, + found the stored PROPOSAL under it, and opened a second negotiation at + round one -- a silent restart that discards every concession already + made. + """ + _seed_negotiable(mock_storage) + + with ( + patch("ad_seller.storage.factory.get_storage", return_value=mock_storage), + patch("ad_seller.events.helpers.emit_event", new_callable=AsyncMock), + ): + async with client as c: + first = await c.post( + "/api/v1/negotiations/messages", + json={ + "idempotency_key": "idem-cont-1", + "action": "counter", + "quote_id": QUOTE_ID, + "buyer_price": {"amount_micros": 22_000_000, "currency": "USD"}, + }, + ) + assert first.status_code == 200, first.text + negotiation_id = first.json()["negotiation_id"] + + second = await c.post( + "/api/v1/negotiations/messages", + json={ + "idempotency_key": "idem-cont-2", + "action": "counter", + "proposal_id": PROPOSAL_ID, + "quote_id": QUOTE_ID, + "buyer_price": {"amount_micros": 23_000_000, "currency": "USD"}, + }, + ) + + assert second.status_code == 200, second.text + body = second.json() + assert body["negotiation_id"] == negotiation_id, "continuation started a new negotiation" + assert body["round"]["round_number"] == 2, "continuation restarted the round numbering" + # One record, under the id round one opened it with. No second record + # minted under the proposal id the continuation led with. + assert len(mock_storage._store[f"negotiation:{QUOTE_ID}"]["rounds"]) == 2 + assert f"negotiation:{PROPOSAL_ID}" not in mock_storage._store + + async def test_resolution_prefers_an_existing_record_over_message_order(self, mock_storage): + """Unit view of the same thing: an id that resolves to a stored + negotiation wins over an id that merely came first on the message.""" + mock_storage._store[f"quote:{QUOTE_ID}"] = _available_quote() + mock_storage._store[f"negotiation:{QUOTE_ID}"] = _accepted_negotiation( + proposal_id=QUOTE_ID, rounds=[] + ) + + with patch("ad_seller.storage.factory.get_storage", return_value=mock_storage): + resolved = await negotiation_service.resolve_negotiation_key( + proposal_id=PROPOSAL_ID, negotiation_id="neg-agreed01", quote_id=QUOTE_ID + ) + + assert resolved == QUOTE_ID + + async def test_a_negotiation_id_alone_is_not_a_storage_key(self, mock_storage): + """Known limitation, deliberately left to the structural fix: no + negotiation is stored under its own ``neg-`` id, so a message carrying + ONLY negotiation_id has nothing to resolve against. + + This is not a silent restart -- there is no proposal or quote under a + ``neg-`` id either, so the round is refused with a 404 rather than + quietly beginning again. Re-keying storage on negotiation_id is what + makes this case resolvable. + """ + mock_storage._store[f"quote:{QUOTE_ID}"] = _available_quote() + mock_storage._store[f"negotiation:{QUOTE_ID}"] = _accepted_negotiation( + proposal_id=QUOTE_ID, rounds=[] + ) + + with patch("ad_seller.storage.factory.get_storage", return_value=mock_storage): + resolved = await negotiation_service.resolve_negotiation_key( + negotiation_id="neg-agreed01" + ) + assert resolved == "neg-agreed01" + + with pytest.raises(HTTPException) as exc: + await negotiation_service.counter_proposal( + resolved, buyer_price=23.0, buyer_context=_buyer_context() + ) + + assert exc.value.status_code == 404 + + async def test_unresolvable_ids_mint_under_the_first_supplied_id(self, mock_storage): + """Nothing stored under any supplied id: the key is the first one + supplied, the previous or-chain order, so opens are unchanged.""" + with patch("ad_seller.storage.factory.get_storage", return_value=mock_storage): + resolved = await negotiation_service.resolve_negotiation_key( + proposal_id=PROPOSAL_ID, negotiation_id="neg-fresh01", quote_id=QUOTE_ID + ) + + assert resolved == PROPOSAL_ID + + async def test_no_ids_at_all_resolves_to_none(self, mock_storage): + """The router's 400 path: no negotiation context supplied.""" + with patch("ad_seller.storage.factory.get_storage", return_value=mock_storage): + assert await negotiation_service.resolve_negotiation_key() is None + + +# ============================================================================= +# Step 3: an accepted negotiation with no price must not book at list price +# ============================================================================= + + +class TestAcceptedWithoutPriceFailsLoudly: + async def test_accepted_negotiation_with_no_rounds_refuses_to_book(self, mock_storage): + """A negotiation that concluded accepted with no agreed price is a + corrupt record. Booking it at the un-negotiated quoted price is the + worst available outcome, so booking fails instead.""" + mock_storage._store[f"quote:{QUOTE_ID}"] = _available_quote() + mock_storage._store[f"negotiation:{PROPOSAL_ID}"] = _accepted_negotiation(rounds=[]) + mock_storage._store[f"negotiation_by_quote:{QUOTE_ID}"] = PROPOSAL_ID + + with patch("ad_seller.storage.factory.get_storage", return_value=mock_storage): + with pytest.raises(HTTPException) as exc: + await deal_service.book_deal(_booking_request()) + + assert exc.value.status_code == 500 + assert exc.value.detail["error"] == "negotiation_price_unresolved" + # And nothing was booked at the list price behind the buyer's back. + assert not [k for k in mock_storage._store if k.startswith("deal:")] + assert mock_storage._store[f"quote:{QUOTE_ID}"]["status"] == "available" + + async def test_accepted_negotiation_with_a_null_seller_price_refuses_to_book( + self, mock_storage + ): + """Same guard, reached via a final round whose price is missing rather + than via an empty rounds list.""" + neg = _accepted_negotiation( + rounds=[ + { + "round_number": 1, + "buyer_price": 22.0, + "seller_price": None, + "action": "accept", + "rationale": "corrupt", + "timestamp": datetime.utcnow().isoformat(), + } + ] + ) + mock_storage._store[f"quote:{QUOTE_ID}"] = _available_quote() + mock_storage._store[f"negotiation:{PROPOSAL_ID}"] = neg + mock_storage._store[f"negotiation_by_quote:{QUOTE_ID}"] = PROPOSAL_ID + + with patch("ad_seller.storage.factory.get_storage", return_value=mock_storage): + with pytest.raises(HTTPException) as exc: + await deal_service.book_deal(_booking_request()) + + assert exc.value.status_code == 500 + assert exc.value.detail["error"] == "negotiation_price_unresolved" From 0309acb1c2cf12bc07c8702973a988cf51d7b084 Mon Sep 17 00:00:00 2001 From: ATC964 <82515918+atc964@users.noreply.github.com> Date: Tue, 22 Sep 2026 13:37:18 -0400 Subject: [PATCH 2/2] fix: gate quote binding on quote ownership A negotiation message may name the quote its agreed price will apply to; an accepted negotiation is then indexed by negotiation_by_quote:{quote_id} and rewrites the booked final_cpm when that quote is booked. The quote_id was taken off the wire unchecked, so buyer A could negotiate its own proposal down, accept while naming buyer B's quote id, and B's booking would silently carry A's price. Binding a quote_id onto a negotiation (counter_proposal and apply_terminal_action, covering both the message's quote_id field and quote-led opens) now requires that the caller own the quote, using the same ownership record the booking route enforces (the quote-history buyer_id written from the verified pricing key at issue time): - An unowned quote reads as 404 quote_not_found, byte-identical to a nonexistent one, so the route is not an existence oracle over other tenants' quote ids (same pattern as the change-request routes). - A quote_id resolving to neither a quote nor its history record is refused the same way: it could never be booked, so binding it is a buyer bug or an attack, and dropping it silently would hide both. - The route's auth is optional and anonymous identity fields are self-asserted, so only a key-derived identity counts for ownership; every other caller is treated as public, fail-closed. - No recorded buyer (legacy) or a recorded 'public' buyer keeps today's behavior: nothing buyer-specific to enforce, binding allowed. Seven new tests, including the end-to-end control: the cross-tenant accept is refused, no pointer is written, and the victim's booking carries the quoted price. Verified by disabling the check and watching the five enforcement tests fail. --- .../interfaces/api/routers/negotiation.py | 6 +- src/ad_seller/services/negotiation_service.py | 79 ++++- .../test_negotiation_quote_correlation.py | 291 ++++++++++++++++++ 3 files changed, 374 insertions(+), 2 deletions(-) diff --git a/src/ad_seller/interfaces/api/routers/negotiation.py b/src/ad_seller/interfaces/api/routers/negotiation.py index 144e9621..971f5020 100644 --- a/src/ad_seller/interfaces/api/routers/negotiation.py +++ b/src/ad_seller/interfaces/api/routers/negotiation.py @@ -245,7 +245,11 @@ async def post_negotiation_message( # the agreed state, and an accept indexes the negotiation by its # quote so booking that quote can find the agreed price. status_data = await negotiation_service.apply_terminal_action( - proposal_id, message.action.value, buyer_price, quote_id=message.quote_id + proposal_id, + message.action.value, + buyer_price, + quote_id=message.quote_id, + buyer_context=buyer_context, ) response = cm.terminal_round_response(status_data, message.action, buyer_price) diff --git a/src/ad_seller/services/negotiation_service.py b/src/ad_seller/services/negotiation_service.py index 6389836a..839a9980 100644 --- a/src/ad_seller/services/negotiation_service.py +++ b/src/ad_seller/services/negotiation_service.py @@ -98,6 +98,73 @@ async def resolve_negotiation_key( return candidates[0] +def _quote_not_found_detail(quote_id: str) -> dict[str, str]: + """The exact detail shape ``quote_service.get_quote`` emits for a missing + quote — reused verbatim so an unowned quote is indistinguishable from a + nonexistent one (no existence oracle; house pattern, PR #98).""" + return {"error": "quote_not_found", "message": f"Quote '{quote_id}' not found."} + + +async def _verify_quote_binding(storage: Any, quote_id: str, buyer_context: Any) -> None: + """Gate binding a ``quote_id`` onto a negotiation on quote ownership. + + A negotiation message may name the quote its agreed price will apply to; + an accepted negotiation is then indexed by that quote and REWRITES the + booked ``final_cpm`` when the quote is booked. Without this check, buyer A + could accept its own negotiation while naming buyer B's quote id, and B's + booking would silently carry A's price. The caller must own the quote it + binds — the same ownership the booking route enforces before honouring a + quote (``routers/deals.py``). + + Rules, mirroring the booking route's, fail-closed on identity: + + - The quote must resolve at all (the live quote record or its history + entry). Binding an unresolvable id is refused loudly — it could never + be booked, so the binding is either a buyer bug or an attack, and + silently dropping it would hide both. + - Ownership comes from the quote-history record's ``buyer_id`` (written + at quote issue time from the verified pricing key). No record, or no + recorded buyer: legacy data, nothing to enforce — allowed, as at + booking. A ``"public"`` buyer asserted no identity at quote time, so + there is no ownership to enforce. + - The caller's pricing key counts only when it is key-derived + (``authentication_method == "api_key"``). This route's auth is + OPTIONAL, and an anonymous caller's identity fields are self-asserted + wire data — EP-5.2 floors the TIER but the pricing key would still + echo whatever ids the body claimed, so trusting it here would let an + anonymous caller impersonate the victim's identity and pass the very + check aimed at it. Every other caller is treated as public. + - A mismatch reads as 404 ``quote_not_found``, byte-identical to the + nonexistent case: a distinguishable "exists but not yours" would turn + this route into an existence oracle over every tenant's quote ids. + """ + from ..storage.quote_history import QuoteHistoryStore + + quote = await storage.get_quote(quote_id) + try: + history_record = await QuoteHistoryStore(storage).get_quote(quote_id) + except Exception: + history_record = None + if not isinstance(history_record, dict): + history_record = None + + if quote is None and history_record is None: + raise HTTPException(status_code=404, detail=_quote_not_found_detail(quote_id)) + + quoted_buyer_id = history_record.get("buyer_id") if history_record else None + if not quoted_buyer_id or quoted_buyer_id == "public": + return + + caller_key = ( + buyer_context.get_pricing_key() + if buyer_context is not None + and getattr(buyer_context, "authentication_method", None) == "api_key" + else "public" + ) + if quoted_buyer_id != caller_key: + raise HTTPException(status_code=404, detail=_quote_not_found_detail(quote_id)) + + async def submit_proposal( request: Any, buyer_context: Any, catalog: dict[str, Any] ) -> dict[str, Any]: @@ -333,8 +400,12 @@ async def counter_proposal( ) # Retain the quote this negotiation concerns. Backfills a negotiation - # opened before the quote was named; first one seen wins. + # opened before the quote was named; first one seen wins. Binding is + # gated on ownership: this id is what routes an agreed price onto a + # booked deal, so a caller may only bind a quote issued to it — whether + # the id arrived in the message's quote_id field or as a quote-led key. if quote_id and not history.quote_id: + await _verify_quote_binding(storage, quote_id, buyer_context) history.quote_id = quote_id # Evaluate buyer's offer @@ -389,6 +460,7 @@ async def apply_terminal_action( action: str, buyer_price: Optional[float] = None, quote_id: Optional[str] = None, + buyer_context: Optional[Any] = None, ) -> dict[str, Any]: """Record a buyer's terminal accept/reject on the stored negotiation. @@ -400,6 +472,10 @@ async def apply_terminal_action( ``quote_id`` backfills the quote this negotiation concerns when the accept is the first message to name it. On accept, this is the point the quote index is written, which is what lets booking find the agreed price. + Binding a NEW quote id is gated on quote ownership against + ``buyer_context`` (see ``_verify_quote_binding``); a ``None`` context is + treated as an anonymous (public) caller, so it fails closed against + buyer-specific quotes rather than skipping the check. No-ops (returning current status) when no stored negotiation exists or it is already terminal, preserving the read-only legacy behavior. @@ -413,6 +489,7 @@ async def apply_terminal_action( history = NegotiationHistory(**data) if history.status == "active" and action in ("accept", "reject"): if quote_id and not history.quote_id: + await _verify_quote_binding(storage, quote_id, buyer_context) history.quote_id = quote_id last = history.rounds[-1] if history.rounds else None seller_price = last.seller_price if last else history.base_price diff --git a/tests/unit/test_negotiation_quote_correlation.py b/tests/unit/test_negotiation_quote_correlation.py index a94f155c..41e7dd88 100644 --- a/tests/unit/test_negotiation_quote_correlation.py +++ b/tests/unit/test_negotiation_quote_correlation.py @@ -615,6 +615,297 @@ async def test_no_ids_at_all_resolves_to_none(self, mock_storage): assert await negotiation_service.resolve_negotiation_key() is None +# ============================================================================= +# Step 2c: only the quote's own buyer may bind it to a negotiation +# ============================================================================= + + +VICTIM_PRICING_KEY = "advertiser:adv-victim-9" + + +def _quote_history_record(quote_id=QUOTE_ID, buyer_id=VICTIM_PRICING_KEY): + """The record the quotes route writes at issue time (QuoteHistoryStore).""" + return { + "quote_id": quote_id, + "buyer_id": buyer_id, + "product_id": PRODUCT_ID, + "quoted_cpm": QUOTED_CPM, + "quoted_at": datetime.utcnow().isoformat(), + "expires_at": (datetime.utcnow() + timedelta(hours=23)).isoformat(), + } + + +def _api_key_buyer_context(seat="seat-test-1", agency="agency-test-1", adv="adv-test-1"): + """A KEY-derived context — the only identity quote ownership trusts.""" + from ad_seller.interfaces.api.deps import _build_buyer_context + + record = MagicMock() + record.identity = BuyerIdentity(seat_id=seat, agency_id=agency, advertiser_id=adv) + return _build_buyer_context(api_key_record=record) + + +class TestQuoteBindingOwnership: + """The quote a negotiation binds is the quote whose booked price it will + REWRITE, so binding is gated on ownership. An unowned quote reads as 404 + ``quote_not_found``, byte-identical to a nonexistent one (no existence + oracle — house pattern, PR #98).""" + + async def test_cross_tenant_accept_cannot_rewrite_the_victims_booked_price( + self, client, mock_storage + ): + """THE attack, end to end: buyer A negotiates its own proposal down, + accepts while naming buyer B's quote id, then B books its quote. The + accept must be refused, no quote pointer written, and B's deal must + book at B's quoted price -- not A's negotiated one.""" + _seed_negotiable(mock_storage) + # The quote belongs to the VICTIM, not to the caller the client + # fixture authenticates (advertiser:adv-test-1). + mock_storage._store[f"quote_history:{QUOTE_ID}"] = _quote_history_record() + + with ( + patch("ad_seller.storage.factory.get_storage", return_value=mock_storage), + patch("ad_seller.events.helpers.emit_event", new_callable=AsyncMock), + ): + async with client as c: + # A negotiates its OWN proposal -- no quote named, this is fine. + counter = await c.post( + "/api/v1/negotiations/messages", + json={ + "idempotency_key": "idem-xt-counter", + "action": "counter", + "proposal_id": PROPOSAL_ID, + "buyer_price": {"amount_micros": 22_000_000, "currency": "USD"}, + }, + ) + assert counter.status_code == 200, counter.text + + # The accept names B's quote: refused, and as NOT-FOUND, so a + # probe cannot distinguish "exists but not yours" from + # "does not exist" (no existence oracle). + accept = await c.post( + "/api/v1/negotiations/messages", + json={ + "idempotency_key": "idem-xt-accept", + "action": "accept", + "proposal_id": PROPOSAL_ID, + "quote_id": QUOTE_ID, + }, + ) + assert accept.status_code == 404, accept.text + assert accept.json()["detail"]["error"] == "quote_not_found" + + # Nothing bound, nothing indexed, negotiation NOT concluded. + assert mock_storage._store[f"negotiation:{PROPOSAL_ID}"]["quote_id"] is None + assert mock_storage._store[f"negotiation:{PROPOSAL_ID}"]["status"] == "active" + assert f"negotiation_by_quote:{QUOTE_ID}" not in mock_storage._store + + # B books its quote: full price, no negotiation suffix. + app.dependency_overrides[_get_optional_api_key_record] = lambda: MagicMock( + identity=BuyerIdentity( + seat_id="seat-victim-9", + agency_id="agency-victim-9", + advertiser_id="adv-victim-9", + ) + ) + victim = httpx.AsyncClient(transport=ASGITransport(app=app), base_url="http://test") + async with victim as v: + booked = await v.post( + "/api/v1/deals", + json={ + "idempotency_key": "idem-xt-book", + "quote_id": QUOTE_ID, + "buyer_identity": { + "seat_id": "seat-victim-9", + "agency_id": "agency-victim-9", + "advertiser_id": "adv-victim-9", + }, + }, + ) + + assert booked.status_code == 200, booked.text + pricing = booked.json()["deal"]["pricing"] + assert pricing["final_cpm"]["amount_micros"] == round(QUOTED_CPM * 1_000_000) + assert "Negotiated to" not in pricing["rationale"] + + async def test_own_quote_still_binds_and_books_the_negotiated_price(self, client, mock_storage): + """The legitimate flow with an ownership record present: the caller + negotiates naming ITS OWN quote, accepts, books -- the deal carries + the agreed price.""" + _seed_negotiable(mock_storage) + # The quote was issued to the very buyer the client authenticates. + mock_storage._store[f"quote_history:{QUOTE_ID}"] = _quote_history_record( + buyer_id="advertiser:adv-test-1" + ) + + with ( + patch("ad_seller.storage.factory.get_storage", return_value=mock_storage), + patch("ad_seller.events.helpers.emit_event", new_callable=AsyncMock), + ): + async with client as c: + counter = await c.post( + "/api/v1/negotiations/messages", + json={ + "idempotency_key": "idem-own-counter", + "action": "counter", + "proposal_id": PROPOSAL_ID, + "quote_id": QUOTE_ID, + "buyer_price": {"amount_micros": 22_000_000, "currency": "USD"}, + }, + ) + assert counter.status_code == 200, counter.text + + accept = await c.post( + "/api/v1/negotiations/messages", + json={ + "idempotency_key": "idem-own-accept", + "action": "accept", + "proposal_id": PROPOSAL_ID, + "quote_id": QUOTE_ID, + }, + ) + assert accept.status_code == 200, accept.text + assert accept.json()["status"] == "accepted" + + booked = await c.post( + "/api/v1/deals", + json={ + "idempotency_key": "idem-own-book", + "quote_id": QUOTE_ID, + "buyer_identity": { + "seat_id": "seat-test-1", + "agency_id": "agency-test-1", + "advertiser_id": "adv-test-1", + }, + }, + ) + + assert booked.status_code == 200, booked.text + agreed = mock_storage._store[f"negotiation:{PROPOSAL_ID}"]["rounds"][-1]["seller_price"] + assert agreed != QUOTED_CPM + pricing = booked.json()["deal"]["pricing"] + assert pricing["final_cpm"]["amount_micros"] == round(agreed * 1_000_000) + assert f"Negotiated to ${agreed:.2f} CPM" in pricing["rationale"] + + async def test_binding_an_unresolvable_quote_id_is_refused(self, client, mock_storage): + """A quote_id resolving to neither a quote nor a history record could + never be booked, so binding it is a buyer bug or an attack -- refused + loudly rather than silently dropped.""" + _seed_negotiable(mock_storage) + + with ( + patch("ad_seller.storage.factory.get_storage", return_value=mock_storage), + patch("ad_seller.events.helpers.emit_event", new_callable=AsyncMock), + ): + async with client as c: + counter = await c.post( + "/api/v1/negotiations/messages", + json={ + "idempotency_key": "idem-ghost-counter", + "action": "counter", + "proposal_id": PROPOSAL_ID, + "quote_id": "qt-nosuchquote", + "buyer_price": {"amount_micros": 22_000_000, "currency": "USD"}, + }, + ) + + assert counter.status_code == 404, counter.text + assert counter.json()["detail"]["error"] == "quote_not_found" + # The refused round left nothing behind. + assert f"negotiation:{PROPOSAL_ID}" not in mock_storage._store + + async def test_quote_led_open_of_a_foreign_quote_is_refused(self, mock_storage): + """The sibling vector: the same foreign quote id sent as the + negotiation KEY (quote-led open) rather than in the quote_id field + binds through the same site and is refused the same way.""" + mock_storage._store[f"quote:{QUOTE_ID}"] = _available_quote() + mock_storage._store[f"product:{PRODUCT_ID}"] = _stored_product() + mock_storage._store[f"quote_history:{QUOTE_ID}"] = _quote_history_record() + + with ( + patch("ad_seller.storage.factory.get_storage", return_value=mock_storage), + patch("ad_seller.events.helpers.emit_event", new_callable=AsyncMock), + ): + with pytest.raises(HTTPException) as exc: + await negotiation_service.counter_proposal( + QUOTE_ID, buyer_price=22.0, buyer_context=_api_key_buyer_context() + ) + + assert exc.value.status_code == 404 + assert exc.value.detail["error"] == "quote_not_found" + assert f"negotiation:{QUOTE_ID}" not in mock_storage._store + + async def test_anonymous_self_asserted_identity_cannot_bind_a_buyer_quote(self, mock_storage): + """This route's auth is OPTIONAL and an anonymous caller's identity + fields are self-asserted wire data. Asserting the victim's ids must + not pass the ownership check aimed at exactly that caller: only a + KEY-derived identity counts, everyone else is public.""" + _seed_negotiable(mock_storage) + mock_storage._store[f"quote_history:{QUOTE_ID}"] = _quote_history_record() + + from ad_seller.interfaces.api.deps import _build_buyer_context + + impostor = _build_buyer_context( + buyer_tier="advertiser", agency_id="agency-victim-9", advertiser_id="adv-victim-9" + ) + assert impostor.get_pricing_key() == VICTIM_PRICING_KEY # the claim... + + with ( + patch("ad_seller.storage.factory.get_storage", return_value=mock_storage), + patch("ad_seller.events.helpers.emit_event", new_callable=AsyncMock), + ): + with pytest.raises(HTTPException) as exc: # ...does not bind. + await negotiation_service.counter_proposal( + PROPOSAL_ID, buyer_price=22.0, buyer_context=impostor, quote_id=QUOTE_ID + ) + + assert exc.value.status_code == 404 + assert exc.value.detail["error"] == "quote_not_found" + + async def test_a_public_quote_stays_negotiable_without_ownership(self, mock_storage): + """A quote issued to ``buyer_id="public"`` asserted no identity at + quote time -- nothing buyer-specific was priced, so there is no + ownership to enforce (same rule as booking).""" + _seed_negotiable(mock_storage) + mock_storage._store[f"quote_history:{QUOTE_ID}"] = _quote_history_record(buyer_id="public") + + with ( + patch("ad_seller.storage.factory.get_storage", return_value=mock_storage), + patch("ad_seller.events.helpers.emit_event", new_callable=AsyncMock), + ): + result = await negotiation_service.counter_proposal( + PROPOSAL_ID, + buyer_price=22.0, + buyer_context=_buyer_context(), + quote_id=QUOTE_ID, + ) + + assert result["round_number"] == 1 + assert mock_storage._store[f"negotiation:{PROPOSAL_ID}"]["quote_id"] == QUOTE_ID + + async def test_terminal_bind_without_a_context_fails_closed(self, mock_storage): + """``apply_terminal_action`` called with a quote to bind but no buyer + context treats the caller as public: a buyer-specific quote does not + bind, rather than the check being skipped.""" + _seed_negotiable(mock_storage) + mock_storage._store[f"quote_history:{QUOTE_ID}"] = _quote_history_record() + + with ( + patch("ad_seller.storage.factory.get_storage", return_value=mock_storage), + patch("ad_seller.events.helpers.emit_event", new_callable=AsyncMock), + ): + await negotiation_service.counter_proposal( + PROPOSAL_ID, buyer_price=22.0, buyer_context=_buyer_context() + ) + with pytest.raises(HTTPException) as exc: + await negotiation_service.apply_terminal_action( + PROPOSAL_ID, "accept", quote_id=QUOTE_ID + ) + + assert exc.value.status_code == 404 + assert mock_storage._store[f"negotiation:{PROPOSAL_ID}"]["status"] == "active" + assert f"negotiation_by_quote:{QUOTE_ID}" not in mock_storage._store + + # ============================================================================= # Step 3: an accepted negotiation with no price must not book at list price # =============================================================================