From 7845d9a1d9ef55be9d3ac761ee29f5ffa7e45744 Mon Sep 17 00:00:00 2001 From: CarterPerez-dev Date: Sun, 14 Jun 2026 02:38:11 -0400 Subject: [PATCH] m1-7 --- .../ja3-ja4-tls-fingerprinting/Cargo.lock | 4 +- .../ja3-ja4-tls-fingerprinting/Cargo.toml | 2 +- .../crates/tlsfp-core/Cargo.toml | 1 - .../crates/tlsfp-core/src/error.rs | 3 - .../crates/tlsfp-core/src/lib.rs | 1 + .../crates/tlsfp-core/src/pipeline/decode.rs | 121 ++- .../crates/tlsfp-core/src/pipeline/mod.rs | 222 ++++- .../crates/tlsfp-core/src/pipeline/source.rs | 3 + .../crates/tlsfp-core/src/quic.rs | 782 ++++++++++++++++++ .../crates/tlsfp-core/tests/pipeline.rs | 51 ++ .../crates/tlsfp/Cargo.toml | 3 + .../crates/tlsfp/src/cli.rs | 183 +++- .../crates/tlsfp/src/live.rs | 562 +++++++++++++ .../crates/tlsfp/src/main.rs | 1 + 14 files changed, 1877 insertions(+), 62 deletions(-) create mode 100644 PROJECTS/intermediate/ja3-ja4-tls-fingerprinting/crates/tlsfp-core/src/quic.rs create mode 100644 PROJECTS/intermediate/ja3-ja4-tls-fingerprinting/crates/tlsfp/src/live.rs diff --git a/PROJECTS/intermediate/ja3-ja4-tls-fingerprinting/Cargo.lock b/PROJECTS/intermediate/ja3-ja4-tls-fingerprinting/Cargo.lock index 1759a05a..4befdd6c 100644 --- a/PROJECTS/intermediate/ja3-ja4-tls-fingerprinting/Cargo.lock +++ b/PROJECTS/intermediate/ja3-ja4-tls-fingerprinting/Cargo.lock @@ -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", diff --git a/PROJECTS/intermediate/ja3-ja4-tls-fingerprinting/Cargo.toml b/PROJECTS/intermediate/ja3-ja4-tls-fingerprinting/Cargo.toml index 5cf2cdc8..887f6f47 100644 --- a/PROJECTS/intermediate/ja3-ja4-tls-fingerprinting/Cargo.toml +++ b/PROJECTS/intermediate/ja3-ja4-tls-fingerprinting/Cargo.toml @@ -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" diff --git a/PROJECTS/intermediate/ja3-ja4-tls-fingerprinting/crates/tlsfp-core/Cargo.toml b/PROJECTS/intermediate/ja3-ja4-tls-fingerprinting/crates/tlsfp-core/Cargo.toml index 63237bba..febc79e0 100644 --- a/PROJECTS/intermediate/ja3-ja4-tls-fingerprinting/crates/tlsfp-core/Cargo.toml +++ b/PROJECTS/intermediate/ja3-ja4-tls-fingerprinting/crates/tlsfp-core/Cargo.toml @@ -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 diff --git a/PROJECTS/intermediate/ja3-ja4-tls-fingerprinting/crates/tlsfp-core/src/error.rs b/PROJECTS/intermediate/ja3-ja4-tls-fingerprinting/crates/tlsfp-core/src/error.rs index df68f4fd..35f47b04 100644 --- a/PROJECTS/intermediate/ja3-ja4-tls-fingerprinting/crates/tlsfp-core/src/error.rs +++ b/PROJECTS/intermediate/ja3-ja4-tls-fingerprinting/crates/tlsfp-core/src/error.rs @@ -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 = core::result::Result; diff --git a/PROJECTS/intermediate/ja3-ja4-tls-fingerprinting/crates/tlsfp-core/src/lib.rs b/PROJECTS/intermediate/ja3-ja4-tls-fingerprinting/crates/tlsfp-core/src/lib.rs index 50fb0aed..d20f1ccc 100644 --- a/PROJECTS/intermediate/ja3-ja4-tls-fingerprinting/crates/tlsfp-core/src/lib.rs +++ b/PROJECTS/intermediate/ja3-ja4-tls-fingerprinting/crates/tlsfp-core/src/lib.rs @@ -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}; diff --git a/PROJECTS/intermediate/ja3-ja4-tls-fingerprinting/crates/tlsfp-core/src/pipeline/decode.rs b/PROJECTS/intermediate/ja3-ja4-tls-fingerprinting/crates/tlsfp-core/src/pipeline/decode.rs index 6db1b4bd..5c204b33 100644 --- a/PROJECTS/intermediate/ja3-ja4-tls-fingerprinting/crates/tlsfp-core/src/pipeline/decode.rs +++ b/PROJECTS/intermediate/ja3-ja4-tls-fingerprinting/crates/tlsfp-core/src/pipeline/decode.rs @@ -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, Skip> { +/// network with the TLS and QUIC it cares about. +pub fn decode_frame(link: i32, data: &[u8]) -> Result, 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, 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 { 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) )); } diff --git a/PROJECTS/intermediate/ja3-ja4-tls-fingerprinting/crates/tlsfp-core/src/pipeline/mod.rs b/PROJECTS/intermediate/ja3-ja4-tls-fingerprinting/crates/tlsfp-core/src/pipeline/mod.rs index fc98eba6..83e951a2 100644 --- a/PROJECTS/intermediate/ja3-ja4-tls-fingerprinting/crates/tlsfp-core/src/pipeline/mod.rs +++ b/PROJECTS/intermediate/ja3-ja4-tls-fingerprinting/crates/tlsfp-core/src/pipeline/mod.rs @@ -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, + crypto: CryptoAssembler, + largest_pn: Option, + 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, + quic_flows: HashMap, 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 = 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)] diff --git a/PROJECTS/intermediate/ja3-ja4-tls-fingerprinting/crates/tlsfp-core/src/pipeline/source.rs b/PROJECTS/intermediate/ja3-ja4-tls-fingerprinting/crates/tlsfp-core/src/pipeline/source.rs index 8a05413f..42c783c8 100644 --- a/PROJECTS/intermediate/ja3-ja4-tls-fingerprinting/crates/tlsfp-core/src/pipeline/source.rs +++ b/PROJECTS/intermediate/ja3-ja4-tls-fingerprinting/crates/tlsfp-core/src/pipeline/source.rs @@ -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. diff --git a/PROJECTS/intermediate/ja3-ja4-tls-fingerprinting/crates/tlsfp-core/src/quic.rs b/PROJECTS/intermediate/ja3-ja4-tls-fingerprinting/crates/tlsfp-core/src/quic.rs new file mode 100644 index 00000000..d5d3c69c --- /dev/null +++ b/PROJECTS/intermediate/ja3-ja4-tls-fingerprinting/crates/tlsfp-core/src/quic.rs @@ -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, 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::::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::::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, +} + +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 { + 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) -> Result { + 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, + pending: BTreeMap>, + 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::::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 { + ::hex::decode(s).unwrap() + } + + /// The protected client Initial packet from RFC 9001 Appendix A.2. + fn rfc9001_a2_protected_packet() -> Vec { + 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", + )) + } +} diff --git a/PROJECTS/intermediate/ja3-ja4-tls-fingerprinting/crates/tlsfp-core/tests/pipeline.rs b/PROJECTS/intermediate/ja3-ja4-tls-fingerprinting/crates/tlsfp-core/tests/pipeline.rs index 3c670e94..8b264b4b 100644 --- a/PROJECTS/intermediate/ja3-ja4-tls-fingerprinting/crates/tlsfp-core/tests/pipeline.rs +++ b/PROJECTS/intermediate/ja3-ja4-tls-fingerprinting/crates/tlsfp-core/tests/pipeline.rs @@ -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"); diff --git a/PROJECTS/intermediate/ja3-ja4-tls-fingerprinting/crates/tlsfp/Cargo.toml b/PROJECTS/intermediate/ja3-ja4-tls-fingerprinting/crates/tlsfp/Cargo.toml index 80fbf70b..b6a77ed3 100644 --- a/PROJECTS/intermediate/ja3-ja4-tls-fingerprinting/crates/tlsfp/Cargo.toml +++ b/PROJECTS/intermediate/ja3-ja4-tls-fingerprinting/crates/tlsfp/Cargo.toml @@ -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 diff --git a/PROJECTS/intermediate/ja3-ja4-tls-fingerprinting/crates/tlsfp/src/cli.rs b/PROJECTS/intermediate/ja3-ja4-tls-fingerprinting/crates/tlsfp/src/cli.rs index 47b061ba..5442e42f 100644 --- a/PROJECTS/intermediate/ja3-ja4-tls-fingerprinting/crates/tlsfp/src/cli.rs +++ b/PROJECTS/intermediate/ja3-ja4-tls-fingerprinting/crates/tlsfp/src/cli.rs @@ -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 = None; + + let capture_failure: Option = 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, +) { + 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); + } +} diff --git a/PROJECTS/intermediate/ja3-ja4-tls-fingerprinting/crates/tlsfp/src/live.rs b/PROJECTS/intermediate/ja3-ja4-tls-fingerprinting/crates/tlsfp/src/live.rs new file mode 100644 index 00000000..de45a236 --- /dev/null +++ b/PROJECTS/intermediate/ja3-ja4-tls-fingerprinting/crates/tlsfp/src/live.rs @@ -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); + +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, + staged: CapturedFrame, + link_type: i32, + stop: Arc, + failure: Arc>, + failure_reported: bool, + thread: Option>>, +} + +impl LiveSource { + /// Opens an interface by name and starts capturing immediately. + pub fn open(interface: &str, config: &LiveConfig) -> Result { + 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>, 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 { + 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>, 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>, 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, + sender: &flume::Sender, + stop: &AtomicBool, + failure: &OnceLock, +) -> Option { + 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, micros: impl Into) -> 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 = 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, + failure: Arc>, + ) -> 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::(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::(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); + } +} diff --git a/PROJECTS/intermediate/ja3-ja4-tls-fingerprinting/crates/tlsfp/src/main.rs b/PROJECTS/intermediate/ja3-ja4-tls-fingerprinting/crates/tlsfp/src/main.rs index ce70db7b..d89db785 100644 --- a/PROJECTS/intermediate/ja3-ja4-tls-fingerprinting/crates/tlsfp/src/main.rs +++ b/PROJECTS/intermediate/ja3-ja4-tls-fingerprinting/crates/tlsfp/src/main.rs @@ -2,6 +2,7 @@ // main.rs mod cli; +mod live; use anyhow::Result; use clap::Parser;