-dev:完成了token效验的替换工作

.kiro/specs/sso-token-unification/design.md

@@ -0,0 +1,444 @@
+# Design Document: SSO Token 统一认证
+
+## Overview
+
+将标注平台从"本地 JWT 签发 + SSO 身份认证"模式迁移为"SSO 统一 token 签发"模式。核心变化是：后端不再签发 JWT，而是直接透传 SSO 中心的 token；中间件通过调用 SSO 的 `/oauth/userinfo` 端点验证 token 有效性，并引入内存缓存减少对 SSO 中心的请求压力。
+
+### 改造前后对比
+
+```
+改造前:
+  用户 → SSO 登录 → 获取 SSO token → 后端用 SSO token 拉取用户信息 → 后端签发本地 JWT → 前端用本地 JWT 访问 API
+
+改造后:
+  用户 → SSO 登录 → 获取 SSO token → 后端透传 SSO token 给前端 → 前端用 SSO token 访问 API → 后端向 SSO 验证 token
+```
+
+## Architecture
+
+```mermaid
+sequenceDiagram
+    participant User as 用户/前端
+    participant Platform as 标注平台后端
+    participant SSO as SSO 认证中心
+    participant Cache as Token 缓存
+    participant DB as 本地数据库
+
+    Note over User,SSO: === SSO 登录流程 ===
+    User->>Platform: GET /api/oauth/login
+    Platform->>User: 返回 authorization_url + state
+    User->>SSO: 重定向到 SSO 登录页
+    SSO->>User: 用户授权后重定向回 callback (带 code)
+    User->>Platform: GET /api/oauth/callback?code=xxx&state=yyy
+    Platform->>SSO: POST /oauth/token (用 code 换 token)
+    SSO->>Platform: 返回 access_token + refresh_token
+    Platform->>SSO: GET /oauth/userinfo (用 access_token)
+    SSO->>Platform: 返回用户信息 {code:0, data:{...}}
+    Platform->>DB: 同步用户信息到本地
+    Platform->>Cache: 缓存 token → user 映射
+    Platform->>User: 返回 SSO access_token + refresh_token + user
+
+    Note over User,SSO: === API 请求认证流程 ===
+    User->>Platform: API 请求 (Authorization: Bearer SSO_token)
+    Platform->>Cache: 查询 token 缓存
+    alt 缓存命中
+        Cache->>Platform: 返回缓存的用户信息
+    else 缓存未命中
+        Platform->>SSO: GET /oauth/userinfo (验证 token)
+        SSO->>Platform: 返回用户信息
+        Platform->>Cache: 缓存 token → user 映射
+    end
+    Platform->>User: 返回 API 响应
+
+    Note over User,SSO: === Token 刷新流程 ===
+    User->>Platform: POST /api/oauth/refresh (refresh_token)
+    Platform->>SSO: POST /oauth/token (grant_type=refresh_token)
+    SSO->>Platform: 返回新的 access_token + refresh_token
+    Platform->>User: 返回新 tokens
+```
+
+## Components and Interfaces
+
+### 后端组件变更
+
+#### 1. TokenCacheService（新增）
+
+内存级 token 缓存服务，减少对 SSO 中心的请求。
+
+```python
+class TokenCacheService:
+    """SSO Token 内存缓存服务"""
+    
+    def __init__(self, ttl_seconds: int = 300):
+        """
+        Args:
+            ttl_seconds: 缓存过期时间，默认 300 秒（5 分钟）
+        """
+        self._cache: Dict[str, CacheEntry] = {}
+        self._ttl = ttl_seconds
+    
+    def get(self, token: str) -> Optional[Dict]:
+        """查询缓存，返回用户信息或 None"""
+        ...
+    
+    def set(self, token: str, user_info: Dict) -> None:
+        """写入缓存"""
+        ...
+    
+    def invalidate(self, token: str) -> None:
+        """使指定 token 缓存失效"""
+        ...
+    
+    def clear(self) -> None:
+        """清空所有缓存"""
+        ...
+```
+
+#### 2. AuthMiddleware（改造）
+
+从本地 JWT 验证改为 SSO token 验证 + 缓存。
+
+```python
+# 改造前：
+payload = JWTService.verify_token(token, "access")
+
+# 改造后：
+# 1. 先查缓存
+user_info = token_cache.get(token)
+if not user_info:
+    # 2. 缓存未命中，调 SSO userinfo 验证
+    user_info = await sso_verify_token(token)
+    # 3. 验证成功则写入缓存
+    token_cache.set(token, user_info)
+```
+
+关键变化：
+- 中间件改为 async（需要异步调用 SSO）
+- 移除对 `JWTService.verify_token` 的依赖
+- 新增 SSO 不可用时返回 503 的处理
+
+#### 3. OAuthService（改造）
+
+- `exchange_code_for_token`: 保持不变，继续用授权码换 token
+- `get_user_info`: 保持不变，继续获取用户信息
+- `sync_user_from_oauth`: 保持不变，继续同步用户到本地
+- 新增 `refresh_sso_token`: 转发 refresh 请求到 SSO 中心
+- 新增 `verify_sso_token`: 用 SSO userinfo 端点验证 token
+
+```python
+@staticmethod
+async def verify_sso_token(access_token: str) -> Dict[str, Any]:
+    """
+    通过 SSO userinfo 端点验证 token 并获取用户信息
+    
+    Returns:
+        用户信息字典 {id, username, email, ...}
+    
+    Raises:
+        HTTPException(401): token 无效
+        HTTPException(503): SSO 不可用
+    """
+    ...
+
+@staticmethod
+async def refresh_sso_token(refresh_token: str) -> Dict[str, Any]:
+    """
+    向 SSO 中心刷新 token
+    
+    Returns:
+        新的 token 信息 {access_token, refresh_token, ...}
+    
+    Raises:
+        HTTPException(401): refresh_token 无效
+        HTTPException(503): SSO 不可用
+    """
+    ...
+```
+
+#### 4. OAuth Router（改造）
+
+- `/api/oauth/callback`: 不再调用 `JWTService`，直接返回 SSO token
+- `/api/oauth/refresh`（新增）: 转发 refresh 请求到 SSO 中心
+- `/api/oauth/login`: 保持不变
+- `/api/oauth/status`: 保持不变
+
+#### 5. Auth Router（移除）
+
+移除以下端点：
+- `POST /api/auth/register` — 不再支持本地注册
+- `POST /api/auth/login` — 不再支持本地登录
+- `POST /api/auth/refresh` — 刷新改走 `/api/oauth/refresh`
+- `GET /api/auth/me` — 改为 `GET /api/oauth/me`（从 SSO 验证后的 request.state 获取）
+
+#### 6. JWTService（移除）
+
+整个 `jwt_service.py` 不再需要，因为不再本地签发或验证 JWT。
+
+#### 7. AuthService（精简）
+
+移除 `register_user`、`login_user`、`refresh_tokens` 方法。
+保留 `get_current_user`（从数据库查询用户详情）。
+
+### 前端组件变更
+
+#### 1. OAuth Callback（改造）
+
+回调后存储的是 SSO token 而非本地 JWT：
+
+```typescript
+// 改造前：
+setAuth({
+  tokens: {
+    access_token: response.access_token,   // 本地 JWT
+    refresh_token: response.refresh_token,  // 本地 JWT
+    token_type: response.token_type,
+  },
+  user: response.user,
+});
+
+// 改造后（结构不变，内容变了）：
+setAuth({
+  tokens: {
+    access_token: response.access_token,   // SSO token
+    refresh_token: response.refresh_token,  // SSO refresh token
+    token_type: response.token_type,
+  },
+  user: response.user,
+});
+```
+
+实际上前端存储结构不需要变化，只是 token 的来源变了。
+
+#### 2. API Client Token Refresh（改造）
+
+刷新端点从 `/api/auth/refresh` 改为 `/api/oauth/refresh`：
+
+```typescript
+// 改造前：
+const response = await axios.post(`${API_BASE_URL}/api/auth/refresh`, {
+  refresh_token: tokens.refresh_token,
+});
+
+// 改造后：
+const response = await axios.post(`${API_BASE_URL}/api/oauth/refresh`, {
+  refresh_token: tokens.refresh_token,
+});
+```
+
+#### 3. Login Page（改造）
+
+移除本地登录表单，直接发起 SSO 登录：
+
+```typescript
+// 改造后的 LoginForm：
+// 页面加载时自动检查 OAuth 状态并跳转 SSO
+useEffect(() => {
+  startOAuthLogin();
+}, []);
+```
+
+#### 4. Register Page（移除）
+
+移除注册页面和相关路由。
+
+#### 5. Auth Middleware Public Paths（更新）
+
+```python
+PUBLIC_PATHS = {
+    "/",
+    "/health",
+    "/docs",
+    "/openapi.json",
+    "/redoc",
+    "/api/oauth/status",
+    "/api/oauth/login",
+    "/api/oauth/callback",
+    "/api/oauth/refresh",
+}
+```
+
+## Data Models
+
+### Token 缓存数据结构
+
+```python
+@dataclass
+class CacheEntry:
+    user_info: Dict[str, Any]  # {id, username, email, role, ...}
+    created_at: float          # time.time() 时间戳
+    
+    def is_expired(self, ttl: float) -> bool:
+        return (time.time() - self.created_at) > ttl
+```
+
+### SSO Userinfo 响应格式
+
+根据 SSO demo 代码，SSO 中心的 userinfo 响应格式为：
+
+```json
+{
+  "code": 0,
+  "data": {
+    "id": "user_123",
+    "username": "zhangsan",
+    "email": "zhangsan@example.com",
+    "name": "张三"
+  }
+}
+```
+
+### OAuth Token 响应格式
+
+```json
+{
+  "access_token": "eyJhbGciOiJSUzI1NiIs...",
+  "refresh_token": "dGhpcyBpcyBhIHJlZnJlc2g...",
+  "token_type": "bearer",
+  "expires_in": 3600
+}
+```
+
+或包装格式：
+
+```json
+{
+  "code": 0,
+  "data": {
+    "access_token": "eyJhbGciOiJSUzI1NiIs...",
+    "refresh_token": "dGhpcyBpcyBhIHJlZnJlc2g...",
+    "token_type": "bearer",
+    "expires_in": 3600
+  }
+}
+```
+
+### 配置变更
+
+```yaml
+# 移除 jwt 配置段
+# jwt:
+#   secret_key: "..."
+#   algorithm: "HS256"
+#   access_token_expire_minutes: 60
+#   refresh_token_expire_days: 7
+
+# 保留并增强 oauth 配置
+oauth:
+  enabled: true
+  base_url: "http://192.168.92.61:8000"
+  client_id: "nlKLQJdJK3f5ub7UDfQ_E71z2Lo3YSQx"
+  client_secret: "wh0HU_9T83rYMjfLFToNxFOKcrk_8H7Ba_27nNGlPqtTf9ROCytsOgp2ue0ol5mm"
+  redirect_uri: "http://localhost:4200/auth/callback"
+  scope: "profile email"
+  authorize_endpoint: "/oauth/login"
+  token_endpoint: "/oauth/token"
+  userinfo_endpoint: "/oauth/userinfo"
+  revoke_endpoint: "/oauth/revoke"
+  token_cache_ttl: 300  # 新增：缓存 TTL（秒）
+```
+
+### 数据库变更
+
+无数据库 schema 变更。`users` 表保持不变，继续通过 `oauth_provider` 和 `oauth_id` 关联 SSO 用户。
+
+
+## Correctness Properties
+
+*A property is a characteristic or behavior that should hold true across all valid executions of a system — essentially, a formal statement about what the system should do. Properties serve as the bridge between human-readable specifications and machine-verifiable correctness guarantees.*
+
+### Property 1: Cache hit returns cached user info
+
+*For any* token that has been previously verified and cached (within TTL), querying the cache with that token should return the same user information that was originally cached, without making any call to the SSO center.
+
+**Validates: Requirements 3.1, 3.2**
+
+### Property 2: Successful SSO verification populates cache
+
+*For any* valid SSO token, after the Auth_Middleware successfully verifies it against the SSO center, the token-to-user mapping should be present in the cache, and a subsequent cache lookup for the same token should return the user info.
+
+**Validates: Requirements 3.4**
+
+### Property 3: Invalid tokens are rejected
+
+*For any* token that the SSO center rejects (returns non-zero code or HTTP error), the Auth_Middleware should return HTTP 401 and the token should not be cached.
+
+**Validates: Requirements 3.5**
+
+### Property 4: Cache entries expire after TTL
+
+*For any* cached token entry, after the configured TTL has elapsed, the cache should return None for that token, forcing re-verification against the SSO center.
+
+**Validates: Requirements 3.7**
+
+### Property 5: User sync invariant
+
+*For any* SSO user information received during OAuth callback, after `sync_user_from_oauth` completes, the local database should contain a user record where `oauth_provider` equals "sso", `oauth_id` matches the SSO-provided ID, and `username` and `email` match the latest SSO-provided values.
+
+**Validates: Requirements 5.2, 5.3, 5.4**
+
+## Error Handling
+
+| 场景 | HTTP 状态码 | 错误类型 | 描述 |
+|------|------------|---------|------|
+| 缺少 Authorization header | 401 | `missing_token` | 请求未携带认证令牌 |
+| Bearer 格式错误 | 401 | `invalid_token_format` | Authorization header 格式不正确 |
+| SSO 验证失败（token 无效） | 401 | `invalid_token` | SSO 中心拒绝该 token |
+| SSO 验证失败（token 过期） | 401 | `token_expired` | SSO token 已过期 |
+| SSO 中心不可用 | 503 | `sso_unavailable` | 无法连接 SSO 认证中心 |
+| OAuth 未启用 | 503 | `sso_not_configured` | OAuth/SSO 配置未启用 |
+| Refresh token 无效 | 401 | `invalid_refresh_token` | SSO 刷新令牌无效或已过期 |
+| 用户信息同步失败 | 500 | `user_sync_error` | 无法将 SSO 用户同步到本地数据库 |
+
+### 错误响应格式
+
+保持与现有系统一致：
+
+```json
+{
+  "detail": "错误描述信息",
+  "error_type": "error_type_code"
+}
+```
+
+## Testing Strategy
+
+### 测试框架
+
+- 后端：`pytest` + `pytest-asyncio`（异步测试）
+- 属性测试：`hypothesis`（Python property-based testing 库）
+- 前端：现有的 Jest 测试框架
+
+### 单元测试
+
+1. **TokenCacheService 测试**
+   - 缓存写入和读取
+   - 缓存过期
+   - 缓存失效（invalidate）
+   - 并发访问安全性
+
+2. **OAuthService 新方法测试**
+   - `verify_sso_token` 成功/失败场景
+   - `refresh_sso_token` 成功/失败场景
+   - SSO 响应格式兼容性（标准格式 vs 包装格式）
+
+3. **AuthMiddleware 测试**
+   - 公开路径跳过认证
+   - 缓存命中路径
+   - 缓存未命中 → SSO 验证路径
+   - 各种错误场景
+
+4. **前端测试**
+   - API client 的 token refresh 端点变更
+   - OAuth callback 存储 SSO token
+
+### 属性测试
+
+每个属性测试至少运行 100 次迭代。
+
+- **Property 1**: 生成随机 token 和用户信息，写入缓存后立即读取，验证返回值一致
+- **Property 2**: 模拟 SSO 验证成功，验证缓存被填充
+- **Property 3**: 生成随机无效 token，验证都返回 401 且不被缓存
+- **Property 4**: 生成随机 token，写入缓存，模拟时间流逝超过 TTL，验证缓存返回 None
+- **Property 5**: 生成随机 SSO 用户信息，执行 sync，验证数据库记录正确
+
+测试标注格式：`Feature: sso-token-unification, Property {N}: {property_text}`

