From 7754100f77efce365bece5890429eebfc3ab7ba6 Mon Sep 17 00:00:00 2001 From: encodeous Date: Sun, 1 Nov 2026 14:24:23 -0500 Subject: [PATCH] breaking(UAPI): decouple nylon from WireGuard IPC --- cmd/probe.go | 7 +- cmd/reload.go | 7 +- cmd/render.go | 11 + cmd/status.go | 17 +- cmd/trace.go | 29 +-- core/ipc_client.go | 90 -------- core/ipc_handler.go | 305 +++++++++++++++++---------- core/nylon.go | 4 +- core/nylon_wireguard.go | 8 +- core/observability.go | 10 +- core/observability_test.go | 4 +- core/sys_darwin.go | 14 -- core/sys_linux.go | 15 -- core/sys_physical.go | 6 +- core/sys_virtual.go | 14 +- core/sys_windows.go | 14 -- core/sys_wireguard.go | 77 +------ e2e/ipc_test.go | 50 +++-- e2e/probe_test.go | 2 +- go.mod | 4 +- go.sum | 4 +- integration/ipc_test.go | 233 ++++++++++---------- ipc/client.go | 84 ++++++++ ipc/ipc.go | 98 +++++++++ ipc/ipc_test.go | 39 ++++ ipc/socket_unix.go | 45 ++++ ipc/socket_unix_test.go | 91 ++++++++ ipc/socket_windows.go | 35 ++++ protocol/nylon_ipc.pb.go | 420 +++++++++++++++++++++++++++---------- protocol/nylon_ipc.proto | 46 +++- 30 files changed, 1149 insertions(+), 634 deletions(-) delete mode 100644 core/ipc_client.go create mode 100644 ipc/client.go create mode 100644 ipc/ipc.go create mode 100644 ipc/ipc_test.go create mode 100644 ipc/socket_unix.go create mode 100644 ipc/socket_unix_test.go create mode 100644 ipc/socket_windows.go diff --git a/cmd/probe.go b/cmd/probe.go index 22ba9fc7..8016c5f5 100644 --- a/cmd/probe.go +++ b/cmd/probe.go @@ -5,7 +5,6 @@ import ( "os" "time" - "github.com/encodeous/nylon/core" "github.com/encodeous/nylon/protocol" "github.com/spf13/cobra" ) @@ -23,7 +22,7 @@ var probeCmd = &cobra.Command{ fmt.Fprintln(os.Stderr, "Error: timeout must be positive") os.Exit(1) } - resp, err := core.SendIPCRequest(itf, &protocol.IpcRequest{ + resp, err := ipcRequest(itf, &protocol.IpcRequest{ Request: &protocol.IpcRequest_Probe{Probe: &protocol.ProbeRequest{ PeerId: args[0], TimeoutMs: durationMillis(timeout), @@ -33,10 +32,6 @@ var probeCmd = &cobra.Command{ fmt.Fprintln(os.Stderr, "Error:", err) os.Exit(1) } - if !resp.Ok { - fmt.Fprintln(os.Stderr, "Error:", resp.Error) - os.Exit(1) - } if jsonOut { printJSON(resp) return diff --git a/cmd/reload.go b/cmd/reload.go index bf07e00c..58b055bb 100644 --- a/cmd/reload.go +++ b/cmd/reload.go @@ -4,7 +4,6 @@ import ( "fmt" "os" - "github.com/encodeous/nylon/core" "github.com/encodeous/nylon/protocol" "github.com/spf13/cobra" ) @@ -17,17 +16,13 @@ var reloadCmd = &cobra.Command{ itf, _ := cmd.Flags().GetString("interface") jsonOut, _ := cmd.Flags().GetBool("json") - resp, err := core.SendIPCRequest(itf, &protocol.IpcRequest{ + resp, err := ipcRequest(itf, &protocol.IpcRequest{ Request: &protocol.IpcRequest_Reload{Reload: &protocol.ReloadRequest{}}, }) if err != nil { fmt.Fprintln(os.Stderr, "Error:", err) os.Exit(1) } - if !resp.Ok { - fmt.Fprintln(os.Stderr, "Error:", resp.Error) - os.Exit(1) - } if jsonOut { printJSON(resp) return diff --git a/cmd/render.go b/cmd/render.go index 42b37f14..af569984 100644 --- a/cmd/render.go +++ b/cmd/render.go @@ -5,6 +5,7 @@ import ( "strings" "time" + "github.com/encodeous/nylon/ipc" "github.com/encodeous/nylon/protocol" "github.com/encodeous/nylon/state" "google.golang.org/protobuf/encoding/protojson" @@ -163,6 +164,16 @@ func formatDurationNs(ns int64) string { return time.Duration(ns).String() } +// ipcRequest sends one request to the nylon serving interface itf. +func ipcRequest(itf string, req *protocol.IpcRequest) (*protocol.IpcResponse, error) { + client, err := ipc.Dial(itf) + if err != nil { + return nil, err + } + defer client.Close() + return client.Do(req) +} + func printJSON(resp *protocol.IpcResponse) { m := protojson.MarshalOptions{Indent: " ", EmitUnpopulated: true} data, _ := m.Marshal(resp) diff --git a/cmd/status.go b/cmd/status.go index aa194a48..b22bbabc 100644 --- a/cmd/status.go +++ b/cmd/status.go @@ -5,7 +5,6 @@ import ( "os" "strings" - "github.com/encodeous/nylon/core" "github.com/encodeous/nylon/protocol" "github.com/encodeous/nylon/state" "github.com/moby/term" @@ -23,17 +22,13 @@ var statusCmd = &cobra.Command{ showFull, _ := cmd.Flags().GetBool("full") noColor, _ := cmd.Flags().GetBool("no-color") - resp, err := core.SendIPCRequest(itf, &protocol.IpcRequest{ + resp, err := ipcRequest(itf, &protocol.IpcRequest{ Request: &protocol.IpcRequest_Status{Status: &protocol.StatusRequest{}}, }) if err != nil { fmt.Fprintln(os.Stderr, "Error:", err) os.Exit(1) } - if !resp.Ok { - fmt.Fprintln(os.Stderr, "Error:", resp.Error) - os.Exit(1) - } if jsonOut { printJSON(resp) return @@ -103,12 +98,12 @@ func renderStatus(s *protocol.StatusResponse, opts statusRenderOptions) { if best != nil { bestMetric = best.Metric } - wg := neigh.GetWireguard() + peerStats := neigh.GetTransport() statHeaders := []string{"best metric", "latest handshake", "tx", "rx"} - statRow := []string{metricText(p, bestMetric), formatHandshake(wg.LatestHandshakeUnix), formatBytes(wg.TxBytes), formatBytes(wg.RxBytes)} - if wg.Endpoint != nil { - statHeaders = append(statHeaders, "wireguard endpoint") - statRow = append(statRow, *wg.Endpoint) + statRow := []string{metricText(p, bestMetric), formatHandshake(peerStats.LatestHandshakeUnix), formatBytes(peerStats.TxBytes), formatBytes(peerStats.RxBytes)} + if peerStats.Endpoint != nil { + statHeaders = append(statHeaders, "transport endpoint") + statRow = append(statRow, *peerStats.Endpoint) } printTable(p, 2, statHeaders, [][]string{statRow}) if len(neigh.Endpoints) > 0 { diff --git a/cmd/trace.go b/cmd/trace.go index 64e1b9c4..155d09a6 100644 --- a/cmd/trace.go +++ b/cmd/trace.go @@ -4,7 +4,7 @@ import ( "fmt" "os" - "github.com/encodeous/nylon/core" + "github.com/encodeous/nylon/ipc" "github.com/encodeous/nylon/protocol" "github.com/spf13/cobra" ) @@ -15,23 +15,16 @@ var traceCmd = &cobra.Command{ GroupID: "ny", Run: func(cmd *cobra.Command, args []string) { itf, _ := cmd.Flags().GetString("interface") - first := true - err := core.SendIPCStream(itf, &protocol.IpcRequest{ - Request: &protocol.IpcRequest_Trace{Trace: &protocol.TraceRequest{}}, - }, func(resp *protocol.IpcResponse) error { - if first { - first = false - if !resp.Ok { - fmt.Fprintln(os.Stderr, "Error:", resp.Error) - os.Exit(1) - } - return nil - } - if t := resp.GetTrace(); t != nil { - fmt.Print(t.Line) - } - return nil - }) + client, err := ipc.Dial(itf) + if err == nil { + defer client.Close() + err = client.Stream(&protocol.IpcRequest{ + Request: &protocol.IpcRequest_Trace{Trace: &protocol.TraceRequest{}}, + }, func(resp *protocol.IpcResponse) bool { + fmt.Print(resp.GetTrace().GetLine()) + return true + }) + } if err != nil { fmt.Fprintln(os.Stderr, "Error:", err) os.Exit(1) diff --git a/core/ipc_client.go b/core/ipc_client.go deleted file mode 100644 index b501c2a5..00000000 --- a/core/ipc_client.go +++ /dev/null @@ -1,90 +0,0 @@ -package core - -import ( - "bufio" - "fmt" - - "github.com/encodeous/nylon/protocol" - "github.com/encodeous/polyamide-wireguard-go/ipc" -) - -func SendIPCRequest(itf string, req *protocol.IpcRequest) (*protocol.IpcResponse, error) { - conn, err := ipc.UAPIDial(itf) - if err != nil { - return nil, fmt.Errorf("connect to %s: %w", itf, err) - } - defer conn.Close() - - rw := bufio.NewReadWriter(bufio.NewReader(conn), bufio.NewWriter(conn)) - - if _, err := rw.WriteString("get=nylon\n"); err != nil { - return nil, err - } - - data, err := pjMarshal.Marshal(req) - if err != nil { - return nil, err - } - if _, err := rw.Write(data); err != nil { - return nil, err - } - if _, err := rw.WriteString("\n"); err != nil { - return nil, err - } - if err := rw.Flush(); err != nil { - return nil, err - } - - line, err := rw.ReadBytes('\n') - if err != nil { - return nil, fmt.Errorf("read response: %w", err) - } - - resp := &protocol.IpcResponse{} - if err := pjUnmarshal.Unmarshal(line, resp); err != nil { - return nil, fmt.Errorf("unmarshal response: %w", err) - } - return resp, nil -} - -func SendIPCStream(itf string, req *protocol.IpcRequest, handler func(*protocol.IpcResponse) error) error { - conn, err := ipc.UAPIDial(itf) - if err != nil { - return fmt.Errorf("connect to %s: %w", itf, err) - } - defer conn.Close() - - rw := bufio.NewReadWriter(bufio.NewReader(conn), bufio.NewWriter(conn)) - - if _, err := rw.WriteString("get=nylon\n"); err != nil { - return err - } - - data, err := pjMarshal.Marshal(req) - if err != nil { - return err - } - if _, err := rw.Write(data); err != nil { - return err - } - if _, err := rw.WriteString("\n"); err != nil { - return err - } - if err := rw.Flush(); err != nil { - return err - } - - for { - line, err := rw.ReadBytes('\n') - if err != nil { - return nil // stream ended - } - resp := &protocol.IpcResponse{} - if err := pjUnmarshal.Unmarshal(line, resp); err != nil { - return fmt.Errorf("unmarshal stream response: %w", err) - } - if err := handler(resp); err != nil { - return err - } - } -} diff --git a/core/ipc_handler.go b/core/ipc_handler.go index 2cb8ede5..fa2e5121 100644 --- a/core/ipc_handler.go +++ b/core/ipc_handler.go @@ -6,108 +6,203 @@ import ( "context" "errors" "fmt" + "io" + "net" "os" "slices" + "sync" "time" + "github.com/encodeous/nylon/ipc" "github.com/encodeous/nylon/polyamide" "github.com/encodeous/nylon/protocol" "github.com/encodeous/nylon/state" "github.com/goccy/go-yaml" - "google.golang.org/protobuf/encoding/protojson" ) -var ErrIPCStatusHandled = errors.New("nylon IPC response handled") - -var pjMarshal = protojson.MarshalOptions{EmitUnpopulated: true} -var pjUnmarshal = protojson.UnmarshalOptions{DiscardUnknown: true} - const ( defaultIPCProbeTimeout = 2 * time.Second maxIPCProbeTimeout = 10 * time.Second + // maxIPCInFlight bounds the requests one connection can have running. + maxIPCInFlight = 16 ) -func writeResponse(rw *bufio.ReadWriter, resp *protocol.IpcResponse) error { - data, err := pjMarshal.Marshal(resp) - if err != nil { - return fmt.Errorf("marshal response: %w", err) - } - _, err = rw.Write(data) - if err != nil { - return err - } - _, err = rw.WriteString("\n") - if err != nil { - return err - } - return rw.Flush() +// ipcServer accepts connections on nylon's socket. +type ipcServer struct { + listener net.Listener + workers sync.WaitGroup +} + +// serve accepts connections until the listener closes. +func (s *ipcServer) serve(n *Nylon) { + s.workers.Add(1) + go func() { + defer s.workers.Done() + for n.Context.Err() == nil { + c, err := s.listener.Accept() + if err != nil { + if n.Context.Err() != nil || errors.Is(err, net.ErrClosed) { + return + } + n.Log.Debug("IPC accept failed", "err", err) + continue + } + s.workers.Add(1) + go func() { + defer s.workers.Done() + if err := ServeIPC(n, c); err != nil { + n.Log.Debug("IPC connection ended", "err", err) + } + }() + } + }() } -func errResponse(msg string) *protocol.IpcResponse { - return &protocol.IpcResponse{Ok: false, Error: msg} +func (s *ipcServer) close() error { + err := s.listener.Close() + s.workers.Wait() + return err } -func readRequest(rw *bufio.ReadWriter) (*protocol.IpcRequest, error) { - line, err := rw.ReadBytes('\n') +// ServeIPC answers requests on conn until the client hangs up or nylon stops. It closes conn. +func ServeIPC(n *Nylon, conn net.Conn) error { + defer conn.Close() + stop := context.AfterFunc(n.Context, func() { _ = conn.Close() }) + defer stop() + + r := bufio.NewReader(conn) + version, err := ipc.ReadHello(r) if err != nil { - return nil, err + return fmt.Errorf("read hello: %w", err) + } + // Answer either way, so a client with another version can say which. + if err := ipc.WriteHello(conn); err != nil { + return err } - req := &protocol.IpcRequest{} - if err := pjUnmarshal.Unmarshal(line, req); err != nil { - return nil, fmt.Errorf("unmarshal request: %w", err) + if version != ipc.Version { + return ipc.ErrVersion{Local: ipc.Version, Remote: version} } - return req, nil -} -func HandleNylonIPC(n *Nylon, rw *bufio.ReadWriter) error { - req, err := readRequest(rw) - if err != nil { - if err := writeResponse(rw, errResponse(err.Error())); err != nil { + ctx, cancel := context.WithCancel(n.Context) + s := &ipcSession{n: n, conn: conn, active: make(map[uint32]context.CancelFunc)} + defer func() { + cancel() + _ = conn.Close() // unblocks a request stuck writing to a client that stopped reading + s.requests.Wait() + }() + for { + req := &protocol.IpcRequest{} + if err := ipc.ReadFrame(r, req); err != nil { + if errors.Is(err, io.EOF) || errors.Is(err, net.ErrClosed) { + return nil + } + // The stream can't be trusted after a bad frame, so say why and hang up. + _ = conn.SetWriteDeadline(time.Now().Add(time.Second)) + _ = s.send(&protocol.IpcResponse{Error: ipc.Errorf(protocol.IpcError_INVALID_REQUEST, "bad request: %v", err)}) return err } - return ErrIPCStatusHandled + if c := req.GetCancel(); c != nil { + s.cancel(c.Id) + continue + } + s.start(ctx, req) } +} - // trace is blocking, so we dont dispatch - if _, ok := req.Request.(*protocol.IpcRequest_Trace); ok { - return handleTrace(n, rw) - } - if _, ok := req.Request.(*protocol.IpcRequest_Probe); ok { - resp := handleIPCProbe(n, req.GetProbe()) - if err := writeResponse(rw, resp); err != nil { - return err - } - return ErrIPCStatusHandled +// ipcSession is one client connection. Its requests run concurrently. +type ipcSession struct { + n *Nylon + conn net.Conn + writeMu sync.Mutex + mu sync.Mutex + active map[uint32]context.CancelFunc + requests sync.WaitGroup +} + +func (s *ipcSession) send(resp *protocol.IpcResponse) error { + s.writeMu.Lock() + defer s.writeMu.Unlock() + return ipc.WriteFrame(s.conn, resp) +} + +func (s *ipcSession) start(ctx context.Context, req *protocol.IpcRequest) { + id := req.Id + s.mu.Lock() + _, dup := s.active[id] + busy := len(s.active) >= maxIPCInFlight + var cancel context.CancelFunc + if !dup && !busy { + ctx, cancel = context.WithCancel(ctx) + s.active[id] = cancel + } + s.mu.Unlock() + switch { + case dup: + _ = s.send(&protocol.IpcResponse{Id: id, Error: ipc.Errorf(protocol.IpcError_INVALID_REQUEST, "request %d is already running", id)}) + return + case busy: + _ = s.send(&protocol.IpcResponse{Id: id, Error: ipc.Errorf(protocol.IpcError_UNAVAILABLE, "too many requests in flight")}) + return + } + s.requests.Add(1) + go func() { + defer s.requests.Done() + defer s.cancel(id) + handleIPCRequest(ctx, s.n, req, func(resp *protocol.IpcResponse) error { + resp.Id = id + return s.send(resp) + }) + }() +} + +func (s *ipcSession) cancel(id uint32) { + s.mu.Lock() + defer s.mu.Unlock() + if cancel, ok := s.active[id]; ok { + cancel() + delete(s.active, id) } +} + +func errResponse(code protocol.IpcError_Code, format string, args ...any) *protocol.IpcResponse { + return &protocol.IpcResponse{Error: ipc.Errorf(code, format, args...)} +} + +// handleIPCRequest answers req through send, which a trace calls once per event. +func handleIPCRequest(ctx context.Context, n *Nylon, req *protocol.IpcRequest, send func(*protocol.IpcResponse) error) { + var resp *protocol.IpcResponse + switch req.Request.(type) { + case *protocol.IpcRequest_Status: + resp = dispatchIPC(ctx, n, func() *protocol.IpcResponse { return handleStatus(n, req.GetStatus()) }) + case *protocol.IpcRequest_Reload: + resp = dispatchIPC(ctx, n, func() *protocol.IpcResponse { return handleIPCReload(n, req.GetReload()) }) + case *protocol.IpcRequest_Probe: + resp = handleIPCProbe(ctx, n, req.GetProbe()) + case *protocol.IpcRequest_Trace: + handleTrace(ctx, n, send) + return + default: + resp = errResponse(protocol.IpcError_INVALID_REQUEST, "unknown request") + } + _ = send(resp) +} +// dispatchIPC runs fn on the dispatcher, which owns nylon's state. +func dispatchIPC(ctx context.Context, n *Nylon, fn func() *protocol.IpcResponse) *protocol.IpcResponse { done := make(chan *protocol.IpcResponse, 1) n.Dispatch(func() error { - var resp *protocol.IpcResponse - switch req.Request.(type) { - case *protocol.IpcRequest_Status: - resp = handleStatus(n, req.GetStatus()) - case *protocol.IpcRequest_Reload: - resp = handleIPCReload(n, req.GetReload()) - default: - resp = errResponse("unknown method") - } - done <- resp + done <- fn() return nil }) - - var resp *protocol.IpcResponse select { - case resp = <-done: - case <-n.Context.Done(): - resp = errResponse("nylon shutting down") + case resp := <-done: + return resp + case <-ctx.Done(): + return errResponse(protocol.IpcError_UNAVAILABLE, "nylon shutting down") case <-time.After(n.IPCDispatchTimeout): // nylon is too busy to handle IPC requests - resp = errResponse("timed out waiting for dispatch") + return errResponse(protocol.IpcError_UNAVAILABLE, "timed out waiting for dispatch") } - if err := writeResponse(rw, resp); err != nil { - return err - } - return ErrIPCStatusHandled } func handleStatus(n *Nylon, req *protocol.StatusRequest) *protocol.IpcResponse { @@ -122,22 +217,21 @@ func handleStatus(n *Nylon, req *protocol.StatusRequest) *protocol.IpcResponse { txBytes := uint64(0) rxBytes := uint64(0) - wgStats := wireGuardPeerStats(n) - for _, stat := range wgStats { + peerStats := transportPeerStats(n) + for _, stat := range peerStats { txBytes += stat.TxBytes rxBytes += stat.RxBytes } listenPort := uint32(n.LocalCfg.Port) - // The WireGuard transport knows its bound port once it starts. - if n.wireGuard != nil { - if port := n.wireGuard.transport.ListenPort(); port != 0 { + // A transport that binds a port knows which once it starts. + if t, ok := n.defaultTransport().(interface{ ListenPort() uint16 }); ok { + if port := t.ListenPort(); port != 0 { listenPort = uint32(port) } } return &protocol.IpcResponse{ - Ok: true, Response: &protocol.IpcResponse_Status{Status: &protocol.StatusResponse{ Node: &protocol.NodeStatus{ NodeId: string(n.LocalCfg.Id), @@ -157,7 +251,7 @@ func handleStatus(n *Nylon, req *protocol.StatusRequest) *protocol.IpcResponse { RxBytes: rxBytes, }, }, - Neighbours: buildNeighbours(n, wgStats), + Neighbours: buildNeighbours(n, peerStats), Routes: buildRouteTables(n), FeasibilityDistances: buildFeasibilityDistances(n), }}, @@ -193,7 +287,7 @@ func buildSeqnos(n *Nylon) []*protocol.SeqnoEntry { return entries } -func buildNeighbours(n *Nylon, wgStats map[state.NyPublicKey]polyamide.PeerStats) []*protocol.NeighbourInfo { +func buildNeighbours(n *Nylon, peerStats map[state.NyPublicKey]polyamide.PeerStats) []*protocol.NeighbourInfo { ids := slices.Clone(n.GetPeers(n.LocalCfg.Id)) slices.Sort(ids) neighbours := make([]*protocol.NeighbourInfo, 0, len(ids)) @@ -206,7 +300,7 @@ func buildNeighbours(n *Nylon, wgStats map[state.NyPublicKey]polyamide.PeerStats eps = n.buildEndpoints(neigh) routes = buildNeighRoutes(neigh) } - stat := wgStats[cfg.PubKey] + stat := peerStats[cfg.PubKey] neighbours = append(neighbours, &protocol.NeighbourInfo{ PeerId: string(id), PublicKey: keyString(cfg.PubKey), @@ -214,7 +308,7 @@ func buildNeighbours(n *Nylon, wgStats map[state.NyPublicKey]polyamide.PeerStats Endpoints: eps, Routes: routes, Advertised: advertisementsForNode(n, id), - Wireguard: wireGuardPeerStatsProto(stat), + Transport: transportPeerStatsProto(stat), }) } return neighbours @@ -321,7 +415,7 @@ func advertisementsForNode(n *Nylon, id state.NodeId) []*protocol.Advertisement return ads } -func wireGuardPeerStats(n *Nylon) map[state.NyPublicKey]polyamide.PeerStats { +func transportPeerStats(n *Nylon) map[state.NyPublicKey]polyamide.PeerStats { stats := make(map[state.NyPublicKey]polyamide.PeerStats) if len(n.Transports) == 0 { return stats @@ -333,7 +427,7 @@ func wireGuardPeerStats(n *Nylon) map[state.NyPublicKey]polyamide.PeerStats { } return stats } -func wireGuardPeerStatsProto(stat polyamide.PeerStats) *protocol.WireGuardPeerStats { +func transportPeerStatsProto(stat polyamide.PeerStats) *protocol.TransportPeerStats { var handshake int64 if !stat.LastHandshake.IsZero() { handshake = stat.LastHandshake.UnixNano() @@ -342,7 +436,7 @@ func wireGuardPeerStatsProto(stat polyamide.PeerStats) *protocol.WireGuardPeerSt if stat.PreferredEndpoint != nil { endpoint = stat.PreferredEndpoint.Address() } - return &protocol.WireGuardPeerStats{LatestHandshakeUnix: handshake, TxBytes: stat.TxBytes, RxBytes: stat.RxBytes, PersistentKeepaliveInterval: uint32(stat.Keepalive / time.Second), Endpoint: &endpoint} + return &protocol.TransportPeerStats{LatestHandshakeUnix: handshake, TxBytes: stat.TxBytes, RxBytes: stat.RxBytes, PersistentKeepaliveInterval: uint32(stat.Keepalive / time.Second), Endpoint: &endpoint} } func sourceProto(source state.Source) *protocol.Source { @@ -426,31 +520,31 @@ func sortRouteTableEntries(entries []*protocol.RouteTableEntry) { }) } -func handleIPCProbe(n *Nylon, req *protocol.ProbeRequest) *protocol.IpcResponse { +func handleIPCProbe(ctx context.Context, n *Nylon, req *protocol.ProbeRequest) *protocol.IpcResponse { probeTimeout := ipcProbeTimeout(req) - dispatchCtx, dispatchCancel := context.WithTimeout(n.Context, n.IPCDispatchTimeout) + dispatchCtx, dispatchCancel := context.WithTimeout(ctx, n.IPCDispatchTimeout) defer dispatchCancel() probes, err := NewDispatchFuture(n, func() ([]Future[*protocol.EndpointProbeResult], error) { return n.sendEndpointProbes(state.NodeId(req.PeerId), probeTimeout) }).Await(dispatchCtx) if err != nil { - if errors.Is(err, context.DeadlineExceeded) { - return errResponse("timed out waiting for dispatch") + if ctxErr := dispatchCtx.Err(); ctxErr != nil && errors.Is(err, ctxErr) { + return errResponse(protocol.IpcError_UNAVAILABLE, "timed out waiting for dispatch") } - return errResponse(err.Error()) + // the only other failure: the peer isn't a neighbour + return errResponse(protocol.IpcError_NOT_FOUND, "%v", err) } - probeCtx, probeCancel := context.WithTimeout(n.Context, probeTimeout+n.IPCDispatchTimeout) + probeCtx, probeCancel := context.WithTimeout(ctx, probeTimeout+n.IPCDispatchTimeout) defer probeCancel() results, err := AwaitAll(probeCtx, probes) if err != nil { if ctxErr := probeCtx.Err(); ctxErr != nil && errors.Is(err, ctxErr) { - return errResponse(err.Error()) + return errResponse(protocol.IpcError_UNAVAILABLE, "%v", err) } } return &protocol.IpcResponse{ - Ok: true, Response: &protocol.IpcResponse_Probe{Probe: &protocol.ProbeResponse{Results: results}}, } } @@ -465,11 +559,11 @@ func ipcProbeTimeout(req *protocol.ProbeRequest) time.Duration { func handleIPCReload(n *Nylon, req *protocol.ReloadRequest) *protocol.IpcResponse { data, err := os.ReadFile(n.ConfigPath) if err != nil { - return errResponse(fmt.Sprintf("read file: %v", err)) + return errResponse(protocol.IpcError_INTERNAL, "read file: %v", err) } var cfg state.CentralCfg if err = yaml.Unmarshal(data, &cfg); err != nil { - return errResponse(fmt.Sprintf("parse config: %v", err)) + return errResponse(protocol.IpcError_INVALID_REQUEST, "parse config: %v", err) } // We're running on the dispatch goroutine, so call ApplyCentralConfig // directly rather than re-dispatching (which would deadlock). @@ -489,46 +583,35 @@ func handleIPCReload(n *Nylon, req *protocol.ReloadRequest) *protocol.IpcRespons case ApplyRestartRequired: protoResult = protocol.ReloadResult_RESTART_REQUIRED } - return &protocol.IpcResponse{ - Ok: result != ApplyRejected, + resp := &protocol.IpcResponse{ Response: &protocol.IpcResponse_Reload{Reload: &protocol.ReloadResponse{ Result: protoResult, Message: msg, }}, } + if result == ApplyRejected { + resp.Error = ipc.Errorf(protocol.IpcError_INVALID_REQUEST, "config rejected: %s", msg) + } + return resp } -func handleTrace(n *Nylon, rw *bufio.ReadWriter) error { +// handleTrace sends each trace line until ctx ends: the client cancels, hangs up, or nylon stops. +func handleTrace(ctx context.Context, n *Nylon, send func(*protocol.IpcResponse) error) { if !n.DBG_trace_tc { - if err := writeResponse(rw, errResponse("tracing not enabled; restart with --dbg-trace-tc")); err != nil { - return err - } - return ErrIPCStatusHandled - } - // Send initial OK - if err := writeResponse(rw, &protocol.IpcResponse{Ok: true}); err != nil { - return err + _ = send(errResponse(protocol.IpcError_FAILED_PRECONDITION, "tracing not enabled; restart with --dbg-trace-tc")) + return } - ctx, cancel := context.WithCancel(context.Background()) - go func() { - _, _ = rw.ReadByte() // wait for EOF / disconnect - cancel() - }() ch := make(chan interface{}) n.Trace.Register(ch) defer n.Trace.Unregister(ch) for { select { case <-ctx.Done(): - return ErrIPCStatusHandled + return case msg := <-ch: if str, ok := msg.(string); ok { - resp := &protocol.IpcResponse{ - Ok: true, - Response: &protocol.IpcResponse_Trace{Trace: &protocol.TraceEvent{Line: str}}, - } - if err := writeResponse(rw, resp); err != nil { - return ErrIPCStatusHandled + if err := send(&protocol.IpcResponse{Response: &protocol.IpcResponse_Trace{Trace: &protocol.TraceEvent{Line: str}}}); err != nil { + return } } } diff --git a/core/nylon.go b/core/nylon.go index ca9f48be..e78d7bb9 100644 --- a/core/nylon.go +++ b/core/nylon.go @@ -68,8 +68,8 @@ type Nylon struct { // peerReceived is when each peer last received a packet. It ranks links without health. peerReceived map[polyamide.Peer]time.Time tunWorkers sync.WaitGroup - // wireGuard serves legacy UAPI. It is nil for custom runtimes. - wireGuard *legacyWireGuard + // ipc serves nylon's socket. It is nil for custom runtimes. + ipc *ipcServer resourceCleanup sync.Once // systemConfigured is set once startup begins configuring the host with // pre-up commands. Only then does cleanup undo host configuration, so a diff --git a/core/nylon_wireguard.go b/core/nylon_wireguard.go index d32bf974..9263f167 100644 --- a/core/nylon_wireguard.go +++ b/core/nylon_wireguard.go @@ -38,8 +38,8 @@ func (n *Nylon) initTransport() error { return err } } - if n.wireGuard != nil { - n.wireGuard.serve(n.Context, n.Log) + if n.ipc != nil { + n.ipc.serve(n) } n.startTUNReader() n.watchTUNEvents() @@ -108,8 +108,8 @@ func (n *Nylon) cleanupTransport() error { } } var err error - if n.wireGuard != nil { - err = errors.Join(err, n.wireGuard.close()) + if n.ipc != nil { + err = errors.Join(err, n.ipc.close()) } if n.Tun != nil { err = errors.Join(err, n.Tun.Close()) diff --git a/core/observability.go b/core/observability.go index 63971df2..01b90db5 100644 --- a/core/observability.go +++ b/core/observability.go @@ -176,12 +176,12 @@ func writePrometheusMetrics(w io.Writer, status *protocol.StatusResponse) { for _, neigh := range status.Neighbours { labels := map[string]string{"peer": neigh.PeerId} - wg := neigh.GetWireguard() - metrics.metric("nylon_wireguard_peer_transmit_bytes_total", "WireGuard bytes transmitted to a peer.", "counter", labels, float64(wg.TxBytes)) - metrics.metric("nylon_wireguard_peer_receive_bytes_total", "WireGuard bytes received from a peer.", "counter", labels, float64(wg.RxBytes)) + stats := neigh.GetTransport() + metrics.metric("nylon_wireguard_peer_transmit_bytes_total", "WireGuard bytes transmitted to a peer.", "counter", labels, float64(stats.TxBytes)) + metrics.metric("nylon_wireguard_peer_receive_bytes_total", "WireGuard bytes received from a peer.", "counter", labels, float64(stats.RxBytes)) handshake := float64(0) - if wg.LatestHandshakeUnix > 0 { - handshake = float64(wg.LatestHandshakeUnix / int64(time.Second)) + if stats.LatestHandshakeUnix > 0 { + handshake = float64(stats.LatestHandshakeUnix / int64(time.Second)) } metrics.metric("nylon_wireguard_peer_latest_handshake_seconds", "Unix time of the latest WireGuard handshake.", "gauge", labels, handshake) for _, endpoint := range neigh.Endpoints { diff --git a/core/observability_test.go b/core/observability_test.go index 9e017e99..017dac9d 100644 --- a/core/observability_test.go +++ b/core/observability_test.go @@ -66,12 +66,12 @@ func TestPrometheusMetrics(t *testing.T) { Neighbours: []*protocol.NeighbourInfo{ { PeerId: "bob", - Wireguard: &protocol.WireGuardPeerStats{ + Transport: &protocol.TransportPeerStats{ TxBytes: 7, LatestHandshakeUnix: 1786226710556904600, }, }, - {PeerId: "eve", Wireguard: &protocol.WireGuardPeerStats{TxBytes: 5}}, + {PeerId: "eve", Transport: &protocol.TransportPeerStats{TxBytes: 5}}, }, } var buf bytes.Buffer diff --git a/core/sys_darwin.go b/core/sys_darwin.go index abe1164d..bcb87a2e 100644 --- a/core/sys_darwin.go +++ b/core/sys_darwin.go @@ -6,22 +6,8 @@ import ( "net/netip" "github.com/encodeous/nylon/polyamide" - "github.com/encodeous/polyamide-wireguard-go/ipc" ) -func InitUAPI(logger *slog.Logger, itfName string) (net.Listener, error) { - fileUAPI, err := ipc.UAPIOpen(itfName) - if err != nil { - return nil, err - } - - uapi, err := ipc.UAPIListen(itfName, fileUAPI) - if err != nil { - return nil, err - } - return uapi, nil -} - func InitInterface(logger *slog.Logger, ifName string) error { return nil } diff --git a/core/sys_linux.go b/core/sys_linux.go index d2964a44..64d225ce 100644 --- a/core/sys_linux.go +++ b/core/sys_linux.go @@ -2,27 +2,12 @@ package core import ( "log/slog" - "net" "net/netip" "github.com/encodeous/nylon/polyamide" "github.com/encodeous/nylon/state" - "github.com/encodeous/polyamide-wireguard-go/ipc" ) -func InitUAPI(logger *slog.Logger, itfName string) (net.Listener, error) { - fileUAPI, err := ipc.UAPIOpen(itfName) - if err != nil { - return nil, err - } - - uapi, err := ipc.UAPIListen(itfName, fileUAPI) - if err != nil { - return nil, err - } - return uapi, nil -} - func InitInterface(logger *slog.Logger, ifName string) error { err := Exec(logger, "ip", "link", "set", ifName, "up") if err != nil { diff --git a/core/sys_physical.go b/core/sys_physical.go index a252c8ab..ca8c5a9c 100644 --- a/core/sys_physical.go +++ b/core/sys_physical.go @@ -7,6 +7,7 @@ import ( "net" "runtime" + "github.com/encodeous/nylon/ipc" "github.com/encodeous/nylon/polyamide" "github.com/encodeous/polyamide-wireguard-go/device" "github.com/encodeous/polyamide-wireguard-go/tun" @@ -35,14 +36,15 @@ func newDefaultRuntime(n *Nylon) (Runtime, error) { actual, err := host.Name() var listener net.Listener if err == nil { - listener, err = InitUAPI(n.Log, actual) + // named after nylon's interface, not the OS's (utunN on macOS), so `nylon -i ` finds it + listener, err = ipc.Listen(n.InterfaceName) } if err != nil { host.Close() transport.Close() return Runtime{}, err } - n.wireGuard = &legacyWireGuard{transport: transport, listener: listener} + n.ipc = &ipcServer{listener: listener} if n.NoTun { n.Log.Info("Created userspace-only WireGuard device", "name", actual) } else { diff --git a/core/sys_virtual.go b/core/sys_virtual.go index ac190581..ed34f8be 100644 --- a/core/sys_virtual.go +++ b/core/sys_virtual.go @@ -4,8 +4,6 @@ package core import ( "fmt" - - "github.com/encodeous/polyamide-wireguard-go/adapter" ) // The integration environment supplies a complete runtime. Nylon does not know @@ -19,15 +17,5 @@ func newDefaultRuntime(n *Nylon) (Runtime, error) { if !ok { return Runtime{}, fmt.Errorf("expected aux config vnet runtime factory") } - rt, err := factory.NewRuntime(n) - if err != nil { - return Runtime{}, err - } - // Tests use the WireGuard transport's UAPI without a socket. - if len(rt.Transports) != 0 { - if wg, ok := rt.Transports[0].(*adapter.Transport); ok { - n.wireGuard = &legacyWireGuard{transport: wg} - } - } - return rt, nil + return factory.NewRuntime(n) } diff --git a/core/sys_windows.go b/core/sys_windows.go index 1df1c183..26941ac4 100644 --- a/core/sys_windows.go +++ b/core/sys_windows.go @@ -6,25 +6,11 @@ import ( "net" "net/netip" "strconv" - "strings" "github.com/encodeous/nylon/polyamide" - "github.com/encodeous/polyamide-wireguard-go/ipc" "github.com/kmahyyg/go-network-compo/wintypes" ) -func InitUAPI(logger *slog.Logger, itfName string) (net.Listener, error) { - uapi, err := ipc.UAPIListen(itfName) - if err != nil && strings.Contains(err.Error(), "This security ID may not be assigned as the owner of this object") { - logger.Warn("UAPI not started. Nylon needs to be run with SYSTEM privileges. See: https://github.com/WireGuard/wgctrl-go/issues/141") - return nil, nil - } - if err != nil { - return nil, err - } - return uapi, nil -} - func InitInterface(logger *slog.Logger, ifName string) error { return nil } diff --git a/core/sys_wireguard.go b/core/sys_wireguard.go index f664aac8..d913718d 100644 --- a/core/sys_wireguard.go +++ b/core/sys_wireguard.go @@ -1,86 +1,13 @@ package core import ( - "bufio" - "context" - "errors" - "log/slog" - "net" - "sync" - "github.com/encodeous/nylon/log" "github.com/encodeous/nylon/polyamide" "github.com/encodeous/polyamide-wireguard-go/adapter" "github.com/encodeous/polyamide-wireguard-go/conn" - "github.com/encodeous/polyamide-wireguard-go/device" ) -// NewWireGuardTransport creates the WireGuard transport for n and serves Nylon IPC -// over its UAPI socket. A nil newBind uses the platform UDP bind. +// NewWireGuardTransport creates the WireGuard transport for n. A nil newBind uses the platform UDP bind. func NewWireGuardTransport(n *Nylon, newBind func() conn.Bind) (*adapter.Transport, error) { - return adapter.New(adapter.Options{PrivateKey: polyamide.PrivateKey(n.Key), ListenPort: n.Port, NewBind: newBind, Logger: n.Log.With("module", log.ScopePolyamide), Debug: n.DBG_log_wireguard, - ControlHandlers: map[string]func(*bufio.ReadWriter) error{"get=nylon\n": func(w *bufio.ReadWriter) error { - err := HandleNylonIPC(n, w) - if errors.Is(err, ErrIPCStatusHandled) { - return device.ErrIPCStatusHandled - } - return err - }}, - }) -} - -// legacyWireGuard serves WireGuard's UAPI, which the wg tool and Nylon's IPC use. -// It exists for compatibility: only the platform runtimes create it, and it only -// covers the WireGuard transport. -type legacyWireGuard struct { - transport *adapter.Transport - listener net.Listener // nil when there is no UAPI socket - workers sync.WaitGroup -} - -// serve accepts UAPI connections until the listener closes. -func (w *legacyWireGuard) serve(ctx context.Context, logger *slog.Logger) { - if w.listener == nil { - return - } - w.workers.Add(1) - go func() { - defer w.workers.Done() - for ctx.Err() == nil { - c, err := w.listener.Accept() - if err != nil { - if ctx.Err() != nil || errors.Is(err, net.ErrClosed) { - return - } - logger.Debug("UAPI accept failed", "err", err) - continue - } - w.workers.Add(1) - go func() { - defer w.workers.Done() - defer c.Close() - stop := context.AfterFunc(ctx, func() { _ = c.Close() }) - defer stop() - w.transport.HandleUAPI(c) - }() - } - }() -} - -func (w *legacyWireGuard) close() error { - var err error - if w.listener != nil { - err = w.listener.Close() - } - w.workers.Wait() - return err -} - -// HandleUAPI serves one legacy UAPI connection with the WireGuard transport. -func (n *Nylon) HandleUAPI(c net.Conn) error { - if n.wireGuard == nil { - return polyamide.ErrUnsupported - } - n.wireGuard.transport.HandleUAPI(c) - return nil + return adapter.New(adapter.Options{PrivateKey: polyamide.PrivateKey(n.Key), ListenPort: n.Port, NewBind: newBind, Logger: n.Log.With("module", log.ScopePolyamide), Debug: n.DBG_log_wireguard}) } diff --git a/e2e/ipc_test.go b/e2e/ipc_test.go index 816f1f63..31df0b3e 100644 --- a/e2e/ipc_test.go +++ b/e2e/ipc_test.go @@ -9,7 +9,8 @@ import ( "github.com/encodeous/nylon/state" ) -func TestIPCRawSocketJSONHasNoErrnoTrailer(t *testing.T) { +// Speaks the wire format byte by byte, so a change to it shows up here even if client and server change together. +func TestIPCRawSocketBinaryProtocol(t *testing.T) { h := NewHarness(t) configDir := h.SetupTestDir() @@ -29,31 +30,48 @@ func TestIPCRawSocketJSONHasNoErrnoTrailer(t *testing.T) { NodeConfigPath: nodePath, }) + // hello, then a frame holding IpcRequest{id: 1, status: {}} stdout, _, err := h.Exec("node1", []string{ "python3", "-c", - `import socket + `import socket, struct s=socket.socket(socket.AF_UNIX, socket.SOCK_STREAM) -s.connect("/var/run/wireguard/nylon0.sock") -s.sendall(b'get=nylon\n{"status":{}}\n') -line=s.recv(65535).decode() -print(line, end="") -s.settimeout(0.2) -try: - extra=s.recv(65535).decode() - print(extra, end="") -except TimeoutError: - pass +s.connect("/var/run/nylon/nylon0.sock") +s.sendall(b"NYLN\x00\x01" + struct.pack(">I", 4) + b"\x08\x01\x12\x00") +def read(n): + data=b"" + while len(data) < n: + chunk=s.recv(n - len(data)) + if not chunk: + raise EOFError + data+=chunk + return data +print(read(6).hex()) +size=struct.unpack(">I", read(4))[0] +print(read(size)[:3].hex()) s.close() `, }) if err != nil { t.Fatalf("raw IPC request failed: %v\nstdout:\n%s", err, stdout) } - if strings.Contains(stdout, "errno=") { - t.Fatalf("raw IPC response included UAPI errno trailer:\n%s", stdout) + lines := strings.Fields(stdout) + if len(lines) != 2 { + t.Fatalf("unexpected output:\n%s", stdout) } - if !strings.Contains(stdout, `"ok":true`) { - t.Fatalf("raw IPC response did not include a successful protojson response:\n%s", stdout) + if lines[0] != "4e594c4e0001" { + t.Fatalf("hello was %s, want NYLN version 1", lines[0]) + } + // id 1 (08 01), then field 3, the status (1a), and no field 2, an error + if lines[1] != "08011a" { + t.Fatalf("response started %s, want a status for request 1", lines[1]) + } + + // nylon doesn't serve WireGuard's UAPI, so the wg tool can't see or change its transport. + if stdout, _, err := h.Exec("node1", []string{"test", "-e", "/var/run/wireguard/nylon0.sock"}); err == nil { + t.Fatalf("WireGuard UAPI socket exists:\n%s", stdout) + } + if stdout, _, err := h.Exec("node1", []string{"wg", "show", "nylon0"}); err == nil { + t.Fatalf("wg can talk to nylon:\n%s", stdout) } } diff --git a/e2e/probe_test.go b/e2e/probe_test.go index 9b5e8112..e57262e3 100644 --- a/e2e/probe_test.go +++ b/e2e/probe_test.go @@ -161,7 +161,7 @@ func waitForProbeResult(t *testing.T, h *Harness, nodeName, peerName, address st last := "" for time.Now().Before(deadline) { resp, stdout, stderr, err := runProbeJSON(h, nodeName, peerName, probeTimeout) - if err == nil && resp.GetOk() { + if err == nil && resp.GetError() == nil { for _, result := range resp.GetProbe().GetResults() { if result.GetAddress() == address && check(result) { return result diff --git a/go.mod b/go.mod index 1f6b50b9..4cb87982 100644 --- a/go.mod +++ b/go.mod @@ -6,7 +6,7 @@ require ( github.com/digineo/go-ping v1.2.0 github.com/dustin/go-broadcast v0.0.0-20211018055107-71439988bd91 github.com/encodeous/metric v0.0.0-20251111175231-f339c2f7c4bd - github.com/encodeous/polyamide-wireguard-go v0.0.0-20260930194717-ff21c4e94337 + github.com/encodeous/polyamide-wireguard-go v0.0.0-20261101192016-21817349e7b4 github.com/encodeous/tint v1.2.0 github.com/gaissmai/bart v0.25.0 github.com/goccy/go-yaml v1.19.1 @@ -24,6 +24,7 @@ require ( go.uber.org/goleak v1.3.0 go4.org/netipx v0.0.0-20231129151722-fdeea329fbba golang.org/x/crypto v0.52.0 + golang.org/x/sys v0.47.0 google.golang.org/protobuf v1.36.10 ) @@ -81,7 +82,6 @@ require ( go.opentelemetry.io/otel/trace v1.45.0 // indirect golang.org/x/net v0.55.0 // indirect golang.org/x/sync v0.20.0 // indirect - golang.org/x/sys v0.47.0 // indirect golang.org/x/text v0.37.0 // indirect golang.zx2c4.com/wintun v0.0.0-20230126152724-0fa3db229ce2 // indirect gopkg.in/yaml.v3 v3.0.1 // indirect diff --git a/go.sum b/go.sum index 8b941540..77d6c7d8 100644 --- a/go.sum +++ b/go.sum @@ -43,8 +43,8 @@ github.com/ebitengine/purego v0.10.0 h1:QIw4xfpWT6GWTzaW5XEKy3HXoqrJGx1ijYHzTF0/ github.com/ebitengine/purego v0.10.0/go.mod h1:iIjxzd6CiRiOG0UyXP+V1+jWqUXVjPKLAI0mRfJZTmQ= github.com/encodeous/metric v0.0.0-20251111175231-f339c2f7c4bd h1:B32Ob80QTv5MomcVt709TsiWyD0QrpUYtnwW1jQFNlE= github.com/encodeous/metric v0.0.0-20251111175231-f339c2f7c4bd/go.mod h1:DiXCPJtfZYioejF9zv9wfs3TXqWWglKGQ20DsBNVWVw= -github.com/encodeous/polyamide-wireguard-go v0.0.0-20260930194717-ff21c4e94337 h1:HuwThz05/Khp6lZ4yHjIZHWCf0y5Flr2JOP6nBE6wQQ= -github.com/encodeous/polyamide-wireguard-go v0.0.0-20260930194717-ff21c4e94337/go.mod h1:SulefP9J0mccyOLDA+5Aa5nO569HF9YgkSRmx48wkQA= +github.com/encodeous/polyamide-wireguard-go v0.0.0-20261101192016-21817349e7b4 h1:kr+P9WO7u+hPmOkx7XtuSA7Al8Ngdfr9KJc5gNQSDVo= +github.com/encodeous/polyamide-wireguard-go v0.0.0-20261101192016-21817349e7b4/go.mod h1:SulefP9J0mccyOLDA+5Aa5nO569HF9YgkSRmx48wkQA= github.com/encodeous/tint v1.2.0 h1:1Y+32Iu+C8MXBoNjsM4YDf6iAkcks7csAI9f7b4fr8k= github.com/encodeous/tint v1.2.0/go.mod h1:pyfyH+fKmtmIuWVWeMcSJjJKVJSb9sVlIfVsW3jkP+Q= github.com/felixge/httpsnoop v1.0.4 h1:NFTV2Zj1bL4mc9sqWACXbQFVBBg2W3GPvqp8/ESS2Wg= diff --git a/integration/ipc_test.go b/integration/ipc_test.go index f3dce55f..de4b78b5 100644 --- a/integration/ipc_test.go +++ b/integration/ipc_test.go @@ -3,22 +3,23 @@ package integration import ( - "bufio" - "bytes" + "encoding/binary" "errors" + "io" "net" "os" + "sync" "testing" "time" "github.com/encodeous/nylon/core" + "github.com/encodeous/nylon/ipc" "github.com/encodeous/nylon/protocol" "github.com/encodeous/nylon/state" "github.com/goccy/go-yaml" "github.com/stretchr/testify/assert" "github.com/stretchr/testify/require" "go.uber.org/goleak" - "google.golang.org/protobuf/encoding/protojson" ) func setupTwoNodeHarness(t *testing.T) (*VirtualHarness, chan error) { @@ -39,25 +40,35 @@ func setupTwoNodeHarness(t *testing.T) (*VirtualHarness, chan error) { return vh, errs } -func ipcCall(t *testing.T, n *core.Nylon, req *protocol.IpcRequest) *protocol.IpcResponse { +// ipcClient connects a client to n's IPC server over an in-memory pipe. +func ipcClient(t *testing.T, n *core.Nylon) *ipc.Client { t.Helper() - m := protojson.MarshalOptions{EmitUnpopulated: true} - data, err := m.Marshal(req) + client, server := net.Pipe() + done := make(chan struct{}) + go func() { + defer close(done) + _ = core.ServeIPC(n, server) + }() + t.Cleanup(func() { + _ = client.Close() + <-done + }) + c, err := ipc.NewClient(client) require.NoError(t, err) - data = append(data, '\n') + return c +} - var outBuf bytes.Buffer - rw := bufio.NewReadWriter( - bufio.NewReader(bytes.NewReader(data)), - bufio.NewWriter(&outBuf), - ) - err = core.HandleNylonIPC(n, rw) - require.True(t, errors.Is(err, core.ErrIPCStatusHandled), "unexpected error: %v", err) +func ipcCall(t *testing.T, n *core.Nylon, req *protocol.IpcRequest) (*protocol.IpcResponse, error) { + t.Helper() + return ipcClient(t, n).Do(req) +} - resp := &protocol.IpcResponse{} - um := protojson.UnmarshalOptions{DiscardUnknown: true} - require.NoError(t, um.Unmarshal(outBuf.Bytes(), resp)) - return resp +// ipcErrorCode is the code of the daemon's error, failing the test for any other kind of error. +func ipcErrorCode(t *testing.T, err error) protocol.IpcError_Code { + t.Helper() + var ipcErr *ipc.Error + require.True(t, errors.As(err, &ipcErr), "expected a daemon error, got %v", err) + return ipcErr.Code } func TestIPCStatus(t *testing.T) { @@ -69,11 +80,10 @@ func TestIPCStatus(t *testing.T) { time.Sleep(3 * time.Second) a := vh.Nylons[vh.IndexOf("a")].Load() - resp := ipcCall(t, a, &protocol.IpcRequest{ + resp, err := ipcCall(t, a, &protocol.IpcRequest{ Request: &protocol.IpcRequest_Status{Status: &protocol.StatusRequest{}}, }) - - assert.True(t, resp.Ok) + require.NoError(t, err) s := resp.GetStatus() require.NotNil(t, s.GetNode()) assert.Equal(t, "a", s.GetNode().NodeId) @@ -112,11 +122,10 @@ func TestIPCProbeReportsTimeout(t *testing.T) { time.Sleep(500 * time.Millisecond) a := vh.Nylons[vh.IndexOf("a")].Load() - resp := ipcCall(t, a, &protocol.IpcRequest{ + resp, err := ipcCall(t, a, &protocol.IpcRequest{ Request: &protocol.IpcRequest_Probe{Probe: &protocol.ProbeRequest{PeerId: "b", TimeoutMs: 100}}, }) - - require.True(t, resp.Ok, resp.Error) + require.NoError(t, err) results := resp.GetProbe().GetResults() require.NotEmpty(t, results) for _, result := range results { @@ -146,18 +155,23 @@ func TestIPCReloadConfig(t *testing.T) { a := vh.Nylons[vh.IndexOf("a")].Load() a.ConfigPath = tmpFile - done := make(chan *protocol.IpcResponse, 1) + client := ipcClient(t, a) + type result struct { + resp *protocol.IpcResponse + err error + } + done := make(chan result, 1) go func() { - resp := ipcCall(t, a, &protocol.IpcRequest{ + resp, err := client.Do(&protocol.IpcRequest{ Request: &protocol.IpcRequest_Reload{Reload: &protocol.ReloadRequest{}}, }) - done <- resp + done <- result{resp, err} }() select { - case resp := <-done: - assert.True(t, resp.Ok) - r := resp.GetReload() + case res := <-done: + require.NoError(t, res.err) + r := res.resp.GetReload() assert.Equal(t, protocol.ReloadResult_NOOP, r.Result) case err := <-errs: t.Fatal(err) @@ -174,117 +188,126 @@ func TestIPCProbeNonNeighbour(t *testing.T) { time.Sleep(1 * time.Second) a := vh.Nylons[vh.IndexOf("a")].Load() - resp := ipcCall(t, a, &protocol.IpcRequest{ + _, err := ipcCall(t, a, &protocol.IpcRequest{ Request: &protocol.IpcRequest_Probe{Probe: &protocol.ProbeRequest{PeerId: "nonexistent"}}, }) - - assert.False(t, resp.Ok) - assert.Contains(t, resp.Error, "not a neighbour") + assert.Equal(t, protocol.IpcError_NOT_FOUND, ipcErrorCode(t, err)) + assert.Contains(t, err.Error(), "not a neighbour") } func TestIPCTraceDisabled(t *testing.T) { defer goleak.VerifyNone(t) - vh, errs := setupTwoNodeHarness(t) + vh, _ := setupTwoNodeHarness(t) defer vh.Stop() time.Sleep(1 * time.Second) a := vh.Nylons[vh.IndexOf("a")].Load() - done := make(chan *protocol.IpcResponse, 1) - a.Dispatch(func() error { - // Trace should fail since DBG_trace_tc is false by default - resp := ipcCall(t, a, &protocol.IpcRequest{ - Request: &protocol.IpcRequest_Trace{Trace: &protocol.TraceRequest{}}, - }) - done <- resp - return nil + // DBG_trace_tc is false by default + _, err := ipcCall(t, a, &protocol.IpcRequest{ + Request: &protocol.IpcRequest_Trace{Trace: &protocol.TraceRequest{}}, }) - - select { - case resp := <-done: - assert.False(t, resp.Ok) - assert.Contains(t, resp.Error, "tracing not enabled") - case err := <-errs: - t.Fatal(err) - case <-time.After(10 * time.Second): - t.Fatal("timeout") - } + assert.Equal(t, protocol.IpcError_FAILED_PRECONDITION, ipcErrorCode(t, err)) + assert.Contains(t, err.Error(), "tracing not enabled") } +// A frame that isn't an IpcRequest gets an error response, then the server hangs up. func TestIPCMalformedRequest(t *testing.T) { defer goleak.VerifyNone(t) - vh, errs := setupTwoNodeHarness(t) + vh, _ := setupTwoNodeHarness(t) defer vh.Stop() time.Sleep(1 * time.Second) a := vh.Nylons[vh.IndexOf("a")].Load() - done := make(chan bool, 1) - a.Dispatch(func() error { - // Send garbage - var outBuf bytes.Buffer - rw := bufio.NewReadWriter( - bufio.NewReader(bytes.NewReader([]byte("not json\n"))), - bufio.NewWriter(&outBuf), - ) - err := core.HandleNylonIPC(a, rw) - // Should write an error response, not crash - assert.True(t, errors.Is(err, core.ErrIPCStatusHandled), "unexpected error: %v", err) - resp := &protocol.IpcResponse{} - um := protojson.UnmarshalOptions{DiscardUnknown: true} - assert.NoError(t, um.Unmarshal(outBuf.Bytes(), resp)) - assert.False(t, resp.Ok) - done <- true - return nil - }) + client, server := net.Pipe() + defer client.Close() + go core.ServeIPC(a, server) - select { - case <-done: - case err := <-errs: - t.Fatal(err) - case <-time.After(10 * time.Second): - t.Fatal("timeout") - } + require.NoError(t, ipc.WriteHello(client)) + version, err := ipc.ReadHello(client) + require.NoError(t, err) + require.Equal(t, ipc.Version, version) + + garbage := []byte{0xff, 0xff, 0xff} + frame := binary.BigEndian.AppendUint32(nil, uint32(len(garbage))) + _, err = client.Write(append(frame, garbage...)) + require.NoError(t, err) + + resp := &protocol.IpcResponse{} + require.NoError(t, ipc.ReadFrame(client, resp)) + assert.Equal(t, protocol.IpcError_INVALID_REQUEST, resp.GetError().GetCode()) + _, err = client.Read(make([]byte, 1)) + assert.ErrorIs(t, err, io.EOF) } -func TestIPCSocketResponseHasNoUAPIErrnoTrailer(t *testing.T) { +// A client with another protocol version learns the server's, and the server hangs up. +func TestIPCVersionMismatch(t *testing.T) { defer goleak.VerifyNone(t) - vh, errs := setupTwoNodeHarness(t) + vh, _ := setupTwoNodeHarness(t) defer vh.Stop() - time.Sleep(1 * time.Second) - a := vh.Nylons[vh.IndexOf("a")].Load() client, server := net.Pipe() defer client.Close() - go a.HandleUAPI(server) + served := make(chan error, 1) + go func() { served <- core.ServeIPC(a, server) }() - m := protojson.MarshalOptions{EmitUnpopulated: true} - data, err := m.Marshal(&protocol.IpcRequest{ - Request: &protocol.IpcRequest_Status{Status: &protocol.StatusRequest{}}, - }) + _, err := client.Write([]byte{'N', 'Y', 'L', 'N', 0xff, 0xff}) require.NoError(t, err) - - _, err = client.Write(append(append([]byte("get=nylon\n"), data...), '\n')) + version, err := ipc.ReadHello(client) require.NoError(t, err) + assert.Equal(t, ipc.Version, version) + assert.ErrorAs(t, <-served, &ipc.ErrVersion{}) +} - reader := bufio.NewReader(client) - line, err := reader.ReadBytes('\n') - require.NoError(t, err) +// Requests on one connection run concurrently, and cancelling one ends it early. +func TestIPCConcurrentRequestsAndCancel(t *testing.T) { + defer goleak.VerifyNone(t) + vh := &VirtualHarness{} + a1 := "192.168.53.1:1234" + b1 := "192.168.53.2:1234" + vh.NewNode("a", "10.0.0.1/32") + vh.NewNode("b", "10.0.0.2/32") + vh.Central.Graph = []string{"a, b"} + vh.Endpoints = map[string]state.NodeId{a1: "a", b1: "b"} + vh.AddLink(a1, b1).WithPacketLoss(1) + vh.AddLink(b1, a1).WithPacketLoss(1) + vh.Start() + defer vh.Stop() - resp := &protocol.IpcResponse{} - um := protojson.UnmarshalOptions{DiscardUnknown: true} - require.NoError(t, um.Unmarshal(line, resp)) - require.True(t, resp.Ok) + time.Sleep(500 * time.Millisecond) - require.NoError(t, client.SetReadDeadline(time.Now().Add(100*time.Millisecond))) - extra, err := reader.ReadString('\n') - require.Error(t, err) - require.NotContains(t, extra, "errno=") + a := vh.Nylons[vh.IndexOf("a")].Load() + client, server := net.Pipe() + defer client.Close() + go core.ServeIPC(a, server) + require.NoError(t, ipc.WriteHello(client)) + _, err := ipc.ReadHello(client) + require.NoError(t, err) + require.NoError(t, client.SetDeadline(time.Now().Add(10*time.Second))) - select { - case err := <-errs: - t.Fatal(err) - default: + var writeMu sync.Mutex + send := func(req *protocol.IpcRequest) { + writeMu.Lock() + defer writeMu.Unlock() + require.NoError(t, ipc.WriteFrame(client, req)) + } + read := func() *protocol.IpcResponse { + resp := &protocol.IpcResponse{} + require.NoError(t, ipc.ReadFrame(client, resp)) + return resp } + // Every packet is lost, so the probe waits out its whole timeout. + start := time.Now() + send(&protocol.IpcRequest{Id: 1, Request: &protocol.IpcRequest_Probe{Probe: &protocol.ProbeRequest{PeerId: "b", TimeoutMs: 5000}}}) + send(&protocol.IpcRequest{Id: 2, Request: &protocol.IpcRequest_Status{Status: &protocol.StatusRequest{}}}) + resp := read() + require.Equal(t, uint32(2), resp.Id, "status should not wait for the probe") + require.Equal(t, "a", resp.GetStatus().GetNode().GetNodeId()) + + send(&protocol.IpcRequest{Request: &protocol.IpcRequest_Cancel{Cancel: &protocol.CancelRequest{Id: 1}}}) + resp = read() + require.Equal(t, uint32(1), resp.Id) + require.Less(t, time.Since(start), 4*time.Second, "the cancelled probe should end before its timeout") } diff --git a/ipc/client.go b/ipc/client.go new file mode 100644 index 00000000..15f490ea --- /dev/null +++ b/ipc/client.go @@ -0,0 +1,84 @@ +package ipc + +import ( + "bufio" + "fmt" + "net" + + "github.com/encodeous/nylon/protocol" +) + +// Client is a connection to a nylon daemon. It sends one request at a time; the protocol allows more, but the +// CLI never needs them. +type Client struct { + conn net.Conn + r *bufio.Reader + nextID uint32 +} + +// Dial connects to the daemon serving the nylon interface itf. +func Dial(itf string) (*Client, error) { + conn, err := dial(itf) + if err != nil { + return nil, fmt.Errorf("connect to nylon on %s: %w", itf, err) + } + c, err := NewClient(conn) + if err != nil { + _ = conn.Close() + return nil, err + } + return c, nil +} + +// NewClient says hello over conn. The client owns conn from then on. +func NewClient(conn net.Conn) (*Client, error) { + if err := WriteHello(conn); err != nil { + return nil, err + } + r := bufio.NewReader(conn) + version, err := ReadHello(r) + if err != nil { + return nil, fmt.Errorf("read hello: %w", err) + } + if version != Version { + return nil, ErrVersion{Local: Version, Remote: version} + } + return &Client{conn: conn, r: r}, nil +} + +func (c *Client) Close() error { return c.conn.Close() } + +// Do sends req and returns its response. A response carrying an error comes back as *Error. +func (c *Client) Do(req *protocol.IpcRequest) (*protocol.IpcResponse, error) { + var resp *protocol.IpcResponse + err := c.Stream(req, func(r *protocol.IpcResponse) bool { + resp = r + return false + }) + return resp, err +} + +// Stream sends req and passes each response to fn until fn returns false, the daemon reports an error, or the +// connection closes. Stopping early leaves the request running; close the client to end it. +func (c *Client) Stream(req *protocol.IpcRequest, fn func(*protocol.IpcResponse) bool) error { + c.nextID++ + req.Id = c.nextID + if err := WriteFrame(c.conn, req); err != nil { + return err + } + for { + resp := &protocol.IpcResponse{} + if err := ReadFrame(c.r, resp); err != nil { + return err + } + if resp.Id != req.Id { + continue // a response to a request this client gave up on + } + if e := resp.GetError(); e != nil { + return &Error{Code: e.Code, Message: e.Message} + } + if !fn(resp) { + return nil + } + } +} diff --git a/ipc/ipc.go b/ipc/ipc.go new file mode 100644 index 00000000..df9e3ea4 --- /dev/null +++ b/ipc/ipc.go @@ -0,0 +1,98 @@ +// Package ipc is nylon's control protocol: the socket the daemon listens on, and how messages cross it. +// protocol/nylon_ipc.proto describes the wire format. +package ipc + +import ( + "bytes" + "encoding/binary" + "errors" + "fmt" + "io" + + "github.com/encodeous/nylon/protocol" + "google.golang.org/protobuf/proto" +) + +// Version is the protocol version +const Version uint16 = 1 + +// MaxFrame bounds a frame, so a bad peer can't make the other side allocate without limit. +const MaxFrame = 16 << 20 + +var magic = [4]byte{'N', 'Y', 'L', 'N'} + +// ErrVersion means the other side speaks a different protocol version. +type ErrVersion struct{ Local, Remote uint16 } + +func (e ErrVersion) Error() string { + return fmt.Sprintf("nylon IPC version mismatch: this side speaks %d, the other speaks %d", e.Local, e.Remote) +} + +var ErrNotNylon = errors.New("not a nylon IPC socket") + +// WriteHello sends this side's hello. +func WriteHello(w io.Writer) error { + var hello [6]byte + copy(hello[:4], magic[:]) + binary.BigEndian.PutUint16(hello[4:], Version) + _, err := w.Write(hello[:]) + return err +} + +// ReadHello reads the other side's hello and returns its version. +func ReadHello(r io.Reader) (uint16, error) { + var hello [6]byte + if _, err := io.ReadFull(r, hello[:]); err != nil { + return 0, err + } + if !bytes.Equal(hello[:4], magic[:]) { + return 0, ErrNotNylon + } + return binary.BigEndian.Uint16(hello[4:]), nil +} + +// WriteFrame sends m as one frame. Callers sharing a writer must serialize calls. +func WriteFrame(w io.Writer, m proto.Message) error { + data, err := proto.Marshal(m) + if err != nil { + return err + } + if len(data) > MaxFrame { + return fmt.Errorf("frame of %d bytes is over the %d byte limit", len(data), MaxFrame) + } + frame := make([]byte, 4+len(data)) + binary.BigEndian.PutUint32(frame, uint32(len(data))) + copy(frame[4:], data) + _, err = w.Write(frame) + return err +} + +// ReadFrame reads one frame into m. +func ReadFrame(r io.Reader, m proto.Message) error { + var size [4]byte + if _, err := io.ReadFull(r, size[:]); err != nil { + return err + } + n := binary.BigEndian.Uint32(size[:]) + if n > MaxFrame { + return fmt.Errorf("frame of %d bytes is over the %d byte limit", n, MaxFrame) + } + data := make([]byte, n) + if _, err := io.ReadFull(r, data); err != nil { + return err + } + return proto.Unmarshal(data, m) +} + +// Error is a request the daemon couldn't do. +type Error struct { + Code protocol.IpcError_Code + Message string +} + +func (e *Error) Error() string { return e.Message } + +// Errorf builds a response error for a request. +func Errorf(code protocol.IpcError_Code, format string, args ...any) *protocol.IpcError { + return &protocol.IpcError{Code: code, Message: fmt.Sprintf(format, args...)} +} diff --git a/ipc/ipc_test.go b/ipc/ipc_test.go new file mode 100644 index 00000000..3f10d692 --- /dev/null +++ b/ipc/ipc_test.go @@ -0,0 +1,39 @@ +package ipc + +import ( + "bytes" + "encoding/binary" + "testing" + + "github.com/encodeous/nylon/protocol" + "github.com/stretchr/testify/require" +) + +func TestFrameRoundTrip(t *testing.T) { + var buf bytes.Buffer + sent := &protocol.IpcRequest{Id: 7, Request: &protocol.IpcRequest_Probe{Probe: &protocol.ProbeRequest{PeerId: "bob"}}} + require.NoError(t, WriteFrame(&buf, sent)) + require.Equal(t, uint32(buf.Len()-4), binary.BigEndian.Uint32(buf.Bytes())) + + got := &protocol.IpcRequest{} + require.NoError(t, ReadFrame(&buf, got)) + require.Equal(t, uint32(7), got.Id) + require.Equal(t, "bob", got.GetProbe().GetPeerId()) +} + +// A length over the limit is refused before anything is allocated for it. +func TestFrameTooLarge(t *testing.T) { + frame := binary.BigEndian.AppendUint32(nil, MaxFrame+1) + require.ErrorContains(t, ReadFrame(bytes.NewReader(frame), &protocol.IpcRequest{}), "over the") +} + +func TestHello(t *testing.T) { + var buf bytes.Buffer + require.NoError(t, WriteHello(&buf)) + version, err := ReadHello(&buf) + require.NoError(t, err) + require.Equal(t, Version, version) + + _, err = ReadHello(bytes.NewReader([]byte("errno="))) + require.ErrorIs(t, err, ErrNotNylon) +} diff --git a/ipc/socket_unix.go b/ipc/socket_unix.go new file mode 100644 index 00000000..ec938f0c --- /dev/null +++ b/ipc/socket_unix.go @@ -0,0 +1,45 @@ +//go:build !windows + +package ipc + +import ( + "errors" + "fmt" + "net" + "os" + "path/filepath" + "syscall" + "time" +) + +// Dir holds the sockets, one per interface. Only root can use them. +var Dir = "/var/run/nylon" + +// Path is where the daemon for interface itf listens. +func Path(itf string) string { + return filepath.Join(Dir, itf+".sock") +} + +// Listen opens the socket for interface itf. It fails if another daemon is already serving it, and replaces a +// socket left behind by one that exited. +func Listen(itf string) (net.Listener, error) { + if err := os.MkdirAll(Dir, 0o755); err != nil { + return nil, err + } + path := Path(itf) + if conn, err := net.DialTimeout("unix", path, time.Second); err == nil { + _ = conn.Close() + return nil, fmt.Errorf("another nylon is already serving %s", path) + } + if err := os.Remove(path); err != nil && !errors.Is(err, os.ErrNotExist) { + return nil, err + } + // owner only, from the moment it exists + old := syscall.Umask(0o077) + defer syscall.Umask(old) + return net.Listen("unix", path) +} + +func dial(itf string) (net.Conn, error) { + return net.Dial("unix", Path(itf)) +} diff --git a/ipc/socket_unix_test.go b/ipc/socket_unix_test.go new file mode 100644 index 00000000..4375ed47 --- /dev/null +++ b/ipc/socket_unix_test.go @@ -0,0 +1,91 @@ +//go:build !windows + +package ipc + +import ( + "net" + "os" + "testing" + + "github.com/encodeous/nylon/protocol" + "github.com/stretchr/testify/require" +) + +// useTempDir points the sockets at a fresh folder. Unix socket paths are short, so not t.TempDir(). +func useTempDir(t *testing.T) { + dir, err := os.MkdirTemp("/tmp", "nylon-ipc") + require.NoError(t, err) + old := Dir + Dir = dir + t.Cleanup(func() { + Dir = old + _ = os.RemoveAll(dir) + }) +} + +func TestListenIsOwnerOnly(t *testing.T) { + useTempDir(t) + l, err := Listen("nylon") + require.NoError(t, err) + defer l.Close() + info, err := os.Stat(Path("nylon")) + require.NoError(t, err) + require.Zero(t, info.Mode().Perm()&0o077, "group and others must not have access, got %v", info.Mode().Perm()) +} + +func TestListenRefusesARunningDaemon(t *testing.T) { + useTempDir(t) + l, err := Listen("nylon") + require.NoError(t, err) + defer l.Close() + _, err = Listen("nylon") + require.ErrorContains(t, err, "already serving") +} + +// A daemon that crashed leaves its socket behind; the next one replaces it. +func TestListenReplacesAStaleSocket(t *testing.T) { + useTempDir(t) + l, err := net.Listen("unix", Path("nylon")) + require.NoError(t, err) + l.(*net.UnixListener).SetUnlinkOnClose(false) + require.NoError(t, l.Close()) + _, err = os.Stat(Path("nylon")) + require.NoError(t, err, "the stale socket should still exist") + + l, err = Listen("nylon") + require.NoError(t, err) + require.NoError(t, l.Close()) +} + +// The client against a minimal server over the real socket. +func TestDialAndDo(t *testing.T) { + useTempDir(t) + l, err := Listen("nylon") + require.NoError(t, err) + defer l.Close() + go func() { + conn, err := l.Accept() + if err != nil { + return + } + defer conn.Close() + if _, err := ReadHello(conn); err != nil { + return + } + _ = WriteHello(conn) + req := &protocol.IpcRequest{} + if ReadFrame(conn, req) != nil { + return + } + _ = WriteFrame(conn, &protocol.IpcResponse{Id: req.Id, Error: Errorf(protocol.IpcError_NOT_FOUND, "no peer %s", req.GetProbe().GetPeerId())}) + }() + + c, err := Dial("nylon") + require.NoError(t, err) + defer c.Close() + _, err = c.Do(&protocol.IpcRequest{Request: &protocol.IpcRequest_Probe{Probe: &protocol.ProbeRequest{PeerId: "bob"}}}) + var ipcErr *Error + require.ErrorAs(t, err, &ipcErr) + require.Equal(t, protocol.IpcError_NOT_FOUND, ipcErr.Code) + require.Equal(t, "no peer bob", ipcErr.Message) +} diff --git a/ipc/socket_windows.go b/ipc/socket_windows.go new file mode 100644 index 00000000..fafe0d2f --- /dev/null +++ b/ipc/socket_windows.go @@ -0,0 +1,35 @@ +package ipc + +import ( + "net" + "time" + + "github.com/encodeous/polyamide-wireguard-go/ipc/namedpipe" + "golang.org/x/sys/windows" +) + +// Only SYSTEM and Administrators can open it, at high integrity. +var securityDescriptor *windows.SECURITY_DESCRIPTOR + +func init() { + var err error + securityDescriptor, err = windows.SecurityDescriptorFromString("O:SYD:P(A;;GA;;;SY)(A;;GA;;;BA)S:(ML;;NWNRNX;;;HI)") + if err != nil { + panic(err) + } +} + +// Path is where the daemon for interface itf listens. Only administrators can create pipes under ProtectedPrefix, +// so another program can't pose as nylon. +func Path(itf string) string { + return `\\.\pipe\ProtectedPrefix\Administrators\nylon\` + itf +} + +// Listen opens the pipe for interface itf. +func Listen(itf string) (net.Listener, error) { + return (&namedpipe.ListenConfig{SecurityDescriptor: securityDescriptor}).Listen(Path(itf)) +} + +func dial(itf string) (net.Conn, error) { + return namedpipe.DialTimeout(Path(itf), 5*time.Second) +} diff --git a/protocol/nylon_ipc.pb.go b/protocol/nylon_ipc.pb.go index 16bcf0b9..a826c1f8 100644 --- a/protocol/nylon_ipc.pb.go +++ b/protocol/nylon_ipc.pb.go @@ -1,7 +1,7 @@ // Code generated by protoc-gen-go. DO NOT EDIT. // versions: -// protoc-gen-go v1.36.11 -// protoc v3.21.12 +// protoc-gen-go v1.36.10 +// protoc v7.36.2 // source: protocol/nylon_ipc.proto package protocol @@ -128,6 +128,61 @@ func (EndpointProbeStatus) EnumDescriptor() ([]byte, []int) { return file_protocol_nylon_ipc_proto_rawDescGZIP(), []int{1} } +type IpcError_Code int32 + +const ( + IpcError_INTERNAL IpcError_Code = 0 + IpcError_INVALID_REQUEST IpcError_Code = 1 + IpcError_UNAVAILABLE IpcError_Code = 2 // nylon is busy or shutting down; worth retrying + IpcError_FAILED_PRECONDITION IpcError_Code = 3 // e.g. trace without --dbg-trace-tc + IpcError_NOT_FOUND IpcError_Code = 4 +) + +// Enum value maps for IpcError_Code. +var ( + IpcError_Code_name = map[int32]string{ + 0: "INTERNAL", + 1: "INVALID_REQUEST", + 2: "UNAVAILABLE", + 3: "FAILED_PRECONDITION", + 4: "NOT_FOUND", + } + IpcError_Code_value = map[string]int32{ + "INTERNAL": 0, + "INVALID_REQUEST": 1, + "UNAVAILABLE": 2, + "FAILED_PRECONDITION": 3, + "NOT_FOUND": 4, + } +) + +func (x IpcError_Code) Enum() *IpcError_Code { + p := new(IpcError_Code) + *p = x + return p +} + +func (x IpcError_Code) String() string { + return protoimpl.X.EnumStringOf(x.Descriptor(), protoreflect.EnumNumber(x)) +} + +func (IpcError_Code) Descriptor() protoreflect.EnumDescriptor { + return file_protocol_nylon_ipc_proto_enumTypes[2].Descriptor() +} + +func (IpcError_Code) Type() protoreflect.EnumType { + return &file_protocol_nylon_ipc_proto_enumTypes[2] +} + +func (x IpcError_Code) Number() protoreflect.EnumNumber { + return protoreflect.EnumNumber(x) +} + +// Deprecated: Use IpcError_Code.Descriptor instead. +func (IpcError_Code) EnumDescriptor() ([]byte, []int) { + return file_protocol_nylon_ipc_proto_rawDescGZIP(), []int{26, 0} +} + type StatusRequest struct { state protoimpl.MessageState `protogen:"open.v1"` unknownFields protoimpl.UnknownFields @@ -732,7 +787,8 @@ func (x *EndpointInfo) GetStabilizedRttNs() int64 { return 0 } -type WireGuardPeerStats struct { +// What the transport reports about a peer. Byte counts include transport headers. +type TransportPeerStats struct { state protoimpl.MessageState `protogen:"open.v1"` LatestHandshakeUnix int64 `protobuf:"varint,1,opt,name=latest_handshake_unix,json=latestHandshakeUnix,proto3" json:"latest_handshake_unix,omitempty"` TxBytes uint64 `protobuf:"varint,2,opt,name=tx_bytes,json=txBytes,proto3" json:"tx_bytes,omitempty"` @@ -743,20 +799,20 @@ type WireGuardPeerStats struct { sizeCache protoimpl.SizeCache } -func (x *WireGuardPeerStats) Reset() { - *x = WireGuardPeerStats{} +func (x *TransportPeerStats) Reset() { + *x = TransportPeerStats{} mi := &file_protocol_nylon_ipc_proto_msgTypes[11] ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) ms.StoreMessageInfo(mi) } -func (x *WireGuardPeerStats) String() string { +func (x *TransportPeerStats) String() string { return protoimpl.X.MessageStringOf(x) } -func (*WireGuardPeerStats) ProtoMessage() {} +func (*TransportPeerStats) ProtoMessage() {} -func (x *WireGuardPeerStats) ProtoReflect() protoreflect.Message { +func (x *TransportPeerStats) ProtoReflect() protoreflect.Message { mi := &file_protocol_nylon_ipc_proto_msgTypes[11] if x != nil { ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) @@ -768,40 +824,40 @@ func (x *WireGuardPeerStats) ProtoReflect() protoreflect.Message { return mi.MessageOf(x) } -// Deprecated: Use WireGuardPeerStats.ProtoReflect.Descriptor instead. -func (*WireGuardPeerStats) Descriptor() ([]byte, []int) { +// Deprecated: Use TransportPeerStats.ProtoReflect.Descriptor instead. +func (*TransportPeerStats) Descriptor() ([]byte, []int) { return file_protocol_nylon_ipc_proto_rawDescGZIP(), []int{11} } -func (x *WireGuardPeerStats) GetLatestHandshakeUnix() int64 { +func (x *TransportPeerStats) GetLatestHandshakeUnix() int64 { if x != nil { return x.LatestHandshakeUnix } return 0 } -func (x *WireGuardPeerStats) GetTxBytes() uint64 { +func (x *TransportPeerStats) GetTxBytes() uint64 { if x != nil { return x.TxBytes } return 0 } -func (x *WireGuardPeerStats) GetRxBytes() uint64 { +func (x *TransportPeerStats) GetRxBytes() uint64 { if x != nil { return x.RxBytes } return 0 } -func (x *WireGuardPeerStats) GetPersistentKeepaliveInterval() uint32 { +func (x *TransportPeerStats) GetPersistentKeepaliveInterval() uint32 { if x != nil { return x.PersistentKeepaliveInterval } return 0 } -func (x *WireGuardPeerStats) GetEndpoint() string { +func (x *TransportPeerStats) GetEndpoint() string { if x != nil && x.Endpoint != nil { return *x.Endpoint } @@ -816,7 +872,7 @@ type NeighbourInfo struct { Endpoints []*EndpointInfo `protobuf:"bytes,4,rep,name=endpoints,proto3" json:"endpoints,omitempty"` Routes []*NeighRoute `protobuf:"bytes,5,rep,name=routes,proto3" json:"routes,omitempty"` Advertised []*Advertisement `protobuf:"bytes,6,rep,name=advertised,proto3" json:"advertised,omitempty"` - Wireguard *WireGuardPeerStats `protobuf:"bytes,7,opt,name=wireguard,proto3" json:"wireguard,omitempty"` + Transport *TransportPeerStats `protobuf:"bytes,7,opt,name=transport,proto3" json:"transport,omitempty"` unknownFields protoimpl.UnknownFields sizeCache protoimpl.SizeCache } @@ -893,9 +949,9 @@ func (x *NeighbourInfo) GetAdvertised() []*Advertisement { return nil } -func (x *NeighbourInfo) GetWireguard() *WireGuardPeerStats { +func (x *NeighbourInfo) GetTransport() *TransportPeerStats { if x != nil { - return x.Wireguard + return x.Transport } return nil } @@ -1592,14 +1648,60 @@ func (x *TraceEvent) GetLine() string { return "" } +type CancelRequest struct { + state protoimpl.MessageState `protogen:"open.v1"` + Id uint32 `protobuf:"varint,1,opt,name=id,proto3" json:"id,omitempty"` // the request to stop + unknownFields protoimpl.UnknownFields + sizeCache protoimpl.SizeCache +} + +func (x *CancelRequest) Reset() { + *x = CancelRequest{} + mi := &file_protocol_nylon_ipc_proto_msgTypes[24] + ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) + ms.StoreMessageInfo(mi) +} + +func (x *CancelRequest) String() string { + return protoimpl.X.MessageStringOf(x) +} + +func (*CancelRequest) ProtoMessage() {} + +func (x *CancelRequest) ProtoReflect() protoreflect.Message { + mi := &file_protocol_nylon_ipc_proto_msgTypes[24] + if x != nil { + ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) + if ms.LoadMessageInfo() == nil { + ms.StoreMessageInfo(mi) + } + return ms + } + return mi.MessageOf(x) +} + +// Deprecated: Use CancelRequest.ProtoReflect.Descriptor instead. +func (*CancelRequest) Descriptor() ([]byte, []int) { + return file_protocol_nylon_ipc_proto_rawDescGZIP(), []int{24} +} + +func (x *CancelRequest) GetId() uint32 { + if x != nil { + return x.Id + } + return 0 +} + type IpcRequest struct { state protoimpl.MessageState `protogen:"open.v1"` + Id uint32 `protobuf:"varint,1,opt,name=id,proto3" json:"id,omitempty"` // chosen by the client, echoed in each response // Types that are valid to be assigned to Request: // // *IpcRequest_Status // *IpcRequest_Probe // *IpcRequest_Reload // *IpcRequest_Trace + // *IpcRequest_Cancel Request isIpcRequest_Request `protobuf_oneof:"request"` unknownFields protoimpl.UnknownFields sizeCache protoimpl.SizeCache @@ -1607,7 +1709,7 @@ type IpcRequest struct { func (x *IpcRequest) Reset() { *x = IpcRequest{} - mi := &file_protocol_nylon_ipc_proto_msgTypes[24] + mi := &file_protocol_nylon_ipc_proto_msgTypes[25] ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) ms.StoreMessageInfo(mi) } @@ -1619,7 +1721,7 @@ func (x *IpcRequest) String() string { func (*IpcRequest) ProtoMessage() {} func (x *IpcRequest) ProtoReflect() protoreflect.Message { - mi := &file_protocol_nylon_ipc_proto_msgTypes[24] + mi := &file_protocol_nylon_ipc_proto_msgTypes[25] if x != nil { ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) if ms.LoadMessageInfo() == nil { @@ -1632,7 +1734,14 @@ func (x *IpcRequest) ProtoReflect() protoreflect.Message { // Deprecated: Use IpcRequest.ProtoReflect.Descriptor instead. func (*IpcRequest) Descriptor() ([]byte, []int) { - return file_protocol_nylon_ipc_proto_rawDescGZIP(), []int{24} + return file_protocol_nylon_ipc_proto_rawDescGZIP(), []int{25} +} + +func (x *IpcRequest) GetId() uint32 { + if x != nil { + return x.Id + } + return 0 } func (x *IpcRequest) GetRequest() isIpcRequest_Request { @@ -1678,24 +1787,37 @@ func (x *IpcRequest) GetTrace() *TraceRequest { return nil } +func (x *IpcRequest) GetCancel() *CancelRequest { + if x != nil { + if x, ok := x.Request.(*IpcRequest_Cancel); ok { + return x.Cancel + } + } + return nil +} + type isIpcRequest_Request interface { isIpcRequest_Request() } type IpcRequest_Status struct { - Status *StatusRequest `protobuf:"bytes,1,opt,name=status,proto3,oneof"` + Status *StatusRequest `protobuf:"bytes,2,opt,name=status,proto3,oneof"` } type IpcRequest_Probe struct { - Probe *ProbeRequest `protobuf:"bytes,2,opt,name=probe,proto3,oneof"` + Probe *ProbeRequest `protobuf:"bytes,3,opt,name=probe,proto3,oneof"` } type IpcRequest_Reload struct { - Reload *ReloadRequest `protobuf:"bytes,3,opt,name=reload,proto3,oneof"` + Reload *ReloadRequest `protobuf:"bytes,4,opt,name=reload,proto3,oneof"` } type IpcRequest_Trace struct { - Trace *TraceRequest `protobuf:"bytes,4,opt,name=trace,proto3,oneof"` + Trace *TraceRequest `protobuf:"bytes,5,opt,name=trace,proto3,oneof"` +} + +type IpcRequest_Cancel struct { + Cancel *CancelRequest `protobuf:"bytes,6,opt,name=cancel,proto3,oneof"` } func (*IpcRequest_Status) isIpcRequest_Request() {} @@ -1706,10 +1828,64 @@ func (*IpcRequest_Reload) isIpcRequest_Request() {} func (*IpcRequest_Trace) isIpcRequest_Request() {} +func (*IpcRequest_Cancel) isIpcRequest_Request() {} + +type IpcError struct { + state protoimpl.MessageState `protogen:"open.v1"` + Code IpcError_Code `protobuf:"varint,1,opt,name=code,proto3,enum=proto.IpcError_Code" json:"code,omitempty"` + Message string `protobuf:"bytes,2,opt,name=message,proto3" json:"message,omitempty"` + unknownFields protoimpl.UnknownFields + sizeCache protoimpl.SizeCache +} + +func (x *IpcError) Reset() { + *x = IpcError{} + mi := &file_protocol_nylon_ipc_proto_msgTypes[26] + ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) + ms.StoreMessageInfo(mi) +} + +func (x *IpcError) String() string { + return protoimpl.X.MessageStringOf(x) +} + +func (*IpcError) ProtoMessage() {} + +func (x *IpcError) ProtoReflect() protoreflect.Message { + mi := &file_protocol_nylon_ipc_proto_msgTypes[26] + if x != nil { + ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) + if ms.LoadMessageInfo() == nil { + ms.StoreMessageInfo(mi) + } + return ms + } + return mi.MessageOf(x) +} + +// Deprecated: Use IpcError.ProtoReflect.Descriptor instead. +func (*IpcError) Descriptor() ([]byte, []int) { + return file_protocol_nylon_ipc_proto_rawDescGZIP(), []int{26} +} + +func (x *IpcError) GetCode() IpcError_Code { + if x != nil { + return x.Code + } + return IpcError_INTERNAL +} + +func (x *IpcError) GetMessage() string { + if x != nil { + return x.Message + } + return "" +} + type IpcResponse struct { state protoimpl.MessageState `protogen:"open.v1"` - Ok bool `protobuf:"varint,1,opt,name=ok,proto3" json:"ok,omitempty"` - Error string `protobuf:"bytes,2,opt,name=error,proto3" json:"error,omitempty"` + Id uint32 `protobuf:"varint,1,opt,name=id,proto3" json:"id,omitempty"` + Error *IpcError `protobuf:"bytes,2,opt,name=error,proto3" json:"error,omitempty"` // set when the request failed // Types that are valid to be assigned to Response: // // *IpcResponse_Status @@ -1723,7 +1899,7 @@ type IpcResponse struct { func (x *IpcResponse) Reset() { *x = IpcResponse{} - mi := &file_protocol_nylon_ipc_proto_msgTypes[25] + mi := &file_protocol_nylon_ipc_proto_msgTypes[27] ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) ms.StoreMessageInfo(mi) } @@ -1735,7 +1911,7 @@ func (x *IpcResponse) String() string { func (*IpcResponse) ProtoMessage() {} func (x *IpcResponse) ProtoReflect() protoreflect.Message { - mi := &file_protocol_nylon_ipc_proto_msgTypes[25] + mi := &file_protocol_nylon_ipc_proto_msgTypes[27] if x != nil { ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) if ms.LoadMessageInfo() == nil { @@ -1748,21 +1924,21 @@ func (x *IpcResponse) ProtoReflect() protoreflect.Message { // Deprecated: Use IpcResponse.ProtoReflect.Descriptor instead. func (*IpcResponse) Descriptor() ([]byte, []int) { - return file_protocol_nylon_ipc_proto_rawDescGZIP(), []int{25} + return file_protocol_nylon_ipc_proto_rawDescGZIP(), []int{27} } -func (x *IpcResponse) GetOk() bool { +func (x *IpcResponse) GetId() uint32 { if x != nil { - return x.Ok + return x.Id } - return false + return 0 } -func (x *IpcResponse) GetError() string { +func (x *IpcResponse) GetError() *IpcError { if x != nil { return x.Error } - return "" + return nil } func (x *IpcResponse) GetResponse() isIpcResponse_Response { @@ -1883,7 +2059,7 @@ const file_protocol_nylon_ipc_proto_rawDesc = "" + "\x0ffiltered_rtt_ns\x18\a \x01(\x03R\rfilteredRttNs\x12*\n" + "\x11stabilized_rtt_ns\x18\b \x01(\x03R\x0fstabilizedRttNsB\v\n" + "\t_resolved\"\xf0\x01\n" + - "\x12WireGuardPeerStats\x122\n" + + "\x12TransportPeerStats\x122\n" + "\x15latest_handshake_unix\x18\x01 \x01(\x03R\x13latestHandshakeUnix\x12\x19\n" + "\btx_bytes\x18\x02 \x01(\x04R\atxBytes\x12\x19\n" + "\brx_bytes\x18\x03 \x01(\x04R\arxBytes\x12B\n" + @@ -1900,7 +2076,7 @@ const file_protocol_nylon_ipc_proto_rawDesc = "" + "\n" + "advertised\x18\x06 \x03(\v2\x14.proto.AdvertisementR\n" + "advertised\x127\n" + - "\twireguard\x18\a \x01(\v2\x19.proto.WireGuardPeerStatsR\twireguard\"W\n" + + "\ttransport\x18\a \x01(\v2\x19.proto.TransportPeerStatsR\ttransport\"W\n" + "\x0fRouteTableEntry\x12\x16\n" + "\x06prefix\x18\x01 \x01(\tR\x06prefix\x12\x0e\n" + "\x02nh\x18\x02 \x01(\tR\x02nh\x12\x1c\n" + @@ -1959,17 +2135,30 @@ const file_protocol_nylon_ipc_proto_rawDesc = "" + "\amessage\x18\x02 \x01(\tR\amessage\" \n" + "\n" + "TraceEvent\x12\x12\n" + - "\x04line\x18\x01 \x01(\tR\x04line\"\xd1\x01\n" + + "\x04line\x18\x01 \x01(\tR\x04line\"\x1f\n" + + "\rCancelRequest\x12\x0e\n" + + "\x02id\x18\x01 \x01(\rR\x02id\"\x91\x02\n" + "\n" + - "IpcRequest\x12.\n" + - "\x06status\x18\x01 \x01(\v2\x14.proto.StatusRequestH\x00R\x06status\x12+\n" + - "\x05probe\x18\x02 \x01(\v2\x13.proto.ProbeRequestH\x00R\x05probe\x12.\n" + - "\x06reload\x18\x03 \x01(\v2\x14.proto.ReloadRequestH\x00R\x06reload\x12+\n" + - "\x05trace\x18\x04 \x01(\v2\x13.proto.TraceRequestH\x00R\x05traceB\t\n" + - "\arequest\"\xfa\x01\n" + + "IpcRequest\x12\x0e\n" + + "\x02id\x18\x01 \x01(\rR\x02id\x12.\n" + + "\x06status\x18\x02 \x01(\v2\x14.proto.StatusRequestH\x00R\x06status\x12+\n" + + "\x05probe\x18\x03 \x01(\v2\x13.proto.ProbeRequestH\x00R\x05probe\x12.\n" + + "\x06reload\x18\x04 \x01(\v2\x14.proto.ReloadRequestH\x00R\x06reload\x12+\n" + + "\x05trace\x18\x05 \x01(\v2\x13.proto.TraceRequestH\x00R\x05trace\x12.\n" + + "\x06cancel\x18\x06 \x01(\v2\x14.proto.CancelRequestH\x00R\x06cancelB\t\n" + + "\arequest\"\xb2\x01\n" + + "\bIpcError\x12(\n" + + "\x04code\x18\x01 \x01(\x0e2\x14.proto.IpcError.CodeR\x04code\x12\x18\n" + + "\amessage\x18\x02 \x01(\tR\amessage\"b\n" + + "\x04Code\x12\f\n" + + "\bINTERNAL\x10\x00\x12\x13\n" + + "\x0fINVALID_REQUEST\x10\x01\x12\x0f\n" + + "\vUNAVAILABLE\x10\x02\x12\x17\n" + + "\x13FAILED_PRECONDITION\x10\x03\x12\r\n" + + "\tNOT_FOUND\x10\x04\"\x8b\x02\n" + "\vIpcResponse\x12\x0e\n" + - "\x02ok\x18\x01 \x01(\bR\x02ok\x12\x14\n" + - "\x05error\x18\x02 \x01(\tR\x05error\x12/\n" + + "\x02id\x18\x01 \x01(\rR\x02id\x12%\n" + + "\x05error\x18\x02 \x01(\v2\x0f.proto.IpcErrorR\x05error\x12/\n" + "\x06status\x18\x03 \x01(\v2\x15.proto.StatusResponseH\x00R\x06status\x12,\n" + "\x05probe\x18\x04 \x01(\v2\x14.proto.ProbeResponseH\x00R\x05probe\x12/\n" + "\x06reload\x18\x05 \x01(\v2\x15.proto.ReloadResponseH\x00R\x06reload\x12)\n" + @@ -2000,75 +2189,81 @@ func file_protocol_nylon_ipc_proto_rawDescGZIP() []byte { return file_protocol_nylon_ipc_proto_rawDescData } -var file_protocol_nylon_ipc_proto_enumTypes = make([]protoimpl.EnumInfo, 2) -var file_protocol_nylon_ipc_proto_msgTypes = make([]protoimpl.MessageInfo, 26) +var file_protocol_nylon_ipc_proto_enumTypes = make([]protoimpl.EnumInfo, 3) +var file_protocol_nylon_ipc_proto_msgTypes = make([]protoimpl.MessageInfo, 28) var file_protocol_nylon_ipc_proto_goTypes = []any{ (ReloadResult)(0), // 0: proto.ReloadResult (EndpointProbeStatus)(0), // 1: proto.EndpointProbeStatus - (*StatusRequest)(nil), // 2: proto.StatusRequest - (*ProbeRequest)(nil), // 3: proto.ProbeRequest - (*ReloadRequest)(nil), // 4: proto.ReloadRequest - (*TraceRequest)(nil), // 5: proto.TraceRequest - (*Source)(nil), // 6: proto.Source - (*FD)(nil), // 7: proto.FD - (*PubRoute)(nil), // 8: proto.PubRoute - (*NeighRoute)(nil), // 9: proto.NeighRoute - (*SelRoute)(nil), // 10: proto.SelRoute - (*Advertisement)(nil), // 11: proto.Advertisement - (*EndpointInfo)(nil), // 12: proto.EndpointInfo - (*WireGuardPeerStats)(nil), // 13: proto.WireGuardPeerStats - (*NeighbourInfo)(nil), // 14: proto.NeighbourInfo - (*RouteTableEntry)(nil), // 15: proto.RouteTableEntry - (*RouteTables)(nil), // 16: proto.RouteTables - (*SeqnoEntry)(nil), // 17: proto.SeqnoEntry - (*FeasibilityDistance)(nil), // 18: proto.FeasibilityDistance - (*NodeStats)(nil), // 19: proto.NodeStats - (*NodeStatus)(nil), // 20: proto.NodeStatus - (*StatusResponse)(nil), // 21: proto.StatusResponse - (*EndpointProbeResult)(nil), // 22: proto.EndpointProbeResult - (*ProbeResponse)(nil), // 23: proto.ProbeResponse - (*ReloadResponse)(nil), // 24: proto.ReloadResponse - (*TraceEvent)(nil), // 25: proto.TraceEvent - (*IpcRequest)(nil), // 26: proto.IpcRequest - (*IpcResponse)(nil), // 27: proto.IpcResponse + (IpcError_Code)(0), // 2: proto.IpcError.Code + (*StatusRequest)(nil), // 3: proto.StatusRequest + (*ProbeRequest)(nil), // 4: proto.ProbeRequest + (*ReloadRequest)(nil), // 5: proto.ReloadRequest + (*TraceRequest)(nil), // 6: proto.TraceRequest + (*Source)(nil), // 7: proto.Source + (*FD)(nil), // 8: proto.FD + (*PubRoute)(nil), // 9: proto.PubRoute + (*NeighRoute)(nil), // 10: proto.NeighRoute + (*SelRoute)(nil), // 11: proto.SelRoute + (*Advertisement)(nil), // 12: proto.Advertisement + (*EndpointInfo)(nil), // 13: proto.EndpointInfo + (*TransportPeerStats)(nil), // 14: proto.TransportPeerStats + (*NeighbourInfo)(nil), // 15: proto.NeighbourInfo + (*RouteTableEntry)(nil), // 16: proto.RouteTableEntry + (*RouteTables)(nil), // 17: proto.RouteTables + (*SeqnoEntry)(nil), // 18: proto.SeqnoEntry + (*FeasibilityDistance)(nil), // 19: proto.FeasibilityDistance + (*NodeStats)(nil), // 20: proto.NodeStats + (*NodeStatus)(nil), // 21: proto.NodeStatus + (*StatusResponse)(nil), // 22: proto.StatusResponse + (*EndpointProbeResult)(nil), // 23: proto.EndpointProbeResult + (*ProbeResponse)(nil), // 24: proto.ProbeResponse + (*ReloadResponse)(nil), // 25: proto.ReloadResponse + (*TraceEvent)(nil), // 26: proto.TraceEvent + (*CancelRequest)(nil), // 27: proto.CancelRequest + (*IpcRequest)(nil), // 28: proto.IpcRequest + (*IpcError)(nil), // 29: proto.IpcError + (*IpcResponse)(nil), // 30: proto.IpcResponse } var file_protocol_nylon_ipc_proto_depIdxs = []int32{ - 6, // 0: proto.PubRoute.source:type_name -> proto.Source - 7, // 1: proto.PubRoute.fd:type_name -> proto.FD - 8, // 2: proto.NeighRoute.pub_route:type_name -> proto.PubRoute - 8, // 3: proto.SelRoute.pub_route:type_name -> proto.PubRoute - 12, // 4: proto.NeighbourInfo.endpoints:type_name -> proto.EndpointInfo - 9, // 5: proto.NeighbourInfo.routes:type_name -> proto.NeighRoute - 11, // 6: proto.NeighbourInfo.advertised:type_name -> proto.Advertisement - 13, // 7: proto.NeighbourInfo.wireguard:type_name -> proto.WireGuardPeerStats - 10, // 8: proto.RouteTables.selected:type_name -> proto.SelRoute - 15, // 9: proto.RouteTables.forward:type_name -> proto.RouteTableEntry - 15, // 10: proto.RouteTables.exit:type_name -> proto.RouteTableEntry - 6, // 11: proto.FeasibilityDistance.source:type_name -> proto.Source - 7, // 12: proto.FeasibilityDistance.fd:type_name -> proto.FD - 11, // 13: proto.NodeStatus.advertised:type_name -> proto.Advertisement - 17, // 14: proto.NodeStatus.seqnos:type_name -> proto.SeqnoEntry - 19, // 15: proto.NodeStatus.stats:type_name -> proto.NodeStats - 20, // 16: proto.StatusResponse.node:type_name -> proto.NodeStatus - 14, // 17: proto.StatusResponse.neighbours:type_name -> proto.NeighbourInfo - 16, // 18: proto.StatusResponse.routes:type_name -> proto.RouteTables - 18, // 19: proto.StatusResponse.feasibility_distances:type_name -> proto.FeasibilityDistance + 7, // 0: proto.PubRoute.source:type_name -> proto.Source + 8, // 1: proto.PubRoute.fd:type_name -> proto.FD + 9, // 2: proto.NeighRoute.pub_route:type_name -> proto.PubRoute + 9, // 3: proto.SelRoute.pub_route:type_name -> proto.PubRoute + 13, // 4: proto.NeighbourInfo.endpoints:type_name -> proto.EndpointInfo + 10, // 5: proto.NeighbourInfo.routes:type_name -> proto.NeighRoute + 12, // 6: proto.NeighbourInfo.advertised:type_name -> proto.Advertisement + 14, // 7: proto.NeighbourInfo.transport:type_name -> proto.TransportPeerStats + 11, // 8: proto.RouteTables.selected:type_name -> proto.SelRoute + 16, // 9: proto.RouteTables.forward:type_name -> proto.RouteTableEntry + 16, // 10: proto.RouteTables.exit:type_name -> proto.RouteTableEntry + 7, // 11: proto.FeasibilityDistance.source:type_name -> proto.Source + 8, // 12: proto.FeasibilityDistance.fd:type_name -> proto.FD + 12, // 13: proto.NodeStatus.advertised:type_name -> proto.Advertisement + 18, // 14: proto.NodeStatus.seqnos:type_name -> proto.SeqnoEntry + 20, // 15: proto.NodeStatus.stats:type_name -> proto.NodeStats + 21, // 16: proto.StatusResponse.node:type_name -> proto.NodeStatus + 15, // 17: proto.StatusResponse.neighbours:type_name -> proto.NeighbourInfo + 17, // 18: proto.StatusResponse.routes:type_name -> proto.RouteTables + 19, // 19: proto.StatusResponse.feasibility_distances:type_name -> proto.FeasibilityDistance 1, // 20: proto.EndpointProbeResult.status:type_name -> proto.EndpointProbeStatus - 22, // 21: proto.ProbeResponse.results:type_name -> proto.EndpointProbeResult + 23, // 21: proto.ProbeResponse.results:type_name -> proto.EndpointProbeResult 0, // 22: proto.ReloadResponse.result:type_name -> proto.ReloadResult - 2, // 23: proto.IpcRequest.status:type_name -> proto.StatusRequest - 3, // 24: proto.IpcRequest.probe:type_name -> proto.ProbeRequest - 4, // 25: proto.IpcRequest.reload:type_name -> proto.ReloadRequest - 5, // 26: proto.IpcRequest.trace:type_name -> proto.TraceRequest - 21, // 27: proto.IpcResponse.status:type_name -> proto.StatusResponse - 23, // 28: proto.IpcResponse.probe:type_name -> proto.ProbeResponse - 24, // 29: proto.IpcResponse.reload:type_name -> proto.ReloadResponse - 25, // 30: proto.IpcResponse.trace:type_name -> proto.TraceEvent - 31, // [31:31] is the sub-list for method output_type - 31, // [31:31] is the sub-list for method input_type - 31, // [31:31] is the sub-list for extension type_name - 31, // [31:31] is the sub-list for extension extendee - 0, // [0:31] is the sub-list for field type_name + 3, // 23: proto.IpcRequest.status:type_name -> proto.StatusRequest + 4, // 24: proto.IpcRequest.probe:type_name -> proto.ProbeRequest + 5, // 25: proto.IpcRequest.reload:type_name -> proto.ReloadRequest + 6, // 26: proto.IpcRequest.trace:type_name -> proto.TraceRequest + 27, // 27: proto.IpcRequest.cancel:type_name -> proto.CancelRequest + 2, // 28: proto.IpcError.code:type_name -> proto.IpcError.Code + 29, // 29: proto.IpcResponse.error:type_name -> proto.IpcError + 22, // 30: proto.IpcResponse.status:type_name -> proto.StatusResponse + 24, // 31: proto.IpcResponse.probe:type_name -> proto.ProbeResponse + 25, // 32: proto.IpcResponse.reload:type_name -> proto.ReloadResponse + 26, // 33: proto.IpcResponse.trace:type_name -> proto.TraceEvent + 34, // [34:34] is the sub-list for method output_type + 34, // [34:34] is the sub-list for method input_type + 34, // [34:34] is the sub-list for extension type_name + 34, // [34:34] is the sub-list for extension extendee + 0, // [0:34] is the sub-list for field type_name } func init() { file_protocol_nylon_ipc_proto_init() } @@ -2079,13 +2274,14 @@ func file_protocol_nylon_ipc_proto_init() { file_protocol_nylon_ipc_proto_msgTypes[10].OneofWrappers = []any{} file_protocol_nylon_ipc_proto_msgTypes[11].OneofWrappers = []any{} file_protocol_nylon_ipc_proto_msgTypes[20].OneofWrappers = []any{} - file_protocol_nylon_ipc_proto_msgTypes[24].OneofWrappers = []any{ + file_protocol_nylon_ipc_proto_msgTypes[25].OneofWrappers = []any{ (*IpcRequest_Status)(nil), (*IpcRequest_Probe)(nil), (*IpcRequest_Reload)(nil), (*IpcRequest_Trace)(nil), + (*IpcRequest_Cancel)(nil), } - file_protocol_nylon_ipc_proto_msgTypes[25].OneofWrappers = []any{ + file_protocol_nylon_ipc_proto_msgTypes[27].OneofWrappers = []any{ (*IpcResponse_Status)(nil), (*IpcResponse_Probe)(nil), (*IpcResponse_Reload)(nil), @@ -2096,8 +2292,8 @@ func file_protocol_nylon_ipc_proto_init() { File: protoimpl.DescBuilder{ GoPackagePath: reflect.TypeOf(x{}).PkgPath(), RawDescriptor: unsafe.Slice(unsafe.StringData(file_protocol_nylon_ipc_proto_rawDesc), len(file_protocol_nylon_ipc_proto_rawDesc)), - NumEnums: 2, - NumMessages: 26, + NumEnums: 3, + NumMessages: 28, NumExtensions: 0, NumServices: 0, }, diff --git a/protocol/nylon_ipc.proto b/protocol/nylon_ipc.proto index 45ba8551..04ed9107 100644 --- a/protocol/nylon_ipc.proto +++ b/protocol/nylon_ipc.proto @@ -66,7 +66,8 @@ message EndpointInfo { int64 stabilized_rtt_ns = 8; } -message WireGuardPeerStats { +// What the transport reports about a peer. Byte counts include transport headers. +message TransportPeerStats { int64 latest_handshake_unix = 1; uint64 tx_bytes = 2; uint64 rx_bytes = 3; @@ -81,7 +82,7 @@ message NeighbourInfo { repeated EndpointInfo endpoints = 4; repeated NeighRoute routes = 5; repeated Advertisement advertised = 6; - WireGuardPeerStats wireguard = 7; + TransportPeerStats transport = 7; } message RouteTableEntry { @@ -162,18 +163,47 @@ message TraceEvent { string line = 1; } +// Wire format, on the nylon socket: /var/run/nylon/.sock, or on Windows +// \\.\pipe\ProtectedPrefix\Administrators\nylon\. +// +// Both sides first send a hello: the 4 bytes "NYLN" and a big-endian uint16 protocol version. A server that doesn't +// speak the client's version sends its own hello, then closes the connection. +// +// After that, every message is a frame: a big-endian uint32 length, then that many bytes of binary protobuf. Clients +// send IpcRequest frames; the server answers each with IpcResponse frames carrying the request's id. Requests on one +// connection run concurrently, so responses can arrive in any order. A trace request gets a response per event until +// it is cancelled or the connection closes. + +message CancelRequest { + uint32 id = 1; // the request to stop +} + message IpcRequest { + uint32 id = 1; // chosen by the client, echoed in each response oneof request { - StatusRequest status = 1; - ProbeRequest probe = 2; - ReloadRequest reload = 3; - TraceRequest trace = 4; + StatusRequest status = 2; + ProbeRequest probe = 3; + ReloadRequest reload = 4; + TraceRequest trace = 5; + CancelRequest cancel = 6; } } +message IpcError { + enum Code { + INTERNAL = 0; + INVALID_REQUEST = 1; + UNAVAILABLE = 2; // nylon is busy or shutting down; worth retrying + FAILED_PRECONDITION = 3; // e.g. trace without --dbg-trace-tc + NOT_FOUND = 4; + } + Code code = 1; + string message = 2; +} + message IpcResponse { - bool ok = 1; - string error = 2; + uint32 id = 1; + IpcError error = 2; // set when the request failed oneof response { StatusResponse status = 3; ProbeResponse probe = 4;