Mengumpulkan log audit NetApp Console (sebelumnya BlueXP)

Didukung di:

Dokumen ini menjelaskan cara menyerap log audit NetApp Console (sebelumnya BlueXP) ke Google Security Operations menggunakan Google Cloud Storage V2.

NetApp Console adalah bidang kontrol terpadu untuk mengelola layanan data dan penyimpanan multi-cloud hybrid di seluruh lingkungan lokal dan cloud. Layanan Audit mencatat operasi yang dilakukan oleh layanan Console, termasuk alamat IP asal, ruang kerja, agen Console yang digunakan, dan data telemetri lainnya yang berguna untuk analisis forensik dan persyaratan kepatuhan.

Sebelum memulai

Pastikan Anda memiliki prasyarat berikut:

  • Instance Google SecOps
  • Project GCP dengan Cloud Storage, Cloud Run, Pub/Sub, dan Cloud Scheduler 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 administratif ke Konsol NetApp dengan izin untuk membuat akun layanan
  • ID akun NetApp Console Anda

Mengonfigurasi akses API Konsol NetApp

Agar Google SecOps dapat mengambil log audit, Anda perlu membuat akun layanan dengan izin yang sesuai dan membuat kredensial API.

Mendapatkan ID akun Anda

  1. Buka Konsol NetApp menggunakan browser.
  2. Login menggunakan kredensial NetApp Console atau kredensial NetApp Support Site Anda.
  3. Klik drop-down Akun di bagian atas halaman.
  4. Klik Kelola Akun untuk akun yang dipilih.
  5. Di bagian Ringkasan, salin nilai ID Akun.

Membuat akun layanan

  1. Di Konsol NetApp, buka Administration > Identity and access.
  2. Pilih Anggota.
  3. Pilih Tambahkan anggota.
  4. Untuk Jenis Anggota, pilih Akun layanan.
  5. Masukkan nama untuk akun layanan (misalnya, Google SecOps Integration).
  6. Biarkan Gunakan autentikasi JWT kunci pribadi tidak dicentang untuk menggunakan autentikasi rahasia klien.

  7. Di bagian Pilih organisasi, folder, atau project, pilih organisasi Anda.

  8. Untuk Kategori, pilih Organisasi.

  9. Untuk Role, pilih Organization viewer.

  10. Klik Tambahkan.

Mencatat kredensial API

Setelah membuat akun layanan, dialog akan menampilkan kredensial Anda:

  • Client ID: ID klien unik Anda (misalnya, TvPPs4SeM5smEElsGmdDUznljhN3YY8s)
  • Rahasia Klien: Kunci rahasia API Anda
  1. Download atau salin Client ID dan Rahasia Klien ke lokasi yang aman.
  2. Klik Close.

Memverifikasi izin

Untuk memverifikasi bahwa akun layanan memiliki izin yang diperlukan:

  1. Login ke NetApp Console.
  2. Buka Administrasi > Identitas dan akses.
  3. Pilih Anggota.
  4. Temukan akun layanan yang Anda buat dan pastikan akun tersebut memiliki peran Organization viewer.
  5. Jika peran tidak ditetapkan, klik nama akun layanan dan perbarui peran menjadi Organization Viewer.

Menguji akses API

  • Uji kredensial Anda sebelum melanjutkan integrasi:

    # Replace with your actual credentials
    CLIENT_ID="your-client-id"
    CLIENT_SECRET="your-client-secret"
    ACCOUNT_ID="your-account-id"
    
    # Obtain access token
    TOKEN=$(curl -s -X POST "https://netapp-cloud-account.auth0.com/oauth/token" \
        -H "Content-Type: application/json" \
        -d '{
            "grant_type": "client_credentials",
            "client_id": "'"${CLIENT_ID}"'",
            "client_secret": "'"${CLIENT_SECRET}"'",
            "audience": "https://api.cloud.netapp.com"
        }' | python3 -c "import sys,json; print(json.load(sys.stdin)['access_token'])")
    
    # Test audit API access
    curl -s -H "Authorization: Bearer ${TOKEN}" \
        "https://cloudmanager.cloud.netapp.com/audit/${ACCOUNT_ID}?offset=0" \
        | python3 -m json.tool
    

Respons yang berhasil akan menampilkan objek JSON yang berisi catatan audit untuk akun yang ditentukan.

