Forwarders

Publishing captured protocol exchanges to a monitoring system.

py20305.forwarders.types

The message format this client publishes captured traffic in.

:class:ProtocolMessage wraps one IEEE 2030.5 exchange -- request or response, with its endpoints, timing and payload -- in a protocol-agnostic envelope. The forwarder publishes these; a monitoring system subscribes to them. Nothing in the client's own operation depends on this module, so a deployment that does not forward anything never touches it.

The envelope is deliberately more general than 2030.5 needs, because it is consumed by systems that also ingest other protocols and want one shape for all of them. That is why :class:ProtocolMetadata carries a free-form extra dict and why every optional field is omitted from the serialized form rather than emitted as null: consumers match on presence, and adding a field is a compatible change.

Serialization contract, as emitted by :meth:ProtocolMessage.to_dict:

• version, protocol, direction, timestamp, client_id, forwarder_id, payload, source, hash and is_valid are always present.
• destination, protocol_data and validation_error appear only when set.
• timestamp is ISO 8601 with an offset. A trailing Z is rewritten to +00:00 on construction.
• hash is a deterministic UUIDv3 over protocol, client id, timestamp and payload, so a consumer can deduplicate replays without coordinating with the producer.

WireDirection

Bases: StrEnum

Which way the message was travelling, as carried on the wire.

UPSTREAM is device toward server; DOWNSTREAM is server toward device.

Deliberately not named MessageDirection: :mod:py20305.forwarders.base defines a capture-side enum by that name with the same members, and two same-named enums compare unequal across modules without any error to say so. Distinct names make the conversion at the boundary explicit.

Source code in src/py20305/forwarders/types.py

class WireDirection(StrEnum):
    """Which way the message was travelling, as carried on the wire.

    ``UPSTREAM`` is device toward server; ``DOWNSTREAM`` is server toward
    device.

    Deliberately not named ``MessageDirection``:
    :mod:`py20305.forwarders.base` defines a capture-side enum by
    that name with the same members, and two same-named enums compare unequal
    across modules without any error to say so. Distinct names make the
    conversion at the boundary explicit.
    """

    UPSTREAM = "upstream"
    DOWNSTREAM = "downstream"

Protocol

Bases: StrEnum

The application protocol a message was captured from.

Source code in src/py20305/forwarders/types.py

class Protocol(StrEnum):
    """The application protocol a message was captured from."""

    IEEE_2030_5 = "2030.5"
    #: Southbound device traffic. A consumer reading this channel tells the
    #: two apart by this field, which is why device telemetry rides the same
    #: envelope rather than defining a second one.
    MODBUS = "modbus"
    #: A southbound exchange over something this vocabulary does not name --
    #: a custom or in-process connector. Recording it as `modbus` would be a
    #: claim about the wire that is simply untrue, and a consumer filtering on
    #: protocol would then act on a device that speaks something else. The
    #: wire value is ``generic``, matching the consumer contract's catch-all;
    #: any other spelling fails enum parsing and schema validation on the
    #: receiving side.
    GENERIC = "generic"
    #: Deprecated alias for :attr:`GENERIC`, kept so source written against
    #: 0.2.x keeps importing -- the *wire* spelling was always the member
    #: value, so nothing serialized changes and ``from_string("other")``
    #: still rejects the obsolete wire value.
    OTHER = "generic"

    @classmethod
    def from_string(cls, value: str) -> Protocol:
        """Parse a protocol from its wire value."""
        normalized = value.lower().strip()
        for member in cls:
            if member.value.lower() == normalized:
                return member
        raise ValueError(f"Unknown protocol: {value}")

from_string classmethod

from_string(value: str) -> Protocol

Parse a protocol from its wire value.

Source code in src/py20305/forwarders/types.py

@classmethod
def from_string(cls, value: str) -> Protocol:
    """Parse a protocol from its wire value."""
    normalized = value.lower().strip()
    for member in cls:
        if member.value.lower() == normalized:
            return member
    raise ValueError(f"Unknown protocol: {value}")

PayloadEnvelope dataclass

A payload together with what it is, so a consumer need not guess.

2030.5 bodies are XML, but a forwarded message may also carry JSON or an opaque blob, and the consumer has to tell them apart without sniffing. Construct through the from_* factories rather than directly.

Source code in src/py20305/forwarders/types.py

@dataclass
class PayloadEnvelope:
    """A payload together with what it is, so a consumer need not guess.

    2030.5 bodies are XML, but a forwarded message may also carry JSON or an
    opaque blob, and the consumer has to tell them apart without sniffing.
    Construct through the ``from_*`` factories rather than directly.
    """

    #: Common content types, as class constants so callers don't retype them.
    JSON = "application/json"
    XML = "application/xml"
    TEXT = "text/plain"
    BINARY = "application/octet-stream"

    content_type: str
    data: str
    encoding: str = "utf-8"

    @classmethod
    def from_dict(cls, data: dict[str, Any]) -> PayloadEnvelope:
        """Wrap a Python dict as a JSON payload."""
        return cls(content_type=cls.JSON, data=json.dumps(data), encoding="utf-8")

    @classmethod
    def from_xml(cls, xml_string: str) -> PayloadEnvelope:
        """Wrap an XML string. The usual case for 2030.5."""
        return cls(content_type=cls.XML, data=xml_string, encoding="utf-8")

    @classmethod
    def from_binary(cls, binary_data: bytes) -> PayloadEnvelope:
        """Wrap raw bytes, base64-encoded so the envelope stays JSON-safe."""
        return cls(
            content_type=cls.BINARY,
            data=base64.b64encode(binary_data).decode("ascii"),
            encoding="base64",
        )

    @classmethod
    def from_text(cls, text: str, content_type: str = "text/plain") -> PayloadEnvelope:
        """Wrap plain text, or text of a caller-specified content type."""
        return cls(content_type=content_type, data=text, encoding="utf-8")

    @classmethod
    def infer_from_string(cls, data: str) -> PayloadEnvelope:
        """Guess the content type of a string payload.

        Tries JSON, then XML, then falls back to plain text. Only for callers
        holding a string with no type information; prefer a ``from_*``
        factory whenever the type is known.
        """
        stripped = data.strip()

        if stripped.startswith(("{", "[")):
            try:
                json.loads(stripped)
                return cls(content_type=cls.JSON, data=data, encoding="utf-8")
            except json.JSONDecodeError:
                pass

        if stripped.startswith("<?xml") or (stripped.startswith("<") and ">" in stripped):
            return cls(content_type=cls.XML, data=data, encoding="utf-8")

        return cls(content_type=cls.TEXT, data=data, encoding="utf-8")

    def to_dict(self) -> dict[str, Any] | None:
        """Parse a JSON payload back to a dict, or ``None`` if it isn't one."""
        if self.content_type == self.JSON:
            result = json.loads(self.data)
            if isinstance(result, dict):
                return result
        return None

    def to_bytes(self) -> bytes:
        """Return the payload as bytes, decoding base64 when that's the encoding."""
        if self.encoding == "base64":
            return base64.b64decode(self.data)
        return self.data.encode(self.encoding)

    def to_string(self) -> str:
        """Return the payload's raw string form."""
        return self.data

    @property
    def is_json(self) -> bool:
        """Whether this payload is JSON."""
        return self.content_type == self.JSON

    @property
    def is_xml(self) -> bool:
        """Whether this payload is XML."""
        return self.content_type == self.XML

    @property
    def is_binary(self) -> bool:
        """Whether this payload is an opaque blob."""
        return self.content_type == self.BINARY

    @property
    def is_text(self) -> bool:
        """Whether this payload is plain text."""
        return self.content_type == self.TEXT

    def serialize(self) -> dict[str, str]:
        """Serialize for transport."""
        return {
            "content_type": self.content_type,
            "data": self.data,
            "encoding": self.encoding,
        }

    @classmethod
    def deserialize(cls, envelope_dict: dict[str, str]) -> PayloadEnvelope:
        """Rebuild from a :meth:`serialize` result."""
        return cls(
            content_type=envelope_dict["content_type"],
            data=envelope_dict["data"],
            encoding=envelope_dict.get("encoding", "utf-8"),
        )

is_json property

is_json: bool

Whether this payload is JSON.

is_xml property

is_xml: bool

Whether this payload is XML.

is_binary property

is_binary: bool

Whether this payload is an opaque blob.

is_text property

is_text: bool

Whether this payload is plain text.

from_dict classmethod

from_dict(data: dict[str, Any]) -> PayloadEnvelope

Wrap a Python dict as a JSON payload.

Source code in src/py20305/forwarders/types.py

@classmethod
def from_dict(cls, data: dict[str, Any]) -> PayloadEnvelope:
    """Wrap a Python dict as a JSON payload."""
    return cls(content_type=cls.JSON, data=json.dumps(data), encoding="utf-8")

from_xml classmethod

from_xml(xml_string: str) -> PayloadEnvelope

Wrap an XML string. The usual case for 2030.5.

Source code in src/py20305/forwarders/types.py

@classmethod
def from_xml(cls, xml_string: str) -> PayloadEnvelope:
    """Wrap an XML string. The usual case for 2030.5."""
    return cls(content_type=cls.XML, data=xml_string, encoding="utf-8")

from_binary classmethod

from_binary(binary_data: bytes) -> PayloadEnvelope

Wrap raw bytes, base64-encoded so the envelope stays JSON-safe.

Source code in src/py20305/forwarders/types.py

@classmethod
def from_binary(cls, binary_data: bytes) -> PayloadEnvelope:
    """Wrap raw bytes, base64-encoded so the envelope stays JSON-safe."""
    return cls(
        content_type=cls.BINARY,
        data=base64.b64encode(binary_data).decode("ascii"),
        encoding="base64",
    )

from_text classmethod

from_text(text: str, content_type: str = 'text/plain') -> PayloadEnvelope

Wrap plain text, or text of a caller-specified content type.

Source code in src/py20305/forwarders/types.py

@classmethod
def from_text(cls, text: str, content_type: str = "text/plain") -> PayloadEnvelope:
    """Wrap plain text, or text of a caller-specified content type."""
    return cls(content_type=content_type, data=text, encoding="utf-8")

infer_from_string classmethod

infer_from_string(data: str) -> PayloadEnvelope

Guess the content type of a string payload.

Tries JSON, then XML, then falls back to plain text. Only for callers holding a string with no type information; prefer a from_* factory whenever the type is known.

Source code in src/py20305/forwarders/types.py

@classmethod
def infer_from_string(cls, data: str) -> PayloadEnvelope:
    """Guess the content type of a string payload.

    Tries JSON, then XML, then falls back to plain text. Only for callers
    holding a string with no type information; prefer a ``from_*``
    factory whenever the type is known.
    """
    stripped = data.strip()

    if stripped.startswith(("{", "[")):
        try:
            json.loads(stripped)
            return cls(content_type=cls.JSON, data=data, encoding="utf-8")
        except json.JSONDecodeError:
            pass

    if stripped.startswith("<?xml") or (stripped.startswith("<") and ">" in stripped):
        return cls(content_type=cls.XML, data=data, encoding="utf-8")

    return cls(content_type=cls.TEXT, data=data, encoding="utf-8")

to_dict

to_dict() -> dict[str, Any] | None

Parse a JSON payload back to a dict, or None if it isn't one.

Source code in src/py20305/forwarders/types.py

def to_dict(self) -> dict[str, Any] | None:
    """Parse a JSON payload back to a dict, or ``None`` if it isn't one."""
    if self.content_type == self.JSON:
        result = json.loads(self.data)
        if isinstance(result, dict):
            return result
    return None

to_bytes

to_bytes() -> bytes

Return the payload as bytes, decoding base64 when that's the encoding.

Source code in src/py20305/forwarders/types.py

def to_bytes(self) -> bytes:
    """Return the payload as bytes, decoding base64 when that's the encoding."""
    if self.encoding == "base64":
        return base64.b64decode(self.data)
    return self.data.encode(self.encoding)

to_string

to_string() -> str

Return the payload's raw string form.

Source code in src/py20305/forwarders/types.py

def to_string(self) -> str:
    """Return the payload's raw string form."""
    return self.data

serialize

serialize() -> dict[str, str]

Serialize for transport.

Source code in src/py20305/forwarders/types.py

def serialize(self) -> dict[str, str]:
    """Serialize for transport."""
    return {
        "content_type": self.content_type,
        "data": self.data,
        "encoding": self.encoding,
    }

deserialize classmethod

deserialize(envelope_dict: dict[str, str]) -> PayloadEnvelope

Rebuild from a :meth:serialize result.

Source code in src/py20305/forwarders/types.py

@classmethod
def deserialize(cls, envelope_dict: dict[str, str]) -> PayloadEnvelope:
    """Rebuild from a :meth:`serialize` result."""
    return cls(
        content_type=envelope_dict["content_type"],
        data=envelope_dict["data"],
        encoding=envelope_dict.get("encoding", "utf-8"),
    )

NetworkEndpoint dataclass

One end of the connection a message was seen on.

mac is carried because the format defines it and a capture-side producer populates it from the L2 frame. This client sees TCP sessions rather than frames, so it leaves it unset -- but a message read back from the wire has to preserve it rather than silently drop it.

Source code in src/py20305/forwarders/types.py

@dataclass
class NetworkEndpoint:
    """One end of the connection a message was seen on.

    ``mac`` is carried because the format defines it and a capture-side
    producer populates it from the L2 frame. This client sees TCP sessions
    rather than frames, so it leaves it unset -- but a message read back from
    the wire has to preserve it rather than silently drop it.
    """

    ip: str
    port: int
    mac: str | None = None

    def __post_init__(self) -> None:
        """Normalize the loosely-typed fields.

        A port may arrive as a string, since configuration often supplies one.
        A MAC is normalized to lowercase colon-separated form, accepting either
        separator, because the format defines one spelling and a consumer
        comparing addresses should not have to guess which it received.

        Raises:
            ValueError: If ``mac`` is set and is not six hex pairs.
        """
        if isinstance(self.port, str):
            self.port = int(self.port)
        if self.mac is not None:
            if not isinstance(self.mac, str) or not _MAC_RE.match(self.mac):
                raise ValueError(
                    f"mac must be six hex pairs separated by ':' or '-', got {self.mac!r}"
                )
            self.mac = self.mac.lower().replace("-", ":")

    def to_dict(self) -> dict[str, Any]:
        """Serialize for transport, omitting ``mac`` when unset."""
        result: dict[str, Any] = {"ip": self.ip, "port": self.port}
        if self.mac is not None:
            result["mac"] = self.mac
        return result

    @classmethod
    def from_dict(cls, data: dict[str, Any]) -> NetworkEndpoint:
        """Rebuild from a :meth:`to_dict` result."""
        return cls(ip=data["ip"], port=data["port"], mac=data.get("mac"))

to_dict

to_dict() -> dict[str, Any]

Serialize for transport, omitting mac when unset.

Source code in src/py20305/forwarders/types.py

def to_dict(self) -> dict[str, Any]:
    """Serialize for transport, omitting ``mac`` when unset."""
    result: dict[str, Any] = {"ip": self.ip, "port": self.port}
    if self.mac is not None:
        result["mac"] = self.mac
    return result

from_dict classmethod

from_dict(data: dict[str, Any]) -> NetworkEndpoint

Rebuild from a :meth:to_dict result.

Source code in src/py20305/forwarders/types.py

@classmethod
def from_dict(cls, data: dict[str, Any]) -> NetworkEndpoint:
    """Rebuild from a :meth:`to_dict` result."""
    return cls(ip=data["ip"], port=data["port"], mac=data.get("mac"))

ProtocolMetadata dataclass

Protocol-level detail about a message, beyond its bytes.

Every field is optional and omitted from the serialized form when unset, so a consumer reads what the producer knew rather than a wall of nulls. extra carries anything this class does not name.

Source code in src/py20305/forwarders/types.py

@dataclass
class ProtocolMetadata:
    """Protocol-level detail about a message, beyond its bytes.

    Every field is optional and omitted from the serialized form when unset,
    so a consumer reads what the producer knew rather than a wall of nulls.
    ``extra`` carries anything this class does not name.
    """

    lfdi: str | None = None
    message_type: str | None = None
    http_method: str | None = None
    uri: str | None = None

    #: The format's own free-form bag, round-tripped nested under ``extra``.
    extra: dict[str, Any] = field(default_factory=dict)

    #: Top-level keys this class does not model, kept where they were found.
    #:
    #: The format defines many fields for protocols this client does not
    #: speak, and modelling them here would be dead weight. But a message read
    #: from the wire and written back out has to keep its shape: folding those
    #: keys into ``extra`` would move them a level deeper and quietly rewrite
    #: another producer's message.
    passthrough: dict[str, Any] = field(default_factory=dict)

    def to_dict(self) -> dict[str, Any]:
        """Serialize, omitting unset fields and empty containers."""
        result: dict[str, Any] = {}
        for name in self.__dataclass_fields__:
            if name in {"extra", "passthrough"}:
                continue
            value = getattr(self, name)
            if value is not None:
                result[name] = value
        result.update(self.passthrough)
        if self.extra:
            result["extra"] = self.extra
        return result

    @classmethod
    def from_dict(cls, data: dict[str, Any]) -> ProtocolMetadata:
        """Rebuild from a :meth:`to_dict` result.

        Raises:
            TypeError: If ``extra`` is present and is not a mapping. Silently
                discarding it would lose data that the sender considered part
                of the message.
        """
        known = {f for f in cls.__dataclass_fields__ if f not in {"extra", "passthrough"}}
        raw_extra = data.get("extra")
        if raw_extra is not None and not isinstance(raw_extra, dict):
            raise TypeError(
                f"protocol_data 'extra' must be a mapping, got {type(raw_extra).__name__}"
            )

        kwargs: dict[str, Any] = {}
        passthrough: dict[str, Any] = {}
        for key, value in data.items():
            if key == "extra":
                continue
            if key in known:
                kwargs[key] = value
            else:
                passthrough[key] = value
        return cls(**kwargs, extra=dict(raw_extra or {}), passthrough=passthrough)

to_dict

to_dict() -> dict[str, Any]

Serialize, omitting unset fields and empty containers.

Source code in src/py20305/forwarders/types.py

def to_dict(self) -> dict[str, Any]:
    """Serialize, omitting unset fields and empty containers."""
    result: dict[str, Any] = {}
    for name in self.__dataclass_fields__:
        if name in {"extra", "passthrough"}:
            continue
        value = getattr(self, name)
        if value is not None:
            result[name] = value
    result.update(self.passthrough)
    if self.extra:
        result["extra"] = self.extra
    return result

from_dict classmethod

from_dict(data: dict[str, Any]) -> ProtocolMetadata

Rebuild from a :meth:to_dict result.

Raises:

Type	Description
TypeError	If extra is present and is not a mapping. Silently discarding it would lose data that the sender considered part of the message.

Source code in src/py20305/forwarders/types.py

@classmethod
def from_dict(cls, data: dict[str, Any]) -> ProtocolMetadata:
    """Rebuild from a :meth:`to_dict` result.

    Raises:
        TypeError: If ``extra`` is present and is not a mapping. Silently
            discarding it would lose data that the sender considered part
            of the message.
    """
    known = {f for f in cls.__dataclass_fields__ if f not in {"extra", "passthrough"}}
    raw_extra = data.get("extra")
    if raw_extra is not None and not isinstance(raw_extra, dict):
        raise TypeError(
            f"protocol_data 'extra' must be a mapping, got {type(raw_extra).__name__}"
        )

    kwargs: dict[str, Any] = {}
    passthrough: dict[str, Any] = {}
    for key, value in data.items():
        if key == "extra":
            continue
        if key in known:
            kwargs[key] = value
        else:
            passthrough[key] = value
    return cls(**kwargs, extra=dict(raw_extra or {}), passthrough=passthrough)

ProtocolMessage dataclass

One captured protocol exchange, ready to publish.

Construct it with whatever you have -- payload accepts a :class:PayloadEnvelope, a dict, a list or a string, and is normalized to an envelope on construction -- then call :meth:to_dict or :meth:to_json.

Source code in src/py20305/forwarders/types.py

@dataclass
class ProtocolMessage:
    """One captured protocol exchange, ready to publish.

    Construct it with whatever you have -- ``payload`` accepts a
    :class:`PayloadEnvelope`, a dict, a list or a string, and is normalized
    to an envelope on construction -- then call :meth:`to_dict` or
    :meth:`to_json`.
    """

    protocol: Protocol
    direction: WireDirection
    client_id: str
    payload: PayloadEnvelope | dict[str, Any] | list[Any] | str

    source: NetworkEndpoint
    destination: NetworkEndpoint | None = None

    #: Identifies the system that produced or relayed this message, so a
    #: consumer aggregating several sites can tell them apart.
    forwarder_id: str = ""

    protocol_data: ProtocolMetadata = field(default_factory=ProtocolMetadata)

    timestamp: str = field(default_factory=lambda: datetime.now(UTC).isoformat())
    hash: str = ""

    is_valid: bool = True
    validation_error: str | None = None

    def __post_init__(self) -> None:
        """Normalize the loosely-typed fields and compute the hash.

        Raises:
            ValueError: If ``client_id`` is empty or ``timestamp`` is not
                ISO 8601. Both are caught here rather than at the consumer,
                where the message would be undiagnosable.
        """
        if isinstance(self.protocol, str):
            self.protocol = Protocol.from_string(self.protocol)

        if isinstance(self.direction, str):
            self.direction = WireDirection(self.direction)

        if isinstance(self.payload, dict):
            self.payload = PayloadEnvelope.from_dict(self.payload)
        elif isinstance(self.payload, list):
            self.payload = PayloadEnvelope(
                content_type=PayloadEnvelope.JSON,
                data=json.dumps(self.payload),
                encoding="utf-8",
            )
        elif isinstance(self.payload, str):
            self.payload = PayloadEnvelope.infer_from_string(self.payload)

        if self.timestamp.endswith("Z"):
            self.timestamp = self.timestamp[:-1] + "+00:00"

        try:
            datetime.fromisoformat(self.timestamp)
        except (ValueError, TypeError) as exc:
            raise ValueError(
                "timestamp must be ISO 8601 format "
                f"(e.g. '2024-01-15T12:00:00+00:00'): {exc}"
            ) from exc

        if not self.client_id or not self.client_id.strip():
            raise ValueError("client_id must be a non-empty string")

        if self.forwarder_id:
            self.forwarder_id = self.forwarder_id.strip()

        if not self.hash:
            self.hash = self._generate_hash()

    def _generate_hash(self) -> str:
        """Derive a deduplication key from the message's identifying fields."""
        payload_str = (
            self.payload.data
            if isinstance(self.payload, PayloadEnvelope)
            else str(self.payload)
        )
        unique = f"{self.protocol.value}:{self.client_id}:{self.timestamp}:{payload_str}"
        return str(uuid.uuid3(uuid.NAMESPACE_DNS, unique))

    def to_dict(self) -> dict[str, Any]:
        """Serialize for transport. See the module docstring for the contract."""
        payload_data: dict[str, str] | dict[str, Any] | list[Any] | str
        if isinstance(self.payload, PayloadEnvelope):
            payload_data = self.payload.serialize()
        else:  # pragma: no cover -- __post_init__ normalizes every other shape
            payload_data = self.payload

        result: dict[str, Any] = {
            "version": _VERSION,
            "protocol": self.protocol.value,
            "direction": self.direction.value,
            "timestamp": self.timestamp,
            "client_id": self.client_id,
            "forwarder_id": self.forwarder_id,
            "payload": payload_data,
            "source": self.source.to_dict(),
            "hash": self.hash,
            "is_valid": self.is_valid,
        }

        if self.destination:
            result["destination"] = self.destination.to_dict()

        protocol_data = self.protocol_data.to_dict()
        if protocol_data:
            result["protocol_data"] = protocol_data

        if self.validation_error:
            result["validation_error"] = self.validation_error

        return result

    def to_json(self) -> str:
        """Serialize to a JSON string."""
        return json.dumps(self.to_dict())

    @classmethod
    def from_dict(cls, data: dict[str, Any]) -> ProtocolMessage:
        """Rebuild a message from a :meth:`to_dict` result.

        The inverse of the serialization contract, so a consumer can read
        what a producer wrote without hand-unpacking the envelope. Unknown
        keys under ``protocol_data`` survive the round trip via
        :attr:`ProtocolMetadata.extra`, which is what lets a consumer built
        against this version read a message from a later one.

        Metadata keys this class does not model are preserved in
        :attr:`ProtocolMetadata.passthrough` and re-emitted where they were
        found, so a message from a producer modelling more fields keeps its
        shape rather than having them folded a level deeper.

        ``hash`` is taken from the input rather than recomputed. It is the
        producer's deduplication key, and recomputing it here would silently
        change it whenever a field this class does not model was dropped.

        Raises:
            KeyError: If a key the contract says is always present is missing.
        """
        payload = data["payload"]
        if isinstance(payload, dict) and {"content_type", "data"} <= payload.keys():
            payload = PayloadEnvelope.deserialize(payload)

        destination = data.get("destination")
        protocol_data = data.get("protocol_data")

        return cls(
            protocol=Protocol.from_string(data["protocol"]),
            direction=WireDirection(data["direction"]),
            client_id=data["client_id"],
            payload=payload,
            source=NetworkEndpoint.from_dict(data["source"]),
            destination=NetworkEndpoint.from_dict(destination) if destination else None,
            forwarder_id=data.get("forwarder_id", ""),
            protocol_data=(
                ProtocolMetadata.from_dict(protocol_data)
                if protocol_data
                else ProtocolMetadata()
            ),
            timestamp=data["timestamp"],
            hash=data.get("hash", ""),
            is_valid=data.get("is_valid", True),
            validation_error=data.get("validation_error"),
        )

    @classmethod
    def from_json(cls, raw: str | bytes) -> ProtocolMessage:
        """Rebuild a message from a :meth:`to_json` result."""
        return cls.from_dict(json.loads(raw))

