From c79cc1ab9e5373650231d3b352a4638dcbad6f32 Mon Sep 17 00:00:00 2001 From: latte Date: Wed, 25 Feb 2026 20:49:08 +0100 Subject: [PATCH] Add PUBLIC_BASE_URL and refine OAuth scopes --- .claude/settings.local.json | 4 +- .env.example | 3 + docs/configuration.md | 1 + docs/deployment.md | 1 + docs/troubleshooting.md | 75 +++++++++++ openapi-gpt.yaml | 7 +- src/aegis_gitea_mcp/config.py | 22 ++++ src/aegis_gitea_mcp/server.py | 207 +++++++++++++++++++++++++++++- tests/test_server.py | 232 ++++++++++++++++++++++++++++++++++ 9 files changed, 541 insertions(+), 11 deletions(-) create mode 100644 docs/troubleshooting.md diff --git a/.claude/settings.local.json b/.claude/settings.local.json index 227e67a..4d5d448 100644 --- a/.claude/settings.local.json +++ b/.claude/settings.local.json @@ -2,7 +2,9 @@ "permissions": { "allow": [ "Bash(python -m pytest:*)", - "Bash(python:*)" + "Bash(python:*)", + "Bash(docker compose:*)", + "Bash(findstr:*)" ] } } diff --git a/.env.example b/.env.example index d866ebe..8267978 100644 --- a/.env.example +++ b/.env.example @@ -16,6 +16,9 @@ OAUTH_CACHE_TTL_SECONDS=300 # MCP server configuration MCP_HOST=127.0.0.1 MCP_PORT=8080 +# Optional external URL used in OAuth metadata when running behind reverse proxies. +# Example: PUBLIC_BASE_URL=https://gitea-mcp.hiddenden.cafe +PUBLIC_BASE_URL= ALLOW_INSECURE_BIND=false # Logging / observability diff --git a/docs/configuration.md b/docs/configuration.md index 0e91104..7300ce1 100644 --- a/docs/configuration.md +++ b/docs/configuration.md @@ -23,6 +23,7 @@ cp .env.example .env |---|---|---|---| | `MCP_HOST` | No | `127.0.0.1` | Interface to bind to | | `MCP_PORT` | No | `8080` | Port to listen on | +| `PUBLIC_BASE_URL` | No | empty | Public HTTPS base URL advertised in OAuth metadata (recommended behind reverse proxy) | | `ALLOW_INSECURE_BIND` | No | `false` | Explicit opt-in required for `0.0.0.0` bind | | `LOG_LEVEL` | No | `INFO` | `DEBUG`, `INFO`, `WARNING`, `ERROR`, `CRITICAL` | | `STARTUP_VALIDATE_GITEA` | No | `true` | Validate OIDC discovery endpoint at startup | diff --git a/docs/deployment.md b/docs/deployment.md index 60679e5..22214ca 100644 --- a/docs/deployment.md +++ b/docs/deployment.md @@ -44,6 +44,7 @@ Workflows live in `.gitea/workflows/`: ## Production Recommendations - Place MCP behind TLS reverse proxy. +- Set `PUBLIC_BASE_URL=https://` so OAuth metadata advertises HTTPS endpoints. - Restrict inbound traffic to expected clients. - Persist and monitor audit logs. - Monitor `/metrics` and auth-failure events. diff --git a/docs/troubleshooting.md b/docs/troubleshooting.md new file mode 100644 index 0000000..45e6a33 --- /dev/null +++ b/docs/troubleshooting.md @@ -0,0 +1,75 @@ +# Troubleshooting + +## "Internal server error (-32603)" from ChatGPT + +**Symptom:** ChatGPT shows `Internal server error` with JSON-RPC error code `-32603` when trying to use Gitea tools. + +**Cause:** The OAuth token stored by ChatGPT was issued without Gitea API scopes (e.g. `read:repository`). This happens when the initial authorization request didn't include the correct `scope` parameter. The token passes OIDC validation (openid/profile/email) but gets **403 Forbidden** from Gitea's REST API. + +**Fix:** +1. In Gitea: Go to **Settings > Applications > Authorized OAuth2 Applications** and revoke the MCP application. +2. In ChatGPT: Go to **Settings > Connected apps** and disconnect the Gitea integration. +3. Re-authorize: Use the ChatGPT integration again. It will trigger a fresh OAuth flow with the correct scopes (`read:repository`). + +**Verification:** Check the server logs for `oauth_auth_summary`. A working token shows: +``` +oauth_auth_summary: api_probe=pass login=alice +``` +A scopeless token shows: +``` +oauth_token_lacks_api_scope: status=403 login=alice +``` + +## "Gitea rejected the API call" (403) + +**Symptom:** Tool calls return 403 with a message about re-authorizing. + +**Cause:** Same root cause as above — the OAuth token doesn't have the required Gitea API scopes. The middleware's API scope probe detected this and returned a clear error instead of letting it fail deep in the tool handler. + +**Fix:** Same as above — revoke and re-authorize. + +## ChatGPT caches stale tokens + +**Symptom:** After fixing the OAuth configuration, ChatGPT still sends the old token. + +**Cause:** ChatGPT caches access tokens and doesn't automatically re-authenticate when the server configuration changes. + +**Fix:** +1. In ChatGPT: **Settings > Connected apps** > disconnect the integration. +2. Start a new conversation and use the integration again — this forces a fresh OAuth flow. + +## How OAuth scopes work with Gitea + +Gitea's OAuth2/OIDC implementation uses **granular scopes** for API access: + +| Scope | Access | +|-------|--------| +| `read:repository` | Read repositories, issues, PRs, files | +| `write:repository` | Create/edit issues, PRs, comments, files | +| `openid` | OIDC identity (login, email) | + +When an OAuth application requests authorization, the `scope` parameter in the authorize URL determines what permissions the resulting token has. If only OIDC scopes are requested (e.g. `openid profile email`), the token will validate via the userinfo endpoint but will be rejected by Gitea's REST API with 403. + +The MCP server's `openapi-gpt.yaml` file controls which scopes ChatGPT requests. Ensure it includes: +```yaml +scopes: + read:repository: "Read access to Gitea repositories" + write:repository: "Write access to Gitea repositories" +``` + +## Reading the `oauth_auth_summary` log + +Every authenticated request emits a structured log line: + +| Field | Description | +|-------|-------------| +| `token_type` | `jwt` or `opaque` | +| `scopes_observed` | Scopes extracted from the token/userinfo | +| `scopes_effective` | Final scopes after implicit grants | +| `api_probe` | `pass`, `fail:403`, `fail:401`, `skip:cached`, `skip:error` | +| `login` | Gitea username | + +- `api_probe=pass` — token works for Gitea API calls +- `api_probe=fail:403` — token lacks API scopes, request rejected with re-auth guidance +- `api_probe=skip:cached` — previous probe passed, cached result used +- `api_probe=skip:error` — network error during probe, request allowed to proceed diff --git a/openapi-gpt.yaml b/openapi-gpt.yaml index d36f47d..6384cda 100644 --- a/openapi-gpt.yaml +++ b/openapi-gpt.yaml @@ -22,11 +22,12 @@ components: # The token URL must point to the MCP server's OAuth proxy endpoint. tokenUrl: "https://YOUR_MCP_SERVER_DOMAIN/oauth/token" scopes: - read: "Read access to Gitea repositories" + read:repository: "Read access to Gitea repositories" + write:repository: "Write access to Gitea repositories" security: - gitea_oauth: - - read + - read:repository paths: /mcp/tools: @@ -63,7 +64,7 @@ paths: so only repositories and data accessible to the user will be returned. security: - gitea_oauth: - - read + - read:repository requestBody: required: true content: diff --git a/src/aegis_gitea_mcp/config.py b/src/aegis_gitea_mcp/config.py index aa5aebf..cfee191 100644 --- a/src/aegis_gitea_mcp/config.py +++ b/src/aegis_gitea_mcp/config.py @@ -46,6 +46,13 @@ class Settings(BaseSettings): default=False, description="Allow binding to 0.0.0.0 (disabled by default for local hardening)", ) + public_base_url: HttpUrl | None = Field( + default=None, + description=( + "Public externally-reachable base URL for this MCP server. " + "When set, OAuth metadata endpoints use this URL for absolute links." + ), + ) # Logging and observability log_level: str = Field(default="INFO", description="Application logging level") @@ -204,6 +211,14 @@ class Settings(BaseSettings): raise ValueError(f"log_level must be one of {_ALLOWED_LOG_LEVELS}") return normalized + @field_validator("public_base_url", mode="before") + @classmethod + def normalize_public_base_url(cls, value: object) -> object: + """Treat empty PUBLIC_BASE_URL as unset.""" + if isinstance(value, str) and not value.strip(): + return None + return value + @field_validator("gitea_token") @classmethod def validate_token_not_empty(cls, value: str) -> str: @@ -298,6 +313,13 @@ class Settings(BaseSettings): """Get Gitea base URL as normalized string.""" return str(self.gitea_url).rstrip("/") + @property + def public_base(self) -> str | None: + """Get normalized public base URL when explicitly configured.""" + if self.public_base_url is None: + return None + return str(self.public_base_url).rstrip("/") + _settings: Settings | None = None diff --git a/src/aegis_gitea_mcp/server.py b/src/aegis_gitea_mcp/server.py index 83921bd..5fc9ec2 100644 --- a/src/aegis_gitea_mcp/server.py +++ b/src/aegis_gitea_mcp/server.py @@ -17,7 +17,11 @@ from pydantic import BaseModel, Field, ValidationError from aegis_gitea_mcp.audit import get_audit_logger from aegis_gitea_mcp.automation import AutomationError, AutomationManager from aegis_gitea_mcp.config import get_settings -from aegis_gitea_mcp.gitea_client import GiteaClient +from aegis_gitea_mcp.gitea_client import ( + GiteaAuthenticationError, + GiteaAuthorizationError, + GiteaClient, +) from aegis_gitea_mcp.logging_utils import configure_logging from aegis_gitea_mcp.mcp_protocol import ( AVAILABLE_TOOLS, @@ -74,6 +78,36 @@ logger = logging.getLogger(__name__) READ_SCOPE = "read:repository" WRITE_SCOPE = "write:repository" +# Cache of tokens verified to have Gitea API scope. +# Key: hash of token prefix, Value: monotonic expiry time. +_api_scope_cache: dict[str, float] = {} +_API_SCOPE_CACHE_TTL = 60 # seconds + +_REAUTH_GUIDANCE = ( + "Your OAuth token lacks Gitea API scopes (e.g. read:repository). " + "Revoke the authorization in Gitea (Settings > Applications > Authorized OAuth2 Applications) " + "and in ChatGPT (Settings > Connected apps), then re-authorize." +) + + +def _has_required_scope(required_scope: str, granted_scopes: set[str]) -> bool: + """Return whether granted scopes satisfy the required MCP tool scope.""" + normalized = {scope.strip().lower() for scope in granted_scopes if scope and scope.strip()} + expanded = set(normalized) + + # Compatibility: broad repository scopes imply both read and write repository access. + if "repository" in normalized or "repo" in normalized: + expanded.update({READ_SCOPE, WRITE_SCOPE}) + if "write:repo" in normalized: + expanded.add(WRITE_SCOPE) + if "read:repo" in normalized: + expanded.add(READ_SCOPE) + if WRITE_SCOPE in expanded: + expanded.add(READ_SCOPE) + + return required_scope in expanded + + app = FastAPI( title="AegisGitea MCP Server", description="Security-first MCP server for controlled AI access to self-hosted Gitea", @@ -131,7 +165,9 @@ TOOL_HANDLERS: dict[str, ToolHandler] = { def _oauth_metadata_url(request: Request) -> str: """Build absolute metadata URL for OAuth challenge responses.""" - return f"{str(request.base_url).rstrip('/')}/.well-known/oauth-protected-resource" + settings = get_settings() + base_url = settings.public_base or str(request.base_url).rstrip("/") + return f"{base_url}/.well-known/oauth-protected-resource" def _oauth_unauthorized_response( @@ -276,8 +312,82 @@ async def authenticate_and_rate_limit( if user_data: set_gitea_user_token(access_token) - set_gitea_user_login(str(user_data.get("login", "unknown"))) - set_gitea_user_scopes(user_data.get("scopes", [])) + login = str(user_data.get("login", "unknown")) + set_gitea_user_login(login) + + observed_scopes: list[str] = list(user_data.get("scopes", [])) + + # Gitea's OIDC tokens only carry standard scopes (openid, profile, email), + # not granular Gitea scopes like read:repository. When a token is + # successfully validated the user has already authorized this OAuth app, + # so we grant read:repository implicitly (and write:repository when + # write_mode is enabled). The Gitea API itself still enforces per-repo + # permissions on every call made with the user's token. + effective_scopes: set[str] = set(observed_scopes) + effective_scopes.add(READ_SCOPE) + if settings.write_mode: + effective_scopes.add(WRITE_SCOPE) + set_gitea_user_scopes(effective_scopes) + + # Probe: verify the token actually works for Gitea's REST API. + # Try both "token" and "Bearer" header formats since Gitea may + # accept OAuth tokens differently depending on version/config. + import hashlib + import time as _time + + token_hash = hashlib.sha256(access_token.encode()).hexdigest()[:16] + now = _time.monotonic() + probe_result = "skip:cached" + token_type = "jwt" if access_token.count(".") == 2 else "opaque" + + if token_hash not in _api_scope_cache or now >= _api_scope_cache[token_hash]: + try: + probe_status = None + async with httpx.AsyncClient( + timeout=settings.request_timeout_seconds + ) as probe_client: + # Try "token" format first (Gitea PAT style) + probe_resp = await probe_client.get( + f"{settings.gitea_base_url}/api/v1/user", + headers={"Authorization": f"token {access_token}"}, + ) + probe_status = probe_resp.status_code + # If "token" format fails, try "Bearer" (OAuth2 standard) + if probe_status in (401, 403): + probe_resp = await probe_client.get( + f"{settings.gitea_base_url}/api/v1/user", + headers={"Authorization": f"Bearer {access_token}"}, + ) + probe_status = probe_resp.status_code + + if probe_status in (401, 403): + probe_result = f"fail:{probe_status}" + logger.warning( + "oauth_token_lacks_api_scope", + extra={ + "status": probe_status, + "login": login, + "token_type": token_type, + "scopes_observed": observed_scopes, + }, + ) + else: + probe_result = "pass" + _api_scope_cache[token_hash] = now + _API_SCOPE_CACHE_TTL + except httpx.RequestError: + probe_result = "skip:error" + logger.debug("oauth_api_scope_probe_network_error") + + logger.info( + "oauth_auth_summary", + extra={ + "token_type": token_type, + "scopes_observed": observed_scopes, + "scopes_effective": sorted(effective_scopes), + "api_probe": probe_result, + "login": login, + }, + ) return await call_next(request) @@ -384,7 +494,7 @@ async def oauth_authorization_server_metadata(request: Request) -> JSONResponse: from this server without needing to know the Gitea URL upfront. """ settings = get_settings() - base_url = str(request.base_url).rstrip("/") + base_url = settings.public_base or str(request.base_url).rstrip("/") gitea_base = settings.gitea_base_url return JSONResponse( @@ -418,6 +528,7 @@ async def oauth_token_proxy(request: Request) -> JSONResponse: code = form_data.get("code") redirect_uri = form_data.get("redirect_uri", "") + code_verifier = form_data.get("code_verifier", "") # ChatGPT sends the client_id and client_secret (that were configured in the GPT Action # settings) in the POST body. Use those directly; fall back to env vars if not provided. client_id = form_data.get("client_id") or settings.gitea_oauth_client_id @@ -434,6 +545,8 @@ async def oauth_token_proxy(request: Request) -> JSONResponse: "grant_type": "authorization_code", "redirect_uri": redirect_uri, } + if code_verifier: + payload["code_verifier"] = code_verifier try: async with httpx.AsyncClient(timeout=30) as client: @@ -447,12 +560,25 @@ async def oauth_token_proxy(request: Request) -> JSONResponse: raise HTTPException(status_code=502, detail="Failed to reach Gitea token endpoint") from exc if response.status_code != 200: + logger.error( + "oauth_token_exchange_failed", + extra={"status": response.status_code, "body": response.text[:500]}, + ) raise HTTPException( status_code=response.status_code, detail="Token exchange failed with Gitea", ) - return JSONResponse(content=response.json()) + token_data = response.json() + logger.info( + "oauth_token_exchange_ok", + extra={ + "token_type": token_data.get("token_type"), + "scope": token_data.get("scope", ""), + "expires_in": token_data.get("expires_in"), + }, + ) + return JSONResponse(content=token_data) @app.get("/metrics") @@ -519,7 +645,7 @@ async def _execute_tool_call( required_scope = WRITE_SCOPE if tool_def.write_operation else READ_SCOPE granted_scopes = set(get_gitea_user_scopes()) - if required_scope not in granted_scopes: + if not _has_required_scope(required_scope, granted_scopes): audit.log_access_denied( tool_name=tool_name, reason=f"insufficient_scope:{required_scope}", @@ -623,6 +749,40 @@ async def call_tool(request: MCPToolCallRequest) -> JSONResponse: ) raise HTTPException(status_code=400, detail=error_message) from exc + except GiteaAuthorizationError as exc: + audit.log_tool_invocation( + tool_name=request.tool, + correlation_id=correlation_id, + result_status="error", + error="gitea_authorization_error", + ) + logger.warning("gitea_authorization_error: %s", exc) + return JSONResponse( + status_code=403, + content=MCPToolCallResponse( + success=False, + error=_REAUTH_GUIDANCE, + correlation_id=correlation_id, + ).model_dump(), + ) + + except GiteaAuthenticationError as exc: + audit.log_tool_invocation( + tool_name=request.tool, + correlation_id=correlation_id, + result_status="error", + error="gitea_authentication_error", + ) + logger.warning("gitea_authentication_error: %s", exc) + return JSONResponse( + status_code=401, + content=MCPToolCallResponse( + success=False, + error="Gitea rejected the token. Please re-authenticate.", + correlation_id=correlation_id, + ).model_dump(), + ) + except Exception: # Security decision: do not leak stack traces or raw exception messages. error_message = "Internal server error" @@ -747,6 +907,39 @@ async def sse_message_handler(request: Request) -> JSONResponse: "error": {"code": -32000, "message": str(exc.detail)}, } ) + except GiteaAuthorizationError as exc: + audit.log_tool_invocation( + tool_name=str(tool_name), + correlation_id=correlation_id, + result_status="error", + error="gitea_authorization_error", + ) + logger.warning("gitea_authorization_error: %s", exc) + return JSONResponse( + content={ + "jsonrpc": "2.0", + "id": message_id, + "error": {"code": -32000, "message": _REAUTH_GUIDANCE}, + } + ) + except GiteaAuthenticationError as exc: + audit.log_tool_invocation( + tool_name=str(tool_name), + correlation_id=correlation_id, + result_status="error", + error="gitea_authentication_error", + ) + logger.warning("gitea_authentication_error: %s", exc) + return JSONResponse( + content={ + "jsonrpc": "2.0", + "id": message_id, + "error": { + "code": -32000, + "message": "Gitea rejected the token. Please re-authenticate.", + }, + } + ) except Exception as exc: audit.log_tool_invocation( tool_name=str(tool_name), diff --git a/tests/test_server.py b/tests/test_server.py index 56a3ac9..44e2af4 100644 --- a/tests/test_server.py +++ b/tests/test_server.py @@ -31,6 +31,8 @@ def oauth_env(monkeypatch: pytest.MonkeyPatch) -> None: monkeypatch.setenv("GITEA_OAUTH_CLIENT_SECRET", "test-client-secret") monkeypatch.setenv("ENVIRONMENT", "test") monkeypatch.setenv("STARTUP_VALIDATE_GITEA", "false") + monkeypatch.setenv("WRITE_MODE", "false") + monkeypatch.setenv("PUBLIC_BASE_URL", "") @pytest.fixture @@ -104,6 +106,52 @@ def test_oauth_authorization_server_metadata(client: TestClient) -> None: assert payload["scopes_supported"] == ["read:repository", "write:repository"] +def test_oauth_metadata_uses_public_base_url(monkeypatch: pytest.MonkeyPatch) -> None: + """Public base URL is used for externally advertised OAuth metadata links.""" + monkeypatch.setenv("GITEA_URL", "https://gitea.example.com") + monkeypatch.setenv("OAUTH_MODE", "true") + monkeypatch.setenv("GITEA_OAUTH_CLIENT_ID", "test-client-id") + monkeypatch.setenv("GITEA_OAUTH_CLIENT_SECRET", "test-client-secret") + monkeypatch.setenv("PUBLIC_BASE_URL", "https://mcp.example.com") + monkeypatch.setenv("ENVIRONMENT", "test") + monkeypatch.setenv("STARTUP_VALIDATE_GITEA", "false") + + from aegis_gitea_mcp.server import app + + client = TestClient(app) + metadata_response = client.get("/.well-known/oauth-authorization-server") + assert metadata_response.status_code == 200 + payload = metadata_response.json() + assert payload["token_endpoint"] == "https://mcp.example.com/oauth/token" + + challenge_response = client.post( + "/mcp/tool/call", + json={"tool": "list_repositories", "arguments": {}}, + ) + assert challenge_response.status_code == 401 + challenge = challenge_response.headers["WWW-Authenticate"] + assert ( + 'resource_metadata="https://mcp.example.com/.well-known/oauth-protected-resource"' + in challenge + ) + + +def test_scope_compatibility_write_implies_read() -> None: + """write:repository grants read-level access for read tools.""" + from aegis_gitea_mcp.server import READ_SCOPE, _has_required_scope + + assert _has_required_scope(READ_SCOPE, {"write:repository"}) + + +def test_scope_compatibility_repository_aliases() -> None: + """Legacy/broad repository scopes satisfy MCP read/write requirements.""" + from aegis_gitea_mcp.server import READ_SCOPE, WRITE_SCOPE, _has_required_scope + + assert _has_required_scope(READ_SCOPE, {"repository"}) + assert _has_required_scope(WRITE_SCOPE, {"repository"}) + assert _has_required_scope(WRITE_SCOPE, {"repo"}) + + def test_list_tools_without_auth(client: TestClient) -> None: """Tool listing remains discoverable without auth.""" response = client.get("/mcp/tools") @@ -313,3 +361,187 @@ async def test_startup_event_succeeds_when_discovery_ready( mock_client_cls.return_value.__aexit__ = AsyncMock(return_value=False) await server.startup_event() + + +def test_probe_scopeless_token_returns_401( + oauth_env: None, mock_oauth_validation: None, monkeypatch: pytest.MonkeyPatch +) -> None: + """Middleware returns 401 with re-auth guidance when Gitea probe returns 403.""" + from aegis_gitea_mcp import server + + server._api_scope_cache.clear() + + mock_probe_response = MagicMock() + mock_probe_response.status_code = 403 + + with patch("aegis_gitea_mcp.server.httpx.AsyncClient") as mock_client_cls: + mock_client = AsyncMock() + mock_client.get = AsyncMock(return_value=mock_probe_response) + mock_client_cls.return_value.__aenter__ = AsyncMock(return_value=mock_client) + mock_client_cls.return_value.__aexit__ = AsyncMock(return_value=False) + + from aegis_gitea_mcp.server import app + + client = TestClient(app, raise_server_exceptions=False) + response = client.post( + "/mcp/tool/call", + headers={"Authorization": "Bearer valid-read"}, + json={"tool": "list_repositories", "arguments": {}}, + ) + + assert response.status_code == 401 + assert "WWW-Authenticate" in response.headers + + +def test_probe_valid_token_proceeds( + oauth_env: None, mock_oauth_validation: None, monkeypatch: pytest.MonkeyPatch +) -> None: + """Middleware allows request through when Gitea probe returns 200.""" + from aegis_gitea_mcp import server + + server._api_scope_cache.clear() + + mock_probe_response = MagicMock() + mock_probe_response.status_code = 200 + + async def _fake_execute(tool_name: str, arguments: dict, correlation_id: str) -> dict: + return {"ok": True} + + monkeypatch.setattr("aegis_gitea_mcp.server._execute_tool_call", _fake_execute) + + with patch("aegis_gitea_mcp.server.httpx.AsyncClient") as mock_client_cls: + mock_client = AsyncMock() + mock_client.get = AsyncMock(return_value=mock_probe_response) + mock_client_cls.return_value.__aenter__ = AsyncMock(return_value=mock_client) + mock_client_cls.return_value.__aexit__ = AsyncMock(return_value=False) + + from aegis_gitea_mcp.server import app + + client = TestClient(app, raise_server_exceptions=False) + response = client.post( + "/mcp/tool/call", + headers={"Authorization": "Bearer valid-read"}, + json={"tool": "list_repositories", "arguments": {}}, + ) + + assert response.status_code == 200 + assert response.json()["success"] is True + + +def test_gitea_authorization_error_returns_403( + oauth_env: None, mock_oauth_validation: None, monkeypatch: pytest.MonkeyPatch +) -> None: + """GiteaAuthorizationError from tool handler returns 403 with re-auth guidance.""" + from aegis_gitea_mcp import server + from aegis_gitea_mcp.gitea_client import GiteaAuthorizationError + + server._api_scope_cache.clear() + + async def _fake_execute(tool_name: str, arguments: dict, correlation_id: str) -> dict: + raise GiteaAuthorizationError("403 Forbidden") + + monkeypatch.setattr("aegis_gitea_mcp.server._execute_tool_call", _fake_execute) + + mock_probe_response = MagicMock() + mock_probe_response.status_code = 200 + + with patch("aegis_gitea_mcp.server.httpx.AsyncClient") as mock_client_cls: + mock_client = AsyncMock() + mock_client.get = AsyncMock(return_value=mock_probe_response) + mock_client_cls.return_value.__aenter__ = AsyncMock(return_value=mock_client) + mock_client_cls.return_value.__aexit__ = AsyncMock(return_value=False) + + from aegis_gitea_mcp.server import app + + client = TestClient(app, raise_server_exceptions=False) + response = client.post( + "/mcp/tool/call", + headers={"Authorization": "Bearer valid-read"}, + json={"tool": "list_repositories", "arguments": {}}, + ) + + assert response.status_code == 403 + body = response.json() + assert body["success"] is False + assert "re-authorize" in body["error"].lower() + + +def test_gitea_authentication_error_returns_401( + oauth_env: None, mock_oauth_validation: None, monkeypatch: pytest.MonkeyPatch +) -> None: + """GiteaAuthenticationError from tool handler returns 401.""" + from aegis_gitea_mcp import server + from aegis_gitea_mcp.gitea_client import GiteaAuthenticationError + + server._api_scope_cache.clear() + + async def _fake_execute(tool_name: str, arguments: dict, correlation_id: str) -> dict: + raise GiteaAuthenticationError("401 Unauthorized") + + monkeypatch.setattr("aegis_gitea_mcp.server._execute_tool_call", _fake_execute) + + mock_probe_response = MagicMock() + mock_probe_response.status_code = 200 + + with patch("aegis_gitea_mcp.server.httpx.AsyncClient") as mock_client_cls: + mock_client = AsyncMock() + mock_client.get = AsyncMock(return_value=mock_probe_response) + mock_client_cls.return_value.__aenter__ = AsyncMock(return_value=mock_client) + mock_client_cls.return_value.__aexit__ = AsyncMock(return_value=False) + + from aegis_gitea_mcp.server import app + + client = TestClient(app, raise_server_exceptions=False) + response = client.post( + "/mcp/tool/call", + headers={"Authorization": "Bearer valid-read"}, + json={"tool": "list_repositories", "arguments": {}}, + ) + + assert response.status_code == 401 + body = response.json() + assert body["success"] is False + assert "re-authenticate" in body["error"].lower() + + +def test_sse_gitea_authorization_error_returns_jsonrpc_error( + oauth_env: None, mock_oauth_validation: None, monkeypatch: pytest.MonkeyPatch +) -> None: + """GiteaAuthorizationError in SSE handler returns JSON-RPC -32000 with guidance.""" + from aegis_gitea_mcp import server + from aegis_gitea_mcp.gitea_client import GiteaAuthorizationError + + server._api_scope_cache.clear() + + async def _fake_execute(tool_name: str, arguments: dict, correlation_id: str) -> dict: + raise GiteaAuthorizationError("403 Forbidden") + + monkeypatch.setattr("aegis_gitea_mcp.server._execute_tool_call", _fake_execute) + + mock_probe_response = MagicMock() + mock_probe_response.status_code = 200 + + with patch("aegis_gitea_mcp.server.httpx.AsyncClient") as mock_client_cls: + mock_client = AsyncMock() + mock_client.get = AsyncMock(return_value=mock_probe_response) + mock_client_cls.return_value.__aenter__ = AsyncMock(return_value=mock_client) + mock_client_cls.return_value.__aexit__ = AsyncMock(return_value=False) + + from aegis_gitea_mcp.server import app + + client = TestClient(app, raise_server_exceptions=False) + response = client.post( + "/mcp/sse", + headers={"Authorization": "Bearer valid-read"}, + json={ + "jsonrpc": "2.0", + "id": "err-1", + "method": "tools/call", + "params": {"name": "list_repositories", "arguments": {}}, + }, + ) + + assert response.status_code == 200 + body = response.json() + assert body["error"]["code"] == -32000 + assert "re-authorize" in body["error"]["message"].lower()