Spaces:

basyx
/

MediaRouter

Running

Harden workspace asset metadata path checks

c4a7267 verified 10 days ago

8.74 kB

from __future__ import annotations

import asyncio
import json
import re
from datetime import datetime, timezone
from pathlib import Path
from typing import Any
from uuid import UUID

from fastapi import APIRouter, HTTPException, Query, Request, Response, status
from pydantic import BaseModel, ConfigDict, Field, model_validator
from sqlalchemy import func, or_, select
from sqlalchemy.exc import IntegrityError

from app.core.exceptions import NotFoundError
from app.generation.models import GenerationJob, GenerationRequest
from app.projects.models import Project, ProjectRenderJob
from app.security.assets import CanonicalAssetNotFoundError
from app.security.models import CanonicalMediaAsset, CanonicalMediaVariant

router = APIRouter(prefix="/v1/assets", tags=["assets"])


class AssetMetadataUpdate(BaseModel):
    model_config = ConfigDict(extra="forbid")
    metadata: dict[str, Any] = Field(default_factory=dict)

    @model_validator(mode="after")
    def metadata_is_bounded_json(self) -> "AssetMetadataUpdate":
        try:
            encoded = json.dumps(self.metadata, ensure_ascii=False, allow_nan=False)
        except (TypeError, ValueError) as exc:
            raise ValueError("metadata must contain JSON-compatible values") from exc
        if len(encoded.encode("utf-8")) > 16_384:
            raise ValueError("metadata must not exceed 16 KiB")
        return self


def _identity(request: Request) -> tuple[str, str]:
    context = getattr(request.state, "auth", None)
    if context is None or not context.workspace_id or not context.user_id:
        raise HTTPException(status_code=403, detail="An active workspace membership is required.")
    return context.workspace_id, context.user_id


def _view(record: CanonicalMediaAsset, output_dir: Path) -> dict[str, Any]:
    output_root = output_dir.resolve()
    root = (output_root / record.request_id).resolve()
    path = (root / record.filename).resolve()
    safe = (
        bool(re.fullmatch(r"[0-9a-fA-F-]{16,64}", record.request_id))
        and bool(re.fullmatch(r"[A-Za-z0-9][A-Za-z0-9._-]{0,254}", record.filename))
        and root.parent == output_root
        and path.parent == root
    )
    try:
        available = safe and path.is_file() and path.stat().st_size == record.file_size
    except OSError:
        available = False
    return {
        "id": record.id,
        "workspace_id": record.workspace_id,
        "project_id": record.project_id,
        "request_id": record.request_id,
        "filename": record.filename,
        "mime_type": record.mime_type,
        "file_size": record.file_size,
        "metadata": record.metadata_json or {},
        "file_available": available,
        "created_at": record.created_at,
        "updated_at": record.updated_at,
    }


@router.get("", response_model=dict[str, Any])
async def list_assets(
    request: Request,
    project_id: UUID | None = Query(default=None),
    q: str | None = Query(default=None, min_length=1, max_length=200),
    offset: int = Query(default=0, ge=0),
    limit: int = Query(default=100, ge=1, le=500),
) -> dict[str, Any]:
    workspace_id, user_id = _identity(request)
    container = request.app.state.container
    filters = [CanonicalMediaAsset.workspace_id == workspace_id]
    if project_id is not None:
        filters.append(CanonicalMediaAsset.project_id == str(project_id))
    if q:
        term = "%" + q.strip().replace("%", "\\%").replace("_", "\\_") + "%"
        filters.append(or_(
            CanonicalMediaAsset.filename.ilike(term, escape="\\"),
            CanonicalMediaAsset.mime_type.ilike(term, escape="\\"),
        ))
    async with container.security_database.tenant_session(
        workspace_id=workspace_id, user_id=user_id
    ) as session:
        total = await session.scalar(
            select(func.count()).select_from(CanonicalMediaAsset).where(*filters)
        )
        records = (await session.scalars(
            select(CanonicalMediaAsset)
            .where(*filters)
            .order_by(CanonicalMediaAsset.created_at.desc(), CanonicalMediaAsset.id)
            .offset(offset)
            .limit(limit)
        )).all()
        return {
            "items": [_view(record, container.settings.output_dir) for record in records],
            "total": int(total or 0),
            "offset": offset,
            "limit": limit,
        }


