本頁面由 Cloud Translation API 翻譯而成。

收集 Team Cymru Scout 威脅情報資料

支援的國家/地區：

Google SecOps SIEM

本文說明如何使用 Amazon S3，將 Team Cymru Scout 威脅情報資料擷取至 Google Security Operations。

事前準備

請確認您已完成下列事前準備事項：

Google SecOps 執行個體
Team Cymru Scout 租戶的特殊存取權
AWS 的特殊存取權 (S3、IAM、Lambda、EventBridge)

取得 Team Cymru Scout 的必要條件

登入 Team Cymru Scout Platform。
前往 API 金鑰網頁。
按一下 [建立] 按鈕。
視需要提供金鑰的說明。
按一下「建立金鑰」按鈕，即可產生 API 金鑰。
複製下列詳細資料並儲存在安全位置：
- SCOUT_API_KEY - API 存取金鑰
- SCOUT_BASE_URL - Scout API 基本網址

為 Google SecOps 設定 AWS S3 值區和 IAM

按照這份使用者指南建立 Amazon S3 bucket：建立 bucket。
儲存 bucket 的「名稱」和「地區」，以供日後參考 (例如 team-cymru-scout-ti)。
請按照這份使用者指南建立使用者：建立 IAM 使用者。
選取建立的「使用者」。
選取「安全憑證」分頁標籤。
在「Access Keys」部分中，按一下「Create Access Key」。
選取「第三方服務」做為「用途」。
點選「下一步」。
選用：新增說明標記。
按一下「建立存取金鑰」。
按一下「下載 CSV 檔案」，儲存「存取金鑰」和「私密存取金鑰」以供日後參考。
按一下 [完成]。
選取「權限」分頁標籤。
在「權限政策」部分中，按一下「新增權限」。
選取「新增權限」。
選取「直接附加政策」。
搜尋 AmazonS3FullAccess 政策。
選取政策。
點選「下一步」。
按一下「Add permissions」。

設定 S3 上傳的身分與存取權管理政策和角色

在 AWS 控制台中，前往「IAM」>「Policies」。
按一下「建立政策」>「JSON」分頁。

輸入下列政策：

{
"Version": "2012-10-17",
"Statement": [
    {
    "Sid": "AllowPutObjects",
    "Effect": "Allow",
    "Action": "s3:PutObject",
    "Resource": "arn:aws:s3:::team-cymru-scout-ti/*"
    },
    {
    "Sid": "AllowGetStateObject",
    "Effect": "Allow",
    "Action": "s3:GetObject",
    "Resource": "arn:aws:s3:::team-cymru-scout-ti/team-cymru/scout-ti/state.json"
    }
]
}

如果您輸入其他 bucket 名稱，請替換 team-cymru-scout-ti。

依序點選「下一步」>「建立政策」。
依序前往「IAM」>「Roles」>「Create role」>「AWS service」>「Lambda」。
附加新建立的政策。
為角色命名 TeamCymruScoutToS3Role，然後按一下「建立角色」。

建立 Lambda 函式

在 AWS 控制台中，依序前往「Lambda」>「Functions」>「Create function」。
按一下「從頭開始撰寫」。
請提供下列設定詳細資料：

設定值

名稱 team_cymru_scout_ti_to_s3

執行階段 Python 3.13

架構 x86_64

執行角色 TeamCymruScoutToS3Role

設定	值
名稱	`team_cymru_scout_ti_to_s3`
執行階段	Python 3.13
架構	x86_64
執行角色	`TeamCymruScoutToS3Role`

建立函式後，開啟「程式碼」分頁，刪除存根並輸入下列程式碼 (team_cymru_scout_ti_to_s3.py)：

