pbwraith/gitea-saas-core
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity
f39343ea81
gitea-saas-core/tests/test_client_api.py
Claude f39343ea81 gitea-saas-core#41: client-facing surface — per-client keys, checkout, /api/me, self-serve work submission, Stripe webhook 
- client_api.py: ClientKeyStore (SHA-256 hashed at rest, scoped, revocable),
  checkout with per-client rate limit, /api/me status, /api/me/issues submit,
  signature-authenticated /api/webhooks/stripe reusing the issue-#3 seam, and
  operator key-issuance routes; compound assembly guard replaces the billing
  operator before_request with identical operator-route semantics so scoped
  client endpoints are reachable and valid client keys get 403 on operator
  routes (cross-tenant + operator denials).
- auth.py: PUBLIC_PATHS + /api/webhooks/stripe; FORBIDDEN_BODY/
  forbidden_response; hash_client_key (no new compare_digest call site).
- schema.sql: client_api_keys table (hash + prefix only).
- app.py: mounts the client surface (stripe_builder/gitea injectable).
- tests/test_client_api.py (18 tests) + assembled-app coverage in test_app.py.
  Full suite: 293 passed, 1 skipped.
2026-09-05 04:34:50 -05:00

552 lines
22 KiB
Python
Raw Blame History

"""Hermetic tests for the client-facing surface (issue #41).
Per-client API keys (hashed at rest, scoped, revocable), self-serve
checkout (rate-limited), the Stripe webhook credit path (idempotent via
the issue-#3 seam), ``GET /api/me`` client status, self-serve work
submission, and the cross-tenant / operator-route denial guarantees.
Zero network: Flask test client + temp SQLite DB + MockStripeClient +
FakeGitea. Run with:
python -m pytest tests/test_client_api.py -q
"""
import json
import sys
from pathlib import Path
from typing import Optional
import pytest
sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
from gitea_saas.auth import (API_KEY_HEADER, FORBIDDEN_BODY, # noqa: E402
UNAUTHORIZED_BODY, hash_client_key)
from gitea_saas.billing_core import BillingCore # noqa: E402
from gitea_saas.client_api import (ClientKeyStore, ClientStatus, # noqa: E402
ClientSurface, register_client_routes)
from gitea_saas.config import ServerConfig # noqa: E402
from gitea_saas.stripe_client import MockStripeClient # noqa: E402
API_KEY = "operator-test-key-41"
AUTH = {API_KEY_HEADER: API_KEY}
# Explicit mock signature: tests only, never a production default.
MOCK_SIG = "test-webhook-sig-41"
class FakeGitea:
"""In-memory create_org_repo/create_issue seam (factory contract)."""
def __init__(self):
self.repos = set()
self.issues = []
def create_org_repo(self, name):
self.repos.add(name)
def create_issue(self, repo, title, body, labels=None):
number = len(self.issues) + 1
self.issues.append({"repo": repo, "title": title, "body": body,
"labels": labels, "number": number})
return number
def _mock_stripe_builder():
return MockStripeClient(webhook_signature=MOCK_SIG)
def _cfg(tmp_path: Path, api_key: Optional[str] = API_KEY) -> ServerConfig:
return ServerConfig(api_key=api_key, port=8080, db_path=tmp_path / "saas.db")
def _webhook_payload(event_id="evt_41_1", client_id=1, amount=2500):
return {"id": event_id, "type": "payment.succeeded",
"data": {"object": {"metadata": {"client_id": str(client_id)},
"amount_total": amount}}}
@pytest.fixture()
def client_factory(tmp_path):
"""Build ONE assembled app per test (cached); returns a zero-arg builder."""
state = {}
def build():
if "app" in state:
return state["app"], state["gitea"]
from gitea_saas.app import create_app
gitea = FakeGitea()
application = create_app(_cfg(tmp_path, API_KEY),
stripe_builder=_mock_stripe_builder,
gitea=gitea)
application.config.update(TESTING=True)
state["app"] = application
state["gitea"] = gitea
return application, gitea
build()
return build
@pytest.fixture()
def client(client_factory):
application, _ = client_factory()
return application.test_client()
@pytest.fixture()
def gitea(client_factory):
return client_factory()[1]
# ---------------------------------------------------------------------------
# helpers: provision clients + issue keys through the operator API
# ---------------------------------------------------------------------------
def _create_client(client, name: str) -> int:
resp = client.post("/api/clients", json={"name": name}, headers=AUTH)
assert resp.status_code == 201
return resp.get_json()["id"]
def _issue_key(client, client_id: int) -> dict:
resp = client.post(f"/api/clients/{client_id}/keys", headers=AUTH)
assert resp.status_code == 201
return resp.get_json()
def _client_header(raw_key: str) -> dict:
return {API_KEY_HEADER: raw_key}
# ---------------------------------------------------------------------------
# ClientKeyStore unit tests (acceptance criterion 2: hashed at rest)
# ---------------------------------------------------------------------------
def test_keys_stored_hashed_plaintext_never_in_db(tmp_path):
core = BillingCore(tmp_path / "saas.db")
ok, rec = core.create_client("HashCo")
store = ClientKeyStore(tmp_path / "saas.db")
ok, issued = store.issue(rec["id"])
assert ok
raw = issued["key"]
# The plaintext key is not anywhere in the DB — not as a value, not as
# a column. The stored material is hash + display prefix only.
rows = store._conn.execute(
"SELECT key_hash, key_prefix FROM client_api_keys").fetchall()
assert len(rows) == 1
stored_hash, stored_prefix = rows[0]["key_hash"], rows[0]["key_prefix"]
assert hash_client_key(raw) == stored_hash
assert stored_hash != raw
assert stored_prefix != raw # prefix is a truncated display hint
assert raw not in stored_hash
# Dump the whole DB as text: plaintext must not appear anywhere.
blob = " ".join(str(v) for row in store._conn.execute(
"SELECT * FROM client_api_keys") for v in row)
assert raw not in blob
def test_resolve_client_id_scoped_and_revoked(tmp_path):
core = BillingCore(tmp_path / "saas.db")
_, a = core.create_client("Alice")
_, b = core.create_client("Bob")
store = ClientKeyStore(tmp_path / "saas.db")
_, ka = store.issue(a["id"])
_, kb = store.issue(b["id"])
assert store.resolve_client_id(ka["key"]) == a["id"]
assert store.resolve_client_id(kb["key"]) == b["id"]
assert store.resolve_client_id("saas_wrong") is None
assert store.resolve_client_id("") is None
assert store.resolve_client_id(None) is None
# Revoke A's key only; B's keeps resolving.
ok, _ = store.revoke(a["id"], ka["key_id"])
assert ok
assert store.resolve_client_id(ka["key"]) is None
assert store.resolve_client_id(kb["key"]) == b["id"]
def test_revoke_is_scoped_no_cross_tenant_revoke(tmp_path):
core = BillingCore(tmp_path / "saas.db")
_, a = core.create_client("Alice")
_, b = core.create_client("Bob")
store = ClientKeyStore(tmp_path / "saas.db")
_, ka = store.issue(a["id"])
ok, details = store.revoke(b["id"], ka["key_id"])
assert not ok and (details.get("error") or {}).get("code") == "KEY_NOT_FOUND"
assert store.resolve_client_id(ka["key"]) == a["id"] # still active
ok, details = store.revoke(a["id"], 999999)
assert not ok and (details.get("error") or {}).get("code") == "KEY_NOT_FOUND"
def test_list_and_revoke_all_never_expose_secrets(tmp_path):
core = BillingCore(tmp_path / "saas.db")
_, a = core.create_client("Alice")
store = ClientKeyStore(tmp_path / "saas.db")
_, k1 = store.issue(a["id"])
_, k2 = store.issue(a["id"])
ok, listing = store.list_keys(a["id"])
assert ok and listing["count"] == 2
for item in listing["keys"]:
assert "key" not in item and "key_hash" not in item
ok, res = store.revoke_all(a["id"])
assert ok and res["revoked_count"] == 2
assert store.resolve_client_id(k1["key"]) is None
assert store.resolve_client_id(k2["key"]) is None
def test_issue_unknown_client(tmp_path):
BillingCore(tmp_path / "saas.db")
store = ClientKeyStore(tmp_path / "saas.db")
ok, details = store.issue(999)
assert not ok and (details.get("error") or {}).get("code") == "UNKNOWN_CLIENT"
# ---------------------------------------------------------------------------
# Operator key management HTTP routes
# ---------------------------------------------------------------------------
def test_operator_issues_and_revokes_keys(client):
cid = _create_client(client, "KeyCo")
issued = _issue_key(client, cid)
assert issued["client_id"] == cid
assert issued["key"].startswith("saas_")
assert issued["prefix"] == issued["key"][:10]
assert len(issued["key"]) > 20
# Listing shows metadata only.
listing = client.get(f"/api/clients/{cid}/keys", headers=AUTH)
assert listing.status_code == 200
assert listing.get_json()["count"] == 1
assert "key" not in listing.get_json()["keys"][0]
# The key works on the client surface.
me = client.get("/api/me", headers=_client_header(issued["key"]))
assert me.status_code == 200
assert me.get_json()["client_id"] == cid
# Revoke via the operator route: key stops working.
rev = client.delete(f"/api/clients/{cid}/keys/{issued['key_id']}",
headers=AUTH)
assert rev.status_code == 200
assert client.get("/api/me",
headers=_client_header(issued["key"])).status_code == 401
def test_key_routes_require_operator_key(client):
cid = _create_client(client, "LockedCo")
assert client.post(f"/api/clients/{cid}/keys").status_code == 401
assert client.get(f"/api/clients/{cid}/keys").status_code == 401
# A CLIENT key (other tenant) is 403 on key issuance routes.
key = _issue_key(client, cid)
other = _create_client(client, "OtherCo")
resp = client.post(f"/api/clients/{other}/keys",
headers=_client_header(key["key"]))
assert resp.status_code == 403
assert resp.get_json() == FORBIDDEN_BODY
# ---------------------------------------------------------------------------
# Tenancy: a client key never reaches another client's resources, and never
# reaches an operator route (acceptance criterion 1 + security note)
# ---------------------------------------------------------------------------
def test_cross_tenant_and_operator_denials(client):
a_id = _create_client(client, "TenantA")
b_id = _create_client(client, "TenantB")
key_a = _issue_key(client, a_id)["key"]
# Client A's key on client B's resources: 403 everywhere.
for path in (
f"/api/factory/{b_id}/issues",
f"/api/factory/{b_id}/lock",
f"/api/factory/{b_id}/unlock",
f"/api/clients/{b_id}",
f"/api/clients/{b_id}/ledger",
f"/api/clients/{b_id}/credits",
):
resp = client.post(path, json={}, headers=_client_header(key_a))
assert resp.status_code == 403, path
assert resp.get_json() == FORBIDDEN_BODY
# Operator routes: /api/clients, /api/factory/*, /dashboard.
assert client.get("/api/clients", headers=_client_header(key_a)) \
.status_code == 403
assert client.post("/api/factory/provision",
json={"client_id": b_id},
headers=_client_header(key_a)).status_code == 403
assert client.get("/dashboard", headers=_client_header(key_a)) \
.status_code == 403
assert client.get("/api/dashboard/summary", headers=_client_header(key_a)) \
.status_code == 403
# And the operator key never works on the client surface.
assert client.get("/api/me", headers=AUTH).status_code == 401
assert client.get("/api/me").get_json() == UNAUTHORIZED_BODY
assert client.post("/api/checkout", json={"amount_minor": 100},
headers=AUTH).status_code == 401
def test_each_client_sees_only_its_own_me(client):
a_id = _create_client(client, "SelfScopeA")
b_id = _create_client(client, "SelfScopeB")
key_a = _issue_key(client, a_id)["key"]
key_b = _issue_key(client, b_id)["key"]
me_a = client.get("/api/me", headers=_client_header(key_a)).get_json()
me_b = client.get("/api/me", headers=_client_header(key_b)).get_json()
assert me_a["client_id"] == a_id
assert me_b["client_id"] == b_id
assert me_a["name"] == "SelfScopeA"
assert me_b["name"] == "SelfScopeB"
# ---------------------------------------------------------------------------
# Checkout (rate-limited) + webhook credit (idempotent) — acceptance 3
# ---------------------------------------------------------------------------
def test_checkout_creates_session_scoped_to_caller(client, tmp_path):
cid = _create_client(client, "CheckoutCo")
key = _issue_key(client, cid)["key"]
resp = client.post("/api/checkout",
json={"amount_minor": 2500,
"success_url": "https://example.test/ok"},
headers=_client_header(key))
assert resp.status_code == 201
body = resp.get_json()
assert set(body) == {"checkout_url", "session_id"}
assert body["checkout_url"].startswith("https://checkout.stripe.com/mock/")
assert body["session_id"].startswith("cs_mock_")
def test_checkout_webhook_credit_exactly_once(client, tmp_path):
"""Acceptance 3: checkout -> webhook -> balance credit, replay no-op."""
cid = _create_client(client, "FundCo")
key = _issue_key(client, cid)["key"]
checkout = client.post("/api/checkout", json={"amount_minor": 2500},
headers=_client_header(key))
assert checkout.status_code == 201
payload = _webhook_payload(client_id=cid, amount=2500)
headers = {"Stripe-Signature": MOCK_SIG}
first = client.post("/api/webhooks/stripe",
data=json.dumps(payload), # raw bytes path
content_type="application/json", headers=headers)
assert first.status_code == 200
assert first.get_json()["applied"] is True
balance = client.get("/api/me", headers=_client_header(key)).get_json()
assert balance["balance_minor"] == 2500
# Replayed webhook (same event id): exactly once — no double credit.
replay = client.post("/api/webhooks/stripe",
data=json.dumps(payload),
content_type="application/json", headers=headers)
assert replay.status_code == 200
balance_after = client.get("/api/me",
headers=_client_header(key)).get_json()
assert balance_after["balance_minor"] == 2500
def test_webhook_rejects_bad_signature_and_bad_payload(client):
cid = _create_client(client, "WebhookCo")
payload = json.dumps(_webhook_payload(client_id=cid)).encode()
resp = client.post("/api/webhooks/stripe", data=payload,
content_type="application/json",
headers={"Stripe-Signature": "wrong-sig"})
assert resp.status_code == 400
assert resp.get_json()["error"]["code"] == "invalid_signature"
# No signature header at all: same generic 400.
resp = client.post("/api/webhooks/stripe", data=payload,
content_type="application/json")
assert resp.status_code == 400
def test_checkout_validation(client):
cid = _create_client(client, "BadCo")
key = _issue_key(client, cid)["key"]
h = _client_header(key)
assert client.post("/api/checkout", json={"amount_minor": 0},
headers=h).status_code == 400
assert client.post("/api/checkout", json={"amount_minor": "50"},
headers=h).status_code == 400
assert client.post("/api/checkout", json={"amount_minor": 100,
"currency": "too-long"},
headers=h).status_code == 400
assert client.post("/api/checkout",
json={"amount_minor": 100,
"success_url": "javascript:alert(1)"},
headers=h).status_code == 400
assert client.post("/api/checkout", json="nope",
headers=h).status_code == 400
def test_checkout_rate_limited_per_client(tmp_path):
"""Rate-limit /api/checkout per client (security note)."""
db = tmp_path / "saas.db"
core = BillingCore(db)
ok, rec = core.create_client("RatedCo")
store = ClientKeyStore(db)
ok, issued = store.issue(rec["id"])
import flask
from gitea_saas.client_api import ClientStatus
app = flask.Flask(__name__)
app.config["API_KEY"] = API_KEY
surface = ClientSurface(core, store, ClientStatus(core, db),
_mock_stripe_builder,
checkout_rate=2, checkout_window_s=60)
register_client_routes(app, surface,
gitea=FakeGitea(),
config=type("C", (), {"db_path": db})())
app.config.update(TESTING=True)
c = app.test_client()
h = _client_header(issued["key"])
assert c.post("/api/checkout", json={"amount_minor": 100},
headers=h).status_code == 201
assert c.post("/api/checkout", json={"amount_minor": 100},
headers=h).status_code == 201
third = c.post("/api/checkout", json={"amount_minor": 100}, headers=h)
assert third.status_code == 429
assert third.get_json()["error"]["code"] == "rate_limited"
assert third.headers.get("Retry-After") is not None
# Other client's limiter is its own (no cross-tenant throttling).
ok, rec2 = core.create_client("OtherRated")
ok, issued2 = store.issue(rec2["id"])
assert c.post("/api/checkout", json={"amount_minor": 100},
headers=_client_header(issued2["key"])).status_code == 201
def test_stripe_unconfigured_fails_closed(tmp_path, monkeypatch):
"""No Stripe mode/creds -> checkout & webhook 503, never a 200."""
monkeypatch.delenv("SAAS_STRIPE_MODE", raising=False)
monkeypatch.delenv("STRIPE_SECRET_KEY", raising=False)
monkeypatch.delenv("STRIPE_WEBHOOK_SECRET", raising=False)
from gitea_saas.app import create_app
application = create_app(_cfg(tmp_path, API_KEY), gitea=FakeGitea())
application.config.update(TESTING=True)
c = application.test_client()
cid = c.post("/api/clients", json={"name": "NoStripe"},
headers=AUTH).get_json()["id"]
key = c.post(f"/api/clients/{cid}/keys", headers=AUTH).get_json()["key"]
checkout = c.post("/api/checkout", json={"amount_minor": 100},
headers=_client_header(key))
assert checkout.status_code == 503
assert checkout.get_json()["error"]["code"] == "stripe_unconfigured"
webhook = c.post("/api/webhooks/stripe",
data=json.dumps(_webhook_payload()).encode(),
content_type="application/json",
headers={"Stripe-Signature": "anything"})
assert webhook.status_code == 503
# ---------------------------------------------------------------------------
# /api/me status (balance, queued, in-flight, delivered)
# ---------------------------------------------------------------------------
def test_me_reports_balance_and_counts(client, gitea, tmp_path):
cid = _create_client(client, "StatusCo")
key = _issue_key(client, cid)["key"]
# Unfunded submit -> queued (no Gitea issue created).
resp = client.post("/api/me/issues",
json={"repo": "statusco-repo",
"title": "First task",
"acceptance_body": "do it"},
headers=_client_header(key))
assert resp.status_code == 202
assert resp.get_json()["delivered"] is False
me = client.get("/api/me", headers=_client_header(key)).get_json()
assert me["queued_items"] == 1
assert me["in_flight_items"] == 0
assert me["delivered_count"] == 0
assert me["balance_minor"] == 0
# Fund + lock: submits now queue; unlock flushes them to Gitea, which
# is what the factory records as "shipped" (in-flight, awaiting merge).
core = BillingCore(tmp_path / "saas.db")
ok, _ = core.credit(cid, 10000, "fund-1")
assert ok
assert client.post(f"/api/factory/{cid}/lock", json={},
headers=AUTH).status_code == 200
resp = client.post("/api/me/issues",
json={"repo": "statusco-repo",
"title": "Second task",
"acceptance_body": "do it too"},
headers=_client_header(key))
assert resp.status_code == 202
assert resp.get_json()["reason"] == "locked"
me = client.get("/api/me", headers=_client_header(key)).get_json()
assert me["queued_items"] == 2 # unfunded + locked
assert me["in_flight_items"] == 0
flush = client.post(f"/api/factory/{cid}/unlock", json={}, headers=AUTH)
assert flush.status_code == 200
assert flush.get_json()["flushed"] == 2
assert len(gitea.issues) == 2
me = client.get("/api/me", headers=_client_header(key)).get_json()
assert me["queued_items"] == 0
assert me["in_flight_items"] == 2 # shipped to Gitea, not delivered
assert me["balance_minor"] == 10000
# Delivery debit (what delivery.py writes on merge) -> delivered count.
ok, _ = core.debit(cid, 500, "deliver:statusco-repo:1:sha",
ref_type="pr", ref_id="statusco-repo#1")
assert ok
me = client.get("/api/me", headers=_client_header(key)).get_json()
assert me["delivered_count"] == 1
assert me["balance_minor"] == 9500
def test_me_issues_validation_and_scoping(client, tmp_path):
cid = _create_client(client, "SubmitCo")
other = _create_client(client, "OtherSubmit")
key = _issue_key(client, cid)["key"]
other_key = _issue_key(client, other)["key"]
h = _client_header(key)
assert client.post("/api/me/issues", json={"repo": "../evil",
"title": "x"},
headers=h).status_code == 400
assert client.post("/api/me/issues", json={"repo": "good-repo",
"title": ""},
headers=h).status_code == 400
# The submit is ALWAYS scoped to the caller — a queued row can only
# ever belong to the key's own client_id (factory enforces existence).
resp = client.post("/api/me/issues",
json={"repo": "submitco-repo", "title": "Task"},
headers=_client_header(other_key))
assert resp.status_code == 202
me_other = client.get("/api/me", headers=_client_header(other_key)).get_json()
assert me_other["queued_items"] == 1
me_self = client.get("/api/me", headers=h).get_json()
assert me_self["queued_items"] == 0
# ---------------------------------------------------------------------------
# Assembled-app reachability + fail-closed 503 (no SAAS_API_KEY)
# ---------------------------------------------------------------------------
def test_assembled_app_has_client_routes(client_factory, tmp_path, monkeypatch):
monkeypatch.delenv("SAAS_API_KEY", raising=False)
from gitea_saas.app import create_app
application = create_app(_cfg(tmp_path, api_key=None), gitea=FakeGitea())
application.config.update(TESTING=True)
c = application.test_client()
# No SAAS_API_KEY: every authed route (client surface included) is 503.
assert c.get("/api/me").status_code == 503
assert c.post("/api/checkout", json={"amount_minor": 100}).status_code == 503
assert c.post("/api/me/issues", json={}).status_code == 503
Reference in New Issue View Git Blame Copy Permalink