跨帳號批量掃描應急漏洞 - Security Center

高危漏洞被公開披露後，管理多個雲帳號的企業需要快速評估所有帳號的受影響情況。本文介紹如何通過 Python 指令碼調用Security Center ModifyEmgVulSubmit API，結合資來源目錄（RD）自動擷取成員帳號列表，實現跨帳號批量掃描應急漏洞。

背景

假設某互連網企業在資來源目錄（RD）中管理著 150 個成員帳號。Linux 系統組件 polkit pkexec 被曝出本地提權漏洞（CVE-2021-4034），安全團隊需立即評估全集團的受影響狀況。

傳統營運模式面臨以下難題：

手動響應滯後：逐一登入 150 個帳號發起掃描，耗時數小時甚至數天，極易錯失漏洞收斂的最佳視窗期。
掃描覆蓋死角：依靠人工登記帳號列表，難以確保新加入 RD 的成員帳號被納入掃描範圍。
自動化設計門檻：簡單的並髮腳本若缺乏 API 節流機制，會因觸發雲端流控導致任務批量失敗。
風險視角孤島：掃描結果分散，安全主管無法在一個介面查看全集團的漏洞概覽及處置進度。

方案概述

本方案在本地運行 Python 指令碼，通過資來源目錄自動擷取所有成員帳號，調用Security Center API 批量下發應急漏洞掃描任務。掃描完成後，通過 Agentic SOC 的Log Service集中查看全部帳號的漏洞掃描結果。整體流程如下：

營運人員在本地通過專用 RAM 使用者的 AccessKey 運行 Python 指令碼。指令碼首先調用 DescribeEmgVulItem API 驗證配置中的漏洞名稱是否有效，然後調用資來源目錄 ListAccounts API 自動擷取所有成員帳號，最後逐一調用Security Center ModifyEmgVulSubmit API 下發掃描任務。掃描結果通過 Agentic SOC 匯聚到Log Service（SLS），營運人員在一個介面即可查看全部帳號的漏洞狀態。

實施步驟

步驟一：配置環境與許可權

準備環境：
- 一個具備管理員權限的阿里雲企業認證帳號。
- 開通資來源目錄。
- 資來源目錄成員均已開通Security Center（進階版及以上），並在Agentic SOC中完成日誌接入。詳情請參見接入配置。
- Python 3.6及以上的本地環境。
使用阿里雲帳號建立RAM使用者（例如VulnScanner），並為其授予以下許可權：
- AliyunYundunSASFullAccess（必需）：用於調用Security Center API 下發掃描任務。本文使用系統策略以快速驗證，生產環境建議建立自訂策略僅授予ModifyEmgVulSubmit 和 DescribeEmgVulItem 的調用許可權。
- AliyunResourceDirectoryReadOnlyAccess（推薦）：用於自動擷取資來源目錄下的所有成員帳號。若不授予此許可權，需要在設定檔中手動維護帳號列表。
啟用OpenAPI調用並妥善儲存AccessKey（在步驟三：執行掃描時使用）。
重要
嚴禁使用阿里雲主帳號的 AccessKey 執行指令碼。請務必使用上述專用 RAM 使用者的 AccessKey，並定期輪轉以降低泄露風險。

步驟二：下載指令碼並配置掃描參數

下載以下三個檔案至本地工作目錄：

scan_manager.py：主程式，負責遍曆帳號並調用掃描 API。
config.ini：設定檔，定義服務地址、掃描目標漏洞等參數。
requirements.txt：Python 依賴庫列表。

scan_manager.py

# -*- coding: utf-8 -*-
# This file is auto-generated, don't edit it. Thanks.
import os
import sys
import time
import configparser
from typing import List, Dict

from alibabacloud_credentials.client import Client as CredentialClient
from alibabacloud_sas20181203 import models as sas_20181203_models
from alibabacloud_sas20181203.client import Client as Sas20181203Client
from alibabacloud_tea_openapi import models as open_api_models
from alibabacloud_tea_util import models as util_models
from alibabacloud_tea_util.client import Client as UtilClient

