diff --git a/.env.example b/.env.example index 0b6ed02..b268642 100644 --- a/.env.example +++ b/.env.example @@ -50,7 +50,11 @@ XMPP_HOST=example.com XMPP_PORT=5222 XMPP_TO=parent@example.com XMPP_RESOURCE=pronote-sync -XMPP_USE_TLS=true +# Mode TLS : "starttls" (port 5222, défaut), "direct" (port 5223) ou "disabled" (localhost uniquement) +XMPP_TLS_MODE=starttls +# XMPP_USE_TLS=true/false est obsolète mais encore supporté (alias de XMPP_TLS_MODE) +XMPP_CONNECT_TIMEOUT=15 +XMPP_CLEANUP_TIMEOUT=10 XMPP_TIMEOUT=30 # --- IA (optionnelle) --- diff --git a/.secrets.baseline b/.secrets.baseline index 8934a65..24b06df 100644 --- a/.secrets.baseline +++ b/.secrets.baseline @@ -139,9 +139,8 @@ "type": "Hex High Entropy String", "filename": "GUIDE_DEV_PYTHON.md", "hashed_secret": "90bd1b48e958257948487b90bee080ba5ed00caa", - "is_verified": true, - "line_number": 5085, - "is_secret": false + "is_verified": false, + "line_number": 5103 } ], "tests/unit/test_caldav_gateway.py": [ @@ -175,7 +174,16 @@ "is_verified": false, "line_number": 28 } + ], + "tests/unit/test_xmpp_channel.py": [ + { + "type": "Secret Keyword", + "filename": "tests/unit/test_xmpp_channel.py", + "hashed_secret": "b01bcded8826562355b3e32a8b31e214ff533511", + "is_verified": false, + "line_number": 1011 + } ] }, - "generated_at": "2026-09-11T09:57:08Z" + "generated_at": "2026-09-11T14:59:15Z" } diff --git a/GUIDE_DEV_PYTHON.md b/GUIDE_DEV_PYTHON.md index f2a0ca8..37f41f5 100644 --- a/GUIDE_DEV_PYTHON.md +++ b/GUIDE_DEV_PYTHON.md @@ -275,7 +275,8 @@ Le projet utilise **`pydantic-settings`** pour valider et charger la configurati > ⚠️ **Décision d'implémentation** : > `XMPP_RECIPIENT` a été renommé en `XMPP_TO` dans l'implémentation (aligné avec §10.2.1). -> Des variables XMPP supplémentaires ont été ajoutées : `XMPP_ENABLED`, `XMPP_HOST`, `XMPP_PORT`, `XMPP_RESOURCE`, `XMPP_USE_TLS`, `XMPP_TIMEOUT`. +> Des variables XMPP supplémentaires ont été ajoutées : `XMPP_ENABLED`, `XMPP_HOST`, `XMPP_PORT`, `XMPP_RESOURCE`, `XMPP_TLS_MODE`, `XMPP_TIMEOUT`, `XMPP_CONNECT_TIMEOUT`, `XMPP_CLEANUP_TIMEOUT`. +> `XMPP_USE_TLS` reste supporté mais est **obsolète** (remplacé par `XMPP_TLS_MODE`). > Une section `BLOG_ENABLED` et `BLOG_RSS_URL` a été ajoutée dans `.env.example`. Les variables Pronote sont obligatoires selon les sources activées : @@ -4066,7 +4067,9 @@ Si le besoin évolue (ex: **plusieurs destinataires**), les étapes suivantes so | `XMPP_PORT` | Port XMPP (5222 pour STARTTLS, 5223 pour TLS direct). | `5222` | `int` | ❌ Non | | `XMPP_TO` | Destinataire unique (ex: `parent@exemple.org`). | `None` | `str` | ✅ Oui | | `XMPP_RESOURCE` | Ressource XMPP (ex: `pronote-sync`). | `"pronote-sync"` | `str` | ❌ Non | -| `XMPP_USE_TLS` | Utiliser TLS pour la connexion. | `True` | `bool` | ❌ Non | +| `XMPP_TLS_MODE` | Mode TLS : `starttls` (défaut), `direct` ou `disabled` (localhost uniquement). | `"starttls"` | `Literal` | ❌ Non | +| `XMPP_CONNECT_TIMEOUT` | Timeout de connexion (secondes). | `15` | `float` | ❌ Non | +| `XMPP_CLEANUP_TIMEOUT` | Timeout de nettoyage (secondes). | `10` | `float` | ❌ Non | | `XMPP_TIMEOUT` | Timeout de connexion (secondes). | `30` | `int` | ❌ Non | **⚠️ Notes** : @@ -4074,6 +4077,7 @@ Si le besoin évolue (ex: **plusieurs destinataires**), les étapes suivantes so - **Pas de variables PubSub** : `XMPP_PUBSUB_NODE`, `XMPP_ROOM`, `XMPP_SUBSCRIBERS` **ne doivent pas être introduites** pour l'instant. - **Sécurité** : `XMPP_JID`, `XMPP_PASSWORD` et `XMPP_TO` **ne doivent jamais apparaître** dans les logs, erreurs ou fixtures. - **Standardisation** : `XMPP_TO` est mappé sur le champ `to` dans le modèle Pydantic. +- **`XMPP_USE_TLS` obsolète** : remplacée par `XMPP_TLS_MODE` (`starttls`, `direct` ou `disabled`) ; `XMPP_USE_TLS=true` → `direct` et `XMPP_USE_TLS=false` → `starttls`, avec un `DeprecationWarning`. #### 10.2.2 Exemple de configuration dans `.env` @@ -4086,13 +4090,19 @@ XMPP_HOST=exemple.org XMPP_PORT=5222 XMPP_TO=parent@exemple.org XMPP_RESOURCE=pronote-digest -XMPP_USE_TLS=true +# XMPP_TLS_MODE : "starttls" (défaut), "direct" (port 5223) ou "disabled" (localhost uniquement) +XMPP_TLS_MODE=starttls +# XMPP_USE_TLS=true/false est obsolète mais encore supporté (alias de XMPP_TLS_MODE) +XMPP_CONNECT_TIMEOUT=15 +XMPP_CLEANUP_TIMEOUT=10 XMPP_TIMEOUT=30 ``` #### 10.2.3 Modèle Pydantic pour la configuration XMPP ```python +from typing import Literal + from pydantic import SecretStr, Field from pydantic_settings import BaseSettings, SettingsConfigDict @@ -4118,14 +4128,22 @@ class XmppSettings(BaseSettings): port: int = Field(default=5222, ge=1, le=65535) to: str | None = None resource: str = "pronote-sync" - use_tls: bool = True - timeout: int = Field(default=30, gt=0) + tls_mode: Literal["direct", "starttls", "disabled"] = "starttls" + use_tls: bool | None = Field( + default=None, + deprecated="Utiliser tls_mode à la place (XMPP_USE_TLS obsolète).", + ) + timeout: float = Field(default=30, gt=0) + connect_timeout: float = Field(default=15, gt=0) + cleanup_timeout: float = Field(default=10, gt=0) ``` > **⚠️ Mapping TLS** : -> - `use_tls=True` → **TLS direct** (port 5223, `enable_direct_tls=True`, `enable_starttls=False`). -> - `use_tls=False` → **STARTTLS** (port 5222, `enable_starttls=True`, `enable_direct_tls=False`). -> La validation refuse `use_tls=False` si `host` n'est pas un hôte de boucle locale (`localhost`, `127.0.0.1`, `::1`). +> - `tls_mode="starttls"` (défaut) → **STARTTLS** (port 5222, `enable_starttls=True`, `enable_direct_tls=False`). +> - `tls_mode="direct"` → **TLS direct** (port 5223, `enable_direct_tls=True`, `enable_starttls=False`). +> - `tls_mode="disabled"` → TLS désactivé ; **uniquement** autorisé sur un hôte de boucle locale (`localhost`, `127.0.0.1`, `::1`). +> - `XMPP_USE_TLS` (`use_tls`) est **obsolète** : `use_tls=True` → `tls_mode="direct"` et +> `use_tls=False` → `tls_mode="starttls"` (avec `DeprecationWarning`). --- diff --git a/README.LLM.md b/README.LLM.md index 799d9b7..6d481f3 100644 --- a/README.LLM.md +++ b/README.LLM.md @@ -71,7 +71,8 @@ The following variables can be safely pre-configured in `/etc/pronote-sync/prono - `THEORETICAL_WEEK_ANCHOR_DATE`, `THEORETICAL_WEEK_ANCHOR_TYPE` - **XMPP:** - - `XMPP_ENABLED`, `XMPP_HOST`, `XMPP_PORT`, `XMPP_USE_TLS`, `XMPP_TIMEOUT`, `XMPP_RESOURCE` + - `XMPP_ENABLED`, `XMPP_HOST`, `XMPP_PORT`, `XMPP_TLS_MODE`, `XMPP_TIMEOUT`, `XMPP_CONNECT_TIMEOUT`, `XMPP_CLEANUP_TIMEOUT`, `XMPP_RESOURCE` + - `XMPP_USE_TLS` is deprecated but still supported (aliased to `XMPP_TLS_MODE`) - **AI:** - `AI_ENABLED`, `AI_PROVIDER`, `AI_BASE_URL`, `AI_MODEL`, `AI_ALLOW_INSECURE_HTTP` diff --git a/pronote_sync/channels/xmpp.py b/pronote_sync/channels/xmpp.py index 8fdd6f0..e334ee5 100644 --- a/pronote_sync/channels/xmpp.py +++ b/pronote_sync/channels/xmpp.py @@ -36,6 +36,24 @@ logger = logging.getLogger(__name__) __all__ = ["XmppChannel", "SyncXmppChannel", "XmppMessage"] +async def _cancel_pending( + *futures: asyncio.Future[bool], +) -> None: + """Annule les futures/tâches encore en attente et supprime le bruit. + + À appeler avant chaque retour anticipé de :meth:`XmppChannel.send_async` + afin qu'aucune tentative de connexion ne survive au retour de la méthode. + + :param futures: Futures ou tâches à annuler (les déjà terminées sont + ignorées pour la cancellation mais attendues pour purger l'attente). + :rtype: None + """ + for future in futures: + if not future.done(): + future.cancel() + await asyncio.gather(*futures, return_exceptions=True) + + def _secret_values(settings: XmppSettings) -> tuple[SecretStr | str, ...]: """Rassemble les secrets du canal XMPP pour le masquage des logs. @@ -220,11 +238,14 @@ class XmppChannel: """Exécute le flux asynchrone d'envoi XMPP (U2). Connecte le client ``slixmpp`` avec un hôte et un port explicites, - configure TLS avant la connexion, puis attend l'un des événements - ``session_start``, ``failed_auth`` ou ``disconnected`` sous un - timeout unique avant d'envoyer un message direct ``chat`` au + configure TLS avant la connexion selon ``tls_mode`` (``direct``, + ``starttls`` ou ``disabled``), attend la connexion sous + ``connect_timeout`` puis l'un des événements ``session_start``, + ``failed_auth``, ``connection_failed`` ou ``disconnected`` sous + ``timeout`` avant d'envoyer un message direct ``chat`` au destinataire configuré. La déconnexion est garantie par un bloc - ``try/finally``. Aucun secret n'est journalisé (SEC-XMPP-02). + ``try/finally`` borné par ``cleanup_timeout``. Aucun secret n'est + journalisé (SEC-XMPP-02). :param message: Message final à envoyer. :return: ``True`` si l'envoi a réussi (ou a été simulé en dry-run), @@ -244,55 +265,132 @@ class XmppChannel: logger.warning("Destinataire XMPP manquant.") return False - # Create typed client - client = ClientXMPP( - jid_str, - self.settings.password.get_secret_value() if self.settings.password else "", - ) - - # Configure TLS BEFORE connect - if self.settings.use_tls: - # TLS direct (port 5223 typically) - client.enable_direct_tls = True - client.enable_starttls = False - else: - # STARTTLS (port 5222 typically) - client.enable_starttls = True - client.enable_direct_tls = False - - # Register handlers + # Client typed lazily: the construction is done inside the try block so that + # any error is caught and converted to ``False`` (channel contract: never raise) + client: ClientXMPP | None = None session_future: asyncio.Future[bool] = asyncio.get_event_loop().create_future() + failure_kind = "disconnected" + # Declared before the ``try`` so the exception handlers (CancelledError and + # Exception) can cancel any task still pending from ``asyncio.wait()`` + connect_future: asyncio.Future[bool] | None = None + session_task: asyncio.Future[bool] | None = None def on_session_start(event: object) -> None: if not session_future.done(): session_future.set_result(True) def on_failed_auth(event: object) -> None: + nonlocal failure_kind if not session_future.done(): + failure_kind = "failed_auth" + session_future.set_result(False) + + def on_connection_failed(event: object) -> None: + nonlocal failure_kind + if not session_future.done(): + failure_kind = "connection_failed" session_future.set_result(False) def on_disconnected(event: object) -> None: if not session_future.done(): session_future.set_result(False) - client.add_event_handler("session_start", on_session_start) - client.add_event_handler("failed_auth", on_failed_auth) - client.add_event_handler("disconnected", on_disconnected) - try: + # Create typed client + client = ClientXMPP( + jid_str, + self.settings.password.get_secret_value() if self.settings.password else "", + ) + + # Configure TLS BEFORE connect (canonical tls_mode) + match self.settings.tls_mode: + case "direct": + client.enable_direct_tls = True + client.enable_starttls = False + case "starttls": + client.enable_direct_tls = False + client.enable_starttls = True + case "disabled": + client.enable_direct_tls = False + client.enable_starttls = False + + # Register handlers + client.add_event_handler("session_start", on_session_start) + client.add_event_handler("failed_auth", on_failed_auth) + client.add_event_handler("connection_failed", on_connection_failed) + client.add_event_handler("disconnected", on_disconnected) + # Connect with explicit host and port - connect_future = client.connect(self.settings.host, self.settings.port) - await connect_future # connect() returns a Future, not a coroutine + connect_future = asyncio.ensure_future( + client.connect(self.settings.host, self.settings.port) + ) - # Wait for one of the three events under a single timeout - try: - success = await asyncio.wait_for(session_future, timeout=self.settings.timeout) - except TimeoutError: - logger.warning("Délai d'attente de session XMPP dépassé.") - return False + # Wrap the session future in a task so that cancelling pending tasks + # during the concurrent wait never cancels ``session_future`` itself + async def _await_session() -> bool: + return await session_future - if not success: - logger.warning("Échec d'authentification ou déconnexion XMPP.") + session_task = asyncio.ensure_future(_await_session()) + + # Wait for the connection and the session event concurrently, bounded by + # connect_timeout as the global time limit: a ``connection_failed`` event + # can thus trigger an early return before the connect timeout expires + done, _pending = await asyncio.wait( + {connect_future, session_task}, + timeout=self.settings.connect_timeout, + return_when=asyncio.FIRST_COMPLETED, + ) + + if session_task in done: + if not session_future.result(): + # ``connection_failed``/``failed_auth``/``disconnected`` fired + # before the connection was resolved: immediate failure (fail fast) + if failure_kind == "connection_failed": + logger.warning("Échec de connexion réseau XMPP.") + else: + logger.warning("Échec d'authentification ou déconnexion XMPP.") + await _cancel_pending(connect_future, session_task) + return False + # ``session_start`` fired: the connection succeeded even if the + # connect future is still pending; proceed to send the message + await _cancel_pending(connect_future, session_task) + elif connect_future in done: + # The connection resolved: surface a connect error (redacted) if any + if not connect_future.cancelled(): + connect_exc = connect_future.exception() + if connect_exc is not None and isinstance(connect_exc, Exception): + logger.warning( + "Échec de connexion XMPP : %s", + redact_exception( + connect_exc, + extra_secrets=_secret_values(self.settings), + ), + ) + await _cancel_pending(connect_future, session_task) + return False + + # Connection established: wait for a session event under ``timeout`` + try: + success = await asyncio.wait_for( + asyncio.shield(session_future), timeout=self.settings.timeout + ) + except TimeoutError: + logger.warning("Délai d'attente de session XMPP dépassé.") + await _cancel_pending(connect_future, session_task) + return False + if not success: + if failure_kind == "connection_failed": + logger.warning("Échec de connexion réseau XMPP.") + else: + logger.warning("Échec d'authentification ou déconnexion XMPP.") + await _cancel_pending(connect_future, session_task) + return False + else: + # connect_timeout expired: cancel everything and fail fast + logger.warning( + "Délai de connexion XMPP dépassé (%ss).", self.settings.connect_timeout + ) + await _cancel_pending(connect_future, session_task) return False # Send the message @@ -300,19 +398,39 @@ class XmppChannel: client.send_message(mto=JID(self.settings.to), mbody=formatted, mtype="chat") return True + except asyncio.CancelledError: + # Contrat du canal : toujours retourner un booléen, même en cas + # d'annulation de la tâche appelante (cleanup exécuté par le finally). + logger.debug("Envoi XMPP annulé avant la fin de l'opération.") + pending = [f for f in (connect_future, session_task) if f is not None] + if pending: + await _cancel_pending(*pending) + return False except Exception as exc: - redacted = redact_exception(exc) + redacted = redact_exception(exc, extra_secrets=_secret_values(self.settings)) extra = _secret_values(self.settings) logger.warning("Erreur XMPP: %s", redact_secrets(redacted, extra_secrets=extra)) + pending = [f for f in (connect_future, session_task) if f is not None] + if pending: + await _cancel_pending(*pending) return False finally: - try: - disconnect_future = client.disconnect() - await disconnect_future - except Exception as cleanup_exc: - logger.debug( - "Erreur lors de la déconnexion XMPP: %s", redact_exception(cleanup_exc) - ) + if client is not None: + try: + disconnect_future = client.disconnect() + await asyncio.wait_for(disconnect_future, timeout=self.settings.cleanup_timeout) + except asyncio.CancelledError: + logger.debug("Déconnexion XMPP annulée.") + except TimeoutError: + logger.debug( + "Délai de déconnexion XMPP dépassé (%ss), abandon.", + self.settings.cleanup_timeout, + ) + except Exception as cleanup_exc: + logger.debug( + "Erreur lors de la déconnexion XMPP : %s", + redact_exception(cleanup_exc, extra_secrets=_secret_values(self.settings)), + ) class SyncXmppChannel: @@ -363,7 +481,7 @@ class SyncXmppChannel: try: return asyncio.run(self._channel.send_async(message)) except Exception as exc: - redacted = redact_exception(exc) + redacted = redact_exception(exc, extra_secrets=_secret_values(self.settings)) redacted = redact_secrets(redacted, extra_secrets=_secret_values(self.settings)) logger.warning("XMPP : erreur lors de l'envoi synchrone : %s", redacted) return False diff --git a/pronote_sync/config/settings.py b/pronote_sync/config/settings.py index cde7b24..7964114 100644 --- a/pronote_sync/config/settings.py +++ b/pronote_sync/config/settings.py @@ -8,6 +8,7 @@ depuis les variables d'environnement (préfixées par groupe) et le fichier from __future__ import annotations +import warnings from datetime import date from typing import Literal from urllib.parse import urlparse @@ -18,6 +19,7 @@ from pydantic import ( ValidationInfo, field_serializer, field_validator, + model_validator, ) from pydantic_settings import BaseSettings, SettingsConfigDict @@ -158,10 +160,16 @@ class XmppSettings(BaseSettings): Contraintes de champs : ``port`` est borné entre 1 et 65535 et ``timeout`` doit être strictement positif. - Politique TLS : la désactivation de TLS (``use_tls`` à ``False``) n'est - autorisée que sur un hôte de boucle locale (``localhost``, ``127.0.0.1``, - ``::1``). Dans tout autre cas, une erreur de validation est levée, - indépendamment de l'état du champ ``enabled``. + Politique TLS : le mode ``tls_mode`` détermine la négociation TLS + (``direct``, ``starttls`` ou ``disabled``). Le mode ``disabled`` n'est + autorisé que sur un hôte de boucle locale (``localhost``, ``127.0.0.1``, + ``::1``) ; ``starttls`` et ``direct`` sont permis pour tous les hôtes. + + Compatibilité : le champ historique ``use_tls`` (booléen) est un alias + obsolète ; ``use_tls=True`` mappe vers ``tls_mode="direct"`` et + ``use_tls=False`` vers ``tls_mode="starttls"``, avec un + :pyexc:`DeprecationWarning`. La valeur brute fournie reste lisible via + ``use_tls`` (``None`` si non fournie). """ model_config = SettingsConfigDict( @@ -178,33 +186,61 @@ class XmppSettings(BaseSettings): port: int = Field(default=5222, ge=1, le=65535) to: str | None = None resource: str = "pronote-sync" - use_tls: bool = True - timeout: int = Field(default=30, gt=0) + tls_mode: Literal["direct", "starttls", "disabled"] = "starttls" + use_tls: bool | None = Field( + default=None, + deprecated="Utiliser tls_mode à la place (XMPP_USE_TLS obsolète).", + ) + timeout: float = Field(default=30, gt=0) + connect_timeout: float = Field(default=15, gt=0) + cleanup_timeout: float = Field(default=10, gt=0) - @field_validator("use_tls") + @model_validator(mode="before") @classmethod - def _validate_tls_policy(cls, v: bool, info: ValidationInfo) -> bool: - """Refuse la désactivation de TLS hors des hôtes de boucle locale. + def _migrate_use_tls(cls, data: object) -> object: + """Mappe l'alias obsolète ``use_tls`` vers le mode canonique ``tls_mode``. - La règle s'applique quel que soit l'état du champ ``enabled``. Le - message d'erreur ne contient aucune valeur sensible (``jid``, + ``use_tls=True`` devient ``tls_mode="direct"`` et ``use_tls=False`` + devient ``tls_mode="starttls"`` ; un :pyexc:`DeprecationWarning` est + émis à chaque usage explicite de l'alias. ``tls_mode`` fourni + explicitement prend le pas sur l'alias. + + :param data: Données d'entrée du modèle (dict ou autre). + :return: Données d'entrée avec ``tls_mode`` dérivé de ``use_tls``. + :rtype: object + """ + if not isinstance(data, dict) or "use_tls" not in data: + return data + warnings.warn( + "XMPP_USE_TLS est obsolète : utiliser XMPP_TLS_MODE " + "('direct', 'starttls' ou 'disabled').", + DeprecationWarning, + stacklevel=2, + ) + if data.get("tls_mode") is None: + data["tls_mode"] = "direct" if data["use_tls"] else "starttls" + return data + + @model_validator(mode="after") + def _validate_tls_policy(self) -> XmppSettings: + """Refuse le mode ``disabled`` hors des hôtes de boucle locale. + + La règle s'applique quel que soit l'état du champ ``enabled``. Les + modes ``starttls`` et ``direct`` sont autorisés pour tous les hôtes. + Le message d'erreur ne contient aucune valeur sensible (``jid``, ``password``, ``to``). - :param v: Valeur du champ ``use_tls`` à valider. - :param info: Contexte de validation (accès aux autres champs). - :return: La valeur validée inchangée. - :rtype: bool - :raises ValueError: Si ``use_tls`` est ``False`` et que ``host`` + :return: L'instance validée inchangée. + :rtype: XmppSettings + :raises ValueError: Si ``tls_mode`` est ``disabled`` et que ``host`` n'est pas un hôte de boucle locale. """ - if v is False: - host = info.data.get("host", "") - if host not in _XMPP_LOOPBACK_HOSTS: - raise ValueError( - "TLS désactivé n'est autorisé que sur les hôtes de loopback " - "(localhost, 127.0.0.1, ::1)." - ) from None - return v + if self.tls_mode == "disabled" and self.host not in _XMPP_LOOPBACK_HOSTS: + raise ValueError( + "TLS désactivé n'est autorisé que sur les hôtes de loopback " + "(localhost, 127.0.0.1, ::1)." + ) from None + return self class AISettings(BaseSettings): diff --git a/tests/integration/test_xmpp_integration.py b/tests/integration/test_xmpp_integration.py index dfc2e1c..4f551e0 100644 --- a/tests/integration/test_xmpp_integration.py +++ b/tests/integration/test_xmpp_integration.py @@ -67,7 +67,7 @@ class FakeClientXMPP: :param handler: Fonction gestionnaire. :raises: AssertionError si l'événement n'est pas supporté. """ - if name not in ("session_start", "failed_auth", "disconnected"): + if name not in ("session_start", "failed_auth", "connection_failed", "disconnected"): raise AssertionError(f"Unsupported event: {name}") self.handlers.setdefault(name, []).append(handler) diff --git a/tests/unit/test_xmpp_channel.py b/tests/unit/test_xmpp_channel.py index 958e833..4f8d051 100644 --- a/tests/unit/test_xmpp_channel.py +++ b/tests/unit/test_xmpp_channel.py @@ -17,7 +17,7 @@ from unittest.mock import patch import pytest from pydantic import SecretStr -from pronote_sync.channels.xmpp import XmppChannel, XmppMessage +from pronote_sync.channels.xmpp import SyncXmppChannel, XmppChannel, XmppMessage from pronote_sync.config.settings import XmppSettings from pronote_sync.models.agenda import Lesson, TheoreticalLesson from pronote_sync.models.blog import BlogArticle, ExternalInfo @@ -48,11 +48,15 @@ class FakeClientXMPP: self._should_disconnect_early = False self._host_used: str | None = None self._port_used: int | None = None + # New modes for realistic failure simulation + self._connect_mode: str = "ok" # "ok", "pending", "connection_failed" + self._disconnect_mode: str = "ok" # "ok", "pending" + self._fire_connection_failed: bool = False def add_event_handler( self, name: str, pointer: Callable[..., object], disposable: bool = False ) -> None: - if name not in ("session_start", "failed_auth", "disconnected"): + if name not in ("session_start", "failed_auth", "connection_failed", "disconnected"): raise AssertionError(f"Unsupported event: {name}") self.handlers.setdefault(name, []).append(pointer) @@ -63,10 +67,24 @@ class FakeClientXMPP: self.connected = True self._host_used = host self._port_used = port - # Schedule event handlers to fire after connect returns - loop.call_soon(self._fire_events) - future.set_result(True) - return future + + if self._connect_mode == "pending": + # Never resolves - simulates connection timeout + return future + elif self._connect_mode == "connection_failed": + # Resolves connect future but fires connection_failed event + loop.call_soon(self._fire_connection_failed_event) + future.set_result(True) + return future + else: + # Schedule event handlers to fire after connect returns + loop.call_soon(self._fire_events) + future.set_result(True) + return future + + def _fire_connection_failed_event(self) -> None: + """Fire connection_failed event for testing.""" + self._fire("connection_failed") def _fire_events(self) -> None: if self._should_disconnect_early: @@ -86,8 +104,13 @@ class FakeClientXMPP: loop = asyncio.get_event_loop() future: asyncio.Future[bool] = loop.create_future() self.disconnected = True - future.set_result(True) - return future + + if self._disconnect_mode == "pending": + # Never resolves - simulates cleanup timeout + return future + else: + future.set_result(True) + return future def send_message( self, mto: object, mbody: str | None = None, mtype: str | None = None, **kwargs: object @@ -681,6 +704,339 @@ class TestXmppChannelSend: jid_arg = call_args.args[0] assert jid_arg == "bot@example.com/myresource" + @patch("pronote_sync.channels.xmpp.ClientXMPP", new=FakeClientXMPP) + @pytest.mark.asyncio + async def test_connect_timeout_returns_false(self, caplog: pytest.LogCaptureFixture) -> None: + """Test que connect_timeout retourne False quand connect() ne résout pas. + + :param caplog: Fixture pytest pour capturer les logs. + """ + settings = XmppSettings( + enabled=True, + jid="bot@example.com", + password=SecretStr("secret123"), + host="xmpp.example.com", + port=5222, + to="parent@example.com", + resource="pronote-sync", + tls_mode="starttls", + connect_timeout=0.05, + timeout=30, + cleanup_timeout=0.01, + ) + + class PendingConnectClient(FakeClientXMPP): + def __init__(self, jid: str, password: str) -> None: + super().__init__(jid, password) + self._connect_mode = "pending" + + with patch("pronote_sync.channels.xmpp.ClientXMPP", new=PendingConnectClient): + channel = XmppChannel(settings, dry_run=False) + msg = XmppMessage(target_date=date(2025, 9, 7), synthesis=None, external_info=None) + result = await channel.send_async(msg) + assert result is False + # Vérifier que le log contient "connexion" + logs = caplog.text + assert "connexion" in logs.lower() + + @pytest.mark.asyncio + async def test_cancelled_error_cancels_pending_tasks(self) -> None: + """Test que l'annulation de send_async annule les tâches encore en attente. + + Un ``connect()`` qui ne résout jamais et une annulation de la tâche + appelante doivent entraîner la cancellation de la future de connexion + (idempotence de ``_cancel_pending`` sur les chemins d'exception). + """ + settings = XmppSettings( + enabled=True, + jid="bot@example.com", + password=SecretStr("secret123"), + host="xmpp.example.com", + port=5222, + to="parent@example.com", + resource="pronote-sync", + tls_mode="starttls", + connect_timeout=15, + timeout=30, + cleanup_timeout=0.01, + ) + + class PendingConnectClient(FakeClientXMPP): + """Client dont ``connect()`` retourne une future jamais résolue.""" + + def __init__(self, jid: str, password: str) -> None: + super().__init__(jid, password) + self._connect_mode = "pending" + self.connect_future: asyncio.Future[bool] | None = None + + def connect( + self, host: str | None = None, port: int | None = None + ) -> asyncio.Future[bool]: + future = super().connect(host, port) + self.connect_future = future + return future + + with patch("pronote_sync.channels.xmpp.ClientXMPP", new=PendingConnectClient): + channel = XmppChannel(settings, dry_run=False) + msg = XmppMessage(target_date=date(2025, 9, 7), synthesis=None, external_info=None) + client = PendingConnectClient("bot@example.com", "secret123") + with patch("pronote_sync.channels.xmpp.ClientXMPP", new=lambda j, p: client): + task = asyncio.ensure_future(channel.send_async(msg)) + # Laisse asyncio.wait démarrer et la future de connexion rester en attente + await asyncio.sleep(0.05) + task.cancel() + result = await task + assert result is False + assert client.connect_future is not None + assert client.connect_future.cancelled() + + @patch("pronote_sync.channels.xmpp.ClientXMPP", new=FakeClientXMPP) + @pytest.mark.asyncio + async def test_connection_failed_event_returns_false( + self, caplog: pytest.LogCaptureFixture + ) -> None: + """Test que connection_failed event retourne False rapidement. + + :param caplog: Fixture pytest pour capturer les logs. + """ + settings = XmppSettings( + enabled=True, + jid="bot@example.com", + password=SecretStr("secret123"), + host="xmpp.example.com", + port=5222, + to="parent@example.com", + resource="pronote-sync", + tls_mode="starttls", + connect_timeout=15, + timeout=30, + cleanup_timeout=0.01, + ) + + class ConnectionFailedClient(FakeClientXMPP): + def __init__(self, jid: str, password: str) -> None: + super().__init__(jid, password) + self._connect_mode = "connection_failed" + + with patch("pronote_sync.channels.xmpp.ClientXMPP", new=ConnectionFailedClient): + channel = XmppChannel(settings, dry_run=False) + msg = XmppMessage(target_date=date(2025, 9, 7), synthesis=None, external_info=None) + result = await channel.send_async(msg) + assert result is False + # Vérifier que le log contient une mention d'échec réseau + logs = caplog.text + assert "réseau" in logs.lower() or "connexion" in logs.lower() + + @patch("pronote_sync.channels.xmpp.ClientXMPP", new=FakeClientXMPP) + @pytest.mark.asyncio + async def test_cleanup_timeout_does_not_hang(self, caplog: pytest.LogCaptureFixture) -> None: + """Test que cleanup_timeout ne bloque pas quand disconnect() ne résout pas. + + :param caplog: Fixture pytest pour capturer les logs. + """ + import logging + + caplog.set_level(logging.DEBUG) + + settings = XmppSettings( + enabled=True, + jid="bot@example.com", + password=SecretStr("secret123"), + host="xmpp.example.com", + port=5222, + to="parent@example.com", + resource="pronote-sync", + tls_mode="starttls", + connect_timeout=0.05, + timeout=30, + cleanup_timeout=0.01, + ) + + class HangingDisconnectClient(FakeClientXMPP): + def __init__(self, jid: str, password: str) -> None: + super().__init__(jid, password) + self._disconnect_mode = "pending" + + with patch("pronote_sync.channels.xmpp.ClientXMPP", new=HangingDisconnectClient): + channel = XmppChannel(settings, dry_run=False) + msg = XmppMessage(target_date=date(2025, 9, 7), synthesis=None, external_info=None) + # Should complete quickly despite hanging disconnect + result = await asyncio.wait_for(channel.send_async(msg), timeout=0.5) + assert result is True + # Vérifier que le log contient une mention de timeout de nettoyage + logs = caplog.text + assert "déconnexion" in logs.lower() + + @patch("pronote_sync.channels.xmpp.ClientXMPP", new=FakeClientXMPP) + @pytest.mark.asyncio + async def test_tls_mode_direct_config(self) -> None: + """Test que tls_mode='direct' configure enable_direct_tls=True et enable_starttls=False. + + :return: None + """ + settings = XmppSettings( + enabled=True, + jid="bot@example.com", + password=SecretStr("secret123"), + host="xmpp.example.com", + port=5222, + to="parent@example.com", + resource="pronote-sync", + tls_mode="direct", + timeout=30, + ) + + class InspectClient(FakeClientXMPP): + def __init__(self, jid: str, password: str) -> None: + super().__init__(jid, password) + + with patch("pronote_sync.channels.xmpp.ClientXMPP") as mock_cls: + mock_cls.return_value = InspectClient("bot@example.com", "secret123") + channel = XmppChannel(settings, dry_run=False) + msg = XmppMessage(target_date=date(2025, 9, 7), synthesis=None, external_info=None) + await channel.send_async(msg) + client_instance = mock_cls.return_value + assert client_instance.enable_direct_tls is True + assert client_instance.enable_starttls is False + + @patch("pronote_sync.channels.xmpp.ClientXMPP", new=FakeClientXMPP) + @pytest.mark.asyncio + async def test_tls_mode_starttls_config(self) -> None: + """Test que tls_mode='starttls' configure enable_starttls=True et enable_direct_tls=False. + + :return: None + """ + settings = XmppSettings( + enabled=True, + jid="bot@example.com", + password=SecretStr("secret123"), + host="xmpp.example.com", + port=5222, + to="parent@example.com", + resource="pronote-sync", + tls_mode="starttls", + timeout=30, + ) + + class InspectClient(FakeClientXMPP): + def __init__(self, jid: str, password: str) -> None: + super().__init__(jid, password) + + with patch("pronote_sync.channels.xmpp.ClientXMPP") as mock_cls: + mock_cls.return_value = InspectClient("bot@example.com", "secret123") + channel = XmppChannel(settings, dry_run=False) + msg = XmppMessage(target_date=date(2025, 9, 7), synthesis=None, external_info=None) + await channel.send_async(msg) + client_instance = mock_cls.return_value + assert client_instance.enable_starttls is True + assert client_instance.enable_direct_tls is False + + @patch("pronote_sync.channels.xmpp.ClientXMPP", new=FakeClientXMPP) + @pytest.mark.asyncio + async def test_tls_mode_disabled_config(self) -> None: + """Test que tls_mode='disabled' avec host='127.0.0.1' configure TLS désactivé. + + :return: None + """ + settings = XmppSettings( + enabled=True, + jid="bot@example.com", + password=SecretStr("secret123"), + host="127.0.0.1", + port=5222, + to="parent@example.com", + resource="pronote-sync", + tls_mode="disabled", + timeout=30, + ) + + class InspectClient(FakeClientXMPP): + def __init__(self, jid: str, password: str) -> None: + super().__init__(jid, password) + + with patch("pronote_sync.channels.xmpp.ClientXMPP") as mock_cls: + mock_cls.return_value = InspectClient("bot@example.com", "secret123") + channel = XmppChannel(settings, dry_run=False) + msg = XmppMessage(target_date=date(2025, 9, 7), synthesis=None, external_info=None) + await channel.send_async(msg) + client_instance = mock_cls.return_value + assert client_instance.enable_direct_tls is False + assert client_instance.enable_starttls is False + + @patch("pronote_sync.channels.xmpp.ClientXMPP", new=FakeClientXMPP) + @pytest.mark.asyncio + async def test_no_secret_leak_in_connection_failure_log( + self, caplog: pytest.LogCaptureFixture + ) -> None: + """Test qu'aucun secret ne fuit dans les logs en cas d'échec de connexion. + + :param caplog: Fixture pytest pour capturer les logs. + """ + sentinel_password = "SECRET_PASSWORD_XMPP_12345" # pragma: allowlist secret + settings = XmppSettings( + enabled=True, + jid="bot@example.com", + password=SecretStr(sentinel_password), + host="xmpp.example.com", + port=5222, + to="parent@example.com", + resource="pronote-sync", + tls_mode="starttls", + timeout=30, + ) + + class ConnectionFailedClient(FakeClientXMPP): + def __init__(self, jid: str, password: str) -> None: + super().__init__(jid, password) + self._connect_mode = "connection_failed" + + with patch("pronote_sync.channels.xmpp.ClientXMPP", new=ConnectionFailedClient): + channel = XmppChannel(settings, dry_run=False) + msg = XmppMessage(target_date=date(2025, 9, 7), synthesis=None, external_info=None) + await channel.send_async(msg) + + # Vérifier que le mot de passe sentinelle n'apparaît pas dans les logs + logs = caplog.text + assert sentinel_password not in logs + + @patch("pronote_sync.channels.xmpp.ClientXMPP", new=FakeClientXMPP) + @pytest.mark.asyncio + async def test_no_secret_leak_in_connect_timeout_log( + self, caplog: pytest.LogCaptureFixture + ) -> None: + """Test qu'aucun secret ne fuit dans les logs en cas de timeout de connexion. + + :param caplog: Fixture pytest pour capturer les logs. + """ + sentinel_password = "SECRET_PASSWORD_XMPP_12345" + settings = XmppSettings( + enabled=True, + jid="bot@example.com", + password=SecretStr(sentinel_password), + host="xmpp.example.com", + port=5222, + to="parent@example.com", + resource="pronote-sync", + tls_mode="starttls", + connect_timeout=0.05, + timeout=30, + cleanup_timeout=0.01, + ) + + class PendingConnectClient(FakeClientXMPP): + def __init__(self, jid: str, password: str) -> None: + super().__init__(jid, password) + self._connect_mode = "pending" + + with patch("pronote_sync.channels.xmpp.ClientXMPP", new=PendingConnectClient): + channel = XmppChannel(settings, dry_run=False) + msg = XmppMessage(target_date=date(2025, 9, 7), synthesis=None, external_info=None) + await channel.send_async(msg) + + # Vérifier que le mot de passe sentinelle n'apparaît pas dans les logs + logs = caplog.text + assert sentinel_password not in logs + @patch("pronote_sync.channels.xmpp.ClientXMPP", new=FakeClientXMPP) @pytest.mark.asyncio async def test_send_async_tls_direct_config(self) -> None: @@ -816,6 +1172,328 @@ class TestXmppChannelSend: result = await channel.send_async(msg) assert result is True + @patch("pronote_sync.channels.xmpp.ClientXMPP", new=FakeClientXMPP) + @pytest.mark.asyncio + async def test_send_async_connect_raises_exception_returns_false(self) -> None: + """Test que connect() levant une exception retourne False. + + :return: None + """ + settings = XmppSettings( + enabled=True, + jid="bot@example.com", + password=SecretStr("secret123"), + host="xmpp.example.com", + port=5222, + to="parent@example.com", + resource="pronote-sync", + tls_mode="starttls", + timeout=30, + ) + + class ConnectExceptionClient(FakeClientXMPP): + def connect( + self, host: str | None = None, port: int | None = None + ) -> asyncio.Future[bool]: + raise ConnectionError("Network unreachable") + + with patch("pronote_sync.channels.xmpp.ClientXMPP", new=ConnectExceptionClient): + channel = XmppChannel(settings, dry_run=False) + msg = XmppMessage(target_date=date(2025, 9, 7), synthesis=None, external_info=None) + result = await channel.send_async(msg) + assert result is False + + @patch("pronote_sync.channels.xmpp.ClientXMPP", new=FakeClientXMPP) + @pytest.mark.asyncio + async def test_send_async_connect_future_error_returns_false( + self, caplog: pytest.LogCaptureFixture + ) -> None: + """Test que la future de connect() résolue avec une exception retourne False. + + Le message d'erreur doit être expurgé et ne pas contenir les secrets. + + :param caplog: Fixture pytest pour capturer les logs. + """ + sentinel_password = "SECRET_PASSWORD_XMPP_CONNECT_FUT" # pragma: allowlist secret + settings = XmppSettings( + enabled=True, + jid="bot@example.com", + password=SecretStr(sentinel_password), + host="xmpp.example.com", + port=5222, + to="parent@example.com", + resource="pronote-sync", + tls_mode="starttls", + timeout=30, + ) + + class ConnectFutureErrorClient(FakeClientXMPP): + def connect( + self, host: str | None = None, port: int | None = None + ) -> asyncio.Future[bool]: + loop = asyncio.get_event_loop() + future: asyncio.Future[bool] = loop.create_future() + future.set_exception(ConnectionError("Network unreachable")) + return future + + with patch("pronote_sync.channels.xmpp.ClientXMPP", new=ConnectFutureErrorClient): + channel = XmppChannel(settings, dry_run=False) + msg = XmppMessage(target_date=date(2025, 9, 7), synthesis=None, external_info=None) + result = await channel.send_async(msg) + assert result is False + assert "Échec de connexion XMPP" in caplog.text + assert sentinel_password not in caplog.text + + @patch("pronote_sync.channels.xmpp.ClientXMPP", new=FakeClientXMPP) + @pytest.mark.asyncio + async def test_send_async_session_failure_after_connect_returns_false( + self, caplog: pytest.LogCaptureFixture + ) -> None: + """Test qu'un échec de session après connexion résolue retourne False. + + Le connect future est résolu avec succès, puis l'événement + ``failed_auth`` arrive pendant l'attente de session. + + :param caplog: Fixture pytest pour capturer les logs. + """ + settings = XmppSettings( + enabled=True, + jid="bot@example.com", + password=SecretStr("secret123"), + host="xmpp.example.com", + port=5222, + to="parent@example.com", + resource="pronote-sync", + tls_mode="starttls", + timeout=5, + cleanup_timeout=0.01, + ) + + class LateAuthFailClient(FakeClientXMPP): + def connect( + self, host: str | None = None, port: int | None = None + ) -> asyncio.Future[bool]: + loop = asyncio.get_event_loop() + future: asyncio.Future[bool] = loop.create_future() + self.connected = True + self._host_used = host + self._port_used = port + loop.call_later(0.01, lambda: self._fire("failed_auth")) + future.set_result(True) + return future + + with patch("pronote_sync.channels.xmpp.ClientXMPP", new=LateAuthFailClient): + channel = XmppChannel(settings, dry_run=False) + msg = XmppMessage(target_date=date(2025, 9, 7), synthesis=None, external_info=None) + result = await channel.send_async(msg) + assert result is False + assert "authentification" in caplog.text.lower() + + @patch("pronote_sync.channels.xmpp.ClientXMPP", new=FakeClientXMPP) + @pytest.mark.asyncio + async def test_send_async_cancelled_returns_false( + self, caplog: pytest.LogCaptureFixture + ) -> None: + """Test que l'annulation de la tâche retourne False sans lever. + + Le contrat du canal impose un retour booléen : ``CancelledError`` doit + être interceptée et convertie en ``False``. + + :param caplog: Fixture pytest pour capturer les logs. + """ + import logging + + caplog.set_level(logging.DEBUG) + settings = XmppSettings( + enabled=True, + jid="bot@example.com", + password=SecretStr("secret123"), + host="xmpp.example.com", + port=5222, + to="parent@example.com", + resource="pronote-sync", + tls_mode="starttls", + connect_timeout=15, + timeout=30, + cleanup_timeout=0.01, + ) + + class PendingConnectClient(FakeClientXMPP): + def __init__(self, jid: str, password: str) -> None: + super().__init__(jid, password) + self._connect_mode = "pending" + + with patch("pronote_sync.channels.xmpp.ClientXMPP", new=PendingConnectClient): + channel = XmppChannel(settings, dry_run=False) + msg = XmppMessage(target_date=date(2025, 9, 7), synthesis=None, external_info=None) + task = asyncio.ensure_future(channel.send_async(msg)) + await asyncio.sleep(0.001) + task.cancel() + result = await asyncio.wait_for(task, timeout=1.0) + assert result is False + assert "annulé" in caplog.text.lower() + + +class TestSecretValues: + """Tests pour la fonction _secret_values.""" + + def test_secret_values_with_all_secrets(self) -> None: + """Test que _secret_values retourne tous les secrets. + + :return: None + """ + from pronote_sync.channels.xmpp import _secret_values + + settings = XmppSettings( + jid="bot@example.com", + password=SecretStr("secret123"), + to="parent@example.com", + ) + secrets = _secret_values(settings) + assert len(secrets) == 3 + assert "bot@example.com" in secrets + assert settings.password in secrets + assert "parent@example.com" in secrets + + def test_secret_values_with_none_values(self) -> None: + """Test que _secret_values filtre les valeurs None. + + :return: None + """ + from pronote_sync.channels.xmpp import _secret_values + + settings = XmppSettings( + jid=None, + password=None, + to=None, + ) + secrets = _secret_values(settings) + assert len(secrets) == 0 + + def test_secret_values_with_some_none(self) -> None: + """Test que _secret_values gère les valeurs partiellement None. + + :return: None + """ + from pronote_sync.channels.xmpp import _secret_values + + settings = XmppSettings( + jid="bot@example.com", + password=None, + to="parent@example.com", + ) + secrets = _secret_values(settings) + assert len(secrets) == 2 + assert "bot@example.com" in secrets + assert "parent@example.com" in secrets + + +class TestSyncXmppChannel: + """Tests unitaires pour la classe SyncXmppChannel.""" + + @patch("pronote_sync.channels.xmpp.ClientXMPP", new=FakeClientXMPP) + def test_sync_send_dry_run_returns_true(self) -> None: + """Test que SyncXmppChannel en dry_run retourne True. + + :return: None + """ + settings = XmppSettings( + enabled=True, + jid="bot@example.com", + password=SecretStr("secret123"), + host="xmpp.example.com", + port=5222, + to="parent@example.com", + resource="pronote-sync", + use_tls=True, + timeout=30, + ) + channel = SyncXmppChannel(settings, dry_run=True) + msg = XmppMessage(target_date=date(2025, 9, 7), synthesis=None, external_info=None) + result = channel.send(msg) + assert result is True + + @patch("pronote_sync.channels.xmpp.ClientXMPP", new=FakeClientXMPP) + def test_sync_send_success_returns_true(self) -> None: + """Test que SyncXmppChannel.send retourne True en cas de succès. + + :return: None + """ + settings = XmppSettings( + enabled=True, + jid="bot@example.com", + password=SecretStr("secret123"), + host="xmpp.example.com", + port=5222, + to="parent@example.com", + resource="pronote-sync", + use_tls=True, + timeout=30, + ) + channel = SyncXmppChannel(settings, dry_run=False) + msg = XmppMessage(target_date=date(2025, 9, 7), synthesis=None, external_info=None) + result = channel.send(msg) + assert result is True + + @patch("pronote_sync.channels.xmpp.ClientXMPP", new=FakeClientXMPP) + def test_sync_send_exception_returns_false(self) -> None: + """Test que SyncXmppChannel.send retourne False en cas d'exception. + + :return: None + """ + settings = XmppSettings( + enabled=True, + jid="bot@example.com", + password=SecretStr("secret123"), + host="xmpp.example.com", + port=5222, + to="parent@example.com", + resource="pronote-sync", + use_tls=True, + timeout=30, + ) + + class ErrorClient(FakeClientXMPP): + def connect( + self, host: str | None = None, port: int | None = None + ) -> asyncio.Future[bool]: + raise RuntimeError("Connection failed") + + with patch("pronote_sync.channels.xmpp.ClientXMPP", new=ErrorClient): + channel = SyncXmppChannel(settings, dry_run=False) + msg = XmppMessage(target_date=date(2025, 9, 7), synthesis=None, external_info=None) + result = channel.send(msg) + assert result is False + + @patch("pronote_sync.channels.xmpp.ClientXMPP", new=FakeClientXMPP) + def test_sync_send_asyncio_run_error_returns_false( + self, caplog: pytest.LogCaptureFixture + ) -> None: + """Test que send retourne False si asyncio.run lève une exception. + + :param caplog: Fixture pytest pour capturer les logs. + """ + settings = XmppSettings( + enabled=True, + jid="bot@example.com", + password=SecretStr("secret123"), + host="xmpp.example.com", + port=5222, + to="parent@example.com", + resource="pronote-sync", + use_tls=True, + timeout=30, + ) + channel = SyncXmppChannel(settings, dry_run=False) + msg = XmppMessage(target_date=date(2025, 9, 7), synthesis=None, external_info=None) + with patch( + "pronote_sync.channels.xmpp.asyncio.run", + side_effect=RuntimeError("Boucle événementielle indisponible"), + ): + result = channel.send(msg) + assert result is False + assert "erreur" in caplog.text.lower() + class TestXmppChannelSecurity: """Tests de sécurité pour XmppChannel (non-fuite de secrets). diff --git a/tests/unit/test_xmpp_settings.py b/tests/unit/test_xmpp_settings.py index cdfc1de..eb78e00 100644 --- a/tests/unit/test_xmpp_settings.py +++ b/tests/unit/test_xmpp_settings.py @@ -99,19 +99,52 @@ class TestTimeoutConstraints: class TestTlsPolicy: """Tests de la politique TLS pour le champ use_tls.""" - def test_use_tls_false_with_remote_host_rejected(self) -> None: - """Vérifie que use_tls=False avec un hôte distant est rejeté. + def test_tls_disabled_with_remote_host_rejected(self) -> None: + """Vérifie que tls_mode="disabled" avec un hôte distant est rejeté. - :raises ValidationError: Si use_tls=False et host n'est pas une boucle locale. + :raises ValidationError: Si tls_mode="disabled" et host n'est pas une boucle locale. """ with pytest.raises(ValidationError) as exc_info: - XmppSettings(use_tls=False, host="talk.example.com") - assert "use_tls" in str(exc_info.value).lower() or "tls" in str(exc_info.value).lower() + XmppSettings(tls_mode="disabled", host="talk.example.com") + assert "tls" in str(exc_info.value).lower() - def test_use_tls_false_with_remote_host_rejected_when_enabled(self) -> None: + def test_tls_disabled_with_remote_host_rejected_when_enabled(self) -> None: """Vérifie que la politique TLS s'applique même quand le canal est activé.""" with pytest.raises(ValidationError): - XmppSettings(enabled=True, use_tls=False, host="talk.example.com") + XmppSettings(enabled=True, tls_mode="disabled", host="talk.example.com") + + def test_use_tls_false_maps_to_starttls(self) -> None: + """Vérifie que l'alias obsolète use_tls=False mappe vers tls_mode='starttls'. + + :return: Vérifie le mapping et l'émission d'un DeprecationWarning. + :rtype: None + """ + with pytest.warns(DeprecationWarning): + settings = XmppSettings(use_tls=False, host="talk.example.com") + assert settings.tls_mode == "starttls" + assert settings.host == "talk.example.com" + + def test_use_tls_true_maps_to_direct(self) -> None: + """Vérifie que l'alias obsolète use_tls=True mappe vers tls_mode='direct'. + + :return: Vérifie le mapping et l'émission d'un DeprecationWarning. + :rtype: None + """ + with pytest.warns(DeprecationWarning): + settings = XmppSettings(use_tls=True, host="talk.example.com") + assert settings.tls_mode == "direct" + assert settings.host == "talk.example.com" + + def test_default_tls_mode_is_starttls(self) -> None: + """Vérifie que tls_mode vaut 'starttls' par défaut avec les timeouts par défaut. + + :return: Vérifie tls_mode='starttls', connect_timeout=15, cleanup_timeout=10. + :rtype: None + """ + settings = XmppSettings() + assert settings.tls_mode == "starttls" + assert settings.connect_timeout == 15 + assert settings.cleanup_timeout == 10 def test_use_tls_false_with_localhost_allowed(self) -> None: """Vérifie que use_tls=False avec localhost est autorisé. @@ -120,7 +153,7 @@ class TestTlsPolicy: :rtype: None """ settings = XmppSettings(use_tls=False, host="localhost") - assert settings.use_tls is False + assert settings.tls_mode == "starttls" assert settings.host == "localhost" def test_use_tls_false_with_127_allowed(self) -> None: @@ -130,7 +163,7 @@ class TestTlsPolicy: :rtype: None """ settings = XmppSettings(use_tls=False, host="127.0.0.1") - assert settings.use_tls is False + assert settings.tls_mode == "starttls" assert settings.host == "127.0.0.1" def test_use_tls_false_with_ipv6_loopback_allowed(self) -> None: @@ -140,7 +173,7 @@ class TestTlsPolicy: :rtype: None """ settings = XmppSettings(use_tls=False, host="::1") - assert settings.use_tls is False + assert settings.tls_mode == "starttls" assert settings.host == "::1" def test_use_tls_true_with_remote_host_allowed(self) -> None: @@ -150,7 +183,7 @@ class TestTlsPolicy: :rtype: None """ settings = XmppSettings(use_tls=True, host="talk.example.com") - assert settings.use_tls is True + assert settings.tls_mode == "direct" assert settings.host == "talk.example.com" def test_use_tls_true_with_empty_host_allowed(self) -> None: @@ -160,7 +193,7 @@ class TestTlsPolicy: :rtype: None """ settings = XmppSettings(use_tls=True, host="") - assert settings.use_tls is True + assert settings.tls_mode == "direct" assert settings.host == "" @@ -173,7 +206,7 @@ class TestNoSecretInErrorMessages: Crée une instance avec des valeurs sensibles et vérifie que l'erreur de validation ne contient pas ces valeurs dans son message. - :raises ValidationError: Si use_tls=False avec un hôte non-local. + :raises ValidationError: Si tls_mode="disabled" avec un hôte non-local. :return: Vérifie que le message d'erreur ne contient pas les secrets. :rtype: None """ @@ -184,7 +217,7 @@ class TestNoSecretInErrorMessages: with pytest.raises(ValidationError) as exc_info: XmppSettings( - use_tls=False, + tls_mode="disabled", host="talk.example.com", jid=sentinel_jid, password=sentinel_password, @@ -197,3 +230,72 @@ class TestNoSecretInErrorMessages: assert "test_password_123" not in error_message assert "test_to@example.com" not in error_message assert "secret" not in error_message + + +class TestXmppSettingsTlsMode: + """Tests supplémentaires pour les validateurs de XmppSettings.""" + + def test_use_tls_true_maps_to_direct_with_warning(self) -> None: + """Vérifie que use_tls=True mappe vers tls_mode='direct' avec DeprecationWarning. + + :return: Vérifie le mapping et l'émission d'un DeprecationWarning. + :rtype: None + """ + with pytest.warns(DeprecationWarning): + settings = XmppSettings(use_tls=True, host="talk.example.com") + assert settings.tls_mode == "direct" + + def test_use_tls_false_maps_to_starttls_with_warning(self) -> None: + """Vérifie que use_tls=False mappe vers tls_mode='starttls' avec DeprecationWarning. + + :return: Vérifie le mapping et l'émission d'un DeprecationWarning. + :rtype: None + """ + with pytest.warns(DeprecationWarning): + settings = XmppSettings(use_tls=False, host="talk.example.com") + assert settings.tls_mode == "starttls" + + def test_tls_mode_disabled_rejected_for_remote(self) -> None: + """Vérifie que tls_mode='disabled' avec un hôte distant est rejeté. + + :raises ValidationError: Si tls_mode="disabled" et host n'est pas une boucle locale. + """ + with pytest.raises(ValidationError) as exc_info: + XmppSettings(tls_mode="disabled", host="example.com") + assert "tls" in str(exc_info.value).lower() + + def test_tls_mode_disabled_allowed_for_loopback(self) -> None: + """Vérifie que tls_mode='disabled' avec host='127.0.0.1' est valide. + + :return: Vérifie que XmppSettings(tls_mode='disabled', host='127.0.0.1') est valide. + :rtype: None + """ + settings = XmppSettings(tls_mode="disabled", host="127.0.0.1") + assert settings.tls_mode == "disabled" + assert settings.host == "127.0.0.1" + + def test_connect_timeout_must_be_positive(self) -> None: + """Vérifie que connect_timeout doit être strictement positif. + + :raises ValidationError: Si connect_timeout <= 0. + """ + with pytest.raises(ValidationError) as exc_info: + XmppSettings(connect_timeout=0) + assert "connect_timeout" in str(exc_info.value).lower() + + with pytest.raises(ValidationError) as exc_info: + XmppSettings(connect_timeout=-1) + assert "connect_timeout" in str(exc_info.value).lower() + + def test_cleanup_timeout_must_be_positive(self) -> None: + """Vérifie que cleanup_timeout doit être strictement positif. + + :raises ValidationError: Si cleanup_timeout <= 0. + """ + with pytest.raises(ValidationError) as exc_info: + XmppSettings(cleanup_timeout=0) + assert "cleanup_timeout" in str(exc_info.value).lower() + + with pytest.raises(ValidationError) as exc_info: + XmppSettings(cleanup_timeout=-1) + assert "cleanup_timeout" in str(exc_info.value).lower()