fix: add QR token state lock primitive

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 ``<state_file>.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.