222 lines
9.5 KiB
Python
222 lines
9.5 KiB
Python
"""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": "<token>", "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()
|