Mengumpulkan log Oracle NetSuite

Didukung di:

Google SecOps SIEM

Dokumen ini menjelaskan cara menyerap log Oracle NetSuite ke Google Security Operations menggunakan Cloud Storage V2.

Oracle NetSuite adalah platform perencanaan sumber daya perusahaan (ERP) berbasis cloud yang menyediakan kemampuan pengelolaan bisnis yang komprehensif, termasuk pengelolaan keuangan, CRM, e-commerce, dan pengelolaan inventaris. NetSuite membuat log audit untuk login pengguna, perubahan sistem, dan aktivitas API yang dapat dikueri melalui SuiteQL REST API dan diserap ke Google SecOps untuk pemantauan keamanan dan kepatuhan.

Sebelum memulai

Pastikan Anda memiliki prasyarat berikut:

• Instance Google SecOps
• Project Google Cloud dengan Cloud Storage, Cloud Run, Pub/Sub, dan Cloud Scheduler API yang diaktifkan
• Izin untuk membuat dan mengelola bucket Cloud Storage
• Izin untuk mengelola kebijakan Identity and Access Management (IAM) di bucket Cloud Storage
• Izin untuk membuat layanan Cloud Run, topik Pub/Sub, dan tugas Cloud Scheduler
• Akses administrator ke akun Oracle NetSuite Anda
• Fitur Layanan Web REST dan OAuth 2.0 diaktifkan di akun NetSuite Anda
• ID akun NetSuite Anda (dapat ditemukan di Setup > Company > Company Information)

Mengaktifkan fitur NetSuite yang diperlukan

1. Login ke akun NetSuite Anda sebagai Administrator.
2. Buka Setup > Company > Enable Features.
3. Klik subtab SuiteCloud.
4. Di bagian SuiteTalk (Web Services), pilih kotak berikut:
   • Layanan Web REST
5. Di bagian Manage Authentication, centang kotak berikut:
   • OAuth 2.0
6. Klik Simpan.

Mengonfigurasi kredensial OAuth 2.0 Oracle NetSuite

Agar fungsi Cloud Run dapat mengambil log audit dari NetSuite, Anda perlu membuat data integrasi dengan autentikasi kredensial klien OAuth 2.0.

Membuat data integrasi

1. Login ke akun NetSuite Anda sebagai Administrator.
2. Buka Setup > Integration > Manage Integrations > New.
3. Di kolom Name, masukkan Google SecOps Integration.
4. Di kolom Description, masukkan Integration with Google SecOps for audit log ingestion.
5. Di kolom State, pilih Enabled.
6. Klik subtab Authentication.
7. Di bagian OAuth 2.0, konfigurasikan hal berikut:
   • Pilih Client Credentials (Machine to Machine) Grant
   • Pilih REST Web Services
8. Hapus centang pada semua kotak di bagian Token-based Authentication.
9. Klik Simpan.

Mencatat kredensial OAuth 2.0

Setelah menyimpan data integrasi, NetSuite akan menampilkan kredensial OAuth 2.0 Anda:

• Client ID: ID klien unik Anda

• Rahasia Klien: Kunci rahasia API Anda

  Catatan: Segera salin dan simpan Client ID dan Client Secret. NetSuite hanya menampilkan nilai ini sekali. Jika Anda kehilangan kredensial tersebut, Anda harus mereset kredensial pada data integrasi.

Membuat sertifikat untuk kredensial klien OAuth 2.0

Alur kredensial klien OAuth 2.0 memerlukan sertifikat untuk penandatanganan JWT. Buat sertifikat ECDSA menggunakan OpenSSL:

1. Di komputer yang telah menginstal OpenSSL, jalankan perintah berikut untuk membuat kunci pribadi dan sertifikat:

   openssl req -new -x509 -newkey ec \
       -pkeyopt ec_paramgen_curve:prime256v1 \
       -pkeyopt ec_param_enc:named_curve \
       -nodes -days 730 \
       -out public.pem -keyout private.pem
   

2. Saat diminta, masukkan kolom subjek sertifikat (Negara, Negara Bagian, Organisasi, dan sebagainya). Untuk Common Name, masukkan Google SecOps Integration.

3. Perintah ini akan menghasilkan dua file:

   • private.pem: Kunci pribadi (jaga keamanannya; digunakan untuk penandatanganan JWT di fungsi Cloud Run)
   • public.pem: Sertifikat publik (upload ke NetSuite)
   Catatan: NetSuite mendukung sertifikat ECDSA (ES256, ES512) dan RSA-PSS (PS256). Dokumen ini menggunakan ECDSA dengan kurva prime256v1 (ES256). Sertifikat memiliki masa berlaku maksimum 2 tahun.

