收集 Group-IB 威脅情報記錄

支援的國家/地區：

本文說明如何使用 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 金鑰。

建立 Google Cloud Storage 值區

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

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

    設定 值
    為 bucket 命名 輸入全域不重複的名稱 (例如 groupib-ti-logs)
    位置類型 根據需求選擇 (區域、雙區域、多區域)
    位置 選取位置 (例如 us-central1)
    儲存空間級別 標準 (建議用於經常存取的記錄)
    存取控管 統一 (建議)

  6. 點選「建立」。

部署 Cloud 函式來提取 Group-IB 資料

建立 Cloud 函式，從 Group-IB TI&A API 提取威脅情資資料，並以 NDJSON 檔案的形式寫入 GCS bucket，供 Google SecOps 擷取。

建立 Cloud 函式

  1. 前往 Google Cloud Console。
  2. 前往 Cloud Functions。
  3. 按一下「Create Function」(建立函式)。

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

    設定 值
    環境 第 2 代
    函式名稱 groupib-to-gcs
    區域 選取最靠近 GCS bucket 的區域
    觸發條件類型 HTTPS
    驗證 需要驗證
    分配的記憶體 512 MB (如果擷取大型集合，請增加此值)
    逾時 540 秒

  5. 點選「下一步」。

  6. 將「執行階段」設為 Python 3.11 (或更新版本)。

  7. 將「Entry point」(進入點) 設為 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. 點選「Deploy」(部署)。

設定環境變數

  1. 部署完成後，請前往函式詳細資料頁面。
  2. 按一下 [編輯]。
  3. 展開「執行階段、建構作業、連線和安全性設定」專區。

  4. 在「Runtime environment variables」(執行階段環境變數) 下方，新增下列變數：

    變數 值
    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 bucket 名稱 (例如 groupib-ti-logs)
    GCS_PREFIX 記錄檔的前置字串 (例如 groupib-ti)
    GIB_COLLECTIONS 要擷取的集合清單 (以半形逗號分隔，詳情請參閱下文)
    DEFAULT_DAYS_BACK 首次執行時要回溯的天數 (預設值：3)

  5. 點選「Deploy」(部署)。

可用的 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 網路犯罪 (HI) 威脅報告，內含 IOC 和 MITRE ATT&CK 對應 20
hi/threat_actor 網路罪犯威脅發動者檔案 100
apt/threat APT (國家級) 威脅報告，內含 IOC 和 MITRE ATT&CK 對應 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 開放式 Proxy IP 位址 100
suspicious_ip/socks_proxy SOCKS Proxy IP 位址 100

  • 範例：如要收集遭盜用的憑證、C2 基礎架構和 APT 資料，請執行下列操作：

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

排定 Cloud 函式

使用 Cloud Scheduler 定期觸發函式。

  1. 前往 Google Cloud 控制台的 Cloud Scheduler。
  2. 點選「建立工作」。

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

    設定 值
    名稱 groupib-to-gcs-schedule
    區域 與 Cloud Function 相同的區域
    頻率 0 */1 * * * (每小時) 或 0 0 * * * (每天)
    時區 選取時區

  4. 在「設定執行作業」下方：

    • 目標類型：選取「HTTP」。
    • 網址：輸入 Cloud 函式觸發網址。
    • HTTP 方法：選取「POST」。
    • 驗證標頭：選取「新增 OIDC 權杖」。
    • 服務帳戶：選取具備 roles/cloudfunctions.invoker 權限的服務帳戶。

  5. 點選「建立」。

測試資料匯出功能

  1. 在 Cloud Scheduler 控制台中，按一下工作旁的「強制執行」，手動觸發函式。
  2. 在 Cloud Logging 中查看 Cloud Functions 記錄，確認資料已從 Group-IB 擷取。
  3. 前往 Google Cloud 控制台的「Cloud Storage」>「Buckets」頁面。
  4. 按一下 bucket 名稱 (例如 groupib-ti-logs)。
  5. 前往前置字串資料夾 (例如 groupib-ti/)。

  6. 確認 bucket 中是否顯示新的 .ndjson 檔案。

擷取 Google SecOps 服務帳戶

Google SecOps 會使用專屬服務帳戶，從 GCS bucket 讀取資料。您必須授予這個服務帳戶值區存取權。

取得服務帳戶電子郵件地址

  1. 依序前往「SIEM 設定」>「動態饋給」。
  2. 按一下「新增動態消息」。
  3. 按一下「設定單一動態饋給」。
  4. 在「動態饋給名稱」欄位中輸入動態饋給名稱 (例如 Group-IB Threat Intelligence)。
  5. 選取「Google Cloud Storage V2」做為「來源類型」。
  6. 選取「Group-IB Threat Intelligence」做為「記錄類型」。

  7. 按一下「取得服務帳戶」。系統會顯示專屬的服務帳戶電子郵件地址，例如：

    chronicle-12345678@chronicle-gcp-prod.iam.gserviceaccount.com

  8. 複製這個電子郵件地址，以便在下一步中使用。

將 IAM 權限授予 Google SecOps 服務帳戶

Google SecOps 服務帳戶需要 GCS bucket 的「Storage 物件檢視者」角色。

  1. 依序前往「Cloud Storage」>「Buckets」。
  2. 按一下 bucket 名稱 (例如 groupib-ti-logs)。
  3. 前往「權限」分頁標籤。
  4. 按一下「授予存取權」。
  5. 請提供下列設定詳細資料：
    • 新增主體：貼上 Google SecOps 服務帳戶電子郵件地址。
    • 指派角色：選取「Storage 物件檢視者」。

  6. 按一下 [儲存]。

在 Google SecOps 中設定資訊提供，擷取 Group-IB 威脅情報記錄

  1. 依序前往「SIEM 設定」>「動態饋給」。
  2. 按一下「新增動態消息」。
  3. 按一下「設定單一動態饋給」。
  4. 在「動態饋給名稱」欄位中輸入動態饋給名稱 (例如 Group-IB Threat Intelligence)。
  5. 選取「Google Cloud Storage V2」做為「來源類型」。
  6. 選取「Group-IB Threat Intelligence」做為「記錄類型」。
  7. 點選「下一步」。

  8. 指定下列輸入參數的值：

    • 儲存空間 bucket URL：輸入 GCS bucket URI，並加上前置路徑：

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

      更改項目：

      • groupib-ti-logs：您的 GCS bucket 名稱。
      • groupib-ti：儲存記錄的前置字元/資料夾路徑。

    • 來源刪除選項：根據偏好設定選取刪除選項：

      • 永不：移轉後一律不刪除任何檔案 (建議用於測試)。
      • 刪除已轉移的檔案：成功轉移檔案後刪除檔案。
      • 刪除已轉移的檔案和空白目錄：成功轉移檔案後，刪除檔案和空白目錄。

    • 檔案存在時間上限：包含在過去天數內修改的檔案。預設值為 180 天。

    • 資產命名空間：資產命名空間。

    • 擷取標籤：要套用至這個動態饋給事件的標籤。

  9. 點選「下一步」。

  10. 在「Finalize」(完成) 畫面中檢查新的動態饋給設定，然後按一下「Submit」(提交)。

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