.kiro/specs/sso-token-unification/requirements.md

@@ -0,0 +1,96 @@
+# Requirements Document
+
+## Introduction
+
+将标注平台的用户认证体系从"本地 JWT 签发"迁移为"SSO 中心统一签发 token"。当前系统在 OAuth 回调后自行生成 JWT token 用于后续 API 认证，改造后将直接使用 SSO 中心签发的 access_token 和 refresh_token，实现统一认证体系。本地用户名/密码登录和注册功能将被移除，所有认证统一走 SSO。
+
+## Glossary
+
+- **SSO_Center**: 外部 SSO 认证中心，负责用户身份认证和 token 签发，地址由 `oauth.base_url` 配置
+- **Platform**: 本标注平台后端服务（FastAPI 应用）
+- **Frontend**: 本标注平台前端应用（React/TypeScript）
+- **SSO_Token**: SSO 认证中心签发的 access_token，用于 API 请求认证
+- **SSO_Refresh_Token**: SSO 认证中心签发的 refresh_token，用于刷新 SSO_Token
+- **Auth_Middleware**: 后端认证中间件，负责验证每个请求的 token 有效性
+- **Token_Introspection**: 向 SSO_Center 验证 token 有效性的过程
+- **User_Sync**: 从 SSO 用户信息同步到本地数据库的过程
+
+## Requirements
+
+### Requirement 1: 移除本地认证
+
+**User Story:** As a platform administrator, I want to remove local username/password authentication, so that all users authenticate through the unified SSO system.
+
+#### Acceptance Criteria
+
+1. WHEN the Platform starts, THE Platform SHALL only support SSO-based authentication (local login and registration endpoints removed)
+2. WHEN a request is made to the removed `/api/auth/login` endpoint, THE Platform SHALL return HTTP 404
+3. WHEN a request is made to the removed `/api/auth/register` endpoint, THE Platform SHALL return HTTP 404
+4. THE Frontend SHALL remove the local login form and registration form, displaying only the SSO login button
+
+### Requirement 2: SSO Token 透传
+
+**User Story:** As a developer, I want the platform to use SSO-issued tokens directly, so that the authentication system is unified with the SSO center.
+
+#### Acceptance Criteria
+
+1. WHEN the OAuth callback is completed successfully, THE Platform SHALL return the SSO_Token and SSO_Refresh_Token directly to the Frontend (instead of generating local JWT tokens)
+2. WHEN the Frontend makes API requests, THE Frontend SHALL attach the SSO_Token in the Authorization header as a Bearer token
+3. THE Platform SHALL NOT generate or sign any JWT tokens locally for user authentication purposes
+
+### Requirement 3: SSO Token 验证
+
+**User Story:** As a platform administrator, I want every API request to be validated against the SSO center, so that revoked or expired tokens are rejected immediately.
+
+#### Acceptance Criteria
+
+1. WHEN a protected API request is received, THE Auth_Middleware SHALL first check the local token cache for a valid cached session
+2. WHEN the token is found in the local cache, THE Auth_Middleware SHALL use the cached user information and attach it to the request state
+3. WHEN the token is not found in the local cache, THE Auth_Middleware SHALL validate the SSO_Token by calling the SSO_Center's `/oauth/userinfo` endpoint
+4. WHEN the SSO_Center returns a response with `code=0` and user data in the `data` field, THE Auth_Middleware SHALL cache the token-to-user mapping and attach user information to the request state
+5. WHEN the SSO_Center returns an error or the token is invalid, THE Auth_Middleware SHALL return HTTP 401 with an appropriate error message
+6. WHEN the SSO_Center is unreachable, THE Auth_Middleware SHALL return HTTP 503 with a service unavailable error message
+7. THE Auth_Middleware SHALL support a configurable cache TTL (default 5 minutes) to balance between performance and token revocation responsiveness
+
+### Requirement 4: Token 刷新
+
+**User Story:** As a user, I want my session to be refreshed automatically when my token expires, so that I don't have to re-login frequently.
+
+#### Acceptance Criteria
+
+1. WHEN the Frontend receives a 401 response indicating token expiration, THE Frontend SHALL attempt to refresh the token using the SSO_Refresh_Token
+2. WHEN refreshing the token, THE Platform SHALL forward the refresh request to the SSO_Center's token endpoint with `grant_type=refresh_token`
+3. WHEN the SSO_Center returns new tokens, THE Platform SHALL return the new SSO_Token and SSO_Refresh_Token to the Frontend
+4. WHEN the refresh request fails, THE Frontend SHALL clear stored tokens and redirect the user to the SSO login flow
+
+### Requirement 5: 用户信息同步
+
+**User Story:** As a platform administrator, I want user information to be synchronized from SSO on each login, so that user data stays up-to-date.
+
+#### Acceptance Criteria
+
+1. WHEN a user completes SSO login via the OAuth callback, THE Platform SHALL fetch user information from the SSO_Center and sync it to the local database
+2. WHEN the SSO user does not exist in the local database, THE Platform SHALL create a new user record with the SSO-provided information
+3. WHEN the SSO user already exists in the local database, THE Platform SHALL update the username and email if they have changed
+4. THE Platform SHALL store the `oauth_provider` as "sso" and the `oauth_id` from the SSO_Center for each synced user
+
+### Requirement 6: 前端认证流程改造
+
+**User Story:** As a user, I want a seamless SSO login experience, so that I can access the platform through the unified authentication system.
+
+#### Acceptance Criteria
+
+1. WHEN a user visits the login page, THE Frontend SHALL automatically initiate the SSO login flow (redirect to SSO_Center)
+2. WHEN the OAuth callback returns tokens, THE Frontend SHALL store the SSO_Token and SSO_Refresh_Token in localStorage
+3. WHEN the user clicks logout, THE Frontend SHALL clear stored tokens and redirect to the SSO_Center's logout endpoint (if available)
+4. WHEN the Frontend detects an invalid or expired token and refresh fails, THE Frontend SHALL redirect the user to the SSO login flow
+
+### Requirement 7: 配置简化
+
+**User Story:** As a developer, I want the configuration to reflect the new SSO-only authentication model, so that the system is easy to understand and maintain.
+
+#### Acceptance Criteria
+
+1. THE Platform SHALL remove the local JWT `secret_key` configuration since tokens are no longer signed locally
+2. THE Platform SHALL retain the `oauth` configuration section for SSO_Center connection settings
+3. WHEN the `oauth.enabled` configuration is set to false, THE Platform SHALL return HTTP 503 for all authentication-related requests with a message indicating SSO is not configured

.kiro/specs/sso-token-unification/tasks.md

@@ -0,0 +1,119 @@
+# Implementation Plan: SSO Token 统一认证
+
+## Overview
+
+将标注平台从本地 JWT 签发迁移为 SSO 统一 token 认证。按照后端核心组件 → 中间件改造 → 路由改造 → 前端适配 → 清理的顺序实施，确保每一步都可验证。
+
+## Tasks
+
+- [x] 1. 实现 TokenCacheService
+  - [x] 1.1 创建 `backend/services/token_cache_service.py`，实现内存级 token 缓存
+    - 实现 `CacheEntry` 数据类（user_info + created_at）
+    - 实现 `get(token)` 方法：查询缓存，检查 TTL，过期则删除并返回 None
+    - 实现 `set(token, user_info)` 方法：写入缓存
+    - 实现 `invalidate(token)` 和 `clear()` 方法
+    - 支持通过构造函数配置 TTL（默认 300 秒）
+    - _Requirements: 3.1, 3.2, 3.7_
+  - [ ]* 1.2 编写 TokenCacheService 属性测试
+    - **Property 1: Cache hit returns cached user info**
+    - **Property 4: Cache entries expire after TTL**
+    - **Validates: Requirements 3.1, 3.2, 3.7**
+
+- [x] 2. 改造 OAuthService，新增 SSO token 验证和刷新方法
+  - [x] 2.1 在 `backend/services/oauth_service.py` 中新增 `verify_sso_token` 方法
+    - 调用 SSO `/oauth/userinfo` 端点验证 token
+    - 处理 `{"code": 0, "data": {...}}` 包装格式和标准格式
+    - SSO 不可用时抛出 HTTPException(503)
+    - token 无效时抛出 HTTPException(401)
+    - _Requirements: 3.3, 3.4, 3.5, 3.6_
+  - [x] 2.2 在 `backend/services/oauth_service.py` 中新增 `refresh_sso_token` 方法
+    - 向 SSO `/oauth/token` 端点发送 `grant_type=refresh_token` 请求
+    - 处理包装格式和标准格式的响应
+    - _Requirements: 4.2, 4.3_
+  - [ ]* 2.3 编写 OAuthService 新方法的单元测试
+    - 测试 `verify_sso_token` 成功/失败/SSO 不可用场景
+    - 测试 `refresh_sso_token` 成功/失败场景
+    - _Requirements: 3.3, 3.4, 3.5, 3.6, 4.2, 4.3_
+
+- [x] 3. 改造 AuthMiddleware
+  - [x] 3.1 重写 `backend/middleware/auth_middleware.py` 的 `dispatch` 方法
+    - 改为先查 TokenCacheService 缓存
+    - 缓存未命中时调用 `OAuthService.verify_sso_token` 验证
+    - 验证成功后写入缓存并同步用户到本地数据库
+    - 移除对 JWTService 的依赖
+    - 更新 PUBLIC_PATHS（移除 auth 端点，添加 oauth/refresh）
+    - 处理 SSO 不可用（503）和 token 无效（401）
+    - _Requirements: 3.1, 3.2, 3.3, 3.4, 3.5, 3.6_
+  - [ ]* 3.2 编写 AuthMiddleware 属性测试
+    - **Property 2: Successful SSO verification populates cache**
+    - **Property 3: Invalid tokens are rejected**
+    - **Validates: Requirements 3.4, 3.5**
+
+- [x] 4. Checkpoint - 后端核心组件验证
+  - Ensure all tests pass, ask the user if questions arise.
+
+- [x] 5. 改造 OAuth Router 和移除 Auth Router
+  - [x] 5.1 改造 `backend/routers/oauth.py` 的 callback 端点
+    - 移除 `JWTService.create_access_token` 和 `JWTService.create_refresh_token` 调用
+    - 直接返回 SSO 的 access_token 和 refresh_token
+    - 保留用户信息同步逻辑
+    - _Requirements: 2.1, 5.1, 5.2, 5.3, 5.4_
+  - [x] 5.2 在 `backend/routers/oauth.py` 中新增 `/api/oauth/refresh` 端点
+    - 接收 refresh_token，调用 `OAuthService.refresh_sso_token`
+    - 返回新的 access_token 和 refresh_token
+    - _Requirements: 4.2, 4.3_
+  - [x] 5.3 在 `backend/routers/oauth.py` 中新增 `/api/oauth/me` 端点
+    - 从 `request.state.user` 获取用户信息（由中间件填充）
+    - 查询本地数据库返回完整用户信息
+    - _Requirements: 2.2_
+  - [x] 5.4 移除 `backend/routers/auth.py` 中的 login、register、refresh 端点
+    - 保留文件但只保留 `/api/auth/me` 端点（重定向到 `/api/oauth/me`）或直接移除整个 router
+    - 从 `backend/main.py` 中移除 auth router 的注册
+    - _Requirements: 1.1, 1.2, 1.3_
+  - [ ]* 5.5 编写 User Sync 属性测试
+    - **Property 5: User sync invariant**
+    - **Validates: Requirements 5.2, 5.3, 5.4**
+
+- [x] 6. 配置和清理
+  - [x] 6.1 更新 `backend/config.py`
+    - 移除 JWT 相关配置（secret_key, algorithm, expire 等）
+    - 新增 `token_cache_ttl` 配置项
+    - 更新 `config.dev.yaml` 和 `config.prod.yaml`
+    - _Requirements: 7.1, 7.2, 7.3_
+  - [x] 6.2 移除不再需要的后端文件
+    - 删除 `backend/services/jwt_service.py`
+    - 精简 `backend/services/auth_service.py`（移除 register_user, login_user, refresh_tokens）
+    - 移除 `backend/schemas/auth.py` 中不再需要的 schema（UserRegister, UserLogin, TokenRefresh, TokenPayload）
+    - _Requirements: 1.1, 2.3_
+
+- [x] 7. 前端适配
+  - [x] 7.1 更新 `web/apps/lq_label/src/services/api.ts` 中的 token refresh 逻辑
+    - 将 refresh 端点从 `/api/auth/refresh` 改为 `/api/oauth/refresh`
+    - 更新 request interceptor 中跳过 token 的路径列表
+    - _Requirements: 4.1, 4.4_
+  - [x] 7.2 改造 `web/apps/lq_label/src/components/login-form/login-form.tsx`
+    - 移除本地登录表单，页面加载时自动发起 SSO 登录
+    - 保留加载状态和错误提示
+    - _Requirements: 6.1, 1.4_
+  - [x] 7.3 移除注册相关前端代码
+    - 移除 `register-form` 组件
+    - 从 `app.tsx` 路由中移除 `/register` 路由
+    - 从 `auth-service.ts` 中移除 `register` 函数
+    - _Requirements: 1.1, 1.4_
+  - [x] 7.4 更新 `web/apps/lq_label/src/services/auth-service.ts`
+    - 移除 `login`、`register`、`refreshToken` 函数
+    - 更新 `getCurrentUser` 端点为 `/api/oauth/me`
+    - _Requirements: 1.1, 2.2_
+
+- [x] 8. Final checkpoint - 全面验证
+  - Ensure all tests pass, ask the user if questions arise.
+
+## Notes
+
+- Tasks marked with `*` are optional and can be skipped for faster MVP
+- Each task references specific requirements for traceability
+- Checkpoints ensure incremental validation
+- Property tests validate universal correctness properties
+- Unit tests validate specific examples and edge cases
+- 前端改动较小，主要是端点 URL 变更和登录页简化
+- 数据库无 schema 变更，用户表结构保持不变

