איסוף יומנים של Proofpoint Secure Email Relay

נתמך ב:

במאמר הזה מוסבר איך להטמיע יומנים של Proofpoint Secure Email Relay ב-Google Security Operations באמצעות Cloud Storage V2.

‫Proofpoint Secure Email Relay (SER) הוא שרת ממסר מנוהל לדואר שנוצר על ידי אפליקציות, כמו הודעות טרנזקציה והתראות. ה-SER מדווח על התנועה הזו באמצעות Reporting API, אחד מממשקי ה-API של הגדרת ה-SER. ה-API מחזיר נתוני שימוש מצטברים: מיקום התפוקה המורשית, נפח יומי ומספר ההודעות, ומונים לכל משתמש בשרת הדואר. הכלי לניתוח מבצע מיפוי של הנתונים האלה למודל הנתונים המאוחד (UDM), ומתעד אותם כתוויות לצד זהות המשתמש בממסר.

לפני שמתחילים

חשוב לוודא שמתקיימות הדרישות המוקדמות הבאות:

  • מופע של Google SecOps
  • פרויקט ב- Google Cloud עם Cloud Storage API מופעל
  • הרשאות ליצירה ולניהול של קטגוריות ב-Cloud Storage
  • הרשאות לניהול כללי מדיניות של ניהול זהויות והרשאות גישה (IAM) בדליים של Cloud Storage
  • הרשאות ליצירת שירותי Cloud Run, נושאי Pub/Sub ומשימות Cloud Scheduler
  • גישת אדמין לפורטל Proofpoint Secure Email Relay, כולל הרשאה ליצור מפתחות API

יצירת קטגוריה של Cloud Storage

  1. עוברים אל Google Cloud המסוף.
  2. בוחרים פרויקט או יוצרים פרויקט חדש.
  3. בתפריט הניווט, עוברים אל Cloud Storage > Buckets.
  4. לוחצים על Create bucket.

  5. מזינים את פרטי ההגדרה הבאים:

    הגדרה ערך
    Name your bucket (שם הקטגוריה) מזינים שם ייחודי גלובלית (לדוגמה, proofpoint-ser-logs)
    סוג מיקום בוחרים לפי הצרכים שלכם (אזור, בשני אזורים, במספר אזורים)
    Location בוחרים את המיקום (לדוגמה, us-central1).
    סוג אחסון (storage class) רגיל (מומלץ ליומנים שמתבצעת אליהם גישה לעיתים קרובות)
    בקרת גישה אחיד (מומלץ)
    כלים להגנה אופציונלי: הפעלת ניהול גרסאות של אובייקטים או מדיניות שמירת נתונים

  6. לוחצים על יצירה.

איסוף פרטי כניסה ל-Proofpoint SER API

הגישה ל-Reporting API ניתנת באמצעות מפתח API שהונפק משירות ניהול מפתחות ה-API של Proofpoint. המפתח לא נוצר בתוך מסוף SER עצמו, ולכן הוא לא מופיע באף אחת מהאפשרויות של הניווט ב-SER.

יצירה של מפתח API

  1. נכנסים לפורטל Proofpoint Secure Email Relay באמצעות פרטי כניסה של אדמין.
  2. פותחים את ממשק למעבר בין חשבונות בפינה הימנית העליונה.
  3. עוברים אל Services > API Key Management (שירותים > ניהול מפתחות API).
  4. לוחצים על Create Key (יצירת מפתח).
  5. בוחרים את המוצר Secure Email Relay.

  6. מעתיקים את הערכים הבאים ומאחסנים אותם בצורה מאובטחת:

    • ‫Key: משמש כ-client_id
    • סוד: משמש כclient_secret

אימות ההרשאות

שירות ניהול מפתחות API משותף לכמה מוצרים של Proofpoint, ולכן הגישה אליו היא דרך ממשק למעבר בין חשבונות ולא דרך תפריט הניווט של SER. כדי לקבל גישה לממשקי ה-API של הגדרות SER, כולל Reporting API, צריך לבחור באפשרות Secure Email Relay בתור המוצר כשיוצרים את המפתח.

אם האפשרות ניהול מפתחות API לא מופיעה בממשק למעבר בין חשבונות, סימן שלחשבון שלכם חסרה הרשאת ניהול API. כדי לקבל הרשאה, צריך לפנות לאדמין של Proofpoint או לצוות ניהול החשבון.

