update PUT /Users endpoint

ee/api/chalicelib/core/users.py

@@ -143,6 +143,37 @@ def reset_member(tenant_id, editor_id, user_id_to_update):
     return {"data": {"invitationLink": generate_new_invitation(user_id_to_update)}}
 
 
+def update_scim_user(
+    user_id: int,
+    tenant_id: int,
+    email: str,
+):
+    with pg_client.PostgresClient() as cur:
+        cur.execute(
+            cur.mogrify(
+                """
+                WITH u AS (
+                    UPDATE public.users
+                    SET email = %(email)s
+                    WHERE
+                        users.user_id = %(user_id)s
+                        AND users.tenant_id = %(tenant_id)s
+                        AND users.deleted_at IS NULL
+                    RETURNING *
+                )
+                SELECT *
+                FROM u;
+                """,
+                {
+                    "tenant_id": tenant_id,
+                    "user_id": user_id,
+                    "email": email,
+                }
+            )
+        )
+        return helper.dict_to_camel_case(cur.fetchone())
+
+
 def update(tenant_id, user_id, changes, output=True):
     AUTH_KEYS = ["password", "invitationToken", "invitedAt", "changePwdExpireAt", "changePwdToken"]
     if len(changes.keys()) == 0:
@@ -1124,7 +1155,7 @@ def restore_scim_user(
                         api_key = default,
                         jwt_iat = NULL,
                         weekly_report = default
-                    WHERE user_id = %(user_id)s
+                    WHERE users.user_id = %(user_id)s
                     RETURNING *
                 )
                 SELECT *

ee/api/routers/scim.py

@@ -84,6 +84,18 @@ def _uniqueness_error_response():
     )
 
 
+def _mutability_error_response():
+    return JSONResponse(
+        status_code=400,
+        content={
+            "schemas": ["urn:ietf:params:scim:api:messages:2.0:Error"],
+            "detail": "The attempted modification is not compatible with the target attribute's mutability or current state.",
+            "status": "400",
+            "scimType": "mutability",
+        }
+    )
+
+
 @public_app.get("/ResourceTypes", dependencies=[Depends(auth_required)])
 async def get_resource_types(filter_param: str | None = Query(None, alias="filter")):
     if filter_param is not None:
@@ -329,42 +341,33 @@ async def create_user(r: UserRequest, tenant_id = Depends(auth_required)):
     return response
 
 
-@public_app.put("/Users/{user_id}", dependencies=[Depends(auth_required)])
+@public_app.put("/Users/{user_id}")
-def update_user(user_id: str, r: UserRequest):
+def update_user(user_id: str, r: UserRequest, tenant_id = Depends(auth_required)):
-    """Update SCIM User"""
+    db_resource = users.get_scim_user_by_id(user_id, tenant_id)
-    tenant_id = 1
+    if not db_resource:
-    user = users.get_by_uuid(user_id, tenant_id)
+        return _not_found_error_response(user_id)
-    if not user:
+    current_scim_resource = _convert_db_user_to_scim_user(db_resource).model_dump(mode="json", exclude_none=True)
-        raise HTTPException(status_code=404, detail="User not found")
+    changes = r.model_dump(mode="json")
-
+    schema = SCHEMA_IDS_TO_SCHEMA_DETAILS["urn:ietf:params:scim:schemas:core:2.0:User"]
-    changes = r.model_dump(mode='json', exclude={"schemas", "emails", "name", "locale", "groups", "password", "active"}) # some of these should be added later if necessary
-    nested_changes = r.model_dump(mode='json', include={"name", "emails"})
-    mapping = {"userName": "email", "displayName": "name", "externalId": "internal_id"} # mapping between scim schema field names and local database model, can be done as config?
-    for k, v in mapping.items():
-        if k in changes:
-            changes[v] = changes.pop(k)
-    changes["data"] = {}
-    for k, v in nested_changes.items():
-        value_to_insert = v[0] if k == "emails" else v
-        changes["data"][k] = value_to_insert
     try:
-        users.update(tenant_id, user["userId"], changes)
+        valid_mutable_changes = scim_helpers.filter_mutable_attributes(schema, changes, current_scim_resource)
-        res = UserResponse(
+    except ValueError:
-            schemas = ["urn:ietf:params:scim:schemas:core:2.0:User"],
+        # todo(jon): will need to add a test for this once we have an immutable field
-            id = user["data"]["userId"],
+        return _mutability_error_response()
-            userName = r.userName,
+    try:
-            name = r.name,
+        updated_db_resource = users.update_scim_user(
-            emails = r.emails,
+            user_id,
-            displayName = r.displayName,
+            tenant_id,
-            locale = r.locale,
+            email=valid_mutable_changes["userName"],
-            externalId = r.externalId,
-            active = r.active, # ignore for now, since, can't insert actual timestamp
-            groups = [], # ignore
         )
-
+        updated_scim_resource = _convert_db_user_to_scim_user(updated_db_resource)
-        return JSONResponse(status_code=201, content=res.model_dump(mode='json'))
+        return JSONResponse(
-    except Exception as e:
+            status_code=200,
-        raise HTTPException(status_code=500, detail=str(e))
+            content=updated_scim_resource.model_dump(mode="json", exclude_none=True),
+        )
+    except Exception:
+        # note(jon): for now, this is the only error that would happen when updating the scim user
+        return _uniqueness_error_response()
 
 
 @public_app.delete("/Users/{user_id}")

ee/api/routers/scim_helpers.py

@@ -103,3 +103,40 @@ def exclude_attributes(resource: dict[str, Any], exclude_list: list[str]) -> dic
             else:
                 new_resource[key] = value
     return new_resource
+
+
+def filter_mutable_attributes(schema: dict[str, Any], requested_changes: dict[str, Any], current: dict[str, Any]) -> dict[str, Any]:
+    attributes = {attr.get("name"): attr for attr in schema.get("attributes", [])}
+
+    valid_changes = {}
+
+    for attr_name, new_value in requested_changes.items():
+        attr_def = attributes.get(attr_name)
+        if not attr_def:
+            # Unknown attribute: ignore per RFC 7644
+            continue
+
+        mutability = attr_def.get("mutability", "readWrite")
+
+        if mutability == "readWrite" or mutability == "writeOnly":
+            valid_changes[attr_name] = new_value
+
+        elif mutability == "readOnly":
+            # Cannot modify read-only attributes: ignore
+            continue
+
+        elif mutability == "immutable":
+            # Only valid if the new value matches the current value exactly
+            current_value = current_values.get(attr_name)
+            if new_value != current_value:
+                raise ValueError(
+                    f"Attribute '{attr_name}' is immutable (cannot change). "
+                    f"Current value: {current_value!r}, attempted change: {new_value!r}"
+                )
+            # If it matches, no change is needed (already set)
+
+        else:
+            # Unknown mutability: default to safe behavior (ignore)
+            continue
+
+    return valid_changes