איסוף יומנים של Apache Hadoop

גרסת המנתח: 3.0

נתמך ב:

במאמר הזה מוסבר איך להטמיע יומני Apache Hadoop ב-Google Security Operations באמצעות סוכן Bindplane. המנתח קודם מחלץ שדות מיומני Hadoop גולמיים באמצעות תבניות Grok שמבוססות על פורמטים נפוצים של יומני Hadoop. לאחר מכן, הוא ממפה את השדות שחולצו לשדות המתאימים בסכימה של Unified Data Model ‏(UDM), מבצע המרות של סוגי נתונים ומעשיר את הנתונים בהקשר נוסף.

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

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

  • מכונה של Google SecOps
  • מארח עם Windows 2016 ואילך או Linux עם systemd
  • אם אתם משתמשים בשרת proxy, ודאו שהיציאות של חומת האש פתוחות בהתאם לדרישות של סוכן Bindplane.
  • גישה מורשית לקובצי התצורה של אשכול Apache Hadoop

קבלת פרטי כניסה ל-Google SecOps בשביל סוכן Bindplane

כדי לאמת את סוכן Bindplane ב-Google SecOps, בוחרים באחת מהשיטות הבאות:

  • פריסות חדשות: משתמשים ב-Chronicle API (אפשרות 1).
  • פריסות קודמות: ממשיכים להשתמש בקובץ האימות להטמעת נתונים (אפשרות 2) עד שה-Ingestion API יופסק ב-20 ביולי 2027.
  1. נכנסים למסוףGoogle Cloud ובוחרים את הפרויקט ב- Google Cloud שמקושר למכונה של Google SecOps.
  2. נכנסים לדף IAM & Admin > Service Accounts.
  3. לוחצים על Create service account.
  4. מזינים את פרטי ההגדרה הבאים:
    • Service account name: מזינים שם לחשבון (לדוגמה, bindplane-ingest).
    • Service account description: מזינים תיאור (לדוגמה, Bindplane agent log ingestion).
  5. לוחצים על Create and continue.
  6. בקטע Grant this service account access to project, לוחצים על Select a role ובוחרים באפשרות Chronicle Editor.
  7. לוחצים על המשך ואז על סיום.
  8. לוחצים על חשבון השירות שיצרתם ועוברים לכרטיסייה Keys.
  9. לוחצים על Add key > Create new key.
  10. בוחרים באפשרות JSON ולוחצים על Create. קובץ המפתח יורד למחשב.
  11. שומרים את קובץ המפתח בתור ingestion-auth.json בספריית Bindplane collector (/opt/observiq-otel-collector/ingestion-auth.json ב-Linux או C:\Program Files\observIQ OpenTelemetry Collector\ingestion-auth.json ב-Windows), או מעדכנים את creds_file_path ב-config.yaml כך שיתאים לנתיב שבו שמרתם אותו.
  12. עוברים אל IAM & Admin > Settings ומעתיקים את הערך של Project number. תצטרכו אותו כדי להגדיר את הייצוא.

אפשרות 2: הורדה של קובץ האימות להטמעת נתונים (גרסה קודמת)

  1. נכנסים למסוף Google SecOps.
  2. עוברים אל הגדרות > ניהול נתונים > סוכני איסוף. אם אתם משתמשים בפלטפורמת SIEM בלבד, עוברים אל הגדרות > סוכני איסוף.
  3. מורידים את קובץ האימות להטמעת נתונים.
  4. שומרים את הקובץ בצורה מאובטחת במערכת שבה יותקן הסוכן של Bindplane.

איך מקבלים את מספר הלקוח ב-Google SecOps

  1. נכנסים למסוף Google SecOps.
  2. עוברים אל הגדרות > ניהול > פרופיל. אם אתם משתמשים בפלטפורמה של SIEM בלבד, עוברים אל הגדרות > פרופיל.
  3. מעתיקים את מספר הלקוח מהקטע פרטי הארגון ושומרים אותו.

התקנת הסוכן של Bindplane

מתקינים את הסוכן של Bindplane במערכת ההפעלה Windows או Linux לפי ההוראות הבאות.

התקנת חלונות

  1. פותחים את שורת הפקודה או את PowerShell כאדמין.

  2. מריצים את הפקודה הבאה:

    msiexec /i "https://github.com/observIQ/bindplane-agent/releases/latest/download/observiq-otel-collector.msi" /quiet

התקנה ב-Linux

  1. פותחים טרמינל עם הרשאות root או sudo.

  2. מריצים את הפקודה הבאה:

    sudo sh -c "$(curl -fsSlL https://github.com/observiq/bindplane-agent/releases/latest/download/install_unix.sh)" install_unix.sh

