Mengumpulkan log Proofpoint Secure Email Relay

Didukung di:

Dokumen ini menjelaskan cara menyerap log Proofpoint Secure Email Relay ke Google Security Operations menggunakan Cloud Storage V2.

Proofpoint Secure Email Relay (SER) adalah relai terkelola untuk email yang dibuat aplikasi, seperti pesan transaksi dan notifikasi. SER melaporkan traffic tersebut melalui Reporting API-nya, salah satu API konfigurasi SER. API ini menampilkan angka penggunaan gabungan: posisi throughput berlisensi, jumlah dan volume pesan harian, serta penghitung untuk setiap pengguna relay. Parser memetakan angka tersebut ke Model Data Terpadu (UDM), mencatatnya sebagai label bersama identitas pengguna relay.

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 Cloud Run, topik Pub/Sub, dan tugas Cloud Scheduler
  • Akses administrator ke portal Proofpoint Secure Email Relay, termasuk izin untuk membuat kunci API

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, proofpoint-ser-logs)
    Location type Pilih berdasarkan kebutuhan Anda (Region, Dual-region, Multi-region)
    Location Pilih lokasi (misalnya, us-central1)
    Storage class Standar (direkomendasikan untuk log yang sering diakses)
    Access control Seragam (direkomendasikan)
    Alat perlindungan Opsional: Aktifkan pembuatan versi objek atau kebijakan retensi

  6. Klik Create.

Kumpulkan kredensial Proofpoint SER API

Akses ke Reporting API diberikan oleh kunci API yang dikeluarkan dari layanan Pengelolaan Kunci API Proofpoint. Kunci tidak dibuat di dalam konsol SER itu sendiri, itulah sebabnya kunci tidak muncul di salah satu entri navigasi SER.

Membuat kunci API

  1. Login ke portal Proofpoint Secure Email Relay dengan kredensial administrator.
  2. Buka Pengalih Aplikasi di pojok kiri atas.
  3. Buka Services > API Key Management.
  4. Klik Buat Kunci.
  5. Pilih Secure Email Relay sebagai produk.

  6. Salin dan simpan nilai berikut dengan aman:

    • Kunci: digunakan sebagai client_id
    • Secret: digunakan sebagai client_secret

Verifikasi izin

Layanan Pengelolaan Kunci API digunakan bersama di beberapa produk Proofpoint, itulah sebabnya Anda mengaksesnya melalui Pengalih Aplikasi, bukan menu navigasi SER. Untuk mendapatkan akses ke API konfigurasi SER, termasuk Reporting API, Anda harus memilih Secure Email Relay sebagai produk saat membuat kunci.

Jika Pengelolaan Kunci API tidak muncul di Pengalih Aplikasi, berarti akun Anda tidak memiliki izin pengelolaan API. Hubungi administrator Proofpoint atau tim akun Anda agar mereka dapat memberi Anda izin.

Menguji akses API

  • Reporting API menggunakan pemberian kredensial klien OAuth 2.0. Tukar kunci dan secret dengan token pembawa, lalu panggil API dengan token tersebut.

    # Replace with the values you copied
CLIENT_ID="<your-key>"
CLIENT_SECRET="<your-secret>"

# 1. Exchange the key and secret for an access token
ACCESS_TOKEN=$(curl -s -X POST "https://auth.proofpoint.com/v1/token" \
  -H "Content-Type: application/x-www-form-urlencoded" \
  -d "grant_type=client_credentials" \
  -d "client_id=${CLIENT_ID}" \
  -d "client_secret=${CLIENT_SECRET}" \
  | python3 -c "import json,sys; print(json.load(sys.stdin)['access_token'])")

# 2. Call the Reporting API with the token
curl -v -H "Authorization: Bearer ${ACCESS_TOKEN}" \
  "https://reporting.ser.proofpoint.com/v1/usage/overview"

