收集 Code42 Incydr 核心数据集

支持的平台：

Google SecOps SIEM

本文档介绍了如何使用 Google Cloud Storage 将 Code42 Incydr 核心数据集（用户、审核、支持请求和可选的文件事件）提取到 Google Security Operations。

Code42 Incydr 是一款内部风险管理解决方案，可通过实时监控端点、云服务和电子邮件中的文件活动，检测、调查和应对跨设备的数据渗漏。

准备工作

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

Google SecOps 实例
已启用 Cloud Storage API 的 GCP 项目
创建和管理 GCS 存储分区的权限
管理 GCS 存储分区的 IAM 政策的权限
创建 Cloud Run 服务、Pub/Sub 主题和 Cloud Scheduler 作业的权限
以“内部风险管理员”角色对 Code42 Incydr API 或管理控制台进行特权访问

创建 Google Cloud Storage 存储分区

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

提供以下配置详细信息：

设置	值
为存储分区命名	输入一个全局唯一的名称（例如 `code42-incydr-logs`）
位置类型	根据您的需求进行选择（区域级、双区域级、多区域级）
位置	选择相应位置（例如 `us-central1`）
存储类别	标准（建议用于经常访问的日志）
访问权限控制	统一（推荐）
保护工具	可选：启用对象版本控制或保留政策

点击创建。

收集 Code42 Incydr API 凭据

创建 API 客户端

登录 Code42 Incydr Web 控制台。
依次前往管理 > 集成 > API 客户端。
点击创建新的 API 客户端。
在创建新的 API 客户端对话框中，输入客户端的名称（例如 Google Security Operations Integration）。
请复制以下详细信息并将其保存在安全的位置：
- 客户端 ID：API 客户端标识符。
- 密钥：API 客户端密钥。
点击创建。

确定 API 基准网址

API 基本网址取决于您的 Code42 Incydr 控制台网址。在 Incydr 开发者门户或租户的环境文档中验证您的 API 网关网址。

常见默认值：

控制台网址	API 基本网址
`https://console.us.code42.com`	`https://api.us.code42.com`
`https://console.us2.code42.com`	`https://api.us2.code42.com`
`https://console.ie.code42.com`	`https://api.ie.code42.com`
`https://console.gov.code42.com`	`https://api.gov.code42.com`

验证 API 客户端权限

API 客户端必须具有适当的权限才能访问所需的端点：

在 Code42 Incydr 控制台中，依次前往管理 > 集成 > API 客户端。
点击您创建的 API 客户端名称。
验证 API 客户端是否具有以下范围的访问权限：
- 用户：对用户数据的读取权限
- 审核日志：对审核日志的读取权限
- 支持请求：对支持请求数据的读取权限
- 文件事件（可选）：对文件事件数据的读取权限
注意：如果您没有看到这些选项或无法访问某些端点，请与您的 Code42 管理员联系，以获取必要的权限。确保您的 API 客户端具有所需的 Incydr API 访问权限级别（完全访问权限或基本访问权限），如 Code42 Incydr API 访问权限级别文档中所述。

测试 API 访问权限

在继续进行集成之前，请先测试您的凭据：

# Replace with your actual credentials
CLIENT_ID="your-client-id"
CLIENT_SECRET="your-client-secret"
API_BASE="https://api.us.code42.com"

# Get OAuth token
TOKEN=$(curl -s -X POST "${API_BASE}/v1/oauth/token" \
  -u "${CLIENT_ID}:${CLIENT_SECRET}" \
  -H "Content-Type: application/x-www-form-urlencoded" \
  -d "grant_type=client_credentials" | jq -r '.access_token')

# Test API access
curl -v -H "Authorization: Bearer ${TOKEN}" "${API_BASE}/v1/users?pageSize=1"

为 Cloud Run 函数创建服务账号

Cloud Run 函数需要一个服务账号，该账号具有向 GCS 存储分区写入内容以及被 Pub/Sub 调用的权限。

创建服务账号

在 GCP 控制台中，依次前往 IAM 和管理 > 服务账号。
点击创建服务账号。
提供以下配置详细信息：
- 服务账号名称：输入 code42-incydr-collector-sa。
- 服务账号说明：输入 Service account for Cloud Run function to collect Code42 Incydr logs。
点击创建并继续。
在向此服务账号授予对项目的访问权限部分中，添加以下角色：
1. 点击选择角色。
2. 搜索并选择 Storage Object Admin。
3. 点击 + 添加其他角色。
4. 搜索并选择 Cloud Run Invoker。
5. 点击 + 添加其他角色。
6. 搜索并选择 Cloud Functions Invoker。
点击继续。
点击完成。

