自动化脚本如何清理未使用OAuth令牌

wen 实用脚本 2026-07-04 22

本文目录导读：

自动化脚本如何清理未使用OAuth令牌

核心原则
方案一：基于数据库直接清理（最常见，高效）
方案二：通过Identity Provider (IdP) API 清理（更安全，适用于云服务或标准授权服务器）
方案三：基于日志分析（无最后使用时间字段）
最佳实践建议

清理未使用的OAuth令牌是保障系统安全、减少资源浪费的重要操作，由于OAuth令牌（包括访问令牌和刷新令牌）通常存储在数据库或缓存中，自动化脚本的核心逻辑是识别“僵尸令牌”（已过期、已撤销或长期未使用的令牌）并执行删除或失效操作。

以下是几种不同场景下的自动化脚本方案及最佳实践：

核心原则

切勿删除未过期的有效令牌：只清理已过期或已被用户/管理员主动撤销的令牌。
识别“未使用”的令牌：这通常指令牌的最后使用时间（last_used_at） 远早于当前时间，且已超过最大允许的不活跃周期（90天未使用），如果系统没有记录最后使用时间，可以跳过此项,仅处理过期令牌。
优雅处理刷新令牌：如果清理了刷新令牌，用户下次访问时需要重新登录,需确保通知用户或按策略执行。

基于数据库直接清理（最常见，高效）

如果你的OAuth令牌存储在数据库（如MySQL, PostgreSQL）中,这是最直接的方法。

脚本逻辑：删除数据库中 last_used_at 早于某个时间点且 expires_at 已过的令牌,或直接删除所有已过期的令牌。

示例脚本（Python + SQL）：

import pymysql
from datetime import datetime, timedelta
import logging
# 配置日志
logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s')
# 数据库配置
DB_CONFIG = {
    'host': 'your-db-host',
    'user': 'your-db-user',
    'password': 'your-db-password',
    'database': 'oauth_db',
}
def cleanup_stale_oauth_tokens(days_inactive=90):
    """
    清理指定天数内未使用且已过期的OAuth令牌。
    对于未记录最后使用时间的令牌，仅清理已过期的。
    """
    try:
        connection = pymysql.connect(**DB_CONFIG)
        cursor = connection.cursor()
        # 计算截止时间点
        cutoff_date = datetime.now() - timedelta(days=days_inactive)
        # 策略1：清理“未使用+已过期”的令牌（推荐）
        sql_cleanup_used = """
            DELETE FROM oauth_access_tokens
            WHERE (last_used_at IS NULL AND expires_at < NOW())
               OR (last_used_at IS NOT NULL AND last_used_at < %s AND expires_at < NOW())
        """
        cursor.execute(sql_cleanup_used, (cutoff_date,))
        deleted_access_tokens = cursor.rowcount
        logging.info(f"删除了 {deleted_access_tokens} 个未使用/过期的访问令牌")
        # 同时清理关联的刷新令牌（如果你的存储结构是分开的）
        sql_cleanup_refresh = """
            DELETE FROM oauth_refresh_tokens
            WHERE (last_used_at IS NULL AND expires_at < NOW())
               OR (last_used_at IS NOT NULL AND last_used_at < %s AND expires_at < NOW())
        """
        cursor.execute(sql_cleanup_refresh, (cutoff_date,))
        deleted_refresh_tokens = cursor.rowcount
        logging.info(f"删除了 {deleted_refresh_tokens} 个未使用/过期的刷新令牌")
        # 可选：清理孤立令牌（用户已删除，但令牌还在）
        # 需根据你的用户表外键关系调整
        # sql_orphaned = """
        #     DELETE at FROM oauth_access_tokens at
        #     LEFT JOIN users u ON at.user_id = u.id
        #     WHERE u.id IS NULL
        # """
        # cursor.execute(sql_orphaned)
        # logging.info(f"删除了 {cursor.rowcount} 个孤立的访问令牌")
        connection.commit()
        logging.info("令牌清理完成")
    except Exception as e:
        logging.error(f"清理令牌时出错: {e}")
        if connection:
            connection.rollback()
    finally:
        if cursor:
            cursor.close()
        if connection:
            connection.close()
if __name__ == "__main__":
    # 每天凌晨2点执行
    cleanup_stale_oauth_tokens(days_inactive=90)

部署方式：将此脚本设置为一个定时任务（Linux Crontab / Windows Task Scheduler）。

通过Identity Provider (IdP) API 清理（更安全，适用于云服务或标准授权服务器）

如果你的OAuth服务器有管理API（如Auth0, Keycloak, Okta, AWS Cognito），直接调用API是更佳选择,因为数据库结构可能不开放。

示例脚本（Python + Requests, 针对Keycloak）：

