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

收集 URLScan IO 記錄

支援的國家/地區：

Google SecOps SIEM

本文說明如何使用 Amazon S3，將 URLScan IO 記錄擷取至 Google Security Operations。

事前準備

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

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

取得 URLScan IO 的必要條件

登入 URLScan IO。
按一下個人資料圖示。
從選單中選取「API 金鑰」。
如果您尚未拿到 API 金鑰：
- 按一下「建立 API 金鑰」按鈕。
- 輸入 API 金鑰的說明 (例如 Google SecOps Integration)。
- 選取金鑰的權限 (如要取得唯讀存取權，請選取「讀取」權限)。
- 點按「Generate API Key」(產生 API 金鑰)。
複製下列詳細資料並儲存在安全位置：
- API_KEY：產生的 API 金鑰字串 (格式：xxxxxxxx-xxxx-xxxx-xxxx-xxxxxxxxxxxx)
- API 基礎網址：https://urlscan.io/api/v1 (所有使用者皆相同)
請注意 API 配額限制：
- 免費帳戶：每天最多可發出 1, 000 次 API 呼叫，每分鐘最多 60 次
- 專業版帳戶：根據訂閱方案提供較高的上限
如要將搜尋範圍限制在貴機構的掃描結果，請記下：
- 使用者 ID：您的使用者名稱或電子郵件地址 (用於 user: 搜尋篩選器)
- 團隊 ID：如果使用團隊功能 (適用於 team: 搜尋篩選器)

為 Google SecOps 設定 AWS S3 值區和 IAM

按照這份使用者指南建立 Amazon S3 bucket：建立 bucket。
儲存 bucket 的「名稱」和「地區」，以供日後參考 (例如 urlscan-logs-bucket)。
請按照這份使用者指南建立使用者：建立 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:::urlscan-logs-bucket/*"
    },
    {
      "Sid": "AllowGetStateObject",
      "Effect": "Allow",
      "Action": "s3:GetObject",
      "Resource": "arn:aws:s3:::urlscan-logs-bucket/urlscan/state.json"
    }
  ]
}

如果您輸入其他 bucket 名稱，請替換 urlscan-logs-bucket。

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

建立 Lambda 函式

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

設定值

名稱 urlscan-collector

執行階段 Python 3.13

架構 x86_64

執行角色 urlscan-lambda-role

設定	值
名稱	`urlscan-collector`
執行階段	Python 3.13
架構	x86_64
執行角色	`urlscan-lambda-role`

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

import json
import os
import boto3
from datetime import datetime, timedelta
import urllib3
import base64

s3 = boto3.client('s3')
http = urllib3.PoolManager()

def lambda_handler(event, context):
    # Environment variables
    bucket = os.environ['S3_BUCKET']
    prefix = os.environ['S3_PREFIX']
    state_key = os.environ['STATE_KEY']
    api_key = os.environ['API_KEY']
    api_base = os.environ['API_BASE']
    search_query = os.environ.get('SEARCH_QUERY', 'date:>now-1h')
    page_size = int(os.environ.get('PAGE_SIZE', '100'))
    max_pages = int(os.environ.get('MAX_PAGES', '10'))

    # Load state
    state = load_state(bucket, state_key)
    last_run = state.get('last_run')

    # Prepare search query
    if last_run:
        # Adjust search query based on last run
        search_time = datetime.fromisoformat(last_run)
        time_diff = datetime.utcnow() - search_time
        hours = int(time_diff.total_seconds() / 3600) + 1
        search_query = f'date:>now-{hours}h'

    # Search for scans
    headers = {'API-Key': api_key}
    all_results = []

    for page in range(max_pages):
        search_url = f"{api_base}/search/"
        params = {
            'q': search_query,
            'size': page_size,
            'offset': page * page_size
        }

        # Make search request
        response = http.request(
            'GET',
            search_url,
            fields=params,
            headers=headers
        )

        if response.status != 200:
            print(f"Search failed: {response.status}")
            break

        search_data = json.loads(response.data.decode('utf-8'))
        results = search_data.get('results', [])

        if not results:
            break

        # Fetch full result for each scan
        for result in results:
            uuid = result.get('task', {}).get('uuid')
            if uuid:
                result_url = f"{api_base}/result/{uuid}/"
                result_response = http.request(
                    'GET',
                    result_url,
                    headers=headers
                )

                if result_response.status == 200:
                    full_result = json.loads(result_response.data.decode('utf-8'))
                    all_results.append(full_result)
                else:
                    print(f"Failed to fetch result for {uuid}: {result_response.status}")

        # Check if we have more pages
        if len(results) < page_size:
            break

    # Write results to S3
    if all_results:
        now = datetime.utcnow()
        file_key = f"{prefix}year={now.year}/month={now.month:02d}/day={now.day:02d}/hour={now.hour:02d}/urlscan_{now.strftime('%Y%m%d_%H%M%S')}.json"

        # Create NDJSON content
        ndjson_content = '\n'.join([json.dumps(r, separators=(',', ':')) for r in all_results])

        # Upload to S3
        s3.put_object(
            Bucket=bucket,
            Key=file_key,
            Body=ndjson_content.encode('utf-8'),
            ContentType='application/x-ndjson'
        )

        print(f"Uploaded {len(all_results)} results to s3://{bucket}/{file_key}")

    # Update state
    state['last_run'] = datetime.utcnow().isoformat()
    save_state(bucket, state_key, state)

    return {
        'statusCode': 200,
        'body': json.dumps({
            'message': f'Processed {len(all_results)} scan results',
            'location': f"s3://{bucket}/{prefix}"
        })
    }

def load_state(bucket, key):
    try:
        response = s3.get_object(Bucket=bucket, Key=key)
        return json.loads(response['Body'].read())
    except s3.exceptions.NoSuchKey:
        return {}
    except Exception as e:
        print(f"Error loading state: {e}")
        return {}

def save_state(bucket, key, state):
    try:
        s3.put_object(
            Bucket=bucket,
            Key=key,
            Body=json.dumps(state),
            ContentType='application/json'
        )
    except Exception as e:
        print(f"Error saving state: {e}")

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

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

鍵	範例值
`S3_BUCKET`	`urlscan-logs-bucket`
`S3_PREFIX`	`urlscan/`
`STATE_KEY`	`urlscan/state.json`
`API_KEY`	`<your-api-key>`
`API_BASE`	`https://urlscan.io/api/v1`
`SEARCH_QUERY`	`date:>now-1h`
`PAGE_SIZE`	`100`
`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 函式 urlscan-collector。
- 名稱：urlscan-collector-1h。
按一下「建立時間表」。

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

前往 AWS 控制台 > IAM > Users。
點選 [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:::urlscan-logs-bucket/*"
    },
    {
      "Effect": "Allow",
      "Action": ["s3:ListBucket"],
      "Resource": "arn:aws:s3:::urlscan-logs-bucket"
    }
  ]
}

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

在 Google SecOps 中設定動態消息，擷取 URLScan IO 記錄

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

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