Documentation Index
Fetch the complete documentation index at: https://docs.mirobody.ai/llms.txt
Use this file to discover all available pages before exploring further.
健康设备提供商使用 OAuth 进行安全的用户授权。Mirobody 支持 OAuth 1.0 和 OAuth 2.0 协议。
OAuth 1.0
使用者：Garmin, Fitbit (旧版)
OAuth 2.0
使用者：Whoop, Apple Health, 大多数现代 API
OAuth 2.0 实现
大多数现代健康 API 使用 OAuth 2.0。以下是其实现方式：
1. 发起授权
async def link(self, user_id: str, return_url: str = None) -> Dict:
"""发起 OAuth 2.0 授权"""
# 生成用于 CSRF 保护的 state 参数
state = self._generate_state()
await self._store_state(user_id, state, return_url)
# 构建授权 URL
auth_params = {
"client_id": self.client_id,
"response_type": "code",
"redirect_uri": self.redirect_url,
"scope": self.scopes,
"state": state
}
auth_url = f"{self.auth_url}?{urlencode(auth_params)}"
return {
"link_web_url": auth_url,
"state": state
}
2. 处理回调
async def callback(self, user_id: str, code: str, state: str, **kwargs) -> Dict:
"""处理 OAuth 回调并使用 code 交换令牌"""
# 验证 state 参数
stored_state = await self._get_stored_state(user_id)
if state != stored_state:
raise ValueError("Invalid state parameter")
# 使用授权码交换访问令牌
token_data = {
"grant_type": "authorization_code",
"code": code,
"redirect_uri": self.redirect_url,
"client_id": self.client_id,
"client_secret": self.client_secret
}
async with aiohttp.ClientSession() as session:
async with session.post(self.token_url, data=token_data) as resp:
tokens = await resp.json()
# 安全地存储令牌
await self._store_tokens(user_id, tokens)
return {"success": True, "message": "Provider linked successfully"}
3. 令牌刷新
async def _refresh_token(self, user_id: str) -> Dict:
"""刷新过期的访问令牌"""
tokens = await self._get_tokens(user_id)
refresh_token = tokens.get("refresh_token")
if not refresh_token:
raise ValueError("No refresh token available")
token_data = {
"grant_type": "refresh_token",
"refresh_token": refresh_token,
"client_id": self.client_id,
"client_secret": self.client_secret
}
async with aiohttp.ClientSession() as session:
async with session.post(self.token_url, data=token_data) as resp:
new_tokens = await resp.json()
# 更新存储的令牌
await self._store_tokens(user_id, new_tokens)
return new_tokens
OAuth 1.0 实现
某些提供商 (如 Garmin) 使用 OAuth 1.0：
1. 请求令牌 (Request Token)
async def link(self, user_id: str, return_url: str = None) -> Dict:
"""发起 OAuth 1.0 流程"""
from requests_oauthlib import OAuth1Session
# 存储返回 URL
if return_url:
await self._store_return_url(user_id, return_url)
# 创建 OAuth1 会话
oauth = OAuth1Session(
client_key=self.client_id,
client_secret=self.client_secret,
callback_uri=self.redirect_url
)
# 获取请求令牌
request_token = oauth.fetch_request_token(self.request_token_url)
# 临时存储请求令牌
await self._store_request_token(user_id, request_token)
# 构建授权 URL
auth_url = oauth.authorization_url(self.auth_url)
return {"link_web_url": auth_url}
2. 交换访问令牌
async def callback(self, user_id: str, oauth_token: str, oauth_verifier: str, **kwargs) -> Dict:
"""使用 verifier 交换访问令牌"""
from requests_oauthlib import OAuth1Session
# 检索请求令牌
request_token = await self._get_request_token(user_id)
# 使用请求令牌创建 OAuth1 会话
oauth = OAuth1Session(
client_key=self.client_id,
client_secret=self.client_secret,
resource_owner_key=request_token["oauth_token"],
resource_owner_secret=request_token["oauth_token_secret"],
verifier=oauth_verifier
)
# 获取访问令牌
access_token = oauth.fetch_access_token(self.access_token_url)
# 存储访问令牌
await self._store_tokens(user_id, access_token)
return {"success": True}
安全最佳实践
- 在存入数据库前加密令牌
- 使用安全的密钥管理 (例如环境变量)
- 绝不记录令牌或凭据日志
- 在支持的情况下实现令牌轮换 (Token Rotation)
针对 OAuth 2.0：
- 生成加密强度高的随机 state
- 临时存储 state (10-15 分钟 TTL)
- 在回调中验证 state
- 防止 CSRF 攻击
- 优雅地处理过期令牌
- 使用指数退避算法重试失败请求
- 记录错误时避免暴露敏感数据
- 向用户提供清晰的错误消息
完整参考
如需包含所有必需方法和模式的详细实现，请参阅：
