diff --git a/easytier-core/src/gateway/dataplane/mod.rs b/easytier-core/src/gateway/dataplane/mod.rs index b8b2a8d7..9ec13971 100644 --- a/easytier-core/src/gateway/dataplane/mod.rs +++ b/easytier-core/src/gateway/dataplane/mod.rs @@ -176,7 +176,19 @@ const TCP_ENTRY: FlowKind = FlowKind::Tcp; const TCP_LISTEN_ENTRY: FlowKind = FlowKind::TcpListen; type FlowSet = Arc>; -type DataPlaneTcpIo = Box; +pub(super) enum DataPlaneTcpIo { + Smoltcp(crate::gateway::smoltcp::TcpStream), + External(Box), +} + +impl DataPlaneTcpIo { + fn as_mut(&mut self) -> &mut dyn TcpProxyStream { + match self { + Self::Smoltcp(stream) => stream, + Self::External(stream) => &mut **stream, + } + } +} pub(crate) struct DataPlaneRuntime where @@ -688,7 +700,7 @@ where generation.ensure_open()?; Ok(DataPlaneTcpStream::new( - Box::new(stream), + DataPlaneTcpIo::Smoltcp(stream), local_addr, Some(data_plane_ref), DataPlaneTcpStreamRoute::Outbound { _flow: flow }, @@ -710,7 +722,7 @@ where .await?; let local_addr = socket.local_addr().map_err(DataPlaneError::from)?; Ok(DataPlaneTcpStream::new( - Box::new(socket), + DataPlaneTcpIo::External(Box::new(socket)), local_addr, None, DataPlaneTcpStreamRoute::External, @@ -748,7 +760,7 @@ where ))) .await?; Ok(DataPlaneTcpStream::new( - stream, + DataPlaneTcpIo::External(stream), source_addr, None, DataPlaneTcpStreamRoute::External, diff --git a/easytier-core/src/gateway/dataplane/tcp.rs b/easytier-core/src/gateway/dataplane/tcp.rs index dea9a521..d645ada9 100644 --- a/easytier-core/src/gateway/dataplane/tcp.rs +++ b/easytier-core/src/gateway/dataplane/tcp.rs @@ -25,7 +25,7 @@ pub(super) enum DataPlaneTcpStreamRoute { /// A TCP stream created by the data-plane API. pub struct DataPlaneTcpStream { - pub(super) stream: DataPlaneTcpIo, + stream: Option, pub(super) local_addr: SocketAddr, pub(super) _route: DataPlaneTcpStreamRoute, pub(super) _data_plane_lease: Option, @@ -53,7 +53,7 @@ impl DataPlaneTcpStream { generation: DataPlaneIoGuard, ) -> Self { Self { - stream, + stream: Some(stream), local_addr, _data_plane_lease: data_plane_lease, _route: route, @@ -68,6 +68,21 @@ impl DataPlaneTcpStream { } } +impl Drop for DataPlaneTcpStream { + fn drop(&mut self) { + if let Some(DataPlaneTcpIo::Smoltcp(stream)) = self.stream.take() { + if self.generation.is_closed() { + return; + } + // The application is done, but FIN retransmissions and the peer's + // FIN still need this flow and the network stack until TCP closes. + let route = std::mem::replace(&mut self._route, DataPlaneTcpStreamRoute::External); + let lease = self._data_plane_lease.take(); + stream.close((route, lease)); + } + } +} + impl DataPlaneTcpListener { pub fn local_addr(&self) -> SocketAddr { self.local_addr @@ -99,7 +114,7 @@ impl DataPlaneTcpListener { FlowData::DataPlaneRoute, ); let accepted = DataPlaneTcpStream::new( - Box::new(stream), + DataPlaneTcpIo::Smoltcp(stream), local_addr, Some(self.data_plane_lease.clone()), DataPlaneTcpStreamRoute::Accepted { _flow: flow }, @@ -119,7 +134,7 @@ impl AsyncRead for DataPlaneTcpStream { if this.generation.ensure_open().is_err() || this.read_closed.as_mut().poll(cx).is_ready() { return Poll::Ready(Err(this.generation.closed_io_error())); } - Pin::new(&mut this.stream).poll_read(cx, buf) + Pin::new(this.stream.as_mut().unwrap().as_mut()).poll_read(cx, buf) } } @@ -134,7 +149,7 @@ impl AsyncWrite for DataPlaneTcpStream { { return Poll::Ready(Err(this.generation.closed_io_error())); } - Pin::new(&mut this.stream).poll_write(cx, buf) + Pin::new(this.stream.as_mut().unwrap().as_mut()).poll_write(cx, buf) } fn poll_flush(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll> { @@ -143,7 +158,7 @@ impl AsyncWrite for DataPlaneTcpStream { { return Poll::Ready(Err(this.generation.closed_io_error())); } - Pin::new(&mut this.stream).poll_flush(cx) + Pin::new(this.stream.as_mut().unwrap().as_mut()).poll_flush(cx) } fn poll_shutdown( @@ -155,6 +170,6 @@ impl AsyncWrite for DataPlaneTcpStream { { return Poll::Ready(Err(this.generation.closed_io_error())); } - Pin::new(&mut this.stream).poll_shutdown(cx) + Pin::new(this.stream.as_mut().unwrap().as_mut()).poll_shutdown(cx) } } diff --git a/easytier-core/src/gateway/dataplane/tests.rs b/easytier-core/src/gateway/dataplane/tests.rs index 250bd00e..d22f1052 100644 --- a/easytier-core/src/gateway/dataplane/tests.rs +++ b/easytier-core/src/gateway/dataplane/tests.rs @@ -542,9 +542,29 @@ async fn listener_and_accepted_stream_own_independent_flow_lifetimes() { assert_eq!(&buf, b"after-listener-drop"); drop(client); - assert_eq!(a.gateway.entries.count(), 0); + assert_eq!(a.gateway.entries.count(), 1); + assert_eq!(server.read(&mut buf).await.unwrap(), 0); + // The closed application no longer reads, but TCP must keep draining + // incoming data so the peer's queued data and FIN can finish sending. + tokio::time::timeout(timeout, async { + server.write_all(&vec![0; 1024 * 1024]).await.unwrap(); + server.shutdown().await.unwrap(); + }) + .await + .unwrap(); drop(server); - assert_eq!(b.gateway.entries.count(), 0); + + tokio::time::timeout(Duration::from_secs(15), async { + while a.gateway.entries.count() != 0 + || b.gateway.entries.count() != 0 + || a.gateway.net.lock().await.is_some() + || b.gateway.net.lock().await.is_some() + { + tokio::time::sleep(Duration::from_millis(10)).await; + } + }) + .await + .expect("TCP teardown must release the last flow and stack leases"); stop_data_plane_pair(&a, &b).await; } diff --git a/easytier-core/src/gateway/smoltcp/mod.rs b/easytier-core/src/gateway/smoltcp/mod.rs index 9e8128f4..8f354e8a 100644 --- a/easytier-core/src/gateway/smoltcp/mod.rs +++ b/easytier-core/src/gateway/smoltcp/mod.rs @@ -3,5 +3,5 @@ mod tokio_smoltcp; pub(super) use stack::{SmolTcpStack, output_dst_ip}; pub(super) use tokio_smoltcp::{ - BufferSize, Net, NetConfig, TcpListener, UdpSocket, channel_device, + BufferSize, Net, NetConfig, TcpListener, TcpStream, UdpSocket, channel_device, }; diff --git a/easytier-core/src/gateway/smoltcp/tokio_smoltcp/reactor.rs b/easytier-core/src/gateway/smoltcp/tokio_smoltcp/reactor.rs index 46dd7f48..a5664c39 100644 --- a/easytier-core/src/gateway/smoltcp/tokio_smoltcp/reactor.rs +++ b/easytier-core/src/gateway/smoltcp/tokio_smoltcp/reactor.rs @@ -24,7 +24,10 @@ use std::{ sync::Arc, task::{Context as TaskContext, Poll, Waker}, }; -use tokio::{pin, select, sync::Notify}; +use tokio::{ + pin, select, + sync::{Notify, mpsc}, +}; use crate::foundation::time::sleep; @@ -50,11 +53,18 @@ struct TcpListenerRegistry { type SharedTcpListenerRegistry = Arc>; +struct ClosingTcpSocket { + handle: OwnedSocketHandle, + deadline: Instant, + _keep_alive: Box, +} + pub(crate) struct Reactor { notify: Arc, iface: BufferInterface, socket_allocator: SocketAlloctor, tcp_listeners: SharedTcpListenerRegistry, + closing_tcp: mpsc::UnboundedSender, } async fn receive( @@ -203,12 +213,14 @@ fn update_tcp_listeners( wakers } +#[allow(clippy::too_many_arguments)] async fn run( mut async_iface: impl super::device::AsyncDevice, iface: BufferInterface, mut device: BufferDevice, socket_allocator: SocketAlloctor, tcp_listeners: SharedTcpListenerRegistry, + mut closing_tcp: mpsc::UnboundedReceiver, notify: Arc, stopper: Arc, ) -> io::Result<()> { @@ -219,21 +231,36 @@ async fn run( .max_burst_size .unwrap_or(MAX_BURST_SIZE); let mut recv_buf = VecDeque::with_capacity(max_burst_size); + let mut closing = Vec::::new(); + let mut closed = Vec::new(); pin!(timer); loop { let packets = device.take_send_queue(); async_iface.send_all(&mut iter(packets).map(Ok)).await?; + // Keep routing and stack leases until the last protocol output has + // left the device queue. Drop handles outside the socket-set lock. + closed.clear(); + while let Ok(socket) = closing_tcp.try_recv() { + closing.push(socket); + } if recv_buf.is_empty() && device.need_wait() { let start = Instant::now(); - let deadline = { + let mut deadline = { iface .lock() .poll_delay(start, &socket_allocator.sockets().lock()) .unwrap_or(default_timeout) }; + for socket in &closing { + deadline = deadline.min(if socket.deadline <= start { + Duration::ZERO + } else { + socket.deadline - start + }); + } if deadline != Duration::ZERO { timer @@ -243,6 +270,7 @@ async fn run( _ = &mut timer => {}, _ = receive(&mut async_iface,&mut recv_buf) => {} _ = notify.notified() => {} + Some(socket) = closing_tcp.recv() => closing.push(socket), _ = stopper.notified() => break, }; } @@ -290,6 +318,21 @@ async fn run( iface.poll(now, &mut device, &mut sockets); listener_wakers.extend(update_tcp_listeners(&mut listeners, &mut sockets)); + if prepare_tcp_close(now, &closing, &mut sockets) { + iface.poll(now, &mut device, &mut sockets); + } + let mut index = 0; + while index < closing.len() { + let socket = sockets.get::(*closing[index].handle); + // Closed can still have an RST to dispatch. TIME-WAIT must also + // remain available to acknowledge a retransmitted peer FIN. + if socket.state() == tcp::State::Closed && socket.remote_endpoint().is_none() { + closed.push(closing.swap_remove(index)); + } else { + index += 1; + } + } + // wake up all closed sockets (smoltcp seems have a bug that it doesn't wake up closed sockets) for (_, socket) in sockets.iter_mut() { if let Socket::Tcp(tcp) = socket @@ -309,6 +352,30 @@ async fn run( Ok(()) } +fn prepare_tcp_close( + now: Instant, + closing: &[ClosingTcpSocket], + sockets: &mut SocketSet<'static>, +) -> bool { + let mut changed = false; + for entry in closing { + let socket = sockets.get_mut::(*entry.handle); + // There is no application reader anymore. Keep the receive window + // open so queued peer data cannot prevent its FIN from reaching us. + if socket.can_recv() { + let _ = socket.recv(|data| (data.len(), ())); + changed = true; + } + // The normal TCP timeout measures inactivity, so ACKs can extend it. + // A released application resource has an absolute teardown deadline. + if now >= entry.deadline && socket.state() != tcp::State::Closed { + socket.abort(); + changed = true; + } + } + changed +} + impl Reactor { pub fn new( async_device: impl super::device::AsyncDevice, @@ -321,12 +388,14 @@ impl Reactor { let notify = Arc::new(Notify::new()); let socket_allocator = SocketAlloctor::new(buffer_size); let tcp_listeners = Arc::new(Mutex::new(TcpListenerRegistry::default())); + let (closing_tx, closing_rx) = mpsc::unbounded_channel(); let fut = run( async_device, iface.clone(), device, socket_allocator.clone(), tcp_listeners.clone(), + closing_rx, notify.clone(), stopper, ); @@ -337,6 +406,7 @@ impl Reactor { iface, socket_allocator, tcp_listeners, + closing_tcp: closing_tx, }, fut, ) @@ -350,6 +420,21 @@ impl Reactor { |sockets: &mut smoltcp::iface::SocketSet<'_>| sockets.get_mut::(handle), ) } + pub(super) fn close_tcp( + &self, + handle: OwnedSocketHandle, + keep_alive: Box, + deadline: Instant, + ) { + self.get_socket::(*handle).close(); + // The receiver and all closing sockets belong to run(), so stopping + // the reactor also releases them without a detached cleanup task. + let _ = self.closing_tcp.send(ClosingTcpSocket { + handle, + deadline, + _keep_alive: keep_alive, + }); + } pub fn context(&self) -> MappedMutexGuard<'_, Context> { MutexGuard::map(self.iface.lock(), |iface| iface.context()) } diff --git a/easytier-core/src/gateway/smoltcp/tokio_smoltcp/socket.rs b/easytier-core/src/gateway/smoltcp/tokio_smoltcp/socket.rs index 714e5c46..ba38a33e 100644 --- a/easytier-core/src/gateway/smoltcp/tokio_smoltcp/socket.rs +++ b/easytier-core/src/gateway/smoltcp/tokio_smoltcp/socket.rs @@ -97,6 +97,15 @@ pub struct TcpStream { } impl TcpStream { + /// Release the application stream while the reactor finishes TCP teardown. + pub(crate) fn close(self, keep_alive: impl Send + 'static) { + self.reactor.close_tcp( + self.handle, + Box::new(keep_alive), + smoltcp::time::Instant::now() + super::socket_allocator::TCP_TIMEOUT, + ); + } + pub(super) async fn connect( reactor: Arc, local_endpoint: IpEndpoint, @@ -331,21 +340,21 @@ impl UdpSocket { #[cfg(test)] mod tests { - use std::time::Duration as StdDuration; + use std::{sync::Arc, time::Duration as StdDuration}; use smoltcp::{ iface::Config, phy::{DeviceCapabilities, Medium}, socket::tcp, time::Duration, - wire::{HardwareAddress, Ipv4Address, TcpControl, TcpSeqNumber}, + wire::{HardwareAddress, Ipv4Address, Ipv4Packet, TcpControl, TcpPacket, TcpSeqNumber}, }; use super::super::{ Net, NetConfig, channel_device, test_utils::{TcpPackets, recv_tcp, recv_tcp_for_port}, }; - use super::{TCP_LISTENER_MAX_PENDING, TcpListener}; + use super::{TCP_LISTENER_MAX_PENDING, TcpListener, TcpStream}; const LISTEN_ADDR: Ipv4Address = Ipv4Address::new(10, 126, 126, 1); const LISTEN_PORT: u16 = 34569; @@ -417,6 +426,136 @@ mod tests { .collect() } + async fn connected_stream( + net: &Net, + ingress: &tokio::sync::mpsc::Sender>>, + egress: &mut tokio::sync::mpsc::Receiver>, + ) -> TcpStream { + let mut listener = net + .tcp_bind("10.126.126.1:34569".parse().unwrap()) + .await + .unwrap(); + let client_addr = Ipv4Address::new(10, 126, 126, 2); + let sequence = begin_handshake(ingress, egress, client_addr, 40000, 1000).await; + finish_handshake(ingress, client_addr, 40000, 1000, sequence).await; + tokio::time::timeout(StdDuration::from_secs(1), listener.accept()) + .await + .unwrap() + .unwrap() + .0 + } + + #[tokio::test] + async fn closing_stream_retransmits_fin_and_retains_time_wait() { + let (net, ingress, mut egress) = test_net(); + let stream = connected_stream(&net, &ingress, &mut egress).await; + let reactor = stream.reactor.clone(); + let keep_alive = Arc::new(()); + let retained = Arc::downgrade(&keep_alive); + stream.close(keep_alive); + + let fin = recv_tcp(&mut egress).await; + assert_eq!(fin.control, TcpControl::Fin); + // Lose the first FIN. The application handle is gone, but the reactor + // must still retransmit the same sequence number. + let retransmit = tokio::time::timeout(StdDuration::from_secs(3), egress.recv()) + .await + .unwrap() + .unwrap(); + let ipv4 = Ipv4Packet::new_checked(&retransmit).unwrap(); + let tcp = TcpPacket::new_checked(ipv4.payload()).unwrap(); + assert!(tcp.fin()); + assert_eq!(tcp.seq_number(), fin.sequence); + + let client_addr = Ipv4Address::new(10, 126, 126, 2); + let peer_fin = PACKETS.fin(client_addr, 40000, TcpSeqNumber(1001), fin.sequence + 1); + ingress.send(Ok(peer_fin.clone())).await.unwrap(); + let ack = recv_tcp(&mut egress).await; + assert_eq!(ack.control, TcpControl::None); + assert_eq!(ack.ack_number, Some(TcpSeqNumber(1002))); + assert!(retained.upgrade().is_some()); + + // Lose our final ACK. TIME-WAIT must acknowledge the repeated FIN. + ingress.send(Ok(peer_fin)).await.unwrap(); + let ack = recv_tcp(&mut egress).await; + assert_eq!(ack.control, TcpControl::None); + assert_eq!(ack.ack_number, Some(TcpSeqNumber(1002))); + tokio::time::timeout(StdDuration::from_secs(15), async { + while retained.upgrade().is_some() { + tokio::time::sleep(StdDuration::from_millis(10)).await; + } + }) + .await + .expect("TIME-WAIT must eventually release the socket and its leases"); + assert_eq!( + reactor.socket_allocator().sockets().lock().iter().count(), + 0 + ); + } + + #[tokio::test] + async fn closing_stream_deadline_sends_reset_and_releases_socket() { + let (net, ingress, mut egress) = test_net(); + let stream = connected_stream(&net, &ingress, &mut egress).await; + let reactor = stream.reactor.clone(); + let keep_alive = Arc::new(()); + let retained = Arc::downgrade(&keep_alive); + // Exercise the same reactor deadline with a short test budget. + reactor.close_tcp( + stream.handle, + Box::new(keep_alive), + smoltcp::time::Instant::now() + Duration::from_millis(150), + ); + let fin = recv_tcp(&mut egress).await; + assert_eq!(fin.control, TcpControl::Fin); + ingress + .send(Ok(PACKETS.ack( + Ipv4Address::new(10, 126, 126, 2), + 40000, + TcpSeqNumber(1001), + fin.sequence + 1, + ))) + .await + .unwrap(); + // The peer ACKs our FIN but never closes its own half (FIN-WAIT-2). + let reset = recv_tcp(&mut egress).await; + assert_eq!(reset.control, TcpControl::Rst); + tokio::time::timeout(StdDuration::from_secs(1), async { + while retained.upgrade().is_some() { + tokio::task::yield_now().await; + } + }) + .await + .unwrap(); + assert_eq!( + reactor.socket_allocator().sockets().lock().iter().count(), + 0 + ); + } + + #[tokio::test] + async fn stopping_reactor_releases_closing_streams() { + let (net, ingress, mut egress) = test_net(); + let stream = connected_stream(&net, &ingress, &mut egress).await; + let reactor = stream.reactor.clone(); + let keep_alive = Arc::new(()); + let retained = Arc::downgrade(&keep_alive); + stream.close(keep_alive); + assert_eq!(recv_tcp(&mut egress).await.control, TcpControl::Fin); + drop(net); + tokio::time::timeout(StdDuration::from_secs(1), async { + while retained.upgrade().is_some() { + tokio::task::yield_now().await; + } + }) + .await + .expect("stopping the reactor must release unfinished TCP teardown"); + assert_eq!( + reactor.socket_allocator().sockets().lock().iter().count(), + 0 + ); + } + #[tokio::test] async fn duplicate_syn_does_not_consume_another_backlog_slot() { let (net, ingress, mut egress) = test_net(); diff --git a/easytier-core/src/gateway/smoltcp/tokio_smoltcp/test_utils.rs b/easytier-core/src/gateway/smoltcp/tokio_smoltcp/test_utils.rs index 8ed40d94..aa3cff2a 100644 --- a/easytier-core/src/gateway/smoltcp/tokio_smoltcp/test_utils.rs +++ b/easytier-core/src/gateway/smoltcp/tokio_smoltcp/test_utils.rs @@ -114,6 +114,7 @@ pub(crate) struct TcpPacketSummary { pub(crate) control: TcpControl, pub(crate) sequence: TcpSeqNumber, pub(crate) dst_port: u16, + pub(crate) ack_number: Option, } pub(crate) async fn recv_tcp(receiver: &mut mpsc::Receiver>) -> TcpPacketSummary { @@ -133,6 +134,7 @@ pub(crate) async fn recv_tcp(receiver: &mut mpsc::Receiver>) -> TcpPacke control: repr.control, sequence: repr.seq_number, dst_port: repr.dst_port, + ack_number: repr.ack_number, } } diff --git a/easytier-go/internal/artifact/easytier_core.wasm b/easytier-go/internal/artifact/easytier_core.wasm index db02e556..3036135b 100644 Binary files a/easytier-go/internal/artifact/easytier_core.wasm and b/easytier-go/internal/artifact/easytier_core.wasm differ diff --git a/easytier-go/internal/artifact/provenance.go b/easytier-go/internal/artifact/provenance.go index afed2653..8e5891cb 100644 --- a/easytier-go/internal/artifact/provenance.go +++ b/easytier-go/internal/artifact/provenance.go @@ -3,6 +3,6 @@ package artifact const ( - Commit = "c51437c673a71f69b4b028c8d566a7b23b33e65c" - SHA256 = "17d6dd4561c1f1e818e7d70ed031c3cf8ea5dcfc1fd0461a6753f1f23f5fa6ab" + Commit = "5f1eaa7c9f9d3e89e9b75a6d87531a436fae40c4" + SHA256 = "de332a01eb4141e2e86ecee89500c40cc7d07a822a381868f95bb231bfc3739c" ) diff --git a/easytier-go/proto/api/config/api_config.pb.go b/easytier-go/proto/api/config/api_config.pb.go index 909bfa07..03d695cc 100644 --- a/easytier-go/proto/api/config/api_config.pb.go +++ b/easytier-go/proto/api/config/api_config.pb.go @@ -93,8 +93,11 @@ type InstanceConfigPatch struct { VpnPortalClients []*VpnPortalClientPatch `protobuf:"bytes,15,rep,name=vpn_portal_clients,json=vpnPortalClients,proto3" json:"vpn_portal_clients,omitempty"` ManagedCredentials *manage.ManagedCredentialSet `protobuf:"bytes,16,opt,name=managed_credentials,json=managedCredentials,proto3" json:"managed_credentials,omitempty"` PreferPeerRelay *bool `protobuf:"varint,17,opt,name=prefer_peer_relay,json=preferPeerRelay,proto3,oneof" json:"prefer_peer_relay,omitempty"` - unknownFields protoimpl.UnknownFields - sizeCache protoimpl.SizeCache + // Applies with routes: true preserves an empty manual list; false restores + // automatic routes. Omission preserves legacy RoutePatch behavior. + EnableManualRoutes *bool `protobuf:"varint,18,opt,name=enable_manual_routes,json=enableManualRoutes,proto3,oneof" json:"enable_manual_routes,omitempty"` + unknownFields protoimpl.UnknownFields + sizeCache protoimpl.SizeCache } func (x *InstanceConfigPatch) Reset() { @@ -246,6 +249,13 @@ func (x *InstanceConfigPatch) GetPreferPeerRelay() bool { return false } +func (x *InstanceConfigPatch) GetEnableManualRoutes() bool { + if x != nil && x.EnableManualRoutes != nil { + return *x.EnableManualRoutes + } + return false +} + type VpnPortalClientPatch struct { state protoimpl.MessageState `protogen:"open.v1"` Action ConfigPatchAction `protobuf:"varint,1,opt,name=action,proto3,enum=api.config.ConfigPatchAction" json:"action,omitempty"` @@ -867,7 +877,7 @@ var File_api_config_proto protoreflect.FileDescriptor const file_api_config_proto_rawDesc = "" + "\n" + "\x10api_config.proto\x12\n" + - "api.config\x1a\fcommon.proto\x1a\tacl.proto\x1a\x12api_instance.proto\x1a\x10api_manage.proto\"\x86\t\n" + + "api.config\x1a\fcommon.proto\x1a\tacl.proto\x1a\x12api_instance.proto\x1a\x10api_manage.proto\"\xd6\t\n" + "\x13InstanceConfigPatch\x12\x1f\n" + "\bhostname\x18\x01 \x01(\tH\x00R\bhostname\x88\x01\x01\x12)\n" + "\x04ipv4\x18\x02 \x01(\v2\x10.common.Ipv4InetH\x01R\x04ipv4\x88\x01\x01\x12)\n" + @@ -889,7 +899,8 @@ const file_api_config_proto_rawDesc = "" + "\x12disable_relay_data\x18\x0e \x01(\bH\aR\x10disableRelayData\x88\x01\x01\x12N\n" + "\x12vpn_portal_clients\x18\x0f \x03(\v2 .api.config.VpnPortalClientPatchR\x10vpnPortalClients\x12Q\n" + "\x13managed_credentials\x18\x10 \x01(\v2 .api.manage.ManagedCredentialSetR\x12managedCredentials\x12/\n" + - "\x11prefer_peer_relay\x18\x11 \x01(\bH\bR\x0fpreferPeerRelay\x88\x01\x01B\v\n" + + "\x11prefer_peer_relay\x18\x11 \x01(\bH\bR\x0fpreferPeerRelay\x88\x01\x01\x125\n" + + "\x14enable_manual_routes\x18\x12 \x01(\bH\tR\x12enableManualRoutes\x88\x01\x01B\v\n" + "\t_hostnameB\a\n" + "\x05_ipv4B\a\n" + "\x05_ipv6B\x06\n" + @@ -898,7 +909,8 @@ const file_api_config_proto_rawDesc = "" + "\x16_ipv6_public_addr_autoB\x1a\n" + "\x18_ipv6_public_addr_prefixB\x15\n" + "\x13_disable_relay_dataB\x14\n" + - "\x12_prefer_peer_relay\"\x88\x01\n" + + "\x12_prefer_peer_relayB\x17\n" + + "\x15_enable_manual_routes\"\x88\x01\n" + "\x14VpnPortalClientPatch\x125\n" + "\x06action\x18\x01 \x01(\x0e2\x1d.api.config.ConfigPatchActionR\x06action\x129\n" + "\x06client\x18\x02 \x01(\v2!.api.manage.VpnPortalClientConfigR\x06client\"x\n" + diff --git a/easytier-go/proto/provenance.go b/easytier-go/proto/provenance.go index 4c6318e4..10755a04 100644 --- a/easytier-go/proto/provenance.go +++ b/easytier-go/proto/provenance.go @@ -3,6 +3,6 @@ package proto const ( - EasyTierCommit = "c51437c673a71f69b4b028c8d566a7b23b33e65c" - SchemaSHA256 = "e49a4af09e5b66089eebb73fda78ffca7e8468a5a938cf7a593d689881686f0a" + EasyTierCommit = "5f1eaa7c9f9d3e89e9b75a6d87531a436fae40c4" + SchemaSHA256 = "33516bc93a3fd43df341ce081a904c1f8c5fd00ea5c33d78925e2359312a926c" ) diff --git a/easytier-go/proto/web/web.pb.go b/easytier-go/proto/web/web.pb.go index ece7eb55..a6c9fff8 100644 --- a/easytier-go/proto/web/web.pb.go +++ b/easytier-go/proto/web/web.pb.go @@ -95,8 +95,11 @@ type HeartbeatRequest struct { SupportConfigSource bool `protobuf:"varint,9,opt,name=support_config_source,json=supportConfigSource,proto3" json:"support_config_source,omitempty"` FailedNetworkInstances []*common.UUID `protobuf:"bytes,10,rep,name=failed_network_instances,json=failedNetworkInstances,proto3" json:"failed_network_instances,omitempty"` SupportHeartbeatPolicy bool `protobuf:"varint,11,opt,name=support_heartbeat_policy,json=supportHeartbeatPolicy,proto3" json:"support_heartbeat_policy,omitempty"` - unknownFields protoimpl.UnknownFields - sizeCache protoimpl.SizeCache + // Supports explicit manual mode patches and typed empty-manual readback. + // Older peers silently ignore unknown patch fields; missing means false. + SupportManualRoutesPatch bool `protobuf:"varint,12,opt,name=support_manual_routes_patch,json=supportManualRoutesPatch,proto3" json:"support_manual_routes_patch,omitempty"` + unknownFields protoimpl.UnknownFields + sizeCache protoimpl.SizeCache } func (x *HeartbeatRequest) Reset() { @@ -206,6 +209,13 @@ func (x *HeartbeatRequest) GetSupportHeartbeatPolicy() bool { return false } +func (x *HeartbeatRequest) GetSupportManualRoutesPatch() bool { + if x != nil { + return x.SupportManualRoutesPatch + } + return false +} + type HeartbeatResponse struct { state protoimpl.MessageState `protogen:"open.v1"` HeartbeatIntervalMs *uint32 `protobuf:"varint,1,opt,name=heartbeat_interval_ms,json=heartbeatIntervalMs,proto3,oneof" json:"heartbeat_interval_ms,omitempty"` @@ -346,7 +356,7 @@ const file_web_proto_rawDesc = "" + "\fDeviceOsInfo\x12\x17\n" + "\aos_type\x18\x01 \x01(\tR\x06osType\x12\x18\n" + "\aversion\x18\x02 \x01(\tR\aversion\x12\"\n" + - "\fdistribution\x18\x03 \x01(\tR\fdistribution\"\x9d\x04\n" + + "\fdistribution\x18\x03 \x01(\tR\fdistribution\"\xdc\x04\n" + "\x10HeartbeatRequest\x12+\n" + "\n" + "machine_id\x18\x01 \x01(\v2\f.common.UUIDR\tmachineId\x12%\n" + @@ -362,7 +372,8 @@ const file_web_proto_rawDesc = "" + "\x15support_config_source\x18\t \x01(\bR\x13supportConfigSource\x12F\n" + "\x18failed_network_instances\x18\n" + " \x03(\v2\f.common.UUIDR\x16failedNetworkInstances\x128\n" + - "\x18support_heartbeat_policy\x18\v \x01(\bR\x16supportHeartbeatPolicy\"\xb6\x01\n" + + "\x18support_heartbeat_policy\x18\v \x01(\bR\x16supportHeartbeatPolicy\x12=\n" + + "\x1bsupport_manual_routes_patch\x18\f \x01(\bR\x18supportManualRoutesPatch\"\xb6\x01\n" + "\x11HeartbeatResponse\x127\n" + "\x15heartbeat_interval_ms\x18\x01 \x01(\rH\x00R\x13heartbeatIntervalMs\x88\x01\x01\x125\n" + "\x14heartbeat_timeout_ms\x18\x02 \x01(\rH\x01R\x12heartbeatTimeoutMs\x88\x01\x01B\x18\n" + diff --git a/easytier-go/tests/host_dataplane_test.go b/easytier-go/tests/host_dataplane_test.go index b6027f12..880fc87f 100644 --- a/easytier-go/tests/host_dataplane_test.go +++ b/easytier-go/tests/host_dataplane_test.go @@ -186,6 +186,143 @@ func testTCPDataPlane( case <-ctx.Done(): t.Fatalf("blocked TCP read did not wake: %v", ctx.Err()) } + + // FIN gate: Close must half-close the stream so the peer's read ends with + // io.EOF. Releasing the resource without shutdown_write drops the socket + // instead, which leaves the peer ESTABLISHED forever (the exit-node leak). + if err := serverConnection.SetReadDeadline(time.Now().Add(5 * time.Second)); err != nil { + t.Fatalf("set server read deadline: %v", err) + } + if _, err := serverConnection.Read(make([]byte, 1)); !errors.Is(err, io.EOF) { + t.Fatalf("server read after client close = %v, want io.EOF (FIN not sent)", err) + } + if err := serverConnection.Close(); err != nil { + t.Fatalf("server close after EOF: %v", err) + } +} + +// Application deadlines must not prevent the guest from closing TCP. +func TestTCPCloseAfterExpiredDeadlines(t *testing.T) { + past := time.Now().Add(-time.Second) + for name, expire := range map[string]func(net.Conn) error{ + "write": func(c net.Conn) error { return c.SetWriteDeadline(past) }, + "full": func(c net.Conn) error { return c.SetDeadline(past) }, + } { + t.Run(name, func(t *testing.T) { + ctx, cancel := context.WithTimeout(context.Background(), 45*time.Second) + defer cancel() + host, server, client := startDataPlanePair(t, ctx) + defer host.Close(ctx) + defer server.Close(ctx) + defer client.Close(ctx) + + listener := listenTCPEventually(t, ctx, server) + defer listener.Close() + accepted := make(chan connectionResult, 1) + go func() { + connection, err := listener.Accept() + accepted <- connectionResult{connection: connection, err: err} + }() + clientConnection := dialEventually(t, ctx, client, + fmt.Sprintf("10.144.0.101:%d", listener.Addr().(*net.TCPAddr).Port)) + var serverConnection net.Conn + select { + case result := <-accepted: + if result.err != nil { + t.Fatalf("accept EasyTier TCP: %v", result.err) + } + serverConnection = result.connection + case <-ctx.Done(): + t.Fatalf("accept EasyTier TCP: %v", ctx.Err()) + } + + if err := expire(clientConnection); err != nil { + t.Fatalf("set expired client deadline: %v", err) + } + if err := clientConnection.Close(); err != nil { + t.Fatalf("client close after expired deadline: %v", err) + } + if err := serverConnection.SetReadDeadline(time.Now().Add(5 * time.Second)); err != nil { + t.Fatalf("set server read deadline: %v", err) + } + if _, err := serverConnection.Read(make([]byte, 1)); !errors.Is(err, io.EOF) { + t.Fatalf("server read after expired-deadline close = %v, want io.EOF", err) + } + if err := expire(serverConnection); err != nil { + t.Fatalf("set expired server deadline: %v", err) + } + if err := serverConnection.Close(); err != nil { + t.Fatalf("server close after expired deadline: %v", err) + } + }) + } +} + +func TestTCPCloseWithBlockedWrite(t *testing.T) { + ctx, cancel := context.WithTimeout(context.Background(), 45*time.Second) + defer cancel() + host, server, client := startDataPlanePair(t, ctx) + defer host.Close(ctx) + defer server.Close(ctx) + defer client.Close(ctx) + + listener := listenTCPEventually(t, ctx, server) + defer listener.Close() + accepted := make(chan connectionResult, 1) + go func() { + connection, err := listener.Accept() + accepted <- connectionResult{connection: connection, err: err} + }() + clientConnection := dialEventually(t, ctx, client, + fmt.Sprintf("10.144.0.101:%d", listener.Addr().(*net.TCPAddr).Port)) + defer clientConnection.Close() + var serverConnection net.Conn + select { + case result := <-accepted: + if result.err != nil { + t.Fatalf("accept EasyTier TCP: %v", result.err) + } + serverConnection = result.connection + case <-ctx.Done(): + t.Fatalf("accept EasyTier TCP: %v", ctx.Err()) + } + defer serverConnection.Close() + if err := serverConnection.SetReadDeadline(time.Now().Add(10 * time.Second)); err != nil { + t.Fatal(err) + } + + written := make(chan error, 1) + go func() { + _, err := clientConnection.Write(make([]byte, 1024*1024)) + written <- err + }() + // Observe real data before withholding reads to fill the TCP windows. + if _, err := io.ReadFull(serverConnection, make([]byte, 1)); err != nil { + t.Fatalf("read start of blocked write: %v", err) + } + select { + case err := <-written: + t.Fatalf("large write completed before Close: %v", err) + case <-time.After(100 * time.Millisecond): + } + if err := clientConnection.Close(); err != nil { + t.Fatalf("close with blocked write: %v", err) + } + select { + case err := <-written: + if !errors.Is(err, net.ErrClosed) { + t.Fatalf("blocked write after Close = %v, want net.ErrClosed", err) + } + case <-ctx.Done(): + t.Fatalf("Close did not unblock Write: %v", ctx.Err()) + } + // Bytes already accepted by TCP precede FIN; draining them must end in EOF. + if _, err := io.Copy(io.Discard, serverConnection); err != nil { + t.Fatalf("server read after Close = %v, want EOF", err) + } + if err := serverConnection.Close(); err != nil { + t.Fatalf("server close after EOF: %v", err) + } } func testUDPDataPlane(