feat(schema): add Alert -> enodia.event.v1 reference wrapper
This commit is contained in:
parent
54bd8ff1f4
commit
65f5be6420
3 changed files with 22 additions and 0 deletions
|
|
@ -29,6 +29,9 @@ Required fields:
|
||||||
| `host` | string | Hostname at emission time. |
|
| `host` | string | Hostname at emission time. |
|
||||||
| `<event_type>` | object | Payload, under a key equal to `event_type` (e.g. `alert` holds an `enodia.alert.v1` object). |
|
| `<event_type>` | object | Payload, under a key equal to `event_type` (e.g. `alert` holds an `enodia.alert.v1` object). |
|
||||||
|
|
||||||
|
The reference encoder lives in `enodia_sentinel/event.py`
|
||||||
|
(`build_event`, `from_alert`); `tests/test_event_envelope.py` pins the contract.
|
||||||
|
|
||||||
## `enodia.alert.v1`
|
## `enodia.alert.v1`
|
||||||
|
|
||||||
Single detection object, used inside alert snapshots and by commands such as
|
Single detection object, used inside alert snapshots and by commands such as
|
||||||
|
|
|
||||||
|
|
@ -44,3 +44,8 @@ def build_event(
|
||||||
"host": host,
|
"host": host,
|
||||||
event_type: payload,
|
event_type: payload,
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|
||||||
|
def from_alert(alert, *, host: str | None = None, timestamp: str | None = None) -> dict:
|
||||||
|
"""Wrap an ``Alert`` (via its ``to_dict()``) in an ``alert`` event envelope."""
|
||||||
|
return build_event("alert", alert.to_dict(), host=host, timestamp=timestamp)
|
||||||
|
|
|
||||||
|
|
@ -42,6 +42,20 @@ class TestEventEnvelope(unittest.TestCase):
|
||||||
def test_event_types_is_closed_set(self):
|
def test_event_types_is_closed_set(self):
|
||||||
self.assertEqual(event.EVENT_TYPES, frozenset({"alert", "incident", "status"}))
|
self.assertEqual(event.EVENT_TYPES, frozenset({"alert", "incident", "status"}))
|
||||||
|
|
||||||
|
def test_from_alert_wraps_alert_payload(self):
|
||||||
|
from enodia_sentinel.alert import Alert, Severity
|
||||||
|
alert = Alert(
|
||||||
|
Severity.CRITICAL, "reverse_shell", "rsh:4242",
|
||||||
|
"pid=4242 peer=[8.8.8.8:4444]", (4242,),
|
||||||
|
sid=100010, classtype="command-and-control")
|
||||||
|
env = event.from_alert(alert, host="host-a")
|
||||||
|
self.assertEqual(env["event_type"], "alert")
|
||||||
|
self.assertEqual(env["host"], "host-a")
|
||||||
|
# Payload is the untouched enodia.alert.v1 object.
|
||||||
|
self.assertEqual(env["alert"], alert.to_dict())
|
||||||
|
self.assertEqual(env["alert"]["signature"], "reverse_shell")
|
||||||
|
self.assertEqual(env["alert"]["sid"], 100010)
|
||||||
|
|
||||||
|
|
||||||
if __name__ == "__main__":
|
if __name__ == "__main__":
|
||||||
unittest.main()
|
unittest.main()
|
||||||
|
|
|
||||||
Loading…
Add table
Add a link
Reference in a new issue