יצירה של שאילתות מרובות שלבים

נתמך ב:

במאמר הזה מוסבר איך שאילתות מרובות שלבים ב-YARA-L מאפשרות להזין את הפלט של שלב אחד בשאילתה ישירות לקלט של שלב עוקב. התהליך הזה נותן לכם יותר שליטה בהמרת הנתונים מאשר שאילתה אחת ומונוליטית.

שילוב שאילתות מרובות שלבים עם תכונות קיימות

שאילתות מרובות שלבים פועלות בשילוב עם התכונות הקיימות הבאות ב-Google Security Operations:

  • כללי זיהוי מורכבים: שאילתות רב-שלביות משלימות כללי זיהוי מורכבים על ידי גישור הפער בין זיהוי אוטומטי לחקירה פעילה. בעוד שכללים מורכבים מצטיינים בזיהוי מתאמים מורכבים של כמה אירועים על פני חלונות זמן מורחבים, שאילתות רב-שלביות מאפשרות לאנליסטים לעבור מהזיהויים האלה לחיפושים איטרטיביים בזמן אמת כדי לאמת באופן מיידי את האיום המתפתח ולהגדיר את היקף ההשפעה שלו.

  • טווחים של זמן וכללים של כמה אירועים: אתם יכולים להשתמש בשאילתות מרובות שלבים כדי לזהות אנומליות על ידי השוואה בין חלונות זמן שונים בנתונים. לדוגמה, אפשר להשתמש בשאילתות הראשוניות כדי ליצור בסיס השוואה לאורך תקופה ממושכת, ואז להשתמש בשאילתות מאוחרות יותר כדי להעריך את הפעילות האחרונה בהשוואה לבסיס ההשוואה הזה. אפשר גם להשתמש בכללים מרובי-אירועים כדי ליצור השוואה דומה.

שאילתות מרובות שלבים ב-YARA-L נתמכות גם בלוחות בקרה וגם בחיפוש.

הצטרפות עוזרת לקשר בין נתונים ממספר מקורות כדי לספק יותר הקשר לחקירה. קישור של אירועים קשורים, ישויות ונתונים אחרים מאפשר לכם לחקור תרחישי תקיפה מורכבים. מידע נוסף מפורט במאמר בנושא שימוש בצירופים בחיפוש.

שיקולים עיקריים

כשמגדירים שאילתה מרובת שלבים, חשוב לשים לב לנקודות הבאות:

  • שלב הגבלה: שאילתות מרובות שלבים צריכות להכיל בין שלב אחד לארבעה שלבים עם שם, בנוסף לשלב הבסיס.
  • תחביר של הזמנה: תמיד צריך להגדיר את התחביר של השלב עם השם לפני שמגדירים את התחביר של שלב הבסיס.

מומלץ לעיין בבעיות הידועות הבאות ובפתרונות העקיפים המומלצים כשמטמיעים שאילתות מרובות שלבים:

  • כל השאילתות הרב-שלביות מתנהגות כמו שאילתות חיפוש סטטיסטיות (הפלט מורכב מנתונים סטטיסטיים נצברים ולא מאירועים לא נצברים או משורות בטבלת נתונים).

  • הביצועים של שאילתות איחוד עם UDM ואירועים של ישויות בצד אחד יכולים להיות נמוכים בגלל הגודל של מערך הנתונים הזה. מומלץ מאוד לסנן את האירועים של UDM ושל הישות בצד של ה-join כמה שיותר (לדוגמה, סינון לפי סוג האירוע).

הנחיות כלליות לגבי שיטות מומלצות זמינות במאמר בנושא שיטות מומלצות ל-YARA-L 2.0. מידע ספציפי על הצטרפות זמין במאמר בנושא שיטות מומלצות.

מונחים חשובים

בהקשר של צירופים, שלב עם חלון הוא שלב עם קטע match שמכיל חלון. לעומת זאת, שלב של טבלה לא מוציא חלונות.

יצירה של שאילתת YARA-L מרובת שלבים

כדי ליצור שאילתת YARA-L מרובת שלבים, מבצעים את השלבים הבאים.

מבנה ותחביר של שלבים

עוברים אל חקירה > חיפוש. כשמגדירים את שלבי השאילתה, צריך לפעול לפי המבנה הבא:

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

