"""Configuration de l'application via ``pydantic-settings``. Chaque sous-groupe de configuration est un modèle ``BaseSettings`` dédié, chargé depuis les variables d'environnement (préfixées par groupe) et le fichier ``.env``. Tous les champs disposent de valeurs par défaut sûres afin que ``Settings()`` fonctionne même sans fichier de configuration présent. """ from __future__ import annotations import warnings from datetime import date from typing import Annotated, Literal from urllib.parse import urlparse from pydantic import ( AfterValidator, BaseModel, ConfigDict, Field, SecretStr, field_serializer, model_validator, ) from pydantic_settings import BaseSettings, SettingsConfigDict from pronote_sync.utils.redaction import redact_url _EXTERNAL_ENDPOINT_SCHEMES: frozenset[str] = frozenset({"http", "https"}) _LOOPBACK_HOSTS: frozenset[str] = frozenset({"localhost", "127.0.0.1", "::1"}) def _validate_external_endpoint_url(value: SecretStr) -> SecretStr: """Valide la structure et le schéma réseau d'une URL d'endpoint externe. Le socle commun accepte uniquement les schémas ``http`` et ``https``, avec un hôte obligatoire. Les credentials embarqués (``user:pass@host``) sont refusés afin qu'aucun secret ne soit transporté dans l'URL. La restriction ``https``/HTTP loopback est ensuite affinée par chaque connecteur. :param value: URL potentiellement sensible à valider. :return: URL validée, toujours encapsulée dans ``SecretStr``. :rtype: SecretStr :raises ValueError: Si l'URL est malformée, sans hôte, utilise un schéma non réseau ou contient des credentials. """ is_valid = False try: parsed = urlparse(value.get_secret_value()) _ = parsed.port is_valid = ( parsed.scheme in _EXTERNAL_ENDPOINT_SCHEMES and parsed.hostname is not None and parsed.username is None and parsed.password is None ) except ValueError: pass if not is_valid: raise ValueError("Endpoint externe invalide : URL ou schéma non supporté") from None return value EndpointUrl = Annotated[SecretStr, AfterValidator(_validate_external_endpoint_url)] class ExternalEndpoint(BaseModel): """Représente un endpoint externe potentiellement sensible. Le socle accepte uniquement les transports réseau ``https`` et ``http`` (hôte obligatoire, sans credentials embarqués). Chaque connecteur restreint ensuite cette liste selon sa propre politique de sécurité. L'URL reste encapsulée dans :class:`pydantic.SecretStr` et sa sérialisation conserve uniquement une représentation expurgée. """ model_config = ConfigDict(extra="forbid", frozen=True, hide_input_in_errors=True) url: EndpointUrl @field_serializer("url") def _serialize_url(self, value: SecretStr) -> str: """Expurge l'URL lors de la sérialisation. :param value: URL encapsulée à sérialiser. :return: URL expurgée. :rtype: str """ return redact_url(value.get_secret_value()) class PronoteSettings(BaseSettings): """Paramètres d'accès à Pronote (flux iCal et API ``pronotepy``). Les variables d'environnement correspondantes sont préfixées par ``PRONOTE_``. """ model_config = SettingsConfigDict( env_file=".env", env_nested_delimiter="__", extra="ignore", env_prefix="PRONOTE_", hide_input_in_errors=True, ) endpoint: ExternalEndpoint | None = None ical_endpoint: ExternalEndpoint | None = None ical_url: SecretStr | None = Field( default=None, exclude=True, deprecated="Utiliser ical_endpoint.url à la place (PRONOTE_ICAL_URL obsolète).", ) username: str | None = None password: SecretStr | None = None ent: str | None = None url: str | None = Field( default=None, exclude=True, deprecated="Utiliser endpoint.url à la place (PRONOTE_URL obsolète).", ) account_type: Literal["student", "parent"] = "parent" agenda_source: Literal["auto", "ical", "pronotepy"] = "auto" homework_source: Literal["auto", "ical", "pronotepy"] = "auto" messages_source: Literal["pronotepy"] = "pronotepy" auth_mode: Literal["password", "qr_token"] = "password" qr_code_file: str | None = None qr_pin: SecretStr | None = None account_pin: SecretStr | None = None @model_validator(mode="before") @classmethod def _migrate_legacy_endpoints(cls, data: object) -> object: """Migre les URL Pronote historiques vers les endpoints communs. :param data: Données brutes du modèle. :return: Données complétées avec les endpoints si nécessaire. :rtype: object """ if not isinstance(data, dict): return data migrated_data = data.copy() if migrated_data.get("url") is not None: warnings.warn( "PRONOTE_URL est obsolète : utiliser PRONOTE_ENDPOINT__URL.", DeprecationWarning, stacklevel=2, ) if migrated_data.get("endpoint") is None: migrated_data["endpoint"] = {"url": migrated_data["url"]} if migrated_data.get("ical_url") is not None: warnings.warn( "PRONOTE_ICAL_URL est obsolète : utiliser PRONOTE_ICAL_ENDPOINT__URL.", DeprecationWarning, stacklevel=2, ) if migrated_data.get("ical_endpoint") is None: migrated_data["ical_endpoint"] = {"url": migrated_data["ical_url"]} return migrated_data @model_validator(mode="after") def _validate_endpoint_policies(self) -> PronoteSettings: """Applique les transports autorisés aux deux endpoints Pronote. L'API Pronote et le flux iCal exigent tous deux HTTPS : aucun fichier local n'est accepté. :return: Instance validée inchangée. :rtype: PronoteSettings :raises ValueError: Si un endpoint n'utilise pas HTTPS. """ if ( self.endpoint is not None and urlparse(self.endpoint.url.get_secret_value()).scheme != "https" ): raise ValueError("URL Pronote invalide : HTTPS requis") from None if ( self.ical_endpoint is not None and urlparse(self.ical_endpoint.url.get_secret_value()).scheme != "https" ): raise ValueError("URL iCal Pronote invalide : HTTPS requis") from None return self @field_serializer("qr_pin") def _serialize_qr_pin(self, value: SecretStr | None) -> str | None: """Masque le code PIN QR lors de la sérialisation (repr, str, JSON). :param value: Valeur du champ ``qr_pin``. :return: ``"**********"`` si la valeur est définie, ``None`` sinon. :rtype: str | None """ if value is None: return None return "**********" @field_serializer("account_pin") def _serialize_account_pin(self, value: SecretStr | None) -> str | None: """Masque le PIN du compte lors de la sérialisation. :param value: Valeur du PIN de second facteur du compte. :return: ``"**********"`` si la valeur est définie, ``None`` sinon. :rtype: str | None """ if value is None: return None return "**********" class CalDAVSettings(BaseSettings): """Paramètres d'accès au serveur CalDAV de destination. Les variables d'environnement correspondantes sont préfixées par ``CALDAV_``. L'URL est traitée comme potentiellement sensible (au même titre que ``PRONOTE_ICAL_ENDPOINT__URL``) : elle est de type ``SecretStr`` et masquée lors de la sérialisation. Par défaut, seul HTTPS est accepté ; HTTP n'est toléré que pour un hôte de boucle locale (``localhost``, ``127.0.0.1``, ``::1``) lorsque ``allow_insecure_http`` vaut ``True``. Le nouvel endpoint se configure avec ``CALDAV_ENDPOINT__URL`` ; ``CALDAV_URL`` reste temporairement pris en charge avec un avertissement de dépréciation. """ model_config = SettingsConfigDict( env_file=".env", env_nested_delimiter="__", extra="ignore", env_prefix="CALDAV_", hide_input_in_errors=True, ) allow_insecure_http: bool = False endpoint: ExternalEndpoint | None = None url: SecretStr | None = Field( default=None, exclude=True, deprecated="Utiliser endpoint.url à la place (CALDAV_URL obsolète).", ) username: str | None = None password: SecretStr | None = None calendar_path: str = "/pronote-sync/" @model_validator(mode="before") @classmethod def _migrate_legacy_url(cls, data: object) -> object: """Migre ``url`` vers l'endpoint commun avec un avertissement. L'alias historique ``CALDAV_URL`` conserve ses sémantiques passées pendant la transition : lorsqu'il embarque des identifiants (``user:pass@hôte``), l'endpoint est construit sans revalidation pour ne pas casser une configuration existante, alors que le contrat canonique ``CALDAV_ENDPOINT__URL`` refuse désormais les credentials embarqués. La politique de transport (HTTPS, ou HTTP loopback uniquement avec ``allow_insecure_http``) reste appliquée ensuite. :param data: Données brutes du modèle. :return: Données complétées avec ``endpoint`` si nécessaire. :rtype: object """ if not isinstance(data, dict) or data.get("url") is None: return data migrated_data = data.copy() warnings.warn( "CALDAV_URL est obsolète : utiliser CALDAV_ENDPOINT__URL.", DeprecationWarning, stacklevel=2, ) if migrated_data.get("endpoint") is None: legacy_url = migrated_data["url"] raw_url = ( legacy_url.get_secret_value() if isinstance(legacy_url, SecretStr) else str(legacy_url) ) parsed = urlparse(raw_url) if parsed.username is not None or parsed.password is not None: # Alias obsolète : grandfathered, on préserve l'URL telle quelle. migrated_data["endpoint"] = ExternalEndpoint.model_construct( url=legacy_url if isinstance(legacy_url, SecretStr) else SecretStr(legacy_url) ) else: migrated_data["endpoint"] = {"url": legacy_url} return migrated_data @model_validator(mode="after") def _validate_endpoint_policy(self) -> CalDAVSettings: """Applique la politique HTTPS/HTTP loopback propre à CalDAV. HTTPS est toujours accepté. HTTP n'est accepté que pour un hôte de boucle locale (``localhost``, ``127.0.0.1``, ``::1``) et uniquement lorsque ``allow_insecure_http`` vaut ``True``. Les autres schémas du socle commun sont refusés pour ce connecteur. :return: Instance validée inchangée. :rtype: CalDAVSettings :raises ValueError: Si le schéma n'est pas supporté ou si l'URL HTTP n'est pas autorisée. """ if self.endpoint is None: return self raw_url = self.endpoint.url.get_secret_value() parsed = urlparse(raw_url) if parsed.scheme not in ("http", "https"): raise ValueError("URL CalDAV invalide : schéma non supporté") from None if parsed.scheme == "https": return self # HTTP — check allow_insecure_http flag and loopback if not self.allow_insecure_http: raise ValueError( "URL CalDAV non sécurisée : HTTPS requis (ou activer " "CALDAV_ALLOW_INSECURE_HTTP pour localhost)" ) from None hostname = parsed.hostname or "" if hostname not in _LOOPBACK_HOSTS: raise ValueError( "URL CalDAV non sécurisée : HTTP autorisé uniquement pour localhost" ) from None return self _XMPP_LOOPBACK_HOSTS: frozenset[str] = frozenset({"localhost", "127.0.0.1", "::1"}) class XmppSettings(BaseSettings): """Paramètres du canal de notifications XMPP (désactivé par défaut). Tous les champs ont des valeurs par défaut afin que le canal XMPP reste inactif tant qu'il n'est pas explicitement activé. Les variables d'environnement correspondantes sont préfixées par ``XMPP_``. Contraintes de champs : ``port`` est borné entre 1 et 65535 et ``timeout`` doit être strictement positif. Politique TLS : le mode ``tls_mode`` détermine la négociation TLS (``direct``, ``starttls`` ou ``disabled``). Le mode ``disabled`` n'est autorisé que sur un hôte de boucle locale (``localhost``, ``127.0.0.1``, ``::1``) ; ``starttls`` et ``direct`` sont permis pour tous les hôtes. Compatibilité : le champ historique ``use_tls`` (booléen) est un alias obsolète ; ``use_tls=True`` mappe vers ``tls_mode="direct"`` et ``use_tls=False`` vers ``tls_mode="starttls"``, avec un :pyexc:`DeprecationWarning`. La valeur brute fournie reste lisible via ``use_tls`` (``None`` si non fournie). """ model_config = SettingsConfigDict( env_file=".env", extra="ignore", env_prefix="XMPP_", hide_input_in_errors=True, ) enabled: bool = False jid: str | None = None password: SecretStr | None = None host: str = "" port: int = Field(default=5222, ge=1, le=65535) to: str | None = None resource: str = "pronote-sync" tls_mode: Literal["direct", "starttls", "disabled"] = "starttls" use_tls: bool | None = Field( default=None, deprecated="Utiliser tls_mode à la place (XMPP_USE_TLS obsolète).", ) timeout: float = Field(default=30, gt=0) connect_timeout: float = Field(default=15, gt=0) cleanup_timeout: float = Field(default=10, gt=0) @model_validator(mode="before") @classmethod def _migrate_use_tls(cls, data: object) -> object: """Mappe l'alias obsolète ``use_tls`` vers le mode canonique ``tls_mode``. ``use_tls=True`` devient ``tls_mode="direct"`` et ``use_tls=False`` devient ``tls_mode="starttls"`` ; un :pyexc:`DeprecationWarning` est émis à chaque usage explicite de l'alias. ``tls_mode`` fourni explicitement prend le pas sur l'alias. :param data: Données d'entrée du modèle (dict ou autre). :return: Données d'entrée avec ``tls_mode`` dérivé de ``use_tls``. :rtype: object """ if not isinstance(data, dict) or "use_tls" not in data: return data warnings.warn( "XMPP_USE_TLS est obsolète : utiliser XMPP_TLS_MODE " "('direct', 'starttls' ou 'disabled').", DeprecationWarning, stacklevel=2, ) if data.get("tls_mode") is None: data["tls_mode"] = "direct" if data["use_tls"] else "starttls" return data @model_validator(mode="after") def _validate_tls_policy(self) -> XmppSettings: """Refuse le mode ``disabled`` hors des hôtes de boucle locale. La règle s'applique quel que soit l'état du champ ``enabled``. Les modes ``starttls`` et ``direct`` sont autorisés pour tous les hôtes. Le message d'erreur ne contient aucune valeur sensible (``jid``, ``password``, ``to``). :return: L'instance validée inchangée. :rtype: XmppSettings :raises ValueError: Si ``tls_mode`` est ``disabled`` et que ``host`` n'est pas un hôte de boucle locale. """ if self.tls_mode == "disabled" and self.host not in _XMPP_LOOPBACK_HOSTS: raise ValueError( "TLS désactivé n'est autorisé que sur les hôtes de loopback " "(localhost, 127.0.0.1, ::1)." ) from None return self class AISettings(BaseSettings): """Paramètres de la synthèse par IA (désactivée par défaut). Les variables d'environnement correspondantes sont préfixées par ``AI_``. Le provider ``openai-compatible`` permet d'utiliser n'importe quelle API compatible OpenAI via ``AI_BASE_URL`` ; les URLs en HTTP ne sont alors acceptées que si ``AI_ALLOW_INSECURE_HTTP`` vaut ``true``. """ model_config = SettingsConfigDict(env_file=".env", extra="ignore", env_prefix="AI_") enabled: bool = False provider: Literal["openai", "litellm", "openai-compatible"] = "openai" base_url: str | None = None api_key: SecretStr | None = None allow_insecure_http: bool = False model: str | None = None class BlogSettings(BaseSettings): """Paramètres de la source RSS du blog du collège (désactivée par défaut). Les variables d'environnement correspondantes sont préfixées par ``BLOG_``. """ model_config = SettingsConfigDict( env_file=".env", env_nested_delimiter="__", extra="ignore", env_prefix="BLOG_", hide_input_in_errors=True, ) enabled: bool = False endpoint: ExternalEndpoint = Field( default_factory=lambda: ExternalEndpoint( url=SecretStr("https://blogpeda.ac-bordeaux.fr/cjeliote/?feed=rss2") ) ) rss_url: str | None = Field( default=None, exclude=True, deprecated="Utiliser endpoint.url à la place (BLOG_RSS_URL obsolète).", ) @model_validator(mode="before") @classmethod def _migrate_legacy_rss_url(cls, data: object) -> object: """Migre ``rss_url`` vers l'endpoint commun avec un avertissement. :param data: Données brutes du modèle. :return: Données complétées avec ``endpoint`` si nécessaire. :rtype: object """ if not isinstance(data, dict) or data.get("rss_url") is None: return data migrated_data = data.copy() warnings.warn( "BLOG_RSS_URL est obsolète : utiliser BLOG_ENDPOINT__URL.", DeprecationWarning, stacklevel=2, ) if migrated_data.get("endpoint") is None: migrated_data["endpoint"] = {"url": migrated_data["rss_url"]} return migrated_data @model_validator(mode="after") def _validate_endpoint_policy(self) -> BlogSettings: """Refuse les transports non sûrs pour le flux RSS. Seul HTTPS est accepté : aucun fichier local n'est lu depuis un endpoint externe. :return: Instance validée inchangée. :rtype: BlogSettings :raises ValueError: Si le schéma n'est pas ``https``. """ scheme = urlparse(self.endpoint.url.get_secret_value()).scheme if scheme != "https": raise ValueError("URL RSS invalide : HTTPS requis") from None return self class AppSettings(BaseSettings): """Paramètres généraux de l'application, sans préfixe d'environnement. Contient notamment la fenêtre de synchronisation en jours (``SYNC_PAST_DAYS`` / ``SYNC_FUTURE_DAYS``) et la configuration de l'agenda théorique (``THEORETICAL_AGENDA_PATH``, ``THEORETICAL_WEEK_ANCHOR_DATE``, ``THEORETICAL_WEEK_ANCHOR_TYPE`` ainsi que ``SCHOOL_HOLIDAYS_PATH`` pour les vacances scolaires). """ model_config = SettingsConfigDict(env_file=".env", extra="ignore") dry_run: bool = False log_level: str = "INFO" theoretical_agenda_path: str | None = None school_holidays_path: str | None = None theoretical_week_anchor_date: date | None = None theoretical_week_anchor_type: Literal["even", "odd"] | None = None sync_past_days: int = Field(default=7, ge=0) sync_future_days: int = Field(default=30, ge=0) class Settings(BaseSettings): """Configuration racine du pipeline ``pronote-sync``. Agrège les sous-groupes de configuration : Pronote, CalDAV, XMPP, IA, blog et paramètres généraux de l'application. """ model_config = SettingsConfigDict(env_file=".env", extra="ignore") pronote: PronoteSettings = Field(default_factory=PronoteSettings) caldav: CalDAVSettings = Field(default_factory=CalDAVSettings) xmpp: XmppSettings = Field(default_factory=XmppSettings) ai: AISettings = Field(default_factory=AISettings) blog: BlogSettings = Field(default_factory=BlogSettings) app: AppSettings = Field(default_factory=AppSettings) def redaction_secrets(self) -> tuple[SecretStr, ...]: """Énumère tous les secrets configurés pour la rédaction. Collecte les valeurs :class:`pydantic.SecretStr` non vides présentes dans les sous-configurations (URL iCal, mots de passe, code PIN QR et clé API IA). Les valeurs vides ou ``None`` sont filtrées ; les doublons sont supprimés. :return: Tuple de secrets à masquer dans les messages d'erreur. :rtype: tuple[SecretStr, ...] """ secrets = [ self.pronote.endpoint.url if self.pronote.endpoint is not None else None, self.pronote.ical_endpoint.url if self.pronote.ical_endpoint is not None else None, self.pronote.password, self.pronote.qr_pin, self.pronote.account_pin, self.caldav.endpoint.url if self.caldav.endpoint is not None else None, self.caldav.password, self.xmpp.password, self.ai.api_key, self.blog.endpoint.url, ] return tuple(dict.fromkeys(secret for secret in secrets if secret is not None))