From e0745f4babb5f8fa0f84f33cf24772fb428fd8f6 Mon Sep 17 00:00:00 2001 From: w568w <1278297578@qq.com> Date: Thu, 4 Jun 2026 17:17:41 +0800 Subject: [PATCH] feat: Add data plane support to FFI (#2287) 1. Overview This PR adds data plane APIs to easytier-ffi: TCP Outbound: - data_plane_tcp_connect - data_plane_tcp_read - data_plane_tcp_write - data_plane_tcp_close TCP Listener: - data_plane_tcp_bind - data_plane_tcp_accept - data_plane_tcp_listener_close UDP: - data_plane_udp_bind - data_plane_udp_send_to - data_plane_udp_recv_from - data_plane_udp_close 2. Key Changes The main changes are focused on: - easytier-contrib/easytier-ffi/src/lib.rs: Added FFI interfaces; made ERROR_MSG thread-safe. - easytier/src/gateway/socks5.rs: Bridges the data plane to the existing Socks5 server logic. - Added EasyTierUdpSocket, mainly wrapping ref-counting and critical object (e.g., Socks5EntrySet) hold & drop logic, and exposing common fields (e.g., local_addr). - Extended Socks5Server functionality to expose TCP and UDP socket creation interfaces for FFI calls. - Other files: Mostly pass-through logic. - Added a relatively large Go usage example. --- .gitignore | 3 + Cargo.lock | 2 + easytier-contrib/easytier-ffi/Cargo.toml | 6 + .../easytier-ffi/examples/go/README.md | 87 ++ .../easytier-ffi/examples/go/easytier.go | 549 +++++++++++++ .../easytier-ffi/examples/go/easytier_test.go | 140 ++++ .../easytier-ffi/examples/go/go.mod | 5 + easytier-contrib/easytier-ffi/src/lib.rs | 777 +++++++++++++++++- easytier/Cargo.toml | 1 + easytier/src/gateway/socks5.rs | 60 +- easytier/src/gateway/socks5/dataplane.rs | 535 ++++++++++++ easytier/src/instance/instance.rs | 5 + easytier/src/instance_manager.rs | 55 ++ easytier/src/launcher.rs | 132 +++ 14 files changed, 2336 insertions(+), 21 deletions(-) create mode 100644 easytier-contrib/easytier-ffi/examples/go/README.md create mode 100644 easytier-contrib/easytier-ffi/examples/go/easytier.go create mode 100644 easytier-contrib/easytier-ffi/examples/go/easytier_test.go create mode 100644 easytier-contrib/easytier-ffi/examples/go/go.mod create mode 100644 easytier/src/gateway/socks5/dataplane.rs diff --git a/.gitignore b/.gitignore index 9d45b2a2..57cb46ef 100644 --- a/.gitignore +++ b/.gitignore @@ -43,3 +43,6 @@ easytier-gui/src-tauri/*.sys .direnv .flake-profile + +# contrib +go.sum diff --git a/Cargo.lock b/Cargo.lock index eee0ffb5..cd7b36c6 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -2401,6 +2401,8 @@ dependencies = [ "once_cell", "serde", "serde_json", + "tokio", + "tokio-util", "uuid", ] diff --git a/easytier-contrib/easytier-ffi/Cargo.toml b/easytier-contrib/easytier-ffi/Cargo.toml index 52e66241..83e3c755 100644 --- a/easytier-contrib/easytier-ffi/Cargo.toml +++ b/easytier-contrib/easytier-ffi/Cargo.toml @@ -6,6 +6,10 @@ edition.workspace = true [lib] crate-type = ["cdylib"] +[features] +default = ["ffi-dataplane"] +ffi-dataplane = ["easytier/ffi-dataplane"] + [dependencies] easytier = { path = "../../easytier" } @@ -15,3 +19,5 @@ dashmap = "6.0" serde = { version = "1.0", features = ["derive"] } serde_json = "1" uuid = "1.17.0" +tokio = { version = "1", features = ["rt-multi-thread", "io-util", "time", "sync", "macros"] } +tokio-util = "0.7" diff --git a/easytier-contrib/easytier-ffi/examples/go/README.md b/easytier-contrib/easytier-ffi/examples/go/README.md new file mode 100644 index 00000000..e3ad4607 --- /dev/null +++ b/easytier-contrib/easytier-ffi/examples/go/README.md @@ -0,0 +1,87 @@ +# 1. Go FFI Demo + +This demo wraps EasyTier FFI data-plane TCP as Go `net.Conn` and `net.Listener`. +It can connect to an SSH server through EasyTier and read its banner, or accept a +TCP connection from another EasyTier peer and run a small ping/pong exchange. + +## 1.1. Build the FFI library + +Run from the repository root: + +```sh +cargo build -p easytier-ffi --features ffi-dataplane +``` + +The demo loads the debug library by default: + +```text +target/debug/libeasytier_ffi.so +``` + +To use another library path, export `EASYTIER_FFI_LIB=/path/to/libeasytier_ffi.so`. + +## 1.2. Configure the EasyTier config + +`EASYTIER_FFI_CONFIG` is a string of the EasyTier config in TOML format which is passed to the FFI library. For example: + +```sh +export EASYTIER_FFI_CONFIG='instance_name = "default" +ipv4 = "10.0.0.1" +peers = ["tcp://123.123.123.123:11010"] + +[network_identity] +network_name = "testnet" +network_secret = "mysecret" + +[flags] +no_tun = true # disable tun device to avoid permission issues. +' +``` + +You should configure with your own real values. + +Set the local instance name and a SSH server target to connect through EasyTier: + +```sh +export EASYTIER_FFI_INSTANCE=default +export EASYTIER_FFI_TARGET=10.0.0.2:22 +``` + +To run the TCP listen integration test in the same `go test` process as the SSH +test, use a separate instance name and config: + +```sh +export EASYTIER_FFI_LISTEN_CONFIG='instance_name = "listener" +ipv4 = "10.0.0.3" +peers = ["tcp://123.123.123.123:11010"] + +[network_identity] +network_name = "testnet" +network_secret = "mysecret" + +[flags] +no_tun = true +' +export EASYTIER_FFI_LISTEN_INSTANCE=listener +export EASYTIER_FFI_LISTEN_PORT=12345 +``` + +## 1.3. Run the demo + +`goffi` is built without cgo on Linux, so run the test with `CGO_ENABLED=0`: + +```sh +cd easytier-contrib/easytier-ffi/examples/go +CGO_ENABLED=0 go test -v ./... +``` + +Expected output includes an SSH banner similar to: + +```text +attempt 1: got banner "SSH-2.0-..." +PASS +``` + +For `TestTCPListenIntegration`, connect from another EasyTier peer to the local +EasyTier IPv4 address and `EASYTIER_FFI_LISTEN_PORT`, send `ping`, and expect +`pong` in response. diff --git a/easytier-contrib/easytier-ffi/examples/go/easytier.go b/easytier-contrib/easytier-ffi/examples/go/easytier.go new file mode 100644 index 00000000..15859ad1 --- /dev/null +++ b/easytier-contrib/easytier-ffi/examples/go/easytier.go @@ -0,0 +1,549 @@ +package easytierffi + +import ( + "context" + "errors" + "fmt" + "io" + "net" + "os" + "runtime" + "strconv" + "strings" + "sync/atomic" + "time" + "unsafe" + + "github.com/go-webgpu/goffi/ffi" + "github.com/go-webgpu/goffi/types" +) + +const defaultTimeout = 30 * time.Second + +type Native struct { + lib unsafe.Pointer + + runNetworkInstance symCall + getErrorMsg symCall + freeString symCall + tcpConnect symCall + tcpBind symCall + tcpAccept symCall + tcpRead symCall + tcpWrite symCall + tcpClose symCall + tcpListenerClose symCall +} + +type Conn struct { + native *Native + handle uint64 + local net.Addr + remote net.Addr + closed atomic.Bool + rd atomicDeadline + wd atomicDeadline +} + +type Listener struct { + native *Native + handle uint64 + addr net.Addr + closed atomic.Bool +} + +type symCall struct { + fn unsafe.Pointer + cif types.CallInterface +} + +type atomicDeadline struct{ v atomic.Int64 } + +type timeoutError string + +func Open(path string) (*Native, error) { + lib, err := ffi.LoadLibrary(path) + if err != nil { + return nil, err + } + n := &Native{lib: lib} + if err := n.bind(); err != nil { + ffi.FreeLibrary(lib) + return nil, err + } + return n, nil +} + +func (n *Native) Close() error { + if n.lib == nil { + return nil + } + ffi.FreeLibrary(n.lib) + n.lib = nil + return nil +} + +func (n *Native) RunNetworkInstance(config string) error { + defer pinErrorThread()() + cfg := cString(config) + cfgPtr := unsafe.Pointer(&cfg[0]) + var ret int32 + err := n.runNetworkInstance.call(unsafe.Pointer(&ret), unsafe.Pointer(&cfgPtr)) + runtime.KeepAlive(cfg) + if err != nil { + return err + } + if ret != 0 { + return n.lastError() + } + return nil +} + +func (n *Native) DialContext(ctx context.Context, instance, network, address string) (net.Conn, error) { + if network != "tcp" && network != "tcp4" && network != "tcp6" { + return nil, net.UnknownNetworkError(network) + } + ip, port, err := parseIPPort(address) + if err != nil { + return nil, err + } + timeout := defaultTimeout + if deadline, ok := ctx.Deadline(); ok { + timeout = time.Until(deadline) + } + if timeout <= 0 { + return nil, context.DeadlineExceeded + } + if err := ctx.Err(); err != nil { + return nil, err + } + handle, local, err := n.tcpConnectTo(instance, ip.String(), uint16(port), timeout) + if err != nil { + return nil, err + } + return &Conn{native: n, handle: handle, local: local, remote: &net.TCPAddr{IP: ip, Port: port}}, nil +} + +func (n *Native) ListenContext(ctx context.Context, instance, network, address string) (net.Listener, error) { + if network != "tcp" && network != "tcp4" && network != "tcp6" { + return nil, net.UnknownNetworkError(network) + } + port, err := parseListenPort(address) + if err != nil { + return nil, err + } + timeout := defaultTimeout + if deadline, ok := ctx.Deadline(); ok { + timeout = time.Until(deadline) + } + if timeout <= 0 { + return nil, context.DeadlineExceeded + } + if err := ctx.Err(); err != nil { + return nil, err + } + handle, local, err := n.tcpBindTo(instance, uint16(port), timeout) + if err != nil { + return nil, err + } + return &Listener{native: n, handle: handle, addr: local}, nil +} + +func (c *Conn) Read(b []byte) (int, error) { + if c.closed.Load() { + return 0, net.ErrClosed + } + n, err := c.native.tcpReadFrom(c.handle, b, c.rd.timeout(defaultTimeout)) + if err != nil { + return 0, opError("read", c.remote, err) + } + if n == 0 { + return 0, io.EOF + } + return n, nil +} + +func (c *Conn) Write(b []byte) (int, error) { + if c.closed.Load() { + return 0, net.ErrClosed + } + n, err := c.native.tcpWriteTo(c.handle, b, c.wd.timeout(defaultTimeout)) + if err != nil { + return 0, opError("write", c.remote, err) + } + return n, nil +} + +func (c *Conn) Close() error { + if !c.closed.CompareAndSwap(false, true) { + return net.ErrClosed + } + return c.native.tcpCloseHandle(c.handle) +} + +func (c *Conn) LocalAddr() net.Addr { return c.local } +func (c *Conn) RemoteAddr() net.Addr { return c.remote } +func (c *Conn) SetDeadline(t time.Time) error { c.rd.set(t); c.wd.set(t); return nil } +func (c *Conn) SetReadDeadline(t time.Time) error { c.rd.set(t); return nil } +func (c *Conn) SetWriteDeadline(t time.Time) error { c.wd.set(t); return nil } + +func (l *Listener) Accept() (net.Conn, error) { + if l.closed.Load() { + return nil, net.ErrClosed + } + for { + handle, local, peer, err := l.native.tcpAcceptFrom(l.handle, defaultTimeout) + if err == nil { + return &Conn{native: l.native, handle: handle, local: local, remote: peer}, nil + } + if l.closed.Load() { + return nil, net.ErrClosed + } + var netErr net.Error + if errors.As(err, &netErr) && netErr.Timeout() { + continue + } + return nil, opError("accept", l.addr, err) + } +} + +func (l *Listener) Close() error { + if !l.closed.CompareAndSwap(false, true) { + return net.ErrClosed + } + return l.native.tcpListenerCloseHandle(l.handle) +} + +func (l *Listener) Addr() net.Addr { return l.addr } + +func (n *Native) bind() error { + return errors.Join( + n.bindSym(&n.runNetworkInstance, "run_network_instance", types.SInt32TypeDescriptor, types.PointerTypeDescriptor), + n.bindSym(&n.getErrorMsg, "get_error_msg", types.VoidTypeDescriptor, types.PointerTypeDescriptor), + n.bindSym(&n.freeString, "free_string", types.VoidTypeDescriptor, types.PointerTypeDescriptor), + n.bindSym(&n.tcpConnect, "data_plane_tcp_connect", types.UInt64TypeDescriptor, types.PointerTypeDescriptor, types.PointerTypeDescriptor, types.UInt16TypeDescriptor, types.UInt64TypeDescriptor, types.PointerTypeDescriptor, types.PointerTypeDescriptor), + n.bindSym(&n.tcpBind, "data_plane_tcp_bind", types.UInt64TypeDescriptor, types.PointerTypeDescriptor, types.UInt16TypeDescriptor, types.UInt64TypeDescriptor, types.PointerTypeDescriptor, types.PointerTypeDescriptor), + n.bindSym(&n.tcpAccept, "data_plane_tcp_accept", types.UInt64TypeDescriptor, types.UInt64TypeDescriptor, types.UInt64TypeDescriptor, types.PointerTypeDescriptor, types.PointerTypeDescriptor, types.PointerTypeDescriptor, types.PointerTypeDescriptor), + n.bindSym(&n.tcpRead, "data_plane_tcp_read", types.SInt32TypeDescriptor, types.UInt64TypeDescriptor, types.PointerTypeDescriptor, types.UInt32TypeDescriptor, types.UInt64TypeDescriptor), + n.bindSym(&n.tcpWrite, "data_plane_tcp_write", types.SInt32TypeDescriptor, types.UInt64TypeDescriptor, types.PointerTypeDescriptor, types.UInt32TypeDescriptor, types.UInt64TypeDescriptor), + n.bindSym(&n.tcpClose, "data_plane_tcp_close", types.SInt32TypeDescriptor, types.UInt64TypeDescriptor), + n.bindSym(&n.tcpListenerClose, "data_plane_tcp_listener_close", types.SInt32TypeDescriptor, types.UInt64TypeDescriptor), + ) +} + +func (n *Native) bindSym(dst *symCall, name string, ret *types.TypeDescriptor, args ...*types.TypeDescriptor) error { + sym, err := ffi.GetSymbol(n.lib, name) + if err != nil { + return err + } + if err := ffi.PrepareCallInterface(&dst.cif, types.DefaultCall, ret, args); err != nil { + return err + } + dst.fn = sym + return nil +} + +func (s *symCall) call(ret unsafe.Pointer, args ...unsafe.Pointer) error { + // `ffi.CallFunction` and libffi `ffi_call` are safe to invoke concurrently + // because `cif` is prepared once during binding and only read afterwards. + return ffi.CallFunction(&s.cif, s.fn, ret, args) +} + +func (n *Native) tcpConnectTo(instance, ip string, port uint16, timeout time.Duration) (uint64, *net.TCPAddr, error) { + defer pinErrorThread()() + inst := cString(instance) + dst := cString(ip) + instPtr := unsafe.Pointer(&inst[0]) + dstPtr := unsafe.Pointer(&dst[0]) + timeoutMS := uint64(timeout / time.Millisecond) + var handle uint64 + var outIP unsafe.Pointer + outIPArg := unsafe.Pointer(&outIP) + var outPort uint16 + outPortArg := unsafe.Pointer(&outPort) + err := n.tcpConnect.call( + unsafe.Pointer(&handle), + unsafe.Pointer(&instPtr), + unsafe.Pointer(&dstPtr), + unsafe.Pointer(&port), + unsafe.Pointer(&timeoutMS), + unsafe.Pointer(&outIPArg), + unsafe.Pointer(&outPortArg), + ) + runtime.KeepAlive(inst) + runtime.KeepAlive(dst) + if err != nil { + return 0, nil, err + } + if handle == 0 { + return 0, nil, n.lastError() + } + return handle, n.takeTCPAddr(outIP, outPort), nil +} + +func (n *Native) tcpBindTo(instance string, port uint16, timeout time.Duration) (uint64, *net.TCPAddr, error) { + defer pinErrorThread()() + inst := cString(instance) + instPtr := unsafe.Pointer(&inst[0]) + timeoutMS := uint64(timeout / time.Millisecond) + var handle uint64 + var outIP unsafe.Pointer + outIPArg := unsafe.Pointer(&outIP) + var outPort uint16 + outPortArg := unsafe.Pointer(&outPort) + err := n.tcpBind.call( + unsafe.Pointer(&handle), + unsafe.Pointer(&instPtr), + unsafe.Pointer(&port), + unsafe.Pointer(&timeoutMS), + unsafe.Pointer(&outIPArg), + unsafe.Pointer(&outPortArg), + ) + runtime.KeepAlive(inst) + if err != nil { + return 0, nil, err + } + if handle == 0 { + return 0, nil, n.lastError() + } + return handle, n.takeTCPAddr(outIP, outPort), nil +} + +func (n *Native) tcpAcceptFrom(handle uint64, timeout time.Duration) (uint64, *net.TCPAddr, *net.TCPAddr, error) { + defer pinErrorThread()() + timeoutMS := uint64(timeout / time.Millisecond) + var stream uint64 + var outLocalIP unsafe.Pointer + outLocalIPArg := unsafe.Pointer(&outLocalIP) + var outLocalPort uint16 + outLocalPortArg := unsafe.Pointer(&outLocalPort) + var outPeerIP unsafe.Pointer + outPeerIPArg := unsafe.Pointer(&outPeerIP) + var outPeerPort uint16 + outPeerPortArg := unsafe.Pointer(&outPeerPort) + err := n.tcpAccept.call( + unsafe.Pointer(&stream), + unsafe.Pointer(&handle), + unsafe.Pointer(&timeoutMS), + unsafe.Pointer(&outLocalIPArg), + unsafe.Pointer(&outLocalPortArg), + unsafe.Pointer(&outPeerIPArg), + unsafe.Pointer(&outPeerPortArg), + ) + if err != nil { + return 0, nil, nil, err + } + if stream == 0 { + return 0, nil, nil, n.lastError() + } + return stream, n.takeTCPAddr(outLocalIP, outLocalPort), n.takeTCPAddr(outPeerIP, outPeerPort), nil +} + +func (n *Native) tcpReadFrom(handle uint64, buf []byte, timeout time.Duration) (int, error) { + if len(buf) == 0 { + return 0, nil + } + defer pinErrorThread()() + var ret int32 + bufPtr := unsafe.Pointer(&buf[0]) + length := uint32(len(buf)) + timeoutMS := uint64(timeout / time.Millisecond) + err := n.tcpRead.call(unsafe.Pointer(&ret), unsafe.Pointer(&handle), unsafe.Pointer(&bufPtr), unsafe.Pointer(&length), unsafe.Pointer(&timeoutMS)) + runtime.KeepAlive(buf) + if err != nil { + return 0, err + } + if ret < 0 { + return 0, n.lastError() + } + return int(ret), nil +} + +func (n *Native) tcpWriteTo(handle uint64, buf []byte, timeout time.Duration) (int, error) { + if len(buf) == 0 { + return 0, nil + } + defer pinErrorThread()() + var ret int32 + bufPtr := unsafe.Pointer(&buf[0]) + length := uint32(len(buf)) + timeoutMS := uint64(timeout / time.Millisecond) + err := n.tcpWrite.call(unsafe.Pointer(&ret), unsafe.Pointer(&handle), unsafe.Pointer(&bufPtr), unsafe.Pointer(&length), unsafe.Pointer(&timeoutMS)) + runtime.KeepAlive(buf) + if err != nil { + return 0, err + } + if ret < 0 { + return 0, n.lastError() + } + return int(ret), nil +} + +func (n *Native) tcpCloseHandle(handle uint64) error { + defer pinErrorThread()() + var ret int32 + if err := n.tcpClose.call(unsafe.Pointer(&ret), unsafe.Pointer(&handle)); err != nil { + return err + } + if ret != 0 { + return n.lastError() + } + return nil +} + +func (n *Native) tcpListenerCloseHandle(handle uint64) error { + defer pinErrorThread()() + var ret int32 + if err := n.tcpListenerClose.call(unsafe.Pointer(&ret), unsafe.Pointer(&handle)); err != nil { + return err + } + if ret != 0 { + return n.lastError() + } + return nil +} + +// pinErrorThread ties an FFI op to the get_error_msg that reads its result: the +// Rust side stores the last error in a thread-local, so the goroutine must not +// migrate to another OS thread between the two calls. Use as `defer pinErrorThread()()` +// at the start of any wrapper that reports failures through lastError. +func pinErrorThread() func() { + runtime.LockOSThread() + return runtime.UnlockOSThread +} + +func (n *Native) lastError() error { + var out unsafe.Pointer + outArg := unsafe.Pointer(&out) + if err := n.getErrorMsg.call(nil, unsafe.Pointer(&outArg)); err != nil { + return err + } + if out == nil { + return errors.New("easytier ffi call failed") + } + msg := readCString(out) + _ = n.freeCString(out) + if strings.Contains(msg, "timed out") { + return timeoutError(msg) + } + return errors.New(msg) +} + +func (n *Native) freeCString(ptr unsafe.Pointer) error { + if ptr == nil { + return nil + } + return n.freeString.call(nil, unsafe.Pointer(&ptr)) +} + +func (n *Native) takeTCPAddr(ipPtr unsafe.Pointer, port uint16) *net.TCPAddr { + if ipPtr == nil { + return nil + } + ip := net.ParseIP(readCString(ipPtr)) + _ = n.freeCString(ipPtr) + return &net.TCPAddr{IP: ip, Port: int(port)} +} + +func (d *atomicDeadline) set(t time.Time) { + if t.IsZero() { + d.v.Store(0) + return + } + d.v.Store(t.UnixNano()) +} + +func (d *atomicDeadline) timeout(fallback time.Duration) time.Duration { + ns := d.v.Load() + if ns == 0 { + return fallback + } + remaining := time.Until(time.Unix(0, ns)) + if remaining <= 0 { + return time.Millisecond + } + return remaining +} + +func (e timeoutError) Error() string { return string(e) } +func (e timeoutError) Timeout() bool { return true } +func (e timeoutError) Temporary() bool { return true } + +func opError(op string, addr net.Addr, err error) error { + return &net.OpError{Op: op, Net: "easytier", Addr: addr, Err: err} +} + +func parseIPPort(address string) (net.IP, int, error) { + host, portStr, err := net.SplitHostPort(address) + if err != nil { + return nil, 0, err + } + ip := net.ParseIP(host) + if ip == nil { + return nil, 0, fmt.Errorf("easytier ffi requires an IP address, got %q", host) + } + port, err := strconv.ParseUint(portStr, 10, 16) + if err != nil { + return nil, 0, err + } + return ip, int(port), nil +} + +func parseListenPort(address string) (int, error) { + host, portStr, err := net.SplitHostPort(address) + if err != nil { + return 0, err + } + if host != "" { + ip := net.ParseIP(host) + if ip == nil { + return 0, fmt.Errorf("easytier ffi requires an IP address, got %q", host) + } + if !ip.IsUnspecified() { + return 0, fmt.Errorf("easytier ffi listen address must be unspecified, got %q", host) + } + } + port, err := strconv.ParseUint(portStr, 10, 16) + if err != nil { + return 0, err + } + return int(port), nil +} + +func cString(s string) []byte { + if strings.ContainsRune(s, 0) { + panic("easytier ffi string contains NUL") + } + return append([]byte(s), 0) +} + +func readCString(ptr unsafe.Pointer) string { + if ptr == nil { + return "" + } + var b []byte + for p := uintptr(ptr); ; p++ { + c := *(*byte)(unsafe.Pointer(p)) + if c == 0 { + return string(b) + } + b = append(b, c) + } +} + +func defaultLibraryPath() string { + if p := os.Getenv("EASYTIER_FFI_LIB"); p != "" { + return p + } + switch runtime.GOOS { + case "darwin": + return "../../../../target/debug/libeasytier_ffi.dylib" + case "windows": + return "..\\..\\..\\..\\target\\debug\\easytier_ffi.dll" + default: + return "../../../../target/debug/libeasytier_ffi.so" + } +} + +var _ net.Conn = (*Conn)(nil) +var _ net.Listener = (*Listener)(nil) diff --git a/easytier-contrib/easytier-ffi/examples/go/easytier_test.go b/easytier-contrib/easytier-ffi/examples/go/easytier_test.go new file mode 100644 index 00000000..56943864 --- /dev/null +++ b/easytier-contrib/easytier-ffi/examples/go/easytier_test.go @@ -0,0 +1,140 @@ +package easytierffi + +import ( + "context" + "fmt" + "io" + "net" + "os" + "strconv" + "strings" + "testing" + "time" +) + +func TestSSHIntegration(t *testing.T) { + config := os.Getenv("EASYTIER_FFI_CONFIG") + instance := os.Getenv("EASYTIER_FFI_INSTANCE") + target := os.Getenv("EASYTIER_FFI_TARGET") + if config == "" || instance == "" || target == "" { + t.Skip("set EASYTIER_FFI_CONFIG, EASYTIER_FFI_INSTANCE and EASYTIER_FFI_TARGET to run integration test") + } + + n, err := Open(defaultLibraryPath()) + if err != nil { + t.Fatal(err) + } + defer n.Close() + + if err := n.RunNetworkInstance(config); err != nil { + t.Fatal(err) + } + + ctx, cancel := context.WithTimeout(context.Background(), 120*time.Second) + defer cancel() + + var lastErr error + for attempt := 1; ctx.Err() == nil; attempt++ { + conn, err := n.DialContext(ctx, instance, "tcp", target) + if err != nil { + lastErr = err + t.Logf("attempt %d: dial failed: %v", attempt, err) + time.Sleep(3 * time.Second) + continue + } + + _ = conn.SetReadDeadline(time.Now().Add(10 * time.Second)) + buf := make([]byte, 128) + nn, err := conn.Read(buf) + _ = conn.Close() + if err != nil { + lastErr = err + t.Logf("attempt %d: read failed: %v", attempt, err) + time.Sleep(3 * time.Second) + continue + } + banner := string(buf[:nn]) + if !strings.HasPrefix(banner, "SSH-") { + t.Fatalf("attempt %d: expected SSH banner, got %q", attempt, banner) + } + t.Logf("attempt %d: got banner %q", attempt, strings.TrimRight(banner, "\r\n")) + return + } + t.Fatalf("never got SSH banner, last err: %v", lastErr) +} + +func TestTCPListenIntegration(t *testing.T) { + config := os.Getenv("EASYTIER_FFI_LISTEN_CONFIG") + instance := os.Getenv("EASYTIER_FFI_LISTEN_INSTANCE") + listenPort := os.Getenv("EASYTIER_FFI_LISTEN_PORT") + if config == "" || instance == "" || listenPort == "" { + t.Skip("set EASYTIER_FFI_LISTEN_CONFIG, EASYTIER_FFI_LISTEN_INSTANCE and EASYTIER_FFI_LISTEN_PORT to run integration test") + } + port, err := strconv.ParseUint(listenPort, 10, 16) + if err != nil { + t.Fatal(err) + } + + n, err := Open(defaultLibraryPath()) + if err != nil { + t.Fatal(err) + } + defer n.Close() + + if err := n.RunNetworkInstance(config); err != nil { + t.Fatal(err) + } + + ctx, cancel := context.WithTimeout(context.Background(), 120*time.Second) + defer cancel() + + // Data-plane readiness is asynchronous: the instance must finish starting + // before the data plane accepts binds. Retry until ready or ctx expires. + var listener net.Listener + for attempt := 1; ; attempt++ { + listener, err = n.ListenContext(ctx, instance, "tcp", net.JoinHostPort("0.0.0.0", strconv.Itoa(int(port)))) + if err == nil { + break + } + if ctx.Err() != nil { + t.Fatalf("bind never succeeded, last err: %v", err) + } + t.Logf("attempt %d: bind failed: %v", attempt, err) + time.Sleep(3 * time.Second) + } + t.Logf("listening on %s; connect from another EasyTier peer and send ping", listener.Addr()) + + accepted := make(chan error, 1) + go func() { + conn, err := listener.Accept() + if err != nil { + accepted <- err + return + } + defer conn.Close() + + _ = conn.SetDeadline(time.Now().Add(10 * time.Second)) + buf := make([]byte, 4) + if _, err := io.ReadFull(conn, buf); err != nil { + accepted <- err + return + } + if string(buf) != "ping" { + accepted <- fmt.Errorf("expected %q, got %q", "ping", string(buf)) + return + } + _, err = conn.Write([]byte("pong")) + accepted <- err + }() + + select { + case err := <-accepted: + _ = listener.Close() + if err != nil { + t.Fatal(err) + } + case <-ctx.Done(): + _ = listener.Close() + t.Fatal(ctx.Err()) + } +} diff --git a/easytier-contrib/easytier-ffi/examples/go/go.mod b/easytier-contrib/easytier-ffi/examples/go/go.mod new file mode 100644 index 00000000..cecc1de2 --- /dev/null +++ b/easytier-contrib/easytier-ffi/examples/go/go.mod @@ -0,0 +1,5 @@ +module easytierffi-example + +go 1.25 + +require github.com/go-webgpu/goffi v0.4.1 diff --git a/easytier-contrib/easytier-ffi/src/lib.rs b/easytier-contrib/easytier-ffi/src/lib.rs index e9b65be4..a912c591 100644 --- a/easytier-contrib/easytier-ffi/src/lib.rs +++ b/easytier-contrib/easytier-ffi/src/lib.rs @@ -1,18 +1,68 @@ -use std::sync::Mutex; +use std::cell::RefCell; + +#[cfg(feature = "ffi-dataplane")] +use std::{ + future::Future, + net::{IpAddr, SocketAddr}, + sync::{ + Arc, + atomic::{AtomicU64, Ordering}, + }, + time::Duration, +}; use dashmap::DashMap; +#[cfg(feature = "ffi-dataplane")] +use easytier::launcher::{DataPlaneTcpListener, DataPlaneTcpStream, DataPlaneUdpSocket}; use easytier::{ common::config::{ConfigFileControl, ConfigLoader as _, TomlConfigLoader}, instance_manager::NetworkInstanceManager, }; +#[cfg(feature = "ffi-dataplane")] +use tokio::io::{AsyncReadExt, AsyncWriteExt, ReadHalf, WriteHalf}; +#[cfg(feature = "ffi-dataplane")] +use tokio_util::sync::CancellationToken; static INSTANCE_NAME_ID_MAP: once_cell::sync::Lazy> = once_cell::sync::Lazy::new(DashMap::new); static INSTANCE_MANAGER: once_cell::sync::Lazy = once_cell::sync::Lazy::new(NetworkInstanceManager::new); -static ERROR_MSG: once_cell::sync::Lazy>> = - once_cell::sync::Lazy::new(|| Mutex::new(Vec::new())); +thread_local! { + // # Thread Safety + // set_error_msg and get_error_msg must be called on the same thread to + // get correct error. And since `Handle::block_on` polls the top-level + // future on the calling thread, set_error_msg always runs on the same + // thread as the corresponding get_error_msg. + static ERROR_MSG: RefCell> = const { RefCell::new(Vec::new()) }; +} +#[cfg(feature = "ffi-dataplane")] +static NEXT_DATA_PLANE_HANDLE: AtomicU64 = AtomicU64::new(1); +#[cfg(feature = "ffi-dataplane")] +static DATA_PLANE_HANDLES: once_cell::sync::Lazy> = + once_cell::sync::Lazy::new(DashMap::new); + +#[cfg(feature = "ffi-dataplane")] +struct DataPlaneHandle { + instance_id: uuid::Uuid, + runtime: tokio::runtime::Handle, + // Cancelled by close() to wake any in-flight op on this handle. + close_token: CancellationToken, + resource: DataPlaneResource, +} + +#[cfg(feature = "ffi-dataplane")] +struct TcpHalves { + read: tokio::sync::Mutex>, + write: tokio::sync::Mutex>, +} + +#[cfg(feature = "ffi-dataplane")] +enum DataPlaneResource { + Tcp(Arc), + TcpListener(Arc>), + Udp(Arc), +} #[repr(C)] pub struct KeyValuePair { @@ -21,11 +71,199 @@ pub struct KeyValuePair { } fn set_error_msg(msg: &str) { - let bytes = msg.as_bytes(); - let mut msg_buf = ERROR_MSG.lock().unwrap(); - let len = bytes.len(); - msg_buf.resize(len, 0); - msg_buf[..len].copy_from_slice(bytes); + ERROR_MSG.with(|cell| { + let mut buf = cell.borrow_mut(); + buf.clear(); + buf.extend_from_slice(msg.as_bytes()); + }); +} + +// Several helper functions for FFI data plane operations to facilitate logic reuse. + +#[cfg(feature = "ffi-dataplane")] +fn next_handle() -> u64 { + NEXT_DATA_PLANE_HANDLE.fetch_add(1, Ordering::Relaxed) +} + +#[cfg(feature = "ffi-dataplane")] +fn timeout_duration(timeout_ms: u64) -> Duration { + Duration::from_millis(timeout_ms) +} + +#[cfg(feature = "ffi-dataplane")] +unsafe fn cstr_to_string(ptr: *const std::ffi::c_char, name: &str) -> Option { + if ptr.is_null() { + set_error_msg(&format!("{} is null", name)); + return None; + } + Some( + unsafe { std::ffi::CStr::from_ptr(ptr) } + .to_string_lossy() + .into_owned(), + ) +} + +#[cfg(feature = "ffi-dataplane")] +fn get_instance_id(inst_name: &str) -> Option { + INSTANCE_NAME_ID_MAP.get(inst_name).map(|id| *id.value()) +} + +#[cfg(feature = "ffi-dataplane")] +fn parse_socket_addr(host: &str, port: u16) -> Option { + let ip = match host.parse::() { + Ok(ip) => ip, + Err(e) => { + set_error_msg(&format!("failed to parse ip address: {}", e)); + return None; + } + }; + Some(SocketAddr::new(ip, port)) +} + +/// Encode an IP address for FFI return. Returns `*mut c_char` to match +/// `CString::into_raw`; caller releases it via `free_string`. +#[cfg(feature = "ffi-dataplane")] +fn into_ffi_ip_cstring(ip: IpAddr) -> Option<*mut std::ffi::c_char> { + match std::ffi::CString::new(ip.to_string()) { + Ok(s) => Some(s.into_raw()), + Err(e) => { + set_error_msg(&format!("failed to encode ip: {}", e)); + None + } + } +} + +#[cfg(feature = "ffi-dataplane")] +fn get_runtime_handle( + inst_id: &uuid::Uuid, + deadline: std::time::Instant, +) -> Option { + let remaining = deadline.saturating_duration_since(std::time::Instant::now()); + let Some(rt) = INSTANCE_MANAGER.data_plane_wait_runtime_handle(inst_id, remaining) else { + set_error_msg("instance runtime is not ready"); + return None; + }; + Some(rt) +} + +#[cfg(feature = "ffi-dataplane")] +fn insert_tcp_stream_handle( + instance_id: uuid::Uuid, + runtime: tokio::runtime::Handle, + stream: DataPlaneTcpStream, +) -> u64 { + let (rd, wr) = tokio::io::split(stream); + let handle = next_handle(); + DATA_PLANE_HANDLES.insert( + handle, + DataPlaneHandle { + instance_id, + runtime, + close_token: CancellationToken::new(), + resource: DataPlaneResource::Tcp(Arc::new(TcpHalves { + read: tokio::sync::Mutex::new(rd), + write: tokio::sync::Mutex::new(wr), + })), + }, + ); + handle +} + +#[cfg(feature = "ffi-dataplane")] +fn get_tcp_stream( + handle: u64, +) -> Option<(Arc, tokio::runtime::Handle, CancellationToken)> { + let Some(h) = DATA_PLANE_HANDLES.get(&handle) else { + set_error_msg("tcp stream handle not found"); + return None; + }; + match &h.resource { + DataPlaneResource::Tcp(halves) => { + Some((halves.clone(), h.runtime.clone(), h.close_token.clone())) + } + DataPlaneResource::TcpListener(_) | DataPlaneResource::Udp(_) => { + set_error_msg("handle is not a tcp stream"); + None + } + } +} + +#[cfg(feature = "ffi-dataplane")] +fn get_tcp_listener( + handle: u64, +) -> Option<( + Arc>, + tokio::runtime::Handle, + CancellationToken, + uuid::Uuid, +)> { + let Some(h) = DATA_PLANE_HANDLES.get(&handle) else { + set_error_msg("tcp listener handle not found"); + return None; + }; + match &h.resource { + DataPlaneResource::TcpListener(listener) => Some(( + listener.clone(), + h.runtime.clone(), + h.close_token.clone(), + h.instance_id, + )), + DataPlaneResource::Tcp(_) | DataPlaneResource::Udp(_) => { + set_error_msg("handle is not a tcp listener"); + None + } + } +} + +#[cfg(feature = "ffi-dataplane")] +fn get_udp_socket( + handle: u64, +) -> Option<( + Arc, + tokio::runtime::Handle, + CancellationToken, +)> { + let Some(h) = DATA_PLANE_HANDLES.get(&handle) else { + set_error_msg("udp socket handle not found"); + return None; + }; + match &h.resource { + DataPlaneResource::Udp(socket) => { + Some((socket.clone(), h.runtime.clone(), h.close_token.clone())) + } + DataPlaneResource::Tcp(_) | DataPlaneResource::TcpListener(_) => { + set_error_msg("handle is not a udp socket"); + None + } + } +} + +/// Run an IO op on the resource's owning runtime, supporting +/// timeout and cancellation. +#[cfg(feature = "ffi-dataplane")] +async fn run_with_cancel( + close_token: &CancellationToken, + timeout_ms: u64, + error_prefix: &str, + op: F, +) -> Option> +where + F: Future>, +{ + tokio::select! { + biased; + _ = close_token.cancelled() => { + set_error_msg(&format!("{}: handle closed", error_prefix)); + None + } + res = tokio::time::timeout(timeout_duration(timeout_ms), op) => match res { + Ok(r) => Some(r), + Err(_) => { + set_error_msg(&format!("{} timed out", error_prefix)); + None + } + } + } } /// # Safety @@ -58,19 +296,23 @@ pub unsafe extern "C" fn set_tun_fd( } /// # Safety -/// Get the last error message +/// Get the last error message produced on the calling thread. The returned +/// pointer (if non-null) must be released via `free_string`. #[unsafe(no_mangle)] pub unsafe extern "C" fn get_error_msg(out: *mut *const std::ffi::c_char) { - let msg_buf = ERROR_MSG.lock().unwrap(); - if msg_buf.is_empty() { - unsafe { - *out = std::ptr::null(); + let cstr = ERROR_MSG.with(|cell| { + let buf = cell.borrow(); + if buf.is_empty() { + None + } else { + std::ffi::CString::new(&buf[..]).ok() } - return; - } - let cstr = std::ffi::CString::new(&msg_buf[..]).unwrap(); + }); unsafe { - *out = cstr.into_raw(); + *out = match cstr { + Some(s) => s.into_raw() as *const std::ffi::c_char, + None => std::ptr::null(), + }; } } @@ -155,6 +397,8 @@ pub unsafe extern "C" fn retain_network_instance( return -1; } INSTANCE_NAME_ID_MAP.clear(); + #[cfg(feature = "ffi-dataplane")] + DATA_PLANE_HANDLES.clear(); return 0; } @@ -176,13 +420,19 @@ pub unsafe extern "C" fn retain_network_instance( .filter_map(|name| INSTANCE_NAME_ID_MAP.get(name).map(|id| *id)) .collect(); - if let Err(e) = INSTANCE_MANAGER.retain_network_instance(inst_ids) { + if let Err(e) = INSTANCE_MANAGER.retain_network_instance(inst_ids.clone()) { set_error_msg(&format!("failed to retain instances: {}", e)); return -1; } INSTANCE_NAME_ID_MAP.retain(|k, _| inst_names.contains(k)); + // FIXME: `DATA_PLANE_HANDLES.retain()` could trigger drop and cleanup of TCP halves, + // but `retain_network_instance()` has shutdown the server. + // Maybe move this line before `retain_network_instance` to allow graceful close of TCP (TCP FIN)? + #[cfg(feature = "ffi-dataplane")] + DATA_PLANE_HANDLES.retain(|_, handle| inst_ids.contains(&handle.instance_id)); + 0 } @@ -237,6 +487,497 @@ pub unsafe extern "C" fn collect_network_infos( index as std::ffi::c_int } +/// # Safety +/// Open a TCP stream through an EasyTier instance data plane. Returns 0 on +/// failure. On success, writes the local socket address chosen for this +/// connection into `out_local_ip` (a heap-allocated C string the caller must +/// release via `free_string`) and `out_local_port`. Both out pointers must be +/// non-null. +#[cfg(feature = "ffi-dataplane")] +#[unsafe(no_mangle)] +pub unsafe extern "C" fn data_plane_tcp_connect( + inst_name: *const std::ffi::c_char, + dst_ip: *const std::ffi::c_char, + dst_port: std::ffi::c_ushort, + timeout_ms: u64, + out_local_ip: *mut *const std::ffi::c_char, + out_local_port: *mut std::ffi::c_ushort, +) -> u64 { + if out_local_ip.is_null() || out_local_port.is_null() { + set_error_msg("output pointer is null"); + return 0; + } + let Some(inst_name) = (unsafe { cstr_to_string(inst_name, "inst_name") }) else { + return 0; + }; + let Some(dst_ip) = (unsafe { cstr_to_string(dst_ip, "dst_ip") }) else { + return 0; + }; + let Some(inst_id) = get_instance_id(&inst_name) else { + set_error_msg("instance not found"); + return 0; + }; + let Some(dst_addr) = parse_socket_addr(&dst_ip, dst_port) else { + return 0; + }; + let deadline = std::time::Instant::now() + timeout_duration(timeout_ms); + let Some(runtime) = get_runtime_handle(&inst_id, deadline) else { + return 0; + }; + + let remaining = deadline.saturating_duration_since(std::time::Instant::now()); + let result = + runtime.block_on(INSTANCE_MANAGER.data_plane_tcp_connect(&inst_id, dst_addr, remaining)); + match result { + Ok(stream) => { + let local_addr = stream.local_addr(); + let Some(local_ip) = into_ffi_ip_cstring(local_addr.ip()) else { + return 0; + }; + let handle = insert_tcp_stream_handle(inst_id, runtime, stream); + unsafe { + *out_local_ip = local_ip as *const std::ffi::c_char; + *out_local_port = local_addr.port(); + } + handle + } + Err(e) => { + set_error_msg(&format!("failed to connect tcp data plane: {}", e)); + 0 + } + } +} + +/// # Safety +/// Bind a TCP listener through an EasyTier instance data plane. Returns 0 on +/// failure. The local address actually bound is written into `out_local_ip` / +/// `out_local_port`; the caller must release `*out_local_ip` via `free_string`. +#[cfg(feature = "ffi-dataplane")] +#[unsafe(no_mangle)] +pub unsafe extern "C" fn data_plane_tcp_bind( + inst_name: *const std::ffi::c_char, + local_port: std::ffi::c_ushort, + timeout_ms: u64, + out_local_ip: *mut *const std::ffi::c_char, + out_local_port: *mut std::ffi::c_ushort, +) -> u64 { + if out_local_ip.is_null() || out_local_port.is_null() { + set_error_msg("output pointer is null"); + return 0; + } + let Some(inst_name) = (unsafe { cstr_to_string(inst_name, "inst_name") }) else { + return 0; + }; + let Some(inst_id) = get_instance_id(&inst_name) else { + set_error_msg("instance not found"); + return 0; + }; + let deadline = std::time::Instant::now() + timeout_duration(timeout_ms); + let Some(runtime) = get_runtime_handle(&inst_id, deadline) else { + return 0; + }; + + let remaining = deadline.saturating_duration_since(std::time::Instant::now()); + let result = + runtime.block_on(INSTANCE_MANAGER.data_plane_tcp_bind(&inst_id, local_port, remaining)); + match result { + Ok(listener) => { + let local_addr = listener.local_addr(); + let Some(local_ip) = into_ffi_ip_cstring(local_addr.ip()) else { + return 0; + }; + let handle = next_handle(); + DATA_PLANE_HANDLES.insert( + handle, + DataPlaneHandle { + instance_id: inst_id, + runtime, + close_token: CancellationToken::new(), + resource: DataPlaneResource::TcpListener(Arc::new(tokio::sync::Mutex::new( + listener, + ))), + }, + ); + unsafe { + *out_local_ip = local_ip as *const std::ffi::c_char; + *out_local_port = local_addr.port(); + } + handle + } + Err(e) => { + set_error_msg(&format!("failed to bind tcp data plane: {}", e)); + 0 + } + } +} + +/// # Safety +/// Accept one connection from a TCP data-plane listener. Returns a TCP stream +/// handle, or 0 on failure. Local and peer addresses are written into out +/// parameters; returned IP strings must be released via `free_string`. +#[cfg(feature = "ffi-dataplane")] +#[unsafe(no_mangle)] +pub unsafe extern "C" fn data_plane_tcp_accept( + handle: u64, + timeout_ms: u64, + out_local_ip: *mut *const std::ffi::c_char, + out_local_port: *mut std::ffi::c_ushort, + out_peer_ip: *mut *const std::ffi::c_char, + out_peer_port: *mut std::ffi::c_ushort, +) -> u64 { + if out_local_ip.is_null() + || out_local_port.is_null() + || out_peer_ip.is_null() + || out_peer_port.is_null() + { + set_error_msg("output pointer is null"); + return 0; + } + let Some((listener, runtime, close_token, instance_id)) = get_tcp_listener(handle) else { + return 0; + }; + + let ret = runtime.block_on(async move { + let mut listener = listener.lock().await; + run_with_cancel( + &close_token, + timeout_ms, + "tcp data plane accept", + listener.accept(), + ) + .await + }); + + match ret { + Some(Ok((stream, peer_addr))) => { + let local_addr = stream.local_addr(); + let Some(local_ip) = into_ffi_ip_cstring(local_addr.ip()) else { + return 0; + }; + let Some(peer_ip) = into_ffi_ip_cstring(peer_addr.ip()) else { + free_string(local_ip); + return 0; + }; + let stream_handle = insert_tcp_stream_handle(instance_id, runtime, stream); + unsafe { + *out_local_ip = local_ip as *const std::ffi::c_char; + *out_local_port = local_addr.port(); + *out_peer_ip = peer_ip as *const std::ffi::c_char; + *out_peer_port = peer_addr.port(); + } + stream_handle + } + Some(Err(e)) => { + set_error_msg(&format!("failed to accept tcp data plane: {}", e)); + 0 + } + None => 0, + } +} + +/// # Safety +/// Read from a TCP data-plane stream. +#[cfg(feature = "ffi-dataplane")] +#[unsafe(no_mangle)] +pub unsafe extern "C" fn data_plane_tcp_read( + handle: u64, + buf: *mut std::ffi::c_uchar, + len: u32, + timeout_ms: u64, +) -> std::ffi::c_int { + if buf.is_null() { + set_error_msg("buf is null"); + return -1; + } + let Some((halves, runtime, close_token)) = get_tcp_stream(handle) else { + return -1; + }; + // Safety: caller-owned buffer outlives this blocking call. + let buf = unsafe { std::slice::from_raw_parts_mut(buf, len as usize) }; + runtime.block_on(async move { + let mut rd = halves.read.lock().await; + match run_with_cancel( + &close_token, + timeout_ms, + "failed to read tcp data plane", + rd.read(buf), + ) + .await + { + Some(Ok(n)) => n as std::ffi::c_int, + Some(Err(e)) => { + set_error_msg(&format!("failed to read tcp data plane: {}", e)); + -1 + } + None => -1, + } + }) +} + +/// # Safety +/// Write to a TCP data-plane stream. +#[cfg(feature = "ffi-dataplane")] +#[unsafe(no_mangle)] +pub unsafe extern "C" fn data_plane_tcp_write( + handle: u64, + buf: *const std::ffi::c_uchar, + len: u32, + timeout_ms: u64, +) -> std::ffi::c_int { + if buf.is_null() { + set_error_msg("buf is null"); + return -1; + } + let Some((halves, runtime, close_token)) = get_tcp_stream(handle) else { + return -1; + }; + let total = len as usize; + // Safety: caller-owned buffer outlives this blocking call. + let buf = unsafe { std::slice::from_raw_parts(buf, total) }; + runtime.block_on(async move { + let mut wr = halves.write.lock().await; + // Use `write_all` to honor `net.Conn::Write` semantics on the Go side + // (must write everything or return an error); single `write()` can + // silently short-write and corrupt streams that the caller assumes are + // fully written. + match run_with_cancel( + &close_token, + timeout_ms, + "failed to write tcp data plane", + wr.write_all(buf), + ) + .await + { + Some(Ok(())) => total as std::ffi::c_int, + Some(Err(e)) => { + set_error_msg(&format!("failed to write tcp data plane: {}", e)); + -1 + } + None => -1, + } + }) +} + +#[cfg(feature = "ffi-dataplane")] +#[unsafe(no_mangle)] +pub extern "C" fn data_plane_tcp_close(handle: u64) -> std::ffi::c_int { + let Some((_, h)) = DATA_PLANE_HANDLES.remove_if(&handle, |_, e| { + matches!(e.resource, DataPlaneResource::Tcp(_)) + }) else { + set_error_msg(if DATA_PLANE_HANDLES.contains_key(&handle) { + "handle is not a tcp stream" + } else { + "tcp stream handle not found" + }); + return -1; + }; + h.close_token.cancel(); + if let DataPlaneResource::Tcp(halves) = h.resource { + // Best-effort half-close; if write half is in use, the in-flight call + // observes the cancel token and releases the lock shortly after. + h.runtime.spawn(async move { + if let Ok(mut wr) = halves.write.try_lock() { + let _ = wr.shutdown().await; + } + }); + } + 0 +} + +#[cfg(feature = "ffi-dataplane")] +#[unsafe(no_mangle)] +pub extern "C" fn data_plane_tcp_listener_close(handle: u64) -> std::ffi::c_int { + let Some((_, h)) = DATA_PLANE_HANDLES.remove_if(&handle, |_, e| { + matches!(e.resource, DataPlaneResource::TcpListener(_)) + }) else { + set_error_msg(if DATA_PLANE_HANDLES.contains_key(&handle) { + "handle is not a tcp listener" + } else { + "tcp listener handle not found" + }); + return -1; + }; + h.close_token.cancel(); + 0 +} + +/// # Safety +/// Bind a UDP socket through an EasyTier instance data plane. Returns 0 on +/// failure. The local address actually bound (which may differ from the +/// requested port when `local_port == 0`) is written into `out_local_ip` / +/// `out_local_port`; the caller must release `*out_local_ip` via `free_string`. +#[cfg(feature = "ffi-dataplane")] +#[unsafe(no_mangle)] +pub unsafe extern "C" fn data_plane_udp_bind( + inst_name: *const std::ffi::c_char, + local_port: std::ffi::c_ushort, + timeout_ms: u64, + out_local_ip: *mut *const std::ffi::c_char, + out_local_port: *mut std::ffi::c_ushort, +) -> u64 { + if out_local_ip.is_null() || out_local_port.is_null() { + set_error_msg("output pointer is null"); + return 0; + } + let Some(inst_name) = (unsafe { cstr_to_string(inst_name, "inst_name") }) else { + return 0; + }; + let Some(inst_id) = get_instance_id(&inst_name) else { + set_error_msg("instance not found"); + return 0; + }; + let deadline = std::time::Instant::now() + timeout_duration(timeout_ms); + let Some(runtime) = get_runtime_handle(&inst_id, deadline) else { + return 0; + }; + + let remaining = deadline.saturating_duration_since(std::time::Instant::now()); + let result = + runtime.block_on(INSTANCE_MANAGER.data_plane_udp_bind(&inst_id, local_port, remaining)); + match result { + Ok(socket) => { + let local_addr = socket.local_addr(); + let Some(local_ip) = into_ffi_ip_cstring(local_addr.ip()) else { + return 0; + }; + let handle = next_handle(); + DATA_PLANE_HANDLES.insert( + handle, + DataPlaneHandle { + instance_id: inst_id, + runtime, + close_token: CancellationToken::new(), + resource: DataPlaneResource::Udp(Arc::new(socket)), + }, + ); + unsafe { + *out_local_ip = local_ip as *const std::ffi::c_char; + *out_local_port = local_addr.port(); + } + handle + } + Err(e) => { + set_error_msg(&format!("failed to bind udp data plane: {}", e)); + 0 + } + } +} + +/// # Safety +/// Send a datagram through a UDP data-plane socket. +#[cfg(feature = "ffi-dataplane")] +#[unsafe(no_mangle)] +pub unsafe extern "C" fn data_plane_udp_send_to( + handle: u64, + dst_ip: *const std::ffi::c_char, + dst_port: std::ffi::c_ushort, + buf: *const std::ffi::c_uchar, + len: u32, + timeout_ms: u64, +) -> std::ffi::c_int { + if buf.is_null() { + set_error_msg("buf is null"); + return -1; + } + let Some(dst_ip) = (unsafe { cstr_to_string(dst_ip, "dst_ip") }) else { + return -1; + }; + let Some(dst_addr) = parse_socket_addr(&dst_ip, dst_port) else { + return -1; + }; + let Some((socket, runtime, close_token)) = get_udp_socket(handle) else { + return -1; + }; + let total = len as usize; + // Safety: caller-owned buffer outlives this blocking call. + let buf = unsafe { std::slice::from_raw_parts(buf, total) }; + runtime.block_on(async move { + match run_with_cancel( + &close_token, + timeout_ms, + "failed to send udp data plane", + socket.send_to(buf, dst_addr), + ) + .await + { + Some(Ok(n)) => n as std::ffi::c_int, + Some(Err(e)) => { + set_error_msg(&format!("failed to send udp data plane: {}", e)); + -1 + } + None => -1, + } + }) +} + +/// # Safety +/// Receive a datagram from a UDP data-plane socket. +#[cfg(feature = "ffi-dataplane")] +#[unsafe(no_mangle)] +pub unsafe extern "C" fn data_plane_udp_recv_from( + handle: u64, + buf: *mut std::ffi::c_uchar, + len: u32, + out_ip: *mut *const std::ffi::c_char, + out_port: *mut std::ffi::c_ushort, + timeout_ms: u64, +) -> std::ffi::c_int { + if buf.is_null() || out_ip.is_null() || out_port.is_null() { + set_error_msg("output pointer is null"); + return -1; + } + let Some((socket, runtime, close_token)) = get_udp_socket(handle) else { + return -1; + }; + let total = len as usize; + // Safety: caller-owned buffer outlives this blocking call. + let buf = unsafe { std::slice::from_raw_parts_mut(buf, total) }; + let ret = runtime.block_on(run_with_cancel( + &close_token, + timeout_ms, + "udp data plane receive", + socket.recv_from(buf), + )); + + match ret { + Some(Ok((n, addr))) => { + // The returned ip pointer must be released by the caller via + // `free_string` (which calls `CString::from_raw`, matching + // `CString::into_raw` here). + let Some(ip_cstr) = into_ffi_ip_cstring(addr.ip()) else { + return -1; + }; + unsafe { + *out_ip = ip_cstr as *const std::ffi::c_char; + *out_port = addr.port() as std::ffi::c_ushort; + } + n as std::ffi::c_int + } + Some(Err(e)) => { + set_error_msg(&format!("failed to receive udp data plane: {}", e)); + -1 + } + None => -1, + } +} + +#[cfg(feature = "ffi-dataplane")] +#[unsafe(no_mangle)] +pub extern "C" fn data_plane_udp_close(handle: u64) -> std::ffi::c_int { + let Some((_, h)) = DATA_PLANE_HANDLES.remove_if(&handle, |_, e| { + matches!(e.resource, DataPlaneResource::Udp(_)) + }) else { + set_error_msg(if DATA_PLANE_HANDLES.contains_key(&handle) { + "handle is not a udp socket" + } else { + "udp socket handle not found" + }); + return -1; + }; + h.close_token.cancel(); + 0 +} + #[cfg(test)] mod tests { use super::*; diff --git a/easytier/Cargo.toml b/easytier/Cargo.toml index a31fe43f..80ed550a 100644 --- a/easytier/Cargo.toml +++ b/easytier/Cargo.toml @@ -391,6 +391,7 @@ websocket = [ ] smoltcp = ["dep:smoltcp"] socks5 = ["smoltcp"] +ffi-dataplane = ["socks5"] jemalloc = ["dep:jemallocator", "dep:jemalloc-sys"] jemalloc-prof = [ "jemalloc", diff --git a/easytier/src/gateway/socks5.rs b/easytier/src/gateway/socks5.rs index 0cd5d07d..fe98fd0b 100644 --- a/easytier/src/gateway/socks5.rs +++ b/easytier/src/gateway/socks5.rs @@ -52,6 +52,12 @@ use crate::{ peers::{PeerPacketFilter, peer_manager::PeerManager}, }; +#[cfg(feature = "ffi-dataplane")] +mod dataplane; + +#[cfg(feature = "ffi-dataplane")] +pub use dataplane::{DataPlaneTcpListener, DataPlaneTcpStream, DataPlaneUdpSocket}; + enum SocksUdpSocket { UdpSocket(Arc), SmolUdpSocket(super::tokio_smoltcp::UdpSocket), @@ -136,11 +142,17 @@ impl AsyncWrite for SocksTcpStream { enum Socks5EntryData { Tcp(TcpListener), // hold a binded socket to hold the tcp port + #[cfg(feature = "ffi-dataplane")] + // a data-plane routing entry that owns no resource. the entry_type in the + // key distinguishes a listen route from an actively outbound route. + DataPlaneRoute, Udp((Arc, UdpClientKey)), // hold the socket to send data to dst } const UDP_ENTRY: u8 = 1; const TCP_ENTRY: u8 = 2; +#[cfg(feature = "ffi-dataplane")] +const TCP_LISTEN_ENTRY: u8 = 3; #[derive(Debug, Eq, PartialEq, Hash, Clone)] struct Socks5Entry { @@ -477,6 +489,11 @@ pub struct Socks5Server { kcp_endpoint: Mutex>>, socks5_enabled: Arc, + #[cfg(feature = "ffi-dataplane")] + data_plane_refs: Arc, + // Tracks whether the smoltcp `net` is ready for data-plane callers. + #[cfg(feature = "ffi-dataplane")] + data_plane_net_ready: tokio::sync::watch::Sender, cancel_tokens: Arc>, port_forward_list_change_notifier: Arc, entry_count: Arc, @@ -508,14 +525,27 @@ impl PeerPacketFilter for Socks5Server { let Some(tcp_packet) = TcpPacket::new(ipv4.payload()) else { return Some(packet); }; - Socks5Entry { + let entry = Socks5Entry { dst: SocketAddr::new(ipv4.get_source().into(), tcp_packet.get_source()), src: SocketAddr::new( ipv4.get_destination().into(), tcp_packet.get_destination(), ), entry_type: TCP_ENTRY, - } + }; + #[cfg(feature = "ffi-dataplane")] + let entry = if self.entries.contains_key(&entry) { + // Case 1: it is an established connection that has an exactly matched inbound. + entry + } else { + // Case 2: it could be a new TCP SYN packet that has not been accepted. + Socks5Entry { + src: entry.src, + dst: SocketAddr::new(IpAddr::V4(Ipv4Addr::UNSPECIFIED), 0), + entry_type: TCP_LISTEN_ENTRY, + } + }; + entry } IpNextHeaderProtocols::Udp => { @@ -591,6 +621,10 @@ impl Socks5Server { kcp_endpoint: Mutex::new(None), socks5_enabled: Arc::new(AtomicBool::new(false)), + #[cfg(feature = "ffi-dataplane")] + data_plane_refs: Arc::new(AtomicUsize::new(0)), + #[cfg(feature = "ffi-dataplane")] + data_plane_net_ready: tokio::sync::watch::channel(false).0, cancel_tokens: Arc::new(DashMap::new()), port_forward_list_change_notifier: Arc::new(Notify::new()), entry_count: Arc::new(AtomicUsize::new(0)), @@ -608,11 +642,25 @@ impl Socks5Server { let cancel_tokens = self.cancel_tokens.clone(); let port_forward_list_change_notifier = self.port_forward_list_change_notifier.clone(); let socks5_enabled = self.socks5_enabled.clone(); + #[cfg(feature = "ffi-dataplane")] + let data_plane_refs = self.data_plane_refs.clone(); + #[cfg(feature = "ffi-dataplane")] + let data_plane_net_ready = self.data_plane_net_ready.clone(); self.tasks.lock().unwrap().spawn(async move { let mut prev_ipv4 = None; loop { - if cancel_tokens.is_empty() && !socks5_enabled.load(Ordering::Relaxed) { + #[cfg(feature = "ffi-dataplane")] + let data_plane_active = data_plane_refs.load(Ordering::Relaxed) > 0; + #[cfg(not(feature = "ffi-dataplane"))] + let data_plane_active = false; + + if cancel_tokens.is_empty() + && !socks5_enabled.load(Ordering::Relaxed) + && !data_plane_active + { let _ = net.lock().await.take(); + #[cfg(feature = "ffi-dataplane")] + let _ = data_plane_net_ready.send_replace(false); port_forward_list_change_notifier.notified().await; continue; } @@ -637,8 +685,14 @@ impl Socks5Server { packet_recv.clone(), entries.clone(), )); + // Wake any data-plane callers waiting in + // `wait_data_plane_net` for the smoltcp net to appear. + #[cfg(feature = "ffi-dataplane")] + let _ = data_plane_net_ready.send_replace(true); } else { let _ = net.lock().await.take(); + #[cfg(feature = "ffi-dataplane")] + let _ = data_plane_net_ready.send_replace(false); } } diff --git a/easytier/src/gateway/socks5/dataplane.rs b/easytier/src/gateway/socks5/dataplane.rs new file mode 100644 index 00000000..5dc4aacd --- /dev/null +++ b/easytier/src/gateway/socks5/dataplane.rs @@ -0,0 +1,535 @@ +//! Data-plane access built on top of the `Socks5Server` smoltcp stack. +//! +//! This module exposes TCP streams and UDP sockets (mainly for FFI callers that +//! send traffic through EasyTier without creating OS-level proxy listeners). +//! +//! Typical usage: +//! +//! ```ignore +//! let instance = Instance::new(cfg); +//! instance.run().await?; +//! let socks5_server = instance.get_socks5_server(); +//! +//! let socket = socks5_server.data_plane_udp_bind(local_port, timeout).await?; +//! socket.send_to(buf, peer_addr).await?; +//! ``` + +use std::{ + net::{IpAddr, Ipv4Addr, SocketAddr}, + pin::Pin, + sync::{ + Arc, + atomic::{AtomicUsize, Ordering}, + }, + task::{Context, Poll}, + time::{Duration, Instant}, +}; + +use anyhow::Context as _; +use dashmap::mapref::entry::Entry; +use tokio::io::{AsyncRead, AsyncWrite}; + +use crate::{common::error::Error, gateway::fast_socks5::server::AsyncTcpConnector}; + +use super::{ + Socks5AutoConnector, Socks5Entry, Socks5EntryData, Socks5EntrySet, Socks5Server, + SocksTcpStream, SocksUdpSocket, TCP_ENTRY, TCP_LISTEN_ENTRY, UDP_ENTRY, UdpClientKey, +}; +use crate::gateway::tokio_smoltcp::{Net, TcpListener}; + +struct DataPlaneRef { + refs: Arc, + notifier: Arc, +} + +/// A route-table entry whose lifetime is tied to this value: constructing it +/// reserves the route and bumps the active-entry count, dropping it removes the +/// route and drops the count back. +struct OwnedRouteEntry { + entries: Socks5EntrySet, + entry_count: Arc, + entry: Socks5Entry, +} + +impl OwnedRouteEntry { + /// Inserts the route, replacing any existing entry for the same key. + fn register( + entries: Socks5EntrySet, + entry_count: Arc, + entry: Socks5Entry, + ) -> Self { + if entries + .insert(entry.clone(), Socks5EntryData::DataPlaneRoute) + .is_none() + { + entry_count.fetch_add(1, Ordering::Relaxed); + } + Self { + entries, + entry_count, + entry, + } + } + + /// Inserts the route only if the key is free, returning `None` on conflict. + fn try_register( + entries: Socks5EntrySet, + entry_count: Arc, + entry: Socks5Entry, + ) -> Option { + match entries.entry(entry.clone()) { + Entry::Occupied(_) => return None, + Entry::Vacant(vacant) => { + vacant.insert(Socks5EntryData::DataPlaneRoute); + entry_count.fetch_add(1, Ordering::Relaxed); + } + } + Some(Self { + entries, + entry_count, + entry, + }) + } +} + +impl Drop for OwnedRouteEntry { + fn drop(&mut self) { + if self.entries.remove(&self.entry).is_some() { + self.entry_count.fetch_sub(1, Ordering::Relaxed); + } + } +} + +/// Tracks how an established data-plane TCP stream keeps its inbound route alive. +/// +/// The two variants capture the intrinsic asymmetry between the connect and +/// accept paths. An outbound stream reserved a source port through the +/// [`Socks5AutoConnector`], which owns the matching route entry and clears it on +/// drop. An accepted stream instead inherits its port and peer from the +/// listener, so it carries merely an [`OwnedRouteEntry`]. +enum DataPlaneTcpStreamRoute { + Outbound(Socks5AutoConnector), + Accepted(OwnedRouteEntry), +} + +/// A TCP stream created by the data plane API. +/// Can be either an actively requested outbound connection or an outbound request accepted from a TCP listener. +pub struct DataPlaneTcpStream { + stream: SocksTcpStream, + local_addr: SocketAddr, + _data_plane_ref: DataPlaneRef, + _route: DataPlaneTcpStreamRoute, +} + +/// A TCP listener created by the data plane API. +/// It accepts inbound connections and produces [`DataPlaneTcpStream`]s. +pub struct DataPlaneTcpListener { + listener: TcpListener, + local_addr: SocketAddr, + entries: Socks5EntrySet, + entry_count: Arc, + _listen_route: OwnedRouteEntry, + _data_plane_ref: DataPlaneRef, +} + +pub struct DataPlaneUdpSocket { + socket: Arc, + entries: Socks5EntrySet, + entry_count: Arc, + local_addr: SocketAddr, + _data_plane_ref: DataPlaneRef, +} + +impl Drop for DataPlaneRef { + fn drop(&mut self) { + if self.refs.fetch_sub(1, Ordering::Relaxed) == 1 { + self.notifier.notify_one(); + } + } +} + +impl DataPlaneTcpStream { + pub fn local_addr(&self) -> SocketAddr { + self.local_addr + } +} + +impl DataPlaneTcpListener { + pub fn local_addr(&self) -> SocketAddr { + self.local_addr + } + + pub async fn accept(&mut self) -> Result<(DataPlaneTcpStream, SocketAddr), std::io::Error> { + let (stream, peer_addr) = self.listener.accept().await?; + let local_addr = stream.local_addr()?; + let route = OwnedRouteEntry::register( + self.entries.clone(), + self.entry_count.clone(), + Socks5Entry { + src: local_addr, + dst: peer_addr, + entry_type: TCP_ENTRY, + }, + ); + let accepted = DataPlaneTcpStream { + stream: SocksTcpStream::SmolTcp(stream), + local_addr, + _data_plane_ref: self._data_plane_ref.clone(), + _route: DataPlaneTcpStreamRoute::Accepted(route), + }; + Ok((accepted, peer_addr)) + } +} + +impl AsyncRead for DataPlaneTcpStream { + fn poll_read( + self: Pin<&mut Self>, + cx: &mut Context<'_>, + buf: &mut tokio::io::ReadBuf<'_>, + ) -> Poll> { + Pin::new(&mut self.get_mut().stream).poll_read(cx, buf) + } +} + +impl AsyncWrite for DataPlaneTcpStream { + fn poll_write( + self: Pin<&mut Self>, + cx: &mut Context<'_>, + buf: &[u8], + ) -> Poll> { + Pin::new(&mut self.get_mut().stream).poll_write(cx, buf) + } + + fn poll_flush(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll> { + Pin::new(&mut self.get_mut().stream).poll_flush(cx) + } + + fn poll_shutdown( + self: Pin<&mut Self>, + cx: &mut Context<'_>, + ) -> Poll> { + Pin::new(&mut self.get_mut().stream).poll_shutdown(cx) + } +} + +impl DataPlaneUdpSocket { + pub fn local_addr(&self) -> SocketAddr { + self.local_addr + } + + pub async fn send_to(&self, buf: &[u8], addr: SocketAddr) -> Result { + let key = Socks5Entry { + src: self.local_addr, + dst: addr, + entry_type: UDP_ENTRY, + }; + if let Entry::Vacant(entry) = self.entries.entry(key) { + entry.insert(Socks5EntryData::Udp(( + self.socket.clone(), + UdpClientKey { + client_addr: self.local_addr, + dst_addr: addr, + }, + ))); + self.entry_count.fetch_add(1, Ordering::Relaxed); + } + self.socket.send_to(buf, addr).await + } + + pub async fn recv_from(&self, buf: &mut [u8]) -> Result<(usize, SocketAddr), std::io::Error> { + self.socket.recv_from(buf).await + } +} + +impl Drop for DataPlaneUdpSocket { + fn drop(&mut self) { + self.entries.retain(|_, data| match data { + Socks5EntryData::Udp((socket, _)) if Arc::ptr_eq(socket, &self.socket) => { + self.entry_count.fetch_sub(1, Ordering::Relaxed); + false + } + _ => true, + }); + } +} + +impl Clone for DataPlaneRef { + fn clone(&self) -> Self { + self.refs.fetch_add(1, Ordering::Relaxed); + Self { + refs: self.refs.clone(), + notifier: self.notifier.clone(), + } + } +} + +impl Socks5Server { + fn acquire_data_plane_ref(&self) -> DataPlaneRef { + self.data_plane_refs.fetch_add(1, Ordering::Relaxed); + self.port_forward_list_change_notifier.notify_one(); + DataPlaneRef { + refs: self.data_plane_refs.clone(), + notifier: self.port_forward_list_change_notifier.clone(), + } + } + + async fn wait_data_plane_net( + &self, + deadline: Instant, + ) -> Result<(cidr::Ipv4Inet, Arc), Error> { + let mut ready = self.data_plane_net_ready.subscribe(); + loop { + if let Some(net) = self + .net + .lock() + .await + .as_ref() + .map(|net| (net.ipv4_addr, net.smoltcp_net.clone())) + { + return Ok(net); + } + + let now = Instant::now(); + if now >= deadline { + return Err(anyhow::anyhow!("data plane net is not ready").into()); + } + let _ = tokio::time::timeout(deadline - now, ready.wait_for(|ready| *ready)).await; + } + } + + pub async fn data_plane_tcp_connect( + &self, + dst_addr: SocketAddr, + timeout: Duration, + ) -> Result { + let data_plane_ref = self.acquire_data_plane_ref(); + let deadline = Instant::now() + timeout; + let (ipv4_addr, smoltcp_net) = self.wait_data_plane_net(deadline).await?; + // FIXME: This is the data-plane source address reserved for route + // matching. `Socks5AutoConnector` may fall back to direct TCP for + // non-virtual destinations, so this is not always the OS socket's + // local address. + let local_port = smoltcp_net.get_port(); + let local_addr = SocketAddr::new(IpAddr::V4(ipv4_addr.address()), local_port); + let connector = Socks5AutoConnector { + #[cfg(feature = "kcp")] + kcp_endpoint: self.kcp_endpoint.lock().await.clone(), + peer_mgr: self.peer_manager.clone(), + entries: self.entries.clone(), + smoltcp_net: Some(smoltcp_net), + src_addr: local_addr, + entry_count: self.entry_count.clone(), + inner_connector: parking_lot::Mutex::new(None), + }; + + let remaining = deadline.saturating_duration_since(Instant::now()); + let inner_timeout_s = remaining.as_secs().saturating_add(1); + let stream = + tokio::time::timeout(remaining, connector.tcp_connect(dst_addr, inner_timeout_s)) + .await + .with_context(|| "data plane tcp connect timeout")? + .map_err(anyhow::Error::from)?; + Ok(DataPlaneTcpStream { + stream, + local_addr, + _data_plane_ref: data_plane_ref, + _route: DataPlaneTcpStreamRoute::Outbound(connector), + }) + } + + pub async fn data_plane_tcp_bind( + &self, + local_port: u16, + timeout: Duration, + ) -> Result { + let data_plane_ref = self.acquire_data_plane_ref(); + let deadline = Instant::now() + timeout; + let (ipv4_addr, smoltcp_net) = self.wait_data_plane_net(deadline).await?; + let bind_addr = SocketAddr::new(IpAddr::V4(ipv4_addr.address()), local_port); + let listener = smoltcp_net.tcp_bind(bind_addr).await?; + let local_addr = listener.local_addr()?; + let listen_route = OwnedRouteEntry::try_register( + self.entries.clone(), + self.entry_count.clone(), + Socks5Entry { + src: local_addr, + dst: SocketAddr::new(IpAddr::V4(Ipv4Addr::UNSPECIFIED), 0), + entry_type: TCP_LISTEN_ENTRY, + }, + ) + .ok_or_else(|| anyhow::anyhow!("data plane tcp listener already exists"))?; + + Ok(DataPlaneTcpListener { + listener, + local_addr, + entries: self.entries.clone(), + entry_count: self.entry_count.clone(), + _listen_route: listen_route, + _data_plane_ref: data_plane_ref, + }) + } + + pub async fn data_plane_udp_bind( + &self, + local_port: u16, + timeout: Duration, + ) -> Result { + let data_plane_ref = self.acquire_data_plane_ref(); + let deadline = Instant::now() + timeout; + let (ipv4_addr, smoltcp_net) = self.wait_data_plane_net(deadline).await?; + let bind_addr = SocketAddr::new(IpAddr::V4(ipv4_addr.address()), local_port); + let smol = smoltcp_net.udp_bind(bind_addr).await?; + let local_addr = smol.local_addr()?; + let socket = Arc::new(SocksUdpSocket::SmolUdpSocket(smol)); + + Ok(DataPlaneUdpSocket { + socket, + entries: self.entries.clone(), + entry_count: self.entry_count.clone(), + local_addr, + _data_plane_ref: data_plane_ref, + }) + } +} + +#[cfg(test)] +mod tests { + use std::time::Duration; + + use tokio::io::{AsyncReadExt, AsyncWriteExt}; + + use super::Socks5Server; + use crate::peers::peer_manager::PeerManager; + use crate::peers::tests::{connect_peer_manager, create_mock_peer_manager}; + use crate::tunnel::common::tests::wait_for_condition; + + /// A peer and its data-plane server. `Socks5Server` only holds a `Weak` + /// reference to the `PeerManager`, so the manager must be kept alive by the + /// test for the server's smoltcp <-> peer routing to work. + struct Endpoint { + _peer: std::sync::Arc, + server: std::sync::Arc, + ip: cidr::Ipv4Inet, + } + + /// Brings up two peers connected by a ring tunnel, each with a virtual IPv4 + /// and a running `Socks5Server`, and waits until the route to `b`'s IPv4 is + /// visible from `a`. `run(None)` leaves the kcp endpoint unset, so the + /// connect path goes through smoltcp, matching the listener side under test. + async fn setup_pair() -> (Endpoint, Endpoint) { + let a = create_mock_peer_manager().await; + let b = create_mock_peer_manager().await; + connect_peer_manager(a.clone(), b.clone()).await; + + let a_ip: cidr::Ipv4Inet = "10.126.126.1/24".parse().unwrap(); + let b_ip: cidr::Ipv4Inet = "10.126.126.2/24".parse().unwrap(); + a.get_global_ctx().set_ipv4(Some(a_ip)); + b.get_global_ctx().set_ipv4(Some(b_ip)); + + let server_a = Socks5Server::new(a.get_global_ctx(), a.clone(), None); + let server_b = Socks5Server::new(b.get_global_ctx(), b.clone(), None); + server_a.run(None).await.unwrap(); + server_b.run(None).await.unwrap(); + + wait_for_condition( + || async { + a.get_route() + .get_peer_id_by_ipv4(&b_ip.address()) + .await + .is_some() + }, + Duration::from_secs(10), + ) + .await; + + ( + Endpoint { + _peer: a, + server: server_a, + ip: a_ip, + }, + Endpoint { + _peer: b, + server: server_b, + ip: b_ip, + }, + ) + } + + #[tokio::test] + async fn data_plane_tcp_pingpong() { + let (ep_a, ep_b) = setup_pair().await; + let (server_a, server_b, b_ip) = (ep_a.server, ep_b.server, ep_b.ip); + let timeout = Duration::from_secs(10); + + let mut listener = server_b.data_plane_tcp_bind(0, timeout).await.unwrap(); + let listen_addr = + std::net::SocketAddr::new(b_ip.address().into(), listener.local_addr().port()); + + let accept = tokio::spawn(async move { + let (mut stream, _peer) = listener.accept().await.unwrap(); + let mut buf = [0u8; 4]; + stream.read_exact(&mut buf).await.unwrap(); + assert_eq!(&buf, b"ping"); + stream.write_all(b"pong").await.unwrap(); + stream.flush().await.unwrap(); + // Hold the listener and stream until the client has read the reply. + tokio::time::sleep(Duration::from_secs(1)).await; + }); + + let mut client = server_a + .data_plane_tcp_connect(listen_addr, timeout) + .await + .unwrap(); + client.write_all(b"ping").await.unwrap(); + client.flush().await.unwrap(); + let mut buf = [0u8; 4]; + client.read_exact(&mut buf).await.unwrap(); + assert_eq!(&buf, b"pong"); + + accept.await.unwrap(); + } + + #[tokio::test] + async fn data_plane_udp_pingpong() { + let (ep_a, ep_b) = setup_pair().await; + let (server_a, a_ip, server_b, b_ip) = (ep_a.server, ep_a.ip, ep_b.server, ep_b.ip); + let timeout = Duration::from_secs(10); + + let sock_a = server_a.data_plane_udp_bind(0, timeout).await.unwrap(); + let sock_b = server_b.data_plane_udp_bind(0, timeout).await.unwrap(); + let addr_a = std::net::SocketAddr::new(a_ip.address().into(), sock_a.local_addr().port()); + let addr_b = std::net::SocketAddr::new(b_ip.address().into(), sock_b.local_addr().port()); + + // UDP data-plane routes are connected-style: a socket only accepts + // inbound datagrams from a peer it has already sent to, because the + // route entry is registered by `send_to`. Prime b's route toward a so + // the upcoming ping is routed instead of dropped at b's packet filter. + // This datagram is dropped at a (a has no route yet) and is not awaited. + sock_b.send_to(b"warmup", addr_a).await.unwrap(); + + sock_a.send_to(b"ping", addr_b).await.unwrap(); + let mut buf = [0u8; 16]; + let (n, from) = tokio::time::timeout(timeout, sock_b.recv_from(&mut buf)) + .await + .expect("recv ping timed out") + .unwrap(); + assert_eq!(&buf[..n], b"ping"); + assert_eq!(from, addr_a); + + sock_b.send_to(b"pong", addr_a).await.unwrap(); + // a may also receive the stray warmup datagram (it arrives once a has + // registered its route by sending the ping above), so skip anything + // that is not the reply. + loop { + let (n, from) = tokio::time::timeout(timeout, sock_a.recv_from(&mut buf)) + .await + .expect("recv pong timed out") + .unwrap(); + if &buf[..n] == b"pong" { + assert_eq!(from, addr_b); + break; + } + } + } +} diff --git a/easytier/src/instance/instance.rs b/easytier/src/instance/instance.rs index 0f009088..d6c2f156 100644 --- a/easytier/src/instance/instance.rs +++ b/easytier/src/instance/instance.rs @@ -1143,6 +1143,11 @@ impl Instance { self.peer_manager.clone() } + #[cfg(feature = "ffi-dataplane")] + pub fn get_socks5_server(&self) -> Arc { + self.socks5_server.clone() + } + pub async fn close_peer_conn( &mut self, peer_id: PeerId, diff --git a/easytier/src/instance_manager.rs b/easytier/src/instance_manager.rs index 229cb86a..e708c89f 100644 --- a/easytier/src/instance_manager.rs +++ b/easytier/src/instance_manager.rs @@ -1,3 +1,5 @@ +#[cfg(feature = "ffi-dataplane")] +use crate::launcher::{DataPlaneTcpListener, DataPlaneTcpStream, DataPlaneUdpSocket}; use dashmap::DashMap; use std::fmt::{Display, Formatter}; use std::{collections::BTreeMap, path::PathBuf, sync::Arc}; @@ -171,6 +173,59 @@ impl NetworkInstanceManager { tokio::runtime::Runtime::new()?.block_on(self.collect_network_infos()) } + #[cfg(feature = "ffi-dataplane")] + pub async fn data_plane_tcp_connect( + &self, + instance_id: &uuid::Uuid, + dst_addr: std::net::SocketAddr, + timeout: std::time::Duration, + ) -> Result { + let instance = self + .instance_map + .get(instance_id) + .ok_or_else(|| anyhow::anyhow!("instance {} not found", instance_id))?; + instance.data_plane_tcp_connect(dst_addr, timeout).await + } + + #[cfg(feature = "ffi-dataplane")] + pub async fn data_plane_tcp_bind( + &self, + instance_id: &uuid::Uuid, + local_port: u16, + timeout: std::time::Duration, + ) -> Result { + let instance = self + .instance_map + .get(instance_id) + .ok_or_else(|| anyhow::anyhow!("instance {} not found", instance_id))?; + instance.data_plane_tcp_bind(local_port, timeout).await + } + + #[cfg(feature = "ffi-dataplane")] + pub async fn data_plane_udp_bind( + &self, + instance_id: &uuid::Uuid, + local_port: u16, + timeout: std::time::Duration, + ) -> Result { + let instance = self + .instance_map + .get(instance_id) + .ok_or_else(|| anyhow::anyhow!("instance {} not found", instance_id))?; + instance.data_plane_udp_bind(local_port, timeout).await + } + + #[cfg(feature = "ffi-dataplane")] + pub fn data_plane_wait_runtime_handle( + &self, + instance_id: &uuid::Uuid, + timeout: std::time::Duration, + ) -> Option { + self.instance_map + .get(instance_id) + .and_then(|inst| inst.wait_runtime_handle(timeout)) + } + pub async fn get_network_info( &self, instance_id: &uuid::Uuid, diff --git a/easytier/src/launcher.rs b/easytier/src/launcher.rs index f4f4649e..a5662475 100644 --- a/easytier/src/launcher.rs +++ b/easytier/src/launcher.rs @@ -2,6 +2,10 @@ use crate::common::config::{ ConfigFileControl, ConfigSource, PortForwardConfig, parse_mapped_listener_urls, process_secure_mode_cfg, }; +#[cfg(feature = "ffi-dataplane")] +use crate::gateway::socks5::Socks5Server; +#[cfg(feature = "ffi-dataplane")] +pub use crate::gateway::socks5::{DataPlaneTcpListener, DataPlaneTcpStream, DataPlaneUdpSocket}; use crate::proto::api::{self, manage}; use crate::proto::rpc_types::controller::BaseController; use crate::rpc_service::InstanceRpcService; @@ -45,6 +49,13 @@ struct EasyTierData { tun_fd: (mpsc::Sender, Mutex>>), event_subscriber: RwLock>, instance_stop_notifier: Arc, + #[cfg(feature = "ffi-dataplane")] + data_plane: tokio::sync::watch::Sender>>, + #[cfg(feature = "ffi-dataplane")] + runtime_handle: ( + parking_lot::Mutex>, + parking_lot::Condvar, + ), } impl Default for EasyTierData { @@ -56,6 +67,10 @@ impl Default for EasyTierData { events: RwLock::new(VecDeque::new()), tun_fd: (sender, Mutex::new(Some(receiver))), instance_stop_notifier: Arc::new(tokio::sync::Notify::new()), + #[cfg(feature = "ffi-dataplane")] + data_plane: tokio::sync::watch::channel(None).0, + #[cfg(feature = "ffi-dataplane")] + runtime_handle: (parking_lot::Mutex::new(None), parking_lot::Condvar::new()), } } } @@ -160,6 +175,10 @@ impl EasyTierLauncher { instance.run().await?; + #[cfg(feature = "ffi-dataplane")] + data.data_plane + .send_replace(Some(instance.get_socks5_server())); + api_service .write() .unwrap() @@ -214,6 +233,13 @@ impl EasyTierLauncher { } .unwrap(); + #[cfg(feature = "ffi-dataplane")] + { + let (lock, cvar) = &data.runtime_handle; + *lock.lock() = Some(rt.handle().clone()); + cvar.notify_all(); + } + let stop_notifier = Arc::new(tokio::sync::Notify::new()); let stop_notifier_clone = stop_notifier.clone(); @@ -263,6 +289,43 @@ impl EasyTierLauncher { } } } + + #[cfg(feature = "ffi-dataplane")] + pub fn get_data_plane(&self) -> Option> { + self.data.data_plane.borrow().clone() + } + + /// Waits up to `deadline` for the data-plane server to be published. + #[cfg(feature = "ffi-dataplane")] + pub async fn wait_data_plane( + &self, + deadline: tokio::time::Instant, + ) -> Option> { + let mut rx = self.data.data_plane.subscribe(); + loop { + if let Some(server) = rx.borrow_and_update().clone() { + return Some(server); + } + if tokio::time::timeout_at(deadline, rx.changed()) + .await + .is_err() + { + return None; + } + } + } + + /// Blocks up to `timeout` for the runtime handle to be published. + #[cfg(feature = "ffi-dataplane")] + pub fn wait_runtime_handle( + &self, + timeout: std::time::Duration, + ) -> Option { + let (lock, cvar) = &self.data.runtime_handle; + let mut guard = lock.lock(); + cvar.wait_while_for(&mut guard, |h| h.is_none(), timeout); + guard.clone() + } } impl Default for EasyTierLauncher { @@ -454,6 +517,75 @@ impl NetworkInstance { .as_ref() .and_then(|launcher| launcher.get_api_service()) } + + /// Waits up to `timeout` for the data-plane server to come up, returning it + /// together with the deadline so the caller can spend the remaining budget + /// on the actual operation. + #[cfg(feature = "ffi-dataplane")] + async fn wait_data_plane( + &self, + timeout: std::time::Duration, + ) -> anyhow::Result<(Arc, tokio::time::Instant)> { + let deadline = tokio::time::Instant::now() + timeout; + let launcher = self + .launcher + .as_ref() + .ok_or_else(|| anyhow::anyhow!("data plane is not ready"))?; + let server = launcher + .wait_data_plane(deadline) + .await + .ok_or_else(|| anyhow::anyhow!("data plane is not ready"))?; + Ok((server, deadline)) + } + + #[cfg(feature = "ffi-dataplane")] + pub async fn data_plane_tcp_connect( + &self, + dst_addr: SocketAddr, + timeout: std::time::Duration, + ) -> anyhow::Result { + let (server, deadline) = self.wait_data_plane(timeout).await?; + server + .data_plane_tcp_connect(dst_addr, deadline - tokio::time::Instant::now()) + .await + .map_err(Into::into) + } + + #[cfg(feature = "ffi-dataplane")] + pub async fn data_plane_tcp_bind( + &self, + local_port: u16, + timeout: std::time::Duration, + ) -> anyhow::Result { + let (server, deadline) = self.wait_data_plane(timeout).await?; + server + .data_plane_tcp_bind(local_port, deadline - tokio::time::Instant::now()) + .await + .map_err(Into::into) + } + + #[cfg(feature = "ffi-dataplane")] + pub async fn data_plane_udp_bind( + &self, + local_port: u16, + timeout: std::time::Duration, + ) -> anyhow::Result { + let (server, deadline) = self.wait_data_plane(timeout).await?; + server + .data_plane_udp_bind(local_port, deadline - tokio::time::Instant::now()) + .await + .map_err(Into::into) + } + + #[cfg(feature = "ffi-dataplane")] + pub fn wait_runtime_handle( + &self, + timeout: std::time::Duration, + ) -> Option { + self.launcher + .as_ref() + .and_then(|launcher| launcher.wait_runtime_handle(timeout)) + } } pub fn add_proxy_network_to_config(