flowscope 0.25.2

//! [`FlowExtractor`] trait and its supporting types.
//!
//! Implement this trait to teach the rest of `netring-flow` (the
//! tracker, the reassembler hook) what counts as a flow in your
//! domain. Built-in implementations live in [`crate::extract`].

use bitflags::bitflags;

use crate::view::PacketView;

/// Extract a flow descriptor from one packet.
///
/// Implementations are called once per packet on the hot path —
/// keep them cheap and stateless. Most return `Some(_)`; malformed,
/// non-IP, or out-of-scope packets return `None` and are skipped.
///
/// # Bidirectional flows
///
/// If you want A→B and B→A merged into one flow, your extractor
/// must produce the **same `Key`** for both orientations and report
/// each packet's direction via [`Extracted::orientation`]. The
/// built-in [`crate::extract::FiveTuple::bidirectional`] does this
/// by sorting `(addr, port)` pairs.
///
/// # Bounds
///
/// `Send + Sync + 'static` is required so a tracker generic over
/// this trait can be used from any task / thread.
pub trait FlowExtractor: Send + Sync + 'static {
    /// The flow key. Equality + hashability identify the flow.
    type Key: Eq + std::hash::Hash + Clone + Send + Sync + 'static;

    /// Extract a flow descriptor from `view`.
    ///
    /// Returns `None` if this packet is not part of any flow you
    /// want to track (skipped, malformed, encap-only, ARP, etc.).
    fn extract(&self, view: PacketView<'_>) -> Option<Extracted<Self::Key>>;
}

/// Result of extracting one packet.
///
/// `key` identifies the flow. `orientation` says whether `view` was
/// in the canonical direction or reversed. `l4`, `tcp` and `l4_meta`
/// carry pre-parsed protocol data that the tracker, the reassemblers
/// and the session engines reuse without re-parsing.
///
/// Offsets in `tcp` and `l4_meta` are relative to the frame the
/// **outermost** extractor was called with — decap combinators
/// ([`crate::extract::InnerVxlan`], …) rebase the inner extractor's
/// offsets with [`Extracted::rebased`], so the tracker can slice the
/// original frame.
#[derive(Debug, Clone)]
#[non_exhaustive]
pub struct Extracted<K> {
    /// The flow this packet belongs to.
    pub key: K,

    /// Orientation of *this packet* relative to the canonical form
    /// of `key`. `Forward` if the natural src→dst direction matches
    /// the key's a→b ordering; `Reverse` if the extractor swapped
    /// to canonicalize.
    ///
    /// The tracker translates this into [`crate::FlowSide`] (Initiator
    /// / Responder) based on which orientation it saw first.
    pub orientation: Orientation,

    /// L4 protocol if the extractor identified one. Drives the
    /// tracker's choice of timeout and whether to engage TCP state.
    pub l4: Option<L4Proto>,

    /// Pre-parsed TCP info for TCP packets. If `Some`, the tracker
    /// runs the TCP state machine without re-parsing; if `None`,
    /// TCP-specific events (Established, history string) won't fire
    /// for this flow.
    ///
    /// Built-in extractors fill this for ~zero extra cost; custom
    /// extractors that don't care about TCP can leave it `None`.
    pub tcp: Option<TcpInfo>,

    /// Ports and L4 payload location, when known. The session
    /// engines use it for port-based slot selection and to hand
    /// datagram parsers their payload without re-parsing the frame.
    /// `None` makes them fall back to parsing the frame as plain
    /// Ethernet (wrong for tunnelled traffic). Built-in extractors
    /// fill it. New in 0.25.0.
    pub l4_meta: Option<L4Meta>,
}

impl<K> Extracted<K> {
    /// Construct an extraction result (no [`L4Meta`]; add it with
    /// [`Self::with_l4_meta`]).
    pub fn new(
        key: K,
        orientation: Orientation,
        l4: Option<L4Proto>,
        tcp: Option<TcpInfo>,
    ) -> Self {
        Self {
            key,
            orientation,
            l4,
            tcp,
            l4_meta: None,
        }
    }

    /// Attach ports / L4 payload location. New in 0.25.0.
    pub fn with_l4_meta(mut self, meta: L4Meta) -> Self {
        self.l4_meta = Some(meta);
        self
    }

