커넥터 로직 빌드

다음에서 지원:

이 문서에서는 통합 개발 환경 (IDE)에서 통합을 만든 후 커넥터를 통합에 연결하여 새 커넥터를 빌드하는 방법을 설명합니다.

이메일 커넥터 통합 만들기

이메일 커넥터 통합을 만들려면 다음 단계를 따르세요.

  1. 응답 > IDE 창에서 추가 를 클릭하여 새 IDE 항목을 추가합니다.
  2. 통합 을 선택하고 통합 이름을 Email Connector로 지정합니다.
  3. 만들기 를 클릭합니다. 통합이 기본 아이콘과 함께 사이드바에 표시됩니다.
  4. more_vert 맞춤 통합 구성 을 클릭하고 다음 설정을 정의합니다.
    • 설명
    • 아이콘
    • Python 종속 항목
    • 통합 매개변수
  5. 저장 을 클릭합니다.

이메일 커넥터 만들기

이메일 커넥터를 만들려면 다음 단계를 따르세요.

  1. 응답 > IDE 창에서 추가 를 클릭하여 새 IDE 항목을 추가합니다.
  2. 커넥터 라디오 버튼을 선택하고 커넥터 이름을 My Email Connector로 지정합니다.
  3. 목록에서 통합 이메일 커넥터 를 선택하여 커넥터를 통합과 연결합니다.
  4. 만들기 를 클릭합니다.

커넥터 매개변수 설정

커넥터를 만든 후 커넥터 매개변수를 설정합니다.

  1. 표에 따라 커넥터 매개변수를 설정합니다.

    2. 매개변수 설명 유형
    Username 필수. IMAP 사용자 이름. 커넥터가 Google SecOps 플랫폼으로 이메일을 가져오는 이메일 주소입니다. 기본값은 email@gmail.com입니다. 문자열
    Password 필수. 커넥터가 Google SecOps 플랫폼으로 이메일을 가져오는 데 사용하는 이메일 주소의 IMAP (인터넷 메시지 접속 프로토콜) 비밀번호입니다. 비밀번호
    IMAP Port 필수. IMAP에서 원격 서버의 이메일에 액세스하는 데 사용하는 특정 네트워크 포트입니다. 예: 993 (기본값) int
    IMAP Server Address 필수. IMAP 계정의 수신 메일 서버입니다. 예: imap.gmail.com (기본값) 문자열
    Folder to check for emails 선택사항. 지정된 폴더에서만 이메일을 가져옵니다. 예: Inbox (기본값) inbox
  2. 다음 필드 세부정보를 입력합니다.
    1. 제품 필드 이름 = device_product: 알림의 제품 이름에 할당할 원시 필드 값을 결정합니다. 코드에서 Mail (제품)로 정의된 관련 필드를 찾습니다.
      event["device_product"] = PRODUCT #The PRODUCT constant is `"Mail"
    2. 이벤트 필드 이름 = event_name: 이벤트 유형 필드에 할당할 원시 필드 값을 결정합니다. Suspicious email로 정의된 코드에서 관련 필드를 찾습니다.
      event["event_name"] = Suspicious email

이메일 커넥터 수정

이메일 커넥터의 매개변수를 수정하려면 다음 단계를 따르세요.

  1. 내 이메일 커넥터에 대해 생성된 다음 코드를 복사하여 IDE에 붙여넣고 안내를 따릅니다. 
    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. 커넥터 코드를 복사한 후 가져와야 하는 필수 모듈을 검토한 다음 기본 함수로 진행합니다. 기본 함수에서 호출되는 각 메서드는 나중에 더 자세히 설명합니다.

관련 가져오기

Python 모듈에는 정의되고 구현된 함수, 클래스 또는 변수 집합이 있습니다. 다음 함수를 모두 구현하려면 스크립트로 이러한 모듈을 가져옵니다.

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

기본 함수

