freeswitch-sofia-trace-parser 0.6.1

use std::collections::HashMap;
use std::fs::File;
use std::io::{self, Read};
use std::process;

use clap::Parser;
use regex::Regex;
use tracing::{debug, warn};

use freeswitch_sofia_trace_parser::types::{Direction, SipMessageType};
use freeswitch_sofia_trace_parser::{
    FrameIterator, GrepFilter, MessageIterator, ParseError, ParseStats, ParsedMessageIterator,
    ParsedSipMessage, SipMessage,
};

enum OutputMode {
    Summary,
    Full,
    Headers,
    Body,
}

#[derive(Parser)]
#[command(
    name = "freeswitch-sofia-trace-parser",
    about = "Parse and filter FreeSWITCH mod_sofia SIP trace dump files"
)]
struct Cli {
    /// Dump files to parse (- for stdin, default: stdin)
    files: Vec<String>,

    /// Include SIP method (request + responses via CSeq), repeatable
    #[arg(short, long = "method", value_name = "VERB")]
    method: Vec<String>,

    /// Exclude SIP method (request + responses via CSeq), repeatable
    #[arg(short = 'x', long = "exclude", value_name = "VERB")]
    exclude: Vec<String>,

    /// Match Call-ID by regex
    #[arg(short = 'c', long = "call-id", value_name = "REGEX")]
    call_id: Option<String>,

    /// Filter by direction (recv/sent)
    #[arg(short, long, value_name = "DIR")]
    direction: Option<String>,

    /// Match address by regex
    #[arg(short, long, value_name = "REGEX")]
    address: Option<String>,

    /// Match header value by regex (NAME=REGEX), repeatable
    #[arg(short = 'H', long = "header", value_name = "NAME=REGEX")]
    header: Vec<String>,

    /// Match regex against message body
    #[arg(short = 'b', long = "body-grep", value_name = "REGEX")]
    body_grep: Option<String>,

    /// Match regex against full reconstructed SIP message
    #[arg(short = 'g', long = "grep", value_name = "REGEX")]
    grep: Option<String>,

    /// Output all messages sharing Call-IDs with matched messages
    #[arg(short = 'D', long = "dialog")]
    dialog: bool,

    /// Include OPTIONS messages (excluded by default)
    #[arg(long = "all-methods")]
    all_methods: bool,

    /// Show full SIP message content
    #[arg(long, group = "output_mode")]
    full: bool,

    /// Show headers only, no body
    #[arg(long, group = "output_mode")]
    headers: bool,

    /// Show body only
    #[arg(long, group = "output_mode")]
    body: bool,

    /// Show raw reassembled bytes (level 2)
    #[arg(long, group = "output_mode")]
    raw: bool,

    /// Show raw frames (level 1)
    #[arg(long, group = "output_mode")]
    frames: bool,

    /// Show statistics summary
    #[arg(long, group = "output_mode")]
    stats: bool,

    /// Report unparsed input regions to stderr
    #[arg(long)]
    unparsed: bool,

    /// Skip grep separator filtering (faster when reading files directly)
    #[arg(long)]
    no_grep_filter: bool,

    /// Increase verbosity (-v info, -vv debug, -vvv trace)
    #[arg(short, long, action = clap::ArgAction::Count)]
    verbose: u8,
}

struct CompiledFilters {
    methods: Vec<String>,
    excludes: Vec<String>,
    exclude_options: bool,
    call_id: Option<Regex>,
    direction: Option<Direction>,
    address: Option<Regex>,
    headers: Vec<(String, Regex)>,
    body_grep: Option<Regex>,
    grep: Option<Regex>,
}

impl CompiledFilters {
    fn is_excluded(&self, msg: &ParsedSipMessage) -> bool {
        let method = msg.method().unwrap_or("");

        if self.exclude_options && method.eq_ignore_ascii_case("OPTIONS") {
            return true;
        }

        if !self.excludes.is_empty() && self.excludes.iter().any(|m| m.eq_ignore_ascii_case(method))
        {
            return true;
        }

        false
    }