to_dict

to_dict() -> dict[str, Any]

Serialize for transport. See the module docstring for the contract.

Source code in src/py20305/forwarders/types.py

def to_dict(self) -> dict[str, Any]:
    """Serialize for transport. See the module docstring for the contract."""
    payload_data: dict[str, str] | dict[str, Any] | list[Any] | str
    if isinstance(self.payload, PayloadEnvelope):
        payload_data = self.payload.serialize()
    else:  # pragma: no cover -- __post_init__ normalizes every other shape
        payload_data = self.payload

    result: dict[str, Any] = {
        "version": _VERSION,
        "protocol": self.protocol.value,
        "direction": self.direction.value,
        "timestamp": self.timestamp,
        "client_id": self.client_id,
        "forwarder_id": self.forwarder_id,
        "payload": payload_data,
        "source": self.source.to_dict(),
        "hash": self.hash,
        "is_valid": self.is_valid,
    }

    if self.destination:
        result["destination"] = self.destination.to_dict()

    protocol_data = self.protocol_data.to_dict()
    if protocol_data:
        result["protocol_data"] = protocol_data

    if self.validation_error:
        result["validation_error"] = self.validation_error

    return result

to_json

to_json() -> str

Serialize to a JSON string.

Source code in src/py20305/forwarders/types.py

def to_json(self) -> str:
    """Serialize to a JSON string."""
    return json.dumps(self.to_dict())

from_dict classmethod

from_dict(data: dict[str, Any]) -> ProtocolMessage

Rebuild a message from a :meth:to_dict result.

The inverse of the serialization contract, so a consumer can read what a producer wrote without hand-unpacking the envelope. Unknown keys under protocol_data survive the round trip via :attr:ProtocolMetadata.extra, which is what lets a consumer built against this version read a message from a later one.

Metadata keys this class does not model are preserved in :attr:ProtocolMetadata.passthrough and re-emitted where they were found, so a message from a producer modelling more fields keeps its shape rather than having them folded a level deeper.

hash is taken from the input rather than recomputed. It is the producer's deduplication key, and recomputing it here would silently change it whenever a field this class does not model was dropped.

Raises:

Type	Description
KeyError	If a key the contract says is always present is missing.

Source code in src/py20305/forwarders/types.py

@classmethod
def from_dict(cls, data: dict[str, Any]) -> ProtocolMessage:
    """Rebuild a message from a :meth:`to_dict` result.

    The inverse of the serialization contract, so a consumer can read
    what a producer wrote without hand-unpacking the envelope. Unknown
    keys under ``protocol_data`` survive the round trip via
    :attr:`ProtocolMetadata.extra`, which is what lets a consumer built
    against this version read a message from a later one.

    Metadata keys this class does not model are preserved in
    :attr:`ProtocolMetadata.passthrough` and re-emitted where they were
    found, so a message from a producer modelling more fields keeps its
    shape rather than having them folded a level deeper.

    ``hash`` is taken from the input rather than recomputed. It is the
    producer's deduplication key, and recomputing it here would silently
    change it whenever a field this class does not model was dropped.

    Raises:
        KeyError: If a key the contract says is always present is missing.
    """
    payload = data["payload"]
    if isinstance(payload, dict) and {"content_type", "data"} <= payload.keys():
        payload = PayloadEnvelope.deserialize(payload)

    destination = data.get("destination")
    protocol_data = data.get("protocol_data")

    return cls(
        protocol=Protocol.from_string(data["protocol"]),
        direction=WireDirection(data["direction"]),
        client_id=data["client_id"],
        payload=payload,
        source=NetworkEndpoint.from_dict(data["source"]),
        destination=NetworkEndpoint.from_dict(destination) if destination else None,
        forwarder_id=data.get("forwarder_id", ""),
        protocol_data=(
            ProtocolMetadata.from_dict(protocol_data)
            if protocol_data
            else ProtocolMetadata()
        ),
        timestamp=data["timestamp"],
        hash=data.get("hash", ""),
        is_valid=data.get("is_valid", True),
        validation_error=data.get("validation_error"),
    )

from_json classmethod

from_json(raw: str | bytes) -> ProtocolMessage

Rebuild a message from a :meth:to_json result.

Source code in src/py20305/forwarders/types.py

@classmethod
def from_json(cls, raw: str | bytes) -> ProtocolMessage:
    """Rebuild a message from a :meth:`to_json` result."""
    return cls.from_dict(json.loads(raw))

py20305.forwarders.manager

Forwarder manager for routing messages to multiple forwarders.

The ForwarderManager acts as a facade that routes MessageFrame objects to one or more registered forwarders (MQTT, file, console, etc.).

ForwarderManager

Routes messages to multiple forwarders.

The manager handles: - Registration of forwarders - Lifecycle management (start/stop all forwarders) - Message routing to all registered forwarders - Aggregated statistics

Usage

manager = ForwarderManager() manager.add_forwarder(mqtt_adapter) await manager.start()

Queue messages - they're routed to all forwarders

manager.queue_message(frame)

await manager.stop()

Source code in src/py20305/forwarders/manager.py

class ForwarderManager:
    """Routes messages to multiple forwarders.

    The manager handles:
    - Registration of forwarders
    - Lifecycle management (start/stop all forwarders)
    - Message routing to all registered forwarders
    - Aggregated statistics

    Usage:
        manager = ForwarderManager()
        manager.add_forwarder(mqtt_adapter)
        await manager.start()

        # Queue messages - they're routed to all forwarders
        manager.queue_message(frame)

        await manager.stop()
    """

    def __init__(self) -> None:
        """Initialize the forwarder manager."""
        self._forwarders: list[BaseForwarder] = []
        #: Forwarders whose `start` raised. Identity-keyed, because a
        #: forwarder is not necessarily hashable by value.
        self._failed: set[BaseForwarder] = set()
        self._running = False
        self._client_lfdi: str | None = None
        self._forwarder_id: str | None = None
        self._source_host: str | None = None
        self._events_dropped_not_running = 0
        #: Events a forwarder did not take -- its start failed, or queueing
        #: raised -- by forwarder.
        self._events_dropped_by_failed: dict[str, int] = {}
        self._connection_observer: LinkObserver | None = None

    @property
    def can_deliver_events(self) -> bool:
        """Return whether an event queued now would reach at least one forwarder.

        A forwarder counts when it started and, if it reports live readiness
        through ``can_deliver``, is ready now. A forwarder that does not report
        it counts as ready once started.
        """
        return self._running and any(
            f not in self._failed and getattr(f, "can_deliver", True) for f in self._forwarders
        )

    @property
    def connection_observer(self) -> LinkObserver | None:
        """Where forwarders report their own broker connection outcomes."""
        return self._connection_observer

    @connection_observer.setter
    def connection_observer(self, value: LinkObserver | None) -> None:
        """Set the connection observer and propagate it to forwarders that support it.

        Set it before :meth:`start` so the first broker connection is reported.
        """
        self._connection_observer = value
        for forwarder in self._forwarders:
            if hasattr(forwarder, "connection_observer"):
                forwarder.connection_observer = value

    @property
    def running(self) -> bool:
        """Return whether the manager is running."""
        return self._running

    @property
    def forwarders(self) -> list[BaseForwarder]:
        """Return list of registered forwarders."""
        return list(self._forwarders)

    @property
    def client_lfdi(self) -> str | None:
        """Return this client's own LFDI, used as an identifier fallback."""
        return self._client_lfdi

    @client_lfdi.setter
    def client_lfdi(self, value: str | None) -> None:
        """Set this client's own LFDI and propagate it to forwarders that use it."""
        self._client_lfdi = value
        for forwarder in self._forwarders:
            if hasattr(forwarder, "client_lfdi"):
                forwarder.client_lfdi = value

    @property
    def forwarder_id(self) -> str | None:
        """Return the configured forwarder ID."""
        return self._forwarder_id

    @forwarder_id.setter
    def forwarder_id(self, value: str | None) -> None:
        """Set forwarder ID and propagate to forwarders that support it."""
        self._forwarder_id = value
        for forwarder in self._forwarders:
            if hasattr(forwarder, "forwarder_id"):
                forwarder.forwarder_id = value

    @property
    def source_host(self) -> str | None:
        """Return source hostname for forwarded messages."""
        return self._source_host

    @source_host.setter
    def source_host(self, value: str | None) -> None:
        """Set source hostname and propagate to forwarders that support it."""
        self._source_host = value
        for forwarder in self._forwarders:
            if hasattr(forwarder, "source_host"):
                forwarder.source_host = value

    def add_forwarder(self, forwarder: BaseForwarder) -> None:
        """Register a forwarder.

        Args:
            forwarder: Forwarder instance to add
        """
        if forwarder in self._forwarders:
            logger.warning("Forwarder %s already registered", forwarder.name)
            return

        self._forwarders.append(forwarder)

        # Propagate properties to new forwarder
        if self._client_lfdi and hasattr(forwarder, "client_lfdi"):
            forwarder.client_lfdi = self._client_lfdi
        if self._forwarder_id and hasattr(forwarder, "forwarder_id"):
            forwarder.forwarder_id = self._forwarder_id
        if self._source_host and hasattr(forwarder, "source_host"):
            forwarder.source_host = self._source_host
        if self._connection_observer is not None and hasattr(forwarder, "connection_observer"):
            forwarder.connection_observer = self._connection_observer

        logger.info("Registered forwarder: %s", forwarder.name)

    def remove_forwarder(self, forwarder: BaseForwarder) -> bool:
        """Unregister a forwarder.

        Args:
            forwarder: Forwarder instance to remove

        Returns:
            True if removed, False if not found
        """
        try:
            self._forwarders.remove(forwarder)
            logger.info("Removed forwarder: %s", forwarder.name)
            return True
        except ValueError:
            return False

    async def start(self) -> None:
        """Start all registered forwarders.

        Forwarders that fail to start are logged but don't prevent
        other forwarders from starting.
        """
        if self._running:
            logger.warning("ForwarderManager already running")
            return

        logger.info("Starting ForwarderManager with %d forwarders", len(self._forwarders))

        from py20305.diagnostics import report

        for forwarder in self._forwarders:
            try:
                await forwarder.start()
            except Exception as e:
                self._failed.add(forwarder)
                report(
                    "warnings",
                    f"Failed to start forwarder {forwarder.name}: {e}",
                    source="forwarder",
                    dedup_key=f"forwarder_lifecycle:{forwarder.name}:start",
                    details={"forwarder": forwarder.name, "phase": "start", "error": str(e)},
                )

        self._running = True

    async def stop(self) -> None:
        """Stop all registered forwarders gracefully."""
        if not self._running:
            return

        from py20305.diagnostics import report

        logger.info("Stopping ForwarderManager")

        for forwarder in self._forwarders:
            try:
                await forwarder.stop()
            except Exception as e:
                report(
                    "warnings",
                    f"Error stopping forwarder {forwarder.name}: {e}",
                    source="forwarder",
                    dedup_key=f"forwarder_lifecycle:{forwarder.name}:stop",
                    details={"forwarder": forwarder.name, "phase": "stop", "error": str(e)},
                )

        self._running = False
        logger.info("ForwarderManager stopped")

    def queue_message(self, frame: MessageFrame) -> None:
        """Route a message to all registered forwarders.

        Messages are queued to each forwarder. Errors in one forwarder
        don't affect others.

        Args:
            frame: Message frame to route
        """
        if not self._running:
            logger.debug("ForwarderManager not running, dropping message")
            return

        for forwarder in self._forwarders:
            if forwarder in self._failed:
                continue
            try:
                forwarder.queue_message(frame)
            except Exception as e:
                logger.error("Error queueing to forwarder %s: %s", forwarder.name, e)

    def failed_forwarders(self) -> list[BaseForwarder]:
        """Registered forwarders whose ``start`` raised and has not since succeeded.

        A forwarder whose broker was unreachable at startup drops everything
        handed to it, while the manager reports itself running -- so "the
        manager started" is not the same question as "anything is being
        delivered", and a caller retrying needs the second.

        Recorded when the start fails rather than read back off the forwarder:
        ``BaseForwarder`` does not require a ``running`` attribute, so asking
        for one would classify any implementation without it as healthy and
        never retry the very forwarders this exists for.
        """
        return [f for f in self._forwarders if f in self._failed]

    async def retry_failed(self) -> int:
        """Try to start the forwarders that are not running. Returns how many came up.

        Separate from :meth:`start` so a caller can drive it on whatever
        cadence suits it. Without this a broker that is briefly unreachable
        when the client boots -- which is exactly when a broker restarted
        alongside it would be -- silently disables forwarding for the lifetime
        of the process.
        """
        recovered = 0
        for forwarder in self.failed_forwarders():
            try:
                await forwarder.start()
            except Exception as e:
                logger.debug("Forwarder %s still unavailable: %s", forwarder.name, e)
                continue
            self._failed.discard(forwarder)
            recovered += 1
            logger.info("Forwarder %s reconnected", forwarder.name)
        return recovered

    def queue_event(self, event: EventFrame) -> None:
        """Route an event to all registered forwarders.

        Mirrors :meth:`queue_message`: non-blocking, and one forwarder's
        failure does not affect the others. Forwarders that do not carry
        events drop them.

        Args:
            event: Event frame to route
        """
        if not self._running:
            # Counted: an audit consumer sees the loss as a sequence gap, and
            # an operator needs a number on this side that explains it.
            self._events_dropped_not_running += 1
            logger.debug("ForwarderManager not running, dropping event")
            return

        for forwarder in self._forwarders:
            if forwarder in self._failed:
                # Its start failed, so it is not running and would drop the
                # event without a word while the manager reports itself running.
                # This count is what explains the gap a sequence-checking
                # consumer then sees.
                self._count_event_drop(forwarder)
                continue
            try:
                forwarder.queue_event(event)
            except Exception as e:
                self._count_event_drop(forwarder)
                logger.error("Error queueing event to forwarder %s: %s", forwarder.name, e)

    def _count_event_drop(self, forwarder: BaseForwarder) -> None:
        """Count one event a forwarder did not take, under a key unique to it.

        Its name, unless another registered forwarder shares the name, when its
        position is added: two forwarders on one counter would hide which of
        them is losing records.
        """
        name = forwarder.name
        if sum(1 for other in self._forwarders if other.name == name) > 1:
            name = f"{name}#{self._forwarders.index(forwarder)}"
        self._events_dropped_by_failed[name] = self._events_dropped_by_failed.get(name, 0) + 1

    def queue_telemetry(self, frame: TelemetryFrame) -> None:
        """Route measured device state to all registered forwarders.

        A forwarder that does not carry telemetry drops it (the default
        implementation on ``AbstractForwarder``), so this is safe to call
        regardless of which forwarders are registered.
        """
        if not self._running:
            logger.debug("ForwarderManager not running, dropping telemetry")
            return

        for forwarder in self._forwarders:
            if forwarder in self._failed:
                continue
            try:
                forwarder.queue_telemetry(frame)
            except Exception as e:
                logger.error("Error queueing telemetry to forwarder %s: %s", forwarder.name, e)

    def get_statistics(self) -> dict[str, Any]:
        """Return aggregated statistics from all forwarders."""
        return {
            "running": self._running,
            "forwarder_count": len(self._forwarders),
            "client_lfdi": self._client_lfdi,
            "events_dropped_not_running": self._events_dropped_not_running,
            "events_dropped_by_forwarder": dict(self._events_dropped_by_failed),
            "forwarders": {f.name: f.get_statistics() for f in self._forwarders},
        }

    @classmethod
    async def from_config(cls, config: ForwarderConfig) -> ForwarderManager:
        """Create and configure a ForwarderManager from config.

        Args:
            config: Forwarder configuration

        Returns:
            Configured ForwarderManager instance
        """
        manager = cls()

        if config.mqtt is not None and config.mqtt.enabled:
            try:
                from py20305.forwarders.mqtt_adapter import MQTTForwarderAdapter
                from py20305.forwarders.mqtt_forwarder import MQTTForwarder

                mqtt = MQTTForwarder(config.mqtt)
                manager.add_forwarder(MQTTForwarderAdapter(mqtt))
            except Exception as e:
                from py20305.diagnostics import report

                report(
                    "errors",
                    f"Failed to instantiate MQTT forwarder: {e}",
                    source="mqtt",
                    details={"error": str(e)},
                )

        return manager

can_deliver_events property

can_deliver_events: bool

Return whether an event queued now would reach at least one forwarder.

A forwarder counts when it started and, if it reports live readiness through can_deliver, is ready now. A forwarder that does not report it counts as ready once started.

connection_observer property writable

connection_observer: LinkObserver | None

Where forwarders report their own broker connection outcomes.

running property

running: bool

Return whether the manager is running.

forwarders property

forwarders: list[BaseForwarder]

Return list of registered forwarders.

client_lfdi property writable

client_lfdi: str | None

Return this client's own LFDI, used as an identifier fallback.

forwarder_id property writable

forwarder_id: str | None

Return the configured forwarder ID.

source_host property writable

source_host: str | None

Return source hostname for forwarded messages.

add_forwarder

add_forwarder(forwarder: BaseForwarder) -> None

Register a forwarder.

Parameters:

Name	Type	Description	Default
forwarder	BaseForwarder	Forwarder instance to add	required

Source code in src/py20305/forwarders/manager.py

def add_forwarder(self, forwarder: BaseForwarder) -> None:
    """Register a forwarder.

    Args:
        forwarder: Forwarder instance to add
    """
    if forwarder in self._forwarders:
        logger.warning("Forwarder %s already registered", forwarder.name)
        return

    self._forwarders.append(forwarder)

    # Propagate properties to new forwarder
    if self._client_lfdi and hasattr(forwarder, "client_lfdi"):
        forwarder.client_lfdi = self._client_lfdi
    if self._forwarder_id and hasattr(forwarder, "forwarder_id"):
        forwarder.forwarder_id = self._forwarder_id
    if self._source_host and hasattr(forwarder, "source_host"):
        forwarder.source_host = self._source_host
    if self._connection_observer is not None and hasattr(forwarder, "connection_observer"):
        forwarder.connection_observer = self._connection_observer

    logger.info("Registered forwarder: %s", forwarder.name)

remove_forwarder

remove_forwarder(forwarder: BaseForwarder) -> bool

Unregister a forwarder.

Parameters:

Name	Type	Description	Default
forwarder	BaseForwarder	Forwarder instance to remove	required

Returns:

Type	Description
bool	True if removed, False if not found

Source code in src/py20305/forwarders/manager.py

def remove_forwarder(self, forwarder: BaseForwarder) -> bool:
    """Unregister a forwarder.

    Args:
        forwarder: Forwarder instance to remove

    Returns:
        True if removed, False if not found
    """
    try:
        self._forwarders.remove(forwarder)
        logger.info("Removed forwarder: %s", forwarder.name)
        return True
    except ValueError:
        return False

start async

start() -> None

Start all registered forwarders.

Forwarders that fail to start are logged but don't prevent other forwarders from starting.

Source code in src/py20305/forwarders/manager.py

async def start(self) -> None:
    """Start all registered forwarders.

    Forwarders that fail to start are logged but don't prevent
    other forwarders from starting.
    """
    if self._running:
        logger.warning("ForwarderManager already running")
        return

    logger.info("Starting ForwarderManager with %d forwarders", len(self._forwarders))

    from py20305.diagnostics import report

    for forwarder in self._forwarders:
        try:
            await forwarder.start()
        except Exception as e:
            self._failed.add(forwarder)
            report(
                "warnings",
                f"Failed to start forwarder {forwarder.name}: {e}",
                source="forwarder",
                dedup_key=f"forwarder_lifecycle:{forwarder.name}:start",
                details={"forwarder": forwarder.name, "phase": "start", "error": str(e)},
            )

    self._running = True

stop async

stop() -> None

Stop all registered forwarders gracefully.

Source code in src/py20305/forwarders/manager.py

async def stop(self) -> None:
    """Stop all registered forwarders gracefully."""
    if not self._running:
        return

    from py20305.diagnostics import report

    logger.info("Stopping ForwarderManager")

    for forwarder in self._forwarders:
        try:
            await forwarder.stop()
        except Exception as e:
            report(
                "warnings",
                f"Error stopping forwarder {forwarder.name}: {e}",
                source="forwarder",
                dedup_key=f"forwarder_lifecycle:{forwarder.name}:stop",
                details={"forwarder": forwarder.name, "phase": "stop", "error": str(e)},
            )

    self._running = False
    logger.info("ForwarderManager stopped")

queue_message

queue_message(frame: MessageFrame) -> None

Route a message to all registered forwarders.

Messages are queued to each forwarder. Errors in one forwarder don't affect others.

Parameters:

Name	Type	Description	Default
frame	MessageFrame	Message frame to route	required

Source code in src/py20305/forwarders/manager.py

def queue_message(self, frame: MessageFrame) -> None:
    """Route a message to all registered forwarders.

    Messages are queued to each forwarder. Errors in one forwarder
    don't affect others.

    Args:
        frame: Message frame to route
    """
    if not self._running:
        logger.debug("ForwarderManager not running, dropping message")
        return

    for forwarder in self._forwarders:
        if forwarder in self._failed:
            continue
        try:
            forwarder.queue_message(frame)
        except Exception as e:
            logger.error("Error queueing to forwarder %s: %s", forwarder.name, e)

failed_forwarders

failed_forwarders() -> list[BaseForwarder]

Registered forwarders whose start raised and has not since succeeded.

A forwarder whose broker was unreachable at startup drops everything handed to it, while the manager reports itself running -- so "the manager started" is not the same question as "anything is being delivered", and a caller retrying needs the second.

Recorded when the start fails rather than read back off the forwarder: BaseForwarder does not require a running attribute, so asking for one would classify any implementation without it as healthy and never retry the very forwarders this exists for.

Source code in src/py20305/forwarders/manager.py

