收集 Digital Shadows SearchLight 日志

支持的平台：

Google SecOps SIEM

注意： 此功能受《Google Security Operations 服务专用条款》的正式发布前产品条款约束。我们对正式发布前功能的支持可能有限，并且对正式发布前功能的更改与其他正式发布前版本可能不兼容。如需了解详情，请参阅 Google SecOps 技术支持服务准则和 Google SecOps 服务专用条款。

本文档介绍了如何使用 Amazon S3 将 Digital Shadows SearchLight 日志注入到 Google Security Operations。解析器从 JSON 日志中提取安全事件数据。它会初始化统一数据模型 (UDM) 字段、解析 JSON 载荷、将相关字段映射到 UDM 架构、使用 grok 模式提取电子邮件和主机名等实体，并在 UDM 事件中构建 security_result 和 metadata 对象。

准备工作

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

• Google SecOps 实例。
• 对 Digital Shadows SearchLight 租户的特权访问权限。
• 对 AWS（S3、Identity and Access Management (IAM)、Lambda、EventBridge）的特权访问权限。

收集 Digital Shadows SearchLight 前提条件（ID、API 密钥、组织 ID、令牌）

1. 登录 Digital Shadows SearchLight 门户。
2. 依次前往设置 > API 凭据。
3. 创建新的 API 客户端或密钥对。
4. 复制以下详细信息并将其保存在安全的位置：
   • API 密钥
   • API Secret
   • 账号 ID
   • API 基准网址：https://api.searchlight.app/v1 或 https://portal-digitalshadows.com/api/v1

为 Google SecOps 配置 AWS S3 存储桶和 IAM

1. 按照以下用户指南创建 Amazon S3 存储桶：创建存储桶
2. 保存存储桶名称和区域以供日后参考（例如 digital-shadows-logs）。
3. 按照以下用户指南创建用户：创建 IAM 用户。
4. 选择创建的用户。
5. 选择安全凭据标签页。
6. 在访问密钥部分中，点击创建访问密钥。
7. 选择第三方服务作为使用情形。
8. 点击下一步。
9. 可选：添加说明标记。
10. 点击创建访问密钥。
11. 点击下载 .CSV 文件，保存访问密钥和秘密访问密钥，以供日后参考。
12. 点击完成。
13. 选择权限标签页。
14. 在权限政策部分中，点击添加权限。
15. 选择添加权限。
16. 选择直接附加政策。
17. 搜索 AmazonS3FullAccess 政策。
18. 选择相应政策。
19. 点击下一步。
20. 点击添加权限。

为 S3 上传配置 IAM 政策和角色

1. 在 AWS 控制台中，依次前往 IAM > 政策。
2. 依次点击创建政策 > JSON 标签页。
3. 复制并粘贴以下政策。

4. 政策 JSON（如果您输入了其他存储桶名称，请替换 digital-shadows-logs）：

   {
     "Version": "2012-10-17",
     "Statement": [
       {
         "Sid": "AllowPutObjects",
         "Effect": "Allow",
         "Action": "s3:PutObject",
         "Resource": "arn:aws:s3:::digital-shadows-logs/*"
       },
       {
         "Sid": "AllowGetStateObject",
         "Effect": "Allow",
         "Action": "s3:GetObject",
         "Resource": "arn:aws:s3:::digital-shadows-logs/digital-shadows-searchlight/state.json"
       }
     ]
   }
   

5. 依次点击下一步 > 创建政策。

6. 依次前往 IAM > 角色 > 创建角色 > AWS 服务 > Lambda。

7. 附加新创建的政策。

8. 将角色命名为 digital-shadows-lambda-role，然后点击创建角色。

创建 Lambda 函数

1. 在 AWS 控制台中，依次前往 Lambda > 函数 > 创建函数。
2. 点击从头开始创作。

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

   设置	值
   名称	digital-shadows-collector
   运行时	Python 3.13
   架构	x86_64
   执行角色	digital-shadows-lambda-role

