收集 Fortra Digital Guardian (舊稱 Digital Guardian EDR) 記錄

剖析器版本：3.0

本文說明如何使用 Google Cloud Storage V2 和 Cloud Run 函式，將 Fortra Digital Guardian (舊稱 Digital Guardian EDR) 記錄檔擷取至 Google Security Operations。

Fortra Digital Guardian 是一套全面的資料遺失防護和端點偵測與應變平台，可掌握端點、網路和雲端應用程式的系統、使用者和資料事件。Analytics & Reporting Cloud (ARC) 服務提供進階分析、工作流程和報表功能，可全面保護資料。

Cloud Run 函式會使用 OAuth 2.0 向 ARC Export API 進行驗證、擷取匯出資料、確認書籤以繼續處理下一個區塊、將結果以 NDJSON 格式寫入 GCS bucket，而 Google SecOps 會透過 GCS V2 資訊提供擷取這些資料。

事前準備

請確認您已完成下列事前準備事項：

• Google SecOps 執行個體。
• 已啟用下列 API 的 Google Cloud 專案：
  • Cloud Storage
  • Cloud Run 函式
  • Cloud Scheduler
  • Pub/Sub
  • Cloud Build
• 建立及管理 GCS 值區、Cloud Run 函式、Pub/Sub 主題和 Cloud Scheduler 工作的權限。
• 具備 Digital Guardian Management Console (DGMC) 的特殊權限。
• 存取 Digital Guardian Analytics & Reporting Cloud (ARC) 租戶設定。
• 在 DGMC 中設定 Cloud 服務的管理員權限。
• 在 DGMC 中建立的匯出設定檔，且包含有效的 GUID。

建立 Google Cloud Storage bucket

1. 前往 Google Cloud 控制台。
2. 選取專案或建立新專案。
3. 在導覽選單中，依序前往「Cloud Storage」>「bucket」。
4. 按一下「建立值區」。

5. 請提供下列設定詳細資料：

   設定	值
   為 bucket 命名	輸入全域不重複的名稱 (例如 digitalguardian-edr-logs)
   位置類型	根據需求選擇 (區域、雙區域、多區域)
   位置	選取最接近執行個體的位置 (例如 us-central1)
   儲存空間級別	標準 (建議用於經常存取的記錄)
   存取控管	統一 (建議)
   保護工具	選用：啟用物件版本管理或保留政策

6. 點選「建立」。

收集 Digital Guardian API 憑證

取得 API 憑證

從 ARC 租戶設定取得用戶端 ID 和用戶端密鑰

1. 登入 Digital Guardian ARC 入口網站。
2. 前往「ARC 租戶設定」。
3. 複製並儲存下列值：
   • 租戶 ID：這是您的用戶端 ID。
   • 驗證權杖：這是您的用戶端密鑰。

從 DGMC 取得閘道網址

1. 登入 Digital Guardian Management Console (DGMC)。
2. 依序前往「系統」>「設定」>「雲端服務」。
3. 在「API 存取權」部分，記錄：
   • Access Gateway 基準網址：例如 https://accessgw-usw.msp.digitalguardian.com
   • 授權伺服器網址：例如 https://authsrv.msp.digitalguardian.com/as/token.oauth2

建立及設定匯出設定檔

1. 在 DGMC 中，依序前往「管理」>「報告」>「匯出設定檔」。
2. 按一下「建立匯出設定檔」。
3. 將「匯出格式」設為「JSON 平面資料表」。
4. 按一下「儲存」，然後從設定檔詳細資料複製 GUID (格式：xxxxxxxx-xxxx-xxxx-xxxx-xxxxxxxxxxxx)。

建立 Cloud Run 函式

