收集 URLScan IO 日志
支持的平台：
Google SecOps
SIEM
本文档介绍了如何使用 Google Cloud Storage 将 URLScan IO 日志注入到 Google Security Operations。URLScan IO 是一项分析网站的服务，可提供有关网站行为、安全性和性能的详细信息。它会扫描网址并生成包含屏幕截图、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 密钥，请执行以下操作：
- 点击创建 API 密钥按钮。
- 输入 API 密钥的说明（例如
Google SecOps Integration）。 - 点击 Generate API Key。
- 复制以下详细信息并将其保存在安全的位置：
- API_KEY：生成的 API 密钥字符串（格式：
xxxxxxxx-xxxx-xxxx-xxxx-xxxxxxxxxxxx） - API 基础网址：
https://urlscan.io/api/v1（所有用户都一样）
- API_KEY：生成的 API 密钥字符串（格式：
- 请注意您的 API 配额限制：
- 免费账号和专业版账号会受到每分钟、每小时和每天的限制，具体限制因操作而异。查看您的个人配额或 API 速率限制标头，了解确切的限制。
- 如需了解详情，请参阅 URLScan IO API 速率限制文档。
如果您需要将搜索范围限制为仅限组织扫描，请记下以下内容：
- 用户标识符：您的用户名或电子邮件地址（用于
user:搜索过滤条件） - 团队标识符：如果使用团队功能（与
team:搜索过滤条件搭配使用）
- 用户标识符：您的用户名或电子邮件地址（用于
验证 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"
预期响应：HTTP 200，包含搜索结果的 JSON。
如果您收到 HTTP 401 或 403 错误，请验证您的 API 密钥是否正确且未过期。
创建 Google Cloud Storage 存储桶
- 前往 Google Cloud 控制台。
- 选择您的项目或创建一个新项目。
- 在导航菜单中，依次前往 Cloud Storage > 存储桶。
- 点击创建存储分区。
提供以下配置详细信息：
设置 值 为存储桶命名 输入一个全局唯一的名称（例如 urlscan-logs-bucket）位置类型 根据您的需求进行选择（区域级、双区域级、多区域级） 位置 选择位置（例如 us-central1）存储类别 标准（建议用于频繁访问的日志） 访问权限控制 均匀（推荐） 保护工具 可选：启用对象版本控制或保留政策 点击创建。
为 Cloud Run 函数创建服务账号
Cloud Run 函数需要一个服务账号，该账号具有写入 GCS 存储桶和被 Pub/Sub 调用的权限。
创建服务账号
- 在 GCP 控制台中，依次前往 IAM 和管理 > 服务账号。
- 点击创建服务账号。
- 提供以下配置详细信息：
- 服务账号名称：输入
urlscan-collector-sa。 - 服务账号说明：输入
Service account for Cloud Run function to collect URLScan IO logs。
- 服务账号名称：输入
- 点击创建并继续。
- 在向此服务账号授予对项目的访问权限部分，添加以下角色：
- 点击选择角色。
- 搜索并选择 Storage Object Admin。
- 点击 + 添加其他角色。
- 搜索并选择 Cloud Run Invoker。
- 点击 + 添加其他角色。
- 搜索并选择 Cloud Functions Invoker。
- 点击继续。
- 点击完成。
以下操作需要这些角色：
- Storage Object Admin：将日志写入 GCS 存储桶并管理状态文件
- Cloud Run Invoker：允许 Pub/Sub 调用函数
- Cloud Functions Invoker：允许调用函数
授予对 GCS 存储桶的 IAM 权限
向服务账号授予对 GCS 存储桶的写入权限：
- 前往 Cloud Storage > 存储桶。
- 点击您的存储桶名称。
- 前往权限标签页。
- 点击授予访问权限。
- 提供以下配置详细信息：
- 添加主账号：输入服务账号电子邮件地址（例如
urlscan-collector-sa@PROJECT_ID.iam.gserviceaccount.com）。 - 分配角色：选择 Storage Object Admin。
- 添加主账号：输入服务账号电子邮件地址（例如
- 点击保存。
创建 Pub/Sub 主题
创建一个 Pub/Sub 主题，供 Cloud Scheduler 发布消息，并供 Cloud Run 函数订阅。
- 在 GCP 控制台中，前往 Pub/Sub > 主题。
- 点击创建主题。
- 提供以下配置详细信息：
- 主题 ID：输入
urlscan-logs-trigger。 - 将其他设置保留为默认值。
- 主题 ID：输入
- 点击创建。
创建 Cloud Run 函数以收集日志
Cloud Run 函数由来自 Cloud Scheduler 的 Pub/Sub 消息触发，用于从 URLScan IO API 中提取日志并将其写入 GCS。
- 在 GCP 控制台中，前往 Cloud Run。
- 点击创建服务。
- 选择函数（使用内嵌编辑器创建函数）。
在配置部分中，提供以下配置详细信息：
设置 值 Service 名称 urlscan-collector区域 选择与您的 GCS 存储桶匹配的区域（例如 us-central1）运行时 选择 Python 3.12 或更高版本 在触发器（可选）部分中：
- 点击 + 添加触发器。
- 选择 Cloud Pub/Sub。
- 在选择 Cloud Pub/Sub 主题中，选择 Pub/Sub 主题 (
urlscan-logs-trigger)。 - 点击保存。
在身份验证部分中：
- 选择需要进行身份验证。
- 检查 Identity and Access Management (IAM)。
向下滚动并展开容器、网络、安全性。
前往安全标签页：
- 服务账号：选择服务账号 (
urlscan-collector-sa)。
- 服务账号：选择服务账号 (
前往容器标签页：
- 点击变量和密钥。
- 为每个环境变量点击 + 添加变量：
变量名称 示例值 说明 GCS_BUCKETurlscan-logs-bucketGCS 存储桶名称 GCS_PREFIXurlscan/日志文件的前缀 STATE_KEYurlscan/state.json状态文件路径 API_KEYyour-urlscan-api-keyURLScan IO API 密钥 API_BASEhttps://urlscan.io/api/v1API 基本网址 SEARCH_QUERYdate:>now-1h“搜索查询”过滤条件 PAGE_SIZE100每页记录数 MAX_PAGES10要提取的最大网页数 在变量和密钥部分中，向下滚动到请求：
- 请求超时：输入
600秒（10 分钟）。
- 请求超时：输入
前往设置标签页：
- 在资源部分中：
- 内存：选择 512 MiB 或更高值。
- CPU：选择 1。
- 在资源部分中：
在修订版本伸缩部分中：
- 实例数下限：输入
0。 - 实例数上限：输入
100（或根据预期负载进行调整）。
- 实例数下限：输入
点击创建。
等待服务创建完成（1-2 分钟）。
创建服务后，系统会自动打开内嵌代码编辑器。
添加函数代码
- 在函数入口点中输入 main
在内嵌代码编辑器中，创建两个文件：
- 第一个文件：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- 第二个文件：requirements.txt:：
functions-framework==3.* google-cloud-storage==2.* urllib3>=2.0.0点击部署以保存并部署函数。
等待部署完成（2-3 分钟）。
创建 Cloud Scheduler 作业
Cloud Scheduler 会定期向 Pub/Sub 主题发布消息，从而触发 Cloud Run function。
- 在 GCP 控制台中，前往 Cloud Scheduler。
- 点击创建作业。
提供以下配置详细信息：
设置 值 名称 urlscan-collector-hourly区域 选择与 Cloud Run 函数相同的区域 频率 0 * * * *（每小时整点）时区 选择时区（建议选择世界协调时间 [UTC]） 目标类型 Pub/Sub 主题 选择 Pub/Sub 主题 ( urlscan-logs-trigger)消息正文 {}（空 JSON 对象）点击创建。
时间表频率选项
根据日志量和延迟时间要求选择频次：
频率 Cron 表达式 使用场景 每隔 5 分钟 */5 * * * *大容量、低延迟 每隔 15 分钟 */15 * * * *搜索量中等 每小时 0 * * * *标准（推荐） 每 6 小时 0 */6 * * *低量批处理 每天 0 0 * * *历史数据收集
测试集成
- 在 Cloud Scheduler 控制台中，找到您的作业 (
urlscan-collector-hourly)。 - 点击强制运行以手动触发作业。
- 等待几秒钟。
- 前往 Cloud Run > 服务。
- 点击函数名称 (
urlscan-collector)。 - 点击日志标签页。
验证函数是否已成功执行。查找以下项：
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 设置 > Feed。
- 点击添加新 Feed。
- 点击配置单个 Feed。
- 在 Feed 名称字段中，输入 Feed 的名称（例如
URLScan IO logs）。 - 选择 Google Cloud Storage V2 作为来源类型。
- 选择 URLScan IO 作为日志类型。
点击获取服务账号。系统会显示一个唯一的服务账号电子邮件地址，例如：
chronicle-12345678@chronicle-gcp-prod.iam.gserviceaccount.com复制此电子邮件地址，以便在下一步中使用。
向 Google SecOps 服务账号授予 IAM 权限
Google SecOps 服务账号需要您的 GCS 存储桶的 Storage Object Viewer 角色。
- 前往 Cloud Storage > 存储桶。
- 点击您的存储桶名称。
- 前往权限标签页。
- 点击授予访问权限。
- 提供以下配置详细信息：
- 添加主账号：粘贴 Google SecOps 服务账号电子邮件地址。
- 分配角色：选择 Storage Object Viewer。
点击保存。
在 Google SecOps 中配置 Feed 以注入 URLScan IO 日志
- 前往 SIEM 设置 > Feed。
- 点击添加新 Feed。
- 点击配置单个 Feed。
- 在 Feed 名称字段中，输入 Feed 的名称（例如
URLScan IO logs）。 - 选择 Google Cloud Storage V2 作为来源类型。
- 选择 URLScan IO 作为日志类型。
- 点击下一步。
为以下输入参数指定值：
存储桶网址：输入带有前缀路径的 GCS 存储桶 URI：
gs://urlscan-logs-bucket/urlscan/将
urlscan-logs-bucket：您的 GCS 存储桶名称。urlscan/：存储日志的可选前缀/文件夹路径（留空表示根目录）。示例：
- 根存储桶：
gs://urlscan-logs-bucket/ - 带前缀：
gs://urlscan-logs-bucket/urlscan/
- 根存储桶：
来源删除选项：根据您的偏好选择删除选项：
- 永不：转移后永不删除任何文件（建议用于测试）。
- 删除已转移的文件：成功转移后删除文件。
删除已转移的文件和空目录：成功转移后删除文件和空目录。
文件存在时间上限：包含在过去指定天数内修改过的文件。默认值为 180 天。
资源命名空间：资源命名空间。
注入标签：要应用于此 Feed 中事件的标签。
点击下一步。
在最终确定界面中查看新的 Feed 配置，然后点击提交。
UDM 映射表
| 日志字段 | UDM 映射 | 逻辑 |
|---|---|---|
_additional |
additional |
已重命名/已映射 |
has_principal |
metadata.event_type |
已映射：true → NETWORK_CONNECTION、true → STATUS_UPDATE |
_network |
network |
已重命名/已映射 |
_principal |
principal |
已重命名/已映射 |
_target |
target |
已重命名/已映射 |
| 不适用 | metadata.event_type |
常量：GENERIC_EVENT |
| 不适用 | metadata.product_name |
常量：URLSCAN.IO |
| 不适用 | metadata.vendor_name |
常量：URLSCAN.IO |
更新日志
需要更多帮助？获得社区成员和 Google SecOps 专业人士的解答。