收集 Qualys 资产上下文日志

支持：

Google SecOps SIEM

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

Qualys 资产上下文提供来自 Qualys 平台的主机资产清单数据（IP 地址、DNS 名称、操作系统、云提供商元数据和标签）。由于此数据是从 Qualys API 中提取的，因此 Cloud Run 函数会定期查询 Qualys 主机列表 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 账号（具有 Manager 角色的用户，或具有 API 访问权限且可以查看您要收集的资产的角色）

创建 Cloud Storage 存储桶

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

收集 Qualys API 凭据

登录 Qualys Web 应用。
确定或创建 API 用户：
- 前往用户 > 用户管理。
- 使用现有用户，或依次点击 新建 > 用户 并分配 Manager 角色（或具有 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 主机列表 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。

在容器 > 变量和 Secret 标签页上，添加以下环境变量：

变量名称	示例值
`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 * * *（每六小时一次）。
- 时区：选择UTC。
- 目标类型：选择 Pub/Sub。
- 主题：选择 qualys-asset-context-trigger。
- 消息正文：输入{}。
点击创建。
如需验证设置，请点击作业中的强制运行 ，然后检查 Cloud Run 日志和存储桶，以确认主机记录已写入。

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

前往 SIEM 设置 > Feed。
点击 Add New Feed （添加新 Feed）。
在下一页上，点击配置单个 Feed 。
在 Feed 名称 字段中，输入 Feed 的名称（例如 Qualys asset context logs）。
选择 Google Cloud Storage V2 作为来源类型 。
选择 Qualys 资产上下文 作为日志类型 。
点击 Chronicle 服务账号 字段旁边的获取服务账号 ，然后复制服务帐号电子邮件地址。
点击下一步 。
为以下输入参数指定值：
- 存储分区 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`	从更改日志映射

更新日志

查看此解析器的更新日志