Merge pull request #77 from splunk-soar-connectors/grokas/PAPP-37549-es-polling

Author: grokas-splunk

README.md

@@ -577,6 +577,8 @@ This table lists the configuration variables required to operate MS Graph for Of
 
 VARIABLE | REQUIRED | TYPE | DESCRIPTION
 -------- | -------- | ---- | -----------
+**es_security_domain** | optional | string | Security domain for ES findings |
+**es_urgency** | optional | string | Urgency level for ES findings |
 **tenant** | required | string | Tenant ID (e.g. 1e309abf-db6c-XXXX-a1d2-XXXXXXXXXXXX) |
 **client_id** | required | string | Application ID |
 **auth_type** | required | string | Authentication type to use for connectivity |
@@ -606,6 +608,7 @@ VARIABLE | REQUIRED | TYPE | DESCRIPTION
 
 [test connectivity](#action-test-connectivity) - test connectivity <br>
 [on poll](#action-on-poll) - on poll <br>
+[on es poll](#action-on-es-poll) - Poll for new emails and create ES findings for each email. <br>
 [block sender](#action-block-sender) - Add a sender to the blocked senders list <br>
 [copy email](#action-copy-email) - Copy an email to a folder <br>
 [create folder](#action-create-folder) - Create a new mail folder <br>
@@ -677,6 +680,29 @@ PARAMETER | REQUIRED | DESCRIPTION | TYPE | CONTAINS
 
 No Output
 
+## action: 'on es poll'
+
+Poll for new emails and create ES findings for each email.
+
+Type: **ingest** <br>
+Read only: **True**
+
+Callback action for the on_es_poll ingest functionality
+
+#### Action Parameters
+
+PARAMETER | REQUIRED | DESCRIPTION | TYPE | CONTAINS
+--------- | -------- | ----------- | ---- | --------
+**start_time** | optional | Start of time range, in epoch time (milliseconds). | numeric | |
+**end_time** | optional | End of time range, in epoch time (milliseconds). | numeric | |
+**container_count** | optional | Maximum number of container records to query for. | numeric | |
+**es_base_url** | required | Base URL for the Splunk Enterprise Security API | string | |
+**es_session_key** | required | Session token for the Splunk Enterprise Security API | string | |
+
+#### Action Output
+
+No Output
+
 ## action: 'block sender'
 
 Add a sender to the blocked senders list

pyproject.toml

@@ -9,7 +9,7 @@ dependencies = [
     "beautifulsoup4>=4.14.2",
     "msal>=1.31.0",
     "requests>=2.32.0",
-    "splunk-soar-sdk>=3.8.0",
+    "splunk-soar-sdk>=3.8.2",
 ]
 
 # [tool.uv.sources]

src/app.py

@@ -17,7 +17,7 @@
 import os
 import re
 import time
-from collections.abc import Iterator
+from collections.abc import Generator, Iterator
 from html import unescape
 from pathlib import Path
 
@@ -38,7 +38,8 @@
 from soar_sdk.logging import getLogger
 from soar_sdk.models.artifact import Artifact
 from soar_sdk.models.container import Container
-from soar_sdk.params import OnPollParams
+from soar_sdk.models.finding import Finding, FindingAttachment
+from soar_sdk.params import OnESPollParams, OnPollParams
 
 from .consts import (
     MSGOFFICE365_CONTAINER_DESCRIPTION,
@@ -763,6 +764,106 @@ def on_poll(
         state["first_run"] = False
 
 
+@app.on_es_poll()
+def on_es_poll(
+    params: OnESPollParams, soar: SOARClient, asset: Asset
+) -> Generator[Finding, int | None]:
+    """Poll for new emails and create ES findings for each email."""
+    helper = MsGraphHelper(soar, asset)
+    helper.get_token()
+
+    state = getattr(asset, "ingest_state", None) or {}
+    email_address = asset.email_address
+    if not email_address:
+        raise ValueError("Email address is required for ES polling")
+
+    folder = asset.folder or MSGOFFICE365_DEFAULT_FOLDER
+    folder_id = folder
+    if asset.get_folder_id:
+        resolved_id = helper.get_folder_id(folder, email_address)
+        if resolved_id:
+            folder_id = resolved_id
+
+    is_first_run = state.get("es_first_run", True)
+    max_emails = asset.first_run_max_emails if is_first_run else asset.max_containers
+    last_time = state.get("es_last_time")
+
+    endpoint = f"/users/{email_address}/mailFolders/{folder_id}/messages"
+    select_fields = ",".join(MSGOFFICE365_SELECT_PARAMETER_LIST)
+    api_params = {
+        "$select": select_fields,
+        "$top": str(min(max_emails, MSGOFFICE365_PER_PAGE_COUNT)),
+        "$orderby": MSGOFFICE365_ORDERBY_RECEIVED_DESC
+        if asset.ingest_manner == "latest first"
+        else "receivedDateTime asc",
+    }
+
+    if last_time:
+        api_params["$filter"] = f"receivedDateTime gt {last_time}"
+
+    emails_processed = 0
+    latest_time = last_time
+
+    while emails_processed < max_emails:
+        resp = helper.make_rest_call_helper(endpoint, params=api_params)
+        emails = resp.get("value", [])
+
+        if not emails:
+            break
+
+        for email_data in emails:
+            if emails_processed >= max_emails:
+                break
+
+            email_time = email_data.get("receivedDateTime")
+            if email_time and (not latest_time or email_time > latest_time):
+                latest_time = email_time
+
+            email_id = email_data.get("id")
+            subject = email_data.get("subject") or email_id
+
+            attachments = []
+            try:
+                eml_content = helper.make_rest_call_helper(
+                    f"/users/{email_address}/messages/{email_id}/$value",
+                    download=True,
+                )
+                if eml_content:
+                    if isinstance(eml_content, str):
+                        eml_content = eml_content.encode("utf-8")
+                    attachments.append(
+                        FindingAttachment(
+                            file_name=f"{subject[:50]}.eml",
+                            data=eml_content,
+                        )
+                    )
+            except Exception as e:
+                logger.warning(f"Failed to fetch email EML: {e}")
+
+            yield Finding(
+                rule_title=f"Email: {subject[:100]}"
+                if subject
+                else f"Email ID: {email_id}",
+                attachments=attachments if attachments else None,
+            )
+
+            emails_processed += 1
+
+        next_link = resp.get("@odata.nextLink")
+        if not next_link or emails_processed >= max_emails:
+            break
+        api_params = None
+        resp = helper.make_rest_call_helper(endpoint, nextLink=next_link)
+
+    if latest_time:
+        state["es_last_time"] = latest_time
+        state["es_first_run"] = False
+        if hasattr(asset, "ingest_state"):
+            asset.ingest_state.put_all(state)
+
+    logger.info(f"Processed {emails_processed} emails for ES findings")
+
+
 # Import action modules to register them with the app
 from .actions import (  # noqa: F401
     block_sender,