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

add tunnel handler and connector

Browse Source
This commit is contained in:
ginuerzh
2023-10-15 15:39:25 +08:00
parent 033151a770
commit 497915f465
24 changed files with 1375 additions and 63 deletions
Download Patch File Download Diff File Expand all files Collapse all files

19
handler/relay/bind.go
View File

@ -2,6 +2,8 @@ package relay
import (
"context"
"crypto/md5"
"encoding/hex"
"fmt"
"net"
"time"
@ -181,7 +183,7 @@ func (h *relayHandler) bindUDP(ctx context.Context, conn net.Conn, network, addr
return nil
}
func (h *relayHandler) handleBindTunnel(ctx context.Context, conn net.Conn, network string, tunnelID relay.TunnelID, log logger.Logger) (err error) {
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,
@ -198,9 +200,11 @@ func (h *relayHandler) handleBindTunnel(ctx context.Context, conn net.Conn, netw
connectorID = relay.NewUDPConnectorID(uuid[:])
}
addr := ":0"
if h.ep != nil {
addr = h.ep.Addr().String()
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)
@ -221,10 +225,11 @@ func (h *relayHandler) handleBindTunnel(ctx context.Context, conn net.Conn, netw
}
h.pool.Add(tunnelID, NewConnector(connectorID, session))
log.Debugf("tunnel %s connector %s/%s established", tunnelID, connectorID, network)
if h.recorder.Recorder != nil {
h.recorder.Recorder.Record(ctx, tunnelID[:])
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
}

15
handler/relay/connect.go
View File

@ -143,15 +143,12 @@ func (h *relayHandler) handleConnectTunnel(ctx context.Context, conn net.Conn, n
tid = parseTunnelID(ingress.Get(ctx, host))
}
// client is not an public entrypoint.
if h.md.entryPointID.IsZero() || !tunnelID.Equal(h.md.entryPointID) {
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
}
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, tid, 3, log)

26
handler/relay/handler.go
View File

@ -13,12 +13,10 @@ import (
"github.com/go-gost/core/hop"
"github.com/go-gost/core/listener"
md "github.com/go-gost/core/metadata"
"github.com/go-gost/core/recorder"
"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"
xrecorder "github.com/go-gost/x/recorder"
"github.com/go-gost/x/registry"
xservice "github.com/go-gost/x/service"
)
@ -35,13 +33,12 @@ func init() {
}
type relayHandler struct {
hop hop.Hop
router *chain.Router
md metadata
options handler.Options
ep service.Service
pool *ConnectorPool
recorder recorder.RecorderObject
hop hop.Hop
router *chain.Router
md metadata
options handler.Options
ep service.Service
pool *ConnectorPool
}
func NewHandler(opts ...handler.Option) handler.Handler {
@ -66,15 +63,6 @@ func (h *relayHandler) Init(md md.Metadata) (err error) {
h.router = chain.NewRouter(chain.LoggerRouterOption(h.options.Logger))
}
if opts := h.router.Options(); opts != nil {
for _, ro := range opts.Recorders {
if ro.Record == xrecorder.RecorderServiceHandlerRelayTunnelEndpoint {
h.recorder = ro
break
}
}
}
if err = h.initEntryPoint(); err != nil {
return
}
@ -248,7 +236,7 @@ func (h *relayHandler) Handle(ctx context.Context, conn net.Conn, opts ...handle
return h.handleConnect(ctx, conn, network, address, log)
case relay.CmdBind:
if !tunnelID.IsZero() {
return h.handleBindTunnel(ctx, conn, network, tunnelID, log)
return h.handleBindTunnel(ctx, conn, network, address, tunnelID, log)
}
defer conn.Close()

3
handler/relay/metadata.go
View File

@ -9,7 +9,6 @@ import (
"github.com/go-gost/core/logger"
mdata "github.com/go-gost/core/metadata"
mdutil "github.com/go-gost/core/metadata/util"
"github.com/go-gost/relay"
xingress "github.com/go-gost/x/ingress"
"github.com/go-gost/x/registry"
)
@ -22,7 +21,6 @@ type metadata struct {
hash string
directTunnel bool
entryPoint string
entryPointID relay.TunnelID
entryPointProxyProtocol int
ingress ingress.Ingress
}
@ -53,7 +51,6 @@ func (h *relayHandler) parseMetadata(md mdata.Metadata) (err error) {
h.md.directTunnel = mdutil.GetBool(md, "tunnel.direct")
h.md.entryPoint = mdutil.GetString(md, entryPoint)
h.md.entryPointID = parseTunnelID(mdutil.GetString(md, entryPointID))
h.md.entryPointProxyProtocol = mdutil.GetInt(md, entryPointProxyProtocol)
h.md.ingress = registry.IngressRegistry().Get(mdutil.GetString(md, "ingress"))

67
handler/tunnel/bind.go Normal file
View File

@ -0,0 +1,67 @@
package tunnel
import (
"context"
"crypto/md5"
"encoding/hex"
"net"
"github.com/go-gost/core/logger"
"github.com/go-gost/relay"
"github.com/go-gost/x/internal/util/mux"
"github.com/google/uuid"
)
func (h *tunnelHandler) handleBind(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)
if err != nil {
return
}
h.pool.Add(tunnelID, NewConnector(connectorID, session))
if h.md.ingress != nil {
h.md.ingress.Set(ctx, addr, tunnelID.String())
}
if h.recorder.Recorder != nil {
h.recorder.Recorder.Record(ctx, tunnelID[:])
}
log.Debugf("%s/%s: tunnel=%s, connector=%s established", addr, network, tunnelID, connectorID)
return
}

26
handler/tunnel/conn.go Normal file
View File

@ -0,0 +1,26 @@
package tunnel
import (
"bytes"
"net"
)
type tcpConn struct {
net.Conn
wbuf bytes.Buffer
}
func (c *tcpConn) Read(b []byte) (n int, err error) {
return c.Conn.Read(b)
}
func (c *tcpConn) Write(b []byte) (n int, err error) {
n = len(b) // force byte length consistent
if c.wbuf.Len() > 0 {
c.wbuf.Write(b) // append the data to the cached header
_, err = c.wbuf.WriteTo(c.Conn)
return
}
_, err = c.Conn.Write(b)
return
}

105
handler/tunnel/connect.go Normal file
View File

@ -0,0 +1,105 @@
package tunnel
import (
"context"
"fmt"
"net"
"strconv"
"time"
"github.com/go-gost/core/logger"
"github.com/go-gost/relay"
xnet "github.com/go-gost/x/internal/net"
)
func (h *tunnelHandler) handleConnect(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))
}
// client is not an public entrypoint.
if h.md.entryPointID.IsZero() || !tunnelID.Equal(h.md.entryPointID) {
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, tid, 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())
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
}

