Repository: pion/dtls
Pull request #112 (2019-08-28 19:35:54)

Fragmenting sent handshake messages
#### Description
Previously handshake messages being sent out would not be fragmented. Most handshake messages are pretty small but a couple of exceptions, such as Certificate, mean the packet size can be greater than the MTU, which then causes some packet loss and the handshake would never be able to complete. This changes the `internalSend` function to fragment handshakes to the set MTU.

We set `MTU` as a config option, which defaults to `1200`

#### Reference issue


RyanGordon: This is ready for review now. A couple of notes:

I don't _love_ the `Marshal()` interface, but it didn't really make sense to fit it anywhere else. I can't fragment before `packet.Marshal()`, because the handshake is just an in-memory struct still; we have no idea the full size of the handshake message itself. And I can't do it after because we need to manipulate the record and handshake headers to support fragmentation properly to spec. It might take a larger refactor to have a cleaner interface. Open to thoughts here.

I also realizing I needed to tackle the
`// sequenceNumber and messageSequence line up, may need to be re-evaluated`
tech debt because the possibility of fragmentation means these will not align anymore, so there is slightly more code tweaked to fix this case.

One last change was a small improvement to the way the "handshake completed signaling" works so that we don't end up in a situation where we could cause a panic calling an already-closed channel. I can split this out if desired / I may move this change to https://github.com/pion/dtls/pull/113
Sean-Der: I am going to spend some time with this, but I am not a big fan of how the MTU is split out all over the code base.

I need to think over it, but it would be really nice if we didn't have to change `Marshal()` but just had a single function that took care of fragmenting for us.

 I don't think there is much danger to double parse/allocate since this isn't a hot path. I do think this could start to introduce substantial code complexity. Maybe this is the only way we can do it, but I will think it over and come back! 
Sean-Der: @RyanGordon re handshake complete stuff, merge away! Thank you so much for putting it in a smaller PR, it makes it so much easier to just worry about a single thing.
Sean-Der: What do you think of keeping something in memory that maps `original sequence number` -> `fragmented`

Then only when it comes time to encrypt and send we can say `We are sending N`, but `N-1`  was actually these three so we need to account for that.

This adds more state/complexity to that function. But I just think in the long run it will pay off to have something that is very contained/testable. 
RyanGordon: I agree it would be nicer if we could keep `Marshal()` the same. I am going to think more on this but the basic complexity is the way the inheritance works with marshaling:

recordHeader.Marshal -> handshakeHeader.Marshal -> handshake.Marshal

If we reversed it, then it becomes much easier to keep the fragmenting and other stateful logic out of Marshal

i.e. very rough-cut:
```
internalSend(message) {
  var packets [][]byte
  if message.type == handshake {
    handshakes := fragment_handshake(message.handshake)
    for handshake in handshakes {
       record = ...
       packets = append(packets, record.Marshal())
   }
  } else {
   packets = [message.Marshal()]
  }

  // send packets
}

fragment_handshake(handshake) {
  content := handshake.content.Marshal()
  origHeader := handshake.header
  fragments = split(content, MTU_SIZE)
  
  handshakes = []
  for fragment in fragments {
     fragmentHeader = copy(origHeader)
     fragmentHeader.offset = ...
     fragmentHeader.len = ...
     fragmentHeader.content = fragment
     handshakes = append(handshakes, fragmentHeader)
  }
  
  return handshakes
}

```
Sean-Der: This is fantastic @RyanGordon!

I am really impressed how fast you rewrote/fixed up this PR. Reading through the code I like it a lot more, it is just nice that it is only in one place.

I think that is the hardest part about OpenSSL. A lot of the TLS code duplicates logic/ a lot of things are interwoven so it is hard to follow.

This is a merge from me!
Sean-Der: So checking out the PR I think that if we fire messages for a flight multiple times this will cause issues.

It looks like we increment the sequenceNumber every time we send a fragment. I think this a good design decision.

However if we re-run a flight multiple times the SequenceNumber will not account for that. I think the only change we need to make is to record `sequenceNumber at start of flight` and then we should be good! Then every time we re-run that flight we need to reset back.

Happy to take care of it also! 
Sean-Der: Another thing I realized, we currently might send different packets with the same sequenceNumber.

If this flight handler has run multiple times https://github.com/pion/dtls/blob/master/server_handlers.go#L481 we don't increment the global sequenceNumber counter until we have moved onto the next flight. 

If we hit an error during handshaking https://github.com/pion/dtls/blob/master/conn.go#L391 we will reuse a sequence number that has already been used. The global sequenceNumber will just be lower than what has actually gone out.

We need to handle sequenceNumbers and replaying flights better in general. The code is confusing error prone. Maybe we should 'replay flights' by pulling them from the local handshake cache? Then we can increment the global sequenceNumber cache w/e we want.


