Implement SCTP Negotiation Acceleration Protocol (SNAP) (#449)

This commit adds a createAssociationWithOutOfBandTokens method
that creates the SCTP association with a local and remote sctp-init
token as described in draft-hancke-tsvwg-snap-00.

Tokens can be generated using the newly added GenerateOutOfBandToken
method and set in the sctp.Config as LocalSctpInit and RemoteSctpInit.

This allows the peer connection to generate and negotiate the
sctp-init attribute in the SDP which skips two network round trips.

See https://datatracker.ietf.org/doc/draft-hancke-tsvwg-snap/

Author: fippo

association.go

@@ -31,6 +31,15 @@ const defaultSCTPSrcDstPort = 5000
 // Use global random generator to properly seed by crypto grade random.
 var globalMathRandomGenerator = randutil.NewMathRandomGenerator() // nolint:gochecknoglobals
 
+// Generates a non-zero Initiate tag.
+func generateInitiateTag() uint32 {
+	for {
+		if u := globalMathRandomGenerator.Uint32(); u != 0 {
+			return u
+		}
+	}
+}
+
 // Association errors.
 var (
 	ErrChunk                         = errors.New("abort chunk, with following errors")
@@ -315,6 +324,12 @@ type Association struct {
 	tlrStartTime time.Time // time of first recovery RTT
 }
 
+type snapConfig struct {
+	// Local and remote SCTP init to use for SNAP
+	localInit  []byte
+	remoteInit []byte
+}
+
 // Config collects the arguments to createAssociation construction into
 // a single structure.
 type Config struct {
@@ -340,6 +355,9 @@ type Config struct {
 
 	// RACK config options
 	rack rackSettings
+
+	// SNAP/sctp-init
+	snapConfig *snapConfig
 }
 
 // Server accepts a SCTP stream over a conn.
@@ -385,11 +403,37 @@ func createClientWithContext(ctx context.Context, config Config) (*Association,
 	return createClientWithOptionsWithContext(ctx, config)
 }
 
+func createSNAPAssociation(config *Config) (*Association, error) {
+	// SNAP, aka sctp-init in the SDP.
+	remote := &chunkInit{}
+	err := remote.unmarshal(config.snapConfig.remoteInit)
+	if err != nil {
+		return nil, err
+	}
+	local := &chunkInit{}
+	err = local.unmarshal(config.snapConfig.localInit)
+	if err != nil {
+		return nil, err
+	}
+	assoc := createAssociationFromConfigWithTsn(config, local.initialTSN)
+	assoc.initWithOutOfBandTokens(local, remote)
+
+	return assoc, nil
+}
+
 func createClientWithOptionsWithContext(ctx context.Context, opts ...ClientOption) (*Association, error) {
+	config, err := buildClientConfig(opts...)
+	if err != nil {
+		return nil, err
+	}
+	if config.snapConfig != nil && len(config.snapConfig.remoteInit) != 0 && len(config.snapConfig.localInit) != 0 {
+		return createSNAPAssociation(config)
+	}
 	assoc, err := createClientAssociation(opts...)
 	if err != nil {
 		return nil, err
 	}
+
 	assoc.initClient()
 
 	select {
@@ -431,7 +475,7 @@ func createServerAssociation(opts ...ServerOption) (*Association, error) {
 		return nil, err
 	}
 
-	return createAssociationFromConfig(cfg), nil
+	return createAssociationFromConfig(cfg)
 }
 
 func (a *Association) initServer() {
@@ -512,7 +556,7 @@ func createClientAssociation(opts ...ClientOption) (*Association, error) {
 		return nil, err
 	}
 
-	return createAssociationFromConfig(cfg), nil
+	return createAssociationFromConfig(cfg)
 }
 
 func (a *Association) initClient() {
@@ -589,6 +633,8 @@ func (c Config) applyClient(cfg *Config) error { //nolint:dupl,cyclop
 
 	cfg.rack = c.rack
 
+	cfg.snapConfig = c.snapConfig
+
 	return nil
 }
 
@@ -614,7 +660,13 @@ func buildClientConfig(opts ...ClientOption) (*Config, error) {
 	return cfg, nil
 }
 
-func createAssociationFromConfig(cfg *Config) *Association {
+func createAssociationFromConfig(cfg *Config) (*Association, error) {
+	tsn := globalMathRandomGenerator.Uint32()
+
+	return createAssociationFromConfigWithTsn(cfg, tsn), nil
+}
+
+func createAssociationFromConfigWithTsn(cfg *Config, tsn uint32) *Association {
 	maxReceiveBufferSize := cfg.MaxReceiveBufferSize
 	if maxReceiveBufferSize == 0 {
 		maxReceiveBufferSize = initialRecvBufSize
@@ -632,7 +684,6 @@ func createAssociationFromConfig(cfg *Config) *Association {
 
 	rtoMax := cfg.RTOMax
 
-	tsn := globalMathRandomGenerator.Uint32()
 	assoc := &Association{
 		netConn:              cfg.NetConn,
 		maxReceiveBufferSize: maxReceiveBufferSize,
@@ -650,7 +701,7 @@ func createAssociationFromConfig(cfg *Config) *Association {
 		controlQueue:            newControlQueue(),
 		mtu:                     mtu,
 		maxPayloadSize:          mtu - (commonHeaderSize + dataChunkHeaderSize),
-		myVerificationTag:       globalMathRandomGenerator.Uint32(),
+		myVerificationTag:       generateInitiateTag(),
 		initialTSN:              tsn,
 		myNextTSN:               tsn,
 		myNextRSN:               tsn,
@@ -717,6 +768,43 @@ func createAssociationFromConfig(cfg *Config) *Association {
 	return assoc
 }
 
+func (a *Association) initWithOutOfBandTokens(localInit *chunkInit, remoteInit *chunkInit) {
+	a.lock.Lock()
+	defer a.lock.Unlock()
+
+	go a.readLoop()
+	go a.writeLoop()
+
+	a.payloadQueue.init(remoteInit.initialTSN - 1)
+	a.myMaxNumInboundStreams = min16(localInit.numInboundStreams, remoteInit.numInboundStreams)
+	a.myMaxNumOutboundStreams = min16(localInit.numOutboundStreams, remoteInit.numOutboundStreams)
+	a.setRWND(remoteInit.advertisedReceiverWindowCredit)
+	a.peerVerificationTag = remoteInit.initiateTag
+	a.sourcePort = defaultSCTPSrcDstPort
+	a.destinationPort = defaultSCTPSrcDstPort
+	for _, param := range remoteInit.params {
+		switch v := param.(type) { // nolint:gocritic
+		case *paramSupportedExtensions:
+			for _, t := range v.ChunkTypes {
+				if t == ctForwardTSN {
+					a.log.Debugf("[%s] use ForwardTSN (on init)", a.name)
+					a.useForwardTSN = true
+				}
+			}
+		case *paramZeroChecksumAcceptable:
+			a.sendZeroChecksum = v.edmid == dtlsErrorDetectionMethod
+		}
+	}
+
+	if !a.useForwardTSN {
+		a.log.Warnf("[%s] not using ForwardTSN (on init)", a.name)
+	}
+
+	a.ssthresh = a.RWND()
+
+	a.setState(established)
+}
+
 // caller must hold a.lock.
 func (a *Association) sendInit() error {
 	a.log.Debugf("[%s] sending INIT", a.name)
@@ -1714,7 +1802,7 @@ func (a *Association) handleInitAck(pkt *packet, initChunkAck *chunkInitAck) err
 	a.setRWND(initChunkAck.advertisedReceiverWindowCredit)
 	a.log.Debugf("[%s] initial rwnd=%d", a.name, a.RWND())
 
-	// RFC 4690 Sec 7.2.1
+	// RFC 4960 Sec 7.2.1
 	//  o  The initial value of ssthresh MAY be arbitrarily high (for
 	//     example, implementations MAY use the size of the receiver
 	//     advertised window).
@@ -4222,3 +4310,39 @@ func (a *Association) sendActiveHeartbeatLocked() {
 	})
 	a.awakeWriteLoop()
 }
+
+// GenerateOutOfBandToken generates an out-of-band connection token (i.e. a
+// serialized SCTP INIT chunk) for use with SNAP.
+func GenerateOutOfBandToken(opts ...ClientOption) ([]byte, error) {
+	config := &Config{}
+	config.applyDefaults()
+
+	for _, opt := range opts {
+		if opt == nil {
+			continue
+		}
+		if err := opt.applyClient(config); err != nil {
+			return nil, err
+		}
+	}
+
+	config.applyDefaults()
+
+	init := &chunkInit{}
+	init.initialTSN = globalMathRandomGenerator.Uint32()
+	init.numOutboundStreams = math.MaxUint16
+	init.numInboundStreams = math.MaxUint16
+	init.initiateTag = generateInitiateTag()
+	init.advertisedReceiverWindowCredit = config.MaxReceiveBufferSize
+	setSupportedExtensions(&init.chunkInitCommon)
+
+	if config.EnableZeroChecksum {
+		init.params = append(init.params, &paramZeroChecksumAcceptable{edmid: dtlsErrorDetectionMethod})
+	}
+	_, err := init.check()
+	if err != nil {
+		return nil, err
+	}
+
+	return init.marshal()
+}

association_options.go

@@ -163,3 +163,18 @@ func WithCwndCAStep(cwndCAStep uint32) AssociationOption {
 		return nil
 	})
 }
+
+// WithSNAP enables SNAP, https://datatracker.ietf.org/doc/draft-hancke-tsvwg-snap/.
+func WithSNAP(localSctpInit []byte, remoteSctpInit []byte) AssociationOption {
+	return sharedOption(func(c *Config) error {
+		if len(localSctpInit) == 0 || len(remoteSctpInit) == 0 {
+			return errInvalidSnapToken
+		}
+		c.snapConfig = &snapConfig{
+			localInit:  localSctpInit,
+			remoteInit: remoteSctpInit,
+		}
+
+		return nil
+	})
+}

association_options_test.go

@@ -68,6 +68,14 @@ func TestAssociationOptions_Validation(t *testing.T) {
 		err := WithRTOMax(0).applyServer(&cfg)
 		assert.ErrorIs(t, err, errInvalidRTOMax)
 	})
+
+	t.Run("snap nil arguments", func(t *testing.T) {
+		var cfg Config
+		err := WithSNAP(nil, nil).applyServer(&cfg)
+		assert.ErrorIs(t, err, errInvalidSnapToken)
+		err = WithSNAP([]byte{}, []byte{}).applyServer(&cfg)
+		assert.ErrorIs(t, err, errInvalidSnapToken)
+	})
 }
 
 func TestClientWithOptions_ValidatesOptionValues(t *testing.T) {

association_test.go

@@ -4842,7 +4842,8 @@ func TestAbortStillSendsWhenWriteLoopClosing(t *testing.T) {
 		EnableZeroChecksum: false,
 	}
 	cfg.applyDefaults()
-	assoc := createAssociationFromConfig(&cfg)
+	assoc, err := createAssociationFromConfig(&cfg)
+	assert.NoError(t, err)
 	assoc.initServer()
 
 	// Simulate the problematic timing: writeLoop is sitting in its select and
@@ -4921,7 +4922,8 @@ func TestAbort_WaitsForAbortWriteAttempt(t *testing.T) {
 		MaxMessageSize: 1200,
 	}
 	cfg.applyDefaults()
-	assoc := createAssociationFromConfig(&cfg)
+	assoc, err := createAssociationFromConfig(&cfg)
+	assert.NoError(t, err)
 	assoc.initServer()
 
 	assoc.Abort("test")
@@ -4944,7 +4946,8 @@ func TestAbortSentChClosedWhenAbortMarshalFails(t *testing.T) {
 		MaxMessageSize: 1200,
 	}
 	cfg.applyDefaults()
-	assoc := createAssociationFromConfig(&cfg)
+	assoc, err := createAssociationFromConfig(&cfg)
+	assert.NoError(t, err)
 	assoc.initServer()
 
 	assoc.lock.Lock()
@@ -4961,3 +4964,94 @@ func TestAbortSentChClosedWhenAbortMarshalFails(t *testing.T) {
 		require.Fail(t, "abortSentCh was not closed when ABORT marshaling failed")
 	}
 }
+
+func TestAssociationSNAPInvalidInit(t *testing.T) {
+	br := test.NewBridge()
+	tokenConfig := Config{
+		MaxReceiveBufferSize: 65535,
+		EnableZeroChecksum:   false,
+	}
+	init, err := GenerateOutOfBandToken(tokenConfig)
+	assert.NoError(t, err)
+
+	_, err = ClientWithOptions(
+		WithNetConn(br.GetConn0()),
+		WithSNAP([]byte{1, 2}, init))
+	assert.ErrorIs(t, err, ErrChunkHeaderTooSmall)
+
+	_, err = ClientWithOptions(
+		WithNetConn(br.GetConn1()),
+		WithSNAP(init, []byte{1, 2}))
+	assert.ErrorIs(t, err, ErrChunkHeaderTooSmall)
+}
+
+func TestAssociationSNAP(t *testing.T) {
+	lim := test.TimeOut(time.Second * 10)
+	defer lim.Stop()
+
+	loggerFactory := logging.NewDefaultLoggerFactory()
+	br := test.NewBridge()
+
+	// Use GenerateOutOfBandToken to create the init chunks
+	tokenConfig := Config{
+		MaxReceiveBufferSize: 65535,
+		EnableZeroChecksum:   false,
+	}
+	initA, err := GenerateOutOfBandToken(tokenConfig)
+	assert.NoError(t, err)
+
+	initB, err := GenerateOutOfBandToken(tokenConfig)
+	assert.NoError(t, err)
+
+	assocA, err := ClientWithOptions(
+		WithName("a"),
+		WithNetConn(br.GetConn0()),
+		WithLoggerFactory(loggerFactory),
+		WithSNAP(initA, initB))
+	assert.NoError(t, err)
+	assert.NotNil(t, assocA)
+
+	assocB, err := ClientWithOptions(
+		WithName("b"),
+		WithNetConn(br.GetConn1()),
+		WithLoggerFactory(loggerFactory),
+		WithSNAP(initB, initA))
+	assert.NoError(t, err)
+	assert.NotNil(t, assocB)
+
+	const si uint16 = 1
+	const msg = "SNAP is snappy"
+
+	streamA, err := assocA.OpenStream(si, PayloadTypeWebRTCBinary)
+	assert.NoError(t, err)
+
+	_, err = streamA.WriteSCTP([]byte(msg), PayloadTypeWebRTCBinary)
+	assert.NoError(t, err)
+
+	br.Process()
+
+	accepted := make(chan *Stream)
+	go func() {
+		s, errAccept := assocB.AcceptStream()
+		assert.NoError(t, errAccept)
+		accepted <- s
+	}()
+
+	flushBuffers(br, assocA, assocB)
+
+	var streamB *Stream
+	select {
+	case streamB = <-accepted:
+	case <-time.After(5 * time.Second):
+		assert.Fail(t, "timed out waiting for accept stream")
+	}
+
+	buf := make([]byte, 64)
+	n, ppi, err := streamB.ReadSCTP(buf)
+	assert.NoError(t, err, "ReadSCTP failed")
+	assert.Equal(t, len(msg), n, "unexpected length of received data")
+	assert.Equal(t, PayloadTypeWebRTCBinary, ppi, "unexpected ppi")
+	assert.Equal(t, msg, string(buf[:n]))
+
+	closeAssociationPair(br, assocA, assocB)
+}

errors.go

@@ -33,4 +33,7 @@ var (
 	// errInvalidRackWcDelAck indicates the receiver worst-case delayed-ACK for PTO when only 1 packet in flight
 	// was set to < 0.
 	errInvalidRackWcDelAck = errors.New("RackWcDelAck was set to <= 0")
+
+	// errInvalidSnapToken indicates a SNAP token that is not parseable.
+	errInvalidSnapToken = errors.New("SNAP token is invalid")
 )