From a0cbee881776b6a6858277a9a68de0bb8af42dce Mon Sep 17 00:00:00 2001 From: ginuerzh Date: Thu, 17 Oct 2024 21:57:20 +0800 Subject: [PATCH] use http.transport for non-connect http tunnel --- handler/http/handler.go | 257 +++++++++++++++++++++----------- handler/http/metadata.go | 5 +- handler/http2/handler.go | 3 +- handler/http3/handler.go | 3 +- handler/redirect/tcp/handler.go | 1 + handler/redirect/udp/handler.go | 3 +- handler/relay/connect.go | 5 +- handler/socks/v4/handler.go | 3 +- handler/socks/v5/connect.go | 5 +- handler/ss/handler.go | 3 +- handler/sshd/handler.go | 5 +- handler/tunnel/connect.go | 1 + handler/tunnel/entrypoint.go | 13 +- handler/unix/handler.go | 1 + metadata/util/util.go | 4 + 15 files changed, 213 insertions(+), 99 deletions(-) diff --git a/handler/http/handler.go b/handler/http/handler.go index b944abbe..0a6d77f0 100644 --- a/handler/http/handler.go +++ b/handler/http/handler.go @@ -40,20 +40,37 @@ import ( stats_wrapper "github.com/go-gost/x/observer/stats/wrapper" xrecorder "github.com/go-gost/x/recorder" "github.com/go-gost/x/registry" + "golang.org/x/net/http/httpguts" ) +type recorderObjectCtxKey struct{} + +var ( + ctxKeyRecorderObject = &recorderObjectCtxKey{} +) + +func contextWithRecorderObject(ctx context.Context, ro *xrecorder.HandlerRecorderObject) context.Context { + return context.WithValue(ctx, ctxKeyRecorderObject, ro) +} + +func recorderObjectFromContext(ctx context.Context) *xrecorder.HandlerRecorderObject { + v, _ := ctx.Value(ctxKeyRecorderObject).(*xrecorder.HandlerRecorderObject) + return v +} + func init() { registry.HandlerRegistry().Register("http", NewHandler) } type httpHandler struct { - md metadata - options handler.Options - stats *stats_util.HandlerStats - limiter traffic.TrafficLimiter - cancel context.CancelFunc - recorder recorder.RecorderObject - certPool tls_util.CertPool + md metadata + options handler.Options + stats *stats_util.HandlerStats + limiter traffic.TrafficLimiter + cancel context.CancelFunc + recorder recorder.RecorderObject + certPool tls_util.CertPool + transport http.RoundTripper } func NewHandler(opts ...handler.Option) handler.Handler { @@ -95,6 +112,13 @@ func (h *httpHandler) Init(md md.Metadata) error { h.certPool = tls_util.NewMemoryCertPool() } + h.transport = &http.Transport{ + DialContext: h.dial, + IdleConnTimeout: 30 * time.Second, + ResponseHeaderTimeout: h.md.readTimeout, + DisableKeepAlives: !h.md.keepalive, + } + return nil } @@ -153,7 +177,8 @@ func (h *httpHandler) Handle(ctx context.Context, conn net.Conn, opts ...handler return rate_limiter.ErrRateLimit } - req, err := http.ReadRequest(bufio.NewReader(conn)) + br := bufio.NewReader(conn) + req, err := http.ReadRequest(br) if err != nil { log.Error(err) return err @@ -164,6 +189,8 @@ func (h *httpHandler) Handle(ctx context.Context, conn net.Conn, opts ...handler ro.ClientIP = clientIP.String() } + conn = xnet.NewReadWriteConn(br, conn, conn) + return h.handleRequest(ctx, conn, req, ro, log) } @@ -205,7 +232,8 @@ func (h *httpHandler) handleRequest(ctx context.Context, conn net.Conn, req *htt ro.Host = addr fields := map[string]any{ - "dst": addr, + "dst": addr, + "host": addr, } if u, _, _ := h.basicProxyAuth(req.Header.Get("Proxy-Authorization")); u != "" { @@ -249,6 +277,18 @@ func (h *httpHandler) handleRequest(ctx context.Context, conn net.Conn, req *htt ro.HTTP.Response.Header = resp.Header }() + if req.Method == "PRI" || + (req.Method != http.MethodConnect && req.URL.Scheme != "http") { + resp.StatusCode = http.StatusBadRequest + + if log.IsLevelEnabled(logger.TraceLevel) { + dump, _ := httputil.DumpResponse(resp, false) + log.Trace(string(dump)) + } + + return resp.Write(conn) + } + clientID, ok := h.authenticate(ctx, conn, req, resp, log) if !ok { return errors.New("authentication failed") @@ -271,38 +311,6 @@ func (h *httpHandler) handleRequest(ctx context.Context, conn net.Conn, req *htt return h.handleUDP(ctx, conn, ro, log) } - if req.Method == "PRI" || - (req.Method != http.MethodConnect && req.URL.Scheme != "http") { - resp.StatusCode = http.StatusBadRequest - - if log.IsLevelEnabled(logger.TraceLevel) { - dump, _ := httputil.DumpResponse(resp, false) - log.Trace(string(dump)) - } - - return resp.Write(conn) - } - - switch h.md.hash { - case "host": - ctx = ctxvalue.ContextWithHash(ctx, &ctxvalue.Hash{Source: addr}) - } - - var buf bytes.Buffer - cc, err := h.options.Router.Dial(ctxvalue.ContextWithBuffer(ctx, &buf), network, addr) - ro.Route = buf.String() - if err != nil { - resp.StatusCode = http.StatusServiceUnavailable - - if log.IsLevelEnabled(logger.TraceLevel) { - dump, _ := httputil.DumpResponse(resp, false) - log.Trace(string(dump)) - } - resp.Write(conn) - return err - } - defer cc.Close() - { rw := traffic_wrapper.WrapReadWriter( h.limiter, @@ -327,12 +335,25 @@ func (h *httpHandler) handleRequest(ctx context.Context, conn net.Conn, req *htt } if req.Method != http.MethodConnect { - ro2 := &xrecorder.HandlerRecorderObject{} - *ro2 = *ro - ro.Time = time.Time{} - return h.handleProxy(ctx, conn, cc, req, ro2, log) + return h.handleProxy(ctx, conn, req, ro, log) } + ctx = contextWithRecorderObject(ctx, ro) + ctx = ctxvalue.ContextWithLogger(ctx, log) + cc, err := h.dial(ctx, "tcp", addr) + + if err != nil { + resp.StatusCode = http.StatusServiceUnavailable + + if log.IsLevelEnabled(logger.TraceLevel) { + dump, _ := httputil.DumpResponse(resp, false) + log.Trace(string(dump)) + } + resp.Write(conn) + return err + } + defer cc.Close() + resp.StatusCode = http.StatusOK resp.Status = "200 Connection established" @@ -404,13 +425,21 @@ func (h *httpHandler) handleRequest(ctx context.Context, conn net.Conn, req *htt return nil } -func (h *httpHandler) handleProxy(ctx context.Context, rw io.ReadWriter, cc net.Conn, req *http.Request, ro *xrecorder.HandlerRecorderObject, log logger.Logger) error { - if shouldClose, err := h.proxyRoundTrip(ctx, rw, cc, req, ro, log); err != nil || shouldClose { +func (h *httpHandler) handleProxy(ctx context.Context, conn net.Conn, req *http.Request, ro *xrecorder.HandlerRecorderObject, log logger.Logger) error { + pStats := stats.Stats{} + conn = stats_wrapper.WrapConn(conn, &pStats) + + ro.Time = time.Time{} + + if err := h.proxyRoundTrip(ctx, conn, req, ro, &pStats, log); err != nil { return err } + br := bufio.NewReader(conn) for { - req, err := http.ReadRequest(bufio.NewReader(rw)) + pStats.Reset() + + req, err := http.ReadRequest(br) if err != nil { if errors.Is(err, io.EOF) || errors.Is(err, net.ErrClosed) { return nil @@ -423,40 +452,47 @@ func (h *httpHandler) handleProxy(ctx context.Context, rw io.ReadWriter, cc net. log.Trace(string(dump)) } - if shouldClose, err := h.proxyRoundTrip(ctx, rw, cc, req, ro, log); err != nil || shouldClose { + if err := h.proxyRoundTrip(ctx, conn, req, ro, &pStats, log); err != nil { return err } } } -func (h *httpHandler) proxyRoundTrip(ctx context.Context, rw io.ReadWriter, cc net.Conn, req *http.Request, ro *xrecorder.HandlerRecorderObject, log logger.Logger) (close bool, err error) { - close = true +func (h *httpHandler) proxyRoundTrip(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 - if req == nil { - return + host := req.Host + if _, port, _ := net.SplitHostPort(host); port == "" { + host = net.JoinHostPort(strings.Trim(host, "[]"), "80") } + ro.Host = host + ro.Time = time.Now() - start := time.Now() - ro.Time = start + log = log.WithFields(map[string]any{ + "host": host, + }) 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), + "inputBytes": ro.InputBytes, + "outputBytes": ro.OutputBytes, }).Infof("%s >-< %s", ro.RemoteAddr, req.Host) }() - if clientIP := xhttp.GetClientIP(req); clientIP != nil { - ro.ClientIP = clientIP.String() - } ro.HTTP = &xrecorder.HTTPRecorderObject{ Host: req.Host, Proto: req.Proto, @@ -468,6 +504,16 @@ func (h *httpHandler) proxyRoundTrip(ctx context.Context, rw io.ReadWriter, cc n 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) + } + } // HTTP/1.0 if req.ProtoMajor == 1 && req.ProtoMinor == 0 { @@ -478,15 +524,19 @@ func (h *httpHandler) proxyRoundTrip(ctx context.Context, rw io.ReadWriter, cc n } } - if !h.md.keepalive { - req.Header.Set("Connection", "close") - } - req.Header.Del("Proxy-Authorization") req.Header.Del("Proxy-Connection") req.Header.Del("Gost-Target") req.Header.Del("X-Gost-Target") + 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 := h.recorder.Options; opts != nil && opts.HTTPBody { if req.Body != nil { @@ -499,34 +549,22 @@ func (h *httpHandler) proxyRoundTrip(ctx context.Context, rw io.ReadWriter, cc n } } - res := &http.Response{ - ProtoMajor: req.ProtoMajor, - ProtoMinor: req.ProtoMinor, - Header: http.Header{}, - StatusCode: http.StatusServiceUnavailable, - } - ro.HTTP.StatusCode = res.StatusCode + ctx = ctxvalue.ContextWithClientAddr(ctx, ctxvalue.ClientAddr(clientAddr)) + ctx = contextWithRecorderObject(ctx, ro) + ctx = ctxvalue.ContextWithLogger(ctx, log) - if err = req.Write(cc); err != nil { + resp, err := h.transport.RoundTrip(req.WithContext(ctx)) + if err != nil { res.Write(rw) return } + defer resp.Body.Close() if reqBody != nil { ro.HTTP.Request.Body = reqBody.Content() ro.HTTP.Request.ContentLength = reqBody.Length() } - cc.SetReadDeadline(time.Now().Add(h.md.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() - cc.SetReadDeadline(time.Time{}) - ro.HTTP.StatusCode = resp.StatusCode ro.HTTP.Response.Header = resp.Header.Clone() ro.HTTP.Response.ContentLength = resp.ContentLength @@ -545,6 +583,10 @@ func (h *httpHandler) proxyRoundTrip(ctx context.Context, rw io.ReadWriter, cc n resp.ProtoMinor = req.ProtoMinor } + if resp.StatusCode == http.StatusSwitchingProtocols { + return h.handleUpgradeResponse(rw, req, resp) + } + var respBody *xhttp.Body if opts := h.recorder.Options; opts != nil && opts.HTTPBody { maxSize := opts.MaxBodySize @@ -556,8 +598,7 @@ func (h *httpHandler) proxyRoundTrip(ctx context.Context, rw io.ReadWriter, cc n } if err = resp.Write(rw); err != nil { - log.Errorf("write response: %v", err) - return + return fmt.Errorf("write response: %v", err) } if respBody != nil { @@ -565,7 +606,53 @@ func (h *httpHandler) proxyRoundTrip(ctx context.Context, rw io.ReadWriter, cc n ro.HTTP.Response.ContentLength = respBody.Length() } - return resp.Close, nil + return +} + +func (h *httpHandler) dial(ctx context.Context, network, addr string) (conn net.Conn, err error) { + switch h.md.hash { + case "host": + ctx = ctxvalue.ContextWithHash(ctx, &ctxvalue.Hash{Source: addr}) + } + + if log := ctxvalue.LoggerFromContext(ctx); log != nil { + log.Debugf("dial: new connection to host %s", addr) + } + + var buf bytes.Buffer + conn, err = h.options.Router.Dial(ctxvalue.ContextWithBuffer(ctx, &buf), network, addr) + if ro := recorderObjectFromContext(ctx); ro != nil { + ro.Route = buf.String() + } + + return +} + +func upgradeType(h http.Header) string { + if !httpguts.HeaderValuesContainsToken(h["Connection"], "Upgrade") { + return "" + } + return h.Get("Upgrade") +} + +func (h *httpHandler) 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") + } + + res.Body = nil + if err := res.Write(rw); err != nil { + return fmt.Errorf("response write: %v", err) + } + + return xnet.Transport(rw, backConn) } func (h *httpHandler) decodeServerName(s string) (string, error) { diff --git a/handler/http/metadata.go b/handler/http/metadata.go index 13579117..385b4024 100644 --- a/handler/http/metadata.go +++ b/handler/http/metadata.go @@ -53,7 +53,10 @@ func (h *httpHandler) parseMetadata(md mdata.Metadata) error { h.md.header = hd } - h.md.keepalive = mdutil.GetBool(md, "http.keepalive", "keepalive") + h.md.keepalive = true + if mdutil.IsExists(md, "http.keepalive", "keepalive") { + h.md.keepalive = mdutil.GetBool(md, "http.keepalive", "keepalive") + } if pr := mdutil.GetString(md, "probeResist", "probe_resist"); pr != "" { if ss := strings.SplitN(pr, ":", 2); len(ss) == 2 { diff --git a/handler/http2/handler.go b/handler/http2/handler.go index e96ecb9b..0715fc2e 100644 --- a/handler/http2/handler.go +++ b/handler/http2/handler.go @@ -178,7 +178,8 @@ func (h *http2Handler) roundTrip(ctx context.Context, w http.ResponseWriter, req ro.Host = host fields := map[string]any{ - "dst": host, + "dst": host, + "host": host, } if u, _, _ := h.basicProxyAuth(req.Header.Get("Proxy-Authorization")); u != "" { fields["user"] = u diff --git a/handler/http3/handler.go b/handler/http3/handler.go index 01c50231..b38cc2f3 100644 --- a/handler/http3/handler.go +++ b/handler/http3/handler.go @@ -124,7 +124,8 @@ func (h *http3Handler) roundTrip(ctx context.Context, w http.ResponseWriter, req } log = log.WithFields(map[string]any{ - "dst": fmt.Sprintf("%s/%s", target.Addr, "tcp"), + "dst": fmt.Sprintf("%s/%s", target.Addr, "tcp"), + "host": target.Addr, }) log.Debugf("%s >> %s", req.RemoteAddr, addr) diff --git a/handler/redirect/tcp/handler.go b/handler/redirect/tcp/handler.go index 46e9171a..42c3d5b8 100644 --- a/handler/redirect/tcp/handler.go +++ b/handler/redirect/tcp/handler.go @@ -131,6 +131,7 @@ func (h *redirectHandler) Handle(ctx context.Context, conn net.Conn, opts ...han log = log.WithFields(map[string]any{ "dst": fmt.Sprintf("%s/%s", dstAddr, dstAddr.Network()), + "host": dstAddr.String(), }) if h.md.sniffing { diff --git a/handler/redirect/udp/handler.go b/handler/redirect/udp/handler.go index a4e4a1a0..fe6e349d 100644 --- a/handler/redirect/udp/handler.go +++ b/handler/redirect/udp/handler.go @@ -108,7 +108,8 @@ func (h *redirectHandler) Handle(ctx context.Context, conn net.Conn, opts ...han ro.Host = dstAddr.String() log = log.WithFields(map[string]any{ - "dst": fmt.Sprintf("%s/%s", dstAddr, dstAddr.Network()), + "dst": fmt.Sprintf("%s/%s", dstAddr, dstAddr.Network()), + "host": dstAddr.String(), }) log.Debugf("%s >> %s", conn.RemoteAddr(), dstAddr) diff --git a/handler/relay/connect.go b/handler/relay/connect.go index ce0f12d1..bf81059a 100644 --- a/handler/relay/connect.go +++ b/handler/relay/connect.go @@ -33,8 +33,9 @@ func (h *relayHandler) handleConnect(ctx context.Context, conn net.Conn, network } log = log.WithFields(map[string]any{ - "dst": fmt.Sprintf("%s/%s", address, network), - "cmd": "connect", + "dst": fmt.Sprintf("%s/%s", address, network), + "cmd": "connect", + "host": address, }) log.Debugf("%s >> %s/%s", conn.RemoteAddr(), address, network) diff --git a/handler/socks/v4/handler.go b/handler/socks/v4/handler.go index f6021d7f..f9686c72 100644 --- a/handler/socks/v4/handler.go +++ b/handler/socks/v4/handler.go @@ -196,7 +196,8 @@ func (h *socks4Handler) handleConnect(ctx context.Context, conn net.Conn, req *g addr := req.Addr.String() log = log.WithFields(map[string]any{ - "dst": addr, + "dst": addr, + "host": addr, }) log.Debugf("%s >> %s", conn.RemoteAddr(), addr) diff --git a/handler/socks/v5/connect.go b/handler/socks/v5/connect.go index 9389ff05..a1d938f5 100644 --- a/handler/socks/v5/connect.go +++ b/handler/socks/v5/connect.go @@ -23,8 +23,9 @@ import ( func (h *socks5Handler) handleConnect(ctx context.Context, conn net.Conn, network, address string, ro *xrecorder.HandlerRecorderObject, log logger.Logger) error { log = log.WithFields(map[string]any{ - "dst": fmt.Sprintf("%s/%s", address, network), - "cmd": "connect", + "dst": fmt.Sprintf("%s/%s", address, network), + "cmd": "connect", + "host": address, }) log.Debugf("%s >> %s", conn.RemoteAddr(), address) diff --git a/handler/ss/handler.go b/handler/ss/handler.go index e482797f..3cf4e2e8 100644 --- a/handler/ss/handler.go +++ b/handler/ss/handler.go @@ -148,7 +148,8 @@ func (h *ssHandler) Handle(ctx context.Context, conn net.Conn, opts ...handler.H conn.SetReadDeadline(time.Time{}) log = log.WithFields(map[string]any{ - "dst": addr.String(), + "dst": addr.String(), + "host": addr.String(), }) log.Debugf("%s >> %s", conn.RemoteAddr(), addr) diff --git a/handler/sshd/handler.go b/handler/sshd/handler.go index 93e40f84..298206c7 100644 --- a/handler/sshd/handler.go +++ b/handler/sshd/handler.go @@ -140,8 +140,9 @@ func (h *forwardHandler) handleDirectForward(ctx context.Context, conn *sshd_uti ro.Host = targetAddr log = log.WithFields(map[string]any{ - "dst": fmt.Sprintf("%s/%s", targetAddr, "tcp"), - "cmd": "connect", + "dst": fmt.Sprintf("%s/%s", targetAddr, "tcp"), + "cmd": "connect", + "host": targetAddr, }) log.Debugf("%s >> %s", conn.RemoteAddr(), targetAddr) diff --git a/handler/tunnel/connect.go b/handler/tunnel/connect.go index 4682a6d9..ce9f423d 100644 --- a/handler/tunnel/connect.go +++ b/handler/tunnel/connect.go @@ -16,6 +16,7 @@ func (h *tunnelHandler) handleConnect(ctx context.Context, req *relay.Request, c "dst": fmt.Sprintf("%s/%s", dstAddr, network), "cmd": "connect", "tunnel": tunnelID.String(), + "host": dstAddr, }) resp := relay.Response{ diff --git a/handler/tunnel/entrypoint.go b/handler/tunnel/entrypoint.go index 72c3966d..27d9f855 100644 --- a/handler/tunnel/entrypoint.go +++ b/handler/tunnel/entrypoint.go @@ -180,7 +180,16 @@ func (ep *entrypoint) httpRoundTrip(ctx context.Context, rw io.ReadWriter, req * *ro2 = *ro ro = ro2 + host := req.Host + if _, port, _ := net.SplitHostPort(host); port == "" { + host = net.JoinHostPort(strings.Trim(host, "[]"), "80") + } + ro.Host = host ro.Time = time.Now() + + log = log.WithFields(map[string]any{ + "host": host, + }) log.Infof("%s <-> %s", ro.RemoteAddr, req.Host) defer func() { if err != nil { @@ -344,6 +353,7 @@ func (h *entrypoint) dial(ctx context.Context, network, addr string) (conn net.C if log == nil { log = h.log } + log.Debugf("dial: new connection to host %s", addr) if tunnelID.IsZero() { return nil, fmt.Errorf("%w %s", ErrTunnelRoute, addr) @@ -358,7 +368,6 @@ func (h *entrypoint) dial(ctx context.Context, network, addr string) (conn net.C } log = log.WithFields(map[string]any{ - "host": addr, "tunnel": tunnelID.String(), }) @@ -374,7 +383,7 @@ func (h *entrypoint) dial(ctx context.Context, network, addr string) (conn net.C if err != nil { return } - log.Debugf("new connection to tunnel: %s, connector: %s", tunnelID, cid) + log.Debugf("dial: connected to host %s, tunnel: %s, connector: %s", addr, tunnelID, cid) if node == h.node { clientAddr := ctxvalue.ClientAddrFromContext(ctx) diff --git a/handler/unix/handler.go b/handler/unix/handler.go index a52243d2..7041119d 100644 --- a/handler/unix/handler.go +++ b/handler/unix/handler.go @@ -126,6 +126,7 @@ func (h *unixHandler) Handle(ctx context.Context, conn net.Conn, opts ...handler log = log.WithFields(map[string]any{ "node": target.Name, "dst": target.Addr, + "host": target.Addr, }) ro.Host = target.Addr diff --git a/metadata/util/util.go b/metadata/util/util.go index 5fa13c0f..c5dad69f 100644 --- a/metadata/util/util.go +++ b/metadata/util/util.go @@ -9,6 +9,10 @@ import ( ) func IsExists(md metadata.Metadata, keys ...string) bool { + if md == nil { + return false + } + for _, key := range keys { if md.IsExists(key) { return true