איסוף יומנים של BeyondTrust Endpoint Privilege Management (EPM)

נתמך ב:

במאמר הזה מוסבר איך להטמיע יומנים של BeyondTrust Endpoint Privilege Management ‏ (EPM) ב-Google Security Operations באמצעות Cloud Storage. הכלי לניתוח מתמקד בהמרת נתוני יומן גולמיים בפורמט JSON מ-BeyondTrust Endpoint לפורמט מובנה שתואם ל-UDM של Google SecOps. הוא מאתחל קודם ערכי ברירת מחדל לשדות שונים, ואז מנתח את מטען הנתונים (payload) בפורמט JSON, ולאחר מכן ממפה שדות ספציפיים מהיומן הגולמי לשדות UDM תואמים באובייקט event.idm.read_only_udm.

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

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

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

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

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

  5. מספקים את פרטי ההגדרה הבאים:

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

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

איסוף פרטי הכניסה ל-API של BeyondTrust EPM

  1. נכנסים למסוף האינטרנט של BeyondTrust Privilege Management כאדמינים.
  2. עוברים אל הגדרות > הגדרות > הגדרות API.
  3. לוחצים על Create an API Account (יצירת חשבון API).
  4. מספקים את פרטי ההגדרה הבאים:
    • שם: מזינים Google SecOps Collector.
    • גישה ל-API: מפעילים את האפשרות דיווח (קריאה בלבד). /management-api/v3/Events/FromStartDate נקודת הקצה שבה נעשה שימוש על ידי כלי האיסוף הזה שייכת לקטגוריית ההרשאות דיווח. מפעילים גם את ביקורת (קריאה בלבד) רק אם רוצים לאסוף גם נתוני ביקורת על פעילות במסוף האינטרנט.
  5. מעתיקים את מזהה הלקוח ואת הסוד של הלקוח ושומרים אותם.

  6. מעתיקים את כתובת הבסיס של ה-API שמופיעה בחלק העליון של הדף API Settings (הגדרות API). בדרך כלל זה הנתיב https://<your-tenant>-services.pm.beyondtrustcloud.com. השימוש שלך יהיה כ-BPT_API_URL.

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

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

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

  1. במסוף Google Cloud , נכנסים אל IAM & Admin > Service Accounts.
  2. לוחצים על יצירת חשבון שירות.
  3. מספקים את פרטי ההגדרה הבאים:
    • שם חשבון השירות: מזינים beyondtrust-epm-collector-sa.
    • תיאור חשבון השירות: מזינים Service account for Cloud Run function to collect BeyondTrust EPM 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. לוחצים על שם הקטגוריה.
  3. עוברים לכרטיסייה Permissions.
  4. לוחצים על הענקת גישה.
  5. מספקים את פרטי ההגדרה הבאים:
    • Add principals (הוספת ישויות מורשות): מזינים את כתובת האימייל בחשבון השירות. לדוגמה, beyondtrust-epm-collector-sa@PROJECT_ID.iam.gserviceaccount.com.
    • הקצאת תפקידים: בוחרים באפשרות אדמין של אובייקטים באחסון.
  6. לוחצים על Save.

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

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

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

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

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

  1. במסוף Google Cloud , עוברים אל Cloud Run.
  2. לוחצים על יצירת שירות.
  3. בוחרים באפשרות פונקציה. משתמשים בעורך הישיר כדי ליצור פונקציה.

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

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

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

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

  6. בקטע Authentication:

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

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

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

    • חשבון שירות: בוחרים את חשבון השירות beyondtrust-epm-collector-sa.

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

    1. לוחצים על Variables & Secrets (משתנים וסודות).
    2. לוחצים על + הוספת משתנה לכל משתנה סביבתי:
    שם המשתנה ערך לדוגמה
    GCS_BUCKET beyondtrust-epm-logs
    GCS_PREFIX beyondtrust-epm/
    STATE_KEY beyondtrust-epm-state.json
    BPT_API_URL https://yourtenant-services.pm.beyondtrustcloud.com
    CLIENT_ID your-client-id
    CLIENT_SECRET your-client-secret
    RECORD_SIZE 1000
    MAX_BATCHES 50
    LOOKBACK_HOURS 24
    • ‫RECORD_SIZE: רשומות לכל בקשה. ‫/Events/FromStartDate מקבל ערכים מ-1 עד 1,000. הפונקציה מגבילה ערכים גבוהים יותר.
    • ‫MAX_BATCHES: מגבלת בטיחות על בקשות בכל הרצה. אם הפעלה מגיעה לשם, הסמן עדיין מתקדם למה שנכתב, כך שההפעלה הבאה ממשיכה משם ולא חוזרת על העבודה.
    • ‫LOOKBACK_HOURS: כמה זמן אחורה מגיעה ההרצה הראשונה. ‫BeyondTrust שומרת אירועים פסיביים (קודים 106,‏ 107,‏ 603,‏ 706) למשך 30 ימים, ואת כל שאר האירועים למשך 90 ימים, כך שחיפוש עם חלון מבט לאחור ארוך יותר לא יחזיר תוצאות ישנות יותר.
    • ‫LOOKBACK_HOURS: כמה זמן אחורה צריך להגיע בהרצה הראשונה, לפני שנוצר סמן.

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

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

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

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

  12. גוללים אל סביבת הפעלה:

    • בוחרים באפשרות ברירת מחדל (מומלץ).

  13. בקטע שינוי קנה מידה של העריכה:

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

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

  15. ממתינים ליצירת השירות (1-2 דקות).

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

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

  1. מזינים main בFunction entry point

  2. בעורך הקוד המוטבע, יוצרים שני קבצים:

    • הקובץ הראשון – main.py:
    import hashlib
