Google 会使用 AI 技术将内容翻译成您偏好的语言。AI 翻译可能包含错误。

收集 Qualys 资源上下文日志

支持的平台：

Google SecOps SIEM

本文档介绍了如何使用 Google Cloud Storage 将 Qualys 资产上下文日志注入到 Google Security Operations。

Qualys 资产上下文提供来自 Qualys 平台的主机资产清单数据（IP 地址、DNS 名称、操作系统、云提供商元数据和标记）。由于此数据是从 Qualys API 中提取的，因此 Cloud Run 函数会定期查询 Qualys Host List API，将 XML 响应转换为 JSON，并将其写入 Cloud Storage 存储桶，供 Google SecOps 提取。

准备工作

请确保满足以下前提条件：

Google SecOps 实例
启用了 Cloud Storage、Cloud Run、Pub/Sub 和 Cloud Scheduler API 的 Google Cloud 项目
创建和管理 Cloud Storage 存储桶、服务账号、Cloud Run 函数、Pub/Sub 主题和 Cloud Scheduler 作业的权限
具有 API 访问权限的 Qualys 账号（具有管理员角色的用户，或具有 API 访问权限且可查看您要收集的资产的角色）

创建 Cloud Storage 存储桶

前往 Google Cloud 控制台。
选择您的项目或创建一个新项目。
前往 Cloud Storage > 存储桶。
点击创建存储分区。
提供以下配置详细信息：
- 指定存储桶的名称：输入一个全局唯一的名称（例如 qualys-asset-context-logs）。
- 位置类型：根据您的需求进行选择（区域、双区域或多区域）。
- 位置：选择位置（例如 us-central1）。
- 存储类别：选择 Standard。
- 访问权限控制：选择统一。
点击创建。

收集 Qualys API 凭据

登录 Qualys Web 应用。
确定或创建 API 用户：
- 依次前往用户 > 用户管理。
- 使用现有用户，或依次点击新建 > 用户，然后分配管理员角色（或具有 API 访问权限的角色）。
记录用户的用户名和密码。
确定您的 Qualys API 服务器网址（该网址特定于您的 Qualys 平台，例如 https://qualysapi.qg3.apps.qualys.com）。如需查找该网址，请参阅确定您的 Qualys 平台。

注意： Qualys FO API 要求每个请求都包含 X-Requested-With 标头。Cloud Run 函数会自动设置此标头。

为 Cloud Run 函数创建服务账号

前往 IAM 和管理 > 服务账号。
点击创建服务账号。
提供以下配置详细信息：
- 服务账号名称：输入 qualys-asset-context-sa。
- 服务账号说明：输入 Service account for Cloud Run function to collect Qualys asset context logs。
点击创建并继续。
授予以下角色：
- Storage Object Admin：将主机记录写入存储桶。
- Cloud Run Invoker：允许 Pub/Sub 调用该函数。
点击继续，然后点击完成。

向服务账号授予对 Cloud Storage 存储桶的访问权限

前往 Cloud Storage > 存储桶。
点击您的存储桶名称。
前往权限标签页。
点击授予访问权限。
提供以下配置详细信息：
- 添加主账号：输入服务账号电子邮件地址 (qualys-asset-context-sa@<PROJECT_ID>.iam.gserviceaccount.com)。
- 分配角色：选择 Storage Object Admin。
点击保存。

创建 Pub/Sub 主题

依次前往 Pub/Sub > 主题。
点击创建主题。
在主题 ID 字段中，输入 qualys-asset-context-trigger。
点击创建。

创建 Cloud Run 函数以收集日志

Cloud Run 函数由 Cloud Scheduler 中的 Pub/Sub 消息触发。它会调用 Qualys FO Host List API，将 XML 响应转换为 JSON，并将主机记录写入存储桶。

前往 Cloud Run。
点击创建服务。
选择函数以使用内嵌代码编辑器。
提供以下配置详细信息：
- 服务名称：输入 qualys-asset-context-collector。
- 区域：选择与您的存储桶匹配的区域（例如 us-central1）。
- 运行时：选择 Python 3.12 或更高版本。
在触发器部分中：
1. 点击添加触发器。
2. 选择 Cloud Pub/Sub。
3. 在选择 Cloud Pub/Sub 主题中，选择 qualys-asset-context-trigger。
4. 点击保存。
在身份验证部分中，选择需要进行身份验证。
展开容器、网络、安全性，然后在安全性标签页上将服务账号设置为 qualys-asset-context-sa。

