Merge pull request #818 from cbcoutinho/fix/app-password-loginname-validation
fix(api): validate app password against Nextcloud using loginName, not UID
This commit is contained in:
@@ -238,6 +238,24 @@ async def provision_app_password(request: Request) -> JSONResponse:
|
|||||||
status_code=400,
|
status_code=400,
|
||||||
)
|
)
|
||||||
|
|
||||||
|
# Parse optional scopes and the Nextcloud loginName from the request body
|
||||||
|
# up front. Nextcloud authenticates app passwords against the *loginName*,
|
||||||
|
# which can differ from the UID — e.g. OIDC-provisioned users whose UID is
|
||||||
|
# their display name (UID "Chris Coutinho", loginName "chris@coutinho.io").
|
||||||
|
# Use the loginName for the BasicAuth validation below, falling back to the
|
||||||
|
# path user_id for legacy callers that don't send one (where UID ==
|
||||||
|
# loginName).
|
||||||
|
scopes = None
|
||||||
|
nc_username = None
|
||||||
|
try:
|
||||||
|
body = await request.json()
|
||||||
|
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
|
||||||
|
|
||||||
# Get Nextcloud host from settings
|
# Get Nextcloud host from settings
|
||||||
settings = get_settings()
|
settings = get_settings()
|
||||||
nextcloud_host = settings.nextcloud_host
|
nextcloud_host = settings.nextcloud_host
|
||||||
@@ -249,7 +267,10 @@ async def provision_app_password(request: Request) -> JSONResponse:
|
|||||||
status_code=500,
|
status_code=500,
|
||||||
)
|
)
|
||||||
|
|
||||||
# Validate app password against Nextcloud
|
# Validate app password against Nextcloud. BasicAuth places the user-id
|
||||||
|
# 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:
|
try:
|
||||||
async with nextcloud_httpx_client(
|
async with nextcloud_httpx_client(
|
||||||
timeout=NEXTCLOUD_VALIDATION_TIMEOUT
|
timeout=NEXTCLOUD_VALIDATION_TIMEOUT
|
||||||
@@ -258,7 +279,7 @@ async def provision_app_password(request: Request) -> JSONResponse:
|
|||||||
test_url = f"{nextcloud_host}/ocs/v1.php/cloud/user"
|
test_url = f"{nextcloud_host}/ocs/v1.php/cloud/user"
|
||||||
response = await client.get(
|
response = await client.get(
|
||||||
test_url,
|
test_url,
|
||||||
auth=(username, app_password),
|
auth=(login_name, app_password),
|
||||||
params={"format": "json"},
|
params={"format": "json"},
|
||||||
headers={"OCS-APIRequest": "true"},
|
headers={"OCS-APIRequest": "true"},
|
||||||
)
|
)
|
||||||
@@ -274,10 +295,11 @@ async def provision_app_password(request: Request) -> JSONResponse:
|
|||||||
status_code=401,
|
status_code=401,
|
||||||
)
|
)
|
||||||
|
|
||||||
# Verify the user ID from response matches
|
# 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()
|
data = response.json()
|
||||||
ocs_user_id = data.get("ocs", {}).get("data", {}).get("id")
|
ocs_user_id = data.get("ocs", {}).get("data", {}).get("id")
|
||||||
if ocs_user_id != username:
|
if ocs_user_id != path_user_id:
|
||||||
logger.warning("User ID mismatch in OCS response")
|
logger.warning("User ID mismatch in OCS response")
|
||||||
_record_rate_limit_attempt(path_user_id, success=False)
|
_record_rate_limit_attempt(path_user_id, success=False)
|
||||||
return JSONResponse(
|
return JSONResponse(
|
||||||
@@ -292,16 +314,6 @@ async def provision_app_password(request: Request) -> JSONResponse:
|
|||||||
status_code=500,
|
status_code=500,
|
||||||
)
|
)
|
||||||
|
|
||||||
# Parse optional scopes and username from request body
|
|
||||||
scopes = None
|
|
||||||
nc_username = None
|
|
||||||
try:
|
|
||||||
body = await request.json()
|
|
||||||
scopes = body.get("scopes") # list[str] | None
|
|
||||||
nc_username = body.get("username") # Nextcloud loginName
|
|
||||||
except Exception:
|
|
||||||
pass # No JSON body = legacy call without scopes
|
|
||||||
|
|
||||||
# Store the validated app password
|
# Store the validated app password
|
||||||
try:
|
try:
|
||||||
storage = await _get_app_password_storage(request)
|
storage = await _get_app_password_storage(request)
|
||||||
|
|||||||
@@ -248,6 +248,64 @@ async def test_provision_app_password_success(temp_storage, mocker):
|
|||||||
stored_password = await temp_storage.get_app_password("testuser")
|
stored_password = await temp_storage.get_app_password("testuser")
|
||||||
assert stored_password == "aaaaa-bbbbb-ccccc-ddddd-eeeee"
|
assert stored_password == "aaaaa-bbbbb-ccccc-ddddd-eeeee"
|
||||||
|
|
||||||
|
# Legacy callers send no loginName in the body → the OCS validation falls
|
||||||
|
# back to authenticating as the UID (here UID == loginName).
|
||||||
|
_, get_kwargs = mock_client.get.call_args
|
||||||
|
assert get_kwargs["auth"] == ("testuser", "aaaaa-bbbbb-ccccc-ddddd-eeeee")
|
||||||
|
|
||||||
|
|
||||||
|
async def test_provision_app_password_uses_loginname_not_uid(temp_storage, mocker):
|
||||||
|
"""Regression: when the Nextcloud UID differs from the loginName (e.g.
|
||||||
|
OIDC-provisioned users whose UID is their display name — UID
|
||||||
|
"Chris Coutinho", loginName "chris@coutinho.io"), the OCS BasicAuth
|
||||||
|
validation must authenticate as the loginName from the request body, not
|
||||||
|
the UID. Authenticating as the UID is rejected by Nextcloud with HTTP 401.
|
||||||
|
"""
|
||||||
|
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,
|
||||||
|
),
|
||||||
|
)
|
||||||
|
|
||||||
|
# OCS validation succeeds and reports the UID as the account id.
|
||||||
|
mock_response = MagicMock()
|
||||||
|
mock_response.status_code = 200
|
||||||
|
mock_response.json.return_value = {"ocs": {"data": {"id": "Chris Coutinho"}}}
|
||||||
|
|
||||||
|
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,
|
||||||
|
)
|
||||||
|
|
||||||
|
app = create_test_app(temp_storage)
|
||||||
|
client = TestClient(app)
|
||||||
|
|
||||||
|
pw = "aaaaa-bbbbb-ccccc-ddddd-eeeee"
|
||||||
|
# A literal space in the path is encoded by the client and decoded back to
|
||||||
|
# the UID; the BasicAuth username matches that UID.
|
||||||
|
response = client.post(
|
||||||
|
"/api/v1/users/Chris Coutinho/app-password",
|
||||||
|
headers={"Authorization": create_basic_auth_header("Chris Coutinho", pw)},
|
||||||
|
json={"username": "chris@coutinho.io"},
|
||||||
|
)
|
||||||
|
|
||||||
|
assert response.status_code == 200
|
||||||
|
assert response.json()["success"] is True
|
||||||
|
|
||||||
|
# The OCS BasicAuth used the loginName from the body, not the UID.
|
||||||
|
_, get_kwargs = mock_client.get.call_args
|
||||||
|
assert get_kwargs["auth"] == ("chris@coutinho.io", pw)
|
||||||
|
|
||||||
|
# Stored under the UID (the identity key).
|
||||||
|
assert await temp_storage.get_app_password("Chris Coutinho") == pw
|
||||||
|
|
||||||
|
|
||||||
async def test_provision_app_password_nextcloud_validation_fails(mocker):
|
async def test_provision_app_password_nextcloud_validation_fails(mocker):
|
||||||
"""Test that failed Nextcloud validation returns 401."""
|
"""Test that failed Nextcloud validation returns 401."""
|
||||||
|
|||||||
Reference in New Issue
Block a user