def failed_forwarders(self) -> list[BaseForwarder]:
    """Registered forwarders whose ``start`` raised and has not since succeeded.

    A forwarder whose broker was unreachable at startup drops everything
    handed to it, while the manager reports itself running -- so "the
    manager started" is not the same question as "anything is being
    delivered", and a caller retrying needs the second.

    Recorded when the start fails rather than read back off the forwarder:
    ``BaseForwarder`` does not require a ``running`` attribute, so asking
    for one would classify any implementation without it as healthy and
    never retry the very forwarders this exists for.
    """
    return [f for f in self._forwarders if f in self._failed]

retry_failed async

retry_failed() -> int

Try to start the forwarders that are not running. Returns how many came up.

Separate from :meth:start so a caller can drive it on whatever cadence suits it. Without this a broker that is briefly unreachable when the client boots -- which is exactly when a broker restarted alongside it would be -- silently disables forwarding for the lifetime of the process.

Source code in src/py20305/forwarders/manager.py

async def retry_failed(self) -> int:
    """Try to start the forwarders that are not running. Returns how many came up.

    Separate from :meth:`start` so a caller can drive it on whatever
    cadence suits it. Without this a broker that is briefly unreachable
    when the client boots -- which is exactly when a broker restarted
    alongside it would be -- silently disables forwarding for the lifetime
    of the process.
    """
    recovered = 0
    for forwarder in self.failed_forwarders():
        try:
            await forwarder.start()
        except Exception as e:
            logger.debug("Forwarder %s still unavailable: %s", forwarder.name, e)
            continue
        self._failed.discard(forwarder)
        recovered += 1
        logger.info("Forwarder %s reconnected", forwarder.name)
    return recovered

queue_event

queue_event(event: EventFrame) -> None

Route an event to all registered forwarders.

Mirrors :meth:queue_message: non-blocking, and one forwarder's failure does not affect the others. Forwarders that do not carry events drop them.

Parameters:

Name	Type	Description	Default
event	EventFrame	Event frame to route	required

Source code in src/py20305/forwarders/manager.py

def queue_event(self, event: EventFrame) -> None:
    """Route an event to all registered forwarders.

    Mirrors :meth:`queue_message`: non-blocking, and one forwarder's
    failure does not affect the others. Forwarders that do not carry
    events drop them.

    Args:
        event: Event frame to route
    """
    if not self._running:
        # Counted: an audit consumer sees the loss as a sequence gap, and
        # an operator needs a number on this side that explains it.
        self._events_dropped_not_running += 1
        logger.debug("ForwarderManager not running, dropping event")
        return

    for forwarder in self._forwarders:
        if forwarder in self._failed:
            # Its start failed, so it is not running and would drop the
            # event without a word while the manager reports itself running.
            # This count is what explains the gap a sequence-checking
            # consumer then sees.
            self._count_event_drop(forwarder)
            continue
        try:
            forwarder.queue_event(event)
        except Exception as e:
            self._count_event_drop(forwarder)
            logger.error("Error queueing event to forwarder %s: %s", forwarder.name, e)

queue_telemetry

queue_telemetry(frame: TelemetryFrame) -> None

Route measured device state to all registered forwarders.

A forwarder that does not carry telemetry drops it (the default implementation on AbstractForwarder), so this is safe to call regardless of which forwarders are registered.

Source code in src/py20305/forwarders/manager.py

def queue_telemetry(self, frame: TelemetryFrame) -> None:
    """Route measured device state to all registered forwarders.

    A forwarder that does not carry telemetry drops it (the default
    implementation on ``AbstractForwarder``), so this is safe to call
    regardless of which forwarders are registered.
    """
    if not self._running:
        logger.debug("ForwarderManager not running, dropping telemetry")
        return

    for forwarder in self._forwarders:
        if forwarder in self._failed:
            continue
        try:
            forwarder.queue_telemetry(frame)
        except Exception as e:
            logger.error("Error queueing telemetry to forwarder %s: %s", forwarder.name, e)

get_statistics

get_statistics() -> dict[str, Any]

Return aggregated statistics from all forwarders.

Source code in src/py20305/forwarders/manager.py

def get_statistics(self) -> dict[str, Any]:
    """Return aggregated statistics from all forwarders."""
    return {
        "running": self._running,
        "forwarder_count": len(self._forwarders),
        "client_lfdi": self._client_lfdi,
        "events_dropped_not_running": self._events_dropped_not_running,
        "events_dropped_by_forwarder": dict(self._events_dropped_by_failed),
        "forwarders": {f.name: f.get_statistics() for f in self._forwarders},
    }

from_config async classmethod

from_config(config: ForwarderConfig) -> ForwarderManager

Create and configure a ForwarderManager from config.

Parameters:

Name	Type	Description	Default
config	ForwarderConfig	Forwarder configuration	required

Returns:

Type	Description
ForwarderManager	Configured ForwarderManager instance

Source code in src/py20305/forwarders/manager.py

@classmethod
async def from_config(cls, config: ForwarderConfig) -> ForwarderManager:
    """Create and configure a ForwarderManager from config.

    Args:
        config: Forwarder configuration

    Returns:
        Configured ForwarderManager instance
    """
    manager = cls()

    if config.mqtt is not None and config.mqtt.enabled:
        try:
            from py20305.forwarders.mqtt_adapter import MQTTForwarderAdapter
            from py20305.forwarders.mqtt_forwarder import MQTTForwarder

            mqtt = MQTTForwarder(config.mqtt)
            manager.add_forwarder(MQTTForwarderAdapter(mqtt))
        except Exception as e:
            from py20305.diagnostics import report

            report(
                "errors",
                f"Failed to instantiate MQTT forwarder: {e}",
                source="mqtt",
                details={"error": str(e)},
            )

    return manager

py20305.forwarders.config

Configuration for message forwarders.

Configuration is loaded from the config file's forwarders map. Each forwarder type is an optional typed field on ForwarderConfig.

MQTTForwarderConfig

Bases: _StrictForwarderModel

Configuration for MQTT message forwarding.

Supports both mTLS connections (e.g., AWS IoT Core) and plain MQTT brokers. When cert_path/key_path/ca_path are all omitted, connects without TLS. When any cert path is provided, all three are required and validated.

Source code in src/py20305/forwarders/config.py

class MQTTForwarderConfig(_StrictForwarderModel):
    """Configuration for MQTT message forwarding.

    Supports both mTLS connections (e.g., AWS IoT Core) and plain MQTT brokers.
    When cert_path/key_path/ca_path are all omitted, connects without TLS.
    When any cert path is provided, all three are required and validated.
    """

    endpoint: str = Field(description="MQTT broker hostname or IP")
    port: int = Field(default=8883, description="MQTT broker port (8883 for TLS, 1883 for plain)")
    cert_path: Path | None = Field(default=None, description="Path to client certificate (.pem)")
    key_path: Path | None = Field(default=None, description="Path to client private key (.key)")
    ca_path: Path | None = Field(default=None, description="Path to CA certificate (.pem)")
    topic_base: str = Field(default="csip_client", description="Base topic for publishing")
    client_id_prefix: str = Field(default="csip_client", description="Prefix for MQTT client ID")
    enabled: bool = Field(default=True, description="Whether forwarding is enabled")

    @property
    def tls_enabled(self) -> bool:
        """Return whether TLS is configured."""
        return self.cert_path is not None

    @field_validator("port")
    @classmethod
    def validate_port(cls, v: int) -> int:
        if not 1 <= v <= 65535:
            raise ValueError("Port must be between 1 and 65535")
        return v

    @model_validator(mode="after")
    def validate_cert_paths(self) -> MQTTForwarderConfig:
        """Validate certificate file configuration.

        If all cert paths are None, TLS is disabled (plain MQTT).
        If any cert path is provided, all three must be present and point to
        existing files.
        """
        if not self.enabled:
            return self

        cert_fields = {
            "cert_path": self.cert_path,
            "key_path": self.key_path,
            "ca_path": self.ca_path,
        }
        provided = {k for k, v in cert_fields.items() if v is not None}
        missing = set(cert_fields) - provided

        # All None = plain MQTT, no validation needed
        if not provided:
            return self

        # Partial = error
        if missing:
            raise ValueError(
                f"Partial TLS configuration: {', '.join(sorted(provided))} provided "
                f"but {', '.join(sorted(missing))} missing. "
                f"Provide all three cert paths for TLS, or omit all for plain MQTT."
            )

        # All provided = validate files exist.
        #
        # Absolute paths only. A relative one is resolved against the directory
        # of the configuration file, which happens after validation and is not
        # visible here -- checking it now would test it against the working
        # directory instead, so a correct configuration would fail whenever the
        # client was started from anywhere but the config file's own directory.
        # The client's own TLS paths are treated the same way, and an
        # unreadable file surfaces when the context is built.
        for field_name, path in cert_fields.items():
            assert path is not None  # mypy narrowing
            if not path.is_absolute():
                continue
            if not path.exists():
                raise ValueError(f"{field_name} does not exist: {path}")
            if not path.is_file():
                raise ValueError(f"{field_name} is not a file: {path}")

        return self

tls_enabled property

tls_enabled: bool

Return whether TLS is configured.

validate_cert_paths

validate_cert_paths() -> MQTTForwarderConfig

Validate certificate file configuration.

If all cert paths are None, TLS is disabled (plain MQTT). If any cert path is provided, all three must be present and point to existing files.

Source code in src/py20305/forwarders/config.py

@model_validator(mode="after")
def validate_cert_paths(self) -> MQTTForwarderConfig:
    """Validate certificate file configuration.

    If all cert paths are None, TLS is disabled (plain MQTT).
    If any cert path is provided, all three must be present and point to
    existing files.
    """
    if not self.enabled:
        return self

    cert_fields = {
        "cert_path": self.cert_path,
        "key_path": self.key_path,
        "ca_path": self.ca_path,
    }
    provided = {k for k, v in cert_fields.items() if v is not None}
    missing = set(cert_fields) - provided

    # All None = plain MQTT, no validation needed
    if not provided:
        return self

    # Partial = error
    if missing:
        raise ValueError(
            f"Partial TLS configuration: {', '.join(sorted(provided))} provided "
            f"but {', '.join(sorted(missing))} missing. "
            f"Provide all three cert paths for TLS, or omit all for plain MQTT."
        )

    # All provided = validate files exist.
    #
    # Absolute paths only. A relative one is resolved against the directory
    # of the configuration file, which happens after validation and is not
    # visible here -- checking it now would test it against the working
    # directory instead, so a correct configuration would fail whenever the
    # client was started from anywhere but the config file's own directory.
    # The client's own TLS paths are treated the same way, and an
    # unreadable file surfaces when the context is built.
    for field_name, path in cert_fields.items():
        assert path is not None  # mypy narrowing
        if not path.is_absolute():
            continue
        if not path.exists():
            raise ValueError(f"{field_name} does not exist: {path}")
        if not path.is_file():
            raise ValueError(f"{field_name} is not a file: {path}")

    return self

ConnectionTelemetryConfig

Bases: _StrictForwarderModel

Configuration for reporting this client's own connection outcomes.

Off by default: an operator who has not asked for connection telemetry should not start publishing it on an upgrade.

Source code in src/py20305/forwarders/config.py

class ConnectionTelemetryConfig(_StrictForwarderModel):
    """Configuration for reporting this client's own connection outcomes.

    Off by default: an operator who has not asked for connection telemetry
    should not start publishing it on an upgrade.
    """

    enabled: bool = Field(
        default=False,
        description="Whether to report connection attempts and failures",
    )
    topic_suffix: str = Field(
        default="out/connection-events",
        description="Topic to publish connection events on, relative to the forwarder topic base",
    )
    coalesce_successes: bool = Field(
        default=True,
        description=(
            "Whether successful connections on one link are collapsed into one event per "
            "window. Set false where every connection attempt must be logged as its own "
            "record. Failures are never coalesced."
        ),
    )
    coalesce_window_seconds: float = Field(
        default=60.0,
        description=(
            "How long successful connections are collapsed into one event. "
            "Failures are never coalesced. Zero emits one event per success."
        ),
    )

    @field_validator("coalesce_window_seconds")
    @classmethod
    def validate_window(cls, v: float) -> float:
        # NaN and infinity satisfy `not < 0` and then blow up downstream when
        # the window converts to integer milliseconds -- reject them here,
        # where the operator sees a configuration error instead of a crash.
        if not math.isfinite(v) or v < 0:
            raise ValueError("coalesce_window_seconds must be a finite, non-negative number")
        return v

    @field_validator("topic_suffix")
    @classmethod
    def validate_topic_suffix(cls, v: str) -> str:
        cleaned = v.strip().strip("/")
        if not cleaned:
            raise ValueError("topic_suffix must not be empty")
        # Publishing connection events onto the protocol-message topic would
        # put OCSF envelopes where a consumer expects captured protocol
        # payloads. The two shapes are different and nothing downstream would
        # flag the mix, so it is rejected here rather than discovered there.
        if cleaned == PROTOCOL_MESSAGE_TOPIC_SUFFIX:
            raise ValueError(
                f"topic_suffix must not be {PROTOCOL_MESSAGE_TOPIC_SUFFIX!r} -- that topic "
                "carries captured protocol messages, not connection events"
            )
        # MQTT reserves these for subscription filters; a publisher using them
        # produces a topic no subscriber matches the way the operator intended.
        if "+" in cleaned or "#" in cleaned:
            raise ValueError("topic_suffix must not contain the MQTT wildcards '+' or '#'")
        return cleaned

DeviceTelemetryConfig

Bases: _StrictForwarderModel

Configuration for reporting southbound device reads and writes.

Off by default: an operator who has not asked for southbound telemetry should not start publishing their device traffic on an upgrade.

Source code in src/py20305/forwarders/config.py

class DeviceTelemetryConfig(_StrictForwarderModel):
    """Configuration for reporting southbound device reads and writes.

    Off by default: an operator who has not asked for southbound telemetry
    should not start publishing their device traffic on an upgrade.
    """

    enabled: bool = Field(
        default=False,
        description="Whether to report device reads and writes to the monitoring system",
    )
    topic_suffix: str = Field(
        default="",
        description=(
            "Topic to publish device telemetry on, relative to the forwarder topic base. "
            "Empty (the default) publishes alongside the existing protocol messages, which "
            "is where the rest of the captured traffic already goes."
        ),
    )

    @field_validator("topic_suffix")
    @classmethod
    def validate_topic_suffix(cls, v: str) -> str:
        """Normalize the suffix, allowing empty to mean "the default topic".

        Wildcards are rejected: they are subscription filters, not topic
        names, and a suffix like ``out/#`` would validate here and then
        publish to a literal-``#`` topic no subscriber matches.
        """
        cleaned = v.strip().strip("/")
        if "+" in cleaned or "#" in cleaned:
            raise ValueError("topic_suffix must not contain the MQTT wildcards '+' or '#'")
        return cleaned

validate_topic_suffix classmethod

validate_topic_suffix(v: str) -> str

Normalize the suffix, allowing empty to mean "the default topic".

Wildcards are rejected: they are subscription filters, not topic names, and a suffix like out/# would validate here and then publish to a literal-# topic no subscriber matches.

Source code in src/py20305/forwarders/config.py

@field_validator("topic_suffix")
@classmethod
def validate_topic_suffix(cls, v: str) -> str:
    """Normalize the suffix, allowing empty to mean "the default topic".

    Wildcards are rejected: they are subscription filters, not topic
    names, and a suffix like ``out/#`` would validate here and then
    publish to a literal-``#`` topic no subscriber matches.
    """
    cleaned = v.strip().strip("/")
    if "+" in cleaned or "#" in cleaned:
        raise ValueError("topic_suffix must not contain the MQTT wildcards '+' or '#'")
    return cleaned

AuditConfig

Bases: _StrictForwarderModel

Configuration for the control audit trail.

One switch for both halves: device writes, enriched with what caused them, and the event lifecycle and loss-of-communications records on their own topic. Off by default, like the other telemetry streams.

Writes are published without the high-volume reads, so an operator can keep an audit trail without turning on full device telemetry. With both on, each write is still published once.

Source code in src/py20305/forwarders/config.py

class AuditConfig(_StrictForwarderModel):
    """Configuration for the control audit trail.

    One switch for both halves: device writes, enriched with what caused them,
    and the event lifecycle and loss-of-communications records on their own
    topic. Off by default, like the other telemetry streams.

    Writes are published without the high-volume reads, so an operator can
    keep an audit trail without turning on full device telemetry. With both on,
    each write is still published once.
    """

    enabled: bool = Field(
        default=False,
        description=(
            "Whether to publish device writes with their cause, and event lifecycle "
            "and loss-of-communications records"
        ),
    )
    topic_suffix: str = Field(
        default="out/der-events",
        description=(
            "Topic for lifecycle and loss-of-communications records, relative to the "
            "forwarder topic base. Device writes stay on the device telemetry topic."
        ),
    )

    @field_validator("topic_suffix")
    @classmethod
    def validate_topic_suffix(cls, v: str) -> str:
        cleaned = v.strip().strip("/")
        # An indexer that routes on the outbound prefix would not pick up a
        # record published anywhere else, so the trail would be lost silently.
        if not cleaned.startswith("out/") or cleaned == "out/":
            raise ValueError("topic_suffix must start with 'out/' and name a topic under it")
        if "+" in cleaned or "#" in cleaned:
            raise ValueError("topic_suffix must not contain the MQTT wildcards '+' or '#'")
        return cleaned

ForwarderConfig

Bases: _StrictForwarderModel

Top-level forwarder configuration.

Each forwarder type is an optional typed field. This replaces the previous list-of-ForwarderSpec approach with named fields.

Source code in src/py20305/forwarders/config.py

class ForwarderConfig(_StrictForwarderModel):
    """Top-level forwarder configuration.

    Each forwarder type is an optional typed field. This replaces the
    previous list-of-ForwarderSpec approach with named fields.
    """

    schema_dir: Path | None = Field(
        default=None,
        description="Path to IEEE 2030.5 XSD schemas for message validation",
    )
    mqtt: MQTTForwarderConfig | None = Field(
        default=None,
        description="MQTT forwarder configuration",
    )
    retry_interval_seconds: int = Field(
        default=60,
        ge=0,
        description=(
            "How often to retry a forwarder whose broker was unreachable. "
            "Zero disables retrying, leaving a failed forwarder stopped for "
            "the lifetime of the process."
        ),
    )
    connection_telemetry: ConnectionTelemetryConfig = Field(
        default_factory=ConnectionTelemetryConfig,
        description="Reporting of this client's own connection attempts and failures",
    )
    device_telemetry: DeviceTelemetryConfig = Field(
        default_factory=DeviceTelemetryConfig,
        description="Reporting of southbound device reads and writes",
    )
    audit: AuditConfig = Field(
        default_factory=AuditConfig,
        description="Control audit trail: enriched device writes and event lifecycle records",
    )

    @model_validator(mode="after")
    def validate_audit_topic_is_its_own(self) -> ForwarderConfig:
        """Keep the lifecycle records off every other stream's topic.

        Checked whether or not the other streams are enabled: turning one on
        later must not quietly start mixing two payload shapes on one topic.
        """
        topic = self.audit.topic_suffix
        others = {
            "protocol messages": PROTOCOL_MESSAGE_TOPIC_SUFFIX,
            "device_telemetry": self.device_telemetry.topic_suffix or PROTOCOL_MESSAGE_TOPIC_SUFFIX,
            "connection_telemetry": self.connection_telemetry.topic_suffix,
            "measured device state": TELEMETRY_TOPIC_SUFFIX,
        }
        for name, other in others.items():
            if topic == other:
                raise ValueError(
                    f"audit.topic_suffix {topic!r} is also the {name} topic; "
                    "lifecycle records need a topic of their own"
                )
        return self

    @model_validator(mode="after")
    def validate_telemetry_topics_differ(self) -> ForwarderConfig:
        """Keep the two telemetry streams off one another's topic.

        They carry different payload shapes -- OCSF connection events and
        protocol-message envelopes -- and a consumer reading one would
        silently receive the other mixed in. Device telemetry's empty default
        means the protocol-message topic, so the comparison is on effective
        topics.
        """
        if not (self.connection_telemetry.enabled and self.device_telemetry.enabled):
            return self
        device_topic = self.device_telemetry.topic_suffix or PROTOCOL_MESSAGE_TOPIC_SUFFIX
        if device_topic == self.connection_telemetry.topic_suffix:
            raise ValueError(
                f"connection_telemetry and device_telemetry both publish to {device_topic!r}; "
                "they carry different payload shapes and must use different topics"
            )
        return self

    def has_enabled_forwarders(self) -> bool:
        """Return True if any forwarder is configured and enabled."""
        return bool(self.mqtt is not None and self.mqtt.enabled)

validate_audit_topic_is_its_own

validate_audit_topic_is_its_own() -> ForwarderConfig

Keep the lifecycle records off every other stream's topic.

Checked whether or not the other streams are enabled: turning one on later must not quietly start mixing two payload shapes on one topic.

Source code in src/py20305/forwarders/config.py

@model_validator(mode="after")
def validate_audit_topic_is_its_own(self) -> ForwarderConfig:
    """Keep the lifecycle records off every other stream's topic.

    Checked whether or not the other streams are enabled: turning one on
    later must not quietly start mixing two payload shapes on one topic.
    """
    topic = self.audit.topic_suffix
    others = {
        "protocol messages": PROTOCOL_MESSAGE_TOPIC_SUFFIX,
        "device_telemetry": self.device_telemetry.topic_suffix or PROTOCOL_MESSAGE_TOPIC_SUFFIX,
        "connection_telemetry": self.connection_telemetry.topic_suffix,
        "measured device state": TELEMETRY_TOPIC_SUFFIX,
    }
    for name, other in others.items():
        if topic == other:
            raise ValueError(
                f"audit.topic_suffix {topic!r} is also the {name} topic; "
                "lifecycle records need a topic of their own"
            )
    return self

validate_telemetry_topics_differ

validate_telemetry_topics_differ() -> ForwarderConfig

Keep the two telemetry streams off one another's topic.

They carry different payload shapes -- OCSF connection events and protocol-message envelopes -- and a consumer reading one would silently receive the other mixed in. Device telemetry's empty default means the protocol-message topic, so the comparison is on effective topics.

Source code in src/py20305/forwarders/config.py

@model_validator(mode="after")
def validate_telemetry_topics_differ(self) -> ForwarderConfig:
    """Keep the two telemetry streams off one another's topic.

    They carry different payload shapes -- OCSF connection events and
    protocol-message envelopes -- and a consumer reading one would
    silently receive the other mixed in. Device telemetry's empty default
    means the protocol-message topic, so the comparison is on effective
    topics.
    """
    if not (self.connection_telemetry.enabled and self.device_telemetry.enabled):
        return self
    device_topic = self.device_telemetry.topic_suffix or PROTOCOL_MESSAGE_TOPIC_SUFFIX
    if device_topic == self.connection_telemetry.topic_suffix:
        raise ValueError(
            f"connection_telemetry and device_telemetry both publish to {device_topic!r}; "
            "they carry different payload shapes and must use different topics"
        )
    return self

has_enabled_forwarders

has_enabled_forwarders() -> bool

Return True if any forwarder is configured and enabled.

Source code in src/py20305/forwarders/config.py

def has_enabled_forwarders(self) -> bool:
    """Return True if any forwarder is configured and enabled."""
    return bool(self.mqtt is not None and self.mqtt.enabled)

py20305.forwarders.connection_telemetry

Connection telemetry -- reporting this client's own connection outcomes.

The forwarder already carries this client's captured IEEE 2030.5 exchanges to a security-monitoring system, and the device-telemetry channel carries its southbound reads and writes. This module adds the third channel: the client's own connection outcomes. A network sensor on a span port can approximate the connection attempts a monitoring platform wants logged. It cannot produce the error log at all -- a certificate that fails validation, a redirect loop, a 500 from the server are outcomes this client knows and a capture cannot recover from inside TLS.