Membuat pemetaan kredensial klien OAuth 2.0

1. Di NetSuite, buka Setup > Integration > Manage Authentication > OAuth 2.0 Client Credentials (M2M) Setup.
2. Klik Buat Baru.
3. Dalam dialog, konfigurasi setelan berikut:
   • Entity: Pilih karyawan atau pengguna yang akan dikaitkan dengan permintaan API.
   • Peran: Pilih peran yang memiliki izin Login menggunakan Token Akses OAuth 2.0 (lihat Izin yang diperlukan). Tanpa izin ini, peran tidak akan muncul dalam daftar ini.
   • Application: Pilih Google SecOps Integration.
4. Klik Choose File dan upload file sertifikat public.pem.
5. Klik Simpan.
6. Setelah menyimpan, catat nilai Certificate ID yang ditampilkan dalam daftar pemetaan. Nilai ini digunakan sebagai kid (ID Kunci) di header JWT.

Izin yang diperlukan

Peran yang ditetapkan ke pemetaan OAuth 2.0 harus memiliki izin berikut:

Izin	Tingkat Akses	Tujuan
Login menggunakan Token Akses OAuth 2.0	Penuh	Mengizinkan peran menggunakan login token OAuth 2.0
Layanan Web REST	Penuh	Menjalankan permintaan REST API
Jejak Audit Login	Penuh	Mengakses data jejak audit login
SuiteAnalytics Workbook	Lihat	Mengakses data analisis melalui SuiteQL
Catatan Sistem untuk Analytics dan REST	Lihat	Membuat kueri data SystemNote melalui SuiteQL

Catatan: Tanpa System Notes for Analytics and REST, kueri SuiteQL di SystemNote hanya menampilkan catatan sistem untuk perubahan yang dibuat sendiri oleh pengguna integrasi. Permintaan masih berhasil, sehingga tidak ada yang melaporkan error, sementara data perubahan tidak pernah sampai.

Membuat peran khusus untuk integrasi

1. Buka Setup > Users/Roles > Manage Roles > New.
2. Di kolom Name, masukkan Google SecOps Integration Role.
3. Klik subtab Permissions.
4. Klik sub-subtab Penyiapan.
5. Tambahkan izin berikut:
   • Log in using OAuth 2.0 Access Tokens: Tetapkan ke Full
   • Layanan Web REST: Tetapkan ke Penuh
   • Jejak Audit Login: Tetapkan ke Full
6. Klik sub-subtab Laporan.
7. Tambahkan izin berikut:
   • SuiteAnalytics Workbook: Tetapkan ke View
8. Klik sub-subtab Daftar.
9. Tambahkan izin berikut:
   • System Notes for Analytics and REST: Tetapkan ke View
10. Klik Simpan.

Memverifikasi izin

Untuk memverifikasi bahwa peran memiliki izin yang diperlukan, lakukan hal berikut:

1. Login ke NetSuite.
2. Buka Setup > Users/Roles > Manage Roles.
3. Temukan Google SecOps Integration Role, lalu klik Edit.
4. Pastikan izin yang tercantum di Izin yang diperlukan ada di subtab Izin.

Menguji akses API

