From 81337bd9e314dad3b1402d69e47c483386ba6c63 Mon Sep 17 00:00:00 2001 From: curo1305 Date: Fri, 12 Jun 2026 10:27:48 +0200 Subject: [PATCH] =?UTF-8?q?feat(08-05):=20decompose=20documents=20API=20mo?= =?UTF-8?q?nolith=20into=20documents/=20package=20=E2=80=94=20CODE-02=20CO?= =?UTF-8?q?DE-08?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - 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 --- backend/api/documents.py | 852 ------------------------------ backend/api/documents/__init__.py | 28 + backend/api/documents/content.py | 145 +++++ backend/api/documents/crud.py | 388 ++++++++++++++ backend/api/documents/shared.py | 45 ++ backend/api/documents/upload.py | 366 +++++++++++++ 6 files changed, 972 insertions(+), 852 deletions(-) delete mode 100644 backend/api/documents.py create mode 100644 backend/api/documents/__init__.py create mode 100644 backend/api/documents/content.py create mode 100644 backend/api/documents/crud.py create mode 100644 backend/api/documents/shared.py create mode 100644 backend/api/documents/upload.py diff --git a/backend/api/documents.py b/backend/api/documents.py deleted file mode 100644 index 9b11c8f..0000000 --- a/backend/api/documents.py +++ /dev/null @@ -1,852 +0,0 @@ -""" -Document API endpoints for DocuVault — Phase 3 Wave 2 / Phase 5 Plan 06. - -Implements the presigned PUT upload flow (D-04, D-05): - POST /api/documents/upload-url — create pending Document row, return presigned URL - POST /api/documents/{id}/confirm — stat MinIO for authoritative size, atomic quota UPDATE - -Cloud upload path (D-10, D-14, D-15 — Phase 5 Plan 06): - POST /api/documents/upload — multipart upload with target_backend parameter; - cloud backends bypass presigned URL and use direct put_object() - -Preserved endpoints (auth guards added in Plan 03-03): - GET /api/documents — list documents - GET /api/documents/{id} — get document metadata - DELETE /api/documents/{id} — delete document (decrements quota atomically) - POST /api/documents/{id}/classify — reclassify document topics - GET /api/documents/{id}/content — stream document bytes (all backends, Phase 5 Plan 06) - -NOTE (Wave 2): No auth guards on any endpoint yet — Plan 03-03 adds get_current_user -to all handlers. The doc.user_id=None guard in /confirm is a Wave 2 placeholder. -""" -from __future__ import annotations - -import urllib.parse -import uuid -from pathlib import Path -from typing import Optional - -import structlog as _structlog -_log = _structlog.get_logger(__name__) - -from fastapi import APIRouter, Depends, Form, HTTPException, Query, Request, UploadFile, File, status -from fastapi.responses import JSONResponse, StreamingResponse -from pydantic import BaseModel, Field, field_validator -from sqlalchemy import select, text, func -from sqlalchemy.ext.asyncio import AsyncSession - -from config import settings -from db.models import CloudConnection, Document, Folder, Quota, Share, User -from deps.auth import get_regular_user -from deps.db import get_db -from deps.utils import get_client_ip -from services import classifier, storage -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 tasks.document_tasks import extract_and_classify - -try: - from minio.error import S3Error -except ImportError: - # Fallback for test environments where minio is not installed - S3Error = Exception # type: ignore[assignment,misc] - -from storage.exceptions import CloudConnectionError - -# Valid cloud backend slugs (T-05-06-01: validated against allowlist, not user-supplied string) -_CLOUD_PROVIDERS = frozenset({"google_drive", "onedrive", "nextcloud", "webdav"}) - -router = APIRouter(prefix="/api/documents", tags=["documents"]) - - -# ── Request models ──────────────────────────────────────────────────────────── - -class UploadUrlRequest(BaseModel): - filename: str - content_type: str - - -class DocumentPatch(BaseModel): - """Pydantic model for PATCH /api/documents/{doc_id}. - - Optional fields — model_fields_set distinguishes "not provided" from "set to null". - At least one field must be present in model_fields_set (enforced in the handler). - - T-05-09-01: explicit field declaration prevents mass assignment. - T-05-09-02: only filename and folder_id are accepted — no other fields can be set. - """ - filename: Optional[str] = Field(None, min_length=1, max_length=255) - folder_id: Optional[uuid.UUID] = None - - @field_validator("filename") - @classmethod - def filename_no_path_separators(cls, v: Optional[str]) -> Optional[str]: - if v is not None and ("/" in v or "\\" in v): - raise ValueError("filename must not contain path separators") - return v - - -# ── 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", - } - - -# ── GET /api/documents ──────────────────────────────────────────────────────── - -@router.get("") -@account_limiter.limit("100/minute") -async def list_documents( - request: Request, - topic: Optional[str] = Query(None), - page: int = Query(1, ge=1), - per_page: int = Query(20, ge=1, le=100), - sort: str = Query("date"), - order: str = Query("desc"), - folder_id: Optional[str] = Query(None), - q: Optional[str] = Query(None), - session: AsyncSession = Depends(get_db), - current_user: User = Depends(get_regular_user), -): - """List documents with optional sort, folder filter, and full-text search. - - D-16: requires authenticated regular user (get_regular_user rejects admins). - Returns only documents belonging to the current user. - - FOLD-05: sort by name|date|size; order asc|desc; folder_id filter; - q full-text search via plainto_tsquery (PostgreSQL only — silently skipped - on SQLite when function is unavailable). FTS scope is always scoped to - current_user.id (T-04-03-02). - - Backward-compat: when sort/order/folder_id/q are not provided, behaviour - is identical to the pre-Phase-4 implementation. - """ - request.state.current_user = current_user - # If no new params used, fall through to the legacy storage.list_metadata path - # to preserve full backward compatibility with topic filtering. - if folder_id is None and q is None and sort == "date" and order == "desc": - docs = await storage.list_metadata(session, user_id=current_user.id, topic=topic) - total = len(docs) - start = (page - 1) * per_page - # Add is_shared field (Phase 4 addition) - shared_result = await session.execute( - select(Share.document_id).where(Share.owner_id == current_user.id) - ) - shared_ids = {row[0] for row in shared_result.fetchall()} - items = [] - for d in docs[start : start + per_page]: - doc_id_str = d.get("id", "") - try: - doc_uuid = uuid.UUID(doc_id_str) - except (ValueError, AttributeError): - doc_uuid = None - d["is_shared"] = doc_uuid in shared_ids if doc_uuid else False - items.append(d) - return {"items": items, "total": total, "page": page, "per_page": per_page} - - # New path: direct ORM query with sort/filter/FTS - from db.models import DocumentTopic, Topic # noqa: PLC0415 (avoid circular at module top) - - stmt = select(Document).where(Document.user_id == current_user.id) - - # Topic filter (join-based, same as list_metadata) - if topic is not None: - stmt = ( - stmt.join(DocumentTopic, DocumentTopic.document_id == Document.id) - .join(Topic, Topic.id == DocumentTopic.topic_id) - .where(Topic.name == topic) - ) - - # Folder filter - if folder_id is not None: - try: - folder_uuid = uuid.UUID(folder_id) - except ValueError: - raise HTTPException(status_code=404, detail="Folder not found") - stmt = stmt.where(Document.folder_id == folder_uuid) - - # Sort - sort_col = Document.created_at # default: date - if sort == "name": - sort_col = Document.filename - elif sort == "size": - sort_col = Document.size_bytes - - order_fn = sort_col.asc if order == "asc" else sort_col.desc - stmt = stmt.order_by(order_fn()) - - # Full-text search — plainto_tsquery on extracted_text (PostgreSQL only) - # Falls back to unfiltered if the DB dialect doesn't support @@ (e.g. SQLite in test env) - fts_requested = q is not None and len(q) >= 2 - if fts_requested: - fts_stmt = stmt.where( - func.to_tsvector("english", func.coalesce(Document.extracted_text, "")).op("@@")( - func.plainto_tsquery("english", q) - ) - ) - try: - result = await session.execute(fts_stmt) - except Exception: - result = await session.execute(stmt) - else: - result = await session.execute(stmt) - docs_orm = result.scalars().all() - - # is_shared subquery - shared_result = await session.execute( - select(Share.document_id).where(Share.owner_id == current_user.id) - ) - shared_ids = {row[0] for row in shared_result.fetchall()} - - # Serialize - all_items = [] - for doc in docs_orm: - from services.storage import _doc_to_dict, _load_topic_names # noqa: PLC0415 - topic_names = await _load_topic_names(session, doc.id) - d = _doc_to_dict(doc, topic_names) - d["is_shared"] = doc.id in shared_ids - all_items.append(d) - - total = len(all_items) - start = (page - 1) * per_page - return { - "items": all_items[start : start + per_page], - "total": total, - "page": page, - "per_page": per_page, - } - - -# ── GET /api/documents/{doc_id} ─────────────────────────────────────────────── - -@router.get("/{doc_id}") -@account_limiter.limit("100/minute") -async def get_document( - request: Request, - doc_id: str, - session: AsyncSession = Depends(get_db), - current_user: User = Depends(get_regular_user), -): - """Return document metadata by ID. - - D-16: requires authenticated regular user. Asserts ownership — cross-user - access returns 404 (not 403) to avoid information leakage (T-03-11). - """ - request.state.current_user = current_user - try: - uid = uuid.UUID(doc_id) - except ValueError: - raise HTTPException(404, "Document not found") - - doc = await session.get(Document, uid) - if doc is None: - raise HTTPException(404, "Document not found") - - is_recipient = False - if doc.user_id != current_user.id: - # Allow recipients of an active share to view the document - share_result = await session.execute( - select(Share).where( - Share.document_id == uid, - Share.recipient_id == current_user.id, - ) - ) - if share_result.scalar_one_or_none() is None: - raise HTTPException(404, "Document not found") - is_recipient = True - - meta = await storage.get_metadata(session, doc_id) - if meta is None: - raise HTTPException(404, "Document not found") - # T-04-04-03: recipients get metadata only — extracted_text excluded (consistent with /shares/received) - if is_recipient: - meta.pop("extracted_text", None) - return meta - - -# ── PATCH /api/documents/{doc_id} ──────────────────────────────────────────── - -@router.patch("/{doc_id}") -@account_limiter.limit("100/minute") -async def patch_document( - request: Request, - doc_id: str, - body: DocumentPatch, - session: AsyncSession = Depends(get_db), - current_user: User = Depends(get_regular_user), -): - """Update document metadata (filename and/or folder_id). - - T-05-09-01: get_regular_user dep rejects admins (403) and unauthenticated (401). - T-05-09-01: ownership check — non-owner gets 404 to avoid leaking document IDs (D-16). - T-05-09-02: response uses storage.get_metadata() which excludes credentials_enc and - password_hash via the _doc_to_dict whitelist. - - At least one field must be provided — empty body returns 422. - folder_id=null moves the document to the root (no folder). - """ - request.state.current_user = current_user - try: - uid = uuid.UUID(doc_id) - except ValueError: - raise HTTPException(404, "Document not found") - - doc = await session.get(Document, uid) - if doc is None or doc.user_id != current_user.id: - raise HTTPException(404, "Document not found") - - # Require at least one field to be set (model_fields_set tracks provided fields) - if not body.model_fields_set: - raise HTTPException(422, "At least one field (filename, folder_id) must be provided") - - if "filename" in body.model_fields_set and body.filename is not None: - doc.filename = body.filename - - if "folder_id" in body.model_fields_set: - # folder_id=null → move to root (no folder); folder_id= → move to folder - if body.folder_id is not None: - target = await session.get(Folder, body.folder_id) - if target is None or target.user_id != current_user.id: - raise HTTPException(404, "Folder not found") - doc.folder_id = body.folder_id - - await session.commit() - - meta = await storage.get_metadata(session, doc_id) - if meta is None: - raise HTTPException(404, "Document not found") - return meta - - -# ── DELETE /api/documents/{doc_id} ─────────────────────────────────────────── - -@router.delete("/{doc_id}") -@account_limiter.limit("100/minute") -async def delete_document( - doc_id: str, - request: Request, - remove_only: bool = Query(default=False), - session: AsyncSession = Depends(get_db), - current_user: User = Depends(get_regular_user), -): - """Delete a document and decrement quota atomically. - - For cloud-stored documents: - - Default path: attempt cloud provider delete first; on failure return - {success: false, cloud_delete_failed: true} (HTTP 200) so the frontend - can offer a "Remove from app" fallback (T-06.2-03-02). - - remove_only=true: skip cloud delete, remove DB row only, skip quota decrement. - - Cloud docs always use skip_quota=True (never charged MinIO quota, T-06.2-03-01). - - D-16: requires authenticated regular user. Asserts ownership — cross-user - delete returns 404 (not 403) to avoid information leakage (T-03-11). - """ - request.state.current_user = current_user - try: - uid = uuid.UUID(doc_id) - except ValueError: - raise HTTPException(404, "Document not found") - - doc = await session.get(Document, uid) - if doc is None or doc.user_id != current_user.id: - raise HTTPException(404, "Document not found") - - is_cloud = doc.storage_backend != "minio" - _doc_size = doc.size_bytes - _doc_id = doc.id - _ip = get_client_ip(request) - - # Cloud routing: attempt provider delete unless remove_only is set - if is_cloud and not remove_only: - try: - cloud_backend = await get_storage_backend_for_document(doc, current_user, session) - await cloud_backend.delete_object(doc.object_key) - except Exception as exc: - _log.warning("cloud_delete_failed", provider=doc.storage_backend, error=str(exc)) - return JSONResponse( - status_code=200, - content={ - "success": False, - "cloud_delete_failed": True, - "detail": "Cloud provider delete failed. You can remove from app only.", - }, - ) - - # auto_commit=False defers the commit so the audit log write below happens - # in the same transaction — avoids the split-transaction gap (WR-08). - ok = await storage.delete_document(session, doc_id, skip_quota=is_cloud, auto_commit=False) - if not ok: - raise HTTPException(404, "Document not found") - - # D-13: document deleted event — written in the same transaction as the delete (WR-08). - await write_audit_log( - session, - event_type="document.deleted", - user_id=current_user.id, - actor_id=current_user.id, - resource_id=_doc_id, - ip_address=_ip, - metadata_={"size_bytes": _doc_size}, - ) - await session.commit() - - return {"success": True} - - -# ── POST /api/documents/{doc_id}/classify ──────────────────────────────────── - -@router.post("/{doc_id}/classify") -@account_limiter.limit("100/minute") -async def classify_document( - request: Request, - doc_id: str, - session: AsyncSession = Depends(get_db), - current_user: User = Depends(get_regular_user), -): - """Re-queue a document for classification via Celery (D-11). - - Sets doc.status='processing', commits, dispatches extract_and_classify.delay(), - and returns {'document_id': str, 'status': 'processing'}. - - D-16: requires authenticated regular user. Asserts ownership — cross-user - classify returns 404 (not 403) to avoid information leakage (T-03-11). - T-07-10: ownership enforced here; IDOR returns 404 per STATE.md policy. - """ - request.state.current_user = current_user - try: - uid = uuid.UUID(doc_id) - except ValueError: - raise HTTPException(404, "Document not found") - - doc = await session.get(Document, uid) - if doc is None or doc.user_id != current_user.id: - raise HTTPException(404, "Document not found") - - doc.status = "processing" - await session.commit() - - extract_and_classify.delay(str(doc.id)) - - return {"document_id": str(doc.id), "status": "processing"} - - -# ── Range header parsing helper ─────────────────────────────────────────────── - -def _parse_range(range_header: str, file_size: int) -> tuple: - """Parse a 'bytes=X-Y' Range header and return (start, end). - - Returns (start, end) where both are inclusive byte offsets. - Raises HTTP 416 on any invalid or out-of-bounds range. - - T-04-05-03: validates start <= end, start >= 0, end < file_size. - """ - try: - h = range_header.replace("bytes=", "").split("-") - start = int(h[0]) if h[0] != "" else 0 - end = int(h[1]) if h[1] != "" else file_size - 1 - except (ValueError, IndexError): - raise HTTPException(status.HTTP_416_RANGE_NOT_SATISFIABLE) - if start > end or start < 0 or end >= file_size: - raise HTTPException(status.HTTP_416_RANGE_NOT_SATISFIABLE) - return start, end - - -# ── GET /api/documents/{doc_id}/content ────────────────────────────────────── - -@router.get("/{doc_id}/content") -@account_limiter.limit("100/minute") -async def stream_document_content( - doc_id: str, - request: Request, - session: AsyncSession = Depends(get_db), - current_user: User = Depends(get_regular_user), -): - """Stream document bytes directly from MinIO (DOC-02). - - T-04-05-01: uses get_regular_user — admin role → 403 (critical security invariant). - T-04-05-02: bytes fetched via get_object() ONLY — presigned_get_url() never called. - T-04-05-03: Range header validated via _parse_range(); invalid range → 416. - T-04-05-04: access gated on ownership OR active Share.recipient_id. - - Returns 200 (or 206 for Range requests) with: - Content-Type: doc.content_type - Content-Disposition: inline; filename="" - Accept-Ranges: bytes - Content-Length: - """ - 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: - raise HTTPException(status_code=404, detail="Document not found") - - # Access control: owner OR share recipient (T-04-05-04) - if doc.user_id != current_user.id: - result = await session.execute( - select(Share).where( - Share.document_id == doc.id, - Share.recipient_id == current_user.id, - ) - ) - share = result.scalar_one_or_none() - if share is None: - raise HTTPException(status_code=404, detail="Document not found") - - # Fetch bytes from the correct backend — get_storage_backend_for_document handles - # all backends (MinIO, Google Drive, OneDrive, Nextcloud, WebDAV) transparently - # (D-15, T-04-05-02). NEVER via presigned URL for cloud backends (D-14). - try: - storage_backend = await get_storage_backend_for_document(doc, current_user, session) - file_bytes = await storage_backend.get_object(doc.object_key) - except CloudConnectionError as exc: - raise HTTPException( - status_code=503, - detail="Cloud connection requires re-authentication. Please reconnect in Settings.", - ) from exc - except HTTPException: - raise - except Exception as exc: - raise HTTPException( - status_code=502, - detail="Cloud backend unreachable. Please try again or reconnect in Settings.", - ) from exc - file_size = len(file_bytes) - - safe_name = urllib.parse.quote(doc.filename, safe='') - headers = { - "content-type": doc.content_type, - "content-disposition": f"inline; filename*=UTF-8''{safe_name}", - "accept-ranges": "bytes", - "content-length": str(file_size), - } - - range_header = request.headers.get("range") - if range_header: - start, end = _parse_range(range_header, file_size) - chunk = file_bytes[start : end + 1] - headers["content-range"] = f"bytes {start}-{end}/{file_size}" - headers["content-length"] = str(len(chunk)) - return StreamingResponse( - iter([chunk]), - status_code=206, - headers=headers, - ) - - return StreamingResponse( - iter([file_bytes]), - status_code=200, - headers=headers, - ) diff --git a/backend/api/documents/__init__.py b/backend/api/documents/__init__.py new file mode 100644 index 0000000..51cfb16 --- /dev/null +++ b/backend/api/documents/__init__.py @@ -0,0 +1,28 @@ +"""Document API package — router aggregator. + +Aggregates upload_router, crud_router, and content_router under a single +APIRouter with prefix="/api/documents". This is the ONLY file in the package +that sets the prefix — sub-routers declare no prefix (D-04). + +main.py import unchanged after decomposition: + from api.documents import router as documents_router + app.include_router(documents_router) +""" +from fastapi import APIRouter + +from api.documents.upload import router as upload_router +from api.documents.crud import router as crud_router, list_documents +from api.documents.content import router as content_router +# Re-export for test monkeypatching: tests patch api.documents.extract_and_classify.delay +# (same Celery task object as upload.py uses — patching .delay affects all callers) +from tasks.document_tasks import extract_and_classify # noqa: F401 +from storage import get_storage_backend_for_document # noqa: F401 + +router = APIRouter(prefix="/api/documents", tags=["documents"]) +router.include_router(upload_router) +# list_documents registered directly on parent: FastAPI 0.100+ disallows include_router +# when the include prefix AND route path are both empty strings (our sub-router pattern +# uses no prefix per D-04, and the list endpoint has path=""). +router.add_api_route("", list_documents, methods=["GET"]) +router.include_router(crud_router) +router.include_router(content_router) diff --git a/backend/api/documents/content.py b/backend/api/documents/content.py new file mode 100644 index 0000000..6cd99f7 --- /dev/null +++ b/backend/api/documents/content.py @@ -0,0 +1,145 @@ +"""Document content streaming endpoint. + +Endpoints: + GET /{doc_id}/content — stream document bytes from any backend (MinIO or cloud) + +Sub-router carries NO prefix — prefix="/api/documents" lives in __init__.py (D-04). + +Security: + T-04-05-01: uses get_regular_user — admin role → 403 (critical security invariant). + T-04-05-02: bytes fetched via get_object() ONLY — presigned_get_url() never called. + T-04-05-03: Range header validated via _parse_range(); invalid range → 416. + T-04-05-04: access gated on ownership OR active Share.recipient_id. + DOC-04 / SEC-04: ownership assertion before serving bytes. +""" +from __future__ import annotations + +import urllib.parse +import uuid + +from fastapi import APIRouter, Depends, HTTPException, Request, status +from fastapi.responses import StreamingResponse +from sqlalchemy import select +from sqlalchemy.ext.asyncio import AsyncSession + +from db.models import Document, Share, User +from deps.auth import get_regular_user +from deps.db import get_db +from services.rate_limiting import account_limiter +from storage.exceptions import CloudConnectionError + +router = APIRouter() + + +# ── Range header parsing helper ─────────────────────────────────────────────── + +def _parse_range(range_header: str, file_size: int) -> tuple: + """Parse a 'bytes=X-Y' Range header and return (start, end). + + Returns (start, end) where both are inclusive byte offsets. + Raises HTTP 416 on any invalid or out-of-bounds range. + + T-04-05-03: validates start <= end, start >= 0, end < file_size. + """ + try: + h = range_header.replace("bytes=", "").split("-") + start = int(h[0]) if h[0] != "" else 0 + end = int(h[1]) if h[1] != "" else file_size - 1 + except (ValueError, IndexError): + raise HTTPException(status.HTTP_416_RANGE_NOT_SATISFIABLE) + if start > end or start < 0 or end >= file_size: + raise HTTPException(status.HTTP_416_RANGE_NOT_SATISFIABLE) + return start, end + + +# ── GET /api/documents/{doc_id}/content ────────────────────────────────────── + +@router.get("/{doc_id}/content") +@account_limiter.limit("100/minute") +async def stream_document_content( + doc_id: str, + request: Request, + session: AsyncSession = Depends(get_db), + current_user: User = Depends(get_regular_user), +): + """Stream document bytes directly from MinIO (DOC-02). + + T-04-05-01: uses get_regular_user — admin role → 403 (critical security invariant). + T-04-05-02: bytes fetched via get_object() ONLY — presigned_get_url() never called. + T-04-05-03: Range header validated via _parse_range(); invalid range → 416. + T-04-05-04: access gated on ownership OR active Share.recipient_id. + + Returns 200 (or 206 for Range requests) with: + Content-Type: doc.content_type + Content-Disposition: inline; filename="" + Accept-Ranges: bytes + Content-Length: + """ + 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: + raise HTTPException(status_code=404, detail="Document not found") + + # Access control: owner OR share recipient (T-04-05-04) + if doc.user_id != current_user.id: + result = await session.execute( + select(Share).where( + Share.document_id == doc.id, + Share.recipient_id == current_user.id, + ) + ) + share = result.scalar_one_or_none() + if share is None: + raise HTTPException(status_code=404, detail="Document not found") + + # Fetch bytes from the correct backend — get_storage_backend_for_document handles + # all backends (MinIO, Google Drive, OneDrive, Nextcloud, WebDAV) transparently + # (D-15, T-04-05-02). NEVER via presigned URL for cloud backends (D-14). + try: + import api.documents as _doc_pkg # late import allows test monkeypatching via api.documents + storage_backend = await _doc_pkg.get_storage_backend_for_document(doc, current_user, session) + file_bytes = await storage_backend.get_object(doc.object_key) + except CloudConnectionError as exc: + raise HTTPException( + status_code=503, + detail="Cloud connection requires re-authentication. Please reconnect in Settings.", + ) from exc + except HTTPException: + raise + except Exception as exc: + raise HTTPException( + status_code=502, + detail="Cloud backend unreachable. Please try again or reconnect in Settings.", + ) from exc + file_size = len(file_bytes) + + safe_name = urllib.parse.quote(doc.filename, safe='') + headers = { + "content-type": doc.content_type, + "content-disposition": f"inline; filename*=UTF-8''{safe_name}", + "accept-ranges": "bytes", + "content-length": str(file_size), + } + + range_header = request.headers.get("range") + if range_header: + start, end = _parse_range(range_header, file_size) + chunk = file_bytes[start : end + 1] + headers["content-range"] = f"bytes {start}-{end}/{file_size}" + headers["content-length"] = str(len(chunk)) + return StreamingResponse( + iter([chunk]), + status_code=206, + headers=headers, + ) + + return StreamingResponse( + iter([file_bytes]), + status_code=200, + headers=headers, + ) diff --git a/backend/api/documents/crud.py b/backend/api/documents/crud.py new file mode 100644 index 0000000..c6d1963 --- /dev/null +++ b/backend/api/documents/crud.py @@ -0,0 +1,388 @@ +"""Document CRUD endpoints — list, get, patch, delete, and re-classify. + +Endpoints: + GET "" — list documents with sort, folder filter, and FTS (list_documents) + GET /{doc_id} — get document metadata (get_document) + PATCH /{doc_id} — update filename and/or folder_id (patch_document) + DELETE /{doc_id} — delete document, decrement quota atomically (delete_document) + POST /{doc_id}/classify — re-queue Celery classification (classify_document, D-08) + +Sub-router carries NO prefix — prefix="/api/documents" lives in __init__.py (D-04). + +Security: + T-03-11: ownership assertion on every resource endpoint — cross-user access returns 404. + T-05-09-01: get_regular_user dep rejects admins (403) and unauthenticated (401). + T-05-09-02: response uses storage.get_metadata() whitelist — no credentials_enc, no password_hash. + T-06.2-03-01: cloud documents skip MinIO quota decrement. + T-06.2-03-02: cloud delete failure returns {success: false, cloud_delete_failed: true} (HTTP 200). + T-07-10: classify endpoint — IDOR returns 404 per ownership assertion. +""" +from __future__ import annotations + +import uuid +from typing import Optional + +import structlog as _structlog + +_log = _structlog.get_logger(__name__) + +from fastapi import APIRouter, Depends, HTTPException, Query, Request, status +from fastapi.responses import JSONResponse +from sqlalchemy import func, select +from sqlalchemy.ext.asyncio import AsyncSession + +from db.models import Document, Folder, Share, User +from deps.auth import get_regular_user +from deps.db import get_db +from deps.utils import get_client_ip +from services import classifier, storage +from services.audit import write_audit_log +from services.rate_limiting import account_limiter +from storage import get_storage_backend_for_document as _get_storage_backend_for_document +from tasks.document_tasks import extract_and_classify + +from api.documents.shared import DocumentPatch, _CLOUD_PROVIDERS + +router = APIRouter() + + +# ── GET /api/documents ──────────────────────────────────────────────────────── +# Route registered on parent router in __init__.py (FastAPI 0.100+ disallows +# include_router when both the include prefix and route path are empty strings). + +@account_limiter.limit("100/minute") +async def list_documents( + request: Request, + topic: Optional[str] = Query(None), + page: int = Query(1, ge=1), + per_page: int = Query(20, ge=1, le=100), + sort: str = Query("date"), + order: str = Query("desc"), + folder_id: Optional[str] = Query(None), + q: Optional[str] = Query(None), + session: AsyncSession = Depends(get_db), + current_user: User = Depends(get_regular_user), +): + """List documents with optional sort, folder filter, and full-text search. + + D-16: requires authenticated regular user (get_regular_user rejects admins). + Returns only documents belonging to the current user. + + FOLD-05: sort by name|date|size; order asc|desc; folder_id filter; + q full-text search via plainto_tsquery (PostgreSQL only — silently skipped + on SQLite when function is unavailable). FTS scope is always scoped to + current_user.id (T-04-03-02). + + Backward-compat: when sort/order/folder_id/q are not provided, behaviour + is identical to the pre-Phase-4 implementation. + """ + request.state.current_user = current_user + # If no new params used, fall through to the legacy storage.list_metadata path + # to preserve full backward compatibility with topic filtering. + if folder_id is None and q is None and sort == "date" and order == "desc": + docs = await storage.list_metadata(session, user_id=current_user.id, topic=topic) + total = len(docs) + start = (page - 1) * per_page + # Add is_shared field (Phase 4 addition) + shared_result = await session.execute( + select(Share.document_id).where(Share.owner_id == current_user.id) + ) + shared_ids = {row[0] for row in shared_result.fetchall()} + items = [] + for d in docs[start : start + per_page]: + doc_id_str = d.get("id", "") + try: + doc_uuid = uuid.UUID(doc_id_str) + except (ValueError, AttributeError): + doc_uuid = None + d["is_shared"] = doc_uuid in shared_ids if doc_uuid else False + items.append(d) + return {"items": items, "total": total, "page": page, "per_page": per_page} + + # New path: direct ORM query with sort/filter/FTS + from db.models import DocumentTopic, Topic # noqa: PLC0415 (avoid circular at module top) + + stmt = select(Document).where(Document.user_id == current_user.id) + + # Topic filter (join-based, same as list_metadata) + if topic is not None: + stmt = ( + stmt.join(DocumentTopic, DocumentTopic.document_id == Document.id) + .join(Topic, Topic.id == DocumentTopic.topic_id) + .where(Topic.name == topic) + ) + + # Folder filter + if folder_id is not None: + try: + folder_uuid = uuid.UUID(folder_id) + except ValueError: + raise HTTPException(status_code=404, detail="Folder not found") + stmt = stmt.where(Document.folder_id == folder_uuid) + + # Sort + sort_col = Document.created_at # default: date + if sort == "name": + sort_col = Document.filename + elif sort == "size": + sort_col = Document.size_bytes + + order_fn = sort_col.asc if order == "asc" else sort_col.desc + stmt = stmt.order_by(order_fn()) + + # Full-text search — plainto_tsquery on extracted_text (PostgreSQL only) + # Falls back to unfiltered if the DB dialect doesn't support @@ (e.g. SQLite in test env) + fts_requested = q is not None and len(q) >= 2 + if fts_requested: + fts_stmt = stmt.where( + func.to_tsvector("english", func.coalesce(Document.extracted_text, "")).op("@@")( + func.plainto_tsquery("english", q) + ) + ) + try: + result = await session.execute(fts_stmt) + except Exception: + result = await session.execute(stmt) + else: + result = await session.execute(stmt) + docs_orm = result.scalars().all() + + # is_shared subquery + shared_result = await session.execute( + select(Share.document_id).where(Share.owner_id == current_user.id) + ) + shared_ids = {row[0] for row in shared_result.fetchall()} + + # Serialize + all_items = [] + for doc in docs_orm: + from services.storage import _doc_to_dict, _load_topic_names # noqa: PLC0415 + topic_names = await _load_topic_names(session, doc.id) + d = _doc_to_dict(doc, topic_names) + d["is_shared"] = doc.id in shared_ids + all_items.append(d) + + total = len(all_items) + start = (page - 1) * per_page + return { + "items": all_items[start : start + per_page], + "total": total, + "page": page, + "per_page": per_page, + } + + +# ── GET /api/documents/{doc_id} ─────────────────────────────────────────────── + +@router.get("/{doc_id}") +@account_limiter.limit("100/minute") +async def get_document( + request: Request, + doc_id: str, + session: AsyncSession = Depends(get_db), + current_user: User = Depends(get_regular_user), +): + """Return document metadata by ID. + + D-16: requires authenticated regular user. Asserts ownership — cross-user + access returns 404 (not 403) to avoid information leakage (T-03-11). + """ + request.state.current_user = current_user + try: + uid = uuid.UUID(doc_id) + except ValueError: + raise HTTPException(404, "Document not found") + + doc = await session.get(Document, uid) + if doc is None: + raise HTTPException(404, "Document not found") + + is_recipient = False + if doc.user_id != current_user.id: + # Allow recipients of an active share to view the document + share_result = await session.execute( + select(Share).where( + Share.document_id == uid, + Share.recipient_id == current_user.id, + ) + ) + if share_result.scalar_one_or_none() is None: + raise HTTPException(404, "Document not found") + is_recipient = True + + meta = await storage.get_metadata(session, doc_id) + if meta is None: + raise HTTPException(404, "Document not found") + # T-04-04-03: recipients get metadata only — extracted_text excluded (consistent with /shares/received) + if is_recipient: + meta.pop("extracted_text", None) + return meta + + +# ── PATCH /api/documents/{doc_id} ──────────────────────────────────────────── + +@router.patch("/{doc_id}") +@account_limiter.limit("100/minute") +async def patch_document( + request: Request, + doc_id: str, + body: DocumentPatch, + session: AsyncSession = Depends(get_db), + current_user: User = Depends(get_regular_user), +): + """Update document metadata (filename and/or folder_id). + + T-05-09-01: get_regular_user dep rejects admins (403) and unauthenticated (401). + T-05-09-01: ownership check — non-owner gets 404 to avoid leaking document IDs (D-16). + T-05-09-02: response uses storage.get_metadata() which excludes credentials_enc and + password_hash via the _doc_to_dict whitelist. + + At least one field must be provided — empty body returns 422. + folder_id=null moves the document to the root (no folder). + """ + request.state.current_user = current_user + try: + uid = uuid.UUID(doc_id) + except ValueError: + raise HTTPException(404, "Document not found") + + doc = await session.get(Document, uid) + if doc is None or doc.user_id != current_user.id: + raise HTTPException(404, "Document not found") + + # Require at least one field to be set (model_fields_set tracks provided fields) + if not body.model_fields_set: + raise HTTPException(422, "At least one field (filename, folder_id) must be provided") + + if "filename" in body.model_fields_set and body.filename is not None: + doc.filename = body.filename + + if "folder_id" in body.model_fields_set: + # folder_id=null → move to root (no folder); folder_id= → move to folder + if body.folder_id is not None: + target = await session.get(Folder, body.folder_id) + if target is None or target.user_id != current_user.id: + raise HTTPException(404, "Folder not found") + doc.folder_id = body.folder_id + + await session.commit() + + meta = await storage.get_metadata(session, doc_id) + if meta is None: + raise HTTPException(404, "Document not found") + return meta + + +# ── DELETE /api/documents/{doc_id} ─────────────────────────────────────────── + +@router.delete("/{doc_id}") +@account_limiter.limit("100/minute") +async def delete_document( + doc_id: str, + request: Request, + remove_only: bool = Query(default=False), + session: AsyncSession = Depends(get_db), + current_user: User = Depends(get_regular_user), +): + """Delete a document and decrement quota atomically. + + For cloud-stored documents: + - Default path: attempt cloud provider delete first; on failure return + {success: false, cloud_delete_failed: true} (HTTP 200) so the frontend + can offer a "Remove from app" fallback (T-06.2-03-02). + - remove_only=true: skip cloud delete, remove DB row only, skip quota decrement. + - Cloud docs always use skip_quota=True (never charged MinIO quota, T-06.2-03-01). + + D-16: requires authenticated regular user. Asserts ownership — cross-user + delete returns 404 (not 403) to avoid information leakage (T-03-11). + """ + request.state.current_user = current_user + try: + uid = uuid.UUID(doc_id) + except ValueError: + raise HTTPException(404, "Document not found") + + doc = await session.get(Document, uid) + if doc is None or doc.user_id != current_user.id: + raise HTTPException(404, "Document not found") + + is_cloud = doc.storage_backend != "minio" + _doc_size = doc.size_bytes + _doc_id = doc.id + _ip = get_client_ip(request) + + # Cloud routing: attempt provider delete unless remove_only is set + if is_cloud and not remove_only: + try: + import api.documents as _doc_pkg # late import allows test monkeypatching via api.documents + _gsb = _doc_pkg.get_storage_backend_for_document + cloud_backend = await _gsb(doc, current_user, session) + await cloud_backend.delete_object(doc.object_key) + except Exception as exc: + _log.warning("cloud_delete_failed", provider=doc.storage_backend, error=str(exc)) + return JSONResponse( + status_code=200, + content={ + "success": False, + "cloud_delete_failed": True, + "detail": "Cloud provider delete failed. You can remove from app only.", + }, + ) + + # auto_commit=False defers the commit so the audit log write below happens + # in the same transaction — avoids the split-transaction gap (WR-08). + ok = await storage.delete_document(session, doc_id, skip_quota=is_cloud, auto_commit=False) + if not ok: + raise HTTPException(404, "Document not found") + + # D-13: document deleted event — written in the same transaction as the delete (WR-08). + await write_audit_log( + session, + event_type="document.deleted", + user_id=current_user.id, + actor_id=current_user.id, + resource_id=_doc_id, + ip_address=_ip, + metadata_={"size_bytes": _doc_size}, + ) + await session.commit() + + return {"success": True} + + +# ── POST /api/documents/{doc_id}/classify ──────────────────────────────────── + +@router.post("/{doc_id}/classify") +@account_limiter.limit("100/minute") +async def classify_document( + request: Request, + doc_id: str, + session: AsyncSession = Depends(get_db), + current_user: User = Depends(get_regular_user), +): + """Re-queue a document for classification via Celery (D-11). + + Sets doc.status='processing', commits, dispatches extract_and_classify.delay(), + and returns {'document_id': str, 'status': 'processing'}. + + D-16: requires authenticated regular user. Asserts ownership — cross-user + classify returns 404 (not 403) to avoid information leakage (T-03-11). + T-07-10: ownership enforced here; IDOR returns 404 per STATE.md policy. + Placed in crud.py per D-08: same ownership-check pattern as get/patch/delete. + """ + request.state.current_user = current_user + try: + uid = uuid.UUID(doc_id) + except ValueError: + raise HTTPException(404, "Document not found") + + doc = await session.get(Document, uid) + if doc is None or doc.user_id != current_user.id: + raise HTTPException(404, "Document not found") + + doc.status = "processing" + await session.commit() + + extract_and_classify.delay(str(doc.id)) + + return {"document_id": str(doc.id), "status": "processing"} diff --git a/backend/api/documents/shared.py b/backend/api/documents/shared.py new file mode 100644 index 0000000..5cd61b9 --- /dev/null +++ b/backend/api/documents/shared.py @@ -0,0 +1,45 @@ +"""Shared constants and Pydantic request models for the documents API package. + +CODE-08: Single definition of _CLOUD_PROVIDERS, UploadUrlRequest, and DocumentPatch. +These are imported by upload.py and crud.py — never duplicated. + +T-05-06-01: _CLOUD_PROVIDERS is an allowlist frozenset; target_backend validated + against it (never against user-supplied strings). +T-05-09-01: DocumentPatch fields declared explicitly — mass assignment prevented. +T-05-09-02: filename_no_path_separators validator preserved verbatim (path traversal + defense at the API boundary — D-11 analysis: stays in Pydantic model). +""" +from __future__ import annotations + +import uuid +from typing import Optional + +from pydantic import BaseModel, Field, field_validator + +# Valid cloud backend slugs (T-05-06-01: validated against allowlist, not user-supplied string) +_CLOUD_PROVIDERS = frozenset({"google_drive", "onedrive", "nextcloud", "webdav"}) + + +class UploadUrlRequest(BaseModel): + filename: str + content_type: str + + +class DocumentPatch(BaseModel): + """Pydantic model for PATCH /api/documents/{doc_id}. + + Optional fields — model_fields_set distinguishes "not provided" from "set to null". + At least one field must be present in model_fields_set (enforced in the handler). + + T-05-09-01: explicit field declaration prevents mass assignment. + T-05-09-02: only filename and folder_id are accepted — no other fields can be set. + """ + filename: Optional[str] = Field(None, min_length=1, max_length=255) + folder_id: Optional[uuid.UUID] = None + + @field_validator("filename") + @classmethod + def filename_no_path_separators(cls, v: Optional[str]) -> Optional[str]: + if v is not None and ("/" in v or "\\" in v): + raise ValueError("filename must not contain path separators") + return v diff --git a/backend/api/documents/upload.py b/backend/api/documents/upload.py new file mode 100644 index 0000000..5e9823a --- /dev/null +++ b/backend/api/documents/upload.py @@ -0,0 +1,366 @@ +"""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", + }