איסוף יומנים של 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.
  4. לוחצים על Create Key.
  5. בוחרים באפשרות Secure Email Relay בתור המוצר.

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

    • מפתח: משמש בתור client_id
    • סוד: משמש בתור client_secret

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

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

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

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

  • ‫Reporting API משתמש בהענקת פרטי כניסה של לקוח OAuth 2.0. מחליפים את המפתח והסוד בטוקן למוכ"ז, ואז קוראים ל-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. לוחצים על Create Service Account.
  3. מזינים את פרטי ההגדרה הבאים:
    • Service account name: מזינים proofpoint-ser-collector-sa
    • Service account description: מזינים 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 ובוחרים בו.
    5. לוחצים על + הוספת תפקיד.
    6. מחפשים את התפקיד Cloud Functions Invoker ובוחרים אותו.
  6. לוחצים על Continue.
  7. לוחצים על סיום.

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

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

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

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

  1. נכנסים לדף Cloud Storage > Buckets.
  2. לוחצים על שם הקטגוריה (לדוגמה, proofpoint-ser-logs).
  3. עוברים לכרטיסייה Permissions.
  4. לוחצים על הענקת גישה.
  5. מזינים את פרטי ההגדרה הבאים:
    • הוספת חשבונות משתמשים: מזינים את כתובת האימייל של חשבון השירות (לדוגמה, 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. לוחצים על Create topic.
  3. מזינים את פרטי ההגדרה הבאים:
    • מזהה הנושא: מזינים proofpoint-ser-trigger
    • משאירים את שאר ההגדרות כברירת מחדל.
  4. לוחצים על יצירה.

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

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

  1. במסוף GCP, נכנסים ל-Cloud Run.
  2. לוחצים על Create service.
  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. עוברים לכרטיסייה אבטחה:

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

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

    1. לוחצים על Variables & Secrets.
    2. לוחצים על ‎+ Add variable לכל משתנה סביבה:
    שם משתנה ערך לדוגמה תיאור
    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 נקודת קצה של טוקן OAuth 2.0
    API_BASE https://reporting.ser.proofpoint.com כתובת ה-URL הבסיסית של SER Reporting API
    CLIENT_ID your-key המפתח מניהול מפתחות ה-API
    CLIENT_SECRET your-secret הסוד מניהול מפתחות ה-API
    PAGE_SIZE 50 מספר המשתמשים ב-Relay שביקשת לכל דף
    MAX_PAGES 200 מספר הדפים המקסימלי לעימוד של משתמשי ממסר
    REPORT_DAY_OFFSET 1 דיווח על היום הזה לפני מספר הימים הזה
    MAX_BACKFILL_DAYS 7 ברוב הימים, הרצה אחת תמלא את הנתונים החסרים
    SEEN_RETENTION_DAYS 7 מספר הימים שבהם מפתחות לביטול כפילויות נשמרים במצב

  10. בקטע Variables & Secrets, עוברים אל Requests:

    • ‫Request timeout: מזינים 600 שניות (10 דקות)

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

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

  12. בקטע Revision scaling:

    • Minimum number of instances: מזינים 0
    • Maximum number of instances: מזינים 100 (או משנים בהתאם לעומס הצפוי)

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

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

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

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

  1. בשדה Entry point, מזינים 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. לוחצים על Create Job.

  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. לוחצים על הכרטיסייה Logs.

  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: הטוקן למוכ"ז לא נשלח או שהוא כבר לא תקף.
  • ‫HTTP 429: הגבלת קצב של יצירת בקשות. הפונקציה מנסה שוב עם השהיה, ואם המגבלה נמשכת, ההרצה נכשלת והיום נשאר להרצה הבאה.
  • משתני סביבה חסרים: ודאו שכל המשתנים הנדרשים מוגדרים.

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

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

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

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

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

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

    • גיל הקובץ המקסימלי: כולל קבצים ששונו במספר הימים האחרונים (ברירת המחדל היא 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.
    • הקצאת תפקידים: בוחרים את שני התפקידים הבאים:
      • צפייה באובייקטים של אחסון: קריאת אובייקטים של יומנים.
      • קריאה בקטגוריה באחסון מדור קודם: קריאת המטא-נתונים של הקטגוריה. אם בחרתם באפשרות המחיקה Delete transferred files and empty directories, בחרו במקום זאת באפשרות Storage Legacy Bucket Writer, שגם מעניקה את הרשאת המחיקה.

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

טבלת מיפוי UDM

שדה ביומן מיפוי UDM Logic (פונקציה לוגית)
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.