fix(security): F02 (Astra P0) — globale Login-Identitaet von Mandantenverwaltung trennen
Vorher: Ein Mandanten-Admin (users:write) konnte die globale User.email und das Passwort JEDES Mitglieds seines Mandanten aendern. User ist aber mandantenuebergreifend — derselbe Datensatz traegt Passwort und Systemadmin-Flag; der Passwort-Reset nutzt die veraenderbare Adresse. Ein Admin aus Mandant A konnte so die globale Reset-Adresse eines gemeinsamen Benutzers umlenken (Astra-Repro: globale Feldaenderung isoliert reproduziert). Fix (routes/users.py update_user): - email/new_password fuer FREMDE User -> 403 global_identity_forbidden (nur Selbstservice oder echter System-Admin) - is_active fuer MEHRMANDANTEN-User durch Tenant-Admin -> 403 multi_tenant_status_forbidden (Deaktivierung waere global sperrend; Single-Mandanten-Mitglieder duerfen wie bisher deaktiviert werden) - is_system_admin-Eskalationscheck unberuehrt (war schon korrekt) Abnahme (Astra): Ein Tenant-Verwalter kann weder die globale E-Mail- Adresse noch den globalen Aktivstatus eines gemeinsamen Benutzers veraendern — erfuellt. Tests: test_user_service.py 13/13 (5 neue F02-Tests: fremde E-Mail 403, fremdes Passwort 403, Mehrmandanten-Deaktivierung 403, Name-Aenderung bleibt 200, Selbstservice bleibt 200). ruff clean.
This commit is contained in:
+136
-9
@@ -8,7 +8,7 @@ from __future__ import annotations
|
||||
import pytest
|
||||
from httpx import AsyncClient
|
||||
|
||||
from tests.conftest import ORIGIN_HEADER, seed_tenant_and_users, login_client
|
||||
from tests.conftest import ORIGIN_HEADER, login_client, seed_tenant_and_users
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
@@ -17,7 +17,7 @@ class TestUserServiceCRUD:
|
||||
|
||||
async def test_list_users_as_admin(self, client: AsyncClient, db_session):
|
||||
"""Admin can list users in their tenant."""
|
||||
seed = await seed_tenant_and_users(db_session)
|
||||
await seed_tenant_and_users(db_session)
|
||||
await login_client(client, "admin@tenanta.com")
|
||||
|
||||
resp = await client.get("/api/v1/users", headers=ORIGIN_HEADER)
|
||||
@@ -31,7 +31,7 @@ class TestUserServiceCRUD:
|
||||
|
||||
async def test_list_users_as_viewer_forbidden(self, client: AsyncClient, db_session):
|
||||
"""Viewer may or may not list users depending on default permissions."""
|
||||
seed = await seed_tenant_and_users(db_session)
|
||||
await seed_tenant_and_users(db_session)
|
||||
await login_client(client, "viewer@tenanta.com")
|
||||
|
||||
resp = await client.get("/api/v1/users", headers=ORIGIN_HEADER)
|
||||
@@ -39,7 +39,7 @@ class TestUserServiceCRUD:
|
||||
|
||||
async def test_create_user_as_admin(self, client: AsyncClient, db_session):
|
||||
"""Admin can create a new user in their tenant."""
|
||||
seed = await seed_tenant_and_users(db_session)
|
||||
await seed_tenant_and_users(db_session)
|
||||
await login_client(client, "admin@tenanta.com")
|
||||
|
||||
resp = await client.post(
|
||||
@@ -58,7 +58,7 @@ class TestUserServiceCRUD:
|
||||
|
||||
async def test_create_user_as_viewer_forbidden(self, client: AsyncClient, db_session):
|
||||
"""Viewer cannot create users."""
|
||||
seed = await seed_tenant_and_users(db_session)
|
||||
await seed_tenant_and_users(db_session)
|
||||
await login_client(client, "viewer@tenanta.com")
|
||||
|
||||
resp = await client.post(
|
||||
@@ -75,7 +75,7 @@ class TestUserServiceCRUD:
|
||||
|
||||
async def test_create_user_duplicate_email(self, client: AsyncClient, db_session):
|
||||
"""Cannot create user with existing email — should return error, not 500."""
|
||||
seed = await seed_tenant_and_users(db_session)
|
||||
await seed_tenant_and_users(db_session)
|
||||
await login_client(client, "admin@tenanta.com")
|
||||
|
||||
resp = await client.post(
|
||||
@@ -92,7 +92,7 @@ class TestUserServiceCRUD:
|
||||
|
||||
async def test_update_user_as_admin(self, client: AsyncClient, db_session):
|
||||
"""Admin can update a user."""
|
||||
seed = await seed_tenant_and_users(db_session)
|
||||
await seed_tenant_and_users(db_session)
|
||||
await login_client(client, "admin@tenanta.com")
|
||||
|
||||
# Get user list first
|
||||
@@ -112,7 +112,7 @@ class TestUserServiceCRUD:
|
||||
|
||||
async def test_delete_user_as_admin(self, client: AsyncClient, db_session):
|
||||
"""Admin can delete a user."""
|
||||
seed = await seed_tenant_and_users(db_session)
|
||||
await seed_tenant_and_users(db_session)
|
||||
await login_client(client, "admin@tenanta.com")
|
||||
|
||||
resp = await client.get("/api/v1/users", headers=ORIGIN_HEADER)
|
||||
@@ -128,7 +128,7 @@ class TestUserServiceCRUD:
|
||||
|
||||
async def test_cross_tenant_user_isolation(self, client: AsyncClient, db_session):
|
||||
"""Admin A cannot see users from tenant B."""
|
||||
seed = await seed_tenant_and_users(db_session)
|
||||
await seed_tenant_and_users(db_session)
|
||||
await login_client(client, "admin@tenanta.com")
|
||||
|
||||
resp = await client.get("/api/v1/users", headers=ORIGIN_HEADER)
|
||||
@@ -137,3 +137,130 @@ class TestUserServiceCRUD:
|
||||
users = data if isinstance(data, list) else data.get("items", [])
|
||||
emails = [u["email"] for u in users]
|
||||
assert "admin@tenantb.com" not in emails
|
||||
|
||||
|
||||
class TestF02GlobalIdentityGuard:
|
||||
"""F02 (Astra P0): tenant administration must not change global login
|
||||
identity fields of other users.
|
||||
|
||||
User is a GLOBAL record shared across tenants (password, email,
|
||||
is_system_admin). A tenant admin (users:write) may manage membership
|
||||
and tenant-scoped fields — but must NOT change another member's
|
||||
global login identity (password-reset address, credentials) or the
|
||||
global activation status of a multi-tenant user.
|
||||
"""
|
||||
|
||||
async def _get_user(self, client: AsyncClient, email: str) -> dict:
|
||||
resp = await client.get("/api/v1/users", headers=ORIGIN_HEADER)
|
||||
assert resp.status_code == 200
|
||||
data = resp.json()
|
||||
users = data if isinstance(data, list) else data.get("items", [])
|
||||
return next(u for u in users if u["email"] == email)
|
||||
|
||||
async def test_tenant_admin_cannot_change_foreign_email(self, client: AsyncClient, db_session):
|
||||
"""Tenant admin changing ANOTHER member's email → 403."""
|
||||
await seed_tenant_and_users(db_session)
|
||||
await login_client(client, "admin@tenanta.com")
|
||||
viewer = await self._get_user(client, "viewer@tenanta.com")
|
||||
|
||||
resp = await client.patch(
|
||||
f"/api/v1/users/{viewer['id']}",
|
||||
json={"email": "attacker-controlled@evil.de"},
|
||||
headers=ORIGIN_HEADER,
|
||||
)
|
||||
assert resp.status_code == 403
|
||||
assert resp.json().get("detail", {}).get("code") in ("global_identity_forbidden",)
|
||||
|
||||
async def test_tenant_admin_cannot_change_foreign_password(self, client: AsyncClient, db_session):
|
||||
"""Tenant admin setting ANOTHER member's password → 403."""
|
||||
await seed_tenant_and_users(db_session)
|
||||
await login_client(client, "admin@tenanta.com")
|
||||
viewer = await self._get_user(client, "viewer@tenanta.com")
|
||||
|
||||
resp = await client.patch(
|
||||
f"/api/v1/users/{viewer['id']}",
|
||||
json={"new_password": "NewHackedPass123!", "current_password": "whatever"},
|
||||
headers=ORIGIN_HEADER,
|
||||
)
|
||||
assert resp.status_code == 403
|
||||
assert resp.json().get("detail", {}).get("code") in ("global_identity_forbidden",)
|
||||
|
||||
async def test_tenant_admin_cannot_deactivate_multi_tenant_user(self, client: AsyncClient, db_session):
|
||||
"""admin@tenanta.com is seeded with a SECOND membership (tenant B).
|
||||
A tenant admin deactivating this multi-tenant user's GLOBAL status → 403.
|
||||
|
||||
Deactivation would lock the user out of every tenant, not just this one.
|
||||
"""
|
||||
seed = await seed_tenant_and_users(db_session)
|
||||
# editor_a edits the multi-tenant admin_a — editors lack users:write,
|
||||
# so log in as admin_a won't work (self-service rules differ).
|
||||
# Instead: viewer_a is single-tenant, admin_a is multi-tenant.
|
||||
# We need a second admin in tenant A without extra fixtures: use the
|
||||
# system-admin path — a non-system-admin tenant admin tries to
|
||||
# deactivate the multi-tenant admin_a from tenant A.
|
||||
await login_client(client, "admin@tenanta.com")
|
||||
# admin_a changing their OWN status hits self-modification prevention
|
||||
# first — so instead verify via a fresh single-tenant admin in A.
|
||||
# The seeded admin_a is multi-tenant; add a temporary admin:
|
||||
from app.core.auth import hash_password
|
||||
from app.models.user import User, UserTenant
|
||||
|
||||
tenant_a_id = seed["tenant_a"].id
|
||||
admin_role_id = seed["admin_role_a"].id
|
||||
other_admin = User(
|
||||
email="otheradmin@tenanta.com",
|
||||
name="Other Admin",
|
||||
password_hash=hash_password("TestPass123!"),
|
||||
is_active=True,
|
||||
preferences={},
|
||||
)
|
||||
db_session.add(other_admin)
|
||||
await db_session.flush()
|
||||
ut = UserTenant(
|
||||
user_id=other_admin.id,
|
||||
tenant_id=tenant_a_id,
|
||||
is_default=True,
|
||||
role="admin",
|
||||
role_id=admin_role_id,
|
||||
)
|
||||
db_session.add(ut)
|
||||
await db_session.commit()
|
||||
|
||||
await login_client(client, "otheradmin@tenanta.com")
|
||||
admin_a = await self._get_user(client, "admin@tenanta.com")
|
||||
|
||||
resp = await client.patch(
|
||||
f"/api/v1/users/{admin_a['id']}",
|
||||
json={"is_active": False},
|
||||
headers=ORIGIN_HEADER,
|
||||
)
|
||||
assert resp.status_code == 403
|
||||
assert resp.json().get("detail", {}).get("code") in ("multi_tenant_status_forbidden",)
|
||||
|
||||
async def test_tenant_admin_can_still_rename_member(self, client: AsyncClient, db_session):
|
||||
"""Regression: tenant administration keeps working for tenant-scoped
|
||||
fields (name, role) — the guard only protects global identity."""
|
||||
await seed_tenant_and_users(db_session)
|
||||
await login_client(client, "admin@tenanta.com")
|
||||
viewer = await self._get_user(client, "viewer@tenanta.com")
|
||||
|
||||
resp = await client.patch(
|
||||
f"/api/v1/users/{viewer['id']}",
|
||||
json={"name": "Renamed Viewer"},
|
||||
headers=ORIGIN_HEADER,
|
||||
)
|
||||
assert resp.status_code == 200
|
||||
assert resp.json()["name"] == "Renamed Viewer"
|
||||
|
||||
async def test_self_service_email_change_still_allowed(self, client: AsyncClient, db_session):
|
||||
"""Self-service: users can still change THEIR OWN email."""
|
||||
await seed_tenant_and_users(db_session)
|
||||
await login_client(client, "admin@tenanta.com")
|
||||
me = await self._get_user(client, "admin@tenanta.com")
|
||||
|
||||
resp = await client.patch(
|
||||
f"/api/v1/users/{me['id']}",
|
||||
json={"email": "admin@tenanta.com"}, # same value, but goes through the guard
|
||||
headers=ORIGIN_HEADER,
|
||||
)
|
||||
assert resp.status_code == 200
|
||||
|
||||
Reference in New Issue
Block a user