Mengumpulkan log platform MuleSoft Anypoint

Didukung di:

Dokumen ini menjelaskan cara menyerap peristiwa jejak audit dari log platform MuleSoft Anypoint ke Google Security Operations menggunakan Cloud Storage.

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 fungsi Cloud Run, topik Pub/Sub, dan tugas Cloud Scheduler
  • Izin untuk membuat akun layanan
  • Akses istimewa ke MuleSoft Anypoint Platform

Mendapatkan ID Organisasi MuleSoft

  1. Login ke Anypoint Platform.
  2. Buka Pengelolaan Akses > Organisasi.
  3. Di tabel Grup Bisnis, klik nama organisasi Anda.
  4. Salin ID organisasi (misalnya, 0a12b3c4-d5e6-789f-1021-1a2b34cd5e6f).

Atau, buka MuleSoft Business Groups dan salin ID dari URL.

Buat Aplikasi Terhubung MuleSoft

  1. Login ke Anypoint Platform.
  2. Buka Pengelolaan Akses > Aplikasi Terhubung > Buat Aplikasi.
  3. Berikan detail konfigurasi berikut:
    • Nama aplikasi: Masukkan nama unik (misalnya, Google SecOps export).
    • Pilih Aplikasi bertindak atas namanya sendiri (kredensial klien).
  4. Klik Add scopes > Audit Log Viewer > Next.
  5. Pilih setiap Grup Bisnis yang lognya Anda butuhkan.
  6. Klik Berikutnya > Tambahkan cakupan.

  7. Klik Save, lalu salin Client ID dan Client Secret.

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
    Memberi nama bucket Anda Masukkan nama yang unik secara global (misalnya, mulesoft-audit-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.

Buat akun layanan untuk fungsi Cloud Run

Cloud Run Function memerlukan akun layanan dengan izin untuk menulis ke bucket Cloud Storage.

Membuat akun layanan

  1. Di Konsol GCP, buka IAM & Admin > Service Accounts.
  2. Klik Create Service Account.
  3. Berikan detail konfigurasi berikut:
    • Service account name: Masukkan mulesoft-logs-collector-sa.
    • Deskripsi akun layanan: Masukkan Service account for Cloud Run function to collect MuleSoft Anypoint logs.
  4. Klik Create and Continue.
  5. Di bagian Grant this service account access to project:
    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: Izinkan Pub/Sub memanggil fungsi
  • Cloud Functions Invoker: Mengizinkan pemanggilan fungsi

Memberikan izin IAM pada bucket Cloud Storage

Berikan izin write akun layanan di bucket Cloud Storage sebagai berikut:

  1. Buka Cloud Storage > Buckets.
  2. Klik nama bucket Anda.
  3. Buka tab Izin.
  4. Klik Grant access.
  5. Berikan detail konfigurasi berikut:
    • Tambahkan prinsipal: Masukkan email akun layanan (misalnya, mulesoft-logs-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 di-subscribe oleh fungsi Cloud Run sebagai berikut:

  1. Di Konsol GCP, buka Pub/Sub > Topics.
  2. Klik Create topic.
  3. Berikan detail konfigurasi berikut:
    • ID Topik: Masukkan mulesoft-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 dari MuleSoft Anypoint API dan menuliskannya ke Cloud Storage.

  1. Di Konsol GCP, buka Cloud Run.
  2. Klik Write a function.

  3. Di bagian Configure, berikan detail konfigurasi berikut:

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

  4. Di bagian Pemicu, klik Tambahkan pemicu, lalu pilih Pemicu Pub/Sub.

  5. Di panel Pemicu Eventarc, berikan detail konfigurasi berikut:

    • Nama pemicu: Biarkan nama yang dibuat atau masukkan nama untuk pemicu.
    • Jenis pemicu: Pilih Sumber Google.
    • Penyedia peristiwa: Pilih Pub/Sub.
    • Jenis peristiwa: Pilih google.cloud.pubsub.topic.v1.messagePublished.
    • Select a Cloud Pub/Sub topic: Pilih topik mulesoft-audit-trigger.
    • Region: Pilih region yang sama dengan fungsi.
    • Akun layanan: Pilih akun layanan mulesoft-logs-collector-sa.

  6. Klik Save trigger.

  7. Di bagian Authentication:

    1. Pilih Wajibkan autentikasi.
    2. PeriksaIdentity and Access Management (IAM).

  8. Buka dan luaskan Containers, Networking, Security.

  9. Buka tab Security:

    • Service account: Pilih akun layanan mulesoft-logs-collector-sa.

  10. Buka tab Containers:

    1. Klik Variables & Secrets.
    2. Klik + Tambahkan variabel untuk setiap variabel lingkungan:
    Nama Variabel Nilai Contoh Deskripsi
    GCS_BUCKET mulesoft-audit-logs Nama bucket Cloud Storage
    GCS_PREFIX mulesoft-audit Awalan untuk file log
    STATE_KEY mulesoft-audit-state.json Jalur status, di luar awalan log
    MULE_ORG_ID your_org_id ID Organisasi MuleSoft
    CLIENT_ID your_client_id ID klien Aplikasi Terhubung
    CLIENT_SECRET your_client_secret Rahasia klien Aplikasi Terhubung
    MAX_RECORDS 10000 Jumlah data maks per proses
    PAGE_SIZE 200 Entri per halaman (maksimum API adalah 200)
    LOOKBACK_HOURS 24 Periode lihat balik awal pada operasi pertama

  11. Buka tab Variables & Secrets ke Requests:

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

  12. Buka tab Setelan di Penampung:

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

  13. Buka Lingkungan eksekusi:

    • Pilih Default (direkomendasikan).

  14. Di bagian Penskalaan revisi:

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

  15. Klik Create.

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

  17. Setelah layanan dibuat, editor kode inline akan otomatis terbuka.

Menambahkan kode fungsi

  1. Masukkan main di Function entry point

  2. Di editor kode inline, buat dua file:

    • File pertama - main.py:

      import functions_framework
from google.cloud import storage
from google.cloud.exceptions import NotFound
import hashlib
import json
import os
import urllib3
from urllib.parse import urlencode
from datetime import datetime, timezone, timedelta
import time

# Initialize HTTP client with timeouts
http = urllib3.PoolManager(
    timeout=urllib3.Timeout(connect=5.0, read=30.0),
    retries=False,
)

# Initialize Storage client
storage_client = storage.Client()

# MuleSoft API endpoints
TOKEN_URL = "https://anypoint.mulesoft.com/accounts/api/v2/oauth2/token"
AUDIT_URL = "https://anypoint.mulesoft.com/audit/v2/organizations/{org_id}/query?cursorPagination=true"

# Environment variables
GCS_BUCKET = os.environ.get('GCS_BUCKET')
GCS_PREFIX = os.environ.get('GCS_PREFIX', 'mulesoft-audit')
# STATE_KEY must stay OUTSIDE GCS_PREFIX. The feed ingests every object under
# its bucket URI and, with a deletion option selected, deletes what it
# transferred. A state file inside the prefix would be ingested as log data and
# then deleted, resetting collection and re-ingesting duplicates.
STATE_KEY = os.environ.get('STATE_KEY', 'mulesoft-audit-state.json')
MULE_ORG_ID = os.environ.get('MULE_ORG_ID')
CLIENT_ID = os.environ.get('CLIENT_ID')
CLIENT_SECRET = os.environ.get('CLIENT_SECRET')
MAX_RECORDS = int(os.environ.get('MAX_RECORDS', '10000'))
# The Audit Log Query API returns at most 200 entries per JSON query.
PAGE_SIZE = min(int(os.environ.get('PAGE_SIZE', '200')), 200)
LOOKBACK_HOURS = int(os.environ.get('LOOKBACK_HOURS', '24'))
# The query re-reads this many minutes before the watermark so that events the
# platform records late are still collected. Re-read events are removed by hash.
OVERLAP_MINUTES = int(os.environ.get('OVERLAP_MINUTES', '2'))

class FetchError(Exception):
    """Raised when the MuleSoft API call fails.

    The watermark must never advance on a failed fetch, otherwise the events in
    the failed window are skipped permanently.
    """

def to_unix_millis(dt: datetime) -> int:
    """Convert datetime to Unix epoch milliseconds."""
    if dt.tzinfo is None:
        dt = dt.replace(tzinfo=timezone.utc)
    return int(dt.astimezone(timezone.utc).timestamp() * 1000)

def parse_datetime(value: str) -> datetime:
    """Parse an ISO 8601 datetime string into an aware datetime."""
    if value.endswith("Z"):
        value = value[:-1] + "+00:00"
    dt = datetime.fromisoformat(value)
    if dt.tzinfo is None:
        dt = dt.replace(tzinfo=timezone.utc)
    return dt

def event_keys(event: dict) -> set:
    """Return the identity this event can be recognised by.

    Audit log entries carry no unique event ID, so a record counts as already
    collected when its full JSON content was seen before. The overlap window
    returns the same JSON again, which this catches without any knowledge of
    the payload schema.
    """
    return {'sha256:' + hashlib.sha256(
        json.dumps(event, sort_keys=True, ensure_ascii=False).encode('utf-8')
    ).hexdigest()}

def event_time(event: dict) -> str:
    """Return the event timestamp as an ISO 8601 string, or an empty string.

    Audit log entries carry the event time in `timestamp` as Unix milliseconds.
    """
    raw = event.get('timestamp')
    if raw is None or raw == '':
        return ''
    try:
        return datetime.fromtimestamp(int(raw) / 1000, tz=timezone.utc).isoformat()
    except (TypeError, ValueError):
        return ''

@functions_framework.cloud_event
def main(cloud_event):
    """Fetch MuleSoft Anypoint audit logs and write them to Cloud Storage.

    Args:
        cloud_event: CloudEvent object containing the Pub/Sub message.
    """
    if not all([GCS_BUCKET, MULE_ORG_ID, CLIENT_ID, CLIENT_SECRET]):
        # Raise rather than return: a bare return acks the Pub/Sub message and
        # reports the run as successful, silently discarding the schedule tick.
        raise RuntimeError('Missing required environment variables')

    bucket = storage_client.bucket(GCS_BUCKET)
    state = load_state(bucket, STATE_KEY)

    now = datetime.now(timezone.utc)
    watermark = None
    if state.get('last_event_time'):
        watermark = parse_datetime(state['last_event_time'])
    seen_keys = set(state.get('seen_keys', []))

    if watermark is None:
        start_time = now - timedelta(hours=LOOKBACK_HOURS)
    else:
        start_time = watermark - timedelta(minutes=OVERLAP_MINUTES)

    print(f"Fetching audit logs from {start_time.isoformat()} to {now.isoformat()}")

    token = get_token(CLIENT_ID, CLIENT_SECRET)

    # A FetchError here propagates: the run fails, the watermark is untouched,
    # and the next invocation retries the same window.
    records, newest_event_time = fetch_audit(
        org_id=MULE_ORG_ID,
        token=token,
        start_time=start_time,
        end_time=now,
        page_size=PAGE_SIZE,
        max_records=MAX_RECORDS,
    )

    # Remove events already written by an earlier run. Without this, the overlap
    # window re-emits its events on every invocation.
    fresh = [r for r in records if not (event_keys(r) & seen_keys)]
    print(f"Fetched {len(records)} records, {len(fresh)} new after deduplication")

    if not fresh:
        print("No new audit log records found. Watermark left unchanged.")
        return

    if not newest_event_time:
        # A watermark that cannot be derived from the data is not a watermark.
        # Advancing on wall-clock here would skip every late-recorded event.
        raise FetchError('Records were returned but no event timestamp could be parsed')

    timestamp = now.strftime('%Y%m%dT%H%M%SZ')
    object_key = f"{GCS_PREFIX}/audit_{timestamp}.ndjson"
    blob = bucket.blob(object_key)

    ndjson = '\n'.join(json.dumps(record, ensure_ascii=False) for record 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}")

    # Advance the watermark only after the data is durably written.
    new_watermark = parse_datetime(newest_event_time)
    if watermark and new_watermark < watermark:
        new_watermark = watermark

    # Retain only the hashes still inside the overlap window, so the state file
    # stays small while covering every event the next query can re-read.
    cutoff = new_watermark - timedelta(minutes=OVERLAP_MINUTES)
    retained = set()
    for record in records:
        stamp = event_time(record)
        if not stamp:
            continue
        if parse_datetime(stamp) >= cutoff:
            retained.update(event_keys(record))

    save_state(bucket, STATE_KEY, {
        'last_event_time': new_watermark.isoformat(),
        'seen_keys': sorted(retained),
    })

    print(f"Successfully processed {len(fresh)} records")

def load_state(bucket, key):
    """Read the collector state from Cloud Storage.

    Only a missing object is treated as a cold start. Any other error is raised:
    swallowing it would silently reset collection to the full lookback window
    and re-ingest that entire 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: dict):
    """Write the collector state to Cloud Storage.

    Failures are raised, not logged. If the state write fails after the data was
    uploaded, the next run repeats the same window and duplicates it.
    """
    blob = bucket.blob(key)
    blob.upload_from_string(
        json.dumps(state, indent=2),
        content_type='application/json',
    )
    print(f"Saved state: last_event_time={state.get('last_event_time')}")

def retry_after_seconds(response, backoff: float) -> int:
    """Parse Retry-After defensively; it may be an HTTP date, which int() cannot parse."""
    raw = response.headers.get('Retry-After')
    try:
        return int(raw) if raw else int(backoff)
    except (TypeError, ValueError):
        return int(backoff)

def get_token(client_id: str, client_secret: str) -> str:
    """Get an OAuth 2.0 access token from MuleSoft using client credentials."""
    body = urlencode({
        'grant_type': 'client_credentials',
        'client_id': client_id,
        'client_secret': client_secret,
    }).encode('utf-8')

    backoff = 1.0
    for attempt in range(5):
        try:
            response = http.request(
                'POST',
                TOKEN_URL,
                body=body,
                headers={'Content-Type': 'application/x-www-form-urlencoded'},
            )
        except Exception as e:
            raise FetchError(f'Token request failed: {e}') from e

        if response.status == 429:
            wait = retry_after_seconds(response, backoff)
            print(f"Rate limited (429) on token request. Retrying after {wait}s...")
            time.sleep(wait)
            backoff = min(backoff * 2, 30.0)
            continue

        if response.status != 200:
            raise FetchError(f'Failed to get token: HTTP {response.status} {response.data.decode("utf-8", "replace")}')

        try:
            return json.loads(response.data.decode('utf-8'))['access_token']
        except (json.JSONDecodeError, KeyError) as e:
            raise FetchError(f'Malformed token response: {e}') from e

    raise FetchError('Rate limited repeatedly on token request; giving up without advancing the watermark')

def fetch_audit(org_id: str, token: str, start_time: datetime, end_time: datetime, page_size: int, max_records: int):
    """Fetch audit log entries from the MuleSoft Audit Log Query API.

    Uses cursor pagination. The API returns entries in ascending timestamp order,
    which a time watermark requires: with newest-first ordering a truncated run
    would advance the watermark past unread events.

    Returns:
        Tuple of (records list, newest_event_time ISO string).

    Raises:
        FetchError: on any API or transport failure, so that the caller does not
            mistake a failed fetch for an empty result and advance the watermark.
    """
    endpoint = AUDIT_URL.format(org_id=org_id)
    headers = {
        'Authorization': f'Bearer {token}',
        'Content-Type': 'application/json',
        'Accept': 'application/json',
    }

    records = []
    newest_time = None
    cursor = None
    page_num = 0
    backoff = 1.0
    rate_limit_retries = 0
    MAX_RATE_LIMIT_RETRIES = 5

    while True:
        page_num += 1

        if len(records) >= max_records:
            # Stop cleanly. The remaining events stay ahead of the watermark and
            # are collected by the next run.
            print(f"Reached max_records limit ({max_records})")
            break

        # Rebuild the full query on every page; the time window must never be dropped.
        body = {
            'startDate': to_unix_millis(start_time),
            'endDate': to_unix_millis(end_time),
            'limit': page_size,
            'ascending': True,
        }
        if cursor:
            body['cursor'] = cursor

        try:
            response = http.request(
                'POST',
                endpoint,
                body=json.dumps(body).encode('utf-8'),
                headers=headers,
            )
        except Exception as e:
            raise FetchError(f'Request to {endpoint} failed: {e}') from e

        if response.status == 429:
            rate_limit_retries += 1
            if rate_limit_retries > MAX_RATE_LIMIT_RETRIES:
                raise FetchError('Rate limited repeatedly; giving up without advancing the watermark')
            wait = retry_after_seconds(response, backoff)
            print(f"Rate limited (429). Retrying after {wait}s...")
            time.sleep(wait)
            backoff = min(backoff * 2, 30.0)
            continue

        backoff = 1.0
        rate_limit_retries = 0

        if response.status != 200:
            raise FetchError(f'HTTP {response.status} from {endpoint}: {response.data.decode("utf-8", "replace")}')

        try:
            data = json.loads(response.data.decode('utf-8'))
        except json.JSONDecodeError as e:
            raise FetchError(f'Malformed JSON response from {endpoint}: {e}') from e

        page_results = data.get('data') or []
        if not page_results:
            print("No more results (empty page)")
            break

        print(f"Page {page_num}: Retrieved {len(page_results)} entries")
        records.extend(page_results)

        for event in page_results:
            stamp = event_time(event)
            if not stamp:
                continue
            if newest_time is None or parse_datetime(stamp) > parse_datetime(newest_time):
                newest_time = stamp

        cursor = data.get('cursor')
        if not cursor:
            break

    print(f"Retrieved {len(records)} total entries from {page_num} pages")
    return records, newest_time

    • 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).

Pertimbangan penting

  • Pembatasan kapasitas: Endpoint Kueri Log Audit menerapkan pembatasan kapasitas per IP di tiga bidang kontrol. Bidang kontrol AS mengizinkan 700 permintaan per menit per IP, sedangkan bidang kontrol Uni Eropa dan Pemerintah mengizinkan 40 permintaan per menit per IP. Fungsi ini menerapkan backoff eksponensial untuk menangani pembatasan kapasitas secara otomatis.

  • Masa berlaku token: Masa berlaku token akses biasanya berakhir sekitar 30 hingga 60 menit setelah dikeluarkan. Fungsi ini meminta token baru untuk setiap eksekusi, yang berada dalam periode tersebut.

  • Penghapusan duplikat dan state file: Fungsi ini menyimpan watermark waktu peristiwa terbaru dan hash peristiwa di dalam jendela tumpang-tindih dalam state file. Setiap proses membaca ulang dua menit sebelum tanda air untuk mengambil entri yang terlambat direkam dan menghapus entri yang sudah ditulis, sehingga percobaan ulang, proses manual, atau penundaan penjadwal tidak menghasilkan peristiwa duplikat. Panggilan API yang gagal akan menggagalkan operasi dan membuat watermark tidak berubah, sehingga operasi berikutnya akan mencoba kembali jendela yang sama.

  • Retensi log audit: Log audit memiliki periode retensi default selama satu tahun. Jika organisasi Anda dibuat sebelum 10 Juli 2023 dan Anda tidak mengubah periode retensi secara manual, periode retensinya adalah enam tahun. Download log secara berkala jika Anda perlu menyimpannya di luar periode retensi yang dikonfigurasi.

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 mulesoft-audit-collector-hourly
    Region Pilih region yang sama dengan fungsi Cloud Run
    Frekuensi 0 * * * * (setiap jam, tepat pada jamnya)
    Zona Waktu Pilih zona waktu (disarankan UTC)
    Jenis target Pub/Sub
    Topik Pilih topik mulesoft-audit-trigger
    Isi pesan {} (objek JSON kosong)

  4. Klik Create.

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 > mulesoft-audit-collector > Logs.
  4. Pastikan fungsi berhasil dieksekusi.
  5. Periksa bucket Cloud Storage untuk mengonfirmasi bahwa log telah ditulis.

Mengonfigurasi feed di Google SecOps untuk menyerap log MuleSoft

  1. Buka Setelan SIEM > Feed.
  2. Klik Tambahkan Feed Berita Baru.
  3. Klik Konfigurasi satu feed.
  4. Di kolom Nama feed, masukkan nama untuk feed (misalnya, MuleSoft Logs).
  5. Pilih Google Cloud Storage V2 sebagai Source type.
  6. Pilih Mulesoft sebagai Log type.

  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. Anda akan menggunakannya pada langkah berikutnya.

  9. Klik Berikutnya.

  10. Tentukan nilai untuk parameter input berikut:

    • URL bucket penyimpanan: Masukkan URI bucket Cloud Storage:

      gs://mulesoft-audit-logs/
      • Ganti mulesoft-audit-logs dengan nama bucket yang sebenarnya.

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

  11. Klik Berikutnya.

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

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.
    • Assign roles: Pilih kedua peran 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.

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