Google uses AI technology to translate content into your preferred language. AI translations can contain errors.

פיתוח המחבר

נתמך ב:

Google secops SOAR

במאמר הזה מוסבר איך ליצור מחבר חדש בסביבת פיתוח משולבת (IDE) על ידי יצירת שילוב וקישור מחבר אליו.

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

כדי ליצור שילוב של מחבר אימייל:

בחלון תגובה > IDE, לוחצים על הוספה כדי להוסיף פריט IDE חדש.
בוחרים באפשרות Integration (שילוב) ונותנים לשילוב שם Email Connector.
לוחצים על יצירה. השילוב מופיע בסרגל הצד עם סמל ברירת מחדל.
לוחצים על more_vert הגדרת שילוב בהתאמה אישית ומגדירים את ההגדרות הבאות:

תיאור
סמל
תלויות של Python
פרמטרים של שילוב

לוחצים על Save.

יצירת מחבר אימייל

כדי ליצור מחבר אימייל:

בחלון תגובה > IDE, לוחצים על הוספה כדי להוסיף פריט IDE חדש.
לוחצים על כפתור הבחירה Connector (מחבר) ונותנים למחבר שם My Email Connector.
ברשימה, בוחרים בשילוב Email Connector כדי לשייך את המחבר לשילוב.
לוחצים על יצירה.

הגדרת הפרמטרים של המחבר

אחרי שיוצרים את המחבר, מגדירים את הפרמטרים שלו:

מגדירים את הפרמטרים של המחבר בהתאם לטבלה:

פרמטר	תיאור	סוג
`Username`	חובה. שם המשתמש ב-IMAP. כתובת האימייל שממנה המחבר שולף את האימיילים אל פלטפורמת Google SecOps. ערך ברירת המחדל הוא `email@gmail.com`.	מחרוזת
`Password`	חובה. סיסמת Internet Message Access Protocol ‏ (IMAP) לכתובת האימייל שבה המחבר משתמש כדי למשוך אימיילים אל פלטפורמת Google SecOps.	סיסמה
`IMAP Port`	חובה. יציאת הרשת הספציפית שמשמשת את IMAP כדי לגשת לאימיילים משרת מרוחק. לדוגמה: `993` (ערך ברירת מחדל).	int
`IMAP Server Address`	חובה. שרת הדואר הנכנס של חשבון IMAP. לדוגמה, `imap.gmail.com` (ערך ברירת המחדל).	מחרוזת
`Folder to check for emails`	אופציונלי. הכלי שולף אימיילים רק מהתיקייה שצוינה. לדוגמה: `Inbox` (ערך ברירת מחדל).	תיבת דואר

מזינים את הפרטים בשדות הבאים:

שם שדה המוצר = device_product: קובע את ערך השדה הגולמי שיוקצה לשם המוצר בהתראה. מחפשים את השדה שקשור למוצר בקוד, שהוגדר כ-Mail (product).
event["device_product"] = PRODUCT #The PRODUCT constant is `"Mail"
‫Event Field Name = event_name: קובע את ערך השדה הגולמי שיוקצה לשדה של סוג האירוע. מאתרים את השדה שקשור לכך בקוד, שמוגדר כ-Suspicious email.
event["event_name"] = Suspicious email

עריכת מחבר האימייל

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

מעתיקים את הקוד הבא שנוצר עבור My Email Connector, מדביקים אותו בסביבת הפיתוח המשולבת ופועלים לפי ההוראות:

from SiemplifyConnectors import SiemplifyConnectorExecution
from SiemplifyConnectorsDataModel import AlertInfo
from SiemplifyUtils import output_handler, convert_datetime_to_unix_time, convert_string_to_datetime
import email, imaplib, sys, re

