Compare commits
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
328eaeec88 |
+4
-11
@@ -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
@@ -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
@@ -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
@@ -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`
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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, 42–50 | ✅ `sources/ical.py` | **Cohérent** |
|
||||
| `PRONOTE_ENDPOINT__URL` | ✅ Ligne 3 | ✅ `client.py` (ligne 294) | **Cohérent** |
|
||||
| `PRONOTE_ICAL_URL` | ✅ Lignes 2, 42–50 | ✅ `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
@@ -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))
|
||||
|
||||
@@ -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
|
||||
)
|
||||
|
||||
@@ -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,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
|
||||
|
||||
@@ -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.
|
||||
|
||||
@@ -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,
|
||||
|
||||
@@ -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).
|
||||
|
||||
@@ -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()
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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:
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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
|
||||
):
|
||||
|
||||
@@ -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"))
|
||||
|
||||
@@ -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,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)
|
||||
|
||||
@@ -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
@@ -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
|
||||
|
||||
@@ -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
@@ -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
@@ -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
|
||||
|
||||
Reference in New Issue
Block a user