diff --git a/ctx/value.go b/ctx/value.go index 1e1f6c3b..349aa089 100644 --- a/ctx/value.go +++ b/ctx/value.go @@ -3,6 +3,8 @@ package ctx import ( "bytes" "context" + + "github.com/go-gost/core/logger" ) // clientAddrKey saves the client address. @@ -92,3 +94,18 @@ func BufferFromContext(ctx context.Context) *bytes.Buffer { v, _ := ctx.Value(keyBuffer).(*bytes.Buffer) return v } + +type loggerKey struct{} + +var ( + keyLogger = &loggerKey{} +) + +func ContextWithLogger(ctx context.Context, log logger.Logger) context.Context { + return context.WithValue(ctx, keyLogger, log) +} + +func LoggerFromContext(ctx context.Context) logger.Logger { + v, _ := ctx.Value(keyLogger).(logger.Logger) + return v +} diff --git a/handler/tunnel/connect.go b/handler/tunnel/connect.go index 21a22b47..4682a6d9 100644 --- a/handler/tunnel/connect.go +++ b/handler/tunnel/connect.go @@ -35,7 +35,7 @@ func (h *tunnelHandler) handleConnect(ctx context.Context, req *relay.Request, c // client is a public entrypoint. if tunnelID.Equal(h.md.entryPointID) { resp.WriteTo(conn) - return h.ep.handle(ctx, conn) + return h.ep.Handle(ctx, conn) } if !h.md.directTunnel { diff --git a/handler/tunnel/entrypoint.go b/handler/tunnel/entrypoint.go index 352ccde7..49552bee 100644 --- a/handler/tunnel/entrypoint.go +++ b/handler/tunnel/entrypoint.go @@ -3,8 +3,14 @@ package tunnel import ( "bufio" "context" + "errors" + "fmt" + "io" "net" + "net/http" + "net/http/httputil" "strconv" + "strings" "time" "github.com/go-gost/core/handler" @@ -19,29 +25,47 @@ import ( admission "github.com/go-gost/x/admission/wrapper" ctxvalue "github.com/go-gost/x/ctx" xnet "github.com/go-gost/x/internal/net" + xhttp "github.com/go-gost/x/internal/net/http" "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" + "golang.org/x/net/http/httpguts" ) -type entrypoint struct { - node string - service string - pool *ConnectorPool - ingress ingress.Ingress - sd sd.SD - log logger.Logger - recorder recorder.RecorderObject - keepalive bool - readTimeout time.Duration +const ( + defaultBodySize = 1024 * 1024 // 1MB +) + +type recorderObjectCtxKey struct{} + +var ( + ctxKeyRecorderObject = &recorderObjectCtxKey{} +) + +func contextWithRecorderObject(ctx context.Context, ro *xrecorder.HandlerRecorderObject) context.Context { + return context.WithValue(ctx, ctxKeyRecorderObject, ro) } -func (ep *entrypoint) handle(ctx context.Context, conn net.Conn) (err error) { - defer conn.Close() +func recorderObjectFromContext(ctx context.Context) *xrecorder.HandlerRecorderObject { + v, _ := ctx.Value(ctxKeyRecorderObject).(*xrecorder.HandlerRecorderObject) + return v +} - start := time.Now() +type entrypoint struct { + node string + service string + pool *ConnectorPool + ingress ingress.Ingress + sd sd.SD + log logger.Logger + recorder recorder.RecorderObject + transport http.RoundTripper +} + +func (ep *entrypoint) Handle(ctx context.Context, conn net.Conn) (err error) { + defer conn.Close() ro := &xrecorder.HandlerRecorderObject{ Node: ep.node, @@ -49,7 +73,7 @@ func (ep *entrypoint) handle(ctx context.Context, conn net.Conn) (err error) { RemoteAddr: conn.RemoteAddr().String(), LocalAddr: conn.LocalAddr().String(), Network: "tcp", - Time: start, + Time: time.Now(), SID: string(ctxvalue.SidFromContext(ctx)), } ro.ClientIP, _, _ = net.SplitHostPort(conn.RemoteAddr().String()) @@ -70,18 +94,17 @@ func (ep *entrypoint) handle(ctx context.Context, conn net.Conn) (err error) { } ro.InputBytes = pStats.Get(stats.KindInputBytes) ro.OutputBytes = pStats.Get(stats.KindOutputBytes) - ro.Duration = time.Since(start) + ro.Duration = time.Since(ro.Time) if err := ro.Record(ctx, ep.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", conn.RemoteAddr(), conn.LocalAddr()) }() br := bufio.NewReader(conn) - v, err := br.Peek(1) if err != nil { return err @@ -89,13 +112,287 @@ func (ep *entrypoint) handle(ctx context.Context, conn net.Conn) (err error) { conn = xnet.NewReadWriteConn(br, conn, conn) if v[0] == relay.Version1 { - return ep.handleConnect(ctx, conn, log) + return ep.handleConnect(ctx, conn, ro, log) } - return ep.handleHTTP(ctx, conn, ro, log) } -func (ep *entrypoint) handleConnect(ctx context.Context, conn net.Conn, log logger.Logger) error { +func (ep *entrypoint) handleHTTP(ctx context.Context, conn net.Conn, ro *xrecorder.HandlerRecorderObject, log logger.Logger) (err error) { + pStats := stats.Stats{} + conn = stats_wrapper.WrapConn(conn, &pStats) + + br := bufio.NewReader(conn) + req, err := http.ReadRequest(br) + if err != nil { + return err + } + + if log.IsLevelEnabled(logger.TraceLevel) { + dump, _ := httputil.DumpRequest(req, false) + log.Trace(string(dump)) + } + + 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(), + }, + } + + ro.Time = time.Time{} + + if err := ep.httpRoundTrip(ctx, conn, req, ro, &pStats, log); err != nil { + log.Error(err) + return err + } + + for { + pStats.Reset() + + req, err := http.ReadRequest(br) + if err != nil { + if errors.Is(err, io.EOF) || errors.Is(err, net.ErrClosed) { + return nil + } + return err + } + + if log.IsLevelEnabled(logger.TraceLevel) { + dump, _ := httputil.DumpRequest(req, false) + log.Trace(string(dump)) + } + + if err := ep.httpRoundTrip(ctx, conn, req, ro, &pStats, log); err != nil { + return err + } + } +} + +func (ep *entrypoint) httpRoundTrip(ctx context.Context, rw io.ReadWriter, req *http.Request, ro *xrecorder.HandlerRecorderObject, pStats *stats.Stats, log logger.Logger) (err error) { + ro2 := &xrecorder.HandlerRecorderObject{} + *ro2 = *ro + ro = ro2 + + ro.Time = time.Now() + log.Infof("%s <-> %s", ro.RemoteAddr, req.Host) + 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(ro.Time) + if err := ro.Record(ctx, ep.recorder.Recorder); err != nil { + log.Errorf("record: %v", err) + } + + log.WithFields(map[string]any{ + "duration": time.Since(ro.Time), + }).Infof("%s >-< %s", ro.RemoteAddr, req.Host) + }() + + if req.URL.Scheme == "" { + req.URL.Scheme = "http" + } + if req.URL.Host == "" { + req.URL.Host = req.Host + } + + 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) + } + } + + res := &http.Response{ + ProtoMajor: req.ProtoMajor, + ProtoMinor: req.ProtoMinor, + Header: http.Header{}, + StatusCode: http.StatusServiceUnavailable, + } + ro.HTTP.StatusCode = res.StatusCode + + var reqBody *xhttp.Body + if opts := ep.recorder.Options; opts != nil && opts.HTTPBody { + if req.Body != nil { + maxSize := opts.MaxBodySize + if maxSize <= 0 { + maxSize = defaultBodySize + } + reqBody = xhttp.NewBody(req.Body, maxSize) + req.Body = reqBody + } + } + + ctx = ctxvalue.ContextWithClientAddr(ctx, ctxvalue.ClientAddr(clientAddr)) + ctx = contextWithRecorderObject(ctx, ro) + ctx = ctxvalue.ContextWithLogger(ctx, log) + + resp, err := ep.transport.RoundTrip(req.WithContext(ctx)) + if err != nil { + if errors.Is(err, ErrTunnelRoute) || errors.Is(err, ErrPrivateTunnel) { + res.StatusCode = http.StatusBadGateway + ro.HTTP.StatusCode = http.StatusBadGateway + } + res.Write(rw) + return + } + defer resp.Body.Close() + + if reqBody != nil { + ro.HTTP.Request.Body = reqBody.Content() + ro.HTTP.Request.ContentLength = reqBody.Length() + } + + ro.HTTP.StatusCode = resp.StatusCode + ro.HTTP.Response.Header = resp.Header + ro.HTTP.Response.ContentLength = resp.ContentLength + + if log.IsLevelEnabled(logger.TraceLevel) { + dump, _ := httputil.DumpResponse(resp, false) + log.Trace(string(dump)) + } + + var respBody *xhttp.Body + if opts := ep.recorder.Options; opts != nil && opts.HTTPBody { + maxSize := opts.MaxBodySize + if maxSize <= 0 { + maxSize = defaultBodySize + } + respBody = xhttp.NewBody(resp.Body, maxSize) + resp.Body = respBody + } + + if err = resp.Write(rw); err != nil { + log.Errorf("write response: %v", err) + return + } + + if respBody != nil { + ro.HTTP.Response.Body = respBody.Content() + ro.HTTP.Response.ContentLength = respBody.Length() + } + + if resp.StatusCode == http.StatusSwitchingProtocols { + return ep.handleUpgradeResponse(rw, req, resp) + } + + return +} + +func upgradeType(h http.Header) string { + if !httpguts.HeaderValuesContainsToken(h["Connection"], "Upgrade") { + return "" + } + return h.Get("Upgrade") +} + +func (ep *entrypoint) handleUpgradeResponse(rw io.ReadWriter, req *http.Request, res *http.Response) error { + reqUpType := upgradeType(req.Header) + resUpType := upgradeType(res.Header) + if !strings.EqualFold(reqUpType, resUpType) { + return fmt.Errorf("backend tried to switch protocol %q when %q was requested", resUpType, reqUpType) + } + + backConn, ok := res.Body.(io.ReadWriteCloser) + if !ok { + return fmt.Errorf("internal error: 101 switching protocols response with non-writable body") + } + + return xnet.Transport(rw, backConn) +} + +func (h *entrypoint) dial(ctx context.Context, network, addr string) (conn net.Conn, err error) { + var tunnelID relay.TunnelID + if h.ingress != nil { + if rule := h.ingress.GetRule(ctx, addr); rule != nil { + tunnelID = parseTunnelID(rule.Endpoint) + } + } + + log := ctxvalue.LoggerFromContext(ctx) + if log == nil { + log = h.log + } + + if tunnelID.IsZero() { + return nil, fmt.Errorf("%w %s", ErrTunnelRoute, addr) + } + + if ro := recorderObjectFromContext(ctx); ro != nil { + ro.ClientID = tunnelID.String() + } + + if tunnelID.IsPrivate() { + return nil, fmt.Errorf("%w: tunnel %s is private for host %s", ErrPrivateTunnel, tunnelID, addr) + } + + log = log.WithFields(map[string]any{ + "host": addr, + "tunnel": tunnelID.String(), + }) + + d := &Dialer{ + node: h.node, + pool: h.pool, + sd: h.sd, + retry: 3, + timeout: 15 * time.Second, + log: log, + } + conn, node, cid, err := d.Dial(ctx, "tcp", tunnelID.String()) + if err != nil { + return + } + log.Debugf("new connection to tunnel: %s, connector: %s", tunnelID, cid) + + if node == h.node { + clientAddr := ctxvalue.ClientAddrFromContext(ctx) + var features []relay.Feature + af := &relay.AddrFeature{} + af.ParseFrom(string(clientAddr)) + features = append(features, af) // src address + + af = &relay.AddrFeature{} + af.ParseFrom(addr) + features = append(features, af) // dst address + + if _, err = (&relay.Response{ + Version: relay.Version1, + Status: relay.StatusOK, + Features: features, + }).WriteTo(conn); err != nil { + conn.Close() + return nil, err + } + } + + return conn, nil +} + +func (ep *entrypoint) handleConnect(ctx context.Context, conn net.Conn, ro *xrecorder.HandlerRecorderObject, log logger.Logger) (err error) { req := relay.Request{} if _, err := req.ReadFrom(conn); err != nil { return err @@ -137,6 +434,8 @@ func (ep *entrypoint) handleConnect(ctx context.Context, conn net.Conn, log logg return ErrTunnelID } + ro.ClientID = tunnelID.String() + d := Dialer{ pool: ep.pool, retry: 3, @@ -228,5 +527,5 @@ func (h *entrypointHandler) Init(md md.Metadata) (err error) { } func (h *entrypointHandler) Handle(ctx context.Context, conn net.Conn, opts ...handler.HandleOption) error { - return h.ep.handle(ctx, conn) + return h.ep.Handle(ctx, conn) } diff --git a/handler/tunnel/handler.go b/handler/tunnel/handler.go index 3d4d4701..a4af23df 100644 --- a/handler/tunnel/handler.go +++ b/handler/tunnel/handler.go @@ -5,6 +5,7 @@ import ( "errors" "fmt" "net" + "net/http" "strconv" "time" @@ -32,6 +33,8 @@ var ( ErrTunnelID = errors.New("invalid tunnel ID") ErrTunnelNotAvailable = errors.New("tunnel not available") ErrUnauthorized = errors.New("unauthorized") + ErrTunnelRoute = errors.New("no route to host") + ErrPrivateTunnel = errors.New("private tunnel") ) func init() { @@ -87,9 +90,14 @@ 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, } + h.ep.transport = &http.Transport{ + DialContext: h.ep.dial, + IdleConnTimeout: 30 * time.Second, + ResponseHeaderTimeout: h.md.entryPointReadTimeout, + DisableKeepAlives: !h.md.entryPointKeepalive, + } + for _, ro := range h.options.Recorders { if ro.Record == xrecorder.RecorderServiceHandler { h.ep.recorder = ro diff --git a/handler/tunnel/metadata.go b/handler/tunnel/metadata.go index 8d778442..f260f9c5 100644 --- a/handler/tunnel/metadata.go +++ b/handler/tunnel/metadata.go @@ -42,7 +42,12 @@ func (h *tunnelHandler) parseMetadata(md mdata.Metadata) (err error) { 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.entryPointKeepalive = true + if mdutil.IsExists(md, "entrypoint.keepalive") { + 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 diff --git a/handler/tunnel/sniffing.go b/handler/tunnel/sniffing.go deleted file mode 100644 index 1faa15c9..00000000 --- a/handler/tunnel/sniffing.go +++ /dev/null @@ -1,328 +0,0 @@ -package tunnel - -import ( - "bufio" - "context" - "errors" - "fmt" - "io" - "net" - "net/http" - "net/http/httputil" - "strings" - "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" -) - -const ( - defaultBodySize = 1024 * 1024 // 1MB -) - -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 - } - - if log.IsLevelEnabled(logger.TraceLevel) { - dump, _ := httputil.DumpRequest(req, false) - log.Trace(string(dump)) - } - - 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) { - return nil - } - return err - } - - if log.IsLevelEnabled(logger.TraceLevel) { - dump, _ := httputil.DumpRequest(req, false) - log.Trace(string(dump)) - } - - if shouldClose, err := h.httpRoundTrip(ctx, conn, cc, req, ro, &pStats, log); err != nil || shouldClose { - return err - } - } -} - -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 - - ro2 := &xrecorder.HandlerRecorderObject{} - *ro2 = *ro - ro = ro2 - - ro.Time = time.Now() - log.Infof("%s <-> %s", ro.RemoteAddr, req.Host) - 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(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(ro.Time), - }).Infof("%s >-< %s", ro.RemoteAddr, req.Host) - }() - - 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(), - }, - } - - res := &http.Response{ - ProtoMajor: req.ProtoMajor, - ProtoMinor: req.ProtoMinor, - Header: http.Header{}, - StatusCode: http.StatusServiceUnavailable, - } - - // HTTP/1.0 - if req.ProtoMajor == 1 && req.ProtoMinor == 0 { - if strings.ToLower(req.Header.Get("Connection")) == "keep-alive" { - req.Header.Del("Connection") - } else { - req.Header.Set("Connection", "close") - } - } - - if !h.keepalive { - req.Header.Set("Connection", "close") - } - - var reqBody *xhttp.Body - if opts := h.recorder.Options; opts != nil && opts.HTTPBody { - if req.Body != nil { - maxSize := opts.MaxBodySize - if maxSize <= 0 { - maxSize = defaultBodySize - } - reqBody = xhttp.NewBody(req.Body, maxSize) - req.Body = reqBody - } - } - - if err = req.Write(cc); err != nil { - log.Errorf("send request: %v", err) - res.Write(rw) - return - } - - if reqBody != nil { - ro.HTTP.Request.Body = reqBody.Content() - ro.HTTP.Request.ContentLength = reqBody.Length() - } - - 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) - res.Write(rw) - return - } - defer resp.Body.Close() - xio.SetReadDeadline(cc, time.Time{}) - - ro.HTTP.StatusCode = resp.StatusCode - ro.HTTP.Response.Header = resp.Header - ro.HTTP.Response.ContentLength = resp.ContentLength - - if log.IsLevelEnabled(logger.TraceLevel) { - dump, _ := httputil.DumpResponse(resp, false) - log.Trace(string(dump)) - } - - // HTTP/1.0 - if req.ProtoMajor == 1 && req.ProtoMinor == 0 { - if !resp.Close { - resp.Header.Set("Connection", "keep-alive") - } - resp.ProtoMajor = req.ProtoMajor - resp.ProtoMinor = req.ProtoMinor - } - - var respBody *xhttp.Body - if opts := h.recorder.Options; opts != nil && opts.HTTPBody { - maxSize := opts.MaxBodySize - if maxSize <= 0 { - maxSize = defaultBodySize - } - respBody = xhttp.NewBody(resp.Body, maxSize) - resp.Body = respBody - } - - if err = resp.Write(rw); err != nil { - log.Errorf("write response: %v", err) - return - } - - if respBody != nil { - ro.HTTP.Response.Body = respBody.Content() - ro.HTTP.Response.ContentLength = respBody.Length() - } - - if req.Header.Get("Upgrade") == "websocket" { - xnet.Transport(rw, cc) - } - - return resp.Close, nil -} diff --git a/internal/util/forwarder/sniffer.go b/internal/util/forwarder/sniffer.go index 23f503f5..ccdd5f23 100644 --- a/internal/util/forwarder/sniffer.go +++ b/internal/util/forwarder/sniffer.go @@ -176,6 +176,7 @@ func (h *Sniffer) HandleHTTP(ctx context.Context, conn net.Conn, opts ...HandleO log.Debugf("connected to node %s(%s)", node.Name, node.Addr) ro.Time = time.Time{} + shouldClose, err := h.httpRoundTrip(ctx, conn, cc, node, req, &pStats, &ho) if err != nil || shouldClose { return err @@ -510,7 +511,7 @@ func (h *Sniffer) httpRoundTrip(ctx context.Context, rw, cc io.ReadWriter, node ro.HTTP.Response.ContentLength = respBody.Length() } - if req.Header.Get("Upgrade") == "websocket" { + if resp.StatusCode == http.StatusSwitchingProtocols { xnet.Transport(rw, cc) } diff --git a/internal/util/sniffing/sniffer.go b/internal/util/sniffing/sniffer.go index b37365cc..c4369154 100644 --- a/internal/util/sniffing/sniffer.go +++ b/internal/util/sniffing/sniffer.go @@ -365,7 +365,7 @@ func (h *Sniffer) httpRoundTrip(ctx context.Context, rw, cc io.ReadWriter, req * ro.HTTP.Response.ContentLength = respBody.Length() } - if req.Header.Get("Upgrade") == "websocket" { + if resp.StatusCode == http.StatusSwitchingProtocols { xnet.Transport(rw, cc) }