From a11a4e709cd9db6fa07d20bc0ebbcd166beb032f Mon Sep 17 00:00:00 2001 From: Chris Coutinho Date: Mon, 1 Jun 2026 12:56:44 +0200 Subject: [PATCH 1/3] fix(api): return 401 not 500 on failed app-password OCS validation (#824) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit provision_app_password validated credentials against OCS v1 (/ocs/v1.php/cloud/user), which always returns HTTP 200 — even on auth failure, where the real status lives in ocs.meta.statuscode (997) and ocs.data comes back as an empty list []. The status_code != 200 guard therefore never fired, execution fell through to [].get("id"), and the resulting AttributeError escaped as an unhandled 500. This blocked background vector indexing for any user whose supplied loginName didn't resolve (e.g. display name "Admin" vs loginName "admin"). Extract a shared _validate_nextcloud_credentials helper that: - queries OCS v2 (/ocs/v2.php), which maps the OCS status onto the HTTP status, so a failed credential is a real 401; - parses the payload defensively (isinstance guards) so a non-dict ocs.data can never raise; - returns a clean 502 for an unreachable Nextcloud or a non-JSON body. delete_app_password shared the same v1.php dead-guard bug, which made its credential check a no-op (any valid-format password passed) — an auth bypass on deletion. Route it through the same helper and accept the loginName from the request body (mirroring provisioning) so OIDC users whose UID differs from their loginName are not regressed. Adds unit regression tests for the OCS failure payload, non-dict data, and non-JSON response, plus a login-flow integration test that provisions with capitalized ("Admin") and spaced ("Test User") loginNames and asserts a 401 rather than a 500. Co-Authored-By: Claude Opus 4.8 (1M context) --- nextcloud_mcp_server/api/passwords.py | 197 ++++++++++++------ .../test_app_password_loginname_mismatch.py | 87 ++++++++ .../test_management_app_password_endpoints.py | 190 ++++++++++++++++- 3 files changed, 412 insertions(+), 62 deletions(-) create mode 100644 tests/server/login_flow/test_app_password_loginname_mismatch.py diff --git a/nextcloud_mcp_server/api/passwords.py b/nextcloud_mcp_server/api/passwords.py index 6f29892f..bd07cfde 100644 --- a/nextcloud_mcp_server/api/passwords.py +++ b/nextcloud_mcp_server/api/passwords.py @@ -39,6 +39,11 @@ APP_PASSWORD_PATTERN = re.compile(r"^[a-zA-Z0-9-]{20,256}$") # Timeout for Nextcloud API validation requests (seconds) NEXTCLOUD_VALIDATION_TIMEOUT = 10.0 +# OCS meta status codes that indicate success. OCS v1 (``/ocs/v1.php``) reports +# 100; OCS v2 (``/ocs/v2.php``) reports 200. We query v2 (see +# ``_validate_nextcloud_credentials``) but accept both for robustness. +_OCS_SUCCESS_STATUSCODES = frozenset({100, 200}) + # Rate limiting configuration for app password provisioning # Limits: 5 attempts per user per hour RATE_LIMIT_MAX_ATTEMPTS = 5 @@ -182,6 +187,94 @@ async def _get_app_password_storage(request: Request) -> RefreshTokenStorage: return storage +async def _validate_nextcloud_credentials( + nextcloud_host: str, login_name: str, password: str +) -> tuple[str | None, JSONResponse | None]: + """Validate a credential against Nextcloud and return the account UID. + + Authenticates against the OCS ``/cloud/user`` endpoint as ``login_name``. + Nextcloud keys app-password auth on the *loginName*, which may differ from + the UID (e.g. OIDC-provisioned users, or the ``admin`` account whose display + name is ``Admin``). + + Queries OCS **v2** (``/ocs/v2.php``) deliberately. OCS **v1** + (``/ocs/v1.php``) always returns HTTP 200 — even on auth failure, where it + wraps the real status in ``ocs.meta.statuscode`` (997 = unauthenticated) and + returns ``ocs.data`` as an empty list ``[]``. v2 maps the OCS status onto + the HTTP status, so a failed credential is a real 401. The payload is also + parsed defensively so a non-dict ``ocs.data`` can never raise — this is the + crash behind issue #824 (``AttributeError: 'list' object has no attribute + 'get'`` escaping as an unhandled 500). + + Returns: + ``(ocs_user_id, None)`` on success, otherwise ``(None, error_response)`` + with a ready-to-return :class:`JSONResponse`: 401 for an invalid + credential, 502 when Nextcloud is unreachable or returns something we + cannot parse. + """ + try: + async with nextcloud_httpx_client( + timeout=NEXTCLOUD_VALIDATION_TIMEOUT + ) as client: + response = await client.get( + f"{nextcloud_host}/ocs/v2.php/cloud/user", + auth=(login_name, password), + params={"format": "json"}, + headers={"OCS-APIRequest": "true"}, + ) + except httpx.RequestError as e: + logger.error("Failed to reach Nextcloud for credential validation: %s", e) + return None, JSONResponse( + {"success": False, "error": "Failed to validate credentials"}, + status_code=502, + ) + + # v2.php maps an OCS auth failure onto a real HTTP status (e.g. 401). + if response.status_code != 200: + logger.warning("Credential validation failed: HTTP %s", response.status_code) + return None, JSONResponse( + {"success": False, "error": "Invalid app password"}, + status_code=401, + ) + + # Parse defensively: even on HTTP 200 the body may be malformed or carry a + # non-dict ``ocs.data`` (the v1.php auth-failure shape, kept as a guard + # against surprises). Never call ``.get`` on something that isn't a dict. + try: + payload = response.json() + except ValueError as e: + logger.error("Nextcloud returned a non-JSON OCS response: %s", e) + return None, JSONResponse( + {"success": False, "error": "Unexpected response from Nextcloud"}, + status_code=502, + ) + + ocs = payload.get("ocs") if isinstance(payload, dict) else None + meta = ocs.get("meta") if isinstance(ocs, dict) else None + statuscode = meta.get("statuscode") if isinstance(meta, dict) else None + ocs_data = ocs.get("data") if isinstance(ocs, dict) else None + ocs_user_id = ocs_data.get("id") if isinstance(ocs_data, dict) else None + + # Treat a non-success OCS status, or a payload we can't read a user id from, + # as a failed validation rather than crashing. ``statuscode`` is ``None`` + # when ``meta`` is absent; fall back to "did we get a user id?" so a minimal + # but valid response still passes. + if ( + statuscode is not None and statuscode not in _OCS_SUCCESS_STATUSCODES + ) or not ocs_user_id: + logger.warning( + "Credential validation failed: OCS statuscode=%s, user_id present=%s", + statuscode, + ocs_user_id is not None, + ) + return None, JSONResponse( + {"success": False, "error": "Invalid app password"}, + status_code=401, + ) + + return ocs_user_id, None + + async def provision_app_password(request: Request) -> JSONResponse: """POST /api/v1/users/{user_id}/app-password - Store app password for background sync. @@ -271,47 +364,21 @@ async def provision_app_password(request: Request) -> JSONResponse: # literally in the header (RFC 7617 — no URL-encoding) and Nextcloud keys # app-password auth on the loginName, so authenticate as the loginName, not # the UID. - try: - async with nextcloud_httpx_client( - timeout=NEXTCLOUD_VALIDATION_TIMEOUT - ) as client: - # Use OCS API to verify credentials - test_url = f"{nextcloud_host}/ocs/v1.php/cloud/user" - response = await client.get( - test_url, - auth=(login_name, app_password), - params={"format": "json"}, - headers={"OCS-APIRequest": "true"}, - ) + ocs_user_id, error_response = await _validate_nextcloud_credentials( + nextcloud_host, login_name, app_password + ) + if error_response is not None: + _record_rate_limit_attempt(path_user_id, success=False) + return error_response - if response.status_code != 200: - logger.warning( - "App password validation failed for user: HTTP %s", - response.status_code, - ) - _record_rate_limit_attempt(path_user_id, success=False) - return JSONResponse( - {"success": False, "error": "Invalid app password"}, - status_code=401, - ) - - # Verify the authenticated account maps to the path user_id (UID): - # the loginName must resolve to the UID claimed in the URL path. - data = response.json() - ocs_user_id = data.get("ocs", {}).get("data", {}).get("id") - if ocs_user_id != path_user_id: - logger.warning("User ID mismatch in OCS response") - _record_rate_limit_attempt(path_user_id, success=False) - return JSONResponse( - {"success": False, "error": "User ID mismatch"}, - status_code=403, - ) - - except httpx.RequestError as e: - logger.error("Failed to validate app password: %s", e) + # Verify the authenticated account maps to the path user_id (UID): the + # loginName must resolve to the UID claimed in the URL path. + if ocs_user_id != path_user_id: + logger.warning("User ID mismatch in OCS response") + _record_rate_limit_attempt(path_user_id, success=False) return JSONResponse( - {"success": False, "error": "Failed to validate credentials"}, - status_code=500, + {"success": False, "error": "User ID mismatch"}, + status_code=403, ) # Store the validated app password @@ -402,34 +469,44 @@ async def delete_app_password(request: Request) -> JSONResponse: if error_response is not None: return error_response - # Validate credentials against Nextcloud + # Nextcloud keys app-password auth on the loginName, which can differ from + # the UID (OIDC-provisioned users). Use the loginName from the body when + # present, falling back to the path UID for legacy callers — same contract + # as provisioning. + nc_username = None + try: + body = await request.json() + nc_username = body.get("username") + except Exception: + pass # No JSON body = legacy call without an explicit loginName + login_name = nc_username or username + + # Validate credentials against Nextcloud. Shares the OCS v2 + defensive + # parsing path with provisioning (issue #824): OCS v1 always returned HTTP + # 200, so the old ``!= 200`` guard never fired and a bad credential could + # silently pass — a genuine auth bypass on deletion. settings = get_settings() nextcloud_host = settings.nextcloud_host - try: - async with nextcloud_httpx_client( - timeout=NEXTCLOUD_VALIDATION_TIMEOUT - ) as client: - test_url = f"{nextcloud_host}/ocs/v1.php/cloud/user" - response = await client.get( - test_url, - auth=(username, password), - params={"format": "json"}, - headers={"OCS-APIRequest": "true"}, - ) - - if response.status_code != 200: - return JSONResponse( - {"success": False, "error": "Invalid credentials"}, - status_code=401, - ) - except httpx.RequestError as e: - logger.error("Failed to validate credentials: %s", e) + if not nextcloud_host: + logger.error("NEXTCLOUD_HOST not configured") return JSONResponse( - {"success": False, "error": "Failed to validate credentials"}, + {"success": False, "error": "Server not configured"}, status_code=500, ) + _, error_response = await _validate_nextcloud_credentials( + nextcloud_host, login_name, password + ) + if error_response is not None: + # Preserve the historical "Invalid credentials" wording for this route. + if error_response.status_code == 401: + return JSONResponse( + {"success": False, "error": "Invalid credentials"}, + status_code=401, + ) + return error_response + try: storage = await _get_app_password_storage(request) deleted = await storage.delete_app_password(username) diff --git a/tests/server/login_flow/test_app_password_loginname_mismatch.py b/tests/server/login_flow/test_app_password_loginname_mismatch.py new file mode 100644 index 00000000..457058bc --- /dev/null +++ b/tests/server/login_flow/test_app_password_loginname_mismatch.py @@ -0,0 +1,87 @@ +"""Integration test for issue #824 against the live login-flow MCP server. + +`POST /api/v1/users/{user_id}/app-password` validates the supplied BasicAuth +credential against Nextcloud's OCS `/cloud/user` endpoint. Nextcloud keys +app-password auth on the *loginName*, which differs from the display name — +e.g. the admin account's display name is `Admin` (capital A) while its +loginName is `admin`, and a user "Test User" (with a space) has a distinct +loginName/UID. + +When the supplied loginName does not authenticate, OCS v1 (`/ocs/v1.php`) +returns **HTTP 200** with `ocs.meta.statuscode: 997` and `ocs.data: []`. The +old handler gated auth failure on the HTTP status (`!= 200`), so it never +fired, fell through to `[].get("id")`, and raised +`AttributeError: 'list' object has no attribute 'get'` — escaping as an +unhandled **500**. The fix queries OCS v2 and parses defensively, so a failed +validation is a clean **401**. + +These tests exercise the live `mcp-login-flow` container (port 8004), which +performs the OCS round-trip against the real Nextcloud. The credentials are +deliberately wrong, so validation fails for every UID shape under test — +capitals (`Admin`) and spaces (`Test User`) — which is exactly the path that +used to 500. +""" + +import httpx +import pytest + +LOGIN_FLOW_API_BASE_URL = "http://localhost:8004" + +pytestmark = [pytest.mark.integration, pytest.mark.login_flow] + +# A syntactically valid app password (matches APP_PASSWORD_PATTERN) that is not +# a real credential for any account — so the OCS validation always fails. +_WRONG_APP_PASSWORD = "aaaaa-bbbbb-ccccc-ddddd-eeeee" + + +def _basic_auth_header(username: str, password: str) -> str: + return httpx.BasicAuth(username, password)._auth_header + + +@pytest.mark.parametrize( + "user_id", + [ + pytest.param("Admin", id="capitalized-display-name"), + pytest.param("Test User", id="display-name-with-space"), + ], +) +async def test_provision_with_unresolvable_loginname_returns_401_not_500(user_id): + """A loginName that fails OCS validation yields 401, never 500 (#824). + + Pre-fix, the capitalized/spaced loginName produced an OCS v1 ``200 + + data: []`` payload that crashed parsing with an unhandled 500. + """ + async with httpx.AsyncClient(timeout=30.0) as client: + response = await client.post( + f"{LOGIN_FLOW_API_BASE_URL}/api/v1/users/{user_id}/app-password", + headers={"Authorization": _basic_auth_header(user_id, _WRONG_APP_PASSWORD)}, + ) + + assert response.status_code == 401, ( + f"provisioning for {user_id!r} returned {response.status_code} " + f"(expected 401, not 500): {response.text}" + ) + assert "Invalid app password" in response.json().get("error", "") + + +async def test_provision_with_mismatched_loginname_body_returns_401_not_500(): + """The body-supplied loginName is what's validated; a non-resolving + loginName (display name with a space) still yields 401, not 500 (#824). + + Mirrors the production call shape where the UID in the path differs from the + Nextcloud loginName sent in the JSON body. + """ + async with httpx.AsyncClient(timeout=30.0) as client: + response = await client.post( + f"{LOGIN_FLOW_API_BASE_URL}/api/v1/users/testuser/app-password", + headers={ + "Authorization": _basic_auth_header("testuser", _WRONG_APP_PASSWORD) + }, + json={"username": "Test User"}, + ) + + assert response.status_code == 401, ( + f"provisioning returned {response.status_code} " + f"(expected 401, not 500): {response.text}" + ) + assert "Invalid app password" in response.json().get("error", "") diff --git a/tests/unit/test_management_app_password_endpoints.py b/tests/unit/test_management_app_password_endpoints.py index 7b62ac3e..8973cb05 100644 --- a/tests/unit/test_management_app_password_endpoints.py +++ b/tests/unit/test_management_app_password_endpoints.py @@ -356,6 +356,142 @@ async def test_provision_app_password_nextcloud_validation_fails(mocker): assert "Invalid app password" in response.json()["error"] +def _mock_ocs_client(mocker, *, status_code: int, json_payload=None, json_error=None): + """Build a mocked ``nextcloud_httpx_client`` returning a canned OCS response. + + ``json_payload`` sets ``response.json()`` return value; ``json_error`` makes + ``response.json()`` raise (simulating a non-JSON body). + """ + mock_response = MagicMock() + mock_response.status_code = status_code + if json_error is not None: + mock_response.json.side_effect = json_error + else: + mock_response.json.return_value = json_payload + + mock_client = AsyncMock() + mock_client.get = AsyncMock(return_value=mock_response) + mock_client.__aenter__ = AsyncMock(return_value=mock_client) + mock_client.__aexit__ = AsyncMock() + mocker.patch( + "nextcloud_mcp_server.api.passwords.nextcloud_httpx_client", + return_value=mock_client, + ) + return mock_client + + +def _provision_only_app(): + return Starlette( + routes=[ + Route( + "/api/v1/users/{user_id}/app-password", + provision_app_password, + methods=["POST"], + ), + ] + ) + + +async def test_provision_app_password_ocs_v1_failure_payload_returns_401(mocker): + """Regression for #824: an OCS auth-failure payload (HTTP 200 + + ``meta.statuscode: 997`` + ``data: []``) must return 401, never 500. + + OCS v1 always returns HTTP 200; the old ``status_code != 200`` guard never + fired, so execution fell through to ``[].get("id")`` and raised + ``AttributeError: 'list' object has no attribute 'get'`` as an unhandled + 500. + """ + mocker.patch( + "nextcloud_mcp_server.api.passwords.get_settings", + return_value=MagicMock( + nextcloud_host="http://localhost:8080", + nextcloud_verify_ssl=True, + nextcloud_ca_bundle=None, + ), + ) + _mock_ocs_client( + mocker, + status_code=200, + json_payload={"ocs": {"meta": {"statuscode": 997}, "data": []}}, + ) + + client = TestClient(_provision_only_app()) + response = client.post( + "/api/v1/users/Admin/app-password", + headers={ + "Authorization": create_basic_auth_header( + "Admin", "aaaaa-bbbbb-ccccc-ddddd-eeeee" + ) + }, + ) + + assert response.status_code == 401 + assert "Invalid app password" in response.json()["error"] + + +async def test_provision_app_password_nondict_data_does_not_500(mocker): + """Regression for #824: a non-dict ``ocs.data`` under HTTP 200 (list, or + ``null``) is handled gracefully (401), not as an unhandled 500/ + AttributeError.""" + mocker.patch( + "nextcloud_mcp_server.api.passwords.get_settings", + return_value=MagicMock( + nextcloud_host="http://localhost:8080", + nextcloud_verify_ssl=True, + nextcloud_ca_bundle=None, + ), + ) + + for bad_data in ([], None, "unexpected"): + passwords._rate_limit_attempts.clear() + _mock_ocs_client( + mocker, + status_code=200, + json_payload={"ocs": {"data": bad_data}}, + ) + client = TestClient(_provision_only_app()) + response = client.post( + "/api/v1/users/testuser/app-password", + headers={ + "Authorization": create_basic_auth_header( + "testuser", "aaaaa-bbbbb-ccccc-ddddd-eeeee" + ) + }, + ) + assert response.status_code == 401, f"data={bad_data!r} should yield 401" + assert "Invalid app password" in response.json()["error"] + + +async def test_provision_app_password_non_json_response_returns_502(mocker): + """A non-JSON OCS body under HTTP 200 returns a clean 502, not a 500.""" + mocker.patch( + "nextcloud_mcp_server.api.passwords.get_settings", + return_value=MagicMock( + nextcloud_host="http://localhost:8080", + nextcloud_verify_ssl=True, + nextcloud_ca_bundle=None, + ), + ) + _mock_ocs_client( + mocker, + status_code=200, + json_error=ValueError("Expecting value: line 1 column 1 (char 0)"), + ) + + client = TestClient(_provision_only_app()) + response = client.post( + "/api/v1/users/testuser/app-password", + headers={ + "Authorization": create_basic_auth_header( + "testuser", "aaaaa-bbbbb-ccccc-ddddd-eeeee" + ) + }, + ) + + assert response.status_code == 502 + assert "Unexpected response" in response.json()["error"] + + async def test_get_app_password_status_provisioned(temp_storage, mocker): """Test checking status when app password is provisioned.""" # Store an app password @@ -441,9 +577,12 @@ async def test_delete_app_password_success(temp_storage, mocker): ), ) - # Mock httpx client for Nextcloud validation + # Mock httpx client for Nextcloud validation (OCS v2 success shape) mock_response = MagicMock() mock_response.status_code = 200 + mock_response.json.return_value = { + "ocs": {"meta": {"statuscode": 200}, "data": {"id": "testuser"}} + } mock_client = AsyncMock() mock_client.get = AsyncMock(return_value=mock_response) @@ -489,9 +628,12 @@ async def test_delete_app_password_not_found(temp_storage, mocker): ), ) - # Mock httpx client for Nextcloud validation + # Mock httpx client for Nextcloud validation (OCS v2 success shape) mock_response = MagicMock() mock_response.status_code = 200 + mock_response.json.return_value = { + "ocs": {"meta": {"statuscode": 200}, "data": {"id": "testuser"}} + } mock_client = AsyncMock() mock_client.get = AsyncMock(return_value=mock_response) @@ -570,6 +712,50 @@ async def test_delete_app_password_invalid_credentials(mocker): assert "Invalid credentials" in response.json()["error"] +async def test_delete_app_password_ocs_failure_payload_returns_401(mocker): + """Regression for #824: delete shares the OCS v2 + defensive parsing path. + + An OCS auth-failure payload (HTTP 200 + ``data: []``) must reject the + deletion with 401 — previously the ``!= 200`` guard on v1.php never fired, + so a wrong password silently passed validation (an auth bypass on delete). + """ + mocker.patch( + "nextcloud_mcp_server.api.passwords.get_settings", + return_value=MagicMock( + nextcloud_host="http://localhost:8080", + nextcloud_verify_ssl=True, + nextcloud_ca_bundle=None, + ), + ) + _mock_ocs_client( + mocker, + status_code=200, + json_payload={"ocs": {"meta": {"statuscode": 997}, "data": []}}, + ) + + app = Starlette( + routes=[ + Route( + "/api/v1/users/{user_id}/app-password", + delete_app_password, + methods=["DELETE"], + ), + ] + ) + client = TestClient(app) + response = client.delete( + "/api/v1/users/testuser/app-password", + headers={ + "Authorization": create_basic_auth_header( + "testuser", "aaaaa-bbbbb-ccccc-ddddd-eeeee" + ) + }, + ) + + assert response.status_code == 401 + assert "Invalid credentials" in response.json()["error"] + + async def test_delete_app_password_username_mismatch(): """Test that username mismatch returns 403 for deletion.""" app = Starlette( From 3c55d03ecf9f36e001c1fff6f211c9583d8a3597 Mon Sep 17 00:00:00 2001 From: Chris Coutinho Date: Mon, 1 Jun 2026 13:04:42 +0200 Subject: [PATCH 2/3] fix(api): block cross-user delete and address review feedback (#824) Review of PR #825 surfaced an auth bypass introduced by adding loginName support to delete_app_password: with the OCS-resolved UID discarded, a user could authenticate as their own loginName (via the request body) while targeting another user's path and delete the victim's stored app password. Add the same UID-mismatch guard provisioning already has, so the authenticated account must own the path UID (403 otherwise). Also: - integration test: build the BasicAuth header via base64 instead of httpx.BasicAuth._auth_header (private attribute); mark the throwaway test credential NOSONAR(S2068). - unit tests: cover the httpx.RequestError -> 502 branch, the standard OCS v2 success shape (meta.statuscode 200), and the cross-user delete 403 guard. Co-Authored-By: Claude Opus 4.8 (1M context) --- nextcloud_mcp_server/api/passwords.py | 15 ++- .../test_app_password_loginname_mismatch.py | 10 +- .../test_management_app_password_endpoints.py | 121 ++++++++++++++++++ 3 files changed, 142 insertions(+), 4 deletions(-) diff --git a/nextcloud_mcp_server/api/passwords.py b/nextcloud_mcp_server/api/passwords.py index bd07cfde..70ca0e2d 100644 --- a/nextcloud_mcp_server/api/passwords.py +++ b/nextcloud_mcp_server/api/passwords.py @@ -495,7 +495,7 @@ async def delete_app_password(request: Request) -> JSONResponse: status_code=500, ) - _, error_response = await _validate_nextcloud_credentials( + ocs_user_id, error_response = await _validate_nextcloud_credentials( nextcloud_host, login_name, password ) if error_response is not None: @@ -507,6 +507,19 @@ async def delete_app_password(request: Request) -> JSONResponse: ) return error_response + # The authenticated account must be the UID whose password is being deleted. + # ``_extract_basic_auth`` only checks the BasicAuth *name* field equals the + # path UID, not that the supplied credential authenticates as that account — + # without this guard a user could authenticate with their own loginName (via + # the body) while targeting another user's path and delete the victim's + # stored password. + if ocs_user_id != path_user_id: + logger.warning("User ID mismatch in OCS response for delete") + return JSONResponse( + {"success": False, "error": "User ID mismatch"}, + status_code=403, + ) + try: storage = await _get_app_password_storage(request) deleted = await storage.delete_app_password(username) diff --git a/tests/server/login_flow/test_app_password_loginname_mismatch.py b/tests/server/login_flow/test_app_password_loginname_mismatch.py index 457058bc..9f835686 100644 --- a/tests/server/login_flow/test_app_password_loginname_mismatch.py +++ b/tests/server/login_flow/test_app_password_loginname_mismatch.py @@ -22,6 +22,8 @@ capitals (`Admin`) and spaces (`Test User`) — which is exactly the path that used to 500. """ +import base64 + import httpx import pytest @@ -30,12 +32,14 @@ LOGIN_FLOW_API_BASE_URL = "http://localhost:8004" pytestmark = [pytest.mark.integration, pytest.mark.login_flow] # A syntactically valid app password (matches APP_PASSWORD_PATTERN) that is not -# a real credential for any account — so the OCS validation always fails. -_WRONG_APP_PASSWORD = "aaaaa-bbbbb-ccccc-ddddd-eeeee" +# a real credential for any account — so the OCS validation always fails. This +# is a throwaway test fixture, not a real secret. +_WRONG_APP_PASSWORD = "aaaaa-bbbbb-ccccc-ddddd-eeeee" # NOSONAR(S2068) def _basic_auth_header(username: str, password: str) -> str: - return httpx.BasicAuth(username, password)._auth_header + credentials = base64.b64encode(f"{username}:{password}".encode()).decode() + return f"Basic {credentials}" @pytest.mark.parametrize( diff --git a/tests/unit/test_management_app_password_endpoints.py b/tests/unit/test_management_app_password_endpoints.py index 8973cb05..7e191b16 100644 --- a/tests/unit/test_management_app_password_endpoints.py +++ b/tests/unit/test_management_app_password_endpoints.py @@ -492,6 +492,79 @@ async def test_provision_app_password_non_json_response_returns_502(mocker): assert "Unexpected response" in response.json()["error"] +async def test_provision_app_password_request_error_returns_502(mocker): + """A transport-level failure reaching Nextcloud returns a clean 502.""" + import httpx + + mocker.patch( + "nextcloud_mcp_server.api.passwords.get_settings", + return_value=MagicMock( + nextcloud_host="http://localhost:8080", + nextcloud_verify_ssl=True, + nextcloud_ca_bundle=None, + ), + ) + + mock_client = AsyncMock() + mock_client.get = AsyncMock(side_effect=httpx.ConnectError("connection refused")) + mock_client.__aenter__ = AsyncMock(return_value=mock_client) + # return_value=False so the context manager does not suppress the raised error + mock_client.__aexit__ = AsyncMock(return_value=False) + mocker.patch( + "nextcloud_mcp_server.api.passwords.nextcloud_httpx_client", + return_value=mock_client, + ) + + client = TestClient(_provision_only_app()) + response = client.post( + "/api/v1/users/testuser/app-password", + headers={ + "Authorization": create_basic_auth_header( + "testuser", "aaaaa-bbbbb-ccccc-ddddd-eeeee" + ) + }, + ) + + assert response.status_code == 502 + assert "Failed to validate credentials" in response.json()["error"] + + +async def test_provision_app_password_standard_v2_success(temp_storage, mocker): + """A standard OCS v2 success payload (``meta.statuscode: 200`` + ``data``) + provisions normally — exercises the ``statuscode in success`` branch rather + than the no-meta fallback.""" + mocker.patch( + "nextcloud_mcp_server.api.passwords.get_settings", + return_value=MagicMock( + nextcloud_host="http://localhost:8080", + nextcloud_verify_ssl=True, + nextcloud_ca_bundle=None, + ), + ) + _mock_ocs_client( + mocker, + status_code=200, + json_payload={"ocs": {"meta": {"statuscode": 200}, "data": {"id": "testuser"}}}, + ) + + client = TestClient(create_test_app(temp_storage)) + response = client.post( + "/api/v1/users/testuser/app-password", + headers={ + "Authorization": create_basic_auth_header( + "testuser", "aaaaa-bbbbb-ccccc-ddddd-eeeee" + ) + }, + ) + + assert response.status_code == 200 + assert response.json()["success"] is True + assert ( + await temp_storage.get_app_password("testuser") + == "aaaaa-bbbbb-ccccc-ddddd-eeeee" + ) + + async def test_get_app_password_status_provisioned(temp_storage, mocker): """Test checking status when app password is provisioned.""" # Store an app password @@ -756,6 +829,54 @@ async def test_delete_app_password_ocs_failure_payload_returns_401(mocker): assert "Invalid credentials" in response.json()["error"] +async def test_delete_app_password_cross_user_uid_mismatch_returns_403( + temp_storage, mocker +): + """Regression: the authenticated account must own the path UID. + + An attacker authenticating as their own loginName (via the body) while + targeting another user's path must be rejected with 403 — otherwise they + could delete the victim's stored password. The OCS validation resolves to a + different UID than the path, so the UID-mismatch guard must fire. + """ + await temp_storage.store_app_password("victim", "aaaaa-bbbbb-ccccc-ddddd-eeeee") + + mocker.patch( + "nextcloud_mcp_server.api.passwords.get_settings", + return_value=MagicMock( + nextcloud_host="http://localhost:8080", + nextcloud_verify_ssl=True, + nextcloud_ca_bundle=None, + ), + ) + # The supplied credential authenticates, but as "attacker", not "victim". + _mock_ocs_client( + mocker, + status_code=200, + json_payload={"ocs": {"meta": {"statuscode": 200}, "data": {"id": "attacker"}}}, + ) + + app = create_test_app(temp_storage) + client = TestClient(app) + response = client.request( + "DELETE", + "/api/v1/users/victim/app-password", + headers={ + "Authorization": create_basic_auth_header( + "victim", "fffff-ggggg-hhhhh-iiiii-jjjjj" + ) + }, + json={"username": "attacker-loginname"}, + ) + + assert response.status_code == 403 + assert "mismatch" in response.json()["error"].lower() + # Victim's password must be untouched. + assert ( + await temp_storage.get_app_password("victim") == "aaaaa-bbbbb-ccccc-ddddd-eeeee" + ) + + async def test_delete_app_password_username_mismatch(): """Test that username mismatch returns 403 for deletion.""" app = Starlette( From be360edab0b7809448726960eacdc430a759283a Mon Sep 17 00:00:00 2001 From: Chris Coutinho Date: Mon, 1 Jun 2026 15:15:48 +0200 Subject: [PATCH 3/3] fix(api): distinguish Nextcloud 5xx from auth failure; tighten body parse (#824) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Address PR #825 review round 2: - _validate_nextcloud_credentials now only maps OCS HTTP 401/403 to a 401 "invalid credential"; any other non-200 (5xx, 503 maintenance mode) surfaces as 502 "Nextcloud returned a server error" so ops don't chase a phantom bad password when Nextcloud is actually down. - The client-facing 401 message is now a parameter, so delete_app_password keeps its "Invalid credentials" wording without unwrapping/rebuilding the helper's JSONResponse. - Body parsing catches (ValueError, UnicodeDecodeError) instead of bare Exception, and guards body.get behind isinstance(body, dict) — no longer swallows RuntimeError/AttributeError or a non-object JSON body. - Add a unit test asserting 500/503 -> 502. Co-Authored-By: Claude Opus 4.8 (1M context) --- nextcloud_mcp_server/api/passwords.py | 56 ++++++++++++------- .../test_management_app_password_endpoints.py | 27 +++++++++ 2 files changed, 64 insertions(+), 19 deletions(-) diff --git a/nextcloud_mcp_server/api/passwords.py b/nextcloud_mcp_server/api/passwords.py index 70ca0e2d..2cf365a0 100644 --- a/nextcloud_mcp_server/api/passwords.py +++ b/nextcloud_mcp_server/api/passwords.py @@ -188,7 +188,11 @@ async def _get_app_password_storage(request: Request) -> RefreshTokenStorage: async def _validate_nextcloud_credentials( - nextcloud_host: str, login_name: str, password: str + nextcloud_host: str, + login_name: str, + password: str, + *, + invalid_credential_error: str = "Invalid app password", ) -> tuple[str | None, JSONResponse | None]: """Validate a credential against Nextcloud and return the account UID. @@ -206,11 +210,15 @@ async def _validate_nextcloud_credentials( crash behind issue #824 (``AttributeError: 'list' object has no attribute 'get'`` escaping as an unhandled 500). + Args: + invalid_credential_error: client-facing error string for the 401 path, + so callers can keep their own wording. + Returns: ``(ocs_user_id, None)`` on success, otherwise ``(None, error_response)`` - with a ready-to-return :class:`JSONResponse`: 401 for an invalid - credential, 502 when Nextcloud is unreachable or returns something we - cannot parse. + with a ready-to-return :class:`JSONResponse`: **401** for an invalid + credential, **502** when Nextcloud is unreachable, errors out (5xx / + maintenance mode), or returns something we cannot parse. """ try: async with nextcloud_httpx_client( @@ -229,13 +237,22 @@ async def _validate_nextcloud_credentials( status_code=502, ) - # v2.php maps an OCS auth failure onto a real HTTP status (e.g. 401). - if response.status_code != 200: + # v2.php maps an OCS auth failure onto a real HTTP status. Only 401/403 mean + # "bad credential" — anything else non-200 (5xx, 503 maintenance mode) is a + # Nextcloud-side problem and must surface as 502, not a misleading "invalid + # password" that sends ops chasing the wrong cause. + if response.status_code in (401, 403): logger.warning("Credential validation failed: HTTP %s", response.status_code) return None, JSONResponse( - {"success": False, "error": "Invalid app password"}, + {"success": False, "error": invalid_credential_error}, status_code=401, ) + if response.status_code != 200: + logger.error("Nextcloud OCS returned HTTP %s", response.status_code) + return None, JSONResponse( + {"success": False, "error": "Nextcloud returned a server error"}, + status_code=502, + ) # Parse defensively: even on HTTP 200 the body may be malformed or carry a # non-dict ``ocs.data`` (the v1.php auth-failure shape, kept as a guard @@ -268,7 +285,7 @@ async def _validate_nextcloud_credentials( ocs_user_id is not None, ) return None, JSONResponse( - {"success": False, "error": "Invalid app password"}, + {"success": False, "error": invalid_credential_error}, status_code=401, ) @@ -342,10 +359,11 @@ async def provision_app_password(request: Request) -> JSONResponse: nc_username = None try: body = await request.json() + except (ValueError, UnicodeDecodeError): + body = None # No / malformed JSON body = legacy call without extras + if isinstance(body, dict): scopes = body.get("scopes") # list[str] | None nc_username = body.get("username") # Nextcloud loginName - except Exception: - pass # No JSON body = legacy call without scopes / loginName login_name = nc_username or username @@ -476,9 +494,10 @@ async def delete_app_password(request: Request) -> JSONResponse: nc_username = None try: body = await request.json() + except (ValueError, UnicodeDecodeError): + body = None # No / malformed JSON body = legacy call without a loginName + if isinstance(body, dict): nc_username = body.get("username") - except Exception: - pass # No JSON body = legacy call without an explicit loginName login_name = nc_username or username # Validate credentials against Nextcloud. Shares the OCS v2 + defensive @@ -495,16 +514,15 @@ async def delete_app_password(request: Request) -> JSONResponse: status_code=500, ) + # Keep this route's historical "Invalid credentials" wording via the helper + # rather than unwrapping and rebuilding its response. ocs_user_id, error_response = await _validate_nextcloud_credentials( - nextcloud_host, login_name, password + nextcloud_host, + login_name, + password, + invalid_credential_error="Invalid credentials", ) if error_response is not None: - # Preserve the historical "Invalid credentials" wording for this route. - if error_response.status_code == 401: - return JSONResponse( - {"success": False, "error": "Invalid credentials"}, - status_code=401, - ) return error_response # The authenticated account must be the UID whose password is being deleted. diff --git a/tests/unit/test_management_app_password_endpoints.py b/tests/unit/test_management_app_password_endpoints.py index 7e191b16..7ad924b8 100644 --- a/tests/unit/test_management_app_password_endpoints.py +++ b/tests/unit/test_management_app_password_endpoints.py @@ -429,6 +429,33 @@ async def test_provision_app_password_ocs_v1_failure_payload_returns_401(mocker) assert "Invalid app password" in response.json()["error"] +async def test_provision_app_password_nextcloud_5xx_returns_502(mocker): + """A Nextcloud server error / maintenance mode (5xx) surfaces as 502, not a + misleading 401 that blames the user's password.""" + mocker.patch( + "nextcloud_mcp_server.api.passwords.get_settings", + return_value=MagicMock( + nextcloud_host="http://localhost:8080", + nextcloud_verify_ssl=True, + nextcloud_ca_bundle=None, + ), + ) + for status_code in (500, 503): + passwords._rate_limit_attempts.clear() + _mock_ocs_client(mocker, status_code=status_code, json_payload={}) + client = TestClient(_provision_only_app()) + response = client.post( + "/api/v1/users/testuser/app-password", + headers={ + "Authorization": create_basic_auth_header( + "testuser", "aaaaa-bbbbb-ccccc-ddddd-eeeee" + ) + }, + ) + assert response.status_code == 502, f"HTTP {status_code} should map to 502" + assert "server error" in response.json()["error"].lower() + + async def test_provision_app_password_nondict_data_does_not_500(mocker): """Regression for #824: a non-dict ``ocs.data`` under HTTP 200 (list, or ``null``) is handled gracefully (401), not as an unhandled 500/