4. 创建函数后，打开 Code 标签页，删除桩代码并粘贴以下代码 (digital-shadows-collector.py)。

   import json
   import os
   import base64
   import logging
   import time
   from datetime import datetime, timedelta, timezone
   from urllib.parse import urlencode
   
   import boto3
   import urllib3
   
   logger = logging.getLogger()
   logger.setLevel(logging.INFO)
   
   HTTP = urllib3.PoolManager(retries=False)
   
   def _basic_auth_header(key: str, secret: str) -> str:
       token = base64.b64encode(f"{key}:{secret}".encode("utf-8")).decode("utf-8")
       return f"Basic {token}"
   
   def _load_state(s3, bucket, key, default_days=30) -> str:
       """Return ISO8601 checkpoint (UTC)."""
       try:
           obj = s3.get_object(Bucket=bucket, Key=key)
           state = json.loads(obj["Body"].read().decode("utf-8"))
           ts = state.get("last_timestamp")
           if ts:
               return ts
       except s3.exceptions.NoSuchKey:
           pass
       except Exception as e:
           logger.warning(f"State read error: {e}")
       return (datetime.now(timezone.utc) - timedelta(days=default_days)).isoformat()
   
   def _save_state(s3, bucket, key, ts: str) -> None:
       s3.put_object(
           Bucket=bucket,
           Key=key,
           Body=json.dumps({"last_timestamp": ts}).encode("utf-8"),
           ContentType="application/json",
       )
   
   def _get_json(url: str, headers: dict, params: dict, backoff_s=2, max_retries=3) -> dict:
       qs = f"?{urlencode(params)}" if params else ""
       for attempt in range(max_retries):
           r = HTTP.request("GET", f"{url}{qs}", headers=headers)
           if r.status == 200:
               return json.loads(r.data.decode("utf-8"))
           if r.status in (429, 500, 502, 503, 504):
               wait = backoff_s * (2 ** attempt)
               logger.warning(f"HTTP {r.status} from DS API, retrying in {wait}s")
               time.sleep(wait)
               continue
           raise RuntimeError(f"DS API error {r.status}: {r.data[:200]}")
       raise RuntimeError("Exceeded retry budget for DS API")
   
   def _collect(api_base, headers, path, since_ts, account_id, page_size, max_pages, time_param):
       items = []
       for page in range(max_pages):
           params = {
               "limit": page_size,
               "offset": page * page_size,
               time_param: since_ts,
           }
           if account_id:
               params["account-id"] = account_id
   
           data = _get_json(f"{api_base}/{path}", headers, params)
           batch = data.get("items") or data.get("data") or []
           if not batch:
               break
           items.extend(batch)
           if len(batch) < page_size:
               break
       return items
   
   def lambda_handler(event, context):
       # Required
       s3_bucket  = os.environ["S3_BUCKET"]
       api_key    = os.environ["DS_API_KEY"]
       api_secret = os.environ["DS_API_SECRET"]
   
       # Optional / defaults
       s3_prefix  = os.environ.get("S3_PREFIX", "digital-shadows-searchlight/")
       state_key  = os.environ.get("STATE_KEY", "digital-shadows-searchlight/state.json")
       api_base   = os.environ.get("API_BASE", "https://api.searchlight.app/v1")
       account_id = os.environ.get("DS_ACCOUNT_ID", "")
       page_size  = int(os.environ.get("PAGE_SIZE", "100"))
       max_pages  = int(os.environ.get("MAX_PAGES", "10"))
   
       s3 = boto3.client("s3")
       last_ts = _load_state(s3, s3_bucket, state_key)
       logger.info(f"Checkpoint: {last_ts}")
   
       headers = {
           "Authorization": _basic_auth_header(api_key, api_secret),
           "Accept": "application/json",
           "User-Agent": "Chronicle-DigitalShadows-S3/1.0",
       }
   
       records = []
   
       # Incidents (time filter often 'published-after' or 'updated-since' depending on tenancy)
       incidents = _collect(api_base, headers, "incidents", last_ts, account_id, page_size, max_pages, time_param="published-after")
       for incident in incidents:
           incident['_source_type'] = 'incident'
       records.extend(incidents)
   
       # Intelligence incidents (alerts)
       intel_incidents = _collect(api_base, headers, "intel-incidents", last_ts, account_id, page_size, max_pages, time_param="published-after")
       for intel in intel_incidents:
           intel['_source_type'] = 'intelligence_incident'
       records.extend(intel_incidents)
   
       # Indicators (IOCs)
       indicators = _collect(api_base, headers, "indicators", last_ts, account_id, page_size, max_pages, time_param="lastUpdated-after")
       for indicator in indicators:
           indicator['_source_type'] = 'ioc'
       records.extend(indicators)
   
       if records:
           # Choose newest timestamp seen in this batch
           newest = max(
               (r.get("updated") or r.get("raised") or r.get("lastUpdated") or last_ts)
               for r in records
           )
           key = f"{s3_prefix}digital_shadows_{datetime.now(timezone.utc).strftime('%Y%m%d_%H%M%S')}.json"
           body = "\n".join(json.dumps(r, separators=(",", ":")) for r in records).encode("utf-8")
   
           s3.put_object(
               Bucket=s3_bucket,
               Key=key,
               Body=body,
               ContentType="application/x-ndjson",
           )
           _save_state(s3, s3_bucket, state_key, newest)
           msg = f"Wrote {len(records)} records to s3://{s3_bucket}/{key}"
       else:
           msg = "No new records"
   
       logger.info(msg)
       return {"statusCode": 200, "body": msg}
   