בדיקת הגישה ל-API

  • ‫Reporting API משתמש במענק פרטי הכניסה של לקוח OAuth 2.0. מחליפים את המפתח והסוד בטוקן מסוג bearer, ואז מפעילים את ה-API עם הטוקן הזה.

    # 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"

קריאה מוצלחת מחזירה אובייקט JSON שהאיבר data שלו מכיל את הערכים throughputLimit, licenseStartDate, licenseEndDate ואת ממוצעי התפוקה.

יצירת חשבון שירות לפונקציית Cloud Run

פונקציית Cloud Run צריכה חשבון שירות עם הרשאות כתיבה לקטגוריית Cloud Storage והרשאות הפעלה על ידי Pub/Sub.

יצירת חשבון השירות

  1. במסוף GCP, נכנסים אל IAM & Admin > Service Accounts.
  2. לוחצים על יצירת חשבון שירות.
  3. מזינים את פרטי ההגדרה הבאים:
    • שם חשבון השירות: מזינים proofpoint-ser-collector-sa
    • תיאור חשבון השירות: מזינים Service account for Cloud Run function to collect Proofpoint Secure Email Relay logs
  4. לוחצים על יצירה והמשך.
  5. בקטע Grant this service account access to project (מתן גישה של חשבון השירות הזה לפרויקט), מוסיפים את התפקידים הבאים:
    1. לוחצים על בחירת תפקיד.
    2. מחפשים את Storage Object Admin ובוחרים בו.
    3. לוחצים על + הוספת תפקיד נוסף.
    4. מחפשים את התפקיד Cloud Run Invoker (הפעלת שירותים ב-Cloud Run) ולוחצים עליו.
    5. לוחצים על + הוספת תפקיד נוסף.
    6. מחפשים את האפשרות Cloud Functions Invoker (הפעלת פונקציות של Cloud Functions) ולוחצים עליה.
  6. לוחצים על Continue.
  7. לוחצים על סיום.

התפקידים האלה נדרשים כדי:

  • ‫Storage Object Admin: כתיבת יומנים לקטגוריה של Cloud Storage וניהול קובצי מצב
  • ‫Cloud Run Invoker: מתן הרשאה ל-Pub/Sub להפעיל את הפונקציה
  • Cloud Functions Invoker: מאפשר הפעלה של פונקציות

הענקת הרשאות IAM בקטגוריית Cloud Storage

נותנים לחשבון השירות הרשאות כתיבה בקטגוריה של Cloud Storage:

  1. נכנסים אל Cloud Storage > Buckets.
  2. לוחצים על שם הקטגוריה (לדוגמה, proofpoint-ser-logs).
  3. עוברים לכרטיסייה Permissions.
  4. לוחצים על הענקת גישה.
  5. מזינים את פרטי ההגדרה הבאים:
    • Add principals (הוספת ישויות מורשות): מזינים את כתובת האימייל בחשבון השירות (לדוגמה, proofpoint-ser-collector-sa@PROJECT_ID.iam.gserviceaccount.com).
    • הקצאת תפקידים: בוחרים באפשרות אדמין של אובייקט אחסון.
  6. לוחצים על Save.

יוצרים נושא Pub/Sub

יוצרים נושא Pub/Sub ש-Cloud Scheduler יפרסם בו ופונקציית Cloud Run תירשם אליו.

  1. במסוף GCP, עוברים אל Pub/Sub > Topics.
  2. לוחצים על יצירת נושא.
  3. מזינים את פרטי ההגדרה הבאים:
    • מזהה הנושא: מזינים proofpoint-ser-trigger
    • השארת ההגדרות האחרות כברירת מחדל
  4. לוחצים על יצירה.

יצירת פונקציית Cloud Run לאיסוף יומנים

