Desenvolver o conector

Compatível com:

Este documento explica como criar um conector no ambiente de desenvolvimento integrado (IDE) ao criar uma integração e vincular um conector a ela.

Criar uma integração de conector de e-mail

Para criar uma integração de conector de e-mail, siga estas etapas:

  1. Na janela Resposta > IDE, clique em Adicionar para adicionar um novo item de ambiente de desenvolvimento integrado.
  2. Selecione Integração e nomeie a integração como Email Connector.
  3. Clique em Criar. A integração aparece na barra lateral com um ícone padrão.
  4. Clique em more_vert Configurar integração personalizada e defina estas configurações:
    • Descrição
    • Ícone
    • Dependências do Python
    • Parâmetros de integração
  5. Clique em Salvar.

Criar um conector de e-mail

Para criar um conector de e-mail, siga estas etapas:

  1. Na janela Resposta > IDE, clique em Adicionar para adicionar um novo item de ambiente de desenvolvimento integrado.
  2. Selecione o botão de opção Conector e nomeie o conector como My Email Connector.
  3. Na lista, selecione a integração Conector de e-mail para associar o conector à integração.
  4. Clique em Criar.

Definir os parâmetros do conector

Depois de criar o conector, defina os parâmetros dele:

  1. Defina os parâmetros do conector de acordo com a tabela:

    2. Parâmetro Descrição Tipo
    Username Obrigatório. Nome de usuário do IMAP. O endereço de e-mail de onde o conector extrai os e-mails para a plataforma do Google SecOps. O valor padrão é email@gmail.com. string
    Password Obrigatório. Senha do protocolo Internet Message Access Protocol (IMAP) para o endereço de e-mail que o conector usa para extrair e-mails para a plataforma do Google SecOps. senha
    IMAP Port Obrigatório. A porta de rede específica usada pelo IMAP para acessar e-mails de um servidor remoto. Por exemplo: 993 (valor padrão). int
    IMAP Server Address Obrigatório. O servidor de e-mail de entrada de uma conta IMAP. Por exemplo, imap.gmail.com (valor padrão). string
    Folder to check for emails Opcional. Extrai e-mails apenas da pasta especificada. Por exemplo: Inbox (valor padrão). inbox
  2. Insira os seguintes detalhes do campo:
    1. Nome do campo do produto = device_product: determina o valor do campo bruto a ser atribuído ao nome do produto do alerta. Encontre o campo relacionado no código, que foi definido como Mail (produto).
      event["device_product"] = PRODUCT #The PRODUCT constant is `"Mail"
    2. Nome do campo de evento = event_name: determina o valor do campo bruto a ser atribuído ao campo de tipo de evento. Encontre o campo relacionado no código, definido como Suspicious email.
      event["event_name"] = Suspicious email

Editar o conector de e-mail

Para editar os parâmetros do conector de e-mail, siga estas etapas:

  1. Copie o código a seguir criado para Meu conector de e-mail, cole-o no ambiente de desenvolvimento integrado e siga as instruções: 
    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)
  2. Depois de copiar o código do conector, revise os módulos necessários para importação e prossiga para a função principal. Cada método chamado da função principal será discutido mais tarde em mais detalhes.

Importações relevantes

Um módulo do Python tem um conjunto de funções, classes ou variáveis definidas e implementadas. Para implementar todas as funções a seguir, importe esses módulos para o script:

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

Função principal

