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

add transport for tunnel entrypoint

Browse Source
This commit is contained in:
ginuerzh
2024-10-15 18:47:36 +08:00
parent 5bba004324
commit 51ca9f620a
8 changed files with 357 additions and 355 deletions
Download Patch File Download Diff File Expand all files Collapse all files
ctx/value.go
+17
View File
@@ -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
}
handler/tunnel/connect.go
+1 -1
View File
@@ -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 {
handler/tunnel/entrypoint.go
+320 -21
View File
@@ -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)
}
handler/tunnel/handler.go
+10 -2
View File
@@ -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
handler/tunnel/metadata.go
+6 -1
View File
@@ -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
handler/tunnel/sniffing.go
-328
View File
@@ -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
}
internal/util/forwarder/sniffer.go
+2 -1
View File
@@ -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)
}
internal/util/sniffing/sniffer.go
+1 -1
View File
@@ -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)
}