收集 Group-IB 威胁情报日志

支持的平台：

Google SecOps SIEM

本文档介绍了如何使用 Google Cloud Storage 将 Group-IB 威胁情报日志注入到 Google Security Operations。

Group-IB Threat Intelligence & Attribution (TI&A) 是一个网络威胁情报平台，可提供有关威胁行为者、失陷指标 (IOC)、恶意软件、命令和控制 (C2) 基础设施、遭入侵的凭据、钓鱼式攻击活动和漏洞的实时数据。它会汇总来自开放网络、深网和暗网来源的情报，使安全团队能够主动检测和应对威胁。

准备工作

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

• Google SecOps 实例
• 已启用 API 访问权限的 Group-IB TI&A 账号
• Group-IB TI&A 门户网站的访问权限 (tap.group-ib.com)
• 已启用以下 API 的 Google Cloud 云项目：
  • Cloud Storage API
  • Cloud Functions API
  • Cloud Scheduler API
  • Cloud Build API

生成 Group-IB API 密钥

1. 访问 https://tap.group-ib.com，登录 Group-IB TI&A 门户。
2. 点击右上角的名称，然后选择个人资料。
3. 点击前往我的设置。
4. 前往安全和访问权限标签页。
5. 在个人令牌部分，点击生成新令牌。

6. 复制并妥善保存 API 密钥。

   注意： API 密钥会与您的电子邮件地址一起用作 HTTP 基本身份验证的密码。请妥善保管。如果您丢失了密钥，则必须生成一个新密钥。

重要提示： 请与 Group-IB 支持团队联系，将您的 Cloud Function（或 NAT 网关）的 IP 地址添加到许可名单中，以免 API 请求被阻止。

创建 Google Cloud Storage 存储分区

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

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

   设置	值
   为存储桶命名	输入一个全局唯一的名称（例如 groupib-ti-logs）
   位置类型	根据您的需求进行选择（区域级、双区域级、多区域级）
   位置	选择营业地点（例如 us-central1）
   存储类别	标准（建议用于经常访问的日志）
   访问权限控制	均匀（推荐）

6. 点击创建。

部署 Cloud Functions 函数以拉取 Group-IB 数据

创建一个 Cloud Function，用于从 Group-IB TI&A API 中注入威胁情报数据，并将其作为 NDJSON 文件写入 GCS 存储桶，以供 Google SecOps 注入。

创建 Cloud Function

1. 转到 Google Cloud Console。
2. 前往 Cloud Functions。
3. 点击创建函数。

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

   设置	值
   环境	第 2 代
   函数名称	groupib-to-gcs
   区域	选择离您的 GCS 存储桶最近的区域
   触发器类型	HTTPS
   Authentication	需要进行身份验证
   分配的内存	512 MB（如果提取大型集合，则增加此值）
   超时	540 秒

5. 点击下一步。

6. 将运行时设置为 Python 3.11（或更高版本）。

7. 将入口点设置为 main。

