Compare commits
4
Commits
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
7387f9a78d | ||
|
|
894f5d137a | ||
|
|
f261fed1af | ||
|
|
e6f0659cbf |
+2
-2
@@ -156,7 +156,7 @@
|
|||||||
"filename": "tests/unit/test_caldav_gateway.py",
|
"filename": "tests/unit/test_caldav_gateway.py",
|
||||||
"hashed_secret": "1c58bd92003bbaa0538e249fff6ee19a270dec5f",
|
"hashed_secret": "1c58bd92003bbaa0538e249fff6ee19a270dec5f",
|
||||||
"is_verified": false,
|
"is_verified": false,
|
||||||
"line_number": 763
|
"line_number": 794
|
||||||
}
|
}
|
||||||
],
|
],
|
||||||
"tests/unit/test_caldav_security.py": [
|
"tests/unit/test_caldav_security.py": [
|
||||||
@@ -185,5 +185,5 @@
|
|||||||
}
|
}
|
||||||
]
|
]
|
||||||
},
|
},
|
||||||
"generated_at": "2026-09-12T17:57:39Z"
|
"generated_at": "2026-09-12T22:12:56Z"
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -97,7 +97,10 @@ class BlogRSSState:
|
|||||||
redact_exception(exc),
|
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.
|
"""Sauvegarde l'état dans le fichier JSON de manière atomique.
|
||||||
|
|
||||||
La sortie est déterministe : ``known_guids`` est trié
|
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
|
jamais laisser un fichier partiel en cas d'interruption. En cas
|
||||||
d'erreur d'écriture, une erreur est journalisée sans être
|
d'erreur d'écriture, une erreur est journalisée sans être
|
||||||
propagée et le fichier temporaire est supprimé.
|
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:
|
if not self._persistence_enabled:
|
||||||
return
|
return True
|
||||||
|
known_guids, etag, last_modified = state or (
|
||||||
|
self._known_guids,
|
||||||
|
self._etag,
|
||||||
|
self._last_modified,
|
||||||
|
)
|
||||||
payload = {
|
payload = {
|
||||||
"version": _STATE_VERSION,
|
"version": _STATE_VERSION,
|
||||||
"known_guids": sorted(self._known_guids),
|
"known_guids": sorted(known_guids),
|
||||||
"etag": self._etag,
|
"etag": etag,
|
||||||
"last_modified": self._last_modified,
|
"last_modified": last_modified,
|
||||||
}
|
}
|
||||||
tmp_file = self._state_file.with_suffix(".tmp")
|
tmp_file = self._state_file.with_suffix(".tmp")
|
||||||
try:
|
try:
|
||||||
with open(tmp_file, "w", encoding="utf-8") as handle:
|
with open(tmp_file, "w", encoding="utf-8") as handle:
|
||||||
json.dump(payload, handle, indent=2)
|
json.dump(payload, handle, indent=2)
|
||||||
tmp_file.replace(self._state_file)
|
tmp_file.replace(self._state_file)
|
||||||
|
return True
|
||||||
except Exception as exc:
|
except Exception as exc:
|
||||||
logger.error(
|
logger.error(
|
||||||
"Impossible d'écrire le fichier d'état blog RSS %s : %s.",
|
"Impossible d'écrire le fichier d'état blog RSS %s : %s.",
|
||||||
@@ -134,6 +147,7 @@ class BlogRSSState:
|
|||||||
"Nettoyage du fichier temporaire échoué : %s",
|
"Nettoyage du fichier temporaire échoué : %s",
|
||||||
redact_exception(cleanup_exc),
|
redact_exception(cleanup_exc),
|
||||||
)
|
)
|
||||||
|
return False
|
||||||
|
|
||||||
def get_known_guids(self) -> frozenset[str]:
|
def get_known_guids(self) -> frozenset[str]:
|
||||||
"""Renvoie une copie immuable des GUID d'articles déjà connus.
|
"""Renvoie une copie immuable des GUID d'articles déjà connus.
|
||||||
@@ -168,10 +182,13 @@ class BlogRSSState:
|
|||||||
"""
|
"""
|
||||||
if result.not_modified:
|
if result.not_modified:
|
||||||
return
|
return
|
||||||
self._known_guids.update(article.id for article in result.articles)
|
new_state = (
|
||||||
self._etag = result.etag
|
self._known_guids | {article.id for article in result.articles},
|
||||||
self._last_modified = result.last_modified
|
result.etag,
|
||||||
self._save()
|
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]:
|
def get_cache_headers(self) -> tuple[str | None, str | None]:
|
||||||
"""Renvoie les en-têtes de cache HTTP mémorisés.
|
"""Renvoie les en-têtes de cache HTTP mémorisés.
|
||||||
|
|||||||
@@ -191,7 +191,11 @@ class CalDAVGateway:
|
|||||||
for vevent in component.walk("VEVENT"):
|
for vevent in component.walk("VEVENT"):
|
||||||
managed = vevent.get(MANAGED_PROPERTY)
|
managed = vevent.get(MANAGED_PROPERTY)
|
||||||
if managed is not None and str(managed) == MANAGED_VALUE:
|
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)
|
canonical_uid = normalize_pronote_uid(raw_uid)
|
||||||
result.append((raw_uid, canonical_uid, vevent))
|
result.append((raw_uid, canonical_uid, vevent))
|
||||||
except Exception as exc:
|
except Exception as exc:
|
||||||
|
|||||||
@@ -26,12 +26,12 @@ _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|secret|token)"
|
||||||
@@ -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"))
|
||||||
|
|||||||
@@ -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"})
|
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
|
# Ensure trailing newline
|
||||||
|
|||||||
@@ -327,6 +327,37 @@ def test_list_managed_events_returns_only_managed(
|
|||||||
assert str(vevent.get("UID")) == "test-uid-123"
|
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(
|
def test_list_managed_events_not_connected_raises(
|
||||||
caldav_settings: CalDAVSettings,
|
caldav_settings: CalDAVSettings,
|
||||||
) -> None:
|
) -> None:
|
||||||
|
|||||||
@@ -236,6 +236,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:
|
||||||
|
|||||||
Reference in New Issue
Block a user