fix(security): neutraliser le contexte des erreurs expurgées

pronote_sync/pipeline/steps/fetch.py

@@ -121,6 +121,7 @@ def fetch_step(
     :raises PronoteAuthRotationError: Si une rotation du token d'authentification
         pronotepy est nécessaire : propagée telle quelle jusqu'au pipeline.
     """
+    critical_error: PipelineCriticalError | None = None
     try:
         lessons, school_events = fetcher.fetch_agenda()
         target_date = resolve_target_date(today or date.today(), lessons, school_events)
@@ -130,9 +131,11 @@ def fetch_step(
     except PronoteAuthRotationError:
         raise
     except Exception as exc:
-        raise PipelineCriticalError(
+        critical_error = PipelineCriticalError(
             f"Récupération Pronote impossible : {redact_exception(exc)}", step="fetch"
-        ) from None
+        )
+    if critical_error is not None:
+        raise critical_error from None
 
     messages, warnings = _fetch_optional_messages(fetcher)
     return (

pronote_sync/pipeline/steps/fetch_blog.py

@@ -5,7 +5,7 @@ from __future__ import annotations
 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.utils.redaction import redact_exception
+from pronote_sync.utils.redaction import redact_exception, redact_secrets
 
 
 def fetch_blog_step(client: BlogRSSClient | None, state: BlogRSSState | None) -> BlogRSSFetchResult:
@@ -24,15 +24,18 @@ def fetch_blog_step(client: BlogRSSClient | None, state: BlogRSSState | None) ->
     """
     if client is None or state is None:
         return BlogRSSFetchResult()
+    error_message: str | None = None
     try:
         etag, last_modified = state.get_cache_headers()
         result = client.fetch_and_parse(
             known_guids=state.get_known_guids(), etag=etag, last_modified=last_modified
         )
         if result.error is not None:
-            raise RuntimeError(result.error) from None
+            error_message = f"Récupération du blog échouée : {redact_secrets(result.error)}"
-        if not result.not_modified and not result.articles:
+        elif not result.not_modified and not result.articles:
             state.update_cache_headers(result.etag, result.last_modified)
-        return result
     except Exception as exc:
-        raise RuntimeError(f"Récupération du blog échouée : {redact_exception(exc)}") from None
+        error_message = f"Récupération du blog échouée : {redact_exception(exc)}"
+    if error_message is not None:
+        raise RuntimeError(error_message) from None
+    return result

pronote_sync/sources/pronote/auth_state.py

@@ -218,6 +218,7 @@ class PronoteAuthState:
         }
         tmp_file = self._state_file.with_suffix(".tmp")
         fd: int | None = None
+        write_error: PronoteSyncError | None = None
         try:
             # Nettoie un éventuel fichier temporaire stale laissé par une exécution interrompue.
             if tmp_file.exists():
@@ -259,10 +260,12 @@ class PronoteAuthState:
                     "Nettoyage du fichier temporaire d'état d'authentification Pronote échoué : %s",
                     redact_exception(cleanup_exc),
                 )
