skyplatform-iam 2.2.2

SkyPlatform IAM认证SDK，提供FastAPI中间件和认证路由

SkyPlatform IAM SDK

SkyPlatform IAM 认证 SDK，提供 FastAPI 中间件和 IAM 服务连接功能，简化第三方服务的认证集成。

功能特性

• FastAPI 纯 ASGI 中间件: 自动拦截请求进行 Token 验证和权限检查
• 统一初始化: 使用 init_skyplatform_iam 一键设置认证功能
• 全局客户端单例: 通过 get_iam_client() 在任何地方访问 IAM 功能
• JWKS 本地验签: 服务间调用（SERVICE-TOKEN）使用 JWKS 公钥本地验证，无需回调 IAM
• HTTP 连接池 + 自动重试: 内置 requests.Session 连接池和网络错误自动重试
• 灵活配置: 通过显式 AuthConfig 进行代码配置
• 白名单机制: 支持通配符匹配的白名单路径配置
• 完善的异常体系: 自定义异常层级，便于错误处理

快速开始

安装

pip install skyplatform-iam

使用规范

1. 通过 config = AuthConfig(...) 构造配置（必须显式传入）
2. 通过 init_skyplatform_iam(app, config) 注册中间件（启动时自动获取 machine_token）
3. 后续直接使用 get_iam_client() 获取单例客户端（自动校验与续期 machine_token）

基本使用

一键初始化（推荐）

from fastapi import FastAPI
from skyplatform_iam import init_skyplatform_iam, AuthConfig

app = FastAPI()

config = AuthConfig(
    skyplatform_iam_host="http://127.0.0.1:5001",
    server_name="my_service",
    access_key="your_access_key",
    whitelist_paths=["/docs", "/redoc", "/openapi.json", "/health"],
)

init_skyplatform_iam(app, config)


@app.get("/protected")
async def protected_endpoint(request):
    user = request.state.user
    return {"message": "访问成功", "user": user}

在业务代码中使用 IAM 客户端

from skyplatform_iam import get_iam_client

# 在已调用 init_skyplatform_iam 之后，获取单例客户端
iam_client = get_iam_client()

自定义配置

from skyplatform_iam import init_skyplatform_iam, AuthConfig

config = AuthConfig(
    skyplatform_iam_host="https://your-iam-host.com",
    server_name="your-server-name",
    access_key="your-access-key",
    whitelist_paths=[
        "/docs", "/redoc", "/openapi.json",
        "/health", "/public",
        "/auth/register", "/auth/login",
    ],
    enable_debug=True,
    enable_sdk_logging=True,
)

init_skyplatform_iam(app, config)

内置方法

ConnectSkyplatformIam 类提供以下方法：

用户认证

方法	说明	返回值
register(...)	用户注册	成功返回用户信息 dict，失败返回 None
login_with_password(...)	账号密码登录	成功返回含 access_token / refresh_token 的 dict，失败返回 None
logout(token)	用户登出	成功返回 True，失败返回 False

Token 管理

方法	说明	返回值
verify_token(token, api, method)	Token 验证和权限检查	成功返回用户信息 dict，无权限抛 403，无效返回 None
refresh_token(refresh_token)	刷新 Token	成功返回含新 token 的 dict，失败返回 None

用户配置

方法	说明	返回值
add_custom_config(user_id, config_name, config_value)	添加/更新自定义配置	成功返回 dict，失败返回 None
get_custom_configs(user_id)	获取用户所有自定义配置	成功返回 dict，失败返回 None
delete_custom_config(user_id, config_name)	删除自定义配置	成功返回 dict，失败返回 None

用户管理

方法	说明	返回值
get_space_users(search, user_status, ...)	获取空间用户列表（分页）	成功返回含分页数据的 dict，失败返回 None

服务间调用

方法	说明	返回值
get_service_token(target_service, ttl)	获取服务间调用令牌	成功返回 token 字符串，失败返回空串
build_service_call_headers(target_service, ttl)	构建服务间调用请求头	返回含 SERVICE-TOKEN 的 dict
get_authorized_services()	获取本服务被授权调用的目标服务列表	返回服务名列表

使用示例

用户注册与登录

from skyplatform_iam import get_iam_client

iam = get_iam_client()

# 注册
result = iam.register(
    cred_type="username",
    cred_value="testuser",
    password="password123",
    nickname="测试用户",
)
if result:
    print(f"注册成功: {result}")

# 登录
data = iam.login_with_password(
    cred_type="username",
    cred_value="testuser",
    password="password123",
)
if data:
    access_token = data["access_token"]
    refresh_token = data["refresh_token"]
    print(f"登录成功: {access_token}")

# 登出
iam.logout(token=access_token)

Token 验证与刷新

# Token 验证
user_info = iam.verify_token(
    token="user_token",
    api="/api/protected",
    method="GET",
)
if user_info:
    print(f"Token 有效，用户ID: {user_info['user_id']}")

