EIEI/wordcloud
5
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fix: add authenticated product image contract

Browse Source
This commit is contained in:
broccoli
2026-09-12 16:19:41 +08:00
parent fe62a7f721
commit f900be30ba
4 changed files with 152 additions and 16 deletions
Download Patch File Download Diff File Expand all files Collapse all files
backend/service/app.py
+80 -14
View File
@@ -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)
backend/service/schemas.py
+14 -2
View File
@@ -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):
backend/tests/test_product_archive.py
+55
View File
@@ -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 = [