    fn matches(&self, msg: &ParsedSipMessage) -> bool {
        if self.is_excluded(msg) {
            return false;
        }

        if !self.methods.is_empty() {
            let method = msg.method().unwrap_or("");
            if !self.methods.iter().any(|m| m.eq_ignore_ascii_case(method)) {
                return false;
            }
        }

        if let Some(ref re) = self.call_id {
            match msg.call_id() {
                Some(cid) if re.is_match(cid) => {}
                _ => return false,
            }
        }

        if let Some(dir) = self.direction {
            if msg.direction != dir {
                return false;
            }
        }

        if let Some(ref re) = self.address {
            if !re.is_match(&msg.address) {
                return false;
            }
        }

        for (name, re) in &self.headers {
            let matched = msg
                .headers
                .iter()
                .filter(|(k, _)| k.eq_ignore_ascii_case(name))
                .any(|(_, v)| re.is_match(v));
            if !matched {
                return false;
            }
        }

        if let Some(ref re) = self.body_grep {
            let body_str = msg.body_text();
            if !re.is_match(&body_str) {
                return false;
            }
        }

        if let Some(ref re) = self.grep {
            let full = msg.to_bytes();
            let full_str = String::from_utf8_lossy(&full);
            if !re.is_match(&full_str) {
                return false;
            }
        }

        true
    }
}

fn compile_regex(pattern: &str, label: &str) -> Regex {
    match Regex::new(pattern) {
        Ok(re) => re,
        Err(e) => {
            eprintln!("invalid {label} regex '{pattern}': {e}");
            process::exit(2);
        }
    }
}

fn compile_filters(cli: &Cli) -> CompiledFilters {
    let methods: Vec<String> = cli.method.iter().map(|m| m.to_uppercase()).collect();
    let excludes: Vec<String> = cli.exclude.iter().map(|m| m.to_uppercase()).collect();

    let exclude_options = !cli.all_methods && !methods.iter().any(|m| m == "OPTIONS");

    let call_id = cli.call_id.as_ref().map(|p| compile_regex(p, "call-id"));

    let direction = cli.direction.as_ref().map(|d| match d.as_str() {
        "recv" => Direction::Recv,
        "sent" => Direction::Sent,
        other => {
            eprintln!("invalid direction '{other}': expected recv or sent");
            process::exit(2);
        }
    });

    let address = cli.address.as_ref().map(|p| compile_regex(p, "address"));

    let mut headers = Vec::new();
    for spec in &cli.header {
        let eq = match spec.find('=') {
            Some(pos) => pos,
            None => {
                eprintln!("invalid header filter '{spec}': expected NAME=REGEX");
                process::exit(2);
            }
        };
        let name = spec[..eq].to_string();
        let re = compile_regex(&spec[eq + 1..], &format!("header {name}"));
        headers.push((name, re));
    }

    let body_grep = cli
        .body_grep
        .as_ref()
        .map(|p| compile_regex(p, "body-grep"));
    let grep = cli.grep.as_ref().map(|p| compile_regex(p, "grep"));

    CompiledFilters {
        methods,
        excludes,
        exclude_options,
        call_id,
        direction,
        address,
        headers,
        body_grep,
        grep,
    }
}

fn output_mode(cli: &Cli) -> OutputMode {
    if cli.full {
        OutputMode::Full
    } else if cli.headers {
        OutputMode::Headers
    } else if cli.body {
        OutputMode::Body
    } else {
        OutputMode::Summary
    }
}

fn open_input(files: &[String], grep_filter: bool) -> Box<dyn Read> {
    let raw: Box<dyn Read> = if files.is_empty() || (files.len() == 1 && files[0] == "-") {
        Box::new(io::stdin().lock())
    } else {
        let mut readers: Vec<Box<dyn Read>> = Vec::new();
        for path in files {
            if path == "-" {
                readers.push(Box::new(io::stdin().lock()));
            } else {
                match File::open(path) {
                    Ok(f) => readers.push(Box::new(f)),
                    Err(e) => {
                        eprintln!("{path}: {e}");
                        process::exit(1);
                    }
                }
            }
        }

        if readers.len() == 1 {
            readers.remove(0)
        } else {
            let mut chain: Box<dyn Read> = readers.remove(0);
            for r in readers {
                chain = Box::new(chain.chain(r));
            }
            chain
        }
    };
    if grep_filter {
        Box::new(GrepFilter::new(raw))
    } else {
        raw
    }
}

fn init_tracing(verbose: u8) {
    let level = match verbose {
        0 => "warn",
        1 => "info",
        2 => "debug",
        _ => "trace",
    };
    tracing_subscriber::fmt()
        .with_env_filter(
            tracing_subscriber::EnvFilter::try_from_default_env().unwrap_or_else(|_| level.into()),
        )
        .with_writer(io::stderr)
        .init();
}

