Mengumpulkan log Cisco Application Centric Infrastructure (ACI)

Didukung di:

Google SecOps SIEM

Catatan: Fitur ini tercakup dalam Persyaratan Penawaran Pra-GA dari Persyaratan Khusus Layanan Google Security Operations. Fitur pra-GA mungkin memiliki dukungan terbatas, dan perubahan pada fitur pra-GA mungkin tidak kompatibel dengan versi pra-GA lainnya. Untuk mengetahui informasi selengkapnya, lihat panduan Layanan Dukungan Teknis Google SecOps dan Persyaratan Khusus Layanan Google SecOps.

Dokumen ini menjelaskan cara menyerap log Cisco Application Centric Infrastructure (ACI) ke Google Security Operations. Parser pertama-tama mencoba memproses log Cisco ACI yang masuk sebagai pesan syslog menggunakan pola Grok. Jika penguraian syslog gagal, maka diasumsikan bahwa pesan dalam format JSON dan akan diuraikan sesuai dengan itu. Terakhir, kolom yang diekstrak dipetakan ke model data terpadu (UDM).

Integrasi ini mendukung dua metode:

• Opsi 1: Format Syslog menggunakan agen Bindplane
• Opsi 2: Format JSON menggunakan Google Cloud Storage menggunakan APIC REST API

Setiap opsi bersifat mandiri dan dapat diterapkan secara independen berdasarkan persyaratan infrastruktur dan preferensi format log Anda.

Opsi 1: Syslog menggunakan agen Bindplane

Opsi ini mengonfigurasi fabric Cisco ACI untuk mengirim pesan syslog ke agen Bindplane, yang meneruskannya ke Google Security Operations untuk dianalisis.

Sebelum memulai

Pastikan Anda memiliki prasyarat berikut:

• Instance Google SecOps
• Host Windows 2016 atau yang lebih baru atau Linux dengan systemd
• Jika beroperasi dari balik proxy, pastikan port firewall terbuka sesuai dengan persyaratan agen Bindplane
• Akses istimewa ke konsol Cisco APIC

Mendapatkan file autentikasi penyerapan Google SecOps

1. Buka Setelan SIEM > Agen Pengumpulan.
2. Download File Autentikasi Penyerapan.
3. Simpan file dengan aman di sistem tempat BindPlane akan diinstal.

Mendapatkan ID pelanggan Google SecOps

1. Buka Setelan SIEM > Profil.
2. Salin dan simpan ID Pelanggan dari bagian Detail Organisasi.

Menginstal agen Bindplane

Instal agen Bindplane di sistem operasi Windows atau Linux Anda sesuai dengan petunjuk berikut.

Penginstalan Windows

1. Buka Command Prompt atau PowerShell sebagai administrator.

2. Jalankan perintah berikut:

   msiexec /i "https://github.com/observIQ/bindplane-agent/releases/latest/download/observiq-otel-collector.msi" /quiet
   

Penginstalan Linux

1. Buka terminal dengan hak istimewa root atau sudo.

2. Jalankan perintah berikut:

   sudo sh -c "$(curl -fsSlL https://github.com/observiq/bindplane-agent/releases/latest/download/install_unix.sh)" install_unix.sh
   

Untuk opsi penginstalan tambahan, lihat Panduan penginstalan agen BindPlane.

Mengonfigurasi agen BindPlane untuk menyerap Syslog dan mengirimkannya ke Google SecOps

Mengakses file konfigurasi

1. Cari file config.yaml. Biasanya, file ini berada di direktori /etc/bindplane-agent/ di Linux atau di direktori penginstalan di Windows.
2. Buka file menggunakan editor teks (misalnya, nano, vi, atau Notepad).

3. Edit file config.yaml:

   receivers:
     udplog:
       # Replace the port and IP address as required
       listen_address: "0.0.0.0:514"
   
   exporters:
     chronicle/chronicle_w_labels:
       compression: gzip
       # Adjust the path to the credentials file you downloaded
       creds_file_path: '/path/to/ingestion-authentication-file.json'
       # Replace with your actual customer ID
       customer_id: <CUSTOMER_ID>
       endpoint: malachiteingestion-pa.googleapis.com
       # Add optional ingestion labels for better organization
       log_type: 'CISCO_ACI'
       raw_log_field: body
       ingestion_labels:
         service:
   
   pipelines:
     logs/source0__chronicle_w_labels-0:
       receivers:
         - udplog
       exporters:
         - chronicle/chronicle_w_labels
   

   • Ganti kode berikut:
     • Ganti port dan alamat IP sesuai kebutuhan di infrastruktur Anda.
     • Ganti <CUSTOMER_ID> dengan ID pelanggan yang sebenarnya.
     • Ganti /path/to/ingestion-authentication-file.json dengan jalur tempat file autentikasi disimpan.

Mulai ulang agen Bindplane untuk menerapkan perubahan