Panggilan yang berhasil akan menampilkan objek JSON yang anggota data-nya berisi throughputLimit, licenseStartDate, licenseEndDate, dan rata-rata throughput.

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 GCP Console, buka IAM & Admin > Service Accounts.
  2. Klik Create Service Account.
  3. Berikan detail konfigurasi berikut:
    • Nama akun layanan: Masukkan proofpoint-ser-collector-sa
    • Deskripsi akun layanan: Masukkan Service account for Cloud Run function to collect Proofpoint Secure Email Relay 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 (misalnya, proofpoint-ser-logs).
  3. Buka tab Izin.
  4. Klik Grant access.
  5. Berikan detail konfigurasi berikut:
    • Tambahkan prinsipal: Masukkan email akun layanan (misalnya, proofpoint-ser-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 Konsol GCP, buka Pub/Sub > Topics.
  2. Klik Create topic.
  3. Berikan detail konfigurasi berikut:
    • ID Topik: Masukkan proofpoint-ser-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 laporan penggunaan dari SER Reporting API dan menuliskannya 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 Konfigurasi, berikan detail konfigurasi berikut:

    Setelan Nilai
    Nama layanan proofpoint-ser-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 Pub/Sub (proofpoint-ser-trigger).
    4. Klik Simpan.

  6. Di bagian Authentication:

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

  7. Buka dan luaskan Containers, Networking, Security.

  8. Buka tab Security:

    • Akun layanan: Pilih akun layanan (proofpoint-ser-collector-sa)

  9. Buka tab Containers:

    1. Klik Variables & Secrets.
    2. Klik + Add variable untuk setiap variabel lingkungan:
    Nama Variabel Nilai Contoh Deskripsi
    GCS_BUCKET proofpoint-ser-logs Nama bucket Cloud Storage
    GCS_PREFIX ser-logs Awalan untuk file log
    STATE_KEY ser-logs-state.json Jalur status, di luar awalan log
    TOKEN_URL https://auth.proofpoint.com/v1/token Endpoint token OAuth 2.0
    API_BASE https://reporting.ser.proofpoint.com URL dasar SER Reporting API
    CLIENT_ID your-key Key dari Pengelolaan Kunci API
    CLIENT_SECRET your-secret Secret dari Pengelolaan Kunci API
    PAGE_SIZE 50 Pengguna Relay yang diminta per halaman
    MAX_PAGES 200 Batas halaman untuk penomoran halaman pengguna relai
    REPORT_DAY_OFFSET 1 Laporkan pada hari ini sekian hari yang lalu
    MAX_BACKFILL_DAYS 7 Sebagian besar hari, satu kali penayangan iklan akan mengisi ulang
    SEEN_RETENTION_DAYS 7 Jumlah hari kunci penghapusan duplikat disimpan dalam status

  10. Di bagian Variabel & Secret, buka Requests:

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

  11. Buka tab Setelan:

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

  12. Di bagian Penskalaan revisi:

    • Minimum number of instances: Masukkan 0
    • Jumlah maksimum instance: Masukkan 100 (atau sesuaikan berdasarkan perkiraan beban)

  13. Klik Create.

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

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

Tambahkan 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
from google.cloud.exceptions import NotFound
import hashlib
import json
import os
import time
import urllib.parse
import urllib3
from datetime import date, datetime, timezone, timedelta

# 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()

# Environment variables
GCS_BUCKET = os.environ.get('GCS_BUCKET')
GCS_PREFIX = os.environ.get('GCS_PREFIX', 'ser-logs')
# 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', 'ser-logs-state.json')
TOKEN_URL = os.environ.get('TOKEN_URL', 'https://auth.proofpoint.com/v1/token')
API_BASE = os.environ.get('API_BASE', 'https://reporting.ser.proofpoint.com')
CLIENT_ID = os.environ.get('CLIENT_ID')
CLIENT_SECRET = os.environ.get('CLIENT_SECRET')
PAGE_SIZE = int(os.environ.get('PAGE_SIZE', '50'))
MAX_PAGES = int(os.environ.get('MAX_PAGES', '200'))
# The Reporting API filters on whole calendar days, so the collector reports on
# a day that has already finished rather than on a partial one.
REPORT_DAY_OFFSET = int(os.environ.get('REPORT_DAY_OFFSET', '1'))
MAX_BACKFILL_DAYS = int(os.environ.get('MAX_BACKFILL_DAYS', '7'))
SEEN_RETENTION_DAYS = int(os.environ.get('SEEN_RETENTION_DAYS', '7'))

MAX_RATE_LIMIT_RETRIES = 5

class FetchError(Exception):
    """Raised when a Proofpoint SER API call fails.

    Collection must fail loudly. Returning an empty result on an API error is
    indistinguishable from a day with no traffic, and would let the run record
    a day it never actually read.
    """

def get_access_token():
    """Exchange the API key and secret for an OAuth 2.0 bearer token.

    The SER configuration APIs use the client credentials grant with the
    credentials in the form body. The returned token is sent to each API
    host as an Authorization: Bearer header.
    """
    body = urllib.parse.urlencode({
        'grant_type': 'client_credentials',
        'client_id': CLIENT_ID,
        'client_secret': CLIENT_SECRET,
    })
    headers = {
        'Content-Type': 'application/x-www-form-urlencoded',
        'Accept': 'application/json',
    }

    try:
        response = http.request('POST', TOKEN_URL, body=body, headers=headers)
    except Exception as e:
        raise FetchError(f'Token request to {TOKEN_URL} failed: {e}') from e

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

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

    if not token:
        raise FetchError('Token response did not contain an access_token')

    return token

def call_api(token, method, path, body=None):
    """Call one Reporting API endpoint and return the decoded JSON body.

    Every failure raises. A caller that mistook an error for an empty report
    would record the day as collected and never come back to it.
    """
    url = f'{API_BASE}{path}'
    headers = {
        'Authorization': f'Bearer {token}',
        'Accept': 'application/json',
        'Content-Type': 'application/json',
        'User-Agent': 'GoogleSecOps-ProofpointSERCollector/1.0',
    }
    payload = json.dumps(body) if body is not None else None
    backoff = 1.0
    retries = 0

    while True:
        try:
            response = http.request(method, url, body=payload, headers=headers)
        except Exception as e:
            raise FetchError(f'Request to {url} failed: {e}') from e

        if response.status == 429:
            retries += 1
            if retries > MAX_RATE_LIMIT_RETRIES:
                raise FetchError(f'Rate limited repeatedly by {url}; giving up without recording the day')
            raw_retry_after = response.headers.get('Retry-After')
            try:
                # Retry-After may also be an HTTP date, which int() cannot parse.
                delay = int(raw_retry_after) if raw_retry_after else int(backoff)
            except (TypeError, ValueError):
                delay = int(backoff)
            print(f'Rate limited (429) on {path}. Retrying after {delay}s...')
            time.sleep(delay)
            backoff = min(backoff * 2, 30.0)
            continue

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

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

def as_list(value):
    """Return a response data member as a list.

    The Reporting API returns data as an object on some endpoints and as an
    array on others, and the published schema declares neither.
    """
    if value is None:
        return []
    if isinstance(value, list):
        return value
    return [value]

def collect_day(token, day, include_overview):
    """Collect one reporting day and return the records to write.

    The nesting of each record is chosen to match what the parser expects.
    Rows carrying acceptedMessages are emitted bare, because the parser nests
    those under data itself. The other two responses are emitted whole, so
    that their data and metadata envelope survives to the parser.
    """
    records = []

    # Licensed throughput position. This endpoint takes no date and reports
    # the tenant's standing right now, so it is collected once per run rather
    # than once per day: a backfill would otherwise write the same snapshot
    # against every day it catches up on.
    if include_overview:
        records.append(call_api(token, 'GET', '/v1/usage/overview'))

    # Volume in bytes for the day. One envelope per row, so that the
    # totalThroughput in metadata stays attached to the row it describes.
    data_trend = call_api(token, 'POST', '/v1/usage/data-trend', {
        'dates': day,
        'interval': 'day',
    })
    for element in as_list(data_trend.get('data')):
        records.append({'data': [element], 'metadata': data_trend.get('metadata', {})})

    # Message counters for the day.
    message_trend = call_api(token, 'POST', '/v1/usage/message-trend', {
        'dates': day,
        'interval': 'day',
    })
    records.extend(as_list(message_trend.get('data')))

    # Per relay user counters, paginated.
    page = 1
    while True:
        if page > MAX_PAGES:
            raise FetchError(f'Relay user pagination for {day} exceeded MAX_PAGES ({MAX_PAGES})')

        payload = call_api(token, 'POST', '/v1/usage/relay-users', {
            'dates': day,
            'pageNum': page,
            'pageSize': PAGE_SIZE,
        })
        elements = as_list(payload.get('data'))
        if not elements:
            break

        records.extend(elements)

        pagination = (payload.get('metadata') or {}).get('pagination') or {}
        total_pages = pagination.get('totalPages')
        if not total_pages or page >= total_pages:
            break
        page += 1

    return records

def record_key(record):
    """Return the deduplication identity of one record.

    Reporting rows are aggregates and carry no identifier of their own, so
    identity is the content hash. Keys are held per reporting day, which makes
    a repeated run of the same day a no-op without ever suppressing a new day
    whose figures happen to be identical.
    """
    return 'sha256:' + hashlib.sha256(
        json.dumps(record, sort_keys=True, ensure_ascii=False).encode('utf-8')
    ).hexdigest()

def pending_days(last_report_date, target):
    """Return the reporting days still to collect, oldest first."""
    if not last_report_date:
        start = target
    else:
        start = date.fromisoformat(last_report_date) + timedelta(days=1)

    if start > target:
        return []

    if (target - start).days >= MAX_BACKFILL_DAYS:
        start = target - timedelta(days=MAX_BACKFILL_DAYS - 1)
        print(f'Backfill capped at {MAX_BACKFILL_DAYS} days. Days before {start.isoformat()} are not collected.')

    return [start + timedelta(days=offset) for offset in range((target - start).days + 1)]

def prune_seen(seen, target):
    """Drop deduplication keys for days outside the retention window."""
    cutoff = target - timedelta(days=SEEN_RETENTION_DAYS)
    return {day: keys for day, keys in seen.items() if date.fromisoformat(day) >= cutoff}

@functions_framework.cloud_event
def main(cloud_event):
    """Fetch Proofpoint SER usage reports and write them to Cloud Storage.

    Args:
        cloud_event: CloudEvent object containing the Pub/Sub message.
    """
    if not all([GCS_BUCKET, 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)
    seen = state.get('seen') or {}

    now = datetime.now(timezone.utc)
    target = (now - timedelta(days=REPORT_DAY_OFFSET)).date()

    days = pending_days(state.get('last_report_date'), target)
    if not days:
        print(f'Nothing to collect. {target.isoformat()} is already recorded.')
        return

    token = get_access_token()

    for day in days:
        key = day.isoformat()
        print(f'Collecting reporting day {key}')

        # A FetchError here propagates: the run fails, the state is left at the
        # last day that was fully written, and the next run retries this day.
        records = collect_day(token, key, include_overview=(day == days[-1]))

        day_seen = set(seen.get(key, []))
        fresh = []
        for record in records:
            digest = record_key(record)
            if digest in day_seen:
                continue
            day_seen.add(digest)
            fresh.append(record)

        print(f'{key}: fetched {len(records)} records, {len(fresh)} new after deduplication')

        if fresh:
            timestamp = now.strftime('%Y%m%dT%H%M%SZ')
            object_key = f'{GCS_PREFIX}/usage_{key}_{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}')

        # Record the day only after its data is durably written.
        seen[key] = sorted(day_seen)
        state['last_report_date'] = key
        state['seen'] = prune_seen(seen, target)
        save_state(bucket, STATE_KEY, state)

    print(f'Successfully processed {len(days)} reporting day(s)')

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 restart collection and re-ingest the
    whole backfill window.
    """
    blob = bucket.blob(key)
    try:
        return json.loads(blob.download_as_text())
    except NotFound:
        print('No state file found. Starting from the most recent completed day.')
        return {}

def save_state(bucket, key, state):
    """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 day 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_report_date={state.get("last_report_date")}')

  • File kedua - requirements.txt:

    functions-framework==3.*
google-cloud-storage==2.*
urllib3>=2.0.0
  1. Klik Deploy untuk menyimpan dan men-deploy fungsi.

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

Buat tugas Cloud Scheduler

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

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

  3. Berikan detail konfigurasi berikut:

    Setelan Nilai
    Nama proofpoint-ser-collector-daily
    Region Pilih region yang sama dengan fungsi Cloud Run
    Frekuensi 0 2 * * * (harian, pukul 02.00)
    Zona Waktu Pilih zona waktu (UTC direkomendasikan)
    Jenis target Pub/Sub
    Topik Pilih topik Pub/Sub (proofpoint-ser-trigger)
    Isi pesan {} (objek JSON kosong)

  4. Klik Create.

Opsi frekuensi jadwal

Reporting API mengagregasi menurut hari kalender, sehingga hari yang selesai dikumpulkan satu kali. Jalankan tugas setiap hari, setelah hari pelaporan ditutup di zona waktu tenant Anda:

Frekuensi Ekspresi Cron Kasus Penggunaan
Setiap hari pukul 02.00 0 2 * * * Standar (direkomendasikan)
Setiap hari pukul 06.00 0 6 * * * Penyewa yang pelaporannya diselesaikan nanti
Dua kali sehari 0 2,14 * * * Menambahkan percobaan ulang pada hari yang sama jika proses pertama gagal

Run yang menemukan harinya sudah direkam akan keluar tanpa menulis apa pun, sehingga run tambahan dikenai biaya satu panggilan API dan tidak pernah menduplikasi data.

Menguji integrasi

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

  7. Pastikan fungsi berhasil dieksekusi. Cari:

    Collecting reporting day YYYY-MM-DD
YYYY-MM-DD: fetched X records, X new after deduplication
Wrote X records to gs://proofpoint-ser-logs/ser-logs/usage_YYYY-MM-DD_YYYYMMDDTHHMMSSZ.ndjson
Saved state: last_report_date=YYYY-MM-DD
Successfully processed 1 reporting day(s)

  8. Buka Cloud Storage > Buckets.

  9. Klik nama bucket Anda (proofpoint-ser-logs).

  10. Buka folder awalan (ser-logs/).

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

Jika Anda melihat error dalam log:

  • HTTP 400 dari endpoint token: Kunci atau rahasia salah, atau kunci telah berakhir. invalid_client berarti pasangan ditolak; buat kunci pengganti.
  • HTTP 401 dari Reporting API: Token pembawa tidak dikirim atau tidak lagi valid.
  • HTTP 429: Pembatasan laju. Fungsi akan mencoba lagi dengan penundaan dan gagal menjalankan jika batas tetap ada, sehingga hari tersebut akan dialihkan ke eksekusi berikutnya.
  • Variabel lingkungan tidak ada: Periksa apakah semua variabel yang diperlukan telah ditetapkan.

Mengambil akun layanan Google SecOps

Google SecOps menggunakan akun layanan unik untuk membaca data dari bucket 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, Proofpoint SER Logs).
  5. Pilih Google Cloud Storage V2 sebagai Source type.
  6. Pilih ProofPoint Secure Email Relay sebagai Jenis log.
  7. Klik Get Service Account.

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

    chronicle-12345678@chronicle-gcp-prod.iam.gserviceaccount.com

  9. Salin alamat email ini untuk digunakan di langkah berikutnya.

  10. Klik Berikutnya.

  11. Tentukan nilai untuk parameter input berikut:

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

      gs://proofpoint-ser-logs/ser-logs/
      • Ganti:
        • proofpoint-ser-logs: Nama bucket Cloud Storage Anda.
        • ser-logs: Awalan atau jalur folder opsional tempat log disimpan (kosongkan untuk root).

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

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

    • Namespace aset: Namespace aset

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

  12. Klik Berikutnya.

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

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 (misalnya, proofpoint-ser-logs).
  3. Buka tab Izin.
  4. Klik Grant access.
  5. Berikan detail konfigurasi berikut:
    • Add principals: Tempel email akun layanan Google SecOps
    • Tetapkan peran: Pilih keduanya:
      • 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.

