RedBear-OS/netstack/src/observer.rs

//! Packet observer / capture facility — mirrors Linux 7.1's AF_PACKET + tcpdump.
//!
//! Provides a ring buffer that captures packets flowing through the network
//! stack. Exposed via `/scheme/netcfg/capture` for reading by diagnostic tools.
//!
//! Usage:
//!   cat /scheme/netcfg/capture/read   →  drain captured packets (hex dump)
//!   cat /scheme/netcfg/capture/count  →  stats
//!   echo tcp > /scheme/netcfg/capture/filter  →  capture only TCP
//!   echo > /scheme/netcfg/capture/enable      →  start
//!   echo > /scheme/netcfg/capture/disable     →  stop
//!   echo tcp port 80 > /scheme/netcfg/capture/filter  →  TCP port 80 only

use std::cell::RefCell;
use std::rc::Rc;
use std::sync::atomic::{AtomicBool, AtomicU64, Ordering};

const MAX_CAPTURE_PACKETS: usize = 256;
const MAX_CAPTURE_BYTES: usize = 65536;

pub struct Observer {
    enabled: AtomicBool,
    buffer: RefCell<Vec<Vec<u8>>>,
    filter: RefCell<CaptureFilter>,
    max_packets: usize,
    total_captured: AtomicU64,
}

#[derive(Debug, Clone)]
struct CaptureFilter {
    proto: Option<u8>,
    port: Option<u16>,
}

impl CaptureFilter {
    fn new() -> Self {
        Self { proto: None, port: None }
    }

    fn matches(&self, packet: &[u8]) -> bool {
        if self.proto.is_none() && self.port.is_none() {
            return true;
        }
        if packet.len() < 20 {
            return false; // too short to determine protocol/port
        }
        let version = packet[0] >> 4;
        let (proto, src_port_offset, dst_port_offset) = match version {
            4 => {
                if packet.len() < 24 { return true; }
                let ihl = (packet[0] & 0x0f) as usize * 4;
                if packet.len() < ihl + 4 { return true; }
                (packet[9], ihl, ihl + 2)
            }
            6 => {
                if packet.len() < 48 { return true; }
                (packet[6], 40, 42)
            }
            _ => return true,
        };
        if let Some(p) = self.proto {
            if proto != p { return false; }
        }
        if let Some(port) = self.port {
            let sp = u16::from_be_bytes([packet[src_port_offset], packet[src_port_offset + 1]]);
            let dp = u16::from_be_bytes([packet[dst_port_offset], packet[dst_port_offset + 1]]);
            if sp != port && dp != port { return false; }
        }
        true
    }

    fn from_str(s: &str) -> Self {
        let mut filter = Self::new();
        let lower = s.trim().to_lowercase();
        let parts: Vec<&str> = lower.split_whitespace().collect();
        let mut i = 0;
        while i < parts.len() {
            match parts[i] {
                "tcp" => filter.proto = Some(6),
                "udp" => filter.proto = Some(17),
                "icmp" => filter.proto = Some(1),
                "port" if i + 1 < parts.len() => {
                    if let Ok(p) = parts[i + 1].parse::<u16>() {
                        filter.port = Some(p);
                        i += 1;
                    }
                }
                _ => {}
            }
            i += 1;
        }
        filter
    }
}

pub type ObserverRef = Rc<Observer>;

impl Observer {
    pub fn new() -> ObserverRef {
        Rc::new(Observer {
            enabled: AtomicBool::new(false),
            buffer: RefCell::new(Vec::with_capacity(MAX_CAPTURE_PACKETS)),
            filter: RefCell::new(CaptureFilter::new()),
            max_packets: MAX_CAPTURE_PACKETS,
            total_captured: AtomicU64::new(0),
        })
    }

    pub fn enable(&self) {
        self.enabled.store(true, Ordering::Relaxed);
    }

    pub fn disable(&self) {
        self.enabled.store(false, Ordering::Relaxed);
    }

    pub fn is_enabled(&self) -> bool {
        self.enabled.load(Ordering::Relaxed)
    }

    pub fn set_filter(&self, filter_str: &str) {
        *self.filter.borrow_mut() = CaptureFilter::from_str(filter_str);
    }

    pub fn filter_str(&self) -> String {
        let f = self.filter.borrow();
        let mut s = String::new();
        if let Some(p) = f.proto {
            s.push_str(match p { 6 => "tcp", 17 => "udp", 1 => "icmp", _ => "?" });
        }
        if let Some(port) = f.port {
            if !s.is_empty() { s.push(' '); }
            s.push_str(&format!("port {}", port));
        }
        if s.is_empty() { s.push_str("any"); }
        s
    }

    pub fn capture(&self, packet: &[u8]) {
        if !self.is_enabled() {
            return;
        }
        if !self.filter.borrow().matches(packet) {
            return;
        }
        let mut buf = self.buffer.borrow_mut();
        if buf.len() >= self.max_packets {
            buf.remove(0);
        }
        // Truncate to per-packet size limit so a single large packet
        // can't exhaust the total capture buffer.
        let per_packet_limit = MAX_CAPTURE_BYTES / self.max_packets.max(1);
        let truncated_len = packet.len().min(per_packet_limit);
        let mut copy = Vec::with_capacity(truncated_len);
        copy.extend_from_slice(&packet[..truncated_len]);
        buf.push(copy);
        self.total_captured.fetch_add(1, Ordering::Relaxed);
    }

    pub fn drain_hex(&self) -> String {
        let mut out = String::new();
        let mut buf = self.buffer.borrow_mut();
        for packet in buf.drain(..) {
            out.push_str(&format!("# {} bytes\n", packet.len()));
            for chunk in packet.chunks(16) {
                for byte in chunk {
                    out.push_str(&format!("{:02x} ", byte));
                }
                out.push('\n');
            }
            out.push('\n');
        }
        out
    }

    pub fn len(&self) -> usize {
        self.buffer.borrow().len()
    }

    pub fn total(&self) -> u64 {
        self.total_captured.load(Ordering::Relaxed)
    }
}