m1-7
This commit is contained in:
CarterPerez-dev
parent
e4207b343a
commit
7845d9a1d9
|
|
@ -1422,8 +1422,11 @@ dependencies = [
|
|||
"pcap",
|
||||
"pcap-parser",
|
||||
"rusqlite",
|
||||
"rustix",
|
||||
"serde",
|
||||
"serde_json",
|
||||
"smallvec",
|
||||
"thiserror",
|
||||
"tlsfp-core",
|
||||
"tokio",
|
||||
"tower",
|
||||
|
|
@ -1438,7 +1441,6 @@ version = "0.1.0"
|
|||
dependencies = [
|
||||
"aes",
|
||||
"aes-gcm",
|
||||
"ctr",
|
||||
"etherparse",
|
||||
"hex",
|
||||
"hkdf",
|
||||
|
|
|
|||
|
|
@ -32,6 +32,7 @@ pcap = "2"
|
|||
|
||||
tokio = { version = "1", features = ["rt-multi-thread", "macros", "sync", "signal"] }
|
||||
flume = "0.11"
|
||||
rustix = { version = "1", features = ["std", "event"] }
|
||||
|
||||
rusqlite = { version = "0.32", features = ["bundled"] }
|
||||
|
||||
|
|
@ -47,7 +48,6 @@ clap = { version = "4", features = ["derive"] }
|
|||
hkdf = "0.12"
|
||||
aes = "0.8"
|
||||
aes-gcm = "0.10"
|
||||
ctr = "0.9"
|
||||
|
||||
[workspace.lints.rust]
|
||||
unsafe_code = "forbid"
|
||||
|
|
|
|||
|
|
@ -26,7 +26,6 @@ pcap-parser.workspace = true
|
|||
hkdf.workspace = true
|
||||
aes.workspace = true
|
||||
aes-gcm.workspace = true
|
||||
ctr.workspace = true
|
||||
|
||||
[dev-dependencies]
|
||||
hex.workspace = true
|
||||
|
|
|
|||
|
|
@ -49,9 +49,6 @@ pub enum ParseError {
|
|||
|
||||
#[error("QUIC header protection or AEAD removal failed")]
|
||||
QuicCryptoFailure,
|
||||
|
||||
#[error("no CRYPTO frame carrying a ClientHello was present")]
|
||||
NoCryptoClientHello,
|
||||
}
|
||||
|
||||
pub type Result<T> = core::result::Result<T, ParseError>;
|
||||
|
|
|
|||
|
|
@ -21,6 +21,7 @@ pub mod ja4t;
|
|||
pub mod ja4x;
|
||||
pub mod parse;
|
||||
pub mod pipeline;
|
||||
pub mod quic;
|
||||
pub mod registry;
|
||||
|
||||
pub use error::{ParseError, Result};
|
||||
|
|
|
|||
|
|
@ -117,24 +117,50 @@ pub struct DecodedSegment<'pkt> {
|
|||
pub payload: &'pkt [u8],
|
||||
}
|
||||
|
||||
/// Why a frame produced no segment. The distinction only feeds counters, but
|
||||
/// the counters are how an operator learns what a capture was made of.
|
||||
/// One UDP datagram decoded out of a captured frame.
|
||||
///
|
||||
/// Addresses are directional like a TCP segment's. The payload is the UDP
|
||||
/// data, which the pipeline hands to the QUIC layer; a datagram that turns
|
||||
/// out not to be QUIC is simply ignored there, since UDP carries far more
|
||||
/// than QUIC and the fingerprinter only reads the handshake it understands.
|
||||
#[derive(Debug)]
|
||||
pub struct DecodedDatagram<'pkt> {
|
||||
pub src: SocketAddr,
|
||||
pub dst: SocketAddr,
|
||||
pub payload: &'pkt [u8],
|
||||
}
|
||||
|
||||
/// One transport payload decoded out of a captured frame.
|
||||
///
|
||||
/// The decoder surfaces the two transports the pipeline fingerprints: TCP,
|
||||
/// which carries TLS and HTTP, and UDP, which carries QUIC. Everything else
|
||||
/// is a [`Skip`].
|
||||
#[derive(Debug)]
|
||||
pub enum Decoded<'pkt> {
|
||||
Tcp(DecodedSegment<'pkt>),
|
||||
Udp(DecodedDatagram<'pkt>),
|
||||
}
|
||||
|
||||
/// Why a frame produced no transport payload. The distinction only feeds
|
||||
/// counters, but the counters are how an operator learns what a capture was
|
||||
/// made of.
|
||||
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
|
||||
pub enum Skip {
|
||||
UnsupportedLinkType,
|
||||
NotIp,
|
||||
NotTcp,
|
||||
NotTransport,
|
||||
Malformed,
|
||||
}
|
||||
|
||||
/// Decodes a captured frame down to its TCP segment, if it has one.
|
||||
/// Decodes a captured frame down to its transport payload, if it has one the
|
||||
/// pipeline fingerprints.
|
||||
///
|
||||
/// VLAN tags, including stacked QinQ, are stepped over by etherparse. Frames
|
||||
/// the decoder does not understand are skipped with a reason rather than
|
||||
/// failing the capture: a fingerprinting pipeline must shrug off the GRE
|
||||
/// tunnel, the ARP chatter, and the malformed frame that share every real
|
||||
/// network with the TLS it cares about.
|
||||
pub fn decode_frame(link: i32, data: &[u8]) -> Result<DecodedSegment<'_>, Skip> {
|
||||
/// network with the TLS and QUIC it cares about.
|
||||
pub fn decode_frame(link: i32, data: &[u8]) -> Result<Decoded<'_>, Skip> {
|
||||
let sliced = match link {
|
||||
link_type::ETHERNET => SlicedPacket::from_ethernet(data),
|
||||
link_type::LINUX_SLL => SlicedPacket::from_linux_sll(data),
|
||||
|
|
@ -170,26 +196,32 @@ pub fn decode_frame(link: i32, data: &[u8]) -> Result<DecodedSegment<'_>, Skip>
|
|||
Some(NetSlice::Arp(_)) | None => return Err(Skip::NotIp),
|
||||
};
|
||||
|
||||
let Some(TransportSlice::Tcp(tcp)) = &sliced.transport else {
|
||||
return Err(Skip::NotTcp);
|
||||
};
|
||||
match &sliced.transport {
|
||||
Some(TransportSlice::Tcp(tcp)) => {
|
||||
let flags = TcpFlags::from_header(tcp.slice());
|
||||
let syn_fingerprint = flags
|
||||
.syn()
|
||||
.then(|| tcp_fingerprint_input(tcp.window_size(), tcp.options()));
|
||||
|
||||
let flags = TcpFlags::from_header(tcp.slice());
|
||||
let syn_fingerprint = flags
|
||||
.syn()
|
||||
.then(|| tcp_fingerprint_input(tcp.window_size(), tcp.options()));
|
||||
|
||||
Ok(DecodedSegment {
|
||||
src: SocketAddr::new(src_ip, tcp.source_port()),
|
||||
dst: SocketAddr::new(dst_ip, tcp.destination_port()),
|
||||
tcp: TcpMeta {
|
||||
seq: tcp.sequence_number(),
|
||||
flags,
|
||||
window_size: tcp.window_size(),
|
||||
},
|
||||
syn_fingerprint,
|
||||
payload: tcp.payload(),
|
||||
})
|
||||
Ok(Decoded::Tcp(DecodedSegment {
|
||||
src: SocketAddr::new(src_ip, tcp.source_port()),
|
||||
dst: SocketAddr::new(dst_ip, tcp.destination_port()),
|
||||
tcp: TcpMeta {
|
||||
seq: tcp.sequence_number(),
|
||||
flags,
|
||||
window_size: tcp.window_size(),
|
||||
},
|
||||
syn_fingerprint,
|
||||
payload: tcp.payload(),
|
||||
}))
|
||||
}
|
||||
Some(TransportSlice::Udp(udp)) => Ok(Decoded::Udp(DecodedDatagram {
|
||||
src: SocketAddr::new(src_ip, udp.source_port()),
|
||||
dst: SocketAddr::new(dst_ip, udp.destination_port()),
|
||||
payload: udp.payload(),
|
||||
})),
|
||||
_ => Err(Skip::NotTransport),
|
||||
}
|
||||
}
|
||||
|
||||
/// Walks raw TCP options into the JA4T input.
|
||||
|
|
@ -244,10 +276,17 @@ pub fn tcp_fingerprint_input(window_size: u16, options: &[u8]) -> TcpFingerprint
|
|||
|
||||
#[cfg(test)]
|
||||
mod tests {
|
||||
use super::{Skip, TcpFlags, decode_frame, link_type, tcp_fingerprint_input};
|
||||
use super::{Decoded, Skip, TcpFlags, decode_frame, link_type, tcp_fingerprint_input};
|
||||
use crate::ja4t::ja4t;
|
||||
use etherparse::PacketBuilder;
|
||||
|
||||
fn tcp_segment(link: i32, data: &[u8]) -> super::DecodedSegment<'_> {
|
||||
match decode_frame(link, data) {
|
||||
Ok(Decoded::Tcp(seg)) => seg,
|
||||
other => panic!("expected a TCP segment, got {other:?}"),
|
||||
}
|
||||
}
|
||||
|
||||
fn tcp_frame(payload: &[u8]) -> Vec<u8> {
|
||||
let builder = PacketBuilder::ethernet2([1; 6], [2; 6])
|
||||
.ipv4([10, 0, 0, 1], [10, 0, 0, 2], 64)
|
||||
|
|
@ -274,7 +313,7 @@ mod tests {
|
|||
#[test]
|
||||
fn decodes_an_ethernet_tcp_frame() {
|
||||
let frame = tcp_frame(b"hello");
|
||||
let seg = decode_frame(link_type::ETHERNET, &frame).unwrap();
|
||||
let seg = tcp_segment(link_type::ETHERNET, &frame);
|
||||
assert_eq!(seg.src.to_string(), "10.0.0.1:40000");
|
||||
assert_eq!(seg.dst.to_string(), "10.0.0.2:443");
|
||||
assert_eq!(seg.tcp.seq, 1000);
|
||||
|
|
@ -290,28 +329,44 @@ mod tests {
|
|||
let mut frame = Vec::with_capacity(builder.size(0));
|
||||
builder.write(&mut frame, &[]).unwrap();
|
||||
|
||||
let seg = decode_frame(link_type::ETHERNET, &frame).unwrap();
|
||||
let seg = tcp_segment(link_type::ETHERNET, &frame);
|
||||
assert_eq!(seg.dst.port(), 443);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn non_tcp_and_garbage_are_skips_not_panics() {
|
||||
fn decodes_a_udp_datagram() {
|
||||
let builder = PacketBuilder::ethernet2([1; 6], [2; 6])
|
||||
.ipv4([10, 0, 0, 1], [10, 0, 0, 2], 64)
|
||||
.udp(5000, 53);
|
||||
.udp(50000, 443);
|
||||
let mut udp = Vec::with_capacity(builder.size(4));
|
||||
builder.write(&mut udp, &[0xde, 0xad, 0xbe, 0xef]).unwrap();
|
||||
|
||||
let Ok(Decoded::Udp(datagram)) = decode_frame(link_type::ETHERNET, &udp) else {
|
||||
panic!("expected a UDP datagram");
|
||||
};
|
||||
assert_eq!(datagram.src.to_string(), "10.0.0.1:50000");
|
||||
assert_eq!(datagram.dst.port(), 443);
|
||||
assert_eq!(datagram.payload, &[0xde, 0xad, 0xbe, 0xef]);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn non_transport_and_garbage_are_skips_not_panics() {
|
||||
let builder = PacketBuilder::ethernet2([1; 6], [2; 6])
|
||||
.ipv4([10, 0, 0, 1], [10, 0, 0, 2], 64)
|
||||
.icmpv4_echo_request(1, 1);
|
||||
let mut icmp = Vec::with_capacity(builder.size(0));
|
||||
builder.write(&mut icmp, &[]).unwrap();
|
||||
|
||||
assert!(matches!(
|
||||
decode_frame(link_type::ETHERNET, &udp),
|
||||
Err(Skip::NotTcp)
|
||||
decode_frame(link_type::ETHERNET, &icmp),
|
||||
Err(Skip::NotTransport)
|
||||
));
|
||||
assert!(matches!(
|
||||
decode_frame(link_type::ETHERNET, &[0x01, 0x02]),
|
||||
Err(Skip::Malformed)
|
||||
));
|
||||
assert!(matches!(
|
||||
decode_frame(147, &udp),
|
||||
decode_frame(147, &icmp),
|
||||
Err(Skip::UnsupportedLinkType)
|
||||
));
|
||||
}
|
||||
|
|
|
|||
|
|
@ -22,12 +22,18 @@ use std::collections::HashMap;
|
|||
|
||||
use serde::Serialize;
|
||||
|
||||
use crate::ja3::{ja3, ja3_string};
|
||||
use crate::ja4::{Transport, ja4};
|
||||
use crate::ja4t::ja4t;
|
||||
use crate::pipeline::decode::{Skip, decode_frame};
|
||||
use crate::parse::parse_client_hello;
|
||||
use crate::pipeline::decode::{Decoded, DecodedDatagram, DecodedSegment, Skip, decode_frame};
|
||||
use crate::pipeline::event::{FingerprintEvent, StreamEvent};
|
||||
use crate::pipeline::flow::{FlowKey, PushOutcome, ReassemblyLimits, StreamReassembler};
|
||||
use crate::pipeline::source::{PacketSource, RawFrame, SourceError};
|
||||
use crate::pipeline::tls::StreamProtocol;
|
||||
use crate::quic::{
|
||||
ClientHelloState, CryptoAssembler, InitialKeys, InitialPacket, walk_crypto_frames,
|
||||
};
|
||||
|
||||
/// Tuning knobs for the pipeline.
|
||||
///
|
||||
|
|
@ -85,9 +91,10 @@ pub struct Counters {
|
|||
pub frames: u64,
|
||||
pub bytes: u64,
|
||||
pub tcp_segments: u64,
|
||||
pub udp_datagrams: u64,
|
||||
pub skipped_unsupported_link_type: u64,
|
||||
pub skipped_not_ip: u64,
|
||||
pub skipped_not_tcp: u64,
|
||||
pub skipped_not_transport: u64,
|
||||
pub skipped_malformed: u64,
|
||||
pub flows_created: u64,
|
||||
pub flows_evicted_idle: u64,
|
||||
|
|
@ -96,6 +103,14 @@ pub struct Counters {
|
|||
pub events: u64,
|
||||
pub streams_capped: u64,
|
||||
pub unfinished_tls_streams: u64,
|
||||
/// QUIC long header Initial packets observed, both directions.
|
||||
pub quic_initials: u64,
|
||||
/// Client Initials whose protection was removed and payload decrypted.
|
||||
/// The gap to `quic_initials` is mostly server Initials, which a passive
|
||||
/// observer cannot open, and is exactly the honesty an operator needs.
|
||||
pub quic_decrypted: u64,
|
||||
/// Initials carrying a QUIC version this build has no salt for.
|
||||
pub quic_version_unsupported: u64,
|
||||
}
|
||||
|
||||
/// One direction of one tracked flow.
|
||||
|
|
@ -133,6 +148,35 @@ impl FlowState {
|
|||
}
|
||||
}
|
||||
|
||||
/// One tracked QUIC conversation.
|
||||
///
|
||||
/// QUIC needs far less per flow state than TCP. There is one cryptographic
|
||||
/// stream to reassemble, not two byte streams, and the only message this
|
||||
/// pipeline reads from it is the ClientHello, which lives entirely in the
|
||||
/// client's first flight of Initial packets. Once the client keys are locked
|
||||
/// from the first Initial that authenticates, every later Initial on the flow
|
||||
/// reuses them, and `done` retires the flow the moment the ClientHello is in
|
||||
/// hand or the stream proves it will never hold one.
|
||||
struct QuicFlow {
|
||||
keys: Option<InitialKeys>,
|
||||
crypto: CryptoAssembler,
|
||||
largest_pn: Option<u64>,
|
||||
done: bool,
|
||||
last_seen_nanos: u64,
|
||||
}
|
||||
|
||||
impl QuicFlow {
|
||||
fn new(max_crypto_bytes: usize) -> Self {
|
||||
Self {
|
||||
keys: None,
|
||||
crypto: CryptoAssembler::new(max_crypto_bytes),
|
||||
largest_pn: None,
|
||||
done: false,
|
||||
last_seen_nanos: 0,
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/// The passive fingerprinting engine.
|
||||
///
|
||||
/// Feed it frames, take events out through the sink closure. The pipeline is
|
||||
|
|
@ -142,6 +186,7 @@ impl FlowState {
|
|||
pub struct Pipeline {
|
||||
config: PipelineConfig,
|
||||
flows: HashMap<FlowKey, FlowState>,
|
||||
quic_flows: HashMap<FlowKey, QuicFlow>,
|
||||
counters: Counters,
|
||||
}
|
||||
|
||||
|
|
@ -151,6 +196,7 @@ impl Pipeline {
|
|||
Self {
|
||||
config,
|
||||
flows: HashMap::new(),
|
||||
quic_flows: HashMap::new(),
|
||||
counters: Counters::default(),
|
||||
}
|
||||
}
|
||||
|
|
@ -173,27 +219,36 @@ impl Pipeline {
|
|||
Ok(())
|
||||
}
|
||||
|
||||
/// Processes one captured frame.
|
||||
/// Processes one captured frame, dispatching to the TCP or QUIC path.
|
||||
pub fn feed(&mut self, frame: &RawFrame<'_>, sink: &mut impl FnMut(FingerprintEvent)) {
|
||||
self.counters.frames += 1;
|
||||
self.counters.bytes += frame.data.len() as u64;
|
||||
|
||||
let segment = match decode_frame(frame.link_type, frame.data) {
|
||||
Ok(segment) => segment,
|
||||
Err(skip) => {
|
||||
match skip {
|
||||
Skip::UnsupportedLinkType => {
|
||||
self.counters.skipped_unsupported_link_type += 1;
|
||||
}
|
||||
Skip::NotIp => self.counters.skipped_not_ip += 1,
|
||||
Skip::NotTcp => self.counters.skipped_not_tcp += 1,
|
||||
Skip::Malformed => self.counters.skipped_malformed += 1,
|
||||
}
|
||||
return;
|
||||
match decode_frame(frame.link_type, frame.data) {
|
||||
Ok(Decoded::Tcp(segment)) => {
|
||||
self.counters.tcp_segments += 1;
|
||||
self.feed_tcp(&segment, frame, sink);
|
||||
}
|
||||
};
|
||||
self.counters.tcp_segments += 1;
|
||||
Ok(Decoded::Udp(datagram)) => {
|
||||
self.counters.udp_datagrams += 1;
|
||||
self.feed_quic(&datagram, frame, sink);
|
||||
}
|
||||
Err(skip) => match skip {
|
||||
Skip::UnsupportedLinkType => self.counters.skipped_unsupported_link_type += 1,
|
||||
Skip::NotIp => self.counters.skipped_not_ip += 1,
|
||||
Skip::NotTransport => self.counters.skipped_not_transport += 1,
|
||||
Skip::Malformed => self.counters.skipped_malformed += 1,
|
||||
},
|
||||
}
|
||||
}
|
||||
|
||||
/// Feeds one TCP segment into its flow's reassembler and protocol layer.
|
||||
fn feed_tcp(
|
||||
&mut self,
|
||||
segment: &DecodedSegment<'_>,
|
||||
frame: &RawFrame<'_>,
|
||||
sink: &mut impl FnMut(FingerprintEvent),
|
||||
) {
|
||||
let (key, direction) = FlowKey::from_pair(segment.src, segment.dst);
|
||||
if !self.flows.contains_key(&key) {
|
||||
if self.flows.len() >= self.config.max_flows {
|
||||
|
|
@ -261,6 +316,105 @@ impl Pipeline {
|
|||
}
|
||||
}
|
||||
|
||||
/// Feeds one UDP datagram into its QUIC flow, decrypting any client
|
||||
/// Initial packets and reassembling the ClientHello they carry.
|
||||
///
|
||||
/// A datagram may coalesce several QUIC packets, so the parse walks them
|
||||
/// in turn. Each Initial is opened with the flow's locked client keys, or,
|
||||
/// before any are locked, with keys derived from that packet's own
|
||||
/// Destination Connection ID; the AEAD tag is what confirms a packet was
|
||||
/// a client Initial rather than a server one this observer cannot read.
|
||||
/// Once the contiguous CRYPTO stream holds a complete ClientHello it is
|
||||
/// fingerprinted exactly like a TCP one, only carrying the QUIC transport
|
||||
/// marker, and the flow is retired.
|
||||
fn feed_quic(
|
||||
&mut self,
|
||||
datagram: &DecodedDatagram<'_>,
|
||||
frame: &RawFrame<'_>,
|
||||
sink: &mut impl FnMut(FingerprintEvent),
|
||||
) {
|
||||
let (key, direction) = FlowKey::from_pair(datagram.src, datagram.dst);
|
||||
if !self.quic_flows.contains_key(&key) {
|
||||
if self.quic_flows.len() >= self.config.max_flows {
|
||||
self.evict_quic(frame.ts_nanos);
|
||||
}
|
||||
self.quic_flows
|
||||
.insert(key, QuicFlow::new(self.config.max_assembled_bytes));
|
||||
}
|
||||
let Some(flow) = self.quic_flows.get_mut(&key) else {
|
||||
return;
|
||||
};
|
||||
flow.last_seen_nanos = flow.last_seen_nanos.max(frame.ts_nanos);
|
||||
if flow.done {
|
||||
return;
|
||||
}
|
||||
|
||||
let mut offset = 0usize;
|
||||
while offset < datagram.payload.len() {
|
||||
let packet = match InitialPacket::parse(datagram.payload, offset) {
|
||||
Ok(packet) => packet,
|
||||
Err(crate::error::ParseError::UnsupportedQuicVersion(_)) => {
|
||||
self.counters.quic_version_unsupported += 1;
|
||||
break;
|
||||
}
|
||||
Err(_) => break,
|
||||
};
|
||||
self.counters.quic_initials += 1;
|
||||
offset = packet.next_offset;
|
||||
|
||||
let opened = if let Some(keys) = flow.keys.as_ref() {
|
||||
packet.open(keys, flow.largest_pn).ok()
|
||||
} else {
|
||||
let candidate = InitialKeys::client(packet.dcid);
|
||||
match packet.open(&candidate, None) {
|
||||
Ok(opened) => {
|
||||
flow.keys = Some(candidate);
|
||||
Some(opened)
|
||||
}
|
||||
Err(_) => None,
|
||||
}
|
||||
};
|
||||
|
||||
let Some(opened) = opened else {
|
||||
continue;
|
||||
};
|
||||
self.counters.quic_decrypted += 1;
|
||||
flow.largest_pn = Some(
|
||||
flow.largest_pn
|
||||
.map_or(opened.packet_number, |seen| seen.max(opened.packet_number)),
|
||||
);
|
||||
let crypto = &mut flow.crypto;
|
||||
let _ = walk_crypto_frames(&opened.frames, |off, data| crypto.push(off, data));
|
||||
}
|
||||
|
||||
match flow.crypto.client_hello() {
|
||||
ClientHelloState::Ready(body) => {
|
||||
if let Ok(hello) = parse_client_hello(body) {
|
||||
let (src, dst) = direction.addresses(&key);
|
||||
sink(FingerprintEvent {
|
||||
ts_nanos: frame.ts_nanos,
|
||||
src,
|
||||
dst,
|
||||
event: StreamEvent::ClientHello {
|
||||
ja3: ja3(&hello),
|
||||
ja3_raw: ja3_string(&hello),
|
||||
ja4: ja4(&hello, Transport::Quic),
|
||||
sni: hello.server_name().map(str::to_owned),
|
||||
alpn: hello
|
||||
.alpn_protocols()
|
||||
.first()
|
||||
.map(|p| String::from_utf8_lossy(p).into_owned()),
|
||||
},
|
||||
});
|
||||
self.counters.events += 1;
|
||||
}
|
||||
flow.done = true;
|
||||
}
|
||||
ClientHelloState::NotClientHello | ClientHelloState::Abandoned => flow.done = true,
|
||||
ClientHelloState::Incomplete => {}
|
||||
}
|
||||
}
|
||||
|
||||
/// Settles the books at end of capture.
|
||||
///
|
||||
/// Streams that were recognized as TLS but never produced a complete
|
||||
|
|
@ -279,6 +433,7 @@ impl Pipeline {
|
|||
}
|
||||
}
|
||||
self.flows.clear();
|
||||
self.quic_flows.clear();
|
||||
}
|
||||
|
||||
/// Sheds flows when the table is full: everything idle past the timeout
|
||||
|
|
@ -327,6 +482,39 @@ impl Pipeline {
|
|||
}
|
||||
}
|
||||
}
|
||||
|
||||
/// Sheds QUIC flows under the same policy as the TCP table: retire the
|
||||
/// idle and the already harvested first, and if none qualify, the single
|
||||
/// stalest flow, so a fresh handshake is never turned away for a dead one.
|
||||
fn evict_quic(&mut self, now_nanos: u64) {
|
||||
let timeout = self.config.idle_timeout_nanos;
|
||||
let idle: Vec<FlowKey> = self
|
||||
.quic_flows
|
||||
.iter()
|
||||
.filter(|(_, flow)| {
|
||||
now_nanos.saturating_sub(flow.last_seen_nanos) > timeout || flow.done
|
||||
})
|
||||
.map(|(key, _)| *key)
|
||||
.collect();
|
||||
|
||||
if idle.is_empty() {
|
||||
let stalest = self
|
||||
.quic_flows
|
||||
.iter()
|
||||
.min_by_key(|(_, flow)| flow.last_seen_nanos)
|
||||
.map(|(key, _)| *key);
|
||||
if let Some(key) = stalest {
|
||||
self.quic_flows.remove(&key);
|
||||
self.counters.flows_evicted_pressure += 1;
|
||||
}
|
||||
return;
|
||||
}
|
||||
|
||||
for key in idle {
|
||||
self.quic_flows.remove(&key);
|
||||
self.counters.flows_evicted_idle += 1;
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
|
|
|
|||
|
|
@ -45,6 +45,9 @@ pub enum SourceError {
|
|||
|
||||
#[error("malformed capture: {0}")]
|
||||
Malformed(String),
|
||||
|
||||
#[error("capture source failed: {0}")]
|
||||
Capture(String),
|
||||
}
|
||||
|
||||
/// One link layer frame as captured, with the metadata needed to decode it.
|
||||
|
|
|
|||
|
|
@ -0,0 +1,782 @@
|
|||
// ©AngelaMos | 2026
|
||||
// quic.rs
|
||||
|
||||
//! QUIC version 1 Initial packet decryption, from RFC 9001 and RFC 9000.
|
||||
//!
|
||||
//! A QUIC ClientHello is not sent in the clear. It travels inside CRYPTO
|
||||
//! frames in one or more Initial packets, and those packets are encrypted,
|
||||
//! even though the key is derivable by anyone who sees the connection start.
|
||||
//! The point of that encryption is not secrecy. It is ossification defence:
|
||||
//! by protecting the Initial under a key derived from a connection ID that
|
||||
//! every observer can read, QUIC forces middleboxes to either implement the
|
||||
//! whole scheme or leave the packet alone, which keeps the wire format free
|
||||
//! to evolve. A passive fingerprinter is the rare observer that genuinely
|
||||
//! wants to read the handshake, so it implements the whole scheme.
|
||||
//!
|
||||
//! The recipe, from RFC 9001 Section 5.2 and Section 5.4, is:
|
||||
//!
|
||||
//! 1. The initial secret is `HKDF-Extract(initial_salt, dcid)`, where `dcid`
|
||||
//! is the Destination Connection ID of the client's first Initial packet
|
||||
//! and `initial_salt` is a constant fixed by the QUIC version.
|
||||
//! 2. The client traffic secret is `HKDF-Expand-Label(initial_secret,
|
||||
//! "client in", "", 32)`, and from it come the AEAD key, the AEAD IV, and
|
||||
//! the header protection key, each its own `HKDF-Expand-Label`.
|
||||
//! 3. Header protection masks the low bits of the first byte and the whole
|
||||
//! packet number with `AES-ECB(hp, sample)`, where `sample` is sixteen
|
||||
//! bytes of ciphertext taken four bytes past the start of the packet
|
||||
//! number field. Removing it reveals the packet number length and value.
|
||||
//! 4. The payload is `AEAD_AES_128_GCM`, with the unprotected header as
|
||||
//! associated data and a nonce built by XORing the packet number into the
|
||||
//! IV.
|
||||
//!
|
||||
//! Everything here is bounds checked and allocation light. A malformed or
|
||||
//! truncated Initial is an ordinary error, never a panic, because this code
|
||||
//! runs against whatever a network hands it.
|
||||
|
||||
use std::collections::BTreeMap;
|
||||
|
||||
use aes::Aes128;
|
||||
use aes::cipher::{BlockEncrypt, KeyInit as _, generic_array::GenericArray};
|
||||
use aes_gcm::Aes128Gcm;
|
||||
use aes_gcm::aead::AeadInPlace;
|
||||
use hkdf::Hkdf;
|
||||
use sha2::Sha256;
|
||||
use smallvec::SmallVec;
|
||||
|
||||
use crate::error::{ParseError, Result};
|
||||
use crate::registry::handshake_type;
|
||||
|
||||
/// The QUIC version 1 code point, from RFC 9000.
|
||||
pub const VERSION_1: u32 = 0x0000_0001;
|
||||
|
||||
/// The version 1 Initial salt, from RFC 9001 Section 5.2. This constant is
|
||||
/// what binds a set of Initial keys to one QUIC version: a different version
|
||||
/// uses a different salt, so keys derived under the wrong salt simply fail
|
||||
/// the AEAD tag.
|
||||
const INITIAL_SALT_V1: [u8; 20] = [
|
||||
0x38, 0x76, 0x2c, 0xf7, 0xf5, 0x59, 0x34, 0xb3, 0x4d, 0x17, 0x9a, 0xe6, 0xa4, 0xc8, 0x0c, 0xad,
|
||||
0xcc, 0xbb, 0x7f, 0x0a,
|
||||
];
|
||||
|
||||
const KEY_LEN: usize = 16;
|
||||
const IV_LEN: usize = 12;
|
||||
const HP_LEN: usize = 16;
|
||||
|
||||
/// The header protection sample is sixteen bytes, the AES block size.
|
||||
const SAMPLE_LEN: usize = 16;
|
||||
|
||||
/// The AEAD authentication tag occupies the last sixteen bytes of the
|
||||
/// protected payload.
|
||||
const TAG_LEN: usize = 16;
|
||||
|
||||
/// Header protection assumes a four byte packet number when locating the
|
||||
/// sample, because the true length is not known until protection is removed.
|
||||
const PN_SAMPLE_OFFSET: usize = 4;
|
||||
|
||||
/// A connection ID is at most twenty bytes in QUIC version 1 (RFC 9000
|
||||
/// Section 17.2). A longer length field marks the packet as not version 1
|
||||
/// framing and is rejected rather than trusted.
|
||||
const MAX_CID_LEN: usize = 20;
|
||||
|
||||
/// The high bit of the first byte marks a long header; the next bit is the
|
||||
/// version independent fixed bit. Initial is long header packet type `0b00`.
|
||||
const LONG_HEADER_FORM: u8 = 0x80;
|
||||
const PACKET_TYPE_MASK: u8 = 0x30;
|
||||
const PACKET_TYPE_INITIAL: u8 = 0x00;
|
||||
|
||||
/// Header protection masks the low four bits of a long header's first byte.
|
||||
const LONG_HEADER_PN_MASK: u8 = 0x0f;
|
||||
|
||||
/// The two low bits of the unprotected first byte hold the packet number
|
||||
/// length, biased by one.
|
||||
const PN_LEN_MASK: u8 = 0x03;
|
||||
|
||||
/// QUIC frame type codes that may appear in an Initial packet, from RFC 9000
|
||||
/// Section 12.4. Anything outside this set means the decrypted bytes are not
|
||||
/// a well formed Initial payload, which is treated as a decryption failure
|
||||
/// rather than guessed past.
|
||||
mod frame {
|
||||
pub const PADDING: u8 = 0x00;
|
||||
pub const PING: u8 = 0x01;
|
||||
pub const ACK: u8 = 0x02;
|
||||
pub const ACK_ECN: u8 = 0x03;
|
||||
pub const CRYPTO: u8 = 0x06;
|
||||
pub const CONNECTION_CLOSE: u8 = 0x1c;
|
||||
}
|
||||
|
||||
/// Reads a QUIC variable length integer, returning the value and the number
|
||||
/// of bytes it consumed.
|
||||
///
|
||||
/// The length lives in the top two bits of the first byte, giving a one, two,
|
||||
/// four, or eight byte encoding. A buffer too short for the indicated length
|
||||
/// is an error, not a truncated read.
|
||||
fn read_varint(buf: &[u8]) -> Result<(u64, usize)> {
|
||||
let first = *buf.first().ok_or(short(1, 0))?;
|
||||
let len = 1usize << (first >> 6);
|
||||
let bytes = buf.get(..len).ok_or(short(len, buf.len()))?;
|
||||
let mut value = u64::from(first & 0x3f);
|
||||
for &b in &bytes[1..] {
|
||||
value = (value << 8) | u64::from(b);
|
||||
}
|
||||
Ok((value, len))
|
||||
}
|
||||
|
||||
const fn short(needed: usize, have: usize) -> ParseError {
|
||||
ParseError::Truncated { needed, have }
|
||||
}
|
||||
|
||||
/// Decodes a truncated packet number into the full value, from RFC 9000
|
||||
/// Appendix A.3.
|
||||
///
|
||||
/// The wire carries only the low bits of the packet number; the rest is
|
||||
/// inferred from the largest packet number already seen so that the value
|
||||
/// lands in the window around what was expected next. The very first Initial
|
||||
/// on a flow has no prior packet, so its truncated value is taken as is,
|
||||
/// which is correct because a connection's first packet numbers start at zero
|
||||
/// and never need the window arithmetic.
|
||||
fn decode_packet_number(largest: Option<u64>, truncated: u64, pn_nbits: u32) -> u64 {
|
||||
let Some(largest) = largest else {
|
||||
return truncated;
|
||||
};
|
||||
let expected = largest + 1;
|
||||
let win = 1u64 << pn_nbits;
|
||||
let hwin = win / 2;
|
||||
let mask = win - 1;
|
||||
let candidate = (expected & !mask) | truncated;
|
||||
if candidate + hwin <= expected && candidate < (1u64 << 62) - win {
|
||||
return candidate + win;
|
||||
}
|
||||
if candidate > expected + hwin && candidate >= win {
|
||||
return candidate - win;
|
||||
}
|
||||
candidate
|
||||
}
|
||||
|
||||
/// Builds the `HkdfLabel` structure that QUIC and TLS 1.3 feed to
|
||||
/// HKDF-Expand, from RFC 8446 Section 7.1.
|
||||
///
|
||||
/// The structure is the output length as a sixteen bit integer, then the
|
||||
/// label prefixed by its own one byte length, then an empty context prefixed
|
||||
/// by a zero length byte. QUIC always uses the `tls13 ` prefix and an empty
|
||||
/// context for Initial keys.
|
||||
fn hkdf_expand_label(secret: &[u8], label: &[u8], out: &mut [u8]) {
|
||||
const PREFIX: &[u8] = b"tls13 ";
|
||||
let mut info: SmallVec<[u8; 32]> = SmallVec::new();
|
||||
let out_len = u16::try_from(out.len()).expect("Initial key material is never that long");
|
||||
info.extend_from_slice(&out_len.to_be_bytes());
|
||||
info.push(u8::try_from(PREFIX.len() + label.len()).expect("Initial labels are short"));
|
||||
info.extend_from_slice(PREFIX);
|
||||
info.extend_from_slice(label);
|
||||
info.push(0);
|
||||
|
||||
let hk = Hkdf::<Sha256>::from_prk(secret).expect("the secret is one SHA-256 block");
|
||||
hk.expand(&info, out)
|
||||
.expect("Initial output never exceeds the HKDF limit");
|
||||
}
|
||||
|
||||
/// The AEAD and header protection keys for one direction of one QUIC Initial.
|
||||
pub struct InitialKeys {
|
||||
key: [u8; KEY_LEN],
|
||||
iv: [u8; IV_LEN],
|
||||
hp: [u8; HP_LEN],
|
||||
}
|
||||
|
||||
impl InitialKeys {
|
||||
/// Derives the client's Initial keys from a Destination Connection ID,
|
||||
/// for QUIC version 1.
|
||||
///
|
||||
/// A passive observer derives these from the connection ID alone, with no
|
||||
/// secret input, which is exactly why the AEAD tag rather than secrecy is
|
||||
/// what tells a client Initial apart from a server one: only a packet the
|
||||
/// client actually protected under these keys will verify.
|
||||
#[must_use]
|
||||
pub fn client(dcid: &[u8]) -> Self {
|
||||
let (initial_secret, _) = Hkdf::<Sha256>::extract(Some(&INITIAL_SALT_V1), dcid);
|
||||
let mut client_secret = [0u8; 32];
|
||||
hkdf_expand_label(&initial_secret, b"client in", &mut client_secret);
|
||||
|
||||
let mut keys = Self {
|
||||
key: [0u8; KEY_LEN],
|
||||
iv: [0u8; IV_LEN],
|
||||
hp: [0u8; HP_LEN],
|
||||
};
|
||||
hkdf_expand_label(&client_secret, b"quic key", &mut keys.key);
|
||||
hkdf_expand_label(&client_secret, b"quic iv", &mut keys.iv);
|
||||
hkdf_expand_label(&client_secret, b"quic hp", &mut keys.hp);
|
||||
keys
|
||||
}
|
||||
}
|
||||
|
||||
/// One QUIC Initial packet located within a UDP datagram, still protected.
|
||||
///
|
||||
/// The fields outside header protection, the connection IDs, the token, and
|
||||
/// the length, are read on construction; the first byte and the packet number
|
||||
/// stay protected until [`InitialPacket::open`] removes protection and
|
||||
/// decrypts the payload. The datagram may carry several coalesced packets, so
|
||||
/// `next_offset` says where the following packet begins.
|
||||
pub struct InitialPacket<'pkt> {
|
||||
datagram: &'pkt [u8],
|
||||
/// The Destination Connection ID, which seeds key derivation.
|
||||
pub dcid: &'pkt [u8],
|
||||
/// Absolute offset of this packet's first byte within the datagram.
|
||||
start: usize,
|
||||
/// Absolute offset of the packet number field within the datagram.
|
||||
pn_offset: usize,
|
||||
/// Absolute offset one past this packet's protected payload.
|
||||
end: usize,
|
||||
/// Absolute offset where the next coalesced packet would begin.
|
||||
pub next_offset: usize,
|
||||
}
|
||||
|
||||
/// A successfully opened Initial: its decrypted frame bytes and packet number.
|
||||
pub struct OpenedInitial {
|
||||
pub packet_number: u64,
|
||||
pub frames: Vec<u8>,
|
||||
}
|
||||
|
||||
impl<'pkt> InitialPacket<'pkt> {
|
||||
/// Parses one long header Initial packet starting at `start` in a
|
||||
/// datagram, reading only the unprotected header fields.
|
||||
///
|
||||
/// Returns [`ParseError::NotQuicInitial`] when the bytes at `start` are
|
||||
/// not a long header Initial, and [`ParseError::UnsupportedQuicVersion`]
|
||||
/// when the packet is a long header of a QUIC version this code has no
|
||||
/// salt for. Both are routine on real traffic and feed counters rather
|
||||
/// than failing a capture.
|
||||
pub fn parse(datagram: &'pkt [u8], start: usize) -> Result<Self> {
|
||||
let buf = datagram.get(start..).ok_or(ParseError::NotQuicInitial)?;
|
||||
let first = *buf.first().ok_or(ParseError::NotQuicInitial)?;
|
||||
if first & LONG_HEADER_FORM == 0 {
|
||||
return Err(ParseError::NotQuicInitial);
|
||||
}
|
||||
|
||||
let version = u32::from_be_bytes(
|
||||
buf.get(1..5)
|
||||
.ok_or(ParseError::NotQuicInitial)?
|
||||
.try_into()
|
||||
.expect("a four byte slice is four bytes"),
|
||||
);
|
||||
if version != VERSION_1 {
|
||||
return Err(ParseError::UnsupportedQuicVersion(version));
|
||||
}
|
||||
if first & PACKET_TYPE_MASK != PACKET_TYPE_INITIAL {
|
||||
return Err(ParseError::NotQuicInitial);
|
||||
}
|
||||
|
||||
let mut pos = 5usize;
|
||||
let dcid = read_cid(buf, &mut pos)?;
|
||||
let _scid = read_cid(buf, &mut pos)?;
|
||||
|
||||
let (token_len, n) = read_varint(buf.get(pos..).ok_or(ParseError::NotQuicInitial)?)
|
||||
.map_err(|_| ParseError::NotQuicInitial)?;
|
||||
pos += n;
|
||||
let token_len = usize::try_from(token_len).map_err(|_| ParseError::NotQuicInitial)?;
|
||||
pos = pos
|
||||
.checked_add(token_len)
|
||||
.ok_or(ParseError::NotQuicInitial)?;
|
||||
|
||||
let (length, n) = read_varint(buf.get(pos..).ok_or(ParseError::NotQuicInitial)?)
|
||||
.map_err(|_| ParseError::NotQuicInitial)?;
|
||||
pos += n;
|
||||
let length = usize::try_from(length).map_err(|_| ParseError::NotQuicInitial)?;
|
||||
|
||||
let pn_offset = start + pos;
|
||||
let end = pn_offset
|
||||
.checked_add(length)
|
||||
.ok_or(ParseError::NotQuicInitial)?;
|
||||
if end > datagram.len() {
|
||||
return Err(ParseError::NotQuicInitial);
|
||||
}
|
||||
|
||||
Ok(Self {
|
||||
datagram,
|
||||
dcid,
|
||||
start,
|
||||
pn_offset,
|
||||
end,
|
||||
next_offset: end,
|
||||
})
|
||||
}
|
||||
|
||||
/// Removes header protection and decrypts the payload, returning the
|
||||
/// cleartext QUIC frames.
|
||||
///
|
||||
/// `largest_pn` is the largest packet number already decrypted on this
|
||||
/// flow, used to reconstruct the full packet number for the AEAD nonce.
|
||||
/// A failure to authenticate is reported as
|
||||
/// [`ParseError::QuicCryptoFailure`]; for a passive observer this most
|
||||
/// often means the packet was a server Initial, protected under keys the
|
||||
/// observer did not derive, rather than corruption.
|
||||
pub fn open(&self, keys: &InitialKeys, largest_pn: Option<u64>) -> Result<OpenedInitial> {
|
||||
let sample_start = self.pn_offset + PN_SAMPLE_OFFSET;
|
||||
let sample = self
|
||||
.datagram
|
||||
.get(sample_start..sample_start + SAMPLE_LEN)
|
||||
.ok_or(ParseError::QuicCryptoFailure)?;
|
||||
|
||||
let mask = header_protection_mask(&keys.hp, sample);
|
||||
|
||||
let protected_first = self.datagram[self.start];
|
||||
let first = protected_first ^ (mask[0] & LONG_HEADER_PN_MASK);
|
||||
let pn_len = usize::from(first & PN_LEN_MASK) + 1;
|
||||
|
||||
let pn_end = self.pn_offset + pn_len;
|
||||
if pn_end > self.end {
|
||||
return Err(ParseError::QuicCryptoFailure);
|
||||
}
|
||||
let mut pn_bytes = [0u8; 4];
|
||||
let mut truncated = 0u64;
|
||||
for i in 0..pn_len {
|
||||
let clear = self.datagram[self.pn_offset + i] ^ mask[1 + i];
|
||||
pn_bytes[i] = clear;
|
||||
truncated = (truncated << 8) | u64::from(clear);
|
||||
}
|
||||
let pn_nbits = u32::try_from(pn_len * 8).expect("the packet number is at most four bytes");
|
||||
let packet_number = decode_packet_number(largest_pn, truncated, pn_nbits);
|
||||
|
||||
let mut header: SmallVec<[u8; 64]> = SmallVec::new();
|
||||
header.push(first);
|
||||
header.extend_from_slice(&self.datagram[self.start + 1..self.pn_offset]);
|
||||
header.extend_from_slice(&pn_bytes[..pn_len]);
|
||||
|
||||
let ciphertext = self
|
||||
.datagram
|
||||
.get(pn_end..self.end)
|
||||
.ok_or(ParseError::QuicCryptoFailure)?;
|
||||
if ciphertext.len() < TAG_LEN {
|
||||
return Err(ParseError::QuicCryptoFailure);
|
||||
}
|
||||
let split = ciphertext.len() - TAG_LEN;
|
||||
let mut frames = ciphertext[..split].to_vec();
|
||||
let tag = GenericArray::from_slice(&ciphertext[split..]);
|
||||
|
||||
let nonce = aead_nonce(&keys.iv, packet_number);
|
||||
let cipher =
|
||||
Aes128Gcm::new_from_slice(&keys.key).map_err(|_| ParseError::QuicCryptoFailure)?;
|
||||
cipher
|
||||
.decrypt_in_place_detached(GenericArray::from_slice(&nonce), &header, &mut frames, tag)
|
||||
.map_err(|_| ParseError::QuicCryptoFailure)?;
|
||||
|
||||
Ok(OpenedInitial {
|
||||
packet_number,
|
||||
frames,
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
/// Reads a one byte length prefixed connection ID, advancing the cursor.
|
||||
fn read_cid<'pkt>(buf: &'pkt [u8], pos: &mut usize) -> Result<&'pkt [u8]> {
|
||||
let len = usize::from(*buf.get(*pos).ok_or(ParseError::NotQuicInitial)?);
|
||||
if len > MAX_CID_LEN {
|
||||
return Err(ParseError::NotQuicInitial);
|
||||
}
|
||||
let start = *pos + 1;
|
||||
let cid = buf
|
||||
.get(start..start + len)
|
||||
.ok_or(ParseError::NotQuicInitial)?;
|
||||
*pos = start + len;
|
||||
Ok(cid)
|
||||
}
|
||||
|
||||
/// Computes the five byte header protection mask, `AES-ECB(hp, sample)`.
|
||||
fn header_protection_mask(hp: &[u8; HP_LEN], sample: &[u8]) -> [u8; 5] {
|
||||
let cipher = Aes128::new_from_slice(hp).expect("the header protection key is sixteen bytes");
|
||||
let mut block = GenericArray::clone_from_slice(sample);
|
||||
cipher.encrypt_block(&mut block);
|
||||
let mut mask = [0u8; 5];
|
||||
mask.copy_from_slice(&block[..5]);
|
||||
mask
|
||||
}
|
||||
|
||||
/// Builds the AEAD nonce by XORing the packet number into the static IV, from
|
||||
/// RFC 9001 Section 5.3.
|
||||
fn aead_nonce(iv: &[u8; IV_LEN], packet_number: u64) -> [u8; IV_LEN] {
|
||||
let mut nonce = *iv;
|
||||
let pn = packet_number.to_be_bytes();
|
||||
for i in 0..8 {
|
||||
nonce[IV_LEN - 8 + i] ^= pn[i];
|
||||
}
|
||||
nonce
|
||||
}
|
||||
|
||||
/// Walks the frames in a decrypted Initial payload, handing each CRYPTO
|
||||
/// frame's offset and bytes to `on_crypto`.
|
||||
///
|
||||
/// Only the frame types RFC 9000 Section 12.4 permits in an Initial packet
|
||||
/// are recognized. An unrecognized type means the bytes are not a valid
|
||||
/// Initial payload after all, reported as [`ParseError::QuicCryptoFailure`];
|
||||
/// since the AEAD tag already authenticated these bytes, that points at a
|
||||
/// version or framing this code does not model rather than at an attacker.
|
||||
pub fn walk_crypto_frames(frames: &[u8], mut on_crypto: impl FnMut(u64, &[u8])) -> Result<()> {
|
||||
let mut pos = 0usize;
|
||||
while pos < frames.len() {
|
||||
let ty = frames[pos];
|
||||
pos += 1;
|
||||
match ty {
|
||||
frame::PADDING | frame::PING => {}
|
||||
frame::ACK | frame::ACK_ECN => {
|
||||
let (_largest, n) = read_varint(&frames[pos..])?;
|
||||
pos += n;
|
||||
let (_delay, n) = read_varint(&frames[pos..])?;
|
||||
pos += n;
|
||||
let (range_count, n) = read_varint(&frames[pos..])?;
|
||||
pos += n;
|
||||
let (_first_range, n) = read_varint(&frames[pos..])?;
|
||||
pos += n;
|
||||
for _ in 0..range_count {
|
||||
let (_gap, n) = read_varint(&frames[pos..])?;
|
||||
pos += n;
|
||||
let (_len, n) = read_varint(&frames[pos..])?;
|
||||
pos += n;
|
||||
}
|
||||
if ty == frame::ACK_ECN {
|
||||
for _ in 0..3 {
|
||||
let (_ecn, n) = read_varint(&frames[pos..])?;
|
||||
pos += n;
|
||||
}
|
||||
}
|
||||
}
|
||||
frame::CRYPTO => {
|
||||
let (offset, n) = read_varint(&frames[pos..])?;
|
||||
pos += n;
|
||||
let (len, n) = read_varint(&frames[pos..])?;
|
||||
pos += n;
|
||||
let len = usize::try_from(len).map_err(|_| ParseError::QuicCryptoFailure)?;
|
||||
let end = pos.checked_add(len).ok_or(ParseError::QuicCryptoFailure)?;
|
||||
let data = frames.get(pos..end).ok_or(ParseError::QuicCryptoFailure)?;
|
||||
on_crypto(offset, data);
|
||||
pos = end;
|
||||
}
|
||||
frame::CONNECTION_CLOSE => {
|
||||
let (_code, n) = read_varint(&frames[pos..])?;
|
||||
pos += n;
|
||||
let (_frame_type, n) = read_varint(&frames[pos..])?;
|
||||
pos += n;
|
||||
let (reason_len, n) = read_varint(&frames[pos..])?;
|
||||
pos += n;
|
||||
let reason_len =
|
||||
usize::try_from(reason_len).map_err(|_| ParseError::QuicCryptoFailure)?;
|
||||
pos = pos
|
||||
.checked_add(reason_len)
|
||||
.filter(|&p| p <= frames.len())
|
||||
.ok_or(ParseError::QuicCryptoFailure)?;
|
||||
}
|
||||
_ => return Err(ParseError::QuicCryptoFailure),
|
||||
}
|
||||
}
|
||||
Ok(())
|
||||
}
|
||||
|
||||
/// Reassembles the cleartext handshake stream that CRYPTO frames carry.
|
||||
///
|
||||
/// CRYPTO frames are the QUIC analogue of a TCP byte stream: each carries an
|
||||
/// absolute offset, frames can arrive out of order and span packets, and a
|
||||
/// ClientHello routinely splits across several. This is the QUIC counterpart
|
||||
/// of the TCP reassembler, kept separate because CRYPTO offsets are sixty
|
||||
/// four bit and start at zero per stream rather than at a negotiated sequence
|
||||
/// number. Contiguous bytes from offset zero accumulate in one buffer;
|
||||
/// everything ahead of the write cursor parks until the gap before it fills.
|
||||
/// Both buffers are capped so a hostile sender cannot turn the assembler into
|
||||
/// a memory bomb.
|
||||
pub struct CryptoAssembler {
|
||||
assembled: Vec<u8>,
|
||||
pending: BTreeMap<u64, Vec<u8>>,
|
||||
pending_bytes: usize,
|
||||
max_bytes: usize,
|
||||
overflowed: bool,
|
||||
}
|
||||
|
||||
impl CryptoAssembler {
|
||||
#[must_use]
|
||||
pub fn new(max_bytes: usize) -> Self {
|
||||
Self {
|
||||
assembled: Vec::new(),
|
||||
pending: BTreeMap::new(),
|
||||
pending_bytes: 0,
|
||||
max_bytes,
|
||||
overflowed: false,
|
||||
}
|
||||
}
|
||||
|
||||
/// Returns the contiguous handshake bytes assembled from offset zero.
|
||||
#[must_use]
|
||||
pub fn data(&self) -> &[u8] {
|
||||
&self.assembled
|
||||
}
|
||||
|
||||
/// Adds one CRYPTO frame's bytes at their stream offset.
|
||||
///
|
||||
/// Bytes wholly before the write cursor are duplicates and ignored; bytes
|
||||
/// straddling it extend the contiguous run and then pull in any parked
|
||||
/// segments the new bytes made contiguous; bytes wholly ahead of it park.
|
||||
/// Anything that would push either buffer past its cap is dropped and the
|
||||
/// assembler latches an overflow so the flow can be abandoned.
|
||||
pub fn push(&mut self, offset: u64, data: &[u8]) {
|
||||
if data.is_empty() || self.overflowed {
|
||||
return;
|
||||
}
|
||||
let Ok(cursor) = u64::try_from(self.assembled.len()) else {
|
||||
self.overflowed = true;
|
||||
return;
|
||||
};
|
||||
|
||||
if offset > cursor {
|
||||
self.park(offset, data);
|
||||
return;
|
||||
}
|
||||
let skip = usize::try_from(cursor - offset).unwrap_or(usize::MAX);
|
||||
if skip >= data.len() {
|
||||
return;
|
||||
}
|
||||
if !self.extend(&data[skip..]) {
|
||||
return;
|
||||
}
|
||||
self.drain_pending();
|
||||
}
|
||||
|
||||
fn extend(&mut self, bytes: &[u8]) -> bool {
|
||||
if self.assembled.len() + bytes.len() > self.max_bytes {
|
||||
self.overflowed = true;
|
||||
return false;
|
||||
}
|
||||
self.assembled.extend_from_slice(bytes);
|
||||
true
|
||||
}
|
||||
|
||||
fn park(&mut self, offset: u64, data: &[u8]) {
|
||||
if self.pending_bytes + data.len() > self.max_bytes {
|
||||
self.overflowed = true;
|
||||
return;
|
||||
}
|
||||
if let Some(existing) = self.pending.get(&offset) {
|
||||
if existing.len() >= data.len() {
|
||||
return;
|
||||
}
|
||||
self.pending_bytes -= existing.len();
|
||||
}
|
||||
self.pending_bytes += data.len();
|
||||
self.pending.insert(offset, data.to_vec());
|
||||
}
|
||||
|
||||
fn drain_pending(&mut self) {
|
||||
while let Some((&offset, _)) = self.pending.iter().next() {
|
||||
let Ok(cursor) = u64::try_from(self.assembled.len()) else {
|
||||
self.overflowed = true;
|
||||
return;
|
||||
};
|
||||
if offset > cursor {
|
||||
break;
|
||||
}
|
||||
let segment = self.pending.remove(&offset).expect("offset just observed");
|
||||
self.pending_bytes -= segment.len();
|
||||
let skip = usize::try_from(cursor - offset).unwrap_or(usize::MAX);
|
||||
if skip < segment.len() && !self.extend(&segment[skip..]) {
|
||||
return;
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/// Returns the body of the leading ClientHello once it is fully present.
|
||||
///
|
||||
/// The QUIC Initial cryptographic stream begins with the ClientHello
|
||||
/// handshake message: a one byte type, a three byte length, then the
|
||||
/// body. This reports the body once the contiguous bytes hold all of it,
|
||||
/// reports that the stream does not begin with a ClientHello, or reports
|
||||
/// that more bytes are still needed.
|
||||
#[must_use]
|
||||
pub fn client_hello(&self) -> ClientHelloState<'_> {
|
||||
if self.assembled.len() < 4 {
|
||||
return if self.overflowed {
|
||||
ClientHelloState::Abandoned
|
||||
} else {
|
||||
ClientHelloState::Incomplete
|
||||
};
|
||||
}
|
||||
if self.assembled[0] != handshake_type::CLIENT_HELLO {
|
||||
return ClientHelloState::NotClientHello;
|
||||
}
|
||||
let len = usize::from(self.assembled[1]) << 16
|
||||
| usize::from(self.assembled[2]) << 8
|
||||
| usize::from(self.assembled[3]);
|
||||
match self.assembled.get(4..4 + len) {
|
||||
Some(body) => ClientHelloState::Ready(body),
|
||||
None if self.overflowed => ClientHelloState::Abandoned,
|
||||
None => ClientHelloState::Incomplete,
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/// What [`CryptoAssembler::client_hello`] found in the assembled bytes.
|
||||
#[derive(Debug, PartialEq, Eq)]
|
||||
pub enum ClientHelloState<'a> {
|
||||
/// The full ClientHello body is present.
|
||||
Ready(&'a [u8]),
|
||||
/// More CRYPTO bytes are needed before the ClientHello is complete.
|
||||
Incomplete,
|
||||
/// The stream does not begin with a ClientHello and never will.
|
||||
NotClientHello,
|
||||
/// A buffer cap was hit; the flow should be given up.
|
||||
Abandoned,
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
mod tests {
|
||||
use super::{
|
||||
ClientHelloState, CryptoAssembler, InitialKeys, InitialPacket, decode_packet_number,
|
||||
hkdf_expand_label, read_varint, walk_crypto_frames,
|
||||
};
|
||||
|
||||
/// RFC 9000 Appendix A.1 sample variable length integer decodings.
|
||||
#[test]
|
||||
fn varint_matches_rfc9000_appendix_a1() {
|
||||
assert_eq!(
|
||||
read_varint(&hex("c2197c5eff14e88c")).unwrap(),
|
||||
(151_288_809_941_952_652, 8)
|
||||
);
|
||||
assert_eq!(read_varint(&hex("9d7f3e7d")).unwrap(), (494_878_333, 4));
|
||||
assert_eq!(read_varint(&hex("7bbd")).unwrap(), (15_293, 2));
|
||||
assert_eq!(read_varint(&hex("25")).unwrap(), (37, 1));
|
||||
assert_eq!(read_varint(&hex("4025")).unwrap(), (37, 2));
|
||||
}
|
||||
|
||||
/// RFC 9000 Appendix A.3 sample packet number decoding.
|
||||
#[test]
|
||||
fn packet_number_decode_matches_rfc9000_appendix_a3() {
|
||||
assert_eq!(
|
||||
decode_packet_number(Some(0xa82f_30ea), 0x9b32, 16),
|
||||
0xa82f_9b32
|
||||
);
|
||||
assert_eq!(decode_packet_number(None, 2, 8), 2);
|
||||
}
|
||||
|
||||
/// RFC 9001 Appendix A.1 derives a known set of client Initial keys from
|
||||
/// the sample Destination Connection ID.
|
||||
#[test]
|
||||
fn client_initial_keys_match_rfc9001_appendix_a1() {
|
||||
let dcid = hex("8394c8f03e515708");
|
||||
let keys = InitialKeys::client(&dcid);
|
||||
assert_eq!(keys.key.to_vec(), hex("1f369613dd76d5467730efcbe3b1a22d"));
|
||||
assert_eq!(keys.iv.to_vec(), hex("fa044b2f42a3fd3b46fb255c"));
|
||||
assert_eq!(keys.hp.to_vec(), hex("9f50449e04a0e810283a1e9933adedd2"));
|
||||
}
|
||||
|
||||
/// RFC 9001 Appendix A.1 also pins the intermediate expand label output.
|
||||
#[test]
|
||||
fn expand_label_reproduces_rfc9001_client_secret() {
|
||||
let dcid = hex("8394c8f03e515708");
|
||||
let (initial_secret, _) =
|
||||
hkdf::Hkdf::<sha2::Sha256>::extract(Some(&super::INITIAL_SALT_V1), &dcid);
|
||||
let mut client_secret = [0u8; 32];
|
||||
hkdf_expand_label(&initial_secret, b"client in", &mut client_secret);
|
||||
assert_eq!(
|
||||
client_secret.to_vec(),
|
||||
hex("c00cf151ca5be075ed0ebfb5c80323c42d6b7db67881289af4008f1f6c357aea")
|
||||
);
|
||||
}
|
||||
|
||||
/// RFC 9001 Appendix A.2 carries the full protected client Initial. Its
|
||||
/// decryption must recover the documented CRYPTO frame and a parseable
|
||||
/// ClientHello.
|
||||
#[test]
|
||||
fn opens_rfc9001_appendix_a2_client_initial() {
|
||||
let datagram = rfc9001_a2_protected_packet();
|
||||
let packet = InitialPacket::parse(&datagram, 0).unwrap();
|
||||
assert_eq!(packet.dcid, &hex("8394c8f03e515708")[..]);
|
||||
|
||||
let keys = InitialKeys::client(packet.dcid);
|
||||
let opened = packet.open(&keys, None).unwrap();
|
||||
assert_eq!(opened.packet_number, 2);
|
||||
|
||||
let expected_head = hex("060040f1010000ed0303ebf8fa56f12939b9584a3896472ec40bb863cfd3e868");
|
||||
assert_eq!(&opened.frames[..expected_head.len()], &expected_head[..]);
|
||||
|
||||
let mut assembler = CryptoAssembler::new(1 << 16);
|
||||
walk_crypto_frames(&opened.frames, |offset, data| assembler.push(offset, data)).unwrap();
|
||||
let ClientHelloState::Ready(body) = assembler.client_hello() else {
|
||||
panic!("expected a complete ClientHello");
|
||||
};
|
||||
let hello = crate::parse::parse_client_hello(body).unwrap();
|
||||
assert!(!hello.cipher_suites.is_empty());
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn assembler_orders_out_of_order_crypto_frames() {
|
||||
let mut a = CryptoAssembler::new(1 << 16);
|
||||
a.push(5, b"world");
|
||||
assert_eq!(a.client_hello(), ClientHelloState::Incomplete);
|
||||
a.push(0, b"hello");
|
||||
assert_eq!(a.data(), b"helloworld");
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn assembler_ignores_pure_duplicates_and_overlap() {
|
||||
let mut a = CryptoAssembler::new(1 << 16);
|
||||
a.push(0, b"hello");
|
||||
a.push(0, b"hel");
|
||||
a.push(2, b"llo world");
|
||||
assert_eq!(a.data(), b"hello world");
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn assembler_latches_overflow_past_the_cap() {
|
||||
let mut a = CryptoAssembler::new(8);
|
||||
a.push(0, b"12345678");
|
||||
a.push(8, b"9");
|
||||
assert_eq!(a.data(), b"12345678");
|
||||
assert_eq!(a.client_hello(), ClientHelloState::NotClientHello);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn non_quic_bytes_are_not_an_initial() {
|
||||
let udp = b"GET / HTTP/1.1\r\n";
|
||||
assert!(InitialPacket::parse(udp, 0).is_err());
|
||||
}
|
||||
|
||||
fn hex(s: &str) -> Vec<u8> {
|
||||
::hex::decode(s).unwrap()
|
||||
}
|
||||
|
||||
/// The protected client Initial packet from RFC 9001 Appendix A.2.
|
||||
fn rfc9001_a2_protected_packet() -> Vec<u8> {
|
||||
hex(concat!(
|
||||
"c000000001088394c8f03e5157080000449e7b9aec34d1b1c98dd7689fb8ec11",
|
||||
"d242b123dc9bd8bab936b47d92ec356c0bab7df5976d27cd449f63300099f399",
|
||||
"1c260ec4c60d17b31f8429157bb35a1282a643a8d2262cad67500cadb8e7378c",
|
||||
"8eb7539ec4d4905fed1bee1fc8aafba17c750e2c7ace01e6005f80fcb7df6212",
|
||||
"30c83711b39343fa028cea7f7fb5ff89eac2308249a02252155e2347b63d58c5",
|
||||
"457afd84d05dfffdb20392844ae812154682e9cf012f9021a6f0be17ddd0c208",
|
||||
"4dce25ff9b06cde535d0f920a2db1bf362c23e596d11a4f5a6cf3948838a3aec",
|
||||
"4e15daf8500a6ef69ec4e3feb6b1d98e610ac8b7ec3faf6ad760b7bad1db4ba3",
|
||||
"485e8a94dc250ae3fdb41ed15fb6a8e5eba0fc3dd60bc8e30c5c4287e53805db",
|
||||
"059ae0648db2f64264ed5e39be2e20d82df566da8dd5998ccabdae053060ae6c",
|
||||
"7b4378e846d29f37ed7b4ea9ec5d82e7961b7f25a9323851f681d582363aa5f8",
|
||||
"9937f5a67258bf63ad6f1a0b1d96dbd4faddfcefc5266ba6611722395c906556",
|
||||
"be52afe3f565636ad1b17d508b73d8743eeb524be22b3dcbc2c7468d54119c74",
|
||||
"68449a13d8e3b95811a198f3491de3e7fe942b330407abf82a4ed7c1b311663a",
|
||||
"c69890f4157015853d91e923037c227a33cdd5ec281ca3f79c44546b9d90ca00",
|
||||
"f064c99e3dd97911d39fe9c5d0b23a229a234cb36186c4819e8b9c5927726632",
|
||||
"291d6a418211cc2962e20fe47feb3edf330f2c603a9d48c0fcb5699dbfe58964",
|
||||
"25c5bac4aee82e57a85aaf4e2513e4f05796b07ba2ee47d80506f8d2c25e50fd",
|
||||
"14de71e6c418559302f939b0e1abd576f279c4b2e0feb85c1f28ff18f58891ff",
|
||||
"ef132eef2fa09346aee33c28eb130ff28f5b766953334113211996d20011a198",
|
||||
"e3fc433f9f2541010ae17c1bf202580f6047472fb36857fe843b19f5984009dd",
|
||||
"c324044e847a4f4a0ab34f719595de37252d6235365e9b84392b061085349d73",
|
||||
"203a4a13e96f5432ec0fd4a1ee65accdd5e3904df54c1da510b0ff20dcc0c77f",
|
||||
"cb2c0e0eb605cb0504db87632cf3d8b4dae6e705769d1de354270123cb11450e",
|
||||
"fc60ac47683d7b8d0f811365565fd98c4c8eb936bcab8d069fc33bd801b03ade",
|
||||
"a2e1fbc5aa463d08ca19896d2bf59a071b851e6c239052172f296bfb5e724047",
|
||||
"90a2181014f3b94a4e97d117b438130368cc39dbb2d198065ae3986547926cd2",
|
||||
"162f40a29f0c3c8745c0f50fba3852e566d44575c29d39a03f0cda721984b6f4",
|
||||
"40591f355e12d439ff150aab7613499dbd49adabc8676eef023b15b65bfc5ca0",
|
||||
"6948109f23f350db82123535eb8a7433bdabcb909271a6ecbcb58b936a88cd4e",
|
||||
"8f2e6ff5800175f113253d8fa9ca8885c2f552e657dc603f252e1a8e308f76f0",
|
||||
"be79e2fb8f5d5fbbe2e30ecadd220723c8c0aea8078cdfcb3868263ff8f09400",
|
||||
"54da48781893a7e49ad5aff4af300cd804a6b6279ab3ff3afb64491c85194aab",
|
||||
"760d58a606654f9f4400e8b38591356fbf6425aca26dc85244259ff2b19c41b9",
|
||||
"f96f3ca9ec1dde434da7d2d392b905ddf3d1f9af93d1af5950bd493f5aa731b4",
|
||||
"056df31bd267b6b90a079831aaf579be0a39013137aac6d404f518cfd4684064",
|
||||
"7e78bfe706ca4cf5e9c5453e9f7cfd2b8b4c8d169a44e55c88d4a9a7f9474241",
|
||||
"e221af44860018ab0856972e194cd934",
|
||||
))
|
||||
}
|
||||
}
|
||||
|
|
@ -46,6 +46,17 @@ fn client_hellos(events: &[FingerprintEvent]) -> Vec<(&str, &str)> {
|
|||
.collect()
|
||||
}
|
||||
|
||||
/// Returns the ClientHello events whose JA4 transport marker is QUIC.
|
||||
fn quic_client_hellos(events: &[FingerprintEvent]) -> Vec<&FingerprintEvent> {
|
||||
events
|
||||
.iter()
|
||||
.filter(|e| match &e.event {
|
||||
StreamEvent::ClientHello { ja4, .. } => ja4.hash.starts_with('q'),
|
||||
_ => false,
|
||||
})
|
||||
.collect()
|
||||
}
|
||||
|
||||
fn ja4s_hashes(events: &[FingerprintEvent]) -> Vec<(&str, &str)> {
|
||||
events
|
||||
.iter()
|
||||
|
|
@ -112,6 +123,46 @@ fn chrome_cloudflare_tcp_handshake_reproduces_published_ja4() {
|
|||
);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn chrome_cloudflare_quic_initial_decrypts_to_published_q_ja4() {
|
||||
// The same capture carries the same browser over both transports. The TCP
|
||||
// ClientHello above and this QUIC one share a JA4_a prefix family but
|
||||
// differ where the transport and ALPN do, which is the whole reason JA4
|
||||
// records the transport. The QUIC value below was reproduced from the
|
||||
// decrypted Initial by an independent decryptor and recomputed from its
|
||||
// raw string with a shell digest before being pinned.
|
||||
let events = fingerprint("chrome-cloudflare-quic-with-secrets.pcapng");
|
||||
|
||||
let quic = quic_client_hellos(&events);
|
||||
assert_eq!(quic.len(), 1, "expected exactly one QUIC ClientHello");
|
||||
let StreamEvent::ClientHello { ja4, sni, alpn, .. } = &quic[0].event else {
|
||||
unreachable!("filtered to ClientHello");
|
||||
};
|
||||
assert_eq!(ja4.hash, "q13d0310h3_55b375c5d22e_cd85d2d88918");
|
||||
assert_eq!(sni.as_deref(), Some("cloudflare-quic.com"));
|
||||
assert_eq!(alpn.as_deref(), Some("h3"));
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn quic_several_tls_frames_reassembles_split_client_hello() {
|
||||
// This capture's ClientHello is delivered across several CRYPTO frames in
|
||||
// one Initial, exercising the offset based reassembler. It lands on the
|
||||
// identical JA4 as the chrome capture despite being a different client,
|
||||
// because JA4 sorts the cipher and extension lists before hashing.
|
||||
let events = fingerprint("quic-with-several-tls-frames.pcapng");
|
||||
|
||||
let quic = quic_client_hellos(&events);
|
||||
assert_eq!(quic.len(), 1);
|
||||
let StreamEvent::ClientHello { ja4, .. } = &quic[0].event else {
|
||||
unreachable!("filtered to ClientHello");
|
||||
};
|
||||
assert_eq!(ja4.hash, "q13d0310h3_55b375c5d22e_cd85d2d88918");
|
||||
assert_eq!(
|
||||
ja4.raw,
|
||||
"q13d0310h3_1301,1302,1303_000a,000d,001b,002b,002d,0033,0039,4469_0403,0804,0401,0503,0805,0501,0806,0601,0201"
|
||||
);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn tls_handshake_reproduces_published_ja4s_raw_and_hash() {
|
||||
let events = fingerprint("tls-handshake.pcapng");
|
||||
|
|
|
|||
|
|
@ -36,3 +36,6 @@ serde.workspace = true
|
|||
serde_json.workspace = true
|
||||
clap.workspace = true
|
||||
etherparse.workspace = true
|
||||
smallvec.workspace = true
|
||||
thiserror.workspace = true
|
||||
rustix.workspace = true
|
||||
|
|
|
|||
|
|
@ -5,7 +5,9 @@ use anyhow::{Context, Result};
|
|||
use clap::{Parser, Subcommand};
|
||||
use tracing_subscriber::EnvFilter;
|
||||
|
||||
use tlsfp_core::{PcapFileSource, Pipeline, PipelineConfig};
|
||||
use tlsfp_core::{FingerprintEvent, PcapFileSource, Pipeline, PipelineConfig, SourceError};
|
||||
|
||||
use crate::live::{DEFAULT_BPF_FILTER, LiveConfig, LiveSource};
|
||||
|
||||
/// JA3/JA4 TLS fingerprinting tool.
|
||||
///
|
||||
|
|
@ -36,9 +38,44 @@ pub enum Command {
|
|||
},
|
||||
|
||||
/// Capture live from a network interface and fingerprint in real time.
|
||||
///
|
||||
/// Live capture opens a raw socket, which an unprivileged user cannot
|
||||
/// do. Rather than running the whole tool as root, grant the binary the
|
||||
/// two capabilities libpcap needs:
|
||||
///
|
||||
/// sudo setcap cap_net_raw,cap_net_admin=eip "$(command -v tlsfp)"
|
||||
///
|
||||
/// cap_net_raw opens the capture socket, cap_net_admin covers
|
||||
/// promiscuous mode, and =eip marks both permitted, inheritable, and
|
||||
/// effective when the binary runs. File capabilities live on the binary
|
||||
/// itself, so repeat the grant after every rebuild, and point it at
|
||||
/// target/debug/tlsfp or target/release/tlsfp when running from cargo.
|
||||
///
|
||||
/// The default --filter keeps all TCP (TLS lives on any port, JA4T
|
||||
/// wants the SYNs) plus UDP 443 for QUIC, and drops everything else in
|
||||
/// the kernel. Tighten it on busy links, for example:
|
||||
///
|
||||
/// tlsfp live eth0 --filter "tcp port 443"
|
||||
///
|
||||
/// Stop with ctrl-c: the first one drains and prints final counters, a
|
||||
/// second one exits immediately.
|
||||
#[command(verbatim_doc_comment)]
|
||||
Live {
|
||||
/// Interface name, for example eth0.
|
||||
/// Interface name, for example eth0, or any for every interface.
|
||||
interface: String,
|
||||
|
||||
/// Emit one JSON object per event instead of readable lines.
|
||||
#[arg(long)]
|
||||
json: bool,
|
||||
|
||||
/// BPF filter compiled into the kernel before capture begins.
|
||||
#[arg(long, default_value = DEFAULT_BPF_FILTER)]
|
||||
filter: String,
|
||||
|
||||
/// Capture only traffic the interface would normally receive
|
||||
/// instead of switching it to promiscuous mode.
|
||||
#[arg(long)]
|
||||
no_promisc: bool,
|
||||
},
|
||||
|
||||
/// Serve the web dashboard and HTTP API.
|
||||
|
|
@ -57,15 +94,21 @@ impl Cli {
|
|||
_ => "tlsfp=trace",
|
||||
};
|
||||
let filter = EnvFilter::try_from_default_env().unwrap_or_else(|_| EnvFilter::new(default));
|
||||
tracing_subscriber::fmt().with_env_filter(filter).init();
|
||||
tracing_subscriber::fmt()
|
||||
.with_env_filter(filter)
|
||||
.with_writer(std::io::stderr)
|
||||
.init();
|
||||
}
|
||||
|
||||
pub fn run(self) -> Result<()> {
|
||||
match self.command {
|
||||
Command::Pcap { path, json } => run_pcap(&path, json),
|
||||
Command::Live { interface } => {
|
||||
anyhow::bail!("live capture on {interface} is not wired up yet")
|
||||
}
|
||||
Command::Live {
|
||||
interface,
|
||||
json,
|
||||
filter,
|
||||
no_promisc,
|
||||
} => run_live(&interface, json, filter, !no_promisc),
|
||||
Command::Serve { bind } => {
|
||||
anyhow::bail!("dashboard on {bind} is not wired up yet")
|
||||
}
|
||||
|
|
@ -113,8 +156,12 @@ fn run_pcap(path: &std::path::Path, json: bool) -> Result<()> {
|
|||
tracing::info!(
|
||||
frames = counters.frames,
|
||||
tcp_segments = counters.tcp_segments,
|
||||
udp_datagrams = counters.udp_datagrams,
|
||||
events = counters.events,
|
||||
flows = counters.flows_created,
|
||||
quic_initials = counters.quic_initials,
|
||||
quic_decrypted = counters.quic_decrypted,
|
||||
quic_version_unsupported = counters.quic_version_unsupported,
|
||||
unfinished_tls_streams = counters.unfinished_tls_streams,
|
||||
segments_dropped = counters.segments_dropped,
|
||||
"capture processed"
|
||||
|
|
@ -124,3 +171,127 @@ fn run_pcap(path: &std::path::Path, json: bool) -> Result<()> {
|
|||
}
|
||||
Ok(())
|
||||
}
|
||||
|
||||
/// Captures from an interface until ctrl-c and fingerprints in real time.
|
||||
///
|
||||
/// The capture itself runs on a dedicated OS thread inside [`LiveSource`];
|
||||
/// this function owns the tokio side of the bridge. The runtime is built
|
||||
/// here rather than in main so the file path stays a plain synchronous
|
||||
/// program.
|
||||
fn run_live(interface: &str, json: bool, filter: String, promiscuous: bool) -> Result<()> {
|
||||
let config = LiveConfig {
|
||||
filter,
|
||||
promiscuous,
|
||||
};
|
||||
let source = LiveSource::open(interface, &config)?;
|
||||
tracing::info!(interface, filter = %config.filter, "live capture started");
|
||||
|
||||
let runtime = tokio::runtime::Builder::new_multi_thread()
|
||||
.enable_all()
|
||||
.build()
|
||||
.context("building the async runtime")?;
|
||||
runtime.block_on(drive_live(source, json))
|
||||
}
|
||||
|
||||
/// Drains the live source through the same pipeline the file path uses.
|
||||
///
|
||||
/// Events flush per line so the stream is followable as it happens. The
|
||||
/// first ctrl-c asks the capture thread to stop and lets the channel
|
||||
/// drain, which makes the final counters trustworthy; a second ctrl-c
|
||||
/// exits without ceremony. A closed stdout pipe is a normal way for a
|
||||
/// live session to end, so it stops the capture instead of reporting an
|
||||
/// error.
|
||||
async fn drive_live(mut source: LiveSource, json: bool) -> Result<()> {
|
||||
let stop = source.stop_handle();
|
||||
tokio::spawn(async move {
|
||||
if tokio::signal::ctrl_c().await.is_ok() {
|
||||
tracing::info!("ctrl-c received; draining capture");
|
||||
stop.stop();
|
||||
}
|
||||
if tokio::signal::ctrl_c().await.is_ok() {
|
||||
std::process::exit(130);
|
||||
}
|
||||
});
|
||||
|
||||
let mut pipeline = Pipeline::new(PipelineConfig::default());
|
||||
let stdout = std::io::stdout().lock();
|
||||
let mut out = std::io::BufWriter::new(stdout);
|
||||
let mut write_failure: Option<std::io::Error> = None;
|
||||
|
||||
let capture_failure: Option<SourceError> = loop {
|
||||
let received = source.next_frame_async().await;
|
||||
let frame = match received {
|
||||
Ok(Some(frame)) => frame,
|
||||
Ok(None) => break None,
|
||||
Err(error) => break Some(error),
|
||||
};
|
||||
pipeline.feed(&frame, &mut |event| {
|
||||
write_live_event(&mut out, &event, json, &mut write_failure);
|
||||
});
|
||||
if write_failure.is_some() {
|
||||
break None;
|
||||
}
|
||||
};
|
||||
pipeline.finish();
|
||||
|
||||
let counters = pipeline.counters();
|
||||
tracing::info!(
|
||||
frames = counters.frames,
|
||||
tcp_segments = counters.tcp_segments,
|
||||
udp_datagrams = counters.udp_datagrams,
|
||||
events = counters.events,
|
||||
flows = counters.flows_created,
|
||||
quic_initials = counters.quic_initials,
|
||||
quic_decrypted = counters.quic_decrypted,
|
||||
quic_version_unsupported = counters.quic_version_unsupported,
|
||||
unfinished_tls_streams = counters.unfinished_tls_streams,
|
||||
segments_dropped = counters.segments_dropped,
|
||||
"live capture stopped"
|
||||
);
|
||||
|
||||
if let Some(stats) = source.close() {
|
||||
tracing::info!(
|
||||
received = stats.received,
|
||||
kernel_dropped = stats.kernel_dropped,
|
||||
interface_dropped = stats.interface_dropped,
|
||||
"kernel capture statistics"
|
||||
);
|
||||
if stats.kernel_dropped > 0 || stats.interface_dropped > 0 {
|
||||
tracing::warn!(
|
||||
"the kernel dropped frames; an absent fingerprint is not proof of an absent handshake"
|
||||
);
|
||||
}
|
||||
}
|
||||
|
||||
match (write_failure, capture_failure) {
|
||||
(Some(error), _) if error.kind() == std::io::ErrorKind::BrokenPipe => Ok(()),
|
||||
(Some(error), _) => Err(anyhow::Error::from(error).context("writing events to stdout")),
|
||||
(None, Some(error)) => Err(anyhow::Error::from(error).context("live capture failed")),
|
||||
(None, None) => Ok(()),
|
||||
}
|
||||
}
|
||||
|
||||
/// Writes one event and flushes it immediately, recording the first
|
||||
/// failure instead of panicking inside the pipeline's sink. Later calls
|
||||
/// become no-ops once a write has failed.
|
||||
fn write_live_event(
|
||||
out: &mut impl std::io::Write,
|
||||
event: &FingerprintEvent,
|
||||
json: bool,
|
||||
failure: &mut Option<std::io::Error>,
|
||||
) {
|
||||
if failure.is_some() {
|
||||
return;
|
||||
}
|
||||
let result = if json {
|
||||
serde_json::to_writer(&mut *out, event)
|
||||
.map_err(std::io::Error::from)
|
||||
.and_then(|()| writeln!(out))
|
||||
.and_then(|()| out.flush())
|
||||
} else {
|
||||
writeln!(out, "{event}").and_then(|()| out.flush())
|
||||
};
|
||||
if let Err(error) = result {
|
||||
*failure = Some(error);
|
||||
}
|
||||
}
|
||||
|
|
|
|||
|
|
@ -0,0 +1,562 @@
|
|||
// ©AngelaMos | 2026
|
||||
// live.rs
|
||||
|
||||
//! Live capture behind the same [`PacketSource`] seam the file path uses.
|
||||
//!
|
||||
//! libpcap blocks, tokio must not, and the two meet here. A dedicated OS
|
||||
//! thread owns the activated capture handle and does nothing but pull
|
||||
//! frames and push them into a bounded channel. A `std::thread` rather
|
||||
//! than `spawn_blocking` because an indefinite capture loop parked on the
|
||||
//! blocking pool would pin one of its slots forever. The channel is
|
||||
//! bounded so a slow consumer backs pressure up into the kernel ring
|
||||
//! buffer, where overload becomes a counted drop in [`CaptureStats`]
|
||||
//! instead of unbounded memory growth here. The consumer side is
|
||||
//! [`LiveSource`]: a synchronous [`PacketSource`] for anything that wants
|
||||
//! to block, and [`LiveSource::next_frame_async`] for the tokio side of
|
||||
//! the bridge. Both feed the exact pipeline the pcap file path uses.
|
||||
//!
|
||||
//! The capture handle runs non-blocking and the thread waits on `poll`
|
||||
//! with a bounded timeout, not on libpcap's own read timeout. That read
|
||||
//! timeout cannot be relied on to bound a read: on a silent interface
|
||||
//! several platforms never start its timer, so a `next_packet` call would
|
||||
//! park until the next frame arrives, which on an idle link can be never.
|
||||
//! Owning the wait through `poll` is what lets the loop notice a stop
|
||||
//! request promptly no matter how quiet the wire is.
|
||||
|
||||
use std::io::ErrorKind;
|
||||
use std::sync::Arc;
|
||||
use std::sync::OnceLock;
|
||||
use std::sync::atomic::{AtomicBool, Ordering};
|
||||
use std::thread::JoinHandle;
|
||||
|
||||
use pcap::{Active, Capture, Device};
|
||||
use rustix::event::{PollFd, PollFlags, Timespec, poll};
|
||||
use smallvec::SmallVec;
|
||||
use thiserror::Error;
|
||||
|
||||
use tlsfp_core::{PacketSource, RawFrame, SourceError};
|
||||
|
||||
/// Kernel side prefilter applied before frames ever reach userspace.
|
||||
///
|
||||
/// Every TCP segment stays: TLS runs on any port, JA4T wants the SYNs, and
|
||||
/// JA4H wants cleartext HTTP wherever it appears. UDP narrows to 443,
|
||||
/// where QUIC lives. Everything else, ARP, ICMP, DNS, mDNS, DHCP, drops in
|
||||
/// the kernel for the cost of a BPF program, not a context switch.
|
||||
pub const DEFAULT_BPF_FILTER: &str = "tcp or (udp and port 443)";
|
||||
|
||||
/// Full frames, never clipped. A truncated segment would punch a hole in
|
||||
/// TCP reassembly and silently end fingerprinting for that stream, and
|
||||
/// offloads like GRO can hand the capture socket frames far beyond the
|
||||
/// wire MTU.
|
||||
const SNAPLEN: i32 = 65_535;
|
||||
|
||||
/// How long the capture thread waits in `poll` before looping back to
|
||||
/// re-check the stop flag. This is the upper bound on shutdown latency on
|
||||
/// an idle interface, and it is paid only as latency, never as busy work.
|
||||
const POLL_TIMEOUT_MILLIS: i64 = 100;
|
||||
|
||||
/// Kernel capture buffer. This is the shock absorber while the channel is
|
||||
/// full: bursts queue here, and only when it overflows does the kernel
|
||||
/// drop, visibly, into [`CaptureStats::kernel_dropped`].
|
||||
const KERNEL_BUFFER_BYTES: i32 = 4 * 1024 * 1024;
|
||||
|
||||
/// Frames in flight between the capture thread and the consumer. Bounded
|
||||
/// so the bridge applies backpressure instead of growing without limit.
|
||||
const CHANNEL_CAPACITY: usize = 512;
|
||||
|
||||
/// Frames at or under this size live inline in the channel message, no
|
||||
/// heap allocation per frame. Sized to cover an ethernet frame at the
|
||||
/// usual 1500 MTU plus VLAN tags; offload jumbos spill to the heap.
|
||||
const INLINE_FRAME_BYTES: usize = 2048;
|
||||
|
||||
const NANOS_PER_SECOND: u64 = 1_000_000_000;
|
||||
const NANOS_PER_MILLI: i64 = 1_000_000;
|
||||
const NANOS_PER_MICRO: u64 = 1_000;
|
||||
|
||||
/// Errors that prevent a live capture from starting.
|
||||
#[derive(Debug, Error)]
|
||||
pub enum LiveError {
|
||||
#[error("{0}")]
|
||||
Open(String),
|
||||
|
||||
#[error("invalid BPF filter {filter:?}: {source}")]
|
||||
Filter { filter: String, source: pcap::Error },
|
||||
|
||||
#[error("failed to start the capture thread: {0}")]
|
||||
Spawn(std::io::Error),
|
||||
}
|
||||
|
||||
/// Knobs for a live capture session.
|
||||
#[derive(Debug, Clone)]
|
||||
pub struct LiveConfig {
|
||||
/// BPF program text compiled into the kernel before capture begins.
|
||||
pub filter: String,
|
||||
/// Whether to ask the interface for traffic beyond its own addresses.
|
||||
pub promiscuous: bool,
|
||||
}
|
||||
|
||||
impl Default for LiveConfig {
|
||||
fn default() -> Self {
|
||||
Self {
|
||||
filter: DEFAULT_BPF_FILTER.to_owned(),
|
||||
promiscuous: true,
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/// Final tallies from the kernel side of a finished capture.
|
||||
///
|
||||
/// `kernel_dropped` is the honesty number: frames the kernel discarded
|
||||
/// because the consumer fell behind. When it is nonzero, an absence of
|
||||
/// fingerprints is not evidence of an absence of handshakes.
|
||||
#[derive(Debug, Clone, Copy, Default)]
|
||||
pub struct CaptureStats {
|
||||
pub received: u64,
|
||||
pub kernel_dropped: u64,
|
||||
pub interface_dropped: u64,
|
||||
}
|
||||
|
||||
/// Asks the capture thread to stop, from any thread or task.
|
||||
///
|
||||
/// The request is honored within one poll timeout: the capture loop
|
||||
/// checks the flag between packets, and an idle interface wakes it every
|
||||
/// [`POLL_TIMEOUT_MILLIS`]. Relaxed ordering is enough because the flag
|
||||
/// carries no data of its own: every captured frame crosses the channel
|
||||
/// and every error crosses the `OnceLock`, each of which synchronizes
|
||||
/// itself, and the join in [`LiveSource::close`] is the final barrier.
|
||||
#[derive(Debug, Clone)]
|
||||
pub struct StopHandle(Arc<AtomicBool>);
|
||||
|
||||
impl StopHandle {
|
||||
pub fn stop(&self) {
|
||||
self.0.store(true, Ordering::Relaxed);
|
||||
}
|
||||
}
|
||||
|
||||
/// One frame copied out of the kernel buffer, owned so it can cross the
|
||||
/// thread boundary.
|
||||
struct CapturedFrame {
|
||||
ts_nanos: u64,
|
||||
data: SmallVec<[u8; INLINE_FRAME_BYTES]>,
|
||||
}
|
||||
|
||||
/// Frames from a live interface, behind the [`PacketSource`] seam.
|
||||
///
|
||||
/// Construction activates the interface, installs the BPF filter, and
|
||||
/// spawns the capture thread. From then on this type is the receiving end
|
||||
/// of the bridge: [`PacketSource::next_frame`] blocks for synchronous
|
||||
/// consumers, [`LiveSource::next_frame_async`] suspends for tokio ones,
|
||||
/// and both hand out frames borrowing from an internal staging buffer
|
||||
/// exactly like [`tlsfp_core::PcapFileSource`] does.
|
||||
pub struct LiveSource {
|
||||
receiver: flume::Receiver<CapturedFrame>,
|
||||
staged: CapturedFrame,
|
||||
link_type: i32,
|
||||
stop: Arc<AtomicBool>,
|
||||
failure: Arc<OnceLock<String>>,
|
||||
failure_reported: bool,
|
||||
thread: Option<JoinHandle<Option<CaptureStats>>>,
|
||||
}
|
||||
|
||||
impl LiveSource {
|
||||
/// Opens an interface by name and starts capturing immediately.
|
||||
pub fn open(interface: &str, config: &LiveConfig) -> Result<Self, LiveError> {
|
||||
let inactive = Capture::from_device(interface)
|
||||
.map_err(|error| open_error(interface, &error))?
|
||||
.promisc(config.promiscuous)
|
||||
.snaplen(SNAPLEN)
|
||||
.immediate_mode(true)
|
||||
.buffer_size(KERNEL_BUFFER_BYTES);
|
||||
let mut capture = inactive
|
||||
.open()
|
||||
.map_err(|error| open_error(interface, &error))?;
|
||||
capture
|
||||
.filter(&config.filter, true)
|
||||
.map_err(|source| LiveError::Filter {
|
||||
filter: config.filter.clone(),
|
||||
source,
|
||||
})?;
|
||||
let capture = capture
|
||||
.setnonblock()
|
||||
.map_err(|error| open_error(interface, &error))?;
|
||||
let link_type = capture.get_datalink().0;
|
||||
|
||||
let (sender, receiver) = flume::bounded(CHANNEL_CAPACITY);
|
||||
let stop = Arc::new(AtomicBool::new(false));
|
||||
let failure = Arc::new(OnceLock::new());
|
||||
let thread = std::thread::Builder::new()
|
||||
.name("tlsfp-capture".to_owned())
|
||||
.spawn({
|
||||
let stop = Arc::clone(&stop);
|
||||
let failure = Arc::clone(&failure);
|
||||
move || capture_loop(capture, &sender, &stop, &failure)
|
||||
})
|
||||
.map_err(LiveError::Spawn)?;
|
||||
|
||||
Ok(Self {
|
||||
receiver,
|
||||
staged: CapturedFrame {
|
||||
ts_nanos: 0,
|
||||
data: SmallVec::new(),
|
||||
},
|
||||
link_type,
|
||||
stop,
|
||||
failure,
|
||||
failure_reported: false,
|
||||
thread: Some(thread),
|
||||
})
|
||||
}
|
||||
|
||||
/// Returns a handle that requests a graceful stop of the capture.
|
||||
pub fn stop_handle(&self) -> StopHandle {
|
||||
StopHandle(Arc::clone(&self.stop))
|
||||
}
|
||||
|
||||
/// The async twin of [`PacketSource::next_frame`] for the tokio side
|
||||
/// of the bridge: suspends instead of blocking, then stages and
|
||||
/// returns the frame through the same internals.
|
||||
pub async fn next_frame_async(&mut self) -> Result<Option<RawFrame<'_>>, SourceError> {
|
||||
let received = self.receiver.recv_async().await;
|
||||
match received {
|
||||
Ok(frame) => Ok(Some(self.stage(frame))),
|
||||
Err(_) => self.drained(),
|
||||
}
|
||||
}
|
||||
|
||||
/// Stops the capture, waits for the thread to finish, and returns the
|
||||
/// kernel's final counts. `None` when the statistics could not be
|
||||
/// read.
|
||||
pub fn close(mut self) -> Option<CaptureStats> {
|
||||
self.stop.store(true, Ordering::Relaxed);
|
||||
let thread = self.thread.take();
|
||||
drop(self);
|
||||
thread.and_then(|handle| handle.join().ok()).flatten()
|
||||
}
|
||||
|
||||
fn stage(&mut self, frame: CapturedFrame) -> RawFrame<'_> {
|
||||
self.staged = frame;
|
||||
RawFrame {
|
||||
ts_nanos: self.staged.ts_nanos,
|
||||
link_type: self.link_type,
|
||||
data: &self.staged.data,
|
||||
}
|
||||
}
|
||||
|
||||
/// Maps channel disconnection to the trait's vocabulary: a recorded
|
||||
/// capture failure surfaces as an error exactly once, then the source
|
||||
/// reads as exhausted; a clean shutdown reads as exhausted from the
|
||||
/// start.
|
||||
fn drained(&mut self) -> Result<Option<RawFrame<'_>>, SourceError> {
|
||||
if self.failure_reported {
|
||||
return Ok(None);
|
||||
}
|
||||
match self.failure.get() {
|
||||
Some(message) => {
|
||||
self.failure_reported = true;
|
||||
Err(SourceError::Capture(message.clone()))
|
||||
}
|
||||
None => Ok(None),
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
impl PacketSource for LiveSource {
|
||||
fn next_frame(&mut self) -> Result<Option<RawFrame<'_>>, SourceError> {
|
||||
match self.receiver.recv() {
|
||||
Ok(frame) => Ok(Some(self.stage(frame))),
|
||||
Err(_) => self.drained(),
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/// Dropping the source signals the capture thread instead of joining it.
|
||||
/// The receiver field drops right after, which unblocks a capture thread
|
||||
/// parked on a full channel, so the thread always exits on its own within
|
||||
/// one poll timeout. [`LiveSource::close`] is the path that also waits
|
||||
/// and collects statistics.
|
||||
impl Drop for LiveSource {
|
||||
fn drop(&mut self) {
|
||||
self.stop.store(true, Ordering::Relaxed);
|
||||
}
|
||||
}
|
||||
|
||||
/// The capture thread: wait for readability, drain the ring, repeat.
|
||||
///
|
||||
/// Each turn polls the capture fd with a bounded timeout. A timeout is the
|
||||
/// heartbeat that lets an idle interface still notice a stop request. When
|
||||
/// the fd is readable the loop drains every queued packet before polling
|
||||
/// again, since one readiness signal can cover a burst. A full channel
|
||||
/// makes `send` block, which stops the draining, which lets the kernel
|
||||
/// buffer absorb the burst and count what it sheds. The non-blocking
|
||||
/// handle reports an empty ring as `TimeoutExpired`, the signal to go wait
|
||||
/// again. An error or hangup latched on the fd, the mark of an interface
|
||||
/// going down, is drained of any last packets and then ends the loop,
|
||||
/// because `poll` keeps reporting a latched condition as ready and would
|
||||
/// otherwise spin a core over an empty ring. Any other capture or poll
|
||||
/// error is recorded for the consumer and ends the loop.
|
||||
fn capture_loop(
|
||||
mut capture: Capture<Active>,
|
||||
sender: &flume::Sender<CapturedFrame>,
|
||||
stop: &AtomicBool,
|
||||
failure: &OnceLock<String>,
|
||||
) -> Option<CaptureStats> {
|
||||
let timeout = Timespec {
|
||||
tv_sec: 0,
|
||||
tv_nsec: POLL_TIMEOUT_MILLIS * NANOS_PER_MILLI,
|
||||
};
|
||||
'capture: while !stop.load(Ordering::Relaxed) {
|
||||
let revents = {
|
||||
let mut fds = [PollFd::new(&capture, PollFlags::IN)];
|
||||
match poll(&mut fds, Some(&timeout)) {
|
||||
Ok(0) | Err(rustix::io::Errno::INTR) => None,
|
||||
Ok(_) => Some(fds[0].revents()),
|
||||
Err(error) => {
|
||||
let _ = failure.set(format!("waiting on the capture failed: {error}"));
|
||||
break 'capture;
|
||||
}
|
||||
}
|
||||
};
|
||||
let Some(revents) = revents else {
|
||||
continue;
|
||||
};
|
||||
loop {
|
||||
if stop.load(Ordering::Relaxed) {
|
||||
break 'capture;
|
||||
}
|
||||
match capture.next_packet() {
|
||||
Ok(packet) => {
|
||||
let frame = CapturedFrame {
|
||||
ts_nanos: timeval_to_nanos(
|
||||
packet.header.ts.tv_sec,
|
||||
packet.header.ts.tv_usec,
|
||||
),
|
||||
data: SmallVec::from_slice(packet.data),
|
||||
};
|
||||
if sender.send(frame).is_err() {
|
||||
break 'capture;
|
||||
}
|
||||
}
|
||||
Err(pcap::Error::TimeoutExpired) => break,
|
||||
Err(error) => {
|
||||
let _ = failure.set(error.to_string());
|
||||
break 'capture;
|
||||
}
|
||||
}
|
||||
}
|
||||
if revents.intersects(PollFlags::ERR | PollFlags::HUP | PollFlags::NVAL) {
|
||||
let _ = failure.set("the capture interface reported an error or hangup".to_owned());
|
||||
break 'capture;
|
||||
}
|
||||
}
|
||||
capture.stats().ok().map(|stat| CaptureStats {
|
||||
received: u64::from(stat.received),
|
||||
kernel_dropped: u64::from(stat.dropped),
|
||||
interface_dropped: u64::from(stat.if_dropped),
|
||||
})
|
||||
}
|
||||
|
||||
/// Converts a capture timestamp to nanoseconds since the epoch, the unit
|
||||
/// [`RawFrame`] carries. Generic over the integer widths because libc's
|
||||
/// timeval fields differ across platforms. Saturates rather than wraps,
|
||||
/// and clamps timestamps from before the epoch to zero.
|
||||
fn timeval_to_nanos(sec: impl Into<i64>, micros: impl Into<i64>) -> u64 {
|
||||
let sec = u64::try_from(sec.into()).unwrap_or(0);
|
||||
let micros = u64::try_from(micros.into()).unwrap_or(0);
|
||||
sec.saturating_mul(NANOS_PER_SECOND)
|
||||
.saturating_add(micros.saturating_mul(NANOS_PER_MICRO))
|
||||
}
|
||||
|
||||
/// Builds the open failure message an operator can act on: the underlying
|
||||
/// error, the capability grant when the cause is permissions, and the
|
||||
/// interfaces libpcap can actually see.
|
||||
fn open_error(interface: &str, error: &pcap::Error) -> LiveError {
|
||||
use std::fmt::Write as _;
|
||||
|
||||
let mut message = format!("cannot open capture on {interface}: {error}");
|
||||
if is_permission_denied(error) {
|
||||
let binary = std::env::current_exe().map_or_else(
|
||||
|_| "$(command -v tlsfp)".to_owned(),
|
||||
|path| path.display().to_string(),
|
||||
);
|
||||
let _ = write!(
|
||||
message,
|
||||
"\nlive capture needs CAP_NET_RAW; grant it once per build with:\n sudo setcap cap_net_raw,cap_net_admin=eip {binary}"
|
||||
);
|
||||
}
|
||||
if let Ok(devices) = Device::list() {
|
||||
if !devices.is_empty() {
|
||||
let names: Vec<String> = devices.into_iter().map(|device| device.name).collect();
|
||||
let _ = write!(message, "\navailable interfaces: {}", names.join(", "));
|
||||
}
|
||||
}
|
||||
LiveError::Open(message)
|
||||
}
|
||||
|
||||
/// libpcap has no structured permission error; it reports activation
|
||||
/// failures as text. The two phrasings Linux produces both name the
|
||||
/// problem, which is enough to decide whether the setcap hint belongs in
|
||||
/// the message.
|
||||
fn is_permission_denied(error: &pcap::Error) -> bool {
|
||||
match error {
|
||||
pcap::Error::PcapError(message) => {
|
||||
let lower = message.to_ascii_lowercase();
|
||||
lower.contains("permission") || lower.contains("not permitted")
|
||||
}
|
||||
pcap::Error::IoError(kind) => *kind == ErrorKind::PermissionDenied,
|
||||
_ => false,
|
||||
}
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
mod tests {
|
||||
use std::sync::Arc;
|
||||
use std::sync::OnceLock;
|
||||
use std::sync::atomic::AtomicBool;
|
||||
|
||||
use etherparse::PacketBuilder;
|
||||
use smallvec::SmallVec;
|
||||
|
||||
use tlsfp_core::{PacketSource, Pipeline, PipelineConfig, SourceError};
|
||||
|
||||
use super::{CapturedFrame, LiveConfig, LiveSource, timeval_to_nanos};
|
||||
|
||||
fn source_from_parts(
|
||||
receiver: flume::Receiver<CapturedFrame>,
|
||||
failure: Arc<OnceLock<String>>,
|
||||
) -> LiveSource {
|
||||
LiveSource {
|
||||
receiver,
|
||||
staged: CapturedFrame {
|
||||
ts_nanos: 0,
|
||||
data: SmallVec::new(),
|
||||
},
|
||||
link_type: 1,
|
||||
stop: Arc::new(AtomicBool::new(false)),
|
||||
failure,
|
||||
failure_reported: false,
|
||||
thread: None,
|
||||
}
|
||||
}
|
||||
|
||||
fn frame(ts_nanos: u64, data: &[u8]) -> CapturedFrame {
|
||||
CapturedFrame {
|
||||
ts_nanos,
|
||||
data: SmallVec::from_slice(data),
|
||||
}
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn frames_cross_the_bridge_in_order_and_end_cleanly() {
|
||||
let (sender, receiver) = flume::bounded(8);
|
||||
let producer = std::thread::spawn(move || {
|
||||
for i in 0..100u64 {
|
||||
sender.send(frame(i, &i.to_be_bytes())).unwrap();
|
||||
}
|
||||
});
|
||||
|
||||
let mut source = source_from_parts(receiver, Arc::new(OnceLock::new()));
|
||||
for i in 0..100u64 {
|
||||
let staged = source.next_frame().unwrap().unwrap();
|
||||
assert_eq!(staged.ts_nanos, i);
|
||||
assert_eq!(staged.link_type, 1);
|
||||
assert_eq!(staged.data, i.to_be_bytes());
|
||||
}
|
||||
assert!(source.next_frame().unwrap().is_none());
|
||||
producer.join().unwrap();
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn capture_failure_surfaces_once_then_reads_exhausted() {
|
||||
let (sender, receiver) = flume::bounded::<CapturedFrame>(1);
|
||||
let failure = Arc::new(OnceLock::new());
|
||||
failure.set("the interface went away".to_owned()).unwrap();
|
||||
drop(sender);
|
||||
|
||||
let mut source = source_from_parts(receiver, failure);
|
||||
let error = source.next_frame().unwrap_err();
|
||||
assert!(matches!(
|
||||
error,
|
||||
SourceError::Capture(message) if message.contains("went away")
|
||||
));
|
||||
assert!(source.next_frame().unwrap().is_none());
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn live_frames_feed_the_same_pipeline_as_files() {
|
||||
let request = b"GET / HTTP/1.1\r\nHost: example.com\r\nAccept: */*\r\n\r\n";
|
||||
let builder = PacketBuilder::ethernet2([1; 6], [2; 6])
|
||||
.ipv4([10, 0, 0, 1], [10, 0, 0, 2], 64)
|
||||
.tcp(40000, 80, 1000, 64240);
|
||||
let mut bytes = Vec::with_capacity(builder.size(request.len()));
|
||||
builder.write(&mut bytes, request).unwrap();
|
||||
|
||||
let (sender, receiver) = flume::bounded(8);
|
||||
sender.send(frame(7, &bytes)).unwrap();
|
||||
drop(sender);
|
||||
|
||||
let mut source = source_from_parts(receiver, Arc::new(OnceLock::new()));
|
||||
let mut pipeline = Pipeline::new(PipelineConfig::default());
|
||||
let mut events = Vec::new();
|
||||
pipeline
|
||||
.run(&mut source, |event| events.push(event))
|
||||
.unwrap();
|
||||
|
||||
assert_eq!(events.len(), 1);
|
||||
assert!(events[0].to_string().contains("http_request"));
|
||||
assert_eq!(events[0].src.to_string(), "10.0.0.1:40000");
|
||||
assert_eq!(pipeline.counters().tcp_segments, 1);
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn async_bridge_yields_frames_then_reads_exhausted() {
|
||||
let (sender, receiver) = flume::bounded(2);
|
||||
let mut source = source_from_parts(receiver, Arc::new(OnceLock::new()));
|
||||
|
||||
sender.send(frame(1, &[0xab])).unwrap();
|
||||
drop(sender);
|
||||
|
||||
let staged = source.next_frame_async().await.unwrap().unwrap();
|
||||
assert_eq!(staged.ts_nanos, 1);
|
||||
assert_eq!(staged.data, [0xab]);
|
||||
assert!(source.next_frame_async().await.unwrap().is_none());
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn async_bridge_surfaces_capture_failure() {
|
||||
let (sender, receiver) = flume::bounded::<CapturedFrame>(1);
|
||||
let failure = Arc::new(OnceLock::new());
|
||||
failure.set("device vanished".to_owned()).unwrap();
|
||||
drop(sender);
|
||||
|
||||
let mut source = source_from_parts(receiver, failure);
|
||||
let error = source.next_frame_async().await.unwrap_err();
|
||||
assert!(matches!(error, SourceError::Capture(_)));
|
||||
assert!(source.next_frame_async().await.unwrap().is_none());
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn timeval_conversion_scales_clamps_and_saturates() {
|
||||
assert_eq!(timeval_to_nanos(1i64, 500_000i64), 1_500_000_000);
|
||||
assert_eq!(timeval_to_nanos(0i64, 7i64), 7_000);
|
||||
assert_eq!(timeval_to_nanos(-5i64, 0i64), 0);
|
||||
assert_eq!(timeval_to_nanos(0i64, -1i64), 0);
|
||||
assert_eq!(timeval_to_nanos(i64::MAX, 999_999i64), u64::MAX);
|
||||
}
|
||||
|
||||
#[test]
|
||||
#[ignore = "needs CAP_NET_RAW or root: sudo -E cargo test -p tlsfp -- --ignored"]
|
||||
fn loopback_capture_sees_a_syn() {
|
||||
let config = LiveConfig {
|
||||
filter: "tcp and port 39999".to_owned(),
|
||||
promiscuous: false,
|
||||
};
|
||||
let mut source = LiveSource::open("lo", &config).unwrap();
|
||||
let _ = std::net::TcpStream::connect(("127.0.0.1", 39999));
|
||||
|
||||
let seen = source.next_frame().unwrap().expect("a SYN on loopback");
|
||||
assert!(!seen.data.is_empty());
|
||||
|
||||
let stats = source.close().expect("final capture statistics");
|
||||
assert!(stats.received >= 1);
|
||||
}
|
||||
}
|
||||
|
|
@ -2,6 +2,7 @@
|
|||
// main.rs
|
||||
|
||||
mod cli;
|
||||
mod live;
|
||||
|
||||
use anyhow::Result;
|
||||
use clap::Parser;
|
||||
|
|
|
|||
Loading…
Reference in New Issue