fix(auth): harden simple auth and router imports

apps/api/app/api/routes/__init__.py

@@ -1,56 +1,5 @@
-from fastapi import APIRouter
+"""Compatibility exports for the canonical API router module."""
 
-from app.api.routes.asset_libraries import router as asset_libraries_router
-from app.api.routes.assets import router as assets_router
-# from app.api.routes.auth import router as auth_router  # 暂时注释，待修复循环导入
-from app.api.routes.auth_simple import router as auth_router  # 临时简化版
-from app.api.routes.classification_jobs import router as classification_jobs_router
-from app.api.routes.health import router as health_router
-from app.api.routes.ingest_jobs import router as ingest_jobs_router
-from app.api.routes.project_management import router as project_management_router
-from app.api.routes.projects import router as projects_router
-from app.api.routes.upload import router as upload_router
-
-api_router = APIRouter(prefix="/api/v1")
-
-api_router.include_router(
-    auth_router,
-    tags=["认证"],
-)
-api_router.include_router(
-    projects_router,
-    prefix="/projects",
-    tags=["项目管理"],
-)
-api_router.include_router(
-    asset_libraries_router,
-    prefix="/asset-libraries",
-    tags=["素材库管理"],
-)
-api_router.include_router(
-    assets_router,
-    prefix="/assets",
-    tags=["素材资产"],
-)
-api_router.include_router(
-    ingest_jobs_router,
-    prefix="/ingest-jobs",
-    tags=["导入任务"],
-)
-api_router.include_router(
-    classification_jobs_router,
-    prefix="/classification-jobs",
-    tags=["分类任务"],
-)
-api_router.include_router(
-    upload_router,
-    prefix="/upload",
-    tags=["文件上传"],
-)
-api_router.include_router(
-    project_management_router,
-    prefix="/project-management",
-    tags=["项目推进管理"],
-)
+from app.api.router import api_router, health_router
 
 __all__ = ["api_router", "health_router"]

apps/api/app/api/routes/auth_simple.py

@@ -1,23 +1,25 @@
 """
-简化版认证 API（使用 SQLAlchemy ORM）
+认证 API（SQLAlchemy ORM）
 """
-from fastapi import APIRouter, HTTPException, status, Depends
-from pydantic import BaseModel
-from sqlalchemy.orm import Session
+from datetime import datetime, timedelta, timezone
 import hashlib
 import secrets
-from datetime import datetime
 
+import jwt
+from fastapi import APIRouter, Depends, HTTPException, status
+from pydantic import BaseModel, EmailStr
+from sqlalchemy.orm import Session
+
+from app.config import settings
 from app.dependencies import get_db_session
 from packages.adapters.sqlalchemy_impl.models import UserModel
+from packages.domain.auth import password_hasher, password_validator
 
 router = APIRouter(prefix="/auth", tags=["认证"])
 
 
-# ==================== Models ====================
-
 class RegisterRequest(BaseModel):
-    email: str
+    email: EmailStr
     password: str
     username: str
     display_name: str
@@ -32,7 +34,7 @@ class RegisterResponse(BaseModel):
 
 
 class LoginRequest(BaseModel):
-    email: str
+    email: EmailStr
     password: str
 
 
@@ -43,115 +45,128 @@ class LoginResponse(BaseModel):
     email: str
     username: str
     display_name: str
+    expires_in: int
 
 
-# ==================== Helper Functions ====================
+ACCESS_TOKEN_EXPIRE_MINUTES = 30
+JWT_ALGORITHM = "HS256"
+LEGACY_SHA256_HEX_LENGTH = 64
 
-def hash_password(password: str) -> str:
-    """简单的密码哈希（仅用于测试）"""
+
+def _normalize_email(email: str) -> str:
+    return email.strip().lower()
+
+
+def _normalize_username(username: str) -> str:
+    return username.strip()
+
+
+def _is_legacy_sha256_hash(password_hash: str) -> bool:
+    return len(password_hash) == LEGACY_SHA256_HEX_LENGTH and all(
+        char in "0123456789abcdef" for char in password_hash.lower()
+    )
+
+
+def _legacy_sha256(password: str) -> str:
     return hashlib.sha256(password.encode()).hexdigest()
 
 
