import os import logging import yaml log = logging.getLogger("ainventory") _config = {} class ConfigError(Exception): """Raised when configuration is invalid or missing.""" pass def _deep_merge(base, source): """Recursively merge dictionaries.""" for key, value in source.items(): if isinstance(value, dict) and key in base and isinstance(base[key], dict): _deep_merge(base[key], value) else: base[key] = value def _map_secrets(config, secrets): """Map secrets from secrets.yaml (flattened) to the nested config structure.""" mapping = { "JWT_SECRET_KEY": ("auth", "jwt_secret_key"), "GEMINI_API_KEY": ("ai", "gemini_api_key"), "CLAUDE_API_KEY": ("ai", "claude_api_key"), "DATABASE_PASSWORD": ("database", "password"), "LDAP_PASSWORD": ("auth", "ldap_password") } for secret_key, config_path in mapping.items(): if secret_key in secrets: target = config for p in config_path[:-1]: if p not in target: target[p] = {} target = target[p] target[config_path[-1]] = secrets[secret_key] def _to_bool(val): """Convert string to boolean.""" if isinstance(val, bool): return val return str(val).lower() in ("true", "1", "yes", "on") def _apply_env_overrides(config): """Apply environment variable overrides (D-06 load order).""" # Mapping of environment variables to config paths # Format: ENV_VAR: (path, type_converter) env_mapping = { "BACKEND_DATABASE_SQLITE_PATH": (("database", "sqlite_path"), str), "BACKEND_DATABASE_WAL_MODE": (("database", "wal_mode"), _to_bool), "BACKEND_DATABASE_LOG_RETENTION_DAYS": (("database", "log_retention_days"), int), "BACKEND_AI_PRIMARY_AI_PROVIDER": (("ai", "primary_ai_provider"), str), "PRIMARY_AI_PROVIDER": (("ai", "primary_ai_provider"), str), "BACKEND_AI_FALLBACK_PROVIDER": (("ai", "fallback_provider"), str), "BACKEND_AI_GEMINI_API_KEY": (("ai", "gemini_api_key"), str), "GEMINI_API_KEY": (("ai", "gemini_api_key"), str), "BACKEND_AI_CLAUDE_API_KEY": (("ai", "claude_api_key"), str), "CLAUDE_API_KEY": (("ai", "claude_api_key"), str), "BACKEND_AUTH_JWT_SECRET_KEY": (("auth", "jwt_secret_key"), str), "JWT_SECRET_KEY": (("auth", "jwt_secret_key"), str), "BACKEND_AUTH_LDAP_ENABLED": (("auth", "ldap_enabled"), _to_bool), "LDAP_ENABLED": (("auth", "ldap_enabled"), _to_bool), "BACKEND_AUTH_LDAP_SERVER": (("auth", "ldap_server"), str), "LDAP_SERVER": (("auth", "ldap_server"), str), "BACKEND_AUTH_LDAP_BASE_DN": (("auth", "ldap_base_dn"), str), "LDAP_BASE_DN": (("auth", "ldap_base_dn"), str), "BACKEND_AUTH_LDAP_USER_TEMPLATE": (("auth", "ldap_user_template"), str), "LDAP_USER_TEMPLATE": (("auth", "ldap_user_template"), str), "BACKEND_AUTH_LDAP_GROUPS_DN": (("auth", "ldap_groups_dn"), str), "LDAP_GROUPS_DN": (("auth", "ldap_groups_dn"), str), "BACKEND_AUTH_LDAP_USE_TLS": (("auth", "ldap_use_tls"), _to_bool), "LDAP_USE_TLS": (("auth", "ldap_use_tls"), _to_bool), "BACKEND_AUTH_LDAP_IGNORE_CERT": (("auth", "ldap_ignore_cert"), _to_bool), "LDAP_IGNORE_CERT": (("auth", "ldap_ignore_cert"), _to_bool), "BACKEND_AUTH_PASSWORD_CACHE_PATH": (("auth", "password_cache_path"), str), "BACKEND_LOGGING_LOG_LEVEL": (("logging", "log_level"), str), "LOG_LEVEL": (("logging", "log_level"), str), "BACKEND_LOGGING_LOG_ROTATION_SIZE_MB": (("logging", "log_rotation_size_mb"), int), "BACKEND_LOGGING_LOG_ROTATION_COUNT": (("logging", "log_rotation_count"), int), "BACKEND_APPLICATION_DATA_DIR": (("application", "data_dir"), str), "DATA_DIR": (("application", "data_dir"), str), "BACKEND_APPLICATION_LOGS_DIR": (("application", "logs_dir"), str), "LOGS_DIR": (("application", "logs_dir"), str), "BACKEND_APPLICATION_CORS_ORIGINS": (("application", "cors_origins"), str), "EXTRA_ALLOWED_ORIGINS": (("application", "cors_origins"), str), "ALLOWED_ORIGINS": (("application", "cors_origins"), str), "SERVER_IP": (("application", "server_ip"), str), "FRONTEND_PORT": (("application", "frontend_port"), str), "FRONTEND_SSL_PORT": (("application", "frontend_ssl_port"), str), "BACKEND_PORT": (("application", "backend_port"), str), "BACKEND_SSL_PORT": (("application", "backend_ssl_port"), str), } sensitive_vars = [ "JWT_SECRET_KEY", "GEMINI_API_KEY", "CLAUDE_API_KEY", "BACKEND_AUTH_JWT_SECRET_KEY", "BACKEND_AI_GEMINI_API_KEY", "BACKEND_AI_CLAUDE_API_KEY", "LDAP_PASSWORD", "DATABASE_PASSWORD" ] for env_var, (path, converter) in env_mapping.items(): val = os.getenv(env_var) if val is not None: try: converted_val = converter(val) target = config for p in path[:-1]: if p not in target: target[p] = {} target = target[p] target[path[-1]] = converted_val # Mask sensitive values in logs log_val = "********" if env_var in sensitive_vars else converted_val log.info(f"ℹ️ Override {'.'.join(path)} from environment ({env_var}): {log_val}") except Exception as e: log.warning(f"⚠️ Failed to convert env var {env_var}='{val}': {e}") def load_config() -> dict: """Load config from YAML files with env var overrides (D-06 load order).""" global _config # Base directory is backend/ base_dir = os.path.dirname(os.path.abspath(__file__)) # Project root is one level up project_root = os.path.dirname(base_dir) config_dir = os.path.join(project_root, "config") # 1. Define Defaults config = { "database": { "sqlite_path": "data/inventory.db", "wal_mode": True, "log_retention_days": 30 }, "ai": { "primary_ai_provider": "gemini", "fallback_provider": "claude", "gemini_api_key": "", "claude_api_key": "" }, "auth": { "jwt_secret_key": "change_me_in_production", "ldap_enabled": False, "ldap_server": "", "ldap_base_dn": "", "ldap_user_template": "uid={username},ou=people,dc=ldap,dc=lan", "ldap_groups_dn": "ou=groups", "ldap_use_tls": True, "ldap_ignore_cert": False, "ldap_role_mappings": [ {"group": "inventory_admins", "role": "admin"}, {"group": "inventory_users", "role": "user"} ], "password_cache_path": "data/.passwords" }, "logging": { "log_level": "INFO", "log_rotation_size_mb": 10, "log_rotation_count": 5 }, "application": { "data_dir": "./data", "logs_dir": "./logs", "cors_origins": "http://localhost:8917", "server_ip": "localhost", "frontend_port": "8917", "frontend_ssl_port": "8919", "backend_port": "8918", "backend_ssl_port": "8918" }, "features": { "ai_extraction_enabled": True, "offline_sync_enabled": True, "audit_logging_enabled": True } } # 2. Load backend.yaml backend_yaml_path = os.path.join(config_dir, "backend.yaml") if os.path.exists(backend_yaml_path): try: with open(backend_yaml_path, 'r') as f: yaml_data = yaml.safe_load(f) or {} _deep_merge(config, yaml_data) log.info(f"✅ Loaded configuration from {backend_yaml_path}") except Exception as e: log.warning(f"⚠️ Failed to load {backend_yaml_path}: {e}") else: log.warning(f"ℹ️ {backend_yaml_path} not found. Using defaults.") # 3. Load secrets.yaml secrets_yaml_path = os.path.join(config_dir, "secrets.yaml") if os.path.exists(secrets_yaml_path): try: with open(secrets_yaml_path, 'r') as f: secrets_data = yaml.safe_load(f) or {} _map_secrets(config, secrets_data) log.info(f"✅ Loaded secrets from {secrets_yaml_path}") except Exception as e: log.warning(f"⚠️ Failed to load {secrets_yaml_path}: {e}") # 4. Load network.yaml (SSOT for infrastructure) network_yaml_path = os.path.join(config_dir, "network.yaml") if os.path.exists(network_yaml_path): try: with open(network_yaml_path, 'r') as f: network_data = yaml.safe_load(f) or {} # Map infrastructure settings to application domain infra = network_data.get("application", {}) if "server_ip" in infra: config["application"]["server_ip"] = infra["server_ip"] if "cors_origins" in infra: # Append or override? Plan says establish as master. # We merge with any existing ones. existing = config["application"].get("cors_origins", "") if existing: config["application"]["cors_origins"] = f"{infra['cors_origins']},{existing}" else: config["application"]["cors_origins"] = infra["cors_origins"] # Map ports to application domain for CORS logic ports = network_data.get("ports", {}) for key, val in ports.items(): config["application"][key] = val # Map SSL state config["application"]["ssl_enabled"] = network_data.get("ssl", {}).get("ssl_enabled", False) log.info(f"✅ Loaded network topology from {network_yaml_path}") except Exception as e: log.warning(f"⚠️ Failed to load {network_yaml_path}: {e}") # 5. Apply Environment Overrides (D-06) _apply_env_overrides(config) _config = config return _config def get_config() -> dict: """Get loaded config.""" if not _config: load_config() return _config def validate_config(config: dict) -> bool: """Validate config has all required values.""" required = [ ("auth", "jwt_secret_key"), ("ai", "primary_ai_provider"), ] missing = [] for path in required: val = config for p in path: val = val.get(p) if not val or val == "change_me_in_production" or val == "CHANGE_ME_IN_PRODUCTION_MIN_32_CHARS": missing.append(".".join(path)) if missing: raise ConfigError(f"Missing or default required configuration: {', '.join(missing)}") # Validate enums valid_providers = ["gemini", "claude"] if config["ai"]["primary_ai_provider"] not in valid_providers: raise ConfigError(f"Invalid primary_ai_provider: {config['ai']['primary_ai_provider']}. Must be one of {valid_providers}") valid_log_levels = ["DEBUG", "INFO", "WARNING", "ERROR"] if config["logging"]["log_level"].upper() not in valid_log_levels: raise ConfigError(f"Invalid log_level: {config['logging']['log_level']}. Must be one of {valid_log_levels}") log.info(f"✅ Config validated: primary_ai_provider={config['ai']['primary_ai_provider']}, log_level={config['logging']['log_level']}") return True # Auto-run if imported try: load_config() validate_config(_config) except ConfigError as ce: log.error(f"❌ Configuration error: {ce}") except Exception as e: log.error(f"❌ Unexpected error during config load: {e}")