README.md

@@ -0,0 +1,2 @@
+# 标注平台使用流程
+

backend/config.dev.yaml

@@ -1,12 +1,5 @@
 # 开发环境配置
 
-# JWT 配置
-jwt:
-  secret_key: "dev-secret-key-for-local-development"
-  algorithm: "HS256"
-  access_token_expire_minutes: 60
-  refresh_token_expire_days: 7
-
 # OAuth 2.0 单点登录配置
 oauth:
   enabled: true
@@ -21,6 +14,7 @@ oauth:
   token_endpoint: "/oauth/token"
   userinfo_endpoint: "/oauth/userinfo"
   revoke_endpoint: "/oauth/revoke"
+  token_cache_ttl: 300  # Token 缓存 TTL（秒）
 
 # 数据库配置 (MySQL)
 database:

backend/config.prod.yaml

@@ -1,13 +1,5 @@
 # 生产环境配置
 
-# JWT 配置
-jwt:
-  # 与开发环境保持一致，确保 token 兼容
-  secret_key: "dev-secret-key-for-local-development"
-  algorithm: "HS256"
-  access_token_expire_minutes: 60
-  refresh_token_expire_days: 7
-
 # OAuth 2.0 单点登录配置
 oauth:
   enabled: true
@@ -22,6 +14,7 @@ oauth:
   token_endpoint: "/oauth/token"
   userinfo_endpoint: "/oauth/userinfo"
   revoke_endpoint: "/oauth/revoke"
+  token_cache_ttl: 300  # Token 缓存 TTL（秒）
 
 # 数据库配置 (MySQL)
 database:

backend/config.py

@@ -1,14 +1,12 @@
 """
 Application configuration module.
-Manages JWT and OAuth settings from YAML configuration file.
+Manages OAuth/SSO settings from YAML configuration file.
 Supports dev/prod environments via APP_ENV environment variable.
 """
 import os
-import secrets
 import logging
 import yaml
 from pathlib import Path
-from typing import Dict, Any
 
 logger = logging.getLogger(__name__)
 
@@ -18,11 +16,11 @@ def get_config_path() -> Path:
     根据 APP_ENV 环境变量获取配置文件路径
     APP_ENV=prod -> config.prod.yaml
     APP_ENV=dev -> config.dev.yaml
-    默认 -> config.yaml (兼容旧配置)
+    默认 -> config.dev.yaml
     """
     app_env = os.getenv("APP_ENV", "").lower()
     base_path = Path(__file__).parent
-    
+
     if app_env == "prod":
         config_file = base_path / "config.prod.yaml"
         logger.info("使用生产环境配置: config.prod.yaml")
@@ -30,51 +28,40 @@ def get_config_path() -> Path:
         config_file = base_path / "config.dev.yaml"
         logger.info("使用开发环境配置: config.dev.yaml")
     else:
-        # 兼容旧的 config.yaml
         print("默认使用开发环境")
         config_file = base_path / "config.dev.yaml"
         if app_env:
-            logger.warning(f"未知的 APP_ENV 值: {app_env}，使用默认 config.yaml")
-    
+            logger.warning(f"未知的 APP_ENV 值: {app_env}，使用默认 config.dev.yaml")
+
     return config_file
 
 
 class Settings:
-    """Application settings loaded from config.yaml."""
-    
+    """Application settings loaded from config YAML."""
+
     def __init__(self):
         """Load configuration from YAML file."""
         config_path = get_config_path()
-        
+
         if not config_path.exists():
             raise FileNotFoundError(f"配置文件不存在: {config_path}")
-        
+
         with open(config_path, 'r', encoding='utf-8') as f:
             config = yaml.safe_load(f)
-        
-        # 记录当前环境（统一转小写）
+
         self.APP_ENV = os.getenv("APP_ENV", "default").lower()
         print(f"[Config] APP_ENV={self.APP_ENV}, 配置文件={config_path}")
-        
-        # JWT Settings
-        jwt_config = config.get('jwt', {})
-        self.JWT_SECRET_KEY = jwt_config.get('secret_key', secrets.token_urlsafe(32))
-        self.JWT_ALGORITHM = jwt_config.get('algorithm', 'HS256')
-        self.ACCESS_TOKEN_EXPIRE_MINUTES = jwt_config.get('access_token_expire_minutes', 15)
-        self.REFRESH_TOKEN_EXPIRE_DAYS = jwt_config.get('refresh_token_expire_days', 7)
-        
+
         # Database Settings (MySQL only)
         db_config = config.get('database', {})
-        
-        # MySQL Settings
         mysql_config = db_config.get('mysql', {})
         self.MYSQL_HOST = mysql_config.get('host', 'localhost')
         self.MYSQL_PORT = mysql_config.get('port', 3306)
         self.MYSQL_USER = mysql_config.get('user', 'root')
         self.MYSQL_PASSWORD = mysql_config.get('password', '')
         self.MYSQL_DATABASE = mysql_config.get('database', 'annotation_platform')
-        
-        # OAuth Settings
+
+        # OAuth/SSO Settings
         oauth_config = config.get('oauth', {})
         self.OAUTH_ENABLED = oauth_config.get('enabled', False)
         self.OAUTH_BASE_URL = oauth_config.get('base_url', '')
@@ -82,24 +69,21 @@ class Settings:
         self.OAUTH_CLIENT_SECRET = oauth_config.get('client_secret', '')
         self.OAUTH_REDIRECT_URI = oauth_config.get('redirect_uri', '')
         self.OAUTH_SCOPE = oauth_config.get('scope', 'profile email')
-        
+
         # OAuth Endpoints
         self.OAUTH_AUTHORIZE_ENDPOINT = oauth_config.get('authorize_endpoint', '/oauth/authorize')
         self.OAUTH_TOKEN_ENDPOINT = oauth_config.get('token_endpoint', '/oauth/token')
         self.OAUTH_USERINFO_ENDPOINT = oauth_config.get('userinfo_endpoint', '/oauth/userinfo')
         self.OAUTH_REVOKE_ENDPOINT = oauth_config.get('revoke_endpoint', '/oauth/revoke')
-        
+
+        # Token Cache TTL (seconds)
+        self.TOKEN_CACHE_TTL = oauth_config.get('token_cache_ttl', 300)
+
         # Server Settings
         server_config = config.get('server', {})
         self.SERVER_HOST = server_config.get('host', '0.0.0.0')
         self.SERVER_PORT = server_config.get('port', 8000)
         self.SERVER_RELOAD = server_config.get('reload', True)
-        
-        # Warn if using default JWT secret in production
-        if self.APP_ENV == "prod" and self.JWT_SECRET_KEY in ['your-secret-key-here', 'CHANGE_THIS_TO_A_SECURE_RANDOM_KEY']:
-            logger.warning("生产环境使用默认 JWT_SECRET_KEY，请立即修改 config.prod.yaml！")
-        elif self.JWT_SECRET_KEY == 'your-secret-key-here':
-            logger.warning(f"使用默认 JWT_SECRET_KEY，生产环境请修改配置文件！(当前环境: {self.APP_ENV})")
 
 
 # Create settings instance

backend/main.py

@@ -1,14 +1,22 @@
 """
 FastAPI application entry point.
-Provides RESTful API for the annotation platform with JWT authentication.
+Provides RESTful API for the annotation platform with SSO authentication.
 """
+import logging
 from fastapi import FastAPI
 from fastapi.middleware.cors import CORSMiddleware
 from contextlib import asynccontextmanager
 from database import init_database
-from routers import project, task, annotation, auth, oauth, user, template, statistics, export, external
+from routers import project, task, annotation, oauth, user, template, statistics, export, external
 from middleware.auth_middleware import AuthMiddleware
 
+# 配置日志
+logging.basicConfig(
+    level=logging.DEBUG,
+    format='%(asctime)s - %(name)s - %(levelname)s - %(message)s',
+    datefmt='%Y-%m-%d %H:%M:%S'
+)
+
 
 @asynccontextmanager
 async def lifespan(app: FastAPI):
@@ -25,7 +33,7 @@ async def lifespan(app: FastAPI):
 # Create FastAPI application instance
 app = FastAPI(
     title="Annotation Platform API",
-    description="RESTful API for data annotation management with JWT authentication",
+    description="RESTful API for data annotation management with SSO authentication",
     version="1.0.0",
     lifespan=lifespan
 )
@@ -46,7 +54,6 @@ app.add_middleware(
 app.add_middleware(AuthMiddleware)
 
 # Include routers
-app.include_router(auth.router)
 app.include_router(oauth.router)
 app.include_router(project.router)
 app.include_router(task.router)

backend/middleware/auth_middleware.py

@@ -1,57 +1,67 @@
 """
-Authentication Middleware for JWT token verification.
-Validates JWT tokens and attaches user info to request state.
+Authentication Middleware for SSO token verification.
+Validates SSO tokens via the SSO center's userinfo endpoint,
+with an in-memory cache to reduce external calls.
 """
+import logging
 from fastapi import Request, HTTPException, status
 from fastapi.responses import JSONResponse
 from starlette.middleware.base import BaseHTTPMiddleware
-from services.jwt_service import JWTService
-import jwt
+from services.token_cache_service import TokenCacheService
+from services.oauth_service import OAuthService
+from config import settings
+
+logger = logging.getLogger(__name__)
+
+# 全局 token 缓存实例
+# SSO token 有效期 600 秒，缓存设置为 550 秒（留 50 秒余量）
+token_cache = TokenCacheService(
+    ttl_seconds=getattr(settings, 'TOKEN_CACHE_TTL', 550)
+)
 
 
 class AuthMiddleware(BaseHTTPMiddleware):
     """
-    Authentication middleware for JWT token verification.
-    Validates JWT tokens and attaches user info to request state.
+    SSO Token 认证中间件。
+    先查本地缓存，未命中则调用 SSO userinfo 端点验证。
     """
-    
-    # Public endpoints that don't require authentication
+
     PUBLIC_PATHS = {
         "/",
         "/health",
         "/docs",
         "/openapi.json",
         "/redoc",
-        "/api/auth/register",
-        "/api/auth/login",
-        "/api/auth/refresh",
         "/api/oauth/status",
         "/api/oauth/login",
-        "/api/oauth/callback"
+        "/api/oauth/callback",
+        "/api/oauth/refresh",
     }
-    
+
     async def dispatch(self, request: Request, call_next):
-        """
-        Process each request through authentication.
-        
-        Args:
-            request: FastAPI Request object
-            call_next: Next middleware or route handler
-            
-        Returns:
-            Response from next handler or error response
-        """
         # Skip authentication for public paths
+        logger.debug(f"AuthMiddleware: path={request.url.path}, method={request.method}")
         if request.url.path in self.PUBLIC_PATHS:
+            logger.debug(f"Skipping auth for public path: {request.url.path}")
             return await call_next(request)
-        
+
         # Skip authentication for OPTIONS requests (CORS preflight)
         if request.method == "OPTIONS":
             return await call_next(request)
-        
+
+        # Check if OAuth/SSO is enabled
+        if not settings.OAUTH_ENABLED:
+            return JSONResponse(
+                status_code=status.HTTP_503_SERVICE_UNAVAILABLE,
+                content={
+                    "detail": "SSO 认证未配置",
+                    "error_type": "sso_not_configured"
+                }
+            )
+
         # Extract token from Authorization header
         auth_header = request.headers.get("Authorization")
-        
+
         if not auth_header:
             return JSONResponse(
                 status_code=status.HTTP_401_UNAUTHORIZED,
@@ -60,7 +70,7 @@ class AuthMiddleware(BaseHTTPMiddleware):
                     "error_type": "missing_token"
                 }
             )
-        
+
         # Verify Bearer token format
         parts = auth_header.split()
         if len(parts) != 2 or parts[0].lower() != "bearer":
@@ -71,51 +81,67 @@ class AuthMiddleware(BaseHTTPMiddleware):
                     "error_type": "invalid_token_format"
                 }
             )
-        
-        token = parts[1]
-        
+
+        sso_token = parts[1]
+
         try:
-            # Verify and decode token
-            payload = JWTService.verify_token(token, "access")
-            
-            if not payload:
-                return JSONResponse(
-                    status_code=status.HTTP_401_UNAUTHORIZED,
-                    content={
-                        "detail": "无效的认证令牌",
-                        "error_type": "invalid_token"
-                    }
-                )
-            
+            # 1. 先查本地缓存
+            user_info = token_cache.get(sso_token)
+
+            if user_info is None:
+                # 2. 缓存未命中，调 SSO profile 验证（含角色信息）
+                user_info = await OAuthService.verify_sso_token(sso_token)
+
+                # 3. 同步用户到本地数据库（更新角色）
+                try:
+                    OAuthService.sync_user_from_oauth(user_info)
+                except Exception as sync_err:
+                    logger.warning(f"用户同步失败（不影响认证）: {sync_err}")
+
+                # 4. 写入缓存
+                token_cache.set(sso_token, user_info)
+
+            # 提取用户信息
+            user_id = user_info.get("id") or user_info.get("sub")
+            username = (
+                user_info.get("username")
+                or user_info.get("preferred_username")
+                or user_info.get("name")
+            )
+            email = user_info.get("email", "")
+            role = user_info.get("role", "viewer")
+
             # Attach user info to request state
             request.state.user = {
-                "id": payload["sub"],
-                "username": payload["username"],
-                "email": payload["email"],
-                "role": payload["role"]
+                "id": str(user_id),
+                "username": username,
+                "email": email,
+                "role": role,
             }
-            
-            # Continue to route handler
+
             response = await call_next(request)
             return response
-            
-        except jwt.ExpiredSignatureError:
-            return JSONResponse(
-                status_code=status.HTTP_401_UNAUTHORIZED,
-                content={
-                    "detail": "认证令牌已过期",
-                    "error_type": "token_expired"
-                }
-            )
-        except jwt.InvalidTokenError:
+
+        except HTTPException as e:
+            error_type = "invalid_token"
+            if e.status_code == 503:
+                error_type = "sso_unavailable"
+            elif e.status_code == 401:
+                # SSO 返回 401 说明 token 过期或无效，统一标记为 token_expired
+                # 让前端有机会用 refresh_token 刷新
+                error_type = "token_expired"
+                # 同时清除本地缓存中的过期 token
+                token_cache.invalidate(sso_token)
+
             return JSONResponse(
-                status_code=status.HTTP_401_UNAUTHORIZED,
+                status_code=e.status_code,
                 content={
-                    "detail": "无效的认证令牌",
-                    "error_type": "invalid_token"
+                    "detail": e.detail,
+                    "error_type": error_type
                 }
             )
         except Exception as e:
+            logger.error(f"认证过程发生错误: {e}")
             return JSONResponse(
                 status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
                 content={
@@ -128,34 +154,34 @@ class AuthMiddleware(BaseHTTPMiddleware):
 def require_role(*allowed_roles: str):
     """
     Decorator to check user role.
