gitea_saas/stripe_client.py

@@ -0,0 +1,246 @@
+"""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.
+
+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
+``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 hmac
+import os
+from typing import Optional
+
+
+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``. The
+    signature is REQUIRED -- it has no default, so a real deployment can
+    never inherit a published constant.
+    """
+
+    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] = []
+        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 (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)
+        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,
+                        mock: Optional[bool] = None,
+                        webhook_signature: Optional[str] = None) -> object:
+    """Factory: mock only on explicit opt-in; otherwise fail closed.
+
+    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"))
+    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]:
+    """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}

tests/test_stripe_client.py

@@ -0,0 +1,311 @@
+"""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,
+    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",
+    "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(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",
+    )
+    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(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}
+    assert client.processed_events == [event]
+
+
+def test_bad_signature_raises_signature_error():
+    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(webhook_signature=MOCK_SIG)
+    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(webhook_signature=MOCK_SIG)
+
+    payload = payload_bytes(member_payload("evt_credit", "payment.succeeded",
+                                           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["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["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(webhook_signature=MOCK_SIG)
+    event = client.process_webhook(
+        payload_bytes(member_payload("evt_1", "checkout.session.completed",
+                                     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["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(webhook_signature=MOCK_SIG)
+    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(webhook_signature=MOCK_SIG)
+    # Missing amount_total.
+    event = client.process_webhook(
+        payload_bytes({"id": "evt_i", "type": "payment.succeeded",
+                       "data": {"object": {"metadata": {"client_id": c["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(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["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: fail-closed credential resolution (bounce fix)
+# ---------------------------------------------------------------------------
+
+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)
+    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_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["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["id"])
+    assert ok and balance["balance"] == 0  # unchanged: nothing was credited