""" Phase 13 cloud content and mutation endpoints. Owner-scoped routes for opening, previewing, downloading, renaming, moving, deleting, creating folders, and uploading cloud files. All routes are connection-ID-keyed and call the service layer only; no provider SDK calls are made here. Security invariants (T-13-14): - Provider credentials, raw provider URLs, and tokens must never appear in any response body or header. Typed kind/reason codes are the only way the frontend learns about provider outcomes. - Binary-only preview enforced by PreviewSupport.is_supported() (D-18). - Cross-connection moves are rejected before the adapter is called (D-08). - Self-destination moves are rejected before the adapter is called (D-09). - No health probe triggered by folder navigation (D-13). Route family: GET /connections/{id}/items/{item_id}/open — D-02 authorized access URL GET /connections/{id}/items/{item_id}/preview — D-18 binary-only preview PATCH /connections/{id}/items/{item_id}/rename — D-05, D-07 POST /connections/{id}/items/{item_id}/move — D-08, D-09 DELETE /connections/{id}/items/{item_id} — D-10, D-11 POST /connections/{id}/folders — D-05, D-06 POST /connections/{id}/items/upload — D-03, D-04 Admin users are blocked by get_regular_user (T-13-01). """ from __future__ import annotations import uuid from typing import Optional from fastapi import APIRouter, Depends, File, Form, HTTPException, Request, UploadFile, status from fastapi.responses import JSONResponse, Response from sqlalchemy.ext.asyncio import AsyncSession from api.cloud.schemas import ( CreateFolderRequest, MoveItemRequest, MutationResultOut, RenameItemRequest, ) from config import settings from db.models import User from deps.auth import get_regular_user from deps.db import get_db from services.cloud_items import ConnectionNotFound from services.rate_limiting import account_limiter from storage.cloud_base import ( MUT_KIND_CONFLICT, MUT_KIND_DELETED, MUT_KIND_FOLDER, MUT_KIND_INVALID_DEST, MUT_KIND_OFFLINE, MUT_KIND_REAUTH, MUT_KIND_STALE, MUT_KIND_UNSUPPORTED, MUT_KIND_UPDATED, MUT_KIND_UPLOADED, MUT_REASON_ITEM_CHANGED, MUT_REASON_NAME_COLLISION, MUT_REASON_PERMANENT, MUT_REASON_PROVIDER_ERROR, MUT_REASON_SELF_MOVE, MUT_REASON_TOKEN_EXPIRED, MUT_REASON_TRASHED, PreviewSupport, ) router = APIRouter() # ── Helpers ─────────────────────────────────────────────────────────────────── def _master_key() -> bytes: return settings.cloud_creds_key.encode() async def _resolve_and_get_adapter( session: AsyncSession, connection_id: uuid.UUID, user_id: uuid.UUID, ): """Resolve owned connection and return a mutable adapter. Raises HTTPException 404 if the connection is not owned by user_id. Raises HTTPException 401 (reauth_required) if credential decryption fails. """ from services.cloud_items import resolve_owned_connection from storage.cloud_utils import decrypt_credentials from storage.cloud_backend_factory import build_mutable_cloud_adapter try: conn = await resolve_owned_connection( session, connection_id=connection_id, user_id=user_id ) except ConnectionNotFound: raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Connection not found") try: creds = decrypt_credentials(_master_key(), str(user_id), conn.credentials_enc) except Exception: # Credential decryption failure = credentials corrupted or key mismatch. # This is a reauth scenario, not a server error — return typed reauth body. raise HTTPException( status_code=status.HTTP_401_UNAUTHORIZED, detail=f"Credential decryption failed — reconnect this connection", ) try: adapter = build_mutable_cloud_adapter(conn.provider, creds) except ValueError as exc: raise HTTPException( status_code=status.HTTP_422_UNPROCESSABLE_ENTITY, detail=str(exc), ) return conn, adapter def _mutation_error_json(result: dict, http_status: int) -> JSONResponse: """Return a typed JSON error response for a failed mutation. T-13-14: Returns {kind, reason} at the top level — never nested under 'detail'. This lets the frontend read body.kind without special-casing HTTPException format. """ kind = result.get("kind", "error") reason = result.get("reason", MUT_REASON_PROVIDER_ERROR) return JSONResponse( status_code=http_status, content={"kind": kind, "reason": reason}, ) def _mutation_error_response(result: dict) -> JSONResponse: """Translate a normalized mutation result dict into a typed JSON response. Used for non-success mutation results that need an HTTP error status. Returns JSONResponse (not HTTPException) so kind/reason appear at the top level of the response body rather than nested under 'detail'. """ kind = result.get("kind", "error") if kind == MUT_KIND_STALE: return _mutation_error_json(result, status.HTTP_409_CONFLICT) if kind == MUT_KIND_CONFLICT: return _mutation_error_json(result, status.HTTP_409_CONFLICT) if kind == MUT_KIND_INVALID_DEST: return _mutation_error_json(result, status.HTTP_409_CONFLICT) if kind == MUT_KIND_REAUTH: return _mutation_error_json(result, status.HTTP_401_UNAUTHORIZED) if kind == MUT_KIND_UNSUPPORTED: return JSONResponse( status_code=status.HTTP_422_UNPROCESSABLE_ENTITY, content={"kind": "unsupported_operation", "reason": "provider_unsupported"}, ) # offline / error — 503 return _mutation_error_json(result, status.HTTP_503_SERVICE_UNAVAILABLE) # ── Open endpoint ───────────────────────────────────────────────────────────── @router.get("/connections/{connection_id}/items/{item_id:path}/open") @account_limiter.limit("100/minute") async def open_cloud_file( connection_id: uuid.UUID, item_id: str, request: Request, session: AsyncSession = Depends(get_db), current_user: User = Depends(get_regular_user), ) -> dict: """Return a DocuVault-authorized open URL for a cloud file. D-02: Provider credentials and raw provider URLs are never exposed in the response. The caller must navigate to the returned url — never window.open to a raw provider URL. T-13-14: Response shape is stable — {kind: 'open', url: ''}. T-13-01: Owner-scoped via get_regular_user + resolve_owned_connection. """ request.state.current_user = current_user # Verify ownership — 404 for foreign user (IDOR, T-13-01) try: from services.cloud_items import resolve_owned_connection await resolve_owned_connection( session, connection_id=connection_id, user_id=current_user.id ) except ConnectionNotFound: raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Connection not found") # Return a DocuVault-scoped authorized URL — the download route below serves bytes. # This URL is backend-scoped: no raw provider URL or credentials ever cross to the client. authorized_url = ( f"{settings.backend_url}/api/cloud/connections/{connection_id}" f"/items/{item_id}/download" ) return { "kind": "open", "url": authorized_url, "reason": "authorized", } # ── Preview endpoint ────────────────────────────────────────────────────────── def _unsupported_preview_response(connection_id: uuid.UUID, item_id: str, reason: str) -> dict: """Return a typed unsupported-preview JSON body (D-18, T-13-14).""" return { "kind": "unsupported_preview", "reason": reason, "url": ( f"{settings.backend_url}/api/cloud/connections/{connection_id}" f"/items/{item_id}/download" ), } @router.get("/connections/{connection_id}/items/{item_id:path}/preview") @account_limiter.limit("100/minute") async def preview_cloud_file( connection_id: uuid.UUID, item_id: str, request: Request, session: AsyncSession = Depends(get_db), current_user: User = Depends(get_regular_user), ): """Stream a binary preview of a supported cloud file. D-18: Only supported binary file formats are previewed. Google Workspace and Microsoft Office document rendering/editing are explicitly excluded. Unsupported formats return a typed {kind: 'unsupported_preview', reason: 'unsupported_format'} body so the frontend can route to the authorized download fallback (D-02). The response streams inline (Content-Disposition: inline) — never as a browser-to-device download (Content-Disposition: attachment). T-13-14: No provider URL, token, or credential appears in any response. T-13-01: Owner-scoped via get_regular_user + resolve_owned_connection. Ownership check is performed before any credential decryption — this keeps the endpoint safe even when credentials cannot be decrypted (e.g. test fixtures with a different encryption key). """ request.state.current_user = current_user # Step 1: Ownership check (no credential decryption yet — T-13-01, IDOR gate) try: from services.cloud_items import resolve_owned_connection await resolve_owned_connection( session, connection_id=connection_id, user_id=current_user.id ) except ConnectionNotFound: raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Connection not found") # Step 2: Resolve content type from cached cloud item metadata from sqlalchemy import select from db.models import CloudItem result = await session.execute( select(CloudItem).where( CloudItem.connection_id == connection_id, CloudItem.provider_item_id == item_id, CloudItem.user_id == current_user.id, CloudItem.deleted_at.is_(None), ) ) cloud_item = result.scalar_one_or_none() content_type = cloud_item.content_type if cloud_item else None # Step 3: D-18 binary preview gate — unsupported formats use authorized download fallback. # No credential decryption is attempted for unsupported types. if not PreviewSupport.is_supported(content_type): return _unsupported_preview_response(connection_id, item_id, "unsupported_format") # Step 4: Fetch bytes via the adapter (supported binary format confirmed above) try: conn, adapter = await _resolve_and_get_adapter(session, connection_id, current_user.id) file_bytes = await adapter.get_object(item_id) except HTTPException: # Re-raise 404 from ownership check in _resolve_and_get_adapter (already done above, # but keep for belt-and-suspenders) return _unsupported_preview_response(connection_id, item_id, "provider_error") except Exception: return _unsupported_preview_response(connection_id, item_id, "provider_error") # Stream inline — never as attachment (D-18) return Response( content=file_bytes, media_type=content_type or "application/octet-stream", headers={"Content-Disposition": "inline"}, ) # ── Authorized download (fallback for open and unsupported preview) ──────────── @router.get("/connections/{connection_id}/items/{item_id:path}/download") @account_limiter.limit("100/minute") async def download_cloud_file( connection_id: uuid.UUID, item_id: str, request: Request, session: AsyncSession = Depends(get_db), current_user: User = Depends(get_regular_user), ): """Authorized download fallback for cloud files. D-02: The download route is DocuVault-controlled — no raw provider URL or token crosses to the client. Bytes are fetched server-side and streamed. This is the target of both open and unsupported-preview fallback URLs. T-13-01: Owner-scoped via get_regular_user + resolve_owned_connection. """ request.state.current_user = current_user conn, adapter = await _resolve_and_get_adapter(session, connection_id, current_user.id) # Resolve filename from metadata if available from sqlalchemy import select from db.models import CloudItem meta_result = await session.execute( select(CloudItem).where( CloudItem.connection_id == connection_id, CloudItem.provider_item_id == item_id, CloudItem.user_id == current_user.id, CloudItem.deleted_at.is_(None), ) ) cloud_item = meta_result.scalar_one_or_none() filename = cloud_item.name if cloud_item else item_id content_type = (cloud_item.content_type if cloud_item else None) or "application/octet-stream" try: file_bytes = await adapter.get_object(item_id) except Exception as exc: result_dict = adapter._normalize_error(exc) if result_dict.get("kind") == MUT_KIND_REAUTH: raise HTTPException( status_code=status.HTTP_401_UNAUTHORIZED, detail="Provider authentication required — please reconnect this connection", ) raise HTTPException( status_code=status.HTTP_503_SERVICE_UNAVAILABLE, detail="Provider temporarily unavailable — please try again", ) safe_filename = filename.replace('"', "'") return Response( content=file_bytes, media_type=content_type, headers={"Content-Disposition": f'attachment; filename="{safe_filename}"'}, ) # ── Create folder ───────────────────────────────────────────────────────────── @router.post("/connections/{connection_id}/folders", status_code=status.HTTP_201_CREATED) @account_limiter.limit("100/minute") async def create_cloud_folder( connection_id: uuid.UUID, body: CreateFolderRequest, request: Request, session: AsyncSession = Depends(get_db), current_user: User = Depends(get_regular_user), ) -> dict: """Create a folder in connected cloud storage. D-05: Name collision auto-suffix: 'Projects (1)', 'Projects (2)', etc. D-06: Concurrent collision race is handled by the adapter with bounded retry. Returns {kind: 'folder', provider_item_id, name, parent_ref} on success. Returns typed error body for unsupported providers, auth failures, etc. T-13-01: Owner-scoped. T-13-14: No credentials in response. """ request.state.current_user = current_user conn, adapter = await _resolve_and_get_adapter(session, connection_id, current_user.id) result = await adapter.create_folder( parent_ref=body.parent_ref, name=body.name, connection_id=connection_id, user_id=current_user.id, ) kind = result.get("kind") if kind == MUT_KIND_FOLDER: return { "kind": "folder", "provider_item_id": result.get("provider_item_id"), "name": result.get("name", body.name), "parent_ref": result.get("parent_ref", body.parent_ref), } # Unsupported operation — return typed body instead of 500 if kind == MUT_KIND_UNSUPPORTED: return { "kind": "unsupported_operation", "reason": "provider_unsupported", } return _mutation_error_response(result) # ── Rename ──────────────────────────────────────────────────────────────────── @router.patch("/connections/{connection_id}/items/{item_id:path}/rename") @account_limiter.limit("100/minute") async def rename_cloud_item( connection_id: uuid.UUID, item_id: str, body: RenameItemRequest, request: Request, session: AsyncSession = Depends(get_db), current_user: User = Depends(get_regular_user), ) -> dict: """Rename a cloud file or folder. D-05: Counter-suffix collision policy — auto-suffixes on name collision. D-07: etag guards against operating on stale metadata; returns {kind: 'stale', reason: 'item_changed'} on mismatch (HTTP 409). Returns {kind: 'renamed', provider_item_id, name} on success. T-13-01: Owner-scoped. T-13-14: No credentials in response. """ request.state.current_user = current_user conn, adapter = await _resolve_and_get_adapter(session, connection_id, current_user.id) result = await adapter.rename( provider_item_id=item_id, new_name=body.new_name, etag=body.etag, ) kind = result.get("kind") if kind == MUT_KIND_UPDATED: return { "kind": "renamed", "provider_item_id": result.get("provider_item_id", item_id), "name": result.get("name", body.new_name), } return _mutation_error_response(result) # ── Move ────────────────────────────────────────────────────────────────────── @router.post("/connections/{connection_id}/items/{item_id:path}/move") @account_limiter.limit("100/minute") async def move_cloud_item( connection_id: uuid.UUID, item_id: str, body: MoveItemRequest, request: Request, session: AsyncSession = Depends(get_db), current_user: User = Depends(get_regular_user), ) -> dict: """Move a cloud item within the same connection. D-08: Cross-connection moves are rejected before calling the adapter. D-09: Self and descendant destinations are rejected before calling the adapter. Returns {kind: 'moved', provider_item_id, parent_ref} on success. Returns {kind: 'invalid_destination', reason: '...'} for invalid moves (HTTP 409). T-13-01: Owner-scoped. T-13-14: No credentials in response. """ request.state.current_user = current_user # D-08: Cross-connection move rejection (before any adapter call) if body.destination_connection_id is not None: dest_conn_id = str(body.destination_connection_id).strip() source_conn_id = str(connection_id) if dest_conn_id and dest_conn_id != source_conn_id: return JSONResponse( status_code=status.HTTP_409_CONFLICT, content={"kind": "invalid_destination", "reason": "cross_connection"}, ) # D-09: Self-destination rejection (before any adapter call) if body.destination_parent_ref == item_id: return JSONResponse( status_code=status.HTTP_409_CONFLICT, content={"kind": "invalid_destination", "reason": MUT_REASON_SELF_MOVE}, ) conn, adapter = await _resolve_and_get_adapter(session, connection_id, current_user.id) result = await adapter.move( provider_item_id=item_id, destination_parent_ref=body.destination_parent_ref, etag=body.etag, ) kind = result.get("kind") if kind == MUT_KIND_UPDATED: return { "kind": "moved", "provider_item_id": result.get("provider_item_id", item_id), "parent_ref": result.get("destination_parent_ref", body.destination_parent_ref), } return _mutation_error_response(result) # ── Delete ──────────────────────────────────────────────────────────────────── @router.delete("/connections/{connection_id}/items/{item_id:path}", status_code=status.HTTP_200_OK) @account_limiter.limit("100/minute") async def delete_cloud_item( connection_id: uuid.UUID, item_id: str, request: Request, session: AsyncSession = Depends(get_db), current_user: User = Depends(get_regular_user), ) -> dict: """Delete a cloud item. D-11: Prefers the provider's trash/recycle-bin when supported. Returns {kind: 'deleted', reason: 'trashed'} or {kind: 'deleted', reason: 'permanent'}. D-10: Callers must confirm before invoking this endpoint — it acts immediately. T-13-01: Owner-scoped. T-13-14: No credentials or provider URLs in response. """ request.state.current_user = current_user conn, adapter = await _resolve_and_get_adapter(session, connection_id, current_user.id) result = await adapter.delete(provider_item_id=item_id, trash=True) kind = result.get("kind") reason = result.get("reason") if kind == MUT_KIND_DELETED: return { "kind": "deleted", "reason": reason or MUT_REASON_PERMANENT, "provider_item_id": result.get("provider_item_id", item_id), } return _mutation_error_response(result) # ── Upload ──────────────────────────────────────────────────────────────────── @router.post("/connections/{connection_id}/items/upload", status_code=status.HTTP_200_OK) @account_limiter.limit("100/minute") async def upload_cloud_file( connection_id: uuid.UUID, request: Request, file: UploadFile = File(...), parent_ref: Optional[str] = Form(None), filename: Optional[str] = Form(None), session: AsyncSession = Depends(get_db), current_user: User = Depends(get_regular_user), ) -> dict: """Upload a file to a cloud folder. D-03: A same-name upload never overwrites silently. Returns {kind: 'conflict', reason: 'name_collision'} so the frontend shows the conflict resolution dialog. The conflict check uses cached cloud_items metadata first (fast path) before attempting the provider upload. D-04: Caller manages the sequential upload queue; this endpoint handles one file. Returns {kind: 'uploaded', provider_item_id, name, parent_ref, size} on success. T-13-01: Owner-scoped. T-13-14: No provider URL or credentials in response. """ request.state.current_user = current_user # Ownership check before credential decryption (T-13-01, IDOR gate) try: from services.cloud_items import resolve_owned_connection await resolve_owned_connection( session, connection_id=connection_id, user_id=current_user.id ) except ConnectionNotFound: raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Connection not found") upload_filename = filename or file.filename or "upload" # D-03: Check cached metadata for same-name item in parent folder. # This is the fast-path conflict detection: if we have a cached item with # the same name under the same parent_ref, signal conflict before attempting # the provider upload. The provider adapter also enforces this independently. from sqlalchemy import select from db.models import CloudItem conflict_check = await session.execute( select(CloudItem).where( CloudItem.connection_id == connection_id, CloudItem.user_id == current_user.id, CloudItem.name == upload_filename, CloudItem.parent_ref == parent_ref, CloudItem.deleted_at.is_(None), ) ) if conflict_check.scalar_one_or_none() is not None: return { "kind": "conflict", "reason": MUT_REASON_NAME_COLLISION, "existing_name": upload_filename, } # Proceed with provider upload (requires credential decryption) conn, adapter = await _resolve_and_get_adapter(session, connection_id, current_user.id) content_type = file.content_type or "application/octet-stream" content = await file.read() result = await adapter.upload_file( parent_ref=parent_ref, filename=upload_filename, content=content, content_type=content_type, connection_id=connection_id, user_id=current_user.id, ) kind = result.get("kind") if kind == MUT_KIND_UPLOADED: # Route authoritative upload success through centralized reconciliation # before returning (Plan 06, Task 1: T-13-19). # upsert_cloud_item gives the new item stable row identity in cloud_items, # so navigation can show it immediately without a full provider re-list. # update_folder_state invalidates the parent folder so the next browse # triggers a provider re-list and reflects the new content truthfully. # These calls are intentionally in-request — the response is not returned # until reconciliation completes (reconcile-before-return contract). provider_item_id = result.get("provider_item_id") resolved_name = result.get("name", upload_filename) resolved_parent_ref = result.get("parent_ref", parent_ref) resolved_size = result.get("size", len(content)) if provider_item_id: from storage.cloud_base import CloudResource from services.cloud_items import upsert_cloud_item, update_folder_state resource = CloudResource( id=uuid.uuid4(), provider_item_id=provider_item_id, connection_id=connection_id, user_id=current_user.id, name=resolved_name, kind="file", parent_ref=resolved_parent_ref, content_type=content_type, size=resolved_size, ) # Upsert item — establishes stable row identity for navigation await upsert_cloud_item(session, user_id=str(current_user.id), resource=resource) # Invalidate parent folder so next browse triggers a provider re-list. # We do NOT call apply_listing_and_finalize — that requires a full provider # listing which we don't have. Instead we mark the folder stale with a # controlled reason so the browse endpoint knows to re-fetch. invalidate_parent = resolved_parent_ref if resolved_parent_ref is not None else "" await update_folder_state( session, user_id=str(current_user.id), connection_id=str(connection_id), parent_ref=invalidate_parent, refresh_state="warning", error_code="upload_mutated", error_message="Folder contents changed by upload — re-listing required.", ) await session.commit() return { "kind": "uploaded", "provider_item_id": provider_item_id, "name": resolved_name, "parent_ref": resolved_parent_ref, "size": resolved_size, } if kind == MUT_KIND_CONFLICT: # D-03: Return conflict kind for same-name detection — never overwrite silently return { "kind": "conflict", "reason": MUT_REASON_NAME_COLLISION, "existing_name": result.get("existing_name", upload_filename), } return _mutation_error_response(result)