• Uji kredensial Anda sebelum melanjutkan integrasi:

  # Replace with your actual values
  ACCOUNT_ID="your-account-id"
  CLIENT_ID="your-client-id"
  CERTIFICATE_ID="your-certificate-id"
  PRIVATE_KEY_FILE="private.pem"
  
  # Generate JWT (requires Python 3 with PyJWT and cryptography)
  pip install PyJWT cryptography
  
  TOKEN=$(python3 -c "
  import jwt, time, json
  with open('${PRIVATE_KEY_FILE}', 'r') as f:
      key = f.read()
  now = int(time.time())
  payload = {
      'iss': '${CLIENT_ID}',
      'scope': ['restlets', 'rest_webservices'],
      'aud': 'https://${ACCOUNT_ID}.suitetalk.api.netsuite.com/services/rest/auth/oauth2/v1/token',
      'iat': now,
      'exp': now + 3600
  }
  print(jwt.encode(payload, key, algorithm='ES256', headers={'kid': '${CERTIFICATE_ID}', 'typ': 'JWT'}))
  ")
  
  # Obtain access token
  ACCESS_TOKEN=$(curl -s -X POST \
      "https://${ACCOUNT_ID}.suitetalk.api.netsuite.com/services/rest/auth/oauth2/v1/token" \
      -H "Content-Type: application/x-www-form-urlencoded" \
      -d "grant_type=client_credentials&client_assertion_type=urn%3Aietf%3Aparams%3Aoauth%3Aclient-assertion-type%3Ajwt-bearer&client_assertion=${TOKEN}" \
      | python3 -c "import sys,json; print(json.load(sys.stdin)['access_token'])")
  
  # Test SuiteQL query against the login audit trail
  curl -s -X POST \
      "https://${ACCOUNT_ID}.suitetalk.api.netsuite.com/services/rest/query/v1/suiteql?limit=5" \
      -H "Authorization: Bearer ${ACCESS_TOKEN}" \
      -H "Content-Type: application/json" \
      -H "Prefer: transient" \
      -d '{"q": "SELECT Date, Status, User, IPAddress FROM LoginAudit ORDER BY Date DESC"}' \
      | python3 -m json.tool
  
  # Test SuiteQL query against system notes
  curl -s -X POST \
      "https://${ACCOUNT_ID}.suitetalk.api.netsuite.com/services/rest/query/v1/suiteql?limit=5" \
      -H "Authorization: Bearer ${ACCESS_TOKEN}" \
      -H "Content-Type: application/json" \
      -H "Prefer: transient" \
      -d '{"q": "SELECT Date, RecordID, Field, BUILTIN.DF(SystemNote.Name) AS EmployeeName FROM SystemNote ORDER BY Date DESC"}' \
      | python3 -m json.tool
  

Respons yang berhasil akan menampilkan objek JSON dengan array items yang berisi data audit login.

Kueri kedua harus menampilkan catatan sistem terbaru, termasuk perubahan yang dibuat oleh pengguna selain pengguna integrasi. Array items yang kosong, atau yang hanya mencantumkan pengguna integrasi di EmployeeName, berarti peran tersebut tidak memiliki izin System Notes for Analytics and REST.

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
   Memberi nama bucket Anda	Masukkan nama yang unik secara global (misalnya, netsuite-audit-logs)
   Location type	Pilih sesuai kebutuhan Anda (Region, Dual-region, Multi-region)
   Location	Pilih lokasi (misalnya, us-central1)
   Storage class	Standard (direkomendasikan untuk log yang sering diakses)
   Kontrol akses	Seragam (direkomendasikan)
   Alat perlindungan	Opsional: Mengaktifkan pembuatan versi objek atau kebijakan retensi

6. Klik Create.

Membuat akun layanan untuk fungsi Cloud Run

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

Membuat akun layanan

1. Di konsol Google Cloud , buka IAM & Admin > Service Accounts.
2. Klik Create Service Account.
3. Berikan detail konfigurasi berikut:
   • Service account name: Masukkan netsuite-audit-collector-sa
   • Service account description: Masukkan Service account for Cloud Run function to collect Oracle NetSuite 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.

Peran ini diperlukan untuk:

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

Memberikan izin IAM pada bucket Cloud Storage

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

Membuat topik Pub/Sub

Buat topik Pub/Sub yang akan dipublikasikan oleh Cloud Scheduler dan di-subscribe oleh fungsi Cloud Run.

1. Di Konsol GCP, buka Pub/Sub > Topics.
2. Klik Create topic.
3. Berikan detail konfigurasi berikut:
   • ID Topik: Masukkan netsuite-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 Oracle NetSuite SuiteQL REST API dan menulisnya ke Cloud Storage.

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

4. Di bagian Configure, berikan detail konfigurasi berikut:

   Setelan	Nilai
   Nama layanan	netsuite-audit-collector
   Region	Pilih region yang cocok dengan bucket Cloud Storage Anda (misalnya, us-central1)
   Runtime	Pilih Python 3.12 atau yang lebih baru

5. Di bagian Trigger (optional):

   1. Klik + Tambahkan pemicu.
   2. Pilih Cloud Pub/Sub.
   3. Di Select a Cloud Pub/Sub topic, pilih netsuite-audit-trigger.
   4. Klik Simpan.

6. Di bagian Authentication:

   1. Pilih Wajibkan autentikasi.
   2. Pilih Identity and Access Management (IAM).
   Catatan: Pub/Sub akan otomatis menangani autentikasi saat memanggil fungsi.

7. Scroll ke bawah dan luaskan Containers, Networking, Security.

8. Buka tab Keamanan:

   • Service account: Pilih netsuite-audit-collector-sa

9. Buka tab Penampung:

   1. Klik Variables & Secrets.
   2. Klik + Add variable untuk setiap variabel lingkungan:

   Nama Variabel	Nilai Contoh	Deskripsi
   GCS_BUCKET	netsuite-audit-logs	Nama bucket Cloud Storage
   GCS_PREFIX	netsuite-audit	Awalan untuk file log
   STATE_KEY	netsuite-audit-state.json	Jalur status, di luar awalan log
   NS_ACCOUNT_ID	123456	ID akun NetSuite
   NS_CLIENT_ID	your-client-id	Client ID OAuth 2.0
   NS_CERTIFICATE_ID	your-certificate-id	ID sertifikat (kid) dari pemetaan M2M
   NS_PRIVATE_KEY	-----BEGIN EC PRIVATE KEY-----...	Konten kunci pribadi ECDSA
   LOOKBACK_HOURS	24	Periode lihat balik awal
   PAGE_SIZE	1000	Data per halaman SuiteQL (maks. 1.000)
   MAX_RECORDS	50000	Data maksimum per operasi

   Catatan: STATE_KEY sengaja mengarah ke luar awalan log. Feed menyerap setiap objek di URI bucket-nya dan, jika opsi penghapusan dipilih, akan menghapus objek yang ditransfer. Menjaga state file di luar awalan tersebut akan mencegahnya diserap sebagai data log dan dihapus, yang akan mereset koleksi dan menyerap ulang peristiwa duplikat.

10. Di bagian Variables & Secrets, scroll ke bawah ke Requests:

    • Request timeout: Masukkan 600 detik (10 menit)

11. Buka tab Setelan:

    • Di bagian Resources:
      • Memory: Pilih 512 MiB atau lebih tinggi
      • CPU: Pilih 1

12. Di bagian Revision scaling:

    • 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:

   • File pertama - main.py:

     import functions_framework
     from google.cloud import storage
     import json
     import os
     import urllib3
     import urllib.parse
     import jwt
     import time
     import hashlib
     from datetime import datetime, timezone, timedelta
     
     http = urllib3.PoolManager(
         timeout=urllib3.Timeout(connect=10.0, read=60.0),
         retries=False,
     )
     
     storage_client = storage.Client()
     
     GCS_BUCKET = os.environ.get('GCS_BUCKET')
     GCS_PREFIX = os.environ.get('GCS_PREFIX', 'netsuite-audit')
     STATE_KEY = os.environ.get('STATE_KEY', 'netsuite-audit-state.json')
     NS_ACCOUNT_ID = os.environ.get('NS_ACCOUNT_ID')
     NS_CLIENT_ID = os.environ.get('NS_CLIENT_ID')
     NS_CERTIFICATE_ID = os.environ.get('NS_CERTIFICATE_ID')
     NS_PRIVATE_KEY = os.environ.get('NS_PRIVATE_KEY', '').replace('\\n', '\n')
     LOOKBACK_HOURS = int(os.environ.get('LOOKBACK_HOURS', '24'))
     PAGE_SIZE = int(os.environ.get('PAGE_SIZE', '1000'))
     MAX_RECORDS = int(os.environ.get('MAX_RECORDS', '50000'))
     
     # Neither SuiteQL query selects a unique row identifier, so the content hash
     # is the only identity available.
     ID_FIELDS = ()
     
     def event_keys(event):
         """Return every identity this event can be recognised by."""
         keys = {'sha256:' + hashlib.sha256(
             json.dumps(event, sort_keys=True, ensure_ascii=False).encode('utf-8')
         ).hexdigest()}
         for field in ID_FIELDS:
             value = event.get(field)
             if value:
                 keys.add(f'{field}:{value}')
         return keys
     
     @functions_framework.cloud_event
     def main(cloud_event):
         if not all([GCS_BUCKET, NS_ACCOUNT_ID, NS_CLIENT_ID, NS_CERTIFICATE_ID, NS_PRIVATE_KEY]):
             print('Error: Missing required environment variables')
             return
     
         try:
             bucket = storage_client.bucket(GCS_BUCKET)
             state = load_state(bucket)
             seen_keys = load_seen(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 logs from {last_time.isoformat()} to {now.isoformat()}")
     
             access_token = get_access_token()
     
             login_audits = fetch_login_audit(access_token, last_time, now)
             system_notes = fetch_system_notes(access_token, last_time, now)
     
             all_records = []
             for record in login_audits:
                 record['_netsuite_record_type'] = 'login_audit'
                 all_records.append(record)
             for record in system_notes:
                 record['_netsuite_record_type'] = 'system_note'
                 all_records.append(record)
     
             if not all_records:
                 print("No new records found.")
                 save_state(bucket, now.isoformat())
                 return
     
             # Drop the rows the overlap window re-read from the previous run
             fresh = [r for r in all_records if not (event_keys(r) & seen_keys)]
             all_keys = set()
             for record in all_records:
                 all_keys |= event_keys(record)
     
             if not fresh:
                 print("No new records found. Watermark left unchanged.")
                 return
     
             timestamp = now.strftime('%Y%m%d_%H%M%S')
             object_key = f"{GCS_PREFIX}/netsuite_audit_{timestamp}.ndjson"
             blob = bucket.blob(object_key)
     
             ndjson = '\n'.join(
                 [json.dumps(r, ensure_ascii=False, default=str) for r in fresh]
             ) + '\n'
             blob.upload_from_string(ndjson, content_type='application/x-ndjson')
     
             print(f"Wrote {len(fresh)} records to gs://{GCS_BUCKET}/{object_key}")
     
             # Remember every fetched row so the next overlap window skips it
             save_seen(bucket, all_keys)
     
             newest = find_newest_time(login_audits, system_notes)
             save_state(bucket, newest if newest else now.isoformat())
     
             print(f"Successfully processed {len(fresh)} records "
                     f"(login_audit: {len(login_audits)}, system_note: {len(system_notes)})")
     
         except Exception as e:
             print(f'Error processing logs: {str(e)}')
             raise
     
     def get_access_token():
         token_url = (
             f"https://{NS_ACCOUNT_ID}.suitetalk.api.netsuite.com"
             f"/services/rest/auth/oauth2/v1/token"
         )
     
         now = int(time.time())
         payload = {
             'iss': NS_CLIENT_ID,
             'scope': ['restlets', 'rest_webservices'],
             'aud': token_url,
             'iat': now,
             'exp': now + 3600,
         }
     
         client_assertion = jwt.encode(
             payload,
             NS_PRIVATE_KEY,
             algorithm='ES256',
             headers={'kid': NS_CERTIFICATE_ID, 'typ': 'JWT'},
         )
     
         body = urllib.parse.urlencode({
             'grant_type': 'client_credentials',
             'client_assertion_type': 'urn:ietf:params:oauth:client-assertion-type:jwt-bearer',
             'client_assertion': client_assertion,
         }).encode('utf-8')
     
         response = http.request(
             'POST', token_url,
             body=body,
             headers={'Content-Type': 'application/x-www-form-urlencoded'},
         )
     
         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'))
         access_token = data.get('access_token')
         if not access_token:
             raise Exception("No access_token in token response")
     
         print("Successfully obtained NetSuite OAuth 2.0 access token")
         return access_token
     
     def run_suiteql(access_token, query, page_size=1000, max_records=50000):
         base_url = (
             f"https://{NS_ACCOUNT_ID}.suitetalk.api.netsuite.com"
             f"/services/rest/query/v1/suiteql"
         )
     
         headers = {
             'Authorization': f'Bearer {access_token}',
             'Content-Type': 'application/json',
             'Prefer': 'transient',
         }
     
         all_items = []
         offset = 0
         backoff = 1.0
     
         while True:
             if len(all_items) >= max_records:
                 print(f"Reached max_records limit ({max_records})")
                 break
     
             url = f"{base_url}?limit={page_size}&offset={offset}"
             body = json.dumps({'q': query}).encode('utf-8')
     
             try:
                 response = http.request('POST', url, body=body, 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, 60.0)
                     continue
     
                 backoff = 1.0
     
                 if response.status != 200:
                     print(f"SuiteQL error: {response.status} - "
                             f"{response.data.decode('utf-8')}")
                     break
     
                 data = json.loads(response.data.decode('utf-8'))
                 items = data.get('items', [])
     
                 if not items:
                     break
     
                 all_items.extend(items)
                 page_num = (offset // page_size) + 1
                 print(f"Page {page_num}: Retrieved {len(items)} records "
                         f"(total: {len(all_items)})")
     
                 has_more = data.get('hasMore', False)
                 if not has_more:
                     break
     
                 offset += len(items)
     
             except Exception as e:
                 print(f"Error executing SuiteQL query: {e}")
                 break
     
         return all_items
     
     def fetch_login_audit(access_token, start_time, end_time):
         start_str = start_time.strftime('%m/%d/%Y %H:%M:%S')
         end_str = end_time.strftime('%m/%d/%Y %H:%M:%S')
     
         query = (
             "SELECT "
             "LoginAudit.Date, "
             "LoginAudit.Status, "
             "LoginAudit.User, "
             "BUILTIN.DF(LoginAudit.User) AS Username, "
             "LoginAudit.EmailAddress, "
             "LoginAudit.Role, "
             "BUILTIN.DF(LoginAudit.Role) AS Rolename, "
             "LoginAudit.IPAddress, "
             "LoginAudit.UserAgent, "
             "LoginAudit.Detail, "
             "LoginAudit.oAuthAppName, "
             "LoginAudit.oAuthAccessTokenName, "
             "LoginAudit.requestUri "
             "FROM LoginAudit "
             f"WHERE LoginAudit.Date >= TO_DATE('{start_str}', 'MM/DD/YYYY HH24:MI:SS') "
             f"AND LoginAudit.Date <= TO_DATE('{end_str}', 'MM/DD/YYYY HH24:MI:SS') "
             "ORDER BY LoginAudit.Date ASC"
         )
     
         print("Fetching LoginAudit records...")
         return run_suiteql(access_token, query, PAGE_SIZE, MAX_RECORDS)
     
     def fetch_system_notes(access_token, start_time, end_time):
         start_str = start_time.strftime('%m/%d/%Y %H:%M:%S')
         end_str = end_time.strftime('%m/%d/%Y %H:%M:%S')
     
         query = (
             "SELECT "
             "SystemNote.Date, "
             "SystemNote.RecordTypeID, "
             "BUILTIN.DF(SystemNote.RecordTypeID) AS RecordType, "
             "SystemNote.RecordID, "
             "SystemNote.Field, "
             "SystemNote.OldValue, "
             "SystemNote.NewValue, "
             "SystemNote.Name AS EmployeeID, "
             "BUILTIN.DF(SystemNote.Name) AS EmployeeName, "
             "SystemNote.Role, "
             "BUILTIN.DF(SystemNote.Role) AS RoleName, "
             "SystemNote.Context, "
             "BUILTIN.DF(SystemNote.Context) AS ContextName "
             "FROM SystemNote "
             f"WHERE SystemNote.Date >= TO_DATE('{start_str}', 'MM/DD/YYYY HH24:MI:SS') "
             f"AND SystemNote.Date <= TO_DATE('{end_str}', 'MM/DD/YYYY HH24:MI:SS') "
             "ORDER BY SystemNote.Date ASC"
         )
     
         print("Fetching SystemNote records...")
         return run_suiteql(access_token, query, PAGE_SIZE, MAX_RECORDS)
     
     def find_newest_time(login_audits, system_notes):
         newest = None
         for r in login_audits:
             t = r.get('date')
             if t and (newest is None or t > newest):
                 newest = t
         for r in system_notes:
             t = r.get('date')
             if t and (newest is None or t > newest):
                 newest = t
         return newest
     
     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}")
             raise
         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}")
             raise
     
     def load_seen(bucket):
         try:
             blob = bucket.blob(STATE_KEY + '.seen')
             if blob.exists():
                 return set(json.loads(blob.download_as_text()))
         except Exception as e:
             print(f"Warning: Could not load seen keys: {e}")
             raise
         return set()
     
     def save_seen(bucket, keys):
         try:
             blob = bucket.blob(STATE_KEY + '.seen')
             blob.upload_from_string(
                 json.dumps(sorted(keys)),
                 content_type='application/json'
             )
             print(f"Saved {len(keys)} event keys for deduplication")
         except Exception as e:
             print(f"Warning: Could not save seen keys: {e}")
             raise
     

   Catatan: Skrip ini disediakan sebagai contoh dan tidak didukung secara resmi. Kami menerima masukan tentang cara meningkatkan kualitasnya. Untuk mengirimkan masukan, klik Kirim Masukan > Masukan Produk Keamanan Google SecOps di halaman ini.

   • File kedua - requirements.txt:

     functions-framework==3.*
     google-cloud-storage==2.*
     urllib3>=2.0.0
     PyJWT[crypto]>=2.8.0
     

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

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

   Catatan: Konfigurasi pemicu Pub/Sub otomatis membuat langganan dan izin yang diperlukan.

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	netsuite-audit-collector-hourly
   Region	Pilih region yang sama dengan fungsi Cloud Run
   Frekuensi	0 * * * * (setiap jam, tepat pada jamnya)
   Zona Waktu	Pilih zona waktu (UTC direkomendasikan)
   Jenis target	Pub/Sub
   Topik	Pilih netsuite-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 15 menit	*/15 * * * *	Volume tinggi, latensi rendah
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 (netsuite-audit-collector-hourly).
2. Klik Force run untuk memicu tugas secara manual.
3. Tunggu beberapa detik.
4. Buka Cloud Run > Services.
5. Klik netsuite-audit-collector.
6. Klik tab Logs.

7. Pastikan fungsi berhasil dieksekusi. Cari:

   Fetching logs from YYYY-MM-DDTHH:MM:SS+00:00 to YYYY-MM-DDTHH:MM:SS+00:00
   Successfully obtained NetSuite OAuth 2.0 access token
   Fetching LoginAudit records...
   Page 1: Retrieved X records (total: X)
   Fetching SystemNote records...
   Page 1: Retrieved X records (total: X)
   Wrote X records to gs://netsuite-audit-logs/netsuite-audit/netsuite_audit_YYYYMMDD_HHMMSS.ndjson
   Successfully processed X records (login_audit: X, system_note: X)
   

8. Buka Cloud Storage > Buckets.

9. Klik netsuite-audit-logs.

10. Buka folder netsuite-audit/.

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

Jika Anda melihat error di log, atau jenis data tidak ada dalam output:

• HTTP 401: Pastikan variabel lingkungan NS_CLIENT_ID, NS_CERTIFICATE_ID, dan NS_PRIVATE_KEY sudah benar.
• HTTP 403: Pastikan peran NetSuite memiliki izin login menggunakan izin OAuth 2.0 Access Tokens (Full), REST Web Services (Full), dan Login Audit Trail (Full).
• Tidak ada data system_note dalam output: Peran tidak memiliki izin System Notes for Analytics and REST (View). Kueri SuiteQL masih berhasil dan hanya menampilkan catatan untuk perubahan yang dilakukan oleh pengguna integrasi itu sendiri, sehingga tidak ada error yang mencapai log.
• HTTP 429: Pembatasan kapasitas - fungsi akan otomatis mencoba lagi dengan backoff eksponensial.
• Variabel lingkungan tidak ada: Pastikan semua variabel yang diperlukan telah ditetapkan dalam konfigurasi fungsi Cloud Run.

Mengambil akun layanan Google SecOps

1. Buka Setelan > Pengelolaan Data > Feed. Jika Anda menggunakan platform khusus SIEM, buka Settings > Feeds.
2. Klik Tambahkan Feed Berita Baru.
3. Klik Konfigurasi satu feed.
4. Di kolom Nama feed, masukkan nama untuk feed (misalnya, NetSuite Audit Logs).
5. Pilih Google Cloud Storage V2 sebagai Source type.
6. Pilih Oracle NetSuite sebagai Log type.
7. Klik Get Service Account.

8. Email akun layanan yang unik akan ditampilkan, misalnya:

   [e-mail]
   

9. Salin alamat email ini untuk digunakan di 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 dua peran di bucket Cloud Storage Anda: Storage Object Viewer untuk membaca objek log, dan peran level bucket untuk membaca metadata bucket.

1. Buka Cloud Storage > Buckets.
2. Klik netsuite-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 kedua opsi berikut:
     • Storage Object Viewer: membaca objek log.
     • Storage Legacy Bucket Reader: membaca metadata bucket. Jika Anda memilih opsi penghapusan Delete transferred files and empty directories, pilih Storage Legacy Bucket Writer, yang juga memberikan izin penghapusan.

6. Klik Simpan.

   Catatan: Berikan kedua peran. Storage Object Viewer tidak memberikan izin level bucket yang diperlukan feed, dan peran Storage Legacy Bucket tidak memberikan izin untuk membaca konten objek.

Menyelesaikan konfigurasi feed Google SecOps

1. Kembali ke wizard pembuatan feed Google SecOps.
2. Klik Berikutnya.

3. Tentukan nilai untuk parameter input berikut:

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

     gs://netsuite-audit-logs/netsuite-audit/
     

   Catatan: Selalu sertakan garis miring (/) di akhir URI.

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

     • Jangan pernah menghapus file: Jangan pernah menghapus file dari sumber (direkomendasikan untuk pengujian).
     • Hapus file yang ditransfer dan direktori kosong: Hapus file dan direktori kosong dari sumber setelah pengambilan berhasil diselesaikan.

   • Usia File Maksimum: Sertakan file yang diubah dalam jumlah hari terakhir (defaultnya adalah 180 hari)

   • Namespace aset: Namespace aset

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

4. Klik Berikutnya.

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

Batas kapasitas NetSuite API

NetSuite REST API memiliki batas tata kelola berikut:

Jenis Batas	Nilai
Batas konkurensi	10 permintaan serentak per integrasi
Hasil kueri SuiteQL per halaman	Maksimum 1.000 baris
Total hasil SuiteQL	100.000 baris per kueri

Fungsi Cloud Run secara otomatis menangani pembatasan kapasitas dengan backoff eksponensial.

Tabel pemetaan UDM

Kolom Log	Pemetaan UDM	Logika
Login_Audit_Trail_Access_Token_Name_label	additional.fields	Digabung
Login_Audit_Trail_Detail_label	additional.fields	Digabung
additional_preview	additional.fields	Digabung
detail_label	additional.fields	Digabung
role_change_label	additional.fields	Digabung
sec_challenge_label	additional.fields	Digabung
oauthaccesstokenname	extensions.auth.auth_details	Dipetakan secara langsung
has_principal	extensions.auth.type	Dipetakan: true → AUTHTYPE_UNSPECIFIED
column7	metadata.description	Dipetakan secara langsung
msg	metadata.description	Dipetakan secara langsung
Role_Change_Date	metadata.event_timestamp	Diurai sebagai M/d/yyyy h:mm a
date	metadata.event_timestamp	Diurai sebagai M/d/yyyy h:mm:ss a
result.Login_Audit_Trail_Date	metadata.event_timestamp	Diurai sebagai LDAP
ts	metadata.event_timestamp	Diurai sebagai MMM d HH:mm:ss
has_principal	metadata.event_type	Dipetakan: true → USER_LOGIN, true → STATUS_UPDATE
has_principal_user	metadata.event_type	Dipetakan: true → USER_UNCATEGORIZED
column1	metadata.product_event_type	Dipetakan secara langsung
log_type	metadata.product_event_type	Dipetakan secara langsung
useragent	network.http.parsed_user_agent	Dipetakan secara langsung
Login_Audit_Trail_User_Agent	network.http.user_agent	Dipetakan secara langsung
result.Login_Audit_Trail_User_Agent	network.http.user_agent	Dipetakan secara langsung
useragent	network.http.user_agent	Dipetakan secara langsung
Login_Audit_Trail_Application	principal.application	Dipetakan secara langsung
column10	principal.application	Dipetakan secara langsung
oauthappname	principal.application	Dipetakan secara langsung
result.host	principal.asset.hostname	Dipetakan secara langsung
column3	principal.asset.ip	Digabung
ipaddress	principal.asset.ip	Digabung
result.Login_Audit_Trail_IP_Address	principal.asset.ip	Dipetakan secara langsung
src_ip	principal.asset.ip	Digabung
result.host	principal.hostname	Dipetakan secara langsung
column3	principal.ip	Digabung
ipaddress	principal.ip	Digabung
result.Login_Audit_Trail_IP_Address	principal.ip	Dipetakan secara langsung
src_ip	principal.ip	Digabung
Login_Audit_Trail_Role_label	principal.user.attribute.labels	Digabung
login_audits_role_label	principal.user.attribute.labels	Digabung
supervisor_label	principal.user.attribute.labels	Digabung
role	principal.user.attribute.roles	Digabung
Email	principal.user.email_addresses	Dipetakan: ^.+@.+$ → Email
column2	principal.user.email_addresses	Digabung
emailaddress	principal.user.email_addresses	Dipetakan: ^.+@.+$ → emailaddress
result.Login_Audit_Trail_Email_Address	principal.user.email_addresses	Digabung
Job_Title	principal.user.title	Dipetakan secara langsung
Name	principal.user.user_display_name	Dipetakan secara langsung
Login_Audit_Trail_User	principal.user.userid	Dipetakan secara langsung
result.Login_Audit_Trail_User	principal.user.userid	Dipetakan secara langsung
user	principal.user.userid	Dipetakan secara langsung
Login_Audit_Trail_Security_Challenge_label	security_result.about.resource.attribute.labels	Digabung
security_result_action	security_result.action	Digabung
Login_Audit_Trail_Status	security_result.summary	Dipetakan secara langsung
result.Login_Audit_Trail_Status	security_result.summary	Dipetakan secara langsung
api_type_label	target.resource.attribute.labels	Digabung
Login_Audit_Trail_Request_URI	target.url	Dipetakan secara langsung
result.Login_Audit_Trail_Request_URI	target.url	Dipetakan secara langsung
T/A	extensions.auth.type	Konstanta: AUTHTYPE_UNSPECIFIED
T/A	metadata.event_type	Konstanta: USER_LOGIN
T/A	network.http.parsed_user_agent	Konstanta: parseduseragent

Log Perubahan

Melihat Log Perubahan untuk parser ini

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