# CONSTANTS
CONNECTOR_NAME = "Mail"
VENDOR = "Mail"
PRODUCT = "Mail"
DEFAULT_PRIORITY = 60 # Default is Medium
RULE_GENERATOR_EXAMPLE = "Mail"
DEFAULT_FOLDER_TO_CHECK_INBOX = "inbox"
DEFAULT_MESSAGES_TO_READ_UNSEEN = "UNSEEN"
URLS_REGEX = r"(?i)\b(?:http(?:s)?:\/\/)?(?:www\.)?[a-zA-Z0-9:%_\+~#=][a-zA-Z0-9:%\._\+~#=]{1,255}\.[a-z]{2,6}\b(?:[-a-zA-Z0-9@:%_\+.~#?&//=]*)"


def create_alert(siemplify, alert_id, email_message_data, datetime_in_unix_time, created_event):
    """Returns an alert which is one event that contains one unread email message"""
    siemplify.LOGGER.info(f"Started processing Alert {alert_id}")

    create_event = None
    alert_info = AlertInfo()
    # Initializes alert_info
    alert_info.display_id = f"{alert_id}" # Each alert needs to have a unique id, otherwise it won't create a case with the same alert id.
    alert_info.ticket_id = f"{alert_id}" # In default, ticket_id = display_id. However, if for some reason the external alert id is different from the display_id, you can save the original external alert id in the "ticket_id" field. 
    alert_info.name = email_message_data['Subject']
    alert_info.rule_generator = RULE_GENERATOR_EXAMPLE # The name of the siem rule which causes the creation of the alert. 
    alert_info.start_time = datetime_in_unix_time # Time should be saved in UnixTime. You may use SiemplifyUtils.convert_datetime_to_unix_time, or SiemplifyUtils.convert_string_to_datetime 
    alert_info.end_time = datetime_in_unix_time # Time should be saved in UnixTime. You may use SiemplifyUtils.convert_datetime_to_unix_time, or SiemplifyUtils.convert_string_to_datetime
    alert_info.priority = 60 # Informative = -1,Low = 40,Medium = 60,High = 80,Critical = 100.
    alert_info.device_vendor = VENDOR # The field will be fetched from the Original Alert. If you build this alert manually, state the source vendor of the data. (ie: Microsoft, Mcafee) 
    alert_info.device_product = PRODUCT # The field will be fetched from the Original Alert. If you build this alert manually, state the source product of the data. (ie: ActiveDirectory, AntiVirus)

    siemplify.LOGGER.info(f"Events creating started for alert {alert_id}")
    try:
        if created_event is not None:
            alert_info.events.append(created_event)
            siemplify.LOGGER.info(f"Added Event {alert_id} to Alert {alert_id}")
    # Raise an exception if failed to process the event
    except Exception as e:
        siemplify.LOGGER.error(f"Failed to process event {alert_id}")
        siemplify.LOGGER.exception(e)
    return alert_info


def create_event(siemplify, alert_id, email_message_data, all_found_url_in_emails_body_list, datetime_in_unix_time):
    """Returns the digested data of a single unread email"""
    siemplify.LOGGER.info(f"--- Started processing Event: alert_id: {alert_id} | event_id: {alert_id}") 
    event = {} 
    event["StartTime"] = datetime_in_unix_time # Time should be saved in UnixTime. You may use SiemplifyUtils.convert_datetime_to_unix_time, or SiemplifyUtils.convert_string_to_datetime 
    event["EndTime"] = datetime_in_unix_time # Time should be saved in UnixTime. You may use SiemplifyUtils.convert_datetime_to_unix_time, or SiemplifyUtils.convert_string_to_datetime 
    event["event_name"] = "Suspicious email" 
    event["device_product"] = PRODUCT # ie: "device_product" is the field name that describes the product the event originated from. 
    event["Subject"] = email_message_data["Subject"] 
    event["SourceUserName"] = email_message_data["From"] 
    event["DestinationUserName"] = email_message_data["To"] 
    event["found_url"] = ",".join(all_found_url_in_emails_body_list) 
    siemplify.LOGGER.info(f"---Finished processing Event: alert_id: {alert_id} | event_id: {alert_id}")
    return event