RyanGordon: > So checking out the PR I think that if we fire messages for a flight multiple times this will cause issues.
> 
> It looks like we increment the sequenceNumber every time we send a fragment. That works great for the happy case.
> 
> However if we re-run a flight multiple times the SequenceNumber will not account for that. I think the only change we need to make is to record `sequenceNumber at start of flight` and then we should be good. Then every time we re-run that flight we need to reset back.
> 
> Happy to take care of it also!

Hmm, I think this is actually how it works because `c.handshakeMessageSequence` is used for the handshake header `messageSequence` field and `c.state.localSequenceNumber` is only used for the record header `sequenceNumber` field

`c.handshakeMessageSequence` is assigned based on the old logic that would correctly set the handshake header `messageSequence` to the right value for the state of the sequence, including replays.

And now `c.state.localSequenceNumber` is only used for the record and is incremented every time a record is sent so it is always unique for each sent record (even if the record contains retransmission content). At least according to RFC6347 4.1.2.6 seems to say that the record layer sequence number should always be unique or duplicate packets could be dropped by the receiver depending on the library implementation.
Sean-Der: @RyanGordon 

(These are just assumptions, please correct me where I am wrong!)

* https://github.com/pion/dtls/blob/bc94af06cd09f3ec7fb246e5e0a264382da4e19e/conn.go#L379 always increments the SequenceNumber. We will never have a message with the same sequence number twice.

* This means that each call to internal send will use new sequenceNumbers https://github.com/pion/dtls/blob/bc94af06cd09f3ec7fb246e5e0a264382da4e19e/client_handlers.go#L292

`pion/dtls` itself is unable to handle gaps in sequence numbers. It is very explicit about the SequenceNumber that is used for certain messages https://github.com/pion/dtls/blob/bc94af06cd09f3ec7fb246e5e0a264382da4e19e/server_handlers.go#L188
RyanGordon: Hmm okay, that is interesting. I need to do more research into this to make sure I am understanding the spec correctly and that this doesn't introduce breaking behavior. (Currently) My understanding is that the record "sequence number" (I believe) should always be incrementing even on re-transmit. The handshake "message sequence" is the one that should have the specific ordering and be re-used depending on the state of the handshake and used to know where in the handshake the sender is.

I will be double checking what the spec behavior _should be_, and what impact these changes  would have and what would need to be changed to avoid any impact
Sean-Der: Fantastic, sorry for this review being a doozy! I really appreciate how much you are doing to land this.

I would see what OpenSSL does on retransmissions (if it increments the SequenceNumber), and honestly if E2E Lossy passes I think we are in great shape. 

I really need to make a nice way for us to do E2E Lossy against OpenSSL.
daenney: > I really need to make a nice way for us to do E2E Lossy against OpenSSL.
I'm working on that. Once I've recovered from a 9hr timezone shift I'll submit
Sean-Der: > > I really need to make a nice way for us to do E2E Lossy against OpenSSL.
> > I'm working on that. Once I've recovered from a 9hr timezone shift I'll submit

Well that is just a fantastic surprise. Thanks for grabbing that!
RyanGordon: Ok, this is ready for a final review @Sean-Der, et al. Tests should pass but golangci-lint seems to be doing something weird (related to Go 1.13 release?)
Sean-Der: Nice @RyanGordon !

Mind rebasing this off of you 1.12 pin I will make sure I review this tonight! If it passes everything that is for sure going to be a merge from me.

really nice work