This module turns those outcomes into OCSF Network Activity events (the vendored ocsf module beside this one) and publishes them on their own topic. What lives here is the mapping from this client's error taxonomy onto the schema, and the coalescing the volume of a polling client requires.

Two rules come from the compliance argument rather than from the schema, and the event class enforces both at construction:

• A failure carries its reason. That reason is the record's entire value.
• Failures are never coalesced. Successes may be collapsed into a window carrying bounds and a count; collapsing failures would discard the reasons.

The emitter implements the client's :class:~py20305.client.observer.ConnectionObserver seam, so wiring it up is one assignment: client.http.connection_observer = emitter.

A host application can report its other links through the same emitter: :meth:ConnectionTelemetryEmitter.record_connection and :meth:ConnectionTelemetryEmitter.record_connection_failure take the link's role, service, direction and endpoints. Every event carries one role:<role> label in metadata.labels (see :class:ConnectionRole) and the connection's direction in connection_info.direction_id, so a consumer can tell the links apart.

ConnectionRole

Bases: StrEnum

Which of the host application's links a connection event describes.

Serialized as one role:<value> label in metadata.labels (see :func:role_label). A consumer keys on these strings, so they are part of the wire contract: add a member rather than renaming one.

Source code in src/py20305/forwarders/connection_telemetry.py

class ConnectionRole(StrEnum):
    """Which of the host application's links a connection event describes.

    Serialized as one ``role:<value>`` label in ``metadata.labels`` (see
    :func:`role_label`). A consumer keys on these strings, so they are part of
    the wire contract: add a member rather than renaming one.
    """

    #: The IEEE 2030.5 client's link to the utility server.
    NORTHBOUND = "northbound"
    #: A connection to a DER device (SunSpec Modbus, DNP3, OCPP and others).
    SOUTHBOUND = "southbound"
    #: An inbound connection to the host's own management console or API.
    MANAGEMENT = "management"
    #: An inbound connection to one of the host's own protocol listeners.
    SERVER = "server"
    #: The connection to the broker that carries this telemetry.
    TELEMETRY = "telemetry"

Window dataclass

A closed window of successful connections, ready to emit.

Attributes:

Name	Type	Description
count	int	Attempts the window represents.
start_ms	int	First attempt, as an OCSF timestamp.
end_ms	int	Last attempt, as an OCSF timestamp.
src	Endpoint | None	The source endpoint observed when the window opened, or the first one seen inside it. Carried with the window rather than re-read at emission, because the success that closes a window belongs to a different attempt than the one that opened it.
dst	Endpoint | None	The destination endpoint observed together with src.

Source code in src/py20305/forwarders/connection_telemetry.py

@dataclass(frozen=True)
class Window:
    """A closed window of successful connections, ready to emit.

    Attributes:
        count: Attempts the window represents.
        start_ms: First attempt, as an OCSF timestamp.
        end_ms: Last attempt, as an OCSF timestamp.
        src: The source endpoint observed when the window opened, or the
            first one seen inside it. Carried with the window rather than
            re-read at emission, because the success that closes a window
            belongs to a different attempt than the one that opened it.
        dst: The destination endpoint observed together with ``src``.
    """

    count: int
    start_ms: int
    end_ms: int
    src: Endpoint | None = None
    dst: Endpoint | None = None

Outcome dataclass

A connection outcome classified for OCSF.

Attributes:

Name	Type	Description
activity_id	NetworkActivityId	FAIL / REFUSE when nothing was established; OPEN when the connection worked and the exchange over it did not.
detail	str	Why it happened. Required on every failure.
status_code	str | None	The peer's own code, when it gave one.

Source code in src/py20305/forwarders/connection_telemetry.py

@dataclass(frozen=True)
class Outcome:
    """A connection outcome classified for OCSF.

    Attributes:
        activity_id: ``FAIL`` / ``REFUSE`` when nothing was established;
            ``OPEN`` when the connection worked and the exchange over it did
            not.
        detail: Why it happened. Required on every failure.
        status_code: The peer's own code, when it gave one.
    """

    activity_id: NetworkActivityId
    detail: str
    status_code: str | None = None

    def __post_init__(self) -> None:
        """Remove URL credentials from the reason text, then bound it.

        Done at the boundary so no caller can bypass it. The order matters:
        clipping first can cut off the ``@`` that ends a URL's userinfo, and
        the scrubber then no longer recognizes the URL and leaves the
        password in place.
        """
        object.__setattr__(self, "detail", _bounded(_scrub_credentials(self.detail)))
        if self.status_code is not None:
            object.__setattr__(self, "status_code", _bounded_code(self.status_code))

CoalescingWindow

Collapses successful connections on one link into one event per window.

A polling client would otherwise put more volume on the collector's flow path than the passive capture beside it. A coalesced record still has to answer the question a connection log asks -- was there an attempt at time T -- so it carries the window's bounds and the attempt count, and every attempt in it shared one link: role, direction, service and peer. It is not a sample.

Only successes are coalesced. Failures never enter here.

A window keeps the endpoints observed when it opened. The success that closes a window is a different attempt from the one that opened it, so reading the current endpoints at close time would report a connection that carried none of the successes being described. A coalesced record therefore names the source port of the first connection in the window only.

Not synchronized: every caller runs on the event loop, and none of the methods here await, so a window update cannot be interleaved with another.

Source code in src/py20305/forwarders/connection_telemetry.py

class CoalescingWindow:
    """Collapses successful connections on one link into one event per window.

    A polling client would otherwise put more volume on the collector's flow
    path than the passive capture beside it. A coalesced record still has to
    answer the question a connection log asks -- was there an attempt at time
    T -- so it carries the window's bounds and the attempt count, and every
    attempt in it shared one link: role, direction, service and peer. It is
    not a sample.

    Only successes are coalesced. Failures never enter here.

    A window keeps the endpoints observed when it opened. The success that
    *closes* a window is a different attempt from the one that opened it, so
    reading the current endpoints at close time would report a connection
    that carried none of the successes being described. A coalesced record
    therefore names the source port of the first connection in the window
    only.

    Not synchronized: every caller runs on the event loop, and none of the
    methods here await, so a window update cannot be interleaved with another.
    """

    def __init__(self, window_seconds: float = DEFAULT_COALESCE_WINDOW_SECONDS) -> None:
        """Create a window.

        Args:
            window_seconds: How long a window stays open. Zero disables
                coalescing, emitting one event per success.
        """
        if window_seconds < 0:
            raise ValueError(f"window_seconds must not be negative, got {window_seconds}")
        self._window_ms = int(window_seconds * 1000)
        self._count = 0
        self._start_ms: int | None = None
        self._end_ms: int | None = None
        self._src: Endpoint | None = None
        self._dst: Endpoint | None = None

    @property
    def window_ms(self) -> int:
        """Window length in milliseconds."""
        return self._window_ms

    @property
    def pending(self) -> int:
        """Successes recorded but not yet emitted."""
        return self._count

    @property
    def start_ms(self) -> int | None:
        """Start of the open window, or ``None`` when no window is open."""
        return self._start_ms

    def record(
        self, now_ms: int, src: Endpoint | None = None, dst: Endpoint | None = None
    ) -> Window | None:
        """Record one success, returning a window to emit when one closes.

        Args:
            now_ms: The time of the success, in epoch milliseconds.
            src: The source endpoint of the connection this success was
                carried on, when it is known.
            dst: The destination endpoint of that connection, when it is known.

        Returns:
            The closed window when this success closed one, otherwise ``None``.
            The returned window never includes the success that closed it --
            that one opens the next window, so no attempt is counted twice or
            dropped -- and it carries the endpoints observed when it opened,
            not the ones live at close time.
        """
        if self._window_ms == 0:
            return Window(1, now_ms, now_ms, src, dst)

        if self._start_ms is None:
            self._open(now_ms, src, dst)
            return None

        if now_ms - self._start_ms >= self._window_ms:
            closed = self._snapshot()
            self._open(now_ms, src, dst)
            return closed

        self._count += 1
        self._end_ms = now_ms
        # A window that opened without a source endpoint adopts the first one
        # seen inside it: better a connection that carried some of these
        # successes than none at all. Its destination comes with it, since
        # both describe the same connection.
        if self._src is None and src is not None:
            self._src = src
            self._dst = dst if dst is not None else self._dst
        elif self._dst is None:
            self._dst = dst
        return None

    def _open(self, now_ms: int, src: Endpoint | None, dst: Endpoint | None) -> None:
        """Start a new window at ``now_ms`` with the given endpoints."""
        self._start_ms = now_ms
        self._end_ms = now_ms
        self._count = 1
        self._src = src
        self._dst = dst

    def _snapshot(self) -> Window:
        """Return the current window as it stands, for emission."""
        assert self._start_ms is not None
        return Window(
            self._count, self._start_ms, self._end_ms or self._start_ms, self._src, self._dst
        )

    def take_expired(self, now_ms: int) -> Window | None:
        """Close the window if its configured length has elapsed.

        The window normally closes when the success *after* it arrives. A link
        whose successes stop -- it starts failing, or goes idle -- would
        otherwise hold its last successes unreported until shutdown, so other
        activity through the emitter offers the clock a chance to close an
        expired window.

        Returns:
            The expired window, or ``None`` when nothing is open or it is
            still inside its length.
        """
        if self._window_ms == 0 or self._start_ms is None or self._count == 0:
            return None
        if now_ms - self._start_ms < self._window_ms:
            return None
        return self.flush()

    def flush(self) -> Window | None:
        """Close the open window, if any.

        Called on shutdown so the successes accumulated since the last close
        are reported rather than discarded -- an unreported attempt is a gap
        in the connection log.

        Returns:
            The pending window, or ``None`` when nothing is pending.
        """
        if self._start_ms is None or self._count == 0:
            return None
        closed = self._snapshot()
        self._start_ms = None
        self._end_ms = None
        self._count = 0
        self._src = None
        self._dst = None
        return closed

window_ms property

window_ms: int

Window length in milliseconds.

pending property

pending: int

Successes recorded but not yet emitted.

start_ms property

start_ms: int | None

Start of the open window, or None when no window is open.

record

record(now_ms: int, src: Endpoint | None = None, dst: Endpoint | None = None) -> Window | None

Record one success, returning a window to emit when one closes.

Parameters:

Name	Type	Description	Default
now_ms	int	The time of the success, in epoch milliseconds.	required
src	Endpoint | None	The source endpoint of the connection this success was carried on, when it is known.	None
dst	Endpoint | None	The destination endpoint of that connection, when it is known.	None

Returns:

Type	Description
Window | None	The closed window when this success closed one, otherwise None.
Window | None	The returned window never includes the success that closed it --
Window | None	that one opens the next window, so no attempt is counted twice or
Window | None	dropped -- and it carries the endpoints observed when it opened,
Window | None	not the ones live at close time.

Source code in src/py20305/forwarders/connection_telemetry.py

def record(
    self, now_ms: int, src: Endpoint | None = None, dst: Endpoint | None = None
) -> Window | None:
    """Record one success, returning a window to emit when one closes.

    Args:
        now_ms: The time of the success, in epoch milliseconds.
        src: The source endpoint of the connection this success was
            carried on, when it is known.
        dst: The destination endpoint of that connection, when it is known.

    Returns:
        The closed window when this success closed one, otherwise ``None``.
        The returned window never includes the success that closed it --
        that one opens the next window, so no attempt is counted twice or
        dropped -- and it carries the endpoints observed when it opened,
        not the ones live at close time.
    """
    if self._window_ms == 0:
        return Window(1, now_ms, now_ms, src, dst)

    if self._start_ms is None:
        self._open(now_ms, src, dst)
        return None

    if now_ms - self._start_ms >= self._window_ms:
        closed = self._snapshot()
        self._open(now_ms, src, dst)
        return closed

    self._count += 1
    self._end_ms = now_ms
    # A window that opened without a source endpoint adopts the first one
    # seen inside it: better a connection that carried some of these
    # successes than none at all. Its destination comes with it, since
    # both describe the same connection.
    if self._src is None and src is not None:
        self._src = src
        self._dst = dst if dst is not None else self._dst
    elif self._dst is None:
        self._dst = dst
    return None

take_expired

take_expired(now_ms: int) -> Window | None

Close the window if its configured length has elapsed.

The window normally closes when the success after it arrives. A link whose successes stop -- it starts failing, or goes idle -- would otherwise hold its last successes unreported until shutdown, so other activity through the emitter offers the clock a chance to close an expired window.

Returns:

Type	Description
Window | None	The expired window, or None when nothing is open or it is
Window | None	still inside its length.

Source code in src/py20305/forwarders/connection_telemetry.py

def take_expired(self, now_ms: int) -> Window | None:
    """Close the window if its configured length has elapsed.

    The window normally closes when the success *after* it arrives. A link
    whose successes stop -- it starts failing, or goes idle -- would
    otherwise hold its last successes unreported until shutdown, so other
    activity through the emitter offers the clock a chance to close an
    expired window.

    Returns:
        The expired window, or ``None`` when nothing is open or it is
        still inside its length.
    """
    if self._window_ms == 0 or self._start_ms is None or self._count == 0:
        return None
    if now_ms - self._start_ms < self._window_ms:
        return None
    return self.flush()

flush

flush() -> Window | None

Close the open window, if any.

Called on shutdown so the successes accumulated since the last close are reported rather than discarded -- an unreported attempt is a gap in the connection log.

Returns:

Type	Description
Window | None	The pending window, or None when nothing is pending.

Source code in src/py20305/forwarders/connection_telemetry.py

def flush(self) -> Window | None:
    """Close the open window, if any.

    Called on shutdown so the successes accumulated since the last close
    are reported rather than discarded -- an unreported attempt is a gap
    in the connection log.

    Returns:
        The pending window, or ``None`` when nothing is pending.
    """
    if self._start_ms is None or self._count == 0:
        return None
    closed = self._snapshot()
    self._start_ms = None
    self._end_ms = None
    self._count = 0
    self._src = None
    self._dst = None
    return closed

LinkObserver

Bases: Protocol

Receives connection outcomes from the links this package opens or accepts.

:class:ConnectionTelemetryEmitter implements it. A host application can put its own object in front of the emitter, for example to replace the emitter at runtime or to pass calls from worker threads to the event loop. Implementations must never raise.

Source code in src/py20305/forwarders/connection_telemetry.py

class LinkObserver(Protocol):
    """Receives connection outcomes from the links this package opens or accepts.

    :class:`ConnectionTelemetryEmitter` implements it. A host application can
    put its own object in front of the emitter, for example to replace the
    emitter at runtime or to pass calls from worker threads to the event loop.
    Implementations must never raise.
    """

    def record_connection(
        self,
        *,
        role: ConnectionRole | str,
        service: str | None,
        direction: ConnectionDirectionId = ConnectionDirectionId.OUTBOUND,
        dst: Endpoint | None = None,
        src: Endpoint | None = None,
        url: str | None = None,
        now_ms: int | None = None,
    ) -> None:
        """Record one successful connection."""

    def record_connection_failure(
        self,
        *,
        role: ConnectionRole | str,
        service: str | None,
        reason: BaseException | str,
        direction: ConnectionDirectionId = ConnectionDirectionId.OUTBOUND,
        dst: Endpoint | None = None,
        src: Endpoint | None = None,
        url: str | None = None,
        activity_id: NetworkActivityId | None = None,
        status_code: str | None = None,
        now_ms: int | None = None,
    ) -> None:
        """Record one failed connection with its reason."""

record_connection

record_connection(*, role: ConnectionRole | str, service: str | None, direction: ConnectionDirectionId = ConnectionDirectionId.OUTBOUND, dst: Endpoint | None = None, src: Endpoint | None = None, url: str | None = None, now_ms: int | None = None) -> None

Record one successful connection.

Source code in src/py20305/forwarders/connection_telemetry.py

def record_connection(
    self,
    *,
    role: ConnectionRole | str,
    service: str | None,
    direction: ConnectionDirectionId = ConnectionDirectionId.OUTBOUND,
    dst: Endpoint | None = None,
    src: Endpoint | None = None,
    url: str | None = None,
    now_ms: int | None = None,
) -> None:
    """Record one successful connection."""

record_connection_failure

record_connection_failure(*, role: ConnectionRole | str, service: str | None, reason: BaseException | str, direction: ConnectionDirectionId = ConnectionDirectionId.OUTBOUND, dst: Endpoint | None = None, src: Endpoint | None = None, url: str | None = None, activity_id: NetworkActivityId | None = None, status_code: str | None = None, now_ms: int | None = None) -> None

Record one failed connection with its reason.

Source code in src/py20305/forwarders/connection_telemetry.py

def record_connection_failure(
    self,
    *,
    role: ConnectionRole | str,
    service: str | None,
    reason: BaseException | str,
    direction: ConnectionDirectionId = ConnectionDirectionId.OUTBOUND,
    dst: Endpoint | None = None,
    src: Endpoint | None = None,
    url: str | None = None,
    activity_id: NetworkActivityId | None = None,
    status_code: str | None = None,
    now_ms: int | None = None,
) -> None:
    """Record one failed connection with its reason."""

ConnectionTelemetryEmitter

Turns connection outcomes into events on the forwarder transport.

One instance per host application. Attached as the IEEE 2030.5 client's connection observer (client.http.connection_observer = emitter), the client calls :meth:begin_request / :meth:record_success / :meth:record_failure / :meth:on_connect / :meth:flush through that seam, and those events carry the northbound role. The host reports its other links through :meth:record_connection and :meth:record_connection_failure.

Successes are coalesced per link into windows, unless the configuration turns coalescing off, in which case every success is its own event. Failures are always one event each.

Events produced while the transport cannot deliver them -- the broker is unreachable, or the forwarder has not started -- are held, up to a bound, and sent once it can. A broker outage is itself a connection event, and its record would otherwise be lost to the outage it describes.

Nothing here may raise into the caller. Telemetry that can break the connection it describes is worse than absent telemetry, so every entry point swallows its own errors and logs them.

Source code in src/py20305/forwarders/connection_telemetry.py

