refactor(ia): unifier le contrat AI_BASE_URL entre les providers #43

Merged
OpenCode merged 2 commits from refactor/issue-17-ai-base-url into main 2026-09-12 20:00:42 +02:00
6 changed files with 653 additions and 91 deletions
+7
View File
@@ -66,6 +66,13 @@ AI_BASE_URL=https://api.openai.com/v1
# AI_API_KEY= # AI_API_KEY=
# AI_MODEL=gpt-4o-mini # exemple recommandé, non activé par défaut # AI_MODEL=gpt-4o-mini # exemple recommandé, non activé par défaut
# NOTE : la validation structurelle de AI_BASE_URL s'applique à TOUS les
# providers (openai, litellm, openai-compatible) : HTTPS obligatoire sauf si
# AI_ALLOW_INSECURE_HTTP=true, aucun credential embarqué (user:pass@hôte),
# aucun paramètre sensible dans la query string (token, key, api_key,
# secret, password, auth), et aucune manipulation automatique de /v1.
# Seul le provider openai-compatible exige AI_BASE_URL et AI_MODEL.
# Exemple : OpenRouter (HTTPS) # Exemple : OpenRouter (HTTPS)
# AI_PROVIDER=openai-compatible # AI_PROVIDER=openai-compatible
# AI_BASE_URL=https://openrouter.ai/api/v1 # AI_BASE_URL=https://openrouter.ai/api/v1
+2 -2
View File
@@ -140,7 +140,7 @@
"filename": "GUIDE_DEV_PYTHON.md", "filename": "GUIDE_DEV_PYTHON.md",
"hashed_secret": "90bd1b48e958257948487b90bee080ba5ed00caa", "hashed_secret": "90bd1b48e958257948487b90bee080ba5ed00caa",
"is_verified": false, "is_verified": false,
"line_number": 5133 "line_number": 5186
} }
], ],
"tests/unit/test_caldav_gateway.py": [ "tests/unit/test_caldav_gateway.py": [
@@ -185,5 +185,5 @@
} }
] ]
}, },
"generated_at": "2026-09-12T17:18:46Z" "generated_at": "2026-09-12T17:57:39Z"
} }
+90 -37
View File
@@ -307,10 +307,10 @@ d'un besoin réel et testé.
| `THEORETICAL_WEEK_ANCHOR_TYPE` | Parité de la semaine de référence (`even` ou `odd`). | `None` | `Literal["even", "odd"] \| None`| | `THEORETICAL_WEEK_ANCHOR_TYPE` | Parité de la semaine de référence (`even` ou `odd`). | `None` | `Literal["even", "odd"] \| None`|
| `AI_ENABLED` | Activer la synthèse IA. | `False` | `bool` | | `AI_ENABLED` | Activer la synthèse IA. | `False` | `bool` |
| `AI_PROVIDER` | Fournisseur IA (`openai`, `openai-compatible` ou `litellm`). | `openai` | `Literal["openai", "litellm", "openai-compatible"]` | | `AI_PROVIDER` | Fournisseur IA (`openai`, `openai-compatible` ou `litellm`). | `openai` | `Literal["openai", "litellm", "openai-compatible"]` |
| `AI_BASE_URL` | URL de base pour l'API IA (ex: OpenAI compatible). | `None` | `str \| None`| | `AI_BASE_URL` | URL de base pour l'API IA (validée structurellement pour tous les providers, voir ci-dessous). | `None` | `str \| None`|
| `AI_API_KEY` | Clé API pour l'API IA. | `None` | `SecretStr` | | `AI_API_KEY` | Clé API pour l'API IA. | `None` | `SecretStr` |
| `AI_MODEL` | Modèle IA à utiliser (exemple recommandé : `gpt-4o-mini`). | `None` | `str \| None`| | `AI_MODEL` | Modèle IA à utiliser (exemple recommandé : `gpt-4o-mini`). | `None` | `str \| None`|
| `AI_ALLOW_INSECURE_HTTP` | Autoriser HTTP (non sécurisé) pour `openai-compatible` uniquement. | `False` | `bool` | | `AI_ALLOW_INSECURE_HTTP` | Autoriser HTTP (non sécurisé) pour tous les providers (openai, litellm, openai-compatible). | `False` | `bool` |
| `DRY_RUN` | Simulation sans sortie distante ni état local persistant ; incompatible avec `qr_token`. | `False` | `bool` | | `DRY_RUN` | Simulation sans sortie distante ni état local persistant ; incompatible avec `qr_token`. | `False` | `bool` |
| `LOG_LEVEL` | Niveau de log (`DEBUG`, `INFO`, `WARNING`, `ERROR`). | `INFO` | `str` | | `LOG_LEVEL` | Niveau de log (`DEBUG`, `INFO`, `WARNING`, `ERROR`). | `INFO` | `str` |
@@ -394,7 +394,7 @@ LOG_LEVEL=INFO
> `CalDAVSettings.calendar_path` a pour valeur par défaut `"/pronote-sync/"`. > `CalDAVSettings.calendar_path` a pour valeur par défaut `"/pronote-sync/"`.
> `XmppSettings.resource` a pour valeur par défaut `"pronote-sync"`. > `XmppSettings.resource` a pour valeur par défaut `"pronote-sync"`.
> > ``AISettings.provider`` accepte également ``openai-compatible`` (réutilise ``OpenAISynthesisProvider`` avec un ``base_url`` personnalisé). > > ``AISettings.provider`` accepte également ``openai-compatible`` (réutilise ``OpenAISynthesisProvider`` avec un ``base_url`` personnalisé).
> > ``AISettings.allow_insecure_http`` (défaut ``False``) autorise les URLs HTTP pour le provider ``openai-compatible`` uniquement. > > ``AISettings.allow_insecure_http`` (défaut ``False``) autorise les URLs HTTP pour les trois providers (``openai``, ``litellm`` et ``openai-compatible``).
```python ```python
from typing import Literal from typing import Literal
@@ -3925,16 +3925,26 @@ L'import de `litellm` est conditionnel avec `try/except ImportError` → `None`.
| ``openai-compatible`` | Proxy ou serveur compatible OpenAI | ``OpenAISynthesisProvider`` | | ``openai-compatible`` | Proxy ou serveur compatible OpenAI | ``OpenAISynthesisProvider`` |
| ``litellm`` | Bibliothèque LiteLLM embarquée | ``LiteLLMSynthesisProvider`` | | ``litellm`` | Bibliothèque LiteLLM embarquée | ``LiteLLMSynthesisProvider`` |
Pour le provider ``openai-compatible``, la validation de la configuration est stricte : Pour le provider ``openai-compatible``, ``AI_BASE_URL`` et ``AI_MODEL`` sont requis ; pour
``openai`` et ``litellm``, ils sont optionnels. La validation structurelle de ``AI_BASE_URL``
(partagée via ``_validate_base_url``) s'applique de façon identique aux trois providers dès que
l'URL est renseignée :
- ``AI_BASE_URL`` est requis. - URL parsable par ``urlparse`` (``ValueError`` ou schéma vide → refusée) et hostname non vide.
- ``AI_MODEL`` est requis et ne doit pas être vide. - Schéma limité à ``http``/``https`` ; ``http`` refusé sauf si ``AI_ALLOW_INSECURE_HTTP=true``.
- ``AI_API_KEY`` est requis (MVP). - Credentials dans l'URL (``user:pass@host``) refusés.
- L'URL doit utiliser le schéma ``https`` sauf si ``AI_ALLOW_INSECURE_HTTP=true``. - Paramètres sensibles dans la *query string* refusés, y compris sans valeur
- Les credentials dans l'URL sont refusés. (``token``, ``key``, ``api_key``, ``secret``, ``password``, ``auth``).
- Les paramètres sensibles dans la *query string* sont refusés.
- Aucune manipulation automatique de ``/v1`` n'est effectuée. - Aucune manipulation automatique de ``/v1`` n'est effectuée.
- Si la configuration est incomplète, la factory retourne ``None`` avec un avertissement (mode dégradé). - En cas de violation, un avertissement est journalisé (URL masquée via ``redact_url``) et la
factory retourne ``None`` (mode dégradé) ; la factory ne lève jamais d'exception et ne fait
aucun appel réseau.
| Provider | `AI_BASE_URL` | `AI_MODEL` | Validation structurelle |
|---------------------|---------------|------------|---------------------------------------------|
| ``openai`` | optionnel | optionnel | `_validate_base_url("openai", ...)` |
| ``litellm`` | optionnel | optionnel | `_validate_base_url("litellm", ...)` |
| ``openai-compatible`` | requis | requis | `_validate_openai_compatible_config` (présence puis `_validate_base_url`) |
La politique hors réseau de la table des modèles litellm est gérée par `LITELLM_LOCAL_MODEL_COST_MAP=true`. Les tests utilisent `pytest.importorskip("litellm")`. La politique hors réseau de la table des modèles litellm est gérée par `LITELLM_LOCAL_MODEL_COST_MAP=true`. Les tests utilisent `pytest.importorskip("litellm")`.
@@ -3950,37 +3960,69 @@ from ..utils.redaction import redact_url
logger = logging.getLogger(__name__) logger = logging.getLogger(__name__)
_SENSITIVE_QUERY_PARAMS = {"token", "key", "api_key", "secret", "password", "auth"}
def _validate_base_url(provider: str, url: str, allow_insecure_http: bool) -> str | None:
"""Valide structurellement une URL de base IA, partagée entre providers."""
try:
parsed = urlparse(url)
if not parsed.scheme:
logger.warning("URL invalide pour le provider %s : %s", provider, redact_url(url))
return None
if not parsed.hostname:
logger.warning(
"URL sans hostname pour le provider %s : %s", provider, redact_url(url)
)
return None
if parsed.scheme not in ("http", "https"):
logger.warning(
"Schéma d'URL non supporté pour le provider %s : %s", provider, redact_url(url)
)
return None
if parsed.scheme == "http" and not allow_insecure_http:
logger.warning(
"URL HTTP non autorisée sans AI_ALLOW_INSECURE_HTTP=true pour le provider %s : %s",
provider,
redact_url(url),
)
return None
if parsed.username is not None or parsed.password is not None:
logger.warning(
"Credentials dans l'URL refusés pour le provider %s : %s",
provider,
redact_url(url),
)
return None
param_names = [
name.lower() for name, _ in parse_qsl(parsed.query, keep_blank_values=True)
]
if any(name in _SENSITIVE_QUERY_PARAMS for name in param_names):
logger.warning(
"Paramètres sensibles dans l'URL refusés pour le provider %s : %s",
provider,
redact_url(url),
)
return None
# Accéder à parsed.port peut lever ValueError (port invalide/hors bornes).
parsed.port # noqa: B018
except ValueError:
logger.warning("URL invalide pour le provider %s : %s", provider, redact_url(url))
return None
return url
def _validate_openai_compatible_config( def _validate_openai_compatible_config(
url: str | None, model: str | None, allow_insecure_http: bool url: str | None, model: str | None, allow_insecure_http: bool
) -> str | None: ) -> str | None:
"""Valide la configuration du provider ``openai-compatible``.""" """Valide la configuration du provider ``openai-compatible``."""
if not url or not model: if not url:
logger.warning("URL de base requise pour le provider openai-compatible")
return None return None
try: if not model:
parsed = urlparse(url) logger.warning("Modèle requis pour le provider openai-compatible")
except ValueError:
logger.warning("URL invalide : %s", redact_url(url))
return None return None
if not parsed.hostname: return _validate_base_url("openai-compatible", url, allow_insecure_http)
logger.warning("URL sans hostname : %s", redact_url(url))
return None
if parsed.scheme not in ("http", "https"):
return None
if parsed.scheme == "http" and not allow_insecure_http:
return None
if parsed.username is not None or parsed.password is not None:
logger.warning("Credentials dans l'URL refusés : %s", redact_url(url))
return None
sensitive_names = {"token", "key", "api_key", "secret", "password", "auth"}
param_names = [
name.lower() for name, _ in parse_qsl(parsed.query, keep_blank_values=True)
]
if any(name in sensitive_names for name in param_names):
logger.warning(
"Paramètres sensibles dans l'URL refusés : %s", redact_url(url)
)
return None
return url
def get_synthesis_provider(settings: AISettings) -> SynthesisProvider | None: def get_synthesis_provider(settings: AISettings) -> SynthesisProvider | None:
@@ -3989,7 +4031,10 @@ def get_synthesis_provider(settings: AISettings) -> SynthesisProvider | None:
Retourne ``None`` lorsque la synthèse IA est désactivée ou qu'aucune clé Retourne ``None`` lorsque la synthèse IA est désactivée ou qu'aucune clé
API n'est configurée. Pour le provider ``litellm``, le paquet ``litellm`` API n'est configurée. Pour le provider ``litellm``, le paquet ``litellm``
(extra ``ai-litellm``) est requis : s'il est absent, un avertissement est (extra ``ai-litellm``) est requis : s'il est absent, un avertissement est
journalisé et ``None`` est retourné. journalisé et ``None`` est retourné. Pour ``openai`` et ``litellm``,
une ``base_url`` éventuelle est validée par :func:`_validate_base_url` ;
pour le provider ``openai-compatible``, la configuration est validée par
:func:`_validate_openai_compatible_config`.
:param settings: Paramètres IA. :param settings: Paramètres IA.
:return: Le fournisseur configuré, ou ``None`` si désactivé ou sans clé API. :return: Le fournisseur configuré, ou ``None`` si désactivé ou sans clé API.
@@ -4009,6 +4054,10 @@ def get_synthesis_provider(settings: AISettings) -> SynthesisProvider | None:
except ImportError: except ImportError:
logger.warning("Extra 'ai-litellm' requis pour le provider litellm") logger.warning("Extra 'ai-litellm' requis pour le provider litellm")
return None return None
if base_url is not None and (
_validate_base_url("litellm", base_url, settings.allow_insecure_http) is None
):
return None
return LiteLLMSynthesisProvider(api_key=settings.api_key, base_url=base_url, model=model) return LiteLLMSynthesisProvider(api_key=settings.api_key, base_url=base_url, model=model)
if settings.provider == "openai-compatible": if settings.provider == "openai-compatible":
@@ -4019,6 +4068,10 @@ def get_synthesis_provider(settings: AISettings) -> SynthesisProvider | None:
return None return None
return OpenAISynthesisProvider(api_key=settings.api_key, base_url=url, model=model) return OpenAISynthesisProvider(api_key=settings.api_key, base_url=url, model=model)
if base_url is not None and (
_validate_base_url("openai", base_url, settings.allow_insecure_http) is None
):
return None
return OpenAISynthesisProvider(api_key=settings.api_key, base_url=base_url, model=model) return OpenAISynthesisProvider(api_key=settings.api_key, base_url=base_url, model=model)
``` ```
+1 -1
View File
@@ -75,7 +75,7 @@ The following variables can be safely pre-configured in `/etc/pronote-sync/prono
- `XMPP_USE_TLS` is deprecated but still supported (aliased to `XMPP_TLS_MODE`) - `XMPP_USE_TLS` is deprecated but still supported (aliased to `XMPP_TLS_MODE`)
- **AI:** - **AI:**
- `AI_ENABLED`, `AI_PROVIDER`, `AI_BASE_URL`, `AI_MODEL`, `AI_ALLOW_INSECURE_HTTP` - `AI_ENABLED`, `AI_PROVIDER`, `AI_BASE_URL`, `AI_MODEL`, `AI_ALLOW_INSECURE_HTTP` — la validation structurelle de `AI_BASE_URL` s'applique à tous les providers (HTTPS sauf `AI_ALLOW_INSECURE_HTTP=true`, pas de credentials ni de paramètres sensibles dans l'URL, pas de `/v1` automatique) ; seul `openai-compatible` exige `AI_BASE_URL` et `AI_MODEL`.
- **Blog:** - **Blog:**
- `BLOG_ENABLED`, `BLOG_RSS_URL` - `BLOG_ENABLED`, `BLOG_RSS_URL`
+87 -50
View File
@@ -15,20 +15,83 @@ logger = logging.getLogger(__name__)
__all__ = ["get_synthesis_provider", "SynthesisProvider", "OpenAISynthesisProvider"] __all__ = ["get_synthesis_provider", "SynthesisProvider", "OpenAISynthesisProvider"]
_SENSITIVE_QUERY_PARAMS = {"token", "key", "api_key", "secret", "password", "auth"}
def _validate_base_url(provider: str, url: str, allow_insecure_http: bool) -> str | None:
"""Valide structurellement une URL de base IA, partagée entre providers.
Applique les règles structurelles identiques aux trois providers
(``openai``, ``litellm`` et ``openai-compatible``) : URL parsable par
``urlparse`` (``ValueError`` rejeté), hostname non vide, schéma limité
à ``http``/``https`` (HTTP refusé sauf si ``allow_insecure_http`` vaut
``True``), absence d'identifiants dans le netloc et de paramètres
sensibles dans la requête (y compris les paramètres sans valeur). L'URL
est retournée strictement inchangée : aucune manipulation automatique
du suffixe ``/v1`` n'est effectuée. En cas de violation, un
avertissement est journalisé (l'URL est toujours masquée via
:func:`redact_url`) et ``None`` est retourné ; la fonction ne lève
jamais d'exception et n'effectue aucun appel réseau.
:param provider: Nom du provider (utilisé pour le message d'avertissement).
:param url: URL de base à valider (non vide).
:param allow_insecure_http: Autorise ou non les URLs en HTTP.
:return: L'URL validée, strictement inchangée, ou ``None`` si invalide.
:rtype: str | None
"""
try:
parsed = urlparse(url)
if not parsed.scheme:
logger.warning("URL invalide pour le provider %s : %s", provider, redact_url(url))
return None
if not parsed.hostname:
logger.warning("URL sans hostname pour le provider %s : %s", provider, redact_url(url))
return None
if parsed.scheme not in ("http", "https"):
logger.warning(
"Schéma d'URL non supporté pour le provider %s : %s", provider, redact_url(url)
)
return None
if parsed.scheme == "http" and not allow_insecure_http:
logger.warning(
"URL HTTP non autorisée sans AI_ALLOW_INSECURE_HTTP=true pour le provider %s : %s",
provider,
redact_url(url),
)
return None
if parsed.username is not None or parsed.password is not None:
logger.warning(
"Credentials dans l'URL refusés pour le provider %s : %s",
provider,
redact_url(url),
)
return None
param_names = [name.lower() for name, _ in parse_qsl(parsed.query, keep_blank_values=True)]
if any(name in _SENSITIVE_QUERY_PARAMS for name in param_names):
logger.warning(
"Paramètres sensibles dans l'URL refusés pour le provider %s : %s",
provider,
redact_url(url),
)
return None
# Accéder à parsed.port peut lever ValueError (port invalide/hors bornes).
parsed.port # noqa: B018
except ValueError:
logger.warning("URL invalide pour le provider %s : %s", provider, redact_url(url))
return None
return url
def _validate_openai_compatible_config( def _validate_openai_compatible_config(
url: str | None, model: str | None, allow_insecure_http: bool url: str | None, model: str | None, allow_insecure_http: bool
) -> str | None: ) -> str | None:
"""Valide la configuration du provider ``openai-compatible``. """Valide la configuration du provider ``openai-compatible``.
Vérifie la présence de l'URL de base et du modèle, le schéma de l'URL Vérifie d'abord la présence de l'URL de base et du modèle (spécifique
(HTTPS obligatoire, HTTP accepté uniquement si ``allow_insecure_http`` à ``openai-compatible``), puis délègue les règles structurelles
vaut ``True``), la présence d'un hostname non vide, l'absence partagées à :func:`_validate_base_url`. En cas d'échec, un
d'identifiants dans le netloc et de paramètres sensibles dans la avertissement est journalisé et ``None`` est retourné : la synthèse IA
requête (y compris les paramètres sans valeur). Une URL malformée se dégrade silencieusement, sans jamais lever d'exception.
(``ValueError`` levé par ``urlparse``) est également rejetée. En cas
d'échec, un avertissement est journalisé (l'URL est toujours masquée
via :func:`redact_url`) et ``None`` est retourné : la synthèse IA se
dégrade silencieusement, sans jamais lever d'exception.
:param url: URL de base de l'API compatible OpenAI. :param url: URL de base de l'API compatible OpenAI.
:param model: Identifiant du modèle à utiliser. :param model: Identifiant du modèle à utiliser.
@@ -43,42 +106,7 @@ def _validate_openai_compatible_config(
if not model: if not model:
logger.warning("Modèle requis pour le provider openai-compatible") logger.warning("Modèle requis pour le provider openai-compatible")
return None return None
return _validate_base_url("openai-compatible", url, allow_insecure_http)
try:
parsed = urlparse(url)
except ValueError:
logger.warning(
"URL invalide pour le provider openai-compatible : %s",
redact_url(url),
)
return None
if not parsed.hostname:
logger.warning(
"URL sans hostname pour le provider openai-compatible : %s",
redact_url(url),
)
return None
if parsed.scheme not in ("http", "https"):
logger.warning(
"Schéma d'URL non supporté pour le provider openai-compatible : %s",
redact_url(url),
)
return None
if parsed.scheme == "http" and not allow_insecure_http:
logger.warning(
"URL HTTP non autorisée sans AI_ALLOW_INSECURE_HTTP=true : %s",
redact_url(url),
)
return None
if parsed.username is not None or parsed.password is not None:
logger.warning("Credentials dans l'URL refusés : %s", redact_url(url))
return None
sensitive_names = {"token", "key", "api_key", "secret", "password", "auth"}
param_names = [name.lower() for name, _ in parse_qsl(parsed.query, keep_blank_values=True)]
if any(name in sensitive_names for name in param_names):
logger.warning("Paramètres sensibles dans l'URL refusés : %s", redact_url(url))
return None
return url
def get_synthesis_provider(settings: AISettings) -> SynthesisProvider | None: def get_synthesis_provider(settings: AISettings) -> SynthesisProvider | None:
@@ -87,14 +115,15 @@ def get_synthesis_provider(settings: AISettings) -> SynthesisProvider | None:
Retourne ``None`` lorsque la synthèse IA est désactivée ou qu'aucune clé Retourne ``None`` lorsque la synthèse IA est désactivée ou qu'aucune clé
API n'est configurée. Pour le provider ``litellm``, le paquet ``litellm`` API n'est configurée. Pour le provider ``litellm``, le paquet ``litellm``
(extra ``ai-litellm``) est requis : s'il est absent, un avertissement est (extra ``ai-litellm``) est requis : s'il est absent, un avertissement est
journalisé et ``None`` est retourné. Pour le provider journalisé et ``None`` est retourné. Pour ``openai`` et ``litellm``,
``openai-compatible``, la configuration (URL de base et modèle) est une ``base_url`` éventuelle est validée par :func:`_validate_base_url` ;
validée par :func:`_validate_openai_compatible_config` ; en cas de pour le provider ``openai-compatible``, la configuration (URL de base
rejet, ``None`` est retourné avec un avertissement. et modèle) est validée par :func:`_validate_openai_compatible_config` ;
en cas de rejet, ``None`` est retourné avec un avertissement.
:param settings: Paramètres IA. :param settings: Paramètres IA.
:return: Le fournisseur configuré, ou ``None`` si désactivé, sans clé API :return: Le fournisseur configuré, ou ``None`` si désactivé, sans clé API
ou avec une configuration ``openai-compatible`` invalide. ou avec une configuration invalide.
:rtype: SynthesisProvider | None :rtype: SynthesisProvider | None
""" """
if not settings.enabled: if not settings.enabled:
@@ -111,6 +140,10 @@ def get_synthesis_provider(settings: AISettings) -> SynthesisProvider | None:
except ImportError: except ImportError:
logger.warning("Extra 'ai-litellm' requis pour le provider litellm") logger.warning("Extra 'ai-litellm' requis pour le provider litellm")
return None return None
if base_url is not None and (
_validate_base_url("litellm", base_url, settings.allow_insecure_http) is None
):
return None
return LiteLLMSynthesisProvider(api_key=settings.api_key, base_url=base_url, model=model) return LiteLLMSynthesisProvider(api_key=settings.api_key, base_url=base_url, model=model)
if settings.provider == "openai-compatible": if settings.provider == "openai-compatible":
@@ -121,4 +154,8 @@ def get_synthesis_provider(settings: AISettings) -> SynthesisProvider | None:
return None return None
return OpenAISynthesisProvider(api_key=settings.api_key, base_url=url, model=model) return OpenAISynthesisProvider(api_key=settings.api_key, base_url=url, model=model)
if base_url is not None and (
_validate_base_url("openai", base_url, settings.allow_insecure_http) is None
):
return None
return OpenAISynthesisProvider(api_key=settings.api_key, base_url=base_url, model=model) return OpenAISynthesisProvider(api_key=settings.api_key, base_url=base_url, model=model)
+466 -1
View File
@@ -11,7 +11,7 @@ factory de sélection, en vérifiant :
from __future__ import annotations from __future__ import annotations
from datetime import date, datetime, time from datetime import date, datetime, time
from typing import TYPE_CHECKING, Any from typing import TYPE_CHECKING, Any, Literal
from unittest.mock import MagicMock from unittest.mock import MagicMock
import pytest import pytest
@@ -29,6 +29,7 @@ from pronote_sync.models.diff import AgendaChange, AgendaChangeType, AgendaDiff
from pronote_sync.models.message import Message, MessageType from pronote_sync.models.message import Message, MessageType
from pronote_sync.models.synthesis import SynthesisInput from pronote_sync.models.synthesis import SynthesisInput
from pronote_sync.synthesis import get_synthesis_provider from pronote_sync.synthesis import get_synthesis_provider
from pronote_sync.synthesis.litellm import LiteLLMSynthesisProvider
from pronote_sync.synthesis.openai import OpenAISynthesisProvider from pronote_sync.synthesis.openai import OpenAISynthesisProvider
from pronote_sync.synthesis.provider import SynthesisProvider from pronote_sync.synthesis.provider import SynthesisProvider
@@ -872,6 +873,70 @@ def test_openai_compatible_valid_https_url_accepted() -> None:
assert isinstance(result, OpenAISynthesisProvider) assert isinstance(result, OpenAISynthesisProvider)
def test_openai_compatible_url_returned_unchanged() -> None:
"""Vérifie que l'URL est retournée strictement inchangée, sans manipulation de /v1."""
custom_url = "https://api.example.com/custom/path?query=value"
settings = AISettings(
enabled=True,
api_key=SecretStr("test"),
provider="openai-compatible",
base_url=custom_url,
model="test-model",
)
result = get_synthesis_provider(settings)
assert isinstance(result, OpenAISynthesisProvider)
# OpenAI SDK appends a trailing slash to base_url, so we check the string representation
assert str(result._client.base_url).rstrip("/") == custom_url
def test_openai_url_returned_unchanged() -> None:
"""Vérifie que l'URL est retournée strictement inchangée pour openai."""
custom_url = "https://api.example.com/custom/path?query=value"
settings = AISettings(
enabled=True,
api_key=SecretStr("test"),
provider="openai",
base_url=custom_url,
)
result = get_synthesis_provider(settings)
assert isinstance(result, OpenAISynthesisProvider)
# OpenAI SDK appends a trailing slash to base_url, so we check the string representation
assert str(result._client.base_url).rstrip("/") == custom_url
def test_litellm_url_returned_unchanged() -> None:
"""Vérifie que l'URL est retournée strictement inchangée pour litellm."""
pytest.importorskip("litellm")
custom_url = "https://api.example.com/custom/path?query=value"
settings = AISettings(
enabled=True,
api_key=SecretStr("test"),
provider="litellm",
base_url=custom_url,
)
result = get_synthesis_provider(settings)
assert result is not None
assert isinstance(result, LiteLLMSynthesisProvider)
assert result._base_url == custom_url
def test_openai_compatible_malformed_port_refused(
caplog: pytest.LogCaptureFixture,
) -> None:
"""Vérifie qu'un port malformé est refusé pour openai-compatible."""
settings = AISettings(
enabled=True,
api_key=SecretStr("test"),
provider="openai-compatible",
base_url="https://host:bad/v1",
model="test-model",
)
result = get_synthesis_provider(settings)
assert result is None
assert "URL invalide" in caplog.text
assert "openai-compatible" in caplog.text
def test_openai_compatible_http_refused_by_default( def test_openai_compatible_http_refused_by_default(
caplog: pytest.LogCaptureFixture, caplog: pytest.LogCaptureFixture,
) -> None: ) -> None:
@@ -887,6 +952,7 @@ def test_openai_compatible_http_refused_by_default(
result = get_synthesis_provider(settings) result = get_synthesis_provider(settings)
assert result is None assert result is None
assert "URL HTTP non autorisée sans AI_ALLOW_INSECURE_HTTP=true" in caplog.text assert "URL HTTP non autorisée sans AI_ALLOW_INSECURE_HTTP=true" in caplog.text
assert "openai-compatible" in caplog.text
def test_openai_compatible_http_accepted_with_allow_insecure_http() -> None: def test_openai_compatible_http_accepted_with_allow_insecure_http() -> None:
@@ -917,6 +983,7 @@ def test_openai_compatible_credentials_in_url_refused(
result = get_synthesis_provider(settings) result = get_synthesis_provider(settings)
assert result is None assert result is None
assert "Credentials dans l'URL refusés" in caplog.text assert "Credentials dans l'URL refusés" in caplog.text
assert "openai-compatible" in caplog.text
def test_openai_compatible_sensitive_query_params_refused( def test_openai_compatible_sensitive_query_params_refused(
@@ -933,6 +1000,24 @@ def test_openai_compatible_sensitive_query_params_refused(
result = get_synthesis_provider(settings) result = get_synthesis_provider(settings)
assert result is None assert result is None
assert "Paramètres sensibles dans l'URL refusés" in caplog.text assert "Paramètres sensibles dans l'URL refusés" in caplog.text
assert "openai-compatible" in caplog.text
def test_openai_compatible_sensitive_query_params_valueless_refused(
caplog: pytest.LogCaptureFixture,
) -> None:
"""Vérifie que les query params sensibles sans valeur sont refusés pour openai-compatible."""
settings = AISettings(
enabled=True,
api_key=SecretStr("test"),
provider="openai-compatible",
base_url="https://host/v1?token",
model="test-model",
)
result = get_synthesis_provider(settings)
assert result is None
assert "Paramètres sensibles dans l'URL refusés" in caplog.text
assert "openai-compatible" in caplog.text
def test_openai_compatible_connection_error_returns_none( def test_openai_compatible_connection_error_returns_none(
@@ -991,3 +1076,383 @@ def test_openai_compatible_factory_no_network_calls(
assert isinstance(result, OpenAISynthesisProvider) assert isinstance(result, OpenAISynthesisProvider)
mock_get.assert_not_called() mock_get.assert_not_called()
mock_post.assert_not_called() mock_post.assert_not_called()
# --- Tests de validation AI_BASE_URL pour openai et litellm (Issue #17) ---
def test_openai_valid_https_base_url_accepted() -> None:
"""Vérifie qu'une URL HTTPS valide est acceptée pour openai."""
settings = AISettings(
enabled=True,
api_key=SecretStr("test"),
provider="openai",
base_url="https://api.openai.com/v1",
)
result = get_synthesis_provider(settings)
assert isinstance(result, OpenAISynthesisProvider)
def test_openai_http_refused_by_default(
caplog: pytest.LogCaptureFixture,
) -> None:
"""Vérifie que HTTP est refusé par défaut pour openai."""
settings = AISettings(
enabled=True,
api_key=SecretStr("test"),
provider="openai",
base_url="http://127.0.0.1:11434/v1",
allow_insecure_http=False,
)
result = get_synthesis_provider(settings)
assert result is None
assert "URL HTTP non autorisée sans AI_ALLOW_INSECURE_HTTP=true" in caplog.text
assert "openai" in caplog.text
def test_openai_http_accepted_with_allow_insecure_http() -> None:
"""Vérifie que HTTP est accepté avec allow_insecure_http=True pour openai."""
settings = AISettings(
enabled=True,
api_key=SecretStr("test"),
provider="openai",
base_url="http://127.0.0.1:11434/v1",
allow_insecure_http=True,
)
result = get_synthesis_provider(settings)
assert isinstance(result, OpenAISynthesisProvider)
def test_openai_credentials_in_url_refused(
caplog: pytest.LogCaptureFixture,
) -> None:
"""Vérifie que les credentials dans l'URL sont refusés pour openai."""
settings = AISettings(
enabled=True,
api_key=SecretStr("test"),
provider="openai",
base_url="https://user:pass@host/v1", # pragma: allowlist secret
)
result = get_synthesis_provider(settings)
assert result is None
assert "Credentials dans l'URL refusés" in caplog.text
assert "openai" in caplog.text
def test_openai_sensitive_query_params_refused(
caplog: pytest.LogCaptureFixture,
) -> None:
"""Vérifie que les query params sensibles sont refusés pour openai."""
settings = AISettings(
enabled=True,
api_key=SecretStr("test"),
provider="openai",
base_url="https://host/v1?token=secret",
)
result = get_synthesis_provider(settings)
assert result is None
assert "Paramètres sensibles dans l'URL refusés" in caplog.text
assert "openai" in caplog.text
def test_openai_sensitive_query_params_valueless_refused(
caplog: pytest.LogCaptureFixture,
) -> None:
"""Vérifie que les query params sensibles sans valeur sont refusés pour openai."""
settings = AISettings(
enabled=True,
api_key=SecretStr("test"),
provider="openai",
base_url="https://host/v1?token",
)
result = get_synthesis_provider(settings)
assert result is None
assert "Paramètres sensibles dans l'URL refusés" in caplog.text
assert "openai" in caplog.text
def test_openai_malformed_url_refused(
caplog: pytest.LogCaptureFixture,
) -> None:
"""Vérifie qu'une URL malformée est refusée pour openai."""
settings = AISettings(
enabled=True,
api_key=SecretStr("test"),
provider="openai",
base_url="not-a-valid-url",
)
result = get_synthesis_provider(settings)
assert result is None
assert "URL invalide" in caplog.text
assert "openai" in caplog.text
def test_openai_no_hostname_url_refused(
caplog: pytest.LogCaptureFixture,
) -> None:
"""Vérifie qu'une URL sans hostname est refusée pour openai."""
settings = AISettings(
enabled=True,
api_key=SecretStr("test"),
provider="openai",
base_url="https:///v1",
)
result = get_synthesis_provider(settings)
assert result is None
assert "URL sans hostname" in caplog.text
assert "openai" in caplog.text
def test_openai_malformed_port_refused(
caplog: pytest.LogCaptureFixture,
) -> None:
"""Vérifie qu'un port malformé est refusé pour openai."""
settings = AISettings(
enabled=True,
api_key=SecretStr("test"),
provider="openai",
base_url="https://host:bad/v1",
)
result = get_synthesis_provider(settings)
assert result is None
assert "URL invalide" in caplog.text
assert "openai" in caplog.text
def test_openai_no_network_calls_during_validation(
mocker: MockerFixture,
) -> None:
"""Vérifie qu'aucun appel réseau n'est effectué pendant la validation pour openai."""
mock_get = mocker.patch("requests.get")
mock_post = mocker.patch("requests.post")
settings = AISettings(
enabled=True,
api_key=SecretStr("test"),
provider="openai",
base_url="https://api.example.com/v1",
)
result = get_synthesis_provider(settings)
assert isinstance(result, OpenAISynthesisProvider)
mock_get.assert_not_called()
mock_post.assert_not_called()
def test_openai_sentinel_key_not_in_logs(
caplog: pytest.LogCaptureFixture,
) -> None:
"""Vérifie qu'une clé sentinelle est absente des logs pour openai."""
sentinel = "sk-SENTINEL-OPENAI-BASE-URL-12345"
settings = AISettings(
enabled=True,
api_key=SecretStr(sentinel),
provider="openai",
base_url="https://user:pass@host/v1", # pragma: allowlist secret
)
result = get_synthesis_provider(settings)
assert result is None
assert sentinel not in caplog.text
def test_litellm_valid_https_base_url_accepted() -> None:
"""Vérifie qu'une URL HTTPS valide est acceptée pour litellm."""
pytest.importorskip("litellm")
settings = AISettings(
enabled=True,
api_key=SecretStr("test"),
provider="litellm",
base_url="https://api.litellm.ai/v1",
)
result = get_synthesis_provider(settings)
assert result is not None
def test_litellm_http_refused_by_default(
caplog: pytest.LogCaptureFixture,
) -> None:
"""Vérifie que HTTP est refusé par défaut pour litellm."""
pytest.importorskip("litellm")
settings = AISettings(
enabled=True,
api_key=SecretStr("test"),
provider="litellm",
base_url="http://127.0.0.1:11434/v1",
allow_insecure_http=False,
)
result = get_synthesis_provider(settings)
assert result is None
assert "URL HTTP non autorisée sans AI_ALLOW_INSECURE_HTTP=true" in caplog.text
assert "litellm" in caplog.text
def test_litellm_http_accepted_with_allow_insecure_http() -> None:
"""Vérifie que HTTP est accepté avec allow_insecure_http=True pour litellm."""
pytest.importorskip("litellm")
settings = AISettings(
enabled=True,
api_key=SecretStr("test"),
provider="litellm",
base_url="http://127.0.0.1:11434/v1",
allow_insecure_http=True,
)
result = get_synthesis_provider(settings)
assert result is not None
def test_litellm_credentials_in_url_refused(
caplog: pytest.LogCaptureFixture,
) -> None:
"""Vérifie que les credentials dans l'URL sont refusés pour litellm."""
pytest.importorskip("litellm")
settings = AISettings(
enabled=True,
api_key=SecretStr("test"),
provider="litellm",
base_url="https://user:pass@host/v1", # pragma: allowlist secret
)
result = get_synthesis_provider(settings)
assert result is None
assert "Credentials dans l'URL refusés" in caplog.text
assert "litellm" in caplog.text
def test_litellm_sensitive_query_params_refused(
caplog: pytest.LogCaptureFixture,
) -> None:
"""Vérifie que les query params sensibles sont refusés pour litellm."""
pytest.importorskip("litellm")
settings = AISettings(
enabled=True,
api_key=SecretStr("test"),
provider="litellm",
base_url="https://host/v1?token=secret",
)
result = get_synthesis_provider(settings)
assert result is None
assert "Paramètres sensibles dans l'URL refusés" in caplog.text
assert "litellm" in caplog.text
def test_litellm_sensitive_query_params_valueless_refused(
caplog: pytest.LogCaptureFixture,
) -> None:
"""Vérifie que les query params sensibles sans valeur sont refusés pour litellm."""
pytest.importorskip("litellm")
settings = AISettings(
enabled=True,
api_key=SecretStr("test"),
provider="litellm",
base_url="https://host/v1?token",
)
result = get_synthesis_provider(settings)
assert result is None
assert "Paramètres sensibles dans l'URL refusés" in caplog.text
assert "litellm" in caplog.text
def test_litellm_malformed_url_refused(
caplog: pytest.LogCaptureFixture,
) -> None:
"""Vérifie qu'une URL malformée est refusée pour litellm."""
pytest.importorskip("litellm")
settings = AISettings(
enabled=True,
api_key=SecretStr("test"),
provider="litellm",
base_url="not-a-valid-url",
)
result = get_synthesis_provider(settings)
assert result is None
assert "URL invalide" in caplog.text
assert "litellm" in caplog.text
def test_litellm_no_hostname_url_refused(
caplog: pytest.LogCaptureFixture,
) -> None:
"""Vérifie qu'une URL sans hostname est refusée pour litellm."""
pytest.importorskip("litellm")
settings = AISettings(
enabled=True,
api_key=SecretStr("test"),
provider="litellm",
base_url="https:///v1",
)
result = get_synthesis_provider(settings)
assert result is None
assert "URL sans hostname" in caplog.text
assert "litellm" in caplog.text
def test_litellm_malformed_port_refused(
caplog: pytest.LogCaptureFixture,
) -> None:
"""Vérifie qu'un port malformé est refusé pour litellm."""
pytest.importorskip("litellm")
settings = AISettings(
enabled=True,
api_key=SecretStr("test"),
provider="litellm",
base_url="https://host:bad/v1",
)
result = get_synthesis_provider(settings)
assert result is None
assert "URL invalide" in caplog.text
assert "litellm" in caplog.text
def test_litellm_no_network_calls_during_validation(
mocker: MockerFixture,
) -> None:
"""Vérifie qu'aucun appel réseau n'est effectué pendant la validation pour litellm."""
pytest.importorskip("litellm")
mock_get = mocker.patch("requests.get")
mock_post = mocker.patch("requests.post")
settings = AISettings(
enabled=True,
api_key=SecretStr("test"),
provider="litellm",
base_url="https://api.example.com/v1",
)
result = get_synthesis_provider(settings)
assert result is not None
mock_get.assert_not_called()
mock_post.assert_not_called()
def test_litellm_sentinel_key_not_in_logs(
caplog: pytest.LogCaptureFixture,
) -> None:
"""Vérifie qu'une clé sentinelle est absente des logs pour litellm."""
pytest.importorskip("litellm")
sentinel = "sk-SENTINEL-LITELLM-BASE-URL-67890"
settings = AISettings(
enabled=True,
api_key=SecretStr(sentinel),
provider="litellm",
base_url="https://user:pass@host/v1", # pragma: allowlist secret
)
result = get_synthesis_provider(settings)
assert result is None
assert sentinel not in caplog.text
@pytest.mark.parametrize("provider", ["openai", "litellm", "openai-compatible"])
def test_all_providers_http_refused_same_warning(
provider: Literal["openai", "litellm", "openai-compatible"], caplog: pytest.LogCaptureFixture
) -> None:
"""Vérifie que tous les providers émettent le même message d'avertissement pour HTTP refusé."""
if provider == "litellm":
pytest.importorskip("litellm")
settings = AISettings(
enabled=True,
api_key=SecretStr("test"),
provider=provider,
base_url="http://127.0.0.1:11434/v1",
allow_insecure_http=False,
model="test-model" if provider == "openai-compatible" else None,
)
result = get_synthesis_provider(settings)
assert result is None
assert "URL HTTP non autorisée sans AI_ALLOW_INSECURE_HTTP=true" in caplog.text