diff --git a/lib/go/README.md b/lib/go/README.md index 9a47c53277..03d246cb01 100644 --- a/lib/go/README.md +++ b/lib/go/README.md @@ -287,3 +287,12 @@ larger `MaxFrameSize` in its own configuration, not only in the server's: conf := &thrift.TConfiguration{ MaxFrameSize: 64 * 1024 * 1024, } + +A note about the frames TFramedTransport writes +=============================================== + +`TFramedTransport.Flush` holds the frames it writes to the same limit, through +the same check. A larger frame is not written: `Flush` returns a +`TProtocolException` of type `SIZE_LIMIT` and drops the frame, where it +previously returned no error and wrote a frame `readFrame` refuses. The note +above on raising `MaxFrameSize` applies to it as well. diff --git a/lib/go/thrift/framed_transport.go b/lib/go/thrift/framed_transport.go index 9f403da55e..1a4772b78c 100644 --- a/lib/go/thrift/framed_transport.go +++ b/lib/go/thrift/framed_transport.go @@ -202,11 +202,15 @@ func (p *TFramedTransport) WriteString(s string) (n int, err error) { func (p *TFramedTransport) Flush(ctx context.Context) error { size := p.writeBuf.Len() - if uint64(size) > uint64(math.MaxUint32) { - return NewTTransportException(UNKNOWN_TRANSPORT_EXCEPTION, fmt.Sprintf("frame too large: %d bytes exceeds uint32 max", size)) - } - defer bufPool.put(&p.writeBuf) + + // readFrame refuses a frame larger than the configured maximum, and so does + // a peer holding the same configuration. THeaderTransport.Flush holds the + // frames it writes to the same limit. A refused frame is dropped with the + // buffer. + if err := checkWriteFrameSize(p.cfg, size); err != nil { + return err + } buf := p.buffer[:4] binary.BigEndian.PutUint32(buf, uint32(size)) _, err := p.transport.Write(buf) diff --git a/lib/go/thrift/framed_transport_test.go b/lib/go/thrift/framed_transport_test.go index 0aaf4b80e8..c994ae203b 100644 --- a/lib/go/thrift/framed_transport_test.go +++ b/lib/go/thrift/framed_transport_test.go @@ -168,3 +168,50 @@ func TestTFramedTransportEmptyFrames(t *testing.T) { ) } } + +// Flush refuses a frame that a TFramedTransport holding the same configuration +// refuses to read, before writing any of it, and the transport stays usable. +// It refuses it the way THeaderTransport.Flush refuses one, through the check +// the two share. +func TestTFramedTransportFlushFrameSizeLimit(t *testing.T) { + const limit = 1024 + conf := &TConfiguration{MaxFrameSize: limit} + ctx := context.Background() + + t.Run("at-limit", func(t *testing.T) { + buf := NewTMemoryBuffer() + writer := NewTFramedTransportConf(buf, conf) + payload := bytes.Repeat([]byte("x"), limit) + writer.Write(payload) + if err := writer.Flush(ctx); err != nil { + t.Fatalf("Flush refused a frame of exactly %d bytes: %v", limit, err) + } + reader := NewTFramedTransportConf(buf, conf) + read := make([]byte, limit) + if _, err := io.ReadFull(reader, read); err != nil { + t.Fatalf("reading the frame back: %v", err) + } + if !bytes.Equal(read, payload) { + t.Error("payload read back differs from the payload written") + } + }) + + t.Run("over-limit", func(t *testing.T) { + out := &flushCountingTransport{in: bytes.NewReader(nil)} + writer := NewTFramedTransportConf(out, conf) + writer.Write(bytes.Repeat([]byte("x"), limit+1)) + requireFlushSizeLimit(t, writer.Flush(ctx)) + if out.written != 0 { + t.Errorf("Flush wrote %d bytes of a frame it refused, want 0", out.written) + } + + // The refused frame is dropped, so the next one goes out on its own. + writer.Write([]byte("next")) + if err := writer.Flush(ctx); err != nil { + t.Fatalf("Flush after a refused frame: %v", err) + } + if want := 4 + len("next"); out.written != want { + t.Errorf("Flush after a refused frame wrote %d bytes, want %d", out.written, want) + } + }) +} diff --git a/lib/go/thrift/header_transport.go b/lib/go/thrift/header_transport.go index fe2aff2abb..77abcd344a 100644 --- a/lib/go/thrift/header_transport.go +++ b/lib/go/thrift/header_transport.go @@ -628,12 +628,13 @@ func (t *THeaderTransport) Write(p []byte) (int, error) { return t.writeBuffer.Write(p) } -// checkWriteFrameSize refuses a frame of size bytes that ReadFrame, holding the -// same configuration, would refuse. THeaderMaxFrameSize is below the largest -// length the frame's 32-bit length word can carry, so a frame that passes is -// written with its true length. -func (t *THeaderTransport) checkWriteFrameSize(size int) error { - if uint64(size) > uint64(THeaderMaxFrameSize) || int64(size) > int64(t.cfg.GetMaxFrameSize()) { +// checkWriteFrameSize refuses a frame of size bytes that a reader holding cfg +// would refuse. THeaderMaxFrameSize is below the largest length the frame's +// 32-bit length word can carry, so a frame that passes is written with its +// true length. TFramedTransport.Flush writes the same length word and applies +// the same check. +func checkWriteFrameSize(cfg *TConfiguration, size int) error { + if uint64(size) > uint64(THeaderMaxFrameSize) || int64(size) > int64(cfg.GetMaxFrameSize()) { return NewTProtocolExceptionWithType( SIZE_LIMIT, fmt.Errorf("frame too large: %d bytes", size), @@ -733,7 +734,7 @@ func (t *THeaderTransport) Flush(ctx context.Context) error { } // First write frame length - if err := t.checkWriteFrameSize(payload.Len()); err != nil { + if err := checkWriteFrameSize(t.cfg, payload.Len()); err != nil { return err } buf := t.buffer[:size32] @@ -747,7 +748,7 @@ func (t *THeaderTransport) Flush(ctx context.Context) error { } case clientFramedBinary, clientFramedCompact: - if err := t.checkWriteFrameSize(t.writeBuffer.Len()); err != nil { + if err := checkWriteFrameSize(t.cfg, t.writeBuffer.Len()); err != nil { return err } buf := t.buffer[:size32]