import json
import os
import re
import time
import urllib3
import uuid
from datetime import datetime, timedelta, timezone
from urllib.parse import urlencode

import functions_framework
from google.cloud import storage
from google.cloud.exceptions import NotFound

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

# Initialize Storage client
storage_client = storage.Client()

TOKEN_PATH = '/oauth/connect/token'
# Events/FromStartDate is the endpoint BeyondTrust documents for bulk extraction:
# "A new API is exposed to extract the events in bulk." It takes only StartDate and
# RecordSize. Events/search is not usable here: it requires an OperatingSystem value,
# it is a single string with no documented "all" value, so it can only ever return one
# operating system per call.
EVENTS_PATH = '/management-api/v3/Events/FromStartDate'
OAUTH_SCOPE = 'urn:management:api'
# RecordSize accepts 1 to 1000.
MAX_RECORD_SIZE = 1000

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

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

def rfc3339(dt: datetime) -> str:
    """Render a datetime the way StartDate accepts it: milliseconds and a literal Z."""
    return dt.astimezone(timezone.utc).strftime('%Y-%m-%dT%H:%M:%S.%f')[:-3] + 'Z'

def parse_iso(stamp: str) -> datetime:
    """Parse any ISO 8601 timestamp shape the API emits.

    event.ingested comes back with a +00:00 offset and up to seven fractional
    digits, e.g. 2026-08-03T13:29:55.1109163+00:00. fromisoformat's tolerance for
    long fractions varies across Python versions, so the fraction is trimmed to
    microseconds first.
    """
    text = stamp.strip()
    if text.endswith('Z'):
        text = text[:-1] + '+00:00'
    text = re.sub(r'\.(\d{1,6})\d*', r'.\1', text, count=1)
    parsed = datetime.fromisoformat(text)
    if parsed.tzinfo is None:
        parsed = parsed.replace(tzinfo=timezone.utc)
    return parsed

def canonical(stamp: str) -> str:
    """Re-render an API timestamp into the one form StartDate accepts.

    The API returns event.ingested as +00:00-offset ISO 8601 but rejects that
    same shape as a StartDate value with 400 "Invalid Start date format": it
    accepts only Z-suffixed values. Every timestamp that came from the API must
    pass through here before being sent back or persisted.
    """
    return rfc3339(parse_iso(stamp))

def event_id(evt: dict) -> str:
    """Return the event's identity for deduplication.

    Falls back to a content hash when event.id is absent, so an id-less event
    still deduplicates instead of being re-ingested on every boundary re-read.
    """
    explicit = str((evt.get('event') or {}).get('id') or '')
    if explicit:
        return explicit
    digest = hashlib.sha256(
        json.dumps(evt, sort_keys=True, ensure_ascii=False).encode('utf-8')
    ).hexdigest()
    return f'sha256:{digest}'

def event_ingested(evt: dict) -> str:
    """Return the Elastic ingestion timestamp, which is what StartDate filters on."""
    return str((evt.get('event') or {}).get('ingested') or '')

