Compare commits

..
18 changed files with 364 additions and 434 deletions
+3 -5
View File
@@ -1,7 +1,6 @@
# --- Pronote ---
PRONOTE_ICAL_ENDPOINT__URL=https://college.ent/pronote/ical/Edt_Jean.ics?icalsecurise=REPLACE_ME&version=2024
PRONOTE_ENDPOINT__URL=https://college.ent/pronote/parent.html
# Anciens noms temporairement supportés avec un avertissement : PRONOTE_ICAL_URL, PRONOTE_URL
PRONOTE_ICAL_URL=https://college.ent/pronote/ical/Edt_Jean.ics?icalsecurise=REPLACE_ME&version=2024
PRONOTE_URL=https://college.ent/pronote/parent.html
PRONOTE_ACCOUNT_TYPE=parent
PRONOTE_USERNAME=parent.dupont
PRONOTE_PASSWORD=your_secure_password
@@ -95,8 +94,7 @@ AI_BASE_URL=https://api.openai.com/v1
# --- Blog ---
BLOG_ENABLED=false
BLOG_ENDPOINT__URL=https://blogpeda.ac-bordeaux.fr/cjeliote/?feed=rss2
# Ancien nom temporairement supporté avec un avertissement de dépréciation : BLOG_RSS_URL
BLOG_RSS_URL=https://blogpeda.ac-bordeaux.fr/cjeliote/?feed=rss2
# --- Divers ---
DRY_RUN=false
+2 -2
View File
@@ -140,7 +140,7 @@
"filename": "GUIDE_DEV_PYTHON.md",
"hashed_secret": "90bd1b48e958257948487b90bee080ba5ed00caa",
"is_verified": false,
"line_number": 5188
"line_number": 5186
}
],
"tests/unit/test_caldav_gateway.py": [
@@ -185,5 +185,5 @@
}
]
},
"generated_at": "2026-09-13T09:54:04Z"
"generated_at": "2026-09-12T22:12:56Z"
}
+29 -33
View File
@@ -259,8 +259,8 @@ Le projet utilise **`pydantic-settings`** pour valider et charger la configurati
| Variable | Description | Exemple (anonymisé) | Type |
|------------------------------|-----------------------------------------------------------------------------|---------------------------------------------|---------------|
| `PRONOTE_ENDPOINT__URL` | URL de la page Pronote via l'endpoint commun. | `https://college.ent/pronote/parent.html` | `ExternalEndpoint` |
| `PRONOTE_ICAL_ENDPOINT__URL` | URL du flux iCal Pronote via l'endpoint commun (contient `icalsecurise`). | `https://college.ent/pronote/ical/...` | `ExternalEndpoint` |
| `PRONOTE_URL` | URL de la page Pronote utilisée par `pronotepy` (page parent). | `https://college.ent/pronote/parent.html` | `str` |
| `PRONOTE_ICAL_URL` | URL du flux iCal Pronote (contient `icalsecurise`). | `https://college.ent/pronote/ical/...` | `SecretStr` |
| `PRONOTE_USERNAME` | Identifiant Pronote (si `pronotepy` utilisé). | `parent.dupont` | `str` |
| `PRONOTE_PASSWORD` | Mot de passe Pronote (si `pronotepy` utilisé). | `SecretStr` (masqué) | `SecretStr` |
| `PRONOTE_ENT` | Slug ENT supporté, résolu vers une fonction de `pronotepy.ent`. | `monbureaunumerique` | `str` |
@@ -277,22 +277,17 @@ Le projet utilise **`pydantic-settings`** pour valider et charger la configurati
> `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_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_ENDPOINT__URL` a été ajoutée dans `.env.example`.
> `BLOG_RSS_URL` reste temporairement accepté avec un avertissement de dépréciation.
> `CALDAV_URL` reste temporairement accepté avec un avertissement de dépréciation ; utiliser
> `CALDAV_ENDPOINT__URL` pour la nouvelle configuration.
> `PRONOTE_URL` et `PRONOTE_ICAL_URL` restent temporairement acceptés avec un avertissement de
> dépréciation ; utiliser respectivement `PRONOTE_ENDPOINT__URL` et `PRONOTE_ICAL_ENDPOINT__URL`.
> 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 :
- la source iCal exige `PRONOTE_ICAL_ENDPOINT__URL` ;
- la source `pronotepy` exige `PRONOTE_ENDPOINT__URL`, `PRONOTE_USERNAME` et
- la source iCal exige `PRONOTE_ICAL_URL` ;
- la source `pronotepy` exige `PRONOTE_URL`, `PRONOTE_USERNAME` et
`PRONOTE_PASSWORD` ;
- `PRONOTE_ENT` reste optionnel pour une connexion directe, mais, s'il est fourni, son slug doit
appartenir à une liste fermée et être résolu vers la fonction correspondante de `pronotepy.ent`.
`PRONOTE_ENDPOINT__URL` et `PRONOTE_ICAL_ENDPOINT__URL` sont deux contrats distincts : l'un ne doit jamais être déduit
`PRONOTE_URL` et `PRONOTE_ICAL_URL` sont deux contrats distincts : l'un ne doit jamais être déduit
de l'autre. Le cas d'usage actuel est un compte parent ; le client à construire est donc
`pronotepy.ParentClient`. Une généralisation à plusieurs profils ne sera ajoutée qu'en présence
d'un besoin réel et testé.
@@ -328,8 +323,8 @@ d'un besoin réel et testé.
```ini
# --- Pronote ---
PRONOTE_ENDPOINT__URL=https://college.ent/pronote/parent.html
PRONOTE_ICAL_ENDPOINT__URL=https://college.ent/pronote/ical/Edt_Jean.ics?icalsecurise=REPLACE_ME&version=2024
PRONOTE_URL=https://college.ent/pronote/parent.html
PRONOTE_ICAL_URL=https://college.ent/pronote/ical/Edt_Jean.ics?icalsecurise=REPLACE_ME&version=2024
PRONOTE_USERNAME=parent.dupont
PRONOTE_PASSWORD=your_secure_password
PRONOTE_ENT=monbureaunumerique
@@ -340,7 +335,7 @@ PRONOTE_HOMEWORK_SOURCE=auto
PRONOTE_MESSAGES_SOURCE=pronotepy
# --- CalDAV ---
CALDAV_ENDPOINT__URL=https://caldav.example.com/calendars/user/pronote/
CALDAV_URL=https://caldav.example.com/calendars/user/pronote/
CALDAV_ALLOW_INSECURE_HTTP=false
CALDAV_USERNAME=user@example.com
CALDAV_PASSWORD=your_caldav_password
@@ -394,7 +389,7 @@ LOG_LEVEL=INFO
> L'implémentation utilise le style moderne de Pydantic v2 : `model_config = ConfigDict(frozen=True)` au lieu de `class Config`, pas de `json_encoders` (la sérialisation ISO est native en v2), `str | None` au lieu de `Optional[str]`, `list[str]` au lieu de `List[str]`.
> `AISettings.enabled` a pour valeur par défaut `False`.
> `XmppSettings` est entièrement défini en §10.2.3 avec tous les champs optionnels (valeurs par défaut) pour que `Settings()` fonctionne sans `.env`.
> `BlogSettings` a été ajouté (§5 bis.9.2) avec `enabled=False` et un endpoint RSS par défaut.
> `BlogSettings` a été ajouté (§5 bis.9.2) avec `enabled=False` et `rss_url` par défaut.
> `sync_past_days` et `sync_future_days` sont dans `AppSettings`, et non `CalDAVSettings`.
> `CalDAVSettings.calendar_path` a pour valeur par défaut `"/pronote-sync/"`.
> `XmppSettings.resource` a pour valeur par défaut `"pronote-sync"`.
@@ -408,9 +403,9 @@ from pydantic_settings import BaseSettings, SettingsConfigDict
class PronoteSettings(BaseSettings):
model_config = SettingsConfigDict(env_prefix="PRONOTE_", env_file=".env", env_nested_delimiter="__", extra="ignore")
endpoint: ExternalEndpoint | None = None
ical_endpoint: ExternalEndpoint | None = None
model_config = SettingsConfigDict(env_prefix="PRONOTE_", env_file=".env", extra="ignore")
url: str | None = None
ical_url: SecretStr | None = None
username: str | None = None
password: SecretStr | None = None
ent: str | None = None
@@ -1118,7 +1113,7 @@ La déduplication des articles du blog repose sur leur **GUID** (ou leur URL si
Aucun fichier d'état local n'est utilisé : l'état est géré en mémoire par run.
# Initialisation
rss_client = BlogRSSClient(rss_url=settings.blog.endpoint.url.get_secret_value())
rss_client = BlogRSSClient(rss_url=settings.blog.rss_url)
blog_state = ## (section obsolète supprimée)()
# Récupération des nouveaux articles
@@ -1223,7 +1218,7 @@ Ajouter les variables suivantes dans la configuration :
| **Variable** | **Description** | **Valeur par défaut** | **Type** |
|----------------------------|-------------------------------------------------------------------------------|-----------------------|-------------------|
| `BLOG_ENABLED` | Activer la récupération du blog. | `False` | `bool` |
| `BLOG_ENDPOINT__URL` | URL du flux RSS via l'endpoint commun (masquée en `SecretStr`). | `https://blogpeda.ac-bordeaux.fr/cjeliote/?feed=rss2` | `ExternalEndpoint` |
| `BLOG_RSS_URL` | URL du flux RSS du blog. | `https://blogpeda.ac-bordeaux.fr/cjeliote/?feed=rss2` | `str` |
#### 5 bis.9.2 Modèle Pydantic pour la configuration du blog
@@ -1235,7 +1230,10 @@ from pydantic_settings import BaseSettings, SettingsConfigDict
class BlogSettings(BaseSettings):
model_config = SettingsConfigDict(env_prefix="BLOG_", env_file=".env", extra="ignore")
enabled: bool = Field(False, description="Activer la récupération du blog")
endpoint: ExternalEndpoint
rss_url: str = Field(
"https://blogpeda.ac-bordeaux.fr/cjeliote/?feed=rss2",
description="URL du flux RSS du blog",
)
```
**Intégration dans `Settings`** :
@@ -1255,7 +1253,7 @@ class Settings(BaseSettings):
```ini
# --- Blog du collège ---
BLOG_ENABLED=true
BLOG_ENDPOINT__URL=https://blogpeda.ac-bordeaux.fr/cjeliote/?feed=rss2
BLOG_RSS_URL=https://blogpeda.ac-bordeaux.fr/cjeliote/?feed=rss2
```
---
@@ -2707,7 +2705,7 @@ Utilisation de la bibliothèque [`caldav`](https://pypi.org/project/caldav/) (Py
#### API réelle (`caldav>=1.3.0`)
- **Connexion** : `caldav.DAVClient(url, username, password)` — les paramètres proviennent
de `CalDAVSettings` (`CALDAV_ENDPOINT__URL`, `CALDAV_USERNAME`, `CALDAV_PASSWORD`).
de `CalDAVSettings` (`CALDAV_URL`, `CALDAV_USERNAME`, `CALDAV_PASSWORD`).
- **Résolution du calendrier** : `DAVClient.principal()` puis `principal.calendars()` ;
sélectionner le calendrier dont l'URL correspond à **`CalDAVSettings.calendar_path`**
(ex: `/pronote-sync/`). La résolution ne se fait **pas** par nom de calendrier :
@@ -4612,7 +4610,7 @@ def get_channel(settings: XmppSettings, dry_run: bool = False) -> Channel | None
- **XMPP** : Si l'envoi échoue → **logger l'erreur** mais continuer le pipeline.
- **Erreurs critiques** :
- **Aucune source disponible** (iCal + pronotepy échouent) → **échec explicite** avec message clair.
- **Configuration invalide** (ex: `PRONOTE_ICAL_ENDPOINT__URL` manquant) → **échec explicite**.
- **Configuration invalide** (ex: `PRONOTE_ICAL_URL` manquant) → **échec explicite**.
### 11.2 Hiérarchie des erreurs
@@ -5343,14 +5341,12 @@ def mock_xmpp_channel():
def sample_settings():
"""Retourne une configuration de test."""
from pydantic import SecretStr
from pronote_sync.config.settings import (
AISettings, AppSettings, CalDAVSettings, ExternalEndpoint, PronoteSettings, Settings, XmppSettings,
)
from pronote_sync.config.settings import Settings, PronoteSettings, CalDAVSettings, XmppSettings, AISettings, AppSettings
return Settings(
pronote=PronoteSettings(
endpoint=ExternalEndpoint(url=SecretStr("https://test.ent/pronote/parent.html")),
ical_endpoint=ExternalEndpoint(url=SecretStr("https://test.ent/pronote/ical/test.ics")),
url="https://test.ent/pronote/parent.html",
ical_url=SecretStr("https://test.ent/pronote/ical/test.ics"),
username="test_user",
password=SecretStr("test_password"),
ent="monbureaunumerique",
@@ -5438,7 +5434,7 @@ def test_pipeline_full(mock_requests_get, mock_caldav_client, mock_ai_provider,
# Configurer le client CalDAV
caldav_client = CalDAVClient(
url=sample_settings.caldav.endpoint.url if sample_settings.caldav.endpoint else None,
url=sample_settings.caldav.url,
username=sample_settings.caldav.username,
password=sample_settings.caldav.password,
allow_insecure_http=sample_settings.caldav.allow_insecure_http,
@@ -5633,7 +5629,7 @@ TOTAL 1000 10 99%
| **Risque** | **Mesure de mitigation** | **Vérification** | **Statut** |
|-------------------------------------|----------------------------------------------------------------------------------------|-------------------------------------------|------------|
| Tokens dans le code | Utiliser `pydantic-settings` + `SecretStr` pour les variables d'environnement. | `grep -r "icalsecurise\|password\|api_key" src/` | ❌ Interdit |
| Tokens dans les logs | Masquage systématique via `RedactingFormatter` (voir [Section 4.2](#42-implémentation)). | Tests avec `PRONOTE_ICAL_ENDPOINT__URL` contenant un token. | ✅ Obligatoire |
| Tokens dans les logs | Masquage systématique via `RedactingFormatter` (voir [Section 4.2](#42-implémentation)). | Tests avec `PRONOTE_ICAL_URL` contenant un token. | ✅ Obligatoire |
| Tokens dans les erreurs | Masquage dans les messages d'erreur (voir `redact_url` et `redact_secrets`). | Tests avec URLs contenant des tokens. | ✅ Obligatoire |
| Tokens dans les fixtures | **Anonymiser** toutes les fixtures (pas de tokens réels). | Vérification manuelle des fixtures. | ✅ Obligatoire |
| Tokens dans les commits Git | Utiliser `.gitignore` pour `.env` et `pre-commit` pour bloquer les secrets. | `git grep "icalsecurise\|password" -- .` (contenu suivi courant) | ❌ Interdit |
@@ -5952,10 +5948,10 @@ Exemple de ligne cron (exécution tous les jours à 18h) :
|---------------------------------------|------------------------------------------------------------------------------------|------------------------------------------------------------------------------|
| Échec de la récupération iCal | Token `icalsecurise` expiré ou invalide. | Régénérer le token depuis Pronote. |
| Échec de la connexion Pronote (`pronotepy`) | Identifiants incorrects ou ENT non supporté. | Vérifier `PRONOTE_USERNAME`, `PRONOTE_PASSWORD`, `PRONOTE_ENT`. |
| Échec de la connexion CalDAV | URL, identifiant ou mot de passe CalDAV incorrect, ou HTTP non autorisé pour l'hôte. | Vérifier `CALDAV_ENDPOINT__URL`, `CALDAV_USERNAME`, `CALDAV_PASSWORD`, `CALDAV_ALLOW_INSECURE_HTTP`. |
| Échec de la connexion CalDAV | URL, identifiant ou mot de passe CalDAV incorrect, ou HTTP non autorisé pour l'hôte. | Vérifier `CALDAV_URL`, `CALDAV_USERNAME`, `CALDAV_PASSWORD`, `CALDAV_ALLOW_INSECURE_HTTP`. |
| Échec de la connexion XMPP | Identifiant ou mot de passe XMPP incorrect. | Vérifier `XMPP_JID`, `XMPP_PASSWORD`. |
| Échec de la synthèse IA | Clé API IA invalide ou modèle non disponible. | Vérifier `AI_API_KEY`, `AI_BASE_URL`, `AI_MODEL`. |
| Aucun cours récupéré | Flux iCal vide ou `pronotepy` non configuré. | Vérifier `PRONOTE_ICAL_ENDPOINT__URL` ou les identifiants `pronotepy`. |
| Aucun cours récupéré | Flux iCal vide ou `pronotepy` non configuré. | Vérifier `PRONOTE_ICAL_URL` ou les identifiants `pronotepy`. |
| Doublons dans les devoirs | Problème de déduplication. | Vérifier la logique de déduplication (voir [Section 5.1.4](#514-déduplication-des-devoirs)). |
| Synchronisation CalDAV lente | Trop d'événements à synchroniser. | Réduire `SYNC_PAST_DAYS` ou `SYNC_FUTURE_DAYS`. |
+4 -4
View File
@@ -94,7 +94,7 @@ https://{etablissement}.index-education.net/pronote/ical/Edt_{prenom}.ics?icalse
🔹 **Source** : ⚠️ Recommandation du projet (inspirée des bonnes pratiques générales de sécurité).
### Intégration dans `pronote-sync`
- **Paramètre** : `PRONOTE_ICAL_ENDPOINT__URL` (ex. `.env.example` ligne 2).
- **Paramètre** : `PRONOTE_ICAL_URL` (ex. `.env.example` ligne 2).
- **Comportement** :
- Prioritaire en mode `PRONOTE_AGENDA_SOURCE=auto`.
- Si l'URL est invalide ou expire, repli automatique vers `pronotepy` (si `PRONOTE_AGENDA_SOURCE=auto`).
@@ -126,7 +126,7 @@ Connexion via le protocole propriétaire de Pronote (JSON sur HTTPS), avec **chi
### Intégration dans `pronote-sync`
- **Paramètres** :
- `PRONOTE_ENDPOINT__URL` (ex. `.env.example` ligne 3).
- `PRONOTE_URL` (ex. `.env.example` ligne 3).
- `PRONOTE_USERNAME`, `PRONOTE_PASSWORD`.
- `PRONOTE_ENT` (slug dans `_ENT_NAMES`).
- `PRONOTE_ACCOUNT_TYPE` (ex. `parent`).
@@ -288,8 +288,8 @@ Mécanisme d'appairage par QR code pour les appareils mobiles, **contournant l'a
### Alignement avec `.env.example`
| Paramètre | Document | Code | Statut |
|-----------|----------|------|--------|
| `PRONOTE_ICAL_ENDPOINT__URL` | ✅ Lignes 2, 4250 | ✅ `sources/ical.py` | **Cohérent** |
| `PRONOTE_ENDPOINT__URL` | ✅ Ligne 3 | ✅ `client.py` (ligne 294) | **Cohérent** |
| `PRONOTE_ICAL_URL` | ✅ Lignes 2, 4250 | ✅ `sources/ical.py` | **Cohérent** |
| `PRONOTE_URL` | ✅ Ligne 3 | ✅ `client.py` (ligne 294) | **Cohérent** |
| `PRONOTE_ENT` | ✅ Ligne 7 | ✅ `client.py` (ligne 297, `_ENT_NAMES`) | **Cohérent** |
| `PRONOTE_AUTH_MODE=qr_token` | ✅ Ligne 23 | ✅ `client.py` (ligne 334) | **Cohérent** |
| `PRONOTE_QR_CODE_FILE` | ✅ Ligne 24 | ✅ `client.py` (ligne 387) | **Cohérent** |
+15 -122
View File
@@ -88,28 +88,13 @@ class PronoteSettings(BaseSettings):
``PRONOTE_``.
"""
model_config = SettingsConfigDict(
env_file=".env",
env_nested_delimiter="__",
extra="ignore",
env_prefix="PRONOTE_",
)
model_config = SettingsConfigDict(env_file=".env", extra="ignore", env_prefix="PRONOTE_")
endpoint: ExternalEndpoint | None = None
ical_endpoint: ExternalEndpoint | None = None
ical_url: SecretStr | None = Field(
default=None,
exclude=True,
deprecated="Utiliser ical_endpoint.url à la place (PRONOTE_ICAL_URL obsolète).",
)
ical_url: SecretStr | None = None
username: str | None = None
password: SecretStr | None = None
ent: str | None = None
url: str | None = Field(
default=None,
exclude=True,
deprecated="Utiliser endpoint.url à la place (PRONOTE_URL obsolète).",
)
url: str | None = None
account_type: Literal["student", "parent"] = "parent"
agenda_source: Literal["auto", "ical", "pronotepy"] = "auto"
homework_source: Literal["auto", "ical", "pronotepy"] = "auto"
@@ -119,57 +104,17 @@ class PronoteSettings(BaseSettings):
qr_pin: SecretStr | None = None
account_pin: SecretStr | None = None
@model_validator(mode="before")
@classmethod
def _migrate_legacy_endpoints(cls, data: object) -> object:
"""Migre les URL Pronote historiques vers les endpoints communs.
@field_serializer("ical_url")
def _serialize_ical_url(self, value: SecretStr | None) -> str | None:
"""Masque l'URL iCal lors de la sérialisation (repr, str, JSON).
:param data: Données brutes du modèle.
:return: Données complétées avec les endpoints si nécessaire.
:rtype: object
:param value: Valeur du champ ``ical_url``.
:return: ``"**********"`` si la valeur est définie, ``None`` sinon.
:rtype: str | None
"""
if not isinstance(data, dict):
return data
migrated_data = data.copy()
if migrated_data.get("url") is not None:
warnings.warn(
"PRONOTE_URL est obsolète : utiliser PRONOTE_ENDPOINT__URL.",
DeprecationWarning,
stacklevel=2,
)
if migrated_data.get("endpoint") is None:
migrated_data["endpoint"] = {"url": migrated_data["url"]}
if migrated_data.get("ical_url") is not None:
warnings.warn(
"PRONOTE_ICAL_URL est obsolète : utiliser PRONOTE_ICAL_ENDPOINT__URL.",
DeprecationWarning,
stacklevel=2,
)
if migrated_data.get("ical_endpoint") is None:
migrated_data["ical_endpoint"] = {"url": migrated_data["ical_url"]}
return migrated_data
@model_validator(mode="after")
def _validate_endpoint_policies(self) -> PronoteSettings:
"""Applique les transports autorisés aux deux endpoints Pronote.
L'API Pronote utilise HTTPS. Le flux iCal accepte également ``file``
afin de préserver les fixtures locales injectées.
:return: Instance validée inchangée.
:rtype: PronoteSettings
:raises ValueError: Si un endpoint utilise un schéma interdit.
"""
if (
self.endpoint is not None
and urlparse(self.endpoint.url.get_secret_value()).scheme != "https"
):
raise ValueError("URL Pronote invalide : HTTPS requis") from None
if self.ical_endpoint is not None and urlparse(
self.ical_endpoint.url.get_secret_value()
).scheme not in {"https", "file"}:
raise ValueError("URL iCal Pronote invalide : HTTPS ou file requis") from None
return self
if value is None:
return None
return "**********"
@field_serializer("qr_pin")
def _serialize_qr_pin(self, value: SecretStr | None) -> str | None:
@@ -407,60 +352,10 @@ class BlogSettings(BaseSettings):
``BLOG_``.
"""
model_config = SettingsConfigDict(
env_file=".env",
env_nested_delimiter="__",
extra="ignore",
env_prefix="BLOG_",
)
model_config = SettingsConfigDict(env_file=".env", extra="ignore", env_prefix="BLOG_")
enabled: bool = False
endpoint: ExternalEndpoint = Field(
default_factory=lambda: ExternalEndpoint(
url=SecretStr("https://blogpeda.ac-bordeaux.fr/cjeliote/?feed=rss2")
)
)
rss_url: str | None = Field(
default=None,
exclude=True,
deprecated="Utiliser endpoint.url à la place (BLOG_RSS_URL obsolète).",
)
@model_validator(mode="before")
@classmethod
def _migrate_legacy_rss_url(cls, data: object) -> object:
"""Migre ``rss_url`` vers l'endpoint commun avec un avertissement.
:param data: Données brutes du modèle.
:return: Données complétées avec ``endpoint`` si nécessaire.
:rtype: object
"""
if not isinstance(data, dict) or data.get("rss_url") is None:
return data
migrated_data = data.copy()
warnings.warn(
"BLOG_RSS_URL est obsolète : utiliser BLOG_ENDPOINT__URL.",
DeprecationWarning,
stacklevel=2,
)
if migrated_data.get("endpoint") is None:
migrated_data["endpoint"] = {"url": migrated_data["rss_url"]}
return migrated_data
@model_validator(mode="after")
def _validate_endpoint_policy(self) -> BlogSettings:
"""Refuse les transports non sûrs pour le flux RSS de production.
Le transport ``file`` reste autorisé pour les fixtures locales.
:return: Instance validée inchangée.
:rtype: BlogSettings
:raises ValueError: Si le schéma n'est ni ``https`` ni ``file``.
"""
scheme = urlparse(self.endpoint.url.get_secret_value()).scheme
if scheme not in {"https", "file"}:
raise ValueError("URL RSS invalide : HTTPS ou file requis") from None
return self
rss_url: str = "https://blogpeda.ac-bordeaux.fr/cjeliote/?feed=rss2"
class AppSettings(BaseSettings):
@@ -513,8 +408,7 @@ class Settings(BaseSettings):
:rtype: tuple[SecretStr, ...]
"""
secrets = [
self.pronote.endpoint.url if self.pronote.endpoint is not None else None,
self.pronote.ical_endpoint.url if self.pronote.ical_endpoint is not None else None,
self.pronote.ical_url,
self.pronote.password,
self.pronote.qr_pin,
self.pronote.account_pin,
@@ -522,6 +416,5 @@ class Settings(BaseSettings):
self.caldav.password,
self.xmpp.password,
self.ai.api_key,
self.blog.endpoint.url,
]
return tuple(dict.fromkeys(secret for secret in secrets if secret is not None))
+1 -5
View File
@@ -138,11 +138,7 @@ class PipelineRunner:
comparator = (
AgendaComparator(theoretical_provider) if theoretical_provider is not None else None
)
blog_client = (
BlogRSSClient(settings.blog.endpoint.url.get_secret_value())
if settings.blog.enabled
else None
)
blog_client = BlogRSSClient(settings.blog.rss_url) if settings.blog.enabled else None
blog_state = (
BlogRSSState(persistence_enabled=persistence_enabled) if settings.blog.enabled else None
)
+5 -2
View File
@@ -121,6 +121,7 @@ def fetch_step(
:raises PronoteAuthRotationError: Si une rotation du token d'authentification
pronotepy est nécessaire : propagée telle quelle jusqu'au pipeline.
"""
critical_error: PipelineCriticalError | None = None
try:
lessons, school_events = fetcher.fetch_agenda()
target_date = resolve_target_date(today or date.today(), lessons, school_events)
@@ -130,9 +131,11 @@ def fetch_step(
except PronoteAuthRotationError:
raise
except Exception as exc:
raise PipelineCriticalError(
critical_error = PipelineCriticalError(
f"Récupération Pronote impossible : {redact_exception(exc)}", step="fetch"
) from None
)
if critical_error is not None:
raise critical_error from None
messages, warnings = _fetch_optional_messages(fetcher)
return (
+8 -5
View File
@@ -5,7 +5,7 @@ from __future__ import annotations
from pronote_sync.sources.blog.result import BlogRSSFetchResult
from pronote_sync.sources.blog.rss import BlogRSSClient
from pronote_sync.sources.blog.state import BlogRSSState
from pronote_sync.utils.redaction import redact_exception
from pronote_sync.utils.redaction import redact_exception, redact_secrets
def fetch_blog_step(client: BlogRSSClient | None, state: BlogRSSState | None) -> BlogRSSFetchResult:
@@ -24,15 +24,18 @@ def fetch_blog_step(client: BlogRSSClient | None, state: BlogRSSState | None) ->
"""
if client is None or state is None:
return BlogRSSFetchResult()
error_message: str | None = None
try:
etag, last_modified = state.get_cache_headers()
result = client.fetch_and_parse(
known_guids=state.get_known_guids(), etag=etag, last_modified=last_modified
)
if result.error is not None:
raise RuntimeError(result.error) from None
if not result.not_modified and not result.articles:
error_message = f"Récupération du blog échouée : {redact_secrets(result.error)}"
elif not result.not_modified and not result.articles:
state.update_cache_headers(result.etag, result.last_modified)
return result
except Exception as exc:
raise RuntimeError(f"Récupération du blog échouée : {redact_exception(exc)}") from None
error_message = f"Récupération du blog échouée : {redact_exception(exc)}"
if error_message is not None:
raise RuntimeError(error_message) from None
return result
+5 -2
View File
@@ -218,6 +218,7 @@ class PronoteAuthState:
}
tmp_file = self._state_file.with_suffix(".tmp")
fd: int | None = None
write_error: PronoteSyncError | None = None
try:
# Nettoie un éventuel fichier temporaire stale laissé par une exécution interrompue.
if tmp_file.exists():
@@ -259,10 +260,12 @@ class PronoteAuthState:
"Nettoyage du fichier temporaire d'état d'authentification Pronote échoué : %s",
redact_exception(cleanup_exc),
)
raise PronoteSyncError(
write_error = PronoteSyncError(
f"Impossible d'écrire le fichier d'état d'authentification Pronote "
f"{redact_secrets(str(self._state_file))}."
) from None
)
if write_error is not None:
raise write_error from None
def clear(self) -> None:
"""Supprime le fichier d'état d'authentification.
+3 -3
View File
@@ -294,18 +294,18 @@ class PronoteClient:
est manquant, ou si l'ENT fourni est inconnu.
:raises pronotepy.PronoteAPIError: Si la connexion à Pronote échoue.
"""
endpoint = self._settings.endpoint
url = self._settings.url
username = self._settings.username
password = self._settings.password
ent = self._settings.ent
if endpoint is None or username is None or password is None:
if url is None or username is None or password is None:
raise ValueError("url, username et password sont requis pour pronotepy")
resolver = _resolve_ent(ent) if ent is not None else None
client_class: type[pronotepy.Client] = (
pronotepy.ParentClient if self._settings.account_type == "parent" else pronotepy.Client
)
self._client = client_class(
pronote_url=endpoint.url.get_secret_value(),
pronote_url=url,
username=username,
password=password.get_secret_value(),
ent=resolver,
+74 -69
View File
@@ -141,10 +141,10 @@ class PronoteFetcher:
def _is_ical_configured(self) -> bool:
"""Vérifie que la source iCal est configurée.
:return: ``True`` si ``ical_endpoint`` est défini, ``False`` sinon.
:return: ``True`` si ``ical_url`` est défini, ``False`` sinon.
:rtype: bool
"""
return self._settings.pronote.ical_endpoint is not None
return self._settings.pronote.ical_url is not None
def _is_pronotepy_configured(self) -> bool:
"""Vérifie si la source pronotepy est utilisable selon le mode d'authentification.
@@ -155,13 +155,13 @@ class PronoteFetcher:
"""
pronote = self._settings.pronote
if pronote.auth_mode == "qr_token":
# En mode qr_token, seul PRONOTE_ENDPOINT__URL est requis.
# En mode qr_token, seul PRONOTE_URL est requis.
# Le QR code et le PIN ne sont nécessaires que pour l'enrôlement initial.
# Les exécutions suivantes utilisent le token persisté.
return pronote.endpoint is not None
return pronote.url is not None
# En mode password, URL + identifiant + mot de passe sont requis.
return (
pronote.endpoint is not None
pronote.url is not None
and pronote.username is not None
and pronote.password is not None
)
@@ -171,16 +171,16 @@ class PronoteFetcher:
:return: Tuple ``(cours, événements scolaires)``.
:rtype: tuple[list[Lesson], list[SchoolEvent]]
:raises ValueError: Si ``ical_endpoint`` n'est pas configuré ou si le flux est invalide.
:raises ValueError: Si ``ical_url`` n'est pas configuré ou si le flux est invalide.
:raises OSError: Si le fichier iCal local est illisible.
:raises requests.RequestException: Si la récupération HTTP échoue.
"""
if self._cache_ical_for_run and self._run_ical_agenda is not None:
return self._run_ical_agenda
ical_endpoint = self._settings.pronote.ical_endpoint
if ical_endpoint is None:
raise ValueError("PRONOTE_ICAL_ENDPOINT__URL est requis pour la source iCal")
raw_ical = fetch_ical(ical_endpoint.url.get_secret_value())
ical_url = self._settings.pronote.ical_url
if ical_url is None:
raise ValueError("PRONOTE_ICAL_URL est requis pour la source iCal")
raw_ical = fetch_ical(ical_url.get_secret_value())
lessons, _, school_events = parse_ical(raw_ical)
result = (lessons, school_events)
if self._cache_ical_for_run:
@@ -210,7 +210,7 @@ class PronoteFetcher:
Les modes explicites ``ICAL`` et ``PRONOTEPY`` désignent la seule
source utilisée, sans aucun repli. En mode ``AUTO``, iCal est
primaire si ``ical_endpoint`` est configuré (repli pronotepy si la
primaire si ``ical_url`` est configuré (repli pronotepy si la
configuration pronotepy est complète), sinon pronotepy sans repli.
:return: Tuple ``(source primaire, source de repli ou ``None``)``.
@@ -269,34 +269,36 @@ class PronoteFetcher:
primary,
redact_exception(exc),
)
if fallback is None:
raise PipelineCriticalError(
f"Impossible de récupérer l'agenda : la source {primary} a échoué"
) from None
logger.info("Repli sur %s pour l'agenda.", fallback)
try:
lessons, school_events = self._fetch_agenda_source(fallback)
except PronoteAuthRotationError:
raise
except Exception as exc:
logger.error(
"Échec de la récupération %s pour l'agenda : %s",
fallback,
redact_exception(exc),
)
raise PipelineCriticalError(
f"Impossible de récupérer l'agenda : les sources {primary}"
f" et {fallback} ont échoué"
) from None
if not lessons:
logger.warning(
"Le repli %s pour l'agenda a retourné un résultat vide après l'échec "
"de %s : impossible de distinguer une absence de cours d'un échec "
"silencieux.",
fallback,
primary,
)
return lessons, school_events
if fallback is None:
raise PipelineCriticalError(
f"Impossible de récupérer l'agenda : la source {primary} a échoué"
) from None
logger.info("Repli sur %s pour l'agenda.", fallback)
fallback_result: tuple[list[Lesson], list[SchoolEvent]] | None = None
try:
fallback_result = self._fetch_agenda_source(fallback)
except PronoteAuthRotationError:
raise
except Exception as exc:
logger.error(
"Échec de la récupération %s pour l'agenda : %s",
fallback,
redact_exception(exc),
)
if fallback_result is None:
raise PipelineCriticalError(
f"Impossible de récupérer l'agenda : les sources {primary} et {fallback} ont échoué"
) from None
lessons, school_events = fallback_result
if not lessons:
logger.warning(
"Le repli %s pour l'agenda a retourné un résultat vide après l'échec "
"de %s : impossible de distinguer une absence de cours d'un échec "
"silencieux.",
fallback,
primary,
)
return lessons, school_events
def _fetch_homework_ical(self, target_date: date) -> list[Homework]:
"""Récupère les devoirs depuis le flux iCal pour la date cible.
@@ -304,7 +306,7 @@ class PronoteFetcher:
:param target_date: Date cible pour laquelle collecter les devoirs.
:return: Liste des devoirs.
:rtype: list[Homework]
:raises ValueError: Si ``ical_endpoint`` n'est pas configuré ou si le flux est invalide.
:raises ValueError: Si ``ical_url`` n'est pas configuré ou si le flux est invalide.
:raises OSError: Si le fichier iCal local est illisible.
:raises requests.RequestException: Si la récupération HTTP échoue.
"""
@@ -335,7 +337,7 @@ class PronoteFetcher:
Les modes explicites ``ICAL`` et ``PRONOTEPY`` désignent la seule
source utilisée, sans aucun repli. En mode ``AUTO``, iCal est
primaire si ``ical_endpoint`` est configuré (repli pronotepy si la
primaire si ``ical_url`` est configuré (repli pronotepy si la
configuration pronotepy est complète), sinon pronotepy sans repli.
:return: Tuple ``(source primaire, source de repli ou ``None``)``.
@@ -396,34 +398,37 @@ class PronoteFetcher:
primary,
redact_exception(exc),
)
if fallback is None:
raise PipelineCriticalError(
f"Impossible de récupérer les devoirs : la source {primary} a échoué"
) from None
logger.info("Repli sur %s pour les devoirs.", fallback)
try:
homeworks = self._fetch_homework_source(fallback, target_date)
except PronoteAuthRotationError:
raise
except Exception as exc:
logger.error(
"Échec de la récupération %s pour les devoirs : %s",
fallback,
redact_exception(exc),
)
raise PipelineCriticalError(
f"Impossible de récupérer les devoirs : les sources {primary}"
f" et {fallback} ont échoué"
) from None
if not homeworks:
logger.warning(
"Le repli %s pour les devoirs a retourné un résultat vide après "
"l'échec de %s : impossible de distinguer une absence de devoirs "
"d'un échec silencieux.",
fallback,
primary,
)
return homeworks
if fallback is None:
raise PipelineCriticalError(
f"Impossible de récupérer les devoirs : la source {primary} a échoué"
) from None
logger.info("Repli sur %s pour les devoirs.", fallback)
fallback_result: list[Homework] | None = None
try:
fallback_result = self._fetch_homework_source(fallback, target_date)
except PronoteAuthRotationError:
raise
except Exception as exc:
logger.error(
"Échec de la récupération %s pour les devoirs : %s",
fallback,
redact_exception(exc),
)
if fallback_result is None:
raise PipelineCriticalError(
f"Impossible de récupérer les devoirs : les sources {primary}"
f" et {fallback} ont échoué"
) from None
homeworks = fallback_result
if not homeworks:
logger.warning(
"Le repli %s pour les devoirs a retourné un résultat vide après "
"l'échec de %s : impossible de distinguer une absence de devoirs "
"d'un échec silencieux.",
fallback,
primary,
)
return homeworks
def fetch_messages(self) -> list[Message]:
"""Récupère les messages des discussions Pronote (toujours via pronotepy).
+5 -2
View File
@@ -85,6 +85,7 @@ class JsonTheoreticalAgendaProvider:
self._file_path: str = file_path
self._parity_service: WeekParityService | None = parity_service
self._holiday_calendar: SchoolHolidayCalendar | None = holiday_calendar
load_error: PronoteSyncError | None = None
try:
content = Path(file_path).read_text(encoding="utf-8")
parsed = TheoreticalAgendaFile.model_validate_json(content)
@@ -94,9 +95,11 @@ class JsonTheoreticalAgendaProvider:
redact_secrets(str(file_path)),
redact_exception(exc),
)
raise PronoteSyncError(
load_error = PronoteSyncError(
f"Le fichier d'agenda théorique est invalide : {redact_secrets(str(file_path))}"
) from None
)
if load_error is not None:
raise load_error from None
self._lessons: tuple[TheoreticalLessonEntry, ...] = parsed.lessons
if self._parity_service is None and any(
entry.week in ("even", "odd") for entry in self._lessons
+5 -2
View File
@@ -77,6 +77,7 @@ class SchoolHolidayCalendar:
raise PronoteSyncError(
f"Le fichier de vacances scolaires est introuvable : {redact_secrets(str(path))}"
) from None
load_error: PronoteSyncError | None = None
try:
data: Any = json.loads(path.read_text(encoding="utf-8"))
file_model: SchoolHolidayFile = SchoolHolidayFile.model_validate(data)
@@ -86,9 +87,11 @@ class SchoolHolidayCalendar:
redact_secrets(str(path)),
redact_exception(exc),
)
raise PronoteSyncError(
load_error = PronoteSyncError(
f"Le fichier de vacances scolaires est invalide : {redact_secrets(str(path))}"
) from None
)
if load_error is not None:
raise load_error from None
self._periods = file_model.periods
def is_holiday(self, target_date: date) -> bool:
+31 -8
View File
@@ -26,15 +26,15 @@ _TEXT_SUFFIXES = frozenset(
{".conf", ".ini", ".json", ".md", ".py", ".service", ".timer", ".toml", ".txt", ".yaml", ".yml"}
)
_LITERAL_SECRET_RE = re.compile(
r"(?ix)\b[a-z0-9_]*(?:api[_-]?key|access[_-]?token|auth(?:orization)?|icalsecurise|password|secret|token)"
r"\s*[:=]\s*['\"][^'\"\r\n]{3,}['\"]"
r"(?ix)\b[a-z0-9_]*(?:api[_-]?key|access[_-]?token|auth(?:orization)?|icalsecurise|password|pin|secret|token)"
r"\s*[:=]\s*['\"](?P<value>[^'\"\r\n]{3,})['\"]"
)
_UNQUOTED_SECRET_RE = re.compile(
r"(?ix)\b[a-z0-9_]*(?:api[_-]?key|access[_-]?token|auth(?:orization)?|icalsecurise|password|secret|token)"
r"\s*[:=]\s*[a-z0-9][a-z0-9._~+/-]{2,}"
r"(?ix)\b[a-z0-9_]*(?:api[_-]?key|access[_-]?token|auth(?:orization)?|icalsecurise|password|pin|secret|token)"
r"\s*[:=]\s*(?P<value>[a-z0-9][a-z0-9._~+/-]{2,})"
)
_URL_SECRET_RE = re.compile(
r"(?ix)[?&](?:api[_-]?key|access[_-]?token|auth(?:orization)?|icalsecurise|password|secret|token)"
r"(?ix)[?&](?:api[_-]?key|access[_-]?token|auth(?:orization)?|icalsecurise|password|pin|secret|token)"
r"=([^&#\s]{3,})"
)
_URL_PLACEHOLDER_RE = re.compile(
@@ -47,6 +47,13 @@ _URL_PLACEHOLDER_RE = re.compile(
r")$"
)
_EXTRA_NAMES = frozenset({"pronote_sync"})
_ASSIGNMENT_PLACEHOLDER_RE = re.compile(
r"(?ix)^(?:"
r"<(?:pin|secret|valeur|value|token|jeton)>|"
r"(?:change|replace|your)[_-]?(?:me|here|value|valeur|pin|password|secret)|"
r"(?:placeholder|example|local-not-required)"
r")$"
)
@dataclass(frozen=True)
@@ -68,6 +75,16 @@ CommandRunner = Callable[..., subprocess.CompletedProcess[str]]
ContentProvider = Callable[[Path], str | None]
def _is_assignment_placeholder(value: str) -> bool:
"""Indique si une valeur d'affectation est un placeholder documentaire.
:param value: Valeur extraite d'une affectation sensible.
:return: ``True`` si la valeur ne représente pas un secret réel.
:rtype: bool
"""
return _ASSIGNMENT_PLACEHOLDER_RE.fullmatch(value.strip()) is not None
def _is_candidate(path: Path) -> bool:
"""Indique si un chemin peut être analysé comme fichier texte.
@@ -188,9 +205,15 @@ def find_secrets(
for number, line in enumerate(content.splitlines(), start=1):
if _ALLOWLIST_MARKER in line:
continue
is_literal_secret = _LITERAL_SECRET_RE.search(line) or (
relative_path.suffix in _UNQUOTED_CONFIG_SUFFIXES
and _UNQUOTED_SECRET_RE.search(line)
literal_match = _LITERAL_SECRET_RE.search(line)
unquoted_match = (
_UNQUOTED_SECRET_RE.search(line)
if relative_path.suffix in _UNQUOTED_CONFIG_SUFFIXES
else None
)
is_literal_secret = any(
match is not None and not _is_assignment_placeholder(match.group("value"))
for match in (literal_match, unquoted_match)
)
if is_literal_secret:
findings.append(SecretFinding(relative_path, number, "affectation-litterale"))
+73
View File
@@ -118,6 +118,27 @@ def test_main_detects_sensitive_url_parameter(
assert sentinel not in output
def test_main_detects_pin_url_parameter_without_disclosing_its_value(
secret_checker: ModuleType, tmp_path: Path, capsys: CaptureFixture[str]
) -> None:
"""Vérifie qu'un PIN dans une query string déclenche un échec sans fuite.
:param secret_checker: Module du script sous test.
:param tmp_path: Répertoire temporaire représentant un dépôt.
:param capsys: Fixture de capture de sortie.
:return: None
"""
sentinel = "m14-url-pin-sentinel"
(tmp_path / "settings.yaml").write_text(
f"url: https://example.invalid/api?pin={sentinel}\n", encoding="utf-8"
) # secret-check: allow
assert secret_checker.main([], root=tmp_path) == 1
output = capsys.readouterr().out
assert "settings.yaml:1 (parametre-url)" in output
assert sentinel not in output
def test_main_ignores_documentation_url_placeholders(
secret_checker: ModuleType, tmp_path: Path, capsys: CaptureFixture[str]
) -> None:
@@ -236,6 +257,58 @@ def test_main_detects_prefixed_secret_assignment(
assert sentinel not in output
def test_main_detects_pronote_pin_assignments_without_disclosing_value(
secret_checker: ModuleType, tmp_path: Path, capsys: CaptureFixture[str]
) -> None:
"""Détecte les PIN Pronote littéraux et non quotés sans afficher leur valeur.
:param secret_checker: Module du script sous test.
:param tmp_path: Répertoire temporaire représentant un dépôt.
:param capsys: Fixture de capture de sortie.
:return: None
"""
literal_pin = "pin-literal-sentinel"
unquoted_pin = "pin-unquoted-sentinel"
(tmp_path / "settings.py").write_text(f'PRONOTE_QR_PIN = "{literal_pin}"\n', encoding="utf-8")
(tmp_path / "settings.yaml").write_text(
f"PRONOTE_ACCOUNT_PIN: {unquoted_pin}\n", encoding="utf-8"
)
assert secret_checker.main([], root=tmp_path) == 1
output = capsys.readouterr().out
assert "settings.py:1" in output
assert "settings.yaml:1" in output
assert literal_pin not in output
assert unquoted_pin not in output
@pytest.mark.parametrize(
"line",
[
'PRONOTE_QR_PIN = "<valeur>"',
"# PRONOTE_ACCOUNT_PIN doit rester dans le fichier d'environnement local",
],
)
def test_main_ignores_pronote_pin_placeholders_and_descriptions(
secret_checker: ModuleType,
tmp_path: Path,
capsys: CaptureFixture[str],
line: str,
) -> None:
"""Ignore les placeholders et descriptions de PIN sans affectation réelle.
:param secret_checker: Module du script sous test.
:param tmp_path: Répertoire temporaire représentant un dépôt.
:param capsys: Fixture de capture de sortie.
:param line: Ligne documentaire à analyser.
:return: None
"""
(tmp_path / "guide.py").write_text(line + "\n", encoding="utf-8")
assert secret_checker.main([], root=tmp_path) == 0
assert "OK:" in capsys.readouterr().out
def test_main_detects_short_secret_assignment(
secret_checker: ModuleType, tmp_path: Path, capsys: CaptureFixture[str]
) -> None:
+3 -153
View File
@@ -13,14 +13,7 @@ import pytest
from pydantic import SecretStr, ValidationError
from pronote_sync.config.env import load_settings
from pronote_sync.config.settings import (
AppSettings,
BlogSettings,
CalDAVSettings,
ExternalEndpoint,
PronoteSettings,
Settings,
)
from pronote_sync.config.settings import AppSettings, PronoteSettings, Settings
if TYPE_CHECKING:
from _pytest.monkeypatch import MonkeyPatch
@@ -80,7 +73,7 @@ def test_secretstr_masking_ical_url() -> None:
# Vérification de la sérialisation JSON
json_str = settings.model_dump_json()
assert "SECRET_TOKEN" not in json_str
assert "REDACTED" in json_str
assert "**********" in json_str
def test_secretstr_masking_password() -> None:
@@ -135,150 +128,7 @@ def test_url_from_pronote_url_env_var(monkeypatch: MonkeyPatch) -> None:
test_url = "https://example.index-education.net/pronote/parent.html"
monkeypatch.setenv("PRONOTE_URL", test_url)
settings = load_settings()
assert settings.pronote.endpoint is not None
assert settings.pronote.endpoint.url.get_secret_value() == test_url
@pytest.mark.parametrize(
"url",
[
"https://endpoint.example.test/api",
"http://localhost:8080/test",
"file:///tmp/fixture.ics",
],
)
def test_external_endpoint_accepts_supported_schemes(url: str) -> None:
"""Vérifie le socle commun des URL d'endpoints autorisés.
:param url: URL représentative du schéma à valider.
:return: None
"""
endpoint = ExternalEndpoint(url=SecretStr(url))
assert endpoint.url.get_secret_value() == url
@pytest.mark.parametrize(
"url",
["ftp://endpoint.example.test", "https:///missing-host", "https://host:bad", "file://"],
)
def test_external_endpoint_rejects_invalid_urls_without_leak(url: str) -> None:
"""Vérifie que le socle rejette les URL invalides sans les afficher.
:param url: URL invalide à refuser.
:return: None
"""
with pytest.raises(ValidationError) as exc_info:
ExternalEndpoint(url=SecretStr(url))
assert url not in str(exc_info.value)
def test_external_endpoint_is_immutable_and_redacted() -> None:
"""Vérifie le contrat immuable et expurgé du value object.
:return: None
"""
endpoint = ExternalEndpoint(url=SecretStr("https://user:secret@example.test/calendar"))
with pytest.raises(ValidationError):
endpoint.url = SecretStr("https://other.example.test")
assert "secret" not in endpoint.model_dump_json()
assert "REDACTED" in endpoint.model_dump_json()
def test_caldav_endpoint_loads_from_nested_environment(monkeypatch: MonkeyPatch) -> None:
"""Vérifie le chargement du nouvel endpoint CalDAV depuis l'environnement.
:param monkeypatch: Fixture pytest pour modifier temporairement l'environnement.
:return: None
"""
monkeypatch.setenv("CALDAV_ENDPOINT__URL", "https://caldav.example.test/dav")
settings = load_settings()
assert settings.caldav.endpoint is not None
assert settings.caldav.endpoint.url.get_secret_value() == "https://caldav.example.test/dav"
def test_caldav_legacy_url_migrates_with_warning() -> None:
"""Vérifie la migration temporaire du réglage CalDAV historique.
:return: None
"""
with pytest.warns(DeprecationWarning, match="CALDAV_URL"):
settings = CalDAVSettings(url=SecretStr("https://caldav.example.test/dav"))
assert settings.endpoint is not None
assert settings.endpoint.url.get_secret_value() == "https://caldav.example.test/dav"
def test_pronote_endpoints_load_from_nested_environment(monkeypatch: MonkeyPatch) -> None:
"""Vérifie le chargement des deux endpoints Pronote depuis l'environnement.
:param monkeypatch: Fixture pytest pour modifier temporairement l'environnement.
:return: None
"""
monkeypatch.setenv("PRONOTE_ENDPOINT__URL", "https://pronote.example.test/parent.html")
monkeypatch.setenv("PRONOTE_ICAL_ENDPOINT__URL", "file:///tmp/pronote.ics")
settings = load_settings()
assert settings.pronote.endpoint is not None
assert settings.pronote.ical_endpoint is not None
assert (
settings.pronote.endpoint.url.get_secret_value()
== "https://pronote.example.test/parent.html"
)
assert settings.pronote.ical_endpoint.url.get_secret_value() == "file:///tmp/pronote.ics"
@pytest.mark.parametrize(
("field", "url", "message"),
[
("endpoint", "http://pronote.example.test", "HTTPS requis"),
("ical_endpoint", "http://pronote.example.test/calendar", "HTTPS ou file requis"),
],
)
def test_pronote_endpoint_policy_rejects_insecure_url(field: str, url: str, message: str) -> None:
"""Vérifie la politique de transport des endpoints Pronote.
:param field: Nom du champ endpoint à alimenter.
:param url: URL non sûre à refuser.
:param message: Fragment attendu du message sûr.
:return: None
"""
endpoint = ExternalEndpoint(url=SecretStr(url))
with pytest.raises(ValidationError, match=message) as exc_info:
if field == "endpoint":
PronoteSettings(endpoint=endpoint)
else:
PronoteSettings(ical_endpoint=endpoint)
assert url not in str(exc_info.value)
def test_blog_endpoint_loads_from_nested_environment(monkeypatch: MonkeyPatch) -> None:
"""Vérifie le chargement du nouvel endpoint RSS depuis l'environnement.
:param monkeypatch: Fixture pytest pour modifier temporairement l'environnement.
:return: None
"""
monkeypatch.setenv("BLOG_ENDPOINT__URL", "https://blog.example.test/feed")
settings = load_settings()
assert settings.blog.endpoint.url.get_secret_value() == "https://blog.example.test/feed"
def test_blog_legacy_rss_url_migrates_with_warning() -> None:
"""Vérifie la migration temporaire du réglage RSS historique.
:return: None
"""
with pytest.warns(DeprecationWarning, match="BLOG_RSS_URL"):
settings = BlogSettings(rss_url="https://blog.example.test/feed")
assert settings.endpoint.url.get_secret_value() == "https://blog.example.test/feed"
def test_blog_endpoint_rejects_insecure_url_without_leak() -> None:
"""Vérifie que l'URL RSS HTTP est refusée sans être exposée.
:return: None
"""
url = "http://user:secret@blog.example.test/feed" # pragma: allowlist secret
with pytest.raises(ValidationError) as exc_info:
BlogSettings(endpoint=ExternalEndpoint(url=SecretStr(url)))
assert url not in str(exc_info.value)
assert settings.pronote.url == test_url
def test_auth_mode_default_password() -> None:
+83
View File
@@ -0,0 +1,83 @@
"""Régressions sur le contexte des exceptions expurgées."""
from __future__ import annotations
import ast
import traceback
from datetime import date
from pathlib import Path
import pytest
from pronote_sync.errors import PipelineCriticalError
from pronote_sync.models.agenda import Lesson, SchoolEvent
from pronote_sync.models.homework import Homework
from pronote_sync.models.message import Message
from pronote_sync.pipeline.steps.fetch import fetch_step
class _FailingFetcher:
"""Fetcher minimal qui expose une erreur externe porteuse d'un secret."""
def fetch_agenda(self) -> tuple[list[Lesson], list[SchoolEvent]]:
"""Déclenche une erreur externe pendant la récupération critique."""
raise RuntimeError("password=fetch-context-secret")
def fetch_homework(self, target_date: date) -> list[Homework]:
"""Retourne une liste vide pour compléter le protocole du fetcher."""
return []
def fetch_messages(self) -> list[Message]:
"""Retourne une liste vide pour compléter le protocole du fetcher."""
return []
def fetch_informations(self) -> list[Message]:
"""Retourne une liste vide pour compléter le protocole du fetcher."""
return []
def test_fetch_step_does_not_retain_external_exception_context() -> None:
"""Vérifie qu'une erreur critique ne conserve ni secret ni contexte externe."""
with pytest.raises(PipelineCriticalError) as exc_info:
fetch_step(_FailingFetcher())
error = exc_info.value
formatted = "".join(traceback.format_exception(error))
assert "fetch-context-secret" not in str(error)
assert "fetch-context-secret" not in formatted
assert error.__cause__ is None
assert error.__context__ is None
def test_production_raise_from_none_is_never_inside_except() -> None:
"""Vérifie structurellement que les exceptions expurgées sont levées hors des handlers."""
root = Path(__file__).parents[2] / "pronote_sync"
violations: list[str] = []
class Visitor(ast.NodeVisitor):
"""Collecte les levées ``from None`` imbriquées dans un handler."""
def __init__(self) -> None:
self._inside_except = False
def visit_ExceptHandler(self, node: ast.ExceptHandler) -> None:
"""Visite un handler en marquant son périmètre lexical."""
previous = self._inside_except
self._inside_except = True
self.generic_visit(node)
self._inside_except = previous
def visit_Raise(self, node: ast.Raise) -> None:
"""Signale une levée ``from None`` dans un handler."""
if (
self._inside_except
and isinstance(node.cause, ast.Constant)
and node.cause.value is None
):
violations.append(f"{path}:{node.lineno}")
self.generic_visit(node)
for path in sorted(root.rglob("*.py")):
Visitor().visit(ast.parse(path.read_text(encoding="utf-8")))
assert violations == []
+15 -17
View File
@@ -251,7 +251,7 @@ def test_fetch_agenda_auto_both_fail(mock_fetcher: PronoteFetcher) -> None:
:rtype: None
"""
# Disable pronotepy so fallback is None
mock_fetcher._settings.pronote.endpoint = None
mock_fetcher._settings.pronote.url = None
with (
patch("pronote_sync.sources.pronote.fallback.fetch_ical") as m_fetch_ical,
@@ -280,7 +280,7 @@ def test_fetch_agenda_ical_mode_failure(mock_fetcher: PronoteFetcher) -> None:
"""
# Override settings to use ical mode explicitly and disable fallback
mock_fetcher._settings.pronote.agenda_source = "ical"
mock_fetcher._settings.pronote.endpoint = None
mock_fetcher._settings.pronote.url = None
with (
patch("pronote_sync.sources.pronote.fallback.fetch_ical") as m_fetch_ical,
@@ -311,7 +311,7 @@ def test_fetch_agenda_pronotepy_mode_failure(mock_fetcher: PronoteFetcher) -> No
"""
# Override settings to use pronotepy mode explicitly and disable fallback
mock_fetcher._settings.pronote.agenda_source = "pronotepy"
mock_fetcher._settings.pronote.ical_endpoint = None
mock_fetcher._settings.pronote.ical_url = None
client = MagicMock()
client.get_lessons.side_effect = OSError("Pronote API error")
@@ -438,7 +438,7 @@ def test_fetch_homework_auto_both_fail(mock_fetcher: PronoteFetcher) -> None:
target_date = date(2025, 9, 10)
# Disable pronotepy so fallback is None
mock_fetcher._settings.pronote.endpoint = None
mock_fetcher._settings.pronote.url = None
with (
patch("pronote_sync.sources.pronote.fallback.fetch_ical") as m_fetch_ical,
@@ -531,7 +531,7 @@ def test_no_secrets_in_error_messages(
:rtype: None
"""
# Disable pronotepy so fallback is None to trigger PipelineCriticalError
mock_fetcher._settings.pronote.endpoint = None
mock_fetcher._settings.pronote.url = None
with (
patch("pronote_sync.sources.pronote.fallback.fetch_ical") as m_fetch_ical,
@@ -628,8 +628,8 @@ def test_fetch_agenda_no_source_configured_raises(mock_fetcher: PronoteFetcher)
:rtype: None
"""
# Disable both sources
mock_fetcher._settings.pronote.ical_endpoint = None
mock_fetcher._settings.pronote.endpoint = None
mock_fetcher._settings.pronote.ical_url = None
mock_fetcher._settings.pronote.url = None
with pytest.raises(PipelineCriticalError) as exc_info:
mock_fetcher.fetch_agenda()
@@ -912,10 +912,8 @@ def test_fetch_agenda_ical_url_none_raises_value_error(mock_fetcher: PronoteFetc
:return: None
:rtype: None
"""
mock_fetcher._settings.pronote.ical_endpoint = None
with pytest.raises(
ValueError, match="PRONOTE_ICAL_ENDPOINT__URL est requis pour la source iCal"
):
mock_fetcher._settings.pronote.ical_url = None
with pytest.raises(ValueError, match="PRONOTE_ICAL_URL est requis pour la source iCal"):
mock_fetcher._fetch_agenda_ical()
@@ -930,7 +928,7 @@ def test_agenda_sources_auto_only_pronotepy_configured(mock_fetcher: PronoteFetc
:rtype: None
"""
mock_fetcher._settings.pronote.agenda_source = "auto"
mock_fetcher._settings.pronote.ical_endpoint = None
mock_fetcher._settings.pronote.ical_url = None
primary, fallback = mock_fetcher._agenda_sources()
assert primary == "pronotepy"
assert fallback is None
@@ -1022,7 +1020,7 @@ def test_homework_sources_explicit_ical_mode_strict(mock_fetcher: PronoteFetcher
assert fallback is None
# Without pronotepy configured
mock_fetcher._settings.pronote.endpoint = None
mock_fetcher._settings.pronote.url = None
primary, fallback = mock_fetcher._homework_sources()
assert primary == "ical"
assert fallback is None
@@ -1048,7 +1046,7 @@ def test_homework_sources_explicit_pronotepy_mode_strict(
assert fallback is None
# Without ical configured
mock_fetcher._settings.pronote.ical_endpoint = None
mock_fetcher._settings.pronote.ical_url = None
primary, fallback = mock_fetcher._homework_sources()
assert primary == "pronotepy"
assert fallback is None
@@ -1065,7 +1063,7 @@ def test_homework_sources_auto_only_pronotepy_configured(mock_fetcher: PronoteFe
:rtype: None
"""
mock_fetcher._settings.pronote.homework_source = "auto"
mock_fetcher._settings.pronote.ical_endpoint = None
mock_fetcher._settings.pronote.ical_url = None
primary, fallback = mock_fetcher._homework_sources()
assert primary == "pronotepy"
assert fallback is None
@@ -1081,8 +1079,8 @@ def test_homework_sources_auto_no_source_configured_raises(mock_fetcher: Pronote
:rtype: None
"""
mock_fetcher._settings.pronote.homework_source = "auto"
mock_fetcher._settings.pronote.ical_endpoint = None
mock_fetcher._settings.pronote.endpoint = None
mock_fetcher._settings.pronote.ical_url = None
mock_fetcher._settings.pronote.url = None
with pytest.raises(PipelineCriticalError) as exc_info:
mock_fetcher._homework_sources()