• Untuk memulai ulang agen Bindplane di Linux, jalankan perintah berikut:

  sudo systemctl restart bindplane-agent
  

• Untuk memulai ulang agen Bindplane di Windows, Anda dapat menggunakan konsol Services atau memasukkan perintah berikut:

  net stop BindPlaneAgent && net start BindPlaneAgent
  

Mengonfigurasi penerusan Syslog di Cisco ACI

Mengonfigurasi Kontrak Pengelolaan Out-of-Band

1. Login ke konsol Cisco APIC.
2. Buka Tenants > mgmt > Contracts > Filters.
3. Klik Buat Filter.
4. Berikan detail konfigurasi berikut:
   • Nama: Masukkan syslog-udp-514.
   • Nama Entri: Masukkan syslog.
   • EtherType: Pilih IP.
   • IP Protocol: Pilih UDP.
   • Rentang Port Tujuan Dari: Masukkan 514.
   • Rentang Port Tujuan Ke: Masukkan 514.
5. Klik Kirim.

Buat Kontrak Pengelolaan

1. Buka Tenants > mgmt > Contracts > Standard.
2. Klik Buat Kontrak.
3. Berikan detail konfigurasi berikut:
   • Nama: Masukkan mgmt-syslog-contract.
   • Cakupan: Pilih Konteks.
4. Klik Kirim.
5. Perluas kontrak, lalu klik Subjek.
6. Klik Buat Subjek Kontrak.
7. Berikan detail konfigurasi berikut:
   • Nama: Masukkan syslog-subject.
   • Terapkan Kedua Arah: Centang opsi ini.
8. Klik Kirim.
9. Luaskan subjek dan klik Filter.
10. Klik Create Filter Binding.
11. Pilih filter syslog-udp-514.
12. Klik Kirim.

Mengonfigurasi Grup Tujuan Syslog

1. Buka Admin > Pengumpul Data Eksternal > Tujuan Pemantauan > Syslog.
2. Klik kanan Syslog, lalu pilih Create Syslog Monitoring Destination Group.
3. Berikan detail konfigurasi berikut:
   • Nama: Masukkan Chronicle-Syslog-Group.
   • Admin State: Pilih Enabled.
   • Format: Pilih aci.
4. Klik Berikutnya.
5. Dalam dialog Create Syslog Monitoring Destination:
   • Nama: Masukkan Chronicle-BindPlane.
   • Host: Masukkan alamat IP server agen Bindplane Anda.
   • Port: Masukkan 514.
   • Admin State: Pilih Enabled.
   • Tingkat keparahan: Pilih informasi (untuk merekam log mendetail).
6. Klik Kirim.

Mengonfigurasi Kebijakan Pemantauan

Kebijakan Pemantauan Fabric

1. Buka Fabric > Fabric Policies > Policies > Monitoring > Common Policy.
2. Perluas Callhome/Smart Callhome/SNMP/Syslog/TACACS.
3. Klik kanan Syslog, lalu pilih Create Syslog Source.
4. Berikan detail konfigurasi berikut:
   • Nama: Masukkan Chronicle-Fabric-Syslog.
   • Log Audit: Centang untuk menyertakan peristiwa audit.
   • Peristiwa: Centang untuk menyertakan peristiwa sistem.
   • Kesalahan: Centang untuk menyertakan peristiwa kesalahan.
   • Log Sesi: Centang untuk menyertakan log sesi.
   • Grup Tujuan: Pilih Chronicle-Syslog-Group.
5. Klik Kirim.

Kebijakan Pemantauan Akses

1. Buka Fabric > Access Policies > Policies > Monitoring > Default Policy.
2. Perluas Callhome/Smart Callhome/SNMP/Syslog.
3. Klik kanan Syslog, lalu pilih Create Syslog Source.
4. Berikan detail konfigurasi berikut:
   • Nama: Masukkan Chronicle-Access-Syslog.
   • Log Audit: Centang untuk menyertakan peristiwa audit.
   • Peristiwa: Centang untuk menyertakan peristiwa sistem.
   • Kesalahan: Centang untuk menyertakan peristiwa kesalahan.
   • Log Sesi: Centang untuk menyertakan log sesi.
   • Grup Tujuan: Pilih Chronicle-Syslog-Group.
5. Klik Kirim.

Mengonfigurasi Kebijakan Pesan Syslog Sistem

1. Buka Fabric > Fabric Policies > Policies > Monitoring > Common Policy.
2. Luaskan Syslog Messages Policies.
3. Klik default.
4. Di bagian Filter Fasilitas:
   • Fasilitas: Pilih default.
   • Tingkat Keparahan Minimum: Ubah menjadi informasi.
5. Klik Kirim.

Opsi 2: JSON menggunakan Google Cloud Storage