必须拥有这些角色，才能：

Storage Object Admin：将日志写入 GCS 存储分区并管理状态文件
Cloud Run Invoker：允许 Pub/Sub 调用函数
Cloud Functions Invoker：允许调用函数

授予对 GCS 存储分区的 IAM 权限

向服务账号授予对 GCS 存储分区的写入权限：

前往 Cloud Storage > 存储分区。
点击您的存储分区名称（例如 code42-incydr-logs）。
前往权限标签页。
点击授予访问权限。
提供以下配置详细信息：
- 添加主账号：输入服务账号电子邮件地址（例如 code42-incydr-collector-sa@your-project.iam.gserviceaccount.com）。
- 分配角色：选择 Storage Object Admin。
点击保存。

创建发布/订阅主题

创建一个 Pub/Sub 主题，供 Cloud Scheduler 发布消息，并供 Cloud Run 函数订阅。

在 GCP 控制台中，前往 Pub/Sub > 主题。
点击创建主题。
提供以下配置详细信息：
- 主题 ID：输入 code42-incydr-trigger。
- 将其他设置保留为默认值。
点击创建。

创建 Cloud Run 函数以收集日志

Cloud Run 函数由来自 Cloud Scheduler 的 Pub/Sub 消息触发，用于从 Code42 Incydr API 中提取日志并将其写入 GCS。

在 GCP 控制台中，前往 Cloud Run。
点击创建服务。
选择函数（使用内嵌编辑器创建函数）。
在配置部分中，提供以下配置详细信息：

设置值

Service 名称 code42-incydr-collector

区域选择与您的 GCS 存储分区匹配的区域（例如 us-central1）

运行时 选择 Python 3.12 或更高版本
在触发器（可选）部分中：
1. 点击 + 添加触发器。
2. 选择 Cloud Pub/Sub。
3. 在选择 Cloud Pub/Sub 主题部分，选择主题 code42-incydr-trigger。
4. 点击保存。
在身份验证部分中：
1. 选择需要进行身份验证。
2. 检查 Identity and Access Management (IAM)。
注意：调用函数时，Pub/Sub 会自动处理身份验证。
向下滚动并展开容器、网络、安全性。
前往安全标签页：
- 服务账号：选择服务账号 code42-incydr-collector-sa。

设置	值
Service 名称	`code42-incydr-collector`
区域	选择与您的 GCS 存储分区匹配的区域（例如 `us-central1`）
运行时	选择 Python 3.12 或更高版本

前往容器标签页：

点击变量和密钥。
为每个环境变量点击 + 添加变量：

变量名称	示例值	说明
`INCYDR_BASE_URL`	`https://api.us.code42.com`	您租户的 API 基础网址
`INCYDR_CLIENT_ID`	`your-client-id`	API 客户端 ID
`INCYDR_CLIENT_SECRET`	`your-client-secret`	API 客户端密钥
`GCS_BUCKET`	`code42-incydr-logs`	GCS 存储分区名称
`GCS_PREFIX`	`code42/`	日志文件的前缀
`PAGE_SIZE`	`500`	每页记录数
`LOOKBACK_MINUTES`	`60`	初始回溯期
`STREAMS`	`users,audit,cases`	以英文逗号分隔的数据流
`FE_QUERY_JSON`	``	可选：文件事件查询 JSON
`FE_PAGE_SIZE`	`1000`	可选：文件事件页面大小

在变量和密钥标签页中，向下滚动到请求：
- 请求超时：输入 600 秒（10 分钟）。
前往容器中的设置标签页：
- 在资源部分中：
  - 内存：选择 1024 MiB 或更高值。
  - CPU：选择 1。
- 点击完成。
滚动到执行环境：
- 选择默认（推荐）。
在修订版本扩缩部分中：
- 实例数下限：输入 0。
- 实例数上限：输入 100（或根据预期负载进行调整）。
点击创建。
等待服务创建完成（1-2 分钟）。
创建服务后，系统会自动打开内嵌代码编辑器。

添加函数代码

在函数入口点中输入 main

在内嵌代码编辑器中，创建两个文件：

第一个文件：main.py:：

import functions_framework
from google.cloud import storage
import json
import os
import urllib3
from datetime import datetime, timedelta, timezone
import time

# Initialize HTTP client
http = urllib3.PoolManager(
    timeout=urllib3.Timeout(connect=5.0, read=30.0),
    retries=False,
)

# Initialize Storage client
storage_client = storage.Client()