198
handler/tunnel/handler.go Normal file
View File

@ -0,0 +1,198 @@
package tunnel
import (
"context"
"errors"
"net"
"strconv"
"time"
"github.com/go-gost/core/chain"
"github.com/go-gost/core/handler"
md "github.com/go-gost/core/metadata"
"github.com/go-gost/core/recorder"
"github.com/go-gost/relay"
auth_util "github.com/go-gost/x/internal/util/auth"
xrecorder "github.com/go-gost/x/recorder"
"github.com/go-gost/x/registry"
)
var (
ErrBadVersion = errors.New("relay: bad version")
ErrUnknownCmd = errors.New("relay: unknown command")
ErrTunnelID = errors.New("tunnel: invalid tunnel ID")
ErrUnauthorized = errors.New("relay: unauthorized")
ErrRateLimit = errors.New("relay: rate limiting exceeded")
)
func init() {
registry.HandlerRegistry().Register("tunnel", NewHandler)
}
type tunnelHandler struct {
router *chain.Router
md metadata
options handler.Options
pool *ConnectorPool
recorder recorder.RecorderObject
}
func NewHandler(opts ...handler.Option) handler.Handler {
options := handler.Options{}
for _, opt := range opts {
opt(&options)
}
return &tunnelHandler{
options: options,
pool: NewConnectorPool(),
}
}
func (h *tunnelHandler) Init(md md.Metadata) (err error) {
if err := h.parseMetadata(md); err != nil {
return err
}
h.router = h.options.Router
if h.router == nil {
h.router = chain.NewRouter(chain.LoggerRouterOption(h.options.Logger))
}
if opts := h.router.Options(); opts != nil {
for _, ro := range opts.Recorders {
if ro.Record == xrecorder.RecorderServiceHandlerTunnelEndpoint {
h.recorder = ro
break
}
}
}
return nil
}
func (h *tunnelHandler) Handle(ctx context.Context, conn net.Conn, opts ...handler.HandleOption) (err error) {
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() {
if err != nil {
conn.Close()
}
log.WithFields(map[string]any{
"duration": time.Since(start),
}).Infof("%s >< %s", conn.RemoteAddr(), conn.LocalAddr())
}()
ctx = auth_util.ContextWithClientAddr(ctx, auth_util.ClientAddr(conn.RemoteAddr().String()))
if !h.checkRateLimit(conn.RemoteAddr()) {
return ErrRateLimit
}
if h.md.readTimeout > 0 {
conn.SetReadDeadline(time.Now().Add(h.md.readTimeout))
}
req := relay.Request{}
if _, err := req.ReadFrom(conn); err != nil {
return err
}
conn.SetReadDeadline(time.Time{})
resp := relay.Response{
Version: relay.Version1,
Status: relay.StatusOK,
}
if req.Version != relay.Version1 {
resp.Status = relay.StatusBadRequest
resp.WriteTo(conn)
return ErrBadVersion
}
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:
if feature, _ := f.(*relay.UserAuthFeature); feature != nil {
user, pass = feature.Username, feature.Password
}
case relay.FeatureAddr:
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
}
}
}
if tunnelID.IsZero() {
resp.Status = relay.StatusBadRequest
resp.WriteTo(conn)
return ErrTunnelID
}
if user != "" {
log = log.WithFields(map[string]any{"user": user})
}
if h.options.Auther != nil {
id, ok := h.options.Auther.Authenticate(ctx, user, pass)
if !ok {
resp.Status = relay.StatusUnauthorized
resp.WriteTo(conn)
return ErrUnauthorized
}
ctx = auth_util.ContextWithID(ctx, auth_util.ID(id))
}
network := networkID.String()
if (req.Cmd & relay.FUDP) == relay.FUDP {
network = "udp"
}
switch req.Cmd & relay.CmdMask {
case relay.CmdConnect:
defer conn.Close()
return h.handleConnect(ctx, conn, network, address, tunnelID, log)
case relay.CmdBind:
return h.handleBind(ctx, conn, network, address, tunnelID, log)
default:
resp.Status = relay.StatusBadRequest
resp.WriteTo(conn)
return ErrUnknownCmd
}
}
// Close implements io.Closer interface.
func (h *tunnelHandler) Close() error {
return nil
}
func (h *tunnelHandler) checkRateLimit(addr net.Addr) bool {
if h.options.RateLimiter == nil {
return true
}
host, _, _ := net.SplitHostPort(addr.String())
if limiter := h.options.RateLimiter.Limiter(host); limiter != nil {
return limiter.Allow(1)
}
return true
}

62
handler/tunnel/metadata.go Normal file
View File

@ -0,0 +1,62 @@
package tunnel
import (
"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"
"github.com/go-gost/relay"
xingress "github.com/go-gost/x/ingress"
"github.com/go-gost/x/registry"
)
type metadata struct {
readTimeout time.Duration
hash string
directTunnel bool
entryPointID relay.TunnelID
ingress ingress.Ingress
}
func (h *tunnelHandler) parseMetadata(md mdata.Metadata) (err error) {
const (
readTimeout = "readTimeout"
entryPointID = "entrypoint.id"
hash = "hash"
)
h.md.readTimeout = mdutil.GetDuration(md, readTimeout)
h.md.hash = mdutil.GetString(md, hash)
h.md.directTunnel = mdutil.GetBool(md, "tunnel.direct")
h.md.entryPointID = parseTunnelID(mdutil.GetString(md, entryPointID))
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",
})),
)
}
}
return
}

200
handler/tunnel/tunnel.go Normal file
View File

@ -0,0 +1,200 @@
package tunnel
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
}