From ba48d625dd6bcdd5ead1faaa3be683a1d2832939 Mon Sep 17 00:00:00 2001 From: curo1305 Date: Tue, 23 Jun 2026 15:50:37 +0200 Subject: [PATCH] feat(14-04): add analysis API routes and typed schemas MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - POST /analysis/connections/{id}/estimate: scope estimate without bytes (T-14-04) - POST /analysis/connections/{id}/jobs: enqueue job, returns job_id (ANALYZE-01) - GET /analysis/jobs, GET /analysis/jobs/{id}: job status list and detail - POST /analysis/jobs/{id}/cancel: batch cancel (queued items immediate; running cooperative) - POST /analysis/jobs/{id}/items/{item_id}/cancel|skip|retry: per-item controls (ANALYZE-05) - AnalysisEstimateRequest/Out, AnalysisEnqueueRequest/Out, AnalysisJobOut: typed schemas - AnalysisJobOut: both simple (waiting/working/done/skipped/failed) and per-stage counts - All routes use get_regular_user — admin blocked (T-14-01) - Response schemas exclude credentials_enc, object_key, raw provider URLs (T-14-02) - 14/14 test_cloud_analysis_api.py pass; 7 new analysis security tests pass (33/33 total) --- backend/api/cloud/analysis.py | 513 ++++++++++++++++++++++++++- backend/api/cloud/schemas.py | 140 ++++++++ backend/tests/test_cloud_security.py | 202 +++++++++++ 3 files changed, 845 insertions(+), 10 deletions(-) diff --git a/backend/api/cloud/analysis.py b/backend/api/cloud/analysis.py index 59453a1..aeda92c 100644 --- a/backend/api/cloud/analysis.py +++ b/backend/api/cloud/analysis.py @@ -5,25 +5,518 @@ Provides the /analysis sub-router tree under /api/cloud. All routes are owner-scoped and blocked for admin accounts (get_regular_user enforced at each leaf router). -Currently registered routes: - GET /analysis/cache — cache usage/settings (cloud/cache.py) - PATCH /analysis/cache/settings — update settings (cloud/cache.py) +Route family (Plan 03 cache + Plan 04 jobs/controls): + GET /analysis/cache — cache usage/settings + PATCH /analysis/cache/settings — update settings + POST /analysis/connections/{id}/estimate — scope estimate (no bytes) + POST /analysis/connections/{id}/jobs — enqueue analysis job + GET /analysis/jobs — list jobs + GET /analysis/jobs/{job_id} — job status + POST /analysis/jobs/{job_id}/cancel — cancel batch + POST /analysis/jobs/{job_id}/items/{item_id}/cancel — cancel single item + POST /analysis/jobs/{job_id}/items/{item_id}/skip — skip single item + POST /analysis/jobs/{job_id}/items/{item_id}/retry — retry failed item -Future routes (Phase 14 plans 04-09): - POST /analysis/connections/{id}/estimate - POST /analysis/connections/{id}/jobs - GET /analysis/jobs/{id} - POST /analysis/jobs/{id}/cancel - POST /analysis/jobs/{id}/items/{item_id}/skip +Security invariants (T-14-01, T-14-02, T-14-03): + - All routes use get_regular_user — admin accounts blocked. + - All routes are owner-scoped: connection_id and job_id resolved against user. + - Response schemas never include credentials_enc, object_key, or provider URLs. + - No provider mutation methods are called in this module. Module is importable as api.cloud.analysis for tests and callers that verify router registration before endpoint implementation is complete. """ from __future__ import annotations -from fastapi import APIRouter +import uuid +from typing import List, Optional + +from fastapi import APIRouter, Depends, HTTPException, Query, Request, status +from fastapi.responses import JSONResponse +from sqlalchemy.ext.asyncio import AsyncSession from api.cloud.cache import router as cache_router +from api.cloud.schemas import ( + AnalysisControlOut, + AnalysisEnqueueOut, + AnalysisEnqueueRequest, + AnalysisEstimateOut, + AnalysisEstimateRequest, + AnalysisJobItemOut, + AnalysisJobOut, +) +from db.models import User +from deps.auth import get_regular_user +from deps.db import get_db +from services.cloud_analysis import ( + AnalysisItemNotFound, + AnalysisJobNotFound, + cancel_job, + cancel_job_item, + enqueue_analysis_job, + estimate_scope, + get_analysis_job, + list_analysis_jobs, + list_job_items, + retry_job_item, + skip_job_item, + InvalidJobState, +) +from services.cloud_items import ConnectionNotFound +from services.rate_limiting import account_limiter + router = APIRouter() router.include_router(cache_router) + +# ── Estimate ─────────────────────────────────────────────────────────────────── + + +@router.post( + "/analysis/connections/{connection_id}/estimate", + response_model=AnalysisEstimateOut, +) +@account_limiter.limit("60/minute") +async def estimate_analysis_scope( + request: Request, + connection_id: uuid.UUID, + body: AnalysisEstimateRequest, + session: AsyncSession = Depends(get_db), + current_user: User = Depends(get_regular_user), +) -> AnalysisEstimateOut: + """Estimate analysis scope without downloading provider bytes. + + Returns counts and byte totals for the requested scope. No bytes are + fetched from the provider. No provider mutations are made. + + T-14-04: bytes never downloaded during estimate. + T-14-03: no provider mutations. + T-14-01: connection must be owned by current_user. + """ + request.state.current_user = current_user + + try: + result = await estimate_scope( + session, + user_id=current_user.id, + connection_id=connection_id, + scope=body.scope, + provider_item_ids=body.provider_item_ids, + recursive=body.recursive, + ) + except ConnectionNotFound as exc: + raise HTTPException( + status_code=status.HTTP_404_NOT_FOUND, + detail=str(exc), + ) from exc + except ValueError as exc: + raise HTTPException( + status_code=status.HTTP_422_UNPROCESSABLE_ENTITY, + detail=str(exc), + ) from exc + + return AnalysisEstimateOut( + supported_count=result.supported_count, + unsupported_count=result.unsupported_count, + total_provider_bytes=result.total_provider_bytes, + recursive=result.recursive, + is_partial=result.is_partial, + scope_kind=result.scope_kind, + ) + + +# ── Enqueue ──────────────────────────────────────────────────────────────────── + + +@router.post( + "/analysis/connections/{connection_id}/jobs", + response_model=AnalysisEnqueueOut, + status_code=202, +) +@account_limiter.limit("30/minute") +async def enqueue_job( + request: Request, + connection_id: uuid.UUID, + body: AnalysisEnqueueRequest, + session: AsyncSession = Depends(get_db), + current_user: User = Depends(get_regular_user), +) -> AnalysisEnqueueOut: + """Enqueue an analysis job for the requested scope. + + Creates a CloudAnalysisJob and populates job items. Already-current items + are skipped without downloading bytes. Returns a stable job_id for polling. + + Accepts failure_behavior override for this job (D-11). + + T-14-01: connection must be owned by current_user. + T-14-03: no provider mutations. + T-14-04: no bytes downloaded for already-current checks. + """ + request.state.current_user = current_user + + # Validate failure_behavior before calling the service + valid_fb = {"pause_batch", "continue_item"} + if body.failure_behavior not in valid_fb: + raise HTTPException( + status_code=status.HTTP_422_UNPROCESSABLE_ENTITY, + detail=f"Invalid failure_behavior {body.failure_behavior!r}. Valid values: {sorted(valid_fb)}", + ) + + try: + result = await enqueue_analysis_job( + session, + user_id=current_user.id, + connection_id=connection_id, + scope=body.scope, + provider_item_ids=body.provider_item_ids, + recursive=body.recursive, + failure_behavior=body.failure_behavior, + ) + except ConnectionNotFound as exc: + raise HTTPException( + status_code=status.HTTP_404_NOT_FOUND, + detail=str(exc), + ) from exc + except ValueError as exc: + raise HTTPException( + status_code=status.HTTP_422_UNPROCESSABLE_ENTITY, + detail=str(exc), + ) from exc + + await session.commit() + + return AnalysisEnqueueOut( + job_id=str(result.job_id), + status="queued", + total_count=result.total_count, + queued_count=result.queued_count, + already_current_count=result.already_current_count, + unsupported_count=result.unsupported_count, + ) + + +# ── Job status / list ────────────────────────────────────────────────────────── + + +def _build_job_out(job, *, detail: bool = False) -> AnalysisJobOut: + """Build a credential-free AnalysisJobOut from a CloudAnalysisJob row. + + Simple mode aggregates internal statuses into human-friendly labels: + waiting = queued + working = downloading + extracting + classifying + done = indexed + already_current + skipped = cancelled + unsupported + failed = failed + + Detailed mode includes per-stage counts. + """ + # Simple aggregate labels + waiting_count = job.queued_count or 0 + working_count = ( + (job.downloading_count or 0) + + (job.extracting_count or 0) + + (job.classifying_count or 0) + ) + done_count = (job.indexed_count or 0) + (job.already_current_count or 0) + skipped_count = (job.cancelled_count or 0) + (job.unsupported_count or 0) + failed_count = job.failed_count or 0 + + out = AnalysisJobOut( + job_id=str(job.id), + connection_id=str(job.connection_id), + scope_kind=job.scope_kind, + status=job.status, + failure_behavior=job.failure_behavior, + recursive=job.recursive, + total_count=job.total_count or 0, + waiting_count=waiting_count, + working_count=working_count, + done_count=done_count, + skipped_count=skipped_count, + failed_count=failed_count, + # Per-stage counts always included (0 when not in detailed mode) + queued_count=job.queued_count or 0, + downloading_count=job.downloading_count or 0, + extracting_count=job.extracting_count or 0, + classifying_count=job.classifying_count or 0, + indexed_count=job.indexed_count or 0, + already_current_count=job.already_current_count or 0, + cancelled_count=job.cancelled_count or 0, + unsupported_count=job.unsupported_count or 0, + created_at=job.created_at, + started_at=job.started_at, + finished_at=job.finished_at, + ) + + return out + + +@router.get( + "/analysis/jobs", + response_model=List[AnalysisJobOut], +) +@account_limiter.limit("120/minute") +async def list_jobs( + request: Request, + connection_id: Optional[uuid.UUID] = Query(default=None), + job_status: Optional[str] = Query(default=None, alias="status"), + detail: bool = Query(default=False), + session: AsyncSession = Depends(get_db), + current_user: User = Depends(get_regular_user), +) -> List[AnalysisJobOut]: + """List analysis jobs owned by the current user. + + Optional filters: connection_id, status. + Use ?detail=true for per-stage counts (ANALYZE-04). + """ + request.state.current_user = current_user + + jobs = await list_analysis_jobs( + session, + user_id=current_user.id, + connection_id=connection_id, + status=job_status, + ) + return [_build_job_out(j, detail=detail) for j in jobs] + + +@router.get( + "/analysis/jobs/{job_id}", + response_model=AnalysisJobOut, +) +@account_limiter.limit("120/minute") +async def get_job_status( + request: Request, + job_id: uuid.UUID, + detail: bool = Query(default=False), + session: AsyncSession = Depends(get_db), + current_user: User = Depends(get_regular_user), +) -> AnalysisJobOut: + """Return job status for a single analysis job. + + Use ?detail=true for per-stage item counts (ANALYZE-04). + Response never includes credentials, object_key, or raw provider data. + """ + request.state.current_user = current_user + + try: + job = await get_analysis_job(session, job_id=job_id, user_id=current_user.id) + except AnalysisJobNotFound as exc: + raise HTTPException( + status_code=status.HTTP_404_NOT_FOUND, + detail=str(exc), + ) from exc + + return _build_job_out(job, detail=detail) + + +# ── Cancel job ───────────────────────────────────────────────────────────────── + + +@router.post( + "/analysis/jobs/{job_id}/cancel", + response_model=AnalysisControlOut, +) +@account_limiter.limit("30/minute") +async def cancel_analysis_job( + request: Request, + job_id: uuid.UUID, + session: AsyncSession = Depends(get_db), + current_user: User = Depends(get_regular_user), +) -> AnalysisControlOut: + """Cancel a queued or running analysis batch. + + Queued items transition to cancelled immediately. + Running items are marked for cooperative cancellation. + + Returns a typed result — not HTTPException — so kind/reason appear at + response top level (Phase 13 mutation response style). + """ + request.state.current_user = current_user + + try: + await cancel_job(session, job_id=job_id, user_id=current_user.id) + except AnalysisJobNotFound as exc: + raise HTTPException( + status_code=status.HTTP_404_NOT_FOUND, + detail=str(exc), + ) from exc + except InvalidJobState: + # Already terminal — still return 200 with typed reason + return AnalysisControlOut( + kind="cancelled", + reason="already_terminal", + job_id=str(job_id), + ) + + await session.commit() + + return AnalysisControlOut( + kind="cancelled", + reason="batch_cancelled", + job_id=str(job_id), + ) + + +# ── Cancel item ──────────────────────────────────────────────────────────────── + + +@router.post( + "/analysis/jobs/{job_id}/items/{item_id}/cancel", + response_model=AnalysisControlOut, +) +@account_limiter.limit("60/minute") +async def cancel_analysis_item( + request: Request, + job_id: uuid.UUID, + item_id: uuid.UUID, + session: AsyncSession = Depends(get_db), + current_user: User = Depends(get_regular_user), +) -> AnalysisControlOut: + """Cancel a single queued or in-flight job item. + + item_id is the cloud_item_id (DocuVault stable UUID). + """ + request.state.current_user = current_user + + try: + await cancel_job_item( + session, + job_id=job_id, + item_id=item_id, + user_id=current_user.id, + ) + except AnalysisJobNotFound as exc: + raise HTTPException( + status_code=status.HTTP_404_NOT_FOUND, + detail=str(exc), + ) from exc + except AnalysisItemNotFound as exc: + raise HTTPException( + status_code=status.HTTP_404_NOT_FOUND, + detail=str(exc), + ) from exc + except InvalidJobState as exc: + return JSONResponse( + status_code=status.HTTP_409_CONFLICT, + content={"kind": "invalid_state", "reason": str(exc)}, + ) + + await session.commit() + + return AnalysisControlOut( + kind="cancelled", + reason="item_cancelled", + job_id=str(job_id), + item_id=str(item_id), + ) + + +# ── Skip item ────────────────────────────────────────────────────────────────── + + +@router.post( + "/analysis/jobs/{job_id}/items/{item_id}/skip", + response_model=AnalysisControlOut, +) +@account_limiter.limit("60/minute") +async def skip_analysis_item( + request: Request, + job_id: uuid.UUID, + item_id: uuid.UUID, + session: AsyncSession = Depends(get_db), + current_user: User = Depends(get_regular_user), +) -> AnalysisControlOut: + """Skip a queued or failed job item. + + Marks the item as cancelled (skip is a user-initiated cancel variant). + item_id is the cloud_item_id (DocuVault stable UUID). + """ + request.state.current_user = current_user + + try: + await skip_job_item( + session, + job_id=job_id, + item_id=item_id, + user_id=current_user.id, + ) + except AnalysisJobNotFound as exc: + raise HTTPException( + status_code=status.HTTP_404_NOT_FOUND, + detail=str(exc), + ) from exc + except AnalysisItemNotFound as exc: + raise HTTPException( + status_code=status.HTTP_404_NOT_FOUND, + detail=str(exc), + ) from exc + except InvalidJobState as exc: + return JSONResponse( + status_code=status.HTTP_409_CONFLICT, + content={"kind": "invalid_state", "reason": str(exc)}, + ) + + await session.commit() + + return AnalysisControlOut( + kind="skipped", + reason="item_skipped", + job_id=str(job_id), + item_id=str(item_id), + ) + + +# ── Retry item ───────────────────────────────────────────────────────────────── + + +@router.post( + "/analysis/jobs/{job_id}/items/{item_id}/retry", + response_model=AnalysisControlOut, +) +@account_limiter.limit("30/minute") +async def retry_analysis_item( + request: Request, + job_id: uuid.UUID, + item_id: uuid.UUID, + session: AsyncSession = Depends(get_db), + current_user: User = Depends(get_regular_user), +) -> AnalysisControlOut: + """Retry a failed job item from the beginning. + + Resets the item to queued status, increments retry_count, clears error fields. + item_id is the cloud_item_id (DocuVault stable UUID). + """ + request.state.current_user = current_user + + try: + await retry_job_item( + session, + job_id=job_id, + item_id=item_id, + user_id=current_user.id, + ) + except AnalysisJobNotFound as exc: + raise HTTPException( + status_code=status.HTTP_404_NOT_FOUND, + detail=str(exc), + ) from exc + except AnalysisItemNotFound as exc: + raise HTTPException( + status_code=status.HTTP_404_NOT_FOUND, + detail=str(exc), + ) from exc + except InvalidJobState as exc: + return JSONResponse( + status_code=status.HTTP_409_CONFLICT, + content={"kind": "invalid_state", "reason": str(exc)}, + ) + + await session.commit() + + return AnalysisControlOut( + kind="retried", + reason="item_requeued", + job_id=str(job_id), + item_id=str(item_id), + ) diff --git a/backend/api/cloud/schemas.py b/backend/api/cloud/schemas.py index 594dd03..88da3bf 100644 --- a/backend/api/cloud/schemas.py +++ b/backend/api/cloud/schemas.py @@ -255,3 +255,143 @@ class CacheSettingsUpdateRequest(BaseModel): ge=1, description="Preferred byte ceiling for the local byte cache", ) + + +# ── Phase 14 analysis request schemas ───────────────────────────────────────── + +class AnalysisEstimateRequest(BaseModel): + """Request body for POST /analysis/connections/{id}/estimate. + + scope: "file" | "selection" | "folder" | "connection" + provider_item_ids: Required for file/selection/folder scope. Omitted for connection scope. + recursive: Expand folder subtree recursively (folder scope only; connection always recursive). + """ + + scope: str = Field(..., description="file | selection | folder | connection") + provider_item_ids: Optional[List[str]] = Field( + default=None, + description="Provider item IDs for file/selection/folder scope", + ) + recursive: bool = Field( + default=False, + description="Expand folder children recursively (folder scope)", + ) + + +class AnalysisEnqueueRequest(BaseModel): + """Request body for POST /analysis/connections/{id}/jobs. + + scope: "file" | "selection" | "folder" | "connection" + provider_item_ids: Required for file/selection/folder scope. + recursive: Expand folder subtree recursively. + failure_behavior: "pause_batch" (default) | "continue_item" (D-11). + """ + + scope: str = Field(..., description="file | selection | folder | connection") + provider_item_ids: Optional[List[str]] = Field(default=None) + recursive: bool = Field(default=False) + failure_behavior: str = Field( + default="pause_batch", + description="pause_batch | continue_item", + ) + + +# ── Phase 14 analysis response schemas ──────────────────────────────────────── + +class AnalysisEstimateOut(BaseModel): + """Estimate response — no credentials, bytes, or object_key (T-14-02). + + supported_count: Files that can be analysed. + unsupported_count: Items that cannot be analysed (unsupported type, folder). + total_provider_bytes: Sum of provider_size across supported items. + recursive: Whether the estimate was recursive. + is_partial: True when metadata expansion was incomplete. + scope_kind: Echo of the requested scope. + """ + + supported_count: int = 0 + unsupported_count: int = 0 + total_provider_bytes: int = 0 + recursive: bool = False + is_partial: bool = False + scope_kind: str + + +class AnalysisJobOut(BaseModel): + """Job status response — no credentials or object_key (T-14-02). + + Exposes both simple and detailed aggregate counts. Caller requests detail + via ?detail=true query parameter. + + Simple (default): waiting_count, working_count, done_count, + skipped_count, failed_count, total_count. + Detailed: queued_count, downloading_count, extracting_count, + classifying_count, indexed_count, already_current_count, + cancelled_count, failed_count, unsupported_count. + """ + + job_id: str + connection_id: str + scope_kind: str + status: str + failure_behavior: str + recursive: bool = False + total_count: int = 0 + + # Simple labels (always present) + waiting_count: int = 0 + working_count: int = 0 + done_count: int = 0 + skipped_count: int = 0 + failed_count: int = 0 + + # Detailed labels (always present — 0 when detail=false, per-stage values when detail=true) + queued_count: int = 0 + downloading_count: int = 0 + extracting_count: int = 0 + classifying_count: int = 0 + indexed_count: int = 0 + already_current_count: int = 0 + cancelled_count: int = 0 + unsupported_count: int = 0 + + created_at: Optional[datetime] = None + started_at: Optional[datetime] = None + finished_at: Optional[datetime] = None + + +class AnalysisEnqueueOut(BaseModel): + """Enqueue response — returns a stable job_id for status polling (ANALYZE-01). + + No credentials or object_key in response (T-14-02). + """ + + job_id: str + status: str + total_count: int = 0 + queued_count: int = 0 + already_current_count: int = 0 + unsupported_count: int = 0 + + +class AnalysisJobItemOut(BaseModel): + """Per-item job status — no credentials, object_key, or raw provider data (T-14-02).""" + + id: str + cloud_item_id: str + provider_item_id: str + status: str + error_code: Optional[str] = None + error_message: Optional[str] = None + retry_count: int = 0 + created_at: Optional[datetime] = None + finished_at: Optional[datetime] = None + + +class AnalysisControlOut(BaseModel): + """Generic control result for cancel/skip/retry operations.""" + + kind: str # "cancelled" | "skipped" | "retried" + reason: Optional[str] = None + job_id: Optional[str] = None + item_id: Optional[str] = None diff --git a/backend/tests/test_cloud_security.py b/backend/tests/test_cloud_security.py index 74edfc2..a67a165 100644 --- a/backend/tests/test_cloud_security.py +++ b/backend/tests/test_cloud_security.py @@ -704,3 +704,205 @@ async def test_cache_settings_limit_above_tier_cap_returns_422(async_client, db_ assert resp.status_code == 422, ( f"Cache limit above tier cap must return 422, got {resp.status_code}" ) + + +# ── Phase 14: Analysis job API security (T-14-01, T-14-02, T-14-03) ──────────── + + +async def _create_analysis_connection(session, user_id, provider="google_drive"): + """Helper: create a cloud connection for analysis security tests.""" + from db.models import CloudConnection + from storage.cloud_utils import encrypt_credentials + from config import settings as app_settings + + master_key = app_settings.cloud_creds_key.encode() + creds_enc = encrypt_credentials( + master_key, + str(user_id), + {"access_token": "tok", "refresh_token": "ref"}, + ) + conn = CloudConnection( + id=_uuid.uuid4(), + user_id=user_id, + provider=provider, + display_name="Security Test Connection", + credentials_enc=creds_enc, + status="ACTIVE", + ) + session.add(conn) + await session.commit() + return conn + + +async def _create_analysis_item(session, user_id, connection_id, name="doc.pdf"): + """Helper: create a cloud item for analysis security tests.""" + from db.models import CloudItem + + item = CloudItem( + id=_uuid.uuid4(), + user_id=user_id, + connection_id=connection_id, + provider_item_id=f"sec-pitem-{_uuid.uuid4().hex[:8]}", + name=name, + kind="file", + content_type="application/pdf", + provider_size=102400, + etag="etag-sec-v1", + analysis_status="pending", + ) + session.add(item) + await session.commit() + return item + + +async def test_analysis_estimate_admin_blocked(async_client, db_session): + """T-14-01: Admin cannot access analysis estimate endpoint.""" + admin_ctx = await _create_user_and_token(db_session, role="admin") + user_ctx = await _create_user_and_token(db_session, role="user") + conn = await _create_analysis_connection(db_session, user_ctx["user"].id) + + resp = await async_client.post( + f"/api/cloud/analysis/connections/{conn.id}/estimate", + json={"scope": "connection", "recursive": True}, + headers=admin_ctx["headers"], + ) + + assert resp.status_code in (403, 404), ( + f"Admin must not access analysis estimate — got {resp.status_code}" + ) + + +async def test_analysis_enqueue_admin_blocked(async_client, db_session): + """T-14-01: Admin cannot enqueue analysis jobs.""" + admin_ctx = await _create_user_and_token(db_session, role="admin") + user_ctx = await _create_user_and_token(db_session, role="user") + conn = await _create_analysis_connection(db_session, user_ctx["user"].id) + + resp = await async_client.post( + f"/api/cloud/analysis/connections/{conn.id}/jobs", + json={"scope": "connection", "recursive": True}, + headers=admin_ctx["headers"], + ) + + assert resp.status_code in (403, 404), ( + f"Admin must not enqueue analysis jobs — got {resp.status_code}" + ) + + +async def test_analysis_estimate_unauthenticated_blocked(async_client, db_session): + """Analysis estimate requires authentication.""" + conn_id = str(_uuid.uuid4()) + + resp = await async_client.post( + f"/api/cloud/analysis/connections/{conn_id}/estimate", + json={"scope": "connection", "recursive": True}, + ) + + assert resp.status_code in (401, 403), ( + f"Unauthenticated analysis estimate must be rejected — got {resp.status_code}" + ) + + +async def test_analysis_enqueue_response_excludes_credentials(async_client, db_session): + """T-14-02: Enqueue response must not contain credentials or object_key.""" + from unittest.mock import patch as _patch + + user_ctx = await _create_user_and_token(db_session, role="user") + conn = await _create_analysis_connection(db_session, user_ctx["user"].id) + item = await _create_analysis_item(db_session, user_ctx["user"].id, conn.id) + + with _patch("services.cloud_analysis.get_adapter", return_value=MagicMock()): + resp = await async_client.post( + f"/api/cloud/analysis/connections/{conn.id}/jobs", + json={"scope": "file", "provider_item_ids": [item.provider_item_id]}, + headers=user_ctx["headers"], + ) + + assert resp.status_code in (200, 202), ( + f"Enqueue must succeed — got {resp.status_code}: {resp.text}" + ) + body_text = resp.text + + # T-14-02: No credential or internal storage fields in response + for field in ("credentials_enc", "access_token", "refresh_token", "object_key"): + assert field not in body_text, ( + f"T-14-02: Sensitive field '{field}' must not appear in enqueue response" + ) + + +async def test_analysis_job_status_excludes_credentials(async_client, db_session): + """T-14-02: Job status response must not contain credentials or object_key.""" + from unittest.mock import patch as _patch + + user_ctx = await _create_user_and_token(db_session, role="user") + conn = await _create_analysis_connection(db_session, user_ctx["user"].id) + item = await _create_analysis_item(db_session, user_ctx["user"].id, conn.id) + + with _patch("services.cloud_analysis.get_adapter", return_value=MagicMock()): + enqueue_resp = await async_client.post( + f"/api/cloud/analysis/connections/{conn.id}/jobs", + json={"scope": "file", "provider_item_ids": [item.provider_item_id]}, + headers=user_ctx["headers"], + ) + + assert enqueue_resp.status_code in (200, 202) + job_id = enqueue_resp.json().get("job_id") + assert job_id is not None + + status_resp = await async_client.get( + f"/api/cloud/analysis/jobs/{job_id}", + headers=user_ctx["headers"], + ) + assert status_resp.status_code == 200 + body_text = status_resp.text + + for field in ("credentials_enc", "access_token", "refresh_token", "object_key"): + assert field not in body_text, ( + f"T-14-02: Sensitive field '{field}' must not appear in job status response" + ) + + +async def test_foreign_user_cannot_access_analysis_estimate(async_client, db_session): + """T-14-01: Foreign user cannot estimate analysis scope for another user's connection.""" + owner_ctx = await _create_user_and_token(db_session, role="user") + foreign_ctx = await _create_user_and_token(db_session, role="user") + owner_conn = await _create_analysis_connection(db_session, owner_ctx["user"].id) + + resp = await async_client.post( + f"/api/cloud/analysis/connections/{owner_conn.id}/estimate", + json={"scope": "connection", "recursive": True}, + headers=foreign_ctx["headers"], + ) + + assert resp.status_code in (403, 404), ( + f"Foreign user must not estimate owner's connection — got {resp.status_code}" + ) + + +async def test_foreign_user_cannot_read_analysis_job_status(async_client, db_session): + """T-14-01: Foreign user cannot read job status for another user's job.""" + from unittest.mock import patch as _patch + + owner_ctx = await _create_user_and_token(db_session, role="user") + foreign_ctx = await _create_user_and_token(db_session, role="user") + owner_conn = await _create_analysis_connection(db_session, owner_ctx["user"].id) + owner_item = await _create_analysis_item(db_session, owner_ctx["user"].id, owner_conn.id) + + with _patch("services.cloud_analysis.get_adapter", return_value=MagicMock()): + enqueue_resp = await async_client.post( + f"/api/cloud/analysis/connections/{owner_conn.id}/jobs", + json={"scope": "file", "provider_item_ids": [owner_item.provider_item_id]}, + headers=owner_ctx["headers"], + ) + + assert enqueue_resp.status_code in (200, 202) + job_id = enqueue_resp.json().get("job_id") + + resp = await async_client.get( + f"/api/cloud/analysis/jobs/{job_id}", + headers=foreign_ctx["headers"], + ) + + assert resp.status_code in (403, 404), ( + f"Foreign user must not read owner's job — got {resp.status_code}" + )