diff --git a/apps/api/app/config.py b/apps/api/app/config.py index da2b61152..81f35b11c 100755 --- a/apps/api/app/config.py +++ b/apps/api/app/config.py @@ -79,6 +79,27 @@ class Settings(BaseSettings): OSS_ACCESS_KEY_ID: str = "" OSS_ACCESS_KEY_SECRET: str = "" OSS_BUCKET_NAME: str = "xiaoxia-autocut" + + @field_validator("OSS_ACCESS_KEY_ID", mode="before") + @classmethod + def validate_oss_access_key_id(cls, v): + if (v is None or v == "") and os.getenv("APP_ENV", "development") != "development": + raise ValueError( + "OSS_ACCESS_KEY_ID must be set via environment variable in non-development environments. " + "Check the server .env file (e.g. /var/lib/xiaoxia-saas-staging/.env)." + ) + return v or "" + + @field_validator("OSS_ACCESS_KEY_SECRET", mode="before") + @classmethod + def validate_oss_access_key_secret(cls, v): + if (v is None or v == "") and os.getenv("APP_ENV", "development") != "development": + raise ValueError( + "OSS_ACCESS_KEY_SECRET must be set via environment variable in non-development environments. " + "Check the server .env file (e.g. /var/lib/xiaoxia-saas-staging/.env)." + ) + return v or "" + OSS_DIRECT_UPLOAD_MAX_MB: int = Field( default=2000, validation_alias=AliasChoices("OSS_DIRECT_UPLOAD_MAX_MB", "MAX_UPLOAD_SIZE_MB"), diff --git a/apps/api/app/core/storage.py b/apps/api/app/core/storage.py index 566a5d3c7..df064c86d 100644 --- a/apps/api/app/core/storage.py +++ b/apps/api/app/core/storage.py @@ -34,13 +34,18 @@ class OSSStorageService: if has_key_id and has_key_secret: if oss2 is not None: try: + # P0-2 修复：oss2.Bucket 的 endpoint 必须带 https:// 前缀， + # 否则 sign_url 默认生成 HTTP URL。 + bucket_endpoint = settings.OSS_ENDPOINT + if not bucket_endpoint.startswith(("http://", "https://")): + bucket_endpoint = f"https://{bucket_endpoint}" auth = oss2.Auth( settings.OSS_ACCESS_KEY_ID, settings.OSS_ACCESS_KEY_SECRET, ) self.bucket = oss2.Bucket( auth, - settings.OSS_ENDPOINT, + bucket_endpoint, settings.OSS_BUCKET_NAME, ) logger.info( @@ -64,6 +69,26 @@ class OSSStorageService: self.access_key_secret = settings.OSS_ACCESS_KEY_SECRET self.endpoint = settings.OSS_ENDPOINT + def diagnose(self) -> None: + """启动诊断：输出 OSS 配置状态，帮助排查预签名 URL 问题。""" + key_id_display = ( + f"{self.access_key_id[:4]}...{self.access_key_id[-4:]}" if len(self.access_key_id) > 8 else "(empty)" + ) + logger.info( + "[OSS诊断] endpoint=%s bucket_name=%s access_key_id=%s", + self.endpoint, + self.bucket_name, + key_id_display, + ) + if self.bucket is None: + logger.error( + "[OSS诊断] ❌ bucket=None — 预签名URL不可用！" + "原因: OSS_ACCESS_KEY_ID/OSS_ACCESS_KEY_SECRET 未配置或 oss2 未安装。" + "请检查服务器 .env 文件（如 /var/lib/xiaoxia-saas-staging/.env）" + ) + else: + logger.info("[OSS诊断] ✅ bucket 已配置，预签名URL可用") + def _is_local_generated_url(self, storage_key_or_url: str) -> bool: parsed = urlparse(storage_key_or_url) path = parsed.path if parsed.scheme else storage_key_or_url @@ -167,8 +192,7 @@ class OSSStorageService: if self._is_local_generated_url(storage_key_or_url): return storage_key_or_url logger.warning( - "get_download_url: OSS bucket not configured, returning raw URL. " - "storage_key_or_url=%s", + "get_download_url: OSS bucket not configured, returning raw URL. " "storage_key_or_url=%s", storage_key_or_url[:200], ) return self.get_url(self._normalize_storage_key(storage_key_or_url)) @@ -184,8 +208,7 @@ class OSSStorageService: return signed except Exception: logger.exception( - "get_download_url: sign_url failed, falling back to raw URL. " - "storage_key=%s", + "get_download_url: sign_url failed, falling back to raw URL. " "storage_key=%s", storage_key[:200], ) return self.get_url(storage_key) @@ -256,4 +279,5 @@ def get_storage_service() -> OSSStorageService: global _storage_service if _storage_service is None: _storage_service = OSSStorageService() + _storage_service.diagnose() return _storage_service diff --git a/tests/unit/test_p02_oss_credential_validation.py b/tests/unit/test_p02_oss_credential_validation.py new file mode 100644 index 000000000..deaca7dcc --- /dev/null +++ b/tests/unit/test_p02_oss_credential_validation.py @@ -0,0 +1,158 @@ +"""P0-2 修复：OSS 凭证验证 + 启动诊断。 + +验证： +1. 非开发环境 OSS_ACCESS_KEY_ID/SECRET 为空时启动失败 +2. 开发环境允许空凭证 +3. diagnose() 方法正确输出配置状态 +""" + +import logging +import os +import sys +from unittest.mock import MagicMock, patch + +import pytest + + +def _fresh_settings(env: str): + """清除 config 模块缓存，以指定 APP_ENV 重新导入 Settings。 + + 为非开发环境预设 OSS 环境变量，确保模块级 get_settings() 能成功完成导入。 + 测试方法内可根据需要清除这些变量来测试验证器。 + """ + for mod_name in [m for m in list(sys.modules) if "app.config" in m]: + del sys.modules[mod_name] + os.environ["APP_ENV"] = env + # 非开发环境下，为模块级导入提供有效凭证（避免导入时验证失败） + if env != "development": + os.environ.setdefault("OSS_ACCESS_KEY_ID", "test-key-for-import") + os.environ.setdefault("OSS_ACCESS_KEY_SECRET", "test-secret-for-import") + # 重置单例，让测试方法自行控制实例化 + from apps.api.app import config as _cfg + from apps.api.app.config import Settings + + _cfg._settings = None + return Settings + + +class TestOSSCredentialValidation: + """测试 OSS 凭证验证器（直接调用验证器类方法）。""" + + def test_empty_oss_key_id_rejected_in_staging(self): + """非开发环境 OSS_ACCESS_KEY_ID 为空应报错。""" + Settings = _fresh_settings("staging") + with pytest.raises(Exception, match="OSS_ACCESS_KEY_ID"): + Settings.validate_oss_access_key_id("") + + def test_empty_oss_key_secret_rejected_in_staging(self): + """非开发环境 OSS_ACCESS_KEY_SECRET 为空应报错。""" + Settings = _fresh_settings("staging") + with pytest.raises(Exception, match="OSS_ACCESS_KEY_SECRET"): + Settings.validate_oss_access_key_secret("") + + def test_empty_oss_credentials_allowed_in_development(self): + """开发环境允许空 OSS 凭证。""" + Settings = _fresh_settings("development") + assert Settings.validate_oss_access_key_id("") == "" + assert Settings.validate_oss_access_key_secret("") == "" + + def test_valid_credentials_pass_validation(self): + """有效凭证应通过验证。""" + Settings = _fresh_settings("staging") + assert Settings.validate_oss_access_key_id("test-key-id") == "test-key-id" + assert Settings.validate_oss_access_key_secret("test-key-secret") == "test-key-secret" + + def test_valid_credentials_instantiation_succeeds(self): + """有效凭证应能成功创建 Settings 实例。""" + os.environ.pop("OSS_ACCESS_KEY_ID", None) + os.environ.pop("OSS_ACCESS_KEY_SECRET", None) + Settings = _fresh_settings("staging") + os.environ["OSS_ACCESS_KEY_ID"] = "test-key-id" + os.environ["OSS_ACCESS_KEY_SECRET"] = "test-key-secret" + s = Settings(_env_file=None) + assert s.OSS_ACCESS_KEY_ID == "test-key-id" + assert s.OSS_ACCESS_KEY_SECRET == "test-key-secret" + + +class TestOSSDiagnose: + """测试 OSSStorageService.diagnose() 方法。""" + + @patch("apps.api.app.core.storage.oss2", None) + @patch("apps.api.app.core.storage.get_settings") + def test_diagnose_logs_error_when_bucket_none(self, mock_settings, caplog): + """bucket=None 时 diagnose 应输出 ERROR 日志。""" + from apps.api.app.core.storage import OSSStorageService + + mock_settings.return_value.OSS_BUCKET_NAME = "test-bucket" + mock_settings.return_value.OSS_ENDPOINT = "oss-cn-test.com" + mock_settings.return_value.OSS_ACCESS_KEY_ID = "" + mock_settings.return_value.OSS_ACCESS_KEY_SECRET = "" + + service = OSSStorageService() + assert service.bucket is None + + with caplog.at_level(logging.ERROR, logger="apps.api.app.core.storage"): + service.diagnose() + + assert any("❌" in record.message for record in caplog.records) + + @patch("apps.api.app.core.storage.oss2") + @patch("apps.api.app.core.storage.get_settings") + def test_diagnose_logs_success_when_bucket_configured(self, mock_settings, mock_oss2, caplog): + """bucket 已配置时 diagnose 应输出成功日志。""" + from apps.api.app.core.storage import OSSStorageService + + mock_settings.return_value.OSS_BUCKET_NAME = "test-bucket" + mock_settings.return_value.OSS_ENDPOINT = "oss-cn-test.com" + mock_settings.return_value.OSS_ACCESS_KEY_ID = "test-key-id" + mock_settings.return_value.OSS_ACCESS_KEY_SECRET = "test-key-secret" + mock_oss2.Bucket.return_value = MagicMock() + + service = OSSStorageService() + assert service.bucket is not None + + with caplog.at_level(logging.INFO, logger="apps.api.app.core.storage"): + service.diagnose() + + assert any("OSS诊断" in record.message for record in caplog.records) + + +class TestOSSHTTPSEndpoint: + """测试 P0-2 真正根因：sign_url 必须返回 HTTPS URL。""" + + @patch("apps.api.app.core.storage.oss2") + @patch("apps.api.app.core.storage.get_settings") + def test_endpoint_without_scheme_gets_https_prefix(self, mock_settings, mock_oss2): + """endpoint 无 scheme 时应自动加 https://，确保 sign_url 生成 HTTPS URL。""" + from apps.api.app.core.storage import OSSStorageService + + mock_settings.return_value.OSS_BUCKET_NAME = "test-bucket" + mock_settings.return_value.OSS_ENDPOINT = "oss-cn-hangzhou.aliyuncs.com" + mock_settings.return_value.OSS_ACCESS_KEY_ID = "test-key-id" + mock_settings.return_value.OSS_ACCESS_KEY_SECRET = "test-key-secret" + mock_oss2.Bucket.return_value = MagicMock() + + OSSStorageService() + + # 验证传给 oss2.Bucket 的 endpoint 带了 https:// + call_args = mock_oss2.Bucket.call_args + endpoint_passed = call_args[0][1] # 第二个位置参数 + assert endpoint_passed == "https://oss-cn-hangzhou.aliyuncs.com" + + @patch("apps.api.app.core.storage.oss2") + @patch("apps.api.app.core.storage.get_settings") + def test_endpoint_with_existing_https_not_doubled(self, mock_settings, mock_oss2): + """endpoint 已有 https:// 时不应重复添加。""" + from apps.api.app.core.storage import OSSStorageService + + mock_settings.return_value.OSS_BUCKET_NAME = "test-bucket" + mock_settings.return_value.OSS_ENDPOINT = "https://oss-cn-hangzhou.aliyuncs.com" + mock_settings.return_value.OSS_ACCESS_KEY_ID = "test-key-id" + mock_settings.return_value.OSS_ACCESS_KEY_SECRET = "test-key-secret" + mock_oss2.Bucket.return_value = MagicMock() + + OSSStorageService() + + call_args = mock_oss2.Bucket.call_args + endpoint_passed = call_args[0][1] + assert endpoint_passed == "https://oss-cn-hangzhou.aliyuncs.com"