-    
+
     Usage:
         @require_role("admin", "annotator")
         async def my_endpoint(request: Request):
             ...
-    
+
     Args:
         allowed_roles: Tuple of allowed role names
-        
+
     Returns:
         Decorator function
     """
     def decorator(func):
         async def wrapper(request: Request, *args, **kwargs):
             user = getattr(request.state, "user", None)
-            
+
             if not user:
                 raise HTTPException(
                     status_code=status.HTTP_401_UNAUTHORIZED,
                     detail="未认证"
                 )
-            
+
             if user["role"] not in allowed_roles:
                 raise HTTPException(
                     status_code=status.HTTP_403_FORBIDDEN,
                     detail="权限不足"
                 )
-            
+
             return await func(request, *args, **kwargs)
         return wrapper
     return decorator

backend/routers/auth.py

@@ -1,124 +1,7 @@
 """
-Authentication API router.
-Provides endpoints for user registration, login, token refresh, and user info.
+Authentication API router (DEPRECATED).
+Local authentication has been removed. All authentication now goes through SSO.
+See routers/oauth.py for the active authentication endpoints.
 """
-from fastapi import APIRouter, HTTPException, status, Request
-from schemas.auth import (
-    UserRegister, UserLogin, TokenResponse,
-    TokenRefresh, UserResponse
-)
-from services.auth_service import AuthService
-
-router = APIRouter(
-    prefix="/api/auth",
-    tags=["authentication"]
-)
-
-
-@router.post("/register", status_code=status.HTTP_201_CREATED)
-async def register(user_data: UserRegister):
-    """
-    Register a new user.
-    
-    Args:
-        user_data: User registration data
-        
-    Returns:
-        Success message with user_id
-        
-    Raises:
-        HTTPException: 409 if username or email already exists
-        HTTPException: 400 if validation fails
-    """
-    user = AuthService.register_user(
-        username=user_data.username,
-        email=user_data.email,
-        password=user_data.password
-    )
-    
-    return {
-        "message": "用户注册成功",
-        "user_id": user.id
-    }
-
-
-@router.post("/login", response_model=TokenResponse)
-async def login(credentials: UserLogin):
-    """
-    Authenticate user and return JWT tokens.
-    
-    Args:
-        credentials: User login credentials
-        
-    Returns:
-        Access token, refresh token, and user info
-        
-    Raises:
-        HTTPException: 401 if credentials are invalid
-    """
-    result = AuthService.login_user(
-        username=credentials.username,
-        password=credentials.password
-    )
-    
-    return TokenResponse(
-        access_token=result["access_token"],
-        refresh_token=result["refresh_token"],
-        user=UserResponse(**result["user"])
-    )
-
-
-@router.post("/refresh", response_model=TokenResponse)
-async def refresh_token(token_data: TokenRefresh):
-    """
-    Refresh access token using refresh token.
-    
-    Args:
-        token_data: Refresh token
-        
-    Returns:
-        New access token and refresh token
-        
-    Raises:
-        HTTPException: 401 if refresh token is invalid or expired
-    """
-    result = AuthService.refresh_tokens(token_data.refresh_token)
-    
-    return TokenResponse(
-        access_token=result["access_token"],
-        refresh_token=result["refresh_token"],
-        user=UserResponse(**result["user"])
-    )
-
-
-@router.get("/me", response_model=UserResponse)
-async def get_current_user(request: Request):
-    """
-    Get current authenticated user info.
-    
-    Args:
-        request: FastAPI Request with user info in state
-        
-    Returns:
-        Current user information
-        
-    Raises:
-        HTTPException: 401 if not authenticated
-    """
-    user_data = getattr(request.state, "user", None)
-    
-    if not user_data:
-        raise HTTPException(
-            status_code=status.HTTP_401_UNAUTHORIZED,
-            detail="未认证"
-        )
-    
-    user = AuthService.get_current_user(user_data["id"])
-    
-    return UserResponse(
-        id=user.id,
-        username=user.username,
-        email=user.email,
-        role=user.role,
-        created_at=user.created_at
-    )
+# This file is intentionally left minimal.
+# All auth endpoints have been moved to /api/oauth/*

backend/routers/oauth.py

@@ -1,16 +1,18 @@
 """
 OAuth 2.0 认证路由
-处理 OAuth 登录流程
+处理 SSO 登录流程、token 刷新和用户信息查询。
+所有认证统一走 SSO，不再本地签发 JWT。
 """
-from fastapi import APIRouter, HTTPException, Query
-from fastapi.responses import RedirectResponse
+import logging
+from fastapi import APIRouter, HTTPException, Query, Request, status
 from pydantic import BaseModel
 from typing import Optional
 from config import settings
 from services.oauth_service import OAuthService
-from services.jwt_service import JWTService
-from schemas.auth import TokenResponse, UserResponse
+from services.auth_service import AuthService
+from middleware.auth_middleware import token_cache
 
+logger = logging.getLogger(__name__)
 router = APIRouter(prefix="/api/oauth", tags=["oauth"])
 
 
@@ -20,106 +22,180 @@ class OAuthLoginResponse(BaseModel):
     state: str
 
 
+class SSOTokenResponse(BaseModel):
+    """SSO Token 响应（透传 SSO 中心的 token）"""
+    access_token: str
+    refresh_token: str
+    token_type: str = "bearer"
+    user: dict
+
+
+class RefreshRequest(BaseModel):
+    """Token 刷新请求"""
+    refresh_token: str
+
+
+class UserResponse(BaseModel):
+    """用户信息响应"""
+    id: str
+    username: str
+    email: str
+    role: str
+    created_at: str
+
+
 @router.get("/login", response_model=OAuthLoginResponse)
 async def oauth_login():
     """
-    启动 OAuth 登录流程
-    
-    生成授权 URL 和 state 参数，前端需要保存 state 并重定向到授权 URL
-    
-    Returns:
-        包含授权 URL 和 state 的响应
+    启动 OAuth 登录流程。
+    生成授权 URL 和 state 参数，前端需要保存 state 并重定向到授权 URL。
     """
     if not settings.OAUTH_ENABLED:
-        raise HTTPException(status_code=400, detail="OAuth 登录未启用")
-    
-    # 生成 state 参数
+        raise HTTPException(status_code=503, detail="SSO 认证未配置")
+
     state = OAuthService.generate_state()
-    
-    # 构建授权 URL
     authorization_url = OAuthService.get_authorization_url(state)
-    
+
     return OAuthLoginResponse(
         authorization_url=authorization_url,
         state=state
     )
 
 
-@router.get("/callback", response_model=TokenResponse)
+@router.get("/callback")
 async def oauth_callback(
     code: str = Query(..., description="OAuth 授权码"),
     state: str = Query(..., description="State 参数"),
 ):
     """
-    OAuth 回调端点
-    
-    处理 OAuth 认证中心的回调，用授权码换取令牌，获取用户信息，
-    并创建或更新本地用户记录
-    
-    Args:
-        code: OAuth 授权码
-        state: State 参数（前端需要验证）
-        
-    Returns:
-        JWT tokens 和用户信息
+    OAuth 回调端点。
+    用授权码换取 SSO token，获取用户信息并同步到本地数据库，
+    直接返回 SSO 的 access_token 和 refresh_token（不再本地签发 JWT）。
     """
-    if not settings.OAUTH_ENABLED:
-        raise HTTPException(status_code=400, detail="OAuth 登录未启用")
+    logger.info(f"OAuth callback received: code={code[:10]}..., state={state[:10]}...")
     
+    if not settings.OAUTH_ENABLED:
+        raise HTTPException(status_code=503, detail="SSO 认证未配置")
+
     try:
-        # 1. 用授权码换取访问令牌
+        # 1. 用授权码换取 SSO token
+        logger.debug("Exchanging code for token...")
         token_data = await OAuthService.exchange_code_for_token(code)
         access_token = token_data.get("access_token")
-        
+        refresh_token = token_data.get("refresh_token", "")
+
         if not access_token:
             raise HTTPException(status_code=400, detail="未能获取访问令牌")
-        
-        # 2. 使用访问令牌获取用户信息
-        oauth_user_info = await OAuthService.get_user_info(access_token)
-        
-        # 3. 同步用户到本地数据库
-        user = OAuthService.sync_user_from_oauth(oauth_user_info)
-        
-        # 4. 生成本地 JWT tokens
-        user_data = {
-            "id": user.id,
-            "username": user.username,
-            "email": user.email,
-            "role": user.role
-        }
-        
-        jwt_access_token = JWTService.create_access_token(user_data)
-        jwt_refresh_token = JWTService.create_refresh_token(user_data)
-        
-        # 5. 返回 tokens 和用户信息
-        return TokenResponse(
-            access_token=jwt_access_token,
-            refresh_token=jwt_refresh_token,
+
+        # 2. 使用 SSO token 获取完整用户信息（含角色）
+        logger.debug("Verifying SSO token and getting user info...")
+        try:
+            user_info = await OAuthService.verify_sso_token(access_token)
+            logger.debug(f"User info: {user_info.get('username')}, role: {user_info.get('role')}")
+        except HTTPException as e:
+            logger.error(f"verify_sso_token failed: status={e.status_code}, detail={e.detail}")
+            raise
+        except Exception as e:
+            logger.error(f"verify_sso_token unexpected error: {e}", exc_info=True)
+            raise
+
+        # 3. 同步用户到本地数据库（含角色映射）
+        logger.debug("Syncing user to local database...")
+        try:
+            user = OAuthService.sync_user_from_oauth(user_info)
+            logger.debug(f"User synced: id={user.id}, username={user.username}, role={user.role}")
+        except Exception as e:
+            logger.error(f"sync_user_from_oauth failed: {e}", exc_info=True)
+            raise
+
+        # 4. 缓存 token → 用户信息映射
+        token_cache.set(access_token, user_info)
+
+        # 5. 直接返回 SSO token（不再本地签发 JWT）
+        logger.info(f"OAuth login successful for user: {user.username}")
+        return SSOTokenResponse(
+            access_token=access_token,
+            refresh_token=refresh_token,
             token_type="bearer",
-            user=UserResponse(
-                id=user.id,
-                username=user.username,
-                email=user.email,
-                role=user.role,
-                created_at=user.created_at
-            )
+            user={
+                "id": user.id,
+                "username": user.username,
+                "email": user.email,
+                "role": user.role,
+                "created_at": str(user.created_at)
+            }
         )
-        
+
+    except HTTPException:
+        raise
     except Exception as e:
+        logger.error(f"OAuth callback error: {e}", exc_info=True)
         raise HTTPException(
             status_code=400,
             detail=f"OAuth 登录失败: {str(e)}"
         )
 
 
-@router.get("/status")
-async def oauth_status():
+@router.post("/refresh")
+async def oauth_refresh(request_body: RefreshRequest):
     """