Membuat bucket Google Cloud Storage

  1. Buka Konsol Google Cloud.
  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, netapp-bluexp-audit-logs)
    Location type Pilih sesuai kebutuhan Anda (Region, Dual-region, Multi-region)
    Location Pilih lokasi (misalnya, us-central1)
    Kelas penyimpanan Standar (direkomendasikan untuk log yang sering diakses)
    Kontrol akses Seragam (direkomendasikan)
    Alat perlindungan Opsional: Mengaktifkan pembuatan versi objek atau kebijakan retensi
  6. Klik Create.

Buat akun layanan untuk fungsi Cloud Run

  1. Di Konsol GCP, buka IAM & Admin > Service Accounts.
  2. Klik Create Service Account.
  3. Berikan detail konfigurasi berikut:
    • Nama akun layanan: Masukkan netapp-bluexp-collector-sa
    • Deskripsi akun layanan: Masukkan Service account for Cloud Run function to collect NetApp Console audit logs
  4. Klik Create and Continue.
  5. Di bagian Grant this service account access to 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 Done.

Memberikan izin IAM pada bucket GCS

  1. Buka Cloud Storage > Buckets.
  2. Klik nama bucket Anda (netapp-bluexp-audit-logs).
  3. Buka tab Izin.
  4. Klik Grant access.
  5. Berikan detail konfigurasi berikut:
    • Add principals: Masukkan email akun layanan (netapp-bluexp-collector-sa@PROJECT_ID.iam.gserviceaccount.com)
    • Assign roles: Pilih Storage Object Admin
  6. Klik Simpan.

Membuat topik Pub/Sub

  1. Di Konsol GCP, buka Pub/Sub > Topics.
  2. Klik Create topic.
  3. Berikan detail konfigurasi berikut:
    • ID Topik: Masukkan netapp-bluexp-audit-trigger
    • Biarkan setelan lainnya sesuai default
  4. Klik Create.

Membuat fungsi Cloud Run untuk mengumpulkan log

