收集 DigiCert 稽核記錄
支援的國家/地區：
Google SecOps
SIEM
本文說明如何使用 Amazon S3，將 DigiCert 稽核記錄擷取至 Google Security Operations。
事前準備
- Google SecOps 執行個體
- DigiCert CertCentral 的特殊存取權 (具備管理員角色的 API 金鑰)
- AWS (S3、IAM、Lambda、EventBridge) 的具備權限存取權
取得 DigiCert API 金鑰和報表 ID
- 在 CertCentral 中，依序前往「帳戶」>「API 金鑰」，然後建立 API 金鑰 (
X-DC-DEVKEY)。 - 在「報表」> 報表庫中，建立 稽核記錄報表 (JSON 格式)，並記下報表 ID (UUID)。
- 您也可以使用報表記錄，找出現有報表的 ID。
為 Google SecOps 設定 AWS S3 值區和 IAM
- 按照本使用指南建立 Amazon S3 值區：建立值區
- 儲存 bucket 的「名稱」和「區域」，以供日後參考 (例如
digicert-logs)。 - 按照這份使用者指南建立使用者：建立 IAM 使用者。
- 選取建立的「使用者」。
- 選取「安全憑證」分頁標籤。
- 在「Access Keys」部分中，按一下「Create Access Key」。
- 選取「第三方服務」做為「用途」。
- 點選「下一步」。
- 選用：新增說明標記。
- 按一下「建立存取金鑰」。
- 按一下「下載 CSV 檔案」，儲存「存取金鑰」和「私密存取金鑰」，以供日後使用。
- 按一下 [完成]。
- 選取 [權限] 分頁標籤。
- 在「Permissions policies」(權限政策) 區段中，按一下「Add permissions」(新增權限)。
- 選取「新增權限」。
- 選取「直接附加政策」
- 搜尋並選取 AmazonS3FullAccess 政策。
- 點選「下一步」。
- 按一下「Add permissions」。
設定 S3 上傳的身分與存取權管理政策和角色
- 依序前往 AWS 管理控制台 > IAM > 政策 > 建立政策 > JSON 分頁。
輸入下列政策：
{ "Version": "2012-10-17", "Statement": [ { "Sid": "AllowPutDigiCertObjects", "Effect": "Allow", "Action": ["s3:PutObject"], "Resource": "arn:aws:s3:::digicert-logs/*" }, { "Sid": "AllowGetStateObject", "Effect": "Allow", "Action": ["s3:GetObject"], "Resource": "arn:aws:s3:::digicert-logs/digicert/logs/state.json" } ] }- 如果您輸入的值區名稱不同，請替換
digicert-logs。
- 如果您輸入的值區名稱不同，請替換
依序點選「Next」>「Create policy」。
依序前往「IAM」>「Roles」>「Create role」>「AWS service」>「Lambda」。
附加新建立的政策。
為角色命名
WriteDigicertToS3Role，然後按一下「建立角色」。
建立 Lambda 函式
- 在 AWS 控制台中，依序前往「Lambda」>「Functions」>「Create function」。
- 按一下「從頭開始撰寫」。
請提供下列設定詳細資料：
設定 值 名稱 digicert_audit_logs_to_s3執行階段 Python 3.13 架構 x86_64 執行角色 WriteDigicertToS3Role建立函式後，開啟「程式碼」分頁，刪除存根並輸入下列程式碼 (
digicert_audit_logs_to_s3.py)：#!/usr/bin/env python3 import datetime as dt, gzip, io, json, os, time, uuid, zipfile from typing import Any, Dict, Iterable, List, Tuple from urllib import request, parse, error import boto3 from botocore.exceptions import ClientError API_BASE = "https://api.digicert.com/reports/v1" USER_AGENT = "secops-digicert-reports/1.0" s3 = boto3.client("s3") def _now() -> dt.datetime: return dt.datetime.now(dt.timezone.utc) def _http(method: str, url: str, api_key: str, body: bytes | None = None, timeout: int = 30, max_retries: int = 5) -> Tuple[int, Dict[str,str], bytes]: headers = {"X-DC-DEVKEY": api_key, "Content-Type": "application/json", "User-Agent": USER_AGENT} attempt, backoff = 0, 1.0 while True: req = request.Request(url=url, method=method, headers=headers, data=body) try: with request.urlopen(req, timeout=timeout) as resp: status, h = resp.status, {k.lower(): v for k, v in resp.headers.items()} data = resp.read() if 500 <= status <= 599 and attempt < max_retries: attempt += 1; time.sleep(backoff); backoff *= 2; continue return status, h, data except error.HTTPError as e: status, h = e.code, {k.lower(): v for k, v in (e.headers or {}).items()} if status == 429 and attempt < max_retries: ra = h.get("retry-after"); delay = float(ra) if ra and ra.isdigit() else backoff attempt += 1; time.sleep(delay); backoff *= 2; continue if 500 <= status <= 599 and attempt < max_retries: attempt += 1; time.sleep(backoff); backoff *= 2; continue raise except error.URLError: if attempt < max_retries: attempt += 1; time.sleep(backoff); backoff *= 2; continue raise def start_report_run(api_key: str, report_id: str, timeout: int) -> None: st, _, body = _http("POST", f"{API_BASE}/report/{report_id}/run", api_key, b"{}", timeout) if st not in (200, 201): raise RuntimeError(f"Start run failed: {st} {body[:200]!r}") def list_report_history(api_key: str, *, status_filter: str | None = None, report_type: str | None = None, limit: int = 100, sort_by: str = "report_start_date", sort_direction: str = "DESC", timeout: int = 30, offset: int = 0) -> Dict[str, Any]: qs = {"limit": str(limit), "offset": str(offset), "sort_by": sort_by, "sort_direction": sort_direction} if status_filter: qs["status"] = status_filter if report_type: qs["report_type"] = report_type st, _, body = _http("GET", f"{API_BASE}/report/history?{parse.urlencode(qs)}", api_key, timeout=timeout) if st != 200: raise RuntimeError(f"History failed: {st} {body[:200]!r}") return json.loads(body.decode("utf-8")) def find_ready_run(api_key: str, report_id: str, started_not_before: dt.datetime, timeout: int, max_wait_seconds: int, poll_interval: int) -> str: deadline = time.time() + max_wait_seconds while time.time() < deadline: hist = list_report_history(api_key, status_filter="READY", report_type="audit-logs", limit=200, timeout=timeout).get("report_history", []) for it in hist: if it.get("report_identifier") != report_id or not it.get("report_run_identifier"): continue try: rsd = dt.datetime.strptime(it.get("report_start_date",""), "%Y-%m-%d %H:%M:%S").replace(tzinfo=dt.timezone.utc) except Exception: rsd = started_not_before if rsd + dt.timedelta(seconds=60) >= started_not_before: return it["report_run_identifier"] time.sleep(poll_interval) raise TimeoutError("READY run not found in time") def get_json_rows(api_key: str, report_id: str, run_id: str, timeout: int) -> List[Dict[str, Any]]: st, h, body = _http("GET", f"{API_BASE}/report/{report_id}/{run_id}/json", api_key, timeout=timeout) if st != 200: raise RuntimeError(f"Get JSON failed: {st} {body[:200]!r}") if "application/zip" in h.get("content-type","").lower() or body[:2] == b"PK": with zipfile.ZipFile(io.BytesIO(body)) as zf: name = next((n for n in zf.namelist() if n.lower().endswith(".json")), None) if not name: raise RuntimeError("ZIP has no JSON") rows = json.loads(zf.read(name).decode("utf-8")) else: rows = json.loads(body.decode("utf-8")) if not isinstance(rows, list): raise RuntimeError("Unexpected JSON format") return rows def load_state(bucket: str, key: str) -> Dict[str, Any]: try: return json.loads(s3.get_object(Bucket=bucket, Key=key)["Body"].read().decode("utf-8")) except ClientError as e: if e.response["Error"]["Code"] in ("NoSuchKey","404"): return {} raise def save_state(bucket: str, key: str, state: Dict[str, Any]) -> None: s3.put_object(Bucket=bucket, Key=key, Body=json.dumps(state).encode("utf-8"), ContentType="application/json") def write_ndjson_gz(bucket: str, prefix: str, rows: Iterable[Dict[str, Any]], run_id: str) -> str: ts = _now().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")) s3.put_object(Bucket=bucket, Key=key, Body=buf.getvalue(), ContentType="application/x-ndjson", ContentEncoding="gzip") return key def lambda_handler(event: Dict[str, Any], context: Any) -> Dict[str, Any]: api_key = os.environ["DIGICERT_API_KEY"] report_id = os.environ["DIGICERT_REPORT_ID"] bucket = os.environ["S3_BUCKET"] prefix = os.environ.get("S3_PREFIX", "digicert/logs").rstrip("/") state_key = os.environ.get("STATE_KEY", f"{prefix}/state.json") 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")) state = load_state(bucket, state_key) if state_key else {} last_run = state.get("last_run_id") started = _now() start_report_run(api_key, report_id, timeout) run_id = find_ready_run(api_key, report_id, started, timeout, max_wait, poll_int) if last_run and last_run == run_id: return {"status":"skip", "report_run_identifier": run_id} rows = get_json_rows(api_key, report_id, run_id, timeout) key = write_ndjson_gz(bucket, prefix, rows, run_id) if state_key: save_state(bucket, state_key, {"last_run_id": run_id, "last_success_at": _now().isoformat(), "last_s3_key": key, "rows_count": len(rows)}) return {"status":"ok", "report_identifier": report_id, "report_run_identifier": run_id, "rows": len(rows), "s3_key": key}依序前往「Configuration」>「Environment variables」>「Edit」>「Add new environment variable」。
輸入下列環境變數，並將 換成您的值：
鍵 範例 S3_BUCKETdigicert-logsS3_PREFIXdigicert/logs/STATE_KEYdigicert/logs/state.jsonDIGICERT_API_KEYxxxxxxxxxxxxxxxxxxxxxxxxDIGICERT_REPORT_ID88de5e19-ec57-4d70-865d-df953b062574REQUEST_TIMEOUT30POLL_INTERVAL10MAX_WAIT_SECONDS300建立函式後，請留在函式頁面 (或依序開啟「Lambda」>「Functions」>「your-function」)。
選取「設定」分頁標籤。
在「一般設定」面板中，按一下「編輯」。
將「逾時」變更為「15 分鐘 (900 秒)」，然後按一下「儲存」。
建立 EventBridge 排程
- 依序前往「Amazon EventBridge」>「Scheduler」>「Create schedule」。
- 提供下列設定詳細資料：
- 週期性時間表：費率 (
1 hour)。 - 目標：您的 Lambda 函式。
- 名稱：
digicert-audit-1h。
- 週期性時間表：費率 (
- 按一下「建立時間表」。
選用：為 Google SecOps 建立唯讀 IAM 使用者和金鑰
- 在 AWS 控制台中，依序前往「IAM」>「Users」，然後按一下「Add users」。
- 提供下列設定詳細資料：
- 使用者：輸入不重複的名稱 (例如
secops-reader) - 存取權類型：選取「存取金鑰 - 程式輔助存取」
- 按一下「建立使用者」。
- 使用者：輸入不重複的名稱 (例如
- 附加最低讀取權限政策 (自訂)：依序選取「使用者」
secops-reader「權限」「新增權限」「直接附加政策」「建立政策」 在 JSON 編輯器中輸入下列政策：
{ "Version": "2012-10-17", "Statement": [ { "Effect": "Allow", "Action": ["s3:GetObject"], "Resource": "arn:aws:s3:::<your-bucket>/*" }, { "Effect": "Allow", "Action": ["s3:ListBucket"], "Resource": "arn:aws:s3:::<your-bucket>" } ] }Name =
secops-reader-policy。依序點選「建立政策」> 搜尋/選取 >「下一步」>「新增權限」。
為
secops-reader建立存取金鑰：依序點選「安全憑證」>「存取金鑰」>「建立存取金鑰」。下載 CSV (這些值會輸入至動態饋給)。
在 Google SecOps 中設定資訊提供，擷取 DigiCert 記錄
- 依序前往「SIEM 設定」>「動態饋給」。
- 按一下「新增動態消息」。
- 在「動態饋給名稱」欄位中輸入動態饋給名稱 (例如
DigiCert Audit Logs)。 - 選取「Amazon S3 V2」做為「來源類型」。
- 選取「Digicert」做為「記錄類型」。
- 點選「下一步」。
- 指定下列輸入參數的值：
- S3 URI：
s3://digicert-logs/digicert/logs/ - 來源刪除選項：根據偏好選取刪除選項。
- 檔案存在時間上限：預設為 180 天。
- 存取金鑰 ID：具有 S3 值區存取權的使用者存取金鑰。
- 存取密鑰：具有 S3 bucket 存取權的使用者私密金鑰。
- 資產命名空間：資產命名空間。
- 擷取標籤：要套用至這個動態饋給事件的標籤。
- S3 URI：
- 點選「下一步」。
- 在「Finalize」畫面上檢查新的動態饋給設定，然後按一下「Submit」。
還有其他問題嗎？向社群成員和 Google SecOps 專業人員尋求答案。