267 lines
10 KiB
Python
267 lines
10 KiB
Python
"""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
|