Mengumpulkan log BeyondTrust Endpoint Privilege Management (EPM)

Didukung di:

Google SecOps SIEM

Dokumen ini menjelaskan cara menyerap log BeyondTrust Endpoint Privilege Management (EPM) ke Google Security Operations menggunakan Cloud Storage. Parser ini berfokus pada transformasi data log JSON mentah dari BeyondTrust Endpoint menjadi format terstruktur yang sesuai dengan UDM Google SecOps. Pertama-tama, nilai default untuk berbagai kolom diinisialisasi, lalu payload JSON diurai, dan selanjutnya kolom tertentu dari log mentah dipetakan ke kolom UDM yang sesuai dalam objek event.idm.read_only_udm.

Sebelum memulai

Pastikan Anda memiliki prasyarat berikut:

• Instance Google SecOps
• Project Google Cloud dengan Cloud Storage API diaktifkan
• Izin untuk membuat dan mengelola bucket Cloud Storage
• Izin untuk mengelola kebijakan Identity and Access Management (IAM) pada bucket Cloud Storage
• Izin untuk membuat layanan fungsi Cloud Run, topik Pub/Sub, dan tugas Cloud Scheduler
• Akses istimewa ke tenant atau API BeyondTrust Endpoint Privilege Management

Membuat bucket 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, beyondtrust-epm-logs)
   Location type	Pilih berdasarkan 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: Aktifkan pembuatan versi objek atau kebijakan retensi

6. Klik Create.

Mengumpulkan kredensial BeyondTrust EPM API

1. Login ke konsol web BeyondTrust Privilege Management sebagai administrator.
2. Buka Konfigurasi > Setelan > Setelan API.
3. Klik Create an API Account.
4. Berikan detail konfigurasi berikut:
   • Nama: Masukkan Google SecOps Collector.
   • Akses API: Aktifkan Pelaporan (Hanya Baca). Endpoint /management-api/v3/Events/FromStartDate yang digunakan oleh pengumpul ini termasuk dalam kategori izin Pelaporan. Aktifkan Audit (Hanya Baca) juga hanya jika Anda juga ingin mengumpulkan audit aktivitas konsol web.
5. Salin dan simpan Client ID dan Client Secret.

6. Salin URL dasar API yang ditampilkan di bagian atas halaman Setelan API. Biasanya https://<your-tenant>-services.pm.beyondtrustcloud.com. Anda akan menggunakannya sebagai BPT_API_URL.

   Catatan: Salin nilai dari konsol, bukan membuatnya. Tenant yang dikelola di app.beyondtrust.io menggunakan host yang berbeda.

Buat 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:
   • Nama akun layanan: Masukkan beyondtrust-epm-collector-sa.
   • Deskripsi akun layanan: Masukkan Service account for Cloud Run function to collect BeyondTrust EPM 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 file status
• Cloud Run Invoker: Mengizinkan Pub/Sub memanggil fungsi
• Cloud Functions Invoker: Mengizinkan pemanggilan fungsi

Memberikan izin IAM pada bucket Cloud Storage

Beri akun layanan izin tulis di bucket Cloud Storage:

1. Buka Cloud Storage > Buckets.
2. Klik nama bucket Anda.
3. Buka tab Izin.
4. Klik Grant access.
5. Berikan detail konfigurasi berikut:
   • Add principals: Masukkan email akun layanan. Contoh, beyondtrust-epm-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 KonsolGoogle Cloud , buka Pub/Sub > Topics.
2. Klik Create topic.
3. Berikan detail konfigurasi berikut:
   • ID Topik: Masukkan beyondtrust-epm-trigger.
   • Biarkan setelan lainnya sesuai default.
4. Klik Create.

Membuat fungsi Cloud Run untuk mengumpulkan log

Fungsi Cloud Run dipicu oleh pesan Pub/Sub dari Cloud Scheduler untuk mengambil log dari BeyondTrust EPM API dan menuliskannya ke Cloud Storage.

1. Di konsol Google Cloud , buka Cloud Run.
2. Klik Create service.
3. Pilih Fungsi. Gunakan editor inline untuk membuat fungsi.

4. Di bagian Konfigurasi, berikan detail konfigurasi berikut:

   Setelan	Nilai
   Nama layanan	beyondtrust-epm-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 Pemicu (opsional):

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

6. Di bagian Authentication:

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

7. Buka dan luaskan Containers, Networking, Security.

