Merge pull request #825 from cbcoutinho/fix/provision-app-password-ocs-nondict-824
fix(api): return 401 not 500 on failed app-password OCS validation (#824)
This commit is contained in:
@@ -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,111 @@ async def _get_app_password_storage(request: Request) -> RefreshTokenStorage:
|
||||
return storage
|
||||
|
||||
|
||||
async def _validate_nextcloud_credentials(
|
||||
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.
|
||||
|
||||
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).
|
||||
|
||||
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, errors out (5xx /
|
||||
maintenance mode), 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. 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_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
|
||||
# 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_credential_error},
|
||||
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.
|
||||
|
||||
@@ -249,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
|
||||
|
||||
@@ -271,34 +382,15 @@ 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"},
|
||||
)
|
||||
|
||||
if response.status_code != 200:
|
||||
logger.warning(
|
||||
"App password validation failed for user: HTTP %s",
|
||||
response.status_code,
|
||||
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 JSONResponse(
|
||||
{"success": False, "error": "Invalid app password"},
|
||||
status_code=401,
|
||||
)
|
||||
return error_response
|
||||
|
||||
# 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")
|
||||
# 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)
|
||||
@@ -307,13 +399,6 @@ async def provision_app_password(request: Request) -> JSONResponse:
|
||||
status_code=403,
|
||||
)
|
||||
|
||||
except httpx.RequestError as e:
|
||||
logger.error("Failed to validate app password: %s", e)
|
||||
return JSONResponse(
|
||||
{"success": False, "error": "Failed to validate credentials"},
|
||||
status_code=500,
|
||||
)
|
||||
|
||||
# Store the validated app password
|
||||
try:
|
||||
storage = await _get_app_password_storage(request)
|
||||
@@ -402,32 +487,55 @@ 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()
|
||||
except (ValueError, UnicodeDecodeError):
|
||||
body = None # No / malformed JSON body = legacy call without a loginName
|
||||
if isinstance(body, dict):
|
||||
nc_username = body.get("username")
|
||||
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 not nextcloud_host:
|
||||
logger.error("NEXTCLOUD_HOST not configured")
|
||||
return JSONResponse(
|
||||
{"success": False, "error": "Server not configured"},
|
||||
status_code=500,
|
||||
)
|
||||
|
||||
if response.status_code != 200:
|
||||
return JSONResponse(
|
||||
{"success": False, "error": "Invalid credentials"},
|
||||
status_code=401,
|
||||
# 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,
|
||||
invalid_credential_error="Invalid credentials",
|
||||
)
|
||||
except httpx.RequestError as e:
|
||||
logger.error("Failed to validate credentials: %s", e)
|
||||
if error_response is not None:
|
||||
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": "Failed to validate credentials"},
|
||||
status_code=500,
|
||||
{"success": False, "error": "User ID mismatch"},
|
||||
status_code=403,
|
||||
)
|
||||
|
||||
try:
|
||||
|
||||
@@ -0,0 +1,91 @@
|
||||
"""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 base64
|
||||
|
||||
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. 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:
|
||||
credentials = base64.b64encode(f"{username}:{password}".encode()).decode()
|
||||
return f"Basic {credentials}"
|
||||
|
||||
|
||||
@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", "")
|
||||
@@ -356,6 +356,242 @@ 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_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/
|
||||
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_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
|
||||
@@ -441,9 +677,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 +728,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 +812,98 @@ 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_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