gitea-saas-core/tests/test_auth.py

"""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