From 018f9258acf142e1423932b756785236d5793423 Mon Sep 17 00:00:00 2001 From: Pat Whittingslow Date: Sat, 3 Jan 2026 12:41:57 -0300 Subject: [PATCH] Add tcp.Listener (#15) * begin reviewing TCP listener * begin adding listener tests * adding TestExchange implementation * split out legacy and Step tests * move listener to tcp package * fix up tcp tests taking very long * add xnet.TCPPool and work on TCPPool semantics * xnet: TCPPool only accepts established connections * xnet: getting to bottom of panic in xnet.Listener * xnet: improve listener tests * fix several data races in testing fixtures * fix more synchronization things in listener test * finalize tcplistener test --- errors.go | 8 +- examples/stack/main.go | 4 +- examples/xnet/main.go | 2 + internal/ip.go | 20 + internet/node-tcplistener.go | 233 ---------- internet/stackbasic_test.go | 5 + internet/tcplistener_test.go | 398 +++++++++++++++++ tcp/conn.go | 2 +- tcp/control.go | 2 +- tcp/handler.go | 13 +- tcp/internal_test.go | 252 +++++++++-- tcp/listener.go | 256 +++++++++++ tcp/tcp_legacy_test.go | 665 ++++++++++++++++++++++++++++ tcp/tcp_test.go | 815 ++++++++++++----------------------- tcp/txqueue_test.go | 7 +- x/xnet/stack-async.go | 16 + x/xnet/tcppool.go | 152 +++++++ x/xnet/xnet_listener_test.go | 225 ++++++++++ x/xnet/xnet_test.go | 62 +-- 19 files changed, 2288 insertions(+), 849 deletions(-) delete mode 100644 internet/node-tcplistener.go create mode 100644 internet/tcplistener_test.go create mode 100644 tcp/listener.go create mode 100644 tcp/tcp_legacy_test.go create mode 100644 x/xnet/tcppool.go create mode 100644 x/xnet/xnet_listener_test.go diff --git a/errors.go b/errors.go index ca86c8f..a00b818 100644 --- a/errors.go +++ b/errors.go @@ -19,9 +19,11 @@ type errGeneric uint8 // Generic errors common to internet functioning. const ( - _ errGeneric = iota // non-initialized err - ErrPacketDrop // packet dropped - ErrBadCRC // incorrect checksum + _ errGeneric = iota // non-initialized err + ErrPacketDrop // packet dropped + ErrBadCRC // incorrect checksum + ErrZeroSource // zero source(port/addr) + ErrZeroDestination // zero destination(port/addr) ) func (err errGeneric) Error() string { diff --git a/examples/stack/main.go b/examples/stack/main.go index 9370c3e..5e3602b 100644 --- a/examples/stack/main.go +++ b/examples/stack/main.go @@ -233,8 +233,8 @@ func (stack *Stack) Send(b []byte) (int, error) { return stack.ethernet.Encapsulate(b, -1, 0) } -func (stack *Stack) OpenTCPListener(port uint16) (*internet.NodeTCPListener, error) { - var listener internet.NodeTCPListener +func (stack *Stack) OpenTCPListener(port uint16) (*tcp.Listener, error) { + var listener tcp.Listener err := listener.Reset(port, naiveTCPPool{}) if err != nil { return nil, err diff --git a/examples/xnet/main.go b/examples/xnet/main.go index f54a755..7c5d8d5 100644 --- a/examples/xnet/main.go +++ b/examples/xnet/main.go @@ -1,3 +1,5 @@ +//go:build !tinygo && linux + package main import ( diff --git a/internal/ip.go b/internal/ip.go index 77a9263..362d07c 100644 --- a/internal/ip.go +++ b/internal/ip.go @@ -67,3 +67,23 @@ func IsZeroed[T comparable](a ...T) bool { } return true } + +// DeleteZeroed deletes zero values in-place contained within the +// slice and returns the modified slice without zero values. +// Does not modify capacity. +func DeleteZeroed[T comparable](a []T) []T { + var z T + off := 0 + deleted := false + for i := 0; i < len(a); i++ { + if a[i] != z { + if deleted { + a[off] = a[i] + } + off++ + } else if !deleted { + deleted = true + } + } + return a[:off] +} diff --git a/internet/node-tcplistener.go b/internet/node-tcplistener.go deleted file mode 100644 index 9b1aa14..0000000 --- a/internet/node-tcplistener.go +++ /dev/null @@ -1,233 +0,0 @@ -package internet - -import ( - "bytes" - "errors" - "log/slog" - "net" - - "github.com/soypat/lneto" - "github.com/soypat/lneto/internal" - "github.com/soypat/lneto/tcp" -) - -var _ StackNode = (*NodeTCPListener)(nil) - -type tcpPool interface { - GetTCP() (*tcp.Conn, tcp.Value) - PutTCP(*tcp.Conn) -} - -type NodeTCPListener struct { - connID uint64 - // ready have received a - ready []*tcp.Conn - accepted []*tcp.Conn - - port uint16 - poolGet func() (*tcp.Conn, tcp.Value) - poolReturn func(*tcp.Conn) -} - -// LocalPort implements [StackNode]. -func (listener *NodeTCPListener) LocalPort() uint16 { return listener.port } - -// ConnectionID implements [StackNode]. -func (listener *NodeTCPListener) ConnectionID() *uint64 { return &listener.connID } - -// Protocol implements [StackNode]. -func (listener *NodeTCPListener) Protocol() uint64 { return uint64(lneto.IPProtoTCP) } - -func (listener *NodeTCPListener) Close() error { - if listener.isClosed() { - return errors.New("already closed") - } - listener.connID++ - listener.port = 0 - return nil -} - -func (listener *NodeTCPListener) Reset(port uint16, pool tcpPool) error { - if port == 0 { - return errZeroPort - } else if pool == nil { - return errors.New("nil TCP pool") - } - *listener = NodeTCPListener{ - connID: listener.connID + 1, - port: port, - poolGet: pool.GetTCP, - poolReturn: pool.PutTCP, - ready: listener.ready[:0], - accepted: listener.accepted[:0], - } - return nil -} - -func (listener *NodeTCPListener) NumberOfReadyToAccept() (nready int) { - if listener.isClosed() { - return 0 - } - for _, conn := range listener.ready { - if conn == nil { - continue - } - nready++ - } - return nready -} - -func (listener *NodeTCPListener) TryAccept() (*tcp.Conn, error) { - if listener.isClosed() { - return nil, net.ErrClosed - } - listener.maintainConns() - for i, conn := range listener.ready { - if conn == nil { - continue - } - listener.accepted = append(listener.accepted, conn) - listener.ready[i] = nil // discard from ready. - return conn, nil - } - return nil, errors.New("no conns available") -} - -// Encapsulate implements [StackNode]. -func (listener *NodeTCPListener) Encapsulate(carrierData []byte, offsetToIP, offsetToFrame int) (int, error) { - if listener.isClosed() { - return 0, net.ErrClosed - } - for i, conn := range listener.accepted { - if conn == nil { - continue - } - n, err := conn.Encapsulate(carrierData, offsetToIP, offsetToFrame) - if err != nil { - err = listener.maintainConn(listener.accepted, i, err) - } - if n == 0 { - continue - } - return n, err - } - return 0, nil -} - -// Demux implements [StackNode]. -func (listener *NodeTCPListener) Demux(carrierData []byte, tcpFrameOffset int) error { - if listener.isClosed() { - return net.ErrClosed - } - tfrm, err := tcp.NewFrame(carrierData[tcpFrameOffset:]) - if err != nil { - return err - } - srcaddr, _, _, _, err := internal.GetIPAddr(carrierData) - if err != nil { - return err - } - dst := tfrm.DestinationPort() - if dst != listener.port { - return errors.New("not our port") - } - src := tfrm.SourcePort() - // Try to demux in accepted: - demuxed, err := listener.tryDemux(listener.accepted, src, srcaddr, carrierData, tcpFrameOffset) - if demuxed { - return err - } - demuxed, err = listener.tryDemux(listener.ready, src, srcaddr, carrierData, tcpFrameOffset) - if demuxed { - return err - } - // Connection not in ready nor accepted. - _, flags := tfrm.OffsetAndFlags() - if flags != tcp.FlagSYN { - return nil // Not a synchronizing packet, drop it. - } - conn, iss := listener.poolGet() - if conn == nil { - slog.Error("tcpListener:no-free-conn") - return nil - } - err = conn.OpenListen(dst, iss) - if err != nil { - slog.Error("NodeTCPListener:open", slog.String("err", err.Error())) - return err // This should not happend - } - err = conn.Demux(carrierData, tcpFrameOffset) - if err != nil { - conn.Abort() - slog.Error("NodeTCPListener:demux", slog.String("err", err.Error())) - return nil - } - listener.ready = append(listener.ready, conn) - return nil -} - -func (listener *NodeTCPListener) tryDemux(conns []*tcp.Conn, remotePort uint16, remoteAddr, carrierData []byte, tcpFrameOffset int) (demuxed bool, err error) { - idx := getConn(conns, remotePort, remoteAddr) - if idx >= 0 { - err := conns[idx].Demux(carrierData, tcpFrameOffset) - if err != nil { - err = listener.maintainConn(conns, idx, err) - } - return true, err - } - return false, nil -} - -func (listener *NodeTCPListener) maintainAccepted(connIdx int, err error) { - if err == net.ErrClosed { - conn := listener.accepted[connIdx] - listener.poolReturn(conn) - listener.accepted[connIdx] = nil - } -} - -func (listener *NodeTCPListener) isClosed() bool { - return listener.port == 0 -} - -func (listener *NodeTCPListener) maintainConns() { - listener.accepted = removeZeros(listener.accepted) - listener.ready = removeZeros(listener.ready) -} - -func removeZeros[S ~[]E, E comparable](s S) S { - var z E - putIdx := 0 - for i := range s { - if s[i] != z { - s[putIdx] = s[i] - putIdx++ - } - } - return s[:putIdx] -} - -func getConn(conns []*tcp.Conn, remotePort uint16, remoteAddr []byte) int { - for i, conn := range conns { - if conn == nil { - continue - } - gotPort := conn.RemotePort() - gotaddr := conn.RemoteAddr() - if remotePort == gotPort && bytes.Equal(remoteAddr, gotaddr) { - return i - } - } - return -1 -} - -func (listener *NodeTCPListener) maintainConn(conns []*tcp.Conn, idx int, err error) error { - if err == net.ErrClosed { - println("CLOSING CONN") - conn := conns[idx] - listener.poolReturn(conn) - conns[idx] = nil - return nil // avoid closing listener entirely. - } - return err -} diff --git a/internet/stackbasic_test.go b/internet/stackbasic_test.go index 2e93702..c86552c 100644 --- a/internet/stackbasic_test.go +++ b/internet/stackbasic_test.go @@ -60,6 +60,11 @@ func expectExchange(t *testing.T, from, to *StackIP, buf []byte) { func setupClientServerEstablished(t *testing.T, rng *rand.Rand, client, server *StackIP, connClient, connServer *tcp.Conn) { t.Helper() setupClientServer(t, rng, client, server, connClient, connServer) + testClientServerEstablish(t, client, server, connClient, connServer) +} + +func testClientServerEstablish(t *testing.T, client, server *StackIP, connClient, connServer *tcp.Conn) { + t.Helper() var buf [2048]byte nextToSend := client nextToRecv := server diff --git a/internet/tcplistener_test.go b/internet/tcplistener_test.go new file mode 100644 index 0000000..300bfac --- /dev/null +++ b/internet/tcplistener_test.go @@ -0,0 +1,398 @@ +package internet + +import ( + "math/rand" + "net/netip" + "testing" + + "github.com/soypat/lneto/tcp" +) + +func TestListener_SingleConnection(t *testing.T) { + rng := rand.New(rand.NewSource(1)) + var clientStack, serverStack StackIP + var clientConn, serverConn tcp.Conn + var listener tcp.Listener + + pool := newMockTCPPool(1, 3, 2048) + + // Use existing setup but replace server's conn registration with listener. + setupClientServer(t, rng, &clientStack, &serverStack, &clientConn, &serverConn) + serverConn.Abort() + serverPort := uint16(80) + if err := listener.Reset(serverPort, pool); err != nil { + t.Fatal(err) + } + if err := serverStack.Register(&listener); err != nil { + t.Fatal(err) + } + + var buf [2048]byte + + // Complete full handshake before TryAccept (TryAccept only works for ESTABLISHED). + // Client sends SYN. + expectExchange(t, &clientStack, &serverStack, buf[:]) + if listener.NumberOfReadyToAccept() != 0 { + t.Fatalf("after SYN: expected 0 ready (not established yet), got %d", listener.NumberOfReadyToAccept()) + } + // Server sends SYN-ACK. + expectExchange(t, &serverStack, &clientStack, buf[:]) + if listener.NumberOfReadyToAccept() != 0 { + t.Fatalf("after SYN: expected 0 ready (not established yet), got %d", listener.NumberOfReadyToAccept()) + } + // Client sends ACK. + expectExchange(t, &clientStack, &serverStack, buf[:]) + + // Now connection is ESTABLISHED, TryAccept should work. + if listener.NumberOfReadyToAccept() != 1 { + t.Fatalf("after handshake: expected 1 ready, got %d", listener.NumberOfReadyToAccept()) + } + acceptedConn, err := listener.TryAccept() + if err != nil { + t.Fatalf("TryAccept: %v", err) + } + if listener.NumberOfReadyToAccept() != 0 { + t.Fatalf("after accept: expected 0 ready, got %d", listener.NumberOfReadyToAccept()) + } + if acceptedConn.State() != tcp.StateEstablished { + t.Fatalf("accepted conn: expected StateEstablished, got %s", acceptedConn.State()) + } + if clientConn.State() != tcp.StateEstablished { + t.Fatalf("client conn: expected StateEstablished, got %s", clientConn.State()) + } +} + +func TestListener_AcceptAfterEstablished(t *testing.T) { + rng := rand.New(rand.NewSource(1)) + var client1Stack, serverStack StackIP + var client1Conn, serverConn tcp.Conn + var listener tcp.Listener + pool := newMockTCPPool(2, 3, 2048) + + // Setup server with listener. + setupClientServer(t, rng, &client1Stack, &serverStack, &client1Conn, &serverConn) + serverConn.Abort() + serverPort := uint16(80) + if err := listener.Reset(serverPort, pool); err != nil { + t.Fatal(err) + } + if err := serverStack.Register(&listener); err != nil { + t.Fatal(err) + } + + var buf [2048]byte + + // Complete full handshake for client1. + expectExchange(t, &client1Stack, &serverStack, buf[:]) // SYN + expectExchange(t, &serverStack, &client1Stack, buf[:]) // SYN-ACK + expectExchange(t, &client1Stack, &serverStack, buf[:]) // ACK + + // Now TryAccept client1. + if listener.NumberOfReadyToAccept() != 1 { + t.Fatalf("after client1 handshake: expected 1 ready, got %d", listener.NumberOfReadyToAccept()) + } + accepted1, err := listener.TryAccept() + if err != nil { + t.Fatalf("TryAccept client1: %v", err) + } else if listener.NumberOfReadyToAccept() != 0 { + t.Fatalf("after accepting conn: expected 0 ready, got %d", listener.NumberOfReadyToAccept()) + } + if accepted1.State() != tcp.StateEstablished { + t.Fatalf("accepted1: expected StateEstablished, got %s", accepted1.State()) + } + + // Setup second client and verify we can still accept. + var client2Stack StackIP + var client2Conn tcp.Conn + setupClient(t, &client2Stack, &client2Conn, serverStack.Addr(), serverPort, 1338) + + // Complete full handshake for client2. + expectExchange(t, &client2Stack, &serverStack, buf[:]) // SYN + expectExchange(t, &serverStack, &client2Stack, buf[:]) // SYN-ACK + expectExchange(t, &client2Stack, &serverStack, buf[:]) // ACK + + // Now TryAccept client2. + if listener.NumberOfReadyToAccept() != 1 { + t.Fatalf("after client2 handshake: expected 1 ready, got %d", listener.NumberOfReadyToAccept()) + } + accepted2, err := listener.TryAccept() + if err != nil { + t.Fatalf("TryAccept client2: %v", err) + } else if listener.NumberOfReadyToAccept() != 0 { + t.Fatalf("after client2 accept: expected 0 ready, got %d", listener.NumberOfReadyToAccept()) + } + if accepted2.State() != tcp.StateEstablished { + t.Fatalf("accepted2: expected StateEstablished, got %s", accepted2.State()) + } +} + +func TestListener_MultiConn(t *testing.T) { + const numClients = 5 + rng := rand.New(rand.NewSource(1)) + var serverStack StackIP + var serverConn tcp.Conn + var listener tcp.Listener + pool := newMockTCPPool(numClients, 3, 2048) + + // Create slices for clients. + clientStacks := make([]StackIP, numClients) + clientConns := make([]tcp.Conn, numClients) + acceptedConns := make([]*tcp.Conn, numClients) + + // Setup server with listener using setupClientServer for first client to get server configured. + setupClientServer(t, rng, &clientStacks[0], &serverStack, &clientConns[0], &serverConn) + serverConn.Abort() + serverPort := uint16(80) + if err := listener.Reset(serverPort, pool); err != nil { + t.Fatal(err) + } + if err := serverStack.Register(&listener); err != nil { + t.Fatal(err) + } + + // Setup remaining clients. + for i := 1; i < numClients; i++ { + clientPort := uint16(1337 + i) + setupClient(t, &clientStacks[i], &clientConns[i], serverStack.Addr(), serverPort, clientPort) + } + + var buf [2048]byte + + // Complete full handshakes for all clients. + for i := 0; i < numClients; i++ { + expectExchange(t, &clientStacks[i], &serverStack, buf[:]) // SYN + expectExchange(t, &serverStack, &clientStacks[i], buf[:]) // SYN-ACK + expectExchange(t, &clientStacks[i], &serverStack, buf[:]) // ACK + } + if listener.NumberOfReadyToAccept() != numClients { + t.Fatalf("after all handshakes: expected %d ready, got %d", numClients, listener.NumberOfReadyToAccept()) + } + if pool.NumberOfAcquired() != numClients { + t.Fatalf("pool should have %d acquired, got %d", numClients, pool.NumberOfAcquired()) + } + + // Accept all connections. + for i := 0; i < numClients; i++ { + var err error + acceptedConns[i], err = listener.TryAccept() + if err != nil { + t.Fatalf("TryAccept client %d: %v", i, err) + } + } + if listener.NumberOfReadyToAccept() != 0 { + t.Fatalf("after all accepts: expected 0 ready, got %d", listener.NumberOfReadyToAccept()) + } + + // Verify all connections established. + for i := 0; i < numClients; i++ { + if clientConns[i].State() != tcp.StateEstablished { + t.Errorf("client %d: expected StateEstablished, got %s", i, clientConns[i].State()) + } + if acceptedConns[i].State() != tcp.StateEstablished { + t.Errorf("accepted %d: expected StateEstablished, got %s", i, acceptedConns[i].State()) + } + } + + // Test data exchange: client -> server. + for i := 0; i < numClients; i++ { + msg := []byte("hello from client " + string('0'+byte(i))) + n, err := clientConns[i].Write(msg) + if err != nil { + t.Fatalf("client %d write: %v", i, err) + } + if n != len(msg) { + t.Fatalf("client %d write: wrote %d, expected %d", i, n, len(msg)) + } + } + + // Exchange data packets from all clients to server. + for i := 0; i < numClients; i++ { + expectExchange(t, &clientStacks[i], &serverStack, buf[:]) + } + + // Read data on server side and verify. + for i := 0; i < numClients; i++ { + expected := "hello from client " + string('0'+byte(i)) + var readBuf [64]byte + n, err := acceptedConns[i].Read(readBuf[:]) + if err != nil { + t.Fatalf("server read %d: %v", i, err) + } + if string(readBuf[:n]) != expected { + t.Errorf("server read %d: got %q, expected %q", i, string(readBuf[:n]), expected) + } + } + + // Test data exchange: server -> client. + for i := 0; i < numClients; i++ { + msg := []byte("reply to client " + string('0'+byte(i))) + n, err := acceptedConns[i].Write(msg) + if err != nil { + t.Fatalf("server %d write: %v", i, err) + } + if n != len(msg) { + t.Fatalf("server %d write: wrote %d, expected %d", i, n, len(msg)) + } + } + + // Exchange data packets from server to all clients. + for i := 0; i < numClients; i++ { + expectExchange(t, &serverStack, &clientStacks[i], buf[:]) + } + + // Read responses on client side and verify. + for i := 0; i < numClients; i++ { + expected := "reply to client " + string('0'+byte(i)) + var readBuf [64]byte + n, err := clientConns[i].Read(readBuf[:]) + if err != nil { + t.Fatalf("client read %d: %v", i, err) + } + if string(readBuf[:n]) != expected { + t.Errorf("client read %d: got %q, expected %q", i, string(readBuf[:n]), expected) + } + } + + // Close connections, alternating between client-initiated and server-initiated. + for i := 0; i < numClients; i++ { + var closer, responder *StackIP + var closerConn, responderConn *tcp.Conn + var serverClosed bool + whoCloses := "client" + whoResponds := "server" + expectStates := func(ctx string, wantCloserState, wantResponderState tcp.State) { + t.Helper() + if closerConn.State() != wantCloserState { + t.Errorf("%s: %s closer want %s, got %s", ctx, whoCloses, wantCloserState, closerConn.State()) + } + if responderConn.State() != wantResponderState { + t.Errorf("%s: %s respon want %s, got %s", ctx, whoResponds, wantResponderState, responderConn.State()) + } + } + if i%2 == 0 { + // Client initiates close. + closer, responder = &clientStacks[i], &serverStack + closerConn, responderConn = &clientConns[i], acceptedConns[i] + } else { + // Server initiates close. + serverClosed = true + whoCloses, whoResponds = whoResponds, whoCloses + closer, responder = &serverStack, &clientStacks[i] + closerConn, responderConn = acceptedConns[i], &clientConns[i] + } + _ = serverClosed // Used for context in debugging. + + // Closer calls Close(), FIN not sent yet. + if err := closerConn.Close(); err != nil { + t.Fatalf("conn %d close: %v", i, err) + } + expectStates("after-close()", tcp.StateEstablished, tcp.StateEstablished) + + // Closer sends FIN -> responder receives, goes to CLOSE-WAIT. + expectExchange(t, closer, responder, buf[:]) + expectStates("after-FIN", tcp.StateFinWait1, tcp.StateCloseWait) + + // Responder sends ACK -> closer goes to FIN-WAIT-2. + expectExchange(t, responder, closer, buf[:]) + expectStates("after-ACK", tcp.StateFinWait2, tcp.StateCloseWait) + + // Responder closes and sends FIN -> closer goes to TIME-WAIT. + if err := responderConn.Close(); err != nil { + t.Fatalf("conn %d responder close: %v", i, err) + } + expectExchange(t, responder, closer, buf[:]) + expectStates("after-resp-FIN", tcp.StateTimeWait, tcp.StateLastAck) + + // Closer sends final ACK -> responder goes to CLOSED. + expectExchange(t, closer, responder, buf[:]) + expectStates("after-final-ACK", tcp.StateTimeWait, tcp.StateClosed) + } +} + +// tryExchange attempts an exchange but doesn't fail if no data to send. +func tryExchange(t *testing.T, from, to *StackIP, buf []byte) { + t.Helper() + n, err := from.Encapsulate(buf, -1, 0) + if err != nil || n == 0 { + return // No data to send. + } + _ = to.Demux(buf[:n], 0) // Ignore errors during close. +} + +func setupClient(t *testing.T, client *StackIP, conn *tcp.Conn, serverAddr netip.Addr, serverPort, clientPort uint16) { + t.Helper() + bufsize := 2048 + clientIP := netip.AddrFrom4([4]byte{192, 168, 1, byte(clientPort % 256)}) + client.Reset(clientIP, 1) + err := conn.Configure(tcp.ConnConfig{ + RxBuf: make([]byte, bufsize), + TxBuf: make([]byte, bufsize), + TxPacketQueueSize: 3, + }) + if err != nil { + t.Fatal(err) + } + serverAddrPort := netip.AddrPortFrom(serverAddr, serverPort) + err = conn.OpenActive(clientPort, serverAddrPort, 100) + if err != nil { + t.Fatal(err) + } + err = client.Register(conn) + if err != nil { + t.Fatal(err) + } +} + +// mockTCPPool implements tcpPool for testing. +type mockTCPPool struct { + naqcuired int + conns []tcp.Conn + acquired []bool + nextISS tcp.Value +} + +func newMockTCPPool(n, queuesize, bufsize int) *mockTCPPool { + pool := &mockTCPPool{ + acquired: make([]bool, n), + conns: make([]tcp.Conn, n), + } + for i := range pool.conns { + err := pool.conns[i].Configure(tcp.ConnConfig{ + RxBuf: make([]byte, bufsize), + TxBuf: make([]byte, bufsize), + TxPacketQueueSize: queuesize, + }) + if err != nil { + panic(err) + } + } + return pool +} + +func (p *mockTCPPool) GetTCP() (*tcp.Conn, tcp.Value) { + for i := range p.conns { + if !p.acquired[i] { + p.acquired[i] = true + p.nextISS += 1000 + p.naqcuired++ + return &p.conns[i], p.nextISS + } + } + return nil, 0 +} + +func (p *mockTCPPool) PutTCP(conn *tcp.Conn) { + for i := range p.conns { + if &p.conns[i] == conn { + p.conns[i].Abort() + p.acquired[i] = false + p.naqcuired-- + return + } + } + panic("conn does not belong to this pool") +} + +func (p *mockTCPPool) NumberOfAcquired() int { + return p.naqcuired +} diff --git a/tcp/conn.go b/tcp/conn.go index e5e5580..e83dea5 100644 --- a/tcp/conn.go +++ b/tcp/conn.go @@ -249,7 +249,7 @@ func (conn *Conn) Read(b []byte) (int, error) { } conn.trace("TCPConn.Read:start") backoff := internal.NewBackoff(internal.BackoffTCPConn) - for conn.h.BufferedInput() == 0 && conn.State() == StateEstablished { + for conn.BufferedInput() == 0 && conn.State() == StateEstablished { if err := conn.checkPipe(connid, &conn.rdead); err != nil { return 0, err } diff --git a/tcp/control.go b/tcp/control.go index e767e41..ec53190 100644 --- a/tcp/control.go +++ b/tcp/control.go @@ -533,7 +533,7 @@ func (tcb *ControlBlock) Close() (err error) { err = errConnNotExist case StateCloseWait: tcb._state = StateLastAck - tcb.pending = [2]Flags{FlagFIN, FlagACK} + tcb.pending = [2]Flags{FlagFIN | FlagACK, 0} case StateListen, StateSynSent: // In Listen State there is no established connection. // In SynSent the remote endpoint is not yet synchronized and upon receiving an RST will abort connection. diff --git a/tcp/handler.go b/tcp/handler.go index 155f447..f50611e 100644 --- a/tcp/handler.go +++ b/tcp/handler.go @@ -83,13 +83,12 @@ func (h *Handler) RemotePort() uint16 { // OpenActive opens an "active" TCP connection to a known remote port. The caller holds knowledge of the IP address. // OpenActive is used by TCP Clients to initiate a connection. func (h *Handler) OpenActive(localPort, remotePort uint16, iss Value) error { - if h.bufRx.Size() < minBufferSize || h.bufTx.Size() < minBufferSize { + if remotePort == 0 { + return lneto.ErrZeroDestination + } else if h.bufRx.Size() < minBufferSize || h.bufTx.Size() < minBufferSize { return errBufferTooSmall - } - if h.scb.State() != StateClosed && h.scb.State() != StateTimeWait { + } else if h.scb.State() != StateClosed && h.scb.State() != StateTimeWait { return errNeedClosedTCBToOpen - } else if remotePort == 0 { - return errors.New("zero port on open call") } // reset/Abort prepares a SCB for active connection by resetting state to closed. h.scb.reset() @@ -101,7 +100,9 @@ func (h *Handler) OpenActive(localPort, remotePort uint16, iss Value) error { // OpenListen prepares a passive TCP connection where the Handler acts as a server. // OpenListen is used by TCP Servers to begin listening for remote connections. func (h *Handler) OpenListen(localPort uint16, iss Value) error { - if h.bufRx.Size() < minBufferSize || h.bufTx.Size() < minBufferSize { + if localPort == 0 { + return lneto.ErrZeroSource + } else if h.bufRx.Size() < minBufferSize || h.bufTx.Size() < minBufferSize { return errBufferTooSmall } // Open will fail unless SCB in closed state. diff --git a/tcp/internal_test.go b/tcp/internal_test.go index a2cb593..b29b925 100644 --- a/tcp/internal_test.go +++ b/tcp/internal_test.go @@ -6,72 +6,162 @@ import ( "testing" ) +const logExchange = false + // Here we define internal testing helpers that may be used in any *_test.go file // but are not exported. -// Exchange represents a single exchange of segments. -type Exchange struct { - Outgoing *Segment - Incoming *Segment - WantPending *Segment // Expected pending segment. If nil not checked. - WantState State // Expected end state. - WantPeerState State // Expected end state of peer. Not necessary when calling HelperExchange but can aid with logging information. +// ExchangeTest defines a complete TCP exchange scenario with initial state for both peers. +// Use Run() to execute the test from both perspectives, or RunA()/RunB() individually. +type ExchangeTest struct { + ISSA Value // Initial Send Sequence for peer A. + ISSB Value // Initial Send Sequence for peer B. + WindowA Size // A's receive window size. + WindowB Size // B's receive window size. + InitStateA State // A's state before exchanges. + InitStateB State // B's state before exchanges. + Steps []SegmentStep +} +type StepAction uint8 + +const ( + _ StepAction = iota + StepASends + StepBSends + StepACloses + StepBCloses +) + +// SegmentStep defines a single segment exchange with resulting states for both peers. +type SegmentStep struct { + Seg Segment // The segment being exchanged. + Action StepAction + + // States after the segment is processed. + AState State + BState State + + // Pending segments after the step (nil if none expected). + APending *Segment + BPending *Segment } -func (tcb *ControlBlock) HelperExchange(t *testing.T, exchange []Exchange) { +// Run executes the test from both peers' perspectives as subtests. +func (et ExchangeTest) Run(t *testing.T) { + t.Helper() + t.Run("PeerA", func(t *testing.T) { + t.Helper() + et.RunA(t) + }) + t.Run("PeerB", func(t *testing.T) { + t.Helper() + et.RunB(t) + }) +} + +// RunA executes the test from peer A's perspective. +func (et ExchangeTest) RunA(t *testing.T) { + t.Helper() + var tcb ControlBlock + tcb.HelperInitState(et.InitStateA, et.ISSA, et.ISSA, et.WindowA) + if et.InitStateA.hasIRS() { + tcb.HelperInitRcv(et.ISSB, et.ISSB, et.WindowB) + } + tcb.HelperSteps(t, et.Steps, true) +} + +// RunB executes the test from peer B's perspective. +func (et ExchangeTest) RunB(t *testing.T) { + t.Helper() + var tcb ControlBlock + tcb.HelperInitState(et.InitStateB, et.ISSB, et.ISSB, et.WindowB) + if et.InitStateB.hasIRS() { + tcb.HelperInitRcv(et.ISSA, et.ISSA, et.WindowA) + } + tcb.HelperSteps(t, et.Steps, false) +} + +// HelperSteps processes segment steps from a specific peer's perspective, calling Close() when indicated. +func (tcb *ControlBlock) HelperSteps(t *testing.T, steps []SegmentStep, isPeerA bool) { t.Helper() var i int - var ex Exchange + var st SegmentStep defer func() { if t.Failed() { - t.Errorf("exchange failed:\nwant: %s\ngot: %s", - ex.RFC9293String(ex.WantState, ex.WantPeerState), - ex.RFC9293String(tcb._state, ex.WantPeerState), - ) + peer := "B" + if isPeerA { + peer = "A" + } + t.Errorf("step[%d] failed (peer %s)", i, peer) } }() - const pfx = "exchange" - t.Log(tcb._state, "Exchange start") - for i, ex = range exchange { - if ex.Outgoing != nil && ex.Incoming != nil { - t.Fatalf(pfx+"[%d] cannot send and receive in the same exchange, please split into two exchanges.", i) - } else if ex.Outgoing == nil && ex.Incoming == nil { - t.Fatalf(pfx+"[%d] must send or receive a segment.", i) + const pfx = "step" + if logExchange { + t.Log(tcb._state, "Steps start, isPeerA:", isPeerA) + } + for i, st = range steps { + // Determine if this peer should close before this step. + nop := isPeerA && st.Action == StepBCloses || !isPeerA && st.Action == StepACloses + if nop { + continue } - if ex.Outgoing != nil { - prevInflight := tcb.snd.inFlight() - err := tcb.Send(*ex.Outgoing) - gotSent := tcb.snd.inFlight() - prevInflight + switch st.Action { + default: + panic("unknown action") + case StepACloses, StepBCloses: + err := tcb.Close() if err != nil { - t.Fatalf(pfx+"[%d] snd: %s\nseg=%+v\nrcv=%+v\nsnd=%+v", i, err, *ex.Outgoing, tcb.rcv, tcb.snd) - } else if gotSent != ex.Outgoing.LEN() { - t.Fatalf(pfx+"[%d] snd: expected %d data sent, calculated inflight %d", i, ex.Outgoing.LEN(), gotSent) + t.Fatalf(pfx+"[%d] Close: %s", i, err) } - } - if ex.Incoming != nil { - err := tcb.Recv(*ex.Incoming) - if err != nil { - msg := fmt.Sprintf(pfx+"[%d] rcv: %s\nseg=%+v\nrcv=%+v\nsnd=%+v", i, err, *ex.Incoming, tcb.rcv, tcb.snd) - if IsDroppedErr(err) { - t.Log(msg) - } else { - t.Fatal(msg) + case StepASends, StepBSends: + // Determine if this peer sends or receives. + isSender := isPeerA && st.Action == StepASends || !isPeerA && st.Action == StepBSends + seg := st.Seg + if isSender { + prevInflight := tcb.snd.inFlight() + err := tcb.Send(seg) + gotSent := tcb.snd.inFlight() - prevInflight + if err != nil { + t.Fatalf(pfx+"[%d] snd: %s\nseg=%+v\nrcv=%+v\nsnd=%+v", i, err, seg, tcb.rcv, tcb.snd) + } else if gotSent != seg.LEN() { + t.Fatalf(pfx+"[%d] snd: expected %d data sent, calculated inflight %d", i, seg.LEN(), gotSent) + } + } else if tcb._state != StateTimeWait { // TODO: should we support receiving in TimeWait? + err := tcb.Recv(seg) + if err != nil { + msg := fmt.Sprintf(pfx+"[%d] rcv: %s\nseg=%+v\nrcv=%+v\nsnd=%+v", i, err, seg, tcb.rcv, tcb.snd) + if IsDroppedErr(err) { + t.Log(msg) + } else { + t.Fatal(msg) + } } } } - - t.Log(ex.RFC9293String(tcb._state, ex.WantPeerState)) + // Select expected state and pending based on which peer we are. + var wantState State + var wantPending *Segment + if isPeerA { + wantState = st.AState + wantPending = st.APending + } else { + wantState = st.BState + wantPending = st.BPending + } + if logExchange { + t.Logf(pfx+"[%d] state=%s (want=%s)", i, tcb._state, wantState) + } state := tcb.State() - if state != ex.WantState { - t.Errorf(pfx+"[%d] unexpected state:\n got=%s\nwant=%s", i, state, ex.WantState) + if state != wantState { + t.Errorf(pfx+"[%d] unexpected state:\n got=%s\nwant=%s", i, state, wantState) } pending, ok := tcb.PendingSegment(0) - if !ok && ex.WantPending != nil { - t.Fatalf(pfx+"[%d] pending:got none, want=%+v", i, *ex.WantPending) - } else if ex.WantPending != nil && pending != *ex.WantPending { - t.Fatalf(pfx+"[%d] pending:\n got=%+v\nwant=%+v", i, pending, *ex.WantPending) - } else if ok && ex.WantPending == nil { + if !ok && wantPending != nil { + t.Fatalf(pfx+"[%d] pending:got none, want=%+v", i, *wantPending) + } else if wantPending != nil && pending != *wantPending { + t.Fatalf(pfx+"[%d] pending:\n got=%+v\nwant=%+v", i, pending, *wantPending) + } else if ok && wantPending == nil { t.Fatalf(pfx+"[%d] pending:\n got=%+v\nwant=none", i, pending) } } @@ -191,3 +281,75 @@ func (ex *Exchange) RFC9293String(A, B State) string { } return StringExchange(seg, A, B, !sentByA) } + +// Exchange represents a single exchange of segments. +// TODO: replace [Exchange] tests with [ExchageTest]. +type Exchange struct { + Outgoing *Segment + Incoming *Segment + WantPending *Segment // Expected pending segment. If nil not checked. + WantState State // Expected end state. + WantPeerState State // Expected end state of peer. Not necessary when calling HelperExchange but can aid with logging information. +} + +func (tcb *ControlBlock) HelperExchange(t *testing.T, exchange []Exchange) { + t.Helper() + var i int + var ex Exchange + defer func() { + if t.Failed() { + t.Errorf("exchange failed:\nwant: %s\ngot: %s", + ex.RFC9293String(ex.WantState, ex.WantPeerState), + ex.RFC9293String(tcb._state, ex.WantPeerState), + ) + } + }() + const pfx = "exchange" + if logExchange { + t.Log(tcb._state, "Exchange start") + } + for i, ex = range exchange { + if ex.Outgoing != nil && ex.Incoming != nil { + t.Fatalf(pfx+"[%d] cannot send and receive in the same exchange, please split into two exchanges.", i) + } else if ex.Outgoing == nil && ex.Incoming == nil { + t.Fatalf(pfx+"[%d] must send or receive a segment.", i) + } + if ex.Outgoing != nil { + prevInflight := tcb.snd.inFlight() + err := tcb.Send(*ex.Outgoing) + gotSent := tcb.snd.inFlight() - prevInflight + if err != nil { + t.Fatalf(pfx+"[%d] snd: %s\nseg=%+v\nrcv=%+v\nsnd=%+v", i, err, *ex.Outgoing, tcb.rcv, tcb.snd) + } else if gotSent != ex.Outgoing.LEN() { + t.Fatalf(pfx+"[%d] snd: expected %d data sent, calculated inflight %d", i, ex.Outgoing.LEN(), gotSent) + } + } + if ex.Incoming != nil { + err := tcb.Recv(*ex.Incoming) + if err != nil { + msg := fmt.Sprintf(pfx+"[%d] rcv: %s\nseg=%+v\nrcv=%+v\nsnd=%+v", i, err, *ex.Incoming, tcb.rcv, tcb.snd) + if IsDroppedErr(err) { + t.Log(msg) + } else { + t.Fatal(msg) + } + } + } + if logExchange { + t.Log(ex.RFC9293String(tcb._state, ex.WantPeerState)) + } + + state := tcb.State() + if state != ex.WantState { + t.Errorf(pfx+"[%d] unexpected state:\n got=%s\nwant=%s", i, state, ex.WantState) + } + pending, ok := tcb.PendingSegment(0) + if !ok && ex.WantPending != nil { + t.Fatalf(pfx+"[%d] pending:got none, want=%+v", i, *ex.WantPending) + } else if ex.WantPending != nil && pending != *ex.WantPending { + t.Fatalf(pfx+"[%d] pending:\n got=%+v\nwant=%+v", i, pending, *ex.WantPending) + } else if ok && ex.WantPending == nil { + t.Fatalf(pfx+"[%d] pending:\n got=%+v\nwant=none", i, pending) + } + } +} diff --git a/tcp/listener.go b/tcp/listener.go new file mode 100644 index 0000000..a7f50c7 --- /dev/null +++ b/tcp/listener.go @@ -0,0 +1,256 @@ +package tcp + +import ( + "bytes" + "errors" + "log/slog" + "net" + "sync" + + "github.com/soypat/lneto" + "github.com/soypat/lneto/internal" +) + +// pool is a [sync.Pool] like +type pool interface { + GetTCP() (*Conn, Value) + PutTCP(*Conn) +} + +type Listener struct { + connID uint64 + mu sync.Mutex + // incoming stores connections that are potential candidates for acceptance. + incoming []*Conn + // accepted stores all connections that have been accepted and are open. + accepted []*Conn + port uint16 + poolGet func() (*Conn, Value) + poolReturn func(*Conn) +} + +// LocalPort implements [StackNode]. +func (listener *Listener) LocalPort() uint16 { + listener.mu.Lock() + defer listener.mu.Unlock() + return listener.port +} + +// ConnectionID implements [StackNode]. +func (listener *Listener) ConnectionID() *uint64 { return &listener.connID } + +// Protocol implements [StackNode]. +func (listener *Listener) Protocol() uint64 { return uint64(lneto.IPProtoTCP) } + +func (listener *Listener) Close() error { + listener.mu.Lock() + defer listener.mu.Unlock() + if listener.isClosed() { + return errors.New("already closed") + } + listener.connID++ + listener.port = 0 + return nil +} + +func (listener *Listener) Reset(port uint16, pool pool) error { + if port == 0 { + return errZeroDstPort + } else if pool == nil { + return errors.New("nil TCP pool") + } + listener.mu.Lock() + defer listener.mu.Unlock() + *listener = Listener{ + mu: listener.mu, + connID: listener.connID + 1, + port: port, + poolGet: pool.GetTCP, + poolReturn: pool.PutTCP, + incoming: listener.incoming[:0], + accepted: listener.accepted[:0], + } + return nil +} + +func (listener *Listener) NumberOfReadyToAccept() (nready int) { + listener.mu.Lock() + defer listener.mu.Unlock() + if listener.isClosed() { + return 0 + } + for _, conn := range listener.incoming { + if conn == nil || conn.State() != StateEstablished { + continue + } + nready++ + } + return nready +} + +// TryAccept polls the list of ready connections that have been established +func (listener *Listener) TryAccept() (*Conn, error) { + listener.mu.Lock() + defer listener.mu.Unlock() + if listener.isClosed() { + return nil, net.ErrClosed + } + listener.maintainConns() + for i, conn := range listener.incoming { + if conn == nil || conn.State() != StateEstablished { + continue + } + listener.accepted = append(listener.accepted, conn) + listener.incoming[i] = nil // discard from ready. + return conn, nil + } + return nil, errors.New("no conns available") +} + +// Encapsulate implements [StackNode]. +func (listener *Listener) Encapsulate(carrierData []byte, offsetToIP, offsetToFrame int) (int, error) { + listener.mu.Lock() + defer listener.mu.Unlock() + if listener.isClosed() { + return 0, net.ErrClosed + } + // First try incoming connections (for handshake SYN-ACK). + for i, conn := range listener.incoming { + if conn == nil || conn.State() == StateEstablished { + // Nil or already established. + continue + } + n, err := conn.Encapsulate(carrierData, offsetToIP, offsetToFrame) + if err != nil { + err = listener.maintainConn(listener.incoming, i, err) + } + if n == 0 { + continue + } + return n, err + } + // Then try accepted connections. + for i, conn := range listener.accepted { + if conn == nil { + continue + } + n, err := conn.Encapsulate(carrierData, offsetToIP, offsetToFrame) + if err != nil { + err = listener.maintainConn(listener.accepted, i, err) + } + if n == 0 { + continue + } + return n, err + } + return 0, nil +} + +// Demux implements [StackNode]. +func (listener *Listener) Demux(carrierData []byte, tcpFrameOffset int) error { + listener.mu.Lock() + defer listener.mu.Unlock() + if listener.isClosed() { + return net.ErrClosed + } + tfrm, err := NewFrame(carrierData[tcpFrameOffset:]) + if err != nil { + return err + } + srcaddr, _, _, _, err := internal.GetIPAddr(carrierData) + if err != nil { + return err + } + dst := tfrm.DestinationPort() + if dst != listener.port { + return errors.New("not our port") + } + src := tfrm.SourcePort() + // Try to demux in accepted: + demuxed, err := listener.tryDemux(listener.accepted, src, srcaddr, carrierData, tcpFrameOffset) + if demuxed { + return err + } + demuxed, err = listener.tryDemux(listener.incoming, src, srcaddr, carrierData, tcpFrameOffset) + if demuxed { + return err + } + // Connection not in ready nor accepted. + _, flags := tfrm.OffsetAndFlags() + if flags != FlagSYN { + return lneto.ErrPacketDrop // Not a synchronizing packet, drop it. + } + conn, iss := listener.poolGet() + if conn == nil { + slog.Error("tcpListener:no-free-conn") + return lneto.ErrPacketDrop + } + err = conn.OpenListen(dst, iss) + if err != nil { + listener.poolReturn(conn) + slog.Error("Listener:open", slog.String("err", err.Error())) + return err // This should not happend + } + err = conn.Demux(carrierData, tcpFrameOffset) + if err != nil { + listener.poolReturn(conn) + slog.Error("Listener:demux", slog.String("err", err.Error())) + return lneto.ErrPacketDrop + } + listener.incoming = append(listener.incoming, conn) + return nil +} + +func (listener *Listener) tryDemux(conns []*Conn, remotePort uint16, remoteAddr, carrierData []byte, tcpFrameOffset int) (demuxed bool, err error) { + idx := getConn(conns, remotePort, remoteAddr) + if idx >= 0 { + err := conns[idx].Demux(carrierData, tcpFrameOffset) + if err != nil { + err = listener.maintainConn(conns, idx, err) + } + return true, err + } + return false, nil +} + +func (listener *Listener) isClosed() bool { + return listener.port == 0 +} + +func (listener *Listener) maintainConns() { + listener.accepted = internal.DeleteZeroed(listener.accepted) + for i := range listener.incoming { + if listener.incoming[i] == nil { + continue + } + if listener.incoming[i].State() > StateEstablished || listener.incoming[i].State().IsClosed() { + // Something went wrong in handshake or pool aborted/closed the connection. + listener.poolReturn(listener.incoming[i]) + listener.incoming[i] = nil + } + } + listener.incoming = internal.DeleteZeroed(listener.incoming) +} + +func getConn(conns []*Conn, remotePort uint16, remoteAddr []byte) int { + for i, conn := range conns { + if conn == nil { + continue + } + gotPort := conn.RemotePort() + gotaddr := conn.RemoteAddr() + if remotePort == gotPort && bytes.Equal(remoteAddr, gotaddr) { + return i + } + } + return -1 +} + +func (listener *Listener) maintainConn(conns []*Conn, idx int, err error) error { + if err == net.ErrClosed { + listener.poolReturn(conns[idx]) + conns[idx] = nil + return nil // avoid closing listener entirely. + } + return err +} diff --git a/tcp/tcp_legacy_test.go b/tcp/tcp_legacy_test.go new file mode 100644 index 0000000..ec5bf71 --- /dev/null +++ b/tcp/tcp_legacy_test.go @@ -0,0 +1,665 @@ +package tcp_test + +import ( + "strconv" + "testing" + + "github.com/soypat/lneto" + "github.com/soypat/lneto/ethernet" + "github.com/soypat/lneto/ipv4" + "github.com/soypat/lneto/tcp" +) + +/* + Section 3.5 of RFC 9293: Basic 3-way handshake for connection synchronization. + TCP Peer A TCP Peer B + + 1. CLOSED LISTEN + + 2. SYN-SENT --> --> SYN-RECEIVED + + 3. ESTABLISHED <-- <-- SYN-RECEIVED + + 4. ESTABLISHED --> --> ESTABLISHED + + 5. ESTABLISHED --> --> ESTABLISHED +*/ +func TestExchange_rfc9293_figure6(t *testing.T) { + const issA, issB, windowA, windowB = 100, 300, 1000, 1000 + exchangeA := []tcp.Exchange{ + { // A sends SYN to B. + Outgoing: &tcp.Segment{SEQ: issA, Flags: tcp.FlagSYN, WND: windowA}, + WantState: tcp.StateSynSent, + WantPeerState: tcp.StateSynRcvd, + }, + { // A receives SYNACK from B thus establishing the connection on A's side. + Incoming: &tcp.Segment{SEQ: issB, ACK: issA + 1, Flags: SYNACK, WND: windowB}, + WantState: tcp.StateEstablished, + WantPending: &tcp.Segment{SEQ: issA + 1, ACK: issB + 1, Flags: tcp.FlagACK, WND: windowA}, + WantPeerState: tcp.StateSynRcvd, + }, + { // A sends ACK to B, which leaves connection established on their side. Three way handshake complete by now. + Outgoing: &tcp.Segment{SEQ: issA + 1, ACK: issB + 1, Flags: tcp.FlagACK, WND: windowA}, + WantState: tcp.StateEstablished, + WantPeerState: tcp.StateEstablished, + }, + } + var tcbA tcp.ControlBlock + tcbA.HelperInitState(tcp.StateSynSent, issA, issA, windowA) + tcbA.HelperExchange(t, exchangeA) + segA, ok := tcbA.PendingSegment(0) + if ok { + t.Error("unexpected Client pending segment after establishment: ", segA) + } + exchangeB := reverseExchange(exchangeA) + + var tcbB tcp.ControlBlock + tcbB.HelperInitState(tcp.StateListen, issB, issB, windowB) + tcbB.HelperExchange(t, exchangeB) // TODO remove [:3] after snd.UNA bugfix + segB, ok := tcbB.PendingSegment(0) + if ok { + t.Error("unexpected Listener pending segment after establishment: ", segB) + } +} + +/* + Section 3.5 of RFC 9293: Simultaneous Connection Synchronization (SYN). + TCP Peer A TCP Peer B + + 1. CLOSED CLOSED + + 2. SYN-SENT --> ... + + 3. SYN-RECEIVED <-- <-- SYN-SENT + + 4. ... --> SYN-RECEIVED + + 5. SYN-RECEIVED --> ... + + 6. ESTABLISHED <-- <-- SYN-RECEIVED + + 7. ... --> ESTABLISHED +*/ +func TestExchange_rfc9293_figure7(t *testing.T) { + const issA, issB, windowA, windowB = 100, 300, 1000, 1000 + exchangeA := []tcp.Exchange{ + 0: { // A sends SYN to B. + Outgoing: &tcp.Segment{SEQ: issA, Flags: tcp.FlagSYN, WND: windowA}, + WantState: tcp.StateSynSent, + }, + 1: { // A receives a SYN with no ACK from B. + Incoming: &tcp.Segment{SEQ: issB, Flags: tcp.FlagSYN, WND: windowB}, + WantState: tcp.StateSynRcvd, + WantPending: &tcp.Segment{SEQ: issA, ACK: issB + 1, Flags: SYNACK, WND: windowA}, + }, + 2: { // A sends SYNACK to B. + Outgoing: &tcp.Segment{SEQ: issA, ACK: issB + 1, Flags: SYNACK, WND: windowA}, + WantState: tcp.StateSynRcvd, + }, + 3: { // A receives ACK from B. + Incoming: &tcp.Segment{SEQ: issB, ACK: issA + 1, Flags: SYNACK, WND: windowA}, + WantState: tcp.StateEstablished, + }, + } + var tcbA tcp.ControlBlock + tcbA.HelperInitState(tcp.StateSynSent, issA, issA, windowA) + tcbA.HelperExchange(t, exchangeA) +} + +/* + Recovery from Old Duplicate SYN + TCP Peer A TCP Peer B + + 1. CLOSED LISTEN + + 2. SYN-SENT --> ... + + 3. (duplicate) ... --> SYN-RECEIVED + + 4. SYN-SENT <-- <-- SYN-RECEIVED + + 5. SYN-SENT --> --> LISTEN + + 6. ... --> SYN-RECEIVED + + 7. ESTABLISHED <-- <-- SYN-RECEIVED + + 8. ESTABLISHED --> --> ESTABLISHED +*/ +func TestExchange_rfc9293_figure8(t *testing.T) { + const issA, issB, windowA, windowB = 100, 300, 1000, 1000 + const issAold = 90 + const issBNew = issB + 100 + exchangeA := []tcp.Exchange{ + 0: { // A sends new SYN to B (which is not received). + Outgoing: &tcp.Segment{SEQ: issA, Flags: tcp.FlagSYN, WND: windowA}, + WantState: tcp.StateSynSent, + WantPeerState: tcp.StateSynRcvd, + }, + 1: { // Receive SYN from B acking an old "duplicate" SYN. + Incoming: &tcp.Segment{SEQ: issB, ACK: issAold + 1, Flags: SYNACK, WND: windowB}, + WantState: tcp.StateSynSent, + WantPending: &tcp.Segment{SEQ: issAold + 1, Flags: tcp.FlagRST, WND: windowA}, + WantPeerState: tcp.StateSynRcvd, + }, + 2: { // A sends RST to B and makes segment believable by using the old SEQ. + Outgoing: &tcp.Segment{SEQ: issAold + 1, Flags: tcp.FlagRST, WND: windowA}, + WantState: tcp.StateSynSent, + WantPeerState: tcp.StateListen, + }, + 3: { // A sends a duplicate SYN to B. + Outgoing: &tcp.Segment{SEQ: issA, Flags: tcp.FlagSYN, WND: windowA}, + WantState: tcp.StateSynSent, + WantPeerState: tcp.StateSynRcvd, + }, + 4: { // B SYNACKs new SYN. + Incoming: &tcp.Segment{SEQ: issBNew, ACK: issA + 1, Flags: SYNACK, WND: windowB}, + WantState: tcp.StateEstablished, + WantPending: &tcp.Segment{SEQ: issA + 1, ACK: issBNew + 1, Flags: tcp.FlagACK, WND: windowA}, + WantPeerState: tcp.StateSynRcvd, + }, + 5: { // B receives ACK from A. + Outgoing: &tcp.Segment{SEQ: issA + 1, ACK: issBNew + 1, Flags: tcp.FlagACK, WND: windowA}, + WantState: tcp.StateEstablished, + WantPeerState: tcp.StateEstablished, + }, + } + var tcbA tcp.ControlBlock + tcbA.HelperInitState(tcp.StateSynSent, issA, issA, windowA) + tcbA.HelperExchange(t, exchangeA) + + exchangeB := []tcp.Exchange{ + 0: { // B receives old SYN from A. + Incoming: &tcp.Segment{SEQ: issAold, Flags: tcp.FlagSYN, WND: windowA}, + WantState: tcp.StateSynRcvd, + WantPending: &tcp.Segment{SEQ: issB, ACK: issAold + 1, Flags: SYNACK, WND: windowB}, + }, + 1: { // B SYNACKs old SYN. + Outgoing: &tcp.Segment{SEQ: issB, ACK: issAold + 1, Flags: SYNACK, WND: windowB}, + WantState: tcp.StateSynRcvd, + }, + 2: { // B receives RST from A. + Incoming: &tcp.Segment{SEQ: issAold + 1, Flags: tcp.FlagRST, WND: windowA}, + WantState: tcp.StateListen, + }, + 3: { // B receives new SYN from A. + Incoming: &tcp.Segment{SEQ: issA, Flags: tcp.FlagSYN, WND: windowA}, + WantState: tcp.StateSynRcvd, + WantPending: &tcp.Segment{SEQ: issBNew, ACK: issA + 1, Flags: SYNACK, WND: windowB}, + }, + 4: { // B SYNACKs new SYN. + Outgoing: &tcp.Segment{SEQ: issBNew, ACK: issA + 1, Flags: SYNACK, WND: windowB}, + WantState: tcp.StateSynRcvd, + }, + 5: { // B receives ACK from A. + Incoming: &tcp.Segment{SEQ: issA + 1, ACK: issBNew + 1, Flags: tcp.FlagACK, WND: windowA}, + WantState: tcp.StateEstablished, + }, + } + var tcbB tcp.ControlBlock + tcbB.HelperInitState(tcp.StateListen, issB, issB, windowB) + tcbB.HelperExchange(t, exchangeB) +} + +/* + Figure 12: Normal Close Sequence + TCP Peer A TCP Peer B + 1. ESTABLISHED ESTABLISHED + + 2. (Close) + FIN-WAIT-1 --> --> CLOSE-WAIT + + 3. FIN-WAIT-2 <-- <-- CLOSE-WAIT + + 4. (Close) + TIME-WAIT <-- <-- LAST-ACK + + 5. TIME-WAIT --> --> CLOSED + + 6. (2 MSL) + CLOSED +*/ +func TestExchange_rfc9293_figure12(t *testing.T) { + const issA, issB, windowA, windowB = 100, 300, 1000, 1000 + exchangeA := []tcp.Exchange{ + 0: { // A sends FIN|ACK to B to begin closing connection. + Outgoing: &tcp.Segment{SEQ: issA, ACK: issB, Flags: FINACK, WND: windowA}, + WantState: tcp.StateFinWait1, + WantPeerState: tcp.StateCloseWait, + }, + 1: { // A receives ACK from B. + Incoming: &tcp.Segment{SEQ: issB, ACK: issA + 1, Flags: tcp.FlagACK, WND: windowB}, + WantState: tcp.StateFinWait2, + WantPeerState: tcp.StateCloseWait, + // TODO(soypat): WantPending should be nil here? Perhaps fix test by modifying rcvFinWait1 pending result. + WantPending: &tcp.Segment{SEQ: issA + 1, ACK: issB, Flags: tcp.FlagACK, WND: windowA}, + }, + 2: { // A receives FIN|ACK from B. + Incoming: &tcp.Segment{SEQ: issB, ACK: issA + 1, Flags: FINACK, WND: windowB}, + WantState: tcp.StateTimeWait, + WantPending: &tcp.Segment{SEQ: issA + 1, ACK: issB + 1, Flags: tcp.FlagACK, WND: windowA}, + WantPeerState: tcp.StateLastAck, + }, + 3: { // A sends ACK to B. + Outgoing: &tcp.Segment{SEQ: issA + 1, ACK: issB + 1, Flags: tcp.FlagACK, WND: windowA}, + WantState: tcp.StateTimeWait, // Technically we should be in TimeWait here. + WantPeerState: tcp.StateClosed, + }, + } + var tcbA tcp.ControlBlock + 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) +} + +/* +Figure 12: Normal Close Sequence from Peer B (passive close) perspective. + + TCP Peer A TCP Peer B + + 1. ESTABLISHED ESTABLISHED + + 2. (Close) + FIN-WAIT-1 --> --> CLOSE-WAIT + + 3. FIN-WAIT-2 <-- <-- CLOSE-WAIT + + 4. (Close) + TIME-WAIT <-- <-- LAST-ACK + + 5. TIME-WAIT --> --> CLOSED + +This test validates the passive close (B side) behavior where B receives +a FIN from A, acknowledges it, then later closes and sends its own FIN. +*/ +func TestExchange_rfc9293_figure12_peerB(t *testing.T) { + const issA, issB, windowA, windowB = 100, 300, 1000, 1000 + // Note: After B sends an ACK in CLOSE-WAIT, the implementation auto-queues FIN|ACK. + // This is an optimization that combines steps 3 and 4 of RFC 9293 Figure 12. + exchangeB := []tcp.Exchange{ + 0: { // B receives FIN|ACK from A, goes to CLOSE-WAIT with pending ACK. + Incoming: &tcp.Segment{SEQ: issA, ACK: issB, Flags: FINACK, WND: windowA}, + WantState: tcp.StateCloseWait, + WantPending: &tcp.Segment{SEQ: issB, ACK: issA + 1, Flags: tcp.FlagACK, WND: windowB}, + }, + 1: { // B sends ACK to A. Implementation auto-queues FIN|ACK for close. + Outgoing: &tcp.Segment{SEQ: issB, ACK: issA + 1, Flags: tcp.FlagACK, WND: windowB}, + WantState: tcp.StateCloseWait, + WantPending: &tcp.Segment{SEQ: issB, ACK: issA + 1, Flags: FINACK, WND: windowB}, + }, + 2: { // B sends FIN|ACK to A, goes to LAST-ACK. + Outgoing: &tcp.Segment{SEQ: issB, ACK: issA + 1, Flags: FINACK, WND: windowB}, + WantState: tcp.StateLastAck, + }, + 3: { // B receives final ACK from A, goes to CLOSED. + Incoming: &tcp.Segment{SEQ: issA + 1, ACK: issB + 1, Flags: tcp.FlagACK, WND: windowA}, + WantState: tcp.StateClosed, + }, + } + var tcbB tcp.ControlBlock + tcbB.HelperInitState(tcp.StateEstablished, issB, issB, windowB) + tcbB.HelperInitRcv(issA, issA, windowA) + tcbB.HelperExchange(t, exchangeB) +} + +/* + Figure 12: Simultaneous Close Sequence + TCP Peer A TCP Peer B + + 1. ESTABLISHED ESTABLISHED + + 2. (Close) (Close) + FIN-WAIT-1 --> ... FIN-WAIT-1 + <-- <-- + ... --> + + 3. CLOSING --> ... CLOSING + <-- <-- + ... --> + + 4. TIME-WAIT TIME-WAIT + (2 MSL) (2 MSL) + CLOSED CLOSED +*/ +func TestExchange_rfc9293_figure13(t *testing.T) { + const issA, issB, windowA, windowB = 100, 300, 1000, 1000 + exchangeA := []tcp.Exchange{ + 0: { // A sends FIN|ACK to B to begin closing connection. + Outgoing: &tcp.Segment{SEQ: issA, ACK: issB, Flags: FINACK, WND: windowA}, + WantState: tcp.StateFinWait1, + }, + 1: { // A receives FIN|ACK from B, who sent packet before receiving A's FINACK. + Incoming: &tcp.Segment{SEQ: issB, ACK: issA, Flags: FINACK, WND: windowB}, + WantState: tcp.StateClosing, + WantPending: &tcp.Segment{SEQ: issA + 1, ACK: issB + 1, Flags: tcp.FlagACK, WND: windowA}, + }, + 2: { // A sends ACK to B. + Outgoing: &tcp.Segment{SEQ: issA + 1, ACK: issB + 1, Flags: tcp.FlagACK, WND: windowA}, + WantState: tcp.StateTimeWait, + }, + } + var tcbA tcp.ControlBlock + tcbA.HelperInitState(tcp.StateEstablished, issA, issA, windowA) + tcbA.HelperInitRcv(issB, issB, windowB) + tcbA.HelperExchange(t, exchangeA) + + // No need to test B since exchange is completely symmetric. +} + +// Check no duplicate ack is sent during establishment. +func TestExchange_noDupAckDuringEstablished(t *testing.T) { + var tcbA tcp.ControlBlock + const issA, issB, windowA, windowB = 300, 334222749, 256, 64240 + synseg := tcp.ClientSynSegment(issA, windowA) + + // err := tcbA.Open(issA, issA, tcp.StateSynSent) + tcbA.SetRecvWindow(windowA) + // if err != nil { + // t.Fatal(err) + // } + establishA := []tcp.Exchange{ + 0: { // A sends SYN to B. + Outgoing: &synseg, + WantState: tcp.StateSynSent, + }, + 1: { // B sends SYN to A. + Incoming: &tcp.Segment{SEQ: issB, ACK: 0, WND: windowB, Flags: tcp.FlagSYN}, + WantPending: &tcp.Segment{SEQ: issA, ACK: issB + 1, WND: windowA, Flags: SYNACK}, + WantState: tcp.StateSynRcvd, + }, + 2: { // Send SYNACK to B. + Outgoing: &tcp.Segment{SEQ: issA, ACK: issB + 1, WND: windowA, Flags: SYNACK}, + WantState: tcp.StateSynRcvd, + }, + 3: { // B ACKs SYNACK, thus establishing the connection on both sides. + Incoming: &tcp.Segment{SEQ: issB + 1, ACK: issA + 1, WND: windowB, Flags: tcp.FlagACK}, + WantState: tcp.StateEstablished, + }, + } + tcbA.HelperExchange(t, establishA) + if tcbA.State() != tcp.StateEstablished { + t.Fatal("expected established state") + } + checkNoPending(t, &tcbA) + const datasize = 5 + dataExA := []tcp.Exchange{ + 0: { // B sends PSH|ACK to A with data. + Incoming: &tcp.Segment{SEQ: issB + 1, ACK: issA + 1, WND: windowB, Flags: PSHACK, DATALEN: datasize}, + WantPending: &tcp.Segment{SEQ: issA + 1, ACK: issB + 1 + datasize, WND: windowA, Flags: tcp.FlagACK}, + WantState: tcp.StateEstablished, + }, + 1: { // A ACKs B's data. + Outgoing: &tcp.Segment{SEQ: issA + 1, ACK: issB + 1 + datasize, WND: windowA, Flags: tcp.FlagACK}, + WantState: tcp.StateEstablished, + }, + 2: { // A sends PSH|ACK to B with data, same amount, as if echoing. + Outgoing: &tcp.Segment{SEQ: issA + 1, ACK: issB + 1 + datasize, WND: windowA, Flags: PSHACK, DATALEN: datasize}, + WantState: tcp.StateEstablished, + }, + // 3: { // B ACKs A's data. + // Incoming: &tcp.Segment{SEQ: issB + 1 + datasize, ACK: issA + 1 + datasize, WND: windowB, Flags: tcp.FlagACK}, + // WantPending: nil, + // WantState: tcp.StateEstablished, + // }, + } + tcbA.HelperExchange(t, dataExA) + checkNoPending(t, &tcbA) + tcbA.Recv(tcp.Segment{SEQ: issB + 1 + datasize, ACK: issA + 1 + datasize, WND: windowB, Flags: tcp.FlagACK}) + checkNoPending(t, &tcbA) +} + +// This test reenacts a full client-server interaction in the sending and receiving +// of the 12 byte message "hello world\n" over TCP. +func TestExchange_helloworld(t *testing.T) { + // Client Transmission Control Block. + var tcbA tcp.ControlBlock + const windowA, windowB = 502, 4096 + const issA, issB = 0x5e722b7d, 0xbe6e4c0f + const datalen = 12 + + exchangeA := []tcp.Exchange{ + 0: { // A sends SYN to B. + Outgoing: &tcp.Segment{SEQ: issA, Flags: tcp.FlagSYN, WND: windowA}, + WantState: tcp.StateSynSent, + WantPeerState: tcp.StateSynRcvd, + }, + 1: { // A receives SYNACK from B. + Incoming: &tcp.Segment{SEQ: issB, ACK: issA + 1, Flags: SYNACK, WND: windowB}, + WantState: tcp.StateEstablished, + WantPending: &tcp.Segment{SEQ: issA + 1, ACK: issB + 1, Flags: tcp.FlagACK, WND: windowA}, + WantPeerState: tcp.StateSynRcvd, + }, + 2: { // A sends ACK to B thus establishing connection. + Outgoing: &tcp.Segment{SEQ: issA + 1, ACK: issB + 1, Flags: tcp.FlagACK, WND: windowA}, + WantState: tcp.StateEstablished, + WantPeerState: tcp.StateEstablished, + }, + 3: { // A sends PSH|ACK to B with 12 byte message: "hello world\n" + Outgoing: &tcp.Segment{SEQ: issA + 1, ACK: issB + 1, Flags: PSHACK, WND: windowA, DATALEN: datalen}, + WantState: tcp.StateEstablished, + WantPeerState: tcp.StateEstablished, + }, + 4: { // A receives ACK from B of last message. + Incoming: &tcp.Segment{SEQ: issB + 1, ACK: issA + 1 + datalen, Flags: tcp.FlagACK, WND: windowB}, + WantState: tcp.StateEstablished, + WantPeerState: tcp.StateEstablished, + }, + 5: { // A receives PSH|ACK from B with echoed 12 byte message: "hello world\n" + Incoming: &tcp.Segment{SEQ: issB + 1, ACK: issA + 1 + datalen, Flags: PSHACK, WND: windowB, DATALEN: datalen}, + WantState: tcp.StateEstablished, + WantPending: &tcp.Segment{SEQ: issA + 1 + datalen, ACK: issB + 1 + datalen, Flags: tcp.FlagACK, WND: windowA}, + WantPeerState: tcp.StateEstablished, + }, + 6: { // A ACKs B's message. + Outgoing: &tcp.Segment{SEQ: issA + 1 + datalen, ACK: issB + 1 + datalen, Flags: tcp.FlagACK, WND: windowA}, + WantState: tcp.StateEstablished, + WantPeerState: tcp.StateEstablished, + }, + 7: { // A sends PSH|ACK to B with SECOND 12 byte message. + Outgoing: &tcp.Segment{SEQ: issA + 1 + datalen, ACK: issB + 1 + datalen, Flags: PSHACK, WND: windowA, DATALEN: datalen}, + WantState: tcp.StateEstablished, + WantPeerState: tcp.StateEstablished, + }, + 8: { // A receives PSH|ACK that acks last message and contains echoed of SECOND 12 byte message. + Incoming: &tcp.Segment{SEQ: issB + 1 + datalen, ACK: issA + 1 + 2*datalen, Flags: PSHACK, WND: windowB, DATALEN: datalen}, + WantState: tcp.StateEstablished, + WantPending: &tcp.Segment{SEQ: issA + 1 + 2*datalen, ACK: issB + 1 + 2*datalen, Flags: tcp.FlagACK, WND: windowA}, + WantPeerState: tcp.StateEstablished, + }, + 9: { // A ACKs B's SECOND message. + Outgoing: &tcp.Segment{SEQ: issA + 1 + 2*datalen, ACK: issB + 1 + 2*datalen, Flags: tcp.FlagACK, WND: windowA}, + WantState: tcp.StateEstablished, + WantPeerState: tcp.StateEstablished, + }, + 10: { // A sends FIN|ACK to B to close connection. + Outgoing: &tcp.Segment{SEQ: issA + 1 + 2*datalen, ACK: issB + 1 + 2*datalen, Flags: FINACK, WND: windowA}, + WantState: tcp.StateFinWait1, + WantPeerState: tcp.StateCloseWait, + }, + 11: { // A receives B's ACK of FIN. + Incoming: &tcp.Segment{SEQ: issB + 1 + 2*datalen, ACK: issA + 2 + 2*datalen, Flags: tcp.FlagACK, WND: windowB}, + WantState: tcp.StateFinWait2, + WantPending: &tcp.Segment{SEQ: issA + 2 + 2*datalen, ACK: issB + 1 + 2*datalen, Flags: tcp.FlagACK, WND: windowA}, + WantPeerState: tcp.StateCloseWait, + }, + } + // The client starts in the SYN_SENT state with a random sequence number. + gotServerSeg, _ := parseSegment(t, exchangeHelloWorld[0]) + tcbA.HelperInitState(tcp.StateSynSent, gotServerSeg.SEQ, gotServerSeg.SEQ, windowB) + tcbA.HelperExchange(t, exchangeA) +} + +func reverseExchange(exchange []tcp.Exchange) []tcp.Exchange { + if len(exchange) == 0 { + panic("len(exchange) != len(states) or empty exchange: " + strconv.Itoa(len(exchange))) + } + firstIsIn := exchange[0].Incoming != nil + if firstIsIn { + panic("please start with an outgoing segment to reverse exchange for best test results") + } + out := make([]tcp.Exchange, len(exchange)) + for i := range exchange { + isLast := i == len(exchange)-1 + isOut := exchange[i].Outgoing != nil + out[i].WantState, out[i].WantPeerState = exchange[i].WantPeerState, exchange[i].WantState + if isOut { + out[i].Incoming = exchange[i].Outgoing + if !isLast { + out[i].WantPending = exchange[i+1].Incoming + } + } else { + out[i].Outgoing = exchange[i].Incoming + } + } + return out +} + +func parseSegment(t *testing.T, b []byte) (tcp.Segment, []byte) { + var vld lneto.Validator + t.Helper() + efrm, err := ethernet.NewFrame(b) + if err != nil { + t.Fatal(err) + } + if efrm.EtherTypeOrSize() != ethernet.TypeIPv4 { + t.Fatalf("not IPv4") + } + efrm.ValidateSize(&vld) + if err := vld.ErrPop(); err != nil { + t.Fatal(vld.ErrPop()) + } + ifrm, err := ipv4.NewFrame(efrm.Payload()) + if err != nil { + t.Fatal(err) + } + if ifrm.Protocol() != 6 { + t.Fatalf("not TCP") + } + v, _ := ifrm.VersionAndIHL() + if v != 4 { + t.Fatal("invalid IP version", v) + } + ifrm.ValidateSize(&vld) + if err := vld.ErrPop(); err != nil { + t.Fatal(vld.ErrPop()) + } + + ipl := ifrm.Payload() + tfrm, err := tcp.NewFrame(ipl) + if err != nil { + t.Fatal(err) + } + tfrm.ValidateSize(&vld) + if err := vld.ErrPop(); err != nil { + t.Fatal(err) + } + _ = tfrm.String() + payload := tfrm.Payload() + return tfrm.Segment(len(payload)), payload +} + +func TestUnexpectedStateClosing(t *testing.T) { + // TCB is a server which returns an HTTP response and receives a FINACK. + var tcb tcp.ControlBlock + const httpLen = 1192 + const issA, issB, windowA, windowB = 1, 127, 2000, 2000 + tcb.HelperInitState(tcp.StateEstablished, issA, issA, windowA) + tcb.HelperInitRcv(issB, issB, windowB) + + ex := []tcp.Exchange{ + 0: { // Server sends HTTP response. + Outgoing: &tcp.Segment{SEQ: issA, ACK: issB, Flags: PSHACK, WND: windowA, DATALEN: httpLen}, + WantState: tcp.StateEstablished, + }, + 1: { // Client sends an ACK to server. + Incoming: &tcp.Segment{SEQ: issB, ACK: issA + httpLen, Flags: tcp.FlagACK, WND: windowB}, + WantState: tcp.StateEstablished, + }, + 2: { // Client sends FIN|ACK to server. + Incoming: &tcp.Segment{SEQ: issB, ACK: issA + httpLen, Flags: FINACK, WND: windowB}, + WantPending: &tcp.Segment{SEQ: issA + httpLen, ACK: issB + 1, Flags: tcp.FlagACK, WND: windowA}, + WantState: tcp.StateCloseWait, + }, + 3: { // Server sends out FINACK. + Outgoing: &tcp.Segment{SEQ: issA + httpLen, ACK: issB + 1, Flags: FINACK, WND: windowA}, + WantState: tcp.StateLastAck, + }, + 4: { // Client sends back ACK. + Incoming: &tcp.Segment{SEQ: issB + 1, ACK: issA + httpLen + 1, Flags: tcp.FlagACK, WND: windowB}, + WantState: tcp.StateClosed, + }, + } + 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/tcp/tcp_test.go b/tcp/tcp_test.go index 9d4a2d5..debbb4f 100644 --- a/tcp/tcp_test.go +++ b/tcp/tcp_test.go @@ -2,12 +2,8 @@ package tcp_test import ( "math/rand" - "strconv" "testing" - "github.com/soypat/lneto" - "github.com/soypat/lneto/ethernet" - "github.com/soypat/lneto/ipv4" "github.com/soypat/lneto/tcp" ) @@ -17,6 +13,109 @@ const ( PSHACK = tcp.FlagPSH | tcp.FlagACK ) +// TestExchangeTest_PassiveClose_FINACKRegression is a regression test for the bug where +// Close() in CLOSE-WAIT state set pending = [FlagFIN, FlagACK] (separate slots) instead +// of pending = [FlagFIN|FlagACK, 0]. This caused PendingSegment() to return only FIN +// without ACK since it only reads pending[0]. +func TestExchangeTest_PassiveClose_FINACKRegression(t *testing.T) { + const issA, issB, windowA, windowB = 100, 300, 1000, 1000 + test := tcp.ExchangeTest{ + ISSA: issA, + ISSB: issB, + WindowA: windowA, + WindowB: windowB, + InitStateA: tcp.StateEstablished, + InitStateB: tcp.StateEstablished, + Steps: []tcp.SegmentStep{ + 0: { // A sends FIN|ACK to B. B goes to CLOSE-WAIT. + Seg: tcp.Segment{SEQ: issA, ACK: issB, Flags: FINACK, WND: windowA}, + Action: tcp.StepASends, + AState: tcp.StateFinWait1, + BState: tcp.StateCloseWait, + BPending: &tcp.Segment{SEQ: issB, ACK: issA + 1, Flags: tcp.FlagACK, WND: windowB}, + }, + 1: { // B sends ACK to A. Auto-queues FIN|ACK in CLOSE-WAIT. + Seg: tcp.Segment{SEQ: issB, ACK: issA + 1, Flags: tcp.FlagACK, WND: windowB}, + Action: tcp.StepBSends, + AState: tcp.StateFinWait2, + BState: tcp.StateCloseWait, + BPending: &tcp.Segment{SEQ: issB, ACK: issA + 1, Flags: FINACK, WND: windowB}, + }, + 2: { // B calls Close(). Goes to LAST-ACK. Pending must be FIN|ACK combined. + // This is the regression check: Close() must NOT overwrite the auto-queued + // FIN|ACK with [FlagFIN, FlagACK] in separate pending slots. + Action: tcp.StepBCloses, + AState: tcp.StateFinWait2, // A unchanged. + BState: tcp.StateLastAck, + BPending: &tcp.Segment{SEQ: issB, ACK: issA + 1, Flags: FINACK, WND: windowB}, + }, + 3: { // B sends FIN|ACK to A. + Seg: tcp.Segment{SEQ: issB, ACK: issA + 1, Flags: FINACK, WND: windowB}, + Action: tcp.StepBSends, + AState: tcp.StateTimeWait, + BState: tcp.StateLastAck, + BPending: nil, + }, + 4: { // A sends final ACK to B. B goes to CLOSED. + Seg: tcp.Segment{SEQ: issA + 1, ACK: issB + 1, Flags: tcp.FlagACK, WND: windowA}, + Action: tcp.StepASends, + AState: tcp.StateTimeWait, + BState: tcp.StateClosed, + }, + }, + } + test.RunB(t) // Only run B's perspective since that's where Close() is called. +} + +// TestExchangeTest_figure12 demonstrates ExchangeTest which defines both peers' states +// symmetrically and runs tests from both perspectives with a single definition. +func TestExchangeTest_figure12(t *testing.T) { + const issA, issB, windowA, windowB = 100, 300, 1000, 1000 + test := tcp.ExchangeTest{ + ISSA: issA, + ISSB: issB, + WindowA: windowA, + WindowB: windowB, + InitStateA: tcp.StateEstablished, + InitStateB: tcp.StateEstablished, + Steps: []tcp.SegmentStep{ + 0: { // A sends FIN|ACK to B. + Seg: tcp.Segment{SEQ: issA, ACK: issB, Flags: FINACK, WND: windowA}, + Action: tcp.StepASends, + AState: tcp.StateFinWait1, + BState: tcp.StateCloseWait, + APending: nil, + BPending: &tcp.Segment{SEQ: issB, ACK: issA + 1, Flags: tcp.FlagACK, WND: windowB}, + }, + 1: { // B sends ACK to A. (Auto-queues FIN|ACK in CLOSE-WAIT) + Seg: tcp.Segment{SEQ: issB, ACK: issA + 1, Flags: tcp.FlagACK, WND: windowB}, + Action: tcp.StepBSends, + AState: tcp.StateFinWait2, + BState: tcp.StateCloseWait, + APending: &tcp.Segment{SEQ: issA + 1, ACK: issB, Flags: tcp.FlagACK, WND: windowA}, // TODO: should be nil? + BPending: &tcp.Segment{SEQ: issB, ACK: issA + 1, Flags: FINACK, WND: windowB}, + }, + 2: { // B sends FIN|ACK to A. + Seg: tcp.Segment{SEQ: issB, ACK: issA + 1, Flags: FINACK, WND: windowB}, + Action: tcp.StepBSends, + AState: tcp.StateTimeWait, + BState: tcp.StateLastAck, + APending: &tcp.Segment{SEQ: issA + 1, ACK: issB + 1, Flags: tcp.FlagACK, WND: windowA}, + BPending: nil, + }, + 3: { // A sends final ACK to B. + Seg: tcp.Segment{SEQ: issA + 1, ACK: issB + 1, Flags: tcp.FlagACK, WND: windowA}, + Action: tcp.StepASends, + AState: tcp.StateTimeWait, + BState: tcp.StateClosed, + APending: nil, + BPending: nil, + }, + }, + } + test.Run(t) // Runs both PeerA and PeerB subtests. +} + /* Section 3.5 of RFC 9293: Basic 3-way handshake for connection synchronization. TCP Peer A TCP Peer B @@ -31,42 +130,39 @@ const ( 5. ESTABLISHED --> --> ESTABLISHED */ -func TestExchange_rfc9293_figure6(t *testing.T) { +func TestExchangeTest_rfc9293_figure6(t *testing.T) { const issA, issB, windowA, windowB = 100, 300, 1000, 1000 - exchangeA := []tcp.Exchange{ - { // A sends SYN to B. - Outgoing: &tcp.Segment{SEQ: issA, Flags: tcp.FlagSYN, WND: windowA}, - WantState: tcp.StateSynSent, - WantPeerState: tcp.StateSynRcvd, - }, - { // A receives SYNACK from B thus establishing the connection on A's side. - Incoming: &tcp.Segment{SEQ: issB, ACK: issA + 1, Flags: SYNACK, WND: windowB}, - WantState: tcp.StateEstablished, - WantPending: &tcp.Segment{SEQ: issA + 1, ACK: issB + 1, Flags: tcp.FlagACK, WND: windowA}, - WantPeerState: tcp.StateSynRcvd, - }, - { // A sends ACK to B, which leaves connection established on their side. Three way handshake complete by now. - Outgoing: &tcp.Segment{SEQ: issA + 1, ACK: issB + 1, Flags: tcp.FlagACK, WND: windowA}, - WantState: tcp.StateEstablished, - WantPeerState: tcp.StateEstablished, + test := tcp.ExchangeTest{ + ISSA: issA, + ISSB: issB, + WindowA: windowA, + WindowB: windowB, + InitStateA: tcp.StateSynSent, + InitStateB: tcp.StateListen, + Steps: []tcp.SegmentStep{ + 0: { // A sends SYN to B. + Seg: tcp.Segment{SEQ: issA, Flags: tcp.FlagSYN, WND: windowA}, + Action: tcp.StepASends, + AState: tcp.StateSynSent, + BState: tcp.StateSynRcvd, + BPending: &tcp.Segment{SEQ: issB, ACK: issA + 1, Flags: SYNACK, WND: windowB}, + }, + 1: { // B sends SYNACK to A. + Seg: tcp.Segment{SEQ: issB, ACK: issA + 1, Flags: SYNACK, WND: windowB}, + Action: tcp.StepBSends, + AState: tcp.StateEstablished, + BState: tcp.StateSynRcvd, + APending: &tcp.Segment{SEQ: issA + 1, ACK: issB + 1, Flags: tcp.FlagACK, WND: windowA}, + }, + 2: { // A sends ACK to B. Three-way handshake complete. + Seg: tcp.Segment{SEQ: issA + 1, ACK: issB + 1, Flags: tcp.FlagACK, WND: windowA}, + Action: tcp.StepASends, + AState: tcp.StateEstablished, + BState: tcp.StateEstablished, + }, }, } - var tcbA tcp.ControlBlock - tcbA.HelperInitState(tcp.StateSynSent, issA, issA, windowA) - tcbA.HelperExchange(t, exchangeA) - segA, ok := tcbA.PendingSegment(0) - if ok { - t.Error("unexpected Client pending segment after establishment: ", segA) - } - exchangeB := reverseExchange(exchangeA) - - var tcbB tcp.ControlBlock - tcbB.HelperInitState(tcp.StateListen, issB, issB, windowB) - tcbB.HelperExchange(t, exchangeB) // TODO remove [:3] after snd.UNA bugfix - segB, ok := tcbB.PendingSegment(0) - if ok { - t.Error("unexpected Listener pending segment after establishment: ", segB) - } + test.Run(t) } /* @@ -87,30 +183,43 @@ func TestExchange_rfc9293_figure6(t *testing.T) { 7. ... --> ESTABLISHED */ -func TestExchange_rfc9293_figure7(t *testing.T) { +func TestExchangeTest_rfc9293_figure7(t *testing.T) { const issA, issB, windowA, windowB = 100, 300, 1000, 1000 - exchangeA := []tcp.Exchange{ - 0: { // A sends SYN to B. - Outgoing: &tcp.Segment{SEQ: issA, Flags: tcp.FlagSYN, WND: windowA}, - WantState: tcp.StateSynSent, - }, - 1: { // A receives a SYN with no ACK from B. - Incoming: &tcp.Segment{SEQ: issB, Flags: tcp.FlagSYN, WND: windowB}, - WantState: tcp.StateSynRcvd, - WantPending: &tcp.Segment{SEQ: issA, ACK: issB + 1, Flags: SYNACK, WND: windowA}, - }, - 2: { // A sends SYNACK to B. - Outgoing: &tcp.Segment{SEQ: issA, ACK: issB + 1, Flags: SYNACK, WND: windowA}, - WantState: tcp.StateSynRcvd, - }, - 3: { // A receives ACK from B. - Incoming: &tcp.Segment{SEQ: issB, ACK: issA + 1, Flags: SYNACK, WND: windowA}, - WantState: tcp.StateEstablished, + // NOTE: Simultaneous SYN can only be tested from one perspective because + // the "simultaneous" nature means segments cross in flight. In a sequential + // test, when B receives A's SYN first, B goes to SYN-RECEIVED, not SYN-SENT. + test := tcp.ExchangeTest{ + ISSA: issA, + ISSB: issB, + WindowA: windowA, + WindowB: windowB, + InitStateA: tcp.StateSynSent, + InitStateB: tcp.StateSynSent, + Steps: []tcp.SegmentStep{ + 0: { // A sends SYN to B (crosses with B's SYN). + Seg: tcp.Segment{SEQ: issA, Flags: tcp.FlagSYN, WND: windowA}, + Action: tcp.StepASends, + AState: tcp.StateSynSent, + }, + 1: { // A receives SYN from B (no ACK - B hasn't received A's SYN yet). + Seg: tcp.Segment{SEQ: issB, Flags: tcp.FlagSYN, WND: windowB}, + Action: tcp.StepBSends, + AState: tcp.StateSynRcvd, + APending: &tcp.Segment{SEQ: issA, ACK: issB + 1, Flags: SYNACK, WND: windowA}, + }, + 2: { // A sends SYNACK to B. + Seg: tcp.Segment{SEQ: issA, ACK: issB + 1, Flags: SYNACK, WND: windowA}, + Action: tcp.StepASends, + AState: tcp.StateSynRcvd, + }, + 3: { // A receives SYNACK from B. + Seg: tcp.Segment{SEQ: issB, ACK: issA + 1, Flags: SYNACK, WND: windowB}, + Action: tcp.StepBSends, + AState: tcp.StateEstablished, + }, }, } - var tcbA tcp.ControlBlock - tcbA.HelperInitState(tcp.StateSynSent, issA, issA, windowA) - tcbA.HelperExchange(t, exchangeA) + test.RunA(t) // Only A's perspective - B's is fundamentally different in sequential test. } /* @@ -132,146 +241,104 @@ func TestExchange_rfc9293_figure7(t *testing.T) { 7. ESTABLISHED <-- <-- SYN-RECEIVED 8. ESTABLISHED --> --> ESTABLISHED + +NOTE: This test is asymmetric. A and B have different views because B receives an +old duplicate SYN that A never sent. Cannot use Run() - must test each perspective separately. */ -func TestExchange_rfc9293_figure8(t *testing.T) { +func TestExchangeTest_rfc9293_figure8(t *testing.T) { const issA, issB, windowA, windowB = 100, 300, 1000, 1000 const issAold = 90 - const issBNew = issB + 100 - exchangeA := []tcp.Exchange{ - 0: { // A sends new SYN to B (which is not received). - Outgoing: &tcp.Segment{SEQ: issA, Flags: tcp.FlagSYN, WND: windowA}, - WantState: tcp.StateSynSent, - WantPeerState: tcp.StateSynRcvd, - }, - 1: { // Receive SYN from B acking an old "duplicate" SYN. - Incoming: &tcp.Segment{SEQ: issB, ACK: issAold + 1, Flags: SYNACK, WND: windowB}, - WantState: tcp.StateSynSent, - WantPending: &tcp.Segment{SEQ: issAold + 1, Flags: tcp.FlagRST, WND: windowA}, - WantPeerState: tcp.StateSynRcvd, - }, - 2: { // A sends RST to B and makes segment believable by using the old SEQ. - Outgoing: &tcp.Segment{SEQ: issAold + 1, Flags: tcp.FlagRST, WND: windowA}, - WantState: tcp.StateSynSent, - WantPeerState: tcp.StateListen, - }, - 3: { // A sends a duplicate SYN to B. - Outgoing: &tcp.Segment{SEQ: issA, Flags: tcp.FlagSYN, WND: windowA}, - WantState: tcp.StateSynSent, - WantPeerState: tcp.StateSynRcvd, - }, - 4: { // B SYNACKs new SYN. - Incoming: &tcp.Segment{SEQ: issBNew, ACK: issA + 1, Flags: SYNACK, WND: windowB}, - WantState: tcp.StateEstablished, - WantPending: &tcp.Segment{SEQ: issA + 1, ACK: issBNew + 1, Flags: tcp.FlagACK, WND: windowA}, - WantPeerState: tcp.StateSynRcvd, - }, - 5: { // B receives ACK from A. - Outgoing: &tcp.Segment{SEQ: issA + 1, ACK: issBNew + 1, Flags: tcp.FlagACK, WND: windowA}, - WantState: tcp.StateEstablished, - WantPeerState: tcp.StateEstablished, - }, - } - var tcbA tcp.ControlBlock - tcbA.HelperInitState(tcp.StateSynSent, issA, issA, windowA) - tcbA.HelperExchange(t, exchangeA) + const issBnew = 400 - exchangeB := []tcp.Exchange{ - 0: { // B receives old SYN from A. - Incoming: &tcp.Segment{SEQ: issAold, Flags: tcp.FlagSYN, WND: windowA}, - WantState: tcp.StateSynRcvd, - WantPending: &tcp.Segment{SEQ: issB, ACK: issAold + 1, Flags: SYNACK, WND: windowB}, - }, - 1: { // B SYNACKs old SYN. - Outgoing: &tcp.Segment{SEQ: issB, ACK: issAold + 1, Flags: SYNACK, WND: windowB}, - WantState: tcp.StateSynRcvd, - }, - 2: { // B receives RST from A. - Incoming: &tcp.Segment{SEQ: issAold + 1, Flags: tcp.FlagRST, WND: windowA}, - WantState: tcp.StateListen, - }, - 3: { // B receives new SYN from A. - Incoming: &tcp.Segment{SEQ: issA, Flags: tcp.FlagSYN, WND: windowA}, - WantState: tcp.StateSynRcvd, - WantPending: &tcp.Segment{SEQ: issBNew, ACK: issA + 1, Flags: SYNACK, WND: windowB}, - }, - 4: { // B SYNACKs new SYN. - Outgoing: &tcp.Segment{SEQ: issBNew, ACK: issA + 1, Flags: SYNACK, WND: windowB}, - WantState: tcp.StateSynRcvd, - }, - 5: { // B receives ACK from A. - Incoming: &tcp.Segment{SEQ: issA + 1, ACK: issBNew + 1, Flags: tcp.FlagACK, WND: windowA}, - WantState: tcp.StateEstablished, - }, - } - var tcbB tcp.ControlBlock - tcbB.HelperInitState(tcp.StateListen, issB, issB, windowB) - tcbB.HelperExchange(t, exchangeB) + // Test from A's perspective: A sends SYN, gets wrong SYNACK, sends RST, + // gets correct SYNACK, sends ACK. + t.Run("PeerA", func(t *testing.T) { + var tcbA tcp.ControlBlock + tcbA.HelperInitState(tcp.StateSynSent, issA, issA, windowA) + + stepsA := []tcp.SegmentStep{ + 0: { // A sends SYN (step 2 in figure). + Seg: tcp.Segment{SEQ: issA, Flags: tcp.FlagSYN, WND: windowA}, + Action: tcp.StepASends, + AState: tcp.StateSynSent, + }, + 1: { // A receives SYNACK with wrong ACK (acking old duplicate SYN). + Seg: tcp.Segment{SEQ: issB, ACK: issAold + 1, Flags: SYNACK, WND: windowB}, + Action: tcp.StepBSends, + AState: tcp.StateSynSent, // A stays in SYN-SENT because ACK is wrong. + APending: &tcp.Segment{SEQ: issAold + 1, Flags: tcp.FlagRST, WND: windowA}, + }, + 2: { // A sends RST to reject the bad SYNACK. + Seg: tcp.Segment{SEQ: issAold + 1, Flags: tcp.FlagRST, WND: windowA}, + Action: tcp.StepASends, + AState: tcp.StateSynSent, + }, + 3: { // A sends duplicate SYN (retransmit, step 6 arrival at B). + Seg: tcp.Segment{SEQ: issA, Flags: tcp.FlagSYN, WND: windowA}, + Action: tcp.StepASends, + AState: tcp.StateSynSent, + }, + 4: { // A receives correct SYNACK from B with new ISS. + Seg: tcp.Segment{SEQ: issBnew, ACK: issA + 1, Flags: SYNACK, WND: windowB}, + Action: tcp.StepBSends, + AState: tcp.StateEstablished, + APending: &tcp.Segment{SEQ: issA + 1, ACK: issBnew + 1, Flags: tcp.FlagACK, WND: windowA}, + }, + 5: { // A sends ACK to complete handshake. + Seg: tcp.Segment{SEQ: issA + 1, ACK: issBnew + 1, Flags: tcp.FlagACK, WND: windowA}, + Action: tcp.StepASends, + AState: tcp.StateEstablished, + }, + } + tcbA.HelperSteps(t, stepsA, true) + }) + + // Test from B's perspective: B receives old SYN, sends SYNACK, receives RST, + // receives real SYN, sends new SYNACK, receives ACK. + t.Run("PeerB", func(t *testing.T) { + var tcbB tcp.ControlBlock + tcbB.HelperInitState(tcp.StateListen, issB, issB, windowB) + + stepsB := []tcp.SegmentStep{ + 0: { // B receives old duplicate SYN (step 3). + Seg: tcp.Segment{SEQ: issAold, Flags: tcp.FlagSYN, WND: windowA}, + Action: tcp.StepASends, + BState: tcp.StateSynRcvd, + BPending: &tcp.Segment{SEQ: issB, ACK: issAold + 1, Flags: SYNACK, WND: windowB}, + }, + 1: { // B sends SYNACK for old SYN. + Seg: tcp.Segment{SEQ: issB, ACK: issAold + 1, Flags: SYNACK, WND: windowB}, + Action: tcp.StepBSends, + BState: tcp.StateSynRcvd, + }, + 2: { // B receives RST, goes back to LISTEN. + Seg: tcp.Segment{SEQ: issAold + 1, Flags: tcp.FlagRST, WND: windowA}, + Action: tcp.StepASends, + BState: tcp.StateListen, + }, + 3: { // B receives real SYN (step 6). + Seg: tcp.Segment{SEQ: issA, Flags: tcp.FlagSYN, WND: windowA}, + Action: tcp.StepASends, + BState: tcp.StateSynRcvd, + BPending: &tcp.Segment{SEQ: issBnew, ACK: issA + 1, Flags: SYNACK, WND: windowB}, + }, + 4: { // B sends new SYNACK. + Seg: tcp.Segment{SEQ: issBnew, ACK: issA + 1, Flags: SYNACK, WND: windowB}, + Action: tcp.StepBSends, + BState: tcp.StateSynRcvd, + }, + 5: { // B receives ACK, connection established. + Seg: tcp.Segment{SEQ: issA + 1, ACK: issBnew + 1, Flags: tcp.FlagACK, WND: windowA}, + Action: tcp.StepASends, + BState: tcp.StateEstablished, + }, + } + tcbB.HelperSteps(t, stepsB, false) + }) } /* - Figure 12: Normal Close Sequence - TCP Peer A TCP Peer B - 1. ESTABLISHED ESTABLISHED - - 2. (Close) - FIN-WAIT-1 --> --> CLOSE-WAIT - - 3. FIN-WAIT-2 <-- <-- CLOSE-WAIT - - 4. (Close) - TIME-WAIT <-- <-- LAST-ACK - - 5. TIME-WAIT --> --> CLOSED - - 6. (2 MSL) - CLOSED -*/ -func TestExchange_rfc9293_figure12(t *testing.T) { - const issA, issB, windowA, windowB = 100, 300, 1000, 1000 - exchangeA := []tcp.Exchange{ - 0: { // A sends FIN|ACK to B to begin closing connection. - Outgoing: &tcp.Segment{SEQ: issA, ACK: issB, Flags: FINACK, WND: windowA}, - WantState: tcp.StateFinWait1, - WantPeerState: tcp.StateCloseWait, - }, - 1: { // A receives ACK from B. - Incoming: &tcp.Segment{SEQ: issB, ACK: issA + 1, Flags: tcp.FlagACK, WND: windowB}, - WantState: tcp.StateFinWait2, - WantPeerState: tcp.StateCloseWait, - // TODO(soypat): WantPending should be nil here? Perhaps fix test by modifying rcvFinWait1 pending result. - WantPending: &tcp.Segment{SEQ: issA + 1, ACK: issB, Flags: tcp.FlagACK, WND: windowA}, - }, - 2: { // A receives FIN|ACK from B. - Incoming: &tcp.Segment{SEQ: issB, ACK: issA + 1, Flags: FINACK, WND: windowB}, - WantState: tcp.StateTimeWait, - WantPending: &tcp.Segment{SEQ: issA + 1, ACK: issB + 1, Flags: tcp.FlagACK, WND: windowA}, - WantPeerState: tcp.StateLastAck, - }, - 3: { // A sends ACK to B. - Outgoing: &tcp.Segment{SEQ: issA + 1, ACK: issB + 1, Flags: tcp.FlagACK, WND: windowA}, - WantState: tcp.StateTimeWait, // Technically we should be in TimeWait here. - WantPeerState: tcp.StateClosed, - }, - } - var tcbA tcp.ControlBlock - 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) -} - -/* - Figure 12: Simultaneous Close Sequence + Figure 13: Simultaneous Close Sequence TCP Peer A TCP Peer B 1. ESTABLISHED ESTABLISHED @@ -289,182 +356,38 @@ func TestExchange_rfc9293_figure12(t *testing.T) { (2 MSL) (2 MSL) CLOSED CLOSED */ -func TestExchange_rfc9293_figure13(t *testing.T) { +func TestExchangeTest_rfc9293_figure13(t *testing.T) { const issA, issB, windowA, windowB = 100, 300, 1000, 1000 - exchangeA := []tcp.Exchange{ - 0: { // A sends FIN|ACK to B to begin closing connection. - Outgoing: &tcp.Segment{SEQ: issA, ACK: issB, Flags: FINACK, WND: windowA}, - WantState: tcp.StateFinWait1, - }, - 1: { // A receives FIN|ACK from B, who sent packet before receiving A's FINACK. - Incoming: &tcp.Segment{SEQ: issB, ACK: issA, Flags: FINACK, WND: windowB}, - WantState: tcp.StateClosing, - WantPending: &tcp.Segment{SEQ: issA + 1, ACK: issB + 1, Flags: tcp.FlagACK, WND: windowA}, - }, - 2: { // A sends ACK to B. - Outgoing: &tcp.Segment{SEQ: issA + 1, ACK: issB + 1, Flags: tcp.FlagACK, WND: windowA}, - WantState: tcp.StateTimeWait, + // NOTE: Simultaneous close can only be tested from one perspective because + // the "simultaneous" nature means FINs cross in flight. In a sequential test, + // when B receives A's FIN first, B goes to CLOSE-WAIT, not CLOSING. + test := tcp.ExchangeTest{ + ISSA: issA, + ISSB: issB, + WindowA: windowA, + WindowB: windowB, + InitStateA: tcp.StateEstablished, + InitStateB: tcp.StateEstablished, + Steps: []tcp.SegmentStep{ + 0: { // A sends FIN|ACK to B (crosses with B's FIN|ACK). + Seg: tcp.Segment{SEQ: issA, ACK: issB, Flags: FINACK, WND: windowA}, + Action: tcp.StepASends, + AState: tcp.StateFinWait1, + }, + 1: { // A receives FIN|ACK from B (B sent before receiving A's FIN). + Seg: tcp.Segment{SEQ: issB, ACK: issA, Flags: FINACK, WND: windowB}, + Action: tcp.StepBSends, + AState: tcp.StateClosing, + APending: &tcp.Segment{SEQ: issA + 1, ACK: issB + 1, Flags: tcp.FlagACK, WND: windowA}, + }, + 2: { // A sends ACK to B. + Seg: tcp.Segment{SEQ: issA + 1, ACK: issB + 1, Flags: tcp.FlagACK, WND: windowA}, + Action: tcp.StepASends, + AState: tcp.StateTimeWait, + }, }, } - var tcbA tcp.ControlBlock - tcbA.HelperInitState(tcp.StateEstablished, issA, issA, windowA) - tcbA.HelperInitRcv(issB, issB, windowB) - tcbA.HelperExchange(t, exchangeA) - - // No need to test B since exchange is completely symmetric. -} - -// Check no duplicate ack is sent during establishment. -func TestExchange_noDupAckDuringEstablished(t *testing.T) { - var tcbA tcp.ControlBlock - const issA, issB, windowA, windowB = 300, 334222749, 256, 64240 - synseg := tcp.ClientSynSegment(issA, windowA) - - // err := tcbA.Open(issA, issA, tcp.StateSynSent) - tcbA.SetRecvWindow(windowA) - // if err != nil { - // t.Fatal(err) - // } - establishA := []tcp.Exchange{ - 0: { // A sends SYN to B. - Outgoing: &synseg, - WantState: tcp.StateSynSent, - }, - 1: { // B sends SYN to A. - Incoming: &tcp.Segment{SEQ: issB, ACK: 0, WND: windowB, Flags: tcp.FlagSYN}, - WantPending: &tcp.Segment{SEQ: issA, ACK: issB + 1, WND: windowA, Flags: SYNACK}, - WantState: tcp.StateSynRcvd, - }, - 2: { // Send SYNACK to B. - Outgoing: &tcp.Segment{SEQ: issA, ACK: issB + 1, WND: windowA, Flags: SYNACK}, - WantState: tcp.StateSynRcvd, - }, - 3: { // B ACKs SYNACK, thus establishing the connection on both sides. - Incoming: &tcp.Segment{SEQ: issB + 1, ACK: issA + 1, WND: windowB, Flags: tcp.FlagACK}, - WantState: tcp.StateEstablished, - }, - } - tcbA.HelperExchange(t, establishA) - if tcbA.State() != tcp.StateEstablished { - t.Fatal("expected established state") - } - checkNoPending(t, &tcbA) - const datasize = 5 - dataExA := []tcp.Exchange{ - 0: { // B sends PSH|ACK to A with data. - Incoming: &tcp.Segment{SEQ: issB + 1, ACK: issA + 1, WND: windowB, Flags: PSHACK, DATALEN: datasize}, - WantPending: &tcp.Segment{SEQ: issA + 1, ACK: issB + 1 + datasize, WND: windowA, Flags: tcp.FlagACK}, - WantState: tcp.StateEstablished, - }, - 1: { // A ACKs B's data. - Outgoing: &tcp.Segment{SEQ: issA + 1, ACK: issB + 1 + datasize, WND: windowA, Flags: tcp.FlagACK}, - WantState: tcp.StateEstablished, - }, - 2: { // A sends PSH|ACK to B with data, same amount, as if echoing. - Outgoing: &tcp.Segment{SEQ: issA + 1, ACK: issB + 1 + datasize, WND: windowA, Flags: PSHACK, DATALEN: datasize}, - WantState: tcp.StateEstablished, - }, - // 3: { // B ACKs A's data. - // Incoming: &tcp.Segment{SEQ: issB + 1 + datasize, ACK: issA + 1 + datasize, WND: windowB, Flags: tcp.FlagACK}, - // WantPending: nil, - // WantState: tcp.StateEstablished, - // }, - } - tcbA.HelperExchange(t, dataExA) - checkNoPending(t, &tcbA) - tcbA.Recv(tcp.Segment{SEQ: issB + 1 + datasize, ACK: issA + 1 + datasize, WND: windowB, Flags: tcp.FlagACK}) - checkNoPending(t, &tcbA) -} - -// This test reenacts a full client-server interaction in the sending and receiving -// of the 12 byte message "hello world\n" over TCP. -func TestExchange_helloworld(t *testing.T) { - // Client Transmission Control Block. - var tcbA tcp.ControlBlock - const windowA, windowB = 502, 4096 - const issA, issB = 0x5e722b7d, 0xbe6e4c0f - const datalen = 12 - - exchangeA := []tcp.Exchange{ - 0: { // A sends SYN to B. - Outgoing: &tcp.Segment{SEQ: issA, Flags: tcp.FlagSYN, WND: windowA}, - WantState: tcp.StateSynSent, - WantPeerState: tcp.StateSynRcvd, - }, - 1: { // A receives SYNACK from B. - Incoming: &tcp.Segment{SEQ: issB, ACK: issA + 1, Flags: SYNACK, WND: windowB}, - WantState: tcp.StateEstablished, - WantPending: &tcp.Segment{SEQ: issA + 1, ACK: issB + 1, Flags: tcp.FlagACK, WND: windowA}, - WantPeerState: tcp.StateSynRcvd, - }, - 2: { // A sends ACK to B thus establishing connection. - Outgoing: &tcp.Segment{SEQ: issA + 1, ACK: issB + 1, Flags: tcp.FlagACK, WND: windowA}, - WantState: tcp.StateEstablished, - WantPeerState: tcp.StateEstablished, - }, - 3: { // A sends PSH|ACK to B with 12 byte message: "hello world\n" - Outgoing: &tcp.Segment{SEQ: issA + 1, ACK: issB + 1, Flags: PSHACK, WND: windowA, DATALEN: datalen}, - WantState: tcp.StateEstablished, - WantPeerState: tcp.StateEstablished, - }, - 4: { // A receives ACK from B of last message. - Incoming: &tcp.Segment{SEQ: issB + 1, ACK: issA + 1 + datalen, Flags: tcp.FlagACK, WND: windowB}, - WantState: tcp.StateEstablished, - WantPeerState: tcp.StateEstablished, - }, - 5: { // A receives PSH|ACK from B with echoed 12 byte message: "hello world\n" - Incoming: &tcp.Segment{SEQ: issB + 1, ACK: issA + 1 + datalen, Flags: PSHACK, WND: windowB, DATALEN: datalen}, - WantState: tcp.StateEstablished, - WantPending: &tcp.Segment{SEQ: issA + 1 + datalen, ACK: issB + 1 + datalen, Flags: tcp.FlagACK, WND: windowA}, - WantPeerState: tcp.StateEstablished, - }, - 6: { // A ACKs B's message. - Outgoing: &tcp.Segment{SEQ: issA + 1 + datalen, ACK: issB + 1 + datalen, Flags: tcp.FlagACK, WND: windowA}, - WantState: tcp.StateEstablished, - WantPeerState: tcp.StateEstablished, - }, - 7: { // A sends PSH|ACK to B with SECOND 12 byte message. - Outgoing: &tcp.Segment{SEQ: issA + 1 + datalen, ACK: issB + 1 + datalen, Flags: PSHACK, WND: windowA, DATALEN: datalen}, - WantState: tcp.StateEstablished, - WantPeerState: tcp.StateEstablished, - }, - 8: { // A receives PSH|ACK that acks last message and contains echoed of SECOND 12 byte message. - Incoming: &tcp.Segment{SEQ: issB + 1 + datalen, ACK: issA + 1 + 2*datalen, Flags: PSHACK, WND: windowB, DATALEN: datalen}, - WantState: tcp.StateEstablished, - WantPending: &tcp.Segment{SEQ: issA + 1 + 2*datalen, ACK: issB + 1 + 2*datalen, Flags: tcp.FlagACK, WND: windowA}, - WantPeerState: tcp.StateEstablished, - }, - 9: { // A ACKs B's SECOND message. - Outgoing: &tcp.Segment{SEQ: issA + 1 + 2*datalen, ACK: issB + 1 + 2*datalen, Flags: tcp.FlagACK, WND: windowA}, - WantState: tcp.StateEstablished, - WantPeerState: tcp.StateEstablished, - }, - 10: { // A sends FIN|ACK to B to close connection. - Outgoing: &tcp.Segment{SEQ: issA + 1 + 2*datalen, ACK: issB + 1 + 2*datalen, Flags: FINACK, WND: windowA}, - WantState: tcp.StateFinWait1, - WantPeerState: tcp.StateCloseWait, - }, - 11: { // A receives B's ACK of FIN. - Incoming: &tcp.Segment{SEQ: issB + 1 + 2*datalen, ACK: issA + 2 + 2*datalen, Flags: tcp.FlagACK, WND: windowB}, - WantState: tcp.StateFinWait2, - WantPending: &tcp.Segment{SEQ: issA + 2 + 2*datalen, ACK: issB + 1 + 2*datalen, Flags: tcp.FlagACK, WND: windowA}, - WantPeerState: tcp.StateCloseWait, - }, - } - // The client starts in the SYN_SENT state with a random sequence number. - gotServerSeg, _ := parseSegment(t, exchangeHelloWorld[0]) - tcbA.HelperInitState(tcp.StateSynSent, gotServerSeg.SEQ, gotServerSeg.SEQ, windowB) - tcbA.HelperExchange(t, exchangeA) - - // TODO(soypat): fix exchange reversal. - return - exchangeB := reverseExchange(exchangeA) - - exchangeB[7].WantPending = nil // Is an unpredicable action. - var tcbB tcp.ControlBlock - tcbB.HelperInitState(tcp.StateListen, issB, issB, windowB) - tcbB.HelperInitRcv(issA, issA, windowA) - tcbB.HelperExchange(t, exchangeB) + test.RunA(t) // Only A's perspective - B's is fundamentally different in sequential test. } func TestResetEstablished(t *testing.T) { @@ -524,136 +447,6 @@ func TestFinackClose(t *testing.T) { } } -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()) - } - } -} - -func parseSegment(t *testing.T, b []byte) (tcp.Segment, []byte) { - var vld lneto.Validator - t.Helper() - efrm, err := ethernet.NewFrame(b) - if err != nil { - t.Fatal(err) - } - if efrm.EtherTypeOrSize() != ethernet.TypeIPv4 { - t.Fatalf("not IPv4") - } - efrm.ValidateSize(&vld) - if err := vld.ErrPop(); err != nil { - t.Fatal(vld.ErrPop()) - } - ifrm, err := ipv4.NewFrame(efrm.Payload()) - if err != nil { - t.Fatal(err) - } - if ifrm.Protocol() != 6 { - t.Fatalf("not TCP") - } - v, _ := ifrm.VersionAndIHL() - if v != 4 { - t.Fatal("invalid IP version", v) - } - ifrm.ValidateSize(&vld) - if err := vld.ErrPop(); err != nil { - t.Fatal(vld.ErrPop()) - } - - ipl := ifrm.Payload() - tfrm, err := tcp.NewFrame(ipl) - if err != nil { - t.Fatal(err) - } - tfrm.ValidateSize(&vld) - if err := vld.ErrPop(); err != nil { - t.Fatal(err) - } - _ = tfrm.String() - payload := tfrm.Payload() - return tfrm.Segment(len(payload)), payload -} - -func reverseExchange(exchange []tcp.Exchange) []tcp.Exchange { - if len(exchange) == 0 { - panic("len(exchange) != len(states) or empty exchange: " + strconv.Itoa(len(exchange))) - } - firstIsIn := exchange[0].Incoming != nil - if firstIsIn { - panic("please start with an outgoing segment to reverse exchange for best test results") - } - out := make([]tcp.Exchange, len(exchange)) - for i := range exchange { - isLast := i == len(exchange)-1 - isOut := exchange[i].Outgoing != nil - out[i].WantState, out[i].WantPeerState = exchange[i].WantPeerState, exchange[i].WantState - if isOut { - out[i].Incoming = exchange[i].Outgoing - if !isLast { - out[i].WantPending = exchange[i+1].Incoming - } - } else { - out[i].Outgoing = exchange[i].Incoming - } - } - return out -} - func checkNoPending(t *testing.T, tcb *tcp.ControlBlock) bool { t.Helper() // We extensively test the API for inadvertent state modification in a HasPending or PendingSegment call. @@ -701,40 +494,6 @@ var exchangeHelloWorld = [][]byte{ 11: []byte("\xd8\x5e\xd3\x43\x03\xeb\x28\xcd\xc1\x05\x4d\xbb\x08\x00\x45\x00\x00\x28\x00\x00\x40\x00\x40\x06\xb6\x5b\xc0\xa8\x01\x91\xc0\xa8\x01\x93\x04\xd2\x84\x96\xbe\x6e\x4c\x28\x5e\x72\x2b\x97\x50\x10\x10\x00\xfd\x56\x00\x00\x00\x00\x00\x00\x00\x00"), } -func TestUnexpectedStateClosing(t *testing.T) { - // TCB is a server which returns an HTTP response and receives a FINACK. - var tcb tcp.ControlBlock - const httpLen = 1192 - const issA, issB, windowA, windowB = 1, 127, 2000, 2000 - tcb.HelperInitState(tcp.StateEstablished, issA, issA, windowA) - tcb.HelperInitRcv(issB, issB, windowB) - - ex := []tcp.Exchange{ - 0: { // Server sends HTTP response. - Outgoing: &tcp.Segment{SEQ: issA, ACK: issB, Flags: PSHACK, WND: windowA, DATALEN: httpLen}, - WantState: tcp.StateEstablished, - }, - 1: { // Client sends an ACK to server. - Incoming: &tcp.Segment{SEQ: issB, ACK: issA + httpLen, Flags: tcp.FlagACK, WND: windowB}, - WantState: tcp.StateEstablished, - }, - 2: { // Client sends FIN|ACK to server. - Incoming: &tcp.Segment{SEQ: issB, ACK: issA + httpLen, Flags: FINACK, WND: windowB}, - WantPending: &tcp.Segment{SEQ: issA + httpLen, ACK: issB + 1, Flags: tcp.FlagACK, WND: windowA}, - WantState: tcp.StateCloseWait, - }, - 3: { // Server sends out FINACK. - Outgoing: &tcp.Segment{SEQ: issA + httpLen, ACK: issB + 1, Flags: FINACK, WND: windowA}, - WantState: tcp.StateLastAck, - }, - 4: { // Client sends back ACK. - Incoming: &tcp.Segment{SEQ: issB + 1, ACK: issA + httpLen + 1, Flags: tcp.FlagACK, WND: windowB}, - WantState: tcp.StateClosed, - }, - } - tcb.HelperExchange(t, ex[:]) -} - // This corresponds to https://github.com/soypat/seqs/issues/19 // The bug consisted of a panic condition encountered when using wget client with a seqs based server. // Thanks to @knieriem for finding this and the detailed report they submitted. diff --git a/tcp/txqueue_test.go b/tcp/txqueue_test.go index 3ee6043..78613fe 100644 --- a/tcp/txqueue_test.go +++ b/tcp/txqueue_test.go @@ -27,8 +27,8 @@ func TestRingTx_op(t *testing.T) { dataSent := make([]byte, 0, maxBuf*10) var rtx ringTx rng := rand.New(rand.NewSource(0)) - for iseed := int64(0); iseed < 10000; iseed++ { - rng.Seed(iseed) + for iseed := int64(0); iseed < 1000; iseed++ { + rng.Seed(iseed + rng.Int63()) for itest := 0; itest < 32; itest++ { bufsize := rng.Intn(maxBuf/2) + maxBuf/2 iss := Value(0) @@ -209,9 +209,7 @@ func TestTxQueue_multipacket(t *testing.T) { rng := rand.New(rand.NewSource(3)) var wbuf, rbuf [mtu]byte for itest := 0; itest < 32; itest++ { - t.Log(itest) rng.Seed(int64(itest)) - println(itest) err := rtx.Reset(internalbuff, maxPkts, iss) if err != nil { t.Fatal(err) @@ -258,7 +256,6 @@ func TestTxQueue_multipacket(t *testing.T) { } acked := 0 for acked < roff { - t.Log(acked) maxToack := min(roff-acked, maxWriteSize) toack := rng.Intn(maxToack) + 1 // t.Log("\n", rtx.string()) diff --git a/x/xnet/stack-async.go b/x/xnet/stack-async.go index 84430c9..c91d012 100644 --- a/x/xnet/stack-async.go +++ b/x/xnet/stack-async.go @@ -219,6 +219,12 @@ func (s *StackAsync) Addr() netip.Addr { return s.ip.Addr() } +func (s *StackAsync) SetSubnet(subnetMask netip.Prefix) { + s.mu.Lock() + defer s.mu.Unlock() + s.subnet = subnetMask +} + func (s *StackAsync) SetHardwareAddress(hw [6]byte) error { s.mu.Lock() defer s.mu.Unlock() @@ -286,6 +292,16 @@ func (s *StackAsync) ListenTCP(conn *tcp.Conn, localPort uint16) (err error) { return nil } +func (s *StackAsync) RegisterListener(listener *tcp.Listener) (err error) { + s.mu.Lock() + defer s.mu.Unlock() + lport := listener.LocalPort() + if lport == 0 { + return lneto.ErrZeroSource + } + return s.tcps.Register(listener, nil) +} + var errNoDNSServer = errors.New("no DNS server- did DHCP complete? You can set a predetermined DNS server in Stack configuration") func (s *StackAsync) StartLookupIP(host string) error { diff --git a/x/xnet/tcppool.go b/x/xnet/tcppool.go new file mode 100644 index 0000000..c3a9081 --- /dev/null +++ b/x/xnet/tcppool.go @@ -0,0 +1,152 @@ +package xnet + +import ( + "errors" + "log/slog" + "sync" + "time" + + "github.com/soypat/lneto/tcp" +) + +// TCPPool implements tcp.pool. +type TCPPool struct { + mu sync.Mutex + naqcuired int + conns []tcp.Conn + acquiredAt []time.Time + closingAt []time.Time + abortedAt []time.Time + nextISS tcp.Value + _now func() time.Time + estbTimeout time.Duration + closingTimeout time.Duration +} + +func _() { + var l tcp.Listener + l.Reset(0, &TCPPool{}) // compile time guarantee of interface implementation. +} + +type TCPPoolConfig struct { + PoolSize int + QueueSize int + BufferSize int + ConnLogger *slog.Logger + Now func() time.Time + // EstablishedTimeout sets the timeout for a TCP connection since it is acquired until it is established. + // If the connection does not establish in this time it will be closed by the pool. + EstablishedTimeout time.Duration + // ClosingTimeout sets the timeout for a TCP connection to close and be returned to Pool. + // If the connection is not closed in this time it will be aborted by the pool. + ClosingTimeout time.Duration +} + +func NewTCPPool(cfg TCPPoolConfig) (*TCPPool, error) { + if cfg.EstablishedTimeout <= 0 || cfg.ClosingTimeout <= 0 { + return nil, errors.New("invalid timeout") + } + n := cfg.PoolSize + bufsize := cfg.BufferSize + pool := &TCPPool{ + acquiredAt: make([]time.Time, n), + closingAt: make([]time.Time, n), + abortedAt: make([]time.Time, n), + conns: make([]tcp.Conn, n), + _now: cfg.Now, + estbTimeout: cfg.EstablishedTimeout, + closingTimeout: cfg.ClosingTimeout, + } + bufSpace := make([]byte, 2*n*bufsize) + for i := range pool.conns { + bufoff := 2 * i * bufsize + err := pool.conns[i].Configure(tcp.ConnConfig{ + RxBuf: bufSpace[bufoff : bufoff+bufsize], + TxBuf: bufSpace[bufoff+bufsize : bufoff+2*bufsize], + TxPacketQueueSize: cfg.QueueSize, + Logger: cfg.ConnLogger, + }) + if err != nil { + return nil, err + } + } + return pool, nil +} + +func (p *TCPPool) GetTCP() (*tcp.Conn, tcp.Value) { + p.mu.Lock() + defer p.mu.Unlock() + for i := range p.conns { + if p.acquiredAt[i].IsZero() { + p.acquiredAt[i] = p.now() + p.nextISS += 1000 + p.naqcuired++ + return &p.conns[i], p.nextISS + } + } + return nil, 0 +} + +func (p *TCPPool) PutTCP(conn *tcp.Conn) { + for i := range p.conns { + if &p.conns[i] == conn { + p.mu.Lock() + p.conns[i].Abort() + p.acquiredAt[i] = time.Time{} + p.abortedAt[i] = time.Time{} + p.closingAt[i] = time.Time{} + p.naqcuired-- + p.mu.Unlock() + return + } + } + panic("conn does not belong to this pool") +} + +func (p *TCPPool) CheckTimeouts() { + for i := range p.conns { + st := p.conns[i].State() + if st == tcp.StateEstablished { + continue + } + p.mu.Lock() + acq := p.acquiredAt[i] + p.mu.Unlock() + if acq.IsZero() { + continue + } else if st.IsPreestablished() && p.since(acq) > p.estbTimeout { + // Was acquired and did not reach establishment state so we close. + // This is part of a syn-flood defense mechanism. + p.conns[i].Close() + } else if st.IsClosed() || st.IsClosing() { + p.mu.Lock() + if p.closingAt[i].IsZero() { + p.closingAt[i] = p.now() + } else if p.abortedAt[i].IsZero() && p.since(p.closingAt[i]) > p.closingTimeout { + p.abortedAt[i] = p.now() + p.conns[i].Abort() + } else if p.since(p.abortedAt[i]) > 10*time.Second { + println("connection aborted and still not returned to TCPPool") + } + p.mu.Unlock() + } + } +} + +func (p *TCPPool) since(t time.Time) time.Duration { + if p._now == nil { + return time.Since(t) + } + return p._now().Sub(t) +} + +func (p *TCPPool) now() time.Time { + if p._now == nil { + return time.Now() + } + return p._now() +} + +func (p *TCPPool) NumberOfAcquired() int { + return p.naqcuired +} diff --git a/x/xnet/xnet_listener_test.go b/x/xnet/xnet_listener_test.go new file mode 100644 index 0000000..d8c9753 --- /dev/null +++ b/x/xnet/xnet_listener_test.go @@ -0,0 +1,225 @@ +package xnet + +import ( + "net/netip" + "testing" + "time" + + "github.com/soypat/lneto/tcp" +) + +func TestStackAsyncListener_SingleConnection(t *testing.T) { + const seed int64 = 1234 + const MTU = 1500 + const svPort = 80 + const clPort = 1337 + + // Create two stacks. + client, sv := new(StackAsync), new(StackAsync) + err := client.Reset(StackConfig{ + Hostname: "Client", + RandSeed: seed, + StaticAddress: netip.AddrFrom4([4]byte{10, 0, 0, 1}), + MaxTCPConns: 1, + HardwareAddress: [6]byte{0xbe, 0xef, 0, 0, 0, 1}, + MTU: MTU, + }) + if err != nil { + t.Fatal(err) + } + err = sv.Reset(StackConfig{ + Hostname: "Server", + RandSeed: ^seed, + StaticAddress: netip.AddrFrom4([4]byte{10, 0, 0, 2}), + MaxTCPConns: 1, // Note: We use listener, not direct TCP conn registration. + HardwareAddress: [6]byte{0xbe, 0xef, 0, 0, 0, 2}, + MTU: MTU, + }) + if err != nil { + t.Fatal(err) + } + client.SetGateway6(sv.HardwareAddress()) + sv.SetGateway6(client.HardwareAddress()) + + // Create client connection. + var clConn tcp.Conn + err = clConn.Configure(tcp.ConnConfig{ + RxBuf: make([]byte, MTU), + TxBuf: make([]byte, MTU), + TxPacketQueueSize: 4, + }) + if err != nil { + t.Fatal(err) + } + + // Create pool and listener for server. + pool, err := NewTCPPool(TCPPoolConfig{ + PoolSize: 1, + QueueSize: 4, + BufferSize: MTU, + EstablishedTimeout: 10e9, + ClosingTimeout: 10e9, + }) + if err != nil { + t.Fatal(err) + } + + var listener tcp.Listener + err = listener.Reset(svPort, pool) + if err != nil { + t.Fatal(err) + } + err = sv.RegisterListener(&listener) + if err != nil { + t.Fatal(err) + } + + // Client dials server. + err = client.DialTCP(&clConn, clPort, netip.AddrPortFrom(sv.Addr(), svPort)) + if err != nil { + t.Fatal(err) + } + + tst := testerFrom(t, MTU) + + // Complete TCP handshake. + tst.TestTCPHandshake(client, sv) + + // After handshake, TryAccept should work. + if listener.NumberOfReadyToAccept() != 1 { + t.Fatalf("after handshake: expected 1 ready, got %d", listener.NumberOfReadyToAccept()) + } + svConn, err := listener.TryAccept() + if err != nil { + t.Fatalf("TryAccept: %v", err) + } + if listener.NumberOfReadyToAccept() != 0 { + t.Fatalf("after accept: expected 0 ready, got %d", listener.NumberOfReadyToAccept()) + } + + // Verify both connections are established. + if clConn.State() != tcp.StateEstablished { + t.Fatalf("client: expected StateEstablished, got %s", clConn.State()) + } + if svConn.State() != tcp.StateEstablished { + t.Fatalf("server: expected StateEstablished, got %s", svConn.State()) + } + + // Test data exchange: client -> server. + sendData := []byte("hello from client") + tst.TestTCPEstablishedSingleData(client, sv, &clConn, svConn, sendData) + + // Test data exchange: server -> client. + replyData := []byte("hello from server") + tst.TestTCPEstablishedSingleData(sv, client, svConn, &clConn, replyData) + + // Test close (client-initiated). + tst.TestTCPClose(client, sv, &clConn, svConn) +} + +func TestStackAsyncListener_MultiSequentialConn(t *testing.T) { + const seed int64 = 1234 + const MTU = 1500 + const svPort = 80 + const clPort = 1337 + const poolsize = 10 + const bufsize = 128 + // Create two stacks. + sv := new(StackAsync) + err := sv.Reset(StackConfig{ + Hostname: "Server", + RandSeed: ^seed, + StaticAddress: netip.AddrFrom4([4]byte{10, 0, 0, 2}), + MaxTCPConns: 1, // Note: We use listener, not direct TCP conn registration. + HardwareAddress: [6]byte{0xbe, 0xef, 0, 0, 0, 2}, + MTU: MTU, + }) + if err != nil { + t.Fatal(err) + } + + // Create pool and listener for server. + pool, err := NewTCPPool(TCPPoolConfig{ + PoolSize: poolsize, + QueueSize: 4, + BufferSize: bufsize, + EstablishedTimeout: 10e9, + ClosingTimeout: 10e9, + }) + if err != nil { + t.Fatal(err) + } + + var listener tcp.Listener + err = listener.Reset(svPort, pool) + if err != nil { + t.Fatal(err) + } + err = sv.RegisterListener(&listener) + if err != nil { + t.Fatal(err) + } + caddr := netip.AddrFrom4([4]byte{10, 0, 0, 1}) + chw := [6]byte{0xbe, 0xef, 0, 0, 0, 1} + sv.SetGateway6(chw) + tst := testerFrom(t, MTU) + doRequest := func(caddrp netip.AddrPort, sleep time.Duration, data []byte) { + var client StackAsync + err := client.Reset(StackConfig{ + Hostname: "Client", + RandSeed: seed, + StaticAddress: caddrp.Addr(), + MaxTCPConns: 1, + HardwareAddress: chw, + MTU: MTU, + }) + if err != nil { + panic(err) + } + client.SetGateway6(sv.HardwareAddress()) + // Create client connection. + var clConn tcp.Conn + err = clConn.Configure(tcp.ConnConfig{ + RxBuf: make([]byte, bufsize), + TxBuf: make([]byte, bufsize), + TxPacketQueueSize: 4, + }) + if err != nil { + t.Fatal(err) + } + // Client dials server. + err = client.DialTCP(&clConn, caddrp.Port(), netip.AddrPortFrom(sv.Addr(), svPort)) + if err != nil { + t.Fatal(err) + } + // Complete TCP handshake. + tst.TestTCPHandshake(&client, sv) + // After handshake, TryAccept should work. + if listener.NumberOfReadyToAccept() != 1 { + t.Fatalf("after handshake: expected 1 ready, got %d", listener.NumberOfReadyToAccept()) + } + svconn, err := listener.TryAccept() + if err != nil { + t.Fatal(err) + } else if svconn.RemotePort() != clConn.LocalPort() || + [4]byte(svconn.RemoteAddr()) != client.Addr().As4() { + t.Fatal("race condition to listener acquisition") + } + // Verify both connections are established. + if clConn.State() != tcp.StateEstablished { + t.Fatalf("client: expected StateEstablished, got %s", clConn.State()) + } + if len(data) > 0 { + tst.TestTCPEstablishedSingleData(&client, sv, &clConn, svconn, data) + } + if sleep > 0 { + time.Sleep(sleep) + } + tst.TestTCPClose(&client, sv, &clConn, svconn) + } + + for range 1000 { + caddr := caddr.Next() + doRequest(netip.AddrPortFrom(caddr, uint16(sv.Prand32())), 0, []byte("HTTP 1.0\r\n")) + } +} diff --git a/x/xnet/xnet_test.go b/x/xnet/xnet_test.go index a3d1102..aaa99ea 100644 --- a/x/xnet/xnet_test.go +++ b/x/xnet/xnet_test.go @@ -5,6 +5,7 @@ import ( "errors" "math/rand" "net/netip" + "sync" "testing" "github.com/soypat/lneto/arp" @@ -14,6 +15,8 @@ import ( ) const ( + logExchange = false + synack = tcp.FlagSYN | tcp.FlagACK pshack = tcp.FlagPSH | tcp.FlagACK finack = tcp.FlagFIN | tcp.FlagACK @@ -143,12 +146,12 @@ func testerFrom(t *testing.T, mtu int) *tester { } type tester struct { - t *testing.T - cap pcap.PacketBreakdown - frmbuf []pcap.Frame - buf []byte - exch []tcpExpectExchange - lastSeg tcp.Segment + t *testing.T + + cap pcap.PacketBreakdown + frmbuf []pcap.Frame + bufmu sync.Mutex + buf []byte } type tcpExpectExchange struct { @@ -177,7 +180,7 @@ func (tst *tester) TestTCPSetupAndEstablish(svStack, clStack *StackAsync, svConn func (tst *tester) TestTCPHandshake(stack1, stack2 *StackAsync) { tst.t.Helper() - tst.exch = append(tst.exch[:0], []tcpExpectExchange{ + exch := [...]tcpExpectExchange{ { SourceIdx: 0, WantFlags: tcp.FlagSYN, @@ -194,8 +197,8 @@ func (tst *tester) TestTCPHandshake(stack1, stack2 *StackAsync) { }, noExchange(0), noExchange(1), - }...) - for _, wants := range tst.exch { + } + for _, wants := range exch { tst.TCPExchange(wants, stack1, stack2) } } @@ -217,7 +220,7 @@ func (tst *tester) TestTCPEstablishedSingleData(srcStack, dstStack *StackAsync, t.Fatal(err) } nprev := dstConn.BufferedInput() - tst.exch = append(tst.exch[:0], []tcpExpectExchange{ + exch := [...]tcpExpectExchange{ { SourceIdx: 0, WantFlags: pshack, @@ -230,10 +233,12 @@ func (tst *tester) TestTCPEstablishedSingleData(srcStack, dstStack *StackAsync, }, noExchange(0), noExchange(1), - }...) - for _, wants := range tst.exch { + } + for _, wants := range exch { tst.TCPExchange(wants, srcStack, dstStack) } + tst.bufmu.Lock() + defer tst.bufmu.Unlock() n, err := dstConn.Read(tst.buf) if err != nil { t.Errorf("reading back data %q on conn2: %s", sendData, err) @@ -249,6 +254,7 @@ func (tst *tester) TestTCPEstablishedSingleData(srcStack, dstStack *StackAsync, t.Errorf("expected to read back %q from conn, got %q", sendData, got) } } + setzero(tst.buf[:n]) } @@ -263,7 +269,7 @@ func (tst *tester) TestTCPClose(stack1, stack2 *StackAsync, conn1, conn2 *tcp.Co if err != nil { t.Fatal(err) } - tst.exch = append(tst.exch[:0], []tcpExpectExchange{ + exch := [...]tcpExpectExchange{ { SourceIdx: 0, WantFlags: finack, // Closer sends FINACK @@ -284,19 +290,22 @@ func (tst *tester) TestTCPClose(stack1, stack2 *StackAsync, conn1, conn2 *tcp.Co }, noExchange(0), noExchange(1), - }...) - t.Log(conn1.State().String(), conn2.State().String()) - for i, exch := range tst.exch { + } + if logExchange { + t.Log(conn1.State().String(), conn2.State().String()) + } + for i, exch := range exch { failed := t.Failed() - tst.TCPExchange(exch, stack1, stack2) + seg := tst.TCPExchange(exch, stack1, stack2) if !failed && t.Failed() { t.Error(i, exch.SourceIdx, "close failure") } if exch.WantFlags == 0 { continue } - - t.Log(i, tcp.StringExchange(tst.lastSeg, conn1.State(), conn2.State(), exch.SourceIdx != 0)) + if logExchange { + t.Log(i, tcp.StringExchange(seg, conn1.State(), conn2.State(), exch.SourceIdx != 0)) + } } state1 := conn1.State() @@ -315,8 +324,9 @@ func (tst *tester) TestTCPClose(stack1, stack2 *StackAsync, conn1, conn2 *tcp.Co } } -func (tst *tester) TCPExchange(expect tcpExpectExchange, stack1, stack2 *StackAsync) { - tst.lastSeg = tcp.Segment{} +func (tst *tester) TCPExchange(expect tcpExpectExchange, stack1, stack2 *StackAsync) tcp.Segment { + tst.bufmu.Lock() + defer tst.bufmu.Unlock() var src, dst *StackAsync defer func(failed bool) { if !failed && tst.t.Failed() { @@ -328,7 +338,6 @@ func (tst *tester) TCPExchange(expect tcpExpectExchange, stack1, stack2 *StackAs t.Helper() buf := tst.buf[:cap(tst.buf)] nodata := expect.WantFlags == 0 - switch expect.SourceIdx { case 0: src, dst = stack1, stack2 @@ -337,17 +346,18 @@ func (tst *tester) TCPExchange(expect tcpExpectExchange, stack1, stack2 *StackAs default: panic("OOB") } + n, err := src.Encapsulate(buf[:], -1, 0) if err != nil { t.Fatal(err) } else if n == 0 { if nodata { - return // No data sent and no data expected. + return tcp.Segment{} // No data sent and no data expected. } t.Error("zero bits sent") } else if nodata && n > 0 { t.Error("expected no data sent and got data") - return + return tcp.Segment{} } defer setzero(buf[:n]) @@ -379,7 +389,6 @@ func (tst *tester) TCPExchange(expect tcpExpectExchange, stack1, stack2 *StackAs payload := tfrm.Payload() seg := tfrm.Segment(len(payload)) - tst.lastSeg = seg if !bytes.Equal(payload, expect.WantData) { t.Errorf("mismatched data sent, \nwant=%q\ngot=%q\n", expect.WantData, payload) } @@ -390,11 +399,14 @@ func (tst *tester) TCPExchange(expect tcpExpectExchange, stack1, stack2 *StackAs if err != nil { t.Fatal(err) } + return seg } func (tst *tester) ARPExchangeOnly(querying, target *StackAsync) { t := tst.t t.Helper() + tst.bufmu.Lock() + defer tst.bufmu.Unlock() buf := tst.buf[:cap(tst.buf)] // === PHASE 1: ARP Request from querying stack ===