Leopoldadmin/leocrm
1
0
Fork 0
Code Issues 363 Pull Requests Actions 1 Packages Projects Releases 1 Wiki Activity

fix(security): F02 (Astra P0) — globale Login-Identitaet von Mandantenverwaltung trennen

Browse Source 
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:
Agent Zero
2026-09-17 22:58:38 +02:00
parent f2a7206c7d
commit 824686c673
2 changed files with 176 additions and 11 deletions
Download Patch File Download Diff File Expand all files Collapse all files
app/routes/users.py
+40 -2
View File
@@ -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
tests/test_user_service.py
+136 -9
View File
@@ -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