diff --git a/backend/app/api/docs/documents/initiate_v2.md b/backend/app/api/docs/documents/initiate_v2.md new file mode 100644 index 000000000..77d658dec --- /dev/null +++ b/backend/app/api/docs/documents/initiate_v2.md @@ -0,0 +1,11 @@ +Open a v2 upload session: get a URL and form fields to send a document straight to Kaapi's object storage. + +Step 1 of the three-step flow: + +1. `POST /api/v2/documents/uploads` with the filename — returns a `document_id`, an `upload_url`, and `upload_fields`. +2. Upload the file with a single `multipart/form-data` POST to `upload_url`: send every entry in `upload_fields` as a form field, then the file **last** in a field named `file`. No auth header on this call. +3. `PUT /api/v2/documents/{document_id}` to create the document — no body needed. + +The filename is validated here (an unsupported type fails before any upload) and travels with the file, so registration in step 3 uses it automatically. Nothing is persisted at this step — hence the `200` rather than a `201` — the `document_id` only becomes a document once you register it. + +Maximum file size is 25 MB, enforced by storage as the file uploads: a larger file is rejected outright with `400 EntityTooLarge` and nothing is stored. `upload_url` is valid for `expires_in` seconds (the effective value after server-side capping, which may be shorter than requested); open a new upload session if it lapses. diff --git a/backend/app/api/docs/documents/register_v2.md b/backend/app/api/docs/documents/register_v2.md new file mode 100644 index 000000000..30239c8be --- /dev/null +++ b/backend/app/api/docs/documents/register_v2.md @@ -0,0 +1,7 @@ +Register the document at the `document_id` issued by `POST /api/v2/documents/uploads`, from the file uploaded to its pre-signed URL. + +Final step of the v2 upload flow, and it takes no request body. The uploaded object is moved to its permanent location, the document row is created with the filename captured at step 1, and the response carries a fresh signed URL for reading the file back. + +Errors: `400` if nothing was uploaded for that `document_id` (the upload never happened or the URL lapsed); `409` if the `document_id` was already registered — open a new upload session in that case. + +The 25 MB cap is enforced by storage while the file uploads, so an oversized file never reaches this step. Document transformation is not available on v2. Use `POST /api/v1/documents` if you need a `target_format` conversion. diff --git a/backend/app/api/main.py b/backend/app/api/main.py index 6fd97fd1f..c7807ac19 100644 --- a/backend/app/api/main.py +++ b/backend/app/api/main.py @@ -12,6 +12,7 @@ cron, doc_transformation_job, documents, + documents_v2, evaluations, features, fine_tuning, @@ -89,8 +90,10 @@ # v2 API surface (mounted at settings.API_V2_STR). Only the endpoints that differ -# from v1 live here — currently the judged run trigger. Everything else stays v1. +# from v1 live here — currently the judged run trigger and the pre-signed document +# upload flow. Everything else stays v1. api_v2_router = APIRouter() +api_v2_router.include_router(documents_v2.router) api_v2_router.include_router(evaluations_v2_router) api_v2_router.include_router(evaluations_dataset_v2_router) api_v2_router.include_router(evaluations_prompt_improvement_v2_router) diff --git a/backend/app/api/routes/documents_v2.py b/backend/app/api/routes/documents_v2.py new file mode 100644 index 000000000..9f3a39d33 --- /dev/null +++ b/backend/app/api/routes/documents_v2.py @@ -0,0 +1,97 @@ +"""v2 document upload: pre-signed POST to a pending key, then registration.""" + +from pathlib import Path +from uuid import UUID, uuid4 + +from fastapi import APIRouter, Depends +from fastapi import Path as FastPath + +from app.api.deps import AuthContextDep, SessionDep +from app.api.permissions import Permission, require_permission +from app.core.cloud import get_cloud_storage +from app.models import ( + DocumentPublic, + DocumentUploadInitiateResponse, + DocumentUploadRequest, +) +from app.services.collections.helpers import MAX_DOC_SIZE_MB +from app.services.documents.registration import ( + register_uploaded_document, + validate_filename_format, +) +from app.utils import APIResponse, load_description + +router = APIRouter(prefix="/documents", tags=["Documents v2"]) + +UPLOAD_URL_EXPIRY_SECONDS = 3600 +MAX_UPLOAD_BYTES = MAX_DOC_SIZE_MB * 1024 * 1024 + + +@router.post( + "/uploads", + description=load_description("documents/initiate_v2.md"), + response_model=APIResponse[DocumentUploadInitiateResponse], + dependencies=[Depends(require_permission(Permission.REQUIRE_PROJECT))], +) +def create_upload_url( + session: SessionDep, + current_user: AuthContextDep, + request: DocumentUploadRequest, +) -> APIResponse[DocumentUploadInitiateResponse]: + validate_filename_format(request.filename) + + storage = get_cloud_storage(session=session, project_id=current_user.project_.id) + document_id = uuid4() + ticket = storage.create_upload_ticket( + Path(str(document_id)), + filename=request.filename, + max_bytes=MAX_UPLOAD_BYTES, + expires_in=UPLOAD_URL_EXPIRY_SECONDS, + ) + + return APIResponse[DocumentUploadInitiateResponse].success_response( + DocumentUploadInitiateResponse( + document_id=document_id, + upload_url=ticket.url, + upload_fields=ticket.fields, + expires_in=ticket.expires_in, + ), + metadata={ + "next_step": ( + f"Upload the file to the pre-signed S3 URL with the provided fields " + f"(file part last), then PUT /api/v2/documents/{document_id} to register the document." + ) + }, + ) + + +@router.put( + "/{document_id}", + description=load_description("documents/register_v2.md"), + status_code=201, + response_model=APIResponse[DocumentPublic], + dependencies=[Depends(require_permission(Permission.REQUIRE_PROJECT))], +) +def register_document( + session: SessionDep, + current_user: AuthContextDep, + document_id: UUID = FastPath( + description="Document id issued by the upload session" + ), +) -> APIResponse[DocumentPublic]: + document = register_uploaded_document( + session=session, + project_id=current_user.project_.id, + document_id=document_id, + ) + + storage = get_cloud_storage(session=session, project_id=current_user.project_.id) + document_schema = DocumentPublic.model_validate(document, from_attributes=True) + document_schema.signed_url = storage.get_signed_url(document.object_store_url) + + return APIResponse[DocumentPublic].success_response( + document_schema, + metadata={ + "note": "Document registered. The upload URL is spent and cannot be reused." + }, + ) diff --git a/backend/app/core/cloud/__init__.py b/backend/app/core/cloud/__init__.py index b6b0b08ec..2611b76bd 100644 --- a/backend/app/core/cloud/__init__.py +++ b/backend/app/core/cloud/__init__.py @@ -3,6 +3,9 @@ AmazonCloudStorageClient, CloudStorage, CloudStorageError, + ObjectNotFoundError, + StoredObject, + UploadTicket, get_cloud_storage, upload_audio_to_gcs, ) diff --git a/backend/app/core/cloud/storage.py b/backend/app/core/cloud/storage.py index a65d43fc5..9dade4f54 100644 --- a/backend/app/core/cloud/storage.py +++ b/backend/app/core/cloud/storage.py @@ -7,10 +7,10 @@ import functools as ft from pathlib import Path from dataclasses import dataclass, asdict -from urllib.parse import ParseResult, quote, urlparse, urlunparse +from urllib.parse import ParseResult, quote, unquote, urlparse, urlunparse from abc import ABC, abstractmethod -from typing import Any +from typing import Any, NamedTuple import boto3 from fastapi import UploadFile from botocore.exceptions import ClientError @@ -48,6 +48,41 @@ class CloudStorageError(Exception): pass +class ObjectNotFoundError(CloudStorageError): + pass + + +MISSING_OBJECT_CODES = ("404", "NoSuchKey") + + +def _to_storage_error(err: ClientError, url: str) -> CloudStorageError: + """Map a botocore ClientError onto the storage exception hierarchy.""" + message = f'AWS Error: "{err}" ({url})' + code = err.response.get("Error", {}).get("Code") + if code in MISSING_OBJECT_CODES: + return ObjectNotFoundError(message) + return CloudStorageError(message) + + +# S3 user-metadata key that carries the client's filename, signed into the upload +# ticket so the client cannot change it and registration can read it back. +FILENAME_METADATA_KEY = "filename" + + +class UploadTicket(NamedTuple): + url: str + # Form fields the client must POST alongside the file, the file part last. + fields: dict[str, str] + # Effective expiry after capping, which may be shorter than what the caller asked for. + expires_in: int + + +class StoredObject(NamedTuple): + size_kb: float + # The client filename recovered from object metadata, None if the object carries none. + filename: str | None + + class AmazonCloudStorageClient: @ft.cached_property def client(self): @@ -137,11 +172,37 @@ def from_url(cls, url: str): return cls(Bucket=url.netloc, Key=str(path)) +# Unregistered uploads. Unrelated to the staging *environment*: this is a holding +# area, and one literal-prefix lifecycle rule reaps it for every project. +PENDING_PREFIX = "pending" +PENDING_TTL_DAYS = 1 + + class CloudStorage(ABC): def __init__(self, project_id: int, storage_path: UUID): self.project_id = project_id self.storage_path = str(storage_path) + def url_for(self, file_path: Path, is_pending: bool = False) -> SimpleStorageName: + """Resolve a project-relative path into a fully qualified storage name. + + The single place a key is built — callers never join storage_path themselves. + + Args: + file_path: Path relative to the project's storage root. + is_pending: Place the key under ``PENDING_PREFIX`` — the holding area for + uploads that have not been registered yet. An S3 lifecycle rule deletes + everything there after ``PENDING_TTL_DAYS`` (1 day), so pass True only + for objects that are meant to disappear if registration never happens. + Registered documents use the default (False) and are never expired. + """ + if file_path.is_absolute(): + raise ValueError("file_path must be relative to the project's storage root") + roots = ( + (PENDING_PREFIX, self.storage_path) if is_pending else (self.storage_path,) + ) + return SimpleStorageName(Path(*roots, file_path).as_posix()) + @abstractmethod def put(self, source: UploadFile, filepath: Path) -> SimpleStorageName: """Upload a file to storage""" @@ -162,6 +223,11 @@ def get_file_size_kb(self, url: str) -> float: """Return the file size in KB""" pass + @abstractmethod + def head(self, url: str) -> StoredObject: + """Return the object's size and the filename recorded in its metadata.""" + pass + @abstractmethod def get_signed_url( self, url: str, expires_in: int = 3600, filename: str | None = None @@ -169,6 +235,29 @@ def get_signed_url( """Generate a signed URL with an optional expiry""" pass + @abstractmethod + def create_upload_ticket( + self, + file_path: Path, + *, + filename: str, + max_bytes: int, + expires_in: int = 3600, + ) -> UploadTicket: + """Create a pre-signed POST the client uploads a file directly to. + + The ticket caps the body at max_bytes (S3 rejects anything larger at the + edge) and pins filename into signed metadata so the client cannot change it. + Always resolves under PENDING_PREFIX: nothing is ever presigned to a final + key, or an abandoned upload would be indistinguishable from a document. + """ + pass + + @abstractmethod + def copy(self, source_url: str, destination: Path) -> SimpleStorageName: + """Server-side copy of an existing object to a project-relative path""" + pass + @abstractmethod def delete(self, url: str) -> None: """Delete a file from storage""" @@ -181,10 +270,7 @@ def __init__(self, project_id: int, storage_path: UUID): self.aws = AmazonCloudStorageClient() def put(self, source: UploadFile, file_path: Path) -> SimpleStorageName: - if file_path.is_absolute(): - raise ValueError("file_path must be relative to the project's storage root") - key = Path(self.storage_path) / file_path - destination = SimpleStorageName(key.as_posix()) + destination = self.url_for(file_path) kwargs = asdict(destination) try: @@ -225,7 +311,7 @@ def stream(self, url: str) -> StreamingBody: f"{{'project_id': '{self.project_id}', 'bucket': '{_mask(name.Bucket)}', 'key': '{_mask(name.Key)}', 'error': '{str(err)}'}}", exc_info=True, ) - raise CloudStorageError(f'AWS Error: "{err}" ({url})') from err + raise _to_storage_error(err, url) from err def get(self, url: str) -> bytes: name = SimpleStorageName.from_url(url) @@ -244,7 +330,7 @@ def get(self, url: str) -> bytes: f"{{'project_id': '{self.project_id}', 'bucket': '{_mask(name.Bucket)}', 'key': '{_mask(name.Key)}', 'error': '{str(err)}'}}", exc_info=True, ) - raise CloudStorageError(f'AWS Error: "{err}" ({url})') from err + raise _to_storage_error(err, url) from err def get_file_size_kb(self, url: str) -> float: name = SimpleStorageName.from_url(url) @@ -264,7 +350,25 @@ def get_file_size_kb(self, url: str) -> float: f"{{'project_id': '{self.project_id}', 'bucket': '{_mask(name.Bucket)}', 'key': '{_mask(name.Key)}', 'error': '{str(err)}'}}", exc_info=True, ) - raise CloudStorageError(f'AWS Error: "{err}" ({url})') from err + raise _to_storage_error(err, url) from err + + def head(self, url: str) -> StoredObject: + name = SimpleStorageName.from_url(url) + try: + response = self.aws.client.head_object(**asdict(name)) + except ClientError as err: + logger.error( + f"[AmazonCloudStorage.head] AWS head object error | " + f"{{'project_id': '{self.project_id}', 'bucket': '{_mask(name.Bucket)}', 'key': '{_mask(name.Key)}', 'error': '{str(err)}'}}", + exc_info=True, + ) + raise _to_storage_error(err, url) from err + + encoded = response.get("Metadata", {}).get(FILENAME_METADATA_KEY) + return StoredObject( + size_kb=round(response["ContentLength"] / 1024, 2), + filename=unquote(encoded) if encoded else None, + ) # Maximum allowed expiry for signed URLs (24 hours) MAX_SIGNED_URL_EXPIRY = 86400 @@ -305,6 +409,64 @@ def get_signed_url( ) raise CloudStorageError(f'AWS Error: "{err}" ({url})') from err + def create_upload_ticket( + self, + file_path: Path, + *, + filename: str, + max_bytes: int, + expires_in: int = 3600, + ) -> UploadTicket: + """ + Pre-signed POST the client uploads to, under PENDING_PREFIX. The size cap is + enforced by S3 at the edge, and the filename is signed into metadata so it + cannot be swapped between issuing the ticket and registration. + """ + expires_in = min(expires_in, self.MAX_SIGNED_URL_EXPIRY) + + name = self.url_for(file_path, is_pending=True) + encoded = quote(filename) + meta_field = f"x-amz-meta-{FILENAME_METADATA_KEY}" + try: + post = self.aws.client.generate_presigned_post( + name.Bucket, + name.Key, + Fields={meta_field: encoded}, + Conditions=[ + ["content-length-range", 1, max_bytes], + {meta_field: encoded}, + ], + ExpiresIn=expires_in, + ) + return UploadTicket( + url=post["url"], fields=post["fields"], expires_in=expires_in + ) + except ClientError as err: + logger.error( + f"[AmazonCloudStorage.create_upload_ticket] AWS presign error | " + f"{{'project_id': '{self.project_id}', 'bucket': '{_mask(name.Bucket)}', 'key': '{_mask(name.Key)}', 'error': '{str(err)}'}}", + exc_info=True, + ) + raise _to_storage_error(err, str(name)) from err + + def copy(self, source_url: str, destination: Path) -> SimpleStorageName: + source = SimpleStorageName.from_url(source_url) + target = self.url_for(destination) + try: + self.aws.client.copy_object( + Bucket=target.Bucket, + Key=target.Key, + CopySource={"Bucket": source.Bucket, "Key": source.Key}, + ) + return target + except ClientError as err: + logger.error( + f"[AmazonCloudStorage.copy] AWS copy error | " + f"{{'project_id': '{self.project_id}', 'bucket': '{_mask(target.Bucket)}', 'source_key': '{_mask(source.Key)}', 'key': '{_mask(target.Key)}', 'error': '{str(err)}'}}", + exc_info=True, + ) + raise _to_storage_error(err, source_url) from err + def delete(self, url: str) -> None: name = SimpleStorageName.from_url(url) kwargs = asdict(name) @@ -320,7 +482,7 @@ def delete(self, url: str) -> None: f"{{'project_id': '{self.project_id}', 'bucket': '{_mask(name.Bucket)}', 'key': '{_mask(name.Key)}', 'error': '{str(err)}'}}", exc_info=True, ) - raise CloudStorageError(f'AWS Error: "{err}" ({url})') from err + raise _to_storage_error(err, url) from err def get_cloud_storage(session: Session, project_id: int) -> CloudStorage: diff --git a/backend/app/crud/document/document.py b/backend/app/crud/document/document.py index 1acf6cda3..7790624b5 100644 --- a/backend/app/crud/document/document.py +++ b/backend/app/crud/document/document.py @@ -33,6 +33,10 @@ def read_one(self, doc_id: UUID) -> Document: return result + def exists(self, doc_id: UUID) -> bool: + # Ignores deleted_at and project scope: the PK stays taken after a soft delete. + return self.session.get(Document, doc_id) is not None + def read_many( self, skip: int | None = None, diff --git a/backend/app/models/__init__.py b/backend/app/models/__init__.py index 863506047..8a1559573 100644 --- a/backend/app/models/__init__.py +++ b/backend/app/models/__init__.py @@ -86,6 +86,8 @@ DocTransformationJobsPublic, Document, DocumentPublic, + DocumentUploadInitiateResponse, + DocumentUploadRequest, DocumentUploadResponse, TransformationJobInfo, TransformedDocumentPublic, diff --git a/backend/app/models/document.py b/backend/app/models/document.py index 3f3c80996..37e051daf 100644 --- a/backend/app/models/document.py +++ b/backend/app/models/document.py @@ -117,6 +117,27 @@ class DocumentUploadResponse(DocumentPublic): transformation_job: TransformationJobInfo | None = None +class DocumentUploadRequest(SQLModel): + filename: str = Field( + min_length=1, + max_length=255, + description="Original filename including its extension, e.g. report.pdf", + ) + + +class DocumentUploadInitiateResponse(SQLModel): + document_id: UUID = Field( + description="Identifier to register the document under; the registration endpoint takes it as a path parameter" + ) + upload_url: str = Field(description="URL to POST the file to") + upload_fields: dict[str, str] = Field( + description="Form fields to send with the file in the multipart POST, the file part last" + ) + expires_in: int = Field( + description="Effective lifetime of the upload URL in seconds, after server-side capping" + ) + + class DocTransformationJobPublic(SQLModel): job_id: UUID source_document_id: UUID diff --git a/backend/app/services/documents/registration.py b/backend/app/services/documents/registration.py new file mode 100644 index 000000000..16cb3c5ca --- /dev/null +++ b/backend/app/services/documents/registration.py @@ -0,0 +1,80 @@ +"""v2 upload policy: promote what the client uploaded into a document row.""" + +from pathlib import Path +from uuid import UUID + +from fastapi import HTTPException +from sqlalchemy.exc import IntegrityError +from sqlmodel import Session + +from app.core.cloud import get_cloud_storage +from app.core.cloud.storage import ObjectNotFoundError +from app.crud import DocumentCrud +from app.models import Document +from app.services.doctransform.registry import get_file_format + +DUPLICATE_DOCUMENT_DETAIL = ( + "This document_id is already registered. Request a new upload URL." +) +MISSING_UPLOAD_DETAIL = ( + "No uploaded file found for this document_id. Upload the file to the " + "pre-signed URL before registering it." +) + + +def validate_filename_format(filename: str) -> str: + """Resolve the document format from the extension; HTTPException(400) if unsupported.""" + try: + return get_file_format(filename) + except ValueError as e: + raise HTTPException(status_code=400, detail=str(e)) + + +def register_uploaded_document( + *, + session: Session, + project_id: int, + document_id: UUID, +) -> Document: + """Promote an uploaded object into a document row, moving it to its final key. + + The filename comes from the object's own signed metadata, not the request, so + it cannot differ from what the upload ticket was issued for. Size is already + capped by the ticket, so the copied object is measured only to record its size. + """ + document_crud = DocumentCrud(session, project_id) + if document_crud.exists(document_id): + raise HTTPException(status_code=409, detail=DUPLICATE_DOCUMENT_DETAIL) + + storage = get_cloud_storage(session=session, project_id=project_id) + pending_url = str(storage.url_for(Path(str(document_id)), is_pending=True)) + + # Copy first, then read the frozen final key: the pending object stays writable + # through its ticket, so measuring it directly would race the copy. + try: + object_store_url = storage.copy(pending_url, Path(str(document_id))) + except ObjectNotFoundError: + raise HTTPException(status_code=400, detail=MISSING_UPLOAD_DETAIL) + + stored = storage.head(str(object_store_url)) + filename = stored.filename or str(document_id) + + try: + document = document_crud.update( + Document( + id=document_id, + fname=filename, + file_size_kb=stored.size_kb, + object_store_url=str(object_store_url), + project_id=project_id, + ) + ) + except IntegrityError: + # Two completions can both clear exists(); the PK decides, and the object is the winner's. + session.rollback() + raise HTTPException(status_code=409, detail=DUPLICATE_DOCUMENT_DETAIL) + + # Only once the row is committed: a failed insert leaves the upload retryable, + # and an abandoned pending object expires on its own. + storage.delete(pending_url) + return document diff --git a/backend/app/tests/api/routes/documents/test_route_document_register_v2.py b/backend/app/tests/api/routes/documents/test_route_document_register_v2.py new file mode 100644 index 000000000..62fd67a3c --- /dev/null +++ b/backend/app/tests/api/routes/documents/test_route_document_register_v2.py @@ -0,0 +1,224 @@ +from unittest.mock import patch +from urllib.parse import quote +from uuid import UUID, uuid4 + +import pytest +import requests +from botocore.exceptions import ClientError +from fastapi.testclient import TestClient +from httpx import Response +from moto import mock_aws +from sqlmodel import Session + +from app.core.cloud import AmazonCloudStorageClient +from app.core.config import settings +from app.core.db import engine +from app.core.util import now +from app.crud import DocumentCrud +from app.models import Document +from app.tests.utils.auth import TestAuthContext +from app.tests.utils.document import DocumentMaker + +DOCUMENTS_ROUTE = f"{settings.API_V2_STR}/documents" +UPLOADS_ROUTE = f"{DOCUMENTS_ROUTE}/uploads" + + +def pending_key(auth: TestAuthContext, document_id: UUID) -> str: + # No extension: the filename rides in signed metadata, not in the key. + return f"pending/{auth.project.storage_path}/{document_id}" + + +def final_key(auth: TestAuthContext, document_id: UUID) -> str: + return f"{auth.project.storage_path}/{document_id}" + + +def put_pending( + auth: TestAuthContext, document_id: UUID, body: bytes, filename: str +) -> None: + """Simulate a client upload: the object plus the filename the ticket would have pinned.""" + AmazonCloudStorageClient().client.put_object( + Bucket=settings.AWS_S3_BUCKET, + Key=pending_key(auth, document_id), + Body=body, + Metadata={"filename": quote(filename)}, + ) + + +def assert_absent(key: str) -> None: + with pytest.raises(ClientError) as excinfo: + AmazonCloudStorageClient().client.head_object( + Bucket=settings.AWS_S3_BUCKET, Key=key + ) + assert excinfo.value.response["Error"]["Code"] == "404" + + +def register(client: TestClient, auth: TestAuthContext, document_id: UUID) -> Response: + return client.put( + f"{DOCUMENTS_ROUTE}/{document_id}", + headers={"X-API-KEY": auth.key}, + ) + + +@mock_aws +@pytest.mark.usefixtures("aws_credentials") +class TestDocumentRegisterV2: + def test_registers_pending_object_under_its_final_key( + self, + db: Session, + client: TestClient, + user_api_key: TestAuthContext, + ) -> None: + AmazonCloudStorageClient().create() + document_id = uuid4() + put_pending(user_api_key, document_id, b"x" * 2048, "report.pdf") + + response = register(client, user_api_key, document_id) + + assert response.status_code == 201 + data = response.json()["data"] + assert data["id"] == str(document_id) + assert data["fname"] == "report.pdf" + assert data["signed_url"] + assert "cannot be reused" in response.json()["metadata"]["note"] + + document = db.get(Document, document_id) + assert document is not None + assert document.fname == "report.pdf" + assert document.file_size_kb == 2.0 + assert document.object_store_url == ( + f"s3://{settings.AWS_S3_BUCKET}/{final_key(user_api_key, document_id)}" + ) + assert document.project_id == user_api_key.project_id + + assert_absent(pending_key(user_api_key, document_id)) + + def test_missing_object_is_rejected( + self, + db: Session, + client: TestClient, + user_api_key: TestAuthContext, + ) -> None: + AmazonCloudStorageClient().create() + document_id = uuid4() + + response = register(client, user_api_key, document_id) + + assert response.status_code == 400 + assert "No uploaded file found" in response.json()["error"] + assert db.get(Document, document_id) is None + + def test_duplicate_document_id_is_rejected( + self, + db: Session, + client: TestClient, + user_api_key: TestAuthContext, + ) -> None: + AmazonCloudStorageClient().create() + existing = next(DocumentMaker(project_id=user_api_key.project_id, session=db)) + db.add(existing) + db.commit() + put_pending(user_api_key, existing.id, b"x" * 1024, "report.pdf") + + response = register(client, user_api_key, existing.id) + + assert response.status_code == 409 + assert "already registered" in response.json()["error"] + + db.refresh(existing) + assert existing.fname != "report.pdf" + + def test_concurrent_registration_loses_the_insert_race( + self, + db: Session, + client: TestClient, + user_api_key: TestAuthContext, + ) -> None: + AmazonCloudStorageClient().create() + winner = next(DocumentMaker(project_id=user_api_key.project_id, session=db)) + document_id, original_fname = winner.id, winner.fname + # Committed outside the test transaction: the winner of the race is another + # request's session, so the request under test can only hit the PK constraint. + with Session(engine) as outside: + outside.add(winner) + outside.commit() + put_pending(user_api_key, document_id, b"x" * 1024, "report.pdf") + + try: + # exists() returning False simulates the racing request that also saw no row. + with patch.object(DocumentCrud, "exists", return_value=False): + response = register(client, user_api_key, document_id) + + assert response.status_code == 409 + assert "already registered" in response.json()["error"] + + survivor = db.get(Document, document_id) + assert survivor is not None + assert survivor.fname == original_fname + finally: + with Session(engine) as cleanup: + row = cleanup.get(Document, document_id) + if row is not None: + cleanup.delete(row) + cleanup.commit() + + def test_soft_deleted_document_id_is_rejected( + self, + db: Session, + client: TestClient, + user_api_key: TestAuthContext, + ) -> None: + AmazonCloudStorageClient().create() + deleted = next(DocumentMaker(project_id=user_api_key.project_id, session=db)) + deleted.deleted_at = now() + db.add(deleted) + db.commit() + + response = register(client, user_api_key, deleted.id) + + assert response.status_code == 409 + assert "already registered" in response.json()["error"] + + def test_missing_api_key_is_unauthorized(self, client: TestClient) -> None: + response = client.put(f"{DOCUMENTS_ROUTE}/{uuid4()}") + + assert response.status_code == 401 + + +@mock_aws +@pytest.mark.usefixtures("aws_credentials") +class TestDocumentUploadRoundTripV2: + def test_uploads_then_post_then_register( + self, + db: Session, + client: TestClient, + user_api_key: TestAuthContext, + ) -> None: + AmazonCloudStorageClient().create() + + init = client.post( + UPLOADS_ROUTE, + headers={"X-API-KEY": user_api_key.key}, + json={"filename": "handbook.pdf"}, + ) + assert init.status_code == 200 + data = init.json()["data"] + document_id = UUID(data["document_id"]) + + upload = requests.post( + data["upload_url"], + data=data["upload_fields"], + files={"file": ("handbook.pdf", b"y" * 3072)}, + ) + assert upload.status_code in (200, 204) + + response = register(client, user_api_key, document_id) + + assert response.status_code == 201 + assert response.json()["data"]["id"] == str(document_id) + + document = db.get(Document, document_id) + assert document is not None + assert document.object_store_url == ( + f"s3://{settings.AWS_S3_BUCKET}/{final_key(user_api_key, document_id)}" + ) + assert_absent(pending_key(user_api_key, document_id)) diff --git a/backend/app/tests/api/routes/documents/test_route_document_uploads_v2.py b/backend/app/tests/api/routes/documents/test_route_document_uploads_v2.py new file mode 100644 index 000000000..2adaf7e59 --- /dev/null +++ b/backend/app/tests/api/routes/documents/test_route_document_uploads_v2.py @@ -0,0 +1,142 @@ +from urllib.parse import quote + +import pytest +from fastapi.testclient import TestClient +from moto import mock_aws +from sqlmodel import Session + +from app.core.cloud import AmazonCloudStorageClient +from app.core.config import settings +from app.models import Document +from app.tests.utils.auth import TestAuthContext + +UPLOADS_ROUTE = f"{settings.API_V2_STR}/documents/uploads" + + +def pending_key(auth: TestAuthContext, document_id: str) -> str: + # No extension: the filename lives in signed metadata, not in the key. + return f"pending/{auth.project.storage_path}/{document_id}" + + +@mock_aws +@pytest.mark.usefixtures("aws_credentials") +class TestDocumentUploadsV2: + def test_returns_presigned_post_for_pending_key( + self, + db: Session, + client: TestClient, + user_api_key: TestAuthContext, + ) -> None: + AmazonCloudStorageClient().create() + + response = client.post( + UPLOADS_ROUTE, + headers={"X-API-KEY": user_api_key.key}, + json={"filename": "quarterly-report.pdf"}, + ) + + assert response.status_code == 200 + data = response.json()["data"] + assert data["expires_in"] == 3600 + + fields = data["upload_fields"] + assert fields["key"] == pending_key(user_api_key, data["document_id"]) + assert "x-amz-signature" in fields + assert "register" in response.json()["metadata"]["next_step"] + + def test_upload_target_is_the_pending_key_not_the_final_one( + self, + db: Session, + client: TestClient, + user_api_key: TestAuthContext, + ) -> None: + AmazonCloudStorageClient().create() + + response = client.post( + UPLOADS_ROUTE, + headers={"X-API-KEY": user_api_key.key}, + json={"filename": "quarterly-report.pdf"}, + ) + + data = response.json()["data"] + key = data["upload_fields"]["key"] + assert key == pending_key(user_api_key, data["document_id"]) + assert key != f"{user_api_key.project.storage_path}/{data['document_id']}" + + def test_filename_is_pinned_in_signed_metadata( + self, + db: Session, + client: TestClient, + user_api_key: TestAuthContext, + ) -> None: + AmazonCloudStorageClient().create() + + response = client.post( + UPLOADS_ROUTE, + headers={"X-API-KEY": user_api_key.key}, + json={"filename": "Quarterly Report.pdf"}, + ) + + # Signed as a field so the client cannot change it; URL-encoded for the header. + fields = response.json()["data"]["upload_fields"] + assert fields["x-amz-meta-filename"] == quote("Quarterly Report.pdf") + + def test_does_not_create_document_row( + self, + db: Session, + client: TestClient, + user_api_key: TestAuthContext, + ) -> None: + AmazonCloudStorageClient().create() + + response = client.post( + UPLOADS_ROUTE, + headers={"X-API-KEY": user_api_key.key}, + json={"filename": "notes.txt"}, + ) + + from uuid import UUID + + document_id = UUID(response.json()["data"]["document_id"]) + assert db.get(Document, document_id) is None + + def test_unsupported_extension_is_rejected( + self, + client: TestClient, + user_api_key: TestAuthContext, + ) -> None: + response = client.post( + UPLOADS_ROUTE, + headers={"X-API-KEY": user_api_key.key}, + json={"filename": "notes.xyz"}, + ) + + assert response.status_code == 400 + assert "Unsupported file extension: .xyz" in response.json()["error"] + + def test_blank_filename_is_rejected( + self, + client: TestClient, + user_api_key: TestAuthContext, + ) -> None: + response = client.post( + UPLOADS_ROUTE, + headers={"X-API-KEY": user_api_key.key}, + json={"filename": ""}, + ) + + assert response.status_code == 422 + + def test_missing_api_key_is_unauthorized(self, client: TestClient) -> None: + response = client.post(UPLOADS_ROUTE, json={"filename": "notes.txt"}) + + assert response.status_code == 401 + + def test_invalid_api_key_is_unauthorized(self, client: TestClient) -> None: + response = client.post( + UPLOADS_ROUTE, + headers={"X-API-KEY": "ApiKey not-a-real-key"}, + json={"filename": "notes.txt"}, + ) + + assert response.status_code == 401 diff --git a/backend/app/tests/conftest.py b/backend/app/tests/conftest.py index dfa6fa18c..4a4ce9f41 100644 --- a/backend/app/tests/conftest.py +++ b/backend/app/tests/conftest.py @@ -78,6 +78,16 @@ def seed_baseline( yield +@pytest.fixture +def aws_credentials(monkeypatch: pytest.MonkeyPatch) -> None: + """Dummy AWS creds for moto. setenv (not os.environ) so they unwind after the test.""" + monkeypatch.setenv("AWS_ACCESS_KEY_ID", "testing") + monkeypatch.setenv("AWS_SECRET_ACCESS_KEY", "testing") + monkeypatch.setenv("AWS_SECURITY_TOKEN", "testing") + monkeypatch.setenv("AWS_SESSION_TOKEN", "testing") + monkeypatch.setenv("AWS_DEFAULT_REGION", settings.AWS_DEFAULT_REGION) + + @pytest.fixture(scope="function") def client(db: Session) -> Generator[TestClient, None, None]: app.dependency_overrides[get_db] = lambda: db diff --git a/backend/app/tests/core/cloud/test_storage.py b/backend/app/tests/core/cloud/test_storage.py index e7ffb8cab..f5d7f7a54 100644 --- a/backend/app/tests/core/cloud/test_storage.py +++ b/backend/app/tests/core/cloud/test_storage.py @@ -1,14 +1,26 @@ """Tests for app.core.cloud.storage helpers.""" +from pathlib import Path from types import SimpleNamespace from unittest.mock import MagicMock, patch +from urllib.parse import quote from uuid import uuid4 +import pytest +from botocore.exceptions import ClientError +from moto import mock_aws + from app.core.cloud.storage import ( GCS_SCOPES, AmazonCloudStorage, + AmazonCloudStorageClient, + CloudStorageError, + ObjectNotFoundError, + SimpleStorageName, + _to_storage_error, build_gcp_sa_credentials, ) +from app.core.config import settings def test_build_gcp_sa_credentials_passes_key_and_scopes(): @@ -22,6 +34,175 @@ def test_build_gcp_sa_credentials_passes_key_and_scopes(): assert creds is mock_from_info.return_value +def client_error(code: str, operation: str = "HeadObject") -> ClientError: + return ClientError({"Error": {"Code": code, "Message": code}}, operation) + + +class TestUrlFor: + def test_joins_the_projects_storage_path(self) -> None: + storage = AmazonCloudStorage(project_id=1, storage_path=uuid4()) + + name = storage.url_for(Path("report.pdf")) + + assert name.Key == f"{storage.storage_path}/report.pdf" + assert name.Bucket == settings.AWS_S3_BUCKET + + def test_pending_prefix_leads_the_key(self) -> None: + storage = AmazonCloudStorage(project_id=1, storage_path=uuid4()) + + name = storage.url_for(Path("report.pdf"), is_pending=True) + + # Ahead of storage_path, not after it: one literal-prefix rule must match every project. + assert name.Key == f"pending/{storage.storage_path}/report.pdf" + + def test_absolute_path_is_rejected(self) -> None: + storage = AmazonCloudStorage(project_id=1, storage_path=uuid4()) + + with pytest.raises(ValueError, match="must be relative"): + storage.url_for(Path("/etc/passwd")) + + +class TestToStorageError: + @pytest.mark.parametrize("code", ["404", "NoSuchKey"]) + def test_missing_object_codes_become_object_not_found(self, code: str) -> None: + error = _to_storage_error(client_error(code), "s3://bucket/key") + + assert isinstance(error, ObjectNotFoundError) + assert "s3://bucket/key" in str(error) + + def test_other_codes_stay_generic(self) -> None: + error = _to_storage_error(client_error("AccessDenied"), "s3://bucket/key") + + assert isinstance(error, CloudStorageError) + assert not isinstance(error, ObjectNotFoundError) + + +@mock_aws +@pytest.mark.usefixtures("aws_credentials") +class TestCreateUploadTicket: + def test_targets_the_pending_key(self) -> None: + storage = AmazonCloudStorage(project_id=1, storage_path=uuid4()) + document_id = uuid4() + + ticket = storage.create_upload_ticket( + Path(str(document_id)), filename="report.pdf", max_bytes=25 * 1024 * 1024 + ) + + assert ticket.fields["key"] == f"pending/{storage.storage_path}/{document_id}" + assert "x-amz-signature" in ticket.fields + + def test_filename_is_pinned_url_encoded(self) -> None: + storage = AmazonCloudStorage(project_id=1, storage_path=uuid4()) + + ticket = storage.create_upload_ticket( + Path(str(uuid4())), filename="my report.pdf", max_bytes=1024 + ) + + assert ticket.fields["x-amz-meta-filename"] == quote("my report.pdf") + + def test_expiry_is_capped_at_one_day(self) -> None: + storage = AmazonCloudStorage(project_id=1, storage_path=uuid4()) + + ticket = storage.create_upload_ticket( + Path("f"), filename="a.pdf", max_bytes=1024, expires_in=7 * 24 * 3600 + ) + + assert ticket.expires_in == 86400 + + def test_shorter_expiry_is_preserved(self) -> None: + storage = AmazonCloudStorage(project_id=1, storage_path=uuid4()) + + ticket = storage.create_upload_ticket( + Path("f"), filename="a.pdf", max_bytes=1024, expires_in=600 + ) + + assert ticket.expires_in == 600 + + def test_aws_error_is_wrapped(self) -> None: + storage = AmazonCloudStorage(project_id=1, storage_path=uuid4()) + + with patch.object( + storage.aws.client, + "generate_presigned_post", + side_effect=client_error("AccessDenied", "PutObject"), + ): + with pytest.raises(CloudStorageError, match="AccessDenied"): + storage.create_upload_ticket( + Path("f"), filename="a.pdf", max_bytes=1024 + ) + + +@mock_aws +@pytest.mark.usefixtures("aws_credentials") +class TestHead: + def test_returns_size_and_decoded_filename(self) -> None: + aws = AmazonCloudStorageClient() + aws.create() + storage = AmazonCloudStorage(project_id=1, storage_path=uuid4()) + name = storage.url_for(Path(str(uuid4())), is_pending=True) + aws.client.put_object( + Bucket=name.Bucket, + Key=name.Key, + Body=b"x" * 2048, + Metadata={"filename": quote("my report.pdf")}, + ) + + stored = storage.head(str(name)) + + assert stored.size_kb == 2.0 + assert stored.filename == "my report.pdf" + + def test_filename_is_none_without_metadata(self) -> None: + aws = AmazonCloudStorageClient() + aws.create() + storage = AmazonCloudStorage(project_id=1, storage_path=uuid4()) + name = storage.url_for(Path(str(uuid4()))) + aws.client.put_object(Bucket=name.Bucket, Key=name.Key, Body=b"x" * 1024) + + stored = storage.head(str(name)) + + assert stored.size_kb == 1.0 + assert stored.filename is None + + def test_missing_key_raises_object_not_found(self) -> None: + AmazonCloudStorageClient().create() + storage = AmazonCloudStorage(project_id=1, storage_path=uuid4()) + url = str(storage.url_for(Path(str(uuid4())))) + + with pytest.raises(ObjectNotFoundError): + storage.head(url) + + +@mock_aws +@pytest.mark.usefixtures("aws_credentials") +class TestCopy: + def test_copies_the_source_bytes_to_the_destination_key(self) -> None: + aws = AmazonCloudStorageClient() + aws.create() + storage = AmazonCloudStorage(project_id=1, storage_path=uuid4()) + source = storage.url_for(Path("waiting.pdf"), is_pending=True) + aws.client.put_object( + Bucket=source.Bucket, Key=source.Key, Body=b"pending bytes" + ) + destination = Path("final.pdf") + + target = storage.copy(str(source), destination) + + assert target == SimpleStorageName( + Key=f"{storage.storage_path}/final.pdf", Bucket=settings.AWS_S3_BUCKET + ) + copied = aws.client.get_object(Bucket=target.Bucket, Key=target.Key) + assert copied["Body"].read() == b"pending bytes" + + def test_missing_source_raises_object_not_found(self) -> None: + AmazonCloudStorageClient().create() + storage = AmazonCloudStorage(project_id=1, storage_path=uuid4()) + source = str(storage.url_for(Path("absent.pdf"), is_pending=True)) + + with pytest.raises(ObjectNotFoundError): + storage.copy(source, Path("final.pdf")) + + def _amazon_storage_with_mock_client(mock_client): storage = AmazonCloudStorage(project_id=1, storage_path=uuid4()) storage.aws = SimpleNamespace(client=mock_client) diff --git a/docs/wiki/modules/knowledge-base.md b/docs/wiki/modules/knowledge-base.md index 179b0c69d..b585165a4 100644 --- a/docs/wiki/modules/knowledge-base.md +++ b/docs/wiki/modules/knowledge-base.md @@ -6,7 +6,8 @@ Deep dive: `docs/architecture/kaapi-knowledge-base-ARCHITECTURE.md` (§3 upload, All paths relative to `backend/app/`. ## Routes -- `api/routes/documents.py` — upload/list +- `api/routes/documents.py` — upload/list (v1 multipart upload, the only path that transforms) +- `api/routes/documents_v2.py` — v2 pre-signed upload: `POST /documents/uploads` → 200 (issues a pre-signed POST to a pending key; nothing persisted) then `PUT /documents/{document_id}` → 201 (registers the pending object; no body); no transformation - `api/routes/collections.py`, `api/routes/collection_job.py` — collection CRUD + job status - `api/routes/doc_transformation_job.py` — transform job status @@ -22,7 +23,7 @@ All paths relative to `backend/app/`. ## Services / CRUD - `services/collections/` — `create_collection.py`, `delete_collection.py`, `providers/`, `helpers.py` -- `services/documents/` — upload path +- `services/documents/` — `helpers.py` (v1 upload path), `registration.py` (v2 upload policy), `validator.py` - `services/doctransform/` — `job.py`, `registry.py`, `transformer.py`, `zerox_transformer.py` - `crud/collection/`, `crud/document/`, `crud/document_collection.py`, `crud/rag/`, `crud/file.py` @@ -35,6 +36,15 @@ All paths relative to `backend/app/`. ## Gotchas - `signed_url` behaviour is chosen per request by the `download` query param (default false = inline, as before). `put()` stores the upload's `Content-Type`, so a bare presigned URL for a PDF opens in a tab (CSV/XLSX datasets happen to download, which is why only KB docs looked broken). `get_signed_url(..., filename=...)` adds `ResponseContentDisposition: attachment`; `_signed_url()` in `services/documents/helpers.py` passes `fname` only when `download` is set, and the flag threads through `build_document_schema(s)` / `build_job_schema(s)` from `api/routes/documents.py`, `doc_transformation_job.py` and `collections.py`. An attachment URL will not render in an `