-    获取 OAuth 配置状态
+    Token 刷新端点。
+    将 refresh 请求转发到 SSO 中心，返回新的 token。
+    """
+    logger.info(f"Token refresh requested, refresh_token={request_body.refresh_token[:20]}...")
     
-    Returns:
-        OAuth 是否启用及相关配置信息
+    if not settings.OAUTH_ENABLED:
+        raise HTTPException(status_code=503, detail="SSO 认证未配置")
+
+    try:
+        token_data = await OAuthService.refresh_sso_token(request_body.refresh_token)
+        logger.info("Token refresh successful")
+        
+        return {
+            "access_token": token_data.get("access_token"),
+            "refresh_token": token_data.get("refresh_token", ""),
+            "token_type": token_data.get("token_type", "bearer"),
+        }
+    except HTTPException as e:
+        logger.error(f"Token refresh failed: status={e.status_code}, detail={e.detail}")
+        raise
+    except Exception as e:
+        logger.error(f"Token refresh unexpected error: {e}", exc_info=True)
+        raise HTTPException(
+            status_code=400,
+            detail=f"Token 刷新失败: {str(e)}"
+        )
+
+
+@router.get("/me", response_model=UserResponse)
+async def get_current_user(request: Request):
+    """
+    获取当前认证用户信息。
+    用户信息由 AuthMiddleware 从 SSO 验证后填充到 request.state。
     """
+    user_data = getattr(request.state, "user", None)
+
+    if not user_data:
+        raise HTTPException(
+            status_code=status.HTTP_401_UNAUTHORIZED,
+            detail="未认证"
+        )
+
+    # 从本地数据库获取完整用户信息
+    user = AuthService.get_current_user(user_data["id"])
+
+    return UserResponse(
+        id=user.id,
+        username=user.username,
+        email=user.email,
+        role=user.role,
+        created_at=str(user.created_at)
+    )
+
+
+@router.get("/status")
+async def oauth_status():
+    """获取 OAuth 配置状态"""
     return {
         "enabled": settings.OAUTH_ENABLED,
         "provider": "SSO" if settings.OAUTH_ENABLED else None,

backend/schemas/auth.py

@@ -1,25 +1,11 @@
 """
 Authentication schemas for request/response validation.
-Defines Pydantic models for user registration, login, and token management.
+Simplified for SSO-only authentication.
 """
-from pydantic import BaseModel, EmailStr, Field
-from typing import Optional, TYPE_CHECKING
+from pydantic import BaseModel, Field
 from datetime import datetime
 
 
-class UserRegister(BaseModel):
-    """User registration request schema."""
-    username: str = Field(..., min_length=3, max_length=50, description="用户名，3-50个字符")
-    email: EmailStr = Field(..., description="有效的邮箱地址")
-    password: str = Field(..., min_length=8, max_length=100, description="密码，至少8个字符")
-
-
-class UserLogin(BaseModel):
-    """User login request schema."""
-    username: str = Field(..., description="用户名")
-    password: str = Field(..., description="密码")
-
-
 class UserResponse(BaseModel):
     """User response schema."""
     id: str = Field(..., description="用户ID")
@@ -27,30 +13,6 @@ class UserResponse(BaseModel):
     email: str = Field(..., description="邮箱")
     role: str = Field(..., description="用户角色")
     created_at: datetime = Field(..., description="创建时间")
-    
+
     class Config:
         from_attributes = True
-
-
-class TokenResponse(BaseModel):
-    """Token response schema."""
-    access_token: str = Field(..., description="访问令牌")
-    refresh_token: str = Field(..., description="刷新令牌")
-    token_type: str = Field(default="bearer", description="令牌类型")
-    user: UserResponse = Field(..., description="用户信息")
-
-
-class TokenRefresh(BaseModel):
-    """Token refresh request schema."""
-    refresh_token: str = Field(..., description="刷新令牌")
-
-
-class TokenPayload(BaseModel):
-    """JWT token payload schema."""
-    sub: str = Field(..., description="用户ID")
-    username: str = Field(..., description="用户名")
-    email: str = Field(..., description="邮箱")
-    role: str = Field(..., description="角色")
-    exp: datetime = Field(..., description="过期时间")
-    iat: datetime = Field(..., description="签发时间")
-    type: str = Field(..., description="令牌类型: access 或 refresh")

backend/services/__init__.py

@@ -2,11 +2,13 @@
 Business logic services package.
 """
 from .auth_service import AuthService
-from .jwt_service import JWTService
+from .oauth_service import OAuthService
+from .token_cache_service import TokenCacheService
 from .export_service import ExportService
 
 __all__ = [
     "AuthService",
-    "JWTService",
+    "OAuthService",
+    "TokenCacheService",
     "ExportService",
 ]

backend/services/auth_service.py

@@ -1,229 +1,27 @@
 """
-Authentication Service for user management and authentication.
-Handles user registration, login, token refresh, and user queries.
+Authentication Service for user queries.
+Local registration, login, and token refresh have been removed.
+All authentication now goes through SSO (see oauth_service.py).
 """
-import bcrypt
-import uuid
-from typing import Dict
 from database import get_db_connection
 from models import User
-from services.jwt_service import JWTService
 from fastapi import HTTPException, status
 
 
 class AuthService:
-    """Service for authentication operations."""
-    
-    @staticmethod
-    def register_user(username: str, email: str, password: str) -> User:
-        """
-        Register a new user.
-        
-        Args:
-            username: Unique username
-            email: Unique email address
-            password: Plain text password
-            
-        Returns:
-            Created User object
-            
-        Raises:
-            HTTPException: 409 if username or email already exists
-        """
-        with get_db_connection() as conn:
-            cursor = conn.cursor()
-            
-            # Check username uniqueness
-            cursor.execute(
-                "SELECT id FROM users WHERE username = ?",
-                (username,)
-            )
-            if cursor.fetchone():
-                raise HTTPException(
-                    status_code=status.HTTP_409_CONFLICT,
-                    detail="用户名已被使用"
-                )
-            
-            # Check email uniqueness
-            cursor.execute(
-                "SELECT id FROM users WHERE email = ?",
-                (email,)
-            )
-            if cursor.fetchone():
-                raise HTTPException(
-                    status_code=status.HTTP_409_CONFLICT,
-                    detail="邮箱已被使用"
-                )
-            
-            # Hash password
-            password_hash = bcrypt.hashpw(
-                password.encode('utf-8'),
-                bcrypt.gensalt()
-            ).decode('utf-8')
-            
-            # Create user
-            user_id = f"user_{uuid.uuid4().hex[:12]}"
-            cursor.execute("""
-                INSERT INTO users (
-                    id, username, email, password_hash, role
-                )
-                VALUES (?, ?, ?, ?, ?)
-            """, (user_id, username, email, password_hash, "annotator"))
-            
-            # Fetch created user
-            cursor.execute(
-                "SELECT * FROM users WHERE id = ?",
-                (user_id,)
-            )
-            row = cursor.fetchone()
-            return User.from_row(row)
-    
-    @staticmethod
-    def login_user(username: str, password: str) -> Dict:
-        """
-        Authenticate user and generate tokens.
-        
-        Args:
-            username: Username
-            password: Plain text password
-            
-        Returns:
-            Dict containing access_token, refresh_token, and user info
-            
-        Raises:
-            HTTPException: 401 if credentials are invalid
-        """
-        with get_db_connection() as conn:
-            cursor = conn.cursor()
-            
-            # Find user
-            cursor.execute(
-                "SELECT * FROM users WHERE username = ?",
-                (username,)
-            )
-            row = cursor.fetchone()
-            
-            if not row:
-                raise HTTPException(
-                    status_code=status.HTTP_401_UNAUTHORIZED,
-                    detail="用户名或密码错误"
-                )
-            
-            user = User.from_row(row)
-            
-            # Verify password
-            try:
-                password_valid = bcrypt.checkpw(
-                    password.encode('utf-8'),
-                    user.password_hash.encode('utf-8')
-                )
-            except (ValueError, TypeError) as e:
-                # Invalid salt or hash format - password hash is corrupted or not bcrypt
-                # This can happen if password was stored in plaintext or different format
-                raise HTTPException(
-                    status_code=status.HTTP_401_UNAUTHORIZED,
-                    detail="用户名或密码错误"
-                )
-            
-            if not password_valid:
-                raise HTTPException(
-                    status_code=status.HTTP_401_UNAUTHORIZED,
-                    detail="用户名或密码错误"
-                )
-            
-            # Generate tokens
-            user_data = {
-                "id": user.id,
-                "username": user.username,
-                "email": user.email,
-                "role": user.role,
-                "created_at": user.created_at
-            }
-            
-            access_token = JWTService.create_access_token(user_data)
-            refresh_token = JWTService.create_refresh_token(user_data)
-            
-            return {
-                "access_token": access_token,
-                "refresh_token": refresh_token,
-                "user": user_data
-            }
-    
-    @staticmethod
-    def refresh_tokens(refresh_token: str) -> Dict:
-        """
-        Refresh access token using refresh token.
-        
-        Args:
-            refresh_token: Valid refresh token
-            
-        Returns:
-            Dict containing new access_token and refresh_token
-            
-        Raises:
-            HTTPException: 401 if refresh token is invalid or expired
-        """
-        try:
-            payload = JWTService.verify_token(refresh_token, "refresh")
-            if not payload:
-                raise HTTPException(
-                    status_code=status.HTTP_401_UNAUTHORIZED,
-                    detail="无效的刷新令牌"
-                )
-            
-            user_id = payload["sub"]
-            
-            # Fetch user from database
-            with get_db_connection() as conn:
-                cursor = conn.cursor()
-                cursor.execute(
-                    "SELECT * FROM users WHERE id = ?",
-                    (user_id,)
-                )
-                row = cursor.fetchone()
-                
-                if not row:
-                    raise HTTPException(
-                        status_code=status.HTTP_401_UNAUTHORIZED,
-                        detail="用户不存在"
-                    )
-                
-                user = User.from_row(row)
-                user_data = {
-                    "id": user.id,
-                    "username": user.username,
-                    "email": user.email,
-                    "role": user.role,
-                    "created_at": user.created_at
-                }
-                
-                # Generate new tokens (token rotation)
-                new_access_token = JWTService.create_access_token(user_data)
-                new_refresh_token = JWTService.create_refresh_token(user_data)
-                
-                return {
-                    "access_token": new_access_token,
-                    "refresh_token": new_refresh_token,
-                    "user": user_data
-                }
-                
-        except Exception as e:
-            raise HTTPException(
-                status_code=status.HTTP_401_UNAUTHORIZED,
-                detail="刷新令牌已过期或无效，请重新登录"
-            )
-    
+    """Service for user query operations."""
+
     @staticmethod
     def get_current_user(user_id: str) -> User:
         """
         Get user by ID.
-        
+
         Args:
             user_id: User unique identifier
-            
+
         Returns:
             User object
-            
+
         Raises:
             HTTPException: 404 if user not found
         """
@@ -234,11 +32,11 @@ class AuthService:
                 (user_id,)
             )
             row = cursor.fetchone()
-            
+
             if not row:
                 raise HTTPException(
                     status_code=status.HTTP_404_NOT_FOUND,
                     detail="用户不存在"
                 )
-            
+
             return User.from_row(row)

backend/services/jwt_service.py

@@ -1,100 +0,0 @@
-"""
-JWT Service for token generation and validation.
-Handles creation and verification of access and refresh tokens.
-"""
-from datetime import datetime, timedelta
-from typing import Dict, Optional
-import jwt
-from config import settings
-
-
-class JWTService:
-    """Service for JWT token operations."""
-    
-    @staticmethod
-    def create_access_token(user_data: Dict) -> str:
-        """
-        Create access token with 15 minutes expiration.
-        
-        Args:
-            user_data: Dict containing user_id, username, email, role
-            
-        Returns:
-            Encoded JWT token string
-        """
-        expire = datetime.utcnow() + timedelta(
-            minutes=settings.ACCESS_TOKEN_EXPIRE_MINUTES
-        )
-        payload = {
-            "sub": user_data["id"],
-            "username": user_data["username"],
-            "email": user_data["email"],
-            "role": user_data["role"],
-            "exp": expire,
-            "iat": datetime.utcnow(),
-            "type": "access"
-        }
-        return jwt.encode(
-            payload,
-            settings.JWT_SECRET_KEY,
-            algorithm=settings.JWT_ALGORITHM
-        )
-    
-    @staticmethod
-    def create_refresh_token(user_data: Dict) -> str:
-        """
-        Create refresh token with 7 days expiration.
-        
-        Args:
-            user_data: Dict containing user_id
-            
-        Returns:
-            Encoded JWT token string
-        """
-        expire = datetime.utcnow() + timedelta(
-            days=settings.REFRESH_TOKEN_EXPIRE_DAYS
-        )
-        payload = {
-            "sub": user_data["id"],
-            "exp": expire,
-            "iat": datetime.utcnow(),
-            "type": "refresh"
-        }
-        return jwt.encode(
-            payload,
-            settings.JWT_SECRET_KEY,
-            algorithm=settings.JWT_ALGORITHM
-        )
-    
-    @staticmethod
-    def verify_token(token: str, token_type: str = "access") -> Optional[Dict]:
-        """
-        Verify and decode JWT token.
-        
-        Args:
-            token: JWT token string
-            token_type: Expected token type (access or refresh)
-            
-        Returns:
-            Decoded payload dict or None if invalid
-            
-        Raises:
-            jwt.ExpiredSignatureError: Token has expired
-            jwt.InvalidTokenError: Token is invalid
-        """
-        try:
-            payload = jwt.decode(
-                token,
-                settings.JWT_SECRET_KEY,
-                algorithms=[settings.JWT_ALGORITHM]
-            )
-            
-            # Verify token type
-            if payload.get("type") != token_type:
-                return None
-                
-            return payload
-        except jwt.ExpiredSignatureError:
-            raise
-        except jwt.InvalidTokenError:
-            raise

backend/services/oauth_service.py

@@ -1,15 +1,47 @@
 """
 OAuth 2.0 认证服务
-处理与 OAuth 认证中心的交互
+处理与 OAuth 认证中心的交互，包括 token 验证和刷新
 """
 import httpx
+import logging
 import secrets
 from typing import Dict, Any, Optional
 from datetime import datetime
+from fastapi import HTTPException, status
 from config import settings
 from models import User
 from database import get_db_connection
 