Opsi ini menggunakan APIC REST API untuk mengumpulkan peristiwa, kesalahan, dan log audit berformat JSON dari fabric Cisco ACI dan menyimpannya di Google Cloud Storage untuk penyerapan Google SecOps.

Sebelum memulai

Pastikan Anda memiliki prasyarat berikut:

• Instance Google SecOps
• Project GCP dengan Cloud Storage API yang diaktifkan
• Izin untuk membuat dan mengelola bucket GCS
• Izin untuk mengelola kebijakan IAM di bucket GCS
• Izin untuk membuat layanan Cloud Run, topik Pub/Sub, dan tugas Cloud Scheduler
• Akses istimewa ke konsol Cisco APIC

Mengumpulkan prasyarat APIC ACI Cisco

Mendapatkan kredensial APIC

1. Login ke konsol Cisco APIC menggunakan HTTPS.

2. Buka Admin > AAA (di APIC 6.0+) atau Admin > Authentication > AAA (di rilis yang lebih lama).

   Catatan: Jalur menu AAA berubah mulai dari APIC 6.0(1).

3. Buat atau gunakan pengguna lokal yang sudah ada dengan hak istimewa yang sesuai.

4. Salin dan simpan detail berikut di lokasi yang aman:

   • Nama Pengguna APIC: Pengguna lokal dengan akses baca ke data pemantauan
   • Sandi APIC: Sandi pengguna
   • URL APIC: URL HTTPS APIC Anda (misalnya, https://apic.example.com)

Catatan: Akun pengguna harus memiliki izin baca ke class tertentu yang dikueri (faultInst, eventRecord, aaaModLR) sesuai dengan kontrol akses berbasis peran Cisco.

Membuat bucket Google Cloud Storage

1. Buka Google Cloud console.
2. Pilih project Anda atau buat project baru.
3. Di menu navigasi, buka Cloud Storage > Buckets.
4. Klik Create bucket.

5. Berikan detail konfigurasi berikut:

   Setelan	Nilai
   Beri nama bucket Anda	Masukkan nama yang unik secara global (misalnya, cisco-aci-logs)
   Location type	Pilih berdasarkan kebutuhan Anda (Region, Dual-region, Multi-region)
   Lokasi	Pilih lokasi (misalnya, us-central1)
   Kelas penyimpanan	Standar (direkomendasikan untuk log yang sering diakses)
   Access control	Seragam (direkomendasikan)
   Alat perlindungan	Opsional: Aktifkan pembuatan versi objek atau kebijakan retensi

6. Klik Buat.

Buat akun layanan untuk Cloud Run Function

Fungsi Cloud Run memerlukan akun layanan dengan izin untuk menulis ke bucket GCS dan dipanggil oleh Pub/Sub.

Membuat akun layanan

1. Di GCP Console, buka IAM & Admin > Service Accounts.
2. Klik Create Service Account.
3. Berikan detail konfigurasi berikut:
   • Nama akun layanan: Masukkan cisco-aci-collector-sa.
   • Deskripsi akun layanan: Masukkan Service account for Cloud Run function to collect Cisco ACI logs.
4. Klik Create and Continue.
5. Di bagian Berikan akun layanan ini akses ke project, tambahkan peran berikut:
   1. Klik Pilih peran.
   2. Telusuri dan pilih Storage Object Admin.
   3. Klik + Add another role.
   4. Telusuri dan pilih Cloud Run Invoker.
   5. Klik + Add another role.
   6. Telusuri dan pilih Cloud Functions Invoker.
6. Klik Lanjutkan.
7. Klik Selesai.

Peran ini diperlukan untuk:

• Storage Object Admin: Menulis log ke bucket GCS dan mengelola file status
• Cloud Run Invoker: Mengizinkan Pub/Sub memanggil fungsi
• Cloud Functions Invoker: Mengizinkan pemanggilan fungsi

Memberikan izin IAM pada bucket GCS

Beri akun layanan (cisco-aci-collector-sa) izin tulis di bucket GCS:

1. Buka Cloud Storage > Buckets.
2. Klik nama bucket Anda (misalnya, cisco-aci-logs).
3. Buka tab Izin.
4. Klik Grant access.
5. Berikan detail konfigurasi berikut:
   • Tambahkan prinsipal: Masukkan email akun layanan (misalnya, cisco-aci-collector-sa@PROJECT_ID.iam.gserviceaccount.com).
   • Tetapkan peran: Pilih Storage Object Admin.
6. Klik Simpan.

Membuat topik Pub/Sub

Buat topik Pub/Sub yang akan dipublikasikan oleh Cloud Scheduler dan akan dilanggan oleh fungsi Cloud Run.

1. Di GCP Console, buka Pub/Sub > Topics.
2. Klik Create topic.
3. Berikan detail konfigurasi berikut:
   • ID Topik: Masukkan cisco-aci-trigger.
   • Biarkan setelan lainnya tetap default.
4. Klik Buat.

Membuat fungsi Cloud Run untuk mengumpulkan log

Fungsi Cloud Run akan dipicu oleh pesan Pub/Sub dari Cloud Scheduler untuk mengambil log dari Cisco APIC REST API dan menuliskannya ke GCS.

1. Di GCP Console, buka Cloud Run.
2. Klik Create service.
3. Pilih Function (gunakan editor inline untuk membuat fungsi).

4. Di bagian Konfigurasi, berikan detail konfigurasi berikut:

   Setelan	Nilai
   Nama layanan	cisco-aci-collector
   Wilayah	Pilih region yang cocok dengan bucket GCS Anda (misalnya, us-central1)
   Runtime	Pilih Python 3.12 atau yang lebih baru

5. Di bagian Pemicu (opsional):

   1. Klik + Tambahkan pemicu.
   2. Pilih Cloud Pub/Sub.
   3. Di Select a Cloud Pub/Sub topic, pilih topik Pub/Sub (cisco-aci-trigger).
   4. Klik Simpan.

6. Di bagian Authentication:

   • Pilih Wajibkan autentikasi.
   • Pilih Identity and Access Management (IAM).

Catatan: Pub/Sub secara otomatis menangani autentikasi saat memanggil fungsi.

1. Scroll ke dan luaskan Containers, Networking, Security.
2. Buka tab Security:
   • Akun layanan: Pilih akun layanan (cisco-aci-collector-sa).

3. Buka tab Containers:

   • Klik Variables & Secrets.

   • Klik + Tambahkan variabel untuk setiap variabel lingkungan:

     Nama Variabel	Nilai Contoh	Deskripsi
     GCS_BUCKET	cisco-aci-logs	Nama bucket GCS
     GCS_PREFIX	cisco-aci-events	Awalan untuk file log
     STATE_KEY	cisco-aci-events/state.json	Jalur file status
     APIC_URL	https://apic.example.com	URL HTTPS APIC
     APIC_USERNAME	your-apic-username	Nama pengguna APIC
     APIC_PASSWORD	your-apic-password	Sandi APIC
     PAGE_SIZE	100	Data per halaman
     MAX_PAGES	10	Halaman maks per proses

   Catatan: Untuk lingkungan produksi, sebaiknya simpan kredensial APIC di Google Secret Manager, bukan di variabel lingkungan.

4. Di bagian Variabel & Secret, scroll ke Permintaan:

   • Waktu tunggu permintaan: Masukkan 300 detik (5 menit).

5. Buka tab Setelan:

   • Di bagian Materi:
     • Memori: Pilih 512 MiB atau yang lebih tinggi.
     • CPU: Pilih 1.

6. Di bagian Penskalaan revisi:

   • Jumlah minimum instance: Masukkan 0.
   • Jumlah maksimum instance: Masukkan 100 (atau sesuaikan berdasarkan perkiraan beban).

7. Klik Buat.

8. Tunggu hingga layanan dibuat (1-2 menit).

9. Setelah layanan dibuat, editor kode inline akan terbuka secara otomatis.

Menambahkan kode fungsi

1. Masukkan main di kolom Entry point.

2. Di editor kode inline, buat dua file:

   • File pertama: main.py:

   import functions_framework
   from google.cloud import storage
   import json
   import os
   import urllib3
   from datetime import datetime, timezone, timedelta
   import logging
   
   # Configure logging
   logger = logging.getLogger()
   logger.setLevel(logging.INFO)
   
   # Initialize HTTP client with timeouts
   http = urllib3.PoolManager(
       timeout=urllib3.Timeout(connect=5.0, read=60.0),
       retries=False,
   )
   
   # Initialize Storage client
   storage_client = storage.Client()
   
   # Environment variables
   GCS_BUCKET = os.environ.get('GCS_BUCKET')
   GCS_PREFIX = os.environ.get('GCS_PREFIX', 'cisco-aci-events')
   STATE_KEY = os.environ.get('STATE_KEY', 'cisco-aci-events/state.json')
   APIC_URL = os.environ.get('APIC_URL')
   APIC_USERNAME = os.environ.get('APIC_USERNAME')
   APIC_PASSWORD = os.environ.get('APIC_PASSWORD')
   PAGE_SIZE = int(os.environ.get('PAGE_SIZE', '100'))
   MAX_PAGES = int(os.environ.get('MAX_PAGES', '10'))
   
   @functions_framework.cloud_event
   def main(cloud_event):
       """
       Cloud Run function triggered by Pub/Sub to fetch Cisco ACI logs and write to GCS.
   
       Args:
           cloud_event: CloudEvent object containing Pub/Sub message
       """
   
       if not all([GCS_BUCKET, APIC_URL, APIC_USERNAME, APIC_PASSWORD]):
           logger.error('Error: Missing required environment variables')
           return
   
       try:
           # Get GCS bucket
           bucket = storage_client.bucket(GCS_BUCKET)
   
           # Load state
           state = load_state(bucket, STATE_KEY)
   
           # Determine time window
           last_timestamp = state.get('last_timestamp')
           if not last_timestamp:
               last_timestamp = (datetime.utcnow() - timedelta(hours=1)).isoformat() + 'Z'
   
           logger.info(f"Starting Cisco ACI data collection for bucket: {GCS_BUCKET}")
   
           # Authenticate to APIC
           session_token = authenticate_apic(APIC_URL, APIC_USERNAME, APIC_PASSWORD)
           headers = {
               'Cookie': f'APIC-cookie={session_token}',
               'Accept': 'application/json',
               'Content-Type': 'application/json'
           }
   
           # Data types to collect
           data_types = ['faultInst', 'eventRecord', 'aaaModLR']
           all_collected_data = []
   
           for data_type in data_types:
               logger.info(f"Collecting {data_type} data")
               collected_data = collect_aci_data(
                   APIC_URL,
                   headers,
                   data_type,
                   last_timestamp,
                   PAGE_SIZE,
                   MAX_PAGES
               )
   
               # Tag each record with its type
               for record in collected_data:
                   record['_data_type'] = data_type
   
               all_collected_data.extend(collected_data)
               logger.info(f"Collected {len(collected_data)} {data_type} records")
   
           logger.info(f"Total records collected: {len(all_collected_data)}")
   
           # Store data in GCS if any were collected
           if all_collected_data:
               timestamp_str = datetime.utcnow().strftime('%Y%m%d_%H%M%S')
               s3_key = f"{GCS_PREFIX}/cisco_aci_events_{timestamp_str}.ndjson"
   
               # Convert to NDJSON format (one JSON object per line)
               ndjson_content = '\n'.join(json.dumps(record) for record in all_collected_data)
   
               # Upload to GCS
               blob = bucket.blob(s3_key)
               blob.upload_from_string(
                   ndjson_content,
                   content_type='application/x-ndjson'
               )
   
               logger.info(f"Uploaded {len(all_collected_data)} records to gs://{GCS_BUCKET}/{s3_key}")
   
               # Update state file with latest timestamp from collected data
               latest_timestamp = get_latest_timestamp_from_records(all_collected_data)
               if not latest_timestamp:
                   latest_timestamp = datetime.utcnow().isoformat() + 'Z'
   
               update_state(bucket, STATE_KEY, latest_timestamp)
           else:
               logger.info("No new log records found.")
   
           logger.info(f"Successfully processed {len(all_collected_data)} records")
   
       except Exception as e:
           logger.error(f'Error processing logs: {str(e)}')
           raise
   
   def authenticate_apic(apic_url, username, password):
       """Authenticate to APIC and return session token"""
       login_url = f"{apic_url}/api/aaaLogin.json"
       login_data = {
           "aaaUser": {
               "attributes": {
                   "name": username,
                   "pwd": password
               }
           }
       }
   
       response = http.request(
           'POST',
           login_url,
           body=json.dumps(login_data).encode('utf-8'),
           headers={'Content-Type': 'application/json'},
           timeout=30
       )
   
       if response.status != 200:
           raise RuntimeError(f"APIC authentication failed: {response.status} {response.data[:256]!r}")
   
       response_data = json.loads(response.data.decode('utf-8'))
       token = response_data['imdata'][0]['aaaLogin']['attributes']['token']
       logger.info("Successfully authenticated to APIC")
       return token
   
   def collect_aci_data(apic_url, headers, data_type, last_timestamp, page_size, max_pages):
       """Collect data from APIC REST API with pagination"""
       all_data = []
       page = 0
   
       while page < max_pages:
           # Build API URL with pagination and time filters
           api_url = f"{apic_url}/api/class/{data_type}.json"
           params = [
               f'page-size={page_size}',
               f'page={page}',
               f'order-by={data_type}.created|asc'
           ]
   
           # Add time filter to prevent duplicates
           if last_timestamp:
               params.append(f'query-target-filter=gt({data_type}.created,"{last_timestamp}")')
   
           full_url = f"{api_url}?{'&'.join(params)}"
   
           logger.info(f"Fetching {data_type} page {page} from APIC")
   
           # Make API request
           response = http.request('GET', full_url, headers=headers, timeout=60)
   
           if response.status != 200:
               logger.error(f"API request failed: {response.status} {response.data[:256]!r}")
               break
   
           data = json.loads(response.data.decode('utf-8'))
           records = data.get('imdata', [])
   
           if not records:
               logger.info(f"No more {data_type} records found")
               break
   
           # Extract the actual data from APIC format
           extracted_records = []
           for record in records:
               if data_type in record:
                   extracted_records.append(record[data_type])
   
           all_data.extend(extracted_records)
           page += 1
   
           # If we got less than page_size records, we've reached the end
           if len(records) < page_size:
               break
   
       return all_data
   
   def get_last_timestamp(bucket, state_key):
       """Get the last run timestamp from GCS state file"""
       try:
           blob = bucket.blob(state_key)
           if blob.exists():
               state_data = blob.download_as_text()
               state = json.loads(state_data)
               return state.get('last_timestamp')
       except Exception as e:
           logger.warning(f"Error reading state file: {str(e)}")
   
       return None
   
   def get_latest_timestamp_from_records(records):
       """Get the latest timestamp from collected records to prevent missing events"""
       if not records:
           return None
   
       latest = None
       latest_time = None
   
       for record in records:
           try:
               # Handle both direct attributes and nested structure
               attrs = record.get('attributes', record)
               created = attrs.get('created')
               modTs = attrs.get('modTs')
   
               # Use created or modTs as fallback
               timestamp = created or modTs
   
               if timestamp:
                   if latest_time is None or timestamp > latest_time:
                       latest_time = timestamp
                       latest = record
           except Exception as e:
               logger.debug(f"Error parsing timestamp from record: {e}")
               continue
   
       return latest_time
   
   def update_state(bucket, state_key, timestamp):
       """Update the state file with the current timestamp"""
       try:
           state_data = {
               'last_timestamp': timestamp,
               'updated_at': datetime.utcnow().isoformat() + 'Z'
           }
           blob = bucket.blob(state_key)
           blob.upload_from_string(
               json.dumps(state_data),
               content_type='application/json'
           )
           logger.info(f"Updated state file with timestamp: {timestamp}")
       except Exception as e:
           logger.error(f"Error updating state file: {str(e)}")
   
   def load_state(bucket, key):
       """Load state from GCS."""
       try:
           blob = bucket.blob(key)
           if blob.exists():
               state_data = blob.download_as_text()
               return json.loads(state_data)
       except Exception as e:
           logger.warning(f"Could not load state: {e}")
   
       return {}
   

   • File kedua: requirements.txt:

   functions-framework==3.*
   google-cloud-storage==2.*
   urllib3>=2.0.0
   

3. Klik Deploy untuk menyimpan dan men-deploy fungsi.

4. Tunggu hingga deployment selesai (2-3 menit).

Catatan: Konfigurasi pemicu Pub/Sub akan otomatis membuat langganan dan izin yang diperlukan. URL APIC harus dapat dijangkau dari lingkungan jaringan fungsi Cloud Run, dan aturan firewall yang sesuai harus mengizinkan HTTPS (TCP/443) dari fungsi ke APIC.

Buat tugas Cloud Scheduler

Cloud Scheduler memublikasikan pesan ke topik Pub/Sub secara berkala, sehingga memicu fungsi Cloud Run.

1. Di GCP Console, buka Cloud Scheduler.
2. Klik Create Job.

3. Berikan detail konfigurasi berikut:

   Setelan	Nilai
   Nama	cisco-aci-collector-15m
   Wilayah	Pilih region yang sama dengan fungsi Cloud Run
   Frekuensi	*/15 * * * * (setiap 15 menit)
   Zona waktu	Pilih zona waktu (UTC direkomendasikan)
   Jenis target	Pub/Sub
   Topik	Pilih topik Pub/Sub (cisco-aci-trigger)
   Isi pesan	{} (objek JSON kosong)

4. Klik Buat.

Opsi frekuensi jadwal

• Pilih frekuensi berdasarkan volume log dan persyaratan latensi:

  Frekuensi	Ekspresi Cron	Kasus Penggunaan
  Setiap 5 menit	*/5 * * * *	Volume tinggi, latensi rendah
  Setiap 15 menit	*/15 * * * *	Volume sedang (direkomendasikan)
  Setiap jam	0 * * * *	Standar
  Setiap 6 jam	0 */6 * * *	Volume rendah, pemrosesan batch
  Harian	0 0 * * *	Pengumpulan data historis

Menguji integrasi

1. Di konsol Cloud Scheduler, temukan tugas Anda (cisco-aci-collector-15m).
2. Klik Force run untuk memicu tugas secara manual.
3. Tunggu beberapa detik.
4. Buka Cloud Run > Services.
5. Klik nama fungsi Anda (cisco-aci-collector).
6. Klik tab Logs.

7. Pastikan fungsi berhasil dieksekusi. Cari hal berikut:

   Starting Cisco ACI data collection for bucket: cisco-aci-logs
   Successfully authenticated to APIC
   Collecting faultInst data
   Collected X faultInst records
   Collecting eventRecord data
   Collected X eventRecord records
   Collecting aaaModLR data
   Collected X aaaModLR records
   Total records collected: X
   Uploaded X records to gs://cisco-aci-logs/cisco-aci-events/cisco_aci_events_YYYYMMDD_HHMMSS.ndjson
   Successfully processed X records
   

8. Buka Cloud Storage > Buckets.

9. Klik nama bucket Anda (cisco-aci-logs).

10. Buka folder awalan (cisco-aci-events/).

11. Pastikan file .ndjson baru dibuat dengan stempel waktu saat ini.

Jika Anda melihat error dalam log:

• HTTP 401: Periksa kredensial APIC di variabel lingkungan
• HTTP 403: Pastikan akun APIC memiliki izin baca ke class faultInst, eventRecord, dan aaaModLR
• Error koneksi: Verifikasi bahwa fungsi Cloud Run dapat mencapai URL APIC di TCP/443
• Variabel lingkungan tidak ada: Periksa apakah semua variabel yang diperlukan telah ditetapkan

Mengambil akun layanan Google SecOps

Google SecOps menggunakan akun layanan unik untuk membaca data dari bucket GCS Anda. Anda harus memberi akun layanan ini akses ke bucket Anda.

Dapatkan email akun layanan

1. Buka Setelan SIEM > Feed.
2. Klik Tambahkan Feed Baru.
3. Klik Konfigurasi satu feed.
4. Di kolom Nama feed, masukkan nama untuk feed (misalnya, Cisco ACI JSON logs).
5. Pilih Google Cloud Storage V2 sebagai Source type.
6. Pilih Cisco Application Centric Infrastructure sebagai Log type.

7. Klik Get Service Account. Email akun layanan yang unik akan ditampilkan, misalnya:

   [e-mail]
   

8. Salin alamat email ini. Anda akan menggunakannya pada langkah berikutnya.

Catatan: Setiap instance Google SecOps memiliki akun layanan yang unik. Jangan gunakan akun layanan dari dokumentasi atau contoh lain.

Memberikan izin IAM ke akun layanan Google SecOps

Akun layanan Google SecOps memerlukan peran Storage Object Viewer di bucket GCS Anda.

1. Buka Cloud Storage > Buckets.
2. Klik nama bucket Anda (cisco-aci-logs).
3. Buka tab Izin.
4. Klik Grant access.
5. Berikan detail konfigurasi berikut:
   • Add principals: Tempel email akun layanan Google SecOps.
   • Tetapkan peran: Pilih Storage Object Viewer.
6. Klik Simpan.

Catatan: Jika Anda berencana menggunakan opsi penghapusan "Hapus file yang ditransfer" atau "Hapus file yang ditransfer dan direktori kosong", berikan peran Storage Object Admin, bukan Storage Object Viewer.

Mengonfigurasi feed di Google SecOps untuk menyerap log Cisco ACI

1. Buka Setelan SIEM > Feed.
2. Klik Tambahkan Feed Baru.
3. Klik Konfigurasi satu feed.
4. Di kolom Nama feed, masukkan nama untuk feed (misalnya, Cisco ACI JSON logs).
5. Pilih Google Cloud Storage V2 sebagai Source type.
6. Pilih Cisco Application Centric Infrastructure sebagai Log type.
7. Klik Berikutnya.

8. Tentukan nilai untuk parameter input berikut:

   • URL bucket penyimpanan: Masukkan URI bucket GCS dengan jalur awalan:

     gs://cisco-aci-logs/cisco-aci-events/
     

     • Ganti:
       • cisco-aci-logs: Nama bucket GCS Anda.
       • cisco-aci-events: Jalur folder/awalan tempat log disimpan.
   Catatan: Selalu sertakan garis miring (/) di akhir URI.

   • Opsi penghapusan sumber: Pilih opsi penghapusan sesuai preferensi Anda:

     • Jangan pernah: Tidak pernah menghapus file apa pun setelah transfer (direkomendasikan untuk pengujian).
     • Hapus file yang ditransfer: Menghapus file setelah transfer berhasil.
     • Hapus file yang ditransfer dan direktori kosong: Menghapus file dan direktori kosong setelah transfer berhasil.
     Catatan: Jika Anda memilih opsi penghapusan, akun layanan harus memiliki peran Storage Object Admin, bukan Storage Object Viewer. Perbarui izin IAM yang sesuai.

   • Usia File Maksimum: Menyertakan file yang diubah dalam beberapa hari terakhir. Defaultnya adalah 180 hari.

   • Namespace aset: Namespace aset.

   • Label penyerapan: Label yang diterapkan ke peristiwa dari feed ini.

9. Klik Berikutnya.

10. Tinjau konfigurasi feed baru Anda di layar Selesaikan, lalu klik Kirim.

Tabel pemetaan UDM

Kolom Log	Pemetaan UDM	Logika
@timestamp	read_only_udm.metadata.event_timestamp	Nilai diambil dari kolom log mentah '@timestamp' dan diuraikan sebagai stempel waktu.
aci_tag	read_only_udm.metadata.product_log_id	Nilai diambil dari kolom log mentah 'aci_tag'.
cisco_timestamp	-	Tidak dipetakan.
DIP	read_only_udm.target.ip	Nilai diambil dari kolom log mentah 'DIP'.
DPort	read_only_udm.target.port	Nilai diambil dari kolom log mentah 'DPort' dan dikonversi menjadi bilangan bulat.
deskripsi	read_only_udm.security_result.description	Nilai diambil dari kolom log mentah 'description'.
fault_cause	read_only_udm.additional.fields.value.string_value	Nilai diambil dari kolom log mentah 'fault_cause'. Kunci ditetapkan ke 'Fault Cause'.
hostname	read_only_udm.principal.hostname	Nilai diambil dari kolom log mentah 'hostname'.
lifecycle_state	read_only_udm.metadata.product_event_type	Nilai diambil dari kolom log mentah 'lifecycle_state'.
log.source.address	-	Tidak dipetakan.
logstash.collect.host	-	Tidak dipetakan.
logstash.collect.timestamp	read_only_udm.metadata.collected_timestamp	Nilai diambil dari kolom log mentah 'logstash.collect.timestamp' dan diuraikan sebagai stempel waktu.
logstash.ingest.host	read_only_udm.intermediary.hostname	Nilai diambil dari kolom log mentah 'logstash.ingest.host'.
logstash.irm_environment	read_only_udm.additional.fields.value.string_value	Nilai diambil dari kolom log mentah 'logstash.irm_environment'. Kuncinya ditetapkan ke 'IRM_Environment'.
logstash.irm_region	read_only_udm.additional.fields.value.string_value	Nilai diambil dari kolom log mentah 'logstash.irm_region'. Kuncinya ditetapkan ke 'IRM_Region'.
logstash.irm_site	read_only_udm.additional.fields.value.string_value	Nilai diambil dari kolom log mentah 'logstash.irm_site'. Kunci ditetapkan ke 'IRM_Site'.
logstash.process.host	read_only_udm.intermediary.hostname	Nilai diambil dari kolom log mentah 'logstash.process.host'.
pesan	-	Tidak dipetakan.
message_class	-	Tidak dipetakan.
message_code	-	Tidak dipetakan.
message_content	-	Tidak dipetakan.
message_dn	-	Tidak dipetakan.
message_type	read_only_udm.metadata.product_event_type	Nilai diambil dari kolom log mentah 'message_type' setelah menghapus tanda kurung siku.
node_link	read_only_udm.principal.process.file.full_path	Nilai diambil dari kolom log mentah 'node_link'.
PktLen	read_only_udm.network.received_bytes	Nilai diambil dari kolom log mentah 'PktLen' dan dikonversi menjadi bilangan bulat yang tidak bertanda tangan.
program	-	Tidak dipetakan.
Proto	read_only_udm.network.ip_protocol	Nilai diambil dari kolom log mentah 'Proto', dikonversi menjadi bilangan bulat, dan dipetakan ke nama protokol IP yang sesuai (misalnya, 6 -> TCP).
SIP	read_only_udm.principal.ip	Nilai diambil dari kolom log mentah 'SIP'.
SPort	read_only_udm.principal.port	Nilai diambil dari kolom log mentah 'SPort' dan dikonversi menjadi bilangan bulat.
syslog_facility	-	Tidak dipetakan.
syslog_facility_code	-	Tidak dipetakan.
syslog_host	read_only_udm.principal.ip, read_only_udm.observer.ip	Nilai diambil dari kolom log mentah 'syslog_host'.
syslog_prog	-	Tidak dipetakan.
syslog_severity	read_only_udm.security_result.severity_details	Nilai diambil dari kolom log mentah 'syslog_severity'.
syslog_severity_code	read_only_udm.security_result.severity	Nilai diambil dari kolom log mentah 'syslog_severity_code' dan dipetakan ke tingkat keparahan yang sesuai: 5, 6, 7 -> INFORMASI; 3, 4 -> SEDANG; 0, 1, 2 -> TINGGI.
syslog5424_pri	-	Tidak dipetakan.
Vlan-Id	read_only_udm.principal.resource.id	Nilai diambil dari kolom log mentah 'Vlan-Id'.
-	read_only_udm.metadata.event_type	Logika: Jika 'SIP' atau 'hostname' ada dan 'Proto' ada, tetapkan ke 'NETWORK_CONNECTION'. Jika 'SIP', 'hostname', atau 'syslog_host' ada, tetapkan ke 'STATUS_UPDATE'. Jika tidak, setel ke 'GENERIC_EVENT'.
-	read_only_udm.metadata.log_type	Logika: Setel ke 'CISCO_ACI'.
-	read_only_udm.metadata.vendor_name	Logika: Disetel ke 'Cisco'.
-	read_only_udm.metadata.product_name	Logika: Setel ke 'ACI'.

Perlu bantuan lebih lanjut? Dapatkan jawaban dari anggota Komunitas dan profesional Google SecOps.