5. 依次前往配置 > 环境变量。

6. 依次点击修改 > 添加新的环境变量。

7. 输入下表中提供的环境变量，并将示例值替换为您的值。

   环境变量

   键	示例值
   S3_BUCKET	digital-shadows-logs
   S3_PREFIX	digital-shadows-searchlight/
   STATE_KEY	digital-shadows-searchlight/state.json
   DS_API_KEY	<your-6-character-api-key>
   DS_API_SECRET	<your-32-character-api-secret>
   API_BASE	https://api.searchlight.app/v1（或https://portal-digitalshadows.com/api/v1）
   DS_ACCOUNT_ID	<your-account-id>（大多数租户都需要）
   PAGE_SIZE	100
   MAX_PAGES	10

8. 创建函数后，请停留在其页面上（或依次打开 Lambda > 函数 > 您的函数）。

9. 选择配置标签页。

10. 在常规配置面板中，点击修改。

11. 将超时更改为 5 分钟（300 秒），然后点击保存。

    注意： 超时时间是一个硬性上限，这意味着如果函数在计时器达到 5 分钟时仍在运行，AWS Lambda 将终止该函数。任何较短的执行时间都只会正常结束；您只需为实际运行时间付费。

创建 EventBridge 计划

1. 依次前往 Amazon EventBridge > 调度器 > 创建调度。
2. 提供以下配置详细信息：
   • 周期性安排：费率 (1 hour)。
   • 目标：您的 Lambda 函数 digital-shadows-collector。
   • 名称：digital-shadows-collector-1h。
3. 点击创建时间表。

（可选）为 Google SecOps 创建只读 IAM 用户和密钥

1. 依次前往 AWS 控制台 > IAM > 用户。
2. 点击 Add users（添加用户）。
3. 提供以下配置详细信息：
   • 用户：输入 secops-reader。
   • 访问类型：选择访问密钥 - 以程序化方式访问。
4. 点击创建用户。
5. 附加最低限度的读取政策（自定义）：依次选择用户 > secops-reader > 权限 > 添加权限 > 直接附加政策 > 创建政策。

6. JSON：

   {
     "Version": "2012-10-17",
     "Statement": [
       {
         "Effect": "Allow",
         "Action": ["s3:GetObject"],
         "Resource": "arn:aws:s3:::digital-shadows-logs/*"
       },
       {
         "Effect": "Allow",
         "Action": ["s3:ListBucket"],
         "Resource": "arn:aws:s3:::digital-shadows-logs"
       }
     ]
   }
   

7. 名称 = secops-reader-policy。

8. 依次点击创建政策 > 搜索/选择 > 下一步 > 添加权限。

9. 为 secops-reader 创建访问密钥：安全凭据 > 访问密钥。

10. 点击创建访问密钥。

11. 下载 .CSV。（您需要将这些值粘贴到 Feed 中）。

在 Google SecOps 中配置 Feed 以注入 Digital Shadows SearchLight 日志

1. 依次前往 SIEM 设置> Feed。
2. 点击 + 添加新 Feed。
3. 在Feed 名称字段中，输入 Feed 的名称（例如 Digital Shadows SearchLight logs）。
4. 选择 Amazon S3 V2 作为来源类型。
5. 选择 Digital Shadows SearchLight 作为日志类型。
6. 点击下一步。
7. 为以下输入参数指定值：
   • S3 URI：s3://digital-shadows-logs/digital-shadows-searchlight/
   • 来源删除选项：根据您的偏好选择删除选项。
   • 文件存在时间上限：包含在过去指定天数内修改的文件。默认值为 180 天。
   • 访问密钥 ID：有权访问 S3 存储桶的用户访问密钥。
   • 私有访问密钥：具有 S3 存储桶访问权限的用户私有密钥。
   • 资产命名空间：资产命名空间。
   • 注入标签：应用于此 Feed 中事件的标签。
8. 点击下一步。
9. 在最终确定界面中查看新的 Feed 配置，然后点击提交。

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