기본 함수는 스크립트의 진입점입니다. Python 인터프리터는 코드를 순차적으로 실행하고 스크립트에 정의된 각 메서드를 호출합니다.

  1. 커넥터 매개변수를 추출합니다. siemplify.extract_connector_param 함수를 사용하여 커넥터 (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. get_email_messages_data (`imap_host`, `imap_port`, `username`, `password`, `folder_to_check) 함수를 사용하여 읽지 않은 이메일의 모든 정보를 수집합니다. 
    # Getting the digested email message data 
    email_messages_data_list = get_email_messages_data(imap_host, imap_port, username, password, folder_to_check)
  3. 이메일에서 모든 정보를 받은 후 정보가 수집되었는지 확인한 다음 각 이메일에서 다음 작업을 실행합니다.
    # 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']로 메시지 날짜를 추출한 다음 Google SecOps 함수를 사용하여 이 날짜/시간을 Unix 시간으로 변환합니다.
    string_to_datetime = convert_string_to_datetime(datetime_email_message) 
datetime_in_unix_time = convert_datetime_to_unix_time(string_to_datetime)
  4. find_url_in_email_message_body(siemplify_email_messages_data_list) 함수를 사용하여 이메일 메시지 본문에서 URL을 검색합니다. URL이 발견되면 플레이북의 다른 제품을 사용하여 악성 URL인지 확인합니다.
    5. found_urls_in_email_body = find_url_in_email_message_body(siemplify, email_messages_data_list)
  5. 각 이메일 메시지의 고유 ID를 추출하고 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. Google SecOps 플랫폼으로 알림을 수집하는 데 필요한 모든 세부정보를 추출한 후 알림과 이벤트를 만듭니다.
    # 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. 생성된 알림과 생성된 이벤트를 검증합니다. 검증 후 알림을 알림 목록에 추가합니다. 
        # 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. 지정된 사용자의 받은편지함에 읽지 않은 이메일이 없는 시나리오에서는 다음 코드를 추가합니다. 
    else:
    siemplify.LOGGER.info(f"The inbox for user {username} has no unread emails")
  9. 알림 목록을 시스템에 반환하면 각 알림이 케이스 대기열에 케이스로 표시됩니다. 
        # Returning all the created alerts to the cases module in Siemplify 
    siemplify.return_package(alerts)
  10. 커넥터 구성에서 설정한 시간 내에 Main 함수를 실행합니다. 
    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)

읽지 않은 이메일 메시지 가져오기

읽지 않은 이메일 메시지 가져오기 함수는 Imap 및 Email 모듈을 사용하여 이메일에 연결하고 이메일 메시지 세부정보를 가져옵니다. 또한 읽지 않은 모든 이메일 메시지의 모든 정보가 포함된 목록을 반환합니다.

  1. 기본 클래스에서 함수를 사용합니다: 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. imap module을 사용하여 이메일에 연결합니다. 
        # Login to email using 'imap' module
    mail = imaplib.IMAP4_SSL(imap_host, imap_port)
    mail.login(username, password)
  3. 읽지 않은 메시지를 확인할 이메일의 폴더를 결정합니다. 이 예시에서는 받은편지함 폴더 (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. 읽지 않은 모든 메시지 DEFAULT_MESSAGES_TO_READ_UNSEEN = "UNSEEN"를 수집한 다음 이 데이터를 목록으로 변환합니다. 
    # 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

이벤트 만들기

이벤트 만들기 함수는 각 이메일 메시지 구성요소를 이벤트 필드에 각각 연결하여 이벤트를 만듭니다.

  1. 기본 클래스에서 함수를 사용하여 이벤트를 만듭니다: 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. 이벤트 필드의 사전을 만듭니다. 필수 필드는 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. 각 알림에는 하나 이상의 이벤트가 포함됩니다. 이 예시에서 알림에는 하나의 이벤트(이메일 메시지 1개)가 포함됩니다. 따라서 이벤트를 만든 후 모든 이벤트 정보가 포함된 알림을 만듭니다.

알림 정보를 만들고 알림 정보 특성 필드 초기화

이 함수는 알림을 만듭니다. 각 알림에는 하나 이상의 이벤트가 포함됩니다. 이 경우 각 알림에는 기본적으로 이메일 메시지 1개인 이벤트 1개가 포함됩니다.

  1. 기본 클래스에서 알림을 만듭니다.
    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. 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. 알림을 만든 후 이벤트를 검증하고 alert_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

이메일 본문 함수에서 URL 찾기

이메일 본문에서 URL 찾기 함수는 이메일 본문에서 URL을 검색합니다. 이 함수를 사용하려면 다음 단계를 따르세요.

  1. 각 이메일 메시지에서 일반 텍스트 콘텐츠가 포함된 메일 본문의 부분을 찾습니다.
    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. 본문에 필수 콘텐츠가 포함되어 있으면 이 정보를 email_message_body = part.get_payload()로 로드하고 정규 표현식을 사용하여 URL을 검색합니다.
    3. 
URLS_REGEX=r"http[s]?://(?:[a-zA-Z]|[0-9]|[$-_@.&+]|[!*\(\),]|(?:%[0-9a-fA-F][0-9a-fA-F]))+"
  3. 이 예시에서는 이메일 본문에서 URL을 추출합니다.
    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. 커넥터 코드를 검토한 후 선택한 Gmail 받은편지함에서 플랫폼으로 케이스를 수집하도록 커넥터를 구성할 수 있습니다.

도움이 더 필요하신가요? 커뮤니티 회원 및 Google SecOps 전문가에게 문의하여 답변을 받으세요.