איסוף יומנים של 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
- נכנסים אל מסוף Google Cloud.
- בוחרים פרויקט או יוצרים פרויקט חדש.
- בתפריט הניווט, עוברים אל Cloud Storage > Buckets.
- לוחצים על Create bucket.
מספקים את פרטי ההגדרה הבאים:
הגדרה ערך Name your bucket (שם הקטגוריה) מזינים שם ייחודי גלובלית (לדוגמה, netflow-logs-bucket)סוג המיקום בוחרים לפי הצרכים (אזור, שני אזורים, מספר אזורים) Location בוחרים את המיקום (לדוגמה, us-central1).סוג אחסון (storage class) רגיל (מומלץ ליומנים שמתבצעת אליהם גישה לעיתים קרובות) בקרת גישה אחיד (מומלץ) כלים להגנה אופציונלי: הפעלת ניהול גרסאות של אובייקטים או מדיניות שמירת נתונים לוחצים על יצירה.
יצירת חשבון שירות לפונקציית Cloud Run
פונקציית Cloud Run צריכה חשבון שירות עם הרשאות כתיבה לקטגוריה של GCS והרשאה להפעלה על ידי Pub/Sub.
יצירת חשבון שירות
- במסוף GCP, עוברים אל IAM & Admin > Service Accounts.
- לוחצים על יצירת חשבון שירות.
- מספקים את פרטי ההגדרה הבאים:
- שם חשבון השירות: מזינים
netflow-collector-sa - תיאור חשבון השירות: מזינים
Service account for Cloud Run function to collect netflow logs
- שם חשבון השירות: מזינים
- לוחצים על יצירה והמשך.
- בקטע Grant this service account access to project (מתן גישה של חשבון השירות הזה לפרויקט), מוסיפים את התפקידים הבאים:
- לוחצים על בחירת תפקיד.
- מחפשים את Storage Object Admin ולוחצים עליו.
- לוחצים על + הוספת תפקיד נוסף.
- מחפשים את האפשרות Cloud Run Invoker (הפעלת שירותים ב-Cloud Run) ולוחצים עליה.
- לוחצים על + הוספת תפקיד נוסף.
- מחפשים את האפשרות Cloud Functions Invoker (הפעלת פונקציות של Cloud Functions) ולוחצים עליה.
- לוחצים על Continue.
- לוחצים על סיום.
התפקידים האלה נדרשים כדי:
- אדמין של אובייקט אחסון: כתיבת יומנים לקטגוריית GCS וניהול קבצי מצב
- Cloud Run Invoker: מתן הרשאה ל-Pub/Sub להפעיל את הפונקציה
- Cloud Functions Invoker: מאפשר הפעלה של פונקציות
מתן הרשאות IAM בקטגוריית GCS
נותנים לחשבון השירות הרשאות כתיבה בקטגוריית GCS:
- עוברים אל Cloud Storage > Buckets (קטגוריות).
- לוחצים על שם הקטגוריה (לדוגמה,
netflow-logs-bucket). - עוברים לכרטיסייה Permissions.
- לוחצים על הענקת גישה.
- מספקים את פרטי ההגדרה הבאים:
- Add principals (הוספת ישויות מורשות): מזינים את כתובת האימייל בחשבון של חשבון השירות (לדוגמה,
netflow-collector-sa@PROJECT_ID.iam.gserviceaccount.com). - הקצאת תפקידים: בוחרים באפשרות אדמין של אובייקטים באחסון.
- Add principals (הוספת ישויות מורשות): מזינים את כתובת האימייל בחשבון של חשבון השירות (לדוגמה,
- לוחצים על Save.
יצירת נושא Pub/Sub
יוצרים נושא Pub/Sub ש-Cloud Scheduler יפרסם בו ופונקציית Cloud Run תירשם אליו.
- במסוף GCP, עוברים אל Pub/Sub > Topics.
- לוחצים על יצירת נושא.
- מספקים את פרטי ההגדרה הבאים:
- מזהה הנושא: מזינים
netflow-trigger - השארת ההגדרות האחרות כברירת מחדל
- מזהה הנושא: מזינים
- לוחצים על יצירה.
יצירת פונקציית Cloud Run לאיסוף יומני NetFlow
פונקציית Cloud Run תפעיל OpenTelemetry Collector שמקבל נתוני netflow ומייצא אותם ל-GCS.
- במסוף GCP, עוברים אל Cloud Run.
- לוחצים על יצירת שירות.
- בוחרים באפשרות Function (שימוש בעורך ישיר ליצירת פונקציה).
בקטע Configure (הגדרה), מזינים את פרטי ההגדרה הבאים:
הגדרה ערך שם השירות netflow-collectorאזור בוחרים אזור שתואם ל-GCS Bucket (לדוגמה, us-central1)זמן ריצה בוחרים באפשרות Python 3.12 ואילך. בקטע טריגר (אופציונלי):
- לוחצים על + הוספת טריגר.
- בוחרים באפשרות Cloud Pub/Sub.
- בקטע Select a Cloud Pub/Sub topic, בוחרים את נושא ה-Pub/Sub
netflow-trigger. - לוחצים על Save.
בקטע Authentication:
- בוחרים באפשרות דרישת אימות.
- בודקים את ניהול זהויות והרשאות גישה (IAM).
גוללים למטה ומרחיבים את Containers, Networking, Security (מאגרי תגים, רשתות, אבטחה).
עוברים לכרטיסייה אבטחה:
- חשבון שירות: בוחרים את חשבון השירות
netflow-collector-sa
- חשבון שירות: בוחרים את חשבון השירות
עוברים לכרטיסייה מאגרי תגים:
- לוחצים על Variables & Secrets (משתנים וסודות).
- לוחצים על + הוספת משתנה לכל משתנה סביבתי:
שם המשתנה ערך לדוגמה תיאור GCS_BUCKETnetflow-logs-bucketשם קטגוריית GCS GCS_PREFIXnetflowקידומת לקובצי יומן NETFLOW_PORT2055יציאה למקלט NetFlow NETFLOW_SCHEMEnetflowסוג הסכימה: netflow, sflow או ipfix NETFLOW_SOCKETS4מספר שקעי UDP NETFLOW_WORKERS8מספר העובדים של פענוח בקטע Variables & Secrets, גוללים למטה אל Requests:
- זמן קצוב לתפוגת בקשה: מזינים
600שניות (10 דקות)
- זמן קצוב לתפוגת בקשה: מזינים
עוברים לכרטיסייה הגדרות:
- בקטע Resources (מקורות מידע):
- זיכרון: בוחרים באפשרות 1 GiB או יותר
- מעבד: בוחרים באפשרות 2.
- בקטע Resources (מקורות מידע):
בקטע שינוי קנה מידה של העריכה:
- מספר מינימלי של מופעים: מזינים
1 - מספר מקרים מקסימלי: מזינים
10
- מספר מינימלי של מופעים: מזינים
לוחצים על יצירה.
ממתינים ליצירת השירות (1-2 דקות).
אחרי שיוצרים את השירות, עורך הקוד המוטבע נפתח באופן אוטומטי.
הוספת קוד פונקציה
- מזינים main בשדה Entry point.
בעורך הקוד המוטבע, יוצרים שלושה קבצים:
- 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"]לוחצים על Deploy כדי לשמור ולפרוס את הפונקציה.
ממתינים לסיום הפריסה (3-5 דקות).
הגדרת מכשירי רשת לשליחת נתוני NetFlow
מגדירים את מכשירי הרשת (נתבים, מתגים, חומות אש) לייצוא נתוני netflow, sflow או IPFIX לפונקציית Cloud Run.
קבלת כתובת IP חיצונית של פונקציית Cloud Run
- במסוף GCP, עוברים אל Cloud Run > Services.
- לוחצים על שם הפונקציה
netflow-collector. - מעתיקים את כתובת ה-URL שמוצגת בחלק העליון (לדוגמה,
https://netflow-collector-xxxxx-uc.a.run.app). - שם המארח מחולץ מכתובת ה-URL.
כדי לפתור את כתובת ה-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.
- במסוף GCP, עוברים אל Cloud Scheduler.
- לוחצים על Create Job (יצירת משימה).
מספקים את פרטי ההגדרה הבאים:
הגדרה ערך שם netflow-collector-hourlyאזור בחירה באותו אזור כמו פונקציית Cloud Run תדירות 0 * * * *(כל שעה, בדיוק בשעה)אזור זמן בחירת אזור זמן (מומלץ לבחור UTC) סוג הטירגוט Pub/Sub נושא בוחרים את נושא ה-Pub/Sub netflow-triggerגוף ההודעה {}(אובייקט JSON ריק)לוחצים על יצירה.
אפשרויות לתדירות התזמון
בחירת התדירות בהתאם לנפח היומן ולדרישות זמן האחזור:
| תדירות | ביטוי Cron | תרחיש לדוגמה |
|---|---|---|
| כל 5 דקות | */5 * * * * |
נפח גבוה, זמן אחזור נמוך |
| כל 15 דקות | */15 * * * * |
נפח חיפושים בינוני |
| כל שעה | 0 * * * * |
רגיל (מומלץ) |
| כל 6 שעות | 0 */6 * * * |
נפח נמוך, עיבוד באצווה |
בדיקת השילוב
- במסוף Cloud Scheduler, מוצאים את הג'וב
netflow-collector-hourly. - לוחצים על הפעלת המשימה בכוח כדי להפעיל את המשימה באופן ידני.
- מחכים כמה שניות.
- עוברים אל Cloud Run > Services.
- לוחצים על שם הפונקציה
netflow-collector. - לוחצים על הכרטיסייה Logs (יומנים).
מוודאים שהפונקציה בוצעה בהצלחה. מחפשים:
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עוברים אל Cloud Storage > Buckets (קטגוריות).
לוחצים על שם הקטגוריה
netflow-logs-bucket.עוברים לתיקייה עם הקידומת
netflow/.מוודאים שנוצר קובץ
.ndjsonחדש עם חותמת הזמן הנוכחית.
אם מופיעות שגיאות ביומנים:
- האיסוף הסתיים באופן לא צפוי: צריך לבדוק את ההגדרה של מקלט netflow
- אין נתונים להעלאה: מוודאים שמכשירי הרשת שולחים נתוני NetFlow לכתובת ה-IP ולפורט הנכונים
- ההרשאה נדחתה: צריך לבדוק שלחשבון השירות יש תפקיד אדמין של אובייקט אחסון
- שגיאה בקישור ליציאה: מוודאים שהיציאה 2055 לא נמצאת כבר בשימוש
אחזור חשבון השירות של Google SecOps
Google SecOps משתמש בחשבון שירות ייחודי כדי לקרוא נתונים מהקטגוריה של GCS. צריך להעניק לחשבון השירות הזה גישה לדלי.
קבלת כתובת האימייל בחשבון השירות
- עוברים אל SIEM Settings > Feeds (הגדרות SIEM > פידים).
- לוחצים על הוספת פיד חדש.
- לוחצים על Configure a single feed (הגדרת פיד יחיד).
- בשדה שם הפיד, מזינים שם לפיד (לדוגמה,
OpenTelemetry Netflow Logs). - בוחרים באפשרות Google Cloud Storage V2 בתור סוג המקור.
- בוחרים באפשרות NETFLOW_OTEL בתור סוג היומן.
לוחצים על קבלת חשבון שירות. תוצג כתובת אימייל ייחודית של חשבון שירות. לדוגמה:
chronicle-12345678@chronicle-gcp-prod.iam.gserviceaccount.comמעתיקים את כתובת האימייל הזו לשימוש בשלב הבא.
לוחצים על הבא.
מציינים ערכים לפרמטרים הבאים של הקלט:
כתובת ה-URL של קטגוריית האחסון: מזינים את ה-URI של קטגוריית GCS עם נתיב הקידומת:
gs://netflow-logs-bucket/netflow/
אפשרות למחיקת המקור: בוחרים את אפשרות המחיקה לפי ההעדפה שלכם:
- אף פעם: לא נמחקים קבצים אחרי ההעברה (מומלץ לבדיקות).
- מחיקת קבצים שהועברו: הקבצים נמחקים אחרי שההעברה מסתיימת.
מחיקת קבצים שהועברו וספריות ריקות: מחיקת קבצים וספריות ריקות אחרי שההעברה מסתיימת בהצלחה.
הגיל המקסימלי של הקובץ: כולל קבצים ששונו במספר הימים האחרון (ברירת המחדל היא 180 ימים)
מרחב השמות של הנכס: מרחב השמות של הנכס
תוויות להוספה: התווית שתתווסף לאירועים מהפיד הזה
לוחצים על הבא.
בודקים את ההגדרות של הפיד החדש במסך סיום ולוחצים על שליחה.
הענקת הרשאות IAM לחשבון השירות של Google SecOps
לחשבון השירות של Google SecOps צריך להיות התפקיד צפייה באובייקט אחסון בקטגוריה של GCS.
- עוברים אל Cloud Storage > Buckets (קטגוריות).
- לוחצים על שם הקטגוריה
netflow-logs-bucket. - עוברים לכרטיסייה Permissions.
- לוחצים על הענקת גישה.
- מספקים את פרטי ההגדרה הבאים:
- Add principals (הוספת ישויות מורשות): מדביקים את כתובת האימייל בחשבון של חשבון השירות של Google SecOps
- הקצאת תפקידים: בוחרים באפשרות צפייה באובייקטים באחסון.
- לוחצים על 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.