security.py

"""Secure configuration management for the TNFR engine.

This module provides utilities for loading configuration from environment
variables with validation and secure defaults. It ensures that sensitive
credentials are never hardcoded in source code.

Security Principles:
- Never hardcode secrets, API keys, or passwords
- Load sensitive values from environment variables
- Provide secure defaults for development
- Validate configuration before use
- Read the process environment; loading an environment file is the caller's job
- Sanitize credentials in logs to prevent exposure
- Secure memory management for secrets
- Credential rotation and TTL support
"""

from __future__ import annotations

import os
import secrets
import time
import warnings
from datetime import datetime, timedelta, timezone
from typing import Any, Callable
from urllib.parse import quote, unquote, urlparse, urlunparse

from .parsing import parse_bool


class ConfigurationError(Exception):
    """Raised when configuration is invalid or missing required values."""


class SecurityAuditWarning(UserWarning):
    """Warning for security audit findings that don't stop execution."""


def get_env_variable(
    name: str,
    default: str | None = None,
    required: bool = False,
    secret: bool = False,
) -> str | None:
    """Get an environment variable with validation.

    Parameters
    ----------
    name : str
        The name of the environment variable to retrieve.
    default : str, optional
        Default value if the environment variable is not set.
    required : bool, default=False
        If True, raise ConfigurationError if the variable is not set.
    secret : bool, default=False
        If True, this is a sensitive value (password, token, etc.).
        Warnings will be issued if using defaults for secrets.

    Returns
    -------
    str or None
        The value of the environment variable, or the default value.

    Raises
    ------
    ConfigurationError
        If required=True and the variable is not set.

    Examples
    --------
    >>> # Get optional configuration with default
    >>> log_level = get_env_variable("TNFR_LOG_LEVEL", default="INFO")

    >>> # Get required secret (will raise if not set)
    >>> api_token = get_env_variable(
    ...     "GITHUB_TOKEN",
    ...     required=True,
    ...     secret=True
    ... )

    >>> # Get optional secret (will warn if using default)
    >>> redis_password = get_env_variable(
    ...     "REDIS_PASSWORD",
    ...     default="",
    ...     secret=True
    ... )
    """
    value = os.environ.get(name)

    if value is None:
        if required:
            raise ConfigurationError(
                f"Required environment variable '{name}' is not set. "
                f"Please set it in the process environment."
            )
        if secret and default is not None:
            warnings.warn(
                f"Using default value for secret '{name}'. "
                f"set the environment variable for production use.",
                stacklevel=2,
            )
        return default

    return value


def load_pypi_credentials() -> dict[str, str | None]:
    """Load PyPI publishing credentials from environment.

    Returns
    -------
    dict
        Dictionary containing username, password, and repository settings.

    Notes
    -----
    This function reads from multiple environment variables to support
    different tools (twine, poetry, etc.):

    - PYPI_USERNAME or TWINE_USERNAME
    - PYPI_PASSWORD, PYPI_API_TOKEN, or TWINE_PASSWORD
    - PYPI_REPOSITORY (defaults to 'pypi')

    Best Practice
    -------------
    Use API tokens instead of passwords:
    - PYPI_USERNAME=__token__
    - PYPI_PASSWORD=pypi-XXXXXXXXXXXXXXXXXXXX...

    Note: Example uses 'XXX' pattern to avoid triggering security scanners.
    Actual PyPI tokens follow format: pypi-AgEIcHlwaS5vcmcC...

    See Also
    --------
    https://pypi.org/help/#apitoken : PyPI API token documentation
    """
    username = os.environ.get("PYPI_USERNAME") or os.environ.get("TWINE_USERNAME")
    password = (
        os.environ.get("PYPI_PASSWORD")
        or os.environ.get("PYPI_API_TOKEN")
        or os.environ.get("TWINE_PASSWORD")
    )
    repository = os.environ.get("PYPI_REPOSITORY", "pypi")

    return {
        "username": username,
        "password": password,
        "repository": repository,
    }


