From a97ac1f30a3021f7747b19b8062a7470c0126861 Mon Sep 17 00:00:00 2001 From: ginuerzh Date: Tue, 22 Oct 2024 22:51:33 +0800 Subject: [PATCH] fix http body recorder --- admission/admission.go | 19 +++++++++++++ auth/auth.go | 26 ++++++++++++++++++ bypass/bypass.go | 43 ++++++++++++++++++++++++++++++ config/parsing/hop/parse.go | 4 +-- config/parsing/node/parse.go | 6 ++--- config/parsing/service/parse.go | 15 ++++++----- handler/http/handler.go | 19 +++++++------ handler/tunnel/bind.go | 12 ++++++--- handler/tunnel/connect.go | 2 +- handler/tunnel/dialer.go | 4 +-- handler/tunnel/entrypoint.go | 19 +++++++------ internal/util/forwarder/sniffer.go | 20 ++++++++------ internal/util/sniffing/sniffer.go | 18 ++++++++----- sd/plugin/grpc.go | 3 ++- 14 files changed, 159 insertions(+), 51 deletions(-) diff --git a/admission/admission.go b/admission/admission.go index 6be5de32..2723594b 100644 --- a/admission/admission.go +++ b/admission/admission.go @@ -271,3 +271,22 @@ func (p *localAdmission) Close() error { } return nil } + +type admissionGroup struct { + admissions []admission.Admission +} + +func AdmissionGroup(admissions ...admission.Admission) admission.Admission { + return &admissionGroup{ + admissions: admissions, + } +} + +func (p *admissionGroup) Admit(ctx context.Context, addr string, opts ...admission.Option) bool { + for _, admission := range p.admissions { + if admission != nil && !admission.Admit(ctx, addr, opts...) { + return false + } + } + return true +} diff --git a/auth/auth.go b/auth/auth.go index ed6c5069..ef35f03c 100644 --- a/auth/auth.go +++ b/auth/auth.go @@ -252,3 +252,29 @@ func (p *authenticator) Close() error { } return nil } + +type authenticatorGroup struct { + authers []auth.Authenticator +} + +func AuthenticatorGroup(authers ...auth.Authenticator) auth.Authenticator { + return &authenticatorGroup{ + authers: authers, + } +} + +func (p *authenticatorGroup) Authenticate(ctx context.Context, user, password string, opts ...auth.Option) (string, bool) { + if len(p.authers) == 0 { + return "", false + } + for _, auther := range p.authers { + if auther == nil { + continue + } + + if id, ok := auther.Authenticate(ctx, user, password, opts...); ok { + return id, ok + } + } + return "", false +} diff --git a/bypass/bypass.go b/bypass/bypass.go index a2709d19..05dd559b 100644 --- a/bypass/bypass.go +++ b/bypass/bypass.go @@ -6,6 +6,7 @@ import ( "errors" "io" "net" + "slices" "strings" "sync" "time" @@ -288,3 +289,45 @@ func (bp *localBypass) Close() error { } return nil } + +type bypassGroup struct { + bypasses []bypass.Bypass +} + +func BypassGroup(bypasses ...bypass.Bypass) bypass.Bypass { + return &bypassGroup{ + bypasses: bypasses, + } +} + +func (p *bypassGroup) Contains(ctx context.Context, network, addr string, opts ...bypass.Option) bool { + var whitelist, blacklist []bool + for _, bypass := range p.bypasses { + result := bypass.Contains(ctx, network, addr, opts...) + if bypass.IsWhitelist() { + whitelist = append(whitelist, result) + } else { + blacklist = append(blacklist, result) + } + } + status := false + if len(whitelist) > 0 { + if slices.Contains(whitelist, false) { + status = false + } else { + status = true + } + } + if !status && len(blacklist) > 0 { + if slices.Contains(blacklist, true) { + status = true + } else { + status = false + } + } + return status +} + +func (p *bypassGroup) IsWhitelist() bool { + return false +} diff --git a/config/parsing/hop/parse.go b/config/parsing/hop/parse.go index e2905c96..afe833f1 100644 --- a/config/parsing/hop/parse.go +++ b/config/parsing/hop/parse.go @@ -4,10 +4,10 @@ import ( "crypto/tls" "strings" - "github.com/go-gost/core/bypass" "github.com/go-gost/core/chain" "github.com/go-gost/core/hop" "github.com/go-gost/core/logger" + xbypass "github.com/go-gost/x/bypass" "github.com/go-gost/x/config" "github.com/go-gost/x/config/parsing" bypass_parser "github.com/go-gost/x/config/parsing/bypass" @@ -115,7 +115,7 @@ func ParseHop(cfg *config.HopConfig, log logger.Logger) (hop.Hop, error) { xhop.NameOption(cfg.Name), xhop.NodeOption(nodes...), xhop.SelectorOption(sel), - xhop.BypassOption(bypass.BypassGroup(bypass_parser.List(cfg.Bypass, cfg.Bypasses...)...)), + xhop.BypassOption(xbypass.BypassGroup(bypass_parser.List(cfg.Bypass, cfg.Bypasses...)...)), xhop.ReloadPeriodOption(cfg.Reload), xhop.LoggerOption(log.WithFields(map[string]any{ "kind": "hop", diff --git a/config/parsing/node/parse.go b/config/parsing/node/parse.go index c49f1dac..69a9b1ea 100644 --- a/config/parsing/node/parse.go +++ b/config/parsing/node/parse.go @@ -6,14 +6,13 @@ import ( "regexp" "strings" - "github.com/go-gost/core/bypass" "github.com/go-gost/core/chain" "github.com/go-gost/core/connector" "github.com/go-gost/core/dialer" "github.com/go-gost/core/logger" "github.com/go-gost/core/metadata" - mdutil "github.com/go-gost/x/metadata/util" xauth "github.com/go-gost/x/auth" + xbypass "github.com/go-gost/x/bypass" xchain "github.com/go-gost/x/chain" "github.com/go-gost/x/config" "github.com/go-gost/x/config/parsing" @@ -21,6 +20,7 @@ import ( bypass_parser "github.com/go-gost/x/config/parsing/bypass" tls_util "github.com/go-gost/x/internal/util/tls" mdx "github.com/go-gost/x/metadata" + mdutil "github.com/go-gost/x/metadata/util" "github.com/go-gost/x/registry" ) @@ -149,7 +149,7 @@ func ParseNode(hop string, cfg *config.NodeConfig, log logger.Logger) (*chain.No opts := []chain.NodeOption{ chain.TransportNodeOption(tr), - chain.BypassNodeOption(bypass.BypassGroup(bypass_parser.List(cfg.Bypass, cfg.Bypasses...)...)), + chain.BypassNodeOption(xbypass.BypassGroup(bypass_parser.List(cfg.Bypass, cfg.Bypasses...)...)), chain.ResoloverNodeOption(registry.ResolverRegistry().Get(cfg.Resolver)), chain.HostMapperNodeOption(registry.HostsRegistry().Get(cfg.Hosts)), chain.MetadataNodeOption(nm), diff --git a/config/parsing/service/parse.go b/config/parsing/service/parse.go index 7433e647..3dc94e0d 100644 --- a/config/parsing/service/parse.go +++ b/config/parsing/service/parse.go @@ -6,9 +6,7 @@ import ( "strings" "time" - "github.com/go-gost/core/admission" "github.com/go-gost/core/auth" - "github.com/go-gost/core/bypass" "github.com/go-gost/core/chain" "github.com/go-gost/core/handler" "github.com/go-gost/core/hop" @@ -18,6 +16,9 @@ import ( "github.com/go-gost/core/recorder" "github.com/go-gost/core/selector" "github.com/go-gost/core/service" + xadmission "github.com/go-gost/x/admission" + xauth "github.com/go-gost/x/auth" + xbypass "github.com/go-gost/x/bypass" xchain "github.com/go-gost/x/chain" "github.com/go-gost/x/config" "github.com/go-gost/x/config/parsing" @@ -84,7 +85,7 @@ func ParseService(cfg *config.ServiceConfig) (service.Service, error) { } var auther auth.Authenticator if len(authers) > 0 { - auther = auth.AuthenticatorGroup(authers...) + auther = xauth.AuthenticatorGroup(authers...) } admissions := admission_parser.List(cfg.Admission, cfg.Admissions...) @@ -155,7 +156,7 @@ func ParseService(cfg *config.ServiceConfig) (service.Service, error) { listener.AutherOption(auther), listener.AuthOption(auth_parser.Info(cfg.Listener.Auth)), listener.TLSConfigOption(tlsConfig), - listener.AdmissionOption(admission.AdmissionGroup(admissions...)), + listener.AdmissionOption(xadmission.AdmissionGroup(admissions...)), listener.TrafficLimiterOption(registry.TrafficLimiterRegistry().Get(cfg.Limiter)), listener.ConnLimiterOption(registry.ConnLimiterRegistry().Get(cfg.CLimiter)), listener.ServiceOption(cfg.Name), @@ -235,7 +236,7 @@ func ParseService(cfg *config.ServiceConfig) (service.Service, error) { auther = nil if len(authers) > 0 { - auther = auth.AuthenticatorGroup(authers...) + auther = xauth.AuthenticatorGroup(authers...) } var recorders []recorder.RecorderObject @@ -277,7 +278,7 @@ func ParseService(cfg *config.ServiceConfig) (service.Service, error) { handler.RouterOption(xchain.NewRouter(routerOpts...)), handler.AutherOption(auther), handler.AuthOption(auth_parser.Info(cfg.Handler.Auth)), - handler.BypassOption(bypass.BypassGroup(bypass_parser.List(cfg.Bypass, cfg.Bypasses...)...)), + handler.BypassOption(xbypass.BypassGroup(bypass_parser.List(cfg.Bypass, cfg.Bypasses...)...)), handler.TLSConfigOption(tlsConfig), handler.RateLimiterOption(registry.RateLimiterRegistry().Get(cfg.RLimiter)), handler.TrafficLimiterOption(registry.TrafficLimiterRegistry().Get(cfg.Handler.Limiter)), @@ -309,7 +310,7 @@ func ParseService(cfg *config.ServiceConfig) (service.Service, error) { } s := xservice.NewService(cfg.Name, ln, h, - xservice.AdmissionOption(admission.AdmissionGroup(admissions...)), + xservice.AdmissionOption(xadmission.AdmissionGroup(admissions...)), xservice.PreUpOption(preUp), xservice.PreDownOption(preDown), xservice.PostUpOption(postUp), diff --git a/handler/http/handler.go b/handler/http/handler.go index 58a44c84..520c4a5f 100644 --- a/handler/http/handler.go +++ b/handler/http/handler.go @@ -561,17 +561,18 @@ func (h *httpHandler) proxyRoundTrip(ctx context.Context, rw io.ReadWriter, req ctx = ctxvalue.ContextWithLogger(ctx, log) 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() } + if err != nil { + res.Write(rw) + return + } + defer resp.Body.Close() + ro.HTTP.StatusCode = resp.StatusCode ro.HTTP.Response.Header = resp.Header.Clone() ro.HTTP.Response.ContentLength = resp.ContentLength @@ -607,15 +608,17 @@ func (h *httpHandler) proxyRoundTrip(ctx context.Context, rw io.ReadWriter, req resp.Body = respBody } - if err = resp.Write(rw); err != nil { - return fmt.Errorf("write response: %v", err) - } + err = resp.Write(rw) if respBody != nil { ro.HTTP.Response.Body = respBody.Content() ro.HTTP.Response.ContentLength = respBody.Length() } + if err != nil { + return fmt.Errorf("write response: %v", err) + } + return } diff --git a/handler/tunnel/bind.go b/handler/tunnel/bind.go index 5d76c3ef..9492f275 100644 --- a/handler/tunnel/bind.go +++ b/handler/tunnel/bind.go @@ -37,11 +37,15 @@ func (h *tunnelHandler) handleBind(ctx context.Context, conn net.Conn, network, v := md5.Sum([]byte(tunnelID.String())) endpoint := hex.EncodeToString(v[:8]) - addr := address - host, port, _ := net.SplitHostPort(addr) - if host == "" { - addr = net.JoinHostPort(endpoint, port) + host, port, _ := net.SplitHostPort(address) + if host == "" || h.md.ingress == nil { + host = endpoint + } else if host != endpoint { + if rule := h.md.ingress.GetRule(ctx, host); rule != nil && rule.Endpoint != tunnelID.String() { + host = endpoint + } } + addr := net.JoinHostPort(host, port) af := &relay.AddrFeature{} err = af.ParseFrom(addr) diff --git a/handler/tunnel/connect.go b/handler/tunnel/connect.go index ce9f423d..229d1a23 100644 --- a/handler/tunnel/connect.go +++ b/handler/tunnel/connect.go @@ -72,7 +72,7 @@ func (h *tunnelHandler) handleConnect(ctx context.Context, req *relay.Request, c } defer cc.Close() - log.Debugf("new connection to tunnel: %s, connector: %s", tunnelID, cid) + log.Debugf("connected to node=%s tunnel=%s connector=%s", node, tunnelID, cid) if node == h.id { if _, err := resp.WriteTo(conn); err != nil { diff --git a/handler/tunnel/dialer.go b/handler/tunnel/dialer.go index f86f387a..1b03a0fa 100644 --- a/handler/tunnel/dialer.go +++ b/handler/tunnel/dialer.go @@ -68,11 +68,11 @@ func (d *Dialer) Dial(ctx context.Context, network string, tid string) (conn net } node = service.Node - cid = service.Name + cid = service.ID dialer := net.Dialer{ Timeout: d.timeout, } - conn, err = dialer.DialContext(ctx, network, service.Address) + conn, err = dialer.DialContext(ctx, "tcp", service.Address) return } diff --git a/handler/tunnel/entrypoint.go b/handler/tunnel/entrypoint.go index eebfc039..57660bfd 100644 --- a/handler/tunnel/entrypoint.go +++ b/handler/tunnel/entrypoint.go @@ -324,6 +324,12 @@ func (ep *entrypoint) httpRoundTrip(ctx context.Context, rw io.ReadWriter, req * ctx = ctxvalue.ContextWithLogger(ctx, log) resp, err := ep.transport.RoundTrip(req.WithContext(ctx)) + + if reqBody != nil { + ro.HTTP.Request.Body = reqBody.Content() + ro.HTTP.Request.ContentLength = reqBody.Length() + } + if err != nil { if errors.Is(err, ErrTunnelRoute) || errors.Is(err, ErrPrivateTunnel) { res.StatusCode = http.StatusBadGateway @@ -334,11 +340,6 @@ func (ep *entrypoint) httpRoundTrip(ctx context.Context, rw io.ReadWriter, req * } 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 @@ -365,15 +366,17 @@ func (ep *entrypoint) httpRoundTrip(ctx context.Context, rw io.ReadWriter, req * resp.Body = respBody } - if err = resp.Write(rw); err != nil { - return fmt.Errorf("write response: %v", err) - } + err = resp.Write(rw) if respBody != nil { ro.HTTP.Response.Body = respBody.Content() ro.HTTP.Response.ContentLength = respBody.Length() } + if err != nil { + return fmt.Errorf("write response: %v", err) + } + return } diff --git a/internal/util/forwarder/sniffer.go b/internal/util/forwarder/sniffer.go index c01a6110..28dc6d6f 100644 --- a/internal/util/forwarder/sniffer.go +++ b/internal/util/forwarder/sniffer.go @@ -453,16 +453,18 @@ func (h *Sniffer) httpRoundTrip(ctx context.Context, rw, cc io.ReadWriter, node } } - if err = req.Write(cc); err != nil { - res.Write(rw) - return - } + err = req.Write(cc) if reqBody != nil { ro.HTTP.Request.Body = reqBody.Content() ro.HTTP.Request.ContentLength = reqBody.Length() } + if err != nil { + res.Write(rw) + return + } + xio.SetReadDeadline(cc, time.Now().Add(h.ReadTimeout)) resp, err := http.ReadResponse(bufio.NewReader(cc), req) if err != nil { @@ -514,16 +516,18 @@ func (h *Sniffer) httpRoundTrip(ctx context.Context, rw, cc io.ReadWriter, node resp.Body = respBody } - if err = resp.Write(rw); err != nil { - log.Errorf("write response: %v", err) - return - } + err = resp.Write(rw) if respBody != nil { ro.HTTP.Response.Body = respBody.Content() ro.HTTP.Response.ContentLength = respBody.Length() } + if err != nil { + log.Errorf("write response: %v", err) + return + } + return resp.Close, nil } diff --git a/internal/util/sniffing/sniffer.go b/internal/util/sniffing/sniffer.go index 9dbb1370..ef9ab3b3 100644 --- a/internal/util/sniffing/sniffer.go +++ b/internal/util/sniffing/sniffer.go @@ -323,15 +323,17 @@ func (h *Sniffer) httpRoundTrip(ctx context.Context, rw, cc io.ReadWriter, req * } } - if err = req.Write(cc); err != nil { - return - } + err = req.Write(cc) if reqBody != nil { ro.HTTP.Request.Body = reqBody.Content() ro.HTTP.Request.ContentLength = reqBody.Length() } + if err != nil { + return + } + xio.SetReadDeadline(cc, time.Now().Add(h.ReadTimeout)) resp, err := http.ReadResponse(bufio.NewReader(cc), req) if err != nil { @@ -377,16 +379,18 @@ func (h *Sniffer) httpRoundTrip(ctx context.Context, rw, cc io.ReadWriter, req * resp.Body = respBody } - if err = resp.Write(rw); err != nil { - err = fmt.Errorf("write response: %w", err) - return - } + err = resp.Write(rw) if respBody != nil { ro.HTTP.Response.Body = respBody.Content() ro.HTTP.Response.ContentLength = respBody.Length() } + if err != nil { + err = fmt.Errorf("write response: %w", err) + return + } + return resp.Close, nil } diff --git a/sd/plugin/grpc.go b/sd/plugin/grpc.go index 0ef7c01c..b8fe5a4b 100644 --- a/sd/plugin/grpc.go +++ b/sd/plugin/grpc.go @@ -117,8 +117,9 @@ func (p *grpcPlugin) Get(ctx context.Context, name string) ([]*sd.Service, error continue } services = append(services, &sd.Service{ - Node: v.Node, + ID: v.Id, Name: v.Name, + Node: v.Node, Network: v.Network, Address: v.Address, })