🛡️ Sentinel: [CRITICAL] Fix TOCTOU vulnerability and unbound memory DoS in file reads

.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-03-22 - [TOCTOU and Bypassed Size Limits in File Reads]
+**Vulnerability:** A Time-of-Check to Time-of-Use (TOCTOU) vulnerability existed where file size was checked via `path.stat().st_size` combined with `path.exists()` before a full unbounded read `path.read_text()`. This could allow device files (e.g. `/dev/zero`) to report 0 bytes and bypass the size check, then block/crash the application by continuously streaming data into memory during `read_text()`.
+**Learning:** Relying on `path.stat().st_size` for size enforcement before reading is insecure and susceptible to TOCTOU. Furthermore, `path.exists()` allows non-regular files (like device streams) to be parsed, incorrectly reporting their size as 0, thereby completely bypassing naive checks.
+**Prevention:** To prevent unbound memory DoS during file reads: always verify the path is a regular file with `path.is_file()`, eliminate `st_size` checks, and implement a strict bounded read pattern using `f.read(limit + 1)`, manually raising an error if the read content exceeds `limit`.

src/python_learning_orchestrated/adapters/checkpoint_store.py

@@ -231,12 +231,19 @@ 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:
-        raise ValueError(f"Checkpoint file {path} exceeds 10MB size limit")
-    parsed = json.loads(path.read_text(encoding="utf-8"))
-    return parsed if isinstance(parsed, dict) else {}
+
+    with path.open("r", 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")
+
+        if not content.strip():
+            return {}
+
+        parsed = json.loads(content)
+        return parsed if isinstance(parsed, dict) else {}
 
 
 def _write_json(path: Path, payload: dict[str, object]) -> None:

src/python_learning_orchestrated/adapters/json_file_practice_repository.py

@@ -103,19 +103,27 @@ 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")
-            if not content.strip():
-                return {"items": [], "attempts": []}
-            parsed = json.loads(content)
-            return parsed if isinstance(parsed, dict) else {"items": [], "attempts": []}
+            with self._file_path.open("r", 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)
+                return (
+                    parsed
+                    if isinstance(parsed, dict)
+                    else {"items": [], "attempts": []}
+                )
         except (OSError, json.JSONDecodeError):
             return {"items": [], "attempts": []}
 

src/python_learning_orchestrated/adapters/json_file_progress_repository.py

@@ -40,27 +40,30 @@ def reset_progress(self, user_id: str) -> None:
 
     def _load_storage(self) -> dict[str, LessonProgress]:
         """Load all persisted progress payloads."""
-        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 repository file {self._file_path} exceeds 10MB size limit"
-            )
 
         try:
-            content = self._file_path.read_text(encoding="utf-8")
-            if not content.strip():
-                return {}
-
-            data = json.loads(content)
-            if not isinstance(data, dict):
-                return {}
-
-            normalized: dict[str, LessonProgress] = {}
-            for user_id, progress in data.items():
-                if isinstance(user_id, str) and isinstance(progress, dict):
-                    normalized[user_id] = cast(LessonProgress, progress)
-            return normalized
+            with self._file_path.open("r", 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 {}
+
+                data = json.loads(content)
+                if not isinstance(data, dict):
+                    return {}
+
+                normalized: dict[str, LessonProgress] = {}
+                for user_id, progress in data.items():
+                    if isinstance(user_id, str) and isinstance(progress, dict):
+                        normalized[user_id] = cast(LessonProgress, progress)
+                return normalized
         except (json.JSONDecodeError, OSError):
             return {}
 

src/python_learning_orchestrated/adapters/json_file_progress_snapshot_store.py

@@ -34,14 +34,20 @@ 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"))
+            with self._file_path.open("r", 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"
+                    )
+                if not content.strip():
+                    return {}
+                parsed = json.loads(content)
         except (OSError, json.JSONDecodeError):
             return {}
         return parsed if isinstance(parsed, dict) else {}