diff --git a/config/parsing/service/parse.go b/config/parsing/service/parse.go index 53a87228..7433e647 100644 --- a/config/parsing/service/parse.go +++ b/config/parsing/service/parse.go @@ -14,7 +14,6 @@ import ( "github.com/go-gost/core/hop" "github.com/go-gost/core/listener" "github.com/go-gost/core/logger" - mdutil "github.com/go-gost/x/metadata/util" "github.com/go-gost/core/observer/stats" "github.com/go-gost/core/recorder" "github.com/go-gost/core/selector" @@ -30,6 +29,7 @@ import ( selector_parser "github.com/go-gost/x/config/parsing/selector" tls_util "github.com/go-gost/x/internal/util/tls" "github.com/go-gost/x/metadata" + mdutil "github.com/go-gost/x/metadata/util" "github.com/go-gost/x/registry" xservice "github.com/go-gost/x/service" "github.com/vishvananda/netns" @@ -73,6 +73,7 @@ func ParseService(cfg *config.ServiceConfig) (service.Service, error) { } if tlsConfig == nil { tlsConfig = parsing.DefaultTLSConfig().Clone() + tls_util.SetTLSOptions(tlsConfig, tlsCfg.Options) } authers := auth_parser.List(cfg.Listener.Auther, cfg.Listener.Authers...) @@ -222,6 +223,7 @@ func ParseService(cfg *config.ServiceConfig) (service.Service, error) { } if tlsConfig == nil { tlsConfig = parsing.DefaultTLSConfig().Clone() + tls_util.SetTLSOptions(tlsConfig, tlsCfg.Options) } authers = auth_parser.List(cfg.Handler.Auther, cfg.Handler.Authers...) diff --git a/handler/forward/local/handler.go b/handler/forward/local/handler.go index 8a8eaa40..e7fa2ec3 100644 --- a/handler/forward/local/handler.go +++ b/handler/forward/local/handler.go @@ -6,20 +6,23 @@ import ( "context" "errors" "fmt" - "io" "net" + "strconv" "time" "github.com/go-gost/core/chain" "github.com/go-gost/core/handler" "github.com/go-gost/core/hop" md "github.com/go-gost/core/metadata" + "github.com/go-gost/core/observer/stats" "github.com/go-gost/core/recorder" ctxvalue "github.com/go-gost/x/ctx" - xio "github.com/go-gost/x/internal/io" xnet "github.com/go-gost/x/internal/net" + "github.com/go-gost/x/internal/util/forwarder" "github.com/go-gost/x/internal/util/sniffing" + tls_util "github.com/go-gost/x/internal/util/tls" rate_limiter "github.com/go-gost/x/limiter/rate" + stats_wrapper "github.com/go-gost/x/observer/stats/wrapper" xrecorder "github.com/go-gost/x/recorder" "github.com/go-gost/x/registry" ) @@ -35,6 +38,7 @@ type forwardHandler struct { md metadata options handler.Options recorder recorder.RecorderObject + certPool tls_util.CertPool } func NewHandler(opts ...handler.Option) handler.Handler { @@ -60,6 +64,10 @@ func (h *forwardHandler) Init(md md.Metadata) (err error) { } } + if h.md.certificate != nil && h.md.privateKey != nil { + h.certPool = tls_util.NewMemoryCertPool() + } + return } @@ -77,6 +85,7 @@ func (h *forwardHandler) Handle(ctx context.Context, conn net.Conn, opts ...hand Service: h.options.Service, RemoteAddr: conn.RemoteAddr().String(), LocalAddr: conn.LocalAddr().String(), + Network: "tcp", Time: start, SID: string(ctxvalue.SidFromContext(ctx)), } @@ -87,14 +96,21 @@ func (h *forwardHandler) Handle(ctx context.Context, conn net.Conn, opts ...hand "local": conn.LocalAddr().String(), "sid": ctxvalue.SidFromContext(ctx), }) - 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) - ro.Record(ctx, h.recorder.Recorder) + if err := ro.Record(ctx, h.recorder.Recorder); err != nil { + log.Errorf("record: %v", err) + } log.WithFields(map[string]any{ "duration": time.Since(start), @@ -111,7 +127,6 @@ func (h *forwardHandler) Handle(ctx context.Context, conn net.Conn, opts ...hand } ro.Network = network - var rw io.ReadWriter = conn var proto string if network == "tcp" && h.md.sniffing { if h.md.sniffingTimeout > 0 { @@ -126,15 +141,42 @@ func (h *forwardHandler) Handle(ctx context.Context, conn net.Conn, opts ...hand conn.SetReadDeadline(time.Time{}) } - rw = xio.NewReadWriter(br, conn) + dial := func(ctx context.Context, network, address string) (net.Conn, error) { + var buf bytes.Buffer + cc, err := h.options.Router.Dial(ctxvalue.ContextWithBuffer(ctx, &buf), "tcp", address) + ro.Route = buf.String() + return cc, err + } + sniffer := &forwarder.Sniffer{ + Recorder: h.recorder.Recorder, + RecorderOptions: h.recorder.Options, + Certificate: h.md.certificate, + PrivateKey: h.md.privateKey, + NegotiatedProtocol: h.md.alpn, + CertPool: h.certPool, + MitmBypass: h.md.mitmBypass, + ReadTimeout: h.md.readTimeout, + } + + conn = xnet.NewReadWriteConn(br, conn, conn) switch proto { case sniffing.ProtoHTTP: - ro2 := &xrecorder.HandlerRecorderObject{} - *ro2 = *ro - ro.Time = time.Time{} - return h.handleHTTP(ctx, rw, ro2, log) + return sniffer.HandleHTTP(ctx, conn, + forwarder.WithDial(dial), + forwarder.WithHop(h.hop), + forwarder.WithBypass(h.options.Bypass), + forwarder.WithHTTPKeepalive(h.md.httpKeepalive), + forwarder.WithRecorderObject(ro), + forwarder.WithLog(log), + ) case sniffing.ProtoTLS: - return h.handleTLS(ctx, xnet.NewReadWriteConn(rw, rw, conn), ro, log) + return sniffer.HandleTLS(ctx, conn, + forwarder.WithDial(dial), + forwarder.WithHop(h.hop), + forwarder.WithBypass(h.options.Bypass), + forwarder.WithRecorderObject(ro), + forwarder.WithLog(log), + ) } } @@ -191,7 +233,7 @@ func (h *forwardHandler) Handle(ctx context.Context, conn net.Conn, opts ...hand t := time.Now() log.Infof("%s <-> %s", conn.RemoteAddr(), target.Addr) - xnet.Transport(rw, cc) + xnet.Transport(conn, cc) log.WithFields(map[string]any{ "duration": time.Since(t), }).Infof("%s >-< %s", conn.RemoteAddr(), target.Addr) @@ -210,3 +252,27 @@ func (h *forwardHandler) checkRateLimit(addr net.Addr) bool { return true } + +func convertAddr(addr net.Addr) net.Addr { + host, sp, _ := net.SplitHostPort(addr.String()) + ip := net.ParseIP(host) + port, _ := strconv.Atoi(sp) + + if ip == nil || ip.Equal(net.IPv6zero) { + ip = net.IPv4zero + } + + switch addr.Network() { + case "tcp", "tcp4", "tcp6": + return &net.TCPAddr{ + IP: ip, + Port: port, + } + + default: + return &net.UDPAddr{ + IP: ip, + Port: port, + } + } +} diff --git a/handler/forward/local/metadata.go b/handler/forward/local/metadata.go index cd6305bd..2b76d418 100644 --- a/handler/forward/local/metadata.go +++ b/handler/forward/local/metadata.go @@ -17,6 +17,7 @@ type metadata struct { sniffing bool sniffingTimeout time.Duration + httpKeepalive bool certificate *x509.Certificate privateKey crypto.PrivateKey @@ -33,6 +34,8 @@ func (h *forwardHandler) parseMetadata(md mdata.Metadata) (err error) { h.md.sniffing = mdutil.GetBool(md, "sniffing") h.md.sniffingTimeout = mdutil.GetDuration(md, "sniffing.timeout") + h.md.httpKeepalive = mdutil.GetBool(md, "http.keepalive") + certFile := mdutil.GetString(md, "mitm.certFile", "mitm.caCertFile") keyFile := mdutil.GetString(md, "mitm.keyFile", "mitm.caKeyFile") if certFile != "" && keyFile != "" { diff --git a/handler/forward/local/sniffing.go b/handler/forward/local/sniffing.go deleted file mode 100644 index b0df864d..00000000 --- a/handler/forward/local/sniffing.go +++ /dev/null @@ -1,492 +0,0 @@ -package local - -import ( - "bufio" - "bytes" - "context" - "crypto/tls" - "encoding/hex" - "errors" - "fmt" - "io" - "net" - "net/http" - "net/http/httputil" - "strings" - "time" - - "github.com/go-gost/core/bypass" - "github.com/go-gost/core/chain" - "github.com/go-gost/core/hop" - "github.com/go-gost/core/logger" - dissector "github.com/go-gost/tls-dissector" - xbypass "github.com/go-gost/x/bypass" - "github.com/go-gost/x/config" - 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/util/sniffing" - tls_util "github.com/go-gost/x/internal/util/tls" - xrecorder "github.com/go-gost/x/recorder" -) - -const ( - defaultBodySize = 1024 * 1024 // 1MB -) - -func (h *forwardHandler) handleHTTP(ctx context.Context, rw io.ReadWriter, ro *xrecorder.HandlerRecorderObject, log logger.Logger) error { - br := bufio.NewReader(rw) - req, err := http.ReadRequest(br) - if err != nil { - return err - } - - if log.IsLevelEnabled(logger.TraceLevel) { - dump, _ := httputil.DumpRequest(req, false) - log.Trace(string(dump)) - } - - shouldClose, err := h.httpRoundTrip(ctx, rw, req, ro, log) - if err != nil || shouldClose { - return err - } - - for { - 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, rw, req, ro, log); err != nil || shouldClose { - return err - } - } -} - -func (h *forwardHandler) httpRoundTrip(ctx context.Context, rw io.ReadWriter, req *http.Request, ro *xrecorder.HandlerRecorderObject, log logger.Logger) (close bool, err error) { - close = true - - start := time.Now() - ro.Time = start - - log.Infof("%s <-> %s", ro.RemoteAddr, req.Host) - defer func() { - ro.Duration = time.Since(start) - if err != nil { - ro.Err = err.Error() - } - if err := ro.Record(ctx, h.recorder.Recorder); err != nil { - log.Errorf("record: %v", err) - } - - log.WithFields(map[string]any{ - "duration": time.Since(start), - }).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(), - }, - } - - 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)) - } - } - - // 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") - } - } - - 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") - } - - if bp := h.options.Bypass; bp != nil && - bp.Contains(ctx, "tcp", host, bypass.WithPathOption(req.RequestURI)) { - log.Debugf("bypass: %s %s", host, req.RequestURI) - res.StatusCode = http.StatusForbidden - ro.HTTP.StatusCode = res.StatusCode - res.Write(rw) - err = xbypass.ErrBypass - return - } - - target := &chain.Node{ - Addr: host, - } - if h.hop != nil { - target = h.hop.Select(ctx, - hop.HostSelectOption(host), - hop.ProtocolSelectOption(sniffing.ProtoHTTP), - hop.PathSelectOption(req.URL.Path), - ) - } - if target == nil { - log.Warnf("node for %s not found", host) - res.StatusCode = http.StatusBadGateway - ro.HTTP.StatusCode = res.StatusCode - res.Write(rw) - err = errors.New("node not available") - return - } - - ro.Host = target.Addr - - log = log.WithFields(map[string]any{ - "host": req.Host, - "node": target.Name, - "dst": target.Addr, - }) - log.Debugf("find node for host %s -> %s(%s)", host, target.Name, target.Addr) - - var bodyRewrites []chain.HTTPBodyRewriteSettings - if httpSettings := target.Options().HTTP; httpSettings != nil { - if auther := httpSettings.Auther; auther != nil { - username, password, _ := req.BasicAuth() - id, ok := auther.Authenticate(ctx, username, password) - if !ok { - res.StatusCode = http.StatusUnauthorized - ro.HTTP.StatusCode = res.StatusCode - res.Header.Set("WWW-Authenticate", "Basic") - log.Warnf("node %s(%s) 401 unauthorized", target.Name, target.Addr) - res.Write(rw) - err = errors.New("unauthorized") - return - } - ctx = ctxvalue.ContextWithClientID(ctx, ctxvalue.ClientID(id)) - } - - if httpSettings.Host != "" { - req.Host = httpSettings.Host - } - for k, v := range httpSettings.Header { - req.Header.Set(k, v) - } - - for _, re := range httpSettings.RewriteURL { - if re.Pattern.MatchString(req.URL.Path) { - if s := re.Pattern.ReplaceAllString(req.URL.Path, re.Replacement); s != "" { - req.URL.Path = s - break - } - } - } - - bodyRewrites = httpSettings.RewriteBody - } - - var buf bytes.Buffer - cc, err := h.options.Router.Dial(ctxvalue.ContextWithBuffer(ctx, &buf), "tcp", target.Addr) - ro.Route = buf.String() - if err != nil { - // TODO: the router itself may be failed due to the failed node in the router, - // the dead marker may be a wrong operation. - if marker := target.Marker(); marker != nil { - marker.Mark() - } - log.Warnf("connect to node %s(%s) failed: %v", target.Name, target.Addr, err) - res.Write(rw) - return - } - if marker := target.Marker(); marker != nil { - marker.Reset() - } - // TODO: re-use the connection - defer cc.Close() - - log.Debugf("connect to node %s(%s)", target.Name, target.Addr) - - if tlsSettings := target.Options().TLS; tlsSettings != nil { - cfg := &tls.Config{ - ServerName: tlsSettings.ServerName, - InsecureSkipVerify: !tlsSettings.Secure, - } - tls_util.SetTLSOptions(cfg, &config.TLSOptions{ - MinVersion: tlsSettings.Options.MinVersion, - MaxVersion: tlsSettings.Options.MaxVersion, - CipherSuites: tlsSettings.Options.CipherSuites, - ALPN: tlsSettings.Options.ALPN, - }) - cc = tls.Client(cc, cfg) - } - - 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 { - res.Write(rw) - return - } - - 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 - 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 - } - - if err = h.rewriteBody(resp, bodyRewrites...); err != nil { - log.Errorf("rewrite body: %v", err) - return - } - - 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 -} - -func (h *forwardHandler) handleTLS(ctx context.Context, conn net.Conn, ro *xrecorder.HandlerRecorderObject, log logger.Logger) error { - buf := new(bytes.Buffer) - - clientHello, err := dissector.ParseClientHello(io.TeeReader(conn, buf)) - if err != nil { - return err - } - - ro.TLS = &xrecorder.TLSRecorderObject{ - ServerName: clientHello.ServerName, - ClientHello: hex.EncodeToString(buf.Bytes()), - } - if len(clientHello.SupportedProtos) > 0 { - ro.TLS.Proto = clientHello.SupportedProtos[0] - } - - host := clientHello.ServerName - if _, _, err := net.SplitHostPort(host); err != nil { - host = net.JoinHostPort(strings.Trim(host, "[]"), "0") - } - - var target *chain.Node - if host != "" { - target = &chain.Node{ - Addr: host, - } - } - if h.hop != nil { - target = h.hop.Select(ctx, - hop.HostSelectOption(host), - hop.ProtocolSelectOption(sniffing.ProtoTLS), - ) - } - if target == nil { - err := errors.New("target not available") - log.Error(err) - return err - } - - addr := target.Addr - if opts := target.Options(); opts != nil { - switch opts.Network { - case "unix": - ro.Network = opts.Network - default: - if _, _, err := net.SplitHostPort(addr); err != nil { - addr += ":0" - } - } - } - ro.Host = addr - - log = log.WithFields(map[string]any{ - "host": host, - "node": target.Name, - "dst": fmt.Sprintf("%s/%s", addr, ro.Network), - }) - - log.Debugf("%s >> %s", ro.RemoteAddr, addr) - - var routeBuf bytes.Buffer - cc, err := h.options.Router.Dial(ctxvalue.ContextWithBuffer(ctx, &routeBuf), ro.Network, addr) - ro.Route = routeBuf.String() - if err != nil { - log.Error(err) - // TODO: the router itself may be failed due to the failed node in the router, - // the dead marker may be a wrong operation. - if marker := target.Marker(); marker != nil { - marker.Mark() - } - return err - } - defer cc.Close() - if marker := target.Marker(); marker != nil { - marker.Reset() - } - - if _, err := buf.WriteTo(cc); err != nil { - return err - } - - cc.SetReadDeadline(time.Now().Add(h.md.readTimeout)) - serverHello, err := dissector.ParseServerHello(io.TeeReader(cc, buf)) - cc.SetReadDeadline(time.Time{}) - - if serverHello != nil { - ro.TLS.CipherSuite = tls_util.CipherSuite(serverHello.CipherSuite).String() - ro.TLS.CompressionMethod = serverHello.CompressionMethod - if serverHello.Proto != "" { - ro.TLS.Proto = serverHello.Proto - } - if serverHello.Version > 0 { - ro.TLS.Version = tls_util.Version(serverHello.Version).String() - } - } - - if buf.Len() > 0 { - ro.TLS.ServerHello = hex.EncodeToString(buf.Bytes()) - } - - if _, err := buf.WriteTo(conn); err != nil { - return err - } - - t := time.Now() - log.Infof("%s <-> %s", ro.RemoteAddr, addr) - xnet.Transport(conn, cc) - log.WithFields(map[string]any{ - "duration": time.Since(t), - }).Infof("%s >-< %s", ro.RemoteAddr, addr) - - return err -} - -func (h *forwardHandler) rewriteBody(resp *http.Response, rewrites ...chain.HTTPBodyRewriteSettings) error { - if resp == nil || len(rewrites) == 0 || resp.ContentLength <= 0 { - return nil - } - - if encoding := resp.Header.Get("Content-Encoding"); encoding != "" { - return nil - } - - body, err := drainBody(resp.Body) - if err != nil || body == nil { - return err - } - - contentType, _, _ := strings.Cut(resp.Header.Get("Content-Type"), ";") - for _, rewrite := range rewrites { - rewriteType := rewrite.Type - if rewriteType == "" { - rewriteType = "text/html" - } - if rewriteType != "*" && !strings.Contains(rewriteType, contentType) { - continue - } - - body = rewrite.Pattern.ReplaceAll(body, rewrite.Replacement) - } - - resp.Body = io.NopCloser(bytes.NewReader(body)) - resp.ContentLength = int64(len(body)) - - return nil -} - -func drainBody(b io.ReadCloser) (body []byte, err error) { - if b == nil || b == http.NoBody { - // No copying needed. Preserve the magic sentinel meaning of NoBody. - return nil, nil - } - var buf bytes.Buffer - if _, err = buf.ReadFrom(b); err != nil { - return nil, err - } - if err = b.Close(); err != nil { - return nil, err - } - return buf.Bytes(), nil -} diff --git a/handler/forward/remote/handler.go b/handler/forward/remote/handler.go index 7647829b..e6158f1d 100644 --- a/handler/forward/remote/handler.go +++ b/handler/forward/remote/handler.go @@ -6,7 +6,6 @@ import ( "context" "errors" "fmt" - "io" "net" "strconv" "time" @@ -15,13 +14,16 @@ import ( "github.com/go-gost/core/handler" "github.com/go-gost/core/hop" mdata "github.com/go-gost/core/metadata" + "github.com/go-gost/core/observer/stats" "github.com/go-gost/core/recorder" ctxvalue "github.com/go-gost/x/ctx" - xio "github.com/go-gost/x/internal/io" xnet "github.com/go-gost/x/internal/net" "github.com/go-gost/x/internal/net/proxyproto" + "github.com/go-gost/x/internal/util/forwarder" "github.com/go-gost/x/internal/util/sniffing" + tls_util "github.com/go-gost/x/internal/util/tls" rate_limiter "github.com/go-gost/x/limiter/rate" + stats_wrapper "github.com/go-gost/x/observer/stats/wrapper" xrecorder "github.com/go-gost/x/recorder" "github.com/go-gost/x/registry" ) @@ -36,6 +38,7 @@ type forwardHandler struct { md metadata options handler.Options recorder recorder.RecorderObject + certPool tls_util.CertPool } func NewHandler(opts ...handler.Option) handler.Handler { @@ -61,6 +64,10 @@ func (h *forwardHandler) Init(md mdata.Metadata) (err error) { } } + if h.md.certificate != nil && h.md.privateKey != nil { + h.certPool = tls_util.NewMemoryCertPool() + } + return } @@ -78,6 +85,7 @@ func (h *forwardHandler) Handle(ctx context.Context, conn net.Conn, opts ...hand Service: h.options.Service, RemoteAddr: conn.RemoteAddr().String(), LocalAddr: conn.LocalAddr().String(), + Network: "tcp", Time: start, SID: string(ctxvalue.SidFromContext(ctx)), } @@ -88,14 +96,21 @@ func (h *forwardHandler) Handle(ctx context.Context, conn net.Conn, opts ...hand "local": conn.LocalAddr().String(), "sid": ctxvalue.SidFromContext(ctx), }) - 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) - ro.Record(ctx, h.recorder.Recorder) + if err := ro.Record(ctx, h.recorder.Recorder); err != nil { + log.Errorf("record: %v", err) + } log.WithFields(map[string]any{ "duration": time.Since(start), @@ -112,7 +127,6 @@ func (h *forwardHandler) Handle(ctx context.Context, conn net.Conn, opts ...hand } ro.Network = network - var rw io.ReadWriter = conn var proto string if network == "tcp" && h.md.sniffing { if h.md.sniffingTimeout > 0 { @@ -127,15 +141,42 @@ func (h *forwardHandler) Handle(ctx context.Context, conn net.Conn, opts ...hand conn.SetReadDeadline(time.Time{}) } - rw = xio.NewReadWriter(br, conn) + dial := func(ctx context.Context, network, address string) (net.Conn, error) { + var buf bytes.Buffer + cc, err := h.options.Router.Dial(ctxvalue.ContextWithBuffer(ctx, &buf), "tcp", address) + ro.Route = buf.String() + return cc, err + } + sniffer := &forwarder.Sniffer{ + Recorder: h.recorder.Recorder, + RecorderOptions: h.recorder.Options, + Certificate: h.md.certificate, + PrivateKey: h.md.privateKey, + NegotiatedProtocol: h.md.alpn, + CertPool: h.certPool, + MitmBypass: h.md.mitmBypass, + ReadTimeout: h.md.readTimeout, + } + + conn = xnet.NewReadWriteConn(br, conn, conn) switch proto { case sniffing.ProtoHTTP: - ro2 := &xrecorder.HandlerRecorderObject{} - *ro2 = *ro - ro.Time = time.Time{} - return h.handleHTTP(ctx, rw, ro2, log) + return sniffer.HandleHTTP(ctx, conn, + forwarder.WithDial(dial), + forwarder.WithHop(h.hop), + forwarder.WithBypass(h.options.Bypass), + forwarder.WithHTTPKeepalive(h.md.httpKeepalive), + forwarder.WithRecorderObject(ro), + forwarder.WithLog(log), + ) case sniffing.ProtoTLS: - return h.handleTLS(ctx, xnet.NewReadWriteConn(rw, rw, conn), ro, log) + return sniffer.HandleTLS(ctx, conn, + forwarder.WithDial(dial), + forwarder.WithHop(h.hop), + forwarder.WithBypass(h.options.Bypass), + forwarder.WithRecorderObject(ro), + forwarder.WithLog(log), + ) } } @@ -190,7 +231,7 @@ func (h *forwardHandler) Handle(ctx context.Context, conn net.Conn, opts ...hand t := time.Now() log.Infof("%s <-> %s", conn.RemoteAddr(), target.Addr) - xnet.Transport(rw, cc) + xnet.Transport(conn, cc) log.WithFields(map[string]any{ "duration": time.Since(t), }).Infof("%s >-< %s", conn.RemoteAddr(), target.Addr) diff --git a/handler/forward/remote/metadata.go b/handler/forward/remote/metadata.go index d0014213..cdddc42c 100644 --- a/handler/forward/remote/metadata.go +++ b/handler/forward/remote/metadata.go @@ -1,29 +1,57 @@ package remote import ( + "crypto" + "crypto/tls" + "crypto/x509" "time" + "github.com/go-gost/core/bypass" mdata "github.com/go-gost/core/metadata" mdutil "github.com/go-gost/x/metadata/util" + "github.com/go-gost/x/registry" ) type metadata struct { - readTimeout time.Duration + readTimeout time.Duration + proxyProtocol int + sniffing bool sniffingTimeout time.Duration - proxyProtocol int + httpKeepalive bool + + certificate *x509.Certificate + privateKey crypto.PrivateKey + alpn string + mitmBypass bypass.Bypass } func (h *forwardHandler) parseMetadata(md mdata.Metadata) (err error) { - const ( - readTimeout = "readTimeout" - sniffing = "sniffing" - proxyProtocol = "proxyProtocol" - ) + h.md.readTimeout = mdutil.GetDuration(md, "readTimeout") + if h.md.readTimeout <= 0 { + h.md.readTimeout = 15 * time.Second + } + h.md.proxyProtocol = mdutil.GetInt(md, "proxyProtocol") - h.md.readTimeout = mdutil.GetDuration(md, readTimeout) - h.md.sniffing = mdutil.GetBool(md, sniffing) + h.md.sniffing = mdutil.GetBool(md, "sniffing") h.md.sniffingTimeout = mdutil.GetDuration(md, "sniffing.timeout") - h.md.proxyProtocol = mdutil.GetInt(md, proxyProtocol) + + h.md.httpKeepalive = mdutil.GetBool(md, "http.keepalive") + + certFile := mdutil.GetString(md, "mitm.certFile", "mitm.caCertFile") + keyFile := mdutil.GetString(md, "mitm.keyFile", "mitm.caKeyFile") + if certFile != "" && keyFile != "" { + tlsCert, err := tls.LoadX509KeyPair(certFile, keyFile) + if err != nil { + return err + } + h.md.certificate, err = x509.ParseCertificate(tlsCert.Certificate[0]) + if err != nil { + return err + } + h.md.privateKey = tlsCert.PrivateKey + } + h.md.alpn = mdutil.GetString(md, "mitm.alpn") + h.md.mitmBypass = registry.BypassRegistry().Get(mdutil.GetString(md, "mitm.bypass")) return } diff --git a/handler/forward/remote/sniffing.go b/handler/forward/remote/sniffing.go deleted file mode 100644 index ec264dd4..00000000 --- a/handler/forward/remote/sniffing.go +++ /dev/null @@ -1,495 +0,0 @@ -package remote - -import ( - "bufio" - "bytes" - "context" - "crypto/tls" - "encoding/hex" - "errors" - "fmt" - "io" - "net" - "net/http" - "net/http/httputil" - "strings" - "time" - - "github.com/go-gost/core/bypass" - "github.com/go-gost/core/chain" - "github.com/go-gost/core/hop" - "github.com/go-gost/core/logger" - dissector "github.com/go-gost/tls-dissector" - xbypass "github.com/go-gost/x/bypass" - "github.com/go-gost/x/config" - 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" - "github.com/go-gost/x/internal/util/sniffing" - tls_util "github.com/go-gost/x/internal/util/tls" - xrecorder "github.com/go-gost/x/recorder" -) - -const ( - defaultBodySize = 1024 * 1024 // 1MB -) - -func (h *forwardHandler) handleHTTP(ctx context.Context, rw io.ReadWriter, ro *xrecorder.HandlerRecorderObject, log logger.Logger) error { - br := bufio.NewReader(rw) - req, err := http.ReadRequest(br) - if err != nil { - return err - } - - if log.IsLevelEnabled(logger.TraceLevel) { - dump, _ := httputil.DumpRequest(req, false) - log.Trace(string(dump)) - } - - shouldClose, err := h.httpRoundTrip(ctx, rw, req, ro, log) - if err != nil || shouldClose { - return err - } - - for { - 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, rw, req, ro, log); err != nil || shouldClose { - return err - } - } -} - -func (h *forwardHandler) httpRoundTrip(ctx context.Context, rw io.ReadWriter, req *http.Request, ro *xrecorder.HandlerRecorderObject, log logger.Logger) (close bool, err error) { - close = true - - start := time.Now() - ro.Time = start - - log.Infof("%s <-> %s", ro.RemoteAddr, req.Host) - defer func() { - ro.Duration = time.Since(start) - if err != nil { - ro.Err = err.Error() - } - if err := ro.Record(ctx, h.recorder.Recorder); err != nil { - log.Errorf("record: %v", err) - } - - log.WithFields(map[string]any{ - "duration": time.Since(start), - }).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(), - }, - } - - 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)) - } - - // 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") - } - } - - 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") - } - - if bp := h.options.Bypass; bp != nil && - bp.Contains(ctx, "tcp", host, bypass.WithPathOption(req.RequestURI)) { - log.Debugf("bypass: %s %s", host, req.RequestURI) - res.StatusCode = http.StatusForbidden - ro.HTTP.StatusCode = res.StatusCode - res.Write(rw) - err = xbypass.ErrBypass - return - } - - target := &chain.Node{ - Addr: host, - } - if h.hop != nil { - target = h.hop.Select(ctx, - hop.HostSelectOption(host), - hop.ProtocolSelectOption(sniffing.ProtoHTTP), - hop.PathSelectOption(req.URL.Path), - ) - } - if target == nil { - log.Warnf("node for %s not found", host) - res.StatusCode = http.StatusBadGateway - ro.HTTP.StatusCode = res.StatusCode - res.Write(rw) - err = errors.New("node not available") - return - } - - ro.Host = target.Addr - - log = log.WithFields(map[string]any{ - "host": req.Host, - "node": target.Name, - "dst": target.Addr, - }) - log.Debugf("find node for host %s -> %s(%s)", host, target.Name, target.Addr) - - var bodyRewrites []chain.HTTPBodyRewriteSettings - if httpSettings := target.Options().HTTP; httpSettings != nil { - if auther := httpSettings.Auther; auther != nil { - username, password, _ := req.BasicAuth() - id, ok := auther.Authenticate(ctx, username, password) - if !ok { - res.StatusCode = http.StatusUnauthorized - ro.HTTP.StatusCode = res.StatusCode - res.Header.Set("WWW-Authenticate", "Basic") - log.Warnf("node %s(%s) 401 unauthorized", target.Name, target.Addr) - res.Write(rw) - err = errors.New("unauthorized") - return - } - ctx = ctxvalue.ContextWithClientID(ctx, ctxvalue.ClientID(id)) - } - - if httpSettings.Host != "" { - req.Host = httpSettings.Host - } - for k, v := range httpSettings.Header { - req.Header.Set(k, v) - } - - for _, re := range httpSettings.RewriteURL { - if re.Pattern.MatchString(req.URL.Path) { - if s := re.Pattern.ReplaceAllString(req.URL.Path, re.Replacement); s != "" { - req.URL.Path = s - break - } - } - } - - bodyRewrites = httpSettings.RewriteBody - } - - var buf bytes.Buffer - cc, err := h.options.Router.Dial(ctxvalue.ContextWithBuffer(ctx, &buf), "tcp", target.Addr) - ro.Route = buf.String() - if err != nil { - // TODO: the router itself may be failed due to the failed node in the router, - // the dead marker may be a wrong operation. - if marker := target.Marker(); marker != nil { - marker.Mark() - } - log.Warnf("connect to node %s(%s) failed: %v", target.Name, target.Addr, err) - res.Write(rw) - return - } - if marker := target.Marker(); marker != nil { - marker.Reset() - } - // TODO: re-use the connection - defer cc.Close() - - log.Debugf("connect to node %s(%s)", target.Name, target.Addr) - - if tlsSettings := target.Options().TLS; tlsSettings != nil { - cfg := &tls.Config{ - ServerName: tlsSettings.ServerName, - InsecureSkipVerify: !tlsSettings.Secure, - } - tls_util.SetTLSOptions(cfg, &config.TLSOptions{ - MinVersion: tlsSettings.Options.MinVersion, - MaxVersion: tlsSettings.Options.MaxVersion, - CipherSuites: tlsSettings.Options.CipherSuites, - ALPN: tlsSettings.Options.ALPN, - }) - cc = tls.Client(cc, cfg) - } - - remoteAddr, _ := net.ResolveTCPAddr("tcp", clientAddr) - localAddr, _ := net.ResolveTCPAddr("tcp", ro.LocalAddr) - cc = proxyproto.WrapClientConn(h.md.proxyProtocol, remoteAddr, localAddr, cc) - - 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 { - res.Write(rw) - return - } - - if reqBody != nil { - ro.HTTP.Request.Body = reqBody.Content() - ro.HTTP.Request.ContentLength = reqBody.Length() - } - - cc.SetReadDeadline(time.Now().Add(30 * time.Second)) - 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 - 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 - } - - if err = h.rewriteBody(resp, bodyRewrites...); err != nil { - log.Errorf("rewrite body: %v", err) - return - } - - 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 -} - -func (h *forwardHandler) handleTLS(ctx context.Context, conn net.Conn, ro *xrecorder.HandlerRecorderObject, log logger.Logger) error { - buf := new(bytes.Buffer) - - clientHello, err := dissector.ParseClientHello(io.TeeReader(conn, buf)) - if err != nil { - return err - } - - ro.TLS = &xrecorder.TLSRecorderObject{ - ServerName: clientHello.ServerName, - ClientHello: hex.EncodeToString(buf.Bytes()), - } - if len(clientHello.SupportedProtos) > 0 { - ro.TLS.Proto = clientHello.SupportedProtos[0] - } - - host := clientHello.ServerName - if _, _, err := net.SplitHostPort(host); err != nil { - host = net.JoinHostPort(strings.Trim(host, "[]"), "0") - } - - var target *chain.Node - if host != "" { - target = &chain.Node{ - Addr: host, - } - } - if h.hop != nil { - target = h.hop.Select(ctx, - hop.HostSelectOption(host), - hop.ProtocolSelectOption(sniffing.ProtoTLS), - ) - } - if target == nil { - err := errors.New("target not available") - log.Error(err) - return err - } - - addr := target.Addr - if opts := target.Options(); opts != nil { - switch opts.Network { - case "unix": - ro.Network = opts.Network - default: - if _, _, err := net.SplitHostPort(addr); err != nil { - addr += ":0" - } - } - } - ro.Host = addr - - log = log.WithFields(map[string]any{ - "host": host, - "node": target.Name, - "dst": fmt.Sprintf("%s/%s", addr, ro.Network), - }) - - log.Debugf("%s >> %s", ro.RemoteAddr, addr) - - var routeBuf bytes.Buffer - cc, err := h.options.Router.Dial(ctxvalue.ContextWithBuffer(ctx, &routeBuf), ro.Network, addr) - ro.Route = routeBuf.String() - if err != nil { - log.Error(err) - // TODO: the router itself may be failed due to the failed node in the router, - // the dead marker may be a wrong operation. - if marker := target.Marker(); marker != nil { - marker.Mark() - } - return err - } - defer cc.Close() - if marker := target.Marker(); marker != nil { - marker.Reset() - } - - if _, err := buf.WriteTo(cc); err != nil { - return err - } - - cc.SetReadDeadline(time.Now().Add(30 * time.Second)) - serverHello, err := dissector.ParseServerHello(io.TeeReader(cc, buf)) - cc.SetReadDeadline(time.Time{}) - - if serverHello != nil { - ro.TLS.CipherSuite = tls_util.CipherSuite(serverHello.CipherSuite).String() - ro.TLS.CompressionMethod = serverHello.CompressionMethod - if serverHello.Proto != "" { - ro.TLS.Proto = serverHello.Proto - } - if serverHello.Version > 0 { - ro.TLS.Version = tls_util.Version(serverHello.Version).String() - } - } - - if buf.Len() > 0 { - ro.TLS.ServerHello = hex.EncodeToString(buf.Bytes()) - } - - if _, err := buf.WriteTo(conn); err != nil { - return err - } - - t := time.Now() - log.Infof("%s <-> %s", ro.RemoteAddr, addr) - xnet.Transport(conn, cc) - log.WithFields(map[string]any{ - "duration": time.Since(t), - }).Infof("%s >-< %s", ro.RemoteAddr, addr) - - return err -} - -func (h *forwardHandler) rewriteBody(resp *http.Response, rewrites ...chain.HTTPBodyRewriteSettings) error { - if resp == nil || len(rewrites) == 0 || resp.ContentLength <= 0 { - return nil - } - - if encoding := resp.Header.Get("Content-Encoding"); encoding != "" { - return nil - } - - body, err := drainBody(resp.Body) - if err != nil || body == nil { - return err - } - - contentType, _, _ := strings.Cut(resp.Header.Get("Content-Type"), ";") - for _, rewrite := range rewrites { - rewriteType := rewrite.Type - if rewriteType == "" { - rewriteType = "text/html" - } - if rewriteType != "*" && !strings.Contains(rewriteType, contentType) { - continue - } - - body = rewrite.Pattern.ReplaceAll(body, rewrite.Replacement) - } - - resp.Body = io.NopCloser(bytes.NewReader(body)) - resp.ContentLength = int64(len(body)) - - return nil -} - -func drainBody(b io.ReadCloser) (body []byte, err error) { - if b == nil || b == http.NoBody { - // No copying needed. Preserve the magic sentinel meaning of NoBody. - return nil, nil - } - var buf bytes.Buffer - if _, err = buf.ReadFrom(b); err != nil { - return nil, err - } - if err = b.Close(); err != nil { - return nil, err - } - return buf.Bytes(), nil -} diff --git a/handler/sni/handler.go b/handler/sni/handler.go index a78b6248..4d6b500e 100644 --- a/handler/sni/handler.go +++ b/handler/sni/handler.go @@ -86,7 +86,6 @@ func (h *sniHandler) Handle(ctx context.Context, conn net.Conn, opts ...handler. "local": conn.LocalAddr().String(), "sid": ctxvalue.SidFromContext(ctx), }) - log.Infof("%s <> %s", conn.RemoteAddr(), conn.LocalAddr()) pStats := stats.Stats{} diff --git a/internal/util/forwarder/sniffer.go b/internal/util/forwarder/sniffer.go new file mode 100644 index 00000000..3ceae946 --- /dev/null +++ b/internal/util/forwarder/sniffer.go @@ -0,0 +1,957 @@ +package forwarder + +import ( + "bufio" + "bytes" + "context" + "crypto" + "crypto/tls" + "crypto/x509" + "encoding/hex" + "errors" + "fmt" + "io" + "net" + "net/http" + "net/http/httputil" + "strings" + "time" + + "github.com/go-gost/core/bypass" + "github.com/go-gost/core/chain" + "github.com/go-gost/core/hop" + "github.com/go-gost/core/logger" + "github.com/go-gost/core/observer/stats" + "github.com/go-gost/core/recorder" + dissector "github.com/go-gost/tls-dissector" + xbypass "github.com/go-gost/x/bypass" + "github.com/go-gost/x/config" + 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" + "github.com/go-gost/x/internal/util/sniffing" + tls_util "github.com/go-gost/x/internal/util/tls" + stats_wrapper "github.com/go-gost/x/observer/stats/wrapper" + xrecorder "github.com/go-gost/x/recorder" + "golang.org/x/net/http2" +) + +const ( + // Default max body size to record. + DefaultBodySize = 1024 * 1024 // 1MB +) + +var ( + DefaultCertPool = tls_util.NewMemoryCertPool() +) + +type HandleOptions struct { + Dial func(ctx context.Context, network, address string) (net.Conn, error) + + HTTPKeepalive bool + Node *chain.Node + Hop hop.Hop + Bypass bypass.Bypass + RecorderObject *xrecorder.HandlerRecorderObject + Log logger.Logger +} + +type HandleOption func(opts *HandleOptions) + +func WithDial(dial func(ctx context.Context, network, address string) (net.Conn, error)) HandleOption { + return func(opts *HandleOptions) { + opts.Dial = dial + } +} + +func WithHTTPKeepalive(keepalive bool) HandleOption { + return func(opts *HandleOptions) { + opts.HTTPKeepalive = keepalive + } +} + +func WithNode(node *chain.Node) HandleOption { + return func(opts *HandleOptions) { + opts.Node = node + } +} + +func WithHop(hop hop.Hop) HandleOption { + return func(opts *HandleOptions) { + opts.Hop = hop + } +} + +func WithBypass(bypass bypass.Bypass) HandleOption { + return func(opts *HandleOptions) { + opts.Bypass = bypass + } +} + +func WithRecorderObject(ro *xrecorder.HandlerRecorderObject) HandleOption { + return func(opts *HandleOptions) { + opts.RecorderObject = ro + } +} + +func WithLog(log logger.Logger) HandleOption { + return func(opts *HandleOptions) { + opts.Log = log + } +} + +type Sniffer struct { + Recorder recorder.Recorder + RecorderOptions *recorder.Options + + // MITM TLS termination + Certificate *x509.Certificate + PrivateKey crypto.PrivateKey + NegotiatedProtocol string + CertPool tls_util.CertPool + MitmBypass bypass.Bypass + + ReadTimeout time.Duration +} + +func (h *Sniffer) HandleHTTP(ctx context.Context, conn net.Conn, opts ...HandleOption) error { + var ho HandleOptions + for _, opt := range opts { + opt(&ho) + } + + pStats := stats.Stats{} + conn = stats_wrapper.WrapConn(conn, &pStats) + + br := bufio.NewReader(conn) + req, err := http.ReadRequest(br) + if err != nil { + return err + } + + if ho.Log.IsLevelEnabled(logger.TraceLevel) { + dump, _ := httputil.DumpRequest(req, false) + ho.Log.Trace(string(dump)) + } + + ro := ho.RecorderObject + 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)) + } + + // http/2 + if req.Method == "PRI" && len(req.Header) == 0 && req.URL.Path == "*" && req.Proto == "HTTP/2.0" { + return h.serveH2(ctx, xnet.NewReadWriteConn(br, conn, conn), &ho) + } + + node, cc, err := h.dial(ctx, conn, req, &ho) + if err != nil { + return err + } + defer cc.Close() + + log := ho.Log + 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 + } + + 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, node, req, &pStats, &ho); err != nil || shouldClose { + return err + } + } +} + +func (h *Sniffer) dial(ctx context.Context, conn net.Conn, req *http.Request, ho *HandleOptions) (node *chain.Node, cc net.Conn, err error) { + dial := ho.Dial + if dial == nil { + dial = (&net.Dialer{}).DialContext + } + + if node = ho.Node; node != nil { + cc, err = dial(ctx, "tcp", node.Addr) + return + } + + ro := ho.RecorderObject + + res := &http.Response{ + ProtoMajor: req.ProtoMajor, + ProtoMinor: req.ProtoMinor, + Header: http.Header{}, + StatusCode: http.StatusServiceUnavailable, + } + + host := req.Host + if host != "" { + if _, _, err := net.SplitHostPort(host); err != nil { + host = net.JoinHostPort(strings.Trim(host, "[]"), "80") + } + ro.Host = host + ho.Log = ho.Log.WithFields(map[string]any{ + "host": host, + }) + + if ho.Bypass != nil && + ho.Bypass.Contains(ctx, "tcp", host, bypass.WithPathOption(req.RequestURI)) { + ho.Log.Debugf("bypass: %s %s", host, req.RequestURI) + res.StatusCode = http.StatusForbidden + ro.HTTP.StatusCode = res.StatusCode + res.Write(conn) + return nil, nil, xbypass.ErrBypass + } + } + + node = &chain.Node{ + Addr: host, + } + if ho.Hop != nil { + node = ho.Hop.Select(ctx, + hop.HostSelectOption(host), + hop.ProtocolSelectOption(sniffing.ProtoHTTP), + hop.PathSelectOption(req.URL.Path), + ) + } + if node == nil { + ho.Log.Warnf("node for %s not found", host) + res.StatusCode = http.StatusBadGateway + ro.HTTP.StatusCode = res.StatusCode + res.Write(conn) + return nil, nil, errors.New("node not available") + } + + ro.Host = node.Addr + ho.Log = ho.Log.WithFields(map[string]any{ + "node": node.Name, + "dst": node.Addr, + }) + ho.Log.Debugf("find node for host %s -> %s(%s)", host, node.Name, node.Addr) + + cc, err = dial(ctx, "tcp", node.Addr) + if err != nil { + // TODO: the router itself may be failed due to the failed node in the router, + // the dead marker may be a wrong operation. + if marker := node.Marker(); marker != nil { + marker.Mark() + } + ho.Log.Warnf("connect to node %s(%s) failed: %v", node.Name, node.Addr, err) + res.Write(conn) + return + } + if marker := node.Marker(); marker != nil { + marker.Reset() + } + + if tlsSettings := node.Options().TLS; tlsSettings != nil { + cfg := &tls.Config{ + ServerName: tlsSettings.ServerName, + InsecureSkipVerify: !tlsSettings.Secure, + } + tls_util.SetTLSOptions(cfg, &config.TLSOptions{ + MinVersion: tlsSettings.Options.MinVersion, + MaxVersion: tlsSettings.Options.MaxVersion, + CipherSuites: tlsSettings.Options.CipherSuites, + ALPN: tlsSettings.Options.ALPN, + }) + cc = tls.Client(cc, cfg) + } + return +} + +func (h *Sniffer) serveH2(ctx context.Context, conn net.Conn, ho *HandleOptions) error { + const expectedBody = "SM\r\n\r\n" + + buf := make([]byte, len(expectedBody)) + n, err := io.ReadFull(conn, buf) + if err != nil { + return fmt.Errorf("h2: error reading client preface: %s", err) + } + if string(buf[:n]) != expectedBody { + return errors.New("h2: invalid client preface") + } + + ro := ho.RecorderObject + log := ho.Log + + ro.Time = time.Time{} + + tr := &http2.Transport{ + DialTLSContext: func(ctx context.Context, network, addr string, cfg *tls.Config) (net.Conn, error) { + node, cc, err := h.dialTLS(ctx, addr, ho) + if err != nil { + return nil, err + } + ho.Log.Debugf("connected to node %s(%s)", node.Name, node.Addr) + return cc, nil + }, + } + defer tr.CloseIdleConnections() + + (&http2.Server{}).ServeConn(conn, &http2.ServeConnOpts{ + Context: ctx, + SawClientPreface: true, + Handler: &h2Handler{ + transport: tr, + recorder: h.Recorder, + recorderOptions: h.RecorderOptions, + recorderObject: ro, + log: log, + }, + }) + return nil +} + +func (h *Sniffer) httpRoundTrip(ctx context.Context, rw, cc io.ReadWriter, node *chain.Node, req *http.Request, pStats *stats.Stats, ho *HandleOptions) (close bool, err error) { + close = true + + log := ho.Log + ro := &xrecorder.HandlerRecorderObject{} + *ro = *ho.RecorderObject + + 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); 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 !ho.HTTPKeepalive { + req.Header.Set("Connection", "close") + } + + var bodyRewrites []chain.HTTPBodyRewriteSettings + if httpSettings := node.Options().HTTP; httpSettings != nil { + if auther := httpSettings.Auther; auther != nil { + username, password, _ := req.BasicAuth() + id, ok := auther.Authenticate(ctx, username, password) + if !ok { + res.StatusCode = http.StatusUnauthorized + ro.HTTP.StatusCode = res.StatusCode + res.Header.Set("WWW-Authenticate", "Basic") + log.Warnf("node %s(%s) 401 unauthorized", node.Name, node.Addr) + res.Write(rw) + err = errors.New("unauthorized") + return + } + ctx = ctxvalue.ContextWithClientID(ctx, ctxvalue.ClientID(id)) + } + + if httpSettings.Host != "" { + req.Host = httpSettings.Host + } + for k, v := range httpSettings.Header { + req.Header.Set(k, v) + } + + for _, re := range httpSettings.RewriteURL { + if re.Pattern.MatchString(req.URL.Path) { + if s := re.Pattern.ReplaceAllString(req.URL.Path, re.Replacement); s != "" { + req.URL.Path = s + break + } + } + } + + bodyRewrites = httpSettings.RewriteBody + } + + var reqBody *xhttp.Body + if opts := h.RecorderOptions; 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 { + 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) + 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 + } + + if err = h.rewriteBody(resp, bodyRewrites...); err != nil { + log.Errorf("rewrite body: %v", err) + return + } + + var respBody *xhttp.Body + if opts := h.RecorderOptions; 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 +} + +func (h *Sniffer) rewriteBody(resp *http.Response, rewrites ...chain.HTTPBodyRewriteSettings) error { + if resp == nil || len(rewrites) == 0 || resp.ContentLength <= 0 { + return nil + } + + if encoding := resp.Header.Get("Content-Encoding"); encoding != "" { + return nil + } + + body, err := drainBody(resp.Body) + if err != nil || body == nil { + return err + } + + contentType, _, _ := strings.Cut(resp.Header.Get("Content-Type"), ";") + for _, rewrite := range rewrites { + rewriteType := rewrite.Type + if rewriteType == "" { + rewriteType = "text/html" + } + if rewriteType != "*" && !strings.Contains(rewriteType, contentType) { + continue + } + + body = rewrite.Pattern.ReplaceAll(body, rewrite.Replacement) + } + + resp.Body = io.NopCloser(bytes.NewReader(body)) + resp.ContentLength = int64(len(body)) + + return nil +} + +func drainBody(b io.ReadCloser) (body []byte, err error) { + if b == nil || b == http.NoBody { + // No copying needed. Preserve the magic sentinel meaning of NoBody. + return nil, nil + } + var buf bytes.Buffer + if _, err = buf.ReadFrom(b); err != nil { + return nil, err + } + if err = b.Close(); err != nil { + return nil, err + } + return buf.Bytes(), nil +} + +func (h *Sniffer) HandleTLS(ctx context.Context, conn net.Conn, opts ...HandleOption) error { + var ho HandleOptions + for _, opt := range opts { + opt(&ho) + } + + buf := new(bytes.Buffer) + clientHello, err := dissector.ParseClientHello(io.TeeReader(conn, buf)) + if err != nil { + return err + } + + ro := ho.RecorderObject + ro.TLS = &xrecorder.TLSRecorderObject{ + ServerName: clientHello.ServerName, + ClientHello: hex.EncodeToString(buf.Bytes()), + } + if len(clientHello.SupportedProtos) > 0 { + ro.TLS.Proto = clientHello.SupportedProtos[0] + } + + ctx = ctxvalue.ContextWithClientAddr(ctx, ctxvalue.ClientAddr(ro.RemoteAddr)) + + host := clientHello.ServerName + if host != "" { + if _, _, err := net.SplitHostPort(host); err != nil { + host = net.JoinHostPort(strings.Trim(host, "[]"), "443") + } + ro.Host = host + + if ho.Bypass != nil && ho.Bypass.Contains(ctx, "tcp", host) { + return xbypass.ErrBypass + } + } + + node, cc, err := h.dialTLS(ctx, host, &ho) + if err != nil { + return err + } + defer cc.Close() + ho.Node = node + + log := ho.Log + log.Debugf("connected to node %s(%s)", node.Name, node.Addr) + + if h.Certificate != nil && h.PrivateKey != nil && + len(clientHello.SupportedProtos) > 0 && (clientHello.SupportedProtos[0] == "h2" || clientHello.SupportedProtos[0] == "http/1.1") { + if host == "" { + host = ro.Host + } + if h.MitmBypass == nil || !h.MitmBypass.Contains(ctx, "tcp", host) { + return h.terminateTLS(ctx, xnet.NewReadWriteConn(io.MultiReader(buf, conn), conn, conn), cc, clientHello, &ho) + } + } + + if _, err := buf.WriteTo(cc); err != nil { + return err + } + + xio.SetReadDeadline(cc, time.Now().Add(h.ReadTimeout)) + serverHello, err := dissector.ParseServerHello(io.TeeReader(cc, buf)) + xio.SetReadDeadline(cc, time.Time{}) + + if serverHello != nil { + ro.TLS.CipherSuite = tls_util.CipherSuite(serverHello.CipherSuite).String() + ro.TLS.CompressionMethod = serverHello.CompressionMethod + if serverHello.Proto != "" { + ro.TLS.Proto = serverHello.Proto + } + if serverHello.Version > 0 { + ro.TLS.Version = tls_util.Version(serverHello.Version).String() + } + } + + if buf.Len() > 0 { + ro.TLS.ServerHello = hex.EncodeToString(buf.Bytes()) + } + + if _, err := buf.WriteTo(conn); err != nil { + return err + } + + log.Infof("%s <-> %s", ro.RemoteAddr, ro.Host) + xnet.Transport(conn, cc) + log.WithFields(map[string]any{ + "duration": time.Since(ro.Time), + }).Infof("%s >-< %s", ro.RemoteAddr, ro.Host) + + return err +} + +func (h *Sniffer) dialTLS(ctx context.Context, host string, ho *HandleOptions) (node *chain.Node, cc net.Conn, err error) { + dial := ho.Dial + if dial == nil { + dial = (&net.Dialer{}).DialContext + } + + if node = ho.Node; node != nil { + cc, err = dial(ctx, "tcp", node.Addr) + return + } + + if host != "" { + node = &chain.Node{ + Addr: host, + } + } + if ho.Hop != nil { + node = ho.Hop.Select(ctx, + hop.HostSelectOption(host), + hop.ProtocolSelectOption(sniffing.ProtoTLS), + ) + } + if node == nil { + err = errors.New("node not available") + return + } + + ro := ho.RecorderObject + addr := node.Addr + if opts := node.Options(); opts != nil { + switch opts.Network { + case "unix": + ro.Network = opts.Network + default: + if _, _, err := net.SplitHostPort(addr); err != nil { + addr += ":443" + } + } + } + ro.Host = addr + + ho.Log = ho.Log.WithFields(map[string]any{ + "host": host, + "node": node.Name, + "dst": fmt.Sprintf("%s/%s", addr, ro.Network), + }) + ho.Log.Debugf("find node for host %s -> %s(%s)", host, node.Name, addr) + + cc, err = dial(ctx, ro.Network, addr) + if err != nil { + // TODO: the router itself may be failed due to the failed node in the router, + // the dead marker may be a wrong operation. + if marker := node.Marker(); marker != nil { + marker.Mark() + } + ho.Log.Warnf("connect to node %s(%s) failed: %v", node.Name, node.Addr, err) + return + } + + if marker := node.Marker(); marker != nil { + marker.Reset() + } + + if tlsSettings := node.Options().TLS; tlsSettings != nil { + cfg := &tls.Config{ + ServerName: tlsSettings.ServerName, + InsecureSkipVerify: !tlsSettings.Secure, + } + tls_util.SetTLSOptions(cfg, &config.TLSOptions{ + MinVersion: tlsSettings.Options.MinVersion, + MaxVersion: tlsSettings.Options.MaxVersion, + CipherSuites: tlsSettings.Options.CipherSuites, + ALPN: tlsSettings.Options.ALPN, + }) + cc = tls.Client(cc, cfg) + } + return +} + +func (h *Sniffer) terminateTLS(ctx context.Context, conn, cc net.Conn, clientHello *dissector.ClientHelloInfo, ho *HandleOptions) error { + ro := ho.RecorderObject + log := ho.Log + + nextProtos := clientHello.SupportedProtos + if h.NegotiatedProtocol != "" { + nextProtos = []string{h.NegotiatedProtocol} + } + + cfg := &tls.Config{ + ServerName: clientHello.ServerName, + NextProtos: nextProtos, + CipherSuites: clientHello.CipherSuites, + } + if cfg.ServerName == "" { + cfg.InsecureSkipVerify = true + } + clientConn := tls.Client(cc, cfg) + if err := clientConn.HandshakeContext(ctx); err != nil { + return err + } + + cs := clientConn.ConnectionState() + ro.TLS.CipherSuite = tls_util.CipherSuite(cs.CipherSuite).String() + ro.TLS.Proto = cs.NegotiatedProtocol + ro.TLS.Version = tls_util.Version(cs.Version).String() + + host := cfg.ServerName + if host == "" { + if host = cs.PeerCertificates[0].Subject.CommonName; host == "" { + host = ro.Host + } + } + if h, _, _ := net.SplitHostPort(host); h != "" { + host = h + } + + negotiatedProtocol := cs.NegotiatedProtocol + if h.NegotiatedProtocol != "" { + negotiatedProtocol = h.NegotiatedProtocol + } + nextProtos = nil + if negotiatedProtocol != "" { + nextProtos = []string{negotiatedProtocol} + } + + // cache the tls server handshake record. + wb := &bytes.Buffer{} + conn = xnet.NewReadWriteConn(conn, io.MultiWriter(wb, conn), conn) + + serverConn := tls.Server(conn, &tls.Config{ + NextProtos: nextProtos, + GetCertificate: func(chi *tls.ClientHelloInfo) (*tls.Certificate, error) { + certPool := h.CertPool + if certPool == nil { + certPool = DefaultCertPool + } + serverName := chi.ServerName + if serverName == "" { + serverName = host + } + cert, err := certPool.Get(serverName) + if cert != nil { + pool := x509.NewCertPool() + pool.AddCert(h.Certificate) + if _, err = cert.Verify(x509.VerifyOptions{ + DNSName: serverName, + Roots: pool, + }); err != nil { + log.Warnf("verify cached certificate for %s: %v", serverName, err) + cert = nil + } + } + if cert == nil { + cert, err = tls_util.GenerateCertificate(serverName, 7*24*time.Hour, h.Certificate, h.PrivateKey) + certPool.Put(serverName, cert) + } + if err != nil { + return nil, err + } + + return &tls.Certificate{ + Certificate: [][]byte{cert.Raw}, + PrivateKey: h.PrivateKey, + }, nil + }, + }) + err := serverConn.HandshakeContext(ctx) + if record, _ := dissector.ReadRecord(wb); record != nil { + wb.Reset() + record.WriteTo(wb) + ro.TLS.ServerHello = hex.EncodeToString(wb.Bytes()) + } + if err != nil { + return err + } + + opts := []HandleOption{ + WithDial(func(ctx context.Context, network, address string) (net.Conn, error) { + return clientConn, nil + }), + WithHTTPKeepalive(true), + WithNode(ho.Node), + WithRecorderObject(ro), + WithLog(log), + } + return h.HandleHTTP(ctx, serverConn, opts...) +} + +type h2Handler struct { + transport http.RoundTripper + recorder recorder.Recorder + recorderOptions *recorder.Options + recorderObject *xrecorder.HandlerRecorderObject + log logger.Logger +} + +func (h *h2Handler) ServeHTTP(w http.ResponseWriter, r *http.Request) { + log := h.log + + ro := &xrecorder.HandlerRecorderObject{} + *ro = *h.recorderObject + ro.Time = time.Now() + + var err error + log.Infof("%s <-> %s", ro.RemoteAddr, r.Host) + defer func() { + ro.Duration = time.Since(ro.Time) + if err != nil { + ro.Err = err.Error() + } + if err := ro.Record(r.Context(), h.recorder); err != nil { + log.Errorf("record: %v", err) + } + + log.WithFields(map[string]any{ + "duration": time.Since(ro.Time), + }).Infof("%s >-< %s", ro.RemoteAddr, r.Host) + }() + + if clientIP := xhttp.GetClientIP(r); clientIP != nil { + ro.ClientIP = clientIP.String() + } + ro.HTTP = &xrecorder.HTTPRecorderObject{ + Host: r.Host, + Proto: r.Proto, + Scheme: "https", + Method: r.Method, + URI: r.RequestURI, + Request: xrecorder.HTTPRequestRecorderObject{ + ContentLength: r.ContentLength, + Header: r.Header.Clone(), + }, + } + + if log.IsLevelEnabled(logger.TraceLevel) { + dump, _ := httputil.DumpRequest(r, false) + log.Trace(string(dump)) + } + + url := r.URL + url.Scheme = "https" + url.Host = r.Host + req := &http.Request{ + Method: r.Method, + URL: url, + Host: r.Host, + Header: r.Header, + Body: r.Body, + ContentLength: r.ContentLength, + Trailer: r.Trailer, + } + + var reqBody *xhttp.Body + if opts := h.recorderOptions; 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 + } + } + + resp, err := h.transport.RoundTrip(req.WithContext(r.Context())) + if reqBody != nil { + ro.HTTP.Request.Body = reqBody.Content() + ro.HTTP.Request.ContentLength = reqBody.Length() + } + if err != nil { + log.Error(err) + w.WriteHeader(http.StatusServiceUnavailable) + return + } + defer resp.Body.Close() + + 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)) + } + + h.setHeader(w, resp.Header) + w.WriteHeader(resp.StatusCode) + + var respBody *xhttp.Body + if opts := h.recorderOptions; opts != nil && opts.HTTPBody { + maxSize := opts.MaxBodySize + if maxSize <= 0 { + maxSize = DefaultBodySize + } + respBody = xhttp.NewBody(resp.Body, maxSize) + resp.Body = respBody + } + + io.Copy(w, resp.Body) + + if respBody != nil { + ro.HTTP.Response.Body = respBody.Content() + ro.HTTP.Response.ContentLength = respBody.Length() + } +} + +func (h *h2Handler) setHeader(w http.ResponseWriter, header http.Header) { + for k, v := range header { + for i := range v { + w.Header().Add(k, v[i]) + } + } +} diff --git a/internal/util/sniffing/sniffer.go b/internal/util/sniffing/sniffer.go index b78964ac..b7a00c3a 100644 --- a/internal/util/sniffing/sniffer.go +++ b/internal/util/sniffing/sniffer.go @@ -23,6 +23,7 @@ import ( "github.com/go-gost/core/recorder" dissector "github.com/go-gost/tls-dissector" xbypass "github.com/go-gost/x/bypass" + 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" @@ -97,14 +98,11 @@ type Sniffer struct { } func (h *Sniffer) HandleHTTP(ctx context.Context, conn net.Conn, opts ...HandleOption) error { - var o HandleOptions + var ho HandleOptions for _, opt := range opts { - opt(&o) + opt(&ho) } - ro := o.RecorderObject - log := o.Log - pStats := stats.Stats{} conn = stats_wrapper.WrapConn(conn, &pStats) @@ -113,11 +111,31 @@ func (h *Sniffer) HandleHTTP(ctx context.Context, conn net.Conn, opts ...HandleO if err != nil { return err } + + log := ho.Log if log.IsLevelEnabled(logger.TraceLevel) { dump, _ := httputil.DumpRequest(req, false) log.Trace(string(dump)) } + ro := ho.RecorderObject + 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)) + + // http/2 + if req.Method == "PRI" && len(req.Header) == 0 && req.URL.Path == "*" && req.Proto == "HTTP/2.0" { + return h.serveH2(ctx, xnet.NewReadWriteConn(br, conn, conn), &ho) + } + host := req.Host if host != "" { if _, _, err := net.SplitHostPort(host); err != nil { @@ -129,59 +147,12 @@ func (h *Sniffer) HandleHTTP(ctx context.Context, conn net.Conn, opts ...HandleO "host": host, }) - if o.Bypass != nil && o.Bypass.Contains(ctx, "tcp", host) { + if ho.Bypass != nil && ho.Bypass.Contains(ctx, "tcp", host) { return xbypass.ErrBypass } } - // http/2 - if req.Method == "PRI" && len(req.Header) == 0 && req.URL.Path == "*" && req.Proto == "HTTP/2.0" { - const expectedBody = "SM\r\n\r\n" - - buf := make([]byte, len(expectedBody)) - n, err := io.ReadFull(br, buf) - if err != nil { - return fmt.Errorf("h2: error reading client preface: %s", err) - } - if string(buf[:n]) != expectedBody { - return errors.New("h2: invalid client preface") - } - - ro.Time = time.Time{} - - tr := &http2.Transport{ - DialTLSContext: func(ctx context.Context, network, addr string, cfg *tls.Config) (net.Conn, error) { - if dial := o.DialTLS; dial != nil { - return dial(ctx, network, addr, cfg) - } - - cc, err := (&net.Dialer{}).DialContext(ctx, network, addr) - if err != nil { - return nil, err - } - - cc = tls.Client(cc, cfg) - return cc, nil - }, - } - defer tr.CloseIdleConnections() - - h2s := &http2.Server{} - h2s.ServeConn(xnet.NewReadWriteConn(br, conn, conn), &http2.ServeConnOpts{ - Context: ctx, - SawClientPreface: true, - Handler: &h2Handler{ - transport: tr, - recorder: h.Recorder, - recorderOptions: h.RecorderOptions, - recorderObject: ro, - log: log, - }, - }) - return nil - } - - dial := o.Dial + dial := ho.Dial if dial == nil { dial = (&net.Dialer{}).DialContext } @@ -220,6 +191,54 @@ func (h *Sniffer) HandleHTTP(ctx context.Context, conn net.Conn, opts ...HandleO } } +func (h *Sniffer) serveH2(ctx context.Context, conn net.Conn, ho *HandleOptions) error { + const expectedBody = "SM\r\n\r\n" + + buf := make([]byte, len(expectedBody)) + n, err := io.ReadFull(conn, buf) + if err != nil { + return fmt.Errorf("h2: error reading client preface: %s", err) + } + if string(buf[:n]) != expectedBody { + return errors.New("h2: invalid client preface") + } + + ro := ho.RecorderObject + log := ho.Log + + ro.Time = time.Time{} + + tr := &http2.Transport{ + DialTLSContext: func(ctx context.Context, network, addr string, cfg *tls.Config) (net.Conn, error) { + if dial := ho.DialTLS; dial != nil { + return dial(ctx, network, addr, cfg) + } + + cc, err := (&net.Dialer{}).DialContext(ctx, network, addr) + if err != nil { + return nil, err + } + + cc = tls.Client(cc, cfg) + return cc, nil + }, + } + defer tr.CloseIdleConnections() + + (&http2.Server{}).ServeConn(conn, &http2.ServeConnOpts{ + Context: ctx, + SawClientPreface: true, + Handler: &h2Handler{ + transport: tr, + recorder: h.Recorder, + recorderOptions: h.RecorderOptions, + recorderObject: ro, + log: log, + }, + }) + return nil +} + func (h *Sniffer) 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 @@ -245,9 +264,6 @@ func (h *Sniffer) httpRoundTrip(ctx context.Context, rw, cc io.ReadWriter, req * }).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, @@ -345,9 +361,9 @@ func (h *Sniffer) httpRoundTrip(ctx context.Context, rw, cc io.ReadWriter, req * } func (h *Sniffer) HandleTLS(ctx context.Context, conn net.Conn, opts ...HandleOption) error { - var o HandleOptions + var ho HandleOptions for _, opt := range opts { - opt(&o) + opt(&ho) } buf := new(bytes.Buffer) @@ -356,7 +372,9 @@ func (h *Sniffer) HandleTLS(ctx context.Context, conn net.Conn, opts ...HandleOp return err } - ro := o.RecorderObject + log := ho.Log + + ro := ho.RecorderObject ro.TLS = &xrecorder.TLSRecorderObject{ ServerName: clientHello.ServerName, ClientHello: hex.EncodeToString(buf.Bytes()), @@ -365,6 +383,8 @@ func (h *Sniffer) HandleTLS(ctx context.Context, conn net.Conn, opts ...HandleOp ro.TLS.Proto = clientHello.SupportedProtos[0] } + ctx = ctxvalue.ContextWithClientAddr(ctx, ctxvalue.ClientAddr(ro.RemoteAddr)) + host := clientHello.ServerName if host != "" { if _, _, err := net.SplitHostPort(host); err != nil { @@ -372,12 +392,12 @@ func (h *Sniffer) HandleTLS(ctx context.Context, conn net.Conn, opts ...HandleOp } ro.Host = host - if o.Bypass != nil && o.Bypass.Contains(ctx, "tcp", host) { + if ho.Bypass != nil && ho.Bypass.Contains(ctx, "tcp", host) { return xbypass.ErrBypass } } - dial := o.Dial + dial := ho.Dial if dial == nil { dial = (&net.Dialer{}).DialContext } @@ -393,7 +413,7 @@ func (h *Sniffer) HandleTLS(ctx context.Context, conn net.Conn, opts ...HandleOp host = ro.Host } if h.MitmBypass == nil || !h.MitmBypass.Contains(ctx, "tcp", host) { - return h.terminateTLS(ctx, xnet.NewReadWriteConn(io.MultiReader(buf, conn), conn, conn), cc, clientHello, &o) + return h.terminateTLS(ctx, xnet.NewReadWriteConn(io.MultiReader(buf, conn), conn, conn), cc, clientHello, &ho) } } @@ -424,18 +444,18 @@ func (h *Sniffer) HandleTLS(ctx context.Context, conn net.Conn, opts ...HandleOp return err } - o.Log.Infof("%s <-> %s", ro.RemoteAddr, ro.Host) + log.Infof("%s <-> %s", ro.RemoteAddr, ro.Host) xnet.Transport(conn, cc) - o.Log.WithFields(map[string]any{ + log.WithFields(map[string]any{ "duration": time.Since(ro.Time), }).Infof("%s >-< %s", ro.RemoteAddr, ro.Host) return err } -func (h *Sniffer) terminateTLS(ctx context.Context, conn, cc net.Conn, clientHello *dissector.ClientHelloInfo, o *HandleOptions) error { - ro := o.RecorderObject - log := o.Log +func (h *Sniffer) terminateTLS(ctx context.Context, conn, cc net.Conn, clientHello *dissector.ClientHelloInfo, ho *HandleOptions) error { + ro := ho.RecorderObject + log := ho.Log nextProtos := clientHello.SupportedProtos if h.NegotiatedProtocol != "" {