יצירת לוגיקה של מחבר
במאמר הזה מוסבר איך ליצור מחבר חדש בסביבת הפיתוח המשולבת (IDE) על ידי יצירת אינטגרציה ואז קישור של מחבר אליה.
יצירת שילוב של מחבר אימייל
כדי ליצור שילוב של מחבר אימייל, פועלים לפי השלבים הבאים:
- בחלון Automation > IDE, לוחצים על Add כדי להוסיף פריט IDE חדש.
- בוחרים באפשרות Integration (שילוב) ונותנים לשילוב את השם
Email Connector. - לוחצים על יצירה. השילוב מופיע בסרגל הצד עם סמל ברירת מחדל.
- לוחצים על more_vert הגדרת שילוב בהתאמה אישית וקובעים את ההגדרות הבאות:
- תיאור
- סמל
- יחסי תלות של Python
- פרמטרים של שילוב
- לוחצים על Save.
יצירת מחבר אימייל
כדי ליצור מחבר לאימייל:
- בחלון Automation > IDE, לוחצים על Add כדי להוסיף פריט IDE חדש.
- בוחרים את כפתור הבחירה Connector ונותנים למחבר את השם
My Email Connector. - ברשימה, בוחרים את השילוב Email Connector כדי לשייך את המחבר לשילוב.
- לוחצים על יצירה.
הגדרת הפרמטרים של המחבר
אחרי שיוצרים את המחבר, מגדירים את הפרמטרים שלו:
- מגדירים את הפרמטרים של המחבר לפי הטבלה:
- מזינים את פרטי השדות הבאים:
- שם שדה המוצר = device_product: קובע את ערך השדה הגולמי שיוקצה לשם המוצר של ההתראה. מוצאים את השדה הרלוונטי בקוד, שהוגדר בתור
Mail(מוצר).event["device_product"] = PRODUCT #The PRODUCT constant is `"Mail" - שם שדה האירוע = event_name: קובע את ערך השדה הגולמי שיוקצה לשדה של סוג האירוע. מוצאים את השדה הרלוונטי בקוד, שמוגדר בתור
Suspicious email.event["event_name"] = Suspicious email
| פרמטר | תיאור | סוג |
|---|---|---|
Username |
חובה. שם המשתמש ב-IMAP. כתובת האימייל שממנה המחבר מושך את האימיילים לפלטפורמת Google SecOps. ערך ברירת המחדל הוא email@gmail.com. |
מחרוזת |
Password |
חובה. סיסמת IMAP (Internet Message Access Protocol) לכתובת האימייל שבה המחבר משתמש כדי למשוך אימיילים לפלטפורמת Google SecOps. | סיסמה |
IMAP Port |
חובה. יציאת הרשת הספציפית שמשמשת את ה-IMAP כדי לגשת לאימיילים משרת מרוחק. לדוגמה: 993 (ערך ברירת המחדל). |
int |
IMAP Server Address |
חובה. שרת האימיילים הנכנסים של חשבון IMAP. לדוגמה, imap.gmail.com (ערך ברירת המחדל). |
מחרוזת |
Folder to check for emails |
אופציונלי. שליפת אימיילים רק מהתיקייה שצוינה. לדוגמה: Inbox (ערך ברירת המחדל). |
תיבת דואר נכנס, אינבוקס |
עריכת המחבר לאימייל
כדי לערוך את הפרמטרים של מחבר האימייל, פועלים לפי השלבים הבאים:
- מעתיקים את הקוד הבא שנוצר בשביל My Email Connector, מדביקים אותו ב-IDE ופועלים לפי ההוראות:
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) - אחרי שמעתיקים את הקוד של המחבר, בודקים את המודולים שצריך לייבא ואז ממשיכים לפונקציה הראשית. בהמשך נדון בפירוט בכל method שנקראת מהפונקציה הראשית.
ייבוא רלוונטי
מודול 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']ואז ממיר את התאריך והשעה האלה לזמן Unix epoch באמצעות פונקציות של 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, צריך להשתמש במוצרים אחרים במדריך שלנו כדי לבדוק אם היא זדונית. - מחלצים את המזהה הייחודי של כל הודעת אימייל ומקצים אותו למשתנה
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)
found_urls_in_email_body = find_url_in_email_message_body(siemplify, email_messages_data_list)
קבלת הודעת האימייל שלא נקראה
הפונקציה 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"): - אוספים את כל ההודעות שלא נקראו
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
# 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)
יוצרים את האירוע.
הפונקציה Create the event יוצרת את האירוע על ידי שיוך של כל רכיב בהודעת האימייל לשדות האירוע, בהתאמה.
- במחלקה הראשית, יוצרים את האירוע באמצעות הפונקציה:
create_event(siemplify, alert_id, email_message_data, all_found_url_in_emails_body_list, datetime_in_unix_time) -
יוצרים מילון לשדות של האירוע. שדות החובה הם
event["StartTime"], event["EndTime"], event["event_name"] and event["device_product"]: - כל התראה מכילה אירוע אחד או יותר. בדוגמה הזו, ההתראה מכילה אירוע אחד: הודעת אימייל אחת. לכן, אחרי שיוצרים את האירוע, צריך ליצור את ההתראה שמכילה את כל המידע על האירוע.
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 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 בתוכן האימייל
הפונקציה חיפוש כתובת ה-URL בתוכן האימייל סורקת את תוכן האימייל כדי למצוא כתובות URL. כדי להשתמש בפונקציה הזו, פועלים לפי השלבים הבאים:
- בכל הודעת אימייל, מאתרים את החלק בגוף ההודעה שמכיל תוכן בטקסט פשוט:
- אם גוף ההודעה מכיל את התוכן הנדרש, טוענים את המידע הזה באמצעות
email_message_body = part.get_payload()ומחפשים כתובות URL באמצעות הביטוי הרגולרי: - בדוגמה הזו, המערכת מחלצת כתובות URL מתוכן האימייל:
- אחרי שבודקים את קוד המחבר, אפשר להגדיר מחבר שיקלוט פניות לפלטפורמה מתיבת דואר נכנס נבחרת ב-Gmail.
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
URLS_REGEX=r"http[s]?://(?:[a-zA-Z]|[0-9]|[$-_@.&+]|[!*\(\),]|(?:%[0-9a-fA-F][0-9a-fA-F]))+"
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 הבעיה עדיין לא נפתרה? קבלת תשובות מחברי הקהילה וממומחי Google SecOps.