Tabel pemetaan UDM

Kolom Log Pemetaan UDM Logika
acceptedMessages_label additional.fields Digabung
acceptedThroughput_label additional.fields Digabung
average30DayThroughput_label additional.fields Digabung
average7DayThroughput_label additional.fields Digabung
averageDailyThroughput_label additional.fields Digabung
avgAcceptedMessageSize_label additional.fields Digabung
blockedMessages_label additional.fields Digabung
data_totalThroughput_label additional.fields Digabung
deliveredMessages_label additional.fields Digabung
licenseEndDate_label additional.fields Digabung
licenseStartDate_label additional.fields Digabung
map_label additional.fields Digabung
metadata_totalThroughput_label additional.fields Digabung
quarantinedMessages_label additional.fields Digabung
rejectedMessages_label additional.fields Digabung
remainingThroughput_label additional.fields Digabung
requestedMessages_label additional.fields Digabung
requestedThroughput_label additional.fields Digabung
sentMessages_label additional.fields Digabung
throughputForecast_label additional.fields Digabung
throughputLimit_label additional.fields Digabung
throughput_label additional.fields Digabung
totalMessages_label additional.fields Digabung
undeliveredMessages_label additional.fields Digabung
data.name metadata.description Dipetakan secara langsung
desc metadata.description Dipetakan secara langsung
data.date metadata.event_timestamp Diurai sebagai yyyy-MM-dd
event_type metadata.event_type Dipetakan secara langsung
fromEnvelope network.email.bounce_address Dipetakan secara langsung
fromHeader network.email.from Dipetakan secara langsung
applicationName principal.administrative_domain Dipetakan secara langsung
principal_host principal.asset.hostname Dipetakan secara langsung
principal_host principal.hostname Dipetakan secara langsung
principal_port principal.port Dipetakan secara langsung
data.relayUserId principal.user.product_object_id Dipetakan secara langsung
userId principal.user.product_object_id Dipetakan secara langsung
applicationUserName principal.user.user_display_name Dipetakan secara langsung
senderName target.administrative_domain Dipetakan secara langsung
senderId target.user.product_object_id Dipetakan secara langsung
T/A metadata.product_name Konstanta: PROOFPOINT SER
T/A metadata.vendor_name Konstanta: PROOFPOINT

Log Perubahan

Melihat Log Perubahan untuk parser ini

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