AegisGitea-MCP/tests/test_oauth_flow_security.py

"""Threat cases of the per-user OAuth flow, exercised over the real HTTP app.

Covered: token expiry, refresh rotation and reuse detection, revocation, code
replay, PKCE and client binding, state replay and expiry, the confused-deputy
attack (browser binding), consent CSRF and clickjacking headers, redirect URI
and resource manipulation, upstream failures, upstream refresh, the static
bearer next to OAuth, and secrets in logs, audit entries and results.
"""

from __future__ import annotations

import json
import logging
from typing import Any

import pytest
from pytest_httpx import HTTPXMock
from starlette.testclient import TestClient

from aegis_gitea_mcp.oauth.provider import TOKEN_PREFIX
from tests.test_oauth_helpers import (
    API,
    APP_CLIENT_ID,
    APP_CLIENT_SECRET,
    BASE,
    BEARER_TOKEN,
    CLIENT_STATE,
    CSRF_FIELD,
    ENCRYPTION_KEY,
    GITEA,
    GITEA_CODE,
    GITEA_TOKEN,
    REDIRECT,
    RESOURCE,
    TOKEN_URL,
    USER_REFRESH,
    USER_TOKEN,
    browser,
    build_oauth_app,
    exchange_code,
    form_of,
    gitea_requests,
    mcp_client,
    mock_gitea_login,
    payload,
    pkce,
    query_of,
    refresh,
    register,
    sign_in,
    start_flow,
)


@pytest.fixture(autouse=True)
def _env(mock_env: None) -> None:
    """The audit logger and the rate limiter read the process-wide settings."""


def mcp_status(client: TestClient, token: str) -> int:
    """HTTP status of an MCP request with ``token`` (401 when the token is not accepted)."""
    response = client.post("/mcp", json={}, headers={"Authorization": f"Bearer {token}"})
    return response.status_code


def consent_form(client: TestClient) -> tuple[str, str]:
    """Register, call /authorize and open the consent page; return (request id, csrf)."""
    registration = register(client).json()
    authorize = client.get(
        "/authorize",
        params={
            "client_id": registration["client_id"],
            "response_type": "code",
            "code_challenge": pkce()[1],
            "state": CLIENT_STATE,
        },
    )
    consent_url = authorize.headers["location"]
    match = CSRF_FIELD.search(client.get(consent_url).text)
    assert match is not None
    return query_of(consent_url)["request"], match.group(1)


def callback_after_consent(client: TestClient, **params: str) -> Any:
    """Run the flow up to Gitea and call our callback with ``params``."""
    registration = register(client).json()
    gitea_url = start_flow(client, registration, pkce()[1])
    return client.get("/oauth/callback", params={"state": query_of(gitea_url)["state"], **params})


# Discovery and the static bearer in the same app


def test_metadata_documents_and_the_bare_protected_resource_path() -> None:
    client = browser(build_oauth_app().app)

    suffixed = client.get("/.well-known/oauth-protected-resource/mcp")
    bare = client.get("/.well-known/oauth-protected-resource")
    server = client.get("/.well-known/oauth-authorization-server").json()

    assert suffixed.status_code == bare.status_code == 200
    assert bare.json() == suffixed.json()
    assert bare.json()["resource"] == RESOURCE
    assert bare.json()["authorization_servers"] == [BASE]
    assert bare.json()["scopes_supported"] == ["gitea"]
    assert server["issuer"] == BASE
    assert server["registration_endpoint"] == f"{BASE}/register"
    assert server["revocation_endpoint"] == f"{BASE}/revoke"
    assert server["code_challenge_methods_supported"] == ["S256"]


def test_unauthenticated_mcp_request_gets_the_oauth_challenge() -> None:
    response = browser(build_oauth_app().app).post("/mcp", json={})

    assert response.status_code == 401
    assert (
        f'resource_metadata="{BASE}/.well-known/oauth-protected-resource/mcp"'
        in response.headers["WWW-Authenticate"]
    )


