Security Center:Otomatisasi pemindaian kerentanan darurat lintas akun

Langkah 1: Siapkan lingkungan dan izin

1. Verifikasi prasyarat berikut:

   • Akun Alibaba Cloud yang telah diverifikasi sebagai perusahaan dengan hak istimewa administrator.

   • Aktifkan Resource Directory.

   • Cloud Security Center (edisi Advanced atau lebih tinggi) diaktifkan di semua akun anggota RD, dengan konfigurasi ingest log Agentic SOC (Konfigurasi akses).

   • Python 3.6 atau versi lebih baru telah diinstal di mesin lokal.

2. Buat Pengguna RAM (misalnya, VulnScanner) dan berikan izin berikut:

   • AliyunYundunSASFullAccess (Wajib): Memberikan akses ke API pemindaian Cloud Security Center. Topik ini menggunakan kebijakan sistem untuk validasi cepat. Di lingkungan produksi, buat kebijakan kustom yang hanya memberikan izin ModifyEmgVulSubmit dan DescribeEmgVulItem.

   • AliyunResourceDirectoryReadOnlyAccess (Disarankan): Memungkinkan penemuan otomatis semua akun anggota RD. Tanpa izin ini, Anda harus mencantumkan ID akun secara manual dalam file konfigurasi.

3. Aktifkan akses OpenAPI dan simpan AccessKey. Anda memerlukannya di Langkah 3: Jalankan pemindaian.

   Penting

   Jangan pernah menggunakan AccessKey Akun Alibaba Cloud Anda (akun root) untuk menjalankan skrip. Selalu gunakan AccessKey Pengguna RAM khusus yang dibuat di atas, dan putar secara berkala untuk meminimalkan risiko kebocoran kredensial.

Langkah 2: Unduh skrip dan konfigurasikan parameter pemindaian