def load_github_credentials() -> dict[str, str | None]:
    """Load GitHub API credentials from environment.

    Returns
    -------
    dict
        Dictionary containing token and repository information.

    Notes
    -----
    This function reads GITHUB_TOKEN and GITHUB_REPOSITORY environment
    variables commonly set in GitHub Actions and other CI environments.

    Best Practice
    -------------
    Use fine-grained personal access tokens with minimal scopes:
    - For security scans: read:security_events
    - For releases: contents:write, packages:write

    See Also
    --------
    https://docs.github.com/en/authentication/keeping-your-account-and-data-secure/creating-a-personal-access-token
    """
    token = os.environ.get("GITHUB_TOKEN")
    repository = os.environ.get("GITHUB_REPOSITORY")

    return {
        "token": token,
        "repository": repository,
    }


def _parse_redis_database(value: str) -> int:
    """Normalize the database index for URL and individual-variable inputs."""
    try:
        database = int(value)
    except (TypeError, ValueError) as exc:
        raise ConfigurationError("REDIS_DB must be a nonnegative integer") from exc
    if database < 0:
        raise ConfigurationError("REDIS_DB must be a nonnegative integer")
    return database


def load_redis_config(validate_url: bool = True) -> dict[str, Any]:
    """Load Redis connection configuration from environment.

    Parameters
    ----------
    validate_url : bool, default=True
        Whether to validate the constructed Redis URL.

    Returns
    -------
    dict
        Dictionary containing Redis connection parameters.

    Notes
    -----
    Supports standard Redis configuration variables:

    - REDIS_HOST (default: 'localhost')
    - REDIS_PORT (default: 6379)
    - REDIS_PASSWORD (optional)
    - REDIS_DB (default: 0)
    - REDIS_USE_TLS (default: False)
    - REDIS_URL (alternative: full URL, overrides individual params)

    URL passwords are percent-decoded for use as connection parameters.
    Database indices must be nonnegative integers in either input form.

    Security
    --------
    Always use authentication (REDIS_PASSWORD) in production.
    Enable TLS (REDIS_USE_TLS=true) for network connections.
    URLs with credentials are validated and sanitized for logging.

    See Also
    --------
    tnfr.utils.RedisCacheLayer : Redis cache implementation
    SecureCredentialValidator : URL validation and sanitization
    """
    # Check if full URL is provided
    redis_url = get_env_variable("REDIS_URL", default=None)

    if redis_url:
        # Validate URL if requested
        if validate_url:
            SecureCredentialValidator.validate_redis_url(redis_url)

        # Parse URL to extract components
        parsed = urlparse(redis_url)

        return {
            "host": parsed.hostname or "localhost",
            "port": 6379 if parsed.port is None else parsed.port,
            "password": (
                unquote(parsed.password) if parsed.password is not None else None
            ),
            "db": _parse_redis_database(parsed.path.lstrip("/") or "0"),
            "ssl": parsed.scheme == "rediss",
            "url": redis_url,
        }

    # Load from individual variables
    host = get_env_variable("REDIS_HOST", default="localhost")
    port_str = get_env_variable("REDIS_PORT", default="6379")
    password = get_env_variable("REDIS_PASSWORD", default=None, secret=True)
    db_str = get_env_variable("REDIS_DB", default="0")
    use_tls_str = get_env_variable("REDIS_USE_TLS", default="false")

    try:
        port = int(port_str)
    except ValueError:
        raise ConfigurationError(f"REDIS_PORT must be an integer, got: {port_str}")

    # Validate port range
    if not (1 <= port <= 65535):
        raise ConfigurationError(f"REDIS_PORT must be between 1 and 65535, got: {port}")

    db = _parse_redis_database(db_str)

    try:
        use_tls = parse_bool(use_tls_str)
    except ValueError as exc:
        raise ConfigurationError("REDIS_USE_TLS must be a boolean") from exc

    # Construct URL for validation
    if validate_url:
        scheme = "rediss" if use_tls else "redis"
        url_host = f"[{host}]" if ":" in host and not host.startswith("[") else host
        if password:
            url = f"{scheme}://:{quote(password, safe='')}@{url_host}:{port}/{db}"
        else:
            url = f"{scheme}://{url_host}:{port}/{db}"
        SecureCredentialValidator.validate_redis_url(url)

    return {
        "host": host,
        "port": port,
        "password": password,
        "db": db,
        "ssl": use_tls,
    }