# Environment variables
BASE = os.environ.get("INCYDR_BASE_URL", "").rstrip("/")
CID = os.environ.get("INCYDR_CLIENT_ID", "")
CSECRET = os.environ.get("INCYDR_CLIENT_SECRET", "")
BUCKET = os.environ.get("GCS_BUCKET", "")
PREFIX_BASE = os.environ.get("GCS_PREFIX", "code42/")
PAGE_SIZE = int(os.environ.get("PAGE_SIZE", "500"))
LOOKBACK_MINUTES = int(os.environ.get("LOOKBACK_MINUTES", "60"))
STREAMS = [s.strip() for s in os.environ.get("STREAMS", "users").split(",") if s.strip()]
FE_QUERY_JSON = os.environ.get("FE_QUERY_JSON", "").strip()
FE_PAGE_SIZE = int(os.environ.get("FE_PAGE_SIZE", "1000"))

def now_utc():
    return datetime.now(timezone.utc)

def iso_minus(minutes: int):
    return (now_utc() - timedelta(minutes=minutes)).strftime("%Y-%m-%dT%H:%M:%SZ")

def put_json(bucket, prefix: str, page_label: str, data):
    ts = now_utc().strftime("%Y/%m/%d/%H%M%S")
    key = f"{PREFIX_BASE}{prefix}{ts}-{page_label}.json"
    blob = bucket.blob(key)
    blob.upload_from_string(json.dumps(data), content_type='application/json')
    return key

def get_token():
    """Get OAuth 2.0 access token using client credentials flow."""
    token_url = f"{BASE}/v1/oauth/token"

    # Encode credentials
    import base64
    credentials = f"{CID}:{CSECRET}"
    encoded_credentials = base64.b64encode(credentials.encode('utf-8')).decode('utf-8')

    headers = {
        'Authorization': f'Basic {encoded_credentials}',
        'Content-Type': 'application/x-www-form-urlencoded',
        'Accept': 'application/json'
    }

    body = 'grant_type=client_credentials'

    backoff = 1.0
    max_retries = 3

    for attempt in range(max_retries):
        response = http.request('POST', token_url, body=body, headers=headers)

        if response.status == 429:
            retry_after = int(response.headers.get('Retry-After', str(int(backoff))))
            print(f"Rate limited (429) on token request. Retrying after {retry_after}s...")
            time.sleep(retry_after)
            backoff = min(backoff * 2, 30.0)
            continue

        if response.status != 200:
            raise RuntimeError(f"Failed to get access token: {response.status} - {response.data.decode('utf-8')}")

        data = json.loads(response.data.decode('utf-8'))
        return data['access_token']

    raise RuntimeError(f"Failed to get token after {max_retries} retries due to rate limiting")

def auth_header():
    token = get_token()
    return {
        "Authorization": f"Bearer {token}",
        "Accept": "application/json"
    }

def http_get(path: str, params: dict = None, headers: dict = None):
    url = f"{BASE}{path}"
    if params:
        from urllib.parse import urlencode
        url += "?" + urlencode(params)

    backoff = 1.0
    max_retries = 3

    for attempt in range(max_retries):
        response = http.request('GET', url, headers=headers)

        if response.status == 429:
            retry_after = int(response.headers.get('Retry-After', str(int(backoff))))
            print(f"Rate limited (429). Retrying after {retry_after}s...")
            time.sleep(retry_after)
            backoff = min(backoff * 2, 30.0)
            continue

        return response.data

    raise RuntimeError(f"Failed after {max_retries} retries due to rate limiting")

def http_post_json(path: str, body: dict, headers: dict = None):
    url = f"{BASE}{path}"

    backoff = 1.0
    max_retries = 3

    for attempt in range(max_retries):
        response = http.request(
            'POST',
            url,
            body=json.dumps(body),
            headers={**headers, 'Content-Type': 'application/json'}
        )

        if response.status == 429:
            retry_after = int(response.headers.get('Retry-After', str(int(backoff))))
            print(f"Rate limited (429). Retrying after {retry_after}s...")
            time.sleep(retry_after)
            backoff = min(backoff * 2, 30.0)
            continue

        return response.data

    raise RuntimeError(f"Failed after {max_retries} retries due to rate limiting")

# USERS (/v1/users)
def pull_users(bucket, hdrs):
    next_token = None
    pages = 0
    while True:
        params = {"active": "true", "blocked": "false", "pageSize": PAGE_SIZE}
        if next_token:
            params["pageToken"] = next_token
        raw = http_get("/v1/users", params, hdrs)
        data = json.loads(raw.decode('utf-8'))
        put_json(bucket, "users/", f"users-page-{pages}", data)
        pages += 1
        next_token = data.get("nextPageToken") or data.get("next_page_token")
        if not next_token:
            break
    return pages

