"""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. :param persistence_enabled: Si ``False``, charge l'état existant mais ne modifie jamais le fichier d'état. ``True`` par défaut. """ def __init__( self, state_file: Path | str = ".pronote_auth_state.json", persistence_enabled: bool = True, ) -> 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. :param persistence_enabled: Si ``False``, charge l'état existant mais désactive toutes les écritures ou suppressions sur disque. ``True`` par défaut. """ self._state_file = Path(state_file) self._persistence_enabled = persistence_enabled self._in_memory_credentials: dict[str, str] | None = None self._in_memory_state_cleared = False 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 self._in_memory_state_cleared: return None if self._in_memory_credentials is not None: return self._in_memory_credentials.copy() 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. """ if not self._persistence_enabled: self._in_memory_credentials = credentials.copy() self._in_memory_state_cleared = False return 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. Lorsque la persistance est désactivée, elle efface uniquement les credentials conservés en mémoire et ne modifie jamais le fichier d'état. :raises OSError: Si la suppression du fichier existant échoue. """ self._in_memory_credentials = None if not self._persistence_enabled: self._in_memory_state_cleared = True return 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()