async def test_static_bearer_still_runs_as_admin_next_to_oauth(httpx_mock: HTTPXMock) -> None:
    oauth = build_oauth_app()
    httpx_mock.add_response(url=f"{API}/user", json={"login": "root"}, is_reusable=True)
    httpx_mock.add_response(url=f"{API}/repos/octo/demo", json={"full_name": "octo/demo"})

    async with mcp_client(oauth.app, BEARER_TOKEN) as mcp:
        result = await mcp.call_tool("gitea_repo_get", {"owner": "octo", "repo": "demo"})

    assert payload(result)["identity"] == {"login": "root", "mode": "admin-token", "sudo": None}
    (request,) = gitea_requests(httpx_mock, f"{API}/repos/octo/demo")
    assert request.headers["Authorization"] == f"token {GITEA_TOKEN}"


async def test_sudo_from_an_oauth_user_is_sent_with_the_users_own_token(
    httpx_mock: HTTPXMock,
) -> None:
    oauth = build_oauth_app()
    flow = sign_in(browser(oauth.app), httpx_mock)
    httpx_mock.add_response(url=f"{API}/user", status_code=403, json={"message": "no sudo"})

    async with mcp_client(oauth.app, flow.tokens["access_token"]) as mcp:
        result = await mcp.call_tool("gitea_user_get_current", {"sudo": "root"})

    assert result.is_error and payload(result)["error"]["kind"] == "forbidden"
    request = gitea_requests(httpx_mock, f"{API}/user")[-1]
    assert request.headers["Authorization"] == f"Bearer {USER_TOKEN}"
    assert request.headers["Sudo"] == "root"


# Our tokens: expiry, refresh rotation, revocation


def test_expired_access_token_is_rejected_with_401(httpx_mock: HTTPXMock) -> None:
    oauth = build_oauth_app()
    client = browser(oauth.app)
    flow = sign_in(client, httpx_mock)

    oauth.clock.advance(3601)

    assert mcp_status(client, flow.tokens["access_token"]) == 401


def test_refresh_rotates_both_tokens_and_reuse_revokes_the_grant(httpx_mock: HTTPXMock) -> None:
    oauth = build_oauth_app()
    with browser(oauth.app) as client:
        flow = sign_in(client, httpx_mock)
        old_access, old_refresh = flow.tokens["access_token"], flow.tokens["refresh_token"]

        rotated = refresh(client, flow, old_refresh)
        assert rotated.status_code == 200
        new = rotated.json()
        assert new["access_token"] != old_access and new["refresh_token"] != old_refresh
        assert mcp_status(client, old_access) == 401
        assert mcp_status(client, new["access_token"]) != 401

        replay = refresh(client, flow, old_refresh)

        assert replay.status_code == 400 and replay.json()["error"] == "invalid_grant"
        # Reuse of a rotated-out refresh token signals theft: the whole grant ends.
        assert mcp_status(client, new["access_token"]) == 401
        assert refresh(client, flow, new["refresh_token"]).json()["error"] == "invalid_grant"


def test_refresh_honours_scope_rules(httpx_mock: HTTPXMock) -> None:
    oauth = build_oauth_app()
    client = browser(oauth.app)
    flow = sign_in(client, httpx_mock)

    widened = refresh(client, flow, flow.tokens["refresh_token"], scope="gitea admin")
    same = refresh(client, flow, flow.tokens["refresh_token"], scope="gitea")

    assert widened.json()["error"] == "invalid_scope"
    assert same.status_code == 200 and same.json()["scope"] == "gitea"


