Compare commits
8
Commits
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
97a14d9543 | ||
|
|
5dc0907ad4 | ||
|
|
c8c282bf75 | ||
|
|
6609d5c4ca | ||
|
|
839fdd202a | ||
|
|
dee5fe8eff | ||
|
|
7387f9a78d | ||
|
|
894f5d137a |
+1
-2
@@ -94,8 +94,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
@@ -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"
|
||||||
}
|
}
|
||||||
|
|||||||
+7
-4
@@ -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
|
||||||
```
|
```
|
||||||
|
|
||||||
---
|
---
|
||||||
|
|||||||
@@ -352,60 +352,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):
|
||||||
@@ -466,6 +416,5 @@ class Settings(BaseSettings):
|
|||||||
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))
|
||||||
|
|||||||
@@ -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
|
||||||
)
|
)
|
||||||
|
|||||||
@@ -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 (
|
||||||
|
|||||||
@@ -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
|
||||||
|
|||||||
@@ -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.
|
||||||
|
|||||||
@@ -274,8 +274,9 @@ class PronoteFetcher:
|
|||||||
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)
|
||||||
|
fallback_result: tuple[list[Lesson], list[SchoolEvent]] | None = None
|
||||||
try:
|
try:
|
||||||
lessons, school_events = self._fetch_agenda_source(fallback)
|
fallback_result = self._fetch_agenda_source(fallback)
|
||||||
except PronoteAuthRotationError:
|
except PronoteAuthRotationError:
|
||||||
raise
|
raise
|
||||||
except Exception as exc:
|
except Exception as exc:
|
||||||
@@ -284,10 +285,11 @@ class PronoteFetcher:
|
|||||||
fallback,
|
fallback,
|
||||||
redact_exception(exc),
|
redact_exception(exc),
|
||||||
)
|
)
|
||||||
|
if fallback_result is None:
|
||||||
raise PipelineCriticalError(
|
raise PipelineCriticalError(
|
||||||
f"Impossible de récupérer l'agenda : les sources {primary}"
|
f"Impossible de récupérer l'agenda : les sources {primary} et {fallback} ont échoué"
|
||||||
f" et {fallback} ont échoué"
|
|
||||||
) from None
|
) from None
|
||||||
|
lessons, school_events = fallback_result
|
||||||
if not lessons:
|
if not lessons:
|
||||||
logger.warning(
|
logger.warning(
|
||||||
"Le repli %s pour l'agenda a retourné un résultat vide après l'échec "
|
"Le repli %s pour l'agenda a retourné un résultat vide après l'échec "
|
||||||
@@ -401,8 +403,9 @@ class PronoteFetcher:
|
|||||||
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)
|
||||||
|
fallback_result: list[Homework] | None = None
|
||||||
try:
|
try:
|
||||||
homeworks = self._fetch_homework_source(fallback, target_date)
|
fallback_result = self._fetch_homework_source(fallback, target_date)
|
||||||
except PronoteAuthRotationError:
|
except PronoteAuthRotationError:
|
||||||
raise
|
raise
|
||||||
except Exception as exc:
|
except Exception as exc:
|
||||||
@@ -411,10 +414,12 @@ class PronoteFetcher:
|
|||||||
fallback,
|
fallback,
|
||||||
redact_exception(exc),
|
redact_exception(exc),
|
||||||
)
|
)
|
||||||
|
if fallback_result is None:
|
||||||
raise PipelineCriticalError(
|
raise PipelineCriticalError(
|
||||||
f"Impossible de récupérer les devoirs : les sources {primary}"
|
f"Impossible de récupérer les devoirs : les sources {primary}"
|
||||||
f" et {fallback} ont échoué"
|
f" et {fallback} ont échoué"
|
||||||
) from None
|
) from None
|
||||||
|
homeworks = fallback_result
|
||||||
if not homeworks:
|
if not homeworks:
|
||||||
logger.warning(
|
logger.warning(
|
||||||
"Le repli %s pour les devoirs a retourné un résultat vide après "
|
"Le repli %s pour les devoirs a retourné un résultat vide après "
|
||||||
|
|||||||
@@ -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
|
||||||
|
|||||||
@@ -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:
|
||||||
|
|||||||
@@ -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"))
|
||||||
|
|||||||
@@ -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:
|
||||||
|
|||||||
@@ -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 == []
|
||||||
Reference in New Issue
Block a user