Patch:
diff --git a/client_handlers.go b/client_handlers.go
index a70b160..0d9abd4 100644
--- a/client_handlers.go
+++ b/client_handlers.go
@@ -3,6 +3,7 @@ package dtls
 import (
 	"bytes"
 	"fmt"
+	"sync/atomic"
 )
 
 func initalizeCipherSuite(c *Conn, h *handshakeMessageServerKeyExchange) (*alert, error) {
@@ -171,7 +172,7 @@ func clientHandshakeHandler(c *Conn) (*alert, error) {
 			if alertPtr, err := handleSingleHandshake(expectedMessages[0].data); err != nil {
 				return alertPtr, err
 			}
-			c.state.localSequenceNumber++
+			c.handshakeMessageSequence++
 		case expectedMessages[1] != nil:
 			if alertPtr, err := handleSingleHandshake(expectedMessages[1].data); err != nil {
 				return alertPtr, err
@@ -220,7 +221,7 @@ func clientHandshakeHandler(c *Conn) (*alert, error) {
 			}
 		}
 
-		c.state.localSequenceNumber++
+		c.handshakeMessageSequence++
 		c.currFlight.set(flight5)
 	case flight5:
 		expectedMessages := c.handshakeCache.pull(
@@ -234,7 +235,8 @@ func clientHandshakeHandler(c *Conn) (*alert, error) {
 		}
 
 		c.setLocalEpoch(1)
-		c.state.localSequenceNumber = 1
+		c.handshakeMessageSequence = 1
+		atomic.StoreUint64(&c.state.localSequenceNumber, 1)
 		c.handshakeDoneSignal.Close()
 	default:
 		return &alert{alertLevelFatal, alertUnexpectedMessage}, fmt.Errorf("client asked to handle unknown flight (%d)", c.currFlight.get())
@@ -289,13 +291,11 @@ func clientFlightHandler(c *Conn) (bool, *alert, error) {
 
 		c.internalSend(&recordLayer{
 			recordLayerHeader: recordLayerHeader{
-				sequenceNumber:  c.state.localSequenceNumber,
 				protocolVersion: protocolVersion1_2,
 			},
 			content: &handshake{
-				// sequenceNumber and messageSequence line up, may need to be re-evaluated
 				handshakeHeader: handshakeHeader{
-					messageSequence: uint16(c.state.localSequenceNumber),
+					messageSequence: uint16(c.handshakeMessageSequence),
 				},
 				handshakeMessage: &handshakeMessageClientHello{
 					version:            protocolVersion1_2,
@@ -313,23 +313,21 @@ func clientFlightHandler(c *Conn) (bool, *alert, error) {
 			return true, nil, nil
 		}
 
-		sequenceNumber := c.state.localSequenceNumber
+		messageSequence := c.handshakeMessageSequence
 		if c.remoteRequestedCertificate {
 			c.internalSend(&recordLayer{
 				recordLayerHeader: recordLayerHeader{
-					sequenceNumber:  c.state.localSequenceNumber,
 					protocolVersion: protocolVersion1_2,
 				},
 				content: &handshake{
-					// sequenceNumber and messageSequence line up, may need to be re-evaluated
 					handshakeHeader: handshakeHeader{
-						messageSequence: uint16(c.state.localSequenceNumber),
+						messageSequence: uint16(messageSequence),
 					},
 					handshakeMessage: &handshakeMessageCertificate{
 						certificate: c.localCertificate,
 					}},
 			}, false)
-			sequenceNumber++
+			messageSequence++
 		}
 
 		clientKeyExchange := &handshakeMessageClientKeyExchange{}
@@ -341,19 +339,17 @@ func clientFlightHandler(c *Conn) (bool, *alert, error) {
 
 		c.internalSend(&recordLayer{
 			recordLayerHeader: recordLayerHeader{
-				sequenceNumber:  sequenceNumber,
 				protocolVersion: protocolVersion1_2,
 			},
 			content: &handshake{
-				// sequenceNumber and messageSequence line up, may need to be re-evaluated
 				handshakeHeader: handshakeHeader{
-					messageSequence: uint16(sequenceNumber),
+					messageSequence: uint16(messageSequence),
 				},
 				handshakeMessage: clientKeyExchange,
 			},
 		}, false)
 
-		sequenceNumber++
+		messageSequence++
 
 		serverKeyExchangeData := c.handshakeCache.pullAndMerge(
 			handshakeCachePullRule{handshakeTypeServerKeyExchange, false},
@@ -411,13 +407,11 @@ func clientFlightHandler(c *Conn) (bool, *alert, error) {
 
 			c.internalSend(&recordLayer{
 				recordLayerHeader: recordLayerHeader{
-					sequenceNumber:  sequenceNumber,
 					protocolVersion: protocolVersion1_2,
 				},
 				content: &handshake{
-					// sequenceNumber and messageSequence line up, may need to be re-evaluated
 					handshakeHeader: handshakeHeader{
-						messageSequence: uint16(sequenceNumber),
+						messageSequence: uint16(messageSequence),
 					},
 					handshakeMessage: &handshakeMessageCertificateVerify{
 						hashAlgorithm:      HashAlgorithmSHA256,
@@ -425,12 +419,11 @@ func clientFlightHandler(c *Conn) (bool, *alert, error) {
 						signature:          c.localCertificateVerify,
 					}},
 			}, false)
-			sequenceNumber++
+			messageSequence++
 		}
 
 		c.internalSend(&recordLayer{
 			recordLayerHeader: recordLayerHeader{
-				sequenceNumber:  sequenceNumber,
 				protocolVersion: protocolVersion1_2,
 			},
 			content: &changeCipherSpec{},
@@ -456,17 +449,16 @@ func clientFlightHandler(c *Conn) (bool, *alert, error) {
 			}
 		}
 
-		// TODO: Fix hard-coded epoch & sequenceNumber, taking retransmitting into account.
+		// TODO: Fix hard-coded epoch, taking retransmitting into account.
+		atomic.StoreUint64(&c.state.localSequenceNumber, 0)
 		c.internalSend(&recordLayer{
 			recordLayerHeader: recordLayerHeader{
 				epoch:           1,
-				sequenceNumber:  0, // sequenceNumber restarts per epoch
 				protocolVersion: protocolVersion1_2,
 			},
 			content: &handshake{
-				// sequenceNumber and messageSequence line up, may need to be re-evaluated
 				handshakeHeader: handshakeHeader{
-					messageSequence: uint16(sequenceNumber), // KeyExchange + 1
+					messageSequence: uint16(messageSequence),
 				},
 				handshakeMessage: &handshakeMessageFinished{
 					verifyData: c.localVerifyData,
diff --git a/config.go b/config.go
index 28baa18..dc620dd 100644
--- a/config.go
+++ b/config.go
@@ -81,6 +81,10 @@ type Config struct {
 	// ConnectTimeout is the timeout threshold for new connection handshakes
 	// to complete (default is 30 seconds)
 	ConnectTimeout *time.Duration
+
+	// MTU is the length at which handshake messages will be fragmented to
+	// fit within the maximum transmission unit (default is 1200 bytes)
+	MTU int
 }
 
 const defaultConnectTimeout = 30 * time.Second
@@ -90,6 +94,8 @@ func ConnectTimeoutOption(timeout time.Duration) *time.Duration {
 	return &timeout
 }
 
+const defaultMTU = 1200 // bytes
+
 // PSKCallback is called once we have the remote's PSKIdentityHint.
 // If the remote provided none it will be nil
 type PSKCallback func([]byte) ([]byte, error)
diff --git a/conn.go b/conn.go
index d6a2617..441db86 100644
--- a/conn.go
+++ b/conn.go
@@ -45,6 +45,8 @@ type Conn struct {
 
 	connectTimeout time.Duration
 
+	maximumTransmissionUnit int
+
 	remoteRequestedCertificate bool // Did we get a CertificateRequest
 
 	localSRTPProtectionProfiles []SRTPProtectionProfile // Available SRTPProtectionProfiles, if empty no SRTP support
@@ -73,6 +75,7 @@ type Conn struct {
 	rootCAs               *x509.CertPool
 	serverName            string
 
+	handshakeMessageSequence       int
 	handshakeMessageHandler        handshakeMessageHandler
 	flightHandler                  flightHandler
 	handshakeDoneSignal            *Closer
@@ -126,6 +129,11 @@ func createConn(nextConn net.Conn, flightHandler flightHandler, handshakeMessage
 		connectTimeout = math.MaxInt64 * time.Nanosecond
 	}
 
+	mtu := config.MTU
+	if mtu <= 0 {
+		mtu = defaultMTU
+	}
+
 	handshakeDoneSignal := NewCloser()
 
 	c := &Conn{
@@ -136,6 +144,7 @@ func createConn(nextConn net.Conn, flightHandler flightHandler, handshakeMessage
 		handshakeMessageHandler:     handshakeMessageHandler,
 		flightHandler:               flightHandler,
 		connectTimeout:              connectTimeout,
+		maximumTransmissionUnit:     mtu,
 		localCertificate:            config.Certificate,
 		localPrivateKey:             config.PrivateKey,
 		clientAuth:                  config.ClientAuth,
@@ -267,14 +276,12 @@ func (c *Conn) Write(p []byte) (int, error) {
 	c.internalSend(&recordLayer{
 		recordLayerHeader: recordLayerHeader{
 			epoch:           c.getLocalEpoch(),
-			sequenceNumber:  c.state.localSequenceNumber,
 			protocolVersion: protocolVersion1_2,
 		},
 		content: &applicationData{
 			data: p,
 		},
 	}, true)
-	c.state.localSequenceNumber++
 
 	return len(p), nil
 }
@@ -341,30 +348,114 @@ func (c *Conn) ExportKeyingMaterial(label string, context []byte, length int) ([
 	return prfPHash(c.state.masterSecret, seed, length, c.state.cipherSuite.hashFunc())
 }
 
-func (c *Conn) internalSend(pkt *recordLayer, shouldEncrypt bool) {
+func (c *Conn) internalSend(record *recordLayer, shouldEncrypt bool) {
+	var packets [][]byte
 
-	raw, err := pkt.Marshal()
-	if err != nil {
-		c.stopWithError(err)
-		return
-	}
+	if h, ok := record.content.(*handshake); ok {
+		handshakeRaw, err := record.Marshal()
+		if err != nil {
+			c.stopWithError(err)
+			return
+		}
 
-	if h, ok := pkt.content.(*handshake); ok {
 		c.log.Tracef("[handshake] -> %s", h.handshakeHeader.handshakeType.String())
-		c.handshakeCache.push(raw[recordLayerHeaderSize:], h.handshakeHeader.messageSequence, h.handshakeHeader.handshakeType, c.state.isClient)
-	}
+		c.handshakeCache.push(handshakeRaw[recordLayerHeaderSize:], h.handshakeHeader.messageSequence, h.handshakeHeader.handshakeType, c.state.isClient)
 
-	if shouldEncrypt {
-		raw, err = c.state.cipherSuite.encrypt(pkt, raw)
+		handshakeFragments, err := c.fragmentHandshake(h)
 		if err != nil {
 			c.stopWithError(err)
 			return
 		}
+
+		for _, handshakeFragment := range handshakeFragments {
+			recordLayerHeader := &recordLayerHeader{
+				contentType:     record.recordLayerHeader.contentType,
+				contentLen:      uint16(len(handshakeFragment)),
+				protocolVersion: record.recordLayerHeader.protocolVersion,
+				epoch:           record.recordLayerHeader.epoch,
+				sequenceNumber:  atomic.LoadUint64(&c.state.localSequenceNumber),
+			}
+
+			atomic.AddUint64(&c.state.localSequenceNumber, 1)
+
+			recordLayerHeaderBytes, err := recordLayerHeader.Marshal()
+			if err != nil {
+				c.stopWithError(err)
+				return
+			}
+
+			packet := append(recordLayerHeaderBytes, handshakeFragment...)
+			packets = append(packets, packet)
+		}
+	} else {
+		record.recordLayerHeader.sequenceNumber = atomic.LoadUint64(&c.state.localSequenceNumber)
+		atomic.AddUint64(&c.state.localSequenceNumber, 1)
+
+		packet, err := record.Marshal()
+		if err != nil {
+			c.stopWithError(err)
+			return
+		}
+
+		packets = [][]byte{packet}
 	}
 
-	if _, err := c.nextConn.Write(raw); err != nil {
-		c.stopWithError(err)
+	for _, packet := range packets {
+		if shouldEncrypt {
+			var err error
+			packet, err = c.state.cipherSuite.encrypt(record, packet)
+			if err != nil {
+				c.stopWithError(err)
+				return
+			}
+		}
+
+		if _, err := c.nextConn.Write(packet); err != nil {
+			c.stopWithError(err)
+			return
+		}
+	}
+}
+
+func (c *Conn) fragmentHandshake(h *handshake) ([][]byte, error) {
+	content, err := h.handshakeMessage.Marshal()
+	if err != nil {
+		return nil, err
+	}
+
+	fragmentedHandshakes := make([][]byte, 0)
+
+	contentFragments := splitBytes(content, c.maximumTransmissionUnit)
+	if len(contentFragments) == 0 {
+		contentFragments = [][]byte{
+			{},
+		}
+	}
+
+	offset := 0
+	for _, contentFragment := range contentFragments {
+		contentFragmentLen := len(contentFragment)
+
+		handshakeHeaderFragment := &handshakeHeader{
+			handshakeType:   h.handshakeHeader.handshakeType,
+			length:          h.handshakeHeader.length,
+			messageSequence: h.handshakeHeader.messageSequence,
+			fragmentOffset:  uint32(offset),
+			fragmentLength:  uint32(contentFragmentLen),
+		}
+
+		offset += contentFragmentLen
+
+		handshakeHeaderFragmentRaw, err := handshakeHeaderFragment.Marshal()
+		if err != nil {
+			return nil, err
+		}
+
+		fragmentedHandshake := append(handshakeHeaderFragmentRaw, contentFragment...)
+		fragmentedHandshakes = append(fragmentedHandshakes, fragmentedHandshake)
 	}
+
+	return fragmentedHandshakes, nil
 }
 
 func (c *Conn) inboundLoop() {
@@ -482,7 +573,6 @@ func (c *Conn) notify(level alertLevel, desc alertDescription) {
 	c.internalSend(&recordLayer{
 		recordLayerHeader: recordLayerHeader{
 			epoch:           c.getLocalEpoch(),
-			sequenceNumber:  c.state.localSequenceNumber,
 			protocolVersion: protocolVersion1_2,
 		},
 		content: &alert{
@@ -490,8 +580,6 @@ func (c *Conn) notify(level alertLevel, desc alertDescription) {
 			alertDescription: desc,
 		},
 	}, c.isHandshakeCompletedSuccessfully())
-
-	c.state.localSequenceNumber++
 }
 
 func (c *Conn) setHandshakeCompletedSuccessfully() {
diff --git a/server_handlers.go b/server_handlers.go
index 8d0617a..67a089f 100644
--- a/server_handlers.go
+++ b/server_handlers.go
@@ -3,6 +3,7 @@ package dtls
 import (
 	"bytes"
 	"fmt"
+	"sync/atomic"
 )
 
 func serverHandshakeHandler(c *Conn) (*alert, error) {
@@ -18,7 +19,7 @@ func serverHandshakeHandler(c *Conn) (*alert, error) {
 				if !bytes.Equal(c.cookie, h.cookie) {
 					return &alert{alertLevelFatal, alertAccessDenied}, errCookieMismatch
 				}
-				c.state.localSequenceNumber = 1
+				c.handshakeMessageSequence = 1
 				c.currFlight.set(flight4)
 				break
 			}
@@ -252,16 +253,16 @@ func serverHandshakeHandler(c *Conn) (*alert, error) {
 
 		switch {
 		case c.localPSKIdentityHint != nil:
-			c.state.localSequenceNumber = 4
+			c.handshakeMessageSequence = 4
 		case c.localPSKCallback != nil:
-			c.state.localSequenceNumber = 3
+			c.handshakeMessageSequence = 3
 		case c.clientAuth > NoClientCert:
-			c.state.localSequenceNumber = 6
+			c.handshakeMessageSequence = 6
 		default:
-			c.state.localSequenceNumber = 5
+			c.handshakeMessageSequence = 5
 		}
-		c.setLocalEpoch(1)
 
+		c.setLocalEpoch(1)
 		c.currFlight.set(flight6)
 	}
 	return nil, nil
@@ -277,13 +278,11 @@ func serverFlightHandler(c *Conn) (bool, *alert, error) {
 	case flight2:
 		c.internalSend(&recordLayer{
 			recordLayerHeader: recordLayerHeader{
-				sequenceNumber:  c.state.localSequenceNumber,
 				protocolVersion: protocolVersion1_2,
 			},
 			content: &handshake{
-				// sequenceNumber and messageSequence line up, may need to be re-evaluated
 				handshakeHeader: handshakeHeader{
-					messageSequence: uint16(c.state.localSequenceNumber),
+					messageSequence: uint16(c.handshakeMessageSequence),
 				},
 				handshakeMessage: &handshakeMessageHelloVerifyRequest{
 					version: protocolVersion1_2,
@@ -316,16 +315,14 @@ func serverFlightHandler(c *Conn) (bool, *alert, error) {
 			}...)
 		}
 
-		sequenceNumber := c.state.localSequenceNumber
+		messageSequence := c.handshakeMessageSequence
 		c.internalSend(&recordLayer{
 			recordLayerHeader: recordLayerHeader{
-				sequenceNumber:  sequenceNumber,
 				protocolVersion: protocolVersion1_2,
 			},
 			content: &handshake{
-				// sequenceNumber and messageSequence line up, may need to be re-evaluated
 				handshakeHeader: handshakeHeader{
-					messageSequence: uint16(sequenceNumber),
+					messageSequence: uint16(messageSequence),
 				},
 				handshakeMessage: &handshakeMessageServerHello{
 					version:           protocolVersion1_2,
@@ -335,24 +332,22 @@ func serverFlightHandler(c *Conn) (bool, *alert, error) {
 					extensions:        extensions,
 				}},
 		}, false)
-		sequenceNumber++
+		messageSequence++
 
 		if c.localPSKCallback == nil {
 			c.internalSend(&recordLayer{
 				recordLayerHeader: recordLayerHeader{
-					sequenceNumber:  sequenceNumber,
 					protocolVersion: protocolVersion1_2,
 				},
 				content: &handshake{
-					// sequenceNumber and messageSequence line up, may need to be re-evaluated
 					handshakeHeader: handshakeHeader{
-						messageSequence: uint16(sequenceNumber),
+						messageSequence: uint16(messageSequence),
 					},
 					handshakeMessage: &handshakeMessageCertificate{
 						certificate: c.localCertificate,
 					}},
 			}, false)
-			sequenceNumber++
+			messageSequence++
 
 			if len(c.localKeySignature) == 0 {
 				serverRandom, err := c.state.localRandom.Marshal()
@@ -373,13 +368,11 @@ func serverFlightHandler(c *Conn) (bool, *alert, error) {
 
 			c.internalSend(&recordLayer{
 				recordLayerHeader: recordLayerHeader{
-					sequenceNumber:  sequenceNumber,
 					protocolVersion: protocolVersion1_2,
 				},
 				content: &handshake{
-					// sequenceNumber and messageSequence line up, may need to be re-evaluated
 					handshakeHeader: handshakeHeader{
-						messageSequence: uint16(sequenceNumber),
+						messageSequence: uint16(messageSequence),
 					},
 					handshakeMessage: &handshakeMessageServerKeyExchange{
 						ellipticCurveType:  ellipticCurveTypeNamedCurve,
@@ -390,18 +383,16 @@ func serverFlightHandler(c *Conn) (bool, *alert, error) {
 						signature:          c.localKeySignature,
 					}},
 			}, false)
-			sequenceNumber++
+			messageSequence++
 
 			if c.clientAuth > NoClientCert {
 				c.internalSend(&recordLayer{
 					recordLayerHeader: recordLayerHeader{
-						sequenceNumber:  sequenceNumber,
 						protocolVersion: protocolVersion1_2,
 					},
 					content: &handshake{
-						// sequenceNumber and messageSequence line up, may need to be re-evaluated
 						handshakeHeader: handshakeHeader{
-							messageSequence: uint16(sequenceNumber),
+							messageSequence: uint16(messageSequence),
 						},
 						handshakeMessage: &handshakeMessageCertificateRequest{
 							certificateTypes: []clientCertificateType{clientCertificateTypeRSASign, clientCertificateTypeECDSASign},
@@ -416,7 +407,7 @@ func serverFlightHandler(c *Conn) (bool, *alert, error) {
 						},
 					},
 				}, false)
-				sequenceNumber++
+				messageSequence++
 			}
 		} else if c.localPSKIdentityHint != nil {
 			/* To help the client in selecting which identity to use, the server
@@ -427,30 +418,26 @@ func serverFlightHandler(c *Conn) (bool, *alert, error) {
 			 */
 			c.internalSend(&recordLayer{
 				recordLayerHeader: recordLayerHeader{
-					sequenceNumber:  sequenceNumber,
 					protocolVersion: protocolVersion1_2,
 				},
 				content: &handshake{
-					// sequenceNumber and messageSequence line up, may need to be re-evaluated
 					handshakeHeader: handshakeHeader{
-						messageSequence: uint16(sequenceNumber),
+						messageSequence: uint16(messageSequence),
 					},
 					handshakeMessage: &handshakeMessageServerKeyExchange{
 						identityHint: c.localPSKIdentityHint,
 					}},
 			}, false)
-			sequenceNumber++
+			messageSequence++
 		}
 
 		c.internalSend(&recordLayer{
 			recordLayerHeader: recordLayerHeader{
-				sequenceNumber:  sequenceNumber,
 				protocolVersion: protocolVersion1_2,
 			},
 			content: &handshake{
-				// sequenceNumber and messageSequence line up, may need to be re-evaluated
 				handshakeHeader: handshakeHeader{
-					messageSequence: uint16(sequenceNumber),
+					messageSequence: uint16(messageSequence),
 				},
 				handshakeMessage: &handshakeMessageServerHelloDone{},
 			},
@@ -458,7 +445,6 @@ func serverFlightHandler(c *Conn) (bool, *alert, error) {
 	case flight6:
 		c.internalSend(&recordLayer{
 			recordLayerHeader: recordLayerHeader{
-				sequenceNumber:  c.state.localSequenceNumber,
 				protocolVersion: protocolVersion1_2,
 			},
 			content: &changeCipherSpec{},
@@ -485,16 +471,15 @@ func serverFlightHandler(c *Conn) (bool, *alert, error) {
 			}
 		}
 
+		atomic.StoreUint64(&c.state.localSequenceNumber, 0)
 		c.internalSend(&recordLayer{
 			recordLayerHeader: recordLayerHeader{
 				epoch:           1,
-				sequenceNumber:  0, // sequenceNumber restarts per epoch
 				protocolVersion: protocolVersion1_2,
 			},
 			content: &handshake{
-				// sequenceNumber and messageSequence line up, may need to be re-evaluated
 				handshakeHeader: handshakeHeader{
-					messageSequence: uint16(c.state.localSequenceNumber), // KeyExchange + 1
+					messageSequence: uint16(c.handshakeMessageSequence),
 				},
 
 				handshakeMessage: &handshakeMessageFinished{
diff --git a/state.go b/state.go
index c48bae2..822d078 100644
--- a/state.go
+++ b/state.go
@@ -75,7 +75,7 @@ func (s *State) serialize() (*serializedState, error) {
 		RemoteEpoch:           s.remoteEpoch.Load().(uint16),
 		CipherSuiteID:         uint16(s.cipherSuite.ID()),
 		MasterSecret:          s.masterSecret,
-		SequenceNumber:        s.localSequenceNumber,
+		SequenceNumber:        atomic.LoadUint64(&s.localSequenceNumber),
 		LocalRandom:           localRnd,
 		RemoteRandom:          remoteRnd,
 		SRTPProtectionProfile: uint16(s.srtpProtectionProfile),
@@ -117,7 +117,7 @@ func (s *State) deserialize(serialized serializedState) error {
 		return err
 	}
 
-	s.localSequenceNumber = serialized.SequenceNumber
+	atomic.StoreUint64(&s.localSequenceNumber, serialized.SequenceNumber)
 	s.srtpProtectionProfile = SRTPProtectionProfile(serialized.SRTPProtectionProfile)
 
 	// Set remote certificate
diff --git a/util.go b/util.go
index 7baf5c2..0613d2e 100644
--- a/util.go
+++ b/util.go
@@ -152,3 +152,18 @@ func findMatchingCipherSuite(a, b []cipherSuite) (cipherSuite, bool) {
 	}
 	return nil, false
 }
+
+func splitBytes(bytes []byte, splitLen int) [][]byte {
+	splitBytes := make([][]byte, 0)
+	numBytes := len(bytes)
+	for i := 0; i < numBytes; i += splitLen {
+		j := i + splitLen
+		if j > numBytes {
+			j = numBytes
+		}
+
+		splitBytes = append(splitBytes, bytes[i:j])
+	}
+
+	return splitBytes
+}


Test patch:
diff --git a/e2e/e2e_lossy_test.go b/e2e/e2e_lossy_test.go
index 0e92fde..a50f84b 100644
--- a/e2e/e2e_lossy_test.go
+++ b/e2e/e2e_lossy_test.go
@@ -32,6 +32,7 @@ func TestPionE2ELossy(t *testing.T) {
 		LossChanceRange int
 		DoClientAuth    bool
 		CipherSuites    []dtls.CipherSuiteID
+		MTU             int
 	}{
 		{
 			LossChanceRange: 0,
@@ -77,7 +78,23 @@ func TestPionE2ELossy(t *testing.T) {
 			LossChanceRange: 50,
 			CipherSuites:    []dtls.CipherSuiteID{dtls.TLS_ECDHE_ECDSA_WITH_AES_256_CBC_SHA},
 		},
+		{
+			LossChanceRange: 10,
+			MTU:             100,
+			DoClientAuth:    true,
+		},
+		{
+			LossChanceRange: 20,
+			MTU:             100,
+			DoClientAuth:    true,
+		},
+		{
+			LossChanceRange: 50,
+			MTU:             100,
+			DoClientAuth:    true,
+		},
 	} {
+
 		rand.Seed(time.Now().UTC().UnixNano())
 		chosenLoss := rand.Intn(9) + test.LossChanceRange
 		serverDone := make(chan error)
@@ -92,7 +109,9 @@ func TestPionE2ELossy(t *testing.T) {
 				FlightInterval:     flightInterval,
 				CipherSuites:       test.CipherSuites,
 				InsecureSkipVerify: true,
+				MTU:                test.MTU,
 			}
+
 			if test.DoClientAuth {
 				cfg.Certificate = clientCert
 				cfg.PrivateKey = clientKey
@@ -110,10 +129,11 @@ func TestPionE2ELossy(t *testing.T) {
 				Certificate:    serverCert,
 				PrivateKey:     serverKey,
 				FlightInterval: flightInterval,
+				MTU:            test.MTU,
 			}
+
 			if test.DoClientAuth {
 				cfg.ClientAuth = dtls.RequireAnyClientCert
-
 			}
 
 			if _, err := dtls.Server(br.GetConn1(), cfg); err != nil {
diff --git a/e2e/e2e_test.go b/e2e/e2e_test.go
index fdc6edd..519cc3e 100644
--- a/e2e/e2e_test.go
+++ b/e2e/e2e_test.go
@@ -232,3 +232,33 @@ func TestPionE2ESimplePSK(t *testing.T) {
 		assertE2ECommunication(cfg, cfg, serverPort, t)
 	}
 }
+
+func TestPionE2EMTUs(t *testing.T) {
+	lim := test.TimeOut(time.Second * 30)
+	defer lim.Stop()
+
+	report := test.CheckRoutines(t)
+	defer report()
+
+	serverPort := randomPort(t)
+
+	for _, mtu := range []int{
+		10000,
+		1000,
+		100,
+	} {
+		cert, key, err := dtls.GenerateSelfSigned()
+		if err != nil {
+			t.Fatal(err)
+		}
+
+		cfg := &dtls.Config{
+			Certificate:        cert,
+			PrivateKey:         key,
+			CipherSuites:       []dtls.CipherSuiteID{dtls.TLS_ECDHE_ECDSA_WITH_AES_128_GCM_SHA256},
+			InsecureSkipVerify: true,
+			MTU:                mtu,
+		}
+		assertE2ECommunication(cfg, cfg, serverPort, t)
+	}
+}