收集 Fortra Digital Guardian（以前称为 Digital Guardian EDR）日志

解析器版本： 3.0

本文档介绍了如何使用 Cloud Run 函数通过 Google Cloud Storage V2 将 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 存储桶，Google SecOps 通过 GCS V2 Feed 提取这些结果。

准备工作

请确保您满足以下前提条件：

• Google SecOps 实例。
• 已启用以下 API 的 Google Cloud 云项目：
  • Cloud Storage
  • Cloud Run functions
  • 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 存储桶

1. 前往 Google Cloud 控制台。
2. 选择您的项目或创建一个新项目。
3. 在导航菜单中，依次前往 Cloud Storage > 存储分区。
4. 点击创建存储桶 。

5. 提供以下配置详细信息：

   设置	值
   指定存储桶的名称	输入一个全局唯一的名称（例如 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. 依次前往系统 > 配置 > Cloud 服务。
3. 在 API 访问权限 部分中，记录：
   • 访问网关基础网址：例如 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 服务帐号并配置 Feed

1. 依次前往 SIEM 设置 > Feed。
2. 点击 Add New Feed （添加新 Feed）。
3. 选择 Google Cloud Storage V2 作为来源类型 。
4. 选择 Digital Guardian EDR 作为日志类型 。
5. 点击 Get Service Account （获取服务账号），然后复制电子邮件地址。
6. 指定存储桶网址：gs://digitalguardian-edr-logs/digitalguardian_edr/（确保包含尾部斜杠）。

UDM 映射表

更新日志

查看此解析器的更新日志

需要更多帮助？获得社区成员和 Google SecOps 专业人士的解答。