שילוב של Apache Kafka עם Google SecOps

במאמר הזה מוסבר איך לשלב את Apache Kafka עם Google Security Operations ‏ (Google SecOps).

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

האינטגרציה עם Apache Kafka יכולה לתת מענה לתרחישי השימוש הבאים:

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

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

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

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

לפני שמגדירים את השילוב של Apache Kafka ב-Google SecOps, צריך לוודא שמתקיימות הדרישות המוקדמות הבאות:

  • שרת Apache Kafka: צריך לוודא שיש לכם גישה לשרת Apache Kafka פעיל עם ברוקרים ונושאים של Kafka שהוגדרו כנדרש.

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

    us-docker.pkg.dev/siem-ar-public/images/agent-debian:latest

פרמטרים של שילוב

השילוב של Apache Kafka דורש את הפרמטרים הבאים:

פרמטר תיאור
Kafka brokers

חובה.

רשימה מופרדת בפסיקים של ברוקרים של Kafka להתחברות, בפורמט hostname:port.
Use TLS for connection

זה שינוי אופציונלי.

אם האפשרות הזו נבחרת, השילוב משתמש בהצפנת TLS לאימות.

הפרמטר הזה דורש אישור של רשות אישורים (CA).

ההגדרה לא מופעלת כברירת מחדל.
Use SASL PLAIN with TLS for connection

זה שינוי אופציונלי.

אם האפשרות הזו נבחרת, השילוב משתמש במנגנון של שם משתמש וסיסמה ב-SASL PLAIN לצורך אימות.

האפשרות הזו נתמכת רק בהצפנת TLS, ונדרשים בה שם משתמש וסיסמה של SASL ואישור CA.

ההגדרה לא מופעלת כברירת מחדל.
CA certificate of Kafka server

זה שינוי אופציונלי.

אישור ה-CA שמשמש לאימות הזהות של שרת Kafka.

חובה לכלול את הפרמטר הזה אם SASL מופעל.
Client certificate

זה שינוי אופציונלי.

האישור של הלקוח לאימות TLS בו-זמני (mTLS) עם שרת Kafka.

הפרמטר הזה נדרש אם מופעל פרוטוקול TLS הדדי (mTLS).
Client certificate key

זה שינוי אופציונלי.

המפתח הפרטי שמתאים לאישור של הלקוח, שמשמש לאימות הדדי של TLS.

הפרמטר הזה נדרש אם מופעל פרוטוקול TLS הדדי (mTLS).
Client certificate key password

זה שינוי אופציונלי.

הסיסמה שמשמשת לפענוח המפתח הפרטי של אישור הלקוח.

הפרמטר הזה נדרש אם מופעל פרוטוקול TLS הדדי (mTLS).
SASL PLAIN Username

זה שינוי אופציונלי.

שם המשתמש לאימות SASL PLAIN עם ברוקרים של Kafka.

חובה לכלול את הפרמטר הזה אם SASL מופעל.
SASL PLAIN Password

זה שינוי אופציונלי.

הסיסמה לאימות SASL PLAIN עם ברוקרי Kafka.

חובה לכלול את הפרמטר הזה אם SASL מופעל.

הוראות להגדרת שילוב ב-Google SecOps מופיעות במאמר הגדרת שילובים.

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

פעולות

מידע נוסף על פעולות זמין במאמרים איך מגיבים לפעולות שממתינות לאישור ב-Your Workdesk ואיך מבצעים פעולה ידנית.

פינג

משתמשים בפעולה Ping כדי לבדוק את הקישוריות ל-Apache Kafka.

הפעולה הזו לא מופעלת על ישויות ב-Google SecOps.

הוספת נתונים לפעולה

אין.

פלט של פעולות

הפעולה Ping מספקת את הפלטים הבאים:

סוג הפלט של הפעולה זמינות
קיר המארז לא זמין
קישור לקיר של הפנייה לא זמין
טבלת קיר עם תושבת לא זמין
טבלת העשרה לא זמין
תוצאה בפורמט JSON לא זמין
הודעות פלט יש גישה
תוצאת הסקריפט. יש גישה
הודעות פלט

הפעולה Ping יכולה להחזיר את הודעות הפלט הבאות:

הודעת פלט תיאור ההודעה

Successfully connected to the Apache Kafka server with the provided connection parameters!

 הפעולה הסתיימה בהצלחה.
Failed to connect to the Apache Kafka server! Error is ERROR_REASON

הפעולה נכשלה.

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

בטבלה הבאה מפורט הערך של פלט תוצאת הסקריפט כשמשתמשים בפעולה Ping:

