איסוף יומנים של OpenTelemetry Netflow Receiver

נתמך ב:

במאמר הזה מוסבר איך להטמיע יומנים של OpenTelemetry Netflow Receiver ב-Google Security Operations באמצעות Google Cloud Storage V2.

רכיב OpenTelemetry Netflow Receiver הוא רכיב בקוד פתוח שממתין לתנועת נתונים של netflow,‏ sflow ו-IPFIX UDP וממיר אותה לרשומות יומן של OpenTelemetry. ההרשאה הזו מאפשרת ניטור וניתוח של תעבורת נתונים ברשת, כולל זיהוי פרוטוקולים, ניתוח נפח התעבורה, מעקב אחר השימוש ביציאות וסטטיסטיקות של בייטים וחבילות.

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

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

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

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

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

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

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

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

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

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

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

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

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

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

פונקציית Cloud Run תפעיל OpenTelemetry Collector שמקבל נתוני netflow ומייצא אותם ל-GCS.

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

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

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

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

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

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

    1. לוחצים על Variables & Secrets (משתנים וסודות).
    2. לוחצים על + הוספת משתנה לכל משתנה סביבתי:
    שם המשתנה ערך לדוגמה תיאור
    GCS_BUCKET netflow-logs-bucket שם קטגוריית GCS
    GCS_PREFIX netflow קידומת לקובצי יומן
    NETFLOW_PORT 2055 יציאה למקלט NetFlow
    NETFLOW_SCHEME netflow סוג הסכימה: netflow,‏ sflow או ipfix
    NETFLOW_SOCKETS 4 מספר שקעי UDP
    NETFLOW_WORKERS 8 מספר העובדים של פענוח
  10. בקטע Variables & Secrets, גוללים למטה אל Requests:

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

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

    • מספר מינימלי של מופעים: מזינים 1
    • מספר מקרים מקסימלי: מזינים 10
  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 subprocess
    import tempfile
    import signal
    import time
    from datetime import datetime, timezone
    
    # Initialize Storage client
    storage_client = storage.Client()
    
    # Environment variables
    GCS_BUCKET = os.environ.get('GCS_BUCKET')
    GCS_PREFIX = os.environ.get('GCS_PREFIX', 'netflow')
    NETFLOW_PORT = os.environ.get('NETFLOW_PORT', '2055')
    NETFLOW_SCHEME = os.environ.get('NETFLOW_SCHEME', 'netflow')
    NETFLOW_SOCKETS = os.environ.get('NETFLOW_SOCKETS', '4')
    NETFLOW_WORKERS = os.environ.get('NETFLOW_WORKERS', '8')
    
    # Global process handle
    collector_process = None
    
    def create_collector_config():
      """Create OpenTelemetry Collector configuration."""
      timestamp = datetime.now(timezone.utc).strftime('%Y%m%d_%H%M%S')
      file_path = f"/tmp/netflow_{timestamp}.ndjson"
    
      config = {
        'receivers': {
          'netflow': {
            'scheme': NETFLOW_SCHEME,
            'hostname': '0.0.0.0',
            'port': int(NETFLOW_PORT),
            'sockets': int(NETFLOW_SOCKETS),
            'workers': int(NETFLOW_WORKERS),
            'queue_size': 5000
          }
        },
        'processors': {
          'batch': {
            'timeout': '10s',
            'send_batch_size': 1000
          }
        },
        'exporters': {
          'file': {
            'path': file_path,
            'format': 'json'
          }
        },
        'service': {
          'pipelines': {
            'logs': {
              'receivers': ['netflow'],
              'processors': ['batch'],
              'exporters': ['file']
            }
          },
          'telemetry': {
            'logs': {
              'level': 'info'
            }
          }
        }
      }
    
      config_path = '/tmp/otel-config.yaml'
      with open(config_path, 'w') as f:
        import yaml
        yaml.dump(config, f)
    
      return config_path, file_path
    
    def upload_to_gcs(file_path):
      """Upload netflow logs to GCS."""
      if not os.path.exists(file_path) or os.path.getsize(file_path) == 0:
        print(f"No data to upload from {file_path}")
        return
    
      bucket = storage_client.bucket(GCS_BUCKET)
      timestamp = datetime.now(timezone.utc).strftime('%Y%m%d_%H%M%S')
      object_key = f"{GCS_PREFIX}/netflow_{timestamp}.ndjson"
      blob = bucket.blob(object_key)
    
      blob.upload_from_filename(file_path, content_type='application/x-ndjson')
      print(f"Uploaded {os.path.getsize(file_path)} bytes to gs://{GCS_BUCKET}/{object_key}")
    
    def signal_handler(signum, frame):
      """Handle shutdown signals."""
      global collector_process
      if collector_process:
        print("Shutting down collector...")
        collector_process.terminate()
        collector_process.wait(timeout=10)
    
    @functions_framework.cloud_event
    def main(cloud_event):
      """
      Cloud Run function to run OpenTelemetry Collector for netflow collection.
    
      Args:
        cloud_event: CloudEvent object containing Pub/Sub message
      """
      global collector_process
    
      if not GCS_BUCKET:
        print('Error: GCS_BUCKET environment variable not set')
        return
    
      try:
        # Set up signal handlers
        signal.signal(signal.SIGTERM, signal_handler)
        signal.signal(signal.SIGINT, signal_handler)
    
        # Create collector configuration
        config_path, file_path = create_collector_config()
        print(f"Created collector config at {config_path}")
        print(f"Netflow receiver listening on {NETFLOW_SCHEME}://0.0.0.0:{NETFLOW_PORT}")
    
        # Start OpenTelemetry Collector
        collector_process = subprocess.Popen(
          ['/otelcol-contrib', '--config', config_path],
          stdout=subprocess.PIPE,
          stderr=subprocess.PIPE,
          text=True
        )
    
        print(f"Started OpenTelemetry Collector (PID: {collector_process.pid})")
    
        # Run for collection period (e.g., 5 minutes)
        collection_time = 300
        print(f"Collecting netflow data for {collection_time} seconds...")
    
        start_time = time.time()
        while time.time() - start_time < collection_time:
          if collector_process.poll() is not None:
            stdout, stderr = collector_process.communicate()
            print(f"Collector exited unexpectedly")
            print(f"STDOUT: {stdout}")
            print(f"STDERR: {stderr}")
            break
          time.sleep(10)
    
        # Stop collector
        if collector_process.poll() is None:
          print("Stopping collector...")
          collector_process.terminate()
          collector_process.wait(timeout=10)
    
        # Upload collected data
        upload_to_gcs(file_path)
    
        # Cleanup
        if os.path.exists(file_path):
          os.remove(file_path)
        if os.path.exists(config_path):
          os.remove(config_path)
    
        print("Netflow collection completed successfully")
    
      except Exception as e:
        print(f'Error during netflow collection: {str(e)}')
        if collector_process and collector_process.poll() is None:
          collector_process.terminate()
        raise
    
    • requirements.txt:
    functions-framework==3.*
    google-cloud-storage==2.*
    PyYAML==6.*
    
    • ‫Dockerfile:
    FROM python:3.12-slim
    
    # Install OpenTelemetry Collector Contrib
    RUN apt-get update && apt-get install -y wget && \
      wget https://github.com/open-telemetry/opentelemetry-collector-releases/releases/download/v0.144.0/otelcol-contrib_0.144.0_linux_amd64.deb && \
      dpkg -i otelcol-contrib_0.144.0_linux_amd64.deb && \
      rm otelcol-contrib_0.144.0_linux_amd64.deb && \
      apt-get clean
    
    # Set working directory
    WORKDIR /app
    
    # Copy requirements and install
    COPY requirements.txt .
    RUN pip install --no-cache-dir -r requirements.txt
    
    # Copy function code
    COPY main.py .
    
    # Expose netflow port
    EXPOSE 2055/udp
    
    # Run function
    CMD ["functions-framework", "--target=main", "--port=8080"]
    
  3. לוחצים על Deploy כדי לשמור ולפרוס את הפונקציה.

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