8. Buka tab Security:

   • Akun layanan: Pilih akun layanan beyondtrust-epm-collector-sa.

9. Buka tab Containers:

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

   Nama Variabel	Nilai Contoh
   GCS_BUCKET	beyondtrust-epm-logs
   GCS_PREFIX	beyondtrust-epm/
   STATE_KEY	beyondtrust-epm-state.json
   BPT_API_URL	https://yourtenant-services.pm.beyondtrustcloud.com
   CLIENT_ID	your-client-id
   CLIENT_SECRET	your-client-secret
   RECORD_SIZE	1000
   MAX_BATCHES	50
   LOOKBACK_HOURS	24

   • RECORD_SIZE: catatan per permintaan. /Events/FromStartDate menerima 1 hingga 1000; fungsi ini akan membatasi nilai yang lebih tinggi.
   • MAX_BATCHES: batas keamanan pada permintaan per eksekusi. Jika dijalankan, kursor akan tetap maju ke bagian yang ditulis, sehingga proses berikutnya akan dilanjutkan dari sana, bukan mengulangi pekerjaan.
   • LOOKBACK_HOURS: seberapa jauh mundur jangkauan operasi pertama. BeyondTrust menyimpan peristiwa pasif (kode 106, 107, 603, 706) selama 30 hari dan semua peristiwa lainnya selama 90 hari, sehingga periode lihat balik yang lebih lama tidak akan menampilkan apa pun yang lebih lama dari itu.
   • LOOKBACK_HOURS: seberapa jauh untuk menjangkau pada proses pertama, sebelum ada kursor.
   Catatan: Cakupan OAuth (urn:management:api) dan jalur API adalah nilai tetap dan ditetapkan dalam kode fungsi, sehingga bukan variabel lingkungan.
   Catatan: STATE_KEY sengaja menunjuk di luar awalan log. Feed menyerap setiap objek di bawah URI bucket-nya dan, saat opsi penghapusan dipilih, menghapus apa yang ditransfernya. Dengan menyimpan file status di luar awalan tersebut, file tidak akan diproses sebagai data log dan tidak akan dihapus, yang akan mereset pengumpulan dan memproses ulang peristiwa duplikat.

10. Di tab Variables & Secrets, buka Requests:

    • Waktu tunggu permintaan: Masukkan 600 detik (10 menit).

11. Buka tab Setelan di Penampung:

    • Di bagian Referensi:
      • Memori: Pilih 512 MiB atau lebih tinggi.
      • CPU: Pilih 1.
    • Klik Done.

12. Scroll ke Lingkungan eksekusi:

    • Pilih Default (direkomendasikan).

13. Di bagian Penskalaan revisi:

    • Minimum number of instances: Masukkan 0.
    • Maximum number of instances: Masukkan 100 (atau sesuaikan berdasarkan perkiraan beban).

14. Klik Create.

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

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

Menambahkan kode fungsi

1. Masukkan main di Function entry point

