From a52cd9a4c2de1c9a8092e8812ae531006f208b18 Mon Sep 17 00:00:00 2001 From: ginuerzh Date: Thu, 16 Dec 2021 19:03:56 +0800 Subject: [PATCH] add kcp dialer --- cmd/gost/config.go | 3 +- cmd/gost/register.go | 1 + go.mod | 31 ++-- go.sum | 1 - pkg/chain/chain.go | 6 +- pkg/chain/transport.go | 9 +- pkg/{listener => common/util}/kcp/config.go | 6 +- pkg/common/util/kcp/kcp.go | 2 +- pkg/common/util/ss/ss.go | 3 +- pkg/connector/relay/conn.go | 3 +- pkg/dialer/dialer.go | 2 +- pkg/dialer/kcp/conn.go | 56 ++++++ pkg/dialer/kcp/dialer.go | 183 ++++++++++++++++++++ pkg/dialer/kcp/metadata.go | 46 +++++ pkg/dialer/option.go | 12 ++ pkg/dialer/tcp/dialer.go | 23 --- pkg/dialer/tcp/metadata.go | 10 +- pkg/listener/kcp/listener.go | 32 ++-- pkg/listener/kcp/metadata.go | 44 ++++- 19 files changed, 395 insertions(+), 78 deletions(-) rename pkg/{listener => common/util}/kcp/config.go (95%) create mode 100644 pkg/dialer/kcp/conn.go create mode 100644 pkg/dialer/kcp/dialer.go create mode 100644 pkg/dialer/kcp/metadata.go diff --git a/cmd/gost/config.go b/cmd/gost/config.go index 1ea1591..b024d02 100644 --- a/cmd/gost/config.go +++ b/cmd/gost/config.go @@ -153,7 +153,8 @@ func chainFromConfig(cfg *config.ChainConfig) *chain.Chain { tr := (&chain.Transport{}). WithConnector(cr). - WithDialer(d) + WithDialer(d). + WithAddr(v.Addr) node := chain.NewNode(v.Name, v.Addr). WithTransport(tr). diff --git a/cmd/gost/register.go b/cmd/gost/register.go index b1a531e..5068cfd 100644 --- a/cmd/gost/register.go +++ b/cmd/gost/register.go @@ -16,6 +16,7 @@ import ( _ "github.com/go-gost/gost/pkg/dialer/ftcp" _ "github.com/go-gost/gost/pkg/dialer/http2" _ "github.com/go-gost/gost/pkg/dialer/http2/h2" + _ "github.com/go-gost/gost/pkg/dialer/kcp" _ "github.com/go-gost/gost/pkg/dialer/tcp" _ "github.com/go-gost/gost/pkg/dialer/udp" diff --git a/go.mod b/go.mod index 1b838cb..1e73590 100644 --- a/go.mod +++ b/go.mod @@ -5,36 +5,23 @@ go 1.17 require ( github.com/aead/chacha20 v0.0.0-20180709150244-8b13a72661da // indirect github.com/asaskevich/govalidator v0.0.0-20210307081110-f21760c49a8d + github.com/cheekybits/genny v1.0.0 // indirect github.com/coreos/go-iptables v0.5.0 // indirect + github.com/fsnotify/fsnotify v1.5.1 // indirect github.com/ginuerzh/tls-dissector v0.0.2-0.20201202075250-98fa925912da github.com/go-gost/gosocks4 v0.0.1 github.com/go-gost/gosocks5 v0.3.1-0.20211109033403-d894d75b7f09 github.com/go-gost/relay v0.1.1-0.20211123134818-8ef7fd81ffd7 github.com/go-gost/tls-dissector v0.0.2-0.20211125135007-2b5d5bd9c07e + github.com/go-task/slim-sprig v0.0.0-20210107165309-348f09dbbbc0 // indirect github.com/gobwas/glob v0.2.3 github.com/golang/snappy v0.0.3 github.com/google/gopacket v1.1.19 // indirect github.com/gorilla/websocket v1.4.2 - github.com/lucas-clemente/quic-go v0.24.0 - github.com/shadowsocks/go-shadowsocks2 v0.1.4 - github.com/shadowsocks/shadowsocks-go v0.0.0-20200409064450-3e585ff90601 - github.com/sirupsen/logrus v1.8.1 - github.com/spf13/viper v1.9.0 - github.com/xtaci/kcp-go/v5 v5.6.1 - github.com/xtaci/smux v1.5.15 - github.com/xtaci/tcpraw v1.2.25 - golang.org/x/crypto v0.0.0-20210817164053-32db794688a5 - golang.org/x/net v0.0.0-20211209124913-491a49abca63 - gopkg.in/yaml.v2 v2.4.0 -) - -require ( - github.com/cheekybits/genny v1.0.0 // indirect - github.com/fsnotify/fsnotify v1.5.1 // indirect - github.com/go-task/slim-sprig v0.0.0-20210107165309-348f09dbbbc0 // indirect github.com/hashicorp/hcl v1.0.0 // indirect github.com/klauspost/cpuid v1.3.1 // indirect github.com/klauspost/reedsolomon v1.9.9 // indirect + github.com/lucas-clemente/quic-go v0.24.0 github.com/magiconair/properties v1.8.5 // indirect github.com/marten-seemann/qtls-go1-16 v0.1.4 // indirect github.com/marten-seemann/qtls-go1-17 v0.1.0 // indirect @@ -45,19 +32,29 @@ require ( github.com/pelletier/go-toml v1.9.4 // indirect github.com/pkg/errors v0.9.1 // indirect github.com/riobard/go-bloom v0.0.0-20200614022211-cdc8013cb5b3 // indirect + github.com/shadowsocks/go-shadowsocks2 v0.1.4 + github.com/shadowsocks/shadowsocks-go v0.0.0-20200409064450-3e585ff90601 + github.com/sirupsen/logrus v1.8.1 github.com/spf13/afero v1.6.0 // indirect github.com/spf13/cast v1.4.1 // indirect github.com/spf13/jwalterweatherman v1.1.0 // indirect github.com/spf13/pflag v1.0.5 // indirect + github.com/spf13/viper v1.9.0 github.com/subosito/gotenv v1.2.0 // indirect github.com/templexxx/cpu v0.0.7 // indirect github.com/templexxx/xorsimd v0.4.1 // indirect github.com/tjfoc/gmsm v1.3.2 // indirect + github.com/xtaci/kcp-go/v5 v5.6.1 + github.com/xtaci/smux v1.5.15 + github.com/xtaci/tcpraw v1.2.25 + golang.org/x/crypto v0.0.0-20210817164053-32db794688a5 golang.org/x/mod v0.4.2 // indirect + golang.org/x/net v0.0.0-20211209124913-491a49abca63 golang.org/x/sys v0.0.0-20210823070655-63515b42dcdf // indirect golang.org/x/text v0.3.6 // indirect golang.org/x/tools v0.1.5 // indirect golang.org/x/xerrors v0.0.0-20200804184101-5ec99f83aff1 // indirect gopkg.in/ini.v1 v1.63.2 // indirect gopkg.in/tomb.v1 v1.0.0-20141024135613-dd632973f1e7 // indirect + gopkg.in/yaml.v2 v2.4.0 ) diff --git a/go.sum b/go.sum index 868264b..83b69fb 100644 --- a/go.sum +++ b/go.sum @@ -521,7 +521,6 @@ golang.org/x/net v0.0.0-20210226172049-e18ecbb05110/go.mod h1:m0MpNAwzfU5UDzcl9v golang.org/x/net v0.0.0-20210316092652-d523dce5a7f4/go.mod h1:RBQZq4jEuRlivfhVLdyRGr576XBO4/greRjx4P4O3yc= golang.org/x/net v0.0.0-20210405180319-a5a99cb37ef4/go.mod h1:p54w0d4576C0XHj96bSt6lcn1PtDYWL6XObtHCRCNQM= golang.org/x/net v0.0.0-20210428140749-89ef3d95e781/go.mod h1:OJAsFXCWl8Ukc7SiCT/9KSuxbyM7479/AVlXFRxuMCk= -golang.org/x/net v0.0.0-20210503060351-7fd8e65b6420 h1:a8jGStKg0XqKDlKqjLrXn0ioF5MH36pT7Z0BRTqLhbk= golang.org/x/net v0.0.0-20210503060351-7fd8e65b6420/go.mod h1:9nx3DQGgdP8bBQD5qxJ1jj9UTztislL4KSBs9R2vV5Y= golang.org/x/net v0.0.0-20211209124913-491a49abca63 h1:iocB37TsdFuN6IBRZ+ry36wrkoV51/tl5vOWqkcPGvY= golang.org/x/net v0.0.0-20211209124913-491a49abca63/go.mod h1:9nx3DQGgdP8bBQD5qxJ1jj9UTztislL4KSBs9R2vV5Y= diff --git a/pkg/chain/chain.go b/pkg/chain/chain.go index 29efd3f..81aa69d 100644 --- a/pkg/chain/chain.go +++ b/pkg/chain/chain.go @@ -32,8 +32,10 @@ func (c *Chain) GetRouteFor(network, address string) (r *Route) { } if node.transport.IsMultiplex() { - tr := node.transport.Copy().WithRoute(r) - node = node.Copy().WithTransport(tr) + tr := node.transport.Copy(). + WithRoute(r) + node = node.Copy(). + WithTransport(tr) r = &Route{} } diff --git a/pkg/chain/transport.go b/pkg/chain/transport.go index 220847b..496115e 100644 --- a/pkg/chain/transport.go +++ b/pkg/chain/transport.go @@ -9,6 +9,7 @@ import ( ) type Transport struct { + addr string route *Route dialer dialer.Dialer connector connector.Connector @@ -51,7 +52,8 @@ func (tr *Transport) dialOptions() []dialer.DialOption { func (tr *Transport) Handshake(ctx context.Context, conn net.Conn) (net.Conn, error) { var err error if hs, ok := tr.dialer.(dialer.Handshaker); ok { - conn, err = hs.Handshake(ctx, conn) + conn, err = hs.Handshake(ctx, conn, + dialer.AddrHandshakeOption(tr.addr)) if err != nil { return nil, err } @@ -84,3 +86,8 @@ func (tr *Transport) WithRoute(r *Route) *Transport { tr.route = r return tr } + +func (tr *Transport) WithAddr(addr string) *Transport { + tr.addr = addr + return tr +} diff --git a/pkg/listener/kcp/config.go b/pkg/common/util/kcp/config.go similarity index 95% rename from pkg/listener/kcp/config.go rename to pkg/common/util/kcp/config.go index f87aa0e..85cfb0b 100644 --- a/pkg/listener/kcp/config.go +++ b/pkg/common/util/kcp/config.go @@ -8,8 +8,8 @@ import ( ) var ( - // Salt is the default salt for KCP cipher. - Salt = "kcp-go" + // DefaultSalt is the default salt for KCP cipher. + DefaultSalt = "kcp-go" ) var ( @@ -78,7 +78,7 @@ func (c *Config) Init() { } } -func blockCrypt(key, crypt, salt string) (block kcp.BlockCrypt) { +func BlockCrypt(key, crypt, salt string) (block kcp.BlockCrypt) { pass := pbkdf2.Key([]byte(key), []byte(salt), 4096, 32, sha1.New) switch crypt { diff --git a/pkg/common/util/kcp/kcp.go b/pkg/common/util/kcp/kcp.go index 93250aa..22def39 100644 --- a/pkg/common/util/kcp/kcp.go +++ b/pkg/common/util/kcp/kcp.go @@ -12,7 +12,7 @@ type kcpCompStreamConn struct { r *snappy.Reader } -func KCPCompStreamConn(conn net.Conn) net.Conn { +func CompStreamConn(conn net.Conn) net.Conn { return &kcpCompStreamConn{ Conn: conn, w: snappy.NewBufferedWriter(conn), diff --git a/pkg/common/util/ss/ss.go b/pkg/common/util/ss/ss.go index ca9b81e..1a26bb3 100644 --- a/pkg/common/util/ss/ss.go +++ b/pkg/common/util/ss/ss.go @@ -51,7 +51,8 @@ func (c *shadowConn) Write(b []byte) (n int, err error) { n = len(b) // force byte length consistent if c.wbuf.Len() > 0 { c.wbuf.Write(b) // append the data to the cached header - _, err = c.wbuf.WriteTo(c.Conn) + _, err = c.Conn.Write(c.wbuf.Bytes()) + c.wbuf.Reset() return } _, err = c.Conn.Write(b) diff --git a/pkg/connector/relay/conn.go b/pkg/connector/relay/conn.go index c3354e5..11b29f9 100644 --- a/pkg/connector/relay/conn.go +++ b/pkg/connector/relay/conn.go @@ -34,7 +34,8 @@ func (c *tcpConn) Write(b []byte) (n int, err error) { n = len(b) // force byte length consistent if c.wbuf.Len() > 0 { c.wbuf.Write(b) // append the data to the cached header - _, err = c.wbuf.WriteTo(c.Conn) + _, err = c.Conn.Write(c.wbuf.Bytes()) + c.wbuf.Reset() return } _, err = c.Conn.Write(b) diff --git a/pkg/dialer/dialer.go b/pkg/dialer/dialer.go index 03b6204..89cc36b 100644 --- a/pkg/dialer/dialer.go +++ b/pkg/dialer/dialer.go @@ -14,7 +14,7 @@ type Dialer interface { } type Handshaker interface { - Handshake(ctx context.Context, conn net.Conn) (net.Conn, error) + Handshake(ctx context.Context, conn net.Conn, opts ...HandshakeOption) (net.Conn, error) } type Multiplexer interface { diff --git a/pkg/dialer/kcp/conn.go b/pkg/dialer/kcp/conn.go new file mode 100644 index 0000000..c46a9f4 --- /dev/null +++ b/pkg/dialer/kcp/conn.go @@ -0,0 +1,56 @@ +package kcp + +import ( + "net" + + "github.com/xtaci/smux" +) + +type muxSession struct { + conn net.Conn + session *smux.Session +} + +func (session *muxSession) GetConn() (net.Conn, error) { + return session.session.OpenStream() +} + +func (session *muxSession) Accept() (net.Conn, error) { + return session.session.AcceptStream() +} + +func (session *muxSession) Close() error { + if session.session == nil { + return nil + } + return session.session.Close() +} + +func (session *muxSession) IsClosed() bool { + if session.session == nil { + return true + } + return session.session.IsClosed() +} + +func (session *muxSession) NumStreams() int { + return session.session.NumStreams() +} + +type fakeTCPConn struct { + raddr net.Addr + net.PacketConn +} + +func (c *fakeTCPConn) Read(b []byte) (n int, err error) { + n, _, err = c.ReadFrom(b) + return +} + +func (c *fakeTCPConn) Write(b []byte) (n int, err error) { + return c.WriteTo(b, c.raddr) +} + +func (c *fakeTCPConn) RemoteAddr() net.Addr { + return c.raddr +} diff --git a/pkg/dialer/kcp/dialer.go b/pkg/dialer/kcp/dialer.go new file mode 100644 index 0000000..b0b2270 --- /dev/null +++ b/pkg/dialer/kcp/dialer.go @@ -0,0 +1,183 @@ +package kcp + +import ( + "context" + "errors" + "net" + "sync" + "time" + + kcp_util "github.com/go-gost/gost/pkg/common/util/kcp" + "github.com/go-gost/gost/pkg/dialer" + "github.com/go-gost/gost/pkg/logger" + md "github.com/go-gost/gost/pkg/metadata" + "github.com/go-gost/gost/pkg/registry" + "github.com/xtaci/kcp-go/v5" + "github.com/xtaci/smux" + "github.com/xtaci/tcpraw" +) + +func init() { + registry.RegisterDialer("kcp", NewDialer) +} + +type kcpDialer struct { + sessions map[string]*muxSession + sessionMutex sync.Mutex + logger logger.Logger + md metadata +} + +func NewDialer(opts ...dialer.Option) dialer.Dialer { + options := &dialer.Options{} + for _, opt := range opts { + opt(options) + } + + return &kcpDialer{ + sessions: make(map[string]*muxSession), + logger: options.Logger, + } +} + +func (d *kcpDialer) Init(md md.Metadata) (err error) { + if err = d.parseMetadata(md); err != nil { + return + } + + d.md.config.Init() + + return nil +} + +// IsMultiplex implements dialer.Multiplexer interface. +func (d *kcpDialer) IsMultiplex() bool { + return true +} + +func (d *kcpDialer) Dial(ctx context.Context, addr string, opts ...dialer.DialOption) (conn net.Conn, err error) { + var options dialer.DialOptions + for _, opt := range opts { + opt(&options) + } + + d.sessionMutex.Lock() + defer d.sessionMutex.Unlock() + + session, ok := d.sessions[addr] + if session != nil && session.IsClosed() { + delete(d.sessions, addr) // session is dead + ok = false + } + if !ok { + raddr, err := net.ResolveUDPAddr("udp", addr) + if err != nil { + return nil, err + } + + if d.md.config.TCP { + pc, err := tcpraw.Dial("tcp", addr) + if err != nil { + return nil, err + } + conn = &fakeTCPConn{ + raddr: raddr, + PacketConn: pc, + } + } else { + conn, err = net.ListenUDP("udp", nil) + if err != nil { + return nil, err + } + } + session = &muxSession{conn: conn} + d.sessions[addr] = session + } + + return session.conn, err +} + +// Handshake implements dialer.Handshaker +func (d *kcpDialer) Handshake(ctx context.Context, conn net.Conn, options ...dialer.HandshakeOption) (net.Conn, error) { + opts := &dialer.HandshakeOptions{} + for _, option := range options { + option(opts) + } + config := d.md.config + + d.sessionMutex.Lock() + defer d.sessionMutex.Unlock() + + if d.md.handshakeTimeout > 0 { + conn.SetDeadline(time.Now().Add(d.md.handshakeTimeout)) + defer conn.SetDeadline(time.Time{}) + } + + session, ok := d.sessions[opts.Addr] + if !ok || session.session == nil { + s, err := d.initSession(opts.Addr, conn, config) + if err != nil { + d.logger.Error(err) + conn.Close() + delete(d.sessions, opts.Addr) + return nil, err + } + session = s + d.sessions[opts.Addr] = session + } + cc, err := session.GetConn() + if err != nil { + session.Close() + delete(d.sessions, opts.Addr) + return nil, err + } + + return cc, nil +} + +func (d *kcpDialer) initSession(addr string, conn net.Conn, config *kcp_util.Config) (*muxSession, error) { + pc, ok := conn.(net.PacketConn) + if !ok { + return nil, errors.New("wrong connection type") + } + + kcpconn, err := kcp.NewConn(addr, + kcp_util.BlockCrypt(config.Key, config.Crypt, kcp_util.DefaultSalt), + config.DataShard, config.ParityShard, pc) + if err != nil { + return nil, err + } + + kcpconn.SetStreamMode(true) + kcpconn.SetWriteDelay(false) + kcpconn.SetNoDelay(config.NoDelay, config.Interval, config.Resend, config.NoCongestion) + kcpconn.SetWindowSize(config.SndWnd, config.RcvWnd) + kcpconn.SetMtu(config.MTU) + kcpconn.SetACKNoDelay(config.AckNodelay) + + if config.DSCP > 0 { + if err := kcpconn.SetDSCP(config.DSCP); err != nil { + d.logger.Warn("SetDSCP: ", err) + } + } + if err := kcpconn.SetReadBuffer(config.SockBuf); err != nil { + d.logger.Warn("SetReadBuffer: ", err) + } + if err := kcpconn.SetWriteBuffer(config.SockBuf); err != nil { + d.logger.Warn("SetWriteBuffer: ", err) + } + + // stream multiplex + smuxConfig := smux.DefaultConfig() + smuxConfig.MaxReceiveBuffer = config.SockBuf + smuxConfig.KeepAliveInterval = time.Duration(config.KeepAlive) * time.Second + var cc net.Conn = kcpconn + if !config.NoComp { + cc = kcp_util.CompStreamConn(kcpconn) + } + session, err := smux.Client(cc, smuxConfig) + if err != nil { + return nil, err + } + return &muxSession{conn: conn, session: session}, nil +} diff --git a/pkg/dialer/kcp/metadata.go b/pkg/dialer/kcp/metadata.go new file mode 100644 index 0000000..85e3d26 --- /dev/null +++ b/pkg/dialer/kcp/metadata.go @@ -0,0 +1,46 @@ +package kcp + +import ( + "encoding/json" + "time" + + kcp_util "github.com/go-gost/gost/pkg/common/util/kcp" + md "github.com/go-gost/gost/pkg/metadata" +) + +type metadata struct { + handshakeTimeout time.Duration + config *kcp_util.Config +} + +func (d *kcpDialer) parseMetadata(md md.Metadata) (err error) { + const ( + config = "config" + handshakeTimeout = "handshakeTimeout" + ) + + if mm, _ := md.Get(config).(map[interface{}]interface{}); len(mm) > 0 { + m := make(map[string]interface{}) + for k, v := range mm { + if sk, ok := k.(string); ok { + m[sk] = v + } + } + b, err := json.Marshal(m) + if err != nil { + return err + } + cfg := &kcp_util.Config{} + if err := json.Unmarshal(b, cfg); err != nil { + return err + } + d.md.config = cfg + } + + if d.md.config == nil { + d.md.config = kcp_util.DefaultConfig + } + + d.md.handshakeTimeout = md.GetDuration(handshakeTimeout) + return +} diff --git a/pkg/dialer/option.go b/pkg/dialer/option.go index 44d2fc4..ceee723 100644 --- a/pkg/dialer/option.go +++ b/pkg/dialer/option.go @@ -30,3 +30,15 @@ func DialFuncDialOption(dialf func(ctx context.Context, addr string) (net.Conn, opts.DialFunc = dialf } } + +type HandshakeOptions struct { + Addr string +} + +type HandshakeOption func(opts *HandshakeOptions) + +func AddrHandshakeOption(addr string) HandshakeOption { + return func(opts *HandshakeOptions) { + opts.Addr = addr + } +} diff --git a/pkg/dialer/tcp/dialer.go b/pkg/dialer/tcp/dialer.go index dfa27e0..9c72cd2 100644 --- a/pkg/dialer/tcp/dialer.go +++ b/pkg/dialer/tcp/dialer.go @@ -40,33 +40,10 @@ func (d *tcpDialer) Dial(ctx context.Context, addr string, opts ...dialer.DialOp opt(&options) } - dial := options.DialFunc - if dial != nil { - conn, err := dial(ctx, addr) - if err != nil { - d.logger.Error(err) - } else { - d.logger.WithFields(map[string]interface{}{ - "src": conn.LocalAddr().String(), - "dst": addr, - }).Debug("dial with dial func") - } - return conn, err - } - var netd net.Dialer conn, err := netd.DialContext(ctx, "tcp", addr) if err != nil { d.logger.Error(err) - } else { - d.logger.WithFields(map[string]interface{}{ - "src": conn.LocalAddr().String(), - "dst": addr, - }).Debug("dial direct") } return conn, err } - -func (d *tcpDialer) parseMetadata(md md.Metadata) (err error) { - return -} diff --git a/pkg/dialer/tcp/metadata.go b/pkg/dialer/tcp/metadata.go index 9afea73..20ff7bd 100644 --- a/pkg/dialer/tcp/metadata.go +++ b/pkg/dialer/tcp/metadata.go @@ -1,6 +1,10 @@ package tcp -import "time" +import ( + "time" + + md "github.com/go-gost/gost/pkg/metadata" +) const ( dialTimeout = "dialTimeout" @@ -13,3 +17,7 @@ const ( type metadata struct { dialTimeout time.Duration } + +func (d *tcpDialer) parseMetadata(md md.Metadata) (err error) { + return +} diff --git a/pkg/listener/kcp/listener.go b/pkg/listener/kcp/listener.go index 12b6b14..4c067c5 100644 --- a/pkg/listener/kcp/listener.go +++ b/pkg/listener/kcp/listener.go @@ -19,12 +19,12 @@ func init() { } type kcpListener struct { - addr string - md metadata - ln *kcp.Listener - connChan chan net.Conn - errChan chan error - logger logger.Logger + addr string + ln *kcp.Listener + cqueue chan net.Conn + errChan chan error + logger logger.Logger + md metadata } func NewListener(opts ...listener.Option) listener.Listener { @@ -44,11 +44,7 @@ func (l *kcpListener) Init(md md.Metadata) (err error) { } config := l.md.config - if config == nil { - config = DefaultConfig - } config.Init() - l.md.config = config var ln *kcp.Listener @@ -59,10 +55,10 @@ func (l *kcpListener) Init(md md.Metadata) (err error) { return } ln, err = kcp.ServeConn( - blockCrypt(config.Key, config.Crypt, Salt), config.DataShard, config.ParityShard, conn) + kcp_util.BlockCrypt(config.Key, config.Crypt, kcp_util.DefaultSalt), config.DataShard, config.ParityShard, conn) } else { ln, err = kcp.ListenWithOptions(l.addr, - blockCrypt(config.Key, config.Crypt, Salt), config.DataShard, config.ParityShard) + kcp_util.BlockCrypt(config.Key, config.Crypt, kcp_util.DefaultSalt), config.DataShard, config.ParityShard) } if err != nil { return @@ -81,7 +77,7 @@ func (l *kcpListener) Init(md md.Metadata) (err error) { } l.ln = ln - l.connChan = make(chan net.Conn, l.md.connQueueSize) + l.cqueue = make(chan net.Conn, l.md.backlog) l.errChan = make(chan error, 1) go l.listenLoop() @@ -92,7 +88,7 @@ func (l *kcpListener) Init(md md.Metadata) (err error) { func (l *kcpListener) Accept() (conn net.Conn, err error) { var ok bool select { - case conn = <-l.connChan: + case conn = <-l.cqueue: case err, ok = <-l.errChan: if !ok { err = listener.ErrClosed @@ -142,7 +138,7 @@ func (l *kcpListener) mux(conn net.Conn) { smuxConfig.KeepAliveInterval = time.Duration(l.md.config.KeepAlive) * time.Second if !l.md.config.NoComp { - conn = kcp_util.KCPCompStreamConn(conn) + conn = kcp_util.CompStreamConn(conn) } mux, err := smux.Server(conn, smuxConfig) @@ -155,17 +151,17 @@ func (l *kcpListener) mux(conn net.Conn) { for { stream, err := mux.AcceptStream() if err != nil { - l.logger.Error("accept stream:", err) + l.logger.Error("accept stream: ", err) return } select { - case l.connChan <- stream: + case l.cqueue <- stream: case <-stream.GetDieCh(): stream.Close() default: stream.Close() - l.logger.Error("connection queue is full") + l.logger.Warnf("connection queue is full, client %s discarded", stream.RemoteAddr()) } } } diff --git a/pkg/listener/kcp/metadata.go b/pkg/listener/kcp/metadata.go index 8b89741..0e8f9a6 100644 --- a/pkg/listener/kcp/metadata.go +++ b/pkg/listener/kcp/metadata.go @@ -1,23 +1,53 @@ package kcp -import md "github.com/go-gost/gost/pkg/metadata" +import ( + "encoding/json" + + kcp_util "github.com/go-gost/gost/pkg/common/util/kcp" + md "github.com/go-gost/gost/pkg/metadata" +) const ( - defaultQueueSize = 128 + defaultBacklog = 128 ) type metadata struct { - config *Config - - connQueueSize int + config *kcp_util.Config + backlog int } func (l *kcpListener) parseMetadata(md md.Metadata) (err error) { const ( - connQueueSize = "connQueueSize" + backlog = "backlog" + config = "config" ) - l.md.connQueueSize = md.GetInt(connQueueSize) + if mm, _ := md.Get(config).(map[interface{}]interface{}); len(mm) > 0 { + m := make(map[string]interface{}) + for k, v := range mm { + if sk, ok := k.(string); ok { + m[sk] = v + } + } + b, err := json.Marshal(m) + if err != nil { + return err + } + cfg := &kcp_util.Config{} + if err := json.Unmarshal(b, cfg); err != nil { + return err + } + l.md.config = cfg + } + + if l.md.config == nil { + l.md.config = kcp_util.DefaultConfig + } + + l.md.backlog = md.GetInt(backlog) + if l.md.backlog <= 0 { + l.md.backlog = defaultBacklog + } return }