Add PUBLIC_BASE_URL and refine OAuth scopes
Some checks failed
docker / lint (push) Has been cancelled
docker / test (push) Has been cancelled
docker / docker-build (push) Has been cancelled
lint / lint (push) Has been cancelled
test / test (push) Has been cancelled

This commit is contained in:
2026-02-25 20:49:08 +01:00
parent 59e1ea53a8
commit c79cc1ab9e
9 changed files with 541 additions and 11 deletions

View File

@@ -2,7 +2,9 @@
"permissions": { "permissions": {
"allow": [ "allow": [
"Bash(python -m pytest:*)", "Bash(python -m pytest:*)",
"Bash(python:*)" "Bash(python:*)",
"Bash(docker compose:*)",
"Bash(findstr:*)"
] ]
} }
} }

View File

@@ -16,6 +16,9 @@ OAUTH_CACHE_TTL_SECONDS=300
# MCP server configuration # MCP server configuration
MCP_HOST=127.0.0.1 MCP_HOST=127.0.0.1
MCP_PORT=8080 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 ALLOW_INSECURE_BIND=false
# Logging / observability # Logging / observability

View File

@@ -23,6 +23,7 @@ cp .env.example .env
|---|---|---|---| |---|---|---|---|
| `MCP_HOST` | No | `127.0.0.1` | Interface to bind to | | `MCP_HOST` | No | `127.0.0.1` | Interface to bind to |
| `MCP_PORT` | No | `8080` | Port to listen on | | `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 | | `ALLOW_INSECURE_BIND` | No | `false` | Explicit opt-in required for `0.0.0.0` bind |
| `LOG_LEVEL` | No | `INFO` | `DEBUG`, `INFO`, `WARNING`, `ERROR`, `CRITICAL` | | `LOG_LEVEL` | No | `INFO` | `DEBUG`, `INFO`, `WARNING`, `ERROR`, `CRITICAL` |
| `STARTUP_VALIDATE_GITEA` | No | `true` | Validate OIDC discovery endpoint at startup | | `STARTUP_VALIDATE_GITEA` | No | `true` | Validate OIDC discovery endpoint at startup |

View File

@@ -44,6 +44,7 @@ Workflows live in `.gitea/workflows/`:
## Production Recommendations ## Production Recommendations
- Place MCP behind TLS reverse proxy. - Place MCP behind TLS reverse proxy.
- Set `PUBLIC_BASE_URL=https://<your-mcp-domain>` so OAuth metadata advertises HTTPS endpoints.
- Restrict inbound traffic to expected clients. - Restrict inbound traffic to expected clients.
- Persist and monitor audit logs. - Persist and monitor audit logs.
- Monitor `/metrics` and auth-failure events. - Monitor `/metrics` and auth-failure events.

75
docs/troubleshooting.md Normal file
View File

@@ -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

View File

@@ -22,11 +22,12 @@ components:
# The token URL must point to the MCP server's OAuth proxy endpoint. # The token URL must point to the MCP server's OAuth proxy endpoint.
tokenUrl: "https://YOUR_MCP_SERVER_DOMAIN/oauth/token" tokenUrl: "https://YOUR_MCP_SERVER_DOMAIN/oauth/token"
scopes: scopes:
read: "Read access to Gitea repositories" read:repository: "Read access to Gitea repositories"
write:repository: "Write access to Gitea repositories"
security: security:
- gitea_oauth: - gitea_oauth:
- read - read:repository
paths: paths:
/mcp/tools: /mcp/tools:
@@ -63,7 +64,7 @@ paths:
so only repositories and data accessible to the user will be returned. so only repositories and data accessible to the user will be returned.
security: security:
- gitea_oauth: - gitea_oauth:
- read - read:repository
requestBody: requestBody:
required: true required: true
content: content:

View File

@@ -46,6 +46,13 @@ class Settings(BaseSettings):
default=False, default=False,
description="Allow binding to 0.0.0.0 (disabled by default for local hardening)", 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 # Logging and observability
log_level: str = Field(default="INFO", description="Application logging level") 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}") raise ValueError(f"log_level must be one of {_ALLOWED_LOG_LEVELS}")
return normalized 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") @field_validator("gitea_token")
@classmethod @classmethod
def validate_token_not_empty(cls, value: str) -> str: def validate_token_not_empty(cls, value: str) -> str:
@@ -298,6 +313,13 @@ class Settings(BaseSettings):
"""Get Gitea base URL as normalized string.""" """Get Gitea base URL as normalized string."""
return str(self.gitea_url).rstrip("/") 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 _settings: Settings | None = None

View File

