mirror of
https://github.com/MemeMeow-Studio/MemeMeow.git
synced 2026-10-18 13:42:36 +08:00
infstellar
2257f39300
Replace directory-scoped image APIs with flat PostgreSQL-backed storage and stable meme_id resource contracts. Remove directory endpoints, sidecar assumptions, and path-based media access while preserving task, upload, rename, delete, repair, and search workflows. Archive the image ingestion and library OpenSpec change with migration and contract coverage. 将按目录组织的图片接口替换为基于 PostgreSQL 的扁平存储和稳定 meme_id 资源契约。删除目录接口、sidecar 假设和路径媒体访问，同时保留任务、上传、重命名、删除、修复与检索流程。归档图片摄取与图片库 OpenSpec，并补充迁移和契约测试。 BREAKING CHANGE: image and context APIs now require stable meme_id values; directory-based endpoints and path media URLs are removed. 破坏性变更：图片和语境接口现在要求稳定 meme_id，目录接口与路径媒体 URL 已移除。
73 lines
3.3 KiB
Python
73 lines
3.3 KiB
Python
"""扁平图片存储标识解析器，阻止路径穿越和符号链接逃逸。"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import re
|
|
from pathlib import Path
|
|
|
|
from fastapi import HTTPException
|
|
|
|
|
|
SUPPORTED_EXTENSIONS = {".png", ".jpg", ".jpeg", ".gif"}
|
|
SAFE_NAME = re.compile(r"^[^/\\\x00-\x1f\x7f]+$")
|
|
INTERNAL_STORAGE_NAMES = frozenset({".staging", ".quarantine"})
|
|
|
|
|
|
def validate_business_storage_key(value: str) -> str:
|
|
"""校验业务图片只能使用当前 scope 根目录下的单个安全文件名。
|
|
|
|
该函数供 API、repository、StorageCoordinator 和 migration 共用；BlobStore
|
|
的内部 ``.staging``/``.quarantine`` key 不经过这里校验。
|
|
"""
|
|
if not isinstance(value, str) or not value or value in {".", ".."}:
|
|
raise ValueError("invalid_storage_key")
|
|
if not SAFE_NAME.fullmatch(value) or Path(value).name != value:
|
|
raise ValueError("invalid_storage_key")
|
|
if value in INTERNAL_STORAGE_NAMES or value.startswith(tuple(f"{name}." for name in INTERNAL_STORAGE_NAMES)):
|
|
raise ValueError("reserved_storage_key")
|
|
if value.strip(" .") != value:
|
|
raise ValueError("invalid_storage_key")
|
|
if Path(value).suffix.lower() not in SUPPORTED_EXTENSIONS:
|
|
raise ValueError("unsupported_format")
|
|
return value
|
|
|
|
|
|
class PathResolver:
|
|
"""将客户端相对标识解析到图片根目录内的真实文件。"""
|
|
|
|
def __init__(self, image_root: Path):
|
|
self.root = image_root.expanduser().resolve()
|
|
self.root.mkdir(parents=True, exist_ok=True)
|
|
|
|
def resolve_file(self, filename: str, *, must_exist: bool = True) -> Path:
|
|
"""解析图片文件并校验扩展名、根目录边界与符号链接。"""
|
|
try:
|
|
validate_business_storage_key(filename)
|
|
except ValueError as exc:
|
|
raise HTTPException(400, detail={"error": "invalid_filename", "message": "文件名非法"})
|
|
lexical = self.root / filename
|
|
if lexical.is_symlink():
|
|
raise HTTPException(403, detail={"error": "path_forbidden", "message": "符号链接文件不允许访问"})
|
|
candidate = lexical.resolve()
|
|
if self.root not in candidate.parents or candidate == self.root:
|
|
raise HTTPException(403, detail={"error": "path_forbidden", "message": "路径超出图片根目录"})
|
|
if must_exist and (not candidate.exists() or not candidate.is_file()):
|
|
raise HTTPException(404, detail={"error": "file_not_found", "message": "图片不存在"})
|
|
return candidate
|
|
|
|
def relative(self, path: Path) -> str:
|
|
"""返回统一的 POSIX 相对标识，不暴露服务器绝对路径。"""
|
|
if path == self.root or path.resolve() == self.root:
|
|
return ""
|
|
if path.is_symlink():
|
|
raise HTTPException(403, detail={"error": "path_forbidden", "message": "符号链接文件不允许访问"})
|
|
resolved = path.resolve()
|
|
if resolved == self.root or self.root not in resolved.parents:
|
|
raise HTTPException(403, detail={"error": "path_forbidden", "message": "路径超出图片根目录"})
|
|
return resolved.relative_to(self.root).as_posix()
|
|
|
|
def media_url(self, path: Path) -> str:
|
|
"""生成受控媒体接口 URL。"""
|
|
relative = self.relative(path)
|
|
return f"/media/{Path(relative).name}"
|