Fungsi Cloud Run akan dipicu oleh pesan Pub/Sub dari Cloud Scheduler untuk mengambil log audit dari NetApp Console Audit API dan menuliskannya ke GCS.

  1. Di Konsol GCP, buka Cloud Run.
  2. Klik Create service.
  3. Pilih Function (gunakan editor edit langsung untuk membuat fungsi).
  4. Di bagian Configure, berikan detail konfigurasi berikut:

    Setelan Nilai
    Nama layanan netapp-bluexp-collector
    Region 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 netapp-bluexp-audit-trigger.
    4. Klik Simpan.
  6. Di bagian Authentication:

    1. Pilih Wajibkan autentikasi.
    2. Periksa Identity and Access Management (IAM).
  7. Scroll ke bawah dan luaskan Containers, Networking, Security.

  8. Buka tab Security:

    • Akun layanan: Pilih netapp-bluexp-collector-sa
  9. Buka tab Containers:

    1. Klik Variables & Secrets.
    2. Klik + Add variable untuk setiap variabel lingkungan:
    Nama Variabel Nilai Contoh Deskripsi
    GCS_BUCKET netapp-bluexp-audit-logs Nama bucket GCS
    GCS_PREFIX netapp-bluexp-audit Awalan untuk file log
    STATE_KEY netapp-bluexp-audit/state.json Jalur file status
    NETAPP_CLIENT_ID your-client-id Client ID akun layanan NetApp Console
    NETAPP_CLIENT_SECRET your-client-secret Rahasia Klien akun layanan NetApp Console
    NETAPP_ACCOUNT_ID account-AbCdEfGh ID akun NetApp Console
    MAX_RECORDS 5000 Jumlah data maks per proses
    LOOKBACK_HOURS 24 Periode lihat balik awal
  10. Di bagian Variables & Secrets, scroll ke bawah ke Requests:

    • Waktu tunggu permintaan: Masukkan 600 detik (10 menit)
  11. Buka tab Setelan:

    • Di bagian Referensi:
      • Memori: Pilih 512 MiB atau lebih tinggi
      • CPU: Pilih 1
  12. Di bagian Penskalaan revisi:

    • Minimum number of instances: Masukkan 0
    • Maximum number of instances: Masukkan 100
  13. Klik Create.

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

  15. 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:

    • main.py:
    import functions_framework
    from google.cloud import storage
    import json
    import os
    import urllib3
    from datetime import datetime, timezone, timedelta
    import time
    
    http = urllib3.PoolManager(
      timeout=urllib3.Timeout(connect=5.0, read=30.0),
      retries=False,
    )
    
    storage_client = storage.Client()
    
    GCS_BUCKET = os.environ.get('GCS_BUCKET')
    GCS_PREFIX = os.environ.get('GCS_PREFIX', 'netapp-bluexp-audit')
    STATE_KEY = os.environ.get('STATE_KEY', 'netapp-bluexp-audit/state.json')
    NETAPP_CLIENT_ID = os.environ.get('NETAPP_CLIENT_ID')
    NETAPP_CLIENT_SECRET = os.environ.get('NETAPP_CLIENT_SECRET')
    NETAPP_ACCOUNT_ID = os.environ.get('NETAPP_ACCOUNT_ID')
    MAX_RECORDS = int(os.environ.get('MAX_RECORDS', '5000'))
    LOOKBACK_HOURS = int(os.environ.get('LOOKBACK_HOURS', '24'))
    
    TOKEN_URL = 'https://netapp-cloud-account.auth0.com/oauth/token'
    API_BASE = 'https://cloudmanager.cloud.netapp.com'
    AUDIENCE = 'https://api.cloud.netapp.com'
    
    def to_unix_millis(dt):
      if dt.tzinfo is None:
        dt = dt.replace(tzinfo=timezone.utc)
      dt = dt.astimezone(timezone.utc)
      return int(dt.timestamp() * 1000)
    
    def parse_datetime(value):
      if value.endswith('Z'):
        value = value[:-1] + '+00:00'
      return datetime.fromisoformat(value)
    
    def get_access_token():
      body = json.dumps({
        'grant_type': 'client_credentials',
        'client_id': NETAPP_CLIENT_ID,
        'client_secret': NETAPP_CLIENT_SECRET,
        'audience': AUDIENCE,
      }).encode('utf-8')
    
      response = http.request(
        'POST', TOKEN_URL,
        body=body,
        headers={'Content-Type': 'application/json'}
      )
    
      if response.status != 200:
        raise Exception(
          f"Token request failed: {response.status} - "
          f"{response.data.decode('utf-8')}"
        )
    
      data = json.loads(response.data.decode('utf-8'))
      token = data.get('access_token')
      if not token:
        raise Exception('No access_token in token response')
    
      print('Successfully obtained NetApp Console access token')
      return token
    
    @functions_framework.cloud_event
    def main(cloud_event):
      if not all([GCS_BUCKET, NETAPP_CLIENT_ID, NETAPP_CLIENT_SECRET, NETAPP_ACCOUNT_ID]):
        print('Error: Missing required environment variables')
        return
    
      try:
        bucket = storage_client.bucket(GCS_BUCKET)
        state = load_state(bucket)
        now = datetime.now(timezone.utc)
    
        if isinstance(state, dict) and state.get('last_event_time'):
          try:
            last_val = state['last_event_time']
            if last_val.endswith('Z'):
              last_val = last_val[:-1] + '+00:00'
            last_time = datetime.fromisoformat(last_val)
            last_time = last_time - timedelta(minutes=2)
          except Exception as e:
            print(f'Warning: Could not parse last_event_time: {e}')
            last_time = now - timedelta(hours=LOOKBACK_HOURS)
        else:
          last_time = now - timedelta(hours=LOOKBACK_HOURS)
    
        print(f'Fetching audit logs from {last_time.isoformat()} to {now.isoformat()}')
    
        token = get_access_token()
    
        records, newest_time = fetch_audit_logs(token, last_time, now)
    
        if not records:
          print('No new audit records found.')
          save_state(bucket, now.isoformat())
          return
    
        timestamp = now.strftime('%Y%m%d_%H%M%S')
        object_key = f'{GCS_PREFIX}/netapp_bluexp_audit_{timestamp}.ndjson'
        blob = bucket.blob(object_key)
    
        ndjson = '\n'.join(
          [json.dumps(r, ensure_ascii=False, default=str) for r in records]
        ) + '\n'
        blob.upload_from_string(ndjson, content_type='application/x-ndjson')
    
        print(f'Wrote {len(records)} records to gs://{GCS_BUCKET}/{object_key}')
    
        save_state(bucket, newest_time if newest_time else now.isoformat())
    
        print(f'Successfully processed {len(records)} audit records')
    
      except Exception as e:
        print(f'Error processing audit logs: {str(e)}')
        raise
    
    def fetch_audit_logs(token, start_time, end_time):
      endpoint = f'{API_BASE}/audit/{NETAPP_ACCOUNT_ID}'
    
      headers = {
        'Authorization': f'Bearer {token}',
        'Accept': 'application/json',
        'User-Agent': 'GoogleSecOps-NetAppBlueXPCollector/1.0',
      }
    
      start_millis = to_unix_millis(start_time)
      end_millis = to_unix_millis(end_time)
    
      records = []
      newest_time = None
      offset = 0
      page_num = 0
      backoff = 1.0
    
      while True:
        page_num += 1
    
        if len(records) >= MAX_RECORDS:
          print(f'Reached max_records limit ({MAX_RECORDS})')
          break
    
        url = (
          f'{endpoint}'
          f'?fromLastModified={start_millis}'
          f'&toLastModified={end_millis}'
          f'&offset={offset}'
        )
    
        try:
          response = http.request('GET', url, headers=headers)
    
          if response.status == 429:
            retry_after = int(response.headers.get('Retry-After', str(int(backoff))))
            print(f'Rate limited (429). Retrying after {retry_after}s...')
            time.sleep(retry_after)
            backoff = min(backoff * 2, 30.0)
            continue
    
          backoff = 1.0
    
          if response.status != 200:
            print(f'HTTP Error: {response.status}')
            response_text = response.data.decode('utf-8')
            print(f'Response body: {response_text}')
            return [], None
    
          data = json.loads(response.data.decode('utf-8'))
    
          page_results = data.get('auditEntries', [])
    
          if not page_results:
            print('No more results (empty page)')
            break
    
          print(f'Page {page_num}: Retrieved {len(page_results)} audit records')
          records.extend(page_results)
    
          for entry in page_results:
            try:
              last_modified = entry.get('lastModified')
              if last_modified:
                entry_dt = datetime.fromtimestamp(
                  last_modified / 1000, tz=timezone.utc
                )
                entry_time = entry_dt.isoformat()
                if newest_time is None or parse_datetime(entry_time) > parse_datetime(newest_time):
                  newest_time = entry_time
            except Exception as e:
              print(f'Warning: Could not parse entry time: {e}')
    
          count = data.get('count', len(page_results))
          if count < 100:
            print(f'Reached last page (count={count})')
            break
    
          offset += count
    
        except Exception as e:
          print(f'Error fetching audit logs: {e}')
          return [], None
    
      print(f'Retrieved {len(records)} total audit records from {page_num} pages')
      return records, newest_time
    
    def load_state(bucket):
      try:
        blob = bucket.blob(STATE_KEY)
        if blob.exists():
          return json.loads(blob.download_as_text())
      except Exception as e:
        print(f'Warning: Could not load state: {e}')
      return {}
    
    def save_state(bucket, last_event_time_iso):
      try:
        state = {
          'last_event_time': last_event_time_iso,
          'last_run': datetime.now(timezone.utc).isoformat(),
        }
        blob = bucket.blob(STATE_KEY)
        blob.upload_from_string(
          json.dumps(state, indent=2),
          content_type='application/json'
        )
        print(f'Saved state: last_event_time={last_event_time_iso}')
      except Exception as e:
        print(f'Warning: Could not save state: {e}')
    
    • 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).

