From 824686c67317b46ce009616ecd8b558e1ef49946 Mon Sep 17 00:00:00 2001 From: Agent Zero Date: Thu, 17 Sep 2026 22:58:38 +0200 Subject: [PATCH] =?UTF-8?q?fix(security):=20F02=20(Astra=20P0)=20=E2=80=94?= =?UTF-8?q?=20globale=20Login-Identitaet=20von=20Mandantenverwaltung=20tre?= =?UTF-8?q?nnen?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 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. --- app/routes/users.py | 42 ++++++++++- tests/test_user_service.py | 145 ++++++++++++++++++++++++++++++++++--- 2 files changed, 176 insertions(+), 11 deletions(-) diff --git a/app/routes/users.py b/app/routes/users.py index 6bc99f0..c5f9577 100644 --- a/app/routes/users.py +++ b/app/routes/users.py @@ -7,7 +7,7 @@ from typing import Any from fastapi import APIRouter, Depends, HTTPException, Query, Response, status from pydantic import BaseModel, Field -from sqlalchemy import select +from sqlalchemy import func, select from sqlalchemy.ext.asyncio import AsyncSession from app.core.audit import log_audit @@ -16,7 +16,7 @@ from app.core.db import get_db from app.core.notifications import post_system_message from app.core.permissions import invalidate_permission_cache from app.deps import get_current_user, require_permission -from app.models.user import User +from app.models.user import User, UserTenant from app.schemas.user import PaginatedUsers, UserCreate, UserResponse, UserUpdate from app.services.owner_transfer_service import transfer_ownership from app.services.user_service import _UNSET, user_service @@ -219,6 +219,44 @@ async def update_user( detail={"detail": "Only system admin can change system admin flag", "code": "admin_flag_forbidden"}, ) + # F02 (Astra P0): global identity fields vs. tenant administration. + # User.email, password and is_active live on the GLOBAL user record + # (shared across tenants). A tenant admin (users:write) must not change + # another member's global login identity: that would change the + # password-reset address / login credentials of a user who may also + # belong to other tenants. Allowed only as verified self-service or + # by a real system admin. + acting_is_system_admin = bool(current_user.get("is_system_admin")) + is_self = uid == acting_user_id + if not is_self and not acting_is_system_admin: + if body.email is not None or body.new_password is not None: + raise HTTPException( + status_code=status.HTTP_403_FORBIDDEN, + detail={ + "detail": "Global identity (email/password) can only be changed by the user themselves or a system admin", + "code": "global_identity_forbidden", + }, + ) + if body.is_active is not None: + # Global activation status: a tenant admin may deactivate a + # member of THEIR tenant, but only if the user belongs solely + # to this tenant. For multi-tenant users, deactivation here + # would lock them out of every other tenant too. + membership_q = await db.execute( + select(func.count()).select_from(UserTenant).where( + UserTenant.user_id == uid + ) + ) + membership_count = membership_q.scalar() or 0 + if membership_count > 1: + raise HTTPException( + status_code=status.HTTP_403_FORBIDDEN, + detail={ + "detail": "User belongs to multiple tenants — global activation status can only be changed by a system admin", + "code": "multi_tenant_status_forbidden", + }, + ) + # Determine if role_id was explicitly sent (Pydantic v2) role_id_sent = "role_id" in body.model_fields_set diff --git a/tests/test_user_service.py b/tests/test_user_service.py index 5cdd1dd..979a93e 100644 --- a/tests/test_user_service.py +++ b/tests/test_user_service.py @@ -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