# --- 資來源目錄用戶端的匯入 ---
from alibabacloud_resourcemanager20200331 import models as resourcemanager_20200331_models
from alibabacloud_resourcemanager20200331.client import Client as ResourceManager20200331Client

class Sample:
    def __init__(self):
        pass

    @staticmethod
    def create_client(endpoint: str) -> Sas20181203Client:
        credential = CredentialClient()
        config = open_api_models.Config(
            credential=credential
        )
        config.endpoint = endpoint
        return Sas20181203Client(config)

    @staticmethod
    def resolve_vul_names(client: Sas20181203Client, target_names: List[str]) -> Dict[str, str]:
        """通過 DescribeEmgVulItem API 將控制台漏洞名稱解析為 API 所需的漏洞名稱。

        將控制台應急漏洞頁面的漏洞名稱作為 VulName 參數傳給 DescribeEmgVulItem，
        從返回結果中提取 Name 欄位，用於後續調用 ModifyEmgVulSubmit。

        Args:
            client: Security Center用戶端。
            target_names: 設定檔中的漏洞名稱列表（控制台顯示名稱）。

        Returns:
            Dict，key 為 API 漏洞名稱（用於 ModifyEmgVulSubmit），value 為顯示名稱（用於日誌輸出）。
        """
        runtime = util_models.RuntimeOptions()

        resolved: Dict[str, str] = {}
        for target in target_names:
            try:
                request = sas_20181203_models.DescribeEmgVulItemRequest(
                    vul_name=target
                )
                response = client.describe_emg_vul_item_with_options(request, runtime)
                vul_items = response.body.grouped_vul_items or []

                if vul_items:
                    vul_name = vul_items[0].name
                    resolved[vul_name] = target
                    print(f"  ✅ '{target}' → {vul_name}")
                else:
                    print(f"  ⚠️ 未找到匹配的應急漏洞: '{target}'")
            except Exception as e:
                print(f"  ❌ 查詢漏洞 '{target}' 失敗: {e}")

        return resolved

    @staticmethod
    def main(args: List[str]) -> None:
        # --- 從 config.ini 檔案讀取配置 ---
        config = configparser.ConfigParser(delimiters=('=',))
        # 指定以 utf-8 編碼讀取，避免中文亂碼
        config.read('config.ini', encoding='utf-8')

        try:
            # 讀取掃描配置
            endpoint = config.get('scan_config', 'endpoint')
            scan_targets_str = config.get('scan_config', 'scan_targets')
            target_names = [name.strip() for name in scan_targets_str.split(',') if name.strip()]
            account_ids_str = config.get('scan_config', 'member_uids')
            account_ids = [uid.strip() for uid in account_ids_str.split(',') if uid.strip()]
            max_qps = config.getint('scan_config', 'max_qps')

        except (configparser.NoSectionError, configparser.NoOptionError) as e:
            print(f"❌ 設定檔 'config.ini' 讀取錯誤: {e}")
            print("請確保 config.ini 檔案存在且包含 [scan_config] 部分和所有必需的鍵。")
            sys.exit(1)

        # --- 資產自動探索邏輯 ---
        # 如果設定檔中的 member_uids 為空白，則嘗試從資來源目錄自動擷取
        if not account_ids:
            print("ℹ️ 'member_uids' 為空白，嘗試從資來源目錄自動探索成員帳號...")
            try:
                # 資源管理服務的通用 Endpoint
                rm_endpoint = 'resourcemanager.aliyuncs.com'

                # 建立資來源目錄用戶端
                credential = CredentialClient()
                rm_config = open_api_models.Config(credential=credential, endpoint=rm_endpoint)
                rm_client = ResourceManager20200331Client(rm_config)

                all_accounts = []
                page_number = 1
                while True:
                    list_accounts_request = resourcemanager_20200331_models.ListAccountsRequest(
                        page_number=page_number,
                        page_size=100  # 使用最大 PageSize 減少請求次數
                    )
                    # 調用 ListAccounts API，迴圈處理分頁
                    response = rm_client.list_accounts(list_accounts_request)
                    if response.body.accounts and response.body.accounts.account:
                        current_page_accounts = response.body.accounts.account
                        all_accounts.extend(current_page_accounts)
                        # 當返回的記錄數小於 PageSize 時，表示已是最後一頁
                        if len(current_page_accounts) < 100:
                            break
                        page_number += 1
                    else:
                        break  # 沒有更多帳號

                # 從返回結果中提取帳號ID
                account_ids = [acc.account_id for acc in all_accounts]

                if account_ids:
                    print(f"✅ 成功從資來源目錄發現 {len(account_ids)} 個成員帳號。")
                else:
                    print("⚠️ 未能從資來源目錄發現任何成員帳號。請檢查許可權或資來源目錄狀態。")
            except Exception as e:
                print(f"❌ 從資來源目錄自動探索帳號失敗: {e}")
                print("   請確保 RAM 使用者擁有 'AliyunResourceDirectoryReadOnlyAccess' 許可權，或在 config.ini 中手動設定 'member_uids'。")
                # 如果自動探索失敗且手動也未配置，則退出
                if not any(account_ids):
                    sys.exit(1)
        # --- 配置讀取結束 ---

        client = Sample.create_client(endpoint)
        runtime = util_models.RuntimeOptions()

        # --- 通過 API 解析漏洞名稱 ---
        print("正在查詢應急漏洞列表...")
        vul_map = Sample.resolve_vul_names(client, target_names)

        if not vul_map:
            print("⚠️ 沒有匹配到任何漏洞，程式退出。")
            return

        vul_names = list(vul_map.keys())
        print(f"✅ 成功匹配 {len(vul_names)} 個漏洞")

        total_tasks = len(vul_names) * len(account_ids)
        if total_tasks == 0:
            print("⚠️  沒有在 config.ini 中找到要執行的漏洞或帳號，程式退出。")
            return

        print(f"開始處理 {len(vul_names)} 個漏洞 × {len(account_ids)} 個帳號 = {total_tasks} 個掃描任務")
        print(f"QPS控製為 {max_qps}，預計完成時間: {total_tasks / max_qps:.1f} 秒")

        success_count = 0
        failure_count = 0
        start_time = time.time()
        last_request_time = 0
        MIN_INTERVAL = 1.0 / max_qps
        request_count = 0

        for name in vul_names:
            for account_id in account_ids:
                request_count += 1

                # QPS控制
                current_time = time.time()
                elapsed_since_last = current_time - last_request_time
                if elapsed_since_last < MIN_INTERVAL:
                    time.sleep(MIN_INTERVAL - elapsed_since_last)
                last_request_time = time.time()

                display_name = vul_map.get(name, name)

                try:
                    modify_emg_vul_submit_request = sas_20181203_models.ModifyEmgVulSubmitRequest(
                        name=name,
                        user_agreement='yes',
                        resource_directory_account_id=account_id
                    )

                    client.modify_emg_vul_submit_with_options(modify_emg_vul_submit_request, runtime)
                    success_count += 1
                    elapsed_time = time.time() - start_time
                    avg_qps = request_count / elapsed_time if elapsed_time > 0 else 0

                    print(f"✅ [{request_count}/{total_tasks}] 成功: 漏洞='{display_name}', 帳號ID={account_id} "
                          f"(當前平均QPS: {avg_qps:.1f})")

                except Exception as error:
                    failure_count += 1
                    print(f"❌ [{request_count}/{total_tasks}] 失敗: 漏洞='{display_name}', 帳號ID={account_id}")

                    error_message = getattr(error, 'message', str(error))
                    print(f"   錯誤資訊: {error_message}")
                    recommend_message = getattr(error, 'data', {}).get('Recommend')
                    if recommend_message:
                        print(f"   診斷建議: {recommend_message}")

                    error_str = str(error).lower()
                    if 'throttling' in error_str or 'qps' in error_str:
                        print("   ⚠️  檢測到API流控，自動增加等待時間...")
                        time.sleep(2)

        total_time = time.time() - start_time
        final_qps = request_count / total_time if total_time > 0 else 0

        print(f"\n{'=' * 50}")
        print(f"處理完成! 總耗時: {total_time:.1f} 秒")
        print(f"成功: {success_count}, 失敗: {failure_count}, 總計: {total_tasks}")
        print(f"實際平均QPS: {final_qps:.1f} (目標QPS: {max_qps})")
        print(f"{'=' * 50}")

