startup_validation.py

import ipaddress
import json
import re
from typing import Any, Dict, List

from config import Settings, settings

POSTGRES_RESERVED_CONNECTIONS = 10
MIN_PRODUCTION_SECRET_LENGTH = 20
PLACEHOLDER_PATTERN = re.compile(
    r"^\$\{.+\}$|<.+>|CHANGE_ME|REPLACE_ME|FROM_SECRET",
    re.IGNORECASE,
)


class StartupValidationError(Exception):
    def __init__(self, errors: List[Dict[str, Any]]) -> None:
        self.errors = errors
        super().__init__("Startup validation failed")


def validate_settings(current_settings: Settings = settings) -> Dict[str, Any]:
    errors: List[Dict[str, Any]] = []

    if current_settings.repository_backend not in ["json", "postgres"]:
        errors.append({
            "field": "REPOSITORY_BACKEND",
            "message": "Must be either 'json' or 'postgres'",
        })

    if current_settings.api_key_auth_backend not in ["env", "database"]:
        errors.append({
            "field": "API_KEY_AUTH_BACKEND",
            "message": "Must be either 'env' or 'database'",
        })

    if current_settings.api_process_model not in ["uvicorn", "gunicorn"]:
        errors.append({
            "field": "API_PROCESS_MODEL",
            "message": "Must be either 'uvicorn' or 'gunicorn'",
        })

    if current_settings.repository_backend == "postgres":
        require_non_empty(
            errors,
            field="DATABASE_URL",
            value=current_settings.database_url,
        )

    if (
        current_settings.api_key_auth_backend == "database"
        and current_settings.repository_backend != "postgres"
    ):
        errors.append({
            "field": "API_KEY_AUTH_BACKEND",
            "message": "Database API key backend requires REPOSITORY_BACKEND=postgres",
        })

    require_non_empty(
        errors,
        field="REDIS_URL",
        value=current_settings.redis_url,
    )
    require_non_empty(
        errors,
        field="DEEPSEEK_BASE_URL",
        value=current_settings.deepseek_base_url,
    )
    require_non_empty(
        errors,
        field="DEEPSEEK_MODEL",
        value=current_settings.deepseek_model,
    )

    if current_settings.app_env in ["docker", "prod", "production"]:
        require_non_empty(
            errors,
            field="DEEPSEEK_API_KEY",
            value=current_settings.deepseek_api_key,
        )
        if (
            current_settings.api_key_auth_backend == "env"
            and not current_settings.api_key
            and not current_settings.api_key_tenant_map
        ):
            errors.append({
                "field": "API_KEY",
                "message": "Must set API_KEY or API_KEY_TENANT_MAP",
            })

    if current_settings.app_env in ["prod", "production"]:
        if current_settings.grafana_admin_user.strip().lower() == "admin":
            errors.append({
                "field": "GRAFANA_ADMIN_USER",
                "message": "Must not use the default Grafana admin user in production",
            })
        if current_settings.grafana_admin_password == "admin":
            errors.append({
                "field": "GRAFANA_ADMIN_PASSWORD",
                "message": "Must not use the default Grafana password in production",
            })
        validate_production_guardrails(errors, current_settings)

    validate_api_key_tenant_map(errors, current_settings.api_key_tenant_map)
    validate_cidr_list(
        errors,
        "PLATFORM_OPERATOR_ALLOWED_CIDRS",
        current_settings.platform_operator_allowed_cidrs,
    )
    validate_cidr_list(
        errors,
        "TRUSTED_PROXY_CIDRS",
        current_settings.trusted_proxy_cidrs,
    )
    require_non_negative_number(
        errors,
        "DEEPSEEK_INPUT_TOKEN_PRICE_PER_1K",
        current_settings.deepseek_input_token_price_per_1k,
    )
    require_non_negative_number(
        errors,
        "DEEPSEEK_OUTPUT_TOKEN_PRICE_PER_1K",
        current_settings.deepseek_output_token_price_per_1k,
    )
    require_positive_int(
        errors,
        "DEEPSEEK_TIMEOUT_SECONDS",
        current_settings.deepseek_timeout_seconds,
    )
    require_positive_int(
        errors,
        "DEEPSEEK_MAX_CONCURRENCY",
        current_settings.deepseek_max_concurrency,
    )
    require_positive_int(
        errors,
        "DEEPSEEK_CIRCUIT_BREAKER_FAILURE_THRESHOLD",
        current_settings.deepseek_circuit_breaker_failure_threshold,
    )
    require_positive_int(
        errors,
        "DEEPSEEK_CIRCUIT_BREAKER_RECOVERY_SECONDS",
        current_settings.deepseek_circuit_breaker_recovery_seconds,
    )
    require_positive_int(errors, "POSTGRES_POOL_SIZE", current_settings.postgres_pool_size)
    require_non_negative_int(
        errors,
        "POSTGRES_MAX_OVERFLOW",
        current_settings.postgres_max_overflow,
    )
    require_non_negative_int(
        errors,
        "POSTGRES_MAX_CONNECTIONS",
        current_settings.postgres_max_connections,
    )
    require_positive_int(
        errors,
        "POSTGRES_POOL_RECYCLE_SECONDS",
        current_settings.postgres_pool_recycle_seconds,
    )
    require_positive_int(
        errors,
        "REDIS_SOCKET_TIMEOUT_SECONDS",
        current_settings.redis_socket_timeout_seconds,
    )
    require_positive_int(
        errors,
        "REDIS_HEALTH_CHECK_INTERVAL_SECONDS",
        current_settings.redis_health_check_interval_seconds,
    )

    require_positive_int(errors, "MAX_RUN_RETRIES", current_settings.max_run_retries)
    require_positive_int(errors, "MAX_AGENT_RETRIES", current_settings.max_agent_retries)
    require_positive_int(
        errors,
        "RETRY_BASE_DELAY_SECONDS",
        current_settings.retry_base_delay_seconds,
    )
    require_positive_int(
        errors,
        "RETRY_MAX_DELAY_SECONDS",
        current_settings.retry_max_delay_seconds,
    )
    require_positive_int(
        errors,
        "WORKER_HEARTBEAT_INTERVAL_SECONDS",
        current_settings.worker_heartbeat_interval_seconds,
    )
    require_positive_int(
        errors,
        "WORKER_HEARTBEAT_STALE_SECONDS",
        current_settings.worker_heartbeat_stale_seconds,
    )
    require_positive_int(errors, "WORKER_CONCURRENCY", current_settings.worker_concurrency)
    require_non_negative_int(
        errors,
        "WORKER_TENANT_INFLIGHT_LIMIT",
        current_settings.worker_tenant_inflight_limit,
    )
    require_positive_int(errors, "API_WORKERS", current_settings.api_workers)
    require_positive_int(
        errors,
        "API_TIMEOUT_SECONDS",
        current_settings.api_timeout_seconds,
    )
    require_positive_int(
        errors,
        "API_GRACEFUL_TIMEOUT_SECONDS",
        current_settings.api_graceful_timeout_seconds,
    )
    require_positive_int(
        errors,
        "API_KEEP_ALIVE_SECONDS",
        current_settings.api_keep_alive_seconds,
    )
    require_positive_int(
        errors,
        "API_MAX_REQUEST_BODY_BYTES",
        current_settings.api_max_request_body_bytes,
    )
    require_non_negative_int(
        errors,
        "API_RATE_LIMIT_PER_MINUTE",
        current_settings.api_rate_limit_per_minute,
    )
    require_non_negative_int(
        errors,
        "TENANT_RUN_QUOTA_PER_DAY",
        current_settings.tenant_run_quota_per_day,
    )
    require_non_negative_int(
        errors,
        "TENANT_MONTHLY_RUN_QUOTA",
        current_settings.tenant_monthly_run_quota,
    )
    require_non_negative_int(
        errors,
        "TENANT_MONTHLY_TOKEN_QUOTA",
        current_settings.tenant_monthly_token_quota,
    )
    require_non_negative_number(
        errors,
        "TENANT_MONTHLY_COST_QUOTA_USD",
        current_settings.tenant_monthly_cost_quota_usd,
    )
    require_positive_int(
        errors,
        "DATA_RETENTION_RUN_DAYS",
        current_settings.data_retention_run_days,
    )
    require_positive_int(
        errors,
        "DATA_RETENTION_AUDIT_EVENT_DAYS",
        current_settings.data_retention_audit_event_days,
    )
    require_positive_int(
        errors,
        "DATA_RETENTION_USAGE_LEDGER_DAYS",
        current_settings.data_retention_usage_ledger_days,
    )
    require_positive_int(
        errors,
        "TENANT_DELETION_GRACE_DAYS",
        current_settings.tenant_deletion_grace_days,
    )

    return {
        "valid": not errors,
        "errors": errors,
    }