הפונקציה של Cloud Run תופעל על ידי הודעות Pub/Sub מ-Cloud Scheduler כדי לאחזר דוחות שימוש מ-SER Reporting API ולכתוב אותם ב-Cloud Storage.

  1. במסוף GCP, עוברים אל Cloud Run.
  2. לוחצים על יצירת שירות.
  3. בוחרים באפשרות Function (שימוש בעורך מוטבע ליצירת פונקציה).

  4. בקטע Configure (הגדרה), מזינים את פרטי ההגדרה הבאים:

    הגדרה ערך
    שם השירות proofpoint-ser-collector
    אזור בוחרים אזור שמתאים לקטגוריה של Cloud Storage (לדוגמה, us-central1)
    זמן ריצה בוחרים באפשרות Python 3.12 ואילך

  5. בקטע טריגר (אופציונלי):

    1. לוחצים על + הוספת טריגר.
    2. בוחרים באפשרות Cloud Pub/Sub.
    3. בקטע Select a Cloud Pub/Sub topic, בוחרים את נושא ה-Pub/Sub ‏ (proofpoint-ser-trigger).
    4. לוחצים על Save.

  6. בקטע אימות:

    1. לוחצים על דרישת אימות.
    2. בודקים את ניהול זהויות והרשאות גישה (IAM).

  7. עוברים אל Containers, Networking, Security (מאגרי מידע, רשתות, אבטחה) ומרחיבים אותו.

  8. עוברים לכרטיסייה אבטחה:

    • חשבון שירות: בוחרים את חשבון השירות (proofpoint-ser-collector-sa)

  9. עוברים לכרטיסייה מאגרי תגים:

    1. לוחצים על Variables & Secrets (משתנים וסודות).
    2. לוחצים על + הוספת משתנה לכל משתנה סביבתי:
    שם המשתנה ערך לדוגמה תיאור
    GCS_BUCKET proofpoint-ser-logs שם הקטגוריה של Cloud Storage
    GCS_PREFIX ser-logs קידומת לקובצי יומן
    STATE_KEY ser-logs-state.json נתיב המצב, מחוץ לקידומת היומן
    TOKEN_URL https://auth.proofpoint.com/v1/token נקודת קצה (endpoint) של טוקן OAuth 2.0
    API_BASE https://reporting.ser.proofpoint.com כתובת הבסיס של SER Reporting API
    CLIENT_ID your-key המפתח מניהול מפתחות API
    CLIENT_SECRET your-secret הסוד מניהול מפתחות API
    PAGE_SIZE 50 מספר המשתמשים שביקשו להשתמש בשרת הדואר לניתוב הודעות בכל דף
    MAX_PAGES 200 המספר המקסימלי של דפים להחלפת דפים (פגינציה) של משתמש שרת ממסר
    REPORT_DAY_OFFSET 1 דיווח על היום הזה לפני מספר הימים הזה
    MAX_BACKFILL_DAYS 7 ברוב הימים, הרצה אחת תבצע מילוי חוסרים
    SEEN_RETENTION_DAYS 7 מספר הימים שבהם מפתחות לביטול כפילויות נשמרים במצב

  10. בקטע Variables & Secrets (משתנים וסודות), עוברים אל Requests (בקשות):

    • זמן קצוב לתפוגת בקשה: מזינים 600 שניות (10 דקות)

  11. עוברים לכרטיסייה הגדרות:

    • בקטע מקורות מידע:
      • זיכרון: בוחרים באפשרות 512MiB או יותר
      • מעבד: בוחרים באפשרות 1

  12. בקטע Revision scaling:

    • מספר מינימלי של מופעים: מזינים 0
    • מספר מקסימלי של מופעים: מזינים 100 (או משנים בהתאם לעומס הצפוי)

  13. לוחצים על יצירה.

  14. ממתינים ליצירת השירות (דקה או שתיים).

  15. אחרי שיוצרים את השירות, עורך הקוד המוטבע ייפתח באופן אוטומטי.

הוספת קוד הפונקציה

  1. מזינים main בשדה נקודת כניסה.
  2. בעורך הקוד המוטבע, יוצרים שני קבצים:

  • הקובץ הראשון – 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")}')

  • קובץ שני – requirements.txt:

    functions-framework==3.*
google-cloud-storage==2.*
urllib3>=2.0.0
  1. לוחצים על Deploy (פריסה) כדי לשמור ולפרוס את הפונקציה.

  2. ממתינים עד שהפריסה תושלם (2-3 דקות).

יצירת משימה ב-Cloud Scheduler

‫Cloud Scheduler יפרסם הודעות בנושא Pub/Sub במרווחי זמן קבועים, ויפעיל את הפונקציה של Cloud Run.

  1. במסוף GCP, עוברים אל Cloud Scheduler.
  2. לוחצים על יצירת משימה.

  3. מזינים את פרטי ההגדרה הבאים:

    הגדרה ערך
    שם proofpoint-ser-collector-daily
    אזור בחירה באותו אזור כמו פונקציית Cloud Run
    תדירות ‫0 2 * * * (יומי, בשעה 02:00)
    אזור זמן בחירת אזור זמן (מומלץ להשתמש ב-UTC)
    סוג הטירגוט Pub/Sub
    נושא בוחרים את נושא Pub/Sub (proofpoint-ser-trigger).
    גוף ההודעה ‫{} (אובייקט JSON ריק)

  4. לוחצים על יצירה.

