diff --git a/bypass/bypass.go b/bypass/bypass.go index 3da5cb76..2d1781b5 100644 --- a/bypass/bypass.go +++ b/bypass/bypass.go @@ -3,6 +3,7 @@ package bypass import ( "bufio" "context" + "errors" "io" "net" "strings" @@ -15,6 +16,10 @@ import ( "github.com/go-gost/x/internal/matcher" ) +var ( + ErrBypass = errors.New("bypass") +) + type options struct { whitelist bool matchers []string diff --git a/chain/router.go b/chain/router.go index b6b5aaf2..e777c1fa 100644 --- a/chain/router.go +++ b/chain/router.go @@ -10,6 +10,7 @@ import ( "github.com/go-gost/core/chain" "github.com/go-gost/core/logger" "github.com/go-gost/core/recorder" + ctxvalue "github.com/go-gost/x/ctx" xnet "github.com/go-gost/x/internal/net" ) @@ -48,7 +49,11 @@ func (r *Router) Dial(ctx context.Context, network, address string) (conn net.Co } r.record(ctx, recorder.RecorderServiceRouterDialAddress, []byte(host)) - conn, err = r.dial(ctx, network, address) + log := r.options.Logger.WithFields(map[string]any{ + "sid": ctxvalue.SidFromContext(ctx), + }) + + conn, err = r.dial(ctx, network, address, log) if err != nil { r.record(ctx, recorder.RecorderServiceRouterDialAddressError, []byte(host)) return @@ -69,22 +74,19 @@ func (r *Router) record(ctx context.Context, name string, data []byte) error { for _, rec := range r.options.Recorders { if rec.Record == name { - err := rec.Recorder.Record(ctx, data) - if err != nil { - r.options.Logger.Errorf("record %s: %v", name, err) - } - return err + return rec.Recorder.Record(ctx, data) } } return nil } -func (r *Router) dial(ctx context.Context, network, address string) (conn net.Conn, err error) { +func (r *Router) dial(ctx context.Context, network, address string, log logger.Logger) (conn net.Conn, err error) { count := r.options.Retries + 1 if count <= 0 { count = 1 } - r.options.Logger.Debugf("dial %s/%s", address, network) + + log.Debugf("dial %s/%s", address, network) for i := 0; i < count; i++ { ctx := ctx @@ -95,9 +97,9 @@ func (r *Router) dial(ctx context.Context, network, address string) (conn net.Co } var ipAddr string - ipAddr, err = xnet.Resolve(ctx, "ip", address, r.options.Resolver, r.options.HostMapper, r.options.Logger) + ipAddr, err = xnet.Resolve(ctx, "ip", address, r.options.Resolver, r.options.HostMapper, log) if err != nil { - r.options.Logger.Error(err) + log.Error(err) break } @@ -106,13 +108,13 @@ func (r *Router) dial(ctx context.Context, network, address string) (conn net.Co route = r.options.Chain.Route(ctx, network, ipAddr, chain.WithHostRouteOption(address)) } - if r.options.Logger.IsLevelEnabled(logger.DebugLevel) { + if log.IsLevelEnabled(logger.DebugLevel) { buf := bytes.Buffer{} for _, node := range routePath(route) { fmt.Fprintf(&buf, "%s@%s > ", node.Name, node.Addr) } fmt.Fprintf(&buf, "%s", ipAddr) - r.options.Logger.Debugf("route(retry=%d) %s", i, buf.String()) + log.Debugf("route(retry=%d) %s", i, buf.String()) } if route == nil { @@ -122,12 +124,12 @@ func (r *Router) dial(ctx context.Context, network, address string) (conn net.Co chain.InterfaceDialOption(r.options.IfceName), chain.NetnsDialOption(r.options.Netns), chain.SockOptsDialOption(r.options.SockOpts), - chain.LoggerDialOption(r.options.Logger), + chain.LoggerDialOption(log), ) if err == nil { break } - r.options.Logger.Errorf("route(retry=%d) %s", i, err) + log.Errorf("route(retry=%d) %s", i, err) } return @@ -138,7 +140,12 @@ func (r *Router) Bind(ctx context.Context, network, address string, opts ...chai if count <= 0 { count = 1 } - r.options.Logger.Debugf("bind on %s/%s", address, network) + + log := r.options.Logger.WithFields(map[string]any{ + "sid": ctxvalue.SidFromContext(ctx), + }) + + log.Debugf("bind on %s/%s", address, network) for i := 0; i < count; i++ { ctx := ctx @@ -157,13 +164,13 @@ func (r *Router) Bind(ctx context.Context, network, address string, opts ...chai } } - if r.options.Logger.IsLevelEnabled(logger.DebugLevel) { + if log.IsLevelEnabled(logger.DebugLevel) { buf := bytes.Buffer{} for _, node := range routePath(route) { fmt.Fprintf(&buf, "%s@%s > ", node.Name, node.Addr) } fmt.Fprintf(&buf, "%s", address) - r.options.Logger.Debugf("route(retry=%d) %s", i, buf.String()) + log.Debugf("route(retry=%d) %s", i, buf.String()) } if route == nil { @@ -173,7 +180,7 @@ func (r *Router) Bind(ctx context.Context, network, address string, opts ...chai if err == nil { break } - r.options.Logger.Errorf("route(retry=%d) %s", i, err) + log.Errorf("route(retry=%d) %s", i, err) } return diff --git a/config/config.go b/config/config.go index d5f2d637..3cd58dde 100644 --- a/config/config.go +++ b/config/config.go @@ -284,8 +284,9 @@ type TCPRecorder struct { } type HTTPRecorder struct { - URL string `json:"url" yaml:"url"` - Timeout time.Duration `json:"timeout"` + URL string `yaml:"url" json:"url"` + Timeout time.Duration `yaml:",omitempty" json:"timeout,omitempty"` + Header map[string]string `yaml:",omitempty" json:"header,omitempty"` } type RedisRecorder struct { diff --git a/config/parsing/recorder/parse.go b/config/parsing/recorder/parse.go index 60d675f2..d2cba61d 100644 --- a/config/parsing/recorder/parse.go +++ b/config/parsing/recorder/parse.go @@ -2,6 +2,7 @@ package recorder import ( "crypto/tls" + "net/http" "strings" "github.com/go-gost/core/recorder" @@ -51,7 +52,14 @@ func ParseRecorder(cfg *config.RecorderConfig) (r recorder.Recorder) { } if cfg.HTTP != nil && cfg.HTTP.URL != "" { - return xrecorder.HTTPRecorder(cfg.HTTP.URL, xrecorder.TimeoutHTTPRecorderOption(cfg.HTTP.Timeout)) + h := http.Header{} + for k, v := range cfg.HTTP.Header { + h.Add(k, v) + } + return xrecorder.HTTPRecorder(cfg.HTTP.URL, + xrecorder.TimeoutHTTPRecorderOption(cfg.HTTP.Timeout), + xrecorder.HeaderHTTPRecorderOption(h), + ) } if cfg.Redis != nil && diff --git a/config/parsing/service/parse.go b/config/parsing/service/parse.go index 79ad47f3..4436a0bc 100644 --- a/config/parsing/service/parse.go +++ b/config/parsing/service/parse.go @@ -278,6 +278,7 @@ func ParseService(cfg *config.ServiceConfig) (service.Service, error) { handler.RateLimiterOption(registry.RateLimiterRegistry().Get(cfg.RLimiter)), handler.TrafficLimiterOption(registry.TrafficLimiterRegistry().Get(cfg.Handler.Limiter)), handler.ObserverOption(registry.ObserverRegistry().Get(cfg.Handler.Observer)), + handler.RecordersOption(recorders...), handler.LoggerOption(handlerLogger), handler.ServiceOption(cfg.Name), handler.NetnsOption(netnsIn), diff --git a/go.mod b/go.mod index 81f4dd4a..203c8c47 100644 --- a/go.mod +++ b/go.mod @@ -9,7 +9,7 @@ require ( github.com/asaskevich/govalidator v0.0.0-20210307081110-f21760c49a8d github.com/gin-contrib/cors v1.6.0 github.com/gin-gonic/gin v1.9.1 - github.com/go-gost/core v0.1.2 + github.com/go-gost/core v0.1.3 github.com/go-gost/gosocks4 v0.0.1 github.com/go-gost/gosocks5 v0.4.2 github.com/go-gost/plugin v0.1.1 diff --git a/go.sum b/go.sum index 90f1d7a2..4626c733 100644 --- a/go.sum +++ b/go.sum @@ -53,8 +53,8 @@ github.com/gin-contrib/sse v0.1.0 h1:Y/yl/+YNO8GZSjAhjMsSuLt29uWRFHdHYUb5lYOV9qE github.com/gin-contrib/sse v0.1.0/go.mod h1:RHrZQHXnP2xjPF+u1gW/2HnVO7nvIa9PG3Gm+fLHvGI= github.com/gin-gonic/gin v1.9.1 h1:4idEAncQnU5cB7BeOkPtxjfCSye0AAm1R0RVIqJ+Jmg= github.com/gin-gonic/gin v1.9.1/go.mod h1:hPrL7YrpYKXt5YId3A/Tnip5kqbEAP+KLuI3SUcPTeU= -github.com/go-gost/core v0.1.2 h1:uWGLXEcfqkLYwvpGutXN2eIGXLOUGQX7L1QTe3NUDeA= -github.com/go-gost/core v0.1.2/go.mod h1:WGI43jOka7FAsSAwi/fSMaqxdR+E339ycb4NBGlFr6A= +github.com/go-gost/core v0.1.3 h1:Azq50of/OEv89/4THFbwNLidXK1fIwsMRVcgQUlvJug= +github.com/go-gost/core v0.1.3/go.mod h1:WGI43jOka7FAsSAwi/fSMaqxdR+E339ycb4NBGlFr6A= github.com/go-gost/gosocks4 v0.0.1 h1:+k1sec8HlELuQV7rWftIkmy8UijzUt2I6t+iMPlGB2s= github.com/go-gost/gosocks4 v0.0.1/go.mod h1:3B6L47HbU/qugDg4JnoFPHgJXE43Inz8Bah1QaN9qCc= github.com/go-gost/gosocks5 v0.4.2 h1:IianxHTkACPqCwiOAT3MHoMdSUl+SEPSRu1ikawC1Pc= diff --git a/handler/dns/handler.go b/handler/dns/handler.go index 2cd4534d..1880a944 100644 --- a/handler/dns/handler.go +++ b/handler/dns/handler.go @@ -15,8 +15,12 @@ import ( "github.com/go-gost/core/hosts" "github.com/go-gost/core/logger" md "github.com/go-gost/core/metadata" + "github.com/go-gost/core/recorder" + ctxvalue "github.com/go-gost/x/ctx" xhop "github.com/go-gost/x/hop" resolver_util "github.com/go-gost/x/internal/util/resolver" + rate_limiter "github.com/go-gost/x/limiter/rate" + xrecorder "github.com/go-gost/x/recorder" "github.com/go-gost/x/registry" "github.com/go-gost/x/resolver/exchanger" "github.com/miekg/dns" @@ -37,6 +41,7 @@ type dnsHandler struct { hostMapper hosts.HostMapper md metadata options handler.Options + recorder recorder.RecorderObject } func NewHandler(opts ...handler.Option) handler.Handler { @@ -108,6 +113,13 @@ func (h *dnsHandler) Init(md md.Metadata) (err error) { h.exchangers["default"] = ex } + for _, ro := range h.options.Recorders { + if ro.Record == xrecorder.RecorderServiceHandler { + h.recorder = ro + break + } + } + return } @@ -116,24 +128,44 @@ func (h *dnsHandler) Forward(hop hop.Hop) { h.hop = hop } -func (h *dnsHandler) Handle(ctx context.Context, conn net.Conn, opts ...handler.HandleOption) error { +func (h *dnsHandler) Handle(ctx context.Context, conn net.Conn, opts ...handler.HandleOption) (err error) { defer conn.Close() start := time.Now() + + ro := &xrecorder.HandlerRecorderObject{ + Service: h.options.Service, + Network: conn.LocalAddr().Network(), + RemoteAddr: conn.RemoteAddr().String(), + LocalAddr: conn.LocalAddr().String(), + Time: start, + SID: string(ctxvalue.SidFromContext(ctx)), + } + log := h.options.Logger.WithFields(map[string]any{ "remote": conn.RemoteAddr().String(), "local": conn.LocalAddr().String(), + "sid": ctxvalue.SidFromContext(ctx), }) log.Infof("%s <> %s", conn.RemoteAddr(), conn.LocalAddr()) 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.Warnf("recorder: %v", err) + } + log.WithFields(map[string]any{ "duration": time.Since(start), }).Infof("%s >< %s", conn.RemoteAddr(), conn.LocalAddr()) + }() if !h.checkRateLimit(conn.RemoteAddr()) { - return nil + return rate_limiter.ErrRateLimit } b := bufpool.Get(h.md.bufferSize) @@ -145,7 +177,7 @@ func (h *dnsHandler) Handle(ctx context.Context, conn net.Conn, opts ...handler. return err } - reply, err := h.request(ctx, b[:n], log) + reply, err := h.request(ctx, b[:n], ro, log) if err != nil { return err } @@ -170,7 +202,7 @@ func (h *dnsHandler) checkRateLimit(addr net.Addr) bool { return true } -func (h *dnsHandler) request(ctx context.Context, msg []byte, log logger.Logger) ([]byte, error) { +func (h *dnsHandler) request(ctx context.Context, msg []byte, ro *xrecorder.HandlerRecorderObject, log logger.Logger) ([]byte, error) { mq := dns.Msg{} if err := mq.Unpack(msg); err != nil { log.Error(err) @@ -181,6 +213,14 @@ func (h *dnsHandler) request(ctx context.Context, msg []byte, log logger.Logger) return nil, errors.New("msg: empty question") } + ro.DNS = &xrecorder.DNSRecorderObject{ + ID: int(mq.Id), + Name: mq.Question[0].Name, + Class: dns.Class(mq.Question[0].Qclass).String(), + Type: dns.Type(mq.Question[0].Qtype).String(), + Question: mq.String(), + } + resolver_util.AddSubnetOpt(&mq, h.md.clientIP) if log.IsLevelEnabled(logger.TraceLevel) { @@ -188,13 +228,15 @@ func (h *dnsHandler) request(ctx context.Context, msg []byte, log logger.Logger) } var mr *dns.Msg - if log.IsLevelEnabled(logger.TraceLevel) { - defer func() { - if mr != nil { + defer func() { + if mr != nil { + if log.IsLevelEnabled(logger.TraceLevel) { log.Trace(mr.String()) } - }() - } + + ro.DNS.Answer = mr.String() + } + }() if h.options.Bypass != nil && mq.Question[0].Qclass == dns.ClassINET { if h.options.Bypass.Contains(context.Background(), "udp", strings.Trim(mq.Question[0].Name, ".")) { @@ -214,10 +256,12 @@ func (h *dnsHandler) request(ctx context.Context, msg []byte, log logger.Logger) // only cache for single question message. if len(mq.Question) == 1 { var ttl time.Duration - mr, ttl = h.cache.Load(resolver_util.NewCacheKey(&mq.Question[0])) + mr, ttl = h.cache.Load(ctx, resolver_util.NewCacheKey(&mq.Question[0])) if mr != nil { mr.Id = mq.Id if int32(ttl.Seconds()) > 0 { + ro.DNS.Cached = true + log.Debugf("message %d (cached): %s", mq.Id, mq.Question[0].String()) b := bufpool.Get(h.md.bufferSize) return mr.PackBuffer(b) @@ -225,6 +269,12 @@ func (h *dnsHandler) request(ctx context.Context, msg []byte, log logger.Logger) } } + ex := h.selectExchanger(ctx, strings.Trim(mq.Question[0].Name, ".")) + if ex == nil { + return nil, fmt.Errorf("exchange not found for %s", mq.Question[0].Name) + } + ro.Host = ex.String() + if mr != nil && h.md.async { b := bufpool.Get(h.md.bufferSize) reply, err := mr.PackBuffer(b) @@ -234,15 +284,21 @@ func (h *dnsHandler) request(ctx context.Context, msg []byte, log logger.Logger) h.cache.RefreshTTL(resolver_util.NewCacheKey(&mq.Question[0])) log.Debugf("exchange message %d (async): %s", mq.Id, mq.Question[0].String()) - go h.exchange(ctx, &mq) + go h.exchange(ctx, ex, &mq) return reply, nil } log.Debugf("exchange message %d: %s", mq.Id, mq.Question[0].String()) - return h.exchange(ctx, &mq) + mr, err := h.exchange(ctx, ex, &mq) + if err != nil { + return nil, err + } + + b := bufpool.Get(h.md.bufferSize) + return mr.PackBuffer(b) } -func (h *dnsHandler) exchange(ctx context.Context, mq *dns.Msg) ([]byte, error) { +func (h *dnsHandler) exchange(ctx context.Context, ex exchanger.Exchanger, mq *dns.Msg) (*dns.Msg, error) { b := bufpool.Get(h.md.bufferSize) defer bufpool.Put(b) @@ -251,12 +307,6 @@ func (h *dnsHandler) exchange(ctx context.Context, mq *dns.Msg) ([]byte, error) return nil, err } - ex := h.selectExchanger(ctx, strings.Trim(mq.Question[0].Name, ".")) - if ex == nil { - err = fmt.Errorf("exchange not found for %s", mq.Question[0].Name) - return nil, err - } - reply, err := ex.Exchange(ctx, query) if err != nil { return nil, err @@ -268,10 +318,10 @@ func (h *dnsHandler) exchange(ctx context.Context, mq *dns.Msg) ([]byte, error) } if len(mq.Question) == 1 { key := resolver_util.NewCacheKey(&mq.Question[0]) - h.cache.Store(key, mr, h.md.ttl) + h.cache.Store(ctx, key, mr, h.md.ttl) } - return reply, nil + return mr, nil } // lookup host mapper diff --git a/handler/forward/local/handler.go b/handler/forward/local/handler.go index eedbeb7c..3b78a09f 100644 --- a/handler/forward/local/handler.go +++ b/handler/forward/local/handler.go @@ -21,12 +21,16 @@ import ( "github.com/go-gost/core/hop" "github.com/go-gost/core/logger" md "github.com/go-gost/core/metadata" + "github.com/go-gost/core/recorder" + 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" "github.com/go-gost/x/internal/util/forward" tls_util "github.com/go-gost/x/internal/util/tls" + rate_limiter "github.com/go-gost/x/limiter/rate" + xrecorder "github.com/go-gost/x/recorder" "github.com/go-gost/x/registry" ) @@ -37,9 +41,10 @@ func init() { } type forwardHandler struct { - hop hop.Hop - md metadata - options handler.Options + hop hop.Hop + md metadata + options handler.Options + recorder recorder.RecorderObject } func NewHandler(opts ...handler.Option) handler.Handler { @@ -58,6 +63,13 @@ func (h *forwardHandler) Init(md md.Metadata) (err error) { return } + for _, ro := range h.options.Recorders { + if ro.Record == xrecorder.RecorderServiceHandler { + h.recorder = ro + break + } + } + return } @@ -66,24 +78,42 @@ func (h *forwardHandler) Forward(hop hop.Hop) { h.hop = hop } -func (h *forwardHandler) Handle(ctx context.Context, conn net.Conn, opts ...handler.HandleOption) error { +func (h *forwardHandler) Handle(ctx context.Context, conn net.Conn, opts ...handler.HandleOption) (err error) { defer conn.Close() start := time.Now() + + ro := &xrecorder.HandlerRecorderObject{ + Service: h.options.Service, + RemoteAddr: conn.RemoteAddr().String(), + LocalAddr: conn.LocalAddr().String(), + Time: start, + SID: string(ctxvalue.SidFromContext(ctx)), + } + log := h.options.Logger.WithFields(map[string]any{ "remote": conn.RemoteAddr().String(), "local": conn.LocalAddr().String(), + "sid": ctxvalue.SidFromContext(ctx), }) log.Infof("%s <> %s", conn.RemoteAddr(), conn.LocalAddr()) defer func() { + if !ro.Time.IsZero() { + if err != nil { + ro.Err = err.Error() + } + ro.Duration = time.Since(start) + ro.Record(ctx, h.recorder.Recorder) + } + log.WithFields(map[string]any{ "duration": time.Since(start), }).Infof("%s >< %s", conn.RemoteAddr(), conn.LocalAddr()) }() if !h.checkRateLimit(conn.RemoteAddr()) { - return nil + return rate_limiter.ErrRateLimit } network := "tcp" @@ -106,7 +136,11 @@ func (h *forwardHandler) Handle(ctx context.Context, conn net.Conn, opts ...hand } if protocol == forward.ProtoHTTP { - h.handleHTTP(ctx, xio.NewReadWriteCloser(rw, rw, conn), conn.RemoteAddr(), log) + ro2 := &xrecorder.HandlerRecorderObject{} + *ro2 = *ro + ro.Time = time.Time{} + + h.handleHTTP(ctx, xio.NewReadWriteCloser(rw, rw, conn), conn.RemoteAddr(), ro2, log) return nil } @@ -144,6 +178,9 @@ func (h *forwardHandler) Handle(ctx context.Context, conn net.Conn, opts ...hand } } + ro.Network = network + ro.Host = addr + log = log.WithFields(map[string]any{ "host": host, "node": target.Name, @@ -177,7 +214,7 @@ func (h *forwardHandler) Handle(ctx context.Context, conn net.Conn, opts ...hand return nil } -func (h *forwardHandler) handleHTTP(ctx context.Context, rw io.ReadWriteCloser, remoteAddr net.Addr, log logger.Logger) (err error) { +func (h *forwardHandler) handleHTTP(ctx context.Context, rw io.ReadWriteCloser, remoteAddr net.Addr, ro *xrecorder.HandlerRecorderObject, log logger.Logger) (err error) { br := bufio.NewReader(rw) for { @@ -202,6 +239,27 @@ func (h *forwardHandler) handleHTTP(ctx context.Context, rw io.ReadWriteCloser, log.Trace(string(dump)) } + start := time.Now() + ro.Time = start + ro.HTTP = &xrecorder.HTTPRecorderObject{ + Host: req.Host, + Proto: req.Proto, + Scheme: req.URL.Scheme, + Method: req.Method, + URI: req.RequestURI, + RequestHeader: req.Header.Clone(), + } + defer func() { + if err != nil { + ro.HTTP.StatusCode = resp.StatusCode + ro.HTTP.ResponseHeader = resp.Header + + ro.Duration = time.Since(start) + ro.Err = err.Error() + ro.Record(ctx, h.recorder.Recorder) + } + }() + host := req.Host if _, _, err := net.SplitHostPort(host); err != nil { host = net.JoinHostPort(strings.Trim(host, "[]"), "80") @@ -209,7 +267,8 @@ func (h *forwardHandler) handleHTTP(ctx context.Context, rw io.ReadWriteCloser, if bp := h.options.Bypass; bp != nil && bp.Contains(ctx, "tcp", host, bypass.WithPathOption(req.RequestURI)) { log.Debugf("bypass: %s %s", host, req.RequestURI) resp.StatusCode = http.StatusForbidden - return resp.Write(rw) + resp.Write(rw) + return xbypass.ErrBypass } if addr := getRealClientAddr(req, remoteAddr); addr != remoteAddr { @@ -236,6 +295,8 @@ func (h *forwardHandler) handleHTTP(ctx context.Context, rw io.ReadWriteCloser, return resp.Write(rw) } + ro.Host = target.Addr + log = log.WithFields(map[string]any{ "host": req.Host, "node": target.Name, @@ -305,14 +366,15 @@ func (h *forwardHandler) handleHTTP(ctx context.Context, rw io.ReadWriteCloser, cc = tls.Client(cc, cfg) } - if err := req.Write(cc); err != nil { + if err = req.Write(cc); err != nil { cc.Close() log.Warnf("send request to node %s(%s): %v", target.Name, target.Addr, err) - return resp.Write(rw) + resp.Write(rw) + return err } if req.Header.Get("Upgrade") == "websocket" { - err := xnet.Transport(cc, xio.NewReadWriter(br, rw)) + err = xnet.Transport(cc, xio.NewReadWriter(br, rw)) if err == nil { err = io.EOF } @@ -322,12 +384,28 @@ func (h *forwardHandler) handleHTTP(ctx context.Context, rw io.ReadWriteCloser, go func() { defer cc.Close() - res, err := http.ReadResponse(bufio.NewReader(cc), req) + var err error + var res *http.Response + + defer func() { + ro.Duration = time.Since(start) + if err != nil { + ro.Err = err.Error() + } + if res != nil && ro.HTTP != nil { + ro.HTTP.ResponseHeader = res.Header + ro.HTTP.StatusCode = res.StatusCode + } + ro.Record(ctx, h.recorder.Recorder) + }() + + res, err = http.ReadResponse(bufio.NewReader(cc), req) if err != nil { log.Warnf("read response from node %s(%s): %v", target.Name, target.Addr, err) resp.Write(rw) return } + defer res.Body.Close() if log.IsLevelEnabled(logger.TraceLevel) { dump, _ := httputil.DumpResponse(res, false) @@ -338,7 +416,7 @@ func (h *forwardHandler) handleHTTP(ctx context.Context, rw io.ReadWriteCloser, defer rw.Close() } - if err := h.rewriteBody(res, bodyRewrites...); err != nil { + if err = h.rewriteBody(res, bodyRewrites...); err != nil { rw.Close() log.Errorf("rewrite body: %v", err) return diff --git a/handler/forward/remote/handler.go b/handler/forward/remote/handler.go index 0231df01..345adac4 100644 --- a/handler/forward/remote/handler.go +++ b/handler/forward/remote/handler.go @@ -22,6 +22,8 @@ import ( "github.com/go-gost/core/logger" mdata "github.com/go-gost/core/metadata" mdutil "github.com/go-gost/core/metadata/util" + "github.com/go-gost/core/recorder" + 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" @@ -29,6 +31,8 @@ import ( "github.com/go-gost/x/internal/net/proxyproto" "github.com/go-gost/x/internal/util/forward" tls_util "github.com/go-gost/x/internal/util/tls" + rate_limiter "github.com/go-gost/x/limiter/rate" + xrecorder "github.com/go-gost/x/recorder" "github.com/go-gost/x/registry" ) @@ -38,9 +42,10 @@ func init() { } type forwardHandler struct { - hop hop.Hop - md metadata - options handler.Options + hop hop.Hop + md metadata + options handler.Options + recorder recorder.RecorderObject } func NewHandler(opts ...handler.Option) handler.Handler { @@ -59,6 +64,13 @@ func (h *forwardHandler) Init(md mdata.Metadata) (err error) { return } + for _, ro := range h.options.Recorders { + if ro.Record == xrecorder.RecorderServiceHandler { + h.recorder = ro + break + } + } + return } @@ -67,24 +79,42 @@ func (h *forwardHandler) Forward(hop hop.Hop) { h.hop = hop } -func (h *forwardHandler) Handle(ctx context.Context, conn net.Conn, opts ...handler.HandleOption) error { +func (h *forwardHandler) Handle(ctx context.Context, conn net.Conn, opts ...handler.HandleOption) (err error) { defer conn.Close() start := time.Now() + + ro := &xrecorder.HandlerRecorderObject{ + Service: h.options.Service, + RemoteAddr: conn.RemoteAddr().String(), + LocalAddr: conn.LocalAddr().String(), + Time: start, + SID: string(ctxvalue.SidFromContext(ctx)), + } + log := h.options.Logger.WithFields(map[string]any{ "remote": conn.RemoteAddr().String(), "local": conn.LocalAddr().String(), + "sid": ctxvalue.SidFromContext(ctx), }) log.Infof("%s <> %s", conn.RemoteAddr(), conn.LocalAddr()) defer func() { + if !ro.Time.IsZero() { + if err != nil { + ro.Err = err.Error() + } + ro.Duration = time.Since(start) + ro.Record(ctx, h.recorder.Recorder) + } + log.WithFields(map[string]any{ "duration": time.Since(start), }).Infof("%s >< %s", conn.RemoteAddr(), conn.LocalAddr()) }() if !h.checkRateLimit(conn.RemoteAddr()) { - return nil + return rate_limiter.ErrRateLimit } network := "tcp" @@ -108,7 +138,11 @@ func (h *forwardHandler) Handle(ctx context.Context, conn net.Conn, opts ...hand } } if protocol == forward.ProtoHTTP { - h.handleHTTP(ctx, xio.NewReadWriteCloser(rw, rw, conn), conn.RemoteAddr(), localAddr, log) + ro2 := &xrecorder.HandlerRecorderObject{} + *ro2 = *ro + ro.Time = time.Time{} + + h.handleHTTP(ctx, xio.NewReadWriteCloser(rw, rw, conn), conn.RemoteAddr(), localAddr, ro2, log) return nil } @@ -143,6 +177,9 @@ func (h *forwardHandler) Handle(ctx context.Context, conn net.Conn, opts ...hand } } + ro.Network = network + ro.Host = target.Addr + log = log.WithFields(map[string]any{ "host": host, "node": target.Name, @@ -178,7 +215,7 @@ func (h *forwardHandler) Handle(ctx context.Context, conn net.Conn, opts ...hand return nil } -func (h *forwardHandler) handleHTTP(ctx context.Context, rw io.ReadWriteCloser, remoteAddr net.Addr, localAddr net.Addr, log logger.Logger) (err error) { +func (h *forwardHandler) handleHTTP(ctx context.Context, rw io.ReadWriteCloser, remoteAddr net.Addr, localAddr net.Addr, ro *xrecorder.HandlerRecorderObject, log logger.Logger) (err error) { br := bufio.NewReader(rw) for { @@ -202,6 +239,27 @@ func (h *forwardHandler) handleHTTP(ctx context.Context, rw io.ReadWriteCloser, log.Trace(string(dump)) } + start := time.Now() + ro.Time = start + ro.HTTP = &xrecorder.HTTPRecorderObject{ + Host: req.Host, + Proto: req.Proto, + Scheme: req.URL.Scheme, + Method: req.Method, + URI: req.RequestURI, + RequestHeader: req.Header.Clone(), + } + defer func() { + if err != nil { + ro.HTTP.StatusCode = resp.StatusCode + ro.HTTP.ResponseHeader = resp.Header + + ro.Duration = time.Since(start) + ro.Err = err.Error() + ro.Record(ctx, h.recorder.Recorder) + } + }() + host := req.Host if _, _, err := net.SplitHostPort(host); err != nil { host = net.JoinHostPort(strings.Trim(host, "[]"), "80") @@ -209,7 +267,8 @@ func (h *forwardHandler) handleHTTP(ctx context.Context, rw io.ReadWriteCloser, if bp := h.options.Bypass; bp != nil && bp.Contains(ctx, "tcp", host, bypass.WithPathOption(req.RequestURI)) { log.Debugf("bypass: %s %s", host, req.RequestURI) resp.StatusCode = http.StatusForbidden - return resp.Write(rw) + resp.Write(rw) + return xbypass.ErrBypass } if addr := getRealClientAddr(req, remoteAddr); addr != remoteAddr { @@ -236,6 +295,8 @@ func (h *forwardHandler) handleHTTP(ctx context.Context, rw io.ReadWriteCloser, return resp.Write(rw) } + ro.Host = target.Addr + log = log.WithFields(map[string]any{ "host": req.Host, "node": target.Name, @@ -306,10 +367,11 @@ func (h *forwardHandler) handleHTTP(ctx context.Context, rw io.ReadWriteCloser, cc = proxyproto.WrapClientConn(h.md.proxyProtocol, remoteAddr, localAddr, cc) - if err := req.Write(cc); err != nil { + if err = req.Write(cc); err != nil { cc.Close() log.Warnf("send request to node %s(%s): %v", target.Name, target.Addr, err) - return resp.Write(rw) + resp.Write(rw) + return err } if req.Header.Get("Upgrade") == "websocket" { @@ -323,12 +385,28 @@ func (h *forwardHandler) handleHTTP(ctx context.Context, rw io.ReadWriteCloser, go func() { defer cc.Close() - res, err := http.ReadResponse(bufio.NewReader(cc), req) + var err error + var res *http.Response + + defer func() { + ro.Duration = time.Since(start) + if err != nil { + ro.Err = err.Error() + } + if res != nil && ro.HTTP != nil { + ro.HTTP.ResponseHeader = res.Header + ro.HTTP.StatusCode = res.StatusCode + } + ro.Record(ctx, h.recorder.Recorder) + }() + + res, err = http.ReadResponse(bufio.NewReader(cc), req) if err != nil { log.Warnf("read response from node %s(%s): %v", target.Name, target.Addr, err) resp.Write(rw) return } + defer res.Body.Close() if log.IsLevelEnabled(logger.TraceLevel) { dump, _ := httputil.DumpResponse(res, false) @@ -339,7 +417,7 @@ func (h *forwardHandler) handleHTTP(ctx context.Context, rw io.ReadWriteCloser, defer rw.Close() } - if err := h.rewriteBody(res, bodyRewrites...); err != nil { + if err = h.rewriteBody(res, bodyRewrites...); err != nil { rw.Close() log.Errorf("rewrite body: %v", err) return diff --git a/handler/http/handler.go b/handler/http/handler.go index 4f19849a..08e034f9 100644 --- a/handler/http/handler.go +++ b/handler/http/handler.go @@ -24,13 +24,17 @@ import ( "github.com/go-gost/core/logger" md "github.com/go-gost/core/metadata" "github.com/go-gost/core/observer/stats" + "github.com/go-gost/core/recorder" + xbypass "github.com/go-gost/x/bypass" ctxvalue "github.com/go-gost/x/ctx" xio "github.com/go-gost/x/internal/io" netpkg "github.com/go-gost/x/internal/net" limiter_util "github.com/go-gost/x/internal/util/limiter" stats_util "github.com/go-gost/x/internal/util/stats" + rate_limiter "github.com/go-gost/x/limiter/rate" traffic_wrapper "github.com/go-gost/x/limiter/traffic/wrapper" stats_wrapper "github.com/go-gost/x/observer/stats/wrapper" + xrecorder "github.com/go-gost/x/recorder" "github.com/go-gost/x/registry" ) @@ -39,11 +43,12 @@ func init() { } type httpHandler struct { - md metadata - options handler.Options - stats *stats_util.HandlerStats - limiter traffic.TrafficLimiter - cancel context.CancelFunc + md metadata + options handler.Options + stats *stats_util.HandlerStats + limiter traffic.TrafficLimiter + cancel context.CancelFunc + recorder recorder.RecorderObject } func NewHandler(opts ...handler.Option) handler.Handler { @@ -74,15 +79,29 @@ func (h *httpHandler) Init(md md.Metadata) error { h.limiter = limiter_util.NewCachedTrafficLimiter(limiter, 30*time.Second, 60*time.Second) } + for _, ro := range h.options.Recorders { + if ro.Record == xrecorder.RecorderServiceHandler { + h.recorder = ro + break + } + } + return nil } -func (h *httpHandler) Handle(ctx context.Context, conn net.Conn, opts ...handler.HandleOption) error { +func (h *httpHandler) Handle(ctx context.Context, conn net.Conn, opts ...handler.HandleOption) (err error) { defer conn.Close() - // ctx = sx.ContextWithHash(ctx, &sx.Hash{}) - start := time.Now() + + ro := &xrecorder.HandlerRecorderObject{ + Service: h.options.Service, + RemoteAddr: conn.RemoteAddr().String(), + LocalAddr: conn.LocalAddr().String(), + Time: start, + SID: string(ctxvalue.SidFromContext(ctx)), + } + log := h.options.Logger.WithFields(map[string]any{ "remote": conn.RemoteAddr().String(), "local": conn.LocalAddr().String(), @@ -90,13 +109,21 @@ func (h *httpHandler) Handle(ctx context.Context, conn net.Conn, opts ...handler }) log.Infof("%s <> %s", conn.RemoteAddr(), conn.LocalAddr()) defer func() { + if !ro.Time.IsZero() { + if err != nil { + ro.Err = err.Error() + } + ro.Duration = time.Since(start) + ro.Record(ctx, h.recorder.Recorder) + } + log.WithFields(map[string]any{ "duration": time.Since(start), }).Infof("%s >< %s", conn.RemoteAddr(), conn.LocalAddr()) }() if !h.checkRateLimit(conn.RemoteAddr()) { - return nil + return rate_limiter.ErrRateLimit } req, err := http.ReadRequest(bufio.NewReader(conn)) @@ -106,7 +133,7 @@ func (h *httpHandler) Handle(ctx context.Context, conn net.Conn, opts ...handler } defer req.Body.Close() - return h.handleRequest(ctx, conn, req, log) + return h.handleRequest(ctx, conn, req, ro, log) } func (h *httpHandler) Close() error { @@ -116,7 +143,7 @@ func (h *httpHandler) Close() error { return nil } -func (h *httpHandler) handleRequest(ctx context.Context, conn net.Conn, req *http.Request, log logger.Logger) error { +func (h *httpHandler) handleRequest(ctx context.Context, conn net.Conn, req *http.Request, ro *xrecorder.HandlerRecorderObject, log logger.Logger) error { if !req.URL.IsAbs() && govalidator.IsDNSName(req.Host) { req.URL.Scheme = "http" } @@ -125,6 +152,7 @@ func (h *httpHandler) handleRequest(ctx context.Context, conn net.Conn, req *htt if network != "udp" { network = "tcp" } + ro.Network = network // Try to get the actual host. // Compatible with GOST 2.x. @@ -133,15 +161,13 @@ func (h *httpHandler) handleRequest(ctx context.Context, conn net.Conn, req *htt req.Host = h } } - req.Header.Del("Gost-Target") - if v := req.Header.Get("X-Gost-Target"); v != "" { if h, err := h.decodeServerName(v); err == nil { req.Host = h } } - req.Header.Del("X-Gost-Target") + ro.Host = req.Host addr := req.Host if _, port, _ := net.SplitHostPort(addr); port == "" { addr = net.JoinHostPort(strings.Trim(addr, "[]"), "80") @@ -150,8 +176,10 @@ func (h *httpHandler) handleRequest(ctx context.Context, conn net.Conn, req *htt fields := map[string]any{ "dst": addr, } + if u, _, _ := h.basicProxyAuth(req.Header.Get("Proxy-Authorization")); u != "" { fields["user"] = u + ro.Client = u } log = log.WithFields(fields) @@ -170,14 +198,26 @@ func (h *httpHandler) handleRequest(ctx context.Context, conn net.Conn, req *htt if resp.Header == nil { resp.Header = http.Header{} } - if resp.Header.Get("Proxy-Agent") == "" { resp.Header.Set("Proxy-Agent", h.md.proxyAgent) } + ro.HTTP = &xrecorder.HTTPRecorderObject{ + Host: req.Host, + Proto: req.Proto, + Scheme: req.URL.Scheme, + Method: req.Method, + URI: req.RequestURI, + RequestHeader: req.Header.Clone(), + } + defer func() { + ro.HTTP.StatusCode = resp.StatusCode + ro.HTTP.ResponseHeader = resp.Header + }() + clientID, ok := h.authenticate(ctx, conn, req, resp, log) if !ok { - return nil + return errors.New("authenication failed") } ctx = ctxvalue.ContextWithClientID(ctx, ctxvalue.ClientID(clientID)) @@ -189,8 +229,8 @@ func (h *httpHandler) handleRequest(ctx context.Context, conn net.Conn, req *htt log.Trace(string(dump)) } log.Debug("bypass: ", addr) - - return resp.Write(conn) + resp.Write(conn) + return xbypass.ErrBypass } if network == "udp" { @@ -209,8 +249,6 @@ func (h *httpHandler) handleRequest(ctx context.Context, conn net.Conn, req *htt return resp.Write(conn) } - req.Header.Del("Proxy-Authorization") - switch h.md.hash { case "host": ctx = ctxvalue.ContextWithHash(ctx, &ctxvalue.Hash{Source: addr}) @@ -249,7 +287,11 @@ func (h *httpHandler) handleRequest(ctx context.Context, conn net.Conn, req *htt } if req.Method != http.MethodConnect { - return h.handleProxy(xio.NewReadWriteCloser(rw, rw, conn), cc, req, log) + ro2 := &xrecorder.HandlerRecorderObject{} + *ro2 = *ro + ro.Time = time.Time{} + + return h.handleProxy(ctx, xio.NewReadWriteCloser(rw, rw, conn), cc, req, ro2, log) } resp.StatusCode = http.StatusOK @@ -274,7 +316,7 @@ func (h *httpHandler) handleRequest(ctx context.Context, conn net.Conn, req *htt return nil } -func (h *httpHandler) handleProxy(rw io.ReadWriteCloser, cc io.ReadWriter, req *http.Request, log logger.Logger) (err error) { +func (h *httpHandler) handleProxy(ctx context.Context, rw io.ReadWriteCloser, cc io.ReadWriter, req *http.Request, ro *xrecorder.HandlerRecorderObject, log logger.Logger) (err error) { roundTrip := func(req *http.Request) error { if req == nil { return nil @@ -287,6 +329,21 @@ func (h *httpHandler) handleProxy(rw io.ReadWriteCloser, cc io.ReadWriter, req * StatusCode: http.StatusServiceUnavailable, } + start := time.Now() + ro.Time = start + + if ro.HTTP != nil { + ro.HTTP = &xrecorder.HTTPRecorderObject{ + Host: req.Host, + Proto: req.Proto, + Scheme: req.URL.Scheme, + Method: req.Method, + URI: req.RequestURI, + RequestHeader: req.Header.Clone(), + StatusCode: resp.StatusCode, + } + } + // HTTP/1.0 if req.ProtoMajor == 1 && req.ProtoMinor == 0 { if strings.ToLower(req.Header.Get("Connection")) == "keep-alive" { @@ -296,20 +353,44 @@ func (h *httpHandler) handleProxy(rw io.ReadWriteCloser, cc io.ReadWriter, req * } } + req.Header.Del("Proxy-Authorization") req.Header.Del("Proxy-Connection") + req.Header.Del("Gost-Target") + req.Header.Del("X-Gost-Target") if err = req.Write(cc); err != nil { resp.Write(rw) + + ro.Duration = time.Since(start) + ro.Err = err.Error() + ro.Record(ctx, h.recorder.Recorder) + return err } go func() { - res, err := http.ReadResponse(bufio.NewReader(cc), req) + var err error + var res *http.Response + + defer func() { + ro.Duration = time.Since(start) + if err != nil { + ro.Err = err.Error() + } + if res != nil && ro.HTTP != nil { + ro.HTTP.ResponseHeader = res.Header + ro.HTTP.StatusCode = res.StatusCode + } + ro.Record(ctx, h.recorder.Recorder) + }() + + res, err = http.ReadResponse(bufio.NewReader(cc), req) if err != nil { h.options.Logger.Errorf("read response: %v", err) resp.Write(rw) return } + defer res.Body.Close() if log.IsLevelEnabled(logger.TraceLevel) { dump, _ := httputil.DumpResponse(res, false) diff --git a/handler/http2/handler.go b/handler/http2/handler.go index 55500378..35b67195 100644 --- a/handler/http2/handler.go +++ b/handler/http2/handler.go @@ -24,13 +24,17 @@ import ( "github.com/go-gost/core/logger" md "github.com/go-gost/core/metadata" "github.com/go-gost/core/observer/stats" + "github.com/go-gost/core/recorder" + xbypass "github.com/go-gost/x/bypass" ctxvalue "github.com/go-gost/x/ctx" xio "github.com/go-gost/x/internal/io" netpkg "github.com/go-gost/x/internal/net" limiter_util "github.com/go-gost/x/internal/util/limiter" stats_util "github.com/go-gost/x/internal/util/stats" + rate_limiter "github.com/go-gost/x/limiter/rate" traffic_wrapper "github.com/go-gost/x/limiter/traffic/wrapper" stats_wrapper "github.com/go-gost/x/observer/stats/wrapper" + xrecorder "github.com/go-gost/x/recorder" "github.com/go-gost/x/registry" ) @@ -39,11 +43,12 @@ func init() { } type http2Handler struct { - md metadata - options handler.Options - stats *stats_util.HandlerStats - limiter traffic.TrafficLimiter - cancel context.CancelFunc + md metadata + options handler.Options + stats *stats_util.HandlerStats + limiter traffic.TrafficLimiter + cancel context.CancelFunc + recorder recorder.RecorderObject } func NewHandler(opts ...handler.Option) handler.Handler { @@ -74,31 +79,55 @@ func (h *http2Handler) Init(md md.Metadata) error { h.limiter = limiter_util.NewCachedTrafficLimiter(limiter, 30*time.Second, 60*time.Second) } + for _, ro := range h.options.Recorders { + if ro.Record == xrecorder.RecorderServiceHandler { + h.recorder = ro + break + } + } + return nil } -func (h *http2Handler) Handle(ctx context.Context, conn net.Conn, opts ...handler.HandleOption) error { +func (h *http2Handler) Handle(ctx context.Context, conn net.Conn, opts ...handler.HandleOption) (err error) { defer conn.Close() start := time.Now() + + ro := &xrecorder.HandlerRecorderObject{ + Service: h.options.Service, + RemoteAddr: conn.RemoteAddr().String(), + LocalAddr: conn.LocalAddr().String(), + Network: "tcp", + Time: start, + SID: string(ctxvalue.SidFromContext(ctx)), + } + log := h.options.Logger.WithFields(map[string]any{ "remote": conn.RemoteAddr().String(), "local": conn.LocalAddr().String(), + "sid": ctxvalue.SidFromContext(ctx), }) log.Infof("%s <> %s", conn.RemoteAddr(), conn.LocalAddr()) defer func() { + if err != nil { + ro.Err = err.Error() + } + ro.Duration = time.Since(start) + ro.Record(ctx, h.recorder.Recorder) + log.WithFields(map[string]any{ "duration": time.Since(start), }).Infof("%s >< %s", conn.RemoteAddr(), conn.LocalAddr()) }() if !h.checkRateLimit(conn.RemoteAddr()) { - return nil + return rate_limiter.ErrRateLimit } v, ok := conn.(md.Metadatable) if !ok || v == nil { - err := errors.New("wrong connection type") + err = errors.New("wrong connection type") log.Error(err) return err } @@ -107,6 +136,7 @@ func (h *http2Handler) Handle(ctx context.Context, conn net.Conn, opts ...handle return h.roundTrip(ctx, md.Get("w").(http.ResponseWriter), md.Get("r").(*http.Request), + ro, log, ) } @@ -121,7 +151,7 @@ func (h *http2Handler) Close() error { // NOTE: there is an issue (golang/go#43989) will cause the client hangs // when server returns an non-200 status code, // May be fixed in go1.18. -func (h *http2Handler) roundTrip(ctx context.Context, w http.ResponseWriter, req *http.Request, log logger.Logger) error { +func (h *http2Handler) roundTrip(ctx context.Context, w http.ResponseWriter, req *http.Request, ro *xrecorder.HandlerRecorderObject, log logger.Logger) error { // Try to get the actual host. // Compatible with GOST 2.x. if v := req.Header.Get("Gost-Target"); v != "" { @@ -129,15 +159,13 @@ func (h *http2Handler) roundTrip(ctx context.Context, w http.ResponseWriter, req req.Host = h } } - req.Header.Del("Gost-Target") - if v := req.Header.Get("X-Gost-Target"); v != "" { if h, err := h.decodeServerName(v); err == nil { req.Host = h } } - req.Header.Del("X-Gost-Target") + ro.Host = req.Host addr := req.Host if _, port, _ := net.SplitHostPort(addr); port == "" { addr = net.JoinHostPort(strings.Trim(addr, "[]"), "80") @@ -148,6 +176,7 @@ func (h *http2Handler) roundTrip(ctx context.Context, w http.ResponseWriter, req } if u, _, _ := h.basicProxyAuth(req.Header.Get("Proxy-Authorization")); u != "" { fields["user"] = u + ro.Client = u } log = log.WithFields(fields) @@ -164,25 +193,41 @@ func (h *http2Handler) roundTrip(ctx context.Context, w http.ResponseWriter, req resp := &http.Response{ ProtoMajor: 2, ProtoMinor: 0, - Header: http.Header{}, + Header: w.Header(), Body: io.NopCloser(bytes.NewReader([]byte{})), } + ro.HTTP = &xrecorder.HTTPRecorderObject{ + Host: req.Host, + Proto: req.Proto, + Scheme: req.URL.Scheme, + Method: req.Method, + URI: req.RequestURI, + RequestHeader: req.Header.Clone(), + } + defer func() { + ro.HTTP.StatusCode = resp.StatusCode + ro.HTTP.ResponseHeader = resp.Header + }() + clientID, ok := h.authenticate(ctx, w, req, resp, log) if !ok { - return nil + return errors.New("authenication failed") } ctx = ctxvalue.ContextWithClientID(ctx, ctxvalue.ClientID(clientID)) if h.options.Bypass != nil && h.options.Bypass.Contains(ctx, "tcp", addr) { - w.WriteHeader(http.StatusForbidden) + resp.StatusCode = http.StatusForbidden + w.WriteHeader(resp.StatusCode) log.Debug("bypass: ", addr) - return nil + return xbypass.ErrBypass } // delete the proxy related headers. req.Header.Del("Proxy-Authorization") req.Header.Del("Proxy-Connection") + req.Header.Del("Gost-Target") + req.Header.Del("X-Gost-Target") switch h.md.hash { case "host": @@ -192,12 +237,14 @@ func (h *http2Handler) roundTrip(ctx context.Context, w http.ResponseWriter, req cc, err := h.options.Router.Dial(ctx, "tcp", addr) if err != nil { log.Error(err) - w.WriteHeader(http.StatusServiceUnavailable) + resp.StatusCode = http.StatusServiceUnavailable + w.WriteHeader(resp.StatusCode) return err } defer cc.Close() if req.Method == http.MethodConnect { + resp.StatusCode = http.StatusOK w.WriteHeader(http.StatusOK) if fw, ok := w.(http.Flusher); ok { fw.Flush() @@ -209,6 +256,7 @@ func (h *http2Handler) roundTrip(ctx context.Context, w http.ResponseWriter, req conn, _, err := hj.Hijack() if err != nil { log.Error(err) + resp.StatusCode = http.StatusInternalServerError w.WriteHeader(http.StatusInternalServerError) return err } @@ -253,6 +301,8 @@ func (h *http2Handler) roundTrip(ctx context.Context, w http.ResponseWriter, req } // TODO: forward request + resp.StatusCode = http.StatusBadRequest + w.WriteHeader(resp.StatusCode) return nil } diff --git a/handler/http3/handler.go b/handler/http3/handler.go index 8be8c933..01c50231 100644 --- a/handler/http3/handler.go +++ b/handler/http3/handler.go @@ -60,6 +60,7 @@ func (h *http3Handler) Handle(ctx context.Context, conn net.Conn, opts ...handle log := h.options.Logger.WithFields(map[string]any{ "remote": conn.RemoteAddr().String(), "local": conn.LocalAddr().String(), + "sid": ctxvalue.SidFromContext(ctx), }) log.Infof("%s <> %s", conn.RemoteAddr(), conn.LocalAddr()) defer func() { diff --git a/handler/redirect/tcp/handler.go b/handler/redirect/tcp/handler.go index e8e548b5..170087ad 100644 --- a/handler/redirect/tcp/handler.go +++ b/handler/redirect/tcp/handler.go @@ -18,9 +18,14 @@ import ( "github.com/go-gost/core/handler" "github.com/go-gost/core/logger" md "github.com/go-gost/core/metadata" + "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" netpkg "github.com/go-gost/x/internal/net" + rate_limiter "github.com/go-gost/x/limiter/rate" + xrecorder "github.com/go-gost/x/recorder" "github.com/go-gost/x/registry" ) @@ -31,8 +36,9 @@ func init() { } type redirectHandler struct { - md metadata - options handler.Options + md metadata + options handler.Options + recorder recorder.RecorderObject } func NewHandler(opts ...handler.Option) handler.Handler { @@ -51,6 +57,13 @@ func (h *redirectHandler) Init(md md.Metadata) (err error) { return } + for _, ro := range h.options.Recorders { + if ro.Record == xrecorder.RecorderServiceHandler { + h.recorder = ro + break + } + } + return } @@ -58,20 +71,39 @@ func (h *redirectHandler) Handle(ctx context.Context, conn net.Conn, opts ...han defer conn.Close() start := time.Now() + + ro := &xrecorder.HandlerRecorderObject{ + Service: h.options.Service, + Network: "tcp", + RemoteAddr: conn.RemoteAddr().String(), + LocalAddr: conn.LocalAddr().String(), + Time: start, + SID: string(ctxvalue.SidFromContext(ctx)), + } + log := h.options.Logger.WithFields(map[string]any{ "remote": conn.RemoteAddr().String(), "local": conn.LocalAddr().String(), + "sid": ctxvalue.SidFromContext(ctx), }) log.Infof("%s <> %s", conn.RemoteAddr(), conn.LocalAddr()) defer func() { + if !ro.Time.IsZero() { + if err != nil { + ro.Err = err.Error() + } + ro.Duration = time.Since(start) + ro.Record(ctx, h.recorder.Recorder) + } + log.WithFields(map[string]any{ "duration": time.Since(start), }).Infof("%s >< %s", conn.RemoteAddr(), conn.LocalAddr()) }() if !h.checkRateLimit(conn.RemoteAddr()) { - return nil + return rate_limiter.ErrRateLimit } var dstAddr net.Addr @@ -86,6 +118,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()), }) @@ -111,7 +145,7 @@ func (h *redirectHandler) Handle(ctx context.Context, conn net.Conn, opts ...han // try to sniff HTTP traffic if isHTTP(string(hdr[:])) { - return h.handleHTTP(ctx, rw, conn.RemoteAddr(), dstAddr, log) + return h.handleHTTP(ctx, rw, conn.RemoteAddr(), dstAddr, ro, log) } } @@ -119,7 +153,7 @@ func (h *redirectHandler) Handle(ctx context.Context, conn net.Conn, opts ...han if h.options.Bypass != nil && h.options.Bypass.Contains(ctx, dstAddr.Network(), dstAddr.String()) { log.Debug("bypass: ", dstAddr) - return nil + return xbypass.ErrBypass } cc, err := h.options.Router.Dial(ctx, dstAddr.Network(), dstAddr.String()) @@ -139,17 +173,27 @@ func (h *redirectHandler) Handle(ctx context.Context, conn net.Conn, opts ...han return nil } -func (h *redirectHandler) handleHTTP(ctx context.Context, rw io.ReadWriter, raddr, dstAddr net.Addr, log logger.Logger) error { +func (h *redirectHandler) handleHTTP(ctx context.Context, rw io.ReadWriter, raddr, dstAddr net.Addr, ro *xrecorder.HandlerRecorderObject, log logger.Logger) error { req, err := http.ReadRequest(bufio.NewReader(rw)) if err != nil { return err } + ro.HTTP = &xrecorder.HTTPRecorderObject{ + Host: req.Host, + Proto: req.Proto, + Scheme: req.URL.Scheme, + Method: req.Method, + URI: req.RequestURI, + RequestHeader: req.Header, + } + if log.IsLevelEnabled(logger.TraceLevel) { dump, _ := httputil.DumpRequest(req, false) log.Trace(string(dump)) } + ro.Host = req.Host host := req.Host if _, _, err := net.SplitHostPort(host); err != nil { host = net.JoinHostPort(strings.Trim(host, "[]"), "80") @@ -160,12 +204,15 @@ func (h *redirectHandler) handleHTTP(ctx context.Context, rw io.ReadWriter, radd if h.options.Bypass != nil && h.options.Bypass.Contains(ctx, "tcp", host, bypass.WithPathOption(req.RequestURI)) { log.Debugf("bypass: %s %s", host, req.RequestURI) - return nil + return xbypass.ErrBypass } cc, err := h.options.Router.Dial(ctx, "tcp", host) if err != nil { log.Error(err) + if !h.md.sniffingFallback { + return err + } } if cc == nil { @@ -190,23 +237,28 @@ func (h *redirectHandler) handleHTTP(ctx context.Context, rw io.ReadWriter, radd return err } - var rw2 io.ReadWriter = cc - if log.IsLevelEnabled(logger.TraceLevel) { - var buf bytes.Buffer - resp, err := http.ReadResponse(bufio.NewReader(io.TeeReader(cc, &buf)), req) - if err != nil { - log.Error(err) - return err - } - defer resp.Body.Close() + br := bufio.NewReader(cc) + resp, err := http.ReadResponse(br, req) + if err != nil { + log.Error(err) + return err + } + defer resp.Body.Close() + ro.HTTP.StatusCode = resp.StatusCode + ro.HTTP.ResponseHeader = resp.Header + + if log.IsLevelEnabled(logger.TraceLevel) { dump, _ := httputil.DumpResponse(resp, false) log.Trace(string(dump)) - - rw2 = xio.NewReadWriter(io.MultiReader(&buf, cc), cc) } - netpkg.Transport(rw, rw2) + if err := resp.Write(rw); err != nil { + log.Error(err) + return err + } + + netpkg.Transport(rw, xio.NewReadWriter(br, cc)) return nil } diff --git a/handler/redirect/udp/handler.go b/handler/redirect/udp/handler.go index 3fe66929..38970183 100644 --- a/handler/redirect/udp/handler.go +++ b/handler/redirect/udp/handler.go @@ -8,7 +8,12 @@ import ( "github.com/go-gost/core/handler" md "github.com/go-gost/core/metadata" + "github.com/go-gost/core/recorder" + xbypass "github.com/go-gost/x/bypass" + ctxvalue "github.com/go-gost/x/ctx" netpkg "github.com/go-gost/x/internal/net" + rate_limiter "github.com/go-gost/x/limiter/rate" + xrecorder "github.com/go-gost/x/recorder" "github.com/go-gost/x/registry" ) @@ -17,8 +22,9 @@ func init() { } type redirectHandler struct { - md metadata - options handler.Options + md metadata + options handler.Options + recorder recorder.RecorderObject } func NewHandler(opts ...handler.Option) handler.Handler { @@ -37,30 +43,55 @@ func (h *redirectHandler) Init(md md.Metadata) (err error) { return } + for _, ro := range h.options.Recorders { + if ro.Record == xrecorder.RecorderServiceHandler { + h.recorder = ro + break + } + } + return } -func (h *redirectHandler) Handle(ctx context.Context, conn net.Conn, opts ...handler.HandleOption) error { +func (h *redirectHandler) Handle(ctx context.Context, conn net.Conn, opts ...handler.HandleOption) (err error) { defer conn.Close() start := time.Now() + + ro := &xrecorder.HandlerRecorderObject{ + Service: h.options.Service, + RemoteAddr: conn.RemoteAddr().String(), + LocalAddr: conn.LocalAddr().String(), + Time: start, + SID: string(ctxvalue.SidFromContext(ctx)), + } + log := h.options.Logger.WithFields(map[string]any{ "remote": conn.RemoteAddr().String(), "local": conn.LocalAddr().String(), + "sid": ctxvalue.SidFromContext(ctx), }) log.Infof("%s <> %s", conn.RemoteAddr(), conn.LocalAddr()) defer func() { + if err != nil { + ro.Err = err.Error() + } + ro.Duration = time.Since(start) + ro.Record(ctx, h.recorder.Recorder) + log.WithFields(map[string]any{ "duration": time.Since(start), }).Infof("%s >< %s", conn.RemoteAddr(), conn.LocalAddr()) }() if !h.checkRateLimit(conn.RemoteAddr()) { - return nil + return rate_limiter.ErrRateLimit } dstAddr := conn.LocalAddr() + ro.Network = dstAddr.Network() + ro.Host = dstAddr.String() log = log.WithFields(map[string]any{ "dst": fmt.Sprintf("%s/%s", dstAddr, dstAddr.Network()), @@ -70,7 +101,7 @@ func (h *redirectHandler) Handle(ctx context.Context, conn net.Conn, opts ...han if h.options.Bypass != nil && h.options.Bypass.Contains(ctx, dstAddr.Network(), dstAddr.String()) { log.Debug("bypass: ", dstAddr) - return nil + return xbypass.ErrBypass } cc, err := h.options.Router.Dial(ctx, dstAddr.Network(), dstAddr.String()) diff --git a/handler/relay/connect.go b/handler/relay/connect.go index 34fecf70..42f1b73a 100644 --- a/handler/relay/connect.go +++ b/handler/relay/connect.go @@ -12,6 +12,7 @@ import ( "github.com/go-gost/core/logger" "github.com/go-gost/core/observer/stats" "github.com/go-gost/relay" + xbypass "github.com/go-gost/x/bypass" ctxvalue "github.com/go-gost/x/ctx" xnet "github.com/go-gost/x/internal/net" serial "github.com/go-gost/x/internal/util/serial" @@ -49,8 +50,8 @@ func (h *relayHandler) handleConnect(ctx context.Context, conn net.Conn, network if h.options.Bypass != nil && h.options.Bypass.Contains(ctx, network, address) { log.Debug("bypass: ", address) resp.Status = relay.StatusForbidden - _, err = resp.WriteTo(conn) - return + resp.WriteTo(conn) + return xbypass.ErrBypass } switch h.md.hash { diff --git a/handler/relay/handler.go b/handler/relay/handler.go index ca9c3c81..e8014df6 100644 --- a/handler/relay/handler.go +++ b/handler/relay/handler.go @@ -11,10 +11,13 @@ import ( "github.com/go-gost/core/hop" "github.com/go-gost/core/limiter/traffic" md "github.com/go-gost/core/metadata" + "github.com/go-gost/core/recorder" "github.com/go-gost/relay" ctxvalue "github.com/go-gost/x/ctx" limiter_util "github.com/go-gost/x/internal/util/limiter" stats_util "github.com/go-gost/x/internal/util/stats" + rate_limiter "github.com/go-gost/x/limiter/rate" + xrecorder "github.com/go-gost/x/recorder" "github.com/go-gost/x/registry" ) @@ -22,7 +25,6 @@ var ( ErrBadVersion = errors.New("relay: bad version") ErrUnknownCmd = errors.New("relay: unknown command") ErrUnauthorized = errors.New("relay: unauthorized") - ErrRateLimit = errors.New("relay: rate limiting exceeded") ) func init() { @@ -30,12 +32,13 @@ func init() { } type relayHandler struct { - hop hop.Hop - md metadata - options handler.Options - stats *stats_util.HandlerStats - limiter traffic.TrafficLimiter - cancel context.CancelFunc + hop hop.Hop + md metadata + options handler.Options + stats *stats_util.HandlerStats + limiter traffic.TrafficLimiter + cancel context.CancelFunc + recorder recorder.RecorderObject } func NewHandler(opts ...handler.Option) handler.Handler { @@ -66,6 +69,13 @@ func (h *relayHandler) Init(md md.Metadata) (err error) { h.limiter = limiter_util.NewCachedTrafficLimiter(limiter, 30*time.Second, 60*time.Second) } + for _, ro := range h.options.Recorders { + if ro.Record == xrecorder.RecorderServiceHandler { + h.recorder = ro + break + } + } + return nil } @@ -75,25 +85,40 @@ func (h *relayHandler) Forward(hop hop.Hop) { } func (h *relayHandler) Handle(ctx context.Context, conn net.Conn, opts ...handler.HandleOption) (err error) { + defer conn.Close() + start := time.Now() + + ro := &xrecorder.HandlerRecorderObject{ + Service: h.options.Service, + RemoteAddr: conn.RemoteAddr().String(), + LocalAddr: conn.LocalAddr().String(), + Time: start, + SID: string(ctxvalue.SidFromContext(ctx)), + } + log := h.options.Logger.WithFields(map[string]any{ "remote": conn.RemoteAddr().String(), "local": conn.LocalAddr().String(), + "sid": ctxvalue.SidFromContext(ctx), }) log.Infof("%s <> %s", conn.RemoteAddr(), conn.LocalAddr()) defer func() { if err != nil { - conn.Close() + ro.Err = err.Error() } + ro.Duration = time.Since(start) + ro.Record(ctx, h.recorder.Recorder) + log.WithFields(map[string]any{ "duration": time.Since(start), }).Infof("%s >< %s", conn.RemoteAddr(), conn.LocalAddr()) }() if !h.checkRateLimit(conn.RemoteAddr()) { - return ErrRateLimit + return rate_limiter.ErrRateLimit } if h.md.readTimeout > 0 { @@ -139,6 +164,7 @@ func (h *relayHandler) Handle(ctx context.Context, conn net.Conn, opts ...handle } if user != "" { + ro.Client = user log = log.WithFields(map[string]any{"user": user}) } @@ -156,21 +182,18 @@ func (h *relayHandler) Handle(ctx context.Context, conn net.Conn, opts ...handle if (req.Cmd & relay.FUDP) == relay.FUDP { network = "udp" } + ro.Network = network + ro.Host = address if h.hop != nil { - defer conn.Close() // forward mode return h.handleForward(ctx, conn, network, log) } switch req.Cmd & relay.CmdMask { case 0, relay.CmdConnect: - defer conn.Close() - return h.handleConnect(ctx, conn, network, address, log) case relay.CmdBind: - defer conn.Close() - return h.handleBind(ctx, conn, network, address, log) default: resp.Status = relay.StatusBadRequest diff --git a/handler/serial/handler.go b/handler/serial/handler.go index 313e4306..ca61125a 100644 --- a/handler/serial/handler.go +++ b/handler/serial/handler.go @@ -13,6 +13,7 @@ import ( "github.com/go-gost/core/logger" md "github.com/go-gost/core/metadata" "github.com/go-gost/core/recorder" + ctxvalue "github.com/go-gost/x/ctx" xnet "github.com/go-gost/x/internal/net" serial "github.com/go-gost/x/internal/util/serial" xrecorder "github.com/go-gost/x/recorder" @@ -46,12 +47,10 @@ func (h *serialHandler) Init(md md.Metadata) (err error) { return } - if opts := h.options.Router.Options(); opts != nil { - for _, ro := range opts.Recorders { - if ro.Record == xrecorder.RecorderServiceHandlerSerial { - h.recorder = ro - break - } + for _, ro := range h.options.Recorders { + if ro.Record == xrecorder.RecorderServiceHandlerSerial { + h.recorder = ro + break } } @@ -71,6 +70,7 @@ func (h *serialHandler) Handle(ctx context.Context, conn net.Conn, opts ...handl log = log.WithFields(map[string]any{ "remote": conn.RemoteAddr().String(), "local": conn.LocalAddr().String(), + "sid": ctxvalue.SidFromContext(ctx), }) conn = &recorderConn{ diff --git a/handler/sni/handler.go b/handler/sni/handler.go index 8875ec4e..f1868832 100644 --- a/handler/sni/handler.go +++ b/handler/sni/handler.go @@ -20,10 +20,14 @@ import ( "github.com/go-gost/core/handler" "github.com/go-gost/core/logger" md "github.com/go-gost/core/metadata" + "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" netpkg "github.com/go-gost/x/internal/net" + rate_limiter "github.com/go-gost/x/limiter/rate" + xrecorder "github.com/go-gost/x/recorder" "github.com/go-gost/x/registry" ) @@ -32,8 +36,9 @@ func init() { } type sniHandler struct { - md metadata - options handler.Options + md metadata + options handler.Options + recorder recorder.RecorderObject } func NewHandler(opts ...handler.Option) handler.Handler { @@ -54,27 +59,51 @@ func (h *sniHandler) Init(md md.Metadata) (err error) { return } + for _, ro := range h.options.Recorders { + if ro.Record == xrecorder.RecorderServiceHandler { + h.recorder = ro + break + } + } + return nil } -func (h *sniHandler) Handle(ctx context.Context, conn net.Conn, opts ...handler.HandleOption) error { +func (h *sniHandler) Handle(ctx context.Context, conn net.Conn, opts ...handler.HandleOption) (err error) { defer conn.Close() start := time.Now() + + ro := &xrecorder.HandlerRecorderObject{ + Service: h.options.Service, + RemoteAddr: conn.RemoteAddr().String(), + LocalAddr: conn.LocalAddr().String(), + Network: "tcp", + Time: start, + SID: string(ctxvalue.SidFromContext(ctx)), + } + log := h.options.Logger.WithFields(map[string]any{ "remote": conn.RemoteAddr().String(), "local": conn.LocalAddr().String(), + "sid": ctxvalue.SidFromContext(ctx), }) log.Infof("%s <> %s", conn.RemoteAddr(), conn.LocalAddr()) defer func() { + if err != nil { + ro.Err = err.Error() + } + ro.Duration = time.Since(start) + ro.Record(ctx, h.recorder.Recorder) + log.WithFields(map[string]any{ "duration": time.Since(start), }).Infof("%s >< %s", conn.RemoteAddr(), conn.LocalAddr()) }() if !h.checkRateLimit(conn.RemoteAddr()) { - return nil + return rate_limiter.ErrRateLimit } var hdr [dissector.RecordHeaderLen]byte @@ -88,17 +117,26 @@ func (h *sniHandler) Handle(ctx context.Context, conn net.Conn, opts ...handler. tlsVersion := binary.BigEndian.Uint16(hdr[1:3]) if hdr[0] == dissector.Handshake && (tlsVersion >= tls.VersionTLS10 && tlsVersion <= tls.VersionTLS13) { - return h.handleHTTPS(ctx, rw, conn.RemoteAddr(), log) + return h.handleHTTPS(ctx, rw, conn.RemoteAddr(), ro, log) } - return h.handleHTTP(ctx, rw, conn.RemoteAddr(), log) + return h.handleHTTP(ctx, rw, conn.RemoteAddr(), ro, log) } -func (h *sniHandler) handleHTTP(ctx context.Context, rw io.ReadWriter, raddr net.Addr, log logger.Logger) error { +func (h *sniHandler) handleHTTP(ctx context.Context, rw io.ReadWriter, raddr net.Addr, ro *xrecorder.HandlerRecorderObject, log logger.Logger) error { req, err := http.ReadRequest(bufio.NewReader(rw)) if err != nil { return err } + ro.HTTP = &xrecorder.HTTPRecorderObject{ + Host: req.Host, + Proto: req.Proto, + Scheme: req.URL.Scheme, + Method: req.Method, + URI: req.RequestURI, + RequestHeader: req.Header, + } + if log.IsLevelEnabled(logger.TraceLevel) { dump, _ := httputil.DumpRequest(req, false) log.Trace(string(dump)) @@ -108,13 +146,16 @@ func (h *sniHandler) handleHTTP(ctx context.Context, rw io.ReadWriter, raddr net if _, _, err := net.SplitHostPort(host); err != nil { host = net.JoinHostPort(strings.Trim(host, "[]"), "80") } + + ro.Host = req.Host + log = log.WithFields(map[string]any{ "host": host, }) if h.options.Bypass != nil && h.options.Bypass.Contains(ctx, "tcp", host, bypass.WithPathOption(req.RequestURI)) { log.Debugf("bypass: %s %s", host, req.RequestURI) - return nil + return xbypass.ErrBypass } switch h.md.hash { @@ -142,28 +183,33 @@ func (h *sniHandler) handleHTTP(ctx context.Context, rw io.ReadWriter, raddr net return err } - var rw2 io.ReadWriter = cc - if log.IsLevelEnabled(logger.TraceLevel) { - var buf bytes.Buffer - resp, err := http.ReadResponse(bufio.NewReader(io.TeeReader(cc, &buf)), req) - if err != nil { - log.Error(err) - return err - } - defer resp.Body.Close() + br := bufio.NewReader(cc) + resp, err := http.ReadResponse(br, req) + if err != nil { + log.Error(err) + return err + } + defer resp.Body.Close() + ro.HTTP.StatusCode = resp.StatusCode + ro.HTTP.ResponseHeader = resp.Header + + if log.IsLevelEnabled(logger.TraceLevel) { dump, _ := httputil.DumpResponse(resp, false) log.Trace(string(dump)) - - rw2 = xio.NewReadWriter(io.MultiReader(&buf, cc), cc) } - netpkg.Transport(rw, rw2) + if err := resp.Write(rw); err != nil { + log.Error(err) + return err + } + + netpkg.Transport(rw, xio.NewReadWriter(br, cc)) return nil } -func (h *sniHandler) handleHTTPS(ctx context.Context, rw io.ReadWriter, raddr net.Addr, log logger.Logger) error { +func (h *sniHandler) handleHTTPS(ctx context.Context, rw io.ReadWriter, raddr net.Addr, ro *xrecorder.HandlerRecorderObject, log logger.Logger) error { buf := new(bytes.Buffer) host, err := h.decodeHost(io.TeeReader(rw, buf)) if err != nil { @@ -175,6 +221,8 @@ func (h *sniHandler) handleHTTPS(ctx context.Context, rw io.ReadWriter, raddr ne host = net.JoinHostPort(strings.Trim(host, "[]"), "443") } + ro.Host = host + log = log.WithFields(map[string]any{ "dst": host, }) @@ -182,7 +230,7 @@ func (h *sniHandler) handleHTTPS(ctx context.Context, rw io.ReadWriter, raddr ne if h.options.Bypass != nil && h.options.Bypass.Contains(ctx, "tcp", host) { log.Debug("bypass: ", host) - return nil + return xbypass.ErrBypass } switch h.md.hash { diff --git a/handler/socks/v4/handler.go b/handler/socks/v4/handler.go index d03c7b5f..5266760a 100644 --- a/handler/socks/v4/handler.go +++ b/handler/socks/v4/handler.go @@ -12,13 +12,16 @@ import ( "github.com/go-gost/core/logger" md "github.com/go-gost/core/metadata" "github.com/go-gost/core/observer/stats" + "github.com/go-gost/core/recorder" "github.com/go-gost/gosocks4" ctxvalue "github.com/go-gost/x/ctx" netpkg "github.com/go-gost/x/internal/net" limiter_util "github.com/go-gost/x/internal/util/limiter" stats_util "github.com/go-gost/x/internal/util/stats" + rate_limiter "github.com/go-gost/x/limiter/rate" traffic_wrapper "github.com/go-gost/x/limiter/traffic/wrapper" stats_wrapper "github.com/go-gost/x/observer/stats/wrapper" + xrecorder "github.com/go-gost/x/recorder" "github.com/go-gost/x/registry" ) @@ -33,11 +36,12 @@ func init() { } type socks4Handler struct { - md metadata - options handler.Options - stats *stats_util.HandlerStats - limiter traffic.TrafficLimiter - cancel context.CancelFunc + md metadata + options handler.Options + stats *stats_util.HandlerStats + limiter traffic.TrafficLimiter + cancel context.CancelFunc + recorder recorder.RecorderObject } func NewHandler(opts ...handler.Option) handler.Handler { @@ -68,28 +72,51 @@ func (h *socks4Handler) Init(md md.Metadata) (err error) { h.limiter = limiter_util.NewCachedTrafficLimiter(limiter, 30*time.Second, 60*time.Second) } + for _, ro := range h.options.Recorders { + if ro.Record == xrecorder.RecorderServiceHandler { + h.recorder = ro + break + } + } + return nil } -func (h *socks4Handler) Handle(ctx context.Context, conn net.Conn, opts ...handler.HandleOption) error { +func (h *socks4Handler) Handle(ctx context.Context, conn net.Conn, opts ...handler.HandleOption) (err error) { defer conn.Close() start := time.Now() + ro := &xrecorder.HandlerRecorderObject{ + Service: h.options.Service, + Network: "tcp", + RemoteAddr: conn.RemoteAddr().String(), + LocalAddr: conn.LocalAddr().String(), + Time: start, + SID: string(ctxvalue.SidFromContext(ctx)), + } + log := h.options.Logger.WithFields(map[string]any{ "remote": conn.RemoteAddr().String(), "local": conn.LocalAddr().String(), + "sid": ctxvalue.SidFromContext(ctx), }) log.Infof("%s <> %s", conn.RemoteAddr(), conn.LocalAddr()) defer func() { + if err != nil { + ro.Err = err.Error() + } + ro.Duration = time.Since(start) + ro.Record(ctx, h.recorder.Recorder) + log.WithFields(map[string]any{ "duration": time.Since(start), }).Infof("%s >< %s", conn.RemoteAddr(), conn.LocalAddr()) }() if !h.checkRateLimit(conn.RemoteAddr()) { - return nil + return rate_limiter.ErrRateLimit } if h.md.readTimeout > 0 { @@ -101,6 +128,8 @@ func (h *socks4Handler) Handle(ctx context.Context, conn net.Conn, opts ...handl log.Error(err) return err } + + ro.Host = req.Addr.String() log.Trace(req) conn.SetReadDeadline(time.Time{}) diff --git a/handler/socks/v5/handler.go b/handler/socks/v5/handler.go index 3374f5fb..5e52d034 100644 --- a/handler/socks/v5/handler.go +++ b/handler/socks/v5/handler.go @@ -9,11 +9,14 @@ import ( "github.com/go-gost/core/handler" "github.com/go-gost/core/limiter/traffic" md "github.com/go-gost/core/metadata" + "github.com/go-gost/core/recorder" "github.com/go-gost/gosocks5" ctxvalue "github.com/go-gost/x/ctx" limiter_util "github.com/go-gost/x/internal/util/limiter" "github.com/go-gost/x/internal/util/socks" stats_util "github.com/go-gost/x/internal/util/stats" + rate_limiter "github.com/go-gost/x/limiter/rate" + xrecorder "github.com/go-gost/x/recorder" "github.com/go-gost/x/registry" ) @@ -33,6 +36,7 @@ type socks5Handler struct { stats *stats_util.HandlerStats limiter traffic.TrafficLimiter cancel context.CancelFunc + recorder recorder.RecorderObject } func NewHandler(opts ...handler.Option) handler.Handler { @@ -70,28 +74,51 @@ func (h *socks5Handler) Init(md md.Metadata) (err error) { h.limiter = limiter_util.NewCachedTrafficLimiter(limiter, 30*time.Second, 60*time.Second) } + for _, ro := range h.options.Recorders { + if ro.Record == xrecorder.RecorderServiceHandler { + h.recorder = ro + break + } + } + return } -func (h *socks5Handler) Handle(ctx context.Context, conn net.Conn, opts ...handler.HandleOption) error { +func (h *socks5Handler) Handle(ctx context.Context, conn net.Conn, opts ...handler.HandleOption) (err error) { defer conn.Close() start := time.Now() + ro := &xrecorder.HandlerRecorderObject{ + Service: h.options.Service, + Network: "tcp", + RemoteAddr: conn.RemoteAddr().String(), + LocalAddr: conn.LocalAddr().String(), + Time: start, + SID: string(ctxvalue.SidFromContext(ctx)), + } + log := h.options.Logger.WithFields(map[string]any{ "remote": conn.RemoteAddr().String(), "local": conn.LocalAddr().String(), + "sid": ctxvalue.SidFromContext(ctx), }) log.Infof("%s <> %s", conn.RemoteAddr(), conn.LocalAddr()) defer func() { + if err != nil { + ro.Err = err.Error() + } + ro.Duration = time.Since(start) + ro.Record(ctx, h.recorder.Recorder) + log.WithFields(map[string]any{ "duration": time.Since(start), }).Infof("%s >< %s", conn.RemoteAddr(), conn.LocalAddr()) }() if !h.checkRateLimit(conn.RemoteAddr()) { - return nil + return rate_limiter.ErrRateLimit } if h.md.readTimeout > 0 { @@ -109,12 +136,14 @@ func (h *socks5Handler) Handle(ctx context.Context, conn net.Conn, opts ...handl if clientID := sc.ID(); clientID != "" { ctx = ctxvalue.ContextWithClientID(ctx, ctxvalue.ClientID(clientID)) log = log.WithFields(map[string]any{"user": clientID}) + ro.Client = clientID } conn = sc conn.SetReadDeadline(time.Time{}) address := req.Addr.String() + ro.Host = address switch req.Cmd { case gosocks5.CmdConnect: @@ -124,8 +153,10 @@ func (h *socks5Handler) Handle(ctx context.Context, conn net.Conn, opts ...handl case socks.CmdMuxBind: return h.handleMuxBind(ctx, conn, "tcp", address, log) case gosocks5.CmdUdp: + ro.Network = "udp" return h.handleUDP(ctx, conn, log) case socks.CmdUDPTun: + ro.Network = "udp" return h.handleUDPTun(ctx, conn, "udp", address, log) default: err = ErrUnknownCmd diff --git a/handler/ss/handler.go b/handler/ss/handler.go index c6315cb2..81bb5d82 100644 --- a/handler/ss/handler.go +++ b/handler/ss/handler.go @@ -8,10 +8,13 @@ import ( "github.com/go-gost/core/handler" md "github.com/go-gost/core/metadata" + "github.com/go-gost/core/recorder" "github.com/go-gost/gosocks5" ctxvalue "github.com/go-gost/x/ctx" netpkg "github.com/go-gost/x/internal/net" "github.com/go-gost/x/internal/util/ss" + rate_limiter "github.com/go-gost/x/limiter/rate" + xrecorder "github.com/go-gost/x/recorder" "github.com/go-gost/x/registry" "github.com/shadowsocks/go-shadowsocks2/core" ) @@ -21,9 +24,10 @@ func init() { } type ssHandler struct { - cipher core.Cipher - md metadata - options handler.Options + cipher core.Cipher + md metadata + options handler.Options + recorder recorder.RecorderObject } func NewHandler(opts ...handler.Option) handler.Handler { @@ -50,27 +54,53 @@ func (h *ssHandler) Init(md md.Metadata) (err error) { } } + for _, ro := range h.options.Recorders { + if ro.Record == xrecorder.RecorderServiceHandler { + h.recorder = ro + break + } + } + return } -func (h *ssHandler) Handle(ctx context.Context, conn net.Conn, opts ...handler.HandleOption) error { +func (h *ssHandler) Handle(ctx context.Context, conn net.Conn, opts ...handler.HandleOption) (err error) { defer conn.Close() start := time.Now() + + ro := &xrecorder.HandlerRecorderObject{ + Service: h.options.Service, + Network: "tcp", + RemoteAddr: conn.RemoteAddr().String(), + LocalAddr: conn.LocalAddr().String(), + Time: start, + SID: string(ctxvalue.SidFromContext(ctx)), + } + log := h.options.Logger.WithFields(map[string]any{ "remote": conn.RemoteAddr().String(), "local": conn.LocalAddr().String(), + "sid": ctxvalue.SidFromContext(ctx), }) log.Infof("%s <> %s", conn.RemoteAddr(), conn.LocalAddr()) defer func() { + if !ro.Time.IsZero() { + if err != nil { + ro.Err = err.Error() + } + ro.Duration = time.Since(start) + ro.Record(ctx, h.recorder.Recorder) + } + log.WithFields(map[string]any{ "duration": time.Since(start), }).Infof("%s >< %s", conn.RemoteAddr(), conn.LocalAddr()) }() if !h.checkRateLimit(conn.RemoteAddr()) { - return nil + return rate_limiter.ErrRateLimit } if h.cipher != nil { @@ -87,6 +117,7 @@ func (h *ssHandler) Handle(ctx context.Context, conn net.Conn, opts ...handler.H io.Copy(io.Discard, conn) return err } + ro.Host = addr.String() log = log.WithFields(map[string]any{ "dst": addr.String(), diff --git a/handler/ss/udp/handler.go b/handler/ss/udp/handler.go index ef8cfede..9d33b181 100644 --- a/handler/ss/udp/handler.go +++ b/handler/ss/udp/handler.go @@ -10,8 +10,12 @@ import ( "github.com/go-gost/core/handler" "github.com/go-gost/core/logger" md "github.com/go-gost/core/metadata" + "github.com/go-gost/core/recorder" + ctxvalue "github.com/go-gost/x/ctx" "github.com/go-gost/x/internal/util/relay" "github.com/go-gost/x/internal/util/ss" + rate_limiter "github.com/go-gost/x/limiter/rate" + xrecorder "github.com/go-gost/x/recorder" "github.com/go-gost/x/registry" "github.com/shadowsocks/go-shadowsocks2/core" ) @@ -21,9 +25,10 @@ func init() { } type ssuHandler struct { - cipher core.Cipher - md metadata - options handler.Options + cipher core.Cipher + md metadata + options handler.Options + recorder recorder.RecorderObject } func NewHandler(opts ...handler.Option) handler.Handler { @@ -51,27 +56,51 @@ func (h *ssuHandler) Init(md md.Metadata) (err error) { } } + for _, ro := range h.options.Recorders { + if ro.Record == xrecorder.RecorderServiceHandler { + h.recorder = ro + break + } + } + return } -func (h *ssuHandler) Handle(ctx context.Context, conn net.Conn, opts ...handler.HandleOption) error { +func (h *ssuHandler) Handle(ctx context.Context, conn net.Conn, opts ...handler.HandleOption) (err error) { defer conn.Close() start := time.Now() + + ro := &xrecorder.HandlerRecorderObject{ + Service: h.options.Service, + Network: "udp", + RemoteAddr: conn.RemoteAddr().String(), + LocalAddr: conn.LocalAddr().String(), + Time: start, + SID: string(ctxvalue.SidFromContext(ctx)), + } + log := h.options.Logger.WithFields(map[string]any{ "remote": conn.RemoteAddr().String(), "local": conn.LocalAddr().String(), + "sid": ctxvalue.SidFromContext(ctx), }) log.Infof("%s <> %s", conn.RemoteAddr(), conn.LocalAddr()) defer func() { + if err != nil { + ro.Err = err.Error() + } + ro.Duration = time.Since(start) + ro.Record(ctx, h.recorder.Recorder) + log.WithFields(map[string]any{ "duration": time.Since(start), }).Infof("%s >< %s", conn.RemoteAddr(), conn.LocalAddr()) }() if !h.checkRateLimit(conn.RemoteAddr()) { - return nil + return rate_limiter.ErrRateLimit } pc, ok := conn.(net.PacketConn) @@ -106,14 +135,14 @@ func (h *ssuHandler) Handle(ctx context.Context, conn net.Conn, opts ...handler. t := time.Now() log.Infof("%s <-> %s", conn.LocalAddr(), cc.LocalAddr()) - h.relayPacket(pc, cc, log) + h.relayPacket(ctx, pc, cc, ro, log) log.WithFields(map[string]any{"duration": time.Since(t)}). Infof("%s >-< %s", conn.LocalAddr(), cc.LocalAddr()) return nil } -func (h *ssuHandler) relayPacket(pc1, pc2 net.PacketConn, log logger.Logger) (err error) { +func (h *ssuHandler) relayPacket(ctx context.Context, pc1, pc2 net.PacketConn, ro *xrecorder.HandlerRecorderObject, log logger.Logger) (err error) { bufSize := h.md.bufferSize errc := make(chan error, 2) @@ -128,7 +157,11 @@ func (h *ssuHandler) relayPacket(pc1, pc2 net.PacketConn, log logger.Logger) (er return err } - if h.options.Bypass != nil && h.options.Bypass.Contains(context.Background(), addr.Network(), addr.String()) { + if ro.Host == "" { + ro.Host = addr.String() + } + + if h.options.Bypass != nil && h.options.Bypass.Contains(ctx, addr.Network(), addr.String()) { log.Warn("bypass: ", addr) return nil } @@ -160,7 +193,7 @@ func (h *ssuHandler) relayPacket(pc1, pc2 net.PacketConn, log logger.Logger) (er return err } - if h.options.Bypass != nil && h.options.Bypass.Contains(context.Background(), raddr.Network(), raddr.String()) { + if h.options.Bypass != nil && h.options.Bypass.Contains(ctx, raddr.Network(), raddr.String()) { log.Warn("bypass: ", raddr) return nil } diff --git a/handler/sshd/handler.go b/handler/sshd/handler.go index 884f6295..a6f9e1be 100644 --- a/handler/sshd/handler.go +++ b/handler/sshd/handler.go @@ -12,8 +12,13 @@ import ( "github.com/go-gost/core/handler" "github.com/go-gost/core/logger" md "github.com/go-gost/core/metadata" + "github.com/go-gost/core/recorder" + xbypass "github.com/go-gost/x/bypass" + ctxvalue "github.com/go-gost/x/ctx" netpkg "github.com/go-gost/x/internal/net" sshd_util "github.com/go-gost/x/internal/util/sshd" + rate_limiter "github.com/go-gost/x/limiter/rate" + xrecorder "github.com/go-gost/x/recorder" "github.com/go-gost/x/registry" "golang.org/x/crypto/ssh" ) @@ -28,8 +33,9 @@ func init() { } type forwardHandler struct { - md metadata - options handler.Options + md metadata + options handler.Options + recorder recorder.RecorderObject } func NewHandler(opts ...handler.Option) handler.Handler { @@ -48,26 +54,60 @@ func (h *forwardHandler) Init(md md.Metadata) (err error) { return } + for _, ro := range h.options.Recorders { + if ro.Record == xrecorder.RecorderServiceHandler { + h.recorder = ro + break + } + } + return nil } -func (h *forwardHandler) Handle(ctx context.Context, conn net.Conn, opts ...handler.HandleOption) error { +func (h *forwardHandler) Handle(ctx context.Context, conn net.Conn, opts ...handler.HandleOption) (err error) { defer conn.Close() + start := time.Now() + + ro := &xrecorder.HandlerRecorderObject{ + Service: h.options.Service, + Network: "tcp", + RemoteAddr: conn.RemoteAddr().String(), + LocalAddr: conn.LocalAddr().String(), + Time: start, + SID: string(ctxvalue.SidFromContext(ctx)), + } + log := h.options.Logger.WithFields(map[string]any{ "remote": conn.RemoteAddr().String(), "local": conn.LocalAddr().String(), + "sid": ctxvalue.SidFromContext(ctx), }) + log.Infof("%s <> %s", conn.RemoteAddr(), conn.LocalAddr()) + defer func() { + if !ro.Time.IsZero() { + if err != nil { + ro.Err = err.Error() + } + ro.Duration = time.Since(start) + ro.Record(ctx, h.recorder.Recorder) + } + + log.WithFields(map[string]any{ + "duration": time.Since(start), + }).Infof("%s >< %s", conn.RemoteAddr(), conn.LocalAddr()) + }() + if !h.checkRateLimit(conn.RemoteAddr()) { - return nil + return rate_limiter.ErrRateLimit } switch cc := conn.(type) { case *sshd_util.DirectForwardConn: - return h.handleDirectForward(ctx, cc, log) + return h.handleDirectForward(ctx, cc, ro, log) case *sshd_util.RemoteForwardConn: - return h.handleRemoteForward(ctx, cc, log) + return h.handleRemoteForward(ctx, cc, ro, log) default: err := errors.New("sshd: wrong connection type") log.Error(err) @@ -75,9 +115,10 @@ func (h *forwardHandler) Handle(ctx context.Context, conn net.Conn, opts ...hand } } -func (h *forwardHandler) handleDirectForward(ctx context.Context, conn *sshd_util.DirectForwardConn, log logger.Logger) error { +func (h *forwardHandler) handleDirectForward(ctx context.Context, conn *sshd_util.DirectForwardConn, ro *xrecorder.HandlerRecorderObject, log logger.Logger) error { targetAddr := conn.DstAddr() + ro.Host = targetAddr log = log.WithFields(map[string]any{ "dst": fmt.Sprintf("%s/%s", targetAddr, "tcp"), "cmd": "connect", @@ -87,7 +128,7 @@ func (h *forwardHandler) handleDirectForward(ctx context.Context, conn *sshd_uti if h.options.Bypass != nil && h.options.Bypass.Contains(ctx, "tcp", targetAddr) { log.Debugf("bypass %s", targetAddr) - return nil + return xbypass.ErrBypass } cc, err := h.options.Router.Dial(ctx, "tcp", targetAddr) @@ -106,7 +147,7 @@ func (h *forwardHandler) handleDirectForward(ctx context.Context, conn *sshd_uti return nil } -func (h *forwardHandler) handleRemoteForward(ctx context.Context, conn *sshd_util.RemoteForwardConn, log logger.Logger) error { +func (h *forwardHandler) handleRemoteForward(ctx context.Context, conn *sshd_util.RemoteForwardConn, ro *xrecorder.HandlerRecorderObject, log logger.Logger) error { req := conn.Request() t := tcpipForward{} @@ -117,6 +158,7 @@ func (h *forwardHandler) handleRemoteForward(ctx context.Context, conn *sshd_uti network := "tcp" addr := net.JoinHostPort(t.Host, strconv.Itoa(int(t.Port))) + ro.Host = addr log = log.WithFields(map[string]any{ "dst": fmt.Sprintf("%s/%s", addr, network), diff --git a/handler/tap/handler.go b/handler/tap/handler.go index 66be14ee..e1ab956e 100644 --- a/handler/tap/handler.go +++ b/handler/tap/handler.go @@ -21,6 +21,7 @@ import ( "github.com/go-gost/x/registry" "github.com/shadowsocks/go-shadowsocks2/core" "github.com/shadowsocks/go-shadowsocks2/shadowaead" + ctxvalue "github.com/go-gost/x/ctx" "github.com/songgao/water/waterutil" ) @@ -87,6 +88,7 @@ func (h *tapHandler) Handle(ctx context.Context, conn net.Conn, opts ...handler. log = log.WithFields(map[string]any{ "remote": conn.RemoteAddr().String(), "local": conn.LocalAddr().String(), + "sid": ctxvalue.SidFromContext(ctx), }) log.Infof("%s <> %s", conn.RemoteAddr(), conn.LocalAddr()) diff --git a/handler/tun/handler.go b/handler/tun/handler.go index 94aa93f2..dc2edc60 100644 --- a/handler/tun/handler.go +++ b/handler/tun/handler.go @@ -15,6 +15,7 @@ import ( tun_util "github.com/go-gost/x/internal/util/tun" "github.com/go-gost/x/registry" "github.com/songgao/water/waterutil" + ctxvalue "github.com/go-gost/x/ctx" ) var ( @@ -74,6 +75,7 @@ func (h *tunHandler) Handle(ctx context.Context, conn net.Conn, opts ...handler. log = log.WithFields(map[string]any{ "remote": conn.RemoteAddr().String(), "local": conn.LocalAddr().String(), + "sid": ctxvalue.SidFromContext(ctx), }) log.Infof("%s <> %s", conn.RemoteAddr(), conn.LocalAddr()) diff --git a/handler/tunnel/entrypoint.go b/handler/tunnel/entrypoint.go index 90859ee8..243c00fa 100644 --- a/handler/tunnel/entrypoint.go +++ b/handler/tunnel/entrypoint.go @@ -17,6 +17,7 @@ import ( "github.com/go-gost/core/listener" "github.com/go-gost/core/logger" md "github.com/go-gost/core/metadata" + "github.com/go-gost/core/recorder" "github.com/go-gost/core/sd" "github.com/go-gost/relay" admission "github.com/go-gost/x/admission/wrapper" @@ -25,14 +26,17 @@ import ( "github.com/go-gost/x/internal/net/proxyproto" climiter "github.com/go-gost/x/limiter/conn/wrapper" metrics "github.com/go-gost/x/metrics/wrapper" + xrecorder "github.com/go-gost/x/recorder" ) type entrypoint struct { - node string - pool *ConnectorPool - ingress ingress.Ingress - sd sd.SD - log logger.Logger + node string + service string + pool *ConnectorPool + ingress ingress.Ingress + sd sd.SD + log logger.Logger + recorder recorder.RecorderObject } func (ep *entrypoint) handle(ctx context.Context, conn net.Conn) error { @@ -77,6 +81,40 @@ func (ep *entrypoint) handle(ctx context.Context, conn net.Conn) error { return err } + start := time.Now() + ro := &xrecorder.HandlerRecorderObject{ + Node: ep.node, + Service: ep.service, + RemoteAddr: conn.RemoteAddr().String(), + LocalAddr: conn.LocalAddr().String(), + Network: "tcp", + Host: req.Host, + Time: start, + HTTP: &xrecorder.HTTPRecorderObject{ + Host: req.Host, + Method: req.Method, + Proto: req.Proto, + Scheme: req.URL.Scheme, + URI: req.RequestURI, + RequestHeader: req.Header, + }, + } + defer func() { + if err != nil { + d := time.Since(start) + log.WithFields(map[string]any{ + "duration": d, + }).Debugf("%s >-< %s", conn.RemoteAddr(), req.Host) + + ro.HTTP.StatusCode = resp.StatusCode + ro.HTTP.ResponseHeader = resp.Header + + ro.Duration = d + ro.Err = err.Error() + ro.Record(ctx, ep.recorder.Recorder) + } + }() + if log.IsLevelEnabled(logger.TraceLevel) { dump, _ := httputil.DumpRequest(req, false) log.Trace(string(dump)) @@ -92,16 +130,21 @@ func (ep *entrypoint) handle(ctx context.Context, conn net.Conn) error { } } if tunnelID.IsZero() { - err := fmt.Errorf("no route to host %s", req.Host) + err = fmt.Errorf("no route to host %s", req.Host) log.Error(err) resp.StatusCode = http.StatusBadGateway - return resp.Write(conn) + resp.Write(conn) + return err } + + ro.Client = tunnelID.String() + if tunnelID.IsPrivate() { - err := fmt.Errorf("access denied: tunnel %s is private for host %s", tunnelID, req.Host) + err = fmt.Errorf("access denied: tunnel %s is private for host %s", tunnelID, req.Host) log.Error(err) resp.StatusCode = http.StatusBadGateway - return resp.Write(conn) + resp.Write(conn) + return err } log = log.WithFields(map[string]any{ @@ -116,6 +159,7 @@ func (ep *entrypoint) handle(ctx context.Context, conn net.Conn) error { }) remoteAddr = addr } + ro.RemoteAddr = remoteAddr.String() d := &Dialer{ node: ep.node, @@ -165,14 +209,15 @@ func (ep *entrypoint) handle(ctx context.Context, conn net.Conn) error { } } - if err := req.Write(c); err != nil { + if err = req.Write(c); err != nil { c.Close() log.Errorf("send request: %v", err) - return resp.Write(conn) + resp.Write(conn) + return err } if req.Header.Get("Upgrade") == "websocket" { - err := xnet.Transport(c, xio.NewReadWriter(br, conn)) + err = xnet.Transport(c, xio.NewReadWriter(br, conn)) if err == nil { err = io.EOF } @@ -182,21 +227,35 @@ func (ep *entrypoint) handle(ctx context.Context, conn net.Conn) error { go func() { defer c.Close() - t := time.Now() log.Debugf("%s <-> %s", remoteAddr, host) + var err error + var res *http.Response + defer func() { + d := time.Since(start) log.WithFields(map[string]any{ - "duration": time.Since(t), + "duration": d, }).Debugf("%s >-< %s", remoteAddr, host) + + ro.Duration = d + if err != nil { + ro.Err = err.Error() + } + if res != nil && ro.HTTP != nil { + ro.HTTP.ResponseHeader = res.Header + ro.HTTP.StatusCode = res.StatusCode + } + ro.Record(ctx, ep.recorder.Recorder) }() - res, err := http.ReadResponse(bufio.NewReader(c), req) + res, err = http.ReadResponse(bufio.NewReader(c), req) if err != nil { log.Errorf("read response: %v", err) resp.Write(conn) return } + defer res.Body.Close() if log.IsLevelEnabled(logger.TraceLevel) { dump, _ := httputil.DumpResponse(res, false) diff --git a/handler/tunnel/handler.go b/handler/tunnel/handler.go index 18dad0e3..5ca3eb30 100644 --- a/handler/tunnel/handler.go +++ b/handler/tunnel/handler.go @@ -13,13 +13,13 @@ import ( "github.com/go-gost/core/listener" "github.com/go-gost/core/logger" md "github.com/go-gost/core/metadata" - "github.com/go-gost/core/recorder" "github.com/go-gost/core/service" "github.com/go-gost/relay" ctxvalue "github.com/go-gost/x/ctx" xnet "github.com/go-gost/x/internal/net" limiter_util "github.com/go-gost/x/internal/util/limiter" stats_util "github.com/go-gost/x/internal/util/stats" + rate_limiter "github.com/go-gost/x/limiter/rate" xrecorder "github.com/go-gost/x/recorder" "github.com/go-gost/x/registry" xservice "github.com/go-gost/x/service" @@ -32,7 +32,6 @@ var ( ErrTunnelID = errors.New("invalid tunnel ID") ErrTunnelNotAvailable = errors.New("tunnel not available") ErrUnauthorized = errors.New("unauthorized") - ErrRateLimit = errors.New("rate limiting exceeded") ) func init() { @@ -40,17 +39,16 @@ func init() { } type tunnelHandler struct { - id string - options handler.Options - pool *ConnectorPool - recorder recorder.Recorder - epSvc service.Service - ep *entrypoint - md metadata - log logger.Logger - stats *stats_util.HandlerStats - limiter traffic.TrafficLimiter - cancel context.CancelFunc + id string + options handler.Options + pool *ConnectorPool + epSvc service.Service + ep *entrypoint + md metadata + log logger.Logger + stats *stats_util.HandlerStats + limiter traffic.TrafficLimiter + cancel context.CancelFunc } func NewHandler(opts ...handler.Option) handler.Handler { @@ -79,19 +77,11 @@ func (h *tunnelHandler) Init(md md.Metadata) (err error) { "node": h.id, }) - if opts := h.options.Router.Options(); opts != nil { - for _, ro := range opts.Recorders { - if ro.Record == xrecorder.RecorderServiceHandlerTunnel { - h.recorder = ro.Recorder - break - } - } - } - h.pool = NewConnectorPool(h.id, h.md.sd) h.ep = &entrypoint{ node: h.id, + service: h.options.Service, pool: h.pool, ingress: h.md.ingress, sd: h.md.sd, @@ -103,6 +93,13 @@ func (h *tunnelHandler) Init(md md.Metadata) (err error) { return } + for _, ro := range h.options.Recorders { + if ro.Record == xrecorder.RecorderServiceHandler { + h.ep.recorder = ro + break + } + } + ctx, cancel := context.WithCancel(context.Background()) h.cancel = cancel @@ -174,6 +171,7 @@ func (h *tunnelHandler) Handle(ctx context.Context, conn net.Conn, opts ...handl log := h.log.WithFields(map[string]any{ "remote": conn.RemoteAddr().String(), "local": conn.LocalAddr().String(), + "sid": ctxvalue.SidFromContext(ctx), }) log.Infof("%s <> %s", conn.RemoteAddr(), conn.LocalAddr()) @@ -188,7 +186,7 @@ func (h *tunnelHandler) Handle(ctx context.Context, conn net.Conn, opts ...handl }() if !h.checkRateLimit(conn.RemoteAddr()) { - return ErrRateLimit + return rate_limiter.ErrRateLimit } if h.md.readTimeout > 0 { diff --git a/handler/unix/handler.go b/handler/unix/handler.go index f8194935..1801990c 100644 --- a/handler/unix/handler.go +++ b/handler/unix/handler.go @@ -12,7 +12,10 @@ import ( "github.com/go-gost/core/hop" "github.com/go-gost/core/logger" md "github.com/go-gost/core/metadata" + "github.com/go-gost/core/recorder" + ctxvalue "github.com/go-gost/x/ctx" xnet "github.com/go-gost/x/internal/net" + xrecorder "github.com/go-gost/x/recorder" "github.com/go-gost/x/registry" ) @@ -21,9 +24,10 @@ func init() { } type unixHandler struct { - hop hop.Hop - md metadata - options handler.Options + hop hop.Hop + md metadata + options handler.Options + recorder recorder.RecorderObject } func NewHandler(opts ...handler.Option) handler.Handler { @@ -42,6 +46,13 @@ func (h *unixHandler) Init(md md.Metadata) (err error) { return } + for _, ro := range h.options.Recorders { + if ro.Record == xrecorder.RecorderServiceHandler { + h.recorder = ro + break + } + } + return } @@ -50,20 +61,43 @@ func (h *unixHandler) Forward(hop hop.Hop) { h.hop = hop } -func (h *unixHandler) Handle(ctx context.Context, conn net.Conn, opts ...handler.HandleOption) error { +func (h *unixHandler) Handle(ctx context.Context, conn net.Conn, opts ...handler.HandleOption) (err error) { defer conn.Close() - log := h.options.Logger + start := time.Now() - log = log.WithFields(map[string]any{ + ro := &xrecorder.HandlerRecorderObject{ + Service: h.options.Service, + Network: "unix", + RemoteAddr: conn.RemoteAddr().String(), + LocalAddr: conn.LocalAddr().String(), + Time: start, + SID: string(ctxvalue.SidFromContext(ctx)), + } + + log := h.options.Logger.WithFields(map[string]any{ "remote": conn.RemoteAddr().String(), "local": conn.LocalAddr().String(), + "sid": ctxvalue.SidFromContext(ctx), }) + log.Infof("%s <> %s", conn.RemoteAddr(), conn.LocalAddr()) + defer func() { + if err != nil { + ro.Err = err.Error() + } + ro.Duration = time.Since(start) + ro.Record(ctx, h.recorder.Recorder) + + log.WithFields(map[string]any{ + "duration": time.Since(start), + }).Infof("%s >< %s", conn.RemoteAddr(), conn.LocalAddr()) + }() + if h.hop != nil { target := h.hop.Select(ctx) if target == nil { - err := errors.New("target not available") + err = errors.New("target not available") log.Error(err) return err } @@ -71,6 +105,8 @@ func (h *unixHandler) Handle(ctx context.Context, conn net.Conn, opts ...handler "node": target.Name, "dst": target.Addr, }) + ro.Host = target.Addr + return h.forwardUnix(ctx, conn, target, log) } diff --git a/internal/net/dialer/dialer.go b/internal/net/dialer/dialer.go index f1b10a8a..9f270f7b 100644 --- a/internal/net/dialer/dialer.go +++ b/internal/net/dialer/dialer.go @@ -10,6 +10,7 @@ import ( "time" "github.com/go-gost/core/logger" + ctxvalue "github.com/go-gost/x/ctx" xnet "github.com/go-gost/x/internal/net" "github.com/vishvananda/netns" ) @@ -39,6 +40,9 @@ func (d *Dialer) Dial(ctx context.Context, network, addr string) (conn net.Conn, if log == nil { log = logger.Default() } + log = log.WithFields(map[string]any{ + "sid": ctxvalue.SidFromContext(ctx), + }) if d.Netns != "" { runtime.LockOSThread() diff --git a/internal/net/resovle.go b/internal/net/resovle.go index fb5b8eb5..b7192241 100644 --- a/internal/net/resovle.go +++ b/internal/net/resovle.go @@ -8,6 +8,7 @@ import ( "github.com/go-gost/core/hosts" "github.com/go-gost/core/logger" "github.com/go-gost/core/resolver" + ctxvalue "github.com/go-gost/x/ctx" ) func Resolve(ctx context.Context, network, addr string, r resolver.Resolver, hosts hosts.HostMapper, log logger.Logger) (string, error) { @@ -20,6 +21,13 @@ func Resolve(ctx context.Context, network, addr string, r resolver.Resolver, hos return addr, nil } + if log == nil { + log = logger.Default() + } + log = log.WithFields(map[string]any{ + "sid": ctxvalue.SidFromContext(ctx), + }) + if hosts != nil { if ips, _ := hosts.Lookup(ctx, network, host); len(ips) > 0 { log.Debugf("hit host mapper: %s -> %s", host, ips) diff --git a/internal/util/resolver/cache.go b/internal/util/resolver/cache.go index 17793122..95352562 100644 --- a/internal/util/resolver/cache.go +++ b/internal/util/resolver/cache.go @@ -1,11 +1,13 @@ package resolver import ( + "context" "fmt" "sync" "time" "github.com/go-gost/core/logger" + ctxvalue "github.com/go-gost/x/ctx" "github.com/miekg/dns" ) @@ -44,7 +46,7 @@ func (c *Cache) WithLogger(logger logger.Logger) *Cache { return c } -func (c *Cache) Load(key CacheKey) (msg *dns.Msg, ttl time.Duration) { +func (c *Cache) Load(ctx context.Context, key CacheKey) (msg *dns.Msg, ttl time.Duration) { v, ok := c.m.Load(key) if !ok { return @@ -66,12 +68,19 @@ func (c *Cache) Load(key CacheKey) (msg *dns.Msg, ttl time.Duration) { } ttl = item.ttl - time.Since(item.ts) - c.logger.Debugf("resolver cache hit: %s, ttl: %v", key, ttl) + if log := c.logger; log.IsLevelEnabled(logger.DebugLevel) { + if sid := ctxvalue.SidFromContext(ctx); sid != "" { + log = log.WithFields(map[string]any{ + "sid": sid, + }) + } + log.Debugf("resolver cache hit: %s, ttl: %v", key, ttl) + } return } -func (c *Cache) Store(key CacheKey, mr *dns.Msg, ttl time.Duration) { +func (c *Cache) Store(ctx context.Context, key CacheKey, mr *dns.Msg, ttl time.Duration) { if key == "" || mr == nil || ttl < 0 { return } @@ -98,7 +107,14 @@ func (c *Cache) Store(key CacheKey, mr *dns.Msg, ttl time.Duration) { ttl: ttl, }) - c.logger.Debugf("resolver cache store: %s, ttl: %v", key, ttl) + if log := c.logger; log.IsLevelEnabled(logger.DebugLevel) { + if sid := ctxvalue.SidFromContext(ctx); sid != "" { + log = log.WithFields(map[string]any{ + "sid": sid, + }) + } + log.Debugf("resolver cache store: %s, ttl: %v", key, ttl) + } } func (c *Cache) RefreshTTL(key CacheKey) { diff --git a/limiter/rate/limiter.go b/limiter/rate/limiter.go index f2ef9822..1890f5b0 100644 --- a/limiter/rate/limiter.go +++ b/limiter/rate/limiter.go @@ -1,6 +1,7 @@ package rate import ( + "errors" "sort" "time" @@ -8,6 +9,10 @@ import ( "golang.org/x/time/rate" ) +var ( + ErrRateLimit = errors.New("rate limit") +) + type rlimiter struct { limiter *rate.Limiter } diff --git a/metrics/prom.go b/metrics/prom.go index d84dbe05..2210210e 100644 --- a/metrics/prom.go +++ b/metrics/prom.go @@ -44,13 +44,13 @@ func NewMetrics() metrics.Metrics { Name: string(MetricServiceTransferInputBytesCounter), Help: "Total service input data transfer size in bytes", }, - []string{"host", "service"}), + []string{"host", "service", "client"}), MetricServiceTransferOutputBytesCounter: prometheus.NewCounterVec( prometheus.CounterOpts{ Name: string(MetricServiceTransferOutputBytesCounter), Help: "Total service output data transfer size in bytes", }, - []string{"host", "service"}), + []string{"host", "service", "client"}), MetricServiceHandlerErrorsCounter: prometheus.NewCounterVec( prometheus.CounterOpts{ Name: string(MetricServiceHandlerErrorsCounter), diff --git a/metrics/wrapper/conn.go b/metrics/wrapper/conn.go index f17674aa..d159523a 100644 --- a/metrics/wrapper/conn.go +++ b/metrics/wrapper/conn.go @@ -20,16 +20,21 @@ var ( // serverConn is a server side Conn with metrics supported. type serverConn struct { net.Conn - service string + service string + clientIP string } func WrapConn(service string, c net.Conn) net.Conn { - if !xmetrics.IsEnabled() { + if !xmetrics.IsEnabled() || c == nil { return c } + + host, _, _ := net.SplitHostPort(c.RemoteAddr().String()) + return &serverConn{ - service: service, - Conn: c, + service: service, + Conn: c, + clientIP: host, } } @@ -39,6 +44,7 @@ func (c *serverConn) Read(b []byte) (n int, err error) { xmetrics.MetricServiceTransferInputBytesCounter, metrics.Labels{ "service": c.service, + "client": c.clientIP, }); counter != nil { counter.Add(float64(n)) } @@ -51,6 +57,7 @@ func (c *serverConn) Write(b []byte) (n int, err error) { xmetrics.MetricServiceTransferOutputBytesCounter, metrics.Labels{ "service": c.service, + "client": c.clientIP, }); counter != nil { counter.Add(float64(n)) } @@ -90,10 +97,17 @@ func WrapPacketConn(service string, pc net.PacketConn) net.PacketConn { func (c *packetConn) ReadFrom(p []byte) (n int, addr net.Addr, err error) { n, addr, err = c.PacketConn.ReadFrom(p) + + var clientIP string + if addr != nil { + clientIP, _, _ = net.SplitHostPort(addr.String()) + } + if counter := xmetrics.GetCounter( xmetrics.MetricServiceTransferInputBytesCounter, metrics.Labels{ "service": c.service, + "client": clientIP, }); counter != nil { counter.Add(float64(n)) } @@ -102,10 +116,17 @@ func (c *packetConn) ReadFrom(p []byte) (n int, addr net.Addr, err error) { func (c *packetConn) WriteTo(p []byte, addr net.Addr) (n int, err error) { n, err = c.PacketConn.WriteTo(p, addr) + + var clientIP string + if addr != nil { + clientIP, _, _ = net.SplitHostPort(addr.String()) + } + if counter := xmetrics.GetCounter( xmetrics.MetricServiceTransferOutputBytesCounter, metrics.Labels{ "service": c.service, + "client": clientIP, }); counter != nil { counter.Add(float64(n)) } @@ -155,10 +176,17 @@ func (c *udpConn) SetWriteBuffer(n int) error { func (c *udpConn) Read(b []byte) (n int, err error) { if nc, ok := c.PacketConn.(io.Reader); ok { n, err = nc.Read(b) + + var clientIP string + if addr := c.RemoteAddr(); addr != nil { + clientIP, _, _ = net.SplitHostPort(addr.String()) + } + if counter := xmetrics.GetCounter( xmetrics.MetricServiceTransferInputBytesCounter, metrics.Labels{ "service": c.service, + "client": clientIP, }); counter != nil { counter.Add(float64(n)) } @@ -170,10 +198,17 @@ func (c *udpConn) Read(b []byte) (n int, err error) { func (c *udpConn) ReadFrom(p []byte) (n int, addr net.Addr, err error) { n, addr, err = c.PacketConn.ReadFrom(p) + + var clientIP string + if addr != nil { + clientIP, _, _ = net.SplitHostPort(addr.String()) + } + if counter := xmetrics.GetCounter( xmetrics.MetricServiceTransferInputBytesCounter, metrics.Labels{ "service": c.service, + "client": clientIP, }); counter != nil { counter.Add(float64(n)) } @@ -183,10 +218,17 @@ func (c *udpConn) ReadFrom(p []byte) (n int, addr net.Addr, err error) { func (c *udpConn) ReadFromUDP(b []byte) (n int, addr *net.UDPAddr, err error) { if nc, ok := c.PacketConn.(udp.ReadUDP); ok { n, addr, err = nc.ReadFromUDP(b) + + var clientIP string + if addr != nil { + clientIP, _, _ = net.SplitHostPort(addr.String()) + } + if counter := xmetrics.GetCounter( xmetrics.MetricServiceTransferInputBytesCounter, metrics.Labels{ "service": c.service, + "client": clientIP, }); counter != nil { counter.Add(float64(n)) } @@ -199,10 +241,17 @@ func (c *udpConn) ReadFromUDP(b []byte) (n int, addr *net.UDPAddr, err error) { func (c *udpConn) ReadMsgUDP(b, oob []byte) (n, oobn, flags int, addr *net.UDPAddr, err error) { if nc, ok := c.PacketConn.(udp.ReadUDP); ok { n, oobn, flags, addr, err = nc.ReadMsgUDP(b, oob) + + var clientIP string + if addr != nil { + clientIP, _, _ = net.SplitHostPort(addr.String()) + } + if counter := xmetrics.GetCounter( xmetrics.MetricServiceTransferInputBytesCounter, metrics.Labels{ "service": c.service, + "client": clientIP, }); counter != nil { counter.Add(float64(n)) } @@ -215,10 +264,17 @@ func (c *udpConn) ReadMsgUDP(b, oob []byte) (n, oobn, flags int, addr *net.UDPAd func (c *udpConn) Write(b []byte) (n int, err error) { if nc, ok := c.PacketConn.(io.Writer); ok { n, err = nc.Write(b) + + var clientIP string + if addr := c.RemoteAddr(); addr != nil { + clientIP, _, _ = net.SplitHostPort(addr.String()) + } + if counter := xmetrics.GetCounter( xmetrics.MetricServiceTransferOutputBytesCounter, metrics.Labels{ "service": c.service, + "client": clientIP, }); counter != nil { counter.Add(float64(n)) } @@ -230,10 +286,17 @@ func (c *udpConn) Write(b []byte) (n int, err error) { func (c *udpConn) WriteTo(p []byte, addr net.Addr) (n int, err error) { n, err = c.PacketConn.WriteTo(p, addr) + + var clientIP string + if addr != nil { + clientIP, _, _ = net.SplitHostPort(addr.String()) + } + if counter := xmetrics.GetCounter( xmetrics.MetricServiceTransferOutputBytesCounter, metrics.Labels{ "service": c.service, + "client": clientIP, }); counter != nil { counter.Add(float64(n)) } @@ -243,10 +306,17 @@ func (c *udpConn) WriteTo(p []byte, addr net.Addr) (n int, err error) { func (c *udpConn) WriteToUDP(b []byte, addr *net.UDPAddr) (n int, err error) { if nc, ok := c.PacketConn.(udp.WriteUDP); ok { n, err = nc.WriteToUDP(b, addr) + + var clientIP string + if addr != nil { + clientIP, _, _ = net.SplitHostPort(addr.String()) + } + if counter := xmetrics.GetCounter( xmetrics.MetricServiceTransferOutputBytesCounter, metrics.Labels{ "service": c.service, + "client": clientIP, }); counter != nil { counter.Add(float64(n)) } @@ -259,10 +329,17 @@ func (c *udpConn) WriteToUDP(b []byte, addr *net.UDPAddr) (n int, err error) { func (c *udpConn) WriteMsgUDP(b, oob []byte, addr *net.UDPAddr) (n, oobn int, err error) { if nc, ok := c.PacketConn.(udp.WriteUDP); ok { n, oobn, err = nc.WriteMsgUDP(b, oob, addr) + + var clientIP string + if addr != nil { + clientIP, _, _ = net.SplitHostPort(addr.String()) + } + if counter := xmetrics.GetCounter( xmetrics.MetricServiceTransferOutputBytesCounter, metrics.Labels{ "service": c.service, + "client": clientIP, }); counter != nil { counter.Add(float64(n)) } diff --git a/recorder/http.go b/recorder/http.go index a255cc25..98c1c593 100644 --- a/recorder/http.go +++ b/recorder/http.go @@ -5,6 +5,7 @@ import ( "context" "fmt" "net/http" + "strings" "time" "github.com/go-gost/core/recorder" @@ -12,6 +13,7 @@ import ( type httpRecorderOptions struct { timeout time.Duration + header http.Header } type HTTPRecorderOption func(opts *httpRecorderOptions) @@ -22,9 +24,16 @@ func TimeoutHTTPRecorderOption(timeout time.Duration) HTTPRecorderOption { } } +func HeaderHTTPRecorderOption(header http.Header) HTTPRecorderOption { + return func(opts *httpRecorderOptions) { + opts.header = header + } +} + type httpRecorder struct { url string httpClient *http.Client + header http.Header } // HTTPRecorder records data to HTTP service. @@ -34,11 +43,19 @@ func HTTPRecorder(url string, opts ...HTTPRecorderOption) recorder.Recorder { opt(&options) } + if url == "" { + return nil + } + if !strings.HasPrefix(url, "http") { + url = "http://" + url + } + return &httpRecorder{ url: url, httpClient: &http.Client{ Timeout: options.timeout, }, + header: options.header, } } @@ -48,14 +65,22 @@ func (r *httpRecorder) Record(ctx context.Context, b []byte, opts ...recorder.Re return err } + if r.header != nil { + req.Header = r.header + } + + if req.Header.Get("Content-Type") == "" { + req.Header.Set("Content-Type", "application/json") + } + resp, err := r.httpClient.Do(req) if err != nil { return err } defer resp.Body.Close() - if resp.StatusCode != http.StatusOK { - return fmt.Errorf("%d %s", resp.StatusCode, resp.Status) + if resp.StatusCode >= http.StatusBadRequest { + return fmt.Errorf(resp.Status) } return nil diff --git a/recorder/recorder.go b/recorder/recorder.go index 303b915d..57cab9d1 100644 --- a/recorder/recorder.go +++ b/recorder/recorder.go @@ -1,6 +1,68 @@ package recorder +import ( + "context" + "encoding/json" + "net/http" + "time" + + "github.com/go-gost/core/recorder" +) + const ( + RecorderServiceHandler = "recorder.service.handler" RecorderServiceHandlerSerial = "recorder.service.handler.serial" RecorderServiceHandlerTunnel = "recorder.service.handler.tunnel" ) + +type HTTPRecorderObject struct { + Host string `json:"host"` + Method string `json:"method"` + Proto string `json:"proto"` + Scheme string `json:"scheme"` + URI string `json:"uri"` + StatusCode int `json:"statusCode"` + RequestHeader http.Header `json:"requestHeader"` + ResponseHeader http.Header `json:"responseHeader"` + // RequestBody string + // ResponseBody string +} + +type DNSRecorderObject struct { + ID int `json:"id"` + Name string `json:"name"` + Class string `json:"class"` + Type string `json:"type"` + Question string `json:"question"` + Answer string `json:"answer"` + Cached bool `json:"cached"` +} + +type HandlerRecorderObject struct { + Node string `json:"node,omitempty"` + Service string `json:"service"` + Network string `json:"network"` + RemoteAddr string `json:"remote"` + LocalAddr string `json:"local"` + Host string `json:"host"` + Client string `json:"client,omitempty"` + HTTP *HTTPRecorderObject `json:"http,omitempty"` + DNS *DNSRecorderObject `json:"dns,omitempty"` + Err string `json:"err,omitempty"` + Duration time.Duration `json:"duration"` + Time time.Time `json:"time"` + SID string `json:"sid"` +} + +func (p *HandlerRecorderObject) Record(ctx context.Context, r recorder.Recorder) error { + if p == nil || r == nil { + return nil + } + + data, err := json.Marshal(p) + if err != nil { + return err + } + + return r.Record(ctx, data) +} diff --git a/recorder/tcp.go b/recorder/tcp.go index 1c318c72..039ba20b 100644 --- a/recorder/tcp.go +++ b/recorder/tcp.go @@ -5,11 +5,13 @@ import ( "net" "time" + "github.com/go-gost/core/logger" "github.com/go-gost/core/recorder" ) type tcpRecorderOptions struct { timeout time.Duration + log logger.Logger } type TCPRecorderOption func(opts *tcpRecorderOptions) @@ -20,9 +22,16 @@ func TimeoutTCPRecorderOption(timeout time.Duration) TCPRecorderOption { } } +func LogTCPRecorderOption(log logger.Logger) TCPRecorderOption { + return func(opts *tcpRecorderOptions) { + opts.log = log + } +} + type tcpRecorder struct { addr string dialer *net.Dialer + log logger.Logger } // TCPRecorder records data to TCP service. @@ -37,6 +46,7 @@ func TCPRecorder(addr string, opts ...TCPRecorderOption) recorder.Recorder { dialer: &net.Dialer{ Timeout: options.timeout, }, + log: options.log, } } diff --git a/resolver/resolver.go b/resolver/resolver.go index 5e2b6584..a6b84460 100644 --- a/resolver/resolver.go +++ b/resolver/resolver.go @@ -166,11 +166,11 @@ func (r *localResolver) resolveAsync(ctx context.Context, server *NameServer, ho return } -func (r *localResolver) lookupCache(_ context.Context, server *NameServer, host string) (ips []net.IP, ttl time.Duration, ok bool) { +func (r *localResolver) lookupCache(ctx context.Context, server *NameServer, host string) (ips []net.IP, ttl time.Duration, ok bool) { lookup := func(t uint16, host string) (ips []net.IP, ttl time.Duration, ok bool) { mq := dns.Msg{} mq.SetQuestion(dns.Fqdn(host), t) - mr, ttl := r.cache.Load(resolver_util.NewCacheKey(&mq.Question[0])) + mr, ttl := r.cache.Load(ctx, resolver_util.NewCacheKey(&mq.Question[0])) if mr == nil { return } @@ -223,14 +223,14 @@ func (r *localResolver) resolveIPs(ctx context.Context, server *NameServer, mq * } key := resolver_util.NewCacheKey(&mq.Question[0]) - mr, ttl := r.cache.Load(key) + mr, ttl := r.cache.Load(ctx, key) if ttl <= 0 { resolver_util.AddSubnetOpt(mq, server.ClientIP) mr, err = r.exchange(ctx, server.exchanger, mq) if err != nil { return } - r.cache.Store(key, mr, server.TTL) + r.cache.Store(ctx, key, mr, server.TTL) if r.options.logger.IsLevelEnabled(logger.TraceLevel) { r.options.logger.Trace(mr.String())