if __name__ == '__main__':
    Sample.main(sys.argv[1:])

config.ini

[scan_config]
# Security Center的服務地址（Endpoint），必鬚根據資產所在地區手動修改。
# 例如，上海地區為 tds.cn-shanghai.aliyuncs.com
endpoint = tds.cn-shanghai.aliyuncs.com

# API請求速率（QPS）限制，預設為6次/秒。
max_qps = 6

# 成員帳號阿里雲帳號列表，用英文逗號分隔。
# 如果RAM使用者已授予資來源目錄讀取許可權，請將此項留空，指令碼會自動擷取。
member_uids =

# 要掃描的應急漏洞名稱，多個漏洞用英文逗號分隔。
# 名稱可從Security Center控制台的應急漏洞頁面直接複製。
# 指令碼啟動時會通過 DescribeEmgVulItem API 自動解析為 API 所需的漏洞名稱。
scan_targets = polkit pkexec 本地提權漏洞（CVE-2021-4034）

requirements.txt

alibabacloud_sas20181203
alibabacloud_resourcemanager20200331
alibabacloud_credentials

開啟config.ini，根據實際需求配置以下參數：
- endpoint：根據資產所在地區，修改為對應的Security Center服務存取點。指令碼每次運行針對單個地區，若資產分布在多個地區，需為每個地區分別配置並執行。
- scan_targets：填入需要掃描的應急漏洞名稱，多個漏洞用英文逗號分隔。名稱可從Security Center控制台的應急漏洞頁面直接複製。指令碼啟動時會通過 DescribeEmgVulItem API 自動解析為 API 所需的漏洞名稱。
- max_qps：API 請求速率上限，預設為 6 次/秒，適用於大多數情境。若帳號規模較大且需要更高速率，可在配額中心申請提升Security Center相關 API 的頻率配額。
- member_uids：待掃描的成員帳號 ID 列表。若 RAM 使用者已有資來源目錄讀取許可權，留空即可，指令碼會自動擷取所有成員帳號。

