基于MSAL通过Access Token实现后端Refresh Token获取与邮件访问

一、前端调整：确保获取刷新令牌 #

要实现长期后台访问邮箱，必须请求offline_access权限——仅靠有效期1小时左右的访问令牌（accessToken）无法满足定时任务需求，只有刷新令牌（refreshToken）能用来持续获取新的访问令牌。

修改MSAL配置，补充权限并调整向后端传递的数据：

// 示例MSAL配置，补充offline_access权限
const msalConfig = {
  auth: {
    clientId: "你的客户端ID",
    authority: "https://login.microsoftonline.com/common",
    redirectUri: "你的前端回调地址",
  },
  tokenRequest: {
    scopes: ["Mail.Read", "offline_access"] // 必须添加offline_access
  }
};

const msalInstance = new PublicClientApplication(msalConfig);

const loginPopup = async () => {
  try {
    const loginResponse = await msalInstance.loginPopup(
      msalConfig.tokenRequest
    );
    // 向后端传递刷新令牌、租户ID、用户名等关键信息
    const authDataToSend = {
      refreshToken: loginResponse.refreshToken,
      tenantId: loginResponse.tenantId,
      username: loginResponse.account.username
    };
    props.onSuccess(authDataToSend); // 发送至后端
    console.log("loginResponse", loginResponse);
  } catch (err) {
    console.error(err);
    props.onFailure(err);
  }
};

二、后端：存储令牌并实现刷新逻辑 #

1. 安全存储令牌 #

刷新令牌属于敏感数据，必须加密存储（如数据库加密字段、加密文件）。以下是数据库存储的示例（用SQLAlchemy定义模型）：

from sqlalchemy import Column, String, DateTime
from sqlalchemy.ext.declarative import declarative_base
import datetime

Base = declarative_base()

class UserMailToken(Base):
    __tablename__ = 'user_mail_tokens'
    username = Column(String, primary_key=True)
    refresh_token = Column(String, nullable=False) # 需加密存储
    tenant_id = Column(String, nullable=False)
    created_at = Column(DateTime, default=datetime.datetime.utcnow)

2. 接收前端令牌并存储 #

处理前端POST请求，完成令牌的新增/更新：

from flask import request, jsonify
import json
from your_models import UserMailToken, db
from msal import ConfidentialClientApplication
import consts

@app.route('/save-mail-token', methods=['POST'])
def save_mail_token():
    auth_data = request.get_json()
    # 检查用户是否已存在，存在则更新令牌，不存在则创建新记录
    existing_token = UserMailToken.query.filter_by(username=auth_data['username']).first()
    if existing_token:
        existing_token.refresh_token = auth_data['refreshToken']
        existing_token.tenant_id = auth_data['tenantId']
    else:
        new_token = UserMailToken(
            username=auth_data['username'],
            refresh_token=auth_data['refreshToken'],
            tenant_id=auth_data['tenantId']
        )
        db.session.add(new_token)
    db.session.commit()
    return jsonify({"status": "success"})

3. 令牌刷新与访问令牌获取 #

编写工具函数，用刷新令牌获取有效的访问令牌：

def get_access_token(username):
    # 从数据库读取用户的刷新令牌和租户ID
    user_token = UserMailToken.query.filter_by(username=username).first()
    if not user_token:
        raise ValueError(f"用户{username}未完成邮箱授权")
    
    # 初始化MSAL客户端
    with open(consts.O365_OAUTH_CREDS_PATH) as f:
        creds_data = json.load(f)
    
    app = ConfidentialClientApplication(
        client_id=creds_data["auth"]["client_id"],
        client_credential=creds_data["auth"]["client_secret"],
        authority=f'{creds_data["auth"]["authority_base_uri"]}{user_token.tenant_id}',
    )
    
    # 用刷新令牌换取新的访问令牌
    result = app.acquire_token_by_refresh_token(
        refresh_token=user_token.refresh_token,
        scopes=["Mail.Read"]
    )
    
    if "access_token" in result:
        # 部分场景下会返回新的刷新令牌，需更新存储
        if "refresh_token" in result:
            user_token.refresh_token = result["refresh_token"]
            db.session.commit()
        return result["access_token"]
    else:
        # 刷新失败（令牌过期/用户撤销授权），需提示用户重新登录
        raise Exception(f"令牌刷新失败: {result.get('error_description')}")

三、邮件读取与定时任务 #

1. 调用Graph API读取邮件主题 #

用获取到的访问令牌调用Microsoft Graph API筛选邮件：

import requests

def fetch_target_mails(username):
    access_token = get_access_token(username)
    # 调用Graph API获取邮件主题，可自定义筛选条件
    graph_url = "https://graph.microsoft.com/v1.0/me/messages?$select=subject&$top=100"
    
    headers = {
        "Authorization": f"Bearer {access_token}"
    }
    
    response = requests.get(graph_url, headers=headers)
    if response.status_code == 200:
        messages = response.json().get('value', [])
        # 筛选符合特定条件的邮件主题
        target_subjects = [msg['subject'] for msg in messages if "目标关键词" in msg['subject']]
        return target_subjects
    else:
        raise Exception(f"读取邮件失败: {response.text}")

2. 定时任务实现 #

用APScheduler实现定时邮件检查：

from apscheduler.schedulers.background import BackgroundScheduler

def scheduled_mail_check():
    # 遍历所有已授权用户，执行邮件检查
    all_users = UserMailToken.query.all()
    for user in all_users:
        try:
            target_subjects = fetch_target_mails(user.username)
            # 处理筛选结果，如存入用户通知表、推送至前端等
            print(f"用户{user.username}符合条件的邮件主题: {target_subjects}")
        except Exception as e:
            print(f"处理用户{user.username}邮件失败: {str(e)}")

# 初始化定时任务，示例为每小时执行一次
scheduler = BackgroundScheduler()
scheduler.add_job(scheduled_mail_check, 'interval', hours=1)
scheduler.start()

关键注意事项 #

• 令牌安全：刷新令牌长期有效，必须加密存储，禁止明文暴露。
• 权限最小化：仅请求Mail.Read和offline_access必要权限，避免过度授权。
• 错误处理：令牌刷新失败时，需引导用户重新登录授权；调用Graph API时要处理401、403等权限错误。
• 租户适配：多租户应用需正确存储每个用户的tenantId，确保授权地址匹配。

内容的提问来源于stack exchange，提问作者Bar Ezra