def test_refresh_token_of_another_client_is_rejected(httpx_mock: HTTPXMock) -> None:
    oauth = build_oauth_app()
    with browser(oauth.app) as client:
        flow = sign_in(client, httpx_mock)
        other = register(client).json()

        response = client.post(
            "/token",
            data={
                "grant_type": "refresh_token",
                "refresh_token": flow.tokens["refresh_token"],
                "client_id": other["client_id"],
                "client_secret": other["client_secret"],
            },
        )

        assert response.json()["error"] == "invalid_grant"
        assert mcp_status(client, flow.tokens["access_token"]) != 401


def test_expired_refresh_token_is_rejected(httpx_mock: HTTPXMock) -> None:
    oauth = build_oauth_app()
    client = browser(oauth.app)
    flow = sign_in(client, httpx_mock)

    oauth.clock.advance(31 * 24 * 3600)

    assert refresh(client, flow, flow.tokens["refresh_token"]).json()["error"] == "invalid_grant"


def test_access_token_is_not_a_refresh_token(httpx_mock: HTTPXMock) -> None:
    oauth = build_oauth_app()
    client = browser(oauth.app)
    flow = sign_in(client, httpx_mock)

    response = refresh(client, flow, flow.tokens["access_token"])

    assert response.json()["error"] == "invalid_grant"
    assert mcp_status(client, flow.tokens["refresh_token"]) == 401


@pytest.mark.parametrize("which", ["access_token", "refresh_token"])
def test_revoking_either_token_ends_the_grant(httpx_mock: HTTPXMock, which: str) -> None:
    oauth = build_oauth_app()
    client = browser(oauth.app)
    flow = sign_in(client, httpx_mock)

    revoked = client.post(
        "/revoke",
        data={
            "token": flow.tokens[which],
            "client_id": flow.client_id,
            "client_secret": flow.client_secret,
        },
    )

    assert revoked.status_code == 200
    assert mcp_status(client, flow.tokens["access_token"]) == 401
    assert refresh(client, flow, flow.tokens["refresh_token"]).json()["error"] == "invalid_grant"
    assert oauth.store.sessions == {}


def test_another_client_cannot_revoke_a_grant(httpx_mock: HTTPXMock) -> None:
    oauth = build_oauth_app()
    with browser(oauth.app) as client:
        flow = sign_in(client, httpx_mock)
        other = register(client).json()

        client.post(
            "/revoke",
            data={
                "token": flow.tokens["access_token"],
                "client_id": other["client_id"],
                "client_secret": other["client_secret"],
            },
        )

        assert mcp_status(client, flow.tokens["access_token"]) != 401


# Authorization codes and states


def test_replayed_code_is_rejected_and_revokes_the_tokens_it_produced(
    httpx_mock: HTTPXMock,
) -> None:
    oauth = build_oauth_app()
    client = browser(oauth.app)
    flow = sign_in(client, httpx_mock)
    registration = {"client_id": flow.client_id, "client_secret": flow.client_secret}

    replay = exchange_code(client, registration, flow.code, flow.verifier)

    assert replay.status_code == 400 and replay.json()["error"] == "invalid_grant"
    assert mcp_status(client, flow.tokens["access_token"]) == 401
    assert '"oauth_code_replay"' in oauth.settings.resolved_audit_log_path.read_text("utf-8")


def test_code_needs_the_right_pkce_verifier_and_client(httpx_mock: HTTPXMock) -> None:
    oauth = build_oauth_app()
    client = browser(oauth.app)
    registration = register(client).json()
    other = register(client).json()
    verifier, challenge = pkce()
    gitea_url = start_flow(client, registration, challenge)
    mock_gitea_login(httpx_mock)
    callback = client.get(
        "/oauth/callback", params={"code": GITEA_CODE, "state": query_of(gitea_url)["state"]}
    )
    code = query_of(callback.headers["location"])["code"]

    wrong_verifier = exchange_code(client, registration, code, pkce()[0])
    wrong_client = exchange_code(client, other, code, verifier)
    right = exchange_code(client, registration, code, verifier)

    assert wrong_verifier.json()["error"] == "invalid_grant"
    assert wrong_client.json()["error"] == "invalid_grant"
    assert right.status_code == 200