@router.get("/{asset_id}", response_model=dict[str, Any])
async def get_asset(request: Request, asset_id: UUID) -> dict[str, Any]:
    workspace_id, user_id = _identity(request)
    container = request.app.state.container
    try:
        record = await container.assets.get_owned_by_id(
            workspace_id=workspace_id, user_id=user_id, asset_id=str(asset_id)
        )
        path = container.cleanup.resolve_download(record.request_id, record.filename)
        await container.assets.verify_file(record, path)
    except (CanonicalAssetNotFoundError, NotFoundError) as exc:
        raise HTTPException(status_code=404, detail="Asset was not found or failed integrity verification.") from exc
    return _view(record, container.settings.output_dir)


@router.patch("/{asset_id}", response_model=dict[str, Any])
async def update_asset_metadata(
    request: Request, asset_id: UUID, payload: AssetMetadataUpdate
) -> dict[str, Any]:
    workspace_id, user_id = _identity(request)
    container = request.app.state.container
    async with container.security_database.tenant_session(
        workspace_id=workspace_id, user_id=user_id
    ) as session:
        record = await session.scalar(select(CanonicalMediaAsset).where(
            CanonicalMediaAsset.id == str(asset_id),
            CanonicalMediaAsset.workspace_id == workspace_id,
        ))
        if record is None:
            raise HTTPException(status_code=404, detail="Asset was not found in this workspace.")
        record.metadata_json = payload.metadata
        record.updated_at = datetime.now(timezone.utc)
        await session.commit()
        await session.refresh(record)
        return _view(record, container.settings.output_dir)


