Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
25 changes: 22 additions & 3 deletions backend/app/api/organization.py
Original file line number Diff line number Diff line change
Expand Up @@ -17,6 +17,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])
Expand All @@ -29,11 +34,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)
Expand All @@ -51,17 +56,31 @@ async def admin_update_user(
db: AsyncSession = Depends(get_db),
):
"""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,
Expand Down
93 changes: 93 additions & 0 deletions backend/tests/test_organization_tenant_scope.py
Original file line number Diff line number Diff line change
@@ -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"