go-gost/x
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

Merge branch 'master' into dev

Browse Source
This commit is contained in:
wenyifan 2023-11-15 10:35:36 +08:00
commit 52aa2027d0
26 changed files with 191 additions and 879 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

21
config/parsing/service/parse.go
View File

@ -23,6 +23,7 @@ import (
bypass_parser "github.com/go-gost/x/config/parsing/bypass"
hop_parser "github.com/go-gost/x/config/parsing/hop"
selector_parser "github.com/go-gost/x/config/parsing/selector"
xnet "github.com/go-gost/x/internal/net"
tls_util "github.com/go-gost/x/internal/util/tls"
"github.com/go-gost/x/metadata"
"github.com/go-gost/x/registry"
@ -258,10 +259,18 @@ func parseForwarder(cfg *config.ForwarderConfig) (hop.Hop, error) {
}
for _, node := range cfg.Nodes {
if node != nil {
hc.Nodes = append(hc.Nodes,
&config.NodeConfig{
Name: node.Name,
Addr: node.Addr,
addrs := xnet.AddrPortRange(node.Addr).Addrs()
if len(addrs) == 0 {
addrs = append(addrs, node.Addr)
}
for i, addr := range addrs {
name := node.Name
if i > 0 {
name = fmt.Sprintf("%s-%d", node.Name, i)
}
hc.Nodes = append(hc.Nodes, &config.NodeConfig{
Name: name,
Addr: addr,
Host: node.Host,
Network: node.Network,
Protocol: node.Protocol,
@ -271,8 +280,8 @@ func parseForwarder(cfg *config.ForwarderConfig) (hop.Hop, error) {
HTTP: node.HTTP,
TLS: node.TLS,
Auth: node.Auth,
},
)
})
}
}
}
if len(hc.Nodes) > 0 {

92
connector/relay/bind.go
View File

@ -16,10 +16,6 @@ import (
// Bind implements connector.Binder.
func (c *relayConnector) Bind(ctx context.Context, conn net.Conn, network, address string, opts ...connector.BindOption) (net.Listener, error) {
if !c.md.tunnelID.IsZero() {
return c.bindTunnel(ctx, conn, network, address, c.options.Logger)
}
log := c.options.Logger.WithFields(map[string]any{
"network": network,
"address": address,
@ -43,86 +39,6 @@ func (c *relayConnector) Bind(ctx context.Context, conn net.Conn, network, addre
}
}
func (c *relayConnector) bindTunnel(ctx context.Context, conn net.Conn, network, address string, log logger.Logger) (net.Listener, error) {
addr, cid, err := c.initTunnel(conn, network, address)
if err != nil {
return nil, err
}
log.Infof("create tunnel on %s/%s OK, tunnel=%s, connector=%s", addr, network, c.md.tunnelID.String(), cid)
session, err := mux.ServerSession(conn, c.md.muxCfg)
if err != nil {
return nil, err
}
return &bindListener{
network: network,
addr: addr,
session: session,
logger: log,
}, nil
}
func (c *relayConnector) initTunnel(conn net.Conn, network, address string) (addr net.Addr, cid relay.ConnectorID, err error) {
req := relay.Request{
Version: relay.Version1,
Cmd: relay.CmdBind,
}
if network == "udp" {
req.Cmd |= relay.FUDP
}
if c.options.Auth != nil {
pwd, _ := c.options.Auth.Password()
req.Features = append(req.Features, &relay.UserAuthFeature{
Username: c.options.Auth.Username(),
Password: pwd,
})
}
af := &relay.AddrFeature{}
af.ParseFrom(address)
req.Features = append(req.Features, af,
&relay.TunnelFeature{
ID: c.md.tunnelID.ID(),
},
)
if _, err = req.WriteTo(conn); err != nil {
return
}
// first reply, bind status
resp := relay.Response{}
if _, err = resp.ReadFrom(conn); err != nil {
return
}
if resp.Status != relay.StatusOK {
err = fmt.Errorf("%d: create tunnel %s failed", resp.Status, c.md.tunnelID.String())
return
}
for _, f := range resp.Features {
switch f.Type() {
case relay.FeatureAddr:
if feature, _ := f.(*relay.AddrFeature); feature != nil {
addr = &bindAddr{
network: network,
addr: net.JoinHostPort(feature.Host, strconv.Itoa(int(feature.Port))),
}
}
case relay.FeatureTunnel:
if feature, _ := f.(*relay.TunnelFeature); feature != nil {
cid = relay.NewConnectorID(feature.ID[:])
}
}
}
return
}
func (c *relayConnector) bindTCP(ctx context.Context, conn net.Conn, network, address string, log logger.Logger) (net.Listener, error) {
laddr, err := c.bind(conn, relay.CmdBind, network, address)
if err != nil {
@ -177,6 +93,14 @@ func (c *relayConnector) bind(conn net.Conn, cmd relay.CmdType, network, address
Password: pwd,
})
}
nid := relay.NetworkTCP
if network == "udp" || network == "udp4" || network == "udp6" {
nid = relay.NetworkUDP
}
req.Features = append(req.Features, &relay.NetworkFeature{
Network: nid,
})
fa := &relay.AddrFeature{}
fa.ParseFrom(address)
req.Features = append(req.Features, fa)

10
connector/relay/conn.go
View File

@ -19,6 +19,7 @@ type tcpConn struct {
net.Conn
wbuf *bytes.Buffer
once sync.Once
mu sync.Mutex
}
func (c *tcpConn) Read(b []byte) (n int, err error) {
@ -36,6 +37,10 @@ func (c *tcpConn) Read(b []byte) (n int, err error) {
func (c *tcpConn) Write(b []byte) (n int, err error) {
n = len(b) // force byte length consistent
c.mu.Lock()
defer c.mu.Unlock()
if c.wbuf != nil && c.wbuf.Len() > 0 {
c.wbuf.Write(b) // append the data to the cached header
_, err = c.Conn.Write(c.wbuf.Bytes())
@ -50,6 +55,7 @@ type udpConn struct {
net.Conn
wbuf *bytes.Buffer
once sync.Once
mu sync.Mutex
}
func (c *udpConn) Read(b []byte) (n int, err error) {
@ -88,6 +94,10 @@ func (c *udpConn) Write(b []byte) (n int, err error) {
}
n = len(b)
c.mu.Lock()
defer c.mu.Unlock()
if c.wbuf != nil && c.wbuf.Len() > 0 {
var bb [2]byte
binary.BigEndian.PutUint16(bb[:], uint16(len(b)))

6
connector/relay/connector.go
View File

@ -101,12 +101,6 @@ func (c *relayConnector) Connect(ctx context.Context, conn net.Conn, network, ad
req.Features = append(req.Features, af)
}
if !c.md.tunnelID.IsZero() {
req.Features = append(req.Features, &relay.TunnelFeature{
ID: c.md.tunnelID.ID(),
})
}
if c.md.noDelay {
if _, err := req.WriteTo(conn); err != nil {
return nil, err

11
connector/relay/metadata.go
View File

@ -5,15 +5,12 @@ import (
mdata "github.com/go-gost/core/metadata"
mdutil "github.com/go-gost/core/metadata/util"
"github.com/go-gost/relay"
"github.com/go-gost/x/internal/util/mux"
"github.com/google/uuid"
)
type metadata struct {
connectTimeout time.Duration
noDelay bool
tunnelID relay.TunnelID
muxCfg *mux.Config
}
@ -26,14 +23,6 @@ func (c *relayConnector) parseMetadata(md mdata.Metadata) (err error) {
c.md.connectTimeout = mdutil.GetDuration(md, connectTimeout)
c.md.noDelay = mdutil.GetBool(md, noDelay)
if s := mdutil.GetString(md, "tunnelID", "tunnel.id"); s != "" {
uuid, err := uuid.Parse(s)
if err != nil {
return err
}
c.md.tunnelID = relay.NewTunnelID(uuid[:])
}
c.md.muxCfg = &mux.Config{
Version: mdutil.GetInt(md, "mux.version"),
KeepAliveInterval: mdutil.GetDuration(md, "mux.keepaliveInterval"),

2
go.mod
View File

@ -7,7 +7,7 @@ require (
github.com/asaskevich/govalidator v0.0.0-20210307081110-f21760c49a8d
github.com/gin-contrib/cors v1.3.1
github.com/gin-gonic/gin v1.9.1
github.com/go-gost/core v0.0.0-20231109123312-8e4fc06cf1b7
github.com/go-gost/core v0.0.0-20231113123850-a916f0401649
github.com/go-gost/gosocks4 v0.0.1
github.com/go-gost/gosocks5 v0.4.0
github.com/go-gost/plugin v0.0.0-20231109123346-0ae4157b9d25

2
go.sum
View File

@ -95,6 +95,8 @@ github.com/go-gl/glfw/v3.3/glfw v0.0.0-20191125211704-12ad95a8df72/go.mod h1:tQ2
github.com/go-gl/glfw/v3.3/glfw v0.0.0-20200222043503-6f7a984d4dc4/go.mod h1:tQ2UAYgL5IevRw8kRxooKSPJfGvJ9fJQFa0TUsXzTg8=
github.com/go-gost/core v0.0.0-20231109123312-8e4fc06cf1b7 h1:sDsPtmP51qf8zN/RbZZj/3vNLCoH0sdvpIRwV6TfzvY=
github.com/go-gost/core v0.0.0-20231109123312-8e4fc06cf1b7/go.mod h1:ndkgWVYRLwupVaFFWv8ML1Nr8tD3xhHK245PLpUDg4E=
github.com/go-gost/core v0.0.0-20231113123850-a916f0401649 h1:14iGAk7cqc+aDWtsuY6CWpP0lvC54pA5Izjeh5FdQNs=
github.com/go-gost/core v0.0.0-20231113123850-a916f0401649/go.mod h1:ndkgWVYRLwupVaFFWv8ML1Nr8tD3xhHK245PLpUDg4E=
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.0 h1:EIrOEkpJez4gwHrMa33frA+hHXJyevjp47thpMQsJzI=

2
handler/http/udp.go
View File

@ -71,7 +71,7 @@ func (h *httpHandler) handleUDP(ctx context.Context, conn net.Conn, log logger.L
t := time.Now()
log.Infof("%s <-> %s", conn.RemoteAddr(), pc.LocalAddr())
relay.Run()
relay.Run(ctx)
log.WithFields(map[string]any{
"duration": time.Since(t),
}).Infof("%s >-< %s", conn.RemoteAddr(), pc.LocalAddr())

56
handler/relay/bind.go
View File

@ -2,8 +2,6 @@ package relay
import (
"context"
"crypto/md5"
"encoding/hex"
"fmt"
"net"
"time"
@ -17,7 +15,6 @@ import (
relay_util "github.com/go-gost/x/internal/util/relay"
metrics "github.com/go-gost/x/metrics/wrapper"
xservice "github.com/go-gost/x/service"
"github.com/google/uuid"
)
func (h *relayHandler) handleBind(ctx context.Context, conn net.Conn, network, address string, log logger.Logger) error {
@ -176,60 +173,9 @@ func (h *relayHandler) bindUDP(ctx context.Context, conn net.Conn, network, addr
t := time.Now()
log.Debugf("%s <-> %s", conn.RemoteAddr(), pc.LocalAddr())
r.Run()
r.Run(ctx)
log.WithFields(map[string]any{
"duration": time.Since(t),
}).Debugf("%s >-< %s", conn.RemoteAddr(), pc.LocalAddr())
return nil
}
func (h *relayHandler) handleBindTunnel(ctx context.Context, conn net.Conn, network, address string, tunnelID relay.TunnelID, log logger.Logger) (err error) {
resp := relay.Response{
Version: relay.Version1,
Status: relay.StatusOK,
}
uuid, err := uuid.NewRandom()
if err != nil {
resp.Status = relay.StatusInternalServerError
resp.WriteTo(conn)
return
}
connectorID := relay.NewConnectorID(uuid[:])
if network == "udp" {
connectorID = relay.NewUDPConnectorID(uuid[:])
}
addr := address
if host, port, _ := net.SplitHostPort(addr); host == "" {
v := md5.Sum([]byte(tunnelID.String()))
host = hex.EncodeToString(v[:8])
addr = net.JoinHostPort(host, port)
}
af := &relay.AddrFeature{}
err = af.ParseFrom(addr)
if err != nil {
log.Warn(err)
}
resp.Features = append(resp.Features, af,
&relay.TunnelFeature{
ID: connectorID.ID(),
},
)
resp.WriteTo(conn)
// Upgrade connection to multiplex session.
session, err := mux.ClientSession(conn, h.md.muxCfg)
if err != nil {
return
}
h.pool.Add(tunnelID, NewConnector(connectorID, session))
if h.md.ingress != nil {
h.md.ingress.Set(ctx, addr, tunnelID.String())
}
log.Debugf("%s/%s: tunnel=%s, connector=%s established", address, network, tunnelID, connectorID)
return
}

97
handler/relay/connect.go
View File

@ -6,7 +6,6 @@ import (
"fmt"
"io"
"net"
"strconv"
"time"
"github.com/go-gost/core/logger"
@ -113,99 +112,3 @@ func (h *relayHandler) handleConnect(ctx context.Context, conn net.Conn, network
return nil
}
func (h *relayHandler) handleConnectTunnel(ctx context.Context, conn net.Conn, network, address string, tunnelID relay.TunnelID, log logger.Logger) error {
log = log.WithFields(map[string]any{
"dst": fmt.Sprintf("%s/%s", address, network),
"cmd": "connect",
"tunnel": tunnelID.String(),
})
log.Debugf("%s >> %s/%s", conn.RemoteAddr(), address, network)
resp := relay.Response{
Version: relay.Version1,
Status: relay.StatusOK,
}
host, sp, _ := net.SplitHostPort(address)
if h.options.Bypass != nil && h.options.Bypass.Contains(ctx, network, address) {
log.Debug("bypass: ", address)
resp.Status = relay.StatusForbidden
_, err := resp.WriteTo(conn)
return err
}
var tid relay.TunnelID
if ingress := h.md.ingress; ingress != nil {
tid = parseTunnelID(ingress.Get(ctx, host))
}
if !tid.Equal(tunnelID) && !h.md.directTunnel {
resp.Status = relay.StatusHostUnreachable
resp.WriteTo(conn)
err := fmt.Errorf("no route to host %s", host)
log.Error(err)
return err
}
cc, _, err := getTunnelConn(network, h.pool, tunnelID, 3, log)
if err != nil {
resp.Status = relay.StatusServiceUnavailable
resp.WriteTo(conn)
log.Error(err)
return err
}
defer cc.Close()
log.Debugf("%s >> %s", conn.RemoteAddr(), cc.RemoteAddr())
if h.md.noDelay {
if _, err := resp.WriteTo(conn); err != nil {
log.Error(err)
return err
}
} else {
rc := &tcpConn{
Conn: conn,
}
// cache the header
if _, err := resp.WriteTo(&rc.wbuf); err != nil {
return err
}
conn = rc
}
var features []relay.Feature
af := &relay.AddrFeature{} // source/visitor address
af.ParseFrom(conn.RemoteAddr().String())
features = append(features, af)
if host != "" {
port, _ := strconv.Atoi(sp)
// target host
af = &relay.AddrFeature{
AType: relay.AddrDomain,
Host: host,
Port: uint16(port),
}
features = append(features, af)
}
resp = relay.Response{
Version: relay.Version1,
Status: relay.StatusOK,
Features: features,
}
resp.WriteTo(cc)
t := time.Now()
log.Debugf("%s <-> %s", conn.RemoteAddr(), cc.RemoteAddr())
xnet.Transport(conn, cc)
log.WithFields(map[string]any{
"duration": time.Since(t),
}).Debugf("%s >-< %s", conn.RemoteAddr(), cc.RemoteAddr())
return nil
}

208
handler/relay/entrypoint.go
View File

@ -1,25 +1,17 @@
package relay
import (
"bufio"
"context"
"fmt"
"io"
"net"
"net/http"
"net/http/httputil"
"time"
"github.com/go-gost/core/handler"
"github.com/go-gost/core/ingress"
"github.com/go-gost/core/listener"
"github.com/go-gost/core/logger"
md "github.com/go-gost/core/metadata"
"github.com/go-gost/relay"
admission "github.com/go-gost/x/admission/wrapper"
xnet "github.com/go-gost/x/internal/net"
"github.com/go-gost/x/internal/net/proxyproto"
"github.com/go-gost/x/internal/util/forward"
"github.com/go-gost/x/internal/util/mux"
climiter "github.com/go-gost/x/limiter/conn/wrapper"
limiter "github.com/go-gost/x/limiter/traffic/wrapper"
@ -130,203 +122,3 @@ func (h *tcpHandler) Handle(ctx context.Context, conn net.Conn, opts ...handler.
Debugf("%s >-< %s", conn.RemoteAddr(), cc.RemoteAddr())
return nil
}
type tunnelHandler struct {
pool *ConnectorPool
ingress ingress.Ingress
options handler.Options
}
func newTunnelHandler(pool *ConnectorPool, ingress ingress.Ingress, opts ...handler.Option) handler.Handler {
options := handler.Options{}
for _, opt := range opts {
opt(&options)
}
return &tunnelHandler{
pool: pool,
ingress: ingress,
options: options,
}
}
func (h *tunnelHandler) Init(md md.Metadata) (err error) {
return
}
func (h *tunnelHandler) Handle(ctx context.Context, conn net.Conn, opts ...handler.HandleOption) error {
defer conn.Close()
start := time.Now()
log := h.options.Logger.WithFields(map[string]any{
"remote": conn.RemoteAddr().String(),
"local": conn.LocalAddr().String(),
})
log.Infof("%s <> %s", conn.RemoteAddr(), conn.LocalAddr())
defer func() {
log.WithFields(map[string]any{
"duration": time.Since(start),
}).Infof("%s >< %s", conn.RemoteAddr(), conn.LocalAddr())
}()
var rw io.ReadWriter = conn
var host string
var protocol string
rw, host, protocol, _ = forward.Sniffing(ctx, conn)
h.options.Logger.Debugf("sniffing: host=%s, protocol=%s", host, protocol)
if protocol == forward.ProtoHTTP {
return h.handleHTTP(ctx, conn.RemoteAddr(), rw, log)
}
var tunnelID relay.TunnelID
if h.ingress != nil {
tunnelID = parseTunnelID(h.ingress.Get(ctx, host))
}
if tunnelID.IsZero() {
err := fmt.Errorf("no route to host %s", host)
log.Error(err)
return err
}
if tunnelID.IsPrivate() {
err := fmt.Errorf("access denied: tunnel %s is private for host %s", tunnelID, host)
log.Error(err)
return err
}
log = log.WithFields(map[string]any{
"tunnel": tunnelID.String(),
})
cc, _, err := getTunnelConn("tcp", h.pool, tunnelID, 3, log)
if err != nil {
log.Error(err)
return err
}
defer cc.Close()
log.Debugf("%s >> %s", conn.RemoteAddr(), cc.RemoteAddr())
var features []relay.Feature
af := &relay.AddrFeature{}
af.ParseFrom(conn.RemoteAddr().String()) // client address
features = append(features, af)
if host != "" {
// target host
af := &relay.AddrFeature{}
af.ParseFrom(host)
features = append(features, af)
}
resp := relay.Response{
Version: relay.Version1,
Status: relay.StatusOK,
Features: features,
}
resp.WriteTo(cc)
t := time.Now()
log.Debugf("%s <-> %s", conn.RemoteAddr(), cc.RemoteAddr())
xnet.Transport(rw, cc)
log.WithFields(map[string]any{
"duration": time.Since(t),
}).Debugf("%s >-< %s", conn.RemoteAddr(), cc.RemoteAddr())
return nil
}
func (h *tunnelHandler) handleHTTP(ctx context.Context, raddr net.Addr, rw io.ReadWriter, log logger.Logger) (err error) {
br := bufio.NewReader(rw)
for {
resp := &http.Response{
ProtoMajor: 1,
ProtoMinor: 1,
Header: http.Header{},
StatusCode: http.StatusServiceUnavailable,
}
err = func() error {
req, err := http.ReadRequest(br)
if err != nil {
return err
}
var tunnelID relay.TunnelID
if h.ingress != nil {
tunnelID = parseTunnelID(h.ingress.Get(ctx, req.Host))
}
if tunnelID.IsZero() {
err := fmt.Errorf("no route to host %s", req.Host)
log.Error(err)
resp.StatusCode = http.StatusBadGateway
return resp.Write(rw)
}
if tunnelID.IsPrivate() {
err := fmt.Errorf("access denied: tunnel %s is private for host %s", tunnelID, req.Host)
log.Error(err)
resp.StatusCode = http.StatusBadGateway
return resp.Write(rw)
}
log = log.WithFields(map[string]any{
"host": req.Host,
"tunnel": tunnelID.String(),
})
cc, cid, err := getTunnelConn("tcp", h.pool, tunnelID, 3, log)
if err != nil {
log.Error(err)
return resp.Write(rw)
}
defer cc.Close()
log.Debugf("new connection to tunnel %s(connector %s)", tunnelID, cid)
var features []relay.Feature
af := &relay.AddrFeature{}
af.ParseFrom(raddr.String())
features = append(features, af)
if host := req.Host; host != "" {
if h, _, _ := net.SplitHostPort(host); h == "" {
host = net.JoinHostPort(host, "80")
}
af := &relay.AddrFeature{}
af.ParseFrom(host)
features = append(features, af)
}
(&relay.Response{
Version: relay.Version1,
Status: relay.StatusOK,
Features: features,
}).WriteTo(cc)
if log.IsLevelEnabled(logger.TraceLevel) {
dump, _ := httputil.DumpRequest(req, false)
log.Trace(string(dump))
}
if err := req.Write(cc); err != nil {
log.Warnf("send request to tunnel %s: %v", tunnelID, err)
return resp.Write(rw)
}
res, err := http.ReadResponse(bufio.NewReader(cc), req)
if err != nil {
log.Warnf("read response from tunnel %s: %v", tunnelID, err)
return resp.Write(rw)
}
defer res.Body.Close()
return res.Write(rw)
}()
if err != nil {
// log.Error(err)
break
}
}
return
}

78
handler/relay/handler.go
View File

@ -3,7 +3,6 @@ package relay
import (
"context"
"errors"
"fmt"
"net"
"strconv"
"time"
@ -11,14 +10,11 @@ import (
"github.com/go-gost/core/chain"
"github.com/go-gost/core/handler"
"github.com/go-gost/core/hop"
"github.com/go-gost/core/listener"
md "github.com/go-gost/core/metadata"
"github.com/go-gost/core/service"
"github.com/go-gost/relay"
xnet "github.com/go-gost/x/internal/net"
auth_util "github.com/go-gost/x/internal/util/auth"
"github.com/go-gost/x/registry"
xservice "github.com/go-gost/x/service"
)
var (
@ -38,7 +34,6 @@ type relayHandler struct {
md metadata
options handler.Options
ep service.Service
pool *ConnectorPool
}
func NewHandler(opts ...handler.Option) handler.Handler {
@ -49,7 +44,6 @@ func NewHandler(opts ...handler.Option) handler.Handler {
return &relayHandler{
options: options,
pool: NewConnectorPool(),
}
}
@ -63,68 +57,9 @@ func (h *relayHandler) Init(md md.Metadata) (err error) {
h.router = chain.NewRouter(chain.LoggerRouterOption(h.options.Logger))
}
if err = h.initEntryPoint(); err != nil {
return
}
return nil
}
func (h *relayHandler) initEntryPoint() (err error) {
if h.md.entryPoint == "" {
return
}
network := "tcp"
if xnet.IsIPv4(h.md.entryPoint) {
network = "tcp4"
}
ln, err := net.Listen(network, h.md.entryPoint)
if err != nil {
h.options.Logger.Error(err)
return
}
serviceName := fmt.Sprintf("%s-ep-%s", h.options.Service, ln.Addr())
log := h.options.Logger.WithFields(map[string]any{
"service": serviceName,
"listener": "tcp",
"handler": "ep-tunnel",
"kind": "service",
})
epListener := newTCPListener(ln,
listener.AddrOption(h.md.entryPoint),
listener.ServiceOption(serviceName),
listener.ProxyProtocolOption(h.md.entryPointProxyProtocol),
listener.LoggerOption(log.WithFields(map[string]any{
"kind": "listener",
})),
)
if err = epListener.Init(nil); err != nil {
return
}
epHandler := newTunnelHandler(
h.pool,
h.md.ingress,
handler.ServiceOption(serviceName),
handler.LoggerOption(log.WithFields(map[string]any{
"kind": "handler",
})),
)
if err = epHandler.Init(nil); err != nil {
return
}
h.ep = xservice.NewService(
serviceName, epListener, epHandler,
xservice.LoggerOption(log),
)
go h.ep.Serve()
log.Infof("entrypoint: %s", h.ep.Addr())
return
}
// Forward implements handler.Forwarder.
func (h *relayHandler) Forward(hop hop.Hop) {
h.hop = hop
@ -179,7 +114,6 @@ func (h *relayHandler) Handle(ctx context.Context, conn net.Conn, opts ...handle
var user, pass string
var address string
var networkID relay.NetworkID
var tunnelID relay.TunnelID
for _, f := range req.Features {
switch f.Type() {
case relay.FeatureUserAuth:
@ -190,10 +124,6 @@ func (h *relayHandler) Handle(ctx context.Context, conn net.Conn, opts ...handle
if feature, _ := f.(*relay.AddrFeature); feature != nil {
address = net.JoinHostPort(feature.Host, strconv.Itoa(int(feature.Port)))
}
case relay.FeatureTunnel:
if feature, _ := f.(*relay.TunnelFeature); feature != nil {
tunnelID = relay.NewTunnelID(feature.ID[:])
}
case relay.FeatureNetwork:
if feature, _ := f.(*relay.NetworkFeature); feature != nil {
networkID = feature.Network
@ -230,16 +160,10 @@ func (h *relayHandler) Handle(ctx context.Context, conn net.Conn, opts ...handle
case 0, relay.CmdConnect:
defer conn.Close()
if !tunnelID.IsZero() {
return h.handleConnectTunnel(ctx, conn, network, address, tunnelID, log)
}
return h.handleConnect(ctx, conn, network, address, log)
case relay.CmdBind:
if !tunnelID.IsZero() {
return h.handleBindTunnel(ctx, conn, network, address, tunnelID, log)
}
defer conn.Close()
return h.handleBind(ctx, conn, network, address, log)
default:
resp.Status = relay.StatusBadRequest

61
handler/relay/metadata.go
View File

@ -2,41 +2,29 @@ package relay
import (
"math"
"strings"
"time"
"github.com/go-gost/core/ingress"
"github.com/go-gost/core/logger"
mdata "github.com/go-gost/core/metadata"
mdutil "github.com/go-gost/core/metadata/util"
xingress "github.com/go-gost/x/ingress"
"github.com/go-gost/x/internal/util/mux"
"github.com/go-gost/x/registry"
)
type metadata struct {
readTimeout time.Duration
enableBind bool
udpBufferSize int
noDelay bool
hash string
directTunnel bool
entryPoint string
entryPointProxyProtocol int
ingress ingress.Ingress
muxCfg *mux.Config
readTimeout time.Duration
enableBind bool
udpBufferSize int
noDelay bool
hash string
muxCfg *mux.Config
}
func (h *relayHandler) parseMetadata(md mdata.Metadata) (err error) {
const (
readTimeout = "readTimeout"
enableBind = "bind"
udpBufferSize = "udpBufferSize"
noDelay = "nodelay"
hash = "hash"
entryPoint = "entryPoint"
entryPointID = "entryPoint.id"
entryPointProxyProtocol = "entryPoint.proxyProtocol"
readTimeout = "readTimeout"
enableBind = "bind"
udpBufferSize = "udpBufferSize"
noDelay = "nodelay"
hash = "hash"
)
h.md.readTimeout = mdutil.GetDuration(md, readTimeout)
@ -51,33 +39,6 @@ func (h *relayHandler) parseMetadata(md mdata.Metadata) (err error) {
h.md.hash = mdutil.GetString(md, hash)
h.md.directTunnel = mdutil.GetBool(md, "tunnel.direct")
h.md.entryPoint = mdutil.GetString(md, entryPoint)
h.md.entryPointProxyProtocol = mdutil.GetInt(md, entryPointProxyProtocol)
h.md.ingress = registry.IngressRegistry().Get(mdutil.GetString(md, "ingress"))
if h.md.ingress == nil {
var rules []xingress.Rule
for _, s := range strings.Split(mdutil.GetString(md, "tunnel"), ",") {
ss := strings.SplitN(s, ":", 2)
if len(ss) != 2 {
continue
}
rules = append(rules, xingress.Rule{
Hostname: ss[0],
Endpoint: ss[1],
})
}
if len(rules) > 0 {
h.md.ingress = xingress.NewIngress(
xingress.RulesOption(rules),
xingress.LoggerOption(logger.Default().WithFields(map[string]any{
"kind": "ingress",
})),
)
}
}
h.md.muxCfg = &mux.Config{
Version: mdutil.GetInt(md, "mux.version"),
KeepAliveInterval: mdutil.GetDuration(md, "mux.keepaliveInterval"),

200
handler/relay/tunnel.go
View File

@ -1,200 +0,0 @@
package relay
import (
"fmt"
"net"
"sync"
"sync/atomic"
"time"
"github.com/go-gost/core/logger"
"github.com/go-gost/relay"
"github.com/go-gost/x/internal/util/mux"
"github.com/google/uuid"
)
type Connector struct {
id relay.ConnectorID
t time.Time
s *mux.Session
}
func NewConnector(id relay.ConnectorID, s *mux.Session) *Connector {
c := &Connector{
id: id,
t: time.Now(),
s: s,
}
go c.accept()
return c
}
func (c *Connector) accept() {
for {
conn, err := c.s.Accept()
if err != nil {
logger.Default().Errorf("connector %s: %v", c.id, err)
c.s.Close()
return
}
conn.Close()
}
}
func (c *Connector) ID() relay.ConnectorID {
return c.id
}
func (c *Connector) Session() *mux.Session {
return c.s
}
type Tunnel struct {
id relay.TunnelID
connectors []*Connector
t time.Time
n uint64
mu sync.RWMutex
}
func NewTunnel(id relay.TunnelID) *Tunnel {
t := &Tunnel{
id: id,
t: time.Now(),
}
go t.clean()
return t
}
func (t *Tunnel) ID() relay.TunnelID {
return t.id
}
func (t *Tunnel) AddConnector(c *Connector) {
if c == nil {
return
}
t.mu.Lock()
defer t.mu.Unlock()
t.connectors = append(t.connectors, c)
}
func (t *Tunnel) GetConnector(network string) *Connector {
t.mu.RLock()
defer t.mu.RUnlock()
var connectors []*Connector
for _, c := range t.connectors {
if network == "udp" && c.id.IsUDP() ||
network != "udp" && !c.id.IsUDP() {
connectors = append(connectors, c)
}
}
if len(connectors) == 0 {
return nil
}
n := atomic.AddUint64(&t.n, 1) - 1
return connectors[n%uint64(len(connectors))]
}
func (t *Tunnel) clean() {
ticker := time.NewTicker(30 * time.Second)
for range ticker.C {
t.mu.Lock()
var connectors []*Connector
for _, c := range t.connectors {
if c.Session().IsClosed() {
logger.Default().Debugf("remove tunnel %s connector %s", t.id, c.id)
continue
}
connectors = append(connectors, c)
}
if len(connectors) != len(t.connectors) {
t.connectors = connectors
}
t.mu.Unlock()
}
}
type ConnectorPool struct {
tunnels map[string]*Tunnel
mu sync.RWMutex
}
func NewConnectorPool() *ConnectorPool {
return &ConnectorPool{
tunnels: make(map[string]*Tunnel),
}
}
func (p *ConnectorPool) Add(tid relay.TunnelID, c *Connector) {
p.mu.Lock()
defer p.mu.Unlock()
s := tid.String()
t := p.tunnels[s]
if t == nil {
t = NewTunnel(tid)
p.tunnels[s] = t
}
t.AddConnector(c)
}
func (p *ConnectorPool) Get(network string, tid relay.TunnelID) *Connector {
if p == nil {
return nil
}
p.mu.RLock()
defer p.mu.RUnlock()
t := p.tunnels[tid.String()]
if t == nil {
return nil
}
return t.GetConnector(network)
}
func parseTunnelID(s string) (tid relay.TunnelID) {
if s == "" {
return
}
private := false
if s[0] == '$' {
private = true
s = s[1:]
}
uuid, _ := uuid.Parse(s)
if private {
return relay.NewPrivateTunnelID(uuid[:])
}
return relay.NewTunnelID(uuid[:])
}
func getTunnelConn(network string, pool *ConnectorPool, tid relay.TunnelID, retry int, log logger.Logger) (conn net.Conn, cid relay.ConnectorID, err error) {
if retry <= 0 {
retry = 1
}
for i := 0; i < retry; i++ {
c := pool.Get(network, tid)
if c == nil {
err = fmt.Errorf("tunnel %s not available", tid.String())
break
}
conn, err = c.Session().GetConn()
if err != nil {
log.Error(err)
continue
}
cid = c.id
break
}
return
}

2
handler/socks/v5/udp.go
View File

@ -72,7 +72,7 @@ func (h *socks5Handler) handleUDP(ctx context.Context, conn net.Conn, log logger
WithLogger(log)
r.SetBufferSize(h.md.udpBufferSize)
go r.Run()
go r.Run(ctx)
t := time.Now()
log.Debugf("%s <-> %s", conn.RemoteAddr(), cc.LocalAddr())

2
handler/socks/v5/udp_tun.go
View File

@ -63,7 +63,7 @@ func (h *socks5Handler) handleUDPTun(ctx context.Context, conn net.Conn, network
t := time.Now()
log.Debugf("%s <-> %s", conn.RemoteAddr(), pc.LocalAddr())
r.Run()
r.Run(ctx)
log.WithFields(map[string]any{
"duration": time.Since(t),
}).Debugf("%s >-< %s", conn.RemoteAddr(), pc.LocalAddr())

6
handler/tun/client.go
View File

@ -44,7 +44,7 @@ func (h *tunHandler) handleClient(ctx context.Context, conn net.Conn, raddr stri
ctx, cancel := context.WithCancel(ctx)
defer cancel()
go h.keepAlive(ctx, cc, ips)
go h.keepalive(ctx, cc, ips)
return h.transportClient(conn, cc, log)
}()
@ -57,7 +57,7 @@ func (h *tunHandler) handleClient(ctx context.Context, conn net.Conn, raddr stri
}
}
func (h *tunHandler) keepAlive(ctx context.Context, conn net.Conn, ips []net.IP) {
func (h *tunHandler) keepalive(ctx context.Context, conn net.Conn, ips []net.IP) {
// handshake
keepAliveData := bufpool.Get(keepAliveHeaderLength + len(ips)*net.IPv6len)
defer bufpool.Put(keepAliveData)
@ -130,7 +130,7 @@ func (h *tunHandler) transportClient(tun io.ReadWriter, conn net.Conn, log logge
ipProtocol(waterutil.IPProtocol(header.NextHeader)),
header.PayloadLen, header.TrafficClass)
} else {
log.Warn("unknown packet, discarded")
log.Warnf("unknown packet, discarded(%d)", n)
return nil
}

4
handler/tun/server.go
View File

@ -78,7 +78,7 @@ func (h *tunHandler) transportServer(ctx context.Context, tun io.ReadWriter, con
ipProtocol(waterutil.IPProtocol(header.NextHeader)),
header.PayloadLen, header.TrafficClass)
} else {
log.Warn("unknown packet, discarded")
log.Warnf("unknown packet, discarded(%d)", n)
return nil
}
@ -199,7 +199,7 @@ func (h *tunHandler) transportServer(ctx context.Context, tun io.ReadWriter, con
ipProtocol(waterutil.IPProtocol(header.NextHeader)),
header.PayloadLen, header.TrafficClass)
} else {
log.Warn("unknown packet, discarded")
log.Warnf("unknown packet, discarded(%d): % x", n, b[:n])
return nil
}

4
ingress/ingress.go
View File

@ -259,8 +259,8 @@ func (ing *localIngress) Get(ctx context.Context, host string, opts ...ingress.G
return ep
}
func (ing *localIngress) Set(ctx context.Context, host, endpoint string, opts ...ingress.SetOption) {
func (ing *localIngress) Set(ctx context.Context, host, endpoint string, opts ...ingress.SetOption) bool {
return false
}
func (ing *localIngress) lookup(host string) string {

29
ingress/plugin.go
View File

@ -62,15 +62,20 @@ func (p *grpcPlugin) Get(ctx context.Context, host string, opts ...ingress.GetOp
return r.GetEndpoint()
}
func (p *grpcPlugin) Set(ctx context.Context, host, endpoint string, opts ...ingress.SetOption) {
func (p *grpcPlugin) Set(ctx context.Context, host, endpoint string, opts ...ingress.SetOption) bool {
if p.client == nil {
return
return false
}
p.client.Set(ctx, &proto.SetRequest{
r, _ := p.client.Set(ctx, &proto.SetRequest{
Host: host,
Endpoint: endpoint,
})
if r == nil {
return false
}
return r.Ok
}
func (p *grpcPlugin) Close() error {
@ -94,6 +99,7 @@ type httpPluginSetRequest struct {
}
type httpPluginSetResponse struct {
OK bool `json:"ok"`
}
type httpPlugin struct {
@ -156,9 +162,9 @@ func (p *httpPlugin) Get(ctx context.Context, host string, opts ...ingress.GetOp
return res.Endpoint
}
func (p *httpPlugin) Set(ctx context.Context, host, endpoint string, opts ...ingress.SetOption) {
func (p *httpPlugin) Set(ctx context.Context, host, endpoint string, opts ...ingress.SetOption) bool {
if p.client == nil {
return
return false
}
rb := httpPluginSetRequest{
@ -167,12 +173,12 @@ func (p *httpPlugin) Set(ctx context.Context, host, endpoint string, opts ...ing
}
v, err := json.Marshal(&rb)
if err != nil {
return
return false
}
req, err := http.NewRequestWithContext(ctx, http.MethodPut, p.url, bytes.NewReader(v))
if err != nil {
return
return false
}
if p.header != nil {
@ -181,16 +187,17 @@ func (p *httpPlugin) Set(ctx context.Context, host, endpoint string, opts ...ing
req.Header.Set("Content-Type", "application/json")
resp, err := p.client.Do(req)
if err != nil {
return
return false
}
defer resp.Body.Close()
if resp.StatusCode != http.StatusOK {
return
return false
}
res := httpPluginSetResponse{}
res := httpPluginSetResponse{ }
if err := json.NewDecoder(resp.Body).Decode(&res); err != nil {
return
return false
}
return res.OK
}

70
internal/matcher/matcher.go
View File

@ -1,11 +1,11 @@
package matcher
import (
"fmt"
"net"
"strconv"
"strings"
xnet "github.com/go-gost/x/internal/net"
"github.com/gobwas/glob"
"github.com/yl2chen/cidranger"
)
@ -40,7 +40,7 @@ func (m *ipMatcher) Match(ip string) bool {
}
type addrMatcher struct {
addrs map[string]*PortRange
addrs map[string]*xnet.PortRange
}
// AddrMatcher creates a Matcher with a list of HOST:PORT addresses.
@ -50,7 +50,7 @@ type addrMatcher struct {
// The PORT can be a single port number or port range MIN-MAX(e.g. 0-65535).
func AddrMatcher(addrs []string) Matcher {
matcher := &addrMatcher{
addrs: make(map[string]*PortRange),
addrs: make(map[string]*xnet.PortRange),
}
for _, addr := range addrs {
host, port, _ := net.SplitHostPort(addr)
@ -58,7 +58,10 @@ func AddrMatcher(addrs []string) Matcher {
matcher.addrs[addr] = nil
continue
}
pr, _ := parsePortRange(port)
pr := &xnet.PortRange{}
if err := pr.Parse(port); err != nil {
pr = nil
}
matcher.addrs[host] = pr
}
return matcher
@ -75,13 +78,13 @@ func (m *addrMatcher) Match(addr string) bool {
port, _ := strconv.Atoi(sp)
if pr, ok := m.addrs[host]; ok {
if pr == nil || pr.contains(port) {
if pr == nil || pr.Contains(port) {
return true
}
}
if pr, ok := m.addrs["."+host]; ok {
if pr == nil || pr.contains(port) {
if pr == nil || pr.Contains(port) {
return true
}
}
@ -89,7 +92,7 @@ func (m *addrMatcher) Match(addr string) bool {
for {
if index := strings.IndexByte(host, '.'); index > 0 {
if pr, ok := m.addrs[host[index:]]; ok {
if pr == nil || pr.contains(port) {
if pr == nil || pr.Contains(port) {
return true
}
}
@ -172,7 +175,7 @@ func (m *domainMatcher) Match(domain string) bool {
type wildcardMatcherPattern struct {
glob glob.Glob
pr *PortRange
pr *xnet.PortRange
}
type wildcardMatcher struct {
patterns []wildcardMatcherPattern
@ -187,7 +190,11 @@ func WildcardMatcher(patterns []string) Matcher {
if host == "" {
host = pattern
}
pr, _ := parsePortRange(port)
pr := &xnet.PortRange{}
if err := pr.Parse(port); err != nil {
pr = nil
}
matcher.patterns = append(matcher.patterns, wildcardMatcherPattern{
glob: glob.MustCompile(host),
pr: pr,
@ -208,8 +215,8 @@ func (m *wildcardMatcher) Match(addr string) bool {
}
port, _ := strconv.Atoi(sp)
for _, pattern := range m.patterns {
if pattern.glob.Match(addr) {
if pattern.pr == nil || pattern.pr.contains(port) {
if pattern.glob.Match(host) {
if pattern.pr == nil || pattern.pr.Contains(port) {
return true
}
}
@ -217,44 +224,3 @@ func (m *wildcardMatcher) Match(addr string) bool {
return false
}
type PortRange struct {
Min int
Max int
}
// ParsePortRange parses the s to a PortRange.
// The s can be a single port number and will be converted to port range port-port.
func parsePortRange(s string) (*PortRange, error) {
minmax := strings.Split(s, "-")
switch len(minmax) {
case 1:
port, err := strconv.Atoi(s)
if err != nil {
return nil, err
}
if port < 0 || port > 65535 {
return nil, fmt.Errorf("invalid port: %s", s)
}
return &PortRange{Min: port, Max: port}, nil
case 2:
min, err := strconv.Atoi(minmax[0])
if err != nil {
return nil, err
}
max, err := strconv.Atoi(minmax[1])
if err != nil {
return nil, err
}
return &PortRange{Min: min, Max: max}, nil
default:
return nil, fmt.Errorf("invalid range: %s", s)
}
}
func (pr *PortRange) contains(port int) bool {
return port >= pr.Min && port <= pr.Max
}

72
internal/net/addr.go Normal file
View File

@ -0,0 +1,72 @@
package net
import (
"fmt"
"net"
"strconv"
"strings"
)
// AddrPortRange is the network address with port range supported.
// e.g. 192.168.1.1:0-65535
type AddrPortRange string
func (p AddrPortRange) Addrs() (addrs []string) {
h, sp, err := net.SplitHostPort(string(p))
if err != nil {
return nil
}
pr := PortRange{}
pr.Parse(sp)
for i := pr.Min; i <= pr.Max; i++ {
addrs = append(addrs, net.JoinHostPort(h, strconv.Itoa(i)))
}
return addrs
}
// Port range is a range of port list.
type PortRange struct {
Min int
Max int
}
// Parse parses the s to PortRange.
// The s can be a single port number and will be converted to port range port-port.
func (pr *PortRange) Parse(s string) error {
minmax := strings.Split(s, "-")
switch len(minmax) {
case 1:
port, err := strconv.Atoi(s)
if err != nil {
return err
}
if port < 0 || port > 65535 {
return fmt.Errorf("invalid port: %s", s)
}
pr.Min, pr.Max = port, port
return nil
case 2:
min, err := strconv.Atoi(minmax[0])
if err != nil {
return err
}
max, err := strconv.Atoi(minmax[1])
if err != nil {
return err
}
pr.Min, pr.Max = min, max
return nil
default:
return fmt.Errorf("invalid range: %s", s)
}
}
func (pr *PortRange) Contains(port int) bool {
return port >= pr.Min && port <= pr.Max
}

6
internal/net/udp/relay.go
View File

@ -39,7 +39,7 @@ func (r *Relay) SetBufferSize(n int) {
r.bufferSize = n
}
func (r *Relay) Run() (err error) {
func (r *Relay) Run(ctx context.Context) (err error) {
bufSize := r.bufferSize
if bufSize <= 0 {
bufSize = 4096
@ -58,7 +58,7 @@ func (r *Relay) Run() (err error) {
return err
}
if r.bypass != nil && r.bypass.Contains(context.Background(), "udp", raddr.String()) {
if r.bypass != nil && r.bypass.Contains(ctx, "udp", raddr.String()) {
if r.logger != nil {
r.logger.Warn("bypass: ", raddr)
}
@ -96,7 +96,7 @@ func (r *Relay) Run() (err error) {
return err
}
if r.bypass != nil && r.bypass.Contains(context.Background(), "udp", raddr.String()) {
if r.bypass != nil && r.bypass.Contains(ctx, "udp", raddr.String()) {
if r.logger != nil {
r.logger.Warn("bypass: ", raddr)
}

21
listener/rudp/listener.go
View File

@ -50,12 +50,13 @@ func (l *rudpListener) Init(md md.Metadata) (err error) {
if xnet.IsIPv4(l.options.Addr) {
network = "udp4"
}
laddr, err := net.ResolveUDPAddr(network, l.options.Addr)
if err != nil {
return
if laddr, _ := net.ResolveUDPAddr(network, l.options.Addr); laddr != nil {
l.laddr = laddr
}
if l.laddr == nil {
l.laddr = &bindAddr{addr: l.options.Addr}
}
l.laddr = laddr
l.router = chain.NewRouter(
chain.ChainRouterOption(l.options.Chain),
chain.LoggerRouterOption(l.logger),
@ -116,3 +117,15 @@ func (l *rudpListener) Close() error {
return nil
}
type bindAddr struct {
addr string
}
func (p *bindAddr) Network() string {
return "tcp"
}
func (p *bindAddr) String() string {
return p.addr
}

2
listener/rudp/metadata.go
View File

@ -9,7 +9,7 @@ import (
const (
defaultTTL = 5 * time.Second
defaultReadBufferSize = 4096
defaultReadBufferSize = 1024
defaultReadQueueSize = 1024
defaultBacklog = 128
)

6
registry/ingress.go
View File

@ -38,11 +38,11 @@ func (w *ingressWrapper) Get(ctx context.Context, host string, opts ...ingress.G
return v.Get(ctx, host, opts...)
}
func (w *ingressWrapper) Set(ctx context.Context, host, endpoint string, opts ...ingress.SetOption) {
func (w *ingressWrapper) Set(ctx context.Context, host, endpoint string, opts ...ingress.SetOption) bool {
v := w.r.get(w.name)
if v == nil {
return
return false
}
v.Set(ctx, host, endpoint, opts...)
return v.Set(ctx, host, endpoint, opts...)
}