URLScan IO ログを収集する

以下でサポートされています。

Google SecOps SIEM

このドキュメントでは、Google Cloud Storage を使用して URLScan IO ログを Google Security Operations に取り込む方法について説明します。URLScan IO は、ウェブサイトを分析し、その動作、セキュリティ、パフォーマンスに関する詳細情報を提供するサービスです。URL をスキャンし、スクリーンショット、HTTP トランザクション、DNS レコード、脅威インテリジェンスデータなどの包括的なレポートを生成します。

始める前に

次の前提条件を満たしていることを確認します。

Google SecOps インスタンス
Cloud Storage API が有効になっている GCP プロジェクト
GCS バケットを作成および管理する権限
GCS バケットの IAM ポリシーを管理する権限
Cloud Run サービス、Pub/Sub トピック、Cloud Scheduler ジョブを作成する権限
URLScan IO テナントへの特権アクセス

URLScan IO の前提条件を取得する

URLScan IO にログインします。
プロフィールアイコンをクリックします。
メニューから [API キー] を選択します。
API キーをまだ取得していない場合:
1. [API キーを作成] ボタンをクリックします。
2. API キーの説明を入力します（例: Google SecOps Integration）。
3. [API キーを生成] をクリックします。
次の詳細をコピーして安全な場所に保存します。
- API_KEY: 生成された API キー文字列（形式: xxxxxxxx-xxxx-xxxx-xxxx-xxxxxxxxxxxx）
- API ベース URL: https://urlscan.io/api/v1（すべてのユーザーで同じ）
API の割り当て上限をメモします。
- 無料アカウントと Pro アカウントには、アクションごとに異なる分単位、時間単位、日単位の上限が適用されます。正確な上限については、個人の割り当てまたは API レート制限ヘッダーを確認してください。
- 詳しくは、URLScan IO API のレート制限に関するドキュメントをご覧ください。
検索対象を組織のスキャンのみに制限する必要がある場合は、次の点に注意してください。
- ユーザー識別子: ユーザー名またはメールアドレス（user: 検索フィルタで使用）
- チーム ID: チーム機能を使用している場合（team: 検索フィルタで使用）
注: user: や team: などの検索フィルタは、URLScan IO のドキュメントに記載されている検索構文の一部であり、ログインしていることや Pro 機能を利用していることが前提となる場合があります。高度なクエリの作成については、URLScan IO Search API ドキュメントをご覧ください。

API アクセスを確認する

統合に進む前に、API キーをテストします。

# Replace with your actual API key
API_KEY="your-api-key-here"

# Test API access
curl -v -H "API-Key: ${API_KEY}" "https://urlscan.io/api/v1/search/?q=date:>now-1h&size=1"

想定されるレスポンス: 検索結果を含む JSON を含む HTTP 200。

HTTP 401 または 403 が返された場合は、API キーが正しく、期限切れになっていないことを確認します。

Google Cloud Storage バケットを作成する

Google Cloud Console に移動します。
プロジェクトを選択するか、新しいプロジェクトを作成します。
ナビゲーションメニューで、[Cloud Storage > バケット] に移動します。
[バケットを作成] をクリックします。

次の構成情報を提供してください。

設定	値
バケットに名前を付ける	グローバルに一意の名前（`urlscan-logs-bucket` など）を入力します。
ロケーションタイプ	ニーズに基づいて選択します（リージョン、デュアルリージョン、マルチリージョン）。
ロケーション	ロケーションを選択します（例: `us-central1`）。
ストレージクラス	Standard（頻繁にアクセスされるログにおすすめ）
アクセス制御	均一（推奨）
保護ツール	省略可: オブジェクトのバージョニングまたは保持ポリシーを有効にする

[作成] をクリックします。

Cloud Run functions のサービスアカウントを作成する

Cloud Run 関数には、GCS バケットに書き込み、Pub/Sub によって呼び出される権限を持つサービスアカウントが必要です。

サービスアカウントの作成

