From 65f0276ae3bb1e7225496236f589085bdc8feee3 Mon Sep 17 00:00:00 2001 From: Adam Fisk Date: Fri, 4 Sep 2026 09:31:29 +0100 Subject: [PATCH 1/2] Serialize record emission, because the sequence number is the nonce Write releases wmu around each writeRecord so a slow socket cannot block other writers, and guards re-entry with the flushing flag. That flag only excludes Write from Write. Close reaches writeRecord through writeSized, which takes the wmu Write just freed and never consults flushing, so a close_notify could seal and write concurrently with a data flush. Both halves of writeRecord needed the same lock, for different reasons. sendSeq++ is not atomic, so two sealers could take the same sequence number -- and the sequence number IS the AEAD nonce, so that is nonce reuse under one key. For AES-GCM that is authentication key recovery and a plaintext XOR leak, not a decrypt failure that shows up as a broken connection. Separately, even sealers that take distinct numbers must reach the wire in that order, because the peer decrypts against its own monotonic counter; interleaved raw.Write calls can also split a record. wireMu covers seal, increment and socket write as one step. It cannot be folded into wmu without giving up the property Write's release was there for. The two tests pin the two halves. TestCloseNotifyCannotOvertakeAFlush parks the data record inside the socket write and requires close_notify to wait behind it, then decrypts the recorded wire with a peer Conn -- which is the invariant that actually matters, since a monotonic counter cannot authenticate a reordered record. TestConcurrentWritesAndCloseAreRaceFree runs the same overlap ungated so -race reports the unsynchronised sendSeq if wireMu is ever removed. Without the fix the first fails deterministically and the second reports a data race. Found while wiring yamux over twiddle in getlantern/lantern-box#319, which is the first caller to give one Conn a dedicated write goroutine alongside a reader that closes the session. A single-writer caller cannot reach it. Co-Authored-By: Claude Opus 5 (1M context) Claude-Session: https://claude.ai/code/session_017gn6KuHUL766qQNn8m1fu1 --- conn.go | 20 +++++ conn_race_test.go | 195 ++++++++++++++++++++++++++++++++++++++++++++++ 2 files changed, 215 insertions(+) create mode 100644 conn_race_test.go 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..839fa13 --- /dev/null +++ b/conn_race_test.go @@ -0,0 +1,195 @@ +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") + } + + closed := make(chan struct{}) + go func() { + _ = w.Close() + close(closed) + }() + + // The data record is still parked in the socket write. close_notify must be + // waiting behind it, not already on the wire. + select { + case <-closed: + t.Fatal("close_notify was emitted while a flush held the wire") + case <-time.After(150 * time.Millisecond): + } + if n := gate.count(); n != 0 { + t.Fatalf("%d records reached the wire before the flush was released", n) + } + + 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() +} From d80cb2bac707ed0bbaa8f4e18e28945950b58831 Mon Sep 17 00:00:00 2001 From: Adam Fisk Date: Fri, 4 Sep 2026 09:38:14 +0100 Subject: [PATCH 2/2] address review: bound the negative assertion on something observable The ordering test proved a negative -- close_notify did NOT reach the wire -- against a 150ms window, which passes for the wrong reason if the Close goroutine simply never got scheduled inside it. A false pass there is worse than a flake: the test would go green while not exercising the bug at all. It now signals that the goroutine started, widens the window to 2s, and watches the wire rather than only the Close return. Without wireMu, Close's record is not held by the gate -- only the first write is -- so it lands in the recording directly, which is the symptom itself rather than a proxy for it. Reverting the fix now fails in 10ms on the record count instead of waiting out the window. Co-Authored-By: Claude Opus 5 (1M context) Claude-Session: https://claude.ai/code/session_017gn6KuHUL766qQNn8m1fu1 --- conn_race_test.go | 35 +++++++++++++++++++++++++---------- 1 file changed, 25 insertions(+), 10 deletions(-) diff --git a/conn_race_test.go b/conn_race_test.go index 839fa13..9405c48 100644 --- a/conn_race_test.go +++ b/conn_race_test.go @@ -108,21 +108,36 @@ func TestCloseNotifyCannotOvertakeAFlush(t *testing.T) { 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) }() - - // The data record is still parked in the socket write. close_notify must be - // waiting behind it, not already on the wire. - select { - case <-closed: - t.Fatal("close_notify was emitted while a flush held the wire") - case <-time.After(150 * time.Millisecond): - } - if n := gate.count(); n != 0 { - t.Fatalf("%d records reached the wire before the flush was released", n) + <-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)