From b6a0461739d3e55929027fbd8e28454522d728cb Mon Sep 17 00:00:00 2001 From: pbwraith-worker Date: Tue, 25 Aug 2026 04:22:00 -0500 Subject: [PATCH 1/3] gitea-saas-core#3: Stripe adapter (mockable seam: checkout + webhook -> idempotent credit) MockStripeClient (deterministic, no network, records events, raises StripeSignatureError on wrong sig) + real StripeClient behind env keys (lazy stripe import; CI runs mock-only). build_stripe_client() factory: mock by default, real only when STRIPE_SECRET_KEY + STRIPE_WEBHOOK_SECRET set. apply_payment_event() ties payment.succeeded to core.credit(..., idempotency_key=event_id) so duplicate webhook deliveries never double-credit. 11 hermetic tests (no network). --- gitea_saas/stripe_client.py | 195 +++++++++++++++++++++++++++++++++ tests/test_stripe_client.py | 207 ++++++++++++++++++++++++++++++++++++ 2 files changed, 402 insertions(+) create mode 100644 gitea_saas/stripe_client.py create mode 100644 tests/test_stripe_client.py diff --git a/gitea_saas/stripe_client.py b/gitea_saas/stripe_client.py new file mode 100644 index 0000000..b27e5d5 --- /dev/null +++ b/gitea_saas/stripe_client.py @@ -0,0 +1,195 @@ +"""Stripe adapter as a mockable seam (issue #3). + +Real implementation behind env keys; deterministic mock for tests and CI. +No live Stripe key needed in CI, no network in the mock path. The real +client imports the ``stripe`` library lazily, so this module loads (and +tests run) without it. + +Interface (duck-typed, no HTTP): + * ``create_checkout_session(client_id, amount_minor, currency='usd', + success_url, cancel_url) -> {"checkout_url", "session_id"}`` + * ``process_webhook(payload: bytes, signature: str) + -> {"event_id", "type", "client_id"?, "amount_minor"?}`` + raises ``StripeSignatureError`` on an invalid signature. + +Payments tie-in (server endpoint itself is issue #4): a +``payment.succeeded`` event is applied as +``core.credit(client_id, amount_minor, idempotency_key=event_id)`` via +``apply_payment_event``; duplicate webhook deliveries (same ``event_id``) +are no-ops thanks to the billing core's idempotency. +""" + +from __future__ import annotations + +import json +import os +from typing import Optional + + +class StripeSignatureError(Exception): + """Raised when a webhook signature fails verification.""" + + +class MockStripeClient: + """Deterministic in-memory Stripe adapter. Never touches the network. + + Records every checkout session and processed webhook event so tests can + assert on them. Signature check: any signature that is not the + configured ``webhook_signature`` raises ``StripeSignatureError``. + """ + + def __init__(self, checkout_base_url: str = "https://checkout.stripe.com/mock", + webhook_signature: str = "mock-webhook-secret"): + self.checkout_base_url = checkout_base_url.rstrip("/") + self.webhook_signature = webhook_signature + self.created_checkouts: list[dict] = [] + self.processed_events: list[dict] = [] + + # -- checkout ---------------------------------------------------------- + + def create_checkout_session(self, client_id: int, amount_minor: int, + currency: str = "usd", + success_url: Optional[str] = None, + cancel_url: Optional[str] = None) -> dict: + session_id = f"cs_mock_{client_id}_{amount_minor}_{len(self.created_checkouts) + 1}" + checkout_url = f"{self.checkout_base_url}/{session_id}" + self.created_checkouts.append({ + "client_id": client_id, "amount_minor": amount_minor, "currency": currency, + "session_id": session_id, "checkout_url": checkout_url, + "success_url": success_url, "cancel_url": cancel_url, + }) + return {"checkout_url": checkout_url, "session_id": session_id} + + # -- webhook ----------------------------------------------------------- + + def process_webhook(self, payload: bytes, signature: str) -> dict: + if signature != self.webhook_signature: + raise StripeSignatureError("webhook signature verification failed") + data = json.loads(payload.decode("utf-8")) + event = self._map_event(data) + self.processed_events.append(event) + return event + + @staticmethod + def _map_event(data: dict) -> dict: + """Normalize a Stripe webhook payload to {event_id, type, ...}.""" + event_id = data.get("id", "") + etype = data.get("type", "") + event: dict = {"event_id": event_id, "type": etype, + "client_id": None, "amount_minor": None} + obj = (data.get("data") or {}).get("object") or {} + if etype == "payment.succeeded": + metadata = obj.get("metadata") or {} + event["client_id"] = metadata.get("client_id") + event["amount_minor"] = obj.get("amount_total") + return event + + +class StripeClient: + """Real Stripe adapter (requires the ``stripe`` python package). + + Only used when ``STRIPE_SECRET_KEY`` and ``STRIPE_WEBHOOK_SECRET`` are + both set (see ``build_stripe_client``). CI runs mock-only. + """ + + def __init__(self, secret_key: str, webhook_secret: str): + self.secret_key = secret_key + self.webhook_secret = webhook_secret + + def _stripe(self): + try: + import stripe # lazy: module must load without the package + except ImportError as exc: # pragma: no cover - env not present in CI + raise RuntimeError( + "stripe python library is not installed; cannot use the real StripeClient" + ) from exc + stripe.api_key = self.secret_key + return stripe + + def create_checkout_session(self, client_id: int, amount_minor: int, + currency: str = "usd", + success_url: Optional[str] = None, + cancel_url: Optional[str] = None) -> dict: + stripe = self._stripe() + session = stripe.checkout.Session.create( + mode="payment", + line_items=[{ + "price_data": { + "currency": currency, + "unit_amount": amount_minor, + "product_data": {"name": f"gitea-saas client {client_id} top-up"}, + }, + "quantity": 1, + }], + success_url=success_url, + cancel_url=cancel_url, + metadata={"client_id": str(client_id)}, + ) + return {"checkout_url": session.url, "session_id": session.id} + + def process_webhook(self, payload: bytes, signature: str) -> dict: + stripe = self._stripe() + try: + event = stripe.Webhook.construct_event(payload, signature, self.webhook_secret) + except Exception as exc: # signature verification failure (any stripe error) + raise StripeSignatureError(str(exc)) from exc + obj = event["data"]["object"] + plain = { + "id": event["id"], + "type": event["type"], + "data": {"object": { + "metadata": dict(obj.get("metadata") or {}), + "amount_total": obj.get("amount_total"), + }}, + } + return MockStripeClient._map_event(plain) + + +def build_stripe_client(secret_key: Optional[str] = None, + webhook_secret: Optional[str] = None) -> object: + """Factory: mock by default; real client only when both keys are set. + + Explicit args win over environment variables. + """ + secret_key = secret_key if secret_key is not None else os.environ.get("STRIPE_SECRET_KEY") + webhook_secret = (webhook_secret if webhook_secret is not None + else os.environ.get("STRIPE_WEBHOOK_SECRET")) + if secret_key and webhook_secret: + return StripeClient(secret_key=secret_key, webhook_secret=webhook_secret) + return MockStripeClient() + + +def apply_payment_event(core, event: dict) -> tuple[bool, dict]: + """Server tie-in: apply a processed webhook event through the billing core. + + ``payment.succeeded`` -> ``core.credit(client_id, amount_minor, + idempotency_key=event_id)``. Duplicate webhook deliveries with the same + ``event_id`` are no-ops (billing-core idempotency), never double-credits. + Other event types are ignored with ``(True, {"applied": False})``. + + Returns the billing core's ``(ok, details)`` shape; errors are + structured, never raised. + """ + if not isinstance(event, dict) or event.get("type") != "payment.succeeded": + return True, {"applied": False, "reason": "ignored event type"} + event_id = event.get("event_id") + client_id = event.get("client_id") + amount_minor = event.get("amount_minor") + if not event_id or client_id is None or amount_minor is None: + return False, {"error": { + "code": "INCOMPLETE_EVENT", + "message": "payment.succeeded event missing event_id/client_id/amount_minor", + }} + try: + client_id = int(client_id) + amount_minor = int(amount_minor) + except (TypeError, ValueError): + return False, {"error": { + "code": "INCOMPLETE_EVENT", + "message": "payment.succeeded client_id/amount_minor must be numeric", + }} + ok, details = core.credit(client_id, amount_minor, event_id, + ref_type="stripe-event", ref_id=event_id) + if not ok: + return ok, {"applied": False, **details} + return ok, {"applied": True, **details} diff --git a/tests/test_stripe_client.py b/tests/test_stripe_client.py new file mode 100644 index 0000000..f1c4195 --- /dev/null +++ b/tests/test_stripe_client.py @@ -0,0 +1,207 @@ +"""Hermetic tests for the Stripe adapter seam (issue #3). + +Mock-only, zero network. Run with: + python -m pytest tests/test_stripe_client.py -q +""" + +import json +import sys +from pathlib import Path + +import pytest + +# PEP 420 namespace package: make the project root importable regardless of +# how pytest was invoked. +sys.path.insert(0, str(Path(__file__).resolve().parent.parent)) + +from gitea_saas.billing_core import BillingCore # noqa: E402 +from gitea_saas.stripe_client import ( # noqa: E402 + MockStripeClient, + StripeClient, + StripeSignatureError, + apply_payment_event, + build_stripe_client, +) + +WEBHOOK_PAYLOAD = { + "id": "evt_1A2B3C", + "type": "payment.succeeded", + "data": { + "object": { + "id": "pi_xyz", + "metadata": {"client_id": 7}, + "amount_total": 2500, + "currency": "usd", + } + }, +} + + +def payload_bytes(data=None) -> bytes: + return json.dumps(data or WEBHOOK_PAYLOAD).encode("utf-8") + + +def member_payload(event_id, event_type, client_id=None, amount=None): + return {"id": event_id, "type": event_type, + "data": {"object": {"metadata": {"client_id": client_id}, + "amount_total": amount}}} + + +# --------------------------------------------------------------------------- +# Checkout sessions +# --------------------------------------------------------------------------- + +def test_mock_creates_checkout_session_with_correct_fields(): + client = MockStripeClient() + result = client.create_checkout_session( + client_id=7, amount_minor=2500, currency="usd", + success_url="https://example.test/success", cancel_url="https://example.test/cancel", + ) + assert set(result) == {"checkout_url", "session_id"} + assert result["session_id"] == "cs_mock_7_2500_1" + assert result["checkout_url"].endswith("/" + result["session_id"]) + assert client.created_checkouts == [{ + "client_id": 7, "amount_minor": 2500, "currency": "usd", + "session_id": "cs_mock_7_2500_1", + "checkout_url": f"https://checkout.stripe.com/mock/cs_mock_7_2500_1", + "success_url": "https://example.test/success", + "cancel_url": "https://example.test/cancel", + }] + # Deterministic: next session id increments. + second = client.create_checkout_session(7, 2500) + assert second["session_id"] == "cs_mock_7_2500_2" + + +# --------------------------------------------------------------------------- +# Webhook mapping + signature check +# --------------------------------------------------------------------------- + +def test_handler_maps_payment_succeeded_fields(): + client = MockStripeClient() + event = client.process_webhook(payload_bytes(), client.webhook_signature) + assert event == {"event_id": "evt_1A2B3C", "type": "payment.succeeded", + "client_id": 7, "amount_minor": 2500} + assert client.processed_events == [event] + + +def test_bad_signature_raises_signature_error(): + client = MockStripeClient() + with pytest.raises(StripeSignatureError): + client.process_webhook(payload_bytes(), "wrong-signature") + assert client.processed_events == [] # nothing recorded + + +def test_non_payment_events_map_without_credit_fields(): + client = MockStripeClient() + event = client.process_webhook( + payload_bytes(member_payload("evt_9", "checkout.session.completed")), + client.webhook_signature, + ) + assert event == {"event_id": "evt_9", "type": "checkout.session.completed", + "client_id": None, "amount_minor": None} + + +# --------------------------------------------------------------------------- +# payment.succeeded -> idempotent credit tie-in +# --------------------------------------------------------------------------- + +def test_payment_succeeded_credits_once_and_duplicate_delivery_is_noop(tmp_path): + core = BillingCore(tmp_path / "billing.db") + ok, c = core.create_client("stripe-user") + assert ok + client = MockStripeClient() + + payload = payload_bytes(member_payload("evt_credit", "payment.succeeded", + client_id=c["client_id"], amount=2000)) + ok, details = apply_payment_event(core, client.process_webhook(payload, client.webhook_signature)) + assert ok and details["applied"] is True + assert details["amount_minor"] == 2000 + assert details["balance"] == 2000 + assert details["client_id"] == c["client_id"] + + # Same event_id re-delivered: no double credit. + ok, details = apply_payment_event(core, client.process_webhook(payload, client.webhook_signature)) + assert ok and details["applied"] is True + assert details["replayed"] is True + assert details["balance"] == 2000 + + ok, ledger = core.list_ledger(c["client_id"]) + assert ok and ledger["count"] == 1 # one credit row, applied once + entry = ledger["entries"][0] + assert entry["kind"] == "credit" and entry["idempotency_key"] == "evt_credit" + + +def test_apply_payment_event_ignores_non_payment_events(tmp_path): + core = BillingCore(tmp_path / "billing.db") + ok, c = core.create_client("stripe-user") + assert ok + client = MockStripeClient() + event = client.process_webhook( + payload_bytes(member_payload("evt_1", "checkout.session.completed", + client_id=c["client_id"], amount=5000)), + client.webhook_signature, + ) + ok, details = apply_payment_event(core, event) + assert ok and details["applied"] is False + ok, ledger = core.list_ledger(c["client_id"]) + assert ok and ledger["count"] == 0 + + +def test_apply_payment_event_unknown_client_structured_error(tmp_path): + core = BillingCore(tmp_path / "billing.db") + core.create_client("other") + client = MockStripeClient() + event = client.process_webhook( + payload_bytes(member_payload("evt_u", "payment.succeeded", client_id=999, amount=100)), + client.webhook_signature, + ) + ok, details = apply_payment_event(core, event) + assert ok is False and details["applied"] is False + assert details["error"]["code"] == "UNKNOWN_CLIENT" + + +def test_apply_payment_event_incomplete_event_structured_error(tmp_path): + core = BillingCore(tmp_path / "billing.db") + ok, c = core.create_client("stripe-user") + assert ok + client = MockStripeClient() + # Missing amount_total. + event = client.process_webhook( + payload_bytes({"id": "evt_i", "type": "payment.succeeded", + "data": {"object": {"metadata": {"client_id": c["client_id"]}}}}), + client.webhook_signature, + ) + ok, details = apply_payment_event(core, event) + assert ok is False and details["error"]["code"] == "INCOMPLETE_EVENT" + + +def test_apply_payment_event_coerces_string_metadata_client_id(tmp_path): + core = BillingCore(tmp_path / "billing.db") + ok, c = core.create_client("stripe-user") + assert ok + client = MockStripeClient() + # Stripe metadata values arrive as strings; the tie-in must coerce. + payload = payload_bytes(member_payload("evt_s", "payment.succeeded", + client_id=str(c["client_id"]), amount=750)) + event = client.process_webhook(payload, client.webhook_signature) + ok, details = apply_payment_event(core, event) + assert ok and details["applied"] is True and details["balance"] == 750 + + +# --------------------------------------------------------------------------- +# Factory: mock mode by default +# --------------------------------------------------------------------------- + +def test_mock_mode_enabled_when_env_absent(monkeypatch): + monkeypatch.delenv("STRIPE_SECRET_KEY", raising=False) + monkeypatch.delenv("STRIPE_WEBHOOK_SECRET", raising=False) + assert isinstance(build_stripe_client(), MockStripeClient) + + +def test_real_mode_enabled_when_env_present(monkeypatch): + monkeypatch.setenv("STRIPE_SECRET_KEY", "sk_test_abc") + monkeypatch.setenv("STRIPE_WEBHOOK_SECRET", "whsec_abc") + client = build_stripe_client() + assert isinstance(client, StripeClient) + assert client.secret_key == "sk_test_abc" + assert client.webhook_secret == "whsec_abc" -- 2.43.0 From 6fa667393a7567f0bebc84016c4b02002145a836 Mon Sep 17 00:00:00 2001 From: Claude Date: Tue, 25 Aug 2026 06:00:36 -0500 Subject: [PATCH 2/3] gitea-saas-core#3: fail-closed Stripe credential resolution (bounce fix) build_stripe_client() no longer degrades to the mock on absent config. Mock is selected only on explicit opt-in (mock=True or SAAS_STRIPE_MODE=mock); with mode unset/live, missing STRIPE_SECRET_KEY / STRIPE_WEBHOOK_SECRET raise StripeConfigError naming the variable (never its value). MockStripeClient's webhook_signature is now required (no published default) and verified with hmac.compare_digest. Regression test asserts the forged-webhook scenario leaves the balance unchanged. --- gitea_saas/stripe_client.py | 71 +++++++++++++++++--- tests/test_stripe_client.py | 130 ++++++++++++++++++++++++++++++++---- 2 files changed, 178 insertions(+), 23 deletions(-) diff --git a/gitea_saas/stripe_client.py b/gitea_saas/stripe_client.py index b27e5d5..fec14db 100644 --- a/gitea_saas/stripe_client.py +++ b/gitea_saas/stripe_client.py @@ -12,6 +12,14 @@ Interface (duck-typed, no HTTP): -> {"event_id", "type", "client_id"?, "amount_minor"?}`` raises ``StripeSignatureError`` on an invalid signature. +Credentials resolve FAIL-CLOSED (bounce fix): the mock is selected only on +an explicit opt-in (``mock=True`` or ``SAAS_STRIPE_MODE=mock``) and never as +the fallback for absent configuration. With the mode unset or ``live``, +missing/empty ``STRIPE_SECRET_KEY`` / ``STRIPE_WEBHOOK_SECRET`` raise +``StripeConfigError`` at construction, naming the missing variable (name +only, never the value). The mock requires an explicitly supplied +``webhook_signature`` -- there is no published default. + Payments tie-in (server endpoint itself is issue #4): a ``payment.succeeded`` event is applied as ``core.credit(client_id, amount_minor, idempotency_key=event_id)`` via @@ -22,6 +30,7 @@ are no-ops thanks to the billing core's idempotency. from __future__ import annotations import json +import hmac import os from typing import Optional @@ -30,16 +39,28 @@ class StripeSignatureError(Exception): """Raised when a webhook signature fails verification.""" +class StripeConfigError(Exception): + """Raised when Stripe credentials cannot be resolved (fail-closed). + + Carries only the names of the missing configuration, never its value. + """ + + class MockStripeClient: """Deterministic in-memory Stripe adapter. Never touches the network. Records every checkout session and processed webhook event so tests can assert on them. Signature check: any signature that is not the - configured ``webhook_signature`` raises ``StripeSignatureError``. + configured ``webhook_signature`` raises ``StripeSignatureError``. The + signature is REQUIRED -- it has no default, so a real deployment can + never inherit a published constant. """ - def __init__(self, checkout_base_url: str = "https://checkout.stripe.com/mock", - webhook_signature: str = "mock-webhook-secret"): + def __init__(self, webhook_signature: str, + checkout_base_url: str = "https://checkout.stripe.com/mock"): + if not isinstance(webhook_signature, str) or not webhook_signature: + raise StripeConfigError( + "MockStripeClient requires an explicit webhook_signature") self.checkout_base_url = checkout_base_url.rstrip("/") self.webhook_signature = webhook_signature self.created_checkouts: list[dict] = [] @@ -63,7 +84,8 @@ class MockStripeClient: # -- webhook ----------------------------------------------------------- def process_webhook(self, payload: bytes, signature: str) -> dict: - if signature != self.webhook_signature: + if (not isinstance(signature, str) + or not hmac.compare_digest(signature, self.webhook_signature)): raise StripeSignatureError("webhook signature verification failed") data = json.loads(payload.decode("utf-8")) event = self._map_event(data) @@ -146,17 +168,46 @@ class StripeClient: def build_stripe_client(secret_key: Optional[str] = None, - webhook_secret: Optional[str] = None) -> object: - """Factory: mock by default; real client only when both keys are set. + webhook_secret: Optional[str] = None, + mock: Optional[bool] = None, + webhook_signature: Optional[str] = None) -> object: + """Factory: mock only on explicit opt-in; otherwise fail closed. - Explicit args win over environment variables. + Bounce fix: the mock is NEVER the fallback for absent configuration. + It is selected only when ``mock=True`` or ``SAAS_STRIPE_MODE=mock``; + the mock's ``webhook_signature`` must be supplied explicitly (it has + no default, so a deployment cannot inherit a published constant). + + With the mode unset or ``live`` (or any value other than ``mock``), + ``STRIPE_SECRET_KEY`` and ``STRIPE_WEBHOOK_SECRET`` are both required: + a missing or empty variable raises :class:`StripeConfigError` naming + the variable (its name only -- never its value). Explicit arguments + win over environment variables. """ + mode = (os.environ.get("SAAS_STRIPE_MODE") or "").strip().lower() + if mock is None: + mock = mode == "mock" + if mock: + if not isinstance(webhook_signature, str) or not webhook_signature: + raise StripeConfigError( + "mock mode requires an explicit webhook_signature " + "(there is no default; supply it to build_stripe_client)") + return MockStripeClient(webhook_signature=webhook_signature) secret_key = secret_key if secret_key is not None else os.environ.get("STRIPE_SECRET_KEY") webhook_secret = (webhook_secret if webhook_secret is not None else os.environ.get("STRIPE_WEBHOOK_SECRET")) - if secret_key and webhook_secret: - return StripeClient(secret_key=secret_key, webhook_secret=webhook_secret) - return MockStripeClient() + missing = [name for name, value in (("STRIPE_SECRET_KEY", secret_key), + ("STRIPE_WEBHOOK_SECRET", webhook_secret)) + if not value] + if missing: + raise StripeConfigError( + "missing required Stripe credential(s): " + ", ".join(missing) + + " (set them, or opt into mock mode explicitly via " + "SAAS_STRIPE_MODE=mock / mock=True)") + # Both values are non-empty strings here (checked above); narrow for + # the static checker so no None can reach the real client. + assert secret_key is not None and webhook_secret is not None + return StripeClient(secret_key=secret_key, webhook_secret=webhook_secret) def apply_payment_event(core, event: dict) -> tuple[bool, dict]: diff --git a/tests/test_stripe_client.py b/tests/test_stripe_client.py index f1c4195..7ead475 100644 --- a/tests/test_stripe_client.py +++ b/tests/test_stripe_client.py @@ -18,11 +18,17 @@ from gitea_saas.billing_core import BillingCore # noqa: E402 from gitea_saas.stripe_client import ( # noqa: E402 MockStripeClient, StripeClient, + StripeConfigError, StripeSignatureError, apply_payment_event, build_stripe_client, ) +# Explicit mock signature for the hermetic suite. The point of the bounce +# fix: this constant lives in TESTS ONLY; the production client never has +# a default signature. +MOCK_SIG = "test-mock-secret" + WEBHOOK_PAYLOAD = { "id": "evt_1A2B3C", "type": "payment.succeeded", @@ -52,7 +58,7 @@ def member_payload(event_id, event_type, client_id=None, amount=None): # --------------------------------------------------------------------------- def test_mock_creates_checkout_session_with_correct_fields(): - client = MockStripeClient() + client = MockStripeClient(webhook_signature=MOCK_SIG) result = client.create_checkout_session( client_id=7, amount_minor=2500, currency="usd", success_url="https://example.test/success", cancel_url="https://example.test/cancel", @@ -77,7 +83,7 @@ def test_mock_creates_checkout_session_with_correct_fields(): # --------------------------------------------------------------------------- def test_handler_maps_payment_succeeded_fields(): - client = MockStripeClient() + client = MockStripeClient(webhook_signature=MOCK_SIG) event = client.process_webhook(payload_bytes(), client.webhook_signature) assert event == {"event_id": "evt_1A2B3C", "type": "payment.succeeded", "client_id": 7, "amount_minor": 2500} @@ -85,14 +91,14 @@ def test_handler_maps_payment_succeeded_fields(): def test_bad_signature_raises_signature_error(): - client = MockStripeClient() + client = MockStripeClient(webhook_signature=MOCK_SIG) with pytest.raises(StripeSignatureError): client.process_webhook(payload_bytes(), "wrong-signature") assert client.processed_events == [] # nothing recorded def test_non_payment_events_map_without_credit_fields(): - client = MockStripeClient() + client = MockStripeClient(webhook_signature=MOCK_SIG) event = client.process_webhook( payload_bytes(member_payload("evt_9", "checkout.session.completed")), client.webhook_signature, @@ -109,7 +115,7 @@ def test_payment_succeeded_credits_once_and_duplicate_delivery_is_noop(tmp_path) core = BillingCore(tmp_path / "billing.db") ok, c = core.create_client("stripe-user") assert ok - client = MockStripeClient() + client = MockStripeClient(webhook_signature=MOCK_SIG) payload = payload_bytes(member_payload("evt_credit", "payment.succeeded", client_id=c["client_id"], amount=2000)) @@ -135,7 +141,7 @@ def test_apply_payment_event_ignores_non_payment_events(tmp_path): core = BillingCore(tmp_path / "billing.db") ok, c = core.create_client("stripe-user") assert ok - client = MockStripeClient() + client = MockStripeClient(webhook_signature=MOCK_SIG) event = client.process_webhook( payload_bytes(member_payload("evt_1", "checkout.session.completed", client_id=c["client_id"], amount=5000)), @@ -150,7 +156,7 @@ def test_apply_payment_event_ignores_non_payment_events(tmp_path): def test_apply_payment_event_unknown_client_structured_error(tmp_path): core = BillingCore(tmp_path / "billing.db") core.create_client("other") - client = MockStripeClient() + client = MockStripeClient(webhook_signature=MOCK_SIG) event = client.process_webhook( payload_bytes(member_payload("evt_u", "payment.succeeded", client_id=999, amount=100)), client.webhook_signature, @@ -164,7 +170,7 @@ def test_apply_payment_event_incomplete_event_structured_error(tmp_path): core = BillingCore(tmp_path / "billing.db") ok, c = core.create_client("stripe-user") assert ok - client = MockStripeClient() + client = MockStripeClient(webhook_signature=MOCK_SIG) # Missing amount_total. event = client.process_webhook( payload_bytes({"id": "evt_i", "type": "payment.succeeded", @@ -179,7 +185,7 @@ def test_apply_payment_event_coerces_string_metadata_client_id(tmp_path): core = BillingCore(tmp_path / "billing.db") ok, c = core.create_client("stripe-user") assert ok - client = MockStripeClient() + client = MockStripeClient(webhook_signature=MOCK_SIG) # Stripe metadata values arrive as strings; the tie-in must coerce. payload = payload_bytes(member_payload("evt_s", "payment.succeeded", client_id=str(c["client_id"]), amount=750)) @@ -189,19 +195,117 @@ def test_apply_payment_event_coerces_string_metadata_client_id(tmp_path): # --------------------------------------------------------------------------- -# Factory: mock mode by default +# Factory: fail-closed credential resolution (bounce fix) # --------------------------------------------------------------------------- -def test_mock_mode_enabled_when_env_absent(monkeypatch): +def test_missing_credentials_raise_without_mock_optin(monkeypatch): + """No env keys and no explicit mock opt-in -> construction raises, + naming the missing variables without leaking any value.""" monkeypatch.delenv("STRIPE_SECRET_KEY", raising=False) monkeypatch.delenv("STRIPE_WEBHOOK_SECRET", raising=False) - assert isinstance(build_stripe_client(), MockStripeClient) + monkeypatch.delenv("SAAS_STRIPE_MODE", raising=False) + with pytest.raises(StripeConfigError) as excinfo: + build_stripe_client(secret_key="sk_test_SUPER_SECRET") + message = str(excinfo.value) + assert "STRIPE_WEBHOOK_SECRET" in message + assert "sk_test_SUPER_SECRET" not in message # name only, never the value + # Neither key at all -> both named. + with pytest.raises(StripeConfigError) as excinfo2: + build_stripe_client() + message2 = str(excinfo2.value) + assert "STRIPE_SECRET_KEY" in message2 and "STRIPE_WEBHOOK_SECRET" in message2 -def test_real_mode_enabled_when_env_present(monkeypatch): +def test_only_one_of_two_keys_raises(monkeypatch): + monkeypatch.delenv("STRIPE_SECRET_KEY", raising=False) + monkeypatch.delenv("STRIPE_WEBHOOK_SECRET", raising=False) + monkeypatch.delenv("SAAS_STRIPE_MODE", raising=False) + monkeypatch.setenv("STRIPE_SECRET_KEY", "sk_test_abc") + with pytest.raises(StripeConfigError) as excinfo: + build_stripe_client() + assert "STRIPE_WEBHOOK_SECRET" in str(excinfo.value) + assert "sk_test_abc" not in str(excinfo.value) + monkeypatch.delenv("STRIPE_SECRET_KEY") + monkeypatch.setenv("STRIPE_WEBHOOK_SECRET", "whsec_abc") + with pytest.raises(StripeConfigError) as excinfo2: + build_stripe_client() + assert "STRIPE_SECRET_KEY" in str(excinfo2.value) + assert "whsec_abc" not in str(excinfo2.value) + + +def test_live_mode_still_works_when_both_keys_set(monkeypatch): monkeypatch.setenv("STRIPE_SECRET_KEY", "sk_test_abc") monkeypatch.setenv("STRIPE_WEBHOOK_SECRET", "whsec_abc") + monkeypatch.setenv("SAAS_STRIPE_MODE", "live") client = build_stripe_client() assert isinstance(client, StripeClient) assert client.secret_key == "sk_test_abc" assert client.webhook_secret == "whsec_abc" + + +def test_explicit_mock_optin_with_explicit_signature_works(monkeypatch): + """The suite's seam survives: explicit opt-in + explicit signature.""" + monkeypatch.delenv("STRIPE_SECRET_KEY", raising=False) + monkeypatch.delenv("STRIPE_WEBHOOK_SECRET", raising=False) + monkeypatch.delenv("SAAS_STRIPE_MODE", raising=False) + client = build_stripe_client(mock=True, webhook_signature=MOCK_SIG) + assert isinstance(client, MockStripeClient) + assert client.webhook_signature == MOCK_SIG + # And the seam functions end to end. + event = client.process_webhook(payload_bytes(), MOCK_SIG) + assert event["event_id"] == "evt_1A2B3C" + + +def test_env_mock_mode_requires_explicit_signature(monkeypatch): + monkeypatch.delenv("STRIPE_SECRET_KEY", raising=False) + monkeypatch.delenv("STRIPE_WEBHOOK_SECRET", raising=False) + monkeypatch.setenv("SAAS_STRIPE_MODE", "mock") + # Opt-in via env, but no signature -> fail closed (no published default). + with pytest.raises(StripeConfigError): + build_stripe_client() + client = build_stripe_client(webhook_signature=MOCK_SIG) + assert isinstance(client, MockStripeClient) + assert client.webhook_signature == MOCK_SIG + + +def test_mock_without_signature_fails_closed(): + # No default exists any more: omitting the signature is a hard + # TypeError at construction (fail-closed; nothing to inherit). + with pytest.raises(TypeError): + MockStripeClient() # noqa: E1120 - deliberately missing the arg + with pytest.raises(StripeConfigError): + MockStripeClient(webhook_signature=None) + with pytest.raises(StripeConfigError): + MockStripeClient(webhook_signature="") + with pytest.raises(StripeConfigError): + build_stripe_client(mock=True) + + +def test_forged_webhook_regression_balance_unchanged(tmp_path, monkeypatch): + """Bounce regression: the exact forged-webhook scenario fails closed. + + With credentials absent (and no mock opt-in) construction RAISES, so a + deployment can never silently degrade to a public-constant verifier. + Even on an explicitly-opted-in mock, a webhook signed with the old + published constant is rejected and the balance stays exactly 0 -- + assert the balance, not just the exception, so a future refactor + cannot credit and then raise. + """ + monkeypatch.delenv("STRIPE_SECRET_KEY", raising=False) + monkeypatch.delenv("STRIPE_WEBHOOK_SECRET", raising=False) + monkeypatch.delenv("SAAS_STRIPE_MODE", raising=False) + with pytest.raises(StripeConfigError): + build_stripe_client() + + core = BillingCore(tmp_path / "billing.db") + ok, c = core.create_client("target") + assert ok + # An operator-chosen mock (this deployment's own secret, not a constant). + client = MockStripeClient(webhook_signature="operator-chosen-secret") + forged = payload_bytes(member_payload("evt_forged", "payment.succeeded", + client_id=c["client_id"], amount=500000)) + with pytest.raises(StripeSignatureError): + event = client.process_webhook(forged, "mock-webhook-secret") # old constant + apply_payment_event(core, event) # pragma: no cover - never reached + ok, balance = core.get_balance(c["client_id"]) + assert ok and balance["balance"] == 0 # unchanged: nothing was credited -- 2.43.0 From daeecc078d14546677f31c03689a56cd19b4b7a0 Mon Sep 17 00:00:00 2001 From: Claude Date: Tue, 25 Aug 2026 10:59:56 -0500 Subject: [PATCH 3/3] gitea-saas-core#3: tests: read create_client id key (bounce fix attempt 2, #16 rename) BillingCore.create_client now returns the flat record keyed as 'id' (main's PR #16/#19 rename). The Stripe test fixtures were reading the old 'client_id' return key, failing 5 tests on current main. Only the return value changed; the 'client_id' parameter to member_payload and the client_id metadata key are untouched. --- tests/test_stripe_client.py | 18 +++++++++--------- 1 file changed, 9 insertions(+), 9 deletions(-) diff --git a/tests/test_stripe_client.py b/tests/test_stripe_client.py index 7ead475..31e7ec3 100644 --- a/tests/test_stripe_client.py +++ b/tests/test_stripe_client.py @@ -118,12 +118,12 @@ def test_payment_succeeded_credits_once_and_duplicate_delivery_is_noop(tmp_path) client = MockStripeClient(webhook_signature=MOCK_SIG) payload = payload_bytes(member_payload("evt_credit", "payment.succeeded", - client_id=c["client_id"], amount=2000)) + client_id=c["id"], amount=2000)) ok, details = apply_payment_event(core, client.process_webhook(payload, client.webhook_signature)) assert ok and details["applied"] is True assert details["amount_minor"] == 2000 assert details["balance"] == 2000 - assert details["client_id"] == c["client_id"] + assert details["client_id"] == c["id"] # Same event_id re-delivered: no double credit. ok, details = apply_payment_event(core, client.process_webhook(payload, client.webhook_signature)) @@ -131,7 +131,7 @@ def test_payment_succeeded_credits_once_and_duplicate_delivery_is_noop(tmp_path) assert details["replayed"] is True assert details["balance"] == 2000 - ok, ledger = core.list_ledger(c["client_id"]) + ok, ledger = core.list_ledger(c["id"]) assert ok and ledger["count"] == 1 # one credit row, applied once entry = ledger["entries"][0] assert entry["kind"] == "credit" and entry["idempotency_key"] == "evt_credit" @@ -144,12 +144,12 @@ def test_apply_payment_event_ignores_non_payment_events(tmp_path): client = MockStripeClient(webhook_signature=MOCK_SIG) event = client.process_webhook( payload_bytes(member_payload("evt_1", "checkout.session.completed", - client_id=c["client_id"], amount=5000)), + client_id=c["id"], amount=5000)), client.webhook_signature, ) ok, details = apply_payment_event(core, event) assert ok and details["applied"] is False - ok, ledger = core.list_ledger(c["client_id"]) + ok, ledger = core.list_ledger(c["id"]) assert ok and ledger["count"] == 0 @@ -174,7 +174,7 @@ def test_apply_payment_event_incomplete_event_structured_error(tmp_path): # Missing amount_total. event = client.process_webhook( payload_bytes({"id": "evt_i", "type": "payment.succeeded", - "data": {"object": {"metadata": {"client_id": c["client_id"]}}}}), + "data": {"object": {"metadata": {"client_id": c["id"]}}}}), client.webhook_signature, ) ok, details = apply_payment_event(core, event) @@ -188,7 +188,7 @@ def test_apply_payment_event_coerces_string_metadata_client_id(tmp_path): client = MockStripeClient(webhook_signature=MOCK_SIG) # Stripe metadata values arrive as strings; the tie-in must coerce. payload = payload_bytes(member_payload("evt_s", "payment.succeeded", - client_id=str(c["client_id"]), amount=750)) + client_id=str(c["id"]), amount=750)) event = client.process_webhook(payload, client.webhook_signature) ok, details = apply_payment_event(core, event) assert ok and details["applied"] is True and details["balance"] == 750 @@ -303,9 +303,9 @@ def test_forged_webhook_regression_balance_unchanged(tmp_path, monkeypatch): # An operator-chosen mock (this deployment's own secret, not a constant). client = MockStripeClient(webhook_signature="operator-chosen-secret") forged = payload_bytes(member_payload("evt_forged", "payment.succeeded", - client_id=c["client_id"], amount=500000)) + client_id=c["id"], amount=500000)) with pytest.raises(StripeSignatureError): event = client.process_webhook(forged, "mock-webhook-secret") # old constant apply_payment_event(core, event) # pragma: no cover - never reached - ok, balance = core.get_balance(c["client_id"]) + ok, balance = core.get_balance(c["id"]) assert ok and balance["balance"] == 0 # unchanged: nothing was credited -- 2.43.0