跳转到主要内容

​
概览

健康设备 Providers 使用 OAuth 来安全地完成用户授权。Mirobody Health 同时支持 OAuth 1.0 与 OAuth 2.0 协议。

OAuth 1.0

Used by: Garmin, Fitbit (legacy)

OAuth 2.0

Used by: Whoop, Apple Health, most modern APIs

​
OAuth 2.0 Implementation

大多数现代健康 APIs 使用 OAuth 2.0。实现方式如下：

​
1. Initiate Authorization

async def link(self, user_id: str, return_url: str = None) -> Dict:
    """Initiate OAuth 2.0 authorization"""
    # Generate state parameter for CSRF protection
    state = self._generate_state()
    await self._store_state(user_id, state, return_url)
    
    # Build authorization 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. Handle Callback

async def callback(self, user_id: str, code: str, state: str, **kwargs) -> Dict:
    """Handle OAuth callback and exchange code for tokens"""
    # Verify state parameter
    stored_state = await self._get_stored_state(user_id)
    if state != stored_state:
        raise ValueError("Invalid state parameter")
    
    # Exchange authorization code for access token
    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()
    
    # Store tokens securely
    await self._store_tokens(user_id, tokens)
    
    return {"success": True, "message": "Provider linked successfully"}

​
3. Token Refresh

async def _refresh_token(self, user_id: str) -> Dict:
    """Refresh expired access token"""
    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()
    
    # Update stored tokens
    await self._store_tokens(user_id, new_tokens)
    
    return new_tokens

​
OAuth 1.0 Implementation

部分 Providers（如 Garmin）使用 OAuth 1.0：

​
1. Request Token

async def link(self, user_id: str, return_url: str = None) -> Dict:
    """Initiate OAuth 1.0 flow"""
    from requests_oauthlib import OAuth1Session
    
    # Store return URL
    if return_url:
        await self._store_return_url(user_id, return_url)
    
    # Create OAuth1 session
    oauth = OAuth1Session(
        client_key=self.client_id,
        client_secret=self.client_secret,
        callback_uri=self.redirect_url
    )
    
    # Fetch request token
    request_token = oauth.fetch_request_token(self.request_token_url)
    
    # Store request token temporarily
    await self._store_request_token(user_id, request_token)
    
    # Build authorization URL
    auth_url = oauth.authorization_url(self.auth_url)
    
    return {"link_web_url": auth_url}

​
2. Access Token Exchange

async def callback(self, user_id: str, oauth_token: str, oauth_verifier: str, **kwargs) -> Dict:
    """Exchange verifier for access token"""
    from requests_oauthlib import OAuth1Session
    
    # Retrieve request token
    request_token = await self._get_request_token(user_id)
    
    # Create OAuth1 session with request token
    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
    )
    
    # Fetch access token
    access_token = oauth.fetch_access_token(self.access_token_url)
    
    # Store access token
    await self._store_tokens(user_id, access_token)
    
    return {"success": True}

​
Security Best Practices

  • 数据库存储前对 tokens 加密
  • 使用安全的 key 管理方式（例如环境变量）
  • 不要记录 tokens 或凭据
  • 在支持的情况下实现 token rotation
For OAuth 2.0:
  • 生成加密安全的随机 state
  • 临时存储 state（10–15 分钟 TTL）
  • 在 callback 中校验 state
  • 防止 CSRF attacks
  • 优雅处理过期 tokens
  • 对失败请求使用 exponential backoff 重试
  • 记录错误时避免暴露敏感数据
  • 给用户提供清晰的错误信息

​
完整参考

如需包含所有必需 methods 与模式的详细实现，请参见：

Full Integration Guide

仓库中的完整 Provider 集成指南

Garmin Example

OAuth 1.0 实现示例

Whoop Example

OAuth 2.0 实现示例

Provider Testing

测试策略