```python
#!/usr/bin/env python3
# Lambda: Pull Team Cymru Scout Threat Intelligence exports to S3 (no transform)

import os, json, time
from urllib.request import Request, urlopen
from urllib.error import HTTPError, URLError
import boto3

S3_BUCKET    = os.environ["S3_BUCKET"]
S3_PREFIX    = os.environ.get("S3_PREFIX", "team-cymru/scout-ti/")
STATE_KEY    = os.environ.get("STATE_KEY", "team-cymru/scout-ti/state.json")
WINDOW_SEC   = int(os.environ.get("WINDOW_SECONDS", "3600"))
HTTP_TIMEOUT = int(os.environ.get("HTTP_TIMEOUT", "60"))
HTTP_RETRIES = int(os.environ.get("HTTP_RETRIES", "3"))
MODE         = os.environ.get("MODE", "GET").upper()
API_HEADERS  = json.loads(os.environ.get("API_HEADERS", "{}"))
MAX_PAGES    = int(os.environ.get("MAX_PAGES", "10"))

# GET mode
DOWNLOAD_URL_TEMPLATE = os.environ.get("DOWNLOAD_URL_TEMPLATE", "")
# POST_JSON mode
API_URL            = os.environ.get("API_URL", "")
JSON_BODY_TEMPLATE = os.environ.get("JSON_BODY_TEMPLATE", "")

# Team Cymru Scout specific
SCOUT_BASE_URL = os.environ.get("SCOUT_BASE_URL", "https://api.scout.cymru.com")
SCOUT_API_KEY  = os.environ.get("SCOUT_API_KEY", "")

s3 = boto3.client("s3")

def _iso(ts: float) -> str:
    return time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime(ts))

def _get_state() -> dict:
    try:
        obj = s3.get_object(Bucket=S3_BUCKET, Key=STATE_KEY)
        b = obj["Body"].read()
        return json.loads(b) if b else {}
    except Exception:
        return {}

def _put_state(st: dict):
    s3.put_object(
        Bucket=S3_BUCKET, Key=STATE_KEY,
        Body=json.dumps(st, separators=(",", ":")).encode("utf-8"),
        ContentType="application/json",
    )

def _http(url: str, method: str = "GET", body: bytes | None = None) -> tuple[bytes, str]:
    attempt = 0
    while True:
        try:
            req = Request(url, method=method)
            # Add headers
            headers = API_HEADERS.copy()
            if SCOUT_API_KEY and "Authorization" not in headers:
                headers["Authorization"] = f"Bearer {SCOUT_API_KEY}"
            headers.setdefault("Accept", "application/json")

            for k, v in headers.items():
                req.add_header(k, v)

            if body is not None:
                req.add_header("Content-Type", "application/json")

            with urlopen(req, data=body, timeout=HTTP_TIMEOUT) as r:
                return r.read(), r.headers.get("Content-Type", "application/json")
        except HTTPError as e:
            if e.code in (429, 500, 502, 503, 504) and attempt < HTTP_RETRIES:
                delay = 1 + attempt
                try:
                    delay = int(e.headers.get("Retry-After", delay))
                except Exception:
                    pass
                time.sleep(max(1, delay))
                attempt += 1
                continue
            raise
        except URLError:
            if attempt < HTTP_RETRIES:
                time.sleep(1 + attempt)
                attempt += 1
                continue
            raise

def _write(blob: bytes, ctype: str, from_ts: float, to_ts: float, page: int) -> str:
    date_path = time.strftime("%Y/%m/%d", time.gmtime(to_ts))
    key = f"{S3_PREFIX}/{date_path}/scout_ti_{int(from_ts)}_{int(to_ts)}_p{page:03d}.json"
    s3.put_object(Bucket=S3_BUCKET, Key=key, Body=blob, ContentType=ctype or "application/json")
    return key

def _next_cursor(obj: dict) -> str | None:
    if not isinstance(obj, dict):
        return None

    for container in (obj, obj.get("meta", {}) or {}, obj.get("metadata", {}) or {}):
        for k in ("next", "next_cursor", "nextCursor", "nextPageToken", "continuation", "cursor", "pagedResultsCookie"):
            v = container.get(k)
            if v:
                return str(v)
    return None

def _loop(from_ts: float, to_ts: float) -> dict:
    cursor, page, written = None, 0, 0

    while page < MAX_PAGES:
        if MODE == "GET":
            if DOWNLOAD_URL_TEMPLATE:
                url = (DOWNLOAD_URL_TEMPLATE
                    .replace("{FROM}", _iso(from_ts))
                    .replace("{TO}", _iso(to_ts))
                    .replace("{CURSOR}", cursor or ""))
            else:
                # Default Scout API endpoint (adjust based on actual API)
                url = f"{SCOUT_BASE_URL}/v1/threat-intelligence?start={_iso(from_ts)}&end={_iso(to_ts)}"
                if cursor:
                    url += f"&cursor={cursor}"
            blob, ctype = _http(url, method="GET")
        else:
            assert API_URL and JSON_BODY_TEMPLATE, "API_URL and JSON_BODY_TEMPLATE required for MODE=POST_JSON"
            body = (JSON_BODY_TEMPLATE
                    .replace("{FROM}", _iso(from_ts))
                    .replace("{TO}", _iso(to_ts))
                    .replace("{CURSOR}", cursor or "")).encode("utf-8")
            blob, ctype = _http(API_URL, method="POST", body=body)

        # Normalize to JSON bytes for storage
        try:
            parsed = json.loads(blob.decode("utf-8"))
            normalized = json.dumps(parsed, separators=(",", ":")).encode("utf-8")
            ctype_out = "application/json"
        except Exception:
            normalized = blob
            ctype_out = ctype or "application/octet-stream"

        _ = _write(normalized, ctype_out, from_ts, to_ts, page)
        written += 1
        page += 1

        # Follow cursor if JSON and cursor exists
        try:
            if parsed and isinstance(parsed, dict):
                cursor = _next_cursor(parsed)
            if not cursor:
                break
        except Exception:
            break

    return {"pages": page, "objects": written}

def lambda_handler(event=None, context=None):
    st = _get_state()
    now = time.time()
    from_ts = st.get("last_to_ts") or (now - WINDOW_SEC)
    to_ts = now
    res = _loop(from_ts, to_ts)
    st["last_to_ts"] = to_ts
    _put_state(st)
    return {"ok": True, "window": {"from": _iso(from_ts), "to": _iso(to_ts)}, **res}

if __name__ == "__main__":
    print(lambda_handler())
```

