收集 DigiCert 审核日志
支持的平台：
Google SecOps
SIEM
本文档介绍了如何使用 Google Cloud Storage 将 DigiCert 审核日志提取到 Google Security Operations。DigiCert CertCentral 是一个证书生命周期管理平台，可提供证书操作、用户活动和管理操作的审核日志。
准备工作
确保您满足以下前提条件：
- Google SecOps 实例
- 已启用 Cloud Storage API 的 GCP 项目
- 创建和管理 GCS 存储分区的权限
- 管理 GCS 存储分区的 IAM 政策的权限
- 创建 Cloud Run 服务、Pub/Sub 主题和 Cloud Scheduler 作业的权限
- 对 DigiCert CertCentral 的特权访问权限（具有管理员角色的 API 密钥）
创建 Google Cloud Storage 存储分区
- 前往 Google Cloud 控制台。
- 选择您的项目或创建新项目。
- 在导航菜单中，依次前往 Cloud Storage > 存储分区。
- 点击创建存储分区。
提供以下配置详细信息：
设置 值 为存储分区命名 输入一个全局唯一的名称（例如 digicert-logs）位置类型 根据您的需求进行选择（区域级、双区域级、多区域级） 位置 选择相应位置（例如 us-central1）存储类别 标准（建议用于经常访问的日志） 访问权限控制 统一（推荐） 保护工具 可选：启用对象版本控制或保留政策 点击创建。
收集 DigiCert API 凭据
获取 DigiCert API 密钥
- 登录 DigiCert CertCentral。
- 依次前往账号 > API 密钥。
- 点击 Create API Key。
- 提供以下配置详细信息：
- 名称：输入一个描述性名称（例如
Chronicle Integration）。 - 角色：选择管理员。
- 名称：输入一个描述性名称（例如
- 点击创建。
- 复制并保存 API 密钥 (
X-DC-DEVKEY)。此值不会再次显示。
获取 DigiCert 报告 ID
- 在 DigiCert CertCentral 中，依次前往报告 > 报告库。
- 点击创建报告。
- 提供以下配置详细信息：
- 报告类型：选择审核日志。
- 格式：选择 JSON。
- 名称：输入一个描述性名称（例如
Chronicle Audit Logs）。
- 点击创建。
复制并保存报告 ID（UUID 格式）。
为 Cloud Run 函数创建服务账号
Cloud Run 函数需要一个服务账号，该账号具有向 GCS 存储分区写入内容以及被 Pub/Sub 调用的权限。
创建服务账号
- 在 GCP 控制台中，依次前往 IAM 和管理 > 服务账号。
- 点击创建服务账号。
- 提供以下配置详细信息：
- 服务账号名称：输入
digicert-logs-collector-sa。 - 服务账号说明：输入
Service account for Cloud Run function to collect DigiCert audit logs。
- 服务账号名称：输入
- 点击创建并继续。
- 在向此服务账号授予对项目的访问权限部分中，添加以下角色：
- 点击选择角色。
- 搜索并选择 Storage Object Admin。
- 点击 + 添加其他角色。
- 搜索并选择 Cloud Run Invoker。
- 点击 + 添加其他角色。
- 搜索并选择 Cloud Functions Invoker。
- 点击继续。
- 点击完成。
必须拥有这些角色，才能：
- Storage Object Admin：将日志写入 GCS 存储分区并管理状态文件
- Cloud Run Invoker：允许 Pub/Sub 调用函数
- Cloud Functions Invoker：允许调用函数
授予对 GCS 存储分区的 IAM 权限
向服务账号授予对 GCS 存储分区的写入权限：
- 前往 Cloud Storage > 存储分区。
- 点击您的存储分区名称。
- 前往权限标签页。
- 点击授予访问权限。
- 提供以下配置详细信息：
- 添加主账号：输入服务账号电子邮件地址（例如
digicert-logs-collector-sa@PROJECT_ID.iam.gserviceaccount.com）。 - 分配角色：选择 Storage Object Admin。
- 添加主账号：输入服务账号电子邮件地址（例如
- 点击保存。
创建发布/订阅主题
创建一个 Pub/Sub 主题，供 Cloud Scheduler 发布消息，并供 Cloud Run 函数订阅。
- 在 GCP 控制台中，前往 Pub/Sub > 主题。
- 点击创建主题。
- 提供以下配置详细信息：
- 主题 ID：输入
digicert-audit-trigger。 - 将其他设置保留为默认值。
- 主题 ID：输入
- 点击创建。
创建 Cloud Run 函数以收集日志
Cloud Run 函数由来自 Cloud Scheduler 的 Pub/Sub 消息触发，用于从 DigiCert API 中提取日志并将其写入 GCS。
- 在 GCP 控制台中，前往 Cloud Run。
- 点击创建服务。
- 选择函数（使用内嵌编辑器创建函数）。
在配置部分中，提供以下配置详细信息：
设置 值 Service 名称 digicert-audit-logs-collector区域 选择与您的 GCS 存储分区匹配的区域（例如 us-central1）运行时 选择 Python 3.12 或更高版本 在触发器（可选）部分中：
- 点击 + 添加触发器。
- 选择 Cloud Pub/Sub。
- 在选择 Cloud Pub/Sub 主题中，选择 Pub/Sub 主题 (
digicert-audit-trigger)。 - 点击保存。
在身份验证部分中：
- 选择需要进行身份验证。
- 检查 Identity and Access Management (IAM)。
向下滚动并展开容器、网络、安全性。
前往安全标签页：
- 服务账号：选择服务账号 (
digicert-logs-collector-sa)。
- 服务账号：选择服务账号 (
前往容器标签页：
- 点击变量和密钥。
- 为每个环境变量点击 + 添加变量：
变量名称 示例值 GCS_BUCKETdigicert-logsGCS_PREFIXdigicert/logsSTATE_KEYdigicert/logs/state.jsonDIGICERT_API_KEYxxxxxxxxxxxxxxxxxxxxxxxxDIGICERT_REPORT_ID88de5e19-ec57-4d70-865d-df953b062574REQUEST_TIMEOUT30POLL_INTERVAL10MAX_WAIT_SECONDS300在变量和密钥标签页中，向下滚动到请求：
- 请求超时：输入
900秒（15 分钟）。
- 请求超时：输入
前往容器中的设置标签页：
- 在资源部分中：
- 内存：选择 512 MiB 或更高值。
- CPU：选择 1。
- 点击完成。
- 在资源部分中：
滚动到执行环境：
- 选择默认（推荐）。
在修订版本扩缩部分中：
- 实例数下限：输入
0。 - 实例数上限：输入
100（或根据预期负载进行调整）。
- 实例数下限：输入
点击创建。
等待服务创建完成（1-2 分钟）。
创建服务后，系统会自动打开内嵌代码编辑器。
添加函数代码
- 在函数入口点中输入 main
在内嵌代码编辑器中，创建两个文件：
- 第一个文件：main.py:：
import functions_framework from google.cloud import storage import json import os import urllib3 from datetime import datetime, timezone, timedelta import time import io import gzip import zipfile import uuid # Initialize HTTP client http = urllib3.PoolManager() # Initialize Storage client storage_client = storage.Client() API_BASE = "https://api.digicert.com/reports/v1" USER_AGENT = "secops-digicert-reports/1.0" @functions_framework.cloud_event def main(cloud_event): """ Cloud Run function triggered by Pub/Sub to fetch DigiCert audit logs and write to GCS. Args: cloud_event: CloudEvent object containing Pub/Sub message """ # Get environment variables bucket_name = os.environ.get('GCS_BUCKET') prefix = os.environ.get('GCS_PREFIX', 'digicert/logs').rstrip('/') state_key = os.environ.get('STATE_KEY', f'{prefix}/state.json') api_key = os.environ.get('DIGICERT_API_KEY') report_id = os.environ.get('DIGICERT_REPORT_ID') max_wait = int(os.environ.get('MAX_WAIT_SECONDS', '300')) poll_int = int(os.environ.get('POLL_INTERVAL', '10')) timeout = int(os.environ.get('REQUEST_TIMEOUT', '30')) if not all([bucket_name, api_key, report_id]): print('Error: Missing required environment variables') return try: # Get GCS bucket bucket = storage_client.bucket(bucket_name) # Load state state = load_state(bucket, state_key) last_run = state.get('last_run_id') # Start report run started = datetime.now(timezone.utc) start_report_run(api_key, report_id, timeout) # Wait for report to be ready run_id = find_ready_run(api_key, report_id, started, timeout, max_wait, poll_int) # Skip if same run as last time if last_run and last_run == run_id: print(f'Skipping duplicate run: {run_id}') return # Get report data rows = get_json_rows(api_key, report_id, run_id, timeout) # Write to GCS key = write_ndjson_gz(bucket, prefix, rows, run_id) # Update state save_state(bucket, state_key, { 'last_run_id': run_id, 'last_success_at': datetime.now(timezone.utc).isoformat(), 'last_s3_key': key, 'rows_count': len(rows) }) print(f'Successfully processed {len(rows)} logs to {key}') except Exception as e: print(f'Error processing logs: {str(e)}') raise def http_request(method, url, api_key, body=None, timeout=30, max_retries=5): """Make HTTP request with retry logic.""" headers = { 'X-DC-DEVKEY': api_key, 'Content-Type': 'application/json', 'User-Agent': USER_AGENT } attempt, backoff = 0, 1.0 while True: try: response = http.request( method, url, headers=headers, body=body, timeout=timeout ) status = response.status # Retry on server errors if 500 <= status <= 599 and attempt < max_retries: attempt += 1 time.sleep(backoff) backoff *= 2 continue # Retry on rate limit if status == 429 and attempt < max_retries: retry_after = response.headers.get('Retry-After') delay = float(retry_after) if retry_after and retry_after.isdigit() else backoff attempt += 1 time.sleep(delay) backoff *= 2 continue if status not in (200, 201): raise RuntimeError(f'HTTP {status}: {response.data[:200]}') return status, response.headers, response.data except urllib3.exceptions.HTTPError as e: if attempt < max_retries: attempt += 1 time.sleep(backoff) backoff *= 2 continue raise def start_report_run(api_key, report_id, timeout): """Start a new report run.""" status, _, body = http_request( 'POST', f'{API_BASE}/report/{report_id}/run', api_key, b'{}', timeout ) if status not in (200, 201): raise RuntimeError(f'Start run failed: {status} {body[:200]}') def list_report_history(api_key, status_filter=None, report_type=None, limit=100, timeout=30): """List report history.""" params = { 'limit': str(limit), 'offset': '0', 'sort_by': 'report_start_date', 'sort_direction': 'DESC' } if status_filter: params['status'] = status_filter if report_type: params['report_type'] = report_type query_string = '&'.join([f'{k}={v}' for k, v in params.items()]) url = f'{API_BASE}/report/history?{query_string}' status, _, body = http_request('GET', url, api_key, timeout=timeout) if status != 200: raise RuntimeError(f'History failed: {status} {body[:200]}') return json.loads(body.decode('utf-8')) def find_ready_run(api_key, report_id, started_not_before, timeout, max_wait_seconds, poll_interval): """Find a ready report run.""" deadline = time.time() + max_wait_seconds while time.time() < deadline: hist = list_report_history( api_key, status_filter='READY', limit=200, timeout=timeout ).get('report_history', []) for item in hist: if item.get('report_identifier') != report_id: continue if not item.get('report_run_identifier'): continue try: rsd = datetime.strptime( item.get('report_start_date', ''), '%Y-%m-%d %H:%M:%S' ).replace(tzinfo=timezone.utc) except Exception: rsd = started_not_before if rsd + timedelta(seconds=60) >= started_not_before: return item['report_run_identifier'] time.sleep(poll_interval) raise TimeoutError('READY run not found in time') def get_json_rows(api_key, report_id, run_id, timeout): """Get JSON rows from report.""" status, headers, body = http_request( 'GET', f'{API_BASE}/report/{report_id}/{run_id}/json', api_key, timeout=timeout ) if status != 200: raise RuntimeError(f'Get JSON failed: {status} {body[:200]}') # Check if response is ZIP content_type = headers.get('content-type', '').lower() if 'application/zip' in content_type or body[:2] == b'PK': with zipfile.ZipFile(io.BytesIO(body)) as zf: json_files = [n for n in zf.namelist() if n.lower().endswith('.json')] if not json_files: raise RuntimeError('ZIP has no JSON') rows = json.loads(zf.read(json_files[0]).decode('utf-8')) else: rows = json.loads(body.decode('utf-8')) if not isinstance(rows, list): raise RuntimeError('Unexpected JSON format') return rows def write_ndjson_gz(bucket, prefix, rows, run_id): """Write NDJSON gzipped file to GCS.""" ts = datetime.now(timezone.utc).strftime('%Y/%m/%d/%H%M%S') key = f'{prefix}/{ts}-digicert-audit-{run_id[:8]}-{uuid.uuid4().hex}.json.gz' buf = io.BytesIO() with gzip.GzipFile(fileobj=buf, mode='wb') as gz: for r in rows: gz.write((json.dumps(r, separators=(',', ':')) + '\n').encode('utf-8')) blob = bucket.blob(key) blob.upload_from_string( buf.getvalue(), content_type='application/x-ndjson', content_encoding='gzip' ) return key def load_state(bucket, key): """Load state from GCS.""" try: blob = bucket.blob(key) if blob.exists(): state_data = blob.download_as_text() return json.loads(state_data) except Exception as e: print(f'Warning: Could not load state: {str(e)}') return {} def save_state(bucket, key, state): """Save state to GCS.""" try: blob = bucket.blob(key) blob.upload_from_string( json.dumps(state), content_type='application/json' ) except Exception as e: print(f'Warning: Could not save state: {str(e)}')- 第二个文件：requirements.txt:：
functions-framework==3.* google-cloud-storage==2.* urllib3>=2.0.0点击部署以保存并部署该函数。
等待部署完成（2-3 分钟）。
创建 Cloud Scheduler 作业
Cloud Scheduler 会定期向 Pub/Sub 主题发布消息，从而触发 Cloud Run 函数。
- 在 GCP Console 中，前往 Cloud Scheduler。
- 点击创建作业。
提供以下配置详细信息：
设置 值 名称 digicert-audit-hourly区域 选择与 Cloud Run 函数相同的区域 频率 0 * * * *（每小时一次，在整点时）时区 选择时区（建议选择世界协调时间 [UTC]） 目标类型 Pub/Sub 主题 选择 Pub/Sub 主题 ( digicert-audit-trigger)消息正文 {}（空 JSON 对象）点击创建。
时间表频率选项
根据日志量和延迟时间要求选择频次：
频率 Cron 表达式 使用场景 每隔 5 分钟 */5 * * * *高容量、低延迟 每隔 15 分钟 */15 * * * *搜索量中等 每小时 0 * * * *标准（推荐） 每 6 小时 0 */6 * * *量小、批处理 每天 0 0 * * *历史数据收集
测试调度器作业
- 在 Cloud Scheduler 控制台中，找到您的作业。
- 点击强制运行以手动触发。
- 等待几秒钟，然后前往 Cloud Run > 服务 > digicert-audit-logs-collector > 日志。
- 验证函数是否已成功执行。
- 检查 GCS 存储分区，确认日志已写入。
检索 Google SecOps 服务账号
Google SecOps 使用唯一的服务账号从您的 GCS 存储分区中读取数据。您必须授予此服务账号对您的存储分区的访问权限。
获取服务账号电子邮件地址
- 依次前往 SIEM 设置 > Feed。
- 点击添加新 Feed。
- 点击配置单个 Feed。
- 在Feed 名称字段中，输入 Feed 的名称（例如
DigiCert Audit Logs）。 - 选择 Google Cloud Storage V2 作为来源类型。
- 选择 Digicert 作为日志类型。
点击获取服务账号。系统会显示一个唯一的服务账号电子邮件地址，例如：
chronicle-12345678@chronicle-gcp-prod.iam.gserviceaccount.com复制此电子邮件地址，以便在下一步中使用。
向 Google SecOps 服务账号授予 IAM 权限
Google SecOps 服务账号需要对您的 GCS 存储分区具有 Storage Object Viewer 角色。
- 前往 Cloud Storage > 存储分区。
- 点击您的存储分区名称。
- 前往权限标签页。
- 点击授予访问权限。
- 提供以下配置详细信息：
- 添加主账号：粘贴 Google SecOps 服务账号电子邮件地址。
- 分配角色：选择 Storage Object Viewer。
点击保存。
在 Google SecOps 中配置 Feed 以注入 DigiCert 日志
- 依次前往 SIEM 设置 > Feed。
- 点击添加新 Feed。
- 点击配置单个 Feed。
- 在Feed 名称字段中，输入 Feed 的名称（例如
DigiCert Audit Logs）。 - 选择 Google Cloud Storage V2 作为来源类型。
- 选择 Digicert 作为日志类型。
- 点击下一步。
为以下输入参数指定值：
存储分区网址：输入带有前缀路径的 GCS 存储分区 URI：
gs://digicert-logs/digicert/logs/将
digicert-logs：您的 GCS 存储分区名称。digicert/logs：存储日志的前缀/文件夹路径。
来源删除选项：根据您的偏好选择删除选项：
- 永不：永不删除转移后的任何文件（建议用于测试）。
- 删除已转移的文件：在成功转移后删除文件。
删除已转移的文件和空目录：在成功转移后删除文件和空目录。
文件存在时间上限：包含在过去指定天数内修改的文件。默认值为 180 天。
资产命名空间：资产命名空间。
注入标签：要应用于此 Feed 中事件的标签。
点击下一步。
在最终确定界面中查看新的 Feed 配置，然后点击提交。
需要更多帮助？获得社区成员和 Google SecOps 专业人士的解答。