class ConnectionTelemetryEmitter:
    """Turns connection outcomes into events on the forwarder transport.

    One instance per host application. Attached as the IEEE 2030.5 client's
    connection observer (``client.http.connection_observer = emitter``), the
    client calls :meth:`begin_request` / :meth:`record_success` /
    :meth:`record_failure` / :meth:`on_connect` / :meth:`flush` through that
    seam, and those events carry the ``northbound`` role. The host reports
    its other links through :meth:`record_connection` and
    :meth:`record_connection_failure`.

    Successes are coalesced per link into windows, unless the configuration
    turns coalescing off, in which case every success is its own event.
    Failures are always one event each.

    Events produced while the transport cannot deliver them -- the broker is
    unreachable, or the forwarder has not started -- are held, up to a bound,
    and sent once it can. A broker outage is itself a connection event, and
    its record would otherwise be lost to the outage it describes.

    Nothing here may raise into the caller. Telemetry that can break the
    connection it describes is worse than absent telemetry, so every entry
    point swallows its own errors and logs them.
    """

    def __init__(
        self,
        forwarder: ForwarderManager | None,
        config: ConnectionTelemetryConfig,
        *,
        product_version: str | None = None,
        product_name: str = PRODUCT_NAME,
        vendor_name: str = DEFAULT_VENDOR_NAME,
        max_held_events: int = DEFAULT_MAX_HELD_EVENTS,
    ) -> None:
        """Create an emitter.

        Args:
            forwarder: Where events go. ``None`` disables emission.
            config: Whether telemetry is on, its topic, and how successes are
                coalesced.
            product_version: Version of the producing application, for
                ``metadata.product``.
            product_name: Name of the producing application, for
                ``metadata.product``. A host application passes its own name.
            vendor_name: Publisher of the producing application.
            max_held_events: Most events held while the transport cannot
                deliver them.
        """
        self._forwarder = forwarder
        self._config = config
        self._metadata = {
            role: build_metadata(
                product_version, role=role, product_name=product_name, vendor_name=vendor_name
            )
            for role in ConnectionRole
        }
        #: Failures to *construct or hand off* an event since start -- a bug
        #: in classification, the window, or serialization, or a directly
        #: attached forwarder that raises. The first logs at WARNING, the
        #: rest at DEBUG. A broker outage is not counted here: events are held
        #: while the transport cannot deliver them, and broker-level delivery
        #: failure is visible in the transport's own statistics.
        self.emit_failures = 0
        #: Held events discarded because the hold was full.
        self.held_events_dropped = 0
        self._window_seconds = config.coalesce_window_seconds if config.coalesce_successes else 0.0
        self._links: dict[_LinkKey, _OpenLink] = {}
        self._held: deque[EventFrame] = deque()
        self._max_held = max(0, max_held_events)
        self._server_endpoint: Endpoint | None = None
        self._base_url: str | None = None

    @property
    def enabled(self) -> bool:
        """Whether events will actually be emitted."""
        return self._config.enabled and self._forwarder is not None

    @property
    def coalescing(self) -> bool:
        """Whether successes are collapsed into windows rather than sent one by one."""
        return self._window_seconds > 0

    @property
    def held_events(self) -> int:
        """Events waiting for the transport to become able to deliver them."""
        return len(self._held)

    def set_server(self, host: str | None, port: int | None, base_url: str | None = None) -> None:
        """Record the server this client targets, for events with no live socket.

        A connection that never established has no socket to report, but the
        address the client was trying to reach is exactly what the failure
        record needs to name.
        """
        self._base_url = _redact_url(base_url)
        self._server_endpoint = endpoint_for_host(host, port) if port else None

    # The IEEE 2030.5 client's observer seam.

    def begin_request(self) -> None:
        """Clear any socket left from an earlier request in this task.

        Scopes attribution to "a connection established while handling *this*
        request". Without it, a task that opens a connection and then makes a
        second request over the pool would attribute the first connection's
        local port to the second request -- right only by luck, and wrong as
        soon as the pool hands back a different connection.
        """
        _request_socket.set(None)

    def on_connect(self, pair: SocketPair) -> None:
        """Observe a newly established connection's socket.

        Reached through the client's connector, which is the only place the
        local port is available.
        """
        _request_socket.set(pair)

    def record_success(self, now_ms: int | None = None) -> None:
        """Record one successful IEEE 2030.5 exchange with the server."""
        if not self.enabled:
            return
        try:
            moment = now_ms if now_ms is not None else _now_ms()
            self._emit_expired(moment)
            src, dst = _endpoints(_request_socket.get(), None)
            self._record_window(_NORTHBOUND_KEY, src, dst, self._base_url, moment)
        except Exception:
            self._record_emit_failure("record a success")

    def record_failure(self, exc: BaseException) -> None:
        """Record one failed IEEE 2030.5 connection or exchange, with its reason.

        Never coalesced: each failure keeps its own event and its own reason.
        """
        if not self.enabled:
            return
        try:
            # The failure is also the clock's chance to close expired success
            # windows: those successes happened before this failure, and
            # holding them until the next success -- which may never come --
            # would leave them unreported until shutdown. Guarded on its own:
            # a problem emitting a window must not cost the failure record,
            # which is the more important of the two.
            try:
                self._emit_expired(_now_ms())
            except Exception:
                self._record_emit_failure("close an expired window")
            outcome = classify(exc)
            if outcome is None:
                return
            # The retained socket belongs to a connection this request
            # established. For an exchange failure (OPEN) or a teardown
            # (RESET) that connection is the subject of the record; for a
            # failure that never opened (FAIL, REFUSE) it is an earlier
            # attempt's socket, and attaching its local port would attribute
            # the failure to a connection that did not fail.
            socket_pair = (
                _request_socket.get()
                if outcome.activity_id in (NetworkActivityId.OPEN, NetworkActivityId.RESET)
                else None
            )
            self._emit(
                build_failure_event(
                    outcome,
                    metadata=self._metadata[ConnectionRole.NORTHBOUND],
                    socket_pair=socket_pair,
                    server_endpoint=self._server_endpoint,
                    url=self._base_url,
                )
            )
        except Exception:
            self._record_emit_failure("record a failure")

    # The host application's other links.

    def record_connection(
        self,
        *,
        role: ConnectionRole | str,
        service: str | None,
        direction: ConnectionDirectionId = ConnectionDirectionId.OUTBOUND,
        dst: Endpoint | None = None,
        src: Endpoint | None = None,
        url: str | None = None,
        now_ms: int | None = None,
    ) -> None:
        """Record one successful connection on one of the host's links.

        Args:
            role: Which link the connection belongs to.
            service: Service label for ``dst_endpoint.svc_name``, for example
                ``"modbus"``.
            direction: ``OUTBOUND`` for a connection the host opened,
                ``INBOUND`` for one it accepted.
            dst: The responder: the device or broker for an outbound
                connection, the host's own listener for an inbound one.
                Required for the event to be reported.
            src: The initiator, when its address is known.
            url: URL the connection targeted, if any. Credentials and the
                query string are removed.
            now_ms: Time of the connection in epoch milliseconds; now when
                omitted.
        """
        if not self.enabled:
            return
        try:
            moment = now_ms if now_ms is not None else _now_ms()
            link_role = ConnectionRole(role)
            link_direction = ConnectionDirectionId(direction)
            self._emit_expired(moment)
            key = _link_key(link_role, link_direction, service, src, dst)
            self._record_window(key, src, dst, _redact_url(url), moment)
        except Exception:
            self._record_emit_failure("record a connection")

    def record_connection_failure(
        self,
        *,
        role: ConnectionRole | str,
        service: str | None,
        reason: BaseException | str,
        direction: ConnectionDirectionId = ConnectionDirectionId.OUTBOUND,
        dst: Endpoint | None = None,
        src: Endpoint | None = None,
        url: str | None = None,
        activity_id: NetworkActivityId | None = None,
        status_code: str | None = None,
        now_ms: int | None = None,
    ) -> None:
        """Record one failed connection on one of the host's links, with its reason.

        Never coalesced. The reason is reduced to a bounded line of text with
        credentials in URLs removed; see :func:`link_outcome`.

        Args:
            role: Which link the connection belongs to.
            service: Service label for ``dst_endpoint.svc_name``.
            reason: The exception the connection failed with, or reason text.
            direction: ``OUTBOUND`` or ``INBOUND``.
            dst: The responder. Required for the event to be reported.
            src: The initiator, when its address is known.
            url: URL the connection targeted, if any.
            activity_id: Overrides the classified activity; ``OPEN`` reports
                a connection that opened and whose exchange then failed.
            status_code: The peer's own code, when it gave one.
            now_ms: Time of the failure in epoch milliseconds; now when
                omitted.
        """
        if not self.enabled:
            return
        try:
            moment = now_ms if now_ms is not None else _now_ms()
            link_role = ConnectionRole(role)
            try:
                self._emit_expired(moment)
            except Exception:
                self._record_emit_failure("close an expired window")
            outcome = link_outcome(reason, activity_id=activity_id, status_code=status_code)
            self._emit(
                _failure_event(
                    outcome,
                    metadata=self._metadata[link_role],
                    src=src,
                    dst=dst,
                    url=_redact_url(url),
                    service=service,
                    direction=ConnectionDirectionId(direction),
                    time=moment,
                )
            )
        except Exception:
            self._record_emit_failure("record a connection failure")

    def flush_expired(self, now_ms: int | None = None) -> None:
        """Emit every success window whose length has elapsed, then send held events.

        Windows also close on any later activity through the emitter. A host
        calls this on a timer so a link that went quiet is still reported
        within about one window length, and so events held during a broker
        outage go out once it ends.
        """
        if not self.enabled:
            return
        try:
            self._emit_expired(now_ms if now_ms is not None else _now_ms())
        except Exception:
            self._record_emit_failure("close expired windows")
        self.release_held()

    def flush(self) -> None:
        """Emit every open success window.

        Called when the client or the host closes: successes accumulated since
        the last window closed are still attempts the connection log accounts
        for.
        """
        if not self.enabled:
            return
        try:
            for key in list(self._links):
                self._close_link(key)
        except Exception:
            self._record_emit_failure("flush")
        # With per-attempt logging there may be no open window to close, so
        # closing windows alone would not send events held during an outage.
        self.release_held()

    def release_held(self) -> None:
        """Send held events if the transport can deliver them now."""
        if self._forwarder is None or not self._held:
            return
        try:
            if not self._transport_ready():
                return
            while self._held:
                self._forwarder.queue_event(self._held.popleft())
        except Exception:
            self._record_emit_failure("send held events")

    # Internals.

    def _record_window(
        self,
        key: _LinkKey,
        src: Endpoint | None,
        dst: Endpoint | None,
        url: str | None,
        moment: int,
    ) -> None:
        """Add one success to its link's window, emitting the window when it closes."""
        link = self._links.get(key)
        if link is None:
            if len(self._links) >= MAX_OPEN_WINDOWS:
                # Make room by reporting the oldest window early.
                self._close_link(next(iter(self._links)))
            link = _OpenLink(CoalescingWindow(self._window_seconds), url)
            self._links[key] = link
        closed = link.window.record(moment, src, dst)
        if closed is not None:
            self._emit_window(key, closed, link.url)
            link.url = url
        if link.window.pending == 0:
            del self._links[key]

    def _emit_expired(self, now_ms: int) -> None:
        """Emit and forget every window whose length has elapsed."""
        for key, link in list(self._links.items()):
            closed = link.window.take_expired(now_ms)
            if closed is not None:
                del self._links[key]
                self._emit_window(key, closed, link.url)

    def _close_link(self, key: _LinkKey) -> None:
        """Emit the open window of one link and forget the link."""
        link = self._links.pop(key)
        closed = link.window.flush()
        if closed is not None:
            self._emit_window(key, closed, link.url)

    def _emit_window(self, key: _LinkKey, window: Window, url: str | None) -> None:
        """Build and send the event for one closed success window."""
        northbound = key == _NORTHBOUND_KEY
        self._emit(
            build_coalesced_success_event(
                window,
                metadata=self._metadata[key.role],
                server_endpoint=self._server_endpoint if northbound else None,
                url=self._base_url if northbound else url,
                service=key.service,
                direction=key.direction,
            )
        )

    def _transport_ready(self) -> bool:
        """Return whether an event queued now would reach a running forwarder.

        A forwarder that does not report this is assumed ready.
        """
        return bool(getattr(self._forwarder, "can_deliver_events", True))

    def _record_emit_failure(self, what: str) -> None:
        """Count an emission failure, and say so the first time.

        Swallowing is required -- telemetry must not break the connection it
        describes -- but swallowing silently means an operator relying on
        these records cannot tell a client with no failures from one whose
        telemetry stopped working.
        """
        self.emit_failures += 1
        if self.emit_failures == 1:
            logger.warning(
                "Connection telemetry failed to %s; further failures log at debug",
                what,
                exc_info=True,
            )
        else:
            logger.debug("Connection telemetry failed to %s", what, exc_info=True)

    def _emit(self, event: NetworkActivity | None) -> None:
        """Send one event, or hold it while the transport cannot deliver it."""
        if event is None or self._forwarder is None:
            return
        frame = EventFrame(
            payload=event.to_dict(),
            topic_suffix=self._config.topic_suffix,
            kind="connection-event",
        )
        if not self._transport_ready():
            self._hold(frame)
            return
        self.release_held()
        self._forwarder.queue_event(frame)

    def _hold(self, frame: EventFrame) -> None:
        """Keep one event until the transport can deliver it; drop the oldest at the bound."""
        if self._max_held == 0:
            self._count_held_drop()
            return
        if len(self._held) >= self._max_held:
            self._held.popleft()
            self._count_held_drop()
        self._held.append(frame)

    def _count_held_drop(self) -> None:
        """Count one held event lost to the bound, and warn the first time."""
        self.held_events_dropped += 1
        if self.held_events_dropped == 1:
            logger.warning(
                "Connection telemetry hold is full (%d events) while the transport is "
                "unavailable; dropping the oldest",
                self._max_held,
            )

enabled property

enabled: bool

Whether events will actually be emitted.

coalescing property

coalescing: bool

Whether successes are collapsed into windows rather than sent one by one.

held_events property

held_events: int

Events waiting for the transport to become able to deliver them.

set_server

set_server(host: str | None, port: int | None, base_url: str | None = None) -> None

Record the server this client targets, for events with no live socket.

A connection that never established has no socket to report, but the address the client was trying to reach is exactly what the failure record needs to name.

Source code in src/py20305/forwarders/connection_telemetry.py

def set_server(self, host: str | None, port: int | None, base_url: str | None = None) -> None:
    """Record the server this client targets, for events with no live socket.

    A connection that never established has no socket to report, but the
    address the client was trying to reach is exactly what the failure
    record needs to name.
    """
    self._base_url = _redact_url(base_url)
    self._server_endpoint = endpoint_for_host(host, port) if port else None

begin_request

begin_request() -> None

Clear any socket left from an earlier request in this task.

Scopes attribution to "a connection established while handling this request". Without it, a task that opens a connection and then makes a second request over the pool would attribute the first connection's local port to the second request -- right only by luck, and wrong as soon as the pool hands back a different connection.

Source code in src/py20305/forwarders/connection_telemetry.py

def begin_request(self) -> None:
    """Clear any socket left from an earlier request in this task.

    Scopes attribution to "a connection established while handling *this*
    request". Without it, a task that opens a connection and then makes a
    second request over the pool would attribute the first connection's
    local port to the second request -- right only by luck, and wrong as
    soon as the pool hands back a different connection.
    """
    _request_socket.set(None)

on_connect

on_connect(pair: SocketPair) -> None

Observe a newly established connection's socket.

Reached through the client's connector, which is the only place the local port is available.

Source code in src/py20305/forwarders/connection_telemetry.py

def on_connect(self, pair: SocketPair) -> None:
    """Observe a newly established connection's socket.

    Reached through the client's connector, which is the only place the
    local port is available.
    """
    _request_socket.set(pair)

record_success

record_success(now_ms: int | None = None) -> None

Record one successful IEEE 2030.5 exchange with the server.

Source code in src/py20305/forwarders/connection_telemetry.py

def record_success(self, now_ms: int | None = None) -> None:
    """Record one successful IEEE 2030.5 exchange with the server."""
    if not self.enabled:
        return
    try:
        moment = now_ms if now_ms is not None else _now_ms()
        self._emit_expired(moment)
        src, dst = _endpoints(_request_socket.get(), None)
        self._record_window(_NORTHBOUND_KEY, src, dst, self._base_url, moment)
    except Exception:
        self._record_emit_failure("record a success")

record_failure

record_failure(exc: BaseException) -> None

Record one failed IEEE 2030.5 connection or exchange, with its reason.

Never coalesced: each failure keeps its own event and its own reason.

Source code in src/py20305/forwarders/connection_telemetry.py

def record_failure(self, exc: BaseException) -> None:
    """Record one failed IEEE 2030.5 connection or exchange, with its reason.

    Never coalesced: each failure keeps its own event and its own reason.
    """
    if not self.enabled:
        return
    try:
        # The failure is also the clock's chance to close expired success
        # windows: those successes happened before this failure, and
        # holding them until the next success -- which may never come --
        # would leave them unreported until shutdown. Guarded on its own:
        # a problem emitting a window must not cost the failure record,
        # which is the more important of the two.
        try:
            self._emit_expired(_now_ms())
        except Exception:
            self._record_emit_failure("close an expired window")
        outcome = classify(exc)
        if outcome is None:
            return
        # The retained socket belongs to a connection this request
        # established. For an exchange failure (OPEN) or a teardown
        # (RESET) that connection is the subject of the record; for a
        # failure that never opened (FAIL, REFUSE) it is an earlier
        # attempt's socket, and attaching its local port would attribute
        # the failure to a connection that did not fail.
        socket_pair = (
            _request_socket.get()
            if outcome.activity_id in (NetworkActivityId.OPEN, NetworkActivityId.RESET)
            else None
        )
        self._emit(
            build_failure_event(
                outcome,
                metadata=self._metadata[ConnectionRole.NORTHBOUND],
                socket_pair=socket_pair,
                server_endpoint=self._server_endpoint,
                url=self._base_url,
            )
        )
    except Exception:
        self._record_emit_failure("record a failure")

record_connection

record_connection(*, role: ConnectionRole | str, service: str | None, direction: ConnectionDirectionId = ConnectionDirectionId.OUTBOUND, dst: Endpoint | None = None, src: Endpoint | None = None, url: str | None = None, now_ms: int | None = None) -> None

Record one successful connection on one of the host's links.

Parameters:

Name	Type	Description	Default
role	ConnectionRole | str	Which link the connection belongs to.	required
service	str | None	Service label for dst_endpoint.svc_name, for example "modbus".	required
direction	ConnectionDirectionId	OUTBOUND for a connection the host opened, INBOUND for one it accepted.	OUTBOUND
dst	Endpoint | None	The responder: the device or broker for an outbound connection, the host's own listener for an inbound one. Required for the event to be reported.	None
src	Endpoint | None	The initiator, when its address is known.	None
url	str | None	URL the connection targeted, if any. Credentials and the query string are removed.	None
now_ms	int | None	Time of the connection in epoch milliseconds; now when omitted.	None

Source code in src/py20305/forwarders/connection_telemetry.py

def record_connection(
    self,
    *,
    role: ConnectionRole | str,
    service: str | None,
    direction: ConnectionDirectionId = ConnectionDirectionId.OUTBOUND,
    dst: Endpoint | None = None,
    src: Endpoint | None = None,
    url: str | None = None,
    now_ms: int | None = None,
) -> None:
    """Record one successful connection on one of the host's links.

    Args:
        role: Which link the connection belongs to.
        service: Service label for ``dst_endpoint.svc_name``, for example
            ``"modbus"``.
        direction: ``OUTBOUND`` for a connection the host opened,
            ``INBOUND`` for one it accepted.
        dst: The responder: the device or broker for an outbound
            connection, the host's own listener for an inbound one.
            Required for the event to be reported.
        src: The initiator, when its address is known.
        url: URL the connection targeted, if any. Credentials and the
            query string are removed.
        now_ms: Time of the connection in epoch milliseconds; now when
            omitted.
    """
    if not self.enabled:
        return
    try:
        moment = now_ms if now_ms is not None else _now_ms()
        link_role = ConnectionRole(role)
        link_direction = ConnectionDirectionId(direction)
        self._emit_expired(moment)
        key = _link_key(link_role, link_direction, service, src, dst)
        self._record_window(key, src, dst, _redact_url(url), moment)
    except Exception:
        self._record_emit_failure("record a connection")

record_connection_failure

record_connection_failure(*, role: ConnectionRole | str, service: str | None, reason: BaseException | str, direction: ConnectionDirectionId = ConnectionDirectionId.OUTBOUND, dst: Endpoint | None = None, src: Endpoint | None = None, url: str | None = None, activity_id: NetworkActivityId | None = None, status_code: str | None = None, now_ms: int | None = None) -> None

Record one failed connection on one of the host's links, with its reason.

Never coalesced. The reason is reduced to a bounded line of text with credentials in URLs removed; see :func:link_outcome.

Parameters:

Name	Type	Description	Default
role	ConnectionRole | str	Which link the connection belongs to.	required
service	str | None	Service label for dst_endpoint.svc_name.	required
reason	BaseException | str	The exception the connection failed with, or reason text.	required
direction	ConnectionDirectionId	OUTBOUND or INBOUND.	OUTBOUND
dst	Endpoint | None	The responder. Required for the event to be reported.	None
src	Endpoint | None	The initiator, when its address is known.	None
url	str | None	URL the connection targeted, if any.	None
activity_id	NetworkActivityId | None	Overrides the classified activity; OPEN reports a connection that opened and whose exchange then failed.	None
status_code	str | None	The peer's own code, when it gave one.	None
now_ms	int | None	Time of the failure in epoch milliseconds; now when omitted.	None

Source code in src/py20305/forwarders/connection_telemetry.py

def record_connection_failure(
    self,
    *,
    role: ConnectionRole | str,
    service: str | None,
    reason: BaseException | str,
    direction: ConnectionDirectionId = ConnectionDirectionId.OUTBOUND,
    dst: Endpoint | None = None,
    src: Endpoint | None = None,
    url: str | None = None,
    activity_id: NetworkActivityId | None = None,
    status_code: str | None = None,
    now_ms: int | None = None,
) -> None:
    """Record one failed connection on one of the host's links, with its reason.

    Never coalesced. The reason is reduced to a bounded line of text with
    credentials in URLs removed; see :func:`link_outcome`.

    Args:
        role: Which link the connection belongs to.
        service: Service label for ``dst_endpoint.svc_name``.
        reason: The exception the connection failed with, or reason text.
        direction: ``OUTBOUND`` or ``INBOUND``.
        dst: The responder. Required for the event to be reported.
        src: The initiator, when its address is known.
        url: URL the connection targeted, if any.
        activity_id: Overrides the classified activity; ``OPEN`` reports
            a connection that opened and whose exchange then failed.
        status_code: The peer's own code, when it gave one.
        now_ms: Time of the failure in epoch milliseconds; now when
            omitted.
    """
    if not self.enabled:
        return
    try:
        moment = now_ms if now_ms is not None else _now_ms()
        link_role = ConnectionRole(role)
        try:
            self._emit_expired(moment)
        except Exception:
            self._record_emit_failure("close an expired window")
        outcome = link_outcome(reason, activity_id=activity_id, status_code=status_code)
        self._emit(
            _failure_event(
                outcome,
                metadata=self._metadata[link_role],
                src=src,
                dst=dst,
                url=_redact_url(url),
                service=service,
                direction=ConnectionDirectionId(direction),
                time=moment,
            )
        )
    except Exception:
        self._record_emit_failure("record a connection failure")

flush_expired

flush_expired(now_ms: int | None = None) -> None

Emit every success window whose length has elapsed, then send held events.

Windows also close on any later activity through the emitter. A host calls this on a timer so a link that went quiet is still reported within about one window length, and so events held during a broker outage go out once it ends.

Source code in src/py20305/forwarders/connection_telemetry.py

def flush_expired(self, now_ms: int | None = None) -> None:
    """Emit every success window whose length has elapsed, then send held events.

    Windows also close on any later activity through the emitter. A host
    calls this on a timer so a link that went quiet is still reported
    within about one window length, and so events held during a broker
    outage go out once it ends.
    """
    if not self.enabled:
        return
    try:
        self._emit_expired(now_ms if now_ms is not None else _now_ms())
    except Exception:
        self._record_emit_failure("close expired windows")
    self.release_held()

flush

flush() -> None

Emit every open success window.

Called when the client or the host closes: successes accumulated since the last window closed are still attempts the connection log accounts for.

Source code in src/py20305/forwarders/connection_telemetry.py

def flush(self) -> None:
    """Emit every open success window.

    Called when the client or the host closes: successes accumulated since
    the last window closed are still attempts the connection log accounts
    for.
    """
    if not self.enabled:
        return
    try:
        for key in list(self._links):
            self._close_link(key)
    except Exception:
        self._record_emit_failure("flush")
    # With per-attempt logging there may be no open window to close, so
    # closing windows alone would not send events held during an outage.
    self.release_held()

release_held

release_held() -> None

Send held events if the transport can deliver them now.

Source code in src/py20305/forwarders/connection_telemetry.py

def release_held(self) -> None:
    """Send held events if the transport can deliver them now."""
    if self._forwarder is None or not self._held:
        return
    try:
        if not self._transport_ready():
            return
        while self._held:
            self._forwarder.queue_event(self._held.popleft())
    except Exception:
        self._record_emit_failure("send held events")

role_label

role_label(role: ConnectionRole | str) -> str

Return the metadata.labels entry that names role.

Source code in src/py20305/forwarders/connection_telemetry.py

def role_label(role: ConnectionRole | str) -> str:
    """Return the ``metadata.labels`` entry that names ``role``."""
    return f"{ROLE_LABEL_PREFIX}{ConnectionRole(role).value}"

classify

classify(exc: BaseException) -> Outcome | None

Classify an exception from the request path into an OCSF outcome.

The distinction that matters to whoever reads the log is whether the connection was ever established:

• A transport failure -- connect, timeout, TLS handshake -- means nothing was exchanged, and reports FAIL. A refused connection reports REFUSE, which is more specific and is what the peer actually did.
• An application-layer failure over a connection that did open -- a 500, a 429, a redirect loop, an unusable body -- reports OPEN with a failure status. Reporting it as FAIL would tell a reader the client never reached the server, which would send an investigation the wrong way.

Returns:

Type	Description
Outcome | None	The classified outcome, or None when the exception is not a
Outcome | None	connection outcome at all and nothing should be reported.

Source code in src/py20305/forwarders/connection_telemetry.py

def classify(exc: BaseException) -> Outcome | None:
    """Classify an exception from the request path into an OCSF outcome.

    The distinction that matters to whoever reads the log is whether the
    connection was ever established:

    - A transport failure -- connect, timeout, TLS handshake -- means nothing
      was exchanged, and reports ``FAIL``. A refused connection reports
      ``REFUSE``, which is more specific and is what the peer actually did.
    - An application-layer failure over a connection that *did* open -- a 500,
      a 429, a redirect loop, an unusable body -- reports ``OPEN`` with a
      failure status. Reporting it as ``FAIL`` would tell a reader the client
      never reached the server, which would send an investigation the wrong
      way.

    Returns:
        The classified outcome, or ``None`` when the exception is not a
        connection outcome at all and nothing should be reported.
    """
    # A 204 is a success with no body, not an error. It subclasses
    # Sep2ProtocolError, so it has to be tested before its parent or every
    # empty optional resource would be logged as a failed exchange.
    if isinstance(exc, Sep2NoContentError):
        return None

    if isinstance(exc, Sep2TlsError):
        return Outcome(NetworkActivityId.FAIL, f"TLS failure: {exc}")

    if isinstance(exc, Sep2ConnectionError):
        # The cause chain distinguishes the transport failures OCSF names.
        # A refusal means the host answered and declined; a disconnect or
        # reset means a connection *opened* and was torn down -- reporting
        # that as Fail would tell a reader the client never reached the
        # server, which is not what the socket saw.
        if _has_cause(exc, ConnectionRefusedError):
            return Outcome(NetworkActivityId.REFUSE, f"Connection refused: {exc}")
        if _has_cause(exc, (aiohttp.ServerDisconnectedError, ConnectionResetError)):
            return Outcome(NetworkActivityId.RESET, f"Connection reset by the server: {exc}")
        return Outcome(NetworkActivityId.FAIL, f"Connection failure: {exc}")

    if isinstance(exc, Sep2RateLimitError):
        retry_after = getattr(exc, "retry_after", None)
        suffix = f", retry after {retry_after}s" if retry_after is not None else ""
        return Outcome(NetworkActivityId.OPEN, f"Rate limited by the server{suffix}: {exc}", "429")

    if isinstance(exc, Sep2RedirectError):
        # The Location header is peer-controlled: it can carry userinfo or
        # signed query parameters, so only a stripped-down reference to it is
        # reported -- and the exception text, which embeds the raw header, is
        # not interpolated at all.
        status = getattr(exc, "status_code", None)
        location = getattr(exc, "location", None)
        where = f" to {_safe_url_reference(location)}" if location else ""
        return Outcome(
            NetworkActivityId.OPEN,
            f"Redirected{where}, triggering re-discovery",
            str(status) if status is not None else None,
        )

    if isinstance(exc, Sep2PayloadError):
        # The exception text can embed the unparseable body; identify the
        # failure by where and how big instead of by content.
        path = getattr(exc, "path", None)
        size = getattr(exc, "body_length", None)
        at = f" from GET {path}" if path else ""
        of = f" ({size} bytes)" if size is not None else ""
        return Outcome(NetworkActivityId.OPEN, f"Unusable response body{at}{of}")

    if isinstance(exc, Sep2ProtocolError):
        # The exception text embeds the peer's response body, and a length
        # cap limits volume but not content -- the status code alone is what
        # identifies this failure, so nothing else is reported.
        status = getattr(exc, "status_code", None)
        detail = "Server returned an unexpected status"
        if status is not None:
            detail += f": HTTP {status}"
        return Outcome(
            NetworkActivityId.OPEN,
            detail,
            str(status) if status is not None else None,
        )

    if isinstance(exc, ssl.SSLError):
        return Outcome(NetworkActivityId.FAIL, f"TLS failure: {exc}")

    if isinstance(exc, ConnectionRefusedError):
        return Outcome(NetworkActivityId.REFUSE, f"Connection refused: {exc}")

    if isinstance(exc, ConnectionResetError):
        return Outcome(NetworkActivityId.RESET, f"Connection reset by the server: {exc}")

    if isinstance(exc, TimeoutError):
        return Outcome(NetworkActivityId.FAIL, f"Connection timed out: {exc}")

    if isinstance(exc, OSError):
        return Outcome(NetworkActivityId.FAIL, f"Connection failure: {exc}")

    return None