GCP Console で、[IAM と管理>サービスアカウント] に移動します。
[サービスアカウントを作成] をクリックします。
次の構成の詳細を指定します。
- サービスアカウント名: 「urlscan-collector-sa」と入力します。
- サービスアカウントの説明: 「Service account for Cloud Run function to collect URLScan IO logs」と入力します。
[作成して続行] をクリックします。
[このサービスアカウントにプロジェクトへのアクセスを許可する] セクションで、次のロールを追加します。
1. [ロールを選択] をクリックします。
2. [ストレージオブジェクト管理者] を検索して選択します。
3. [+ 別のロールを追加] をクリックします。
4. [Cloud Run 起動元] を検索して選択します。
5. [+ 別のロールを追加] をクリックします。
6. [Cloud Functions 起動元] を検索して選択します。
[続行] をクリックします。
[完了] をクリックします。

これらのロールは、次の目的で必要です。

Storage オブジェクト管理者: ログを GCS バケットに書き込み、状態ファイルを管理する
Cloud Run 起動元: Pub/Sub が関数を呼び出すことを許可する
Cloud Functions 起動元: 関数の呼び出しを許可する

GCS バケットに対する IAM 権限を付与する

GCS バケットに対する書き込み権限をサービスアカウントに付与します。

[Cloud Storage] > [バケット] に移動します。
バケット名をクリックします。
[権限] タブに移動します。
[アクセス権を付与] をクリックします。
次の構成の詳細を指定します。
- プリンシパルを追加: サービスアカウントのメールアドレス（例: urlscan-collector-sa@PROJECT_ID.iam.gserviceaccount.com）を入力します。
- ロールを割り当てる: [Storage オブジェクト管理者] を選択します。
[保存] をクリックします。

Pub/Sub トピックの作成

Cloud Scheduler がパブリッシュし、Cloud Run functions がサブスクライブする Pub/Sub トピックを作成します。

GCP Console で、[Pub/Sub> トピック] に移動します。
[トピックを作成] をクリックします。
次の構成の詳細を指定します。
- トピック ID: 「urlscan-logs-trigger」と入力します。
- その他の設定はデフォルトのままにします。
[作成] をクリックします。

ログを収集する Cloud Run 関数を作成する

Cloud Run 関数は、Cloud Scheduler からの Pub/Sub メッセージによってトリガーされ、URLScan IO API からログを取得して GCS に書き込みます。

GCP Console で、[Cloud Run] に移動します。
[サービスを作成] をクリックします。
[関数] を選択します（インラインエディタを使用して関数を作成します）。

[構成] セクションで、次の構成の詳細を指定します。

設定	値
サービス名	`urlscan-collector`
リージョン	GCS バケットと一致するリージョンを選択します（例: `us-central1`）。
ランタイム	[Python 3.12] 以降を選択します。

[トリガー（省略可）] セクションで、次の操作を行います。
1. [+ トリガーを追加] をクリックします。
2. [Cloud Pub/Sub] を選択します。
3. [Cloud Pub/Sub トピックを選択してください] で、Pub/Sub トピック（urlscan-logs-trigger）を選択します。
4. [保存] をクリックします。
[認証] セクションで、次の操作を行います。
1. [認証が必要] を選択します。
2. Identity and Access Management（IAM）を確認します。
注: Pub/Sub は、関数を呼び出すときに認証を自動的に処理します。
下にスクロールして、[コンテナ、ネットワーキング、セキュリティ] を開きます。
[セキュリティ] タブに移動します。
- サービスアカウント: サービスアカウントを選択します（urlscan-collector-sa）。

[コンテナ] タブに移動します。

[変数とシークレット] をクリックします。
環境変数ごとに [+ 変数を追加] をクリックします。

変数名	値の例	説明
`GCS_BUCKET`	`urlscan-logs-bucket`	GCS バケット名
`GCS_PREFIX`	`urlscan/`	ログファイルの接頭辞
`STATE_KEY`	`urlscan/state.json`	状態ファイルのパス
`API_KEY`	`your-urlscan-api-key`	URLScan IO API キー
`API_BASE`	`https://urlscan.io/api/v1`	API ベース URL
`SEARCH_QUERY`	`date:>now-1h`	検索クエリフィルタ
`PAGE_SIZE`	`100`	1 ページあたりのレコード数
`MAX_PAGES`	`10`	取得する最大ページ数

[変数とシークレット] セクションで、[リクエスト] まで下にスクロールします。
- リクエストのタイムアウト: 600 秒（10 分）を入力します。
[設定] タブに移動します。
- [リソース] セクションで次の操作を行います。
  - メモリ: 512 MiB 以上を選択します。
  - CPU: [1] を選択します。