stage <stage name> { }

  • סוגריים מסולסלים: מציבים את כל תחביר השלב בתוך סוגריים מסולסלים {}.

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

  • הפניה: כל שלב יכול להפנות לשלבים שהוגדרו מוקדם יותר בשאילתה.

  • שלב הבסיס: לשאילתה חייב להיות שלב בסיס, שעובר עיבוד אחרי כל השלבים עם השמות.

בדוגמה הבאה, השלב daily_stats אוסף נתונים סטטיסטיים יומיים של הרשת:

stage daily_stats {
  metadata.event_type = "NETWORK_CONNECTION"
  $source = principal.hostname
  $target = target.ip
  $source != ""
  $target != ""
  match:
    $source, $target by day
  outcome:
    $exchanged_bytes = sum(network.sent_bytes + network.received_bytes)
}

פלט של שלב הגישה

הפלט של שלב עם שם נגיש לשלבים הבאים באמצעות שדות שלבים. שדות השלב תואמים למשתני match ו-outcome של השלב, ואפשר להשתמש בהם באופן דומה לשדות של מודל הנתונים המאוחד (UDM).

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

$<stage name>.<variable name>

אופציונלי: גישה לחותמות זמן של חלון

אם שלב עם שם משתמש בחלון קופץ, חלון הזזה או חלון עוקב (tumbling window), אפשר לגשת לזמן ההתחלה והסיום של החלון לכל שורת פלט באמצעות השדות השמורים הבאים:

  • $<stage name>.window_start

  • $<stage name>.window_end

השדות window_start ו-window_end הם שדות של מספרים שלמים שמציינים את מספר השניות מאז ראשית זמן יוניקס. הגודל של חלונות בשלבים שונים יכול להיות שונה.

דוגמאות לשאילתות מרובות שלבים

הדוגמאות בקטע הזה ממחישות איך אפשר ליצור שאילתת YARA-L מלאה עם כמה שלבים.

דוגמה: חיפוש חיבורים לרשת עם פעילות חריגה (שעות)

בדוגמה הזו של YARA-L מרובת שלבים מזוהים זוגות של כתובות IP עם פעילות ברשת שהיא גבוהה מהרגיל, והיא מיועדת לזוגות ששומרים על פעילות גבוהה במשך יותר משלוש שעות. השאילתה כוללת שני רכיבים נדרשים: השלב עם השם, hourly_stats, והשלב root.

בשלב hourly_stats מתבצע חיפוש של זוגות principal.ip ו-target.ip עם רמות גבוהות של פעילות ברשת.

בשלב הזה מוחזר ערך שעתי אחד לשדות הבאים:

  • נתונים סטטיסטיים של כתובת ה-IP של המקור (מחרוזת): $hourly_stats.src_ip

  • נתונים סטטיסטיים של כתובת ה-IP של היעד (מחרוזת): $hourly_stats.dst_ip

  • נתונים סטטיסטיים של מספר האירועים (מספר שלם): $hourly_stats.count

  • סטיית תקן של בייטים שהתקבלו (מספר עשרוני): $hourly_stats.std_recd_bytes

  • מספר הבייטים הממוצע שהתקבל (מספר עשרוני): $hourly_stats.avg_recd_bytes

  • שעת ההתחלה של משבצת השעה בשניות מאז ראשית זמן יוניקס (Unix epoch) (מספר שלם): $hourly_stats.window_start

  • שעת הסיום של משבצת השעה בשניות מאז ראשית זמן יוניקס (מספר שלם): $hourly_stats.window_end

בשלב הבסיס מעבדים את הפלט של שלב hourly_stats. הוא מחשב נתונים סטטיסטיים עבור צמדים של principal.ip ו-target.ip עם פעילות שחורגת מהסף שצוין על ידי $hourly_stats. לאחר מכן המערכת מסננת את הזוגות עם יותר משלוש שעות של פעילות גבוהה.


stage hourly_stats {
  metadata.event_type = "NETWORK_CONNECTION"
  $src_ip = principal.ip
  $dst_ip = target.ip
  $src_ip != ""
  $dst_ip != ""

  match:
    $src_ip, $dst_ip by hour

  outcome:
    $count = count(metadata.id)
    $avg_recd_bytes = avg(network.received_bytes)
    $std_recd_bytes = stddev(network.received_bytes)

  condition:
    $avg_recd_bytes > 100 and $std_recd_bytes > 50
}

