openmind/backend/app/config.py

"""Configuration applicative, entierement pilotee par variables d'environnement.

Toute valeur illisible arrete le demarrage avec un message nommant la variable
fautive. Une instance publique ne doit jamais tourner en croyant appliquer une
protection qu'une faute de frappe aurait desactivee.
"""

from __future__ import annotations

import ipaddress
from functools import lru_cache

from pydantic import Field, ValidationError, field_validator, model_validator
from pydantic_settings import BaseSettings, SettingsConfigDict

from app.services.rate_limit import RateLimitSyntaxError, parse_rate


class ConfigurationError(RuntimeError):
    """Configuration inexploitable : le demarrage doit s'arreter ici."""


def _validate_rate(value: str) -> str:
    """Verifie une regle de debit sans la conserver : elle est relue a l'usage."""
    parse_rate(value)  # leve RateLimitSyntaxError si la syntaxe est fautive
    return value


def parse_cidrs(raw: str) -> list[ipaddress.IPv4Network | ipaddress.IPv6Network]:
    """Lit une liste de reseaux CIDR separes par des virgules.

    `strict=False` accepte aussi bien `10.0.0.0/8` qu'une adresse seule comme
    `192.0.2.7`, qui devient alors un reseau /32.
    """
    networks: list[ipaddress.IPv4Network | ipaddress.IPv6Network] = []
    for entry in raw.split(","):
        text = entry.strip()
        if not text:
            continue
        try:
            networks.append(ipaddress.ip_network(text, strict=False))
        except ValueError:
            raise ValueError(
                f"'{text}' n'est pas un reseau valide. Attendu : une adresse ou un CIDR, "
                "par exemple 172.16.0.0/12, 10.0.0.0/8 ou 2001:db8::/32."
            ) from None
    return networks


class Settings(BaseSettings):
    """Parametres du backend.

    Chaque champ est surchargeable via une variable d'environnement de meme nom
    (insensible a la casse) : APP_NAME, DATABASE_URL, MAX_NODES, ...
    """

    model_config = SettingsConfigDict(env_file=".env", extra="ignore", case_sensitive=False)

    app_name: str = "OpenMind"
    app_port: int = 8080
    app_env: str = "production"

    database_url: str = "sqlite:////app/data/openmind.db"
    static_dir: str = "/app/static"

    max_map_size_mb: float = Field(default=10.0, gt=0)
    max_nodes: int = Field(default=5000, gt=0)
    max_title_length: int = Field(default=200, gt=0)

    # --- Exploitation publique -------------------------------------------
    # Instance entierement privee : le partage serveur peut etre coupe sans
    # rien changer au fonctionnement local des cartes.
    sharing_enabled: bool = True

    # Limites de debit, au format "<nombre>/<fenetre>" : 10/hour, 120/minute,
    # 5/second. Seules les valeurs 0, off, none, disabled ou vide desactivent
    # une limite ; toute autre valeur illisible arrete le demarrage.
    share_create_rate_limit: str = "10/hour"
    share_read_rate_limit: str = "120/minute"
    share_write_rate_limit: str = "60/hour"

    # Nombre total de partages heberges. 0 = illimite.
    max_shared_maps: int = Field(default=10000, ge=0)

    # Taille maximale d'un partage. 0 = on retombe sur max_map_size_mb.
    max_share_size_mb: float = Field(default=0.0, ge=0)

    # Duree de vie par defaut d'un nouveau partage, en jours. 0 = permanent.
    default_share_expiry_days: int = Field(default=30, ge=0)
    # Autorise l'option "Jamais" dans l'interface et via l'API.
    allow_permanent_shares: bool = True

    # A n'activer que derriere un reverse proxy de confiance, et uniquement avec
    # trusted_proxy_cidrs renseigne : sinon un client direct pourrait falsifier
    # son adresse et contourner les limites de debit.
    trust_proxy_headers: bool = False
    # Reseaux depuis lesquels X-Forwarded-For est accepte, separes par des
    # virgules. IPv4, IPv6 et CIDR acceptes.
    trusted_proxy_cidrs: str = ""

    # Liste d'origines autorisees, separees par des virgules.
    # Vide = aucune origine externe (l'app est servie par ce meme serveur).
    cors_origins: str = ""

    log_level: str = "INFO"

    @field_validator("share_create_rate_limit", "share_read_rate_limit", "share_write_rate_limit")
    @classmethod
    def _check_rate(cls, value: str) -> str:
        return _validate_rate(value)

    @field_validator("trusted_proxy_cidrs")
    @classmethod
    def _check_cidrs(cls, value: str) -> str:
        parse_cidrs(value)
        return value

    @model_validator(mode="after")
    def _check_proxy_trust(self) -> Settings:
        """Faire confiance a un proxy sans dire lequel ne protege de rien.

        Autant le dire au demarrage : sans reseau declare, aucun en-tete ne
        serait accepte et l'exploitant croirait sa configuration active.
        """
        if self.trust_proxy_headers and not parse_cidrs(self.trusted_proxy_cidrs):
            raise ValueError(
                "TRUST_PROXY_HEADERS=true exige TRUSTED_PROXY_CIDRS : indiquez les reseaux "
                "depuis lesquels X-Forwarded-For peut etre accepte, par exemple "
                "172.16.0.0/12,10.0.0.0/8."
            )
        return self

    @property
    def max_map_size_bytes(self) -> int:
        return int(self.max_map_size_mb * 1024 * 1024)

    @property
    def max_share_size_bytes(self) -> int:
        """Limite propre aux partages, bornee par la limite HTTP globale."""
        if self.max_share_size_mb <= 0:
            return self.max_map_size_bytes
        return min(int(self.max_share_size_mb * 1024 * 1024), self.max_map_size_bytes)

    @property
    def max_share_size_mb_effective(self) -> float:
        return self.max_share_size_bytes / (1024 * 1024)

    @property
    def trusted_proxy_networks(self) -> list[ipaddress.IPv4Network | ipaddress.IPv6Network]:
        return parse_cidrs(self.trusted_proxy_cidrs)

    @property
    def cors_origin_list(self) -> list[str]:
        return [origin.strip() for origin in self.cors_origins.split(",") if origin.strip()]

    @property
    def is_debug(self) -> bool:
        return self.app_env.lower() in {"dev", "development", "debug"}


def _readable_error(error: ValidationError) -> str:
    """Traduit une erreur Pydantic en message nommant la variable fautive."""
    lines = []
    for item in error.errors():
        variable = str(item["loc"][0]).upper() if item.get("loc") else "CONFIGURATION"
        reason = str(item.get("msg", "")).replace("Value error, ", "")
        lines.append(f"  {variable} : {reason}")
    return "\n".join(lines)


@lru_cache
def get_settings() -> Settings:
    try:
        return Settings()
    except RateLimitSyntaxError as error:  # pragma: no cover - filet de securite
        raise ConfigurationError(f"Configuration invalide :\n  {error}") from None
    except ValidationError as error:
        raise ConfigurationError(f"Configuration invalide :\n{_readable_error(error)}") from None