fn print_lossy(bytes: &[u8]) {
    let s = String::from_utf8_lossy(bytes);
    print!("{s}");
    if !s.ends_with('\n') {
        println!();
    }
}

fn format_summary(msg: &ParsedSipMessage) -> String {
    let call_id = msg.call_id().unwrap_or("-");
    format!(
        "{} {} {}/{} {} {}",
        msg.timestamp,
        msg.direction,
        msg.transport,
        msg.address,
        msg.message_type.summary(),
        call_id
    )
}

fn format_frame_header(msg: &ParsedSipMessage) -> String {
    format!(
        "{} {} {}/{} at {} ({} frames) {}",
        msg.direction,
        msg.direction.preposition(),
        msg.transport,
        msg.address,
        msg.timestamp,
        msg.frame_count,
        msg.message_type.summary(),
    )
}

fn output_full(msg: &ParsedSipMessage) {
    println!("{}", format_frame_header(msg));
    print_lossy(&msg.to_bytes());
}

fn output_headers(msg: &ParsedSipMessage) {
    println!("{}", format_frame_header(msg));
    match &msg.message_type {
        SipMessageType::Request { method, uri } => {
            println!("{method} {uri} SIP/2.0");
        }
        SipMessageType::Response { code, reason } => {
            println!("SIP/2.0 {code} {reason}");
        }
    }
    for (name, value) in &msg.headers {
        println!("{name}: {value}");
    }
}

fn output_body(msg: &ParsedSipMessage) {
    if !msg.body.is_empty() {
        print_lossy(&msg.body);
    }
}

fn output_message(mode: &OutputMode, msg: &ParsedSipMessage) {
    match mode {
        OutputMode::Summary => println!("{}", format_summary(msg)),
        OutputMode::Full => output_full(msg),
        OutputMode::Headers => output_headers(msg),
        OutputMode::Body => output_body(msg),
    }
}

fn run_frames(reader: Box<dyn Read>, capture_skipped: bool) -> ParseStats {
    let mut iter = FrameIterator::new(reader).capture_skipped(capture_skipped);
    for result in &mut iter {
        match result {
            Ok(frame) => {
                println!(
                    "{} {} bytes {} {}/{} at {}",
                    frame.direction,
                    frame.byte_count,
                    frame.direction.preposition(),
                    frame.transport,
                    frame.address,
                    frame.timestamp,
                );
                print_lossy(&frame.content);
            }
            Err(ref e) => log_parse_error("frame error", e),
        }
    }
    iter.stats().clone()
}

fn run_raw(reader: Box<dyn Read>, capture_skipped: bool) -> ParseStats {
    let mut iter = MessageIterator::new(reader).capture_skipped(capture_skipped);
    for result in &mut iter {
        match result {
            Ok(msg) => {
                println!(
                    "{} {} {}/{} at {} ({} frames, {} bytes)",
                    msg.direction,
                    msg.direction.preposition(),
                    msg.transport,
                    msg.address,
                    msg.timestamp,
                    msg.frame_count,
                    msg.content.len(),
                );
                print_lossy(&msg.content);
            }
            Err(ref e) => log_parse_error("message error", e),
        }
    }
    iter.parse_stats().clone()
}

