fix(P0-2): sign_url 添加 slash_safe=True 修复签名403

根因: oss2.Bucket.sign_url 默认 slash_safe=False，key 中的 / 会被
编码为 %2F，导致 URL 路径与签名计算时的 canonicalized resource
不一致，OSS 返回 403 签名无效。

修复:
- API 端 storage.py get_download_url 加 slash_safe=True
- Worker 端 oss_helpers.py get_signed_download_url 加 slash_safe=True
- 新增 2 个单元测试验证 slash_safe 参数传递

相关: P0-2 预签名 URL 签名无效问题

apps/api/app/core/storage.py

@@ -199,7 +199,7 @@ class OSSStorageService:
 
         storage_key = self._normalize_storage_key(storage_key_or_url)
         try:
-            signed = self.bucket.sign_url("GET", storage_key, expires_seconds)
+            signed = self.bucket.sign_url("GET", storage_key, expires_seconds, slash_safe=True)
             logger.info(
                 "get_download_url: signed URL generated. storage_key=%s url_prefix=%s",
                 storage_key[:80],

apps/worker/video_processing/oss_helpers.py

@@ -40,6 +40,9 @@ def oss_settings() -> tuple[str, str, str, str] | None:
 def oss_bucket() -> oss2.Bucket | None:
     """获取 OSS Bucket 实例。
 
+    P0-2 修复：endpoint 不带 scheme 时自动补 https:// 前缀，
+    确保 sign_url 等依赖 scheme 的方法返回 HTTPS URL。
+
     Returns:
         oss2.Bucket 实例，配置缺失时返回 None。
     """
@@ -47,6 +50,9 @@ def oss_bucket() -> oss2.Bucket | None:
     if settings is None:
         return None
     access_key_id, access_key_secret, endpoint, bucket_name = settings
+    # endpoint 无 scheme 时补 https://，与 API 端 storage.py 保持一致
+    if not endpoint.startswith(("http://", "https://")):
+        endpoint = f"https://{endpoint}"
     return oss2.Bucket(oss2.Auth(access_key_id, access_key_secret), endpoint, bucket_name)
 
 
@@ -105,13 +111,37 @@ def upload_to_oss(local_path: Path, storage_key: str) -> str | None:
         settings = oss_settings()
         if settings:
             _, _, endpoint, bucket_name = settings
-            return f"https://{bucket_name}.{endpoint.replace('https://', '').replace('http://', '')}/{storage_key}"
+            endpoint_clean = endpoint.replace("https://", "").replace("http://", "")
+            return f"https://{bucket_name}.{endpoint_clean}/{storage_key}"
         return None
     except Exception:
         logger.exception("上传 OSS 失败: %s", storage_key)
         return None
 
 
+def get_signed_download_url(storage_key_or_url: str, expires_seconds: int = 3600) -> str | None:
+    """生成预签名下载 URL（用于私有 bucket 的 URL 校验或临时下载）。
+
+    Args:
+        storage_key_or_url: 存储键或完整 URL（URL 会自动提取 path）
+        expires_seconds: 签名有效期（秒）
+
+    Returns:
+        预签名 URL，失败或 OSS 未配置时返回 None。
+    """
+    bucket = oss_bucket()
+    if bucket is None:
+        return None
+    try:
+        storage_key = normalize_storage_key(storage_key_or_url)
+        signed = bucket.sign_url("GET", storage_key, expires_seconds, slash_safe=True)
+        logger.info("生成预签名URL: key=%s url_prefix=%s", storage_key[:80], signed[:60])
+        return signed
+    except Exception:
+        logger.exception("生成预签名URL失败: %s", storage_key_or_url[:80])
+        return None
+
+
 # ── Asset 解析 ────────────────────────────────────────────────────────────────
 
 

apps/worker/worker_app/tasks/generation.py

@@ -87,6 +87,7 @@ from video_processing.dedup_helpers import create_video_record_and_dedup
 from video_processing.ffmpeg_utils import FFMPEG_BIN, probe_duration, run_ffmpeg
 from video_processing.oss_helpers import (
     download_asset,
+    get_signed_download_url,
     upload_to_oss,
 )
 from video_processing.unified_render_service import UnifiedRenderService
@@ -623,9 +624,24 @@ def generate_video(self, task_id: str) -> dict:
                     f"OSS 上传失败: task_id={task_id}, storage_key={storage_key}, " f"output_path={output_path}"
                 )
 