步驟三：執行掃描

配置身份憑證。將步驟一建立的 RAM 使用者 AccessKey 配置為環境變數。

export ALIBABA_CLOUD_ACCESS_KEY_ID="YOUR_ACCESSKEY_ID"
export ALIBABA_CLOUD_ACCESS_KEY_SECRET="YOUR_ACCESSKEY_SECRET"

安裝依賴。
```
pip3 install -r requirements.txt
```

執行掃描指令碼。

python3 scan_manager.py

指令碼即時列印進度。輸出樣本：

正在查詢應急漏洞列表...
  ✅ 'polkit pkexec 本地提權漏洞（CVE-2021-4034）' → emg_cve-2021-4034:EMG:AVD-2021-4034
✅ 成功匹配 1 個漏洞

ℹ️ 'member_uids' 為空白，嘗試從資來源目錄自動探索成員帳號...
✅ 成功從資來源目錄發現 150 個成員帳號。

開始處理 1 個漏洞 × 150 個帳號 = 150 個掃描任務
QPS控製為 6，預計完成時間: 25.0 秒

✅ [1/150] 成功: 漏洞='polkit pkexec 本地提權漏洞（CVE-2021-4034）', 帳號ID=198XXXXXXXXXXX13 (當前平均QPS: 5.8)
✅ [2/150] 成功: 漏洞='polkit pkexec 本地提權漏洞（CVE-2021-4034）', 帳號ID=142XXXXXXXXXXX84 (當前平均QPS: 5.9)
✅ [3/150] 成功: 漏洞='polkit pkexec 本地提權漏洞（CVE-2021-4034）', 帳號ID=188XXXXXXXXXXX97 (當前平均QPS: 5.7)
...
✅ [150/150] 成功: 漏洞='polkit pkexec 本地提權漏洞（CVE-2021-4034）', 帳號ID=156XXXXXXXXXXX62 (當前平均QPS: 5.9)