-def generate_token() -> str:
-    """生成访问令牌"""
-    return secrets.token_urlsafe(32)
+def _verify_password_with_legacy_upgrade(password: str, user: UserModel, db: Session) -> bool:
+    stored_hash = user.password_hash or ""
+    if password_hasher.verify_password(password, stored_hash):
+        return True
+
+    if _is_legacy_sha256_hash(stored_hash) and secrets.compare_digest(_legacy_sha256(password), stored_hash):
+        user.password_hash = password_hasher.hash_password(password)
+        db.add(user)
+        db.commit()
+        db.refresh(user)
+        return True
+
+    return False
 
 
-# ==================== API Endpoints ====================
+def _create_access_token(user: UserModel) -> tuple[str, int]:
+    expires_delta = timedelta(minutes=ACCESS_TOKEN_EXPIRE_MINUTES)
+    now = datetime.now(timezone.utc)
+    payload = {
+        "sub": user.id,
+        "email": user.email,
+        "type": "user_auth",
+        "iat": now,
+        "exp": now + expires_delta,
+    }
+    token = jwt.encode(payload, settings.JWT_SECRET_KEY, algorithm=JWT_ALGORITHM)
+    return token, int(expires_delta.total_seconds())
+
 
 @router.post("/register", response_model=RegisterResponse, status_code=status.HTTP_201_CREATED)
 async def register(request: RegisterRequest, db: Session = Depends(get_db_session)):
-    """
-    用户注册（简化版）
-    """
-    # 检查邮箱是否已存在
-    existing_user = db.query(UserModel).filter(UserModel.email == request.email).first()
-    
+    email = _normalize_email(request.email)
+    username = _normalize_username(request.username)
+    display_name = request.display_name.strip()
+
+    if not username:
+        raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail="用户名不能为空")
+
+    if not display_name:
+        raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail="显示名称不能为空")
+
+    password_valid, password_error = password_validator.validate(request.password)
+    if not password_valid:
+        raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail=password_error)
+
+    existing_user = db.query(UserModel).filter(UserModel.email == email).first()
     if existing_user:
-        raise HTTPException(
-            status_code=status.HTTP_400_BAD_REQUEST,
-            detail="邮箱已被注册"
-        )
-    
-    # 检查用户名是否已存在
-    if request.username:
-        existing_username = db.query(UserModel).filter(UserModel.username == request.username).first()
-        
-        if existing_username:
-            raise HTTPException(
-                status_code=status.HTTP_400_BAD_REQUEST,
-                detail="用户名已被使用"
-            )
-    
-    # 创建用户
-    user_id = f"user_{secrets.token_hex(8)}"
-    password_hash = hash_password(request.password)
-    
+        raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail="邮箱已被注册")
+
+    existing_username = db.query(UserModel).filter(UserModel.username == username).first()
+    if existing_username:
+        raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail="用户名已被使用")
+
     new_user = UserModel(
-        id=user_id,
-        email=request.email,
-        username=request.username,
-        display_name=request.display_name,
-        password_hash=password_hash,
+        id=f"user_{secrets.token_hex(8)}",
+        email=email,
+        username=username,
+        display_name=display_name,
+        password_hash=password_hasher.hash_password(request.password),
         email_verified=False,
-        created_at=datetime.utcnow()
+        created_at=datetime.now(timezone.utc),
     )
-    
+
     db.add(new_user)
     db.commit()
     db.refresh(new_user)
-    
+
     return RegisterResponse(
         user_id=new_user.id,
         email=new_user.email,
         username=new_user.username or "",
         display_name=new_user.display_name,
-        message="注册成功！"
+        message="注册成功！",
     )
 
 
 @router.post("/login", response_model=LoginResponse)
 async def login(request: LoginRequest, db: Session = Depends(get_db_session)):