def get_cache_secret() -> bytes | None:
    """Get the cache signing secret from environment.

    Returns
    -------
    bytes or None
        The cache secret as bytes, or None if not configured.

    Notes
    -----
    Reads from TNFR_CACHE_SECRET environment variable. The secret should
    be a hex-encoded string (recommended length: 64 characters / 32 bytes).

    Security
    --------
    Use a cryptographically strong random secret:

    >>> import secrets
    >>> secret = secrets.token_hex(32)  # 64-character hex string
    >>> # set TNFR_CACHE_SECRET=<secret> in your environment

    See Also
    --------
    tnfr.utils.ShelveCacheLayer : Shelf cache with signature support
    tnfr.utils.RedisCacheLayer : Redis cache with signature support
    """
    secret_hex = get_env_variable("TNFR_CACHE_SECRET", secret=True)
    if secret_hex is None:
        return None

    try:
        return bytes.fromhex(secret_hex)
    except ValueError as exc:
        raise ConfigurationError(
            f"TNFR_CACHE_SECRET must be a hex-encoded string: {exc}"
        )


def validate_no_hardcoded_secrets(value: str) -> bool:
    """Validate that a string doesn't look like a hardcoded secret.

    Parameters
    ----------
    value : str
        The string to validate.

    Returns
    -------
    bool
        True if the value passes validation.

    Raises
    ------
    ValueError
        If the value appears to be a hardcoded secret.

    Notes
    -----
    This is a heuristic check for common secret patterns:

    - Long alphanumeric strings (potential tokens)
    - Known secret prefixes (ghp_, pypi-, sk-, etc.)
    - Base64-encoded strings

    For production environments, consider using more sophisticated
    tools like `detect-secrets` which employ entropy analysis for
    better accuracy.

    Examples
    --------
    >>> validate_no_hardcoded_secrets("my-password")
    True

    >>> validate_no_hardcoded_secrets("ghp_abcd1234...")
    Traceback (most recent call last):
        ...
    ValueError: Value appears to be a hardcoded GitHub token
    """
    # Check for known secret prefixes
    secret_prefixes = [
        ("ghp_", "GitHub token"),
        ("gho_", "GitHub OAuth token"),
        ("ghu_", "GitHub user token"),
        ("ghs_", "GitHub server token"),
        ("ghr_", "GitHub refresh token"),
        ("pypi-", "PyPI token"),
        ("sk-", "OpenAI API key"),
        ("xoxb-", "Slack bot token"),
        ("xoxp-", "Slack user token"),
    ]

    for prefix, name in secret_prefixes:
        if value.startswith(prefix):
            raise ValueError(f"Value appears to be a hardcoded {name}")

    # Check for suspiciously long alphanumeric strings
    # Note: This is a simple heuristic. For production use, consider
    # entropy-based analysis (e.g., using detect-secrets library)
    if len(value) > 32 and value.replace("-", "").replace("_", "").isalnum():
        # Allow environment variable names (typically uppercase)
        if not value.isupper():
            warnings.warn(
                f"Value looks like it might be a hardcoded secret: {value[:10]}...",
                stacklevel=2,
            )

    return True


