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

Add Google Drive authorization processing and error handling templates

Browse Source 
- Implemented `google_drive_callback.html` for processing Google Drive authorization, including UI for success and error states.
- Added JavaScript functionality for exchanging authorization codes, saving settings, and handling folder selection.
- Created `google_drive_callback_error.html` to display error messages during the authorization process.
This commit is contained in:
Christian Krakau-Louis
2025-04-11 00:13:57 +02:00
parent b258dd2f29
commit 4c50c0aae6
26 changed files with 3297 additions and 727 deletions
Download Patch File Download Diff File Expand all files Collapse all files
app/api/__init__.py
+2
View File
@@ -13,6 +13,7 @@ from app.api.onedrive import router as onedrive_router
from app.api.dropbox import router as dropbox_router
from app.api.openai import router as openai_router
from app.api.azure import router as azure_router
from app.api.google_drive import router as google_drive_router
# Set up logging
logger = logging.getLogger(__name__)
@@ -29,3 +30,4 @@ router.include_router(onedrive_router)
router.include_router(dropbox_router)
router.include_router(openai_router)
router.include_router(azure_router)
router.include_router(google_drive_router)
app/api/dropbox.py
+74 -40
View File
@@ -5,6 +5,9 @@ from fastapi import APIRouter, Request, HTTPException, status, Form
import logging
import os
import requests
import json
from datetime import datetime, timedelta
from typing import Optional
from app.auth import require_login
from app.config import settings
@@ -157,60 +160,91 @@ async def update_dropbox_settings(
@require_login
async def test_dropbox_token(request: Request):
"""
Test if the configured Dropbox refresh token is valid.
Test if the configured Dropbox token is valid and return expiration information.
"""
try:
from app.tasks.upload_to_dropbox import get_dropbox_client
logger.info("Testing Dropbox token validity")
if not settings.dropbox_refresh_token:
logger.warning("No Dropbox refresh token configured")
if not settings.dropbox_refresh_token or not settings.dropbox_app_key or not settings.dropbox_app_secret:
logger.warning("Dropbox credentials not fully configured")
return {
"status": "error",
"message": "No Dropbox refresh token is configured"
"message": "Dropbox credentials are not fully configured"
}
# Check if app key and app secret are configured
if not settings.dropbox_app_key or not settings.dropbox_app_secret:
logger.warning("Dropbox app key or app secret is missing")
# Check token validity by getting current account info
headers = {"Authorization": f"Bearer {settings.dropbox_refresh_token}"}
response = requests.post(
"https://api.dropboxapi.com/2/users/get_current_account",
headers=headers
)
# If token is invalid, try refreshing it
if response.status_code == 401:
logger.info("Dropbox access token invalid or expired, trying to refresh")
# Get a new access token using the refresh token
refresh_url = "https://api.dropbox.com/oauth2/token"
refresh_data = {
"grant_type": "refresh_token",
"refresh_token": settings.dropbox_refresh_token,
"client_id": settings.dropbox_app_key,
"client_secret": settings.dropbox_app_secret
}
refresh_response = requests.post(refresh_url, data=refresh_data)
if refresh_response.status_code != 200:
logger.error(f"Failed to refresh Dropbox token: {refresh_response.text}")
return {
"status": "error",
"message": "Refresh token has expired or is invalid",
"needs_reauth": True
}
token_info = refresh_response.json()
access_token = token_info.get("access_token")
# Try again with the new access token
headers = {"Authorization": f"Bearer {access_token}"}
response = requests.post(
"https://api.dropboxapi.com/2/users/get_current_account",
headers=headers
)
if response.status_code != 200:
logger.error(f"Dropbox token test failed: {response.status_code} {response.text}")
return {
"status": "error",
"message": "Dropbox app key or app secret is missing",
"missing_config": True
"message": f"Token validation failed with status {response.status_code}: {response.text}"
}
# Try to get a client using the configured refresh token
try:
dbx = get_dropbox_client()
# Test connection by getting account info
account = dbx.users_get_current_account()
logger.info(f"Successfully connected to Dropbox as {account.name.display_name}")
return {
"status": "success",
"message": f"Token is valid! Connected as {account.name.display_name}",
"account": account.name.display_name,
"email": account.email
}
except Exception as e:
error_msg = str(e)
logger.error(f"Dropbox token test failed: {error_msg}")
# Determine if this is an authentication error
is_auth_error = "auth" in error_msg.lower() or "invalid" in error_msg.lower()
return {
"status": "error",
"message": f"Token validation failed: {error_msg}",
"is_auth_error": is_auth_error,
"needs_reauth": is_auth_error
}
# Get account info
account_info = response.json()
account_email = account_info.get("email", "Unknown account")
account_name = account_info.get("name", {}).get("display_name", "Unknown user")
# Dropbox refresh tokens don't expire, but we should note that in our response
token_info = {
"expires_in_human": "Never expires (perpetual token)",
"is_perpetual": True
}
logger.info(f"Successfully connected to Dropbox as {account_email}")
return {
"status": "success",
"message": f"Dropbox connection successful",
"account": account_email,
"account_name": account_name,
"token_info": token_info
}
except Exception as e:
logger.exception("Unexpected error testing Dropbox token")
logger.exception(f"Unexpected error testing Dropbox token: {str(e)}")
return {
"status": "error",
"message": f"Unexpected error: {str(e)}"
"message": f"Connection error: {str(e)}"
}
@router.post("/dropbox/save-settings")
app/api/google_drive.py
+506
View File
@@ -0,0 +1,506 @@
"""
Google Drive API endpoints
"""
from fastapi import APIRouter, Request, HTTPException, status, Form
import logging
import os
import requests
import json
from typing import Optional
from datetime import datetime, timedelta
from app.auth import require_login
from app.config import settings
# Set up logging
logger = logging.getLogger(__name__)
router = APIRouter()
@router.post("/google-drive/exchange-token")
@require_login
async def exchange_google_drive_token(
request: Request,
client_id: str = Form(...),
client_secret: str = Form(...),
redirect_uri: str = Form(...),
code: str = Form(...),
folder_id: Optional[str] = Form(None)
):
"""
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)
# 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)}"
)
@router.post("/google-drive/update-settings")
@require_login
async def update_google_drive_settings(
request: Request,
client_id: str = Form(None),
client_secret: str = Form(None),
refresh_token: str = Form(...),
folder_id: str = Form(None),
use_oauth: str = Form("true")
):
"""
Update Google Drive settings in memory
"""
try:
logger.info("Updating Google Drive settings in memory")
# Convert use_oauth string to boolean
use_oauth_bool = use_oauth.lower() in ("true", "1", "yes", "y", "t")
# Update settings in memory
if refresh_token:
settings.google_drive_refresh_token = refresh_token
logger.info("Updated GOOGLE_DRIVE_REFRESH_TOKEN in memory")
if client_id:
settings.google_drive_client_id = client_id
logger.info("Updated GOOGLE_DRIVE_CLIENT_ID in memory")
if client_secret:
settings.google_drive_client_secret = client_secret
logger.info("Updated GOOGLE_DRIVE_CLIENT_SECRET in memory")
if folder_id:
settings.google_drive_folder_id = folder_id
logger.info("Updated GOOGLE_DRIVE_FOLDER_ID in memory")
# Set the OAuth flag
settings.google_drive_use_oauth = use_oauth_bool
logger.info(f"Updated GOOGLE_DRIVE_USE_OAUTH in memory to {use_oauth_bool}")
return {
"status": "success",
"message": "Google Drive settings have been updated in memory"
}
except Exception as e:
logger.exception(f"Unexpected error updating Google Drive settings: {str(e)}")
raise HTTPException(
status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
detail=f"Failed to update Google Drive settings: {str(e)}"
)
@router.get("/google-drive/test-token")
@require_login
async def test_google_drive_token(request: Request):
"""
Test if the configured Google Drive token is valid.
Tests both OAuth and service account approaches based on configuration.
"""
try:
from app.tasks.upload_to_google_drive import get_drive_service_oauth, get_google_drive_service
logger.info("Testing Google Drive token validity")
# Check if OAuth is enabled and configured
if getattr(settings, 'google_drive_use_oauth', False):
if not (settings.google_drive_client_id and
settings.google_drive_client_secret and
settings.google_drive_refresh_token):
logger.warning("Google Drive OAuth credentials not fully configured")
return {
"status": "error",
"message": "Google Drive OAuth credentials are not fully configured"
}
try:
# Test OAuth connection
service = get_drive_service_oauth()
# Get credentials for checking token validity
import google.oauth2.credentials
from google.auth.transport.requests import Request
credentials = google.oauth2.credentials.Credentials(
token=None,
refresh_token=settings.google_drive_refresh_token,
token_uri="https://oauth2.googleapis.com/token",
client_id=settings.google_drive_client_id,
client_secret=settings.google_drive_client_secret
)
# Force a refresh to update the token expiration
if not credentials.valid:
credentials.refresh(Request())
# Get token expiration info
expiration_info = {}
if hasattr(credentials, 'expiry') and credentials.expiry:
now = datetime.now()
expiry = credentials.expiry
time_left = expiry - now
expiration_info = {
"expires_at": expiry.isoformat(),
"expires_in_seconds": max(0, int(time_left.total_seconds())),
"expires_in_human": format_time_remaining(time_left)
}
# Test basic API operation
about = service.about().get(fields="user").execute()
user_email = about.get("user", {}).get("emailAddress", "Unknown")
logger.info(f"Successfully connected to Google Drive as {user_email}")
return {
"status": "success",
"message": f"OAuth token is valid! Connected as {user_email}",
"account": user_email,
"auth_type": "oauth",
"token_info": expiration_info
}
except Exception as e:
error_msg = str(e)
logger.error(f"Google Drive OAuth token test failed: {error_msg}")
# Check if this is a token-related error
if "invalid_grant" in error_msg.lower() or "token" in error_msg.lower():
return {
"status": "error",
"message": f"OAuth token validation failed: {error_msg}",
"needs_reauth": True
}
return {
"status": "error",
"message": f"Connection error: {error_msg}"
}
else:
# Test service account connection
if not settings.google_drive_credentials_json:
logger.warning("Google Drive service account credentials not configured")
return {
"status": "error",
"message": "Google Drive service account credentials are not configured"
}
try:
service = get_google_drive_service()
about = service.about().get(fields="user").execute()
# For service accounts, try to show the delegated user if available
user_email = about.get("user", {}).get("emailAddress", "Unknown")
delegated_user = getattr(settings, 'google_drive_delegate_to', None)
if delegated_user:
user_display = f"{user_email} (delegating as {delegated_user})"
else:
user_display = user_email
logger.info(f"Successfully connected to Google Drive using service account as {user_display}")
return {
"status": "success",
"message": f"Service account is valid! Connected as {user_display}",
"account": user_email,
"auth_type": "service_account"
}
except Exception as e:
error_msg = str(e)
logger.error(f"Google Drive service account test failed: {error_msg}")
return {
"status": "error",
"message": f"Service account validation failed: {error_msg}"
}
except Exception as e:
logger.exception("Unexpected error testing Google Drive token")
return {
"status": "error",
"message": f"Unexpected error: {str(e)}"
}
@router.get("/google-drive/get-token-info")
@require_login
async def get_google_drive_token_info(request: Request):
"""
Get information about the current Google Drive token.
Returns the access token if one exists and is valid.
Used by the frontend to access the Google Picker API.
"""
try:
logger.info("Getting Google Drive token information")
# Check if OAuth is enabled and configured
if not getattr(settings, 'google_drive_use_oauth', False):
logger.warning("OAuth is not enabled, using service account instead")
return {
"status": "error",
"message": "OAuth is not enabled. Service accounts don't support user-facing features like folder picker."
}
if not (settings.google_drive_client_id and
settings.google_drive_client_secret and
settings.google_drive_refresh_token):
logger.warning("Google Drive OAuth credentials not fully configured")
return {
"status": "error",
"message": "Google Drive OAuth credentials are not fully configured"
}
try:
# Get credentials and access token
import google.oauth2.credentials
from google.auth.transport.requests import Request
credentials = google.oauth2.credentials.Credentials(
token=None,
refresh_token=settings.google_drive_refresh_token,
token_uri="https://oauth2.googleapis.com/token",
client_id=settings.google_drive_client_id,
client_secret=settings.google_drive_client_secret
)
# Force a refresh to get a fresh access token
if not credentials.valid:
credentials.refresh(Request())
# Get token expiration info
expiration_info = {}
if hasattr(credentials, 'expiry') and credentials.expiry:
now = datetime.now()
expiry = credentials.expiry
time_left = expiry - now
expiration_info = {
"expires_at": expiry.isoformat(),
"expires_in_seconds": max(0, int(time_left.total_seconds())),
"expires_in_human": format_time_remaining(time_left)
}
# Return the token info
logger.info("Successfully retrieved Google Drive access token")
return {
"status": "success",
"message": "Access token successfully retrieved",
"access_token": credentials.token,
"token_info": expiration_info
}
except Exception as e:
error_msg = str(e)
logger.error(f"Failed to get Google Drive token: {error_msg}")
# Check if this is a token-related error
if "invalid_grant" in error_msg.lower() or "token" in error_msg.lower():
return {
"status": "error",
"message": f"OAuth token retrieval failed: {error_msg}",
"needs_reauth": True
}
return {
"status": "error",
"message": f"Token retrieval error: {error_msg}"
}
except Exception as e:
logger.exception("Unexpected error getting Google Drive token info")
return {
"status": "error",
"message": f"Unexpected error: {str(e)}"
}
def format_time_remaining(time_delta):
"""Format a timedelta into a human-readable string."""
if time_delta.total_seconds() <= 0:
return "Expired"
days = time_delta.days
hours, remainder = divmod(time_delta.seconds, 3600)
minutes, seconds = divmod(remainder, 60)
parts = []
if days > 0:
parts.append(f"{days} day{'s' if days != 1 else ''}")
if hours > 0:
parts.append(f"{hours} hour{'s' if hours != 1 else ''}")
if minutes > 0 and days == 0: # Only show minutes if less than a day
parts.append(f"{minutes} minute{'s' if minutes != 1 else ''}")
return ", ".join(parts)
@router.post("/google-drive/save-settings")
@require_login
async def save_dropbox_settings(
request: Request,
client_id: str = Form(None),
client_secret: str = Form(None),
refresh_token: str = Form(...),
folder_id: str = Form(None),
use_oauth: str = Form("true")
):
"""
Save Google Drive settings to the .env file
"""
try:
# Get the path to the .env file
env_path = os.path.join(os.path.dirname(os.path.dirname(os.path.dirname(__file__))), ".env")
# Convert use_oauth string to boolean
use_oauth_bool = use_oauth.lower() in ("true", "1", "yes", "y", "t")
# Define settings to update
drive_settings = {
"GOOGLE_DRIVE_USE_OAUTH": str(use_oauth_bool).lower()
}
# Only update these if provided
if use_oauth_bool:
if refresh_token:
drive_settings["GOOGLE_DRIVE_REFRESH_TOKEN"] = refresh_token
if client_id:
drive_settings["GOOGLE_DRIVE_CLIENT_ID"] = client_id
if client_secret:
drive_settings["GOOGLE_DRIVE_CLIENT_SECRET"] = client_secret
# Always include folder ID if provided
if folder_id:
drive_settings["GOOGLE_DRIVE_FOLDER_ID"] = folder_id
# Try to update the .env file, but don't fail if it doesn't exist (for Docker containers)
if os.path.exists(env_path):
try:
logger.info(f"Updating Google Drive settings in {env_path}")
# Read the current .env file
with open(env_path, "r") as f:
env_lines = f.readlines()
# Process each line and update or add settings
updated = set()
new_env_lines = []
for line in env_lines:
line = line.rstrip()
is_updated = False
for key, value in drive_settings.items():
if line.startswith(f"{key}=") or line.startswith(f"# {key}="):
if line.startswith("# "): # Uncomment if commented out
line = line[2:]
new_env_lines.append(f"{key}={value}")
updated.add(key)
is_updated = True
break
if not is_updated:
new_env_lines.append(line)
# Add any settings that weren't updated (they weren't in the file)
for key, value in drive_settings.items():
if key not in updated:
new_env_lines.append(f"{key}={value}")
# Write the updated .env file
with open(env_path, "w") as f:
f.write("\n".join(new_env_lines) + "\n")
logger.info("Successfully updated Google Drive settings in .env file")
except Exception as e:
logger.warning(f"Failed to update .env file: {str(e)}, but will continue with in-memory update")
else:
logger.warning(f".env file not found at {env_path}, skipping file update but continuing with in-memory update")
# Update the settings in memory (this always happens)
if refresh_token:
settings.google_drive_refresh_token = refresh_token
if client_id:
settings.google_drive_client_id = client_id
if client_secret:
settings.google_drive_client_secret = client_secret
if folder_id:
settings.google_drive_folder_id = folder_id
# Set OAuth flag
settings.google_drive_use_oauth = use_oauth_bool
logger.info("Successfully updated Google Drive settings in memory")
return {
"status": "success",
"message": "Google Drive settings have been saved",
"in_memory_only": not os.path.exists(env_path)
}
except Exception as e:
logger.exception(f"Unexpected error saving Google Drive settings: {str(e)}")
raise HTTPException(
status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
detail=f"Failed to save Google Drive settings: {str(e)}"
)
app/api/onedrive.py
+127 -37
View File
@@ -5,6 +5,9 @@ from fastapi import APIRouter, Request, HTTPException, status, Form
import logging
import os
import requests
import json
from datetime import datetime, timedelta
from typing import Optional
from app.auth import require_login
from app.config import settings
@@ -112,60 +115,147 @@ async def exchange_onedrive_token(
@require_login
async def test_onedrive_token(request: Request):
"""
Test if the configured OneDrive refresh token is valid.
Provides detailed error information if token is invalid.
Test if the configured OneDrive token is valid and return expiration information.
"""
try:
from app.tasks.upload_to_onedrive import get_onedrive_token
logger.info("Testing OneDrive token validity")
if not settings.onedrive_refresh_token:
logger.warning("No OneDrive refresh token configured")
if not settings.onedrive_refresh_token or not settings.onedrive_client_id or not settings.onedrive_client_secret:
logger.warning("OneDrive credentials not fully configured")
return {
"status": "error",
"message": "No OneDrive refresh token is configured"
"message": "OneDrive credentials are not fully configured"
}
# Check if client ID and client secret are configured
if not settings.onedrive_client_id or not settings.onedrive_client_secret:
logger.warning("OneDrive client ID or client secret is missing")
return {
"status": "error",
"message": "OneDrive client ID or client secret is missing",
"missing_config": True
}
# Refresh token to get a new access token and expiration info
tenant_id = settings.onedrive_tenant_id or "common"
token_url = f"https://login.microsoftonline.com/{tenant_id}/oauth2/v2.0/token"
# Try to get an access token using the configured refresh token
try:
access_token = get_onedrive_token()
# If we got here, token is valid
logger.info("OneDrive token is valid")
return {
"status": "success",
"message": "OneDrive token is valid",
}
except Exception as e:
error_msg = str(e)
logger.error(f"OneDrive token test failed: {error_msg}")
# Determine if this is an invalid_grant error (expired token)
is_expired = "invalid_grant" in error_msg.lower()
refresh_data = {
"client_id": settings.onedrive_client_id,
"client_secret": settings.onedrive_client_secret,
"refresh_token": settings.onedrive_refresh_token,
"grant_type": "refresh_token",
"scope": "offline_access Files.ReadWrite"
}
response = requests.post(token_url, data=refresh_data)
if response.status_code != 200:
logger.error(f"Failed to refresh OneDrive token: {response.text}")
return {
"status": "error",
"message": f"Token validation failed: {error_msg}",
"is_expired": is_expired,
"message": "Refresh token has expired or is invalid",
"needs_reauth": True
}
token_data = response.json()
access_token = token_data.get("access_token")
expires_in = token_data.get("expires_in", 3600) # Default to 1 hour if not specified
# Check if we got a new refresh token (Microsoft sometimes issues a new one)
new_refresh_token = token_data.get("refresh_token")
if new_refresh_token and new_refresh_token != settings.onedrive_refresh_token:
logger.info("Received new refresh token from Microsoft - will update configuration")
# Update refresh token in memory
settings.onedrive_refresh_token = new_refresh_token
# Also try to update .env file if it exists
try:
env_path = os.path.join(os.path.dirname(os.path.dirname(os.path.dirname(__file__))), ".env")
if os.path.exists(env_path):
with open(env_path, "r") as f:
env_lines = f.readlines()
updated_lines = []
updated = False
for line in env_lines:
if line.startswith("ONEDRIVE_REFRESH_TOKEN="):
updated_lines.append(f"ONEDRIVE_REFRESH_TOKEN={new_refresh_token}\n")
updated = True
else:
updated_lines.append(line)
if not updated:
updated_lines.append(f"ONEDRIVE_REFRESH_TOKEN={new_refresh_token}\n")
with open(env_path, "w") as f:
f.writelines(updated_lines)
logger.info("Updated refresh token in .env file")
except Exception as e:
logger.warning(f"Failed to update refresh token in .env file: {e}")
# Test the access token by getting user information
user_info_url = "https://graph.microsoft.com/v1.0/me"
headers = {"Authorization": f"Bearer {access_token}"}
user_response = requests.get(user_info_url, headers=headers)
if user_response.status_code != 200:
logger.error(f"OneDrive token test failed: {user_response.status_code} {user_response.text}")
return {
"status": "error",
"message": f"Token validation failed with status {user_response.status_code}: {user_response.text}"
}
# Get user info
user_info = user_response.json()
display_name = user_info.get("displayName", "Unknown user")
email = user_info.get("userPrincipalName", "Unknown email")
# Calculate expiration time
now = datetime.now()
expiry_time = now + timedelta(seconds=expires_in)
# Format expiration info
time_left = expiry_time - now
token_info = {
"expires_at": expiry_time.isoformat(),
"expires_in_seconds": expires_in,
"expires_in_human": format_time_remaining(time_left),
"refresh_token_validity": "Refresh token is valid for 90 days of inactivity"
}
logger.info(f"Successfully connected to OneDrive as {email}")
return {
"status": "success",
"message": f"OneDrive connection successful",
"account": email,
"account_name": display_name,
"token_info": token_info
}
except Exception as e:
logger.exception("Unexpected error testing OneDrive token")
logger.exception(f"Unexpected error testing OneDrive token: {str(e)}")
return {
"status": "error",
"message": f"Unexpected error: {str(e)}"
"message": f"Connection error: {str(e)}"
}
def format_time_remaining(time_delta):
"""Format a timedelta into a human-readable string."""
if time_delta.total_seconds() <= 0:
return "Expired"
days = time_delta.days
hours, remainder = divmod(time_delta.seconds, 3600)
minutes, seconds = divmod(remainder, 60)
parts = []
if days > 0:
parts.append(f"{days} day{'s' if days != 1 else ''}")
if hours > 0:
parts.append(f"{hours} hour{'s' if hours != 1 else ''}")
if minutes > 0 and days == 0: # Only show minutes if less than a day
parts.append(f"{minutes} minute{'s' if minutes != 1 else ''}")
return ", ".join(parts)
@router.post("/onedrive/save-settings")
@require_login
async def save_onedrive_settings(
app/config.py
+6
View File
@@ -64,6 +64,12 @@ class Settings(BaseSettings):
google_drive_credentials_json: Optional[str] = ""
google_drive_folder_id: Optional[str] = ""
google_drive_delegate_to: Optional[str] = "" # Optional delegated user email
# Google Drive OAuth settings
google_drive_use_oauth: bool = False # Default to service account method
google_drive_client_id: Optional[str] = ""
google_drive_client_secret: Optional[str] = ""
google_drive_refresh_token: Optional[str] = ""
# WebDAV settings
webdav_url: Optional[str] = None
app/tasks/send_to_all.py
+83 -10
View File
@@ -14,6 +14,7 @@ from app.tasks.upload_to_sftp import upload_to_sftp
from app.tasks.upload_to_email import upload_to_email
from app.tasks.upload_to_onedrive import upload_to_onedrive
from app.tasks.upload_to_s3 import upload_to_s3
from app.utils.config_validator import get_provider_status
from app.celery_app import celery
logger = logging.getLogger(__name__)
@@ -33,7 +34,16 @@ def _should_upload_to_paperless():
settings.paperless_host)
def _should_upload_to_google_drive():
return settings.google_drive_credentials_json
# Check for OAuth configuration
if getattr(settings, 'google_drive_use_oauth', False):
return (settings.google_drive_client_id and
settings.google_drive_client_secret and
settings.google_drive_refresh_token and
settings.google_drive_folder_id)
# Or check for service account configuration
else:
return (settings.google_drive_credentials_json and
settings.google_drive_folder_id)
def _should_upload_to_webdav():
return (settings.webdav_url and
@@ -59,18 +69,51 @@ def _should_upload_to_email():
def _should_upload_to_onedrive():
return (settings.onedrive_client_id and
settings.onedrive_client_secret and
(settings.onedrive_refresh_token or
(settings.onedrive_tenant_id and settings.onedrive_tenant_id != "common")))
settings.onedrive_refresh_token)
def _should_upload_to_s3():
return (settings.s3_bucket_name and
settings.aws_access_key_id and
settings.aws_secret_access_key)
@celery.task(base=BaseTaskWithRetry)
def send_to_all_destinations(file_path: str):
"""Distribute a file to all configured storage destinations."""
def get_configured_services_from_validator():
"""
Use the config validator to determine which services are configured properly.
Returns a dictionary with service names as keys and boolean values indicating
whether they're properly configured.
"""
providers = get_provider_status()
service_map = {
"Dropbox": "dropbox",
"NextCloud": "nextcloud",
"Paperless-ngx": "paperless",
"Google Drive": "google_drive",
"WebDAV": "webdav",
"FTP Storage": "ftp",
"SFTP Storage": "sftp",
"Email": "email",
"OneDrive": "onedrive",
"S3 Storage": "s3"
}
result = {}
for provider_name, internal_name in service_map.items():
if provider_name in providers:
result[internal_name] = providers[provider_name].get('configured', False)
return result
@celery.task(base=BaseTaskWithRetry)
def send_to_all_destinations(file_path: str, use_validator=True):
"""
Distribute a file to all configured storage destinations.
Args:
file_path: Path to the file to distribute
use_validator: Whether to use the config validator to determine enabled services
(if False, falls back to individual checks)
"""
if not os.path.exists(file_path):
raise FileNotFoundError(f"File not found: {file_path}")
@@ -131,12 +174,42 @@ def send_to_all_destinations(file_path: str):
},
]
# Optionally get configuration status from validator
configured_services = {}
if use_validator:
try:
configured_services = get_configured_services_from_validator()
logger.info(f"Configured services according to validator: {configured_services}")
except Exception as e:
logger.warning(f"Failed to get configuration from validator: {str(e)}")
use_validator = False
# Process each service
for service in services:
if service["should_upload"]():
logger.info(f"Queueing {file_path} for {service['name']} upload")
task = service["upload_func"].delay(file_path)
results[f"{service['name']}_task_id"] = task.id
service_name = service["name"]
# Determine if service is configured
is_configured = False
if use_validator and service_name in configured_services:
is_configured = configured_services[service_name]
logger.debug(f"{service_name} configuration from validator: {is_configured}")
else:
try:
is_configured = service["should_upload"]()
logger.debug(f"{service_name} configuration from function: {is_configured}")
except Exception as e:
logger.error(f"Error checking configuration for {service_name}: {str(e)}")
is_configured = False
# Queue the upload task if service is configured
if is_configured:
logger.info(f"Queueing {file_path} for {service_name} upload")
try:
task = service["upload_func"].delay(file_path)
results[f"{service_name}_task_id"] = task.id
except Exception as e:
logger.error(f"Failed to queue {service_name} task: {str(e)}")
results[f"{service_name}_error"] = str(e)
return {
"status": "Queued",
app/tasks/upload_to_google_drive.py
+46
View File
@@ -8,6 +8,10 @@ import logging
from googleapiclient.discovery import build
from googleapiclient.http import MediaFileUpload
from google.oauth2.service_account import Credentials
from google.oauth2.credentials import Credentials as OAuthCredentials
from google_auth_oauthlib.flow import Flow
from google.auth.transport.requests import Request
from google.auth.exceptions import RefreshError
from app.config import settings
from app.tasks.retry_config import BaseTaskWithRetry
@@ -15,12 +19,54 @@ from app.celery_app import celery
logger = logging.getLogger(__name__)
def get_drive_service_oauth():
"""
Get Google Drive service using OAuth credentials.
Uses saved refresh token to get a new access token.
"""
try:
# Check for required OAuth settings
if not (settings.google_drive_client_id and
settings.google_drive_client_secret and
settings.google_drive_refresh_token):
logger.error("Google Drive OAuth credentials not fully configured")
return None
# Create credentials object from refresh token
credentials = OAuthCredentials(
None, # No access token initially, will be refreshed
refresh_token=settings.google_drive_refresh_token,
token_uri="https://oauth2.googleapis.com/token",
client_id=settings.google_drive_client_id,
client_secret=settings.google_drive_client_secret,
# Use only drive.file scope
scopes=['https://www.googleapis.com/auth/drive.file']
)
# Refresh the access token
credentials.refresh(Request())
# Build and return the service
service = build('drive', 'v3', credentials=credentials)
return service
except RefreshError as e:
logger.error(f"Failed to refresh Google Drive token: {str(e)}")
raise
except Exception as e:
logger.error(f"Failed to authenticate with Google Drive OAuth: {str(e)}")
return None
def get_google_drive_service():
"""
Authenticate with Google Drive API using service account credentials
and return an authorized service object.
"""
try:
# Check if we should use OAuth instead of service account
if getattr(settings, 'google_drive_use_oauth', False):
return get_drive_service_oauth()
# Load service account credentials from settings
if not settings.google_drive_credentials_json:
logger.error("Google Drive credentials not configured")
app/utils/config_validator.py
+18 -630
View File
@@ -1,633 +1,21 @@
#!/usr/bin/env python3
"""
Configuration validation for the application.
This file serves as a backward-compatible interface to the config_validator package.
"""
import os
import socket
import logging
import inspect
from app.config import settings
# Import and re-export all functions from the new package
from app.utils.config_validator.validators import validate_email_config, validate_storage_configs, check_all_configs
from app.utils.config_validator.masking import mask_sensitive_value
from app.utils.config_validator.providers import get_provider_status
from app.utils.config_validator.settings_display import get_settings_for_display, dump_all_settings
logger = logging.getLogger(__name__)
def validate_email_config():
"""Validates email configuration settings"""
issues = []
# Check for required email settings
if not getattr(settings, 'email_host', None):
issues.append("EMAIL_HOST is not configured")
if not getattr(settings, 'email_port', None):
issues.append("EMAIL_PORT is not configured")
# Test SMTP server connectivity if host is configured
if getattr(settings, 'email_host', None) and getattr(settings, 'email_port', None):
try:
# Attempt to resolve the hostname
socket.gethostbyname(settings.email_host)
except socket.gaierror:
issues.append(f"Cannot resolve email host: {settings.email_host}")
# Check for authentication settings
if not getattr(settings, 'email_username', None):
issues.append("EMAIL_USERNAME is not configured")
if not getattr(settings, 'email_password', None):
issues.append("EMAIL_PASSWORD is not configured")
return issues
def validate_storage_configs():
"""Validates configuration for all storage providers"""
issues = {}
# Validate Dropbox config
dropbox_issues = []
if not (getattr(settings, 'dropbox_app_key', None) and
getattr(settings, 'dropbox_app_secret', None) and
getattr(settings, 'dropbox_refresh_token', None)):
dropbox_issues.append("Dropbox credentials are not fully configured")
issues['dropbox'] = dropbox_issues
# Validate Nextcloud config
nextcloud_issues = []
if not (getattr(settings, 'nextcloud_upload_url', None) and
getattr(settings, 'nextcloud_username', None) and
getattr(settings, 'nextcloud_password', None)):
nextcloud_issues.append("Nextcloud credentials are not fully configured")
issues['nextcloud'] = nextcloud_issues
# Validate SFTP config
sftp_issues = []
if not getattr(settings, 'sftp_host', None):
sftp_issues.append("SFTP_HOST is not configured")
sftp_key_path = getattr(settings, 'sftp_private_key', None)
if sftp_key_path and not os.path.exists(sftp_key_path):
sftp_issues.append(f"SFTP_KEY_PATH file not found: {sftp_key_path}")
if not sftp_key_path and not getattr(settings, 'sftp_password', None):
sftp_issues.append("Neither SFTP_KEY_PATH nor SFTP_PASSWORD is configured")
issues['sftp'] = sftp_issues
# Validate Email sending
email_issues = []
if not getattr(settings, 'email_host', None):
email_issues.append("EMAIL_HOST is not configured")
if not getattr(settings, 'email_default_recipient', None):
email_issues.append("EMAIL_DEFAULT_RECIPIENT is not configured")
issues['email'] = email_issues
# Validate S3
s3_issues = []
if not getattr(settings, 's3_bucket_name', None):
s3_issues.append("S3_BUCKET_NAME is not configured")
if not (getattr(settings, 'aws_access_key_id', None) and
getattr(settings, 'aws_secret_access_key', None)):
s3_issues.append("AWS credentials are not configured")
issues['s3'] = s3_issues
# Validate FTP
ftp_issues = []
if not getattr(settings, 'ftp_host', None):
ftp_issues.append("FTP_HOST is not configured")
if not getattr(settings, 'ftp_username', None):
ftp_issues.append("FTP_USERNAME is not configured")
if not getattr(settings, 'ftp_password', None):
ftp_issues.append("FTP_PASSWORD is not configured")
issues['ftp'] = ftp_issues
# Validate WebDAV
webdav_issues = []
if not getattr(settings, 'webdav_url', None):
webdav_issues.append("WEBDAV_URL is not configured")
if not getattr(settings, 'webdav_username', None):
webdav_issues.append("WEBDAV_USERNAME is not configured")
if not getattr(settings, 'webdav_password', None):
webdav_issues.append("WEBDAV_PASSWORD is not configured")
issues['webdav'] = webdav_issues
# Validate Google Drive
gdrive_issues = []
if not getattr(settings, 'google_drive_credentials_json', None):
gdrive_issues.append("GOOGLE_DRIVE_CREDENTIALS_JSON is not configured")
if not getattr(settings, 'google_drive_folder_id', None):
gdrive_issues.append("GOOGLE_DRIVE_FOLDER_ID is not configured")
issues['google_drive'] = gdrive_issues
# Validate Paperless
paperless_issues = []
if not getattr(settings, 'paperless_host', None):
paperless_issues.append("PAPERLESS_HOST is not configured")
if not getattr(settings, 'paperless_ngx_api_token', None):
paperless_issues.append("PAPERLESS_NGX_API_TOKEN is not configured")
issues['paperless'] = paperless_issues
# Validate OneDrive
onedrive_issues = []
if not (getattr(settings, 'onedrive_client_id', None) and
getattr(settings, 'onedrive_client_secret', None) and
getattr(settings, 'onedrive_refresh_token', None)):
onedrive_issues.append("OneDrive credentials are not fully configured")
issues['onedrive'] = onedrive_issues
# Validate Uptime Kuma
uptime_kuma_issues = []
if not getattr(settings, 'uptime_kuma_url', None):
uptime_kuma_issues.append("UPTIME_KUMA_URL is not configured")
issues['uptime_kuma'] = uptime_kuma_issues
return issues
def mask_sensitive_value(value):
"""
Masks sensitive values like API keys in logs and output
"""
# Return masked value for sensitive data
if value and isinstance(value, str) and len(value) > 8:
return value[:4] + "*" * (len(value) - 4)
return value
def get_provider_status():
"""
Returns status information for all configured providers
"""
providers = {}
# Add AI services first
providers["OpenAI"] = {
"name": "OpenAI",
"icon": "fa-brands fa-openai",
"configured": bool(getattr(settings, 'openai_api_key', None) and
str(getattr(settings, 'openai_api_key', '')).startswith('sk-')),
"enabled": True,
"description": "AI-powered document analysis and metadata extraction",
"details": {
"api_key": mask_sensitive_value(getattr(settings, 'openai_api_key', None)),
"base_url": getattr(settings, 'openai_base_url', 'https://api.openai.com/v1'),
"model": getattr(settings, 'openai_model', 'gpt-4')
}
}
providers["Azure AI"] = {
"name": "Azure AI",
"icon": "fa-solid fa-robot",
"configured": bool(getattr(settings, 'azure_ai_key', None) and
getattr(settings, 'azure_endpoint', None)),
"enabled": True,
"description": "Microsoft Azure Document Intelligence",
"details": {
"api_key": mask_sensitive_value(getattr(settings, 'azure_ai_key', None)),
"endpoint": getattr(settings, 'azure_endpoint', 'Not set'),
"region": getattr(settings, 'azure_region', 'Not set')
}
}
# Add Dropbox configuration - alphabetically ordered providers
providers["Dropbox"] = {
"name": "Dropbox",
"icon": "fa-brands fa-dropbox",
"configured": bool(getattr(settings, 'dropbox_app_key', None) and
getattr(settings, 'dropbox_app_secret', None) and
getattr(settings, 'dropbox_refresh_token', None)),
"enabled": True,
"description": "Upload files to Dropbox cloud storage",
"details": {
"folder": getattr(settings, 'dropbox_folder', 'Not set'),
"app_key": getattr(settings, 'dropbox_app_key', 'Not set'),
"app_secret": mask_sensitive_value(getattr(settings, 'dropbox_app_secret', None)),
"refresh_token": mask_sensitive_value(getattr(settings, 'dropbox_refresh_token', None))
}
}
# Add Email configuration
providers["Email"] = {
"name": "Email",
"icon": "fa-solid fa-envelope",
"configured": bool(getattr(settings, 'email_host', None) and
getattr(settings, 'email_default_recipient', None)),
"enabled": True,
"description": "Send documents via email",
"details": {
"host": getattr(settings, 'email_host', 'Not set'),
"port": getattr(settings, 'email_port', 'Not set'),
"username": getattr(settings, 'email_username', 'Not set'),
"password": mask_sensitive_value(getattr(settings, 'email_password', None)),
"use_tls": getattr(settings, 'email_use_tls', 'Not set'),
"sender": getattr(settings, 'email_sender', 'Not set'),
"default_recipient": getattr(settings, 'email_default_recipient', 'Not set')
}
}
# Add FTP configuration to providers
providers["FTP Storage"] = {
"name": "FTP Storage",
"icon": "fa-solid fa-server",
"configured": bool(getattr(settings, 'ftp_host', None) and
getattr(settings, 'ftp_username', None) and
getattr(settings, 'ftp_password', None)),
"enabled": True,
"description": "Upload files to FTP server",
"details": {
"host": getattr(settings, 'ftp_host', 'Not set'),
"port": getattr(settings, 'ftp_port', 'Not set'),
"username": getattr(settings, 'ftp_username', 'Not set'),
"password": mask_sensitive_value(getattr(settings, 'ftp_password', None)),
"folder": getattr(settings, 'ftp_folder', 'Not set'),
"tls": getattr(settings, 'ftp_use_tls', True),
"allow_plaintext": getattr(settings, 'ftp_allow_plaintext', True)
}
}
# Check Google Drive configuration
providers["Google Drive"] = {
"name": "Google Drive",
"icon": "fa-brands fa-google-drive",
"configured": bool(getattr(settings, 'google_drive_credentials_json', None) and
getattr(settings, 'google_drive_folder_id', None)),
"enabled": True,
"description": "Store documents in Google Drive",
"details": {
"credentials_json": mask_sensitive_value(getattr(settings, 'google_drive_credentials_json', None)),
"folder_id": getattr(settings, 'google_drive_folder_id', 'Not set'),
"delegate": getattr(settings, 'google_drive_delegate_to', 'Not set')
}
}
# Check NextCloud configuration
nextcloud_url = getattr(settings, 'nextcloud_upload_url', 'Not set')
# Extract base URL from WebDAV URL (remove the /remote.php part and everything after it)
nextcloud_base_url = nextcloud_url
if nextcloud_url != 'Not set' and nextcloud_url is not None and '/remote.php' in nextcloud_url:
nextcloud_base_url = nextcloud_url.split('/remote.php')[0]
providers["NextCloud"] = {
"name": "NextCloud",
"icon": "fa-solid fa-cloud",
"configured": bool(getattr(settings, 'nextcloud_upload_url', None) and
getattr(settings, 'nextcloud_username', None) and
getattr(settings, 'nextcloud_password', None)),
"enabled": True,
"description": "Store documents in NextCloud",
"details": {
"url": getattr(settings, 'nextcloud_upload_url', 'Not set'),
"base_url": nextcloud_base_url,
"username": getattr(settings, 'nextcloud_username', 'Not set'),
"password": mask_sensitive_value(getattr(settings, 'nextcloud_password', None)),
"folder": getattr(settings, 'nextcloud_folder', 'Not set')
}
}
# Check OneDrive configuration
providers["OneDrive"] = {
"name": "OneDrive",
"icon": "fa-brands fa-microsoft",
"configured": bool(getattr(settings, 'onedrive_client_id', None) and
getattr(settings, 'onedrive_client_secret', None) and
getattr(settings, 'onedrive_refresh_token', None)),
"enabled": True,
"description": "Store documents in Microsoft OneDrive",
"details": {
"client_id": getattr(settings, 'onedrive_client_id', 'Not set'),
"client_secret": mask_sensitive_value(getattr(settings, 'onedrive_client_secret', None)),
"tenant_id": getattr(settings, 'onedrive_tenant_id', 'Not set'),
"refresh_token": mask_sensitive_value(getattr(settings, 'onedrive_refresh_token', None)),
"folder": getattr(settings, 'onedrive_folder_path', 'Not set')
}
}
# Check Paperless configuration
providers["Paperless-ngx"] = {
"name": "Paperless-ngx",
"icon": "fa-solid fa-file-lines",
"configured": bool(getattr(settings, 'paperless_host', None) and
getattr(settings, 'paperless_ngx_api_token', None)),
"enabled": True,
"description": "Document management system for digital archives",
"details": {
"host": getattr(settings, 'paperless_host', 'Not set'),
"api_token": mask_sensitive_value(getattr(settings, 'paperless_ngx_api_token', None))
}
}
# Check S3 configuration
providers["S3 Storage"] = {
"name": "S3 Storage",
"icon": "fa-brands fa-aws",
"configured": bool(getattr(settings, 's3_bucket_name', None) and
getattr(settings, 'aws_access_key_id', None) and
getattr(settings, 'aws_secret_access_key', None)),
"enabled": True,
"description": "Store documents in S3-compatible object storage",
"details": {
"bucket": getattr(settings, 's3_bucket_name', 'Not set'),
"region": getattr(settings, 'aws_region', 'Not set'),
"access_key_id": getattr(settings, 'aws_access_key_id', 'Not set'),
"secret_access_key": mask_sensitive_value(getattr(settings, 'aws_secret_access_key', None)),
"folder_prefix": getattr(settings, 's3_folder_prefix', 'Not set'),
"storage_class": getattr(settings, 's3_storage_class', 'Not set'),
"acl": getattr(settings, 's3_acl', 'Not set')
}
}
# Check SFTP configuration
providers["SFTP Storage"] = {
"name": "SFTP Storage",
"icon": "fa-solid fa-lock",
"configured": bool(getattr(settings, 'sftp_host', None) and
getattr(settings, 'sftp_username', None) and
(getattr(settings, 'sftp_password', None) or
getattr(settings, 'sftp_private_key', None))),
"enabled": True,
"description": "Upload files to SFTP server",
"details": {
"host": getattr(settings, 'sftp_host', 'Not set'),
"port": getattr(settings, 'sftp_port', 'Not set'),
"username": getattr(settings, 'sftp_username', 'Not set'),
"password": mask_sensitive_value(getattr(settings, 'sftp_password', None)),
"private_key": getattr(settings, 'sftp_private_key', 'Not set'),
"private_key_passphrase": mask_sensitive_value(getattr(settings, 'sftp_private_key_passphrase', None)),
"folder": getattr(settings, 'sftp_folder', 'Not set')
}
}
# Add Uptime Kuma configuration
providers["Uptime Kuma"] = {
"name": "Uptime Kuma",
"icon": "fa-solid fa-heart-pulse",
"configured": bool(getattr(settings, 'uptime_kuma_url', None)),
"enabled": True,
"description": "Server monitoring and status page",
"details": {
"url": getattr(settings, 'uptime_kuma_url', 'Not set'),
"ping_interval": getattr(settings, 'uptime_kuma_ping_interval', 'Not set')
}
}
# Check WebDAV configuration
providers["WebDAV"] = {
"name": "WebDAV",
"icon": "fa-solid fa-globe",
"configured": bool(getattr(settings, 'webdav_url', None) and
getattr(settings, 'webdav_username', None) and
getattr(settings, 'webdav_password', None)),
"enabled": True,
"description": "Store documents on WebDAV servers",
"details": {
"url": getattr(settings, 'webdav_url', 'Not set'),
"username": getattr(settings, 'webdav_username', 'Not set'),
"password": mask_sensitive_value(getattr(settings, 'webdav_password', None)),
"folder": getattr(settings, 'webdav_folder', 'Not set'),
"verify_ssl": getattr(settings, 'webdav_verify_ssl', 'Not set')
}
}
return providers
def dump_all_settings():
"""Log all settings values for diagnostic purposes"""
logger.info("--- DUMPING ALL SETTINGS FOR DIAGNOSTIC PURPOSES ---")
for key in dir(settings):
if not key.startswith('_') and not callable(getattr(settings, key)):
value = getattr(settings, key)
# Mask sensitive values in logs
if key.lower().find('password') >= 0 or key.lower().find('secret') >= 0 or key.lower().find('token') >= 0 or key.lower().find('key') >= 0:
if value:
if isinstance(value, str) and len(value) > 10:
visible_start = max(1, len(value) // 3)
visible_end = max(1, len(value) // 4)
value = f"{value[:visible_start]}{'*' * (len(value) - visible_start - visible_end)}{value[-visible_end:]}"
else:
value = f"{value[:2]}{'*' * (len(value) - 4)}{value[-2:]}" if isinstance(value, str) and len(value) > 4 else "****"
logger.info(f"{key}: {value}")
logger.info("--- END OF SETTINGS DUMP ---")
def get_settings_for_display(show_values=False):
"""
Group settings into logical categories and check if they are configured.
Returns a dictionary with categories as keys and lists of setting items as values.
Each setting item is a dict with name, value, and is_configured.
If show_values is False, sensitive values are masked.
"""
# First include system info with version in result
result = {
"System Info": [
{
"name": "App Version",
"value": settings.version,
"is_configured": True
}
]
}
# Define categories and their settings
categories = {
"Core": [
"debug", # Explicitly include debug setting
"external_hostname",
"workdir",
"database_url",
"redis_url",
"gotenberg_url"
],
"Authentication": [
"auth_enabled",
"authentik_client_id",
"authentik_client_secret",
"authentik_config_url"
],
"Email": [
"email_host",
"email_port",
"email_username",
"email_password",
"email_use_tls",
"email_sender",
"email_default_recipient"
],
"IMAP": [
"imap1_host",
"imap1_port",
"imap1_username",
"imap1_password",
"imap1_ssl",
"imap1_poll_interval_minutes",
"imap1_delete_after_process",
"imap2_host",
"imap2_port",
"imap2_username",
"imap2_password",
"imap2_ssl",
"imap2_poll_interval_minutes",
"imap2_delete_after_process"
],
"Dropbox": [
"dropbox_app_key",
"dropbox_app_secret",
"dropbox_folder",
"dropbox_refresh_token"
],
"NextCloud": [
"nextcloud_upload_url",
"nextcloud_username",
"nextcloud_password",
"nextcloud_folder"
],
"Paperless": [
"paperless_host",
"paperless_ngx_api_token"
],
"Google Drive": [
"google_drive_credentials_json",
"google_drive_folder_id",
"google_drive_delegate_to"
],
"OneDrive": [
"onedrive_client_id",
"onedrive_client_secret",
"onedrive_tenant_id",
"onedrive_refresh_token",
"onedrive_folder_path"
],
"WebDAV": [
"webdav_url",
"webdav_username",
"webdav_password",
"webdav_folder",
"webdav_verify_ssl"
],
"SFTP": [
"sftp_host",
"sftp_port",
"sftp_username",
"sftp_password",
"sftp_folder",
"sftp_private_key",
"sftp_private_key_passphrase"
],
"FTP": [
"ftp_host",
"ftp_port",
"ftp_username",
"ftp_password",
"ftp_folder",
"ftp_use_tls",
"ftp_allow_plaintext"
],
"S3/AWS": [
"aws_access_key_id",
"aws_secret_access_key",
"aws_region",
"s3_bucket_name",
"s3_folder_prefix",
"s3_storage_class",
"s3_acl"
],
"AI Services": [
"openai_api_key",
"openai_base_url",
"openai_model",
"azure_ai_key",
"azure_endpoint",
"azure_region"
],
"Monitoring": [
"uptime_kuma_url",
"uptime_kuma_ping_interval"
]
}
# Handle any settings that don't fit into the predefined categories
all_settings = set([key for key in dir(settings)
if not key.startswith('_') and
not callable(getattr(settings, key)) and
key not in ["model_computed_fields", "model_config",
"model_extra", "model_fields",
"model_fields_set"]])
# Ensure 'version' is excluded since we display it separately
all_settings.discard("version")
categorized_settings = set()
for cat_settings in categories.values():
categorized_settings.update(cat_settings)
uncategorized = all_settings - categorized_settings
if uncategorized:
categories["Other"] = list(uncategorized)
# Build the result
for category, setting_keys in categories.items():
items = []
for key in setting_keys:
if hasattr(settings, key):
value = getattr(settings, key)
# List of patterns that indicate sensitive values
sensitive_patterns = [
'password', 'secret', 'token', 'api_key', 'private_key',
'credentials', 'access_key', 'ai_key'
]
# Check if this is a sensitive value that should be masked
is_sensitive = any(
pattern in key.lower() for pattern in sensitive_patterns
)
# Special handling for "auth" to avoid matching prefixes like "authentik"
if not is_sensitive and "auth" in key.lower():
# Only mark as sensitive if "auth" is a standalone word or at the end
# This avoids matching "authentik" as sensitive
parts = key.lower().split('_')
is_sensitive = any(part == "auth" for part in parts) or key.lower().endswith("auth")
# Mask sensitive values regardless of debug mode
# Other values are only hidden if debug mode is off AND show_values is False
if (is_sensitive or not show_values) and value:
if is_sensitive:
value = mask_sensitive_value(value)
# Check if the setting is configured (has a non-None value)
# For boolean settings, consider them configured even if False
is_configured = value is not None
if is_configured and isinstance(value, str):
is_configured = len(value) > 0
items.append({
"name": key,
"value": value,
"is_configured": is_configured
})
if items: # Only add categories that have items
result[category] = items
return result
def check_all_configs():
"""Run all configuration validations and log results"""
logger.info("Validating application configuration...")
# Check if debug is enabled and dump all settings if it is
if hasattr(settings, 'debug') and settings.debug:
dump_all_settings()
# Check email config
email_issues = validate_email_config()
if email_issues:
logger.warning(f"Email configuration issues: {', '.join(email_issues)}")
else:
logger.info("Email configuration OK")
# Check storage configs
storage_issues = validate_storage_configs()
for provider, issues in storage_issues.items():
if issues:
logger.warning(f"{provider.capitalize()} configuration issues: {', '.join(issues)}")
else:
logger.info(f"{provider.capitalize()} configuration OK")
# Return all identified issues
return {
'email': email_issues,
'storage': storage_issues
}
__all__ = [
'validate_email_config',
'validate_storage_configs',
'mask_sensitive_value',
'get_provider_status',
'get_settings_for_display',
'dump_all_settings',
'check_all_configs'
]
app/utils/config_validator/__init__.py
+25
View File
@@ -0,0 +1,25 @@
"""
Configuration validation package for the application.
"""
from app.utils.config_validator.validators import (
validate_email_config,
validate_storage_configs,
check_all_configs
)
from app.utils.config_validator.masking import mask_sensitive_value
from app.utils.config_validator.providers import get_provider_status
from app.utils.config_validator.settings_display import (
get_settings_for_display,
dump_all_settings
)
__all__ = [
'validate_email_config',
'validate_storage_configs',
'mask_sensitive_value',
'get_provider_status',
'get_settings_for_display',
'dump_all_settings',
'check_all_configs'
]
app/utils/config_validator/masking.py
+12
View File
@@ -0,0 +1,12 @@
"""
Module for masking sensitive information in configuration values
"""
def mask_sensitive_value(value):
"""
Masks sensitive values like API keys in logs and output
"""
# Return masked value for sensitive data
if value and isinstance(value, str) and len(value) > 8:
return value[:4] + "*" * (len(value) - 4)
return value
app/utils/config_validator/providers.py
+256
View File
@@ -0,0 +1,256 @@
"""
Module for handling provider status information
"""
from app.config import settings
from app.utils.config_validator.masking import mask_sensitive_value
def get_provider_status():
"""
Returns status information for all configured providers
"""
providers = {}
# Add AI services first
providers["OpenAI"] = {
"name": "OpenAI",
"icon": "fa-brands fa-openai",
"configured": bool(getattr(settings, 'openai_api_key', None) and
str(getattr(settings, 'openai_api_key', '')).startswith('sk-')),
"enabled": True,
"description": "AI-powered document analysis and metadata extraction",
"details": {
"api_key": mask_sensitive_value(getattr(settings, 'openai_api_key', None)),
"base_url": getattr(settings, 'openai_base_url', 'https://api.openai.com/v1'),
"model": getattr(settings, 'openai_model', 'gpt-4')
}
}
providers["Azure AI"] = {
"name": "Azure AI",
"icon": "fa-solid fa-robot",
"configured": bool(getattr(settings, 'azure_ai_key', None) and
getattr(settings, 'azure_endpoint', None)),
"enabled": True,
"description": "Microsoft Azure Document Intelligence",
"details": {
"api_key": mask_sensitive_value(getattr(settings, 'azure_ai_key', None)),
"endpoint": getattr(settings, 'azure_endpoint', 'Not set'),
"region": getattr(settings, 'azure_region', 'Not set')
}
}
# Add Dropbox configuration - alphabetically ordered providers
providers["Dropbox"] = {
"name": "Dropbox",
"icon": "fa-brands fa-dropbox",
"configured": bool(getattr(settings, 'dropbox_app_key', None) and
getattr(settings, 'dropbox_app_secret', None) and
getattr(settings, 'dropbox_refresh_token', None)),
"enabled": True,
"description": "Upload files to Dropbox cloud storage",
"details": {
"folder": getattr(settings, 'dropbox_folder', 'Not set'),
"app_key": getattr(settings, 'dropbox_app_key', 'Not set'),
"app_secret": mask_sensitive_value(getattr(settings, 'dropbox_app_secret', None)),
"refresh_token": mask_sensitive_value(getattr(settings, 'dropbox_refresh_token', None))
}
}
# Add Email configuration
providers["Email"] = {
"name": "Email",
"icon": "fa-solid fa-envelope",
"configured": bool(getattr(settings, 'email_host', None) and
getattr(settings, 'email_default_recipient', None)),
"enabled": True,
"description": "Send documents via email",
"details": {
"host": getattr(settings, 'email_host', 'Not set'),
"port": getattr(settings, 'email_port', 'Not set'),
"username": getattr(settings, 'email_username', 'Not set'),
"password": mask_sensitive_value(getattr(settings, 'email_password', None)),
"use_tls": getattr(settings, 'email_use_tls', 'Not set'),
"sender": getattr(settings, 'email_sender', 'Not set'),
"default_recipient": getattr(settings, 'email_default_recipient', 'Not set')
}
}
# Add FTP configuration to providers
providers["FTP Storage"] = {
"name": "FTP Storage",
"icon": "fa-solid fa-server",
"configured": bool(getattr(settings, 'ftp_host', None) and
getattr(settings, 'ftp_username', None) and
getattr(settings, 'ftp_password', None)),
"enabled": True,
"description": "Upload files to FTP server",
"details": {
"host": getattr(settings, 'ftp_host', 'Not set'),
"port": getattr(settings, 'ftp_port', 'Not set'),
"username": getattr(settings, 'ftp_username', 'Not set'),
"password": mask_sensitive_value(getattr(settings, 'ftp_password', None)),
"folder": getattr(settings, 'ftp_folder', 'Not set'),
"tls": getattr(settings, 'ftp_use_tls', True),
"allow_plaintext": getattr(settings, 'ftp_allow_plaintext', True)
}
}
# Check Google Drive configuration
gdrive_oauth_configured = bool(getattr(settings, 'google_drive_client_id', None) and
getattr(settings, 'google_drive_client_secret', None) and
getattr(settings, 'google_drive_refresh_token', None))
gdrive_sa_configured = bool(getattr(settings, 'google_drive_credentials_json', None))
# Determine if using OAuth or service account
use_oauth = getattr(settings, 'google_drive_use_oauth', False)
is_configured = (use_oauth and gdrive_oauth_configured) or (not use_oauth and gdrive_sa_configured)
providers["Google Drive"] = {
"name": "Google Drive",
"icon": "fa-brands fa-google-drive",
"configured": is_configured and bool(getattr(settings, 'google_drive_folder_id', None)),
"enabled": True,
"description": "Store documents in Google Drive",
"details": {
"auth_type": "OAuth" if use_oauth else "Service Account",
"client_id": getattr(settings, 'google_drive_client_id', 'Not set') if use_oauth else 'N/A',
"client_secret": mask_sensitive_value(getattr(settings, 'google_drive_client_secret', None)) if use_oauth else 'N/A',
"refresh_token": mask_sensitive_value(getattr(settings, 'google_drive_refresh_token', None)) if use_oauth else 'N/A',
"credentials_json": mask_sensitive_value(getattr(settings, 'google_drive_credentials_json', None)) if not use_oauth else 'N/A',
"folder_id": getattr(settings, 'google_drive_folder_id', 'Not set'),
"delegate": getattr(settings, 'google_drive_delegate_to', 'Not set') if not use_oauth else 'N/A'
}
}
# Check NextCloud configuration
nextcloud_url = getattr(settings, 'nextcloud_upload_url', 'Not set')
# Extract base URL from WebDAV URL (remove the /remote.php part and everything after it)
nextcloud_base_url = nextcloud_url
if nextcloud_url != 'Not set' and nextcloud_url is not None and '/remote.php' in nextcloud_url:
nextcloud_base_url = nextcloud_url.split('/remote.php')[0]
providers["NextCloud"] = {
"name": "NextCloud",
"icon": "fa-solid fa-cloud",
"configured": bool(getattr(settings, 'nextcloud_upload_url', None) and
getattr(settings, 'nextcloud_username', None) and
getattr(settings, 'nextcloud_password', None)),
"enabled": True,
"description": "Store documents in NextCloud",
"details": {
"url": getattr(settings, 'nextcloud_upload_url', 'Not set'),
"base_url": nextcloud_base_url,
"username": getattr(settings, 'nextcloud_username', 'Not set'),
"password": mask_sensitive_value(getattr(settings, 'nextcloud_password', None)),
"folder": getattr(settings, 'nextcloud_folder', 'Not set')
}
}
# Check OneDrive configuration
providers["OneDrive"] = {
"name": "OneDrive",
"icon": "fa-brands fa-microsoft",
"configured": bool(getattr(settings, 'onedrive_client_id', None) and
getattr(settings, 'onedrive_client_secret', None) and
getattr(settings, 'onedrive_refresh_token', None)),
"enabled": True,
"description": "Store documents in Microsoft OneDrive",
"details": {
"client_id": getattr(settings, 'onedrive_client_id', 'Not set'),
"client_secret": mask_sensitive_value(getattr(settings, 'onedrive_client_secret', None)),
"tenant_id": getattr(settings, 'onedrive_tenant_id', 'Not set'),
"refresh_token": mask_sensitive_value(getattr(settings, 'onedrive_refresh_token', None)),
"folder": getattr(settings, 'onedrive_folder_path', 'Not set')
}
}
# Check Paperless configuration
providers["Paperless-ngx"] = {
"name": "Paperless-ngx",
"icon": "fa-solid fa-file-lines",
"configured": bool(getattr(settings, 'paperless_host', None) and
getattr(settings, 'paperless_ngx_api_token', None)),
"enabled": True,
"description": "Document management system for digital archives",
"details": {
"host": getattr(settings, 'paperless_host', 'Not set'),
"api_token": mask_sensitive_value(getattr(settings, 'paperless_ngx_api_token', None))
}
}
# Check S3 configuration
providers["S3 Storage"] = {
"name": "S3 Storage",
"icon": "fa-brands fa-aws",
"configured": bool(getattr(settings, 's3_bucket_name', None) and
getattr(settings, 'aws_access_key_id', None) and
getattr(settings, 'aws_secret_access_key', None)),
"enabled": True,
"description": "Store documents in S3-compatible object storage",
"details": {
"bucket": getattr(settings, 's3_bucket_name', 'Not set'),
"region": getattr(settings, 'aws_region', 'Not set'),
"access_key_id": getattr(settings, 'aws_access_key_id', 'Not set'),
"secret_access_key": mask_sensitive_value(getattr(settings, 'aws_secret_access_key', None)),
"folder_prefix": getattr(settings, 's3_folder_prefix', 'Not set'),
"storage_class": getattr(settings, 's3_storage_class', 'Not set'),
"acl": getattr(settings, 's3_acl', 'Not set')
}
}
# Check SFTP configuration
providers["SFTP Storage"] = {
"name": "SFTP Storage",
"icon": "fa-solid fa-lock",
"configured": bool(getattr(settings, 'sftp_host', None) and
getattr(settings, 'sftp_username', None) and
(getattr(settings, 'sftp_password', None) or
getattr(settings, 'sftp_private_key', None))),
"enabled": True,
"description": "Upload files to SFTP server",
"details": {
"host": getattr(settings, 'sftp_host', 'Not set'),
"port": getattr(settings, 'sftp_port', 'Not set'),
"username": getattr(settings, 'sftp_username', 'Not set'),
"password": mask_sensitive_value(getattr(settings, 'sftp_password', None)),
"private_key": getattr(settings, 'sftp_private_key', 'Not set'),
"private_key_passphrase": mask_sensitive_value(getattr(settings, 'sftp_private_key_passphrase', None)),
"folder": getattr(settings, 'sftp_folder', 'Not set')
}
}
# Add Uptime Kuma configuration
providers["Uptime Kuma"] = {
"name": "Uptime Kuma",
"icon": "fa-solid fa-heart-pulse",
"configured": bool(getattr(settings, 'uptime_kuma_url', None)),
"enabled": True,
"description": "Server monitoring and status page",
"details": {
"url": getattr(settings, 'uptime_kuma_url', 'Not set'),
"ping_interval": getattr(settings, 'uptime_kuma_ping_interval', 'Not set')
}
}
# Check WebDAV configuration
providers["WebDAV"] = {
"name": "WebDAV",
"icon": "fa-solid fa-globe",
"configured": bool(getattr(settings, 'webdav_url', None) and
getattr(settings, 'webdav_username', None) and
getattr(settings, 'webdav_password', None)),
"enabled": True,
"description": "Store documents on WebDAV servers",
"details": {
"url": getattr(settings, 'webdav_url', 'Not set'),
"username": getattr(settings, 'webdav_username', 'Not set'),
"password": mask_sensitive_value(getattr(settings, 'webdav_password', None)),
"folder": getattr(settings, 'webdav_folder', 'Not set'),
"verify_ssl": getattr(settings, 'webdav_verify_ssl', 'Not set')
}
}
return providers
app/utils/config_validator/settings_display.py
+240
View File
@@ -0,0 +1,240 @@
"""
Module for displaying and organizing settings information
"""
import logging
from app.config import settings
from app.utils.config_validator.masking import mask_sensitive_value
logger = logging.getLogger(__name__)
def dump_all_settings():
"""Log all settings values for diagnostic purposes"""
logger.info("--- DUMPING ALL SETTINGS FOR DIAGNOSTIC PURPOSES ---")
for key in dir(settings):
if not key.startswith('_') and not callable(getattr(settings, key)):
value = getattr(settings, key)
# Mask sensitive values in logs
if key.lower().find('password') >= 0 or key.lower().find('secret') >= 0 or key.lower().find('token') >= 0 or key.lower().find('key') >= 0:
if value:
if isinstance(value, str) and len(value) > 10:
visible_start = max(1, len(value) // 3)
visible_end = max(1, len(value) // 4)
value = f"{value[:visible_start]}{'*' * (len(value) - visible_start - visible_end)}{value[-visible_end:]}"
else:
value = f"{value[:2]}{'*' * (len(value) - 4)}{value[-2:]}" if isinstance(value, str) and len(value) > 4 else "****"
logger.info(f"{key}: {value}")
logger.info("--- END OF SETTINGS DUMP ---")
def get_settings_for_display(show_values=False):
"""
Group settings into logical categories and check if they are configured.
Returns a dictionary with categories as keys and lists of setting items as values.
Each setting item is a dict with name, value, and is_configured.
If show_values is False, sensitive values are masked.
"""
# First include system info with version in result
result = {
"System Info": [
{
"name": "App Version",
"value": settings.version,
"is_configured": True
},
{
"name": "Build Date",
"value": settings.build_date,
"is_configured": True
}
]
}
# Define categories and their settings
categories = {
"Core": [
"debug", # Explicitly include debug setting
"external_hostname",
"workdir",
"database_url",
"redis_url",
"gotenberg_url",
"allow_file_delete" # Added allow_file_delete to Core settings
],
"Authentication": [
"auth_enabled",
"authentik_client_id",
"authentik_client_secret",
"authentik_config_url"
],
"Email": [
"email_host",
"email_port",
"email_username",
"email_password",
"email_use_tls",
"email_sender",
"email_default_recipient"
],
"IMAP": [
"imap1_host",
"imap1_port",
"imap1_username",
"imap1_password",
"imap1_ssl",
"imap1_poll_interval_minutes",
"imap1_delete_after_process",
"imap2_host",
"imap2_port",
"imap2_username",
"imap2_password",
"imap2_ssl",
"imap2_poll_interval_minutes",
"imap2_delete_after_process"
],
"Dropbox": [
"dropbox_app_key",
"dropbox_app_secret",
"dropbox_folder",
"dropbox_refresh_token"
],
"NextCloud": [
"nextcloud_upload_url",
"nextcloud_username",
"nextcloud_password",
"nextcloud_folder"
],
"Paperless": [
"paperless_host",
"paperless_ngx_api_token"
],
"Google Drive": [
"google_drive_use_oauth",
"google_drive_client_id",
"google_drive_client_secret",
"google_drive_refresh_token",
"google_drive_credentials_json",
"google_drive_folder_id",
"google_drive_delegate_to"
],
"OneDrive": [
"onedrive_client_id",
"onedrive_client_secret",
"onedrive_tenant_id",
"onedrive_refresh_token",
"onedrive_folder_path"
],
"WebDAV": [
"webdav_url",
"webdav_username",
"webdav_password",
"webdav_folder",
"webdav_verify_ssl"
],
"SFTP": [
"sftp_host",
"sftp_port",
"sftp_username",
"sftp_password",
"sftp_folder",
"sftp_private_key",
"sftp_private_key_passphrase"
],
"FTP": [
"ftp_host",
"ftp_port",
"ftp_username",
"ftp_password",
"ftp_folder",
"ftp_use_tls",
"ftp_allow_plaintext"
],
"S3/AWS": [
"aws_access_key_id",
"aws_secret_access_key",
"aws_region",
"s3_bucket_name",
"s3_folder_prefix",
"s3_storage_class",
"s3_acl"
],
"AI Services": [
"openai_api_key",
"openai_base_url",
"openai_model",
"azure_ai_key",
"azure_endpoint",
"azure_region"
],
"Monitoring": [
"uptime_kuma_url",
"uptime_kuma_ping_interval"
]
}
# Handle any settings that don't fit into the predefined categories
all_settings = set([key for key in dir(settings)
if not key.startswith('_') and
not callable(getattr(settings, key)) and
key not in ["model_computed_fields", "model_config",
"model_extra", "model_fields",
"model_fields_set"]])
# Ensure 'version' is excluded since we display it separately
all_settings.discard("version")
categorized_settings = set()
for cat_settings in categories.values():
categorized_settings.update(cat_settings)
uncategorized = all_settings - categorized_settings
if uncategorized:
categories["Other"] = list(uncategorized)
# Build the result
for category, setting_keys in categories.items():
items = []
for key in setting_keys:
if hasattr(settings, key):
value = getattr(settings, key)
# List of patterns that indicate sensitive values
sensitive_patterns = [
'password', 'secret', 'token', 'api_key', 'private_key',
'credentials', 'access_key', 'ai_key'
]
# Check if this is a sensitive value that should be masked
is_sensitive = any(
pattern in key.lower() for pattern in sensitive_patterns
)
# Special handling for "auth" to avoid matching prefixes like "authentik"
if not is_sensitive and "auth" in key.lower():
# Only mark as sensitive if "auth" is a standalone word or at the end
# This avoids matching "authentik" as sensitive
parts = key.lower().split('_')
is_sensitive = any(part == "auth" for part in parts) or key.lower().endswith("auth")
# Mask sensitive values regardless of debug mode
# Other values are only hidden if debug mode is off AND show_values is False
if (is_sensitive or not show_values) and value:
if is_sensitive:
value = mask_sensitive_value(value)
# Check if the setting is configured (has a non-None value)
# For boolean settings, consider them configured even if False
is_configured = value is not None
if is_configured and isinstance(value, str):
is_configured = len(value) > 0
items.append({
"name": key,
"value": value,
"is_configured": is_configured
})
if items: # Only add categories that have items
result[category] = items
return result
app/utils/config_validator/validators.py
+168
View File
@@ -0,0 +1,168 @@
#!/usr/bin/env python3
import os
import socket
import logging
from app.config import settings
logger = logging.getLogger(__name__)
def validate_email_config():
"""Validates email configuration settings"""
issues = []
# Check for required email settings
if not getattr(settings, 'email_host', None):
issues.append("EMAIL_HOST is not configured")
if not getattr(settings, 'email_port', None):
issues.append("EMAIL_PORT is not configured")
# Test SMTP server connectivity if host is configured
if getattr(settings, 'email_host', None) and getattr(settings, 'email_port', None):
try:
# Attempt to resolve the hostname
socket.gethostbyname(settings.email_host)
except socket.gaierror:
issues.append(f"Cannot resolve email host: {settings.email_host}")
# Check for authentication settings
if not getattr(settings, 'email_username', None):
issues.append("EMAIL_USERNAME is not configured")
if not getattr(settings, 'email_password', None):
issues.append("EMAIL_PASSWORD is not configured")
return issues
def validate_storage_configs():
"""Validates configuration for all storage providers"""
issues = {}
# Validate Dropbox config
dropbox_issues = []
if not (getattr(settings, 'dropbox_app_key', None) and
getattr(settings, 'dropbox_app_secret', None) and
getattr(settings, 'dropbox_refresh_token', None)):
dropbox_issues.append("Dropbox credentials are not fully configured")
issues['dropbox'] = dropbox_issues
# Validate Nextcloud config
nextcloud_issues = []
if not (getattr(settings, 'nextcloud_upload_url', None) and
getattr(settings, 'nextcloud_username', None) and
getattr(settings, 'nextcloud_password', None)):
nextcloud_issues.append("Nextcloud credentials are not fully configured")
issues['nextcloud'] = nextcloud_issues
# Validate SFTP config
sftp_issues = []
if not getattr(settings, 'sftp_host', None):
sftp_issues.append("SFTP_HOST is not configured")
sftp_key_path = getattr(settings, 'sftp_private_key', None)
if sftp_key_path and not os.path.exists(sftp_key_path):
sftp_issues.append(f"SFTP_KEY_PATH file not found: {sftp_key_path}")
if not sftp_key_path and not getattr(settings, 'sftp_password', None):
sftp_issues.append("Neither SFTP_KEY_PATH nor SFTP_PASSWORD is configured")
issues['sftp'] = sftp_issues
# Validate Email sending
email_issues = []
if not getattr(settings, 'email_host', None):
email_issues.append("EMAIL_HOST is not configured")
if not getattr(settings, 'email_default_recipient', None):
email_issues.append("EMAIL_DEFAULT_RECIPIENT is not configured")
issues['email'] = email_issues
# Validate S3
s3_issues = []
if not getattr(settings, 's3_bucket_name', None):
s3_issues.append("S3_BUCKET_NAME is not configured")
if not (getattr(settings, 'aws_access_key_id', None) and
getattr(settings, 'aws_secret_access_key', None)):
s3_issues.append("AWS credentials are not configured")
issues['s3'] = s3_issues
# Validate FTP
ftp_issues = []
if not getattr(settings, 'ftp_host', None):
ftp_issues.append("FTP_HOST is not configured")
if not getattr(settings, 'ftp_username', None):
ftp_issues.append("FTP_USERNAME is not configured")
if not getattr(settings, 'ftp_password', None):
ftp_issues.append("FTP_PASSWORD is not configured")
issues['ftp'] = ftp_issues
# Validate WebDAV
webdav_issues = []
if not getattr(settings, 'webdav_url', None):
webdav_issues.append("WEBDAV_URL is not configured")
if not getattr(settings, 'webdav_username', None):
webdav_issues.append("WEBDAV_USERNAME is not configured")
if not getattr(settings, 'webdav_password', None):
webdav_issues.append("WEBDAV_PASSWORD is not configured")
issues['webdav'] = webdav_issues
# Validate Google Drive
gdrive_issues = []
if not getattr(settings, 'google_drive_credentials_json', None):
gdrive_issues.append("GOOGLE_DRIVE_CREDENTIALS_JSON is not configured")
if not getattr(settings, 'google_drive_folder_id', None):
gdrive_issues.append("GOOGLE_DRIVE_FOLDER_ID is not configured")
issues['google_drive'] = gdrive_issues
# Validate Paperless
paperless_issues = []
if not getattr(settings, 'paperless_host', None):
paperless_issues.append("PAPERLESS_HOST is not configured")
if not getattr(settings, 'paperless_ngx_api_token', None):
paperless_issues.append("PAPERLESS_NGX_API_TOKEN is not configured")
issues['paperless'] = paperless_issues
# Validate OneDrive
onedrive_issues = []
if not (getattr(settings, 'onedrive_client_id', None) and
getattr(settings, 'onedrive_client_secret', None) and
getattr(settings, 'onedrive_refresh_token', None)):
onedrive_issues.append("OneDrive credentials are not fully configured")
issues['onedrive'] = onedrive_issues
# Validate Uptime Kuma
uptime_kuma_issues = []
if not getattr(settings, 'uptime_kuma_url', None):
uptime_kuma_issues.append("UPTIME_KUMA_URL is not configured")
issues['uptime_kuma'] = uptime_kuma_issues
return issues
def check_all_configs():
"""Run all configuration validations and log results"""
from app.utils.config_validator.settings_display import dump_all_settings
logger.info("Validating application configuration...")
# Check if debug is enabled and dump all settings if it is
if hasattr(settings, 'debug') and settings.debug:
dump_all_settings()
# Check email config
email_issues = validate_email_config()
if email_issues:
logger.warning(f"Email configuration issues: {', '.join(email_issues)}")
else:
logger.info("Email configuration OK")
# Check storage configs
storage_issues = validate_storage_configs()
for provider, issues in storage_issues.items():
if issues:
logger.warning(f"{provider.capitalize()} configuration issues: {', '.join(issues)}")
else:
logger.info(f"{provider.capitalize()} configuration OK")
# Return all identified issues
return {
'email': email_issues,
'storage': storage_issues
}
app/views/__init__.py
+2
View File
@@ -10,6 +10,7 @@ from app.views.files import router as files_router
from app.views.status import router as status_router
from app.views.dropbox import router as dropbox_router
from app.views.onedrive import router as onedrive_router
from app.views.google_drive import router as google_drive_router
# Create a combined router
router = APIRouter()
@@ -20,3 +21,4 @@ router.include_router(files_router)
router.include_router(status_router)
router.include_router(dropbox_router)
router.include_router(onedrive_router)
router.include_router(google_drive_router)
app/views/google_drive.py
+118
View File
@@ -0,0 +1,118 @@
"""
Google Drive integration views for setup and OAuth callback.
"""
from fastapi import Request
from fastapi.responses import RedirectResponse
import urllib.parse
from app.views.base import APIRouter, templates, require_login, settings
router = APIRouter()
@router.get("/google-drive-setup")
@require_login
async def google_drive_setup_page(request: Request):
"""
Setup page for the Google Drive integration.
Shows configuration status and setup instructions.
"""
# Check if using OAuth
use_oauth = getattr(settings, 'google_drive_use_oauth', False)
# Check Google Drive OAuth configuration
oauth_configured = bool(settings.google_drive_client_id and
settings.google_drive_client_secret and
settings.google_drive_refresh_token)
# Check Google Drive service account configuration
sa_configured = bool(settings.google_drive_credentials_json)
# Overall configuration status
is_configured = (use_oauth and oauth_configured) or (not use_oauth and sa_configured)
if settings.google_drive_folder_id:
is_configured = is_configured and True
else:
is_configured = False
# Get configuration values to display status (hide sensitive values)
return templates.TemplateResponse(
"google_drive.html",
{
"request": request,
"is_configured": is_configured,
"use_oauth": use_oauth,
"oauth_configured": oauth_configured,
"sa_configured": sa_configured,
"client_id": bool(settings.google_drive_client_id),
"client_id_value": settings.google_drive_client_id or "",
"client_secret": bool(settings.google_drive_client_secret),
"client_secret_value": settings.google_drive_client_secret or "",
"refresh_token": bool(settings.google_drive_refresh_token),
"refresh_token_value": settings.google_drive_refresh_token or "",
"folder_id": settings.google_drive_folder_id or "",
"has_credentials_json": bool(settings.google_drive_credentials_json)
}
)
@router.get("/google-drive-callback")
@require_login
async def google_drive_callback(request: Request, code: str = None, error: str = None, state: str = None):
"""
Callback endpoint for Google Drive OAuth flow.
Now automatically exchanges the code for a token and saves it to the configuration.
"""
if error:
return templates.TemplateResponse(
"google_drive_callback_error.html",
{"request": request, "error": error}
)
if not code:
return templates.TemplateResponse(
"google_drive_callback_error.html",
{"request": request, "error": "No authorization code received from Google"}
)
# Display the processing page with automatic token exchange
return templates.TemplateResponse(
"google_drive_callback.html",
{
"request": request,
"code": code,
"state": state
}
)
@router.get("/google-drive-auth-start")
@require_login
async def google_drive_auth_start(
request: Request,
client_id: str,
redirect_uri: str = None
):
"""
Start the Google Drive OAuth flow by redirecting to Google's authorization page.
"""
if not redirect_uri:
redirect_uri = f"{request.url.scheme}://{request.url.netloc}/google-drive-callback"
# Create the authorization URL with required scopes
# Use only drive.file scope to minimize required permissions
scopes = [
"https://www.googleapis.com/auth/drive.file" # Access to files created or opened by the app
]
scope_str = urllib.parse.quote(' '.join(scopes))
auth_url = (
f"https://accounts.google.com/o/oauth2/auth"
f"?client_id={client_id}"
f"&redirect_uri={urllib.parse.quote(redirect_uri)}"
f"&response_type=code"
f"&scope={scope_str}"
f"&access_type=offline"
f"&prompt=consent" # Force to show consent screen to get refresh token
)
return RedirectResponse(url=auth_url)