1. Unduh tiga file berikut ke direktori kerja Anda:

   • scan_manager.py: Skrip utama. Melakukan iterasi melalui akun dan memanggil API pemindaian untuk masing-masing.

   • config.ini: File konfigurasi. Menentukan titik akhir layanan, kerentanan target, dan parameter pemindaian.

   • requirements.txt: Pustaka Python yang diperlukan.

   scan_manager.py

   # -*- coding: utf-8 -*-
   # Skrip pemindaian batch lintas akun untuk kerentanan darurat
   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
   
   # --- Impor klien Resource Directory ---
   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]:
           """Menggunakan API DescribeEmgVulItem untuk mengonversi nama tampilan kerentanan di konsol menjadi nama internal yang diperlukan oleh API.
   
           API ini meneruskan nama kerentanan dari halaman Kerentanan Darurat di konsol sebagai parameter VulName ke DescribeEmgVulItem
           dan mengekstrak bidang Name dari respons untuk panggilan selanjutnya ke ModifyEmgVulSubmit.
   
           Args:
               client: Klien Cloud Security Center.
               target_names: Daftar nama kerentanan dari file konfigurasi (nama tampilan konsol).
   
           Returns:
               Kamus di mana kunci adalah nama kerentanan API (untuk ModifyEmgVulSubmit) dan nilai adalah nama tampilan (untuk output log).
           """
           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"  ⚠️ Tidak ditemukan kerentanan darurat yang sesuai: '{target}'")
               except Exception as e:
                   print(f"  ❌ Gagal mengkueri kerentanan '{target}': {e}")
   
           return resolved
   
       @staticmethod
       def main(args: List[str]) -> None:
           # --- Baca konfigurasi dari file config.ini ---
           config = configparser.ConfigParser(delimiters=('=',))
           # Baca file menggunakan encoding UTF-8 untuk mencegah karakter rusak.
           config.read('config.ini', encoding='utf-8')
   
           try:
               # Baca konfigurasi pemindaian
               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"❌ Kesalahan membaca file konfigurasi 'config.ini': {e}")
               print("Pastikan file config.ini ada dan berisi bagian [scan_config] dengan semua kunci yang diperlukan.")
               sys.exit(1)
   
           # --- Logika penemuan aset otomatis ---
           # Jika member_uids kosong di file konfigurasi, coba ambil daftar akun secara otomatis dari Resource Directory.
           if not account_ids:
               print("ℹ️ 'member_uids' kosong, mencoba menemukan akun anggota secara otomatis dari Resource Directory...")
               try:
                   # Titik akhir umum untuk Resource Manager.
                   rm_endpoint = 'resourcemanager.aliyuncs.com'
   
                   # Buat klien Resource Directory.
                   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  # Gunakan PageSize maksimum untuk mengurangi jumlah permintaan.
                       )
                       # Panggil API ListAccounts dan tangani paginasi dalam loop.
                       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)
                           # Jika jumlah catatan yang dikembalikan kurang dari PageSize, ini adalah halaman terakhir.
                           if len(current_page_accounts) < 100:
                               break
                           page_number += 1
                       else:
                           break  # Tidak ada akun lagi
   
                   # Ekstrak ID akun dari respons.
                   account_ids = [acc.account_id for acc in all_accounts]
   
                   if account_ids:
                       print(f"✅ Berhasil menemukan {len(account_ids)} akun anggota dari Resource Directory.")
                   else:
                       print("⚠️ Gagal menemukan akun anggota apa pun dari Resource Directory. Periksa izin Anda atau status Resource Directory.")
               except Exception as e:
                   print(f"❌ Gagal menemukan akun secara otomatis dari Resource Directory: {e}")
                   print("   Pastikan Pengguna RAM memiliki izin 'AliyunResourceDirectoryReadOnlyAccess', atau konfigurasikan 'member_uids' secara manual di config.ini.")
                   # Jika penemuan otomatis gagal dan tidak ada konfigurasi manual, keluar.
                       sys.exit(1)
           # --- Akhir pembacaan konfigurasi ---
   
           client = Sample.create_client(endpoint)
           runtime = util_models.RuntimeOptions()
   
           # --- Selesaikan nama kerentanan menggunakan API ---
           print("Mengkueri daftar kerentanan darurat...")
           vul_map = Sample.resolve_vul_names(client, target_names)
   
           if not vul_map:
               print("⚠️ Tidak ditemukan kerentanan yang sesuai. Mengakhiri program.")
               return
   
           vul_names = list(vul_map.keys())
           print(f"✅ Berhasil mencocokkan {len(vul_names)} kerentanan")
   
           total_tasks = len(vul_names) * len(account_ids)
           if total_tasks == 0:
               print("⚠️  Tidak ditemukan kerentanan atau akun dalam config.ini untuk diproses. Mengakhiri program.")
               return
   
           print(f"Mulai memproses {len(vul_names)} kerentanan × {len(account_ids)} akun = {total_tasks} tugas pemindaian")
           print(f"Pembatasan kecepatan QPS diatur pada {max_qps}. Perkiraan waktu penyelesaian: {total_tasks / max_qps:.1f} detik")
   
           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
   
                   # Pembatasan kecepatan 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}] Berhasil: Kerentanan='{display_name}', ID Akun={account_id} "
                             f"(QPS rata-rata saat ini: {avg_qps:.1f})")
   
                   except Exception as error:
                       failure_count += 1
                       print(f"❌ [{request_count}/{total_tasks}] Gagal: Kerentanan='{display_name}', ID Akun={account_id}")
   
                       error_message = getattr(error, 'message', str(error))
                       print(f"   Pesan kesalahan: {error_message}")
                       recommend_message = getattr(error, 'data', {}).get('Recommend')
                       if recommend_message:
                           print(f"   Saran diagnostik: {recommend_message}")
   
                       error_str = str(error).lower()
                       if 'throttling' in error_str or 'qps' in error_str:
                           print("   ⚠️  Terdeteksi pembatasan API. Secara otomatis menambah waktu tunggu...")
                           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"Pemrosesan selesai! Total waktu: {total_time:.1f} detik")
           print(f"Berhasil: {success_count}, Gagal: {failure_count}, Total: {total_tasks}")
           print(f"QPS rata-rata aktual: {final_qps:.1f} (Target QPS: {max_qps})")
           print(f"{'=' * 50}")
   
   
   if __name__ == '__main__':
       Sample.main(sys.argv[1:])
       

   config.ini

   [scan_config]
   # Titik akhir layanan Cloud Security Center. Anda harus mengubahnya sesuai wilayah tempat aset Anda berada.
   # Misalnya, titik akhir untuk wilayah Tiongkok (Shanghai) adalah tds.cn-shanghai.aliyuncs.com.
   endpoint = tds.cn-shanghai.aliyuncs.com
   
   # Batas laju permintaan API, dalam permintaan per detik (QPS). Nilai default adalah 6.
   max_qps = 6
   
   # Daftar ID akun anggota yang dipisahkan koma.
   # Jika Pengguna RAM memiliki izin baca Resource Directory, biarkan kosong. Skrip akan mengambil daftar secara otomatis.
   member_uids =
   
   # Nama kerentanan darurat yang akan dipindai, dipisahkan koma.
   # Anda dapat menyalin nama langsung dari halaman Kerentanan Darurat di konsol Cloud Security Center.
   # Saat startup, skrip menggunakan API DescribeEmgVulItem untuk mengonversi nama-nama ini ke format yang diperlukan oleh API pemindaian.
   scan_targets = polkit pkexec Local Privilege Escalation Vulnerability (CVE-2021-4034)

   requirements.txt

   alibabacloud_sas20181203
   alibabacloud_resourcemanager20200331
   alibabacloud_credentials

