Connector-Logik erstellen
In diesem Dokument wird beschrieben, wie Sie einen neuen Connector in der integrierten Entwicklungsumgebung (IDE) erstellen. Dazu erstellen Sie eine Integration und verknüpfen dann einen Connector damit.
E‑Mail-Connector-Integration erstellen
So erstellen Sie eine E-Mail-Connector-Integration:
- Klicke im Fenster Automatisierung > IDE auf Hinzufügen, um ein neues IDE-Element hinzuzufügen.
- Wählen Sie Integration aus und nennen Sie die Integration
Email Connector. - Klicken Sie auf Erstellen. Die Integration wird in der Seitenleiste mit einem Standardsymbol angezeigt.
- Klicken Sie auf more_vert Benutzerdefinierte Integration konfigurieren und legen Sie die folgenden Einstellungen fest:
- Beschreibung
- Symbol
- Python-Abhängigkeiten
- Integrationsparameter
- Klicken Sie auf Speichern.
E‑Mail-Connector erstellen
So erstellen Sie einen E‑Mail-Connector:
- Klicke im Fenster Automatisierung > IDE auf Hinzufügen, um ein neues IDE-Element hinzuzufügen.
- Wählen Sie das Optionsfeld Connector aus und nennen Sie den Connector
My Email Connector. - Wählen Sie in der Liste die Integration Email Connector aus, um den Connector mit der Integration zu verknüpfen.
- Klicken Sie auf Erstellen.
Connector-Parameter festlegen
Legen Sie nach dem Erstellen des Connectors die Connector-Parameter fest:
- Legen Sie die Connector-Parameter gemäß der Tabelle fest:
- Geben Sie die folgenden Felddetails ein:
- Product Field Name = device_product: Bestimmt den Rohfeldwert, der dem Produktnamen der Benachrichtigung zugewiesen werden soll. Suchen Sie im Code nach dem zugehörigen Feld, das als
Mail(Produkt) definiert wurde.event["device_product"] = PRODUCT #The PRODUCT constant is `"Mail" - Ereignisfeldname = event_name: Bestimmt den Rohfeldwert, der dem Ereignistypsfeld zugewiesen werden soll. Suchen Sie im Code nach dem zugehörigen Feld, das als
Suspicious emaildefiniert ist.event["event_name"] = Suspicious email
| Parameter | Beschreibung | Typ |
|---|---|---|
Username |
Erforderlich. IMAP-Nutzername. Die E‑Mail-Adresse, von der der Connector die E‑Mails in die Google SecOps-Plattform abruft. Der Standardwert ist email@gmail.com. |
String |
Password |
Erforderlich. IMAP-Passwort (Internet Message Access Protocol) für die E-Mail-Adresse, die der Connector verwendet, um E-Mails in die Google SecOps-Plattform zu übertragen. | Passwort |
IMAP Port |
Erforderlich. Der spezifische Netzwerkport, der von IMAP verwendet wird, um auf E‑Mails von einem Remote-Server zuzugreifen. Beispiel: 993 (Standardwert). |
int |
IMAP Server Address |
Erforderlich. Der Server für eingehende E‑Mails für ein IMAP-Konto. Beispiel: imap.gmail.com (Standardwert). |
String |
Folder to check for emails |
Optional: Es werden nur E‑Mails aus dem angegebenen Ordner abgerufen. Beispiel: Inbox (Standardwert). |
inbox |
E‑Mail-Connector bearbeiten
So bearbeiten Sie die Parameter für den E‑Mail-Connector:
- Kopieren Sie den folgenden Code, der für My Email Connector erstellt wurde, fügen Sie ihn in die IDE ein und folgen Sie der Anleitung:
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) - Nachdem Sie den Code des Connectors kopiert haben, prüfen Sie die erforderlichen Module, die importiert werden müssen, und fahren Sie dann mit der Hauptfunktion fort. Jede Methode, die von der Hauptfunktion aufgerufen wird, wird später ausführlicher erläutert.
Relevante Importe
Ein Python-Modul enthält eine Reihe von Funktionen, Klassen oder Variablen, die definiert und implementiert sind. Importieren Sie die folgenden Module in das Skript, um alle folgenden Funktionen zu implementieren:
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
Hauptfunktion
Die Funktion Main ist der Einstiegspunkt des Scripts. Der Python-Interpreter führt den Code sequenziell aus und ruft jede im Script definierte Methode auf.
- Extrahieren Sie die Connector-Parameter. Verwenden Sie die Funktion
siemplify.extract_connector_param, um jeden konfigurierten Parameter für den Connector (username,password,imap_host,imap_port,folder_to_check) zu extrahieren.# 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") - Verwenden Sie die Funktion
get_email_messages_data (`imap_host`, `imap_port`, `username`, `password`, `folder_to_check), um alle Informationen aus den ungelesenen E‑Mails zu erfassen.# Getting the digested email message data email_messages_data_list = get_email_messages_data(imap_host, imap_port, username, password, folder_to_check) - Nachdem Sie alle Informationen aus der E-Mail erhalten haben, bestätigen Sie, dass die Informationen erfasst wurden, und führen Sie dann die folgenden Aktionen für jede E-Mail aus:
# 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
Dieser Code extrahiert das Nachrichtendatum nachdatetime_email_message = message['Date']und konvertiert dieses Datum und diese Uhrzeit dann mithilfe von Google SecOps-Funktionen in die Unix-Epochenzeit:
string_to_datetime = convert_string_to_datetime(datetime_email_message) datetime_in_unix_time = convert_datetime_to_unix_time(string_to_datetime)
- Mit der Funktion
find_url_in_email_message_body(siemplify_email_messages_data_list)können Sie im Text von E‑Mail-Nachrichten nach URLs suchen. Wenn eine URL gefunden wird, verwenden Sie andere Produkte in unserem Playbook, um zu prüfen, ob sie schädlich ist. - Extrahieren Sie die eindeutige ID jeder E‑Mail-Nachricht und weisen Sie sie der Variablen
alert_idzu:# 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','') - Nachdem Sie alle erforderlichen Details extrahiert haben, um die Benachrichtigung in die Google SecOps-Plattform aufzunehmen, erstellen Sie die Benachrichtigung und das Ereignis:
# 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)
- Validieren Sie die erstellte Benachrichtigung und das erstellte Ereignis. Fügen Sie die Benachrichtigung nach der Validierung der Benachrichtigungsliste hinzu.
# 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") - Fügen Sie den folgenden Code für ein Szenario hinzu, in dem der Posteingang des angegebenen Nutzers keine ungelesenen E-Mails enthält:
else: siemplify.LOGGER.info(f"The inbox for user {username} has no unread emails") - Geben Sie die Liste der Benachrichtigungen an das System zurück. Jede Benachrichtigung wird dann als Fall in der Fallwarteschlange angezeigt:
# Returning all the created alerts to the cases module in Siemplify siemplify.return_package(alerts) - Führen Sie die Funktion
Maininnerhalb der in der Connector-Konfiguration festgelegten Zeiten aus: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)
Ungelesene E‑Mail-Nachricht abrufen
Die Funktion Get the unread email message stellt mit den Modulen Imap und Email eine Verbindung zur E‑Mail her und ruft die Details der E‑Mail-Nachricht ab. Außerdem wird eine Liste mit allen Informationen zu allen ungelesenen E‑Mails zurückgegeben.
- Verwenden Sie in der Hauptklasse die Funktion
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 = [] - Stellen Sie mit der
imap moduleeine Verbindung zur E-Mail-Adresse her:# Login to email using 'imap' module mail = imaplib.IMAP4_SSL(imap_host, imap_port) mail.login(username, password) - Bestimmen Sie den Ordner in der E‑Mail, in dem nach ungelesenen Nachrichten gesucht werden soll.
In diesem Beispiel extrahieren Sie E‑Mails aus dem Ordner Posteingang
(DEFAULT_FOLDER_TO_CHECK_INBOX = "inbox"): - Erfassen Sie alle ungelesenen Nachrichten
DEFAULT_MESSAGES_TO_READ_UNSEEN = "UNSEEN"und konvertieren Sie diese Daten dann in eine Liste:# 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)
Ereignis erstellen
Die Funktion Create the event erstellt den Termin, indem sie jede E‑Mail-Nachrichtenkomponente den entsprechenden Terminfeldern zuordnet.
- Erstellen Sie das Ereignis in der Hauptklasse mit der Funktion
create_event(siemplify, alert_id, email_message_data, all_found_url_in_emails_body_list, datetime_in_unix_time). -
Erstellen Sie ein Wörterbuch für die Ereignisfelder. Die Pflichtfelder sind
event["StartTime"], event["EndTime"], event["event_name"] and event["device_product"]: - Jede Benachrichtigung enthält ein oder mehrere Ereignisse. In diesem Beispiel enthält eine Benachrichtigung ein einzelnes Ereignis: eine E‑Mail-Nachricht. Erstellen Sie daher nach dem Erstellen des Ereignisses die Benachrichtigung, die alle Ereignisinformationen enthält.
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
Benachrichtigungsinformationen erstellen und die Felder für die Benachrichtigungsinformationen initialisieren
Diese Funktion erstellt die Benachrichtigung. Jede Benachrichtigung enthält ein oder mehrere Ereignisse. In diesem Fall enthält jede Benachrichtigung ein Ereignis, das im Grunde eine E‑Mail-Nachricht ist.
- Erstellen Sie in der Hauptklasse die Benachrichtigung:
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 - Erstellen Sie die
alert_info-Instanz und initialisieren Sie sie:# 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 - Nachdem Sie die Benachrichtigung erstellt haben, validieren Sie das Ereignis und hängen Sie es an die
alert_info-Attribute an:
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 im E‑Mail-Text finden
Mit der Funktion URL im E‑Mail-Text suchen wird der E‑Mail-Text nach URLs durchsucht. So verwenden Sie diese Funktion:
- Suchen Sie in jeder E‑Mail-Nachricht den Teil des Nachrichtentexts mit Nur-Text-Inhalt:
- Wenn der Text den erforderlichen Inhalt enthält, laden Sie diese Informationen mit
email_message_body = part.get_payload()und suchen Sie mit dem regulären Ausdruck nach URLs: - In diesem Beispiel werden URLs aus dem E‑Mail-Text extrahiert:
- Nachdem Sie den Connector-Code überprüft haben, können Sie einen Connector konfigurieren, um Fälle aus einem ausgewählten Gmail-Posteingang in die Plattform aufzunehmen.
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 Benötigen Sie weitere Hilfe? Antworten von Community-Mitgliedern und Google SecOps-Experten erhalten