"""HTTP routes for object storage — presigned PUT/GET for user uploads. Single namespace ``/api/v1/uploads`` keeps the surface tight. New upload use-cases register a ``purpose`` enum value in :data:`_ALLOWED_PURPOSES` and a per-purpose object-key prefix in :func:`_build_relative_key`; the actual S3 client stays in ``app.core.storage``. The presign-put endpoint signs conditional PUT requests with the caller's identity baked into the key path so admin audit can later trace "who uploaded this". The presign-get endpoint enforces per-purpose ACL by parsing the key shape: the key path itself carries the uploader's person_id (segment after ``{purpose}/{yyyy}/{mm}/{dd}/``), so we can refuse a caller that neither uploaded the object nor holds a read role for that purpose. The presigned URL bypasses Helix auth once minted, so this endpoint is the last line of defence — bare authentication is NOT enough, because any logged-in user could otherwise mint a download URL for any object key whose name they observed (audit logs / shared screens / accidental copy-paste). """ from __future__ import annotations import logging import uuid from collections.abc import Callable, Iterable from typing import Final from fastapi import APIRouter, Depends, HTTPException, Request, status from pydantic import BaseModel, Field from sqlalchemy.ext.asyncio import AsyncSession from app.audit.service import AuditService from app.auth.models import AuthUser from app.core.database import get_db from app.core.dependencies import get_current_user, get_real_ip from app.core.exceptions import AppException from app.core.rbac import can_manage_security, can_read_all_assets from app.core.response import success_response from app.core.storage import ( StorageNotConfiguredError, presign_get, presign_put, ) from app.core.storage_uploads import ( ALLOWED_CONTENT_TYPES, ALLOWED_PURPOSES, build_relative_upload_key, create_upload_grant, safe_upload_filename, ) logger = logging.getLogger(__name__) router = APIRouter(prefix="/api/v1/uploads", tags=["uploads"]) # Each registered purpose maps to a top-level path prefix under the # bucket — adding a new flow means appending here, not editing call # sites. Keys travel through the DB so making them human-readable is # worth the few extra bytes. _ALLOWED_PURPOSES = ALLOWED_PURPOSES # Browsers from mobile (Feishu in-app) frequently produce HEIC/HEIF. # Block obvious script types; we don't need PDF / docx for current # flows (only images), narrow until needed. _ALLOWED_CONTENT_TYPES = ALLOWED_CONTENT_TYPES def _safe_filename(raw: str) -> str: return safe_upload_filename(raw) def _build_relative_key(*, purpose: str, person_id: uuid.UUID, filename: str) -> str: return build_relative_upload_key( purpose=purpose, person_id=person_id, filename=filename, ) def _parse_object_key(key: str) -> tuple[str, uuid.UUID] | None: """Best-effort parse: return ``(purpose, uploader_id)`` for a key produced by :func:`_build_relative_key`. Layout (post storage-prefix): ``{purpose}/{yyyy}/{mm}/{dd}/{person_id}/{filename}``. The optional storage prefix (``""`` in prod, ``"dev/"`` in non-prod, or whatever admin sets in ``platform_settings.storage.s3.key_prefix``) is stripped by scanning for the first segment that matches a known ``_ALLOWED_PURPOSES`` value — robust to any prefix change without consulting the DB. """ # Reject any path-traversal segment up front: the ACL below validates a # parsed slice of the key but the RAW key is what gets handed to S3, so # the parsed shape and the fetched object must not be allowed to diverge. parts = key.split("/") if any(p in ("", ".", "..") for p in parts): return None for i, segment in enumerate(parts): if segment in _ALLOWED_PURPOSES: remaining = parts[i + 1:] # Expect EXACTLY: yyyy / mm / dd / person_id / filename. An exact # length (not a floor) keeps the validated parse byte-aligned # with the real key — a crafted key with extra trailing segments # can no longer pass the uploader-id gate while pointing S3 at a # different object. if len(remaining) != 5: return None try: uploader_id = uuid.UUID(remaining[3]) except (ValueError, IndexError): return None return segment, uploader_id return None #: Public alias — other modules that accept a client-supplied object key #: (currently the feed, which stores keys on ``feed_attachments`` and later #: presigns them itself) must validate the key's SHAPE and OWNERSHIP with #: the same parser this router's ACL uses. Anything that mints a presigned #: GET is an authorisation decision, wherever it lives. parse_object_key = _parse_object_key #: Purposes whose objects the feed is allowed to attach to a post. FEED_PURPOSES: Final[frozenset[str]] = frozenset({ "feed_post_image", "feed_announcement", }) #: Purposes whose objects ANY authenticated employee may read. #: #: This is the one place that decision is recorded, because two places #: recording it is how an ACL and the test that guards it drift apart — #: ``test_every_allowed_purpose_has_an_explicit_read_policy`` asserts #: against this same set, so adding a purpose here is a visible, #: reviewable act rather than a literal quietly edited in two files. #: #: * ``feed_post_image`` / ``feed_announcement`` — company-internal feed #: content. The post-level visibility gate lives inside ``FeedService`` #: (a department-scoped post hides its attachment list from #: out-of-department viewers, so they never learn the ``object_key``). #: This route is the fallback for MCP clients caching keys and the #: mobile webview re-fetching on resume. Public-feed semantics by #: design. #: * ``awareness_material`` — security-awareness comics and article #: images are published TO the whole workforce; every employee seeing #: them is the entire point of the domain. Note the asymmetry with #: ``_PURPOSE_WRITE_GATES`` below: only the security desk uploads #: these, everyone reads them. WORKFORCE_READABLE_PURPOSES: Final[frozenset[str]] = frozenset({ "feed_post_image", "feed_announcement", "awareness_material", }) #: Purposes that may only be UPLOADED by a specific role set. #: #: Historically every purpose was open to any authenticated employee on #: the write side, with the domain's own attach endpoint doing the real #: gating (you can upload a photo, but only an asset admin can pin it to #: a disposal record). That is defensible for evidence photos an employee #: legitimately produces. It is not defensible for a purpose whose whole #: population is authored by one small team: leaving #: ``awareness_material`` open would let any employee spend the org's #: bucket quota on content that only the security desk is ever supposed #: to write. #: #: A purpose absent from this map keeps the historical behaviour — this #: is additive, not a policy change for existing flows. _PURPOSE_WRITE_GATES: Final[dict[str, Callable[[Iterable[str]], bool]]] = { "awareness_material": can_manage_security, } #: Purposes whose objects AI Labs is allowed to consume as input. Same #: contract as :data:`FEED_PURPOSES`: the domain validates a client-supplied #: key's shape and ownership with :func:`parse_object_key` before it ever #: reads the bytes, so an employee cannot point a chat attachment at #: someone else's evidence photo. AI_LABS_PURPOSES: Final[frozenset[str]] = frozenset({ "ai_labs_image", "ai_labs_document", "ai_labs_audio", }) class PresignPutRequest(BaseModel): purpose: str = Field(..., max_length=64) filename: str = Field(..., max_length=200) content_type: str = Field(..., max_length=128) size_bytes: int = Field(..., gt=0) checksum_sha256: str = Field(..., min_length=44, max_length=44) class PresignPutResponse(BaseModel): url: str headers: dict[str, str] object_key: str bucket: str expires_in: int class PresignGetRequest(BaseModel): object_key: str = Field(..., max_length=512) class PresignGetResponse(BaseModel): url: str expires_in: int @router.post("/presign-put", summary="Get a single-create presigned S3 PUT") async def presign_put_endpoint( payload: PresignPutRequest, request: Request, current_user: AuthUser = Depends(get_current_user), db: AsyncSession = Depends(get_db), ): # Per-purpose write gate (see ``_PURPOSE_WRITE_GATES``). Checked # before the grant is minted so a refused caller never consumes # their upload quota, and never lands a ledger row. gate = _PURPOSE_WRITE_GATES.get(payload.purpose) if gate is not None and not gate(current_user.role_names): raise HTTPException( status_code=status.HTTP_403_FORBIDDEN, detail=f"not authorised to upload for purpose '{payload.purpose}'", ) try: grant = await create_upload_grant( db, person_id=current_user.id, purpose=payload.purpose, filename=payload.filename, content_type=payload.content_type, size_bytes=payload.size_bytes, checksum_sha256=payload.checksum_sha256, ) except StorageNotConfiguredError as e: logger.error("storage not configured: %s", e) raise AppException( code="STORAGE_NOT_CONFIGURED", message="storage backend not configured", status_code=503, ) from e await AuditService(db).log( trace_id=request.state.trace_id, actor_type="user", actor_user_id=current_user.id, action="storage.upload.authorize", resource_type="storage_upload_grant", resource_id=str(grant.id), status="success", detail={ "purpose": grant.purpose, "object_key": grant.object_key, "declared_size_bytes": grant.declared_size_bytes, "content_type": grant.content_type, }, ip_address=get_real_ip(request), user_agent=request.headers.get("User-Agent"), ) # Persist the cleanup ledger before issuing the external capability. If # signing or the client upload fails, the orphan sweep can still recover it. await db.commit() try: result = await presign_put( db, object_key=grant.object_key, content_type=grant.content_type, size_bytes=grant.declared_size_bytes, checksum_sha256=grant.checksum_sha256, ) except StorageNotConfiguredError as e: logger.error("storage not configured: %s", e) raise AppException( code="STORAGE_NOT_CONFIGURED", message="storage backend not configured", status_code=503, ) from e return success_response(data=result, request=request) @router.post("/presign-get", summary="Get a presigned S3 GET URL") async def presign_get_endpoint( payload: PresignGetRequest, request: Request, current_user: AuthUser = Depends(get_current_user), db: AsyncSession = Depends(get_db), ): # Per-purpose ACL — the presigned URL bypasses Helix auth once # minted, so bare authentication is NOT enough. Parse the # uploader's person_id out of the key path and refuse callers who # are neither the uploader nor in the purpose's read-role set. # Without this gate any logged-in user who observes an object_key # (audit logs, screenshots, shared screens) can mint a download URL. parsed = _parse_object_key(payload.object_key) if parsed is None: raise HTTPException( status_code=status.HTTP_403_FORBIDDEN, detail="object_key shape not recognised", ) purpose, uploader_id = parsed if purpose == "inventory_evidence": # Uploader may always view their own photo (employees viewing # their own /me/inventory response). Asset-read roles # (global_admin / asset_admin / auditor / management) may view # any — they're the audience for the admin Resolve modal and # the audit trail. if current_user.id != uploader_id and not can_read_all_assets( current_user.role_names, ): raise HTTPException( status_code=status.HTTP_403_FORBIDDEN, detail="not authorised to view this object", ) elif purpose == "asset_disposal_evidence": # Disposal evidence (回收单 / 付款截图 / 租赁归还回执 / 销毁照片) is # uploaded by an asset admin on the 处置 modal and read back on the # asset detail page. Same audience as the asset record it hangs # off: asset-read roles (global_admin / asset_admin / auditor / # management). The uploader clause is kept for symmetry with # ``inventory_evidence`` — an admin whose asset role is later # revoked can still open the objects they themselves uploaded, # which the audit trail already attributes to them anyway. if current_user.id != uploader_id and not can_read_all_assets( current_user.role_names, ): raise HTTPException( status_code=status.HTTP_403_FORBIDDEN, detail="not authorised to view this object", ) elif purpose in WORKFORCE_READABLE_PURPOSES: # Deliberately "if you authenticated, you're in the company" — # see ``WORKFORCE_READABLE_PURPOSES`` for why each member is on # that list. No uploader clause: the read set is already a # superset of the uploader. pass elif purpose in AI_LABS_PURPOSES: # AI Labs inputs are the employee's own private workspace — the # image they asked a question about, the document they attached, # the recording they wanted transcribed. **Uploader only, with no # admin override**, which is deliberate and is the one clause in # this chain that no role can widen: the product tells employees # that nobody at Helix reads what they put into AI Labs, and an # admin-readable escape hatch here would make that untrue. Nothing # in the domain needs one either — its admin surfaces read # content-free counters. if current_user.id != uploader_id: raise HTTPException( status_code=status.HTTP_403_FORBIDDEN, detail="not authorised to view this object", ) else: # New purpose added to ``_ALLOWED_PURPOSES`` without a matching # ACL clause here — refuse rather than fall through to "anyone # authenticated". Forces the next contributor to make an # explicit policy decision when they extend the surface. raise HTTPException( status_code=status.HTTP_403_FORBIDDEN, detail=f"no read policy registered for purpose '{purpose}'", ) try: url = await presign_get(db, object_key=payload.object_key) except StorageNotConfiguredError as e: logger.error("storage not configured: %s", e) raise HTTPException( status_code=status.HTTP_503_SERVICE_UNAVAILABLE, detail="storage backend not configured", ) from e return success_response( data={"url": url, "expires_in": 600}, request=request, )