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.

## 2024-02-23 - [Device File Size Read Bypass / TOCTOU]
**Vulnerability:** `path.stat().st_size` checks before `path.read_text()` are vulnerable to TOCTOU and bypass via device files (like `/dev/zero`), which report size 0 but produce infinite data.
**Learning:** To prevent unbounded memory consumption and DoS, verifying `path.is_file()` first is crucial. However, the most secure approach combines `path.is_file()` with a bounded read pattern (`f.read(limit + 1)`) instead of relying solely on `st_size` beforehand.
**Prevention:** Ensure file adapters always use `path.is_file()` to reject device files and perform bounded reads (`content = f.read(LIMIT + 1); if len(content) > LIMIT: raise ValueError(...)`) instead of relying on `path.stat().st_size`.
8 changes: 5 additions & 3 deletions src/python_learning_orchestrated/adapters/checkpoint_store.py
Original file line number Diff line number Diff line change
Expand Up @@ -231,11 +231,13 @@ 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:
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 +237 to +238

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 used twice here. To improve maintainability and avoid magic numbers, consider defining it as a module-level constant. This would make the code cleaner and easier to update if the limit needs to change.

For example:

_MAX_FILE_SIZE_BYTES = 10 * 1024 * 1024  # 10MB

# ... inside _read_json
with path.open("r", encoding="utf-8") as f:
    content = f.read(_MAX_FILE_SIZE_BYTES + 1)
if len(content) > _MAX_FILE_SIZE_BYTES:
    # ...

This principle applies to the other file adapters modified in this pull request as well.

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 {}
Comment on lines +236 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.

high

This _read_json function does not handle potential OSError or json.JSONDecodeError, which could lead to unhandled exceptions if the file is corrupted or a read error occurs. Other file adapters in this PR, such as JsonFileProgressSnapshotStore, handle these exceptions by returning a default value. For consistency and robustness, it would be beneficial to wrap the logic in a try...except (OSError, json.JSONDecodeError) block and return {} in case of an error.



Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -103,15 +103,17 @@ 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")
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 +111 to +112

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 and avoid magic numbers, it's recommended to define this value as a module-level constant and reuse it. This makes the code cleaner and simplifies future updates to the size limit.

For example:

_MAX_FILE_SIZE_BYTES = 10 * 1024 * 1024  # 10MB

# ... inside _load_storage
with self._file_path.open("r", encoding="utf-8") as f:
    content = f.read(_MAX_FILE_SIZE_BYTES + 1)
if len(content) > _MAX_FILE_SIZE_BYTES:
    # ...

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 @@ -40,15 +40,17 @@ 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")
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 +48 to +49

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 magic number 10 * 1024 * 1024 is used for the file size limit. It's better to extract this into a module-level constant to improve code clarity and make it easier to manage the value in one place.

For example:

_MAX_FILE_SIZE_BYTES = 10 * 1024 * 1024  # 10MB

# ... inside _load_storage
with self._file_path.open("r", encoding="utf-8") as f:
    content = f.read(_MAX_FILE_SIZE_BYTES + 1)
if len(content) > _MAX_FILE_SIZE_BYTES:
    # ...

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 @@ -34,14 +34,16 @@ 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:
Comment on lines +41 to +42

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. To enhance maintainability, this value should be defined as a constant at the module level. This avoids repetition and makes the limit's purpose clearer.

Example:

_MAX_FILE_SIZE_BYTES = 10 * 1024 * 1024  # 10MB

# ... inside _load_payload
with self._file_path.open("r", encoding="utf-8") as f:
    content = f.read(_MAX_FILE_SIZE_BYTES + 1)
if len(content) > _MAX_FILE_SIZE_BYTES:
    # ...

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