anti
bf654b9aed
Shell-session behavioral observation registry layered on core. SPDX: GPL-3.0-or-later (code) / CC-BY-SA-4.0 (attribution-recipes.md).
59 lines
2.2 KiB
Python
59 lines
2.2 KiB
Python
# SPDX-License-Identifier: GPL-3.0-or-later
|
|
"""DECNET bus interop. Aligns BEHAVE Observation with DECNET Event payload shape.
|
|
|
|
DECNET's Event (decnet/bus/base.py:26) carries ``(topic, payload, type, v, ts, id)``.
|
|
A BEHAVE Observation maps onto that envelope as follows:
|
|
|
|
topic = "attacker.observation." + observation.primitive
|
|
payload = observation.model_dump(exclude={"id", "ts", "v"})
|
|
type = observation.primitive
|
|
v = observation.v
|
|
ts = observation.ts
|
|
id = observation.id
|
|
|
|
The publisher must set ``topic`` from the primitive when calling ``bus.publish()``;
|
|
DECNET's bus does not trust topic from the wire (anti-spoofing, base.py:60-76).
|
|
|
|
This module does NOT import DECNET. The adapter speaks dicts; consumers wire it
|
|
to their own bus.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
from typing import Any
|
|
|
|
from .envelope import Observation
|
|
|
|
TOPIC_PREFIX: str = "attacker.observation"
|
|
|
|
|
|
def event_topic_for(primitive: str) -> str:
|
|
"""Return the canonical DECNET bus topic for a BEHAVE primitive."""
|
|
return f"{TOPIC_PREFIX}.{primitive}"
|
|
|
|
|
|
def to_event_payload(obs: Observation) -> dict[str, Any]:
|
|
"""Project an Observation into a dict suitable for ``Event.payload``.
|
|
|
|
Excludes ``id``, ``ts``, and ``v`` because those are carried at the Event
|
|
envelope level by DECNET, not in the payload body.
|
|
"""
|
|
return obs.model_dump(exclude={"id", "ts", "v"}, mode="json")
|
|
|
|
|
|
def from_event_payload(primitive: str, payload: dict[str, Any]) -> Observation:
|
|
"""Reconstruct an Observation from ``(topic-derived primitive, Event.payload)``.
|
|
|
|
The ``primitive`` argument is the trailing segment of the bus topic, NOT a
|
|
field read from the payload — relying on the wire-side ``primitive`` field
|
|
would let a misbehaving publisher spoof observations on topics they don't
|
|
actually publish to. This mirrors DECNET's ``Event.from_dict`` discipline
|
|
(decnet/bus/base.py:60-76).
|
|
"""
|
|
if "primitive" in payload and payload["primitive"] != primitive:
|
|
raise ValueError(
|
|
f"payload.primitive ({payload['primitive']!r}) does not match "
|
|
f"topic-derived primitive ({primitive!r}); refusing to reconstruct"
|
|
)
|
|
return Observation.model_validate({**payload, "primitive": primitive})
|