diff --git a/app/routes/backups.py b/app/routes/backups.py index 407fb52..a4f6c46 100644 --- a/app/routes/backups.py +++ b/app/routes/backups.py @@ -8,7 +8,7 @@ from fastapi import APIRouter, Depends, HTTPException from sqlalchemy.ext.asyncio import AsyncSession 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.services import backup_service @@ -53,16 +53,22 @@ async def create_backup( @router.post( "/{backup_id}/restore", response_model=BackupResponse, - dependencies=[Depends(require_permission("automation:admin"))], + dependencies=[Depends(require_admin)], ) async def restore_backup( backup_id: str, db: AsyncSession = Depends(get_db), - current_user: dict = Depends(get_current_user), + current_user: dict = Depends(require_admin), ): """Restore a database backup. 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"]) try: diff --git a/tests/test_s1_security_guards.py b/tests/test_s1_security_guards.py index fdd401a..4cdb960 100644 --- a/tests/test_s1_security_guards.py +++ b/tests/test_s1_security_guards.py @@ -221,3 +221,45 @@ class TestF03SessionRevocation: assert _rejects("invited") is True # active membership → allow 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" + )