@router.delete("/{asset_id}", status_code=status.HTTP_204_NO_CONTENT)
async def delete_asset(request: Request, asset_id: UUID) -> Response:
    workspace_id, user_id = _identity(request)
    container = request.app.state.container
    file_path: Path | None = None
    async with container.security_database.tenant_session(
        workspace_id=workspace_id, user_id=user_id
    ) as session:
        record = await session.scalar(
            select(CanonicalMediaAsset)
            .where(
                CanonicalMediaAsset.id == str(asset_id),
                CanonicalMediaAsset.workspace_id == workspace_id,
            )
            .with_for_update()
        )
        if record is None:
            raise HTTPException(status_code=404, detail="Asset was not found in this workspace.")
        if record.project_id is not None:
            raise HTTPException(status_code=409, detail="Detach this asset from its project before deleting it.")
        references = (
            select(GenerationRequest.id).where(GenerationRequest.input_asset_id == record.id).limit(1),
            select(GenerationJob.id).where(GenerationJob.output_asset_id == record.id).limit(1),
            select(Project.thumbnail_asset_id).where(Project.thumbnail_asset_id == record.id).limit(1),
            select(ProjectRenderJob.id).where(ProjectRenderJob.output_asset_id == record.id).limit(1),
            select(CanonicalMediaVariant.id).where(CanonicalMediaVariant.asset_id == record.id).limit(1),
        )
        for statement in references:
            if await session.scalar(statement) is not None:
                raise HTTPException(status_code=409, detail="This asset is referenced by a project or job and cannot be deleted.")
        try:
            file_path = container.cleanup.resolve_download(record.request_id, record.filename)
            await container.assets.verify_file(record, file_path)
        except (CanonicalAssetNotFoundError, NotFoundError):
            file_path = None
        try:
            await session.delete(record)
            await session.commit()
        except IntegrityError as exc:
            await session.rollback()
            raise HTTPException(status_code=409, detail="This asset is still referenced and cannot be deleted.") from exc
    if file_path is not None:
        try:
            await asyncio.to_thread(file_path.unlink, missing_ok=True)
        except OSError:
            # The database record is already gone; the periodic output cleanup
            # reconciles any orphaned file without exposing its filesystem path.
            pass
    return Response(status_code=status.HTTP_204_NO_CONTENT)

	from __future__ import annotations

	import asyncio
	import json
	import re
	from datetime import datetime, timezone
	from pathlib import Path
	from typing import Any
	from uuid import UUID

	from fastapi import APIRouter, HTTPException, Query, Request, Response, status
	from pydantic import BaseModel, ConfigDict, Field, model_validator
	from sqlalchemy import func, or_, select
	from sqlalchemy.exc import IntegrityError

	from app.core.exceptions import NotFoundError
	from app.generation.models import GenerationJob, GenerationRequest
	from app.projects.models import Project, ProjectRenderJob
	from app.security.assets import CanonicalAssetNotFoundError
	from app.security.models import CanonicalMediaAsset, CanonicalMediaVariant

	router = APIRouter(prefix="/v1/assets", tags=["assets"])


	class AssetMetadataUpdate(BaseModel):
	model_config = ConfigDict(extra="forbid")
	metadata: dict[str, Any] = Field(default_factory=dict)

	@model_validator(mode="after")
	def metadata_is_bounded_json(self) -> "AssetMetadataUpdate":
	try:
	encoded = json.dumps(self.metadata, ensure_ascii=False, allow_nan=False)
	except (TypeError, ValueError) as exc:
	raise ValueError("metadata must contain JSON-compatible values") from exc
	if len(encoded.encode("utf-8")) > 16_384:
	raise ValueError("metadata must not exceed 16 KiB")
	return self


	def _identity(request: Request) -> tuple[str, str]:
	context = getattr(request.state, "auth", None)
	if context is None or not context.workspace_id or not context.user_id:
	raise HTTPException(status_code=403, detail="An active workspace membership is required.")
	return context.workspace_id, context.user_id


	def _view(record: CanonicalMediaAsset, output_dir: Path) -> dict[str, Any]:
	output_root = output_dir.resolve()
	root = (output_root / record.request_id).resolve()
	path = (root / record.filename).resolve()
	safe = (
	bool(re.fullmatch(r"[0-9a-fA-F-]{16,64}", record.request_id))
	and bool(re.fullmatch(r"[A-Za-z0-9][A-Za-z0-9._-]{0,254}", record.filename))
	and root.parent == output_root
	and path.parent == root
	)
	try:
	available = safe and path.is_file() and path.stat().st_size == record.file_size
	except OSError:
	available = False
	return {
	"id": record.id,
	"workspace_id": record.workspace_id,
	"project_id": record.project_id,
	"request_id": record.request_id,
	"filename": record.filename,
	"mime_type": record.mime_type,
	"file_size": record.file_size,
	"metadata": record.metadata_json or {},
	"file_available": available,
	"created_at": record.created_at,
	"updated_at": record.updated_at,
	}


	@router.get("", response_model=dict[str, Any])
	async def list_assets(
	request: Request,
	project_id: UUID \| None = Query(default=None),
	q: str \| None = Query(default=None, min_length=1, max_length=200),
	offset: int = Query(default=0, ge=0),
	limit: int = Query(default=100, ge=1, le=500),
	) -> dict[str, Any]:
	workspace_id, user_id = _identity(request)
	container = request.app.state.container
	filters = [CanonicalMediaAsset.workspace_id == workspace_id]
	if project_id is not None:
	filters.append(CanonicalMediaAsset.project_id == str(project_id))
	if q:
	term = "%" + q.strip().replace("%", "\\%").replace("_", "\\_") + "%"
	filters.append(or_(
	CanonicalMediaAsset.filename.ilike(term, escape="\\"),
	CanonicalMediaAsset.mime_type.ilike(term, escape="\\"),
	))
	async with container.security_database.tenant_session(
	workspace_id=workspace_id, user_id=user_id
	) as session:
	total = await session.scalar(
	select(func.count()).select_from(CanonicalMediaAsset).where(*filters)
	)
	records = (await session.scalars(
	select(CanonicalMediaAsset)
	.where(*filters)
	.order_by(CanonicalMediaAsset.created_at.desc(), CanonicalMediaAsset.id)
	.offset(offset)
	.limit(limit)
	)).all()
	return {
	"items": [_view(record, container.settings.output_dir) for record in records],
	"total": int(total or 0),
	"offset": offset,
	"limit": limit,
	}


	@router.get("/{asset_id}", response_model=dict[str, Any])
	async def get_asset(request: Request, asset_id: UUID) -> dict[str, Any]:
	workspace_id, user_id = _identity(request)
	container = request.app.state.container
	try:
	record = await container.assets.get_owned_by_id(
	workspace_id=workspace_id, user_id=user_id, asset_id=str(asset_id)
	)
	path = container.cleanup.resolve_download(record.request_id, record.filename)
	await container.assets.verify_file(record, path)
	except (CanonicalAssetNotFoundError, NotFoundError) as exc:
	raise HTTPException(status_code=404, detail="Asset was not found or failed integrity verification.") from exc
	return _view(record, container.settings.output_dir)


	@router.patch("/{asset_id}", response_model=dict[str, Any])
	async def update_asset_metadata(
	request: Request, asset_id: UUID, payload: AssetMetadataUpdate
	) -> dict[str, Any]:
	workspace_id, user_id = _identity(request)
	container = request.app.state.container
	async with container.security_database.tenant_session(
	workspace_id=workspace_id, user_id=user_id
	) as session:
	record = await session.scalar(select(CanonicalMediaAsset).where(
	CanonicalMediaAsset.id == str(asset_id),
	CanonicalMediaAsset.workspace_id == workspace_id,
	))
	if record is None:
	raise HTTPException(status_code=404, detail="Asset was not found in this workspace.")
	record.metadata_json = payload.metadata
	record.updated_at = datetime.now(timezone.utc)
	await session.commit()
	await session.refresh(record)
	return _view(record, container.settings.output_dir)


	@router.delete("/{asset_id}", status_code=status.HTTP_204_NO_CONTENT)
	async def delete_asset(request: Request, asset_id: UUID) -> Response:
	workspace_id, user_id = _identity(request)
	container = request.app.state.container
	file_path: Path \| None = None
	async with container.security_database.tenant_session(
	workspace_id=workspace_id, user_id=user_id
	) as session:
	record = await session.scalar(
	select(CanonicalMediaAsset)
	.where(
	CanonicalMediaAsset.id == str(asset_id),
	CanonicalMediaAsset.workspace_id == workspace_id,
	)
	.with_for_update()
	)
	if record is None:
	raise HTTPException(status_code=404, detail="Asset was not found in this workspace.")
	if record.project_id is not None:
	raise HTTPException(status_code=409, detail="Detach this asset from its project before deleting it.")
	references = (
	select(GenerationRequest.id).where(GenerationRequest.input_asset_id == record.id).limit(1),
	select(GenerationJob.id).where(GenerationJob.output_asset_id == record.id).limit(1),
	select(Project.thumbnail_asset_id).where(Project.thumbnail_asset_id == record.id).limit(1),
	select(ProjectRenderJob.id).where(ProjectRenderJob.output_asset_id == record.id).limit(1),
	select(CanonicalMediaVariant.id).where(CanonicalMediaVariant.asset_id == record.id).limit(1),
	)
	for statement in references:
	if await session.scalar(statement) is not None:
	raise HTTPException(status_code=409, detail="This asset is referenced by a project or job and cannot be deleted.")
	try:
	file_path = container.cleanup.resolve_download(record.request_id, record.filename)
	await container.assets.verify_file(record, file_path)
	except (CanonicalAssetNotFoundError, NotFoundError):
	file_path = None
	try:
	await session.delete(record)
	await session.commit()
	except IntegrityError as exc:
	await session.rollback()
	raise HTTPException(status_code=409, detail="This asset is still referenced and cannot be deleted.") from exc
	if file_path is not None:
	try:
	await asyncio.to_thread(file_path.unlink, missing_ok=True)
	except OSError:
	# The database record is already gone; the periodic output cleanup
	# reconciles any orphaned file without exposing its filesystem path.
	pass
	return Response(status_code=status.HTTP_204_NO_CONTENT)