• 準備函式來源檔案

  • requirements.txt

    functions-framework==3.*
    google-cloud-storage==2.*
    urllib3==2.*
    

  • main.py

    """Cloud Run function to ingest Digital Guardian EDR logs into GCS."""
    
    import json
    import os
    import time
    import urllib.parse
    from datetime import datetime, timezone
    
    import functions_framework
    import urllib3
    from google.cloud import storage
    
    GCS_BUCKET = os.environ["GCS_BUCKET"]
    GCS_PREFIX = os.environ.get("GCS_PREFIX", "digitalguardian_edr")
    STATE_KEY = os.environ.get("STATE_KEY", "digitalguardian_edr_state.json")
    AUTH_SERVER_URL = os.environ["AUTH_SERVER_URL"]
    ARC_SERVER_URL = os.environ["ARC_SERVER_URL"]
    CLIENT_ID = os.environ["CLIENT_ID"]
    CLIENT_SECRET = os.environ["CLIENT_SECRET"]
    EXPORT_PROFILE_GUID = os.environ["EXPORT_PROFILE_GUID"]
    MAX_RECORDS = int(os.environ.get("MAX_RECORDS", "10000"))
    
    http = urllib3.PoolManager()
    gcs = storage.Client()
    
    def _get_access_token() -> str:
            body = urllib.parse.urlencode({
                    "grant_type": "client_credentials",
                    "client_id": CLIENT_ID,
                    "client_secret": CLIENT_SECRET,
                    "scope": "client",
            })
            resp = http.request(
                    "POST",
                    AUTH_SERVER_URL,
                    body=body,
                    headers={"Content-Type": "application/x-www-form-urlencoded"},
            )
            if resp.status != 200:
                    raise RuntimeError(f"OAuth token request failed: {resp.status}")
            return json.loads(resp.data.decode("utf-8"))["access_token"]
    
    def _arc_get(token: str, path: str, retries: int = 5) -> dict:
            url = f"{ARC_SERVER_URL}{path}"
            headers = {"Authorization": f"Bearer {token}", "Accept": "application/json"}
            backoff = 2
            for attempt in range(retries):
                    resp = http.request("GET", url, headers=headers)
                    if resp.status == 200:
                            return json.loads(resp.data.decode("utf-8"))
                    if resp.status == 429:
                            time.sleep(backoff * (2 ** attempt))
                            continue
                    raise RuntimeError(f"ARC API error: {resp.status}")
            raise RuntimeError("ARC API rate limit exceeded.")
    
    def _arc_acknowledge(token: str) -> None:
            url = f"{ARC_SERVER_URL}/rest/1.0/export/{EXPORT_PROFILE_GUID}/acknowledge"
            headers = {"Authorization": f"Bearer {token}", "Accept": "application/json"}
            resp = http.request("POST", url, headers=headers)
            if resp.status not in (200, 204):
                    raise RuntimeError(f"ARC acknowledge failed: {resp.status}")
    
    def _load_state() -> dict:
            bucket = gcs.bucket(GCS_BUCKET)
            blob = bucket.blob(f"{GCS_PREFIX}/{STATE_KEY}")
            return json.loads(blob.download_as_text()) if blob.exists() else {}
    
    def _save_state(state: dict) -> None:
            bucket = gcs.bucket(GCS_BUCKET)
            blob = bucket.blob(f"{GCS_PREFIX}/{STATE_KEY}")
            blob.upload_from_string(json.dumps(state), content_type="application/json")
    
    def _fetch_export(token: str) -> list:
            path = f"/rest/1.0/export/{EXPORT_PROFILE_GUID}"
            data = _arc_get(token, path)
            records = data if isinstance(data, list) else data.get("data", [])
            return records[:MAX_RECORDS]
    
    def _write_ndjson(records: list, run_ts: str) -> str:
            bucket = gcs.bucket(GCS_BUCKET)
            blob_path = f"{GCS_PREFIX}/year={run_ts[:4]}/month={run_ts[5:7]}/day={run_ts[8:10]}/{run_ts}_export.ndjson"
            blob = bucket.blob(blob_path)
            ndjson = "\n".join(json.dumps(r, separators=(",", ":")) for r in records)
            blob.upload_from_string(ndjson, content_type="application/x-ndjson")
            return blob_path
    
    @functions_framework.cloud_event
    def main(cloud_event):
            state = _load_state()
            now = datetime.now(timezone.utc)
            token = _get_access_token()
            records = _fetch_export(token)
            if not records:
                    return "OK"
            run_ts = now.strftime("%Y-%m-%dT%H%M%SZ")
            blob_path = _write_ndjson(records, run_ts)
            _arc_acknowledge(token)
            state["last_run"] = now.isoformat()
            _save_state(state)
            return "OK"
    

擷取 Google SecOps 服務帳戶並設定動態饋給

1. 依序前往「SIEM 設定」>「動態消息」。
2. 按一下「新增動態消息」。
3. 選取「Google Cloud Storage V2」做為「來源類型」。
4. 選取「Digital Guardian EDR」做為「記錄類型」。
5. 按一下「取得服務帳戶」，然後複製電子郵件地址。
6. 指定儲存空間 bucket 網址：gs://digitalguardian-edr-logs/digitalguardian_edr/ (請務必加上結尾斜線)。

UDM 對應表

變更記錄

查看這個剖析器的變更記錄

還有其他問題嗎？向社群成員和 Google SecOps 專業人員尋求答案。