add vtun listener and handler
This commit is contained in:
ginuerzh
@@ -0,0 +1,161 @@
|
||||
// Package iobased provides the implementation of io.ReadWriter
|
||||
// based data-link layer endpoints.
|
||||
package tun
|
||||
|
||||
import (
|
||||
"context"
|
||||
"errors"
|
||||
"io"
|
||||
"sync"
|
||||
|
||||
"github.com/go-gost/core/logger"
|
||||
"gvisor.dev/gvisor/pkg/buffer"
|
||||
"gvisor.dev/gvisor/pkg/tcpip"
|
||||
"gvisor.dev/gvisor/pkg/tcpip/header"
|
||||
"gvisor.dev/gvisor/pkg/tcpip/link/channel"
|
||||
"gvisor.dev/gvisor/pkg/tcpip/stack"
|
||||
)
|
||||
|
||||
const (
|
||||
// Queue length for outbound packet, arriving for read. Overflow
|
||||
// causes packet drops.
|
||||
defaultOutQueueLen = 1 << 10
|
||||
)
|
||||
|
||||
// Endpoint implements the interface of stack.LinkEndpoint from io.ReadWriter.
|
||||
type Endpoint struct {
|
||||
*channel.Endpoint
|
||||
|
||||
// rw is the io.ReadWriter for reading and writing packets.
|
||||
rw io.ReadWriter
|
||||
|
||||
// mtu (maximum transmission unit) is the maximum size of a packet.
|
||||
mtu uint32
|
||||
|
||||
// offset can be useful when perform TUN device I/O with TUN_PI enabled.
|
||||
offset int
|
||||
|
||||
// once is used to perform the init action once when attaching.
|
||||
once sync.Once
|
||||
|
||||
// wg keeps track of running goroutines.
|
||||
wg sync.WaitGroup
|
||||
|
||||
log logger.Logger
|
||||
}
|
||||
|
||||
// New returns stack.LinkEndpoint(.*Endpoint) and error.
|
||||
func newEndpoint(rw io.ReadWriter, mtu uint32, offset int, log logger.Logger) (*Endpoint, error) {
|
||||
if mtu == 0 {
|
||||
return nil, errors.New("MTU size is zero")
|
||||
}
|
||||
|
||||
if rw == nil {
|
||||
return nil, errors.New("RW interface is nil")
|
||||
}
|
||||
|
||||
if offset < 0 {
|
||||
return nil, errors.New("offset must be non-negative")
|
||||
}
|
||||
|
||||
return &Endpoint{
|
||||
Endpoint: channel.New(defaultOutQueueLen, mtu, ""),
|
||||
rw: rw,
|
||||
mtu: mtu,
|
||||
offset: offset,
|
||||
log: log,
|
||||
}, nil
|
||||
}
|
||||
|
||||
// Attach launches the goroutine that reads packets from io.Reader and
|
||||
// dispatches them via the provided dispatcher.
|
||||
func (e *Endpoint) Attach(dispatcher stack.NetworkDispatcher) {
|
||||
e.Endpoint.Attach(dispatcher)
|
||||
e.once.Do(func() {
|
||||
ctx, cancel := context.WithCancel(context.Background())
|
||||
e.wg.Add(2)
|
||||
go func() {
|
||||
e.outboundLoop(ctx)
|
||||
e.wg.Done()
|
||||
}()
|
||||
go func() {
|
||||
e.dispatchLoop(cancel)
|
||||
e.wg.Done()
|
||||
}()
|
||||
})
|
||||
}
|
||||
|
||||
func (e *Endpoint) Wait() {
|
||||
e.wg.Wait()
|
||||
}
|
||||
|
||||
// dispatchLoop dispatches packets to upper layer.
|
||||
func (e *Endpoint) dispatchLoop(cancel context.CancelFunc) {
|
||||
// Call cancel() to ensure (*Endpoint).outboundLoop(context.Context) exits
|
||||
// gracefully after (*Endpoint).dispatchLoop(context.CancelFunc) returns.
|
||||
defer cancel()
|
||||
|
||||
offset, mtu := e.offset, int(e.mtu)
|
||||
|
||||
for {
|
||||
data := make([]byte, offset+mtu)
|
||||
|
||||
n, err := e.rw.Read(data)
|
||||
if err != nil {
|
||||
e.log.Error(err)
|
||||
break
|
||||
}
|
||||
|
||||
e.log.Debugf("read tun: (%d) % x", n, data[:n])
|
||||
|
||||
if n == 0 || n > mtu {
|
||||
continue
|
||||
}
|
||||
|
||||
if !e.IsAttached() {
|
||||
continue /* unattached, drop packet */
|
||||
}
|
||||
|
||||
pkt := stack.NewPacketBuffer(stack.PacketBufferOptions{
|
||||
Payload: buffer.MakeWithData(data[offset : offset+n]),
|
||||
})
|
||||
|
||||
switch header.IPVersion(data[offset:]) {
|
||||
case header.IPv4Version:
|
||||
e.InjectInbound(header.IPv4ProtocolNumber, pkt)
|
||||
case header.IPv6Version:
|
||||
e.InjectInbound(header.IPv6ProtocolNumber, pkt)
|
||||
}
|
||||
pkt.DecRef()
|
||||
}
|
||||
}
|
||||
|
||||
// outboundLoop reads outbound packets from channel, and then it calls
|
||||
// writePacket to send those packets back to lower layer.
|
||||
func (e *Endpoint) outboundLoop(ctx context.Context) {
|
||||
for {
|
||||
pkt := e.ReadContext(ctx)
|
||||
if pkt == nil {
|
||||
break
|
||||
}
|
||||
e.writePacket(pkt)
|
||||
}
|
||||
}
|
||||
|
||||
// writePacket writes outbound packets to the io.Writer.
|
||||
func (e *Endpoint) writePacket(pkt *stack.PacketBuffer) tcpip.Error {
|
||||
defer pkt.DecRef()
|
||||
|
||||
buf := pkt.ToBuffer()
|
||||
defer buf.Release()
|
||||
if e.offset != 0 {
|
||||
v := buffer.NewViewWithData(make([]byte, e.offset))
|
||||
_ = buf.Prepend(v)
|
||||
}
|
||||
|
||||
if _, err := e.rw.Write(buf.Flatten()); err != nil {
|
||||
return &tcpip.ErrInvalidEndpointState{}
|
||||
}
|
||||
|
||||
return nil
|
||||
}
|
||||
@@ -0,0 +1,84 @@
|
||||
package tun
|
||||
|
||||
import (
|
||||
"context"
|
||||
"net"
|
||||
"time"
|
||||
|
||||
"github.com/go-gost/core/handler"
|
||||
md "github.com/go-gost/core/metadata"
|
||||
ctxvalue "github.com/go-gost/x/ctx"
|
||||
"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
|
||||
|
||||
start := time.Now()
|
||||
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())
|
||||
defer func() {
|
||||
log.WithFields(map[string]any{
|
||||
"duration": time.Since(start),
|
||||
}).Infof("%s >< %s", conn.RemoteAddr(), conn.LocalAddr())
|
||||
}()
|
||||
|
||||
ep, err := newEndpoint(conn, 1420, 0, log)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
th := newTransportHandler(log)
|
||||
th.ProcessAsync()
|
||||
defer th.Close()
|
||||
|
||||
stack, err := core.CreateStack(&core.Config{
|
||||
LinkEndpoint: ep,
|
||||
TransportHandler: th,
|
||||
})
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
defer stack.Close()
|
||||
|
||||
log.Debugf("is attached: %v", ep.IsAttached())
|
||||
|
||||
<-ctx.Done()
|
||||
|
||||
return nil
|
||||
}
|
||||
@@ -0,0 +1,12 @@
|
||||
package tun
|
||||
|
||||
import (
|
||||
mdata "github.com/go-gost/core/metadata"
|
||||
)
|
||||
|
||||
type metadata struct {
|
||||
}
|
||||
|
||||
func (h *tunHandler) parseMetadata(md mdata.Metadata) (err error) {
|
||||
return
|
||||
}
|
||||
@@ -0,0 +1,105 @@
|
||||
package tun
|
||||
|
||||
import (
|
||||
"context"
|
||||
"net/netip"
|
||||
"sync"
|
||||
"time"
|
||||
|
||||
"github.com/go-gost/core/logger"
|
||||
"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
|
||||
|
||||
log logger.Logger
|
||||
}
|
||||
|
||||
func newTransportHandler(log logger.Logger) *transportHandler {
|
||||
return &transportHandler{
|
||||
tcpQueue: make(chan adapter.TCPConn),
|
||||
udpQueue: make(chan adapter.UDPConn),
|
||||
udpTimeout: atomic.NewDuration(udpSessionTimeout),
|
||||
procCancel: func() { /* nop */ },
|
||||
log: log,
|
||||
}
|
||||
}
|
||||
|
||||
func (t *transportHandler) HandleTCP(conn adapter.TCPConn) {
|
||||
t.tcpQueue <- conn
|
||||
}
|
||||
|
||||
func (t *transportHandler) HandleUDP(conn adapter.UDPConn) {
|
||||
t.udpQueue <- conn
|
||||
}
|
||||
|
||||
func (t *transportHandler) process(ctx context.Context) {
|
||||
for {
|
||||
select {
|
||||
case conn := <-t.tcpQueue:
|
||||
go t.handleTCPConn(conn)
|
||||
case conn := <-t.udpQueue:
|
||||
go t.handleUDPConn(conn)
|
||||
case <-ctx.Done():
|
||||
return
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// ProcessAsync can be safely called multiple times, but will only be effective once.
|
||||
func (t *transportHandler) ProcessAsync() {
|
||||
t.procOnce.Do(func() {
|
||||
ctx, cancel := context.WithCancel(context.Background())
|
||||
t.procCancel = cancel
|
||||
go t.process(ctx)
|
||||
})
|
||||
}
|
||||
|
||||
// Close closes the Tunnel and releases its resources.
|
||||
func (t *transportHandler) Close() {
|
||||
t.procCancel()
|
||||
}
|
||||
|
||||
func (t *transportHandler) SetUDPTimeout(timeout time.Duration) {
|
||||
t.udpTimeout.Store(timeout)
|
||||
}
|
||||
|
||||
func (t *transportHandler) handleTCPConn(originConn adapter.TCPConn) {
|
||||
defer originConn.Close()
|
||||
|
||||
id := originConn.ID()
|
||||
|
||||
srcIP, _ := netip.AddrFromSlice(id.RemoteAddress.AsSlice())
|
||||
dstIP, _ := netip.AddrFromSlice(id.LocalAddress.AsSlice())
|
||||
|
||||
raddr := netip.AddrPortFrom(srcIP, id.RemotePort)
|
||||
laddr := netip.AddrPortFrom(dstIP, id.LocalPort)
|
||||
|
||||
t.log.Debugf("[TCP] %s <-> %s", raddr.String(), laddr.String())
|
||||
}
|
||||
|
||||
// TODO: Port Restricted NAT support.
|
||||
func (t *transportHandler) handleUDPConn(uc adapter.UDPConn) {
|
||||
defer uc.Close()
|
||||
}
|
||||
Reference in New Issue
Block a user