import requests
from datetime import datetime, timedelta
import logging
logging.basicConfig(level=logging.INFO)
class KeycloakTokenCleaner:
    def __init__(self, base_url, realm, client_id, client_secret):
        self.base_url = base_url.rstrip('/')
        self.realm = realm
        self.client_id = client_id
        self.client_secret = client_secret
        self.management_token = None
    def get_management_token(self):
        """获取管理员级别的access_token"""
        token_url = f"{self.base_url}/realms/{self.realm}/protocol/openid-connect/token"
        payload = {
            'client_id': self.client_id,
            'client_secret': self.client_secret,
            'grant_type': 'client_credentials',
            'scope': 'openid'  # 根据Keycloak角色配置
        }
        response = requests.post(token_url, data=payload)
        response.raise_for_status()
        self.management_token = response.json()['access_token']
        logging.info("成功获取管理令牌")
    def revoke_expired_tokens(self, days_inactive=90):
        """通过Keycloak Admin API撤销长期未使用的用户会话（从而失效其令牌）"""
        if not self.management_token:
            self.get_management_token()
        headers = {'Authorization': f'Bearer {self.management_token}'}
        # 获取所有用户（注意分页）
        users_url = f"{self.base_url}/admin/realms/{self.realm}/users"
        response = requests.get(users_url, headers=headers, params={'max': 1000})
        response.raise_for_status()
        users = response.json()
        cutoff_date = datetime.now() - timedelta(days=days_inactive)
        revoked_count = 0
        for user in users:
            user_id = user['id']
            # 获取用户的会话列表
            sessions_url = f"{self.base_url}/admin/realms/{self.realm}/users/{user_id}/sessions"
            sessions_resp = requests.get(sessions_url, headers=headers)
            sessions_resp.raise_for_status()
            sessions = sessions_resp.json()
            for session in sessions:
                # last_access 通常是Unix时间戳
                last_access = datetime.fromtimestamp(session['lastAccess'] / 1000)  # 假设毫秒
                if last_access < cutoff_date:
                    # 撤销该会话
                    logout_url = f"{self.base_url}/admin/realms/{self.realm}/sessions/{session['id']}"
                    logout_resp = requests.delete(logout_url, headers=headers)
                    if logout_resp.status_code == 204:
                        revoked_count += 1
                        logging.info(f"已撤销用户 {user['username']} 的会话 {session['id']} (最后活跃: {last_access})")
                    else:
                        logging.error(f"撤销会话 {session['id']} 失败: {logout_resp.text}")
        logging.info(f"共撤销了 {revoked_count} 个长期未使用的会话")
        return revoked_count
# 使用示例
cleaner = KeycloakTokenCleaner(
    base_url='https://your-keycloak-server:8443/auth',
    realm='your-realm',
    client_id='admin-cli',  # 或你的管理客户端
    client_secret='your-admin-secret'
)
cleaner.revoke_expired_tokens(days_inactive=90)

优势：基于IdP的lastAccess时间，更准确；无需直接操作数据库。

基于日志分析（无最后使用时间字段）

如果你的系统没有记录last_used_at，但你可以访问访问日志（如API网关日志、Nginx日志）。

脚本逻辑：

从日志中提取所有最近N天内被使用过的令牌ID（Authorization: Bearer xxx）。
从数据库中查询所有已过期令牌，但不在日志列表中，这些日志里从未出现过的过期令牌理论上就是未使用的，但需注意：如果令牌在有效期内但从未被使用，它可能就是“僵尸”。
更激进的策略是：直接删除所有已过期的令牌，因为已过期的令牌本身就是无效的,清理它们不会有副作用。

示例（Shell + SQL）：

#!/bin/bash
# 清理所有已过期的令牌（数据库已记录expires_at）
mysql -u db_user -p'password' oauth_db -e "
DELETE FROM oauth_access_tokens WHERE expires_at < NOW();
DELETE FROM oauth_refresh_tokens WHERE expires_at < NOW();
"
echo "所有已过期的令牌已清理"

最佳实践建议

分步执行，小范围测试：
- 先对生产环境的一小部分用户（如测试用户）执行,观察24小时是否出现异常。
- 先执行SELECT COUNT(*)统计可能会被删除的令牌数,确认影响范围。
保留必要日志：
- 记录被删除的令牌ID、用户ID、过期时间、清理日期。
- 为安全审计提供依据。
通知用户（可选但推荐）：
- 如果清理的是仍未过期但长期未使用的刷新令牌，建议发送邮件通知用户：“您的应用授权令牌已因长期未使用被撤销，请重新授权”。
- 如果仅清理已过期的令牌,通常无需通知。
设置合理的保留期：
- 过期的访问令牌（通常只有1小时有效期）：立即清理,不影响用户。
- 过期的刷新令牌（有效期通常30天或更长）：清理后用户需要重新登录，建议保留期设置为刷新令牌过期后1-7天，以便用户在此期间若意外断连，仍有机会刷新（但技术上过期后无法刷新）。
- 长期未使用的有效令牌：建议不活跃周期为90~180天（取决于应用类型，高频应用可设30天）。
监控清理效果：

清理前后，观察数据库表记录数、API响应时间、授权成功率。

最简单方案：直接运行 DELETE FROM oauth_access_tokens WHERE expires_at < NOW();（Shell脚本 + 定时任务）。
推荐方案：方案一（数据库SQL）或方案二（调用IdP API），并结合“未使用”策略。
安全底线：绝对不要手动删除未过期的令牌，除非你100%确定该用户已不在系统中或已手动撤销。