Correct 4 findings from the independent M11 review: #1 (Critical) — PipelineCriticalError was downgraded to PipelineWarning: - Add except PipelineCriticalError: raise before each except Exception in all 5 non-blocking steps (fetch_blog, compare, caldav_sync, synthesis, send) - Critical errors now propagate to the outer handler and stop the pipeline #2 (Critical) — redact_exception() did not use configured secrets: - Extend redact_exception() with extra_secrets parameter (upward compatible) - Harden redact_secrets(): sort extra_secrets by length descending - Add Settings.redaction_secrets() collecting all 6 SecretStr fields - Add PipelineRunner._redact(exc) using self._redaction_secrets - All except blocks in run() now use self._redact(exc) - CalDAV FAILED-status path uses full redaction_secrets collection #3 (Medium) — BlogRSSClient silently swallowed failures: - Add error field to BlogRSSFetchResult - rss.py sets error on failure paths (except Exception, bozo/invalid feed) - fetch_blog_step raises RuntimeError when result.error is set - PipelineRunner now produces PipelineWarning for blog failures #4 (Medium) — Test coverage at 80%, now 91%: - 11 new integration tests covering blog failure/success, compare failure, CalDAV failure (exception + FAILED status), send False/exception, PipelineCriticalError propagation, secret redaction with sentinel, empty agenda/homework, iCal cache cleanup - Secret redaction test uses mock (no network) and proves configured-secret propagation via non-URL sentinel in RuntimeError Validation: 619 tests pass, ruff/mypy/bandit/pre-commit green, coverage 91%. Co-authored-by: opencode/coder <coder@agents.invalid> Co-authored-by: opencode/test-engineer <test-engineer@agents.invalid>
284 lines
11 KiB
Python
284 lines
11 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
|
|
|
|
from datetime import date
|
|
from typing import Literal
|
|
from urllib.parse import urlparse
|
|
|
|
from pydantic import (
|
|
Field,
|
|
SecretStr,
|
|
ValidationInfo,
|
|
field_serializer,
|
|
field_validator,
|
|
)
|
|
from pydantic_settings import BaseSettings, SettingsConfigDict
|
|
|
|
from pronote_sync.utils.redaction import redact_url
|
|
|
|
|
|
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", extra="ignore", env_prefix="PRONOTE_")
|
|
|
|
ical_url: SecretStr | None = None
|
|
username: str | None = None
|
|
password: SecretStr | None = None
|
|
ent: str | None = None
|
|
pronote_url: str | None = None
|
|
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"
|
|
|
|
@field_serializer("ical_url")
|
|
def _serialize_ical_url(self, value: SecretStr | None) -> str | None:
|
|
"""Masque l'URL iCal lors de la sérialisation (repr, str, JSON).
|
|
|
|
:param value: Valeur du champ ``ical_url``.
|
|
: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_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``.
|
|
"""
|
|
|
|
model_config = SettingsConfigDict(env_file=".env", extra="ignore", env_prefix="CALDAV_")
|
|
|
|
allow_insecure_http: bool = False
|
|
url: SecretStr | None = None
|
|
username: str | None = None
|
|
password: SecretStr | None = None
|
|
calendar_path: str = "/pronote-sync/"
|
|
|
|
@field_serializer("url")
|
|
def _serialize_url(self, value: SecretStr | None) -> str | None:
|
|
"""Masque l'URL CalDAV lors de la sérialisation (repr, str, JSON).
|
|
|
|
:param value: Valeur du champ ``url`` (secret potentiel).
|
|
:return: URL avec les éléments sensibles remplacés par ``REDACTED``,
|
|
ou ``None`` si la valeur est absente.
|
|
:rtype: str | None
|
|
"""
|
|
if value is None:
|
|
return None
|
|
return redact_url(value.get_secret_value())
|
|
|
|
@field_validator("url")
|
|
@classmethod
|
|
def _validate_url_https(cls, v: SecretStr | None, info: ValidationInfo) -> SecretStr | None:
|
|
"""Valide le schéma de l'URL CalDAV (HTTPS obligatoire par défaut).
|
|
|
|
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 messages d'erreur
|
|
ne contiennent jamais l'URL brute (susceptible de contenir des
|
|
identifiants).
|
|
|
|
:param v: Valeur du champ ``url`` à valider.
|
|
:param info: Contexte de validation (accès aux autres champs).
|
|
:return: La valeur validée inchangée.
|
|
:rtype: SecretStr | None
|
|
:raises ValueError: Si le schéma n'est pas supporté ou si l'URL HTTP
|
|
n'est pas autorisée.
|
|
"""
|
|
if v is None:
|
|
return v
|
|
raw_url = v.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 v
|
|
# HTTP — check allow_insecure_http flag and loopback
|
|
allow_insecure = info.data.get("allow_insecure_http", False)
|
|
if not allow_insecure:
|
|
raise ValueError(
|
|
"URL CalDAV non sécurisée : HTTPS requis (ou activer "
|
|
"CALDAV_ALLOW_INSECURE_HTTP pour localhost)"
|
|
) from None
|
|
hostname = parsed.hostname or ""
|
|
loopback_hosts = {"localhost", "127.0.0.1", "::1"}
|
|
if hostname not in loopback_hosts:
|
|
raise ValueError(
|
|
"URL CalDAV non sécurisée : HTTP autorisé uniquement pour localhost"
|
|
) from None
|
|
return v
|
|
|
|
|
|
_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 : la désactivation de TLS (``use_tls`` à ``False``) n'est
|
|
autorisée que sur un hôte de boucle locale (``localhost``, ``127.0.0.1``,
|
|
``::1``). Dans tout autre cas, une erreur de validation est levée,
|
|
indépendamment de l'état du champ ``enabled``.
|
|
"""
|
|
|
|
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"
|
|
use_tls: bool = True
|
|
timeout: int = Field(default=30, gt=0)
|
|
|
|
@field_validator("use_tls")
|
|
@classmethod
|
|
def _validate_tls_policy(cls, v: bool, info: ValidationInfo) -> bool:
|
|
"""Refuse la désactivation de TLS hors des hôtes de boucle locale.
|
|
|
|
La règle s'applique quel que soit l'état du champ ``enabled``. Le
|
|
message d'erreur ne contient aucune valeur sensible (``jid``,
|
|
``password``, ``to``).
|
|
|
|
:param v: Valeur du champ ``use_tls`` à valider.
|
|
:param info: Contexte de validation (accès aux autres champs).
|
|
:return: La valeur validée inchangée.
|
|
:rtype: bool
|
|
:raises ValueError: Si ``use_tls`` est ``False`` et que ``host``
|
|
n'est pas un hôte de boucle locale.
|
|
"""
|
|
if v is False:
|
|
host = info.data.get("host", "")
|
|
if 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 v
|
|
|
|
|
|
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", extra="ignore", env_prefix="BLOG_")
|
|
|
|
enabled: bool = False
|
|
rss_url: str = "https://blogpeda.ac-bordeaux.fr/cjeliote/?feed=rss2"
|
|
|
|
|
|
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 = 7
|
|
sync_future_days: int = 30
|
|
|
|
|
|
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 (Pronote, CalDAV, XMPP, 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.ical_url,
|
|
self.pronote.password,
|
|
self.caldav.url,
|
|
self.caldav.password,
|
|
self.xmpp.password,
|
|
self.ai.api_key,
|
|
]
|
|
return tuple(dict.fromkeys(secret for secret in secrets if secret is not None))
|