use bytes::{Buf, BufMut, Bytes, BytesMut};
use thiserror::Error;
use crate::rfc1928::consts;
use crate::rfc1928::util::target_addr::{AddrError, TargetAddr};
use std::net::{Ipv4Addr, Ipv6Addr, SocketAddr};
pub const MAX_INNER_LEN: usize = u16::MAX as usize;
const FRAME_LEN_FIELD: usize = 2;
#[derive(Debug, Error)]
pub enum WireError {
#[error("datagram frame too large for u16 length prefix: {0} bytes of addr + data (max {1})")]
Oversize(usize, usize),
#[error("address encode: {0}")]
Address(#[from] AddrError),
#[error("empty frame: no address bytes (len == 0 is unreachable from a compliant peer)")]
EmptyFrame,
#[error("unknown ATYP {0:#x}: protocol garbage")]
BadAtyp(u8),
#[error("frame too short for its address form")]
ShortAddress,
#[error("domain form is not valid UTF-8")]
DomainNotUtf8(#[source] std::str::Utf8Error),
}
pub fn addr_encoding_len(addr: &TargetAddr) -> usize {
match addr {
TargetAddr::Ip(SocketAddr::V4(_)) => 1 + 4 + 2,
TargetAddr::Ip(SocketAddr::V6(_)) => 1 + 16 + 2,
TargetAddr::Domain(domain, _) => 1 + 1 + domain.len() + 2,
}
}
pub fn frame(addr: &TargetAddr, payload: &[u8]) -> Result<Bytes, WireError> {
let addr_bytes = addr.to_be_bytes()?;
let inner = addr_bytes
.len()
.checked_add(payload.len())
.unwrap_or(MAX_INNER_LEN + 1);
if inner > MAX_INNER_LEN {
return Err(WireError::Oversize(inner, MAX_INNER_LEN));
}
let mut buf = BytesMut::with_capacity(FRAME_LEN_FIELD + inner);
buf.put_u16(inner as u16);
buf.put_slice(&addr_bytes);
buf.put_slice(payload);
Ok(buf.freeze())
}
pub fn frame_inner(inner: &[u8]) -> Result<Bytes, WireError> {
if inner.len() > MAX_INNER_LEN {
return Err(WireError::Oversize(inner.len(), MAX_INNER_LEN));
}
let mut buf = BytesMut::with_capacity(FRAME_LEN_FIELD + inner.len());
buf.put_u16(inner.len() as u16);
buf.put_slice(inner);
Ok(buf.freeze())
}
#[derive(Debug)]
pub struct DatagramReader {
buf: BytesMut,
current: Option<(usize, BytesMut)>,
}
impl Default for DatagramReader {
fn default() -> Self {
Self::new()
}
}
impl DatagramReader {
pub fn new() -> Self {
Self {
buf: BytesMut::new(),
current: None,
}
}
pub fn feed(&mut self, chunk: &[u8]) -> Vec<Result<(TargetAddr, Bytes), WireError>> {
self.buf.extend_from_slice(chunk);
let mut out = Vec::new();
loop {
let Some((declared, acc)) = self.current.as_mut() else {
if self.buf.len() < FRAME_LEN_FIELD {
break;
}
let declared = u16::from_be_bytes([self.buf[0], self.buf[1]]) as usize;
self.buf.advance(FRAME_LEN_FIELD);
self.current = Some((declared, BytesMut::new()));
continue;
};
if *declared == 0 {
self.current = None;
out.push(Err(WireError::EmptyFrame));
continue;
}
let take = (*declared - acc.len()).min(self.buf.len());
if take > 0 {
acc.extend_from_slice(&self.buf[..take]);
self.buf.advance(take);
}
if acc.len() == *declared {
let inner = std::mem::take(acc).freeze();
self.current = None;
out.push(decode_inner(&inner));
} else {
break;
}
}
out
}
pub fn is_mid_frame(&self) -> bool {
self.current.is_some()
}
pub fn retained_bytes(&self) -> usize {
self.buf.len() + self.current.as_ref().map_or(0, |(_, acc)| acc.len())
}
}
fn decode_inner(inner: &[u8]) -> Result<(TargetAddr, Bytes), WireError> {
if inner.is_empty() {
return Err(WireError::EmptyFrame);
}
let atyp = inner[0];
match atyp {
consts::SOCKS5_ADDR_TYPE_IPV4 => {
if inner.len() < 1 + 4 + 2 {
return Err(WireError::ShortAddress);
}
let ip = Ipv4Addr::new(inner[1], inner[2], inner[3], inner[4]);
let port = u16::from_be_bytes([inner[5], inner[6]]);
Ok((
TargetAddr::Ip(SocketAddr::from((ip, port))),
Bytes::copy_from_slice(&inner[7..]),
))
}
consts::SOCKS5_ADDR_TYPE_IPV6 => {
if inner.len() < 1 + 16 + 2 {
return Err(WireError::ShortAddress);
}
let mut octets = [0u8; 16];
octets.copy_from_slice(&inner[1..17]);
let ip = Ipv6Addr::from(octets);
let port = u16::from_be_bytes([inner[17], inner[18]]);
Ok((
TargetAddr::Ip(SocketAddr::from((ip, port))),
Bytes::copy_from_slice(&inner[19..]),
))
}
consts::SOCKS5_ADDR_TYPE_DOMAIN_NAME => {
if inner.len() < 2 {
return Err(WireError::ShortAddress);
}
let dlen = inner[1] as usize;
if inner.len() < 2 + dlen + 2 {
return Err(WireError::ShortAddress);
}
let domain =
std::str::from_utf8(&inner[2..2 + dlen]).map_err(WireError::DomainNotUtf8)?;
let port_at = 2 + dlen;
let port = u16::from_be_bytes([inner[port_at], inner[port_at + 1]]);
Ok((
TargetAddr::Domain(domain.to_owned(), port),
Bytes::copy_from_slice(&inner[port_at + 2..]),
))
}
other => Err(WireError::BadAtyp(other)),
}
}
#[cfg(test)]
mod tests {
use super::*;
use crate::rfc1928::util::target_addr::AddrError as AddrErr;
use std::net::{SocketAddrV4, SocketAddrV6};
fn v4(port: u16) -> TargetAddr {
TargetAddr::Ip(SocketAddr::V4(SocketAddrV4::new(
Ipv4Addr::new(192, 168, 0, 1),
port,
)))
}
fn v6(port: u16) -> TargetAddr {
TargetAddr::Ip(SocketAddr::V6(SocketAddrV6::new(
Ipv6Addr::LOCALHOST,
port,
0,
0,
)))
}
fn domain(name: &str, port: u16) -> TargetAddr {
TargetAddr::Domain(name.to_owned(), port)
}
fn feed_one(
reader: &mut DatagramReader,
chunk: &[u8],
) -> Option<Result<(TargetAddr, Bytes), WireError>> {
reader.feed(chunk).into_iter().next()
}
#[test]
fn addr_encoding_len_matches_the_rfc_forms() {
assert_eq!(addr_encoding_len(&v4(80)), 7);
assert_eq!(addr_encoding_len(&v6(443)), 19);
assert_eq!(addr_encoding_len(&domain("example.com", 53)), 15);
assert_eq!(addr_encoding_len(&domain(&"a".repeat(255), 53)), 259);
}
#[test]
fn frame_layout_is_len_then_addr_then_payload() {
let addr = v4(80);
let framed = frame(&addr, b"hi").unwrap();
let addr_bytes = addr.to_be_bytes().unwrap();
let inner = (addr_bytes.len() + 2) as u16;
assert_eq!(&framed[..2], &inner.to_be_bytes());
assert_eq!(&framed[2..2 + addr_bytes.len()], &addr_bytes[..]);
assert_eq!(&framed[2 + addr_bytes.len()..], b"hi");
assert_eq!(framed.len(), 2 + inner as usize);
}
#[test]
fn round_trips_all_address_forms() {
for addr in [v4(80), v6(443), domain("example.com", 53)] {
let framed = frame(&addr, b"hello").unwrap();
let mut r = DatagramReader::new();
let dgs = r.feed(&framed);
assert_eq!(dgs.len(), 1);
let (src, payload) = dgs.into_iter().next().unwrap().unwrap();
assert_eq!(src, addr);
assert_eq!(&payload[..], b"hello");
}
}
#[test]
fn empty_payload_is_a_legal_datagram_not_eof() {
for addr in [v4(53), v6(53), domain("example.com", 53)] {
let framed = frame(&addr, b"").unwrap();
assert_eq!(framed.len(), 2 + addr_encoding_len(&addr));
let mut r = DatagramReader::new();
let dgs = r.feed(&framed);
assert_eq!(dgs.len(), 1, "an empty datagram is a real datagram");
let (src, payload) = dgs.into_iter().next().unwrap().unwrap();
assert_eq!(src, addr);
assert!(payload.is_empty());
}
}
#[test]
fn frame_split_across_awkward_chunks_reassembles() {
let framed = frame(&v4(80), b"0123456789").unwrap();
let mut r = DatagramReader::new();
let mut got = None;
for piece in framed.chunks(7) {
for res in r.feed(piece) {
got = Some(res);
}
}
let (src, payload) = got
.expect("datagram reassembled across 7-byte chunks")
.unwrap();
assert_eq!(src, v4(80));
assert_eq!(&payload[..], b"0123456789");
}
#[test]
fn byte_by_byte_feeding_reassembles_batched_frames_in_order() {
let mut buf = BytesMut::new();
buf.extend_from_slice(&frame(&v4(1), b"abc").unwrap());
buf.extend_from_slice(&frame(&domain("example.com", 2), b"de").unwrap());
buf.extend_from_slice(&frame(&v6(3), b"").unwrap());
let wire = buf.freeze();
let mut r = DatagramReader::new();
let mut got = Vec::new();
for piece in wire.chunks(1) {
got.extend(r.feed(piece));
}
assert_eq!(got.len(), 3);
let (src1, p1) = got.remove(0).unwrap();
assert_eq!(src1, v4(1));
assert_eq!(&p1[..], b"abc");
let (src2, p2) = got.remove(0).unwrap();
assert_eq!(src2, domain("example.com", 2));
assert_eq!(&p2[..], b"de");
let (src3, p3) = got.remove(0).unwrap();
assert_eq!(src3, v6(3));
assert!(p3.is_empty());
}
#[test]
fn two_frames_batched_in_one_chunk() {
let mut buf = BytesMut::new();
buf.extend_from_slice(&frame(&v4(1), b"abc").unwrap());
buf.extend_from_slice(&frame(&v6(2), b"de").unwrap());
let mut r = DatagramReader::new();
let mut dgs = r.feed(&buf.freeze());
assert_eq!(dgs.len(), 2);
let (src1, p1) = dgs.remove(0).unwrap();
assert_eq!(src1, v4(1));
assert_eq!(&p1[..], b"abc");
let (src2, p2) = dgs.remove(0).unwrap();
assert_eq!(src2, v6(2));
assert_eq!(&p2[..], b"de");
}
#[test]
fn partial_header_at_chunk_boundary_waits() {
let framed = frame(&v4(80), b"xyz").unwrap();
let mut r = DatagramReader::new();
assert!(feed_one(&mut r, &framed[..1]).is_none());
assert!(
!r.is_mid_frame(),
"a partial header is not a frame in flight"
);
let pair = feed_one(&mut r, &framed[1..]).expect("frame after header completes");
let (src, payload) = pair.unwrap();
assert_eq!(src, v4(80));
assert_eq!(&payload[..], b"xyz");
}
#[test]
fn mid_frame_state_is_observable() {
let framed = frame(&v4(80), b"0123").unwrap();
let mut r = DatagramReader::new();
r.feed(&framed[..3]);
assert!(r.is_mid_frame());
r.feed(&framed[3..]);
assert!(!r.is_mid_frame());
}
#[test]
fn truncated_stream_never_yields_partial_frame() {
let framed = frame(&v4(80), b"payload").unwrap();
let mut r = DatagramReader::new();
r.feed(&framed[..3]);
assert!(r.is_mid_frame());
assert!(feed_one(&mut r, &[]).is_none());
assert!(r.is_mid_frame());
assert_eq!(
r.feed(&framed[3..framed.len() - 1]).len(),
0,
"one byte short must stay buffered, not yield a partial frame"
);
assert!(r.is_mid_frame());
}
#[test]
fn joint_bound_rejects_65507_payload_to_max_domain() {
let addr = domain(&"a".repeat(255), 53);
let err = frame(&addr, &vec![0u8; 65507]).unwrap_err();
assert!(matches!(err, WireError::Oversize(65766, 65535)));
}
#[test]
fn joint_bound_accepts_65507_payload_to_ipv4() {
let addr = v4(53);
let framed = frame(&addr, &vec![0u8; 65507]).unwrap();
assert_eq!(framed.len(), 2 + 65514);
}
#[test]
fn joint_bound_exact_limit_is_legal_and_one_more_rejects() {
for (addr, max_payload) in [
(v4(80), 65528usize),
(v6(80), 65516usize),
(domain(&"a".repeat(255), 80), 65276usize),
] {
let framed = frame(&addr, &vec![0u8; max_payload]).unwrap();
assert_eq!(framed.len(), 2 + MAX_INNER_LEN);
let err = frame(&addr, &vec![0u8; max_payload + 1]).unwrap_err();
assert!(matches!(err, WireError::Oversize(65536, 65535)));
}
}
#[test]
fn frame_inner_frames_the_egress_inner_bytes_exactly_once() {
let inner = {
let mut v = v4(53).to_be_bytes().unwrap().to_vec();
v.extend_from_slice(b"answer");
v
};
let framed = frame_inner(&inner).unwrap();
assert_eq!(&framed[..2], &(inner.len() as u16).to_be_bytes());
assert_eq!(&framed[2..], &inner[..]);
let mut r = DatagramReader::new();
let dgs = r.feed(&framed);
assert_eq!(dgs.len(), 1);
let (src, payload) = dgs.into_iter().next().unwrap().unwrap();
assert_eq!(src, v4(53));
assert_eq!(&payload[..], b"answer");
}
#[test]
fn frame_inner_rejects_inner_bytes_over_the_u16_bound() {
let err = frame_inner(&vec![0u8; MAX_INNER_LEN + 1]).unwrap_err();
assert!(matches!(err, WireError::Oversize(65536, 65535)));
let framed = frame_inner(&vec![0u8; MAX_INNER_LEN]).unwrap();
assert_eq!(framed.len(), 2 + MAX_INNER_LEN);
}
#[test]
fn frame_and_frame_inner_produce_identical_wires() {
let addr = domain("example.com", 53);
let payload = b"payload";
let mut inner = addr.to_be_bytes().unwrap().to_vec();
inner.extend_from_slice(payload);
assert_eq!(frame_inner(&inner).unwrap(), frame(&addr, payload).unwrap());
}
#[test]
fn oversize_rejected_at_frame_time() {
let addr = v4(80);
let err = frame(&addr, &vec![0u8; MAX_INNER_LEN - 6]).unwrap_err();
assert!(matches!(err, WireError::Oversize(65536, 65535)));
}
#[test]
fn domain_too_long_is_an_address_error() {
let addr = domain(&"a".repeat(256), 80);
let err = frame(&addr, b"").unwrap_err();
match err {
WireError::Address(AddrErr::DomainLenTooLong(256)) => {}
other => panic!("expected Address(DomainLenTooLong(256)), got {other:?}"),
}
}
#[test]
fn garbage_atyp_decode_is_typed_not_panic() {
let wire = [0x00, 0x03, 0x7f, 0x01, 0x02];
let mut r = DatagramReader::new();
let dgs = r.feed(&wire);
assert_eq!(dgs.len(), 1);
match dgs.into_iter().next().unwrap() {
Err(WireError::BadAtyp(0x7f)) => {}
other => panic!("expected BadAtyp(0x7f), got {other:?}"),
}
}
#[test]
fn empty_wire_frame_is_a_decode_error() {
let wire = [0x00, 0x00];
let mut r = DatagramReader::new();
let dgs = r.feed(&wire);
assert_eq!(dgs.len(), 1);
match dgs.into_iter().next().unwrap() {
Err(WireError::EmptyFrame) => {}
other => panic!("expected EmptyFrame, got {other:?}"),
}
}
#[test]
fn short_address_frame_is_a_decode_error() {
let wire = [0x00, 0x08, 0x04, 1, 2, 3, 4, 5, 6, 7];
let mut r = DatagramReader::new();
let dgs = r.feed(&wire);
assert_eq!(dgs.len(), 1);
match dgs.into_iter().next().unwrap() {
Err(WireError::ShortAddress) => {}
other => panic!("expected ShortAddress, got {other:?}"),
}
}
#[test]
fn non_utf8_domain_is_a_decode_error() {
let wire = vec![0x00, 0x07, 0x03, 3, 0xff, 0xfe, 0xfd, 0, 80];
let mut r = DatagramReader::new();
let dgs = r.feed(&wire);
assert_eq!(dgs.len(), 1);
match dgs.into_iter().next().unwrap() {
Err(WireError::DomainNotUtf8(_)) => {}
other => panic!("expected DomainNotUtf8, got {other:?}"),
}
}
}