איסוף יומנים של CrowdStrike Falcon Event Streams

גרסת כלי הניתוח: 12.0

נתמך ב:

במאמר הזה מוסבר איך להטמיע יומנים של CrowdStrike Falcon Event Streams ב-Google Security Operations באמצעות Google Cloud Storage V2. ‫CrowdStrike Falcon Event Streams מספק API להזרמת נתונים בזמן אמת, שמעביר נתוני אירועי אבטחה מפלטפורמת Falcon, כולל אירועי זיהוי, אירועי ביקורת, פעילות אימות ועדכונים על תקריות. ‫Event Streams API משתמש בחיבור HTTP מתמשך עם נקודת הקצה /sensors/entities/datafeed/v2 כדי לדחוף אירועים כמעט בזמן אמת, ותומך בהמשכיות מבוססת-היסט כדי להבטיח מסירה מהימנה.

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

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

  • מופע של Google SecOps.
  • פרויקט ב-GCP עם Cloud Storage API מופעל.
  • הרשאות ליצירה ולניהול של קטגוריות GCS.
  • הרשאות לניהול מדיניות IAM בקטגוריות GCS.
  • הרשאות ליצירת שירותי Cloud Run, נושאי Pub/Sub ומשימות Cloud Scheduler.
  • גישה עם הרשאות ל-CrowdStrike Falcon Console עם הרשאות ליצירת לקוחות API.

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

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

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

הגדרת גישה ל-API של CrowdStrike Falcon

יצירת לקוח API

  1. נכנסים למסוף CrowdStrike Falcon.
  2. בתפריט הניווט, עוברים אל Support and resources > Resources and tools > API clients and keys.
  3. לוחצים על Create API client (יצירת לקוח API).
  4. מספקים את פרטי ההגדרה הבאים:
    • שם הלקוח: מזינים שם תיאורי (לדוגמה, Google SecOps Event Streams Integration).
    • תיאור (אופציונלי): מזינים תיאור (לדוגמה, API client for streaming events to GCS).
  5. בקטע API Scopes בוחרים את ההרשאה הבאה:
    • מקורות נתונים של אירועים: בוחרים באפשרות קריאה (נדרש כדי להשתמש ב-Streaming API).
  6. לוחצים על יצירה.

תיעוד פרטי הכניסה ל-API

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

  • ‫Client ID: מחרוזת הקסדצימלית באורך 32 תווים באותיות קטנות.
  • סוד הלקוח: מחרוזת אלפאנומרית באורך 40 תווים.
  • כתובת URL בסיסית: שם הדומיין המוגדר במלואו של CrowdStrike API (לדוגמה, api.crowdstrike.com או api.us-2.crowdstrike.com).

חשוב: צריך להעתיק ולשמור את Client Secret באופן מיידי. הוא מוצג רק פעם אחת ואי אפשר לאחזר אותו בהמשך.

נקודות קצה אזוריות

חברת CrowdStrike פועלת בכמה אזורים עם נקודות קצה שונות של API:

אזור כתובת URL בסיסית
US-1 api.crowdstrike.com
US-2 api.us-2.crowdstrike.com
EU-1 api.eu-1.crowdstrike.com
US-GOV-1 api.laggar.gcw.crowdstrike.com

חשוב: צריך להשתמש בכתובת ה-URL הבסיסית שמתאימה לאזור הכניסה שלכם ל-CrowdStrike Falcon Console.

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

כדי לוודא שללקוח ה-API יש את ההרשאות הנדרשות:

  1. נכנסים למסוף CrowdStrike Falcon.
  2. עוברים אל תמיכה ומשאבים > משאבים וכלים > לקוחות ומפתחות API.
  3. מחפשים את לקוח ה-API שיצרתם ומוודאים שההרשאה Event streams מוגדרת לגישת קריאה.

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

  • לפני שממשיכים בשילוב, חשוב לבדוק את פרטי הכניסה:

    CLIENT_ID="<your-client-id>"
    CLIENT_SECRET="<your-client-secret>"
    BASE_URL="https://api.crowdstrike.com"
    
    # Get OAuth2 token
    TOKEN=$(curl -s -X POST "${BASE_URL}/oauth2/token" \
      -H "Content-Type: application/x-www-form-urlencoded" \
      -d "client_id=${CLIENT_ID}&client_secret=${CLIENT_SECRET}" | python3 -c "import sys,json; print(json.load(sys.stdin)['access_token'])")
    
    # Test Event Streams API access (discover available data feeds)
    curl -s -H "Authorization: Bearer ${TOKEN}" \
      "${BASE_URL}/sensors/entities/datafeed/v2?appId=secops-test&format=flatjson"
    