שם התוצאה של הסקריפט ערך
is_success True או False

מחברים

מידע נוסף על הגדרת מחברים ב-Google SecOps זמין במאמר הוספת נתונים (מחברים).

‫Apache Kafka - Messages Connector

משתמשים במחבר ההודעות של Apache Kafka כדי לאחזר הודעות מ-Apache Kafka.

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

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

מיפוי חומרת ה-JSON

כדי למפות את רמת החומרה של ההתראה, צריך לציין באיזה שדה Apache Kafka - Messages Connector משתמש כדי לקבל את הערך של רמת החומרה בפרמטר Severity Mapping JSON. התגובה של המחבר יכולה להכיל סוגי ערכים, כמו integer, float ו-string.

המחבר Apache Kafka - Messages קורא את הערכים integer ו-float וממפה אותם בהתאם להגדרות של Google SecOps. בטבלה הבאה מוצג המיפוי של הערכים integer לחומרה ב-Google SecOps:

ערך מספרי שלם מידת החומרה הממופה
100 Critical
גווני אפור מ-80 עד 100 High
גווני אפור מ-60 עד 80 Medium
גווני אפור מ-40 עד 60 Low
פחות מ-40 Informational

אם התשובה מכילה את הערך string, Pub/Sub – Messages Connector דורש הגדרה נוספת.

בתחילה, ערך ברירת המחדל יופיע כך:

{
    "Default": 60
}

אם הערכים שנדרשים למיפוי נמצאים במפתח ה-JSON‏ event_severity, הערכים יכולים להיות:

  • "Malicious"
  • "Benign"
  • "Unknown"

כדי לנתח את ערכי המפתח של JSON‏ event_severity ולוודא שאובייקט ה-JSON הוא בפורמט הנכון, מגדירים את הפרמטר Severity Mapping JSON באופן הבא:

{
    "event_severity": {
        "Malicious": 100,
        "Unknown": 60,
        "Benign": -1
    },
    "Default": 50
}

חובה לציין את הערך "Default".

במקרה שבו יש כמה התאמות לאותו אובייקט JSON, Apache Kafka - Messages Connector נותן עדיפות למפתח הראשון של אובייקט JSON.

כדי לעבוד עם שדות שמכילים ערכים של integer או float, צריך להגדיר את המפתח ואת המחרוזת הריקה בפרמטר Severity Mapping JSON:

{
  "Default":"60",
  "integer_field": "",
  "float_field": ""
}

כניסות קלט של מחברים

מחבר ההודעות של Apache Kafka דורש את הפרמטרים הבאים:

פרמטר תיאור
Product Field Name

חובה.

שם השדה שבו מאוחסן שם המוצר.

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

ערך ברירת המחדל הוא Product Name.
Event Field Name

חובה.

השם של השדה שקובע את שם האירוע (סוג משנה).

ערך ברירת המחדל הוא event_type.
Environment Field Name

זה שינוי אופציונלי.

שם השדה שבו מאוחסן שם הסביבה.

אם השדה environment חסר, המחבר משתמש בערך ברירת המחדל.

ערך ברירת המחדל הוא "".
Environment Regex Pattern

זה שינוי אופציונלי.

תבנית של ביטוי רגולרי להפעלה על הערך שנמצא בשדה Environment Field Name. הפרמטר הזה מאפשר לכם לשנות את השדה environment באמצעות לוגיקה של ביטוי רגולרי.

משתמשים בערך ברירת המחדל .* כדי לאחזר את הערך הנדרש Environment Field Name.

אם תבנית הביטוי הרגולרי היא null או ריקה, או אם ערך הסביבה הוא null, התוצאה הסופית של הסביבה היא סביבת ברירת המחדל.
Script Timeout (Seconds)

חובה.

מגבלת הזמן הקצוב לתהליך Python שמריץ את הסקריפט הנוכחי, בשניות.

ערך ברירת המחדל הוא 180.
Kafka brokers

חובה.

רשימה מופרדת בפסיקים של ברוקרים של Kafka להתחברות, בפורמט hostname:port.
Use TLS for connection

זה שינוי אופציונלי.

אם האפשרות הזו נבחרת, השילוב משתמש בהצפנת TLS לאימות.

הפרמטר הזה דורש אישור CA.

ההגדרה לא מופעלת כברירת מחדל.
Use SASL PLAIN with TLS for connection

זה שינוי אופציונלי.

אם האפשרות הזו נבחרת, השילוב משתמש במנגנון של שם משתמש וסיסמה ב-SASL PLAIN לצורך אימות.