$src_ip = $hourly_stats.src_ip
$dst_ip = $hourly_stats.dst_ip
$time_bucket_count = strings.concat(timestamp.get_timestamp($hourly_stats.window_start), "|", $hourly_stats.count)

match:
 $src_ip, $dst_ip

outcome:
 $list = array_distinct($time_bucket_count)
 $count = count_distinct($hourly_stats.window_start)

condition:
 $count > 3

אם משנים את תנאי ההתאמה בשלב הבסיסי באופן הבא, אפשר להוסיף צבירה עם חלון לפי יום לשאילתה הרב-שלבית.

match:
 $src_ip, $dst_ip by day

דוגמה: חיפוש חיבורים לרשת עם פעילות חריגה (באמצעות ציון Z)

השאילתה הרב-שלבית הזו משווה את הפעילות הממוצעת ברשת מדי יום לפעילות של היום באמצעות חישוב ציון Z (מדידת מספר סטיות התקן מהממוצע). השאילתה הזו מחפשת למעשה פעילות רשת גבוהה באופן חריג בין נכסים פנימיים לבין מערכות חיצוניות.

דרישה מוקדמת: כדי שהציון Z המחושב יהיה יעיל, חלון הזמן של השאילתה צריך להיות גדול מ-2 ימים או שווה ל-2 ימים, ולכלול את היום הנוכחי.

השאילתה הרב-שלבית הזו כוללת את השלב daily_stats ואת השלב root, שפועלים יחד כדי לחשב את ציון ה-Z של פעילות הרשת:

  • בשלב daily_stats מתבצעת הצבירה היומית הראשונית. היא מחשבת את סך הבייטים שהועברו בכל יום לכל זוג כתובות IP ‏ (source ו-target) ומחזירה את השדות הבאים של השלב (שמתאימים לעמודות בשורות הפלט):

    • ‫$daily_stats.source: יחיד, מחרוזת
    • ‫$daily_stats.target: יחיד, מחרוזת
    • $daily_stats.exchanged_bytes: יחיד, מספר שלם
    • $daily_stats.window_start: יחיד, מספר שלם
    • $daily_stats.window_end: יחיד, מספר שלם

  • בשלב הבסיס מתבצעת צבירה של daily_stats פלט השלב לכל צמד כתובות IP. הפונקציה מחשבת את הממוצע ואת סטיית התקן של הבייטים היומיים שהועברו בכל טווח החיפוש, יחד עם הבייטים שהועברו היום. המערכת משתמשת בשלושת הערכים המחושבים האלה כדי לקבוע את ציון ה-Z.

  • בפלט מפורטים ציוני ה-Z של כל זוגות כתובות ה-IP של היום, ממוינים בסדר יורד.

// Calculate the total bytes exchanged per day by source and target

stage daily_stats {
  metadata.event_type = "NETWORK_CONNECTION"
  $source = principal.hostname
  $target = target.ip
  $source != ""
  $target != ""

  match:
    $source, $target by day

  outcome:
    $exchanged_bytes = sum(network.sent_bytes + network.received_bytes)
}

// Calculate the average per day over the time window and compare with the bytes exchanged today

$source = $daily_stats.source
$target = $daily_stats.target
$date = timestamp.get_date($daily_stats.window_start)

match:
  $source, $target

outcome:
  $today_bytes = sum(if($date = timestamp.get_date(timestamp.current_seconds()), cast.as_int($daily_stats.exchanged_bytes), 0))
  $average_bytes = window.avg($daily_stats.exchanged_bytes)
  $stddev_bytes = window.stddev($daily_stats.exchanged_bytes)
  $zscore = ($today_bytes - $average_bytes) / $stddev_bytes

order:
  $zscore desc

ייצוא משתנים לא מצטברים משלבים

בשאילתה טיפוסית עם כמה שלבים, הנתונים מועברים בין השלבים בתהליך צבירה, שיכול 'לצמצם' כמה אירועים לסיכום אחד. עם זאת, יש תרחישים שבהם צריך לשמור את הפרטים הספציפיים של כל אירוע – כמו מזהה תהליך ייחודי או שורת פקודה ספציפית – בלי לאבד את רמת הפירוט הזו. כדי לתמוך בזה, אפשר לייצא משתנים ישירות בלי פונקציית קיבוץ.

