Agent Zero
f445aa69d5
Check Cross-Plugin Imports / check (push) Has been cancelled
- common.py neu: alle Safety-/Storage-Helper und Konstanten (exakte Original-Implementierung) - folders_routes.py / sharing_routes.py / search_bulk_routes.py je eigener Router ohne Prefix - routes.py: File-Lifecycle-Kern bleibt physisch (MAX_FILE_SIZE-Test-Patch-Semantik), Rest als Re-Export-Fassade + include_router x3 - Beweis: DMS-Suite 129 Tests = 125 passed + 4 identische Vorbestand-Failures (Baseline-Referenz 1:1), 20/20 Routen via Router-Introspection, ruff clean
109 lines
3.1 KiB
Python
109 lines
3.1 KiB
Python
"""DMS gemeinsame Helper & Konstanten — BUG-018 God-Object-Split."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import os
|
|
import uuid
|
|
|
|
from fastapi import HTTPException
|
|
|
|
OFFICE_EXTENSIONS = {
|
|
".docx": "docx",
|
|
".xlsx": "xlsx",
|
|
".pptx": "pptx",
|
|
}
|
|
|
|
# Max file size: 100 MB
|
|
MAX_FILE_SIZE = 100 * 1024 * 1024
|
|
|
|
|
|
def _parse_uuid(val: str, field: str) -> uuid.UUID:
|
|
try:
|
|
return uuid.UUID(val)
|
|
except (ValueError, TypeError):
|
|
raise HTTPException(
|
|
400, detail={"detail": f"Invalid {field}", "code": "invalid_id"}
|
|
) from None
|
|
|
|
|
|
def _file_storage_path(tenant_id: uuid.UUID, file_id: uuid.UUID) -> str:
|
|
"""Build relative storage path for a file (relative to storage base)."""
|
|
return f"{tenant_id}/{file_id}"
|
|
|
|
|
|
def _get_file_extension(filename: str) -> str:
|
|
"""Extract lowercase extension including dot."""
|
|
return os.path.splitext(filename)[1].lower()
|
|
|
|
|
|
|
|
def _sanitize_filename(filename: str) -> str:
|
|
"""Sanitize a filename for safe use in Content-Disposition headers."""
|
|
import re
|
|
# Extract basename only (strip any path components)
|
|
safe = os.path.basename(filename.replace('\\', '/'))
|
|
# Remove dangerous characters (keep alnum, dot, dash, underscore, space, unicode)
|
|
safe = re.sub(r'[^a-zA-Z0-9.\-_\u00c0-\u017f\u4e00-\u9fff ]', '_', safe)
|
|
# Collapse consecutive dots (path traversal prevention)
|
|
safe = re.sub(r'\.{2,}', '_', safe)
|
|
# Collapse multiple spaces
|
|
safe = re.sub(r' {2,}', ' ', safe)
|
|
# Strip leading dots and whitespace
|
|
safe = safe.lstrip('.').strip()
|
|
# Limit length
|
|
if len(safe) > 200:
|
|
name, ext = safe.rsplit('.', 1) if '.' in safe[:200] else (safe[:200], '')
|
|
safe = name[:200] + ('.' + ext if ext else '')
|
|
return safe or 'file'
|
|
|
|
# Blocked file extensions for security
|
|
BLOCKED_EXTENSIONS = {
|
|
".exe", ".bat", ".cmd", ".sh", ".jar", ".com", ".scr", ".msi",
|
|
".dll", ".vbs", ".ps1", ".app", ".bin", ".reg", ".inf",
|
|
".php", ".py", ".pl", ".asp", ".aspx", ".jsp", ".svg", ".htaccess",
|
|
".phtml", ".pht", ".cgi", ".cfm", ".erb",
|
|
}
|
|
|
|
# Allowed MIME types for upload validation
|
|
ALLOWED_MIME_PREFIXES = {
|
|
"application/pdf",
|
|
"application/msword",
|
|
"application/vnd.openxmlformats-officedocument",
|
|
"application/vnd.oasis.opendocument",
|
|
"application/vnd.ms-excel",
|
|
"application/vnd.ms-powerpoint",
|
|
"application/zip",
|
|
"application/gzip",
|
|
"application/x-tar",
|
|
"application/json",
|
|
"application/xml",
|
|
"application/rtf",
|
|
"application/x-7z-compressed",
|
|
"application/x-rar-compressed",
|
|
"text/plain",
|
|
"text/csv",
|
|
"text/html",
|
|
"text/markdown",
|
|
"image/png",
|
|
"image/jpeg",
|
|
"image/gif",
|
|
"image/webp",
|
|
"image/bmp",
|
|
"image/tiff",
|
|
"image/x-icon",
|
|
"audio/",
|
|
"video/",
|
|
"application/octet-stream",
|
|
}
|
|
|
|
|
|
def _is_blocked_filetype(filename: str) -> bool:
|
|
"""Check if a file has a blocked (dangerous) extension."""
|
|
ext = os.path.splitext(filename)[1].lower()
|
|
return ext in BLOCKED_EXTENSIONS
|
|
|
|
|
|
chunk_size = 1024 * 1024 # 1MB chunks for streaming uploads
|
|
|
|
# ─── Folders ───
|