add gatunad server and winforms client v1 (TCP-only)
Raw-Ethernet tunnel over ethertype 0x6969 to bypass WFP killswitches. Server (Rust, AF_PACKET + classic BPF) relays TCP to 127.0.0.1 services. Client (.NET 8 WinForms, SharpPcap/Npcap) discovers upstreams and exposes local loopback listeners. Wire protocol: 6-byte header, 7 frame types, MANIFEST with labeled upstreams. UDP types reserved, unimplemented.
This commit is contained in:
@@ -0,0 +1,17 @@
|
||||
[package]
|
||||
name = "gatuna"
|
||||
version = "0.1.0"
|
||||
edition = "2021"
|
||||
license = "CC0-1.0"
|
||||
|
||||
[[bin]]
|
||||
name = "gatunad"
|
||||
path = "src/main.rs"
|
||||
|
||||
[dependencies]
|
||||
libc = "0.2"
|
||||
tokio = { version = "1", features = ["rt-multi-thread", "macros", "net", "sync", "io-util"] }
|
||||
clap = { version = "4", features = ["derive"] }
|
||||
pnet_datalink = "0.35"
|
||||
tracing = "0.1"
|
||||
tracing-subscriber = "0.3"
|
||||
@@ -0,0 +1,202 @@
|
||||
//! gatuna wire protocol frame encode/decode.
|
||||
|
||||
pub const ETHERTYPE: u16 = 0x6969;
|
||||
pub const ETH_HEADER_LEN: usize = 14;
|
||||
pub const VERSION: u8 = 1;
|
||||
pub const MAX_PAYLOAD: usize = 1480;
|
||||
|
||||
pub const TYPE_DISCOVER: u8 = 0x01;
|
||||
pub const TYPE_MANIFEST: u8 = 0x02;
|
||||
pub const TYPE_OPEN: u8 = 0x03;
|
||||
pub const TYPE_OPEN_ACK: u8 = 0x04;
|
||||
pub const TYPE_OPEN_NAK: u8 = 0x05;
|
||||
pub const TYPE_DATA: u8 = 0x06;
|
||||
pub const TYPE_CLOSE: u8 = 0x07;
|
||||
pub const TYPE_UDP_OPEN: u8 = 0x08;
|
||||
pub const TYPE_UDP_DATA: u8 = 0x09;
|
||||
pub const TYPE_UDP_CLOSE: u8 = 0x0A;
|
||||
|
||||
pub const PROTO_TCP: u8 = 1;
|
||||
pub const PROTO_UDP: u8 = 2;
|
||||
|
||||
#[allow(dead_code)]
|
||||
pub const REASON_UNSPEC: u8 = 0;
|
||||
pub const REASON_UNKNOWN_UPSTREAM: u8 = 1;
|
||||
pub const REASON_CONNECT_FAILED: u8 = 2;
|
||||
#[allow(dead_code)]
|
||||
pub const REASON_OVERSIZE: u8 = 3;
|
||||
pub const REASON_UNKNOWN_SESSION: u8 = 4;
|
||||
|
||||
#[derive(Clone, Debug)]
|
||||
pub struct UpstreamEntry {
|
||||
pub id: u8,
|
||||
pub proto: u8,
|
||||
pub port: u16,
|
||||
pub label: Option<String>,
|
||||
}
|
||||
|
||||
#[derive(Clone, Debug)]
|
||||
pub enum Frame {
|
||||
Discover,
|
||||
Manifest(Vec<UpstreamEntry>),
|
||||
Open { upstream_id: u8 },
|
||||
OpenAck { session_id: u32, upstream_id: u8 },
|
||||
OpenNak { upstream_id: u8, reason: u8 },
|
||||
Data { session_id: u32, payload: Vec<u8> },
|
||||
Close { session_id: u32, reason: Option<u8> },
|
||||
}
|
||||
|
||||
#[derive(Debug)]
|
||||
pub enum DecodeError {
|
||||
Short,
|
||||
BadVersion(u8),
|
||||
UnknownType(u8),
|
||||
BadPayload(&'static str),
|
||||
}
|
||||
|
||||
impl std::fmt::Display for DecodeError {
|
||||
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
|
||||
match self {
|
||||
DecodeError::Short => write!(f, "frame too short"),
|
||||
DecodeError::BadVersion(v) => write!(f, "unsupported version {v}"),
|
||||
DecodeError::UnknownType(t) => write!(f, "unknown frame type {t:#x}"),
|
||||
DecodeError::BadPayload(m) => write!(f, "bad payload: {m}"),
|
||||
}
|
||||
}
|
||||
}
|
||||
impl std::error::Error for DecodeError {}
|
||||
|
||||
fn encode_entry(buf: &mut Vec<u8>, e: &UpstreamEntry) {
|
||||
let label_bytes = e.label.as_deref().unwrap_or("").as_bytes();
|
||||
let label_len = label_bytes.len().min(255) as u8;
|
||||
buf.push(e.id);
|
||||
buf.push(e.proto);
|
||||
buf.extend_from_slice(&e.port.to_be_bytes());
|
||||
buf.push(label_len);
|
||||
buf.extend_from_slice(&label_bytes[..label_len as usize]);
|
||||
}
|
||||
|
||||
impl Frame {
|
||||
pub fn encode(&self) -> Vec<u8> {
|
||||
let mut buf = Vec::new();
|
||||
match self {
|
||||
Frame::Discover => {
|
||||
buf.extend_from_slice(&[VERSION, TYPE_DISCOVER, 0, 0, 0, 0]);
|
||||
}
|
||||
Frame::Manifest(entries) => {
|
||||
buf.extend_from_slice(&[VERSION, TYPE_MANIFEST, 0, 0, 0, 0]);
|
||||
for e in entries {
|
||||
encode_entry(&mut buf, e);
|
||||
}
|
||||
}
|
||||
Frame::Open { upstream_id } => {
|
||||
buf.extend_from_slice(&[VERSION, TYPE_OPEN, 0, 0, 0, 0, *upstream_id]);
|
||||
}
|
||||
Frame::OpenAck { session_id, upstream_id } => {
|
||||
buf.extend_from_slice(&[VERSION, TYPE_OPEN_ACK]);
|
||||
buf.extend_from_slice(&session_id.to_be_bytes());
|
||||
buf.push(*upstream_id);
|
||||
}
|
||||
Frame::OpenNak { upstream_id, reason } => {
|
||||
buf.extend_from_slice(&[VERSION, TYPE_OPEN_NAK, 0, 0, 0, 0]);
|
||||
buf.push(*upstream_id);
|
||||
buf.push(*reason);
|
||||
}
|
||||
Frame::Data { session_id, payload } => {
|
||||
buf.extend_from_slice(&[VERSION, TYPE_DATA]);
|
||||
buf.extend_from_slice(&session_id.to_be_bytes());
|
||||
buf.extend_from_slice(payload);
|
||||
}
|
||||
Frame::Close { session_id, reason } => {
|
||||
buf.extend_from_slice(&[VERSION, TYPE_CLOSE]);
|
||||
buf.extend_from_slice(&session_id.to_be_bytes());
|
||||
if let Some(r) = reason {
|
||||
buf.push(*r);
|
||||
}
|
||||
}
|
||||
}
|
||||
buf
|
||||
}
|
||||
|
||||
pub fn parse(buf: &[u8]) -> Result<Frame, DecodeError> {
|
||||
if buf.len() < 6 {
|
||||
return Err(DecodeError::Short);
|
||||
}
|
||||
let version = buf[0];
|
||||
if version != VERSION {
|
||||
return Err(DecodeError::BadVersion(version));
|
||||
}
|
||||
let typ = buf[1];
|
||||
let session_id = u32::from_be_bytes([buf[2], buf[3], buf[4], buf[5]]);
|
||||
let payload = &buf[6..];
|
||||
match typ {
|
||||
TYPE_DISCOVER => {
|
||||
if !payload.is_empty() {
|
||||
return Err(DecodeError::BadPayload("DISCOVER must be empty"));
|
||||
}
|
||||
Ok(Frame::Discover)
|
||||
}
|
||||
TYPE_MANIFEST => {
|
||||
let mut entries = Vec::new();
|
||||
let mut i = 0;
|
||||
while i < payload.len() {
|
||||
if i + 5 > payload.len() {
|
||||
return Err(DecodeError::BadPayload("MANIFEST entry truncated"));
|
||||
}
|
||||
let id = payload[i];
|
||||
let proto = payload[i + 1];
|
||||
let port = u16::from_be_bytes([payload[i + 2], payload[i + 3]]);
|
||||
let label_len = payload[i + 4] as usize;
|
||||
i += 5;
|
||||
if i + label_len > payload.len() {
|
||||
return Err(DecodeError::BadPayload("MANIFEST label truncated"));
|
||||
}
|
||||
let label = if label_len == 0 {
|
||||
None
|
||||
} else {
|
||||
Some(String::from_utf8_lossy(&payload[i..i + label_len]).into_owned())
|
||||
};
|
||||
i += label_len;
|
||||
entries.push(UpstreamEntry { id, proto, port, label });
|
||||
}
|
||||
Ok(Frame::Manifest(entries))
|
||||
}
|
||||
TYPE_OPEN => {
|
||||
if payload.len() != 1 {
|
||||
return Err(DecodeError::BadPayload("OPEN payload must be 1 byte"));
|
||||
}
|
||||
Ok(Frame::Open { upstream_id: payload[0] })
|
||||
}
|
||||
TYPE_OPEN_ACK => {
|
||||
if payload.len() != 1 {
|
||||
return Err(DecodeError::BadPayload("OPEN_ACK payload must be 1 byte"));
|
||||
}
|
||||
Ok(Frame::OpenAck { session_id, upstream_id: payload[0] })
|
||||
}
|
||||
TYPE_OPEN_NAK => {
|
||||
if payload.len() != 2 {
|
||||
return Err(DecodeError::BadPayload("OPEN_NAK payload must be 2 bytes"));
|
||||
}
|
||||
Ok(Frame::OpenNak { upstream_id: payload[0], reason: payload[1] })
|
||||
}
|
||||
TYPE_DATA => {
|
||||
if payload.len() > MAX_PAYLOAD {
|
||||
return Err(DecodeError::BadPayload("DATA payload exceeds max"));
|
||||
}
|
||||
Ok(Frame::Data { session_id, payload: payload.to_vec() })
|
||||
}
|
||||
TYPE_CLOSE => {
|
||||
let reason = match payload.len() {
|
||||
0 => None,
|
||||
1 => Some(payload[0]),
|
||||
_ => return Err(DecodeError::BadPayload("CLOSE payload must be 0 or 1 bytes")),
|
||||
};
|
||||
Ok(Frame::Close { session_id, reason })
|
||||
}
|
||||
TYPE_UDP_OPEN | TYPE_UDP_DATA | TYPE_UDP_CLOSE => {
|
||||
Err(DecodeError::BadPayload("UDP frame types not implemented in v1"))
|
||||
}
|
||||
other => Err(DecodeError::UnknownType(other)),
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,246 @@
|
||||
//! Raw Ethernet I/O over `AF_PACKET` with a classic BPF filter on ethertype
|
||||
//! `0x6969`. No eBPF authoring: the filter is a hand-assembled cBPF program
|
||||
//! installed via `SO_ATTACH_FILTER`. The kernel may translate it to eBPF
|
||||
//! internally at attach time; that is transparent and not our concern.
|
||||
|
||||
use std::io::{self, ErrorKind};
|
||||
use std::os::unix::io::{AsRawFd, RawFd};
|
||||
use tokio::io::unix::{AsyncFd, AsyncFdReadyGuard};
|
||||
|
||||
use crate::frame::{ETHERTYPE, ETH_HEADER_LEN};
|
||||
|
||||
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
|
||||
pub struct MacAddr(pub [u8; 6]);
|
||||
|
||||
impl MacAddr {
|
||||
#[allow(dead_code)]
|
||||
pub fn broadcast() -> Self {
|
||||
MacAddr([0xff; 6])
|
||||
}
|
||||
}
|
||||
|
||||
struct LinkFd(RawFd);
|
||||
impl AsRawFd for LinkFd {
|
||||
fn as_raw_fd(&self) -> RawFd {
|
||||
self.0
|
||||
}
|
||||
}
|
||||
impl Drop for LinkFd {
|
||||
fn drop(&mut self) {
|
||||
unsafe {
|
||||
libc::close(self.0);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
pub struct Link {
|
||||
fd: AsyncFd<LinkFd>,
|
||||
ifindex: libc::c_int,
|
||||
pub our_mac: MacAddr,
|
||||
}
|
||||
|
||||
// cBPF program for `ether proto 0x6969`:
|
||||
// 0: ldh [12] load ethertype (host-order u16)
|
||||
// 1: jeq #host(ETHERTYPE), 1, 0 match -> skip to accept; else fall to drop
|
||||
// 2: ret #0 drop
|
||||
// 3: ret #0xFFFF accept (return whole packet)
|
||||
//
|
||||
// The ethertype is carried big-endian on the wire; `ldh` loads it in host
|
||||
// byte order, so the comparison constant must be `u16::from_be(ETHERTYPE)`.
|
||||
fn bpf_program() -> [libc::sock_filter; 4] {
|
||||
const BPF_LD_H_ABS: u16 = 0x28;
|
||||
const BPF_JMP_JEQ_K: u16 = 0x15;
|
||||
const BPF_RET_K: u16 = 0x06;
|
||||
let k = u16::from_be(ETHERTYPE) as u32;
|
||||
[
|
||||
libc::sock_filter { code: BPF_LD_H_ABS, jt: 0, jf: 0, k: 12 },
|
||||
libc::sock_filter { code: BPF_JMP_JEQ_K, jt: 1, jf: 0, k },
|
||||
libc::sock_filter { code: BPF_RET_K, jt: 0, jf: 0, k: 0 },
|
||||
libc::sock_filter { code: BPF_RET_K, jt: 0, jf: 0, k: 0xFFFF },
|
||||
]
|
||||
}
|
||||
|
||||
fn htons(v: u16) -> u16 {
|
||||
v.to_be()
|
||||
}
|
||||
|
||||
fn lookup_mac(iface: &str) -> io::Result<MacAddr> {
|
||||
for ni in pnet_datalink::interfaces() {
|
||||
if ni.name == iface {
|
||||
if let Some(mac) = ni.mac {
|
||||
return Ok(MacAddr(mac.0));
|
||||
}
|
||||
}
|
||||
}
|
||||
Err(io::Error::new(
|
||||
ErrorKind::NotFound,
|
||||
format!("could not determine MAC for interface {iface}"),
|
||||
))
|
||||
}
|
||||
|
||||
impl Link {
|
||||
pub fn open(iface: &str) -> io::Result<Link> {
|
||||
unsafe {
|
||||
let fd = libc::socket(libc::AF_PACKET, libc::SOCK_RAW, htons(libc::ETH_P_ALL as u16));
|
||||
if fd < 0 {
|
||||
return Err(io::last_os_error());
|
||||
}
|
||||
|
||||
// Non-blocking for AsyncFd.
|
||||
let flags = libc::fcntl(fd, libc::F_GETFL);
|
||||
if flags < 0 {
|
||||
let e = io::last_os_error();
|
||||
libc::close(fd);
|
||||
return Err(e);
|
||||
}
|
||||
if libc::fcntl(fd, libc::F_SETFL, flags | libc::O_NONBLOCK) < 0 {
|
||||
let e = io::last_os_error();
|
||||
libc::close(fd);
|
||||
return Err(e);
|
||||
}
|
||||
|
||||
let ciface = match std::ffi::CString::new(iface) {
|
||||
Ok(s) => s,
|
||||
Err(_) => {
|
||||
libc::close(fd);
|
||||
return Err(io::Error::new(
|
||||
ErrorKind::InvalidInput,
|
||||
"interface name contains an interior NUL",
|
||||
));
|
||||
}
|
||||
};
|
||||
let ifindex = libc::if_nametoindex(ciface.as_ptr());
|
||||
if ifindex == 0 {
|
||||
let e = io::last_os_error();
|
||||
libc::close(fd);
|
||||
return Err(e);
|
||||
}
|
||||
|
||||
// Bind to the interface so we only receive frames on it.
|
||||
let mut sll: libc::sockaddr_ll = std::mem::zeroed();
|
||||
sll.sll_family = libc::AF_PACKET as u16;
|
||||
sll.sll_protocol = htons(libc::ETH_P_ALL as u16);
|
||||
sll.sll_ifindex = ifindex as libc::c_int;
|
||||
let r = libc::bind(
|
||||
fd,
|
||||
&sll as *const _ as *const libc::sockaddr,
|
||||
std::mem::size_of::<libc::sockaddr_ll>() as libc::socklen_t,
|
||||
);
|
||||
if r < 0 {
|
||||
let e = io::last_os_error();
|
||||
libc::close(fd);
|
||||
return Err(e);
|
||||
}
|
||||
|
||||
// Classic BPF: only our ethertype reaches userspace.
|
||||
let filt = bpf_program();
|
||||
let prog = libc::sock_fprog {
|
||||
len: filt.len() as u16,
|
||||
filter: filt.as_ptr() as *mut libc::sock_filter,
|
||||
};
|
||||
let r = libc::setsockopt(
|
||||
fd,
|
||||
libc::SOL_SOCKET,
|
||||
libc::SO_ATTACH_FILTER,
|
||||
&prog as *const _ as *const libc::c_void,
|
||||
std::mem::size_of::<libc::sock_fprog>() as libc::socklen_t,
|
||||
);
|
||||
if r < 0 {
|
||||
let e = io::last_os_error();
|
||||
libc::close(fd);
|
||||
return Err(e);
|
||||
}
|
||||
|
||||
let our_mac = lookup_mac(iface)?;
|
||||
|
||||
let async_fd = AsyncFd::new(LinkFd(fd))?;
|
||||
Ok(Link {
|
||||
fd: async_fd,
|
||||
ifindex: ifindex as libc::c_int,
|
||||
our_mac,
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
pub async fn readable(&self) -> io::Result<AsyncFdReadyGuard<'_, LinkFd>> {
|
||||
self.fd.readable().await
|
||||
}
|
||||
|
||||
pub async fn writable(&self) -> io::Result<AsyncFdReadyGuard<'_, LinkFd>> {
|
||||
self.fd.writable().await
|
||||
}
|
||||
|
||||
/// Receive one frame. Returns:
|
||||
/// - `Ok(Some((src, payload)))` for a frame we should process.
|
||||
/// - `Ok(None)` for an ignorable frame (our own outgoing frame, short
|
||||
/// frame, ethertype mismatch) — the caller should keep draining without
|
||||
/// clearing readiness.
|
||||
/// - `Err(WouldBlock)` when no more frames are available — the caller
|
||||
/// should clear readiness and await again.
|
||||
pub fn recv<'a>(&self, buf: &'a mut [u8]) -> io::Result<Option<(MacAddr, &'a [u8])>> {
|
||||
let mut sll: libc::sockaddr_ll = unsafe { std::mem::zeroed() };
|
||||
let mut slen = std::mem::size_of::<libc::sockaddr_ll>() as libc::socklen_t;
|
||||
let n = unsafe {
|
||||
libc::recvfrom(
|
||||
self.fd.as_raw_fd(),
|
||||
buf.as_mut_ptr() as *mut libc::c_void,
|
||||
buf.len(),
|
||||
0,
|
||||
&mut sll as *mut _ as *mut libc::sockaddr,
|
||||
&mut slen,
|
||||
)
|
||||
};
|
||||
if n < 0 {
|
||||
return Err(io::last_os_error());
|
||||
}
|
||||
// Skip frames we transmitted (AF_PACKET with ETH_P_ALL loops them back).
|
||||
if sll.sll_pkttype == libc::PACKET_OUTGOING {
|
||||
return Ok(None);
|
||||
}
|
||||
let pkt = &buf[..n as usize];
|
||||
if pkt.len() < ETH_HEADER_LEN {
|
||||
return Ok(None);
|
||||
}
|
||||
let src = MacAddr([pkt[6], pkt[7], pkt[8], pkt[9], pkt[10], pkt[11]]);
|
||||
// BPF already filtered; double-check for safety against the ethertype.
|
||||
let et = u16::from_be_bytes([pkt[12], pkt[13]]);
|
||||
if et != ETHERTYPE {
|
||||
return Ok(None);
|
||||
}
|
||||
Ok(Some((src, &pkt[ETH_HEADER_LEN..])))
|
||||
}
|
||||
|
||||
/// Send `payload` (our protocol frame) wrapped in an Ethernet header to
|
||||
/// `dst`. The ethertype is `0x6969`. Returns `WouldBlock` if the kernel
|
||||
/// buffer is full; the caller is expected to await writability and retry.
|
||||
pub fn send(&self, dst: MacAddr, payload: &[u8]) -> io::Result<()> {
|
||||
let mut frame = Vec::with_capacity(ETH_HEADER_LEN + payload.len());
|
||||
frame.extend_from_slice(&dst.0);
|
||||
frame.extend_from_slice(&self.our_mac.0);
|
||||
frame.extend_from_slice(ÐERTYPE.to_be_bytes());
|
||||
frame.extend_from_slice(payload);
|
||||
|
||||
let mut sll: libc::sockaddr_ll = unsafe { std::mem::zeroed() };
|
||||
sll.sll_family = libc::AF_PACKET as u16;
|
||||
sll.sll_protocol = htons(libc::ETH_P_ALL as u16);
|
||||
sll.sll_ifindex = self.ifindex;
|
||||
sll.sll_hatype = 1; // ARPHRD_ETHER
|
||||
sll.sll_halen = 6;
|
||||
sll.sll_addr[..6].copy_from_slice(&dst.0);
|
||||
|
||||
let r = unsafe {
|
||||
libc::sendto(
|
||||
self.fd.as_raw_fd(),
|
||||
frame.as_ptr() as *const libc::c_void,
|
||||
frame.len(),
|
||||
0,
|
||||
&sll as *const _ as *const libc::sockaddr,
|
||||
std::mem::size_of::<libc::sockaddr_ll>() as libc::socklen_t,
|
||||
)
|
||||
};
|
||||
if r < 0 {
|
||||
return Err(io::last_os_error());
|
||||
}
|
||||
Ok(())
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,207 @@
|
||||
//! `gatunad` — raw-Ethernet tunnel server.
|
||||
//!
|
||||
//! Listens on an L2 interface for ethertype `0x6969` frames, announces TCP
|
||||
//! upstreams to a discovering client, and relays bidirectional TCP traffic
|
||||
//! between the client and `127.0.0.1:<port>` services.
|
||||
|
||||
mod frame;
|
||||
mod link;
|
||||
mod session;
|
||||
mod upstream;
|
||||
|
||||
use clap::Parser;
|
||||
use std::process::ExitCode;
|
||||
use std::sync::atomic::{AtomicU32, Ordering};
|
||||
use std::sync::{Arc, Mutex};
|
||||
use tokio::net::TcpStream;
|
||||
use tokio::sync::mpsc;
|
||||
use tokio::sync::Mutex as AsyncMutex;
|
||||
use tracing::{error, Level};
|
||||
|
||||
use crate::frame::{
|
||||
Frame, REASON_CONNECT_FAILED, REASON_UNKNOWN_SESSION, REASON_UNKNOWN_UPSTREAM,
|
||||
};
|
||||
use crate::link::Link;
|
||||
use crate::session::{spawn_pump, write_to_session, SessionHandle, SessionStore};
|
||||
use crate::upstream::build_table;
|
||||
|
||||
#[derive(Parser)]
|
||||
#[command(name = "gatunad", version, about = "raw ethernet tunnel server")]
|
||||
struct Args {
|
||||
/// L2 interface name (e.g. eth0).
|
||||
iface: String,
|
||||
/// One or more TCP upstreams as PORT[:label], relayed to 127.0.0.1:PORT.
|
||||
#[arg(num_args = 1..)]
|
||||
ports: Vec<String>,
|
||||
}
|
||||
|
||||
#[tokio::main]
|
||||
async fn main() -> ExitCode {
|
||||
tracing_subscriber::fmt()
|
||||
.with_max_level(Level::ERROR)
|
||||
.with_writer(|| std::io::stdout())
|
||||
.init();
|
||||
|
||||
let args = Args::parse();
|
||||
|
||||
let table = match build_table(&args.ports) {
|
||||
Ok(t) => Arc::new(t),
|
||||
Err(e) => {
|
||||
eprintln!("gatunad: {e}");
|
||||
return ExitCode::FAILURE;
|
||||
}
|
||||
};
|
||||
|
||||
let link = match Link::open(&args.iface) {
|
||||
Ok(l) => Arc::new(l),
|
||||
Err(e) => {
|
||||
eprintln!("gatunad: failed to open interface {}: {e}", args.iface);
|
||||
return ExitCode::FAILURE;
|
||||
}
|
||||
};
|
||||
|
||||
let (tx, mut rx) = mpsc::channel::<(crate::link::MacAddr, Vec<u8>)>(1024);
|
||||
|
||||
// Tx task: sole owner of send-side writes, fed by all session pumps + rx.
|
||||
{
|
||||
let link = Arc::clone(&link);
|
||||
tokio::spawn(async move {
|
||||
while let Some((dst, frame)) = rx.recv().await {
|
||||
loop {
|
||||
let mut guard = match link.writable().await {
|
||||
Ok(g) => g,
|
||||
Err(e) => {
|
||||
error!("writable wait: {e}");
|
||||
break;
|
||||
}
|
||||
};
|
||||
match link.send(dst, &frame) {
|
||||
Ok(()) => break,
|
||||
Err(e) if e.kind() == std::io::ErrorKind::WouldBlock => {
|
||||
guard.clear_ready();
|
||||
continue;
|
||||
}
|
||||
Err(e) => {
|
||||
error!("send failed: {e}");
|
||||
break;
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
});
|
||||
}
|
||||
|
||||
let store: SessionStore = Arc::new(Mutex::new(std::collections::HashMap::new()));
|
||||
let next_id = Arc::new(AtomicU32::new(1));
|
||||
let mut buf = vec![0u8; 65536];
|
||||
|
||||
loop {
|
||||
let mut guard = match link.readable().await {
|
||||
Ok(g) => g,
|
||||
Err(e) => {
|
||||
error!("readable wait: {e}");
|
||||
continue;
|
||||
}
|
||||
};
|
||||
match link.recv(&mut buf) {
|
||||
Ok(Some((src, payload))) => {
|
||||
let frame = match Frame::parse(payload) {
|
||||
Ok(f) => f,
|
||||
Err(e) => {
|
||||
error!("decode from {src:?}: {e}");
|
||||
continue;
|
||||
}
|
||||
};
|
||||
handle_frame(frame, src, &tx, &store, &next_id, &table).await;
|
||||
}
|
||||
Ok(None) => {
|
||||
// Ignorable frame (outgoing/short/mismatch) or transient; do not
|
||||
// clear readiness — keep draining.
|
||||
}
|
||||
Err(e) if e.kind() == std::io::ErrorKind::WouldBlock => {
|
||||
guard.clear_ready();
|
||||
}
|
||||
Err(e) => {
|
||||
error!("recv: {e}");
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
async fn handle_frame(
|
||||
frame: Frame,
|
||||
src: crate::link::MacAddr,
|
||||
tx: &mpsc::Sender<(crate::link::MacAddr, Vec<u8>)>,
|
||||
store: &SessionStore,
|
||||
next_id: &Arc<AtomicU32>,
|
||||
table: &Arc<crate::upstream::UpstreamTable>,
|
||||
) {
|
||||
match frame {
|
||||
Frame::Discover => {
|
||||
let manifest = Frame::Manifest(table.entries());
|
||||
let _ = tx.send((src, manifest.encode())).await;
|
||||
}
|
||||
Frame::Open { upstream_id } => {
|
||||
let port = table.get(upstream_id).map(|u| u.port);
|
||||
match port {
|
||||
Some(port) => {
|
||||
// Spawn so connect() doesn't block the rx loop.
|
||||
let tx = tx.clone();
|
||||
let store = Arc::clone(store);
|
||||
let next = Arc::clone(next_id);
|
||||
tokio::spawn(async move {
|
||||
match TcpStream::connect(("127.0.0.1", port)).await {
|
||||
Ok(stream) => {
|
||||
let sid = next.fetch_add(1, Ordering::Relaxed);
|
||||
let (r, w) = stream.into_split();
|
||||
let w = Arc::new(AsyncMutex::new(w));
|
||||
store.lock().expect("store poisoned").insert(
|
||||
sid,
|
||||
SessionHandle {
|
||||
upstream_id,
|
||||
write: w,
|
||||
},
|
||||
);
|
||||
let ack = Frame::OpenAck { session_id: sid, upstream_id };
|
||||
let _ = tx.send((src, ack.encode())).await;
|
||||
spawn_pump(r, sid, src, tx, store);
|
||||
}
|
||||
Err(e) => {
|
||||
error!("connect 127.0.0.1:{port} failed: {e}");
|
||||
let nak =
|
||||
Frame::OpenNak { upstream_id, reason: REASON_CONNECT_FAILED };
|
||||
let _ = tx.send((src, nak.encode())).await;
|
||||
}
|
||||
}
|
||||
});
|
||||
}
|
||||
None => {
|
||||
let nak = Frame::OpenNak { upstream_id, reason: REASON_UNKNOWN_UPSTREAM };
|
||||
let _ = tx.send((src, nak.encode())).await;
|
||||
}
|
||||
}
|
||||
}
|
||||
Frame::Data { session_id, payload } => {
|
||||
match write_to_session(store, session_id, &payload).await {
|
||||
Ok(()) => {}
|
||||
Err(session::WriteError::UnknownSession) => {
|
||||
let close = Frame::Close {
|
||||
session_id,
|
||||
reason: Some(REASON_UNKNOWN_SESSION),
|
||||
};
|
||||
let _ = tx.send((src, close.encode())).await;
|
||||
}
|
||||
Err(session::WriteError::Io) => {
|
||||
store.lock().expect("store poisoned").remove(&session_id);
|
||||
let close = Frame::Close { session_id, reason: None };
|
||||
let _ = tx.send((src, close.encode())).await;
|
||||
}
|
||||
}
|
||||
}
|
||||
Frame::Close { session_id, reason: _ } => {
|
||||
store.lock().expect("store poisoned").remove(&session_id);
|
||||
}
|
||||
// Not expected from a client; ignore.
|
||||
Frame::Manifest(_) | Frame::OpenAck { .. } | Frame::OpenNak { .. } => {}
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,78 @@
|
||||
//! Per-session state and the localhost→tunnel pump.
|
||||
|
||||
use crate::frame::{Frame, MAX_PAYLOAD};
|
||||
use crate::link::MacAddr;
|
||||
use std::collections::HashMap;
|
||||
use std::sync::{Arc, Mutex};
|
||||
use tokio::io::{AsyncReadExt, AsyncWriteExt};
|
||||
use tokio::net::tcp::OwnedReadHalf;
|
||||
use tokio::sync::mpsc::Sender;
|
||||
use tokio::sync::Mutex as AsyncMutex;
|
||||
|
||||
pub type TxChan = Sender<(MacAddr, Vec<u8>)>;
|
||||
|
||||
#[allow(dead_code)]
|
||||
pub struct SessionHandle {
|
||||
pub upstream_id: u8,
|
||||
pub write: Arc<AsyncMutex<tokio::net::tcp::OwnedWriteHalf>>,
|
||||
}
|
||||
|
||||
pub type SessionStore = Arc<Mutex<HashMap<u32, SessionHandle>>>;
|
||||
|
||||
#[derive(Debug)]
|
||||
pub enum WriteError {
|
||||
UnknownSession,
|
||||
Io,
|
||||
}
|
||||
|
||||
pub async fn write_to_session(
|
||||
store: &SessionStore,
|
||||
id: u32,
|
||||
payload: &[u8],
|
||||
) -> Result<(), WriteError> {
|
||||
let write = {
|
||||
let store = store.lock().expect("store lock poisoned");
|
||||
store.get(&id).map(|h| h.write.clone())
|
||||
};
|
||||
match write {
|
||||
Some(w) => {
|
||||
let mut w = w.lock().await;
|
||||
w.write_all(payload).await.map_err(|_| WriteError::Io)?;
|
||||
Ok(())
|
||||
}
|
||||
None => Err(WriteError::UnknownSession),
|
||||
}
|
||||
}
|
||||
|
||||
/// Spawn the socket→tunnel pump: reads from the localhost TCP stream in
|
||||
/// 1480-byte chunks and emits DATA frames. On EOF/error sends CLOSE and
|
||||
/// removes the session from the store.
|
||||
pub fn spawn_pump(
|
||||
read: OwnedReadHalf,
|
||||
session_id: u32,
|
||||
peer_mac: MacAddr,
|
||||
tx: TxChan,
|
||||
store: SessionStore,
|
||||
) {
|
||||
tokio::spawn(async move {
|
||||
let mut buf = vec![0u8; MAX_PAYLOAD];
|
||||
loop {
|
||||
match read.read(&mut buf).await {
|
||||
Ok(0) => break,
|
||||
Ok(n) => {
|
||||
let frame = Frame::Data {
|
||||
session_id,
|
||||
payload: buf[..n].to_vec(),
|
||||
};
|
||||
if tx.send((peer_mac, frame.encode())).await.is_err() {
|
||||
break;
|
||||
}
|
||||
}
|
||||
Err(_) => break,
|
||||
}
|
||||
}
|
||||
let close = Frame::Close { session_id, reason: None };
|
||||
let _ = tx.send((peer_mac, close.encode())).await;
|
||||
store.lock().expect("store poisoned").remove(&session_id);
|
||||
});
|
||||
}
|
||||
@@ -0,0 +1,83 @@
|
||||
//! Upstream table: cmdline parsing and MANIFEST entry construction.
|
||||
|
||||
use crate::frame::{UpstreamEntry, PROTO_TCP, PROTO_UDP};
|
||||
|
||||
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
|
||||
pub enum Proto {
|
||||
Tcp,
|
||||
Udp,
|
||||
}
|
||||
impl Proto {
|
||||
pub fn as_u8(self) -> u8 {
|
||||
match self {
|
||||
Proto::Tcp => PROTO_TCP,
|
||||
Proto::Udp => PROTO_UDP,
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
#[derive(Clone, Debug)]
|
||||
pub struct Upstream {
|
||||
pub id: u8,
|
||||
pub proto: Proto,
|
||||
pub port: u16,
|
||||
pub label: Option<String>,
|
||||
}
|
||||
|
||||
pub struct UpstreamTable(pub Vec<Upstream>);
|
||||
|
||||
impl UpstreamTable {
|
||||
pub fn get(&self, id: u8) -> Option<&Upstream> {
|
||||
self.0.iter().find(|u| u.id == id)
|
||||
}
|
||||
pub fn entries(&self) -> Vec<UpstreamEntry> {
|
||||
self.0
|
||||
.iter()
|
||||
.map(|u| UpstreamEntry {
|
||||
id: u.id,
|
||||
proto: u.proto.as_u8(),
|
||||
port: u.port,
|
||||
label: u.label.clone(),
|
||||
})
|
||||
.collect()
|
||||
}
|
||||
}
|
||||
|
||||
/// Parse a single `PORT[:label]` argument. v1 only accepts bare integers or
|
||||
/// `int:label`; an explicit `proto:` prefix is reserved for the UDP extension.
|
||||
pub fn parse_port_arg(id: u8, s: &str) -> Result<Upstream, String> {
|
||||
let (port_str, label) = match s.split_once(':') {
|
||||
Some((p, l)) => (p, Some(l.to_string())),
|
||||
None => (s, None),
|
||||
};
|
||||
let port: u16 = port_str
|
||||
.parse()
|
||||
.map_err(|_| format!("invalid port value: {port_str}"))?;
|
||||
if port == 0 {
|
||||
return Err(format!("port must be > 0: {s}"));
|
||||
}
|
||||
if let Some(l) = &label {
|
||||
if l.is_empty() {
|
||||
return Err(format!("empty label: {s}"));
|
||||
}
|
||||
if l.len() > 255 {
|
||||
return Err(format!("label too long (max 255 bytes): {s}"));
|
||||
}
|
||||
}
|
||||
Ok(Upstream { id, proto: Proto::Tcp, port, label })
|
||||
}
|
||||
|
||||
pub fn build_table(args: &[String]) -> Result<UpstreamTable, String> {
|
||||
if args.is_empty() {
|
||||
return Err("at least one PORT is required".into());
|
||||
}
|
||||
if args.len() > 255 {
|
||||
return Err("too many upstreams (max 255)".into());
|
||||
}
|
||||
let mut v = Vec::with_capacity(args.len());
|
||||
for (i, a) in args.iter().enumerate() {
|
||||
let id = (i + 1) as u8;
|
||||
v.push(parse_port_arg(id, a)?);
|
||||
}
|
||||
Ok(UpstreamTable(v))
|
||||
}
|
||||
Reference in New Issue
Block a user