wip

Signed-off-by: Ubuntu <ubuntu@ip-172-31-40-207.us-east-2.compute.internal>

ipn/ipnlocal/local.go

@@ -913,6 +913,12 @@ func (b *LocalBackend) setStateLocked(state ipn.State) {
 	}
 }
 
+func (b *LocalBackend) IPServiceMappings() netmap.IPServiceMappings {
+	b.mu.Lock()
+	defer b.mu.Unlock()
+	return b.ipVIPServiceMap
+}
+
 // setConfigLocked uses the provided config to update the backend's prefs
 // and other state.
 func (b *LocalBackend) setConfigLocked(conf *conffile.Config) error {
@@ -5110,7 +5116,7 @@ func (b *LocalBackend) authReconfigLocked() {
 	}
 
 	oneCGNATRoute := shouldUseOneCGNATRoute(b.logf, b.sys.NetMon.Get(), b.sys.ControlKnobs(), version.OS())
-	rcfg := b.routerConfigLocked(cfg, prefs, oneCGNATRoute)
+	rcfg := b.routerConfigLocked(cfg, prefs, nm, oneCGNATRoute)
 
 	err = b.e.Reconfig(cfg, rcfg, dcfg)
 	if err == wgengine.ErrNoChanges {
@@ -5426,7 +5432,7 @@ func peerRoutes(logf logger.Logf, peers []wgcfg.Peer, cgnatThreshold int) (route
 // routerConfig produces a router.Config from a wireguard config and IPN prefs.
 //
 // b.mu must be held.
-func (b *LocalBackend) routerConfigLocked(cfg *wgcfg.Config, prefs ipn.PrefsView, oneCGNATRoute bool) *router.Config {
+func (b *LocalBackend) routerConfigLocked(cfg *wgcfg.Config, prefs ipn.PrefsView, nm *netmap.NetworkMap, oneCGNATRoute bool) *router.Config {
 	singleRouteThreshold := 10_000
 	if oneCGNATRoute {
 		singleRouteThreshold = 1
@@ -5511,13 +5517,31 @@ func (b *LocalBackend) routerConfigLocked(cfg *wgcfg.Config, prefs ipn.PrefsView
 		}
 	}
 
+	// Get the VIPs for VIP services this node hosts. We will add all locally served VIPs to routes then
+	// we terminate these connection locally in netstack instead of routing to peer.
+	VIPServiceIPs := nm.GetIPVIPServiceMap()
+
 	if slices.ContainsFunc(rs.LocalAddrs, tsaddr.PrefixIs4) {
 		rs.Routes = append(rs.Routes, netip.PrefixFrom(tsaddr.TailscaleServiceIP(), 32))
+		for vip := range VIPServiceIPs {
+			if vip.Is4() {
+				rs.Routes = append(rs.Routes, netip.PrefixFrom(vip, 32))
+			}
+		}
 	}
 	if slices.ContainsFunc(rs.LocalAddrs, tsaddr.PrefixIs6) {
 		rs.Routes = append(rs.Routes, netip.PrefixFrom(tsaddr.TailscaleServiceIPv6(), 128))
+		for vip := range VIPServiceIPs {
+			if vip.Is6() {
+				rs.Routes = append(rs.Routes, netip.PrefixFrom(vip, 128))
+			}
+		}
 	}
 
+	fmt.Println("LocalAddrs are: ", rs.LocalAddrs)
+	fmt.Println("SubnetRoutes are: ", rs.SubnetRoutes)
+	fmt.Println("Routes are: ", rs.Routes)
+
 	return rs
 }
 

wgengine/netstack/link_endpoint.go

@@ -5,6 +5,8 @@ package netstack
 
 import (
 	"context"
+	"log"
+	"net/netip"
 	"sync"
 
 	"gvisor.dev/gvisor/pkg/tcpip"
@@ -278,6 +280,7 @@ func (ep *linkEndpoint) WritePackets(pkts stack.PacketBufferList) (int, tcpip.Er
 	//  control MTU (and by effect TCP MSS in gVisor) we *shouldn't* expect to
 	//  ever overflow 128 slots (see wireguard-go/tun.ErrTooManySegments usage).
 	for _, pkt := range pkts.AsSlice() {
+		logLinkEPOut(pkt)
 		if err := ep.q.Write(pkt); err != nil {
 			if _, ok := err.(*tcpip.ErrNoBufferSpace); !ok && n == 0 {
 				return 0, err
@@ -290,6 +293,68 @@ func (ep *linkEndpoint) WritePackets(pkts stack.PacketBufferList) (int, tcpip.Er
 	return n, nil
 }
 
+func logLinkEPOut(pkt *stack.PacketBuffer) {
+	nh := pkt.NetworkHeader().Slice()
+	th := pkt.TransportHeader().Slice()
+	if len(nh) == 0 || len(th) == 0 {
+		return
+	}
+	if pkt.TransportProtocolNumber != header.TCPProtocolNumber || len(th) < header.TCPMinimumSize {
+		return
+	}
+
+	tcp := header.TCP(th)
+	flags := tcp.Flags()
+
+	// Only log SYN/SYN-ACK/RST (SYN=0x02, ACK=0x10, RST=0x04)
+	syn := flags&header.TCPFlagSyn != 0
+	rst := flags&header.TCPFlagRst != 0
+	if !syn && !rst {
+		return
+	}
+
+	sp := tcp.SourcePort()
+	dp := tcp.DestinationPort()
+
+	var srcIP, dstIP netip.Addr
+	switch pkt.NetworkProtocolNumber {
+	case header.IPv4ProtocolNumber:
+		if len(nh) < header.IPv4MinimumSize {
+			return
+		}
+		ip := header.IPv4(nh)
+		srcIP = addrToNetip(ip.SourceAddress())
+		dstIP = addrToNetip(ip.DestinationAddress())
+
+	case header.IPv6ProtocolNumber:
+		if len(nh) < header.IPv6MinimumSize {
+			return
+		}
+		ip := header.IPv6(nh)
+		srcIP = addrToNetip(ip.SourceAddress())
+		dstIP = addrToNetip(ip.DestinationAddress())
+
+	default:
+		return
+	}
+
+	// If parsing failed, don't log noisy junk.
+	if !srcIP.IsValid() || !dstIP.IsValid() {
+		return
+	}
+
+	log.Printf("linkEP out TCP flags=%#x %s:%d -> %s:%d seq=%d ack=%d",
+		flags, srcIP, sp, dstIP, dp, tcp.SequenceNumber(), tcp.AckNumber())
+}
+
+func addrToNetip(a tcpip.Address) netip.Addr {
+	ip, ok := netip.AddrFromSlice(a.AsSlice())
+	if !ok {
+		return netip.Addr{}
+	}
+	return ip.Unmap()
+}
+
 // Wait implements stack.LinkEndpoint.Wait.
 func (*linkEndpoint) Wait() {}
 

wgengine/netstack/netstack.go

@@ -771,6 +771,11 @@ func (ns *Impl) handleLocalPackets(p *packet.Parsed, t *tstun.Wrapper, gro *gro.
 
 	// Determine if we care about this local packet.
 	dst := p.Dst.Addr()
+	var IPServiceMappings netmap.IPServiceMappings
+	if ns.lb != nil {
+		IPServiceMappings = ns.lb.IPServiceMappings()
+	}
+	serviceName, hasIP := IPServiceMappings[dst]
 	switch {
 	case dst == serviceIP || dst == serviceIPv6:
 		// We want to intercept some traffic to the "service IP" (e.g.
@@ -787,6 +792,30 @@ func (ns *Impl) handleLocalPackets(p *packet.Parsed, t *tstun.Wrapper, gro *gro.
 				return filter.Accept, gro
 			}
 		}
+	case hasIP:
+		if p.IPProto != ipproto.TCP {
+			return filter.Accept, gro
+		}
+		// returns all configured VIP services, since the IPServiceMappings contains
+		// inactive service IPs when node hosts the service, we need to check the
+		// service is active or not before dropping the packet.
+		VIPServices := ns.lb.VIPServices()
+		serviceActive := false
+		for _, svc := range VIPServices {
+			// Even though control only send service IP down when there is a config
+			// for the service, we want to still check that the config still exists
+			// before passing the packet to netstack.
+			if svc.Name == serviceName {
+				serviceActive = svc.Active
+			}
+		}
+		if !serviceActive {
+			return filter.Accept, gro
+		}
+		if debugNetstack() {
+			ns.logf("netstack: intercepting local VIP service packet: proto=%v dst=%v src=%v",
+				p.IPProto, p.Dst, p.Src)
+		}
 	case viaRange.Contains(dst):
 		// We need to handle 4via6 packets leaving the host if the via
 		// route is for this host; otherwise the packet will be dropped
@@ -946,6 +975,55 @@ func (ns *Impl) inject() {
 	inboundBuffs, inboundBuffsSizes := ns.getInjectInboundBuffsSizes()
 	for {
 		pkt := ns.linkEP.ReadContext(ns.ctx)
+		nh := pkt.NetworkHeader().Slice()
+		th := pkt.TransportHeader().Slice()
+
+		var src, dst netip.Addr
+		var sp, dp uint16
+		var syn, ack, rst bool
+
+		if len(nh) > 0 {
+			switch pkt.NetworkProtocolNumber {
+			case header.IPv4ProtocolNumber:
+				ip := header.IPv4(nh)
+
+				sa := ip.SourceAddress()
+				da := ip.DestinationAddress()
+
+				if s, ok := netip.AddrFromSlice(sa.AsSlice()); ok {
+					src = s.Unmap()
+				}
+				if d, ok := netip.AddrFromSlice(da.AsSlice()); ok {
+					dst = d.Unmap()
+				}
+
+			case header.IPv6ProtocolNumber:
+				ip := header.IPv6(nh)
+
+				sa := ip.SourceAddress()
+				da := ip.DestinationAddress()
+
+				if s, ok := netip.AddrFromSlice(sa.AsSlice()); ok {
+					src = s.Unmap()
+				}
+				if d, ok := netip.AddrFromSlice(da.AsSlice()); ok {
+					dst = d.Unmap()
+				}
+			}
+		}
+
+		if pkt.TransportProtocolNumber == header.TCPProtocolNumber && len(th) >= header.TCPMinimumSize {
+			tcp := header.TCP(th)
+			sp = tcp.SourcePort()
+			dp = tcp.DestinationPort()
+			f := tcp.Flags()
+			syn = f&header.TCPFlagSyn != 0
+			ack = f&header.TCPFlagAck != 0
+			rst = f&header.TCPFlagRst != 0
+		}
+
+		ns.logf("inject: dequeued TCP syn=%v ack=%v rst=%v %v:%d -> %v:%d",
+			syn, ack, rst, src, sp, dst, dp)
 		if pkt == nil {
 			if ns.ctx.Err() != nil {
 				// Return without logging.
@@ -965,15 +1043,18 @@ func (ns *Impl) inject() {
 		// send traffic destined for the local device, hence must
 		// be injected 'inbound'.
 		sendToHost := ns.shouldSendToHost(pkt)
+		ns.logf("inject: shouldSendToHost=%v", sendToHost)
 
 		// pkt has a non-zero refcount, so injection methods takes
 		// ownership of one count and will decrement on completion.
 		if sendToHost {
+			ns.logf("inject: InjectInboundPacketBuffer")
 			if err := ns.tundev.InjectInboundPacketBuffer(pkt, inboundBuffs, inboundBuffsSizes); err != nil {
 				ns.logf("netstack inject inbound: %v", err)
 				return
 			}
 		} else {
+			ns.logf("inject: InjectOutboundPacketBuffer")
 			if err := ns.tundev.InjectOutboundPacketBuffer(pkt); err != nil {
 				ns.logf("netstack inject outbound: %v", err)
 				return
@@ -998,12 +1079,32 @@ func (ns *Impl) shouldSendToHost(pkt *stack.PacketBuffer) bool {
 			return true
 		}
 
+		if ns.isVIPServiceIP(srcIP) {
+			dstIP := netip.AddrFrom4(v.DestinationAddress().As4())
+			if ns.isLocalIP(dstIP) {
+				if debugNetstack() {
+					ns.logf("netstack: sending VIP service packet to host: src=%v dst=%v", srcIP, dstIP)
+				}
+				return true
+			}
+		}
+
 	case header.IPv6:
 		srcIP := netip.AddrFrom16(v.SourceAddress().As16())
 		if srcIP == serviceIPv6 {
 			return true
 		}
 
+		if ns.isVIPServiceIP(srcIP) {
+			dstIP := netip.AddrFrom16(v.DestinationAddress().As16())
+			if ns.isLocalIP(dstIP) {
+				if debugNetstack() {
+					ns.logf("netstack: sending VIP service packet to host: src=%v dst=%v", srcIP, dstIP)
+				}
+				return true
+			}
+		}
+
 		if viaRange.Contains(srcIP) {
 			// Only send to the host if this 4via6 route is
 			// something this node handles.
@@ -1349,7 +1450,7 @@ func (ns *Impl) acceptTCP(r *tcp.ForwarderRequest) {
 	getConnOrReset := func(opts ...tcpip.SettableSocketOption) *gonet.TCPConn {
 		ep, err := r.CreateEndpoint(&wq)
 		if err != nil {
-			ns.logf("CreateEndpoint error for %s: %v", stringifyTEI(reqDetails), err)
+			ns.logf("CreateEndpoint error for %s: (%T) %v", stringifyTEI(reqDetails), err, err)
 			r.Complete(true) // sends a RST
 			return nil
 		}