    /// Shift every frame offset (`tcp.payload_offset`,
    /// `l4_meta.payload_offset`) by `delta` bytes. Decap combinators
    /// call this to translate offsets from the inner (or synthetic)
    /// frame they handed their inner extractor into offsets in the
    /// frame they were given. An offset that would become negative
    /// drops the TCP payload / L4 meta rather than pointing at the
    /// wrong bytes. New in 0.25.0.
    pub fn rebased(mut self, delta: isize) -> Self {
        if delta == 0 {
            return self;
        }
        if let Some(t) = self.tcp.as_mut() {
            match t.payload_offset.checked_add_signed(delta) {
                Some(o) => t.payload_offset = o,
                None => t.payload_len = 0,
            }
        }
        if let Some(m) = self.l4_meta.as_mut() {
            match m.payload_offset.checked_add_signed(delta) {
                Some(o) => m.payload_offset = o,
                None => self.l4_meta = None,
            }
        }
        self
    }
}

/// Ports and L4 payload location of one packet (see
/// [`Extracted::l4_meta`]). New in 0.25.0.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
#[non_exhaustive]
pub struct L4Meta {
    /// `(source, destination)` ports, for TCP / UDP / SCTP.
    pub ports: Option<(u16, u16)>,
    /// Offset of the L4 payload in the frame: after the TCP / UDP
    /// header, or the start of the whole ICMP message.
    pub payload_offset: usize,
    /// L4 payload length.
    pub payload_len: usize,
}

impl L4Meta {
    /// Construct.
    pub fn new(ports: Option<(u16, u16)>, payload_offset: usize, payload_len: usize) -> Self {
        Self {
            ports,
            payload_offset,
            payload_len,
        }
    }

    /// The payload bytes inside `frame` (empty when out of range).
    pub fn payload<'a>(&self, frame: &'a [u8]) -> &'a [u8] {
        self.payload_offset
            .checked_add(self.payload_len)
            .and_then(|end| frame.get(self.payload_offset..end))
            .unwrap_or(&[])
    }
}

/// Canonical orientation of a packet relative to its flow key —
/// the **deterministic, address-sorted** direction axis.
///
/// This is one of flowscope's three orthogonal direction axes; keep
/// it distinct from the other two:
///
/// | Axis | Type | Anchored to | Stable under tap-merge? |
/// |------|------|-------------|--------------------------|
/// | **Canonical orientation** | [`Orientation`] | address sort (`a < b`) | **yes** — deterministic |
/// | Logical role | [`FlowSide`](crate::FlowSide) | arrival order / SYN | no — first-seen can race |
/// | Physical capture leg | [`RxMetadata::source_idx`](crate::RxMetadata) | NIC / queue / interface | n/a (orthogonal) |
///
/// `Forward` / `Reverse` are computed purely from the canonical key
/// ordering ([`crate::extract::FiveTupleKey`] sorts endpoints so
/// `a < b`), so the **same wire 5-tuple always yields the same
/// `Orientation` regardless of which packet of the flow was seen
/// first**. That is the property [`FlowSide`](crate::FlowSide) lacks:
/// `FlowSide::Initiator` binds to whichever endpoint's packet *arrived
/// first*, which a two-NIC tap-merge (two queues, a race) can flip.
/// When you need a direction label that two independent captures of
/// the same flow will agree on — Community ID ordering, biflow keying,
/// dedup across capture points — use `Orientation`, not `FlowSide`.
///
/// See `docs/concepts.md` → "Direction, orientation, and capture leg"
/// for the full model and standards mapping (IPFIX `biflowDirection`
/// IE 239 / `observationPointId` IE 138).
///
/// The default is [`Orientation::Forward`] — the natural a→b
/// direction of a freshly-keyed flow.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash, Default)]
#[cfg_attr(feature = "serde", derive(serde::Serialize, serde::Deserialize))]
#[cfg_attr(feature = "serde", serde(rename_all = "snake_case"))]
pub enum Orientation {
    /// Packet's natural src→dst matches the key's a→b ordering.
    #[default]
    Forward,
    /// Extractor swapped src/dst to canonicalize; packet is
    /// flowing from key.b to key.a.
    Reverse,
}

