diff --git a/arp/arp_test.go b/arp/arp_test.go index a1e69bf..3435e74 100644 --- a/arp/arp_test.go +++ b/arp/arp_test.go @@ -2,8 +2,6 @@ package arp import ( "bytes" - "log" - "slices" "testing" "github.com/soypat/lneto" @@ -49,9 +47,9 @@ func TestHandler(t *testing.T) { } // Perform ARP exchange. - expectHWAddr := c2.ourHWAddr + expectHWAddr := c2.ourHWAddr[:] queryAddr := c2.ourProtoAddr - err = c1.StartQuery(nil, queryAddr) + err = c1.StartQuery(queryAddr, false) if err != nil { t.Fatal(err) } @@ -85,11 +83,11 @@ func TestHandler(t *testing.T) { if err != nil { t.Fatal(err) } - hwaddr, err := c1.QueryResult(queryAddr) + hwaddr, err := c1.CacheLookup(queryAddr) if err != nil { - log.Fatal("expected query result:", err) + t.Fatal("expected query result:", err) } else if !bytes.Equal(hwaddr, expectHWAddr) { - log.Fatalf("expected to get hwaddr %x!=%x", hwaddr, expectHWAddr) + t.Fatalf("expected to get hwaddr %x!=%x", hwaddr, expectHWAddr) } n, err = c1.Encapsulate(buf[:], -1, 0) if err != nil { @@ -105,85 +103,6 @@ func TestHandler(t *testing.T) { } } -func TestQueryCompaction(t *testing.T) { - var h Handler - - startQuery := func(addr []byte) { - err := h.StartQuery(nil, addr) - if err != nil { - t.Fatal(err) - } - } - - err := h.Reset(HandlerConfig{ - HardwareAddr: []byte{0xde, 0xad, 0xbe, 0xef, 0x00, 0x00}, - ProtocolAddr: []byte{192, 168, 1, 1}, - MaxQueries: 5, - MaxPending: 1, - HardwareType: 1, - ProtocolType: ethernet.TypeIPv4, - }) - if err != nil { - t.Fatal(err) - } - - // Create multiple queries - addr1 := []byte{192, 168, 1, 10} - addr2 := []byte{192, 168, 1, 20} - addr3 := []byte{192, 168, 1, 30} - - // Start 3 queries - startQuery(addr1) - startQuery(addr2) - startQuery(addr3) - - if len(h.queries) != 3 { - t.Fatalf("expected 3 queries, got %d", len(h.queries)) - } - - // Discard the middle query (addr2) - if err := h.DiscardQuery(addr2); err != nil { - t.Fatal(err) - } - - // Verify addr2 is marked as invalid - hasAddr2 := slices.ContainsFunc(h.queries, func(q queryResult) bool { - return bytes.Equal(q.protoaddr, addr2) - }) - if hasAddr2 { - t.Fatal("addr2 query found after discard") - } - - // Start new queries to trigger compaction - addr4 := []byte{192, 168, 1, 40} - addr5 := []byte{192, 168, 1, 50} - addr6 := []byte{192, 168, 1, 60} - - startQuery(addr4) - startQuery(addr5) - startQuery(addr6) - - // After compaction we should be left with 5 queries - expectedAddrs := [][]byte{addr1, addr3, addr4, addr5, addr6} - - if len(h.queries) != len(expectedAddrs) { - t.Fatalf("after compaction: expected %d queries, got %d", len(expectedAddrs), len(h.queries)) - } - - gotAddrs := [][]byte{} - for _, q := range h.queries { - if !q.isInvalid() { - gotAddrs = append(gotAddrs, q.protoaddr) - } else { - t.Fatalf("invalid query %v should have been removed during compaction", q.protoaddr) - } - } - - if !slices.EqualFunc(gotAddrs, expectedAddrs, bytes.Equal) { - t.Fatalf("expected %v, got %v", expectedAddrs, gotAddrs) - } -} - func validateARP(t *testing.T, buf []byte) { t.Helper() afrm, err := NewFrame(buf) diff --git a/arp/cache.go b/arp/cache.go new file mode 100644 index 0000000..ed72051 --- /dev/null +++ b/arp/cache.go @@ -0,0 +1,167 @@ +package arp + +import ( + "github.com/soypat/lneto" + "github.com/soypat/lneto/ethernet" + "github.com/soypat/lneto/internal" +) + +type cache struct { + entries []entry +} + +// entry is designed for compactness. size=class=24 bytes, same as a slice header on x86. +type entry struct { + addr [16]byte + mac [6]byte + age uint8 + flags eflags +} + +func (e *entry) use(mac [6]byte, proto []byte, flags eflags) { + e.flags = eflagInUse | flags + if copy(e.addr[:], proto) == 16 { + e.flags |= eflagIPv6 + } + e.age = 0 + e.mac = mac +} + +func (e *entry) destroy() { *e = entry{} } + +func (e *entry) put(frame, ourAddr []byte, ourMAC [6]byte, op Operation) (int, error) { + if len(ourMAC) != 6 { + return 0, lneto.ErrInvalidAddr + } + f, err := NewFrame(frame) + if err != nil { + return 0, err + } + f.SetHardware(1, 6) + f.SetOperation(op) + var n int + if e.flags&eflagIPv6 != 0 { + if len(ourAddr) != 16 { + return 0, lneto.ErrInvalidAddr + } else if len(frame) < sizeHeaderv6 { + return 0, lneto.ErrShortBuffer + } + f.SetProtocol(ethernet.TypeIPv6, 16) + hw, addr := f.Sender16() + *hw = ourMAC + copy(addr[:], ourAddr) + hw, addr = f.Target16() + copy(hw[:], e.mac[:]) + copy(addr[:], e.addr[:]) + n = sizeHeaderv6 + } else { + if len(ourAddr) != 4 { + return 0, lneto.ErrInvalidAddr + } + f.SetProtocol(ethernet.TypeIPv4, 4) + hw, addr := f.Sender4() + *hw = ourMAC + copy(addr[:], ourAddr) + hw, addr = f.Target4() + copy(hw[:], e.mac[:]) + copy(addr[:], e.addr[:]) + n = sizeHeaderv4 + } + return n, nil +} + +func (flags eflags) hasAny(bits eflags) bool { return flags&bits != 0 } + +type eflags uint8 + +// unset eflagInUse to signal the entry can be acquired for a new query. +const ( + // eflagInUse set when in use. Discarded/unused entries have this bit unset. + eflagInUse eflags = 1 << iota + // set for IPv6 addressed entries. + eflagIPv6 + // network device queried our address and we must respond to it. + // Both MAC and IP are valid in this case. + eflagPendingResponse + // user asked to query this address and query has yet to be answered. May or may not be sent. + eflagIncomplete + // user asked to query this address and the query has not been sent out yet. + // The MAC address is invalid in this case. + eflagIncompletePendingQuery + // eflagPriority set for prioritized cache entries. These entries are discarded last. + // i.e: set for user created queries, unset for external incoming network queries. + eflagPriority + // trigger callback, you know the drill. + eflagResolveTriggersCallback +) + +func (c *cache) age() { + for i := range c.entries { + if c.entries[i].flags&eflagInUse != 0 && c.entries[i].age < 255 { + c.entries[i].age++ + } + } +} + +func (c *cache) reset(size int) { + internal.SliceReuse(&c.entries, size) + c.entries = c.entries[:cap(c.entries)] // maximize queries given allocation. +} + +func (c *cache) getNextFlagged(entryHasFlags eflags) *entry { + for i := range c.entries { + flags := c.entries[i].flags + if flags&eflagInUse != 0 && flags.hasAny(entryHasFlags) { + return &c.entries[i] + } + } + return nil +} + +func (c *cache) clearFlags(entryHasFlags, clrTheseFlagsIfMatch eflags) { + for i := range c.entries { + // Can clear flags on unused too, simpler. + if c.entries[i].flags&entryHasFlags != 0 { + c.entries[i].flags &^= clrTheseFlagsIfMatch + } + } +} + +func (c *cache) Lookup(addr []byte) *entry { + n := len(addr) + for i := range c.entries { + if c.entries[i].flags&eflagInUse != 0 && internal.BytesEqual(c.entries[i].addr[:n], addr) { + return &c.entries[i] + } + } + return nil +} + +// acquireNext gets next available entry for use. If all are in use evicts +// the oldest passive entry (learned from incoming requests) before touching +// active user queries or pending responses. +func (c *cache) acquireNext() *entry { + const priorityFlags = eflagPendingResponse | eflagIncomplete | eflagPriority + oldest, oldestPassive := 0, -1 + for i := range c.entries { + if c.entries[i].flags&eflagInUse == 0 { + oldest = i + break + } + if !c.entries[i].flags.hasAny(priorityFlags) { + if oldestPassive < 0 || c.entries[i].age > c.entries[oldestPassive].age { + oldestPassive = i + } + } + if c.entries[i].age > c.entries[oldest].age { + oldest = i + } + } + if oldestPassive >= 0 && c.entries[oldest].flags&eflagInUse != 0 { + oldest = oldestPassive + } + c.age() + e := &c.entries[oldest] + e.destroy() + return e +} diff --git a/arp/handler.go b/arp/handler.go index ecb4671..dbcc2a6 100644 --- a/arp/handler.go +++ b/arp/handler.go @@ -1,21 +1,21 @@ package arp import ( - "log/slog" - "github.com/soypat/lneto" "github.com/soypat/lneto/ethernet" "github.com/soypat/lneto/internal" ) type Handler struct { - connID uint64 - ourHWAddr []byte - ourProtoAddr []byte - htype uint16 - protoType ethernet.Type - pendingResponse [][sizeHeaderv6]byte - queries []queryResult + connID uint64 + cache cache + vld lneto.Validator + ourProtoAddr []byte + onresolve func(hw, proto []byte) + + htype uint16 + protoType ethernet.Type + ourHWAddr [6]byte } type HandlerConfig struct { @@ -41,6 +41,10 @@ func (h *Handler) UpdateProtoAddr(protoAddr []byte) error { return nil } +func (h *Handler) SetOnResolveCallback(cb func(hwAddr, protoAddr []byte)) { + h.onresolve = cb +} + func (h *Handler) Reset(cfg HandlerConfig) error { if len(cfg.HardwareAddr) == 0 || len(cfg.HardwareAddr) > 255 || len(cfg.ProtocolAddr) == 0 || len(cfg.ProtocolAddr) > 255 { @@ -48,197 +52,120 @@ func (h *Handler) Reset(cfg HandlerConfig) error { } else if cfg.MaxQueries <= 0 || cfg.MaxPending <= 0 { return lneto.ErrInvalidConfig } + if cfg.HardwareType != 1 || cfg.ProtocolType != ethernet.TypeIPv4 && cfg.ProtocolType != ethernet.TypeIPv6 { + return lneto.ErrUnsupported // We only support common for now. + } *h = Handler{ - connID: h.connID + 1, - ourHWAddr: h.ourHWAddr[:0], - ourProtoAddr: h.ourProtoAddr[:0], - htype: cfg.HardwareType, - protoType: cfg.ProtocolType, - pendingResponse: h.pendingResponse[:0], - queries: h.queries[:0], + connID: h.connID + 1, + ourHWAddr: h.ourHWAddr, + ourProtoAddr: h.ourProtoAddr[:0], + htype: cfg.HardwareType, + protoType: cfg.ProtocolType, + cache: h.cache, } - h.ourHWAddr = append(h.ourHWAddr, cfg.HardwareAddr...) + h.cache.reset(cfg.MaxPending + cfg.MaxQueries) + + h.ourHWAddr = [6]byte(cfg.HardwareAddr) h.ourProtoAddr = append(h.ourProtoAddr, cfg.ProtocolAddr...) - if cap(h.pendingResponse) < cfg.MaxPending { - h.pendingResponse = make([][52]byte, cfg.MaxPending)[:0] - } - if cap(h.queries) < cfg.MaxQueries { - h.queries = make([]queryResult, cfg.MaxQueries)[:0] - } return nil } -type queryResult struct { - protoaddr []byte - hwaddr []byte - dstHw []byte - querysent bool - // inc tracks amount of times the query survived compaction. - // The queries with higher inc will be discarded first. - inc uint16 -} - -func (qr *queryResult) destroy() { - *qr = queryResult{protoaddr: qr.protoaddr[:0], hwaddr: qr.hwaddr[:0]} -} - -func (qr *queryResult) response() []byte { - if len(qr.hwaddr) == 0 { - return nil - } - return qr.hwaddr[:] -} -func (qr *queryResult) isInvalid() bool { return len(qr.protoaddr) == 0 } - // AbortPending drops pending queries and incoming requests. func (h *Handler) AbortPending() { - h.pendingResponse = h.pendingResponse[:0] - h.queries = h.queries[:0] + h.cache.clearFlags(eflagPendingResponse|eflagIncomplete, eflagInUse) } -func (h *Handler) expectSize() int { - return sizeHeader + 2*len(h.ourHWAddr) + 2*len(h.ourProtoAddr) +// CacheSeed pre-populates the cache with a known proto→hardware mapping, making it +// immediately resolvable via [Handler.CacheLookup] without an ARP exchange. +// Seeded entries are evicted before active user queries when the cache is full. +func (h *Handler) CacheSeed(protoAddr, hwAddr []byte) error { + if len(hwAddr) != 6 { + return lneto.ErrUnsupported + } + e := h.cache.acquireNext() + e.use([6]byte(hwAddr), protoAddr, 0) + return nil } -func (h *Handler) QueryResult(protoAddr []byte) (hwAddr []byte, err error) { - for i := range h.queries { - if internal.BytesEqual(protoAddr, h.queries[i].protoaddr) { - if !h.queries[i].querysent { - return nil, errQueryPending - } - mac := h.queries[i].response() - if mac == nil { - return nil, errQueryPending - } - return mac, nil - } +// CacheLookup returns the hardware address for protoAddr if it is resolved in the cache. +// Returns [errQueryPending] if a query is in flight, [errQueryNotFound] if no entry exists. +func (h *Handler) CacheLookup(protoAddr []byte) (hwAddr []byte, err error) { + e := h.cache.Lookup(protoAddr) + if e == nil { + return nil, errQueryNotFound + } else if e.flags.hasAny(eflagIncomplete) { + return nil, errQueryPending } - return nil, errQueryNotFound + return e.mac[:], nil } -func (h *Handler) DiscardQuery(protoAddr []byte) error { - for i := range h.queries { - q := &h.queries[i] - if internal.BytesEqual(protoAddr, q.protoaddr) { - q.destroy() - return nil - } +// CacheRemove cancels a pending query or evicts a cached entry for protoAddr. +func (h *Handler) CacheRemove(protoAddr []byte) error { + e := h.cache.Lookup(protoAddr) + if e == nil { + return errQueryNotFound } - return errQueryNotFound -} - -func (h *Handler) compactQueries() { - validOff := 0 - maxIdx := -1 - maxInc := uint16(0) - for i := 0; i < len(h.queries); i++ { - discard := h.queries[i].isInvalid() - if !discard { - h.queries[i].inc++ - if h.queries[i].inc > maxInc { - maxInc = h.queries[i].inc - maxIdx = i - } - if i != validOff { - // We swap the queries here so that when `StartQuery` extends - // queries slice, we don't have sharing of the internal structures. - // An alternative would be to zero things, however that would incur - // an allocation cost. - h.queries[validOff], h.queries[i] = h.queries[i], h.queries[validOff] - } - validOff++ - } - } - if validOff == len(h.queries) && maxIdx >= 0 { - h.queries[maxIdx].destroy() // Destroy oldest query if unable to compact. - } - h.queries = h.queries[:validOff] + e.destroy() + return nil } // StartQuery queues a query to perform over ARP for the protocol address `proto`. -// The user can additionally specify an dstHWAddr to write query result to on completion. -// If dstHWAddr is nil then query still occurs but no external buffer is written on query completion. -// dstHWAddr must be zeroed out (invalid MAC). -func (h *Handler) StartQuery(dstHWAddr, proto []byte) error { - if len(h.queries) == cap(h.queries) { - h.compactQueries() - if len(h.queries) == cap(h.queries) { - return lneto.ErrExhausted // Should never fail. - } - } +// Use [Handler.SetOnResolveCallback] to asynchronously set an ARP request result. +func (h *Handler) StartQuery(proto []byte, triggerCallback bool) error { if len(proto) != len(h.ourProtoAddr) { return lneto.ErrMismatchLen - } else if dstHWAddr != nil && len(dstHWAddr) != len(h.ourHWAddr) { - return lneto.ErrMismatchLen - } else if dstHWAddr != nil && !internal.IsZeroed(dstHWAddr...) { - return lneto.ErrInvalidConfig } - q := internal.SliceReclaim(&h.queries) - *q = queryResult{ - protoaddr: append(q.protoaddr[:0], proto...), - hwaddr: q.hwaddr[:0], - dstHw: dstHWAddr, + e := h.cache.acquireNext() + e.use([6]byte{}, proto, eflagIncomplete|eflagIncompletePendingQuery|eflagPriority) + if triggerCallback { + e.flags |= eflagResolveTriggersCallback } return nil } -func (h *Handler) Encapsulate(carrierData []byte, offsetToIP, offsetToFrame int) (int, error) { +func (h *Handler) Encapsulate(carrierData []byte, _, offsetToFrame int) (int, error) { b := carrierData[offsetToFrame:] - n := h.expectSize() - if len(b) < n { - return 0, errShortARP + afrm, err := h.newframe(b) + if err != nil { + return 0, err } - if len(h.pendingResponse) > 0 { - // pop frame. - afrm, _ := NewFrame(h.pendingResponse[len(h.pendingResponse)-1][:]) - h.pendingResponse = h.pendingResponse[:len(h.pendingResponse)-1] - afrm.SetOperation(OpReply) - afrm.SwapTargetSender() - hwsender, _ := afrm.Sender() - copy(hwsender, h.ourHWAddr) - n := copy(b, afrm.Clip().RawData()) - tgt, _ := afrm.Target() - trySetEthernetDst(carrierData[:offsetToFrame], tgt) - return n, nil + op := OpReply + e := h.cache.getNextFlagged(eflagPendingResponse) // Prioritize responses. + if e == nil { + e = h.cache.getNextFlagged(eflagIncompletePendingQuery) + if e == nil { + return 0, nil // No action to perform + } + e.flags &^= eflagIncompletePendingQuery + op = OpRequest + } else { + e.flags &^= eflagPendingResponse } - for i := range h.queries { - if h.queries[i].isInvalid() || h.queries[i].querysent { - continue - } - h.queries[i].querysent = true - afrm, _ := NewFrame(b) - afrm.SetHardware(h.htype, uint8(len(h.ourHWAddr))) - afrm.SetProtocol(h.protoType, uint8(len(h.ourProtoAddr))) - afrm.SetOperation(OpRequest) - hwSender, protoSender := afrm.Sender() - copy(hwSender, h.ourHWAddr) - copy(protoSender, h.ourProtoAddr) - hwTarget, protoTarget := afrm.Target() - copy(protoTarget, h.queries[i].protoaddr) - for j := range hwTarget { - hwTarget[j] = 0 - } + // Write Request or Reply, depending on which entry we got. + n, err := e.put(b, h.ourProtoAddr, h.ourHWAddr, op) + if err != nil { + return 0, err + } + switch op { + case OpRequest: broadcast := ethernet.BroadcastAddr() trySetEthernetDst(carrierData[:offsetToFrame], broadcast[:]) - return n, nil + case OpReply: + tgt, _ := afrm.Target() + trySetEthernetDst(carrierData[:offsetToFrame], tgt) } - return 0, nil + return n, nil } func (h *Handler) Demux(ethFrame []byte, frameOffset int) error { - if len(h.pendingResponse) == cap(h.pendingResponse) { - return lneto.ErrExhausted - } - b := ethFrame[frameOffset:] - afrm, err := NewFrame(b) + afrm, err := h.newframe(b) if err != nil { return err } - var vld lneto.Validator - afrm.ValidateSize(&vld) - if vld.HasError() { - return vld.ErrPop() + afrm.ValidateSize(&h.vld) + if h.vld.HasError() { + return h.vld.ErrPop() } htype, hlen := afrm.Hardware() if htype != h.htype || int(hlen) != len(h.ourHWAddr) { @@ -254,35 +181,37 @@ func (h *Handler) Demux(ethFrame []byte, frameOffset int) error { if !internal.BytesEqual(protoaddr, h.ourProtoAddr) { return nil // Not for us. } - h.pendingResponse = h.pendingResponse[:len(h.pendingResponse)+1] // Extend pending buffer. - copy(h.pendingResponse[len(h.pendingResponse)-1][:], afrm.buf) // Set pending buffer. + hw, proto := afrm.Sender() + e := h.cache.acquireNext() + e.use([6]byte(hw), proto, eflagPendingResponse) case OpReply: hwaddr, protoaddr := afrm.Sender() - for i := range h.queries { - q := &h.queries[i] - mac := q.response() - if mac == nil && internal.BytesEqual(q.protoaddr, protoaddr) { - q.hwaddr = append(q.hwaddr, hwaddr...) - if q.dstHw != nil { - if !internal.IsZeroed(q.dstHw...) { - internal.LogAttrs(nil, slog.LevelError, "race-condition:ARP-reused-buffer") - } - // External write to user buffer. - // Copy data and free up this memory. - copy(q.dstHw, hwaddr) - q.inc = 10000 - } - return nil - } + e := h.cache.Lookup(protoaddr) + if e == nil { + return nil + } + copy(e.mac[:], hwaddr) + e.flags &^= eflagIncomplete | eflagIncompletePendingQuery + if e.flags.hasAny(eflagResolveTriggersCallback) && h.onresolve != nil { + h.onresolve(e.mac[:], protoaddr) } - default: return errARPUnsupported } return nil } +func (h *Handler) newframe(b []byte) (Frame, error) { + f, err := NewFrame(b) + if err != nil { + return f, err + } else if h.protoType == ethernet.TypeIPv6 && len(b) < sizeHeaderv6 { + return f, lneto.ErrMismatch + } + return f, nil +} + func trySetEthernetDst(ethFrame []byte, dst []byte) { if len(ethFrame) >= 14 { copy(ethFrame[:6], dst) diff --git a/internet/stack-ethernet.go b/internet/stack-ethernet.go index 1322691..78052f9 100644 --- a/internet/stack-ethernet.go +++ b/internet/stack-ethernet.go @@ -39,6 +39,8 @@ type StackEthernet struct { gwmac [6]byte mtu uint16 acceptMulticast bool + + onSend func(p []byte) // crcupdate set when crc32 has been configured to be appended. crcupdate func(crc uint32, p []byte) uint32 } @@ -63,6 +65,10 @@ func (ls *StackEthernet) HardwareAddr6() [6]byte { return ls.mac } +func (ls *StackEthernet) OnEncapsulate(cb func([]byte)) { + ls.onSend = cb +} + // Reset6 resets the stack with the given parameters. // // Deprecated: Use [StackEthernet.Configure] instead. @@ -194,6 +200,9 @@ func (ls *StackEthernet) Encapsulate(carrierData []byte, offsetToIP, offsetToFra dst[n] = 0 n++ } + if ls.onSend != nil { + ls.onSend(dst[:n]) + } if ls.crcupdate != nil { crc := ls.crcupdate(0, carrierData[offsetToFrame:offsetToFrame+n]) binary.LittleEndian.PutUint32(carrierData[offsetToFrame+n:], crc) diff --git a/x/xnet/stack-async.go b/x/xnet/stack-async.go index 54ba82d..f5f00ef 100644 --- a/x/xnet/stack-async.go +++ b/x/xnet/stack-async.go @@ -39,7 +39,7 @@ type StackAsync struct { dhcpUDP internet.StackUDPPort dhcp dhcpv4.Client dhcpResults DHCPResults - subnet netip.Prefix // Local subnet for ARP resolution. + arpt subnetTable dnsUDP internet.StackUDPPort dns dns.Client @@ -83,6 +83,9 @@ type StackConfig struct { // ICMPQueueLimit sets maximum number of input/output packets queued for processing. // If set to zero ICMP cannot be enabled on the stack. ICMPQueueLimit int + // PassivePeers limits how many subnet peers the stack passively learns MAC addresses for. + // Passively learned entries skip ARP round-trips on the first DialTCP/DialUDP to that peer. + PassivePeers int } func (s *StackAsync) Hostname() string { @@ -94,7 +97,11 @@ func (s *StackAsync) IngressEthernet(ethernetFrame []byte) error { s.mu.Lock() defer s.mu.Unlock() s.totalrecv += uint64(len(ethernetFrame)) - return s.link.Demux(ethernetFrame, 0) + err := s.link.Demux(ethernetFrame, 0) + if err == nil { + s.arpt.learnFromIngressEthernet(ethernetFrame) + } + return err } // EgressEthernet writes the next ethernet frame to send into dstEthernetFrame from the stack. @@ -137,7 +144,7 @@ func (s *StackAsync) MTU() int { } func (s *StackAsync) Reset(cfg StackConfig) error { - if cfg.RandSeed == 0 || cfg.Hostname == "" { + if cfg.RandSeed == 0 || cfg.Hostname == "" || cfg.PassivePeers > 255 { return lneto.ErrInvalidConfig } mac := cfg.HardwareAddress @@ -165,12 +172,18 @@ func (s *StackAsync) Reset(cfg StackConfig) error { return err } s.link.SetAcceptMulticast(cfg.AcceptMulticast) + if cfg.PassivePeers == 0 { + s.link.OnEncapsulate(nil) + } else { + s.link.OnEncapsulate(s.arpt.patchEgressMAC) + } const ipNodes = 3 // 3 IP protocols possible: UDP, TCP, ICMP. err = s.ip.Reset(addr, ipNodes) if err != nil { return err } s.ip.SetAcceptMulticast(cfg.AcceptMulticast) + s.arpt.passivePeers = uint8(cfg.PassivePeers) err = s.resetARP() if err != nil { return err @@ -241,14 +254,16 @@ func (s *StackAsync) resetARP() error { err := s.arp.Reset(arp.HandlerConfig{ HardwareAddr: mac[:], ProtocolAddr: addr.AsSlice(), - MaxQueries: 3, - MaxPending: 3, + MaxQueries: 5, + MaxPending: 5, HardwareType: 1, ProtocolType: proto, }) if err != nil { return err } + s.arpt.reset(10, s.arpt.passivePeers) + s.arp.SetOnResolveCallback(s.arpt.onResolve) err = s.link.Register(&s.arp) if err != nil { return err @@ -308,7 +323,7 @@ func (s *StackAsync) Addr() netip.Addr { func (s *StackAsync) SetSubnet(subnetMask netip.Prefix) { s.mu.Lock() defer s.mu.Unlock() - s.subnet = subnetMask + s.arpt.subnet = subnetMask } func (s *StackAsync) SetHardwareAddress(hw [6]byte) error { @@ -358,10 +373,10 @@ func (s *StackAsync) DialUDP(conn *udp.Conn, localPort uint16, addrp netip.AddrP s.mu.Lock() defer s.mu.Unlock() var mac []byte - if s.subnet.Contains(addrp.Addr()) { + if s.arpt.subnet.Contains(addrp.Addr()) { mac = make([]byte, 6) ip := addrp.Addr().As4() - hw, err := s.arp.QueryResult(ip[:]) + hw, err := s.arp.CacheLookup(ip[:]) if err == nil { // MAC already contained in results. copy(mac, hw) @@ -369,7 +384,7 @@ func (s *StackAsync) DialUDP(conn *udp.Conn, localPort uint16, addrp netip.AddrP // StartQuery starts an ARP query for addresses in this network. // On finishing query MAC is set and thus the StackPort will allow encapsulating // data on that connection. - err = s.arp.StartQuery(mac, ip[:]) + err = s.arpt.startQuery(mac, ip[:], &s.arp) if err != nil { return err } @@ -387,9 +402,9 @@ func (s *StackAsync) DialTCP(conn *tcp.Conn, localPort uint16, addrp netip.AddrP s.mu.Lock() defer s.mu.Unlock() var mac []byte - if s.subnet.Contains(addrp.Addr()) { + if s.arpt.subnet.Contains(addrp.Addr()) { ip := addrp.Addr().As4() - hw, err := s.arp.QueryResult(ip[:]) + hw, err := s.arp.CacheLookup(ip[:]) mac = make([]byte, 6) if err == nil { // Query exists, use pre-existing result. @@ -398,7 +413,7 @@ func (s *StackAsync) DialTCP(conn *tcp.Conn, localPort uint16, addrp netip.AddrP // StartQuery starts an ARP query for addresses in this network. // On finishing query MAC is set and thus the StackPort will allow encapsulating // data on that connection. - err = s.arp.StartQuery(mac, ip[:]) + err = s.arpt.startQuery(mac, ip[:], &s.arp) if err != nil { return err } @@ -576,7 +591,7 @@ func (s *StackAsync) StartResolveHardwareAddress6(ip netip.Addr) error { return lneto.ErrUnsupported } addr := ip.As4() - return s.arp.StartQuery(nil, addr[:]) + return s.arp.StartQuery(addr[:], false) } // ResultResolveHardwareAddress6 @@ -587,7 +602,7 @@ func (s *StackAsync) ResultResolveHardwareAddress6(ip netip.Addr) (hw [6]byte, e return hw, lneto.ErrUnsupported } addr := ip.As4() - hwslice, err := s.arp.QueryResult(addr[:]) + hwslice, err := s.arp.CacheLookup(addr[:]) if err != nil { return hw, err } else if len(hwslice) != 6 { @@ -604,7 +619,7 @@ func (s *StackAsync) DiscardResolveHardwareAddress6(ip netip.Addr) error { return lneto.ErrUnsupported } addr := ip.As4() - return s.arp.DiscardQuery(addr[:]) + return s.arp.CacheRemove(addr[:]) } type DHCPResults struct { @@ -648,7 +663,7 @@ func (stack *StackAsync) AssimilateDHCPResults(results *DHCPResults) error { stack.mu.Lock() defer stack.mu.Unlock() if results.Subnet.IsValid() { - stack.subnet = results.Subnet + stack.arpt.subnet = results.Subnet } if results.AssignedAddr.IsValid() { err := stack.setIPAddr(results.AssignedAddr) diff --git a/x/xnet/stack-blocking.go b/x/xnet/stack-blocking.go index dbcca52..990e211 100644 --- a/x/xnet/stack-blocking.go +++ b/x/xnet/stack-blocking.go @@ -140,7 +140,7 @@ func (s StackBlocking) DoResolveHardwareAddress6(addr netip.Addr, timeout time.D err = errDeadlineExceed // Ensure that if iterations done error is returned. } ip4 := addr.As4() - s.async.arp.DiscardQuery(ip4[:]) + s.async.arp.CacheRemove(ip4[:]) return hw, err } diff --git a/x/xnet/subnet-table.go b/x/xnet/subnet-table.go new file mode 100644 index 0000000..8981dc5 --- /dev/null +++ b/x/xnet/subnet-table.go @@ -0,0 +1,139 @@ +package xnet + +import ( + "encoding/binary" + "net/netip" + + "github.com/soypat/lneto/arp" + "github.com/soypat/lneto/ethernet" + "github.com/soypat/lneto/internal" +) + +// subnetTable manages both passively learned peer MAC/IP tuples and in-flight async ARP resolves. +// +// Layout of resolves slice: +// +// [0 : passivePeers] — owned MAC+IP, permanently retained (learned passively from ingress) +// [passivePeers : len] — externally-owned MAC, evicted by age (pending ARP queries) +type subnetTable struct { + subnet netip.Prefix + resolves []struct { + mac []byte // externally owned for pending entries; owned for passive entries. + ip []byte // always owned by this struct. + age uint16 + } + passivePeers uint8 +} + +func (a *subnetTable) reset(arpentries int, passivePeers uint8) { + a.passivePeers = passivePeers + if a.resolves == nil { + internal.SliceReuse(&a.resolves, arpentries+int(passivePeers)) + a.resolves = a.resolves[:cap(a.resolves)] + } +} + +func (a *subnetTable) learnFromIngressEthernet(ethernetFrame []byte) { + if len(ethernetFrame) > 14+20 && + binary.BigEndian.Uint16(ethernetFrame[12:14]) == uint16(ethernet.TypeIPv4) { + src, _, _, _, _ := internal.GetIPAddr(ethernetFrame[14:]) + a.learnPassive(src, ethernetFrame[6:12]) + } +} + +// learnPassive stores or updates a passively observed MAC/IP tuple in the reserved slots. +// It is a no-op if passivePeers is zero, src is not in the local subnet, or all slots are taken. +func (a *subnetTable) learnPassive(src, mac []byte) { + if a.passivePeers == 0 { + return + } + addr, _ := netip.AddrFromSlice(src) + if !a.subnet.Contains(addr) { + return + } + for i := range a.passivePeers { + v := &a.resolves[i] + if internal.BytesEqual(v.ip, src) { + copy(v.mac, mac) // update in case MAC changed (e.g. NIC swap) + return + } + if len(v.ip) == 0 { + v.ip = append(v.ip, src...) + v.mac = append(v.mac, mac...) + return + } + } +} + +// startQuery copies the MAC into mac immediately if the IP was passively learned, +// otherwise issues an ARP query via h and registers mac as the externally-owned destination. +func (a *subnetTable) startQuery(mac, ip []byte, h *arp.Handler) error { + for i := range a.passivePeers { + v := &a.resolves[i] + if internal.BytesEqual(v.ip, ip) { + copy(mac, v.mac) + return nil + } + } + if err := h.StartQuery(ip, true); err != nil { + return err + } + n := int(a.passivePeers) + oldest := n + for i := n; i < len(a.resolves); i++ { + v := &a.resolves[i] + if len(v.mac) == 0 { + oldest = i + break + } else if v.age > a.resolves[oldest].age { + oldest = i + } + } + for i := n; i < len(a.resolves); i++ { + a.resolves[i].age++ + } + v := &a.resolves[oldest] + v.mac = mac + v.ip = append(v.ip[:0], ip...) + v.age = 0 + return nil +} + +// onResolve is the arp.Handler resolve callback; called when an ARP response arrives. +func (a *subnetTable) onResolve(mac, ip []byte) { + for i := int(a.passivePeers); i < len(a.resolves); i++ { + v := &a.resolves[i] + if internal.BytesEqual(ip, v.ip) { + copy(v.mac, mac) + v.mac = nil + v.ip = v.ip[:0] + return + } + } +} + +// patchEgressMAC is registered as the OnEncapsulate callback on StackEthernet. +// It runs after the payload is written but before CRC is appended, so the CRC +// covers the corrected destination MAC. +func (a *subnetTable) patchEgressMAC(frame []byte) { + if a.passivePeers == 0 || len(frame) < 14+20 || + binary.BigEndian.Uint16(frame[12:14]) != uint16(ethernet.TypeIPv4) { + return + } + efrm, _ := ethernet.NewFrame(frame) + if efrm.IsBroadcast() { + return // broadcast stays broadcast (e.g. DHCP discover). + } + // Server-side connections have no registered MAC; fill from passively learned entries. + _, dstIP, _, _, err := internal.GetIPAddr(frame[14:]) + if err != nil { + return + } + for i := range a.passivePeers { + v := &a.resolves[i] + if internal.BytesEqual(v.ip, dstIP) { + *efrm.DestinationHardwareAddr() = [6]byte(v.mac) + return + } + } +} diff --git a/x/xnet/xnet_arp_test.go b/x/xnet/xnet_arp_test.go index 3dbc1b5..6f94f7b 100644 --- a/x/xnet/xnet_arp_test.go +++ b/x/xnet/xnet_arp_test.go @@ -42,7 +42,7 @@ func TestARPLocal(t *testing.T) { tst := testerFrom(t, mtu) _ = tst tst.ARPExchangeOnly(s1, s2) - hwaddr, err := s1.arp.QueryResult(addr2.Addr().AsSlice()) + hwaddr, err := s1.arp.CacheLookup(addr2.Addr().AsSlice()) if err != nil { t.Fatal(err) } else if !bytes.Equal(hwaddr[:], hw2[:]) { diff --git a/x/xnet/xnet_fuzz_test.go b/x/xnet/xnet_fuzz_test.go index 12d158e..4675681 100644 --- a/x/xnet/xnet_fuzz_test.go +++ b/x/xnet/xnet_fuzz_test.go @@ -283,6 +283,7 @@ func testStackSeeded(t *testing.T, seed1, seed2 int64) { MTU: mtu, HardwareAddress: [6]byte{0x1, 0, 0, 0, 0, v1}, AcceptMulticast: v1%2 == 0, + PassivePeers: 1, } err := s1.Reset(cfg1) if err != nil { @@ -298,6 +299,7 @@ func testStackSeeded(t *testing.T, seed1, seed2 int64) { MTU: mtu, HardwareAddress: [6]byte{0x2, 0, 0, 0, 0, v2}, AcceptMulticast: v2%2 == 0, + PassivePeers: 1, } err = s2.Reset(cfg2) if err != nil { diff --git a/x/xnet/xnet_subnettable_test.go b/x/xnet/xnet_subnettable_test.go new file mode 100644 index 0000000..a358b29 --- /dev/null +++ b/x/xnet/xnet_subnettable_test.go @@ -0,0 +1,155 @@ +package xnet + +import ( + "encoding/binary" + "net/netip" + "testing" + + "github.com/soypat/lneto/ethernet" + "github.com/soypat/lneto/tcp" +) + +// TestSubnetTable_PatchEgressMAC_WhenGatewayMAC captures the bug where patchEgressMAC +// returns early when the Ethernet dst is a gateway MAC (not broadcast), so it never +// patches the destination to the passively-learned client MAC. +func TestSubnetTable_PatchEgressMAC_WhenGatewayMAC(t *testing.T) { + clientIP := [4]byte{10, 0, 0, 1} + clientMAC := [6]byte{0xAA, 0xBB, 0xCC, 0xDD, 0xEE, 0x01} + serverIP := [4]byte{10, 0, 0, 2} + serverMAC := [6]byte{0xAA, 0xBB, 0xCC, 0xDD, 0xEE, 0x02} + gatewayMAC := [6]byte{0xAA, 0xBB, 0xCC, 0xDD, 0xEE, 0xFF} // separate from client + + var st subnetTable + st.reset(4, 2) + st.subnet = netip.MustParsePrefix("10.0.0.0/24") + + // Learn client MAC from a simulated ingress frame (client→server SYN). + ingressFrame := makeMinimalIPv4Frame(serverMAC, clientMAC, clientIP, serverIP) + st.learnFromIngressEthernet(ingressFrame) + + // Simulate egress SYN-ACK: stack uses gateway MAC as Ethernet dst (the bug). + egressFrame := makeMinimalIPv4Frame(gatewayMAC, serverMAC, serverIP, clientIP) + st.patchEgressMAC(egressFrame) + + gotDst := [6]byte(egressFrame[0:6]) + if gotDst != clientMAC { + t.Errorf("patchEgressMAC did not fix Ethernet dst:\n got %x (gateway MAC)\n want %x (client MAC)", gotDst, clientMAC) + } +} + +// TestStackAsync_ListenerSynAckAddressedToClient mirrors the ESP32 hotspot scenario: +// server's gateway is a router (not the client), so the SYN-ACK must use the +// passively-learned client MAC, not the router/gateway MAC. +func TestStackAsync_ListenerSynAckAddressedToClient(t *testing.T) { + const mtu = ethernet.MaxMTU + const svPort = 80 + + clientMAC := [6]byte{0xAA, 0xBB, 0xCC, 0xDD, 0xEE, 0x01} + serverMAC := [6]byte{0xAA, 0xBB, 0xCC, 0xDD, 0xEE, 0x02} + routerMAC := [6]byte{0xAA, 0xBB, 0xCC, 0xDD, 0xEE, 0xFF} // third party — not client + + // Server: gateway = router (not client), but passively learns client MAC from SYN. + var sv StackAsync + err := sv.Reset(StackConfig{ + Hostname: "Server1", + RandSeed: 1234, + StaticAddress: netip.AddrFrom4([4]byte{10, 0, 0, 2}), + MaxActiveTCPPorts: 1, + HardwareAddress: serverMAC, + MTU: mtu, + PassivePeers: 2, + }) + if err != nil { + t.Fatal(err) + } + sv.SetGateway6(routerMAC) + sv.SetSubnet(netip.MustParsePrefix("10.0.0.0/24")) + + pool, err := NewTCPPool(TCPPoolConfig{ + PoolSize: 1, + QueueSize: 4, + TxBufSize: mtu, + RxBufSize: mtu, + EstablishedTimeout: 10e9, + ClosingTimeout: 10e9, + }) + if err != nil { + t.Fatal(err) + } + var listener tcp.Listener + if err = listener.Reset(svPort, pool); err != nil { + t.Fatal(err) + } + if err = sv.RegisterListener(&listener); err != nil { + t.Fatal(err) + } + + // Client: gateway = server MAC (direct L2 path, as in a hotspot WLAN). + var client StackAsync + err = client.Reset(StackConfig{ + Hostname: "Client1", + RandSeed: 5678, + StaticAddress: netip.AddrFrom4([4]byte{10, 0, 0, 1}), + MaxActiveTCPPorts: 1, + HardwareAddress: clientMAC, + MTU: mtu, + }) + if err != nil { + t.Fatal(err) + } + client.SetGateway6(serverMAC) + + var clConn tcp.Conn + if err = clConn.Configure(tcp.ConnConfig{ + RxBuf: make([]byte, mtu), TxBuf: make([]byte, mtu), + TxPacketQueueSize: 4, + }); err != nil { + t.Fatal(err) + } + if err = client.DialTCP(&clConn, 54321, netip.AddrPortFrom(sv.Addr(), svPort)); err != nil { + t.Fatal(err) + } + + buf := make([]byte, mtu+ethernet.MaxOverheadSize) + + // Step 1: client egresses SYN. + n, err := client.EgressEthernet(buf) + if err != nil || n == 0 { + t.Fatalf("client egress SYN: n=%d err=%v", n, err) + } + synDst := [6]byte(buf[0:6]) + if synDst != serverMAC { + t.Fatalf("SYN Ethernet dst wrong: got %x, want server %x", synDst, serverMAC) + } + + // Step 2: server ingresses SYN — passively learns client MAC. + if err = sv.IngressEthernet(buf[:n]); err != nil { + t.Fatalf("server ingress SYN: %v", err) + } + + // Step 3: server egresses SYN-ACK — must be addressed to client, not router. + clear(buf) + n, err = sv.EgressEthernet(buf) + if err != nil || n == 0 { + t.Fatalf("server egress SYN-ACK: n=%d err=%v", n, err) + } + + synackDst := [6]byte(buf[0:6]) + if synackDst != clientMAC { + t.Errorf("SYN-ACK Ethernet dst wrong:\n got %x\n want %x (client MAC)\n note: %x is router MAC", synackDst, clientMAC, routerMAC) + } +} + +// makeMinimalIPv4Frame builds a 35-byte Ethernet+IPv4 frame (no payload, 1 padding byte). +// This is the minimum size that passes both learnFromIngressEthernet (>34) and patchEgressMAC (>=34) checks. +func makeMinimalIPv4Frame(dstMAC, srcMAC [6]byte, srcIP, dstIP [4]byte) []byte { + frame := make([]byte, 35) + copy(frame[0:6], dstMAC[:]) + copy(frame[6:12], srcMAC[:]) + binary.BigEndian.PutUint16(frame[12:14], uint16(ethernet.TypeIPv4)) + frame[14] = 0x45 // IPv4, IHL=5 + frame[22] = 64 // TTL + copy(frame[26:30], srcIP[:]) + copy(frame[30:34], dstIP[:]) + return frame +}