def find_url_in_email_message_body(siemplify, email_messages_data_list):
    """Search for a url in the email body"""
    all_found_url_in_emails_body_list = []
    for message in email_messages_data_list:
        for part in message.walk():
            if part.get_content_maintype() == 'text\plain':
                continue
            email_message_body = part.get_payload()
            all_found_urls = re.findall(URLS_REGEX, str(email_message_body))
            for url in all_found_urls:
                if url not in all_found_url_in_emails_body_list:
                    all_found_url_in_emails_body_list.append(url)


def get_email_messages_data(imap_host, imap_port, username, password, folder_to_check):
    """Returns all unread email messages"""
    email_messages_data_list = []

    # Login to email using 'imap' module
    mail = imaplib.IMAP4_SSL(imap_host, imap_port)
    mail.login(username, password)

    # Determining the default email folder to pull emails from - 'inbox'
    if folder_to_check is None:
        folder_to_check = DEFAULT_FOLDER_TO_CHECK_INBOX

    # Selecting the email folder to pull the data from
    mail.select(folder_to_check)

    # Storing the email message data
    result, data = mail.search(None, DEFAULT_MESSAGES_TO_READ_UNSEEN)

    # If there are several emails collected in the cycle it will split each
    # email message into a separate item in the list chosen_mailbox_items_list
    if len(data) > 0:
        chosen_mailbox_items_list = data[0].split()
        # Iterating each email message and appending to emails_messages_data_list
        for item in chosen_mailbox_items_list:
            typ, email_data = mail.fetch(item, '(RFC 822)')
            # Decoding from binary string to string
            raw_email = email_data[0][1].decode("utf-8")
            # Turning the email data into an email object
            email_message = email.message_from_string(raw_email)
            # Appending the email message data to email_messages_data_list
            email_messages_data_list.append(email_message)
    return email_messages_data_list


@output_handler
def main(is_test_run):
    alerts = [] # The main output of each connector run that contains the alerts data 
    siemplify = SiemplifyConnectorExecution() # Siemplify main SDK wrapper
    siemplify.script_name = CONNECTOR_NAME

    # In case of running a test
    if (is_test_run):
        siemplify.LOGGER.info("This is an \"IDE Play Button\"\\\"Run Connector once\" test run")

    # Extracting the connector's Params
    username = siemplify.extract_connector_param(param_name="Username") 
    password = siemplify.extract_connector_param(param_name="Password") 
    imap_host = siemplify.extract_connector_param(param_name="IMAP Server Address")
    imap_port = siemplify.extract_connector_param(param_name="IMAP Port")
    folder_to_check = siemplify.extract_connector_param(param_name="Folder to check for emails")

    # Getting the digested email message data
    email_messages_data_list = get_email_messages_data(imap_host, imap_port, username, password, folder_to_check)
    # If the email_messages_data_list is not empty
    if len(email_messages_data_list) > 0:
        for message in email_messages_data_list:
            # Converting the email message datetime from string to unix time by SiemplifyUtils functions
            datetime_email_message = message['Date']
            string_to_datetime = convert_string_to_datetime(datetime_email_message)
            datetime_in_unix_time = convert_datetime_to_unix_time(string_to_datetime)
            found_urls_in_email_body = find_url_in_email_message_body(siemplify, email_messages_data_list)
            # Getting the unique id of each email message and removing the suffix '@mail.gmail.com' from the Message-ID, Each alert id can be ingested to the system only once.
            alert_id = message['Message-ID'].replace('@mail.gmail.com','') 
            # Creating the event by calling create_event() function 
            created_event = create_event(siemplify, alert_id, message, found_urls_in_email_body, datetime_in_unix_time) 
            # Creating the alert by calling create_alert() function 
            created_alert = create_alert(siemplify, alert_id, message, datetime_in_unix_time, created_event)
            # Checking that the created_alert is not None 
            if created_alert is not None: 
                alerts.append(created_alert) 
                siemplify.LOGGER.info(f'Added Alert {alert_id} to package results')
    # If the inbox for the user has no unread emails. 
    else: 
        siemplify.LOGGER.info(f'The inbox for user {username} has no unread emails') 
    # Returning all the created alerts to the cases module in Siemplify 
    siemplify.return_package(alerts) 


