跳转到主要内容

​
概览

健康设备提供商使用 OAuth 进行安全的用户授权。Mirobody Health 支持 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 攻击
  • 优雅地处理过期令牌
  • 使用指数退避算法重试失败请求
  • 记录错误时避免暴露敏感数据
  • 向用户提供清晰的错误消息

​
完整参考

如需包含所有必需方法和模式的详细实现，请参阅：

完整集成指南

项目仓库中的完整提供商集成指南

Garmin 示例

OAuth 1.0 实现示例

Whoop 示例

OAuth 2.0 实现示例

提供商测试

测试策略