def next_millisecond(stamp: str) -> str:
    """Return the canonical timestamp one millisecond later.

    Used only when a full batch fits inside a single ingestion millisecond. Without
    this the cursor cannot move and the collector stalls on that timestamp forever.
    """
    return rfc3339(parse_iso(stamp) + timedelta(milliseconds=1))

@functions_framework.cloud_event
def main(cloud_event):
    """Fetch BeyondTrust EPM events and write them to Cloud Storage as NDJSON.

    Args:
            cloud_event: CloudEvent object containing the Pub/Sub message.
    """
    bucket_name = os.environ.get('GCS_BUCKET')
    prefix = os.environ.get('GCS_PREFIX', 'beyondtrust-epm/')
    state_key = os.environ.get('STATE_KEY', 'beyondtrust-epm-state.json')

    api_url = (os.environ.get('BPT_API_URL') or '').rstrip('/')
    client_id = os.environ.get('CLIENT_ID')
    client_secret = os.environ.get('CLIENT_SECRET')
    # Clamp both ends: the API rejects RecordSize outside 1 to 1000, and a
    # misconfigured 0 would otherwise crash-loop on a 400 every run.
    record_size = min(max(int(os.environ.get('RECORD_SIZE', '1000')), 1), MAX_RECORD_SIZE)
    max_batches = int(os.environ.get('MAX_BATCHES', '50'))
    lookback_hours = int(os.environ.get('LOOKBACK_HOURS', '24'))

    if not all([bucket_name, api_url, client_id, client_secret]):
        raise RuntimeError(
            'Missing required environment variables: '
            'GCS_BUCKET, BPT_API_URL, CLIENT_ID, CLIENT_SECRET'
        )

    bucket = storage_client.bucket(bucket_name)
    state = load_state(bucket, state_key)

    # StartDate is a rising cursor on the Elastic ingestion timestamp, not a closed
    # window. There is no EndDate on this endpoint.

    if state.get('last_ingested'):
        # canonical() also repairs state written by the previous script revision,
        # which persisted the API's raw +00:00 form that StartDate rejects.
        try:
            start_date = canonical(state['last_ingested'])
        except (ValueError, TypeError, AttributeError) as e:
            raise RuntimeError(
                f'Unparseable last_ingested in gs://{bucket_name}/{state_key}: '
                f'{state["last_ingested"]!r}. Fix or delete that object; deleting '
                f'restarts collection from the lookback window.'
            ) from e
    else:
        start_date = rfc3339(datetime.now(timezone.utc) - timedelta(hours=lookback_hours))

    seen_ids = set(state.get('seen_ids', []))
    print(f'Collecting events ingested from {start_date}')

    token = get_oauth_token(api_url, client_id, client_secret)
    fresh, cursor, drained = fetch_events(
        api_url, token, start_date, record_size, max_batches, seen_ids
    )

    if not fresh:
        print('No new events. Cursor left unchanged.')
        return

    # Everything that can fail is computed before the upload: a crash between the
    # upload and save_state replays the batch on the next run, so the window where
    # side effects exist without recorded state must stay minimal. The retained ids
    # share the cursor millisecond, because those are exactly the ones the
    # inclusive StartDate will return again.
    retained = {
        event_id(e) for e in fresh
        if event_ingested(e) and canonical(event_ingested(e)) == cursor
    }
    if cursor == start_date:
        # The cursor millisecond did not advance, so ids retained by earlier runs
        # are still on the boundary; dropping them would re-ingest their events.
        retained |= seen_ids

    timestamp = datetime.now(timezone.utc).strftime('%Y%m%dT%H%M%SZ')
    # The random suffix keeps concurrent executions (Pub/Sub delivers at least
    # once) from overwriting each other's object within the same second.
    filename = (
        f'{prefix}beyondtrust-epm-events-{timestamp}-'
        f'{uuid.uuid4().hex[:8]}.ndjson'
    )
    ndjson = '\n'.join(json.dumps(e, ensure_ascii=False) for e in fresh) + '\n'
    bucket.blob(filename).upload_from_string(
        ndjson, content_type='application/x-ndjson'
    )
    print(f'Wrote {len(fresh)} events to gs://{bucket_name}/{filename}')

    save_state(bucket, state_key, {
        'last_ingested': cursor,
        'seen_ids': sorted(retained),
        'updated_at': rfc3339(datetime.now(timezone.utc)),
    })

    if not drained:
        print(
            f'Stopped after {max_batches} batches with more events available. '
            'The cursor advanced, so the next run continues from here.'
        )