==================================================
處理完成! 總耗時: 25.8 秒
成功: 150, 失敗: 0, 總計: 150
實際平均QPS: 5.8 (目標QPS: 6)
==================================================

步驟四：查看掃描結果

使用 RAM 帳號登入Security Center控制台，在左側導覽列選擇 Agentic SOC > 日誌。在日誌庫列表中，選擇標準化日誌 > 漏洞日誌，選擇掃描時間範圍後，單擊查詢/分析，查看漏洞掃描結果。

查詢結果將展示漏洞活動紀錄，日誌表名為 vulnerability_activity。每條記錄包含 vul_name（漏洞名稱，如 polkit pkexec 本地提權漏洞 CVE-2021-4034）和 instance_id（受影響執行個體 ID）等欄位。

重要

應急漏洞僅支援掃描，不支援修複。應急漏洞是針對您伺服器上安裝的軟體進行掃描後發現的風險，需要您根據漏洞詳情中的修複建議，手動對軟體應用升級或進行配置修改，排除安全隱患。￹

生產環境實踐建議

建立常態化掃描機制

應急漏洞的披露不可預測，建議將指令碼納入定期執行計畫，而非僅在漏洞爆發時臨時運行。可通過以下方式實現自動化：

使用 Linux crontab 或 Windows 工作排程器定時執行指令碼（例如每天一次），持續監控新披露的應急漏洞。
結合阿里雲Function Compute（FC）部署為 Serverless 函數，通過定時觸發器調度，無需維護常駐伺服器。

強化 AccessKey 安全管理

定期輪轉：每 90 天輪換一次 RAM 使用者的 AccessKey，降低密鑰泄露風險。
使用執行個體角色：若指令碼運行在 ECS 執行個體上，推薦通過執行個體RAM角色授權，避免在環境變數中暴露 AccessKey。
限制來源 IP：在 RAM 策略中添加 IP 條件，僅允許從營運專用網段調用 API。

常見問題

掃描提交成功，但在 Agentic SOC 的漏洞日誌中看不到記錄

日誌採集未開啟：在 Agentic SOC 的日誌中，確認漏洞日誌的採集開關已開啟。
掃描尚未完成：ModifyEmgVulSubmit 僅提交掃描任務，實際掃描是非同步執行的。根據資產規模，掃描可能需要數分鐘到數十分鐘才能完成。
未檢測到漏洞：漏洞日誌僅記錄實際檢測到的漏洞。如果成員帳號的資產上不存在該應急漏洞，掃描完成後不會產生漏洞日誌記錄。可先登入成員帳號，在漏洞管理 > 應急漏洞頁面確認單個帳號是否有掃描結果。
多帳號日誌未匯聚：成員帳號的日誌需通過 Agentic SOC 匯聚到管理帳號。確認成員帳號已開通Security Center並完成多帳號統一接入配置。

漏洞名稱解析失敗，提示"未找到匹配的應急漏洞"

請確認 config.ini 中 scan_targets 填寫的名稱與Security Center控制台應急漏洞頁面顯示的名稱完全一致（包括括弧和空格）。建議直接從控制台複製漏洞名稱，避免手動輸入導致的字元差異。

如果確認名稱無誤仍然解析失敗，可能該漏洞已過應急響應周期，不再支援通過 DescribeEmgVulItem API 查詢和主動掃描。