gitea-admin/gh-christianlouis-docuelevate
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fix(security): reduce code duplication and fix security issues in OAuth and file handling

Browse Source 
- Extract common OAuth token exchange logic to shared utility (oauth_helper.py)
- Remove sensitive data logging (client_secret, authorization codes)
- Add path traversal validation in resolve_file_path()
- Add input validation for rclone destination parameter
- Replace bare Exception catches with specific exception types
- Use RuntimeError instead of generic Exception for better error handling

Co-authored-by: christianlouis <361235+christianlouis@users.noreply.github.com>
This commit is contained in:
copilot-swe-agent[bot]
2026-02-08 08:26:10 +00:00
parent 6ec4e2b8a7
commit 551b23a80c
6 changed files with 240 additions and 257 deletions
Download Patch File Download Diff File Expand all files Collapse all files
app/api/common.py
+38 -8
View File
@@ -3,8 +3,9 @@ Common utilities for API routes
"""
import logging
import os
from pathlib import Path
from sqlalchemy.orm import Session
from fastapi import Depends
from fastapi import Depends, HTTPException, status
from app.database import SessionLocal
from app.config import settings
@@ -22,15 +23,44 @@ def get_db():
def resolve_file_path(file_path: str, subfolder: str = None) -> str:
"""
Resolves a file path to an absolute path.
Resolves a file path to an absolute path with path traversal protection.
If the path is not absolute, it will be joined with the workdir path.
Optionally, can include a subfolder like 'processed'.
Returns the absolute file path.
Security: Validates that the resolved path stays within the workdir
to prevent path traversal attacks (e.g., ../../etc/passwd).
Args:
file_path: The file path to resolve
subfolder: Optional subfolder within workdir
Returns:
The validated absolute file path
Raises:
HTTPException: If the path attempts to escape the workdir
"""
# Build the base directory
if subfolder:
base_dir = Path(settings.workdir) / subfolder
else:
base_dir = Path(settings.workdir)
# Resolve the file path
if not os.path.isabs(file_path):
if subfolder:
file_path = os.path.join(settings.workdir, subfolder, file_path)
else:
file_path = os.path.join(settings.workdir, file_path)
return file_path
resolved_path = (base_dir / file_path).resolve()
else:
resolved_path = Path(file_path).resolve()
# Ensure the resolved path is within the base directory (path traversal protection)
try:
resolved_path.relative_to(base_dir.resolve())
except ValueError:
# Path is outside the base directory - 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"
)
return str(resolved_path)
app/api/dropbox.py
+25 -78
View File
@@ -11,6 +11,7 @@ from typing import Optional
from app.auth import require_login
from app.config import settings
from app.utils.oauth_helper import exchange_oauth_token
# Set up logging
logger = logging.getLogger(__name__)
@@ -31,84 +32,30 @@ async def exchange_dropbox_token(
Exchange an authorization code for a refresh token from Dropbox.
This is done on the server to avoid exposing client secret in the browser.
"""
try:
logger.info("Starting Dropbox token exchange process")
# Prepare the token request
token_url = "https://api.dropboxapi.com/oauth2/token"
payload = {
'client_id': client_id,
'client_secret': client_secret,
'code': code,
'redirect_uri': redirect_uri,
'grant_type': 'authorization_code'
}
# Log request details (excluding secret)
safe_payload = payload.copy()
safe_payload['client_secret'] = '[REDACTED]'
safe_payload['code'] = f"{code[:5]}...{code[-5:]}" if len(code) > 10 else '[REDACTED]'
logger.info(f"Token exchange request payload: {safe_payload}")
# Make the token request
logger.info("Sending POST request to Dropbox for token exchange")
response = requests.post(token_url, data=payload, timeout=settings.http_request_timeout)
# Check if the request was successful
logger.info(f"Token exchange response status: {response.status_code}")
if response.status_code != 200:
# Log the error response for debugging
try:
error_json = response.json()
logger.error(f"Token exchange failed with status {response.status_code}: {error_json}")
error_detail = error_json
except Exception as json_err:
logger.error(f"Failed to parse error response as JSON: {str(json_err)}")
logger.error(f"Raw response content: {response.content[:500]}") # Limit log size
error_detail = {"error": "Unknown error", "raw_content_snippet": str(response.content[:100])}
raise HTTPException(
status_code=status.HTTP_400_BAD_REQUEST,
detail=f"Token exchange failed: {error_detail}"
)
# Return the token response
token_data = response.json()
# Validate the token response
if "refresh_token" not in token_data:
logger.error(f"Dropbox returned success but no refresh token found in response: {token_data.keys()}")
raise HTTPException(
status_code=status.HTTP_502_BAD_GATEWAY,
detail="Dropbox OAuth server returned success but no refresh token was included"
)
# Calculate token length for logging
refresh_token_length = len(token_data.get("refresh_token", ""))
access_token_length = len(token_data.get("access_token", ""))
logger.info(f"Successfully exchanged authorization code for Dropbox tokens. "
f"Refresh token length: {refresh_token_length}, "
f"Access token length: {access_token_length}")
# Return just what's needed by the frontend
return {
"refresh_token": token_data["refresh_token"],
"access_token": token_data["access_token"],
"expires_in": token_data.get("expires_in", 14400)
}
except HTTPException:
# Re-raise HTTP exceptions as they already have appropriate status codes
raise
except Exception as e:
logger.exception(f"Unexpected error during Dropbox token exchange: {str(e)}")
raise HTTPException(
status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
detail=f"Failed to exchange token: {str(e)}"
)
# Prepare the token request
token_url = "https://api.dropboxapi.com/oauth2/token"
payload = {
'client_id': client_id,
'client_secret': client_secret,
'code': code,
'redirect_uri': redirect_uri,
'grant_type': 'authorization_code'
}
# Use shared OAuth helper (handles secure logging and error handling)
token_data = exchange_oauth_token(
provider_name="Dropbox",
token_url=token_url,
payload=payload
)
# Return just what's needed by the frontend
return {
"refresh_token": token_data["refresh_token"],
"access_token": token_data["access_token"],
"expires_in": token_data.get("expires_in", 14400)
}
@router.post("/dropbox/update-settings")
@require_login
app/api/google_drive.py
+25 -78
View File
@@ -11,6 +11,7 @@ from datetime import datetime, timedelta
from app.auth import require_login
from app.config import settings
from app.utils.oauth_helper import exchange_oauth_token
# Set up logging
logger = logging.getLogger(__name__)
@@ -31,84 +32,30 @@ async def exchange_google_drive_token(
Exchange an authorization code for refresh and access tokens from Google.
This is done on the server to avoid exposing client secret in the browser.
"""
try:
logger.info("Starting Google Drive token exchange process")
# Prepare the token request
token_url = "https://oauth2.googleapis.com/token"
payload = {
'client_id': client_id,
'client_secret': client_secret,
'code': code,
'redirect_uri': redirect_uri,
'grant_type': 'authorization_code'
}
# Log request details (excluding secret)
safe_payload = payload.copy()
safe_payload['client_secret'] = '[REDACTED]'
safe_payload['code'] = f"{code[:5]}...{code[-5:]}" if len(code) > 10 else '[REDACTED]'
logger.info(f"Token exchange request payload: {safe_payload}")
# Make the token request
logger.info("Sending POST request to Google for token exchange")
response = requests.post(token_url, data=payload, timeout=settings.http_request_timeout)
# Check if the request was successful
logger.info(f"Token exchange response status: {response.status_code}")
if response.status_code != 200:
# Log the error response for debugging
try:
error_json = response.json()
logger.error(f"Token exchange failed with status {response.status_code}: {error_json}")
error_detail = error_json
except Exception as json_err:
logger.error(f"Failed to parse error response as JSON: {str(json_err)}")
logger.error(f"Raw response content: {response.content[:500]}") # Limit log size
error_detail = {"error": "Unknown error", "raw_content_snippet": str(response.content[:100])}
raise HTTPException(
status_code=status.HTTP_400_BAD_REQUEST,
detail=f"Token exchange failed: {error_detail}"
)
# Return the token response
token_data = response.json()
# Validate the token response
if "refresh_token" not in token_data:
logger.error(f"Google returned success but no refresh token found in response: {token_data.keys()}")
raise HTTPException(
status_code=status.HTTP_502_BAD_GATEWAY,
detail="Google OAuth server returned success but no refresh token was included"
)
# Calculate token length for logging
refresh_token_length = len(token_data.get("refresh_token", ""))
access_token_length = len(token_data.get("access_token", ""))
logger.info(f"Successfully exchanged authorization code for Google Drive tokens. "
f"Refresh token length: {refresh_token_length}, "
f"Access token length: {access_token_length}")
# Return just what's needed by the frontend
return {
"refresh_token": token_data["refresh_token"],
"access_token": token_data["access_token"],
"expires_in": token_data.get("expires_in", 3600)
}
except HTTPException:
# Re-raise HTTP exceptions as they already have appropriate status codes
raise
except Exception as e:
logger.exception(f"Unexpected error during Google Drive token exchange: {str(e)}")
raise HTTPException(
status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
detail=f"Failed to exchange token: {str(e)}"
)
# Prepare the token request
token_url = "https://oauth2.googleapis.com/token"
payload = {
'client_id': client_id,
'client_secret': client_secret,
'code': code,
'redirect_uri': redirect_uri,
'grant_type': 'authorization_code'
}
# Use shared OAuth helper (handles secure logging and error handling)
token_data = exchange_oauth_token(
provider_name="Google Drive",
token_url=token_url,
payload=payload
)
# Return just what's needed by the frontend
return {
"refresh_token": token_data["refresh_token"],
"access_token": token_data["access_token"],
"expires_in": token_data.get("expires_in", 3600)
}
@router.post("/google-drive/update-settings")
@require_login
app/api/onedrive.py
+25 -79
View File
@@ -11,6 +11,7 @@ from typing import Optional
from app.auth import require_login
from app.config import settings
from app.utils.oauth_helper import exchange_oauth_token
# Set up logging
logger = logging.getLogger(__name__)
@@ -31,85 +32,30 @@ async def exchange_onedrive_token(
Exchange an authorization code for a refresh token.
This is done on the server to avoid exposing client secret in the browser.
"""
try:
logger.info(f"Starting OneDrive token exchange process with tenant_id: {tenant_id}")
# Prepare the token request
token_url = f"https://login.microsoftonline.com/{tenant_id}/oauth2/v2.0/token"
logger.info(f"Using token URL: {token_url}")
payload = {
'client_id': client_id,
'scope': 'https://graph.microsoft.com/.default offline_access',
'code': code,
'redirect_uri': redirect_uri,
'grant_type': 'authorization_code',
'client_secret': client_secret
}
# Log request details (excluding secret)
safe_payload = payload.copy()
safe_payload['client_secret'] = '[REDACTED]'
safe_payload['code'] = f"{code[:5]}...{code[-5:]}" if len(code) > 10 else '[REDACTED]'
logger.info(f"Token exchange request payload: {safe_payload}")
# Make the token request
logger.info("Sending POST request to Microsoft for token exchange")
response = requests.post(token_url, data=payload, timeout=settings.http_request_timeout)
# Check if the request was successful
logger.info(f"Token exchange response status: {response.status_code}")
if response.status_code != 200:
# Log the error response for debugging
try:
error_json = response.json()
logger.error(f"Token exchange failed with status {response.status_code}: {error_json}")
error_detail = error_json
except Exception as json_err:
logger.error(f"Failed to parse error response as JSON: {str(json_err)}")
logger.error(f"Raw response content: {response.content[:500]}") # Limit log size
error_detail = {"error": "Unknown error", "raw_content_snippet": str(response.content[:100])}
raise HTTPException(
status_code=status.HTTP_400_BAD_REQUEST,
detail=f"Token exchange failed: {error_detail}"
)
# Return the token response
token_data = response.json()
# Validate the token response
if "refresh_token" not in token_data:
logger.error(f"Microsoft returned success but no refresh token found in response: {token_data.keys()}")
raise HTTPException(
status_code=status.HTTP_502_BAD_GATEWAY,
detail="Microsoft OAuth server returned success but no refresh token was included"
)
# Calculate token length for logging
refresh_token_length = len(token_data.get("refresh_token", ""))
access_token_length = len(token_data.get("access_token", ""))
logger.info(f"Successfully exchanged authorization code for OneDrive tokens. "
f"Refresh token length: {refresh_token_length}, "
f"Access token length: {access_token_length}")
# Return just what's needed by the frontend
return {
"refresh_token": token_data["refresh_token"],
"expires_in": token_data.get("expires_in", 3600)
}
except HTTPException:
# Re-raise HTTP exceptions as they already have appropriate status codes
raise
except Exception as e:
logger.exception(f"Unexpected error during OneDrive token exchange: {str(e)}")
raise HTTPException(
status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
detail=f"Failed to exchange token: {str(e)}"
)
# Prepare the token request
token_url = f"https://login.microsoftonline.com/{tenant_id}/oauth2/v2.0/token"
payload = {
'client_id': client_id,
'scope': 'https://graph.microsoft.com/.default offline_access',
'code': code,
'redirect_uri': redirect_uri,
'grant_type': 'authorization_code',
'client_secret': client_secret
}
# Use shared OAuth helper (handles secure logging and error handling)
token_data = exchange_oauth_token(
provider_name="OneDrive",
token_url=token_url,
payload=payload
)
# Return just what's needed by the frontend
return {
"refresh_token": token_data["refresh_token"],
"expires_in": token_data.get("expires_in", 3600)
}
@router.get("/onedrive/test-token")
@require_login