שלבים עם שם יכולים לכלול קטע outcome לא מצטבר. המשמעות היא שמשתנים שמוגדרים בקטע outcome הזה מופקים ישירות מהשלב, כך ששלבים עוקבים יכולים לגשת אליהם כשדות של השלב בלי שנדרש צבירה מקובצת.

הפרט הזה מועיל כי הוא מאפשר לכם:

  • שמירה על נאמנות הנתונים: אתם יכולים להעביר את המאפיינים המדויקים של אירוע (לדוגמה, נתיב קובץ ספציפי) לשלב הבא בלי להשתמש בצבירה מלאכותית של 'placeholder' כמו max() או array_distinct().
  • מפחית את מורכבות השאילתות: הוא מפשט את הלוגיקה של YARA-L כי הוא מבטל את הצורך בקטע match או בהוראות קיבוץ רק כדי להעביר ערך משלב 1 לשלב 2.
  • אופטימיזציה של הביצועים: המערכת יכולה להזרים נתונים בין שלבים בצורה יעילה יותר, וכך לקצר את זמני הביצוע של חיפושים מורכבים עם נפח גבוה, כי היא לא משתמשת במנוע הצבירה.

דוגמה: ייצוא של משתנה לא מצטבר

בדוגמה הזו מוסבר איך לייצא משתנים לא מצטברים. שימו לב ללוגיקה הבאה:

  • ‫top_5_bytes_sent מבצע חיפוש של חמשת האירועים עם פעילות הרשת הגבוהה ביותר.

  • בשלב top_5_bytes_sent מוצגים שדות השלב הבאים שמתאימים לעמודות בשורות הפלט:

    • $top_5_bytes_sent.bytes_sent: יחיד, מספר שלם
    • $top_5_bytes_sent.timestamp_seconds: יחיד, מספר שלם

  • בשלב root מחושבות חותמות הזמן המוקדמות והמאוחרות ביותר של חמשת האירועים עם פעילות הרשת הגבוהה ביותר.

stage top_5_bytes_sent {
  metadata.event_type = "NETWORK_CONNECTION"
  network.sent_bytes > 0

  outcome:
    $bytes_sent = network.sent_bytes
    $timestamp_seconds = metadata.event_timestamp.seconds

  order:
    $bytes_sent desc

  limit:
    5
}

outcome:
  $latest_timestamp = timestamp.get_timestamp(max($top_5_bytes_sent.timestamp_seconds))
  $earliest_timestamp = timestamp.get_timestamp(min($top_5_bytes_sent.timestamp_seconds))

הטמעה של חלונות בשאילתות מרובות שלבים

בזיהויים מרובי-שלבים, חלונות מאפשרים להגדיר גבולות זמן ספציפיים לקורלציה של אירועים בתוך שלב. חלוקת הנתונים למקטעים נפרדים של זמן, כמו חלון נע של 5 דקות או חלון עוקב (tumbling window) של שעה, מאפשרת לזהות דפוסים כמו מתקפות כוח ברוטלי או התנהגות של שידור אותות, שמתגלים רק כשמנתחים אותם כרצף מוגבל בזמן.

שאילתות מרובות שלבים תומכות בכל סוגי החלונות (דילוג, הזזה וסיבוב) בשלבים עם שמות. כך תוכלו להעביר נתונים עם הקשר של זמן בין שלבים, למשל להשתמש בשלב 1 כדי לזהות חלון אירועים בתדירות גבוהה ובשלב 2 כדי ליצור קורלציה בין החלון הספציפי הזה לבין פעולות ניהוליות עוקבות.

אם שלב עם שם כולל חלון, אפשר לגשת להתחלת החלון ולסוף החלון של כל שורת פלט באמצעות השדות השמורים הבאים:

  • ‫$stage_window_start: חותמת הזמן של מערכת Unix שמציינת את תחילת חלון הזמן.
  • ‫$stage_window_end: חותמת הזמן של מערכת Unix שמציינת את סוף חלון הזמן.

פרטים נוספים על חלונות זמינים במאמר הלוגיקה של חלונות ב-YARA-L 2.0.