def validate_startup_or_raise(current_settings: Settings = settings) -> None:
    result = validate_settings(current_settings)
    if not result["valid"]:
        raise StartupValidationError(result["errors"])


def require_non_empty(
    errors: List[Dict[str, Any]],
    field: str,
    value: str,
) -> None:
    if not value:
        errors.append({
            "field": field,
            "message": "Must not be empty",
        })


def require_positive_int(
    errors: List[Dict[str, Any]],
    field: str,
    value: int,
) -> None:
    if value <= 0:
        errors.append({
            "field": field,
            "message": "Must be a positive integer",
        })


def require_non_negative_int(
    errors: List[Dict[str, Any]],
    field: str,
    value: int,
) -> None:
    if value < 0:
        errors.append({
            "field": field,
            "message": "Must be a non-negative integer",
        })


def require_non_negative_number(
    errors: List[Dict[str, Any]],
    field: str,
    value: float,
) -> None:
    if value < 0:
        errors.append({
            "field": field,
            "message": "Must be a non-negative number",
        })


def validate_api_key_tenant_map(
    errors: List[Dict[str, Any]],
    raw_value: str,
) -> None:
    if not raw_value:
        return

    try:
        parsed = json.loads(raw_value)
    except json.JSONDecodeError:
        errors.append({
            "field": "API_KEY_TENANT_MAP",
            "message": "Must be a JSON object mapping API keys to tenant ids",
        })
        return

    if not isinstance(parsed, dict):
        errors.append({
            "field": "API_KEY_TENANT_MAP",
            "message": "Must be a JSON object mapping API keys to tenant ids",
        })


