"""Passerelle d'accès au calendrier CalDAV. Ce module fournit :class:`CalDAVGateway`, une passerelle qui isole la bibliothèque ``caldav`` du reste du pipeline de synchronisation. Elle gère la connexion au serveur CalDAV, la résolution du calendrier de destination, la liste des événements gérés par l'outil, ainsi que l'écriture et la suppression d'événements. La passerelle applique des contraintes de sécurité strictes : le mot de passe n'est extrait de son ``SecretStr`` que localement, au moment de créer le client, et aucun secret (mot de passe, URL brute) n'est conservé sur l'instance après la connexion. Toute exception de la bibliothèque ``caldav`` est interceptée puis re-levée sous la forme d'une :class:`~pronote_sync.errors.PronoteSyncError` — sans chaînage — dont le message ne contient aucune donnée sensible. """ from __future__ import annotations import logging from collections.abc import Callable from datetime import datetime from typing import Any, cast from urllib.parse import urlparse import caldav from caldav.lib.error import NotFoundError from icalendar import Component from pydantic import SecretStr from pronote_sync.config.settings import CalDAVSettings from pronote_sync.errors import PronoteSyncError from pronote_sync.sync.serialization import MANAGED_PROPERTY, MANAGED_VALUE from pronote_sync.utils.redaction import redact_exception, redact_secrets, redact_url from pronote_sync.utils.uid import normalize_pronote_uid logger = logging.getLogger(__name__) class CalDAVGateway: """Passerelle d'accès au calendrier CalDAV, isolant la bibliothèque caldav. La passerelle gère la connexion, la résolution du calendrier de destination, la récupération des événements portant le marqueur de gestion (:data:`MANAGED_PROPERTY`), leur écriture et leur suppression. Seuls les paramètres non sensibles nécessaires (``calendar_path``, ``username``) ainsi que l'URL rédigée sont mémorisés sur l'instance ; le mot de passe et l'URL brute ne sont jamais conservés en clair. L'usage typique se fait via le gestionnaire de contexte :: with CalDAVGateway(settings) as gateway: gateway.upsert_event(vcalendar_text, uid) """ def __init__( self, settings: CalDAVSettings, client_factory: Callable[..., Any] | None = None, ) -> None: """Initialise la passerelle avec la configuration CalDAV. Extrait uniquement les paramètres non sensibles nécessaires (``calendar_path``, ``username``) ainsi que l'URL rédigée pour la journalisation. Le mot de passe reste encapsulé dans son ``SecretStr`` et n'est jamais stocké en clair sur l'instance. :param settings: Configuration CalDAV (url, username, password, calendar_path). :param client_factory: Appelable optionnel créant une instance ``DAVClient`` (permet l'injection de dépendances en test). Si ``None``, utilise ``caldav.DAVClient``. """ # ``cast`` nécessaire : mypy ne résout pas le ré-export du module # ``caldav`` (le type de ``caldav.DAVClient`` est vu comme ``object``). self._client_factory: Callable[..., Any] = ( client_factory if client_factory is not None else cast(Callable[..., Any], caldav.DAVClient) ) self._calendar_path: str = settings.calendar_path self._redacted_url: str | None = ( redact_url(settings.url.get_secret_value()) if settings.url else None ) self._username: str | None = settings.username self._url_secret: SecretStr | None = settings.url self._password_secret: SecretStr | None = settings.password self._client: Any = None self._calendar: Any = None def _resolve_calendar(self) -> Any: """Résout le calendrier cible via la découverte CalDAV. Interroge le principal CalDAV puis sa liste de calendriers, et sélectionne celui dont le chemin d'URL correspond au ``calendar_path`` configuré à la frontière d'un composant de chemin (barres obliques finales ignorées, préfixe ``/`` garanti par la normalisation). :return: Le calendrier CalDAV correspondant au chemin configuré. :rtype: Any :raises PronoteSyncError: Si aucun calendrier ne correspond ou si plusieurs calendriers correspondent au chemin configuré. """ principal = self._client.principal() calendars = principal.calendars() normalized_path = self._calendar_path.strip("/") matches: list[Any] = [] for cal in calendars: cal_url = str(cal.url) if hasattr(cal, "url") and cal.url else "" cal_path = urlparse(cal_url).path.strip("/") if cal_path == normalized_path or cal_path.endswith(f"/{normalized_path}"): matches.append(cal) if len(matches) == 0: raise PronoteSyncError( f"Calendrier CalDAV introuvable : {self._redacted_url}" ) from None if len(matches) > 1: raise PronoteSyncError( f"Calendrier CalDAV ambigu : plusieurs calendriers " f"correspondent à '{self._calendar_path}'" ) from None return matches[0] def connect(self) -> None: """Établit la connexion au serveur CalDAV et résout le calendrier. Le mot de passe est extrait de son ``SecretStr`` uniquement pour la création du ``DAVClient``, en variable locale, puis abandonné. Le calendrier cible est résolu par découverte (:meth:`_resolve_calendar`) plutôt que par concaténation d'URL. Toute exception de la bibliothèque ``caldav`` est interceptée, journalisée avec :func:`redact_exception` et re-levée en :class:`PronoteSyncError` — hors du bloc ``except``, afin que ``__context__`` ne retienne aucune exception brute — sans chaînage ni donnée sensible. Les erreurs de résolution du calendrier (message « introuvable » ou « ambigu ») sont propagées telles quelles. :raises PronoteSyncError: Si la configuration est incomplète ou si la connexion au serveur CalDAV échoue. """ if self._url_secret is None or self._username is None or self._password_secret is None: raise PronoteSyncError( "Configuration CalDAV incomplète : url, username et password sont requis" ) from None raw_url = self._url_secret.get_secret_value() password = self._password_secret.get_secret_value() error_msg: str | None = None try: self._client = self._client_factory( url=raw_url, username=self._username, password=password ) self._calendar = self._resolve_calendar() except PronoteSyncError: raise except Exception as exc: error_msg = redact_exception(exc) logger.error("Échec de la connexion CalDAV : %s", error_msg) if error_msg is not None: raise PronoteSyncError(f"Échec de la connexion CalDAV : {self._redacted_url}") from None def list_managed_events(self, start: datetime, end: datetime) -> list[tuple[str, str, Any]]: """Liste les événements gérés par l'outil dans la fenêtre donnée. Interroge le serveur CalDAV sur la fenêtre ``[start, end]`` et ne conserve que les VEVENT portant le marqueur de gestion (:data:`MANAGED_PROPERTY` avec la valeur :data:`MANAGED_VALUE`). Pour chaque VEVENT, l'UID brut tel que stocké sur le serveur est conservé ainsi que sa forme canonique obtenue via :func:`~pronote_sync.utils.uid.normalize_pronote_uid` — la même normalisation que celle appliquée aux événements locaux — afin que le planificateur puisse apparier les événements distants suffixés aux événements Pronote normalisés. :param start: Début de la fenêtre de recherche. :param end: Fin de la fenêtre de recherche. :return: Triplets ``(raw_uid, canonical_uid, vevent)`` pour chaque événement géré trouvé. :rtype: list[tuple[str, str, Any]] :raises PronoteSyncError: Si la passerelle n'est pas connectée ou si la récupération échoue. """ if self._calendar is None: raise PronoteSyncError("Passerelle CalDAV non connectée") from None result: list[tuple[str, str, Any]] = [] error_msg: str | None = None try: events = self._calendar.search(start=start, end=end, event=True, expand=True) for event in events: component: Component = event.icalendar_component for vevent in component.walk("VEVENT"): managed = vevent.get(MANAGED_PROPERTY) if managed is not None and str(managed) == MANAGED_VALUE: raw_uid = str(vevent.get("UID")) canonical_uid = normalize_pronote_uid(raw_uid) result.append((raw_uid, canonical_uid, vevent)) except Exception as exc: error_msg = redact_exception(exc) logger.error("Échec de la récupération des événements CalDAV : %s", error_msg) if error_msg is not None: raise PronoteSyncError( f"Échec de la récupération des événements CalDAV : {self._redacted_url}" ) from None return result def _is_managed_event(self, event: Any) -> bool: """Détermine si un événement distant est géré par pronote-sync. Vérifie la présence du marqueur de gestion (:data:`MANAGED_PROPERTY` avec la valeur :data:`MANAGED_VALUE`) sur au moins un des composants VEVENT de l'événement, selon le même motif que :meth:`list_managed_events`. :param event: Objet événement distant exposant la propriété ``icalendar_component`` retournant un ``icalendar.Calendar``. :return: ``True`` si l'événement porte le marqueur de gestion, ``False`` sinon. :rtype: bool """ component: Component = event.icalendar_component for vevent in component.walk("VEVENT"): managed = vevent.get(MANAGED_PROPERTY) if managed is not None and str(managed) == MANAGED_VALUE: return True return False def upsert_event(self, vcalendar_text: str, uid: str) -> None: """Crée ou met à jour un événement CalDAV identifié par son UID. Recherche d'abord l'événement existant par UID via ``get_event_by_uid`` : s'il est introuvable (``NotFoundError``), un nouvel événement est créé via ``add_event``. S'il existe, son contenu est remplacé puis sauvegardé — uniquement si l'événement est géré par l'outil (marqueur :data:`MANAGED_PROPERTY` avec la valeur :data:`MANAGED_VALUE`). Un événement existant non géré provoque une :class:`PronoteSyncError` explicite et n'est jamais modifié. Toute autre exception est journalisée avec :func:`redact_exception` puis re-levée en :class:`PronoteSyncError` — sans chaînage ni donnée sensible. :param vcalendar_text: Document iCalendar complet (VCALENDAR + VEVENT). :param uid: UID stable de l'événement à créer ou mettre à jour. :raises PronoteSyncError: Si la passerelle n'est pas connectée, si l'événement distant n'est pas géré par l'outil, ou si l'opération échoue. """ if self._calendar is None: raise PronoteSyncError("Passerelle CalDAV non connectée") from None error_msg: str | None = None try: try: event = self._calendar.get_event_by_uid(uid) except NotFoundError: self._calendar.add_event(ical=vcalendar_text) else: if not self._is_managed_event(event): raise PronoteSyncError( "Conflit d'UID : l'événement distant n'est pas géré par pronote-sync" ) from None event.data = vcalendar_text event.save() except PronoteSyncError: raise except Exception as exc: error_msg = redact_exception(exc) logger.error( "Échec de l'écriture d'un événement CalDAV (uid=%s) : %s", redact_secrets(uid), error_msg, ) if error_msg is not None: raise PronoteSyncError( f"Échec de l'écriture d'un événement CalDAV : {self._redacted_url}" ) from None def delete_event(self, uid: str) -> None: """Supprime un événement du calendrier, identifié par son UID. Récupère l'événement distant via ``get_event_by_uid`` : s'il est introuvable (``NotFoundError``), la suppression est un succès idempotent et la méthode retourne silencieusement. S'il existe, il n'est supprimé que s'il est géré par l'outil (marqueur :data:`MANAGED_PROPERTY` avec la valeur :data:`MANAGED_VALUE`) ; un événement non géré est laissé intact, un avertissement est journalisé et la méthode retourne sans erreur. Toute autre exception est journalisée avec :func:`redact_exception` et re-levée en :class:`PronoteSyncError` — sans chaînage ni donnée sensible. :param uid: Identifiant UID de l'événement à supprimer. :raises PronoteSyncError: Si la passerelle n'est pas connectée ou si la suppression échoue. """ if self._calendar is None: raise PronoteSyncError("Passerelle CalDAV non connectée") from None error_msg: str | None = None try: event = self._calendar.get_event_by_uid(uid) if not self._is_managed_event(event): logger.warning( "Suppression refusée : l'événement distant UID=%s n'est pas géré par " "pronote-sync", redact_secrets(uid), ) return event.delete() except NotFoundError: return except Exception as exc: error_msg = redact_exception(exc) logger.error( "Échec de la suppression d'un événement CalDAV (uid=%s) : %s", redact_secrets(uid), error_msg, ) if error_msg is not None: raise PronoteSyncError( f"Échec de la suppression d'un événement CalDAV : {self._redacted_url}" ) from None def close(self) -> None: """Libère les ressources : client, calendrier et secrets. Réinitialise le client, le calendrier et les ``SecretStr`` conservés afin de ne laisser aucune référence à des données sensibles sur l'instance. """ self._client = None self._calendar = None self._url_secret = None self._password_secret = None def __enter__(self) -> CalDAVGateway: """Entre dans le contexte en établissant la connexion. :return: La passerelle connectée. :rtype: CalDAVGateway :raises PronoteSyncError: Si la connexion échoue. """ self.connect() return self def __exit__(self, exc_type: Any, exc_val: Any, exc_tb: Any) -> None: """Quitte le contexte en libérant les ressources. Les exceptions éventuellement en cours ne sont pas interceptées et continuent leur propagation normale. :param exc_type: Type de l'exception en cours, le cas échéant. :param exc_val: Instance de l'exception en cours, le cas échéant. :param exc_tb: Traceback de l'exception en cours, le cas échéant. """ self.close()