def test_expired_code_is_rejected(httpx_mock: HTTPXMock) -> None:
    oauth = build_oauth_app()
    client = browser(oauth.app)
    registration = register(client).json()
    verifier, challenge = pkce()
    gitea_url = start_flow(client, registration, challenge)
    mock_gitea_login(httpx_mock)
    callback = client.get(
        "/oauth/callback", params={"code": GITEA_CODE, "state": query_of(gitea_url)["state"]}
    )
    code = query_of(callback.headers["location"])["code"]

    oauth.clock.advance(301)

    assert exchange_code(client, registration, code, verifier).json()["error"] == "invalid_grant"


@pytest.mark.parametrize("state", [None, "", "made-up-state", "x" * 500])
def test_unknown_state_is_a_generic_400(state: str | None) -> None:
    client = browser(build_oauth_app().app)
    params = {"code": GITEA_CODE}
    if state is not None:
        params["state"] = state

    response = client.get("/oauth/callback", params=params)

    assert response.status_code == 400
    assert "location" not in response.headers
    assert "made-up-state" not in response.text and GITEA_CODE not in response.text
    assert response.headers["Cache-Control"] == "no-store"


def test_expired_and_replayed_states_are_rejected(httpx_mock: HTTPXMock) -> None:
    oauth = build_oauth_app()
    client = browser(oauth.app)
    registration = register(client).json()

    expired_url = start_flow(client, registration, pkce()[1])
    oauth.clock.advance(601)
    expired = client.get(
        "/oauth/callback", params={"code": GITEA_CODE, "state": query_of(expired_url)["state"]}
    )
    gitea_url = start_flow(client, registration, pkce()[1])
    mock_gitea_login(httpx_mock)
    state = query_of(gitea_url)["state"]
    first = client.get("/oauth/callback", params={"code": GITEA_CODE, "state": state})
    replayed = client.get("/oauth/callback", params={"code": GITEA_CODE, "state": state})

    assert expired.status_code == 400
    assert first.status_code == 302
    assert replayed.status_code == 400
    assert len(gitea_requests(httpx_mock, TOKEN_URL)) == 1


def test_callback_in_a_browser_that_did_not_start_the_flow_is_refused(
    httpx_mock: HTTPXMock,
) -> None:
    """Confused deputy: an attacker's Gitea URL opened by a victim must not mint a code."""
    oauth = build_oauth_app()
    attacker = browser(oauth.app)
    registration = register(attacker, redirect_uris=["https://evil.example/cb"]).json()
    authorize = attacker.get(
        "/authorize",
        params={
            "client_id": registration["client_id"],
            "response_type": "code",
            "code_challenge": pkce()[1],
        },
    )
    consent_url = authorize.headers["location"]
    match = CSRF_FIELD.search(attacker.get(consent_url).text)
    assert match is not None
    approved = attacker.post(
        "/oauth/consent",
        data={
            "request": query_of(consent_url)["request"],
            "csrf": match.group(1),
            "decision": "approve",
        },
        headers={"Origin": BASE},
    )
    state = query_of(approved.headers["location"])["state"]

    victim = browser(oauth.app)  # no cookies; Gitea silently redirects the victim here
    response = victim.get("/oauth/callback", params={"code": GITEA_CODE, "state": state})

    assert response.status_code == 400
    assert "location" not in response.headers
    assert gitea_requests(httpx_mock, GITEA) == []
    # The state is spent: the attacker cannot complete it either.
    retry = attacker.get("/oauth/callback", params={"code": GITEA_CODE, "state": state})
    assert retry.status_code == 400
    audit = oauth.settings.resolved_audit_log_path.read_text("utf-8")
    assert '"oauth_callback_binding_mismatch"' in audit


