🛡️ Sentinel: [CRITICAL] Fix TOCTOU and DoS vulnerability in file reading

.jules/sentinel.md

@@ -2,3 +2,7 @@
 **Vulnerability:** Unbounded memory consumption DoS risk due to JSON file parsing without size limits in file-backed repositories and stores.
 **Learning:** `json.loads(file.read_text())` reads the entire file content into memory. This can lead to out-of-memory errors or denial-of-service if an attacker or misconfiguration provides an excessively large file. The application relies heavily on file-backed adapters (e.g., `JsonFilePracticeRepository`, `JsonFileProgressSnapshotStore`, `CheckpointStore`).
 **Prevention:** Implement a strict file size limit check using `path.stat().st_size` (e.g., standardized at 10MB or `10 * 1024 * 1024` bytes) before reading the file content into memory in all file-based adapters.
+## 2025-02-23 - [TOCTOU Vulnerability and OOM DoS in Bounded Reads]
+**Vulnerability:** File size limit checks using `path.stat().st_size` combined with `path.read_text()` create TOCTOU (Time-of-Check to Time-of-Use) vulnerabilities and memory DoS risks (e.g., using device files like `/dev/zero` which report size 0 but have infinite stream).
+**Learning:** Using `is_file()` to enforce regular files before size checking is critical. Also, using a bounded `read()` (e.g., `content = f.read(limit + 1)`) and verifying the read length prevents unbounded memory consumption safely, without relying on `stat()`.
+**Prevention:** Instead of `st_size`, verify `is_file()` first, then use a secure bounded read `content = f.read(10 * 1024 * 1024 + 1)` and raise an error if `len(content) > 10 * 1024 * 1024`.

src/python_learning_orchestrated/adapters/checkpoint_store.py

@@ -233,9 +233,16 @@ def _to_int(value: object, default: int) -> int:
 def _read_json(path: Path) -> dict[str, object]:
     if not path.exists():
         return {}
-    if path.stat().st_size > 10 * 1024 * 1024:
+    if not path.is_file():
+        return {}
+
+    with open(path, encoding="utf-8") as f:
+        content = f.read(10 * 1024 * 1024 + 1)
+
+    if len(content) > 10 * 1024 * 1024:
         raise ValueError(f"Checkpoint file {path} exceeds 10MB size limit")
-    parsed = json.loads(path.read_text(encoding="utf-8"))
+
+    parsed = json.loads(content)
     return parsed if isinstance(parsed, dict) else {}
 
 

src/python_learning_orchestrated/adapters/json_file_practice_repository.py

@@ -105,13 +105,19 @@ def record_attempts(self, attempts: list[Attempt]) -> None:
     def _load_storage(self) -> dict[str, object]:
         if not self._file_path.exists():
             return {"items": [], "attempts": []}
-        if self._file_path.stat().st_size > 10 * 1024 * 1024:
-            raise ValueError(
-                f"Practice repository file {self._file_path} exceeds 10MB size limit"
-            )
+        if not self._file_path.is_file():
+            return {"items": [], "attempts": []}
 
         try:
-            content = self._file_path.read_text(encoding="utf-8")
+            with open(self._file_path, encoding="utf-8") as f:
+                content = f.read(10 * 1024 * 1024 + 1)
+
+            if len(content) > 10 * 1024 * 1024:
+                raise ValueError(
+                    f"Practice repository file {self._file_path} exceeds "
+                    "10MB size limit"
+                )
+
             if not content.strip():
                 return {"items": [], "attempts": []}
             parsed = json.loads(content)

src/python_learning_orchestrated/adapters/json_file_progress_repository.py

@@ -42,13 +42,19 @@ def _load_storage(self) -> dict[str, LessonProgress]:
         """Load all persisted progress payloads."""
         if not self._file_path.exists():
             return {}
-        if self._file_path.stat().st_size > 10 * 1024 * 1024:
-            raise ValueError(
-                f"Progress repository file {self._file_path} exceeds 10MB size limit"
-            )
+        if not self._file_path.is_file():
+            return {}
 
         try:
-            content = self._file_path.read_text(encoding="utf-8")
+            with open(self._file_path, encoding="utf-8") as f:
+                content = f.read(10 * 1024 * 1024 + 1)
+
+            if len(content) > 10 * 1024 * 1024:
+                raise ValueError(
+                    f"Progress repository file {self._file_path} exceeds "
+                    "10MB size limit"
+                )
+
             if not content.strip():
                 return {}
 

src/python_learning_orchestrated/adapters/json_file_progress_snapshot_store.py

@@ -36,12 +36,19 @@ def save(self, snapshot: ProgressSnapshot) -> None:
     def _load_payload(self) -> dict[str, object]:
         if not self._file_path.exists():
             return {}
-        if self._file_path.stat().st_size > 10 * 1024 * 1024:
-            raise ValueError(
-                f"Progress snapshot file {self._file_path} exceeds 10MB size limit"
-            )
+        if not self._file_path.is_file():
+            return {}
+
         try:
-            parsed = json.loads(self._file_path.read_text(encoding="utf-8"))
+            with open(self._file_path, encoding="utf-8") as f:
+                content = f.read(10 * 1024 * 1024 + 1)
+
+            if len(content) > 10 * 1024 * 1024:
+                raise ValueError(
+                    f"Progress snapshot file {self._file_path} exceeds 10MB size limit"
+                )
+
+            parsed = json.loads(content)
         except (OSError, json.JSONDecodeError):
             return {}
         return parsed if isinstance(parsed, dict) else {}