impl Orientation {
    /// The opposite orientation (`Forward` ↔ `Reverse`).
    #[inline]
    #[must_use]
    pub fn flipped(self) -> Self {
        match self {
            Orientation::Forward => Orientation::Reverse,
            Orientation::Reverse => Orientation::Forward,
        }
    }

    /// Stable lowercase slug — `"forward"` / `"reverse"`. Suitable as
    /// a metric label or column value. Mirrors the `as_str` convention
    /// on the other strong-typed enums.
    #[inline]
    #[must_use]
    pub fn as_str(self) -> &'static str {
        match self {
            Orientation::Forward => "forward",
            Orientation::Reverse => "reverse",
        }
    }
}

/// L4 protocol identified by an extractor.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)]
#[cfg_attr(feature = "serde", derive(serde::Serialize, serde::Deserialize))]
#[cfg_attr(
    feature = "serde",
    serde(tag = "kind", content = "value", rename_all = "snake_case")
)]
#[non_exhaustive]
pub enum L4Proto {
    Tcp,
    Udp,
    Icmp,
    IcmpV6,
    Sctp,
    Other(u8),
}

#[cfg(feature = "tracker")]
impl std::fmt::Display for L4Proto {
    /// Lowercase short label matching the `flowscope_*_total{l4=…}`
    /// metric vocabulary (`tcp` / `udp` / `other`). Non-TCP /
    /// non-UDP families collapse to `other`; consumers needing a
    /// finer label match on the variant directly.
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        f.write_str(crate::obs::l4_label(Some(*self)))
    }
}

impl crate::KeyFields for L4Proto {
    /// Uppercase EVE-compatible label (`"TCP"` / `"UDP"` /
    /// `"ICMP"` / `"ICMPv6"` / `"SCTP"`). Returns `None` for
    /// [`L4Proto::Other`] (unknown numeric protocol).
    ///
    /// Sibling to [`L4Proto::canonical_name`] — `proto_str` is
    /// the uppercase Suricata/EVE-shaped variant (Option-
    /// returning for unknown protos); `canonical_name` is the
    /// lowercase always-Some metric-label variant.
    fn proto_str(&self) -> Option<&'static str> {
        Some(match self {
            L4Proto::Tcp => "TCP",
            L4Proto::Udp => "UDP",
            L4Proto::Icmp => "ICMP",
            L4Proto::IcmpV6 => "ICMPv6",
            L4Proto::Sctp => "SCTP",
            L4Proto::Other(_) => return None,
        })
    }
    fn protocol_identifier(&self) -> Option<u8> {
        Some(self.as_u8())
    }
}

impl L4Proto {
    /// Stable lowercase short slug for any `L4Proto`. Always
    /// `Some`-equivalent (never empty, never panics):
    ///
    /// - `Tcp` → `"tcp"`
    /// - `Udp` → `"udp"`
    /// - `Icmp` → `"icmp"`
    /// - `IcmpV6` → `"icmp6"`
    /// - `Sctp` → `"sctp"`
    /// - `Other(_)` → `"other"`
    ///
    /// Sibling to [`crate::KeyFields::proto_str`] (uppercase,
    /// EVE/Suricata schema-shaped, `None` for `Other`). Use
    /// this method for metric labels, log slugs, and
    /// `app_label` fallbacks where lowercase + always-Some is
    /// the right contract.
    ///
    /// Plan 163 (0.14).
    pub fn canonical_name(&self) -> &'static str {
        match self {
            L4Proto::Tcp => "tcp",
            L4Proto::Udp => "udp",
            L4Proto::Icmp => "icmp",
            L4Proto::IcmpV6 => "icmp6",
            L4Proto::Sctp => "sctp",
            L4Proto::Other(_) => "other",
        }
    }

    /// IANA IP protocol number — the wire-format value
    /// carried in the IPv4 `protocol` / IPv6 `next_header`
    /// header field. Used as IPFIX IE 4
    /// (`protocolIdentifier`).
    ///
    /// - `Tcp` → 6
    /// - `Udp` → 17
    /// - `Icmp` → 1
    /// - `IcmpV6` → 58
    /// - `Sctp` → 132
    /// - `Other(n)` → `n`
    ///
    /// New in 0.18.0 (issue #16 sub-piece — needed for the
    /// IE-keyed [`crate::ipfix::FlowRecord`]).
    pub fn as_u8(&self) -> u8 {
        match self {
            L4Proto::Tcp => 6,
            L4Proto::Udp => 17,
            L4Proto::Icmp => 1,
            L4Proto::IcmpV6 => 58,
            L4Proto::Sctp => 132,
            L4Proto::Other(n) => *n,
        }
    }
}

