528 lines
20 KiB
Python
528 lines
20 KiB
Python
"""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({"file", "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 d'une URL d'endpoint externe.
|
|
|
|
: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 ou utilise un schéma inconnu.
|
|
"""
|
|
is_valid = False
|
|
try:
|
|
parsed = urlparse(value.get_secret_value())
|
|
_ = parsed.port
|
|
is_valid = (
|
|
parsed.scheme in _EXTERNAL_ENDPOINT_SCHEMES
|
|
and (parsed.scheme not in {"http", "https"} or parsed.hostname is not None)
|
|
and (parsed.scheme != "file" or bool(parsed.path))
|
|
)
|
|
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 les transports ``https``, ``http`` et ``file``. 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_",
|
|
)
|
|
|
|
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 utilise HTTPS. Le flux iCal accepte également ``file``
|
|
afin de préserver les fixtures locales injectées.
|
|
|
|
:return: Instance validée inchangée.
|
|
:rtype: PronoteSettings
|
|
:raises ValueError: Si un endpoint utilise un schéma interdit.
|
|
"""
|
|
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 not in {"https", "file"}:
|
|
raise ValueError("URL iCal Pronote invalide : HTTPS ou file 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_",
|
|
)
|
|
|
|
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.
|
|
|
|
: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:
|
|
migrated_data["endpoint"] = {"url": migrated_data["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_",
|
|
)
|
|
|
|
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 de production.
|
|
|
|
Le transport ``file`` reste autorisé pour les fixtures locales.
|
|
|
|
:return: Instance validée inchangée.
|
|
:rtype: BlogSettings
|
|
:raises ValueError: Si le schéma n'est ni ``https`` ni ``file``.
|
|
"""
|
|
scheme = urlparse(self.endpoint.url.get_secret_value()).scheme
|
|
if scheme not in {"https", "file"}:
|
|
raise ValueError("URL RSS invalide : HTTPS ou file 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))
|