From 19bf88c7ab4b6b145c4a9007d1255625fa3efe74 Mon Sep 17 00:00:00 2001 From: hrco <4655956+hrco@users.noreply.github.com> Date: Wed, 1 Jul 2026 11:05:17 +0200 Subject: [PATCH] fix: issue + deliver license key on purchase (close money path) subscription_created only counted a seat and returned 200 -- it never generated, stored, or delivered a key, so a paying customer received nothing. Now the webhook mints a signed Ed25519 key (founder=lifetime, pro=365d), persists it idempotently, and a new rate-limited /keys/mine lets the buyer self-retrieve it by purchase email. /keys/free also stores its key so retrieval is uniform. Adds 7 tests covering provisioning, retrieval, idempotency, normalization. Co-Authored-By: Claude Opus 4.8 --- tests/test_key_provisioning.py | 103 +++++++++++++++++++++++++++++++++ xbridge_mcp/http_server.py | 63 +++++++++++++++----- 2 files changed, 152 insertions(+), 14 deletions(-) create mode 100644 tests/test_key_provisioning.py diff --git a/tests/test_key_provisioning.py b/tests/test_key_provisioning.py new file mode 100644 index 0000000..36f5ea5 --- /dev/null +++ b/tests/test_key_provisioning.py @@ -0,0 +1,103 @@ +""" +Tests for the money path: a paid LemonSqueezy webhook must issue + persist a signed +license key, and the buyer must be able to self-retrieve it via /keys/mine. + +Isolated from real state via XBRIDGE_DB_PATH; a throw-away Ed25519 signing key is +injected via XBRIDGE_SIGNING_KEY so _load_signing_key() succeeds. +""" +import base64 +import hashlib +import hmac +import importlib +import json +import os +import tempfile + +os.environ.setdefault("XBRIDGE_DB_PATH", tempfile.mkdtemp(prefix="xbridge-prov-")) + +import pytest +from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey +from cryptography.hazmat.primitives.serialization import Encoding, NoEncryption, PrivateFormat +from starlette.testclient import TestClient + +from xbridge_mcp import http_server + +SECRET = "topsecret" + + +def _signing_key_b64() -> str: + raw = Ed25519PrivateKey.generate().private_bytes(Encoding.Raw, PrivateFormat.Raw, NoEncryption()) + return base64.b64encode(raw).decode() + + +@pytest.fixture +def hs(tmp_path, monkeypatch): + monkeypatch.setenv("XBRIDGE_DB_PATH", str(tmp_path)) + monkeypatch.setenv("LS_SIGNING_SECRET", SECRET) + monkeypatch.setenv("XBRIDGE_SIGNING_KEY", _signing_key_b64()) + importlib.reload(http_server) + yield http_server + for var in ("LS_SIGNING_SECRET", "XBRIDGE_SIGNING_KEY"): + monkeypatch.delenv(var, raising=False) + importlib.reload(http_server) + + +def _post_subscription(client, product_name, email): + body = json.dumps({ + "meta": {"event_name": "subscription_created"}, + "data": {"attributes": {"product_name": product_name, "variant_name": product_name, "user_email": email}}, + }).encode() + sig = hmac.new(SECRET.encode(), body, hashlib.sha256).hexdigest() + return client.post("/webhooks/ls", content=body, headers={"X-Signature": sig, "Content-Type": "application/json"}) + + +class TestPaidProvisioning: + def test_founder_purchase_issues_and_persists_key(self, hs, tmp_path): + client = TestClient(hs.app) + resp = _post_subscription(client, "xBridge Founder", "buyer@example.com") + assert resp.status_code == 200 + assert resp.json()["tier"] == "founder" + stored = json.loads((tmp_path / "issued_keys.json").read_text()) + assert stored["buyer@example.com"]["key"].startswith("xbrdg_v1.") + assert stored["buyer@example.com"]["tier"] == "founder" + + def test_buyer_can_retrieve_key_via_keys_mine(self, hs): + client = TestClient(hs.app) + _post_subscription(client, "xBridge Founder", "buyer@example.com") + resp = client.post("/keys/mine", json={"email": "buyer@example.com"}) + assert resp.status_code == 200 + assert resp.json()["key"].startswith("xbrdg_v1.") + assert resp.json()["tier"] == "founder" + + def test_pro_purchase_issues_pro_key(self, hs): + client = TestClient(hs.app) + resp = _post_subscription(client, "xBridge Pro", "pro@example.com") + assert resp.status_code == 200 + assert resp.json()["tier"] == "pro" + + def test_email_is_normalized_case_insensitive(self, hs): + client = TestClient(hs.app) + _post_subscription(client, "xBridge Founder", "MixedCase@Example.com") + resp = client.post("/keys/mine", json={"email": "mixedcase@example.com"}) + assert resp.status_code == 200 + + def test_repeat_webhook_is_idempotent_seat_and_key(self, hs): + client = TestClient(hs.app) + _post_subscription(client, "xBridge Founder", "buyer@example.com") + first = client.post("/keys/mine", json={"email": "buyer@example.com"}).json()["key"] + _post_subscription(client, "xBridge Founder", "buyer@example.com") + second = client.post("/keys/mine", json={"email": "buyer@example.com"}).json()["key"] + assert first == second # not regenerated + assert TestClient(hs.app).get("/founder-status").json()["founder_seats_used"] == 1 # one seat + + +class TestKeysMineRetrieval: + def test_unknown_email_returns_404(self, hs): + client = TestClient(hs.app) + resp = client.post("/keys/mine", json={"email": "nobody@example.com"}) + assert resp.status_code == 404 + + def test_invalid_email_returns_400(self, hs): + client = TestClient(hs.app) + resp = client.post("/keys/mine", json={"email": "not-an-email"}) + assert resp.status_code == 400 diff --git a/xbridge_mcp/http_server.py b/xbridge_mcp/http_server.py index 698fed0..223c4c3 100644 --- a/xbridge_mcp/http_server.py +++ b/xbridge_mcp/http_server.py @@ -124,26 +124,36 @@ async def webhook_ls(request: Request): attrs = payload.get("data", {}).get("attributes", {}) product_name = attrs.get("product_name", "") - email = attrs.get("user_email", "") + email = (attrs.get("user_email", "") or "").strip().lower() if not email: return JSONResponse({"error": "missing email"}, status_code=400) - if "founder" in product_name.lower() or "founder" in attrs.get("variant_name", "").lower(): + is_founder = "founder" in product_name.lower() or "founder" in attrs.get("variant_name", "").lower() + tier = "founder" if is_founder else "pro" + + # Founder seat accounting — only consume a seat the first time we see this email. + if is_founder: data = _read_counter() - if data["count"] >= FOUNDER_CAP: - return JSONResponse({"error": "Founder tier sold out (50/50)"}, status_code=410) - if email in data["emails"]: - return JSONResponse({"status": "ok", "note": "already claimed founder"}, status_code=200) - data["count"] += 1 - data["emails"].append(email) - _write_counter(data) - tier = "founder" - else: - tier = "pro" + if email not in data["emails"]: + if data["count"] >= FOUNDER_CAP: + return JSONResponse({"error": "Founder tier sold out (50/50)"}, status_code=410) + data["count"] += 1 + data["emails"].append(email) + _write_counter(data) + + # Issue + persist the signed key so the buyer can self-retrieve it via /keys/mine. Idempotent. + keys = _read_keys() + if not keys.get(email, {}).get("key"): + try: + signing_key = _load_signing_key() + except RuntimeError as e: + return JSONResponse({"error": str(e)}, status_code=500) + days = 36500 if is_founder else 365 + keys[email] = {"tier": tier, "key": _make_signed_key(email, tier, days, signing_key), "created": int(time.time())} + _write_keys(keys) print(f"[webhook] {event} | {email} | tier={tier} | founder_count={_read_counter()['count']}/{FOUNDER_CAP}") - return JSONResponse({"status": "ok", "tier": tier, "email": email}) async def keys_free(request: Request): @@ -173,7 +183,7 @@ async def keys_free(request: Request): except RuntimeError as e: return JSONResponse({"error": str(e)}, status_code=500) - keys[email] = {"tier": "free", "created": int(time.time())} + keys[email] = {"tier": "free", "key": key, "created": int(time.time())} _write_keys(keys) print(f"[keys] free key issued to {email}") @@ -202,6 +212,30 @@ async def keys_resend(request: Request): return JSONResponse({"status": "ok", "message": "Check your inbox. If using self-host, your original key is still valid."}) +async def keys_mine(request: Request): + """Self-serve retrieval: return the stored signed key for a purchaser's email.""" + client_ip = request.client.host if request.client else "unknown" + if not _check_and_record_rate_limit("keys_mine_ip", client_ip): + return JSONResponse({"error": "Too many requests. Try again later."}, status_code=429) + + try: + body = await request.json() + except Exception: + return JSONResponse({"error": "invalid json"}, status_code=400) + + email = (body.get("email") or "").strip().lower() + if not email or "@" not in email: + return JSONResponse({"error": "valid email required"}, status_code=400) + + if not _check_and_record_rate_limit("keys_mine_email", email): + return JSONResponse({"error": "Too many requests. Try again later."}, status_code=429) + + entry = _read_keys().get(email) + if not entry or not entry.get("key"): + return JSONResponse({"error": "No key found for this email. If you just purchased, wait a moment and retry."}, status_code=404) + + return JSONResponse({"key": entry["key"], "tier": entry["tier"], "email": email}) + routes = [ Route("/health", health), Route("/", version_info), @@ -209,6 +243,7 @@ async def keys_resend(request: Request): Route("/webhooks/ls", webhook_ls, methods=["POST"]), Route("/keys/free", keys_free, methods=["POST"]), Route("/keys/resend", keys_resend, methods=["POST"]), + Route("/keys/mine", keys_mine, methods=["POST"]), ] app = Starlette(routes=routes)