关于使用haveibeenpwned.com API每日监控Google Workspace账号泄露的问询

完全可以用Google Cloud实现你的需求，下面是一套落地的步骤，结合Google Cloud原生服务和简单代码就能搞定：

一、核心组件选型 #

我们会用到这几个Google Cloud服务：

• Cloud Functions：处理账号获取、API调用、结果解析和邮件发送的核心逻辑
• Cloud Scheduler：设置每日定时触发任务
• Admin SDK：拉取Google Workspace的活跃账号列表
• Secret Manager：安全存储HaveIBeenPwned的API密钥和服务账号凭证
• Gmail API：发送告警邮件报告

二、具体实现步骤 #

1. 准备工作：启用服务并配置权限 #

• 在Google Cloud Console启用Admin SDK、Gmail API、Cloud Functions、Cloud Scheduler和Secret Manager
• 创建服务账号，给它分配Workspace域范围的权限（需要Workspace管理员授权），允许它读取用户列表
• 将HaveIBeenPwned的API密钥、服务账号的JSON密钥文件存储到Secret Manager中，避免硬编码

2. 编写Cloud Function核心逻辑（示例用Python） #

第一步：拉取活跃Workspace账号

调用Admin SDK的Directory API，过滤出未暂停且近期活跃的账号：

from googleapiclient.discovery import build
from google.oauth2 import service_account

def get_active_workspace_users():
    # 从Secret Manager加载服务账号密钥
    credentials = service_account.Credentials.from_service_account_file(
        'service-account-key.json',
        scopes=['https://www.googleapis.com/auth/admin.directory.user.readonly']
    )
    delegated_credentials = credentials.with_subject('admin@your-domain.com')
    service = build('admin', 'directory_v1', credentials=delegated_credentials)
    
    # 查询未暂停的用户，可根据lastLoginTime过滤活跃用户
    results = service.users().list(customer='my_customer', query='isSuspended=false').execute()
    users = results.get('users', [])
    return [user['primaryEmail'] for user in users]

第二步：调用HaveIBeenPwned API查询新增泄露

利用购买的API密钥，指定since参数（昨天的时间戳）来获取新增泄露事件：

import requests
import os
from datetime import datetime, timedelta

def check_pwned_accounts(emails):
    hibp_api_key = os.environ.get('HIBP_API_KEY')
    headers = {'hibp-api-key': hibp_api_key, 'user-agent': 'Google Cloud Pwned Checker'}
    since_date = (datetime.now() - timedelta(days=1)).strftime('%Y-%m-%dT%H:%M:%SZ')
    pwned_results = []
    
    for email in emails:
        try:
            response = requests.get(
                f'https://haveibeenpwned.com/api/v3/breachedaccount/{email}',
                headers=headers,
                params={'since': since_date, 'includeUnverified': False}
            )
            if response.status_code == 200:
                breaches = response.json()
                if breaches:
                    pwned_results.append({
                        'email': email,
                        'breaches': [{'name': b['Name'], 'date': b['BreachDate']} for b in breaches]
                    })
        except Exception as e:
            print(f"Error checking {email}: {str(e)}")
    return pwned_results

第三步：生成报告并发送邮件

用Gmail API发送整理好的告警邮件：

import base64
from googleapiclient.discovery import build

def send_alert_email(pwned_results, delegated_credentials):
    if not pwned_results:
        return
    
    # 构建邮件内容
    email_content = "### 每日数据泄露告警报告\n\n"
    for result in pwned_results:
        email_content += f"- **账号**: {result['email']}\n"
        email_content += "  新增泄露事件:\n"
        for breach in result['breaches']:
            email_content += f"    - {breach['name']} (泄露日期: {breach['date']})\n"
    
    # 调用Gmail API发送邮件
    service = build('gmail', 'v1', credentials=delegated_credentials)
    message = {
        'raw': base64.urlsafe_b64encode(
            f"From: Alert Bot <admin@your-domain.com>\n"
            f"To: security-team@your-domain.com\n"
            f"Subject: 每日数据泄露告警\n\n"
            f"{email_content}".encode('utf-8')
        ).decode('utf-8')
    }
    service.users().messages().send(userId='me', body=message).execute()

3. 设置定时触发 #

在Cloud Scheduler中创建一个每日定时任务（比如每天凌晨2点），选择HTTP目标指向你的Cloud Function URL，或者用Pub/Sub触发器触发Cloud Function，确保任务自动执行。

4. 额外注意事项 #

• 速率限制：HaveIBeenPwned API有调用频率限制，批量查询时可添加短暂延迟（比如每个请求间隔1秒），避免触发限流
• 日志与监控：在Cloud Function中添加日志输出，用Cloud Monitoring设置告警，确保任务执行异常时能及时发现
• 数据过滤：可根据需求调整Workspace账号的过滤条件（比如只查询特定组织单元的账号）

内容的提问来源于stack exchange，提问作者Whitney Christie