2. Di editor kode inline, buat dua file:

   • File pertama - main.py:
   Catatan: Endpoint token adalah /oauth/connect/token. Contoh PowerShell yang dipublikasikan dengan dokumentasi peristiwa ECS menampilkan /oauth/token; contoh tersebut salah. Setiap panduan BeyondTrust API dari 21.7 hingga rilis saat ini mendokumentasikan /oauth/connect/token, dan konektor Sentinel milik BeyondTrust sendiri, alur kerja QRadar, dan aplikasi Splunk semuanya memanggil /oauth/connect/token. Untuk mengonfirmasi endpoint untuk tenant tertentu, minta https://<tenant>-services.pm.beyondtrustcloud.com/oauth/.well-known/openid-configuration dan baca token_endpoint.

   import hashlib
   import json
   import os
   import re
   import time
   import urllib3
   import uuid
   from datetime import datetime, timedelta, timezone
   from urllib.parse import urlencode
   
   import functions_framework
   from google.cloud import storage
   from google.cloud.exceptions import NotFound
   
   # 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()
   
   TOKEN_PATH = '/oauth/connect/token'
   # Events/FromStartDate is the endpoint BeyondTrust documents for bulk extraction:
   # "A new API is exposed to extract the events in bulk." It takes only StartDate and
   # RecordSize. Events/search is not usable here: it requires an OperatingSystem value,
   # it is a single string with no documented "all" value, so it can only ever return one
   # operating system per call.
   EVENTS_PATH = '/management-api/v3/Events/FromStartDate'
   OAUTH_SCOPE = 'urn:management:api'
   # RecordSize accepts 1 to 1000.
   MAX_RECORD_SIZE = 1000
   
   class FetchError(Exception):
       """Raised when the BeyondTrust API call fails.
   
       The cursor must never advance on a failed fetch, otherwise the events in the
       failed window are skipped permanently.
       """
   
   def rfc3339(dt: datetime) -> str:
       """Render a datetime the way StartDate accepts it: milliseconds and a literal Z."""
       return dt.astimezone(timezone.utc).strftime('%Y-%m-%dT%H:%M:%S.%f')[:-3] + 'Z'
   
   def parse_iso(stamp: str) -> datetime:
       """Parse any ISO 8601 timestamp shape the API emits.
   
       event.ingested comes back with a +00:00 offset and up to seven fractional
       digits, e.g. 2026-08-03T13:29:55.1109163+00:00. fromisoformat's tolerance for
       long fractions varies across Python versions, so the fraction is trimmed to
       microseconds first.
       """
       text = stamp.strip()
       if text.endswith('Z'):
           text = text[:-1] + '+00:00'
       text = re.sub(r'\.(\d{1,6})\d*', r'.\1', text, count=1)
       parsed = datetime.fromisoformat(text)
       if parsed.tzinfo is None:
           parsed = parsed.replace(tzinfo=timezone.utc)
       return parsed
   
   def canonical(stamp: str) -> str:
       """Re-render an API timestamp into the one form StartDate accepts.
   
       The API returns event.ingested as +00:00-offset ISO 8601 but rejects that
       same shape as a StartDate value with 400 "Invalid Start date format": it
       accepts only Z-suffixed values. Every timestamp that came from the API must
       pass through here before being sent back or persisted.
       """
       return rfc3339(parse_iso(stamp))
   
   def event_id(evt: dict) -> str:
       """Return the event's identity for deduplication.
   
       Falls back to a content hash when event.id is absent, so an id-less event
       still deduplicates instead of being re-ingested on every boundary re-read.
       """
       explicit = str((evt.get('event') or {}).get('id') or '')
       if explicit:
           return explicit
       digest = hashlib.sha256(
           json.dumps(evt, sort_keys=True, ensure_ascii=False).encode('utf-8')
       ).hexdigest()
       return f'sha256:{digest}'
   
   def event_ingested(evt: dict) -> str:
       """Return the Elastic ingestion timestamp, which is what StartDate filters on."""
       return str((evt.get('event') or {}).get('ingested') or '')
   
   def next_millisecond(stamp: str) -> str:
       """Return the canonical timestamp one millisecond later.
   
       Used only when a full batch fits inside a single ingestion millisecond. Without
       this the cursor cannot move and the collector stalls on that timestamp forever.
       """
       return rfc3339(parse_iso(stamp) + timedelta(milliseconds=1))
   
   @functions_framework.cloud_event
   def main(cloud_event):
       """Fetch BeyondTrust EPM events and write them to Cloud Storage as NDJSON.
   
       Args:
               cloud_event: CloudEvent object containing the Pub/Sub message.
       """
       bucket_name = os.environ.get('GCS_BUCKET')
       prefix = os.environ.get('GCS_PREFIX', 'beyondtrust-epm/')
       state_key = os.environ.get('STATE_KEY', 'beyondtrust-epm-state.json')
   
       api_url = (os.environ.get('BPT_API_URL') or '').rstrip('/')
       client_id = os.environ.get('CLIENT_ID')
       client_secret = os.environ.get('CLIENT_SECRET')
       # Clamp both ends: the API rejects RecordSize outside 1 to 1000, and a
       # misconfigured 0 would otherwise crash-loop on a 400 every run.
       record_size = min(max(int(os.environ.get('RECORD_SIZE', '1000')), 1), MAX_RECORD_SIZE)
       max_batches = int(os.environ.get('MAX_BATCHES', '50'))
       lookback_hours = int(os.environ.get('LOOKBACK_HOURS', '24'))
   
       if not all([bucket_name, api_url, client_id, client_secret]):
           raise RuntimeError(
               'Missing required environment variables: '
               'GCS_BUCKET, BPT_API_URL, CLIENT_ID, CLIENT_SECRET'
           )
   
       bucket = storage_client.bucket(bucket_name)
       state = load_state(bucket, state_key)
   
       # StartDate is a rising cursor on the Elastic ingestion timestamp, not a closed
       # window. There is no EndDate on this endpoint.
   
       if state.get('last_ingested'):
           # canonical() also repairs state written by the previous script revision,
           # which persisted the API's raw +00:00 form that StartDate rejects.
           try:
               start_date = canonical(state['last_ingested'])
           except (ValueError, TypeError, AttributeError) as e:
               raise RuntimeError(
                   f'Unparseable last_ingested in gs://{bucket_name}/{state_key}: '
                   f'{state["last_ingested"]!r}. Fix or delete that object; deleting '
                   f'restarts collection from the lookback window.'
               ) from e
       else:
           start_date = rfc3339(datetime.now(timezone.utc) - timedelta(hours=lookback_hours))
   
       seen_ids = set(state.get('seen_ids', []))
       print(f'Collecting events ingested from {start_date}')
   
       token = get_oauth_token(api_url, client_id, client_secret)
       fresh, cursor, drained = fetch_events(
           api_url, token, start_date, record_size, max_batches, seen_ids
       )
   
       if not fresh:
           print('No new events. Cursor left unchanged.')
           return
   
       # Everything that can fail is computed before the upload: a crash between the
       # upload and save_state replays the batch on the next run, so the window where
       # side effects exist without recorded state must stay minimal. The retained ids
       # share the cursor millisecond, because those are exactly the ones the
       # inclusive StartDate will return again.
       retained = {
           event_id(e) for e in fresh
           if event_ingested(e) and canonical(event_ingested(e)) == cursor
       }
       if cursor == start_date:
           # The cursor millisecond did not advance, so ids retained by earlier runs
           # are still on the boundary; dropping them would re-ingest their events.
           retained |= seen_ids
   
       timestamp = datetime.now(timezone.utc).strftime('%Y%m%dT%H%M%SZ')
       # The random suffix keeps concurrent executions (Pub/Sub delivers at least
       # once) from overwriting each other's object within the same second.
       filename = (
           f'{prefix}beyondtrust-epm-events-{timestamp}-'
           f'{uuid.uuid4().hex[:8]}.ndjson'
       )
       ndjson = '\n'.join(json.dumps(e, ensure_ascii=False) for e in fresh) + '\n'
       bucket.blob(filename).upload_from_string(
           ndjson, content_type='application/x-ndjson'
       )
       print(f'Wrote {len(fresh)} events to gs://{bucket_name}/{filename}')
   
       save_state(bucket, state_key, {
           'last_ingested': cursor,
           'seen_ids': sorted(retained),
           'updated_at': rfc3339(datetime.now(timezone.utc)),
       })
   
       if not drained:
           print(
               f'Stopped after {max_batches} batches with more events available. '
               'The cursor advanced, so the next run continues from here.'
           )
   
   def fetch_events(api_url, token, start_date, record_size, max_batches, seen_ids):
       """Read forward from start_date until the API returns a short batch.
   
       The response envelope has no page count and no next-page token, so a batch
       shorter than record_size is the only documented end-of-data signal.
   
       StartDate is inclusive and the cursor lands on the newest event's timestamp, so
       every batch re-returns the events sharing it. Deduplication by event id therefore
       has to happen as batches arrive, not only between runs.
   
       Returns:
       Tuple of (new events, cursor to persist, whether the stream was drained).
   
       Raises:
       FetchError: on any API or transport failure, so the caller cannot mistake a
       failed fetch for an empty result and advance the cursor.
       """
       headers = {
           'Authorization': f'Bearer {token}',
           'Accept': 'application/json',
       }
   
       all_events = []
       seen = set(seen_ids)
       cursor = start_date
   
       for batch in range(1, max_batches + 1):
           query = urlencode({'StartDate': cursor, 'RecordSize': record_size})
           body = request_with_retry(f'{api_url}{EVENTS_PATH}?{query}', headers)
   
           events = body.get('events') or []
           new = [e for e in events if event_id(e) not in seen]
           seen.update(event_id(e) for e in events if event_id(e))
           all_events.extend(new)
           print(f'Batch {batch}: {len(events)} events, {len(new)} new')
   
           # Termination is judged on what the API returned, not on what survived
           # deduplication: a batch can be full and still be entirely duplicates.
           # The newest stamp in the batch, not events[-1]: ordering is a client-side
           # convention, not a documented guarantee, and a misordered tail would
           # regress the cursor and re-ingest events already written.
           stamps = [canonical(s) for s in (event_ingested(e) for e in events) if s]
   
           if len(events) < record_size:
               # Short batch: the stream is drained.
               if events and not stamps:
                   print(
                       'Warning: no event in the final batch carries event.ingested; '
                       'cursor left unchanged.'
                   )
               if stamps:
                   cursor = max(max(stamps), cursor)
               return all_events, cursor, True
   
           if not stamps:
               raise FetchError(
                   'No event in a full batch carries event.ingested, '
                   'so the cursor cannot advance'
               )
           batch_max = max(stamps)
           if batch_max <= cursor:
               # A full batch fits inside one ingestion millisecond. Stepping past it is the only
               # way to make progress; holding the cursor here stalls collection forever.
               # Events beyond record_size at that exact millisecond are unreachable, which
               # needs more than 1000 events in one millisecond.
               print(
                   f'Warning: a full batch shares ingestion timestamp {batch_max}; '
                   'stepping past it. Events beyond RecordSize at that timestamp are skipped.'
               )
               cursor = next_millisecond(cursor)
           else:
               cursor = batch_max
   
       return all_events, cursor, False
   
   def request_with_retry(url, headers, attempts=4):
       """GET with backoff on 429 and 5xx.
   
       The documented rate limit is 1000 requests per 100 seconds.
       """
       backoff = 1.0
       for attempt in range(1, attempts + 1):
           try:
               response = http.request('GET', url, headers=headers)
           except Exception as e:
               raise FetchError(f'Request to {url} failed: {e}') from e
   
           if response.status in (429, 500, 502, 503, 504) and attempt < attempts:
               retry_after = response.headers.get('Retry-After')
               try:
                   delay = int(retry_after) if retry_after else backoff
               except (TypeError, ValueError):
                   delay = backoff
               # Retry-After is server-controlled input: bound it so a bogus value
               # cannot sleep past the function timeout or crash time.sleep.
               delay = min(max(delay, 1.0), 60.0)
               print(f'HTTP {response.status}. Retrying in {delay}s...')
               time.sleep(delay)
               backoff = min(backoff * 2, 30.0)
               continue
   
           if response.status != 200:
               raise FetchError(
                   f'Request failed: {response.status} {response.data.decode("utf-8")}'
               )
   
           text = response.data.decode('utf-8')
           # BeyondTrust's sample folds "Owner" into "owner" before parsing, but that
           # workaround exists for PowerShell's case-insensitive ConvertFrom-Json.
           # Python parses case-sensitively, and the SecOps parser maps file.Owner.*
           # and file.owner to different UDM fields, so both keys must survive.
           try:
               return json.loads(text)
           except json.JSONDecodeError as e:
               raise FetchError(f'Malformed JSON response: {e}') from e
   
       raise FetchError(f'Giving up on {url} after {attempts} attempts')
   
   def get_oauth_token(api_url, client_id, client_secret):
       """Get an access token using the OAuth client credentials flow.
   
       The token is valid for one hour.
       """
       body = urlencode({
           'grant_type': 'client_credentials',
           'client_id': client_id,
           'client_secret': client_secret,
           'scope': OAUTH_SCOPE,
       })
       response = http.request(
           'POST',
           f'{api_url}{TOKEN_PATH}',
           body=body,
           headers={'Content-Type': 'application/x-www-form-urlencoded'},
       )
       if response.status != 200:
           raise FetchError(
               f'Token request failed: {response.status} '
               f'{response.data.decode("utf-8")}'
           )
       return json.loads(response.data.decode('utf-8'))['access_token']
   
   def load_state(bucket, key):
       """Read the collector state from Cloud Storage.
   
       Only a missing object is a cold start. Any other error is raised: swallowing it
       would reset collection to the lookback window and re-ingest that period.
       """
       blob = bucket.blob(key)
       try:
           return json.loads(blob.download_as_text())
       except NotFound:
           print('No state file found. Starting from the lookback window.')
           return {}
   
   def save_state(bucket, key, state):
       """Write the collector state to Cloud Storage.
   
       Failures are raised, not logged. A run that cannot record its cursor must fail,
       otherwise the next run repeats the same window.
       """
       bucket.blob(key).upload_from_string(
           json.dumps(state, indent=2), content_type='application/json'
       )
       print(f"Saved state: last_ingested={state.get('last_ingested')}")
   

   Catatan: Skrip ini disediakan sebagai contoh dan tidak didukung secara resmi. Kami menerima masukan tentang cara kami dapat meningkatkan kualitasnya. Untuk mengirim 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
   

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.

