From 08ad260606bc2aa101d2517d30e7d9e2b2bb470f Mon Sep 17 00:00:00 2001 From: ginuerzh Date: Sat, 8 Feb 2025 23:07:54 +0800 Subject: [PATCH] add entrypoint for router handler --- connector/router/connector.go | 1 + go.mod | 4 +- go.sum | 4 + handler/router/associate.go | 189 ++++++++++++++++++++-------------- handler/router/entrypoint.go | 62 +++++++++++ handler/router/handler.go | 59 +++++++---- handler/router/metadata.go | 2 +- handler/tun/server.go | 2 +- handler/tunnel/dialer.go | 2 +- listener/tun/tun_linux.go | 6 +- registry/router.go | 3 +- router/plugin/grpc.go | 13 ++- router/plugin/http.go | 21 +++- router/router.go | 13 +-- 14 files changed, 259 insertions(+), 122 deletions(-) create mode 100644 handler/router/entrypoint.go diff --git a/connector/router/connector.go b/connector/router/connector.go index ad21fc36..c6fa9eda 100644 --- a/connector/router/connector.go +++ b/connector/router/connector.go @@ -113,5 +113,6 @@ func (c *routerConnector) Connect(ctx context.Context, conn net.Conn, network, a return nil, err } + log.Debugf("connect to router %s@%s OK", address, c.md.routerID) return conn, nil } diff --git a/go.mod b/go.mod index 5cea2329..4af3ff01 100644 --- a/go.mod +++ b/go.mod @@ -9,10 +9,10 @@ require ( github.com/asaskevich/govalidator v0.0.0-20210307081110-f21760c49a8d github.com/gin-contrib/cors v1.7.2 github.com/gin-gonic/gin v1.10.0 - github.com/go-gost/core v0.3.0 + github.com/go-gost/core v0.3.1 github.com/go-gost/gosocks4 v0.0.1 github.com/go-gost/gosocks5 v0.4.2 - github.com/go-gost/plugin v0.1.1 + github.com/go-gost/plugin v0.2.0 github.com/go-gost/relay v0.5.0 github.com/go-gost/tls-dissector v0.1.1 github.com/go-redis/redis/v8 v8.11.5 diff --git a/go.sum b/go.sum index c24c0a01..4c235ba8 100644 --- a/go.sum +++ b/go.sum @@ -55,12 +55,16 @@ github.com/go-gost/core v0.2.1 h1:oNTeKRcuTYowzWIBEgd/YtDocxzyvsqgWpnn/Q3vuaw= github.com/go-gost/core v0.2.1/go.mod h1:WGI43jOka7FAsSAwi/fSMaqxdR+E339ycb4NBGlFr6A= github.com/go-gost/core v0.3.0 h1:iGiX8aOIC3CmX4en8l2uxcT9+vqhd/7L9C7KjRR2zI0= github.com/go-gost/core v0.3.0/go.mod h1:WGI43jOka7FAsSAwi/fSMaqxdR+E339ycb4NBGlFr6A= +github.com/go-gost/core v0.3.1 h1:Fo1WvyUTlxSF36d31aRC2LE9UPBDQ4hTMv8oQWoP2wE= +github.com/go-gost/core v0.3.1/go.mod h1:WGI43jOka7FAsSAwi/fSMaqxdR+E339ycb4NBGlFr6A= github.com/go-gost/gosocks4 v0.0.1 h1:+k1sec8HlELuQV7rWftIkmy8UijzUt2I6t+iMPlGB2s= github.com/go-gost/gosocks4 v0.0.1/go.mod h1:3B6L47HbU/qugDg4JnoFPHgJXE43Inz8Bah1QaN9qCc= github.com/go-gost/gosocks5 v0.4.2 h1:IianxHTkACPqCwiOAT3MHoMdSUl+SEPSRu1ikawC1Pc= github.com/go-gost/gosocks5 v0.4.2/go.mod h1:1G6I7HP7VFVxveGkoK8mnprnJqSqJjdcASKsdUn4Pp4= github.com/go-gost/plugin v0.1.1 h1:LNoc/Rqwb3ceGhhxwhpjf1SeeYBGe80MJ9E4lpM3qak= github.com/go-gost/plugin v0.1.1/go.mod h1:oN23l+yGDCIP9G3KnDl/I/0zVGOobZUDCB2Z5yYYXts= +github.com/go-gost/plugin v0.2.0 h1:Xxqq3oJi6d2dZ/PqqEpefrUckT6aVz8JUxQ/sb7tmTM= +github.com/go-gost/plugin v0.2.0/go.mod h1:oN23l+yGDCIP9G3KnDl/I/0zVGOobZUDCB2Z5yYYXts= github.com/go-gost/relay v0.5.0 h1:JG1tgy/KWiVXS0ukuVXvbM0kbYuJTWxYpJ5JwzsCf/c= github.com/go-gost/relay v0.5.0/go.mod h1:lcX+23LCQ3khIeASBo+tJ/WbwXFO32/N5YN6ucuYTG8= github.com/go-gost/tls-dissector v0.1.1 h1:2zUOTPzCQAUQ54Rpy0UEi3JPMQSYsIFSeFeKrzmkCoU= diff --git a/handler/router/associate.go b/handler/router/associate.go index 9ba54dc7..ddc7b357 100644 --- a/handler/router/associate.go +++ b/handler/router/associate.go @@ -1,6 +1,7 @@ package router import ( + "bytes" "context" "encoding/binary" "errors" @@ -97,7 +98,7 @@ func (h *routerHandler) handleAssociate(ctx context.Context, conn net.Conn, netw ID: connectorID.String(), Name: clientID, Node: h.id, - Network: network, + Network: "udp", Address: h.md.entryPoint, }) if err != nil { @@ -113,91 +114,127 @@ func (h *routerHandler) handleAssociate(ctx context.Context, conn net.Conn, netw log.Debugf("%s/%s: router=%s, connector=%s, weight=%d established", host, network, routerID, connectorID, connectorID.Weight()) - return h.routing(ctx, conn, routerID, log) -} - -func (h *routerHandler) routing(ctx context.Context, conn net.Conn, routerID relay.TunnelID, log logger.Logger) error { - b := bufpool.Get(h.md.bufferSize) - defer bufpool.Put(b) - - rid := routerID.String() - for { - err := func() error { - n, err := conn.Read(b) - if err != nil { - return err - } + b := bufpool.Get(h.md.bufferSize) - var dstIP net.IP - - if waterutil.IsIPv4(b[:n]) { - header, err := ipv4.ParseHeader(b[:n]) - if err != nil { - log.Warn(err) - return nil - } - - dstIP = header.Dst - - if log.IsLevelEnabled(logger.TraceLevel) { - log.Tracef("%s >> %s %-4s %d/%-4d %-4x %d", - header.Src, header.Dst, xip.Protocol(waterutil.IPv4Protocol(b[:n])), - header.Len, header.TotalLen, header.ID, header.Flags) - } - } else if waterutil.IsIPv6(b[:n]) { - header, err := ipv6.ParseHeader(b[:n]) - if err != nil { - log.Warn(err) - return nil - } - - dstIP = header.Dst - - if log.IsLevelEnabled(logger.TraceLevel) { - log.Tracef("%s >> %s %s %d %d", - header.Src, header.Dst, - xip.Protocol(waterutil.IPProtocol(header.NextHeader)), - header.PayloadLen, header.TrafficClass) - } - } else { - log.Warnf("unknown packet, discarded(%d)", n) - return nil - } - - var route *router.Route - if r := registry.RouterRegistry().Get(rid); r != nil { - route = r.GetRoute(ctx, dstIP, router.IDOption(rid)) - } - if route == nil && h.md.router != nil { - route = h.md.router.GetRoute(ctx, dstIP, router.IDOption(rid)) - } - if route == nil || route.Gateway == "" { - // no route to host, discard - return nil - } - - if log.IsLevelEnabled(logger.TraceLevel) { - log.Tracef("route for %s: %s -> %s", dstIP, route.Dst, route.Gateway) - } - - if c := h.pool.Get(routerID, route.Gateway); c != nil { - if w := c.Writer(); w != nil { - w.Write(b[:n]) - } - } - - return nil - }() + n, err := conn.Read(b) if err != nil { if err == io.EOF { - err = nil + return nil } return err } + + h.handlePacket(ctx, b[:n], routerID, log) } } +func (h *routerHandler) handlePacket(ctx context.Context, data []byte, routerID relay.TunnelID, log logger.Logger) error { + defer bufpool.Put(data) + + var dstIP net.IP + if waterutil.IsIPv4(data) { + header, err := ipv4.ParseHeader(data) + if err != nil { + return err + } + + dstIP = header.Dst + + if log.IsLevelEnabled(logger.TraceLevel) { + log.Tracef("%s >> %s %-4s %d/%-4d %-4x %d", + header.Src, header.Dst, xip.Protocol(waterutil.IPv4Protocol(data)), + header.Len, header.TotalLen, header.ID, header.Flags) + } + } else if waterutil.IsIPv6(data) { + header, err := ipv6.ParseHeader(data) + if err != nil { + return err + } + + dstIP = header.Dst + + if log.IsLevelEnabled(logger.TraceLevel) { + log.Tracef("%s >> %s %s %d %d", + header.Src, header.Dst, + xip.Protocol(waterutil.IPProtocol(header.NextHeader)), + header.PayloadLen, header.TrafficClass) + } + } else { + return fmt.Errorf("unknown packet, discarded(%d)", len(data)) + } + + rid := routerID.String() + + var route *router.Route + if r := registry.RouterRegistry().Get(rid); r != nil { + route = r.GetRoute(ctx, dstIP.String(), router.IDOption(rid)) + } + if route == nil && h.md.router != nil { + route = h.md.router.GetRoute(ctx, dstIP.String(), router.IDOption(rid)) + } + if route == nil || route.Gateway == "" { + // no route to host, discard + return fmt.Errorf("route to host %s", dstIP) + } + + if log.IsLevelEnabled(logger.TraceLevel) { + log.Tracef("route for %s: %s -> %s", dstIP, route.Dst, route.Gateway) + } + + if c := h.pool.Get(routerID, route.Gateway); c != nil { + if w := c.Writer(); w != nil { + w.Write(data) + } + return nil + } + + if h.md.sd == nil { + return nil + } + + clientID := fmt.Sprintf("%s@%s", route.Gateway, routerID) + ss, _ := h.md.sd.Get(ctx, clientID) + + var service *sd.Service + for _, s := range ss { + if s.Node != h.id { + service = s + break + } + } + if service == nil { + return nil + } + + raddr, _ := net.ResolveUDPAddr("udp", service.Address) + if raddr == nil { + return nil + } + + req := &relay.Request{ + Version: relay.Version1, + Cmd: relay.CmdAssociate, + Features: []relay.Feature{ + &relay.TunnelFeature{ + ID: routerID, + }, + &relay.AddrFeature{ + AType: relay.AddrDomain, + Host: route.Gateway, + }, + }, + } + + buf := bytes.Buffer{} + req.WriteTo(&buf) + buf.Write(data) + + h.epConn.WriteTo(buf.Bytes(), raddr) + + return nil +} + type packetConn struct { net.Conn } diff --git a/handler/router/entrypoint.go b/handler/router/entrypoint.go new file mode 100644 index 00000000..8b794a4b --- /dev/null +++ b/handler/router/entrypoint.go @@ -0,0 +1,62 @@ +package router + +import ( + "bytes" + + "github.com/go-gost/core/common/bufpool" + "github.com/go-gost/core/logger" + "github.com/go-gost/relay" +) + +func (h *routerHandler) handleEntrypoint(log logger.Logger) error { + buf := bufpool.Get(h.md.bufferSize) + defer bufpool.Put(buf) + + for { + err := func() error { + n, addr, err := h.epConn.ReadFrom(buf) + if err != nil { + return err + } + + req := relay.Request{} + nn, err := req.ReadFrom(bytes.NewReader(buf[:n])) + if err != nil { + return nil + } + if req.Cmd != relay.CmdAssociate { + return nil + } + + var routerID relay.TunnelID + var gateway string + + for _, f := range req.Features { + switch f.Type() { + case relay.FeatureTunnel: + if feature, _ := f.(*relay.TunnelFeature); feature != nil { + routerID = feature.ID + } + case relay.FeatureAddr: + if feature, _ := f.(*relay.AddrFeature); feature != nil { + gateway = feature.Host + } + } + } + + log.Tracef("redirect from %s to %s@%s", addr, gateway, routerID) + + if c := h.pool.Get(routerID, gateway); c != nil { + if w := c.Writer(); w != nil { + w.Write(buf[nn:]) + } + } + + return nil + }() + + if err != nil { + return err + } + } +} diff --git a/handler/router/handler.go b/handler/router/handler.go index 5a5ee5a2..91e0d2cd 100644 --- a/handler/router/handler.go +++ b/handler/router/handler.go @@ -3,6 +3,7 @@ package router import ( "context" "errors" + "fmt" "net" "strconv" "time" @@ -13,9 +14,9 @@ import ( "github.com/go-gost/core/logger" md "github.com/go-gost/core/metadata" "github.com/go-gost/core/observer" - "github.com/go-gost/core/service" "github.com/go-gost/relay" ctxvalue "github.com/go-gost/x/ctx" + xnet "github.com/go-gost/x/internal/net" limiter_util "github.com/go-gost/x/internal/util/limiter" stats_util "github.com/go-gost/x/internal/util/stats" rate_limiter "github.com/go-gost/x/limiter/rate" @@ -24,12 +25,10 @@ import ( ) var ( - ErrBadVersion = errors.New("bad version") - ErrUnknownCmd = errors.New("unknown command") - ErrRouterID = errors.New("invalid router ID") - ErrRouterNotAvailable = errors.New("router not available") - ErrUnauthorized = errors.New("unauthorized") - ErrTunnelRoute = errors.New("no route to host") + ErrBadVersion = errors.New("bad version") + ErrUnknownCmd = errors.New("unknown command") + ErrRouterID = errors.New("invalid router ID") + ErrUnauthorized = errors.New("unauthorized") ) func init() { @@ -40,7 +39,7 @@ type routerHandler struct { id string options handler.Options pool *ConnectorPool - epSvc service.Service + epConn net.PacketConn md metadata log logger.Logger stats *stats_util.HandlerStats @@ -75,14 +74,9 @@ func (h *routerHandler) Init(md md.Metadata) (err error) { }) h.pool = NewConnectorPool(h.id) - /* - for _, ro := range h.options.Recorders { - if ro.Record == xrecorder.RecorderServiceHandler { - h.ep.recorder = ro - break - } - } - */ + if err = h.initEntrypoint(); err != nil { + return + } ctx, cancel := context.WithCancel(context.Background()) h.cancel = cancel @@ -103,6 +97,35 @@ func (h *routerHandler) Init(md md.Metadata) (err error) { return nil } +func (h *routerHandler) initEntrypoint() (err error) { + if h.md.entryPoint == "" { + return + } + + network := "udp" + if xnet.IsIPv4(h.md.entryPoint) { + network = "udp4" + } + + pc, err := net.ListenPacket(network, h.md.entryPoint) + if err != nil { + h.log.Error(err) + return + } + h.epConn = pc + + log := h.log.WithFields(map[string]any{ + "service": fmt.Sprintf("%s-ep-%s", h.options.Service, pc.LocalAddr()), + "listener": "udp", + "handler": "entrypoint", + "kind": "service", + }) + go h.handleEntrypoint(log) + h.log.Infof("entrypoint: %s", pc.LocalAddr()) + + return +} + func (h *routerHandler) Handle(ctx context.Context, conn net.Conn, opts ...handler.HandleOption) (err error) { defer conn.Close() @@ -212,8 +235,8 @@ func (h *routerHandler) Handle(ctx context.Context, conn net.Conn, opts ...handl // Close implements io.Closer interface. func (h *routerHandler) Close() error { - if h.epSvc != nil { - h.epSvc.Close() + if h.epConn != nil { + h.epConn.Close() } h.pool.Close() diff --git a/handler/router/metadata.go b/handler/router/metadata.go index 3bfa1476..d9b198eb 100644 --- a/handler/router/metadata.go +++ b/handler/router/metadata.go @@ -13,7 +13,7 @@ import ( const ( defaultTTL = 15 * time.Second - defaultBufferSize = 1500 + defaultBufferSize = 4096 ) type metadata struct { diff --git a/handler/tun/server.go b/handler/tun/server.go index 8e070940..9c23cc3e 100644 --- a/handler/tun/server.go +++ b/handler/tun/server.go @@ -272,7 +272,7 @@ func (h *tunHandler) findRouteFor(ctx context.Context, dst net.IP, router router return nil } - if route := router.GetRoute(ctx, dst); route != nil { + if route := router.GetRoute(ctx, dst.String()); route != nil { if gw := net.ParseIP(route.Gateway); gw != nil { if v, ok := h.routes.Load(ipToTunRouteKey(gw)); ok { return v.(net.Addr) diff --git a/handler/tunnel/dialer.go b/handler/tunnel/dialer.go index 1b03a0fa..41de72ca 100644 --- a/handler/tunnel/dialer.go +++ b/handler/tunnel/dialer.go @@ -57,7 +57,7 @@ func (d *Dialer) Dial(ctx context.Context, network string, tid string) (conn net var service *sd.Service for _, s := range ss { d.log.Debugf("%+v", s) - if s.Name != d.node && s.Network == network { + if s.Node != d.node && s.Network == network { service = s break } diff --git a/listener/tun/tun_linux.go b/listener/tun/tun_linux.go index 95c93850..49695c00 100644 --- a/listener/tun/tun_linux.go +++ b/listener/tun/tun_linux.go @@ -52,14 +52,10 @@ func (l *tunListener) addRoutes(ifce *net.Interface) error { if route.Net == nil { continue } - gw := net.ParseIP(route.Gateway) - if gw == nil { - continue - } r := netlink.Route{ Dst: route.Net, - Gw: gw, + Gw: net.ParseIP(route.Gateway), } if r.Gw == nil { r.LinkIndex = ifce.Index diff --git a/registry/router.go b/registry/router.go index e8b46330..fcbd814e 100644 --- a/registry/router.go +++ b/registry/router.go @@ -2,7 +2,6 @@ package registry import ( "context" - "net" "github.com/go-gost/core/router" ) @@ -31,7 +30,7 @@ type routerWrapper struct { r *routerRegistry } -func (w *routerWrapper) GetRoute(ctx context.Context, dst net.IP, opts ...router.Option) *router.Route { +func (w *routerWrapper) GetRoute(ctx context.Context, dst string, opts ...router.Option) *router.Route { v := w.r.get(w.name) if v == nil { return nil diff --git a/router/plugin/grpc.go b/router/plugin/grpc.go index f413d157..ec7f49ee 100644 --- a/router/plugin/grpc.go +++ b/router/plugin/grpc.go @@ -3,7 +3,6 @@ package router import ( "context" "io" - "net" "github.com/go-gost/core/logger" "github.com/go-gost/core/router" @@ -45,21 +44,27 @@ func NewGRPCPlugin(name string, addr string, opts ...plugin.Option) router.Route return p } -func (p *grpcPlugin) GetRoute(ctx context.Context, dst net.IP, opts ...router.Option) *router.Route { +func (p *grpcPlugin) GetRoute(ctx context.Context, dst string, opts ...router.Option) *router.Route { if p.client == nil { return nil } + var options router.Options + for _, opt := range opts { + opt(&options) + } + r, err := p.client.GetRoute(ctx, &proto.GetRouteRequest{ - Dst: dst.String(), + Dst: dst, + Id: options.ID, }) if err != nil { p.log.Error(err) return nil } - return xrouter.ParseRoute(r.Net, r.Gateway) + return xrouter.ParseRoute(r.Dst, r.Gateway) } func (p *grpcPlugin) Close() error { diff --git a/router/plugin/http.go b/router/plugin/http.go index 03cc069a..838526de 100644 --- a/router/plugin/http.go +++ b/router/plugin/http.go @@ -3,7 +3,6 @@ package router import ( "context" "encoding/json" - "net" "net/http" "github.com/go-gost/core/logger" @@ -14,10 +13,12 @@ import ( type httpPluginGetRouteRequest struct { Dst string `json:"dst"` + ID string `json:"id"` } type httpPluginGetRouteResponse struct { Net string `json:"net"` + Dst string `json:"dst"` Gateway string `json:"gateway"` } @@ -46,11 +47,16 @@ func NewHTTPPlugin(name string, url string, opts ...plugin.Option) router.Router } } -func (p *httpPlugin) GetRoute(ctx context.Context, dst net.IP, opts ...router.Option) *router.Route { +func (p *httpPlugin) GetRoute(ctx context.Context, dst string, opts ...router.Option) *router.Route { if p.client == nil { return nil } + var options router.Options + for _, opt := range opts { + opt(&options) + } + req, err := http.NewRequestWithContext(ctx, http.MethodGet, p.url, nil) if err != nil { return nil @@ -61,7 +67,10 @@ func (p *httpPlugin) GetRoute(ctx context.Context, dst net.IP, opts ...router.Op req.Header.Set("Content-Type", "application/json") q := req.URL.Query() - q.Set("dst", dst.String()) + q.Set("dst", dst) + if options.ID != "" { + q.Set("id", options.ID) + } req.URL.RawQuery = q.Encode() resp, err := p.client.Do(req) @@ -79,5 +88,9 @@ func (p *httpPlugin) GetRoute(ctx context.Context, dst net.IP, opts ...router.Op return nil } - return xrouter.ParseRoute(res.Net, res.Gateway) + dstNet := res.Dst + if dstNet == "" { + dstNet = res.Net + } + return xrouter.ParseRoute(dstNet, res.Gateway) } diff --git a/router/router.go b/router/router.go index 78cdf8bb..f3296efe 100644 --- a/router/router.go +++ b/router/router.go @@ -202,17 +202,19 @@ func (p *localRouter) parseRoutes(r io.Reader) (routes []*router.Route, err erro return } -func (p *localRouter) GetRoute(ctx context.Context, dst net.IP, opts ...router.Option) *router.Route { - if dst == nil || p == nil { +func (p *localRouter) GetRoute(ctx context.Context, dst string, opts ...router.Option) *router.Route { + if dst == "" || p == nil { return nil } + dstIP := net.ParseIP(dst) + p.mu.RLock() routes := p.routes p.mu.RUnlock() for _, route := range routes { - if route.Net != nil && route.Net.Contains(dst) { + if route.Dst == dst || route.Net != nil && route.Net.Contains(dstIP) { return route } } @@ -255,11 +257,6 @@ func ParseRoute(dst string, gateway string) *router.Route { } _, ipNet, _ := net.ParseCIDR(dst) - gw := net.ParseIP(gateway) - if gw == nil { - return nil - } - return &router.Route{ Net: ipNet, Dst: dst,