# AUDIT LOG (/v1/audit/log)
def pull_audit(bucket, hdrs):
    start_iso = iso_minus(LOOKBACK_MINUTES)
    next_token = None
    pages = 0
    while True:
        params = {"startTime": start_iso, "pageSize": PAGE_SIZE}
        if next_token:
            params["pageToken"] = next_token
        raw = http_get("/v1/audit/log", params, hdrs)
        try:
            data = json.loads(raw.decode('utf-8'))
            put_json(bucket, "audit/", f"audit-page-{pages}", data)
            next_token = data.get("nextPageToken") or data.get("next_page_token")
            pages += 1
            if not next_token:
                break
        except Exception as e:
            print(f"Error parsing audit log response: {e}")
            # Save raw response
            ts = now_utc().strftime("%Y/%m/%d/%H%M%S")
            key = f"{PREFIX_BASE}audit/{ts}-audit-export.bin"
            blob = bucket.blob(key)
            blob.upload_from_string(raw, content_type='application/octet-stream')
            pages += 1
            break
    return pages

# CASES (/v1/cases)
def pull_cases(bucket, hdrs):
    next_token = None
    pages = 0
    while True:
        params = {"pageSize": PAGE_SIZE}
        if next_token:
            params["pageToken"] = next_token
        raw = http_get("/v1/cases", params, hdrs)
        data = json.loads(raw.decode('utf-8'))
        put_json(bucket, "cases/", f"cases-page-{pages}", data)
        pages += 1
        next_token = data.get("nextPageToken") or data.get("next_page_token")
        if not next_token:
            break
    return pages

# FILE EVENTS (/v2/file-events/search)
def pull_file_events(bucket, hdrs):
    if not FE_QUERY_JSON:
        return 0
    try:
        base_query = json.loads(FE_QUERY_JSON)
    except Exception as e:
        print(f"Error: FE_QUERY_JSON is not valid JSON: {e}")
        return 0

    pages = 0
    next_token = None
    while True:
        body = dict(base_query)
        body["pageSize"] = FE_PAGE_SIZE
        if next_token:
            body["pageToken"] = next_token
        raw = http_post_json("/v2/file-events/search", body, hdrs)
        data = json.loads(raw.decode('utf-8'))
        put_json(bucket, "file_events/", f"fileevents-page-{pages}", data)
        pages += 1
        next_token = (
            data.get("nextPageToken") or 
            data.get("next_page_token") or 
            (data.get("file_events") or {}).get("nextPageToken")
        )
        if not next_token:
            break
    return pages

@functions_framework.cloud_event
def main(cloud_event):
    """
    Cloud Run function triggered by Pub/Sub to fetch logs from Code42 Incydr API and write to GCS.

    Args:
        cloud_event: CloudEvent object containing Pub/Sub message
    """

    if not all([BASE, CID, CSECRET, BUCKET]):
        print('Error: Missing required environment variables')
        return

    try:
        bucket = storage_client.bucket(BUCKET)
        hdrs = auth_header()
        report = {}

        if "users" in STREAMS:
            print("Fetching users...")
            report["users_pages"] = pull_users(bucket, hdrs)
        if "audit" in STREAMS:
            print("Fetching audit logs...")
            report["audit_pages"] = pull_audit(bucket, hdrs)
        if "cases" in STREAMS:
            print("Fetching cases...")
            report["cases_pages"] = pull_cases(bucket, hdrs)
        if "file_events" in STREAMS:
            print("Fetching file events...")
            report["file_events_pages"] = pull_file_events(bucket, hdrs)

        print(f'Successfully processed logs: {json.dumps(report)}')

    except Exception as e:
        print(f'Error processing logs: {str(e)}')
        raise

第二个文件：requirements.txt:：

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

点击部署以保存并部署该函数。
等待部署完成（2-3 分钟）。

注意： Pub/Sub 触发器配置会自动创建必要的订阅和权限。

创建 Cloud Scheduler 作业

Cloud Scheduler 会定期向 Pub/Sub 主题发布消息，从而触发 Cloud Run 函数。

在 GCP Console 中，前往 Cloud Scheduler。
点击创建作业。

提供以下配置详细信息：

设置	值
名称	`code42-incydr-hourly`
区域	选择与 Cloud Run 函数相同的区域
频率	`0 * * * *`（每小时一次，在整点时）
时区	选择时区（建议选择世界协调时间 [UTC]）
目标类型	Pub/Sub
主题	选择主题 `code42-incydr-trigger`
消息正文	`{}`（空 JSON 对象）

点击创建。

时间表频率选项