class SecureCredentialValidator:
    """Robust credential and configuration validator.

    Validates credentials and configuration with strict security criteria
    following TNFR principles of structural coherence and stability.
    """

    ALLOWED_SCHEMES = frozenset(["redis", "rediss"])  # Only secure schemes
    MAX_URL_LENGTH = 512  # Prevent DoS attacks
    MIN_SECRET_LENGTH = 8  # Minimum secret strength

    @staticmethod
    def validate_redis_url(url: str) -> bool:
        """Validate Redis URL with strict security criteria.

        Parameters
        ----------
        url : str
            Redis URL to validate.

        Returns
        -------
        bool
            True if URL is valid.

        Raises
        ------
        ValueError
            If URL fails validation checks.

        Examples
        --------
        >>> SecureCredentialValidator.validate_redis_url("redis://localhost:6379/0")
        True

        >>> SecureCredentialValidator.validate_redis_url("http://evil.com")
        Traceback (most recent call last):
            ...
        ValueError: Unsupported scheme: http
        """
        if not url or not isinstance(url, str):
            raise ValueError("Redis URL must be a non-empty string")

        if len(url) > SecureCredentialValidator.MAX_URL_LENGTH:
            raise ValueError(
                f"Redis URL exceeds maximum length of {SecureCredentialValidator.MAX_URL_LENGTH}"
            )

        try:
            parsed = urlparse(url)
        except Exception as exc:
            raise ValueError(f"Invalid URL format: {exc}")

        if parsed.scheme not in SecureCredentialValidator.ALLOWED_SCHEMES:
            raise ValueError(
                f"Unsupported scheme: {parsed.scheme}. "
                f"Allowed: {', '.join(SecureCredentialValidator.ALLOWED_SCHEMES)}"
            )

        if not parsed.hostname:
            raise ValueError("Redis URL must include a hostname")

        # Validate port if specified
        if parsed.port is not None:
            if not (1 <= parsed.port <= 65535):
                raise ValueError(f"Invalid port number: {parsed.port}")

        return True

    @staticmethod
    def sanitize_for_logging(url: str) -> str:
        """Sanitize URL for safe logging (hide credentials).

        Parameters
        ----------
        url : str
            URL that may contain credentials.

        Returns
        -------
        str
            Sanitized URL with credentials masked.

        Examples
        --------
        >>> SecureCredentialValidator.sanitize_for_logging(
        ...     "redis://user:secret@host:6379/0"
        ... )
        'redis://user:***@host:6379/0'
        """
        if not url:
            return url

        try:
            parsed = urlparse(url)

            # Check if parsing actually succeeded
            if not parsed.scheme and not parsed.netloc:
                # This is not a valid URL
                return "<invalid-url>"
        except Exception:
            # If parsing fails, return a safe placeholder
            return "<invalid-url>"

        # Mask password if present
        if parsed.password:
            # Replace password with ***
            netloc = parsed.netloc
            if "@" in netloc:
                userinfo, hostinfo = netloc.rsplit("@", 1)
                if ":" in userinfo:
                    username, _ = userinfo.split(":", 1)
                    netloc = f"{username}:***@{hostinfo}"
                else:
                    netloc = f"***@{hostinfo}"

            sanitized = parsed._replace(netloc=netloc)
            return urlunparse(sanitized)

        return url

    @staticmethod
    def validate_secret_strength(secret: str | bytes, min_length: int = 8) -> bool:
        """Validate that a secret meets minimum strength requirements.

        Parameters
        ----------
        secret : str or bytes
            The secret to validate.
        min_length : int, default=8
            Minimum required length.

        Returns
        -------
        bool
            True if secret is strong enough.

        Raises
        ------
        ValueError
            If secret is too weak.
        """
        if isinstance(secret, bytes):
            length = len(secret)
            secret_str = secret.decode("utf-8", errors="ignore")
        else:
            length = len(secret)
            secret_str = secret

        # Check for common weak passwords first (before length check)
        # This provides more specific error messages
        weak_passwords = ["password", "123456", "admin", "secret", "test", "changeme"]
        if secret_str.lower() in weak_passwords:
            raise ValueError("Secret matches a known weak password")

        # Then check length
        if length < min_length:
            raise ValueError(f"Secret too short: {length} < {min_length} (minimum)")

        return True


