Desenvolva o conetor

Compatível com:

Este documento explica como criar um novo conetor no ambiente de desenvolvimento integrado (IDE) através da criação de uma integração e, em seguida, da associação de um conetor à mesma.

Crie uma integração de conetor de email

Para criar uma integração de conetor de email, siga estes passos:

  1. Na janela Resposta > IDE, clique em Adicionar para adicionar um novo item IDE.
  2. Selecione Integração e atribua um nome à integração Email Connector.
  3. Clique em Criar. A integração aparece na barra lateral com um ícone predefinido.
  4. Clique em more_vert Configurar integração personalizada e defina estas definições:
    • Descrição
    • Ícone
    • Dependências do Python
    • Parâmetros de integração
  5. Clique em Guardar.

Crie um conetor de email

Para criar um conetor de email, siga estes passos:

  1. Na janela Resposta > IDE, clique em Adicionar para adicionar um novo item IDE.
  2. Selecione o botão de opção Conetor e atribua um nome ao conetor My Email Connector.
  3. Na lista, selecione a integração Email Connector para associar o conetor à integração.
  4. Clique em Criar.

Defina os parâmetros do conetor

Depois de criar o conetor, defina os parâmetros do conetor:

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

    2. Parâmetro Descrição Tipo
    Username Obrigatório. Nome de utilizador IMAP. O endereço de email a partir do qual o conector extrai os emails para a plataforma Google SecOps. O valor predefinido é email@gmail.com. de string
    Password Obrigatório. Palavra-passe do protocolo IMAP (Internet Message Access Protocol) para o endereço de email que o conector usa para transferir emails para a plataforma Google SecOps. palavra-passe
    IMAP Port Obrigatório. A porta de rede específica usada pelo IMAP para aceder a emails a partir de um servidor remoto. Por exemplo: 993 (valor predefinido). int
    IMAP Server Address Obrigatório. O servidor de correio recebido para uma conta IMAP. Por exemplo, imap.gmail.com (valor predefinido). de string
    Folder to check for emails Opcional. Extrai emails apenas da pasta especificada. Por exemplo: Inbox (valor predefinido). caixa de entrada
  2. Introduza os seguintes detalhes dos campos:
    1. Nome do campo do produto = device_product: determina o valor do campo não processado a atribuir 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 do evento = event_name: determina o valor do campo não processado a atribuir ao campo do tipo de evento. Encontre o campo relacionado no código, definido como Suspicious email.
      event["event_name"] = Suspicious email

Edite o conetor de email

Para editar os parâmetros do conector de email, siga estes passos:

  1. Copie o seguinte código criado para My Email Connector, cole-o no IDE 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, reveja os módulos necessários para importar e, em seguida, avance para a função principal. Cada método chamado a partir da função principal será abordado mais tarde com mais detalhe.

Importações relevantes

Um módulo Python tem um conjunto de funções, classes ou variáveis definidas e implementadas. Para implementar todas as seguintes funções, 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 intérprete Python executa o código sequencialmente e chama cada método definido no script.

  1. Extraia os parâmetros do conetor. Use a função siemplify.extract_connector_param para extrair cada parâmetro configurado para o conetor (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 recolher todas as informações dos emails 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 email, confirme que as informações são recolhidas e, em seguida, execute estas ações em cada email:
    # 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
    Este código extrai a data da mensagem por datetime_email_message = message['Date']e, em seguida, converte esta data/hora na hora de época Unix através das 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 email através da função find_url_in_email_message_body(siemplify_email_messages_data_list). Se for encontrado um URL, use outros produtos no nosso manual para verificar se é 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 email 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 carregar o alerta para a plataforma 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 criado e o evento criado. Após a validação, 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. Num cenário em que a caixa de entrada do utilizador especificado não tem emails não lidos, adicione o seguinte código: 
    else:
    siemplify.LOGGER.info(f"The inbox for user {username} has no unread emails")
  9. Devolve a lista de alertas ao sistema e, em seguida, cada alerta é apresentado como um registo na fila de registos: 
        # Returning all the created alerts to the cases module in Siemplify 
    siemplify.return_package(alerts)
  10. Execute a função Main dentro dos horários definidos na configuração do conetor: 
    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 email não lida

A função Get the unread email message (Obter a mensagem de email não lida) liga-se ao email com os módulos Imap e Email e obtém os detalhes da mensagem de email. Também devolve uma lista com todas as informações de todas as mensagens de email não lidas.

  1. A partir da 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. Associe o email através do seguinte: 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 email para verificar se existem mensagens não lidas. Neste exemplo, extrai emails 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. Recolher todas as mensagens não lidas DEFAULT_MESSAGES_TO_READ_UNSEEN = "UNSEEN" e, em seguida, converter estes dados numa 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

Crie o evento

A função Criar o evento cria o evento associando cada componente da mensagem de email aos campos do evento, respetivamente.

  1. A partir da classe principal, crie o evento através da 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 eventos. 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 email. Por conseguinte, depois de criar o evento, crie o alerta que contém todas as informações do evento.

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

Esta função cria o alerta. Cada alerta contém um ou mais eventos. Neste caso, cada alerta contém um evento, que é basicamente uma mensagem de email.

  1. Na turma 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 a instância do alert_info e inicialize-a: 
    # 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 anexe-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 email

A função encontrar o URL no corpo do email analisa o corpo do email à procura de URLs. Para usar esta função, siga estes passos:

  1. Para cada mensagem de email, localize a parte do corpo da mensagem 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 estas informações com email_message_body = part.get_payload() e pesquise URLs através da 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 email:
    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 rever o código do conector, pode configurar um conector para carregar registos para a plataforma a partir de uma caixa de entrada do Gmail selecionada.

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