A função Main é o ponto de entrada do script. O interpretador do Python executa o código sequencialmente e chama cada método definido no script.

  1. Extraia os parâmetros do conector. Use a função siemplify.extract_connector_param para extrair cada parâmetro configurado para o conector (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")
  2. Use a função get_email_messages_data (`imap_host`, `imap_port`, `username`, `password`, `folder_to_check) para coletar todas as informações dos e-mails não lidos. 
    # Getting the digested email message data 
    email_messages_data_list = get_email_messages_data(imap_host, imap_port, username, password, folder_to_check)
  3. Depois de receber todas as informações do e-mail, confirme que elas foram coletadas e faça estas ações em cada e-mail:
    # 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
    Esse código extrai a data da mensagem por datetime_email_message = message['Date'] e depois converte esse horário em horário da época Unix usando funções do Google SecOps:
    string_to_datetime = convert_string_to_datetime(datetime_email_message) 
datetime_in_unix_time = convert_datetime_to_unix_time(string_to_datetime)
  4. Pesquise URLs no corpo da mensagem de e-mail usando a função find_url_in_email_message_body(siemplify_email_messages_data_list). Se um URL for encontrado, use outros produtos no nosso playbook para verificar se ele é malicioso.
    5. found_urls_in_email_body = find_url_in_email_message_body(siemplify, email_messages_data_list)
  5. Extraia o ID exclusivo de cada mensagem de e-mail e atribua-o à variável 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','')
  6. Depois de extrair todos os detalhes necessários para ingerir o alerta na plataforma do Google SecOps, crie o alerta e o evento:
    # 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)
  7. Valide o alerta e o evento criados. Depois de validar, adicione o alerta à lista de alertas. 
        # 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")
  8. Em um cenário em que a caixa de entrada do usuário não tem e-mails não lidos, adicione o seguinte código: 
    else:
    siemplify.LOGGER.info(f"The inbox for user {username} has no unread emails")
  9. Retorne a lista de alertas para o sistema. Cada alerta será apresentado como um caso na fila de casos: 
        # Returning all the created alerts to the cases module in Siemplify 
    siemplify.return_package(alerts)
  10. Execute a função Main nos horários definidos na configuração do conector: 
    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)

Receber a mensagem de e-mail não lida

A função Get the unread email message (Receber a mensagem de e-mail não lida) se conecta ao e-mail com os módulos Imap e Email e recupera os detalhes da mensagem. Ele também retorna uma lista com todas as informações de todas as mensagens de e-mail não lidas.

  1. Na classe principal, use a função: 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 = []
  2. Conecte-se ao e-mail usando o imap module: 
        # Login to email using 'imap' module
    mail = imaplib.IMAP4_SSL(imap_host, imap_port)
    mail.login(username, password)
  3. Determine a pasta no e-mail para verificar se há mensagens não lidas. Neste exemplo, você extrai e-mails da pasta Caixa de entrada (DEFAULT_FOLDER_TO_CHECK_INBOX = "inbox"):
    4. # 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)
  4. Colete todas as mensagens não lidas DEFAULT_MESSAGES_TO_READ_UNSEEN = "UNSEEN" e converta esses dados em uma lista: 
    # 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

Criar o evento

A função Criar o evento associa cada componente da mensagem de e-mail aos campos do evento, respectivamente.

  1. Na classe principal, crie o evento usando a função: create_event(siemplify, alert_id, email_message_data, all_found_url_in_emails_body_list, datetime_in_unix_time)
    2. 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}")
  2. Crie um dicionário para os campos de evento. Os campos obrigatórios são event["StartTime"], event["EndTime"], event["event_name"] and event["device_product"] : 
    3. 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
  3. Cada alerta contém um ou mais eventos. Neste exemplo, um alerta contém um único evento: uma mensagem de e-mail. Portanto, depois de criar o evento, crie o alerta com todas as informações dele.

Crie as informações de alerta e inicialize os campos de características das informações de alerta.

Essa função cria o alerta. Cada alerta contém um ou mais eventos. Nesse caso, cada alerta contém um evento, que é basicamente uma mensagem de e-mail.

  1. Na classe principal, crie o alerta:
    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
  2. Crie e inicialize a instância 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
  3. Depois de criar o alerta, valide o evento e adicione-o às características aert_info:
    4. 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

Encontre o URL na função do corpo do e-mail

A função encontrar o URL no corpo do e-mail verifica o corpo do e-mail em busca de URLs. Para usar essa função, siga estas etapas:

  1. Em cada mensagem de e-mail, localize a parte do corpo com conteúdo de texto simples:
    2. 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
  2. Se o corpo contiver o conteúdo necessário, carregue essas informações com email_message_body = part.get_payload() e pesquise URLs usando a expressão regular:
    3. 
URLS_REGEX=r"http[s]?://(?:[a-zA-Z]|[0-9]|[$-_@.&+]|[!*\(\),]|(?:%[0-9a-fA-F][0-9a-fA-F]))+"
  3. Este exemplo extrai URLs do corpo do e-mail:
    4. 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
  4. Depois de analisar o código do conector, você pode configurar um conector para ingerir casos na plataforma de uma Caixa de entrada do Gmail selecionada.

Precisa de mais ajuda? Receba respostas de membros da comunidade e profissionais do Google SecOps.