收集 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 bucket，供 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 bucket

前往 Google Cloud 控制台。
選取專案或建立新專案。
依序前往「Cloud Storage」>「Buckets」。
按一下「建立值區」。
請提供下列設定詳細資料：
- 為 bucket 命名：輸入全域不重複的名稱 (例如 qualys-asset-context-logs)。
- 位置類型：根據需求選擇 (區域、雙區域或多區域)。
- 位置：選取位置 (例如 us-central1)。
- 儲存空間級別：選取「Standard」。
- 存取控管：選取「統一」。
點選「建立」。

收集 Qualys API 憑證

登入 Qualys 網頁應用程式。
找出或建立 API 使用者：
- 依序前往「Users」>「User Management」。
- 使用現有使用者，或依序點選「新增」> 使用者」，然後指派「管理員」角色 (或具有 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。
點按「Create and continue」(建立並繼續)。
授予下列角色：
- Storage 物件管理員：將主機記錄寫入值區。
- Cloud Run 叫用者：允許 Pub/Sub 叫用函式。
依序點選「繼續」和「完成」。

授予服務帳戶 Cloud Storage bucket 的存取權

依序前往「Cloud Storage」>「Buckets」。
按一下 bucket 名稱。
前往「權限」分頁標籤。
按一下「授予存取權」。
請提供下列設定詳細資料：
- 新增主體：輸入服務帳戶電子郵件地址 (qualys-asset-context-sa@<PROJECT_ID>.iam.gserviceaccount.com)。
- 指派角色：選取「Storage 物件管理員」。
按一下 [儲存]。

建立 Pub/Sub 主題

前往「Pub/Sub」>「主題」。
按一下「建立主題」。
在「Topic ID」(主題 ID) 欄位中輸入 qualys-asset-context-trigger。
點選「建立」。

建立 Cloud Run 函式來收集記錄

Cloud Run 函式是由 Cloud Scheduler 的 Pub/Sub 訊息觸發。這個函式會呼叫 Qualys FO 主機清單 API、將 XML 回應轉換為 JSON，並將主機記錄寫入 bucket。

前往 Cloud Run。
按一下「Create service」(建立服務)。
選取「函式」即可使用內嵌程式碼編輯器。
請提供下列設定詳細資料：
- 服務名稱：輸入 qualys-asset-context-collector。
- Region (區域)：選取與值區相符的區域 (例如 us-central1)。
- 執行階段：選取 Python 3.12 以上版本。
在「Trigger」(觸發條件) 區段：
1. 按一下 [Add trigger] (新增觸發條件)。
2. 選取「Cloud Pub/Sub」。
3. 在「選取 Cloud Pub/Sub 主題」中，選取 qualys-asset-context-trigger。
4. 按一下 [儲存]。
在「驗證」部分，選取「需要驗證」。
展開「Containers, Networking, Security」，然後在「Security」分頁中，將「服務帳戶」設為 qualys-asset-context-sa。

在「Containers」>「Variables & Secrets」分頁中，新增下列環境變數：

變數名稱	範例值
`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`

將「Request timeout」(要求逾時) 設為 600 秒，並將「Memory」(記憶體) 設為 512 MiB 以上。
點按「Create」(建立)。服務建立完成後，系統會開啟內嵌程式碼編輯器。

新增函式程式碼

在「函式進入點」欄位中輸入 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

按一下「Deploy」(部署)，然後等待部署作業完成。

建立 Cloud Scheduler 工作

前往 Cloud Scheduler。
按一下「Create job」(建立工作)。
請提供下列設定詳細資料：
- 「Name」(名稱)：輸入 qualys-asset-context-collector-daily。
- 區域：選取與 Cloud Run 函式相同的區域。
- 頻率：輸入 0 */6 * * * (每六小時)。
- 時區：選取「UTC」。
- 目標類型：選取「Pub/Sub」。
- 主題：選取 qualys-asset-context-trigger。
- 郵件內文：輸入 {}。
點選「建立」。
如要驗證設定，請按一下工作中的「強制執行」，然後檢查 Cloud Run 記錄和值區，確認主機記錄已寫入。

在 Google SecOps 中設定資訊提供，擷取 Qualys 資產內容記錄

依序前往「SIEM 設定」>「動態饋給」。
按一下「新增動態消息」。
在下一個頁面中，按一下「設定單一動態饋給」。
在「動態饋給名稱」欄位中輸入動態饋給名稱 (例如 Qualys asset context logs)。
選取「Google Cloud Storage V2」做為「來源類型」。
選取「Qualys 資產內容」做為「記錄類型」。
按一下「Chronicle 服務帳戶」欄位旁的「取得服務帳戶」，然後複製服務帳戶電子郵件地址。
點選「下一步」。
指定下列輸入參數的值：
- 儲存空間值區 URI：輸入 gs://qualys-asset-context-logs/qualys-asset-context/。這個網址的結尾必須是正斜線 (/)。
- 來源刪除選項：根據偏好設定選取刪除選項。
- 檔案存在時間上限：納入在過去天數內修改的檔案。預設值為 180 天。
點選「下一步」。
在「Finalize」(完成) 畫面中檢查新的動態饋給設定，然後按一下「Submit」(提交)。

授予 Google SecOps 服務帳戶值區存取權

依序前往「Cloud Storage」>「Buckets」。
按一下 bucket 名稱。
前往「權限」分頁標籤。
按一下「授予存取權」。
請提供下列設定詳細資料：
- 新增主體：貼上動態消息設定中的 Google SecOps 服務帳戶電子郵件地址。
- 指派角色：選取「Storage 物件檢視者」 (如果您選擇刪除選項，請選取「Storage 物件管理員」)。
按一下 [儲存]。

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`	從變更記錄對應

變更記錄

查看這個剖析器的變更記錄

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