59 lines
1.9 KiB
Python
59 lines
1.9 KiB
Python
# Copyright (c) 2025 Patrick Motsch
|
|
# All rights reserved.
|
|
"""
|
|
Root access management for system-level operations.
|
|
Provides secure access to root user and DbApp database connector.
|
|
"""
|
|
|
|
import logging
|
|
from modules.connectors.connectorDbPostgre import DatabaseConnector
|
|
from modules.datamodels.datamodelUam import User, UserInDB
|
|
from modules.shared.configuration import APP_CONFIG
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
_rootDbAppConnector = None
|
|
_rootUser = None
|
|
|
|
def getRootDbAppConnector() -> DatabaseConnector:
|
|
"""
|
|
Returns a DatabaseConnector instance for the DbApp database.
|
|
This is used for accessing system tables like AccessRule.
|
|
"""
|
|
global _rootDbAppConnector
|
|
|
|
if _rootDbAppConnector is None:
|
|
_rootDbAppConnector = DatabaseConnector(
|
|
dbHost=APP_CONFIG.get("DB_APP_HOST"),
|
|
dbDatabase=APP_CONFIG.get("DB_APP_DATABASE", "app"),
|
|
dbUser=APP_CONFIG.get("DB_APP_USER"),
|
|
dbPassword=APP_CONFIG.get("DB_APP_PASSWORD_SECRET"),
|
|
dbPort=int(APP_CONFIG.get("DB_APP_PORT", 5432)),
|
|
userId=None # No user context for root connector
|
|
)
|
|
_rootDbAppConnector.initDbSystem()
|
|
|
|
return _rootDbAppConnector
|
|
|
|
def getRootUser() -> User:
|
|
"""
|
|
Returns the root user (initial user from database).
|
|
Used for system-level operations that require root privileges.
|
|
"""
|
|
global _rootUser
|
|
|
|
if _rootUser is None:
|
|
dbApp = getRootDbAppConnector()
|
|
initialUserId = dbApp.getInitialId(UserInDB)
|
|
if not initialUserId:
|
|
raise ValueError("No initial user ID found in database")
|
|
|
|
users = dbApp.getRecordset(UserInDB, recordFilter={"id": initialUserId})
|
|
if not users:
|
|
raise ValueError("Initial user not found in database")
|
|
|
|
user_data = users[0]
|
|
_rootUser = User(**user_data)
|
|
|
|
return _rootUser
|
|
|