if __name__ == '__main__':
# Connectors run in iterations. The interval is configurable from the ConnectorsScreen UI. 
    is_test_run = not (len(sys.argv) < 2 or sys.argv[1] == 'True') 
    main(is_test_run)

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

ייבוא רלוונטי

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

from SiemplifyConnectors import SiemplifyConnectorExecution # This module is responsible for executing the connector 
from SiemplifyConnectorsDataModel import AlertInfo # The data model that contains the alert info class
from SiemplifyUtils import output_handler, convert_datetime_to_unix_time, convert_string_to_datetime # The functions that convert time 
import email, imaplib, sys, re

פונקציה ראשית

הפונקציה Main היא נקודת הכניסה של הסקריפט. מפרש Python מריץ את הקוד באופן רציף וקורא לכל שיטה שמוגדרת בסקריפט.

מחפשים את הפרמטרים של המחבר. משתמשים בפונקציה siemplify.extract_connector_param כדי לחלץ כל פרמטר שהוגדר עבור המחבר (username, ‏ password, ‏ imap_host,‏ imap_port, ‏ folder_to_check).

# Extracting the connector's Params
    username = siemplify.extract_connector_param(param_name="Username") 
    password = siemplify.extract_connector_param(param_name="Password") 
    imap_host = siemplify.extract_connector_param(param_name="IMAP Server Address") 
    imap_port = siemplify.extract_connector_param(param_name="IMAP Port") 
    folder_to_check = siemplify.extract_connector_param(param_name="Folder to check for emails")