אפשרויות לתדירות התזמון

ה-Reporting API צובר נתונים לפי יום קלנדרי, כך שנתונים של יום שהסתיים נאספים פעם אחת. מריצים את העבודה מדי יום, אחרי שהסתיים יום הדיווח באזור הזמן של הדייר:

תדירות ביטוי cron תרחיש לדוגמה
כל יום בשעה 02:00 0 2 * * * רגיל (מומלץ)
כל יום בשעה 06:00 0 6 * * * דיירים שהדיווח שלהם מסתיים מאוחר יותר
פעמיים ביום 0 2,14 * * * הוספת ניסיון חוזר באותו יום אם ההרצה הראשונה נכשלה

אם מריצים את התהליך ביום שכבר נרשם, הוא מסתיים בלי לכתוב כלום, כך שהרצה נוספת עולה קריאה אחת ל-API ולא יוצרת כפילויות של נתונים.

בדיקת השילוב

  1. במסוף Cloud Scheduler, מאתרים את העבודה.
  2. לוחצים על Force run (הפעלה בכוח) כדי להפעיל את המשימה באופן ידני.
  3. ממתינים כמה שניות.
  4. עוברים אל Cloud Run > Services (שירותים).
  5. לוחצים על שם הפונקציה (proofpoint-ser-collector).
  6. לוחצים על הכרטיסייה יומנים.

  7. מוודאים שהפונקציה בוצעה בהצלחה. מחפשים:

    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. נכנסים אל Cloud Storage > Buckets.

  9. לוחצים על שם הקטגוריה (proofpoint-ser-logs).

  10. עוברים לתיקיית הקידומת (ser-logs/).

  11. מוודאים שנוצר קובץ .ndjson חדש עם חותמת הזמן הנוכחית.

אם מופיעות שגיאות ביומנים:

  • ‫HTTP 400 מנקודת הקצה של האסימון: המפתח או הסוד שגויים, או שתוקף המפתח פג. ‫invalid_client – הצמד נדחה, צריך ליצור מפתח חלופי.
  • ‫HTTP 401 מ-Reporting API: טוקן ה-Bearer לא נשלח או שהוא כבר לא תקף.
  • HTTP 429: הגבלת קצב. הפונקציה מנסה שוב עם השהיה לפני ניסיון חוזר (backoff), ואם המכסה נמשכת, היא מפסיקה את ההרצה ומעבירה את היום להרצה הבאה.
  • חסרים משתני סביבה: צריך לוודא שכל המשתנים הנדרשים מוגדרים.

אחזור חשבון השירות של Google SecOps

‫Google SecOps משתמש בחשבון שירות ייחודי כדי לקרוא נתונים מהקטגוריה של Cloud Storage. צריך להעניק לחשבון השירות הזה גישה לדלי.

קבלת כתובת האימייל בחשבון השירות

  1. עוברים אל SIEM Settings > Feeds (הגדרות SIEM > פידים).
  2. לוחצים על הוספת פיד חדש.
  3. לוחצים על הגדרת פיד יחיד.
  4. בשדה שם הפיד, מזינים שם לפיד (לדוגמה, Proofpoint SER Logs).
  5. בוחרים באפשרות Google Cloud Storage V2 בתור סוג המקור.
  6. בוחרים באפשרות ProofPoint Secure Email Relay (ממסר מאובטח של אימייל מ-ProofPoint) בתור סוג היומן.
  7. לוחצים על Get Service Account (קבלת חשבון שירות).

  8. תוצג כתובת האימייל בחשבון שירות ייחודי, לדוגמה:

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

  9. מעתיקים את כתובת האימייל הזו לשימוש בשלב הבא.

  10. לוחצים על הבא.

  11. מציינים ערכים לפרמטרים הבאים של הקלט:

    • כתובת ה-URL של קטגוריית האחסון: מזינים את ה-URI של קטגוריית Cloud Storage עם נתיב הקידומת:

      gs://proofpoint-ser-logs/ser-logs/
      • מחליפים את:
        • ‫proofpoint-ser-logs: שם הקטגוריה שלכם ב-Cloud Storage.
        • ‫ser-logs: קידומת אופציונלית או נתיב תיקייה שבהם מאוחסנים היומנים (משאירים ריק בשביל תיקיית הבסיס).

    • אפשרות למחיקת המקור: בוחרים את אפשרות המחיקה לפי ההעדפה שלכם:

      • אף פעם לא למחוק קבצים: אף פעם לא למחוק קבצים מהמקור (מומלץ לבדיקה).
      • מחיקת קבצים שהועברו וספריות ריקות: מחיקת קבצים וספריות ריקות מהמקור אחרי השלמת אחזור מוצלח.

    • הגיל המקסימלי של הקובץ: כולל קבצים שעברו שינוי במספר הימים האחרון (ברירת המחדל היא 180 ימים)

    • מרחב שמות של נכס: מרחב השמות של הנכס

    • תוויות להוספה: התווית שתתווסף לאירועים מהפיד הזה

  12. לוחצים על הבא.

  13. בודקים את ההגדרות של הפיד החדש במסך סיום ולוחצים על שליחה.