תרחישים נפוצים לדוגמה

  • זיהוי רציף: העברת מסגרת הזמן הספציפית של עלייה חדה שזוהתה בניסיונות כניסה שנכשלו לשלב שני, שבו מתבצע חיפוש של כניסה מוצלחת זמן קצר לאחר מכן.
  • ניתוח משך הזמן: חישוב הזמן שחלף עד לפריצה על ידי השוואה בין $stage_window_start של שלב ניצול לרעה ראשוני לבין חותמות הזמן של אירועים בשלב מאוחר יותר.
  • הגדרת נקודת השוואה היסטורית: שימוש בחלונות כדי להשוות בין מספרי האירועים הנוכחיים לבין משתני הפלט של חלון קודם.

דוגמה: חלון קפיצה

בדוגמה הבאה אפשר לראות איך משתמשים בחלונות קפיצה בשאילתה רב-שלבית:

  • hourly_stats מחפש בשלבים זוגות של כתובות IP עם פעילות רשת גבוהה באותה שעה.

  • הפלט של hourly_stats כולל את השדות הבאים של השלב, שמתאימים לעמודות בשורות הפלט:

    • ‫$hourly_stats.src_ip: יחיד, מחרוזת
    • ‫$hourly_stats.dst_ip: יחיד, מחרוזת
    • $hourly_stats.count: יחיד, מספר שלם
    • ‫$hourly_stats.std_recd_bytes: יחיד, מספר ממשי (float)
    • ‫$hourly_stats.avg_recd_bytes: יחיד, מספר ממשי (float)
    • $hourly_stats.window_start: יחיד, מספר שלם
    • $hourly_stats.window_end: יחיד, מספר שלם

  • המסנן בשלב הבסיסי מסנן זוגות של כתובות IP עם יותר מ-3 שעות של פעילות גבוהה. יכול להיות שיהיה חפיפה בין השעות בגלל השימוש בחלון קפיצה בשלב hourly_stats.

stage hourly_stats {
  metadata.event_type = "NETWORK_CONNECTION"
  $src_ip = principal.ip
  $dst_ip = target.ip
  $src_ip != ""
  $dst_ip != ""

  match:
    $src_ip, $dst_ip over 1h

  outcome:
    $count = count(metadata.id)
    $avg_recd_bytes = avg(network.received_bytes)
    $std_recd_bytes = stddev(network.received_bytes)

  condition:
    $avg_recd_bytes > 100 and $std_recd_bytes > 50
}

$src_ip = $hourly_stats.src_ip
$dst_ip = $hourly_stats.dst_ip
$time_bucket_count = strings.concat(timestamp.get_timestamp($hourly_stats.window_start), "|", $hourly_stats.count)

match:
 $src_ip, $dst_ip

outcome:
 $list = array_distinct($time_bucket_count)
 $count = count_distinct($hourly_stats.window_start)

condition:
 $count > 3

צירופים פנימיים בשאילתות מרובות שלבים

הצטרפות פנימית מאפשרת לכם ליצור קורלציה בין נתונים בשלבים שונים או בסוגים שונים של מקורות, וכך ליצור תהליכי עבודה אנליטיים מורכבים, כמו השוואה בין אירועים בזמן אמת לבין נתוני בסיס סטטיסטיים שחושבו מראש. כשמצטרפים לשלבים, אפשר להעשיר את נתוני הטלמטריה הגולמיים בנתונים עם שמירת מצב (stateful) (כמו חציון או טבלאות חיפוש) כדי לזהות חריגות או איומים מרובי-וקטורים, שמסנן אירועים חד-שלבי לא יזהה.

יש תמיכה בצירופים פנימיים בתוך השלבים של שאילתות מרובות שלבים וביניהם. הפונקציה inner join תומכת בסוגים הבאים:

  • ‫UDM ו-UDM: מתאם בין שני סטים שונים של אירועי אבטחה (לדוגמה, התאמה בין אירוע התחברות לבין גישה לקובץ שמתרחשת אחריו).
  • ‫UDM ו-ECG: מיזוג נתוני אירועים עם מידע מגרף הקשר של ישויות לצורך העשרה של זהויות או נכסים.
  • ‫UDM וטבלת נתונים: צירוף אירועים בזמן אמת לרשימות הפניה סטטיות או לרשימות שהועלו (לדוגמה, רשימה של נכסים בעלי ערך גבוה או טווחי כתובות IP ספציפיים למחלקה).