def fetch_events(api_url, token, start_date, record_size, max_batches, seen_ids):
    """Read forward from start_date until the API returns a short batch.

    The response envelope has no page count and no next-page token, so a batch
    shorter than record_size is the only documented end-of-data signal.

    StartDate is inclusive and the cursor lands on the newest event's timestamp, so
    every batch re-returns the events sharing it. Deduplication by event id therefore
    has to happen as batches arrive, not only between runs.

    Returns:
    Tuple of (new events, cursor to persist, whether the stream was drained).

    Raises:
    FetchError: on any API or transport failure, so the caller cannot mistake a
    failed fetch for an empty result and advance the cursor.
    """
    headers = {
        'Authorization': f'Bearer {token}',
        'Accept': 'application/json',
    }

    all_events = []
    seen = set(seen_ids)
    cursor = start_date

    for batch in range(1, max_batches + 1):
        query = urlencode({'StartDate': cursor, 'RecordSize': record_size})
        body = request_with_retry(f'{api_url}{EVENTS_PATH}?{query}', headers)

        events = body.get('events') or []
        new = [e for e in events if event_id(e) not in seen]
        seen.update(event_id(e) for e in events if event_id(e))
        all_events.extend(new)
        print(f'Batch {batch}: {len(events)} events, {len(new)} new')

        # Termination is judged on what the API returned, not on what survived
        # deduplication: a batch can be full and still be entirely duplicates.
        # The newest stamp in the batch, not events[-1]: ordering is a client-side
        # convention, not a documented guarantee, and a misordered tail would
        # regress the cursor and re-ingest events already written.
        stamps = [canonical(s) for s in (event_ingested(e) for e in events) if s]

        if len(events) < record_size:
            # Short batch: the stream is drained.
            if events and not stamps:
                print(
                    'Warning: no event in the final batch carries event.ingested; '
                    'cursor left unchanged.'
                )
            if stamps:
                cursor = max(max(stamps), cursor)
            return all_events, cursor, True

        if not stamps:
            raise FetchError(
                'No event in a full batch carries event.ingested, '
                'so the cursor cannot advance'
            )
        batch_max = max(stamps)
        if batch_max <= cursor:
            # A full batch fits inside one ingestion millisecond. Stepping past it is the only
            # way to make progress; holding the cursor here stalls collection forever.
            # Events beyond record_size at that exact millisecond are unreachable, which
            # needs more than 1000 events in one millisecond.
            print(
                f'Warning: a full batch shares ingestion timestamp {batch_max}; '
                'stepping past it. Events beyond RecordSize at that timestamp are skipped.'
            )
            cursor = next_millisecond(cursor)
        else:
            cursor = batch_max

    return all_events, cursor, False

def request_with_retry(url, headers, attempts=4):
    """GET with backoff on 429 and 5xx.

    The documented rate limit is 1000 requests per 100 seconds.
    """
    backoff = 1.0
    for attempt in range(1, attempts + 1):
        try:
            response = http.request('GET', url, headers=headers)
        except Exception as e:
            raise FetchError(f'Request to {url} failed: {e}') from e

        if response.status in (429, 500, 502, 503, 504) and attempt < attempts:
            retry_after = response.headers.get('Retry-After')
            try:
                delay = int(retry_after) if retry_after else backoff
            except (TypeError, ValueError):
                delay = backoff
            # Retry-After is server-controlled input: bound it so a bogus value
            # cannot sleep past the function timeout or crash time.sleep.
            delay = min(max(delay, 1.0), 60.0)
            print(f'HTTP {response.status}. Retrying in {delay}s...')
            time.sleep(delay)
            backoff = min(backoff * 2, 30.0)
            continue

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

        text = response.data.decode('utf-8')
        # BeyondTrust's sample folds "Owner" into "owner" before parsing, but that
        # workaround exists for PowerShell's case-insensitive ConvertFrom-Json.
        # Python parses case-sensitively, and the SecOps parser maps file.Owner.*
        # and file.owner to different UDM fields, so both keys must survive.
        try:
            return json.loads(text)
        except json.JSONDecodeError as e:
            raise FetchError(f'Malformed JSON response: {e}') from e

    raise FetchError(f'Giving up on {url} after {attempts} attempts')