8. 将 main.py 的内容替换为以下代码：

   import json
   import os
   import requests
   import functions_framework
   from datetime import datetime, timedelta, timezone
   from urllib.parse import urljoin
   from requests.auth import HTTPBasicAuth
   from google.cloud import storage
   
   GIB_API_URL = os.environ.get('GIB_API_URL', 'https://tap.group-ib.com/api/v2/')
   GIB_USERNAME = os.environ.get('GIB_USERNAME')
   GIB_API_KEY = os.environ.get('GIB_API_KEY')
   GCS_BUCKET = os.environ.get('GCS_BUCKET')
   GCS_PREFIX = os.environ.get('GCS_PREFIX', 'groupib-ti')
   COLLECTIONS = os.environ.get('GIB_COLLECTIONS', 'compromised/account,malware/cnc,apt/threat,hi/threat').split(',')
   DEFAULT_DAYS_BACK = int(os.environ.get('DEFAULT_DAYS_BACK', '3'))
   
   # Max items per request: 100 for most collections, 20 for apt/threat and hi/threat
   BIG_DATA_COLLECTIONS = ['apt/threat', 'hi/threat']
   
   # File to persist seqUpdate values between runs (use GCS for durability)
   STATE_BLOB = '_state/seq_updates.json'
   
   def load_state():
       """Load seqUpdate state from GCS."""
       client = storage.Client()
       bucket = client.bucket(GCS_BUCKET)
       blob = bucket.blob(f"{GCS_PREFIX}/{STATE_BLOB}")
       if blob.exists():
           return json.loads(blob.download_as_text())
       return {}
   
   def save_state(state):
       """Save seqUpdate state to GCS."""
       client = storage.Client()
       bucket = client.bucket(GCS_BUCKET)
       blob = bucket.blob(f"{GCS_PREFIX}/{STATE_BLOB}")
       blob.upload_from_string(json.dumps(state), content_type='application/json')
   
   def gib_request(session, url, params=None):
       """Send authenticated GET request to Group-IB API."""
       resp = session.get(url, params=params)
       if resp.status_code == 301:
           raise Exception('IP not whitelisted by Group-IB. Contact Group-IB support.')
       resp.raise_for_status()
       return resp.json()
   
   def get_seq_update_by_date(session, collection, date_str):
       """Get seqUpdate value for a collection starting from a given date."""
       url = urljoin(GIB_API_URL, 'sequence_list')
       data = gib_request(session, url, {'date': date_str, 'collection': collection})
       return data.get('list', {}).get(collection)
   
   def fetch_collection(session, collection, seq_update):
       """Fetch all new items from a collection starting after the given seqUpdate."""
       limit = 20 if collection in BIG_DATA_COLLECTIONS else 100
       url = urljoin(GIB_API_URL, f"{collection}/updated")
       all_items = []
       last_seq = seq_update
   
       while True:
           data = gib_request(session, url, {'seqUpdate': str(last_seq), 'limit': limit})
           items = data.get('items', [])
           if not items:
               break
           all_items.extend(items)
           last_seq = items[-1].get('seqUpdate')
   
       return all_items, last_seq
   
   def write_to_gcs(items, collection_name):
       """Write items to GCS as NDJSON."""
       if not items:
           return 0
       client = storage.Client()
       bucket = client.bucket(GCS_BUCKET)
       timestamp = datetime.now(timezone.utc).strftime('%Y%m%d_%H%M%S')
       safe_name = collection_name.replace('/', '_')
       blob_path = f"{GCS_PREFIX}/{safe_name}_{timestamp}.ndjson"
       blob = bucket.blob(blob_path)
       ndjson = '\n'.join(json.dumps(item, ensure_ascii=False) for item in items) + '\n'
       blob.upload_from_string(ndjson, content_type='application/x-ndjson')
       return len(items)
   
   @functions_framework.http
   def main(request):
       """Cloud Function entry point."""
       session = requests.Session()
       session.auth = HTTPBasicAuth(GIB_USERNAME, GIB_API_KEY)
       session.headers.update({'Accept': '*/*'})
   
       state = load_state()
       total = 0
   
       for collection in COLLECTIONS:
           collection = collection.strip()
           seq_update = state.get(collection)
   
           if seq_update is None:
               default_date = (datetime.now(timezone.utc) - timedelta(days=DEFAULT_DAYS_BACK)).strftime('%Y-%m-%d')
               seq_update = get_seq_update_by_date(session, collection, default_date)
               if seq_update is None:
                   continue
   
           items, last_seq = fetch_collection(session, collection, seq_update)
           if items:
               write_to_gcs(items, collection)
               total += len(items)
           state[collection] = last_seq
   
       save_state(state)
       return json.dumps({'status': 'success', 'total_items': total}), 200
   

9. 将 requirements.txt 的内容替换为以下依赖项：

   functions-framework==3.*
   requests>=2.28.0
   google-cloud-storage>=2.0.0
   

10. 点击部署。

