diff --git a/tests/integration/test_per_user_oauth.py b/tests/integration/test_per_user_oauth.py new file mode 100644 index 0000000..6e7cb3e --- /dev/null +++ b/tests/integration/test_per_user_oauth.py @@ -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