依序前往「設定」>「環境變數」。
依序點選「編輯」> 新增環境變數。

輸入下列環境變數，並將換成您的值。

鍵	範例值
`S3_BUCKET`	`team-cymru-scout-ti`
`S3_PREFIX`	`team-cymru/scout-ti/`
`STATE_KEY`	`team-cymru/scout-ti/state.json`
`SCOUT_BASE_URL`	`https://api.scout.cymru.com`
`SCOUT_API_KEY`	`your-scout-api-key`
`WINDOW_SECONDS`	`3600`
`HTTP_TIMEOUT`	`60`
`HTTP_RETRIES`	`3`
`MODE`	`GET`或`POST_JSON`
`API_HEADERS`	`{"Authorization":"Bearer <token>","Accept":"application/json"}`
`DOWNLOAD_URL_TEMPLATE`	(GET 模式) 含有 `{FROM}`、`{TO}`、`{CURSOR}` 的自訂網址範本
`API_URL`	(POST_JSON 模式) API 端點網址
`JSON_BODY_TEMPLATE`	(POST_JSON 模式) 含有 `{FROM}`、`{TO}`、`{CURSOR}` 的 JSON 主體
`MAX_PAGES`	`10`

建立函式後，請留在函式頁面 (或依序開啟「Lambda」>「Functions」>「your-function」)。
選取「設定」分頁標籤。
在「一般設定」面板中，按一下「編輯」。
將「Timeout」(逾時間隔) 變更為「5 minutes (300 seconds)」(5 分鐘 (300 秒))，然後按一下「Save」(儲存)。

