68 lines
2.1 KiB
Python
68 lines
2.1 KiB
Python
# Copyright (c) 2026 PowerOn AG
|
|
# All rights reserved.
|
|
"""
|
|
Root access management for system-level operations.
|
|
Provides secure access to root user and DbApp database connector.
|
|
|
|
Bootstrap is guaranteed by app.py lifespan before any access.
|
|
"""
|
|
|
|
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_HOST"),
|
|
dbDatabase="poweron_app",
|
|
dbUser=APP_CONFIG.get("DB_USER"),
|
|
dbPassword=APP_CONFIG.get("DB_PASSWORD_SECRET"),
|
|
dbPort=int(APP_CONFIG.get("DB_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.
|
|
|
|
Raises RuntimeError if no user exists (bootstrap incomplete).
|
|
"""
|
|
global _rootUser
|
|
|
|
if _rootUser is None:
|
|
dbApp = getRootDbAppConnector()
|
|
initialUserId = dbApp.getInitialId(UserInDB)
|
|
|
|
if not initialUserId:
|
|
raise RuntimeError(
|
|
"No root user found - bootstrap incomplete. "
|
|
"Ensure app.py lifespan runs initBootstrap before any service access."
|
|
)
|
|
|
|
users = dbApp.getRecordset(UserInDB, recordFilter={"id": initialUserId})
|
|
if not users:
|
|
raise RuntimeError("Initial user not found in database")
|
|
|
|
user_data = users[0]
|
|
_rootUser = User(**user_data)
|
|
|
|
return _rootUser
|