build_metadata

build_metadata(product_version: str | None = None, *, role: ConnectionRole | str = ConnectionRole.NORTHBOUND, product_name: str = PRODUCT_NAME, vendor_name: str = DEFAULT_VENDOR_NAME) -> Metadata

Build the OCSF metadata block naming the producer and the link's role.

Parameters:

Name	Type	Description	Default
product_version	str | None	Version of the producing application.	None
role	ConnectionRole | str	The link the event describes, recorded as one role:<role> label.	NORTHBOUND
product_name	str	Name of the producing application. Defaults to this library; a host application passes its own name so the event identifies the product that actually emitted it.	PRODUCT_NAME
vendor_name	str	Publisher of the producing application.	DEFAULT_VENDOR_NAME

Source code in src/py20305/forwarders/connection_telemetry.py

def build_metadata(
    product_version: str | None = None,
    *,
    role: ConnectionRole | str = ConnectionRole.NORTHBOUND,
    product_name: str = PRODUCT_NAME,
    vendor_name: str = DEFAULT_VENDOR_NAME,
) -> Metadata:
    """Build the OCSF ``metadata`` block naming the producer and the link's role.

    Args:
        product_version: Version of the producing application.
        role: The link the event describes, recorded as one ``role:<role>``
            label.
        product_name: Name of the producing application. Defaults to this
            library; a host application passes its own name so the event
            identifies the product that actually emitted it.
        vendor_name: Publisher of the producing application.
    """
    return Metadata(
        product=Product(name=product_name, vendor_name=vendor_name, version=product_version),
        labels=[role_label(role)],
    )

endpoint_for_host

endpoint_for_host(host: str | None, port: int | None) -> Endpoint | None

Build an endpoint for a configured host, which may be an address or a name.

OCSF types the endpoint's ip as an IP address. A configured host is often a DNS name, and a name in ip is a schema violation a consumer may reject, so a name goes in hostname.

Returns:

Type	Description
Endpoint | None	The endpoint, or None when host is empty.

Source code in src/py20305/forwarders/connection_telemetry.py

def endpoint_for_host(host: str | None, port: int | None) -> Endpoint | None:
    """Build an endpoint for a configured host, which may be an address or a name.

    OCSF types the endpoint's ``ip`` as an IP address. A configured host is
    often a DNS name, and a name in ``ip`` is a schema violation a consumer may
    reject, so a name goes in ``hostname``.

    Returns:
        The endpoint, or ``None`` when ``host`` is empty.
    """
    if not host:
        return None
    try:
        ipaddress.ip_address(host)
    except ValueError:
        return Endpoint(hostname=host, port=port)
    return Endpoint(ip=host, port=port)

build_failure_event

build_failure_event(outcome: Outcome, *, metadata: Metadata, socket_pair: SocketPair | None, server_endpoint: Endpoint | None, url: str | None = None, service: str | None = SERVICE_LABEL, direction: ConnectionDirectionId = ConnectionDirectionId.OUTBOUND, time: int | None = None) -> NetworkActivity | None

Build one failure event. Failures are never coalesced.

Returns None when there is no destination to report -- an event with neither endpoint says nothing about a connection, and the event class rejects it.

Source code in src/py20305/forwarders/connection_telemetry.py

def build_failure_event(
    outcome: Outcome,
    *,
    metadata: Metadata,
    socket_pair: SocketPair | None,
    server_endpoint: Endpoint | None,
    url: str | None = None,
    service: str | None = SERVICE_LABEL,
    direction: ConnectionDirectionId = ConnectionDirectionId.OUTBOUND,
    time: int | None = None,
) -> NetworkActivity | None:
    """Build one failure event. Failures are never coalesced.

    Returns ``None`` when there is no destination to report -- an event with
    neither endpoint says nothing about a connection, and the event class
    rejects it.
    """
    src, dst = _endpoints(socket_pair, server_endpoint)
    return _failure_event(
        outcome,
        metadata=metadata,
        src=src,
        dst=dst,
        url=url,
        service=service,
        direction=direction,
        time=time,
    )

build_coalesced_success_event

build_coalesced_success_event(window: Window, *, metadata: Metadata, server_endpoint: Endpoint | None, url: str | None = None, service: str | None = SERVICE_LABEL, direction: ConnectionDirectionId = ConnectionDirectionId.OUTBOUND) -> NetworkActivity | None

Build one event standing for a window of successful connections.

server_endpoint is the destination reported when the window observed none, as for a request that reused a pooled connection.

Source code in src/py20305/forwarders/connection_telemetry.py

def build_coalesced_success_event(
    window: Window,
    *,
    metadata: Metadata,
    server_endpoint: Endpoint | None,
    url: str | None = None,
    service: str | None = SERVICE_LABEL,
    direction: ConnectionDirectionId = ConnectionDirectionId.OUTBOUND,
) -> NetworkActivity | None:
    """Build one event standing for a window of successful connections.

    ``server_endpoint`` is the destination reported when the window observed
    none, as for a request that reused a pooled connection.
    """
    count, start_ms, end_ms = window.count, window.start_ms, window.end_ms
    src = window.src
    dst = window.dst if window.dst is not None else server_endpoint
    if dst is None:
        logger.debug("No destination endpoint for connection successes; not reporting")
        return None
    if count == 1:
        return NetworkActivity.for_connection_success(
            metadata=metadata,
            dst_endpoint=dst,
            src_endpoint=src,
            service=service,
            url=url,
            time=end_ms,
            connection_direction=direction,
        )
    return NetworkActivity.for_coalesced_successes(
        metadata=metadata,
        dst_endpoint=dst,
        src_endpoint=src,
        count=count,
        start_time=start_ms,
        end_time=end_ms,
        service=service,
        url=url,
        connection_direction=direction,
    )

link_outcome

link_outcome(reason: BaseException | str, *, activity_id: NetworkActivityId | None = None, status_code: str | None = None) -> Outcome

Classify a failure the host reported on one of its own links.

Unlike :func:classify, this always returns an outcome: the caller has already decided the exception is a connection failure. An exception the IEEE 2030.5 taxonomy does not know is reported as FAIL with its type and message. Credentials in any URL inside the reason are removed.

Parameters:

Name	Type	Description	Default
reason	BaseException | str	The exception, or the reason text.	required
activity_id	NetworkActivityId | None	Overrides the classified activity. OPEN reports a connection that opened and whose exchange then failed.	None
status_code	str | None	The peer's own code, when it gave one.	None

Raises:

Type	Description
ValueError	If activity_id is not a failure activity or OPEN.

Source code in src/py20305/forwarders/connection_telemetry.py

def link_outcome(
    reason: BaseException | str,
    *,
    activity_id: NetworkActivityId | None = None,
    status_code: str | None = None,
) -> Outcome:
    """Classify a failure the host reported on one of its own links.

    Unlike :func:`classify`, this always returns an outcome: the caller has
    already decided the exception is a connection failure. An exception the
    IEEE 2030.5 taxonomy does not know is reported as ``FAIL`` with its type
    and message. Credentials in any URL inside the reason are removed.

    Args:
        reason: The exception, or the reason text.
        activity_id: Overrides the classified activity. ``OPEN`` reports a
            connection that opened and whose exchange then failed.
        status_code: The peer's own code, when it gave one.

    Raises:
        ValueError: If ``activity_id`` is not a failure activity or ``OPEN``.
    """
    if activity_id is not None and NetworkActivityId(activity_id) not in _REPORTABLE_ACTIVITIES:
        raise ValueError(f"activity_id {activity_id!r} does not describe a failed connection")
    if isinstance(reason, BaseException):
        classified = _classify_chain(reason)
        if classified is None:
            classified = Outcome(NetworkActivityId.FAIL, f"Connection failure: {_describe(reason)}")
        detail = classified.detail
        activity = classified.activity_id
        code = classified.status_code
    else:
        detail = reason.strip() or "Connection failure"
        activity = NetworkActivityId.FAIL
        code = None
    return Outcome(
        NetworkActivityId(activity_id) if activity_id is not None else activity,
        detail,
        status_code if status_code is not None else code,
    )

py20305.forwarders.audit

The control audit trail: event lifecycle and loss-of-communications records.

A utility program asks of a control client what it did with each control it was sent: when an event was scheduled, when it went active and on which devices, how it ended, and when the client stopped acting on the server because the server went quiet. The protocol exchanges themselves are already forwarded, but they record what the server said, not what this client decided. These records are the decisions.

Each record is a flat JSON document published on its own topic, so a search index stores every field as a field and an audit query hits one small index.

Delivery is best-effort, as for every stream on this transport. What this module adds is the ability to tell a complete trail from an incomplete one: every record carries boot_id and seq, one counter per client shared with that client's device-write records, so a gap in seq marks a lost record and a new boot_id marks a restart.

Nothing here may raise into event processing. A trail that can stop a control from being applied is worse than no trail.

AuditSequence

One client's audit counter: a random boot_id and a rising seq.

Shared by that client's device-write and lifecycle records, so a single gap check covers both streams. One per client rather than per process: two clients interleaving on one counter would each show a gap at every other record, which reads as loss.

Source code in src/py20305/forwarders/audit.py

class AuditSequence:
    """One client's audit counter: a random ``boot_id`` and a rising ``seq``.

    Shared by that client's device-write and lifecycle records, so a single gap
    check covers both streams. One per client rather than per process: two
    clients interleaving on one counter would each show a gap at every other
    record, which reads as loss.
    """

    def __init__(self) -> None:
        self.boot_id = uuid.uuid4().hex
        # next() on itertools.count is atomic under the GIL, so a write
        # recorded from a worker thread cannot reuse a number.
        self._counter = itertools.count(1)

    def stamp(self) -> dict[str, Any]:
        """The next ``boot_id`` and ``seq`` pair. Each call consumes a number."""
        return {"boot_id": self.boot_id, "seq": next(self._counter)}

stamp

stamp() -> dict[str, Any]

The next boot_id and seq pair. Each call consumes a number.

Source code in src/py20305/forwarders/audit.py

def stamp(self) -> dict[str, Any]:
    """The next ``boot_id`` and ``seq`` pair. Each call consumes a number."""
    return {"boot_id": self.boot_id, "seq": next(self._counter)}

AuditEmitter

Publishes lifecycle and loss-of-communications records.

Built unconfigured and disabled, then pointed at the transport and the operator's configuration once both exist -- the same two-step wiring the device telemetry emitter uses, for the same reason: the forwarder is built after the client.

Source code in src/py20305/forwarders/audit.py

class AuditEmitter:
    """Publishes lifecycle and loss-of-communications records.

    Built unconfigured and disabled, then pointed at the transport and the
    operator's configuration once both exist -- the same two-step wiring the
    device telemetry emitter uses, for the same reason: the forwarder is
    built after the client.
    """

    def __init__(
        self,
        forwarder: ForwarderManager | None = None,
        config: AuditConfig | None = None,
        *,
        client_id: str | None = None,
        sequence: AuditSequence | None = None,
    ) -> None:
        from py20305.forwarders.config import AuditConfig

        self._forwarder = forwarder
        self._config = config if config is not None else AuditConfig()
        self._client_id = client_id or ""
        self._sequence = sequence if sequence is not None else AuditSequence()
        #: Records that failed to build or queue, by kind. A trail that
        #: stopped working must not look like a client with nothing to report.
        self.emit_failures: dict[str, int] = {}

    @property
    def enabled(self) -> bool:
        """Whether records will actually be published."""
        return self._config.enabled and self._forwarder is not None

    @property
    def sequence(self) -> AuditSequence:
        """This client's counter. Hand it to the client's device telemetry
        emitter so device writes number in the same stream."""
        return self._sequence

    def attach_forwarder(self, forwarder: ForwarderManager | None) -> None:
        """Point the emitter at the transport once one exists."""
        self._forwarder = forwarder

    def configure(self, config: AuditConfig, *, client_id: str | None = None) -> None:
        """Apply operator configuration after construction."""
        self._config = config
        if client_id is not None:
            self._client_id = client_id

    def event_transition(
        self,
        *,
        event_mrid: bytes,
        program_href: str,
        from_state: str | None,
        to_state: str,
        effective_start: int,
        effective_duration: int,
        primacy: int,
        lfdis: list[str],
        at: float,
        applied_lfdis: list[str] | None = None,
        rejected_lfdis: list[str] | None = None,
        superseded_by: bytes | None = None,
        superseded_lfdis: list[str] | None = None,
        superseded_modes: dict[str, list[str]] | None = None,
    ) -> None:
        """Record one event moving from one state to another.

        Args:
            event_mrid: The event's mRID.
            program_href: The program the event belongs to.
            from_state: The state before, or ``None`` for an event first seen.
            to_state: ``scheduled``, ``active``, ``completed``, ``cancelled``,
                ``superseded``, ``opted_out``, ``expired`` (already over when
                first received) or ``skipped`` (inside a loss-of-communications
                window the client opted out of).
            effective_start: Start after randomization, epoch seconds.
            effective_duration: Duration after randomization, seconds.
            primacy: The program's primacy.
            lfdis: The devices the event targets.
            at: When the transition happened, on the server's timebase.
            applied_lfdis: On ``active``, the devices whose dispatch returned
                without error -- what was reported to the server as started.
            rejected_lfdis: On ``active``, the devices whose dispatch failed or
                missed the activation deadline.
            superseded_by: On ``superseded``, the superseding event.
            superseded_lfdis: On a partial supersession, the devices it covers.
            superseded_modes: On a partial supersession, the modes per device.
        """
        if not self.enabled:
            return
        try:
            record: dict[str, Any] = {
                "kind": "der_event",
                "event_mrid": mrid_hex(event_mrid),
                "program_href": program_href,
                "from_state": from_state,
                "to_state": to_state,
                "effective_start": effective_start,
                "effective_duration": effective_duration,
                "primacy": primacy,
                "lfdis": sorted(_lfdi(x) for x in lfdis),
                "at": at,
            }
            if applied_lfdis is not None:
                record["applied_lfdis"] = sorted(_lfdi(x) for x in applied_lfdis)
            if rejected_lfdis is not None:
                record["rejected_lfdis"] = sorted(_lfdi(x) for x in rejected_lfdis)
            if superseded_by is not None:
                record["superseded_by"] = mrid_hex(superseded_by)
            if superseded_lfdis is not None:
                record["superseded_lfdis"] = sorted(_lfdi(x) for x in superseded_lfdis)
            if superseded_modes is not None:
                record["superseded_modes"] = {
                    _lfdi(dev): sorted(modes) for dev, modes in superseded_modes.items()
                }
            self._publish(record)
        except Exception:
            self._failed("der_event")

    def comms_loss(
        self,
        *,
        transition: str,
        threshold: int,
        simulated: bool,
        at: float,
        elapsed_seconds: int | None = None,
        duration_seconds: int | None = None,
        resume_after: int | None = None,
    ) -> None:
        """Record a step into or out of loss-of-communications mode.

        Published in the order things happen: ``entered`` before any event is
        opted out, ``recovering`` once contact returns and before the schedule
        is polled again, ``cleared`` once the mode ends. A recovery that fails
        is retried, so ``recovering`` can repeat before one ``cleared``.

        Args:
            transition: ``entered``, ``recovering`` or ``cleared``.
            threshold: The configured silence threshold.
            simulated: Whether an operator-triggered simulation caused it.
            at: When it happened, on the server's timebase.
            elapsed_seconds: On ``entered``, how long the server had been
                silent.
            duration_seconds: On ``recovering`` and ``cleared``, how long the
                client had been in the mode.
            resume_after: On ``cleared``, the epoch after which the schedule
                resumes, when the opted-out window has not yet passed.
        """
        if not self.enabled:
            return
        try:
            record: dict[str, Any] = {
                "kind": "comms_loss",
                "transition": transition,
                "threshold": threshold,
                "simulated": simulated,
                "at": at,
            }
            if transition == "entered":
                record["elapsed_seconds"] = elapsed_seconds
            else:
                record["duration_seconds"] = duration_seconds
            if transition == "cleared":
                record["resume_after"] = resume_after
            self._publish(record)
        except Exception:
            self._failed("comms_loss")

    def late_dispatch(
        self,
        *,
        event_mrid: bytes,
        lfdi: str,
        applied: bool,
        at: float,
        error: str | None = None,
        error_type: str | None = None,
    ) -> None:
        """Record how a dispatch that missed the activation deadline ended.

        The ``active`` record lists such a device as rejected, because that is
        what the server was told. The apply carries on regardless, and this is
        the record of whether it then reached the device.

        Args:
            event_mrid: The event whose control it was.
            lfdi: The device, or its href where no LFDI is known.
            applied: Whether the apply finished without error.
            at: When it finished, on the server's timebase.
            error: The failure, when it failed.
            error_type: The failure's exception class name.
        """
        if not self.enabled:
            return
        try:
            record: dict[str, Any] = {
                "kind": "late_dispatch",
                "event_mrid": mrid_hex(event_mrid),
                "lfdi": _lfdi(lfdi),
                "applied": applied,
                "at": at,
            }
            if error is not None:
                record["error"] = error[:512]
                record["error_type"] = error_type
            self._publish(record)
        except Exception:
            self._failed("late_dispatch")

    def _publish(self, record: dict[str, Any]) -> None:
        assert self._forwarder is not None  # guarded by `enabled`
        record["schema"] = AUDIT_SCHEMA_VERSION
        record["client_id"] = self._client_id
        record.update(self._sequence.stamp())
        self._forwarder.queue_event(
            EventFrame(payload=record, topic_suffix=self._config.topic_suffix, kind="audit")
        )

    def _failed(self, kind: str) -> None:
        count = self.emit_failures.get(kind, 0) + 1
        self.emit_failures[kind] = count
        if count == 1:
            logger.warning(
                "Audit trail failed to record a %s; further failures log at debug",
                kind,
                exc_info=True,
            )
        else:
            logger.debug("Audit trail failed to record a %s", kind, exc_info=True)

enabled property

enabled: bool

Whether records will actually be published.

sequence property

sequence: AuditSequence

This client's counter. Hand it to the client's device telemetry emitter so device writes number in the same stream.

attach_forwarder

attach_forwarder(forwarder: ForwarderManager | None) -> None

Point the emitter at the transport once one exists.

Source code in src/py20305/forwarders/audit.py

def attach_forwarder(self, forwarder: ForwarderManager | None) -> None:
    """Point the emitter at the transport once one exists."""
    self._forwarder = forwarder

configure

configure(config: AuditConfig, *, client_id: str | None = None) -> None

Apply operator configuration after construction.

Source code in src/py20305/forwarders/audit.py

def configure(self, config: AuditConfig, *, client_id: str | None = None) -> None:
    """Apply operator configuration after construction."""
    self._config = config
    if client_id is not None:
        self._client_id = client_id

event_transition

event_transition(*, event_mrid: bytes, program_href: str, from_state: str | None, to_state: str, effective_start: int, effective_duration: int, primacy: int, lfdis: list[str], at: float, applied_lfdis: list[str] | None = None, rejected_lfdis: list[str] | None = None, superseded_by: bytes | None = None, superseded_lfdis: list[str] | None = None, superseded_modes: dict[str, list[str]] | None = None) -> None

Record one event moving from one state to another.

Parameters:

Name	Type	Description	Default
event_mrid	bytes	The event's mRID.	required
program_href	str	The program the event belongs to.	required
from_state	str | None	The state before, or None for an event first seen.	required
to_state	str	scheduled, active, completed, cancelled, superseded, opted_out, expired (already over when first received) or skipped (inside a loss-of-communications window the client opted out of).	required
effective_start	int	Start after randomization, epoch seconds.	required
effective_duration	int	Duration after randomization, seconds.	required
primacy	int	The program's primacy.	required
lfdis	list[str]	The devices the event targets.	required
at	float	When the transition happened, on the server's timebase.	required
applied_lfdis	list[str] | None	On active, the devices whose dispatch returned without error -- what was reported to the server as started.	None
rejected_lfdis	list[str] | None	On active, the devices whose dispatch failed or missed the activation deadline.	None
superseded_by	bytes | None	On superseded, the superseding event.	None
superseded_lfdis	list[str] | None	On a partial supersession, the devices it covers.	None
superseded_modes	dict[str, list[str]] | None	On a partial supersession, the modes per device.	None

Source code in src/py20305/forwarders/audit.py

def event_transition(
    self,
    *,
    event_mrid: bytes,
    program_href: str,
    from_state: str | None,
    to_state: str,
    effective_start: int,
    effective_duration: int,
    primacy: int,
    lfdis: list[str],
    at: float,
    applied_lfdis: list[str] | None = None,
    rejected_lfdis: list[str] | None = None,
    superseded_by: bytes | None = None,
    superseded_lfdis: list[str] | None = None,
    superseded_modes: dict[str, list[str]] | None = None,
) -> None:
    """Record one event moving from one state to another.

    Args:
        event_mrid: The event's mRID.
        program_href: The program the event belongs to.
        from_state: The state before, or ``None`` for an event first seen.
        to_state: ``scheduled``, ``active``, ``completed``, ``cancelled``,
            ``superseded``, ``opted_out``, ``expired`` (already over when
            first received) or ``skipped`` (inside a loss-of-communications
            window the client opted out of).
        effective_start: Start after randomization, epoch seconds.
        effective_duration: Duration after randomization, seconds.
        primacy: The program's primacy.
        lfdis: The devices the event targets.
        at: When the transition happened, on the server's timebase.
        applied_lfdis: On ``active``, the devices whose dispatch returned
            without error -- what was reported to the server as started.
        rejected_lfdis: On ``active``, the devices whose dispatch failed or
            missed the activation deadline.
        superseded_by: On ``superseded``, the superseding event.
        superseded_lfdis: On a partial supersession, the devices it covers.
        superseded_modes: On a partial supersession, the modes per device.
    """
    if not self.enabled:
        return
    try:
        record: dict[str, Any] = {
            "kind": "der_event",
            "event_mrid": mrid_hex(event_mrid),
            "program_href": program_href,
            "from_state": from_state,
            "to_state": to_state,
            "effective_start": effective_start,
            "effective_duration": effective_duration,
            "primacy": primacy,
            "lfdis": sorted(_lfdi(x) for x in lfdis),
            "at": at,
        }
        if applied_lfdis is not None:
            record["applied_lfdis"] = sorted(_lfdi(x) for x in applied_lfdis)
        if rejected_lfdis is not None:
            record["rejected_lfdis"] = sorted(_lfdi(x) for x in rejected_lfdis)
        if superseded_by is not None:
            record["superseded_by"] = mrid_hex(superseded_by)
        if superseded_lfdis is not None:
            record["superseded_lfdis"] = sorted(_lfdi(x) for x in superseded_lfdis)
        if superseded_modes is not None:
            record["superseded_modes"] = {
                _lfdi(dev): sorted(modes) for dev, modes in superseded_modes.items()
            }
        self._publish(record)
    except Exception:
        self._failed("der_event")