תשתמש בפונקציה get_email_messages_data (`imap_host`, `imap_port`, `username`, `password`, `folder_to_check) כדי לאסוף את כל המידע מהאימיילים שלא נקראו.
```
# Getting the digested email message data 
    email_messages_data_list = get_email_messages_data(imap_host, imap_port, username, password, folder_to_check) 
```
אחרי שתקבלו את כל המידע מהאימייל, תאשרו שהמידע נאסף ותבצעו את הפעולות הבאות בכל אימייל:
```
# If the email_messages_data_list is not empty 
if len(email_messages_data_list) > 0:
   for message in email_messages_data_list:
       # Converting the email message datetime from string to unix time by SiemplifyUtils functions
```
הקוד הזה מחלץ את תאריך ההודעה באמצעות datetime_email_message = message['Date'] ואז ממיר את התאריך והשעה האלה לזמן ראשית יוניקס באמצעות הפונקציות של Google SecOps:
```
string_to_datetime = convert_string_to_datetime(datetime_email_message) 
datetime_in_unix_time = convert_datetime_to_unix_time(string_to_datetime) 
```
מחפשים כתובות URL בגוף הודעת האימייל באמצעות הפונקציה find_url_in_email_message_body(siemplify_email_messages_data_list). אם נמצאה כתובת URL, אפשר להשתמש במוצרים אחרים במדריך שלנו כדי לבדוק אם היא זדונית.

found_urls_in_email_body = find_url_in_email_message_body(siemplify, email_messages_data_list)

מחפשים את המזהה הייחודי של כל הודעת אימייל ומקצים אותו למשתנה alert_id:

# Getting the unique id of each email message and removing the suffix '@mail.gmail.com' from the Message-ID, Each alert id can be ingested to the system only once. 
alert_id = message['Message-ID'].replace('@mail.gmail.com','')

אחרי שתחלצו את כל הפרטים הנדרשים כדי להזין את ההתראה לפלטפורמת Google SecOps, תיצרו את ההתראה ואת האירוע:

# Creating the event by calling create_event() function 
created_event = create_event(siemplify, alert_id, message, found_urls_in_email_body, datetime_in_unix_time) 
# Creating the alert by calling create_alert() function 
created_alert = create_alert(siemplify, alert_id, message, datetime_in_unix_time, created_event)

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

    # Checking that the created_alert is not None 
    if created_alert is not None: 
        alerts.append(created_alert)     
        siemplify.LOGGER.info(f"Added Alert {alert_id} to package results")

בתרחיש שבו אין אימיילים עוד לא קראתי בתיבת הדואר הנכנס של המשתמש, מוסיפים את הקוד הבא:
```
else:
    siemplify.LOGGER.info(f"The inbox for user {username} has no unread emails")    
```
רשימת ההתראות מוחזרת למערכת, וכל התראה מוצגת כבקשת תמיכה בתור של בקשות התמיכה:
```
    # Returning all the created alerts to the cases module in Siemplify 
    siemplify.return_package(alerts)
    
```

מריצים את הפונקציה Main בטווח הזמנים שהגדרתם בהגדרות המחבר:

if __name__ == "__main__": 
    # Connectors run in iterations. The interval is configurable from the Connectors UI. 
    is_test_run = not (len(sys.argv) < 2 or sys.argv[1] == 'True')
    main(is_test_run)

קבלת הודעת אימייל שלא נקראה

הפונקציה Get the unread email message מתחברת לאימייל באמצעות המודולים Imap ו-Email, ומאחזרת את פרטי הודעת האימייל. היא גם מחזירה רשימה עם כל המידע של כל הודעות האימייל שלא נקראו.

בכיתה הראשית, משתמשים בפונקציה:

get_email_messages_data(imap_host, imap_port, username, password,
folder_to_check)

def get_email_messages_data(imap_host, imap_port, username, password, folder_to_check):
    """Returns all unread email messages"""
    email_messages_data_list = []

מתחברים לאימייל באמצעות imap module:

    # Login to email using 'imap' module
    mail = imaplib.IMAP4_SSL(imap_host, imap_port)
    mail.login(username, password)

קובעים את התיקייה באימייל שבה רוצים לבדוק אם יש הודעות שלא נקראו. בדוגמה הזו, אתם מחלצים אימיילים מהתיקייה Inbox (תיבת הדואר הנכנס) ‫(DEFAULT_FOLDER_TO_CHECK_INBOX = "inbox"):

# Determining the default email folder to pull emails from - 'inbox'
if folder_to_check is None:
    folder_to_check = DEFAULT_FOLDER_TO_CHECK_INBOX
# Selecting the email folder to pull the data from mail.select(folder_to_check)

איסוף כל ההודעות שלא נקראו DEFAULT_MESSAGES_TO_READ_UNSEEN = "UNSEEN", ואז המרת הנתונים לרשימה:

# Storing the email message data
result, data = mail.search(None, DEFAULT_MESSAGES_TO_READ_UNSEEN)
# If there are several emails collected in the cycle it will split each
# email message into a separate item in the list chosen_mailbox_items_list
if len(data) > 0:
    chosen_mailbox_items_list = data[0].split()
    # Iterating each email message and appending to emails_messages_data_list
    for item in chosen_mailbox_items_list:
        typ, email_data = mail.fetch(item, '(RFC 822)')
        # Decoding from binary string to string
        raw_email = email_data[0][1].decode("utf-8")
        # Turning the email data into an email object
        email_message = email.message_from_string(raw_email)
        # Appending the email message data to email_messages_data_list
        email_messages_data_list.append(email_message)
return email_messages_data_list

יצירת האירוע

הפונקציה Create the event יוצרת את האירוע על ידי שיוך כל רכיב של הודעת האימייל לשדות האירוע, בהתאמה.

בכיתה הראשית, יוצרים את האירוע באמצעות הפונקציה: create_event(siemplify, alert_id, email_message_data, all_found_url_in_emails_body_list, datetime_in_unix_time)

def create_event(siemplify, alert_id, email_message_data, all_found_url_in_emails_body_list, datetime_in_unix_time):
    """Returns the digested data of a single unread email"""
    siemplify.LOGGER.info(f"--- Started processing Event: alert_id: {alert_id} | event_id: {alert_id}")

יוצרים מילון לשדות האירוע. שדות החובה הם event["StartTime"], event["EndTime"], event["event_name"] and event["device_product"]:

event = {} 
event["StartTime"] = datetime_in_unix_time # Time should be saved in UnixTime. You may use SiemplifyUtils.convert_datetime_to_unix_time, or SiemplifyUtils.convert_string_to_datetime 
event["EndTime"] = datetime_in_unix_time # Time should be saved in UnixTime. You may use SiemplifyUtils.convert_datetime_to_unix_time, or SiemplifyUtils.convert_string_to_datetime 
event["event_name"] = "Suspicious email" 
event["device_product"] = PRODUCT # ie: "device_product" is the field name that describes the product the event originated from. 
event["Subject"] = email_message_data["Subject"] 
event["SourceUserName"] = email_message_data["From"] 
event["DestinationUserName"] = email_message_data["To"] 
event["found_url"] = ",".join(all_found_url_in_emails_body_list) 
siemplify.LOGGER.info(f"---Finished processing Event: alert_id: {alert_id} | event_id: {alert_id}")
return event

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

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

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

בכיתה הראשית, יוצרים את ההתראה:

def create_alert(siemplify, alert_id, email_message_data,datetime_in_unix_time, created_event): 
    """Returns an alert which is one event that contains one unread email message"""
    siemplify.LOGGER.info(f"-------------- Started processing Alert {alert_id}")
    create_event = None

יוצרים את מכונת alert_info ומפעילים אותה:

# Initializes the alert_info Characteristics Fields 
alert_info.display_id = f"{alert_id}" 
alert_info.ticket_id = f"{alert_id}" 
alert_info.name = email_message_data['Subject'] 
alert_info.rule_generator = RULE_GENERATOR_EXAMPLE 
alert_info.start_time = datetime_in_unix_time 
alert_info.end_time = datetime_in_unix_time 
alert_info.device_vendor = VENDOR 
alert_info.device_product = PRODUCT

אחרי שיוצרים את ההתראה, מאמתים את האירוע ומצרפים אותו למאפיינים alert_info:

siemplify.LOGGER.info(f"Events creating started for alert {alert_id}")
try:
    if created_event is not None:
        alert_info.events.append(created_event)
    siemplify.LOGGER.info(f"Added Event {alert_id} to Alert {alert_id}")
# Raise an exception if failed to process the event
except Exception as e:
    siemplify.LOGGER.error(f"Failed to process event {alert_id}")
    siemplify.LOGGER.exception(e)
return alert_info

איך מוצאים את כתובת ה-URL בפונקציה של תוכן האימייל

הפונקציה find the URL in the email body סורקת את תוכן האימייל כדי למצוא כתובות URL. כדי להשתמש בפונקציה הזו, פועלים לפי השלבים הבאים:

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

def find_url_in_email_message_body(siemplify, email_messages_data_list):
    """
    Search for a url in the email body,
    """
    all_found_url_in_emails_body_list = []
    for message in email_messages_data_list:
        for part in message.walk():
            if part.get_content_maintype() == 'text\plain':
                continue

אם הגוף מכיל את התוכן הנדרש, טוענים את המידע הזה באמצעות email_message_body = part.get_payload() ומחפשים כתובות URL באמצעות הביטוי הרגולרי:


URLS_REGEX=r"http[s]?://(?:[a-zA-Z]|[0-9]|[$-_@.&+]|[!*\(\),]|(?:%[0-9a-fA-F][0-9a-fA-F]))+"

בדוגמה הזו, כתובות ה-URL מחולצות מתוכן האימייל:

email_message_body = part.get_payload()
all_found_urls = re.findall(URLS_REGEX, str(email_message_body))
for url in all_found_urls:
    if url not in all_found_url_in_emails_body_list:
        all_found_url_in_emails_body_list.append(url)
siemplify.LOGGER.info(f"The URL found: {all_found_url_in_emails_body_list}") 
return all_found_url_in_emails_body_list

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

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