diff --git a/.github/workflows/silkd.yml b/.github/workflows/silkd.yml index 0fcd070c3..497097262 100644 --- a/.github/workflows/silkd.yml +++ b/.github/workflows/silkd.yml @@ -36,4 +36,4 @@ jobs: - name: root-only tests working-directory: silkd - run: sudo -E env "PATH=$PATH" "HOME=$HOME" cargo test --locked --test git_e2e --test fs_e2e --test find_e2e -- --ignored + run: sudo -E env "PATH=$PATH" "HOME=$HOME" cargo test --locked --test git_e2e --test fs_e2e --test find_e2e --test tun_e2e -- --ignored --skip throughput diff --git a/docs/advanced-network.md b/docs/advanced-network.md new file mode 100644 index 000000000..e32880dc7 --- /dev/null +++ b/docs/advanced-network.md @@ -0,0 +1,143 @@ +# Advanced network mode + +The **default network mode** of a none-lane guest is the proxy path: silkd +relays `127.0.0.1:3128` and `127.0.0.1:1080` to the host's +[guarded-egress](egress.md) proxy over vsock. The **advanced network mode** +adds a full L3 path beside it for clients the proxy cannot carry (UDP, QUIC, +WebRTC, clients that ignore proxy variables): silkd creates `tun0` in the +guest and carries raw IP packets over one vsock stream to the host. The guest +still has no NIC, so hibernate, archive, wake and move work as before. + +This page is the contract between silkd and the host. The host side design is +[#471](https://github.com/cocoonstack/sandbox/issues/471). + +## Guest side + +| Item | Value | +| --- | --- | +| Device | `tun0`, `IFF_TUN \| IFF_NO_PI \| IFF_VNET_HDR`, owned by silkd | +| Address | `100.64.0.2/30` in every guest; the host tells guests apart by their vsock stream, never by address | +| Gateway | `100.64.0.1`, default route via `tun0` | +| MTU | 1500 until the host hello names another | +| IPv6 | none; silkd turns IPv6 off on `tun0` | +| Packet stream | guest dials host vsock port **2051** (`_2051` on the host) | + +Both modes run at once. silkd keeps the proxy variables in every exec, so a +client that honours them still reaches `127.0.0.1:3128` over loopback; all +other traffic takes the default route through `tun0`. Loopback never enters +`tun0`, so the guest DNS stub of #472 is unaffected; the tun path does not +intercept DNS. + +### Turning it on and off + +- **`tun_up`** (control channel, port 2048): creates `tun0` with its address + and route, starts the packet stream, answers `done` or an `error` saying why + `tun0` could not be created. On a guest whose `tun0` is already up it ends + the current stream and dials again. sandboxd sends it once a claim of a + tun-enabled pool is live, and again after every wake, unarchive and move, + so the guest drops its old stream at once whatever the VMM did with it (on + Cloud Hypervisor the guest also notices by itself within about 200 ms of a + restore). Warm VMs never get it, so goldens are unchanged. +- **`tun_down`** (control channel): stops the packet stream and removes `tun0` + with its routes before answering `done`, so clients fail at once with + "network unreachable". +- **Boot and reconnect**: silkd dials 2051 at boot and after every stream it + sees end, and lets the host hello decide. At boot a refused dial or a failed + hello means off. Once `tun0` is up, a refused dial or a failed hello is + retried with backoff (50 ms doubling to 1 s); an off answer removes `tun0`. + +The guest switch is advisory. The guest user can be root, create its own TUN +and dial 2051 directly, so the host enforces: it accepts a packet stream only +for a tun-enabled claim, rewrites every source address, and closes the stream +to revoke. + +## Wire protocol (version 1) + +All integers are little-endian. + +### Hello + +Each side sends one 12-byte hello first; the guest sends first, the host +answers. + +| Offset | Size | Field | Guest | Host | +| --- | --- | --- | --- | --- | +| 0 | 4 | magic | `SLKT` | `SLKT` | +| 4 | 2 | version | highest it speaks (1) | the version both use; ≤ the guest's | +| 6 | 2 | offloads | offloads its tun accepts on write | offloads its tun accepts on write | +| 8 | 2 | mtu | 0 | MTU for `tun0`; 0 keeps the current one | +| 10 | 2 | flags | 0 | bit 0 `ON`: packet mode is on for this box | + +Flag bits 1 to 15 are reserved and sent as 0; a peer ignores a bit it does +not know. A later revision uses one to open several streams per guest, one +per `tun0` queue. + +A host that answers without `ON` closes the stream after the hello; the guest +removes `tun0`. A host that refuses a guest answers without `ON` rather than +closing silently. A guest that does not speak the host's version closes the +stream. The guest gives up on a hello that does not arrive in 5 seconds. + +Offload bits are the kernel's `TUN_F_*` values: `0x01` CSUM, `0x02` TSO4, +`0x04` TSO6, `0x08` TSO_ECN, `0x20` USO4, `0x40` USO6. Each side probes its +kernel (USO needs Linux 6.2) and then enables, with `TUNSETOFFLOAD`, only the +offloads both hellos carry; segmentation offloads are dropped when CSUM is not +common. A side never builds a frame whose GSO type the peer did not offer; +packets its kernel queued before a reconnect keep their old type, and a +receiver may drop one it cannot take. + +### Frames + +After the hellos, both directions carry frames back to back: + +``` +u32 len | virtio_net_hdr (10 bytes) | IP packet +``` + +`len` counts the header and the packet: 10 to 65545 bytes. A length outside +that range is a protocol error and ends the stream. The header is the plain +10-byte `struct virtio_net_hdr` (`flags`, `gso_type`, `hdr_len`, `gso_size`, +`csum_start`, `csum_offset`) exactly as `IFF_VNET_HDR` reads and writes it, so +a GSO packet of up to 64 KiB crosses vsock as one frame and is segmented, if +at all, by the receiving kernel. A packet the receiving kernel refuses is +dropped like a packet lost on a link. + +Writers batch: every frame ready at the moment goes out in one stream write. +Readers parse every complete frame one stream read returns; the io_uring path +hands them to the TUN in batched submissions. + +### Reconnect + +There is no session state beyond the hello. After any stream loss the guest +dials again and both sides renegotiate offloads and MTU, so a move to a host +with a different kernel or MTU is safe. Packets in flight during the gap are +lost; TCP retransmits. + +## Host side + +The host side is sandboxd; its design is +[#471](https://github.com/cocoonstack/sandbox/issues/471). Whatever it does +with the packets, it owes the guest this contract: + +- **One stream per claim.** The packet stream arrives on the claim's own + vsock socket, so the stream, not any address in a packet, names the claim. + Every guest uses `100.64.0.2`; the host must not trust a source address the + guest wrote and must deliver only the claim's own return traffic on its + stream, with the destination set back to `100.64.0.2`. +- **Enforcement on the host.** The host answers `ON` only for a claim that is + allowed packet mode, and answers off, closes the stream or refuses the dial + otherwise; revocation closes the stream. Nothing depends on the guest + obeying `tun_down`. +- **Checksums after a rewrite.** A host that changes an address updates the + IPv4 header checksum and the L4 checksum incrementally. For a frame with + `VIRTIO_NET_HDR_F_NEEDS_CSUM` the L4 checksum sits at + `csum_start + csum_offset` and holds the uncomplemented pseudo-header sum; + a UDP checksum of 0 stays 0. With GSO this happens once per frame. +- **Offloads it offered.** The host sends only GSO types the guest's hello + offered and segments the rest itself. + +## Limits + +- IPv4 only. +- One stream per guest; a guest kernel without io_uring, or `SILKD_TUN_IO=epoll` + in silkd's environment, uses the epoll path. +- Windows guests run no silkd and get no advanced network mode. diff --git a/docs/index.md b/docs/index.md index 2a1c6861a..e106d134d 100644 --- a/docs/index.md +++ b/docs/index.md @@ -59,6 +59,8 @@ vsock-only I/O (hardened default); `net=egress` attaches a bridge/CNI NIC. - [Security model](security.md) — trust boundaries, what a compromised guest can reach, the deployment assumptions the guarantees rest on, and the honest limitations +- [Advanced network mode](advanced-network.md) — the L3 `tun0` path beside the + proxies, and its vsock packet protocol - [silkd](silkd.md) — the in-guest daemon: protocol, verb reference, lanes and limits - [Performance](performance.md) — measured latencies and the environments diff --git a/docs/silkd.md b/docs/silkd.md index 03ef6e647..4726b4c17 100644 --- a/docs/silkd.md +++ b/docs/silkd.md @@ -58,6 +58,7 @@ suites — a frame only one side can parse fails CI. | git | `git_clone {url, path, branch?, depth?, auth?}` / `git_status {path}` / `git_add {path, files}` / `git_commit {path, message, author}` / `git_push {path, auth?}` / `git_pull {path, auth?}` / `git_branch {path, action, name?}` | `git_status_result` (porcelain v2), `git_commit_result{hash}`, `git_branches{current, branches}` for `git_branch {action: "list"}`; every other verb terminates with `done`. A status past about 1 MiB of entries carries `truncated: true` with the head of the list, so one frame never exceeds the cap. `auth` is injected as an in-memory header, never written to guest disk | | port | `port_forward {port}` | relays guest TCP 127.0.0.1:port over this connection: `ready` once connected, then `data` both ways. Each direction ends on its own: `data_end` half-closes the guest socket, `done` reports the guest closing its write side, and the RPC ends once both have. A client that only drops the connection leaves the relay open until the guest side closes or writes. Works on both lanes — the no-network lane's only way in | | lsp | `lsp_start {language, root?}` / `lsp_request {server_id}` / `lsp_stop {server_id}` | a broker for the language server a flavor image ships: `lsp_start` spawns the argv named in `/etc/silkd/lsp.d/` (absent on the base image → `not_found` naming the flavor) and answers `lsp_started{server_id}`; `lsp_request` attaches the JSON-RPC byte stream (`ready`, then `data` both ways — silkd pipes bytes, it never parses LSP; `data_end` half-closes the server's stdin) and the stream ending reaps the server (v1 single-shot); `lsp_stop` kills it early, and a server nothing attaches within 5 minutes is reaped | +| tun | `tun_up` / `tun_down` | the [advanced network mode](advanced-network.md) switch: `tun_up` creates `tun0` and starts its packet stream on vsock 2051, `tun_down` removes `tun0` and its routes; both answer `done`, `tun_up` an `error` when `tun0` cannot be created. Advisory: the host enforces | | misc | `info` | `info{version, uptime_secs, procs, sessions}` — sandboxd uses it for readiness; there is no public SDK method. Distinct from the control plane's `GET /v1/info`, which reports node pools and claims | ## Error kinds @@ -94,6 +95,11 @@ forwards the image-baked proxy variables (`http_proxy` and friends) into every exec, so unconfigured clients use the relay without being told; a lane with its own routed network never gets them. +In the [advanced network mode](advanced-network.md) silkd also owns `tun0` +(`100.64.0.2/30`, default route) and carries its packets to the host over +vsock (`CID2:2051`); it asks the host at boot and after every lost stream, and +the proxy variables stay set beside it. + ## Claim env The host writes a claim's [`guest` env entries](sandboxd-api.md#getputpatch-v1sandboxesidenv) diff --git a/protocol/wire/fixtures/v1/req_tun_down.json b/protocol/wire/fixtures/v1/req_tun_down.json new file mode 100644 index 000000000..206836898 --- /dev/null +++ b/protocol/wire/fixtures/v1/req_tun_down.json @@ -0,0 +1 @@ +{"v":1,"op":"tun_down"} diff --git a/protocol/wire/fixtures/v1/req_tun_up.json b/protocol/wire/fixtures/v1/req_tun_up.json new file mode 100644 index 000000000..709327705 --- /dev/null +++ b/protocol/wire/fixtures/v1/req_tun_up.json @@ -0,0 +1 @@ +{"v":1,"op":"tun_up"} diff --git a/protocol/wire/frame.go b/protocol/wire/frame.go index 027103663..052b4744a 100644 --- a/protocol/wire/frame.go +++ b/protocol/wire/frame.go @@ -90,6 +90,8 @@ var ( "lsp_start": decodeReq[LspStart], "lsp_request": decodeReq[LspRequest], "lsp_stop": decodeReq[LspStop], + "tun_up": decodeReq[TunUp], + "tun_down": decodeReq[TunDown], "data": decodeReq[Data], //nolint:goconst // wire tag shared with the response type by design "data_end": decodeReq[DataEnd], } @@ -435,6 +437,16 @@ type LspStop struct { func (LspStop) Op() string { return "lsp_stop" } +// TunUp brings tun0 up and starts its packet stream on vsock port 2051; the host hello may still turn it off. +type TunUp struct{} + +func (TunUp) Op() string { return "tun_up" } + +// TunDown removes tun0 and its routes; the host, not the guest, enforces packet mode. +type TunDown struct{} + +func (TunDown) Op() string { return "tun_down" } + // Data carries one chunk of an upload stream (FsWrite/FsPush payloads). type Data struct { Data []byte `json:"data"` diff --git a/protocol/wire/frame_test.go b/protocol/wire/frame_test.go index 5fb5f2c66..dfa0c94ee 100644 --- a/protocol/wire/frame_test.go +++ b/protocol/wire/frame_test.go @@ -58,8 +58,8 @@ func TestFixtureCorpusRoundTrips(t *testing.T) { seen++ } - if seen != 61 { - t.Fatalf("fixture corpus: %d frames, want exactly 61", seen) + if seen != 63 { + t.Fatalf("fixture corpus: %d frames, want exactly 63", seen) } } diff --git a/silkd/Cargo.lock b/silkd/Cargo.lock index c71bbec74..0ca428311 100644 --- a/silkd/Cargo.lock +++ b/silkd/Cargo.lock @@ -182,6 +182,17 @@ dependencies = [ "libc", ] +[[package]] +name = "io-uring" +version = "0.7.15" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "ed3bd0ecfbb87805f538bb7b32e5239ca0763890c623e349860ecba69469f2bb" +dependencies = [ + "bitflags", + "cfg-if", + "libc", +] + [[package]] name = "itoa" version = "1.0.18" @@ -458,6 +469,7 @@ name = "silkd" version = "0.1.23" dependencies = [ "base64", + "io-uring", "libc", "memchr", "nix", @@ -468,6 +480,7 @@ dependencies = [ "tempfile", "tokio", "tokio-vsock", + "vsock", ] [[package]] diff --git a/silkd/Cargo.toml b/silkd/Cargo.toml index a53246d24..2ed666192 100644 --- a/silkd/Cargo.toml +++ b/silkd/Cargo.toml @@ -18,9 +18,12 @@ serde_json = "1" tokio = { version = "1", features = ["fs", "rt-multi-thread", "macros", "io-util", "net", "process", "sync", "time"] } [target.'cfg(target_os = "linux")'.dependencies] +io-uring = "0.7" tokio-vsock = "0.7" +vsock = "0.5" [dev-dependencies] +nix = { version = "0.31", features = ["sched"] } tempfile = "3" [lints.clippy] diff --git a/silkd/src/lib.rs b/silkd/src/lib.rs index ca16ef605..8c394f73c 100644 --- a/silkd/src/lib.rs +++ b/silkd/src/lib.rs @@ -16,5 +16,8 @@ pub mod server; pub mod session; pub mod sysutil; pub mod tree; +pub mod tun; +#[cfg(target_os = "linux")] +mod tun_uring; pub mod vsock; pub mod watch; diff --git a/silkd/src/main.rs b/silkd/src/main.rs index f2ce3abdb..9049a69ce 100644 --- a/silkd/src/main.rs +++ b/silkd/src/main.rs @@ -33,6 +33,8 @@ async fn main() { let dns = SocketAddrV4::new(net_egress::DNS_ADDR, net_egress::DNS_PORT); tokio::spawn(relay(dns, net_egress::DNS_HOST_VSOCK_PORT)); tokio::spawn(datagrams(dns, net_egress::DNS_DATAGRAM_HOST_VSOCK_PORT)); + #[cfg(target_os = "linux")] + silkd::tun::ask_at_boot(); tokio::spawn(async { if let Err(e) = imds::serve().await { eprintln!("silkd imds: {e}"); diff --git a/silkd/src/proto.rs b/silkd/src/proto.rs index 39c87b2e8..8dde830ac 100644 --- a/silkd/src/proto.rs +++ b/silkd/src/proto.rs @@ -169,6 +169,8 @@ pub enum Request { LspStop { server_id: String, }, + TunUp, + TunDown, Data { #[serde(with = "b64")] data: Vec, @@ -845,7 +847,7 @@ mod tests { seen += 1; } assert_eq!( - seen, 61, + seen, 63, "fixture corpus: adding a verb means adding its fixture" ); } diff --git a/silkd/src/server.rs b/silkd/src/server.rs index 594869507..85a66bfce 100644 --- a/silkd/src/server.rs +++ b/silkd/src/server.rs @@ -9,7 +9,7 @@ use tokio::task::JoinHandle; use crate::proc::{Chunk, Table, write_chunk}; use crate::proto::{self, ErrorKind, ExecReq, ProcInfo, Request, Response}; -use crate::{exec, find, forward, fs, git, lsp, pty, session, tree, watch}; +use crate::{exec, find, forward, fs, git, lsp, pty, session, tree, tun, watch}; /// Shared daemon state handed to every connection. pub struct State { @@ -153,6 +153,14 @@ impl State { return finish_feeder(feeder, res).await; } Request::LspStop { server_id } => self.lsp.stop(writer, &server_id).await, + Request::TunUp => tun_switch(writer, "tun up", tun::up).await, + Request::TunDown => { + tun_switch(writer, "tun down", || { + tun::down(); + Ok(()) + }) + .await + } Request::PortForward { port } => { let (rx, feeder) = spawn_feeder(reader); let res = forward::run(port, rx, writer).await; @@ -366,6 +374,21 @@ where (reader, None) } +/// Runs a blocking tun switch off the async workers and answers `done` or the error. +async fn tun_switch( + w: &mut W, + what: &str, + switch: fn() -> std::io::Result<()>, +) -> std::io::Result<()> { + match tokio::task::spawn_blocking(switch) + .await + .unwrap_or_else(|e| Err(std::io::Error::other(e))) + { + Ok(()) => proto::write_frame(w, &Response::Done).await, + Err(e) => proto::error_frame(w, ErrorKind::Internal, format!("{what}: {e}")).await, + } +} + fn now_secs() -> u64 { SystemTime::now() .duration_since(UNIX_EPOCH) diff --git a/silkd/src/sysutil.rs b/silkd/src/sysutil.rs index ac2516255..04d378696 100644 --- a/silkd/src/sysutil.rs +++ b/silkd/src/sysutil.rs @@ -2,7 +2,9 @@ use std::borrow::Cow; use std::io::Read; -use std::os::fd::{AsRawFd, OwnedFd, RawFd}; +#[cfg(target_os = "linux")] +use std::net::Ipv4Addr; +use std::os::fd::{AsFd, AsRawFd, BorrowedFd, OwnedFd, RawFd}; use std::os::unix::fs::MetadataExt; use std::os::unix::process::ExitStatusExt; use std::path::PathBuf; @@ -157,9 +159,8 @@ pub fn openpty(cols: u16, rows: u16) -> std::io::Result<(OwnedFd, OwnedFd)> { Ok((pty.master, pty.slave)) } -/// Registers an owned fd with the tokio reactor for readiness-driven I/O. -pub fn watch_fd(fd: OwnedFd) -> std::io::Result> { - // SAFETY: an OwnedFd stays open and yields the same raw fd until the AsyncFd that owns it drops. +pub fn watch_fd(fd: T) -> std::io::Result> { + // SAFETY: AsFd's I/O-safety contract keeps the raw fd open while `fd`, and so the AsyncFd, lives. unsafe { AsyncFd::register(fd) }.map_err(|e| e.into_parts().1) } @@ -206,6 +207,138 @@ pub async fn wait_code(child: &mut tokio::process::Child) -> i32 { child.wait().await.map_or(-1, exit_code) } +/// Opens tun device `name` with a virtio-net header leading every packet. +#[cfg(target_os = "linux")] +pub fn tun_open(name: &str) -> std::io::Result { + let file = std::fs::OpenOptions::new() + .read(true) + .write(true) + .open("/dev/net/tun")?; + let mut req = ifreq(name)?; + req.ifr_ifru.ifru_flags = + (libc::IFF_TUN | libc::IFF_NO_PI | libc::IFF_VNET_HDR) as libc::c_short; + // SAFETY: file is an open tun control fd; req is an initialized ifreq that outlives the call. + check(unsafe { libc::ioctl(file.as_raw_fd(), libc::TUNSETIFF, &mut req) })?; + Ok(file.into()) +} + +/// Sets the `TUN_F_*` offloads the tun reader accepts; the kernel refuses a bit it does not know. +#[cfg(target_os = "linux")] +pub fn tun_set_offload(fd: RawFd, flags: u32) -> std::io::Result<()> { + // SAFETY: TUNSETOFFLOAD takes its argument by value. + check(unsafe { libc::ioctl(fd, libc::TUNSETOFFLOAD, libc::c_ulong::from(flags)) }) +} + +/// Assigns `addr/prefix` to interface `name` and brings it up. +#[cfg(target_os = "linux")] +pub fn if_up(name: &str, addr: Ipv4Addr, prefix: u8) -> std::io::Result<()> { + let sock = ctl_socket()?; + let fd = sock.as_raw_fd(); + let mut req = ifreq(name)?; + req.ifr_ifru.ifru_addr = inet_sockaddr(addr); + if_ioctl(fd, libc::SIOCSIFADDR, &mut req)?; + req.ifr_ifru.ifru_netmask = inet_sockaddr(prefix_mask(prefix)); + if_ioctl(fd, libc::SIOCSIFNETMASK, &mut req)?; + if_ioctl(fd, libc::SIOCGIFFLAGS, &mut req)?; + // SAFETY: SIOCGIFFLAGS just filled ifru_flags. + let flags = unsafe { req.ifr_ifru.ifru_flags }; + req.ifr_ifru.ifru_flags = flags | libc::IFF_UP as libc::c_short; + if_ioctl(fd, libc::SIOCSIFFLAGS, &mut req) +} + +#[cfg(target_os = "linux")] +pub fn if_set_mtu(name: &str, mtu: u16) -> std::io::Result<()> { + let sock = ctl_socket()?; + let mut req = ifreq(name)?; + req.ifr_ifru.ifru_mtu = libc::c_int::from(mtu); + if_ioctl(sock.as_raw_fd(), libc::SIOCSIFMTU, &mut req) +} + +#[cfg(target_os = "linux")] +pub fn route_add( + dev: &str, + dst: Ipv4Addr, + prefix: u8, + gateway: Option, +) -> std::io::Result<()> { + let sock = ctl_socket()?; + let name = std::ffi::CString::new(dev)?; + // SAFETY: rtentry is plain data and all-zero is a valid value, a null rt_dev included. + let mut rt: libc::rtentry = unsafe { std::mem::zeroed() }; + rt.rt_dst = inet_sockaddr(dst); + rt.rt_genmask = inet_sockaddr(prefix_mask(prefix)); + rt.rt_flags = libc::RTF_UP; + if let Some(gw) = gateway { + rt.rt_gateway = inet_sockaddr(gw); + rt.rt_flags |= libc::RTF_GATEWAY; + } + rt.rt_dev = name.as_ptr().cast_mut(); + // SAFETY: rt is initialized and rt_dev points into `name`, which outlives the call. + check(unsafe { libc::ioctl(sock.as_raw_fd(), libc::SIOCADDRT as _, &mut rt) }) +} + +pub fn set_nonblocking(fd: BorrowedFd<'_>, on: bool) -> std::io::Result<()> { + let flags = OFlag::from_bits_retain(fcntl(fd, FcntlArg::F_GETFL)?); + let next = if on { + flags | OFlag::O_NONBLOCK + } else { + flags - OFlag::O_NONBLOCK + }; + if next != flags { + fcntl(fd, FcntlArg::F_SETFL(next))?; + } + Ok(()) +} + +pub fn write_all_fd(fd: BorrowedFd<'_>, mut buf: &[u8]) -> std::io::Result<()> { + while !buf.is_empty() { + match nix::unistd::write(fd, buf) { + Ok(n) => buf = &buf[n..], + Err(nix::errno::Errno::EINTR) => {} + Err(e) => return Err(e.into()), + } + } + Ok(()) +} + +/// Fills `buf` from blocking `fd`, failing with TimedOut once `timeout` passes. +pub fn read_exact_fd( + fd: BorrowedFd<'_>, + buf: &mut [u8], + timeout: std::time::Duration, +) -> std::io::Result<()> { + let deadline = std::time::Instant::now() + timeout; + let mut got = 0; + while got < buf.len() { + let left = deadline.saturating_duration_since(std::time::Instant::now()); + let mut pfd = libc::pollfd { + fd: fd.as_raw_fd(), + events: libc::POLLIN, + revents: 0, + }; + let ms = libc::c_int::try_from(left.as_millis()).unwrap_or(libc::c_int::MAX); + // SAFETY: pfd is one initialized pollfd that outlives the call. + match unsafe { libc::poll(&mut pfd, 1, ms) } { + 0 => return Err(std::io::ErrorKind::TimedOut.into()), + r if r < 0 => { + let e = std::io::Error::last_os_error(); + if e.kind() != std::io::ErrorKind::Interrupted { + return Err(e); + } + continue; + } + _ => {} + } + match nix::unistd::read(fd, &mut buf[got..]) { + Ok(0) => return Err(std::io::ErrorKind::UnexpectedEof.into()), + Ok(n) => got += n, + Err(nix::errno::Errno::EINTR) => {} + Err(e) => return Err(e.into()), + } + } + Ok(()) +} + fn make_controlling_tty() -> std::io::Result<()> { // SAFETY: setsid and ioctl take no pointers, so no memory precondition applies. unsafe { @@ -314,6 +447,58 @@ fn lookup_user(user: &str) -> Result<(u32, u32, PathBuf), String> { } } +#[cfg(target_os = "linux")] +fn ctl_socket() -> std::io::Result { + std::net::UdpSocket::bind((Ipv4Addr::UNSPECIFIED, 0)) +} + +#[cfg(target_os = "linux")] +fn ifreq(name: &str) -> std::io::Result { + if name.len() >= libc::IFNAMSIZ || name.as_bytes().contains(&0) { + return Err(std::io::Error::new( + std::io::ErrorKind::InvalidInput, + format!("invalid interface name {name:?}"), + )); + } + // SAFETY: ifreq is plain data and all-zero is a valid value. + let mut req: libc::ifreq = unsafe { std::mem::zeroed() }; + for (d, s) in req.ifr_name.iter_mut().zip(name.bytes()) { + *d = libc::c_char::from_ne_bytes([s]); + } + Ok(req) +} + +#[cfg(target_os = "linux")] +fn if_ioctl(fd: RawFd, request: libc::c_ulong, req: &mut libc::ifreq) -> std::io::Result<()> { + // SAFETY: fd is an open socket; req is an initialized ifreq that outlives the call. + check(unsafe { libc::ioctl(fd, request as _, req) }) +} + +#[cfg(target_os = "linux")] +fn inet_sockaddr(addr: Ipv4Addr) -> libc::sockaddr { + let mut sa_data = [0; 14]; + for (d, s) in sa_data[2..6].iter_mut().zip(addr.octets()) { + *d = libc::c_char::from_ne_bytes([s]); + } + libc::sockaddr { + sa_family: libc::AF_INET as libc::sa_family_t, + sa_data, + } +} + +#[cfg(target_os = "linux")] +fn prefix_mask(prefix: u8) -> Ipv4Addr { + Ipv4Addr::from(u32::MAX.checked_shl(32 - u32::from(prefix)).unwrap_or(0)) +} + +#[cfg(target_os = "linux")] +fn check(rc: libc::c_int) -> std::io::Result<()> { + if rc < 0 { + return Err(std::io::Error::last_os_error()); + } + Ok(()) +} + #[cfg(test)] mod tests { use super::*; diff --git a/silkd/src/tun.rs b/silkd/src/tun.rs new file mode 100644 index 000000000..32e90b4e0 --- /dev/null +++ b/silkd/src/tun.rs @@ -0,0 +1,591 @@ +//! Packet mode: tun0 carries the guest's non-proxy traffic as raw IP frames over one vsock stream to the host. + +use std::io; +use std::net::Ipv4Addr; + +pub const HOST_VSOCK_PORT: u32 = 2051; +/// The guest's address, the same in every box; the host tells boxes apart by their vsock stream. +pub const GUEST_ADDR: Ipv4Addr = Ipv4Addr::new(100, 64, 0, 2); +/// The guest's gateway; names resolve through the loopback DNS relay, never here. +pub const PEER_ADDR: Ipv4Addr = Ipv4Addr::new(100, 64, 0, 1); +pub const PREFIX: u8 = 30; +pub const VERSION: u16 = 1; +pub const HELLO_LEN: usize = 12; +pub const VNET_HDR_LEN: usize = 10; +/// The largest frame body: a virtio-net header and a 64 KiB GSO packet. +pub const MAX_FRAME: usize = VNET_HDR_LEN + 65535; +pub const FLAG_ON: u16 = 0x01; + +/// Offload bits in a hello, equal to the kernel's `TUN_F_*` values. +pub const OFFLOAD_CSUM: u16 = 0x01; +pub const OFFLOAD_TSO4: u16 = 0x02; +pub const OFFLOAD_TSO6: u16 = 0x04; +pub const OFFLOAD_TSO_ECN: u16 = 0x08; +pub const OFFLOAD_USO4: u16 = 0x20; +pub const OFFLOAD_USO6: u16 = 0x40; + +const MAGIC: [u8; 4] = *b"SLKT"; + +/// The 12 bytes each side sends first: magic, version, the offloads its tun accepts, the MTU and flags (host only). +#[derive(Clone, Copy, Debug, PartialEq, Eq)] +pub struct Hello { + pub version: u16, + pub offloads: u16, + pub mtu: u16, + pub flags: u16, +} + +impl Hello { + pub fn encode(&self) -> [u8; HELLO_LEN] { + let mut b = [0u8; HELLO_LEN]; + b[..4].copy_from_slice(&MAGIC); + b[4..6].copy_from_slice(&self.version.to_le_bytes()); + b[6..8].copy_from_slice(&self.offloads.to_le_bytes()); + b[8..10].copy_from_slice(&self.mtu.to_le_bytes()); + b[10..12].copy_from_slice(&self.flags.to_le_bytes()); + b + } + + pub fn decode(b: &[u8; HELLO_LEN]) -> io::Result { + if b[..4] != MAGIC { + return Err(io::Error::new( + io::ErrorKind::InvalidData, + "tun hello: bad magic", + )); + } + let field = |i: usize| u16::from_le_bytes([b[i], b[i + 1]]); + Ok(Hello { + version: field(4), + offloads: field(6), + mtu: field(8), + flags: field(10), + }) + } +} + +#[cfg(target_os = "linux")] +pub use linux::*; + +#[cfg(target_os = "linux")] +mod linux { + use std::os::fd::{AsFd, AsRawFd, BorrowedFd, OwnedFd}; + use std::sync::atomic::{AtomicBool, Ordering}; + use std::sync::{Arc, Mutex}; + use std::time::Duration; + + use tokio::io::unix::AsyncFd; + + use super::*; + use crate::{sysutil, tun_uring}; + + const IFNAME: &str = "tun0"; + const IPV6_OFF: &str = "/proc/sys/net/ipv6/conf/tun0/disable_ipv6"; + const DEFAULT_MTU: u16 = 1500; + /// Per-direction buffer; the sender batches every packet its tun has ready into one stream write. + const BATCH: usize = 256 << 10; + const HELLO_TIMEOUT: Duration = Duration::from_secs(5); + const REDIAL_MIN: Duration = Duration::from_millis(50); + const REDIAL_MAX: Duration = Duration::from_secs(1); + const FULL_OFFLOADS: u16 = + OFFLOAD_CSUM | OFFLOAD_TSO4 | OFFLOAD_TSO6 | OFFLOAD_TSO_ECN | OFFLOAD_USO4 | OFFLOAD_USO6; + /// The set a kernel older than 6.2 accepts, which predates UDP segmentation. + const BASE_OFFLOADS: u16 = OFFLOAD_CSUM | OFFLOAD_TSO4 | OFFLOAD_TSO6 | OFFLOAD_TSO_ECN; + + /// The link thread that owns tun0, if one runs. + static LINK: Mutex>> = Mutex::new(None); + /// Serializes tun_up, tun_down and the boot ask through teardown, so a verb answers only after the last one finished. + static SWITCH: Mutex<()> = Mutex::new(()); + + #[derive(Clone, Copy, Debug, PartialEq, Eq)] + pub enum Io { + Uring, + Epoll, + } + + impl Io { + /// io_uring unless `SILKD_TUN_IO=epoll`; a kernel that refuses io_uring falls back per session. + pub fn from_env() -> Io { + match std::env::var("SILKD_TUN_IO").as_deref() { + Ok("epoll") => Io::Epoll, + _ => Io::Uring, + } + } + } + + #[derive(Debug, PartialEq, Eq)] + pub enum Flow { + /// The host turned packet mode off for this box. + Off, + /// The stream closed; the link dials again. + Closed, + } + + /// A tun device with a virtio-net header on every packet, and the offloads its kernel accepts. + pub struct Tun { + fd: OwnedFd, + accepts: u16, + configured: Arc, + } + + impl Tun { + /// Opens tun `name` with every offload off until a peer states what it accepts. + pub fn open(name: &str) -> io::Result { + let fd = sysutil::tun_open(name)?; + let raw = fd.as_raw_fd(); + let accepts = [FULL_OFFLOADS, BASE_OFFLOADS] + .into_iter() + .find(|&m| sysutil::tun_set_offload(raw, m.into()).is_ok()) + .unwrap_or(0); + sysutil::tun_set_offload(raw, 0)?; + Ok(Tun { + fd, + accepts, + configured: Arc::new(AtomicBool::new(false)), + }) + } + + pub fn accepts(&self) -> u16 { + self.accepts + } + + /// Turns on the offloads both kernels accept. + pub fn agree(&self, peer: u16) -> io::Result<()> { + sysutil::tun_set_offload(self.fd.as_raw_fd(), settable(self.accepts & peer).into()) + } + + /// Gives tun0 the fixed guest address and the default route, and no IPv6; idempotent. + pub fn configure_guest(&self) -> io::Result<()> { + if self.configured() { + return Ok(()); + } + let _ = std::fs::write(IPV6_OFF, "1"); + sysutil::if_set_mtu(IFNAME, DEFAULT_MTU)?; + sysutil::if_up(IFNAME, GUEST_ADDR, PREFIX)?; + sysutil::route_add(IFNAME, Ipv4Addr::UNSPECIFIED, 0, Some(PEER_ADDR))?; + self.configured.store(true, Ordering::Relaxed); + Ok(()) + } + + fn configured(&self) -> bool { + self.configured.load(Ordering::Relaxed) + } + } + + struct Link { + stopping: AtomicBool, + configured: Arc, + stream: Mutex>>, + thread: Mutex>>, + } + + impl Link { + /// Ends the current stream and wakes a waiting redial. + fn redial(&self) { + if let Some(stream) = sysutil::lock(&self.stream).as_ref() { + let _ = stream.shutdown(std::net::Shutdown::Both); + } + if let Some(thread) = sysutil::lock(&self.thread).as_ref() { + thread.thread().unpark(); + } + } + + /// Stops the link and waits until its thread has dropped tun0. + fn stop(&self) { + self.stopping.store(true, Ordering::Relaxed); + self.redial(); + if let Some(thread) = sysutil::lock(&self.thread).take() { + let _ = thread.join(); + } + } + } + + /// Handles `tun_up`: brings tun0 up with its route and starts its link, or makes a running link dial again, since a restored VMM never resets the old stream. + pub fn up() -> io::Result<()> { + let _switch = sysutil::lock(&SWITCH); + loop { + let mut slot = sysutil::lock(&LINK); + let Some(link) = slot.as_ref() else { + let tun = Tun::open(IFNAME)?; + tun.configure_guest()?; + *slot = Some(spawn_link(tun)?); + return Ok(()); + }; + if link.configured.load(Ordering::Relaxed) { + link.redial(); + return Ok(()); + } + let boot = slot.take(); + drop(slot); + if let Some(boot) = boot { + boot.stop(); + } + } + } + + /// Handles `tun_down`: stops the link and removes tun0 with its routes before returning; enforcement stays on the host. + pub fn down() { + let _switch = sysutil::lock(&SWITCH); + let link = sysutil::lock(&LINK).take(); + if let Some(link) = link { + link.stop(); + } + } + + /// Asks the host at boot; a refused dial or an off answer leaves no tun0. + pub fn ask_at_boot() { + let _switch = sysutil::lock(&SWITCH); + let mut slot = sysutil::lock(&LINK); + if slot.is_some() { + return; + } + match Tun::open(IFNAME).and_then(spawn_link) { + Ok(link) => *slot = Some(link), + Err(e) => eprintln!("silkd tun: {e}"), + } + } + + /// Runs one guest connection: hello both ways, then pumps until the stream ends; configures tun0 on the first on. + pub fn session(tun: &Tun, stream: &S, backend: Io) -> io::Result { + let fd = stream.as_fd(); + sysutil::set_nonblocking(fd, false)?; + let hello = Hello { + version: VERSION, + offloads: tun.accepts(), + mtu: 0, + flags: 0, + }; + sysutil::write_all_fd(fd, &hello.encode())?; + let mut b = [0u8; HELLO_LEN]; + sysutil::read_exact_fd(fd, &mut b, HELLO_TIMEOUT)?; + let host = Hello::decode(&b)?; + if host.version != VERSION { + return Err(io::Error::new( + io::ErrorKind::Unsupported, + format!("tun hello: host speaks version {}", host.version), + )); + } + if host.flags & FLAG_ON == 0 { + return Ok(Flow::Off); + } + tun.configure_guest()?; + tun.agree(host.offloads)?; + if host.mtu != 0 { + sysutil::if_set_mtu(IFNAME, host.mtu)?; + } + pump(tun, fd, backend)?; + Ok(Flow::Closed) + } + + /// Moves frames between `tun` and `stream` until either fails or the stream ends; blocks the calling thread. + pub fn pump(tun: &Tun, stream: BorrowedFd<'_>, backend: Io) -> io::Result<()> { + let tun_fd = tun.fd.as_fd(); + if backend == Io::Uring { + match tun_uring::Ring::new() { + Ok(ring) => { + sysutil::set_nonblocking(tun_fd, false)?; + sysutil::set_nonblocking(stream, false)?; + return ended(ring.pump(tun_fd.as_raw_fd(), stream.as_raw_fd())); + } + Err(e) => eprintln!("silkd tun: io_uring unavailable, using epoll: {e}"), + } + } + sysutil::set_nonblocking(tun_fd, true)?; + sysutil::set_nonblocking(stream, true)?; + ended( + tokio::runtime::Builder::new_current_thread() + .enable_io() + .build()? + .block_on(epoll_pump(tun_fd, stream)), + ) + } + + /// A peer that went away while a write was in flight ends the stream the same way an EOF does. + fn ended(res: io::Result<()>) -> io::Result<()> { + match res { + Err(e) + if matches!( + e.kind(), + io::ErrorKind::BrokenPipe + | io::ErrorKind::ConnectionReset + | io::ErrorKind::NotConnected + ) => + { + Ok(()) + } + res => res, + } + } + + fn spawn_link(tun: Tun) -> io::Result> { + let link = Arc::new(Link { + stopping: AtomicBool::new(false), + configured: Arc::clone(&tun.configured), + stream: Mutex::new(None), + thread: Mutex::new(None), + }); + let me = Arc::clone(&link); + let thread = std::thread::Builder::new() + .name("silkd-tun".into()) + .spawn(move || run_link(&me, tun))?; + *sysutil::lock(&link.thread) = Some(thread); + Ok(link) + } + + fn run_link(me: &Arc, tun: Tun) { + let backend = Io::from_env(); + let mut wait = REDIAL_MIN; + while !me.stopping.load(Ordering::Relaxed) { + match vsock::VsockStream::connect_with_cid_port(libc::VMADDR_CID_HOST, HOST_VSOCK_PORT) + { + Ok(stream) => { + let stream = Arc::new(stream); + { + let mut slot = sysutil::lock(&me.stream); + if me.stopping.load(Ordering::Relaxed) { + break; + } + *slot = Some(Arc::clone(&stream)); + } + let flow = session(&tun, &*stream, backend); + *sysutil::lock(&me.stream) = None; + match flow { + Ok(Flow::Off) => break, + Ok(Flow::Closed) => wait = REDIAL_MIN, + Err(_) if !tun.configured() => break, + Err(e) if !me.stopping.load(Ordering::Relaxed) => { + eprintln!("silkd tun: {e}") + } + Err(_) => {} + } + } + Err(_) if !tun.configured() => break, + Err(_) => {} + } + std::thread::park_timeout(wait); + wait = (wait * 2).min(REDIAL_MAX); + } + let mut slot = sysutil::lock(&LINK); + // tun0 closes under LINK, so a racing up() never finds the name still taken. + drop(tun); + if slot.as_ref().is_some_and(|l| Arc::ptr_eq(l, me)) { + *slot = None; + } + } + + async fn epoll_pump(tun: BorrowedFd<'_>, stream: BorrowedFd<'_>) -> io::Result<()> { + let tun = sysutil::watch_fd(tun)?; + let stream = sysutil::watch_fd(stream)?; + tokio::select! { + res = tun_to_stream(&tun, &stream) => res, + res = stream_to_tun(&tun, &stream) => res, + } + } + + async fn tun_to_stream( + tun: &AsyncFd>, + stream: &AsyncFd>, + ) -> io::Result<()> { + let mut buf = vec![0u8; BATCH]; + loop { + let mut guard = tun.readable().await?; + let mut len = 0; + while len + 4 + MAX_FRAME <= BATCH { + let body = &mut buf[len + 4..len + 4 + MAX_FRAME]; + match guard.try_io(|fd| nix::unistd::read(fd.get_ref(), body).map_err(Into::into)) { + Ok(Ok(n)) if n >= VNET_HDR_LEN => { + buf[len..len + 4].copy_from_slice(&(n as u32).to_le_bytes()); + len += 4 + n; + } + Ok(Ok(_)) => {} + Ok(Err(e)) => return Err(e), + Err(_) => break, + } + } + let mut sent = 0; + while sent < len { + let mut guard = stream.writable().await?; + match guard.try_io(|fd| { + nix::unistd::write(fd.get_ref(), &buf[sent..len]).map_err(Into::into) + }) { + Ok(Ok(n)) => sent += n, + Ok(Err(e)) => return Err(e), + Err(_) => {} + } + } + } + } + + async fn stream_to_tun( + tun: &AsyncFd>, + stream: &AsyncFd>, + ) -> io::Result<()> { + let mut buf = vec![0u8; BATCH]; + let (mut start, mut end) = (0, 0); + loop { + while let Some(len) = next_frame(&buf[start..end])? { + send_packet(tun, &buf[start + 4..start + 4 + len]).await; + start += 4 + len; + } + (start, end) = compact(&mut buf, start, end); + let mut guard = stream.readable().await?; + match guard + .try_io(|fd| nix::unistd::read(fd.get_ref(), &mut buf[end..]).map_err(Into::into)) + { + Ok(Ok(0)) => return Ok(()), + Ok(Ok(n)) => end += n, + Ok(Err(e)) => return Err(e), + Err(_) => {} + } + } + } + + async fn send_packet(tun: &AsyncFd>, pkt: &[u8]) { + loop { + let Ok(mut guard) = tun.writable().await else { + return; + }; + // a packet the kernel refuses is dropped, as a lossy link drops it. + if guard + .try_io(|fd| nix::unistd::write(fd.get_ref(), pkt).map_err(Into::into)) + .is_ok() + { + return; + } + } + } +} + +/// Keeps a partial frame in `buf[start..end]` readable: rewinds an empty buffer, and moves the tail to the front only when a whole frame no longer fits behind it. +pub fn compact(buf: &mut [u8], start: usize, end: usize) -> (usize, usize) { + if start == end { + return (0, 0); + } + if buf.len() - end >= 4 + MAX_FRAME { + return (start, end); + } + buf.copy_within(start..end, 0); + (0, end - start) +} + +/// The body length of the complete frame at the head of `buf`, or None while it is still partial. +pub fn next_frame(buf: &[u8]) -> io::Result> { + let Some(head) = buf.first_chunk::<4>() else { + return Ok(None); + }; + let len = u32::from_le_bytes(*head) as usize; + if !(VNET_HDR_LEN..=MAX_FRAME).contains(&len) { + return Err(io::Error::new( + io::ErrorKind::InvalidData, + format!("tun frame length {len}"), + )); + } + Ok((buf.len() >= 4 + len).then_some(len)) +} + +#[cfg(not(target_os = "linux"))] +pub fn up() -> io::Result<()> { + Err(io::Error::new( + io::ErrorKind::Unsupported, + "silkd tun requires linux", + )) +} + +#[cfg(not(target_os = "linux"))] +pub fn down() {} + +/// Trims `bits` to a set `TUNSETOFFLOAD` takes: segmentation needs CSUM, ECN needs TSO, and USO comes in pairs. +#[cfg(any(target_os = "linux", test))] +fn settable(bits: u16) -> u16 { + if bits & OFFLOAD_CSUM == 0 { + return 0; + } + let mut on = bits + & (OFFLOAD_CSUM + | OFFLOAD_TSO4 + | OFFLOAD_TSO6 + | OFFLOAD_TSO_ECN + | OFFLOAD_USO4 + | OFFLOAD_USO6); + if on & (OFFLOAD_TSO4 | OFFLOAD_TSO6) == 0 { + on &= !OFFLOAD_TSO_ECN; + } + if on & (OFFLOAD_USO4 | OFFLOAD_USO6) != OFFLOAD_USO4 | OFFLOAD_USO6 { + on &= !(OFFLOAD_USO4 | OFFLOAD_USO6); + } + on +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn hello_round_trips() { + let h = Hello { + version: VERSION, + offloads: OFFLOAD_CSUM | OFFLOAD_TSO4, + mtu: 1460, + flags: FLAG_ON, + }; + let b = h.encode(); + assert_eq!(&b[..4], b"SLKT"); + assert_eq!(Hello::decode(&b).unwrap(), h); + } + + #[test] + fn hello_rejects_bad_magic() { + let mut b = Hello { + version: VERSION, + offloads: 0, + mtu: 0, + flags: 0, + } + .encode(); + b[0] = b'X'; + assert_eq!( + Hello::decode(&b).unwrap_err().kind(), + io::ErrorKind::InvalidData + ); + } + + #[test] + fn settable_drops_sets_the_kernel_refuses() { + let tso = OFFLOAD_CSUM | OFFLOAD_TSO4 | OFFLOAD_TSO6 | OFFLOAD_TSO_ECN; + assert_eq!(settable(OFFLOAD_TSO4 | OFFLOAD_TSO6), 0); + assert_eq!(settable(OFFLOAD_CSUM | OFFLOAD_TSO_ECN), OFFLOAD_CSUM); + assert_eq!(settable(tso | OFFLOAD_USO4), tso); + assert_eq!( + settable(tso | OFFLOAD_USO4 | OFFLOAD_USO6 | 0x10), + tso | OFFLOAD_USO4 | OFFLOAD_USO6 + ); + } + + #[test] + fn compact_moves_a_partial_frame_only_when_a_frame_no_longer_fits() { + let mut buf = vec![0u8; 2 * (4 + MAX_FRAME)]; + assert_eq!(compact(&mut buf, 7, 7), (0, 0)); + assert_eq!(compact(&mut buf, 10, 20), (10, 20)); + let end = buf.len() - 4; + buf[end - 2] = 0xab; + assert_eq!(compact(&mut buf, end - 3, end), (0, 3)); + assert_eq!(buf[1], 0xab); + } + + #[test] + fn next_frame_waits_for_the_whole_body() { + let mut buf = (12u32).to_le_bytes().to_vec(); + buf.extend_from_slice(&[0; 11]); + assert_eq!(next_frame(&buf).unwrap(), None); + buf.push(0); + assert_eq!(next_frame(&buf).unwrap(), Some(12)); + assert_eq!(next_frame(&buf[..3]).unwrap(), None); + } + + #[test] + fn next_frame_rejects_lengths_out_of_range() { + for len in [0u32, 9, (MAX_FRAME + 1) as u32] { + let err = next_frame(&len.to_le_bytes()).unwrap_err(); + assert_eq!(err.kind(), io::ErrorKind::InvalidData, "{len}"); + } + } +} diff --git a/silkd/src/tun_uring.rs b/silkd/src/tun_uring.rs new file mode 100644 index 000000000..a27b37749 --- /dev/null +++ b/silkd/src/tun_uring.rs @@ -0,0 +1,285 @@ +//! io_uring packet path: tun reads stay queued in the kernel, and one submission carries many tun writes or one batched stream write. + +use std::collections::VecDeque; +use std::io; +use std::os::fd::RawFd; + +use io_uring::{IoUring, opcode, squeue, types}; + +use crate::tun::{MAX_FRAME, VNET_HDR_LEN, compact, next_frame}; + +const ENTRIES: u32 = 256; +/// One stream read of minimal frames queues about 7,700 tun writes; the completion queue holds them all. +const CQ_ENTRIES: u32 = 8192; +/// Tun reads kept queued, one frame per slot. +const SLOTS: usize = 32; +const DOWN_BUF: usize = 256 << 10; + +const TUN_READ: u64 = 1 << 32; +const TUN_WRITE: u64 = 2 << 32; +const STREAM_READ: u64 = 3 << 32; +const STREAM_WRITE: u64 = 4 << 32; +const CANCEL: u64 = 5 << 32; +const KIND_MASK: u64 = !0xffff_ffff; + +/// A ring set up before the session commits to io_uring, so a refusing kernel falls back to epoll. +pub struct Ring(IoUring); + +impl Ring { + pub fn new() -> io::Result { + Ok(Ring( + IoUring::builder().setup_cqsize(CQ_ENTRIES).build(ENTRIES)?, + )) + } + + /// Moves frames between blocking `tun` and `stream` fds until either fails or the stream ends. + pub fn pump(self, tun: RawFd, stream: RawFd) -> io::Result<()> { + let mut p = Pump { + ring: self.0, + inflight: 0, + tun, + stream, + slots: (0..SLOTS) + .map(|_| vec![0u8; 4 + MAX_FRAME].into_boxed_slice()) + .collect(), + ready: VecDeque::with_capacity(SLOTS), + writing: Vec::with_capacity(SLOTS), + iov: Vec::with_capacity(SLOTS), + iov_at: 0, + down: vec![0u8; DOWN_BUF].into_boxed_slice(), + start: 0, + end: 0, + tun_writes: 0, + done: Vec::with_capacity(ENTRIES as usize), + }; + p.run() + } +} + +/// One session's ring and every buffer its queued operations name. +struct Pump { + ring: IoUring, + inflight: usize, + tun: RawFd, + stream: RawFd, + + slots: Vec>, + ready: VecDeque<(usize, usize)>, + writing: Vec, + iov: Vec, + iov_at: usize, + + down: Box<[u8]>, + start: usize, + end: usize, + tun_writes: usize, + + done: Vec<(u64, i32)>, +} + +impl Pump { + fn run(&mut self) -> io::Result<()> { + for i in 0..SLOTS { + self.arm_tun_read(i)?; + } + self.arm_stream_read()?; + loop { + match self.ring.submit_and_wait(1) { + Err(e) if e.kind() == io::ErrorKind::Interrupted => continue, + res => res?, + }; + self.done.clear(); + self.done + .extend(self.ring.completion().map(|c| (c.user_data(), c.result()))); + self.inflight -= self.done.len(); + for i in 0..self.done.len() { + let (ud, res) = self.done[i]; + let id = (ud & !KIND_MASK) as usize; + match ud & KIND_MASK { + TUN_READ => self.tun_read_done(id, res)?, + TUN_WRITE => self.tun_write_done()?, + STREAM_READ => { + if !self.stream_read_done(res)? { + return Ok(()); + } + } + STREAM_WRITE => self.stream_write_done(res)?, + _ => {} + } + } + self.flush_ready()?; + } + } + + fn tun_read_done(&mut self, slot: usize, res: i32) -> io::Result<()> { + if res >= VNET_HDR_LEN as i32 { + let n = res as usize; + self.slots[slot][..4].copy_from_slice(&(n as u32).to_le_bytes()); + self.ready.push_back((slot, 4 + n)); + return Ok(()); + } + if res < 0 && !retryable(res) { + return Err(io::Error::from_raw_os_error(-res)); + } + self.arm_tun_read(slot) + } + + fn tun_write_done(&mut self) -> io::Result<()> { + self.tun_writes -= 1; + if self.tun_writes == 0 { + self.compact_and_read()?; + } + Ok(()) + } + + /// Returns false once the host closed the stream. + fn stream_read_done(&mut self, res: i32) -> io::Result { + match res { + 0 => return Ok(false), + r if r < 0 && retryable(r) => { + self.arm_stream_read()?; + return Ok(true); + } + r if r < 0 => return Err(io::Error::from_raw_os_error(-r)), + r => self.end += r as usize, + } + while let Some(len) = next_frame(&self.down[self.start..self.end])? { + let pkt = self.down[self.start + 4..].as_ptr(); + let write = opcode::Write::new(types::Fd(self.tun), pkt, len as u32) + .build() + .user_data(TUN_WRITE); + self.push(write)?; + self.tun_writes += 1; + self.start += 4 + len; + } + if self.tun_writes == 0 { + self.compact_and_read()?; + } + Ok(true) + } + + fn stream_write_done(&mut self, res: i32) -> io::Result<()> { + if res < 0 { + if retryable(res) { + return self.submit_writev(); + } + return Err(io::Error::from_raw_os_error(-res)); + } + let mut n = res as usize; + while n > 0 && self.iov_at < self.iov.len() { + let v = &mut self.iov[self.iov_at]; + if n >= v.iov_len { + n -= v.iov_len; + self.iov_at += 1; + } else { + v.iov_base = v.iov_base.wrapping_byte_add(n); + v.iov_len -= n; + n = 0; + } + } + if self.iov_at < self.iov.len() { + return self.submit_writev(); + } + while let Some(slot) = self.writing.pop() { + self.arm_tun_read(slot)?; + } + Ok(()) + } + + fn flush_ready(&mut self) -> io::Result<()> { + if !self.writing.is_empty() || self.ready.is_empty() { + return Ok(()); + } + self.iov.clear(); + self.iov_at = 0; + while let Some((slot, len)) = self.ready.pop_front() { + self.writing.push(slot); + self.iov.push(libc::iovec { + iov_base: self.slots[slot].as_mut_ptr().cast(), + iov_len: len, + }); + } + self.submit_writev() + } + + fn submit_writev(&mut self) -> io::Result<()> { + let rest = &self.iov[self.iov_at..]; + let writev = opcode::Writev::new(types::Fd(self.stream), rest.as_ptr(), rest.len() as u32) + .build() + .user_data(STREAM_WRITE); + self.push(writev) + } + + fn arm_tun_read(&mut self, slot: usize) -> io::Result<()> { + let buf = self.slots[slot][4..].as_mut_ptr(); + let read = opcode::Read::new(types::Fd(self.tun), buf, MAX_FRAME as u32) + .build() + .user_data(TUN_READ | slot as u64); + self.push(read) + } + + fn compact_and_read(&mut self) -> io::Result<()> { + (self.start, self.end) = compact(&mut self.down, self.start, self.end); + self.arm_stream_read() + } + + fn arm_stream_read(&mut self) -> io::Result<()> { + let rest = &mut self.down[self.end..]; + let read = opcode::Read::new(types::Fd(self.stream), rest.as_mut_ptr(), rest.len() as u32) + .build() + .user_data(STREAM_READ); + self.push(read) + } + + fn push(&mut self, entry: squeue::Entry) -> io::Result<()> { + queue(&mut self.ring, &entry)?; + self.inflight += 1; + Ok(()) + } +} + +impl Drop for Pump { + fn drop(&mut self) { + let targets = (0..SLOTS as u64) + .map(|i| TUN_READ | i) + .chain([STREAM_READ, STREAM_WRITE]); + for ud in targets { + let cancel = opcode::AsyncCancel::new(ud).build().user_data(CANCEL); + if queue(&mut self.ring, &cancel).is_err() { + break; + } + } + while self.inflight > 0 { + match self.ring.submit_and_wait(1) { + Ok(_) => {} + Err(e) if e.kind() == io::ErrorKind::Interrupted => {} + // the ring cannot report completions; leaking the buffers beats freeing memory the kernel may write. + Err(_) => { + std::mem::forget(std::mem::take(&mut self.slots)); + std::mem::forget(std::mem::take(&mut self.iov)); + std::mem::forget(std::mem::take(&mut self.down)); + return; + } + } + let finished = self + .ring + .completion() + .filter(|c| c.user_data() != CANCEL) + .count(); + self.inflight -= finished; + } + } +} + +/// Pushes entry, submitting to make room while the queue is full. +fn queue(ring: &mut IoUring, entry: &squeue::Entry) -> io::Result<()> { + // SAFETY: every buffer an entry names lives in its Pump, and Pump's Drop waits out each queued entry before freeing them. + while unsafe { ring.submission().push(entry) }.is_err() { + ring.submit()?; + } + Ok(()) +} + +fn retryable(res: i32) -> bool { + res == -libc::EAGAIN || res == -libc::EINTR +} diff --git a/silkd/tests/tun_e2e.rs b/silkd/tests/tun_e2e.rs new file mode 100644 index 000000000..0502e6a8b --- /dev/null +++ b/silkd/tests/tun_e2e.rs @@ -0,0 +1,657 @@ +//! Packet mode end to end: guest tun0 and a host tun in two network namespaces, joined over a stream pair; root only. +#![cfg(target_os = "linux")] +#![allow(clippy::unwrap_used, clippy::expect_used)] + +use std::net::{Ipv4Addr, SocketAddr}; +use std::os::fd::{AsFd, AsRawFd}; +use std::os::unix::net::UnixStream; +use std::sync::Arc; +use std::sync::atomic::{AtomicBool, AtomicU64, Ordering}; +use std::thread::JoinHandle; +use std::time::{Duration, Instant}; + +use nix::sched::{CloneFlags, unshare}; +use tokio::io::{AsyncBufReadExt, AsyncReadExt, AsyncWriteExt, BufReader}; +use tokio::net::{TcpListener, TcpStream, UdpSocket}; + +use silkd::sysutil; +use silkd::tun::{ + self, FLAG_ON, Flow, HELLO_LEN, Hello, Io, PEER_ADDR, Tun, VERSION, VNET_HDR_LEN, +}; + +const CLAIM_ADDR: Ipv4Addr = Ipv4Addr::new(100, 65, 0, 7); +const UPLOAD: u16 = 5001; +const ECHO: u16 = 5002; +const DOWNLOAD: u16 = 5003; +const SINK: u16 = 5004; +const PPS_SINK: u16 = 5005; +const PPS_SOURCE: u16 = 5006; +const GUEST_PPS_PORT: u16 = 6000; +const BULK: u64 = 256 << 20; +const GSO_NONE: u8 = 0; +const GSO_UDP_L4: u8 = 5; +const NEEDS_CSUM: u8 = 1; +const IP_SRC: usize = 12; +const IP_DST: usize = 16; +const PROTO_TCP: u8 = 6; +const PROTO_UDP: u8 = 17; +const UDP_SEGMENT: libc::c_int = 103; +const UDP_DATAGRAMS: usize = 32; +const UDP_SEG: usize = 1000; +const SMALL: usize = 64; +const FLOOD: Duration = Duration::from_secs(2); + +#[derive(Default)] +struct Stats { + frames: AtomicU64, + gso: AtomicU64, + uso: AtomicU64, + largest: AtomicU64, +} + +struct Seen { + upload_peer: String, + echo_peers: Vec, + downloaded: u64, + sink_datagrams: u64, +} + +#[test] +#[ignore = "needs root for network namespaces and /dev/net/tun"] +fn tcp_udp_and_gso_cross_a_rewriting_host() { + for io in [Io::Uring, Io::Epoll] { + let up = Arc::new(Stats::default()); + let down = Arc::new(Stats::default()); + let (seen, accepts) = with_link(io, true, Some((up.clone(), down.clone())), |_| { + guest_workload() + }); + assert_eq!(seen.upload_peer, CLAIM_ADDR.to_string(), "{io:?}"); + assert_eq!(seen.echo_peers, vec![CLAIM_ADDR.to_string(); 16], "{io:?}"); + assert_eq!(seen.downloaded, BULK, "{io:?}"); + assert!( + up.gso.load(Ordering::Relaxed) > 0, + "{io:?}: no guest GSO frame" + ); + assert!( + down.gso.load(Ordering::Relaxed) > 0, + "{io:?}: no host GSO frame" + ); + assert!( + up.largest.load(Ordering::Relaxed) > 16_000, + "{io:?}: MTU-sized frames" + ); + if accepts & tun::OFFLOAD_USO4 != 0 { + assert_eq!(seen.sink_datagrams, UDP_DATAGRAMS as u64, "{io:?}"); + assert!( + up.uso.load(Ordering::Relaxed) > 0, + "{io:?}: no guest USO frame" + ); + } + eprintln!( + "{io:?}: accepts {accepts:#x}; up frames {} gso {} uso {} largest {}; down frames {} gso {} largest {}", + up.frames.load(Ordering::Relaxed), + up.gso.load(Ordering::Relaxed), + up.uso.load(Ordering::Relaxed), + up.largest.load(Ordering::Relaxed), + down.frames.load(Ordering::Relaxed), + down.gso.load(Ordering::Relaxed), + down.largest.load(Ordering::Relaxed), + ); + } +} + +#[test] +#[ignore = "needs root for network namespaces and /dev/net/tun"] +fn full_checksums_cross_when_offloads_are_off() { + for io in [Io::Uring, Io::Epoll] { + let up = Arc::new(Stats::default()); + let down = Arc::new(Stats::default()); + let (seen, _) = with_link(io, false, Some((up.clone(), down.clone())), |_| { + guest_workload() + }); + assert_eq!(seen.upload_peer, CLAIM_ADDR.to_string(), "{io:?}"); + assert_eq!(seen.echo_peers, vec![CLAIM_ADDR.to_string(); 16], "{io:?}"); + assert_eq!(seen.downloaded, BULK, "{io:?}"); + assert_eq!(up.gso.load(Ordering::Relaxed), 0, "{io:?}"); + assert_eq!(down.gso.load(Ordering::Relaxed), 0, "{io:?}"); + } +} + +#[test] +#[ignore = "needs root for network namespaces and /dev/net/tun"] +fn host_off_answer_leaves_no_tun() { + in_netns(|| { + let tun = Tun::open("tun0").unwrap(); + let (guest, host) = UnixStream::pair().unwrap(); + let answer = std::thread::spawn(move || { + let mut b = [0u8; HELLO_LEN]; + sysutil::read_exact_fd(host.as_fd(), &mut b, Duration::from_secs(5)).unwrap(); + assert_eq!(Hello::decode(&b).unwrap().version, VERSION); + let off = Hello { + version: VERSION, + offloads: 0, + mtu: 0, + flags: 0, + }; + sysutil::write_all_fd(host.as_fd(), &off.encode()).unwrap(); + }); + assert_eq!(tun::session(&tun, &guest, Io::Uring).unwrap(), Flow::Off); + answer.join().unwrap(); + drop(tun); + assert!(Tun::open("tun0").is_ok(), "tun0 still held after off"); + }) + .join() + .unwrap(); +} + +#[test] +#[ignore = "needs root; prints throughput, asserts nothing about speed"] +fn throughput_and_per_packet_cost() { + let order = [ + (Io::Uring, true), + (Io::Epoll, true), + (Io::Uring, false), + (Io::Epoll, false), + (Io::Epoll, false), + (Io::Uring, false), + (Io::Epoll, true), + (Io::Uring, true), + ]; + for (io, offloads) in order { + let (r, _) = with_link(io, offloads, None, |pump_tid| { + runtime().block_on(async move { + let t = Instant::now(); + upload(SocketAddr::from((PEER_ADDR, UPLOAD)), 4 * BULK).await; + let up = gbits(4 * BULK, t.elapsed()); + let t = Instant::now(); + let got = download(SocketAddr::from((PEER_ADDR, DOWNLOAD))).await; + let down = gbits(got, t.elapsed()); + let cpu = pump_cpu_ns(pump_tid); + let sent = flood_up().await; + let up_pkts = query(PPS_SINK).await; + let up_cpu = pump_cpu_ns(pump_tid) - cpu; + let cpu = pump_cpu_ns(pump_tid); + let down_pkts = flood_down().await; + let down_cpu = pump_cpu_ns(pump_tid) - cpu; + (up, down, sent, up_pkts, up_cpu, down_pkts, down_cpu) + }) + }); + let (up, down, sent, up_pkts, up_cpu, down_pkts, down_cpu) = r; + let secs = FLOOD.as_secs_f64(); + eprintln!( + "{io:?} offloads={offloads}: tcp up {up:.2} Gbit/s, down {down:.2} Gbit/s; \ + udp{SMALL} up {:.0} kpps ({:.0} ns/pkt, {sent} sent), down {:.0} kpps ({:.0} ns/pkt)", + up_pkts as f64 / secs / 1e3, + up_cpu as f64 / up_pkts.max(1) as f64, + down_pkts as f64 / secs / 1e3, + down_cpu as f64 / down_pkts.max(1) as f64, + ); + } +} + +/// Runs `guest` in the guest namespace while the session runs; `relay` rewrites the guest address the way the host's shared TUN does. +fn with_link( + io: Io, + offloads: bool, + relay: Option<(Arc, Arc)>, + guest: impl FnOnce(i32) -> T + Send + 'static, +) -> (T, u16) { + let (guest_end, host_end, relay_threads) = match relay { + Some((up, down)) => { + let (g, rg) = UnixStream::pair().unwrap(); + let (rh, h) = UnixStream::pair().unwrap(); + let threads = vec![ + forward( + rg.try_clone().unwrap(), + rh.try_clone().unwrap(), + up, + IP_SRC, + tun::GUEST_ADDR, + CLAIM_ADDR, + ), + forward(rh, rg, down, IP_DST, CLAIM_ADDR, tun::GUEST_ADDR), + ]; + (g, h, threads) + } + None => { + let (g, h) = UnixStream::pair().unwrap(); + (g, h, Vec::new()) + } + }; + let stop = Arc::new(AtomicBool::new(false)); + let host_stop = guest_end.try_clone().unwrap(); + let (host, accepts) = spawn_host( + host_end, + io, + offloads, + relay_threads.is_empty(), + stop.clone(), + ); + let out = in_netns(move || { + let tun = Tun::open("tun0").unwrap(); + tun.configure_guest().unwrap(); + let shut = guest_end.try_clone().unwrap(); + let (tid_tx, tid_rx) = std::sync::mpsc::channel(); + let session = std::thread::Builder::new() + .name("guest-pump".into()) + .spawn(move || { + // SAFETY: gettid takes no arguments. + tid_tx.send(unsafe { libc::gettid() }).unwrap(); + tun::session(&tun, &guest_end, io) + }) + .unwrap(); + let tid = tid_rx.recv().unwrap(); + let out = guest(tid); + shut.shutdown(std::net::Shutdown::Both).unwrap(); + assert_eq!(session.join().unwrap().unwrap(), Flow::Closed); + out + }) + .join() + .unwrap(); + stop.store(true, Ordering::Relaxed); + let _ = host_stop.shutdown(std::net::Shutdown::Both); + host.join().unwrap(); + for t in relay_threads { + t.join().unwrap(); + } + (out, accepts) +} + +fn spawn_host( + stream: UnixStream, + io: Io, + offloads: bool, + direct: bool, + stop: Arc, +) -> (JoinHandle<()>, u16) { + let (ready_tx, ready_rx) = std::sync::mpsc::channel(); + let handle = in_netns(move || { + let tun = Tun::open("tunh").unwrap(); + if direct { + sysutil::if_up("tunh", PEER_ADDR, tun::PREFIX).unwrap(); + } else { + sysutil::if_up("tunh", PEER_ADDR, 32).unwrap(); + sysutil::route_add("tunh", Ipv4Addr::new(100, 65, 0, 0), 16, None).unwrap(); + } + let shut = stream.try_clone().unwrap(); + let (bound_tx, bound_rx) = std::sync::mpsc::channel(); + let servers = std::thread::spawn(move || { + runtime().block_on(async move { + let upload = TcpListener::bind((PEER_ADDR, UPLOAD)).await.unwrap(); + let download = TcpListener::bind((PEER_ADDR, DOWNLOAD)).await.unwrap(); + let echo = UdpSocket::bind((PEER_ADDR, ECHO)).await.unwrap(); + let sink = UdpSocket::bind((PEER_ADDR, SINK)).await.unwrap(); + let pps = UdpSocket::bind((PEER_ADDR, PPS_SINK)).await.unwrap(); + let source = TcpListener::bind((PEER_ADDR, PPS_SOURCE)).await.unwrap(); + tokio::spawn(serve_upload(upload)); + tokio::spawn(serve_download(download)); + tokio::spawn(serve_echo(echo)); + tokio::spawn(serve_count(sink, SINK, Some(UDP_SEG))); + tokio::spawn(serve_count(pps, PPS_SINK, None)); + tokio::spawn(serve_flood(source)); + bound_tx.send(()).unwrap(); + while !stop.load(Ordering::Relaxed) { + tokio::time::sleep(Duration::from_millis(20)).await; + } + }); + }); + bound_rx.recv().unwrap(); + ready_tx.send(tun.accepts()).unwrap(); + let fd = stream.as_fd(); + let mut b = [0u8; HELLO_LEN]; + sysutil::read_exact_fd(fd, &mut b, Duration::from_secs(5)).unwrap(); + let guest = Hello::decode(&b).unwrap(); + assert_eq!(guest.version, VERSION); + let mine = if offloads { tun.accepts() } else { 0 }; + let hello = Hello { + version: VERSION, + offloads: mine, + mtu: 1500, + flags: FLAG_ON, + }; + sysutil::write_all_fd(fd, &hello.encode()).unwrap(); + tun.agree(guest.offloads & mine).unwrap(); + let _ = tun::pump(&tun, fd, io); + drop(shut); + servers.join().unwrap(); + }); + let accepts = ready_rx.recv().unwrap(); + (handle, accepts) +} + +fn guest_workload() -> Seen { + runtime().block_on(async { + let upload_peer = upload(SocketAddr::from((PEER_ADDR, UPLOAD)), BULK).await; + let echo = UdpSocket::bind((Ipv4Addr::UNSPECIFIED, 0)).await.unwrap(); + echo.connect((PEER_ADDR, ECHO)).await.unwrap(); + let mut echo_peers = Vec::new(); + for i in 0..16u8 { + echo.send(&[i; 1200]).await.unwrap(); + let mut buf = [0u8; 64]; + let n = tokio::time::timeout(Duration::from_secs(2), echo.recv(&mut buf)) + .await + .unwrap() + .unwrap(); + echo_peers.push(String::from_utf8_lossy(&buf[..n]).into_owned()); + } + let downloaded = download(SocketAddr::from((PEER_ADDR, DOWNLOAD))).await; + let sink = std::net::UdpSocket::bind((Ipv4Addr::UNSPECIFIED, 0)).unwrap(); + let seg = UDP_SEG as libc::c_int; + // SAFETY: seg is a live c_int and the length matches its size. + let rc = unsafe { + libc::setsockopt( + sink.as_raw_fd(), + libc::SOL_UDP, + UDP_SEGMENT, + (&seg as *const libc::c_int).cast(), + size_of::() as libc::socklen_t, + ) + }; + assert_eq!(rc, 0, "UDP_SEGMENT: {}", std::io::Error::last_os_error()); + sink.send_to(&[7u8; UDP_DATAGRAMS * UDP_SEG], (PEER_ADDR, SINK)) + .unwrap(); + let sink_datagrams = query(SINK).await; + Seen { + upload_peer, + echo_peers, + downloaded, + sink_datagrams, + } + }) +} + +fn thread_cpu_ns(tid: i32) -> u64 { + std::fs::read_to_string(format!("/proc/self/task/{tid}/schedstat")) + .ok() + .and_then(|s| s.split_whitespace().next()?.parse().ok()) + .unwrap_or(0) +} + +fn pump_cpu_ns(pump_tid: i32) -> u64 { + let workers = format!("iou-wrk-{pump_tid}"); + let mut total = thread_cpu_ns(pump_tid); + for entry in std::fs::read_dir("/proc/self/task").unwrap().flatten() { + let comm = std::fs::read_to_string(entry.path().join("comm")).unwrap_or_default(); + if comm.trim() == workers + && let Ok(tid) = entry.file_name().to_string_lossy().parse::() + { + total += thread_cpu_ns(tid); + } + } + total +} + +fn runtime() -> tokio::runtime::Runtime { + tokio::runtime::Builder::new_current_thread() + .enable_all() + .build() + .unwrap() +} + +fn in_netns(f: impl FnOnce() -> T + Send + 'static) -> JoinHandle { + std::thread::spawn(move || { + unshare(CloneFlags::CLONE_NEWNET).unwrap(); + f() + }) +} + +fn gbits(bytes: u64, took: Duration) -> f64 { + bytes as f64 * 8.0 / took.as_secs_f64() / 1e9 +} + +async fn upload(target: SocketAddr, bytes: u64) -> String { + let mut s = TcpStream::connect(target).await.unwrap(); + let chunk = vec![0x5au8; 1 << 20]; + let mut left = bytes; + while left > 0 { + let n = left.min(chunk.len() as u64) as usize; + s.write_all(&chunk[..n]).await.unwrap(); + left -= n as u64; + } + s.shutdown().await.unwrap(); + let mut r = BufReader::new(s); + assert_eq!(r.read_u64().await.unwrap(), bytes); + let mut peer = String::new(); + r.read_line(&mut peer).await.unwrap(); + peer +} + +async fn download(target: SocketAddr) -> u64 { + let mut s = TcpStream::connect(target).await.unwrap(); + let mut buf = vec![0u8; 1 << 20]; + let mut got = 0u64; + loop { + let n = s.read(&mut buf).await.unwrap(); + if n == 0 { + return got; + } + got += n as u64; + } +} + +async fn query(port: u16) -> u64 { + tokio::time::sleep(Duration::from_millis(200)).await; + let mut s = TcpStream::connect((PEER_ADDR, port)).await.unwrap(); + s.read_u64().await.unwrap() +} + +async fn flood_up() -> u64 { + tokio::task::spawn_blocking(|| { + let s = std::net::UdpSocket::bind((Ipv4Addr::UNSPECIFIED, 0)).unwrap(); + s.connect((PEER_ADDR, PPS_SINK)).unwrap(); + let payload = [1u8; SMALL]; + let deadline = Instant::now() + FLOOD; + let mut sent = 0u64; + while Instant::now() < deadline { + for _ in 0..64 { + if s.send(&payload).is_ok() { + sent += 1; + } + } + } + sent + }) + .await + .unwrap() +} + +async fn flood_down() -> u64 { + let sink = std::net::UdpSocket::bind((Ipv4Addr::UNSPECIFIED, GUEST_PPS_PORT)).unwrap(); + sink.set_read_timeout(Some(Duration::from_millis(300))) + .unwrap(); + let mut s = TcpStream::connect((PEER_ADDR, PPS_SOURCE)).await.unwrap(); + s.write_u8(1).await.unwrap(); + tokio::task::spawn_blocking(move || { + let mut buf = [0u8; 2048]; + let mut got = 0u64; + while sink.recv(&mut buf).is_ok() { + got += 1; + } + got + }) + .await + .unwrap() +} + +async fn serve_upload(l: TcpListener) { + loop { + let (mut s, peer) = l.accept().await.unwrap(); + tokio::spawn(async move { + let mut buf = vec![0u8; 1 << 20]; + let mut got = 0u64; + while let Ok(n) = s.read(&mut buf).await { + if n == 0 { + break; + } + got += n as u64; + } + let _ = s.write_u64(got).await; + let _ = s.write_all(peer.ip().to_string().as_bytes()).await; + }); + } +} + +async fn serve_download(l: TcpListener) { + loop { + let (mut s, _) = l.accept().await.unwrap(); + tokio::spawn(async move { + let chunk = vec![0xa5u8; 1 << 20]; + for _ in 0..BULK / chunk.len() as u64 { + if s.write_all(&chunk).await.is_err() { + return; + } + } + }); + } +} + +async fn serve_echo(s: UdpSocket) { + let mut buf = [0u8; 2048]; + loop { + let (_, peer) = s.recv_from(&mut buf).await.unwrap(); + let _ = s.send_to(peer.ip().to_string().as_bytes(), peer).await; + } +} + +async fn serve_count(s: UdpSocket, port: u16, size: Option) { + let report = TcpListener::bind((PEER_ADDR, port)).await.unwrap(); + let got = Arc::new(AtomicU64::new(0)); + tokio::spawn({ + let got = Arc::clone(&got); + async move { + let mut buf = [0u8; 2048]; + loop { + let (n, _) = s.recv_from(&mut buf).await.unwrap(); + if size.is_none_or(|want| want == n) { + got.fetch_add(1, Ordering::Relaxed); + } + } + } + }); + loop { + let (mut c, _) = report.accept().await.unwrap(); + let _ = c.write_u64(got.swap(0, Ordering::Relaxed)).await; + } +} + +async fn serve_flood(l: TcpListener) { + loop { + let (mut c, peer) = l.accept().await.unwrap(); + let _ = c.read_u8().await; + let target = SocketAddr::from((peer.ip(), GUEST_PPS_PORT)); + tokio::task::spawn_blocking(move || { + let s = std::net::UdpSocket::bind((PEER_ADDR, 0)).unwrap(); + s.connect(target).unwrap(); + let payload = [2u8; SMALL]; + let deadline = Instant::now() + FLOOD; + while Instant::now() < deadline { + for _ in 0..64 { + let _ = s.send(&payload); + } + } + }); + } +} + +/// Copies frames one way, rewriting the IPv4 address at header offset `at` from `from` to `to`. +fn forward( + r: UnixStream, + mut w: UnixStream, + stats: Arc, + at: usize, + from: Ipv4Addr, + to: Ipv4Addr, +) -> JoinHandle<()> { + std::thread::spawn(move || { + use std::io::{Read, Write}; + + let mut r = std::io::BufReader::with_capacity(256 << 10, r); + let mut hello = [0u8; HELLO_LEN]; + if r.read_exact(&mut hello).is_err() || w.write_all(&hello).is_err() { + return; + } + let mut out = Vec::with_capacity(256 << 10); + let mut frame = vec![0u8; tun::MAX_FRAME]; + loop { + let mut head = [0u8; 4]; + if r.read_exact(&mut head).is_err() { + let _ = w.shutdown(std::net::Shutdown::Write); + return; + } + let len = u32::from_le_bytes(head) as usize; + if r.read_exact(&mut frame[..len]).is_err() { + return; + } + let body = &mut frame[..len]; + rewrite(body, at, from.octets(), to.octets()); + stats.frames.fetch_add(1, Ordering::Relaxed); + stats.largest.fetch_max(len as u64, Ordering::Relaxed); + if body[1] != GSO_NONE { + stats.gso.fetch_add(1, Ordering::Relaxed); + } + if body[1] == GSO_UDP_L4 { + stats.uso.fetch_add(1, Ordering::Relaxed); + } + out.extend_from_slice(&head); + out.extend_from_slice(body); + if frame_buffered(r.buffer()) && out.len() <= 192 << 10 { + continue; + } + let res = w.write_all(&out); + out.clear(); + if res.is_err() { + return; + } + } + }) +} + +fn frame_buffered(buf: &[u8]) -> bool { + tun::next_frame(buf).ok().flatten().is_some() +} + +/// Swaps an IPv4 address at header offset `at`, patching the IP checksum and the L4 checksum (partial or full). +fn rewrite(frame: &mut [u8], at: usize, from: [u8; 4], to: [u8; 4]) { + let partial = frame[0] & NEEDS_CSUM != 0; + let csum_at = usize::from(u16::from_le_bytes([frame[6], frame[7]])) + + usize::from(u16::from_le_bytes([frame[8], frame[9]])); + let ip = &mut frame[VNET_HDR_LEN..]; + if ip[0] >> 4 != 4 || ip[at..at + 4] != from { + return; + } + ip[at..at + 4].copy_from_slice(&to); + let check = u16::from_be_bytes([ip[10], ip[11]]); + ip[10..12].copy_from_slice(&(!adjust(!check, from, to)).to_be_bytes()); + let l4 = usize::from(ip[0] & 0x0f) * 4; + let field = if partial { + csum_at + } else { + match ip[9] { + PROTO_TCP => l4 + 16, + PROTO_UDP => l4 + 6, + _ => return, + } + }; + let sum = u16::from_be_bytes([ip[field], ip[field + 1]]); + let next = if partial { + adjust(sum, from, to) + } else if ip[9] == PROTO_UDP && sum == 0 { + 0 + } else { + !adjust(!sum, from, to) + }; + ip[field..field + 2].copy_from_slice(&next.to_be_bytes()); +} + +/// Moves a ones'-complement sum from one address to another (RFC 1624). +fn adjust(sum: u16, from: [u8; 4], to: [u8; 4]) -> u16 { + let mut s = u32::from(sum); + for i in [0, 2] { + s += u32::from(!u16::from_be_bytes([from[i], from[i + 1]])); + s += u32::from(u16::from_be_bytes([to[i], to[i + 1]])); + } + while s >> 16 != 0 { + s = (s & 0xffff) + (s >> 16); + } + s as u16 +}