-            # HEAD 校验 URL 可访问
-            if not _verify_url_accessible(file_url):
-                raise RuntimeError(f"OSS 上传后 URL 不可访问: file_url={file_url}, " f"storage_key={storage_key}")
+            # P0-2 修复：私有 bucket 下裸 URL 永远 403，改用预签名 URL 校验
+            # 先用预签名 URL 校验，失败则降级为检查文件是否存在（object_exists）
+            verify_url = get_signed_download_url(file_url, expires_seconds=300) or file_url
+            if not _verify_url_accessible(verify_url):
+                # 预签名 URL 也访问失败时，退一步用 object_exists 确认上传成功
+                from video_processing.oss_helpers import oss_bucket, normalize_storage_key
+
+                bucket = oss_bucket()
+                key = normalize_storage_key(file_url)
+                if bucket and bucket.object_exists(key):
+                    logger.info(
+                        "URL 校验失败但 object_exists 确认文件存在，视为上传成功: storage_key=%s", key
+                    )
+                else:
+                    raise RuntimeError(
+                        f"OSS 上传后 URL 不可访问且 object_exists 失败: file_url={file_url}, "
+                        f"storage_key={storage_key}"
+                    )
 
             logger.info("OSS 上传成功: file_url=%s", file_url)
 

tests/unit/test_p02_p03_fixes.py

@@ -284,7 +284,7 @@ class TestGetDownloadUrl:
 
             result = svc.get_download_url("uploads/video.mp4")
 
-            svc.bucket.sign_url.assert_called_once_with("GET", "uploads/video.mp4", 3600)
+            svc.bucket.sign_url.assert_called_once_with("GET", "uploads/video.mp4", 3600, slash_safe=True)
             assert "signed-url" in result
 
     def test_returns_raw_url_when_bucket_none(self):
@@ -332,6 +332,23 @@ class TestGetDownloadUrl:
             key = svc._normalize_storage_key("uploads/video.mp4")
             assert key == "uploads/video.mp4"
 
+
+    def test_sign_url_uses_slash_safe_for_path_keys(self):
+        """包含路径的 key 必须传 slash_safe=True，否则 / 被编码成 %%2F 导致签名403。"""
+        from app.core.storage import OSSStorageService
+
+        with patch.object(OSSStorageService, "__init__", lambda self: None):
+            svc = OSSStorageService()
+            svc.bucket = MagicMock()
+            svc.bucket.sign_url.return_value = "https://bucket.oss.com/generated/tasks/123/out.mp4?sig=xxx"
+            svc.public_url = "https://bucket.oss.com"
+
+            svc.get_download_url("generated/tasks/123/output.mp4")
+
+            call_args = svc.bucket.sign_url.call_args
+            assert call_args.kwargs.get("slash_safe") is True
+
+
     def test_sign_url_failure_falls_back(self):
         """sign_url 异常 → 回退到原始 URL，不崩溃。"""
         from app.core.storage import OSSStorageService

tests/unit/test_p02_worker_oss_fix.py

