feat(08-05): decompose documents API monolith into documents/ package — CODE-02 CODE-08
- 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>
This commit is contained in:
curo1305
co-authored by
Claude Sonnet 4.6
Claude Sonnet 4.6
parent
f01fb0e6d5
commit
81337bd9e3
@@ -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=<uuid> → 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="<filename>"
|
||||
Accept-Ranges: bytes
|
||||
Content-Length: <size>
|
||||
"""
|
||||
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,
|
||||
)
|
||||
@@ -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)
|
||||
@@ -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="<filename>"
|
||||
Accept-Ranges: bytes
|
||||
Content-Length: <size>
|
||||
"""
|
||||
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,
|
||||
)
|
||||
@@ -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=<uuid> → 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"}
|
||||
@@ -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
|
||||
@@ -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",
|
||||
}
|
||||
Reference in New Issue
Block a user