diff --git a/pronote_sync/errors.py b/pronote_sync/errors.py index 0fe1e74..c46fc14 100644 --- a/pronote_sync/errors.py +++ b/pronote_sync/errors.py @@ -38,6 +38,21 @@ class PronoteAuthRotationError(PronoteSyncError): super().__init__(message) +class PronoteAuthStateLockError(PronoteSyncError): + """Erreur levée lorsqu'un autre processus détient l'état d'authentification. + + Cette erreur indique qu'une opération QR code / token concurrente est en + cours. Son message ne contient ni chemin local sensible ni credential. + """ + + def __init__(self, message: str) -> None: + """Initialise l'erreur de contention du verrou d'état. + + :param message: Message actionnable expurgé décrivant la contention. + """ + super().__init__(message) + + class ErrorSeverity(StrEnum): """Niveau de gravité d'une erreur produite par le pipeline.""" diff --git a/pronote_sync/sources/pronote/auth_state.py b/pronote_sync/sources/pronote/auth_state.py index e70d9df..99153d1 100644 --- a/pronote_sync/sources/pronote/auth_state.py +++ b/pronote_sync/sources/pronote/auth_state.py @@ -14,10 +14,13 @@ from __future__ import annotations import json import logging import os +from collections.abc import Generator +from contextlib import contextmanager +from fcntl import LOCK_EX, LOCK_NB, LOCK_UN, flock from pathlib import Path from typing import Any -from pronote_sync.errors import PronoteSyncError +from pronote_sync.errors import PronoteAuthStateLockError, PronoteSyncError from pronote_sync.utils.redaction import redact_exception, redact_secrets logger = logging.getLogger(__name__) @@ -108,6 +111,69 @@ class PronoteAuthState: credentials[key] = value return credentials + @contextmanager + def lock(self) -> Generator[None]: + """Protège une opération d'état par un verrou POSIX non bloquant. + + Le verrou est conservé dans le fichier frère ``.lock`` afin + de survivre à l'écriture atomique du fichier d'état. Le fichier de + verrou reste présent après libération et est créé en ``0600`` pour ne + pas élargir l'accès aux métadonnées de l'état sensible. + + :return: Un gestionnaire de contexte qui tient le verrou exclusif. + :rtype: collections.abc.Generator[None, None, None] + :raises PronoteAuthStateLockError: Si un autre processus détient déjà + le verrou ou si son acquisition échoue. + """ + lock_file = self._state_file.with_name(f"{self._state_file.name}.lock") + descriptor: int | None = None + try: + descriptor = os.open( + str(lock_file), + os.O_RDWR | os.O_CREAT | os.O_NOFOLLOW, + 0o600, + ) + os.fchmod(descriptor, 0o600) + except OSError: + logger.error("Impossible d'ouvrir le verrou d'état d'authentification Pronote.") + if descriptor is not None: + os.close(descriptor) + + if descriptor is None: + raise PronoteAuthStateLockError( + "Impossible d'acquérir le verrou d'état d'authentification Pronote." + ) from None + + is_contended = False + lock_acquisition_failed = False + try: + flock(descriptor, LOCK_EX | LOCK_NB) + except BlockingIOError: + is_contended = True + except OSError: + logger.error("Impossible d'acquérir le verrou d'état d'authentification Pronote.") + os.close(descriptor) + lock_acquisition_failed = True + + if lock_acquisition_failed: + raise PronoteAuthStateLockError( + "Impossible d'acquérir le verrou d'état d'authentification Pronote." + ) from None + + if is_contended: + os.close(descriptor) + raise PronoteAuthStateLockError( + "Une autre opération d'authentification Pronote est déjà en cours." + ) + + try: + yield + finally: + try: + flock(descriptor, LOCK_UN) + finally: + os.close(descriptor) + def save(self, credentials: dict[str, str]) -> None: """Sauvegarde les credentials dans le fichier d'état, de manière atomique. diff --git a/tests/unit/test_pronote_auth_state.py b/tests/unit/test_pronote_auth_state.py index fcc1e30..6fa95d3 100644 --- a/tests/unit/test_pronote_auth_state.py +++ b/tests/unit/test_pronote_auth_state.py @@ -17,10 +17,12 @@ from __future__ import annotations import json import logging import os +from fcntl import LOCK_EX, LOCK_NB, LOCK_UN, flock from pathlib import Path import pytest +from pronote_sync.errors import PronoteAuthStateLockError from pronote_sync.sources.pronote.auth_state import PronoteAuthState @@ -206,3 +208,76 @@ def test_no_credentials_in_logs(tmp_path: Path, caplog: pytest.LogCaptureFixture assert "SENTINEL_USER_ZZZ" not in caplog.text assert "SENTINEL_PASSWORD_ZZZ" not in caplog.text assert "SENTINEL_UUID_ZZZ" not in caplog.text + + +def test_lock_rejects_concurrent_access_with_a_redacted_dedicated_error(tmp_path: Path) -> None: + """Vérifie qu'un verrou concurrent échoue immédiatement sans fuite interne. + + :param tmp_path: Fixture pytest pour un répertoire temporaire. + :return: None + """ + state_file = tmp_path / ".pronote_auth_state.json" + state = PronoteAuthState(state_file) + competing_state = PronoteAuthState(state_file) + + with state.lock(): + assert state_file.with_name(f"{state_file.name}.lock").exists() + with pytest.raises(PronoteAuthStateLockError) as exc_info: + with competing_state.lock(): + pass + + assert "BlockingIOError" not in str(exc_info.value) + assert exc_info.value.__cause__ is None + assert exc_info.value.__context__ is None + + +def test_lock_open_failure_does_not_log_sensitive_lock_path( + tmp_path: Path, caplog: pytest.LogCaptureFixture, monkeypatch: pytest.MonkeyPatch +) -> None: + """Vérifie qu'un échec d'ouverture du verrou ne divulgue pas son chemin. + + :param tmp_path: Fixture pytest pour un répertoire temporaire. + :param caplog: Fixture pytest pour capturer les logs. + :param monkeypatch: Fixture pytest pour remplacer l'ouverture du verrou. + :return: None + """ + sentinel_path = "/SENTINEL_LOCK_PATH_ZZZ/.pronote_auth_state.json.lock" + + def raise_lock_open_error(*args: object, **kwargs: object) -> int: + """Simule un refus d'ouverture portant un chemin sensible.""" + del args, kwargs + raise OSError(13, "Permission denied", sentinel_path) + + monkeypatch.setattr(os, "open", raise_lock_open_error) + + with caplog.at_level(logging.ERROR): + with pytest.raises(PronoteAuthStateLockError) as exc_info: + with PronoteAuthState(tmp_path / ".pronote_auth_state.json").lock(): + pass + + assert "Impossible d'ouvrir le verrou d'état d'authentification Pronote" in caplog.text + assert "SENTINEL_LOCK_PATH_ZZZ" not in caplog.text + assert exc_info.value.__cause__ is None + assert exc_info.value.__context__ is None + + +def test_lock_is_released_when_the_protected_operation_raises(tmp_path: Path) -> None: + """Vérifie que le verrou est libéré même si le bloc protégé échoue. + + :param tmp_path: Fixture pytest pour un répertoire temporaire. + :return: None + """ + state_file = tmp_path / ".pronote_auth_state.json" + lock_file = state_file.with_name(f"{state_file.name}.lock") + state = PronoteAuthState(state_file) + + with pytest.raises(RuntimeError, match="échec simulé"): + with state.lock(): + raise RuntimeError("échec simulé") + + descriptor = os.open(lock_file, os.O_RDWR) + try: + flock(descriptor, LOCK_EX | LOCK_NB) + flock(descriptor, LOCK_UN) + finally: + os.close(descriptor)