🛡️ Sentinel: [CRITICAL] Fix TOCTOU and unbounded memory consumption DoS in JSON adapters

.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 in File Size Checks]
+**Vulnerability:** Time-of-Check to Time-of-Use (TOCTOU) and Device File Bypass in file size limits.
+**Learning:** Checking `path.stat().st_size` is vulnerable to TOCTOU if the file changes between stat and read. Furthermore, checking `st_size` on device files (like `/dev/zero`) returns 0, bypassing the size check, and `read_text()` will then consume unbounded memory.
+**Prevention:** To prevent unbounded memory consumption and TOCTOU, JSON file-backed adapters must use a secure bounded read pattern. Open the file, read up to `limit + 1` bytes, and raise `ValueError` if the length of the read content exceeds the limit. Do not rely solely on `st_size`.

src/python_learning_orchestrated/adapters/checkpoint_store.py

@@ -233,9 +233,13 @@ 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:
-        raise ValueError(f"Checkpoint file {path} exceeds 10MB size limit")
-    parsed = json.loads(path.read_text(encoding="utf-8"))
+    if not path.is_file():
+        raise ValueError(f"Checkpoint path {path} is not a regular file")
+    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(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:
+        if not self._file_path.is_file():
             raise ValueError(
-                f"Practice repository file {self._file_path} exceeds 10MB size limit"
+                f"Practice repository path {self._file_path} is not a regular file"
             )
 
         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:
+        if not self._file_path.is_file():
             raise ValueError(
-                f"Progress repository file {self._file_path} exceeds 10MB size limit"
+                f"Progress repository path {self._file_path} is not a regular file"
             )
 
         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:
+        if not self._file_path.is_file():
             raise ValueError(
-                f"Progress snapshot file {self._file_path} exceeds 10MB size limit"
+                f"Progress snapshot path {self._file_path} is not a regular file"
             )
         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 {}