From b97c80d6bdcec6bbbf8ba3eb3a164070b7659057 Mon Sep 17 00:00:00 2001 From: "copilot-swe-agent[bot]" <198982749+Copilot@users.noreply.github.com> Date: Sun, 8 Feb 2026 08:31:21 +0000 Subject: [PATCH] fix: address code review feedback on security and type hints - Use Optional[int] type hint for timeout parameter in oauth_helper - Replace bare Exception with specific ValueError and JSONDecodeError - Strengthen rclone remote name validation (must start with alphanumeric) - Fix path traversal validation to check against workdir for absolute paths - Add comprehensive comments for security validations Co-authored-by: christianlouis <361235+christianlouis@users.noreply.github.com> --- app/api/common.py | 18 ++++++++++++------ app/tasks/upload_with_rclone.py | 5 +++-- app/utils/oauth_helper.py | 6 +++--- 3 files changed, 18 insertions(+), 11 deletions(-) diff --git a/app/api/common.py b/app/api/common.py index 23a008c5..0cb911d3 100644 --- a/app/api/common.py +++ b/app/api/common.py @@ -42,23 +42,29 @@ def resolve_file_path(file_path: str, subfolder: str = None) -> str: Raises: HTTPException: If the path attempts to escape the workdir """ - # Build the base directory + # Get the workdir as the security boundary + workdir = Path(settings.workdir).resolve() + + # Build the base directory (workdir or workdir/subfolder) if subfolder: - base_dir = Path(settings.workdir) / subfolder + base_dir = workdir / subfolder else: - base_dir = Path(settings.workdir) + base_dir = workdir # Resolve the file path if not os.path.isabs(file_path): + # Relative path: join with base_dir resolved_path = (base_dir / file_path).resolve() else: + # Absolute path: resolve as-is resolved_path = Path(file_path).resolve() - # Ensure the resolved path is within the base directory (path traversal protection) + # Ensure the resolved path is within workdir (path traversal protection) + # This checks both relative and absolute paths against workdir try: - resolved_path.relative_to(base_dir.resolve()) + resolved_path.relative_to(workdir) except ValueError: - # Path is outside the base directory - potential path traversal attack + # Path is outside the workdir - potential path traversal attack logger.warning(f"Path traversal attempt detected: {file_path} -> {resolved_path}") raise HTTPException( status_code=status.HTTP_400_BAD_REQUEST, detail="Invalid file path: path traversal not allowed" diff --git a/app/tasks/upload_with_rclone.py b/app/tasks/upload_with_rclone.py index c718c9eb..a31949d2 100644 --- a/app/tasks/upload_with_rclone.py +++ b/app/tasks/upload_with_rclone.py @@ -33,8 +33,9 @@ def upload_with_rclone(file_path: str, destination: str): # Split and validate destination components remote, remote_path = destination.split(":", 1) - # Validate remote name (alphanumeric, underscore, hyphen only) - if not remote or not all(c.isalnum() or c in ("_", "-") for c in remote): + # Validate remote name to prevent command injection + # Must start with alphanumeric, can contain alphanumeric, underscore, hyphen + if not remote or not remote[0].isalnum() or not all(c.isalnum() or c in ("_", "-") for c in remote): raise ValueError(f"Invalid remote name: {remote}") # Check if rclone is installed and config exists diff --git a/app/utils/oauth_helper.py b/app/utils/oauth_helper.py index 2aad1ed7..6328daef 100644 --- a/app/utils/oauth_helper.py +++ b/app/utils/oauth_helper.py @@ -4,7 +4,7 @@ Shared across multiple OAuth providers to reduce code duplication. """ import logging -from typing import Dict, Any +from typing import Dict, Any, Optional import requests from fastapi import HTTPException, status @@ -14,7 +14,7 @@ logger = logging.getLogger(__name__) def exchange_oauth_token( - provider_name: str, token_url: str, payload: Dict[str, str], timeout: int = None + provider_name: str, token_url: str, payload: Dict[str, str], timeout: Optional[int] = None ) -> Dict[str, Any]: """ Exchange an authorization code for tokens from an OAuth provider. @@ -63,7 +63,7 @@ def exchange_oauth_token( error_type = error_json.get("error", "unknown_error") logger.error(f"Token exchange failed with status {response.status_code}: {error_type}") error_detail = {"error": error_type, "error_description": error_json.get("error_description", "")} - except Exception as json_err: + except (ValueError, requests.exceptions.JSONDecodeError) as json_err: logger.error(f"Failed to parse error response as JSON: {str(json_err)}") error_detail = {"error": "Unknown error", "status_code": response.status_code}