release: money-path key delivery + prior CI/security

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

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,13 +212,38 @@ 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),
     Route("/founder-status", founder_status),
     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)