Compare commits

..
16 changed files with 331 additions and 258 deletions
+2 -5
View File
@@ -28,9 +28,7 @@ PRONOTE_AUTH_MODE=password
# PRONOTE_ACCOUNT_PIN= # PRONOTE_ACCOUNT_PIN=
# --- CalDAV --- # --- CalDAV ---
# Endpoint commun (URL potentiellement sensible, masquée dans les journaux) CALDAV_URL=https://caldav.example.com/calendars/user/pronote/
CALDAV_ENDPOINT__URL=https://caldav.example.com/calendars/user/pronote/
# Ancien nom temporairement supporté avec un avertissement de dépréciation : CALDAV_URL
CALDAV_USERNAME=user@example.com CALDAV_USERNAME=user@example.com
CALDAV_PASSWORD=your_caldav_password CALDAV_PASSWORD=your_caldav_password
CALDAV_CALENDAR_PATH=/pronote-sync/ CALDAV_CALENDAR_PATH=/pronote-sync/
@@ -94,8 +92,7 @@ AI_BASE_URL=https://api.openai.com/v1
# --- Blog --- # --- Blog ---
BLOG_ENABLED=false BLOG_ENABLED=false
BLOG_ENDPOINT__URL=https://blogpeda.ac-bordeaux.fr/cjeliote/?feed=rss2 BLOG_RSS_URL=https://blogpeda.ac-bordeaux.fr/cjeliote/?feed=rss2
# Ancien nom temporairement supporté avec un avertissement de dépréciation : BLOG_RSS_URL
# --- Divers --- # --- Divers ---
DRY_RUN=false DRY_RUN=false
+2 -2
View File
@@ -140,7 +140,7 @@
"filename": "GUIDE_DEV_PYTHON.md", "filename": "GUIDE_DEV_PYTHON.md",
"hashed_secret": "90bd1b48e958257948487b90bee080ba5ed00caa", "hashed_secret": "90bd1b48e958257948487b90bee080ba5ed00caa",
"is_verified": false, "is_verified": false,
"line_number": 5183 "line_number": 5186
} }
], ],
"tests/unit/test_caldav_gateway.py": [ "tests/unit/test_caldav_gateway.py": [
@@ -185,5 +185,5 @@
} }
] ]
}, },
"generated_at": "2026-09-13T09:52:29Z" "generated_at": "2026-09-12T22:12:56Z"
} }
+8 -5
View File
@@ -264,7 +264,7 @@ Le projet utilise **`pydantic-settings`** pour valider et charger la configurati
| `PRONOTE_USERNAME` | Identifiant Pronote (si `pronotepy` utilisé). | `parent.dupont` | `str` | | `PRONOTE_USERNAME` | Identifiant Pronote (si `pronotepy` utilisé). | `parent.dupont` | `str` |
| `PRONOTE_PASSWORD` | Mot de passe Pronote (si `pronotepy` utilisé). | `SecretStr` (masqué) | `SecretStr` | | `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` | | `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_USERNAME` | Identifiant CalDAV. | `user@example.com` | `str` |
| `CALDAV_PASSWORD` | Mot de passe CalDAV. | `SecretStr` (masqué) | `SecretStr` | | `CALDAV_PASSWORD` | Mot de passe CalDAV. | `SecretStr` (masqué) | `SecretStr` |
| `CALDAV_CALENDAR_PATH` | Chemin du calendrier CalDAV de destination. | `/pronote-sync/` | `str` | | `CALDAV_CALENDAR_PATH` | Chemin du calendrier CalDAV de destination. | `/pronote-sync/` | `str` |
@@ -1113,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. Aucun fichier d'état local n'est utilisé : l'état est géré en mémoire par run.
# Initialisation # 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)() blog_state = ## (section obsolète supprimée)()
# Récupération des nouveaux articles # Récupération des nouveaux articles
@@ -1218,7 +1218,7 @@ Ajouter les variables suivantes dans la configuration :
| **Variable** | **Description** | **Valeur par défaut** | **Type** | | **Variable** | **Description** | **Valeur par défaut** | **Type** |
|----------------------------|-------------------------------------------------------------------------------|-----------------------|-------------------| |----------------------------|-------------------------------------------------------------------------------|-----------------------|-------------------|
| `BLOG_ENABLED` | Activer la récupération du blog. | `False` | `bool` | | `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 #### 5 bis.9.2 Modèle Pydantic pour la configuration du blog
@@ -1230,7 +1230,10 @@ from pydantic_settings import BaseSettings, SettingsConfigDict
class BlogSettings(BaseSettings): class BlogSettings(BaseSettings):
model_config = SettingsConfigDict(env_prefix="BLOG_", env_file=".env", extra="ignore") model_config = SettingsConfigDict(env_prefix="BLOG_", env_file=".env", extra="ignore")
enabled: bool = Field(False, description="Activer la récupération du blog") 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`** : **Intégration dans `Settings`** :
@@ -1250,7 +1253,7 @@ class Settings(BaseSettings):
```ini ```ini
# --- Blog du collège --- # --- Blog du collège ---
BLOG_ENABLED=true 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
``` ```
--- ---
+39 -160
View File
@@ -10,76 +10,21 @@ from __future__ import annotations
import warnings import warnings
from datetime import date from datetime import date
from typing import Annotated, Literal from typing import Literal
from urllib.parse import urlparse from urllib.parse import urlparse
from pydantic import ( from pydantic import (
AfterValidator,
BaseModel,
ConfigDict,
Field, Field,
SecretStr, SecretStr,
ValidationInfo,
field_serializer, field_serializer,
field_validator,
model_validator, model_validator,
) )
from pydantic_settings import BaseSettings, SettingsConfigDict from pydantic_settings import BaseSettings, SettingsConfigDict
from pronote_sync.utils.redaction import redact_url from pronote_sync.utils.redaction import redact_url
_EXTERNAL_ENDPOINT_SCHEMES: frozenset[str] = frozenset({"file", "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 d'une URL d'endpoint externe.
: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 ou utilise un schéma inconnu.
"""
is_valid = False
try:
parsed = urlparse(value.get_secret_value())
_ = parsed.port
is_valid = (
parsed.scheme in _EXTERNAL_ENDPOINT_SCHEMES
and (parsed.scheme not in {"http", "https"} or parsed.hostname is not None)
and (parsed.scheme != "file" or bool(parsed.path))
)
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 les transports ``https``, ``http`` et ``file``. 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): class PronoteSettings(BaseSettings):
"""Paramètres d'accès à Pronote (flux iCal et API ``pronotepy``). """Paramètres d'accès à Pronote (flux iCal et API ``pronotepy``).
@@ -146,88 +91,73 @@ class CalDAVSettings(BaseSettings):
Les variables d'environnement correspondantes sont préfixées par Les variables d'environnement correspondantes sont préfixées par
``CALDAV_``. L'URL est traitée comme potentiellement sensible (au même ``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é ; 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``, 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``. ``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( model_config = SettingsConfigDict(env_file=".env", extra="ignore", env_prefix="CALDAV_")
env_file=".env",
env_nested_delimiter="__",
extra="ignore",
env_prefix="CALDAV_",
)
allow_insecure_http: bool = False allow_insecure_http: bool = False
endpoint: ExternalEndpoint | None = None url: SecretStr | None = None
url: SecretStr | None = Field(
default=None,
exclude=True,
deprecated="Utiliser endpoint.url à la place (CALDAV_URL obsolète).",
)
username: str | None = None username: str | None = None
password: SecretStr | None = None password: SecretStr | None = None
calendar_path: str = "/pronote-sync/" calendar_path: str = "/pronote-sync/"
@model_validator(mode="before") @field_serializer("url")
@classmethod def _serialize_url(self, value: SecretStr | None) -> str | None:
def _migrate_legacy_url(cls, data: object) -> object: """Masque l'URL CalDAV lors de la sérialisation (repr, str, JSON).
"""Migre ``url`` vers l'endpoint commun avec un avertissement.
:param data: Données brutes du modèle. :param value: Valeur du champ ``url`` (secret potentiel).
:return: Données complétées avec ``endpoint`` si nécessaire. :return: URL avec les éléments sensibles remplacés par ``REDACTED``,
:rtype: object ou ``None`` si la valeur est absente.
:rtype: str | None
""" """
if not isinstance(data, dict) or data.get("url") is None: if value is None:
return data return None
migrated_data = data.copy() return redact_url(value.get_secret_value())
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
@model_validator(mode="after") @field_validator("url")
def _validate_endpoint_policy(self) -> CalDAVSettings: @classmethod
"""Applique la politique HTTPS/HTTP loopback propre à CalDAV. 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 HTTPS est toujours accepté. HTTP n'est accepté que pour un hôte de
boucle locale (``localhost``, ``127.0.0.1``, ``::1``) et uniquement boucle locale (``localhost``, ``127.0.0.1``, ``::1``) et uniquement
lorsque ``allow_insecure_http`` vaut ``True``. Les autres schémas du lorsque ``allow_insecure_http`` vaut ``True``. Les messages d'erreur
socle commun sont refusés pour ce connecteur. ne contiennent jamais l'URL brute (susceptible de contenir des
identifiants).
:return: Instance validée inchangée. :param v: Valeur du champ ``url`` à valider.
:rtype: CalDAVSettings :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 :raises ValueError: Si le schéma n'est pas supporté ou si l'URL HTTP
n'est pas autorisée. n'est pas autorisée.
""" """
if self.endpoint is None: if v is None:
return self return v
raw_url = self.endpoint.url.get_secret_value() raw_url = v.get_secret_value()
parsed = urlparse(raw_url) parsed = urlparse(raw_url)
if parsed.scheme not in ("http", "https"): if parsed.scheme not in ("http", "https"):
raise ValueError("URL CalDAV invalide : schéma non supporté") from None raise ValueError("URL CalDAV invalide : schéma non supporté") from None
if parsed.scheme == "https": if parsed.scheme == "https":
return self return v
# HTTP — check allow_insecure_http flag and loopback # 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( raise ValueError(
"URL CalDAV non sécurisée : HTTPS requis (ou activer " "URL CalDAV non sécurisée : HTTPS requis (ou activer "
"CALDAV_ALLOW_INSECURE_HTTP pour localhost)" "CALDAV_ALLOW_INSECURE_HTTP pour localhost)"
) from None ) from None
hostname = parsed.hostname or "" 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( raise ValueError(
"URL CalDAV non sécurisée : HTTP autorisé uniquement pour localhost" "URL CalDAV non sécurisée : HTTP autorisé uniquement pour localhost"
) from None ) from None
return self return v
_XMPP_LOOPBACK_HOSTS: frozenset[str] = frozenset({"localhost", "127.0.0.1", "::1"}) _XMPP_LOOPBACK_HOSTS: frozenset[str] = frozenset({"localhost", "127.0.0.1", "::1"})
@@ -352,60 +282,10 @@ class BlogSettings(BaseSettings):
``BLOG_``. ``BLOG_``.
""" """
model_config = SettingsConfigDict( model_config = SettingsConfigDict(env_file=".env", extra="ignore", env_prefix="BLOG_")
env_file=".env",
env_nested_delimiter="__",
extra="ignore",
env_prefix="BLOG_",
)
enabled: bool = False enabled: bool = False
endpoint: ExternalEndpoint = Field( rss_url: str = "https://blogpeda.ac-bordeaux.fr/cjeliote/?feed=rss2"
default_factory=lambda: ExternalEndpoint(
url=SecretStr("https://blogpeda.ac-bordeaux.fr/cjeliote/?feed=rss2")
)
)
rss_url: str | None = Field(
default=None,
exclude=True,
deprecated="Utiliser endpoint.url à la place (BLOG_RSS_URL obsolète).",
)
@model_validator(mode="before")
@classmethod
def _migrate_legacy_rss_url(cls, data: object) -> object:
"""Migre ``rss_url`` vers l'endpoint commun avec un avertissement.
:param data: Données brutes du modèle.
:return: Données complétées avec ``endpoint`` si nécessaire.
:rtype: object
"""
if not isinstance(data, dict) or data.get("rss_url") is None:
return data
migrated_data = data.copy()
warnings.warn(
"BLOG_RSS_URL est obsolète : utiliser BLOG_ENDPOINT__URL.",
DeprecationWarning,
stacklevel=2,
)
if migrated_data.get("endpoint") is None:
migrated_data["endpoint"] = {"url": migrated_data["rss_url"]}
return migrated_data
@model_validator(mode="after")
def _validate_endpoint_policy(self) -> BlogSettings:
"""Refuse les transports non sûrs pour le flux RSS de production.
Le transport ``file`` reste autorisé pour les fixtures locales.
:return: Instance validée inchangée.
:rtype: BlogSettings
:raises ValueError: Si le schéma n'est ni ``https`` ni ``file``.
"""
scheme = urlparse(self.endpoint.url.get_secret_value()).scheme
if scheme not in {"https", "file"}:
raise ValueError("URL RSS invalide : HTTPS ou file requis") from None
return self
class AppSettings(BaseSettings): class AppSettings(BaseSettings):
@@ -462,10 +342,9 @@ class Settings(BaseSettings):
self.pronote.password, self.pronote.password,
self.pronote.qr_pin, self.pronote.qr_pin,
self.pronote.account_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.caldav.password,
self.xmpp.password, self.xmpp.password,
self.ai.api_key, self.ai.api_key,
self.blog.endpoint.url,
] ]
return tuple(dict.fromkeys(secret for secret in secrets if secret is not None)) return tuple(dict.fromkeys(secret for secret in secrets if secret is not None))
+1 -5
View File
@@ -138,11 +138,7 @@ class PipelineRunner:
comparator = ( comparator = (
AgendaComparator(theoretical_provider) if theoretical_provider is not None else None AgendaComparator(theoretical_provider) if theoretical_provider is not None else None
) )
blog_client = ( blog_client = BlogRSSClient(settings.blog.rss_url) if settings.blog.enabled else None
BlogRSSClient(settings.blog.endpoint.url.get_secret_value())
if settings.blog.enabled
else None
)
blog_state = ( blog_state = (
BlogRSSState(persistence_enabled=persistence_enabled) if settings.blog.enabled else None BlogRSSState(persistence_enabled=persistence_enabled) if settings.blog.enabled else None
) )
+5 -2
View File
@@ -121,6 +121,7 @@ def fetch_step(
:raises PronoteAuthRotationError: Si une rotation du token d'authentification :raises PronoteAuthRotationError: Si une rotation du token d'authentification
pronotepy est nécessaire : propagée telle quelle jusqu'au pipeline. pronotepy est nécessaire : propagée telle quelle jusqu'au pipeline.
""" """
critical_error: PipelineCriticalError | None = None
try: try:
lessons, school_events = fetcher.fetch_agenda() lessons, school_events = fetcher.fetch_agenda()
target_date = resolve_target_date(today or date.today(), lessons, school_events) target_date = resolve_target_date(today or date.today(), lessons, school_events)
@@ -130,9 +131,11 @@ def fetch_step(
except PronoteAuthRotationError: except PronoteAuthRotationError:
raise raise
except Exception as exc: except Exception as exc:
raise PipelineCriticalError( critical_error = PipelineCriticalError(
f"Récupération Pronote impossible : {redact_exception(exc)}", step="fetch" f"Récupération Pronote impossible : {redact_exception(exc)}", step="fetch"
) from None )
if critical_error is not None:
raise critical_error from None
messages, warnings = _fetch_optional_messages(fetcher) messages, warnings = _fetch_optional_messages(fetcher)
return ( return (
+8 -5
View File
@@ -5,7 +5,7 @@ from __future__ import annotations
from pronote_sync.sources.blog.result import BlogRSSFetchResult from pronote_sync.sources.blog.result import BlogRSSFetchResult
from pronote_sync.sources.blog.rss import BlogRSSClient from pronote_sync.sources.blog.rss import BlogRSSClient
from pronote_sync.sources.blog.state import BlogRSSState from pronote_sync.sources.blog.state import BlogRSSState
from pronote_sync.utils.redaction import redact_exception from pronote_sync.utils.redaction import redact_exception, redact_secrets
def fetch_blog_step(client: BlogRSSClient | None, state: BlogRSSState | None) -> BlogRSSFetchResult: def fetch_blog_step(client: BlogRSSClient | None, state: BlogRSSState | None) -> BlogRSSFetchResult:
@@ -24,15 +24,18 @@ def fetch_blog_step(client: BlogRSSClient | None, state: BlogRSSState | None) ->
""" """
if client is None or state is None: if client is None or state is None:
return BlogRSSFetchResult() return BlogRSSFetchResult()
error_message: str | None = None
try: try:
etag, last_modified = state.get_cache_headers() etag, last_modified = state.get_cache_headers()
result = client.fetch_and_parse( result = client.fetch_and_parse(
known_guids=state.get_known_guids(), etag=etag, last_modified=last_modified known_guids=state.get_known_guids(), etag=etag, last_modified=last_modified
) )
if result.error is not None: if result.error is not None:
raise RuntimeError(result.error) from None error_message = f"Récupération du blog échouée : {redact_secrets(result.error)}"
if not result.not_modified and not result.articles: elif not result.not_modified and not result.articles:
state.update_cache_headers(result.etag, result.last_modified) state.update_cache_headers(result.etag, result.last_modified)
return result
except Exception as exc: except Exception as exc:
raise RuntimeError(f"Récupération du blog échouée : {redact_exception(exc)}") from None error_message = f"Récupération du blog échouée : {redact_exception(exc)}"
if error_message is not None:
raise RuntimeError(error_message) from None
return result
+5 -2
View File
@@ -218,6 +218,7 @@ class PronoteAuthState:
} }
tmp_file = self._state_file.with_suffix(".tmp") tmp_file = self._state_file.with_suffix(".tmp")
fd: int | None = None fd: int | None = None
write_error: PronoteSyncError | None = None
try: try:
# Nettoie un éventuel fichier temporaire stale laissé par une exécution interrompue. # Nettoie un éventuel fichier temporaire stale laissé par une exécution interrompue.
if tmp_file.exists(): if tmp_file.exists():
@@ -259,10 +260,12 @@ class PronoteAuthState:
"Nettoyage du fichier temporaire d'état d'authentification Pronote échoué : %s", "Nettoyage du fichier temporaire d'état d'authentification Pronote échoué : %s",
redact_exception(cleanup_exc), redact_exception(cleanup_exc),
) )
raise PronoteSyncError( write_error = PronoteSyncError(
f"Impossible d'écrire le fichier d'état d'authentification Pronote " f"Impossible d'écrire le fichier d'état d'authentification Pronote "
f"{redact_secrets(str(self._state_file))}." f"{redact_secrets(str(self._state_file))}."
) from None )
if write_error is not None:
raise write_error from None
def clear(self) -> None: def clear(self) -> None:
"""Supprime le fichier d'état d'authentification. """Supprime le fichier d'état d'authentification.
+61 -56
View File
@@ -269,34 +269,36 @@ class PronoteFetcher:
primary, primary,
redact_exception(exc), redact_exception(exc),
) )
if fallback is None: if fallback is None:
raise PipelineCriticalError( raise PipelineCriticalError(
f"Impossible de récupérer l'agenda : la source {primary} a échoué" f"Impossible de récupérer l'agenda : la source {primary} a échoué"
) from None ) from None
logger.info("Repli sur %s pour l'agenda.", fallback) logger.info("Repli sur %s pour l'agenda.", fallback)
try: fallback_result: tuple[list[Lesson], list[SchoolEvent]] | None = None
lessons, school_events = self._fetch_agenda_source(fallback) try:
except PronoteAuthRotationError: fallback_result = self._fetch_agenda_source(fallback)
raise except PronoteAuthRotationError:
except Exception as exc: raise
logger.error( except Exception as exc:
"Échec de la récupération %s pour l'agenda : %s", logger.error(
fallback, "Échec de la récupération %s pour l'agenda : %s",
redact_exception(exc), fallback,
) redact_exception(exc),
raise PipelineCriticalError( )
f"Impossible de récupérer l'agenda : les sources {primary}" if fallback_result is None:
f" et {fallback} ont échoué" raise PipelineCriticalError(
) from None f"Impossible de récupérer l'agenda : les sources {primary} et {fallback} ont échoué"
if not lessons: ) from None
logger.warning( lessons, school_events = fallback_result
"Le repli %s pour l'agenda a retourné un résultat vide après l'échec " if not lessons:
"de %s : impossible de distinguer une absence de cours d'un échec " logger.warning(
"silencieux.", "Le repli %s pour l'agenda a retourné un résultat vide après l'échec "
fallback, "de %s : impossible de distinguer une absence de cours d'un échec "
primary, "silencieux.",
) fallback,
return lessons, school_events primary,
)
return lessons, school_events
def _fetch_homework_ical(self, target_date: date) -> list[Homework]: def _fetch_homework_ical(self, target_date: date) -> list[Homework]:
"""Récupère les devoirs depuis le flux iCal pour la date cible. """Récupère les devoirs depuis le flux iCal pour la date cible.
@@ -396,34 +398,37 @@ class PronoteFetcher:
primary, primary,
redact_exception(exc), redact_exception(exc),
) )
if fallback is None: if fallback is None:
raise PipelineCriticalError( raise PipelineCriticalError(
f"Impossible de récupérer les devoirs : la source {primary} a échoué" f"Impossible de récupérer les devoirs : la source {primary} a échoué"
) from None ) from None
logger.info("Repli sur %s pour les devoirs.", fallback) logger.info("Repli sur %s pour les devoirs.", fallback)
try: fallback_result: list[Homework] | None = None
homeworks = self._fetch_homework_source(fallback, target_date) try:
except PronoteAuthRotationError: fallback_result = self._fetch_homework_source(fallback, target_date)
raise except PronoteAuthRotationError:
except Exception as exc: raise
logger.error( except Exception as exc:
"Échec de la récupération %s pour les devoirs : %s", logger.error(
fallback, "Échec de la récupération %s pour les devoirs : %s",
redact_exception(exc), fallback,
) redact_exception(exc),
raise PipelineCriticalError( )
f"Impossible de récupérer les devoirs : les sources {primary}" if fallback_result is None:
f" et {fallback} ont échoué" raise PipelineCriticalError(
) from None f"Impossible de récupérer les devoirs : les sources {primary}"
if not homeworks: f" et {fallback} ont échoué"
logger.warning( ) from None
"Le repli %s pour les devoirs a retourné un résultat vide après " homeworks = fallback_result
"l'échec de %s : impossible de distinguer une absence de devoirs " if not homeworks:
"d'un échec silencieux.", logger.warning(
fallback, "Le repli %s pour les devoirs a retourné un résultat vide après "
primary, "l'échec de %s : impossible de distinguer une absence de devoirs "
) "d'un échec silencieux.",
return homeworks fallback,
primary,
)
return homeworks
def fetch_messages(self) -> list[Message]: def fetch_messages(self) -> list[Message]:
"""Récupère les messages des discussions Pronote (toujours via pronotepy). """Récupère les messages des discussions Pronote (toujours via pronotepy).
+5 -2
View File
@@ -85,6 +85,7 @@ class JsonTheoreticalAgendaProvider:
self._file_path: str = file_path self._file_path: str = file_path
self._parity_service: WeekParityService | None = parity_service self._parity_service: WeekParityService | None = parity_service
self._holiday_calendar: SchoolHolidayCalendar | None = holiday_calendar self._holiday_calendar: SchoolHolidayCalendar | None = holiday_calendar
load_error: PronoteSyncError | None = None
try: try:
content = Path(file_path).read_text(encoding="utf-8") content = Path(file_path).read_text(encoding="utf-8")
parsed = TheoreticalAgendaFile.model_validate_json(content) parsed = TheoreticalAgendaFile.model_validate_json(content)
@@ -94,9 +95,11 @@ class JsonTheoreticalAgendaProvider:
redact_secrets(str(file_path)), redact_secrets(str(file_path)),
redact_exception(exc), redact_exception(exc),
) )
raise PronoteSyncError( load_error = PronoteSyncError(
f"Le fichier d'agenda théorique est invalide : {redact_secrets(str(file_path))}" f"Le fichier d'agenda théorique est invalide : {redact_secrets(str(file_path))}"
) from None )
if load_error is not None:
raise load_error from None
self._lessons: tuple[TheoreticalLessonEntry, ...] = parsed.lessons self._lessons: tuple[TheoreticalLessonEntry, ...] = parsed.lessons
if self._parity_service is None and any( if self._parity_service is None and any(
entry.week in ("even", "odd") for entry in self._lessons entry.week in ("even", "odd") for entry in self._lessons
+5 -2
View File
@@ -77,6 +77,7 @@ class SchoolHolidayCalendar:
raise PronoteSyncError( raise PronoteSyncError(
f"Le fichier de vacances scolaires est introuvable : {redact_secrets(str(path))}" f"Le fichier de vacances scolaires est introuvable : {redact_secrets(str(path))}"
) from None ) from None
load_error: PronoteSyncError | None = None
try: try:
data: Any = json.loads(path.read_text(encoding="utf-8")) data: Any = json.loads(path.read_text(encoding="utf-8"))
file_model: SchoolHolidayFile = SchoolHolidayFile.model_validate(data) file_model: SchoolHolidayFile = SchoolHolidayFile.model_validate(data)
@@ -86,9 +87,11 @@ class SchoolHolidayCalendar:
redact_secrets(str(path)), redact_secrets(str(path)),
redact_exception(exc), redact_exception(exc),
) )
raise PronoteSyncError( load_error = PronoteSyncError(
f"Le fichier de vacances scolaires est invalide : {redact_secrets(str(path))}" f"Le fichier de vacances scolaires est invalide : {redact_secrets(str(path))}"
) from None )
if load_error is not None:
raise load_error from None
self._periods = file_model.periods self._periods = file_model.periods
def is_holiday(self, target_date: date) -> bool: def is_holiday(self, target_date: date) -> bool:
+2 -3
View File
@@ -79,12 +79,11 @@ class CalDAVGateway:
else cast(Callable[..., Any], caldav.DAVClient) else cast(Callable[..., Any], caldav.DAVClient)
) )
self._calendar_path: str = settings.calendar_path 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 = ( 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._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._password_secret: SecretStr | None = settings.password
self._client: Any = None self._client: Any = None
self._calendar: Any = None self._calendar: Any = None
+1 -1
View File
@@ -96,7 +96,7 @@ def synchronize(
) )
if ( if (
settings.caldav.endpoint is None settings.caldav.url is None
or settings.caldav.username is None or settings.caldav.username is None
or settings.caldav.password is None or settings.caldav.password is None
): ):
+31 -8
View File
@@ -26,15 +26,15 @@ _TEXT_SUFFIXES = frozenset(
{".conf", ".ini", ".json", ".md", ".py", ".service", ".timer", ".toml", ".txt", ".yaml", ".yml"} {".conf", ".ini", ".json", ".md", ".py", ".service", ".timer", ".toml", ".txt", ".yaml", ".yml"}
) )
_LITERAL_SECRET_RE = re.compile( _LITERAL_SECRET_RE = re.compile(
r"(?ix)\b[a-z0-9_]*(?:api[_-]?key|access[_-]?token|auth(?:orization)?|icalsecurise|password|secret|token)" r"(?ix)\b[a-z0-9_]*(?:api[_-]?key|access[_-]?token|auth(?:orization)?|icalsecurise|password|pin|secret|token)"
r"\s*[:=]\s*['\"][^'\"\r\n]{3,}['\"]" r"\s*[:=]\s*['\"](?P<value>[^'\"\r\n]{3,})['\"]"
) )
_UNQUOTED_SECRET_RE = re.compile( _UNQUOTED_SECRET_RE = re.compile(
r"(?ix)\b[a-z0-9_]*(?:api[_-]?key|access[_-]?token|auth(?:orization)?|icalsecurise|password|secret|token)" r"(?ix)\b[a-z0-9_]*(?:api[_-]?key|access[_-]?token|auth(?:orization)?|icalsecurise|password|pin|secret|token)"
r"\s*[:=]\s*[a-z0-9][a-z0-9._~+/-]{2,}" r"\s*[:=]\s*(?P<value>[a-z0-9][a-z0-9._~+/-]{2,})"
) )
_URL_SECRET_RE = re.compile( _URL_SECRET_RE = re.compile(
r"(?ix)[?&](?:api[_-]?key|access[_-]?token|auth(?:orization)?|icalsecurise|password|secret|token)" r"(?ix)[?&](?:api[_-]?key|access[_-]?token|auth(?:orization)?|icalsecurise|password|pin|secret|token)"
r"=([^&#\s]{3,})" r"=([^&#\s]{3,})"
) )
_URL_PLACEHOLDER_RE = re.compile( _URL_PLACEHOLDER_RE = re.compile(
@@ -47,6 +47,13 @@ _URL_PLACEHOLDER_RE = re.compile(
r")$" r")$"
) )
_EXTRA_NAMES = frozenset({"pronote_sync"}) _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) @dataclass(frozen=True)
@@ -68,6 +75,16 @@ CommandRunner = Callable[..., subprocess.CompletedProcess[str]]
ContentProvider = Callable[[Path], str | None] 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: def _is_candidate(path: Path) -> bool:
"""Indique si un chemin peut être analysé comme fichier texte. """Indique si un chemin peut être analysé comme fichier texte.
@@ -188,9 +205,15 @@ def find_secrets(
for number, line in enumerate(content.splitlines(), start=1): for number, line in enumerate(content.splitlines(), start=1):
if _ALLOWLIST_MARKER in line: if _ALLOWLIST_MARKER in line:
continue continue
is_literal_secret = _LITERAL_SECRET_RE.search(line) or ( literal_match = _LITERAL_SECRET_RE.search(line)
relative_path.suffix in _UNQUOTED_CONFIG_SUFFIXES unquoted_match = (
and _UNQUOTED_SECRET_RE.search(line) _UNQUOTED_SECRET_RE.search(line)
if relative_path.suffix in _UNQUOTED_CONFIG_SUFFIXES
else None
)
is_literal_secret = any(
match is not None and not _is_assignment_placeholder(match.group("value"))
for match in (literal_match, unquoted_match)
) )
if is_literal_secret: if is_literal_secret:
findings.append(SecretFinding(relative_path, number, "affectation-litterale")) findings.append(SecretFinding(relative_path, number, "affectation-litterale"))
+73
View File
@@ -118,6 +118,27 @@ def test_main_detects_sensitive_url_parameter(
assert sentinel not in output 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( def test_main_ignores_documentation_url_placeholders(
secret_checker: ModuleType, tmp_path: Path, capsys: CaptureFixture[str] secret_checker: ModuleType, tmp_path: Path, capsys: CaptureFixture[str]
) -> None: ) -> None:
@@ -236,6 +257,58 @@ def test_main_detects_prefixed_secret_assignment(
assert sentinel not in output 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( def test_main_detects_short_secret_assignment(
secret_checker: ModuleType, tmp_path: Path, capsys: CaptureFixture[str] secret_checker: ModuleType, tmp_path: Path, capsys: CaptureFixture[str]
) -> None: ) -> None:
+83
View File
@@ -0,0 +1,83 @@
"""Régressions sur le contexte des exceptions expurgées."""
from __future__ import annotations
import ast
import traceback
from datetime import date
from pathlib import Path
import pytest
from pronote_sync.errors import PipelineCriticalError
from pronote_sync.models.agenda import Lesson, SchoolEvent
from pronote_sync.models.homework import Homework
from pronote_sync.models.message import Message
from pronote_sync.pipeline.steps.fetch import fetch_step
class _FailingFetcher:
"""Fetcher minimal qui expose une erreur externe porteuse d'un secret."""
def fetch_agenda(self) -> tuple[list[Lesson], list[SchoolEvent]]:
"""Déclenche une erreur externe pendant la récupération critique."""
raise RuntimeError("password=fetch-context-secret")
def fetch_homework(self, target_date: date) -> list[Homework]:
"""Retourne une liste vide pour compléter le protocole du fetcher."""
return []
def fetch_messages(self) -> list[Message]:
"""Retourne une liste vide pour compléter le protocole du fetcher."""
return []
def fetch_informations(self) -> list[Message]:
"""Retourne une liste vide pour compléter le protocole du fetcher."""
return []
def test_fetch_step_does_not_retain_external_exception_context() -> None:
"""Vérifie qu'une erreur critique ne conserve ni secret ni contexte externe."""
with pytest.raises(PipelineCriticalError) as exc_info:
fetch_step(_FailingFetcher())
error = exc_info.value
formatted = "".join(traceback.format_exception(error))
assert "fetch-context-secret" not in str(error)
assert "fetch-context-secret" not in formatted
assert error.__cause__ is None
assert error.__context__ is None
def test_production_raise_from_none_is_never_inside_except() -> None:
"""Vérifie structurellement que les exceptions expurgées sont levées hors des handlers."""
root = Path(__file__).parents[2] / "pronote_sync"
violations: list[str] = []
class Visitor(ast.NodeVisitor):
"""Collecte les levées ``from None`` imbriquées dans un handler."""
def __init__(self) -> None:
self._inside_except = False
def visit_ExceptHandler(self, node: ast.ExceptHandler) -> None:
"""Visite un handler en marquant son périmètre lexical."""
previous = self._inside_except
self._inside_except = True
self.generic_visit(node)
self._inside_except = previous
def visit_Raise(self, node: ast.Raise) -> None:
"""Signale une levée ``from None`` dans un handler."""
if (
self._inside_except
and isinstance(node.cause, ast.Constant)
and node.cause.value is None
):
violations.append(f"{path}:{node.lineno}")
self.generic_visit(node)
for path in sorted(root.rglob("*.py")):
Visitor().visit(ast.parse(path.read_text(encoding="utf-8")))
assert violations == []