fix(security): F23 (Astra P1) — DB-weiter Restore nur noch fuer System-Admins
Vorher: POST /api/v1/backups/{id}/restore war ueber automation:admin
eines Mandanten erreichbar — der Restore bearbeitet aber die GESAMTE
geteilte Datenbank ohne Mandantenfilter. Ein Tenant-Admin haette den
Zustand aller Mandanten ueberschreiben koennen.
Fix: Restore-Route auf require_admin umgestellt (echter System-Admin:
is_system_admin oder *:* via RBAC). Listen/Erstellen/Loeschen von
Backups bleibt mandantenbezogen auf automation:admin.
Abnahme (Astra): Ein Tenant-Admin kann keinen Gesamtrestore ausloesen —
erfuellt (Route-Introspektions-Tests pinnen die Verdrahtung).
Tests: test_s1_security_guards.py 12/12 (2 neue F23-Tests: restore nutzt
require_admin, restore nutzt NICHT require_permission).
This commit is contained in:
@@ -8,7 +8,7 @@ from fastapi import APIRouter, Depends, HTTPException
|
|||||||
from sqlalchemy.ext.asyncio import AsyncSession
|
from sqlalchemy.ext.asyncio import AsyncSession
|
||||||
|
|
||||||
from app.core.db import get_db
|
from app.core.db import get_db
|
||||||
from app.deps import get_current_user, require_permission
|
from app.deps import get_current_user, require_admin, require_permission
|
||||||
from app.schemas.backup import BackupListResponse, BackupResponse
|
from app.schemas.backup import BackupListResponse, BackupResponse
|
||||||
from app.services import backup_service
|
from app.services import backup_service
|
||||||
|
|
||||||
@@ -53,16 +53,22 @@ async def create_backup(
|
|||||||
@router.post(
|
@router.post(
|
||||||
"/{backup_id}/restore",
|
"/{backup_id}/restore",
|
||||||
response_model=BackupResponse,
|
response_model=BackupResponse,
|
||||||
dependencies=[Depends(require_permission("automation:admin"))],
|
dependencies=[Depends(require_admin)],
|
||||||
)
|
)
|
||||||
async def restore_backup(
|
async def restore_backup(
|
||||||
backup_id: str,
|
backup_id: str,
|
||||||
db: AsyncSession = Depends(get_db),
|
db: AsyncSession = Depends(get_db),
|
||||||
current_user: dict = Depends(get_current_user),
|
current_user: dict = Depends(require_admin),
|
||||||
):
|
):
|
||||||
"""Restore a database backup.
|
"""Restore a database backup.
|
||||||
|
|
||||||
WARNING: This is a destructive operation. It drops and recreates the database.
|
WARNING: This is a destructive operation. It drops and recreates the database.
|
||||||
|
|
||||||
|
F23 (Astra P1): a full-database restore is a GLOBAL operations action —
|
||||||
|
it affects every tenant. A tenant admin (automation:admin) must not be
|
||||||
|
able to trigger it: the restore replaces the shared database, not just
|
||||||
|
this tenant's rows. Requires a real system admin (is_system_admin or
|
||||||
|
*:* via the RBAC system).
|
||||||
"""
|
"""
|
||||||
tenant_id = uuid.UUID(current_user["tenant_id"])
|
tenant_id = uuid.UUID(current_user["tenant_id"])
|
||||||
try:
|
try:
|
||||||
|
|||||||
@@ -221,3 +221,45 @@ class TestF03SessionRevocation:
|
|||||||
assert _rejects("invited") is True
|
assert _rejects("invited") is True
|
||||||
# active membership → allow
|
# active membership → allow
|
||||||
assert _rejects("active") is False
|
assert _rejects("active") is False
|
||||||
|
|
||||||
|
|
||||||
|
@pytest.mark.asyncio
|
||||||
|
class TestF23RestoreRequiresSystemAdmin:
|
||||||
|
"""F23 (Astra P1): a full-database restore is a GLOBAL operations
|
||||||
|
action — a tenant admin (automation:admin) must not be able to
|
||||||
|
trigger it. Route introspection pins the dependency wiring.
|
||||||
|
"""
|
||||||
|
|
||||||
|
def _find_restore_route(self):
|
||||||
|
"""Find the POST /{backup_id}/restore route on the backups router."""
|
||||||
|
from app.routes.backups import router
|
||||||
|
|
||||||
|
for route in router.routes:
|
||||||
|
if "restore" in getattr(route, "path", ""):
|
||||||
|
return route
|
||||||
|
return None
|
||||||
|
|
||||||
|
async def test_restore_route_uses_require_admin(self):
|
||||||
|
from app.deps import require_admin
|
||||||
|
|
||||||
|
route = self._find_restore_route()
|
||||||
|
assert route is not None, "restore route not found"
|
||||||
|
deps = list(getattr(route, "dependencies", []))
|
||||||
|
assert any(getattr(d, "dependency", None) is require_admin for d in deps), (
|
||||||
|
"F23: restore route must depend on require_admin (system admin), "
|
||||||
|
"not automation:admin"
|
||||||
|
)
|
||||||
|
|
||||||
|
async def test_restore_route_not_tenant_admin(self):
|
||||||
|
from app.deps import require_permission
|
||||||
|
|
||||||
|
route = self._find_restore_route()
|
||||||
|
assert route is not None
|
||||||
|
deps = list(getattr(route, "dependencies", []))
|
||||||
|
for d in deps:
|
||||||
|
dep_fn = getattr(d, "dependency", None)
|
||||||
|
assert dep_fn is not require_permission, (
|
||||||
|
"F23: restore route must not use require_permission("
|
||||||
|
"automation:admin) — a tenant admin must not trigger a "
|
||||||
|
"full-database restore"
|
||||||
|
)
|
||||||
|
|||||||
Reference in New Issue
Block a user