在容器 > 变量和密钥标签页中，添加以下环境变量：

变量名称	示例值
`GCS_BUCKET`	`qualys-asset-context-logs`
`GCS_PREFIX`	`qualys-asset-context/`
`QUALYS_BASE_URL`	`https://qualysapi.qg3.apps.qualys.com`
`QUALYS_USERNAME`	`your-qualys-username`
`QUALYS_PASSWORD`	`your-qualys-password`
`TRUNCATION_LIMIT`	`1000`
`HTTP_TIMEOUT`	`300`
`MAX_PAGES`	`50`

将请求超时设置为 600 秒，并将内存设置为 512 MiB 或更高值。
点击创建。服务创建完成后，系统会打开内嵌式代码编辑器。

添加函数代码

在函数入口点字段中，输入 main。

在内嵌代码编辑器中，将 main.py 的内容替换为以下代码：

import functions_framework
from google.cloud import storage
import os
import time
import uuid
import json
import base64
import requests
import xmltodict

storage_client = storage.Client()

GCS_BUCKET = os.environ.get('GCS_BUCKET')
GCS_PREFIX = os.environ.get('GCS_PREFIX', 'qualys-asset-context/')
QUALYS_BASE_URL = os.environ.get('QUALYS_BASE_URL')
QUALYS_USERNAME = os.environ.get('QUALYS_USERNAME')
QUALYS_PASSWORD = os.environ.get('QUALYS_PASSWORD')
TRUNCATION_LIMIT = int(os.environ.get('TRUNCATION_LIMIT', '1000'))
HTTP_TIMEOUT = int(os.environ.get('HTTP_TIMEOUT', '300'))
MAX_PAGES = int(os.environ.get('MAX_PAGES', '50'))

def _headers():
    """Qualys FO API requires Basic auth and the X-Requested-With header."""
    token = base64.b64encode(f'{QUALYS_USERNAME}:{QUALYS_PASSWORD}'.encode()).decode()
    return {'Authorization': f'Basic {token}', 'X-Requested-With': 'Cloud Run function'}

def _extract_hosts(xml_text):
    """Parse the FO Host List XML into (list_of_host_dicts, next_page_url)."""
    doc = xmltodict.parse(xml_text)
    response = (doc.get('HOST_LIST_OUTPUT') or {}).get('RESPONSE') or {}
    host_list = (response.get('HOST_LIST') or {}).get('HOST')
    if host_list is None:
        hosts = []
    elif isinstance(host_list, list):
        hosts = host_list
    else:
        hosts = [host_list]
    # When the result is truncated, RESPONSE contains <WARNING><URL>...</URL></WARNING>.
    warning = response.get('WARNING')
    next_url = warning.get('URL') if isinstance(warning, dict) else None
    return hosts, next_url

def _write_hosts(bucket, hosts, page_num):
    """Write one batch of hosts to GCS as JSON, one host per line."""
    ts_path = time.strftime('%Y/%m/%d', time.gmtime())
    uniq = f'{int(time.time() * 1e6)}_{uuid.uuid4().hex[:8]}'
    key = f'{GCS_PREFIX}{ts_path}/qualys_asset_context_p{page_num:03d}_{uniq}.json'
    lines = [json.dumps(host, separators=(',', ':')) for host in hosts]
    bucket.blob(key).upload_from_string('\n'.join(lines), content_type='application/x-ndjson')
    return key

