""" Utility module for configuration management. This module provides a global APP_CONFIG object for accessing configuration from both config.ini files and environment variables stored in .env files, using a flat structure. """ import os import logging import json import base64 import time from typing import Any, Dict, Optional from pathlib import Path from cryptography.fernet import Fernet from cryptography.hazmat.primitives import hashes from cryptography.hazmat.primitives.kdf.pbkdf2 import PBKDF2HMAC # audit_logger imported lazily to avoid circular import # Set up basic logging for configuration loading logging.basicConfig( level=logging.WARNING, format='%(asctime)s - %(levelname)s - %(name)s - %(message)s', handlers=[logging.StreamHandler()] ) # Configure logger logger = logging.getLogger(__name__) class Configuration: """ Configuration class with attribute-style access to flattened configuration. """ def __init__(self): """Initialize the configuration object""" self._data = {} self._configFilePath = None self._envFilePath = None self._configMtime = 0 self._envMtime = 0 self.refresh() def refresh(self): """Reload configuration from files""" self._loadConfig() self._loadEnv() logger.info("Configuration refreshed") def _loadConfig(self): """Load configuration from config.ini file in flattened format""" # Find config.ini file in the gateway directory configPath = Path(__file__).parent.parent.parent / 'config.ini' if not configPath.exists(): logger.warning(f"Configuration file not found at {configPath.absolute()}") return self._configFilePath = configPath currentMtime = os.path.getmtime(configPath) # Skip if file hasn't changed if currentMtime <= self._configMtime: return self._configMtime = currentMtime try: with open(configPath, 'r') as f: lines = f.readlines() i = 0 while i < len(lines): line = lines[i].strip() # Skip empty lines and comments if not line or line.startswith('#'): i += 1 continue # Parse key-value pairs if '=' in line: key, value = line.split('=', 1) key = key.strip() value = value.strip() # Check if value starts with { (JSON object) if value.startswith('{'): # Collect all lines until we find the closing } json_lines = [value] i += 1 brace_count = value.count('{') - value.count('}') while i < len(lines) and brace_count > 0: json_lines.append(lines[i].rstrip('\n')) brace_count += lines[i].count('{') - lines[i].count('}') i += 1 # Join all lines and parse as JSON value = '\n'.join(json_lines) i -= 1 # Adjust for the loop increment # Add to data dictionary self._data[key] = value i += 1 except Exception as e: logger.error(f"Error loading configuration: {e}") def _loadEnv(self): """Load environment variables from .env file""" # Find .env file in the gateway directory envPath = Path(__file__).parent.parent.parent / '.env' if not envPath.exists(): logger.warning(f"Environment file not found at {envPath.absolute()}") return self._envFilePath = envPath currentMtime = os.path.getmtime(envPath) # Skip if file hasn't changed if currentMtime <= self._envMtime: return self._envMtime = currentMtime try: with open(envPath, 'r') as f: lines = f.readlines() i = 0 while i < len(lines): line = lines[i].strip() # Skip empty lines and comments if not line or line.startswith('#'): i += 1 continue # Parse key-value pairs if '=' in line: key, value = line.split('=', 1) key = key.strip() value = value.strip() # Check if value starts with { (JSON object) if value.startswith('{'): # Collect all lines until we find the closing } json_lines = [value] i += 1 brace_count = value.count('{') - value.count('}') while i < len(lines) and brace_count > 0: json_lines.append(lines[i].rstrip('\n')) brace_count += lines[i].count('{') - lines[i].count('}') i += 1 # Join all lines and create the full JSON value full_json_value = '\n'.join(json_lines) self._data[key] = full_json_value else: # Single line value self._data[key] = value i += 1 logger.info(f"Loaded environment variables from {envPath.absolute()}") # Also load system environment variables (don't override existing) for key, value in os.environ.items(): if key not in self._data: self._data[key] = value except Exception as e: logger.error(f"Error loading environment variables: {e}") def checkForUpdates(self): """Check if configuration files have changed and reload if necessary""" if self._configFilePath and os.path.exists(self._configFilePath): currentMtime = os.path.getmtime(self._configFilePath) if currentMtime > self._configMtime: logger.info("Config file has changed, reloading...") self._loadConfig() if self._envFilePath and os.path.exists(self._envFilePath): currentMtime = os.path.getmtime(self._envFilePath) if currentMtime > self._envMtime: logger.info("Environment file has changed, reloading...") self._loadEnv() def get(self, key: str, default: Any = None, user_id: str = "system") -> Any: """Get configuration value with optional default""" self.checkForUpdates() # Check for file changes if key in self._data: value = self._data[key] # Handle secrets (keys ending with _SECRET) if key.endswith("_SECRET"): # Log audit event for secret key access try: from modules.shared.auditLogger import audit_logger audit_logger.logKeyAccess( userId=user_id, mandateId="system", keyName=key, action="decode" ) except Exception: # Don't fail if audit logging fails pass if value.startswith("{") and value.endswith("}"): # Handle JSON secrets (keys ending with _API_KEY that contain JSON) return handleSecretJson(value, userId=user_id, keyName=key) else: return handleSecretText(value, userId=user_id, keyName=key) return value return default def __getattr__(self, name: str) -> Any: """Enable attribute-style access to configuration""" self.checkForUpdates() # Check for file changes value = self.get(name, user_id="system") if value is None: raise AttributeError(f"Configuration key '{name}' not found") return value def __dir__(self) -> list: """Support auto-completion of attributes""" self.checkForUpdates() # Check for file changes return list(self._data.keys()) + super().__dir__() def set(self, key: str, value: Any) -> None: """Set a configuration value (for testing/overrides)""" self._data[key] = value def handleSecretText(value: str, userId: str = "system", keyName: str = "unknown") -> str: """ Handle secret values with encryption/decryption support. Args: value: The secret value to handle (may be encrypted) userId: The user ID making the request (default: "system") keyName: The name of the key being decrypted (default: "unknown") Returns: str: Processed secret value (decrypted if encrypted) """ if _isEncryptedValue(value): return decryptValue(value, userId, keyName) return value def handleSecretJson(value: str, userId: str = "system", keyName: str = "unknown") -> str: """ Handle JSON secret values (like Google service account keys) with encryption/decryption support. Validates that the value is valid JSON after decryption. Args: value: The JSON secret value to handle (may be encrypted) userId: The user ID making the request (default: "system") keyName: The name of the key being decrypted (default: "unknown") Returns: str: Processed JSON secret value (decrypted if encrypted) Raises: ValueError: If the value is not valid JSON after decryption """ # Decrypt if encrypted if _isEncryptedValue(value): decryptedValue = decryptValue(value, userId, keyName) else: decryptedValue = value try: # Validate that it's valid JSON json.loads(decryptedValue) return decryptedValue except json.JSONDecodeError as e: raise ValueError(f"Invalid JSON in secret value: {e}") # Global rate limiting tracking # Structure: {user_id: {key_name: [timestamps]}} _decryption_attempts = {} def _getMasterKey(envType: str = None) -> bytes: """ Get the master key for the specified environment. Args: envType: The environment type (dev, int, prod, etc.). If None, uses current config. Returns: bytes: The master key for encryption/decryption Raises: ValueError: If no master key is found """ # Get the key location from config keyLocation = APP_CONFIG.get('APP_KEY_SYSVAR') if envType is None: envType = APP_CONFIG.get('APP_ENV_TYPE', 'dev') if not keyLocation: raise ValueError("APP_KEY_SYSVAR not configured") # First try to get from environment variable masterKey = os.environ.get(keyLocation) if masterKey: # If found in environment, use it directly return masterKey.encode('utf-8') # If not in environment, try to read from file if os.path.exists(keyLocation): try: with open(keyLocation, 'r') as f: content = f.read().strip() # Parse the key file format: env = key lines = content.split('\n') for line in lines: line = line.strip() if not line or line.startswith('#'): continue if '=' in line: keyEnv, keyValue = line.split('=', 1) keyEnv = keyEnv.strip() keyValue = keyValue.strip() if keyEnv == envType: return keyValue.encode('utf-8') raise ValueError(f"No key found for environment '{envType}' in {keyLocation}") except Exception as e: raise ValueError(f"Error reading key file {keyLocation}: {e}") raise ValueError(f"Master key not found. Checked environment variable '{keyLocation}' and file path") def _deriveEncryptionKey(masterKey: bytes) -> bytes: """ Derive a 32-byte encryption key from the master key using PBKDF2. Args: masterKey: The master key bytes Returns: bytes: 32-byte derived key suitable for Fernet """ # Use a fixed salt for consistency (in production, consider using a random salt stored separately) salt = b'poweron_config_salt_2025' kdf = PBKDF2HMAC( algorithm=hashes.SHA256(), length=32, salt=salt, iterations=100000, ) return base64.urlsafe_b64encode(kdf.derive(masterKey)) def _isEncryptedValue(value: str) -> bool: """ Check if a value is encrypted (starts with environment-specific prefix). Args: value: The value to check Returns: bool: True if encrypted, False otherwise """ if not value or not isinstance(value, str): return False # Check for any environment-specific encryption prefixes return (value.startswith('DEV_ENC:') or value.startswith('INT_ENC:') or value.startswith('PROD_ENC:') or value.startswith('TEST_ENC:') or value.startswith('STAGING_ENC:')) def _getEncryptionPrefix(envType: str) -> str: """ Get the encryption prefix for the given environment type. Args: envType: The environment type (dev, int, prod, etc.) Returns: str: The encryption prefix """ return f"{envType.upper()}_ENC:" def _checkDecryptionRateLimit(userId: str, keyName: str, maxPerSecond: int = 10) -> bool: """ Check if decryption is allowed based on rate limiting (max 10 per second per user per key). Args: userId: The user ID making the request keyName: The name of the key being decrypted maxPerSecond: Maximum decryptions per second (default: 10) Returns: bool: True if allowed, False if rate limited """ currentTime = time.time() # Initialize tracking for this user if not exists if userId not in _decryption_attempts: _decryption_attempts[userId] = {} # Initialize tracking for this key if not exists if keyName not in _decryption_attempts[userId]: _decryption_attempts[userId][keyName] = [] # Clean old attempts (older than 1 second) _decryption_attempts[userId][keyName] = [ timestamp for timestamp in _decryption_attempts[userId][keyName] if currentTime - timestamp < 1.0 ] # Check if we're within rate limit if len(_decryption_attempts[userId][keyName]) >= maxPerSecond: logger.warning(f"Decryption rate limit exceeded for user '{userId}' key '{keyName}' ({maxPerSecond}/sec)") return False # Record this attempt _decryption_attempts[userId][keyName].append(currentTime) return True def encryptValue(value: str, envType: str = None, userId: str = "system", keyName: str = "unknown") -> str: """ Encrypt a value using the master key for the specified environment. Args: value: The plain text value to encrypt envType: The environment type (dev, int, prod). If None, uses current environment. userId: The user ID making the request (default: "system") keyName: The name of the key being encrypted (default: "unknown") Returns: str: The encrypted value with prefix Raises: ValueError: If encryption fails """ if envType is None: envType = APP_CONFIG.get('APP_ENV_TYPE', 'dev') try: masterKey = _getMasterKey(envType) derivedKey = _deriveEncryptionKey(masterKey) fernet = Fernet(derivedKey) # Encrypt the value encryptedBytes = fernet.encrypt(value.encode('utf-8')) encryptedB64 = base64.urlsafe_b64encode(encryptedBytes).decode('utf-8') # Add environment prefix prefix = _getEncryptionPrefix(envType) encryptedValue = f"{prefix}{encryptedB64}" # Log audit event for encryption try: from modules.shared.auditLogger import audit_logger audit_logger.logKeyAccess( userId=userId, mandateId="system", keyName=keyName, action="encrypt" ) except Exception: # Don't fail if audit logging fails pass return encryptedValue except Exception as e: raise ValueError(f"Encryption failed: {e}") def decryptValue(encryptedValue: str, userId: str = "system", keyName: str = "unknown") -> str: """ Decrypt a value using the master key for the current environment. Args: encryptedValue: The encrypted value with prefix userId: The user ID making the request (default: "system") keyName: The name of the key being decrypted (default: "unknown") Returns: str: The decrypted plain text value Raises: ValueError: If decryption fails """ if not _isEncryptedValue(encryptedValue): return encryptedValue # Return as-is if not encrypted # Check rate limiting (10 per second per user per key) if not _checkDecryptionRateLimit(userId, keyName, maxPerSecond=10): raise ValueError(f"Decryption rate limit exceeded for user '{userId}' key '{keyName}' (10/sec)") try: # Extract environment type from prefix if encryptedValue.startswith('DEV_ENC:'): envType = 'dev' prefix = 'DEV_ENC:' elif encryptedValue.startswith('INT_ENC:'): envType = 'int' prefix = 'INT_ENC:' elif encryptedValue.startswith('PROD_ENC:'): envType = 'prod' prefix = 'PROD_ENC:' elif encryptedValue.startswith('TEST_ENC:'): envType = 'test' prefix = 'TEST_ENC:' elif encryptedValue.startswith('STAGING_ENC:'): envType = 'staging' prefix = 'STAGING_ENC:' else: raise ValueError(f"Invalid encryption prefix. Expected DEV_ENC:, INT_ENC:, PROD_ENC:, TEST_ENC:, or STAGING_ENC:") encryptedPart = encryptedValue[len(prefix):] # Get master key for the specific environment and derive encryption key masterKey = _getMasterKey(envType) derivedKey = _deriveEncryptionKey(masterKey) fernet = Fernet(derivedKey) # Decode and decrypt encryptedBytes = base64.urlsafe_b64decode(encryptedPart.encode('utf-8')) decryptedBytes = fernet.decrypt(encryptedBytes) decryptedValue = decryptedBytes.decode('utf-8') # Log audit event for decryption try: from modules.shared.auditLogger import audit_logger audit_logger.logKeyAccess( userId=userId, mandateId="system", keyName=keyName, action="decrypt" ) except Exception: # Don't fail if audit logging fails pass return decryptedValue except Exception as e: raise ValueError(f"Decryption failed: {e}") # Create the global APP_CONFIG instance APP_CONFIG = Configuration()