Export event to Splunk HEC. Args: event: Event to export Returns: True if successful
(self, event: SIEMEvent)
| 42 | |
| 43 | async def export(self, event: SIEMEvent) -> bool: |
| 44 | """Export event to Splunk HEC. |
| 45 | |
| 46 | Args: |
| 47 | event: Event to export |
| 48 | |
| 49 | Returns: |
| 50 | True if successful |
| 51 | """ |
| 52 | url = f"{self.hec_url}/services/collector/event" |
| 53 | |
| 54 | # Build HEC event payload |
| 55 | payload = { |
| 56 | "time": event.timestamp.timestamp(), |
| 57 | "host": event.target or "security-suite", |
| 58 | "source": self.source, |
| 59 | "sourcetype": self.sourcetype, |
| 60 | "index": self.index, |
| 61 | "event": event.to_dict(), |
| 62 | } |
| 63 | |
| 64 | headers = { |
| 65 | "Authorization": f"Splunk {self.hec_token}", |
| 66 | "Content-Type": "application/json", |
| 67 | } |
| 68 | |
| 69 | try: |
| 70 | async with httpx.AsyncClient( |
| 71 | timeout=self.timeout, verify=self.verify_ssl |
| 72 | ) as client: |
| 73 | response = await client.post( |
| 74 | url, |
| 75 | json=payload, |
| 76 | headers=headers, |
| 77 | ) |
| 78 | |
| 79 | if response.status_code == 200: |
| 80 | self.logger.debug(f"Event exported to Splunk: {event.event_type}") |
| 81 | return True |
| 82 | else: |
| 83 | self.logger.error( |
| 84 | f"Splunk HEC error: {response.status_code} - {response.text}" |
| 85 | ) |
| 86 | return False |
| 87 | |
| 88 | except Exception as e: |
| 89 | self.logger.error(f"Splunk export failed: {e}") |
| 90 | return False |
| 91 | |
| 92 | async def export_batch(self, events: list[SIEMEvent]) -> tuple[int, int]: |
| 93 | """Export multiple events in a single request. |
| 94 |