1105 lines
44 KiB
Python
1105 lines
44 KiB
Python
"""
|
|
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 urllib.parse
|
|
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 deps.utils import get_client_ip
|
|
from services.audit import write_audit_log
|
|
from services.cloud_items import (
|
|
ConnectionNotFound,
|
|
update_folder_state,
|
|
upsert_cloud_item,
|
|
)
|
|
from services.cloud_operations import keep_both_name
|
|
from services.rate_limiting import account_limiter
|
|
from storage.cloud_base import (
|
|
CloudResource,
|
|
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,
|
|
)
|
|
|
|
# Maximum collision retry attempts for create-folder (D-06)
|
|
_CREATE_FOLDER_MAX_RETRIES = 5
|
|
|
|
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: '<docuvault-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 401 (credential failure) and 404 (ownership race) — never mask as
|
|
# unsupported_preview. The IDOR guard in Step 1 already passed, but a 401
|
|
# from credential decryption should surface as an auth error to the client.
|
|
raise
|
|
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",
|
|
)
|
|
|
|
# RFC 6266 percent-encoding — prevents header injection via special characters
|
|
# (newlines, semicolons, etc.) that str.replace('"', "'") does not cover.
|
|
encoded_filename = urllib.parse.quote(filename, safe=" !#$&'()*+,-./:;<=>?@[]^_`{|}~")
|
|
return Response(
|
|
content=file_bytes,
|
|
media_type=content_type,
|
|
headers={"Content-Disposition": f"attachment; filename*=UTF-8''{encoded_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: Bounded retry up to _CREATE_FOLDER_MAX_RETRIES when a concurrent client
|
|
takes the candidate name between check and mutation.
|
|
D-07: Stale precondition returned by the provider stops the mutation, marks the
|
|
parent folder as non-fresh, and returns typed {kind: 'stale'} for retry.
|
|
|
|
Returns {kind: 'folder', provider_item_id, name, parent_ref} on success.
|
|
After success, the new folder is upserted into cloud_items and the parent folder
|
|
state is invalidated before the response is returned (reconcile-before-return).
|
|
|
|
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)
|
|
|
|
candidate_name = body.name
|
|
last_result: dict = {}
|
|
|
|
for attempt in range(_CREATE_FOLDER_MAX_RETRIES + 1):
|
|
result = await adapter.create_folder(
|
|
parent_ref=body.parent_ref,
|
|
name=candidate_name,
|
|
connection_id=connection_id,
|
|
user_id=current_user.id,
|
|
)
|
|
last_result = result
|
|
kind = result.get("kind")
|
|
|
|
if kind == MUT_KIND_FOLDER:
|
|
# Success: reconcile before returning (T-13-24, T-13-26)
|
|
provider_item_id = result.get("provider_item_id")
|
|
resolved_name = result.get("name", candidate_name)
|
|
resolved_parent_ref = result.get("parent_ref", body.parent_ref)
|
|
|
|
if provider_item_id:
|
|
resource = CloudResource(
|
|
id=uuid.uuid4(),
|
|
provider_item_id=provider_item_id,
|
|
connection_id=connection_id,
|
|
user_id=current_user.id,
|
|
name=resolved_name,
|
|
kind="folder",
|
|
parent_ref=resolved_parent_ref,
|
|
)
|
|
# Upsert gives the new folder stable row identity in cloud_items
|
|
await upsert_cloud_item(session, user_id=str(current_user.id), resource=resource)
|
|
|
|
# Invalidate parent folder so next browse triggers a provider re-list
|
|
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="create_folder_mutated",
|
|
error_message="Folder contents changed by create-folder — re-listing required.",
|
|
)
|
|
|
|
await session.commit()
|
|
|
|
return {
|
|
"kind": "folder",
|
|
"provider_item_id": provider_item_id,
|
|
"name": resolved_name,
|
|
"parent_ref": resolved_parent_ref,
|
|
}
|
|
|
|
if kind == MUT_KIND_CONFLICT and attempt < _CREATE_FOLDER_MAX_RETRIES:
|
|
# D-05/D-06: auto-suffix with counter and retry
|
|
candidate_name = keep_both_name(body.name, attempt + 1)
|
|
continue
|
|
|
|
if kind == MUT_KIND_STALE:
|
|
# D-07: Stop mutation, mark parent folder as non-fresh, return typed stale result
|
|
stale_parent = body.parent_ref if body.parent_ref is not None else ""
|
|
await update_folder_state(
|
|
session,
|
|
user_id=str(current_user.id),
|
|
connection_id=str(connection_id),
|
|
parent_ref=stale_parent,
|
|
refresh_state="warning",
|
|
error_code="stale_listing",
|
|
error_message="Parent folder changed externally — re-listing required before retry.",
|
|
)
|
|
await session.commit()
|
|
return _mutation_error_response(result)
|
|
|
|
# All other non-success kinds (offline, reauth, unsupported, conflict exhausted)
|
|
break
|
|
|
|
# Unsupported operation — return typed body instead of 500
|
|
if last_result.get("kind") == MUT_KIND_UNSUPPORTED:
|
|
return {
|
|
"kind": "unsupported_operation",
|
|
"reason": "provider_unsupported",
|
|
}
|
|
|
|
return _mutation_error_response(last_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: Rename collision is surfaced to the user (they chose the name explicitly).
|
|
D-07: etag guards against operating on stale metadata. On mismatch:
|
|
stops mutation, marks parent folder as non-fresh, returns
|
|
{kind: 'stale', reason: 'item_changed'} (HTTP 409) so the frontend
|
|
re-lists before retrying.
|
|
|
|
On success: upserts the updated item name in cloud_items and invalidates the
|
|
parent folder state before returning (reconcile-before-return, T-13-26).
|
|
|
|
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:
|
|
# Success: reconcile the updated name in cloud_items before returning (T-13-26)
|
|
resolved_provider_item_id = result.get("provider_item_id", item_id)
|
|
resolved_name = result.get("name", body.new_name)
|
|
|
|
# Resolve the parent_ref from the existing CloudItem so we can invalidate it
|
|
from sqlalchemy import select
|
|
from db.models import CloudItem
|
|
|
|
existing_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),
|
|
)
|
|
)
|
|
existing_item = existing_result.scalar_one_or_none()
|
|
item_parent_ref = existing_item.parent_ref if existing_item else None
|
|
item_kind = existing_item.kind if existing_item else "file"
|
|
item_content_type = existing_item.content_type if existing_item else None
|
|
item_size = existing_item.provider_size if existing_item else None
|
|
|
|
# Upsert the item with the new name — uses provider_item_id as stable identity key
|
|
resource = CloudResource(
|
|
id=uuid.uuid4(),
|
|
provider_item_id=resolved_provider_item_id,
|
|
connection_id=connection_id,
|
|
user_id=current_user.id,
|
|
name=resolved_name,
|
|
kind=item_kind,
|
|
parent_ref=item_parent_ref,
|
|
content_type=item_content_type,
|
|
size=item_size,
|
|
)
|
|
await upsert_cloud_item(session, user_id=str(current_user.id), resource=resource)
|
|
|
|
# Invalidate the parent folder so the next browse reflects the renamed item
|
|
invalidate_parent = item_parent_ref if item_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="rename_mutated",
|
|
error_message="Folder contents changed by rename — re-listing required.",
|
|
)
|
|
|
|
await session.commit()
|
|
|
|
return {
|
|
"kind": "renamed",
|
|
"provider_item_id": resolved_provider_item_id,
|
|
"name": resolved_name,
|
|
}
|
|
|
|
if kind == MUT_KIND_STALE:
|
|
# D-07: Stop mutation, mark parent folder as non-fresh, return typed stale result.
|
|
# Resolve parent_ref from existing CloudItem metadata if available.
|
|
from sqlalchemy import select
|
|
from db.models import CloudItem
|
|
|
|
stale_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),
|
|
)
|
|
)
|
|
stale_item = stale_result.scalar_one_or_none()
|
|
stale_parent = stale_item.parent_ref if stale_item else None
|
|
stale_parent_ref = stale_parent if stale_parent is not None else ""
|
|
|
|
await update_folder_state(
|
|
session,
|
|
user_id=str(current_user.id),
|
|
connection_id=str(connection_id),
|
|
parent_ref=stale_parent_ref,
|
|
refresh_state="warning",
|
|
error_code="stale_listing",
|
|
error_message="Item changed externally — re-listing required before retry.",
|
|
)
|
|
await session.commit()
|
|
return _mutation_error_response(result)
|
|
|
|
return _mutation_error_response(result)
|
|
|
|
|
|
# ── Move ──────────────────────────────────────────────────────────────────────
|
|
|
|
|
|
async def _is_descendant_destination(
|
|
session: AsyncSession,
|
|
connection_id: uuid.UUID,
|
|
user_id: uuid.UUID,
|
|
source_item_id: str,
|
|
dest_parent_ref: str,
|
|
) -> bool:
|
|
"""Return True if dest_parent_ref is a descendant of source_item_id.
|
|
|
|
D-09: The backend must independently reject descendant destinations even if
|
|
the frontend pre-screened them. Walks the ancestor chain of dest_parent_ref
|
|
through the cloud_items table to detect cycles.
|
|
|
|
This is a metadata-only check — no provider calls are made.
|
|
"""
|
|
from sqlalchemy import select
|
|
from db.models import CloudItem
|
|
|
|
visited: set[str] = set()
|
|
current_ref: str | None = dest_parent_ref
|
|
|
|
while current_ref is not None:
|
|
if current_ref in visited:
|
|
# Cycle in ancestor chain — defensive guard
|
|
break
|
|
visited.add(current_ref)
|
|
|
|
if current_ref == source_item_id:
|
|
return True
|
|
|
|
# Look up the parent of current_ref
|
|
result = await session.execute(
|
|
select(CloudItem).where(
|
|
CloudItem.connection_id == connection_id,
|
|
CloudItem.provider_item_id == current_ref,
|
|
CloudItem.user_id == user_id,
|
|
CloudItem.deleted_at.is_(None),
|
|
)
|
|
)
|
|
parent_item = result.scalar_one_or_none()
|
|
current_ref = parent_item.parent_ref if parent_item else None
|
|
|
|
return False
|
|
|
|
|
|
@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.
|
|
D-07: Stale precondition failures stop the move, refresh the source folder, and
|
|
return typed {kind: 'stale', reason: 'item_changed'} for retry.
|
|
|
|
On success: upserts the item's new parent_ref in cloud_items, invalidates both
|
|
source and destination folder states, and writes a metadata-only audit row before
|
|
the response is returned (reconcile-before-return, T-13-26, T-13-29).
|
|
|
|
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},
|
|
)
|
|
|
|
# D-09: Descendant destination rejection — walk ancestor chain via cloud_items metadata.
|
|
# This is a belt-and-suspenders check independent of provider-side validation.
|
|
if body.destination_parent_ref:
|
|
is_descendant = await _is_descendant_destination(
|
|
session,
|
|
connection_id=connection_id,
|
|
user_id=current_user.id,
|
|
source_item_id=item_id,
|
|
dest_parent_ref=body.destination_parent_ref,
|
|
)
|
|
if is_descendant:
|
|
return JSONResponse(
|
|
status_code=status.HTTP_409_CONFLICT,
|
|
content={"kind": "invalid_destination", "reason": "descendant_destination"},
|
|
)
|
|
|
|
# Resolve existing item metadata before the provider call (for reconciliation on success)
|
|
from sqlalchemy import select
|
|
from db.models import CloudItem
|
|
|
|
existing_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),
|
|
)
|
|
)
|
|
existing_item = existing_result.scalar_one_or_none()
|
|
source_parent_ref = existing_item.parent_ref if existing_item else None
|
|
item_kind = existing_item.kind if existing_item else "file"
|
|
item_name = existing_item.name if existing_item else item_id
|
|
item_content_type = existing_item.content_type if existing_item else None
|
|
item_size = existing_item.provider_size if existing_item else None
|
|
|
|
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:
|
|
# Success: reconcile before returning (T-13-26, T-13-29)
|
|
resolved_provider_item_id = result.get("provider_item_id", item_id)
|
|
resolved_dest_parent_ref = result.get("destination_parent_ref", body.destination_parent_ref)
|
|
|
|
# Upsert item with new parent_ref — preserves stable row identity
|
|
resource = CloudResource(
|
|
id=uuid.uuid4(),
|
|
provider_item_id=resolved_provider_item_id,
|
|
connection_id=connection_id,
|
|
user_id=current_user.id,
|
|
name=item_name,
|
|
kind=item_kind,
|
|
parent_ref=resolved_dest_parent_ref,
|
|
content_type=item_content_type,
|
|
size=item_size,
|
|
)
|
|
await upsert_cloud_item(session, user_id=str(current_user.id), resource=resource)
|
|
|
|
# Invalidate source folder (item was removed from here)
|
|
invalidate_source = source_parent_ref if source_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_source,
|
|
refresh_state="warning",
|
|
error_code="move_mutated",
|
|
error_message="Folder contents changed by move — re-listing required.",
|
|
)
|
|
|
|
# Invalidate destination folder (item was added here)
|
|
invalidate_dest = resolved_dest_parent_ref if resolved_dest_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_dest,
|
|
refresh_state="warning",
|
|
error_code="move_mutated",
|
|
error_message="Folder contents changed by move — re-listing required.",
|
|
)
|
|
|
|
# Write metadata-only audit row (T-13-02, T-13-29)
|
|
await write_audit_log(
|
|
session,
|
|
event_type="cloud.item_moved",
|
|
user_id=current_user.id,
|
|
actor_id=current_user.id,
|
|
resource_id=None,
|
|
ip_address=get_client_ip(request),
|
|
metadata_={
|
|
"connection_id": str(connection_id),
|
|
"provider_item_id": resolved_provider_item_id,
|
|
"old_parent_ref": source_parent_ref,
|
|
"new_parent_ref": resolved_dest_parent_ref,
|
|
},
|
|
)
|
|
|
|
await session.commit()
|
|
|
|
return {
|
|
"kind": "moved",
|
|
"provider_item_id": resolved_provider_item_id,
|
|
"parent_ref": resolved_dest_parent_ref,
|
|
}
|
|
|
|
if kind == MUT_KIND_STALE:
|
|
# D-07: Stop mutation, mark source folder as non-fresh, return typed stale result
|
|
stale_source = source_parent_ref if source_parent_ref is not None else ""
|
|
await update_folder_state(
|
|
session,
|
|
user_id=str(current_user.id),
|
|
connection_id=str(connection_id),
|
|
parent_ref=stale_source,
|
|
refresh_state="warning",
|
|
error_code="stale_listing",
|
|
error_message="Item changed externally — re-listing required before retry.",
|
|
)
|
|
await session.commit()
|
|
return _mutation_error_response(result)
|
|
|
|
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-10: Files get a standard confirmation; folders warn that nested contents are included.
|
|
D-11: Prefers the provider's trash/recycle-bin when supported. Returns
|
|
{kind: 'deleted', reason: 'trashed'} or {kind: 'deleted', reason: 'permanent'}.
|
|
On success: marks the parent folder non-fresh and writes a metadata-only audit row
|
|
before the response is returned (reconcile-before-return, T-13-29).
|
|
|
|
T-13-01: Owner-scoped. T-13-14: No credentials or provider URLs in response.
|
|
"""
|
|
request.state.current_user = current_user
|
|
|
|
# Pre-fetch item metadata for disclosure (D-10) and reconciliation
|
|
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),
|
|
)
|
|
)
|
|
existing_item = meta_result.scalar_one_or_none()
|
|
item_kind = existing_item.kind if existing_item else "file"
|
|
item_parent_ref = existing_item.parent_ref if existing_item else None
|
|
|
|
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:
|
|
# Success: reconcile before returning (T-13-29)
|
|
|
|
# Invalidate parent folder so next browse reflects the deletion
|
|
invalidate_parent = item_parent_ref if item_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="delete_mutated",
|
|
error_message="Folder contents changed by delete — re-listing required.",
|
|
)
|
|
|
|
# Write metadata-only audit row — only after authoritative success (T-13-02, T-13-29)
|
|
delete_reason = reason or MUT_REASON_PERMANENT
|
|
await write_audit_log(
|
|
session,
|
|
event_type="cloud.item_deleted",
|
|
user_id=current_user.id,
|
|
actor_id=current_user.id,
|
|
resource_id=None,
|
|
ip_address=get_client_ip(request),
|
|
metadata_={
|
|
"connection_id": str(connection_id),
|
|
"provider_item_id": result.get("provider_item_id", item_id),
|
|
"delete_kind": delete_reason,
|
|
"item_kind": item_kind,
|
|
},
|
|
)
|
|
|
|
await session.commit()
|
|
|
|
# D-10: Disclose folder-vs-file delete semantics to the frontend
|
|
return {
|
|
"kind": "deleted",
|
|
"reason": delete_reason,
|
|
"provider_item_id": result.get("provider_item_id", item_id),
|
|
"item_kind": item_kind,
|
|
"is_folder": item_kind == "folder",
|
|
}
|
|
|
|
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.",
|
|
)
|
|
|
|
# Write metadata-only audit row in the same transaction (Plan 06, Task 2: T-13-20).
|
|
# Only authoritative upload success reaches this path. Non-success outcomes
|
|
# (conflict, offline, reauth_required) are handled in separate branches below
|
|
# and never call write_audit_log — preserving T-13-21 (no false success events).
|
|
# The audit payload is metadata-only: no provider URLs, tokens, bytes, or
|
|
# document text are included (T-13-02).
|
|
await write_audit_log(
|
|
session,
|
|
event_type="cloud.file_uploaded",
|
|
user_id=current_user.id,
|
|
actor_id=current_user.id,
|
|
resource_id=None,
|
|
ip_address=get_client_ip(request),
|
|
metadata_={
|
|
"connection_id": str(connection_id),
|
|
"provider_item_id": provider_item_id,
|
|
"filename": resolved_name,
|
|
"size_bytes": resolved_size,
|
|
"parent_ref": resolved_parent_ref,
|
|
},
|
|
)
|
|
|
|
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)
|