""" Authentication utilities for FLARES application. Provides centralized authentication handling for: - Update API endpoints (S003 fix) - Plugin repository access - Remote configuration loading Security Principles: - Credentials are never logged or printed - Tokens are stored as environment variables when possible - Authentication is optional but recommended for production use Usage: from src.shared.auth_utils import AuthManager # Initialize with token from environment auth = AuthManager() # Add authentication headers to requests headers = auth.get_headers() response = requests.get(url, headers=headers) """ import os import base64 import hashlib from typing import Dict, Optional, Tuple from datetime import datetime, timedelta import logging class AuthManager: """ Centralized authentication manager for FLARES application. Handles: - API token management (stored in environment variables) - Header generation for authenticated requests - Token expiry tracking Environment Variables: FLARES_API_TOKEN: Primary API authentication token FLARES_PLUGIN_TOKEN: Optional plugin repository token FLARES_UPDATE_ENDPOINT_URL: Override default update URL Attributes: api_token (Optional[str]): API token from environment plugin_token (Optional[str]): Plugin repository token update_endpoint_url (str): Default update endpoint """ # Default endpoints DEFAULT_UPDATE_URL = "https://api.flares.research.dezeeuw.ca/v1/updates" DEFAULT_PLUGIN_REGISTRY_URL = "https://plugins.flares.research.dezeeuw.ca/api" def __init__(self, api_token: Optional[str] = None): """ Initialize authentication manager. Args: api_token: Optional API token (falls back to FLARES_API_TOKEN env var) Note: Token is not stored as instance attribute for security. Always read from env. """ self._api_token: Optional[str] = None self._plugin_token: Optional[str] = None # Load from provided token or environment variable if api_token is not None: # Provided programmatically (for testing/CI) self._api_token = api_token elif "FLARES_API_TOKEN" in os.environ: # Load from environment (recommended for production) self._api_token = os.environ.get("FLARES_API_TOKEN") # Plugin token is separate, only loaded if explicitly set if "FLARES_PLUGIN_TOKEN" in os.environ: self._plugin_token = os.environ.get("FLARES_PLUGIN_TOKEN") # Override URLs from environment if configured self.update_endpoint_url = os.environ.get( "FLARES_UPDATE_ENDPOINT_URL", self.DEFAULT_UPDATE_URL ) self.plugin_registry_url = os.environ.get( "FLARES_PLUGIN_REGISTRY_URL", self.DEFAULT_PLUGIN_REGISTRY_URL ) self._logger = logging.getLogger(__name__) @property def has_api_auth(self) -> bool: """Check if API authentication is configured.""" return self._api_token is not None and len(str(self._api_token)) > 0 @property def has_plugin_auth(self) -> bool: """Check if plugin repository authentication is configured.""" return self._plugin_token is not None and len(str(self._plugin_token)) > 0 def get_headers( self, endpoint_type: str = "update", extra_headers: Optional[Dict[str, str]] = None ) -> Dict[str, str]: """ Generate authentication headers for requests. Args: endpoint_type: Type of endpoint ("update" or "plugin") extra_headers: Additional headers to include Returns: Dictionary of headers including authentication if configured Security Note: Headers never contain raw token in logging - use hashed representation only. """ headers = { "User-Agent": "FLARES-Client/1.8.1", "Accept": "application/json", } # Add authentication based on endpoint type if endpoint_type == "update" and self.has_api_auth: headers["Authorization"] = f"Bearer {self._api_token}" self._logger.debug("Adding Bearer token for update endpoint (length=%d)", len(self._api_token)) elif endpoint_type == "plugin" and self.has_plugin_auth: headers["Authorization"] = f"Bearer {self._plugin_token}" self._logger.debug("Adding Bearer token for plugin endpoint") # Merge extra headers (allow override for specific use cases) if extra_headers: headers.update(extra_headers) return headers def create_authenticated_request_config( self, method: str = "GET", auth_type: str = "update" ) -> Dict[str, any]: # type: ignore """ Create a complete request configuration dictionary. Args: method: HTTP method ("GET", "POST", etc.) auth_type: Authentication type ("update" or "plugin") Returns: Dictionary with headers and other request config for requests library Example: >>> auth = AuthManager() >>> config = auth.create_authenticated_request_config() >>> response = requests.get(url, **config) """ return { "headers": self.get_headers(endpoint_type=auth_type), "timeout": 15, "verify": True, # SSL verification (S004) } def rotate_token(self, new_token: str) -> None: """ Rotate API token (for automated key rotation scenarios). Args: new_token: New API token to use Security Note: This bypasses environment variable loading. Use only in controlled environments. """ if not new_token or len(new_token) < 8: raise ValueError("Token must be at least 8 characters") self._api_token = new_token self._logger.warning("API token rotated (length=%d)", len(new_token)) def validate_token_format(self, token: str) -> Tuple[bool, str]: """ Validate token format before use. Args: token: Token string to validate Returns: Tuple of (is_valid, error_message) Validation rules: - Minimum 8 characters - No whitespace or special injection characters - Base64-like character set preferred but not required """ # Length check if len(token) < 8: return False, "Token too short (minimum 8 characters)" if len(token) > 1024: return False, "Token exceeds maximum length" # Check for suspicious characters that could be injection attempts suspicious_patterns = [";", "|", "&", "`", "$", "{", "}", "\n", "\r"] for pattern in suspicious_patterns: if pattern in token: return False, f"Token contains suspicious character: {repr(pattern)}" # Check for URL encoding that might indicate tampering if "%" in token: return False, "Token appears to be URL-encoded" return True, "" def get_update_endpoint_url(auth_manager: Optional[AuthManager] = None) -> str: """ Get the update endpoint URL with optional override. Args: auth_manager: AuthManager instance (optional, uses env var if not provided) Returns: URL string for update endpoint Environment Override: FLARES_UPDATE_ENDPOINT_URL can override the default """ return os.environ.get( "FLARES_UPDATE_ENDPOINT_URL", auth_manager.update_endpoint_url if auth_manager else AuthManager.DEFAULT_UPDATE_URL ) def hash_token_for_logging(token: str) -> str: """ Create a hashed representation of token for safe logging. Args: token: Original token string Returns: First 8 characters + ... (never log full token) Security Note: This is not cryptographically secure - only for display/logging purposes. Never log the actual token value. """ if not token: return "[REDACTED]" # Show first 4 chars, then hash of remainder for uniqueness prefix = token[:4] if len(token) >= 4 else token suffix_hash = hashlib.sha256(token.encode()).hexdigest()[:8] return f"{prefix}...{suffix_hash}" # Convenience singleton for application-wide use _default_auth_manager: Optional[AuthManager] = None def get_default_auth_manager(api_token: Optional[str] = None) -> AuthManager: """ Get or create the default authentication manager. Args: api_token: Override token (only on first call) Returns: Default AuthManager instance """ global _default_auth_manager if _default_auth_manager is None: _default_auth_manager = AuthManager(api_token=api_token) return _default_auth_manager def reset_default_auth_manager() -> None: """Reset the default auth manager (primarily for testing).""" global _default_auth_manager _default_auth_manager = None