def validate_production_guardrails(
    errors: List[Dict[str, Any]],
    current_settings: Settings,
) -> None:
    production_secrets = {
        "DEEPSEEK_API_KEY": current_settings.deepseek_api_key,
        "METRICS_API_KEY": current_settings.metrics_api_key,
        "PLATFORM_OPERATOR_API_KEY": current_settings.platform_operator_api_key,
        "GRAFANA_ADMIN_PASSWORD": current_settings.grafana_admin_password,
    }
    for field, value in production_secrets.items():
        validate_production_secret(errors, field, value)

    if "agent_password" in current_settings.database_url:
        errors.append({
            "field": "DATABASE_URL",
            "message": "Must not use the default Postgres password in production",
        })

    if current_settings.api_rate_limit_per_minute <= 0:
        errors.append({
            "field": "API_RATE_LIMIT_PER_MINUTE",
            "message": "Must enable API rate limiting in production",
        })

    if current_settings.tenant_monthly_run_quota <= 0:
        errors.append({
            "field": "TENANT_MONTHLY_RUN_QUOTA",
            "message": "Must enable monthly run quota in production",
        })

    if not current_settings.metrics_api_key:
        errors.append({
            "field": "METRICS_API_KEY",
            "message": "Must protect metrics endpoint in production",
        })

    if not current_settings.platform_operator_api_key:
        errors.append({
            "field": "PLATFORM_OPERATOR_API_KEY",
            "message": "Must protect platform operator endpoints in production",
        })

    if not current_settings.platform_operator_allowed_cidrs:
        errors.append({
            "field": "PLATFORM_OPERATOR_ALLOWED_CIDRS",
            "message": "Must restrict platform operator endpoints by CIDR allowlist",
        })

    if not current_settings.trusted_proxy_cidrs:
        errors.append({
            "field": "TRUSTED_PROXY_CIDRS",
            "message": "Must configure trusted reverse proxy CIDRs in production",
        })

    if not current_settings.cors_allowed_origins:
        errors.append({
            "field": "CORS_ALLOWED_ORIGINS",
            "message": "Must explicitly configure allowed CORS origins in production",
        })

    if not current_settings.trusted_hosts:
        errors.append({
            "field": "TRUSTED_HOSTS",
            "message": "Must explicitly configure trusted hosts in production",
        })
    validate_public_boundary(errors, current_settings)

    if current_settings.api_process_model != "gunicorn":
        errors.append({
            "field": "API_PROCESS_MODEL",
            "message": "Must use gunicorn process model in production",
        })

    if current_settings.worker_tenant_inflight_limit <= 0:
        errors.append({
            "field": "WORKER_TENANT_INFLIGHT_LIMIT",
            "message": "Must enable per-tenant worker in-flight limiting in production",
        })

    validate_postgres_connection_budget(errors, current_settings)


