如何用Python实现GCP Policy Analyzer服务账号密钥查询并解决401问题

问题概述 #

• 可通过gcloud policy-intelligence query-activity --activity-type=serviceAccountKeyLastAuthentication --project=<project_name>命令查看服务账号密钥的使用情况
• 尝试用Python调用Policy Analyzer REST API时遭遇401认证错误，已启用对应API
• 现有脚本通过google.auth.default()完成其他GCP服务的认证，但Policy Analyzer无官方Python库，最终目标是获取组织内所有服务账号密钥的最后认证时间

解决方案 #

1. 为请求添加正确的认证头 #

你需要利用已获取的credentials生成授权令牌，并将其添加到requests请求的请求头中，具体步骤如下：

• 先从凭据中获取有效访问令牌（过期会自动刷新）
• 将令牌以Authorization: Bearer <token>的形式传入GET请求

示例代码：

import requests
from google.auth.transport.requests import Request

# 复用现有认证逻辑
credentials, project = google.auth.default()
# 确保凭据持有有效令牌
if not credentials.valid:
    credentials.refresh(Request())

# 构造请求URL
url = f"https://policyanalyzer.googleapis.com/v1/projects/{project}/locations/global/activityTypes/serviceAccountKeyLastAuthentication/activities:query"
# 构造过滤参数（如需指定单个密钥，可按此格式调整）
params = {
    "filter": f'activities.full_resource_name="//iam.googleapis.com/projects/{project}/serviceAccounts/{self.sa_email}/keys/{self.key_id}"'
}
# 添加认证头
headers = {
    "Authorization": f"Bearer {credentials.token}"
}

# 发送请求并处理响应
response = requests.get(url, params=params, headers=headers)
response.raise_for_status()  # 触发HTTP错误抛出
activity_data = response.json()

2. 获取组织内所有服务账号密钥的最后认证时间 #

要覆盖整个组织范围，需要遍历组织下的所有项目、服务账号及其密钥，再逐一查询认证时间，流程如下：

1. 使用Cloud Resource Manager API列出组织下的所有项目
2. 对每个项目，通过IAM API获取所有服务账号及对应的密钥
3. 对每个密钥，调用Policy Analyzer API查询最后认证时间

示例流程代码：

import googleapiclient.discovery
from google.auth.transport.requests import Request

# 初始化认证
credentials, _ = google.auth.default()
if not credentials.valid:
    credentials.refresh(Request())

# 1. 获取组织下的所有项目
crm = googleapiclient.discovery.build("cloudresourcemanager", "v3", credentials=credentials)
org_id = "your-organization-id"  # 替换为你的组织ID
projects = []
page_token = None
while True:
    response = crm.projects().list(filter=f"parent.type=organization parent.id={org_id}", pageToken=page_token).execute()
    projects.extend(response.get("projects", []))
    page_token = response.get("nextPageToken")
    if not page_token:
        break

# 2. 遍历项目、服务账号及密钥，查询认证时间
iam = googleapiclient.discovery.build("iam", "v1", credentials=credentials)
policy_analyzer_url = "https://policyanalyzer.googleapis.com/v1/projects/{}/locations/global/activityTypes/serviceAccountKeyLastAuthentication/activities:query"
headers = {"Authorization": f"Bearer {credentials.token}"}

for project in projects:
    project_id = project["projectId"]
    # 获取项目下所有服务账号
    sa_list = iam.projects().serviceAccounts().list(name=f"projects/{project_id}").execute().get("accounts", [])
    for sa in sa_list:
        sa_email = sa["email"]
        # 获取该服务账号的所有密钥
        key_list = iam.projects().serviceAccounts().keys().list(name=f"projects/{project_id}/serviceAccounts/{sa_email}").execute().get("keys", [])
        for key in key_list:
            key_id = key["name"].split("/")[-1]
            # 构造过滤条件并发起请求
            filter_str = f'activities.full_resource_name="//iam.googleapis.com/projects/{project_id}/serviceAccounts/{sa_email}/keys/{key_id}"'
            response = requests.get(
                policy_analyzer_url.format(project_id),
                params={"filter": filter_str},
                headers=headers
            )
            if response.status_code == 200:
                last_auth_time = response.json().get("activities", [{}])[0].get("eventTime")
                print(f"服务账号: {sa_email}, 密钥ID: {key_id}, 最后认证时间: {last_auth_time}")
            else:
                print(f"查询密钥 {key_id} 失败: {response.text}")

3. 必要的权限配置 #

确保使用的账号/服务账号拥有以下权限：

• policyanalyzer.activities.query：用于查询活动数据
• resourcemanager.projects.list：用于遍历组织下的项目
• iam.serviceAccounts.list和iam.serviceAccountKeys.list：用于列出服务账号及其密钥

可通过组合roles/policyanalyzer.viewer、roles/browser、roles/iam.serviceAccountViewer等IAM角色完成授权。

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