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

Merge branch 'main' into refactor/audit-logs-annotated-b008-6104828608171783200

Browse Source
This commit is contained in:
Christian Krakau-Louis
2026-03-16 12:10:32 +01:00
committed by GitHub
parent 8ad90d7da9 0f312160bc
commit e10f0bff42
193 changed files with 189978 additions and 42248 deletions
Download Patch File Download Diff File Expand all files Collapse all files
app/api/dropbox.py
+49 -46
View File
@@ -6,7 +6,7 @@ import logging
import os
from typing import Annotated, Optional
import requests
import httpx
from fastapi import APIRouter, Depends, Form, HTTPException, Request, status
from sqlalchemy.orm import Session
@@ -132,57 +132,60 @@ async def test_dropbox_token(request: Request):
"message": "Dropbox credentials are not fully configured",
}
# 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,
timeout=settings.http_request_timeout,
)
# 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, timeout=settings.http_request_timeout)
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(
async with httpx.AsyncClient() as client:
# Check token validity by getting current account info
headers = {"Authorization": f"Bearer {settings.dropbox_refresh_token}"}
response = await client.post(
"https://api.dropboxapi.com/2/users/get_current_account",
headers=headers,
timeout=settings.http_request_timeout,
)
if response.status_code != 200:
logger.error(f"Dropbox token test failed: {response.status_code} {response.text}")
return {
"status": "error",
"message": f"Token validation failed with status {response.status_code}: {response.text}",
}
# If token is invalid, try refreshing it
if response.status_code == 401:
logger.info("Dropbox access token invalid or expired, trying to refresh")
# Get account info
account_info = response.json()
# 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 = await client.post(
refresh_url, data=refresh_data, timeout=settings.http_request_timeout
)
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 = await client.post(
"https://api.dropboxapi.com/2/users/get_current_account",
headers=headers,
timeout=settings.http_request_timeout,
)
if response.status_code != 200:
logger.error(f"Dropbox token test failed: {response.status_code} {response.text}")
return {
"status": "error",
"message": f"Token validation failed with status {response.status_code}: {response.text}",
}
# 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")
app/api/duplicates.py
+28 -23
View File
@@ -73,33 +73,38 @@ def list_duplicate_groups(
groups = []
total_duplicate_files = 0
for filehash in dup_hashes:
# Find the original (non-duplicate) record with this hash
original = (
db.query(FileRecord)
.filter(FileRecord.filehash == filehash, FileRecord.is_duplicate.is_(False))
.order_by(FileRecord.id.asc())
.first()
if dup_hashes:
# Fetch all matching files (both original and duplicates) in a single batch query
all_records = (
db.query(FileRecord).filter(FileRecord.filehash.in_(dup_hashes)).order_by(FileRecord.id.asc()).all()
)
# Find all duplicate records for this hash
duplicates = (
db.query(FileRecord)
.filter(FileRecord.filehash == filehash, FileRecord.is_duplicate.is_(True))
.order_by(FileRecord.id.asc())
.all()
)
# Group records by hash
originals_by_hash = {}
duplicates_by_hash = {h: [] for h in dup_hashes}
total_duplicate_files += len(duplicates)
for record in all_records:
h = record.filehash
if not record.is_duplicate:
# Store only the first original record per hash, matching the old .first() behaviour
if h not in originals_by_hash:
originals_by_hash[h] = record
else:
duplicates_by_hash[h].append(record)
total_duplicate_files += 1
groups.append(
{
"filehash": filehash,
"original": _file_record_to_dict(original) if original else None,
"duplicates": [_file_record_to_dict(d) for d in duplicates],
"duplicate_count": len(duplicates),
}
)
for filehash in dup_hashes:
original = originals_by_hash.get(filehash)
duplicates = duplicates_by_hash.get(filehash, [])
groups.append(
{
"filehash": filehash,
"original": _file_record_to_dict(original) if original else None,
"duplicates": [_file_record_to_dict(d) for d in duplicates],
"duplicate_count": len(duplicates),
}
)
total_pages = (total_groups + per_page - 1) // per_page if total_groups > 0 else 1
app/api/files.py
+4 -3
View File
@@ -11,6 +11,7 @@ import zipfile
from datetime import datetime, timezone
from typing import Annotated, List, Optional
import aiofiles
from fastapi import APIRouter, Depends, File, HTTPException, Query, Request, UploadFile, status
from fastapi.responses import StreamingResponse
from sqlalchemy import asc, desc
@@ -1277,7 +1278,7 @@ async def ui_upload(request: Request, db: DbSession, file: UploadFile = File(...
# enforcing the size limit during the read so memory usage stays bounded.
try:
written_size = 0
with open(target_path, "wb") as f:
async with aiofiles.open(target_path, "wb") as f:
chunk_size = 65536 # 64 KB chunks
while True:
chunk = await file.read(chunk_size)
@@ -1286,14 +1287,14 @@ async def ui_upload(request: Request, db: DbSession, file: UploadFile = File(...
written_size += len(chunk)
if written_size > max_size:
# Exceeded limit mid-stream; clean up and reject
f.close()
await f.close()
os.remove(target_path)
raise HTTPException(
status_code=413,
detail=f"File too large: exceeded {max_size} bytes during upload. "
f"See SECURITY_AUDIT.md for configuration details.",
)
f.write(chunk)
await f.write(chunk)
except HTTPException:
raise
except Exception as e:
app/api/google_drive.py
+1 -1
View File
@@ -363,7 +363,7 @@ def format_time_remaining(time_delta):
@router.post("/google-drive/save-settings")
@require_login
async def save_dropbox_settings(
async def save_google_drive_settings(
request: Request,
refresh_token: Annotated[str, Form(...)],
client_id: Annotated[Optional[str], Form()] = None,
app/api/integrations.py
+4 -9
View File
@@ -564,7 +564,6 @@ def _test_webdav_connection(config: dict[str, Any] | None, credentials: dict[str
return {"success": False, "message": "Missing required field: url"}
# Only allow http/https to prevent file:// or other custom scheme attacks
import ipaddress
from urllib.parse import urlparse
parsed = urlparse(url)
@@ -574,14 +573,10 @@ def _test_webdav_connection(config: dict[str, Any] | None, credentials: dict[str
# Block requests to private/internal IPs to prevent SSRF
hostname = parsed.hostname or ""
if hostname:
try:
addr = ipaddress.ip_address(hostname)
if addr.is_private or addr.is_loopback or addr.is_link_local:
return {"success": False, "message": "URLs pointing to internal or private networks are not allowed"}
except ValueError:
# Hostname is not an IP literal — allow DNS names through
if hostname in ("localhost", "localhost.localdomain"):
return {"success": False, "message": "URLs pointing to localhost are not allowed"}
from app.utils.network import is_private_ip
if is_private_ip(hostname):
return {"success": False, "message": "URLs pointing to internal or private networks are not allowed"}
try:
import base64
app/api/notifications.py
+9 -10
View File
@@ -452,17 +452,16 @@ async def update_preferences(
)
try:
# Pre-fetch existing preferences for this user to avoid N+1 queries
existing_prefs = (
db.query(UserNotificationPreference).filter(UserNotificationPreference.owner_id == owner_id).all()
)
# Build a fast lookup dictionary keyed by (event_type, channel_type, target_id)
prefs_dict = {(pref.event_type, pref.channel_type, pref.target_id): pref for pref in existing_prefs}
for item in body.preferences:
existing = (
db.query(UserNotificationPreference)
.filter(
UserNotificationPreference.owner_id == owner_id,
UserNotificationPreference.event_type == item.event_type,
UserNotificationPreference.channel_type == item.channel_type,
UserNotificationPreference.target_id == item.target_id,
)
.first()
)
existing = prefs_dict.get((item.event_type, item.channel_type, item.target_id))
if existing:
existing.is_enabled = item.is_enabled
else:
app/api/onedrive.py
+19 -93
View File
@@ -3,7 +3,6 @@ OneDrive API endpoints
"""
import logging
import os
from datetime import datetime, timedelta
from typing import Annotated, Optional
@@ -14,6 +13,7 @@ from sqlalchemy.orm import Session
from app.auth import require_login
from app.config import settings
from app.database import get_db
from app.utils.env_utils import update_env_file
from app.utils.oauth_helper import exchange_oauth_token
from app.utils.settings_service import save_setting_to_db
from app.utils.settings_sync import notify_settings_updated
@@ -115,32 +115,7 @@ async def test_onedrive_token(request: Request):
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}")
update_env_file({"ONEDRIVE_REFRESH_TOKEN": new_refresh_token})
# Persist the rotated refresh token to the database
try:
@@ -246,75 +221,26 @@ async def save_onedrive_settings(
user.get("preferred_username") or user.get("username") or user.get("email") or user.get("id") or "wizard"
)
# Best-effort .env file write
try:
env_path = os.path.join(os.path.dirname(os.path.dirname(os.path.dirname(__file__))), ".env")
if not os.path.exists(env_path):
logger.warning(f".env file not found at {env_path}, skipping file write")
else:
logger.info(f"Updating OneDrive settings in {env_path}")
# Build settings dictionary mapped to database/memory keys
onedrive_settings = {
"onedrive_refresh_token": refresh_token,
"onedrive_client_id": client_id,
"onedrive_client_secret": client_secret,
"onedrive_tenant_id": tenant_id,
"onedrive_folder_path": folder_path,
}
with open(env_path, "r") as f:
env_lines = f.readlines()
# Filter out None values
onedrive_settings = {k: v for k, v in onedrive_settings.items() if v is not None}
onedrive_settings = {"ONEDRIVE_REFRESH_TOKEN": refresh_token}
if client_id:
onedrive_settings["ONEDRIVE_CLIENT_ID"] = client_id
if client_secret:
onedrive_settings["ONEDRIVE_CLIENT_SECRET"] = client_secret
if tenant_id:
onedrive_settings["ONEDRIVE_TENANT_ID"] = tenant_id
if folder_path:
onedrive_settings["ONEDRIVE_FOLDER_PATH"] = folder_path
# Best-effort .env file write using the new utility
env_settings = {k.upper(): v for k, v in onedrive_settings.items()}
update_env_file(env_settings)
updated = set()
new_env_lines = []
for line in env_lines:
stripped_line = line.rstrip()
is_updated = False
for key, value in onedrive_settings.items():
if stripped_line.startswith(f"{key}=") or stripped_line.startswith(f"# {key}="):
new_env_lines.append(f"{key}={value}")
updated.add(key)
is_updated = True
break
if not is_updated:
new_env_lines.append(stripped_line)
for key, value in onedrive_settings.items():
if key not in updated:
new_env_lines.append(f"{key}={value}")
with open(env_path, "w") as f:
f.write("\n".join(new_env_lines) + "\n")
logger.info("Successfully updated OneDrive settings in .env file")
except Exception as env_err:
logger.warning(f"Failed to write .env file (non-fatal): {env_err}")
# Update the settings in memory
if refresh_token:
settings.onedrive_refresh_token = refresh_token
if client_id:
settings.onedrive_client_id = client_id
if client_secret:
settings.onedrive_client_secret = client_secret
if tenant_id:
settings.onedrive_tenant_id = tenant_id
if folder_path:
settings.onedrive_folder_path = folder_path
# Persist to database (primary)
if refresh_token:
save_setting_to_db(db, "onedrive_refresh_token", refresh_token, changed_by=changed_by)
if client_id:
save_setting_to_db(db, "onedrive_client_id", client_id, changed_by=changed_by)
if client_secret:
save_setting_to_db(db, "onedrive_client_secret", client_secret, changed_by=changed_by)
if tenant_id:
save_setting_to_db(db, "onedrive_tenant_id", tenant_id, changed_by=changed_by)
if folder_path:
save_setting_to_db(db, "onedrive_folder_path", folder_path, changed_by=changed_by)
# Update in-memory settings and persist to database dynamically
for key, value in onedrive_settings.items():
setattr(settings, key, value)
save_setting_to_db(db, key, value, changed_by=changed_by)
notify_settings_updated()
app/api/plans.py
+11 -2
View File
@@ -196,11 +196,20 @@ def seed_plans(db: DbSession, _admin: AdminUser) -> dict[str, Any]:
def reorder_plans(body: ReorderBody, db: DbSession, _admin: AdminUser) -> dict[str, Any]:
"""Update sort_order for each plan_id in *body.order* (position = index in list)."""
updated = 0
for sort_order, plan_id in enumerate(body.order):
plan = db.query(SubscriptionPlan).filter(SubscriptionPlan.plan_id == plan_id).first()
# Fetch all requested plans in a single query to avoid N+1
plan_ids = body.order
plans = db.query(SubscriptionPlan).filter(SubscriptionPlan.plan_id.in_(plan_ids)).all()
# Build a map for fast O(1) lookup
plan_map = {p.plan_id: p for p in plans}
for sort_order, plan_id in enumerate(plan_ids):
plan = plan_map.get(plan_id)
if plan:
plan.sort_order = sort_order
updated += 1
try:
db.commit()
except Exception:
app/api/profile.py
+22 -2
View File
@@ -18,7 +18,7 @@ import logging
from hashlib import md5
from typing import Annotated
from fastapi import APIRouter, Depends, File, HTTPException, Request, UploadFile, status
from fastapi import APIRouter, Depends, File, HTTPException, Request, Response, UploadFile, status
from pydantic import BaseModel, Field
from sqlalchemy.orm import Session
@@ -156,7 +156,9 @@ async def get_profile(request: Request, db: DbSession) -> ProfileResponse:
@router.patch("", response_model=ProfileResponse)
@require_login
async def update_profile(body: ProfileUpdateRequest, request: Request, db: DbSession) -> ProfileResponse:
async def update_profile(
body: ProfileUpdateRequest, request: Request, response: Response, db: DbSession
) -> ProfileResponse:
"""Update the current user's editable profile settings."""
user_id = _get_user_id(request)
profile = _get_or_create_profile(db, user_id)
@@ -171,6 +173,24 @@ async def update_profile(body: ProfileUpdateRequest, request: Request, db: DbSes
)
profile.preferred_language = lang or None # type: ignore[assignment]
# Keep session and cookie in sync so detect_language() picks up
# the new preference immediately (without a DB round-trip).
if hasattr(request, "session"):
if lang:
request.session["preferred_language"] = lang
else:
request.session.pop("preferred_language", None)
if lang:
response.set_cookie(
key="docuelevate_lang",
value=lang,
max_age=30 * 24 * 60 * 60,
httponly=False,
samesite="lax",
)
else:
response.delete_cookie(key="docuelevate_lang")
# Validate theme
if body.preferred_theme is not None:
theme = body.preferred_theme.lower().strip()
app/api/shared_links.py
+7 -5
View File
@@ -313,16 +313,18 @@ async def list_shared_links(
active_only: bool = Query(False, description="When true, only return active (non-revoked) links"),
) -> list[dict[str, Any]]:
"""List all shared links created by the authenticated user."""
q = db.query(SharedLink).filter(SharedLink.owner_id == owner_id)
q = (
db.query(SharedLink, FileRecord.original_filename)
.outerjoin(FileRecord, SharedLink.file_id == FileRecord.id)
.filter(SharedLink.owner_id == owner_id)
)
if active_only:
q = q.filter(SharedLink.is_active.is_(True))
links = q.order_by(SharedLink.created_at.desc()).all()
links_with_filenames = q.order_by(SharedLink.created_at.desc()).all()
base_url = str(request.base_url).rstrip("/")
result = []
for link in links:
file_record = db.query(FileRecord).filter(FileRecord.id == link.file_id).first()
filename = file_record.original_filename if file_record else None
for link, filename in links_with_filenames:
result.append(_link_to_dict(link, base_url, filename))
return result
app/api/url_upload.py
+1 -32
View File
@@ -2,7 +2,6 @@
API endpoint for processing files from URLs
"""
import ipaddress
import logging
import mimetypes
import os
@@ -19,6 +18,7 @@ from app.config import settings
from app.tasks.process_document import process_document
from app.utils.allowed_types import ALLOWED_MIME_TYPES
from app.utils.filename_utils import sanitize_filename
from app.utils.network import is_private_ip
# Set up logging
logger = logging.getLogger(__name__)
@@ -42,37 +42,6 @@ class URLUploadRequest(BaseModel):
return v
def is_private_ip(hostname: str) -> bool:
"""
Check if a hostname resolves to a private/internal IP address.
Protects against SSRF attacks by blocking access to internal networks.
"""
try:
# Try to parse as IP address directly
ip = ipaddress.ip_address(hostname)
return ip.is_private or ip.is_loopback or ip.is_link_local or ip.is_reserved
except ValueError:
# Not a direct IP, try to resolve hostname
try:
import socket
# Get all IP addresses for this hostname
addr_info = socket.getaddrinfo(hostname, None)
for info in addr_info:
ip_str = info[4][0]
ip = ipaddress.ip_address(ip_str)
# Block if ANY resolved IP is private/internal
if ip.is_private or ip.is_loopback or ip.is_link_local or ip.is_reserved:
return True
return False
except (socket.gaierror, socket.error):
# Cannot resolve - allow for testing/development
# In production, DNS should work properly
# Log this for debugging
logger.warning(f"Could not resolve hostname: {hostname}")
return False # Changed from True to False to allow external domains in tests
def validate_url_safety(url: str) -> None:
"""
Validate that URL is safe to fetch (SSRF protection).
app/auth.py
+256 -6
View File
@@ -4,12 +4,13 @@ import logging
import pathlib
from datetime import datetime, timezone
from functools import wraps
from urllib.parse import urlparse
from urllib.parse import urlencode, urlparse
from authlib.integrations.starlette_client import OAuth
from fastapi import APIRouter, Depends, Request, status
from fastapi.responses import JSONResponse
from fastapi.templating import Jinja2Templates
from sqlalchemy import func
from sqlalchemy.orm import Session
from starlette.responses import RedirectResponse
@@ -124,8 +125,17 @@ def get_current_user(request: Request):
# Check for Bearer token auth first (API tokens)
api_user = getattr(request.state, "api_token_user", None)
if isinstance(api_user, dict):
logger.debug("[AUTH] get_current_user: resolved from API token (user_id=%s)", api_user.get("id"))
return api_user
return request.session.get("user")
session_user = request.session.get("user")
if session_user:
logger.debug(
"[AUTH] get_current_user: resolved from session (user=%s)",
session_user.get("preferred_username") or session_user.get("email") or session_user.get("id"),
)
else:
logger.debug("[AUTH] get_current_user: no user in session or API token")
return session_user
def _resolve_bearer_user(request: Request, db: Session) -> dict | None:
@@ -140,10 +150,12 @@ def _resolve_bearer_user(request: Request, db: Session) -> dict | None:
"""
auth_header = request.headers.get("authorization", "")
if not isinstance(auth_header, str) or not auth_header.startswith("Bearer "):
logger.debug("[AUTH] _resolve_bearer_user: no Bearer token in Authorization header")
return None
raw_token = auth_header[7:]
if not raw_token or not isinstance(raw_token, str):
logger.debug("[AUTH] _resolve_bearer_user: empty or invalid token after 'Bearer ' prefix")
return None
from app.api.api_tokens import hash_token
@@ -152,8 +164,15 @@ def _resolve_bearer_user(request: Request, db: Session) -> dict | None:
token_hash = hash_token(raw_token)
db_token = db.query(ApiToken).filter(ApiToken.token_hash == token_hash, ApiToken.is_active.is_(True)).first()
if db_token is None:
logger.debug("[AUTH] _resolve_bearer_user: no active API token matched the provided hash")
return None
logger.debug(
"[AUTH] _resolve_bearer_user: matched API token id=%s owner=%s",
db_token.id,
db_token.owner_id,
)
# Update usage tracking
try:
db_token.last_used_at = datetime.now(timezone.utc)
@@ -204,16 +223,18 @@ def require_login(func):
@wraps(func)
async def wrapper(request: Request, *args, **kwargs):
url_path = urlparse(str(request.url)).path
# Check session auth first
if request.session.get("user"):
logger.debug("[AUTH] require_login: session auth OK for %s", url_path)
if inspect.iscoroutinefunction(func):
return await func(*args, request=request, **kwargs)
else:
return func(*args, request=request, **kwargs)
# Fall back to Bearer token auth for API endpoints
url_path = urlparse(str(request.url)).path
if url_path.startswith("/api/"):
logger.debug("[AUTH] require_login: no session, trying Bearer token for %s", url_path)
try:
from app.database import SessionLocal
@@ -227,17 +248,22 @@ def require_login(func):
if api_user:
request.state.api_token_user = api_user
logger.debug(
"[AUTH] require_login: Bearer token auth OK for %s (user=%s)", url_path, api_user.get("id")
)
if inspect.iscoroutinefunction(func):
return await func(*args, request=request, **kwargs)
else:
return func(*args, request=request, **kwargs)
logger.debug("[AUTH] require_login: no valid auth for API endpoint %s — returning 401", url_path)
return JSONResponse(
status_code=status.HTTP_401_UNAUTHORIZED,
content={"error": "Not authenticated"},
)
# Non-API endpoint with no session — redirect to login
logger.debug("[AUTH] require_login: no session for %s — redirecting to /login", url_path)
request.session["redirect_after_login"] = str(request.url)
return RedirectResponse(url="/login", status_code=status.HTTP_302_FOUND)
@@ -254,6 +280,38 @@ def get_gravatar_url(email):
async def login(request: Request):
"""Show login page with appropriate authentication options."""
# Persist the mobile deep-link redirect URI in the session so it survives
# the OAuth provider round-trip and is available when auth completes.
# Accepted schemes:
# • "docuelevate://" — production / EAS builds (custom app scheme)
# • "exp://" — Expo Go development client
# Only custom (non-HTTP) schemes are accepted to prevent open-redirect abuse.
_MOBILE_ALLOWED_SCHEMES = ("docuelevate://", "exp://")
if request.query_params.get("mobile") == "1":
redirect_uri = request.query_params.get("redirect_uri", "")
logger.debug(
"[MOBILE] Login page opened with mobile=1: redirect_uri=%r client_ip=%s",
redirect_uri,
get_client_ip(request),
)
if any(redirect_uri.startswith(s) for s in _MOBILE_ALLOWED_SCHEMES):
request.session["mobile_redirect_uri"] = redirect_uri
logger.info(
"[MOBILE] Mobile redirect URI stored in session: %r",
redirect_uri,
)
else:
logger.warning(
"[MOBILE] Rejected redirect_uri with disallowed scheme: %r (allowed: %s)",
redirect_uri,
", ".join(_MOBILE_ALLOWED_SCHEMES),
)
else:
logger.debug(
"[MOBILE] Login page opened without mobile=1 (standard browser flow) client_ip=%s",
get_client_ip(request),
)
return templates.TemplateResponse(
"login.html",
{
@@ -274,9 +332,15 @@ async def login(request: Request):
async def oauth_login(request: Request):
"""Handle OAuth login flow"""
if not OAUTH_CONFIGURED:
logger.debug("[AUTH] oauth_login: OAuth not configured — redirecting to /login")
return RedirectResponse(url="/login?error=OAuth+not+configured", status_code=status.HTTP_302_FOUND)
redirect_uri = request.url_for("oauth_callback")
logger.debug(
"[AUTH] oauth_login: initiating Authentik OAuth redirect_uri=%s session_keys=%s",
redirect_uri,
list(request.session.keys()),
)
return await oauth.authentik.authorize_redirect(request, redirect_uri)
@@ -291,13 +355,23 @@ async def social_login(request: Request, provider: str):
A redirect to the provider's authorization page, or back to /login on error.
"""
if provider not in SOCIAL_PROVIDERS:
logger.debug(
"[AUTH] social_login: unknown provider=%r (registered=%s)", provider, list(SOCIAL_PROVIDERS.keys())
)
return RedirectResponse(url="/login?error=Unknown+social+provider", status_code=status.HTTP_302_FOUND)
redirect_uri = request.url_for("social_callback", provider=provider)
oauth_client = getattr(oauth, provider, None)
if oauth_client is None:
logger.debug("[AUTH] social_login: provider=%r registered but OAuth client not configured", provider)
return RedirectResponse(url="/login?error=Provider+not+configured", status_code=status.HTTP_302_FOUND)
logger.debug(
"[AUTH] social_login: initiating %s OAuth, redirect_uri=%s session_keys=%s",
provider,
redirect_uri,
list(request.session.keys()),
)
return await oauth_client.authorize_redirect(request, redirect_uri)
@@ -356,27 +430,39 @@ async def social_callback(request: Request, provider: str, db: Session = Depends
A redirect to the user's original destination or the upload page.
"""
if provider not in SOCIAL_PROVIDERS:
logger.debug("[AUTH] social_callback: unknown provider=%r", provider)
return RedirectResponse(url="/login?error=Unknown+social+provider", status_code=status.HTTP_302_FOUND)
oauth_client = getattr(oauth, provider, None)
if oauth_client is None:
logger.debug("[AUTH] social_callback: provider=%r not configured", provider)
return RedirectResponse(url="/login?error=Provider+not+configured", status_code=status.HTTP_302_FOUND)
try:
logger.debug("[AUTH] social_callback: exchanging auth code for provider=%s", provider)
token = await oauth_client.authorize_access_token(request)
# Try standard OIDC userinfo first, fall back to token-embedded userinfo
raw_userinfo = token.get("userinfo")
if not raw_userinfo:
logger.debug("[AUTH] social_callback: no userinfo in token, fetching from userinfo endpoint")
try:
resp = await oauth_client.userinfo(token=token)
raw_userinfo = resp if isinstance(resp, dict) else resp.json() if hasattr(resp, "json") else {}
except Exception:
logger.debug("[AUTH] social_callback: userinfo endpoint failed, using empty dict", exc_info=True)
raw_userinfo = {}
user_data = _normalize_social_userinfo(provider, token, raw_userinfo)
logger.debug(
"[AUTH] social_callback: normalized user_data email=%s sub=%s provider=%s",
user_data.get("email"),
user_data.get("sub"),
provider,
)
if not user_data.get("email"):
logger.debug("[AUTH] social_callback: no email in user_data — aborting")
return RedirectResponse(
url="/login?error=Could+not+retrieve+email+from+provider",
status_code=status.HTTP_302_FOUND,
@@ -407,17 +493,31 @@ async def social_callback(request: Request, provider: str, db: Session = Depends
user_id = (
user_data.get("sub") or user_data.get("preferred_username") or user_data.get("email") or user_data.get("id")
)
# Mobile app flow: issue an inline API token and redirect back to the app.
logger.debug(
"[MOBILE] social_callback: checking for mobile redirect (session has mobile_redirect_uri=%s)",
"mobile_redirect_uri" in request.session,
)
mobile_resp = _create_mobile_redirect(request, db)
if mobile_resp:
logger.info("[MOBILE] social_callback: returning mobile redirect response for provider=%s", provider)
return mobile_resp
if user_id:
profile = db.query(_UserProfile).filter(_UserProfile.user_id == user_id).first()
if profile and not profile.onboarding_completed:
logger.debug("[AUTH] social_callback: user=%s needs onboarding, redirecting", user_id)
post_onboarding = request.session.pop("redirect_after_login", "/upload")
request.session["post_onboarding_redirect"] = post_onboarding
return RedirectResponse(url="/onboarding", status_code=status.HTTP_302_FOUND)
redirect_url = request.session.pop("redirect_after_login", "/upload")
logger.debug("[AUTH] social_callback: login complete, redirecting to %s", redirect_url)
return RedirectResponse(url=redirect_url, status_code=status.HTTP_302_FOUND)
except Exception as e:
logger.warning("[SECURITY] SOCIAL_LOGIN_FAILURE provider=%s error=%s", provider, type(e).__name__)
logger.debug("[AUTH] social_callback: full exception for provider=%s", provider, exc_info=True)
return RedirectResponse(
url="/login?error=Social+login+failed.+Please+try+again.", status_code=status.HTTP_302_FOUND
)
@@ -522,15 +622,23 @@ def _ensure_user_profile(db: Session, user_data: dict, is_admin: bool = False) -
async def oauth_callback(request: Request, db: Session = Depends(get_db)):
"""Handle OAuth callback from provider"""
try:
logger.debug("[AUTH] oauth_callback: exchanging authorization code for token")
token = await oauth.authentik.authorize_access_token(request)
userinfo = token.get("userinfo")
if not userinfo:
logger.debug("[AUTH] oauth_callback: no userinfo in token response — aborting")
return RedirectResponse(
url="/login?error=Failed+to+retrieve+user+information", status_code=status.HTTP_302_FOUND
)
# Store user info in session
user_data = dict(userinfo)
logger.debug(
"[AUTH] oauth_callback: received userinfo email=%s sub=%s groups=%s",
user_data.get("email"),
user_data.get("sub"),
user_data.get("groups", []),
)
# Add Gravatar picture if no picture is provided
if not user_data.get("picture") and user_data.get("email"):
@@ -545,6 +653,12 @@ async def oauth_callback(request: Request, db: Session = Depends(get_db)):
groups = user_data.get("groups", [])
admin_group = (settings.admin_group_name or "admin").strip().lower()
is_admin = admin_group in [group.lower() for group in groups]
logger.debug(
"[AUTH] oauth_callback: admin group check — looking for %r in %s → is_admin=%s",
admin_group,
[g.lower() for g in groups],
is_admin,
)
# Set is_admin flag (defaults to False for OAuth users unless they're in admin group)
user_data["is_admin"] = is_admin
@@ -568,18 +682,34 @@ async def oauth_callback(request: Request, db: Session = Depends(get_db)):
user_id = (
user_data.get("sub") or user_data.get("preferred_username") or user_data.get("email") or user_data.get("id")
)
# Mobile app flow: issue an inline API token and redirect back to the app.
# This check runs before onboarding so native-app users are never sent
# to the web-based onboarding wizard.
logger.debug(
"[MOBILE] oauth_callback: checking for mobile redirect (session has mobile_redirect_uri=%s)",
"mobile_redirect_uri" in request.session,
)
mobile_resp = _create_mobile_redirect(request, db)
if mobile_resp:
logger.info("[MOBILE] oauth_callback: returning mobile redirect response")
return mobile_resp
if user_id:
profile = db.query(_UserProfile).filter(_UserProfile.user_id == user_id).first()
if profile and not profile.onboarding_completed:
logger.debug("[AUTH] oauth_callback: user=%s needs onboarding, redirecting", user_id)
post_onboarding = request.session.pop("redirect_after_login", "/upload")
request.session["post_onboarding_redirect"] = post_onboarding
return RedirectResponse(url="/onboarding", status_code=status.HTTP_302_FOUND)
# Redirect to original destination or default
redirect_url = request.session.pop("redirect_after_login", "/upload")
logger.debug("[AUTH] oauth_callback: login complete, redirecting to %s", redirect_url)
return RedirectResponse(url=redirect_url, status_code=status.HTTP_302_FOUND)
except Exception as e:
logger.warning(f"[SECURITY] OAUTH_LOGIN_FAILURE error={type(e).__name__}")
logger.debug("[AUTH] oauth_callback: full exception details", exc_info=True)
return RedirectResponse(url=f"/login?error=Authentication+failed:+{str(e)}", status_code=status.HTTP_302_FOUND)
@@ -625,6 +755,98 @@ def _record_login_event(
logger.debug("Failed to write login audit event for user=%s", username, exc_info=True)
def _create_mobile_redirect(request: Request, db: Session) -> RedirectResponse | None:
"""Generate a mobile API token and return a redirect to the mobile app.
If ``mobile_redirect_uri`` is stored in the session (set when the login
page was opened with ``?mobile=1&redirect_uri=docuelevate://...``), this
function creates a long-lived API token, appends it as a ``?token=``
query parameter to the redirect URI, and returns the redirect so that
``WebBrowser.openAuthSessionAsync`` in the Expo app intercepts the
deep link and stores the token.
Returns ``None`` when the request is not part of a mobile SSO flow.
Args:
request: The current FastAPI request. The ``user`` dict must already
be stored in ``request.session`` before calling this function.
db: Active database session used to persist the new API token.
Returns:
A ``RedirectResponse`` to the deep-link URI with ``?token=<plaintext>``,
or ``None`` if no mobile redirect URI is pending.
"""
mobile_redirect_uri = request.session.pop("mobile_redirect_uri", None)
if not mobile_redirect_uri:
logger.debug("[MOBILE] _create_mobile_redirect: no mobile_redirect_uri in session — skipping mobile flow")
return None
logger.info(
"[MOBILE] _create_mobile_redirect: mobile flow detected, redirect_uri=%r",
mobile_redirect_uri,
)
user = request.session.get("user") or {}
owner_id = user.get("sub") or user.get("preferred_username") or user.get("email") or user.get("id")
logger.debug(
"[MOBILE] Resolving owner_id from session user: sub=%r preferred_username=%r email=%r id=%r → owner_id=%r",
user.get("sub"),
user.get("preferred_username"),
user.get("email"),
user.get("id"),
owner_id,
)
if not owner_id:
logger.warning("Mobile SSO redirect requested but no owner_id could be resolved from session")
return None
# Lazy imports to avoid circular dependency via app.api.__init__
from app.api.api_tokens import generate_api_token, hash_token
from app.models import ApiToken as _ApiToken
plaintext = generate_api_token()
token_hash_value = hash_token(plaintext)
prefix = plaintext[:12]
db_token = _ApiToken(
owner_id=owner_id,
name="Mobile App",
token_hash=token_hash_value,
token_prefix=prefix,
)
try:
db.add(db_token)
db.commit()
except Exception:
db.rollback()
logger.exception("Failed to create mobile API token for owner_id=%s", owner_id)
return None
# Safely append the token as a query parameter, preserving any existing params.
separator = "&" if "?" in mobile_redirect_uri else "?"
redirect_url = f"{mobile_redirect_uri}{separator}{urlencode({'token': plaintext})}"
# Log the full redirect URL at DEBUG so it's visible when debug logging is enabled.
# At INFO level, log a sanitised version (scheme + host only, token prefix only)
# so the plaintext token is never written to persistent info logs.
parsed = urlparse(redirect_url)
existing_params = f"&{parsed.query.replace(f'token={plaintext}', '')}" if parsed.query else ""
sanitised_url = (
f"{parsed.scheme}://{parsed.netloc}{parsed.path}?token={prefix}…[redacted]{existing_params.rstrip('&')}"
)
logger.info(
"[MOBILE] MOBILE_SSO_TOKEN_ISSUED owner=%s token_id=%s redirect_target=%s",
owner_id,
db_token.id,
sanitised_url,
)
logger.debug(
"[MOBILE] Full redirect URL being sent to client: %s",
redirect_url,
)
return RedirectResponse(url=redirect_url, status_code=status.HTTP_302_FOUND)
async def auth(request: Request, db: Session = Depends(get_db)):
"""Handle local username/password authentication.
@@ -659,8 +881,13 @@ async def auth(request: Request, db: Session = Depends(get_db)):
)
return RedirectResponse(url="/login?error=Invalid+username+or+password", status_code=302)
username_lower = username.lower()
local_user = (
db.query(_LocalUser).filter((_LocalUser.username == username) | (_LocalUser.email == username)).first()
db.query(_LocalUser)
.filter(
(func.lower(_LocalUser.username) == username_lower) | (func.lower(_LocalUser.email) == username_lower)
)
.first()
)
logger.debug(
"[AUTH] LocalUser lookup: username=%r found=%s",
@@ -694,6 +921,17 @@ async def auth(request: Request, db: Session = Depends(get_db)):
logger.info("[SECURITY] LOCAL_LOGIN_SUCCESS user=%s", local_user.email)
_record_login_event(db, request, local_user.email, success=True)
_ensure_user_profile(db, user_data, is_admin=bool(local_user.is_admin))
# Mobile app flow: issue an inline API token and redirect back to the app.
logger.debug(
"[MOBILE] local auth: checking for mobile redirect (session has mobile_redirect_uri=%s)",
"mobile_redirect_uri" in request.session,
)
mobile_resp = _create_mobile_redirect(request, db)
if mobile_resp:
logger.info("[MOBILE] local auth: returning mobile redirect response")
return mobile_resp
profile = db.query(_UserProfile).filter(_UserProfile.user_id == local_user.email).first()
if profile and not profile.onboarding_completed:
post_onboarding = request.session.pop("redirect_after_login", "/upload")
@@ -718,13 +956,13 @@ async def auth(request: Request, db: Session = Depends(get_db)):
logger.debug(
"[AUTH] Admin credential check: admin_configured=%s username_match=%s multi_user_enabled=%s",
admin_configured,
username == settings.admin_username if admin_configured else False,
(username or "").lower() == settings.admin_username.lower() if admin_configured else False,
settings.multi_user_enabled,
)
if (
settings.admin_username
and settings.admin_password
and username == settings.admin_username
and (username or "").lower() == settings.admin_username.lower()
and password == settings.admin_password
):
admin_user_data = {
@@ -739,6 +977,17 @@ async def auth(request: Request, db: Session = Depends(get_db)):
logger.info("[SECURITY] LOCAL_LOGIN_SUCCESS user=%s", username)
_record_login_event(db, request, username, success=True)
_ensure_user_profile(db, admin_user_data, is_admin=True)
# Mobile app flow: issue an inline API token and redirect back to the app.
logger.debug(
"[MOBILE] admin auth: checking for mobile redirect (session has mobile_redirect_uri=%s)",
"mobile_redirect_uri" in request.session,
)
mobile_resp = _create_mobile_redirect(request, db)
if mobile_resp:
logger.info("[MOBILE] admin auth: returning mobile redirect response")
return mobile_resp
redirect_url = request.session.pop("redirect_after_login", "/upload")
return RedirectResponse(url=redirect_url, status_code=302)
else:
@@ -760,6 +1009,7 @@ async def logout(request: Request, db: Session = Depends(get_db)):
username = "unknown"
if isinstance(user, dict):
username = user.get("preferred_username") or user.get("email") or "unknown"
logger.debug("[AUTH] logout: clearing session for user=%s client_ip=%s", username, get_client_ip(request))
logger.info(f"[SECURITY] LOGOUT user={username}")
try:
from app.utils.audit_service import record_event
app/config.py
+45
View File
@@ -48,6 +48,51 @@ class Settings(BaseSettings):
workdir: str
debug: bool = False # Default to False
# Logging level for the application. Accepts standard Python level names:
# DEBUG, INFO, WARNING, ERROR, CRITICAL. When *debug* is True and
# *log_level* has not been explicitly set, the effective level is forced to
# DEBUG so that all ``logger.debug()`` calls produce output.
log_level: str = Field(
default="INFO",
description=(
"Python logging level for the application root logger. "
"Accepts: DEBUG, INFO, WARNING, ERROR, CRITICAL. "
"When DEBUG=True and LOG_LEVEL is not explicitly set, "
"the effective level is automatically lowered to DEBUG."
),
)
# Log output format. ``text`` is the human-readable default.
# ``json`` emits one JSON object per line, ideal for log collectors
# (Promtail, Fluentd, Filebeat, Datadog agent) and SIEM ingestion.
log_format: str = Field(
default="text",
description=(
"Log output format: 'text' (human-readable, default) or "
"'json' (structured JSON lines for SIEM / log aggregation)."
),
)
# Optional syslog forwarding for application logs (not just audit events).
# When enabled, a Python SysLogHandler is added to the root logger so that
# every log message is also sent to the configured syslog receiver.
log_syslog_enabled: bool = Field(
default=False,
description="Forward application logs to a syslog receiver in addition to stdout.",
)
log_syslog_host: str = Field(
default="localhost",
description="Hostname or IP of the syslog receiver for application logs.",
)
log_syslog_port: int = Field(
default=514,
description="Port of the syslog receiver for application logs.",
)
log_syslog_protocol: str = Field(
default="udp",
description="Protocol for syslog transport: 'udp' or 'tcp'.",
)
# Making Dropbox optional
dropbox_enabled: bool = Field(
default=True,
app/database.py
+9 -2
View File
@@ -210,8 +210,10 @@ def _run_schema_migrations(engine: Any) -> None:
if unique_filehash_indexes:
logger.info("Migrating files: dropping unique index on 'filehash'")
with engine.begin() as conn:
preparer = conn.dialect.identifier_preparer
for index in unique_filehash_indexes:
conn.execute(text(f"DROP INDEX IF EXISTS {index['name']}"))
quoted_idx = preparer.quote(index["name"])
conn.execute(text(f"DROP INDEX IF EXISTS {quoted_idx}"))
logger.info("Migration complete: unique index on 'filehash' removed")
except Exception as exc:
logger.warning(f"Skipping filehash unique index drop: {exc}")
@@ -263,12 +265,17 @@ def _ensure_indexes(engine: Any, inspector: Any) -> None:
table_names = inspector.get_table_names()
columns_by_table: dict[str, set[str]] = {}
with engine.begin() as conn:
preparer = conn.dialect.identifier_preparer
for idx_name, table, column in _PERF_INDEXES:
if table in table_names:
if table not in columns_by_table:
columns_by_table[table] = {col["name"] for col in inspector.get_columns(table)}
if column in columns_by_table[table]:
conn.execute(text(f"CREATE INDEX IF NOT EXISTS {idx_name} ON {table} ({column})"))
# SECURITY: Quoted identifiers to prevent SQL injection during index creation
quoted_idx = preparer.quote(idx_name)
quoted_table = preparer.quote(table)
quoted_col = preparer.quote(column)
conn.execute(text(f"CREATE INDEX IF NOT EXISTS {quoted_idx} ON {quoted_table} ({quoted_col})"))
logger.info("Performance indexes ensured")
app/main.py
+113
View File
@@ -1,8 +1,11 @@
#!/usr/bin/env python3
import json as _json_mod
import logging
import os
import pathlib
from contextlib import asynccontextmanager
from datetime import datetime as _dt
from datetime import timezone as _tz
from fastapi import FastAPI, HTTPException, Request, status
from fastapi.middleware.cors import CORSMiddleware
@@ -36,6 +39,114 @@ from app.views import router as frontend_router
# Explicitly include the files router
from app.views.files import router as files_router
# ---------------------------------------------------------------------------
# Configure Python root logging level early so that *all* loggers (including
# those already created via ``logging.getLogger(__name__)`` in other modules)
# respect the configured level.
#
# Standard behaviour (matches Django, Flask, 12-factor conventions):
# • ``LOG_LEVEL`` env var takes precedence when explicitly set.
# • When ``DEBUG=True`` and ``LOG_LEVEL`` is **not** set, the effective
# level is automatically lowered to ``DEBUG``.
# • Default (neither flag set): ``INFO``.
#
# ``LOG_FORMAT=json`` enables structured JSON lines on stdout, suitable for
# Promtail, Fluentd, Filebeat, Datadog, Splunk UF, or any log collector.
#
# ``LOG_SYSLOG_ENABLED=true`` adds a Python SysLogHandler so that every log
# message is also forwarded to the configured syslog receiver — useful for
# traditional (non-container) deployments and centralised SIEM ingestion.
#
# Noisy third-party loggers (httpx, httpcore, authlib, etc.) are pinned to
# WARNING when the app-level is DEBUG to keep output useful.
# ---------------------------------------------------------------------------
_explicit_log_level = os.environ.get("LOG_LEVEL")
if settings.debug and _explicit_log_level is None:
_effective_level = "DEBUG"
else:
_effective_level = settings.log_level.upper()
_effective_level_int = getattr(logging, _effective_level, logging.INFO)
class _JsonFormatter(logging.Formatter):
"""Emit one JSON object per log line for machine consumption.
Fields emitted: ``timestamp``, ``level``, ``logger``, ``message``,
``module``, ``funcName``, ``lineno``, and — when present — ``exc_info``.
Compatible with Grafana Loki, Splunk, ELK, Datadog, and most SIEM tools.
"""
def format(self, record: logging.LogRecord) -> str:
log_entry: dict = {
"timestamp": _dt.fromtimestamp(record.created, tz=_tz.utc).isoformat(),
"level": record.levelname,
"logger": record.name,
"message": record.getMessage(),
"module": record.module,
"funcName": record.funcName,
"lineno": record.lineno,
}
if record.exc_info and record.exc_info[1] is not None:
log_entry["exc_info"] = self.formatException(record.exc_info)
return _json_mod.dumps(log_entry, default=str)
# Choose formatter based on LOG_FORMAT setting
if settings.log_format.lower() == "json":
_handler = logging.StreamHandler()
_handler.setFormatter(_JsonFormatter())
logging.root.handlers = [_handler]
logging.root.setLevel(_effective_level_int)
else:
logging.basicConfig(
level=_effective_level_int,
format="%(asctime)s - %(name)s - %(levelname)s - %(message)s",
datefmt="%Y-%m-%d %H:%M:%S",
force=True,
)
# Optional: forward application logs to a syslog receiver
if settings.log_syslog_enabled:
import logging.handlers as _lh
import socket as _socket
_proto = settings.log_syslog_protocol.lower()
_socktype = _socket.SOCK_STREAM if _proto == "tcp" else _socket.SOCK_DGRAM
_syslog_handler = _lh.SysLogHandler(
address=(settings.log_syslog_host, settings.log_syslog_port),
socktype=_socktype,
)
_syslog_handler.setLevel(_effective_level_int)
# Use the same formatter as stdout (text or JSON)
if settings.log_format.lower() == "json":
_syslog_handler.setFormatter(_JsonFormatter())
else:
_syslog_handler.setFormatter(logging.Formatter("%(name)s - %(levelname)s - %(message)s"))
logging.root.addHandler(_syslog_handler)
# Keep noisy third-party loggers quiet at DEBUG level
if _effective_level_int <= logging.DEBUG:
for _noisy in (
"httpx",
"httpcore",
"authlib",
"urllib3",
"hpack",
"multipart",
"watchfiles",
):
logging.getLogger(_noisy).setLevel(logging.WARNING)
_startup_logger = logging.getLogger(__name__)
_startup_logger.info(
"Root logging level set to %s (debug=%s, format=%s, syslog=%s)",
_effective_level,
settings.debug,
settings.log_format,
settings.log_syslog_enabled,
)
# Load configuration from .env for the session key
config = Config(".env")
# Use settings.session_secret which has proper validation
@@ -261,12 +372,14 @@ _error_templates = Jinja2Templates(directory=str(_error_templates_dir))
# Register the i18n translate helper as a global so error templates can use {{ _("key") }}.
# Error pages use the default language (English); request-specific locale is not needed here.
from app.utils.i18n import SUPPORTED_LANGUAGES as _SUPPORTED_LANGUAGES # noqa: E402
from app.utils.i18n import get_suggested_languages as _get_suggested_languages # noqa: E402
from app.utils.i18n import translate as _translate_fn # noqa: E402
_error_templates.env.globals["_"] = lambda key, **kwargs: _translate_fn(key, "en", **kwargs)
_error_templates.env.globals["min"] = min
_error_templates.env.globals["max"] = max
_error_templates.env.globals["supported_languages"] = _SUPPORTED_LANGUAGES
_error_templates.env.globals["suggested_languages"] = _get_suggested_languages("en", "")
@app.exception_handler(HTTPException)
app/tasks/extract_metadata_with_gpt.py
+192 -191
View File
@@ -1,191 +1,192 @@
#!/usr/bin/env python3
import json
import logging
import os
import re
# Import the shared Celery instance
from app.celery_app import celery
from app.config import settings
from app.database import SessionLocal
from app.models import FileRecord
from app.tasks.embed_metadata_into_pdf import embed_metadata_into_pdf
from app.tasks.retry_config import BaseTaskWithRetry
from app.utils import log_task_progress
from app.utils.ai_provider import get_ai_provider
logger = logging.getLogger(__name__)
def extract_json_from_text(text):
"""
Try to extract a JSON object from the text.
- First, check for a JSON block inside triple backticks.
- If not found, try to extract text from the first '{' to the last '}'.
"""
pattern = r"```(?:json)?\s*(\{.*?\})\s*```"
match = re.search(pattern, text, re.DOTALL)
if match:
return match.group(1)
else:
start = text.find("{")
end = text.rfind("}")
if start != -1 and end != -1 and end > start:
return text[start : end + 1]
return None
@celery.task(base=BaseTaskWithRetry, bind=True)
def extract_metadata_with_gpt(self, filename: str, cleaned_text: str, file_id: int = None):
"""
Uses OpenAI to classify document metadata.
Args:
filename: Can be either a basename (e.g., "file.pdf") or a full path (e.g., "/workdir/processed/file.pdf")
cleaned_text: The extracted text from the document
file_id: Optional file ID for tracking
"""
task_id = self.request.id
logger.info(f"[{task_id}] Starting metadata extraction for: {filename}")
log_task_progress(
task_id,
"extract_metadata_with_gpt",
"in_progress",
f"Extracting metadata for {os.path.basename(filename)}",
file_id=file_id,
)
# Get file_id from database if not provided
if file_id is None:
tmp_dir = os.path.join(settings.workdir, "tmp")
# Handle both basename and full path
if os.path.isabs(filename):
file_path = filename
else:
file_path = os.path.join(tmp_dir, filename)
if os.path.exists(file_path):
with SessionLocal() as db:
file_record = db.query(FileRecord).filter_by(local_filename=file_path).first()
if file_record:
file_id = file_record.id
prompt = (
"You are a specialized document analyzer trained to extract structured metadata from documents.\n"
"Your task is to analyze the given text and return a well-structured JSON object.\n\n"
"Extract and return the following fields:\n"
"1. **filename**: Machine-readable filename "
"(YYYY-MM-DD_DescriptiveTitle, use only letters, numbers, periods, and underscores).\n"
'2. **empfaenger**: The recipient, or "Unknown" if not found.\n'
'3. **absender**: The sender, or "Unknown" if not found.\n'
"4. **correspondent**: The entity or company that issued the document "
'(shortest possible name, e.g., "Amazon" instead of "Amazon EU SARL, German branch").\n'
"5. **kommunikationsart**: One of [Behoerdlicher_Brief, Rechnung, Kontoauszug, Vertrag, "
"Quittung, Privater_Brief, Einladung, Gewerbliche_Korrespondenz, Newsletter, Werbung, Sonstiges].\n"
"6. **kommunikationskategorie**: One of [Amtliche_Postbehoerdliche_Dokumente, "
"Finanz_und_Vertragsdokumente, Geschaeftliche_Kommunikation, "
"Private_Korrespondenz, Sonstige_Informationen].\n"
"7. **document_type**: Precise classification (e.g., Invoice, Contract, Information, Unknown).\n"
"8. **tags**: A list of up to 4 relevant thematic keywords.\n"
'9. **language**: Detected document language (ISO 639-1 code, e.g., "de" or "en").\n'
"10. **title**: A human-readable title summarizing the document content.\n"
"11. **confidence_score**: A numeric value (0-100) indicating the confidence level "
"of the extracted metadata.\n"
"12. **reference_number**: Extracted invoice/order/reference number if available.\n"
"13. **monetary_amounts**: A list of key monetary values detected in the document.\n\n"
"### Important Rules:\n"
"- **OCR Correction**: Assume the text has been corrected for OCR errors.\n"
"- **Tagging**: Max 4 tags, avoiding generic or overly specific terms.\n"
"- **Title**: Concise, no addresses, and contains key identifying features.\n"
"- **Date Selection**: Use the most relevant date if multiple are found.\n"
"- **Output Language**: Maintain the document's original language.\n\n"
f"Extracted text:\n{cleaned_text}\n\n"
"Return only valid JSON with no additional commentary.\n"
)
try:
logger.info(f"[{task_id}] Sending classification request for {filename}...")
log_task_progress(task_id, "call_ai_provider", "in_progress", "Calling AI provider API", file_id=file_id)
provider = get_ai_provider()
model = settings.ai_model or settings.openai_model
content = provider.chat_completion(
messages=[
{"role": "system", "content": "You are an intelligent document classifier."},
{"role": "user", "content": prompt},
],
model=model,
temperature=0,
)
logger.info(f"[{task_id}] Raw classification response for {filename}: {content[:200]}...")
log_task_progress(
task_id,
"call_ai_provider",
"success",
"Received AI provider response",
file_id=file_id,
detail=f"Raw classification response:\n{content}",
)
json_text = extract_json_from_text(content)
if not json_text:
logger.error(f"[{task_id}] Could not find valid JSON in GPT response for {filename}.")
log_task_progress(
task_id,
"extract_metadata_with_gpt",
"failure",
"Invalid JSON in response",
file_id=file_id,
detail=f"Could not parse valid JSON from GPT response.\nRaw response:\n{content}",
)
return {}
metadata = json.loads(json_text)
# SECURITY: Validate filename format from GPT to prevent path traversal
# The prompt requests filenames with only letters, numbers, periods, and underscores
# Enforce this constraint to prevent malicious filenames
suggested_filename = metadata.get("filename", "")
if suggested_filename:
# Check if filename contains only safe characters AND explicitly check for ".."
# Defense in depth: While the regex [\w\-\. ]+ already excludes / and \,
# we explicitly reject ".." to guard against:
# 1. Potential locale-specific \w behavior
# 2. Files literally named ".." which are valid but problematic
# 3. Future code changes that might relax the regex
if not re.match(r"^[\w\-\. ]+$", suggested_filename) or ".." in suggested_filename:
logger.warning(f"[{task_id}] Invalid filename format from GPT: '{suggested_filename}', using fallback")
# Reset to empty to trigger fallback to original filename
metadata["filename"] = ""
logger.info(f"[{task_id}] Extracted metadata: {metadata}")
log_task_progress(
task_id,
"parse_metadata",
"success",
f"Parsed metadata: {list(metadata.keys())}",
file_id=file_id,
detail=f"Extracted metadata:\n{json.dumps(metadata, ensure_ascii=False, indent=2)}",
)
# Trigger the next step: embedding metadata into the PDF
# Pass the filename (can be basename or full path) so embed_metadata_into_pdf can find the file on disk
logger.info(f"[{task_id}] Queueing metadata embedding task")
log_task_progress(
task_id, "extract_metadata_with_gpt", "success", "Metadata extracted, queuing embed task", file_id=file_id
)
embed_metadata_into_pdf.delay(filename, cleaned_text, metadata, file_id)
return {"s3_file": os.path.basename(filename), "metadata": metadata}
except Exception as e:
logger.exception(f"[{task_id}] AI provider classification failed for {filename}: {e}")
log_task_progress(
task_id,
"extract_metadata_with_gpt",
"failure",
f"Exception: {str(e)}",
file_id=file_id,
detail=f"AI provider classification failed for {filename}.\nException: {str(e)}",
)
return {}
#!/usr/bin/env python3
import json
import logging
import os
import re
# Import the shared Celery instance
from app.celery_app import celery
from app.config import settings
from app.database import SessionLocal
from app.models import FileRecord
from app.tasks.embed_metadata_into_pdf import embed_metadata_into_pdf
from app.tasks.retry_config import BaseTaskWithRetry
from app.utils import log_task_progress
from app.utils.ai_provider import get_ai_provider
from app.utils.filename_utils import VALID_FILENAME_RE
logger = logging.getLogger(__name__)
def extract_json_from_text(text):
"""
Try to extract a JSON object from the text.
- First, check for a JSON block inside triple backticks.
- If not found, try to extract text from the first '{' to the last '}'.
"""
pattern = r"```(?:json)?\s*(\{.*?\})\s*```"
match = re.search(pattern, text, re.DOTALL)
if match:
return match.group(1)
else:
start = text.find("{")
end = text.rfind("}")
if start != -1 and end != -1 and end > start:
return text[start : end + 1]
return None
@celery.task(base=BaseTaskWithRetry, bind=True)
def extract_metadata_with_gpt(self, filename: str, cleaned_text: str, file_id: int = None):
"""
Uses OpenAI to classify document metadata.
Args:
filename: Can be either a basename (e.g., "file.pdf") or a full path (e.g., "/workdir/processed/file.pdf")
cleaned_text: The extracted text from the document
file_id: Optional file ID for tracking
"""
task_id = self.request.id
logger.info(f"[{task_id}] Starting metadata extraction for: {filename}")
log_task_progress(
task_id,
"extract_metadata_with_gpt",
"in_progress",
f"Extracting metadata for {os.path.basename(filename)}",
file_id=file_id,
)
# Get file_id from database if not provided
if file_id is None:
tmp_dir = os.path.join(settings.workdir, "tmp")
# Handle both basename and full path
if os.path.isabs(filename):
file_path = filename
else:
file_path = os.path.join(tmp_dir, filename)
if os.path.exists(file_path):
with SessionLocal() as db:
file_record = db.query(FileRecord).filter_by(local_filename=file_path).first()
if file_record:
file_id = file_record.id
prompt = (
"You are a specialized document analyzer trained to extract structured metadata from documents.\n"
"Your task is to analyze the given text and return a well-structured JSON object.\n\n"
"Extract and return the following fields:\n"
"1. **filename**: Machine-readable filename "
"(YYYY-MM-DD_DescriptiveTitle, use only letters, numbers, spaces, dashes, periods, and underscores).\n"
'2. **empfaenger**: The recipient, or "Unknown" if not found.\n'
'3. **absender**: The sender, or "Unknown" if not found.\n'
"4. **correspondent**: The entity or company that issued the document "
'(shortest possible name, e.g., "Amazon" instead of "Amazon EU SARL, German branch").\n'
"5. **kommunikationsart**: One of [Behoerdlicher_Brief, Rechnung, Kontoauszug, Vertrag, "
"Quittung, Privater_Brief, Einladung, Gewerbliche_Korrespondenz, Newsletter, Werbung, Sonstiges].\n"
"6. **kommunikationskategorie**: One of [Amtliche_Postbehoerdliche_Dokumente, "
"Finanz_und_Vertragsdokumente, Geschaeftliche_Kommunikation, "
"Private_Korrespondenz, Sonstige_Informationen].\n"
"7. **document_type**: Precise classification (e.g., Invoice, Contract, Information, Unknown).\n"
"8. **tags**: A list of up to 4 relevant thematic keywords.\n"
'9. **language**: Detected document language (ISO 639-1 code, e.g., "de" or "en").\n'
"10. **title**: A human-readable title summarizing the document content.\n"
"11. **confidence_score**: A numeric value (0-100) indicating the confidence level "
"of the extracted metadata.\n"
"12. **reference_number**: Extracted invoice/order/reference number if available.\n"
"13. **monetary_amounts**: A list of key monetary values detected in the document.\n\n"
"### Important Rules:\n"
"- **OCR Correction**: Assume the text has been corrected for OCR errors.\n"
"- **Tagging**: Max 4 tags, avoiding generic or overly specific terms.\n"
"- **Title**: Concise, no addresses, and contains key identifying features.\n"
"- **Date Selection**: Use the most relevant date if multiple are found.\n"
"- **Output Language**: Maintain the document's original language.\n\n"
f"Extracted text:\n{cleaned_text}\n\n"
"Return only valid JSON with no additional commentary.\n"
)
try:
logger.info(f"[{task_id}] Sending classification request for {filename}...")
log_task_progress(task_id, "call_ai_provider", "in_progress", "Calling AI provider API", file_id=file_id)
provider = get_ai_provider()
model = settings.ai_model or settings.openai_model
content = provider.chat_completion(
messages=[
{"role": "system", "content": "You are an intelligent document classifier."},
{"role": "user", "content": prompt},
],
model=model,
temperature=0,
)
logger.info(f"[{task_id}] Raw classification response for {filename}: {content[:200]}...")
log_task_progress(
task_id,
"call_ai_provider",
"success",
"Received AI provider response",
file_id=file_id,
detail=f"Raw classification response:\n{content}",
)
json_text = extract_json_from_text(content)
if not json_text:
logger.error(f"[{task_id}] Could not find valid JSON in GPT response for {filename}.")
log_task_progress(
task_id,
"extract_metadata_with_gpt",
"failure",
"Invalid JSON in response",
file_id=file_id,
detail=f"Could not parse valid JSON from GPT response.\nRaw response:\n{content}",
)
return {}
metadata = json.loads(json_text)
# SECURITY: Validate filename format from GPT to prevent path traversal
# The prompt requests filenames with only letters, numbers, periods, and underscores
# Enforce this constraint to prevent malicious filenames
suggested_filename = metadata.get("filename", "")
if suggested_filename:
# Check if filename contains only safe characters AND explicitly check for ".."
# Defense in depth: While the regex VALID_FILENAME_PATTERN already excludes / and \,
# we explicitly reject ".." to guard against:
# 1. Potential locale-specific \w behavior
# 2. Files literally named ".." which are valid but problematic
# 3. Future code changes that might relax the regex
if not VALID_FILENAME_RE.match(suggested_filename) or ".." in suggested_filename:
logger.warning(f"[{task_id}] Invalid filename format from GPT: '{suggested_filename}', using fallback")
# Reset to empty to trigger fallback to original filename
metadata["filename"] = ""
logger.info(f"[{task_id}] Extracted metadata: {metadata}")
log_task_progress(
task_id,
"parse_metadata",
"success",
f"Parsed metadata: {list(metadata.keys())}",
file_id=file_id,
detail=f"Extracted metadata:\n{json.dumps(metadata, ensure_ascii=False, indent=2)}",
)
# Trigger the next step: embedding metadata into the PDF
# Pass the filename (can be basename or full path) so embed_metadata_into_pdf can find the file on disk
logger.info(f"[{task_id}] Queueing metadata embedding task")
log_task_progress(
task_id, "extract_metadata_with_gpt", "success", "Metadata extracted, queuing embed task", file_id=file_id
)
embed_metadata_into_pdf.delay(filename, cleaned_text, metadata, file_id)
return {"s3_file": os.path.basename(filename), "metadata": metadata}
except Exception as e:
logger.exception(f"[{task_id}] AI provider classification failed for {filename}: {e}")
log_task_progress(
task_id,
"extract_metadata_with_gpt",
"failure",
f"Exception: {str(e)}",
file_id=file_id,
detail=f"AI provider classification failed for {filename}.\nException: {str(e)}",
)
return {}
app/tasks/upload_to_email.py
+17 -2
View File
@@ -11,6 +11,7 @@ from email.mime.image import MIMEImage
from email.mime.multipart import MIMEMultipart
from email.mime.text import MIMEText
import pypdf
from jinja2 import Environment, FileSystemLoader, select_autoescape
from app.celery_app import celery
@@ -80,8 +81,22 @@ def extract_metadata_from_file(file_path):
except Exception as e:
logger.warning(f"Failed to load metadata from JSON file: {str(e)}")
# TODO: For PDF files, try to extract embedded metadata using PyPDF2
# This would require additional dependencies, so for now we'll just check for external JSON
# Try to extract embedded metadata from PDF
if file_path.lower().endswith(".pdf") and os.path.exists(file_path):
try:
with open(file_path, "rb") as f:
pdf_reader = pypdf.PdfReader(f)
pdf_metadata = pdf_reader.metadata
if pdf_metadata:
# Convert metadata to a standard dictionary
for key, value in pdf_metadata.items():
# Remove the leading slash from PDF metadata keys (e.g., '/Title' -> 'Title')
clean_key = key[1:] if key.startswith("/") else key
metadata[clean_key] = str(value)
logger.info(f"Extracted embedded metadata from PDF: {file_path}")
except Exception as e:
logger.warning(f"Failed to extract metadata from PDF {file_path}: {str(e)}")
return metadata
app/tasks/upload_with_rclone.py
+3 -3
View File
@@ -55,12 +55,12 @@ def upload_with_rclone(self, file_path: str, destination: str):
try:
# Ensure the remote path exists (create folders if needed)
mkdir_cmd = ["rclone", "mkdir", "--config", rclone_config_path, destination]
mkdir_cmd = ["rclone", "mkdir", "--config", rclone_config_path, "--", destination]
subprocess.run(mkdir_cmd, check=True, capture_output=True) # noqa: S603
# Construct the upload command
upload_cmd = ["rclone", "copy", "--config", rclone_config_path, file_path, destination, "--progress"]
upload_cmd = ["rclone", "copy", "--config", rclone_config_path, "--progress", "--", file_path, destination]
log_task_progress(task_id, "rclone_upload", "in_progress", f"Executing rclone copy to {destination}")
@@ -71,7 +71,7 @@ def upload_with_rclone(self, file_path: str, destination: str):
if result.returncode == 0:
# Try to get a public link if possible
try:
link_cmd = ["rclone", "link", "--config", rclone_config_path, f"{destination}/{filename}"]
link_cmd = ["rclone", "link", "--config", rclone_config_path, "--", f"{destination}/{filename}"]
link_result = subprocess.run(link_cmd, capture_output=True, text=True, check=False) # noqa: S603
public_url = link_result.stdout.strip() if link_result.returncode == 0 else None
except (subprocess.SubprocessError, OSError) as e:
app/utils/db_migrate.py
+6 -1
View File
@@ -12,6 +12,7 @@ The utility:
"""
import logging
import re
from typing import Any
from sqlalchemy import MetaData, create_engine, inspect, text
@@ -84,8 +85,12 @@ def preview_migration(source_url: str) -> dict[str, Any]:
total = 0
with src_engine.connect() as conn:
for table_name in tables:
if not re.match(r"^[a-zA-Z0-9_]+$", table_name):
logger.warning(f"Skipping table with invalid name format: {table_name}")
continue
# table_name is safe — sourced from inspect().get_table_names(), not user input
row = conn.execute(text(f'SELECT COUNT(*) FROM "{table_name}"')).fetchone() # noqa: S608
quoted_table = conn.dialect.identifier_preparer.quote(table_name)
row = conn.execute(text(f"SELECT COUNT(*) FROM {quoted_table}")).fetchone() # noqa: S608
count = row[0] if row else 0
result.append({"name": table_name, "row_count": count})
total += count
app/utils/env_utils.py
+55
View File
@@ -0,0 +1,55 @@
import logging
import os
from typing import Dict
logger = logging.getLogger(__name__)
def update_env_file(settings_to_update: Dict[str, str]) -> bool:
"""
Updates the .env file with the given settings (best-effort).
Creates or modifies existing keys.
Args:
settings_to_update: A dictionary mapping uppercase env var names to their new string values.
Returns:
True if the file was successfully updated, False otherwise.
"""
try:
env_path = os.path.join(os.path.dirname(os.path.dirname(os.path.dirname(__file__))), ".env")
if not os.path.exists(env_path):
logger.warning(f".env file not found at {env_path}, skipping file write")
return False
logger.info(f"Updating settings in {env_path}")
with open(env_path, "r") as f:
env_lines = f.readlines()
updated = set()
new_env_lines = []
for line in env_lines:
stripped_line = line.rstrip()
is_updated = False
for key, value in settings_to_update.items():
if stripped_line.startswith(f"{key}=") or stripped_line.startswith(f"# {key}="):
new_env_lines.append(f"{key}={value}")
updated.add(key)
is_updated = True
break
if not is_updated:
new_env_lines.append(stripped_line)
for key, value in settings_to_update.items():
if key not in updated:
new_env_lines.append(f"{key}={value}")
with open(env_path, "w") as f:
f.write("\n".join(new_env_lines) + "\n")
logger.info("Successfully updated settings in .env file")
return True
except Exception as env_err:
logger.warning(f"Failed to write .env file (non-fatal): {env_err}")
return False
app/utils/filename_utils.py
+5
View File
@@ -8,6 +8,11 @@ from pathlib import Path
logger = logging.getLogger(__name__)
# Pattern for valid filenames (alphanumeric, dash, underscore, period, and space)
# Used for validating GPT-provided filenames and other inputs
VALID_FILENAME_PATTERN = r"^[\w\-\. ]+$"
VALID_FILENAME_RE = re.compile(VALID_FILENAME_PATTERN)
def get_unique_filename(original_path: str, check_exists_func: Callable[[str], bool] | None = None) -> str:
"""
app/utils/i18n.py
+91 -15
View File
@@ -2,7 +2,7 @@
Provides a JSON-based translation system for the DocuElevate UI with:
* **49 supported languages** covering European, Asian, Middle-Eastern, and other languages
* **77 supported languages** covering European, Asian, Middle-Eastern, African, and other languages
* Browser ``Accept-Language`` detection with cookie & user-profile persistence
* AI-powered fallback translation via the configured LLM provider
* Locale-aware date, number, and file-size formatting helpers
@@ -52,10 +52,10 @@ SUPPORTED_LANGUAGES: list[dict[str, str]] = [
{"code": "lb", "name": "Luxembourgish", "native": "Lëtzebuergesch", "flag": "🇱🇺"},
{"code": "ca", "name": "Catalan", "native": "Català", "flag": "🏴"},
{"code": "cy", "name": "Welsh", "native": "Cymraeg", "flag": "🏴󠁧󠁢󠁷󠁬󠁳󠁿"}, # Wales subdivision flag (U+1F3F4 + tag chars)
{"code": "fy", "name": "Frisian", "native": "Frysk", "flag": "🇳🇱"},
{"code": "fy", "name": "Western Frisian", "native": "Frysk", "flag": "🇳🇱"},
{"code": "gl", "name": "Galician", "native": "Galego", "flag": "🇪🇸"},
{"code": "li", "name": "Limburgish", "native": "Limburgs", "flag": "🇳🇱"},
{"code": "vls", "name": "West Flemish", "native": "West-Vlams", "flag": "🇧🇪"},
{"code": "vls", "name": "Flemish", "native": "West-Vlams", "flag": "🇧🇪"},
{"code": "nds", "name": "Low German", "native": "Plattdüütsch", "flag": "🇩🇪"},
# --- Tier 3: Central & Eastern European ---
{"code": "pl", "name": "Polish", "native": "Polski", "flag": "🇵🇱"},
@@ -74,25 +74,60 @@ SUPPORTED_LANGUAGES: list[dict[str, str]] = [
# --- Tier 4: Non-EU European, Middle Eastern & African ---
{"code": "tr", "name": "Turkish", "native": "Türkçe", "flag": "🇹🇷"},
{"code": "uk", "name": "Ukrainian", "native": "Українська", "flag": "🇺🇦"},
{"code": "ru", "name": "Russian", "native": "Русский", "flag": "🇷🇺"},
{"code": "he", "name": "Hebrew", "native": "עברית", "flag": "🇮🇱"},
{"code": "ar", "name": "Arabic", "native": "العربية", "flag": "🇸🇦"},
{"code": "fa", "name": "Persian", "native": "فارسی", "flag": "🇮🇷"},
{"code": "af", "name": "Afrikaans", "native": "Afrikaans", "flag": "🇿🇦"},
# --- Tier 5: Asian languages ---
{"code": "zh", "name": "Chinese", "native": "中文", "flag": "🇨🇳"},
{"code": "zh-TW", "name": "Traditional Chinese", "native": "繁體中文", "flag": "🇹🇼"},
{"code": "ja", "name": "Japanese", "native": "日本語", "flag": "🇯🇵"},
{"code": "ko", "name": "Korean", "native": "한국어", "flag": "🇰🇷"},
{"code": "vi", "name": "Vietnamese", "native": "Tiếng Việt", "flag": "🇻🇳"},
{"code": "pa", "name": "Punjabi", "native": "ਪੰਜਾਬੀ", "flag": "🇮🇳"},
{"code": "kn", "name": "Kannada", "native": "ಕನ್ನಡ", "flag": "🇮🇳"},
# --- Tier 6: Constructed & other languages ---
{"code": "hi", "name": "Hindi", "native": "हिन्दी", "flag": "🇮🇳"},
{"code": "bn", "name": "Bengali", "native": "বাংলা", "flag": "🇧🇩"},
{"code": "gu", "name": "Gujarati", "native": "ગુજરાતી", "flag": "🇮🇳"},
{"code": "ml", "name": "Malayalam", "native": "മലയാളം", "flag": "🇮🇳"},
{"code": "mr", "name": "Marathi", "native": "मराठी", "flag": "🇮🇳"},
{"code": "ta", "name": "Tamil", "native": "தமிழ்", "flag": "🇮🇳"},
{"code": "te", "name": "Telugu", "native": "తెలుగు", "flag": "🇮🇳"},
{"code": "ur", "name": "Urdu", "native": "اردو", "flag": "🇵🇰"},
{"code": "si", "name": "Sinhala", "native": "සිංහල", "flag": "🇱🇰"},
{"code": "ne", "name": "Nepali", "native": "नेपाली", "flag": "🇳🇵"},
{"code": "th", "name": "Thai", "native": "ไทย", "flag": "🇹🇭"},
{"code": "km", "name": "Khmer", "native": "ខ្មែរ", "flag": "🇰🇭"},
{"code": "id", "name": "Indonesian", "native": "Bahasa Indonesia", "flag": "🇮🇩"},
{"code": "ms", "name": "Malay", "native": "Bahasa Melayu", "flag": "🇲🇾"},
{"code": "jv", "name": "Javanese", "native": "Basa Jawa", "flag": "🇮🇩"},
{"code": "tl", "name": "Tagalog", "native": "Filipino", "flag": "🇵🇭"},
{"code": "mn", "name": "Mongolian", "native": "Монгол", "flag": "🇲🇳"},
{"code": "kk", "name": "Kazakh", "native": "Қазақ тілі", "flag": "🇰🇿"},
{"code": "uz", "name": "Uzbek", "native": "Oʻzbekcha", "flag": "🇺🇿"},
{"code": "az", "name": "Azerbaijani", "native": "Azərbaycan dili", "flag": "🇦🇿"},
{"code": "hy", "name": "Armenian", "native": "Հայերեն", "flag": "🇦🇲"},
{"code": "ka", "name": "Georgian", "native": "ქართული", "flag": "🇬🇪"},
# --- Tier 6: African languages ---
{"code": "sw", "name": "Swahili", "native": "Kiswahili", "flag": "🇰🇪"},
{"code": "am", "name": "Amharic", "native": "አማርኛ", "flag": "🇪🇹"},
{"code": "ha", "name": "Hausa", "native": "Hausa", "flag": "🇳🇬"},
{"code": "yo", "name": "Yoruba", "native": "Yorùbá", "flag": "🇳🇬"},
{"code": "ig", "name": "Igbo", "native": "Igbo", "flag": "🇳🇬"},
{"code": "zu", "name": "Zulu", "native": "isiZulu", "flag": "🇿🇦"},
# --- Tier 7: Constructed & other languages ---
{"code": "eo", "name": "Esperanto", "native": "Esperanto", "flag": "🌍"},
]
SUPPORTED_LANGUAGE_CODES: set[str] = {lang["code"] for lang in SUPPORTED_LANGUAGES}
DEFAULT_LANGUAGE = "en"
# Lookup map for fast code → language-dict resolution
_LANG_CODE_MAP: dict[str, dict[str, str]] = {lang["code"]: lang for lang in SUPPORTED_LANGUAGES}
# Global-usage order used to fill remaining slots in the smart suggestions list
_POPULAR_LANGUAGE_CODES: list[str] = ["en", "zh", "es", "ar", "fr", "de", "ja", "pt", "hi", "ko"]
# ---------------------------------------------------------------------------
# Translation file loading
# ---------------------------------------------------------------------------
@@ -263,14 +298,10 @@ def detect_language(request: Request) -> str:
return DEFAULT_LANGUAGE
def _parse_accept_language(header: str) -> str | None:
"""Extract the best matching language from an ``Accept-Language`` header.
Parses quality values and returns the highest-priority match among
:data:`SUPPORTED_LANGUAGE_CODES`, or ``None`` if nothing matches.
"""
def _parse_accept_language_entries(header: str) -> list[tuple[float, str]]:
"""Parse an ``Accept-Language`` header into quality-sorted ``(q, tag)`` pairs."""
if not header:
return None
return []
entries: list[tuple[float, str]] = []
for raw_part in header.split(","):
@@ -288,11 +319,17 @@ def _parse_accept_language(header: str) -> str | None:
quality = 1.0
entries.append((quality, lang_tag.strip().lower()))
# Sort by quality descending
entries.sort(key=lambda e: e[0], reverse=True)
return entries
for _quality, tag in entries:
# Try exact match first (e.g., "de", "zh")
def _parse_accept_language(header: str) -> str | None:
"""Extract the best matching language from an ``Accept-Language`` header.
Parses quality values and returns the highest-priority match among
:data:`SUPPORTED_LANGUAGE_CODES`, or ``None`` if nothing matches.
"""
for _quality, tag in _parse_accept_language_entries(header):
code = tag.split("-")[0]
if code in SUPPORTED_LANGUAGE_CODES:
return code
@@ -300,6 +337,45 @@ def _parse_accept_language(header: str) -> str | None:
return None
# Maximum number of languages shown in the compact nav-bar dropdown
_SUGGESTED_LANGUAGES_MAX = 6
def get_suggested_languages(current_locale: str, accept_language_header: str = "") -> list[dict[str, str]]:
"""Return up to :data:`_SUGGESTED_LANGUAGES_MAX` suggested languages for the compact picker.
Selection priority:
1. The currently active language (always included first).
2. Languages listed in the browser's ``Accept-Language`` header.
3. Popular global languages (by estimated speaker count) as fillers.
The resulting list is de-duplicated and capped at
:data:`_SUGGESTED_LANGUAGES_MAX` entries.
"""
candidates: list[str] = []
# 1. Active locale first
if current_locale in SUPPORTED_LANGUAGE_CODES:
candidates.append(current_locale)
# 2. Browser preferences
for _quality, tag in _parse_accept_language_entries(accept_language_header):
if len(candidates) >= _SUGGESTED_LANGUAGES_MAX:
break
code = tag.split("-")[0]
if code in SUPPORTED_LANGUAGE_CODES and code not in candidates:
candidates.append(code)
# 3. Popular language fillers
for code in _POPULAR_LANGUAGE_CODES:
if len(candidates) >= _SUGGESTED_LANGUAGES_MAX:
break
if code not in candidates and code in SUPPORTED_LANGUAGE_CODES:
candidates.append(code)
return [_LANG_CODE_MAP[c] for c in candidates[:_SUGGESTED_LANGUAGES_MAX] if c in _LANG_CODE_MAP]
# ---------------------------------------------------------------------------
# Localization helpers (l10n)
# ---------------------------------------------------------------------------
app/utils/network.py
+34
View File
@@ -0,0 +1,34 @@
import ipaddress
import logging
import socket
logger = logging.getLogger(__name__)
def is_private_ip(hostname: str) -> bool:
"""
Check if a hostname resolves to a private/internal IP address.
Protects against SSRF attacks by blocking access to internal networks.
"""
try:
# Try to parse as IP address directly
ip = ipaddress.ip_address(hostname)
return ip.is_private or ip.is_loopback or ip.is_link_local or ip.is_reserved
except ValueError:
# Not a direct IP, try to resolve hostname
try:
# Get all IP addresses for this hostname
addr_info = socket.getaddrinfo(hostname, None)
for info in addr_info:
ip_str = info[4][0]
ip = ipaddress.ip_address(ip_str)
# Block if ANY resolved IP is private/internal
if ip.is_private or ip.is_loopback or ip.is_link_local or ip.is_reserved:
return True
return False
except (socket.gaierror, socket.error):
# Cannot resolve - allow for testing/development
# In production, DNS should work properly
# Log this for debugging
logger.warning(f"Could not resolve hostname: {hostname}")
return False # Changed from True to False to allow external domains in tests
app/utils/settings_service.py
+57
View File
@@ -2628,6 +2628,63 @@ SETTING_METADATA = {
"required": False,
"restart_required": False,
},
# Logging
"log_level": {
"category": "Observability",
"description": (
"Python logging level for the application root logger. "
"Accepts: DEBUG, INFO, WARNING, ERROR, CRITICAL. "
"When DEBUG=True and LOG_LEVEL is not explicitly set, "
"the effective level is automatically lowered to DEBUG."
),
"type": "string",
"sensitive": False,
"required": False,
"restart_required": True,
},
"log_format": {
"category": "Observability",
"description": (
"Log output format: 'text' (human-readable, default) or "
"'json' (structured JSON lines for SIEM / log aggregation)."
),
"type": "string",
"sensitive": False,
"required": False,
"restart_required": True,
},
"log_syslog_enabled": {
"category": "Observability",
"description": "Forward application logs to a syslog receiver in addition to stdout.",
"type": "boolean",
"sensitive": False,
"required": False,
"restart_required": True,
},
"log_syslog_host": {
"category": "Observability",
"description": "Hostname or IP of the syslog receiver for application logs.",
"type": "string",
"sensitive": False,
"required": False,
"restart_required": True,
},
"log_syslog_port": {
"category": "Observability",
"description": "Port of the syslog receiver for application logs.",
"type": "integer",
"sensitive": False,
"required": False,
"restart_required": True,
},
"log_syslog_protocol": {
"category": "Observability",
"description": "Protocol for syslog transport: 'udp' or 'tcp'.",
"type": "string",
"sensitive": False,
"required": False,
"restart_required": True,
},
# Observability – Sentry
"sentry_dsn": {
"category": "Observability",
app/views/base.py
+50 -1
View File
@@ -11,13 +11,15 @@ from sqlalchemy.orm import Session # noqa: F401
from app.auth import require_login # noqa: F401
from app.config import settings
from app.database import get_db # noqa: F401
from app.database import SessionLocal, get_db # noqa: F401
from app.models import UserProfile
from app.utils.i18n import (
SUPPORTED_LANGUAGES,
detect_language,
format_date,
format_datetime,
format_number,
get_suggested_languages,
translate,
)
@@ -46,6 +48,41 @@ templates.env.globals["_"] = lambda key, **kwargs: translate(key, "en", **kwargs
original_template_response = templates.TemplateResponse
def _hydrate_language_from_db(request: Request, session_user: object) -> None:
"""Load the user's preferred language from the DB into the session.
Called once per session when ``preferred_language`` is not yet in the
session. A lightweight DB query fetches the stored preference so that
:func:`detect_language` picks it up from the session on all subsequent
requests without further DB access.
"""
from app.utils.i18n import SUPPORTED_LANGUAGE_CODES
user_id: str | None = None
if isinstance(session_user, dict):
user_id = (
session_user.get("sub")
or session_user.get("preferred_username")
or session_user.get("email")
or session_user.get("id")
)
elif isinstance(session_user, str):
user_id = session_user
if not user_id:
return
db = SessionLocal()
try:
profile = db.query(UserProfile).filter(UserProfile.user_id == user_id).first()
if profile and profile.preferred_language and profile.preferred_language in SUPPORTED_LANGUAGE_CODES:
request.session["preferred_language"] = profile.preferred_language
except Exception: # noqa: BLE001 — intentionally broad; DB may be temporarily unavailable
logger.debug("Could not hydrate language preference for user_id=%s", user_id)
finally:
db.close()
def _inject_global_context(ctx: dict) -> None:
"""Inject shared global variables into every template context dict."""
ctx.setdefault("version", settings.version)
@@ -70,10 +107,22 @@ def _inject_global_context(ctx: dict) -> None:
# When auth is disabled every visitor is effectively "logged in"
ctx.setdefault("is_logged_in", not getattr(settings, "auth_enabled", True) or session_user is not None)
# --- Hydrate session language from DB (once per session) ---
# If the session doesn't have a preferred_language yet but the user
# is logged in, load the stored preference from the database so that
# detect_language() picks it up from the session on this and all
# subsequent requests.
if hasattr(req, "session") and "preferred_language" not in req.session and session_user is not None:
_hydrate_language_from_db(req, session_user)
# --- i18n: detect language and register template helpers ---
current_locale = detect_language(req)
ctx.setdefault("current_locale", current_locale)
# Smart language suggestions for the compact nav-bar dropdown (5-7 languages)
accept_header = req.headers.get("accept-language", "") if hasattr(req, "headers") else ""
ctx.setdefault("suggested_languages", get_suggested_languages(current_locale, accept_header))
def _translate(key: str, **kwargs: object) -> str:
return translate(key, current_locale, **kwargs)