feat: started to make a TCP state machine

This commit is contained in:
Andrew Rioux 2023-09-20 20:50:04 -04:00
parent f092548a8c
commit ed13defb07
Signed by: andrew.rioux
GPG Key ID: 9B8BAC47C17ABB94
3 changed files with 142 additions and 212 deletions

View File

@ -9,9 +9,13 @@ use pcap_sys::packets::*;
#[tokio::main]
async fn main() -> anyhow::Result<()> {
let target = std::env::args().skip(1).next().ok_or(anyhow!("could not get target IP"))?.parse::<Ipv4Addr>()?;
let target = std::env::args()
.skip(1)
.next()
.ok_or(anyhow!("could not get target IP"))?
.parse::<Ipv4Addr>()?;
let (ifname, src_mac, dst_mac, srcip) = {
let (ifname, _, srcip, src_mac, dst_mac, _) = {
let socket = netlink::Socket::new()?;
let routes = socket.get_routes()?;
@ -19,11 +23,8 @@ async fn main() -> anyhow::Result<()> {
let links = socket.get_links()?;
let addrs = socket.get_addrs()?;
let (ifname, srcip, srcmac, dstmac) =
route::get_macs_and_src_for_ip(&addrs, &routes, &neighs, &links, target)
.ok_or(anyhow!("unable to find a route to the IP"))?;
(ifname, srcmac, dstmac, srcip)
.ok_or(anyhow!("unable to find a route to the IP"))?
};
let mut interface = pcap_sys::new_aggregate_interface(false)?;
@ -56,12 +57,18 @@ async fn main() -> anyhow::Result<()> {
match evt {
EventType::Packet((_, Ok(pkt))) => {
let eth_pkt = pkt.pkt();
let Ok(Layer3Pkt::IPv4Pkt(ip_pkt)) = eth_pkt.get_layer3_pkt() else { continue; };
let Ok(Layer4Pkt::UDP(udp_pkt)) = ip_pkt.get_layer4_packet() else { continue; };
let Ok(Layer3Pkt::IPv4Pkt(ip_pkt)) = eth_pkt.get_layer3_pkt() else {
continue;
};
let Ok(Layer4Pkt::UDP(udp_pkt)) = ip_pkt.get_layer4_packet() else {
continue;
};
let data = udp_pkt.get_data();
let Ok(resp_id) = TryInto::<[u8;4]>::try_into(&data[..4]) else { continue; };
let Ok(resp_id) = TryInto::<[u8; 4]>::try_into(&data[..4]) else {
continue;
};
let resp_id = i32::from_be_bytes(resp_id);
if sent_updates.contains_key(&resp_id) {

View File

@ -577,7 +577,13 @@ impl TCPPacketBuilder {
self
}
pub fn build<I1, I2>(self, srcip: I1, dstip: I2, data: Vec<u8>) -> TCPPacket
pub fn build<I1, I2>(
self,
srcip: I1,
dstip: I2,
data: Vec<u8>,
checksum_offset: Option<u16>,
) -> TCPPacket
where
I1: Into<Ipv4Addr>,
I2: Into<Ipv4Addr>,
@ -626,10 +632,7 @@ impl TCPPacketBuilder {
}
})
.fold(0u32, |acc, e| acc + e)
+ 30;
// + 30 to intentionally deviate from
// the RFC, to make it so that I can identify the packets as sparse
// packets
+ (checksum_offset.unwrap_or(0) as u32);
let checksum = (checksum >> 16) + (checksum & 0xffff);
let checksum = ((checksum >> 16) as u16) + (checksum as u16);
@ -645,3 +648,33 @@ impl TCPPacketBuilder {
}
}
}
#[cfg(test)]
mod test {
use std::net::Ipv4Addr;
use crate::TCPPacketBuilder;
#[test]
fn test_checksum() {
let tcp_packet = TCPPacketBuilder::default()
.srcport(57116)
.dstport(54248)
.seqnumber(0xaed77262)
.acknumber(0)
.syn(true)
.window(64240)
.options(vec![
0x02, 0x04, 0x05, 0xb4, 0x04, 0x02, 0x08, 0x0a, 0xd4, 0x06, 0xdf, 0x18, 0x00, 0x00,
0x00, 0x00, 0x01, 0x03, 0x03, 0x07,
])
.build(
Ipv4Addr::new(10, 104, 18, 67),
Ipv4Addr::new(10, 104, 21, 16),
vec![],
None,
);
assert_eq!(tcp_packet.pkt().checksum(), 0x898d);
}
}

View File

@ -1,18 +1,94 @@
use std::{net::Ipv4Addr, os::fd::AsRawFd, sync::Arc};
use std::{
collections::VecDeque,
net::{Ipv4Addr, SocketAddr},
};
use anyhow::anyhow;
use log::{debug, info, trace};
use nl_sys::{netlink, route};
use smoltcp::{
iface::{Config, Interface, Route, SocketSet},
phy::{wait as phy_wait, Device, Medium, RawSocket},
socket::tcp,
time::Instant,
wire::{EthernetAddress, IpAddress, IpCidr, Ipv4Address},
};
use tokio::io::{unix::AsyncFd, Interest};
use packets::{self, EthernetPacket, IPv4Pkt, TCPPacket, TCPPkt};
use pcap_sys;
struct PeerInfo {
remote_addr: Ipv4Addr,
remote_port: u16,
local_addr: Ipv4Addr,
local_port: u16,
}
#[derive(Default)]
struct TcpSocket {
tx_buffer: VecDeque<u8>,
rx_buffer: VecDeque<u8>,
local_seq_no: u32,
remote_seq_no: u32,
remote_last_ack: Option<u32>,
remote_last_win: u16,
local_rx_last_seq: Option<u32>,
local_rx_last_ack: Option<u32>,
state: TcpState,
peer_info: Option<PeerInfo>,
}
#[derive(Default)]
enum TcpState {
Listen,
SynSent,
SynReceived,
Established,
FinWait1,
FinWait2,
CloseWait,
Closing,
LastAck,
TimeWait,
#[default]
Closed,
}
fn socket_accepts_packet(socket: &TcpSocket, ip: IPv4Pkt<'_>, tcp: TCPPkt<'_>) -> bool {
if let Some(peer) = &socket.peer_info {
peer.local_addr == ip.dest_ip()
&& peer.remote_addr == ip.source_ip()
&& peer.local_port == tcp.dstport()
&& peer.remote_port == tcp.srcport()
} else {
false
}
}
fn connect(
socket: &mut TcpSocket,
remote_addr: Ipv4Addr,
remote_port: u16,
local_addr: Ipv4Addr,
) -> anyhow::Result<TCPPacket> {
socket.state = TcpState::SynSent;
let local_port: u16 = loop {
let port = rand::random();
if port > 40000 {
break port;
}
};
socket.peer_info = Some(PeerInfo {
remote_addr,
remote_port,
local_addr,
local_port,
});
Ok(())
}
fn process_packet(socket: &mut TcpSocket, packet: TCPPkt<'_>) -> anyhow::Result<Option<TCPPacket>> {
match socket.state {
TcpState::SynSent if packet.ack() && packet.syn() => {}
_ => {}
}
Ok(None)
}
#[tokio::main]
async fn main() -> anyhow::Result<()> {
@ -41,193 +117,7 @@ async fn main() -> anyhow::Result<()> {
(routes, res)
};
// dbg!((&ifname, srcip, src_mac, dst_mac));
let socket = TcpSocket::default();
let mut device = RawSocket::new(&ifname, Medium::Ethernet)?;
let mut config = Config::new(EthernetAddress(src_mac).into());
config.random_seed = rand::random();
let mut iface = Interface::new(config, &mut device, Instant::now());
iface.update_ip_addrs(|ip_addrs| {
let o = srcip.octets();
debug!(
"source network ip: {}.{}.{}.{}/{src_snmask}",
o[0], o[1], o[2], o[3]
);
ip_addrs
.push(IpCidr::new(
IpAddress::v4(o[0], o[1], o[2], o[3]),
src_snmask,
))
.unwrap();
});
for route in routes.iter() {
let Some(dst) = route.dst() else {
trace!("failed to get route destination");
continue;
};
let Some(hop) = route.hop_iter().next() else {
trace!("no next hop existed for {dst:?}");
continue;
};
if hop.ifindex() != ifindex {
trace!("hop doesn't match ifindex {ifindex}");
continue;
}
let Some(laddr) = hop.gateway() else {
trace!("couldn't get gateway address for {dst:?}");
continue;
};
if laddr.atype() != Some(libc::AF_INET) {
trace!("unable to load IP info for {dst:?}");
continue;
}
if dst.cidrlen() == 0 {
info!("setting default route via {:?}", &laddr);
iface
.routes_mut()
.add_default_ipv4_route(Ipv4Address::from_bytes(&laddr.hw_address()))?;
iface.routes_mut().update(|routes| {
let lip = laddr.hw_address();
_ = routes.push(Route {
cidr: IpCidr::new(IpAddress::v4(10, 0, 0, 0), 8),
via_router: IpAddress::v4(lip[0], lip[1], lip[2], lip[3]),
expires_at: None,
preferred_until: None,
});
_ = routes.push(Route {
cidr: IpCidr::new(IpAddress::v4(172, 16, 0, 0), 12),
via_router: IpAddress::v4(lip[0], lip[1], lip[2], lip[3]),
expires_at: None,
preferred_until: None,
});
});
} else {
let Some(raddr) = route.dst() else {
continue;
};
if raddr.atype() != Some(libc::AF_INET) {
continue;
}
let lip = laddr.hw_address();
let rip = raddr.hw_address();
info!(
"queueing adding {}.{}.{}.{}/{} via {}.{}.{}.{}",
rip[0],
rip[1],
rip[2],
rip[3],
dst.cidrlen(),
lip[0],
lip[1],
lip[2],
lip[3]
);
iface.routes_mut().update(|routes| {
info!(
"adding {}.{}.{}.{}/{} via {}.{}.{}.{}",
rip[0],
rip[1],
rip[2],
rip[3],
dst.cidrlen(),
lip[0],
lip[1],
lip[2],
lip[3]
);
_ = routes.push(Route {
cidr: IpCidr::new(
IpAddress::v4(rip[0], rip[1], rip[2], rip[3]),
dst.cidrlen() as u8,
),
via_router: IpAddress::v4(lip[0], lip[1], lip[2], lip[3]),
expires_at: None,
preferred_until: None,
})
});
}
}
debug!("routes added:");
iface.routes_mut().update(|r| {
for r in r {
debug!("\t{r:?}");
}
});
let tcp_rx_buffer = tcp::SocketBuffer::new(vec![0; 65536]);
let tcp_tx_buffer = tcp::SocketBuffer::new(vec![0; 65536]);
let tcp_socket = tcp::Socket::new(tcp_rx_buffer, tcp_tx_buffer);
let mut sockets = SocketSet::new(vec![]);
let tcp_handle = sockets.add(tcp_socket);
let port = loop {
let port = rand::random::<u16>();
if port > 40_000 {
break port;
}
};
let socket = sockets.get_mut::<tcp::Socket>(tcp_handle);
socket.connect(iface.context(), (ip, 54248), port)?;
let fd = device.as_raw_fd();
/*let interest = Interest::WRITABLE
.add(Interest::READABLE)
.add(Interest::ERROR)
.add(Interest::PRIORITY);
let afd = AsyncFd::with_interest(fd, interest)?;*/
let mut tcp_active = false;
loop {
let timestamp = Instant::now();
iface.poll(timestamp, &mut device, &mut sockets);
let socket = sockets.get_mut::<tcp::Socket>(tcp_handle);
if socket.is_active() && !tcp_active {
info!("connected!");
tcp_active = true;
} else if !socket.is_active() && tcp_active {
info!("disconnected");
tcp_active = false;
}
tcp_active = socket.is_active();
if !socket.is_active() && !tcp_active {
socket.connect(iface.context(), (ip, 54248), port)?;
info!("connecting...");
}
if socket.can_send() {
socket.send_slice(b"ping")?;
info!("sent data!");
} else if socket.may_recv() {
socket.recv(|data| {
if !data.is_empty() {
match std::str::from_utf8(&data) {
Ok(s) => info!("Data received: {}", s),
Err(_) => info!("Data received: {:?}", data),
}
}
(data.len(), data)
})?;
}
phy_wait(fd, iface.poll_delay(timestamp, &sockets))?;
// drop(afd.ready(interest).await?);
}
Ok(())
}