def test_consent_page_shows_the_client_and_destination_escaped() -> None:
    client = browser(build_oauth_app().app)
    registration = register(client, client_name="<script>alert(1)</script>").json()
    authorize = client.get(
        "/authorize",
        params={
            "client_id": registration["client_id"],
            "response_type": "code",
            "code_challenge": pkce()[1],
        },
    )

    page = client.get(authorize.headers["location"])

    assert page.status_code == 200
    assert "<script>" not in page.text and "&lt;script&gt;" in page.text
    assert "localhost:3000" in page.text
    assert page.headers["X-Frame-Options"] == "DENY"
    assert "frame-ancestors 'none'" in page.headers["Content-Security-Policy"]
    assert page.headers["Cache-Control"] == "no-store"
    cookie = page.headers["set-cookie"].lower()
    assert cookie.startswith("__host-aegis-oauth-csrf=")
    assert "samesite=strict" in cookie and "secure" in cookie and "httponly" in cookie


def test_consent_post_needs_the_csrf_cookie_token_and_origin() -> None:
    oauth = build_oauth_app()
    client = browser(oauth.app)
    request_id, csrf = consent_form(client)
    form = {"request": request_id, "csrf": csrf, "decision": "approve"}

    origin = {"Origin": BASE}
    cross_site = browser(oauth.app).post("/oauth/consent", data=form, headers=origin)  # no cookie
    wrong_token = client.post("/oauth/consent", data={**form, "csrf": "nope"}, headers=origin)
    bad_origin = client.post(
        "/oauth/consent", data=form, headers={"Origin": "https://evil.example"}
    )
    unknown = client.post("/oauth/consent", data={**form, "request": "unknown"}, headers=origin)
    good = client.post("/oauth/consent", data=form, headers=origin)
    again = client.post("/oauth/consent", data=form, headers=origin)

    assert [r.status_code for r in (cross_site, wrong_token, bad_origin, unknown)] == [400] * 4
    assert good.status_code == 303
    assert again.status_code == 400  # single use
    binding = good.headers["set-cookie"].lower()
    assert "__host-aegis-oauth-flow-" in binding and "samesite=lax" in binding


def test_unknown_consent_request_page_is_a_400() -> None:
    client = browser(build_oauth_app().app)

    assert client.get("/oauth/consent", params={"request": "nope"}).status_code == 400
    assert client.get("/oauth/consent").status_code == 400


def test_denying_consent_returns_access_denied_to_the_client() -> None:
    client = browser(build_oauth_app().app)
    request_id, csrf = consent_form(client)

    denied = client.post(
        "/oauth/consent",
        data={"request": request_id, "csrf": csrf, "decision": "deny"},
        headers={"Origin": BASE},
    )

    assert denied.status_code == 303
    assert denied.headers["location"].startswith(REDIRECT)
    assert query_of(denied.headers["location"]) == {"error": "access_denied", "state": CLIENT_STATE}


def test_gitea_error_redirects_access_denied() -> None:
    client = browser(build_oauth_app().app)

    response = callback_after_consent(client, error="access_denied")

    assert response.status_code == 302
    assert query_of(response.headers["location"]) == {
        "error": "access_denied",
        "state": CLIENT_STATE,
    }


@pytest.mark.parametrize("case", ["token_500", "user_401", "no_code"])
def test_upstream_failures_redirect_server_error_without_detail(
    httpx_mock: HTTPXMock, case: str
) -> None:
    oauth = build_oauth_app()
    client = browser(oauth.app)
    if case == "token_500":
        httpx_mock.add_response(
            method="POST", url=TOKEN_URL, status_code=500, text="secret upstream detail"
        )
    elif case == "user_401":
        httpx_mock.add_response(
            method="POST", url=TOKEN_URL, json={"access_token": USER_TOKEN, "token_type": "bearer"}
        )
        httpx_mock.add_response(url=f"{API}/user", status_code=401)

    response = callback_after_consent(client, code="" if case == "no_code" else GITEA_CODE)

    location = response.headers["location"]
    assert query_of(location) == {"error": "server_error", "state": CLIENT_STATE}
    assert "secret upstream detail" not in location
    assert oauth.store.sessions == {}