注意： 逾時是上限，也就是說，如果函式在計時器達到 5 分鐘時仍在執行，AWS Lambda 就會終止函式。如果執行時間較短，系統會正常完成作業，並只收取實際執行時間的費用。

建立 EventBridge 排程

依序前往「Amazon EventBridge」>「Scheduler」>「Create schedule」。
提供下列設定詳細資料：
- 週期性時間表：費率 (1 hour)。
- 目標：您的 Lambda 函式 team_cymru_scout_ti_to_s3。
- 名稱：team-cymru-scout-ti-1h。
按一下「建立時間表」。

選用：為 Google SecOps 建立唯讀 IAM 使用者和金鑰

依序前往 AWS 控制台 > IAM > 使用者 > 新增使用者。
點選 [Add users] (新增使用者)。
提供下列設定詳細資料：
- 使用者：輸入 secops-reader。
- 存取類型：選取「存取金鑰 - 程式輔助存取」。
按一下「建立使用者」。
附加最低讀取權限政策 (自訂)：依序點選「Users」(使用者) >「secops-reader」>「Permissions」(權限) >「Add permissions」(新增權限) >「Attach policies directly」(直接附加政策) >「Create policy」(建立政策)。

在 JSON 編輯器中輸入下列政策：

{
"Version": "2012-10-17",
"Statement": [
    {
    "Effect": "Allow",
    "Action": ["s3:GetObject"],
    "Resource": "arn:aws:s3:::team-cymru-scout-ti/*"
    },
    {
    "Effect": "Allow",
    "Action": ["s3:ListBucket"],
    "Resource": "arn:aws:s3:::team-cymru-scout-ti"
    }
]
}

將名稱設為 secops-reader-policy。
依序前往「建立政策」> 搜尋/選取 >「下一步」>「新增權限」。
依序前往「安全憑證」>「存取金鑰」>「建立存取金鑰」。
下載 CSV (這些值會輸入至動態饋給)。

在 Google SecOps 中設定動態消息，擷取 Team Cymru Scout 威脅情報

依序前往「SIEM 設定」>「動態饋給」。
按一下「新增動態消息」。
在「動態饋給名稱」欄位中輸入動態饋給名稱 (例如 Team Cymru Scout Threat Intelligence)。
選取「Amazon S3 V2」做為「來源類型」。
選取「Team Cymru Scout Threat Intelligence」做為「記錄類型」。
點選「下一步」。
指定下列輸入參數的值：
- S3 URI：s3://team-cymru-scout-ti/team-cymru/scout-ti/
- 來源刪除選項：根據偏好設定選取刪除選項。
- 檔案存在時間上限：包含在過去天數內修改的檔案。預設值為 180 天。
- 存取金鑰 ID：具有 S3 儲存空間存取權的使用者存取金鑰。
- 存取密鑰：具有 S3 bucket 存取權的使用者私密金鑰。
- 資產命名空間：資產命名空間。
- 擷取標籤：套用至這個動態饋給事件的標籤。
點選「下一步」。
在「完成」畫面中檢查新的動態饋給設定，然後按一下「提交」。

支援的 Team Cymru Scout 威脅情報記錄格式

Team Cymru Scout Threat Intelligence 剖析器支援 KV (LEEF) 和 CSV 格式的記錄。

支援的 Team Cymru Scout 威脅情報記錄範例

JSON

{
  "account_name": "dummy_secops_user",
  "account_type": "basic_auth",
  "used_queries": 1414,
  "remaining_queries": 48586,
  "used_queries_percentage": 2.828,
  "query_limit": 50000,
  "used_foundation_queries": 4224,
  "remaining_foundation_queries": 5776,
  "foundation_query_limit": 10000,
  "used_foundation_queries_percentage": 42.24,
  "event_type": "account_usage"
}

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