Buat tugas Cloud Scheduler

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

  1. Di Konsol GCP, buka Cloud Scheduler.
  2. Klik Create Job.
  3. Berikan detail konfigurasi berikut:

    Setelan Nilai
    Nama netapp-bluexp-collector-hourly
    Region Pilih region yang sama dengan fungsi Cloud Run
    Frekuensi 0 * * * * (setiap jam, tepat pada waktunya)
    Zona Waktu Pilih zona waktu (UTC direkomendasikan)
    Jenis target Pub/Sub
    Topik Pilih netapp-bluexp-audit-trigger
    Isi pesan {} (objek JSON kosong)
  4. Klik Create.

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
Setiap jam 0 * * * * Standar (direkomendasikan)
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 (netapp-bluexp-collector-hourly).
  2. Klik Force run untuk memicu tugas secara manual.
  3. Tunggu beberapa detik.
  4. Buka Cloud Run > Services.
  5. Klik netapp-bluexp-collector.
  6. Klik tab Logs.
  7. Pastikan fungsi berhasil dieksekusi. Cari:

    Fetching audit logs from YYYY-MM-DDTHH:MM:SS+00:00 to YYYY-MM-DDTHH:MM:SS+00:00
    Successfully obtained NetApp Console access token
    Page 1: Retrieved X audit records
    Wrote X records to gs://netapp-bluexp-audit-logs/netapp-bluexp-audit/netapp_bluexp_audit_YYYYMMDD_HHMMSS.ndjson
    Successfully processed X audit records
    
  8. Buka Cloud Storage > Buckets.

  9. Klik netapp-bluexp-audit-logs.

  10. Buka folder netapp-bluexp-audit/.

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