comms_loss

comms_loss(*, transition: str, threshold: int, simulated: bool, at: float, elapsed_seconds: int | None = None, duration_seconds: int | None = None, resume_after: int | None = None) -> None

Record a step into or out of loss-of-communications mode.

Published in the order things happen: entered before any event is opted out, recovering once contact returns and before the schedule is polled again, cleared once the mode ends. A recovery that fails is retried, so recovering can repeat before one cleared.

Parameters:

Name	Type	Description	Default
transition	str	entered, recovering or cleared.	required
threshold	int	The configured silence threshold.	required
simulated	bool	Whether an operator-triggered simulation caused it.	required
at	float	When it happened, on the server's timebase.	required
elapsed_seconds	int | None	On entered, how long the server had been silent.	None
duration_seconds	int | None	On recovering and cleared, how long the client had been in the mode.	None
resume_after	int | None	On cleared, the epoch after which the schedule resumes, when the opted-out window has not yet passed.	None

Source code in src/py20305/forwarders/audit.py

def comms_loss(
    self,
    *,
    transition: str,
    threshold: int,
    simulated: bool,
    at: float,
    elapsed_seconds: int | None = None,
    duration_seconds: int | None = None,
    resume_after: int | None = None,
) -> None:
    """Record a step into or out of loss-of-communications mode.

    Published in the order things happen: ``entered`` before any event is
    opted out, ``recovering`` once contact returns and before the schedule
    is polled again, ``cleared`` once the mode ends. A recovery that fails
    is retried, so ``recovering`` can repeat before one ``cleared``.

    Args:
        transition: ``entered``, ``recovering`` or ``cleared``.
        threshold: The configured silence threshold.
        simulated: Whether an operator-triggered simulation caused it.
        at: When it happened, on the server's timebase.
        elapsed_seconds: On ``entered``, how long the server had been
            silent.
        duration_seconds: On ``recovering`` and ``cleared``, how long the
            client had been in the mode.
        resume_after: On ``cleared``, the epoch after which the schedule
            resumes, when the opted-out window has not yet passed.
    """
    if not self.enabled:
        return
    try:
        record: dict[str, Any] = {
            "kind": "comms_loss",
            "transition": transition,
            "threshold": threshold,
            "simulated": simulated,
            "at": at,
        }
        if transition == "entered":
            record["elapsed_seconds"] = elapsed_seconds
        else:
            record["duration_seconds"] = duration_seconds
        if transition == "cleared":
            record["resume_after"] = resume_after
        self._publish(record)
    except Exception:
        self._failed("comms_loss")

late_dispatch

late_dispatch(*, event_mrid: bytes, lfdi: str, applied: bool, at: float, error: str | None = None, error_type: str | None = None) -> None

Record how a dispatch that missed the activation deadline ended.

The active record lists such a device as rejected, because that is what the server was told. The apply carries on regardless, and this is the record of whether it then reached the device.

Parameters:

Name	Type	Description	Default
event_mrid	bytes	The event whose control it was.	required
lfdi	str	The device, or its href where no LFDI is known.	required
applied	bool	Whether the apply finished without error.	required
at	float	When it finished, on the server's timebase.	required
error	str | None	The failure, when it failed.	None
error_type	str | None	The failure's exception class name.	None

Source code in src/py20305/forwarders/audit.py

def late_dispatch(
    self,
    *,
    event_mrid: bytes,
    lfdi: str,
    applied: bool,
    at: float,
    error: str | None = None,
    error_type: str | None = None,
) -> None:
    """Record how a dispatch that missed the activation deadline ended.

    The ``active`` record lists such a device as rejected, because that is
    what the server was told. The apply carries on regardless, and this is
    the record of whether it then reached the device.

    Args:
        event_mrid: The event whose control it was.
        lfdi: The device, or its href where no LFDI is known.
        applied: Whether the apply finished without error.
        at: When it finished, on the server's timebase.
        error: The failure, when it failed.
        error_type: The failure's exception class name.
    """
    if not self.enabled:
        return
    try:
        record: dict[str, Any] = {
            "kind": "late_dispatch",
            "event_mrid": mrid_hex(event_mrid),
            "lfdi": _lfdi(lfdi),
            "applied": applied,
            "at": at,
        }
        if error is not None:
            record["error"] = error[:512]
            record["error_type"] = error_type
        self._publish(record)
    except Exception:
        self._failed("late_dispatch")

mrid_hex

mrid_hex(mrid: bytes | None) -> str | None

An mRID as uppercase hex, the form the mRID element takes on the wire.

Matching the wire form lets a record be joined to the raw control by a plain text match on the identifier.

Source code in src/py20305/forwarders/audit.py

def mrid_hex(mrid: bytes | None) -> str | None:
    """An mRID as uppercase hex, the form the ``mRID`` element takes on the wire.

    Matching the wire form lets a record be joined to the raw control by a
    plain text match on the identifier.
    """
    return mrid.hex().upper() if mrid is not None else None

py20305.forwarders.ocsf

A minimal OCSF Network Activity (class 4001) event, for connection telemetry.

