295 lines
13 KiB
Markdown
295 lines
13 KiB
Markdown
# EVE-Style Event Envelope (Phase 0) Implementation Plan
|
|
|
|
> **For agentic workers:** REQUIRED SUB-SKILL: Use superpowers:subagent-driven-development (recommended) or superpowers:executing-plans to implement this plan task-by-task. Steps use checkbox (`- [ ]`) syntax for tracking.
|
|
|
|
**Goal:** Add an additive, EVE-style event envelope (`enodia.event.v1`) — a single typed wrapper with an `event_type` discriminator — as the reference contract that both today's Python and a future Go agent will emit.
|
|
|
|
**Architecture:** A new pure `enodia_sentinel/event.py` module provides a reference encoder that wraps any payload in a common envelope (`schema`, `event_type`, `timestamp`, `host`, and a payload sub-object keyed by the event type — exactly like Suricata EVE). It is *additive*: it does not change any existing emitter or break existing v1 contracts. Existing objects (starting with `Alert`) get a thin convenience wrapper. Compatibility tests and `SCHEMAS.md` pin the contract.
|
|
|
|
**Tech Stack:** Python 3 standard library only (`dataclasses`/functions, `datetime`, `os`), `unittest`.
|
|
|
|
## Global Constraints
|
|
|
|
- **Stdlib-only.** No third-party imports in `enodia_sentinel/`. (Copied from CLAUDE.md engineering rules.)
|
|
- **Additive v1.** Required v1 fields are never removed, renamed, or retyped; new work only adds. (Copied from `docs/SCHEMAS.md`.)
|
|
- **Every operator-facing JSON contract updates `docs/SCHEMAS.md` and compatibility tests.** (Copied from CLAUDE.md documentation checklist.)
|
|
- **SPDX header** `# SPDX-License-Identifier: GPL-3.0-or-later` as the first line of every new `.py` file. (Matches every existing module.)
|
|
- **Shared checkout:** stage only files this plan creates/modifies; never `git add -A`. (Copied from CLAUDE.md.)
|
|
- **Timestamp format:** `datetime.now().astimezone().isoformat()`. **Host:** `os.uname().nodename`. (Existing convention in `snapshot.py`.)
|
|
- **Test command:** `python3 -m unittest <module> -v` from the repo root.
|
|
|
|
---
|
|
|
|
### Task 1: Event envelope reference encoder
|
|
|
|
**Files:**
|
|
- Modify: `enodia_sentinel/schemas.py` (add one constant)
|
|
- Create: `enodia_sentinel/event.py`
|
|
- Create: `tests/test_event_envelope.py`
|
|
- Modify: `docs/SCHEMAS.md` (add `enodia.event.v1` section)
|
|
|
|
**Interfaces:**
|
|
- Consumes: nothing (leaf module).
|
|
- Produces:
|
|
- `enodia_sentinel.schemas.EVENT_V1: str == "enodia.event.v1"`
|
|
- `enodia_sentinel.event.EVENT_TYPES: frozenset[str]` — the closed v1 set `{"alert", "incident", "status"}`
|
|
- `enodia_sentinel.event.build_event(event_type: str, payload: dict, *, host: str | None = None, timestamp: str | None = None) -> dict` — returns an envelope dict with keys `schema`, `event_type`, `timestamp`, `host`, and `event_type`'s value used as a key holding `payload`. Raises `ValueError` for an `event_type` not in `EVENT_TYPES`.
|
|
|
|
- [ ] **Step 1: Write the failing test**
|
|
|
|
Create `tests/test_event_envelope.py`:
|
|
|
|
```python
|
|
# SPDX-License-Identifier: GPL-3.0-or-later
|
|
"""Compatibility tests for the enodia.event.v1 envelope."""
|
|
import os
|
|
import unittest
|
|
from datetime import datetime
|
|
|
|
from enodia_sentinel import event, schemas
|
|
|
|
|
|
class TestEventEnvelope(unittest.TestCase):
|
|
def test_schema_id_is_stable(self):
|
|
self.assertEqual(schemas.EVENT_V1, "enodia.event.v1")
|
|
|
|
def test_build_event_shape_and_types(self):
|
|
env = event.build_event("alert", {"sid": 100010})
|
|
self.assertEqual(env["schema"], schemas.EVENT_V1)
|
|
self.assertEqual(env["event_type"], "alert")
|
|
self.assertIsInstance(env["timestamp"], str)
|
|
self.assertIsInstance(env["host"], str)
|
|
# EVE-style: payload lives under a key named by event_type.
|
|
self.assertEqual(env["alert"], {"sid": 100010})
|
|
|
|
def test_timestamp_defaults_to_iso8601(self):
|
|
env = event.build_event("status", {"running": True})
|
|
# Parses as ISO-8601 without raising.
|
|
datetime.fromisoformat(env["timestamp"])
|
|
|
|
def test_host_defaults_to_nodename(self):
|
|
env = event.build_event("status", {"running": True})
|
|
self.assertEqual(env["host"], os.uname().nodename)
|
|
|
|
def test_overrides_are_respected(self):
|
|
env = event.build_event(
|
|
"incident", {"id": "x"}, host="host-a", timestamp="2026-07-10T00:00:00-07:00")
|
|
self.assertEqual(env["host"], "host-a")
|
|
self.assertEqual(env["timestamp"], "2026-07-10T00:00:00-07:00")
|
|
|
|
def test_unknown_event_type_rejected(self):
|
|
with self.assertRaises(ValueError):
|
|
event.build_event("nope", {})
|
|
|
|
def test_event_types_is_closed_set(self):
|
|
self.assertEqual(event.EVENT_TYPES, frozenset({"alert", "incident", "status"}))
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|
|
```
|
|
|
|
- [ ] **Step 2: Run test to verify it fails**
|
|
|
|
Run: `python3 -m unittest tests.test_event_envelope -v`
|
|
Expected: FAIL — `ModuleNotFoundError: No module named 'enodia_sentinel.event'` (and `AttributeError` on `schemas.EVENT_V1`).
|
|
|
|
- [ ] **Step 3: Add the schema constant**
|
|
|
|
In `enodia_sentinel/schemas.py`, after the `RECONCILE_V1` line (currently the last constant), add:
|
|
|
|
```python
|
|
EVENT_V1 = "enodia.event.v1"
|
|
```
|
|
|
|
- [ ] **Step 4: Write the module**
|
|
|
|
Create `enodia_sentinel/event.py`:
|
|
|
|
```python
|
|
# SPDX-License-Identifier: GPL-3.0-or-later
|
|
"""EVE-style event envelope: one typed wrapper for every emitted record.
|
|
|
|
Every envelope carries a stable ``schema`` id, an ``event_type`` discriminator,
|
|
a ``timestamp`` and ``host``, and the type-specific payload under a key named by
|
|
the event type (mirroring Suricata's EVE JSON, where an ``alert`` record holds
|
|
its detail under ``"alert"``). This is the reference contract a future Go agent
|
|
must emit byte-for-byte; the Python side is the oracle.
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import os
|
|
from datetime import datetime
|
|
|
|
from .schemas import EVENT_V1
|
|
|
|
# Closed set of v1 discriminators. New types are added additively in later
|
|
# phases (e.g. "dns", "tls", "http", "flow", "anomaly", "capture"); removing or
|
|
# renaming a value is a breaking change.
|
|
EVENT_TYPES = frozenset({"alert", "incident", "status"})
|
|
|
|
|
|
def build_event(
|
|
event_type: str,
|
|
payload: dict,
|
|
*,
|
|
host: str | None = None,
|
|
timestamp: str | None = None,
|
|
) -> dict:
|
|
"""Wrap ``payload`` in an ``enodia.event.v1`` envelope.
|
|
|
|
Raises ``ValueError`` if ``event_type`` is not a known v1 discriminator.
|
|
"""
|
|
if event_type not in EVENT_TYPES:
|
|
raise ValueError(f"unknown event_type: {event_type!r}")
|
|
if host is None:
|
|
host = os.uname().nodename
|
|
if timestamp is None:
|
|
timestamp = datetime.now().astimezone().isoformat()
|
|
return {
|
|
"schema": EVENT_V1,
|
|
"event_type": event_type,
|
|
"timestamp": timestamp,
|
|
"host": host,
|
|
event_type: payload,
|
|
}
|
|
```
|
|
|
|
- [ ] **Step 5: Run test to verify it passes**
|
|
|
|
Run: `python3 -m unittest tests.test_event_envelope -v`
|
|
Expected: PASS (6 tests).
|
|
|
|
- [ ] **Step 6: Document the contract**
|
|
|
|
In `docs/SCHEMAS.md`, add this section immediately after the intro paragraph that ends "...pin the required v1 fields." (before the `## enodia.alert.v1` heading):
|
|
|
|
```markdown
|
|
## `enodia.event.v1`
|
|
|
|
Uniform envelope that wraps every emitted record with a typed discriminator.
|
|
This is the Python↔Go interoperability contract: any implementation must emit
|
|
the same envelope for the same input. Modeled on Suricata's EVE JSON — the
|
|
type-specific payload lives under a key named by `event_type`.
|
|
|
|
Required fields:
|
|
|
|
| Field | Type | Meaning |
|
|
|---|---|---|
|
|
| `schema` | string | `enodia.event.v1`. |
|
|
| `event_type` | string | Discriminator: `alert`, `incident`, or `status` (additive set). |
|
|
| `timestamp` | string | ISO-8601 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). |
|
|
```
|
|
|
|
- [ ] **Step 7: Commit**
|
|
|
|
```bash
|
|
git add enodia_sentinel/schemas.py enodia_sentinel/event.py tests/test_event_envelope.py docs/SCHEMAS.md
|
|
git commit -m "feat(schema): add enodia.event.v1 EVE-style event envelope"
|
|
```
|
|
|
|
---
|
|
|
|
### Task 2: Alert → event convenience wrapper
|
|
|
|
**Files:**
|
|
- Modify: `enodia_sentinel/event.py`
|
|
- Modify: `tests/test_event_envelope.py`
|
|
- Modify: `docs/SCHEMAS.md` (one clarifying line)
|
|
|
|
**Interfaces:**
|
|
- Consumes: `event.build_event` (Task 1); `enodia_sentinel.alert.Alert` with its existing `.to_dict()` method returning keys `sid, severity, signature, classtype, key, detail, pids`.
|
|
- Produces: `enodia_sentinel.event.from_alert(alert, *, host: str | None = None, timestamp: str | None = None) -> dict` — an `enodia.event.v1` envelope with `event_type == "alert"` whose `alert` payload is exactly `alert.to_dict()`.
|
|
|
|
- [ ] **Step 1: Write the failing test**
|
|
|
|
Append to `tests/test_event_envelope.py` inside `TestEventEnvelope` (before the `if __name__` guard):
|
|
|
|
```python
|
|
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)
|
|
```
|
|
|
|
- [ ] **Step 2: Run test to verify it fails**
|
|
|
|
Run: `python3 -m unittest tests.test_event_envelope.TestEventEnvelope.test_from_alert_wraps_alert_payload -v`
|
|
Expected: FAIL — `AttributeError: module 'enodia_sentinel.event' has no attribute 'from_alert'`.
|
|
|
|
- [ ] **Step 3: Add the wrapper**
|
|
|
|
Append to `enodia_sentinel/event.py` (after `build_event`):
|
|
|
|
```python
|
|
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)
|
|
```
|
|
|
|
- [ ] **Step 4: Run test to verify it passes**
|
|
|
|
Run: `python3 -m unittest tests.test_event_envelope -v`
|
|
Expected: PASS (7 tests).
|
|
|
|
- [ ] **Step 5: Document the alert mapping**
|
|
|
|
In `docs/SCHEMAS.md`, in the `enodia.event.v1` table row for `<event_type>`, the text already notes `alert` holds an `enodia.alert.v1` object. Add one line directly beneath the table:
|
|
|
|
```markdown
|
|
The reference encoder lives in `enodia_sentinel/event.py`
|
|
(`build_event`, `from_alert`); `tests/test_event_envelope.py` pins the contract.
|
|
```
|
|
|
|
- [ ] **Step 6: Commit**
|
|
|
|
```bash
|
|
git add enodia_sentinel/event.py tests/test_event_envelope.py docs/SCHEMAS.md
|
|
git commit -m "feat(schema): add Alert -> enodia.event.v1 reference wrapper"
|
|
```
|
|
|
|
---
|
|
|
|
### Task 3: Full-suite regression check
|
|
|
|
**Files:** none (verification only).
|
|
|
|
**Interfaces:** none.
|
|
|
|
- [ ] **Step 1: Run the whole suite to confirm nothing regressed**
|
|
|
|
Run: `python3 -m unittest discover -s tests -v`
|
|
Expected: PASS — the existing suite plus the 7 new envelope tests. No existing schema-contract test changes behavior (the work is purely additive).
|
|
|
|
- [ ] **Step 2: Confirm the additive guarantee held**
|
|
|
|
Run: `git diff --stat HEAD~2 -- enodia_sentinel/`
|
|
Expected: only `schemas.py` (one added line) and the new `event.py` appear — no existing emitter (`alert.py`, `snapshot.py`, `incident.py`, `web.py`) was modified, proving the change is non-breaking.
|
|
|
|
---
|
|
|
|
## Self-Review
|
|
|
|
**Spec coverage (Phase 0 = "Freeze and extend the v1 EVE-style event envelope in SCHEMAS.md; `event_type` discriminator; alert/incident/status as event types; the Python↔Go contract; no language change yet"):**
|
|
- EVE-style envelope with `event_type` discriminator → Task 1 (`build_event`, `EVENT_TYPES`).
|
|
- alert/incident/status as event types → Task 1 (`EVENT_TYPES = {"alert","incident","status"}`); alert reference mapping → Task 2 (`from_alert`). Incident/status wrappers are deferred to their emitter phases; the discriminator set already reserves them additively, satisfying "as event types" at the contract level.
|
|
- Documented in `SCHEMAS.md` → Tasks 1 & 2.
|
|
- Compatibility tests pin the contract → Tasks 1 & 2; regression proven → Task 3.
|
|
- No language change → confirmed; all Python, additive.
|
|
|
|
**Placeholder scan:** none — every code and doc step shows exact content.
|
|
|
|
**Type consistency:** `build_event(event_type, payload, *, host, timestamp)` and `from_alert(alert, *, host, timestamp)` signatures match between the interface blocks, the code steps, and the tests. `EVENT_V1`/`EVENT_TYPES` names are consistent across schemas.py, event.py, and tests. Envelope keys (`schema`, `event_type`, `timestamp`, `host`, `<event_type>`) are identical in code, tests, and the SCHEMAS.md table.
|
|
|
|
**Scope:** single subsystem, additive, independently testable; no decomposition needed.
|