test: per-user OAuth mode against the live instance

2026-10-14 17:57:54 +02:00
1 parent 303bd9aa21
commit fed75b17c8
@@ -0,0 +1,266 @@
+"""Per-user mode against the live instance: calls run as the signed-in Gitea user.
+
+The OAuth store is seeded directly (a session for a non-admin user whose Gitea token is a
+personal access token, plus our own issued access token) instead of running the browser
+flow, then the real HTTP app is driven in process with the MCP SDK client.
+"""
+
+from __future__ import annotations
+
+import secrets
+import time
+from collections.abc import AsyncIterator
+from contextlib import asynccontextmanager
+from dataclasses import dataclass
+from pathlib import Path
+from typing import Any
+
+import httpx2
+import pytest
+from mcp import Client
+from mcp.client.streamable_http import streamable_http_client
+from starlette.applications import Starlette
+
+from aegis_gitea_mcp.audit import AuditLogger
+from aegis_gitea_mcp.auth import MCP_SCOPE
+from aegis_gitea_mcp.config import Settings
+from aegis_gitea_mcp.oauth.components import build_oauth_components
+from aegis_gitea_mcp.oauth.provider import ACCESS_TOKEN_TTL, TOKEN_PREFIX
+from aegis_gitea_mcp.oauth.store import OAuthStore, SessionRecord, TokenRecord, token_hash
+from aegis_gitea_mcp.server import build_server
+from aegis_gitea_mcp.transports import build_http_app
+from tests.integration.conftest import McpSession, RegularUser
+from tests.integration.harness import GiteaInstance
+
+PUBLIC_BASE = "https://mcp.example.test"
+RESOURCE = f"{PUBLIC_BASE}/mcp"
+ENCRYPTION_KEY = secrets.token_urlsafe(48)
+OAUTH_CLIENT_ID = "it-oauth-client"
+
+
+@dataclass
+class OAuthStack:
+    """The HTTP app with its seeded tokens and the secrets the tests must never see."""
+
+    app: Starlette
+    audit_path: Path
+    bearer: str
+    user_login: str
+    user_access_token: str
+    secrets: list[str]
+
+
+async def build_stack(
+    gitea: GiteaInstance, user: RegularUser, tmp_path: Path, monkeypatch: pytest.MonkeyPatch
+) -> OAuthStack:
+    """Build the app with per-user OAuth enabled and a signed-in session for ``user``."""
+    monkeypatch.setenv("GITEA_URL", gitea.base_url)
+    data_dir = tmp_path / "oauth-state"
+    bearer = secrets.token_urlsafe(36)
+    oauth_client_secret = secrets.token_urlsafe(36)
+    settings = Settings(
+        gitea_url=gitea.base_url,
+        gitea_token=gitea.admin_token,
+        mcp_bearer_token=bearer,
+        oauth_enabled=True,
+        gitea_oauth_client_id=OAUTH_CLIENT_ID,
+        gitea_oauth_client_secret=oauth_client_secret,
+        token_encryption_key=ENCRYPTION_KEY,
+        public_base_url=PUBLIC_BASE,
+        data_dir=data_dir,
+        audit_log_path=data_dir / "audit.log",
+        request_timeout_seconds=60,
+    )
+
+    # Seed the encrypted store: one session (the user's own Gitea token) and one of our
+    # access tokens (stored only as a hash), exactly what the browser flow would leave.
+    store = OAuthStore.load(data_dir, ENCRYPTION_KEY)
+    session_id = secrets.token_hex(16)
+    store.sessions[session_id] = SessionRecord(
+        session_id=session_id,
+        client_id=OAUTH_CLIENT_ID,
+        login=user.login,
+        is_admin=False,
+        gitea_access_token=user.token,
+        gitea_refresh_token=None,
+        gitea_expires_at=None,
+        scopes=[MCP_SCOPE],
+        resource=RESOURCE,
+        created_at=time.time(),
+    )
+    access = TOKEN_PREFIX + secrets.token_urlsafe(32)
+    store.tokens[token_hash(access)] = TokenRecord(
+        kind="access",
+        session_id=session_id,
+        client_id=OAUTH_CLIENT_ID,
+        scopes=[MCP_SCOPE],
+        expires_at=time.time() + ACCESS_TOKEN_TTL,
+    )
+    await store.save()
+
+    audit = AuditLogger(settings.resolved_audit_log_path)
+    extras = build_oauth_components(settings, audit=audit)
+    server = build_server(settings, resolver=extras.resolver, executor=extras.executor, audit=audit)
+    app = build_http_app(
+        settings,
+        server=server,
+        token_verifier=extras.token_verifier,
+        auth_server_provider=extras.auth_server_provider,
+        extra_routes=extras.extra_routes,
+        client_registration_options=extras.client_registration_options,
+        revocation_options=extras.revocation_options,
+        rate_limited_paths=extras.rate_limited_paths,
+    )
+    return OAuthStack(
+        app=app,
+        audit_path=settings.resolved_audit_log_path,
+        bearer=bearer,
+        user_login=user.login,
+        user_access_token=access,
+        secrets=[
+            bearer,
+            access,
+            user.token,
+            gitea.admin_token,
+            oauth_client_secret,
+            ENCRYPTION_KEY,
+        ],
+    )
+
+
+@asynccontextmanager
+async def http_session(app: Starlette, token: str, audit_path: Path) -> AsyncIterator[McpSession]:
+    """An MCP session over the real HTTP app (in process) with ``token`` as bearer."""
+    async with (
+        app.router.lifespan_context(app),
+        httpx2.AsyncClient(
+            transport=httpx2.ASGITransport(app=app),
+            headers={"Authorization": f"Bearer {token}"},
+            base_url=PUBLIC_BASE,
+            timeout=60.0,
+        ) as http_client,
+        Client(streamable_http_client(RESOURCE, http_client=http_client)) as client,
+    ):
+        yield McpSession(client, audit_path)
+
+
+@pytest.fixture
+async def stack(
+    gitea: GiteaInstance,
+    regular_user: RegularUser,
+    tmp_path: Path,
+    monkeypatch: pytest.MonkeyPatch,
+) -> OAuthStack:
+    return await build_stack(gitea, regular_user, tmp_path, monkeypatch)
+
+
+async def test_calls_run_as_the_signed_in_user(
+    gitea: GiteaInstance, regular_user: RegularUser, stack: OAuthStack
+) -> None:
+    private_repo = gitea.unique("it-private")
+    created = gitea.api(
+        "POST",
+        f"/admin/users/{gitea.admin_login}/repos",
+        json={"name": private_repo, "private": True, "auto_init": True},
+    )
+    assert created.status_code == 201
+    try:
+        async with http_session(stack.app, stack.user_access_token, stack.audit_path) as mcp:
+            me = await mcp.ok("gitea_user_get_current")
+            assert me.identity == {"login": regular_user.login, "mode": "oauth", "sudo": None}
+            assert me.data["login"] == regular_user.login
+            assert me.data["is_admin"] is False
+
+            # The user can create and read their own repository ...
+            own = gitea.unique("it-own")
+            await mcp.ok("gitea_user_current_create_repo", status=201, body={"name": own})
+            assert (await mcp.ok("gitea_repo_get", owner=regular_user.login, repo=own)).data[
+                "owner"
+            ]["login"] == regular_user.login
+
+            # ... but not another user's private repository: Gitea hides it.
+            hidden = await mcp.call("gitea_repo_get", owner=gitea.admin_login, repo=private_repo)
+            assert hidden.is_error
+            assert hidden.status in (403, 404)
+            assert hidden.error_kind in ("forbidden", "not_found")
+            assert (
+                await mcp.call(
+                    "gitea_repo_get_contents",
+                    owner=gitea.admin_login,
+                    repo=private_repo,
+                    filepath="README.md",
+                )
+            ).status in (403, 404)
+
+            # Admin tools and sudo are refused by Gitea with 403, not by the server.
+            admin_only = await mcp.call("gitea_admin_cron_list")
+            assert admin_only.is_error and admin_only.status == 403
+            assert admin_only.error_kind == "forbidden"
+            assert admin_only.identity["mode"] == "oauth"
+            sudo = await mcp.call("gitea_user_get_current", sudo=gitea.admin_login)
+            assert sudo.is_error and sudo.status == 403
+
+            entries = mcp.audit_entries()
+            assert entries and all(e["identity_mode"] == "oauth" for e in entries)
+            assert all(e["identity_login"] == regular_user.login for e in entries)
+            assert {e["tool_name"] for e in entries} >= {
+                "gitea_user_get_current",
+                "gitea_admin_cron_list",
+            }
+            gitea.api("DELETE", f"/repos/{regular_user.login}/{own}")
+
+            texts = "\n".join(mcp.texts) + stack.audit_path.read_text(encoding="utf-8")
+            for secret in stack.secrets:
+                assert secret not in texts
+    finally:
+        gitea.api("DELETE", f"/repos/{gitea.admin_login}/{private_repo}")
+
+
+async def test_static_bearer_in_the_same_app_runs_as_admin(
+    gitea: GiteaInstance, stack: OAuthStack
+) -> None:
+    async with http_session(stack.app, stack.bearer, stack.audit_path) as mcp:
+        me = await mcp.ok("gitea_user_get_current")
+        assert me.identity == {"login": gitea.admin_login, "mode": "admin-token", "sudo": None}
+        assert me.data["is_admin"] is True
+        cron = await mcp.ok("gitea_admin_cron_list")
+        assert cron.data
+        entries = mcp.audit_entries()
+        assert {e["identity_mode"] for e in entries} == {"admin-token"}
+        texts = "\n".join(mcp.texts)
+        for secret in stack.secrets:
+            assert secret not in texts
+
+
+async def test_unknown_and_expired_tokens_never_reach_gitea(
+    stack: OAuthStack, regular_user: RegularUser
+) -> None:
+    body: dict[str, Any] = {
+        "jsonrpc": "2.0",
+        "id": 1,
+        "method": "initialize",
+        "params": {
+            "protocolVersion": "2025-11-25",
+            "capabilities": {},
+            "clientInfo": {"name": "it", "version": "0"},
+        },
+    }
+    headers = {"Accept": "application/json, text/event-stream"}
+    async with (
+        stack.app.router.lifespan_context(stack.app),
+        httpx2.AsyncClient(
+            transport=httpx2.ASGITransport(app=stack.app), base_url=PUBLIC_BASE
+        ) as http_client,
+    ):
+        for token in (
+            TOKEN_PREFIX + secrets.token_urlsafe(32),  # looks like ours, unknown
+            regular_user.token,  # the user's own Gitea token is not an MCP token
+            secrets.token_urlsafe(36),
+        ):
+            response = await http_client.post(
+                "/mcp", json=body, headers={**headers, "Authorization": f"Bearer {token}"}
+            )
+            assert response.status_code == 401
+            assert token not in response.text
+        anonymous = await http_client.post("/mcp", json=body, headers=headers)
+        assert anonymous.status_code == 401