בדוגמה הבאה מוצג אופן ההגדרה של צירוף ללא התאמה (צירוף שמתבצע בקטע outcome או events ולא בקטע match) בין אירועי UDM לבין שלב של טבלה מחושבת. התבנית הזו מאפשרת לכם לבצע זיהוי סטטיסטי של חריגות, כמו שמוצג בחישוב הבא של סטיית ממוצע מוחלטת (MAD):

  • שלב median: המערכת מחשבת את החציון של הבייטים שנשלחו לכל צמד של מארח מקור וכתובת IP של יעד.
    • ‫$median.host: יחיד, מחרוזת
    • ‫$median.target: יחיד, מחרוזת
    • ‫$median.median: יחיד, מספר ממשי (float)
  • absolute_deviations שלב: מצטרף לכל אירוע UDM עם השורה המתאימה מהשלב החציוני. כך תוכלו לחשב את הסטייה המוחלטת של הבייטים שנשלחו לכל אירוע בנפרד ביחס לקבוצת השוואה.
    • ‫$absolute_deviations.host: יחיד, מחרוזת
    • ‫$absolute_deviations.target: יחיד, מחרוזת
    • ‫$absolute_deviations.absolute_deviation: יחיד, מספר ממשי (float)
  • שלב השורש: המערכת מחשבת את הממוצע של הסטיות המוחלטות האלה בכל האירועים של UDM כדי לקבוע סף חריגה.

דוגמה: הגדרת צירוף ללא התאמה

stage median {
  metadata.event_type = "NETWORK_CONNECTION"
  $host = principal.hostname
  $target = target.ip

  match:
    $host, $target

  outcome:
    $median = window.median(network.sent_bytes, true)
}

stage absolute_deviations {
  metadata.event_type = "NETWORK_CONNECTION"
  $join_host = principal.hostname
  $join_host = $median.host
  $join_target = target.ip[0]
  $join_target = $median.target

  outcome:
    $host = $join_host
    $target = $join_target
    $absolute_deviation = math.abs(network.sent_bytes - $median.median)
}

$host = $absolute_deviations.host
$target = $absolute_deviations.target

match:
  $host, $target

outcome:
  $mean_absolute_deviation = avg($absolute_deviations.absolute_deviation)

דוגמה: צירוף ללא התאמה בין שלב של חלון ושלב של טבלה

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

  • בשלב hourly_stats מחושב המספר הכולל של בייטים שנשלחו לכל צמד של מארח מקור ומארח יעד, ולכל משבצת זמן של שעה.
  • הפלט של שלב hourly_stats כולל את השדות הבאים שמתאימים לעמודות בשורות הפלט:
    • ‫$hourly_stats.source_host: יחיד, מחרוזת
    • ‫$hourly_stats.dst_host: יחיד, מחרוזת
    • ‫$hourly_stats.total_bytes_sent: יחיד, מספר ממשי (float)
    • $hourly_stats.window_start: יחיד, מספר שלם
    • $hourly_stats.window_end: יחיד, מספר שלם
  • בשלב agg_stats מחושבים הממוצע וסטיית התקן של בייטים לשעה לכל צמד של מארח מקור ומארח יעד.

  • הפלט של agg_stats כולל את השדות הבאים של השלב, שמתאימים לעמודות בשורות הפלט:

    • ‫$agg_stats.source_host: יחיד, מחרוזת
    • ‫$agg_stats.dst_host: יחיד, מחרוזת
    • ‫$agg_stats.avg_bytes_sent: יחיד, מספר ממשי (float)
    • ‫$agg_stats.stddev_bytes_sent: יחיד, מספר ממשי (float)

  • בשלב הבסיס, כל שורה מ-hourly_stats מצורפת לשורה מ-agg_stats עבור אותו זוג של מארח מקור ומארח יעד. לכל צמד של מארח מקור ומארח יעד, המערכת מחשבת את ציון ה-z באמצעות סך הבייטים שנשלחו עבור דלי המארחים הזה והנתונים הסטטיסטיים המצטברים.

stage hourly_stats {
 $source_host = principal.hostname
 $dst_host = target.hostname
 principal.hostname != ""
 target.hostname != ""
 match:
   $source_host, $dst_host by hour
 outcome:
   $total_bytes_sent = sum(network.sent_bytes)
}

stage agg_stats {
  $source_host = $hourly_stats.source_host
  $dst_host = $hourly_stats.dst_host
  match:
    $source_host, $dst_host
  outcome:
   $avg_bytes_sent = avg($hourly_stats.total_bytes_sent)
   $stddev_bytes_sent = stddev($hourly_stats.total_bytes_sent)
}