/// Pre-parsed TCP information for a packet.
///
/// Filled by built-in extractors. Decoupled from frame layout so
/// downstream tracker / reassembler logic doesn't need to re-parse.
///
/// `#[non_exhaustive]` since 0.5.0 — additive field changes are
/// unconditionally non-breaking. External consumers read fields by
/// name; internal constructors use struct-literal syntax.
#[derive(Debug, Clone, Copy)]
#[cfg_attr(feature = "serde", derive(serde::Serialize, serde::Deserialize))]
#[non_exhaustive]
pub struct TcpInfo {
    /// Decoded TCP flags.
    pub flags: TcpFlags,
    /// TCP sequence number (host byte order).
    pub seq: u32,
    /// TCP acknowledgment number (host byte order).
    pub ack: u32,
    /// Offset into the frame where the TCP payload begins.
    /// Relative to the frame the extractor was called with.
    pub payload_offset: usize,
    /// Number of payload bytes (zero for pure SYN/ACK/FIN).
    pub payload_len: usize,
    /// TCP receive window from the header (host byte order). Not
    /// scaled — the SYN/SYN-ACK window-scale option is per-flow and
    /// not currently tracked. Consumers wanting the effective window
    /// should pair this with their own wscale tracking until a
    /// per-flow `wscale` lands.
    pub window: u16,
}

bitflags! {
    /// TCP control flags from the TCP header's flags byte.
    #[derive(Debug, Clone, Copy, PartialEq, Eq)]
    pub struct TcpFlags: u8 {
        const FIN = 0b0000_0001;
        const SYN = 0b0000_0010;
        const RST = 0b0000_0100;
        const PSH = 0b0000_1000;
        const ACK = 0b0001_0000;
        const URG = 0b0010_0000;
        const ECE = 0b0100_0000;
        const CWR = 0b1000_0000;
    }
}

#[cfg(feature = "serde")]
impl serde::Serialize for TcpFlags {
    fn serialize<S: serde::Serializer>(&self, s: S) -> Result<S::Ok, S::Error> {
        self.bits().serialize(s)
    }
}

#[cfg(feature = "serde")]
impl<'de> serde::Deserialize<'de> for TcpFlags {
    fn deserialize<D: serde::Deserializer<'de>>(d: D) -> Result<Self, D::Error> {
        let raw = u8::deserialize(d)?;
        Ok(TcpFlags::from_bits_retain(raw))
    }
}

#[cfg(test)]
mod tests {
    use super::*;

    #[test]
    fn tcp_flags_basic() {
        let f = TcpFlags::SYN | TcpFlags::ACK;
        assert!(f.contains(TcpFlags::SYN));
        assert!(f.contains(TcpFlags::ACK));
        assert!(!f.contains(TcpFlags::FIN));
    }

    #[test]
    fn extracted_clone() {
        let e: Extracted<u32> = Extracted {
            key: 42,
            orientation: Orientation::Forward,
            l4: Some(L4Proto::Tcp),
            tcp: Some(TcpInfo {
                flags: TcpFlags::SYN,
                seq: 1,
                ack: 0,
                payload_offset: 54,
                payload_len: 0,
                window: 8192,
            }),
            l4_meta: None,
        };
        let cloned = e.clone();
        assert_eq!(cloned.key, 42);
        assert_eq!(cloned.orientation, Orientation::Forward);
    }

    #[test]
    fn l4_proto_eq() {
        assert_eq!(L4Proto::Tcp, L4Proto::Tcp);
        assert_ne!(L4Proto::Tcp, L4Proto::Udp);
        assert_eq!(L4Proto::Other(1), L4Proto::Other(1));
        assert_ne!(L4Proto::Other(1), L4Proto::Other(2));
    }
}