ros_control/app/config.py

import ipaddress
from functools import lru_cache

from cryptography.fernet import Fernet
from pydantic_settings import BaseSettings, SettingsConfigDict

# минимальная длина секретов, принимаемая при старте (см. insecure_settings)
API_TOKEN_MIN_LEN = 32
SESSION_SECRET_MIN_LEN = 32
ADMIN_PASSWORD_MIN_LEN = 12


class Settings(BaseSettings):
    model_config = SettingsConfigDict(env_file=".env", extra="ignore")

    database_url: str = "sqlite:///./data/ros_control.db"
    secret_key: str = ""  # ключ Fernet для паролей устройств
    session_secret: str = "change-me"
    session_cookie_secure: bool = False  # Secure-флаг cookie сессии; включить за TLS

    admin_user: str = "admin"
    admin_password: str = "change-me"
    api_token: str = "change-me"

    s3_endpoint: str = "https://storage.yandexcloud.net"
    s3_region: str = "ru-central1"
    s3_bucket: str = ""
    s3_access_key: str = ""
    s3_secret_key: str = ""
    s3_prefix: str = "backups"

    backups_cache_ttl: int = 60  # секунд кэш списка бакета живёт без обращения к S3; 0 — кэш выключен

    max_concurrency: int = 10
    ros_timeout: float = 30.0  # секунд на чтение ответа устройства (долгие операции)
    ros_connect_timeout: float = 4.0  # секунд на установление соединения: так быстро определяется недоступность
    poll_interval: int = 30  # период фонового опроса устройств, с; 0 — отключить
    update_check_interval: int = 1800  # как часто в опросе проверять обновления ROS (тяжёлый запрос), с

    # значения по умолчанию для ротации журнала; действующие настройки хранятся в БД и меняются из UI
    events_retention_days: int = 90  # 0 — без ограничения по сроку
    events_max_rows: int = 100000  # 0 — без ограничения по числу записей

    # доверенные прокси (CIDR через запятую, например 10.0.0.0/8,172.16.0.0/12; одиночный адрес — как /32 или /128):
    # только от них принимается X-Forwarded-For; пусто — заголовок всегда игнорируется, IP клиента берётся напрямую
    trusted_proxies: str = ""


@lru_cache
def get_settings() -> Settings:
    return Settings()


@lru_cache
def trusted_networks(value: str) -> tuple[ipaddress.IPv4Network | ipaddress.IPv6Network, ...]:
    """Сети TRUSTED_PROXIES, разобранные из строки настройки; результат кэшируется по значению строки."""
    return tuple(ipaddress.ip_network(part.strip(), strict=False) for part in value.split(",") if part.strip())


def insecure_settings(s: Settings) -> list[str]:
    """Проблемы конфигурации, с которыми приложению нельзя стартовать. Без значений секретов — только названия проблем."""
    problems = []
    if not s.api_token or s.api_token == "change-me":
        problems.append("API_TOKEN: пустой или change-me")
    elif len(s.api_token) < API_TOKEN_MIN_LEN:
        problems.append(f"API_TOKEN: короче {API_TOKEN_MIN_LEN} символов")
    if not s.session_secret or s.session_secret == "change-me":
        problems.append("SESSION_SECRET: пустой или change-me")
    elif len(s.session_secret) < SESSION_SECRET_MIN_LEN:
        problems.append(f"SESSION_SECRET: короче {SESSION_SECRET_MIN_LEN} символов")
    if not s.admin_password or s.admin_password == "change-me":
        problems.append("ADMIN_PASSWORD: пустой или change-me")
    elif len(s.admin_password) < ADMIN_PASSWORD_MIN_LEN:
        problems.append(f"ADMIN_PASSWORD: короче {ADMIN_PASSWORD_MIN_LEN} символов")
    if not s.secret_key:
        problems.append("SECRET_KEY: пустой")
    else:
        try:
            Fernet(s.secret_key.encode())
        except Exception:  # noqa: BLE001
            problems.append("SECRET_KEY: не ключ Fernet")
    if s.trusted_proxies:
        try:
            trusted_networks(s.trusted_proxies)
        except ValueError:
            problems.append(f"TRUSTED_PROXIES: неверная сеть {s.trusted_proxies!r}")
    return problems