-    """
-    用户登录（简化版）
-    """
-    # 查找用户
-    user = db.query(UserModel).filter(UserModel.email == request.email).first()
-    
-    if not user:
-        raise HTTPException(
-            status_code=status.HTTP_401_UNAUTHORIZED,
-            detail="邮箱或密码错误"
-        )
-    
-    # 验证密码
-    password_hash = hash_password(request.password)
-    if password_hash != user.password_hash:
-        raise HTTPException(
-            status_code=status.HTTP_401_UNAUTHORIZED,
-            detail="邮箱或密码错误"
-        )
-    
-    # 生成访问令牌
-    access_token = generate_token()
-    
+    email = _normalize_email(request.email)
+    user = db.query(UserModel).filter(UserModel.email == email).first()
+
+    if not user or not _verify_password_with_legacy_upgrade(request.password, user, db):
+        raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="邮箱或密码错误")
+
+    access_token, expires_in = _create_access_token(user)
+
     return LoginResponse(
         access_token=access_token,
         user_id=user.id,
         email=user.email,
         username=user.username or "",
-        display_name=user.display_name
+        display_name=user.display_name,
+        expires_in=expires_in,
     )
 
 
 @router.get("/me")
 async def get_current_user_info():
-    """
-    获取当前用户信息（临时返回模拟数据）
-    """
-    return {
-        "user_id": "test_user",
-        "email": "test@example.com",
-        "username": "testuser",
-        "display_name": "测试用户"
-    }
+    raise HTTPException(status_code=status.HTTP_501_NOT_IMPLEMENTED, detail="/auth/me requires bearer-token dependency integration")

apps/api/app/config.py

@@ -23,6 +23,8 @@ class Settings(BaseSettings):
     REDIS_URL: str = "redis://localhost:6379/0"
     REDIS_MAX_CONNECTIONS: int = 50
 
+    JWT_SECRET_KEY: str = "your-secret-key-change-in-production"
+
     CELERY_BROKER_URL: str = "redis://localhost:6379/0"
     CELERY_RESULT_BACKEND: str = "redis://localhost:6379/1"
 

apps/api/app/core/storage.py

@@ -1,30 +1,39 @@
 """阿里云 OSS 存储服务"""
-from datetime import timedelta
-from typing import BinaryIO, Optional
+import logging
 from urllib.parse import urlparse
 import os
 
-import oss2
+try:
+    import oss2
+except ImportError:  # pragma: no cover - exercised in minimal local/test environments
+    oss2 = None
 from app.config import get_settings
 
+logger = logging.getLogger(__name__)
+
 
 class OSSStorageService:
     """阿里云 OSS 存储服务"""
 
     def __init__(self):
         settings = get_settings()
-        auth = oss2.Auth(
-            settings.OSS_ACCESS_KEY_ID,
-            settings.OSS_ACCESS_KEY_SECRET
-        )
-        self.bucket = oss2.Bucket(
-            auth,
-            settings.OSS_ENDPOINT,
-            settings.OSS_BUCKET_NAME
-        )
         self.bucket_name = settings.OSS_BUCKET_NAME
         self.public_url = f"https://{settings.OSS_BUCKET_NAME}.{settings.OSS_ENDPOINT}"
         self.local_url_prefix = os.getenv("GENERATED_FILES_URL_PREFIX", "/generated-files")
+        self.bucket = None
+
+        if settings.OSS_ACCESS_KEY_ID and settings.OSS_ACCESS_KEY_SECRET:
+            if oss2 is None:
+                raise RuntimeError("oss2 is required when OSS credentials are configured")
+            auth = oss2.Auth(
+                settings.OSS_ACCESS_KEY_ID,
+                settings.OSS_ACCESS_KEY_SECRET,
+            )
+            self.bucket = oss2.Bucket(
+                auth,
+                settings.OSS_ENDPOINT,
+                settings.OSS_BUCKET_NAME,
+            )
 
     def _is_local_generated_url(self, storage_key_or_url: str) -> bool:
         parsed = urlparse(storage_key_or_url)