fn run_stats(
    reader: Box<dyn Read>,
    filters: &CompiledFilters,
    capture_skipped: bool,
) -> ParseStats {
    let mut method_counts: HashMap<String, usize> = HashMap::new();
    let mut status_counts: HashMap<u16, usize> = HashMap::new();
    let mut direction_counts: HashMap<Direction, usize> = HashMap::new();
    let mut total: usize = 0;
    let mut matched: usize = 0;
    let mut errors: usize = 0;
    let mut total_frames: usize = 0;
    let mut multi_frame_msgs: usize = 0;
    let mut max_frame_count: usize = 0;

    let mut iter = ParsedMessageIterator::new(reader).capture_skipped(capture_skipped);
    for result in &mut iter {
        total += 1;
        match result {
            Ok(msg) => {
                total_frames += msg.frame_count;
                if msg.frame_count > 1 {
                    multi_frame_msgs += 1;
                    max_frame_count = max_frame_count.max(msg.frame_count);
                }
                if !filters.matches(&msg) {
                    continue;
                }
                matched += 1;
                *direction_counts.entry(msg.direction).or_default() += 1;
                match &msg.message_type {
                    SipMessageType::Request { method, .. } => {
                        *method_counts.entry(method.clone()).or_default() += 1;
                    }
                    SipMessageType::Response { code, .. } => {
                        *status_counts.entry(*code).or_default() += 1;
                        if let Some(method) = msg.method() {
                            *method_counts.entry(method.to_string()).or_default() += 1;
                        }
                    }
                }
            }
            Err(_) => errors += 1,
        }
    }
    let stats = iter.parse_stats().clone();

    println!("total: {total}");
    println!("matched: {matched}");
    if errors > 0 {
        println!("parse errors: {errors}");
    }

    if let Some(&n) = direction_counts.get(&Direction::Recv) {
        println!("recv: {n}");
    }
    if let Some(&n) = direction_counts.get(&Direction::Sent) {
        println!("sent: {n}");
    }

    println!("\nreassembly:");
    println!("  frames: {total_frames}");
    println!("  multi-frame messages: {multi_frame_msgs}");
    if max_frame_count > 1 {
        println!("  max frames per message: {max_frame_count}");
    }

    if stats.bytes_read > 0 {
        let parsed_pct =
            ((stats.bytes_read - stats.bytes_skipped) as f64 / stats.bytes_read as f64) * 100.0;
        println!("\ninput:");
        println!("  bytes: {}", stats.bytes_read);
        println!(
            "  parsed: {:.3}% ({}/{})",
            parsed_pct,
            stats.bytes_read - stats.bytes_skipped,
            stats.bytes_read
        );
        if stats.bytes_skipped > 0 {
            println!("  skipped: {} bytes", stats.bytes_skipped);
        }
    }

    let mut methods: Vec<_> = method_counts.into_iter().collect();
    methods.sort_by(|a, b| b.1.cmp(&a.1));
    if !methods.is_empty() {
        println!("\nmethods:");
        for (method, count) in &methods {
            println!("  {method}: {count}");
        }
    }

    let mut statuses: Vec<_> = status_counts.into_iter().collect();
    statuses.sort_by_key(|&(code, _)| code);
    if !statuses.is_empty() {
        println!("\nresponse codes:");
        for (code, count) in &statuses {
            println!("  {code}: {count}");
        }
    }

    stats
}

fn run_filtered(
    reader: Box<dyn Read>,
    mode: &OutputMode,
    filters: &CompiledFilters,
    capture_skipped: bool,
) -> ParseStats {
    let mut iter = ParsedMessageIterator::new(reader).capture_skipped(capture_skipped);
    for result in &mut iter {
        match result {
            Ok(msg) => {
                if !filters.matches(&msg) {
                    continue;
                }
                output_message(mode, &msg);
            }
            Err(ref e) => log_parse_error("parse error", e),
        }
    }
    iter.parse_stats().clone()
}

struct DialogState {
    messages: Vec<SipMessage>,
    matched: bool,
    saw_bye: bool,
    saw_bye_response: bool,
}

