Connector-Logik erstellen
In diesem Dokument wird erläutert, wie Sie in der integrierten Entwicklungsumgebung (IDE) einen neuen Connector erstellen, indem Sie eine Integration erstellen und dann einen Connector damit verknüpfen.
Integration für E‑Mail-Connector erstellen
So erstellen Sie eine Integration für einen E‑Mail-Connector:
- Klicken Sie im Fenster Antwort > IDE auf Hinzufügen, um ein neues IDE-Element hinzuzufügen.
- Wählen Sie Integration aus und geben Sie der Integration den Namen
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:
- Klicken Sie im Fenster Antwort > IDE auf Hinzufügen, um ein neues IDE-Element hinzuzufügen.
- Wählen Sie das Optionsfeld Connector aus und geben Sie dem Connector den Namen
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
Nachdem Sie den Connector erstellt haben, legen Sie die Connector-Parameter fest:
- Legen Sie die Connector-Parameter gemäß der Tabelle fest:
- Geben Sie die folgenden Felddetails ein:
- Produktfeldname = device_product: Bestimmt den Rohwert des Felds, 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 Rohwert des Felds, der dem Feld für den Ereignistyp zugewiesen werden soll. Suchen Sie im Code nach dem zugehörigen Feld, das als
Suspicious emaildefiniert wurde.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 abzurufen. | Passwort |
IMAP Port |
Erforderlich. Der spezifische Netzwerkport, der von IMAP verwendet wird, um auf E‑Mails von einem Remote-Server zuzugreifen. Beispiel: 993 (Standardwert). |
Ganzzahl |
IMAP Server Address |
Erforderlich. Der Posteingangsserver für ein IMAP-Konto. Beispiel: imap.gmail.com (Standardwert). |
String |
Folder to check for emails |
Optional. Ruft E‑Mails nur aus dem angegebenen Ordner ab. Beispiel: Inbox (Standardwert). |
Posteingang |
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 wurden. Um alle folgenden Funktionen zu implementieren, importieren Sie diese Module in das Skript:
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 Skripts. Der Python-Interpreter führt den Code sequenziell aus und ruft jede im Skript definierte Methode auf.
- Extrahieren Sie die Connector-Parameter. Verwenden Sie die Funktion
siemplify.extract_connector_param, um jeden konfigurierten Parameter für den Connector zu extrahieren (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") - 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
Mit diesem Code wird das Datum der Nachricht mitdatetime_email_message = message['Date']extrahiert und dann mit Google SecOps-Funktionen in die Unix-Epochenzeit konvertiert:
string_to_datetime = convert_string_to_datetime(datetime_email_message) datetime_in_unix_time = convert_datetime_to_unix_time(string_to_datetime)
- Suchen Sie mit der Funktion
find_url_in_email_message_body(siemplify_email_messages_data_list)im Text der E‑Mail-Nachricht nach URLs. 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") - Wenn der Posteingang für den angegebenen Nutzer keine ungelesenen E‑Mails enthält, fügen Sie den folgenden Code hinzu:
else: siemplify.LOGGER.info(f"The inbox for user {username} has no unread emails") - Geben Sie die Benachrichtigungsliste 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 Zeiten aus, die Sie in der Connector-Konfiguration festgelegt haben: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 Ungelesene E‑Mail-Nachricht abrufen 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‑Mail-Nachrichten 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 dem
imap moduleeine Verbindung zur E‑Mail 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 werden E‑Mails aus dem Ordner Posteingang
(DEFAULT_FOLDER_TO_CHECK_INBOX = "inbox")extrahiert: - 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 Ereignis erstellen erstellt das Ereignis, indem sie jede E‑Mail-Nachrichtenkomponente den jeweiligen Ereignisfeldern zuordnet.
- Erstellen Sie in der Hauptklasse das Ereignis 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 Instanz
alert_infound 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 Merkmale von
alert_infoan:
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
Funktion zum Suchen der URL im E‑Mail-Text
Die Funktion URL im E‑Mail-Text suchen durchsucht den E‑Mail-Text nach URLs. So verwenden Sie diese Funktion:
- Suchen Sie für jede 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 folgenden 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 so konfigurieren, dass Fälle aus einem ausgewählten Gmail-Posteingang in die Plattform aufgenommen werden.
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