diff --git a/definitions.go b/definitions.go index 8b774da..9ad6b75 100644 --- a/definitions.go +++ b/definitions.go @@ -1,5 +1,42 @@ package lneto +// StackNode is an abstraction of a packet exchanging protocol controller. This is the building block for all protocols, +// from Ethernet to IP to TCP, practically any protocol can be expressed as a StackNode and function completely. +// Today protocols represented by StackNode also include NTP, DNS, DHCP, ARP, ICMP, UDP, mDNS. +// Do note stream based protocols like HTTP are NOT well represented with a StackNode. +type StackNode interface { + // Encapsulate writes the stack node's frame into carrierData[offsetToFrame:] + // along with any other frame or payload the stack node encapsulates. + // The returned integer is amount of bytes written such that carrierData[offsetToFrame:offsetToFrame+n] + // contains written data. Data inside carrierData[:offsetToFrame] usually contains data necessary for + // a StackNode to correctly emit valid frame data: such is the case for TCP packets which require IP + // frame data for checksum calculation. Thus StackNodes must provide fields in their own frame + // required by sub-stacknodes for correct encapsulation; in the case of IPv4/6 this means including fields + // used in pseudo-header checksum like local IP (see [ipv4.CRCWriteUDPPseudo]). + // + // offsetToIP is the offset to the IP frame, if present, else its value should be -1. + // The relation offsetToIP<=offsetToFrame MUST hold. + // + // When [net.ErrClosed] is returned the StackNode should be discarded and any written data passed up normally. + // Errors returned by Encapsulate are "extraordinary" and should not be returned unless the StackNode is receiving invalid carrierData/frameOffset. + Encapsulate(carrierData []byte, offsetToIP, offsetToFrame int) (int, error) + // Demux reads from the argument buffer where frameOffset is the offset of this StackNode's frame first byte. + // The stack node then dispatches(demuxes) the encapsulated frames to its corresponding substack or subnode. + Demux(carrierData []byte, frameOffset int) error + // LocalPort returns the local port of this StackNode or zero if not set/relevant. + LocalPort() uint16 + // Protocol returns a number identifying the protocol used by this [StackNode]. + // Can be an [IPProto] among other types of protocols, i.e: ethernet.Protocol for a link layer [StackNode]. + Protocol() uint64 + // ConnectionID returns the pointer to the connection context number or ConnectionID. + // Stacks should store the original value of ConnectionID (dereference the pointer) on + // registering a [StackNode]. When the value changes this means the registered [StackNode] + // should be discarded since its lifetime has terminated. + ConnectionID() *uint64 + // TODO(pato,ddirect): Do we eventually want to trigger writes to buffers asynchronously? + // SetFlagPending(flagPending func(numPendingEncapsulations int)) +} + //go:generate stringer -type=IPProto,errGeneric -linecomment -output stringers.go . // IPProto represents the IP protocol number. diff --git a/errors.go b/errors.go index 1144087..43bcae0 100644 --- a/errors.go +++ b/errors.go @@ -20,6 +20,13 @@ const ( ErrInvalidField // invalid field ErrInvalidLengthField // invalid length field ErrExhausted // resource exhausted + ErrAlreadyRegistered // protocol already registered + + // Below are potentially good future error additions + // based on one or two encountered use cases, example use case included. + /* + - ErrUnregistered/ErrAborted // connection unregistered. i.e: ICMP client aborted during active ping, ping process returns this. + */ ) func (err errGeneric) Error() string { diff --git a/examples/berkeley-listener/berkeley_server.go b/examples/berkeley-listener/berkeley_server.go index 383bc60..5d072a7 100644 --- a/examples/berkeley-listener/berkeley_server.go +++ b/examples/berkeley-listener/berkeley_server.go @@ -145,7 +145,7 @@ func run() error { } for { - nwrite, err := stack.Encapsulate(buf[:], -1, 0) + nwrite, err := stack.EgressEthernet(buf[:]) if err != nil { log.Println("ERR:ENCAPSULATE", err) } else if nwrite > 0 { @@ -159,13 +159,13 @@ func run() error { log.Fatalf("mismatch written bytes %d!=%d", nwrite, n) } if flagMockClient && mockStack.Addr().IsValid() { - mockStack.Demux(buf[:nwrite], 0) + mockStack.IngressEthernet(buf[:nwrite]) } } if flagMockClient && mockStack.Addr().IsValid() { - n, _ := mockStack.Encapsulate(buf[:], -1, 0) + n, _ := mockStack.EgressEthernet(buf[:]) if n > 0 { - stack.Demux(buf[:n], 0) + stack.IngressEthernet(buf[:n]) } } @@ -181,7 +181,7 @@ func run() error { if err != nil { log.Fatal("goroutine read:", err) } else if nread > 0 { - err = stack.Demux(buf[:nread], 0) + err = stack.IngressEthernet(buf[:nread]) if !errors.Is(err, lneto.ErrPacketDrop) { if err = logFrames("IN", buf[:nread]); err != nil { log.Println("ERR:INLOG", err) diff --git a/examples/httpserver/main.go b/examples/httpserver/main.go index 3ad6fe0..449a7d3 100644 --- a/examples/httpserver/main.go +++ b/examples/httpserver/main.go @@ -144,7 +144,7 @@ func run() (err error) { return err } for { - nwrite, err := stack.Encapsulate(buf[:], -1, 0) + nwrite, err := stack.EgressEthernet(buf[:]) if err != nil { log.Println("ERR:ENCAPSULATE", err) } else if nwrite > 0 { @@ -172,7 +172,7 @@ func run() (err error) { if err != nil { log.Fatal("goroutine read:", err) } else if nread > 0 { - err = stack.Demux(buf[:nread], 0) + err = stack.IngressEthernet(buf[:nread]) if !errors.Is(err, lneto.ErrPacketDrop) { err = logFrames("IN", buf[:nread]) if err != nil { diff --git a/examples/min-working-example/main-mwe.go b/examples/min-working-example/main-mwe.go index 72886ce..39858b8 100644 --- a/examples/min-working-example/main-mwe.go +++ b/examples/min-working-example/main-mwe.go @@ -142,7 +142,7 @@ func stackLoop(ctx context.Context, stack *xnet.StackAsync) { frameLength, _ := network.MaxFrameLength() buf := make([]byte, frameLength) for ctx.Err() == nil { - nwrite, err := stack.Encapsulate(buf[:], -1, 0) + nwrite, err := stack.EgressEthernet(buf[:]) if err != nil { fmt.Println("encaps err:", err) } else if nwrite > 0 { @@ -153,7 +153,7 @@ func stackLoop(ctx context.Context, stack *xnet.StackAsync) { if err != nil { fmt.Println("network read err:", err) } else if nread > 0 { - err = stack.Demux(buf[:nread], 0) + err = stack.IngressEthernet(buf[:nread]) if err != nil && err != lneto.ErrPacketDrop { fmt.Println("demux err:", err) } else { diff --git a/examples/xcurl/main.go b/examples/xcurl/main.go index 35825d9..c37bc77 100644 --- a/examples/xcurl/main.go +++ b/examples/xcurl/main.go @@ -164,7 +164,7 @@ func run() (err error) { return err } for { - nwrite, err := stack.Encapsulate(buf[:], -1, 0) + nwrite, err := stack.EgressEthernet(buf[:]) if err != nil { log.Println("ERR:ENCAPSULATE", err) } else if nwrite > 0 { @@ -193,7 +193,7 @@ func run() (err error) { if err != nil { log.Fatal("groutine read:", err) } else if nread > 0 { - err = stack.Demux(buf[:nread], 0) + err = stack.IngressEthernet(buf[:nread]) if !errors.Is(err, lneto.ErrPacketDrop) { // Only skip logging packet in case of dropped packet. err = logFrames("IN", buf[:nread]) diff --git a/internal/slices.go b/internal/slices.go index 5a49172..06e3d40 100644 --- a/internal/slices.go +++ b/internal/slices.go @@ -79,3 +79,11 @@ func BytesEqual(a, b []byte) bool { } return unsafe.String(&a[0], len(a)) == unsafe.String(&b[0], len(b)) } + +func SliceDequeueFront[T any](a *[]T) T { + s := *a + v := s[0] + n := copy(s, s[1:]) + *a = s[:n] + return v +} diff --git a/internet/definitions.go b/internet/definitions.go index afd361c..9fe7256 100644 --- a/internet/definitions.go +++ b/internet/definitions.go @@ -1,7 +1,6 @@ package internet import ( - "errors" "log/slog" "math" "net" @@ -10,34 +9,6 @@ import ( "github.com/soypat/lneto" ) -// StackNode is an abstraction of a packet exchanging protocol controller. This is the building block for all protocols, -// from Ethernet to IP to TCP, practically any protocol can be expressed as a StackNode and function completely. -type StackNode interface { - // Encapsulate writes the stack node's frame into carrierData[offsetToFrame:] - // along with any other frame or payload the stack node encapsulates. - // The returned integer is amount of bytes written such that carrierData[offsetToFrame:offsetToFrame+n] - // contains written data. Data inside carrierData[:offsetToFrame] usually contains data necessary for - // a StackNode to correctly emit valid frame data: such is the case for TCP packets which require IP - // frame data for checksum calculation. Thus StackNodes must provide fields in their own frame - // required by sub-stacknodes for correct encapsulation; in the case of IPv4/6 this means including fields - // used in pseudo-header checksum like local IP (see [ipv4.CRCWriteUDPPseudo]). - // - // offsetToIP is the offset to the IP frame, if present, else its value should be -1. - // The relation offsetToIP<=offsetToFrame should always hold. - // - // When [net.ErrClosed] is returned the StackNode should be discarded and any written data passed up normally. - // Errors returned by Encapsulate are "extraordinary" and should not be returned unless the StackNode is receiving invalid carrierData/frameOffset. - Encapsulate(carrierData []byte, offsetToIP, offsetToFrame int) (int, error) - // Demux reads from the argument buffer where frameOffset is the offset of this StackNode's frame first byte. - // The stack node then dispatches(demuxes) the encapsulated frames to its corresponding sub-stack-node(s). - Demux(carrierData []byte, frameOffset int) error - LocalPort() uint16 - Protocol() uint64 - // Connect - ConnectionID() *uint64 - // SetFlagPending(flagPending func(numPendingEncapsulations int)) -} - // node is a concrete StackNode as stored in Stacks. Methods are devirtualized for performance benefits, especially on TinyGo. type node struct { currConnID uint64 @@ -71,7 +42,7 @@ func (h *handlers) registerByProto(n node) error { return err } if h.nodeByProto(n.proto) != nil { - return errProtoRegistered + return lneto.ErrAlreadyRegistered } h.nodes = append(h.nodes, n) return nil @@ -83,7 +54,7 @@ func (h *handlers) registerByPortProto(n node) error { return err } if h.nodeByPortProto(n.port, n.proto) != nil { - return errProtoRegistered + return lneto.ErrAlreadyRegistered } h.nodes = append(h.nodes, n) return nil @@ -200,8 +171,7 @@ func (h *handlers) encapsulateAny(buf []byte, offsetIP, offsetThisFrame int) (_ } var ( - errProtoRegistered = errors.New("protocol already registered") - _ = net.ErrClosed + _ = net.ErrClosed ) func (node *node) IsInvalid() bool { @@ -212,7 +182,7 @@ func checkNodeErr(node *node, err error) (discard bool) { return node.IsInvalid() || (err != nil && err == net.ErrClosed) } -func nodeFromStackNode(s StackNode, port uint16, protocol uint64, remoteAddr []byte) node { +func nodeFromStackNode(s lneto.StackNode, port uint16, protocol uint64, remoteAddr []byte) node { if protocol > math.MaxUint16 { panic(">16bit protocol number unsupported") } diff --git a/internet/definitions_go.go b/internet/definitions_go.go index f84e843..da27aa5 100644 --- a/internet/definitions_go.go +++ b/internet/definitions_go.go @@ -2,7 +2,9 @@ package internet -func makecbnode(s StackNode) cbnode { +import "github.com/soypat/lneto" + +func makecbnode(s lneto.StackNode) cbnode { return cbnode{ _s: s, } @@ -10,7 +12,7 @@ func makecbnode(s StackNode) cbnode { type cbnode struct { // Do not access outside of handlers/node logic. - _s StackNode + _s lneto.StackNode } func (s cbnode) Encapsulate(carrierData []byte, offsetToIP, offsetToFrame int) (int, error) { diff --git a/internet/stack-ethernet.go b/internet/stack-ethernet.go index b45b2fd..1985528 100644 --- a/internet/stack-ethernet.go +++ b/internet/stack-ethernet.go @@ -101,6 +101,19 @@ func (ls *StackEthernet) Configure(cfg StackEthernetConfig) error { return nil } +// MaxFrameLength returns the maximum ethernet frame length in bytes, which is the MTU plus the Ethernet header (14 bytes) and CRC (4 bytes if enabled). +// This is the maximum size of an Ethernet frame that can be sent from the stack. +func (ls *StackEthernet) MaxFrameLength() int { + base := int(ls.mtu) + 14 + if ls.crcupdate != nil { + base += 4 + } + return base +} + +// MTU is the Maximum Transmission Unit of the stack corresponding +// to the maximum payload size of an ethernet frame that can be sent through the stack. +// Important to note that the actual ethernet frame size is MTU + Ethernet header (14) + CRC (4 if enabled), this is known as the Maximum Frame Length. func (ls *StackEthernet) MTU() int { return int(ls.mtu) } func (ls *StackEthernet) ConnectionID() *uint64 { return &ls.connID } @@ -109,7 +122,7 @@ func (ls *StackEthernet) LocalPort() uint16 { return 0 } func (ls *StackEthernet) Protocol() uint64 { return 1 } -func (ls *StackEthernet) Register(h StackNode) error { +func (ls *StackEthernet) Register(h lneto.StackNode) error { proto := h.Protocol() if proto > math.MaxUint16 || proto <= 1500 { return lneto.ErrInvalidConfig diff --git a/internet/stack-ip.go b/internet/stack-ip.go index c67966f..645735a 100644 --- a/internet/stack-ip.go +++ b/internet/stack-ip.go @@ -13,7 +13,7 @@ import ( "github.com/soypat/lneto/udp" ) -var _ StackNode = (*StackIP)(nil) +var _ lneto.StackNode = (*StackIP)(nil) type StackIP struct { connID uint64 @@ -205,7 +205,7 @@ func (sb *StackIP) Encapsulate(carrierData []byte, offsetToIP, offsetToFrame int return totalLen, err } -func (sb *StackIP) Register(h StackNode) error { +func (sb *StackIP) Register(h lneto.StackNode) error { proto := h.Protocol() if proto > 255 { return lneto.ErrInvalidConfig @@ -213,6 +213,10 @@ func (sb *StackIP) Register(h StackNode) error { return sb.handlers.registerByPortProto(nodeFromStackNode(h, h.LocalPort(), proto, nil)) } +func (sb *StackIP) IsRegistered(proto lneto.IPProto) bool { + return sb.handlers.nodeByProto(uint16(proto)) != nil +} + func (sb *StackIP) recvicmp(icmpData []byte) error { var crc lneto.CRC791 if crc.PayloadSum16(icmpData) != 0 { diff --git a/internet/stack-ports.go b/internet/stack-ports.go index 3e42cde..7c54cc6 100644 --- a/internet/stack-ports.go +++ b/internet/stack-ports.go @@ -87,7 +87,7 @@ func (ps *StackPorts) Demux(b []byte, offset int) (err error) { // Register registers a port StackNode on StackPorts. // If dstMAC is set to non-nil, length six buffer then -func (ps *StackPorts) Register(h StackNode) error { +func (ps *StackPorts) Register(h lneto.StackNode) error { port := h.LocalPort() proto := h.Protocol() if port <= 0 { @@ -105,7 +105,7 @@ type StackPortsMACFiltered struct { sp StackPorts } -func (mfsp *StackPortsMACFiltered) Register(h StackNode, addr []byte) error { +func (mfsp *StackPortsMACFiltered) Register(h lneto.StackNode, addr []byte) error { port := h.LocalPort() proto := h.Protocol() if port <= 0 { diff --git a/internet/stack-udpport.go b/internet/stack-udpport.go index e8a9bc4..f02ba98 100644 --- a/internet/stack-udpport.go +++ b/internet/stack-udpport.go @@ -16,7 +16,7 @@ type StackUDPPort struct { raddr []byte } -func (sudp *StackUDPPort) SetStackNode(node StackNode, raddr []byte, rmport uint16) { +func (sudp *StackUDPPort) SetStackNode(node lneto.StackNode, raddr []byte, rmport uint16) { sudp.h = nodeFromStackNode(node, node.LocalPort(), node.Protocol(), raddr) sudp.rmport = rmport sudp.raddr = append(sudp.raddr[:0], raddr...) diff --git a/ipv4/icmpv4/client.go b/ipv4/icmpv4/client.go new file mode 100644 index 0000000..342419a --- /dev/null +++ b/ipv4/icmpv4/client.go @@ -0,0 +1,264 @@ +package icmpv4 + +import ( + "slices" + + "github.com/soypat/lneto" + "github.com/soypat/lneto/internal" +) + +var _ lneto.StackNode = (*Client)(nil) // Compile-time guarantee of interface implementation. + +const ( + keyHashCompletedBit = 1 << 31 + keyHashSentBit = 1 << 30 + keyHashBits = (1 << 30) - 1 +) + +type Client struct { + connid uint64 + magic uint32 + _seq uint16 + id uint16 + + outgoingEcho []struct { + // For every ping we send out stores hashes of the data (should include IP likely). + pattern []byte + key uint32 + size uint16 + raddr [4]byte + } + + // responseLengths stores the length of responses received. + // together they should add up to the written length of responseRing. + incomingEcho []struct { + length uint16 + id uint16 + seq uint16 + raddr [4]byte + } + responseRing internal.Ring +} + +type ClientConfig struct { + ResponseQueueBuffer []byte + ResponseQueueLimit int + HashSeed uint32 + // ID is used for Echo (ping) ID field setting. + ID uint16 +} + +func (client *Client) Configure(cfg ClientConfig) error { + if cfg.HashSeed == 0 || len(cfg.ResponseQueueBuffer) < 16 || cfg.ResponseQueueLimit <= 0 { + return lneto.ErrInvalidConfig + } + client.connid++ + internal.SliceReuse(&client.outgoingEcho, cfg.ResponseQueueLimit) + client.responseRing = internal.Ring{Buf: cfg.ResponseQueueBuffer} + client.magic = cfg.HashSeed + client.id = cfg.ID + return nil +} + +func (client *Client) Protocol() uint64 { return uint64(lneto.IPProtoICMP) } + +func (client *Client) LocalPort() uint16 { return 0 } + +func (client *Client) ConnectionID() *uint64 { return &client.connid } + +func (client *Client) Abort() { + client.Reset() + client.connid++ +} + +func (client *Client) Reset() { + client.incomingEcho = client.incomingEcho[:0] + client.outgoingEcho = client.outgoingEcho[:0] + client.responseRing.Reset() +} + +func (client *Client) Demux(carrierData []byte, frameOffset int) error { + rawdata := carrierData[frameOffset:] + ifrm, err := NewFrame(rawdata) + if err != nil { + return err + } + tp := ifrm.Type() + if tp != TypeEcho && tp != TypeEchoReply { + return lneto.ErrPacketDrop + } + var crc lneto.CRC791 + if crc.PayloadSum16(rawdata) != 0 { + return lneto.ErrBadCRC + } + var raddr [4]byte + ipEnabled := frameOffset >= 20 + if ipEnabled { + src, _, _, _, _ := internal.GetIPAddr(carrierData) + if len(src) == 4 { + raddr = [4]byte(src) + } + } + switch tp { + case TypeEcho: + // We received a ping request; not handled client-side. + efrm := FrameEcho{Frame: ifrm} + data := efrm.Data() + n, werr := client.responseRing.Write(data) + if werr != nil { + err = werr + break + } + v := internal.SliceReclaim(&client.incomingEcho) + v.length = uint16(n) + v.id = efrm.Identifier() + v.seq = efrm.SequenceNumber() + v.raddr = raddr + + case TypeEchoReply: + efrm := FrameEcho{Frame: ifrm} + data := efrm.Data() + hash := client.magichash(data, len(data)) & keyHashBits + idx := client.pingidx(hash) + if idx < 0 || (ipEnabled && client.outgoingEcho[idx].raddr != raddr) { + err = lneto.ErrPacketDrop + break + } + client.outgoingEcho[idx].key |= keyHashCompletedBit + + default: + err = lneto.ErrPacketDrop + } + return err +} + +func (client *Client) Encapsulate(carrierData []byte, ipOffset, frameOffset int) (int, error) { + ifrm, err := NewFrame(carrierData[frameOffset:]) + if err != nil { + return 0, err + } + + // Put n bytes of ICMP data. + var n int + var raddr [4]byte + if len(client.incomingEcho) > 0 { + // Priority: send echo reply.1 + inc := client.incomingEcho[0] + efrm := FrameEcho{Frame: ifrm} + efrm.SetType(TypeEchoReply) + efrm.SetIdentifier(inc.id) + efrm.SetSequenceNumber(inc.seq) + dataLen := int(inc.length) + _, rerr := client.responseRing.Read(efrm.Data()[:dataLen]) + if rerr != nil { + return 0, rerr + } + client.incomingEcho = slices.Delete(client.incomingEcho, 0, 1) + n = sizeHeader + dataLen + raddr = inc.raddr + } else if len(client.outgoingEcho) > 0 { + idx := 0 + for idx < len(client.outgoingEcho) { + out := &client.outgoingEcho[idx] + if out.key&keyHashSentBit == 0 { + break + } + idx++ + } + if idx >= len(client.outgoingEcho) { + return 0, nil // No pending to send packet. + } + out := &client.outgoingEcho[idx] + efrm := FrameEcho{Frame: ifrm} + efrm.SetType(TypeEcho) + efrm.SetIdentifier(client.id) + efrm.SetSequenceNumber(client.seq()) + pattern := out.pattern + data := efrm.Data() + size := int(out.size) + written := 0 + for written+len(pattern) <= size && written+len(pattern) <= len(data) { + copy(data[written:], pattern) + written += len(pattern) + } + copy(data[written:written+size%len(pattern)], pattern) + n = sizeHeader + size + raddr = out.raddr + } else { + return 0, nil + } + ifrm.buf = carrierData[frameOffset : frameOffset+n] // Raw buffer set. + ifrm.SetCode(0) + ifrm.SetCRC(0) + var crc lneto.CRC791 + sum := crc.PayloadSum16(carrierData[frameOffset : frameOffset+n]) + ifrm.SetCRC(sum) + if frameOffset >= 20 { + err = internal.SetIPAddrs(carrierData, 0, nil, raddr[:]) + } + return n, err +} + +func (client *Client) seq() uint16 { + client._seq++ + return client._seq +} + +func (client *Client) magichash(pattern []byte, size int) (hash uint32) { + hash = client.magic + i := 0 + n := size / len(pattern) + for i < n { + for _, b := range pattern { + hash = hash*31 + uint32(b) + } + i++ + } + n = size % len(pattern) + for i = 0; i < n; i++ { + hash = hash*31 + uint32(pattern[i]) + } + return hash +} + +func (client *Client) PingStart(remoteAddr [4]byte, pattern []byte, size uint16) (key uint32, err error) { + if int(size) < len(pattern) { + return 0, lneto.ErrInvalidConfig + } else if remoteAddr == [4]byte{} { + return 0, lneto.ErrZeroDestination + } + key = client.magichash(pattern, int(size)) & keyHashBits + v := internal.SliceReclaim(&client.outgoingEcho) + v.key = key + v.size = size + v.pattern = append(v.pattern[:0], pattern...) + v.raddr = remoteAddr + return key, nil +} + +func (client *Client) pingidx(key uint32) int { + for i := range client.outgoingEcho { + if client.outgoingEcho[i].key&keyHashBits == key { + return i + } + } + return -1 +} + +func (client *Client) PingPeek(key uint32) (completed, ok bool) { + idx := client.pingidx(key) + if idx >= 0 { + return client.outgoingEcho[idx].key&keyHashCompletedBit != 0, true + } + return false, false +} + +func (client *Client) PingPop(key uint32) (completed, ok bool) { + idx := client.pingidx(key) + if idx >= 0 { + completed := client.outgoingEcho[idx].key&keyHashCompletedBit != 0 + client.outgoingEcho = slices.Delete(client.outgoingEcho, idx, idx+1) + return completed, true + } + return false, false +} diff --git a/ipv4/icmpv4/client_test.go b/ipv4/icmpv4/client_test.go new file mode 100644 index 0000000..9165439 --- /dev/null +++ b/ipv4/icmpv4/client_test.go @@ -0,0 +1,139 @@ +package icmpv4 + +import ( + "testing" + + "github.com/soypat/lneto/internal" +) + +const ( + testHashSeed = 0xdeadbeef +) + +func TestClients(t *testing.T) { + const sizebuffer = 64 + const queuesize = 2 + var sender, responder Client + err := sender.Configure(ClientConfig{ + ResponseQueueBuffer: make([]byte, sizebuffer), + ResponseQueueLimit: queuesize, + HashSeed: testHashSeed, + }) + if err != nil { + t.Fatal(err) + } + err = responder.Configure(ClientConfig{ + ResponseQueueBuffer: make([]byte, sizebuffer), + ResponseQueueLimit: queuesize, + HashSeed: testHashSeed, + }) + if err != nil { + t.Fatal(err) + } + pattern := []byte("ab12") + size := 8 + var buf [64]byte + key1 := testSingleExchange(t, &sender, &responder, buf[:], pattern, uint16(size)) + completed, ok := sender.PingPop(key1) + if !completed || !ok { + t.Fatal("ping did not complete or not exist") + } +} + +func testSingleExchange(t *testing.T, sender, responder *Client, buf []byte, pattern []byte, size uint16) (senderKey uint32) { + var n int + senderKey, n = testSendEcho(t, sender, buf, pattern, size) + const frameOff = 0 + const ipOff = 0 + completed, ok := sender.PingPeek(senderKey) + if !ok { + t.Error("ping key not exist") + } else if completed { + t.Error("ping completed before response") + } + ifrm, _ := NewFrame(buf[frameOff : frameOff+n]) + efrm := FrameEcho{Frame: ifrm} + id, seq := efrm.Identifier(), efrm.SequenceNumber() + err1 := responder.Demux(buf[:frameOff+n], frameOff) + if err1 != nil { + t.Error("responder demux during single", err1) + } + n, err := responder.Encapsulate(buf, ipOff, frameOff) + if err != nil { + t.Error("responder encaps during single", err) + return + } else if n == 0 && err1 == nil { + t.Error("responder wrote no data") + return + } + ifrm, err = NewFrame(buf[frameOff : frameOff+n]) + if err != nil { + t.Fatal(err) + } + if ifrm.Type() != TypeEchoReply { + t.Fatalf("expected echo reply %d", ifrm.Type()) + } + efrm = FrameEcho{Frame: ifrm} + if efrm.Identifier() != id { + t.Error("mismatched identifier want/got:", id, efrm.Identifier()) + } + if efrm.SequenceNumber() != seq { + t.Error("mismatched sequence number want/got:", seq, efrm.SequenceNumber()) + } + data := efrm.Data() + testPatternMatch(t, data, pattern, int(size)) + err = sender.Demux(buf[:frameOff+n], frameOff) + if err != nil { + t.Error("sender demuxed response", err) + } + completed, ok = sender.PingPeek(senderKey) + if !completed { + t.Error("expected ping to have completed") + } + if !ok { + t.Error("ping key not exist after completion") + } + if completed2, ok2 := sender.PingPeek(senderKey); completed != completed2 || ok != ok2 { + t.Error("change in status after peek") + } + return senderKey +} + +func testSendEcho(t *testing.T, sender *Client, buf []byte, pattern []byte, size uint16) (key uint32, n int) { + t.Helper() + key, err := sender.PingStart([4]byte{1}, pattern, size) + if err != nil { + t.Fatal(err) + } + const frameOff = 0 + const ipOff = 0 + n, err = sender.Encapsulate(buf[:], ipOff, frameOff) + if err != nil { + t.Errorf("sender encapsulate: %v", err) + } + ifrm, err := NewFrame(buf[:n]) + if err != nil { + t.Fatal(err) // only fails in short frame case. + } + if ifrm.Type() != TypeEcho { + t.Errorf("not echo type on send: %d", ifrm.Type()) + } + efrm := FrameEcho{Frame: ifrm} + data := efrm.Data() + testPatternMatch(t, data, pattern, int(size)) + return key, n +} + +func testPatternMatch(t *testing.T, data []byte, pattern []byte, size int) { + t.Helper() + if len(data) != size { + t.Errorf("pattern size mismatch, want %d, got %d", size, len(data)) + } + for i := 0; i < size; i += len(pattern) { + got := data[i:min(len(data), i+len(pattern))] + want := pattern[:len(got)] + if !internal.BytesEqual(got, want) { + t.Errorf("pattern data mismatch at %d, got %s, want %s", i, got, want) + } + } +} diff --git a/ipv4/icmpv4/icmpv4.go b/ipv4/icmpv4/icmpv4.go index 137f59e..29ea007 100644 --- a/ipv4/icmpv4/icmpv4.go +++ b/ipv4/icmpv4/icmpv4.go @@ -6,6 +6,10 @@ import ( "github.com/soypat/lneto" ) +const ( + sizeHeader = 8 +) + type Type uint8 const ( @@ -54,7 +58,7 @@ const ( ) func NewFrame(buf []byte) (Frame, error) { - if len(buf) < 8 { + if len(buf) < sizeHeader { return Frame{}, lneto.ErrShortBuffer } return Frame{buf: buf}, nil diff --git a/stringers.go b/stringers.go index 8e065b1..066952d 100644 --- a/stringers.go +++ b/stringers.go @@ -209,11 +209,12 @@ func _() { _ = x[ErrInvalidField-13] _ = x[ErrInvalidLengthField-14] _ = x[ErrExhausted-15] + _ = x[ErrAlreadyRegistered-16] } -const _errGeneric_name = "lneto-bug(use build tag \"debugheaplog\")packet droppedincorrect checksumzero source(port/addr)zero destination(port/addr)short bufferbuffer fullinvalid addressunsupportedmismatchmismatched lengthinvalid configurationinvalid fieldinvalid length fieldresource exhausted" +const _errGeneric_name = "lneto-bug(use build tag \"debugheaplog\")packet droppedincorrect checksumzero source(port/addr)zero destination(port/addr)short bufferbuffer fullinvalid addressunsupportedmismatchmismatched lengthinvalid configurationinvalid fieldinvalid length fieldresource exhaustedprotocol already registered" -var _errGeneric_index = [...]uint16{0, 39, 53, 71, 93, 120, 132, 143, 158, 169, 177, 194, 215, 228, 248, 266} +var _errGeneric_index = [...]uint16{0, 39, 53, 71, 93, 120, 132, 143, 158, 169, 177, 194, 215, 228, 248, 266, 293} func (i errGeneric) String() string { i -= 1 diff --git a/tcp/tcp_legacy_test.go b/tcp/tcp_legacy_test.go index 67ad52f..ccab0b9 100644 --- a/tcp/tcp_legacy_test.go +++ b/tcp/tcp_legacy_test.go @@ -250,17 +250,6 @@ func TestExchange_rfc9293_figure12(t *testing.T) { tcbA.HelperInitState(tcp.StateEstablished, issA, issA, windowA) tcbA.HelperInitRcv(issB, issB, windowB) tcbA.HelperExchange(t, exchangeA) - // tcbA.HelperExchange(t, exchangeA[:1]) - // tcbA.HelperExchange(t, exchangeA[1:2]) - // tcbA.HelperExchange(t, exchangeA[2:]) - - return - exchangeB := reverseExchange(exchangeA) - exchangeB[1].WantPending = &tcp.Segment{SEQ: issB, ACK: issA + 1, Flags: FINACK, WND: windowB} - var tcbB tcp.ControlBlock - tcbB.HelperInitState(tcp.StateEstablished, issB, issB, windowB) - tcbB.HelperInitRcv(issA, issA, windowA) - tcbB.HelperExchange(t, exchangeB) } /* @@ -608,64 +597,3 @@ func TestUnexpectedStateClosing(t *testing.T) { } tcb.HelperExchange(t, ex[:]) } - -func TestExchange_helloworld_client(t *testing.T) { - return - // Client Transmission Control Block. - var tcb tcp.ControlBlock - // The client starts in the SYN_SENT state with a random sequence number. - gotClientSeg, _ := parseSegment(t, exchangeHelloWorld[0]) - - // We add the SYN state to the client. - tcb.HelperInitState(tcp.StateSynSent, gotClientSeg.SEQ, gotClientSeg.SEQ, gotClientSeg.WND) - err := tcb.Send(gotClientSeg) - if err != nil { - - t.Fatal(err) - } - tcb.HelperPrintSegment(t, false, gotClientSeg) - - segString := func(seg tcp.Segment) string { - return tcb.RelativeAutoSegment(seg).RelativeGoString(0, 0) - } - for i, packet := range exchangeHelloWorld { - if i == 0 { - continue // we already processed first packet. - } - seg, payload := parseSegment(t, packet) - if seg.DATALEN > 0 { - t.Logf("seg[%d] <%s> payload: %q", i, tcb.State(), string(payload)) - } else { - t.Logf("seg[%d] <%s>", i, tcb.State()) - } - isClient := packet[0] == 0x28 - if isClient { - isPSH := seg.Flags&tcp.FlagPSH != 0 - gotClientSeg.Flags |= seg.Flags & (tcp.FlagPSH | tcp.FlagFIN) // Can't predict when client will send FIN. - if isPSH { - gotClientSeg.DATALEN = seg.DATALEN - } - - gotClientSeg.WND = seg.WND // Ignore window field, not a core part of control flow. - if gotClientSeg != seg { - t.Fatalf("client:\n got=%+v\nwant=%+v", segString(gotClientSeg), segString(seg)) - } - err := tcb.Send(gotClientSeg) - if err != nil { - t.Fatalf("incoming %s:\nseg[%d]=%s\nrcv=%+v\nsnd=%+v", err, i, segString(gotClientSeg), tcb.RelativeRecvSpace(), tcb.RelativeSendSpace()) - } - tcb.HelperPrintSegment(t, false, gotClientSeg) - continue // we only pass server packets to the client. - } - err = tcb.Recv(seg) - if err != nil { - t.Fatalf("%s:\nseg[%d]=%s\nrcv=%+v\nsnd=%+v", err, i, segString(seg), tcb.RelativeRecvSpace(), tcb.RelativeSendSpace()) - } - tcb.HelperPrintSegment(t, true, seg) - var ok bool - gotClientSeg, ok = tcb.PendingSegment(0) - if !ok { - t.Fatalf("[%d]: got no segment state=%s", i, tcb.State()) - } - } -} diff --git a/x/xnet/stack-async.go b/x/xnet/stack-async.go index 2255348..84dba0b 100644 --- a/x/xnet/stack-async.go +++ b/x/xnet/stack-async.go @@ -1,6 +1,7 @@ package xnet import ( + "encoding/binary" "errors" "log/slog" "net/netip" @@ -14,6 +15,7 @@ import ( "github.com/soypat/lneto/ethernet" "github.com/soypat/lneto/internal" "github.com/soypat/lneto/internet" + "github.com/soypat/lneto/ipv4/icmpv4" "github.com/soypat/lneto/ntp" "github.com/soypat/lneto/tcp" ) @@ -29,6 +31,7 @@ type StackAsync struct { link internet.StackEthernet ip internet.StackIP arp arp.Handler + icmp icmpv4.Client udps internet.StackPorts tcps internet.StackPortsMACFiltered @@ -72,41 +75,71 @@ type StackConfig struct { MTU uint16 // Accept multicast ethernet and IP packets. Needed for MDNS. AcceptMulticast bool + // ICMPQueueLimit sets maximum number of input/output packets queued for processing. + // If set to zero ICMP cannot be enabled on the stack. + ICMPQueueLimit int } func (s *StackAsync) Hostname() string { return s.hostname } -func (s *StackAsync) Demux(carrierData []byte, etherOff int) error { +// IngressEthernet receives an Ethernet frame from the network and processes it through the stack. The frame should include the Ethernet header and payload and CRC if enabled. +func (s *StackAsync) IngressEthernet(ethernetFrame []byte) error { s.mu.Lock() defer s.mu.Unlock() - s.totalrecv += uint64(len(carrierData) - etherOff) - return s.link.Demux(carrierData, etherOff) + s.totalrecv += uint64(len(ethernetFrame)) + return s.link.Demux(ethernetFrame, 0) } -func (s *StackAsync) Encapsulate(carrierData []byte, offsetToIP, offsetToFrame int) (int, error) { +// EgressEthernet writes the next ethernet frame to send into dstEthernetFrame from the stack. +// The length of dstEthernetFrame should be at least MTU + Ethernet header (14) + CRC (4 if enabled). +func (s *StackAsync) EgressEthernet(dstEthernetFrame []byte) (int, error) { s.mu.Lock() defer s.mu.Unlock() - - n, err := s.link.Encapsulate(carrierData, offsetToIP, offsetToFrame) + n, err := s.link.Encapsulate(dstEthernetFrame, -1, 0) s.totalsent += uint64(n) return n, err } +// IngressIP processes an incoming IP frame through the stack and omits ethernet header processing. +func (s *StackAsync) IngressIP(ipFrame []byte) error { + s.mu.Lock() + defer s.mu.Unlock() + s.totalrecv += uint64(len(ipFrame)) + return s.ip.Demux(ipFrame, 0) +} + +// EgressIP writes the next IP frame to send into dstIPFrame from the stack. The length of dstIPFrame should be at least MTU. +func (s *StackAsync) EgressIP(dstIPFrame []byte) (int, error) { + s.mu.Lock() + defer s.mu.Unlock() + if len(dstIPFrame) < s.link.MTU() { + return 0, lneto.ErrShortBuffer + } + n, err := s.ip.Encapsulate(dstIPFrame, 0, 0) + s.totalsent += uint64(n) + return n, err +} + +// MTU is the Maximum Transmission Unit of the stack corresponding +// to the maximum payload size of an ethernet frame that can be sent through the stack. +// Important to note that the actual ethernet frame size is MTU + Ethernet header (14) + CRC (4 if enabled), this is known as the Maximum Frame Length. func (s *StackAsync) MTU() int { + s.mu.Lock() + defer s.mu.Unlock() return s.link.MTU() } func (s *StackAsync) Reset(cfg StackConfig) error { - s.mu.Lock() - defer s.mu.Unlock() - mac := cfg.HardwareAddress - addr := cfg.StaticAddress - s.prng = uint32(cfg.RandSeed) - if s.prng == 0 { + if cfg.RandSeed == 0 { return lneto.ErrInvalidConfig } + mac := cfg.HardwareAddress + addr := cfg.StaticAddress + s.mu.Lock() + defer s.mu.Unlock() + s.prng = uint32(cfg.RandSeed) s.hostname = cfg.Hostname if !addr.IsValid() { addr = netip.AddrFrom4([4]byte{}) // If static not set DHCP will be performed and address will be zero. @@ -133,7 +166,6 @@ func (s *StackAsync) Reset(cfg StackConfig) error { return err } s.ip.SetAcceptMulticast(cfg.AcceptMulticast) - // err = s.resetARP() if err != nil { return err @@ -144,9 +176,6 @@ func (s *StackAsync) Reset(cfg StackConfig) error { return err } internal.SliceReuse(&s.userUDPs, cfg.MaxUDPConns) - if err != nil { - return err - } // Enable TCP if connections present. if cfg.MaxTCPConns > 0 { @@ -170,6 +199,16 @@ func (s *StackAsync) Reset(cfg StackConfig) error { if err != nil { return err } + if cfg.ICMPQueueLimit > 0 { + err = s.icmp.Configure(icmpv4.ClientConfig{ + ResponseQueueBuffer: make([]byte, cfg.ICMPQueueLimit*64), + ResponseQueueLimit: cfg.ICMPQueueLimit, + HashSeed: s.Prand32(), + }) + if err != nil { + return err + } + } var timebuf [32]time.Time s.sysprec = ntp.CalculateSystemPrecision(time.Now, timebuf[:]) if s.clientID == "" { @@ -211,6 +250,18 @@ func (s *StackAsync) resetARP() error { return nil } +func (s *StackAsync) prandRead(buf []byte) { + i := 0 + for ; i+3 < len(buf); i += 4 { + binary.LittleEndian.PutUint32(buf[i:], s.prand32()) + } + v := s.prand32() + for i < len(buf) { + buf[i] = byte(v >> (8 * (i % 4))) + i++ + } +} + // Prand32 generates a pseudo random 32-bit unsigned integer from the internal state and advances the seed. func (s *StackAsync) Prand32() (randval uint32) { s.mu.Lock() @@ -221,10 +272,7 @@ func (s *StackAsync) Prand32() (randval uint32) { func (s *StackAsync) prand32() uint32 { /* Algorithm "xor" from p. 4 of Marsaglia, "Xorshift RNGs" */ - seed := s.prng - seed ^= seed << 13 - seed ^= seed >> 17 - seed ^= seed << 5 + seed := internal.Prand32(s.prng) s.prng = seed return seed } @@ -282,6 +330,22 @@ func (s *StackAsync) Gateway6() [6]byte { return s.link.Gateway6() } +// EnableICMP registers an ICMP handler to the stack when enabled is true. +// If enabled=false the currently registered ICMP handler is unregistered and state reset. +func (s *StackAsync) EnableICMP(enabled bool) (err error) { + if enabled { + if s.ip.IsRegistered(lneto.IPProtoICMP) { + err = lneto.ErrAlreadyRegistered + } else { + err = s.ip.Register(&s.icmp) + } + + } else { + s.icmp.Abort() + } + return err +} + func (s *StackAsync) DialTCP(conn *tcp.Conn, localPort uint16, addrp netip.AddrPort) (err error) { s.mu.Lock() defer s.mu.Unlock() @@ -337,7 +401,7 @@ func (s *StackAsync) RegisterListener(listener *tcp.Listener) (err error) { // RegisterUDP registers a StackNode on a UDP port with the given remote address and port. // The StackUDPPort wrapping is handled internally. The number of user-registered UDP ports // is limited by [StackConfig.MaxUDPConns]. -func (s *StackAsync) RegisterUDP(node internet.StackNode, remoteAddr []byte, remotePort uint16) error { +func (s *StackAsync) RegisterUDP(node lneto.StackNode, remoteAddr []byte, remotePort uint16) error { s.mu.Lock() defer s.mu.Unlock() idx := len(s.userUDPs) diff --git a/x/xnet/stack-blocking.go b/x/xnet/stack-blocking.go index 916191f..f9eb5fd 100644 --- a/x/xnet/stack-blocking.go +++ b/x/xnet/stack-blocking.go @@ -2,9 +2,11 @@ package xnet import ( "errors" + "net" "net/netip" "time" + "github.com/soypat/lneto" "github.com/soypat/lneto/dhcpv4" "github.com/soypat/lneto/tcp" ) @@ -58,6 +60,33 @@ func (s StackBlocking) DoDHCPv4(reqAddr [4]byte, timeout time.Duration) (*DHCPRe return s.async.ResultDHCP() } +func (s StackBlocking) DoPing(hostAddr netip.Addr, timeout time.Duration) (roundtrip time.Duration, err error) { + if !hostAddr.Is4() { + return 0, lneto.ErrInvalidAddr + } + var buf [16]byte + s.async.prandRead(buf[:]) + key, err := s.async.icmp.PingStart(hostAddr.As4(), buf[:], 56) // size=56 so ICMP size is 64, like linux. + if err != nil { + return 0, err + } + start := time.Now() + sleep := timeout / maxIter + for i := 0; i < maxIter; i++ { + time.Sleep(sleep) + elapsed := time.Since(start) + completed, exists := s.async.icmp.PingPop(key) + if !exists { + return 0, net.ErrClosed // lneto.ErrAborted + } else if completed { + return elapsed, nil + } else if elapsed > timeout { + break + } + } + return 0, errDeadlineExceed +} + func (s StackBlocking) DoNTP(hostAddr netip.Addr, timeout time.Duration) (offset time.Duration, err error) { err = s.async.StartNTP(hostAddr) if err != nil { diff --git a/x/xnet/xnet_bench_test.go b/x/xnet/xnet_bench_test.go index e21f195..d18c8b3 100644 --- a/x/xnet/xnet_bench_test.go +++ b/x/xnet/xnet_bench_test.go @@ -46,23 +46,23 @@ func BenchmarkARPExchange(b *testing.B) { if err != nil { b.Fatal(err) } - n, err := c1.Encapsulate(buf[:], -1, 0) // Send Request. + n, err := c1.EgressEthernet(buf[:]) // Send Request. if err != nil { b.Fatal(err) } else if n == 0 { b.Fatal("expected send of data after first query") } - err = c2.Demux(buf[:n], 0) // Receive request. + err = c2.IngressEthernet(buf[:n]) // Receive request. if err != nil { b.Fatal(err) } - n, err = c2.Encapsulate(buf[:], -1, 0) // Send response. + n, err = c2.EgressEthernet(buf[:]) // Send response. if err != nil { b.Fatal(err) } else if n == 0 { b.Fatal("got no response to request") } - err = c1.Demux(buf[:n], 0) // Receive response. + err = c1.IngressEthernet(buf[:n]) // Receive response. if err != nil { b.Fatal(err) } @@ -140,31 +140,31 @@ func BenchmarkTCPHandshake(b *testing.B) { } // SYN from client. - n, err := client.Encapsulate(pktbuf[:], -1, 0) + n, err := client.EgressEthernet(pktbuf[:]) if err != nil { b.Fatal(err) } - err = sv.Demux(pktbuf[:n], 0) + err = sv.IngressEthernet(pktbuf[:n]) if err != nil { b.Fatal(err) } // SYN-ACK from server. - n, err = sv.Encapsulate(pktbuf[:], -1, 0) + n, err = sv.EgressEthernet(pktbuf[:]) if err != nil { b.Fatal(err) } - err = client.Demux(pktbuf[:n], 0) + err = client.IngressEthernet(pktbuf[:n]) if err != nil { b.Fatal(err) } // ACK from client. - n, err = client.Encapsulate(pktbuf[:], -1, 0) + n, err = client.EgressEthernet(pktbuf[:]) if err != nil { b.Fatal(err) } - err = sv.Demux(pktbuf[:n], 0) + err = sv.IngressEthernet(pktbuf[:n]) if err != nil { b.Fatal(err) } diff --git a/x/xnet/xnet_concurrent_test.go b/x/xnet/xnet_concurrent_test.go index d73920e..e12978c 100644 --- a/x/xnet/xnet_concurrent_test.go +++ b/x/xnet/xnet_concurrent_test.go @@ -159,15 +159,15 @@ func kernelLoop(ctx context.Context, server *StackAsync, clients []StackAsync) { } // Process server outgoing -> route to appropriate client based on dest IP. - if n, _ := server.Encapsulate(buf, -1, 0); n > 0 { + if n, _ := server.EgressEthernet(buf); n > 0 { routePacketToClient(buf[:n], clients) } // Process each client outgoing in randomized order. rng.Shuffle(len(order), func(i, j int) { order[i], order[j] = order[j], order[i] }) for _, idx := range order { - if n, _ := clients[idx].Encapsulate(buf, -1, 0); n > 0 { - server.Demux(buf[:n], 0) // All clients talk to server. + if n, _ := clients[idx].EgressEthernet(buf); n > 0 { + server.IngressEthernet(buf[:n]) // All clients talk to server. } } @@ -184,7 +184,7 @@ func routePacketToClient(pkt []byte, clients []StackAsync) { for i := range clients { if clients[i].Addr() == dstIP { - clients[i].Demux(pkt, 0) + clients[i].IngressEthernet(pkt) return } } diff --git a/x/xnet/xnet_dns_test.go b/x/xnet/xnet_dns_test.go index f8fd94d..a634e4d 100644 --- a/x/xnet/xnet_dns_test.go +++ b/x/xnet/xnet_dns_test.go @@ -49,7 +49,7 @@ func TestDNS_QueryReceivesAnswer(t *testing.T) { // Client sends DNS query. const carrierDataSize = MTU + ethernet.MaxOverheadSize var buf [carrierDataSize]byte - n, err := client.Encapsulate(buf[:], -1, 0) + n, err := client.EgressEthernet(buf[:]) if err != nil { t.Fatal("client Encapsulate failed:", err) } @@ -70,7 +70,7 @@ func TestDNS_QueryReceivesAnswer(t *testing.T) { } // Deliver response to client. - err = client.Demux(responsePkt, 0) + err = client.IngressEthernet(responsePkt) if err != nil { t.Fatal("client Demux failed:", err) } diff --git a/x/xnet/xnet_fuzz_test.go b/x/xnet/xnet_fuzz_test.go index c8a1c1b..7029113 100644 --- a/x/xnet/xnet_fuzz_test.go +++ b/x/xnet/xnet_fuzz_test.go @@ -34,12 +34,12 @@ func FuzzStackAsyncHTTP(f *testing.F) { written := false closed := false for { - n1, err := s1.Encapsulate(buf[:], -1, 0) + n1, err := s1.EgressEthernet(buf[:]) if err != nil { f.Fatal(err) } if n1 > 0 { - err = s2.Demux(buf[:n1], 0) + err = s2.IngressEthernet(buf[:n1]) if err != nil { f.Fatal(err) } @@ -53,10 +53,10 @@ func FuzzStackAsyncHTTP(f *testing.F) { written = true } } - n2, err := s2.Encapsulate(buf[:], -1, 0) + n2, err := s2.EgressEthernet(buf[:]) if n2 > 0 { pktnum++ - err = s1.Demux(buf[:n2], 0) + err = s1.IngressEthernet(buf[:n2]) if err != nil { f.Fatal(err) } @@ -88,7 +88,7 @@ func FuzzStackAsyncHTTP(f *testing.F) { closed := false const maxpkts = 100 for { - n1, err := s1.Encapsulate(buf[:], -1, 0) + n1, err := s1.EgressEthernet(buf[:]) if err != nil { t.Fatal(err) } @@ -97,21 +97,21 @@ func FuzzStackAsyncHTTP(f *testing.F) { n1 = copy(buf[:], a) fixIPTCPCRCs(buf[:n1]) } - s2.Demux(buf[:n1], 0) + s2.IngressEthernet(buf[:n1]) pkt++ if !written && c2.State() >= tcp.StateEstablished { c2.Write(data) written = true } } - n2, err := s2.Encapsulate(buf[:], -1, 0) + n2, err := s2.EgressEthernet(buf[:]) if n2 > 0 { if pkt == pktnum { n2 = copy(buf[:], a) fixIPTCPCRCs(buf[:n2]) } pkt++ - s1.Demux(buf[:n2], 0) + s1.IngressEthernet(buf[:n2]) } if n1 == 0 && n2 == 0 { if !closed { diff --git a/x/xnet/xnet_mdns_test.go b/x/xnet/xnet_mdns_test.go index 8b2cc25..b44e035 100644 --- a/x/xnet/xnet_mdns_test.go +++ b/x/xnet/xnet_mdns_test.go @@ -114,7 +114,7 @@ func TestMDNS_QueryResponse(t *testing.T) { var buf [carrierDataSize]byte // Querier encapsulates query through full stack (Ethernet+IP+UDP+mDNS). - n, err := querierStack.Encapsulate(buf[:], -1, 0) + n, err := querierStack.EgressEthernet(buf[:]) if err != nil || n == 0 { t.Fatal("querier encapsulate:", err, n) } @@ -135,13 +135,13 @@ func TestMDNS_QueryResponse(t *testing.T) { } // Responder demuxes the query (multicast MAC+IP accepted via AcceptMulticast). - err = responderStack.Demux(buf[:n], 0) + err = responderStack.IngressEthernet(buf[:n]) if err != nil { t.Fatal("responder demux:", err) } // Responder encapsulates response. - n, err = responderStack.Encapsulate(buf[:], -1, 0) + n, err = responderStack.EgressEthernet(buf[:]) if err != nil || n == 0 { t.Fatal("responder encapsulate:", err, n) } @@ -165,7 +165,7 @@ func TestMDNS_QueryResponse(t *testing.T) { } // Querier demuxes response. - err = querierStack.Demux(buf[:n], 0) + err = querierStack.IngressEthernet(buf[:n]) if err != nil { t.Fatal("querier demux:", err) } @@ -310,25 +310,25 @@ func mdnsQueryRespond(t *testing.T, querier, responder *StackAsync, buf []byte) t.Helper() // Querier encapsulates query. - n, err := querier.Encapsulate(buf, -1, 0) + n, err := querier.EgressEthernet(buf) if err != nil || n == 0 { t.Fatal("querier encapsulate:", err, n) } // Responder demuxes multicast query directly. - err = responder.Demux(buf[:n], 0) + err = responder.IngressEthernet(buf[:n]) if err != nil { t.Fatal("responder demux:", err) } // Responder encapsulates response. - n, err = responder.Encapsulate(buf, -1, 0) + n, err = responder.EgressEthernet(buf) if err != nil || n == 0 { t.Fatal("responder encapsulate:", err, n) } // Querier demuxes multicast response. - err = querier.Demux(buf[:n], 0) + err = querier.IngressEthernet(buf[:n]) if err != nil { t.Fatal("querier demux:", err) } @@ -382,7 +382,7 @@ func TestMDNS_RealWorldQueries(t *testing.T) { var buf [MTU + ethernet.MaxOverheadSize]byte checkNoData := func(msg string) { t.Helper() - n, err := responderStack.Encapsulate(buf[:], -1, 0) + n, err := responderStack.EgressEthernet(buf[:]) if err != nil { t.Fatal(err) } else if n != 0 { @@ -434,18 +434,18 @@ func TestMDNS_RealWorldQueries(t *testing.T) { ifrm.CRCWriteUDPPseudo(&crc, ufrm.Length()) got := crc.PayloadSum16(ifrm.Payload()) ufrm.SetCRC(got) - err := responderStack.Demux(buf[:14+20+8+msg.Len()], 0) + err := responderStack.IngressEthernet(buf[:14+20+8+msg.Len()]) if err != nil { t.Fatal(err) } } - n, err := responderStack.Encapsulate(buf[:], -1, 0) + n, err := responderStack.EgressEthernet(buf[:]) if err != nil { t.Fatal(err) } else if n < 14+20+8+dns.SizeHeader { t.Error("expected response", n) } - n, err = responderStack.Encapsulate(buf[:], -1, 0) + n, err = responderStack.EgressEthernet(buf[:]) if err != nil { t.Fatal(err) } else if n != 0 { diff --git a/x/xnet/xnet_test.go b/x/xnet/xnet_test.go index d4477a1..7fb88e5 100644 --- a/x/xnet/xnet_test.go +++ b/x/xnet/xnet_test.go @@ -68,7 +68,7 @@ func TestTCPConn_ReadBlocksUntilDataAvailable(t *testing.T) { // Perform packet exchange to deliver data. tst.bufmu.Lock() buf := tst.buf[:cap(tst.buf)] - n, err := client.Encapsulate(buf, -1, 0) + n, err := client.EgressEthernet(buf) if err != nil { tst.bufmu.Unlock() t.Fatal(err) @@ -77,7 +77,7 @@ func TestTCPConn_ReadBlocksUntilDataAvailable(t *testing.T) { tst.bufmu.Unlock() t.Fatal("expected data packet from client") } - err = sv.Demux(buf[:n], 0) + err = sv.IngressEthernet(buf[:n]) tst.bufmu.Unlock() if err != nil { t.Fatal(err) @@ -418,7 +418,7 @@ func (tst *tester) TCPExchange(expect tcpExpectExchange, stack1, stack2 *StackAs panic("OOB") } - n, err := src.Encapsulate(buf[:], -1, 0) + n, err := src.EgressEthernet(buf[:]) if err != nil { t.Fatal(err) } else if n == 0 { @@ -466,7 +466,7 @@ func (tst *tester) TCPExchange(expect tcpExpectExchange, stack1, stack2 *StackAs if seg.Flags != expect.WantFlags { t.Errorf("expected flags %s, got %s", expect.WantFlags.String(), seg.Flags.String()) } - err = dst.Demux(buf[:n], 0) + err = dst.IngressEthernet(buf[:n]) if err != nil { t.Fatal(err) } @@ -481,7 +481,7 @@ func (tst *tester) ARPExchangeOnly(querying, target *StackAsync) { buf := tst.buf[:cap(tst.buf)] // === PHASE 1: ARP Request from querying stack === - n, err := querying.Encapsulate(buf[:], -1, 0) + n, err := querying.EgressEthernet(buf[:]) if err != nil { t.Fatal(err) } else if n == 0 { @@ -526,7 +526,7 @@ func (tst *tester) ARPExchangeOnly(querying, target *StackAsync) { } // Deliver request to target - err = target.Demux(buf[:n], 0) + err = target.IngressEthernet(buf[:n]) if err != nil { t.Fatal("target demux request:", err) } @@ -534,7 +534,7 @@ func (tst *tester) ARPExchangeOnly(querying, target *StackAsync) { // === PHASE 2: ARP Reply from target stack === buf = tst.buf[:cap(tst.buf)] - n, err = target.Encapsulate(buf[:], -1, 0) + n, err = target.EgressEthernet(buf[:]) if err != nil { t.Fatal(err) } else if n == 0 { @@ -574,7 +574,7 @@ func (tst *tester) ARPExchangeOnly(querying, target *StackAsync) { } // Deliver reply to querying stack - err = querying.Demux(buf[:n], 0) + err = querying.IngressEthernet(buf[:n]) if err != nil { t.Fatal("querying demux reply:", err) } @@ -748,7 +748,7 @@ func TestTCPConn_BufferNotClearedOnPassiveClose(t *testing.T) { // Server sends DATA packet to client. tst.bufmu.Lock() buf := tst.buf[:cap(tst.buf)] - n, err := sv.Encapsulate(buf, -1, 0) + n, err := sv.EgressEthernet(buf) if err != nil { tst.bufmu.Unlock() t.Fatal("server encapsulate data:", err) @@ -757,7 +757,7 @@ func TestTCPConn_BufferNotClearedOnPassiveClose(t *testing.T) { tst.bufmu.Unlock() t.Fatal("expected data packet from server") } - err = client.Demux(buf[:n], 0) + err = client.IngressEthernet(buf[:n]) tst.bufmu.Unlock() if err != nil { t.Fatal("client demux data:", err) @@ -771,13 +771,13 @@ func TestTCPConn_BufferNotClearedOnPassiveClose(t *testing.T) { // Client sends ACK for data. tst.bufmu.Lock() buf = tst.buf[:cap(tst.buf)] - n, err = client.Encapsulate(buf, -1, 0) + n, err = client.EgressEthernet(buf) if err != nil { tst.bufmu.Unlock() t.Fatal("client encapsulate ACK:", err) } if n > 0 { - err = sv.Demux(buf[:n], 0) + err = sv.IngressEthernet(buf[:n]) if err != nil { tst.bufmu.Unlock() t.Fatal("server demux ACK:", err) @@ -794,7 +794,7 @@ func TestTCPConn_BufferNotClearedOnPassiveClose(t *testing.T) { // Server sends FIN (enters FIN_WAIT_1). tst.bufmu.Lock() buf = tst.buf[:cap(tst.buf)] - n, err = sv.Encapsulate(buf, -1, 0) + n, err = sv.EgressEthernet(buf) if err != nil { tst.bufmu.Unlock() t.Fatal("server encapsulate FIN:", err) @@ -803,7 +803,7 @@ func TestTCPConn_BufferNotClearedOnPassiveClose(t *testing.T) { tst.bufmu.Unlock() t.Fatal("expected FIN packet from server") } - err = client.Demux(buf[:n], 0) + err = client.IngressEthernet(buf[:n]) tst.bufmu.Unlock() if err != nil { t.Fatal("client demux FIN:", err) @@ -819,13 +819,13 @@ func TestTCPConn_BufferNotClearedOnPassiveClose(t *testing.T) { // Client sends ACK for FIN. tst.bufmu.Lock() buf = tst.buf[:cap(tst.buf)] - n, err = client.Encapsulate(buf, -1, 0) + n, err = client.EgressEthernet(buf) if err != nil { tst.bufmu.Unlock() t.Fatal("client encapsulate ACK:", err) } if n > 0 { - err = sv.Demux(buf[:n], 0) + err = sv.IngressEthernet(buf[:n]) if err != nil { tst.bufmu.Unlock() t.Fatal("server demux ACK:", err) @@ -846,7 +846,7 @@ func TestTCPConn_BufferNotClearedOnPassiveClose(t *testing.T) { // Client sends FIN (enters LAST_ACK). tst.bufmu.Lock() buf = tst.buf[:cap(tst.buf)] - n, err = client.Encapsulate(buf, -1, 0) + n, err = client.EgressEthernet(buf) if err != nil { tst.bufmu.Unlock() t.Fatal("client encapsulate FIN:", err) @@ -855,7 +855,7 @@ func TestTCPConn_BufferNotClearedOnPassiveClose(t *testing.T) { tst.bufmu.Unlock() t.Fatal("expected FIN packet from client") } - err = sv.Demux(buf[:n], 0) + err = sv.IngressEthernet(buf[:n]) tst.bufmu.Unlock() if err != nil { t.Fatal("server demux client FIN:", err) @@ -871,7 +871,7 @@ func TestTCPConn_BufferNotClearedOnPassiveClose(t *testing.T) { // Server sends final ACK. tst.bufmu.Lock() buf = tst.buf[:cap(tst.buf)] - n, err = sv.Encapsulate(buf, -1, 0) + n, err = sv.EgressEthernet(buf) if err != nil { tst.bufmu.Unlock() t.Fatal("server encapsulate final ACK:", err) @@ -880,7 +880,7 @@ func TestTCPConn_BufferNotClearedOnPassiveClose(t *testing.T) { tst.bufmu.Unlock() t.Fatal("expected final ACK from server") } - err = client.Demux(buf[:n], 0) + err = client.IngressEthernet(buf[:n]) tst.bufmu.Unlock() if err != nil { t.Fatal("client demux final ACK:", err) @@ -944,7 +944,7 @@ func TestStackAsync_ICMPEchoChecksum(t *testing.T) { SequenceNumber: 1, Payload: icmpPayload, }) - err = stack.Demux(pkt, 0) + err = stack.IngressEthernet(pkt) if err != nil { t.Fatalf("valid ICMP echo rejected: %v", err) } @@ -958,7 +958,7 @@ func TestStackAsync_ICMPEchoChecksum(t *testing.T) { Payload: icmpPayload, }) pkt = append(pkt, 0xDE, 0xAD, 0xBE, 0xEF) // Simulate Ethernet FCS. - err = stack.Demux(pkt, 0) + err = stack.IngressEthernet(pkt) if err != nil { t.Fatalf("valid ICMP with trailing FCS rejected: %v", err) } @@ -970,7 +970,7 @@ func TestStackAsync_ICMPEchoChecksum(t *testing.T) { Payload: icmpPayload, }) pkt[len(pkt)-1] ^= 0xFF // Flip bits in last payload byte to corrupt ICMP checksum. - err = stack.Demux(pkt, 0) + err = stack.IngressEthernet(pkt) if err == nil { t.Fatal("corrupted ICMP accepted, expected CRC error") } @@ -983,7 +983,7 @@ func TestStackAsync_ICMPEchoChecksum(t *testing.T) { }) pkt[len(pkt)-1] ^= 0xFF // Corrupt ICMP payload. pkt = append(pkt, 0xDE, 0xAD, 0xBE, 0xEF) // Simulate Ethernet FCS. - err = stack.Demux(pkt, 0) + err = stack.IngressEthernet(pkt) if err == nil { t.Fatal("corrupted ICMP with FCS accepted, expected CRC error") }