fn run_dialog(
    reader: Box<dyn Read>,
    mode: &OutputMode,
    filters: &CompiledFilters,
    capture_skipped: bool,
) -> ParseStats {
    let mut dialogs: HashMap<String, DialogState> = HashMap::new();

    let mut iter = MessageIterator::new(reader).capture_skipped(capture_skipped);
    // Single pass: collect messages by Call-ID, track matches
    for result in &mut iter {
        let sip_msg = match result {
            Ok(m) => m,
            Err(ref e) => {
                log_parse_error("message error", e);
                continue;
            }
        };

        let parsed = match sip_msg.parse() {
            Ok(p) => p,
            Err(ref e) => {
                log_parse_error("parse error", e);
                continue;
            }
        };

        if filters.is_excluded(&parsed) {
            continue;
        }

        let call_id = match parsed.call_id() {
            Some(cid) => cid.to_string(),
            None => continue,
        };

        let is_match = filters.matches(&parsed);

        // Detect BYE and BYE responses for pruning
        let is_bye_request = matches!(
            &parsed.message_type,
            SipMessageType::Request { method, .. } if method.eq_ignore_ascii_case("BYE")
        );
        let is_bye_response = matches!(&parsed.message_type, SipMessageType::Response { .. })
            && parsed
                .method()
                .map(|m| m.eq_ignore_ascii_case("BYE"))
                .unwrap_or(false);

        let state = dialogs.entry(call_id).or_insert_with(|| DialogState {
            messages: Vec::new(),
            matched: false,
            saw_bye: false,
            saw_bye_response: false,
        });

        if is_match {
            state.matched = true;
        }
        if is_bye_request {
            state.saw_bye = true;
        }
        if is_bye_response {
            state.saw_bye_response = true;
        }

        state.messages.push(sip_msg);

        // Prune: dialog terminated and never matched
        if state.saw_bye && state.saw_bye_response && !state.matched {
            dialogs.remove(parsed.call_id().unwrap());
        }
    }

    // Output matched dialogs in chronological order
    let mut matched_messages: Vec<SipMessage> = Vec::new();
    for (_, state) in dialogs {
        if state.matched {
            matched_messages.extend(state.messages);
        }
    }

    // Sort by timestamp for chronological output across Call-IDs
    matched_messages.sort_by_key(|m| m.timestamp.sort_key());

    for sip_msg in &matched_messages {
        match sip_msg.parse() {
            Ok(parsed) => output_message(mode, &parsed),
            Err(ref e) => log_parse_error("parse error on output", e),
        }
    }

    iter.parse_stats().clone()
}

fn log_parse_error(context: &str, e: &ParseError) {
    match e {
        ParseError::TransportNoise { .. } => debug!("{context}: {e}"),
        _ => warn!("{context}: {e}"),
    }
}

fn encode_qp(data: &[u8]) -> String {
    use quoted_printable::{encode_with_options, InputMode, Options};
    let opts = Options::default()
        .input_mode(InputMode::Binary)
        .line_length_limit(usize::MAX);
    encode_with_options(data, opts)
}

fn print_unparsed(stats: &ParseStats) {
    for region in &stats.unparsed_regions {
        eprintln!(
            "{}-{} ({} bytes): {}",
            region.offset,
            region.offset + region.length - 1,
            region.length,
            region.reason
        );
        if let Some(data) = &region.data {
            eprintln!("{}", encode_qp(data));
        }
    }
}

fn main() {
    let cli = Cli::parse();
    init_tracing(cli.verbose);

    if cli.dialog && (cli.raw || cli.frames) {
        eprintln!("--dialog is incompatible with --raw and --frames");
        process::exit(2);
    }

    if cli.dialog && cli.stats {
        eprintln!("--dialog is incompatible with --stats");
        process::exit(2);
    }

    let capture = cli.unparsed;

    let stats = if cli.frames {
        run_frames(open_input(&cli.files, !cli.no_grep_filter), capture)
    } else if cli.raw {
        run_raw(open_input(&cli.files, !cli.no_grep_filter), capture)
    } else {
        let filters = compile_filters(&cli);
        let mode = output_mode(&cli);

        if cli.dialog {
            run_dialog(
                open_input(&cli.files, !cli.no_grep_filter),
                &mode,
                &filters,
                capture,
            )
        } else if cli.stats {
            run_stats(
                open_input(&cli.files, !cli.no_grep_filter),
                &filters,
                capture,
            )
        } else {
            run_filtered(
                open_input(&cli.files, !cli.no_grep_filter),
                &mode,
                &filters,
                capture,
            )
        }
    };

    if cli.unparsed {
        print_unparsed(&stats);
    }
}

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

    #[test]
    fn plain_ascii_passthrough() {
        assert_eq!(encode_qp(b"Content-Length: 0"), "Content-Length: 0");
    }

    #[test]
    fn crlf_encoded() {
        assert_eq!(
            encode_qp(b"\r\n\r\nContent-Length: 0"),
            "=0D=0A=0D=0AContent-Length: 0"
        );
    }

    #[test]
    fn binary_encoded() {
        assert_eq!(encode_qp(&[0x00, 0x01, 0xFF]), "=00=01=FF");
    }

    #[test]
    fn equals_sign_encoded() {
        assert_eq!(encode_qp(b"a=b"), "a=3Db");
    }

    #[test]
    fn no_soft_line_breaks() {
        let long = "A".repeat(200);
        let encoded = encode_qp(long.as_bytes());
        assert!(!encoded.contains('='), "should not insert soft line breaks");
        assert_eq!(encoded, long);
    }
}