curo1305andClaude Sonnet 4.6
81337bd9e3
- upload.py: request_upload_url, upload_document, confirm_upload (3 routes) - crud.py: list_documents, get_document, patch_document, delete_document, classify_document (5 routes) - content.py: stream_document_content + _parse_range (1 route) - shared.py: _CLOUD_PROVIDERS, UploadUrlRequest, DocumentPatch (single definitions) - __init__.py: router aggregator prefix=/api/documents, 9 routes total - documents.py monolith deleted; all URL paths unchanged - list_documents registered on parent router to avoid FastAPI empty-path/prefix check - get_storage_backend_for_document re-exported for test monkeypatching compatibility Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
367 lines
14 KiB
Python
367 lines
14 KiB
Python
"""Document upload endpoints — presigned URL flow and direct cloud upload.
|
|
|
|
Endpoints:
|
|
POST /upload-url — create pending Document row, return presigned PUT URL (D-05 step 1)
|
|
POST /upload — direct multipart upload supporting cloud backends (D-10, D-14, D-15)
|
|
POST /{doc_id}/confirm — stat MinIO for authoritative size, enforce quota atomically (D-05 step 3)
|
|
|
|
Sub-router carries NO prefix — prefix="/api/documents" lives in __init__.py (D-04).
|
|
|
|
Security:
|
|
T-03-04: object_key computed server-side using str(current_user.id) — never user-supplied.
|
|
T-03-05: size from backend.stat_object() — never from client.
|
|
T-03-06: atomic SQL UPDATE prevents concurrent over-quota uploads (STORE-03 SC2).
|
|
T-03-11: ownership assertion on confirm — cross-user access returns 404.
|
|
T-03-15: object_key prefix always the authenticated user's id.
|
|
T-05-06-01: target_backend validated against _CLOUD_PROVIDERS allowlist.
|
|
T-05-06-02: CloudConnectionError detail never includes provider error detail.
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import uuid
|
|
from pathlib import Path
|
|
|
|
import structlog as _structlog
|
|
|
|
_log = _structlog.get_logger(__name__)
|
|
|
|
from fastapi import APIRouter, Depends, File, Form, HTTPException, Request, UploadFile, status
|
|
from sqlalchemy import text
|
|
from sqlalchemy.ext.asyncio import AsyncSession
|
|
|
|
from config import settings
|
|
from db.models import CloudConnection, Document, User
|
|
from deps.auth import get_regular_user
|
|
from deps.db import get_db
|
|
from deps.utils import get_client_ip
|
|
from services.audit import write_audit_log
|
|
from services.rate_limiting import account_limiter
|
|
from storage import get_storage_backend, get_storage_backend_for_document
|
|
from storage.cloud_utils import decrypt_credentials
|
|
from storage.exceptions import CloudConnectionError
|
|
from tasks.document_tasks import extract_and_classify
|
|
|
|
try:
|
|
from minio.error import S3Error
|
|
except ImportError:
|
|
S3Error = Exception # type: ignore[assignment,misc]
|
|
|
|
from sqlalchemy import select
|
|
|
|
from api.documents.shared import UploadUrlRequest, _CLOUD_PROVIDERS
|
|
|
|
router = APIRouter()
|
|
|
|
|
|
# ── POST /api/documents/upload-url ───────────────────────────────────────────
|
|
|
|
@router.post("/upload-url")
|
|
@account_limiter.limit("100/minute")
|
|
async def request_upload_url(
|
|
request: Request,
|
|
body: UploadUrlRequest,
|
|
session: AsyncSession = Depends(get_db),
|
|
current_user: User = Depends(get_regular_user),
|
|
):
|
|
"""Create a pending Document row and return a presigned PUT URL.
|
|
|
|
D-05 step 1: FastAPI creates a Document row (status='pending'), generates a
|
|
15-minute presigned PUT URL, returns {upload_url, document_id}.
|
|
Quota is NOT reserved at this step — quota enforcement happens at /confirm.
|
|
|
|
T-03-04: object_key is computed server-side using str(current_user.id); filename
|
|
stored in DB only (CLAUDE.md MinIO key schema).
|
|
T-03-15: object_key prefix is always the authenticated user's id — never user-supplied.
|
|
"""
|
|
request.state.current_user = current_user
|
|
doc_id = uuid.uuid4()
|
|
suffix = Path(body.filename).suffix.lower()
|
|
object_key = f"{current_user.id}/{doc_id}/{uuid.uuid4()}{suffix}"
|
|
|
|
doc = Document(
|
|
id=doc_id,
|
|
user_id=current_user.id,
|
|
filename=body.filename,
|
|
content_type=body.content_type,
|
|
size_bytes=0,
|
|
storage_backend="minio",
|
|
status="pending",
|
|
object_key=object_key,
|
|
)
|
|
session.add(doc)
|
|
await session.commit()
|
|
|
|
upload_url = await get_storage_backend().generate_presigned_put_url(
|
|
object_key, expires_minutes=15
|
|
)
|
|
return {"upload_url": upload_url, "document_id": str(doc_id)}
|
|
|
|
|
|
# ── POST /api/documents/upload ────────────────────────────────────────────────
|
|
|
|
@router.post("/upload")
|
|
@account_limiter.limit("100/minute")
|
|
async def upload_document(
|
|
request: Request,
|
|
file: UploadFile = File(...),
|
|
target_backend: str = Form("minio"),
|
|
cloud_folder_path: str = Form(None),
|
|
session: AsyncSession = Depends(get_db),
|
|
current_user: User = Depends(get_regular_user),
|
|
):
|
|
"""Direct multipart upload endpoint supporting cloud backends (D-10, D-14, D-15).
|
|
|
|
If target_backend == "minio": generates a presigned PUT URL (unchanged MinIO flow).
|
|
If target_backend in ("google_drive", "onedrive", "nextcloud", "webdav"):
|
|
1. Reads file bytes from UploadFile
|
|
2. Loads CloudConnection for current_user.id + target_backend; 404 if not found/not ACTIVE
|
|
3. Decrypts credentials and instantiates the correct backend class
|
|
4. Calls cloud_backend.put_object() to upload directly to the provider
|
|
5. Creates Document with storage_backend=target_backend
|
|
6. Returns {document_id, storage_backend} — no upload_url (cloud upload is synchronous)
|
|
|
|
Cloud uploads do NOT use the atomic quota UPDATE — cloud files are not counted
|
|
against MinIO quota (D-11: separate backends; cloud storage quota is provider-side).
|
|
|
|
Security:
|
|
T-05-06-01: target_backend validated against _CLOUD_PROVIDERS allowlist → 422 on invalid value
|
|
T-05-06-02: CloudConnectionError detail message never includes provider error detail
|
|
"""
|
|
request.state.current_user = current_user
|
|
if target_backend == "minio":
|
|
# MinIO: generate a presigned URL for client-side PUT (existing flow reused)
|
|
doc_id = uuid.uuid4()
|
|
suffix = Path(file.filename or "file").suffix.lower()
|
|
object_key = f"{current_user.id}/{doc_id}/{uuid.uuid4()}{suffix}"
|
|
|
|
doc = Document(
|
|
id=doc_id,
|
|
user_id=current_user.id,
|
|
filename=file.filename or "upload",
|
|
content_type=file.content_type or "application/octet-stream",
|
|
size_bytes=0,
|
|
storage_backend="minio",
|
|
status="pending",
|
|
object_key=object_key,
|
|
)
|
|
session.add(doc)
|
|
await session.commit()
|
|
|
|
upload_url = await get_storage_backend().generate_presigned_put_url(
|
|
object_key, expires_minutes=15
|
|
)
|
|
return {"upload_url": upload_url, "document_id": str(doc_id)}
|
|
|
|
# Cloud backend path
|
|
if target_backend not in _CLOUD_PROVIDERS:
|
|
raise HTTPException(
|
|
status_code=422,
|
|
detail=f"Invalid target_backend '{target_backend}'. Valid values: minio, {', '.join(sorted(_CLOUD_PROVIDERS))}",
|
|
)
|
|
|
|
# Load active CloudConnection for current user + provider (T-05-06-01: user-scoped query)
|
|
result = await session.execute(
|
|
select(CloudConnection).where(
|
|
CloudConnection.user_id == current_user.id,
|
|
CloudConnection.provider == target_backend,
|
|
CloudConnection.status == "ACTIVE",
|
|
)
|
|
)
|
|
conn = result.scalar_one_or_none()
|
|
if conn is None:
|
|
raise HTTPException(
|
|
status_code=404,
|
|
detail=f"No active {target_backend} connection found. Please connect in Settings.",
|
|
)
|
|
|
|
# Decrypt per-user credentials
|
|
master_key = settings.cloud_creds_key.encode()
|
|
credentials = decrypt_credentials(master_key, str(current_user.id), conn.credentials_enc)
|
|
|
|
# Read file bytes
|
|
file_bytes = await file.read()
|
|
filename = file.filename or "upload"
|
|
content_type = file.content_type or "application/octet-stream"
|
|
extension = Path(filename).suffix.lower()
|
|
|
|
doc_id = uuid.uuid4()
|
|
|
|
# Instantiate backend and upload
|
|
if target_backend == "google_drive":
|
|
from storage.google_drive_backend import GoogleDriveBackend # lazy import
|
|
cloud_backend = GoogleDriveBackend(credentials)
|
|
elif target_backend == "onedrive":
|
|
from storage.onedrive_backend import OneDriveBackend # lazy import
|
|
cloud_backend = OneDriveBackend(credentials)
|
|
elif target_backend == "nextcloud":
|
|
from storage.nextcloud_backend import NextcloudBackend # lazy import
|
|
cloud_backend = NextcloudBackend(
|
|
credentials["server_url"],
|
|
credentials["username"],
|
|
credentials["password"],
|
|
)
|
|
elif target_backend == "webdav":
|
|
from storage.webdav_backend import WebDAVBackend # lazy import
|
|
cloud_backend = WebDAVBackend(
|
|
credentials["server_url"],
|
|
credentials["username"],
|
|
credentials["password"],
|
|
)
|
|
|
|
try:
|
|
object_key = await cloud_backend.put_object(
|
|
str(current_user.id),
|
|
str(doc_id),
|
|
file_bytes,
|
|
extension,
|
|
content_type,
|
|
cloud_folder=cloud_folder_path or None,
|
|
original_filename=filename if cloud_folder_path else None,
|
|
)
|
|
except CloudConnectionError as exc:
|
|
raise HTTPException(
|
|
status_code=503,
|
|
detail="Cloud connection requires re-authentication. Please reconnect in Settings.",
|
|
) from exc
|
|
|
|
# Bust folder listing cache so the next GET /folders reflects the new file
|
|
if cloud_folder_path:
|
|
from services.cloud_cache import invalidate_provider_cache # lazy import
|
|
invalidate_provider_cache(str(current_user.id), target_backend)
|
|
|
|
doc = Document(
|
|
id=doc_id,
|
|
user_id=current_user.id,
|
|
filename=filename,
|
|
content_type=content_type,
|
|
size_bytes=len(file_bytes),
|
|
storage_backend=target_backend,
|
|
status="uploaded",
|
|
object_key=object_key,
|
|
)
|
|
session.add(doc)
|
|
|
|
_ip = get_client_ip(request) if request else None
|
|
await write_audit_log(
|
|
session,
|
|
event_type="document.uploaded",
|
|
user_id=current_user.id,
|
|
actor_id=current_user.id,
|
|
resource_id=doc.id,
|
|
ip_address=_ip,
|
|
metadata_={"size_bytes": len(file_bytes), "storage_backend": target_backend},
|
|
)
|
|
await session.commit()
|
|
|
|
extract_and_classify.delay(str(doc.id))
|
|
|
|
return {"document_id": str(doc.id), "storage_backend": target_backend}
|
|
|
|
|
|
# ── POST /api/documents/{doc_id}/confirm ─────────────────────────────────────
|
|
|
|
@router.post("/{doc_id}/confirm")
|
|
@account_limiter.limit("100/minute")
|
|
async def confirm_upload(
|
|
doc_id: str,
|
|
request: Request,
|
|
session: AsyncSession = Depends(get_db),
|
|
current_user: User = Depends(get_regular_user),
|
|
):
|
|
"""Confirm a presigned PUT upload: stat MinIO for size, enforce quota atomically.
|
|
|
|
D-05 step 3: FastAPI reads authoritative file size from MinIO stat_object (never
|
|
from client), runs atomic quota UPDATE, sets status='uploaded', enqueues Celery task.
|
|
|
|
Quota exceeded: HTTP 413 with {"used_bytes": N, "limit_bytes": M, "rejected_bytes": K}
|
|
Upload not found: HTTP 422 (presigned URL may have expired)
|
|
|
|
T-03-05: size always comes from backend.stat_object(doc.object_key) — never client.
|
|
T-03-06: atomic SQL UPDATE prevents concurrent over-quota uploads (STORE-03 SC2).
|
|
T-03-11: ownership assertion — cross-user access returns 404 (D-16).
|
|
"""
|
|
request.state.current_user = current_user
|
|
try:
|
|
uid = uuid.UUID(doc_id)
|
|
except ValueError:
|
|
raise HTTPException(status_code=404, detail="Document not found")
|
|
|
|
doc = await session.get(Document, uid)
|
|
if doc is None or doc.user_id != current_user.id:
|
|
raise HTTPException(status_code=404, detail="Document not found")
|
|
|
|
# Get authoritative file size from MinIO (T-03-05 — never trust client-supplied size)
|
|
try:
|
|
size = await get_storage_backend().stat_object(doc.object_key)
|
|
except Exception as exc:
|
|
code = getattr(exc, "code", "")
|
|
if code == "NoSuchKey":
|
|
raise HTTPException(
|
|
status_code=422,
|
|
detail="Upload not found — presigned URL may have expired",
|
|
)
|
|
raise HTTPException(status_code=502, detail=f"Storage error: {exc}")
|
|
|
|
doc.size_bytes = size
|
|
await session.flush()
|
|
|
|
# Atomic quota enforcement — user_id is always set post-migration (Plan 03-03+)
|
|
result = await session.execute(
|
|
text(
|
|
"UPDATE quotas "
|
|
"SET used_bytes = used_bytes + :delta "
|
|
"WHERE user_id = :uid "
|
|
" AND (used_bytes + :delta) <= limit_bytes "
|
|
"RETURNING used_bytes, limit_bytes"
|
|
),
|
|
{"delta": size, "uid": doc.user_id.hex},
|
|
)
|
|
row = result.fetchone()
|
|
|
|
if row is None:
|
|
# Quota exceeded — fetch current quota state for the 413 body
|
|
quota_result = await session.execute(
|
|
text("SELECT used_bytes, limit_bytes FROM quotas WHERE user_id = :uid"),
|
|
{"uid": doc.user_id.hex},
|
|
)
|
|
q = quota_result.fetchone()
|
|
# Delete the pending Document row and best-effort remove the MinIO object
|
|
await session.delete(doc)
|
|
try:
|
|
await get_storage_backend().delete_object(doc.object_key)
|
|
except Exception:
|
|
pass # MinIO cleanup is best-effort; object TTL will eventually expire
|
|
await session.commit()
|
|
raise HTTPException(
|
|
status_code=413,
|
|
detail={
|
|
"used_bytes": q.used_bytes if q else 0,
|
|
"limit_bytes": q.limit_bytes if q else 0,
|
|
"rejected_bytes": size,
|
|
},
|
|
)
|
|
|
|
used_bytes = row.used_bytes
|
|
|
|
doc.status = "uploaded"
|
|
# D-13: document uploaded event — size_bytes + storage_backend only, NO filename, NO extracted_text (T-04-07-02)
|
|
_ip = get_client_ip(request)
|
|
await write_audit_log(
|
|
session,
|
|
event_type="document.uploaded",
|
|
user_id=current_user.id,
|
|
actor_id=current_user.id,
|
|
resource_id=doc.id,
|
|
ip_address=_ip,
|
|
metadata_={"size_bytes": size, "storage_backend": "minio"},
|
|
)
|
|
await session.commit()
|
|
extract_and_classify.delay(str(doc.id))
|
|
|
|
return {
|
|
"id": str(doc.id),
|
|
"size_bytes": size,
|
|
"used_bytes": used_bytes,
|
|
"status": "uploaded",
|
|
}
|