Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
5 changes: 5 additions & 0 deletions .jules/sentinel.md
Original file line number Diff line number Diff line change
Expand Up @@ -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.

## 2025-03-23 - [TOCTOU and Device File Size Limit Bypass]
**Vulnerability:** File size checks using `path.stat().st_size` can be bypassed if the path points to a device file (like `/dev/zero`, which reports a size of 0). Additionally, there is a Time-of-Check to Time-of-Use (TOCTOU) vulnerability where a file could be modified between the `stat()` check and the `read_text()` call.
**Learning:** Checking file size before reading is insufficient on its own. Device files bypass `st_size` checks, and files can grow between the size check and the read operation, leading to unbounded memory consumption or application hangs.
**Prevention:** Always verify a path is a regular file first using `path.is_file()`. Then, instead of reading the whole file with `read_text()`, open the file and use a bounded read (e.g., `f.read(limit + 1)`) and check if the returned content length exceeds the limit.
12 changes: 9 additions & 3 deletions src/python_learning_orchestrated/adapters/checkpoint_store.py
Original file line number Diff line number Diff line change
Expand Up @@ -233,9 +233,15 @@ 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 file {path} is not a regular file")

with path.open("r", encoding="utf-8") as f:
content = f.read(10 * 1024 * 1024 + 1)
if len(content) > 10 * 1024 * 1024:
Comment on lines +240 to +241

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

medium

The file size limit 10 * 1024 * 1024 is hardcoded. This value is duplicated across multiple files (json_file_practice_repository.py, json_file_progress_repository.py, json_file_progress_snapshot_store.py), making it difficult to maintain. Consider defining it as a constant in a shared module (e.g., a new src/python_learning_orchestrated/adapters/constants.py) and importing it where needed.

For example, in a new constants.py file:

MAX_FILE_SIZE = 10 * 1024 * 1024  # 10MB

Then you can import and use this constant.

Suggested change
content = f.read(10 * 1024 * 1024 + 1)
if len(content) > 10 * 1024 * 1024:
content = f.read(MAX_FILE_SIZE + 1)
if len(content) > MAX_FILE_SIZE:

raise ValueError(f"Checkpoint file {path} exceeds 10MB size limit")

parsed = json.loads(content)
return parsed if isinstance(parsed, dict) else {}
Comment on lines +244 to 245

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

high

An empty checkpoint file will cause json.loads(content) to raise a json.JSONDecodeError because content will be an empty string. This exception is not handled, which could crash the application. It's best to handle this case gracefully by returning an empty dictionary, which is consistent with how a non-existent file is handled.

Suggested change
parsed = json.loads(content)
return parsed if isinstance(parsed, dict) else {}
try:
parsed = json.loads(content)
except json.JSONDecodeError:
return {}
return parsed if isinstance(parsed, dict) else {}



Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -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 file {self._file_path} is not a regular file"
)

try:
content = 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:
Comment on lines +115 to +116

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

medium

The file size limit 10 * 1024 * 1024 is hardcoded here. To improve maintainability, this should be replaced with a shared constant. Please see my other comment about creating a MAX_FILE_SIZE constant.

Suggested change
content = f.read(10 * 1024 * 1024 + 1)
if len(content) > 10 * 1024 * 1024:
content = f.read(MAX_FILE_SIZE + 1)
if len(content) > MAX_FILE_SIZE:

raise ValueError(
f"Practice repository file {self._file_path} "
"exceeds 10MB size limit"
)
if not content.strip():
return {"items": [], "attempts": []}
parsed = json.loads(content)
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -42,13 +42,20 @@ 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 file {self._file_path} is not a regular file"
)

try:
content = 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:
Comment on lines +52 to +53

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

medium

The file size limit 10 * 1024 * 1024 is hardcoded here. To improve maintainability, this should be replaced with a shared constant. Please see my other comment about creating a MAX_FILE_SIZE constant.

Suggested change
content = f.read(10 * 1024 * 1024 + 1)
if len(content) > 10 * 1024 * 1024:
content = f.read(MAX_FILE_SIZE + 1)
if len(content) > MAX_FILE_SIZE:

raise ValueError(
f"Progress repository file {self._file_path} "
"exceeds 10MB size limit"
)

if not content.strip():
return {}

Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -36,12 +36,20 @@ 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 file {self._file_path} is not a regular file"
)

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:
Comment on lines +46 to +47

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

medium

The file size limit 10 * 1024 * 1024 is hardcoded here. To improve maintainability, this should be replaced with a shared constant. Please see my other comment about creating a MAX_FILE_SIZE constant.

Suggested change
content = f.read(10 * 1024 * 1024 + 1)
if len(content) > 10 * 1024 * 1024:
content = f.read(MAX_FILE_SIZE + 1)
if len(content) > MAX_FILE_SIZE:

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 {}
Expand Down
Loading