The Open Cybersecurity Schema Framework is a public, vendor-neutral schema (https://schema.ocsf.io/). This module carries just enough of it to emit one class -- Network Activity, class_uid 4001 -- which is how this client reports its own connection outcomes to a security-monitoring platform: src_endpoint / dst_endpoint / status_id / status_detail map onto what a connection log needs, and a consumer can ingest an established schema without a translation layer.

Vendored rather than depended on: the emitting side needs four factories and a serializer, not a schema toolkit, and a library client should not carry a platform dependency to publish a public format.

Two rules here come from the compliance argument behind connection logging rather than from the schema, and are enforced rather than documented:

• A failure event must carry status_detail. The reason is the entire value of an error record; a failure without one satisfies nobody.
• Failures are never aggregated. Successful connections may be coalesced into a window carrying a count -- a polling client would otherwise emit more connection events than the passive capture beside it -- but collapsing distinct failure reasons into a count destroys the record's only value.

SeverityId

Bases: IntEnum

OCSF severity_id -- how consequential the event is.

Source code in src/py20305/forwarders/ocsf.py

class SeverityId(IntEnum):
    """OCSF ``severity_id`` -- how consequential the event is."""

    UNKNOWN = 0
    INFORMATIONAL = 1
    LOW = 2
    MEDIUM = 3
    HIGH = 4
    CRITICAL = 5
    FATAL = 6
    OTHER = 99

StatusId

Bases: IntEnum

OCSF status_id -- the outcome of the activity the event describes.

Source code in src/py20305/forwarders/ocsf.py

class StatusId(IntEnum):
    """OCSF ``status_id`` -- the outcome of the activity the event describes."""

    UNKNOWN = 0
    SUCCESS = 1
    FAILURE = 2
    OTHER = 99

NetworkActivityId

Bases: IntEnum

OCSF 4001 activity_id.

Source code in src/py20305/forwarders/ocsf.py

class NetworkActivityId(IntEnum):
    """OCSF 4001 ``activity_id``."""

    UNKNOWN = 0
    OPEN = 1
    CLOSE = 2
    RESET = 3
    FAIL = 4
    REFUSE = 5
    TRAFFIC = 6
    LISTEN = 7
    OTHER = 99

ConnectionDirectionId

Bases: IntEnum

OCSF connection_info.direction_id -- which way the connection runs.

Source code in src/py20305/forwarders/ocsf.py

class ConnectionDirectionId(IntEnum):
    """OCSF ``connection_info.direction_id`` -- which way the connection runs."""

    UNKNOWN = 0
    INBOUND = 1
    OUTBOUND = 2
    LATERAL = 3
    LOCAL = 4
    OTHER = 99

Endpoint dataclass

An OCSF network_endpoint -- one end of a connection.

ip and hostname are distinct OCSF attributes with typed meanings: ip must hold an IP address, and a DNS name goes in hostname. A configured server URL usually names a host, while an established socket reports addresses, so either alone identifies an endpoint -- but at least one must be present, or the endpoint identifies nothing.

port may be None: an application-layer client frequently cannot determine the ephemeral local port its socket was assigned, and an unknown port is reported as unknown rather than invented. (This is why the OCSF path does not reuse :class:~py20305.forwarders.types.NetworkEndpoint, whose wire contract requires a port.)

Source code in src/py20305/forwarders/ocsf.py

@dataclass(kw_only=True)
class Endpoint:
    """An OCSF ``network_endpoint`` -- one end of a connection.

    ``ip`` and ``hostname`` are distinct OCSF attributes with typed meanings:
    ``ip`` must hold an IP address, and a DNS name goes in ``hostname``. A
    configured server URL usually names a host, while an established socket
    reports addresses, so either alone identifies an endpoint -- but at least
    one must be present, or the endpoint identifies nothing.

    ``port`` may be ``None``: an application-layer client frequently cannot
    determine the ephemeral local port its socket was assigned, and an unknown
    port is reported as unknown rather than invented. (This is why the OCSF
    path does not reuse :class:`~py20305.forwarders.types.NetworkEndpoint`,
    whose wire contract requires a port.)
    """

    ip: str | None = None
    hostname: str | None = None
    port: int | None = None

    def __post_init__(self) -> None:
        """Reject an endpoint that names nothing."""
        if self.ip is None and self.hostname is None:
            raise ValueError("an endpoint needs an ip or a hostname")

    def to_dict(self) -> dict[str, Any]:
        """Serialize; unset attributes are omitted."""
        result: dict[str, Any] = {}
        if self.ip is not None:
            result["ip"] = self.ip
        if self.hostname is not None:
            result["hostname"] = self.hostname
        if self.port is not None:
            result["port"] = self.port
        return result

to_dict

to_dict() -> dict[str, Any]

Serialize; unset attributes are omitted.

Source code in src/py20305/forwarders/ocsf.py

def to_dict(self) -> dict[str, Any]:
    """Serialize; unset attributes are omitted."""
    result: dict[str, Any] = {}
    if self.ip is not None:
        result["ip"] = self.ip
    if self.hostname is not None:
        result["hostname"] = self.hostname
    if self.port is not None:
        result["port"] = self.port
    return result

Product dataclass

OCSF product object -- the software that produced the event.

Source code in src/py20305/forwarders/ocsf.py

@dataclass(kw_only=True)
class Product:
    """OCSF ``product`` object -- the software that produced the event."""

    name: str
    vendor_name: str = DEFAULT_VENDOR_NAME
    version: str | None = None

    def __post_init__(self) -> None:
        """Reject an empty product name -- it is what identifies the emitter."""
        if not self.name or not self.name.strip():
            raise ValueError("product name must be a non-empty string")
        self.name = self.name.strip()

    def to_dict(self) -> dict[str, Any]:
        """Serialize to an OCSF ``product`` object; unset fields are omitted."""
        result: dict[str, Any] = {"name": self.name, "vendor_name": self.vendor_name}
        if self.version is not None:
            result["version"] = self.version
        return result

to_dict

to_dict() -> dict[str, Any]

Serialize to an OCSF product object; unset fields are omitted.

Source code in src/py20305/forwarders/ocsf.py

def to_dict(self) -> dict[str, Any]:
    """Serialize to an OCSF ``product`` object; unset fields are omitted."""
    result: dict[str, Any] = {"name": self.name, "vendor_name": self.vendor_name}
    if self.version is not None:
        result["version"] = self.version
    return result

Metadata dataclass

OCSF metadata object -- required on every OCSF event.

Attributes:

Name	Type	Description
product	Product	The software that produced the event.
version	str	The OCSF schema version the event conforms to.
labels	list[str]	Free-form labels. Connection telemetry uses one role:<role> label to say which link an event describes.

Source code in src/py20305/forwarders/ocsf.py

@dataclass(kw_only=True)
class Metadata:
    """OCSF ``metadata`` object -- required on every OCSF event.

    Attributes:
        product: The software that produced the event.
        version: The OCSF schema version the event conforms to.
        labels: Free-form labels. Connection telemetry uses one
            ``role:<role>`` label to say which link an event describes.
    """

    product: Product
    version: str = OCSF_SCHEMA_VERSION
    labels: list[str] = field(default_factory=list)

    def to_dict(self) -> dict[str, Any]:
        """Serialize to an OCSF ``metadata`` object; empty labels are omitted."""
        result: dict[str, Any] = {"product": self.product.to_dict(), "version": self.version}
        if self.labels:
            result["labels"] = list(self.labels)
        return result

to_dict

to_dict() -> dict[str, Any]

Serialize to an OCSF metadata object; empty labels are omitted.

Source code in src/py20305/forwarders/ocsf.py

def to_dict(self) -> dict[str, Any]:
    """Serialize to an OCSF ``metadata`` object; empty labels are omitted."""
    result: dict[str, Any] = {"product": self.product.to_dict(), "version": self.version}
    if self.labels:
        result["labels"] = list(self.labels)
    return result

NetworkActivity dataclass

An OCSF Network Activity (4001) event describing one connection outcome.

Prefer the factories -- :meth:for_connection_success, :meth:for_connection_failure, :meth:for_exchange_failure, and :meth:for_coalesced_successes -- over the constructor. They set the activity_id / status_id pair together, which is the part of OCSF conformance no validator can check: emitting Open / Success for a connection that actually failed is structurally valid and semantically wrong.

The two failure factories split on which layer failed, because that is what a reader needs to know first. :meth:for_connection_failure covers transport-level failures -- the connection never opened (Fail, Refuse) or was torn down after it did (Reset). :meth:for_exchange_failure covers a connection that stayed up and whose exchange over it failed, so the activity remains Open and only the status is Failure.

Attributes:

Name	Type	Description
activity_id	NetworkActivityId	What happened to the connection.
metadata	Metadata	The producing software and the OCSF version emitted.
time	int	Event time in epoch milliseconds. For an aggregate this is start_time, per OCSF, so aggregates and discrete events order consistently on one timeline.
severity_id	SeverityId	OCSF severity.
status_id	StatusId	Outcome of the activity.
status_detail	str | None	Why it happened. Required on every failure.
status_code	str | None	Optional producer-specific code, e.g. an HTTP status.
message	str | None	Optional human-readable description.
count	int | None	Number of connections aggregated into this record. Unset for discrete events; > 1 requires the aggregation window.
start_time	int | None	First event in the aggregation window, epoch ms.
end_time	int | None	Last event in the aggregation window, epoch ms.
duration	int | None	Window length in ms. Derived from the bounds when not supplied; an explicit value that disagrees with them is rejected.
src_endpoint	Endpoint | None	The initiator of the connection.
dst_endpoint	Endpoint | None	The responder. At least one endpoint must be present, per the OCSF constraint on the network category.
service	str | None	Service label, e.g. "ieee2030.5". Serialized as dst_endpoint.svc_name, where OCSF carries the responder's service identity.
connection_direction	ConnectionDirectionId	Direction of the connection. Defaults to outbound, the case a client-side emitter reports.
protocol_name	str | None	Transport protocol name, e.g. "tcp".
url	str | None	Optional URL the connection targeted; on a failure it is what identifies which interface was unreachable.

Source code in src/py20305/forwarders/ocsf.py

@dataclass(kw_only=True)
class NetworkActivity:
    """An OCSF Network Activity (4001) event describing one connection outcome.

    Prefer the factories -- :meth:`for_connection_success`,
    :meth:`for_connection_failure`, :meth:`for_exchange_failure`, and
    :meth:`for_coalesced_successes` -- over the constructor. They set the
    ``activity_id`` / ``status_id`` pair together, which is the part of OCSF
    conformance no validator can check: emitting ``Open`` / ``Success`` for a
    connection that actually failed is structurally valid and semantically
    wrong.

    The two failure factories split on which layer failed, because that is
    what a reader needs to know first. :meth:`for_connection_failure` covers
    transport-level failures -- the connection never opened (``Fail``,
    ``Refuse``) or was torn down after it did (``Reset``).
    :meth:`for_exchange_failure` covers a connection that stayed up and whose
    exchange over it failed, so the activity remains ``Open`` and only the
    status is ``Failure``.

    Attributes:
        activity_id: What happened to the connection.
        metadata: The producing software and the OCSF version emitted.
        time: Event time in epoch milliseconds. For an aggregate this is
            ``start_time``, per OCSF, so aggregates and discrete events order
            consistently on one timeline.
        severity_id: OCSF severity.
        status_id: Outcome of the activity.
        status_detail: Why it happened. Required on every failure.
        status_code: Optional producer-specific code, e.g. an HTTP status.
        message: Optional human-readable description.
        count: Number of connections aggregated into this record. Unset for
            discrete events; ``> 1`` requires the aggregation window.
        start_time: First event in the aggregation window, epoch ms.
        end_time: Last event in the aggregation window, epoch ms.
        duration: Window length in ms. Derived from the bounds when not
            supplied; an explicit value that disagrees with them is rejected.
        src_endpoint: The initiator of the connection.
        dst_endpoint: The responder. At least one endpoint must be present,
            per the OCSF constraint on the network category.
        service: Service label, e.g. ``"ieee2030.5"``. Serialized as
            ``dst_endpoint.svc_name``, where OCSF carries the responder's
            service identity.
        connection_direction: Direction of the connection. Defaults to
            outbound, the case a client-side emitter reports.
        protocol_name: Transport protocol name, e.g. ``"tcp"``.
        url: Optional URL the connection targeted; on a failure it is what
            identifies which interface was unreachable.
    """

    CLASS_UID: ClassVar[int] = 4001
    CATEGORY_UID: ClassVar[int] = 4
    CLASS_NAME: ClassVar[str] = "Network Activity"
    CATEGORY_NAME: ClassVar[str] = "Network Activity"

    activity_id: NetworkActivityId
    metadata: Metadata
    time: int = field(default_factory=now_epoch_ms)
    severity_id: SeverityId = SeverityId.INFORMATIONAL
    status_id: StatusId = StatusId.UNKNOWN
    status_detail: str | None = None
    status_code: str | None = None
    message: str | None = None
    count: int | None = None
    start_time: int | None = None
    end_time: int | None = None
    duration: int | None = None
    src_endpoint: Endpoint | None = None
    dst_endpoint: Endpoint | None = None
    service: str | None = None
    connection_direction: ConnectionDirectionId = ConnectionDirectionId.OUTBOUND
    protocol_name: str | None = None
    url: str | None = None

    def __post_init__(self) -> None:
        """Validate the OCSF constraints and the two compliance rules."""
        self.activity_id = NetworkActivityId(self.activity_id)
        self.severity_id = SeverityId(self.severity_id)
        self.status_id = StatusId(self.status_id)
        self.connection_direction = ConnectionDirectionId(self.connection_direction)
        self.time = to_epoch_ms(self.time)
        if self.start_time is not None:
            self.start_time = to_epoch_ms(self.start_time)
        if self.end_time is not None:
            self.end_time = to_epoch_ms(self.end_time)

        if self.count is not None and self.count < 1:
            raise ValueError(f"count must be at least 1 when set, got {self.count}")

        is_aggregate = self.count is not None and self.count > 1

        # OCSF scopes the window attributes to aggregates in both directions:
        # an aggregate must say which interval it covers, and a discrete event
        # uses `time` alone.
        if is_aggregate and (self.start_time is None or self.end_time is None):
            raise ValueError(
                "an aggregated event (count > 1) must carry both start_time and end_time"
            )
        if not is_aggregate and (self.start_time is not None or self.end_time is not None):
            raise ValueError(
                "start_time and end_time describe an aggregation window; "
                "a discrete event uses time alone"
            )

        if self.start_time is not None and self.end_time is not None:
            if self.end_time < self.start_time:
                raise ValueError(
                    f"end_time ({self.end_time}) must not precede start_time ({self.start_time})"
                )
            # OCSF defines duration as the window's length, so an explicit
            # value that disagrees with the bounds is a producer bug rather
            # than a more precise measurement to preserve.
            window = self.end_time - self.start_time
            if self.duration is None:
                self.duration = window
            elif self.duration != window:
                raise ValueError(
                    f"duration ({self.duration}) must equal end_time - start_time ({window})"
                )
        elif self.duration is not None:
            raise ValueError(
                "duration is the aggregation window's length; set start_time and end_time with it"
            )

        # OCSF constrains the network category to at least one endpoint. An
        # event with neither says nothing about a connection.
        if self.src_endpoint is None and self.dst_endpoint is None:
            raise ValueError("at least one of src_endpoint or dst_endpoint must be set")

        # The service label serializes onto the responder, so without one it
        # would be dropped on the way to the wire. Silently losing a documented
        # attribute is worse than refusing the combination.
        if self.service is not None and self.dst_endpoint is None:
            raise ValueError(
                "service is carried on dst_endpoint.svc_name, so it requires a dst_endpoint"
            )

        # A failure record whose reason is missing is the one record the
        # error-logging requirement exists to produce.
        if self.status_id is StatusId.FAILURE and not (
            self.status_detail and self.status_detail.strip()
        ):
            raise ValueError(
                "a failed connection event must carry status_detail explaining the failure"
            )

        # Successful connections may be coalesced into a window; failures keep
        # one event each, because a count of failures with the reasons
        # collapsed cannot answer why any of them failed.
        if is_aggregate and self.status_id is StatusId.FAILURE:
            raise ValueError(
                "failed connections must not be aggregated -- emit one event per failure "
                "so each reason survives"
            )

    @property
    def type_uid(self) -> int:
        """OCSF ``type_uid`` -- ``class_uid * 100 + activity_id``."""
        return self.CLASS_UID * 100 + int(self.activity_id)

    def to_dict(self) -> dict[str, Any]:
        """Serialize to an OCSF 4001 event dictionary."""
        result: dict[str, Any] = {
            "activity_id": int(self.activity_id),
            "activity_name": ACTIVITY_CAPTIONS[self.activity_id],
            "category_uid": self.CATEGORY_UID,
            "category_name": self.CATEGORY_NAME,
            "class_uid": self.CLASS_UID,
            "class_name": self.CLASS_NAME,
            "type_uid": self.type_uid,
            "time": self.time,
            "severity_id": int(self.severity_id),
            "severity": SEVERITY_CAPTIONS[self.severity_id],
            "status_id": int(self.status_id),
            "status": STATUS_CAPTIONS[self.status_id],
            "metadata": self.metadata.to_dict(),
        }

        if self.status_detail is not None:
            result["status_detail"] = self.status_detail
        if self.status_code is not None:
            result["status_code"] = self.status_code
        if self.message is not None:
            result["message"] = self.message
        if self.count is not None:
            result["count"] = self.count
        if self.start_time is not None:
            result["start_time"] = self.start_time
        if self.end_time is not None:
            result["end_time"] = self.end_time
        if self.duration is not None:
            result["duration"] = self.duration

        if self.src_endpoint is not None:
            result["src_endpoint"] = self.src_endpoint.to_dict()
        if self.dst_endpoint is not None:
            dst = self.dst_endpoint.to_dict()
            # The service label rides on the responder, per OCSF.
            if self.service is not None:
                dst["svc_name"] = self.service
            result["dst_endpoint"] = dst

        connection_info: dict[str, Any] = {
            "direction_id": int(self.connection_direction),
            "direction": DIRECTION_CAPTIONS[self.connection_direction],
        }
        if self.protocol_name is not None:
            connection_info["protocol_name"] = self.protocol_name
        result["connection_info"] = connection_info

        if self.url is not None:
            result["url"] = {"url_string": self.url}

        return result

    # Factories -- each sets the activity/status pair together.

    @classmethod
    def for_connection_success(
        cls,
        *,
        metadata: Metadata,
        dst_endpoint: Endpoint,
        src_endpoint: Endpoint | None = None,
        service: str | None = None,
        url: str | None = None,
        time: EventTime | None = None,
        protocol_name: str | None = "tcp",
        connection_direction: ConnectionDirectionId = ConnectionDirectionId.OUTBOUND,
        severity_id: SeverityId = SeverityId.INFORMATIONAL,
        message: str | None = None,
    ) -> NetworkActivity:
        """Report one connection that was established: ``Open`` / ``Success``."""
        return cls(
            activity_id=NetworkActivityId.OPEN,
            status_id=StatusId.SUCCESS,
            severity_id=severity_id,
            metadata=metadata,
            src_endpoint=src_endpoint,
            dst_endpoint=dst_endpoint,
            service=service,
            url=url,
            protocol_name=protocol_name,
            connection_direction=connection_direction,
            message=message,
            time=to_epoch_ms(time) if time is not None else now_epoch_ms(),
        )

    @classmethod
    def for_connection_failure(
        cls,
        *,
        metadata: Metadata,
        dst_endpoint: Endpoint,
        status_detail: str,
        activity_id: NetworkActivityId = NetworkActivityId.FAIL,
        src_endpoint: Endpoint | None = None,
        service: str | None = None,
        url: str | None = None,
        time: EventTime | None = None,
        status_code: str | None = None,
        severity_id: SeverityId = SeverityId.MEDIUM,
        protocol_name: str | None = "tcp",
        connection_direction: ConnectionDirectionId = ConnectionDirectionId.OUTBOUND,
        message: str | None = None,
    ) -> NetworkActivity:
        """Report a transport-level connection failure, and why.

        ``status_detail`` is required: it is the attribute the error record
        exists to carry. A status the peer *returned* is an exchange failure,
        not this -- see :meth:`for_exchange_failure`.

        Raises:
            ValueError: If ``activity_id`` is not a failure activity, or
                ``status_detail`` is empty.
        """
        # Normalize first: the constructor accepts a raw int, so this factory
        # has to as well, and the message below reads `.name`.
        activity_id = NetworkActivityId(activity_id)
        if activity_id not in _FAILURE_ACTIVITIES:
            hint = (
                " -- a connection that opened and whose exchange then "
                "failed is for_exchange_failure()"
                if activity_id is NetworkActivityId.OPEN
                else ""
            )
            raise ValueError(
                f"activity_id for a failure must be one of "
                f"{', '.join(sorted(a.name for a in _FAILURE_ACTIVITIES))}, "
                f"got {activity_id.name}{hint}"
            )

        return cls(
            activity_id=activity_id,
            status_id=StatusId.FAILURE,
            status_detail=status_detail,
            status_code=status_code,
            severity_id=severity_id,
            metadata=metadata,
            src_endpoint=src_endpoint,
            dst_endpoint=dst_endpoint,
            service=service,
            url=url,
            protocol_name=protocol_name,
            connection_direction=connection_direction,
            message=message,
            time=to_epoch_ms(time) if time is not None else now_epoch_ms(),
        )

    @classmethod
    def for_exchange_failure(
        cls,
        *,
        metadata: Metadata,
        dst_endpoint: Endpoint,
        status_detail: str,
        src_endpoint: Endpoint | None = None,
        service: str | None = None,
        url: str | None = None,
        time: EventTime | None = None,
        status_code: str | None = None,
        severity_id: SeverityId = SeverityId.MEDIUM,
        protocol_name: str | None = "tcp",
        connection_direction: ConnectionDirectionId = ConnectionDirectionId.OUTBOUND,
        message: str | None = None,
    ) -> NetworkActivity:
        """Report a connection that opened and whose exchange over it failed.

        The counterpart to :meth:`for_connection_failure`. Here the connection
        worked and stayed up -- a server answered 500, rate-limited the client,
        or sent it round a redirect loop -- so the activity is ``Open`` and
        only the status is ``Failure``. Reporting a 500 as ``Fail`` would tell
        whoever reads the log that the client never reached the peer, sending
        an investigation after a network problem that did not happen;
        reporting it as ``Open`` / ``Success`` because the socket worked would
        lose the failure entirely.

        Raises:
            ValueError: If ``status_detail`` is empty.
        """
        return cls(
            activity_id=NetworkActivityId.OPEN,
            status_id=StatusId.FAILURE,
            status_detail=status_detail,
            status_code=status_code,
            severity_id=severity_id,
            metadata=metadata,
            src_endpoint=src_endpoint,
            dst_endpoint=dst_endpoint,
            service=service,
            url=url,
            protocol_name=protocol_name,
            connection_direction=connection_direction,
            message=message,
            time=to_epoch_ms(time) if time is not None else now_epoch_ms(),
        )

    @classmethod
    def for_coalesced_successes(
        cls,
        *,
        metadata: Metadata,
        dst_endpoint: Endpoint,
        count: int,
        start_time: EventTime,
        end_time: EventTime,
        src_endpoint: Endpoint | None = None,
        service: str | None = None,
        url: str | None = None,
        protocol_name: str | None = "tcp",
        connection_direction: ConnectionDirectionId = ConnectionDirectionId.OUTBOUND,
        severity_id: SeverityId = SeverityId.INFORMATIONAL,
        time: EventTime | None = None,
        message: str | None = None,
    ) -> NetworkActivity:
        """Report a window of successful connections that shared an endpoint.

        A polling client opens connections continuously, and one event per
        connection would put more traffic on the collector than the passive
        capture beside it. A coalesced event stands in for the whole window,
        so it preserves what a per-attempt record would have answered: the
        window's bounds, how many attempts it represents, and the endpoint
        and service they shared. It is not a representative sample.

        ``time`` defaults to ``start_time``: OCSF sets an aggregate's ``time``
        to the earliest event it covers, so aggregates and discrete events
        order consistently on one timeline.

        Raises:
            ValueError: If ``count`` is below 2 or the window is inverted.
        """
        if count < 2:
            raise ValueError(
                f"a coalesced event represents more than one connection, got count={count}; "
                "report a single success with for_connection_success()"
            )

        return cls(
            activity_id=NetworkActivityId.OPEN,
            status_id=StatusId.SUCCESS,
            severity_id=severity_id,
            metadata=metadata,
            src_endpoint=src_endpoint,
            dst_endpoint=dst_endpoint,
            service=service,
            url=url,
            protocol_name=protocol_name,
            connection_direction=connection_direction,
            message=message,
            count=count,
            start_time=to_epoch_ms(start_time),
            end_time=to_epoch_ms(end_time),
            time=to_epoch_ms(time) if time is not None else to_epoch_ms(start_time),
        )

type_uid property

type_uid: int

OCSF type_uid -- class_uid * 100 + activity_id.

to_dict

to_dict() -> dict[str, Any]

Serialize to an OCSF 4001 event dictionary.

Source code in src/py20305/forwarders/ocsf.py

def to_dict(self) -> dict[str, Any]:
    """Serialize to an OCSF 4001 event dictionary."""
    result: dict[str, Any] = {
        "activity_id": int(self.activity_id),
        "activity_name": ACTIVITY_CAPTIONS[self.activity_id],
        "category_uid": self.CATEGORY_UID,
        "category_name": self.CATEGORY_NAME,
        "class_uid": self.CLASS_UID,
        "class_name": self.CLASS_NAME,
        "type_uid": self.type_uid,
        "time": self.time,
        "severity_id": int(self.severity_id),
        "severity": SEVERITY_CAPTIONS[self.severity_id],
        "status_id": int(self.status_id),
        "status": STATUS_CAPTIONS[self.status_id],
        "metadata": self.metadata.to_dict(),
    }

    if self.status_detail is not None:
        result["status_detail"] = self.status_detail
    if self.status_code is not None:
        result["status_code"] = self.status_code
    if self.message is not None:
        result["message"] = self.message
    if self.count is not None:
        result["count"] = self.count
    if self.start_time is not None:
        result["start_time"] = self.start_time
    if self.end_time is not None:
        result["end_time"] = self.end_time
    if self.duration is not None:
        result["duration"] = self.duration

    if self.src_endpoint is not None:
        result["src_endpoint"] = self.src_endpoint.to_dict()
    if self.dst_endpoint is not None:
        dst = self.dst_endpoint.to_dict()
        # The service label rides on the responder, per OCSF.
        if self.service is not None:
            dst["svc_name"] = self.service
        result["dst_endpoint"] = dst

    connection_info: dict[str, Any] = {
        "direction_id": int(self.connection_direction),
        "direction": DIRECTION_CAPTIONS[self.connection_direction],
    }
    if self.protocol_name is not None:
        connection_info["protocol_name"] = self.protocol_name
    result["connection_info"] = connection_info

    if self.url is not None:
        result["url"] = {"url_string": self.url}

    return result

for_connection_success classmethod

for_connection_success(*, metadata: Metadata, dst_endpoint: Endpoint, src_endpoint: Endpoint | None = None, service: str | None = None, url: str | None = None, time: EventTime | None = None, protocol_name: str | None = 'tcp', connection_direction: ConnectionDirectionId = ConnectionDirectionId.OUTBOUND, severity_id: SeverityId = SeverityId.INFORMATIONAL, message: str | None = None) -> NetworkActivity

Report one connection that was established: Open / Success.

Source code in src/py20305/forwarders/ocsf.py

@classmethod
def for_connection_success(
    cls,
    *,
    metadata: Metadata,
    dst_endpoint: Endpoint,
    src_endpoint: Endpoint | None = None,
    service: str | None = None,
    url: str | None = None,
    time: EventTime | None = None,
    protocol_name: str | None = "tcp",
    connection_direction: ConnectionDirectionId = ConnectionDirectionId.OUTBOUND,
    severity_id: SeverityId = SeverityId.INFORMATIONAL,
    message: str | None = None,
) -> NetworkActivity:
    """Report one connection that was established: ``Open`` / ``Success``."""
    return cls(
        activity_id=NetworkActivityId.OPEN,
        status_id=StatusId.SUCCESS,
        severity_id=severity_id,
        metadata=metadata,
        src_endpoint=src_endpoint,
        dst_endpoint=dst_endpoint,
        service=service,
        url=url,
        protocol_name=protocol_name,
        connection_direction=connection_direction,
        message=message,
        time=to_epoch_ms(time) if time is not None else now_epoch_ms(),
    )

for_connection_failure classmethod

for_connection_failure(*, metadata: Metadata, dst_endpoint: Endpoint, status_detail: str, activity_id: NetworkActivityId = NetworkActivityId.FAIL, src_endpoint: Endpoint | None = None, service: str | None = None, url: str | None = None, time: EventTime | None = None, status_code: str | None = None, severity_id: SeverityId = SeverityId.MEDIUM, protocol_name: str | None = 'tcp', connection_direction: ConnectionDirectionId = ConnectionDirectionId.OUTBOUND, message: str | None = None) -> NetworkActivity

Report a transport-level connection failure, and why.

status_detail is required: it is the attribute the error record exists to carry. A status the peer returned is an exchange failure, not this -- see :meth:for_exchange_failure.

Raises:

Type	Description
ValueError	If activity_id is not a failure activity, or status_detail is empty.

Source code in src/py20305/forwarders/ocsf.py

@classmethod
def for_connection_failure(
    cls,
    *,
    metadata: Metadata,
    dst_endpoint: Endpoint,
    status_detail: str,
    activity_id: NetworkActivityId = NetworkActivityId.FAIL,
    src_endpoint: Endpoint | None = None,
    service: str | None = None,
    url: str | None = None,
    time: EventTime | None = None,
    status_code: str | None = None,
    severity_id: SeverityId = SeverityId.MEDIUM,
    protocol_name: str | None = "tcp",
    connection_direction: ConnectionDirectionId = ConnectionDirectionId.OUTBOUND,
    message: str | None = None,
) -> NetworkActivity:
    """Report a transport-level connection failure, and why.

    ``status_detail`` is required: it is the attribute the error record
    exists to carry. A status the peer *returned* is an exchange failure,
    not this -- see :meth:`for_exchange_failure`.

    Raises:
        ValueError: If ``activity_id`` is not a failure activity, or
            ``status_detail`` is empty.
    """
    # Normalize first: the constructor accepts a raw int, so this factory
    # has to as well, and the message below reads `.name`.
    activity_id = NetworkActivityId(activity_id)
    if activity_id not in _FAILURE_ACTIVITIES:
        hint = (
            " -- a connection that opened and whose exchange then "
            "failed is for_exchange_failure()"
            if activity_id is NetworkActivityId.OPEN
            else ""
        )
        raise ValueError(
            f"activity_id for a failure must be one of "
            f"{', '.join(sorted(a.name for a in _FAILURE_ACTIVITIES))}, "
            f"got {activity_id.name}{hint}"
        )

    return cls(
        activity_id=activity_id,
        status_id=StatusId.FAILURE,
        status_detail=status_detail,
        status_code=status_code,
        severity_id=severity_id,
        metadata=metadata,
        src_endpoint=src_endpoint,
        dst_endpoint=dst_endpoint,
        service=service,
        url=url,
        protocol_name=protocol_name,
        connection_direction=connection_direction,
        message=message,
        time=to_epoch_ms(time) if time is not None else now_epoch_ms(),
    )

for_exchange_failure classmethod

for_exchange_failure(*, metadata: Metadata, dst_endpoint: Endpoint, status_detail: str, src_endpoint: Endpoint | None = None, service: str | None = None, url: str | None = None, time: EventTime | None = None, status_code: str | None = None, severity_id: SeverityId = SeverityId.MEDIUM, protocol_name: str | None = 'tcp', connection_direction: ConnectionDirectionId = ConnectionDirectionId.OUTBOUND, message: str | None = None) -> NetworkActivity

Report a connection that opened and whose exchange over it failed.

The counterpart to :meth:for_connection_failure. Here the connection worked and stayed up -- a server answered 500, rate-limited the client, or sent it round a redirect loop -- so the activity is Open and only the status is Failure. Reporting a 500 as Fail would tell whoever reads the log that the client never reached the peer, sending an investigation after a network problem that did not happen; reporting it as Open / Success because the socket worked would lose the failure entirely.

Raises:

Type	Description
ValueError	If status_detail is empty.

Source code in src/py20305/forwarders/ocsf.py

@classmethod
def for_exchange_failure(
    cls,
    *,
    metadata: Metadata,
    dst_endpoint: Endpoint,
    status_detail: str,
    src_endpoint: Endpoint | None = None,
    service: str | None = None,
    url: str | None = None,
    time: EventTime | None = None,
    status_code: str | None = None,
    severity_id: SeverityId = SeverityId.MEDIUM,
    protocol_name: str | None = "tcp",
    connection_direction: ConnectionDirectionId = ConnectionDirectionId.OUTBOUND,
    message: str | None = None,
) -> NetworkActivity:
    """Report a connection that opened and whose exchange over it failed.

    The counterpart to :meth:`for_connection_failure`. Here the connection
    worked and stayed up -- a server answered 500, rate-limited the client,
    or sent it round a redirect loop -- so the activity is ``Open`` and
    only the status is ``Failure``. Reporting a 500 as ``Fail`` would tell
    whoever reads the log that the client never reached the peer, sending
    an investigation after a network problem that did not happen;
    reporting it as ``Open`` / ``Success`` because the socket worked would
    lose the failure entirely.

    Raises:
        ValueError: If ``status_detail`` is empty.
    """
    return cls(
        activity_id=NetworkActivityId.OPEN,
        status_id=StatusId.FAILURE,
        status_detail=status_detail,
        status_code=status_code,
        severity_id=severity_id,
        metadata=metadata,
        src_endpoint=src_endpoint,
        dst_endpoint=dst_endpoint,
        service=service,
        url=url,
        protocol_name=protocol_name,
        connection_direction=connection_direction,
        message=message,
        time=to_epoch_ms(time) if time is not None else now_epoch_ms(),
    )

for_coalesced_successes classmethod

for_coalesced_successes(*, metadata: Metadata, dst_endpoint: Endpoint, count: int, start_time: EventTime, end_time: EventTime, src_endpoint: Endpoint | None = None, service: str | None = None, url: str | None = None, protocol_name: str | None = 'tcp', connection_direction: ConnectionDirectionId = ConnectionDirectionId.OUTBOUND, severity_id: SeverityId = SeverityId.INFORMATIONAL, time: EventTime | None = None, message: str | None = None) -> NetworkActivity

Report a window of successful connections that shared an endpoint.

A polling client opens connections continuously, and one event per connection would put more traffic on the collector than the passive capture beside it. A coalesced event stands in for the whole window, so it preserves what a per-attempt record would have answered: the window's bounds, how many attempts it represents, and the endpoint and service they shared. It is not a representative sample.

time defaults to start_time: OCSF sets an aggregate's time to the earliest event it covers, so aggregates and discrete events order consistently on one timeline.

Raises:

Type	Description
ValueError	If count is below 2 or the window is inverted.

Source code in src/py20305/forwarders/ocsf.py

@classmethod
def for_coalesced_successes(
    cls,
    *,
    metadata: Metadata,
    dst_endpoint: Endpoint,
    count: int,
    start_time: EventTime,
    end_time: EventTime,
    src_endpoint: Endpoint | None = None,
    service: str | None = None,
    url: str | None = None,
    protocol_name: str | None = "tcp",
    connection_direction: ConnectionDirectionId = ConnectionDirectionId.OUTBOUND,
    severity_id: SeverityId = SeverityId.INFORMATIONAL,
    time: EventTime | None = None,
    message: str | None = None,
) -> NetworkActivity:
    """Report a window of successful connections that shared an endpoint.

    A polling client opens connections continuously, and one event per
    connection would put more traffic on the collector than the passive
    capture beside it. A coalesced event stands in for the whole window,
    so it preserves what a per-attempt record would have answered: the
    window's bounds, how many attempts it represents, and the endpoint
    and service they shared. It is not a representative sample.

    ``time`` defaults to ``start_time``: OCSF sets an aggregate's ``time``
    to the earliest event it covers, so aggregates and discrete events
    order consistently on one timeline.

    Raises:
        ValueError: If ``count`` is below 2 or the window is inverted.
    """
    if count < 2:
        raise ValueError(
            f"a coalesced event represents more than one connection, got count={count}; "
            "report a single success with for_connection_success()"
        )

    return cls(
        activity_id=NetworkActivityId.OPEN,
        status_id=StatusId.SUCCESS,
        severity_id=severity_id,
        metadata=metadata,
        src_endpoint=src_endpoint,
        dst_endpoint=dst_endpoint,
        service=service,
        url=url,
        protocol_name=protocol_name,
        connection_direction=connection_direction,
        message=message,
        count=count,
        start_time=to_epoch_ms(start_time),
        end_time=to_epoch_ms(end_time),
        time=to_epoch_ms(time) if time is not None else to_epoch_ms(start_time),
    )

to_epoch_ms

to_epoch_ms(value: EventTime) -> int

Normalize a time value to an OCSF timestamp_t.

OCSF timestamps are milliseconds since the Unix epoch. This accepts a datetime (naive values are read as UTC), an ISO 8601 string, or a number already in epoch milliseconds.

Raises:

Type	Description
TypeError	If value is not a datetime, number, or string.
ValueError	If a string is not parseable as ISO 8601.

Source code in src/py20305/forwarders/ocsf.py

def to_epoch_ms(value: EventTime) -> int:
    """Normalize a time value to an OCSF ``timestamp_t``.

    OCSF timestamps are milliseconds since the Unix epoch. This accepts a
    ``datetime`` (naive values are read as UTC), an ISO 8601 string, or a
    number already in epoch milliseconds.

    Raises:
        TypeError: If ``value`` is not a datetime, number, or string.
        ValueError: If a string is not parseable as ISO 8601.
    """
    if isinstance(value, bool):
        raise TypeError("timestamp must be a datetime, number, or ISO 8601 string, got bool")
    if isinstance(value, datetime):
        moment = value if value.tzinfo is not None else value.replace(tzinfo=UTC)
        return int(moment.timestamp() * 1000)
    if isinstance(value, (int, float)):
        return int(value)
    if isinstance(value, str):
        text = value.strip()
        if text.endswith("Z"):
            text = text[:-1] + "+00:00"
        parsed = datetime.fromisoformat(text)
        if parsed.tzinfo is None:
            parsed = parsed.replace(tzinfo=UTC)
        return int(parsed.timestamp() * 1000)
    raise TypeError(
        f"timestamp must be a datetime, number, or ISO 8601 string, got {type(value).__name__}"
    )

now_epoch_ms

now_epoch_ms() -> int

Current time as an OCSF timestamp_t.

Source code in src/py20305/forwarders/ocsf.py

def now_epoch_ms() -> int:
    """Current time as an OCSF ``timestamp_t``."""
    return int(datetime.now(UTC).timestamp() * 1000)

py20305.client.observer

Connection-outcome observer seam for :class:~py20305.client.http.Sep2Client.

A passive capture beside the client cannot recover the client's own connection outcomes from inside TLS, so the client reports them itself to whoever asks. An embedder implements :class:ConnectionObserver and attaches it at construction (Sep2Client(connection_observer=...)) or afterwards through the connection_observer property; the client then reports each logical request's outcome and each established TCP connection's addresses.

Every callback is invoked from the client's own request path, so implementations must never raise and never block: an observer that throws would fail the request it is describing.

ConnectionObserver

Bases: Protocol

What :class:~py20305.client.http.Sep2Client reports connection outcomes to.

Outcomes are reported per logical request, not per retry attempt: the retry wrapper collapses exhausted transport attempts into one error, so individual attempts are not distinguishable at this seam. A 204 No Content counts as a success — it is a validated contact that happens to signal itself by raising.

Source code in src/py20305/client/observer.py

@runtime_checkable
class ConnectionObserver(Protocol):
    """What :class:`~py20305.client.http.Sep2Client` reports connection outcomes to.

    Outcomes are reported per logical request, not per retry attempt: the
    retry wrapper collapses exhausted transport attempts into one error, so
    individual attempts are not distinguishable at this seam. A 204 No
    Content counts as a success — it is a validated contact that happens to
    signal itself by raising.
    """

    def begin_request(self) -> None:
        """A logical request is starting.

        Called before the first attempt, so the observer can scope any
        per-request attribution (for example, tying an ``on_connect`` that
        fires during the request to that request).
        """

    def record_success(self) -> None:
        """The logical request reached the server and completed."""

    def record_failure(self, exc: BaseException) -> None:
        """The logical request failed with ``exc`` after retries."""

    def on_connect(self, pair: SocketPair) -> None:
        """A TCP connection was established and passed the handshake audit.

        The local port is only knowable at this moment — a pooled connection
        carries many requests on one local port, and the response object has
        already released its connection by the time a caller could ask.
        """

    def flush(self) -> None:
        """The client is closing; emit anything the observer has buffered."""

begin_request

begin_request() -> None

A logical request is starting.

Called before the first attempt, so the observer can scope any per-request attribution (for example, tying an on_connect that fires during the request to that request).

Source code in src/py20305/client/observer.py

def begin_request(self) -> None:
    """A logical request is starting.

    Called before the first attempt, so the observer can scope any
    per-request attribution (for example, tying an ``on_connect`` that
    fires during the request to that request).
    """

record_success

record_success() -> None

The logical request reached the server and completed.

Source code in src/py20305/client/observer.py

def record_success(self) -> None:
    """The logical request reached the server and completed."""

record_failure

record_failure(exc: BaseException) -> None

The logical request failed with exc after retries.

Source code in src/py20305/client/observer.py

def record_failure(self, exc: BaseException) -> None:
    """The logical request failed with ``exc`` after retries."""

on_connect

on_connect(pair: SocketPair) -> None

A TCP connection was established and passed the handshake audit.

The local port is only knowable at this moment — a pooled connection carries many requests on one local port, and the response object has already released its connection by the time a caller could ask.

Source code in src/py20305/client/observer.py

def on_connect(self, pair: SocketPair) -> None:
    """A TCP connection was established and passed the handshake audit.

    The local port is only knowable at this moment — a pooled connection
    carries many requests on one local port, and the response object has
    already released its connection by the time a caller could ask.
    """

flush

flush() -> None

The client is closing; emit anything the observer has buffered.

Source code in src/py20305/client/observer.py

def flush(self) -> None:
    """The client is closing; emit anything the observer has buffered."""