מקורות מידע נוספים בנושא התקנה

לאפשרויות התקנה נוספות, קראו את מדריך ההתקנה של סוכן Bindplane.

הגדרת סוכן Bindplane להטמעת Syslog ושליחה אל Google SecOps

  1. ניגשים לקובץ התצורה:

    • מאתרים את הקובץ config.yaml. בדרך כלל, הוא נמצא בספרייה /opt/observiq-otel-collector/ ב-Linux או בספריית ההתקנה ב-Windows.
    • פותחים את הקובץ באמצעות כלי לעריכת טקסט (לדוגמה, nano,‏ vi או פנקס הרשימות).

  2. עורכים את הקובץ config.yaml:

    receivers:
  udplog:
    # Replace the port and IP address as required
    listen_address: "0.0.0.0:514"

exporters:
  chronicle/chronicle_w_labels:
    protocol: https
    # Region of your instance, for example us, europe, asia-southeast1
    location: <LOCATION>
    # Project number of the Google Cloud project linked to the instance
    project: '<PROJECT_NUMBER>'
    compression: gzip
    # Adjust the path to the credentials file you downloaded in Step 1
    creds_file_path: '/path/to/ingestion-authentication-file.json'
    # Replace with your actual customer ID from Step 2
    customer_id: <CUSTOMER_ID>
    endpoint: chronicle.googleapis.com
    # Add optional ingestion labels for better organization
    log_type: 'HADOOP'
    raw_log_field: body
    ingestion_labels:

service:
  pipelines:
    logs/source0__chronicle_w_labels-0:
      receivers:
        - udplog
      exporters:
        - chronicle/chronicle_w_labels
    • מחליפים את היציאה ואת כתובת ה-IP לפי הצורך בתשתית שלכם.
    • מחליפים את <LOCATION> באזור של המופע של Google SecOps (לדוגמה, us,‏ europe או asia-southeast1). המערכת מוסיפה את הערך הזה בתחילת chronicle.googleapis.com.
    • מחליפים את <PROJECT_NUMBER> במספר הפרויקט שמופיע בקטע IAM & Admin > Settings.
    • מחליפים את <CUSTOMER_ID> במספר הלקוח בפועל.
    • מעדכנים את /path/to/ingestion-authentication-file.json לנתיב הקובץ שבו נשמר קובץ המפתח של חשבון השירות (אפשרות 1) בשלב 1.

מייצא API להטמעה מדור קודם

  • אם אתם משתמשים בקובץ האימות להטמעת נתונים מאפשרות 2, החליפו את הבלוק exporters בקוד הבא. מגדירים את creds_file_path לנתיב של קובץ האימות להטמעת נתונים. אם המכונה נמצאת מחוץ לארה"ב, מחליפים את malachiteingestion-pa.googleapis.com בנקודת הקצה האזורית (לדוגמה, europe-malachiteingestion-pa.googleapis.com).

    exporters:
  chronicle/chronicle_w_labels:
    compression: gzip
    # Adjust the path to the credentials file you downloaded in Step 1
    creds_file_path: '/path/to/ingestion-authentication-file.json'
    # Replace with your actual customer ID from Step 2
    customer_id: <CUSTOMER_ID>
    endpoint: malachiteingestion-pa.googleapis.com
    # Add optional ingestion labels for better organization
    log_type: 'HADOOP'
    raw_log_field: body
    ingestion_labels:

מפעילים מחדש את הסוכן של Bindplane כדי להחיל את השינויים.

  1. כדי להפעיל מחדש את הסוכן של Bindplane ב-Linux, מריצים את הפקודה הבאה:

    sudo systemctl restart observiq-otel-collector

  2. כדי להפעיל מחדש את הסוכן של Bindplane ב-Windows, אפשר להשתמש במסוף Services או להזין את הפקודה הבאה:

    net stop observiq-otel-collector && net start observiq-otel-collector

הגדרת העברת Syslog ב-Apache Hadoop

