icybear/tslink
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Merge remote-tracking branch 'origin/master' into sf

Browse Source
This commit is contained in:
iceBear67
2026-05-31 15:48:07 +08:00
parent 26aa088d58 ea246896e1
commit 5204500d73
7 changed files with 418 additions and 42 deletions
Download Patch File Download Diff File Expand all files Collapse all files
core/forwarder.go
+138 -27
View File
@@ -6,6 +6,7 @@ import (
"io"
"log/slog"
"net"
"net/netip"
"sync"
"time"
@@ -17,7 +18,7 @@ func StartForwarders(ctx context.Context, srv *tsnet.Server, rules map[string][]
for _, rule := range rrs {
rule := rule
tag := tag
slog.Debug("starting forwarder",
slog.Info("starting forwarder",
slog.String("tag", tag),
slog.String("protocol", rule.Protocol),
slog.Int("tailscale_port", rule.TailscalePort),
@@ -42,7 +43,7 @@ func StartConnectors(ctx context.Context, srv *tsnet.Server, rules map[string][]
if rule.LocalAddr != "" {
args = append(args, slog.String("local_addr", rule.LocalAddr))
}
slog.Debug("starting connector", args...)
slog.Info("starting connector", args...)
go runConnector(ctx, srv, rule, tag)
}
}
@@ -87,17 +88,13 @@ func runForwarder(ctx context.Context, srv *tsnet.Server, rule ForwardRule, tag
}
func runTCPForwarder(ctx context.Context, srv *tsnet.Server, rule ForwardRule, logger *slog.Logger) {
ip4, ip6 := srv.TailscaleIPs()
ip := ip4
if !ip.IsValid() {
ip = ip6
}
ip := getSelfTsnetAddr(srv)
ln, err := srv.Listen("tcp", fmt.Sprintf("%s:%d", ip.String(), rule.TailscalePort))
if err != nil {
logger.Error("failed to listen", "error", err)
return
}
logger.Info("listening", slog.String("on", fmt.Sprintf("tailscale:%s:%d", ip.String(), rule.TailscalePort)))
logger.Debug("listening", slog.String("on", fmt.Sprintf("tailscale:%s:%d", ip.String(), rule.TailscalePort)))
go func() {
<-ctx.Done()
@@ -177,32 +174,115 @@ func getConnType(ctx context.Context, srv *tsnet.Server, remoteAddrStr string) s
}
func runUDPForwarder(ctx context.Context, srv *tsnet.Server, rule ForwardRule, logger *slog.Logger) {
ip4, ip6 := srv.TailscaleIPs()
ip := ip4
if !ip.IsValid() {
ip = ip6
}
pc, err := srv.ListenPacket("udp", fmt.Sprintf("%s:%d", ip.String(), rule.TailscalePort))
ip := getSelfTsnetAddr(srv)
ln, err := srv.Listen("udp", fmt.Sprintf("%s:%d", ip.String(), rule.TailscalePort))
if err != nil {
logger.Error("failed to listen", "error", err)
return
}
logger.Info("listening", slog.String("on", fmt.Sprintf("tailscale:%s:%d", ip.String(), rule.TailscalePort)))
relay := &udpRelay{
listenConn: pc,
dialAddr: rule.LocalAddr,
logger: logger,
direction: "local",
sessions: make(map[string]*udpSession),
go func() {
<-ctx.Done()
ln.Close()
}()
for {
conn, err := ln.Accept()
if err != nil {
if ctx.Err() != nil {
return
}
logger.Error("accept error", "error", err)
continue
}
go handleUDPForward(ctx, srv, conn, rule, logger)
}
relay.run(ctx)
}
func handleUDPForward(ctx context.Context, srv *tsnet.Server, conn net.Conn, rule ForwardRule, logger *slog.Logger) {
remoteAddrStr := conn.RemoteAddr().String()
clog := logger.With(slog.String("remote", remoteAddrStr))
lc, err := srv.LocalClient()
if err == nil {
who, err := lc.WhoIs(ctx, remoteAddrStr)
if err == nil {
clog = clog.With(slog.String("user", who.UserProfile.LoginName))
}
}
connType := getConnType(ctx, srv, remoteAddrStr)
clog.Info("accepted connection",
slog.String("conn_type", connType),
slog.String("local_addr", rule.LocalAddr),
)
defer conn.Close()
localConn, err := net.Dial("udp", rule.LocalAddr)
if err != nil {
clog.Error("failed to dial local", "error", err)
return
}
defer localConn.Close()
remoteIP, _, _ := net.SplitHostPort(remoteAddrStr)
var toTs, toLocal int64
var wg sync.WaitGroup
wg.Add(2)
go func() {
defer wg.Done()
buf := make([]byte, 65535)
for {
n, err := conn.Read(buf)
if err != nil {
return
}
toLocal += int64(n)
clog.Debug("inbound udp packet",
slog.String("from_ip", remoteIP),
slog.String("to_ip", rule.LocalAddr),
slog.Int("pkg_size", n),
)
if _, err := localConn.Write(buf[:n]); err != nil {
return
}
}
}()
go func() {
defer wg.Done()
buf := make([]byte, 65535)
for {
n, err := localConn.Read(buf)
if err != nil {
return
}
toTs += int64(n)
localIP, _, _ := net.SplitHostPort(localConn.RemoteAddr().String())
clog.Debug("outbound udp packet",
slog.String("from_ip", localIP),
slog.String("to_ip", remoteIP),
slog.Int("pkg_size", n),
)
if _, err := conn.Write(buf[:n]); err != nil {
return
}
}
}()
wg.Wait()
clog.Info("connection closed", slog.Int64("ts_rx_bytes", toLocal), slog.Int64("ts_tx_bytes", toTs))
}
type udpSession struct {
conn net.Conn
remote net.Addr
lastUse time.Time
conn net.Conn
remote net.Addr
inTailscale bool
lastUse time.Time
}
type udpRelay struct {
@@ -210,6 +290,8 @@ type udpRelay struct {
dialAddr string
logger *slog.Logger
direction string
srv *tsnet.Server
ctx context.Context
mu sync.Mutex
sessions map[string]*udpSession
@@ -255,23 +337,43 @@ func (r *udpRelay) run(ctx context.Context) {
r.mu.Lock()
sess, exists := r.sessions[key]
if !exists {
dialed, err := net.Dial("udp", r.dialAddr)
inTsnet := false
if host, _, err := net.SplitHostPort(r.dialAddr); err == nil {
if dialIP, err := netip.ParseAddr(host); err == nil {
tsnetCIDR := netip.MustParsePrefix("100.64.0.0/10")
inTsnet = tsnetCIDR.Contains(dialIP)
}
}
var dialed net.Conn
if inTsnet {
dialed, err = r.srv.Dial(r.ctx, "udp", r.dialAddr)
} else {
dialed, err = net.Dial("udp", r.dialAddr)
}
if err != nil {
r.mu.Unlock()
r.logger.Error("failed to dial", "error", err)
continue
}
sess = &udpSession{conn: dialed, remote: from, lastUse: time.Now()}
sess = &udpSession{conn: dialed, remote: from, lastUse: time.Now(), inTailscale: inTsnet}
r.sessions[key] = sess
r.mu.Unlock()
r.logger.Debug("new udp session", slog.String("remote", key), slog.String("direction", r.direction))
r.logger.Info("new udp session", slog.String("remote", key), slog.String("direction", r.direction))
go r.readSession(key, sess)
} else {
sess.lastUse = time.Now()
r.mu.Unlock()
}
fromIP, _, _ := net.SplitHostPort(from.String())
toIP, _, _ := net.SplitHostPort(sess.conn.RemoteAddr().String())
r.logger.Debug("outbound udp packet",
slog.String("from_ip", fromIP),
slog.String("to_ip", toIP),
slog.Int("pkg_size", n),
)
if _, err := sess.conn.Write(buf[:n]); err != nil {
r.logger.Error("failed to write", "error", err)
r.removeSession(key)
@@ -287,6 +389,13 @@ func (r *udpRelay) readSession(key string, sess *udpSession) {
r.removeSession(key)
return
}
fromIP, _, _ := net.SplitHostPort(sess.conn.RemoteAddr().String())
toIP, _, _ := net.SplitHostPort(sess.remote.String())
r.logger.Info("udp packet",
slog.String("from_ip", fromIP),
slog.String("to_ip", toIP),
slog.Int("pkg_size", n),
)
if _, err := r.listenConn.WriteTo(buf[:n], sess.remote); err != nil {
r.logger.Error("failed to write back", "error", err)
r.removeSession(key)
@@ -406,6 +515,8 @@ func runUDPConnector(ctx context.Context, srv *tsnet.Server, rule ConnectRule, l
dialAddr: rule.DstAddr,
logger: logger,
direction: "tailscale",
srv: srv,
ctx: ctx,
sessions: make(map[string]*udpSession),
}
relay.run(ctx)
core/tsnet.go
+14 -5
View File
@@ -4,20 +4,27 @@ import (
"context"
fmt2 "fmt"
"log/slog"
"time"
"tailscale.com/ipn"
"tailscale.com/tsnet"
)
func InitTsNet(ctx context.Context, cfg *Core, logger *slog.Logger) (*tsnet.Server, error) {
func InitTsNet(ctx context.Context, cfg *Core, logger *slog.Logger, withDebugLog bool) (*tsnet.Server, error) {
dbgLogger := func(fmt string, args ...interface{}) {}
if withDebugLog {
logger.Warn("Tsnet debug log activated")
dbgLogger = func(fmt string, args ...interface{}) {
logger.With(slog.String("from", "tsnet")).Debug(fmt2.Sprintf(fmt, args...))
}
}
srv := &tsnet.Server{
Hostname: "tslink-" + cfg.Hostname,
AuthKey: cfg.AuthKey,
Ephemeral: cfg.Ephemeral,
Logf: func(fmt string, args ...interface{}) {
logger.With(slog.String("from", "tsnet")).Debug(fmt2.Sprintf(fmt, args...))
},
Logf: dbgLogger,
UserLogf: func(fmt string, args ...interface{}) {
logger.With(slog.String("from", "tsnet")).Info(fmt2.Sprintf(fmt, args...))
},
@@ -34,7 +41,9 @@ func InitTsNet(ctx context.Context, cfg *Core, logger *slog.Logger) (*tsnet.Serv
return nil, err
}
status, err := srv.Up(ctx)
cnclCtx, cancel := context.WithTimeout(ctx, 30*time.Second)
defer cancel()
status, err := srv.Up(cnclCtx)
if err != nil {
logger.With(slog.String("error", err.Error())).Error("bring up tsnet server failed")
return nil, err
core/utils.go
+35 -6
View File
@@ -11,6 +11,7 @@ import (
"time"
"tailscale.com/client/local"
"tailscale.com/ipn/ipnstate"
"tailscale.com/tailcfg"
"tailscale.com/tsnet"
)
@@ -86,7 +87,6 @@ func getPeerFromRules(ctx context.Context, srv *tsnet.Server, rules map[string][
ap, _, err := net.SplitHostPort(rule.DstAddr)
if err != nil {
logger.Debug("error parsing rule", "tag", tag, "dst", rule.DstAddr, "err", err)
// err log
continue
}
addr, err := resolveAddr(ctx, srv, ap)
@@ -109,19 +109,39 @@ func getPeerFromRules(ctx context.Context, srv *tsnet.Server, rules map[string][
func peerConnectivityLogic(ctx context.Context, lc *local.Client, relativePeers []netip.Addr, logger *slog.Logger) {
for _, peer := range relativePeers {
ping, err := lc.Ping(ctx, peer, tailcfg.PingDisco)
loLog := logger.With("peer", peer)
ping, err := func() (*ipnstate.PingResult, error) {
cnclCtx, cancel := context.WithTimeout(ctx, 5*time.Second)
defer cancel()
ping, err := lc.Ping(cnclCtx, peer, tailcfg.PingDisco)
return ping, err
}()
if err != nil {
logger.Debug("connectivity: failed to ping peer", "peer", peer, "err", err)
return
if errors.Is(err, context.DeadlineExceeded) {
loLog.Warn("connectivity: peer ping timeout")
} else {
loLog.Warn("connectivity: failed to ping peer", "err", err)
}
continue
}
peerInfo, err := lc.WhoIs(ctx, peer.String())
if err != nil {
loLog.Warn("failed to get peer info", "err", err)
} else {
loLog = loLog.With("name", peerInfo.Node.ComputedName)
}
var connect string
if ping.DERPRegionCode == "" {
connect = "direct"
} else {
connect = ping.DERPRegionCode
}
logger.Info("connectivity: peer pinged",
"peer", peer,
loLog.Info("connectivity: peer pinged",
"latency", fmt.Sprintf("%.2fms", ping.LatencySeconds*1000),
"connect", connect,
)
@@ -160,3 +180,12 @@ func StartPeerConnectivityDiagnostics(ctx context.Context, logger *slog.Logger,
}
}()
}
func getSelfTsnetAddr(srv *tsnet.Server) netip.Addr {
ip4, ip6 := srv.TailscaleIPs()
ip := ip4
if !ip.IsValid() {
ip = ip6
}
return ip
}