alex
22474cdba5
Backend (ruff): - Fix 106 errors: move imports to top of file (E402) - Remove unused imports (F401) - Add missing imports for undefined names (F821) - Remove unused variables (F841) - Fix test_models.py broken RefreshToken test - Fix test_projects_api.py missing TestClient import Frontend (eslint): - Remove unused imports/variables across 10 files - Fix explicit any types in client.ts and sessions.ts - Clean up empty block statements in terminal.tsx Quality gates: ruff (pass), eslint (pass), tsc --noEmit (pass), pytest (98 passed, 4 pre-existing failures)
99 lines
2.6 KiB
Python
99 lines
2.6 KiB
Python
"""Shared Pydantic validators for API schemas."""
|
|
|
|
|
|
MAX_FOLDER_SIZE_MB = 10
|
|
MAX_FOLDER_SIZE_BYTES = MAX_FOLDER_SIZE_MB * 1024 * 1024
|
|
|
|
|
|
def validate_mount_path(v: str | None) -> str | None:
|
|
"""Validate that a mount path is absolute (starts with /).
|
|
|
|
Args:
|
|
v: Mount path string or None.
|
|
|
|
Returns:
|
|
The validated path, or None if input was None.
|
|
|
|
Raises:
|
|
ValueError: If path is not absolute.
|
|
"""
|
|
if v is None:
|
|
return v
|
|
if not v.startswith("/"):
|
|
raise ValueError("Mount path must be absolute (start with /)")
|
|
return v
|
|
|
|
|
|
def validate_files(v: dict | None, max_size_bytes: int = MAX_FOLDER_SIZE_BYTES) -> dict | None:
|
|
"""Validate file dict for path traversal and size limits.
|
|
|
|
Args:
|
|
v: Dict of {path: content} or None.
|
|
max_size_bytes: Maximum total size in bytes.
|
|
|
|
Returns:
|
|
The validated dict, or None if input was None.
|
|
|
|
Raises:
|
|
ValueError: If path traversal detected or size limit exceeded.
|
|
"""
|
|
if v is None:
|
|
return v
|
|
|
|
total_size = 0
|
|
for path, content in v.items():
|
|
# Check for path traversal
|
|
if ".." in path or path.startswith("/"):
|
|
raise ValueError(f"Invalid file path: {path}")
|
|
total_size += len(content.encode("utf-8"))
|
|
|
|
if total_size > max_size_bytes:
|
|
raise ValueError(f"Total folder size exceeds {max_size_bytes // (1024 * 1024)}MB limit")
|
|
|
|
return v
|
|
|
|
|
|
def validate_env_vars(v: dict | None) -> dict | None:
|
|
"""Validate that environment variables is a JSON object.
|
|
|
|
Args:
|
|
v: Dict of env vars or None.
|
|
|
|
Returns:
|
|
The validated dict, or None if input was None.
|
|
|
|
Raises:
|
|
ValueError: If not a dict.
|
|
"""
|
|
if v is None:
|
|
return v
|
|
if not isinstance(v, dict):
|
|
raise ValueError("environment_variables must be a JSON object")
|
|
return v
|
|
|
|
|
|
def validate_volumes(v: list | None) -> list | None:
|
|
"""Validate volume mounts list.
|
|
|
|
Args:
|
|
v: List of volume dicts or None.
|
|
|
|
Returns:
|
|
The validated list, or None if input was None.
|
|
|
|
Raises:
|
|
ValueError: If not a list or missing required fields.
|
|
"""
|
|
if v is None:
|
|
return v
|
|
if not isinstance(v, list):
|
|
raise ValueError("volumes must be a JSON array")
|
|
for i, vol in enumerate(v):
|
|
if not isinstance(vol, dict):
|
|
raise ValueError(f"Volume at index {i} must be an object")
|
|
if "source" not in vol:
|
|
raise ValueError(f"Volume at index {i} must have 'source' field")
|
|
if "target" not in vol:
|
|
raise ValueError(f"Volume at index {i} must have 'target' field")
|
|
return v
|