Merge pull request #820 from dataelement/agent/debugger/9d168ae7

fix: prevent cross-tenant organization user updates

Author: yaojin3616

backend/app/api/organization.py

@@ -18,6 +18,11 @@
 router = APIRouter(prefix="/org", tags=["organization"])
 
 
+def _is_platform_admin(user: User) -> bool:
+    """Return whether the caller has platform-wide administrative authority."""
+    return user.role == "platform_admin" or bool(getattr(user.identity, "is_platform_admin", False))
+
+
 # ─── Users Management ──────────────────────────────────
 
 @router.get("/users", response_model=list[UserOut])
@@ -30,11 +35,11 @@ async def list_users(
     query = (
         select(User)
         .options(selectinload(User.identity))
-        .where(User.is_active == True)
+        .where(User.is_active)
     )
 
     target_tenant_id = current_user.tenant_id
-    if current_user.role in ("platform_admin", "org_admin") and tenant_id:
+    if _is_platform_admin(current_user) and tenant_id:
         target_tenant_id = tenant_id
     if target_tenant_id:
         query = query.where(User.tenant_id == target_tenant_id)
@@ -52,17 +57,31 @@ async def admin_update_user(
     db: Any = None,
 ):
     """Admin update user profile."""
-    result = await query_dao.execute(db, 
+    query = (
         select(User)
         .options(selectinload(User.identity))
         .where(User.id == user_id)
     )
+    if not _is_platform_admin(current_user):
+        query = query.where(User.tenant_id == current_user.tenant_id)
+
+    result = await query_dao.execute(db, query)
     user = result.scalar_one_or_none()
     if not user:
         raise HTTPException(status_code=404, detail="User not found")
 
     update_data = data.model_dump(exclude_unset=True)
 
+    # Email is stored on the globally shared Identity rather than the tenant User.
+    # An organization administrator must not be able to alter another member's
+    # login and password-reset address, even if that member belongs to this tenant.
+    if (
+        "email" in update_data
+        and not _is_platform_admin(current_user)
+        and user.identity_id != current_user.identity_id
+    ):
+        raise HTTPException(status_code=403, detail="Cannot modify another user's login email")
+
     # Validate email uniqueness within tenant if changing
     if "email" in update_data and update_data["email"] != user.email:
         existing = await query_dao.execute(db, 

backend/tests/test_organization_tenant_scope.py

@@ -0,0 +1,93 @@
+import uuid
+from types import SimpleNamespace
+
+import pytest
+from fastapi import HTTPException
+
+from app.api import organization
+from app.schemas.schemas import UserUpdate
+
+
+class DummyResult:
+    def __init__(self, value=None):
+        self.value = value
+
+    def scalar_one_or_none(self):
+        return self.value
+
+    def scalars(self):
+        return self
+
+    def all(self):
+        return []
+
+
+class RecordingDB:
+    def __init__(self, responses):
+        self.responses = list(responses)
+        self.statements = []
+
+    async def execute(self, statement):
+        self.statements.append(statement)
+        return self.responses.pop(0)
+
+
+def _org_admin(*, tenant_id: uuid.UUID, identity_id: uuid.UUID) -> SimpleNamespace:
+    return SimpleNamespace(
+        role="org_admin",
+        tenant_id=tenant_id,
+        identity_id=identity_id,
+        identity=SimpleNamespace(is_platform_admin=False),
+    )
+
+
+@pytest.mark.asyncio
+async def test_org_admin_cannot_load_user_from_another_tenant_for_update() -> None:
+    tenant_id = uuid.uuid4()
+    db = RecordingDB([DummyResult()])
+
+    with pytest.raises(HTTPException) as raised:
+        await organization.admin_update_user(
+            user_id=uuid.uuid4(),
+            data=UserUpdate(display_name="Changed"),
+            current_user=_org_admin(tenant_id=tenant_id, identity_id=uuid.uuid4()),
+            db=db,
+        )
+
+    assert raised.value.status_code == 404
+    assert "users.tenant_id" in str(db.statements[0])
+
+
+@pytest.mark.asyncio
+async def test_org_admin_cannot_list_users_from_another_tenant() -> None:
+    tenant_id = uuid.uuid4()
+    requested_tenant_id = uuid.uuid4()
+    db = RecordingDB([DummyResult()])
+
+    users = await organization.list_users(
+        tenant_id=requested_tenant_id,
+        current_user=_org_admin(tenant_id=tenant_id, identity_id=uuid.uuid4()),
+        db=db,
+    )
+
+    assert users == []
+    assert db.statements[0].compile().params["tenant_id_1"] == tenant_id
+
+
+@pytest.mark.asyncio
+async def test_org_admin_cannot_change_another_members_global_login_email() -> None:
+    tenant_id = uuid.uuid4()
+    current_identity_id = uuid.uuid4()
+    target = SimpleNamespace(id=uuid.uuid4(), tenant_id=tenant_id, identity_id=uuid.uuid4())
+    db = RecordingDB([DummyResult(target)])
+
+    with pytest.raises(HTTPException) as raised:
+        await organization.admin_update_user(
+            user_id=target.id,
+            data=UserUpdate(email="new-address@example.com"),
+            current_user=_org_admin(tenant_id=tenant_id, identity_id=current_identity_id),
+            db=db,
+        )
+
+    assert raised.value.status_code == 403
+    assert raised.value.detail == "Cannot modify another user's login email"