use std::borrow::Cow;
use std::fmt;
use std::net::SocketAddr;
pub(crate) fn expand_compact(name: &str) -> Option<&'static str> {
let [ch] = name.as_bytes() else {
return None;
};
sip_header::SipHeader::from_compact(*ch).map(|header| header.as_str())
}
pub(crate) fn value_or_compact<'a>(headers: &'a Headers, name: &str) -> Option<&'a str> {
let mut compact = None;
for (key, value) in headers.iter() {
if key.eq_ignore_ascii_case(name) {
return Some(value);
}
if compact.is_none()
&& key.len() == 1
&& expand_compact(key).is_some_and(|full| full.eq_ignore_ascii_case(name))
{
compact = Some(value.as_str());
}
}
compact
}
fn parse_socket_addr(address: &str) -> Option<SocketAddr> {
if let Ok(addr) = address.parse() {
return Some(addr);
}
let (ip, port) = address.strip_prefix('[')?.split_once("]:")?;
Some(SocketAddr::new(ip.parse().ok()?, port.parse().ok()?))
}
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
#[non_exhaustive]
pub enum SkipReason {
PartialFirstFrame,
OversizedFrame,
MidStreamSkip,
ReplayedFrame,
IncompleteFrame,
InvalidHeader,
}
impl fmt::Display for SkipReason {
fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
match self {
SkipReason::PartialFirstFrame => f.write_str("partial first frame"),
SkipReason::OversizedFrame => f.write_str("oversized frame"),
SkipReason::MidStreamSkip => f.write_str("mid-stream skip"),
SkipReason::ReplayedFrame => f.write_str("replayed frame (logrotate)"),
SkipReason::IncompleteFrame => f.write_str("incomplete frame"),
SkipReason::InvalidHeader => f.write_str("invalid header"),
}
}
}
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum SkipTracking {
CountOnly,
TrackRegions,
CaptureData,
}
#[derive(Debug, Clone)]
#[non_exhaustive]
pub struct UnparsedRegion {
pub offset: u64,
pub length: u64,
pub reason: SkipReason,
pub data: Option<Vec<u8>>,
}
#[derive(Debug, Default, Clone)]
pub struct ParseStats {
pub(crate) bytes_read: u64,
pub(crate) bytes_skipped: u64,
pub(crate) incomplete_frames: u64,
pub(crate) incomplete_frame_bytes: u64,
pub(crate) stale_evictions: u64,
pub(crate) stale_evicted_bytes: u64,
pub(crate) non_sip_prefixes: u64,
pub(crate) non_sip_prefix_bytes: u64,
pub(crate) unparsed_regions: Vec<UnparsedRegion>,
}
impl ParseStats {
pub fn bytes_read(&self) -> u64 {
self.bytes_read
}
pub fn bytes_skipped(&self) -> u64 {
self.bytes_skipped
}
pub fn incomplete_frames(&self) -> u64 {
self.incomplete_frames
}
pub fn incomplete_frame_bytes(&self) -> u64 {
self.incomplete_frame_bytes
}
pub fn stale_evictions(&self) -> u64 {
self.stale_evictions
}
pub fn stale_evicted_bytes(&self) -> u64 {
self.stale_evicted_bytes
}
pub fn non_sip_prefixes(&self) -> u64 {
self.non_sip_prefixes
}
pub fn non_sip_prefix_bytes(&self) -> u64 {
self.non_sip_prefix_bytes
}
pub fn unparsed_regions(&self) -> &[UnparsedRegion] {
&self.unparsed_regions
}
pub fn drain_regions(&mut self) -> Vec<UnparsedRegion> {
std::mem::take(&mut self.unparsed_regions)
}
}
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct UnknownKeyword(String);
impl UnknownKeyword {
pub fn as_str(&self) -> &str {
&self.0
}
}
impl fmt::Display for UnknownKeyword {
fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
write!(f, "unknown keyword: {}", self.0)
}
}
impl std::error::Error for UnknownKeyword {}
#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)]
pub enum Direction {
Recv,
Sent,
}
impl fmt::Display for Direction {
fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
f.write_str(self.as_str())
}
}
impl std::str::FromStr for Direction {
type Err = UnknownKeyword;
fn from_str(s: &str) -> Result<Self, Self::Err> {
for candidate in [Direction::Recv, Direction::Sent] {
if s.eq_ignore_ascii_case(candidate.as_str()) {
return Ok(candidate);
}
}
Err(UnknownKeyword(s.to_string()))
}
}
impl Direction {
pub fn as_str(&self) -> &'static str {
match self {
Direction::Recv => "recv",
Direction::Sent => "sent",
}
}
pub fn preposition(&self) -> &'static str {
match self {
Direction::Recv => "from",
Direction::Sent => "to",
}
}
}
#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)]
pub enum Transport {
Tcp,
Udp,
Tls,
Wss,
}
impl Transport {
pub fn as_str(&self) -> &'static str {
match self {
Transport::Tcp => "tcp",
Transport::Udp => "udp",
Transport::Tls => "tls",
Transport::Wss => "wss",
}
}
}
impl fmt::Display for Transport {
fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
f.write_str(self.as_str())
}
}
impl std::str::FromStr for Transport {
type Err = UnknownKeyword;
fn from_str(s: &str) -> Result<Self, Self::Err> {
for candidate in [
Transport::Tcp,
Transport::Udp,
Transport::Tls,
Transport::Wss,
] {
if s.eq_ignore_ascii_case(candidate.as_str()) {
return Ok(candidate);
}
}
Err(UnknownKeyword(s.to_string()))
}
}
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum Timestamp {
TimeOnly {
hour: u8,
min: u8,
sec: u8,
usec: u32,
},
DateTime {
year: u16,
month: u8,
day: u8,
hour: u8,
min: u8,
sec: u8,
usec: u32,
},
}
impl Timestamp {
pub fn time_of_day_secs(&self) -> u32 {
let (h, m, s) = match self {
Timestamp::TimeOnly { hour, min, sec, .. } => (*hour, *min, *sec),
Timestamp::DateTime { hour, min, sec, .. } => (*hour, *min, *sec),
};
h as u32 * 3600 + m as u32 * 60 + s as u32
}
pub fn sort_key(&self) -> (u16, u8, u8, u8, u8, u8, u32) {
match self {
Timestamp::TimeOnly {
hour,
min,
sec,
usec,
} => (0, 0, 0, *hour, *min, *sec, *usec),
Timestamp::DateTime {
year,
month,
day,
hour,
min,
sec,
usec,
} => (*year, *month, *day, *hour, *min, *sec, *usec),
}
}
}
pub(crate) fn days_from_civil(y: i64, m: u32, d: u32) -> i64 {
let y = if m <= 2 { y - 1 } else { y };
let era = if y >= 0 { y } else { y - 399 } / 400;
let yoe = (y - era * 400) as u64;
let m_adj = if m > 2 { m as i64 - 3 } else { m as i64 + 9 } as u64;
let doy = (153 * m_adj + 2) / 5 + d as u64 - 1;
let doe = yoe * 365 + yoe / 4 - yoe / 100 + doy;
era * 146097 + doe as i64 - 719468
}
#[derive(Debug, Default, Clone)]
pub struct StaleClock {
day: u32,
last_time_secs: u32,
now: u64,
last_sweep: u64,
dated: Option<bool>,
reset: bool,
}
impl StaleClock {
pub const TIMEOUT_SECS: u64 = 7200;
pub fn new() -> Self {
Self::default()
}
pub fn observe(&mut self, timestamp: Timestamp) -> u64 {
let dated = matches!(timestamp, Timestamp::DateTime { .. });
if self.dated != Some(dated) {
self.dated = Some(dated);
self.day = 0;
self.last_time_secs = 0;
self.reset = true;
}
let time_secs = timestamp.time_of_day_secs();
self.now = match timestamp {
Timestamp::DateTime {
year, month, day, ..
} => {
let days = days_from_civil(year as i64, month as u32, day as u32).max(0) as u64;
days * 86400 + time_secs as u64
}
Timestamp::TimeOnly { .. } => {
if time_secs < self.last_time_secs && self.last_time_secs - time_secs > 43200 {
self.day += 1;
}
self.day as u64 * 86400 + time_secs as u64
}
};
self.last_time_secs = time_secs;
self.now
}
pub fn now(&self) -> u64 {
self.now
}
pub fn sweep_due(&mut self) -> bool {
if self.reset {
self.reset = false;
self.last_sweep = self.now;
return false;
}
if self.now.saturating_sub(self.last_sweep) >= Self::TIMEOUT_SECS {
self.last_sweep = self.now;
return true;
}
false
}
pub fn is_stale(&self, last_seen: u64) -> bool {
self.now.saturating_sub(last_seen) > Self::TIMEOUT_SECS
}
}
impl fmt::Display for Timestamp {
fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
match self {
Timestamp::TimeOnly {
hour,
min,
sec,
usec,
} => write!(f, "{hour:02}:{min:02}:{sec:02}.{usec:06}"),
Timestamp::DateTime {
year,
month,
day,
hour,
min,
sec,
usec,
} => write!(
f,
"{year:04}-{month:02}-{day:02} {hour:02}:{min:02}:{sec:02}.{usec:06}"
),
}
}
}
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub struct FrameMeta<'a> {
pub direction: Direction,
pub transport: Transport,
pub address: &'a str,
pub timestamp: Timestamp,
}
impl FrameMeta<'_> {
pub fn socket_addr(&self) -> Option<SocketAddr> {
parse_socket_addr(self.address)
}
}
impl fmt::Display for FrameMeta<'_> {
fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
write!(
f,
"{} {} {}/{} at {}",
self.direction,
self.direction.preposition(),
self.transport,
self.address,
self.timestamp
)
}
}
#[derive(Debug, Clone)]
pub struct Frame {
pub direction: Direction,
pub byte_count: usize,
pub transport: Transport,
pub address: String,
pub timestamp: Timestamp,
pub content: Vec<u8>,
}
impl Frame {
pub fn meta(&self) -> FrameMeta<'_> {
FrameMeta {
direction: self.direction,
transport: self.transport,
address: &self.address,
timestamp: self.timestamp,
}
}
pub fn socket_addr(&self) -> Option<SocketAddr> {
self.meta().socket_addr()
}
}
#[derive(Debug, Clone)]
pub struct SipMessage {
pub direction: Direction,
pub transport: Transport,
pub address: String,
pub timestamp: Timestamp,
pub content: Vec<u8>,
pub frame_count: usize,
}
impl SipMessage {
pub fn meta(&self) -> FrameMeta<'_> {
FrameMeta {
direction: self.direction,
transport: self.transport,
address: &self.address,
timestamp: self.timestamp,
}
}
pub fn socket_addr(&self) -> Option<SocketAddr> {
self.meta().socket_addr()
}
}
#[derive(Debug, Clone, PartialEq, Eq)]
pub enum SipMessageType {
Request {
method: String,
uri: String,
},
Response {
code: u16,
reason: String,
},
}
impl fmt::Display for SipMessageType {
fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
match self {
SipMessageType::Request { method, uri } => write!(f, "{method} {uri}"),
SipMessageType::Response { code, reason } => write!(f, "{code} {reason}"),
}
}
}
impl SipMessageType {
pub fn summary(&self) -> Cow<'_, str> {
match self {
SipMessageType::Request { method, .. } => Cow::Borrowed(method),
SipMessageType::Response { code, reason } => Cow::Owned(format!("{code} {reason}")),
}
}
}
#[derive(Debug, Clone, Default, PartialEq, Eq)]
pub struct Headers(Vec<(String, String)>);
impl Headers {
pub fn values<'a>(&'a self, name: &'a str) -> impl Iterator<Item = &'a str> + 'a {
self.0
.iter()
.filter(move |(k, _)| k.eq_ignore_ascii_case(name))
.map(|(_, v)| v.as_str())
}
pub fn value(&self, name: &str) -> Option<&str> {
self.0
.iter()
.find(|(k, _)| k.eq_ignore_ascii_case(name))
.map(|(_, v)| v.as_str())
}
}
impl std::ops::Deref for Headers {
type Target = [(String, String)];
fn deref(&self) -> &Self::Target {
&self.0
}
}
impl std::ops::DerefMut for Headers {
fn deref_mut(&mut self) -> &mut Self::Target {
&mut self.0
}
}
impl From<Vec<(String, String)>> for Headers {
fn from(headers: Vec<(String, String)>) -> Self {
Headers(headers)
}
}
impl FromIterator<(String, String)> for Headers {
fn from_iter<I: IntoIterator<Item = (String, String)>>(iter: I) -> Self {
Headers(iter.into_iter().collect())
}
}
impl<'a> IntoIterator for &'a Headers {
type Item = &'a (String, String);
type IntoIter = std::slice::Iter<'a, (String, String)>;
fn into_iter(self) -> Self::IntoIter {
self.0.iter()
}
}
#[derive(Debug, Clone)]
pub struct ParsedSipMessage {
pub direction: Direction,
pub transport: Transport,
pub address: String,
pub timestamp: Timestamp,
pub message_type: SipMessageType,
pub headers: Headers,
pub body: Vec<u8>,
pub frame_count: usize,
}
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct SipFragment {
pub message_type: Option<SipMessageType>,
pub headers: Headers,
pub body: Vec<u8>,
}
impl SipFragment {
pub fn header_value(&self, name: &str) -> Option<&str> {
self.headers.value(name)
}
pub fn content_type(&self) -> Option<&str> {
value_or_compact(&self.headers, "Content-Type")
}
}
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct MimePart {
pub headers: Headers,
pub body: Vec<u8>,
}
impl MimePart {
pub fn content_type(&self) -> Option<&str> {
value_or_compact(&self.headers, "Content-Type")
}
pub fn header_value(&self, name: &str) -> Option<&str> {
self.headers.value(name)
}
pub fn content_id(&self) -> Option<&str> {
self.header_value("Content-ID")
}
pub fn content_disposition(&self) -> Option<&str> {
self.header_value("Content-Disposition")
}
pub fn content_transfer_encoding(&self) -> Option<&str> {
self.header_value("Content-Transfer-Encoding")
}
}
impl ParsedSipMessage {
pub fn meta(&self) -> FrameMeta<'_> {
FrameMeta {
direction: self.direction,
transport: self.transport,
address: &self.address,
timestamp: self.timestamp,
}
}
pub fn socket_addr(&self) -> Option<SocketAddr> {
self.meta().socket_addr()
}
pub fn call_id(&self) -> Option<&str> {
value_or_compact(&self.headers, "Call-ID")
}
pub fn content_type(&self) -> Option<&str> {
value_or_compact(&self.headers, "Content-Type")
}
pub fn content_length(&self) -> Option<usize> {
value_or_compact(&self.headers, "Content-Length").and_then(|v| v.trim().parse().ok())
}
pub fn cseq(&self) -> Option<&str> {
self.header_value("CSeq")
}
pub fn method(&self) -> Option<&str> {
match &self.message_type {
SipMessageType::Request { method, .. } => Some(method),
SipMessageType::Response { .. } => {
self.cseq().and_then(|cs| cs.split_whitespace().nth(1))
}
}
}
pub fn body_data(&self) -> Cow<'_, str> {
String::from_utf8_lossy(&self.body)
}
pub fn to_bytes(&self) -> Vec<u8> {
let mut out = Vec::new();
match &self.message_type {
SipMessageType::Request { method, uri } => {
out.extend_from_slice(format!("{method} {uri} SIP/2.0\r\n").as_bytes());
}
SipMessageType::Response { code, reason } => {
out.extend_from_slice(format!("SIP/2.0 {code} {reason}\r\n").as_bytes());
}
}
for (name, value) in &self.headers {
out.extend_from_slice(format!("{name}: {value}\r\n").as_bytes());
}
out.extend_from_slice(b"\r\n");
out.extend_from_slice(&self.body);
out
}
pub fn header_value(&self, name: &str) -> Option<&str> {
self.headers.value(name)
}
}
#[cfg(test)]
mod tests {
use super::*;
fn make_parsed(
msg_type: SipMessageType,
headers: Vec<(&str, &str)>,
body: &[u8],
) -> ParsedSipMessage {
ParsedSipMessage {
direction: Direction::Recv,
transport: Transport::Tcp,
address: "10.0.0.1:5060".into(),
timestamp: Timestamp::TimeOnly {
hour: 12,
min: 0,
sec: 0,
usec: 0,
},
message_type: msg_type,
headers: Headers(
headers
.iter()
.map(|(k, v)| (k.to_string(), v.to_string()))
.collect(),
),
body: body.to_vec(),
frame_count: 1,
}
}
fn make_frame(address: &str) -> Frame {
Frame {
direction: Direction::Recv,
byte_count: 0,
transport: Transport::Tcp,
address: address.into(),
timestamp: Timestamp::TimeOnly {
hour: 0,
min: 0,
sec: 0,
usec: 0,
},
content: Vec::new(),
}
}
fn make_message(address: &str) -> SipMessage {
SipMessage {
direction: Direction::Recv,
transport: Transport::Tcp,
address: address.into(),
timestamp: Timestamp::TimeOnly {
hour: 0,
min: 0,
sec: 0,
usec: 0,
},
content: Vec::new(),
frame_count: 1,
}
}
fn parsed_with_address(address: &str) -> ParsedSipMessage {
let mut msg = make_parsed(
SipMessageType::Request {
method: "OPTIONS".into(),
uri: "sip:host".into(),
},
vec![],
b"",
);
msg.address = address.into();
msg
}
#[test]
fn socket_addr_ipv4() {
let addr = make_frame("10.0.0.1:5060").socket_addr().unwrap();
assert!(addr.is_ipv4());
assert_eq!(addr.port(), 5060);
assert_eq!(addr.ip().to_string(), "10.0.0.1");
}
#[test]
fn socket_addr_ipv6_bracketed() {
let addr = make_message("[2001:db8::1]:5061").socket_addr().unwrap();
assert!(addr.is_ipv6());
assert_eq!(addr.port(), 5061);
assert_eq!(addr.ip().to_string(), "2001:db8::1");
}
#[test]
fn socket_addr_ipv4_bracketed() {
let addr = make_frame("[198.51.100.7]:5060").socket_addr().unwrap();
assert!(addr.is_ipv4());
assert_eq!(addr.port(), 5060);
assert_eq!(addr.ip().to_string(), "198.51.100.7");
}
#[test]
fn socket_addr_on_parsed_message() {
let addr = parsed_with_address("192.0.2.4:5080").socket_addr().unwrap();
assert_eq!(addr.port(), 5080);
}
#[test]
fn socket_addr_rejects_non_addresses() {
for bad in [
"345.678.987.654:5060",
"10.0.0.1",
"host.example.test:5060",
"2001:db8::1:5060",
"",
] {
assert!(
make_frame(bad).socket_addr().is_none(),
"should not parse: {bad}"
);
assert!(make_message(bad).socket_addr().is_none());
assert!(parsed_with_address(bad).socket_addr().is_none());
}
}
#[test]
fn to_bytes_request_no_body() {
let msg = make_parsed(
SipMessageType::Request {
method: "OPTIONS".into(),
uri: "sip:host".into(),
},
vec![("Call-ID", "test")],
b"",
);
let bytes = msg.to_bytes();
let text = String::from_utf8(bytes).unwrap();
assert!(text.starts_with("OPTIONS sip:host SIP/2.0\r\n"));
assert!(text.contains("Call-ID: test\r\n"));
assert!(text.ends_with("\r\n\r\n"));
}
#[test]
fn to_bytes_request_with_body() {
let body = b"v=0\r\ns=-\r\n";
let msg = make_parsed(
SipMessageType::Request {
method: "INVITE".into(),
uri: "sip:host".into(),
},
vec![("Call-ID", "test")],
body,
);
let bytes = msg.to_bytes();
assert!(bytes.ends_with(body));
}
#[test]
fn to_bytes_response() {
let msg = make_parsed(
SipMessageType::Response {
code: 200,
reason: "OK".into(),
},
vec![("Call-ID", "resp-test")],
b"",
);
let bytes = msg.to_bytes();
let text = String::from_utf8(bytes).unwrap();
assert!(text.starts_with("SIP/2.0 200 OK\r\n"));
}
#[test]
fn body_data_valid_utf8() {
let msg = make_parsed(
SipMessageType::Request {
method: "MESSAGE".into(),
uri: "sip:host".into(),
},
vec![],
b"hello world",
);
assert_eq!(&*msg.body_data(), "hello world");
}
#[test]
fn body_data_empty() {
let msg = make_parsed(
SipMessageType::Request {
method: "OPTIONS".into(),
uri: "sip:host".into(),
},
vec![],
b"",
);
assert_eq!(&*msg.body_data(), "");
}
#[test]
fn body_data_binary() {
let msg = make_parsed(
SipMessageType::Request {
method: "MESSAGE".into(),
uri: "sip:host".into(),
},
vec![],
&[0xFF, 0xFE],
);
assert!(msg.body_data().contains('\u{FFFD}'));
}
#[test]
fn body_data_preserves_json_escapes() {
let raw = br#"{"key":"value\nwith\\escapes"}"#;
let msg = make_parsed(
SipMessageType::Request {
method: "NOTIFY".into(),
uri: "sip:host".into(),
},
vec![("Content-Type", "application/json")],
raw,
);
assert_eq!(
msg.body_data().as_ref(),
r#"{"key":"value\nwith\\escapes"}"#,
"body_data() must preserve raw escapes"
);
}
}