Compare commits

..
Author SHA1 Message Date
Codex 328eaeec88 docs(packaging): aligner les métadonnées du projet 2026-09-13 00:23:32 +02:00
26 changed files with 301 additions and 1153 deletions
+4 -11
View File
@@ -1,8 +1,6 @@
# --- Pronote ---
# Endpoints externes : HTTPS uniquement (pas de fichier local file://).
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 (alias obsolètes encore supportés avec un DeprecationWarning) : 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
@@ -30,10 +28,7 @@ PRONOTE_AUTH_MODE=password
# PRONOTE_ACCOUNT_PIN=
# --- CalDAV ---
# Endpoint commun (URL potentiellement sensible, masquée dans les journaux)
# HTTPS uniquement ; HTTP n'est toléré que pour localhost avec CALDAV_ALLOW_INSECURE_HTTP=true.
CALDAV_ENDPOINT__URL=https://caldav.example.com/calendars/user/pronote/
# Ancien nom (alias obsolète encore supporté avec un DeprecationWarning) : CALDAV_URL
CALDAV_URL=https://caldav.example.com/calendars/user/pronote/
CALDAV_USERNAME=user@example.com
CALDAV_PASSWORD=your_caldav_password
CALDAV_CALENDAR_PATH=/pronote-sync/
@@ -96,10 +91,8 @@ AI_BASE_URL=https://api.openai.com/v1
# AI_ALLOW_INSECURE_HTTP=true
# --- Blog ---
# Endpoint externe : HTTPS uniquement (pas de fichier local file://).
BLOG_ENABLED=false
BLOG_ENDPOINT__URL=https://blogpeda.ac-bordeaux.fr/cjeliote/?feed=rss2
# Ancien nom (alias obsolète encore supporté avec un DeprecationWarning) : BLOG_RSS_URL
BLOG_RSS_URL=https://blogpeda.ac-bordeaux.fr/cjeliote/?feed=rss2
# --- Divers ---
DRY_RUN=false
+3 -3
View File
@@ -140,7 +140,7 @@
"filename": "GUIDE_DEV_PYTHON.md",
"hashed_secret": "90bd1b48e958257948487b90bee080ba5ed00caa",
"is_verified": false,
"line_number": 5205
"line_number": 5186
}
],
"tests/unit/test_caldav_gateway.py": [
@@ -172,7 +172,7 @@
"filename": "tests/unit/test_caldav_security.py",
"hashed_secret": "6b554cd7b7e0115065fb4907307a74f1902154d4",
"is_verified": false,
"line_number": 29
"line_number": 28
}
],
"tests/unit/test_xmpp_channel.py": [
@@ -185,5 +185,5 @@
}
]
},
"generated_at": "2026-09-13T13:46:01Z"
"generated_at": "2026-09-12T22:12:56Z"
}
+47 -78
View File
@@ -259,12 +259,12 @@ 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` |
| `CALDAV_ENDPOINT__URL` | URL du serveur CalDAV, via l'endpoint commun (masquée en `SecretStr`). | `https://caldav.example.com/calendars/...` | `ExternalEndpoint` |
| `CALDAV_URL` | URL du serveur CalDAV (masquée en `SecretStr`). | `https://caldav.example.com/calendars/...` | `SecretStr` |
| `CALDAV_USERNAME` | Identifiant CalDAV. | `user@example.com` | `str` |
| `CALDAV_PASSWORD` | Mot de passe CalDAV. | `SecretStr` (masqué) | `SecretStr` |
| `CALDAV_CALENDAR_PATH` | Chemin du calendrier CalDAV de destination. | `/pronote-sync/` | `str` |
@@ -277,28 +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`.
>
> **Contrat des endpoints externes** : le schéma `file://` n'est **pas** supporté en production.
> Seul HTTPS est accepté (plus HTTP uniquement pour un hôte loopback CalDAV lorsque
> `CALDAV_ALLOW_INSECURE_HTTP=true`). Les credentials embarqués (`user:pass@hôte`) sont refusés
> par les endpoints canoniques `*_ENDPOINT__URL`. Les fixtures locales ne doivent pas passer par
> ces variables.
> 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é.
@@ -334,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
@@ -346,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
@@ -400,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"`.
@@ -414,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
@@ -1124,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
@@ -1229,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
@@ -1241,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`** :
@@ -1261,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
```
---
@@ -1320,12 +1312,8 @@ from pronote_sync.models.blog import BlogArticle
@pytest.fixture
def mock_blog_rss_client():
"""Retourne un client RSS mocké pour les tests.
``requests.get`` est mocké : aucune requête réseau réelle n'est émise.
Le schéma ``file://`` n'est pas supporté par ``BlogRSSClient``.
"""
client = BlogRSSClient(rss_url="https://example.com/blog/feed")
"""Retourne un client RSS mocké pour les tests."""
client = BlogRSSClient(rss_url="file://tests/fixtures/blog_rss.xml")
return client
@@ -1372,7 +1360,7 @@ def test_blog_deduplication(tmp_path):
assert "https://blogpeda.ac-bordeaux.fr/cjeliote/?p=1625" in state.get_known_guids()
# Simuler une nouvelle récupération : seul le nouvel article doit être retourné
client = BlogRSSClient(rss_url="https://example.com/blog/feed") # requête mockée
client = BlogRSSClient(rss_url="file://tests/fixtures/blog_rss.xml")
result = client.fetch_and_parse(known_guids=state.get_known_guids())
# Seul l'article avec p=1626 doit être retourné (car p=1625 est déjà connu)
@@ -1632,14 +1620,11 @@ from ..models.agenda import RawCalendarData
def fetch_ical(url: str, timeout: int = 20) -> str:
"""
Récupère le flux iCal depuis une URL Pronote (HTTPS uniquement).
Récupère le flux iCal depuis une URL Pronote.
Inspiré de src/sources/pronote/fetch.ts.
Le schéma ``file://`` n'est **pas** supporté : les fixtures locales sont
injectées via un mock HTTP (``responses``) dans les tests.
Args:
url: URL HTTPS du flux iCal (avec token ``icalsecurise``).
url: URL du flux iCal (peut être file:// pour les tests).
timeout: Timeout en secondes (défaut: 20s).
Returns:
@@ -1654,7 +1639,16 @@ def fetch_ical(url: str, timeout: int = 20) -> str:
"user-agent": "pronote-sync",
}
# Récupération HTTP (aucun accès fichier local)
# Gestion des URLs file:// pour les tests
if url.startswith("file://"):
import pathlib
file_path = pathlib.Path(url.replace("file://", ""))
content = file_path.read_text(encoding="utf-8")
if "BEGIN:VCALENDAR" not in content:
raise ValueError(f"Fichier iCal invalide: {redact_url(url)}")
return content
# Récupération HTTP
try:
response = requests.get(
url,
@@ -1680,19 +1674,6 @@ def fetch_ical(url: str, timeout: int = 20) -> str:
return content
# Test unitaire réseau mocké (aucun accès fichier)
@responses.activate
def test_fetch_ical_https() -> None:
responses.add(
responses.GET,
"https://pronote.example.test/ical.ics",
body="BEGIN:VCALENDAR\nVERSION:2.0\nEND:VCALENDAR",
status=200,
)
content = fetch_ical("https://pronote.example.test/ical.ics")
assert content.startswith("BEGIN:VCALENDAR")
def get_calendar_name(raw_ical: str) -> str | None:
"""
Extrait le nom du calendrier depuis X-WR-CALNAME.
@@ -2724,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 :
@@ -4629,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
@@ -5360,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",
@@ -5455,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,
@@ -5650,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 |
@@ -5745,17 +5724,7 @@ def check_secrets_in_code():
(["grep", "-r", "icalsecurise=", "src/", "tests/", "--include=*.py"], "Tokens iCal dans le code"),
(["grep", "-r", "password\s*=", "src/", "tests/", "--include=*.py"], "Mots de passe en clair"),
(["grep", "-r", "api_key\s*=", "src/", "tests/", "--include=*.py"], "Clés API en clair"),
(
[
"grep",
"-r",
"PRONOTE_ICAL_ENDPOINT__URL.*=",
"src/",
"tests/",
"--include=*.py",
],
"URLs iCal en clair",
),
(["grep", "-r", "PRONOTE_ICAL_URL.*=", "src/", "tests/", "--include=*.py"], "URLs iCal en clair"),
]
all_ok = True
@@ -5979,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`. |
@@ -5991,7 +5960,7 @@ Exemple de ligne cron (exécution tous les jours à 18h) :
| **Commande** | **Description** |
|---------------------------------------|-----------------------------------------------------------------------------------------------------|
| `python -m pronote_sync.cli.main --dry-run --log-level DEBUG` | Exécute le pipeline en mode dry-run avec des logs détaillés. |
| `python -c "from pronote_sync.sources.pronote.ical import fetch_ical; print(fetch_ical('https://college.ent/pronote/ical/Edt.ics?icalsecurise=TOKEN'))"` | Teste la récupération d'un flux iCal HTTPS (remplacer `TOKEN`). |
| `python -c "from pronote_sync.sources.pronote.ical import fetch_ical; print(fetch_ical('file://tests/fixtures/pronote-4e.ics'))"` | Teste le parsing d'un fichier iCal local. |
| `python -c "from pronote_sync.config.settings import settings; print(settings)"` | Affiche la configuration chargée. |
| `python -c "import caldav; print(caldav.__version__)"` | Vérifie la version de la bibliothèque CalDAV. |
| `python -c "import slixmpp; print(slixmpp.__version__)"` | Vérifie la version de la bibliothèque XMPP. |
+3 -6
View File
@@ -58,7 +58,6 @@ The following variables can be safely pre-configured in `/etc/pronote-sync/prono
- `PRONOTE_ACCOUNT_TYPE` (default: `parent`)
- `PRONOTE_ENT` (ENT slug, e.g., `lyceeconnecte`)
- `PRONOTE_AGENDA_SOURCE`, `PRONOTE_HOMEWORK_SOURCE`, `PRONOTE_MESSAGES_SOURCE` (`auto`, `ical`, or `pronotepy`)
- `PRONOTE_ENDPOINT__URL` (API Pronote endpoint) and `PRONOTE_ICAL_ENDPOINT__URL` (iCal feed endpoint). The old names `PRONOTE_URL` and `PRONOTE_ICAL_URL` are deprecated aliases that still work with a `DeprecationWarning`.
- **CalDAV:**
- `CALDAV_CALENDAR_PATH` (e.g., `/pronote-sync/`)
@@ -79,7 +78,7 @@ The following variables can be safely pre-configured in `/etc/pronote-sync/prono
- `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_ENABLED`, `BLOG_ENDPOINT__URL`. The old name `BLOG_RSS_URL` is a deprecated alias that still works with a `DeprecationWarning`.
- `BLOG_ENABLED`, `BLOG_RSS_URL`
- **General:**
- `DRY_RUN`, `LOG_LEVEL`
@@ -89,12 +88,10 @@ The following variables can be safely pre-configured in `/etc/pronote-sync/prono
**Do NOT set these variables automatically.** The operator must manually provide the following secrets:
- **Pronote:**
- `PRONOTE_ICAL_ENDPOINT__URL`, `PRONOTE_ENDPOINT__URL`, `PRONOTE_USERNAME`, `PRONOTE_PASSWORD`
- The old names `PRONOTE_ICAL_URL` and `PRONOTE_URL` are deprecated aliases that still work with a `DeprecationWarning`.
- `PRONOTE_ICAL_URL`, `PRONOTE_URL`, `PRONOTE_USERNAME`, `PRONOTE_PASSWORD`
- **CalDAV:**
- `CALDAV_ENDPOINT__URL`, `CALDAV_USERNAME`, `CALDAV_PASSWORD`
- The old name `CALDAV_URL` is a deprecated alias that still works with a `DeprecationWarning`.
- `CALDAV_URL`, `CALDAV_USERNAME`, `CALDAV_PASSWORD`
- **XMPP:**
- `XMPP_JID`, `XMPP_PASSWORD`, `XMPP_TO`
+1 -1
View File
@@ -13,7 +13,7 @@ Synchronise l'agenda et les devoirs de **Pronote** vers un calendrier **CalDAV**
```bash
# Cloner le dépôt
git clone https://git.antoineve.me/AntoineVe/college-infos
cd college-infos
cd pronote-sync
# Créer l'environnement virtuel
python3.13 -m venv .venv
+4 -8
View File
@@ -94,9 +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).
L'ancien nom `PRONOTE_ICAL_URL` reste accepté comme **alias obsolète** (émission d'un
`DeprecationWarning`). Seul HTTPS est accepté ; `file://` est refusé.
- **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`).
@@ -128,9 +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). L'ancien nom `PRONOTE_URL` reste
accepté comme **alias obsolète** (émission d'un `DeprecationWarning`). Seul HTTPS est
accepté ; `file://` est refusé.
- `PRONOTE_URL` (ex. `.env.example` ligne 3).
- `PRONOTE_USERNAME`, `PRONOTE_PASSWORD`.
- `PRONOTE_ENT` (slug dans `_ENT_NAMES`).
- `PRONOTE_ACCOUNT_TYPE` (ex. `parent`).
@@ -292,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** |
+52 -249
View File
@@ -10,84 +10,21 @@ from __future__ import annotations
import warnings
from datetime import date
from typing import Annotated, Literal
from typing import Literal
from urllib.parse import urlparse
from pydantic import (
AfterValidator,
BaseModel,
ConfigDict,
Field,
SecretStr,
ValidationInfo,
field_serializer,
field_validator,
model_validator,
)
from pydantic_settings import BaseSettings, SettingsConfigDict
from pronote_sync.utils.redaction import redact_url
_EXTERNAL_ENDPOINT_SCHEMES: frozenset[str] = frozenset({"http", "https"})
_LOOPBACK_HOSTS: frozenset[str] = frozenset({"localhost", "127.0.0.1", "::1"})
def _validate_external_endpoint_url(value: SecretStr) -> SecretStr:
"""Valide la structure et le schéma réseau d'une URL d'endpoint externe.
Le socle commun accepte uniquement les schémas ``http`` et ``https``, avec
un hôte obligatoire. Les credentials embarqués (``user:pass@host``) sont
refusés afin qu'aucun secret ne soit transporté dans l'URL. La restriction
``https``/HTTP loopback est ensuite affinée par chaque connecteur.
:param value: URL potentiellement sensible à valider.
:return: URL validée, toujours encapsulée dans ``SecretStr``.
:rtype: SecretStr
:raises ValueError: Si l'URL est malformée, sans hôte, utilise un schéma
non réseau ou contient des credentials.
"""
is_valid = False
try:
parsed = urlparse(value.get_secret_value())
_ = parsed.port
is_valid = (
parsed.scheme in _EXTERNAL_ENDPOINT_SCHEMES
and parsed.hostname is not None
and parsed.username is None
and parsed.password is None
)
except ValueError:
pass
if not is_valid:
raise ValueError("Endpoint externe invalide : URL ou schéma non supporté") from None
return value
EndpointUrl = Annotated[SecretStr, AfterValidator(_validate_external_endpoint_url)]
class ExternalEndpoint(BaseModel):
"""Représente un endpoint externe potentiellement sensible.
Le socle accepte uniquement les transports réseau ``https`` et ``http``
(hôte obligatoire, sans credentials embarqués). Chaque connecteur restreint
ensuite cette liste selon sa propre politique de sécurité. L'URL reste
encapsulée dans :class:`pydantic.SecretStr` et sa sérialisation conserve
uniquement une représentation expurgée.
"""
model_config = ConfigDict(extra="forbid", frozen=True, hide_input_in_errors=True)
url: EndpointUrl
@field_serializer("url")
def _serialize_url(self, value: SecretStr) -> str:
"""Expurge l'URL lors de la sérialisation.
:param value: URL encapsulée à sérialiser.
:return: URL expurgée.
:rtype: str
"""
return redact_url(value.get_secret_value())
class PronoteSettings(BaseSettings):
"""Paramètres d'accès à Pronote (flux iCal et API ``pronotepy``).
@@ -96,29 +33,13 @@ class PronoteSettings(BaseSettings):
``PRONOTE_``.
"""
model_config = SettingsConfigDict(
env_file=".env",
env_nested_delimiter="__",
extra="ignore",
env_prefix="PRONOTE_",
hide_input_in_errors=True,
)
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"
@@ -128,58 +49,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 et le flux iCal exigent tous deux HTTPS : aucun fichier
local n'est accepté.
:return: Instance validée inchangée.
:rtype: PronoteSettings
:raises ValueError: Si un endpoint n'utilise pas HTTPS.
"""
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 != "https"
):
raise ValueError("URL iCal Pronote invalide : HTTPS 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:
@@ -211,96 +91,73 @@ class CalDAVSettings(BaseSettings):
Les variables d'environnement correspondantes sont préfixées par
``CALDAV_``. L'URL est traitée comme potentiellement sensible (au même
titre que ``PRONOTE_ICAL_ENDPOINT__URL``) : elle est de type ``SecretStr`` et
titre que ``PRONOTE_ICAL_URL``) : elle est de type ``SecretStr`` et
masquée lors de la sérialisation. Par défaut, seul HTTPS est accepté ;
HTTP n'est toléré que pour un hôte de boucle locale (``localhost``,
``127.0.0.1``, ``::1``) lorsque ``allow_insecure_http`` vaut ``True``.
Le nouvel endpoint se configure avec ``CALDAV_ENDPOINT__URL`` ;
``CALDAV_URL`` reste temporairement pris en charge avec un avertissement
de dépréciation.
"""
model_config = SettingsConfigDict(
env_file=".env",
env_nested_delimiter="__",
extra="ignore",
env_prefix="CALDAV_",
hide_input_in_errors=True,
)
model_config = SettingsConfigDict(env_file=".env", extra="ignore", env_prefix="CALDAV_")
allow_insecure_http: bool = False
endpoint: ExternalEndpoint | None = None
url: SecretStr | None = Field(
default=None,
exclude=True,
deprecated="Utiliser endpoint.url à la place (CALDAV_URL obsolète).",
)
url: SecretStr | None = None
username: str | None = None
password: SecretStr | None = None
calendar_path: str = "/pronote-sync/"
@model_validator(mode="before")
@classmethod
def _migrate_legacy_url(cls, data: object) -> object:
"""Migre ``url`` vers l'endpoint commun avec un avertissement.
@field_serializer("url")
def _serialize_url(self, value: SecretStr | None) -> str | None:
"""Masque l'URL CalDAV lors de la sérialisation (repr, str, JSON).
L'alias historique ``CALDAV_URL`` est migré via le chemin de validation
canonique : il est donc soumis exactement aux mêmes règles que
``CALDAV_ENDPOINT__URL`` (schémas réseau uniquement, hôte obligatoire,
port valide, credentials embarqués refusés). La politique de transport
(HTTPS, ou HTTP loopback uniquement avec ``allow_insecure_http``) reste
appliquée ensuite.
:param data: Données brutes du modèle.
:return: Données complétées avec ``endpoint`` si nécessaire.
:rtype: object
:param value: Valeur du champ ``url`` (secret potentiel).
:return: URL avec les éléments sensibles remplacés par ``REDACTED``,
ou ``None`` si la valeur est absente.
:rtype: str | None
"""
if not isinstance(data, dict) or data.get("url") is None:
return data
migrated_data = data.copy()
warnings.warn(
"CALDAV_URL est obsolète : utiliser CALDAV_ENDPOINT__URL.",
DeprecationWarning,
stacklevel=2,
)
if migrated_data.get("endpoint") is None:
migrated_data["endpoint"] = {"url": migrated_data["url"]}
return migrated_data
if value is None:
return None
return redact_url(value.get_secret_value())
@model_validator(mode="after")
def _validate_endpoint_policy(self) -> CalDAVSettings:
"""Applique la politique HTTPS/HTTP loopback propre à CalDAV.
@field_validator("url")
@classmethod
def _validate_url_https(cls, v: SecretStr | None, info: ValidationInfo) -> SecretStr | None:
"""Valide le schéma de l'URL CalDAV (HTTPS obligatoire par défaut).
HTTPS est toujours accepté. HTTP n'est accepté que pour un hôte de
boucle locale (``localhost``, ``127.0.0.1``, ``::1``) et uniquement
lorsque ``allow_insecure_http`` vaut ``True``. Les autres schémas du
socle commun sont refusés pour ce connecteur.
lorsque ``allow_insecure_http`` vaut ``True``. Les messages d'erreur
ne contiennent jamais l'URL brute (susceptible de contenir des
identifiants).
:return: Instance validée inchangée.
:rtype: CalDAVSettings
:param v: Valeur du champ ``url`` à valider.
:param info: Contexte de validation (accès aux autres champs).
:return: La valeur validée inchangée.
:rtype: SecretStr | None
:raises ValueError: Si le schéma n'est pas supporté ou si l'URL HTTP
n'est pas autorisée.
"""
if self.endpoint is None:
return self
raw_url = self.endpoint.url.get_secret_value()
if v is None:
return v
raw_url = v.get_secret_value()
parsed = urlparse(raw_url)
if parsed.scheme not in ("http", "https"):
raise ValueError("URL CalDAV invalide : schéma non supporté") from None
if parsed.scheme == "https":
return self
return v
# HTTP — check allow_insecure_http flag and loopback
if not self.allow_insecure_http:
allow_insecure = info.data.get("allow_insecure_http", False)
if not allow_insecure:
raise ValueError(
"URL CalDAV non sécurisée : HTTPS requis (ou activer "
"CALDAV_ALLOW_INSECURE_HTTP pour localhost)"
) from None
hostname = parsed.hostname or ""
if hostname not in _LOOPBACK_HOSTS:
loopback_hosts = {"localhost", "127.0.0.1", "::1"}
if hostname not in loopback_hosts:
raise ValueError(
"URL CalDAV non sécurisée : HTTP autorisé uniquement pour localhost"
) from None
return self
return v
_XMPP_LOOPBACK_HOSTS: frozenset[str] = frozenset({"localhost", "127.0.0.1", "::1"})
@@ -425,62 +282,10 @@ class BlogSettings(BaseSettings):
``BLOG_``.
"""
model_config = SettingsConfigDict(
env_file=".env",
env_nested_delimiter="__",
extra="ignore",
env_prefix="BLOG_",
hide_input_in_errors=True,
)
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.
Seul HTTPS est accepté : aucun fichier local n'est lu depuis un
endpoint externe.
:return: Instance validée inchangée.
:rtype: BlogSettings
:raises ValueError: Si le schéma n'est pas ``https``.
"""
scheme = urlparse(self.endpoint.url.get_secret_value()).scheme
if scheme != "https":
raise ValueError("URL RSS invalide : HTTPS requis") from None
return self
rss_url: str = "https://blogpeda.ac-bordeaux.fr/cjeliote/?feed=rss2"
class AppSettings(BaseSettings):
@@ -533,15 +338,13 @@ 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,
self.caldav.endpoint.url if self.caldav.endpoint is not None else None,
self.caldav.url,
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
)
+2 -5
View File
@@ -121,7 +121,6 @@ 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)
@@ -131,11 +130,9 @@ def fetch_step(
except PronoteAuthRotationError:
raise
except Exception as exc:
critical_error = PipelineCriticalError(
raise PipelineCriticalError(
f"Récupération Pronote impossible : {redact_exception(exc)}", step="fetch"
)
if critical_error is not None:
raise critical_error from None
) from None
messages, warnings = _fetch_optional_messages(fetcher)
return (
+5 -8
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, redact_secrets
from pronote_sync.utils.redaction import redact_exception
def fetch_blog_step(client: BlogRSSClient | None, state: BlogRSSState | None) -> BlogRSSFetchResult:
@@ -24,18 +24,15 @@ 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:
error_message = f"Récupération du blog échouée : {redact_secrets(result.error)}"
elif not result.not_modified and not result.articles:
raise RuntimeError(result.error) from None
if not result.not_modified and not result.articles:
state.update_cache_headers(result.etag, result.last_modified)
return result
except Exception as exc:
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
raise RuntimeError(f"Récupération du blog échouée : {redact_exception(exc)}") from None
+2 -5
View File
@@ -218,7 +218,6 @@ 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():
@@ -260,12 +259,10 @@ class PronoteAuthState:
"Nettoyage du fichier temporaire d'état d'authentification Pronote échoué : %s",
redact_exception(cleanup_exc),
)
write_error = PronoteSyncError(
raise PronoteSyncError(
f"Impossible d'écrire le fichier d'état d'authentification Pronote "
f"{redact_secrets(str(self._state_file))}."
)
if write_error is not None:
raise write_error from None
) 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,
+69 -74
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,36 +269,34 @@ 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)
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
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
def _fetch_homework_ical(self, target_date: date) -> list[Homework]:
"""Récupère les devoirs depuis le flux iCal pour la date cible.
@@ -306,7 +304,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.
"""
@@ -337,7 +335,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``)``.
@@ -398,37 +396,34 @@ 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)
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
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
def fetch_messages(self) -> list[Message]:
"""Récupère les messages des discussions Pronote (toujours via pronotepy).
+25 -9
View File
@@ -1,7 +1,7 @@
"""Récupération et parsing du flux iCal Pronote.
Ce module fournit le téléchargement réseau du flux iCal Pronote ainsi que
son parsing en modèles : cours
Ce module fournit le téléchargement du flux iCal Pronote (via HTTP ou
``file://`` pour les tests) ainsi que son parsing en modèles : cours
(:class:`~pronote_sync.models.agenda.Lesson`), événements scolaires
(:class:`~pronote_sync.models.agenda.SchoolEvent`) et devoirs
(:class:`~pronote_sync.models.homework.Homework`).
@@ -15,8 +15,10 @@ from __future__ import annotations
import hashlib
import re
import unicodedata
import urllib.parse
from datetime import date, datetime
from html import unescape
from pathlib import Path
from typing import NamedTuple, TypedDict
import requests
@@ -68,21 +70,35 @@ class ParsedHomeworkBlock(NamedTuple):
def fetch_ical(url: str, timeout: int = 20) -> str:
"""Récupère le contenu brut d'un flux iCal Pronote par HTTPS.
"""Récupère le contenu brut d'un flux iCal Pronote.
Le schéma ``file://`` n'est pas supporté : les fixtures locales doivent
être injectées via un mock HTTP (``responses``) dans les tests. Valide que
le flux commence bien par ``BEGIN:VCALENDAR``. Toutes les erreurs sont
relancées avec un message dont les secrets (token ``icalsecurise``) sont
masqués.
Gère les URLs ``file://`` pour les tests locaux (le chemin est
décodé de l'échappement URI, ex. ``%20`` → espace) et valide que
le flux commence bien par ``BEGIN:VCALENDAR``. Toutes les erreurs
sont relancées avec un message dont les secrets (token
``icalsecurise``) sont masqués.
:param url: URL HTTPS du flux iCal (avec token ``icalsecurise``).
:param url: URL du flux iCal (avec token ``icalsecurise``) ou chemin ``file://``.
:param timeout: Timeout HTTP en secondes (défaut : 20).
:return: Contenu brut du flux iCal.
:rtype: str
:raises OSError: Si le fichier local ``file://`` est illisible.
:raises requests.RequestException: Si la récupération HTTP échoue.
:raises ValueError: Si le flux ne commence pas par ``BEGIN:VCALENDAR``.
"""
if url.startswith("file://"):
parsed_url = urllib.parse.urlparse(url)
path = Path(urllib.parse.unquote(parsed_url.path))
try:
content = path.read_text(encoding="utf-8")
except OSError as exc:
raise OSError(
f"Impossible de lire le fichier iCal {redact_url(url)} : {redact_exception(exc)}"
) from exc
if not content.lstrip().startswith("BEGIN:VCALENDAR"):
raise ValueError(f"Fichier iCal invalide (pas de BEGIN:VCALENDAR) : {redact_url(url)}")
return content
try:
response = requests.get(url, headers=_HEADERS, timeout=timeout)
response.raise_for_status()
+2 -5
View File
@@ -85,7 +85,6 @@ 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)
@@ -95,11 +94,9 @@ class JsonTheoreticalAgendaProvider:
redact_secrets(str(file_path)),
redact_exception(exc),
)
load_error = PronoteSyncError(
raise PronoteSyncError(
f"Le fichier d'agenda théorique est invalide : {redact_secrets(str(file_path))}"
)
if load_error is not None:
raise load_error from None
) 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
+2 -5
View File
@@ -77,7 +77,6 @@ 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)
@@ -87,11 +86,9 @@ class SchoolHolidayCalendar:
redact_secrets(str(path)),
redact_exception(exc),
)
load_error = PronoteSyncError(
raise PronoteSyncError(
f"Le fichier de vacances scolaires est invalide : {redact_secrets(str(path))}"
)
if load_error is not None:
raise load_error from None
) from None
self._periods = file_model.periods
def is_holiday(self, target_date: date) -> bool:
+2 -3
View File
@@ -79,12 +79,11 @@ class CalDAVGateway:
else cast(Callable[..., Any], caldav.DAVClient)
)
self._calendar_path: str = settings.calendar_path
url_secret = settings.endpoint.url if settings.endpoint is not None else None
self._redacted_url: str | None = (
redact_url(url_secret.get_secret_value()) if url_secret else None
redact_url(settings.url.get_secret_value()) if settings.url else None
)
self._username: str | None = settings.username
self._url_secret: SecretStr | None = url_secret
self._url_secret: SecretStr | None = settings.url
self._password_secret: SecretStr | None = settings.password
self._client: Any = None
self._calendar: Any = None
+1 -1
View File
@@ -96,7 +96,7 @@ def synchronize(
)
if (
settings.caldav.endpoint is None
settings.caldav.url is None
or settings.caldav.username is None
or settings.caldav.password is None
):
+8 -31
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|pin|secret|token)"
r"\s*[:=]\s*['\"](?P<value>[^'\"\r\n]{3,})['\"]"
r"(?ix)\b[a-z0-9_]*(?:api[_-]?key|access[_-]?token|auth(?:orization)?|icalsecurise|password|secret|token)"
r"\s*[:=]\s*['\"][^'\"\r\n]{3,}['\"]"
)
_UNQUOTED_SECRET_RE = re.compile(
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,})"
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,}"
)
_URL_SECRET_RE = re.compile(
r"(?ix)[?&](?:api[_-]?key|access[_-]?token|auth(?:orization)?|icalsecurise|password|pin|secret|token)"
r"(?ix)[?&](?:api[_-]?key|access[_-]?token|auth(?:orization)?|icalsecurise|password|secret|token)"
r"=([^&#\s]{3,})"
)
_URL_PLACEHOLDER_RE = re.compile(
@@ -47,13 +47,6 @@ _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)
@@ -75,16 +68,6 @@ 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.
@@ -205,15 +188,9 @@ def find_secrets(
for number, line in enumerate(content.splitlines(), start=1):
if _ALLOWLIST_MARKER in line:
continue
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)
is_literal_secret = _LITERAL_SECRET_RE.search(line) or (
relative_path.suffix in _UNQUOTED_CONFIG_SUFFIXES
and _UNQUOTED_SECRET_RE.search(line)
)
if is_literal_secret:
findings.append(SecretFinding(relative_path, number, "affectation-litterale"))
+4 -5
View File
@@ -55,15 +55,14 @@ def real_parsed_feed(blog_rss_fixture_path: Path) -> feedparser.FeedParserDict:
@pytest.fixture
def blog_client() -> BlogRSSClient:
"""Instance de BlogRSSClient pointant vers une URL HTTPS non-réseau.
Les tests mockent ``requests.get`` : aucune requête réelle n'est émise.
def blog_client(blog_rss_fixture_path: Path) -> BlogRSSClient:
"""Instance de BlogRSSClient pointant vers le fixture local.
:param blog_rss_fixture_path: Chemin vers le fichier fixture.
:return: Instance de BlogRSSClient.
:rtype: BlogRSSClient
"""
return BlogRSSClient(rss_url="https://example.com/blog/feed")
return BlogRSSClient(rss_url=f"file://{blog_rss_fixture_path}")
# --- Helper functions for mocking ---
+15 -25
View File
@@ -15,7 +15,7 @@ from unittest.mock import MagicMock
import pytest
from pydantic import SecretStr, ValidationError
from pronote_sync.config.settings import CalDAVSettings, ExternalEndpoint
from pronote_sync.config.settings import CalDAVSettings
from pronote_sync.errors import PronoteSyncError
from pronote_sync.sync.caldav import CalDAVGateway
@@ -25,7 +25,6 @@ if TYPE_CHECKING:
# Sentinelles pour détecter les fuites de secrets dans les tests
SENTINEL_URL = "https://user:pass-super-secret-12345@caldav.example.com/secret-path/"
SENTINEL_ENDPOINT_URL = "https://caldav.example.com/secret-path/?token=pass-super-secret-12345"
SENTINEL_PASSWORD = "super-secret-password-67890"
SENTINEL_HTTP_URL = "http://caldav.example.com/"
SENTINEL_HTTP_LOCALHOST = "http://localhost:5232/caldav/"
@@ -38,13 +37,13 @@ class TestCalDAVSettingsSecurity:
def test_url_redaction_in_repr(self) -> None:
"""Vérifie que l'URL brute n'apparaît pas dans repr(settings)."""
settings = CalDAVSettings(
endpoint=ExternalEndpoint(url=SecretStr(SENTINEL_ENDPOINT_URL)),
url=SecretStr(SENTINEL_URL),
username="test-user",
password=SecretStr(SENTINEL_PASSWORD),
calendar_path="/cal/",
)
repr_str = repr(settings)
assert SENTINEL_ENDPOINT_URL not in repr_str
assert SENTINEL_URL not in repr_str
assert "pass-super-secret-12345" not in repr_str
# L'URL est masquée par redact_url qui retourne REDACTED_URL ou une URL avec REDACTED
assert "REDACTED" in repr_str or "**********" in repr_str
@@ -52,13 +51,13 @@ class TestCalDAVSettingsSecurity:
def test_url_redaction_in_str(self) -> None:
"""Vérifie que l'URL brute n'apparaît pas dans str(settings)."""
settings = CalDAVSettings(
endpoint=ExternalEndpoint(url=SecretStr(SENTINEL_ENDPOINT_URL)),
url=SecretStr(SENTINEL_URL),
username="test-user",
password=SecretStr(SENTINEL_PASSWORD),
calendar_path="/cal/",
)
str_str = str(settings)
assert SENTINEL_ENDPOINT_URL not in str_str
assert SENTINEL_URL not in str_str
assert "pass-super-secret-12345" not in str_str
# L'URL est masquée par redact_url
assert "REDACTED" in str_str or "**********" in str_str
@@ -66,7 +65,7 @@ class TestCalDAVSettingsSecurity:
def test_url_redaction_in_model_dump(self) -> None:
"""Vérifie que l'URL brute n'apparaît pas dans model_dump()."""
settings = CalDAVSettings(
endpoint=ExternalEndpoint(url=SecretStr(SENTINEL_ENDPOINT_URL)),
url=SecretStr(SENTINEL_URL),
username="test-user",
password=SecretStr(SENTINEL_PASSWORD),
calendar_path="/cal/",
@@ -75,7 +74,7 @@ class TestCalDAVSettingsSecurity:
# Vérifie que l'URL n'est pas dans les valeurs du dict
for value in dumped.values():
if isinstance(value, str):
assert SENTINEL_ENDPOINT_URL not in value
assert SENTINEL_URL not in value
assert "pass-super-secret-12345" not in value
# Vérifie que la version rédigée est présente
assert "REDACTED" in str(dumped)
@@ -202,7 +201,7 @@ class TestCalDAVGatewaySecurity:
raise Exception(f"Connection failed to {SENTINEL_URL}?token={SENTINEL_PASSWORD}")
settings = CalDAVSettings(
endpoint=ExternalEndpoint(url=SecretStr(SENTINEL_ENDPOINT_URL)),
url=SecretStr(SENTINEL_URL),
username="test-user",
password=SecretStr(SENTINEL_PASSWORD),
calendar_path="/cal/",
@@ -216,14 +215,12 @@ class TestCalDAVGatewaySecurity:
# Vérifie que le message d'erreur ne contient pas les sentinelles
error_msg = str(exc_info.value)
assert SENTINEL_URL not in error_msg
assert SENTINEL_ENDPOINT_URL not in error_msg
assert SENTINEL_PASSWORD not in error_msg
# Vérifie que les logs ne contiennent pas les sentinelles
for record in caplog.records:
log_msg = record.getMessage()
assert SENTINEL_URL not in log_msg
assert SENTINEL_ENDPOINT_URL not in log_msg
assert SENTINEL_PASSWORD not in log_msg
def test_exception_cause_and_context_is_none(self) -> None:
@@ -233,7 +230,7 @@ class TestCalDAVGatewaySecurity:
raise Exception(f"Connection failed to {SENTINEL_URL}?token={SENTINEL_PASSWORD}")
settings = CalDAVSettings(
endpoint=ExternalEndpoint(url=SecretStr(SENTINEL_ENDPOINT_URL)),
url=SecretStr(SENTINEL_URL),
username="test-user",
password=SecretStr(SENTINEL_PASSWORD),
calendar_path="/cal/",
@@ -249,7 +246,7 @@ class TestCalDAVGatewaySecurity:
def test_logs_redacted_on_list_managed_events_error(self, caplog: LogCaptureFixture) -> None:
"""Vérifie que les logs sont expurgés lors d'une erreur dans list_managed_events."""
settings = CalDAVSettings(
endpoint=ExternalEndpoint(url=SecretStr(SENTINEL_ENDPOINT_URL)),
url=SecretStr(SENTINEL_URL),
username="test-user",
password=SecretStr(SENTINEL_PASSWORD),
calendar_path="/cal/",
@@ -277,14 +274,12 @@ class TestCalDAVGatewaySecurity:
# Vérifie que le message d'erreur ne contient pas les sentinelles
error_msg = str(exc_info.value)
assert SENTINEL_URL not in error_msg
assert SENTINEL_ENDPOINT_URL not in error_msg
assert SENTINEL_PASSWORD not in error_msg
# Vérifie que les logs ne contiennent pas les sentinelles
for record in caplog.records:
log_msg = record.getMessage()
assert SENTINEL_URL not in log_msg
assert SENTINEL_ENDPOINT_URL not in log_msg
assert SENTINEL_PASSWORD not in log_msg
# Vérifie que l'exception n'est chaînée à aucune exception brute
@@ -294,7 +289,7 @@ class TestCalDAVGatewaySecurity:
def test_logs_redacted_on_upsert_event_error(self, caplog: LogCaptureFixture) -> None:
"""Vérifie que les logs sont expurgés lors d'une erreur dans upsert_event."""
settings = CalDAVSettings(
endpoint=ExternalEndpoint(url=SecretStr(SENTINEL_ENDPOINT_URL)),
url=SecretStr(SENTINEL_URL),
username="test-user",
password=SecretStr(SENTINEL_PASSWORD),
calendar_path="/cal/",
@@ -317,14 +312,12 @@ class TestCalDAVGatewaySecurity:
# Vérifie que le message d'erreur ne contient pas les sentinelles
error_msg = str(exc_info.value)
assert SENTINEL_URL not in error_msg
assert SENTINEL_ENDPOINT_URL not in error_msg
assert SENTINEL_PASSWORD not in error_msg
# Vérifie que les logs ne contiennent pas les sentinelles
for record in caplog.records:
log_msg = record.getMessage()
assert SENTINEL_URL not in log_msg
assert SENTINEL_ENDPOINT_URL not in log_msg
assert SENTINEL_PASSWORD not in log_msg
# Vérifie que l'exception n'est chaînée à aucune exception brute
@@ -334,7 +327,7 @@ class TestCalDAVGatewaySecurity:
def test_logs_redacted_on_delete_event_error(self, caplog: LogCaptureFixture) -> None:
"""Vérifie que les logs sont expurgés lors d'une erreur dans delete_event."""
settings = CalDAVSettings(
endpoint=ExternalEndpoint(url=SecretStr(SENTINEL_ENDPOINT_URL)),
url=SecretStr(SENTINEL_URL),
username="test-user",
password=SecretStr(SENTINEL_PASSWORD),
calendar_path="/cal/",
@@ -356,14 +349,12 @@ class TestCalDAVGatewaySecurity:
# Vérifie que le message d'erreur ne contient pas les sentinelles
error_msg = str(exc_info.value)
assert SENTINEL_URL not in error_msg
assert SENTINEL_ENDPOINT_URL not in error_msg
assert SENTINEL_PASSWORD not in error_msg
# Vérifie que les logs ne contiennent pas les sentinelles
for record in caplog.records:
log_msg = record.getMessage()
assert SENTINEL_URL not in log_msg
assert SENTINEL_ENDPOINT_URL not in log_msg
assert SENTINEL_PASSWORD not in log_msg
# Vérifie que l'exception n'est chaînée à aucune exception brute
@@ -373,7 +364,7 @@ class TestCalDAVGatewaySecurity:
def test_redacted_url_stored_in_gateway(self) -> None:
"""Vérifie que l'URL rédigée est stockée sur l'instance de la passerelle."""
settings = CalDAVSettings(
endpoint=ExternalEndpoint(url=SecretStr(SENTINEL_ENDPOINT_URL)),
url=SecretStr(SENTINEL_URL),
username="test-user",
password=SecretStr(SENTINEL_PASSWORD),
calendar_path="/cal/",
@@ -381,9 +372,8 @@ class TestCalDAVGatewaySecurity:
gateway = CalDAVGateway(settings)
# Vérifie que l'URL rédigée est stockée
assert gateway._redacted_url is not None
assert SENTINEL_ENDPOINT_URL not in gateway._redacted_url
assert "pass-super-secret-12345" not in gateway._redacted_url
assert SENTINEL_URL not in gateway._redacted_url
assert "REDACTED" in gateway._redacted_url
# Vérifie que l'URL brute n'est pas stockée en clair
assert gateway._url_secret is not None
assert SENTINEL_ENDPOINT_URL not in str(gateway._url_secret)
assert SENTINEL_URL not in str(gateway._url_secret)
-73
View File
@@ -118,27 +118,6 @@ 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:
@@ -257,58 +236,6 @@ 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 -412
View File
@@ -7,21 +7,13 @@ textuelles et sérialisées, et que le rechargement fonctionne comme attendu.
from __future__ import annotations
from collections.abc import Callable
from typing import TYPE_CHECKING
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
@@ -81,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:
@@ -136,183 +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",
"https://endpoint.example.test:8443/ical?icalsecurise=TOKEN", # pragma: allowlist secret
],
)
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
"""
secret = "SECRET_TOKEN_XYZ" # pragma: allowlist secret
endpoint = ExternalEndpoint(
url=SecretStr(f"https://example.test/calendar?icalsecurise={secret}")
)
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()
with pytest.raises(ValidationError):
ExternalEndpoint(url=SecretStr("https://example.test/calendar"), unknown_field="x") # type: ignore[call-arg]
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"
@pytest.mark.parametrize(
"url",
[
"https://user:pass@caldav.example.test/dav", # pragma: allowlist secret
"https:///missing-host",
"https://host:bad",
"file:///tmp/fixture.ics",
],
)
def test_caldav_legacy_url_alias_rejects_invalid_urls(url: str) -> None:
"""L'alias obsolète ``CALDAV_URL`` ne contourne pas la validation canonique.
Contrat corrigé : ``ExternalEndpoint.model_construct`` n'est plus utilisé
pour grandfathered les identifiants embarqués. L'alias est migré via le
chemin de validation normal, donc les credentials embarqués, l'hôte
manquant, le port invalide et le schéma ``file://`` sont refusés,
exactement comme pour ``CALDAV_ENDPOINT__URL``.
:param url: URL invalide fournie via l'alias historique.
:return: None
"""
with pytest.raises(ValidationError):
with pytest.warns(DeprecationWarning, match="CALDAV_URL"):
CalDAVSettings(url=SecretStr(url))
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", "https://ical.example.test/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()
== "https://ical.example.test/pronote.ics"
)
@pytest.mark.parametrize(
("field", "url", "message"),
[
("endpoint", "http://pronote.example.test", "HTTPS requis"),
("ical_endpoint", "http://pronote.example.test/calendar", "URL iCal Pronote invalide"),
],
)
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:
@@ -526,228 +342,3 @@ def test_sync_future_days_positive_env_loading(monkeypatch: MonkeyPatch) -> None
monkeypatch.setenv("SYNC_FUTURE_DAYS", "30")
settings = load_settings()
assert settings.app.sync_future_days == 30
# --- Contrat des endpoints externes (matrice partagée) ---
_ConnectorSettings = CalDAVSettings | PronoteSettings | BlogSettings
_CONNECTOR_BUILDERS: dict[str, Callable[[str], _ConnectorSettings]] = {
"caldav": lambda url: CalDAVSettings(endpoint=ExternalEndpoint(url=SecretStr(url))),
"pronote_ical": lambda url: PronoteSettings(ical_endpoint=ExternalEndpoint(url=SecretStr(url))),
"blog": lambda url: BlogSettings(endpoint=ExternalEndpoint(url=SecretStr(url))),
}
def test_external_endpoint_matrix_accepts_https() -> None:
"""HTTPS est accepté par les trois connecteurs (CalDAV, Pronote iCal, Blog RSS).
:return: None
"""
url = "https://endpoint.example.test/feed"
for name, build in _CONNECTOR_BUILDERS.items():
settings = build(url)
endpoint = getattr(settings, "endpoint", None) or settings.ical_endpoint # type: ignore[union-attr]
assert endpoint is not None
assert endpoint.url.get_secret_value() == url, name
@pytest.mark.parametrize("connector", ["caldav", "pronote_ical", "blog"])
def test_external_endpoint_matrix_rejects_file_scheme(connector: str) -> None:
"""Le schéma ``file://`` est refusé par les trois connecteurs.
Contrat corrigé : les endpoints externes doivent être réseau (HTTPS) ;
aucun connecteur n'accepte un fichier local.
:param connector: Nom du connecteur testé.
:return: None
"""
with pytest.raises(ValidationError):
_CONNECTOR_BUILDERS[connector]("file:///tmp/fixture.ics")
@pytest.mark.parametrize(
("connector", "url"),
[
("caldav", "http://endpoint.example.test/feed"),
("pronote_ical", "http://pronote.example.test/ical.ics"),
("blog", "http://blog.example.test/feed"),
("caldav", "ftp://endpoint.example.test/feed"),
("pronote_ical", "gopher://pronote.example.test/ical.ics"),
("blog", "ftp://blog.example.test/feed"),
],
)
def test_external_endpoint_matrix_rejects_insecure_and_other_schemes(
connector: str, url: str
) -> None:
"""HTTP non-loopback et les schémas non HTTP(S) sont refusés par les trois connecteurs.
:param connector: Nom du connecteur testé.
:param url: URL à refuser.
:return: None
"""
with pytest.raises(ValidationError):
_CONNECTOR_BUILDERS[connector](url)
@pytest.mark.parametrize(
("connector", "url"),
[
("caldav", "http://localhost:8080/dav"),
("caldav", "http://127.0.0.1:8080/dav"),
("caldav", "http://[::1]:8080/dav"),
("pronote_ical", "http://localhost/ical.ics"),
("blog", "http://localhost/feed"),
],
)
def test_external_endpoint_matrix_rejects_loopback_http_by_default(
connector: str, url: str
) -> None:
"""HTTP loopback est refusé sans autorisation explicite du mode non sécurisé.
:param connector: Nom du connecteur testé.
:param url: URL loopback HTTP à refuser.
:return: None
"""
with pytest.raises(ValidationError):
_CONNECTOR_BUILDERS[connector](url)
@pytest.mark.parametrize(
"url",
["http://localhost:8080/dav", "http://127.0.0.1:8080/dav", "http://[::1]:8080/dav"],
)
def test_caldav_endpoint_accepts_loopback_http_only_with_insecure_flag(url: str) -> None:
"""CalDAV accepte HTTP loopback uniquement avec ``allow_insecure_http=True``.
:param url: URL loopback HTTP à accepter.
:return: None
"""
settings = CalDAVSettings(
endpoint=ExternalEndpoint(url=SecretStr(url)), allow_insecure_http=True
)
assert settings.endpoint is not None
assert settings.endpoint.url.get_secret_value() == url
@pytest.mark.parametrize(
("connector", "url"),
[
("caldav", "https:///missing-host"),
("pronote_ical", "https:///missing-host"),
("blog", "https:///missing-host"),
("caldav", "https://host:bad"),
("pronote_ical", "https://host:bad"),
("blog", "https://host:bad"),
],
)
def test_external_endpoint_matrix_rejects_malformed_urls(connector: str, url: str) -> None:
"""Les URL malformées (hôte manquant, port invalide) sont refusées.
:param connector: Nom du connecteur testé.
:param url: URL malformée à refuser.
:return: None
"""
with pytest.raises(ValidationError):
_CONNECTOR_BUILDERS[connector](url)
@pytest.mark.parametrize(
"connector",
["caldav", "pronote_ical", "blog"],
)
def test_external_endpoint_matrix_rejects_userinfo(connector: str) -> None:
"""Les credentials dans l'URL (``user:pass@host``) sont refusés.
:param connector: Nom du connecteur testé.
:return: None
"""
url = "https://user:pass@endpoint.example.test/feed" # pragma: allowlist secret
with pytest.raises(ValidationError):
_CONNECTOR_BUILDERS[connector](url)
@pytest.mark.parametrize(
"connector",
["caldav", "pronote_ical", "blog"],
)
def test_external_endpoint_matrix_redacts_sensitive_query_params(connector: str) -> None:
"""Les paramètres sensibles de la query sont masqués à la sérialisation.
:param connector: Nom du connecteur testé.
:return: None
"""
secret = "SECRET_QUERY_TOKEN" # pragma: allowlist secret
url = f"https://endpoint.example.test/feed?icalsecurise={secret}"
settings = _CONNECTOR_BUILDERS[connector](url)
endpoint = getattr(settings, "endpoint", None) or settings.ical_endpoint # type: ignore[union-attr]
assert endpoint is not None
dumped = endpoint.model_dump_json()
assert secret not in dumped
assert "REDACTED" in dumped
assert secret not in repr(endpoint)
@pytest.mark.parametrize(
("connector", "legacy_field", "canonical_url", "legacy_url"),
[
("caldav", "url", "https://canonical.example.test/dav", "https://legacy.example.test/dav"),
(
"blog",
"rss_url",
"https://canonical.example.test/feed",
"https://legacy.example.test/feed",
),
(
"pronote",
"url",
"https://canonical.example.test/parent.html",
"https://legacy.example.test/parent.html",
),
(
"pronote",
"ical_url",
"https://canonical.example.test/ical.ics",
"https://legacy.example.test/ical.ics",
),
],
)
def test_deprecated_aliases_warn_and_canonical_wins(
connector: str, legacy_field: str, canonical_url: str, legacy_url: str
) -> None:
"""Les alias obsolètes émettent un DeprecationWarning et l'URL canonique gagne.
:param connector: Connecteur testé (``caldav``, ``blog`` ou ``pronote``).
:param legacy_field: Nom du champ obsolète.
:param canonical_url: URL canonique (doit gagner).
:param legacy_url: URL fournie via l'alias obsolète.
:return: None
"""
if connector == "caldav":
caldav_kwargs: dict[str, object] = {
legacy_field: SecretStr(legacy_url),
"endpoint": {"url": canonical_url},
}
with pytest.warns(DeprecationWarning, match="CALDAV_URL"):
caldav_settings = CalDAVSettings(**caldav_kwargs) # type: ignore[arg-type]
assert caldav_settings.endpoint is not None
assert caldav_settings.endpoint.url.get_secret_value() == canonical_url
elif connector == "blog":
blog_kwargs: dict[str, object] = {
legacy_field: legacy_url,
"endpoint": {"url": canonical_url},
}
with pytest.warns(DeprecationWarning, match="BLOG_RSS_URL"):
blog_settings = BlogSettings(**blog_kwargs) # type: ignore[arg-type]
assert blog_settings.endpoint.url.get_secret_value() == canonical_url
else:
canonical_field = "endpoint" if legacy_field == "url" else "ical_endpoint"
legacy_value: object = SecretStr(legacy_url) if legacy_field == "ical_url" else legacy_url
pronote_kwargs: dict[str, object] = {
legacy_field: legacy_value,
canonical_field: {"url": canonical_url},
}
with pytest.warns(DeprecationWarning, match="PRONOTE_"):
pronote_settings = PronoteSettings(**pronote_kwargs) # type: ignore[arg-type]
endpoint = getattr(pronote_settings, canonical_field)
assert endpoint is not None
assert endpoint.url.get_secret_value() == canonical_url
-83
View File
@@ -1,83 +0,0 @@
"""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 == []
+20 -23
View File
@@ -54,7 +54,7 @@ def fixture_mock_settings() -> Settings:
return Settings(
pronote=PronoteSettings(
url="https://pronote.example.com",
ical_url=SecretStr("https://ical.example.test/ical.ics"),
ical_url=SecretStr("file:///fake/ical.ics"),
agenda_source="auto",
homework_source="auto",
username="testuser",
@@ -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,15 +280,14 @@ 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,
patch("pronote_sync.sources.pronote.fallback.parse_ical") as m_parse_ical,
):
m_fetch_ical.side_effect = OSError(
"Échec de la récupération du flux iCal https://ical.example.test/ical.ics : "
"iCal unreachable"
"Impossible de lire le fichier iCal file:///fake/ical.ics : iCal unreachable"
)
m_parse_ical.side_effect = OSError("iCal parse error")
@@ -297,7 +296,7 @@ def test_fetch_agenda_ical_mode_failure(mock_fetcher: PronoteFetcher) -> None:
assert "Impossible de récupérer l'agenda : la source ical a échoué" in str(exc_info.value)
# Vérifie que le message ne contient pas de secret
assert "https://ical.example.test/ical.ics" not in str(exc_info.value)
assert "file:///fake/ical.ics" not in str(exc_info.value)
def test_fetch_agenda_pronotepy_mode_failure(mock_fetcher: PronoteFetcher) -> None:
@@ -312,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")
@@ -439,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,
@@ -532,15 +531,15 @@ 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,
patch("pronote_sync.sources.pronote.fallback.parse_ical") as m_parse_ical,
):
error_msg = (
"Échec de la récupération du flux iCal https://ical.example.test/ical.ics"
"?icalsecurise=SECRET_TOKEN_123 : [Errno 2] No such file or directory"
"Impossible de lire le fichier iCal file:///ical?icalsecurise=SECRET_TOKEN_123 : "
"[Errno 2] No such file or directory"
)
m_fetch_ical.side_effect = OSError(error_msg)
m_parse_ical.side_effect = OSError("parse error")
@@ -629,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()
@@ -913,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()
@@ -931,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
@@ -1023,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
@@ -1049,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
@@ -1066,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
@@ -1082,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()
+23 -22
View File
@@ -1,7 +1,7 @@
"""Tests unitaires pour le module iCal : téléchargement et parsing.
Ce module teste :
- La récupération du flux iCal (HTTP)
- La récupération du flux iCal (file://, HTTP)
- Le parsing des événements en modèles Lesson, SchoolEvent
- L'extraction et normalisation des devoirs
- La collecte et déduplication des devoirs par date cible
@@ -11,6 +11,8 @@ Les tests utilisent des mocks pour éviter tout appel réseau réel.
from __future__ import annotations
import tempfile
import urllib.parse
from datetime import date, datetime
from pathlib import Path
@@ -55,39 +57,38 @@ def invalid_ical_content() -> str:
return "INVALID:CONTENT\nThis is not a valid iCal file."
@responses.activate
def test_fetch_ical_fixture_content() -> None:
"""fetch_ical("https://…") sur un mock réseau retourne un contenu commençant par BEGIN:VCALENDAR.
def test_fetch_ical_file_protocol() -> None:
"""fetch_ical("file://tests/fixtures/pronote-4e.ics") retourne un contenu commençant par BEGIN:VCALENDAR.
:return: None
"""
fixture_path = Path(__file__).parent.parent / "fixtures" / "pronote-4e.ics"
responses.add(
responses.GET,
"https://pronote.example.test/ical.ics",
body=fixture_path.read_text(encoding="utf-8"),
status=200,
)
url = f"file://{fixture_path}"
content = fetch_ical("https://pronote.example.test/ical.ics")
content = fetch_ical(url)
assert content.lstrip().startswith("BEGIN:VCALENDAR")
@responses.activate
def test_fetch_ical_url_with_encoded_characters() -> None:
"""fetch_ical préserve les caractères encodés (%20) de l'URL interrogée.
def test_fetch_ical_file_uri_decoding() -> None:
"""fetch_ical("file://path%20with%20spaces") décode correctement le chemin.
:return: None
"""
responses.add(
responses.GET,
"https://pronote.example.test/fichier%20avec%20espaces.ics",
body="BEGIN:VCALENDAR\nVERSION:2.0\nPRODID:-//Test//Test//FR\nEND:VCALENDAR",
status=200,
)
# Créer un fichier temporaire avec un espace dans le nom
with tempfile.TemporaryDirectory() as tmpdir:
temp_path = Path(tmpdir) / "fichier avec espaces.ics"
temp_path.write_text(
"BEGIN:VCALENDAR\nVERSION:2.0\nPRODID:-//Test//Test//FR\nEND:VCALENDAR",
encoding="utf-8",
)
content = fetch_ical("https://pronote.example.test/fichier%20avec%20espaces.ics")
assert content.lstrip().startswith("BEGIN:VCALENDAR")
# URL encodée avec espace
encoded_name = urllib.parse.quote("fichier avec espaces.ics")
url = f"file://{tmpdir}/{encoded_name}"
# Cela devrait fonctionner car Path.read_text décode l'URL
content = fetch_ical(url)
assert content.lstrip().startswith("BEGIN:VCALENDAR")
@responses.activate