diff --git a/conn.go b/conn.go index 0816eb1..103b074 100644 --- a/conn.go +++ b/conn.go @@ -123,6 +123,23 @@ type Conn struct { flushing bool werr error + // wireMu serializes record emission, covering the seal and the socket write + // as one step. + // + // It cannot be folded into wmu. Write deliberately RELEASES wmu around each + // writeRecord so a slow socket does not block other writers, and guards + // re-entry with the flushing flag -- but that flag only excludes Write from + // Write. Close takes the freed wmu and reaches writeRecord through + // writeSized, so a close_notify could seal concurrently with a data flush. + // + // Both halves of writeRecord need the same lock, and for different reasons. + // The sequence number IS the AEAD nonce, so two sealers taking the same + // sendSeq is nonce reuse under one key -- for AES-GCM that is authentication + // key recovery, not a decrypt failure. And even with distinct numbers the + // records must reach the wire in that order, because the peer decrypts + // against its own monotonic counter. + wireMu sync.Mutex + // closeOnce guards the close_notify alert: Close may be called more than // once, and a second alert would itself be the anomaly. closeOnce sync.Once @@ -217,6 +234,9 @@ func (c *Conn) Write(b []byte) (int, error) { } func (c *Conn) writeRecord(typ byte, payload []byte, padTo int) error { + c.wireMu.Lock() + defer c.wireMu.Unlock() + inner := make([]byte, 0, len(payload)+1) inner = append(inner, payload...) inner = append(inner, typ) diff --git a/conn_race_test.go b/conn_race_test.go new file mode 100644 index 0000000..9405c48 --- /dev/null +++ b/conn_race_test.go @@ -0,0 +1,210 @@ +package twiddle + +import ( + "bytes" + "io" + "net" + "sync" + "testing" + "time" +) + +// gateConn records every record handed to the socket and holds the FIRST one +// until the test releases it. Blocking there is what forces the overlap between +// a data flush and close_notify, rather than leaving it to chance. +type gateConn struct { + net.Conn + + entered chan struct{} + release chan struct{} + + mu sync.Mutex + gated bool + records [][]byte +} + +func newGateConn() *gateConn { + return &gateConn{entered: make(chan struct{}), release: make(chan struct{})} +} + +func (c *gateConn) Write(b []byte) (int, error) { + c.mu.Lock() + first := !c.gated + c.gated = true + c.mu.Unlock() + if first { + close(c.entered) + <-c.release + } + c.mu.Lock() + c.records = append(c.records, append([]byte(nil), b...)) + c.mu.Unlock() + return len(b), nil +} + +func (c *gateConn) Close() error { return nil } + +func (c *gateConn) wire() []byte { + c.mu.Lock() + defer c.mu.Unlock() + return bytes.Join(c.records, nil) +} + +func (c *gateConn) count() int { + c.mu.Lock() + defer c.mu.Unlock() + return len(c.records) +} + +// replayConn serves a fixed byte stream to a peer Conn, so the recorded wire +// can be decrypted exactly as the far end would decrypt it. +type replayConn struct { + net.Conn + r *bytes.Reader +} + +func (c *replayConn) Read(b []byte) (int, error) { return c.r.Read(b) } +func (c *replayConn) Close() error { return nil } + +func testSession(t *testing.T) *Session { + t.Helper() + s, err := DeriveSession(make([]byte, 32), make([]byte, 32), TLS_AES_128_GCM_SHA256) + if err != nil { + t.Fatal(err) + } + return s +} + +// A close_notify concurrent with a data flush must not overtake it on the wire. +// +// Write releases wmu around writeRecord so a slow socket cannot block other +// writers; Close reaches writeRecord through writeSized, which takes that freed +// wmu. Before wireMu the two ran concurrently, and because the sequence number +// is the AEAD nonce the peer could not decrypt what came out: it counts +// monotonically, so a record that arrives before the one it was sealed after is +// authenticated against the wrong nonce. +// +// This pins the ordering half deterministically. The other half -- two sealers +// reading the same sendSeq, which is nonce reuse under one key -- is the +// unsynchronised increment, and is what -race reports on the same overlap. +func TestCloseNotifyCannotOvertakeAFlush(t *testing.T) { + sess := testSession(t) + gate := newGateConn() + w, err := NewConn(gate, sess, true, nil) + if err != nil { + t.Fatal(err) + } + + payload := []byte("application data that must arrive first") + wrote := make(chan error, 1) + go func() { + _, err := w.Write(payload) + wrote <- err + }() + + select { + case <-gate.entered: + case <-time.After(3 * time.Second): + t.Fatal("the data record never reached the socket") + } + + started := make(chan struct{}) + closed := make(chan struct{}) + go func() { + close(started) + _ = w.Close() + close(closed) + }() + <-started + + // The data record is still parked in the socket write, so close_notify must + // be waiting behind it rather than already on the wire. + // + // Proving that it did NOT happen needs a bound, and the bound has to outlast + // scheduling delay on a loaded machine: too short and a Close that never got + // to run reads as a Close that correctly waited, which passes for the wrong + // reason. started only proves the goroutine exists, so the loop also watches + // the wire itself -- without wireMu, Close's record is not blocked by the + // gate (only the first write is) and appears there directly, which is the + // symptom rather than a proxy for it. + deadline := time.Now().Add(2 * time.Second) + for time.Now().Before(deadline) { + if n := gate.count(); n != 0 { + t.Fatalf("%d records reached the wire while a flush held it", n) + } + select { + case <-closed: + t.Fatal("close_notify was emitted while a flush held the wire") + default: + } + time.Sleep(5 * time.Millisecond) + } + + close(gate.release) + if err := <-wrote; err != nil { + t.Fatal(err) + } + select { + case <-closed: + case <-time.After(3 * time.Second): + t.Fatal("Close never returned") + } + + // The real assertion: the peer decrypts the recorded wire with its own + // monotonic counter. Any reordering or nonce reuse fails to authenticate. + peer, err := NewConn(&replayConn{r: bytes.NewReader(gate.wire())}, sess, false, nil) + if err != nil { + t.Fatal(err) + } + got := make([]byte, 0, len(payload)) + buf := make([]byte, 512) + for { + n, err := peer.Read(buf) + got = append(got, buf[:n]...) + if err != nil { + if err != io.EOF { + t.Fatalf("peer could not decrypt the recorded wire: %v", err) + } + break + } + } + if !bytes.Equal(got, payload) { + t.Fatalf("peer decrypted %q, want %q", got, payload) + } + if n := gate.count(); n != 2 { + t.Fatalf("wire carries %d records, want the data record and close_notify", n) + } +} + +// The same overlap without the gate, so -race sees the unsynchronised sendSeq +// if wireMu is ever removed. +func TestConcurrentWritesAndCloseAreRaceFree(t *testing.T) { + sess := testSession(t) + server, client := net.Pipe() + defer server.Close() + go func() { + _, _ = io.Copy(io.Discard, server) + }() + + w, err := NewConn(client, sess, true, nil) + if err != nil { + t.Fatal(err) + } + + var wg sync.WaitGroup + for i := 0; i < 8; i++ { + wg.Add(1) + go func() { + defer wg.Done() + for j := 0; j < 32; j++ { + _, _ = w.Write([]byte("0123456789abcdef")) + } + }() + } + wg.Add(1) + go func() { + defer wg.Done() + _ = w.Close() + }() + wg.Wait() +}