-            raise PronoteSyncError(
+            write_error = PronoteSyncError(
                 f"Impossible d'écrire le fichier d'état d'authentification Pronote "
                 f"{redact_secrets(str(self._state_file))}."
-            ) from None
+            )
+        if write_error is not None:
+            raise write_error from None
 
     def clear(self) -> None:
         """Supprime le fichier d'état d'authentification.

pronote_sync/sources/pronote/fallback.py

@@ -274,8 +274,9 @@ class PronoteFetcher:
                 f"Impossible de récupérer l'agenda : la source {primary} a échoué"
             ) from None
         logger.info("Repli sur %s pour l'agenda.", fallback)
+        fallback_result: tuple[list[Lesson], list[SchoolEvent]] | None = None
         try:
-                lessons, school_events = self._fetch_agenda_source(fallback)
+            fallback_result = self._fetch_agenda_source(fallback)
         except PronoteAuthRotationError:
             raise
         except Exception as exc:
@@ -284,10 +285,11 @@ class PronoteFetcher:
                 fallback,
                 redact_exception(exc),
             )
+        if fallback_result is None:
             raise PipelineCriticalError(
-                    f"Impossible de récupérer l'agenda : les sources {primary}"
+                f"Impossible de récupérer l'agenda : les sources {primary} et {fallback} ont échoué"
-                    f" et {fallback} ont échoué"
             ) from None
+        lessons, school_events = fallback_result
         if not lessons:
             logger.warning(
                 "Le repli %s pour l'agenda a retourné un résultat vide après l'échec "
@@ -401,8 +403,9 @@ class PronoteFetcher:
                 f"Impossible de récupérer les devoirs : la source {primary} a échoué"
             ) from None
         logger.info("Repli sur %s pour les devoirs.", fallback)
+        fallback_result: list[Homework] | None = None
         try:
-                homeworks = self._fetch_homework_source(fallback, target_date)
+            fallback_result = self._fetch_homework_source(fallback, target_date)
         except PronoteAuthRotationError:
             raise
         except Exception as exc:
@@ -411,10 +414,12 @@ class PronoteFetcher:
                 fallback,
                 redact_exception(exc),
             )
+        if fallback_result is None:
             raise PipelineCriticalError(
                 f"Impossible de récupérer les devoirs : les sources {primary}"
                 f" et {fallback} ont échoué"
             ) from None
+        homeworks = fallback_result
         if not homeworks:
             logger.warning(
                 "Le repli %s pour les devoirs a retourné un résultat vide après "

pronote_sync/sources/theoretical/file.py

@@ -85,6 +85,7 @@ class JsonTheoreticalAgendaProvider:
         self._file_path: str = file_path
         self._parity_service: WeekParityService | None = parity_service
         self._holiday_calendar: SchoolHolidayCalendar | None = holiday_calendar
+        load_error: PronoteSyncError | None = None
         try:
             content = Path(file_path).read_text(encoding="utf-8")
             parsed = TheoreticalAgendaFile.model_validate_json(content)
@@ -94,9 +95,11 @@ class JsonTheoreticalAgendaProvider:
                 redact_secrets(str(file_path)),
                 redact_exception(exc),
             )
-            raise PronoteSyncError(
+            load_error = PronoteSyncError(
                 f"Le fichier d'agenda théorique est invalide : {redact_secrets(str(file_path))}"
-            ) from None
+            )
+        if load_error is not None:
+            raise load_error from None
         self._lessons: tuple[TheoreticalLessonEntry, ...] = parsed.lessons
         if self._parity_service is None and any(
             entry.week in ("even", "odd") for entry in self._lessons

pronote_sync/sources/theoretical/holidays.py

@@ -77,6 +77,7 @@ class SchoolHolidayCalendar:
             raise PronoteSyncError(
                 f"Le fichier de vacances scolaires est introuvable : {redact_secrets(str(path))}"
             ) from None
+        load_error: PronoteSyncError | None = None
         try:
             data: Any = json.loads(path.read_text(encoding="utf-8"))
             file_model: SchoolHolidayFile = SchoolHolidayFile.model_validate(data)
@@ -86,9 +87,11 @@ class SchoolHolidayCalendar:
                 redact_secrets(str(path)),
                 redact_exception(exc),
             )
-            raise PronoteSyncError(
+            load_error = PronoteSyncError(
                 f"Le fichier de vacances scolaires est invalide : {redact_secrets(str(path))}"
-            ) from None
+            )
+        if load_error is not None:
+            raise load_error from None
         self._periods = file_model.periods
 
     def is_holiday(self, target_date: date) -> bool:

tests/unit/test_exception_context.py

@@ -0,0 +1,83 @@
+"""Régressions sur le contexte des exceptions expurgées."""
+
+from __future__ import annotations
+
+import ast
+import traceback
+from datetime import date
+from pathlib import Path
+
+import pytest
+
+from pronote_sync.errors import PipelineCriticalError
+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
+
+
+class _FailingFetcher:
+    """Fetcher minimal qui expose une erreur externe porteuse d'un secret."""
+
+    def fetch_agenda(self) -> tuple[list[Lesson], list[SchoolEvent]]:
+        """Déclenche une erreur externe pendant la récupération critique."""
+        raise RuntimeError("password=fetch-context-secret")
+
+    def fetch_homework(self, target_date: date) -> list[Homework]:
+        """Retourne une liste vide pour compléter le protocole du fetcher."""
+        return []
+
+    def fetch_messages(self) -> list[Message]:
+        """Retourne une liste vide pour compléter le protocole du fetcher."""
+        return []
+
+    def fetch_informations(self) -> list[Message]:
+        """Retourne une liste vide pour compléter le protocole du fetcher."""
+        return []
+
+
+def test_fetch_step_does_not_retain_external_exception_context() -> None:
+    """Vérifie qu'une erreur critique ne conserve ni secret ni contexte externe."""
+    with pytest.raises(PipelineCriticalError) as exc_info:
+        fetch_step(_FailingFetcher())
+
+    error = exc_info.value
+    formatted = "".join(traceback.format_exception(error))
+    assert "fetch-context-secret" not in str(error)
+    assert "fetch-context-secret" not in formatted
+    assert error.__cause__ is None
+    assert error.__context__ is None
+
+
+def test_production_raise_from_none_is_never_inside_except() -> None:
+    """Vérifie structurellement que les exceptions expurgées sont levées hors des handlers."""
+    root = Path(__file__).parents[2] / "pronote_sync"
+    violations: list[str] = []
+
+    class Visitor(ast.NodeVisitor):
+        """Collecte les levées ``from None`` imbriquées dans un handler."""
+
+        def __init__(self) -> None:
+            self._inside_except = False
+
+        def visit_ExceptHandler(self, node: ast.ExceptHandler) -> None:
+            """Visite un handler en marquant son périmètre lexical."""
+            previous = self._inside_except
+            self._inside_except = True
+            self.generic_visit(node)
+            self._inside_except = previous
+
+        def visit_Raise(self, node: ast.Raise) -> None:
+            """Signale une levée ``from None`` dans un handler."""
+            if (
+                self._inside_except
+                and isinstance(node.cause, ast.Constant)
+                and node.cause.value is None
+            ):
+                violations.append(f"{path}:{node.lineno}")
+            self.generic_visit(node)
+
+    for path in sorted(root.rglob("*.py")):
+        Visitor().visit(ast.parse(path.read_text(encoding="utf-8")))
+
+    assert violations == []