[リビジョンスケーリング] セクションで、次の操作を行います。
- [インスタンスの最小数] に「0」と入力します。
- インスタンスの最大数: 100 と入力します（または、予想される負荷に基づいて調整します）。
[作成] をクリックします。
サービスが作成されるまで待ちます（1 ～ 2 分）。
サービスを作成すると、インラインコードエディタが自動的に開きます。

関数コードを追加する

[関数のエントリポイント] に「main」と入力します。

インラインコードエディタで、次の 2 つのファイルを作成します。

最初のファイル: 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 with timeouts
http = urllib3.PoolManager(
    timeout=urllib3.Timeout(connect=5.0, read=30.0),
    retries=False,
)

# Initialize Storage client
storage_client = storage.Client()

# Environment variables
GCS_BUCKET = os.environ.get('GCS_BUCKET')
GCS_PREFIX = os.environ.get('GCS_PREFIX', 'urlscan/')
STATE_KEY = os.environ.get('STATE_KEY', 'urlscan/state.json')
API_KEY = os.environ.get('API_KEY')
API_BASE = os.environ.get('API_BASE', 'https://urlscan.io/api/v1')
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'))

def parse_datetime(value: str) -> datetime:
    """Parse ISO datetime string to datetime object."""
    if value.endswith("Z"):
        value = value[:-1] + "+00:00"
    return datetime.fromisoformat(value)