הענקת הרשאות IAM לחשבון השירות של Google SecOps

לחשבון השירות של Google SecOps צריך להיות מוקצות שתי הרשאות בקטגוריה של Cloud Storage: צפייה באובייקט אחסון כדי לקרוא את אובייקטי היומן, והרשאה ברמת הקטגוריה כדי לקרוא את המטא-נתונים של הקטגוריה.

  1. נכנסים אל Cloud Storage > Buckets.
  2. לוחצים על שם הקטגוריה (לדוגמה, proofpoint-ser-logs).
  3. עוברים לכרטיסייה Permissions.
  4. לוחצים על הענקת גישה.
  5. מזינים את פרטי ההגדרה הבאים:
    • Add principals (הוספת ישויות מורשות): מדביקים את כתובת האימייל בחשבון של חשבון השירות של Google SecOps
    • הקצאת תפקידים: בוחרים את שתי האפשרויות הבאות:
      • צפייה באובייקט אחסון: קריאת אובייקטים של יומן.
      • ‫Storage Legacy Bucket Reader: קריאת המטא-נתונים של הקטגוריה. אם בחרתם באפשרות המחיקה מחיקת קבצים שהועברו וספריות ריקות, במקום זאת בוחרים באפשרות Storage Legacy Bucket Writer, שמעניקה גם את הרשאת המחיקה.

  6. לוחצים על Save.

טבלת מיפוי UDM

שדה היומן מיפוי UDM לוגיקה
acceptedMessages_label additional.fields ממוזג
acceptedThroughput_label additional.fields ממוזג
average30DayThroughput_label additional.fields ממוזג
average7DayThroughput_label additional.fields ממוזג
averageDailyThroughput_label additional.fields ממוזג
avgAcceptedMessageSize_label additional.fields ממוזג
blockedMessages_label additional.fields ממוזג
data_totalThroughput_label additional.fields ממוזג
deliveredMessages_label additional.fields ממוזג
licenseEndDate_label additional.fields ממוזג
licenseStartDate_label additional.fields ממוזג
map_label additional.fields ממוזג
metadata_totalThroughput_label additional.fields ממוזג
quarantinedMessages_label additional.fields ממוזג
rejectedMessages_label additional.fields ממוזג
remainingThroughput_label additional.fields ממוזג
requestedMessages_label additional.fields ממוזג
requestedThroughput_label additional.fields ממוזג
sentMessages_label additional.fields ממוזג
throughputForecast_label additional.fields ממוזג
throughputLimit_label additional.fields ממוזג
throughput_label additional.fields ממוזג
totalMessages_label additional.fields ממוזג
undeliveredMessages_label additional.fields ממוזג
data.name metadata.description מיפוי ישיר
desc metadata.description מיפוי ישיר
data.date metadata.event_timestamp ניתוח הכתובת כ-yyyy-MM-dd
event_type metadata.event_type מיפוי ישיר
fromEnvelope network.email.bounce_address מיפוי ישיר
fromHeader network.email.from מיפוי ישיר
applicationName principal.administrative_domain מיפוי ישיר
principal_host principal.asset.hostname מיפוי ישיר
principal_host principal.hostname מיפוי ישיר
principal_port principal.port מיפוי ישיר
data.relayUserId principal.user.product_object_id מיפוי ישיר
userId principal.user.product_object_id מיפוי ישיר
applicationUserName principal.user.user_display_name מיפוי ישיר
senderName target.administrative_domain מיפוי ישיר
senderId target.user.product_object_id מיפוי ישיר
לא רלוונטי metadata.product_name קבוע: PROOFPOINT SER
לא רלוונטי metadata.vendor_name קבוע: PROOFPOINT

שנה רישום

צפייה ביומן השינויים של כלי הניתוח הזה

הבעיה עדיין לא נפתרה? קבלת תשובות מחברי הקהילה וממומחי Google SecOps.