From 4c216dd1ca0df472f72ae1d39bc3589790cf19b3 Mon Sep 17 00:00:00 2001 From: Alex Blank Date: Mon, 25 May 2026 14:05:12 +0200 Subject: [PATCH] refactor: extract shared Pydantic validators - Create shared_validators.py with validate_mount_path, validate_files, validate_env_vars, validate_volumes - Refactor config_folders.py to use shared validators - Refactor tool_configs.py to use shared validators - Refactor config_profiles.py to use shared env_vars validator - Reduce ~80 lines of duplicate validation code --- apps/api/src/api/config_folders.py | 47 ++----------- apps/api/src/api/config_profiles.py | 6 +- apps/api/src/api/shared_validators.py | 99 +++++++++++++++++++++++++++ apps/api/src/api/tool_configs.py | 42 ++---------- 4 files changed, 114 insertions(+), 80 deletions(-) create mode 100644 apps/api/src/api/shared_validators.py diff --git a/apps/api/src/api/config_folders.py b/apps/api/src/api/config_folders.py index 559a65d..0bfdc3e 100644 --- a/apps/api/src/api/config_folders.py +++ b/apps/api/src/api/config_folders.py @@ -8,6 +8,7 @@ from pydantic import BaseModel, Field, field_validator from sqlalchemy import select from sqlalchemy.ext.asyncio import AsyncSession +from src.api.shared_validators import validate_files as _validate_files, validate_mount_path as _validate_mount_path from src.auth.dependencies import get_current_user_id, get_db_session from src.models.config_folder import ConfigFolder @@ -15,9 +16,6 @@ logger = logging.getLogger(__name__) router = APIRouter(prefix="/config-folders", tags=["config-folders"]) -MAX_FOLDER_SIZE_MB = 10 -MAX_FOLDER_SIZE_BYTES = MAX_FOLDER_SIZE_MB * 1024 * 1024 - class ConfigFolderCreate(BaseModel): name: str = Field(description="Folder name (unique per user)") @@ -28,24 +26,12 @@ class ConfigFolderCreate(BaseModel): @field_validator("mount_path") @classmethod def validate_mount_path(cls, v: str) -> str: - if not v.startswith("/"): - raise ValueError("Mount path must be absolute (start with /)") - return v + return _validate_mount_path(v) @field_validator("files") @classmethod def validate_files(cls, v: dict) -> dict: - 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_FOLDER_SIZE_BYTES: - raise ValueError(f"Total folder size exceeds {MAX_FOLDER_SIZE_MB}MB limit") - - return v + return _validate_files(v) class ConfigFolderUpdate(BaseModel): @@ -58,29 +44,12 @@ class ConfigFolderUpdate(BaseModel): @field_validator("mount_path") @classmethod def validate_mount_path(cls, v: str | None) -> str | None: - if v is None: - return v - if not v.startswith("/"): - raise ValueError("Mount path must be absolute (start with /)") - return v + return _validate_mount_path(v) @field_validator("files") @classmethod def validate_files(cls, v: dict | None) -> dict | None: - 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_FOLDER_SIZE_BYTES: - raise ValueError(f"Total folder size exceeds {MAX_FOLDER_SIZE_MB}MB limit") - - return v + return _validate_files(v) class ProjectOverrideCreate(BaseModel): @@ -90,11 +59,7 @@ class ProjectOverrideCreate(BaseModel): @field_validator("mount_path") @classmethod def validate_mount_path(cls, v: str | None) -> str | None: - if v is None: - return v - if not v.startswith("/"): - raise ValueError("Mount path must be absolute (start with /)") - return v + return _validate_mount_path(v) class ConfigFolderResponse(BaseModel): diff --git a/apps/api/src/api/config_profiles.py b/apps/api/src/api/config_profiles.py index 08eef3b..573438f 100644 --- a/apps/api/src/api/config_profiles.py +++ b/apps/api/src/api/config_profiles.py @@ -9,6 +9,7 @@ from sqlalchemy import select from sqlalchemy.ext.asyncio import AsyncSession from sqlalchemy.orm import selectinload +from src.api.shared_validators import validate_env_vars as _validate_env_vars from src.auth.dependencies import get_current_user_id, get_db_session from src.models.config_profile import ConfigProfile, ConfigProfileInclude from src.models.project import Project @@ -120,9 +121,10 @@ class ConfigProfileCreate(BaseModel): @field_validator("env_vars") @classmethod def validate_env_vars(cls, v: dict) -> dict: - if not isinstance(v, dict): + result = _validate_env_vars(v) + if result is None: raise ValueError("env_vars must be a JSON object") - return v + return result @field_validator("runtime_hints") @classmethod diff --git a/apps/api/src/api/shared_validators.py b/apps/api/src/api/shared_validators.py new file mode 100644 index 0000000..e430bf8 --- /dev/null +++ b/apps/api/src/api/shared_validators.py @@ -0,0 +1,99 @@ +"""Shared Pydantic validators for API schemas.""" + +from typing import Any + +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 diff --git a/apps/api/src/api/tool_configs.py b/apps/api/src/api/tool_configs.py index caa1cbf..13a5502 100644 --- a/apps/api/src/api/tool_configs.py +++ b/apps/api/src/api/tool_configs.py @@ -8,11 +8,9 @@ from pydantic import BaseModel, Field, field_validator from sqlalchemy import select from sqlalchemy.ext.asyncio import AsyncSession +from src.api.shared_validators import validate_env_vars as _validate_env_vars, validate_volumes as _validate_volumes from src.auth.dependencies import get_current_user_id, get_db_session from src.models.tool_config import ToolConfig -from src.models.tool_type import ToolType - -logger = logging.getLogger(__name__) router = APIRouter(prefix="/tool-configs", tags=["tool-configs"]) @@ -42,27 +40,12 @@ class ToolConfigCreate(BaseModel): @field_validator("environment_variables") @classmethod def validate_env_vars(cls, v: dict | None) -> dict | None: - if v is None: - return v - if not isinstance(v, dict): - raise ValueError("environment_variables must be a JSON object") - return v + return _validate_env_vars(v) @field_validator("volumes") @classmethod def validate_volumes(cls, v: list | None) -> list | None: - 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 + return _validate_volumes(v) class ToolConfigUpdate(BaseModel): @@ -88,27 +71,12 @@ class ToolConfigUpdate(BaseModel): @field_validator("environment_variables") @classmethod def validate_env_vars(cls, v: dict | None) -> dict | None: - if v is None: - return v - if not isinstance(v, dict): - raise ValueError("environment_variables must be a JSON object") - return v + return _validate_env_vars(v) @field_validator("volumes") @classmethod def validate_volumes(cls, v: list | None) -> list | None: - 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 + return _validate_volumes(v) class ToolConfigResponse(BaseModel):