הגדרת מכשירי רשת לשליחת נתוני NetFlow

מגדירים את מכשירי הרשת (נתבים, מתגים, חומות אש) לייצוא נתוני netflow,‏ sflow או IPFIX לפונקציית Cloud Run.

קבלת כתובת IP חיצונית של פונקציית Cloud Run

  1. במסוף GCP, עוברים אל Cloud Run > Services.
  2. לוחצים על שם הפונקציה netflow-collector.
  3. מעתיקים את כתובת ה-URL שמוצגת בחלק העליון (לדוגמה, https://netflow-collector-xxxxx-uc.a.run.app).
  4. שם המארח מחולץ מכתובת ה-URL.
  5. כדי לפתור את כתובת ה-IP, משתמשים ב-nslookup או ב-dig:

    nslookup netflow-collector-xxxxx-uc.a.run.app
    
  • דוגמה: הגדרת Cisco router netflow

      ! Configure netflow exporter
      flow exporter OTEL-EXPORTER
      destination <CLOUD_RUN_IP>
      transport udp 2055
      source <INTERFACE>
      export-protocol netflow-v9
    
      ! Configure flow monitor
      flow monitor OTEL-MONITOR
      exporter OTEL-EXPORTER
      record netflow ipv4 original-input
    
      ! Apply to interface
      interface GigabitEthernet0/0
      ip flow monitor OTEL-MONITOR input
      ip flow monitor OTEL-MONITOR output
    
  • דוגמה: הגדרת NetFlow כללית

    ברוב מכשירי הרשת, מגדירים את ההגדרות הבאות:

    • גרסת Netflow: NetFlow v5,‏ v9 או IPFIX
    • ‫Collector IP: כתובת ה-IP של פונקציית Cloud Run
    • יציאת איסוף: 2055 (או יציאה שהוגדרה)
    • Protocol: ‏ UDP
    • זמן קצוב לתגובה פעילה: 60 שניות (מומלץ)
    • זמן קצוב לתפוגה של חוסר פעילות: 15 שניות (מומלץ)

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

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

  1. במסוף GCP, עוברים אל Cloud Scheduler.
  2. לוחצים על Create Job (יצירת משימה).
  3. מספקים את פרטי ההגדרה הבאים:

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

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

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

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

בדיקת השילוב

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

    Started OpenTelemetry Collector (PID: ...)
    Netflow receiver listening on netflow://0.0.0.0:2055
    Collecting netflow data for 300 seconds...
    Uploaded ... bytes to gs://netflow-logs-bucket/netflow/netflow_YYYYMMDD_HHMMSS.ndjson
    Netflow collection completed successfully
    
  8. עוברים אל Cloud Storage > Buckets (קטגוריות).

  9. לוחצים על שם הקטגוריה netflow-logs-bucket.

  10. עוברים לתיקייה עם הקידומת netflow/.

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

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

  • האיסוף הסתיים באופן לא צפוי: צריך לבדוק את ההגדרה של מקלט netflow
  • אין נתונים להעלאה: מוודאים שמכשירי הרשת שולחים נתוני NetFlow לכתובת ה-IP ולפורט הנכונים
  • ההרשאה נדחתה: צריך לבדוק שלחשבון השירות יש תפקיד אדמין של אובייקט אחסון
  • שגיאה בקישור ליציאה: מוודאים שהיציאה 2055 לא נמצאת כבר בשימוש

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

טבלת מיפוי UDM

בטבלה הבאה מפורט המיפוי של שדות ביומן של OpenTelemetry Netflow Receiver לשדות ב-UDM של Google SecOps:

שדה OpenTelemetry שדה UDM תיאור
source.address principal.ip כתובת ה-IP של המקור
source.port principal.port מספר יציאת המקור
destination.address target.ip כתובת ה-IP של היעד
destination.port target.port מספר יציאת היעד
network.transport network.ip_protocol פרוטוקול העברה (tcp, ‏ udp)
network.type network.ip_version גרסת ה-IP‏ (ipv4, ‏ ipv6)
flow.io.bytes network.sent_bytes סה"כ בייטים שהועברו
flow.io.packets network.sent_packets סה"כ חבילות שהועברו
flow.type metadata.product_log_id סוג התנועה (NETFLOW_V5, ‏ NETFLOW_V9, ‏ SFLOW_5, ‏ IPFIX)
flow.start metadata.event_timestamp שעת ההתחלה של התהליך
flow.end network.session_duration שעת הסיום של התהליך
flow.sampler_address intermediary.ip כתובת ה-IP של הכלי לייצוא נתוני Netflow
flow.tcp_flags network.tcp_flags דגלי TCP
flow.io.bytes event.idm.read_only_udm.network.sent_bytes מופה מיומן השינויים
source.port event.idm.read_only_udm.principal.port מופה מיומן השינויים
destination.port event.idm.read_only_udm.target.port מופה מיומן השינויים
network.transport event.idm.read_only_udm.network.ip_protocol מופה מיומן השינויים
network.type event.idm.read_only_udm.security_result.detection_fields מופה מיומן השינויים
flow.sequence_num event.idm.read_only_udm.network.session_id מופה מיומן השינויים
flow.sampler_address event.idm.read_only_udm.observer.ip מופה מיומן השינויים
flow.type event.idm.read_only_udm.metadata.product_event_type מופה מיומן השינויים

יומן שינויים

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

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