def test_authorize_with_an_unregistered_redirect_uri_never_redirects() -> None:
    client = browser(build_oauth_app().app)
    registration = register(client).json()

    response = client.get(
        "/authorize",
        params={
            "client_id": registration["client_id"],
            "redirect_uri": "https://evil.example/cb",
            "response_type": "code",
            "code_challenge": pkce()[1],
        },
    )

    assert response.status_code == 400
    assert "location" not in response.headers


def test_authorize_for_another_resource_is_invalid_target() -> None:
    client = browser(build_oauth_app().app)
    registration = register(client).json()

    response = client.get(
        "/authorize",
        params={
            "client_id": registration["client_id"],
            "response_type": "code",
            "code_challenge": pkce()[1],
            "state": CLIENT_STATE,
            "resource": "https://other.example/mcp",
        },
    )

    assert response.status_code == 302
    assert query_of(response.headers["location"])["error"] == "invalid_target"


@pytest.mark.parametrize(
    "uri",
    [
        "http://evil.example/cb",
        "javascript:alert(1)",
        "https://user:pw@evil.example/cb",
        "https://claude.ai/cb#frag",
        "custom-scheme://callback",
    ],
)
def test_registration_rejects_unsafe_redirect_uris(uri: str) -> None:
    response = register(browser(build_oauth_app().app), redirect_uris=[uri])

    assert response.status_code == 400
    assert response.json()["error"] in {"invalid_redirect_uri", "invalid_client_metadata"}


@pytest.mark.parametrize(
    "uri",
    [
        "https://claude.ai/api/mcp/auth_callback",
        "http://localhost:51234/callback",
        "http://127.0.0.1:8765/callback",
        "http://[::1]:9000/cb",
    ],
)
def test_registration_accepts_https_and_loopback_redirect_uris(uri: str) -> None:
    response = register(browser(build_oauth_app().app), redirect_uris=[uri])

    assert response.status_code == 201
    assert response.json()["client_secret"]


# Upstream (Gitea) token refresh


async def test_gitea_token_near_expiry_is_refreshed_before_the_call(httpx_mock: HTTPXMock) -> None:
    oauth = build_oauth_app()
    flow = sign_in(browser(oauth.app), httpx_mock, expires_in=30)
    httpx_mock.add_response(
        method="POST",
        url=TOKEN_URL,
        json={
            "access_token": "gitea-user-access-token-NEW-0123456789",
            "refresh_token": "gitea-user-refresh-token-NEW-0123456789",
            "expires_in": 3600,
            "token_type": "bearer",
        },
    )
    httpx_mock.add_response(url=f"{API}/repos/octo/demo", json={"full_name": "octo/demo"})

    async with mcp_client(oauth.app, flow.tokens["access_token"]) as mcp:
        result = await mcp.call_tool("gitea_repo_get", {"owner": "octo", "repo": "demo"})

    assert not result.is_error
    refresh_request = gitea_requests(httpx_mock, TOKEN_URL)[-1]
    assert form_of(refresh_request) == {
        "grant_type": "refresh_token",
        "refresh_token": USER_REFRESH,
        "client_id": APP_CLIENT_ID,
        "client_secret": APP_CLIENT_SECRET,
    }
    (repo_request,) = gitea_requests(httpx_mock, f"{API}/repos/octo/demo")
    assert repo_request.headers["Authorization"] == "Bearer gitea-user-access-token-NEW-0123456789"
    (session,) = oauth.store.sessions.values()
    assert session.gitea_refresh_token == "gitea-user-refresh-token-NEW-0123456789"


