Compare commits

..
Author SHA1 Message Date
Codex dee5fe8eff fix(security): neutraliser le contexte des erreurs expurgées 2026-09-13 00:21:41 +02:00
9 changed files with 179 additions and 151 deletions
+5 -2
View File
@@ -121,6 +121,7 @@ def fetch_step(
:raises PronoteAuthRotationError: Si une rotation du token d'authentification :raises PronoteAuthRotationError: Si une rotation du token d'authentification
pronotepy est nécessaire : propagée telle quelle jusqu'au pipeline. pronotepy est nécessaire : propagée telle quelle jusqu'au pipeline.
""" """
critical_error: PipelineCriticalError | None = None
try: try:
lessons, school_events = fetcher.fetch_agenda() lessons, school_events = fetcher.fetch_agenda()
target_date = resolve_target_date(today or date.today(), lessons, school_events) target_date = resolve_target_date(today or date.today(), lessons, school_events)
@@ -130,9 +131,11 @@ def fetch_step(
except PronoteAuthRotationError: except PronoteAuthRotationError:
raise raise
except Exception as exc: except Exception as exc:
raise PipelineCriticalError( critical_error = PipelineCriticalError(
f"Récupération Pronote impossible : {redact_exception(exc)}", step="fetch" 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) messages, warnings = _fetch_optional_messages(fetcher)
return ( return (
+8 -5
View File
@@ -5,7 +5,7 @@ from __future__ import annotations
from pronote_sync.sources.blog.result import BlogRSSFetchResult from pronote_sync.sources.blog.result import BlogRSSFetchResult
from pronote_sync.sources.blog.rss import BlogRSSClient from pronote_sync.sources.blog.rss import BlogRSSClient
from pronote_sync.sources.blog.state import BlogRSSState 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: 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: if client is None or state is None:
return BlogRSSFetchResult() return BlogRSSFetchResult()
error_message: str | None = None
try: try:
etag, last_modified = state.get_cache_headers() etag, last_modified = state.get_cache_headers()
result = client.fetch_and_parse( result = client.fetch_and_parse(
known_guids=state.get_known_guids(), etag=etag, last_modified=last_modified known_guids=state.get_known_guids(), etag=etag, last_modified=last_modified
) )
if result.error is not None: 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) state.update_cache_headers(result.etag, result.last_modified)
return result
except Exception as exc: 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
+5 -2
View File
@@ -218,6 +218,7 @@ class PronoteAuthState:
} }
tmp_file = self._state_file.with_suffix(".tmp") tmp_file = self._state_file.with_suffix(".tmp")
fd: int | None = None fd: int | None = None
write_error: PronoteSyncError | None = None
try: try:
# Nettoie un éventuel fichier temporaire stale laissé par une exécution interrompue. # Nettoie un éventuel fichier temporaire stale laissé par une exécution interrompue.
if tmp_file.exists(): if tmp_file.exists():
@@ -259,10 +260,12 @@ class PronoteAuthState:
"Nettoyage du fichier temporaire d'état d'authentification Pronote échoué : %s", "Nettoyage du fichier temporaire d'état d'authentification Pronote échoué : %s",
redact_exception(cleanup_exc), redact_exception(cleanup_exc),
) )
raise PronoteSyncError( write_error = PronoteSyncError(
f"Impossible d'écrire le fichier d'état d'authentification Pronote " f"Impossible d'écrire le fichier d'état d'authentification Pronote "
f"{redact_secrets(str(self._state_file))}." f"{redact_secrets(str(self._state_file))}."
) from None )
if write_error is not None:
raise write_error from None
def clear(self) -> None: def clear(self) -> None:
"""Supprime le fichier d'état d'authentification. """Supprime le fichier d'état d'authentification.
+9 -4
View File
@@ -274,8 +274,9 @@ class PronoteFetcher:
f"Impossible de récupérer l'agenda : la source {primary} a échoué" f"Impossible de récupérer l'agenda : la source {primary} a échoué"
) from None ) from None
logger.info("Repli sur %s pour l'agenda.", fallback) logger.info("Repli sur %s pour l'agenda.", fallback)
fallback_result: tuple[list[Lesson], list[SchoolEvent]] | None = None
try: try:
lessons, school_events = self._fetch_agenda_source(fallback) fallback_result = self._fetch_agenda_source(fallback)
except PronoteAuthRotationError: except PronoteAuthRotationError:
raise raise
except Exception as exc: except Exception as exc:
@@ -284,10 +285,11 @@ class PronoteFetcher:
fallback, fallback,
redact_exception(exc), redact_exception(exc),
) )
if fallback_result is None:
raise PipelineCriticalError( 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 ) from None
lessons, school_events = fallback_result
if not lessons: if not lessons:
logger.warning( logger.warning(
"Le repli %s pour l'agenda a retourné un résultat vide après l'échec " "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é" f"Impossible de récupérer les devoirs : la source {primary} a échoué"
) from None ) from None
logger.info("Repli sur %s pour les devoirs.", fallback) logger.info("Repli sur %s pour les devoirs.", fallback)
fallback_result: list[Homework] | None = None
try: try:
homeworks = self._fetch_homework_source(fallback, target_date) fallback_result = self._fetch_homework_source(fallback, target_date)
except PronoteAuthRotationError: except PronoteAuthRotationError:
raise raise
except Exception as exc: except Exception as exc:
@@ -411,10 +414,12 @@ class PronoteFetcher:
fallback, fallback,
redact_exception(exc), redact_exception(exc),
) )
if fallback_result is None:
raise PipelineCriticalError( raise PipelineCriticalError(
f"Impossible de récupérer les devoirs : les sources {primary}" f"Impossible de récupérer les devoirs : les sources {primary}"
f" et {fallback} ont échoué" f" et {fallback} ont échoué"
) from None ) from None
homeworks = fallback_result
if not homeworks: if not homeworks:
logger.warning( logger.warning(
"Le repli %s pour les devoirs a retourné un résultat vide après " "Le repli %s pour les devoirs a retourné un résultat vide après "
+5 -2
View File
@@ -85,6 +85,7 @@ class JsonTheoreticalAgendaProvider:
self._file_path: str = file_path self._file_path: str = file_path
self._parity_service: WeekParityService | None = parity_service self._parity_service: WeekParityService | None = parity_service
self._holiday_calendar: SchoolHolidayCalendar | None = holiday_calendar self._holiday_calendar: SchoolHolidayCalendar | None = holiday_calendar
load_error: PronoteSyncError | None = None
try: try:
content = Path(file_path).read_text(encoding="utf-8") content = Path(file_path).read_text(encoding="utf-8")
parsed = TheoreticalAgendaFile.model_validate_json(content) parsed = TheoreticalAgendaFile.model_validate_json(content)
@@ -94,9 +95,11 @@ class JsonTheoreticalAgendaProvider:
redact_secrets(str(file_path)), redact_secrets(str(file_path)),
redact_exception(exc), redact_exception(exc),
) )
raise PronoteSyncError( load_error = PronoteSyncError(
f"Le fichier d'agenda théorique est invalide : {redact_secrets(str(file_path))}" 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 self._lessons: tuple[TheoreticalLessonEntry, ...] = parsed.lessons
if self._parity_service is None and any( if self._parity_service is None and any(
entry.week in ("even", "odd") for entry in self._lessons entry.week in ("even", "odd") for entry in self._lessons
+5 -2
View File
@@ -77,6 +77,7 @@ class SchoolHolidayCalendar:
raise PronoteSyncError( raise PronoteSyncError(
f"Le fichier de vacances scolaires est introuvable : {redact_secrets(str(path))}" f"Le fichier de vacances scolaires est introuvable : {redact_secrets(str(path))}"
) from None ) from None
load_error: PronoteSyncError | None = None
try: try:
data: Any = json.loads(path.read_text(encoding="utf-8")) data: Any = json.loads(path.read_text(encoding="utf-8"))
file_model: SchoolHolidayFile = SchoolHolidayFile.model_validate(data) file_model: SchoolHolidayFile = SchoolHolidayFile.model_validate(data)
@@ -86,9 +87,11 @@ class SchoolHolidayCalendar:
redact_secrets(str(path)), redact_secrets(str(path)),
redact_exception(exc), redact_exception(exc),
) )
raise PronoteSyncError( load_error = PronoteSyncError(
f"Le fichier de vacances scolaires est invalide : {redact_secrets(str(path))}" 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 self._periods = file_model.periods
def is_holiday(self, target_date: date) -> bool: def is_holiday(self, target_date: date) -> bool:
+7 -30
View File
@@ -26,12 +26,12 @@ _TEXT_SUFFIXES = frozenset(
{".conf", ".ini", ".json", ".md", ".py", ".service", ".timer", ".toml", ".txt", ".yaml", ".yml"} {".conf", ".ini", ".json", ".md", ".py", ".service", ".timer", ".toml", ".txt", ".yaml", ".yml"}
) )
_LITERAL_SECRET_RE = re.compile( _LITERAL_SECRET_RE = re.compile(
r"(?ix)\b[a-z0-9_]*(?:api[_-]?key|access[_-]?token|auth(?:orization)?|icalsecurise|password|pin|secret|token)" r"(?ix)\b[a-z0-9_]*(?:api[_-]?key|access[_-]?token|auth(?:orization)?|icalsecurise|password|secret|token)"
r"\s*[:=]\s*['\"](?P<value>[^'\"\r\n]{3,})['\"]" r"\s*[:=]\s*['\"][^'\"\r\n]{3,}['\"]"
) )
_UNQUOTED_SECRET_RE = re.compile( _UNQUOTED_SECRET_RE = re.compile(
r"(?ix)\b[a-z0-9_]*(?:api[_-]?key|access[_-]?token|auth(?:orization)?|icalsecurise|password|pin|secret|token)" r"(?ix)\b[a-z0-9_]*(?:api[_-]?key|access[_-]?token|auth(?:orization)?|icalsecurise|password|secret|token)"
r"\s*[:=]\s*(?P<value>[a-z0-9][a-z0-9._~+/-]{2,})" r"\s*[:=]\s*[a-z0-9][a-z0-9._~+/-]{2,}"
) )
_URL_SECRET_RE = re.compile( _URL_SECRET_RE = re.compile(
r"(?ix)[?&](?:api[_-]?key|access[_-]?token|auth(?:orization)?|icalsecurise|password|secret|token)" r"(?ix)[?&](?:api[_-]?key|access[_-]?token|auth(?:orization)?|icalsecurise|password|secret|token)"
@@ -47,13 +47,6 @@ _URL_PLACEHOLDER_RE = re.compile(
r")$" r")$"
) )
_EXTRA_NAMES = frozenset({"pronote_sync"}) _EXTRA_NAMES = frozenset({"pronote_sync"})
_ASSIGNMENT_PLACEHOLDER_RE = re.compile(
r"(?ix)^(?:"
r"<(?:pin|secret|valeur|value|token|jeton)>|"
r"(?:change|replace|your)[_-]?(?:me|here|value|valeur|pin|password|secret)|"
r"(?:placeholder|example|local-not-required)"
r")$"
)
@dataclass(frozen=True) @dataclass(frozen=True)
@@ -75,16 +68,6 @@ CommandRunner = Callable[..., subprocess.CompletedProcess[str]]
ContentProvider = Callable[[Path], str | None] ContentProvider = Callable[[Path], str | None]
def _is_assignment_placeholder(value: str) -> bool:
"""Indique si une valeur d'affectation est un placeholder documentaire.
:param value: Valeur extraite d'une affectation sensible.
:return: ``True`` si la valeur ne représente pas un secret réel.
:rtype: bool
"""
return _ASSIGNMENT_PLACEHOLDER_RE.fullmatch(value.strip()) is not None
def _is_candidate(path: Path) -> bool: def _is_candidate(path: Path) -> bool:
"""Indique si un chemin peut être analysé comme fichier texte. """Indique si un chemin peut être analysé comme fichier texte.
@@ -205,15 +188,9 @@ def find_secrets(
for number, line in enumerate(content.splitlines(), start=1): for number, line in enumerate(content.splitlines(), start=1):
if _ALLOWLIST_MARKER in line: if _ALLOWLIST_MARKER in line:
continue continue
literal_match = _LITERAL_SECRET_RE.search(line) is_literal_secret = _LITERAL_SECRET_RE.search(line) or (
unquoted_match = ( relative_path.suffix in _UNQUOTED_CONFIG_SUFFIXES
_UNQUOTED_SECRET_RE.search(line) and _UNQUOTED_SECRET_RE.search(line)
if relative_path.suffix in _UNQUOTED_CONFIG_SUFFIXES
else None
)
is_literal_secret = any(
match is not None and not _is_assignment_placeholder(match.group("value"))
for match in (literal_match, unquoted_match)
) )
if is_literal_secret: if is_literal_secret:
findings.append(SecretFinding(relative_path, number, "affectation-litterale")) findings.append(SecretFinding(relative_path, number, "affectation-litterale"))
-52
View File
@@ -236,58 +236,6 @@ def test_main_detects_prefixed_secret_assignment(
assert sentinel not in output assert sentinel not in output
def test_main_detects_pronote_pin_assignments_without_disclosing_value(
secret_checker: ModuleType, tmp_path: Path, capsys: CaptureFixture[str]
) -> None:
"""Détecte les PIN Pronote littéraux et non quotés sans afficher leur valeur.
:param secret_checker: Module du script sous test.
:param tmp_path: Répertoire temporaire représentant un dépôt.
:param capsys: Fixture de capture de sortie.
:return: None
"""
literal_pin = "pin-literal-sentinel"
unquoted_pin = "pin-unquoted-sentinel"
(tmp_path / "settings.py").write_text(f'PRONOTE_QR_PIN = "{literal_pin}"\n', encoding="utf-8")
(tmp_path / "settings.yaml").write_text(
f"PRONOTE_ACCOUNT_PIN: {unquoted_pin}\n", encoding="utf-8"
)
assert secret_checker.main([], root=tmp_path) == 1
output = capsys.readouterr().out
assert "settings.py:1" in output
assert "settings.yaml:1" in output
assert literal_pin not in output
assert unquoted_pin not in output
@pytest.mark.parametrize(
"line",
[
'PRONOTE_QR_PIN = "<valeur>"',
"# PRONOTE_ACCOUNT_PIN doit rester dans le fichier d'environnement local",
],
)
def test_main_ignores_pronote_pin_placeholders_and_descriptions(
secret_checker: ModuleType,
tmp_path: Path,
capsys: CaptureFixture[str],
line: str,
) -> None:
"""Ignore les placeholders et descriptions de PIN sans affectation réelle.
:param secret_checker: Module du script sous test.
:param tmp_path: Répertoire temporaire représentant un dépôt.
:param capsys: Fixture de capture de sortie.
:param line: Ligne documentaire à analyser.
:return: None
"""
(tmp_path / "guide.py").write_text(line + "\n", encoding="utf-8")
assert secret_checker.main([], root=tmp_path) == 0
assert "OK:" in capsys.readouterr().out
def test_main_detects_short_secret_assignment( def test_main_detects_short_secret_assignment(
secret_checker: ModuleType, tmp_path: Path, capsys: CaptureFixture[str] secret_checker: ModuleType, tmp_path: Path, capsys: CaptureFixture[str]
) -> None: ) -> None:
+83
View File
@@ -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 == []