Class SplunkExporter

modules/siem/splunk.py:11–181

Export events to Splunk via HEC.

class SplunkExporter(SIEMExporter):
    """Export events to Splunk via HEC."""

    def __init__(
        self,
        hec_url: str,
        hec_token: str,
        index: str = "main",
        source: str = "security-suite",
        sourcetype: str = "security:scan",
        verify_ssl: bool = True,
        timeout: float = 10.0,
    ):
        """Initialize Splunk exporter.

        Args:
            hec_url: Splunk HEC URL (e.g., https://splunk:8088)
            hec_token: HEC authentication token
            index: Splunk index to send events to
            source: Event source identifier
            sourcetype: Event sourcetype
            verify_ssl: Whether to verify SSL certificates
            timeout: Request timeout in seconds
        """
        super().__init__()
        self.hec_url = hec_url.rstrip("/")
        self.hec_token = hec_token
        self.index = index
        self.source = source
        self.sourcetype = sourcetype
        self.verify_ssl = verify_ssl
        self.timeout = timeout

    async def export(self, event: SIEMEvent) -> bool:
        """Export event to Splunk HEC.

        Args:
            event: Event to export

        Returns:
            True if successful
        """
        url = f"{self.hec_url}/services/collector/event"

        # Build HEC event payload
        payload = {
            "time": event.timestamp.timestamp(),
            "host": event.target or "security-suite",
            "source": self.source,
            "sourcetype": self.sourcetype,
            "index": self.index,
            "event": event.to_dict(),
        }

        headers = {
            "Authorization": f"Splunk {self.hec_token}",
            "Content-Type": "application/json",
        }