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

add entrypoint for tunnel handler

Browse Source
This commit is contained in:
ginuerzh 2023-10-20 19:12:09 +08:00
parent 1e17320dfb
commit 28824885ab
8 changed files with 499 additions and 24 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

9
bypass/plugin.go
View File

@ -61,8 +61,9 @@ func (p *grpcPlugin) Contains(ctx context.Context, network, addr string, opts ..
&proto.BypassRequest{
Network: network,
Addr: addr,
Host: options.Host,
Client: string(auth_util.IDFromContext(ctx)),
Host: options.Host,
Path: options.Path,
})
if err != nil {
p.log.Error(err)
@ -81,8 +82,9 @@ func (p *grpcPlugin) Close() error {
type httpPluginRequest struct {
Network string `json:"network"`
Addr string `json:"addr"`
Host string `json:"host"`
Client string `json:"client"`
Host string `json:"host"`
Path string `json:"path"`
}
type httpPluginResponse struct {
@ -127,8 +129,9 @@ func (p *httpPlugin) Contains(ctx context.Context, network, addr string, opts ..
rb := httpPluginRequest{
Network: network,
Addr: addr,
Host: options.Host,
Client: string(auth_util.IDFromContext(ctx)),
Host: options.Host,
Path: options.Path,
}
v, err := json.Marshal(&rb)
if err != nil {

20
handler/forward/local/handler.go
View File

@ -14,6 +14,7 @@ import (
"strings"
"time"
"github.com/go-gost/core/bypass"
"github.com/go-gost/core/chain"
"github.com/go-gost/core/handler"
"github.com/go-gost/core/hop"
@ -200,6 +201,21 @@ func (h *forwardHandler) handleHTTP(ctx context.Context, rw io.ReadWriter, remot
return err
}
if log.IsLevelEnabled(logger.TraceLevel) {
dump, _ := httputil.DumpRequest(req, false)
log.Trace(string(dump))
}
host := req.Host
if _, _, err := net.SplitHostPort(host); err != nil {
host = net.JoinHostPort(host, "80")
}
if bp := h.options.Bypass; bp != nil && bp.Contains(ctx, "tcp", host, bypass.WithPathOption(req.RequestURI)) {
log.Debugf("bypass: %s %s", host, req.RequestURI)
resp.StatusCode = http.StatusForbidden
return resp.Write(rw)
}
if addr := getRealClientAddr(req, remoteAddr); addr != remoteAddr {
log = log.WithFields(map[string]any{
"src": addr.String(),
@ -250,10 +266,6 @@ func (h *forwardHandler) handleHTTP(ctx context.Context, rw io.ReadWriter, remot
req.Header.Set(k, v)
}
}
if log.IsLevelEnabled(logger.TraceLevel) {
dump, _ := httputil.DumpRequest(req, false)
log.Trace(string(dump))
}
cc, err = h.router.Dial(ctx, "tcp", target.Addr)
if err != nil {

20
handler/forward/remote/handler.go
View File

@ -14,6 +14,7 @@ import (
"strings"
"time"
"github.com/go-gost/core/bypass"
"github.com/go-gost/core/chain"
"github.com/go-gost/core/handler"
"github.com/go-gost/core/hop"
@ -201,6 +202,21 @@ func (h *forwardHandler) handleHTTP(ctx context.Context, rw io.ReadWriter, remot
return err
}
if log.IsLevelEnabled(logger.TraceLevel) {
dump, _ := httputil.DumpRequest(req, false)
log.Trace(string(dump))
}
host := req.Host
if _, _, err := net.SplitHostPort(host); err != nil {
host = net.JoinHostPort(host, "80")
}
if bp := h.options.Bypass; bp != nil && bp.Contains(ctx, "tcp", host, bypass.WithPathOption(req.RequestURI)) {
log.Debugf("bypass: %s %s", host, req.RequestURI)
resp.StatusCode = http.StatusForbidden
return resp.Write(rw)
}
if addr := getRealClientAddr(req, remoteAddr); addr != remoteAddr {
log = log.WithFields(map[string]any{
"src": addr.String(),
@ -251,10 +267,6 @@ func (h *forwardHandler) handleHTTP(ctx context.Context, rw io.ReadWriter, remot
req.Header.Set(k, v)
}
}
if log.IsLevelEnabled(logger.TraceLevel) {
dump, _ := httputil.DumpRequest(req, false)
log.Trace(string(dump))
}
cc, err = h.router.Dial(ctx, "tcp", target.Addr)
if err != nil {

5
handler/redirect/tcp/handler.go
View File

@ -14,6 +14,7 @@ import (
"strings"
"time"
"github.com/go-gost/core/bypass"
"github.com/go-gost/core/chain"
"github.com/go-gost/core/handler"
"github.com/go-gost/core/logger"
@ -163,8 +164,8 @@ func (h *redirectHandler) handleHTTP(ctx context.Context, rw io.ReadWriter, radd
"host": host,
})
if h.options.Bypass != nil && h.options.Bypass.Contains(ctx, "tcp", host) {
log.Debug("bypass: ", host)
if h.options.Bypass != nil && h.options.Bypass.Contains(ctx, "tcp", host, bypass.WithPathOption(req.RequestURI)) {
log.Debugf("bypass: %s %s", host, req.RequestURI)
return nil
}

5
handler/sni/handler.go
View File

@ -15,6 +15,7 @@ import (
"net/http/httputil"
"time"
"github.com/go-gost/core/bypass"
"github.com/go-gost/core/chain"
"github.com/go-gost/core/handler"
"github.com/go-gost/core/logger"
@ -115,8 +116,8 @@ func (h *sniHandler) handleHTTP(ctx context.Context, rw io.ReadWriter, raddr net
"host": host,
})
if h.options.Bypass != nil && h.options.Bypass.Contains(ctx, "tcp", host) {
log.Debug("bypass: ", host)
if h.options.Bypass != nil && h.options.Bypass.Contains(ctx, "tcp", host, bypass.WithPathOption(req.RequestURI)) {
log.Debugf("bypass: %s %s", host, req.RequestURI)
return nil
}

376
handler/tunnel/entrypoint.go Normal file
View File

@ -0,0 +1,376 @@
package tunnel
import (
"bufio"
"context"
"fmt"
"io"
"net"
"net/http"
"net/http/httputil"
"strconv"
"strings"
"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"
auth_util "github.com/go-gost/x/internal/util/auth"
"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"
metrics "github.com/go-gost/x/metrics/wrapper"
)
type tcpListener struct {
ln net.Listener
options listener.Options
}
func newTCPListener(ln net.Listener, opts ...listener.Option) listener.Listener {
options := listener.Options{}
for _, opt := range opts {
opt(&options)
}
return &tcpListener{
ln: ln,
options: options,
}
}
func (l *tcpListener) Init(md md.Metadata) (err error) {
// l.logger.Debugf("pp: %d", l.options.ProxyProtocol)
ln := l.ln
ln = proxyproto.WrapListener(l.options.ProxyProtocol, ln, 10*time.Second)
ln = metrics.WrapListener(l.options.Service, ln)
ln = admission.WrapListener(l.options.Admission, ln)
ln = limiter.WrapListener(l.options.TrafficLimiter, ln)
ln = climiter.WrapListener(l.options.ConnLimiter, ln)
l.ln = ln
return
}
func (l *tcpListener) Accept() (conn net.Conn, err error) {
return l.ln.Accept()
}
func (l *tcpListener) Addr() net.Addr {
return l.ln.Addr()
}
func (l *tcpListener) Close() error {
return l.ln.Close()
}
type tcpHandler struct {
session *mux.Session
options handler.Options
}
func newTCPHandler(session *mux.Session, opts ...handler.Option) handler.Handler {
options := handler.Options{}
for _, opt := range opts {
opt(&options)
}
return &tcpHandler{
session: session,
options: options,
}
}
func (h *tcpHandler) Init(md md.Metadata) (err error) {
return
}
func (h *tcpHandler) 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())
}()
cc, err := h.session.GetConn()
if err != nil {
log.Error(err)
return err
}
defer cc.Close()
af := &relay.AddrFeature{}
af.ParseFrom(conn.RemoteAddr().String())
resp := relay.Response{
Version: relay.Version1,
Status: relay.StatusOK,
Features: []relay.Feature{af},
}
if _, err := resp.WriteTo(cc); err != nil {
log.Error(err)
return err
}
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
}
type entrypointHandler struct {
pool *ConnectorPool
ingress ingress.Ingress
options handler.Options
}
func newEntrypointHandler(pool *ConnectorPool, ingress ingress.Ingress, opts ...handler.Option) handler.Handler {
options := handler.Options{}
for _, opt := range opts {
opt(&options)
}
return &entrypointHandler{
pool: pool,
ingress: ingress,
options: options,
}
}
func (h *entrypointHandler) Init(md md.Metadata) (err error) {
return
}
func (h *entrypointHandler) 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 *entrypointHandler) 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
}
if log.IsLevelEnabled(logger.TraceLevel) {
dump, _ := httputil.DumpRequest(req, false)
log.Trace(string(dump))
}
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(),
})
if addr := getRealClientAddr(req, raddr); addr != raddr {
log = log.WithFields(map[string]any{
"src": addr.String(),
})
raddr = addr
}
ctx = auth_util.ContextWithClientAddr(ctx, auth_util.ClientAddr(raddr.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()) // src address
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) // dst address
features = append(features, af)
}
(&relay.Response{
Version: relay.Version1,
Status: relay.StatusOK,
Features: features,
}).WriteTo(cc)
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
}
func getRealClientAddr(req *http.Request, raddr net.Addr) net.Addr {
if req == nil {
return nil
}
// cloudflare CDN
sip := req.Header.Get("CF-Connecting-IP")
if sip == "" {
ss := strings.Split(req.Header.Get("X-Forwarded-For"), ",")
if len(ss) > 0 && ss[0] != "" {
sip = ss[0]
}
}
if sip == "" {
sip = req.Header.Get("X-Real-Ip")
}
ip := net.ParseIP(sip)
if ip == nil {
return raddr
}
_, sp, _ := net.SplitHostPort(raddr.String())
port, _ := strconv.Atoi(sp)
return &net.TCPAddr{
IP: ip,
Port: port,
}
}

66
handler/tunnel/handler.go
View File

@ -3,18 +3,23 @@ package tunnel
import (
"context"
"errors"
"fmt"
"net"
"strconv"
"time"
"github.com/go-gost/core/chain"
"github.com/go-gost/core/handler"
"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"
)
var (
@ -33,6 +38,7 @@ type tunnelHandler struct {
router *chain.Router
md metadata
options handler.Options
ep service.Service
pool *ConnectorPool
recorder recorder.RecorderObject
}
@ -68,9 +74,69 @@ func (h *tunnelHandler) Init(md md.Metadata) (err error) {
}
}
if err = h.initEntrypoint(); err != nil {
return
}
return nil
}
func (h *tunnelHandler) 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": "tunnel-ep",
"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 := newEntrypointHandler(
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
}
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{

22
handler/tunnel/metadata.go
View File

@ -15,23 +15,25 @@ import (
)
type metadata struct {
readTimeout time.Duration
noDelay bool
hash string
directTunnel bool
entryPointID relay.TunnelID
ingress ingress.Ingress
muxCfg *mux.Config
readTimeout time.Duration
noDelay bool
hash string
directTunnel bool
entryPoint string
entryPointProxyProtocol int
entryPointID relay.TunnelID
ingress ingress.Ingress
muxCfg *mux.Config
}
func (h *tunnelHandler) parseMetadata(md mdata.Metadata) (err error) {
h.md.readTimeout = mdutil.GetDuration(md, "readTimeout")
h.md.noDelay = mdutil.GetBool(md, "nodelay")
h.md.hash = mdutil.GetString(md, "hash")
h.md.directTunnel = mdutil.GetBool(md, "tunnel.direct")
h.md.entryPoint = mdutil.GetString(md, "entrypoint")
h.md.entryPointID = parseTunnelID(mdutil.GetString(md, "entrypoint.id"))
h.md.entryPointProxyProtocol = mdutil.GetInt(md, "entrypoint.ProxyProtocol")
h.md.ingress = registry.IngressRegistry().Get(mdutil.GetString(md, "ingress"))
if h.md.ingress == nil {
@ -66,5 +68,7 @@ func (h *tunnelHandler) parseMetadata(md mdata.Metadata) (err error) {
MaxStreamBuffer: mdutil.GetInt(md, "mux.maxStreamBuffer"),
}
h.md.hash = mdutil.GetString(md, "hash")
return
}