配置环境变量

1. 部署完成后，前往函数详情页面。
2. 点击修改。
3. 展开运行时、构建、连接和安全设置部分。

4. 在运行时环境变量下，添加以下变量：

   变量	值
   GIB_API_URL	https://tap.group-ib.com/api/v2/（或 https://bt.group-ib.com/api/v2/，具体取决于您所在的区域）
   GIB_USERNAME	您的 Group-IB 账号电子邮件地址
   GIB_API_KEY	您的 Group-IB API 密钥（个人令牌）
   GCS_BUCKET	您的 GCS 存储桶名称（例如 groupib-ti-logs）
   GCS_PREFIX	日志文件的前缀（例如 groupib-ti）
   GIB_COLLECTIONS	要提取的集合的逗号分隔列表（见下文）
   DEFAULT_DAYS_BACK	首次运行时回溯的天数（默认值：3）

5. 点击部署。

   注意： 对于生产环境，请将敏感值 (GIB_API_KEY) 存储在 Secret Manager 中，而不是存储在纯环境变量中。在函数配置中使用引用 Secret 选项来引用它们。

可用的 Group-IB 集合

使用与您的使用场景相关的集合配置 GIB_COLLECTIONS 变量：

集合	说明	最大限制
compromised/account	被盗用的账号凭据（登录名、密码、网域）	100
compromised/card	银行卡遭盗用	100
compromised/mule	钱骡账号	100
compromised/imei	已遭破解的移动设备的 IMEI	100
compromised/file	包含恶意软件归因信息的遭入侵文件	100
attacks/ddos	DDoS 攻击数据（目标 IP、网域）	100
attacks/deface	网站涂改事件	100
attacks/phishing	钓鱼式攻击网址和网域	100
attacks/phishing_kit	钓鱼式攻击工具包哈希和目标品牌	100
bp/phishing	品牌保护 - 钓鱼式攻击事件	100
bp/phishing_kit	品牌保护 - 钓鱼式攻击工具包	100
hi/threat	包含 IOC 和 MITRE ATT&CK 映射的网络犯罪 (HI) 威胁报告	20
hi/threat_actor	网络犯罪威胁行为者个人资料	100
apt/threat	包含 IOC 和 MITRE ATT&CK 映射的 APT（国家级）威胁报告	20
apt/threat_actor	APT 威胁行为者个人资料	100
malware/cnc	命令与控制服务器指标（IP、网域）	100
malware/malware	恶意软件说明和威胁级别	100
malware/targeted_malware	目标恶意软件样本（哈希、文件名）	100
osi/git_leak	Git 代码库数据泄露	100
osi/public_leak	公开数据泄露（粘贴、转储）	100
osi/vulnerability	包含 CVSS 评分的漏洞数据	100
suspicious_ip/tor_node	Tor 退出节点 IP 地址	100
suspicious_ip/open_proxy	开放代理 IP 地址	100
suspicious_ip/socks_proxy	SOCKS 代理 IP 地址	100

• 示例：收集遭入侵的凭据、C2 基础设施和 APT 数据：

  compromised/account,malware/cnc,apt/threat,hi/threat
  

  注意： 由于响应大小较大，apt/threat 和 hi/threat 合集 的每个 API 请求最多只能包含 20 项。Cloud Functions 会自动处理此问题。如果使用 POC 或合作伙伴许可，对数据的访问权限可能仅限 30 天。请勿使用 attacks/phishing 中的 IP 地址进行检测，这可能会导致许多误报。仅关注网址。

安排 Cloud Function

使用 Cloud Scheduler 定期触发该函数。

1. 在 Google Cloud Console 中，前往 Cloud Scheduler。
2. 点击创建作业。

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

   设置	值
   名称	groupib-to-gcs-schedule
   区域	与 Cloud Function 相同的区域
   频率	0 */1 * * *（每小时）或 0 0 * * *（每天）
   时区	选择时区