@@ -17,7 +17,11 @@ from pydantic import BaseModel, Field, ValidationError
from aegis_gitea_mcp.audit import get_audit_logger from aegis_gitea_mcp.audit import get_audit_logger
from aegis_gitea_mcp.automation import AutomationError, AutomationManager from aegis_gitea_mcp.automation import AutomationError, AutomationManager
from aegis_gitea_mcp.config import get_settings 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.logging_utils import configure_logging
from aegis_gitea_mcp.mcp_protocol import ( from aegis_gitea_mcp.mcp_protocol import (
AVAILABLE_TOOLS, AVAILABLE_TOOLS,
@@ -74,6 +78,36 @@ logger = logging.getLogger(__name__)
READ_SCOPE = "read:repository" READ_SCOPE = "read:repository"
WRITE_SCOPE = "write: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( app = FastAPI(
title="AegisGitea MCP Server", title="AegisGitea MCP Server",
description="Security-first MCP server for controlled AI access to self-hosted Gitea", 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: def _oauth_metadata_url(request: Request) -> str:
"""Build absolute metadata URL for OAuth challenge responses.""" """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( def _oauth_unauthorized_response(
@@ -276,8 +312,82 @@ async def authenticate_and_rate_limit(
if user_data: if user_data:
set_gitea_user_token(access_token) set_gitea_user_token(access_token)
set_gitea_user_login(str(user_data.get("login", "unknown"))) login = str(user_data.get("login", "unknown"))
set_gitea_user_scopes(user_data.get("scopes", [])) 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) 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. from this server without needing to know the Gitea URL upfront.
""" """
settings = get_settings() 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 gitea_base = settings.gitea_base_url
return JSONResponse( return JSONResponse(
@@ -418,6 +528,7 @@ async def oauth_token_proxy(request: Request) -> JSONResponse:
code = form_data.get("code") code = form_data.get("code")
redirect_uri = form_data.get("redirect_uri", "") 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 # 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. # 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 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", "grant_type": "authorization_code",
"redirect_uri": redirect_uri, "redirect_uri": redirect_uri,
} }
if code_verifier:
payload["code_verifier"] = code_verifier
try: try:
async with httpx.AsyncClient(timeout=30) as client: 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 raise HTTPException(status_code=502, detail="Failed to reach Gitea token endpoint") from exc
if response.status_code != 200: if response.status_code != 200:
logger.error(
"oauth_token_exchange_failed",
extra={"status": response.status_code, "body": response.text[:500]},
)
raise HTTPException( raise HTTPException(
status_code=response.status_code, status_code=response.status_code,
detail="Token exchange failed with Gitea", 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", "<not returned>"),
"expires_in": token_data.get("expires_in"),
},
)
return JSONResponse(content=token_data)
@app.get("/metrics") @app.get("/metrics")
@@ -519,7 +645,7 @@ async def _execute_tool_call(
required_scope = WRITE_SCOPE if tool_def.write_operation else READ_SCOPE required_scope = WRITE_SCOPE if tool_def.write_operation else READ_SCOPE
granted_scopes = set(get_gitea_user_scopes()) 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( audit.log_access_denied(
tool_name=tool_name, tool_name=tool_name,
reason=f"insufficient_scope:{required_scope}", 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 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: except Exception:
# Security decision: do not leak stack traces or raw exception messages. # Security decision: do not leak stack traces or raw exception messages.
error_message = "Internal server error" error_message = "Internal server error"
@@ -747,6 +907,39 @@ async def sse_message_handler(request: Request) -> JSONResponse:
"error": {"code": -32000, "message": str(exc.detail)}, "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: except Exception as exc:
audit.log_tool_invocation( audit.log_tool_invocation(
tool_name=str(tool_name), tool_name=str(tool_name),

View File

@@ -31,6 +31,8 @@ def oauth_env(monkeypatch: pytest.MonkeyPatch) -> None:
monkeypatch.setenv("GITEA_OAUTH_CLIENT_SECRET", "test-client-secret") monkeypatch.setenv("GITEA_OAUTH_CLIENT_SECRET", "test-client-secret")
monkeypatch.setenv("ENVIRONMENT", "test") monkeypatch.setenv("ENVIRONMENT", "test")
monkeypatch.setenv("STARTUP_VALIDATE_GITEA", "false") monkeypatch.setenv("STARTUP_VALIDATE_GITEA", "false")
monkeypatch.setenv("WRITE_MODE", "false")
monkeypatch.setenv("PUBLIC_BASE_URL", "")
@pytest.fixture @pytest.fixture
@@ -104,6 +106,52 @@ def test_oauth_authorization_server_metadata(client: TestClient) -> None:
assert payload["scopes_supported"] == ["read:repository", "write:repository"] 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: def test_list_tools_without_auth(client: TestClient) -> None:
"""Tool listing remains discoverable without auth.""" """Tool listing remains discoverable without auth."""
response = client.get("/mcp/tools") 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) mock_client_cls.return_value.__aexit__ = AsyncMock(return_value=False)
await server.startup_event() 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()