@functions_framework.cloud_event
def main(cloud_event):
    """
    Cloud Run function triggered by Pub/Sub to fetch URLScan IO results and write to GCS.

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

    if not all([GCS_BUCKET, API_KEY]):
        print('Error: Missing required environment variables')
        return

    try:
        # Get GCS bucket
        bucket = storage_client.bucket(GCS_BUCKET)

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

        # Adjust search query based on last run
        search_query = SEARCH_QUERY
        if last_run:
            try:
                search_time = parse_datetime(last_run)
                time_diff = datetime.now(timezone.utc) - search_time
                hours = int(time_diff.total_seconds() / 3600) + 1
                search_query = f'date:>now-{hours}h'
            except Exception as e:
                print(f'Warning: Could not parse last_run: {e}')

        print(f'Searching with query: {search_query}')

        # Fetch logs
        records, newest_event_time = fetch_logs(
            api_base=API_BASE,
            api_key=API_KEY,
            search_query=search_query,
            page_size=PAGE_SIZE,
            max_pages=MAX_PAGES,
        )

        if not records:
            print("No new log records found.")
            now = datetime.now(timezone.utc)
            save_state(bucket, STATE_KEY, now.isoformat())
            return

        # Write to GCS as NDJSON
        now = datetime.now(timezone.utc)
        file_key = f"{GCS_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"

        ndjson_content = '\n'.join([json.dumps(r, separators=(',', ':')) for r in records])

        blob = bucket.blob(file_key)
        blob.upload_from_string(
            ndjson_content,
            content_type='application/x-ndjson'
        )

        print(f"Uploaded {len(records)} results to gs://{GCS_BUCKET}/{file_key}")

        # Update state with newest event time
        if newest_event_time:
            save_state(bucket, STATE_KEY, newest_event_time)
        else:
            save_state(bucket, STATE_KEY, now.isoformat())

        print(f'Successfully processed {len(records)} scan results')

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

def load_state(bucket, key):
    """Load state from GCS."""
    try:
        blob = bucket.blob(key)
        if blob.exists():
            state_data = blob.download_as_text()
            return json.loads(state_data)
    except Exception as e:
        print(f'Warning: Could not load state: {str(e)}')
    return {}

def save_state(bucket, key, last_event_time_iso: str):
    """Save the last event timestamp to GCS state file."""
    try:
        state = {'last_run': last_event_time_iso}
        blob = bucket.blob(key)
        blob.upload_from_string(
            json.dumps(state, indent=2),
            content_type='application/json'
        )
        print(f"Saved state: last_run={last_event_time_iso}")
    except Exception as e:
        print(f'Warning: Could not save state: {str(e)}')

def fetch_logs(api_base: str, api_key: str, search_query: str, page_size: int, max_pages: int):
    """
    Fetch logs from URLScan IO API with pagination and rate limiting.

    Args:
        api_base: API base URL
        api_key: URLScan IO API key
        search_query: Search query string
        page_size: Number of records per page
        max_pages: Maximum total pages to fetch

    Returns:
        Tuple of (records list, newest_event_time ISO string)
    """

    headers = {
        'API-Key': api_key,
        'Accept': 'application/json',
        'User-Agent': 'GoogleSecOps-URLScanCollector/1.0'
    }

    all_results = []
    newest_time = None
    page_num = 0
    backoff = 1.0
    offset = 0

    while page_num < max_pages:
        page_num += 1

        # Build search URL with pagination
        search_url = f"{api_base}/search/"
        params = [
            f"q={search_query}",
            f"size={page_size}",
            f"offset={offset}"
        ]
        url = f"{search_url}?{'&'.join(params)}"

        try:
            response = http.request('GET', url, headers=headers)

            # Handle rate limiting with exponential backoff
            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

            backoff = 1.0

            if response.status != 200:
                print(f"Search failed: {response.status}")
                response_text = response.data.decode('utf-8')
                print(f"Response body: {response_text}")
                break

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

            if not results:
                print(f"No more results (empty page)")
                break

            print(f"Page {page_num}: Retrieved {len(results)} scan results")

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

                    try:
                        result_response = http.request('GET', result_url, headers=headers)

                        # Handle rate limiting
                        if result_response.status == 429:
                            retry_after = int(result_response.headers.get('Retry-After', '5'))
                            print(f"Rate limited on result fetch. Retrying after {retry_after}s...")
                            time.sleep(retry_after)
                            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)

                            # Track newest event time
                            try:
                                event_time = task.get('time')
                                if event_time:
                                    if newest_time is None or parse_datetime(event_time) > parse_datetime(newest_time):
                                        newest_time = event_time
                            except Exception as e:
                                print(f"Warning: Could not parse event time: {e}")
                        else:
                            print(f"Failed to fetch result for {uuid}: {result_response.status}")
                    except Exception as e:
                        print(f"Error fetching result for {uuid}: {e}")

            # Check if we have more pages
            total = search_data.get('total', 0)
            if offset + len(results) >= total or len(results) < page_size:
                print(f"Reached last page (offset={offset}, results={len(results)}, total={total})")
                break

            offset += len(results)

        except Exception as e:
            print(f"Error fetching logs: {e}")
            return [], None

    print(f"Retrieved {len(all_results)} total records from {page_num} pages")
    return all_results, newest_time

2 つ目のファイル: 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 functions の関数をトリガーします。

GCP Console で、[Cloud Scheduler] に移動します。
[ジョブを作成] をクリックします。

次の構成情報を提供してください。

設定	値
名前	`urlscan-collector-hourly`
リージョン	Cloud Run functions と同じリージョンを選択する
周波数	`0 * * * *`（1 時間ごとに正時）
タイムゾーン	タイムゾーンを選択します（UTC を推奨）。
ターゲットタイプ	Pub/Sub
トピック	Pub/Sub トピック（`urlscan-logs-trigger`）を選択する
メッセージ本文	`{}`（空の JSON オブジェクト）

[作成] をクリックします。

スケジュールの頻度のオプション

ログの量とレイテンシの要件に基づいて頻度を選択します。

頻度	CRON 式	ユースケース
5 分毎	`/5 * * *`	大容量、低レイテンシ
15 分ごと	`/15 * * *`	検索量が普通
1 時間ごと	`0 * * * *`	標準（推奨）
6 時間ごと	`0 /6 * *`	少量、バッチ処理
毎日	`0 0 * * *`	履歴データの収集

統合をテストする

Cloud Scheduler コンソールで、ジョブ（urlscan-collector-hourly）を見つけます。
[強制実行] をクリックして、ジョブを手動でトリガーします。
数秒待ちます。
Cloud Run > サービスに移動します。
関数名（urlscan-collector）をクリックします。
[Logs] タブをクリックします。

関数が正常に実行されたことを確認します。次の内容を確認します。

Searching with query: date:>now-1h
Page 1: Retrieved X scan results
Uploaded X results to gs://bucket-name/urlscan/year=YYYY/month=MM/day=DD/hour=HH/urlscan_YYYYMMDD_HHMMSS.json
Successfully processed X scan results

[Cloud Storage] > [バケット] に移動します。
バケット名をクリックします。
プレフィックスフォルダ（urlscan/）に移動します。
現在のタイムスタンプで新しい .json ファイルが作成されたことを確認します。

ログにエラーが表示された場合:

HTTP 401: 環境変数で API キーを確認する
HTTP 403: API キーの有効期限が切れていないことを確認する
HTTP 429: レート制限 - 関数はバックオフで自動的に再試行されます
環境変数が不足している: 必要な変数がすべて設定されていることを確認します
検索に失敗しました: 検索クエリの構文が正しいことを確認します

Google SecOps サービスアカウントを取得する

Google SecOps は、一意のサービスアカウントを使用して GCS バケットからデータを読み取ります。このサービスアカウントにバケットへのアクセス権を付与する必要があります。

サービスアカウントのメールアドレスを取得する

[SIEM 設定] > [フィード] に移動します。
[Add New Feed] をクリックします。
[単一フィードを設定] をクリックします。
[フィード名] フィールドに、フィードの名前を入力します（例: URLScan IO logs）。
[ソースタイプ] として [Google Cloud Storage V2] を選択します。
[ログタイプ] として [URLScan IO] を選択します。
[サービスアカウントを取得する] をクリックします。一意のサービスアカウントメールアドレスが表示されます（例:）。
```
chronicle-12345678@chronicle-gcp-prod.iam.gserviceaccount.com
```
このメールアドレスをコピーして、次のステップで使用します。

注: 各 Google SecOps インスタンスには一意のサービスアカウントがあります。他のドキュメントや例のサービスアカウントは使用しないでください。

Google SecOps サービスアカウントに IAM 権限を付与する

Google SecOps サービスアカウントには、GCS バケットに対する Storage オブジェクト閲覧者ロールが必要です。

[Cloud Storage] > [バケット] に移動します。
バケット名をクリックします。
[権限] タブに移動します。
[アクセス権を付与] をクリックします。
次の構成の詳細を指定します。
- プリンシパルを追加: Google SecOps サービスアカウントのメールアドレスを貼り付けます。
- ロールを割り当てる: [ストレージオブジェクト閲覧者] を選択します。
[保存] をクリックします。

注: 削除オプションの [転送されたファイルを削除] または [転送されたファイルと空のディレクトリを削除] を使用する場合は、Storage オブジェクト閲覧者ではなく、Storage オブジェクト管理者ロールを付与します。

URLScan IO のログを取り込むように Google SecOps でフィードを構成する

[SIEM 設定] > [フィード] に移動します。
[Add New Feed] をクリックします。
[単一フィードを設定] をクリックします。
[フィード名] フィールドに、フィードの名前を入力します（例: URLScan IO logs）。
[ソースタイプ] として [Google Cloud Storage V2] を選択します。
[ログタイプ] として [URLScan IO] を選択します。
[次へ] をクリックします。
次の入力パラメータの値を指定します。
- ストレージバケットの URL: 接頭辞パスを含む GCS バケット URI を入力します。
```
gs://urlscan-logs-bucket/urlscan/
```
  - 次のように置き換えます。
    - urlscan-logs-bucket: GCS バケット名。
    - urlscan/: ログが保存されるオプションの接頭辞/フォルダパス（ルートの場合は空のままにします）。
      
      例:
      - ルートバケット: gs://urlscan-logs-bucket/
      - 接頭辞あり: gs://urlscan-logs-bucket/urlscan/
  注: URI の末尾には常にスラッシュ（/）を付けます。
- Source deletion option: 必要に応じて削除オプションを選択します。
  - なし: 転送後にファイルを削除しません（テストにおすすめ）。
  - 転送されたファイルを削除する: 転送が完了した後にファイルを削除します。
  - 転送されたファイルと空のディレクトリを削除する: 転送が完了した後にファイルと空のディレクトリを削除します。
    
    注: 削除オプションを選択する場合は、サービスアカウントに Storage オブジェクト閲覧者ではなく、Storage オブジェクト管理者のロールが必要です。必要に応じて IAM 権限を更新します。
- ファイルの最大経過日数: 指定した日数以内に変更されたファイルを含めます。デフォルトは 180 日です。
- アセットの名前空間: アセットの名前空間。
- Ingestion labels: このフィードのイベントに適用されるラベル。
[次へ] をクリックします。
[Finalize] 画面で新しいフィードの設定を確認し、[送信] をクリックします。

ご不明な点がございましたら、コミュニティメンバーや Google SecOps のプロフェッショナルから回答を得ることができます。