From 03638986697e46738c924ce860299e2080ba40ed Mon Sep 17 00:00:00 2001 From: Antoine Van Elstraete Date: Tue, 8 Sep 2026 23:15:06 +0200 Subject: [PATCH] feat: authentification QR code / token pour Pronote MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Ajoute le mode d'authentification PRONOTE_AUTH_MODE=qr_token comme alternative au mode password pour les instances Pronote utilisant HubEduConnect/EduConnect où l'authentification par mot de passe échoue (CAPTCHA, MFA, flux SAML). Nouveaux éléments : - PronoteSettings : auth_mode, qr_code_file, qr_pin (SecretStr) - PronoteAuthState : persistance du token rotatif dans .pronote_auth_state.json (écriture atomique, permissions 0600, symlink-safe via O_EXCL|O_NOFOLLOW) - PronoteClient._connect_qr_token() : token_login avec creds persistés, qrcode_login pour l'enrôlement initial, export_credentials persisté après chaque login réussi - PronoteAuthRotationError : levée en cas d'échec de rotation du token, propagée sans wrapping à travers PronoteFetcher et fetch_step jusqu'à PipelineRunner.run() qui notifie via XMPP (si canal disponible et dry_run inactif) - _is_pronotepy_configured() mode-aware : qr_token ne requiert que PRONOTE_URL - _collect_auth_secrets() : redaction des secrets explicites (token, PIN, jeton QR) dans tous les logs du chemin d'authentification Documentation : - .env.example : PRONOTE_AUTH_MODE, PRONOTE_QR_CODE_FILE, PRONOTE_QR_PIN - AGENTS.md : contrat d'authentification QR code / token - Wiki GuidePronote : section enrôlement, exécutions suivantes, ré-enrôlement Tests (686 passés, couverture 94.87%) : - 5 tests config QR, 9 tests auth_state, 10 tests client QR, 3 tests propagation, 4 tests intégration rotation end-to-end, 4 tests fallback mode-aware - Tests de non-fuite : sentinelles distinctes pour token, PIN, jeton QR Co-authored-by: coder/litellm/coder --- .env.example | 13 + .gitignore | 2 + AGENTS.md | 25 + pronote_sync/config/settings.py | 21 +- pronote_sync/errors.py | 17 + pronote_sync/pipeline/run.py | 45 +- pronote_sync/pipeline/steps/fetch.py | 6 +- pronote_sync/sources/pronote/auth_state.py | 195 +++++++ pronote_sync/sources/pronote/client.py | 270 +++++++++- pronote_sync/sources/pronote/fallback.py | 26 +- tests/integration/test_pipeline_runner.py | 393 +++++++++++++- tests/unit/test_config.py | 76 +++ tests/unit/test_fallback.py | 104 ++++ tests/unit/test_pronote_auth_state.py | 208 ++++++++ tests/unit/test_pronote_client.py | 586 +++++++++++++++++++++ tests/unit/test_rotation_propagation.py | 180 +++++++ 16 files changed, 2128 insertions(+), 39 deletions(-) create mode 100644 pronote_sync/sources/pronote/auth_state.py create mode 100644 tests/unit/test_pronote_auth_state.py create mode 100644 tests/unit/test_rotation_propagation.py diff --git a/.env.example b/.env.example index 12e19ed..6c35742 100644 --- a/.env.example +++ b/.env.example @@ -11,6 +11,19 @@ PRONOTE_AGENDA_SOURCE=auto PRONOTE_HOMEWORK_SOURCE=auto PRONOTE_MESSAGES_SOURCE=pronotepy +# Mode d'authentification Pronote +# "password" (défaut) : authentification classique URL + identifiant + mot de passe +# "qr_token" : authentification par QR code puis token persistant +PRONOTE_AUTH_MODE=password + +# Fichier JSON du QR code Pronote (enrôlement initial, mode qr_token uniquement) +# À générer depuis l'application mobile Pronote. Le QR code expire ~10 minutes. +# PRONOTE_QR_CODE_FILE=/path/to/qr_code.json + +# PIN à 4 chiffres pour l'enrôlement QR code (mode qr_token uniquement) +# SENSIBLE : ne jamais committer cette valeur +# PRONOTE_QR_PIN=1234 + # --- CalDAV --- CALDAV_URL=https://caldav.example.com/calendars/user/pronote/ CALDAV_USERNAME=user@example.com diff --git a/.gitignore b/.gitignore index dd0d1cb..cab0981 100644 --- a/.gitignore +++ b/.gitignore @@ -48,6 +48,8 @@ Thumbs.db # --- Project-specific state files --- .blog_rss_state.json .caldav_sync_state.json +# État d'authentification pronotepy (QR code / token rotation) +.pronote_auth_state.json *.state.json # --- Local scratch / WIP files --- diff --git a/AGENTS.md b/AGENTS.md index 757fd87..aae347a 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -146,6 +146,31 @@ pronote-sync --dry-run - Réutiliser un téléchargement/parsing iCal pour l'agenda et les devoirs pendant un même run, sans cache global ni persistant. +### Contrat d'authentification QR code / token + +- Le mode d'authentification est sélectionné par `PRONOTE_AUTH_MODE` : + - `password` (défaut) : authentification classique via URL, identifiant, mot de passe et ENT. + - `qr_token` : authentification par QR code puis token persistant (pour les instances Pronote + utilisant HubEduConnect/EduConnect où l'authentification par mot de passe échoue). +- En mode `qr_token`, le premier login utilise `pronotepy.qrcode_login(qr_code, pin, uuid)` avec + les paramètres `PRONOTE_QR_CODE_FILE` (chemin du JSON QR) et `PRONOTE_QR_PIN` (PIN SecretStr). +- Après chaque login réussi, les credentials exportées par `pronotepy.export_credentials()` sont + persistées dans `.pronote_auth_state.json` (permissions `0600`, format JSON versionné, écriture + atomique). Le token rotate à chaque session — le fichier doit être mis à jour après chaque run. +- Les logins suivants utilisent `pronotepy.token_login(**credentials)` avec le token persisté. +- En cas d'échec de `token_login` (token expiré/invalide), une `PronoteAuthRotationError` est levée. + Cette erreur se propage sans wrapping à travers `PronoteFetcher` et `fetch_step` jusqu'à + `PipelineRunner.run()`, qui : + - journalise l'erreur (expurgée) ; + - envoie une notification XMPP actionnable si le canal est disponible et `dry_run` est inactif ; + - retourne un résultat dégradé `(None, errors)`. +- `PronoteAuthRotationError` est re-levée telle quelle (`except PronoteAuthRotationError: raise`) + dans toutes les couches d'enveloppement du chemin critique (fetch_agenda, fetch_homework, + fetch_step). Ne pas l'attraper avec `except Exception` sans la re-léver d'abord. +- Le fichier `.pronote_auth_state.json` ne doit jamais être committé (couvert par `.gitignore`). + Son contenu (token vivant) ne doit jamais apparaître dans les logs, les messages d'erreur ou + les notifications XMPP. + ### Contrat du provider `openai-compatible` - Le provider `openai-compatible` réutilise `OpenAISynthesisProvider` avec un `base_url` personnalisé ; aucun nouveau provider n'est créé. - `AI_BASE_URL` et `AI_MODEL` sont requis ; `AI_API_KEY` est requis (MVP). diff --git a/pronote_sync/config/settings.py b/pronote_sync/config/settings.py index aa39a64..cde7b24 100644 --- a/pronote_sync/config/settings.py +++ b/pronote_sync/config/settings.py @@ -42,6 +42,9 @@ class PronoteSettings(BaseSettings): 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 @field_serializer("ical_url") def _serialize_ical_url(self, value: SecretStr | None) -> str | None: @@ -55,6 +58,18 @@ class PronoteSettings(BaseSettings): return None return "**********" + @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 "**********" + class CalDAVSettings(BaseSettings): """Paramètres d'accès au serveur CalDAV de destination. @@ -266,8 +281,9 @@ class Settings(BaseSettings): """É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. + 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, ...] @@ -275,6 +291,7 @@ class Settings(BaseSettings): secrets = [ self.pronote.ical_url, self.pronote.password, + self.pronote.qr_pin, self.caldav.url, self.caldav.password, self.xmpp.password, diff --git a/pronote_sync/errors.py b/pronote_sync/errors.py index ee9e166..0fe1e74 100644 --- a/pronote_sync/errors.py +++ b/pronote_sync/errors.py @@ -21,6 +21,23 @@ class PronoteSyncError(Exception): self.message = message +class PronoteAuthRotationError(PronoteSyncError): + """Erreur de rotation du token d'authentification pronotepy (QR code / token). + + Levée quand le token persisté est invalide ou expiré et qu'un ré-enrôlement + manuel (suppression du fichier d'état + nouveau QR code) est nécessaire. + + :ivar message: Message décrivant l'action à effectuer, sans secret. + """ + + def __init__(self, message: str) -> None: + """Initialise l'erreur de rotation. + + :param message: Message actionnable sans secret (PIN, token, URL). + """ + super().__init__(message) + + class ErrorSeverity(StrEnum): """Niveau de gravité d'une erreur produite par le pipeline.""" diff --git a/pronote_sync/pipeline/run.py b/pronote_sync/pipeline/run.py index 6f4f228..86bba04 100644 --- a/pronote_sync/pipeline/run.py +++ b/pronote_sync/pipeline/run.py @@ -11,7 +11,12 @@ from typing import Protocol, runtime_checkable from pronote_sync.channels import get_channel from pronote_sync.channels.protocol import Channel from pronote_sync.config.settings import Settings -from pronote_sync.errors import PipelineCriticalError, PipelineError, PipelineWarning +from pronote_sync.errors import ( + PipelineCriticalError, + PipelineError, + PipelineWarning, + PronoteAuthRotationError, +) from pronote_sync.models.blog import ExternalInfo from pronote_sync.models.pronote import PronoteData from pronote_sync.models.sync import CalDAVSyncResult, CalDAVSyncStatus @@ -26,6 +31,7 @@ from pronote_sync.pipeline.steps.send import send_step from pronote_sync.pipeline.steps.synthesis import synthesis_step from pronote_sync.sources.blog.rss import BlogRSSClient from pronote_sync.sources.blog.state import BlogRSSState +from pronote_sync.sources.pronote.auth_state import PronoteAuthState from pronote_sync.sources.pronote.client import PronoteClient from pronote_sync.sources.pronote.fallback import PronoteFetcher, PronoteFetcherProtocol from pronote_sync.sources.theoretical import get_theoretical_provider @@ -133,7 +139,15 @@ class PipelineRunner: blog_state = BlogRSSState() if settings.blog.enabled else None return cls( settings=settings, - pronote_fetcher=PronoteFetcher(settings, PronoteClient(settings.pronote)), + pronote_fetcher=PronoteFetcher( + settings, + PronoteClient( + settings.pronote, + auth_state=( + PronoteAuthState() if settings.pronote.auth_mode == "qr_token" else None + ), + ), + ), agenda_comparator=comparator, synthesis_provider=get_synthesis_provider(settings.ai), channel=get_channel(settings.xmpp, dry_run=effective_dry_run), @@ -190,6 +204,11 @@ class PipelineRunner: des étapes facultatives sont converties en :class:`PipelineWarning` afin que les étapes suivantes, notamment XMPP, restent exécutées. + Une :class:`PronoteAuthRotationError` interrompt également l'exécution : + l'erreur est journalisée expurgée, une notification XMPP actionnable est + envoyée (sauf en dry-run ou sans canal), puis un résultat dégradé est + retourné. + :return: Données Pronote normalisées ou ``None``, puis erreurs et avertissements. :rtype: tuple[PronoteData | None, list[PipelineError]] """ @@ -271,6 +290,28 @@ class PipelineRunner: except Exception as exc: self._warn("send", self._redact(exc)) return data, [*self._errors, *self._warnings] + except PronoteAuthRotationError as exc: + error = PipelineCriticalError(self._redact(exc), step="pronote") + logger.error("Erreur critique du pipeline : %s", error.message) + if self._channel is not None and not self._dry_run: + message = XmppMessage( + target_date=now.date(), + synthesis=( + "⚠️ Rotation du token Pronote échouée. Le token d'authentification est " + "expiré ou invalide. Action requise : supprimez le fichier " + ".pronote_auth_state.json et relancez le pipeline avec un nouveau QR " + "code (PRONOTE_QR_CODE_FILE + PRONOTE_QR_PIN)." + ), + external_info=None, + ) + try: + if not send_step(self._channel, message): + self._warn("send", "Le canal XMPP a refusé l'envoi") + except Exception as send_exc: + # L'envoi de la notification est un dernier avertissement : son échec + # ne doit pas masquer l'erreur de rotation, déjà critique. + self._warn("send", self._redact(send_exc)) + self._errors.append(error) except PipelineCriticalError as exc: logger.error("Erreur critique du pipeline : %s", exc.message) self._errors.append(exc) diff --git a/pronote_sync/pipeline/steps/fetch.py b/pronote_sync/pipeline/steps/fetch.py index 842bd52..8c0f270 100644 --- a/pronote_sync/pipeline/steps/fetch.py +++ b/pronote_sync/pipeline/steps/fetch.py @@ -5,7 +5,7 @@ from __future__ import annotations from dataclasses import dataclass from datetime import date -from pronote_sync.errors import PipelineCriticalError, PipelineWarning +from pronote_sync.errors import PipelineCriticalError, PipelineWarning, PronoteAuthRotationError from pronote_sync.models.agenda import Lesson, SchoolEvent from pronote_sync.models.homework import Homework from pronote_sync.models.message import Message @@ -101,6 +101,8 @@ def fetch_step( :return: Données récupérées et avertissements non critiques. :rtype: tuple[FetchedPronoteData, list[PipelineWarning]] :raises PipelineCriticalError: Si l'agenda ou les devoirs ne sont pas disponibles. + :raises PronoteAuthRotationError: Si une rotation du token d'authentification + pronotepy est nécessaire : propagée telle quelle jusqu'au pipeline. """ try: lessons, school_events = fetcher.fetch_agenda() @@ -108,6 +110,8 @@ def fetch_step( homeworks = fetcher.fetch_homework(target_date) except PipelineCriticalError: raise + except PronoteAuthRotationError: + raise except Exception as exc: raise PipelineCriticalError( f"Récupération Pronote impossible : {redact_exception(exc)}", step="fetch" diff --git a/pronote_sync/sources/pronote/auth_state.py b/pronote_sync/sources/pronote/auth_state.py new file mode 100644 index 0000000..e70d9df --- /dev/null +++ b/pronote_sync/sources/pronote/auth_state.py @@ -0,0 +1,195 @@ +"""Persistance des credentials d'authentification par token pronotepy. + +Ce module fournit :class:`PronoteAuthState`, qui stocke et charge les credentials +d'authentification par QR code / token entre les exécutions du pipeline. Le token +pronotepy rotate à chaque session : le fichier d'état doit être mis à jour après +chaque login réussi via :meth:`PronoteAuthState.save`. + +Le fichier d'état est créé avec des permissions ``0600`` car il contient un token +d'authentification vivant. Son contenu n'est jamais journalisé. +""" + +from __future__ import annotations + +import json +import logging +import os +from pathlib import Path +from typing import Any + +from pronote_sync.errors import PronoteSyncError +from pronote_sync.utils.redaction import redact_exception, redact_secrets + +logger = logging.getLogger(__name__) + +_STATE_VERSION = 1 + + +class PronoteAuthState: + """Persiste les credentials d'authentification par token pronotepy entre + les exécutions du pipeline. + + Le fichier d'état contient un dict au format : + {"version": 1, "credentials": {"pronote_url": "...", "username": "...", "password": "", "uuid": "..."}} + + Les credentials sont le retour de pronotepy.Client.export_credentials(), utilisé tel quel + pour token_login(**credentials). Le token rotate à chaque session — le fichier doit être + mis à jour après chaque login réussi. + + :param state_file: Chemin du fichier d'état JSON (``str`` ou + :class:`~pathlib.Path`). ``".pronote_auth_state.json"`` par défaut. + """ + + def __init__(self, state_file: Path | str = ".pronote_auth_state.json") -> None: + """Initialise le gestionnaire d'état d'authentification Pronote. + + Le fichier d'état n'est pas créé à l'initialisation : il n'est écrit + qu'à la première sauvegarde réussie via :meth:`save`. + + :param state_file: Chemin du fichier d'état JSON (``str`` ou + :class:`~pathlib.Path`). ``".pronote_auth_state.json"`` par défaut. + """ + self._state_file = Path(state_file) + + def load(self) -> dict[str, str] | None: + """Charge les credentials d'authentification depuis le fichier d'état. + + Un fichier absent renvoie ``None`` (journalisé en debug). Un fichier + corrompu, une version absente ou non supportée, ou un champ + ``credentials`` invalide renvoient ``None`` avec un avertissement. + Le contenu des credentials n'est jamais journalisé. + + :return: Dict des credentials (``pronote_url``, ``username``, + ``password``, ``uuid``) prêt pour + ``pronotepy.Client.token_login(**credentials)``, ou ``None`` si + aucun état valide n'est disponible. + :rtype: dict[str, str] | None + """ + if not self._state_file.exists(): + logger.debug( + "Fichier d'état d'authentification Pronote %s absent, aucun token à charger.", + redact_secrets(str(self._state_file)), + ) + return None + try: + data: Any = json.loads(self._state_file.read_text(encoding="utf-8")) + except Exception as exc: + logger.warning( + "Impossible de charger le fichier d'état d'authentification Pronote %s : %s, " + "aucun token chargé.", + redact_secrets(str(self._state_file)), + redact_exception(exc), + ) + return None + if not isinstance(data, dict) or data.get("version") != _STATE_VERSION: + logger.warning( + "Fichier d'état d'authentification Pronote %s : version absente ou non supportée, " + "aucun token chargé.", + redact_secrets(str(self._state_file)), + ) + return None + credentials_data = data.get("credentials") + if not isinstance(credentials_data, dict): + logger.warning( + "Fichier d'état d'authentification Pronote %s : champ credentials absent ou invalide, " + "aucun token chargé.", + redact_secrets(str(self._state_file)), + ) + return None + credentials: dict[str, str] = {} + for key, value in credentials_data.items(): + if not isinstance(key, str) or not isinstance(value, str): + logger.warning( + "Fichier d'état d'authentification Pronote %s : champ credentials invalide, " + "aucun token chargé.", + redact_secrets(str(self._state_file)), + ) + return None + credentials[key] = value + return credentials + + def save(self, credentials: dict[str, str]) -> None: + """Sauvegarde les credentials dans le fichier d'état, de manière atomique. + + Le fichier contient ``{"version": 1, "credentials": ...}``. Le JSON est + d'abord écrit dans un fichier temporaire du même répertoire, créé avec + les permissions ``0600`` (lecture seule pour le propriétaire) dès son + ouverture via :func:`os.open` (avec ``O_EXCL`` et ``O_NOFOLLOW`` pour + résister aux attaques par lien symbolique), puis verrouillé via + :func:`os.fchmod` avant toute écriture ; le fichier temporaire remplace + ensuite atomiquement le fichier d'état via :func:`os.replace`. Un + éventuel fichier temporaire stale d'une exécution interrompue est + supprimé avant l'ouverture. Les credentials ne sont jamais journalisés. + + :param credentials: Dict des credentials pronotepy, tel que retourné + par ``pronotepy.Client.export_credentials()``. + :raises PronoteSyncError: Si l'écriture ou le remplacement du fichier + échoue. + """ + payload: dict[str, Any] = { + "version": _STATE_VERSION, + "credentials": credentials, + } + tmp_file = self._state_file.with_suffix(".tmp") + fd: int | None = None + try: + # Nettoie un éventuel fichier temporaire stale laissé par une exécution interrompue. + if tmp_file.exists(): + try: + tmp_file.unlink() + except OSError: + logger.debug( + "Impossible de supprimer le fichier temporaire stale %s, " + "l'ouverture en O_EXCL échouera.", + redact_secrets(str(tmp_file)), + ) + # O_EXCL empêche de créer par-dessus un fichier existant (attaque par lien + # symbolique) et O_NOFOLLOW refuse de suivre un lien symbolique. + fd = os.open( + str(tmp_file), + os.O_WRONLY | os.O_CREAT | os.O_EXCL | os.O_NOFOLLOW, + 0o600, + ) + # Verrouille les permissions en 0600 avant toute écriture, indépendamment de l'umask. + os.fchmod(fd, 0o600) + with os.fdopen(fd, "w", encoding="utf-8") as handle: + json.dump(payload, handle, indent=2) + os.replace(tmp_file, self._state_file) + except Exception as exc: + logger.error( + "Impossible d'écrire le fichier d'état d'authentification Pronote %s : %s.", + redact_secrets(str(self._state_file)), + redact_exception(exc), + ) + if fd is not None: + try: + os.close(fd) + except OSError: + pass + try: + tmp_file.unlink(missing_ok=True) + except Exception as cleanup_exc: + logger.debug( + "Nettoyage du fichier temporaire d'état d'authentification Pronote échoué : %s", + redact_exception(cleanup_exc), + ) + raise PronoteSyncError( + f"Impossible d'écrire le fichier d'état d'authentification Pronote " + f"{redact_secrets(str(self._state_file))}." + ) from None + + def clear(self) -> None: + """Supprime le fichier d'état d'authentification. + + Si le fichier n'existe pas, la méthode ne fait rien et aucune erreur + n'est levée. + + :raises OSError: Si la suppression du fichier existant échoue. + """ + if not self._state_file.exists(): + return + logger.debug( + "Suppression du fichier d'état d'authentification Pronote %s.", + redact_secrets(str(self._state_file)), + ) + self._state_file.unlink() diff --git a/pronote_sync/sources/pronote/client.py b/pronote_sync/sources/pronote/client.py index 297f5f1..1c2d024 100644 --- a/pronote_sync/sources/pronote/client.py +++ b/pronote_sync/sources/pronote/client.py @@ -10,19 +10,24 @@ des cours et des devoirs se propagent pour déclencher le repli iCal. from __future__ import annotations +import json import logging from datetime import date +from pathlib import Path from typing import Any, Protocol +from uuid import uuid4 import pronotepy import pronotepy.ent as pronotepy_ent import requests from pronote_sync.config.settings import PronoteSettings +from pronote_sync.errors import PronoteAuthRotationError from pronote_sync.models.agenda import Lesson, LessonStatus from pronote_sync.models.homework import Homework from pronote_sync.models.message import Message, MessageType -from pronote_sync.utils.redaction import redact_exception +from pronote_sync.sources.pronote.auth_state import PronoteAuthState +from pronote_sync.utils.redaction import redact_exception, redact_secrets from pronote_sync.utils.uid import generate_deterministic_uid, normalize_pronote_uid logger = logging.getLogger(__name__) @@ -91,6 +96,51 @@ def _resolve_ent(ent_name: str) -> Any: return resolver +def _collect_auth_secrets(client: PronoteClient) -> list[str]: + """Collecte toutes les valeurs sensibles d'authentification pour la redaction. + + Rassemble le mot de passe, le PIN QR, le contenu du fichier QR (jeton, + login, url) et les credentials persistés (token, username) afin de les + transmettre comme ``extra_secrets`` aux fonctions de masquage. Une valeur + vide ou ``None`` est ignorée. + + :param client: Le client Pronote dont on collecte les secrets. + :return: Liste des valeurs sensibles à expurger des logs. + :rtype: list[str] + """ + secrets: list[str] = [] + settings = client._settings + # Mot de passe + if settings.password is not None: + secrets.append(settings.password.get_secret_value()) + # PIN QR + if settings.qr_pin is not None: + secrets.append(settings.qr_pin.get_secret_value()) + # Contenu du fichier QR (jeton, login, url) + if settings.qr_code_file is not None: + try: + qr_path = Path(settings.qr_code_file) + qr_data: Any = json.loads(qr_path.read_text(encoding="utf-8")) + for key in ("jeton", "login", "url"): + val = qr_data.get(key) + if isinstance(val, str): + secrets.append(val) + except Exception as exc: + logger.debug( + "Impossible de lire le fichier QR %s : %s", + redact_secrets(settings.qr_code_file), + redact_exception(exc), + ) + # Credentials persistés (token, username du fichier d'état) + if client._auth_state is not None: + creds = client._auth_state.load() + if creds is not None: + for val in creds.values(): + if isinstance(val, str): + secrets.append(val) + return [s for s in secrets if s] + + class PronoteClientProtocol(Protocol): """Interface du client Pronote consommée par la logique de repli.""" @@ -143,24 +193,56 @@ class PronoteClient: exceptions se propager pour déclencher le repli iCal. """ - def __init__(self, settings: PronoteSettings) -> None: + def __init__( + self, + settings: PronoteSettings, + auth_state: PronoteAuthState | None = None, + ) -> None: """Initialise le client Pronote sans se connecter. - :param settings: Paramètres d'accès à Pronote (username, password, ent). + :param settings: Paramètres d'accès à Pronote (username, password, ent, + mode d'authentification, fichier QR et PIN). + :param auth_state: Gestionnaire de persistance du token + d'authentification (optionnel ; requis en mode ``qr_token`` pour + conserver le token entre les exécutions). """ self._settings: PronoteSettings = settings + self._auth_state: PronoteAuthState | None = auth_state self._client: pronotepy.Client | None = None def _connect(self) -> pronotepy.Client: """Crée et connecte le client ``pronotepy`` (connexion paresseuse). - Le client est créé une seule fois puis réutilisé pour les appels - suivants. Le nom d'ENT, s'il est configuré, est résolu via - :func:`_resolve_ent` ; en l'absence d'ENT, ``ent=None`` est transmis - à ``pronotepy`` pour une connexion directe. Le type de compte - (``student`` ou ``parent``) détermine la classe de client utilisée. - L'erreur de connexion est relancée sans journalisation, la méthode - publique appelante étant responsable de la journaliser. + En mode ``password``, utilise l'authentification classique (URL, + username, password, ENT). En mode ``qr_token``, utilise le token + persisté via :class:`PronoteAuthState`, ou procède à l'enrôlement + initial par QR code si aucun token n'est présent. + + :return: Le client ``pronotepy`` connecté. + :rtype: pronotepy.Client + :raises ValueError: Si les credentials requis sont manquants. + :raises PronoteAuthRotationError: Si le token persisté est invalide + (rotation requise) ou si l'enrôlement QR échoue. + :raises pronotepy.PronoteAPIError: Si la connexion échoue. + """ + if self._client is not None: + return self._client + + if self._settings.auth_mode == "qr_token": + self._client = self._connect_qr_token() + else: + self._client = self._connect_password() + return self._client + + def _connect_password(self) -> pronotepy.Client: + """Connecte le client ``pronotepy`` en mode ``password``. + + Le nom d'ENT, s'il est configuré, est résolu via :func:`_resolve_ent` ; + en l'absence d'ENT, ``ent=None`` est transmis à ``pronotepy`` pour une + connexion directe. Le type de compte (``student`` ou ``parent``) + détermine la classe de client utilisée. L'erreur de connexion est + relancée sans journalisation, la méthode publique appelante étant + responsable de la journaliser. :return: Le client ``pronotepy`` connecté. :rtype: pronotepy.Client @@ -168,27 +250,153 @@ class PronoteClient: est manquant, ou si l'ENT fourni est inconnu. :raises pronotepy.PronoteAPIError: Si la connexion à Pronote échoue. """ - if self._client is None: - url = self._settings.url - username = self._settings.username - password = self._settings.password - ent = self._settings.ent - if url is None or username is None or password is None: - raise ValueError("url, username et password sont requis pour pronotepy") - resolver = _resolve_ent(ent) if ent is not None else None - client_class: type[pronotepy.Client] = ( - pronotepy.ParentClient - if self._settings.account_type == "parent" - else pronotepy.Client - ) - self._client = client_class( - pronote_url=url, - username=username, - password=password.get_secret_value(), - ent=resolver, - ) + url = self._settings.url + username = self._settings.username + password = self._settings.password + ent = self._settings.ent + if url is None or username is None or password is None: + raise ValueError("url, username et password sont requis pour pronotepy") + resolver = _resolve_ent(ent) if ent is not None else None + client_class: type[pronotepy.Client] = ( + pronotepy.ParentClient if self._settings.account_type == "parent" else pronotepy.Client + ) + self._client = client_class( + pronote_url=url, + username=username, + password=password.get_secret_value(), + ent=resolver, + ) return self._client + def _connect_qr_token(self) -> pronotepy.Client: + """Connecte via token persisté ou enrôlement par QR code. + + En premier lieu, les credentials persistés (``pronote_url``, username, + ``password``/token, ``uuid``) sont rejoués via + ``pronotepy.Client.token_login`` si :class:`PronoteAuthState` est + disponible et fournit un état. En cas d'échec du login par token + (exception ou client non connecté), une :class:`PronoteAuthRotationError` + est levée immédiatement, sans repli vers l'enrôlement QR : la rotation + du token doit être déclenchée par l'opérateur. L'enrôlement par QR code + n'est tenté que lorsqu'aucun credential n'est persisté (premier login) ; + le nouveau token est ensuite persisté immédiatement. + + :return: Le client ``pronotepy`` connecté. + :rtype: pronotepy.Client + :raises PronoteAuthRotationError: Si le token persisté est invalide + (expiré ou refusé par Pronote), ou si l'enrôlement QR échoue + (fichier QR ou PIN manquant, fichier QR invalide ou expiré). + """ + client_class: type[pronotepy.Client] = ( + pronotepy.ParentClient if self._settings.account_type == "parent" else pronotepy.Client + ) + + # Login par token avec les credentials persistés + if self._auth_state is not None: + creds = self._auth_state.load() + if creds is not None: + try: + client = client_class.token_login(**creds) + if client.logged_in: + self._auth_state.save(client.export_credentials()) + return client + # logged_in est False — le token est invalide + raise PronoteAuthRotationError( + "Le token d'authentification Pronote est invalide (non connecté). " + "Action requise : supprimez le fichier .pronote_auth_state.json " + "et relancez avec un nouveau QR code." + ) from None + except PronoteAuthRotationError: + raise + except Exception as exc: + logger.error( + "Échec du login par token pronotepy : %s", + redact_exception(exc, extra_secrets=_collect_auth_secrets(self)), + ) + # Token expiré/invalide — pas de repli vers l'enrôlement QR + raise PronoteAuthRotationError( + "Le token d'authentification Pronote est expiré ou invalide. " + "Action requise : supprimez le fichier .pronote_auth_state.json " + "et relancez avec un nouveau QR code (PRONOTE_QR_CODE_FILE + " + "PRONOTE_QR_PIN)." + ) from None + + # Enrôlement : premier login via QR code (aucun credential persisté) + client = self._enroll_qr_code(client_class) + # Persister le token rotaté immédiatement + if self._auth_state is not None: + self._auth_state.save(client.export_credentials()) + return client + + def _enroll_qr_code(self, client_class: type[pronotepy.Client]) -> pronotepy.Client: + """Procède à l'enrôlement initial via QR code pronotepy. + + Le fichier QR JSON doit contenir les clés ``login``, ``jeton`` et + ``url``. Le PIN et le contenu du fichier ne sont jamais journalisés ; + les erreurs propagées sont expurgées. + + :param client_class: Classe de client pronotepy à utiliser. + :return: Le client ``pronotepy`` connecté après enrôlement. + :rtype: pronotepy.Client + :raises PronoteAuthRotationError: Si le fichier QR ou le PIN est + manquant, si le fichier QR est illisible ou incomplet, ou si le + login par QR code échoue (PIN invalide ou QR code expiré). + """ + qr_file = self._settings.qr_code_file + qr_pin = self._settings.qr_pin + + if qr_file is None or qr_pin is None: + raise PronoteAuthRotationError( + "Enrôlement QR requis : PRONOTE_QR_CODE_FILE et PRONOTE_QR_PIN sont " + "nécessaires pour le premier login en mode qr_token. Supprimez le " + "fichier .pronote_auth_state.json si présent et relancez avec un " + "QR code frais." + ) from None + + # Read and validate QR code JSON + try: + qr_path = Path(qr_file) + qr_data: Any = json.loads(qr_path.read_text(encoding="utf-8")) + except Exception as exc: + logger.error( + "Fichier QR invalide %s : %s", + redact_secrets(qr_file, extra_secrets=_collect_auth_secrets(self)), + redact_exception(exc, extra_secrets=_collect_auth_secrets(self)), + ) + raise PronoteAuthRotationError( + "Impossible de lire le fichier QR code : " + f"{redact_secrets(qr_file, extra_secrets=_collect_auth_secrets(self))}" + ) from None + + # Validate required keys + for key in ("login", "jeton", "url"): + if key not in qr_data: + raise PronoteAuthRotationError( + f"Le fichier QR code ne contient pas la clé requise : {key}" + ) from None + + pin_value = qr_pin.get_secret_value() + app_uuid = f"pronote-sync-{uuid4().hex}" + + try: + client = client_class.qrcode_login( + qr_code=qr_data, + pin=pin_value, + uuid=app_uuid, + ) + except Exception as exc: + logger.error( + "Échec de l'enrôlement QR : %s", + redact_exception(exc, extra_secrets=_collect_auth_secrets(self)), + ) + raise PronoteAuthRotationError( + "Échec de l'enrôlement par QR code : PIN invalide ou QR code expiré. " + "Générez un nouveau QR code dans l'application Pronote et mettez à " + "jour PRONOTE_QR_CODE_FILE." + ) from None + + return client + def get_messages(self) -> list[Message]: """Récupère les messages des discussions Pronote. @@ -284,6 +492,8 @@ class PronoteClient: :param end: Date de fin de la fenêtre (incluse). :return: Liste des cours. :rtype: list[Lesson] + :raises PronoteAuthRotationError: Si le token persisté est invalide et + qu'aucun ré-enrôlement n'est possible (fichier QR ou PIN manquant). :raises pronotepy.PronoteAPIError: Si l'API Pronote échoue. :raises ValueError: Si la configuration ou l'ENT est invalide. :raises requests.RequestException: Si une requête réseau échoue. @@ -334,6 +544,8 @@ class PronoteClient: :param end: Date de fin de la fenêtre (incluse). :return: Liste des devoirs. :rtype: list[Homework] + :raises PronoteAuthRotationError: Si le token persisté est invalide et + qu'aucun ré-enrôlement n'est possible (fichier QR ou PIN manquant). :raises pronotepy.PronoteAPIError: Si l'API Pronote échoue. :raises ValueError: Si la configuration ou l'ENT est invalide. :raises requests.RequestException: Si une requête réseau échoue. diff --git a/pronote_sync/sources/pronote/fallback.py b/pronote_sync/sources/pronote/fallback.py index 39ef545..257ff8c 100644 --- a/pronote_sync/sources/pronote/fallback.py +++ b/pronote_sync/sources/pronote/fallback.py @@ -23,7 +23,7 @@ from enum import StrEnum from typing import Literal, Protocol from pronote_sync.config.settings import Settings -from pronote_sync.errors import PipelineCriticalError +from pronote_sync.errors import PipelineCriticalError, PronoteAuthRotationError from pronote_sync.models.agenda import Lesson, SchoolEvent from pronote_sync.models.homework import Homework from pronote_sync.models.message import Message @@ -147,13 +147,19 @@ class PronoteFetcher: return self._settings.pronote.ical_url is not None def _is_pronotepy_configured(self) -> bool: - """Vérifie que la source pronotepy est entièrement configurée. + """Vérifie si la source pronotepy est utilisable selon le mode d'authentification. - :return: ``True`` si ``url``, ``username`` et ``password`` - sont tous définis, ``False`` sinon. + :return: ``True`` si pronotepy est configuré pour le mode + d'authentification actif, ``False`` sinon. :rtype: bool """ pronote = self._settings.pronote + if pronote.auth_mode == "qr_token": + # En mode qr_token, seul PRONOTE_URL est requis. + # Le QR code et le PIN ne sont nécessaires que pour l'enrôlement initial. + # Les exécutions suivantes utilisent le token persisté. + return pronote.url is not None + # En mode password, URL + identifiant + mot de passe sont requis. return ( pronote.url is not None and pronote.username is not None @@ -249,10 +255,14 @@ class PronoteFetcher: :return: Tuple ``(cours, événements scolaires)``. :rtype: tuple[list[Lesson], list[SchoolEvent]] :raises PipelineCriticalError: Si toutes les sources tentées échouent. + :raises PronoteAuthRotationError: Si une rotation du token d'authentification + pronotepy est nécessaire : propagée telle quelle, sans repli. """ primary, fallback = self._agenda_sources() try: return self._fetch_agenda_source(primary) + except PronoteAuthRotationError: + raise except Exception as exc: logger.error( "Échec de la récupération %s pour l'agenda : %s", @@ -266,6 +276,8 @@ class PronoteFetcher: logger.info("Repli sur %s pour l'agenda.", fallback) try: lessons, school_events = self._fetch_agenda_source(fallback) + except PronoteAuthRotationError: + raise except Exception as exc: logger.error( "Échec de la récupération %s pour l'agenda : %s", @@ -370,10 +382,14 @@ class PronoteFetcher: :return: Liste des devoirs. :rtype: list[Homework] :raises PipelineCriticalError: Si toutes les sources tentées échouent. + :raises PronoteAuthRotationError: Si une rotation du token d'authentification + pronotepy est nécessaire : propagée telle quelle, sans repli. """ primary, fallback = self._homework_sources() try: return self._fetch_homework_source(primary, target_date) + except PronoteAuthRotationError: + raise except Exception as exc: logger.error( "Échec de la récupération %s pour les devoirs : %s", @@ -387,6 +403,8 @@ class PronoteFetcher: logger.info("Repli sur %s pour les devoirs.", fallback) try: homeworks = self._fetch_homework_source(fallback, target_date) + except PronoteAuthRotationError: + raise except Exception as exc: logger.error( "Échec de la récupération %s pour les devoirs : %s", diff --git a/tests/integration/test_pipeline_runner.py b/tests/integration/test_pipeline_runner.py index 46feffc..85d77d2 100644 --- a/tests/integration/test_pipeline_runner.py +++ b/tests/integration/test_pipeline_runner.py @@ -9,17 +9,19 @@ import pytest from pydantic import SecretStr from pronote_sync.config.settings import AISettings, AppSettings, PronoteSettings, Settings -from pronote_sync.errors import PipelineCriticalError, PipelineWarning +from pronote_sync.errors import PipelineCriticalError, PipelineWarning, PronoteAuthRotationError from pronote_sync.models.agenda import Lesson, LessonStatus, SchoolEvent from pronote_sync.models.blog import BlogArticle from pronote_sync.models.diff import AgendaDiff from pronote_sync.models.homework import Homework from pronote_sync.models.message import Message from pronote_sync.models.sync import CalDAVSyncResult, CalDAVSyncStatus +from pronote_sync.models.xmpp import XmppMessage from pronote_sync.pipeline.run import PipelineRunner from pronote_sync.sources.blog.result import BlogRSSFetchResult from pronote_sync.sources.blog.rss import BlogRSSClient from pronote_sync.sources.blog.state import BlogRSSState +from pronote_sync.sources.pronote.auth_state import PronoteAuthState from pronote_sync.sources.pronote.fallback import PronoteFetcher from pronote_sync.sync.diff import AgendaComparator @@ -393,6 +395,60 @@ def test_from_settings_with_theoretical_agenda_instantiates_comparator( assert isinstance(runner._agenda_comparator, RecordingComparator) +def test_from_settings_password_mode_passes_auth_state_none( + monkeypatch: pytest.MonkeyPatch, +) -> None: + """Password mode (default) constructs PronoteClient with auth_state=None.""" + import pronote_sync.pipeline.run as run_module + + constructed: list[tuple[object, object]] = [] + + class RecordingClient: + """PronoteClient constructor recording the supplied auth_state.""" + + def __init__(self, settings: PronoteSettings, *, auth_state: object) -> None: + """Record the constructor arguments used by the composition root. + + :param settings: Pronote settings supplied by the composition root. + :param auth_state: Auth state handler supplied by the composition root. + """ + constructed.append((settings, auth_state)) + + monkeypatch.setattr(run_module, "PronoteClient", RecordingClient) + + PipelineRunner.from_settings(Settings()) + + assert len(constructed) == 1 + assert constructed[0][1] is None + + +def test_from_settings_qr_token_mode_passes_auth_state_instance( + monkeypatch: pytest.MonkeyPatch, +) -> None: + """QR-token mode constructs PronoteClient with a PronoteAuthState instance.""" + import pronote_sync.pipeline.run as run_module + + constructed: list[tuple[object, object]] = [] + + class RecordingClient: + """PronoteClient constructor recording the supplied auth_state.""" + + def __init__(self, settings: PronoteSettings, *, auth_state: object) -> None: + """Record the constructor arguments used by the composition root. + + :param settings: Pronote settings supplied by the composition root. + :param auth_state: Auth state handler supplied by the composition root. + """ + constructed.append((settings, auth_state)) + + monkeypatch.setattr(run_module, "PronoteClient", RecordingClient) + + PipelineRunner.from_settings(Settings(pronote=PronoteSettings(auth_mode="qr_token"))) + + assert len(constructed) == 1 + assert isinstance(constructed[0][1], PronoteAuthState) + + def test_runner_reuses_ical_download_and_parse_within_one_run( monkeypatch: pytest.MonkeyPatch, pipeline_inputs: tuple[Lesson, Homework], @@ -1063,3 +1119,338 @@ def test_runner_ical_cache_cleanup_on_second_run( "https://pronote.example.test/calendar.ics", "https://pronote.example.test/calendar.ics", ] + + +def test_rotation_error_sends_xmpp_notification() -> None: + """Une PronoteAuthRotationError envoie une notification XMPP puis retourne un résultat dégradé. + + Ce test vérifie que l'erreur de rotation se propage à travers le pipeline réel + (PronoteFetcher → fetch_step → PipelineRunner.run) et déclenche une notification XMPP + avec un message actionnable. + """ + calls: list[str] = [] + channel = StubChannel(calls) + + # Créer un client Pronote qui lève PronoteAuthRotationError + class RotatingPronoteClient: + """Client Pronote qui simule une erreur de rotation de token.""" + + def get_lessons(self, start: date, end: date) -> list[Lesson]: + """Lève l'erreur de rotation lors de la récupération des cours. + + :param start: Début de la fenêtre (ignoré). + :param end: Fin de la fenêtre (ignoré). + :return: Ne retourne jamais. + :raises PronoteAuthRotationError: Toujours. + """ + del start, end + raise PronoteAuthRotationError("Token persisté expiré : ré-enrôlement requis") + + def get_homeworks(self, start: date, end: date) -> list[Homework]: + """Ne devrait pas être appelé si fetch_agenda échoue. + + :param start: Début de la fenêtre (ignoré). + :param end: Fin de la fenêtre (ignoré). + :return: Liste vide. + :rtype: list[Homework] + """ + del start, end + return [] + + def get_messages(self) -> list[Message]: + """Ne devrait pas être appelé si fetch_agenda échoue. + + :return: Liste vide. + :rtype: list[Message] + """ + return [] + + def get_informations(self) -> list[Message]: + """Ne devrait pas être appelé si fetch_agenda échoue. + + :return: Liste vide. + :rtype: list[Message] + """ + return [] + + settings = Settings( + pronote=PronoteSettings( + url="https://pronote.example.com", + username="test", + password=SecretStr("test_password"), + ent="bordeaux", + account_type="parent", + agenda_source="pronotepy", + homework_source="pronotepy", + ) + ) + + runner = PipelineRunner( + settings=settings, + pronote_fetcher=PronoteFetcher(settings, RotatingPronoteClient()), + channel=channel, + now_provider=lambda: datetime(2026, 9, 8, 7, 0), + ) + + data, errors = runner.run() + + assert data is None + assert len(errors) == 1 + assert isinstance(errors[0], PipelineCriticalError) + assert len(channel.messages) == 1 + message = channel.messages[0] + assert isinstance(message, XmppMessage) + assert message.target_date == date(2026, 9, 8) + assert message.synthesis is not None + assert "Rotation" in message.synthesis + assert "token" in message.synthesis + assert "QR code" in message.synthesis + + +def test_rotation_error_no_channel_no_xmpp_send() -> None: + """Sans canal XMPP, l'erreur de rotation ne tente aucun envoi. + + Ce test vérifie que même sans canal XMPP configuré, l'erreur de rotation + est correctement capturée et retournée dans la liste des erreurs. + """ + + class RotatingPronoteClient: + """Client Pronote qui simule une erreur de rotation de token.""" + + def get_lessons(self, start: date, end: date) -> list[Lesson]: + """Lève l'erreur de rotation lors de la récupération des cours. + + :param start: Début de la fenêtre (ignoré). + :param end: Fin de la fenêtre (ignoré). + :return: Ne retourne jamais. + :raises PronoteAuthRotationError: Toujours. + """ + del start, end + raise PronoteAuthRotationError("Token persisté expiré : ré-enrôlement requis") + + def get_homeworks(self, start: date, end: date) -> list[Homework]: + """Ne devrait pas être appelé. + + :param start: Début de la fenêtre (ignoré). + :param end: Fin de la fenêtre (ignoré). + :return: Liste vide. + :rtype: list[Homework] + """ + del start, end + return [] + + def get_messages(self) -> list[Message]: + """Ne devrait pas être appelé. + + :return: Liste vide. + :rtype: list[Message] + """ + return [] + + def get_informations(self) -> list[Message]: + """Ne devrait pas être appelé. + + :return: Liste vide. + :rtype: list[Message] + """ + return [] + + settings = Settings( + pronote=PronoteSettings( + url="https://pronote.example.com", + username="test", + password=SecretStr("test_password"), + ent="bordeaux", + account_type="parent", + agenda_source="pronotepy", + homework_source="pronotepy", + ) + ) + + runner = PipelineRunner( + settings=settings, + pronote_fetcher=PronoteFetcher(settings, RotatingPronoteClient()), + channel=None, + now_provider=lambda: datetime(2026, 9, 8, 7, 0), + ) + + data, errors = runner.run() + + assert data is None + assert len(errors) == 1 + assert isinstance(errors[0], PipelineCriticalError) + + +def test_rotation_error_dry_run_no_xmpp_send() -> None: + """En dry-run, l'erreur de rotation n'envoie aucune notification XMPP. + + Ce test vérifie que même en mode dry-run, l'erreur de rotation est correctement + capturée et retournée, mais aucune notification XMPP n'est envoyée. + """ + + class RotatingPronoteClient: + """Client Pronote qui simule une erreur de rotation de token.""" + + def get_lessons(self, start: date, end: date) -> list[Lesson]: + """Lève l'erreur de rotation lors de la récupération des cours. + + :param start: Début de la fenêtre (ignoré). + :param end: Fin de la fenêtre (ignoré). + :return: Ne retourne jamais. + :raises PronoteAuthRotationError: Toujours. + """ + del start, end + raise PronoteAuthRotationError("Token persisté expiré : ré-enrôlement requis") + + def get_homeworks(self, start: date, end: date) -> list[Homework]: + """Ne devrait pas être appelé. + + :param start: Début de la fenêtre (ignoré). + :param end: Fin de la fenêtre (ignoré). + :return: Liste vide. + :rtype: list[Homework] + """ + del start, end + return [] + + def get_messages(self) -> list[Message]: + """Ne devrait pas être appelé. + + :return: Liste vide. + :rtype: list[Message] + """ + return [] + + def get_informations(self) -> list[Message]: + """Ne devrait pas être appelé. + + :return: Liste vide. + :rtype: list[Message] + """ + return [] + + calls: list[str] = [] + channel = StubChannel(calls) + + settings = Settings( + pronote=PronoteSettings( + url="https://pronote.example.com", + username="test", + password=SecretStr("test_password"), + ent="bordeaux", + account_type="parent", + agenda_source="pronotepy", + homework_source="pronotepy", + messages_source="pronotepy", + auth_mode="password", + qr_code_file=None, + qr_pin=None, + ical_url=None, + ) + ) + + runner = PipelineRunner( + settings=settings, + pronote_fetcher=PronoteFetcher(settings, RotatingPronoteClient()), + channel=channel, + dry_run=True, + now_provider=lambda: datetime(2026, 9, 8, 7, 0), + ) + + data, errors = runner.run() + + assert data is None + assert len(errors) == 1 + assert isinstance(errors[0], PipelineCriticalError) + assert channel.messages == [] + assert "send" not in calls + + +def test_no_secrets_in_xmpp_message() -> None: + """La synthèse XMPP de rotation ne contient aucun secret (token, PIN, URL). + + Ce test vérifie que le message XMPP généré pour une erreur de rotation + ne contient aucun secret sensible, même si l'erreur originale en contenait. + """ + + class RotatingPronoteClient: + """Client Pronote qui simule une erreur de rotation avec secrets dans message.""" + + def get_lessons(self, start: date, end: date) -> list[Lesson]: + """Lève l'erreur de rotation avec message contenant des secrets. + + :param start: Début de la fenêtre (ignoré). + :param end: Fin de la fenêtre (ignoré). + :return: Ne retourne jamais. + :raises PronoteAuthRotationError: Toujours, avec des secrets dans le message. + """ + del start, end + raise PronoteAuthRotationError( + "Token sk-sentinel-token-987654 invalide et PIN 000000 pour " + "https://pronote.sentinel.example/icalsecurise" + ) + + def get_homeworks(self, start: date, end: date) -> list[Homework]: + """Ne devrait pas être appelé. + + :param start: Début de la fenêtre (ignoré). + :param end: Fin de la fenêtre (ignoré). + :return: Liste vide. + :rtype: list[Homework] + """ + del start, end + return [] + + def get_messages(self) -> list[Message]: + """Ne devrait pas être appelé. + + :return: Liste vide. + :rtype: list[Message] + """ + return [] + + def get_informations(self) -> list[Message]: + """Ne devrait pas être appelé. + + :return: Liste vide. + :rtype: list[Message] + """ + return [] + + channel = StubChannel([]) + + settings = Settings( + pronote=PronoteSettings( + url="https://pronote.example.com", + username="test", + password=SecretStr("test_password"), + ent="bordeaux", + account_type="parent", + agenda_source="pronotepy", + homework_source="pronotepy", + ) + ) + + runner = PipelineRunner( + settings=settings, + pronote_fetcher=PronoteFetcher(settings, RotatingPronoteClient()), + channel=channel, + now_provider=lambda: datetime(2026, 9, 8, 7, 0), + ) + + data, errors = runner.run() + + assert data is None + assert len(errors) == 1 + assert len(channel.messages) == 1 + message = channel.messages[0] + assert isinstance(message, XmppMessage) + assert message.synthesis is not None + # Vérifier que les secrets ne sont pas dans le message final + assert "sk-sentinel-token-987654" not in message.synthesis + assert "000000" not in message.synthesis + assert "pronote.sentinel.example" not in message.synthesis + # Vérifier que le message contient les instructions actionnables + assert ".pronote_auth_state.json" in message.synthesis + assert "PRONOTE_QR_CODE_FILE" in message.synthesis + assert "PRONOTE_QR_PIN" in message.synthesis diff --git a/tests/unit/test_config.py b/tests/unit/test_config.py index d5ea40e..81efb2d 100644 --- a/tests/unit/test_config.py +++ b/tests/unit/test_config.py @@ -131,4 +131,80 @@ def test_url_from_pronote_url_env_var(monkeypatch: MonkeyPatch) -> None: assert settings.pronote.url == test_url +def test_auth_mode_default_password() -> None: + """Vérifie que ``auth_mode`` vaut ``"password"`` par défaut. + + :return: None + """ + settings = PronoteSettings() + assert settings.auth_mode == "password" + + +def test_auth_mode_env_qr_token(monkeypatch: MonkeyPatch) -> None: + """Vérifie que ``PRONOTE_AUTH_MODE=qr_token`` est chargé correctement. + + :param monkeypatch: Fixture pytest pour modifier temporairement l'environnement. + :return: None + """ + monkeypatch.setenv("PRONOTE_AUTH_MODE", "qr_token") + settings = load_settings() + assert settings.pronote.auth_mode == "qr_token" + + +def test_qr_pin_loaded_as_secretstr_and_masked(monkeypatch: MonkeyPatch) -> None: + """Vérifie que ``PRONOTE_QR_PIN`` est chargé en ``SecretStr`` et masqué. + + Le PIN ne doit apparaître nulle part dans les représentations textuelles + (str, repr, JSON) : seul le masque ``**********`` est visible. + + :param monkeypatch: Fixture pytest pour modifier temporairement l'environnement. + :return: None + """ + monkeypatch.setenv("PRONOTE_QR_PIN", "123456") + settings = load_settings() + assert isinstance(settings.pronote.qr_pin, SecretStr) + assert settings.pronote.qr_pin.get_secret_value() == "123456" + + str_repr = str(settings) + assert "123456" not in str_repr + assert "**********" in str_repr + + repr_repr = repr(settings) + assert "123456" not in repr_repr + assert "**********" in repr_repr + + json_str = settings.model_dump_json() + assert "123456" not in json_str + assert "**********" in json_str + + +def test_qr_code_file_loaded_as_plain_string(monkeypatch: MonkeyPatch) -> None: + """Vérifie que ``PRONOTE_QR_CODE_FILE`` est chargé comme chaîne simple. + + :param monkeypatch: Fixture pytest pour modifier temporairement l'environnement. + :return: None + """ + monkeypatch.setenv("PRONOTE_QR_CODE_FILE", "/data/qr_code.png") + settings = load_settings() + assert isinstance(settings.pronote.qr_code_file, str) + assert settings.pronote.qr_code_file == "/data/qr_code.png" + + +def test_qr_pin_in_redaction_secrets(monkeypatch: MonkeyPatch) -> None: + """Vérifie que le PIN QR est collecté pour la rédaction des secrets. + + Le ``SecretStr`` du PIN doit figurer dans ``redaction_secrets()`` et sa + représentation textuelle doit rester masquée. + + :param monkeypatch: Fixture pytest pour modifier temporairement l'environnement. + :return: None + """ + monkeypatch.setenv("PRONOTE_QR_PIN", "654321") + settings = load_settings() + secrets = settings.redaction_secrets() + assert settings.pronote.qr_pin in secrets + assert "654321" not in repr(settings.pronote.qr_pin) + assert "**********" in repr(settings.pronote.qr_pin) + + # Ensure trailing newline diff --git a/tests/unit/test_fallback.py b/tests/unit/test_fallback.py index 282fc4e..c6e48ab 100644 --- a/tests/unit/test_fallback.py +++ b/tests/unit/test_fallback.py @@ -1315,6 +1315,110 @@ def test_is_pronotepy_configured_without_ent_returns_true() -> None: assert fetcher._is_pronotepy_configured() is True +def test_is_pronotepy_configured_password_mode_all_set() -> None: + """Test _is_pronotepy_configured() en mode password avec tous les champs définis. + + URL, identifiant et mot de passe sont présents : la fonction retourne True. + + :return: None + :rtype: None + """ + settings = Settings( + pronote=PronoteSettings( + url="https://pronote.example.com", + username="testuser", + password=SecretStr("testpass"), + agenda_source="pronotepy", + homework_source="pronotepy", + ), + app=Settings().app, + ) + + client: _MockPronoteClientProtocol = MagicMock() + fetcher = PronoteFetcher(settings=settings, pronote_client=client) + + assert fetcher._is_pronotepy_configured() is True + + +def test_is_pronotepy_configured_password_mode_missing_password() -> None: + """Test _is_pronotepy_configured() en mode password sans mot de passe. + + Le mot de passe est None : la fonction retourne False. + + :return: None + :rtype: None + """ + settings = Settings( + pronote=PronoteSettings( + url="https://pronote.example.com", + username="testuser", + password=None, + agenda_source="pronotepy", + homework_source="pronotepy", + ), + app=Settings().app, + ) + + client: _MockPronoteClientProtocol = MagicMock() + fetcher = PronoteFetcher(settings=settings, pronote_client=client) + + assert fetcher._is_pronotepy_configured() is False + + +def test_is_pronotepy_configured_qr_token_mode_url_only() -> None: + """Test _is_pronotepy_configured() en mode qr_token avec URL uniquement. + + En mode qr_token, seul l'URL est requis : l'identifiant et le mot de + passe peuvent être absents, la fonction retourne True. + + :return: None + :rtype: None + """ + settings = Settings( + pronote=PronoteSettings( + url="https://pronote.example.com", + username=None, + password=None, + auth_mode="qr_token", + agenda_source="pronotepy", + homework_source="pronotepy", + ), + app=Settings().app, + ) + + client: _MockPronoteClientProtocol = MagicMock() + fetcher = PronoteFetcher(settings=settings, pronote_client=client) + + assert fetcher._is_pronotepy_configured() is True + + +def test_is_pronotepy_configured_qr_token_mode_no_url() -> None: + """Test _is_pronotepy_configured() en mode qr_token sans URL. + + L'URL est None : la fonction retourne False, même si le mode qr_token + ne requiert que PRONOTE_URL. + + :return: None + :rtype: None + """ + settings = Settings( + pronote=PronoteSettings( + url=None, + username=None, + password=None, + auth_mode="qr_token", + agenda_source="pronotepy", + homework_source="pronotepy", + ), + app=Settings().app, + ) + + client: _MockPronoteClientProtocol = MagicMock() + fetcher = PronoteFetcher(settings=settings, pronote_client=client) + + assert fetcher._is_pronotepy_configured() is False + + def test_agenda_sources_auto_without_ical_and_without_ent_returns_pronotepy() -> None: """Test _agenda_sources() en mode AUTO sans iCal URL et sans ent retourne pronotepy. diff --git a/tests/unit/test_pronote_auth_state.py b/tests/unit/test_pronote_auth_state.py new file mode 100644 index 0000000..fcc1e30 --- /dev/null +++ b/tests/unit/test_pronote_auth_state.py @@ -0,0 +1,208 @@ +"""Tests unitaires pour le gestionnaire d'état d'authentification Pronote. + +Ce module valide le comportement de :class:`PronoteAuthState` dans +:mod:`pronote_sync.sources.pronote.auth_state`. Les tests couvrent : + +- Le chargement des credentials (absent, corrompu, version invalide), +- La persistance et le rechargement des credentials, +- Les permissions ``0600`` du fichier d'état, +- La suppression via :meth:`clear`, +- L'absence de fuite des credentials dans les journaux. + +Tous les tests utilisent des fichiers temporaires via la fixture ``tmp_path``. +""" + +from __future__ import annotations + +import json +import logging +import os +from pathlib import Path + +import pytest + +from pronote_sync.sources.pronote.auth_state import PronoteAuthState + + +def test_load_no_file_returns_none(tmp_path: Path) -> None: + """Vérifie qu'un fichier d'état absent renvoie ``None``. + + :param tmp_path: Fixture pytest pour un répertoire temporaire. + :return: None + """ + state = PronoteAuthState(tmp_path / "missing.json") + + assert state.load() is None + + +def test_save_then_load_roundtrip(tmp_path: Path) -> None: + """Vérifie que des credentials sauvegardés sont rechargés à l'identique. + + :param tmp_path: Fixture pytest pour un répertoire temporaire. + :return: None + """ + state_file = tmp_path / "auth.json" + credentials = { + "pronote_url": "https://example.com/pronote", + "username": "parent-1", + "password": "token-123", # pragma: allowlist secret + "uuid": "uuid-456", + } + + state = PronoteAuthState(state_file) + state.save(credentials) + loaded = PronoteAuthState(state_file).load() + + assert loaded == credentials + + +def test_load_corrupted_json_returns_none(tmp_path: Path, caplog: pytest.LogCaptureFixture) -> None: + """Vérifie qu'un fichier JSON corrompu renvoie ``None`` et journalise un avertissement. + + :param tmp_path: Fixture pytest pour un répertoire temporaire. + :param caplog: Fixture pytest pour capturer les logs. + :return: None + """ + state_file = tmp_path / "corrupt.json" + state_file.write_text("not json{", encoding="utf-8") + + with caplog.at_level("WARNING"): + result = PronoteAuthState(state_file).load() + + assert result is None + assert "Impossible de charger le fichier d'état d'authentification Pronote" in caplog.text + + +def test_load_wrong_version_returns_none(tmp_path: Path, caplog: pytest.LogCaptureFixture) -> None: + """Vérifie qu'une version non supportée renvoie ``None`` et journalise un avertissement. + + :param tmp_path: Fixture pytest pour un répertoire temporaire. + :param caplog: Fixture pytest pour capturer les logs. + :return: None + """ + state_file = tmp_path / "wrong_version.json" + state_file.write_text( + json.dumps( + { + "version": 2, + "credentials": { + "pronote_url": "https://example.com", + "username": "u", + "password": "t", + "uuid": "i", + }, + } + ), + encoding="utf-8", + ) + + with caplog.at_level("WARNING"): + result = PronoteAuthState(state_file).load() + + assert result is None + assert "version absente ou non supportée" in caplog.text + + +def test_load_missing_version_returns_none( + tmp_path: Path, caplog: pytest.LogCaptureFixture +) -> None: + """Vérifie qu'un fichier sans champ version renvoie ``None``. + + :param tmp_path: Fixture pytest pour un répertoire temporaire. + :param caplog: Fixture pytest pour capturer les logs. + :return: None + """ + state_file = tmp_path / "missing_version.json" + state_file.write_text( + json.dumps( + { + "credentials": { + "pronote_url": "https://example.com", + "username": "u", + "password": "t", + "uuid": "i", + } + } + ), + encoding="utf-8", + ) + + with caplog.at_level("WARNING"): + result = PronoteAuthState(state_file).load() + + assert result is None + + +def test_clear_removes_file(tmp_path: Path) -> None: + """Vérifie que clear supprime le fichier d'état existant. + + :param tmp_path: Fixture pytest pour un répertoire temporaire. + :return: None + """ + state_file = tmp_path / "auth.json" + state = PronoteAuthState(state_file) + state.save({"pronote_url": "u", "username": "u", "password": "t", "uuid": "i"}) + + assert state_file.exists() + state.clear() + + assert not state_file.exists() + + +def test_clear_no_file_noop(tmp_path: Path) -> None: + """Vérifie que clear ne fait rien quand le fichier n'existe pas. + + :param tmp_path: Fixture pytest pour un répertoire temporaire. + :return: None + """ + state = PronoteAuthState(tmp_path / "missing.json") + + state.clear() + + +def test_save_creates_file_with_0600_permissions(tmp_path: Path) -> None: + """Vérifie que le fichier d'état est créé avec les permissions ``0600``. + + Le fichier contient un token vivant : il doit être lisible uniquement + par le propriétaire. + + :param tmp_path: Fixture pytest pour un répertoire temporaire. + :return: None + """ + state_file = tmp_path / "auth.json" + state = PronoteAuthState(state_file) + state.save({"pronote_url": "u", "username": "u", "password": "t", "uuid": "i"}) + + assert os.stat(state_file).st_mode & 0o777 == 0o600 + + +def test_no_credentials_in_logs(tmp_path: Path, caplog: pytest.LogCaptureFixture) -> None: + """Vérifie qu'aucun contenu des credentials n'apparaît dans les journaux. + + Des sentinelles distinctes sont utilisées pour ``pronote_url``, + ``username``, ``password`` et ``uuid`` ; aucun de ces marqueurs ne doit + apparaître dans les messages journalisés lors d'une sauvegarde, d'un + chargement et d'une suppression. + + :param tmp_path: Fixture pytest pour un répertoire temporaire. + :param caplog: Fixture pytest pour capturer les logs. + :return: None + """ + state_file = tmp_path / "auth.json" + credentials = { + "pronote_url": "https://SENTINEL_URL_ZZZ.example/pronote", + "username": "SENTINEL_USER_ZZZ", + "password": "SENTINEL_PASSWORD_ZZZ", # pragma: allowlist secret + "uuid": "SENTINEL_UUID_ZZZ", + } + + state = PronoteAuthState(state_file) + with caplog.at_level(logging.DEBUG): + state.save(credentials) + state.load() + state.clear() + + assert "SENTINEL_URL_ZZZ" not in caplog.text + assert "SENTINEL_USER_ZZZ" not in caplog.text + assert "SENTINEL_PASSWORD_ZZZ" not in caplog.text + assert "SENTINEL_UUID_ZZZ" not in caplog.text diff --git a/tests/unit/test_pronote_client.py b/tests/unit/test_pronote_client.py index e6e96d9..ad4185d 100644 --- a/tests/unit/test_pronote_client.py +++ b/tests/unit/test_pronote_client.py @@ -7,7 +7,10 @@ utilisent des mocks pour éviter tout accès réseau réel à Pronote. from __future__ import annotations +import json +import logging from datetime import date, datetime +from pathlib import Path import pronotepy import pytest @@ -15,9 +18,11 @@ import pytest_mock from pydantic import SecretStr from pronote_sync.config.settings import PronoteSettings +from pronote_sync.errors import PronoteAuthRotationError from pronote_sync.models.agenda import Lesson, LessonStatus from pronote_sync.models.homework import Homework from pronote_sync.models.message import Message, MessageType +from pronote_sync.sources.pronote.auth_state import PronoteAuthState from pronote_sync.sources.pronote.client import PronoteClient, PronoteClientProtocol # --- Protocol tests --- @@ -613,4 +618,585 @@ def test_get_informations_degraded_on_error( assert messages == [] +# --- QR code / token authentication tests --- + + +def test_connect_password_mode_unchanged( + mocker: pytest_mock.MockerFixture, + pronote_settings: PronoteSettings, +) -> None: + """Vérifie que le mode password conserve le comportement historique. + + :param mocker: Fixture pytest-mock pour le mocking. + :param pronote_settings: Paramètres Pronote valides en mode password. + :return: None + """ + from unittest.mock import Mock + + mock_client = mocker.MagicMock() + mock_client_class = Mock(return_value=mock_client) + mocker.patch("pronotepy.ParentClient", new=mock_client_class) + mocker.patch("pronotepy.Client") + + client = PronoteClient(pronote_settings, auth_state=None) + connected = client._connect() + + assert connected is mock_client + mock_client_class.assert_called_once_with( + pronote_url="https://pronote.example.com", + username="testuser", + password="testpass", # pragma: allowlist secret + ent=mocker.ANY, + ) + # Connexion paresseuse : un second appel réutilise le client déjà créé + client._connect() + assert mock_client_class.call_count == 1 + + +def test_connect_qr_token_with_persisted_creds( + mocker: pytest_mock.MockerFixture, +) -> None: + """Vérifie le login par token persisté en mode qr_token. + + Les credentials chargés depuis :class:`PronoteAuthState` sont rejoués via + ``token_login`` et le token rotate est resauvegardé. + + :param mocker: Fixture pytest-mock pour le mocking. + :return: None + """ + creds = { + "pronote_url": "https://pronote.example.com", + "username": "testuser", + "password": "persisted-token", # pragma: allowlist secret + "uuid": "persisted-uuid", + } + rotated_creds = {**creds, "uuid": "rotated-uuid"} + auth_state = mocker.MagicMock(spec=PronoteAuthState) + auth_state.load.return_value = creds + + mock_client = mocker.MagicMock() + mock_client.logged_in = True + mock_client.export_credentials.return_value = rotated_creds + mocker.patch("pronotepy.ParentClient.token_login", return_value=mock_client) + + settings = PronoteSettings( + url="https://pronote.example.com", + username="testuser", + password=SecretStr("testpass"), + ent=None, + account_type="parent", + auth_mode="qr_token", + ) + client = PronoteClient(settings, auth_state=auth_state) + connected = client._connect() + + assert connected is mock_client + pronotepy.ParentClient.token_login.assert_called_once_with(**creds) # type: ignore[attr-defined] + auth_state.save.assert_called_once_with(rotated_creds) + + +def test_connect_qr_token_no_creds_with_qr_code( + mocker: pytest_mock.MockerFixture, + tmp_path: Path, +) -> None: + """Vérifie l'enrôlement initial par QR code quand aucun token n'est persisté. + + :param mocker: Fixture pytest-mock pour le mocking. + :param tmp_path: Répertoire temporaire de test. + :return: None + """ + qr_file = tmp_path / "qr_code.json" + qr_file.write_text( + json.dumps( + { + "login": "testuser", + "jeton": "qr-jeton", + "url": "https://pronote.example.com", + } + ), + encoding="utf-8", + ) + auth_state = mocker.MagicMock(spec=PronoteAuthState) + auth_state.load.return_value = None + + creds = { + "pronote_url": "https://pronote.example.com", + "username": "testuser", + "password": "new-token", # pragma: allowlist secret + "uuid": "new-uuid", + } + mock_client = mocker.MagicMock() + mock_client.logged_in = True + mock_client.export_credentials.return_value = creds + mocker.patch("pronotepy.ParentClient.qrcode_login", return_value=mock_client) + + settings = PronoteSettings( + url="https://pronote.example.com", + username="testuser", + password=SecretStr("testpass"), + ent=None, + account_type="parent", + auth_mode="qr_token", + qr_code_file=str(qr_file), + qr_pin=SecretStr("123456"), + ) + client = PronoteClient(settings, auth_state=auth_state) + connected = client._connect() + + assert connected is mock_client + qrcode_login = pronotepy.ParentClient.qrcode_login + qrcode_login.assert_called_once() # type: ignore[attr-defined] + kwargs = qrcode_login.call_args.kwargs # type: ignore[attr-defined] + assert kwargs["pin"] == "123456" + assert kwargs["qr_code"] == { + "login": "testuser", + "jeton": "qr-jeton", + "url": "https://pronote.example.com", + } + assert kwargs["uuid"].startswith("pronote-sync-") + auth_state.save.assert_called_once_with(creds) + + +def test_connect_qr_token_token_login_fails_raises_rotation_error( + mocker: pytest_mock.MockerFixture, + tmp_path: Path, +) -> None: + """Vérifie la levée de PronoteAuthRotationError quand le token persisté est invalide. + + En cas d'échec du login par token, aucun repli vers l'enrôlement QR + n'est tenté : l'erreur de rotation est levée immédiatement, même si un + fichier QR est disponible. + + :param mocker: Fixture pytest-mock pour le mocking. + :param tmp_path: Répertoire temporaire de test. + :return: None + """ + qr_file = tmp_path / "qr_code.json" + qr_file.write_text( + json.dumps( + { + "login": "testuser", + "jeton": "qr-jeton", + "url": "https://pronote.example.com", + } + ), + encoding="utf-8", + ) + auth_state = mocker.MagicMock(spec=PronoteAuthState) + auth_state.load.return_value = { + "pronote_url": "https://pronote.example.com", + "username": "testuser", + "password": "expired-token", # pragma: allowlist secret + "uuid": "old-uuid", + } + + token_login = mocker.patch("pronotepy.ParentClient.token_login") + token_login.side_effect = pronotepy.PronoteAPIError("token invalide") + qrcode_login = mocker.patch("pronotepy.ParentClient.qrcode_login") + + settings = PronoteSettings( + url="https://pronote.example.com", + username="testuser", + password=SecretStr("testpass"), + ent=None, + account_type="parent", + auth_mode="qr_token", + qr_code_file=str(qr_file), + qr_pin=SecretStr("123456"), + ) + client = PronoteClient(settings, auth_state=auth_state) + + with pytest.raises(PronoteAuthRotationError) as exc_info: + client._connect() + + token_login.assert_called_once() + qrcode_login.assert_not_called() + auth_state.save.assert_not_called() + message = str(exc_info.value) + assert "expiré ou invalide" in message + assert ".pronote_auth_state.json" in message + assert "PRONOTE_QR_CODE_FILE" in message + + +def test_token_login_failure_raises_rotation_not_enroll( + mocker: pytest_mock.MockerFixture, + tmp_path: Path, +) -> None: + """Vérifie qu'un login par token non connecté lève PronoteAuthRotationError sans enrôlement QR. + + ``token_login`` retourne un client non connecté (``logged_in`` False) : + l'erreur de rotation est levée immédiatement et ``qrcode_login`` n'est + jamais appelé, même avec un QR code disponible. + + :param mocker: Fixture pytest-mock pour le mocking. + :param tmp_path: Répertoire temporaire de test. + :return: None + """ + qr_file = tmp_path / "qr_code.json" + qr_file.write_text( + json.dumps( + { + "login": "testuser", + "jeton": "qr-jeton", + "url": "https://pronote.example.com", + } + ), + encoding="utf-8", + ) + auth_state = mocker.MagicMock(spec=PronoteAuthState) + auth_state.load.return_value = { + "pronote_url": "https://pronote.example.com", + "username": "testuser", + "password": "expired-token", # pragma: allowlist secret + "uuid": "old-uuid", + } + + mock_client = mocker.MagicMock() + mock_client.logged_in = False + token_login = mocker.patch("pronotepy.ParentClient.token_login", return_value=mock_client) + qrcode_login = mocker.patch("pronotepy.ParentClient.qrcode_login") + + settings = PronoteSettings( + url="https://pronote.example.com", + username="testuser", + password=SecretStr("testpass"), + ent=None, + account_type="parent", + auth_mode="qr_token", + qr_code_file=str(qr_file), + qr_pin=SecretStr("123456"), + ) + client = PronoteClient(settings, auth_state=auth_state) + + with pytest.raises(PronoteAuthRotationError) as exc_info: + client._connect() + + token_login.assert_called_once() + qrcode_login.assert_not_called() + auth_state.save.assert_not_called() + message = str(exc_info.value) + assert "non connecté" in message + assert ".pronote_auth_state.json" in message + + +def test_connect_qr_token_no_creds_no_qr_raises_rotation_error( + mocker: pytest_mock.MockerFixture, +) -> None: + """Vérifie la levée de PronoteAuthRotationError sans token persisté ni QR code. + + :param mocker: Fixture pytest-mock pour le mocking. + :return: None + """ + auth_state = mocker.MagicMock(spec=PronoteAuthState) + auth_state.load.return_value = None + + settings = PronoteSettings( + url="https://pronote.example.com", + username="testuser", + password=SecretStr("testpass"), + ent=None, + account_type="parent", + auth_mode="qr_token", + ) + client = PronoteClient(settings, auth_state=auth_state) + + with pytest.raises(PronoteAuthRotationError) as exc_info: + client._connect() + + message = str(exc_info.value) + assert "PRONOTE_QR_CODE_FILE" in message + assert "PRONOTE_QR_PIN" in message + assert ".pronote_auth_state.json" in message + + +def test_connect_qr_token_token_login_fails_no_qr_raises_rotation_error( + mocker: pytest_mock.MockerFixture, +) -> None: + """Vérifie la levée de PronoteAuthRotationError quand le token échoue sans QR. + + :param mocker: Fixture pytest-mock pour le mocking. + :return: None + """ + auth_state = mocker.MagicMock(spec=PronoteAuthState) + auth_state.load.return_value = { + "pronote_url": "https://pronote.example.com", + "username": "testuser", + "password": "expired-token", # pragma: allowlist secret + "uuid": "old-uuid", + } + token_login = mocker.patch("pronotepy.ParentClient.token_login") + token_login.side_effect = pronotepy.PronoteAPIError("token invalide") + + settings = PronoteSettings( + url="https://pronote.example.com", + username="testuser", + password=SecretStr("testpass"), + ent=None, + account_type="parent", + auth_mode="qr_token", + ) + client = PronoteClient(settings, auth_state=auth_state) + + with pytest.raises(PronoteAuthRotationError) as exc_info: + client._connect() + + message = str(exc_info.value) + assert "expiré ou invalide" in message + assert ".pronote_auth_state.json" in message + assert "PRONOTE_QR_CODE_FILE" in message + + +def test_connect_qr_token_invalid_qr_json_raises_rotation_error( + mocker: pytest_mock.MockerFixture, + tmp_path: Path, +) -> None: + """Vérifie la levée de PronoteAuthRotationError pour un fichier QR illisible. + + :param mocker: Fixture pytest-mock pour le mocking. + :param tmp_path: Répertoire temporaire de test. + :return: None + """ + qr_file = tmp_path / "qr_code.json" + qr_file.write_text("{json invalide", encoding="utf-8") + auth_state = mocker.MagicMock(spec=PronoteAuthState) + auth_state.load.return_value = None + + settings = PronoteSettings( + url="https://pronote.example.com", + username="testuser", + password=SecretStr("testpass"), + ent=None, + account_type="parent", + auth_mode="qr_token", + qr_code_file=str(qr_file), + qr_pin=SecretStr("123456"), + ) + client = PronoteClient(settings, auth_state=auth_state) + + with pytest.raises(PronoteAuthRotationError) as exc_info: + client._connect() + + assert "Impossible de lire le fichier QR code" in str(exc_info.value) + + +def test_connect_qr_token_missing_qr_key_raises_rotation_error( + mocker: pytest_mock.MockerFixture, + tmp_path: Path, +) -> None: + """Vérifie la levée de PronoteAuthRotationError quand une clé QR requise manque. + + :param mocker: Fixture pytest-mock pour le mocking. + :param tmp_path: Répertoire temporaire de test. + :return: None + """ + qr_file = tmp_path / "qr_code.json" + qr_file.write_text( + json.dumps({"login": "testuser", "url": "https://pronote.example.com"}), + encoding="utf-8", + ) + auth_state = mocker.MagicMock(spec=PronoteAuthState) + auth_state.load.return_value = None + + settings = PronoteSettings( + url="https://pronote.example.com", + username="testuser", + password=SecretStr("testpass"), + ent=None, + account_type="parent", + auth_mode="qr_token", + qr_code_file=str(qr_file), + qr_pin=SecretStr("123456"), + ) + client = PronoteClient(settings, auth_state=auth_state) + + with pytest.raises(PronoteAuthRotationError) as exc_info: + client._connect() + + assert "jeton" in str(exc_info.value) + + +def test_connect_qr_token_qrcode_login_fails_raises_rotation_error( + mocker: pytest_mock.MockerFixture, + tmp_path: Path, +) -> None: + """Vérifie la levée de PronoteAuthRotationError quand le login QR échoue. + + :param mocker: Fixture pytest-mock pour le mocking. + :param tmp_path: Répertoire temporaire de test. + :return: None + """ + qr_file = tmp_path / "qr_code.json" + qr_file.write_text( + json.dumps( + { + "login": "testuser", + "jeton": "qr-jeton", + "url": "https://pronote.example.com", + } + ), + encoding="utf-8", + ) + auth_state = mocker.MagicMock(spec=PronoteAuthState) + auth_state.load.return_value = None + mocker.patch( + "pronotepy.ParentClient.qrcode_login", + side_effect=pronotepy.exceptions.QRCodeDecryptError("PIN incorrect"), + ) + + settings = PronoteSettings( + url="https://pronote.example.com", + username="testuser", + password=SecretStr("testpass"), + ent=None, + account_type="parent", + auth_mode="qr_token", + qr_code_file=str(qr_file), + qr_pin=SecretStr("123456"), + ) + client = PronoteClient(settings, auth_state=auth_state) + + with pytest.raises(PronoteAuthRotationError) as exc_info: + client._connect() + + message = str(exc_info.value) + assert "PIN invalide ou QR code expiré" in message + assert "PRONOTE_QR_CODE_FILE" in message + + +def test_no_secrets_in_rotation_error_messages( + mocker: pytest_mock.MockerFixture, + tmp_path: Path, + caplog: pytest.LogCaptureFixture, +) -> None: + """Vérifie qu'aucun secret ne fuit dans les erreurs ni les logs de rotation. + + :param mocker: Fixture pytest-mock pour le mocking. + :param tmp_path: Répertoire temporaire de test. + :param caplog: Fixture pytest de capture des logs. + :return: None + """ + sentinel_pin = "SENTINEL_PIN_42" + sentinel_token = "SENTINEL_TOKEN_7" + sentinel_url = "https://sentinel-url.pronote.example.com" + + qr_file = tmp_path / "qr_code.json" + qr_file.write_text( + json.dumps( + { + "login": "testuser", + "jeton": sentinel_token, + "url": sentinel_url, + } + ), + encoding="utf-8", + ) + auth_state = mocker.MagicMock(spec=PronoteAuthState) + auth_state.load.return_value = None + + mocker.patch( + "pronotepy.ParentClient.qrcode_login", + side_effect=pronotepy.exceptions.QRCodeDecryptError( + f"token: {sentinel_token} password: {sentinel_pin}" + ), + ) + + settings = PronoteSettings( + url="https://pronote.example.com", + username="testuser", + password=SecretStr("testpass"), + ent=None, + account_type="parent", + auth_mode="qr_token", + qr_code_file=str(qr_file), + qr_pin=SecretStr(sentinel_pin), + ) + client = PronoteClient(settings, auth_state=auth_state) + + with caplog.at_level(logging.ERROR, logger="pronote_sync.sources.pronote.client"): + with pytest.raises(PronoteAuthRotationError) as exc_info: + client._connect() + + message = str(exc_info.value) + assert sentinel_pin not in message + assert sentinel_token not in message + assert "sentinel-url" not in message + assert caplog.text + assert sentinel_pin not in caplog.text + assert sentinel_token not in caplog.text + assert "sentinel-url" not in caplog.text + + +def test_no_raw_secrets_in_logs( + mocker: pytest_mock.MockerFixture, + tmp_path: Path, + caplog: pytest.LogCaptureFixture, +) -> None: + """Vérifie l'expurgation de secrets bruts sans motif reconnaissable dans les logs. + + Des sentinelles distinctes pour le token persisté, le PIN QR et le jeton + QR sont injectées dans le message d'exception de ``token_login`` sans + motif ``cle=valeur`` ni format d'URL ; elles ne doivent apparaître ni + dans les logs ni dans l'erreur de rotation levée. + + :param mocker: Fixture pytest-mock pour le mocking. + :param tmp_path: Répertoire temporaire de test. + :param caplog: Fixture pytest de capture des logs. + :return: None + """ + sentinel_token = "SENTINEL_RAW_TOKEN_ALPHA" + sentinel_pin = "SENTINEL_RAW_PIN_BRAVO" + sentinel_jeton = "SENTINEL_RAW_JETON_CHARLIE" + + qr_file = tmp_path / "qr_code.json" + qr_file.write_text( + json.dumps( + { + "login": "testuser", + "jeton": sentinel_jeton, + "url": "https://pronote.example.com", + } + ), + encoding="utf-8", + ) + auth_state = mocker.MagicMock(spec=PronoteAuthState) + auth_state.load.return_value = { + "pronote_url": "https://pronote.example.com", + "username": "testuser", + "password": sentinel_token, + "uuid": "old-uuid", + } + + mocker.patch( + "pronotepy.ParentClient.token_login", + side_effect=pronotepy.PronoteAPIError( + f"login refusé {sentinel_token} puis {sentinel_pin} puis {sentinel_jeton}" + ), + ) + + settings = PronoteSettings( + url="https://pronote.example.com", + username="testuser", + password=SecretStr("testpass"), + ent=None, + account_type="parent", + auth_mode="qr_token", + qr_code_file=str(qr_file), + qr_pin=SecretStr(sentinel_pin), + ) + client = PronoteClient(settings, auth_state=auth_state) + + with caplog.at_level(logging.ERROR, logger="pronote_sync.sources.pronote.client"): + with pytest.raises(PronoteAuthRotationError) as exc_info: + client._connect() + + message = str(exc_info.value) + assert sentinel_token not in message + assert sentinel_pin not in message + assert sentinel_jeton not in message + assert caplog.text + assert sentinel_token not in caplog.text + assert sentinel_pin not in caplog.text + assert sentinel_jeton not in caplog.text + + # Ensure trailing newline diff --git a/tests/unit/test_rotation_propagation.py b/tests/unit/test_rotation_propagation.py new file mode 100644 index 0000000..84994b1 --- /dev/null +++ b/tests/unit/test_rotation_propagation.py @@ -0,0 +1,180 @@ +"""Unit tests for PronoteAuthRotationError propagation through each layer. + +These tests verify that the rotation error propagates correctly through the +real call chain without being wrapped in PipelineCriticalError at any layer. +""" + +from __future__ import annotations + +from datetime import date + +import pytest +from pydantic import SecretStr + +from pronote_sync.config.settings import PronoteSettings, Settings +from pronote_sync.errors import PipelineCriticalError, PronoteAuthRotationError +from pronote_sync.models.agenda import Lesson, SchoolEvent +from pronote_sync.models.homework import Homework +from pronote_sync.models.message import Message +from pronote_sync.pipeline.steps.fetch import fetch_step +from pronote_sync.sources.pronote.fallback import PronoteFetcher + + +class StubPronoteClientWithRotationError: + """Stub PronoteClient that raises PronoteAuthRotationError from its methods.""" + + def get_lessons(self, start: date, end: date) -> list[Lesson]: + """Raise rotation error when fetching lessons. + + :param start: Start date (unused). + :param end: End date (unused). + :return: Never returns. + :raises PronoteAuthRotationError: Always. + """ + del start, end + raise PronoteAuthRotationError("Token persisté expiré : ré-enrôlement requis") + + def get_homeworks(self, start: date, end: date) -> list[Homework]: + """Raise rotation error when fetching homeworks. + + :param start: Start date (unused). + :param end: End date (unused). + :return: Never returns. + :raises PronoteAuthRotationError: Always. + """ + del start, end + raise PronoteAuthRotationError("Token persisté expiré : ré-enrôlement requis") + + def get_messages(self) -> list[Message]: + """Return empty messages list. + + :return: Empty list. + :rtype: list[Message] + """ + return [] + + def get_informations(self) -> list[Message]: + """Return empty information messages list. + + :return: Empty list. + :rtype: list[Message] + """ + return [] + + +class StubSettings: + """Minimal settings stub for PronoteFetcher.""" + + def __init__(self) -> None: + """Initialize with minimal configuration.""" + self.pronote = PronoteSettings( + url="https://pronote.example.com", + username="test", + password=SecretStr("test_password"), + ent="bordeaux", + account_type="parent", + agenda_source="pronotepy", + homework_source="pronotepy", + messages_source="pronotepy", + auth_mode="password", + qr_code_file=None, + qr_pin=None, + ical_url=None, + ) + self.app = type("AppSettings", (), {"sync_past_days": 7, "sync_future_days": 7})() + + +class StubFetcherWithRotationError: + """Stub PronoteFetcher that raises PronoteAuthRotationError from its methods.""" + + def __init__(self) -> None: + """Initialize the stub fetcher.""" + self._settings = StubSettings() + self._client = StubPronoteClientWithRotationError() + + def fetch_agenda(self) -> tuple[list[Lesson], list[SchoolEvent]]: + """Raise rotation error when fetching agenda. + + :return: Never returns. + :rtype: tuple[list[Lesson], list[SchoolEvent]] + :raises PronoteAuthRotationError: Always. + """ + raise PronoteAuthRotationError("Token persisté expiré : ré-enrôlement requis") + + def fetch_homework(self, target_date: date) -> list[Homework]: + """Raise rotation error when fetching homework. + + :param target_date: Target date (unused). + :return: Never returns. + :rtype: list[Homework] + :raises PronoteAuthRotationError: Always. + """ + del target_date + raise PronoteAuthRotationError("Token persisté expiré : ré-enrôlement requis") + + def fetch_messages(self) -> list[Message]: + """Return empty messages list. + + :return: Empty list. + :rtype: list[Message] + """ + return [] + + def fetch_informations(self) -> list[Message]: + """Return empty information messages list. + + :return: Empty list. + :rtype: list[Message] + """ + return [] + + +def test_pronote_fetcher_fetch_agenda_propagates_rotation_error() -> None: + """PronoteFetcher.fetch_agenda() propagates PronoteAuthRotationError without wrapping. + + This test verifies that when the underlying PronoteClient raises + PronoteAuthRotationError, the fetcher propagates it directly without + converting it to PipelineCriticalError. + """ + settings = Settings(pronote=StubSettings().pronote) + fetcher = PronoteFetcher(settings, StubPronoteClientWithRotationError()) + + with pytest.raises(PronoteAuthRotationError) as exc_info: + fetcher.fetch_agenda() + + assert "Token persisté expiré" in str(exc_info.value) + assert not isinstance(exc_info.value, PipelineCriticalError) + + +def test_pronote_fetcher_fetch_homework_propagates_rotation_error() -> None: + """PronoteFetcher.fetch_homework() propagates PronoteAuthRotationError without wrapping. + + This test verifies that when the underlying PronoteClient raises + PronoteAuthRotationError, the fetcher propagates it directly without + converting it to PipelineCriticalError. + """ + settings = Settings(pronote=StubSettings().pronote) + fetcher = PronoteFetcher(settings, StubPronoteClientWithRotationError()) + target_date = date(2026, 9, 9) + + with pytest.raises(PronoteAuthRotationError) as exc_info: + fetcher.fetch_homework(target_date) + + assert "Token persisté expiré" in str(exc_info.value) + assert not isinstance(exc_info.value, PipelineCriticalError) + + +def test_fetch_step_propagates_rotation_error() -> None: + """fetch_step() propagates PronoteAuthRotationError without wrapping. + + This test verifies that the pipeline step fetch_step() propagates + PronoteAuthRotationError directly from the fetcher without converting + it to PipelineCriticalError. + """ + fetcher = StubFetcherWithRotationError() + + with pytest.raises(PronoteAuthRotationError) as exc_info: + fetch_step(fetcher, today=date(2026, 9, 8)) + + assert "Token persisté expiré" in str(exc_info.value) + assert not isinstance(exc_info.value, PipelineCriticalError)