diff --git a/purchase_executor.py b/purchase_executor.py index 82c65a130abc..71167b2947c8 100644 --- a/purchase_executor.py +++ b/purchase_executor.py @@ -48,7 +48,7 @@ from pathlib import Path from types import SimpleNamespace from typing import cast -from urllib.parse import urlsplit, urlunsplit +from urllib.parse import quote, urlsplit, urlunsplit import purchase_discovery as discovery import purchase_merchants @@ -132,6 +132,7 @@ def redact_payment_data(value, payment_values: dict): " const m = document.querySelector('#merchant, [data-merchant]');" " return JSON.stringify({url: location.href, text: t.slice(0, 20000)," " merchant: (m && m.textContent || document.title || '').trim()," + " prepared_session: document.documentElement.dataset.hermesPreparedSession === 'authenticated'," " filled: document.documentElement.dataset.hermesPurchaseFilled === '1'}); })()" ) # Set after a successful genuine-input fill; the post-fill re-probe reads this @@ -155,7 +156,6 @@ def redact_payment_data(value, payment_values: dict): r"(?i)subscription|auto[- ]?renew|recurring|billed (?:monthly|annually|yearly)" ) AUTO_RENEW_RE = re.compile(r"(?i)auto(?:matically)?[- ]?renew") -TOTAL_RE = re.compile(r"(?i)total\D{0,10}(\d+\.\d{2})") # --- control-flow signals ---------------------------------------------------- @@ -666,24 +666,120 @@ def page_origin_allowed(page: dict, canonical_domain: str, *, fake_e2e: bool) -> return origin_allowed(page.get("url", ""), canonical_domain, fake_e2e=fake_e2e) +def _checkout_contract(claim: dict, adapter) -> tuple[dict, dict]: + target = claim.get("checkout_target") + terms = claim.get("checkout_terms") + target_keys = {"merchant_id", "product_kind", "product_id", "session_requirement"} + term_keys = { + "product_or_service", "quantity", "quoted_subtotal", "tax", + "mandatory_fees", "final_quoted_total", "currency", + "recurrence_authorization", + } + recurrence_keys = { + "commitment_type", "billing_interval", "renewal_amount", "renewal_date", + "cancellation_deadline", "contract_duration", "cancellation_terms", "auto_renew", + } + if not isinstance(target, dict) or set(target) != target_keys: + raise _Stop("definitive", "checkout_not_ready") + if not isinstance(terms, dict) or set(terms) != term_keys: + raise _Stop("definitive", "checkout_not_ready") + recurrence = terms.get("recurrence_authorization") + if not isinstance(recurrence, dict) or set(recurrence) != recurrence_keys: + raise _Stop("definitive", "checkout_not_ready") + canonical = str(claim.get("canonical_merchant_domain") or "").lower() + if target["merchant_id"] != canonical or ( + adapter is not purchase_merchants.MOCK and adapter.canonical_domain != canonical + ): + raise _Stop("definitive", "checkout_not_ready") + if target["session_requirement"] not in {"anonymous", "authenticated"}: + raise _Stop("definitive", "checkout_not_ready") + product_id = target["product_id"] + if not isinstance(product_id, str) or not product_id or len(product_id) > 160: + raise _Stop("definitive", "checkout_not_ready") + if target["product_kind"] == "domain_registration": + if not re.fullmatch( + r"(?=.{1,253}$)(?:[a-z0-9](?:[a-z0-9-]{0,61}[a-z0-9])?\.)+[a-z]{2,63}", + product_id, + ) or product_id not in str(terms["product_or_service"]).lower().split(): + raise _Stop("definitive", "checkout_not_ready") + elif target["product_kind"] == "merchant_sku": + if not re.fullmatch(r"[A-Za-z0-9][A-Za-z0-9._:-]{0,159}", product_id): + raise _Stop("definitive", "checkout_not_ready") + else: + raise _Stop("definitive", "checkout_not_ready") + quantity = terms["quantity"] + if isinstance(quantity, bool) or not isinstance(quantity, int) or quantity < 1: + raise _Stop("definitive", "checkout_not_ready") + money_fields = ("quoted_subtotal", "tax", "mandatory_fees", "final_quoted_total") + if any(not re.fullmatch(r"\d{1,10}\.\d{2}", str(terms.get(key) or "")) for key in money_fields): + raise _Stop("definitive", "checkout_not_ready") + cents = { + key: int(str(terms[key]).replace(".", "")) + for key in money_fields + } + if cents["quoted_subtotal"] + cents["tax"] + cents["mandatory_fees"] != cents["final_quoted_total"]: + raise _Stop("definitive", "checkout_not_ready") + if ( + terms["product_or_service"] != claim.get("approved_item") + or quantity != claim.get("quantity") + or terms["final_quoted_total"] != claim.get("maximum_total") + or terms["currency"] != claim.get("currency") + or recurrence != claim.get("recurrence_authorization") + ): + raise _Stop("definitive", "checkout_not_ready") + return target, terms + + +def _checkout_url(base_url: str, cart_path: str) -> str: + base = urlsplit(base_url) + relative = urlsplit(cart_path) + if not base.scheme or not base.netloc or relative.scheme or relative.netloc or not relative.path.startswith("/"): + return "" + return urlunsplit((base.scheme, base.netloc, relative.path, relative.query, "")) + + def check_terms(page_text: str, claim: dict) -> list[str]: problems = [] - approved_total = claim["maximum_total"] - totals = TOTAL_RE.findall(page_text) - if not totals: - problems.append("total_missing") - elif any(total != approved_total for total in totals): - problems.append("total_mismatch") - if claim["currency"] not in page_text: + terms = claim["checkout_terms"] + currency = terms["currency"] + if currency not in page_text: problems.append("currency_missing") - if claim["approved_item"].lower() not in page_text.lower(): + labels = { + "subtotal": (r"(?:quoted\s+)?subtotal", terms["quoted_subtotal"]), + "tax": (r"tax", terms["tax"]), + "mandatory_fees": (r"(?:mandatory\s+)?fees", terms["mandatory_fees"]), + "total": (r"(?:final\s+|order\s+)?total", terms["final_quoted_total"]), + } + for name, (label, expected) in labels.items(): + matches = re.findall( + rf"(?im)^\s*(?:{label})\s*:\s*(\d{{1,10}}\.\d{{2}})\s+([A-Z]{{3}})\s*$", + page_text, + ) + if not matches: + problems.append(f"{name}_missing") + elif len(matches) != 1: + problems.append(f"{name}_ambiguous") + else: + amount, found_currency = matches[0] + if amount != expected: + problems.append(f"{name}_mismatch") + if found_currency != currency and "currency_mismatch" not in problems: + problems.append("currency_mismatch") + item_matches = re.findall( + rf"(?im)^\s*{re.escape(terms['product_or_service'])}\s*$", page_text + ) + if not item_matches: problems.append("item_missing") - quantity_match = re.search(r"(?i)quantity\D{0,5}(\d+)", page_text) - if not quantity_match: + elif len(item_matches) != 1: + problems.append("item_ambiguous") + quantities = re.findall(r"(?im)^\s*quantity\s*:\s*(\d+)\s*$", page_text) + if not quantities: problems.append("quantity_missing") - elif int(quantity_match.group(1)) != int(claim["quantity"]): + elif len(quantities) != 1: + problems.append("quantity_ambiguous") + elif int(quantities[0]) != int(terms["quantity"]): problems.append("quantity_mismatch") - recurrence = claim["recurrence_authorization"] + recurrence = terms["recurrence_authorization"] is_recurring_page = bool(RECURRING_RE.search(page_text)) if recurrence["commitment_type"] == "one_time" and is_recurring_page: problems.append("unexpected_recurrence") @@ -860,6 +956,7 @@ def run_once( post_submit_wait: float = POST_SUBMIT_WAIT_SECONDS, sleep=time.sleep, fake_processor_origins: tuple[str, ...] = (), + adapter_for=purchase_merchants.adapter_for, ) -> int: """One attempt: claim → checkout → exactly one terminal report. No retry.""" intent = "Restricted Purchase Executor V0 run for one claimed execution ticket." @@ -933,40 +1030,84 @@ def terminal(action: str, context: dict) -> None: if termination["pending"]: raise _Terminated() audit("claimed", ticket_id=ticket_id, proposal_id=claim["proposal_id"], task_id=task_id) - # Fail closed on an unsupported merchant BEFORE opening credential - # files, filling, or submitting. Production allows Porkbun only; - # fake/staging uses the loopback mock adapter. - adapter = purchase_merchants.adapter_for( - claim["canonical_merchant_domain"], fake_e2e=fake_e2e - ) + # Fail closed on an unsupported merchant and malformed ticket contract + # before navigation, credential access, fill, or submit. + adapter = adapter_for(claim["canonical_merchant_domain"], fake_e2e=fake_e2e) if adapter is None: audit("merchant_rejected", domain=strip_query(claim["canonical_merchant_domain"])) raise _Stop("definitive", "merchant_not_supported") - checkout_url = checkout_url_for(claim) + target, _ = _checkout_contract(claim, adapter) + path = purchase_merchants.cart_path( + adapter, target["product_kind"], target["product_id"], claim["quantity"] + ) + if not path: + raise _Stop("definitive", "checkout_not_ready") + base_url = checkout_url_for(claim) + checkout_url = _checkout_url(base_url, path) if not origin_allowed( checkout_url, claim["canonical_merchant_domain"], fake_e2e=fake_e2e ): raise _Stop("definitive", "wrong_origin") - navigation = browser.navigate(checkout_url, task_id) + + authenticated = target["session_requirement"] == "authenticated" + navigation_url = checkout_url + if authenticated: + handoff_path = adapter.fake_session_handoff_path if ( + fake_e2e and adapter is purchase_merchants.MOCK + ) else "" + if not handoff_path: + raise _Stop("definitive", "login_required") + navigation_url = _checkout_url( + base_url, f"{handoff_path}?return={quote(path, safe='')}" + ) + if not origin_allowed( + navigation_url, claim["canonical_merchant_domain"], fake_e2e=fake_e2e + ): + raise _Stop("definitive", "wrong_origin") + + navigation = browser.navigate(navigation_url, task_id) + if authenticated: + audit("session_handoff", success=bool(navigation.get("success"))) audit("navigated", url=strip_query(checkout_url), success=bool(navigation.get("success"))) if not navigation.get("success"): raise _Stop("definitive", "navigation_failed") - probe = browser.eval_js(PAGE_PROBE_JS, task_id) - if not probe.get("success"): - raise _Stop("definitive", "page_read_failed") - page = probe["result"] - if not page_origin_allowed( - page, claim["canonical_merchant_domain"], fake_e2e=fake_e2e - ): - audit("origin_rejected", url=strip_query(page.get("url", ""))) - raise _Stop("definitive", "wrong_origin") - page_text = page.get("text", "") - if CHALLENGE_RE.search(page_text): - raise _Stop("definitive", "human_challenge_required") - if page.get("filled"): - raise _Stop("definitive", "invalid_checkout_state") + def inspect_precredential(gate: str): + probe = browser.eval_js(PAGE_PROBE_JS, task_id) + if not probe.get("success"): + raise _Stop("definitive", "page_read_failed") + page = probe["result"] + if not page_origin_allowed( + page, claim["canonical_merchant_domain"], fake_e2e=fake_e2e + ): + audit("origin_rejected", url=strip_query(page.get("url", ""))) + raise _Stop("definitive", "wrong_origin") + page_text = page.get("text", "") + if CHALLENGE_RE.search(page_text): + raise _Stop("definitive", "human_challenge_required") + if page.get("filled"): + raise _Stop("definitive", "invalid_checkout_state") + if authenticated and not ( + fake_e2e + and adapter is purchase_merchants.MOCK + and page.get("prepared_session") is True + ): + raise _Stop("definitive", "login_required") + mismatches = check_terms(page_text, claim) + if mismatches: + audit("terms_rejected", gate=gate, mismatches=mismatches) + raise _Stop("definitive", "terms_changed") + audit( + "checkout_validated", gate=gate, + total=claim["maximum_total"], currency=claim["currency"], + ) + return page + + # The prepared cart, session, and exact commercial breakdown pass + # twice before semantic payment discovery begins. + page = inspect_precredential("bootstrap") merchant_name = page.get("merchant", "") + inspect_precredential("prefill") def discover_checkout(bind_controls=True): try: @@ -984,26 +1125,10 @@ def discover_checkout(bind_controls=True): plan = discover_checkout() audit("discovered", **plan.audit()) - mismatches = check_terms(page_text, claim) - if mismatches: - audit("terms_rejected", mismatches=mismatches) - raise _Stop("definitive", "terms_changed") - audit("revalidated", total=claim["maximum_total"], currency=claim["currency"]) - # A second terms + discovery pass closes the pre-credential TOCTOU - # window. No payment file has been opened before both gates pass. - prefill = browser.eval_js(PAGE_PROBE_JS, task_id) - if not prefill.get("success"): - raise _Stop("definitive", "page_read_failed") - prefill_page = prefill["result"] - if not page_origin_allowed( - prefill_page, claim["canonical_merchant_domain"], fake_e2e=fake_e2e - ): - raise _Stop("definitive", "wrong_origin") - if CHALLENGE_RE.search(prefill_page.get("text", "")): - raise _Stop("definitive", "human_challenge_required") - if check_terms(prefill_page.get("text", ""), claim): - raise _Stop("definitive", "terms_changed") + # Preserve V0.2's second discovery fingerprint and terms gate before + # opening any payment credential file. + inspect_precredential("discovery_confirmation") confirmed_plan = discover_checkout() if confirmed_plan.fingerprint != plan.fingerprint: raise _Stop("definitive", "checkout_not_ready") @@ -1029,6 +1154,12 @@ def discover_checkout(bind_controls=True): recheck_text = recheck_page.get("text", "") if CHALLENGE_RE.search(recheck_text): raise _Stop("definitive", "human_challenge_required") + if authenticated and not ( + fake_e2e + and adapter is purchase_merchants.MOCK + and recheck_page.get("prepared_session") is True + ): + raise _Stop("definitive", "login_required") if check_terms(recheck_text, claim): raise _Stop("definitive", "terms_changed") final_plan = discover_checkout(bind_controls=False) @@ -1124,8 +1255,6 @@ def parse_args(argv=None) -> argparse.Namespace: help="loopback-only acceptance mode; refuses any non-loopback URL") parser.add_argument("--checkout-url", default="", help="explicit checkout URL; only valid with --fake-e2e") - parser.add_argument("--checkout-path", default="/checkout", - help="path appended to https://") parser.add_argument("--fake-processor-origin", action="append", default=[], help="exact loopback hosted-field origin; fake-E2E only") parser.add_argument("--state-dir", default=DEFAULT_STATE_DIR, @@ -1163,7 +1292,7 @@ def main(argv=None) -> int: def checkout_url_for(claim: dict) -> str: if args.fake_e2e: return args.checkout_url - return f"https://{claim['canonical_merchant_domain']}{args.checkout_path}" + return f"https://{claim['canonical_merchant_domain']}" def execute() -> int: return run_once( diff --git a/purchase_merchants.py b/purchase_merchants.py index 5503eb7d6781..01ea5ab9d5d6 100644 --- a/purchase_merchants.py +++ b/purchase_merchants.py @@ -19,6 +19,7 @@ from __future__ import annotations from dataclasses import dataclass +from urllib.parse import quote from purchase_discovery import FIELD_NAMES as CARD_FIELDS @@ -28,6 +29,8 @@ class MerchantAdapter: key: str canonical_domain: str checkout_paths: tuple[str, ...] + cart_paths: dict[str, str] | None = None + fake_session_handoff_path: str = "" processor_origins: tuple[str, ...] = () field_hints: dict[str, tuple[str, ...]] | None = None submit_hints: tuple[str, ...] = () @@ -39,6 +42,7 @@ class MerchantAdapter: key="porkbun", canonical_domain="porkbun.com", checkout_paths=("/checkout", "/cart"), + cart_paths={}, processor_origins=(), field_hints={}, fixture="tests/fixtures/porkbun_checkout_v0.html (sanitized 2026-07-19; " @@ -49,7 +53,12 @@ class MerchantAdapter: MOCK = MerchantAdapter( key="mock", canonical_domain="mock.local", - checkout_paths=("/checkout", "/"), + checkout_paths=("/checkout", "/cart", "/"), + cart_paths={ + "domain_registration": "/checkout?product_kind=domain_registration&product_id={product_id}&quantity={quantity}", + "merchant_sku": "/checkout?product_kind=merchant_sku&product_id={product_id}&quantity={quantity}", + }, + fake_session_handoff_path="/__hermes_session_handoff", field_hints={}, fixture="in-repo mock merchant (fake-E2E only)", ) @@ -57,6 +66,14 @@ class MerchantAdapter: _LIVE_ALLOWLIST = {PORKBUN.canonical_domain: PORKBUN} +def cart_path(adapter: MerchantAdapter, product_kind: str, product_id: str, quantity: int) -> str: + """Build the exact adapter-owned cart path, or fail closed with ``""``.""" + template = (adapter.cart_paths or {}).get(product_kind) + if not template or not product_id or isinstance(quantity, bool) or quantity < 1: + return "" + return template.format(product_id=quote(product_id, safe=""), quantity=quantity) + + def adapter_for(canonical_domain: str, *, fake_e2e: bool) -> MerchantAdapter | None: """Resolve the adapter for a claimed merchant, or ``None`` if unsupported. @@ -75,6 +92,10 @@ def _demo() -> None: assert adapter_for("namecheap.com", fake_e2e=False) is None assert adapter_for("anything", fake_e2e=True) is MOCK assert not PORKBUN.processor_origins + assert not cart_path(PORKBUN, "domain_registration", "example.com", 1) + assert cart_path(MOCK, "domain_registration", "example.com", 2) == ( + "/checkout?product_kind=domain_registration&product_id=example.com&quantity=2" + ) assert set((PORKBUN.field_hints or {})).issubset(CARD_FIELDS) print("purchase_merchants demo ok") diff --git a/scripts/purchase_executor_fake_e2e.py b/scripts/purchase_executor_fake_e2e.py index 723406fad7e4..4ab68e89cef8 100644 --- a/scripts/purchase_executor_fake_e2e.py +++ b/scripts/purchase_executor_fake_e2e.py @@ -1,4 +1,4 @@ -"""Fake end-to-end acceptance for semantic purchase discovery (issues #65/#73). +"""Fake end-to-end acceptance for semantic purchase discovery (issues #65/#73/#75). Everything is local and synthetic: * temporary SQLite governance DB seeded with one approved proposal + ticket, @@ -32,6 +32,7 @@ import time import urllib.request import urllib.error +import urllib.parse from datetime import datetime, timedelta, timezone from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer from pathlib import Path @@ -53,21 +54,51 @@ class MockMerchant(BaseHTTPRequestHandler): checkout_html = b"" + authenticated_session = False + requests = [] def log_message(self, *args): pass - def _send(self, body: bytes): + def _send(self, body: bytes, headers=()): self.send_response(200) self.send_header("Content-Type", "text/html") self.send_header("Content-Length", str(len(body))) + for name, value in headers: + self.send_header(name, value) self.end_headers() self.wfile.write(body) def do_GET(self): - self._send(self.checkout_html if self.path.startswith("/checkout") else b"404") + request = urllib.parse.urlsplit(self.path) + self.requests.append(("GET", request.path)) + if request.path == "/__hermes_session_handoff" and self.authenticated_session: + destination = urllib.parse.parse_qs(request.query).get("return", [""])[0] + if destination == "/checkout?product_kind=domain_registration&product_id=example.com&quantity=1": + self.send_response(302) + self.send_header("Location", destination) + self.send_header( + "Set-Cookie", + "hermes_mock_session=ready; Path=/; HttpOnly; SameSite=Strict", + ) + self.send_header("Content-Length", "0") + self.end_headers() + return + if self.path == "/checkout?product_kind=domain_registration&product_id=example.com&quantity=1": + body = self.checkout_html + cookie = self.headers.get("Cookie", "") + if self.authenticated_session and "hermes_mock_session=ready" in cookie: + body = body.replace( + b'', + b'', + 1, + ) + self._send(body) + return + self._send(b"404") def do_POST(self): + self.requests.append(("POST", urllib.parse.urlsplit(self.path).path)) self.rfile.read(int(self.headers.get("Content-Length") or 0)) self._send(CONFIRM_HTML) @@ -130,6 +161,10 @@ def main() -> int: "--hosted-frame", action="store_true", help="serve card fields from a second exact-allowlisted loopback origin", ) + parser.add_argument( + "--authenticated-session", action="store_true", + help="prove a synthetic cookie-backed prepared session in the same browser task", + ) parser.add_argument( "--credentials-dir", default="", help="use externally-provided synthetic credentials (e.g. the staging " @@ -187,6 +222,12 @@ def main() -> int: "free_alternatives_considered": False, "necessary_to_launch": True, "optional_convenience": False, + "checkout_target": { + "merchant_id": "registrar.example", + "product_kind": "domain_registration", + "product_id": "example.com", + "session_requirement": "authenticated" if args.authenticated_session else "anonymous", + }, } created = governance.create_purchase_proposal(db, proposal_payload) proposal_id = created["proposal_id"] @@ -231,9 +272,10 @@ async def execute(packet, update=None, context=None): else: MockMerchant.checkout_html = SAME_PAGE_FIXTURE.read_bytes() + MockMerchant.authenticated_session = args.authenticated_session merchant_server = ThreadingHTTPServer(("127.0.0.1", 0), MockMerchant) threading.Thread(target=merchant_server.serve_forever, daemon=True).start() - checkout_url = f"http://127.0.0.1:{merchant_server.server_address[1]}/checkout" + checkout_url = f"http://127.0.0.1:{merchant_server.server_address[1]}" # --- synthetic credentials ---------------------------------------------- # Either use externally-staged synthetic creds (systemd name-binding) or @@ -286,7 +328,10 @@ async def execute(packet, update=None, context=None): ) print(run.stdout, end="") print(run.stderr, end="", file=sys.stderr) - require(run.returncode == 0, f"executor exit code {run.returncode} (want 0=completed)") + require( + run.returncode == 0, + f"executor exit code {run.returncode} (want 0=completed); requests={MockMerchant.requests}", + ) # --- proofs -------------------------------------------------------------- ledger_after = governance.get_ledger_snapshot(db) @@ -321,6 +366,7 @@ async def execute(packet, update=None, context=None): require("4242424242424242" not in audit, "audit log contains no card number") require('"phase": "discovered"' in audit, "semantic discovery audited") require('"phase": "cleaned_up"' in audit, "browser cleanup audited") + require("hermes_mock_session" not in audit, "session cookie absent from audit") if args.hosted_frame: require(processor_origin in audit, "hosted processor origin audited") else: @@ -333,6 +379,7 @@ async def execute(packet, update=None, context=None): print(json.dumps({ "fake_e2e": "PASS", "checkout_shape": "hosted_frame" if args.hosted_frame else "same_page", + "session_shape": "authenticated" if args.authenticated_session else "anonymous", "proposal_id": proposal_id, "ticket_id": ticket["ticket_id"], "ledger_before": ledger_before, "ledger_after": ledger_after, diff --git a/tests/fixtures/payment_hosted_parent_v0.html b/tests/fixtures/payment_hosted_parent_v0.html index a197e1cddea7..408505040028 100644 --- a/tests/fixtures/payment_hosted_parent_v0.html +++ b/tests/fixtures/payment_hosted_parent_v0.html @@ -6,9 +6,12 @@

