🛡️ Sentinel: [CRITICAL] Fix TOCTOU vulnerability in JSON file adapters

.jules/sentinel.md

@@ -2,3 +2,8 @@
 **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.
+
+## 2024-05-18 - [TOCTOU in file size limit checking]
+**Vulnerability:** Checking file size using `path.stat().st_size` prior to reading is vulnerable to Time-Of-Check to Time-Of-Use (TOCTOU) and out-of-memory DoS, as well as bypassing checks via device files like `/dev/zero` which report size 0.
+**Learning:** `path.exists()` or `path.stat().st_size` checks the state of the filesystem. By the time the file is read, the file could have been changed, or a different file type like a device file could have been placed there, evading the check.
+**Prevention:** Verify it's a regular file first (`path.is_file()`), then read with a strict bound like `f.read(limit + 1)`, and if the read returned size is strictly greater than the limit, throw a size limit exception.

src/python_learning_orchestrated/adapters/checkpoint_store.py

@@ -231,11 +231,17 @@ def _to_int(value: object, default: int) -> int:
 
 
 def _read_json(path: Path) -> dict[str, object]:
-    if not path.exists():
+    if not path.is_file():
         return {}
-    if path.stat().st_size > 10 * 1024 * 1024:
+
+    # Security: Use bounded read (limit + 1) to prevent out-of-memory DoS
+    # from excessively large files or malicious device files (e.g., /dev/zero).
+    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

@@ -103,15 +103,20 @@ def record_attempts(self, attempts: list[Attempt]) -> None:
         self._save_storage(storage)
 
     def _load_storage(self) -> dict[str, object]:
-        if not self._file_path.exists():
+        if not self._file_path.is_file():
             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"
-            )
 
         try:
-            content = self._file_path.read_text(encoding="utf-8")
+            # Security: Use bounded read (limit + 1) to prevent out-of-memory DoS
+            # from excessively large files or malicious device files (e.g., /dev/zero).
+            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_snapshot_store.py

@@ -34,14 +34,18 @@ def save(self, snapshot: ProgressSnapshot) -> None:
         self._save_payload(progress_snapshot_to_payload(snapshot))
 
     def _load_payload(self) -> dict[str, object]:
-        if not self._file_path.exists():
+        if not self._file_path.is_file():
             return {}
-        if self._file_path.stat().st_size > 10 * 1024 * 1024:
-            raise ValueError(
-                f"Progress snapshot file {self._file_path} exceeds 10MB size limit"
-            )
         try:
-            parsed = json.loads(self._file_path.read_text(encoding="utf-8"))
+            # Security: Use bounded read (limit + 1) to prevent out-of-memory DoS
+            # from excessively large files or malicious device files (e.g., /dev/zero).
+            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 {}