diff --git a/.gitignore b/.gitignore new file mode 100644 index 0000000..ea8c4bf --- /dev/null +++ b/.gitignore @@ -0,0 +1 @@ +/target diff --git a/PROTOCOL.md b/PROTOCOL.md new file mode 100644 index 0000000..5748e1b --- /dev/null +++ b/PROTOCOL.md @@ -0,0 +1,184 @@ +# gatuna wire protocol + +All frames ride Ethernet with ethertype `0x6969`. The ethertype is the sole +discriminator; there is no magic number inside the payload. + +## Frame layout + +``` + 0 1 2 3 + 0 1 2 3 4 5 6 7 8 9 0 1 2 3 4 5 6 7 8 9 0 1 2 3 4 5 6 7 8 9 0 1 ++---------------+---------------+-------------------------------+ +| version | type | session_id | ++---------------+---------------+-------------------------------+ +| payload ... | ++---------------------------------------------------------------+ +``` + +- **version** (u8): protocol version. Currently `1`. +- **type** (u8): frame type, see table below. +- **session_id** (u32, big-endian): `0` for non-session frames; the + server-assigned ID for session-scoped frames. +- **payload** (variable): type-dependent. No length field is carried — the + Ethernet frame length from the capture gives the payload extent. + +No CRC, no retransmit, no ordering at this layer. TCP reliability is handled by +the endpoints' TCP stacks; UDP (reserved) will rely on application-level +mechanisms. + +Maximum payload: 1500 (Ethernet MTU) − 14 (eth header) − 6 (our header) = +**1480 bytes**. Larger payloads are not emitted in v1. + +## Frame types + +| Type | Name | Direction | session_id | Payload | +|------|-------------|---------------|------------|----------------------------------| +| 0x01 | DISCOVER | C → broadcast | 0 | empty | +| 0x02 | MANIFEST | S → C | 0 | `id:1, proto:1, port:2` × N | +| 0x03 | OPEN | C → S | 0 | `upstream_id:1` | +| 0x04 | OPEN_ACK | S → C | assigned | `upstream_id:1` | +| 0x05 | OPEN_NAK | S → C | 0 | `upstream_id:1, reason:1` | +| 0x06 | DATA | both | session | raw bytes (≤1480) | +| 0x07 | CLOSE | both | session | optional `reason:1` | + +Reserved (unimplemented in v1; parse returns Err, encode unimplemented): + +| Type | Name | +|------|-------------| +| 0x08 | UDP_OPEN | +| 0x09 | UDP_DATA | +| 0x0A | UDP_CLOSE | + +## Payload field encodings + +### MANIFEST payload + +Variable-length entries, parsed sequentially until the payload is consumed. + +``` + 0 1 2 3 + 0 1 2 3 4 5 6 7 8 9 0 1 2 3 4 5 6 7 8 9 0 1 2 3 4 5 6 7 8 9 0 1 ++---------------+---------------+-------------------------------+ +| id | proto | port (big-endian) | ++---------------+---------------+-------------------------------+ +| label_len | label (UTF-8, label_len bytes) ... | ++---------------+-----------------------------------------------+ +``` + +- **id** (u8): upstream identifier (1-based positional index from `gatunad` + cmdline). +- **proto** (u8): `1 = TCP`, `2 = UDP` (reserved; not emitted in v1). +- **port** (u16, big-endian): the real port on the server's `127.0.0.1`. +- **label_len** (u8): length in bytes of the label that follows. `0` means no + label. +- **label** (`label_len` bytes, UTF-8): human-readable name for the upstream, + taken from the `PORT[:label]` cmdline argument. Maximum 255 bytes. + +To parse: read the 5-byte fixed prefix, then `label_len` bytes, and repeat until +the payload is exhausted. The number of entries is not carried explicitly. + +### OPEN payload + +``` ++---------------+ +| upstream_id | ++---------------+ +``` + +- **upstream_id** (u8): which MANIFEST entry to open. + +### OPEN_ACK payload + +``` ++---------------+ +| upstream_id | ++---------------+ +``` + +- **upstream_id** (u8): echoes the requested upstream. The session is + identified by the `session_id` field in the header, not the payload. + +### OPEN_NAK payload + +``` ++---------------+---------------+ +| upstream_id | reason | ++---------------+---------------+ +``` + +- **upstream_id** (u8): echoes the requested upstream. +- **reason** (u8): see reason codes. + +### DATA payload + +Raw application bytes. Up to 1480 bytes per frame. The `session_id` header +field identifies which session the bytes belong to. + +### CLOSE payload + +``` ++---------------+ +| reason? | ++---------------+ +``` + +- **reason** (u8, optional): present iff payload length ≥ 1. See reason codes. + +## Reason codes + +| Value | Meaning | +|-------|-------------------| +| 0 | unspecified | +| 1 | unknown_upstream | +| 2 | connect_failed | +| 3 | oversize | +| 4 | unknown_session | + +## Discovery flow + +``` +client server + | | + | DISCOVER (dst = broadcast) | + |-------------------------------->| + | | + | MANIFEST (unicast) | + |<--------------------------------| + | | +``` + +The server learns the client's MAC from the DISCOVER frame's source address and +unicasts the MANIFEST back. The server never speaks unsolicited. + +## TCP session lifecycle + +``` +client server + | | + | OPEN { upstream_id } | + |-------------------------------->| + | | TcpStream::connect(127.0.0.1:port) + | | + | OPEN_ACK { session_id } | + |<--------------------------------| (on success) + | OR | + | OPEN_NAK { reason } | + |<--------------------------------| (on failure) + | | + | DATA { session_id, bytes } | + |<------------------------------->| DATA { session_id, bytes } + | | + | CLOSE { session_id, reason? } | + |<------------------------------->| (on EOF, RST, or error) + | | +``` + +- `session_id` is allocated by the server as a monotonically increasing u32 + (starting at 1) from an atomic counter. Collision by wraparound is ignored. +- Either side may send `CLOSE`. The side receiving `CLOSE` tears down its half + and stops emitting frames for that session. +- The server's socket→tunnel pump reads `TcpStream` in 1480-byte chunks and + emits one `DATA` frame per chunk. On `read` returning 0 (FIN) or an error, + it emits `CLOSE` and exits. +- The server's tunnel→socket path writes `DATA` payloads to the `TcpStream` + with `write_all`. On error it emits `CLOSE` and drops the session. diff --git a/README.md b/README.md index e65ee14..bdf1c7a 100644 --- a/README.md +++ b/README.md @@ -1,3 +1,88 @@ # gatuna -raw ethernet tunnel to circumvent WFP killswitches \ No newline at end of file +A raw-Ethernet tunnel for reaching services on a peer machine when a WFP +killswitch on the local machine blocks normal IP traffic. Two programs carry +bytes between their respective loopbacks over a private L2 protocol, bypassing +the IP stack (and therefore the killswitch) entirely. + +## Why this works + +WireGuard-for-Windows installs its killswitch as WFP filters at the ALE layers +only (`ALE_AUTH_CONNECT_V4/V6`, `ALE_AUTH_RECV_ACCEPT_V4/V6`). It never installs +MAC-layer callouts — the L2 code in `tunnel/firewall/` is commented out. Raw +Ethernet frames sent via a packet driver (Npcap on Windows, `AF_PACKET` on +Linux) never enter the IP stack, so they never reach the ALE classify path and +pass unimpeded. See `wireguard-windows/tunnel/firewall/blocker.go:156`. + +## Components + +- `gatunad` — Rust server. Runs on the peer (Linux) that owns the real + services. Announces upstreams and relays TCP between the tunnel and + `127.0.0.1:`. +- `gatuna` (planned) — .NET WinForms client. Runs on the killswitched Windows + box. Discovers the server, presents its upstreams as local loopback + listeners, and hauls bytes over the same L2 protocol. + +This repository builds `gatunad` first. + +## Transport + +- **Medium:** raw Ethernet frames on a shared L2 segment. +- **Ethertype:** `0x6969` (hardcoded). +- **No IP stack involvement.** Frames carry only our 6-byte header + payload. +- **BPF:** the server attaches a classic BPF filter `ether proto 0x6969` to its + `AF_PACKET` socket so it only wakes on our ethertype. No eBPF authoring. + +## Protocol + +See [`PROTOCOL.md`](PROTOCOL.md) for the full wire format. Summary: + +- 6-byte header, big-endian: `[ version:1 ][ type:1 ][ session_id:4 ][ payload:N ]`. +- `version` = `1`. No length field (frame length comes from the capture). No + CRC. Ethertype discriminates our frames from everything else. +- Discovery: client broadcasts `DISCOVER`; server unicasts `MANIFEST` back. +- Sessions: `OPEN` → `OPEN_ACK` (or `OPEN_NAK`) → `DATA`* ↔ `DATA`* → `CLOSE`. +- v1 ships TCP only. UDP frame types are reserved but unimplemented. + +## `gatunad` usage + +``` +gatunad IFACE PORT [PORT...] +``` + +- `IFACE` — L2 interface name (e.g. `eth0`). +- `PORT` — `int[:label]`; a TCP upstream relayed to `127.0.0.1:int`. The label + is optional, max 255 bytes, carried in the MANIFEST for the client to display. + At least one required. Upstream ID = 1-based positional index in cmdline order. + +Example: +``` +sudo ./gatunad eth0 22 8800:site-a 8080:site-b +``` +Exposes upstreams `1→127.0.0.1:22`, `2→127.0.0.1:8800` (label `site-a`), +`3→127.0.0.1:8080` (label `site-b`). + +## Privileges + +`AF_PACKET` requires `CAP_NET_RAW`. Run as root, or grant the binary the +capability once: +``` +sudo setcap cap_net_raw+ep ./target/release/gatunad +``` + +## Logging + +Errors only, to stdout. Normal lifecycle (DISCOVER/OPEN/CLOSE) is silent. + +## v1 limitations + +- TCP only. UDP wire types reserved, code paths stubbed. +- No retransmit at the L2 layer. A dropped `DATA` frame breaks the TCP session + irrecoverably because the localhost socket already ACKed the bytes. Acceptable + on a healthy switched link. +- No auth/crypto. Anyone on the same L2 segment can `DISCOVER` and `OPEN`. +- Single server instance per interface. + +## License + +CC0 1.0 Universal. See [`LICENSE`](LICENSE). diff --git a/gatunad/Cargo.toml b/gatunad/Cargo.toml new file mode 100644 index 0000000..4de11b8 --- /dev/null +++ b/gatunad/Cargo.toml @@ -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" diff --git a/gatunad/src/frame.rs b/gatunad/src/frame.rs new file mode 100644 index 0000000..13d5e88 --- /dev/null +++ b/gatunad/src/frame.rs @@ -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, +} + +#[derive(Clone, Debug)] +pub enum Frame { + Discover, + Manifest(Vec), + Open { upstream_id: u8 }, + OpenAck { session_id: u32, upstream_id: u8 }, + OpenNak { upstream_id: u8, reason: u8 }, + Data { session_id: u32, payload: Vec }, + Close { session_id: u32, reason: Option }, +} + +#[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, 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 { + 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 { + 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)), + } + } +} diff --git a/gatunad/src/link.rs b/gatunad/src/link.rs new file mode 100644 index 0000000..8ec2bb5 --- /dev/null +++ b/gatunad/src/link.rs @@ -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, + 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 { + 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 { + 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::() 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::() 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> { + self.fd.readable().await + } + + pub async fn writable(&self) -> io::Result> { + 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> { + let mut sll: libc::sockaddr_ll = unsafe { std::mem::zeroed() }; + let mut slen = std::mem::size_of::() 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::() as libc::socklen_t, + ) + }; + if r < 0 { + return Err(io::last_os_error()); + } + Ok(()) + } +} diff --git a/gatunad/src/main.rs b/gatunad/src/main.rs new file mode 100644 index 0000000..69d4622 --- /dev/null +++ b/gatunad/src/main.rs @@ -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:` 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, +} + +#[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)>(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)>, + store: &SessionStore, + next_id: &Arc, + table: &Arc, +) { + 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 { .. } => {} + } +} diff --git a/gatunad/src/session.rs b/gatunad/src/session.rs new file mode 100644 index 0000000..c2d9e22 --- /dev/null +++ b/gatunad/src/session.rs @@ -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)>; + +#[allow(dead_code)] +pub struct SessionHandle { + pub upstream_id: u8, + pub write: Arc>, +} + +pub type SessionStore = Arc>>; + +#[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); + }); +} diff --git a/gatunad/src/upstream.rs b/gatunad/src/upstream.rs new file mode 100644 index 0000000..c14ba09 --- /dev/null +++ b/gatunad/src/upstream.rs @@ -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, +} + +pub struct UpstreamTable(pub Vec); + +impl UpstreamTable { + pub fn get(&self, id: u8) -> Option<&Upstream> { + self.0.iter().find(|u| u.id == id) + } + pub fn entries(&self) -> Vec { + 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 { + 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 { + 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)) +} diff --git a/win/gatuna-client/Frame.cs b/win/gatuna-client/Frame.cs new file mode 100644 index 0000000..53d02cb --- /dev/null +++ b/win/gatuna-client/Frame.cs @@ -0,0 +1,156 @@ +namespace gatuna_client; + +using System.Text; + +static class Proto +{ + public const ushort EtherType = 0x6969; + public const int EthHeaderLen = 14; + public const byte Version = 1; + public const int MaxPayload = 1480; + + public const byte TypeDiscover = 0x01; + public const byte TypeManifest = 0x02; + public const byte TypeOpen = 0x03; + public const byte TypeOpenAck = 0x04; + public const byte TypeOpenNak = 0x05; + public const byte TypeData = 0x06; + public const byte TypeClose = 0x07; + + public const byte ProtoTcp = 1; + public const byte ProtoUdp = 2; + + public const byte ReasonUnspecified = 0; + public const byte ReasonUnknownUpstream = 1; + public const byte ReasonConnectFailed = 2; + public const byte ReasonOversize = 3; + public const byte ReasonUnknownSession = 4; +} + +readonly record struct UpstreamEntry( + byte Id, + byte Protocol, + ushort Port, + string? Label) +{ + public string ProtoName => Protocol == Proto.ProtoTcp ? "tcp" : "udp"; +} + +abstract record Frame +{ + internal record Discover : Frame; + internal record Manifest(UpstreamEntry[] Entries) : Frame; + internal record Open(byte UpstreamId) : Frame; + internal record OpenAck(uint SessionId, byte UpstreamId) : Frame; + internal record OpenNak(byte UpstreamId, byte Reason) : Frame; + internal record Data(uint SessionId, byte[] Payload) : Frame; + internal record Close(uint SessionId, byte? Reason) : Frame; +} + +static class FrameCodec +{ + public static byte[] Encode(Frame frame) + { + return frame switch + { + Frame.Discover => + Header(Proto.TypeDiscover, 0), + Frame.Manifest manifest => + BuildManifest(manifest.Entries), + Frame.Open open => + [.. Header(Proto.TypeOpen, 0), open.UpstreamId], + Frame.OpenAck ack => + [.. Header(Proto.TypeOpenAck, ack.SessionId), ack.UpstreamId], + Frame.OpenNak nak => + [.. Header(Proto.TypeOpenNak, 0), nak.UpstreamId, nak.Reason], + Frame.Data data => + [.. Header(Proto.TypeData, data.SessionId), .. data.Payload], + Frame.Close close => + close.Reason.HasValue + ? [.. Header(Proto.TypeClose, close.SessionId), close.Reason.Value] + : Header(Proto.TypeClose, close.SessionId), + _ => throw new InvalidOperationException($"unknown frame type: {frame.GetType()}"), + }; + } + + static byte[] BuildManifest(UpstreamEntry[] entries) + { + using var ms = new MemoryStream(); + ms.Write(Header(Proto.TypeManifest, 0)); + foreach (var e in entries) + { + var labelBytes = Encoding.UTF8.GetBytes(e.Label ?? ""); + var labelLen = (byte)Math.Min(labelBytes.Length, 255); + ms.WriteByte(e.Id); + ms.WriteByte(e.Protocol); + ms.WriteByte((byte)(e.Port >> 8)); + ms.WriteByte((byte)(e.Port & 0xFF)); + ms.WriteByte(labelLen); + if (labelLen > 0) + ms.Write(labelBytes, 0, labelLen); + } + return ms.ToArray(); + } + + static byte[] Header(byte type, uint sessionId) + { + return [ + Proto.Version, type, + (byte)(sessionId >> 24), + (byte)(sessionId >> 16), + (byte)(sessionId >> 8), + (byte)(sessionId & 0xFF), + ]; + } + + public static Frame? Parse(ReadOnlySpan buf) + { + if (buf.Length < 6) + return null; + if (buf[0] != Proto.Version) + return null; + var type = buf[1]; + var sessionId = (uint)(buf[2] << 24 | buf[3] << 16 | buf[4] << 8 | buf[5]); + var payload = buf[6..]; + + return type switch + { + Proto.TypeManifest => ParseManifest(payload), + Proto.TypeOpenAck when payload.Length == 1 => + new Frame.OpenAck(sessionId, payload[0]), + Proto.TypeOpenNak when payload.Length == 2 => + new Frame.OpenNak(payload[0], payload[1]), + Proto.TypeData when payload.Length <= Proto.MaxPayload => + new Frame.Data(sessionId, payload.ToArray()), + Proto.TypeClose when payload.Length is 0 or 1 => + new Frame.Close(sessionId, payload.Length == 1 ? payload[0] : null), + Proto.TypeDiscover when payload.Length == 0 => + new Frame.Discover(), + _ => null, + }; + } + + static Frame.Manifest? ParseManifest(ReadOnlySpan payload) + { + var entries = new List(); + int i = 0; + while (i < payload.Length) + { + if (i + 5 > payload.Length) + return null; + var id = payload[i]; + var proto = payload[i + 1]; + var port = (ushort)(payload[i + 2] << 8 | payload[i + 3]); + var labelLen = payload[i + 4]; + i += 5; + if (i + labelLen > payload.Length) + return null; + string? label = labelLen == 0 + ? null + : Encoding.UTF8.GetString(payload[i..(i + labelLen)]); + i += labelLen; + entries.Add(new UpstreamEntry(id, proto, port, label)); + } + return new Frame.Manifest(entries.ToArray()); + } +} diff --git a/win/gatuna-client/MainForm.cs b/win/gatuna-client/MainForm.cs new file mode 100644 index 0000000..b8b3382 --- /dev/null +++ b/win/gatuna-client/MainForm.cs @@ -0,0 +1,141 @@ +using SharpPcap.LibPcap; + +namespace gatuna_client; + +public partial class MainForm : Form +{ + readonly SessionManager _sessions = new(); + readonly ComboBox _deviceBox = new(); + readonly Button _discoverBtn = new(); + readonly ListView _listView = new(); + readonly Label _statusLabel = new(); + TunnelLink? _link; + + public MainForm() + { + Text = "gatuna"; + Width = 520; + Height = 380; + FormBorderStyle = FormBorderStyle.FixedSingle; + MaximizeBox = false; + MinimizeBox = false; + StartPosition = FormStartPosition.CenterScreen; + InitializeComponents(); + + _sessions.Log += msg => this.Invoke(() => _statusLabel.Text = msg); + _sessions.ManifestReceived += entries => this.Invoke(() => PopulateList(entries)); + + foreach (var d in TunnelLink.ListDevices()) + _deviceBox.Items.Add($"{d.Name} — {d.Interface?.FriendlyName ?? d.Interface?.Description}"); + if (_deviceBox.Items.Count > 0) + _deviceBox.SelectedIndex = 0; + } + + void InitializeComponents() + { + Controls.Add(new Label { Text = "Adapter:", Left = 12, Top = 12, AutoSize = true }); + + _deviceBox.Left = 70; _deviceBox.Top = 9; + _deviceBox.Width = 330; _deviceBox.DropDownStyle = ComboBoxStyle.DropDownList; + Controls.Add(_deviceBox); + + _discoverBtn.Text = "Discover"; + _discoverBtn.Left = 410; _discoverBtn.Top = 8; _discoverBtn.Width = 80; + _discoverBtn.Click += OnDiscover; + Controls.Add(_discoverBtn); + + _listView.Left = 12; _listView.Top = 40; + _listView.Width = 478; _listView.Height = 250; + _listView.View = View.Details; + _listView.FullRowSelect = true; + _listView.CheckBoxes = true; + _listView.Columns.Add("ID", 36); + _listView.Columns.Add("Proto", 50); + _listView.Columns.Add("Port", 56); + _listView.Columns.Add("Label", 160); + _listView.Columns.Add("Mirror", 70); + _listView.ItemChecked += OnItemChecked; + Controls.Add(_listView); + + _statusLabel.Left = 12; _statusLabel.Top = 304; + _statusLabel.Width = 478; _statusLabel.AutoEllipsis = true; + Controls.Add(_statusLabel); + } + + void OnDiscover(object? s, EventArgs e) + { + if (_deviceBox.SelectedIndex < 0) + { + MessageBox.Show("Select a network adapter first."); + return; + } + + if (_link != null) + { + _sessions.StopAll(); + _sessions.DetachLink(); + _link.Dispose(); + } + + var devices = TunnelLink.ListDevices(); + var device = devices[_deviceBox.SelectedIndex]; + _link = new TunnelLink(device); + _link.Log += msg => this.Invoke(() => _statusLabel.Text = msg); + _sessions.AttachLink(_link); + _link.Open(); + _sessions.Discover(); + _statusLabel.Text = "discovering..."; + } + + void PopulateList(UpstreamEntry[] entries) + { + _listView.BeginUpdate(); + _listView.Items.Clear(); + foreach (var up in entries) + { + var item = new ListViewItem(up.Id.ToString()); + item.SubItems.Add(up.ProtoName); + item.SubItems.Add(up.Port.ToString()); + item.SubItems.Add(up.Label ?? ""); + item.SubItems.Add(""); + item.Tag = up; + _listView.Items.Add(item); + } + _listView.EndUpdate(); + } + + void OnItemChecked(object? s, ItemCheckedEventArgs e) + { + if (e.Item.Tag is not UpstreamEntry up) return; + if (e.Item.Checked) + { + var port = _sessions.StartListener(up); + e.Item.SubItems[4].Text = port.ToString(); + } + else + { + if (int.TryParse(e.Item.SubItems[4].Text, out var port) && port > 0) + { + _sessions.StopListener(port); + e.Item.SubItems[4].Text = ""; + } + } + } + + protected override void OnFormClosing(FormClosingEventArgs e) + { + if (e.CloseReason == CloseReason.UserClosing) + { + e.Cancel = true; + Hide(); + return; + } + base.OnFormClosing(e); + } + + public void Shutdown() + { + _sessions.Dispose(); + _link?.Dispose(); + } +} diff --git a/win/gatuna-client/Program.cs b/win/gatuna-client/Program.cs new file mode 100644 index 0000000..caef5dc --- /dev/null +++ b/win/gatuna-client/Program.cs @@ -0,0 +1,40 @@ +namespace gatuna_client; + +static class Program +{ + [STAThread] + static void Main() + { + ApplicationConfiguration.Initialize(); + + var form = new MainForm(); + + using var tray = new NotifyIcon + { + Icon = SystemIcons.Application, + Text = "gatuna", + Visible = true, + }; + + tray.ContextMenuStrip = new ContextMenuStrip(); + tray.ContextMenuStrip.Items.Add("Show", null, (_, _) => + { + form.Show(); + form.Activate(); + }); + tray.ContextMenuStrip.Items.Add("-"); + tray.ContextMenuStrip.Items.Add("Exit", null, (_, _) => + { + form.Shutdown(); + tray.Visible = false; + Application.Exit(); + }); + tray.DoubleClick += (_, _) => + { + form.Show(); + form.Activate(); + }; + + Application.Run(form); + } +} diff --git a/win/gatuna-client/SessionManager.cs b/win/gatuna-client/SessionManager.cs new file mode 100644 index 0000000..68a7631 --- /dev/null +++ b/win/gatuna-client/SessionManager.cs @@ -0,0 +1,284 @@ +using System.Collections.Concurrent; +using System.Net; +using System.Net.Sockets; +using System.Threading.Channels; + +namespace gatuna_client; + +sealed class SessionManager : IDisposable +{ + TunnelLink? _link; + readonly ConcurrentDictionary _sessions = new(); + readonly ConcurrentDictionary _listeners = new(); + + byte[]? _serverMac; + UpstreamEntry[] _upstreams = []; + + // Serialized OPEN: only one outstanding at a time. + readonly object _openLock = new(); + PendingOpen? _pending; + readonly Queue _openQueue = new(); + + public event Action? Log; + public event Action? ManifestReceived; + + public UpstreamEntry[] Upstreams => _upstreams; + public byte[]? ServerMac => _serverMac; + public TunnelLink? Link => _link; + + public void AttachLink(TunnelLink link) + { + _link = link; + link.FrameReceived += HandleFrame; + } + + public void DetachLink() + { + if (_link != null) + _link.FrameReceived -= HandleFrame; + _link = null; + } + + public void Discover() + { + if (_link == null) return; + _link.SendBroadcast(new Frame.Discover()); + } + + public void HandleFrame(Frame frame, byte[] srcMac) + { + switch (frame) + { + case Frame.Manifest manifest: + _serverMac = srcMac; + _upstreams = manifest.Entries; + Log?.Invoke($"manifest: {manifest.Entries.Length} upstreams from {BitConverter.ToString(srcMac)}"); + ManifestReceived?.Invoke(manifest.Entries); + break; + + case Frame.OpenAck ack: + HandleOpenAck(ack, srcMac); + break; + + case Frame.OpenNak nak: + Log?.Invoke($"OPEN_NAK upstream {nak.UpstreamId} reason {nak.Reason}"); + lock (_openLock) + { + if (_pending != null) + _pending.Client.Dispose(); + } + ProcessQueue(); + break; + + case Frame.Data data: + if (_sessions.TryGetValue(data.SessionId, out var session)) + session.Deliver(data.Payload); + else if (_link != null && _serverMac != null) + _link.SendTo(_serverMac, + new Frame.Close(data.SessionId, Proto.ReasonUnknownSession)); + break; + + case Frame.Close close: + if (_sessions.TryRemove(close.SessionId, out var s)) + s.Dispose(); + break; + } + } + + /// + /// Start a local TCP listener for the given upstream. Returns the mirror + /// port, or 0 on failure. + /// + public int StartListener(UpstreamEntry upstream) + { + var listener = new TcpListener(IPAddress.Loopback, 0); + listener.Start(); + var port = ((IPEndPoint)listener.LocalEndpoint).Port; + var state = new ListenerState(listener, upstream); + _listeners[port] = state; + _ = AcceptLoop(state); + Log?.Invoke($"listening 127.0.0.1:{port} -> upstream {upstream.Id} ({upstream.ProtoName}:{upstream.Port})"); + return port; + } + + public void StopListener(int port) + { + if (_listeners.TryRemove(port, out var state)) + { + state.Listener.Stop(); + Log?.Invoke($"stopped listener port {port}"); + } + } + + async Task AcceptLoop(ListenerState state) + { + while (true) + { + TcpClient client; + try + { + client = await state.Listener.AcceptTcpClientAsync(); + } + catch { break; } + EnqueueOpen(client, state.Upstream.Id); + } + } + + void EnqueueOpen(TcpClient client, byte upstreamId) + { + lock (_openLock) + { + if (_pending == null) + { + _pending = new PendingOpen(client, upstreamId); + SendOpen(_pending); + } + else + { + _openQueue.Enqueue(new PendingOpen(client, upstreamId)); + } + } + } + + void SendOpen(PendingOpen po) + { + if (_link == null || _serverMac == null) + { + Log?.Invoke("no server; cannot OPEN"); + po.Client.Dispose(); + return; + } + _link.SendTo(_serverMac, new Frame.Open(po.UpstreamId)); + } + + void ProcessQueue() + { + lock (_openLock) + { + if (_openQueue.Count > 0) + { + _pending = _openQueue.Dequeue(); + SendOpen(_pending); + } + else + { + _pending = null; + } + } + } + + void HandleOpenAck(Frame.OpenAck ack, byte[] srcMac) + { + PendingOpen? po; + lock (_openLock) + po = _pending; + + if (po == null || po.UpstreamId != ack.UpstreamId) + { + Log?.Invoke($"OPEN_ACK upstream {ack.UpstreamId} session {ack.SessionId} — no matching pending"); + return; + } + + var session = new Session( + ack.SessionId, po.Client, srcMac, _link!, + () => _sessions.TryRemove(ack.SessionId, out _), + msg => Log?.Invoke(msg)); + _sessions[ack.SessionId] = session; + session.Start(); + Log?.Invoke($"session {ack.SessionId} upstream {ack.UpstreamId} established"); + ProcessQueue(); + } + + public void StopAll() + { + foreach (var kv in _listeners) + kv.Value.Listener.Stop(); + _listeners.Clear(); + foreach (var s in _sessions.Values) + s.Dispose(); + _sessions.Clear(); + } + + public void Dispose() + { + DetachLink(); + StopAll(); + } +} + +sealed class ListenerState(TcpListener listener, UpstreamEntry upstream) +{ + public TcpListener Listener { get; } = listener; + public UpstreamEntry Upstream { get; } = upstream; +} + +sealed class PendingOpen(TcpClient client, byte upstreamId) +{ + public TcpClient Client { get; } = client; + public byte UpstreamId { get; } = upstreamId; +} + +sealed class Session( + uint sessionId, + TcpClient client, + byte[] serverMac, + TunnelLink link, + Action onClosed, + Action? log) : IDisposable +{ + readonly CancellationTokenSource _cts = new(); + readonly Channel _incoming = Channel.CreateBounded(256); + + public void Start() + { + _ = PumpSocketToTunnel(); + _ = PumpTunnelToSocket(); + } + + public void Deliver(byte[] payload) + { + if (!_incoming.Writer.TryWrite(payload)) + log?.Invoke($"session {sessionId}: incoming channel full"); + } + + async Task PumpSocketToTunnel() + { + try + { + var stream = client.GetStream(); + var buf = new byte[Proto.MaxPayload]; + using var reg = _cts.Token.Register(() => client.Dispose()); + while (!_cts.IsCancellationRequested) + { + var n = await stream.ReadAsync(buf, _cts.Token); + if (n == 0) break; + link.SendTo(serverMac, new Frame.Data(sessionId, buf[..n])); + } + } + catch { } + SendClose(); + onClosed(); + } + + async Task PumpTunnelToSocket() + { + try + { + var stream = client.GetStream(); + await foreach (var payload in _incoming.Reader.ReadAllAsync(_cts.Token)) + await stream.WriteAsync(payload, _cts.Token); + } + catch { } + } + + void SendClose() => link.SendTo(serverMac, new Frame.Close(sessionId, null)); + + public void Dispose() + { + _cts.Cancel(); + _incoming.Writer.TryComplete(); + SendClose(); + try { client.Dispose(); } catch { } + _cts.Dispose(); + } +} diff --git a/win/gatuna-client/TunnelLink.cs b/win/gatuna-client/TunnelLink.cs new file mode 100644 index 0000000..12b3dfa --- /dev/null +++ b/win/gatuna-client/TunnelLink.cs @@ -0,0 +1,99 @@ +using SharpPcap; +using SharpPcap.LibPcap; + +namespace gatuna_client; + +sealed class TunnelLink : IDisposable +{ + LibPcapLiveDevice _device; + readonly byte[] _ourMac = new byte[6]; + + public event Action? FrameReceived; + public event Action? Log; + + public TunnelLink(LibPcapLiveDevice device) + { + _device = device; + } + + public static LibPcapLiveDevice[] ListDevices() + { + return [.. LibPcapLiveDeviceList.Instance + .Where(d => !d.Loopback && d.MacAddress != null)]; + } + + public void Open() + { + _device.Open(new DeviceConfiguration + { + Mode = DeviceModes.Promiscuous | DeviceModes.MaxResponsiveness, + ReadTimeout = 1000, + }); + _device.Filter = $"ether proto 0x{Proto.EtherType:X4}"; + + if (_device.MacAddress?.GetAddressBytes() is { Length: 6 } mac) + { + Buffer.BlockCopy(mac, 0, _ourMac, 0, 6); + } + + _device.OnPacketArrival += OnPacketArrival; + _device.StartCapture(); + Log?.Invoke($"capture started on {_device.Name}"); + } + + void OnPacketArrival(object? sender, PacketCapture capture) + { + var raw = capture.GetPacket(); + var data = raw.Data; + if (data.Length < Proto.EthHeaderLen + 6) + return; + var et = (ushort)(data[12] << 8 | data[13]); + if (et != Proto.EtherType) + return; + var srcMac = new byte[6]; + Buffer.BlockCopy(data, 6, srcMac, 0, 6); + var payload = data.AsSpan(Proto.EthHeaderLen); + if (FrameCodec.Parse(payload) is { } frame) + FrameReceived?.Invoke(frame, srcMac); + } + + public void SendBroadcast(Frame frame) + { + SendRaw([0xFF, 0xFF, 0xFF, 0xFF, 0xFF, 0xFF], FrameCodec.Encode(frame)); + } + + public void SendTo(byte[] dstMac, Frame frame) + { + SendRaw(dstMac, FrameCodec.Encode(frame)); + } + + void SendRaw(byte[] dstMac, byte[] payload) + { + var frame = new byte[Proto.EthHeaderLen + payload.Length]; + Buffer.BlockCopy(dstMac, 0, frame, 0, 6); + Buffer.BlockCopy(_ourMac, 0, frame, 6, 6); + frame[12] = (byte)(Proto.EtherType >> 8); + frame[13] = (byte)(Proto.EtherType & 0xFF); + Buffer.BlockCopy(payload, 0, frame, 14, payload.Length); + try + { + _device.SendPacket(frame); + } + catch (Exception ex) + { + Log?.Invoke($"send failed: {ex.Message}"); + } + } + + public void Dispose() + { + try + { + if (_device.Started) + _device.StopCapture(); + _device.OnPacketArrival -= OnPacketArrival; + _device.Close(); + } + catch { } + } +} diff --git a/win/gatuna-client/app.manifest b/win/gatuna-client/app.manifest new file mode 100644 index 0000000..cd4567a --- /dev/null +++ b/win/gatuna-client/app.manifest @@ -0,0 +1,11 @@ + + + + + + + + + + + diff --git a/win/gatuna-client/gatuna-client.csproj b/win/gatuna-client/gatuna-client.csproj new file mode 100644 index 0000000..4caada3 --- /dev/null +++ b/win/gatuna-client/gatuna-client.csproj @@ -0,0 +1,17 @@ + + + + WinExe + net8.0-windows + gatuna_client + enable + true + enable + app.manifest + + + + + + +