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>
This commit is contained in:
copilot-swe-agent[bot]
2026-02-08 08:31:21 +00:00
parent d2eb9846d3
commit b97c80d6bd
3 changed files with 18 additions and 11 deletions
+12 -6
View File
@@ -42,23 +42,29 @@ def resolve_file_path(file_path: str, subfolder: str = None) -> str:
Raises: Raises:
HTTPException: If the path attempts to escape the workdir 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: if subfolder:
base_dir = Path(settings.workdir) / subfolder base_dir = workdir / subfolder
else: else:
base_dir = Path(settings.workdir) base_dir = workdir
# Resolve the file path # Resolve the file path
if not os.path.isabs(file_path): if not os.path.isabs(file_path):
# Relative path: join with base_dir
resolved_path = (base_dir / file_path).resolve() resolved_path = (base_dir / file_path).resolve()
else: else:
# Absolute path: resolve as-is
resolved_path = Path(file_path).resolve() 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: try:
resolved_path.relative_to(base_dir.resolve()) resolved_path.relative_to(workdir)
except ValueError: 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}") logger.warning(f"Path traversal attempt detected: {file_path} -> {resolved_path}")
raise HTTPException( raise HTTPException(
status_code=status.HTTP_400_BAD_REQUEST, detail="Invalid file path: path traversal not allowed" status_code=status.HTTP_400_BAD_REQUEST, detail="Invalid file path: path traversal not allowed"
+3 -2
View File
@@ -33,8 +33,9 @@ def upload_with_rclone(file_path: str, destination: str):
# Split and validate destination components # Split and validate destination components
remote, remote_path = destination.split(":", 1) remote, remote_path = destination.split(":", 1)
# Validate remote name (alphanumeric, underscore, hyphen only) # Validate remote name to prevent command injection
if not remote or not all(c.isalnum() or c in ("_", "-") for c in remote): # 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}") raise ValueError(f"Invalid remote name: {remote}")
# Check if rclone is installed and config exists # Check if rclone is installed and config exists
+3 -3
View File
@@ -4,7 +4,7 @@ Shared across multiple OAuth providers to reduce code duplication.
""" """
import logging import logging
from typing import Dict, Any from typing import Dict, Any, Optional
import requests import requests
from fastapi import HTTPException, status from fastapi import HTTPException, status
@@ -14,7 +14,7 @@ logger = logging.getLogger(__name__)
def exchange_oauth_token( 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]: ) -> Dict[str, Any]:
""" """
Exchange an authorization code for tokens from an OAuth provider. 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") error_type = error_json.get("error", "unknown_error")
logger.error(f"Token exchange failed with status {response.status_code}: {error_type}") 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", "")} 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)}") logger.error(f"Failed to parse error response as JSON: {str(json_err)}")
error_detail = {"error": "Unknown error", "status_code": response.status_code} error_detail = {"error": "Unknown error", "status_code": response.status_code}