תגובה מוצלחת מחזירה אובייקט JSON שמכיל את resources עם השדות dataFeedURL ו-sessionToken לכל מחיצת זרם זמינה.

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

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

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

  1. במסוף GCP, עוברים אל IAM & Admin > Service Accounts.
  2. לוחצים על יצירת חשבון שירות.
  3. מספקים את פרטי ההגדרה הבאים:
    • שם חשבון השירות: מזינים cs-stream-collector-sa
    • תיאור חשבון השירות: מזינים Service account for Cloud Run function to collect CrowdStrike Event Streams 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. לוחצים על סיום.

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

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

מתן הרשאות IAM בקטגוריית GCS

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

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

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

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

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

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

הפונקציה של Cloud Run תופעל על ידי הודעות Pub/Sub מ-Cloud Scheduler כדי להתחבר ל-CrowdStrike Event Streams API, לצרוך אירועים ולכתוב אותם ב-GCS.

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

    הגדרה ערך
    שם השירות cs-stream-collector
    אזור בוחרים אזור שתואם ל-GCS Bucket (לדוגמה, us-central1)
    זמן ריצה בוחרים באפשרות Python 3.12 ואילך.
  5. בקטע טריגר (אופציונלי):

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

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

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

    • חשבון שירות: בוחרים באפשרות cs-stream-collector-sa
  9. עוברים לכרטיסייה מאגרי תגים:

    1. לוחצים על Variables & Secrets (משתנים וסודות).
    2. לוחצים על + הוספת משתנה לכל משתנה סביבתי:
    שם המשתנה ערך לדוגמה תיאור
    GCS_BUCKET cs-stream-logs שם קטגוריית GCS
    GCS_PREFIX crowdstrike/stream קידומת לקובצי יומן
    STATE_KEY crowdstrike/stream/state.json נתיב קובץ תמונת מצב
    CS_BASE_URL https://api.crowdstrike.com כתובת ה-URL הבסיסית של CrowdStrike API
    CS_CLIENT_ID your-client-id מזהה הלקוח ב-API
    CS_CLIENT_SECRET your-client-secret סוד לקוח API
    CS_APP_ID secops-stream-collector מזהה אפליקציה ייחודי לזרמי אירועים (עד 32 תווים אלפאנומריים)
    STREAM_TIMEOUT 300 זמן קצוב לתפוגה של קריאת הנתונים מהסטרימינג בשניות
    MAX_RECORDS 10000 מספר הרשומות המקסימלי בכל הפעלה
  10. בקטע Variables & Secrets, גוללים למטה אל Requests:

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

    • בקטע Resources (מקורות מידע):
      • זיכרון: בוחרים באפשרות 512MiB או יותר
      • ‫CPU: בוחרים באפשרות 1
  12. בקטע שינוי קנה מידה של העריכה:

    • מספר מינימלי של מופעים: מזינים 0
    • מספר מופעים מקסימלי: מזינים 1 (רק צרכן אחד של הזרם צריך לפעול בכל פעם)
  13. לוחצים על יצירה.

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

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

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

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

    • main.py:

      import functions_framework
      from google.cloud import storage
      import json
      import os
      import urllib3
      from datetime import datetime, timezone
      
      # 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', 'crowdstrike/stream')
      STATE_KEY = os.environ.get('STATE_KEY', 'crowdstrike/stream/state.json')
      CS_BASE_URL = os.environ.get('CS_BASE_URL', 'https://api.crowdstrike.com')
      CS_CLIENT_ID = os.environ.get('CS_CLIENT_ID')
      CS_CLIENT_SECRET = os.environ.get('CS_CLIENT_SECRET')
      CS_APP_ID = os.environ.get('CS_APP_ID', 'secops-stream-collector')
      STREAM_TIMEOUT = int(os.environ.get('STREAM_TIMEOUT', '300'))
      MAX_RECORDS = int(os.environ.get('MAX_RECORDS', '10000'))
      
      def get_oauth_token(base_url: str, client_id: str, client_secret: str) -> str:
          """Get CrowdStrike OAuth2 access token."""
          token_url = f"{base_url}/oauth2/token"
          body = f"client_id={client_id}&client_secret={client_secret}"
          response = http.request(
              'POST', token_url,
              body=body.encode('utf-8'),
              headers={'Content-Type': 'application/x-www-form-urlencoded'}
          )
          if response.status != 200:
              raise Exception(f"OAuth token request failed: {response.status}")
          data = json.loads(response.data.decode('utf-8'))
          return data['access_token']
      
      def refresh_stream_session(base_url: str, token: str, app_id: str, partition: int = 0):
          """Refresh an active stream session to keep it alive."""
          refresh_url = (
              f"{base_url}/sensors/entities/datafeed-actions/v1/{partition}"
              f"?appId={app_id}&action_name=refresh_active_stream_session"
          )
          response = http.request(
              'POST', refresh_url,
              headers={
                  'Authorization': f'Bearer {token}',
                  'Content-Type': 'application/json',
              }
          )
          if response.status == 200:
              print("Stream session refreshed successfully")
          else:
              print(f"Warning: Stream session refresh returned {response.status}")
      
      @functions_framework.cloud_event
      def main(cloud_event):
          """
          Cloud Run function triggered by Pub/Sub to fetch CrowdStrike
          Event Streams data and write to GCS.
          """
      
          if not all([GCS_BUCKET, CS_BASE_URL, CS_CLIENT_ID, CS_CLIENT_SECRET]):
              print('Error: Missing required environment variables')
              return
      
          try:
              bucket = storage_client.bucket(GCS_BUCKET)
      
              # Load state
              state = load_state(bucket, STATE_KEY)
      
              # Get OAuth token
              token = get_oauth_token(CS_BASE_URL, CS_CLIENT_ID, CS_CLIENT_SECRET)
              headers = {
                  'Authorization': f'Bearer {token}',
                  'Accept': 'application/json',
              }
      
              # Discover available data feeds
              discover_url = (
                  f"{CS_BASE_URL}/sensors/entities/datafeed/v2"
                  f"?appId={CS_APP_ID}&format=flatjson"
              )
              response = http.request('GET', discover_url, headers=headers)
      
              if response.status != 200:
                  print(f"Failed to discover data feed: {response.status}")
                  print(f"Response: {response.data.decode('utf-8')}")
                  return
      
              feed_data = json.loads(response.data.decode('utf-8'))
              resources = feed_data.get('resources', [])
      
              if not resources:
                  print("No data feed resources available")
                  return
      
              feed_url = resources[0].get('dataFeedURL', '')
              session_token = resources[0].get(
                  'sessionToken', {}
              ).get('token', '')
      
              if not feed_url or not session_token:
                  print("Missing feed URL or session token")
                  return
      
              # Build stream URL with offset if available
              stream_url = feed_url
              offset = state.get('offset')
              if offset:
                  stream_url = f"{feed_url}&offset={offset}"
      
              # Connect to Event Stream
              stream_headers = {
                  'Authorization': f'Token {session_token}',
                  'Accept': 'application/json',
                  'Connection': 'keep-alive',
              }
      
              stream_http = urllib3.PoolManager(
                  timeout=urllib3.Timeout(
                      connect=10.0, read=float(STREAM_TIMEOUT)
                  ),
                  retries=False,
              )
      
              print(f"Connecting to Event Stream...")
              resp = stream_http.request(
                  'GET', stream_url,
                  headers=stream_headers,
                  preload_content=False
              )
      
              if resp.status != 200:
                  print(f"Stream connection failed: {resp.status}")
                  return
      
              records = []
              latest_offset = offset
              now = datetime.now(timezone.utc)
      
              for line in resp.stream(4096):
                  decoded = line.decode('utf-8').strip()
                  if not decoded:
                      continue
      
                  try:
                      event = json.loads(decoded)
                      records.append(event)
      
                      event_offset = event.get(
                          'metadata', {}
                      ).get('offset')
                      if event_offset is not None:
                          latest_offset = event_offset
      
                      if len(records) >= MAX_RECORDS:
                          print(
                              f"Reached max records limit ({MAX_RECORDS})"
                          )
                          break
                  except json.JSONDecodeError:
                      continue
      
              resp.release_conn()
      
              if not records:
                  print("No new events found in stream.")
                  # Refresh stream session to keep it alive
                  refresh_stream_session(
                      CS_BASE_URL, token, CS_APP_ID
                  )
                  return
      
              # Write to GCS as NDJSON
              timestamp = now.strftime('%Y%m%d_%H%M%S')
              object_key = f"{GCS_PREFIX}/logs_{timestamp}.ndjson"
              blob = bucket.blob(object_key)
      
              ndjson = '\n'.join(
                  [json.dumps(r, ensure_ascii=False) for r in records]
              ) + '\n'
              blob.upload_from_string(
                  ndjson, content_type='application/x-ndjson'
              )
      
              print(
                  f"Wrote {len(records)} records to "
                  f"gs://{GCS_BUCKET}/{object_key}"
              )
      
              # Save state with latest offset
              new_state = {
                  'offset': latest_offset,
                  'last_run': now.isoformat()
              }
              save_state(bucket, STATE_KEY, new_state)
      
              # Refresh stream session to keep it alive
              refresh_stream_session(
                  CS_BASE_URL, token, CS_APP_ID
              )
      
              print(f"Successfully processed {len(records)} events")
      
          except Exception as e:
              print(f'Error processing logs: {str(e)}')
              raise
      
      def load_state(bucket, key):
          """Load state from GCS."""
          try:
              blob = bucket.blob(key)
              if blob.exists():
                  state_data = blob.download_as_text()
                  return json.loads(state_data)
          except Exception as e:
              print(f"Warning: Could not load state: {e}")
          return {}
      
      def save_state(bucket, key, state: dict):
          """Save state to GCS."""
          try:
              blob = bucket.blob(key)
              blob.upload_from_string(
                  json.dumps(state, indent=2),
                  content_type='application/json'
              )
              print(f"Saved state: {state}")
          except Exception as e:
              print(f"Warning: Could not save state: {e}")
      

    requirements.txt:

    ```none
    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. במסוף GCP, עוברים אל Cloud Scheduler.
  2. לוחצים על Create Job (יצירת משימה).
  3. מספקים את פרטי ההגדרה הבאים:

    הגדרה ערך
    שם cs-stream-collector-scheduled
    אזור בחירה באותו אזור כמו פונקציית Cloud Run
    תדירות */15 * * * * (כל 15 דקות)
    אזור זמן בחירת אזור זמן (מומלץ לבחור UTC)
    סוג הטירגוט Pub/Sub
    נושא בחירת cs-stream-trigger
    גוף ההודעה ‫{} (אובייקט JSON ריק)
  4. לוחצים על יצירה.

בדיקת השילוב

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

    Connecting to Event Stream...
    Wrote X records to gs://cs-stream-logs/crowdstrike/stream/logs_YYYYMMDD_HHMMSS.ndjson
    Stream session refreshed successfully
    Successfully processed X events
    
  8. עוברים אל Cloud Storage > Buckets (קטגוריות).

  9. לוחצים על cs-stream-logs.

  10. עוברים לתיקייה crowdstrike/stream/.

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

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

  • ‫HTTP 401: בדיקת פרטי הכניסה ל-API במשתני הסביבה
  • ‫HTTP 403: מוודאים שללקוח ה-API יש הרשאת Scope‏ Event streams: Read
  • ‫HTTP 429: הגבלת קצב – צריך לשנות את התדירות של הכלי לתזמון או להגדיל את STREAM_TIMEOUT
  • חסרים משתני סביבה: צריך לוודא שכל המשתנים הנדרשים מוגדרים
  • אין משאבים של פיד נתונים זמינים: מוודאים שהערך של CS_APP_ID הוא ייחודי ולא חורג מ-32 תווים אלפאנומריים

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

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

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

  1. עוברים אל SIEM Settings > Feeds (הגדרות SIEM > פידים).
  2. לוחצים על הוספת פיד חדש.
  3. לוחצים על Configure a single feed (הגדרת פיד יחיד).
  4. בשדה שם הפיד, מזינים שם לפיד (לדוגמה, CrowdStrike Event Streams logs).
  5. בוחרים באפשרות Google Cloud Storage V2 בתור סוג המקור.
  6. בוחרים באפשרות CrowdStrike Falcon Streaming בתור סוג היומן.
  7. לוחצים על קבלת חשבון שירות.
  8. תוצג כתובת האימייל בחשבון שירות ייחודי, לדוגמה:

    chronicle-12345678@chronicle-gcp-prod.iam.gserviceaccount.com
    
  9. מעתיקים את כתובת האימייל הזו לשימוש בשלב הבא.

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

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

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

      gs://cs-stream-logs/crowdstrike/stream/
      
    • אפשרות למחיקת המקור: בוחרים את אפשרות המחיקה לפי ההעדפה שלכם:

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

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

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

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

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

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

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

לחשבון השירות של Google SecOps צריך להיות התפקיד צפייה באובייקט אחסון בקטגוריה של GCS.

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

דרושה לך עזרה נוספת?

מידע נוסף על פידים של Google Security Operations זמין במסמכי התיעוד של פידים של Google Security Operations. מידע על הדרישות לכל סוג פיד זמין במאמר הגדרת פיד לפי סוג.

לעיון, אפשר לקרוא את מסמכי התיעוד בנושא מנתח הנתונים של CrowdStrike Falcon Event Streams.

הפניה לדלתא של מיפוי UDM

ב-15 בספטמבר 2026, צוות Google SecOps פרסם גרסה חדשה של מנתח הנתונים של CrowdStrike Falcon Event Streams, שכוללת שינויים משמעותיים במיפוי של שדות ביומן של CrowdStrike Falcon Event Streams לשדות ב-UDM.

דלתא של מיפוי שדות ביומן

בטבלה הבאה מפורטות ההתאמות בין שדות ביומן של CrowdStrike Falcon Event Streams לשדות ב-UDM, כפי שהם הוצגו לפני 15 בספטמבר 2026 ואחריו (מפורטות בעמודות מיפוי ישן ומיפוי חדש, בהתאמה):

שדה ביומן מיפוי ישן מיפוי חדש הסיבה לשינוי
deviceId additional.fields principal.asset.asset_id ממופה לשדה UDM ספציפי יותר במקום לתוויות.
event.AuditKeyValues (report_name) security_result.detection_fields target.file.names ממופה לשדה UDM ספציפי יותר במקום לתוויות.
event.IdpPolicyRuleName security_result.detection_fields security_result.rule_name ממופה לשדה UDM ספציפי יותר במקום לתוויות.
event.MitreAttack.TacticID security_result.detection_fields security_result.tactic.id ממופה לשדה UDM ספציפי יותר במקום לתוויות.
event.MitreAttack.TechniqueID security_result.detection_fields security_result.technique.id ממופה לשדה UDM ספציפי יותר במקום לתוויות.
event.NetworkAccesse.RemoteAddress principal.ip target.ip העדכון בוצע כי event.NetworkAccesse.RemoteAddress מייצג את כתובת מכשיר היעד.
event.TacticId security_result.detection_fields security_result.tactic.id ממופה לשדה UDM ספציפי יותר במקום לתוויות.
event.TechniqueId security_result.detection_fields security_result.technique.id ממופה לשדה UDM ספציפי יותר במקום לתוויות.
event.UserId principal.user.userid principal.user.email_addresses,
principal.user.userid
אם event.UserId היא כתובת אימייל, היא ממופה ל-principal.user.email_addresses. אחרת, היא ממופה ל-principal.user.userid.
remoteAddress principal.ip target.ip העדכון בוצע כי remoteAddress מייצג את כתובת מכשיר היעד.
ruleName security_result.detection_fields security_result.rule_name ממופה לשדה UDM ספציפי יותר במקום לתוויות.
serviceName metadata.description,
target.application
target.application הסרנו מיפוי מיותר.

יומן שינויים

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

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