+logger = logging.getLogger(__name__)
+
+# SSO 角色 → 本地角色映射
+SSO_ROLE_MAPPING = {
+    "super_admin": "admin",
+    "label_admin": "admin",
+    "admin": "admin",
+    "labeler": "annotator",
+}
+DEFAULT_LOCAL_ROLE = "viewer"
+
+
+def map_sso_roles_to_local(sso_roles: list, is_superuser: bool = False) -> str:
+    """
+    将 SSO 角色列表映射为本地单一角色。
+    优先级: admin > annotator > viewer
+    """
+    if is_superuser:
+        return "admin"
+
+    local_role = DEFAULT_LOCAL_ROLE
+    for sso_role in sso_roles:
+        mapped = SSO_ROLE_MAPPING.get(sso_role)
+        if mapped == "admin":
+            return "admin"
+        if mapped == "annotator":
+            local_role = "annotator"
+
+    return local_role
+
 
 class OAuthService:
     """OAuth 2.0 认证服务"""
@@ -85,7 +117,9 @@ class OAuthService:
             # 处理不同的响应格式
             if "access_token" in data:
                 return data
-            elif data.get("code") == 0 and "data" in data:
+            # 处理包装格式 {"code": 0, "data": {...}} 或 {"code": "000000", "data": {...}}
+            code = data.get("code")
+            if (code == 0 or code == "000000") and "data" in data:
                 return data["data"]
             else:
                 raise Exception(f"无效的令牌响应格式: {data}")
@@ -120,7 +154,9 @@ class OAuthService:
             # 处理不同的响应格式
             if "sub" in data or "id" in data:
                 return data
-            elif data.get("code") == 0 and "data" in data:
+            # 处理包装格式 {"code": 0, "data": {...}} 或 {"code": "000000", "data": {...}}
+            code = data.get("code")
+            if (code == 0 or code == "000000") and "data" in data:
                 return data["data"]
             else:
                 raise Exception(f"无效的用户信息响应格式: {data}")
@@ -129,7 +165,7 @@ class OAuthService:
     def sync_user_from_oauth(oauth_user_info: Dict[str, Any]) -> User:
         """
         从 OAuth 用户信息同步到本地数据库
-        如果用户不存在则创建，如果存在则更新
+        如果用户不存在则创建，如果存在则更新（包括角色）
         
         Args:
             oauth_user_info: OAuth 返回的用户信息
@@ -151,42 +187,43 @@ class OAuthService:
             if not username:
                 raise ValueError("OAuth 用户信息缺少用户名字段")
             
+            # 计算本地角色
+            sso_roles = oauth_user_info.get("sso_roles") or oauth_user_info.get("roles", [])
+            is_superuser = bool(oauth_user_info.get("is_superuser", False))
+            role = oauth_user_info.get("role") or map_sso_roles_to_local(sso_roles, is_superuser)
+            
             # 查找是否已存在该 OAuth 用户
             cursor.execute(
-                "SELECT * FROM users WHERE oauth_provider = ? AND oauth_id = ?",
+                "SELECT * FROM users WHERE oauth_provider = %s AND oauth_id = %s",
                 ("sso", oauth_id)
             )
             row = cursor.fetchone()
             
             if row:
-                # 用户已存在，更新信息
+                # 用户已存在，更新信息（包括角色）
                 user = User.from_row(row)
                 
-                # 更新用户名和邮箱（如果有变化）
                 cursor.execute("""
                     UPDATE users 
-                    SET username = ?, email = ?
-                    WHERE id = ?
-                """, (username, email, user.id))
+                    SET username = %s, email = %s, role = %s
+                    WHERE id = %s
+                """, (username, email, role, user.id))
                 
                 conn.commit()
                 
                 # 重新查询更新后的用户
-                cursor.execute("SELECT * FROM users WHERE id = ?", (user.id,))
+                cursor.execute("SELECT * FROM users WHERE id = %s", (user.id,))
                 row = cursor.fetchone()
                 return User.from_row(row)
             else:
                 # 新用户，创建记录
                 user_id = f"user_{datetime.now().strftime('%Y%m%d%H%M%S')}_{secrets.token_hex(4)}"
                 
-                # 暂时所有用户都是 annotator 角色（SSO 未提供角色信息）
-                role = "annotator"
-                
                 cursor.execute("""
                     INSERT INTO users (
                         id, username, email, password_hash, role,
                         oauth_provider, oauth_id, created_at
-                    ) VALUES (?, ?, ?, ?, ?, ?, ?, ?)
+                    ) VALUES (%s, %s, %s, %s, %s, %s, %s, %s)
                 """, (
                     user_id,
                     username,
@@ -201,6 +238,148 @@ class OAuthService:
                 conn.commit()
                 
                 # 查询新创建的用户
-                cursor.execute("SELECT * FROM users WHERE id = ?", (user_id,))
+                cursor.execute("SELECT * FROM users WHERE id = %s", (user_id,))
                 row = cursor.fetchone()
                 return User.from_row(row)
+
+    @staticmethod
+    async def verify_sso_token(access_token: str) -> Dict[str, Any]:
+        """
+        通过 SSO 验证 token 并获取用户信息（含角色）。
+        
+        使用 /api/v1/system/users/profile 端点获取完整用户信息，
+        包括 roles 列表和 is_superuser 标记，然后映射为本地角色。
+        
+        Args:
+            access_token: SSO 访问令牌
+            
+        Returns:
+            用户信息字典 {id, username, email, role, ...}
+            
+        Raises:
+            HTTPException(401): token 无效
+            HTTPException(503): SSO 中心不可用
+        """
+        profile_url = f"{settings.OAUTH_BASE_URL}/api/v1/system/users/profile"
+        
+        async with httpx.AsyncClient(timeout=10.0) as client:
+            try:
+                response = await client.get(
+                    profile_url,
+                    headers={"Authorization": f"Bearer {access_token}"}
+                )
+            except httpx.RequestError:
+                raise HTTPException(
+                    status_code=status.HTTP_503_SERVICE_UNAVAILABLE,
+                    detail="SSO 认证中心不可用"
+                )
+        
+        if response.status_code == 401:
+            raise HTTPException(
+                status_code=status.HTTP_401_UNAUTHORIZED,
+                detail="无效的访问令牌"
+            )
+        
+        if response.status_code != 200:
+            raise HTTPException(
+                status_code=status.HTTP_401_UNAUTHORIZED,
+                detail=f"SSO 验证失败 ({response.status_code})"
+            )
+        
+        data = response.json()
+        logger.debug(f"SSO profile response: {data}")
+        
+        # 处理包装格式 {"code": 0, "data": {...}} 或 {"code": "000000", "data": {...}}
+        code = data.get("code")
+        if (code == 0 or code == "000000") and "data" in data:
+            profile = data["data"]
+        elif "id" in data or "username" in data:
+            profile = data
+        else:
+            logger.error(f"Invalid profile response format: {data}")
+            raise HTTPException(
+                status_code=status.HTTP_401_UNAUTHORIZED,
+                detail="无效的访问令牌"
+            )
+        
+        # 提取角色信息并映射
+        sso_roles = profile.get("roles", [])
+        is_superuser = bool(profile.get("is_superuser", False))
+        local_role = map_sso_roles_to_local(sso_roles, is_superuser)
+        
+        logger.info(
+            f"SSO 用户 {profile.get('username')}: "
+            f"sso_roles={sso_roles}, is_superuser={is_superuser} → local_role={local_role}"
+        )
+        
+        # 返回统一格式的用户信息
+        return {
+            "id": profile.get("id"),
+            "username": profile.get("username"),
+            "email": profile.get("email", ""),
+            "role": local_role,
+            "sso_roles": sso_roles,
+            "is_superuser": is_superuser,
+        }
+
+    @staticmethod
+    async def refresh_sso_token(refresh_token: str) -> Dict[str, Any]:
+        """
+        向 SSO 中心刷新 token。
+        
+        Args:
+            refresh_token: SSO 刷新令牌
+            
+        Returns:
+            新的 token 信息 {access_token, refresh_token, ...}
+            
+        Raises:
+            HTTPException(401): refresh_token 无效
+            HTTPException(503): SSO 中心不可用
+        """
+        token_url = f"{settings.OAUTH_BASE_URL}{settings.OAUTH_TOKEN_ENDPOINT}"
+        logger.debug(f"Refreshing token at: {token_url}")
+        
+        async with httpx.AsyncClient(timeout=10.0) as client:
+            try:
+                response = await client.post(
+                    token_url,
+                    data={
+                        "grant_type": "refresh_token",
+                        "refresh_token": refresh_token,
+                        "client_id": settings.OAUTH_CLIENT_ID,
+                        "client_secret": settings.OAUTH_CLIENT_SECRET
+                    },
+                    headers={"Content-Type": "application/x-www-form-urlencoded"}
+                )
+            except httpx.RequestError as e:
+                logger.error(f"SSO refresh request failed: {e}")
+                raise HTTPException(
+                    status_code=status.HTTP_503_SERVICE_UNAVAILABLE,
+                    detail="SSO 认证中心不可用"
+                )
+        
+        logger.debug(f"SSO refresh response: status={response.status_code}")
+        
+        if response.status_code != 200:
+            logger.error(f"SSO refresh failed: {response.status_code}, body={response.text}")
+            raise HTTPException(
+                status_code=status.HTTP_401_UNAUTHORIZED,
+                detail="刷新令牌无效或已过期，请重新登录"
+            )
+        
+        data = response.json()
+        logger.debug(f"SSO refresh response data: {data}")
+        
+        # 处理包装格式 {"code": 0, "data": {...}} 或 {"code": "000000", "data": {...}}
+        code = data.get("code")
+        if (code == 0 or code == "000000") and "data" in data:
+            return data["data"]
+        elif "access_token" in data:
+            return data
+        else:
+            logger.error(f"Invalid refresh response format: {data}")
+            raise HTTPException(
+                status_code=status.HTTP_401_UNAUTHORIZED,
+                detail="刷新令牌无效或已过期，请重新登录"
+            )

backend/services/token_cache_service.py

@@ -0,0 +1,84 @@
+"""
+SSO Token 内存缓存服务
+缓存 token → 用户信息映射，减少对 SSO 中心的请求压力。
+支持可配置的 TTL 过期策略。
+"""
+import time
+import threading
+from dataclasses import dataclass, field
+from typing import Dict, Any, Optional
+
+
+@dataclass
+class CacheEntry:
+    """缓存条目，包含用户信息和创建时间"""
+    user_info: Dict[str, Any]
+    created_at: float = field(default_factory=time.time)
+
+    def is_expired(self, ttl: float) -> bool:
+        """检查缓存条目是否已过期"""
+        return (time.time() - self.created_at) > ttl
+
+
+class TokenCacheService:
+    """
+    SSO Token 内存缓存服务
+
+    通过缓存 token → 用户信息的映射，避免每次 API 请求都调用 SSO 中心验证。
+    缓存条目在 TTL 过期后自动失效，下次访问时会被清除。
+    """
+
+    def __init__(self, ttl_seconds: int = 300):
+        """
+        Args:
+            ttl_seconds: 缓存过期时间（秒），默认 300 秒（5 分钟）
+        """
+        self._cache: Dict[str, CacheEntry] = {}
+        self._ttl = ttl_seconds
+        self._lock = threading.Lock()
+
+    def get(self, token: str) -> Optional[Dict[str, Any]]:
+        """
+        查询缓存，返回用户信息或 None。
+        如果缓存条目已过期，自动删除并返回 None。
+
+        Args:
+            token: SSO access_token
+
+        Returns:
+            用户信息字典，或 None（未命中/已过期）
+        """
+        with self._lock:
+            entry = self._cache.get(token)
+            if entry is None:
+                return None
+            if entry.is_expired(self._ttl):
+                del self._cache[token]
+                return None
+            return entry.user_info
+
+    def set(self, token: str, user_info: Dict[str, Any]) -> None:
+        """
+        写入缓存。
+
+        Args:
+            token: SSO access_token
+            user_info: 用户信息字典
+        """
+        with self._lock:
+            self._cache[token] = CacheEntry(user_info=user_info)
+
+    def invalidate(self, token: str) -> None:
+        """
+        使指定 token 的缓存失效。
+
+        Args:
+            token: SSO access_token
+        """
+        with self._lock:
+            self._cache.pop(token, None)
+
+    def clear(self) -> None:
+        """清空所有缓存"""
+        with self._lock:
+            self._cache.clear()

backend/test/auth_test_helper.py

@@ -0,0 +1,22 @@
+"""
+Test helper for SSO-based authentication.
+Provides utilities to create test tokens by injecting them directly
+into the token cache, bypassing SSO center verification.
+"""
+import uuid
+from middleware.auth_middleware import token_cache
+
+
+def create_test_token(user_data: dict) -> str:
+    """
+    Create a fake SSO token for testing by injecting it into the token cache.
+
+    Args:
+        user_data: Dict with id, username, email, role
+
+    Returns:
+        A fake token string that will be recognized by the auth middleware.
+    """
+    fake_token = f"test_sso_token_{uuid.uuid4().hex}"
+    token_cache.set(fake_token, user_data)
+    return fake_token

backend/test/test_export_api.py

@@ -9,6 +9,7 @@ import uuid
 from fastapi.testclient import TestClient
 from main import app
 from database import get_db_connection, init_database
+from test.auth_test_helper import create_test_token
 
 
 # Test client
@@ -16,12 +17,6 @@ client = TestClient(app)
 
 
 # Test data
-TEST_ADMIN = {
-    "username": f"export_admin_{uuid.uuid4().hex[:8]}",
-    "email": f"export_admin_{uuid.uuid4().hex[:8]}@test.com",
-    "password": "testpassword123"
-}
-
 TEST_PROJECT = {
     "name": "Export Test Project",
     "description": "Project for testing export functionality",
@@ -38,37 +33,25 @@ def setup_database():
 
 @pytest.fixture(scope="module")
 def admin_token(setup_database):
-    """Create admin user and get token."""
-    # Register user
-    response = client.post("/api/auth/register", json=TEST_ADMIN)
-    if response.status_code != 201:
-        # User might already exist, try login
-        pass
-    
-    # Login
-    response = client.post("/api/auth/login", json={
-        "username": TEST_ADMIN["username"],
-        "password": TEST_ADMIN["password"]
-    })
-    
-    if response.status_code != 200:
-        pytest.skip("Could not authenticate admin user")
-    
-    data = response.json()
-    user_id = data["user"]["id"]
-    
-    # Update user role to admin
+    """Create admin user via token cache and get token."""
+    user_id = str(uuid.uuid4())
+    username = f"export_admin_{uuid.uuid4().hex[:8]}"
+    email = f"{username}@test.com"
+
+    # Ensure user exists in local DB
     with get_db_connection() as conn:
         cursor = conn.cursor()
-        cursor.execute("UPDATE users SET role = 'admin' WHERE id = ?", (user_id,))
-    
-    # Re-login to get updated token
-    response = client.post("/api/auth/login", json={
-        "username": TEST_ADMIN["username"],
-        "password": TEST_ADMIN["password"]
+        cursor.execute(
+            "INSERT OR IGNORE INTO users (id, username, email, role) VALUES (?, ?, ?, ?)",
+            (user_id, username, email, "admin"),
+        )
+
+    return create_test_token({
+        "id": user_id,
+        "username": username,
+        "email": email,
+        "role": "admin",
     })
-    
-    return response.json()["access_token"]
 
 
 @pytest.fixture(scope="module")
@@ -358,24 +341,24 @@ class TestExportPermissions:
     
     def test_non_admin_cannot_export(self, setup_database, test_project):
         """Test that non-admin users cannot export data."""
-        # Create a regular user
-        regular_user = {
-            "username": f"regular_user_{uuid.uuid4().hex[:8]}",
-            "email": f"regular_{uuid.uuid4().hex[:8]}@test.com",
-            "password": "testpassword123"
-        }
-        
-        # Register
-        response = client.post("/api/auth/register", json=regular_user)
-        assert response.status_code == 201
-        
-        # Login
-        response = client.post("/api/auth/login", json={
-            "username": regular_user["username"],
-            "password": regular_user["password"]
+        user_id = str(uuid.uuid4())
+        username = f"regular_user_{uuid.uuid4().hex[:8]}"
+        email = f"{username}@test.com"
+
+        # Ensure user exists in local DB as annotator
+        with get_db_connection() as conn:
+            cursor = conn.cursor()
+            cursor.execute(
+                "INSERT OR IGNORE INTO users (id, username, email, role) VALUES (?, ?, ?, ?)",
+                (user_id, username, email, "annotator"),
+            )
+
+        token = create_test_token({
+            "id": user_id,
+            "username": username,
+            "email": email,
+            "role": "annotator",
         })
-        assert response.status_code == 200
-        token = response.json()["access_token"]
         
         # Try to export
         headers = {"Authorization": f"Bearer {token}"}

backend/test/test_external_api_integration.py

@@ -14,6 +14,9 @@ sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
 
 from main import app
 from database import get_db_connection
+from test.auth_test_helper import create_test_token
+import bcrypt
+import uuid
 
 
 @pytest.fixture(scope="module")
@@ -24,64 +27,46 @@ def client():
 
 @pytest.fixture(scope="module")
 def admin_token(client):
-    """Get admin authentication token."""
-    # Try to login as admin
-    response = client.post("/api/auth/login", json={
-        "username": "admin",
-        "password": "admin123"
-    })
-    
-    if response.status_code == 200:
-        return response.json()["access_token"]
-    
-    # If admin doesn't exist, create one
-    response = client.post("/api/auth/register", json={
+    """Get admin authentication token via token cache injection."""
+    admin_id = f"admin_{uuid.uuid4().hex[:8]}"
+    password_hash = bcrypt.hashpw("admin123".encode(), bcrypt.gensalt()).decode()
+
+    with get_db_connection() as conn:
+        cursor = conn.cursor()
+        cursor.execute("""
+            INSERT INTO users (id, username, email, password_hash, role)
+            VALUES (?, ?, ?, ?, 'admin')
+        """, (admin_id, "admin", "admin@test.com", password_hash))
+
+    user_data = {
+        "id": admin_id,
         "username": "admin",
         "email": "admin@test.com",
-        "password": "admin123",
         "role": "admin"
-    })
-    
-    if response.status_code in [200, 201]:
-        return response.json()["access_token"]
-    
-    # Try login again
-    response = client.post("/api/auth/login", json={
-        "username": "admin",
-        "password": "admin123"
-    })
-    return response.json()["access_token"]
+    }
+    return create_test_token(user_data)
 
 
 @pytest.fixture(scope="module")
 def annotator_token(client):
-    """Get annotator authentication token."""
-    # Try to login as annotator
-    response = client.post("/api/auth/login", json={
-        "username": "annotator1",
-        "password": "annotator123"
-    })
-    
-    if response.status_code == 200:
-        return response.json()["access_token"]
-    
-    # If annotator doesn't exist, create one
-    response = client.post("/api/auth/register", json={
+    """Get annotator authentication token via token cache injection."""
+    annotator_id = f"annotator_{uuid.uuid4().hex[:8]}"
+    password_hash = bcrypt.hashpw("annotator123".encode(), bcrypt.gensalt()).decode()
+
+    with get_db_connection() as conn:
+        cursor = conn.cursor()
+        cursor.execute("""
+            INSERT INTO users (id, username, email, password_hash, role)
+            VALUES (?, ?, ?, ?, 'annotator')
+        """, (annotator_id, "annotator1", "annotator1@test.com", password_hash))
+
+    user_data = {
+        "id": annotator_id,
         "username": "annotator1",
         "email": "annotator1@test.com",
-        "password": "annotator123",
         "role": "annotator"
-    })
-    
-    if response.status_code in [200, 201]:
-        return response.json()["access_token"]
-    
-    # Try login again
-    response = client.post("/api/auth/login", json={
-        "username": "annotator1",
-        "password": "annotator123"
-    })
-    return response.json()["access_token"]
+    }
+    return create_test_token(user_data)
 
 
 class TestExternalAPIAuthentication:

backend/test/test_statistics_api.py

@@ -8,7 +8,7 @@ import json
 from fastapi.testclient import TestClient
 from main import app
 from database import get_db_connection, init_database
-from services.jwt_service import JWTService
+from test.auth_test_helper import create_test_token
 import bcrypt
 
 
@@ -42,7 +42,7 @@ def admin_user(setup_database):
         "email": f"admin_{admin_id}@test.com",
         "role": "admin"
     }
-    token = JWTService.create_access_token(user_data)
+    token = create_test_token(user_data)
     
     yield {"token": token, "user_id": admin_id}
     
@@ -71,7 +71,7 @@ def annotator_user(setup_database):
         "email": f"annotator_{annotator_id}@test.com",
         "role": "annotator"
     }
-    token = JWTService.create_access_token(user_data)
+    token = create_test_token(user_data)
     
     yield {"token": token, "user_id": annotator_id}
     

backend/test/test_task_assignment_api.py

@@ -8,7 +8,7 @@ import json
 from fastapi.testclient import TestClient
 from main import app
 from database import get_db_connection, init_database
-from services.jwt_service import JWTService
+from test.auth_test_helper import create_test_token
 import bcrypt
 
 
@@ -42,7 +42,7 @@ def admin_user(setup_database):
         "email": f"admin_{admin_id}@test.com",
         "role": "admin"
     }
-    token = JWTService.create_access_token(user_data)
+    token = create_test_token(user_data)
     
     yield {"token": token, "user_id": admin_id, "user_data": user_data}
     
@@ -74,7 +74,7 @@ def annotator_users(setup_database):
             "email": f"annotator_{annotator_id}@test.com",
             "role": "annotator"
         }
-        token = JWTService.create_access_token(user_data)
+        token = create_test_token(user_data)
         
         annotators.append({"token": token, "user_id": annotator_id, "user_data": user_data})
     

backend/test/test_template_api.py

@@ -7,7 +7,7 @@ import uuid
 from fastapi.testclient import TestClient
 from main import app
 from database import init_database
-from services.jwt_service import JWTService
+from test.auth_test_helper import create_test_token
 import bcrypt
 from database import get_db_connection
 
@@ -42,7 +42,7 @@ def auth_token(setup_database):
         "email": f"user_{user_id}@test.com",
         "role": "annotator"
     }
-    token = JWTService.create_access_token(user_data)
+    token = create_test_token(user_data)
     
     yield token
     

backend/test/test_user_api.py

@@ -7,7 +7,7 @@ import uuid
 from fastapi.testclient import TestClient
 from main import app
 from database import get_db_connection, init_database
-from services.jwt_service import JWTService
+from test.auth_test_helper import create_test_token
 import bcrypt
 
 
@@ -44,7 +44,7 @@ def admin_token(setup_database):
         "email": f"admin_{admin_id}@test.com",
         "role": "admin"
     }
-    token = JWTService.create_access_token(user_data)
+    token = create_test_token(user_data)
     
     yield {"token": token, "user_id": admin_id}
     
@@ -73,7 +73,7 @@ def annotator_token(setup_database):
         "email": f"annotator_{annotator_id}@test.com",
         "role": "annotator"
     }
