feat: 私有桶签名 URL 支持 + TOS 凭证配置

后端 oss.py 新增 generate_presigned_url (TOS V4 Query String Auth),
upload.py 新增 GET /upload/sign-url 端点。前端 api.ts 添加 getSignedUrl
方法,新增 useSignedUrl hook 支持自动缓存和过期刷新。

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
This commit is contained in:
Your Name
2026-02-10 14:54:16 +08:00
co-authored by Claude Opus 4.6
parent 2f24dcfd34
commit 58aed5f201
4 changed files with 250 additions and 2 deletions
+47 -2
View File
@@ -1,12 +1,12 @@
"""
文件上传 API
"""
from fastapi import APIRouter, Depends, HTTPException, status
from fastapi import APIRouter, Depends, HTTPException, Query, status
from pydantic import BaseModel
from typing import Optional
from datetime import datetime
from app.services.oss import generate_upload_policy, get_file_url
from app.services.oss import generate_upload_policy, get_file_url, generate_presigned_url
from app.config import settings
from app.models.user import User
from app.api.deps import get_current_user
@@ -123,3 +123,48 @@ async def file_uploaded(
file_size=request.file_size,
file_type=request.file_type,
)
class SignedUrlResponse(BaseModel):
"""签名 URL 响应"""
signed_url: str
expire_seconds: int
@router.get("/sign-url", response_model=SignedUrlResponse)
async def get_signed_url(
url: str = Query(..., description="文件的原始 URL 或 file_key"),
expire: int = Query(3600, ge=60, le=43200, description="有效期(秒),默认1小时,最长12小时"),
current_user: User = Depends(get_current_user),
):
"""
获取私有桶文件的预签名访问 URL
前端在展示/下载文件前调用此接口,获取带签名的临时访问链接。
支持传入完整 URL 或 file_key。
"""
from app.services.oss import parse_file_key_from_url
# 如果传入的是完整 URL,先解析出 file_key
file_key = url
if url.startswith("http"):
file_key = parse_file_key_from_url(url)
if not file_key:
raise HTTPException(
status_code=status.HTTP_400_BAD_REQUEST,
detail="无效的文件路径",
)
try:
signed_url = generate_presigned_url(file_key, expire_seconds=expire)
except ValueError as e:
raise HTTPException(
status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
detail=str(e),
)
return SignedUrlResponse(
signed_url=signed_url,
expire_seconds=expire,
)
+91
View File
@@ -139,6 +139,97 @@ def get_file_url(file_key: str) -> str:
return f"{host}/{file_key}"
def generate_presigned_url(
file_key: str,
expire_seconds: int = 3600,
) -> str:
"""
为私有桶中的文件生成预签名访问 URL (TOS V4 Query String Auth)
签名流程:
1. 构建 CanonicalRequest
2. 构建 StringToSign
3. 用派生密钥签名
4. 拼接查询参数
Args:
file_key: 文件在 TOS 中的 key
expire_seconds: URL 有效期(秒),默认 1 小时
Returns:
预签名 URL
"""
if not settings.TOS_ACCESS_KEY_ID or not settings.TOS_SECRET_ACCESS_KEY:
raise ValueError("TOS 配置未设置")
from urllib.parse import quote
region = settings.TOS_REGION
endpoint = settings.TOS_ENDPOINT or f"tos-cn-{region}.volces.com"
bucket = settings.TOS_BUCKET_NAME
host = f"{bucket}.{endpoint}"
now_utc = datetime.now(timezone.utc)
date_stamp = now_utc.strftime("%Y%m%d")
tos_date = now_utc.strftime("%Y%m%dT%H%M%SZ")
credential = f"{settings.TOS_ACCESS_KEY_ID}/{date_stamp}/{region}/tos/request"
# 对 file_key 中的路径段分别编码
encoded_key = "/".join(quote(seg, safe="") for seg in file_key.split("/"))
# 查询参数(按字母序排列)
query_params = (
f"X-Tos-Algorithm=TOS4-HMAC-SHA256"
f"&X-Tos-Credential={quote(credential, safe='')}"
f"&X-Tos-Date={tos_date}"
f"&X-Tos-Expires={expire_seconds}"
f"&X-Tos-SignedHeaders=host"
)
# CanonicalRequest
canonical_request = (
f"GET\n"
f"/{encoded_key}\n"
f"{query_params}\n"
f"host:{host}\n"
f"\n"
f"host\n"
f"UNSIGNED-PAYLOAD"
)
# StringToSign
canonical_request_hash = hashlib.sha256(canonical_request.encode()).hexdigest()
string_to_sign = (
f"TOS4-HMAC-SHA256\n"
f"{tos_date}\n"
f"{date_stamp}/{region}/tos/request\n"
f"{canonical_request_hash}"
)
# 派生签名密钥
k_date = hmac.new(
f"TOS4{settings.TOS_SECRET_ACCESS_KEY}".encode(),
date_stamp.encode(),
hashlib.sha256,
).digest()
k_region = hmac.new(k_date, region.encode(), hashlib.sha256).digest()
k_service = hmac.new(k_region, b"tos", hashlib.sha256).digest()
k_signing = hmac.new(k_service, b"request", hashlib.sha256).digest()
# 计算签名
signature = hmac.new(
k_signing,
string_to_sign.encode(),
hashlib.sha256,
).hexdigest()
return (
f"https://{host}/{encoded_key}"
f"?{query_params}"
f"&X-Tos-Signature={signature}"
)
def parse_file_key_from_url(url: str) -> str:
"""
从完整 URL 解析出文件 key