OAUTH_INTEGRATION_GUIDE.md 9.9 KB
OAuth 2.0 单点登录对接方案
概述
本文档详细说明如何将标注平台与 OAuth 2.0 认证中心集成，实现单点登录（SSO）功能。
OAuth 2.0 授权码模式流程
┌─────────┐ ┌──────────────┐
│ │ │ │
│ 用户 │ │ 标注平台 │
│ │ │ (Client) │
└────┬────┘ └──────┬───────┘
│ │
│ 1. 访问标注平台 │
├──────────────────────────────────────────────────────>│
│ │
│ 2. 重定向到 OAuth 登录页 │
│<──────────────────────────────────────────────────────┤
│ │
│ │
│ ┌──────────────────────────────────────────┐ │
│ │ OAuth 认证中心 │ │
│ │ (http://192.168.92.61:8000) │ │
│ └──────────────────────────────────────────┘ │
│ │
│ 3. 用户登录并授权 │
├──────────────────────────────────────────────────────>│
│ │
│ 4. 返回授权码 (code) │
│<──────────────────────────────────────────────────────┤
│ │
│ 5. 携带授权码回调标注平台 │
├──────────────────────────────────────────────────────>│
│ │
│ │ 6. 用授权码换取 token
│ ├────────────────────>
│ │ OAuth
│ │ 7. 返回 access_token
│ │<────────────────────
│ │
│ │ 8. 获取用户信息
│ ├────────────────────>
│ │ OAuth
│ │ 9. 返回用户信息
│ │<────────────────────
│ │
│ 10. 登录成功，建立会话 │
│<──────────────────────────────────────────────────────┤
│ │
OAuth 认证中心信息
基础配置
- OAuth 服务地址:
http://192.168.92.61:8000 - 授权端点:
http://192.168.92.61:8000/oauth/authorize - 令牌端点:
http://192.168.92.61:8000/oauth/token - 用户信息端点:
http://192.168.92.61:8000/oauth/userinfo - 撤销端点:
http://192.168.92.61:8000/oauth/revoke
应用配置（待提供）
- Client ID (应用Key):
待提供 - Client Secret (应用密钥):
待提供 - 回调 URL:
http://localhost:4200/auth/callback(开发环境) - 回调 URL:
http://192.168.92.61:8100/auth/callback(生产环境)
授权参数
response_type: code # 授权码模式
client_id: <YOUR_CLIENT_ID> # 应用标识
redirect_uri: <YOUR_CALLBACK> # 回调地址
scope: profile email # 请求的权限范围
state: <RANDOM_STRING> # 防CSRF攻击的随机字符串
后端实现方案
1. 环境配置
在 backend/.env 中添加 OAuth 配置：
# OAuth 2.0 配置
OAUTH_ENABLED=true
OAUTH_BASE_URL=http://192.168.92.61:8000
OAUTH_CLIENT_ID=<待提供的Client ID>
OAUTH_CLIENT_SECRET=<待提供的Client Secret>
OAUTH_REDIRECT_URI=http://localhost:4200/auth/callback
OAUTH_SCOPE=profile email
2. 更新配置模块
修改 backend/config.py：
from pydantic_settings import BaseSettings
from typing import Optional
class Settings(BaseSettings):
# ... 现有配置 ...
# OAuth 2.0 配置
OAUTH_ENABLED: bool = False
OAUTH_BASE_URL: str = "http://192.168.92.61:8000"
OAUTH_CLIENT_ID: str = ""
OAUTH_CLIENT_SECRET: str = ""
OAUTH_REDIRECT_URI: str = "http://localhost:4200/auth/callback"
OAUTH_SCOPE: str = "profile email"
class Config:
env_file = ".env"
case_sensitive = True
settings = Settings()
3. 创建 OAuth 服务
创建 backend/services/oauth_service.py：
```python """ OAuth 2.0 认证服务 """ import httpx import secrets from typing import Dict, Any, Optional from backend.config import settings from backend.models import User from backend.database import get_db_connection from datetime import datetime
class OAuthService:
"""OAuth 认证服务"""
@staticmethod
def generate_state() -> str:
"""生成随机 state 参数"""
return secrets.token_urlsafe(32)
@staticmethod
def get_authorization_url(state: str) -> str:
"""
构建授权 URL
Args:
state: 防CSRF的随机字符串
Returns:
完整的授权URL
"""
from urllib.parse import urlencode
params = {
"response_type": "code",
"client_id": settings.OAUTH_CLIENT_ID,
"redirect_uri": settings.OAUTH_REDIRECT_URI,
"scope": settings.OAUTH_SCOPE,
"state": state
}
return f"{settings.OAUTH_BASE_URL}/oauth/authorize?{urlencode(params)}"
@staticmethod
async def exchange_code_for_token(code: str) -> Dict[str, Any]:
"""
用授权码换取访问令牌
Args:
code: 授权码
Returns:
令牌信息字典
"""
async with httpx.AsyncClient() as client:
response = await client.post(
f"{settings.OAUTH_BASE_URL}/oauth/token",
data={
"grant_type": "authorization_code",
"code": code,
"redirect_uri": settings.OAUTH_REDIRECT_URI,
"client_id": settings.OAUTH_CLIENT_ID,
"client_secret": settings.OAUTH_CLIENT_SECRET
}
)
if response.status_code != 200:
raise Exception(f"令牌交换失败: {response.text}")
data = response.json()
# 处理不同的响应格式
if "access_token" in data:
return data
elif data.get("code") == 0 and "data" in data:
return data["data"]
else:
raise Exception(f"无效的令牌响应格式: {data}")
@staticmethod
async def get_user_info(access_token: str) -> Dict[str, Any]:
"""
获取用户信息
Args:
access_token: 访问令牌
Returns:
用户信息字典
"""
async with httpx.AsyncClient() as client:
response = await client.get(
f"{settings.OAUTH_BASE_URL}/oauth/userinfo",
headers={"Authorization": f"Bearer {access_token}"}
)
if response.status_code != 200:
raise Exception(f"获取用户信息失败: {response.text}")
data = response.json()
# 处理不同的响应格式
if "sub" in data:
return data
elif data.get("code") == 0 and "data" in data:
return data["data"]
else:
raise Exception(f"无效的用户信息响应格式: {data}")
@staticmethod
async def sync_user_from_oauth(oauth_user_info: Dict[str, Any]) -> User:
"""
从 OAuth 用户信息同步到本地数据库
Args:
oauth_user_info: OAuth 返回的用户信息
Returns:
本地用户对象
"""
with get_db_connection() as conn:
cursor = conn.cursor()
# 提取