112 lines
4.0 KiB
Python
112 lines
4.0 KiB
Python
"""End-to-end per-user OAuth over the real HTTP app: DCR, consent, Gitea, tokens, MCP calls."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import pytest
|
|
from pytest_httpx import HTTPXMock
|
|
|
|
from aegis_gitea_mcp.oauth.provider import TOKEN_PREFIX
|
|
from tests.test_oauth_helpers import (
|
|
API,
|
|
APP_CLIENT_ID,
|
|
APP_CLIENT_SECRET,
|
|
BASE,
|
|
CLIENT_STATE,
|
|
GITEA,
|
|
GITEA_CODE,
|
|
GITEA_TOKEN,
|
|
REDIRECT,
|
|
TOKEN_URL,
|
|
USER_TOKEN,
|
|
browser,
|
|
build_oauth_app,
|
|
exchange_code,
|
|
form_of,
|
|
gitea_requests,
|
|
mcp_client,
|
|
mock_gitea_login,
|
|
payload,
|
|
pkce,
|
|
query_of,
|
|
register,
|
|
start_flow,
|
|
)
|
|
|
|
|
|
@pytest.fixture(autouse=True)
|
|
def _env(mock_env: None) -> None:
|
|
"""The audit logger and the rate limiter read the process-wide settings."""
|
|
|
|
|
|
async def test_full_flow_runs_tool_calls_as_the_signed_in_user(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)
|
|
|
|
# Our server sends the browser to Gitea with its own PKCE, callback and scopes.
|
|
assert gitea_url.startswith(f"{GITEA}/login/oauth/authorize?")
|
|
upstream = query_of(gitea_url)
|
|
assert upstream["client_id"] == APP_CLIENT_ID
|
|
assert upstream["redirect_uri"] == f"{BASE}/oauth/callback"
|
|
assert upstream["response_type"] == "code"
|
|
assert upstream["code_challenge_method"] == "S256"
|
|
assert upstream["code_challenge"] != challenge # ours, not the client's
|
|
assert upstream["scope"] == " ".join(oauth.settings.oauth_scopes)
|
|
assert upstream["state"] != CLIENT_STATE
|
|
|
|
mock_gitea_login(httpx_mock, login="alice")
|
|
callback = client.get(
|
|
"/oauth/callback", params={"code": GITEA_CODE, "state": upstream["state"]}
|
|
)
|
|
assert callback.status_code == 302
|
|
location = callback.headers["location"]
|
|
assert location.startswith(f"{REDIRECT}?")
|
|
back = query_of(location)
|
|
assert back["state"] == CLIENT_STATE and back["code"]
|
|
|
|
# Gitea received the code with our verifier and the app credentials.
|
|
(token_request,) = gitea_requests(httpx_mock, TOKEN_URL)
|
|
form = form_of(token_request)
|
|
assert form["grant_type"] == "authorization_code"
|
|
assert form["code"] == GITEA_CODE
|
|
assert form["client_id"] == APP_CLIENT_ID
|
|
assert form["client_secret"] == APP_CLIENT_SECRET
|
|
assert form["redirect_uri"] == f"{BASE}/oauth/callback"
|
|
assert form["code_verifier"]
|
|
(user_request,) = gitea_requests(httpx_mock, f"{API}/user")
|
|
assert user_request.headers["Authorization"] == f"Bearer {USER_TOKEN}"
|
|
|
|
tokens = exchange_code(client, registration, back["code"], verifier)
|
|
assert tokens.status_code == 200
|
|
body = tokens.json()
|
|
assert body["token_type"] == "Bearer"
|
|
assert body["expires_in"] == 3600
|
|
assert body["scope"] == "gitea"
|
|
assert body["access_token"].startswith(TOKEN_PREFIX)
|
|
assert body["refresh_token"].startswith(TOKEN_PREFIX)
|
|
|
|
httpx_mock.add_response(url=f"{API}/repos/octo/demo", json={"full_name": "octo/demo"})
|
|
httpx_mock.add_response(
|
|
url=f"{API}/admin/users", status_code=403, json={"message": "Only admins"}
|
|
)
|
|
async with mcp_client(oauth.app, body["access_token"]) as mcp:
|
|
result = await mcp.call_tool("gitea_repo_get", {"owner": "octo", "repo": "demo"})
|
|
admin = await mcp.call_tool("gitea_admin_search_users", {})
|
|
|
|
assert not result.is_error
|
|
envelope = payload(result)
|
|
assert envelope["identity"] == {"login": "alice", "mode": "oauth", "sudo": None}
|
|
(repo_request,) = gitea_requests(httpx_mock, f"{API}/repos/octo/demo")
|
|
assert repo_request.headers["Authorization"] == f"Bearer {USER_TOKEN}"
|
|
|
|
assert admin.is_error
|
|
error = payload(admin)
|
|
assert error["error"]["kind"] == "forbidden"
|
|
assert error["identity"]["mode"] == "oauth"
|
|
(admin_request,) = gitea_requests(httpx_mock, f"{API}/admin/users")
|
|
assert admin_request.headers["Authorization"] == f"Bearer {USER_TOKEN}"
|
|
assert GITEA_TOKEN not in admin_request.headers["Authorization"]
|