""" Environment variable validation for thrillwiki project. This module validates environment variables on Django startup to catch configuration errors early. It checks: - Required variables are set - Values have correct types - Values are within valid ranges - URLs are properly formatted - Cross-variable dependencies are satisfied Why python-decouple? - Already used across the project for consistency - Provides type casting and default values - Supports .env files and environment variables """ import logging import re import warnings from typing import Any, Callable, Optional from urllib.parse import urlparse from decouple import config, UndefinedValueError logger = logging.getLogger("thrillwiki") # ============================================================================= # Validation Rules # ============================================================================= # Required environment variables with their validation rules REQUIRED_VARIABLES = { "SECRET_KEY": { "type": str, "min_length": 50, "description": "Django secret key for cryptographic signing", }, "DATABASE_URL": { "type": str, "validator": "url", "description": "Database connection URL", }, } # Optional variables that should be validated if present OPTIONAL_VARIABLES = { "DEBUG": { "type": bool, "default": True, "description": "Debug mode flag", }, "ALLOWED_HOSTS": { "type": str, "description": "Comma-separated list of allowed hosts", }, "REDIS_URL": { "type": str, "validator": "url", "description": "Redis connection URL", }, "EMAIL_PORT": { "type": int, "min_value": 1, "max_value": 65535, "description": "SMTP server port", }, "CACHE_MIDDLEWARE_SECONDS": { "type": int, "min_value": 0, "max_value": 86400, "description": "Cache timeout in seconds", }, "API_RATE_LIMIT_PER_MINUTE": { "type": int, "min_value": 1, "max_value": 10000, "description": "API rate limit per minute", }, "API_RATE_LIMIT_PER_HOUR": { "type": int, "min_value": 1, "max_value": 100000, "description": "API rate limit per hour", }, "SECURE_HSTS_SECONDS": { "type": int, "min_value": 0, "max_value": 31536000 * 2, # Max 2 years "description": "HSTS max-age in seconds", }, "SESSION_COOKIE_AGE": { "type": int, "min_value": 60, "max_value": 86400 * 365, # Max 1 year "description": "Session cookie age in seconds", }, "JWT_ACCESS_TOKEN_LIFETIME_MINUTES": { "type": int, "min_value": 1, "max_value": 1440, # Max 24 hours "description": "JWT access token lifetime in minutes", }, "JWT_REFRESH_TOKEN_LIFETIME_DAYS": { "type": int, "min_value": 1, "max_value": 365, "description": "JWT refresh token lifetime in days", }, "SENTRY_TRACES_SAMPLE_RATE": { "type": float, "min_value": 0.0, "max_value": 1.0, "description": "Sentry trace sampling rate", }, } # Cross-variable validation rules CROSS_VARIABLE_RULES = [ { "name": "production_security", "condition": lambda: config("DEBUG", default=True, cast=bool) is False, "requirements": [ ("SECRET_KEY", lambda v: len(v) >= 50, "must be at least 50 characters"), ("ALLOWED_HOSTS", lambda v: v and v.strip(), "must be set in production"), ], "description": "Production security requirements", }, { "name": "ssl_configuration", "condition": lambda: config("SECURE_SSL_REDIRECT", default=False, cast=bool), "requirements": [ ("SESSION_COOKIE_SECURE", lambda v: v, "should be True with SSL redirect"), ("CSRF_COOKIE_SECURE", lambda v: v, "should be True with SSL redirect"), ], "description": "SSL configuration consistency", }, ] # ============================================================================= # Validation Functions # ============================================================================= def validate_url(value: str) -> bool: """Validate that a value is a valid URL.""" try: result = urlparse(value) return all([result.scheme, result.netloc]) except Exception: return False def validate_email(value: str) -> bool: """Validate that a value is a valid email address.""" email_pattern = r"^[a-zA-Z0-9._%+-]+@[a-zA-Z0-9.-]+\.[a-zA-Z]{2,}$" return bool(re.match(email_pattern, value)) def validate_type(value: Any, expected_type: type) -> bool: """Validate that a value is of the expected type.""" if expected_type == bool: # Special handling for boolean strings return isinstance(value, bool) or str(value).lower() in ( "true", "false", "1", "0", "yes", "no" ) return isinstance(value, expected_type) def validate_range( value: Any, min_value: Optional[Any] = None, max_value: Optional[Any] = None ) -> bool: """Validate that a value is within a specified range.""" if min_value is not None and value < min_value: return False if max_value is not None and value > max_value: return False return True def validate_length(value: str, min_length: int = 0, max_length: int = None) -> bool: """Validate that a string value meets length requirements.""" if len(value) < min_length: return False if max_length is not None and len(value) > max_length: return False return True VALIDATORS = { "url": validate_url, "email": validate_email, } # ============================================================================= # Main Validation Functions # ============================================================================= def validate_variable(name: str, rules: dict) -> list[str]: """ Validate a single environment variable against its rules. Args: name: Environment variable name rules: Validation rules dictionary Returns: List of error messages (empty if valid) """ errors = [] try: # Get the value with appropriate type casting var_type = rules.get("type", str) default = rules.get("default", None) if var_type == bool: value = config(name, default=default, cast=bool) elif var_type == int: value = config(name, default=default, cast=int) elif var_type == float: value = config(name, default=default, cast=float) else: value = config(name, default=default) except UndefinedValueError: errors.append(f"{name}: Required variable is not set") return errors except ValueError as e: errors.append(f"{name}: Invalid value - {e}") return errors # Type validation if not validate_type(value, rules.get("type", str)): errors.append( f"{name}: Expected type {rules['type'].__name__}, " f"got {type(value).__name__}" ) # Length validation (for strings) if isinstance(value, str): min_length = rules.get("min_length", 0) max_length = rules.get("max_length") if not validate_length(value, min_length, max_length): errors.append( f"{name}: Length must be between {min_length} and " f"{max_length or 'unlimited'}" ) # Range validation (for numbers) if isinstance(value, (int, float)): min_value = rules.get("min_value") max_value = rules.get("max_value") if not validate_range(value, min_value, max_value): errors.append( f"{name}: Value must be between {min_value} and {max_value}" ) # Custom validator validator_name = rules.get("validator") if validator_name and validator_name in VALIDATORS: if not VALIDATORS[validator_name](value): errors.append(f"{name}: Failed {validator_name} validation") return errors def validate_cross_rules() -> list[str]: """ Validate cross-variable dependencies. Returns: List of error/warning messages """ errors = [] for rule in CROSS_VARIABLE_RULES: try: # Check if the condition applies if not rule["condition"](): continue # Check each requirement for var_name, check_fn, message in rule["requirements"]: try: value = config(var_name, default=None) if value is not None and not check_fn(value): errors.append( f"{rule['name']}: {var_name} {message}" ) except Exception: errors.append( f"{rule['name']}: Could not validate {var_name}" ) except Exception as e: errors.append(f"Cross-validation error for {rule['name']}: {e}") return errors def validate_all_settings(raise_on_error: bool = False) -> dict: """ Validate all environment variables. Args: raise_on_error: If True, raise ValueError on first error Returns: Dictionary with 'errors' and 'warnings' lists """ result = { "errors": [], "warnings": [], "valid": True, } # Validate required variables for name, rules in REQUIRED_VARIABLES.items(): errors = validate_variable(name, rules) result["errors"].extend(errors) # Validate optional variables (if set) for name, rules in OPTIONAL_VARIABLES.items(): try: # Only validate if the variable is set config(name) errors = validate_variable(name, rules) result["warnings"].extend(errors) # Warnings for optional vars except UndefinedValueError: pass # Optional variable not set, that's fine # Validate cross-variable rules cross_errors = validate_cross_rules() result["warnings"].extend(cross_errors) # Set validity result["valid"] = len(result["errors"]) == 0 # Handle errors if result["errors"]: for error in result["errors"]: logger.error(f"Configuration error: {error}") if raise_on_error: raise ValueError( f"Configuration validation failed: {result['errors']}" ) # Log warnings for warning in result["warnings"]: logger.warning(f"Configuration warning: {warning}") return result def run_startup_validation() -> None: """ Run configuration validation on application startup. This function should be called during Django initialization to catch configuration errors early. """ debug_mode = config("DEBUG", default=True, cast=bool) result = validate_all_settings(raise_on_error=not debug_mode) if result["valid"]: logger.info("Configuration validation passed") else: if debug_mode: for error in result["errors"]: warnings.warn(f"Configuration error: {error}") else: raise ValueError( "Configuration validation failed. Check logs for details." ) # ============================================================================= # Django Management Command Support # ============================================================================= def get_validation_report() -> str: """ Generate a detailed validation report. Returns: Formatted string report """ result = validate_all_settings(raise_on_error=False) lines = ["=" * 60] lines.append("Configuration Validation Report") lines.append("=" * 60) lines.append("") if result["valid"]: lines.append("Status: PASSED") else: lines.append("Status: FAILED") lines.append("") lines.append(f"Errors: {len(result['errors'])}") lines.append(f"Warnings: {len(result['warnings'])}") lines.append("") if result["errors"]: lines.append("-" * 40) lines.append("Errors:") for error in result["errors"]: lines.append(f" - {error}") lines.append("") if result["warnings"]: lines.append("-" * 40) lines.append("Warnings:") for warning in result["warnings"]: lines.append(f" - {warning}") lines.append("") lines.append("=" * 60) return "\n".join(lines)