async def test_gitea_invalid_grant_ends_the_session_without_admin_fallback(
    httpx_mock: HTTPXMock,
) -> None:
    oauth = build_oauth_app()
    client = browser(oauth.app)
    flow = sign_in(client, httpx_mock, expires_in=30)
    httpx_mock.add_response(
        method="POST", url=TOKEN_URL, status_code=400, json={"error": "invalid_grant"}
    )

    async with mcp_client(oauth.app, flow.tokens["access_token"]) as mcp:
        result = await mcp.call_tool("gitea_repo_get", {"owner": "octo", "repo": "demo"})

    assert result.is_error
    error = payload(result)["error"]
    assert error["kind"] == "identity_error"
    assert "sign in again" in error["message"]
    assert gitea_requests(httpx_mock, f"{API}/repos") == []
    assert oauth.store.sessions == {} and oauth.store.tokens == {}
    assert mcp_status(client, flow.tokens["access_token"]) == 401


# Secrets never leak


async def test_no_secret_reaches_logs_audit_or_results(
    httpx_mock: HTTPXMock, caplog: pytest.LogCaptureFixture
) -> None:
    caplog.set_level(logging.DEBUG)
    oauth = build_oauth_app()
    client = browser(oauth.app)
    flow = sign_in(client, httpx_mock, expires_in=30)
    refreshed = "gitea-refreshed-token-0123456789abcdef"
    httpx_mock.add_response(
        method="POST", url=TOKEN_URL, json={"access_token": refreshed, "expires_in": 3600}
    )
    leak = f"{USER_REFRESH} {ENCRYPTION_KEY} {APP_CLIENT_SECRET} {refreshed}"
    httpx_mock.add_response(url=f"{API}/repos/octo/demo", json={"description": leak})

    async with mcp_client(oauth.app, flow.tokens["access_token"]) as mcp:
        result = await mcp.call_tool("gitea_repo_get", {"owner": "octo", "repo": "demo"})
    rotated = refresh(client, flow, flow.tokens["refresh_token"]).json()
    client.post(
        "/revoke",
        data={
            "token": rotated["access_token"],
            "client_id": flow.client_id,
            "client_secret": flow.client_secret,
        },
    )

    text = result.content[0].text  # type: ignore[union-attr]
    for secret in (USER_REFRESH, ENCRYPTION_KEY, APP_CLIENT_SECRET, refreshed):
        assert secret not in text
    assert text.count("[REDACTED]") >= 4
    secrets = [
        USER_TOKEN,
        USER_REFRESH,
        refreshed,
        GITEA_CODE,
        ENCRYPTION_KEY,
        APP_CLIENT_SECRET,
        GITEA_TOKEN,
        BEARER_TOKEN,
        flow.client_secret,
        flow.code,
        flow.verifier,
        flow.tokens["access_token"],
        flow.tokens["refresh_token"],
        rotated["access_token"],
        rotated["refresh_token"],
        query_of(flow.gitea_url)["state"],
        query_of(flow.gitea_url)["code_challenge"],
    ]
    # Lines the test's own HTTP client logs about its requests to our app are not server logs.
    logs = "\n".join(
        record.getMessage()
        for record in caplog.records
        if not (record.name in {"httpx", "httpx2"} and BASE in record.getMessage())
    )
    assert "oauth sign-in: Gitea user alice" in logs
    audit = oauth.settings.resolved_audit_log_path.read_text(encoding="utf-8")
    assert '"oauth_login"' in audit and '"oauth_revoke"' in audit
    for secret in secrets:
        assert secret not in logs, secret
        assert secret not in audit, secret
    assert TOKEN_PREFIX not in json.dumps([vars(token) for token in oauth.store.tokens.values()])


def test_oauth_endpoints_are_rate_limited(monkeypatch: pytest.MonkeyPatch) -> None:
    monkeypatch.setenv("RATE_LIMIT_PER_MINUTE", "2")
    client = browser(build_oauth_app().app)

    statuses = [register(client).status_code for _ in range(3)]

    assert statuses == [201, 201, 429]