-    token = JWTService.create_access_token(user_data)
+    token = create_test_token(user_data)
     
     yield {"token": token, "user_id": annotator_id}
     

web/apps/lq_label/src/app/app.tsx

@@ -5,7 +5,6 @@ import { ThemeProvider } from '../components/theme-provider';
 import { ProtectedRoute } from '../components/protected-route';
 import { AdminRoute } from '../components/admin-route';
 import { LoginForm } from '../components/login-form';
-import { RegisterForm } from '../components/register-form';
 import { OAuthCallback } from '../components/oauth-callback';
 import { isAuthenticatedAtom } from '../atoms/auth-atoms';
 import {
@@ -51,12 +50,6 @@ export function App() {
               isAuthenticated ? <Navigate to="/" replace /> : <LoginForm />
             }
           />
-          <Route
-            path="/register"
-            element={
-              isAuthenticated ? <Navigate to="/" replace /> : <RegisterForm />
-            }
-          />
           <Route path="/auth/callback" element={<OAuthCallback />} />
 
           {/* Protected Routes - Require Authentication */}

web/apps/lq_label/src/atoms/auth-atoms.ts

@@ -14,7 +14,7 @@ export interface User {
   id: string;
   username: string;
   email: string;
-  role: 'annotator' | 'admin';
+  role: 'annotator' | 'admin' | 'viewer';
   created_at: string;
 }
 

web/apps/lq_label/src/components/login-form/login-form.tsx

@@ -1,89 +1,47 @@
 /**
  * Login form component
- * Provides user authentication interface with OAuth support
- * 
- * Requirements: 2.1
+ * Auto-redirects to SSO login. Local login has been removed.
+ *
+ * Requirements: 6.1, 1.4
  */
 import React, { useState, useEffect } from 'react';
-import { useAtom } from 'jotai';
-import { useNavigate, Link } from 'react-router-dom';
-import { login } from '../../services/auth-service';
 import { startOAuthLogin, getOAuthStatus } from '../../services/oauth-service';
-import { loginAtom } from '../../atoms/auth-atoms';
 import { toast } from '../../services/toast';
 import styles from './login-form.module.scss';
 
 export const LoginForm: React.FC = () => {
-  const [, setAuth] = useAtom(loginAtom);
-  const navigate = useNavigate();
+  const [isLoading, setIsLoading] = useState(true);
+  const [error, setError] = useState<string | null>(null);
 
-  const [username, setUsername] = useState('');
-  const [password, setPassword] = useState('');
-  const [isLoading, setIsLoading] = useState(false);
-  const [oauthEnabled, setOauthEnabled] = useState(false);
-
-  // Check if OAuth is enabled
   useEffect(() => {
-    const checkOAuthStatus = async () => {
+    const redirectToSSO = async () => {
       try {
         const status = await getOAuthStatus();
-        setOauthEnabled(status.enabled);
-      } catch (error) {
-        console.error('Failed to check OAuth status:', error);
+        if (status.enabled) {
+          await startOAuthLogin();
+        } else {
+          setError('SSO 登录未启用，请联系管理员');
+          setIsLoading(false);
+        }
+      } catch (err: unknown) {
+        console.error('SSO redirect failed:', err);
+        setError('无法连接到 SSO 服务，请稍后重试');
+        setIsLoading(false);
       }
     };
-    checkOAuthStatus();
+    redirectToSSO();
   }, []);
 
-  const handleSubmit = async (e: React.FormEvent) => {
-    e.preventDefault();
-
-    // Validation
-    if (!username.trim()) {
-      toast.error('请输入用户名');
-      return;
-    }
-
-    if (!password) {
-      toast.error('请输入密码');
-      return;
-    }
-
+  const handleRetry = async () => {
     setIsLoading(true);
-
-    try {
-      // Call login API
-      const response = await login({ username, password });
-
-      // Update auth state
-      setAuth({
-        tokens: {
-          access_token: response.access_token,
-          refresh_token: response.refresh_token,
-          token_type: response.token_type,
-        },
-        user: response.user,
-      });
-
-      // Show success message
-      toast.success(`欢迎回来，${response.user.username}！`);
-
-      // Redirect to home page
-      navigate('/');
-    } catch (error: any) {
-      // Error is already handled by API interceptor
-      console.error('Login failed:', error);
-    } finally {
-      setIsLoading(false);
-    }
-  };
-
-  const handleOAuthLogin = async () => {
+    setError(null);
     try {
       await startOAuthLogin();
-    } catch (error: any) {
-      console.error('OAuth login failed:', error);
-      toast.error('OAuth 登录失败');
+    } catch (err: unknown) {
+      console.error('SSO retry failed:', err);
+      toast.error('SSO 登录失败，请稍后重试');
+      setError('无法连接到 SSO 服务，请稍后重试');
+      setIsLoading(false);
     }
   };
 
@@ -100,136 +58,35 @@ export const LoginForm: React.FC = () => {
               高效、智能的数据标注解决方案，助力 AI 模型训练
             </p>
           </div>
-          
-          <div className={styles.features}>
-            <div className={styles.feature}>
-              <div className={styles.featureIcon}>
-                <svg width="16" height="16" viewBox="0 0 24 24" fill="none">
-                  <path d="M9 11L12 14L22 4" stroke="currentColor" strokeWidth="2" strokeLinecap="round" strokeLinejoin="round"/>
-                  <path d="M21 12V19C21 20.1046 20.1046 21 19 21H5C3.89543 21 3 20.1046 3 19V5C3 3.89543 3.89543 3 5 3H16" stroke="currentColor" strokeWidth="2" strokeLinecap="round"/>
-                </svg>
-              </div>
-              <span className={styles.featureText}>高效标注</span>
-            </div>
-            
-            <div className={styles.feature}>
-              <div className={styles.featureIcon}>
-                <svg width="16" height="16" viewBox="0 0 24 24" fill="none">
-                  <path d="M17 21V19C17 17.9391 16.5786 16.9217 15.8284 16.1716C15.0783 15.4214 14.0609 15 13 15H5C3.93913 15 2.92172 15.4214 2.17157 16.1716C1.42143 16.9217 1 17.9391 1 19V21" stroke="currentColor" strokeWidth="2" strokeLinecap="round" strokeLinejoin="round"/>
-                  <path d="M9 11C11.2091 11 13 9.20914 13 7C13 4.79086 11.2091 3 9 3C6.79086 3 5 4.79086 5 7C5 9.20914 6.79086 11 9 11Z" stroke="currentColor" strokeWidth="2" strokeLinecap="round" strokeLinejoin="round"/>
-                  <path d="M23 21V19C22.9993 18.1137 22.7044 17.2528 22.1614 16.5523C21.6184 15.8519 20.8581 15.3516 20 15.13" stroke="currentColor" strokeWidth="2" strokeLinecap="round" strokeLinejoin="round"/>
-                  <path d="M16 3.13C16.8604 3.35031 17.623 3.85071 18.1676 4.55232C18.7122 5.25392 19.0078 6.11683 19.0078 7.005C19.0078 7.89318 18.7122 8.75608 18.1676 9.45769C17.623 10.1593 16.8604 10.6597 16 10.88" stroke="currentColor" strokeWidth="2" strokeLinecap="round" strokeLinejoin="round"/>
-                </svg>
-              </div>
-              <span className={styles.featureText}>团队协作</span>
-            </div>
-            
-            <div className={styles.feature}>
-              <div className={styles.featureIcon}>
-                <svg width="16" height="16" viewBox="0 0 24 24" fill="none">
-                  <path d="M12 22C17.5228 22 22 17.5228 22 12C22 6.47715 17.5228 2 12 2C6.47715 2 2 6.47715 2 12C2 17.5228 6.47715 22 12 22Z" stroke="currentColor" strokeWidth="2" strokeLinecap="round" strokeLinejoin="round"/>
-                  <path d="M12 6V12L16 14" stroke="currentColor" strokeWidth="2" strokeLinecap="round" strokeLinejoin="round"/>
-                </svg>
-              </div>
-              <span className={styles.featureText}>实时同步</span>
-            </div>
-            
-            <div className={styles.feature}>
-              <div className={styles.featureIcon}>
-                <svg width="16" height="16" viewBox="0 0 24 24" fill="none">
-                  <path d="M12 22C12 22 20 18 20 12V5L12 2L4 5V12C4 18 12 22 12 22Z" stroke="currentColor" strokeWidth="2" strokeLinecap="round" strokeLinejoin="round"/>
-                </svg>
-              </div>
-              <span className={styles.featureText}>安全可靠</span>
-            </div>
-          </div>
         </div>
       </div>
 
-      {/* 右侧区域 - 登录表单 */}
+      {/* 右侧区域 - SSO 登录状态 */}
       <div className={styles.rightSection}>
         <div className={styles.formCard}>
           <div className={styles.formHeader}>
-            <h2 className={styles.title}>欢迎回来</h2>
-            <p className={styles.subtitle}>登录您的账号以继续使用标注平台</p>
+            <h2 className={styles.title}>欢迎使用</h2>
+            <p className={styles.subtitle}>正在为您跳转到统一认证平台</p>
           </div>
 
-          {oauthEnabled && (
-            <>
+          {isLoading && !error && (
+            <div style={{ textAlign: 'center', padding: '2rem 0' }}>
+              <p>正在跳转到 SSO 登录...</p>
+            </div>
+          )}
+
+          {error && (
+            <div style={{ textAlign: 'center', padding: '2rem 0' }}>
+              <p style={{ color: '#ef4444', marginBottom: '1rem' }}>{error}</p>
               <button
                 type="button"
-                onClick={handleOAuthLogin}
+                onClick={handleRetry}
                 className={styles.oauthButton}
               >
-                <svg width="20" height="20" viewBox="0 0 24 24" fill="none">
-                  <path
-                    d="M12 2C6.48 2 2 6.48 2 12C2 17.52 6.48 22 12 22C17.52 22 22 17.52 22 12C22 6.48 17.52 2 12 2ZM12 20C7.59 20 4 16.41 4 12C4 7.59 7.59 4 12 4C16.41 4 20 7.59 20 12C20 16.41 16.41 20 12 20Z"
-                    fill="currentColor"
-                  />
-                  <path
-                    d="M12 6C8.69 6 6 8.69 6 12C6 15.31 8.69 18 12 18C15.31 18 18 15.31 18 12C18 8.69 15.31 6 12 6ZM12 16C9.79 16 8 14.21 8 12C8 9.79 9.79 8 12 8C14.21 8 16 9.79 16 12C16 14.21 14.21 16 12 16Z"
-                    fill="currentColor"
-                  />
-                </svg>
-                <span>使用 SSO 登录</span>
+                <span>重新尝试 SSO 登录</span>
               </button>
-
-              <div className={styles.divider}>
-                <span>或使用账号密码</span>
-              </div>
-            </>
-          )}
-
-          <form onSubmit={handleSubmit} className={styles.form}>
-            <div className={styles.formGroup}>
-              <label htmlFor="username" className={styles.label}>
-                用户名
-              </label>
-              <input
-                id="username"
-                type="text"
-                value={username}
-                onChange={(e) => setUsername(e.target.value)}
-                className={styles.input}
-                placeholder="请输入用户名"
-                disabled={isLoading}
-                autoComplete="username"
-              />
-            </div>
-
-            <div className={styles.formGroup}>
-              <label htmlFor="password" className={styles.label}>
-                密码
-              </label>
-              <input
-                id="password"
-                type="password"
-                value={password}
-                onChange={(e) => setPassword(e.target.value)}
-                className={styles.input}
-                placeholder="请输入密码"
-                disabled={isLoading}
-                autoComplete="current-password"
-              />
             </div>
-
-            <button
-              type="submit"
-              className={styles.submitButton}
-              disabled={isLoading}
-            >
-              {isLoading ? '登录中...' : '登录'}
-            </button>
-          </form>
-
-          <div className={styles.footer}>
-            <p className={styles.footerText}>
-              还没有账号？{' '}
-              <Link to="/register" className={styles.link}>
-                立即注册
-              </Link>
-            </p>
-          </div>
+          )}
         </div>
       </div>
     </div>

web/apps/lq_label/src/components/oauth-callback/oauth-callback.tsx

@@ -27,6 +27,11 @@ export const OAuthCallback: React.FC = () => {
     const processCallback = async () => {
       isProcessingRef.current = true;
       console.log('Starting OAuth callback processing...');
+      
+      // 清除任何旧的认证数据，避免干扰 OAuth 流程
+      localStorage.removeItem('auth_tokens');
+      localStorage.removeItem('current_user');
+      
       try {
         // Log all URL parameters for debugging
         const allParams: Record<string, string> = {};
@@ -80,7 +85,7 @@ export const OAuthCallback: React.FC = () => {
           },
           user: {
             ...response.user,
-            role: response.user.role as 'annotator' | 'admin',
+            role: response.user.role as 'annotator' | 'admin' | 'viewer',
           },
         });
 

web/apps/lq_label/src/main.tsx

@@ -1,4 +1,3 @@
-import { StrictMode } from 'react';
 import * as ReactDOM from 'react-dom/client';
 import { BrowserRouter } from 'react-router-dom';
 
@@ -17,9 +16,7 @@ const root = ReactDOM.createRoot(
   document.getElementById('root') as HTMLElement
 );
 root.render(
-  <StrictMode>
-    <BrowserRouter>
-      <App />
-    </BrowserRouter>
-  </StrictMode>
+  <BrowserRouter>
+    <App />
+  </BrowserRouter>
 );

web/apps/lq_label/src/services/api.ts

@@ -86,11 +86,11 @@ const clearStoredAuth = () => {
  */
 apiClient.interceptors.request.use(
   (config: InternalAxiosRequestConfig) => {
-    // Skip token attachment for auth endpoints
+    // Skip token attachment for OAuth public endpoints
     if (
-      config.url?.includes('/api/auth/register') ||
-      config.url?.includes('/api/auth/login') ||
-      config.url?.includes('/api/auth/refresh')
+      config.url?.includes('/api/oauth/login') ||
+      config.url?.includes('/api/oauth/callback') ||
+      config.url?.includes('/api/oauth/refresh')
     ) {
       return config;
     }
@@ -150,10 +150,8 @@ apiClient.interceptors.response.use(
 
     // Handle 401 Unauthorized errors (token expired or invalid)
     if (error.response?.status === 401) {
-      const errorData = error.response.data as any;
-
-      // If not a retry and error is due to token expiration, try to refresh
-      if (!originalRequest._retry && errorData?.error_type === 'token_expired') {
+      // If not a retry, try to refresh token first
+      if (!originalRequest._retry) {
         // Prevent infinite retry loop
         originalRequest._retry = true;
 
@@ -181,9 +179,9 @@ apiClient.interceptors.response.use(
             throw new Error('No refresh token available');
           }
 
-          // Call refresh token endpoint
+          // Call refresh token endpoint (SSO)
           const response = await axios.post(
-            `${API_BASE_URL || window.location.origin}/api/auth/refresh`,
+            `${API_BASE_URL || window.location.origin}/api/oauth/refresh`,
             {
               refresh_token: tokens.refresh_token,
             }
@@ -191,7 +189,7 @@ apiClient.interceptors.response.use(
 
           const newTokens = {
             access_token: response.data.access_token,
-            refresh_token: response.data.refresh_token,
+            refresh_token: response.data.refresh_token || tokens.refresh_token,
             token_type: response.data.token_type,
           };
 
@@ -230,15 +228,11 @@ apiClient.interceptors.response.use(
           isRefreshing = false;
         }
       } else {
-        // 401 error but not token expiration (invalid credentials, etc.)
-        // OR token refresh already attempted but still failed
-        // Clear auth data and redirect to login
+        // Already retried once, refresh failed — redirect to login
         clearStoredAuth();
         
-        // Show error message and wait a bit before redirecting
         toast.error('认证失败，请重新登录', '认证失败', 2000);
         
-        // Redirect to login page after a short delay to allow toast to show
         setTimeout(() => {
           window.location.href = '/login';
         }, 500);
@@ -1189,3 +1183,9 @@ export async function markProjectCompleted(projectId: string): Promise<Project>
  * Export the configured axios instance for advanced usage
  */
 export { apiClient };
+
+// ============================================================================
+// Token Refresh - DISABLED
+// SSO 不支持 refresh_token grant type，token 过期后需要重新登录
+// 响应拦截器会自动处理 401 错误并跳转到登录页
+// ============================================================================

web/apps/lq_label/src/services/auth-service.ts

@@ -1,104 +1,28 @@
 /**
- * Authentication service for user registration, login, and token management.
- * Provides functions for all authentication-related API calls.
- * 
- * Requirements: 1.1, 2.1, 3.1, 6.1
+ * Authentication service for SSO-based authentication.
+ * Local login/register have been removed — all auth goes through SSO.
+ *
+ * Requirements: 1.1, 2.1, 2.2, 6.1
  */
 import { apiClient } from './api';
-import type { User, AuthTokens } from '../atoms/auth-atoms';
+import type { User } from '../atoms/auth-atoms';
 
 /**
- * User registration data
- */
-export interface RegisterData {
-  username: string;
-  email: string;
-  password: string;
-}
-
-/**
- * User login data
- */
-export interface LoginData {
-  username: string;
-  password: string;
-}
-
-/**
- * Token refresh data
- */
-export interface RefreshTokenData {
-  refresh_token: string;
-}
-
-/**
- * Authentication response (login/register)
- */
-export interface AuthResponse {
-  access_token: string;
-  refresh_token: string;
-  token_type: string;
-  user: User;
-}
-
-/**
- * Register a new user
- * 
- * @param data - Registration data (username, email, password)
- * @returns Authentication response with tokens and user info
- */
-export async function register(data: RegisterData): Promise<AuthResponse> {
-  const response = await apiClient.post<AuthResponse>(
-    '/api/auth/register',
-    data
-  );
-  return response.data;
-}
-
-/**
- * Login with username and password
- * 
- * @param data - Login credentials (username, password)
- * @returns Authentication response with tokens and user info
- */
-export async function login(data: LoginData): Promise<AuthResponse> {
-  const response = await apiClient.post<AuthResponse>('/api/auth/login', data);
-  return response.data;
-}
-
-/**
- * Refresh access token using refresh token
- * 
- * @param refreshToken - Valid refresh token
- * @returns New authentication tokens and user info
- */
-export async function refreshToken(
-  refreshToken: string
-): Promise<AuthResponse> {
-  const response = await apiClient.post<AuthResponse>('/api/auth/refresh', {
-    refresh_token: refreshToken,
-  });
-  return response.data;
-}
-
-/**
- * Get current user information
- * Requires valid access token in Authorization header
- * 
+ * Get current user information from SSO-verified session.
+ * Requires valid SSO access token in Authorization header.
+ *
  * @returns Current user information
  */
 export async function getCurrentUser(): Promise<User> {
-  const response = await apiClient.get<User>('/api/auth/me');
+  const response = await apiClient.get<User>('/api/oauth/me');
   return response.data;
 }
 
 /**
  * Logout (client-side only)
  * Clears tokens from storage
- * Note: Server-side logout is not implemented (stateless JWT)
  */
 export function logout(): void {
-  // Clear tokens from localStorage
   localStorage.removeItem('auth_tokens');
   localStorage.removeItem('current_user');
 }