From f900be30ba7f459323aefcd11612f4179af6eeee Mon Sep 17 00:00:00 2001 From: obroccolio Date: Sat, 12 Sep 2026 16:19:41 +0800 Subject: [PATCH] fix: add authenticated product image contract --- backend/service/app.py | 94 +++++++++++++++++++++++---- backend/service/schemas.py | 16 ++++- backend/tests/test_product_archive.py | 55 ++++++++++++++++ frontend/src/types.ts | 3 + 4 files changed, 152 insertions(+), 16 deletions(-) diff --git a/backend/service/app.py b/backend/service/app.py index 23eeef0..4005765 100644 --- a/backend/service/app.py +++ b/backend/service/app.py @@ -46,10 +46,11 @@ from .schemas import ( Project, ProjectSummary, ProductDetailResponse, - ProductImageRecord, + ProductImageResponse, ProductInput, ProductRecord, ProductVersionArchiveResponse, + ProductVersionDetailResponse, ProductWordcloudArchiveResponse, Template, WordLocation, @@ -1148,6 +1149,19 @@ def _product_archive_service() -> ProductArchiveService: ) +def _product_image_response(row: sqlite3.Row) -> ProductImageResponse: + return ProductImageResponse( + image_id=row["image_id"], + product_id=row["product_id"], + version_id=row["version_id"], + image_path=row["image_path"], + image_type=row["image_type"], + is_cover=bool(row["is_cover"]), + created_at=datetime.fromisoformat(row["created_at"]), + image_url=f"/api/products/{row['product_id']}/images/{row['image_id']}", + ) + + def _version_archive_response(version_id: str) -> ProductVersionArchiveResponse: row = product_archive_store._fetchone( "SELECT * FROM product_versions WHERE version_id = ?", (version_id,) @@ -1172,32 +1186,50 @@ def _version_archive_response(version_id: str) -> ProductVersionArchiveResponse: (version_id,), ) ] + preview_image = product_archive_store._fetchone( + """SELECT * FROM product_images + WHERE product_id = ? AND version_id = ? AND image_type = 'design_preview' + ORDER BY created_at DESC, image_id DESC LIMIT 1""", + (row["product_id"], version_id), + ) + design_preview_path = ( + str(preview_image["image_path"]) + if preview_image and preview_image["image_path"] + else str(PRODUCT_ARCHIVES_DIR / row["product_id"] / version_id / "design-preview.png") + ) return ProductVersionArchiveResponse( version_id=row["version_id"], product_id=row["product_id"], version=row["version"], metadata=json.loads(row["metadata"] or "{}"), created_at=datetime.fromisoformat(row["created_at"]), - design_preview_path=str( - PRODUCT_ARCHIVES_DIR / row["product_id"] / row["version_id"] / "design-preview.png" - ), + design_preview_path=design_preview_path, wordcloud_count=len(archives), wordcloud_archives=archives, ) +def _version_detail_response(version_id: str) -> ProductVersionDetailResponse: + response = _version_archive_response(version_id) + row = product_archive_store._fetchone( + """SELECT * FROM product_images + WHERE product_id = ? AND version_id = ? AND image_type = 'design_preview' + ORDER BY created_at DESC, image_id DESC LIMIT 1""", + (response.product_id, version_id), + ) + return ProductVersionDetailResponse( + **response.model_dump(), + design_preview_image_id=row["image_id"] if row else None, + design_preview_url=( + f"/api/products/{response.product_id}/images/{row['image_id']}" if row else "" + ), + ) + + def _product_detail_response(product_id: str) -> ProductDetailResponse: product = _get_product_or_404(product_id) images = [ - ProductImageRecord( - image_id=row["image_id"], - product_id=row["product_id"], - version_id=row["version_id"], - image_path=row["image_path"], - image_type=row["image_type"], - is_cover=bool(row["is_cover"]), - created_at=datetime.fromisoformat(row["created_at"]), - ) + _product_image_response(row) for row in product_archive_store._fetchall( """SELECT * FROM product_images WHERE product_id = ? ORDER BY created_at, image_id""", @@ -1205,7 +1237,7 @@ def _product_detail_response(product_id: str) -> ProductDetailResponse: ) ] versions = [ - _version_archive_response(row["version_id"]) + _version_detail_response(row["version_id"]) for row in product_archive_store._fetchall( """SELECT version_id FROM product_versions WHERE product_id = ? ORDER BY created_at, version_id""", @@ -1215,6 +1247,40 @@ def _product_detail_response(product_id: str) -> ProductDetailResponse: return ProductDetailResponse(**product.model_dump(), images=images, versions=versions) +@app.get( + "/api/products/{product_id}/images/{image_id}", + response_class=FileResponse, +) +def get_product_image( + request: Request, + product_id: str, + image_id: str, +) -> FileResponse: + _require_orders_auth(request) + _get_product_or_404(product_id) + row = product_archive_store._fetchone( + "SELECT * FROM product_images WHERE product_id = ? AND image_id = ?", + (product_id, image_id), + ) + if row is None or not row["image_path"]: + raise HTTPException(status_code=404, detail="product image not found") + + image_path = Path(row["image_path"]).resolve() + archive_root = PRODUCT_ARCHIVES_DIR.resolve() + if not image_path.is_relative_to(archive_root) or not image_path.is_file(): + raise HTTPException(status_code=404, detail="product image not found") + + media_types = { + ".png": "image/png", + ".jpg": "image/jpeg", + ".jpeg": "image/jpeg", + ".webp": "image/webp", + ".svg": "image/svg+xml", + } + media_type = media_types.get(image_path.suffix.lower(), "application/octet-stream") + return FileResponse(image_path, media_type=media_type, filename=image_path.name) + + @app.post("/api/products", response_model=ProductRecord, status_code=201) def create_product(request: Request, product: ProductInput) -> ProductRecord: _require_orders_auth(request) diff --git a/backend/service/schemas.py b/backend/service/schemas.py index d8f8459..e16a26a 100644 --- a/backend/service/schemas.py +++ b/backend/service/schemas.py @@ -89,9 +89,21 @@ class ProductVersionArchiveResponse(ProductVersionRecord): wordcloud_archives: list[ProductWordcloudArchiveResponse] = Field(default_factory=list) +class ProductImageResponse(ProductImageRecord): + image_url: str = "" + + +class ProductVersionDetailResponse(ProductVersionRecord): + design_preview_image_id: str | None = None + design_preview_url: str = "" + design_preview_path: str = "" + wordcloud_count: int = 0 + wordcloud_archives: list[ProductWordcloudArchiveResponse] = Field(default_factory=list) + + class ProductDetailResponse(ProductRecord): - images: list[ProductImageRecord] = Field(default_factory=list) - versions: list[ProductVersionArchiveResponse] = Field(default_factory=list) + images: list[ProductImageResponse] = Field(default_factory=list) + versions: list[ProductVersionDetailResponse] = Field(default_factory=list) class JobCreateResponse(BaseModel): diff --git a/backend/tests/test_product_archive.py b/backend/tests/test_product_archive.py index 1da4d58..959c89f 100644 --- a/backend/tests/test_product_archive.py +++ b/backend/tests/test_product_archive.py @@ -332,12 +332,67 @@ def test_product_detail_contract_includes_images_and_archive_versions( assert detail["images"][0]["is_cover"] is True assert len(detail["versions"]) == 1 version = detail["versions"][0] + preview_image = detail["images"][0] + expected_image_url = f"/api/products/{product['product_id']}/images/{preview_image['image_id']}" assert version["version_id"] == archived.json()["version_id"] + assert version["design_preview_image_id"] == preview_image["image_id"] + assert version["design_preview_url"] == expected_image_url assert version["wordcloud_count"] == 1 assert version["wordcloud_archives"][0]["source_job_id"] == "job-success" assert Path(version["design_preview_path"]).name == "design-preview.png" +def test_product_detail_image_contract_is_authenticated_and_path_safe( + product_archive_client, prepared_wordcloud_job, tmp_path +): + product = create_product(product_archive_client) + archive_product_version( + product_archive_client, product["product_id"], prepared_wordcloud_job.document + ) + detail = product_archive_client.get( + f"/api/products/{product['product_id']}", headers=orders_auth_header() + ).json() + image = detail["images"][0] + + assert image["image_url"] == f"/api/products/{product['product_id']}/images/{image['image_id']}" + + served = product_archive_client.get( + f"/api/products/{product['product_id']}/images/{image['image_id']}", + headers=orders_auth_header(), + ) + assert served.status_code == 200 + assert served.headers["content-type"] == "image/png" + assert served.content == product_archive_client.archive_root.joinpath( + product["product_id"], image["version_id"], "design-preview.png" + ).read_bytes() + + outside_path = tmp_path / "outside.png" + outside_path.write_bytes(PNG_BYTES) + product_archive_client.product_store._execute( + "UPDATE product_images SET image_path = ? WHERE image_id = ?", + (str(outside_path), image["image_id"]), + ) + unsafe = product_archive_client.get( + f"/api/products/{product['product_id']}/images/{image['image_id']}", + headers=orders_auth_header(), + ) + assert unsafe.status_code == 404 + + unknown = product_archive_client.get( + f"/api/products/{product['product_id']}/images/img_unknown", + headers=orders_auth_header(), + ) + assert unknown.status_code == 404 + + unauthenticated_detail = product_archive_client.get( + f"/api/products/{product['product_id']}" + ) + unauthenticated_image = product_archive_client.get( + f"/api/products/{product['product_id']}/images/{image['image_id']}" + ) + assert unauthenticated_detail.status_code == unauthenticated_image.status_code == 403 + + def test_product_routes_require_orders_auth(product_archive_client): product = create_product(product_archive_client) requests = [ diff --git a/frontend/src/types.ts b/frontend/src/types.ts index 11abbb5..c48e092 100644 --- a/frontend/src/types.ts +++ b/frontend/src/types.ts @@ -288,6 +288,7 @@ export interface ProductImage { image_type: string; is_cover: boolean; created_at: string; + image_url?: string; } export interface ProductWordcloudArchive { @@ -309,6 +310,8 @@ export interface ProductVersion { metadata: Record; created_at: string; design_preview_path: string; + design_preview_image_id?: string | null; + design_preview_url?: string; wordcloud_count: number; wordcloud_archives: ProductWordcloudArchive[]; }