@@ -0,0 +1,286 @@
+"""P0-2 深度修复：Worker 端 OSS 工具函数测试.
+
+测试：
+1. oss_bucket() endpoint 自动补 https:// 前缀
+2. get_signed_download_url() 生成预签名 URL
+3. upload_to_oss() 返回 HTTPS URL
+"""
+
+from __future__ import annotations
+
+import os
+from unittest.mock import MagicMock, patch
+
+import pytest
+
+
+# ── oss_bucket endpoint scheme 修复 ──────────────────────────────────────────
+
+
+class TestOSSBucketEndpointScheme:
+    """测试 oss_bucket() 自动为 endpoint 补 https:// 前缀."""
+
+    def test_endpoint_without_scheme_adds_https(self):
+        """endpoint 不带 scheme 时，自动补 https://."""
+        from video_processing.oss_helpers import oss_bucket
+
+        mock_bucket_instance = MagicMock()
+        with patch.dict(
+            os.environ,
+            {
+                "OSS_ACCESS_KEY_ID": "test-key",
+                "OSS_ACCESS_KEY_SECRET": "test-secret",
+                "OSS_ENDPOINT": "oss-cn-hangzhou.aliyuncs.com",
+                "OSS_BUCKET_NAME": "test-bucket",
+            },
+        ), patch("video_processing.oss_helpers.oss2.Auth") as mock_auth, patch(
+            "video_processing.oss_helpers.oss2.Bucket", return_value=mock_bucket_instance
+        ) as mock_bucket_cls:
+            # 清除缓存，确保重新创建
+            import video_processing.oss_helpers as oss_mod
+
+            bucket = oss_bucket()
+
+            assert bucket is mock_bucket_instance
+            # 验证 endpoint 传的是带 https:// 的
+            call_args = mock_bucket_cls.call_args
+            endpoint_arg = call_args[0][1]  # 第 2 个位置参数是 endpoint
+            assert endpoint_arg.startswith("https://"), (
+                f"endpoint 应该带 https:// 前缀，实际为: {endpoint_arg}"
+            )
+            assert "oss-cn-hangzhou.aliyuncs.com" in endpoint_arg
+
+    def test_endpoint_with_https_keeps_as_is(self):
+        """endpoint 已有 https:// 时，不重复添加."""
+        from video_processing.oss_helpers import oss_bucket
+
+        mock_bucket_instance = MagicMock()
+        with patch.dict(
+            os.environ,
+            {
+                "OSS_ACCESS_KEY_ID": "test-key",
+                "OSS_ACCESS_KEY_SECRET": "test-secret",
+                "OSS_ENDPOINT": "https://oss-cn-hangzhou.aliyuncs.com",
+                "OSS_BUCKET_NAME": "test-bucket",
+            },
+        ), patch("video_processing.oss_helpers.oss2.Auth"), patch(
+            "video_processing.oss_helpers.oss2.Bucket", return_value=mock_bucket_instance
+        ) as mock_bucket_cls:
+            import video_processing.oss_helpers as oss_mod
+
+            bucket = oss_bucket()
+
+            call_args = mock_bucket_cls.call_args
+            endpoint_arg = call_args[0][1]
+            # 不应该出现 https://https:// 这种双重前缀
+            assert endpoint_arg.count("https://") == 1
+            assert endpoint_arg == "https://oss-cn-hangzhou.aliyuncs.com"
+
+    def test_endpoint_with_http_keeps_as_is(self):
+        """endpoint 已有 http:// 时，不修改（保留用户选择）."""
+        from video_processing.oss_helpers import oss_bucket
+
+        mock_bucket_instance = MagicMock()
+        with patch.dict(
+            os.environ,
+            {
+                "OSS_ACCESS_KEY_ID": "test-key",
+                "OSS_ACCESS_KEY_SECRET": "test-secret",
+                "OSS_ENDPOINT": "http://oss-cn-hangzhou.aliyuncs.com",
+                "OSS_BUCKET_NAME": "test-bucket",
+            },
+        ), patch("video_processing.oss_helpers.oss2.Auth"), patch(
+            "video_processing.oss_helpers.oss2.Bucket", return_value=mock_bucket_instance
+        ) as mock_bucket_cls:
+            import video_processing.oss_helpers as oss_mod
+
+            bucket = oss_bucket()
+
+            call_args = mock_bucket_cls.call_args
+            endpoint_arg = call_args[0][1]
+            assert endpoint_arg == "http://oss-cn-hangzhou.aliyuncs.com"
+
+    def test_missing_credentials_returns_none(self):
+        """凭证缺失时返回 None."""
+        from video_processing.oss_helpers import oss_bucket
+
+        with patch.dict(
+            os.environ,
+            {
+                "OSS_ACCESS_KEY_ID": "",
+                "OSS_ACCESS_KEY_SECRET": "",
+                "OSS_ENDPOINT": "oss-cn-hangzhou.aliyuncs.com",
+                "OSS_BUCKET_NAME": "test-bucket",
+            },
+            clear=True,
+        ):
+            import video_processing.oss_helpers as oss_mod
+
+            bucket = oss_bucket()
+            assert bucket is None
+
+
+# ── get_signed_download_url ──────────────────────────────────────────────────
+
+
+class TestGetSignedDownloadUrl:
+    """测试 get_signed_download_url() 预签名 URL 生成."""
+
+    def test_returns_signed_url_with_storage_key(self):
+        """传入 storage key 时，调用 sign_url 并返回结果."""
+        from video_processing.oss_helpers import get_signed_download_url
+
+        mock_bucket = MagicMock()
+        mock_bucket.sign_url.return_value = "https://test-bucket.oss-cn-hangzhou.aliyuncs.com/generated/test.mp4?OSSAccessKeyId=xxx&Expires=xxx&Signature=xxx"
+
+        with patch.dict(
+            os.environ,
+            {
+                "OSS_ACCESS_KEY_ID": "test-key",
+                "OSS_ACCESS_KEY_SECRET": "test-secret",
+                "OSS_ENDPOINT": "oss-cn-hangzhou.aliyuncs.com",
+                "OSS_BUCKET_NAME": "test-bucket",
+            },
+        ), patch("video_processing.oss_helpers.oss2.Auth"), patch(
+            "video_processing.oss_helpers.oss2.Bucket", return_value=mock_bucket
+        ):
+            result = get_signed_download_url("generated/test.mp4", expires_seconds=3600)
+
+            assert result is not None
+            assert "Signature=" in result
+            mock_bucket.sign_url.assert_called_once_with("GET", "generated/test.mp4", 3600, slash_safe=True)
+
+    def test_normalizes_full_url_to_storage_key(self):
+        """传入完整 URL 时，提取 storage key 再生成签名."""
+        from video_processing.oss_helpers import get_signed_download_url
+
+        mock_bucket = MagicMock()
+        mock_bucket.sign_url.return_value = "https://test-bucket.oss-cn-hangzhou.aliyuncs.com/generated/test.mp4?sign=xxx"
+
+        with patch.dict(
+            os.environ,
+            {
+                "OSS_ACCESS_KEY_ID": "test-key",
+                "OSS_ACCESS_KEY_SECRET": "test-secret",
+                "OSS_ENDPOINT": "oss-cn-hangzhou.aliyuncs.com",
+                "OSS_BUCKET_NAME": "test-bucket",
+            },
+        ), patch("video_processing.oss_helpers.oss2.Auth"), patch(
+            "video_processing.oss_helpers.oss2.Bucket", return_value=mock_bucket
+        ):
+            result = get_signed_download_url(
+                "https://test-bucket.oss-cn-hangzhou.aliyuncs.com/generated/test.mp4"
+            )
+
+            mock_bucket.sign_url.assert_called_once()
+            # 验证传给 sign_url 的是纯 storage key，不是完整 URL
+            call_key = mock_bucket.sign_url.call_args[0][1]
+            assert not call_key.startswith("http")
+            assert call_key == "generated/test.mp4"
+
+
+    def test_sign_url_uses_slash_safe_for_path_keys(self):
+        """包含路径的 key 必须传 slash_safe=True，否则 / 被编码成 %2F 导致签名403。"""
+        from video_processing.oss_helpers import get_signed_download_url
+
+        with patch("video_processing.oss_helpers.oss_bucket") as mock_oss_bucket:
+            mock_bucket = MagicMock()
+            mock_oss_bucket.return_value = mock_bucket
+            mock_bucket.sign_url.return_value = (
+                "https://test-bucket.oss-cn-hangzhou.aliyuncs.com/generated/tasks/123/out.mp4?sig=xxx"
+            )
+
+            result = get_signed_download_url("generated/tasks/123/output.mp4")
+
+            assert result is not None
+            call_args = mock_bucket.sign_url.call_args
+            assert call_args.kwargs.get("slash_safe") is True
+
+    def test_returns_none_when_bucket_none(self):
+        """bucket 为 None 时返回 None（不抛异常）."""
+        from video_processing.oss_helpers import get_signed_download_url
+
+        with patch.dict(os.environ, {}, clear=True):
+            result = get_signed_download_url("generated/test.mp4")
+            assert result is None
+
+    def test_sign_url_exception_returns_none(self):
+        """sign_url 抛异常时，返回 None（不向上抛出）."""
+        from video_processing.oss_helpers import get_signed_download_url
+
+        mock_bucket = MagicMock()
+        mock_bucket.sign_url.side_effect = Exception("sign failed")
+
+        with patch.dict(
+            os.environ,
+            {
+                "OSS_ACCESS_KEY_ID": "test-key",
+                "OSS_ACCESS_KEY_SECRET": "test-secret",
+                "OSS_ENDPOINT": "oss-cn-hangzhou.aliyuncs.com",
+                "OSS_BUCKET_NAME": "test-bucket",
+            },
+        ), patch("video_processing.oss_helpers.oss2.Auth"), patch(
+            "video_processing.oss_helpers.oss2.Bucket", return_value=mock_bucket
+        ):
+            result = get_signed_download_url("generated/test.mp4")
+            assert result is None
+
+
+# ── upload_to_oss 返回 HTTPS URL ─────────────────────────────────────────────
+
+
+class TestUploadToOSSReturnsHTTPS:
+    """测试 upload_to_oss() 返回的 URL 始终是 HTTPS."""
+
+    def test_endpoint_without_scheme_returns_https_url(self):
+        """endpoint 不带 scheme 时，返回 HTTPS URL."""
+        from video_processing.oss_helpers import upload_to_oss
+
+        mock_bucket = MagicMock()
+        mock_bucket.put_object_from_file = MagicMock()
+
+        from pathlib import Path
+
+        with patch.dict(
+            os.environ,
+            {
+                "OSS_ACCESS_KEY_ID": "test-key",
+                "OSS_ACCESS_KEY_SECRET": "test-secret",
+                "OSS_ENDPOINT": "oss-cn-hangzhou.aliyuncs.com",
+                "OSS_BUCKET_NAME": "test-bucket",
+            },
+        ), patch("video_processing.oss_helpers.oss2.Auth"), patch(
+            "video_processing.oss_helpers.oss2.Bucket", return_value=mock_bucket
+        ):
+            result = upload_to_oss(Path("/tmp/test.mp4"), "generated/test.mp4")
+
+            assert result is not None
+            assert result.startswith("https://")
+            assert "test-bucket.oss-cn-hangzhou.aliyuncs.com/generated/test.mp4" in result
+
+    def test_endpoint_with_https_returns_clean_url(self):
+        """endpoint 带 https:// 时，URL 里不会有双重 https."""
+        from video_processing.oss_helpers import upload_to_oss
+
+        mock_bucket = MagicMock()
+        mock_bucket.put_object_from_file = MagicMock()
+
+        from pathlib import Path
+
+        with patch.dict(
+            os.environ,
+            {
+                "OSS_ACCESS_KEY_ID": "test-key",
+                "OSS_ACCESS_KEY_SECRET": "test-secret",
+                "OSS_ENDPOINT": "https://oss-cn-hangzhou.aliyuncs.com",
+                "OSS_BUCKET_NAME": "test-bucket",
+            },
+        ), patch("video_processing.oss_helpers.oss2.Auth"), patch(
+            "video_processing.oss_helpers.oss2.Bucket", return_value=mock_bucket
+        ):
+            result = upload_to_oss(Path("/tmp/test.mp4"), "generated/test.mp4")
+
+            assert result is not None
+            assert result.startswith("https://")
+            # 不应该出现 https://https://
+            assert result.count("https://") == 1