4. 在配置执行下：

   • 目标类型：选择 HTTP。
   • 网址：输入 Cloud Function 触发器网址。
   • HTTP 方法：选择 POST。
   • 身份验证标头：选择添加 OIDC 令牌。
   • 服务账号：选择具有 roles/cloudfunctions.invoker 权限的服务账号。

5. 点击创建。

   注意： Cloud Functions 函数使用存储在 GCS 中的 seqUpdate 值跟踪其位置。在首次运行中，该函数会回溯 DEFAULT_DAYS_BACK 天（默认值为 3）。无论调度程序的频率如何，后续运行都会从上次运行结束的位置开始。

测试数据导出

1. 在 Cloud Scheduler 控制台中，点击作业旁边的强制运行，以手动触发函数。
2. 在 Cloud Logging 中查看 Cloud Function 日志，验证是否已从 Group-IB 获取数据。
3. 在 Google Cloud Console 中，前往 Cloud Storage > 存储分区。
4. 点击您的存储桶名称（例如 groupib-ti-logs）。
5. 前往前缀文件夹（例如 groupib-ti/）。

6. 验证存储桶中是否显示新的 .ndjson 文件。

   注意： 每个日志文件都以时间戳命名，格式为 {collection}_{YYYYMMDD_HHMMSS}.ndjson（例如 ioc_common_20260318_120000.ndjson）。

检索 Google SecOps 服务账号

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

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

1. 依次前往 SIEM 设置 > Feed。
2. 点击添加新 Feed。
3. 点击配置单个 Feed。
4. 在 Feed 名称字段中，输入 Feed 的名称（例如 Group-IB Threat Intelligence）。
5. 选择 Google Cloud Storage V2 作为来源类型。
6. 选择 Group-IB 威胁情报作为日志类型。

7. 点击获取服务账号。系统会显示一个唯一的服务账号电子邮件地址，例如：

   [e-mail]
   

8. 复制此电子邮件地址，以便在下一步中使用。

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

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

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

1. 前往 Cloud Storage > 存储分区。
2. 点击您的存储桶名称（例如 groupib-ti-logs）。
3. 前往权限标签页。
4. 点击授予访问权限。
5. 提供以下配置详细信息：
   • 添加主账号：粘贴 Google SecOps 服务账号电子邮件地址。
   • 分配角色：选择 Storage Object Viewer。

6. 点击保存。

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

在 Google SecOps 中配置 Feed 以注入 Group-IB 威胁情报日志

1. 依次前往 SIEM 设置 > Feed。
2. 点击添加新 Feed。
3. 点击配置单个 Feed。
4. 在 Feed 名称字段中，输入 Feed 的名称（例如 Group-IB Threat Intelligence）。
5. 选择 Google Cloud Storage V2 作为来源类型。
6. 选择 Group-IB 威胁情报作为日志类型。
7. 点击下一步。

8. 为以下输入参数指定值：

   • 存储桶网址：输入带有前缀路径的 GCS 存储桶 URI：

     gs://groupib-ti-logs/groupib-ti/
     

     将

     • groupib-ti-logs：您的 GCS 存储桶名称。
     • groupib-ti：存储日志的前缀/文件夹路径。
     注意： 请务必在 URI 末尾添加尾随斜杠 (/)。

   • 来源删除选项：根据您的偏好选择删除选项：

     • 永不：转移后永不删除任何文件（建议用于测试）。
     • 删除已转移的文件：在成功转移后删除文件。
     • 删除已转移的文件和空目录：成功转移后删除文件和空目录。
     注意 ：如果您选择删除选项，服务账号必须具有 Storage Object Admin 角色，而不是 Storage Object Viewer 角色。相应地更新 IAM 权限。

   • 文件存在时间上限：包含在过去指定天数内修改的文件。默认值为 180 天。

   • 资产命名空间：资产命名空间。

   • 注入标签：要应用于此 Feed 中事件的标签。

9. 点击下一步。

10. 在最终确定界面中查看新的 Feed 配置，然后点击提交。

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