use std::marker::PhantomData;
use flowscope::{AnomalyKind, EndReason, FlowSide, FlowStats, L4Proto, TcpInfo, Timestamp};
use crate::protocol::{FlowKey, FlowProtocol, MessageProtocol, Protocol};
pub trait Event: Send + Sync + 'static {
type Payload: Send + Sync + 'static;
fn protocol_marker() -> Option<std::any::TypeId> {
None
}
fn protocol_name() -> &'static str {
"unknown"
}
fn traffic_class() -> crate::protocol::TrafficClass {
crate::protocol::TrafficClass::Any
}
}
impl<P: MessageProtocol> Event for P {
type Payload = P::Message;
fn protocol_marker() -> Option<std::any::TypeId> {
Some(std::any::TypeId::of::<P>())
}
fn protocol_name() -> &'static str {
P::NAME
}
fn traffic_class() -> crate::protocol::TrafficClass {
crate::protocol::TrafficClass::Dispatch(P::dispatch())
}
}
#[non_exhaustive]
pub struct FlowStarted<P: Protocol> {
pub key: FlowKey,
pub l4: Option<L4Proto>,
pub ts: Timestamp,
_marker: PhantomData<fn() -> P>,
}
impl<P: Protocol> FlowStarted<P> {
#[cfg(feature = "bench-zero-alloc")]
pub fn new_for_bench(key: FlowKey, l4: Option<L4Proto>, ts: Timestamp) -> Self {
Self::new(key, l4, ts)
}
#[doc(hidden)]
pub fn new(key: FlowKey, l4: Option<L4Proto>, ts: Timestamp) -> Self {
Self {
key,
l4,
ts,
_marker: PhantomData,
}
}
}
impl<P: FlowProtocol> Event for FlowStarted<P> {
type Payload = FlowStarted<P>;
fn traffic_class() -> crate::protocol::TrafficClass {
crate::protocol::TrafficClass::Dispatch(P::dispatch())
}
}
impl<P: Protocol> std::fmt::Debug for FlowStarted<P> {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
f.debug_struct("FlowStarted")
.field("protocol", &P::NAME)
.field("key", &self.key)
.field("l4", &self.l4)
.field("ts", &self.ts)
.finish()
}
}
#[non_exhaustive]
pub struct FlowEnded<P: Protocol> {
pub key: FlowKey,
pub reason: EndReason,
pub stats: FlowStats,
pub l4: Option<L4Proto>,
pub ts: Timestamp,
_marker: PhantomData<fn() -> P>,
}
impl<P: Protocol> FlowEnded<P> {
#[doc(hidden)]
pub fn new(
key: FlowKey,
reason: EndReason,
stats: FlowStats,
l4: Option<L4Proto>,
ts: Timestamp,
) -> Self {
Self {
key,
reason,
stats,
l4,
ts,
_marker: PhantomData,
}
}
}
impl<P: FlowProtocol> Event for FlowEnded<P> {
type Payload = FlowEnded<P>;
fn traffic_class() -> crate::protocol::TrafficClass {
crate::protocol::TrafficClass::Dispatch(P::dispatch())
}
}
impl<P: Protocol> std::fmt::Debug for FlowEnded<P> {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
f.debug_struct("FlowEnded")
.field("protocol", &P::NAME)
.field("key", &self.key)
.field("reason", &self.reason)
.field("l4", &self.l4)
.field("ts", &self.ts)
.finish()
}
}
#[non_exhaustive]
pub struct FlowEstablished<P: Protocol> {
pub key: FlowKey,
pub ts: Timestamp,
_marker: PhantomData<fn() -> P>,
}
impl<P: Protocol> FlowEstablished<P> {
#[doc(hidden)]
pub fn new(key: FlowKey, ts: Timestamp) -> Self {
Self {
key,
ts,
_marker: PhantomData,
}
}
}
impl<P: FlowProtocol> Event for FlowEstablished<P> {
type Payload = FlowEstablished<P>;
fn traffic_class() -> crate::protocol::TrafficClass {
crate::protocol::TrafficClass::Dispatch(P::dispatch())
}
}
impl<P: Protocol> std::fmt::Debug for FlowEstablished<P> {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
f.debug_struct("FlowEstablished")
.field("protocol", &P::NAME)
.field("key", &self.key)
.field("ts", &self.ts)
.finish()
}
}
#[non_exhaustive]
#[derive(Debug, Clone)]
pub struct FlowPacket {
pub proto: L4Proto,
pub key: FlowKey,
pub side: FlowSide,
pub len: usize,
pub tcp: Option<TcpInfo>,
pub ts: Timestamp,
}
impl FlowPacket {
#[doc(hidden)]
pub fn new(
proto: L4Proto,
key: FlowKey,
side: FlowSide,
len: usize,
tcp: Option<TcpInfo>,
ts: Timestamp,
) -> Self {
Self {
proto,
key,
side,
len,
tcp,
ts,
}
}
}
impl Event for FlowPacket {
type Payload = FlowPacket;
}
#[non_exhaustive]
#[derive(Debug, Clone)]
pub struct TcpRst {
pub key: FlowKey,
pub stats: FlowStats,
pub ts: Timestamp,
pub zero_payload: bool,
}
impl TcpRst {
#[doc(hidden)]
pub fn new(key: FlowKey, stats: FlowStats, ts: Timestamp) -> Self {
let zero_payload = stats.total_bytes() == 0;
Self {
key,
stats,
ts,
zero_payload,
}
}
}
impl Event for TcpRst {
type Payload = TcpRst;
fn traffic_class() -> crate::protocol::TrafficClass {
crate::protocol::TrafficClass::Dispatch(crate::protocol::Dispatch::AllTcp)
}
}
#[cfg(feature = "icmp")]
pub use flowscope::icmp::IcmpFamily;
#[cfg(feature = "icmp")]
#[non_exhaustive]
#[derive(Debug, Clone)]
pub struct IcmpError {
pub family: IcmpFamily,
pub kind: IcmpErrorKind,
pub correlated_flow: Option<FlowKey>,
pub stats: Option<FlowStats>,
pub ts: Timestamp,
}
#[cfg(feature = "icmp")]
#[non_exhaustive]
#[derive(Debug, Clone)]
pub enum IcmpErrorKind {
DestUnreachable(flowscope::icmp::DestUnreachableKind),
TimeExceeded,
ParameterProblem,
MtuSignal(flowscope::icmp::MtuSignalKind),
}
#[cfg(feature = "icmp")]
impl IcmpErrorKind {
pub fn as_str(&self) -> &'static str {
match self {
Self::DestUnreachable(k) => k.as_str(),
Self::TimeExceeded => "time_exceeded",
Self::ParameterProblem => "parameter_problem",
Self::MtuSignal(k) => k.as_str(),
}
}
}
#[cfg(feature = "icmp")]
impl Event for IcmpError {
type Payload = IcmpError;
fn traffic_class() -> crate::protocol::TrafficClass {
crate::protocol::TrafficClass::Dispatch(crate::protocol::Dispatch::Icmp)
}
}
#[cfg(feature = "icmp")]
pub(crate) fn classify_icmp_error(msg: &flowscope::icmp::IcmpMessage) -> Option<IcmpErrorKind> {
if !msg.is_error() {
return None;
}
if let Some(k) = msg.mtu_signal() {
return Some(IcmpErrorKind::MtuSignal(k));
}
if let Some(k) = msg.dest_unreachable_kind() {
return Some(IcmpErrorKind::DestUnreachable(k));
}
match msg.short_kind() {
"time_exceeded" => Some(IcmpErrorKind::TimeExceeded),
"parameter_problem" => Some(IcmpErrorKind::ParameterProblem),
_ => None,
}
}
#[non_exhaustive]
pub struct FlowTick<P: Protocol> {
pub key: FlowKey,
pub stats: FlowStats,
pub ts: Timestamp,
_marker: PhantomData<fn() -> P>,
}
impl<P: Protocol> FlowTick<P> {
#[doc(hidden)]
pub fn new(key: FlowKey, stats: FlowStats, ts: Timestamp) -> Self {
Self {
key,
stats,
ts,
_marker: PhantomData,
}
}
}
impl<P: FlowProtocol> Event for FlowTick<P> {
type Payload = FlowTick<P>;
fn traffic_class() -> crate::protocol::TrafficClass {
crate::protocol::TrafficClass::Dispatch(P::dispatch())
}
}
impl<P: Protocol> std::fmt::Debug for FlowTick<P> {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
f.debug_struct("FlowTick")
.field("protocol", &P::NAME)
.field("key", &self.key)
.field("stats", &self.stats)
.field("ts", &self.ts)
.finish()
}
}
#[non_exhaustive]
pub struct ParserClosed<P: Protocol> {
pub key: FlowKey,
pub parser_kind: &'static str,
pub reason: EndReason,
pub ts: Timestamp,
_marker: PhantomData<fn() -> P>,
}
impl<P: Protocol> ParserClosed<P> {
#[doc(hidden)]
pub fn new(key: FlowKey, parser_kind: &'static str, reason: EndReason, ts: Timestamp) -> Self {
Self {
key,
parser_kind,
reason,
ts,
_marker: PhantomData,
}
}
}
impl<P: Protocol> Event for ParserClosed<P> {
type Payload = ParserClosed<P>;
fn traffic_class() -> crate::protocol::TrafficClass {
crate::protocol::TrafficClass::Dispatch(P::dispatch())
}
}
impl<P: Protocol> std::fmt::Debug for ParserClosed<P> {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
f.debug_struct("ParserClosed")
.field("protocol", &P::NAME)
.field("key", &self.key)
.field("parser_kind", &self.parser_kind)
.field("reason", &self.reason)
.field("ts", &self.ts)
.finish()
}
}
#[derive(Debug)]
#[non_exhaustive]
pub struct AnyFlowAnomaly {
pub key: Option<FlowKey>,
pub kind: AnomalyKind,
pub ts: Timestamp,
}
impl Event for AnyFlowAnomaly {
type Payload = AnyFlowAnomaly;
}
#[derive(Debug, Clone, Copy)]
#[non_exhaustive]
pub struct Tick {
pub now: Timestamp,
pub period: std::time::Duration,
}
impl Tick {
#[doc(hidden)]
pub fn new(now: Timestamp, period: std::time::Duration) -> Self {
Self { now, period }
}
}
impl Event for Tick {
type Payload = Tick;
}
pub use flowscope::FlowSide as Side;
#[cfg(test)]
mod tests {
use super::*;
use crate::protocol::builtin::{Tcp, Udp};
#[test]
fn flow_started_typed_by_protocol() {
fn _accept_tcp(_: &FlowStarted<Tcp>) {}
fn _accept_udp(_: &FlowStarted<Udp>) {}
}
#[cfg(feature = "icmp")]
#[test]
fn classify_and_join_icmpv4_port_unreachable() {
let mut payload = vec![3u8, 3, 0, 0, 0, 0, 0, 0];
payload.extend_from_slice(&[0x45, 0, 0x00, 0x28, 0, 0, 0, 0, 64, 6, 0, 0]);
payload.extend_from_slice(&[10, 0, 0, 1]); payload.extend_from_slice(&[10, 0, 0, 2]); payload.extend_from_slice(&12345u16.to_be_bytes()); payload.extend_from_slice(&80u16.to_be_bytes()); payload.extend_from_slice(&[0, 0, 0, 1]);
let msg = flowscope::icmp::parse_v4(&payload).expect("parses");
let kind = classify_icmp_error(&msg).expect("is an error");
assert_eq!(kind.as_str(), "port_unreachable");
assert!(matches!(kind, IcmpErrorKind::DestUnreachable(_)));
let (_, inner) = msg.error_inner().expect("has inner");
let key =
flowscope::extract::FiveTupleKey::from_inner_canonical(inner).expect("builds a key");
assert_eq!(key.proto, flowscope::L4Proto::Tcp);
let echo = flowscope::icmp::parse_v4(&[8u8, 0, 0, 0, 0x12, 0x34, 0x56, 0x78]).unwrap();
assert!(classify_icmp_error(&echo).is_none());
}
#[test]
fn typed_events_are_send_sync() {
fn assert_send<T: Send>() {}
fn assert_sync<T: Sync>() {}
assert_send::<FlowStarted<Tcp>>();
assert_sync::<FlowStarted<Tcp>>();
assert_send::<FlowEnded<Tcp>>();
assert_sync::<FlowEnded<Tcp>>();
assert_send::<FlowEstablished<Tcp>>();
assert_send::<AnyFlowAnomaly>();
assert_send::<Tick>();
}
#[test]
fn event_trait_blanket_impl_for_protocol_markers() {
#[cfg(feature = "http")]
{
use crate::protocol::builtin::Http;
fn _accept_event<E: Event>() {}
_accept_event::<Http>();
}
}
#[test]
fn flow_started_debug_includes_protocol_name() {
use std::net::{IpAddr, Ipv4Addr, SocketAddr};
let key = flowscope::extract::FiveTupleKey {
proto: flowscope::L4Proto::Tcp,
a: SocketAddr::new(IpAddr::V4(Ipv4Addr::new(10, 0, 0, 1)), 12345),
b: SocketAddr::new(IpAddr::V4(Ipv4Addr::new(10, 0, 0, 2)), 80),
};
let evt = FlowStarted::<Tcp>::new(key, Some(flowscope::L4Proto::Tcp), Timestamp::new(0, 0));
let s = format!("{evt:?}");
assert!(s.contains("tcp"));
}
}