Connector entwickeln
In diesem Dokument wird beschrieben, wie Sie in der integrierten Entwicklungsumgebung (IDE) einen neuen Connector erstellen, indem Sie eine Integration erstellen und dann einen Connector damit verknüpfen.
E‑Mail-Connector-Integration erstellen
So erstellen Sie eine E-Mail-Connector-Integration:
- 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 mit einem Standardsymbol in der Seitenleiste 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 Informationen ein:
- Produktfeldname = 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"
- Event Field Name = event_name: Bestimmt den Rohfeldwert, der dem Feld „Ereignistyp“ zugewiesen werden soll. Suchen Sie im Code nach dem zugehörigen Feld, das als
Suspicious email
.event["event_name"] = Suspicious email
definiert ist.
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. Das 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 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). |
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, sehen Sie sich die erforderlichen Module an, die importiert werden müssen, und fahren Sie dann mit der Hauptfunktion fort. Jede Methode, die von der Hauptfunktion aufgerufen wird, wird später genauer erläutert.
Relevante Importe
Ein Python-Modul enthält eine Reihe von Funktionen, Klassen oder Variablen, die definiert und implementiert sind. Wenn Sie alle folgenden Funktionen implementieren möchten, 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 Verbindungsparameter. Mit der Funktion
siemplify.extract_connector_param
können Sie jeden konfigurierten Parameter für den Connector 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")
- Verwende 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 erhoben 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 Nachrichtendatum mitdatetime_email_message = message['Date']
extrahiert und dann mit Google SecOps-Funktionen in die Unix-Epochenzeit umgewandelt:
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 E-Mail-Nachrichtentext 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_id
zu:# 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 zum Aufnehmen der Benachrichtigung in die Google SecOps-Plattform extrahiert haben, 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)
- Erstellte Benachrichtigung und erstelltes Ereignis validieren 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 des betreffenden Nutzers 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")
- Die Liste der Benachrichtigungen wird an das System zurückgegeben und 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
Main
innerhalb 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 abrufen stellt mit den Modulen Imap
und Email
eine Verbindung zur E‑Mail her und ruft die Details der E‑Mail 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
imap module
eine Verbindung zur E-Mail her:# Login to email using 'imap' module mail = imaplib.IMAP4_SSL(imap_host, imap_port) mail.login(username, password)
- Legen Sie den Ordner in der E‑Mail fest, in dem nach ungelesenen Nachrichten gesucht werden soll.
In diesem Beispiel werden E‑Mails aus dem Ordner Posteingang extrahiert
(DEFAULT_FOLDER_TO_CHECK_INBOX = "inbox")
: - Sammle alle ungelesenen Nachrichten
DEFAULT_MESSAGES_TO_READ_UNSEEN = "UNSEEN"
und wandle diese Daten dann in eine Liste um:# 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
Mit der Funktion Create the event (Ereignis erstellen) wird das Ereignis erstellt, indem jede E-Mail-Nachrichtenkomponente den jeweiligen Ereignisfeldern zugeordnet wird.
- 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 Dictionary 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. 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 Merkmale der Benachrichtigungsinformationen initialisieren
Mit dieser Funktion wird die Benachrichtigung erstellt. Jede Benachrichtigung enthält ein oder mehrere Ereignisse. In diesem Fall enthält jede Benachrichtigung ein Ereignis, das im Grunde einer E-Mail entspricht.
- Erstellen Sie die Benachrichtigung über die Hauptklasse:
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
aert_info
-Merkmale 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 in der Funktion „E-Mail-Text“ suchen
Die Funktion URL im E‑Mail-Text suchen scannt den E‑Mail-Text nach URLs. So verwenden Sie diese Funktion:
- Suchen Sie in jeder E‑Mail-Nachricht den Teil des Nachrichtentexts mit Nur-Text-Inhalt:
- Wenn der Text die erforderlichen Inhalte 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 Anfragen 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