כדי להשתמש באפשרות הזו, צריך לספק שם משתמש וסיסמה של SASL. הוא נתמך רק בהצפנת TLS, שדורשת אישור CA.

ההגדרה לא מופעלת כברירת מחדל.
CA certificate of Kafka server

זה שינוי אופציונלי.

אישור ה-CA שמשמש לאימות הזהות של שרת Kafka.
Client certificate

זה שינוי אופציונלי.

האישור של הלקוח לאימות TLS בו-זמני (mTLS) עם שרת Kafka.
Client certificate key

זה שינוי אופציונלי.

המפתח הפרטי שמתאים לאישור של הלקוח, שמשמש לאימות TLS הדדי.
Client certificate key password

זה שינוי אופציונלי.

הסיסמה שמשמשת לפענוח המפתח הפרטי של אישור הלקוח.
SASL PLAIN Username

זה שינוי אופציונלי.

שם המשתמש לאימות SASL PLAIN עם ברוקרים של Kafka.
SASL PLAIN Password

זה שינוי אופציונלי.

הסיסמה לאימות SASL PLAIN עם ברוקרי Kafka.
Topic

חובה.

נושא Kafka שממנו מאוחזרים האירועים.
Consumer Group ID

זה שינוי אופציונלי.

המזהה של קבוצת הצרכנים שמשמש לאחזור אירועים.

אם לא מספקים ערך, נוצר מזהה ייחודי.
Partitions

זה שינוי אופציונלי.

רשימת מחיצות בפורמט CSV שממנה יאוחזרו הודעות.
Initial Offset

זה שינוי אופציונלי.

המקום שבו המחבר מתחיל לאחזר הודעות ממחיצת Kafka.

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

זה שינוי אופציונלי.

זמן קצוב לתשאול כדי לצרוך הודעה מ-Kafka, בשניות.
Case Name Template

זה שינוי אופציונלי.

תבנית להגדרת שם מותאם אישית של בקשה. המחבר מוסיף מפתח custom_case_name לאירוע.

אפשר להשתמש ב-placeholders בפורמט FIELD_NAME, שמאוכלסים מערכי המחרוזות של האירוע הראשון.

דוגמה: Phishing - EVENT_MAILBOX
Alert Name Template

חובה.

תבנית להגדרת שם ההתראה.

אפשר להשתמש ב-placeholders בפורמט FIELD_NAME, שמאוכלסים מערכי המחרוזות של האירוע הראשון.

דוגמה: Phishing - EVENT_MAILBOX

אם לא מספקים ערך או שהתבנית לא תקינה, המחבר משתמש בשם ברירת מחדל להתראה.
Rule Generator Template

חובה.

תבנית להגדרת מחולל הכללים.

אפשר להשתמש ב-placeholders בפורמט FIELD_NAME, שמאוכלסים מערכי המחרוזות של האירוע הראשון.

דוגמה: Phishing - EVENT_MAILBOX

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

חובה.

שם השדה בהודעת Kafka שמכיל את חותמת הזמן של ההתראה ב-Google SecOps.

אם חותמת הזמן לא בפורמט Unix epoch, צריך להגדיר את הפורמט שלה בפרמטר Timestamp Format.
Timestamp Format

זה שינוי אופציונלי.

הפורמט של חותמת הזמן של ההודעה, נדרש לחותמות זמן שאינן ראשית זמן יוניקס (Unix epoch). משתמשים בקודי פורמט סטנדרטיים של Python strftime.

אם חותמת הזמן לא בפורמט של ראשית זמן יוניקס (Unix epoch) והפרמטר הזה לא מוגדר, המחבר נכשל.
Severity Mapping JSON

חובה.

אובייקט ה-JSON שבו המחבר משתמש כדי לחלץ ולמפות את רמת החומרה מההודעה לסולם העדיפות של Google SecOps.

ערך ברירת המחדל הוא {"Default": "60"}.
Unique ID Field

זה שינוי אופציונלי.

השם של השדה שמשמש כמזהה ייחודי של ההודעה.

אם לא מציינים ערך, המחבר יוצר גיבוב (hash) SHA-256 של תוכן ההודעה ומשתמש בו כמזהה ההודעה.
Max Messages To Fetch

חובה.

המספר המקסימלי של הודעות שהמחבר מעבד בכל איטרציה.

ערך ברירת המחדל הוא 100.
Disable Overflow

זה שינוי אופציונלי.

אם בוחרים באפשרות הזו, המחבר מתעלם ממנגנון הגלישה של Google SecOps.

מופעל כברירת מחדל.
Verify SSL

חובה.

אם האפשרות הזו נבחרה, השילוב מאמת את אישור ה-SSL כשמתחברים לשרת Apache Kafka.

