diff --git a/apps/api/routers/uploads.py b/apps/api/routers/uploads.py index 9124203..f1d6839 100644 --- a/apps/api/routers/uploads.py +++ b/apps/api/routers/uploads.py @@ -1,10 +1,27 @@ -"""``POST /uploads/presign`` — COS 直传预签名(骨架,T5 填充实现)。 +"""``POST /uploads/presign`` — COS 直传预签名(FR-SYS-03 / T-P1-27)。 -FR-SYS-03:图片不经应用服务器;实现契约见 docs/开发任务包/T5-上传预签名.md。 +图片不经应用服务器:本接口只做场景/类型/大小校验并签发 PUT 预签名 URL, +字节流由客户端直传 COS。真实联调需在 `.env` 配 COS 四件套 +(COS_REGION / COS_SECRET_ID / COS_SECRET_KEY + 场景桶配置 +UPLOAD_SCENES_JSON);本地单测注入 FakeSigner,不依赖任何真实 COS。 """ from __future__ import annotations from fastapi import APIRouter +from apps.api.deps import CurrentUserId, StorageServiceDep +from packages.shared.response import ok +from packages.storage.schemas import PresignRequest + router = APIRouter(prefix="/uploads", tags=["uploads"]) + + +@router.post("/presign", summary="签发图片直传 COS 的 PUT 预签名 URL") +async def presign_upload( + user_id: CurrentUserId, + storage: StorageServiceDep, + request: PresignRequest, +) -> dict[str, object]: + presigned = storage.build_presign(user_id=user_id, request=request) + return ok(presigned.model_dump()) diff --git a/packages/storage/schemas.py b/packages/storage/schemas.py new file mode 100644 index 0000000..034160a --- /dev/null +++ b/packages/storage/schemas.py @@ -0,0 +1,31 @@ +"""COS 直传预签名的请求/响应契约(FR-SYS-03 / T-P1-27)。 + +放在 packages/storage 而非 apps/api/schemas:object_key 布局与场景语义属于 +存储域,报告、形象等业务域复用直传时不应该反向依赖 API 层。 +""" + +from __future__ import annotations + +from pydantic import BaseModel, Field + + +class PresignRequest(BaseModel): + """``POST /uploads/presign`` 请求体:客户端上传前的申报信息。 + + size_bytes 只是客户端自报值,仅用于服务端提前拒绝明显超限的请求; + 真正的强制约束靠预签名 URL 签入的 Content-Type 与 COS 桶侧限制兜底。 + """ + + scene: str = Field(min_length=1, max_length=64, description="上传场景,如 report_photo/avatar") + content_type: str = Field(min_length=1, max_length=128, description="MIME 类型,如 image/jpeg") + size_bytes: int = Field(ge=1, description="文件字节数") + + +class PresignResponse(BaseModel): + """签发结果:客户端用 PUT 直传 upload_url,body 与 Content-Type 必须与申报一致。""" + + upload_url: str = Field(description="带签名的 COS PUT URL,过期前有效") + object_key: str = Field(description="COS 对象键(可审计布局:scene/user/日期/uuid)") + expires_in: int = Field(description="URL 有效期(秒),与签发时配置一致") + max_bytes: int = Field(description="服务端允许的最大字节数,客户端用于前置校验") + content_type: str = Field(description="回显申报的 Content-Type,PUT 时必须携带同值") diff --git a/packages/storage/service.py b/packages/storage/service.py index 456fa44..41ec42f 100644 --- a/packages/storage/service.py +++ b/packages/storage/service.py @@ -1,11 +1,171 @@ -"""storage 域 service — 骨架(T5 任务填充实现)。""" +"""storage 域 service — COS 直传 PUT 预签名(FR-SYS-03 / T-P1-27)。 + +只做「签发」,不接收图片字节(图片经客户端直传 COS,应用服务器不中转); +上传后的内容校验属于报告链路(T-P3),不在本模块职责内。 + +COS SDK(qcloud_cos)被隔离在 :class:`CosSigner` Protocol 后面:业务逻辑与 +单测都只面向 Protocol,:class:`QcloudCosSigner` 是全仓库唯一接触 SDK 的地方, +测试注入 FakeSigner 即可零网络验证全部校验与布局规则。 +""" from __future__ import annotations -from typing import Any +from datetime import UTC, datetime +from typing import Protocol +from uuid import uuid4 + +from qcloud_cos import CosConfig, CosS3Client + +from packages.shared.config import Settings +from packages.shared.errors import AppError, ErrorCode +from packages.shared.logging import get_logger +from packages.storage.schemas import PresignRequest, PresignResponse + +logger = get_logger(__name__) + +# MIME → object_key 扩展名。新增可上传类型时,需同步 upload_allowed_types +# 默认值与错误码 UPLOAD_REJECTED 的 hint 文案。 +_CONTENT_TYPE_EXT: dict[str, str] = { + "image/jpeg": "jpg", + "image/png": "png", + "image/webp": "webp", +} + +# 场景配置里的 bucket 别名 → Settings 桶字段名。临时桶 2 小时生命周期(CR-02), +# 正式素材由异步链路搬运确认后才落长期桶,所以 V1 两个场景默认都指向 temp。 +_BUCKET_FIELDS: dict[str, str] = { + "temp": "cos_bucket_temp", + "user": "cos_bucket_user", + "kb": "cos_bucket_kb", + "report": "cos_bucket_report", +} + + +class CosSigner(Protocol): + """预签名能力的最小抽象:业务只依赖这个接口,不依赖具体云厂商。""" + + def presign_put(self, bucket: str, key: str, expires: int, content_type: str) -> str: + """返回带签名的 PUT URL;content_type 签入签名以锁定上传类型。""" + ... + + +class QcloudCosSigner: + """:class:`CosSigner` 的腾讯云 COS 实现。 + + SDK 未提供独立的 Method 枚举模块,按其文档以字符串 ``"PUT"`` 传参; + ``Expired``/``Headers`` 是 ``get_presigned_url`` 的真实参数名。 + """ + + def __init__(self, *, region: str, secret_id: str, secret_key: str) -> None: + # SDK 在密钥为空时抛的是裸 CosClientError(会以未处理异常变成 500 Internal), + # 提前映射为 CONFIG_ERROR,让本地未配 COS 四件套时得到结构化的配置错误。 + if not secret_id or not secret_key: + raise AppError(ErrorCode.CONFIG_ERROR, message="COS 密钥未配置(COS_SECRET_ID/KEY)") + self._client = CosS3Client( + CosConfig(Region=region, SecretId=secret_id, SecretKey=secret_key) + ) + + def presign_put(self, bucket: str, key: str, expires: int, content_type: str) -> str: + # 签名是本地计算(不发网络请求);Content-Type 签入后, + # 客户端用其它类型 PUT 该 URL 会被 COS 直接拒绝。 + # SDK 无类型标注,str() 只为向 mypy 收窄返回类型。 + return str( + self._client.get_presigned_url( + Method="PUT", + Bucket=bucket, + Key=key, + Expired=expires, + Headers={"Content-Type": content_type}, + ) + ) + + +def _parse_allowed_types(raw: str) -> frozenset[str]: + """解析逗号分隔的 MIME 白名单,容忍空串与多余空格(配置是人工维护的)。""" + return frozenset(part.strip() for part in raw.split(",") if part.strip()) class StorageService: - def __init__(self, settings: Any, signer: Any | None = None) -> None: + """签发 COS 直传预签名 URL,并守住场景/类型/大小三重服务端校验。""" + + def __init__(self, settings: Settings, signer: CosSigner | None = None) -> None: self._settings = settings self._signer = signer + + @property + def _active_signer(self) -> CosSigner: + # 惰性构造:只有真实签发请求才创建 COS client,进程启动/测试路径零开销; + # 测试必须注入 FakeSigner,避免任何 SDK 构造。 + if self._signer is None: + self._signer = QcloudCosSigner( + region=self._settings.cos_region, + secret_id=self._settings.cos_secret_id, + secret_key=self._settings.cos_secret_key, + ) + return self._signer + + def build_presign( + self, + user_id: int, + request: PresignRequest, + *, + now: datetime | None = None, + ) -> PresignResponse: + """三重校验通过后签发 PUT 预签名 URL。 + + ``now`` 仅用于测试注入固定时钟;生产调用不传,按当前 UTC 时间分桶。 + """ + scene_config = self._settings.upload_scenes.get(request.scene) + if scene_config is None: + raise AppError(ErrorCode.UPLOAD_REJECTED, message="不支持的上传场景") + + # 类型校验:白名单之外一律拒绝(与大小、场景共用错误码 40010)。 + if request.content_type not in _parse_allowed_types(self._settings.upload_allowed_types): + raise AppError(ErrorCode.UPLOAD_REJECTED) + # 扩展名映射兜底:白名单被运维改成映射表之外的类型时按客户端错误拒绝, + # 而不是在拼 object_key 时抛 KeyError 变成 5xx。 + ext = _CONTENT_TYPE_EXT.get(request.content_type) + if ext is None: + raise AppError(ErrorCode.UPLOAD_REJECTED) + + # 大小校验:客户端自报值超限直接拒绝,省一次无谓的签名计算。 + if request.size_bytes > self._settings.upload_max_bytes: + raise AppError(ErrorCode.UPLOAD_REJECTED) + + # bucket 别名 → 四桶配置。别名写错属于部署配置问题,报 CONFIG_ERROR + # 而不是让客户端背 4xx。 + bucket_alias = scene_config.get("bucket") + bucket_field = _BUCKET_FIELDS.get(bucket_alias) if isinstance(bucket_alias, str) else None + if bucket_field is None: + raise AppError( + ErrorCode.CONFIG_ERROR, + message=f"上传场景 {request.scene} 的 bucket 配置无效", + ) + bucket: str = getattr(self._settings, bucket_field) + + # object_key 布局写死为 {scene}/{user_id}/{yyyyMMdd}/{uuid4hex}.{ext}: + # 按场景分目录、按用户隔离、按日分组、uuid 防猜——这是 COS 侧做 + # 前缀审计与生命周期规则(如临时桶 2h 清理)的基础,改动需过安全评审。 + # 日期取 UTC,保证多实例写入的分组一致。 + day = (now or datetime.now(UTC)).strftime("%Y%m%d") + object_key = f"{request.scene}/{user_id}/{day}/{uuid4().hex}.{ext}" + + expires = self._settings.cos_presign_expire_seconds + url = self._active_signer.presign_put( + bucket=bucket, key=object_key, expires=expires, content_type=request.content_type + ) + logger.info( + "presign_issued", + scene=request.scene, + user_id=user_id, + bucket=bucket, + object_key=object_key, + size_bytes=request.size_bytes, + ) + return PresignResponse( + upload_url=url, + object_key=object_key, + expires_in=expires, + max_bytes=self._settings.upload_max_bytes, + content_type=request.content_type, + ) diff --git a/tests/unit/test_upload_presign.py b/tests/unit/test_upload_presign.py new file mode 100644 index 0000000..de772ed --- /dev/null +++ b/tests/unit/test_upload_presign.py @@ -0,0 +1,228 @@ +"""COS 直传预签名(FR-SYS-03 / T-P1-27)。 + +全部用 FakeSigner 注入,零网络、零 SDK client 构造。object_key 布局 +(场景/用户/日期/uuid)是 COS 侧前缀审计与生命周期规则的基础,逐段精确 +断言;场景/类型/大小三重校验共用 40010,逐条锁死。 +""" + +from __future__ import annotations + +from dataclasses import dataclass, field +from datetime import UTC, datetime +from typing import Any + +import pytest +from pydantic import ValidationError + +from packages.shared.config import Settings +from packages.shared.errors import AppError, ErrorCode +from packages.storage.schemas import PresignRequest +from packages.storage.service import StorageService + +FIXED_NOW = datetime(2026, 8, 14, 1, 2, 3, tzinfo=UTC) + + +@dataclass(frozen=True, slots=True) +class FakeSigner: + """捕获签发参数并回放可预测 URL,替代 QcloudCosSigner(零网络)。""" + + calls: list[dict[str, Any]] = field(default_factory=list) + + def presign_put(self, bucket: str, key: str, expires: int, content_type: str) -> str: + self.calls.append( + {"bucket": bucket, "key": key, "expires": expires, "content_type": content_type} + ) + return f"https://cos.fake/{bucket}/{key}?sign=stub" + + +def make_settings(**overrides: Any) -> Settings: + """显式 kwargs 构造,测试不受 .env 与宿主环境变量影响。 + + upload_scenes 定义了 alias 且未开启 populate_by_name,必须按 + ``UPLOAD_SCENES_JSON`` 传入——按字段名传会被静默忽略。 + """ + options: dict[str, Any] = { + "_env_file": None, + "cos_bucket_temp": "meida-temp-test", + "cos_bucket_user": "meida-user-test", + "cos_bucket_kb": "meida-kb-test", + "cos_bucket_report": "meida-report-test", + "cos_presign_expire_seconds": 321, + "upload_max_bytes": 1024, + "upload_allowed_types": "image/jpeg,image/png,image/webp", + "UPLOAD_SCENES_JSON": {"report_photo": {"bucket": "temp"}, "avatar": {"bucket": "temp"}}, + } + options.update(overrides) + return Settings(**options) + + +def make_service(**overrides: Any) -> tuple[StorageService, FakeSigner]: + signer = FakeSigner() + return StorageService(make_settings(**overrides), signer), signer + + +def test_valid_request_builds_auditable_key_layout() -> None: + service, signer = make_service() + response = service.build_presign( + 7, + PresignRequest(scene="report_photo", content_type="image/jpeg", size_bytes=100), + now=FIXED_NOW, + ) + + # key 布局:{scene}/{user_id}/{yyyyMMdd}/{uuid4hex}.{ext}(写死,改动需过安全评审) + prefix = "report_photo/7/20260814/" + assert response.object_key.startswith(prefix) + filename = response.object_key.removeprefix(prefix) + stem, ext = filename.split(".") + assert ext == "jpg" + assert len(stem) == 32 + assert all(char in "0123456789abcdef" for char in stem) + + # 回显与签发参数 + assert ( + response.upload_url == f"https://cos.fake/meida-temp-test/{response.object_key}?sign=stub" + ) + assert response.expires_in == 321 + assert response.max_bytes == 1024 + assert response.content_type == "image/jpeg" + assert signer.calls == [ + { + "bucket": "meida-temp-test", + "key": response.object_key, + "expires": 321, + "content_type": "image/jpeg", + } + ] + + +def test_two_presigns_for_the_same_request_never_share_a_key() -> None: + service, _ = make_service() + request = PresignRequest(scene="avatar", content_type="image/jpeg", size_bytes=1) + first = service.build_presign(1, request, now=FIXED_NOW) + second = service.build_presign(1, request, now=FIXED_NOW) + assert first.object_key != second.object_key # uuid 防猜 + + +@pytest.mark.parametrize( + ("body", "message"), + [ + ( + PresignRequest(scene="evil", content_type="image/jpeg", size_bytes=1), + "不支持的上传场景", + ), + (PresignRequest(scene="report_photo", content_type="application/pdf", size_bytes=1), None), + (PresignRequest(scene="report_photo", content_type="image/jpeg", size_bytes=1025), None), + ], +) +def test_scene_type_size_rejections_share_error_code_40010( + body: PresignRequest, message: str | None +) -> None: + service, signer = make_service() + with pytest.raises(AppError) as excinfo: + service.build_presign(1, body, now=FIXED_NOW) + assert excinfo.value.code is ErrorCode.UPLOAD_REJECTED + if message is not None: + assert excinfo.value.message == message + assert signer.calls == [] # 拒绝路径绝不触发签名 + + +def test_size_equal_to_limit_is_allowed() -> None: + service, signer = make_service() + service.build_presign( + 1, + PresignRequest(scene="avatar", content_type="image/webp", size_bytes=1024), + now=FIXED_NOW, + ) + assert signer.calls[0]["key"].endswith(".webp") + + +@pytest.mark.parametrize( + ("alias", "expected_bucket"), + [ + ("temp", "meida-temp-test"), + ("user", "meida-user-test"), + ("kb", "meida-kb-test"), + ("report", "meida-report-test"), + ], +) +def test_bucket_alias_maps_to_the_matching_settings_field(alias: str, expected_bucket: str) -> None: + service, signer = make_service(UPLOAD_SCENES_JSON={"report_photo": {"bucket": alias}}) + service.build_presign( + 7, + PresignRequest(scene="report_photo", content_type="image/png", size_bytes=1), + now=FIXED_NOW, + ) + assert signer.calls[0]["bucket"] == expected_bucket + + +def test_allowed_types_string_tolerates_whitespace() -> None: + service, _ = make_service(upload_allowed_types=" image/jpeg , image/png ") + response = service.build_presign( + 1, + PresignRequest(scene="avatar", content_type="image/png", size_bytes=1), + now=FIXED_NOW, + ) + assert response.object_key.endswith(".png") + with pytest.raises(AppError) as excinfo: + service.build_presign( + 1, + PresignRequest(scene="avatar", content_type="image/webp", size_bytes=1), + now=FIXED_NOW, + ) + assert excinfo.value.code is ErrorCode.UPLOAD_REJECTED + + +def test_empty_allowed_types_rejects_everything() -> None: + service, signer = make_service(upload_allowed_types="") + with pytest.raises(AppError) as excinfo: + service.build_presign( + 1, + PresignRequest(scene="avatar", content_type="image/jpeg", size_bytes=1), + now=FIXED_NOW, + ) + assert excinfo.value.code is ErrorCode.UPLOAD_REJECTED + assert signer.calls == [] + + +def test_whitelisted_type_without_extension_mapping_is_rejected() -> None: + # 白名单被扩到扩展名映射表之外的类型(此处模拟 image/gif)时, + # 宁可按 40010 拒绝,也不能在拼 object_key 时抛 KeyError 变成 5xx。 + service, signer = make_service(upload_allowed_types="image/gif") + with pytest.raises(AppError) as excinfo: + service.build_presign( + 1, + PresignRequest(scene="avatar", content_type="image/gif", size_bytes=1), + now=FIXED_NOW, + ) + assert excinfo.value.code is ErrorCode.UPLOAD_REJECTED + assert signer.calls == [] + + +def test_unknown_bucket_alias_is_a_config_error_not_a_client_error() -> None: + service, signer = make_service(UPLOAD_SCENES_JSON={"avatar": {"bucket": "ghost"}}) + with pytest.raises(AppError) as excinfo: + service.build_presign( + 1, + PresignRequest(scene="avatar", content_type="image/jpeg", size_bytes=1), + now=FIXED_NOW, + ) + assert excinfo.value.code is ErrorCode.CONFIG_ERROR + assert signer.calls == [] + + +def test_missing_cos_credentials_is_a_config_error_before_sdk_construction() -> None: + # 未注入 signer 且密钥为空:守卫须在 SDK 构造前抛出(零 SDK 接触、零网络), + # 本地未配 COS 四件套时得到结构化 CONFIG_ERROR 而非未处理 500。 + service = StorageService(make_settings(), None) + with pytest.raises(AppError) as excinfo: + service.build_presign( + 1, + PresignRequest(scene="avatar", content_type="image/jpeg", size_bytes=1), + now=FIXED_NOW, + ) + assert excinfo.value.code is ErrorCode.CONFIG_ERROR + + +def test_request_schema_rejects_non_positive_size() -> None: + with pytest.raises(ValidationError): + PresignRequest(scene="avatar", content_type="image/jpeg", size_bytes=0)