Buat tugas Cloud Scheduler

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

1. Di konsol Google Cloud , buka Cloud Scheduler.
2. Klik Create Job.

3. Berikan detail konfigurasi berikut:

   Setelan	Nilai
   Nama	beyondtrust-epm-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 topik beyondtrust-epm-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 tugas penjadwal

1. Di konsol Cloud Scheduler, temukan tugas Anda.
2. Klik Jalankan paksa untuk memicu secara manual.
3. Tunggu beberapa detik, lalu buka Cloud Run > Services > beyondtrust-epm-collector > Logs.
4. Pastikan fungsi berhasil dieksekusi.
5. Periksa bucket Cloud Storage untuk mengonfirmasi bahwa log telah ditulis.

Mengambil akun layanan Google SecOps

Google SecOps menggunakan akun layanan unik untuk membaca data dari bucket Cloud Storage 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, BeyondTrust EPM logs).
5. Pilih Google Cloud Storage V2 sebagai Source type.
6. Pilih BeyondTrust Endpoint Privilege Management sebagai Jenis log.

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

   [e-mail]
   

8. 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 tingkat bucket untuk membaca metadata bucket.

1. Buka Cloud Storage > Buckets.
2. Klik nama bucket Anda.
3. Buka tab Izin.
4. Klik Grant access.
5. Berikan detail konfigurasi berikut:
   • Add principals: Tempel email akun layanan Google SecOps.
   • Tetapkan peran: Pilih kedua opsi berikut:
   • Storage Object Viewer: membaca objek log.
   • Storage Legacy Bucket Reader: membaca metadata bucket. Jika Anda memilih opsi penghapusan Hapus file yang ditransfer dan direktori kosong, pilih Storage Legacy Bucket Writer, yang juga memberikan izin penghapusan.

