go-gost/x
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

fix race condition

Browse Source
This commit is contained in:
ginuerzh
2023-10-19 23:47:47 +08:00
parent f2fd6554ad
commit cc4310106b
29 changed files with 157 additions and 156 deletions
Download Patch File Download Diff File Expand all files Collapse all files

16
handler/dns/handler.go
View File

@ -141,17 +141,17 @@ func (h *dnsHandler) Handle(ctx context.Context, conn net.Conn, opts ...handler.
b := bufpool.Get(h.md.bufferSize)
defer bufpool.Put(b)
n, err := conn.Read(*b)
n, err := conn.Read(b)
if err != nil {
log.Error(err)
return err
}
reply, err := h.request(ctx, (*b)[:n], log)
reply, err := h.request(ctx, b[:n], log)
if err != nil {
return err
}
defer bufpool.Put(&reply)
defer bufpool.Put(reply)
if _, err = conn.Write(reply); err != nil {
log.Error(err)
@ -203,14 +203,14 @@ func (h *dnsHandler) request(ctx context.Context, msg []byte, log logger.Logger)
log.Debug("bypass: ", mq.Question[0].Name)
mr = (&dns.Msg{}).SetReply(&mq)
b := bufpool.Get(h.md.bufferSize)
return mr.PackBuffer(*b)
return mr.PackBuffer(b)
}
}
mr = h.lookupHosts(ctx, &mq, log)
if mr != nil {
b := bufpool.Get(h.md.bufferSize)
return mr.PackBuffer(*b)
return mr.PackBuffer(b)
}
// only cache for single question message.
@ -222,14 +222,14 @@ func (h *dnsHandler) request(ctx context.Context, msg []byte, log logger.Logger)
if int32(ttl.Seconds()) > 0 {
log.Debugf("message %d (cached): %s", mq.Id, mq.Question[0].String())
b := bufpool.Get(h.md.bufferSize)
return mr.PackBuffer(*b)
return mr.PackBuffer(b)
}
}
}
if mr != nil && h.md.async {
b := bufpool.Get(h.md.bufferSize)
reply, err := mr.PackBuffer(*b)
reply, err := mr.PackBuffer(b)
if err != nil {
return nil, err
}
@ -248,7 +248,7 @@ func (h *dnsHandler) exchange(ctx context.Context, mq *dns.Msg) ([]byte, error)
b := bufpool.Get(h.md.bufferSize)
defer bufpool.Put(b)
query, err := mq.PackBuffer(*b)
query, err := mq.PackBuffer(b)
if err != nil {
return nil, err
}

2
handler/socks/v5/mbind.go
View File

@ -70,7 +70,7 @@ func (h *socks5Handler) muxBindLocal(ctx context.Context, conn net.Conn, network
func (h *socks5Handler) serveMuxBind(ctx context.Context, conn net.Conn, ln net.Listener, log logger.Logger) error {
// Upgrade connection to multiplex stream.
session, err := mux.ClientSession(conn, nil)
session, err := mux.ClientSession(conn, h.md.muxCfg)
if err != nil {
log.Error(err)
return err

12
handler/socks/v5/metadata.go
View File

@ -6,6 +6,7 @@ import (
mdata "github.com/go-gost/core/metadata"
mdutil "github.com/go-gost/core/metadata/util"
"github.com/go-gost/x/internal/util/mux"
)
type metadata struct {
@ -16,6 +17,7 @@ type metadata struct {
udpBufferSize int
compatibilityMode bool
hash string
muxCfg *mux.Config
}
func (h *socks5Handler) parseMetadata(md mdata.Metadata) (err error) {
@ -43,5 +45,15 @@ func (h *socks5Handler) parseMetadata(md mdata.Metadata) (err error) {
h.md.compatibilityMode = mdutil.GetBool(md, compatibilityMode)
h.md.hash = mdutil.GetString(md, hash)
h.md.muxCfg = &mux.Config{
Version: mdutil.GetInt(md, "mux.version"),
KeepAliveInterval: mdutil.GetDuration(md, "mux.keepaliveInterval"),
KeepAliveDisabled: mdutil.GetBool(md, "mux.keepaliveDisabled"),
KeepAliveTimeout: mdutil.GetDuration(md, "mux.keepaliveTimeout"),
MaxFrameSize: mdutil.GetInt(md, "mux.maxFrameSize"),
MaxReceiveBuffer: mdutil.GetInt(md, "mux.maxReceiveBuffer"),
MaxStreamBuffer: mdutil.GetInt(md, "mux.maxStreamBuffer"),
}
return nil
}

8
handler/ss/udp/handler.go
View File

@ -130,7 +130,7 @@ func (h *ssuHandler) relayPacket(pc1, pc2 net.PacketConn, log logger.Logger) (er
b := bufpool.Get(bufSize)
defer bufpool.Put(b)
n, addr, err := pc1.ReadFrom(*b)
n, addr, err := pc1.ReadFrom(b)
if err != nil {
return err
}
@ -140,7 +140,7 @@ func (h *ssuHandler) relayPacket(pc1, pc2 net.PacketConn, log logger.Logger) (er
return nil
}
if _, err = pc2.WriteTo((*b)[:n], addr); err != nil {
if _, err = pc2.WriteTo(b[:n], addr); err != nil {
return err
}
@ -162,7 +162,7 @@ func (h *ssuHandler) relayPacket(pc1, pc2 net.PacketConn, log logger.Logger) (er
b := bufpool.Get(bufSize)
defer bufpool.Put(b)
n, raddr, err := pc2.ReadFrom(*b)
n, raddr, err := pc2.ReadFrom(b)
if err != nil {
return err
}
@ -172,7 +172,7 @@ func (h *ssuHandler) relayPacket(pc1, pc2 net.PacketConn, log logger.Logger) (er
return nil
}
if _, err = pc1.WriteTo((*b)[:n], raddr); err != nil {
if _, err = pc1.WriteTo(b[:n], raddr); err != nil {
return err
}

30
handler/tap/handler.go
View File

@ -196,7 +196,7 @@ func (h *tapHandler) transport(tap net.Conn, conn net.PacketConn, raddr net.Addr
b := bufpool.Get(h.md.bufferSize)
defer bufpool.Put(b)
n, err := tap.Read(*b)
n, err := tap.Read(b)
if err != nil {
select {
case h.exit <- struct{}{}:
@ -208,22 +208,22 @@ func (h *tapHandler) transport(tap net.Conn, conn net.PacketConn, raddr net.Addr
return nil
}
src := waterutil.MACSource((*b)[:n])
dst := waterutil.MACDestination((*b)[:n])
eType := etherType(waterutil.MACEthertype((*b)[:n]))
src := waterutil.MACSource(b[:n])
dst := waterutil.MACDestination(b[:n])
eType := etherType(waterutil.MACEthertype(b[:n]))
log.Debugf("%s >> %s %s %d", src, dst, eType, n)
// client side, deliver frame directly.
if raddr != nil {
_, err := conn.WriteTo((*b)[:n], raddr)
_, err := conn.WriteTo(b[:n], raddr)
return err
}
// server side, broadcast.
if waterutil.IsBroadcast(dst) {
go h.routes.Range(func(k, v any) bool {
conn.WriteTo((*b)[:n], v.(net.Addr))
conn.WriteTo(b[:n], v.(net.Addr))
return true
})
return nil
@ -238,7 +238,7 @@ func (h *tapHandler) transport(tap net.Conn, conn net.PacketConn, raddr net.Addr
return nil
}
if _, err := conn.WriteTo((*b)[:n], addr); err != nil {
if _, err := conn.WriteTo(b[:n], addr); err != nil {
return err
}
@ -258,7 +258,7 @@ func (h *tapHandler) transport(tap net.Conn, conn net.PacketConn, raddr net.Addr
b := bufpool.Get(h.md.bufferSize)
defer bufpool.Put(b)
n, addr, err := conn.ReadFrom(*b)
n, addr, err := conn.ReadFrom(b)
if err != nil &&
err != shadowaead.ErrShortPacket {
return err
@ -267,15 +267,15 @@ func (h *tapHandler) transport(tap net.Conn, conn net.PacketConn, raddr net.Addr
return nil
}
src := waterutil.MACSource((*b)[:n])
dst := waterutil.MACDestination((*b)[:n])
eType := etherType(waterutil.MACEthertype((*b)[:n]))
src := waterutil.MACSource(b[:n])
dst := waterutil.MACDestination(b[:n])
eType := etherType(waterutil.MACEthertype(b[:n]))
log.Debugf("%s >> %s %s %d", src, dst, eType, n)
// client side, deliver frame to tap device.
if raddr != nil {
_, err := tap.Write((*b)[:n])
_, err := tap.Write(b[:n])
return err
}
@ -294,7 +294,7 @@ func (h *tapHandler) transport(tap net.Conn, conn net.PacketConn, raddr net.Addr
if waterutil.IsBroadcast(dst) {
go h.routes.Range(func(k, v any) bool {
if k.(tapRouteKey) != rkey {
conn.WriteTo((*b)[:n], v.(net.Addr))
conn.WriteTo(b[:n], v.(net.Addr))
}
return true
})
@ -302,11 +302,11 @@ func (h *tapHandler) transport(tap net.Conn, conn net.PacketConn, raddr net.Addr
if v, ok := h.routes.Load(hwAddrToTapRouteKey(dst)); ok {
log.Debugf("find route: %s -> %s", dst, v)
_, err := conn.WriteTo((*b)[:n], v.(net.Addr))
_, err := conn.WriteTo(b[:n], v.(net.Addr))
return err
}
if _, err := tap.Write((*b)[:n]); err != nil {
if _, err := tap.Write(b[:n]); err != nil {
select {
case h.exit <- struct{}{}:
default:

42
handler/tun/client.go
View File

@ -62,14 +62,14 @@ func (h *tunHandler) keepAlive(ctx context.Context, conn net.Conn, ips []net.IP)
keepAliveData := bufpool.Get(keepAliveHeaderLength + len(ips)*net.IPv6len)
defer bufpool.Put(keepAliveData)
copy((*keepAliveData)[:4], magicHeader) // magic header
copy((*keepAliveData)[4:20], []byte(h.md.passphrase))
copy(keepAliveData[:4], magicHeader) // magic header
copy(keepAliveData[4:20], []byte(h.md.passphrase))
pos := 20
for _, ip := range ips {
copy((*keepAliveData)[pos:pos+net.IPv6len], ip.To16())
copy(keepAliveData[pos:pos+net.IPv6len], ip.To16())
pos += net.IPv6len
}
if _, err := conn.Write((*keepAliveData)); err != nil {
if _, err := conn.Write(keepAliveData); err != nil {
return
}
@ -84,7 +84,7 @@ func (h *tunHandler) keepAlive(ctx context.Context, conn net.Conn, ips []net.IP)
for {
select {
case <-ticker.C:
if _, err := conn.Write((*keepAliveData)); err != nil {
if _, err := conn.Write(keepAliveData); err != nil {
return
}
h.options.Logger.Debugf("keepalive sended")
@ -103,23 +103,23 @@ func (h *tunHandler) transportClient(tun io.ReadWriter, conn net.Conn, log logge
b := bufpool.Get(h.md.bufferSize)
defer bufpool.Put(b)
n, err := tun.Read(*b)
n, err := tun.Read(b)
if err != nil {
return ErrTun
}
if waterutil.IsIPv4((*b)[:n]) {
header, err := ipv4.ParseHeader((*b)[:n])
if waterutil.IsIPv4(b[:n]) {
header, err := ipv4.ParseHeader(b[:n])
if err != nil {
log.Warn(err)
return nil
}
log.Tracef("%s >> %s %-4s %d/%-4d %-4x %d",
header.Src, header.Dst, ipProtocol(waterutil.IPv4Protocol((*b)[:n])),
header.Src, header.Dst, ipProtocol(waterutil.IPv4Protocol(b[:n])),
header.Len, header.TotalLen, header.ID, header.Flags)
} else if waterutil.IsIPv6((*b)[:n]) {
header, err := ipv6.ParseHeader((*b)[:n])
} else if waterutil.IsIPv6(b[:n]) {
header, err := ipv6.ParseHeader(b[:n])
if err != nil {
log.Warn(err)
return nil
@ -134,7 +134,7 @@ func (h *tunHandler) transportClient(tun io.ReadWriter, conn net.Conn, log logge
return nil
}
_, err = conn.Write((*b)[:n])
_, err = conn.Write(b[:n])
return err
}()
@ -151,13 +151,13 @@ func (h *tunHandler) transportClient(tun io.ReadWriter, conn net.Conn, log logge
b := bufpool.Get(h.md.bufferSize)
defer bufpool.Put(b)
n, err := conn.Read(*b)
n, err := conn.Read(b)
if err != nil {
return err
}
if n == keepAliveHeaderLength && bytes.Equal((*b)[:4], magicHeader) {
ip := net.IP((*b)[4:20])
if n == keepAliveHeaderLength && bytes.Equal(b[:4], magicHeader) {
ip := net.IP(b[4:20])
log.Debugf("keepalive received at %v", ip)
if h.md.keepAlivePeriod > 0 {
@ -166,18 +166,18 @@ func (h *tunHandler) transportClient(tun io.ReadWriter, conn net.Conn, log logge
return nil
}
if waterutil.IsIPv4((*b)[:n]) {
header, err := ipv4.ParseHeader((*b)[:n])
if waterutil.IsIPv4(b[:n]) {
header, err := ipv4.ParseHeader(b[:n])
if err != nil {
log.Warn(err)
return nil
}
log.Tracef("%s >> %s %-4s %d/%-4d %-4x %d",
header.Src, header.Dst, ipProtocol(waterutil.IPv4Protocol((*b)[:n])),
header.Src, header.Dst, ipProtocol(waterutil.IPv4Protocol(b[:n])),
header.Len, header.TotalLen, header.ID, header.Flags)
} else if waterutil.IsIPv6((*b)[:n]) {
header, err := ipv6.ParseHeader((*b)[:n])
} else if waterutil.IsIPv6(b[:n]) {
header, err := ipv6.ParseHeader(b[:n])
if err != nil {
log.Warn(err)
return nil
@ -192,7 +192,7 @@ func (h *tunHandler) transportClient(tun io.ReadWriter, conn net.Conn, log logge
return nil
}
if _, err = tun.Write((*b)[:n]); err != nil {
if _, err = tun.Write(b[:n]); err != nil {
return ErrTun
}
return nil

36
handler/tun/server.go
View File

@ -45,7 +45,7 @@ func (h *tunHandler) transportServer(ctx context.Context, tun io.ReadWriter, con
b := bufpool.Get(h.md.bufferSize)
defer bufpool.Put(b)
n, err := tun.Read(*b)
n, err := tun.Read(b)
if err != nil {
return ErrTun
}
@ -54,8 +54,8 @@ func (h *tunHandler) transportServer(ctx context.Context, tun io.ReadWriter, con
}
var src, dst net.IP
if waterutil.IsIPv4((*b)[:n]) {
header, err := ipv4.ParseHeader((*b)[:n])
if waterutil.IsIPv4(b[:n]) {
header, err := ipv4.ParseHeader(b[:n])
if err != nil {
log.Warnf("parse ipv4 packet header: %v", err)
return nil
@ -63,10 +63,10 @@ func (h *tunHandler) transportServer(ctx context.Context, tun io.ReadWriter, con
src, dst = header.Src, header.Dst
log.Tracef("%s >> %s %-4s %d/%-4d %-4x %d",
src, dst, ipProtocol(waterutil.IPv4Protocol((*b)[:n])),
src, dst, ipProtocol(waterutil.IPv4Protocol(b[:n])),
header.Len, header.TotalLen, header.ID, header.Flags)
} else if waterutil.IsIPv6((*b)[:n]) {
header, err := ipv6.ParseHeader((*b)[:n])
} else if waterutil.IsIPv6(b[:n]) {
header, err := ipv6.ParseHeader(b[:n])
if err != nil {
log.Warnf("parse ipv6 packet header: %v", err)
return nil
@ -90,7 +90,7 @@ func (h *tunHandler) transportServer(ctx context.Context, tun io.ReadWriter, con
log.Debugf("find route: %s -> %s", dst, addr)
if _, err := conn.WriteTo((*b)[:n], addr); err != nil {
if _, err := conn.WriteTo(b[:n], addr); err != nil {
return err
}
return nil
@ -109,16 +109,16 @@ func (h *tunHandler) transportServer(ctx context.Context, tun io.ReadWriter, con
b := bufpool.Get(h.md.bufferSize)
defer bufpool.Put(b)
n, addr, err := conn.ReadFrom(*b)
n, addr, err := conn.ReadFrom(b)
if err != nil {
return err
}
if n == 0 {
return nil
}
if n > keepAliveHeaderLength && bytes.Equal((*b)[:4], magicHeader) {
if n > keepAliveHeaderLength && bytes.Equal(b[:4], magicHeader) {
var peerIPs []net.IP
data := (*b)[keepAliveHeaderLength:n]
data := b[keepAliveHeaderLength:n]
if len(data)%net.IPv6len == 0 {
for len(data) > 0 {
peerIPs = append(peerIPs, net.IP(data[:net.IPv6len]))
@ -139,7 +139,7 @@ func (h *tunHandler) transportServer(ctx context.Context, tun io.ReadWriter, con
if auther := h.options.Auther; auther != nil {
ok := true
key := bytes.TrimRight((*b)[4:20], "\x00")
key := bytes.TrimRight(b[4:20], "\x00")
for _, ip := range peerIPs {
if _, ok = auther.Authenticate(ctx, ip.String(), string(key)); !ok {
break
@ -175,8 +175,8 @@ func (h *tunHandler) transportServer(ctx context.Context, tun io.ReadWriter, con
}
var src, dst net.IP
if waterutil.IsIPv4((*b)[:n]) {
header, err := ipv4.ParseHeader((*b)[:n])
if waterutil.IsIPv4(b[:n]) {
header, err := ipv4.ParseHeader(b[:n])
if err != nil {
log.Warnf("parse ipv4 packet header: %v", err)
return nil
@ -184,10 +184,10 @@ func (h *tunHandler) transportServer(ctx context.Context, tun io.ReadWriter, con
src, dst = header.Src, header.Dst
log.Tracef("%s >> %s %-4s %d/%-4d %-4x %d",
src, dst, ipProtocol(waterutil.IPv4Protocol((*b)[:n])),
src, dst, ipProtocol(waterutil.IPv4Protocol(b[:n])),
header.Len, header.TotalLen, header.ID, header.Flags)
} else if waterutil.IsIPv6((*b)[:n]) {
header, err := ipv6.ParseHeader((*b)[:n])
} else if waterutil.IsIPv6(b[:n]) {
header, err := ipv6.ParseHeader(b[:n])
if err != nil {
log.Warnf("parse ipv6 packet header: %v", err)
return nil
@ -206,11 +206,11 @@ func (h *tunHandler) transportServer(ctx context.Context, tun io.ReadWriter, con
if addr := h.findRouteFor(dst, config.Routes...); addr != nil {
log.Debugf("find route: %s -> %s", dst, addr)
_, err := conn.WriteTo((*b)[:n], addr)
_, err := conn.WriteTo(b[:n], addr)
return err
}
if _, err := tun.Write((*b)[:n]); err != nil {
if _, err := tun.Write(b[:n]); err != nil {
return ErrTun
}
return nil

7
handler/tunnel/metadata.go
View File

@ -14,10 +14,6 @@ import (
"github.com/go-gost/x/registry"
)
const (
defaultMuxVersion = 2
)
type metadata struct {
readTimeout time.Duration
noDelay bool
@ -69,9 +65,6 @@ func (h *tunnelHandler) parseMetadata(md mdata.Metadata) (err error) {
MaxReceiveBuffer: mdutil.GetInt(md, "mux.maxReceiveBuffer"),
MaxStreamBuffer: mdutil.GetInt(md, "mux.maxStreamBuffer"),
}
if h.md.muxCfg.Version == 0 {
h.md.muxCfg.Version = defaultMuxVersion
}
return
}