pbwraith/gitea-saas-core
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Merge pull request 'Consolidate all X-API-Key auth into gitea_saas/auth.py (#24)' (#27) from issue-24 into main

Browse Source
This commit is contained in:
pbwraith 2026-08-30 00:31:11 +02:00
commit b4e577b31b
11 changed files with 497 additions and 90 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

79
gitea_saas/auth.py Normal file
View File

@ -0,0 +1,79 @@
"""Single X-API-Key auth path for the whole service (issue #24).
Every API-key front-end (``server.py``, ``delivery.py``, ``intake_factory.py``,
``dashboard.py``) delegates to this module. The header name, the constant-time
comparison and the 401 body are defined exactly once here; a front-end never
performs its own check or hardcodes the header literal.
``stripe_client.py`` verifies webhook *signatures* (a different secret on a
different verification path) and is explicitly out of scope for this issue —
it keeps its own constant-time signature check.
Public surface:
API_KEY_HEADER the only ``X-API-Key`` header literal in the package
PUBLIC_PATHS deliberately unauthenticated liveness/health routes
key_is_valid() constant-time, fail-closed check — the one API-key
comparison in the whole service
UNAUTHORIZED_BODY canonical 401 payload, byte-identical on every wire
unauthorized_response() Flask Response object carrying that body
"""
from __future__ import annotations
import hmac
from typing import Optional
# The one place the HTTP header literal lives. A `git grep` for the quoted
# header name must match exactly this definition (acceptance criterion 2).
API_KEY_HEADER = "X-API-Key"
# Routes that need no API key. Every front-end guard reads this set, so a
# route's exposure is declared here, not inside one PR's before_request
# (composition finding, 2026-08-26): /health (server.py), /healthz
# (dashboard.py) and /api/factory/health (intake_factory.py) are all
# liveness routes and must stay reachable without a key.
PUBLIC_PATHS = frozenset({
"/health",
"/healthz",
"/api/factory/health",
})
# Canonical 401 payload. All front-ends return exactly this (Flask
# `jsonify(UNAUTHORIZED_BODY)` for HTTP surfaces, the dict itself for the
# delivery `(result, status)` seam), so a client parses every front-end's
# auth failure identically (acceptance criterion 4).
UNAUTHORIZED_BODY = {
"error": {
"code": "UNAUTHORIZED",
"message": "missing or invalid API key",
},
}
def key_is_valid(provided_key: Optional[str],
configured_key: Optional[str]) -> bool:
"""Fail-closed, constant-time API-key check (single implementation).
A request is authorized only when a key is actually configured AND the
provided header value matches it exactly. An unset/empty or non-str
configured key refuses every request; non-str/empty provided values are
rejected before the comparison runs (the constant-time compare raises
``TypeError`` on non-str input).
This is the ONE API-key comparison in the service (acceptance criterion
1) — no front-end does its own.
"""
if not isinstance(configured_key, str) or not configured_key:
return False
if not isinstance(provided_key, str) or not provided_key:
return False
return hmac.compare_digest(provided_key, configured_key)
def unauthorized_response():
"""Flask Response with the canonical 401 body (status 401)."""
from flask import jsonify
resp = jsonify(UNAUTHORIZED_BODY)
resp.status_code = 401
return resp

12
gitea_saas/config.py
View File

@ -5,8 +5,9 @@ coordinated between issues #2 and #6 — see botmanager #21 for the ownership
pattern that motivated it):
* the HTTP server's frozen :class:`ServerConfig` (issue #2) — the billing
API's API-key discipline (``SAAS_API_KEY`` / ``X-API-Key`` header), the
port, and the DB path; and
API's API-key discipline (``SAAS_API_KEY`` env; the ``X-API-Key`` header
constant itself is owned by ``gitea_saas.auth``, issue #24), the port, and
the DB path; and
* the review-gate/policy tuning (issue #6): tier rates/policies
(``SAAS_TIERS_JSON``) plus the reviewer knobs (``REVIEW_*``), returned as
a plain dict by :func:`load_review_config`.
@ -32,8 +33,11 @@ DEFAULT_PORT = 8080
# (in-memory) SQLite database. Real deployments must set SAAS_DB_PATH.
DEFAULT_DB_PATH: Optional[Path] = None
# Header name carrying the API key on every authenticated request.
API_KEY_HEADER = "X-API-Key"
# Header name carrying the API key on every authenticated request. Owned by
# ``gitea_saas.auth`` (issue #24 consolidation); re-exported here so existing
# ``from .config import API_KEY_HEADER`` call sites keep working without a
# second copy of the literal.
from .auth import API_KEY_HEADER
# Tier defaults (issue #6): GOLD auto-merges on a green APPROVE at rate 100
# minor units per delivery; STANDARD is reviewed but operator-merged at rate

55
gitea_saas/dashboard.py
View File

@ -30,7 +30,6 @@ serves the exact same dict.
from __future__ import annotations
import hmac
import html
import os
import sqlite3
@ -38,13 +37,9 @@ from datetime import datetime, timedelta, timezone
from pathlib import Path
from typing import Any, Dict, List, Optional
EMPTY = "\u2014" # em dash empty-state marker
from .auth import API_KEY_HEADER, key_is_valid, unauthorized_response
# Single source for the API-key header name on this branch. The shared
# `config.API_KEY_HEADER` constant lands with issue #2/#6 (neither merged
# yet); when #24 consolidates the service's auth paths, this line is
# replaced by `from .config import API_KEY_HEADER` — no other change.
API_KEY_HEADER = "X-API-Key"
EMPTY = "\u2014" # em dash empty-state marker
DEFAULT_DEBIT_FLOOR_MINOR = 500
@ -412,7 +407,7 @@ def render_html(summary: Dict[str, Any]) -> str:
<table><tr><th>Repo</th><th>PR</th><th>Verdict</th><th>Head</th><th>Reviewed at</th></tr>{rows_rg}</table>
<h2>5 · Ledger flash</h2>
<table><tr><th>Client</th><th>Kind</th><th>Amount</th><th>Ref</th><th>Time</th></tr>{rows_lf}</table>
<p class="note">Listen-only page: bind the server to 127.0.0.1. Live numbers, no caching.</p>
<p class="note">Operator page — requires the same API key as the JSON summary. Live numbers, no caching.</p>
</body></html>"""
@ -435,11 +430,14 @@ def mount_dashboard(app, core=None, db_path=None, api_key=None):
Registered routes:
* ``GET /dashboard`` — HTML operator page (NO auth; the page renders a
loopback-only warning — the server must bind to 127.0.0.1).
* ``GET /api/dashboard/summary`` — JSON payload (``X-API-Key`` auth;
401 on missing/wrong key, 503 when no key is configured — fail-closed).
* ``GET /healthz`` — liveness.
* ``GET /dashboard`` — HTML operator page (the issue #25 guard, expressed
through ``gitea_saas.auth``: constant-time ``key_is_valid`` check, any
refusal — missing, wrong, or unset configured key — is the canonical
401 body; never billing data without a valid key).
* ``GET /api/dashboard/summary`` — JSON payload (``X-API-Key`` auth via
``gitea_saas.auth``; any refusal — missing, wrong, or unset configured
key — is the canonical 401 body, fail-closed).
* ``GET /healthz`` — liveness (public, in ``auth.PUBLIC_PATHS``).
``api_key`` fills ``app.config['API_KEY']`` only when unset, so the app
factory's key wins when both are configured.
@ -462,37 +460,30 @@ def mount_dashboard(app, core=None, db_path=None, api_key=None):
dashboard = Dashboard(core, db_path)
def _guard(view):
from functools import wraps
@wraps(view)
def guarded(*args, **kwargs):
configured = app.config.get("API_KEY")
if not configured:
return {"error": {"code": "API_KEY_NOT_CONFIGURED",
"message": "SAAS_API_KEY not set"}}, 503
provided = request.headers.get(API_KEY_HEADER, "")
if not hmac.compare_digest(provided, configured):
return {"error": {"code": "UNAUTHORIZED",
"message": "bad or missing X-API-Key"}}, 401
return view(*args, **kwargs)
return guarded
@app.get("/healthz")
def healthz():
return jsonify({"ok": True})
@app.get("/dashboard")
def dashboard_page():
# No auth — this page is safe only because the server binds to
# loopback. Recompute from the DB at request time (no caching).
# Guarded through the shared auth module (issues #24/#25): the same
# constant-time key_is_valid check and canonical 401 body as every
# other API-key route — never billing data without a valid key.
# Loopback binding is defense-in-depth only; it was never enforced,
# so it is NOT the authorization boundary. Recompute from the DB at
# request time (no caching).
if not key_is_valid(request.headers.get(API_KEY_HEADER, ""),
app.config.get("API_KEY")):
return unauthorized_response()
summary = dashboard.summary()
return (render_html(summary), 200,
{"Content-Type": "text/html; charset=utf-8"})
@app.get("/api/dashboard/summary")
@_guard
def dashboard_summary():
if not key_is_valid(request.headers.get(API_KEY_HEADER, ""),
app.config.get("API_KEY")):
return unauthorized_response()
return jsonify(dashboard.summary())
return app

29
gitea_saas/delivery.py
View File

@ -28,12 +28,13 @@ a malformed payload, 200 otherwise) -- the server.py webhook route
from __future__ import annotations
import hmac
import json
import os
from collections.abc import Mapping
from typing import Any, Optional
from .auth import UNAUTHORIZED_BODY, key_is_valid
# --- defaults (overridable via env, mirror issue #6's config.py) ----------
DEFAULT_TIERS: dict[str, dict[str, Any]] = {
@ -47,7 +48,6 @@ DEFAULT_FLOOR_MINOR = 500 # SAAS_DEBIT_FLOOR_MINOR ($5.00; matches issue #4)
KEY_PREFIX = "deliver"
# Error codes (structured, mirror billing-core conventions).
UNAUTHORIZED = "UNAUTHORIZED"
BAD_PAYLOAD = "BAD_PAYLOAD"
UNKNOWN_CLIENT = "UNKNOWN_CLIENT"
INTERNAL = "INTERNAL"
@ -120,22 +120,14 @@ class Delivery:
# ------------------------------------------------------------------
def authorized(self, provided_key: Optional[str]) -> bool:
"""Fail-closed, constant-time: a request is authorized only when
the configured ``SAAS_API_KEY`` is set and matches the provided
key exactly.
"""Fail-closed, constant-time API-key check.
The comparison uses :func:`hmac.compare_digest` (no timing side
channel on the shared secret's prefix). ``compare_digest`` raises
``TypeError`` on non-str input, so the ``isinstance`` guard runs
first — unset/empty config and non-str/empty keys all return
``False``.
Single implementation (issue #24): delegates to the shared
``gitea_saas.auth.key_is_valid`` — no comparison lives in this
module. A request is authorized only when ``SAAS_API_KEY`` is set
and matches the provided key exactly.
"""
expected = self.config.get("api_key")
if not expected:
return False
if not isinstance(provided_key, str) or not provided_key:
return False
return hmac.compare_digest(provided_key, expected)
return key_is_valid(provided_key, self.config.get("api_key"))
# ------------------------------------------------------------------
# The debit primitive
@ -279,8 +271,9 @@ class Delivery:
fields, 200 otherwise.
"""
if not self.authorized(provided_key):
return ({"ok": False, "error": {"code": UNAUTHORIZED,
"message": "bad or missing X-API-Key"}}), 401
# Canonical 401 body from gitea_saas.auth (issue #24): identical
# on every front-end's wire.
return (UNAUTHORIZED_BODY, 401)
if not isinstance(payload, dict):
return ({"ok": False, "error": {"code": BAD_PAYLOAD,
"message": "expected JSON object"}}), 400

30
gitea_saas/intake_factory.py
View File

@ -23,7 +23,6 @@ shipped without payment.
from __future__ import annotations
import hmac
import json
import os
import re
@ -33,15 +32,11 @@ from datetime import datetime, timezone
from pathlib import Path
from typing import Any, Dict, List, Optional
# Header carrying the API key. The shared constant lives in the sibling
# ``gitea_saas/config.py`` (issue #2/#6 auth), which lands in a later PR than
# this one; until that module is merged this file falls back to the same
# canonical value. Keeping the single source of truth means the factory
# routes use ``compare_digest`` on the same header name as ``server.py``.
try:
from .config import API_KEY_HEADER # shared with gitea_saas.server
except ImportError: # config.py not merged yet (sibling auth PRs)
API_KEY_HEADER = "X-API-Key"
# Single auth path (issue #24): the header name, the constant-time comparison
# and the 401 body all live in ``gitea_saas.auth``; this module delegates.
# config.py is long merged, so there is no fallback constant here — a failed
# import of a security-relevant constant must be loud, not silently re-hardcoded.
from .auth import API_KEY_HEADER, key_is_valid, unauthorized_response
# Default floor for "funded": a client must hold at least this many minor
# units (default = 5.00 USD) before intake will ship issues.
@ -342,8 +337,8 @@ class IntakeFactory:
# HTTP surface (mounted by the issue #2 Flask app; see register_factory_routes)
# ---------------------------------------------------------------------------
#
# Routes (same auth discipline as gitea_saas.server -- API_KEY_HEADER via
# hmac.compare_digest, fail-closed):
# Routes (single auth path, issue #24 -- gitea_saas.auth.key_is_valid via
# API_KEY_HEADER, fail-closed; /api/factory/health is in PUBLIC_PATHS):
# POST /api/factory/provision {"client_id": int, "repo_name"?: str}
# POST /api/factory/<client_id>/issues {"repo", "title", "acceptance_body",
# "labels"?: [...]}
@ -375,15 +370,10 @@ def register_factory_routes(app: Any, core: Any, gitea: Any, config: Any) -> Non
def _factory() -> IntakeFactory:
return IntakeFactory(core, gitea, config)
def _authorized() -> bool:
sent = request.headers.get(API_KEY_HEADER, "")
return bool(api_key) and hmac.compare_digest(sent, api_key)
def _require_auth():
if not _authorized():
resp = jsonify({"error": "missing or invalid API key", "code": "unauthorized"})
resp.status_code = 401
return resp
sent = request.headers.get(API_KEY_HEADER, "")
if not key_is_valid(sent, api_key):
return unauthorized_response()
return None
@app.get("/api/factory/health")

17
gitea_saas/server.py
View File

@ -44,14 +44,15 @@ tests hermetic (Flask test client + temp DB) and free of any network/Stripe.
from __future__ import annotations
import hmac
import sys
from pathlib import Path
from typing import Any, Dict, Optional, Tuple
from flask import Flask, jsonify, request
from .config import API_KEY_HEADER, load_config
from .auth import (API_KEY_HEADER, PUBLIC_PATHS, key_is_valid,
unauthorized_response)
from .config import load_config
class ApiError(Exception):
@ -219,7 +220,10 @@ def create_app(core: Any, api_key: str) -> Flask:
"""Build the Flask app around a billing ``core`` instance.
``api_key`` is matched against the ``X-API-Key`` header on every route
except ``/health``. Missing or wrong keys get ``401``.
except the liveness routes in ``auth.PUBLIC_PATHS`` (``/health``,
``/healthz``, ``/api/factory/health``). Missing or wrong keys get the
canonical 401 body from ``gitea_saas.auth`` (issue #24 consolidation —
this app delegates, it does not implement its own check).
"""
app = Flask(__name__)
app.config["CORE"] = core
@ -227,13 +231,12 @@ def create_app(core: Any, api_key: str) -> Flask:
@app.before_request
def _require_api_key():
if request.path.rstrip("/") == "/health":
if request.path.rstrip("/") in PUBLIC_PATHS:
return None
sent = request.headers.get(API_KEY_HEADER, "")
expected = app.config["API_KEY"]
if not expected or not isinstance(sent, str) or not hmac.compare_digest(
sent, expected):
return _error(401, "unauthorized", "missing or invalid API key")
if not key_is_valid(sent, expected):
return unauthorized_response()
return None
@app.errorhandler(ApiError)

288
tests/test_auth.py Normal file
View File

@ -0,0 +1,288 @@
"""Acceptance tests for the single X-API-Key auth path (issue #24).
Covers the issue's acceptance criteria against the ONE consolidated
authenticator (``gitea_saas.auth``):
1. exactly one ``compare_digest`` on the API-key path (auth.py; the
``stripe_client.py`` match is webhook-signature verification, which the
issue explicitly scopes out);
2. exactly one ``"X-API-Key"`` literal in the package (``auth.py``);
3. a wrong key of the SAME length is rejected, and an unset configured key
refuses every request rather than serving;
4. the 401 body is byte-identical across all front-ends (server, delivery,
intake factory, dashboard);
5. route exposure is declared in one place (health/liveness routes in
``auth.PUBLIC_PATHS``) and asserted per front-end, including the composed
production app (server app + mounted dashboard).
"""
from __future__ import annotations
import sys
import types
from pathlib import Path
sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
import pytest
import flask
from gitea_saas.auth import (API_KEY_HEADER, PUBLIC_PATHS, UNAUTHORIZED_BODY,
key_is_valid)
from gitea_saas.dashboard import mount_dashboard
from gitea_saas.delivery import Delivery, default_config
from gitea_saas.intake_factory import register_factory_routes
from gitea_saas import server as server_mod
API_KEY = "sekret"
AUTH = {API_KEY_HEADER: API_KEY}
class _Core:
"""Tiny billing-core stub good enough for auth-path tests.
Auth runs BEFORE any core call, so refusal tests never touch it; the
minimal create_client/get_client support the few positive probes.
"""
def create_client(self, name, tier="free", status="active", meta=None):
return True, {"id": 1, "client_id": 1, "name": name, "tier": tier,
"status": status, "meta": meta or {},
"created_at": "2026-01-01T00:00:00+00:00"}
def get_client(self, client_id, include_balance=True):
return True, {"id": client_id, "name": "Acme", "tier": "free",
"status": "active", "meta": {}, "balance": 0,
"created_at": "2026-01-01T00:00:00+00:00"}
# --------------------------------------------------------------------------
# Per-front-end apps (all built the way the owning tests build them)
# --------------------------------------------------------------------------
@pytest.fixture()
def server_app():
app = server_mod.create_app(_Core(), API_KEY)
app.config.update(TESTING=True)
return app
@pytest.fixture()
def delivery():
return Delivery(core=None, config=default_config({"SAAS_API_KEY": API_KEY}))
@pytest.fixture()
def dashboard_app(tmp_path):
a = flask.Flask(__name__)
mount_dashboard(a, core=_Core(), db_path=tmp_path / "dash.db", api_key=API_KEY)
a.config.update(TESTING=True)
return a
@pytest.fixture()
def intake_app():
a = flask.Flask(__name__)
a.config["API_KEY"] = API_KEY
cfg = types.SimpleNamespace(db_path=":memory:", api_key=API_KEY,
debit_floor_minor=500)
register_factory_routes(a, _Core(), types.SimpleNamespace(), cfg)
a.config.update(TESTING=True)
return a
# --------------------------------------------------------------------------
# Acceptance criterion 1 + 2: one implementation, one literal
# --------------------------------------------------------------------------
def _pkg_dir() -> Path:
# gitea_saas is a PEP 420 namespace package (no __init__.py), so resolve
# the package directory from this test file's own location.
return Path(__file__).resolve().parent.parent / "gitea_saas"
def test_single_compare_digest_on_api_key_path():
"""Criterion 1 scoped to the API-key path: exactly ONE comparison call
site on the API-key path (in auth.py). The second package call site is
``stripe_client.py`` webhook-SIGNATURE verification, which the issue
explicitly scopes out (leave it alone). No front-end performs its own
comparison."""
hits = []
for py in sorted(_pkg_dir().glob("*.py")):
for i, line in enumerate(py.read_text(encoding="utf-8").splitlines(), 1):
if "compare_digest(" in line: # call sites, not docstring mentions
hits.append(f"{py.name}:{i}")
assert hits == ["auth.py:70", "stripe_client.py:88"], f"unexpected call sites: {hits}"
def test_single_x_api_key_literal():
"""Criterion 2: exactly one ``\"X-API-Key\"`` literal in the package."""
hits = []
for py in sorted(_pkg_dir().glob("*.py")):
for i, line in enumerate(py.read_text(encoding="utf-8").splitlines(), 1):
if '"X-API-Key"' in line:
hits.append(f"{py.name}:{i}")
assert len(hits) == 1, f"expected exactly one literal, got {hits}"
assert hits[0].startswith("auth.py"), f"the literal must live in auth.py: {hits}"
# --------------------------------------------------------------------------
# key_is_valid (the single implementation)
# --------------------------------------------------------------------------
class TestKeyIsValid:
def test_matching_key(self):
assert key_is_valid(API_KEY, API_KEY) is True
def test_wrong_key_of_same_length_rejected(self):
# criterion 3: same-length wrong key must NOT pass (constant-time path)
wrong = "hunter2".ljust(len(API_KEY), "x")[:len(API_KEY)]
assert len(wrong) == len(API_KEY)
assert wrong != API_KEY
assert key_is_valid(wrong, API_KEY) is False
def test_unset_configured_key_refuses(self):
# criterion 3: unset configured key refuses everything, never serves
assert key_is_valid(API_KEY, None) is False
assert key_is_valid(API_KEY, "") is False
assert key_is_valid(API_KEY, b"sekret") is False # type: ignore[arg-type]
def test_missing_or_non_str_provided(self):
assert key_is_valid(None, API_KEY) is False
assert key_is_valid("", API_KEY) is False
assert key_is_valid(b"sekret", API_KEY) is False # type: ignore[arg-type]
# --------------------------------------------------------------------------
# Criterion 4: byte-identical 401 body from ALL four front-ends
# --------------------------------------------------------------------------
def test_401_body_identical_across_all_front_ends(server_app, delivery,
dashboard_app, intake_app):
# server (Flask)
r = server_app.test_client().get("/api/clients/1")
assert r.status_code == 401 and r.get_json() == UNAUTHORIZED_BODY
# delivery (plain (dict, status) seam)
result, status = delivery.handle_merge_webhook({}, provided_key=None)
assert status == 401 and result == UNAUTHORIZED_BODY
# dashboard (bare app + per-route guard)
r = dashboard_app.test_client().get("/api/dashboard/summary")
assert r.status_code == 401 and r.get_json() == UNAUTHORIZED_BODY
# intake factory (Flask)
r = intake_app.test_client().post("/api/factory/provision",
json={"client_id": 1})
assert r.status_code == 401 and r.get_json() == UNAUTHORIZED_BODY
def test_wrong_key_same_length_401_across_front_ends(server_app, delivery,
dashboard_app, intake_app):
wrong = "hunter2".ljust(len(API_KEY), "x")[:len(API_KEY)]
assert len(wrong) == len(API_KEY) and wrong != API_KEY
assert server_app.test_client().get(
"/api/clients/1", headers={API_KEY_HEADER: wrong}).status_code == 401
r, s = delivery.handle_merge_webhook({}, provided_key=wrong)
assert s == 401 and r == UNAUTHORIZED_BODY
assert dashboard_app.test_client().get(
"/api/dashboard/summary",
headers={API_KEY_HEADER: wrong}).status_code == 401
assert intake_app.test_client().post(
"/api/factory/provision", json={"client_id": 1},
headers={API_KEY_HEADER: wrong}).status_code == 401
def test_unset_configured_key_refuses_every_request(tmp_path):
# server: empty configured key -> deny all, even with a key presented
app = server_mod.create_app(_Core(), "")
app.config.update(TESTING=True)
r = app.test_client().get("/api/clients/1", headers=AUTH)
assert r.status_code == 401 and r.get_json() == UNAUTHORIZED_BODY
# dashboard: no configured key -> 401 (was 503 pre-consolidation)
a = flask.Flask(__name__)
mount_dashboard(a, core=_Core(), db_path=tmp_path / "d2.db", api_key=None)
a.config.update(TESTING=True)
r = a.test_client().get("/api/dashboard/summary")
assert r.status_code == 401 and r.get_json() == UNAUTHORIZED_BODY
# delivery: config with no api_key -> unauthorized
d = Delivery(core=None, config=default_config({}))
assert d.authorized(API_KEY) is False
r, s = d.handle_merge_webhook({}, provided_key=API_KEY)
assert s == 401 and r == UNAUTHORIZED_BODY
# intake: empty configured key -> 401
a2 = flask.Flask(__name__)
a2.config["API_KEY"] = ""
cfg = types.SimpleNamespace(db_path=":memory:", api_key="",
debit_floor_minor=500)
register_factory_routes(a2, _Core(), types.SimpleNamespace(), cfg)
a2.config.update(TESTING=True)
r = a2.test_client().post("/api/factory/provision", json={"client_id": 1},
headers=AUTH)
assert r.status_code == 401 and r.get_json() == UNAUTHORIZED_BODY
# --------------------------------------------------------------------------
# Route exposure: declared in PUBLIC_PATHS, honored by every front-end
# --------------------------------------------------------------------------
def test_public_paths_are_open(server_app, dashboard_app, intake_app):
assert server_app.test_client().get("/health").status_code == 200
assert dashboard_app.test_client().get("/healthz").status_code == 200
assert intake_app.test_client().get("/api/factory/health").status_code == 200
assert "/health" in PUBLIC_PATHS
assert "/healthz" in PUBLIC_PATHS
assert "/api/factory/health" in PUBLIC_PATHS
def test_composed_app_exposure(tmp_path):
"""Mount every front-end onto the ONE production app (server factory)
and assert each registered route's exposure — the composition test no
single PR's suite covers (owner note, 2026-08-26)."""
db = tmp_path / "composed.db"
app = server_mod.create_app(_Core(), API_KEY)
mount_dashboard(app, core=_Core(), db_path=db, api_key=API_KEY)
cfg = types.SimpleNamespace(db_path=db, api_key=API_KEY,
debit_floor_minor=500)
register_factory_routes(app, _Core(), types.SimpleNamespace(), cfg)
app.config.update(TESTING=True)
c = app.test_client()
# public liveness routes — reachable with NO key
assert c.get("/health").status_code == 200
assert c.get("/healthz").status_code == 200
assert c.get("/api/factory/health").status_code == 200
# protected routes — 401 without a key, 200/201 with the key
assert c.get("/api/clients/1").status_code == 401
assert c.get("/api/clients/1", headers=AUTH).status_code == 200
assert c.get("/api/dashboard/summary").status_code == 401
assert c.get("/api/dashboard/summary", headers=AUTH).status_code == 200
# /dashboard (HTML) is guarded on the composed app (server #2 guard)
# AND on the bare dashboard mount (issue #25, expressed through the
# shared auth path — see the bare-mount test below).
assert c.get("/dashboard").status_code == 401
assert c.get("/dashboard", headers=AUTH).status_code == 200
def test_bare_dashboard_mount_guards_page_and_json(tmp_path):
"""The dashboard's OWN mount guards /dashboard (issue #25, through the
shared auth module: constant-time key_is_valid + canonical 401) the same
way it guards /api/dashboard/summary — exposure is not silently
inherited from mount order, and no billing data leaks without a key."""
a = flask.Flask(__name__)
mount_dashboard(a, core=_Core(), db_path=tmp_path / "b.db", api_key=API_KEY)
a.config.update(TESTING=True)
c = a.test_client()
r = c.get("/dashboard")
assert r.status_code == 401
assert r.get_json() == UNAUTHORIZED_BODY
assert "Acme" not in r.get_data(as_text=True)
assert c.get("/dashboard", headers=AUTH).status_code == 200
assert c.get("/api/dashboard/summary").status_code == 401
assert c.get("/healthz").status_code == 200

46
tests/test_dashboard.py
View File

@ -27,6 +27,7 @@ import pytest
import flask
from gitea_saas.auth import UNAUTHORIZED_BODY
from gitea_saas.billing_core import BillingCore
from gitea_saas.dashboard import (Dashboard, EMPTY, render_html,
mount_dashboard, API_KEY_HEADER)
@ -291,7 +292,9 @@ class TestMountContract:
does NOT define or run a second Flask app (review finding: #2 owns
the app factory; three-way server.py conflict removed)."""
client = app.test_client()
assert client.get("/dashboard").status_code == 200
# /dashboard is key-protected (issue #25): open with a valid key.
assert client.get("/dashboard",
headers={API_KEY_HEADER: "sekret"}).status_code == 200
assert client.get("/healthz").status_code == 200
def test_no_second_server_module_in_this_issue(self):
@ -365,12 +368,38 @@ class TestServer:
def test_dashboard_page_open_and_html(self, app, seeded):
client = app.test_client()
r = client.get("/dashboard")
r = client.get("/dashboard", headers={API_KEY_HEADER: "sekret"})
assert r.status_code == 200
assert "text/html" in r.content_type
assert "SaaS dashboard" in r.get_data(as_text=True)
assert "Acme" in r.get_data(as_text=True)
def test_dashboard_page_guard_blocks_anonymous_access(self, app, seeded):
"""GET /dashboard with NO X-API-Key must NOT return billing data
(issue #25): same fail-closed guard as the JSON summary — 401 and no
client/ledger/queue content in the body."""
client = app.test_client()
r = client.get("/dashboard")
# no auth → 401, and no billing data leaks in the body
assert r.status_code == 401
body = r.get_data(as_text=True)
assert "Acme" not in body
assert "Poorco" not in body
assert "SaaS dashboard" not in body
def test_dashboard_wrong_key_same_length_rejected(self, app):
"""A wrong key of the SAME length as the configured key must be
rejected with 401 on the HTML page too (matches the JSON endpoint's
constant-time comparison contract, issue #25)."""
client = app.test_client()
configured = "sekret"
wrong_same_length = "hunter"
assert len(wrong_same_length) == len(configured)
r = client.get("/dashboard",
headers={API_KEY_HEADER: wrong_same_length})
assert r.status_code == 401
assert "Acme" not in r.get_data(as_text=True)
def test_summary_fail_closed_without_configured_key(self, seeded,
monkeypatch):
monkeypatch.delenv("SAAS_API_KEY", raising=False)
@ -379,9 +408,16 @@ class TestServer:
mount_dashboard(a, core=core, db_path=db, api_key=None)
client = a.test_client()
r = client.get("/api/dashboard/summary")
assert r.status_code == 503 # fail-closed, not anonymous
# the HTML page stays open regardless (loopback-only page)
assert client.get("/dashboard").status_code == 200
# Any refusal (missing, wrong, or UNSET configured key) is the same
# canonical 401 body (issue #24) — fail-closed, never anonymous.
assert r.status_code == 401
assert r.get_json() == UNAUTHORIZED_BODY
# The HTML page fails closed the same way (issue #25, expressed
# through the shared auth path): no key configured → 401, and no
# billing data is returned either way.
r2 = client.get("/dashboard")
assert r2.status_code == 401
assert "Acme" not in r2.get_data(as_text=True)
def test_healthz(self, app):
r = app.test_client().get("/healthz")

5
tests/test_delivery.py
View File

@ -13,6 +13,7 @@ import pytest
# regardless of how pytest was invoked.
sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
from gitea_saas.auth import UNAUTHORIZED_BODY # noqa: E402
from gitea_saas.billing_core import BillingCore # noqa: E402
from gitea_saas.delivery import ( # noqa: E402
Delivery,
@ -194,10 +195,12 @@ def test_webhook_without_auth_401(core):
result, status = d.handle_merge_webhook(payload, provided_key=None)
assert status == 401
assert result["error"]["code"] == "UNAUTHORIZED"
# Canonical 401 body (issue #24), byte-identical with every front-end.
assert result == UNAUTHORIZED_BODY
result, status = d.handle_merge_webhook(payload, provided_key="wrong")
assert status == 401
assert result == UNAUTHORIZED_BODY
def test_webhook_with_auth_ok(core):

19
tests/test_intake_factory.py
View File

@ -26,6 +26,7 @@ from pathlib import Path
import pytest
from gitea_saas.auth import UNAUTHORIZED_BODY
from gitea_saas.intake_factory import (
DEFAULT_DEBIT_FLOOR_MINOR,
IntakeFactory,
@ -438,8 +439,26 @@ def _post(app, path, payload=None, key="test-api-key"):
def test_routes_require_api_key(app_client):
resp = app_client.post("/api/factory/provision", json={"client_id": 1})
assert resp.status_code == 401
# Canonical 401 body (issue #24), byte-identical with every front-end.
assert resp.get_json() == UNAUTHORIZED_BODY
resp = app_client.post("/api/factory/1/unlock")
assert resp.status_code == 401
assert resp.get_json() == UNAUTHORIZED_BODY
def test_wrong_key_of_same_length_rejected(app_client):
"""Wrong key of the SAME length as the configured key must be rejected
(issue #24 criterion 3; the constant-time comparison is what makes this
a real check rather than a prefix-time leak)."""
configured = "test-api-key"
wrong_same_length = "hunter-id-123"[:len(configured)]
assert len(wrong_same_length) == len(configured)
assert wrong_same_length != configured
resp = app_client.post("/api/factory/provision",
json={"client_id": 1},
headers={"X-API-Key": wrong_same_length})
assert resp.status_code == 401
assert resp.get_json() == UNAUTHORIZED_BODY
def test_route_provision_submit_unlock_lock_flow(env, app_client):

7
tests/test_server.py
View File

@ -18,6 +18,7 @@ from pathlib import Path
import pytest
from gitea_saas import billing_core, server
from gitea_saas.auth import UNAUTHORIZED_BODY
from gitea_saas.config import API_KEY_HEADER
API_KEY = "test-api-key-secret"
@ -85,14 +86,14 @@ def test_health_is_public(client):
def test_missing_key_is_401(client):
resp = client.get("/api/clients/1")
assert resp.status_code == 401
body = resp.get_json()
assert body["code"] == "unauthorized"
assert resp.get_json() == UNAUTHORIZED_BODY
def test_wrong_key_is_401(client):
resp = client.get("/api/clients/1", headers={API_KEY_HEADER: "nope"})
assert resp.status_code == 401
assert resp.get_json()["code"] == "unauthorized"
# Canonical 401 body (issue #24): byte-identical across every front-end.
assert resp.get_json() == UNAUTHORIZED_BODY
def test_correct_key_is_200(client):