@functions_framework.cloud_event
def main(cloud_event):
    """Triggered by Pub/Sub to pull the Qualys host asset inventory and write it to GCS."""
    if not all([GCS_BUCKET, QUALYS_BASE_URL, QUALYS_USERNAME, QUALYS_PASSWORD]):
        print('Error: missing required environment variables')
        return

    bucket = storage_client.bucket(GCS_BUCKET)
    session = requests.Session()
    session.headers.update(_headers())

    url = f"{QUALYS_BASE_URL.rstrip('/')}/api/2.0/fo/asset/host/"
    params = {
        'action': 'list',
        'details': 'All',
        'truncation_limit': TRUNCATION_LIMIT,
        # show_cloud_tags, host_metadata, and show_tags add the cloud provider
        # tags, cloud instance metadata, and asset tags that the parser maps.
        'show_cloud_tags': 1,
        'host_metadata': 'all',
        'show_tags': 1,
    }

    pages = 0
    total = 0
    try:
        while pages < MAX_PAGES:
            resp = session.get(url, params=params, timeout=HTTP_TIMEOUT)
            resp.raise_for_status()
            hosts, next_url = _extract_hosts(resp.text)
            if hosts:
                _write_hosts(bucket, hosts, pages + 1)
                total += len(hosts)
            pages += 1
            if not next_url:
                break
            # The pagination URL already includes every query parameter.
            url, params = next_url, None

        print(f'Wrote {total} hosts across {pages} pages')
    except Exception as e:
        print(f'Error collecting Qualys asset context: {e}')
        raise

创建一个包含以下内容的 requirements.txt 文件：

functions-framework==3.*
google-cloud-storage==2.*
requests>=2.31.0
xmltodict>=0.13.0

点击部署，然后等待部署完成。

创建 Cloud Scheduler 作业

前往 Cloud Scheduler。
点击创建作业。
提供以下配置详细信息：
- 名称：输入 qualys-asset-context-collector-daily。
- 区域：选择与 Cloud Run 函数相同的区域。
- 频次：输入 0 */6 * * *（每 6 小时一次）。
- 时区：选择 UTC。
- 目标类型：选择 Pub/Sub。
- 主题：选择 qualys-asset-context-trigger。
- 消息正文：输入 {}。
点击创建。
如需验证设置，请点击作业中的强制运行，然后检查 Cloud Run 日志和存储桶，以确认主机记录已写入。

在 Google SecOps 中配置 Feed 以注入 Qualys 资产上下文日志

前往 SIEM 设置 > Feed。
点击添加新 Feed。
在下一页上，点击配置单个 Feed。
在 Feed 名称字段中，输入 Feed 的名称（例如 Qualys asset context logs）。
选择 Google Cloud Storage V2 作为来源类型。
选择 Qualys 资产上下文作为日志类型。
点击编年史服务账号字段旁边的获取服务账号，然后复制服务账号电子邮件地址。
点击下一步。
为以下输入参数指定值：
- 存储桶 URI：输入 gs://qualys-asset-context-logs/qualys-asset-context/。此网址必须以尾随正斜杠 (/) 结尾。
- 源删除选项：根据您的偏好选择删除选项。
- 文件存在时间上限：包含在过去指定天数内修改过的文件。默认值为 180 天。
点击下一步。
在最终确定界面中查看新的 Feed 配置，然后点击提交。

向 Google SecOps 服务账号授予对相应存储桶的访问权限

前往 Cloud Storage > 存储桶。
点击您的存储桶名称。
前往权限标签页。
点击授予访问权限。
提供以下配置详细信息：
- 添加主账号：粘贴从 Feed 设置中复制的 Google SecOps 服务账号电子邮件地址。
- 分配角色：选择 Storage Object Viewer（如果您选择了删除选项，则选择 Storage Object Admin）。
点击保存。

UDM 映射表

日志字段	UDM 映射	逻辑
`ASSET_ID`	`entity.asset.asset_id`	直接映射
`DNS_DATA.HOSTNAME`	`entity.asset.hostname`	直接映射
`IP`	`entity.asset.ip`	已合并
`OS`	`entity.asset.platform_software.platform`	已映射：`(?i)windows` → `WINDOWS`
`ID`	`entity.asset.product_object_id`	直接映射
`dataset`	`relations`	已合并
不适用	`entity.asset.platform_software.platform`	常量：`WINDOWS`
不适用	`metadata.entity_type`	常量：`ASSET`
不适用	`metadata.product_name`	常量：`QUALYS ASSET CONTEXT`
不适用	`metadata.vendor_name`	常量：`QUALYS ASSET CONTEXT`
`DNS_DATA.HOSTNAME`	`entity.entity.asset.hostname`	从更新日志映射

更新日志

查看相应解析器的更新日志