收集 DigiCert 稽核記錄
剖析器版本：1.0
支援的服務：
Google SecOps
SIEM
本文說明如何使用 Google Cloud Storage，將 DigiCert 稽核記錄擷取至 Google Security Operations。DigiCert CertCentral 是憑證生命週期管理平台，提供憑證作業、使用者活動和管理動作的稽核記錄。
事前準備
請確認您符合下列必要條件：
- Google SecOps 執行個體
- 已啟用 Cloud Storage API 的 GCP 專案
- 建立及管理 GCS bucket 的權限
- 具備管理 GCS bucket IAM 政策的權限
- 建立 Cloud Run 服務、Pub/Sub 主題和 Cloud Scheduler 工作的權限
- DigiCert CertCentral 的特殊權限存取權 (具備管理員角色的 API 金鑰)
建立 Google Cloud Storage bucket
- 前往 Google Cloud 控制台。
- 選取專案或建立新專案。
- 在導覽選單中，依序前往「Cloud Storage」>「Bucket」。
- 按一下「建立值區」。
提供下列設定詳細資料：
設定 值 為 bucket 命名 輸入全域不重複的名稱 (例如 digicert-logs)位置類型 請根據需求選擇 (單一區域、雙區域、多區域) 位置 選取位置 (例如 us-central1)儲存空間級別 標準 (建議用於經常存取的記錄) 存取控管 統一 (建議) 保護工具 選用：啟用物件版本管理或資料保留政策 點選「建立」。
收集 DigiCert API 憑證
取得 DigiCert API 金鑰
- 登入 DigiCert CertCentral。
- 依序前往「Account」>「API Keys」。
- 點選「建立 API 金鑰」。
- 提供下列設定詳細資料：
- 名稱：輸入描述性名稱 (例如
Chronicle Integration)。 - 角色：選取「管理員」。
- 名稱：輸入描述性名稱 (例如
- 點選「建立」。
- 複製並儲存 API 金鑰 (
X-DC-DEVKEY)。這個值不會再次顯示。
取得 DigiCert 報告 ID
- 在 DigiCert CertCentral 中，依序前往「Reports」>「Report Library」。
- 點選 [Create Report] (建立報表)。
- 提供下列設定詳細資料：
- 報表類型：選取「稽核記錄」。
- 格式：選取「JSON」。
- 名稱：輸入描述性名稱 (例如
Chronicle Audit Logs)。
- 點選「建立」。
複製並儲存檢舉 ID (UUID 格式)。
建立 Cloud Run 函式的服務帳戶
Cloud Run 函式需要服務帳戶，且該帳戶須具備寫入 GCS bucket 的權限，並可由 Pub/Sub 叫用。
建立服務帳戶
- 在 GCP 控制台中，依序前往「IAM & Admin」(IAM 與管理) >「Service Accounts」(服務帳戶)。
- 按一下「Create Service Account」(建立服務帳戶)。
- 提供下列設定詳細資料：
- 服務帳戶名稱：輸入
digicert-logs-collector-sa。 - 「Service account description」(服務帳戶說明)：輸入
Service account for Cloud Run function to collect DigiCert audit logs。
- 服務帳戶名稱：輸入
- 按一下「建立並繼續」。
- 在「將專案存取權授予這個服務帳戶」部分，新增下列角色：
- 按一下「請選擇角色」。
- 搜尋並選取「Storage 物件管理員」。
- 點選「+ 新增其他角色」。
- 搜尋並選取「Cloud Run Invoker」。
- 點選「+ 新增其他角色」。
- 搜尋並選取「Cloud Functions 叫用者」。
- 按一下「繼續」。
- 按一下「完成」。
這些角色適用於：
- Storage 物件管理員：將記錄檔寫入 GCS bucket 並管理狀態檔案
- Cloud Run 叫用者：允許 Pub/Sub 叫用函式
- Cloud Functions 叫用者：允許叫用函式
授予 GCS bucket 的 IAM 權限
授予服務帳戶 GCS bucket 的寫入權限：
- 依序前往「Cloud Storage」>「Bucket」。
- 按一下 bucket 名稱。
- 前往「權限」分頁。
- 按一下「授予存取權」。
- 提供下列設定詳細資料：
- 新增主體：輸入服務帳戶電子郵件地址 (例如
digicert-logs-collector-sa@PROJECT_ID.iam.gserviceaccount.com)。 - 指派角色：選取「Storage 物件管理員」。
- 新增主體：輸入服務帳戶電子郵件地址 (例如
- 按一下「儲存」。
建立 Pub/Sub 主題
建立 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」。
- 按一下「Create service」(建立服務)。
- 選取「函式」(使用直接修改建立函式)。
在「Configure」(設定) 部分，提供下列設定詳細資料：
設定 值 服務名稱 digicert-audit-logs-collector區域 選取與 GCS bucket 相符的區域 (例如 us-central1)執行階段 選取「Python 3.12」或更新版本 在「Trigger (optional)」(觸發條件 (選用)) 部分：
- 按一下「+ 新增觸發條件」。
- 選取「Cloud Pub/Sub」。
- 在「Select a Cloud Pub/Sub topic」(選取 Cloud Pub/Sub 主題) 中，選擇 Pub/Sub 主題 (
digicert-audit-trigger)。 - 按一下「儲存」。
在「Authentication」(驗證) 部分：
- 選取「需要驗證」。
- Identity and Access Management (IAM)。
向下捲動並展開「容器、網路、安全性」。
前往「安全性」分頁：
- 服務帳戶：選取服務帳戶 (
digicert-logs-collector-sa)。
- 服務帳戶：選取服務帳戶 (
前往「容器」分頁：
- 按一下「Variables & Secrets」(變數與密鑰)。
- 針對每個環境變數，按一下「+ 新增變數」：
變數名稱 範例值 GCS_BUCKETdigicert-logsGCS_PREFIXdigicert/logsSTATE_KEYdigicert/logs/state.jsonDIGICERT_API_KEYxxxxxxxxxxxxxxxxxxxxxxxxDIGICERT_REPORT_ID88de5e19-ec57-4d70-865d-df953b062574REQUEST_TIMEOUT30POLL_INTERVAL10MAX_WAIT_SECONDS300在「變數與密鑰」分頁中，向下捲動至「要求」：
- Request timeout：輸入
900秒 (15 分鐘)。
- Request timeout：輸入
前往「容器」的「設定」分頁：
- 在「Resources」(資源) 部分：
- 「記憶體」：選取「512 MiB」或以上。
- CPU：選取「1」。
- 按一下「完成」。
- 在「Resources」(資源) 部分：
捲動至「Execution environment」(執行環境)：
- 選取「預設」(建議)。
在「Revision scaling」(修訂版本資源調度) 部分：
- 「Minimum number of instances」(執行個體數量下限)：輸入
0。 - 「執行個體數量上限」：輸入
100(或依預期負載調整)。
- 「Minimum number of instances」(執行個體數量下限)：輸入
點選「建立」。
等待服務建立完成 (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 控制台中，前往「Cloud Scheduler」。
- 點選「建立工作」。
提供下列設定詳細資料：
設定 值 名稱 digicert-audit-hourly區域 選取與 Cloud Run 函式相同的區域 頻率 0 * * * *(每小時整點)時區 選取時區 (建議選取世界標準時間) 目標類型 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 bucket，確認記錄已寫入。
擷取 Google SecOps 服務帳戶
Google SecOps 會使用專屬服務帳戶，從 GCS bucket 讀取資料。您必須授予這個服務帳戶值區存取權。
取得服務帳戶電子郵件地址
- 依序前往「設定」>「資料管理」>「動態饋給」。如果使用的是 SIEM 專用平台，請依序前往「設定」>「動態饋給」。
- 按一下「新增動態饋給」。
- 按一下「設定單一動態饋給」。
- 在「Feed name」(動態饋給名稱) 欄位中，輸入動態饋給名稱 (例如
DigiCert Audit Logs)。 - 選取「Google Cloud Storage V2」做為「來源類型」。
- 在「Log type」(記錄類型) 中選取「Digicert」。
按一下「Get Service Account」。系統會顯示專屬服務帳戶電子郵件地址，例如：
chronicle-12345678@chronicle-gcp-prod.iam.gserviceaccount.com複製這個電子郵件地址，下一個步驟將會用到。
將 IAM 權限授予 Google SecOps 服務帳戶
Google SecOps 服務帳戶需要 GCS bucket 的 Storage 物件檢視者角色。
- 依序前往「Cloud Storage」>「Bucket」。
- 按一下 bucket 名稱。
- 前往「權限」分頁。
- 按一下「授予存取權」。
- 提供下列設定詳細資料：
- 新增主體：貼上 Google SecOps 服務帳戶電子郵件地址。
- 指派角色：選取「Storage 物件檢視者」。
按一下「儲存」。
在 Google SecOps 設定動態饋給，以便擷取 DigiCert 記錄
- 依序前往「設定」>「資料管理」>「動態饋給」。如果使用的是 SIEM 專用平台，請依序前往「設定」>「動態饋給」。
- 按一下「新增動態饋給」。
- 按一下「設定單一動態饋給」。
- 在「Feed name」(動態饋給名稱) 欄位中，輸入動態饋給名稱 (例如
DigiCert Audit Logs)。 - 選取「Google Cloud Storage V2」做為「來源類型」。
- 在「Log type」(記錄類型) 中選取「Digicert」。
- 點選「下一步」。
指定下列輸入參數的值：
Storage bucket URL：輸入 GCS bucket URI，並加上前置字元路徑：
gs://digicert-logs/digicert/logs/更改項目：
digicert-logs：GCS bucket 名稱。digicert/logs：：記錄檔的儲存前置字元/資料夾路徑。
來源刪除選項：根據偏好選取刪除選項：
- Never：轉移後一律不刪除任何檔案 (建議用於測試)。
- 刪除已轉移的檔案：在成功轉移後刪除檔案。
刪除已轉移的檔案和空白目錄：成功轉移後，刪除檔案和空白目錄。
檔案存在時間上限：包含過去幾天內修改過的檔案。預設值為 180 天。
Asset namespace：資產命名空間。
擷取標籤：要套用至這個動態饋給事件的標籤。
點選「下一步」。
在「完成」畫面中檢查新的新聞動態消息設定，然後按一下「提交」。
支援的 DigiCert 記錄範例
稽核/活動結構定義
{ "Account ID": "0000000", "Action": "User SSO login", "Action taken via": "CertCentral", "Activity": "User SSO login #9999999\nLogged through SAML SSO", "Adjustment/Ledger ID": "", "Date and time (UTC)": "06-Feb-2025 10:28:04", "Division/Container ID": "111222", "Division/Container name": "Global_Internal_Division", "Domain ID": "", "Domain name": "", "IP address": "192.168.1.100", "IP country": "US", "Message": "Logged through SAML SSO", "Order ID": "", "Order request ID": "", "Organization ID": "", "Organization name": "", "Primary organization name": "Example Corp, Inc.", "Result": "Successful", "Revocation comments": "", "User ID": "9999999", "User name": "John Doe", "Voucher code ID": "", "Voucher order ID": "" }網域中繼資料結構定義
{ "Domain ID": "8888888", "Domain name": "secure.example.com", "Organization ID": "7777777", "Organization name": "Example Corp, Inc.", "DCV email recipients": "", "DCV method": "email", "Domain approver email": "", "Domain expiration date (UTC)": "18-May-2025 08:36:42" }生命週期事件結構定義
{ "event": "certificate_issued", "data": { "order_id": 123456789, "certificate_id": 987654321 } }
變更記錄
還有其他問題嗎？向社群成員和 Google SecOps 專業人員尋求解答。