$source_host = $agg_stats.source_host
$source_host = $hourly_stats.source_host

$dst_host = $agg_stats.dst_host
$dst_host = $hourly_stats.dst_host

outcome:
  $hour_bucket = timestamp.get_timestamp($hourly_stats.window_start)
  $z_score = ($hourly_stats.total_bytes_sent - $agg_stats.avg_bytes_sent)/$agg_stats.stddev_bytes_sent

חיבורים צולבים בשאילתות מרובות שלבים

כשמשתמשים בחיפוש או בלוחות בקרה של Google SecOps, הצטרפות צולבת בשאילתות מרובות שלבים מאפשרת להשוות בין נתוני אירועים ספציפיים של UDM לבין נתונים סטטיסטיים מצטברים שחושבו בשלבים אחרים של YARA-L.

ב-YARA-L, מילת המפתח cross join פועלת עם שלב שמחזיר רק שורה אחת.

כשמשתמשים ב-cross join בין שלב עם מגבלה של 1 לבין מערך נתונים אחר (לדוגמה, אירועים של UDM), השורה היחידה שמופקת מהשלב מצורפת לכל שורה במערך הנתונים השני. כך נתוני האירועים מתעשרים בסטטיסטיקות הכוללות.

דוגמה: איתור פעילות התחברות חריגה

בדוגמה הבאה מזהים את המשתמשים שנכנסים לחשבון בתדירות גבוהה מהרגיל. החישוב מתבצע על ידי השוואה בין מספר ההתחברויות של כל משתמש (בשלב user_login_counts) לבין מספר ההתחברויות הממוצע של כל המשתמשים (בשלב total_users). אפשר למיין בתוצאות החיפוש משתמשים שנכנסו לחשבון מספר פעמים לא רגיל.

לאחר מכן משתמשים במילת המפתח cross join כדי לקשר בין התוצאות משלב total_users לבין התוצאות משלב user_login_counts.

stage user_login_counts {
    $user = principal.user.userid
    metadata.event_type = "USER_LOGIN"
    security_result.action = "ALLOW"

    match:
        $user

    outcome:
        $login_count = count(metadata.id)
}

stage total_users {
    outcome:
        $count = count($user_login_counts.user)
    limit:
        1
}

cross join $total_users, $user_login_counts

$login_count = $user_login_counts.login_count
$user = $user_login_counts.user
$tot_users = $total_users.count

// all users who logged in the same number of times are grouped together.
match:
    $login_count
outcome:
    $num_users = count($user)
    $frequency_percent = (count($user) / max($tot_users) ) * 100

מגבלות

יש אילוצים פונקציונליים ומבניים על שאילתות מרובות שלבים:

דרישות מבניות

כשיוצרים שאילתות, צריך לפעול לפי דרישות המבנה האלה:

  • שלב בסיסי: מותר להשתמש רק בשלב בסיסי אחד לכל שאילתה.
  • שלבים עם שם: אפשר להוסיף עד ארבעה שלבים עם שם.
  • הפניה לשלב: שלב יכול להפנות רק לשלבים שהוגדרו לפניו באופן לוגי באותה שאילתה.
  • שאילתות איחוד (cross join): שאילתת איחוד (cross join) יכולה להפנות רק לשלב שמחזיר שורה אחת. כדי לעמוד בדרישה הזו, צריך לכלול מגבלה (1) בשלב שאליו מתייחסים. היתרון בכך הוא שאפשר לצרף נתון סטטיסטי גלובלי יחיד (כמו מקסימום או ממוצע) לכל שורה של אירוע בנפרד לצורך השוואה.
  • צירופים: מותרים עד ארבעה צירופים שאינם של טבלאות נתונים בכל השלבים.
  • דרישת תוצאה: כל שלב עם שם (לא כולל שלב הבסיס) חייב לכלול מקטע match או מקטע outcome שלא דורש צבירה.outcome

מגבלות של חלונות ותאימות