@@ -48,6 +57,9 @@ class OSSStorageService:
         Returns:
             文件公网 URL
         """
+        if self.bucket is None:
+            raise RuntimeError("OSS storage is not configured")
+
         try:
             # 如果是字符串路径，从本地文件上传
             if isinstance(file_or_path, str):
@@ -84,8 +96,11 @@ class OSSStorageService:
         Returns:
             签名 URL
         """
-        if self._is_local_generated_url(storage_key_or_url):
-            return storage_key_or_url
+        if self.bucket is None:
+            if self._is_local_generated_url(storage_key_or_url):
+                return storage_key_or_url
+            return self.get_url(self._normalize_storage_key(storage_key_or_url))
+
         storage_key = self._normalize_storage_key(storage_key_or_url)
         try:
             return self.bucket.sign_url('GET', storage_key, expires_seconds)
@@ -108,6 +123,9 @@ class OSSStorageService:
             storage_key: 存储键
             local_path: 本地文件路径
         """
+        if self.bucket is None:
+            raise RuntimeError("OSS storage is not configured")
+
         try:
             os.makedirs(os.path.dirname(local_path), exist_ok=True)
             self.bucket.get_object_to_file(storage_key, local_path)
@@ -121,10 +139,13 @@ class OSSStorageService:
         Args:
             storage_key: 存储键
         """
+        if self.bucket is None:
+            return
+
         try:
             self.bucket.delete_object(storage_key)
-        except Exception as e:
-            print(f"Error deleting file from OSS: {e}")
+        except Exception as error:
+            logger.warning("Failed to delete file from OSS", extra={"storage_key": storage_key, "error": str(error)})
 
     def file_exists(self, storage_key: str) -> bool:
         """
@@ -136,6 +157,8 @@ class OSSStorageService:
         Returns:
             是否存在
         """
+        if self.bucket is None:
+            return False
         return self.bucket.object_exists(storage_key)
 
 

docs/全面代码审计报告-2026-06-21.md