‫Apache Hadoop משתמש ב-Log4j לרישום ביומן. מגדירים את Syslog appender המתאים לפי גרסת Log4j, כדי שדמונים של Hadoop (כמו NameNode, ‏ DataNode, ‏ ResourceManager, ‏ NodeManager וכו') יעבירו את היומנים ישירות למקלט ה-syslog (מארח Bindplane). ההגדרה של Log4j מתבצעת באמצעות קבצים (אין ממשק משתמש באינטרנט).

אפשרות 1: הגדרה של Log4j 1.x

  1. מאתרים את הקובץ log4j.properties (בדרך כלל ב-$HADOOP_CONF_DIR/log4j.properties).

  2. מוסיפים לקובץ את ההגדרה הבאה של SyslogAppender:

    # Syslog appender (UDP example)
log4j.appender.SYSLOG=org.apache.log4j.net.SyslogAppender
log4j.appender.SYSLOG.SyslogHost=<BINDPLANE_HOST_IP>:514
log4j.appender.SYSLOG.Facility=LOCAL0
log4j.appender.SYSLOG.FacilityPrinting=true
log4j.appender.SYSLOG.layout=org.apache.log4j.PatternLayout
log4j.appender.SYSLOG.layout.ConversionPattern=%d{ISO8601} level=%p logger=%c thread=%t msg=%m%n

# Example: send NameNode logs to syslog
log4j.logger.org.apache.hadoop.hdfs.server.namenode=INFO,SYSLOG
log4j.additivity.org.apache.hadoop.hdfs.server.namenode=false

# Or attach to root logger to send all Hadoop logs
# log4j.rootLogger=INFO, SYSLOG

  3. מחליפים את <BINDPLANE_HOST_IP> בכתובת ה-IP של מארח Bindplane.

  4. שומרים את הקובץ.

  5. מפעילים מחדש את הדימונים של Hadoop כדי להחיל את השינויים בהגדרות.

אפשרות 2: הגדרת Log4j 2.x

  1. מאתרים את הקובץ log4j2.xml (בדרך כלל ב-$HADOOP_CONF_DIR/log4j2.xml).

  2. מוסיפים לקובץ את ההגדרות הבאות של Syslog appender:

    <Configuration status="WARN">
  <Appenders>
    <!-- UDP example; for TCP use protocol="TCP" -->
    <Syslog name="SYSLOG" format="RFC 5424"
            host="<BINDPLANE_HOST_IP>"
            port="514"
            protocol="UDP"
            facility="LOCAL0"
            appName="hadoop"
            enterpriseNumber="18060"
            mdcId="mdc">
      <PatternLayout pattern="%d{ISO8601} level=%p logger=%c thread=%t msg=%m %X%n"/>
    </Syslog>
  </Appenders>

  <Loggers>
    <!-- Send NameNode logs to syslog -->
    <Logger name="org.apache.hadoop.hdfs.server.namenode" level="info" additivity="false">
      <AppenderRef ref="SYSLOG"/>
    </Logger>

    <!-- Or send all Hadoop logs -->
    <Root level="info">
      <AppenderRef ref="SYSLOG"/>
    </Root>
  </Loggers>
</Configuration>

  3. מחליפים את <BINDPLANE_HOST_IP> בכתובת ה-IP של מארח Bindplane.

  4. שומרים את הקובץ.

  5. מפעילים מחדש את הדימונים של Hadoop כדי להחיל את השינויים בהגדרות.

טבלת מיפוי UDM

שדה ביומן מיפוי UDM Logic (פונקציה לוגית)
מותר security_result.action אם הערך הוא "false", הפעולה היא "BLOCK". אם הערך הוא "true", הפעולה היא "ALLOW".
auth_type additional.fields.key = "auth_type", additional.fields.value.string_value חולץ מהשדה "ugi" באמצעות תבנית grok‏ "%{DATA:suser}@.*auth:%{WORD:auth_type}". הסוגריים והמילה "auth:" יוסרו.
שיחה additional.fields.key = "Call#", additional.fields.value.string_value מיפוי ישיר.
call_context additional.fields.key = "callerContext", additional.fields.value.string_value מיפוי ישיר.
cliIP principal.ip ממופה רק כשהשדה json_data קיים ומנותח בהצלחה כ-JSON.
cmd principal.process.command_line מיפוי ישיר.
cluster_name target.hostname אם הוא קיים, הוא משמש כשם המארח של היעד.
יום metadata.event_timestamp.seconds משמש עם חודש, שנה, שעות, דקות ושניות כדי ליצור את event_timestamp.
description metadata.description מיפוי ישיר.
נהג/ת additional.fields.key = "driver", additional.fields.value.string_value מיפוי ישיר.
dst target.ip או target.hostname או target.file.full_path אם הניתוח ככתובת IP הצליח, המערכת ממפה את הערך לכתובת ה-IP המטורגטת. אם הערך מתחיל ב-"/user", הוא ממופה לנתיב קובץ היעד. אחרת, ממופה לשם המארח של היעד.
dstport target.port ממופה ישירות ומומר למספר שלם.
enforcer security_result.rule_name מיפוי ישיר.
event_count additional.fields.key = "event_count", additional.fields.value.string_value ממופה ישירות ומומר למחרוזת.
fname src.file.full_path מיפוי ישיר.
שעות metadata.event_timestamp.seconds משמש עם חודש, יום, שנה, דקות ושניות כדי ליצור את event_timestamp.
id thought additional.fields.key = "id", additional.fields.value.string_value מיפוי ישיר.
ip principal.ip ממופה לכתובת ה-IP הראשית אחרי הסרת התו "/" בתחילת הכתובת.
json_data מנותח כ-JSON. השדות שחולצו ממופים לשדות UDM תואמים.
logType additional.fields.key = "logType", additional.fields.value.string_value מיפוי ישיר.
הודעה משמש לחילוץ שדות שונים באמצעות תבניות grok.
method network.http.method מיפוי ישיר.
דקות metadata.event_timestamp.seconds משמש עם חודש, יום, שנה, שעות ושניות כדי ליצור את event_timestamp.
month metadata.event_timestamp.seconds משמש עם יום, שנה, שעות, דקות ושניות כדי ליצור את event_timestamp.
הישות שזיהתה את האירוע observer.hostname או observer.ip אם הניתוח ככתובת IP הצליח, המערכת ממפה את הכתובת לכתובת ה-IP של הישות שזיהתה את האירוע. אחרת, ממופה לשם המארח של הצופה.
פרמננט additional.fields.key = "perm", additional.fields.value.string_value מיפוי ישיר.
policy security_result.rule_id ממופה ישירות ומומר למחרוזת.
מוצר metadata.product_name מיפוי ישיר.
product_event metadata.product_event_type מיפוי ישיר. אם הפעולה היא "rename", השדה "dst" ממופה ל-"target.file.full_path".
proto network.application_protocol ממופה ישירות ומומר לאותיות רישיות אם הוא לא "webhdfs".
reason security_result.summary מיפוי ישיר.
מאגר additional.fields.key = "repo", additional.fields.value.string_value מיפוי ישיר.
resType additional.fields.key = "resType", additional.fields.value.string_value מיפוי ישיר.
תוצאה additional.fields.key = "result", additional.fields.value.string_value ממופה ישירות ומומר למחרוזת.
ניסיון חוזר additional.fields.key = "Retry#", additional.fields.value.string_value מיפוי ישיר.
seconds metadata.event_timestamp.seconds משמש עם חודש, יום, שנה, שעות ודקות כדי ליצור את event_timestamp.
seq_num additional.fields.key = "seq_num", additional.fields.value.string_value ממופה ישירות ומומר למחרוזת.
severity security_result.severity ממופה לרמות חומרה שונות על סמך הערך: "INFO",‏ "Info",‏ "info"‏ -> "INFORMATIONAL";‏ "Low",‏ "low",‏ "LOW"‏ -> "LOW";‏ "error",‏ "Error",‏ "WARN",‏ "Warn"‏ -> "MEDIUM";‏ "High",‏ "high",‏ "HIGH"‏ -> "HIGH";‏ "Critical",‏ "critical",‏ "CRITICAL"‏ -> "CRITICAL".
shost principal.hostname משמש כשם המארח הראשי אם הוא שונה מ-"src".
src principal.ip או principal.hostname או observer.ip אם הניתוח ככתובת IP הצליח, מתבצע מיפוי לכתובת ה-IP של הישות המורשית והישות שזיהתה את האירוע. אחרת, ממופה לשם המארח הראשי.
srcport principal.port ממופה ישירות ומומר למספר שלם.
סיכום security_result.summary מיפוי ישיר.
suser principal.user.userid מיפוי ישיר.
תגים additional.fields.key = "tags", additional.fields.value.string_value מיפוי ישיר.
שרשור additional.fields.key = "thread", additional.fields.value.string_value מיפוי ישיר.
טיפ target.ip מיפוי ישיר.
ugi target.hostname משמש כשם המארח של היעד אם השדה log_data לא מכיל את התו "·".
url target.url מיפוי ישיר.
ספק metadata.vendor_name מיפוי ישיר.
גרסה metadata.product_version מיפוי ישיר.
שנה metadata.event_timestamp.seconds משמש עם חודש, יום, שעות, דקות ושניות כדי ליצור את event_timestamp.
לא רלוונטי metadata.event_type מוגדר כברירת מחדל ל-"NETWORK_CONNECTION". הערך משתנה ל-"STATUS_UPDATE" אם לא זוהה יעד.
לא רלוונטי metadata.log_type מגדירים את הערך "HADOOP".
לא רלוונטי security_result.alert_state אם רמת החומרה היא HIGH או CRITICAL, הערך הוא ALERTING.
לא רלוונטי is_alert הערך מוגדר כ-"true" אם רמת החומרה היא "HIGH" או "CRITICAL".
לא רלוונטי is_significant הערך מוגדר כ-"true" אם רמת החומרה היא "HIGH" או "CRITICAL".

יומן שינויים

הצגת יומן השינויים של המנתח הזה

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