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-02-23 - [TOCTOU and DoS via File Stat Checks]
**Vulnerability:** Using `path.stat().st_size` to check file size limits before reading the entire file (TOCTOU), and using `.exists()` instead of `.is_file()`, which allows device files (like `/dev/zero`) to report a size of 0 and bypass the check, leading to unbounded memory consumption DoS.
**Learning:** Checking size before reading is susceptible to Time-of-Check to Time-of-Use race conditions if the file is modified in between. Furthermore, device files report incorrect sizes, circumventing the limit entirely if `.exists()` is used instead of `.is_file()`.
**Prevention:** Always verify a path is a regular file using `path.is_file()` first. To prevent TOCTOU when enforcing size limits, implement a secure bounded read (e.g., `content = f.read(limit + 1)`) and raise a ValueError if the returned length exceeds the limit, rather than relying on pre-read stat checks.
12 changes: 8 additions & 4 deletions src/python_learning_orchestrated/adapters/checkpoint_store.py
Original file line number Diff line number Diff line change
Expand Up @@ -231,11 +231,15 @@ 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"))

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 {}
Comment on lines +237 to 243

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 implementation can be improved in two ways:

  1. Robustness: The code doesn't handle potential OSError from file operations or json.JSONDecodeError from parsing. This could lead to unhandled exceptions, whereas other adapters in the codebase handle these gracefully. For consistency and robustness, this one should too.
  2. Maintainability: The file size limit 10 * 1024 * 1024 is a magic number. Defining it as a constant improves readability and makes it easier to change, especially since it's used across multiple files.

Here's a suggestion that addresses both points. Ideally, the MAX_FILE_SIZE_BYTES constant would be defined at the module level to be shared across all adapters.

    MAX_FILE_SIZE_BYTES = 10 * 1024 * 1024  # 10MB
    try:
        with open(path, encoding="utf-8") as f:
            content = f.read(MAX_FILE_SIZE_BYTES + 1)
            if len(content) > MAX_FILE_SIZE_BYTES:
                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 {}
    except (OSError, json.JSONDecodeError, ValueError):
        return {}



Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -103,15 +103,18 @@ 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 open(self._file_path, 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 a magic number. To improve readability and maintainability, consider defining it as a shared constant (e.g., _MAX_FILE_SIZE_BYTES = 10 * 1024 * 1024) at the module level or in a common constants file and using it here and in the other adapters.

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,18 @@ 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 open(self._file_path, 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 file size limit 10 * 1024 * 1024 is a magic number. To improve readability and maintainability, consider defining it as a shared constant (e.g., _MAX_FILE_SIZE_BYTES = 10 * 1024 * 1024) at the module level or in a common constants file and using it here and in the other adapters.

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,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"))
with open(self._file_path, encoding="utf-8") as f:
content = f.read(10 * 1024 * 1024 + 1)
if len(content) > 10 * 1024 * 1024:
Comment on lines +42 to +43

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 a magic number. To improve readability and maintainability, consider defining it as a shared constant (e.g., _MAX_FILE_SIZE_BYTES = 10 * 1024 * 1024) at the module level or in a common constants file and using it here and in the other adapters.

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