Compare commits
10
Commits
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
5dc0907ad4 | ||
|
|
6609d5c4ca | ||
|
|
839fdd202a | ||
|
|
eef81ea323 | ||
|
|
a21acaa409 | ||
|
|
dee5fe8eff | ||
|
|
7387f9a78d | ||
|
|
894f5d137a | ||
|
|
f261fed1af | ||
|
|
e6f0659cbf |
+2
-2
@@ -156,7 +156,7 @@
|
||||
"filename": "tests/unit/test_caldav_gateway.py",
|
||||
"hashed_secret": "1c58bd92003bbaa0538e249fff6ee19a270dec5f",
|
||||
"is_verified": false,
|
||||
"line_number": 763
|
||||
"line_number": 794
|
||||
}
|
||||
],
|
||||
"tests/unit/test_caldav_security.py": [
|
||||
@@ -185,5 +185,5 @@
|
||||
}
|
||||
]
|
||||
},
|
||||
"generated_at": "2026-09-12T17:57:39Z"
|
||||
"generated_at": "2026-09-12T22:12:56Z"
|
||||
}
|
||||
|
||||
+2
-3
@@ -40,7 +40,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
|
||||
|
||||
## [0.1.0] - 2026-09-08
|
||||
|
||||
Initial release covering milestones M1 through M15.
|
||||
Initial release covering milestones M1 through M15, except the optional Gitea Actions workflow.
|
||||
|
||||
### Added
|
||||
- **M1 (Scaffolding)**: Python project structure with `pyproject.toml`, and tooling configuration for `ruff`, `mypy`, `bandit`, and `pre-commit`.
|
||||
@@ -57,5 +57,4 @@ Initial release covering milestones M1 through M15.
|
||||
- **M12 (CLI entry point)**: `pronote-sync` command with `--dry-run` and `--log-level` options, redacted error display, and safe traceback in DEBUG mode.
|
||||
- **M13 (Tests & coverage)**: 636 tests with 95.67% coverage, test fixtures (`pronote-4e.ics`, `pronote-6e.ics`), shared `conftest.py`, and secret non-leak tests.
|
||||
- **M14 (Deployment)**: systemd service and timer (daily at 18:00), logrotate configuration (daily, rotate 7, compress), `check_secrets.py` pre-deployment scanner, and exploitation guide.
|
||||
- **M15 (Documentation)**: README, README.LLM.md (AI agent setup guide), MIT LICENSE, CHANGELOG, and Gitea Actions CI/CD reference for LXC/VPS (Debian/CentOS).
|
||||
- **Other**: MIT License. Gitea Actions CI/CD reference for LXC/VPS (Debian/CentOS) is planned and optional, not delivered in this release.
|
||||
- **M15 (Documentation)**: README, README.LLM.md (AI agent setup guide), MIT LICENSE, CHANGELOG, and local validation procedures. Gitea Actions CI/CD remains optional and is not delivered in this release.
|
||||
|
||||
@@ -12,8 +12,8 @@ Synchronise l'agenda et les devoirs de **Pronote** vers un calendrier **CalDAV**
|
||||
|
||||
```bash
|
||||
# Cloner le dépôt
|
||||
git clone <repo-url>
|
||||
cd pronote-sync
|
||||
git clone https://git.antoineve.me/AntoineVe/college-infos
|
||||
cd college-infos
|
||||
|
||||
# Créer l'environnement virtuel
|
||||
python3.13 -m venv .venv
|
||||
@@ -48,6 +48,23 @@ pas garantir un état persistant cohérent pendant une simulation.
|
||||
|
||||
---
|
||||
|
||||
## Validation et CI
|
||||
|
||||
Aucun workflow Gitea Actions n'est livré actuellement. Les validations du projet sont donc
|
||||
exécutées localement avec les commandes suivantes :
|
||||
|
||||
```bash
|
||||
pytest
|
||||
ruff check .
|
||||
mypy .
|
||||
bandit -r pronote_sync/
|
||||
```
|
||||
|
||||
`pre-commit run --all-files` regroupe également les contrôles de formatage, typage, sécurité et
|
||||
détection de secrets.
|
||||
|
||||
---
|
||||
|
||||
## 🛠️ Déploiement
|
||||
|
||||
Les artefacts pour **systemd/timer** et **logrotate** sont fournis dans `deploy/`. Voir [docs/exploitation.md](docs/exploitation.md) pour plus de détails.
|
||||
|
||||
@@ -308,5 +308,5 @@ Rédiger la documentation utilisateur et finaliser le projet.
|
||||
|
||||
### Critères d'acceptation
|
||||
- `README.md` permet d'installer et de lancer le projet sans le guide.
|
||||
- Gitea Actions exécute tests + lint + sécurité.
|
||||
- Les procédures locales de test, lint et sécurité sont documentées et exécutables.
|
||||
- Aucun secret dans la documentation.
|
||||
|
||||
@@ -121,6 +121,7 @@ def fetch_step(
|
||||
:raises PronoteAuthRotationError: Si une rotation du token d'authentification
|
||||
pronotepy est nécessaire : propagée telle quelle jusqu'au pipeline.
|
||||
"""
|
||||
critical_error: PipelineCriticalError | None = None
|
||||
try:
|
||||
lessons, school_events = fetcher.fetch_agenda()
|
||||
target_date = resolve_target_date(today or date.today(), lessons, school_events)
|
||||
@@ -130,9 +131,11 @@ def fetch_step(
|
||||
except PronoteAuthRotationError:
|
||||
raise
|
||||
except Exception as exc:
|
||||
raise PipelineCriticalError(
|
||||
critical_error = PipelineCriticalError(
|
||||
f"Récupération Pronote impossible : {redact_exception(exc)}", step="fetch"
|
||||
) from None
|
||||
)
|
||||
if critical_error is not None:
|
||||
raise critical_error from None
|
||||
|
||||
messages, warnings = _fetch_optional_messages(fetcher)
|
||||
return (
|
||||
|
||||
@@ -5,7 +5,7 @@ from __future__ import annotations
|
||||
from pronote_sync.sources.blog.result import BlogRSSFetchResult
|
||||
from pronote_sync.sources.blog.rss import BlogRSSClient
|
||||
from pronote_sync.sources.blog.state import BlogRSSState
|
||||
from pronote_sync.utils.redaction import redact_exception
|
||||
from pronote_sync.utils.redaction import redact_exception, redact_secrets
|
||||
|
||||
|
||||
def fetch_blog_step(client: BlogRSSClient | None, state: BlogRSSState | None) -> BlogRSSFetchResult:
|
||||
@@ -24,15 +24,18 @@ def fetch_blog_step(client: BlogRSSClient | None, state: BlogRSSState | None) ->
|
||||
"""
|
||||
if client is None or state is None:
|
||||
return BlogRSSFetchResult()
|
||||
error_message: str | None = None
|
||||
try:
|
||||
etag, last_modified = state.get_cache_headers()
|
||||
result = client.fetch_and_parse(
|
||||
known_guids=state.get_known_guids(), etag=etag, last_modified=last_modified
|
||||
)
|
||||
if result.error is not None:
|
||||
raise RuntimeError(result.error) from None
|
||||
if not result.not_modified and not result.articles:
|
||||
error_message = f"Récupération du blog échouée : {redact_secrets(result.error)}"
|
||||
elif not result.not_modified and not result.articles:
|
||||
state.update_cache_headers(result.etag, result.last_modified)
|
||||
return result
|
||||
except Exception as exc:
|
||||
raise RuntimeError(f"Récupération du blog échouée : {redact_exception(exc)}") from None
|
||||
error_message = f"Récupération du blog échouée : {redact_exception(exc)}"
|
||||
if error_message is not None:
|
||||
raise RuntimeError(error_message) from None
|
||||
return result
|
||||
|
||||
@@ -97,7 +97,10 @@ class BlogRSSState:
|
||||
redact_exception(exc),
|
||||
)
|
||||
|
||||
def _save(self) -> None:
|
||||
def _save(
|
||||
self,
|
||||
state: tuple[set[str], str | None, str | None] | None = None,
|
||||
) -> bool:
|
||||
"""Sauvegarde l'état dans le fichier JSON de manière atomique.
|
||||
|
||||
La sortie est déterministe : ``known_guids`` est trié
|
||||
@@ -107,20 +110,30 @@ class BlogRSSState:
|
||||
jamais laisser un fichier partiel en cas d'interruption. En cas
|
||||
d'erreur d'écriture, une erreur est journalisée sans être
|
||||
propagée et le fichier temporaire est supprimé.
|
||||
|
||||
:param state: État à sauvegarder ; l'état courant est utilisé par défaut.
|
||||
:return: ``True`` si l'état a été sauvegardé ou si la persistance est désactivée.
|
||||
:rtype: bool
|
||||
"""
|
||||
if not self._persistence_enabled:
|
||||
return
|
||||
return True
|
||||
known_guids, etag, last_modified = state or (
|
||||
self._known_guids,
|
||||
self._etag,
|
||||
self._last_modified,
|
||||
)
|
||||
payload = {
|
||||
"version": _STATE_VERSION,
|
||||
"known_guids": sorted(self._known_guids),
|
||||
"etag": self._etag,
|
||||
"last_modified": self._last_modified,
|
||||
"known_guids": sorted(known_guids),
|
||||
"etag": etag,
|
||||
"last_modified": last_modified,
|
||||
}
|
||||
tmp_file = self._state_file.with_suffix(".tmp")
|
||||
try:
|
||||
with open(tmp_file, "w", encoding="utf-8") as handle:
|
||||
json.dump(payload, handle, indent=2)
|
||||
tmp_file.replace(self._state_file)
|
||||
return True
|
||||
except Exception as exc:
|
||||
logger.error(
|
||||
"Impossible d'écrire le fichier d'état blog RSS %s : %s.",
|
||||
@@ -134,6 +147,7 @@ class BlogRSSState:
|
||||
"Nettoyage du fichier temporaire échoué : %s",
|
||||
redact_exception(cleanup_exc),
|
||||
)
|
||||
return False
|
||||
|
||||
def get_known_guids(self) -> frozenset[str]:
|
||||
"""Renvoie une copie immuable des GUID d'articles déjà connus.
|
||||
@@ -168,10 +182,13 @@ class BlogRSSState:
|
||||
"""
|
||||
if result.not_modified:
|
||||
return
|
||||
self._known_guids.update(article.id for article in result.articles)
|
||||
self._etag = result.etag
|
||||
self._last_modified = result.last_modified
|
||||
self._save()
|
||||
new_state = (
|
||||
self._known_guids | {article.id for article in result.articles},
|
||||
result.etag,
|
||||
result.last_modified,
|
||||
)
|
||||
if self._save(new_state):
|
||||
self._known_guids, self._etag, self._last_modified = new_state
|
||||
|
||||
def get_cache_headers(self) -> tuple[str | None, str | None]:
|
||||
"""Renvoie les en-têtes de cache HTTP mémorisés.
|
||||
|
||||
@@ -218,6 +218,7 @@ class PronoteAuthState:
|
||||
}
|
||||
tmp_file = self._state_file.with_suffix(".tmp")
|
||||
fd: int | None = None
|
||||
write_error: PronoteSyncError | None = None
|
||||
try:
|
||||
# Nettoie un éventuel fichier temporaire stale laissé par une exécution interrompue.
|
||||
if tmp_file.exists():
|
||||
@@ -259,10 +260,12 @@ class PronoteAuthState:
|
||||
"Nettoyage du fichier temporaire d'état d'authentification Pronote échoué : %s",
|
||||
redact_exception(cleanup_exc),
|
||||
)
|
||||
raise PronoteSyncError(
|
||||
write_error = PronoteSyncError(
|
||||
f"Impossible d'écrire le fichier d'état d'authentification Pronote "
|
||||
f"{redact_secrets(str(self._state_file))}."
|
||||
) from None
|
||||
)
|
||||
if write_error is not None:
|
||||
raise write_error from None
|
||||
|
||||
def clear(self) -> None:
|
||||
"""Supprime le fichier d'état d'authentification.
|
||||
|
||||
@@ -269,34 +269,36 @@ class PronoteFetcher:
|
||||
primary,
|
||||
redact_exception(exc),
|
||||
)
|
||||
if fallback is None:
|
||||
raise PipelineCriticalError(
|
||||
f"Impossible de récupérer l'agenda : la source {primary} a échoué"
|
||||
) from None
|
||||
logger.info("Repli sur %s pour l'agenda.", fallback)
|
||||
try:
|
||||
lessons, school_events = self._fetch_agenda_source(fallback)
|
||||
except PronoteAuthRotationError:
|
||||
raise
|
||||
except Exception as exc:
|
||||
logger.error(
|
||||
"Échec de la récupération %s pour l'agenda : %s",
|
||||
fallback,
|
||||
redact_exception(exc),
|
||||
)
|
||||
raise PipelineCriticalError(
|
||||
f"Impossible de récupérer l'agenda : les sources {primary}"
|
||||
f" et {fallback} ont échoué"
|
||||
) from None
|
||||
if not lessons:
|
||||
logger.warning(
|
||||
"Le repli %s pour l'agenda a retourné un résultat vide après l'échec "
|
||||
"de %s : impossible de distinguer une absence de cours d'un échec "
|
||||
"silencieux.",
|
||||
fallback,
|
||||
primary,
|
||||
)
|
||||
return lessons, school_events
|
||||
if fallback is None:
|
||||
raise PipelineCriticalError(
|
||||
f"Impossible de récupérer l'agenda : la source {primary} a échoué"
|
||||
) from None
|
||||
logger.info("Repli sur %s pour l'agenda.", fallback)
|
||||
fallback_result: tuple[list[Lesson], list[SchoolEvent]] | None = None
|
||||
try:
|
||||
fallback_result = self._fetch_agenda_source(fallback)
|
||||
except PronoteAuthRotationError:
|
||||
raise
|
||||
except Exception as exc:
|
||||
logger.error(
|
||||
"Échec de la récupération %s pour l'agenda : %s",
|
||||
fallback,
|
||||
redact_exception(exc),
|
||||
)
|
||||
if fallback_result is None:
|
||||
raise PipelineCriticalError(
|
||||
f"Impossible de récupérer l'agenda : les sources {primary} et {fallback} ont échoué"
|
||||
) from None
|
||||
lessons, school_events = fallback_result
|
||||
if not lessons:
|
||||
logger.warning(
|
||||
"Le repli %s pour l'agenda a retourné un résultat vide après l'échec "
|
||||
"de %s : impossible de distinguer une absence de cours d'un échec "
|
||||
"silencieux.",
|
||||
fallback,
|
||||
primary,
|
||||
)
|
||||
return lessons, school_events
|
||||
|
||||
def _fetch_homework_ical(self, target_date: date) -> list[Homework]:
|
||||
"""Récupère les devoirs depuis le flux iCal pour la date cible.
|
||||
@@ -396,34 +398,37 @@ class PronoteFetcher:
|
||||
primary,
|
||||
redact_exception(exc),
|
||||
)
|
||||
if fallback is None:
|
||||
raise PipelineCriticalError(
|
||||
f"Impossible de récupérer les devoirs : la source {primary} a échoué"
|
||||
) from None
|
||||
logger.info("Repli sur %s pour les devoirs.", fallback)
|
||||
try:
|
||||
homeworks = self._fetch_homework_source(fallback, target_date)
|
||||
except PronoteAuthRotationError:
|
||||
raise
|
||||
except Exception as exc:
|
||||
logger.error(
|
||||
"Échec de la récupération %s pour les devoirs : %s",
|
||||
fallback,
|
||||
redact_exception(exc),
|
||||
)
|
||||
raise PipelineCriticalError(
|
||||
f"Impossible de récupérer les devoirs : les sources {primary}"
|
||||
f" et {fallback} ont échoué"
|
||||
) from None
|
||||
if not homeworks:
|
||||
logger.warning(
|
||||
"Le repli %s pour les devoirs a retourné un résultat vide après "
|
||||
"l'échec de %s : impossible de distinguer une absence de devoirs "
|
||||
"d'un échec silencieux.",
|
||||
fallback,
|
||||
primary,
|
||||
)
|
||||
return homeworks
|
||||
if fallback is None:
|
||||
raise PipelineCriticalError(
|
||||
f"Impossible de récupérer les devoirs : la source {primary} a échoué"
|
||||
) from None
|
||||
logger.info("Repli sur %s pour les devoirs.", fallback)
|
||||
fallback_result: list[Homework] | None = None
|
||||
try:
|
||||
fallback_result = self._fetch_homework_source(fallback, target_date)
|
||||
except PronoteAuthRotationError:
|
||||
raise
|
||||
except Exception as exc:
|
||||
logger.error(
|
||||
"Échec de la récupération %s pour les devoirs : %s",
|
||||
fallback,
|
||||
redact_exception(exc),
|
||||
)
|
||||
if fallback_result is None:
|
||||
raise PipelineCriticalError(
|
||||
f"Impossible de récupérer les devoirs : les sources {primary}"
|
||||
f" et {fallback} ont échoué"
|
||||
) from None
|
||||
homeworks = fallback_result
|
||||
if not homeworks:
|
||||
logger.warning(
|
||||
"Le repli %s pour les devoirs a retourné un résultat vide après "
|
||||
"l'échec de %s : impossible de distinguer une absence de devoirs "
|
||||
"d'un échec silencieux.",
|
||||
fallback,
|
||||
primary,
|
||||
)
|
||||
return homeworks
|
||||
|
||||
def fetch_messages(self) -> list[Message]:
|
||||
"""Récupère les messages des discussions Pronote (toujours via pronotepy).
|
||||
|
||||
@@ -85,6 +85,7 @@ class JsonTheoreticalAgendaProvider:
|
||||
self._file_path: str = file_path
|
||||
self._parity_service: WeekParityService | None = parity_service
|
||||
self._holiday_calendar: SchoolHolidayCalendar | None = holiday_calendar
|
||||
load_error: PronoteSyncError | None = None
|
||||
try:
|
||||
content = Path(file_path).read_text(encoding="utf-8")
|
||||
parsed = TheoreticalAgendaFile.model_validate_json(content)
|
||||
@@ -94,9 +95,11 @@ class JsonTheoreticalAgendaProvider:
|
||||
redact_secrets(str(file_path)),
|
||||
redact_exception(exc),
|
||||
)
|
||||
raise PronoteSyncError(
|
||||
load_error = PronoteSyncError(
|
||||
f"Le fichier d'agenda théorique est invalide : {redact_secrets(str(file_path))}"
|
||||
) from None
|
||||
)
|
||||
if load_error is not None:
|
||||
raise load_error from None
|
||||
self._lessons: tuple[TheoreticalLessonEntry, ...] = parsed.lessons
|
||||
if self._parity_service is None and any(
|
||||
entry.week in ("even", "odd") for entry in self._lessons
|
||||
|
||||
@@ -77,6 +77,7 @@ class SchoolHolidayCalendar:
|
||||
raise PronoteSyncError(
|
||||
f"Le fichier de vacances scolaires est introuvable : {redact_secrets(str(path))}"
|
||||
) from None
|
||||
load_error: PronoteSyncError | None = None
|
||||
try:
|
||||
data: Any = json.loads(path.read_text(encoding="utf-8"))
|
||||
file_model: SchoolHolidayFile = SchoolHolidayFile.model_validate(data)
|
||||
@@ -86,9 +87,11 @@ class SchoolHolidayCalendar:
|
||||
redact_secrets(str(path)),
|
||||
redact_exception(exc),
|
||||
)
|
||||
raise PronoteSyncError(
|
||||
load_error = PronoteSyncError(
|
||||
f"Le fichier de vacances scolaires est invalide : {redact_secrets(str(path))}"
|
||||
) from None
|
||||
)
|
||||
if load_error is not None:
|
||||
raise load_error from None
|
||||
self._periods = file_model.periods
|
||||
|
||||
def is_holiday(self, target_date: date) -> bool:
|
||||
|
||||
@@ -191,7 +191,11 @@ class CalDAVGateway:
|
||||
for vevent in component.walk("VEVENT"):
|
||||
managed = vevent.get(MANAGED_PROPERTY)
|
||||
if managed is not None and str(managed) == MANAGED_VALUE:
|
||||
raw_uid = str(vevent.get("UID"))
|
||||
raw_uid_value = vevent.get("UID")
|
||||
if raw_uid_value is None or not str(raw_uid_value).strip():
|
||||
logger.warning("Événement CalDAV géré sans UID ignoré.")
|
||||
continue
|
||||
raw_uid = str(raw_uid_value)
|
||||
canonical_uid = normalize_pronote_uid(raw_uid)
|
||||
result.append((raw_uid, canonical_uid, vevent))
|
||||
except Exception as exc:
|
||||
|
||||
+6
-5
@@ -7,9 +7,10 @@ name = "pronote-sync"
|
||||
version = "0.1.2"
|
||||
description = "Synchronisation Pronote → CalDAV + XMPP"
|
||||
license = {text = "MIT"}
|
||||
readme = "README.md"
|
||||
requires-python = ">=3.13.5"
|
||||
authors = [
|
||||
{name = "Votre Nom", email = "votre@email.com"}
|
||||
{name = "Antoine Van Elstraete", email = "antoine@van-elstraete.net"}
|
||||
]
|
||||
keywords = ["pronote", "caldav", "xmpp", "sync", "school"]
|
||||
classifiers = [
|
||||
@@ -57,10 +58,10 @@ dev = [
|
||||
pronote-sync = "pronote_sync.cli.main:main"
|
||||
|
||||
[project.urls]
|
||||
Homepage = "https://github.com/votre-utilisateur/pronote-sync"
|
||||
Documentation = "https://github.com/votre-utilisateur/pronote-sync#readme"
|
||||
Repository = "https://github.com/votre-utilisateur/pronote-sync"
|
||||
Issues = "https://github.com/votre-utilisateur/pronote-sync/issues"
|
||||
Homepage = "https://git.antoineve.me/AntoineVe/college-infos"
|
||||
Documentation = "https://git.antoineve.me/AntoineVe/college-infos/wiki"
|
||||
Repository = "https://git.antoineve.me/AntoineVe/college-infos"
|
||||
Issues = "https://git.antoineve.me/AntoineVe/college-infos/issues"
|
||||
|
||||
[tool.setuptools.packages.find]
|
||||
where = ["."]
|
||||
|
||||
@@ -26,15 +26,15 @@ _TEXT_SUFFIXES = frozenset(
|
||||
{".conf", ".ini", ".json", ".md", ".py", ".service", ".timer", ".toml", ".txt", ".yaml", ".yml"}
|
||||
)
|
||||
_LITERAL_SECRET_RE = re.compile(
|
||||
r"(?ix)\b[a-z0-9_]*(?:api[_-]?key|access[_-]?token|auth(?:orization)?|icalsecurise|password|secret|token)"
|
||||
r"\s*[:=]\s*['\"][^'\"\r\n]{3,}['\"]"
|
||||
r"(?ix)\b[a-z0-9_]*(?:api[_-]?key|access[_-]?token|auth(?:orization)?|icalsecurise|password|pin|secret|token)"
|
||||
r"\s*[:=]\s*['\"](?P<value>[^'\"\r\n]{3,})['\"]"
|
||||
)
|
||||
_UNQUOTED_SECRET_RE = re.compile(
|
||||
r"(?ix)\b[a-z0-9_]*(?:api[_-]?key|access[_-]?token|auth(?:orization)?|icalsecurise|password|secret|token)"
|
||||
r"\s*[:=]\s*[a-z0-9][a-z0-9._~+/-]{2,}"
|
||||
r"(?ix)\b[a-z0-9_]*(?:api[_-]?key|access[_-]?token|auth(?:orization)?|icalsecurise|password|pin|secret|token)"
|
||||
r"\s*[:=]\s*(?P<value>[a-z0-9][a-z0-9._~+/-]{2,})"
|
||||
)
|
||||
_URL_SECRET_RE = re.compile(
|
||||
r"(?ix)[?&](?:api[_-]?key|access[_-]?token|auth(?:orization)?|icalsecurise|password|secret|token)"
|
||||
r"(?ix)[?&](?:api[_-]?key|access[_-]?token|auth(?:orization)?|icalsecurise|password|pin|secret|token)"
|
||||
r"=([^&#\s]{3,})"
|
||||
)
|
||||
_URL_PLACEHOLDER_RE = re.compile(
|
||||
@@ -47,6 +47,13 @@ _URL_PLACEHOLDER_RE = re.compile(
|
||||
r")$"
|
||||
)
|
||||
_EXTRA_NAMES = frozenset({"pronote_sync"})
|
||||
_ASSIGNMENT_PLACEHOLDER_RE = re.compile(
|
||||
r"(?ix)^(?:"
|
||||
r"<(?:pin|secret|valeur|value|token|jeton)>|"
|
||||
r"(?:change|replace|your)[_-]?(?:me|here|value|valeur|pin|password|secret)|"
|
||||
r"(?:placeholder|example|local-not-required)"
|
||||
r")$"
|
||||
)
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
@@ -68,6 +75,16 @@ CommandRunner = Callable[..., subprocess.CompletedProcess[str]]
|
||||
ContentProvider = Callable[[Path], str | None]
|
||||
|
||||
|
||||
def _is_assignment_placeholder(value: str) -> bool:
|
||||
"""Indique si une valeur d'affectation est un placeholder documentaire.
|
||||
|
||||
:param value: Valeur extraite d'une affectation sensible.
|
||||
:return: ``True`` si la valeur ne représente pas un secret réel.
|
||||
:rtype: bool
|
||||
"""
|
||||
return _ASSIGNMENT_PLACEHOLDER_RE.fullmatch(value.strip()) is not None
|
||||
|
||||
|
||||
def _is_candidate(path: Path) -> bool:
|
||||
"""Indique si un chemin peut être analysé comme fichier texte.
|
||||
|
||||
@@ -188,9 +205,15 @@ def find_secrets(
|
||||
for number, line in enumerate(content.splitlines(), start=1):
|
||||
if _ALLOWLIST_MARKER in line:
|
||||
continue
|
||||
is_literal_secret = _LITERAL_SECRET_RE.search(line) or (
|
||||
relative_path.suffix in _UNQUOTED_CONFIG_SUFFIXES
|
||||
and _UNQUOTED_SECRET_RE.search(line)
|
||||
literal_match = _LITERAL_SECRET_RE.search(line)
|
||||
unquoted_match = (
|
||||
_UNQUOTED_SECRET_RE.search(line)
|
||||
if relative_path.suffix in _UNQUOTED_CONFIG_SUFFIXES
|
||||
else None
|
||||
)
|
||||
is_literal_secret = any(
|
||||
match is not None and not _is_assignment_placeholder(match.group("value"))
|
||||
for match in (literal_match, unquoted_match)
|
||||
)
|
||||
if is_literal_secret:
|
||||
findings.append(SecretFinding(relative_path, number, "affectation-litterale"))
|
||||
|
||||
@@ -425,4 +425,41 @@ def test_atomic_save_preserves_on_error(tmp_path: Path) -> None:
|
||||
assert state.get_known_guids() == frozenset({"original-guid-1", "original-guid-2", "new-guid"})
|
||||
|
||||
|
||||
def test_acknowledge_does_not_advance_memory_when_save_fails(
|
||||
tmp_path: Path,
|
||||
) -> None:
|
||||
"""Conserve l'état précédent en mémoire si l'acquittement ne peut pas être sauvegardé.
|
||||
|
||||
:param tmp_path: Fixture pytest pour un répertoire temporaire.
|
||||
:return: None
|
||||
"""
|
||||
state_file = tmp_path / "state.json"
|
||||
state = BlogRSSState(state_file)
|
||||
state.add_guids(["existing-guid"])
|
||||
state.update_cache_headers("old-etag", "old-last-modified")
|
||||
article = BlogArticle(
|
||||
id="new-guid",
|
||||
title="Article",
|
||||
url="https://example.com/article",
|
||||
published_at=datetime(2026, 9, 12, 8, 0, tzinfo=UTC),
|
||||
updated_at=None,
|
||||
category=None,
|
||||
author=None,
|
||||
content_html="<p>Contenu</p>",
|
||||
content_text="Contenu",
|
||||
)
|
||||
|
||||
with patch.object(Path, "replace", side_effect=OSError("replace failed")):
|
||||
state.acknowledge(
|
||||
BlogRSSFetchResult(
|
||||
articles=(article,),
|
||||
etag="new-etag",
|
||||
last_modified="new-last-modified",
|
||||
)
|
||||
)
|
||||
|
||||
assert state.get_known_guids() == frozenset({"existing-guid"})
|
||||
assert state.get_cache_headers() == ("old-etag", "old-last-modified")
|
||||
|
||||
|
||||
# Ensure trailing newline
|
||||
|
||||
@@ -327,6 +327,37 @@ def test_list_managed_events_returns_only_managed(
|
||||
assert str(vevent.get("UID")) == "test-uid-123"
|
||||
|
||||
|
||||
def test_list_managed_events_ignores_managed_event_without_uid(
|
||||
caldav_settings: CalDAVSettings,
|
||||
mock_client_factory: MagicMock,
|
||||
caplog: LogCaptureFixture,
|
||||
) -> None:
|
||||
"""Ignore un événement géré sans UID et ne le transmet pas au planificateur.
|
||||
|
||||
:param caldav_settings: Paramètres CalDAV valides.
|
||||
:param mock_client_factory: Usine de clients CalDAV mockée.
|
||||
:param caplog: Capture des journaux de diagnostic.
|
||||
:return: None
|
||||
"""
|
||||
gateway = CalDAVGateway(caldav_settings, client_factory=mock_client_factory)
|
||||
gateway.connect()
|
||||
|
||||
malformed_event = Event()
|
||||
malformed_event.add("SUMMARY", "Événement sans identifiant")
|
||||
malformed_event.add(MANAGED_PROPERTY, MANAGED_VALUE)
|
||||
remote_event = MagicMock()
|
||||
remote_event.icalendar_component = Calendar()
|
||||
remote_event.icalendar_component.add_component(malformed_event)
|
||||
calendar = mock_client_factory.return_value.principal.return_value.calendars.return_value[0]
|
||||
calendar.search.return_value = [remote_event]
|
||||
|
||||
with caplog.at_level(logging.WARNING):
|
||||
result = gateway.list_managed_events(datetime(2026, 1, 1), datetime(2026, 12, 31))
|
||||
|
||||
assert result == []
|
||||
assert "sans UID ignoré" in caplog.text
|
||||
|
||||
|
||||
def test_list_managed_events_not_connected_raises(
|
||||
caldav_settings: CalDAVSettings,
|
||||
) -> None:
|
||||
|
||||
@@ -118,6 +118,27 @@ def test_main_detects_sensitive_url_parameter(
|
||||
assert sentinel not in output
|
||||
|
||||
|
||||
def test_main_detects_pin_url_parameter_without_disclosing_its_value(
|
||||
secret_checker: ModuleType, tmp_path: Path, capsys: CaptureFixture[str]
|
||||
) -> None:
|
||||
"""Vérifie qu'un PIN dans une query string déclenche un échec sans fuite.
|
||||
|
||||
:param secret_checker: Module du script sous test.
|
||||
:param tmp_path: Répertoire temporaire représentant un dépôt.
|
||||
:param capsys: Fixture de capture de sortie.
|
||||
:return: None
|
||||
"""
|
||||
sentinel = "m14-url-pin-sentinel"
|
||||
(tmp_path / "settings.yaml").write_text(
|
||||
f"url: https://example.invalid/api?pin={sentinel}\n", encoding="utf-8"
|
||||
) # secret-check: allow
|
||||
|
||||
assert secret_checker.main([], root=tmp_path) == 1
|
||||
output = capsys.readouterr().out
|
||||
assert "settings.yaml:1 (parametre-url)" in output
|
||||
assert sentinel not in output
|
||||
|
||||
|
||||
def test_main_ignores_documentation_url_placeholders(
|
||||
secret_checker: ModuleType, tmp_path: Path, capsys: CaptureFixture[str]
|
||||
) -> None:
|
||||
@@ -236,6 +257,58 @@ def test_main_detects_prefixed_secret_assignment(
|
||||
assert sentinel not in output
|
||||
|
||||
|
||||
def test_main_detects_pronote_pin_assignments_without_disclosing_value(
|
||||
secret_checker: ModuleType, tmp_path: Path, capsys: CaptureFixture[str]
|
||||
) -> None:
|
||||
"""Détecte les PIN Pronote littéraux et non quotés sans afficher leur valeur.
|
||||
|
||||
:param secret_checker: Module du script sous test.
|
||||
:param tmp_path: Répertoire temporaire représentant un dépôt.
|
||||
:param capsys: Fixture de capture de sortie.
|
||||
:return: None
|
||||
"""
|
||||
literal_pin = "pin-literal-sentinel"
|
||||
unquoted_pin = "pin-unquoted-sentinel"
|
||||
(tmp_path / "settings.py").write_text(f'PRONOTE_QR_PIN = "{literal_pin}"\n', encoding="utf-8")
|
||||
(tmp_path / "settings.yaml").write_text(
|
||||
f"PRONOTE_ACCOUNT_PIN: {unquoted_pin}\n", encoding="utf-8"
|
||||
)
|
||||
|
||||
assert secret_checker.main([], root=tmp_path) == 1
|
||||
output = capsys.readouterr().out
|
||||
assert "settings.py:1" in output
|
||||
assert "settings.yaml:1" in output
|
||||
assert literal_pin not in output
|
||||
assert unquoted_pin not in output
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
"line",
|
||||
[
|
||||
'PRONOTE_QR_PIN = "<valeur>"',
|
||||
"# PRONOTE_ACCOUNT_PIN doit rester dans le fichier d'environnement local",
|
||||
],
|
||||
)
|
||||
def test_main_ignores_pronote_pin_placeholders_and_descriptions(
|
||||
secret_checker: ModuleType,
|
||||
tmp_path: Path,
|
||||
capsys: CaptureFixture[str],
|
||||
line: str,
|
||||
) -> None:
|
||||
"""Ignore les placeholders et descriptions de PIN sans affectation réelle.
|
||||
|
||||
:param secret_checker: Module du script sous test.
|
||||
:param tmp_path: Répertoire temporaire représentant un dépôt.
|
||||
:param capsys: Fixture de capture de sortie.
|
||||
:param line: Ligne documentaire à analyser.
|
||||
:return: None
|
||||
"""
|
||||
(tmp_path / "guide.py").write_text(line + "\n", encoding="utf-8")
|
||||
|
||||
assert secret_checker.main([], root=tmp_path) == 0
|
||||
assert "OK:" in capsys.readouterr().out
|
||||
|
||||
|
||||
def test_main_detects_short_secret_assignment(
|
||||
secret_checker: ModuleType, tmp_path: Path, capsys: CaptureFixture[str]
|
||||
) -> None:
|
||||
|
||||
@@ -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