From d06b862d2456db722e44df23c205ef266cb6ea19 Mon Sep 17 00:00:00 2001 From: Chris Coutinho Date: Tue, 31 Mar 2026 17:09:37 +0200 Subject: [PATCH] fix: require bearer token on provision endpoints (open redirect mitigation) Both /app/provision and /app/provision/status now require a valid Nextcloud OIDC bearer token via the Authorization header, reusing the existing validate_token_and_get_user pattern from the management API. This eliminates the open redirect vulnerability (only authenticated Astrolabe users can trigger the flow) and prevents unauthenticated resource exhaustion via Login Flow v2 session creation. The authenticated user_id from the token replaces the untrusted user_id query parameter. Co-Authored-By: Claude Opus 4.6 (1M context) --- nextcloud_mcp_server/auth/provision_routes.py | 28 ++++++- tests/unit/test_provision_routes.py | 76 ++++++++++++++++--- 2 files changed, 90 insertions(+), 14 deletions(-) diff --git a/nextcloud_mcp_server/auth/provision_routes.py b/nextcloud_mcp_server/auth/provision_routes.py index 6842b223..b0b3fab1 100644 --- a/nextcloud_mcp_server/auth/provision_routes.py +++ b/nextcloud_mcp_server/auth/provision_routes.py @@ -24,6 +24,7 @@ import anyio from starlette.requests import Request from starlette.responses import HTMLResponse, JSONResponse, RedirectResponse +from nextcloud_mcp_server.api.management import validate_token_and_get_user from nextcloud_mcp_server.auth.login_flow import LoginFlowV2Client, rewrite_url_origin from nextcloud_mcp_server.auth.storage import get_shared_storage from nextcloud_mcp_server.config import get_nextcloud_ssl_verify, get_settings @@ -144,20 +145,32 @@ async def _poll_and_store(provision_id: str) -> None: ) -async def provision_page(request: Request) -> RedirectResponse | HTMLResponse: +async def provision_page( + request: Request, +) -> RedirectResponse | HTMLResponse | JSONResponse: """Initiate Login Flow v2 and redirect to Nextcloud's login page. - GET /app/provision?redirect_uri=...&user_id=... + GET /app/provision?redirect_uri=... + + Requires a valid Nextcloud OIDC bearer token (Authorization header). + The authenticated user identity is extracted from the token — the + ``user_id`` query parameter is ignored if present. Initiates Login Flow v2, starts background polling, and redirects the browser to Nextcloud's login/grant page. After the user grants access, the background task stores the app password. The user then navigates back to the redirect_uri (Astrolabe settings). """ + # Authenticate: require a valid Nextcloud OIDC bearer token + try: + user_id, _token_data = await validate_token_and_get_user(request) + except (ValueError, KeyError, AttributeError) as e: + logger.warning(f"Provision request rejected: {e}") + return JSONResponse({"error": "Authentication required"}, status_code=401) + _cleanup_expired_sessions() redirect_uri = request.query_params.get("redirect_uri", "") - user_id = request.query_params.get("user_id", "") if not redirect_uri or not _validate_redirect_uri(redirect_uri): return HTMLResponse( @@ -250,6 +263,8 @@ async def provision_status(request: Request) -> JSONResponse: GET /app/provision/status?id=... + Requires a valid Nextcloud OIDC bearer token (Authorization header). + Returns JSON with status field: - ``"pending"`` — flow in progress, poll again - ``"completed"`` — app password stored, includes ``"username"`` @@ -257,6 +272,13 @@ async def provision_status(request: Request) -> JSONResponse: - ``"error"`` — flow completed but server-side error (e.g. missing app password) - ``"not_found"`` — unknown or already-consumed session (404) """ + # Authenticate: require a valid Nextcloud OIDC bearer token + try: + _user_id, _token_data = await validate_token_and_get_user(request) + except (ValueError, KeyError, AttributeError) as e: + logger.warning(f"Provision status request rejected: {e}") + return JSONResponse({"error": "Authentication required"}, status_code=401) + provision_id = request.query_params.get("id", "") session = _provision_sessions.get(provision_id) diff --git a/tests/unit/test_provision_routes.py b/tests/unit/test_provision_routes.py index a14b5ee0..d809df26 100644 --- a/tests/unit/test_provision_routes.py +++ b/tests/unit/test_provision_routes.py @@ -92,6 +92,16 @@ async def test_render_error_preserves_plain_text(): assert "Provisioning Error" in html_output +# ── Auth helper ────────────────────────────────────────────────────────── + +_MOCK_TOKEN_PATCH = patch( + "nextcloud_mcp_server.auth.provision_routes.validate_token_and_get_user", + new_callable=AsyncMock, + return_value=("alice", {"sub": "alice", "client_id": "astrolabe", "scopes": []}), +) +"""Patch that makes validate_token_and_get_user succeed as user 'alice'.""" + + # ── provision_status tests ─────────────────────────────────────────────── @@ -102,10 +112,23 @@ def _make_request(query_params: dict) -> MagicMock: return request +async def test_provision_status_rejects_missing_token(): + """Missing bearer token returns 401.""" + request = _make_request({"id": "some-id"}) + with patch( + "nextcloud_mcp_server.auth.provision_routes.validate_token_and_get_user", + new_callable=AsyncMock, + side_effect=ValueError("Missing Authorization header"), + ): + response = await provision_status(request) + assert response.status_code == 401 + + async def test_provision_status_not_found(): """Unknown provision ID returns 404.""" request = _make_request({"id": "nonexistent-id"}) - response = await provision_status(request) + with _MOCK_TOKEN_PATCH: + response = await provision_status(request) assert response.status_code == 404 assert response.body is not None @@ -118,7 +141,8 @@ async def test_provision_status_pending(): "expires_at": time.time() + 600, } request = _make_request({"id": provision_id}) - response = await provision_status(request) + with _MOCK_TOKEN_PATCH: + response = await provision_status(request) assert response.status_code == 200 @@ -131,7 +155,8 @@ async def test_provision_status_completed_cleans_up(): "expires_at": time.time() + 600, } request = _make_request({"id": provision_id}) - response = await provision_status(request) + with _MOCK_TOKEN_PATCH: + response = await provision_status(request) assert response.status_code == 200 # Session should be cleaned up after status read assert provision_id not in _provision_sessions @@ -145,7 +170,8 @@ async def test_provision_status_expired_by_ttl(): "expires_at": time.time() - 1, # Already expired } request = _make_request({"id": provision_id}) - response = await provision_status(request) + with _MOCK_TOKEN_PATCH: + response = await provision_status(request) assert response.status_code == 404 assert provision_id not in _provision_sessions @@ -153,17 +179,43 @@ async def test_provision_status_expired_by_ttl(): # ── provision_page tests ───────────────────────────────────────────────── +async def test_provision_page_rejects_missing_token(): + """Missing bearer token returns 401.""" + request = _make_request({"redirect_uri": "https://example.com/callback"}) + with patch( + "nextcloud_mcp_server.auth.provision_routes.validate_token_and_get_user", + new_callable=AsyncMock, + side_effect=ValueError("Missing Authorization header"), + ): + response = await provision_page(request) + assert response.status_code == 401 + + +async def test_provision_page_rejects_invalid_token(): + """Invalid bearer token returns 401.""" + request = _make_request({"redirect_uri": "https://example.com/callback"}) + with patch( + "nextcloud_mcp_server.auth.provision_routes.validate_token_and_get_user", + new_callable=AsyncMock, + side_effect=ValueError("Token validation failed"), + ): + response = await provision_page(request) + assert response.status_code == 401 + + async def test_provision_page_missing_redirect_uri(): """Missing redirect_uri returns 400.""" request = _make_request({}) - response = await provision_page(request) + with _MOCK_TOKEN_PATCH: + response = await provision_page(request) assert response.status_code == 400 async def test_provision_page_invalid_redirect_uri(): """Invalid redirect_uri (javascript:) returns 400.""" request = _make_request({"redirect_uri": "javascript:alert(1)"}) - response = await provision_page(request) + with _MOCK_TOKEN_PATCH: + response = await provision_page(request) assert response.status_code == 400 @@ -172,7 +224,6 @@ async def test_provision_page_skips_if_already_provisioned(): request = _make_request( { "redirect_uri": "https://app.example.com/settings", - "user_id": "alice", } ) @@ -181,10 +232,13 @@ async def test_provision_page_skips_if_already_provisioned(): "app_password": "existing-password", } - with patch( - "nextcloud_mcp_server.auth.provision_routes.get_shared_storage", - new_callable=AsyncMock, - return_value=mock_storage, + with ( + _MOCK_TOKEN_PATCH, + patch( + "nextcloud_mcp_server.auth.provision_routes.get_shared_storage", + new_callable=AsyncMock, + return_value=mock_storage, + ), ): response = await provision_page(request)