def get_oauth_token(api_url, client_id, client_secret):
    """Get an access token using the OAuth client credentials flow.

    The token is valid for one hour.
    """
    body = urlencode({
        'grant_type': 'client_credentials',
        'client_id': client_id,
        'client_secret': client_secret,
        'scope': OAUTH_SCOPE,
    })
    response = http.request(
        'POST',
        f'{api_url}{TOKEN_PATH}',
        body=body,
        headers={'Content-Type': 'application/x-www-form-urlencoded'},
    )
    if response.status != 200:
        raise FetchError(
            f'Token request failed: {response.status} '
            f'{response.data.decode("utf-8")}'
        )
    return json.loads(response.data.decode('utf-8'))['access_token']

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

    Only a missing object is a cold start. Any other error is raised: swallowing it
    would reset collection to the lookback window and re-ingest that period.
    """
    blob = bucket.blob(key)
    try:
        return json.loads(blob.download_as_text())
    except NotFound:
        print('No state file found. Starting from the lookback window.')
        return {}

def save_state(bucket, key, state):
    """Write the collector state to Cloud Storage.

    Failures are raised, not logged. A run that cannot record its cursor must fail,
    otherwise the next run repeats the same window.
    """
    bucket.blob(key).upload_from_string(
        json.dumps(state, indent=2), content_type='application/json'
    )
    print(f"Saved state: last_ingested={state.get('last_ingested')}")

    • הקובץ השני: requirements.txt:
    functions-framework==3.*
google-cloud-storage==2.*
urllib3>=2.0.0

  3. לוחצים על Deploy כדי לשמור ולפרוס את הפונקציה.

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

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

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

  1. נכנסים אל Cloud Scheduler במסוף Google Cloud .
  2. לוחצים על Create Job (יצירת משימה).

  3. מספקים את פרטי ההגדרה הבאים:

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

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

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

בחירת התדירות בהתאם לנפח היומן ולדרישות זמן האחזור:

תדירות ביטוי Cron תרחיש לדוגמה
כל 5 דקות */5 * * * * נפח גבוה, זמן אחזור נמוך
כל 15 דקות */15 * * * * נפח חיפושים בינוני
כל שעה 0 * * * * רגיל (מומלץ)
כל 6 שעות 0 */6 * * * נפח נמוך, עיבוד באצווה
יומי 0 0 * * * איסוף נתונים היסטוריים

בדיקת משימת המתזמן

  1. במסוף Cloud Scheduler, מוצאים את העבודה.
  2. כדי להפעיל את הטריגר באופן ידני, לוחצים על הפעלה מאולצת.
  3. מחכים כמה שניות ועוברים אל Cloud Run > Services > beyondtrust-epm-collector > Logs.
  4. מוודאים שהפונקציה בוצעה בהצלחה.
  5. בודקים את קטגוריה של Cloud Storage כדי לוודא שהיומנים נכתבו.

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

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

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

  1. עוברים אל SIEM Settings > Feeds (הגדרות SIEM > פידים).
  2. לוחצים על הוספת פיד חדש.
  3. לוחצים על Configure a single feed (הגדרת פיד יחיד).
  4. בשדה שם הפיד, מזינים שם לפיד (לדוגמה, BeyondTrust EPM logs).
  5. בוחרים באפשרות Google Cloud Storage V2 בתור סוג המקור.
  6. בוחרים באפשרות BeyondTrust Endpoint Privilege Management בתור סוג היומן.

  7. לוחצים על קבלת חשבון שירות. תוצג כתובת האימייל בחשבון שירות ייחודי, לדוגמה:

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

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

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

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

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

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

הגדרה של פיד ב-Google SecOps להטמעה של יומנים מ-BeyondTrust EPM

  1. עוברים אל SIEM Settings > Feeds (הגדרות SIEM > פידים).
  2. לוחצים על הוספת פיד חדש.
  3. לוחצים על Configure a single feed (הגדרת פיד יחיד).
  4. בשדה שם הפיד, מזינים שם לפיד (לדוגמה, BeyondTrust EPM logs).
  5. בוחרים באפשרות Google Cloud Storage V2 בתור סוג המקור.
  6. בוחרים באפשרות BeyondTrust Endpoint Privilege Management בתור סוג היומן.
  7. לוחצים על הבא.

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

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

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

      • החלפה:

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

      • לדוגמה:

        • קטגוריית הבסיס: gs://beyondtrust-epm-logs/
        • עם קידומת: gs://beyondtrust-epm-logs/beyondtrust-epm/

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

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

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

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

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

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

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

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

טבלת מיפוי UDM

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

יומן שינויים

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

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