andrew.rioux/sparse
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases 1 Wiki Activity

made basic TCP connection to receive data

Browse Source 
somewhat
This commit is contained in:
Andrew Rioux
2023-12-05 00:27:02 -05:00
parent 0bb2871568
commit 8c0ae083fe
13 changed files with 349 additions and 173 deletions
Download Patch File Download Diff File Expand all files Collapse all files

304
tcp-test/client/src/main.rs
View File

@@ -1,12 +1,15 @@
use std::{
collections::VecDeque,
mem::MaybeUninit,
net::{Ipv4Addr, SocketAddr},
time::Duration,
time::{Duration, Instant},
};
use anyhow::{anyhow, bail};
use anyhow::{anyhow, bail, Context};
use ringbuf::LocalRb;
use tokio::{sync::mpsc, task, time::sleep_until};
use tokio_stream::StreamExt;
use log::{debug, error, info, trace};
use nl_sys::{netlink, route};
use packets::{
@@ -14,8 +17,6 @@ use packets::{
Layer3Pkt, Layer4Packet, Layer4Pkt, TCPPacket, TCPPacketBuilder, TCPPkt,
};
use pcap_sys;
use tokio::{sync::mpsc, task};
use tokio_stream::StreamExt;
struct PeerInfo {
remote_addr: Ipv4Addr,
@@ -24,10 +25,17 @@ struct PeerInfo {
local_port: u16,
}
enum Timeout {
Idle { heartbeat: Option<Instant> },
Retransmit {},
}
#[derive(Default)]
struct TcpSocket {
tx_buffer: VecDeque<u8>,
rx_buffer: VecDeque<u8>,
tx_buffer: LocalRb<u8, [MaybeUninit<u8>; 65536]>,
rx_buffer: LocalRb<u8, [MaybeUninit<u8>; 65536]>,
first_ack: u32,
first_rack: u32,
local_seq_no: u32,
remote_seq_no: u32,
remote_last_ack: Option<u32>,
@@ -36,9 +44,12 @@ struct TcpSocket {
local_rx_last_ack: Option<u32>,
state: TcpState,
peer_info: Option<PeerInfo>,
ack_timeout: Option<Duration>,
retransmit_timeout: Option<Duration>,
last_packet_received: Option<Instant>,
}
impl TcpSocket {}
#[derive(Default)]
enum TcpState {
Listen,
@@ -82,11 +93,12 @@ fn socket_accepts_packet(socket: &TcpSocket, ip: &IPv4Pkt<'_>, tcp: &TCPPkt<'_>)
}
}
fn connect(
async fn connect(
socket: &mut TcpSocket,
remote_addr: Ipv4Addr,
remote_port: u16,
local_addr: Ipv4Addr,
send_state_changed: &mpsc::Sender<TcpState>,
) -> (u16, TCPPacket) {
socket.state = TcpState::SynSent;
let local_port: u16 = loop {
@@ -95,6 +107,7 @@ fn connect(
break port;
}
};
let _ = send_state_changed.send(TcpState::SynSent).await;
socket.peer_info = Some(PeerInfo {
remote_addr,
remote_port,
@@ -103,6 +116,7 @@ fn connect(
});
socket.local_seq_no = rand::random();
socket.first_ack = socket.local_seq_no;
let packet = TCPPacketBuilder::default()
.srcport(local_port)
@@ -116,31 +130,213 @@ fn connect(
(local_port, packet)
}
fn process_packet(
fn process_timeout(socket: &mut TcpSocket) -> anyhow::Result<Option<(TCPPacketBuilder, Vec<u8>)>> {
Ok(None)
}
async fn process_packet(
socket: &mut TcpSocket,
packet: TCPPkt<'_>,
) -> anyhow::Result<(Option<(TCPPacketBuilder, Vec<u8>)>, Option<Vec<u8>>)> {
match socket.state {
TcpState::SynSent if packet.ack() && packet.syn() => {}
_ => {}
send_state_changed: &mpsc::Sender<TcpState>,
) -> anyhow::Result<(Vec<(TCPPacketBuilder, Vec<u8>)>, Option<Vec<u8>>)> {
match (&socket.state, &socket.peer_info) {
(TcpState::SynSent, Some(peer)) if packet.ack() && packet.syn() => {
log::debug!("established connection with peer");
socket.remote_last_ack = Some(socket.local_seq_no);
socket.remote_seq_no = packet.seqnumber() + 1;
socket.remote_last_win = packet.window();
socket.first_rack = packet.seqnumber();
socket.local_seq_no += 1;
let tcppacketbuilder = TCPPacketBuilder::default()
.srcport(peer.local_port)
.dstport(peer.remote_port)
.ack(true)
.window(512)
.acknumber(socket.remote_seq_no)
.seqnumber(socket.local_seq_no);
socket.state = TcpState::Established;
let _ = send_state_changed.send(TcpState::Established).await;
Ok((vec![(tcppacketbuilder, vec![])], None))
}
(TcpState::Established, Some(peer)) if packet.fin() => {
log::debug!("Received fin!");
socket.remote_last_ack = Some(socket.local_seq_no);
socket.remote_last_win = packet.window();
socket.remote_seq_no += 1;
socket.state = TcpState::CloseWait;
let _ = send_state_changed.send(TcpState::CloseWait).await;
let ackbuilder = TCPPacketBuilder::default()
.srcport(peer.local_port)
.dstport(peer.remote_port)
.ack(true)
.window(512)
.acknumber(socket.remote_seq_no)
.seqnumber(socket.local_seq_no);
socket.local_seq_no += 1;
let finbuilder = TCPPacketBuilder::default()
.srcport(peer.local_port)
.dstport(peer.remote_port)
.ack(true)
.window(512)
.acknumber(socket.remote_seq_no)
.seqnumber(socket.local_seq_no);
Ok((vec![(ackbuilder, vec![]), (finbuilder, vec![])], None))
}
(TcpState::Established, Some(peer)) if packet.ack() => {
log::debug!("received packet from server",);
socket.remote_last_ack = Some(socket.local_seq_no);
socket.remote_last_win = packet.window();
socket.remote_seq_no += (packet.data().len() & 0xFFFFFFFF) as u32;
let tcppacketbuilder = TCPPacketBuilder::default()
.srcport(peer.local_port)
.dstport(peer.remote_port)
.ack(true)
.window(512)
.acknumber(socket.remote_seq_no)
.seqnumber(socket.local_seq_no);
if !packet.data().is_empty() {
log::debug!("received data: {:?}", packet.data());
}
Ok((
vec![(tcppacketbuilder, vec![])],
match packet.data() {
[] => None,
da => Some(da.to_owned()),
},
))
}
(TcpState::FinWait1, _) if packet.ack() => {
socket.state = TcpState::FinWait2;
let _ = send_state_changed.send(TcpState::FinWait2).await;
Ok((vec![], None))
}
(TcpState::FinWait2, Some(peer)) if packet.fin() => {
socket.state = TcpState::Closed;
let _ = send_state_changed.send(TcpState::Closed).await;
socket.remote_last_ack = Some(socket.local_seq_no);
socket.remote_seq_no = packet.seqnumber() + 1;
socket.remote_last_win = packet.window();
socket.local_seq_no += 1;
let tcppacketbuilder = TCPPacketBuilder::default()
.srcport(peer.local_port)
.dstport(peer.remote_port)
.ack(true)
.window(512)
.acknumber(socket.remote_seq_no)
.seqnumber(socket.local_seq_no);
Ok((vec![(tcppacketbuilder, vec![])], None))
}
_ => Ok((vec![], None)),
}
Ok((None, None))
}
fn resp_tcp(incoming: &TCPPkt<'_>) -> TCPPacketBuilder {
TCPPacketBuilder::default()
.srcport(incoming.dstport())
.dstport(incoming.srcport())
}
fn seq_tcp(socket: &TcpSocket, mut builder: TCPPacketBuilder) -> TCPPacketBuilder {
builder
}
fn recv_data(socket: &mut TcpSocket, data: &mut [u8]) -> anyhow::Result<usize> {
let (mut prod, mut cons) = socket.tx_buffer.split_ref();
let bytes_read = cons.pop_slice(data);
socket.remote_seq_no += (bytes_read & 0xFFFFFFFF) as u32;
Ok(bytes_read)
}
fn send_data(
socket: &mut TcpSocket,
data: Vec<u8>,
) -> anyhow::Result<Option<(TCPPacketBuilder, Vec<u8>)>> {
Ok(None)
) -> anyhow::Result<Vec<(TCPPacketBuilder, Vec<u8>)>> {
let (mut prod, cons) = socket.tx_buffer.split_ref();
if cons.is_empty() {
_ = prod.push_iter(&mut data.into_iter());
Ok(vec![])
} else {
_ = prod.push_iter(&mut data.into_iter());
Ok(vec![])
}
}
fn close_connection(socket: &mut TcpSocket) -> anyhow::Result<TCPPacket> {
socket.state = TcpState::FinWait1;
let peer = socket
.peer_info
.as_ref()
.context("no connection to close")?;
Ok(TCPPacketBuilder::default()
.srcport(peer.local_port)
.dstport(peer.remote_port)
.fin(true)
.ack(true)
.window(512)
.acknumber(
socket
.local_rx_last_ack
.context("information from synchronizing missing")?,
)
.seqnumber(socket.local_seq_no)
.build(peer.local_addr, peer.remote_addr, vec![], None))
}
struct TcpSocketHandle {
state_changed: mpsc::Receiver<TcpState>,
send_channel: mpsc::Sender<Vec<u8>>,
close: mpsc::Sender<()>,
receiver_channel: mpsc::Receiver<Vec<u8>>,
}
async fn use_socket(mut socket_handle: TcpSocketHandle) {
_ = socket_handle.send_channel.send(b"ping".to_vec()).await;
loop {
let state = socket_handle.state_changed.recv().await;
match state {
Some(TcpState::Established) => break,
_ => {}
}
}
log::info!("Connected to server!");
log::info!("disconnecting!");
/*_ = socket_handle.close.send(()).await;
loop {
let state = socket_handle.state_changed.recv().await;
match state {
Some(TcpState::Closed) => break,
_ => {}
}
}*/
log::info!("disconnected!");
match socket_handle.receiver_channel.recv().await {
Some(bytes) => match std::str::from_utf8(&bytes) {
@@ -155,6 +351,10 @@ async fn use_socket(mut socket_handle: TcpSocketHandle) {
log::error!("could not get packets from server")
}
}
/*_ = socket_handle.send_channel.send(b"pong".to_vec()).await;
log::info!("Sent 'pong'!");*/
}
#[tokio::main]
@@ -183,7 +383,6 @@ async fn main() -> anyhow::Result<()> {
};
let mut socket = TcpSocket::default();
let mut interface = pcap_sys::Interface::<pcap_sys::DevDisabled>::new(&ifname)?;
let srcip: u32 = srcip.into();
@@ -207,7 +406,9 @@ async fn main() -> anyhow::Result<()> {
}};
}
let (port, packet) = connect(&mut socket, ip, 54248, srcip);
let (send_state_changed, mut receive_state_changed) = mpsc::channel(64);
let (port, packet) = connect(&mut socket, ip, 54248, srcip, &send_state_changed).await;
_ = receive_state_changed.recv().await;
interface.set_filter(&format!("arp or (inbound and tcp port {port})"), true, None)?;
@@ -221,16 +422,32 @@ async fn main() -> anyhow::Result<()> {
let (send_channel, mut send_packet) = mpsc::channel(1024);
let (receive_packet, receiver_channel) = mpsc::channel(1024);
let (send_state_changed, state_changed) = mpsc::channel(1024);
let (close, mut recv_close) = mpsc::channel(16);
let socket_handle = TcpSocketHandle {
state_changed,
receiver_channel,
send_channel,
close,
};
task::spawn(async move { use_socket(socket_handle) });
let mut packet_queue = VecDeque::new();
loop {
let Some(packet) = tokio::select! {
let deadline = Instant::now()
+ socket
.retransmit_timeout
.unwrap_or_else(|| Duration::from_millis(10));
tokio::select! {
_ = sleep_until(deadline.into()) => {
let Ok(Some((b, d))) = process_timeout(&mut socket) else { continue; };
let pkt = format_packet!(b.build(srcip, ip, d, None));
packet_queue.push_back(pkt);
},
Some(Ok(bytes)) = packets.next() => {
let pkt = bytes.pkt();
@@ -242,13 +459,17 @@ async fn main() -> anyhow::Result<()> {
continue;
}
let Ok((to_send, received)) = process_packet(&mut socket, tcp_pkt) else { continue; };
let Ok((to_send, received)) = process_packet(&mut socket, tcp_pkt, &send_state_changed).await else { continue; };
if let Some(received) = received {
_ = receive_packet.send(received).await;
}
to_send.map(|(b, d)| format_packet!(b.build(srcip, ip, d, None)))
for (b, d) in to_send {
log::trace!("adding packet to send: {b:?}");
let pkt = format_packet!(b.build(srcip, ip, d, None));
packet_queue.push_back(pkt);
}
},
Ok(Layer3Pkt::ARP(arp)) => {
if arp.opcode() != 1 || arp.plen() != 4 || arp.hwlen() != 6 {
@@ -265,23 +486,36 @@ async fn main() -> anyhow::Result<()> {
}
let response = ARPPacket::construct(ARPMode::Reply, ARPProto::IPv4, &src_mac, &sendermac, &queryip.octets(), &senderip);
Some(EthernetPacket::construct(src_mac, sendermac.try_into().unwrap(), &Layer3Packet::ARP(response)))
let resp2 = EthernetPacket::construct(src_mac, sendermac.try_into().unwrap(), &Layer3Packet::ARP(response));
log::trace!("adding packet to send: ARP");
packet_queue.push_back(resp2);
},
_ => continue
};
},
Some(()) = recv_close.recv() => {
log::trace!("adding packet to send: TCP/IP close");
let Ok(to_send) = close_connection(&mut socket) else { continue; };
packet_queue.push_back(format_packet!(to_send));
},
Some(to_send) = send_packet.recv() => {
log::trace!("adding packet to send: TCP/IP message send");
match send_data(&mut socket, to_send) {
Ok(pkts) => {
for (b, d) in pkts {
let pkt = format_packet!(b.build(srcip, ip, d, None));
packet_queue.push_back(pkt);
}
}
_ => continue
}
},
Some(to_send) = send_packet.recv() => {
match send_data(&mut socket, to_send) {
Ok(v) => v.map(|(b, d)| format_packet!(b.build(srcip, ip, d, None))),
Err(_) => continue
}
},
else => { continue; }
} else {
continue;
};
}
_ = packets.sendpacket(packet.pkt());
if let Some(packet) = packet_queue.pop_front() {
log::trace!("sending packet now ({packet:?})");
_ = packets.sendpacket(packet.pkt());
}
}
}