# Copyright (c) 2025 Patrick Motsch # All rights reserved. """ Security Administration routes. MULTI-TENANT: These are SYSTEM-LEVEL operations requiring isSysAdmin=true. No mandate context - SysAdmin manages infrastructure, not data. """ from fastapi import APIRouter, HTTPException, Depends, status, Request, Body from fastapi.responses import FileResponse, JSONResponse from typing import Optional, Dict, Any, List import os import logging from modules.auth import getCurrentUser, limiter, requireSysAdmin from modules.connectors.connectorDbPostgre import DatabaseConnector from modules.interfaces.interfaceDbApp import getRootInterface from modules.datamodels.datamodelUam import User, UserInDB, AuthAuthority from modules.datamodels.datamodelSecurity import Token from modules.shared.configuration import APP_CONFIG logger = logging.getLogger(__name__) router = APIRouter( prefix="/api/admin", tags=["Security Administration"], responses={ 404: {"description": "Not found"}, 400: {"description": "Bad request"}, 401: {"description": "Unauthorized"}, 403: {"description": "Forbidden"}, 500: {"description": "Internal server error"} } ) def _getPoweronDatabases() -> List[str]: """Load databases from PostgreSQL host matching poweron_%.""" dbHost = APP_CONFIG.get("DB_HOST") dbUser = APP_CONFIG.get("DB_USER") dbPassword = APP_CONFIG.get("DB_PASSWORD_SECRET") dbPort = int(APP_CONFIG.get("DB_PORT", 5432)) # Connect to 'postgres' system database to query all databases connector = DatabaseConnector( dbHost=dbHost, dbDatabase="postgres", dbUser=dbUser, dbPassword=dbPassword, dbPort=dbPort, userId=None ) try: with connector.connection.cursor() as cursor: cursor.execute( """ SELECT datname FROM pg_database WHERE datname LIKE 'poweron_%' AND datistemplate = false ORDER BY datname """ ) rows = cursor.fetchall() return [row["datname"] for row in rows if row.get("datname")] finally: connector.close() def _getDatabaseConnector(databaseName: str, userId: str = None) -> DatabaseConnector: """ Create a generic DatabaseConnector for any poweron_* database. Fully dynamic - no interface mapping needed. """ if not databaseName.startswith("poweron_"): raise ValueError(f"Invalid database name: {databaseName}") dbHost = APP_CONFIG.get("DB_HOST") dbUser = APP_CONFIG.get("DB_USER") dbPassword = APP_CONFIG.get("DB_PASSWORD_SECRET") dbPort = int(APP_CONFIG.get("DB_PORT", 5432)) connector = DatabaseConnector( dbHost=dbHost, dbDatabase=databaseName, dbUser=dbUser, dbPassword=dbPassword, dbPort=dbPort, userId=userId ) return connector # ---------------------- # Token listing and revocation # ---------------------- @router.get("/tokens") @limiter.limit("30/minute") def list_tokens( request: Request, currentUser: User = Depends(requireSysAdmin), userId: Optional[str] = None, authority: Optional[str] = None, sessionId: Optional[str] = None, statusFilter: Optional[str] = None, connectionId: Optional[str] = None, ) -> List[Dict[str, Any]]: """ List all tokens in the system. MULTI-TENANT: SysAdmin-only, no mandate filter (system-level view). """ try: appInterface = getRootInterface() recordFilter: Dict[str, Any] = {} if userId: recordFilter["userId"] = userId if authority: recordFilter["authority"] = authority if sessionId: recordFilter["sessionId"] = sessionId if connectionId: recordFilter["connectionId"] = connectionId if statusFilter: recordFilter["status"] = statusFilter # MULTI-TENANT: SysAdmin sees ALL tokens (no mandate filter) # Use interface method to get tokens with flexible filtering tokens = appInterface.getAllTokens(recordFilter=recordFilter) return tokens except HTTPException: raise except Exception as e: logger.error(f"Error listing tokens: {str(e)}") raise HTTPException(status_code=500, detail="Failed to list tokens") @router.post("/tokens/revoke/user") @limiter.limit("30/minute") def revoke_tokens_by_user( request: Request, currentUser: User = Depends(requireSysAdmin), payload: Dict[str, Any] = Body(...) ) -> Dict[str, Any]: """ Revoke all tokens for a user. MULTI-TENANT: SysAdmin-only, can revoke across all mandates. """ try: userId = payload.get("userId") authority = payload.get("authority") reason = payload.get("reason", "sysadmin revoke") if not userId: raise HTTPException(status_code=400, detail="userId is required") appInterface = getRootInterface() # MULTI-TENANT: SysAdmin can revoke any user's tokens (no mandate restriction) count = appInterface.revokeTokensByUser( userId=userId, authority=AuthAuthority(authority) if authority else None, mandateId=None, # SysAdmin: no mandate filter revokedBy=currentUser.id, reason=reason ) return {"revoked": count} except HTTPException: raise except Exception as e: logger.error(f"Error revoking tokens by user: {str(e)}") raise HTTPException(status_code=500, detail="Failed to revoke tokens") @router.post("/tokens/revoke/session") @limiter.limit("30/minute") def revoke_tokens_by_session( request: Request, currentUser: User = Depends(requireSysAdmin), payload: Dict[str, Any] = Body(...) ) -> Dict[str, Any]: """ Revoke all tokens for a specific session. MULTI-TENANT: SysAdmin-only. """ try: userId = payload.get("userId") sessionId = payload.get("sessionId") authority = payload.get("authority", "local") reason = payload.get("reason", "sysadmin session revoke") if not userId or not sessionId: raise HTTPException(status_code=400, detail="userId and sessionId are required") appInterface = getRootInterface() # MULTI-TENANT: SysAdmin can revoke any session (no mandate check) count = appInterface.revokeTokensBySessionId( sessionId=sessionId, userId=userId, authority=AuthAuthority(authority), revokedBy=currentUser.id, reason=reason ) return {"revoked": count} except HTTPException: raise except Exception as e: logger.error(f"Error revoking tokens by session: {str(e)}") raise HTTPException(status_code=500, detail="Failed to revoke session tokens") @router.post("/tokens/revoke/id") @limiter.limit("30/minute") def revoke_token_by_id( request: Request, currentUser: User = Depends(requireSysAdmin), payload: Dict[str, Any] = Body(...) ) -> Dict[str, Any]: """ Revoke a specific token by ID. MULTI-TENANT: SysAdmin-only. """ try: tokenId = payload.get("tokenId") reason = payload.get("reason", "sysadmin revoke") if not tokenId: raise HTTPException(status_code=400, detail="tokenId is required") appInterface = getRootInterface() # MULTI-TENANT: SysAdmin can revoke any token (no mandate check) ok = appInterface.revokeTokenById(tokenId, revokedBy=currentUser.id, reason=reason) return {"revoked": 1 if ok else 0} except HTTPException: raise except Exception as e: logger.error(f"Error revoking token by id: {str(e)}") raise HTTPException(status_code=500, detail="Failed to revoke token") @router.post("/tokens/revoke/mandate") @limiter.limit("10/minute") def revoke_tokens_by_mandate( request: Request, currentUser: User = Depends(requireSysAdmin), payload: Dict[str, Any] = Body(...) ) -> Dict[str, Any]: """ Revoke all tokens for users in a mandate. MULTI-TENANT: SysAdmin-only, can revoke tokens for any mandate. """ try: mandateId = payload.get("mandateId") authority = payload.get("authority", "local") reason = payload.get("reason", "sysadmin mandate revoke") if not mandateId: raise HTTPException(status_code=400, detail="mandateId is required") # MULTI-TENANT: SysAdmin can revoke tokens for any mandate appInterface = getRootInterface() # Get all UserMandate entries for this mandate to find users using interface method userMandates = appInterface.getUserMandatesByMandate(mandateId) total = 0 for um in userMandates: total += appInterface.revokeTokensByUser( userId=um.userId, authority=AuthAuthority(authority) if authority else None, mandateId=None, # Revoke all tokens for user revokedBy=currentUser.id, reason=reason ) return {"revoked": total} except HTTPException: raise except Exception as e: logger.error(f"Error revoking tokens by mandate: {str(e)}") raise HTTPException(status_code=500, detail="Failed to revoke mandate tokens") # ---------------------- # Logs download # ---------------------- @router.get("/logs/{log_name}") @limiter.limit("60/minute") def download_log( request: Request, currentUser: User = Depends(requireSysAdmin), log_name: str = "poweron" ): """ Download server logs. MULTI-TENANT: SysAdmin-only (infrastructure management). """ base_dir = os.path.dirname(os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) # base_dir -> gateway if log_name == "poweron": file_path = os.path.join(base_dir, "poweron.log") elif log_name == "audit": file_path = os.path.join(base_dir, "audit.log") else: raise HTTPException(status_code=400, detail="Unsupported log name") if not os.path.exists(file_path): raise HTTPException(status_code=404, detail=f"{log_name}.log not found") return FileResponse(path=file_path, filename=f"{log_name}.log") # ---------------------- # Database admin # ---------------------- @router.get("/databases") @limiter.limit("10/minute") def list_databases( request: Request, currentUser: User = Depends(requireSysAdmin) ) -> Dict[str, Any]: """ List all poweron_* databases. MULTI-TENANT: SysAdmin-only (infrastructure management). """ try: databases = _getPoweronDatabases() return {"databases": databases} except Exception as e: logger.error(f"Failed to load databases from host: {e}") raise HTTPException(status_code=500, detail="Failed to load databases from host") @router.get("/databases/{database_name}/tables") @limiter.limit("30/minute") def get_database_tables( request: Request, database_name: str, currentUser: User = Depends(requireSysAdmin) ) -> Dict[str, Any]: """ List tables in a database. MULTI-TENANT: SysAdmin-only (infrastructure management). """ if not database_name.startswith("poweron_"): raise HTTPException(status_code=400, detail="Invalid database name format") connector = None try: connector = _getDatabaseConnector(database_name, currentUser.id) tables = connector.getTables() return {"tables": tables} except ValueError as e: raise HTTPException(status_code=400, detail=str(e)) except Exception as e: logger.error(f"Error getting database tables: {str(e)}") raise HTTPException(status_code=500, detail=f"Failed to get database tables: {str(e)}") finally: if connector: connector.close() @router.post("/databases/{database_name}/tables/{table_name}/drop") @limiter.limit("10/minute") def drop_table( request: Request, database_name: str, table_name: str, currentUser: User = Depends(requireSysAdmin), payload: Dict[str, Any] = Body(...) ) -> Dict[str, Any]: """ Drop a table from a database. MULTI-TENANT: SysAdmin-only (infrastructure management). """ if not database_name.startswith("poweron_"): raise HTTPException(status_code=400, detail="Invalid database name format") connector = None try: connector = _getDatabaseConnector(database_name, currentUser.id) conn = connector.connection with conn.cursor() as cursor: # Check if table exists cursor.execute(""" SELECT table_name FROM information_schema.tables WHERE table_schema = 'public' AND table_name = %s """, (table_name,)) if not cursor.fetchone(): raise HTTPException(status_code=404, detail="Table not found") # Drop the table cursor.execute(f'DROP TABLE IF EXISTS "{table_name}" CASCADE') conn.commit() logger.warning(f"Admin drop_table executed by {currentUser.id}: dropped table '{table_name}' from database '{database_name}'") return {"message": f"Table '{table_name}' dropped successfully from database '{database_name}'"} except HTTPException: raise except ValueError as e: raise HTTPException(status_code=400, detail=str(e)) except Exception as e: logger.error(f"Error dropping table: {str(e)}") if connector and connector.connection: connector.connection.rollback() raise HTTPException(status_code=500, detail="Failed to drop table") finally: if connector: connector.close() @router.post("/databases/drop") @limiter.limit("5/minute") def drop_database( request: Request, currentUser: User = Depends(requireSysAdmin), payload: Dict[str, Any] = Body(...) ) -> Dict[str, Any]: """ Drop all tables in a database. MULTI-TENANT: SysAdmin-only (infrastructure management). """ dbName = payload.get("database") if not dbName or not dbName.startswith("poweron_"): raise HTTPException(status_code=400, detail="Invalid database name") # Validate database exists try: configuredDbs = _getPoweronDatabases() except Exception as e: logger.warning(f"Failed to load databases from host: {e}") configuredDbs = [] if configuredDbs and dbName not in configuredDbs: raise HTTPException(status_code=400, detail=f"Database not found. Available: {configuredDbs}") connector = None try: connector = _getDatabaseConnector(dbName, currentUser.id) conn = connector.connection with conn.cursor() as cursor: # Drop all user tables (public schema) cursor.execute(""" SELECT table_name FROM information_schema.tables WHERE table_schema = 'public' AND table_type = 'BASE TABLE' """) tables = [row['table_name'] for row in cursor.fetchall()] dropped = [] for tbl in tables: cursor.execute(f'DROP TABLE IF EXISTS "{tbl}" CASCADE') dropped.append(tbl) conn.commit() logger.warning(f"Admin drop_database executed by {currentUser.id}: dropped tables from '{dbName}': {dropped}") return {"droppedTables": dropped} except ValueError as e: raise HTTPException(status_code=400, detail=str(e)) except Exception as e: logger.error(f"Error dropping database tables: {str(e)}") if connector and connector.connection: connector.connection.rollback() raise HTTPException(status_code=500, detail="Failed to drop database tables") finally: if connector: connector.close()