根据日志量和延迟时间要求选择频次：

频率	Cron 表达式	使用场景
每隔 5 分钟	`/5 * * *`	高容量、低延迟
每隔 15 分钟	`/15 * * *`	搜索量中等
每小时	`0 * * * *`	标准（推荐）
每 6 小时	`0 /6 * *`	量小、批处理
每天	`0 0 * * *`	历史数据收集

测试调度器作业

在 Cloud Scheduler 控制台中，找到您的作业 (code42-incydr-hourly)。
点击强制运行以手动触发。
等待几秒钟，然后前往 Cloud Run > 服务 > code42-incydr-collector > 日志。

验证函数是否已成功执行。期望：

Fetching users...
Fetching audit logs...
Fetching cases...
Successfully processed logs: {"users_pages": X, "audit_pages": Y, "cases_pages": Z}

检查 GCS 存储分区 (code42-incydr-logs) 以确认日志是否已写入。

如果您在日志中看到错误，请执行以下操作：

HTTP 401：检查环境变量中的 API 凭据
HTTP 403：在 Code42 Incydr 控制台中验证 API 客户端是否具有必需的权限
HTTP 429：速率限制 - 函数将自动重试并进行退避
未能获取访问令牌：验证 INCYDR_BASE_URL、INCYDR_CLIENT_ID 和 INCYDR_CLIENT_SECRET 是否正确

检索 Google SecOps 服务账号

Google SecOps 使用唯一的服务账号从您的 GCS 存储分区中读取数据。您必须授予此服务账号对您的存储分区的访问权限。

获取服务账号电子邮件地址

依次前往 SIEM 设置 > Feed。
点击添加新 Feed。
点击配置单个 Feed。
在Feed 名称字段中，输入 Feed 的名称（例如 Code42 Incydr Datasets）。
选择 Google Cloud Storage V2 作为来源类型。
选择 Code42 Incydr 作为日志类型。
点击获取服务账号。系统会显示一个唯一的服务账号电子邮件地址，例如：
```
chronicle-12345678@chronicle-gcp-prod.iam.gserviceaccount.com
```
复制此电子邮件地址，以便在下一步中使用。

注意：每个 Google SecOps 实例都有一个唯一的服务账号。请勿使用其他文档或示例中的服务账号。

向 Google SecOps 服务账号授予 IAM 权限

Google SecOps 服务账号需要对您的 GCS 存储分区具有 Storage Object Viewer 角色。

前往 Cloud Storage > 存储分区。
点击您的存储分区名称 (code42-incydr-logs)。
前往权限标签页。
点击授予访问权限。
提供以下配置详细信息：
- 添加主账号：粘贴 Google SecOps 服务账号电子邮件地址。
- 分配角色：选择 Storage Object Viewer。
点击保存。

注意：如果您打算使用“删除已转移的文件”或“删除已转移的文件和空目录”删除选项，请授予存储对象管理员角色，而不是存储对象查看者角色。

在 Google SecOps 中配置 Feed 以提取 Code42 Incydr 日志

依次前往 SIEM 设置 > Feed。
点击添加新 Feed。
点击配置单个 Feed。
在Feed 名称字段中，输入 Feed 的名称（例如 Code42 Incydr Datasets）。
选择 Google Cloud Storage V2 作为来源类型。
选择 Code42 Incydr 作为日志类型。
点击下一步。
为以下输入参数指定值：
- 存储分区网址：输入带有前缀路径的 GCS 存储分区 URI：
```
gs://code42-incydr-logs/code42/
```
  - 将
    - code42-incydr-logs：您的 GCS 存储分区名称。
    - code42/：存储日志的可选前缀/文件夹路径（留空表示根目录）。
  - 示例：
    - 根存储分区：gs://code42-incydr-logs/
    - 带前缀：gs://code42-incydr-logs/code42/
注意：请务必在 URI 末尾添加尾随斜杠 (/)。
- 来源删除选项：根据您的偏好选择删除选项：
  - 永不：永不删除转移后的任何文件（建议用于测试）。
  - 删除已转移的文件：在成功转移后删除文件。
  - 删除已转移的文件和空目录：在成功转移后删除文件和空目录。
  注意：如果您选择删除选项，服务账号必须具有 Storage Object Admin 角色，而不是 Storage Object Viewer 角色。相应地更新 IAM 权限。
- 文件存在时间上限：包含在过去指定天数内修改的文件。默认值为 180 天。
- 资产命名空间：资产命名空间。
- 注入标签：要应用于此 Feed 中事件的标签。
点击下一步。
在最终确定界面中查看新的 Feed 配置，然后点击提交。