fix(auth_v2): reject SCIM id mutation and SCIM-shape auth errors

PATCH no longer lets a scim:write principal reassign the read-only id
attribute (RFC 7643): any operation whose path targets id, or a no-path value
object carrying id, is rejected with 400 instead of rewriting the record
identifier and clobbering another resource. Authentication failures on the
guarded SCIM routes now render the SCIM Error schema (RFC 7644) for 401 and
403 via a router-scoped route class, preserving the WWW-Authenticate
challenge, rather than the generic {detail} body.
This commit is contained in:
Yassin Kortam 2026-06-10 19:38:20 -07:00
parent 422c4df16e
commit 75509686f7

View file

@ -1,9 +1,20 @@
from __future__ import annotations
from typing import TYPE_CHECKING, Any, Dict, Optional, Type, TypeVar, cast
from typing import (
TYPE_CHECKING,
Any,
Callable,
Coroutine,
Dict,
Optional,
Type,
TypeVar,
cast,
)
from fastapi import APIRouter, Query, Request, Response, Security, status
from fastapi import APIRouter, HTTPException, Query, Request, Response, Security, status
from fastapi.responses import JSONResponse
from fastapi.routing import APIRoute
from pydantic import ValidationError
from scim2_models import (
Bulk,
@ -38,6 +49,29 @@ def _error(status_code: int, detail: str) -> JSONResponse:
)
class _ScimRoute(APIRoute):
"""Render authentication failures with the SCIM Error schema (RFC 7644)."""
def get_route_handler(self) -> Callable[[Request], Coroutine[Any, Any, Response]]:
handler = super().get_route_handler()
async def scim_handler(request: Request) -> Response:
try:
return await handler(request)
except HTTPException as exc:
if exc.status_code not in (
status.HTTP_401_UNAUTHORIZED,
status.HTTP_403_FORBIDDEN,
):
raise
response = _error(exc.status_code, str(exc.detail))
if exc.headers:
response.headers.update(exc.headers)
return response
return scim_handler
async def _parse(request: Request, model: Type[R]) -> R:
body = await request.json()
return model.model_validate(body, scim_ctx=Context.RESOURCE_CREATION_REQUEST)
@ -66,12 +100,20 @@ def _remove_path(data: Dict[str, Any], path: str) -> None:
node.pop(keys[-1], None)
def _targets_read_only_id(op: Any) -> bool:
if op.path is not None:
return op.path.split(".")[0].strip().lower() == "id"
return isinstance(op.value, dict) and any(str(k).lower() == "id" for k in op.value)
def _apply_patch(resource: R, patch: PatchOp) -> R:
data: Dict[str, Any] = resource.model_dump()
for op in patch.operations:
action = op.op.value if hasattr(op.op, "value") else str(op.op)
if op.path is not None and ("[" in op.path or "]" in op.path):
raise ValueError(f"unsupported SCIM patch path filter: {op.path}")
if _targets_read_only_id(op):
raise ValueError("the SCIM id attribute is read-only")
if action == "remove":
if op.path:
_remove_path(data, op.path)
@ -90,6 +132,7 @@ def _dump(resource: Resource, ctx: Context) -> Dict[str, Any]:
def _build_protected_router(auth: AuthSecurity) -> APIRouter:
store = cast(ProvisioningStore, auth.resolver)
protected = APIRouter(
route_class=_ScimRoute,
dependencies=[Security(auth.principal, scopes=["scim:write"])],
)