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

rename vtun to tungo

Browse Source
This commit is contained in:
ginuerzh
2025-07-26 16:17:04 +08:00
parent 281295d02f
commit 35a049fb03
30 changed files with 801 additions and 307 deletions
Download Patch File Download Diff File Expand all files Collapse all files
bypass/bypass.go
+8 -2
View File
@@ -13,6 +13,7 @@ import (
"github.com/go-gost/core/bypass"
"github.com/go-gost/core/logger"
ctxvalue "github.com/go-gost/x/ctx"
"github.com/go-gost/x/internal/loader"
"github.com/go-gost/x/internal/matcher"
xnet "github.com/go-gost/x/internal/net"
@@ -252,10 +253,15 @@ func (bp *localBypass) Contains(ctx context.Context, network, addr string, opts
b := !bp.options.whitelist && matched ||
bp.options.whitelist && !matched
log := bp.options.logger.WithFields(map[string]any{
"sid": ctxvalue.SidFromContext(ctx),
})
if b {
bp.options.logger.Debugf("bypass: %s, whitelist: %t", addr, bp.options.whitelist)
log.Debugf("bypass: %s, whitelist: %t", addr, bp.options.whitelist)
} else {
bp.options.logger.Debugf("pass: %s, whitelist: %t", addr, bp.options.whitelist)
log.Debugf("pass: %s, whitelist: %t", addr, bp.options.whitelist)
}
return b
}
go.mod
+1 -1
View File
@@ -18,6 +18,7 @@ require (
github.com/go-redis/redis/v8 v8.11.5
github.com/gobwas/glob v0.2.3
github.com/golang/snappy v0.0.4
github.com/google/shlex v0.0.0-20191202100458-e7afc7fbc510
github.com/google/uuid v1.6.0
github.com/gorilla/websocket v1.5.3
github.com/miekg/dns v1.1.61
@@ -43,7 +44,6 @@ require (
github.com/xtaci/tcpraw v1.2.25
github.com/yl2chen/cidranger v1.0.2
github.com/zalando/go-keyring v0.2.4
go.uber.org/atomic v1.11.0
golang.org/x/crypto v0.40.0
golang.org/x/exp v0.0.0-20241210194714-1829a127f884
golang.org/x/net v0.42.0
go.sum
+2 -2
View File
@@ -101,6 +101,8 @@ github.com/google/go-cmp v0.6.0/go.mod h1:17dUlkBOakJ0+DkrSSNjCkIjxS6bF9zb3elmeN
github.com/google/gofuzz v1.0.0/go.mod h1:dBl0BpW6vV/+mYPU4Po3pmUjxk6FQPldtuIdl/M65Eg=
github.com/google/gopacket v1.1.19 h1:ves8RnFZPGiFnTS0uPQStjwru6uO6h+nlr9j6fL7kF8=
github.com/google/gopacket v1.1.19/go.mod h1:iJ8V8n6KS+z2U1A8pUwu8bW5SyEMkXJB8Yo/Vo+TKTo=
github.com/google/shlex v0.0.0-20191202100458-e7afc7fbc510 h1:El6M4kTTCOh6aBiKaUGG7oYTSPP8MxqL4YI3kZKwcP4=
github.com/google/shlex v0.0.0-20191202100458-e7afc7fbc510/go.mod h1:pupxD2MaaD3pAXIBCelhxNneeOaAeabZDe5s4K6zSpQ=
github.com/google/uuid v1.6.0 h1:NIvaJDMOsjHA8n1jAhLSgzrAzy1Hgr+hNrb57e+94F0=
github.com/google/uuid v1.6.0/go.mod h1:TIyPZe4MgqvfeYDBFedMoGGpEw/LqOeaOT+nhxU+yHo=
github.com/gorilla/websocket v1.5.3 h1:saDtZ6Pbx/0u+bgYQ3q96pZgCzfhKXGPqt7kZ72aNNg=
@@ -257,8 +259,6 @@ github.com/yl2chen/cidranger v1.0.2/go.mod h1:9U1yz7WPYDwf0vpNWFaeRh0bjwz5RVgRy/
github.com/yuin/goldmark v1.4.13/go.mod h1:6yULJ656Px+3vBD8DxQVa3kxgyrAnzto9xy5taEt/CY=
github.com/zalando/go-keyring v0.2.4 h1:wi2xxTqdiwMKbM6TWwi+uJCG/Tum2UV0jqaQhCa9/68=
github.com/zalando/go-keyring v0.2.4/go.mod h1:HL4k+OXQfJUWaMnqyuSOc0drfGPX2b51Du6K+MRgZMk=
go.uber.org/atomic v1.11.0 h1:ZvwS0R+56ePWxUNi+Atn9dWONBPp/AUETXlHW0DxSjE=
go.uber.org/atomic v1.11.0/go.mod h1:LUxbIzbOniOlMKjJjyPfpl4v+PKK2cNJn91OQbhoJI0=
go.uber.org/mock v0.5.0 h1:KAMbZvZPyBPWgD14IrIQ38QCyjwpvVVV6K/bHl1IwQU=
go.uber.org/mock v0.5.0/go.mod h1:ge71pBPLYDk7QIi1LupWxdAykm7KIEFchiOqd6z7qMM=
go.uber.org/multierr v1.11.0 h1:blXXJkSxSSfBVBlC76pxqeO+LN3aDfLQo+309xJstO0=
handler/vtun/endpoint.go → handler/tungo/endpoint.go
+1 -3
View File
@@ -1,6 +1,4 @@
// Package iobased provides the implementation of io.ReadWriter
// based data-link layer endpoints.
package tun
package tungo
import (
"context"
handler/tungo/handler.go
+180
View File
@@ -0,0 +1,180 @@
package tungo
import (
"context"
"errors"
"net"
"time"
"github.com/go-gost/core/handler"
"github.com/go-gost/core/limiter"
"github.com/go-gost/core/limiter/traffic"
md "github.com/go-gost/core/metadata"
"github.com/go-gost/core/observer"
"github.com/go-gost/core/recorder"
ctxvalue "github.com/go-gost/x/ctx"
stats_util "github.com/go-gost/x/internal/util/stats"
tun_util "github.com/go-gost/x/internal/util/tun"
cache_limiter "github.com/go-gost/x/limiter/traffic/cache"
xrecorder "github.com/go-gost/x/recorder"
"github.com/go-gost/x/registry"
"github.com/xjasonlyu/tun2socks/v2/core"
"github.com/xjasonlyu/tun2socks/v2/core/adapter"
)
func init() {
registry.HandlerRegistry().Register("tungo", NewHandler)
}
type tungoHandler struct {
options handler.Options
md metadata
stats *stats_util.HandlerStats
limiter traffic.TrafficLimiter
cancel context.CancelFunc
recorder recorder.RecorderObject
}
func NewHandler(opts ...handler.Option) handler.Handler {
options := handler.Options{}
for _, opt := range opts {
opt(&options)
}
return &tungoHandler{
options: options,
}
}
func (h *tungoHandler) Init(md md.Metadata) (err error) {
if err = h.parseMetadata(md); err != nil {
return
}
ctx, cancel := context.WithCancel(context.Background())
h.cancel = cancel
if h.options.Observer != nil {
h.stats = stats_util.NewHandlerStats(h.options.Service, h.md.observerResetTraffic)
go h.observeStats(ctx)
}
if h.options.Limiter != nil {
h.limiter = cache_limiter.NewCachedTrafficLimiter(h.options.Limiter,
cache_limiter.RefreshIntervalOption(h.md.limiterRefreshInterval),
cache_limiter.CleanupIntervalOption(h.md.limiterCleanupInterval),
cache_limiter.ScopeOption(limiter.ScopeClient),
)
}
for _, ro := range h.options.Recorders {
if ro.Record == xrecorder.RecorderServiceHandler {
h.recorder = ro
break
}
}
return
}
func (h *tungoHandler) Handle(ctx context.Context, conn net.Conn, opts ...handler.HandleOption) error {
defer conn.Close()
log := h.options.Logger
var config *tun_util.Config
if v, _ := conn.(md.Metadatable); v != nil {
config = v.Metadata().Get("config").(*tun_util.Config)
}
if config == nil {
err := errors.New("tun: wrong connection type")
log.Error(err)
return err
}
start := time.Now()
log = log.WithFields(map[string]any{
"remote": conn.RemoteAddr().String(),
"local": conn.LocalAddr().String(),
"sid": string(ctxvalue.SidFromContext(ctx)),
})
log.Infof("%s <> %s", conn.RemoteAddr(), conn.LocalAddr())
defer func() {
log.WithFields(map[string]any{
"duration": time.Since(start),
}).Infof("%s >< %s", conn.RemoteAddr(), conn.LocalAddr())
}()
th := &transportHandler{
tcpQueue: make(chan adapter.TCPConn),
udpQueue: make(chan adapter.UDPConn),
udpTimeout: h.md.udpTimeout,
procCancel: func() { /* nop */ },
sniffing: h.md.sniffing,
sniffingTimeout: h.md.sniffingTimeout,
sniffingResponseTimeout: h.md.sniffingResponseTimeout,
sniffingFallback: h.md.sniffingFallback,
recorder: h.recorder,
stats: h.stats,
opts: &h.options,
}
th.ProcessAsync()
defer th.Close()
stack, err := core.CreateStack(&core.Config{
LinkEndpoint: newEndpoint(conn, config.MTU, log),
TransportHandler: th,
})
if err != nil {
return err
}
defer stack.Close()
<-ctx.Done()
return nil
}
// Close implements io.Closer interface.
func (h *tungoHandler) Close() error {
if h.cancel != nil {
h.cancel()
}
return nil
}
func (h *tungoHandler) observeStats(ctx context.Context) {
if h.options.Observer == nil {
return
}
var events []observer.Event
ticker := time.NewTicker(h.md.observerPeriod)
defer ticker.Stop()
for {
select {
case <-ticker.C:
if len(events) > 0 {
if err := h.options.Observer.Observe(ctx, events); err == nil {
events = nil
}
break
}
evs := h.stats.Events()
if err := h.options.Observer.Observe(ctx, evs); err != nil {
events = evs
}
case <-ctx.Done():
return
}
}
}
handler/tungo/metadata.go
+46
View File
@@ -0,0 +1,46 @@
package tungo
import (
"time"
mdata "github.com/go-gost/core/metadata"
mdutil "github.com/go-gost/x/metadata/util"
)
type metadata struct {
udpTimeout time.Duration
sniffing bool
sniffingTimeout time.Duration
sniffingResponseTimeout time.Duration
sniffingFallback bool
observerPeriod time.Duration
observerResetTraffic bool
limiterRefreshInterval time.Duration
limiterCleanupInterval time.Duration
}
func (h *tungoHandler) parseMetadata(md mdata.Metadata) (err error) {
h.md.udpTimeout = mdutil.GetDuration(md, "udpTimeout")
h.md.sniffing = mdutil.GetBool(md, "sniffing")
h.md.sniffingTimeout = mdutil.GetDuration(md, "sniffing.timeout")
h.md.sniffingResponseTimeout = mdutil.GetDuration(md, "sniffing.responseTimeout")
h.md.sniffingFallback = mdutil.GetBool(md, "sniffing.fallback")
h.md.observerPeriod = mdutil.GetDuration(md, "observePeriod", "observer.period", "observer.observePeriod")
if h.md.observerPeriod == 0 {
h.md.observerPeriod = 5 * time.Second
}
if h.md.observerPeriod < time.Second {
h.md.observerPeriod = time.Second
}
h.md.observerResetTraffic = mdutil.GetBool(md, "observer.resetTraffic")
h.md.limiterRefreshInterval = mdutil.GetDuration(md, "limiter.refreshInterval")
h.md.limiterCleanupInterval = mdutil.GetDuration(md, "limiter.cleanupInterval")
return
}
handler/tungo/stack.go
+377
View File
@@ -0,0 +1,377 @@
package tungo
import (
"bufio"
"bytes"
"context"
"crypto/tls"
"io"
"math"
"net"
"net/netip"
"strings"
"sync"
"time"
"github.com/go-gost/core/handler"
"github.com/go-gost/core/observer/stats"
"github.com/go-gost/core/recorder"
ctxvalue "github.com/go-gost/x/ctx"
xnet "github.com/go-gost/x/internal/net"
"github.com/go-gost/x/internal/util/sniffing"
stats_util "github.com/go-gost/x/internal/util/stats"
xstats "github.com/go-gost/x/observer/stats"
stats_wrapper "github.com/go-gost/x/observer/stats/wrapper"
xrecorder "github.com/go-gost/x/recorder"
"github.com/rs/xid"
"github.com/xjasonlyu/tun2socks/v2/core/adapter"
)
const (
// udpSessionTimeout is the default timeout for UDP sessions.
udpSessionTimeout = 30 * time.Second
)
var _ adapter.TransportHandler = (*transportHandler)(nil)
type transportHandler struct {
// Unbuffered TCP/UDP queues.
tcpQueue chan adapter.TCPConn
udpQueue chan adapter.UDPConn
// UDP session timeout.
udpTimeout time.Duration
procOnce sync.Once
procCancel context.CancelFunc
sniffing bool
sniffingTimeout time.Duration
sniffingResponseTimeout time.Duration
sniffingFallback bool
stats *stats_util.HandlerStats
recorder recorder.RecorderObject
opts *handler.Options
}
func (h *transportHandler) HandleTCP(conn adapter.TCPConn) {
h.tcpQueue <- conn
}
func (h *transportHandler) HandleUDP(conn adapter.UDPConn) {
h.udpQueue <- conn
}
func (h *transportHandler) process(ctx context.Context) {
for {
select {
case conn := <-h.tcpQueue:
go h.handleTCPConn(conn)
case conn := <-h.udpQueue:
go h.handleUDPConn(conn)
case <-ctx.Done():
return
}
}
}
// ProcessAsync can be safely called multiple times, but will only be effective once.
func (h *transportHandler) ProcessAsync() {
h.procOnce.Do(func() {
ctx, cancel := context.WithCancel(context.Background())
h.procCancel = cancel
go h.process(ctx)
})
}
// Close closes the Tunnel and releases its resources.
func (h *transportHandler) Close() {
h.procCancel()
}
func (h *transportHandler) handleTCPConn(originConn adapter.TCPConn) {
defer originConn.Close()
id := originConn.ID()
remoteIP, _ := netip.AddrFromSlice(id.RemoteAddress.AsSlice())
dstIP, _ := netip.AddrFromSlice(id.LocalAddress.AsSlice())
remoteAddr := netip.AddrPortFrom(remoteIP, id.RemotePort)
dstAddr := netip.AddrPortFrom(dstIP, id.LocalPort)
start := time.Now()
sid := xid.New().String()
ctx := ctxvalue.ContextWithSid(context.Background(), ctxvalue.Sid(sid))
ro := &xrecorder.HandlerRecorderObject{
Service: h.opts.Service,
Network: "tcp",
RemoteAddr: remoteAddr.String(),
Dst: dstAddr.String(),
ClientIP: remoteAddr.String(),
Time: start,
SID: sid,
}
if h, _, _ := net.SplitHostPort(ro.ClientIP); h != "" {
ro.ClientIP = h
}
log := h.opts.Logger.WithFields(map[string]any{
"network": ro.Network,
"remote": ro.RemoteAddr,
"dst": ro.Dst,
"client": ro.ClientIP,
"sid": ro.SID,
})
log.Debugf("%s <> %s", remoteAddr.String(), dstAddr.String())
var err error
var conn net.Conn = originConn
pStats := xstats.Stats{}
conn = stats_wrapper.WrapConn(conn, &pStats)
defer func() {
if err != nil {
ro.Err = err.Error()
}
ro.InputBytes = pStats.Get(stats.KindInputBytes)
ro.OutputBytes = pStats.Get(stats.KindOutputBytes)
ro.Duration = time.Since(start)
log.WithFields(map[string]any{
"src": ro.Src,
"duration": time.Since(start),
"inputBytes": ro.InputBytes,
"outputBytes": ro.OutputBytes,
}).Infof("%s >< %s", remoteAddr.String(), dstAddr.String())
}()
if pstats := h.stats.Stats(""); pstats != nil {
pstats.Add(stats.KindTotalConns, 1)
pstats.Add(stats.KindCurrentConns, 1)
defer pstats.Add(stats.KindCurrentConns, -1)
rw := stats_wrapper.WrapReadWriter(conn, pstats)
conn = xnet.NewReadWriteConn(rw, rw, conn)
}
if h.sniffing {
if h.sniffingTimeout > 0 {
conn.SetReadDeadline(time.Now().Add(h.sniffingTimeout))
}
br := bufio.NewReader(conn)
proto, _ := sniffing.Sniff(ctx, br)
ro.Proto = proto
if h.sniffingTimeout > 0 {
conn.SetReadDeadline(time.Time{})
}
dial := func(ctx context.Context, network, address string) (net.Conn, error) {
log.Debugf("dial %s/%s", address, network)
var cc net.Conn
var err error
if address != "" {
host, _, _ := net.SplitHostPort(address)
if host == "" {
host = address
}
_, port, _ := net.SplitHostPort(dstAddr.String())
address = net.JoinHostPort(strings.Trim(host, "[]"), port)
ro.Host = address
var buf bytes.Buffer
cc, err = h.opts.Router.Dial(ctxvalue.ContextWithBuffer(ctx, &buf), "tcp", address)
ro.Route = buf.String()
if err != nil && !h.sniffingFallback {
return nil, err
}
}
if cc == nil {
var buf bytes.Buffer
cc, err = h.opts.Router.Dial(ctxvalue.ContextWithBuffer(ctx, &buf), "tcp", dstAddr.String())
ro.Route = buf.String()
ro.Host = dstAddr.String()
}
return cc, err
}
dialTLS := func(ctx context.Context, network, address string, cfg *tls.Config) (net.Conn, error) {
return dial(ctx, network, address)
}
sniffer := &sniffing.Sniffer{
Recorder: h.recorder.Recorder,
RecorderOptions: h.recorder.Options,
ReadTimeout: h.sniffingResponseTimeout,
}
conn = xnet.NewReadWriteConn(br, conn, conn)
switch proto {
case sniffing.ProtoHTTP:
sniffer.HandleHTTP(ctx, conn,
sniffing.WithDial(dial),
sniffing.WithDialTLS(dialTLS),
sniffing.WithBypass(h.opts.Bypass),
sniffing.WithRecorderObject(ro),
sniffing.WithLog(log),
)
return
case sniffing.ProtoTLS:
sniffer.HandleTLS(ctx, conn,
sniffing.WithDial(dial),
sniffing.WithDialTLS(dialTLS),
sniffing.WithBypass(h.opts.Bypass),
sniffing.WithRecorderObject(ro),
sniffing.WithLog(log),
)
return
}
}
if h.opts.Bypass != nil &&
h.opts.Bypass.Contains(ctx, "tcp", dstAddr.String()) {
log.Debug("bypass: ", dstAddr)
return
}
var buf bytes.Buffer
cc, err := h.opts.Router.Dial(ctxvalue.ContextWithBuffer(ctx, &buf), "tcp", dstAddr.String())
ro.Route = buf.String()
if err != nil {
log.Errorf("dial %s: %v", dstAddr.String(), err)
return
}
defer cc.Close()
ro.Src = cc.LocalAddr().String()
log = log.WithFields(map[string]any{"src": ro.Src})
t := time.Now()
log.Infof("%s <-> %s", remoteAddr, dstAddr)
xnet.Transport(conn, cc)
log.WithFields(map[string]any{
"duration": time.Since(t),
}).Infof("%s >-< %s", remoteAddr, dstAddr)
}
func (h *transportHandler) handleUDPConn(uc adapter.UDPConn) {
defer uc.Close()
id := uc.ID()
remoteIP, _ := netip.AddrFromSlice(id.RemoteAddress.AsSlice())
dstIP, _ := netip.AddrFromSlice(id.LocalAddress.AsSlice())
remoteAddr := netip.AddrPortFrom(remoteIP, id.RemotePort)
dstAddr := netip.AddrPortFrom(dstIP, id.LocalPort)
start := time.Now()
sid := xid.New().String()
ctx := ctxvalue.ContextWithSid(context.Background(), ctxvalue.Sid(sid))
ro := &xrecorder.HandlerRecorderObject{
Service: h.opts.Service,
Network: "udp",
RemoteAddr: remoteAddr.String(),
Dst: dstAddr.String(),
Time: start,
SID: sid,
}
log := h.opts.Logger.WithFields(map[string]any{
"network": ro.Network,
"remote": ro.RemoteAddr,
"dst": ro.Dst,
"sid": ro.SID,
})
log.Debugf("%s <> %s", remoteAddr.String(), dstAddr.String())
var err error
var conn net.Conn = uc
pStats := xstats.Stats{}
conn = stats_wrapper.WrapConn(conn, &pStats)
defer func() {
if err != nil {
ro.Err = err.Error()
}
ro.InputBytes = pStats.Get(stats.KindInputBytes)
ro.OutputBytes = pStats.Get(stats.KindOutputBytes)
ro.Duration = time.Since(start)
log.WithFields(map[string]any{
"src": ro.Src,
"duration": time.Since(start),
"inputBytes": ro.InputBytes,
"outputBytes": ro.OutputBytes,
}).Infof("%s >< %s", remoteAddr.String(), dstAddr.String())
}()
cc, err := h.opts.Router.Dial(ctx, "udp", dstAddr.String())
if err != nil {
log.Errorf("dial %s: %v", dstAddr.String(), err)
return
}
defer cc.Close()
ro.Src = cc.LocalAddr().String()
log = log.WithFields(map[string]any{"src": ro.Src})
t := time.Now()
log.Infof("%s <-> %s", remoteAddr, dstAddr)
pipeData(conn, cc, h.udpTimeout)
log.WithFields(map[string]any{
"duration": time.Since(t),
}).Infof("%s >-< %s", remoteAddr, dstAddr)
}
func pipeData(conn1, conn2 net.Conn, timeout time.Duration) {
if timeout <= 0 {
timeout = udpSessionTimeout
}
wg := sync.WaitGroup{}
wg.Add(2)
go func() {
defer wg.Done()
copyData(conn1, conn2, timeout)
}()
go func() {
defer wg.Done()
copyData(conn2, conn1, timeout)
}()
wg.Wait()
}
func copyData(dst, src net.Conn, timeout time.Duration) error {
buf := make([]byte, math.MaxUint16)
for {
src.SetReadDeadline(time.Now().Add(timeout))
n, err := src.Read(buf)
if ne, ok := err.(net.Error); ok && ne.Timeout() {
return nil /* ignore I/O timeout */
} else if err == io.EOF {
return nil /* ignore EOF */
} else if err != nil {
return err
}
if _, err = dst.Write(buf[:n]); err != nil {
return err
}
dst.SetReadDeadline(time.Now().Add(timeout))
}
}
handler/vtun/handler.go
-89
View File
@@ -1,89 +0,0 @@
package tun
import (
"context"
"errors"
"net"
"time"
"github.com/go-gost/core/handler"
md "github.com/go-gost/core/metadata"
ctxvalue "github.com/go-gost/x/ctx"
tun_util "github.com/go-gost/x/internal/util/tun"
"github.com/go-gost/x/registry"
"github.com/xjasonlyu/tun2socks/v2/core"
)
func init() {
registry.HandlerRegistry().Register("vtun", NewHandler)
}
type tunHandler struct {
options handler.Options
}
func NewHandler(opts ...handler.Option) handler.Handler {
options := handler.Options{}
for _, opt := range opts {
opt(&options)
}
return &tunHandler{
options: options,
}
}
func (h *tunHandler) Init(md md.Metadata) (err error) {
if err = h.parseMetadata(md); err != nil {
return
}
return
}
func (h *tunHandler) Handle(ctx context.Context, conn net.Conn, opts ...handler.HandleOption) error {
defer conn.Close()
log := h.options.Logger
var config *tun_util.Config
if v, _ := conn.(md.Metadatable); v != nil {
config = v.Metadata().Get("config").(*tun_util.Config)
}
if config == nil {
err := errors.New("tun: wrong connection type")
log.Error(err)
return err
}
start := time.Now()
log = log.WithFields(map[string]any{
"remote": conn.RemoteAddr().String(),
"local": conn.LocalAddr().String(),
"sid": string(ctxvalue.SidFromContext(ctx)),
})
log.Infof("%s <> %s", conn.RemoteAddr(), conn.LocalAddr())
defer func() {
log.WithFields(map[string]any{
"duration": time.Since(start),
}).Infof("%s >< %s", conn.RemoteAddr(), conn.LocalAddr())
}()
th := newTransportHandler(&h.options)
th.ProcessAsync()
defer th.Close()
stack, err := core.CreateStack(&core.Config{
LinkEndpoint: newEndpoint(conn, config.MTU, log),
TransportHandler: th,
})
if err != nil {
return err
}
defer stack.Close()
<-ctx.Done()
return nil
}
handler/vtun/metadata.go
-12
View File
@@ -1,12 +0,0 @@
package tun
import (
mdata "github.com/go-gost/core/metadata"
)
type metadata struct {
}
func (h *tunHandler) parseMetadata(md mdata.Metadata) (err error) {
return
}
handler/vtun/stack.go
-166
View File
@@ -1,166 +0,0 @@
package tun
import (
"context"
"net"
"net/netip"
"sync"
"time"
"github.com/go-gost/core/handler"
"github.com/go-gost/core/observer/stats"
ctxvalue "github.com/go-gost/x/ctx"
xnet "github.com/go-gost/x/internal/net"
xstats "github.com/go-gost/x/observer/stats"
stats_wrapper "github.com/go-gost/x/observer/stats/wrapper"
xrecorder "github.com/go-gost/x/recorder"
"github.com/rs/xid"
"github.com/xjasonlyu/tun2socks/v2/core/adapter"
"go.uber.org/atomic"
)
const (
// tcpConnectTimeout is the default timeout for TCP handshakes.
tcpConnectTimeout = 5 * time.Second
// tcpWaitTimeout implements a TCP half-close timeout.
tcpWaitTimeout = 60 * time.Second
// udpSessionTimeout is the default timeout for UDP sessions.
udpSessionTimeout = 60 * time.Second
)
var _ adapter.TransportHandler = (*transportHandler)(nil)
type transportHandler struct {
// Unbuffered TCP/UDP queues.
tcpQueue chan adapter.TCPConn
udpQueue chan adapter.UDPConn
// UDP session timeout.
udpTimeout *atomic.Duration
procOnce sync.Once
procCancel context.CancelFunc
opts *handler.Options
}
func newTransportHandler(opts *handler.Options) *transportHandler {
return &transportHandler{
tcpQueue: make(chan adapter.TCPConn),
udpQueue: make(chan adapter.UDPConn),
udpTimeout: atomic.NewDuration(udpSessionTimeout),
procCancel: func() { /* nop */ },
opts: opts,
}
}
func (h *transportHandler) HandleTCP(conn adapter.TCPConn) {
h.tcpQueue <- conn
}
func (h *transportHandler) HandleUDP(conn adapter.UDPConn) {
h.udpQueue <- conn
}
func (h *transportHandler) process(ctx context.Context) {
for {
select {
case conn := <-h.tcpQueue:
go h.handleTCPConn(conn)
case conn := <-h.udpQueue:
go h.handleUDPConn(conn)
case <-ctx.Done():
return
}
}
}
// ProcessAsync can be safely called multiple times, but will only be effective once.
func (h *transportHandler) ProcessAsync() {
h.procOnce.Do(func() {
ctx, cancel := context.WithCancel(context.Background())
h.procCancel = cancel
go h.process(ctx)
})
}
// Close closes the Tunnel and releases its resources.
func (h *transportHandler) Close() {
h.procCancel()
}
func (h *transportHandler) SetUDPTimeout(timeout time.Duration) {
h.udpTimeout.Store(timeout)
}
func (h *transportHandler) handleTCPConn(originConn adapter.TCPConn) {
defer originConn.Close()
id := originConn.ID()
remoteIP, _ := netip.AddrFromSlice(id.RemoteAddress.AsSlice())
dstIP, _ := netip.AddrFromSlice(id.LocalAddress.AsSlice())
remoteAddr := netip.AddrPortFrom(remoteIP, id.RemotePort)
dstAddr := netip.AddrPortFrom(dstIP, id.LocalPort)
start := time.Now()
sid := xid.New().String()
ctx := ctxvalue.ContextWithSid(context.Background(), ctxvalue.Sid(sid))
ro := &xrecorder.HandlerRecorderObject{
Service: h.opts.Service,
Network: "tcp",
RemoteAddr: remoteAddr.String(),
Dst: dstAddr.String(),
Time: start,
SID: sid,
}
log := h.opts.Logger.WithFields(map[string]any{"network": "tcp", "sid": ro.SID})
log.Debugf("%s <> %s", remoteAddr.String(), dstAddr.String())
var err error
var conn net.Conn = originConn
pStats := xstats.Stats{}
conn = stats_wrapper.WrapConn(conn, &pStats)
defer func() {
if err != nil {
ro.Err = err.Error()
}
ro.InputBytes = pStats.Get(stats.KindInputBytes)
ro.OutputBytes = pStats.Get(stats.KindOutputBytes)
ro.Duration = time.Since(start)
log.WithFields(map[string]any{
"duration": time.Since(start),
"inputBytes": ro.InputBytes,
"outputBytes": ro.OutputBytes,
}).Infof("%s >< %s", remoteAddr.String(), dstAddr.String())
}()
cc, err := h.opts.Router.Dial(ctx, "tcp", dstAddr.String())
if err != nil {
log.Errorf("dial %s: %v", dstAddr.String(), err)
return
}
defer cc.Close()
ro.Src = cc.LocalAddr().String()
t := time.Now()
log.Infof("%s <-> %s", remoteAddr, dstAddr)
xnet.Transport(conn, cc)
log.WithFields(map[string]any{
"duration": time.Since(t),
}).Infof("%s >-< %s", remoteAddr, dstAddr)
}
// TODO: Port Restricted NAT support.
func (t *transportHandler) handleUDPConn(uc adapter.UDPConn) {
defer uc.Close()
}
internal/net/dialer/dialer.go
+4 -4
View File
@@ -98,7 +98,7 @@ func (d *Dialer) Dial(ctx context.Context, network, addr string) (conn net.Conn,
return
}
log.Debugf("dial %s %v@%s failed: %s", network, ifAddr, ifceName, err)
log.Debugf("dial %s/%s via interface %s@%v failed: %s", addr, network, ifceName, ifAddr, err)
if strict &&
!strings.Contains(err.Error(), "no suitable address found") &&
@@ -113,7 +113,7 @@ func (d *Dialer) Dial(ctx context.Context, network, addr string) (conn net.Conn,
func (d *Dialer) dialOnce(ctx context.Context, network, addr, ifceName string, ifAddr net.Addr, log logger.Logger) (net.Conn, error) {
if ifceName != "" {
log.Debugf("interface: %s %v/%s", ifceName, ifAddr, network)
log.Debugf("dial %s/%s via interface %s@%s", addr, network, ifceName, ifAddr)
}
switch network {
@@ -135,7 +135,7 @@ func (d *Dialer) dialOnce(ctx context.Context, network, addr, ifceName string, i
}
err = sc.Control(func(fd uintptr) {
if ifceName != "" {
if err := bindDevice(fd, ifceName); err != nil {
if err := bindDevice(network, fd, ifceName); err != nil {
log.Warnf("bind device: %v", err)
}
}
@@ -159,7 +159,7 @@ func (d *Dialer) dialOnce(ctx context.Context, network, addr, ifceName string, i
Control: func(network, address string, c syscall.RawConn) error {
return c.Control(func(fd uintptr) {
if ifceName != "" {
if err := bindDevice(fd, ifceName); err != nil {
if err := bindDevice(network, fd, ifceName); err != nil {
log.Warnf("bind device: %v", err)
}
}
internal/net/dialer/dialer_darwin.go
+32
View File
@@ -0,0 +1,32 @@
package dialer
import (
"net"
"syscall"
"golang.org/x/sys/unix"
)
func bindDevice(network string, fd uintptr, ifceName string) error {
if ifceName == "" {
return nil
}
ifce, err := net.InterfaceByName(ifceName)
if err == nil {
return err
}
switch network {
case "tcp", "tcp4", "udp4":
return unix.SetsockoptInt(int(fd), syscall.IPPROTO_IP, syscall.IP_BOUND_IF, ifce.Index)
case "tcp6", "udp6":
return unix.SetsockoptInt(int(fd), syscall.IPPROTO_IPV6, syscall.IPV6_BOUND_IF, ifce.Index)
}
return nil
}
func setMark(fd uintptr, mark int) error {
return nil
}
internal/net/dialer/dialer_linux.go
+1 -1
View File
@@ -4,7 +4,7 @@ import (
"golang.org/x/sys/unix"
)
func bindDevice(fd uintptr, ifceName string) error {
func bindDevice(network string, fd uintptr, ifceName string) error {
if ifceName == "" {
return nil
}
internal/net/dialer/dialer_other.go
+2 -2
View File
@@ -1,8 +1,8 @@
//go:build !linux
//go:build !linux && !windows && !darwin
package dialer
func bindDevice(fd uintptr, ifceName string) error {
func bindDevice(network string, fd uintptr, ifceName string) error {
return nil
}
internal/net/dialer/dialer_windows.go
+51
View File
@@ -0,0 +1,51 @@
package dialer
import (
"encoding/binary"
"net"
"unsafe"
"golang.org/x/sys/windows"
)
const (
IP_UNICAST_IF = 31
IPV6_UNICAST_IF = 31
)
func bindDevice(network string, fd uintptr, ifceName string) error {
if ifceName == "" {
return nil
}
ifce, err := net.InterfaceByName(ifceName)
if err == nil {
return err
}
switch network {
case "tcp", "tcp4", "udp4":
return bindSocketToInterface4(windows.Handle(fd), uint32(ifce.Index))
case "tcp6", "udp6":
return bindSocketToInterface6(windows.Handle(fd), uint32(ifce.Index))
}
return nil
}
func setMark(fd uintptr, mark int) error {
return nil
}
func bindSocketToInterface4(handle windows.Handle, index uint32) error {
// For IPv4, this parameter must be an interface index in network byte order.
// Ref: https://learn.microsoft.com/en-us/windows/win32/winsock/ipproto-ip-socket-options
var bytes [4]byte
binary.BigEndian.PutUint32(bytes[:], index)
index = *(*uint32)(unsafe.Pointer(&bytes[0]))
return windows.SetsockoptInt(handle, windows.IPPROTO_IP, IP_UNICAST_IF, int(index))
}
func bindSocketToInterface6(handle windows.Handle, index uint32) error {
return windows.SetsockoptInt(handle, windows.IPPROTO_IPV6, IPV6_UNICAST_IF, int(index))
}
internal/util/forwarder/sniffer.go
+12
View File
@@ -42,6 +42,10 @@ import (
"golang.org/x/time/rate"
)
const (
DefaultReadTimeout = 30 * time.Second
)
var (
DefaultCertPool = tls_util.NewMemoryCertPool()
)
@@ -124,6 +128,10 @@ func (h *Sniffer) HandleHTTP(ctx context.Context, conn net.Conn, opts ...HandleO
opt(&ho)
}
if h.ReadTimeout <= 0 {
h.ReadTimeout = DefaultReadTimeout
}
pStats := xstats.Stats{}
conn = stats_wrapper.WrapConn(conn, &pStats)
@@ -769,6 +777,10 @@ func (h *Sniffer) HandleTLS(ctx context.Context, conn net.Conn, opts ...HandleOp
opt(&ho)
}
if h.ReadTimeout <= 0 {
h.ReadTimeout = DefaultReadTimeout
}
buf := new(bytes.Buffer)
clientHello, err := dissector.ParseClientHello(io.TeeReader(conn, buf))
if err != nil {
internal/util/sniffing/sniffer.go
+10
View File
@@ -39,6 +39,8 @@ import (
)
const (
DefaultReadTimeout = 30 * time.Second
// DefaultBodySize is the default HTTP body or websocket frame size to record.
DefaultBodySize = 64 * 1024 // 64KB
// MaxBodySize is the maximum HTTP body or websocket frame size to record.
@@ -115,6 +117,10 @@ func (h *Sniffer) HandleHTTP(ctx context.Context, conn net.Conn, opts ...HandleO
opt(&ho)
}
if h.ReadTimeout <= 0 {
h.ReadTimeout = DefaultReadTimeout
}
pStats := xstats.Stats{}
conn = stats_wrapper.WrapConn(conn, &pStats)
@@ -570,6 +576,10 @@ func (h *Sniffer) HandleTLS(ctx context.Context, conn net.Conn, opts ...HandleOp
opt(&ho)
}
if h.ReadTimeout <= 0 {
h.ReadTimeout = DefaultReadTimeout
}
buf := new(bytes.Buffer)
clientHello, err := dissector.ParseClientHello(io.TeeReader(conn, buf))
if err != nil {
internal/util/stats/stats.go
+8
View File
@@ -24,6 +24,10 @@ func NewHandlerStats(service string, resetTraffic bool) *HandlerStats {
}
func (p *HandlerStats) Stats(client string) stats.Stats {
if p == nil {
return nil
}
p.mu.RLock()
pstats := p.stats[client]
p.mu.RUnlock()
@@ -43,6 +47,10 @@ func (p *HandlerStats) Stats(client string) stats.Stats {
}
func (p *HandlerStats) Events() (events []observer.Event) {
if p == nil {
return
}
p.mu.RLock()
defer p.mu.RUnlock()
listener/tun/listener.go
+11 -3
View File
@@ -57,15 +57,21 @@ func (l *tunListener) Init(md mdata.Metadata) (err error) {
if err != nil {
return
}
l.cqueue = make(chan net.Conn)
l.cqueue = make(chan net.Conn, 1)
l.closed = make(chan struct{})
go l.listenLoop()
ctx, done := context.WithCancelCause(context.Background())
go l.listenLoop(done)
<-ctx.Done()
if err := context.Cause(ctx); err != ctx.Err() {
return err
}
return
}
func (l *tunListener) listenLoop() {
func (l *tunListener) listenLoop(ready context.CancelCauseFunc) {
for {
ctx, cancel := context.WithCancel(context.Background())
err := func() error {
@@ -116,6 +122,8 @@ func (l *tunListener) listenLoop() {
cancel()
}
ready(err)
select {
case <-ctx.Done():
case <-l.closed:
listener/tun/tun.go
+7 -4
View File
@@ -2,11 +2,14 @@ package tun
import (
"io"
"math"
"golang.zx2c4.com/wireguard/tun"
)
const (
maxBufSize = 16 * 1024
)
type tunDevice struct {
dev tun.Device
packets int
@@ -65,11 +68,11 @@ func (l *tunListener) createTunDevice() (dev io.ReadWriteCloser, name string, er
return
}
batchSize := 16
batchSize := ifce.BatchSize()
rbufs := make([][]byte, batchSize)
for i := 1; i < len(rbufs); i++ {
rbufs[i] = make([]byte, math.MaxUint16)
rbufs[i] = make([]byte, maxBufSize)
}
dev = &tunDevice{
@@ -77,7 +80,7 @@ func (l *tunListener) createTunDevice() (dev io.ReadWriteCloser, name string, er
sizes: make([]int, batchSize),
rbufs: rbufs,
wbufs: make([][]byte, 1),
wbuf: make([]byte, math.MaxUint16+writeOffset),
wbuf: make([]byte, maxBufSize+writeOffset),
}
name, err = ifce.Name()
listener/tun/tun_unix.go → listener/tun/tun_other.go
View File
listener/vtun/conn.go → listener/tungo/conn.go
+1 -1
View File
@@ -1,4 +1,4 @@
package tun
package tungo
import (
"context"
listener/vtun/listener.go → listener/tungo/listener.go
+13 -5
View File
@@ -1,4 +1,4 @@
package tun
package tungo
import (
"context"
@@ -19,7 +19,7 @@ import (
)
func init() {
registry.ListenerRegistry().Register("vtun", NewListener)
registry.ListenerRegistry().Register("tungo", NewListener)
}
type tunListener struct {
@@ -50,15 +50,21 @@ func (l *tunListener) Init(md mdata.Metadata) (err error) {
l.addr = &Addr{Name: l.options.Addr}
l.cqueue = make(chan net.Conn)
l.cqueue = make(chan net.Conn, 1)
l.closed = make(chan struct{})
go l.listenLoop()
ctx, done := context.WithCancelCause(context.Background())
go l.listenLoop(done)
<-ctx.Done()
if err := context.Cause(ctx); err != ctx.Err() {
return err
}
return
}
func (l *tunListener) listenLoop() {
func (l *tunListener) listenLoop(ready context.CancelCauseFunc) {
for {
ctx, cancel := context.WithCancel(context.Background())
err := func() error {
@@ -108,6 +114,8 @@ func (l *tunListener) listenLoop() {
cancel()
}
ready(err)
select {
case <-ctx.Done():
case <-l.closed:
listener/vtun/metadata.go → listener/tungo/metadata.go
+6 -2
View File
@@ -1,4 +1,4 @@
package tun
package tungo
import (
"net"
@@ -12,7 +12,8 @@ import (
)
const (
defaultMTU = 1420
defaultMTU = 1420
defaultName = "tungo"
)
type metadata struct {
@@ -28,6 +29,9 @@ func (l *tunListener) parseMetadata(md mdata.Metadata) (err error) {
if config.MTU <= 0 {
config.MTU = defaultMTU
}
if config.Name == "" {
config.Name = "tungo"
}
if gw := mdutil.GetString(md, "gw", "tun.gw"); gw != "" {
config.Gateway = net.ParseIP(gw)
}
listener/vtun/tun.go → listener/tungo/tun.go
+8 -5
View File
@@ -1,13 +1,16 @@
package tun
package tungo
import (
"io"
"math"
"github.com/go-gost/core/logger"
"golang.zx2c4.com/wireguard/tun"
)
const (
maxBufSize = 16 * 1024
)
type tunDevice struct {
dev tun.Device
@@ -70,11 +73,11 @@ func (l *tunListener) createTunDevice() (dev io.ReadWriteCloser, name string, er
return
}
batchSize := 16
batchSize := ifce.BatchSize()
rbufs := make([][]byte, batchSize)
for i := 1; i < len(rbufs); i++ {
rbufs[i] = make([]byte, math.MaxUint16)
rbufs[i] = make([]byte, maxBufSize)
}
dev = &tunDevice{
@@ -82,7 +85,7 @@ func (l *tunListener) createTunDevice() (dev io.ReadWriteCloser, name string, er
sizes: make([]int, batchSize),
rbufs: rbufs,
wbufs: make([][]byte, 1),
wbuf: make([]byte, math.MaxUint16+writeOffset),
wbuf: make([]byte, maxBufSize+writeOffset),
log: l.log,
}
name, err = ifce.Name()
listener/vtun/tun_darwin.go → listener/tungo/tun_darwin.go
+1 -1
View File
@@ -1,4 +1,4 @@
package tun
package tungo
import (
"fmt"
listener/vtun/tun_linux.go → listener/tungo/tun_linux.go
+1 -1
View File
@@ -1,4 +1,4 @@
package tun
package tungo
import (
"fmt"
listener/vtun/tun_unix.go → listener/tungo/tun_other.go
+1 -1
View File
@@ -1,6 +1,6 @@
//go:build !linux && !windows && !darwin
package tun
package tungo
import (
"fmt"
listener/vtun/tun_windows.go → listener/tungo/tun_windows.go
+1 -1
View File
@@ -1,4 +1,4 @@
package tun
package tungo
import (
"fmt"
service/service.go
+16 -1
View File
@@ -24,6 +24,7 @@ import (
xnet "github.com/go-gost/x/internal/net"
xmetrics "github.com/go-gost/x/metrics"
xstats "github.com/go-gost/x/observer/stats"
"github.com/google/shlex"
"github.com/rs/xid"
)
@@ -296,12 +297,26 @@ func (s *defaultService) execCmds(phase string, cmds []string) {
}
s.options.logger.Info(cmd)
if err := exec.Command("/bin/sh", "-c", cmd).Run(); err != nil {
if err := s.execCmd(cmd); err != nil {
s.options.logger.Warnf("[%s] %s: %v", phase, cmd, err)
}
}
}
func (s *defaultService) execCmd(cmd string) error {
ss, err := shlex.Split(cmd)
if err != nil {
return err
}
if len(ss) == 0 {
return errors.New("invalid command")
}
c := exec.Command(ss[0], ss[1:]...)
// c.Stdout = os.Stdout
// c.Stderr = os.Stderr
return c.Run()
}
func (s *defaultService) setState(state State) {
s.status.setState(state)