college-infos/pronote_sync/sync/caldav.py

"""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()