# 刷新 Token
new_tokens = iam.refresh_token(refresh_token="your_refresh_token")
if new_tokens:
    print(f"刷新成功: {new_tokens['access_token']}")

用户配置与列表

# 添加配置
iam.add_custom_config(user_id="user_123", config_name="theme", config_value="dark")

# 获取配置
configs = iam.get_custom_configs(user_id="user_123")

# 获取空间用户列表
result = iam.get_space_users(page=1, per_page=10, search="test", user_status="active")

服务间调用

# 构建调用其他服务的请求头
headers = iam.build_service_call_headers("target_service_name")
# headers = {"Content-Type": "application/json", "SERVICE-TOKEN": "..."}

import requests
resp = requests.post("http://target-service/api/data", headers=headers, json={})

中间件功能

鉴权优先级

1. OPTIONS 预检请求 → 直接放行
2. 本地白名单路径 → 放行，request.state.is_whitelist = True
3. SERVICE-TOKEN（服务间调用） → 本地 JWKS 验签，不回调 IAM，request.state.is_service_call = True
4. Authorization Bearer（用户请求） → 调用 IAM /verify 做权限校验

白名单配置

默认内置白名单：/docs、/redoc、/openapi.json、/.well-known/appspecific/com.chrome.devtools.json

config = AuthConfig(
    skyplatform_iam_host="http://127.0.0.1:5000",
    server_name="your_service_name",
    access_key="your_access_key",
)
config.add_whitelist_path("/public")
config.add_whitelist_path("/api/public/*")  # 支持通配符

请求状态属性

中间件验证后会在 request.state 上设置以下属性：

属性	类型	说明
user	dict | None	用户/服务信息
authenticated	bool	是否已认证
is_whitelist	bool	是否白名单路径
is_service_call	bool	是否服务间调用

依赖注入

from skyplatform_iam import get_current_user, get_optional_user
from fastapi import FastAPI, Request

app = FastAPI()


@app.get("/me")
async def me(user: dict = Depends(get_current_user)):
    return user  # 未登录自动返回 401


@app.get("/optional-me")
async def optional_me(user: Optional[dict] = Depends(get_optional_user)):
    return {"user": user}  # 未登录返回 None

异常处理

from skyplatform_iam.exceptions import (
    SkyPlatformAuthException,  # 基础异常
    AuthenticationError,       # 认证失败 (401)
    AuthorizationError,        # 权限不足 (403)
    TokenExpiredError,         # Token 过期 (401)
    TokenInvalidError,         # Token 无效 (401)
    ConfigurationError,        # 配置错误 (500)
    IAMServiceError,           # IAM 服务错误 (500)
    NetworkError,              # 网络连接错误 (503)
)

配置选项

AuthConfig 参数

参数	类型	必填	默认值	说明
skyplatform_iam_host	str	✓	—	IAM 服务地址
server_name	str	✓	—	服务名称
access_key	str	✓	—	访问密钥
machine_token	str	✗	""	预置 machine_token（通常自动获取）
whitelist_paths	List[str]	✗	[]	白名单路径（支持通配符）
ssl_verify	bool	✗	True	是否验证 SSL 证书
token_header	str	✗	"Authorization"	Token 请求头名称
token_prefix	str	✗	"Bearer "	Token 前缀
enable_debug	bool	✗	False	启用调试模式（错误响应含详情）
enable_sdk_logging	bool	✗	True	启用 SDK 日志输出

开发和测试

pip install -e ".[dev]"    # 安装开发依赖
pytest                     # 运行测试
black .                    # 格式化
isort .                    # 排序导入
mypy .                     # 类型检查

兼容性

• Python 3.9+
• FastAPI 0.68.0+
• Pydantic 2.0+

许可证

MIT License

更新日志

v2.2.0

• 中间件改为纯 ASGI 实现，避免 BaseHTTPMiddleware 的流式响应等问题
• 所有 HTTP 请求统一走连接池（requests.Session）+ 自动重试
• async 上下文中的同步 HTTP 调用改用 asyncio.to_thread，不再阻塞事件循环
• 修复 login_with_password / refresh_token 错误覆写 machine_token 的问题
• 统一方法返回类型：成功返回 dict，失败返回 None（不再返回 requests.Response）
• get_space_users 参数 status 重命名为 user_status，避免与内置名冲突
• 提取 verify_service_token_locally 为公共函数，消除 SERVICE-TOKEN 验证逻辑重复
• 消除 decode_jwt_token 的重复实现，统一使用 config.decode_jwt_token
• reload_config 现在完整更新 JWKS 管理器和授权列表
• pydantic 依赖升级到 >=2.0.0
• Python 版本要求统一为 >=3.9

v2.0.0

• 新增 init_skyplatform_iam 统一初始化函数
• 新增 get_iam_client 全局客户端获取函数
• 改进文档和使用示例

v1.0.0

• 初始版本发布
• 提供 FastAPI 中间件和认证路由
• 支持完整的认证功能