6. Klik Simpan.

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

Mengonfigurasi feed di Google SecOps untuk menyerap log BeyondTrust EPM

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, BeyondTrust EPM logs).
5. Pilih Google Cloud Storage V2 sebagai Source type.
6. Pilih BeyondTrust Endpoint Privilege Management sebagai Jenis log.
7. Klik Berikutnya.

8. Tentukan nilai untuk parameter input berikut:

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

     gs://beyondtrust-epm-logs/beyondtrust-epm/
     

     • Ganti:

       • beyondtrust-epm-logs: Nama bucket Cloud Storage Anda.
       • beyondtrust-epm/: Awalan/jalur folder opsional tempat log disimpan (biarkan kosong untuk root).

     • Contoh:

       • Bucket root: gs://beyondtrust-epm-logs/
       • Dengan awalan: gs://beyondtrust-epm-logs/beyondtrust-epm/
     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. Tindakan ini direkomendasikan untuk tujuan pengujian.

     • Hapus file yang ditransfer dan direktori kosong: Menghapus file dan direktori kosong dari sumber setelah pengambilan yang berhasil selesai.

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

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

   • Namespace aset: Namespace aset.

   • Label penyerapan: Label yang akan 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
DomainIdentifier_label	additional.fields	Digabung
WatsonEventType_label	additional.fields	Digabung
action_label	additional.fields	Digabung
activity_id_label	additional.fields	Digabung
app_id_label	additional.fields	Digabung
app_insights_instrumentation_key_label	additional.fields	Digabung
asimov_instrumentation_key_label	additional.fields	Digabung
authorization_request_control_authorization_label	additional.fields	Digabung
bundle_name_label	additional.fields	Digabung
changedBy_label	additional.fields	Digabung
code_signature_subject_name_label	additional.fields	Digabung
collector_api_key_label	additional.fields	Digabung
command_label	additional.fields	Digabung
configuration_application_group_description_label	additional.fields	Digabung
configuration_application_group_identifier_label	additional.fields	Digabung
configuration_application_group_name_label	additional.fields	Digabung
configuration_application_identifier_label	additional.fields	Digabung
configuration_application_type_label	additional.fields	Digabung
configuration_identifier_label	additional.fields	Digabung
configuration_message_identifier_label	additional.fields	Digabung
configuration_message_name_label	additional.fields	Digabung
configuration_message_type_label	additional.fields	Digabung
configuration_name_label	additional.fields	Digabung
configuration_revision_number_label	additional.fields	Digabung
configuration_rule_identifier_label	additional.fields	Digabung
configuration_rule_on_demand_label	additional.fields	Digabung
configuration_rule_script_outcome_rule_affected_label	additional.fields	Digabung
configuration_token_identifier_label	additional.fields	Digabung
configuration_token_name_label	additional.fields	Digabung
configuration_workstyle_description_label	additional.fields	Digabung
configuration_workstyle_identifier_label	additional.fields	Digabung
configuration_workstyle_name_label	additional.fields	Digabung
content_length_label	additional.fields	Digabung
content_type_label	additional.fields	Digabung
domainNetBIOSName_label	additional.fields	Digabung
domain_label	additional.fields	Digabung
entity_label	additional.fields	Digabung
entity_name_label	additional.fields	Digabung
event_action_label	additional.fields	Digabung
exitstatus_label	additional.fields	Digabung
file_hash_md5_label	additional.fields	Digabung
file_hash_sha1_label	additional.fields	Digabung
file_hash_sha256_label	additional.fields	Digabung
file_version_label	additional.fields	Digabung
gid_label	additional.fields	Digabung
group_data_id_label	additional.fields	Digabung
group_id_label	additional.fields	Digabung
group_label	additional.fields	Digabung
handle_label	additional.fields	Digabung
host_name_label	additional.fields	Digabung
host_uptime_labels	additional.fields	Digabung
http_host_label	additional.fields	Digabung
id_label	additional.fields	Digabung
iolog_label	additional.fields	Digabung
is_opted_in_label	additional.fields	Digabung
linenum_label	additional.fields	Digabung
local_identifier_label	additional.fields	Digabung
locale_label	additional.fields	Digabung
master_utcoffset_label	additional.fields	Digabung
masterlocale_label	additional.fields	Digabung
owner_identifier_label	additional.fields	Digabung
parent_entity_id_label	additional.fields	Digabung
parent_process_exec_label	additional.fields	Digabung
parent_process_label	additional.fields	Digabung
pbmasterdnodename_label	additional.fields	Digabung
pipeName_label	additional.fields	Digabung
process_entity_id_label	additional.fields	Digabung
process_hash_label	additional.fields	Digabung
process_name_label	additional.fields	Digabung
process_parent_name_label	additional.fields	Digabung
process_start_time_label	additional.fields	Digabung
processexe_label	additional.fields	Digabung
product_label	additional.fields	Digabung
product_type_label	additional.fields	Digabung
product_version_label	additional.fields	Digabung
requestuser_label	additional.fields	Digabung
runargv_label	additional.fields	Digabung
runcwd_label	additional.fields	Digabung
runeffectivegroup_label	additional.fields	Digabung
runeffectiveuser_label	additional.fields	Digabung
runhost_label	additional.fields	Digabung
schema_version_label	additional.fields	Digabung
sku_name_label	additional.fields	Digabung
telemetry_level_label	additional.fields	Digabung
tenant_id_label	additional.fields	Digabung
type_label	additional.fields	Digabung
uid_label	additional.fields	Digabung
user_id_label	additional.fields	Digabung
user_name_label	additional.fields	Digabung
vs_exe_version_label	additional.fields	Digabung
inter_host	intermediary.hostname	Dipetakan secara langsung
Processes.description	metadata.description	Dipetakan secara langsung
details	metadata.description	Dipetakan secara langsung
event_data.reason	metadata.description	Dipetakan secara langsung
file.pe.description	metadata.description	Dipetakan secara langsung
created	metadata.event_timestamp	Diuraikan sebagai ISO8601
datetime	metadata.event_timestamp	Diuraikan sebagai MMM dd HH:mm:ss
has_principal	metadata.event_type	Dipetakan: true → STATUS_UPDATE
has_user	metadata.event_type	Dipetakan: true → USER_UNCATEGORIZED
parent_working_directory_label	metadata.ingestion_labels	Digabung
working_directory_label	metadata.ingestion_labels	Digabung
auditType	metadata.product_event_type	Dipetakan secara langsung
event_datas.ActionId	metadata.product_log_id	Dipetakan secara langsung
labels.related_item_id	metadata.product_log_id	Dipetakan secara langsung
uniqueid	metadata.product_log_id	Dipetakan secara langsung
masterdversion	metadata.product_version	Dipetakan secara langsung
headers.http_version	network.application_protocol_version	Dipetakan secara langsung
headers.request_method	network.http.method	Dipetakan secara langsung
host.os.platform	principal.administrative_domain	Dipetakan secara langsung
Processes.process	principal.application	Dipetakan secara langsung
agent_ephemeral_id_label	principal.asset.attribute.labels	Digabung
agent_id_label	principal.asset.attribute.labels	Digabung
agent_version_label	principal.asset.attribute.labels	Digabung
ecs_version_label	principal.asset.attribute.labels	Digabung
_hardware	principal.asset.hardware	Digabung
host.hostname	principal.asset.hostname	Dipetakan secara langsung
submithost	principal.asset.hostname	Dipetakan secara langsung
ip_address	principal.asset.ip	Digabung
masterhostip	principal.asset.ip	Digabung
submithostip	principal.asset.ip	Digabung
file.path	principal.file.full_path	Dipetakan secara langsung
lineinfile	principal.file.full_path	Dipetakan secara langsung
host.hostname	principal.hostname	Dipetakan secara langsung
submithost	principal.hostname	Dipetakan secara langsung
ip_address	principal.ip	Digabung
masterhostip	principal.ip	Digabung
submithostip	principal.ip	Digabung
mac	principal.mac	Digabung
host.os.name	principal.platform_version	Dipetakan secara langsung
host.os.version	principal.platform_version	Dipetakan secara langsung
process.command_line	principal.process.command_line	Dipetakan secara langsung
runcommand	principal.process.command_line	Dipetakan secara langsung
process.executable	principal.process.file.full_path	Dipetakan secara langsung
cmd	principal.process.parent_process.command_line	Dipetakan secara langsung
Processes.process_path	principal.process.parent_process.file.full_path	Dipetakan secara langsung
process.parent.executable	principal.process.parent_process.file.full_path	Dipetakan secara langsung
Processes.parent_process_id	principal.process.parent_process.pid	Dipetakan secara langsung
Processes.process_id	principal.process.pid	Dipetakan secara langsung
logpid	principal.process.pid	Dipetakan secara langsung
file.Owner.DomainName	principal.user.company_name	Dipetakan secara langsung
file.Owner.Name	principal.user.user_display_name	Dipetakan secara langsung
runuser	principal.user.user_display_name	Dipetakan secara langsung
userName	principal.user.user_display_name	Dipetakan secara langsung
Processes.user	principal.user.userid	Dipetakan secara langsung
userId	principal.user.userid	Dipetakan secara langsung
userid	principal.user.userid	Dipetakan secara langsung
Processes.user_id	principal.user.windows_sid	Dipetakan secara langsung
EPMWinMac.Configuration.Rule.Action	security_result.action	Digabung
security_result_action	security_result.action	Digabung
event_data.outcome	security_result.category_details	Digabung
host.os.version	security_result.category_details	Digabung
EPMWinMac.Configuration.Application.Description	security_result.description	Dipetakan secara langsung
EPMWinMac.Configuration.Message.Description	security_result.description	Dipetakan secara langsung
host.os.type	src.administrative_domain	Dipetakan secara langsung
file.name	src.file.names	Digabung
host.os.full	src.platform_version	Dipetakan secara langsung
host.os.family	target.administrative_domain	Dipetakan secara langsung
Processes.dest	target.asset.hostname	Dipetakan secara langsung
file.extension	target.file.mime_type	Dipetakan secara langsung
Processes.dest	target.hostname	Dipetakan secara langsung
host.domain	target.hostname	Dipetakan secara langsung
file_DriveType_label	target.resource.attribute.labels	Digabung
file_drive_letter_label	target.resource.attribute.labels	Digabung
owner_label	target.resource.attribute.labels	Digabung
T/A	metadata.event_type	Konstanta: USER_UNCATEGORIZED
T/A	metadata.product_name	Konstanta: Beyondtrust Privilege Management
T/A	metadata.vendor_name	Konstanta: Beyondtrust Privilege Management
T/A	network.application_protocol	Konstanta: HTTP
T/A	principal.platform	Konstanta: MAC

Log Perubahan

Melihat Log Perubahan untuk parser ini

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