2. Buka config.ini dan atur parameter berikut:

   • endpoint: titik akhir layanan Cloud Security Center untuk wilayah tempat aset Anda berada. Skrip menargetkan satu wilayah per eksekusi — jalankan secara terpisah untuk setiap wilayah jika aset tersebar di beberapa wilayah.

   • scan_targets: Nama kerentanan darurat yang akan dipindai, dipisahkan koma. Salin nama langsung dari halaman Urgent Vulnerability di konsol. Saat startup, skrip memanggil DescribeEmgVulItem untuk mengonversi nama tampilan menjadi nama API internal.

   • max_qps: Laju permintaan API maksimum dalam QPS. Default: 6, cukup untuk sebagian besar skenario. Untuk laju lebih tinggi dengan jumlah akun besar, ajukan peningkatan kuota di Quota Center.

   • member_uids: ID akun anggota yang akan dipindai, dipisahkan koma. Biarkan kosong jika Pengguna RAM memiliki izin baca RD — skrip akan menemukan semua akun secara otomatis.

Langkah 3: Jalankan pemindaian

1. Ekspor AccessKey Pengguna RAM (yang dibuat di Langkah 1) sebagai variabel lingkungan.

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

2. Instal dependensi.

   pip3 install -r requirements.txt

3. Jalankan skrip pemindaian.

   python3 scan_manager.py

   Skrip mencetak progres secara real-time. Contoh output:

   Mengkueri daftar kerentanan darurat...
   ✅ 'polkit pkexec Local Privilege Escalation Vulnerability (CVE-2021-4034)' → emg_cve-2021-4034:EMG:AVD-2021-4034
   ✅ Berhasil mencocokkan 1 kerentanan
   
   ℹ️ 'member_uids' kosong, mencoba menemukan akun anggota secara otomatis dari Resource Directory...
   ✅ Berhasil menemukan 150 akun anggota dari Resource Directory.
   
   Mulai memproses 1 kerentanan × 150 akun = 150 tugas pemindaian
   Pembatasan kecepatan QPS diatur pada 6. Perkiraan waktu penyelesaian: 25.0 detik
   
   ✅ [1/150] Berhasil: Kerentanan='polkit pkexec Local Privilege Escalation Vulnerability (CVE-2021-4034)', ID Akun=198XXXXXXXXXXX13 (QPS rata-rata saat ini: 5.8)
   ✅ [2/150] Berhasil: Kerentanan='polkit pkexec Local Privilege Escalation Vulnerability (CVE-2021-4034)', ID Akun=142XXXXXXXXXXX84 (QPS rata-rata saat ini: 5.9)
   ✅ [3/150] Berhasil: Kerentanan='polkit pkexec Local Privilege Escalation Vulnerability (CVE-2021-4034)', ID Akun=188XXXXXXXXXXX97 (QPS rata-rata saat ini: 5.7)
   ...
   ✅ [150/150] Berhasil: Kerentanan='polkit pkexec Local Privilege Escalation Vulnerability (CVE-2021-4034)', ID Akun=156XXXXXXXXXXX62 (QPS rata-rata saat ini: 5.9)
   
   ==================================================
   Pemrosesan selesai! Total waktu: 25.8 detik
   Berhasil: 150, Gagal: 0, Total: 150
   QPS rata-rata aktual: 5.8 (Target QPS: 6)
   ==================================================

Langkah 4: Lihat hasil pemindaian

Login ke Konsol Cloud Security Center sebagai Pengguna RAM. Di panel navigasi kiri, pilih Agentic SOC > Log. Dari daftar Logstore, pilih Standardized Log > Vulnerability Activity. Atur rentang waktu yang mencakup jendela pemindaian Anda dan klik Search & Analyze.

Praktik terbaik untuk lingkungan produksi

FAQ