@@ -0,0 +1,176 @@
+# 全面代码审计报告 - 2026-06-21
+
+## 一、审计目标
+
+对小虾 SaaS 进行一次系统性代码审计，重点不是局部补丁，而是发现并修复：
+
+- 架构分层不一致
+- 重复入口和重复实现
+- 安全与认证缺陷
+- 外部依赖导入副作用
+- 配置真源缺失
+- 可验证的 P0/P1 Bug
+
+## 二、本轮已确认并修复的问题
+
+### 1. P0：简化认证使用 SHA256 存储密码
+
+**涉及文件**：`apps/api/app/api/routes/auth_simple.py`
+
+**问题**：
+- 注册使用 `hashlib.sha256(password)` 保存密码。
+- 登录用同样 SHA256 比对。
+- 这是不可接受的密码存储方案，缺少 salt 和 cost factor。
+
+**根因**：
+- `auth_simple.py` 作为临时实现进入真实路由后没有回收。
+- 项目已有 `PasswordHasher(bcrypt)`，但真实 API 没有复用。
+
+**修复**：
+- 注册改为 `password_hasher.hash_password()`。
+- 登录改为 bcrypt 校验。
+- 对历史 SHA256 用户做透明迁移：首次成功登录后自动升级为 bcrypt。
+- 增加密码强度校验。
+
+### 2. P0：登录返回不可验证随机 token
+
+**涉及文件**：`apps/api/app/api/routes/auth_simple.py`、`apps/api/app/config.py`
+
+**问题**：
+- 登录返回 `secrets.token_urlsafe(32)`。
+- Token 没有签名、没有 payload、没有过期时间、无法被标准认证中间件验证。
+
+**根因**：
+- 临时认证实现绕过了项目已有 JWT 设计。
+- API settings 缺少 `JWT_SECRET_KEY` 配置项。
+
+**修复**：
+- 登录返回 HS256 JWT。
+- JWT payload 包含 `sub/email/type/iat/exp`。
+- 响应新增 `expires_in`。
+- `Settings` 增加 `JWT_SECRET_KEY`，由 `.env` 可覆盖。
+
+### 3. P1：API 路由包存在重复入口和导入副作用
+
+**涉及文件**：`apps/api/app/api/routes/__init__.py`
+
+**问题**：
+- `routes/__init__.py` 维护了一份重复 `api_router`。
+- 正式入口 `app/api/router.py` 也维护一份 `api_router`。
+- 两者内容不一致，旧入口缺少后续生成/成片路由。
+- 导入 `app.api.routes` 会连带导入所有路由和外部依赖。
+
+**根因**：
+- 迁移过程中保留了旧入口，没有明确唯一真源。
+
+**修复**：
+- `routes/__init__.py` 改为兼容转发，只导出 canonical `app.api.router` 中的 `api_router` 和 `health_router`。
+
+### 4. P1：OSS SDK 为硬导入，导致无 OSS 本地环境无法导入 API Router
+
+**涉及文件**：`apps/api/app/core/storage.py`
+
+**问题**：
+- `import oss2` 在模块顶层执行。
+- 本地/测试环境未安装 `oss2` 时，导入任何包含 `generated_videos` 的 API router 都会失败。
+
+**根因**：
+- 外部存储适配器没有做到依赖可选和配置驱动。
+- 无 OSS 配置时，本地 generated 文件已经可以工作，但代码仍强制要求 OSS SDK。
+
+**修复**：
+- `oss2` 改为可选导入。
+- 只有配置了 OSS AK/SK 并实际创建 bucket 时才要求 `oss2`。
+- 未配置 OSS 时，下载 URL 对本地 `/generated-files/` 直接返回，对 OSS URL 回退为公开 URL。
+- 上传/下载等 OSS 专属操作在未配置时返回明确错误。
+
+## 三、已补充测试
+
+### 新增
+
+- `tests/unit/test_auth_simple.py`
+
+覆盖：
+- 登录 token 是可验证 JWT。
+- bcrypt 密码可登录。
+- 历史 SHA256 密码登录后自动升级 bcrypt。
+- 错误密码拒绝且不会写库。
+
+### 已运行通过
+
+```bash
+python -m pytest tests/unit/test_auth_simple.py tests/unit/test_password_hasher.py tests/integration/test_generation_pipeline.py tests/integration/test_projects.py -q
+```
+
+结果：`30 passed`
+
+## 四、仍需继续治理的问题
+
+### P1：认证体系双轨
+
+当前真实路由仍使用 `auth_simple.py`，完整认证实现位于：
+
+- `apps/api/app/api/routes/auth.py`
+- `packages/application/auth/*`
+- `packages/domain/auth/session_store.py`
+- `packages/domain/auth/email_service.py`
+
+问题：
+- 完整认证路由未接入 canonical router。
+- `auth.py` 依赖缺失的 `get_container()`。
+- `auth_simple.py` 仍承担生产入口。
+
+建议：
+1. 先补完整 DI container。
+2. 将 session/email 外部服务移出 domain。
+3. 用完整 auth 替换 auth_simple。
+4. 删除 auth_simple 或改名为测试 fixture。
+
+### P1：Domain 层存在外部基础设施依赖
+
+涉及：
+- `packages/domain/auth/session_store.py` 直接依赖 Redis。
+- `packages/domain/auth/email_service.py` 直接依赖 SMTP。
+
+问题：
+- 违反 Clean Architecture。
+- Domain import 会创建全局外部服务实例。
+- 错误处理使用 `print` 和吞异常。
+
+建议：
+- 抽象 `SessionStore`、`EmailSender` port。
+- Redis/SMTP 实现迁移到 adapters。
+- UseCase 通过构造函数注入 port。
+
+### P1：Repository 体系双轨
+
+当前存在两套持久化体系：
+
+- `packages/adapters/sqlalchemy_impl/*`
+- `packages/adapters/postgres/*`
+
+问题：
+- API 主链路使用 SQLAlchemy。
+- 旧 workspace/auth 代码仍引用 psycopg2/postgres adapters。
+- 容易出现 schema、事务、连接池和模型映射漂移。
+
+建议：
+- 统一到 SQLAlchemy。
+- Postgres psycopg2 adapters 标记 deprecated 后逐步删除。
+- 先迁移 User/Workspace/Member/Invitation。
+
+### P2：临时代码仍在主线
+
+发现：
+- `auth_simple.py` 名称和历史注释仍体现临时方案。
+- 部分测试和文档仍引用旧 MinIO/OSS 混合术语。
+- `__pycache__` 文件出现在工作树扫描中，需确认 `.gitignore` 和仓库状态。
+
+## 五、下一步建议修复顺序
+
+1. 完整 DI container：恢复 `get_container()`，但不要回到全局硬编码。
+2. 认证体系收敛：完整 auth 替换 auth_simple。
+3. 外部服务出 domain：Redis session / SMTP email 移到 adapters。
+4. Repository 统一：优先 User/Workspace 迁移 SQLAlchemy。
+5. 删除死代码：清理旧 routes、旧 postgres adapters、重复文档入口。
+6. 全量测试和 CI：后端 unit/integration + 前端 type-check/build + staging smoke。