class SecureSecretManager:
    """Secure secret management with automatic memory cleanup.

    Manages secrets in memory with secure cleanup to prevent exposure
    through memory dumps. Implements structural coherence principles
    by ensuring secrets maintain integrity throughout their lifecycle.
    """

    def __init__(self) -> None:
        """Initialize secure secret manager."""
        self._secrets: dict[str, bytearray] = {}
        self._access_log: list[tuple[str, float]] = []

    def store_secret(self, key: str, secret: bytes | str) -> None:
        """Store a secret securely.

        Parameters
        ----------
        key : str
            Identifier for the secret.
        secret : bytes or str
            The secret value to store.
        """
        if isinstance(secret, str):
            secret_bytes = secret.encode("utf-8")
        else:
            secret_bytes = secret

        # Store as mutable bytearray for secure clearing
        self._secrets[key] = bytearray(secret_bytes)

    def get_secret(self, key: str) -> bytes:
        """Get a secret with access tracking.

        Parameters
        ----------
        key : str
            Secret identifier.

        Returns
        -------
        bytes
            Copy of the secret (not direct reference).
        """
        self._access_log.append((key, time.time()))
        secret_array = self._secrets.get(key)
        if secret_array is None:
            return b""
        # Return copy to prevent external mutation
        return bytes(secret_array)

    def clear_secret(self, key: str) -> None:
        """Clear a secret from memory securely.

        Parameters
        ----------
        key : str
            Secret identifier to clear.
        """
        if key in self._secrets:
            # Overwrite with random bytes before deletion
            secret_array = self._secrets[key]
            for i in range(len(secret_array)):
                secret_array[i] = secrets.randbits(8) & 0xFF
            del self._secrets[key]

    def clear_all(self) -> None:
        """Clear all secrets from memory."""
        for key in list(self._secrets.keys()):
            self.clear_secret(key)

    def get_access_log(self) -> list[tuple[str, float]]:
        """Get access log for auditing.

        Returns
        -------
        list of tuples
            list of (key, timestamp) tuples.
        """
        return self._access_log.copy()

    def __del__(self) -> None:
        """Cleanup on destruction."""
        self.clear_all()


class CredentialRotationManager:
    """Manages credential rotation with TTL support.

    Implements structural reorganization principle by managing
    credential lifecycle and triggering rotation when coherence
    (validity period) decreases.
    """

    def __init__(
        self,
        rotation_interval: timedelta = timedelta(hours=24),
        warning_threshold: timedelta = timedelta(hours=2),
    ) -> None:
        """Initialize rotation manager.

        Parameters
        ----------
        rotation_interval : timedelta, default=24 hours
            How often credentials should be rotated.
        warning_threshold : timedelta, default=2 hours
            When to warn about upcoming expiration.
        """
        self.rotation_interval = rotation_interval
        self.warning_threshold = warning_threshold
        self._last_rotation: dict[str, datetime] = {}
        self._rotation_callbacks: dict[str, Callable[[], None]] = {}

    def register_credential(
        self,
        credential_key: str,
        rotation_callback: Callable[[], None] | None = None,
    ) -> None:
        """Register a credential for rotation tracking.

        Parameters
        ----------
        credential_key : str
            Identifier for the credential.
        rotation_callback : callable, optional
            Function to call when rotation is needed.
        """
        self._last_rotation[credential_key] = datetime.now(timezone.utc)
        if rotation_callback is not None:
            self._rotation_callbacks[credential_key] = rotation_callback

    def needs_rotation(self, credential_key: str) -> bool:
        """Check if credential needs rotation.

        Parameters
        ----------
        credential_key : str
            Credential identifier.

        Returns
        -------
        bool
            True if rotation is needed.
        """
        last = self._last_rotation.get(credential_key)
        if last is None:
            return True
        age = datetime.now(timezone.utc) - last
        return age >= self.rotation_interval

    def needs_warning(self, credential_key: str) -> bool:
        """Check if credential is nearing expiration.

        Parameters
        ----------
        credential_key : str
            Credential identifier.

        Returns
        -------
        bool
            True if warning should be issued.
        """
        last = self._last_rotation.get(credential_key)
        if last is None:
            return True
        age = datetime.now(timezone.utc) - last
        time_until_rotation = self.rotation_interval - age
        return time_until_rotation <= self.warning_threshold

    def rotate_if_needed(self, credential_key: str) -> bool:
        """Rotate credential if needed.

        Parameters
        ----------
        credential_key : str
            Credential identifier.

        Returns
        -------
        bool
            True if rotation was performed.
        """
        if self.needs_rotation(credential_key):
            callback = self._rotation_callbacks.get(credential_key)
            if callback is not None:
                callback()
            self._last_rotation[credential_key] = datetime.now(timezone.utc)
            return True
        return False

    def get_credential_age(self, credential_key: str) -> timedelta | None:
        """Get age of credential.

        Parameters
        ----------
        credential_key : str
            Credential identifier.

        Returns
        -------
        timedelta or None
            Age of credential, or None if not registered.
        """
        last = self._last_rotation.get(credential_key)
        if last is None:
            return None
        return datetime.now(timezone.utc) - last


