diff --git a/handler/http/handler.go b/handler/http/handler.go index b923b81c..d09ee80e 100644 --- a/handler/http/handler.go +++ b/handler/http/handler.go @@ -241,7 +241,7 @@ func (h *httpHandler) handleRequest(ctx context.Context, conn net.Conn, req *htt clientID, ok := h.authenticate(ctx, conn, req, resp, log) if !ok { - return errors.New("authenication failed") + return errors.New("authentication failed") } ctx = ctxvalue.ContextWithClientID(ctx, ctxvalue.ClientID(clientID)) diff --git a/handler/tunnel/entrypoint.go b/handler/tunnel/entrypoint.go index d7ad1ed9..352ccde7 100644 --- a/handler/tunnel/entrypoint.go +++ b/handler/tunnel/entrypoint.go @@ -12,39 +12,69 @@ 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/observer/stats" "github.com/go-gost/core/recorder" "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" + ctxvalue "github.com/go-gost/x/ctx" xnet "github.com/go-gost/x/internal/net" "github.com/go-gost/x/internal/net/proxyproto" climiter "github.com/go-gost/x/limiter/conn/wrapper" metrics "github.com/go-gost/x/metrics/wrapper" + stats_wrapper "github.com/go-gost/x/observer/stats/wrapper" xrecorder "github.com/go-gost/x/recorder" ) type entrypoint struct { - node string - service string - pool *ConnectorPool - ingress ingress.Ingress - sd sd.SD - log logger.Logger - recorder recorder.RecorderObject + node string + service string + pool *ConnectorPool + ingress ingress.Ingress + sd sd.SD + log logger.Logger + recorder recorder.RecorderObject + keepalive bool + readTimeout time.Duration } -func (ep *entrypoint) handle(ctx context.Context, conn net.Conn) error { +func (ep *entrypoint) handle(ctx context.Context, conn net.Conn) (err error) { defer conn.Close() start := time.Now() + + ro := &xrecorder.HandlerRecorderObject{ + Node: ep.node, + Service: ep.service, + RemoteAddr: conn.RemoteAddr().String(), + LocalAddr: conn.LocalAddr().String(), + Network: "tcp", + Time: start, + SID: string(ctxvalue.SidFromContext(ctx)), + } + ro.ClientIP, _, _ = net.SplitHostPort(conn.RemoteAddr().String()) + log := ep.log.WithFields(map[string]any{ "remote": conn.RemoteAddr().String(), "local": conn.LocalAddr().String(), + "sid": ro.SID, }) - log.Infof("%s <> %s", conn.RemoteAddr(), conn.LocalAddr()) + + pStats := stats.Stats{} + conn = stats_wrapper.WrapConn(conn, &pStats) + defer func() { + if err != nil { + ro.Err = err.Error() + } + ro.InputBytes = pStats.Get(stats.KindInputBytes) + ro.OutputBytes = pStats.Get(stats.KindOutputBytes) + ro.Duration = time.Since(start) + if err := ro.Record(ctx, ep.recorder.Recorder); err != nil { + log.Errorf("record: %v", err) + } + log.WithFields(map[string]any{ "duration": time.Since(start), }).Infof("%s >< %s", conn.RemoteAddr(), conn.LocalAddr()) @@ -56,21 +86,13 @@ func (ep *entrypoint) handle(ctx context.Context, conn net.Conn) error { if err != nil { return err } + + conn = xnet.NewReadWriteConn(br, conn, conn) if v[0] == relay.Version1 { - return ep.handleConnect(ctx, xnet.NewReadWriteConn(br, conn, conn), log) + return ep.handleConnect(ctx, conn, log) } - ro := &xrecorder.HandlerRecorderObject{ - Node: ep.node, - Service: ep.service, - RemoteAddr: conn.RemoteAddr().String(), - LocalAddr: conn.LocalAddr().String(), - Network: "tcp", - Time: start, - } - ro.ClientIP, _, _ = net.SplitHostPort(conn.RemoteAddr().String()) - - return ep.handleHTTP(ctx, xio.NewReadWriter(br, conn), ro, log) + return ep.handleHTTP(ctx, conn, ro, log) } func (ep *entrypoint) handleConnect(ctx context.Context, conn net.Conn, log logger.Logger) error { diff --git a/handler/tunnel/handler.go b/handler/tunnel/handler.go index 5ca3eb30..3d4d4701 100644 --- a/handler/tunnel/handler.go +++ b/handler/tunnel/handler.go @@ -76,7 +76,6 @@ func (h *tunnelHandler) Init(md md.Metadata) (err error) { h.log = h.options.Logger.WithFields(map[string]any{ "node": h.id, }) - h.pool = NewConnectorPool(h.id, h.md.sd) h.ep = &entrypoint{ @@ -88,17 +87,18 @@ func (h *tunnelHandler) Init(md md.Metadata) (err error) { log: h.log.WithFields(map[string]any{ "kind": "entrypoint", }), + keepalive: h.md.entryPointKeepalive, + readTimeout: h.md.entryPointReadTimeout, } - if err = h.initEntrypoint(); err != nil { - return - } - 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 diff --git a/handler/tunnel/metadata.go b/handler/tunnel/metadata.go index 1130e879..8d778442 100644 --- a/handler/tunnel/metadata.go +++ b/handler/tunnel/metadata.go @@ -7,11 +7,11 @@ import ( "github.com/go-gost/core/ingress" "github.com/go-gost/core/logger" mdata "github.com/go-gost/core/metadata" - mdutil "github.com/go-gost/x/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" + mdutil "github.com/go-gost/x/metadata/util" "github.com/go-gost/x/registry" ) @@ -20,29 +20,39 @@ const ( ) type metadata struct { - readTimeout time.Duration + readTimeout time.Duration + entryPoint string entryPointID relay.TunnelID entryPointProxyProtocol int - directTunnel bool - tunnelTTL time.Duration - ingress ingress.Ingress - sd sd.SD - muxCfg *mux.Config - observePeriod time.Duration + entryPointKeepalive bool + entryPointReadTimeout time.Duration + + directTunnel bool + tunnelTTL time.Duration + ingress ingress.Ingress + sd sd.SD + muxCfg *mux.Config + observePeriod time.Duration } func (h *tunnelHandler) parseMetadata(md mdata.Metadata) (err error) { h.md.readTimeout = mdutil.GetDuration(md, "readTimeout") + 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.entryPointKeepalive = mdutil.GetBool(md, "entrypoint.keepalive") + h.md.entryPointReadTimeout = mdutil.GetDuration(md, "entrypoint.readTimeout") + if h.md.entryPointReadTimeout <= 0 { + h.md.entryPointReadTimeout = 15 * time.Second + } + h.md.tunnelTTL = mdutil.GetDuration(md, "tunnel.ttl") if h.md.tunnelTTL <= 0 { h.md.tunnelTTL = defaultTTL } 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 { diff --git a/handler/tunnel/sniffing.go b/handler/tunnel/sniffing.go index dc9dced7..1faa15c9 100644 --- a/handler/tunnel/sniffing.go +++ b/handler/tunnel/sniffing.go @@ -13,10 +13,13 @@ import ( "time" "github.com/go-gost/core/logger" + "github.com/go-gost/core/observer/stats" "github.com/go-gost/relay" ctxvalue "github.com/go-gost/x/ctx" + xio "github.com/go-gost/x/internal/io" xnet "github.com/go-gost/x/internal/net" xhttp "github.com/go-gost/x/internal/net/http" + stats_wrapper "github.com/go-gost/x/observer/stats/wrapper" xrecorder "github.com/go-gost/x/recorder" ) @@ -24,8 +27,11 @@ const ( defaultBodySize = 1024 * 1024 // 1MB ) -func (h *entrypoint) handleHTTP(ctx context.Context, rw io.ReadWriter, ro *xrecorder.HandlerRecorderObject, log logger.Logger) error { - br := bufio.NewReader(rw) +func (h *entrypoint) handleHTTP(ctx context.Context, conn net.Conn, ro *xrecorder.HandlerRecorderObject, log logger.Logger) error { + pStats := stats.Stats{} + conn = stats_wrapper.WrapConn(conn, &pStats) + + br := bufio.NewReader(conn) req, err := http.ReadRequest(br) if err != nil { return err @@ -36,12 +42,85 @@ func (h *entrypoint) handleHTTP(ctx context.Context, rw io.ReadWriter, ro *xreco log.Trace(string(dump)) } - shouldClose, err := h.httpRoundTrip(ctx, rw, req, ro, log) + ro.HTTP = &xrecorder.HTTPRecorderObject{ + Host: req.Host, + Proto: req.Proto, + Scheme: req.URL.Scheme, + Method: req.Method, + URI: req.RequestURI, + Request: xrecorder.HTTPRequestRecorderObject{ + ContentLength: req.ContentLength, + Header: req.Header.Clone(), + }, + } + + if clientIP := xhttp.GetClientIP(req); clientIP != nil { + ro.ClientIP = clientIP.String() + } + + clientAddr := ro.RemoteAddr + if ro.ClientIP != "" { + if _, port, _ := net.SplitHostPort(ro.RemoteAddr); port != "" { + clientAddr = net.JoinHostPort(ro.ClientIP, port) + } + } + ctx = ctxvalue.ContextWithClientAddr(ctx, ctxvalue.ClientAddr(clientAddr)) + + res := &http.Response{ + ProtoMajor: req.ProtoMajor, + ProtoMinor: req.ProtoMinor, + Header: http.Header{}, + StatusCode: http.StatusServiceUnavailable, + } + ro.HTTP.StatusCode = res.StatusCode + + host := req.Host + if _, _, err := net.SplitHostPort(host); err != nil { + host = net.JoinHostPort(strings.Trim(host, "[]"), "80") + } + + cc, node, tunnel, err := h.dial(ctx, conn, host, res, ro, log) + if err != nil { + log.Error(err) + return err + } + defer cc.Close() + + log = log.WithFields(map[string]any{ + "host": host, + "tunnel": tunnel, + "clientIP": ro.ClientIP, + }) + + if node == h.node { + var features []relay.Feature + af := &relay.AddrFeature{} + af.ParseFrom(clientAddr) + features = append(features, af) // src address + + af = &relay.AddrFeature{} + af.ParseFrom(host) + features = append(features, af) // dst address + + if _, err := (&relay.Response{ + Version: relay.Version1, + Status: relay.StatusOK, + Features: features, + }).WriteTo(cc); err != nil { + log.Error(err) + res.Write(conn) + return err + } + } + + shouldClose, err := h.httpRoundTrip(ctx, conn, cc, req, ro, &pStats, log) if err != nil || shouldClose { return err } for { + pStats.Reset() + req, err := http.ReadRequest(br) if err != nil { if errors.Is(err, io.EOF) || errors.Is(err, net.ErrClosed) { @@ -55,30 +134,86 @@ func (h *entrypoint) handleHTTP(ctx context.Context, rw io.ReadWriter, ro *xreco log.Trace(string(dump)) } - if shouldClose, err := h.httpRoundTrip(ctx, rw, req, ro, log); err != nil || shouldClose { + if shouldClose, err := h.httpRoundTrip(ctx, conn, cc, req, ro, &pStats, log); err != nil || shouldClose { return err } } } -func (h *entrypoint) httpRoundTrip(ctx context.Context, rw io.ReadWriter, req *http.Request, ro *xrecorder.HandlerRecorderObject, log logger.Logger) (close bool, err error) { +func (h *entrypoint) dial(ctx context.Context, conn net.Conn, host string, res *http.Response, ro *xrecorder.HandlerRecorderObject, log logger.Logger) (cc net.Conn, node, tunnel string, err error) { + var tunnelID relay.TunnelID + if h.ingress != nil { + if rule := h.ingress.GetRule(ctx, host); rule != nil { + tunnelID = parseTunnelID(rule.Endpoint) + } + } + + if tunnelID.IsZero() { + err = fmt.Errorf("no route to host %s", host) + res.StatusCode = http.StatusBadGateway + ro.HTTP.StatusCode = res.StatusCode + res.Write(conn) + return + } + + ro.ClientID = tunnelID.String() + + if tunnelID.IsPrivate() { + err = fmt.Errorf("access denied: tunnel %s is private for host %s", tunnelID, host) + res.StatusCode = http.StatusBadGateway + ro.HTTP.StatusCode = res.StatusCode + res.Write(conn) + return + } + + tunnel = tunnelID.String() + + log = log.WithFields(map[string]any{ + "host": host, + "tunnel": tunnel, + "clientIP": ro.ClientIP, + }) + + d := &Dialer{ + node: h.node, + pool: h.pool, + sd: h.sd, + retry: 3, + timeout: 15 * time.Second, + log: log, + } + cc, node, cid, err := d.Dial(ctx, "tcp", tunnel) + if err != nil { + res.Write(conn) + return + } + log.Debugf("new connection to tunnel: %s, connector: %s", tunnel, cid) + + return +} + +func (h *entrypoint) httpRoundTrip(ctx context.Context, rw, cc io.ReadWriter, req *http.Request, ro *xrecorder.HandlerRecorderObject, pStats *stats.Stats, log logger.Logger) (close bool, err error) { close = true - start := time.Now() - ro.Time = start + ro2 := &xrecorder.HandlerRecorderObject{} + *ro2 = *ro + ro = ro2 + ro.Time = time.Now() log.Infof("%s <-> %s", ro.RemoteAddr, req.Host) defer func() { - ro.Duration = time.Since(start) if err != nil { ro.Err = err.Error() } + ro.InputBytes = pStats.Get(stats.KindInputBytes) + ro.OutputBytes = pStats.Get(stats.KindOutputBytes) + ro.Duration = time.Since(ro.Time) if err := ro.Record(ctx, h.recorder.Recorder); err != nil { log.Errorf("record: %v", err) } log.WithFields(map[string]any{ - "duration": time.Since(start), + "duration": time.Since(ro.Time), }).Infof("%s >-< %s", ro.RemoteAddr, req.Host) }() @@ -93,15 +228,12 @@ func (h *entrypoint) httpRoundTrip(ctx context.Context, rw io.ReadWriter, req *h Header: req.Header.Clone(), }, } - if clientIP := xhttp.GetClientIP(req); clientIP != nil { - ro.ClientIP = clientIP.String() - } - clientAddr := ro.RemoteAddr - if ro.ClientIP != "" { - if _, port, _ := net.SplitHostPort(ro.RemoteAddr); port != "" { - clientAddr = net.JoinHostPort(ro.ClientIP, port) - } - ctx = ctxvalue.ContextWithClientAddr(ctx, ctxvalue.ClientAddr(clientAddr)) + + res := &http.Response{ + ProtoMajor: req.ProtoMajor, + ProtoMinor: req.ProtoMinor, + Header: http.Header{}, + StatusCode: http.StatusServiceUnavailable, } // HTTP/1.0 @@ -113,87 +245,8 @@ func (h *entrypoint) httpRoundTrip(ctx context.Context, rw io.ReadWriter, req *h } } - res := &http.Response{ - ProtoMajor: req.ProtoMajor, - ProtoMinor: req.ProtoMinor, - Header: http.Header{}, - StatusCode: http.StatusServiceUnavailable, - } - - host := req.Host - if _, _, err := net.SplitHostPort(host); err != nil { - host = net.JoinHostPort(strings.Trim(host, "[]"), "80") - } - - var tunnelID relay.TunnelID - if h.ingress != nil { - if rule := h.ingress.GetRule(ctx, req.Host); rule != nil { - tunnelID = parseTunnelID(rule.Endpoint) - } - } - if tunnelID.IsZero() { - err = fmt.Errorf("no route to host %s", req.Host) - log.Error(err) - res.StatusCode = http.StatusBadGateway - ro.HTTP.StatusCode = res.StatusCode - res.Write(rw) - return - } - - ro.ClientID = tunnelID.String() - - if tunnelID.IsPrivate() { - err = fmt.Errorf("access denied: tunnel %s is private for host %s", tunnelID, req.Host) - log.Error(err) - res.StatusCode = http.StatusBadGateway - ro.HTTP.StatusCode = res.StatusCode - res.Write(rw) - return - } - - log = log.WithFields(map[string]any{ - "host": req.Host, - "tunnel": tunnelID.String(), - "clientIP": ro.ClientIP, - }) - - d := &Dialer{ - node: h.node, - pool: h.pool, - sd: h.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) - res.Write(rw) - return - } - // TODO: re-use connection - defer cc.Close() - log.Debugf("new connection to tunnel: %s, connector: %s", tunnelID, cid) - - if node == h.node { - var features []relay.Feature - af := &relay.AddrFeature{} - af.ParseFrom(clientAddr) - features = append(features, af) // src address - - af = &relay.AddrFeature{} - af.ParseFrom(host) - features = append(features, af) // dst address - - if _, err = (&relay.Response{ - Version: relay.Version1, - Status: relay.StatusOK, - Features: features, - }).WriteTo(cc); err != nil { - log.Error(err) - res.Write(rw) - return - } + if !h.keepalive { + req.Header.Set("Connection", "close") } var reqBody *xhttp.Body @@ -219,7 +272,7 @@ func (h *entrypoint) httpRoundTrip(ctx context.Context, rw io.ReadWriter, req *h ro.HTTP.Request.ContentLength = reqBody.Length() } - cc.SetReadDeadline(time.Now().Add(15 * time.Second)) + xio.SetReadDeadline(cc, time.Now().Add(h.readTimeout)) resp, err := http.ReadResponse(bufio.NewReader(cc), req) if err != nil { log.Errorf("read response: %v", err) @@ -227,7 +280,7 @@ func (h *entrypoint) httpRoundTrip(ctx context.Context, rw io.ReadWriter, req *h return } defer resp.Body.Close() - cc.SetReadDeadline(time.Time{}) + xio.SetReadDeadline(cc, time.Time{}) ro.HTTP.StatusCode = resp.StatusCode ro.HTTP.Response.Header = resp.Header diff --git a/internal/util/forwarder/sniffer.go b/internal/util/forwarder/sniffer.go index 3ceae946..23f503f5 100644 --- a/internal/util/forwarder/sniffer.go +++ b/internal/util/forwarder/sniffer.go @@ -136,6 +136,18 @@ func (h *Sniffer) HandleHTTP(ctx context.Context, conn net.Conn, opts ...HandleO } ro := ho.RecorderObject + ro.HTTP = &xrecorder.HTTPRecorderObject{ + Host: req.Host, + Proto: req.Proto, + Scheme: req.URL.Scheme, + Method: req.Method, + URI: req.RequestURI, + Request: xrecorder.HTTPRequestRecorderObject{ + ContentLength: req.ContentLength, + Header: req.Header.Clone(), + }, + } + if clientIP := xhttp.GetClientIP(req); clientIP != nil { ro.ClientIP = clientIP.String() } @@ -436,6 +448,7 @@ func (h *Sniffer) httpRoundTrip(ctx context.Context, rw, cc io.ReadWriter, node } if err = req.Write(cc); err != nil { + res.Write(rw) return } @@ -448,6 +461,7 @@ func (h *Sniffer) httpRoundTrip(ctx context.Context, rw, cc io.ReadWriter, node resp, err := http.ReadResponse(bufio.NewReader(cc), req) if err != nil { log.Errorf("read response: %v", err) + res.Write(rw) return } defer resp.Body.Close() diff --git a/internal/util/sniffing/sniffer.go b/internal/util/sniffing/sniffer.go index b7a00c3a..b37365cc 100644 --- a/internal/util/sniffing/sniffer.go +++ b/internal/util/sniffing/sniffer.go @@ -119,6 +119,18 @@ func (h *Sniffer) HandleHTTP(ctx context.Context, conn net.Conn, opts ...HandleO } ro := ho.RecorderObject + ro.HTTP = &xrecorder.HTTPRecorderObject{ + Host: req.Host, + Proto: req.Proto, + Scheme: req.URL.Scheme, + Method: req.Method, + URI: req.RequestURI, + Request: xrecorder.HTTPRequestRecorderObject{ + ContentLength: req.ContentLength, + Header: req.Header.Clone(), + }, + } + if clientIP := xhttp.GetClientIP(req); clientIP != nil { ro.ClientIP = clientIP.String() }