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

add sd for tunnel

Browse Source
This commit is contained in:
ginuerzh
2023-10-31 22:59:14 +08:00
parent e8d5e719a4
commit a7166b8206
17 changed files with 795 additions and 173 deletions
Download Patch File Download Diff File Expand all files Collapse all files

15
handler/tunnel/bind.go
View File

@ -8,7 +8,6 @@ import (
"net"
"github.com/go-gost/core/logger"
"github.com/go-gost/core/recorder"
"github.com/go-gost/relay"
"github.com/go-gost/x/internal/util/mux"
"github.com/google/uuid"
@ -59,15 +58,11 @@ func (h *tunnelHandler) handleBind(ctx context.Context, conn net.Conn, network,
if h.md.ingress != nil {
h.md.ingress.Set(ctx, addr, tunnelID.String())
}
if h.recorder != nil {
h.recorder.Record(ctx,
[]byte(fmt.Sprintf("%s:%s", tunnelID, connectorID)),
recorder.MetadataReocrdOption(connectorMetadata{
Op: "add",
Network: network,
Server: conn.LocalAddr().String(),
}),
)
if sd := h.md.sd; sd != nil {
err := sd.Register(ctx, fmt.Sprintf("%s:%s:%s", h.id, tunnelID, connectorID), network, h.md.entryPoint)
if err != nil {
h.log.Error(err)
}
}
log.Debugf("%s/%s: tunnel=%s, connector=%s established", addr, network, tunnelID, connectorID)

82
handler/tunnel/connect.go
View File

@ -11,7 +11,7 @@ import (
xnet "github.com/go-gost/x/internal/net"
)
func (h *tunnelHandler) handleConnect(ctx context.Context, conn net.Conn, network, srcAddr string, dstAddr string, tunnelID relay.TunnelID, log logger.Logger) error {
func (h *tunnelHandler) handleConnect(ctx context.Context, req *relay.Request, conn net.Conn, network, srcAddr string, dstAddr string, tunnelID relay.TunnelID, log logger.Logger) error {
log = log.WithFields(map[string]any{
"dst": fmt.Sprintf("%s/%s", dstAddr, network),
"cmd": "connect",
@ -33,58 +33,68 @@ func (h *tunnelHandler) handleConnect(ctx context.Context, conn net.Conn, networ
host, _, _ := net.SplitHostPort(dstAddr)
// client is a public entrypoint.
if tunnelID.Equal(h.md.entryPointID) && !h.md.entryPointID.IsZero() {
if tunnelID.Equal(h.md.entryPointID) {
resp.WriteTo(conn)
return h.ep.handle(ctx, conn)
}
var tid relay.TunnelID
if ingress := h.md.ingress; ingress != nil && host != "" {
tid = parseTunnelID(ingress.Get(ctx, host))
if !h.md.directTunnel {
var tid relay.TunnelID
if ingress := h.md.ingress; ingress != nil && host != "" {
tid = parseTunnelID(ingress.Get(ctx, host))
}
if !tid.Equal(tunnelID) {
resp.Status = relay.StatusHostUnreachable
resp.WriteTo(conn)
err := fmt.Errorf("no route to host %s", host)
log.Error(err)
return err
}
}
// direct routing
if h.md.directTunnel {
tid = tunnelID
} else if !tid.Equal(tunnelID) {
resp.Status = relay.StatusHostUnreachable
resp.WriteTo(conn)
err := fmt.Errorf("no route to host %s", host)
log.Error(err)
return err
d := Dialer{
node: h.id,
pool: h.pool,
sd: h.md.sd,
retry: 3,
timeout: 15 * time.Second,
log: log,
}
cc, _, err := getTunnelConn(network, h.pool, tid, 3, log)
cc, node, cid, err := d.Dial(ctx, network, tunnelID.String())
if err != nil {
log.Error(err)
resp.Status = relay.StatusServiceUnavailable
resp.WriteTo(conn)
log.Error(err)
return err
}
defer cc.Close()
log.Debugf("%s >> %s", conn.RemoteAddr(), cc.RemoteAddr())
log.Debugf("new connection to tunnel: %s, connector: %s", tunnelID, cid)
if _, err := resp.WriteTo(conn); err != nil {
log.Error(err)
return err
if node == h.id {
if _, err := resp.WriteTo(conn); err != nil {
log.Error(err)
return err
}
resp = relay.Response{
Version: relay.Version1,
Status: relay.StatusOK,
}
af := &relay.AddrFeature{}
af.ParseFrom(srcAddr)
resp.Features = append(resp.Features, af) // src address
af = &relay.AddrFeature{}
af.ParseFrom(dstAddr)
resp.Features = append(resp.Features, af) // dst address
resp.WriteTo(cc)
} else {
req.WriteTo(cc)
}
resp = relay.Response{
Version: relay.Version1,
Status: relay.StatusOK,
}
af := &relay.AddrFeature{}
af.ParseFrom(srcAddr)
resp.Features = append(resp.Features, af) // src address
af = &relay.AddrFeature{}
af.ParseFrom(dstAddr)
resp.Features = append(resp.Features, af) // dst address
resp.WriteTo(cc)
t := time.Now()
log.Debugf("%s <-> %s", conn.RemoteAddr(), cc.RemoteAddr())
xnet.Transport(conn, cc)

76
handler/tunnel/dialer.go Normal file
View File

@ -0,0 +1,76 @@
package tunnel
import (
"context"
"net"
"time"
"github.com/go-gost/core/logger"
"github.com/go-gost/core/sd"
)
type Dialer struct {
node string
pool *ConnectorPool
sd sd.SD
retry int
timeout time.Duration
log logger.Logger
}
func (d *Dialer) Dial(ctx context.Context, network string, tid string) (conn net.Conn, node string, cid string, err error) {
retry := d.retry
retry = 1
for i := 0; i < retry; i++ {
c := d.pool.Get(network, tid)
if c == nil {
break
}
conn, err = c.Session().GetConn()
if err != nil {
d.log.Error(err)
continue
}
node = d.node
cid = c.id.String()
break
}
if conn != nil || err != nil {
return
}
if d.sd == nil {
err = ErrTunnelNotAvailable
return
}
ss, err := d.sd.Get(ctx, tid)
if err != nil {
return
}
var service *sd.Service
for _, s := range ss {
d.log.Debugf("%+v", s)
if s.Name != d.node && s.Network == network {
service = s
break
}
}
if service == nil || service.Address == "" {
err = ErrTunnelNotAvailable
return
}
node = service.Node
cid = service.Name
dialer := net.Dialer{
Timeout: d.timeout,
}
conn, err = dialer.DialContext(ctx, network, service.Address)
return
}

135
handler/tunnel/entrypoint.go
View File

@ -17,6 +17,7 @@ import (
"github.com/go-gost/core/listener"
"github.com/go-gost/core/logger"
md "github.com/go-gost/core/metadata"
"github.com/go-gost/core/sd"
"github.com/go-gost/relay"
admission "github.com/go-gost/x/admission/wrapper"
xio "github.com/go-gost/x/internal/io"
@ -28,8 +29,10 @@ import (
)
type entrypoint struct {
node string
pool *ConnectorPool
ingress ingress.Ingress
sd sd.SD
log logger.Logger
}
@ -51,6 +54,14 @@ func (ep *entrypoint) handle(ctx context.Context, conn net.Conn) error {
br := bufio.NewReader(conn)
v, err := br.Peek(1)
if err != nil {
return err
}
if v[0] == relay.Version1 {
return ep.handleConnect(ctx, xnet.NewBufferReaderConn(conn, br), log)
}
var cc net.Conn
for {
resp := &http.Response{
@ -102,32 +113,42 @@ func (ep *entrypoint) handle(ctx context.Context, conn net.Conn) error {
remoteAddr = addr
}
cc, cid, err := getTunnelConn("tcp", ep.pool, tunnelID, 3, log)
d := &Dialer{
node: ep.node,
pool: ep.pool,
sd: ep.sd,
retry: 3,
timeout: 15 * time.Second,
log: log,
}
cc, node, cid, err := d.Dial(ctx, "tcp", tunnelID.String())
if err != nil {
log.Error(err)
return resp.Write(conn)
}
log.Debugf("new connection to tunnel: %s, connector: %s", tunnelID, cid)
var features []relay.Feature
af := &relay.AddrFeature{}
af.ParseFrom(remoteAddr.String())
features = append(features, af) // src address
host := req.Host
if h, _, _ := net.SplitHostPort(host); h == "" {
host = net.JoinHostPort(host, "80")
}
af = &relay.AddrFeature{}
af.ParseFrom(host)
features = append(features, af) // dst address
(&relay.Response{
Version: relay.Version1,
Status: relay.StatusOK,
Features: features,
}).WriteTo(cc)
if node == ep.node {
var features []relay.Feature
af := &relay.AddrFeature{}
af.ParseFrom(remoteAddr.String())
features = append(features, af) // src address
af = &relay.AddrFeature{}
af.ParseFrom(host)
features = append(features, af) // dst address
(&relay.Response{
Version: relay.Version1,
Status: relay.StatusOK,
Features: features,
}).WriteTo(cc)
}
if err := req.Write(cc); err != nil {
cc.Close()
@ -186,6 +207,90 @@ func (ep *entrypoint) handle(ctx context.Context, conn net.Conn) error {
return nil
}
func (ep *entrypoint) handleConnect(ctx context.Context, conn net.Conn, log logger.Logger) error {
req := relay.Request{}
if _, err := req.ReadFrom(conn); err != nil {
return err
}
resp := relay.Response{
Version: relay.Version1,
Status: relay.StatusOK,
}
var srcAddr, dstAddr string
network := "tcp"
var tunnelID relay.TunnelID
for _, f := range req.Features {
switch f.Type() {
case relay.FeatureAddr:
if feature, _ := f.(*relay.AddrFeature); feature != nil {
v := net.JoinHostPort(feature.Host, strconv.Itoa(int(feature.Port)))
if srcAddr != "" {
dstAddr = v
} else {
srcAddr = v
}
}
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 {
network = feature.Network.String()
}
}
}
if tunnelID.IsZero() {
resp.Status = relay.StatusBadRequest
resp.WriteTo(conn)
return ErrTunnelID
}
d := Dialer{
pool: ep.pool,
retry: 3,
timeout: 15 * time.Second,
log: log,
}
cc, _, cid, err := d.Dial(ctx, network, tunnelID.String())
if err != nil {
log.Error(err)
resp.Status = relay.StatusServiceUnavailable
resp.WriteTo(conn)
return err
}
defer cc.Close()
log.Debugf("new connection to tunnel: %s, connector: %s", tunnelID, cid)
if _, err := resp.WriteTo(conn); err != nil {
log.Error(err)
return err
}
af := &relay.AddrFeature{}
af.ParseFrom(srcAddr)
resp.Features = append(resp.Features, af) // src address
af = &relay.AddrFeature{}
af.ParseFrom(dstAddr)
resp.Features = append(resp.Features, af) // dst address
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
}
func (ep *entrypoint) getRealClientAddr(req *http.Request, raddr net.Addr) net.Addr {
if req == nil {
return nil

60
handler/tunnel/handler.go
View File

@ -8,9 +8,9 @@ import (
"strconv"
"time"
"github.com/go-gost/core/chain"
"github.com/go-gost/core/handler"
"github.com/go-gost/core/listener"
"github.com/go-gost/core/logger"
md "github.com/go-gost/core/metadata"
"github.com/go-gost/core/recorder"
"github.com/go-gost/core/service"
@ -20,14 +20,16 @@ import (
xrecorder "github.com/go-gost/x/recorder"
"github.com/go-gost/x/registry"
xservice "github.com/go-gost/x/service"
"github.com/google/uuid"
)
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")
ErrBadVersion = errors.New("bad version")
ErrUnknownCmd = errors.New("unknown command")
ErrTunnelID = errors.New("invalid tunnel ID")
ErrTunnelNotAvailable = errors.New("tunnel not available")
ErrUnauthorized = errors.New("unauthorized")
ErrRateLimit = errors.New("rate limiting exceeded")
)
func init() {
@ -35,13 +37,14 @@ func init() {
}
type tunnelHandler struct {
router *chain.Router
md metadata
id string
options handler.Options
pool *ConnectorPool
recorder recorder.Recorder
epSvc service.Service
ep *entrypoint
md metadata
log logger.Logger
}
func NewHandler(opts ...handler.Option) handler.Handler {
@ -60,26 +63,33 @@ func (h *tunnelHandler) Init(md md.Metadata) (err error) {
return err
}
h.router = h.options.Router
if h.router == nil {
h.router = chain.NewRouter(chain.LoggerRouterOption(h.options.Logger))
uuid, err := uuid.NewRandom()
if err != nil {
return err
}
h.id = uuid.String()
if opts := h.router.Options(); opts != nil {
h.log = h.options.Logger.WithFields(map[string]any{
"node": h.id,
})
if opts := h.options.Router.Options(); opts != nil {
for _, ro := range opts.Recorders {
if ro.Record == xrecorder.RecorderServiceHandlerTunnelConnector {
if ro.Record == xrecorder.RecorderServiceHandlerTunnel {
h.recorder = ro.Recorder
break
}
}
}
h.pool = NewConnectorPool()
h.pool.WithRecorder(h.recorder)
h.pool = NewConnectorPool(h.id, h.md.sd)
h.ep = &entrypoint{
node: h.id,
pool: h.pool,
ingress: h.md.ingress,
log: h.options.Logger.WithFields(map[string]any{
sd: h.md.sd,
log: h.log.WithFields(map[string]any{
"kind": "entrypoint",
}),
}
@ -102,12 +112,12 @@ func (h *tunnelHandler) initEntrypoint() (err error) {
ln, err := net.Listen(network, h.md.entryPoint)
if err != nil {
h.options.Logger.Error(err)
h.log.Error(err)
return
}
serviceName := fmt.Sprintf("%s-ep-%s", h.options.Service, ln.Addr())
log := h.options.Logger.WithFields(map[string]any{
log := h.log.WithFields(map[string]any{
"service": serviceName,
"listener": "tcp",
"handler": "tunnel-ep",
@ -143,7 +153,7 @@ func (h *tunnelHandler) initEntrypoint() (err error) {
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{
log := h.log.WithFields(map[string]any{
"remote": conn.RemoteAddr().String(),
"local": conn.LocalAddr().String(),
})
@ -189,7 +199,7 @@ func (h *tunnelHandler) Handle(ctx context.Context, conn net.Conn, opts ...handl
var user, pass string
var srcAddr, dstAddr string
var networkID relay.NetworkID
network := "tcp"
var tunnelID relay.TunnelID
for _, f := range req.Features {
switch f.Type() {
@ -212,7 +222,7 @@ func (h *tunnelHandler) Handle(ctx context.Context, conn net.Conn, opts ...handl
}
case relay.FeatureNetwork:
if feature, _ := f.(*relay.NetworkFeature); feature != nil {
networkID = feature.Network
network = feature.Network.String()
}
}
}
@ -237,17 +247,13 @@ func (h *tunnelHandler) Handle(ctx context.Context, conn net.Conn, opts ...handl
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()
log.Debugf("connect: %s >> %s/%s", srcAddr, dstAddr, network)
return h.handleConnect(ctx, conn, network, srcAddr, dstAddr, tunnelID, log)
return h.handleConnect(ctx, &req, conn, network, srcAddr, dstAddr, tunnelID, log)
case relay.CmdBind:
log.Debugf("bind: %s >> %s/%s", srcAddr, dstAddr, network)
return h.handleBind(ctx, conn, network, dstAddr, tunnelID, log)

3
handler/tunnel/metadata.go
View File

@ -8,6 +8,7 @@ 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/core/sd"
"github.com/go-gost/relay"
xingress "github.com/go-gost/x/ingress"
"github.com/go-gost/x/internal/util/mux"
@ -21,6 +22,7 @@ type metadata struct {
entryPointProxyProtocol int
directTunnel bool
ingress ingress.Ingress
sd sd.SD
muxCfg *mux.Config
}
@ -54,6 +56,7 @@ func (h *tunnelHandler) parseMetadata(md mdata.Metadata) (err error) {
)
}
}
h.md.sd = registry.SDRegistry().Get(mdutil.GetString(md, "sd"))
h.md.muxCfg = &mux.Config{
Version: mdutil.GetInt(md, "mux.version"),

91
handler/tunnel/tunnel.go
View File

@ -3,24 +3,17 @@ package tunnel
import (
"context"
"fmt"
"net"
"sync"
"sync/atomic"
"time"
"github.com/go-gost/core/logger"
"github.com/go-gost/core/recorder"
"github.com/go-gost/core/sd"
"github.com/go-gost/relay"
"github.com/go-gost/x/internal/util/mux"
"github.com/google/uuid"
)
type connectorMetadata struct {
Op string
Network string
Server string
}
type Connector struct {
id relay.ConnectorID
t time.Time
@ -58,18 +51,20 @@ func (c *Connector) Session() *mux.Session {
}
type Tunnel struct {
node string
id relay.TunnelID
connectors []*Connector
t time.Time
n uint64
close chan struct{}
mu sync.RWMutex
recorder recorder.Recorder
sd sd.SD
}
func NewTunnel(id relay.TunnelID) *Tunnel {
func NewTunnel(node string, tid relay.TunnelID) *Tunnel {
t := &Tunnel{
id: id,
node: node,
id: tid,
t: time.Now(),
close: make(chan struct{}),
}
@ -77,8 +72,8 @@ func NewTunnel(id relay.TunnelID) *Tunnel {
return t
}
func (t *Tunnel) WithRecorder(recorder recorder.Recorder) {
t.recorder = recorder
func (t *Tunnel) WithSD(sd sd.SD) {
t.sd = sd
}
func (t *Tunnel) ID() relay.TunnelID {
@ -142,30 +137,21 @@ func (t *Tunnel) clean() {
t.mu.Lock()
if len(t.connectors) == 0 {
t.mu.Unlock()
break
}
var connectors []*Connector
for _, c := range t.connectors {
if c.Session().IsClosed() {
logger.Default().Debugf("remove tunnel: %s, connector: %s", t.id, c.id)
if t.recorder != nil {
t.recorder.Record(context.Background(),
[]byte(fmt.Sprintf("%s:%s", t.id, c.id)),
recorder.MetadataReocrdOption(connectorMetadata{
Op: "del",
}),
)
if t.sd != nil {
t.sd.Deregister(context.Background(), fmt.Sprintf("%s:%s:%s", t.node, t.id, c.id))
}
continue
}
connectors = append(connectors, c)
if t.recorder != nil {
t.recorder.Record(context.Background(),
[]byte(fmt.Sprintf("%s:%s", t.id, c.id)),
recorder.MetadataReocrdOption(connectorMetadata{
Op: "set",
}),
)
if t.sd != nil {
t.sd.Renew(context.Background(), fmt.Sprintf("%s:%s:%s", t.node, t.id, c.id))
}
}
if len(connectors) != len(t.connectors) {
@ -179,23 +165,22 @@ func (t *Tunnel) clean() {
}
type ConnectorPool struct {
tunnels map[string]*Tunnel
mu sync.RWMutex
recorder recorder.Recorder
node string
sd sd.SD
tunnels map[string]*Tunnel
mu sync.RWMutex
}
func NewConnectorPool() *ConnectorPool {
func NewConnectorPool(node string, sd sd.SD) *ConnectorPool {
p := &ConnectorPool{
node: node,
sd: sd,
tunnels: make(map[string]*Tunnel),
}
go p.closeIdles()
return p
}
func (p *ConnectorPool) WithRecorder(recorder recorder.Recorder) {
p.recorder = recorder
}
func (p *ConnectorPool) Add(tid relay.TunnelID, c *Connector) {
p.mu.Lock()
defer p.mu.Unlock()
@ -204,15 +189,15 @@ func (p *ConnectorPool) Add(tid relay.TunnelID, c *Connector) {
t := p.tunnels[s]
if t == nil {
t = NewTunnel(tid)
t.WithRecorder(p.recorder)
t = NewTunnel(p.node, tid)
t.WithSD(p.sd)
p.tunnels[s] = t
}
t.AddConnector(c)
}
func (p *ConnectorPool) Get(network string, tid relay.TunnelID) *Connector {
func (p *ConnectorPool) Get(network string, tid string) *Connector {
if p == nil {
return nil
}
@ -220,7 +205,7 @@ func (p *ConnectorPool) Get(network string, tid relay.TunnelID) *Connector {
p.mu.RLock()
defer p.mu.RUnlock()
t := p.tunnels[tid.String()]
t := p.tunnels[tid]
if t == nil {
return nil
}
@ -260,31 +245,3 @@ func parseTunnelID(s string) (tid relay.TunnelID) {
}
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 tid.IsZero() {
err = ErrTunnelID
return
}
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
}