class SecurityAuditor:
    """Security auditor for configuration and environment.

    Implements diagnostic nodal analysis to identify security
    coherence issues and dissonances in configuration.
    """

    SENSITIVE_PATTERNS = frozenset(
        [
            "password",
            "secret",
            "key",
            "token",
            "credential",
            "api_key",
            "apikey",
            "auth",
            "private",
        ]
    )

    WEAK_VALUES = frozenset(
        [
            "password",
            "123456",
            "admin",
            "secret",
            "test",
            "changeme",
            "default",
            "root",
            "toor",
        ]
    )

    def audit_environment_variables(self) -> list[str]:
        """Audit environment variables for security issues.

        Returns
        -------
        list of str
            list of security issues found.
        """
        issues = []

        for var_name in os.environ:
            var_name_lower = var_name.lower()
            var_value = os.environ[var_name]

            # Check if this is a sensitive variable
            is_sensitive = any(
                pattern in var_name_lower for pattern in self.SENSITIVE_PATTERNS
            )

            if is_sensitive:
                # Check for weak values
                if var_value.lower() in self.WEAK_VALUES:
                    issues.append(
                        f"Weak/default value in sensitive variable: {var_name}"
                    )

                # Check for too short secrets
                if len(var_value) < 8:
                    issues.append(
                        f"Secret too short ({len(var_value)} chars) in: {var_name}"
                    )

                # Check if secret looks like a placeholder
                if var_value in ["your-secret", "your-token", "changeme", "..."]:
                    issues.append(f"Placeholder value detected in: {var_name}")

        return issues

    def check_redis_config_security(self) -> list[str]:
        """Check Redis configuration for security issues.

        Returns
        -------
        list of str
            list of security issues found.
        """
        issues = []

        try:
            config = load_redis_config()
        except (ConfigurationError, ValueError):
            return ["Invalid Redis configuration"]
        if not config["password"]:
            issues.append("Redis authentication password not configured")
        if not config["ssl"]:
            issues.append("Redis TLS not enabled - unencrypted connection")

        return issues

    def check_cache_secret_security(self) -> list[str]:
        """Check cache secret configuration.

        Returns
        -------
        list of str
            list of security issues found.
        """
        issues = []

        cache_secret = os.environ.get("TNFR_CACHE_SECRET")
        if not cache_secret:
            issues.append("TNFR_CACHE_SECRET not set - unsigned cache data")
        else:
            # Check if secret is strong enough
            try:
                secret_bytes = bytes.fromhex(cache_secret)
                if len(secret_bytes) < 16:
                    issues.append(
                        f"TNFR_CACHE_SECRET too short: {len(secret_bytes)} bytes "
                        "(recommend 32+ bytes)"
                    )
            except ValueError:
                issues.append("TNFR_CACHE_SECRET is not valid hex")

        return issues

    def run_full_audit(self) -> dict[str, list[str]]:
        """Run complete security audit.

        Returns
        -------
        dict
            Dictionary mapping audit category to list of issues.
        """
        return {
            "environment_variables": self.audit_environment_variables(),
            "redis_config": self.check_redis_config_security(),
            "cache_secret": self.check_cache_secret_security(),
        }


# Global instances for convenience
_global_secret_manager: SecureSecretManager | None = None
_global_rotation_manager: CredentialRotationManager | None = None


def get_secret_manager() -> SecureSecretManager:
    """Get global secret manager instance.

    Returns
    -------
    SecureSecretManager
        Global secret manager instance.
    """
    global _global_secret_manager
    if _global_secret_manager is None:
        _global_secret_manager = SecureSecretManager()
    return _global_secret_manager


def get_rotation_manager() -> CredentialRotationManager:
    """Get global rotation manager instance.

    Returns
    -------
    CredentialRotationManager
        Global rotation manager instance.
    """
    global _global_rotation_manager
    if _global_rotation_manager is None:
        _global_rotation_manager = CredentialRotationManager()
    return _global_rotation_manager