ההגבלות הבאות חלות על השימוש בחלונות ועל המקומות שבהם מריצים שאילתות:

  • תמיכה בתכונות: שאילתות מרובות שלבים פועלות בחיפוש ובמרכזי בקרה, אבל התכונה לא נתמכת בכללים.
  • סוגי חלונות: מומלץ להימנע משילוב של סוגי חלונות שונים בשאילתה אחת.
  • תלות בחלון: שלב שמשתמש בחלון קופץ או בחלון הזזה לא יכול להיות תלוי בשלב אחר שגם משתמש בחלון קופץ או בחלון הזזה.
  • גודל חלון עוקב (tumbling window): למרות שגודל החלונות העוקבים (tumbling window) בשלבים שונים יכול להיות שונה, ההבדל בגודל חייב להיות קטן מ-720x.

דוגמה: הפרש צבירת הנתונים בשלב (לא חוקי)

ההגדרה הבאה לא חוקית כי בחודש יש 44,640 דקות (44,640 / 1 > 720):

שלב: monthly_stats { ... match: by month }

Root: match: by minute

דוגמה: הפרש צבירת הנתונים בשלב (תקין)

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

שלב: daily_stats { ... match: by day }

Root: match: by hour

מכיוון ש-24 (מספר השעות ביממה) קטן מ-720, המערכת יכולה למפות ביעילות את נתוני השלב לשלב הבסיסי.

מגבלות על שלבים ועל שאילתות

לכל שלב בנפרד בשאילתה מרובת שלבים יש אילוצים ספציפיים. רוב המגבלות שחלות על שאילתה חד-שלבית חלות גם על כל שלב בנפרד:

  • דרישת פלט: כל שלב חייב להפיק לפחות משתנה אחד של התאמה או תוצאה (שדה שלב).

  • טווח הזמן של השאילתה:

    • שאילתות רגילות: המקסימום הוא 30 ימים.
    • שאילתות מרובות שלבים עם הצטרפויות ללא התאמה: המקסימום הוא 14 ימים.

  • מגבלות על גודל החלון: גודל החלון המקסימלי (דילוג, הזזה או גלילה) תלוי בשאילתה שלכם, כלומר אם היא כוללת הצטרפות:

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

    • ללא הצטרפות (אירוע יחיד):

    • חלונות קופצים וחלונות נעים: המקסימום הוא יומיים.

    • חלונות מתגלגלים: מספר הימים המקסימלי עולה ל-30.

  • משתני התוצאה המקסימליים:

    • 20 כברירת מחדל
    • ‫50 ללקוחות שהביעו הסכמה למגבלה הגדולה יותר
    • מגבלות על מערכים: במשתנה תוצאה עם ערך מערך מותרים עד 10,000 רכיבים.

  • מגבלות על אירועים לכל שאילתה:

    • מקסימום שני אירועים של UDM
    • מקסימום אירוע אק"ג אחד
    • מקסימום שתי טבלאות נתונים

מגבלות על השירות והביצועים

שאילתות מרובות שלבים כפופות לאותן מגבלות כמו שאילתות סטטיסטיות:

  • שאילתות סטטיסטיות: 120 QPH (API וממשק משתמש).
  • צפיות בחיפוש: 100 צפיות בדקה.
  • תמיכה ב-API: מערכת Google SecOps ו-EventService.UDMSearch API תומכים בצירופים מרובי שלבים, אבל SearchService.UDMSearch API לא תומך בהם. המערכת מאפשרת גם להריץ שאילתות רב-שלביות בלי פעולות איחוד.

הגבלות על אירועים ועל הגדרות גלובליות

אתם צריכים לפעול בהתאם למגבלות האלה ברמת האירוע וברמת הפלטפורמה:

מקסימום אירועים

בשאילתות מרובות שלבים יש הגבלה מחמירה על מספר האירועים שאפשר לעבד בו-זמנית:

  • אירועי UDM: מותר להשתמש ב-2 אירועי UDM לכל היותר.
  • אירועים של גרף הקשר של ישות (ECG): מותר להשתמש באירוע ECG אחד לכל היותר.

מגבלות על שאילתות עם אחזור נתונים גלובלי

המגבלות האלה חלות על כל הפלטפורמה וקובעות עד כמה אחורה בזמן אפשר להסתכל ומה כמות הנתונים ששאילתה מרובת שלבים יכולה להחזיר:

  • טווח הזמן של השאילתה: טווח הזמן המקסימלי לשאילתה רגילה הוא 30 ימים.
  • קבוצת התוצאות הכוללת: הגודל המקסימלי של קבוצת התוצאות הכוללת הוא 10,000 תוצאות.

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