tests/unit/test_auth_simple.py

@@ -0,0 +1,76 @@
+import sys
+from pathlib import Path
+
+import jwt
+
+ROOT = Path(__file__).resolve().parents[2]
+API_ROOT = ROOT / "apps" / "api"
+if str(API_ROOT) not in sys.path:
+    sys.path.insert(0, str(API_ROOT))
+
+import importlib.util
+
+spec = importlib.util.spec_from_file_location("auth_simple", API_ROOT / "app" / "api" / "routes" / "auth_simple.py")
+auth_simple = importlib.util.module_from_spec(spec)
+assert spec.loader is not None
+spec.loader.exec_module(auth_simple)
+
+_create_access_token = auth_simple._create_access_token
+_verify_password_with_legacy_upgrade = auth_simple._verify_password_with_legacy_upgrade
+from app.config import settings
+from packages.adapters.sqlalchemy_impl.models import UserModel
+from packages.domain.auth import password_hasher
+
+
+class DummySession:
+    def __init__(self):
+        self.committed = False
+        self.refreshed = False
+        self.added = []
+
+    def add(self, item):
+        self.added.append(item)
+
+    def commit(self):
+        self.committed = True
+
+    def refresh(self, item):
+        self.refreshed = True
+
+
+def test_create_access_token_returns_verifiable_jwt():
+    user = UserModel(id="user-1", email="user@example.com", username="user", display_name="User")
+
+    token, expires_in = _create_access_token(user)
+    payload = jwt.decode(token, settings.JWT_SECRET_KEY, algorithms=["HS256"])
+
+    assert expires_in == 1800
+    assert payload["sub"] == "user-1"
+    assert payload["email"] == "user@example.com"
+    assert payload["type"] == "user_auth"
+
+
+def test_verify_password_accepts_bcrypt_hash():
+    db = DummySession()
+    user = UserModel(password_hash=password_hasher.hash_password("Password1"))
+
+    assert _verify_password_with_legacy_upgrade("Password1", user, db) is True
+    assert db.committed is False
+
+
+def test_verify_password_upgrades_legacy_sha256_hash():
+    db = DummySession()
+    user = UserModel(password_hash="19513fdc9da4fb72a4a05eb66917548d3c90ff94d5419e1f2363eea89dfee1dd")
+
+    assert _verify_password_with_legacy_upgrade("Password1", user, db) is True
+    assert user.password_hash.startswith("$2")
+    assert db.committed is True
+    assert db.refreshed is True
+
+
+def test_verify_password_rejects_wrong_password():
+    db = DummySession()
+    user = UserModel(password_hash=password_hasher.hash_password("Password1"))
+
+    assert _verify_password_with_legacy_upgrade("WrongPassword1", user, db) is False
+    assert db.committed is False