Repository: pion/sctp
Pull request #130 (2020-07-06 02:13:22)

Fix handling of retransmitted cookie-echo
Relates to pion/webrtc#1270

#### Description
Please see this comment(https://github.com/pion/webrtc/issues/1270#issuecomment-653953743) for details.

#### What's in this PR
* Fixed the issue by sending COOKIE-ACK on the receipt of COOKIE-ECHO even in the Established state.
* Added a reproducible test using vnet, successfully confirmed the problem, and the fix.

codecov[bot]: # [Codecov](https://codecov.io/gh/pion/sctp/pull/130?src=pr&el=h1) Report
> Merging [#130](https://codecov.io/gh/pion/sctp/pull/130?src=pr&el=desc) into [master](https://codecov.io/gh/pion/sctp/commit/82cfade7059d1ae21f97b923df7e6738b171bc1b&el=desc) will **increase** coverage by `0.15%`.
> The diff coverage is `63.63%`.

[![Impacted file tree graph](https://codecov.io/gh/pion/sctp/pull/130/graphs/tree.svg?width=650&height=150&src=pr&token=AB8S3U6xR9)](https://codecov.io/gh/pion/sctp/pull/130?src=pr&el=tree)

```diff
@@            Coverage Diff             @@
##           master     #130      +/-   ##
==========================================
+ Coverage   78.94%   79.09%   +0.15%     
==========================================
  Files          43       43              
  Lines        2798     2799       +1     
==========================================
+ Hits         2209     2214       +5     
+ Misses        454      450       -4     
  Partials      135      135              
```

| Flag | Coverage Δ | |
|---|---|---|
| #go | `79.09% <63.63%> (+0.15%)` | :arrow_up: |
| #wasm | `70.16% <63.63%> (+0.26%)` | :arrow_up: |

| [Impacted Files](https://codecov.io/gh/pion/sctp/pull/130?src=pr&el=tree) | Coverage Δ | |
|---|---|---|
| [association.go](https://codecov.io/gh/pion/sctp/pull/130/diff?src=pr&el=tree#diff-YXNzb2NpYXRpb24uZ28=) | `83.10% <63.63%> (+0.01%)` | :arrow_up: |
| [chunkheader.go](https://codecov.io/gh/pion/sctp/pull/130/diff?src=pr&el=tree#diff-Y2h1bmtoZWFkZXIuZ28=) | `82.14% <0.00%> (+7.14%)` | :arrow_up: |
| [chunk\_cookie\_ack.go](https://codecov.io/gh/pion/sctp/pull/130/diff?src=pr&el=tree#diff-Y2h1bmtfY29va2llX2Fjay5nbw==) | `69.23% <0.00%> (+15.38%)` | :arrow_up: |

------

[Continue to review full report at Codecov](https://codecov.io/gh/pion/sctp/pull/130?src=pr&el=continue).
> **Legend** - [Click here to learn more](https://docs.codecov.io/docs/codecov-delta)
> `Δ = absolute <relative> (impact)`, `ø = not affected`, `? = missing data`
> Powered by [Codecov](https://codecov.io/gh/pion/sctp/pull/130?src=pr&el=footer). Last update [82cfade...0357fc7](https://codecov.io/gh/pion/sctp/pull/130?src=pr&el=lastupdated). Read the [comment docs](https://docs.codecov.io/docs/pull-request-comments).

enobufs: @Sean-Der This is a fix for a fundamental error in handling delayed cookie-echo received in the "established" state (which would happen on lossy connection).  If you could review this at your earliest convenience, that would really be appreciated.
at-wat: ```
$ bash .github/lint-commit-message.sh
Update association.go

Accepted the change.

Co-authored-by: Atsushi Watanabe <atsushi.w@ieee.org>

-------------------------------------------------
The preceding commit message is invalid
it failed 'Separate subject from body with a blank line' of the following checks

* Separate subject from body with a blank line
* Limit the subject line to 50 characters
* Capitalize the subject line
* Do not end the subject line with a period
* Wrap the body at 72 characters
```
Commit message from GitHub web UI uses CR+LF which is currently not supported by our commit message lint script. https://github.com/pion/.goassets/issues/6

`git commit --amend` and remove CR will fix the error.

Patch:
diff --git a/association.go b/association.go
index c8ca49e..88b21db 100644
--- a/association.go
+++ b/association.go
@@ -982,19 +982,27 @@ func (a *Association) handleHeartbeat(c *chunkHeartbeat) []*packet {
 func (a *Association) handleCookieEcho(c *chunkCookieEcho) []*packet {
 	state := a.getState()
 	a.log.Debugf("[%s] COOKIE-ECHO received in state '%s'", a.name, getAssociationStateString(state))
-	if state != closed && state != cookieWait && state != cookieEchoed {
+	switch state {
+	default:
 		return nil
-	}
+	case established:
+		if !bytes.Equal(a.myCookie.cookie, c.cookie) {
+			return nil
+		}
+	case closed, cookieWait, cookieEchoed:
+		if !bytes.Equal(a.myCookie.cookie, c.cookie) {
+			return nil
+		}
 
-	if !bytes.Equal(a.myCookie.cookie, c.cookie) {
-		return nil
-	}
+		a.t1Init.stop()
+		a.storedInit = nil
 
-	a.t1Init.stop()
-	a.storedInit = nil
+		a.t1Cookie.stop()
+		a.storedCookieEcho = nil
 
-	a.t1Cookie.stop()
-	a.storedCookieEcho = nil
+		a.setState(established)
+		a.handshakeCompletedCh <- nil
+	}
 
 	p := &packet{
 		verificationTag: a.peerVerificationTag,
@@ -1002,9 +1010,6 @@ func (a *Association) handleCookieEcho(c *chunkCookieEcho) []*packet {
 		destinationPort: a.destinationPort,
 		chunks:          []chunk{&chunkCookieAck{}},
 	}
-
-	a.setState(established)
-	a.handshakeCompletedCh <- nil
 	return pack(p)
 }
 


Test patch:
diff --git a/association_test.go b/association_test.go
index 2adbb62..84454dd 100644
--- a/association_test.go
+++ b/association_test.go
@@ -2022,84 +2022,6 @@ func TestAssocDelayedAck(t *testing.T) {
 
 		closeAssociationPair(br, a0, a1)
 	})
-
-	t.Run("Second DATA chunk to generate SACK immedidately", func(t *testing.T) {
-		lim := test.TimeOut(time.Second * 10)
-		defer lim.Stop()
-
-		const si uint16 = 6
-		var n int
-		var nPacketsReceived int
-		var ppi PayloadProtocolIdentifier
-		sbuf := make([]byte, 4000) // size should be less than initial cwnd (4380)
-		rbuf := make([]byte, 4000)
-
-		_, err := cryptoRand.Read(sbuf)
-		if !assert.Nil(t, err, "failed to create associations") {
-			return
-		}
-
-		br := test.NewBridge()
-
-		a0, a1, err := createNewAssociationPair(br, ackModeAlwaysDelay, 0)
-		if !assert.Nil(t, err, "failed to create associations") {
-			assert.FailNow(t, "failed due to earlier error")
-		}
-
-		s0, s1, err := establishSessionPair(br, a0, a1, si)
-		assert.Nil(t, err, "failed to establish session pair")
-
-		a0.stats.reset()
-		a1.stats.reset()
-
-		// Writes data (will fragmented)
-		n, err = s0.WriteSCTP(sbuf, PayloadTypeWebRTCBinary)
-		assert.Nil(t, err, "WriteSCTP failed")
-		assert.Equal(t, n, len(sbuf), "unexpected length of received data")
-
-		// Repeat calling br.Tick() until the buffered amount becomes 0
-		for s0.BufferedAmount() > 0 {
-			for {
-				n = br.Tick()
-				if n == 0 {
-					break
-				}
-			}
-
-			for {
-				s1.lock.RLock()
-				readable := s1.reassemblyQueue.isReadable()
-				s1.lock.RUnlock()
-				if !readable {
-					break
-				}
-				n, ppi, err = s1.ReadSCTP(rbuf)
-				if !assert.Nil(t, err, "ReadSCTP failed") {
-					return
-				}
-				assert.Equal(t, len(sbuf), n, "unexpected length of received data")
-				assert.Equal(t, ppi, PayloadTypeWebRTCBinary, "unexpected ppi")
-
-				nPacketsReceived++
-			}
-		}
-
-		br.Process()
-
-		assert.Equal(t, 1, nPacketsReceived, "should be one packet received")
-		assert.Equal(t, 0, s1.getNumBytesInReassemblyQueue(), "reassembly queue should be empty")
-
-		t.Logf("nDATAs      : %d\n", a1.stats.getNumDATAs())
-		t.Logf("nSACKs      : %d\n", a0.stats.getNumSACKs())
-		t.Logf("nAckTimeouts: %d\n", a1.stats.getNumAckTimeouts())
-
-		assert.Equal(t, uint64(4), a1.stats.getNumDATAs(), "DATA chunk count mismatch")
-		assert.True(t, a0.stats.getNumSACKs() < a1.stats.getNumDATAs(), "sack count should less than data")
-		assert.Equal(t, uint64(0), a1.stats.getNumAckTimeouts(), "ackTimeout count mismatch")
-		assert.Equal(t, uint64(0), a0.stats.getNumT3Timeouts(), "should be no retransmit")
-
-		closeAssociationPair(br, a0, a1)
-	})
 }
 
 func TestAssocReset(t *testing.T) {
diff --git a/vnet_test.go b/vnet_test.go
index e445a83..8819e89 100644
--- a/vnet_test.go
+++ b/vnet_test.go
@@ -21,11 +21,13 @@ type vNetEnvConfig struct {
 }
 
 type vNetEnv struct {
-	wan               *vnet.Router
-	net0              *vnet.Net
-	net1              *vnet.Net
-	numToDropData     int
-	numToDropReconfig int
+	wan                 *vnet.Router
+	net0                *vnet.Net
+	net1                *vnet.Net
+	numToDropData       int
+	numToDropReconfig   int
+	numToDropCookieEcho int
+	numToDropCookieAck  int
 }
 
 func (venv *vNetEnv) dropNextDataChunk(numToDrop int) {
@@ -36,6 +38,14 @@ func (venv *vNetEnv) dropNextReconfigChunk(numToDrop int) {
 	venv.numToDropReconfig = numToDrop
 }
 
+func (venv *vNetEnv) dropNextCookieEchoChunk(numToDrop int) {
+	venv.numToDropCookieEcho = numToDrop
+}
+
+func (venv *vNetEnv) dropNextCookieAckChunk(numToDrop int) {
+	venv.numToDropCookieAck = numToDrop
+}
+
 func buildVNetEnv(cfg *vNetEnvConfig) (*vNetEnv, error) {
 	log := cfg.log
 
@@ -88,6 +98,20 @@ func buildVNetEnv(cfg *vNetEnvConfig) (*vNetEnv, error) {
 						log.Infof("Chunk filter:  drop RECONFIG %s", chunk.String())
 						break loop
 					}
+				case *chunkCookieEcho:
+					if venv.numToDropCookieEcho > 0 {
+						toDrop = true
+						venv.numToDropCookieEcho--
+						log.Infof("Chunk filter:  drop %s", chunk.String())
+						break loop
+					}
+				case *chunkCookieAck:
+					if venv.numToDropCookieAck > 0 {
+						toDrop = true
+						venv.numToDropCookieAck--
+						log.Infof("Chunk filter:  drop %s", chunk.String())
+						break loop
+					}
 				}
 			}
 			return !toDrop
@@ -183,7 +207,7 @@ func testRwndFull(t *testing.T, unordered bool) {
 		}
 		defer assoc.Close() // nolint:errcheck
 
-		log.Info("server handlshake complete")
+		log.Info("server handshake complete")
 		close(serverHandshakeDone)
 
 		stream, err := assoc.AcceptStream()
@@ -257,7 +281,7 @@ func testRwndFull(t *testing.T, unordered bool) {
 		}
 		defer assoc.Close() // nolint:errcheck
 
-		log.Info("client handlshake complete")
+		log.Info("client handshake complete")
 		close(clientHandshakeDone)
 
 		stream, err := assoc.OpenStream(777, PayloadTypeWebRTCBinary)
@@ -360,7 +384,7 @@ func TestRwndFull(t *testing.T) {
 }
 
 func testStreamClose(t *testing.T, dropReconfig bool) {
-	lim := test.TimeOut(time.Second * 5)
+	lim := test.TimeOut(time.Second * 10)
 	defer lim.Stop()
 
 	loggerFactory := logging.NewDefaultLoggerFactory()
@@ -408,7 +432,7 @@ func testStreamClose(t *testing.T, dropReconfig bool) {
 		}
 		defer assoc.Close() // nolint:errcheck
 
-		log.Info("server handlshake complete")
+		log.Info("server handshake complete")
 
 		stream, err := assoc.AcceptStream()
 		if !assert.NoError(t, err, "should succeed") {
@@ -457,7 +481,7 @@ func testStreamClose(t *testing.T, dropReconfig bool) {
 		}
 		defer assoc.Close() // nolint:errcheck
 
-		log.Info("client handlshake complete")
+		log.Info("client handshake complete")
 
 		stream, err := assoc.OpenStream(777, PayloadTypeWebRTCBinary)
 		if !assert.NoError(t, err, "should succeed") {
@@ -544,3 +568,116 @@ func TestStreamClose(t *testing.T) {
 		testStreamClose(t, true)
 	})
 }
+
+// this test case reproduces the issue mentioned in
+// https://github.com/pion/webrtc/issues/1270#issuecomment-653953743
+// and confirmes the fix.
+// To reproduce the case mentioned above:
+// * Use simultaneous-open (SCTP)
+// * Drop both of the first COOKIE-ECHO and COOKIE-ACK
+func TestCookieEchoRetransmission(t *testing.T) {
+	lim := test.TimeOut(time.Second * 10)
+	defer lim.Stop()
+
+	loggerFactory := logging.NewDefaultLoggerFactory()
+	log := loggerFactory.NewLogger("test")
+
+	venv, err := buildVNetEnv(&vNetEnvConfig{
+		minDelay:      200 * time.Millisecond,
+		loggerFactory: loggerFactory,
+		log:           log,
+	})
+	if !assert.NoError(t, err, "should succeed") {
+		return
+	}
+	if !assert.NotNil(t, venv, "should not be nil") {
+		return
+	}
+	defer venv.wan.Stop() // nolint:errcheck
+
+	// To cause the cookie echo retransmission, both COOKIE-ECHO
+	// and COOKIE-ACK chunks need to be dropped at the same time.
+	venv.dropNextCookieEchoChunk(1)
+	venv.dropNextCookieAckChunk(1)
+
+	serverHandshakeDone := make(chan struct{})
+	clientHandshakeDone := make(chan struct{})
+	waitAllHandshakeDone := make(chan struct{})
+	clientShutDown := make(chan struct{})
+	serverShutDown := make(chan struct{})
+
+	maxReceiveBufferSize := uint32(64 * 1024)
+
+	// Go routine for Server
+	go func() {
+		defer close(serverShutDown)
+		// connected UDP conn for server
+		conn, err := venv.net0.DialUDP("udp4",
+			&net.UDPAddr{IP: net.ParseIP("1.1.1.1"), Port: 5000},
+			&net.UDPAddr{IP: net.ParseIP("2.2.2.2"), Port: 5000},
+		)
+		if !assert.NoError(t, err, "should succeed") {
+			return
+		}
+		defer conn.Close() // nolint:errcheck
+
+		// server association
+		// using Client for simultaneous open
+		assoc, err := Client(Config{
+			NetConn:              conn,
+			MaxReceiveBufferSize: maxReceiveBufferSize,
+			LoggerFactory:        loggerFactory,
+		})
+		if !assert.NoError(t, err, "should succeed") {
+			return
+		}
+		defer assoc.Close() // nolint:errcheck
+
+		log.Info("server handshake complete")
+		close(serverHandshakeDone)
+		<-waitAllHandshakeDone
+	}()
+
+	// Go routine for Client
+	go func() {
+		defer close(clientShutDown)
+		// connected UDP conn for client
+		conn, err := venv.net1.DialUDP("udp4",
+			&net.UDPAddr{IP: net.ParseIP("2.2.2.2"), Port: 5000},
+			&net.UDPAddr{IP: net.ParseIP("1.1.1.1"), Port: 5000},
+		)
+		if !assert.NoError(t, err, "should succeed") {
+			return
+		}
+
+		// client association
+		assoc, err := Client(Config{
+			NetConn:              conn,
+			MaxReceiveBufferSize: maxReceiveBufferSize,
+			LoggerFactory:        loggerFactory,
+		})
+		if !assert.NoError(t, err, "should succeed") {
+			return
+		}
+		defer assoc.Close() // nolint:errcheck
+
+		log.Info("client handshake complete")
+		close(clientHandshakeDone)
+		<-waitAllHandshakeDone
+	}()
+
+	//
+	// Scenario
+	//
+
+	// wait until both handshake complete
+	<-clientHandshakeDone
+	<-serverHandshakeDone
+	close(waitAllHandshakeDone)
+
+	log.Info("handshake complete")
+
+	<-clientShutDown
+	<-serverShutDown
+	log.Info("all done")
+}