def validate_production_secret(
    errors: List[Dict[str, Any]],
    field: str,
    value: str,
) -> None:
    if not value:
        return
    if PLACEHOLDER_PATTERN.search(value):
        errors.append({
            "field": field,
            "message": f"{field} must be a concrete production secret, not a placeholder",
        })
    if len(value) < MIN_PRODUCTION_SECRET_LENGTH:
        errors.append({
            "field": field,
            "message": (
                f"{field} must be at least "
                f"{MIN_PRODUCTION_SECRET_LENGTH} characters in production"
            ),
        })


def estimate_postgres_connection_budget(current_settings: Settings) -> int:
    per_process_pool_limit = (
        current_settings.postgres_pool_size
        + current_settings.postgres_max_overflow
    )
    runtime_processes = current_settings.api_workers + 2
    return runtime_processes * per_process_pool_limit + POSTGRES_RESERVED_CONNECTIONS


def validate_postgres_connection_budget(
    errors: List[Dict[str, Any]],
    current_settings: Settings,
) -> None:
    if current_settings.postgres_max_connections <= 0:
        errors.append({
            "field": "POSTGRES_MAX_CONNECTIONS",
            "message": "Must set the production Postgres max_connections budget",
        })
        return

    estimated_required = estimate_postgres_connection_budget(current_settings)
    if estimated_required > current_settings.postgres_max_connections:
        errors.append({
            "field": "POSTGRES_MAX_CONNECTIONS",
            "message": (
                "Production Postgres connection budget is too small: "
                f"estimated {estimated_required}, configured "
                f"{current_settings.postgres_max_connections}"
            ),
        })


def validate_public_boundary(
    errors: List[Dict[str, Any]],
    current_settings: Settings,
) -> None:
    validate_production_cors_origins(
        errors,
        current_settings.cors_allowed_origins,
    )
    validate_production_trusted_hosts(
        errors,
        current_settings.trusted_hosts,
    )


def split_csv_values(raw_value: str) -> List[str]:
    return [
        item.strip()
        for item in raw_value.split(",")
        if item.strip()
    ]


def is_local_boundary_value(value: str) -> bool:
    normalized = value.lower()
    return any(
        marker in normalized
        for marker in [
            "localhost",
            "127.0.0.1",
            "[::1]",
            "::1",
        ]
    )


def is_example_boundary_value(value: str) -> bool:
    return "example.com" in value.lower()


def validate_production_cors_origins(
    errors: List[Dict[str, Any]],
    raw_value: str,
) -> None:
    for origin in split_csv_values(raw_value):
        if "*" in origin:
            errors.append({
                "field": "CORS_ALLOWED_ORIGINS",
                "message": "Production CORS origins must not use wildcard values",
            })
        if not origin.startswith("https://"):
            errors.append({
                "field": "CORS_ALLOWED_ORIGINS",
                "message": "Production CORS origins must use https:// origins",
            })
        if is_local_boundary_value(origin):
            errors.append({
                "field": "CORS_ALLOWED_ORIGINS",
                "message": "Production CORS origins must not use localhost or loopback hosts",
            })
        if is_example_boundary_value(origin):
            errors.append({
                "field": "CORS_ALLOWED_ORIGINS",
                "message": "Production CORS origins must not use example.com placeholder hosts",
            })


def validate_production_trusted_hosts(
    errors: List[Dict[str, Any]],
    raw_value: str,
) -> None:
    for host in split_csv_values(raw_value):
        if "*" in host:
            errors.append({
                "field": "TRUSTED_HOSTS",
                "message": "Production trusted hosts must not use wildcard values",
            })
        if "://" in host:
            errors.append({
                "field": "TRUSTED_HOSTS",
                "message": "Production trusted hosts must be hostnames, not URL origins",
            })
        if is_local_boundary_value(host):
            errors.append({
                "field": "TRUSTED_HOSTS",
                "message": "Production trusted hosts must not use localhost or loopback hosts",
            })
        if is_example_boundary_value(host):
            errors.append({
                "field": "TRUSTED_HOSTS",
                "message": "Production trusted hosts must not use example.com placeholder hosts",
            })


def validate_cidr_list(
    errors: List[Dict[str, Any]],
    field: str,
    raw_value: str,
) -> None:
    if not raw_value:
        return

    for raw_cidr in raw_value.split(","):
        cidr = raw_cidr.strip()
        if not cidr:
            continue
        try:
            network = ipaddress.ip_network(cidr, strict=False)
        except ValueError:
            errors.append({
                "field": field,
                "message": f"Invalid CIDR value: {cidr}",
            })
            continue
        if network.prefixlen == 0:
            errors.append({
                "field": field,
                "message": f"{field} must not trust the whole internet: {cidr}",
            })