Jika Anda melihat error di log:

  • HTTP 401: Pastikan variabel lingkungan NETAPP_CLIENT_ID dan NETAPP_CLIENT_SECRET sudah benar
  • HTTP 403: Verifikasi bahwa akun layanan memiliki peran Organization Viewer di Konsol NetApp
  • HTTP 429: Pembatasan kapasitas — fungsi akan otomatis mencoba ulang dengan backoff eksponensial
  • Variabel lingkungan tidak ada: Pastikan semua variabel yang diperlukan ditetapkan dalam konfigurasi fungsi Cloud Run

Mengambil akun layanan Google SecOps

  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, NetApp Console Audit Logs).
  5. Pilih Google Cloud Storage V2 sebagai Source type.
  6. Pilih NetApp Console (sebelumnya BlueXP) sebagai Jenis log.
  7. Klik Get Service Account. Email akun layanan yang unik akan ditampilkan, misalnya:

    chronicle-12345678@chronicle-gcp-prod.iam.gserviceaccount.com
    
  8. Salin alamat email ini untuk digunakan di langkah berikutnya.

  9. Klik Berikutnya.

  10. Tentukan nilai untuk parameter input berikut:

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

      gs://netapp-bluexp-audit-logs/netapp-bluexp-audit/
      
    • 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.

    • Usia File Maksimum: Menyertakan file yang diubah dalam beberapa hari terakhir (default adalah 180 hari)

    • Namespace aset: Namespace aset

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

  11. Klik Berikutnya.

  12. Tinjau konfigurasi feed berita baru Anda di layar Finalize, lalu klik Submit.

Memberikan izin IAM ke akun layanan Google SecOps

  1. Buka Cloud Storage > Buckets.
  2. Klik netapp-bluexp-audit-logs.
  3. Buka tab Izin.
  4. Klik Grant access.
  5. Berikan detail konfigurasi berikut:
    • Add principals: Tempel email akun layanan Google SecOps
    • Assign roles: Pilih Storage Object Viewer
  6. Klik Simpan.

Tabel pemetaan UDM

Kolom Log Pemetaan UDM Logika
endTime additional.fields Digabungkan dengan label dari setiap item jika tidak kosong
responseData additional.fields
weHash additional.fields
wePublicId additional.fields
hasFailedRecord additional.fields
tindakan additional.fields
pelanggan additional.fields
hasRecords additional.fields
workspaceId additional.fields
lastModified additional.fields
startTime additional.fields
agentName metadata.event_type Setel ke STATUS_UPDATE jika agentName tidak kosong, jika tidak, GENERIC_EVENT
agentId metadata.product_deployment_id Nilai disalin langsung jika tidak kosong
requestId metadata.product_log_id Nilai disalin langsung jika tidak kosong
jaringan jaringan Diganti namanya secara langsung jika tidak kosong
agentName principal.asset.hostname Nilai disalin langsung jika tidak kosong
fileName principal.file.names Digabungkan jika tidak kosong
agentName principal.hostname Nilai disalin langsung jika tidak kosong
accountId principal.resource.id Nilai disalin langsung jika tidak kosong
resourceName principal.resource.name Nilai disalin langsung jika tidak kosong
accountId principal.resource.product_object_id Nilai dari accountId jika tidak kosong, atau dari resourceId jika tidak kosong
resourceId principal.resource.product_object_id
principalId principal.user.userid Nilai disalin langsung jika tidak kosong
status security_result.action Disetel ke ALLOW jika status =~ (?i)SUCCESS, BLOCK jika dalam FAILURE atau UNSUCCESSFUL_ATTEMPT
status security_result.action_details Nilai disalin langsung jika tidak kosong
target target Diganti namanya secara langsung jika tidak kosong

Log Perubahan

Melihat Log Perubahan untuk parser ini

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