מופעל כברירת מחדל.
Proxy Server Address

זה שינוי אופציונלי.

הכתובת של שרת ה-Proxy שבו רוצים להשתמש.
Proxy Username

זה שינוי אופציונלי.

שם המשתמש ב-proxy לאימות.
Proxy Password

זה שינוי אופציונלי.

סיסמת ה-proxy לאימות.

כללים של Connectors

המחבר תומך בשרתי proxy.

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

בטבלה הבאה מתואר המיפוי של שדות ההודעות ב-Apache Kafka לשדות ההתראות ב-Google SecOps:

שדה התראה של Siemplify שדה הודעה של Apache Kafka
SourceSystemName הערך מאוכלס על ידי המסגרת.
TicketId הערך של שדה המזהה הייחודי או גיבוב (hash) מסוג SHA-256 של ההודעה.
DisplayId ApacheKafka_{unique id or hash}_{connector identifier}
Name הערך שנוצר על ידי Alert Name Template.
Reason לא רלוונטי
Description לא רלוונטי
DeviceVendor קוד קשיח: Apache Kafka
DeviceProduct ערך ברירת מחדל: Message
Priority ממופה מהפרמטר Severity Mapping JSON.
RuleGenerator הערך שנוצר על ידי Rule Generator Template.
SourceGroupingIdentifier לא רלוונטי
StartTime הומר מ-Timestamp Field.
EndTime הומר מ-Timestamp Field.
Siemplify Alert - Extensions לא רלוונטי
Siemplify Alert - Attachments לא רלוונטי

אירועים של מחברים

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

{
  "notificationConfigName": "organizations/ORGANIZATION_ID/notificationConfigs/soar_connector_CONNECTOR_ID_toxic_notifications_config",
  "finding": {
    "name": "organizations/ORGANIZATION_ID/sources/SOURCE_ID/findings/FINDING_ID",
    "parent": "organizations/ORGANIZATION_ID/sources/SOURCE_ID",
    "resourceName": "//compute.googleapis.com/projects/PROJECT_ID/global/firewalls/FIREWALL_ID",
    "state": "ACTIVE",
    "category": "OPEN_NETBIOS_PORT",
    "externalUri": "https://console.cloud.google.com/networking/firewalls/details/default-allow-rdp?project=PROJECT_ID",
    "sourceProperties": {
      "Recommendation": "Restrict the firewall rules at: https://console.cloud.google.com/networking/firewalls/details/default-allow-rdp?project=PROJECT_ID",
      "ExceptionInstructions": "Add the security mark \"allow_open_netbios_port\" to the asset with a value of \"true\" to prevent this finding from being activated again.",
      "Explanation": "Firewall rules that allow connections from all IP addresses on TCP ports 137-139 or UDP ports 137-139 may expose NetBIOS services to attackers.",
      "ScannerName": "FIREWALL_SCANNER",
      "ResourcePath": [
        "projects/PROJECT_ID/",
        "folders/FOLDER_ID_1/",
        "folders/FOLDER_ID_2/",
        "organizations/ORGANIZATION_ID/"
      ],
      "ExposedService": "NetBIOS",
      "OpenPorts": {
        "TCP": [
          137.0,
          138.0,
          139.0
        ],
        "UDP": [
          137.0,
          138.0,
          139.0
        ]
      },
      "compliance_standards": {
        "iso": [
          {
            "ids": [
              "A.13.1.1"
            ]
          }
        ],
        "pci": [
          {
            "ids": [
              "1.2.1"
            ]
          }
        ],
        "nist": [
          {
            "ids": [
              "SC-7"
            ]
          }
        ]
      },
      "ReactivationCount": 4.0
    },
    "securityMarks": {
      "name": "organizations/ORGANIZATION_ID/sources/SOURCE_ID/findings/FINDING_ID/securityMarks",
      "marks": {
        "USER_ID": "SECURITY_MARK"
      }
    },
    "eventTime": "2024-08-30T14:44:37.973090Z",
    "createTime": "2024-06-24T07:08:54.777Z",
    "propertyDataTypes": {
      "ResourcePath": {
        "listValues": {
          "propertyDataTypes": [
            {
              "primitiveDataType": "STRING"
            }
          ]
        }
      },
      "ReactivationCount": {
        "primitiveDataType": "NUMBER"
      },
      "Explanation": {
        "primitiveDataType": "STRING"
      },
      "ExposedService": {
        "primitiveDataType": "STRING"
      },
      "ScannerName": {
        "primitiveDataType": "STRING"
      }
    }
  }
}

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