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) <noreply@anthropic.com>
This commit is contained in:
co-authored by
Claude Opus 4.8
parent
a11a4e709c
commit
3c55d03ecf
@@ -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)
|
||||
|
||||
@@ -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(
|
||||
|
||||
@@ -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(
|
||||
|
||||
Reference in New Issue
Block a user