Fake Registrar

example.com domain registration

Quantity: 1

+

Subtotal: 20.00 AUD

+

Tax: 2.00 AUD

+

Mandatory fees: 0.00 AUD

Total: 22.00 AUD

+ src="{{PROCESSOR_ORIGIN}}/fields" style="width: 500px; height: 180px">
diff --git a/tests/fixtures/payment_same_page_v0.html b/tests/fixtures/payment_same_page_v0.html index db08df73dc14..01cc3e4e2e6a 100644 --- a/tests/fixtures/payment_same_page_v0.html +++ b/tests/fixtures/payment_same_page_v0.html @@ -6,6 +6,9 @@

Fake Registrar

example.com domain registration

Quantity: 1

+

Subtotal: 20.00 AUD

+

Tax: 2.00 AUD

+

Mandatory fees: 0.00 AUD

Total: 22.00 AUD

diff --git a/tests/test_purchase_executor.py b/tests/test_purchase_executor.py index a67e13feb8a2..a997f168b2c1 100644 --- a/tests/test_purchase_executor.py +++ b/tests/test_purchase_executor.py @@ -8,6 +8,7 @@ import json import os import signal +from dataclasses import replace import stat import subprocess import urllib.error @@ -48,10 +49,27 @@ "cancellation_terms": "No recurring commitment authorized.", "auto_renew": False, }, + "checkout_target": { + "merchant_id": MERCHANT, + "product_kind": "domain_registration", + "product_id": "example.com", + "session_requirement": "anonymous", + }, +} +CLAIM["checkout_terms"] = { + "product_or_service": CLAIM["approved_item"], + "quantity": CLAIM["quantity"], + "quoted_subtotal": "20.00", + "tax": "2.00", + "mandatory_fees": "0.00", + "final_quoted_total": CLAIM["maximum_total"], + "currency": CLAIM["currency"], + "recurrence_authorization": CLAIM["recurrence_authorization"], } CHECKOUT_TEXT = ( "Fake Registrar\nexample.com domain registration\nQuantity: 1\n" + "Subtotal: 20.00 AUD\nTax: 2.00 AUD\nMandatory fees: 0.00 AUD\n" "Total: 22.00 AUD\nPay now" ) CHECKOUT_PAGE = { @@ -60,6 +78,7 @@ "merchant": "Fake Registrar", "has_form": True, "form_action": f"https://{MERCHANT}/pay", + "prepared_session": False, } CONFIRM_PAGE = { "url": f"https://{MERCHANT}/pay", @@ -107,6 +126,19 @@ def _match(field, locator_kind, locator_value, *, confidence=100): ) +TEST_ADAPTER = replace( + purchase_merchants.PORKBUN, + cart_paths={ + "domain_registration": "/checkout?product_kind=domain_registration&product_id={product_id}&quantity={quantity}", + "merchant_sku": "/checkout?product_kind=merchant_sku&product_id={product_id}&quantity={quantity}", + }, +) + + +def _adapter_for_test(domain, *, fake_e2e): + return TEST_ADAPTER if domain == MERCHANT else None + + class FakeBrowser: def __init__(self, pages=None, nav_success=True, fill_ok=True, submit_ok=True, on_eval=None, plan=DISCOVERY_PLAN, discovery_error=None, @@ -212,9 +244,10 @@ def run(browser, bridge, tmp_path, *, fake_e2e=False, token="tok\n"): browser=browser, audit=pe.audit_factory(state), state_dir=state, - checkout_url_for=lambda claim: "https://porkbun.com/checkout", + checkout_url_for=lambda claim: "https://porkbun.com", fake_e2e=fake_e2e, stdin=io.StringIO(token), + adapter_for=_adapter_for_test, post_submit_wait=0.2, sleep=lambda seconds: None, ) @@ -246,6 +279,9 @@ def test_happy_path_completes_once(tmp_path, creds): assert receipt_path.is_file() assert stat.S_IMODE(receipt_path.stat().st_mode) == 0o600 assert browser.cleaned == ["purchase_pt_test"] + assert ("navigate", "https://porkbun.com/checkout?product_kind=domain_registration&product_id=example.com&quantity=1") in browser.calls + first_discovery = next(index for index, call in enumerate(browser.calls) if call[0] == "discover") + assert [kind for kind, _ in browser.calls[:first_discovery]].count("probe") == 2 def test_claim_rejected_touches_nothing(tmp_path, creds): @@ -278,6 +314,156 @@ def test_price_mismatch_aborts_without_submit(tmp_path, creds): assert "submit" not in [kind for kind, _ in browser.calls] +@pytest.mark.parametrize( + "case", + [ + "missing_target", "extra_target", "merchant_mismatch", "product_mismatch", + "domain_prefix", "quantity_mismatch", "total_mismatch", + "inconsistent_components", "recurrence_mismatch", + ], +) +def test_invalid_checkout_contract_stops_before_navigation_or_credentials( + tmp_path, creds, monkeypatch, case, +): + claim = json.loads(json.dumps(CLAIM)) + if case == "missing_target": + claim.pop("checkout_target") + elif case == "extra_target": + claim["checkout_target"]["unexpected"] = "x" + elif case == "merchant_mismatch": + claim["checkout_target"]["merchant_id"] = "evil.example" + elif case == "product_mismatch": + claim["checkout_target"]["product_id"] = "other.example" + elif case == "domain_prefix": + claim["approved_item"] = "notexample.com domain registration" + claim["checkout_terms"]["product_or_service"] = claim["approved_item"] + elif case == "quantity_mismatch": + claim["checkout_terms"]["quantity"] = 2 + elif case == "total_mismatch": + claim["checkout_terms"]["final_quoted_total"] = "21.00" + elif case == "inconsistent_components": + claim["checkout_terms"]["tax"] = "3.00" + else: + claim["checkout_terms"]["recurrence_authorization"]["auto_renew"] = True + + monkeypatch.setattr( + pe, "load_payment_fields", + lambda: (_ for _ in ()).throw(AssertionError("credentials opened")), + ) + browser, bridge = FakeBrowser(), FakeBridge(claim=claim) + assert run(browser, bridge, tmp_path) == pe.EXIT_DEFINITIVE_FAILURE + assert terminal_calls(bridge)[0][1]["failure_category"] == "checkout_not_ready" + assert browser.calls == [] + assert browser.cleaned == ["purchase_pt_test"] + + +def test_sku_contract_matches_cogitator_and_cart_path_is_encoded(): + claim = json.loads(json.dumps(CLAIM)) + claim["approved_item"] = "Basic hosting" + claim["checkout_terms"]["product_or_service"] = claim["approved_item"] + claim["checkout_target"].update(product_kind="merchant_sku", product_id="plan:basic") + + target, _ = pe._checkout_contract(claim, TEST_ADAPTER) + assert purchase_merchants.cart_path( + TEST_ADAPTER, target["product_kind"], target["product_id"], claim["quantity"] + ) == "/checkout?product_kind=merchant_sku&product_id=plan%3Abasic&quantity=1" + + +def test_production_adapter_without_verified_cart_path_fails_before_navigation( + tmp_path, creds, monkeypatch, +): + monkeypatch.setattr( + pe, "load_payment_fields", + lambda: (_ for _ in ()).throw(AssertionError("credentials opened")), + ) + browser, bridge = FakeBrowser(), FakeBridge() + result = pe.run_once( + bridge_post=bridge, + browser=browser, + audit=pe.audit_factory(tmp_path), + state_dir=tmp_path, + checkout_url_for=lambda claim: "https://porkbun.com", + fake_e2e=False, + stdin=io.StringIO("tok\n"), + ) + assert result == pe.EXIT_DEFINITIVE_FAILURE + assert terminal_calls(bridge)[0][1]["failure_category"] == "checkout_not_ready" + assert browser.calls == [] + + +def test_authenticated_checkout_without_prepared_session_is_login_required( + tmp_path, creds, monkeypatch, +): + claim = json.loads(json.dumps(CLAIM)) + claim["checkout_target"]["session_requirement"] = "authenticated" + page = dict(CHECKOUT_PAGE, url="http://127.0.0.1:8000/cart", prepared_session=False) + monkeypatch.setattr( + pe, "load_payment_fields", + lambda: (_ for _ in ()).throw(AssertionError("credentials opened")), + ) + browser, bridge = FakeBrowser(pages=[page, CONFIRM_PAGE]), FakeBridge(claim=claim) + result = pe.run_once( + bridge_post=bridge, + browser=browser, + audit=pe.audit_factory(tmp_path), + state_dir=tmp_path, + checkout_url_for=lambda claimed: "http://127.0.0.1:8000", + fake_e2e=True, + stdin=io.StringIO("tok\n"), + ) + assert result == pe.EXIT_DEFINITIVE_FAILURE + assert terminal_calls(bridge)[0][1]["failure_category"] == "login_required" + assert [kind for kind, _ in browser.calls].count("navigate") == 1 + assert "fill" not in [kind for kind, _ in browser.calls] + + +@pytest.mark.parametrize( + "text,problem", + [ + (CHECKOUT_TEXT.replace("Subtotal: 20.00", "Subtotal: 21.00"), "subtotal_mismatch"), + (CHECKOUT_TEXT.replace("Tax: 2.00 AUD\n", ""), "tax_missing"), + (CHECKOUT_TEXT + "\nMandatory fees: 0.00 AUD", "mandatory_fees_ambiguous"), + (CHECKOUT_TEXT.replace("Total: 22.00", "Total: 23.00"), "total_mismatch"), + (CHECKOUT_TEXT.replace("Tax: 2.00 AUD", "Tax: 2.00 USD"), "currency_mismatch"), + (CHECKOUT_TEXT + "\nQuantity: 2", "quantity_ambiguous"), + (CHECKOUT_TEXT + "\nexample.com domain registration", "item_ambiguous"), + ( + CHECKOUT_TEXT.replace( + "example.com domain registration", + "Other item\nRelated: example.com domain registration", + ), + "item_missing", + ), + ], +) +def test_exact_commercial_breakdown_is_required(text, problem): + assert problem in pe.check_terms(text, CLAIM) + + +@pytest.mark.parametrize( + "text", + [ + CHECKOUT_TEXT.replace("Subtotal: 20.00", "Subtotal: 21.00"), + CHECKOUT_TEXT + "\nQuantity: 2", + CHECKOUT_TEXT + "\nexample.com domain registration", + CHECKOUT_TEXT.replace( + "example.com domain registration", + "Other item\nRelated: example.com domain registration", + ), + ], +) +def test_term_failure_precedes_credential_access(tmp_path, creds, monkeypatch, text): + page = dict(CHECKOUT_PAGE, text=text) + monkeypatch.setattr( + pe, "load_payment_fields", + lambda: (_ for _ in ()).throw(AssertionError("credentials opened")), + ) + browser, bridge = FakeBrowser(pages=[page, CONFIRM_PAGE]), FakeBridge() + assert run(browser, bridge, tmp_path) == pe.EXIT_DEFINITIVE_FAILURE + assert terminal_calls(bridge)[0][1]["failure_category"] == "terms_changed" + assert "discover" not in [kind for kind, _ in browser.calls] + + def test_check_terms_currency_item_quantity_recurrence(): assert pe.check_terms(CHECKOUT_TEXT, CLAIM) == [] assert "currency_missing" in pe.check_terms(CHECKOUT_TEXT.replace("AUD", "USD"), CLAIM) @@ -289,9 +475,13 @@ def test_check_terms_currency_item_quantity_recurrence(): CHECKOUT_TEXT.replace("Quantity: 1\n", ""), CLAIM) assert "unexpected_recurrence" in pe.check_terms( CHECKOUT_TEXT + "\nauto-renews yearly", CLAIM) - monthly = {**CLAIM, "recurrence_authorization": dict( - CLAIM["recurrence_authorization"], commitment_type="subscription", - billing_interval="monthly", auto_renew=True)} + monthly_recurrence = json.loads(json.dumps(CLAIM["recurrence_authorization"])) + monthly_recurrence.update( + commitment_type="subscription", billing_interval="monthly", auto_renew=True + ) + monthly = json.loads(json.dumps(CLAIM)) + monthly["recurrence_authorization"] = monthly_recurrence + monthly["checkout_terms"]["recurrence_authorization"] = monthly_recurrence assert "recurrence_not_shown" in pe.check_terms(CHECKOUT_TEXT, monthly) recurring_text = ( CHECKOUT_TEXT + "\nsubscription\nBilling interval: monthly\nRenewal amount: 22.00\n" @@ -1017,9 +1207,10 @@ def test_unsafe_checkout_url_is_rejected_before_navigation(tmp_path, creds): browser=browser, audit=pe.audit_factory(tmp_path), state_dir=tmp_path, - checkout_url_for=lambda claim: "https://registrar.example@evil.example/checkout", + checkout_url_for=lambda claim: "https://registrar.example@evil.example", fake_e2e=False, stdin=io.StringIO("tok\n"), + adapter_for=_adapter_for_test, ) assert result == pe.EXIT_DEFINITIVE_FAILURE assert browser.calls == []