diff --git a/fakeserver/compress.go b/fakeserver/compress.go index 50bc11b..52d5a83 100644 --- a/fakeserver/compress.go +++ b/fakeserver/compress.go @@ -2,13 +2,13 @@ package fakeserver import ( "bufio" - "bytes" - "compress/flate" "errors" "io" "net" "regexp" + "slices" "strings" + "time" "github.com/go-while/nntp-testkit/internal/inflate" ) @@ -27,7 +27,12 @@ import ( // handlers return and the connection is closed, as RFC 8054 2.2.2 requires. Raw returns the // compressing conn afterwards; keep Raw() from before the call to write bytes past the // compressor (corrupt output for a client test). It fails when compression is already on. -func (c *Conn) StartCompress() error { return c.startCompress(false) } +func (c *Conn) StartCompress() error { return c.startCompress(false, CompressShape{}) } + +// StartCompressShaped is StartCompress with the output stream shaped (CompressShape): flush +// kind, splits, a final block. EmptyFlushWith206 and With206 have no effect here, since the 206 +// is already written; the handlers of the package write them with the 206. +func (c *Conn) StartCompressShaped(shape CompressShape) error { return c.startCompress(false, shape) } // StartCompressOutputOnly is a broken COMPRESS for client tests: after the 206 only what the // server writes is deflated, while the client's bytes are read as they arrive, without @@ -35,7 +40,7 @@ func (c *Conn) StartCompress() error { return c.startCompress(false) } // input. A client that compresses its commands, as RFC 8054 requires, then gets no answer // (its deflated bytes rarely hold a line end) or "500" answers to binary lines, and Lines // records those raw bytes. -func (c *Conn) StartCompressOutputOnly() error { return c.startCompress(true) } +func (c *Conn) StartCompressOutputOnly() error { return c.startCompress(true, CompressShape{}) } // Compressed reports whether StartCompress or StartCompressOutputOnly switched the connection // to compression. @@ -44,16 +49,19 @@ func (c *Conn) Compressed() bool { return ok } -func (c *Conn) startCompress(outputOnly bool) error { +func (c *Conn) startCompress(outputOnly bool, shape CompressShape) error { if c.Compressed() { return errors.New("fakeserver: compression is already active") } - d := &deflateConn{Conn: c.conn, r: c.r} - zw, err := flate.NewWriter(&d.out, flate.DefaultCompression) - if err != nil { - return err + enc := c.enc // the one that wrote the pipelined bytes of the 206, when it did + c.enc = nil + if enc == nil { + var err error + if enc, err = newEncoder(shape); err != nil { + return err + } } - d.zw = zw + d := &deflateConn{Conn: c.conn, r: c.r, enc: enc, sleep: c.sleep, deadline: c.s.deadline} if !outputOnly { pr, pw := io.Pipe() d.r, d.pr = pr, pr @@ -74,11 +82,25 @@ func (c *Conn) startCompress(outputOnly bool) error { // addresses and Close are those of the conn underneath. type deflateConn struct { net.Conn - r io.Reader - pr *io.PipeReader // the inflater's output; nil for StartCompressOutputOnly - zw *flate.Writer // writes into out - out bytes.Buffer // the deflated bytes of one Write, sent in one write - werr error // the first failed write: the stream is broken from there on + r io.Reader + pr *io.PipeReader // the inflater's output; nil for StartCompressOutputOnly + enc *encoder + sleep func(time.Duration) bool // Conn.sleep: bounded by the Deadline, false when the server closed + deadline time.Duration + werr error // the first failed write: the stream is broken from there on + wdl time.Time // the write deadline last set through this conn (zero: none) +} + +// SetWriteDeadline and SetDeadline remember the write deadline, so that send can put back the +// one writeKind set (Drip, LineDelay, Split, Endless) instead of clearing it. +func (d *deflateConn) SetWriteDeadline(t time.Time) error { + d.wdl = t + return d.Conn.SetWriteDeadline(t) +} + +func (d *deflateConn) SetDeadline(t time.Time) error { + d.wdl = t + return d.Conn.SetDeadline(t) } func (d *deflateConn) Read(p []byte) (int, error) { return d.r.Read(p) } @@ -91,29 +113,56 @@ func (d *deflateConn) Close() error { return d.Conn.Close() } -// Write deflates p, flushes the compressor (sync flush) and writes the result to the conn -// underneath in one write, as a server that flushes its output buffer, so the client can -// inflate all of p at once (compress/flate's Writer on its own writes a short input in five -// pieces). After a failed write the deflate stream is broken, so every later Write returns -// that error. +// Write deflates p, flushes the compressor as the shape says (sync flush by default) and +// writes the result to the conn underneath in one write (or the pieces of the shape's Split), +// as a server that flushes its output buffer, so the client can inflate all of p at once +// (compress/flate's Writer on its own writes a short input in five pieces). After a failed +// write the deflate stream is broken, so every later Write returns that error. func (d *deflateConn) Write(p []byte) (int, error) { if d.werr != nil { return 0, d.werr } - d.out.Reset() - if _, err := d.zw.Write(p); err != nil { // into d.out: cannot fail - return 0, err - } - if err := d.zw.Flush(); err != nil { + b, err := d.enc.write(p) + if err != nil { return 0, err } - if _, err := d.Conn.Write(d.out.Bytes()); err != nil { + if err := d.send(b); err != nil { d.werr = err return 0, err } return len(p), nil } +// send writes b to the conn underneath, cut by the shape's Split. The pauses between the +// pieces end at the server's Deadline from the call, with a write deadline set meanwhile. +func (d *deflateConn) send(b []byte) error { + if len(b) == 0 { + return nil + } + sp := d.enc.shape.Split + cuts := sp.cuts(len(b)) + if len(cuts) == 0 { + _, err := d.Conn.Write(b) + return err + } + end := time.Now().Add(d.deadline) + if sp.Gap > 0 && (d.wdl.IsZero() || end.Before(d.wdl)) { + _ = d.Conn.SetWriteDeadline(end) + defer d.Conn.SetWriteDeadline(d.wdl) // what the caller had set, not "none" + } + prev := 0 + for _, k := range append(cuts, len(b)) { + if _, err := d.Conn.Write(b[prev:k]); err != nil { + return err + } + prev = k + if k < len(b) && sp.Gap > 0 && (time.Now().After(end) || !d.sleep(sp.Gap)) { + return errors.New("fakeserver: compressed split ran past the Deadline or the server closed") + } + } + return nil +} + // algorithm is the syntax of an RFC 8054 algorithm name (5.3): 1 to 20 upper-case letters, // digits, "-" or "_"; case-sensitive. var algorithm = regexp.MustCompile(`^[A-Z0-9_-]{1,20}$`) @@ -139,5 +188,25 @@ func (c *Conn) compressAfter(cmd, status string, outputOnly bool) bool { if cmd != "COMPRESS" || (status != "206" && !strings.HasPrefix(status, "206 ")) || c.Compressed() { return true } - return c.startCompress(outputOnly) == nil + return c.startCompress(outputOnly, c.shape) == nil +} + +// pipeline206 returns the compressed bytes the shape writes in the same socket write as the +// 206 reply to a command (nil for none), and keeps the encoder for startCompress. writeKind +// calls it with the reply's status line; armed is set for the command that starts compression +// (COMPRESS, or any command of a Script step with Compress). +func (c *Conn) pipeline206(armed bool, status string) []byte { + if !armed || !c.shape.pipelined() || c.Compressed() || (status != "206" && !strings.HasPrefix(status, "206 ")) { + return nil + } + enc, err := newEncoder(c.shape) + if err != nil { + return nil + } + b, err := enc.flushData(c.shape.With206) + if err != nil { + return nil + } + c.enc = enc + return slices.Clone(b) } diff --git a/fakeserver/compressshape.go b/fakeserver/compressshape.go new file mode 100644 index 0000000..b05a00b --- /dev/null +++ b/fakeserver/compressshape.go @@ -0,0 +1,203 @@ +package fakeserver + +import ( + "bytes" + "compress/flate" + "math/bits" + "slices" + "sort" +) + +// CompressFlush is how the compressed output stream is flushed at the end of every write. +type CompressFlush int + +const ( + // FlushSync ends every write with a sync flush (zlib's Z_SYNC_FLUSH): an empty stored + // block, the marker 00 00 FF FF on a byte boundary. The default. + FlushSync CompressFlush = iota + // FlushPartial ends every write with zlib's Z_PARTIAL_FLUSH, as INN's nnrpd does: an empty + // fixed-Huffman block (10 bits), of which only the whole bytes are sent. The last 0 to 7 + // bits stay in the compressor until the next write; the data before them is complete. + // The data before them is complete: the held bits are only the tail of the empty block's + // end-of-block code, which is zeros. There is no 00 00 FF FF marker in the stream. The data are literals in fixed-Huffman + // blocks, one per write (no matches), so the output is larger than FlushSync's. A client + // that waits for a stored block, or for the end of the block it is in, hangs on this + // stream, as INN's clients must not. + FlushPartial + // FlushFinish ends the deflate stream (a final block) with the first write after the 206: + // that write is deflated and the final empty block follows in the same socket write. Later + // writes are plain bytes on the socket, after the end of the stream (the kit still inflates + // what the client sends). CompressShape.AfterEnd adds bytes behind the final block. + FlushFinish +) + +// CompressShape shapes what the fake server sends on a COMPRESS DEFLATE connection (RFC 8054), +// for client tests: StoreOptions.CompressShape, PeerOptions.CompressShape, Step.CompressShape, +// Conn.StartCompressShaped. The zero value is the default: every write sync-flushed and sent +// in one socket write. +type CompressShape struct { + // Flush is the flush at the end of each write, see CompressFlush. + Flush CompressFlush + + // Split cuts the compressed bytes of each write (reply, Faults piece) into separate socket + // writes: At holds offsets from the start of the write, a negative offset counts from its + // end (with FlushSync -1 .. -3 cut inside the sync marker 00 00 FF FF, -4 before it, and + // with FlushPartial -1 cuts the empty block's last byte), Size cuts a piece every Size + // bytes when At is empty, Gap pauses after each piece but the last, bounded by the + // server's Deadline like Faults.Split. A client that reads between the pieces sees + // compressed bytes that end mid-block or mid-marker. + Split Split + + // EmptyFlushWith206 writes an empty flush block (the marker of Flush) right after the CRLF + // of the "206" in the same socket write as the 206 line, as a server that flushes its + // compressor when it switches it on. A client must keep those bytes for its inflater. The + // 206 is written by the kit's handlers (Store, Peer with Compress or CompressOutputOnly, + // Script step with Compress); a Handler that writes its own 206 cannot use this. Faults on + // the COMPRESS reply (DropAfterBytes, Split offsets) count these bytes too. + EmptyFlushWith206 bool + // With206 is deflated and flushed like a reply and written in the same socket write as the + // 206 (it ends with its own flush, so EmptyFlushWith206 adds nothing then): compressed + // payload pipelined behind the 206, for a client that reads past the 206 line. + With206 []byte + + // AfterEnd is written in the same socket write as the final block, with FlushFinish: + // trailing bytes after the end of the stream (junk, or plain text). + AfterEnd []byte +} + +func (s CompressShape) pipelined() bool { return s.EmptyFlushWith206 || len(s.With206) > 0 } + +// cuts returns the sorted offsets in (0, n) where a write of n bytes is cut. +func (s Split) cuts(n int) []int { + var out []int + for _, a := range s.At { + if a < 0 { + a += n + } + if a > 0 && a < n { + out = append(out, a) + } + } + sort.Ints(out) + out = slices.Compact(out) + if len(s.At) == 0 && s.Size > 0 { + for k := s.Size; k < n; k += s.Size { + out = append(out, k) + } + } + return out +} + +// encoder deflates the output of a compressed connection as its shape says. Not safe for +// concurrent use (one connection goroutine writes). +type encoder struct { + shape CompressShape + zw *flate.Writer // FlushSync and FlushFinish; writes into out + out bytes.Buffer + bw bitWriter // FlushPartial + ended bool // FlushFinish: the stream is over +} + +func newEncoder(s CompressShape) (*encoder, error) { + e := &encoder{shape: s} + if s.Flush != FlushPartial { + zw, err := flate.NewWriter(&e.out, flate.DefaultCompression) + if err != nil { + return nil, err + } + e.zw = zw + } + return e, nil +} + +// flushData deflates p and flushes (sync, or partial for FlushPartial; FlushFinish counts as +// sync here). An empty p is an empty flush block. The result is valid until the next call. +func (e *encoder) flushData(p []byte) ([]byte, error) { + if e.shape.Flush == FlushPartial { + return e.partial(p), nil + } + e.out.Reset() + if _, err := e.zw.Write(p); err != nil { // into e.out: cannot fail + return nil, err + } + if err := e.zw.Flush(); err != nil { + return nil, err + } + return e.out.Bytes(), nil +} + +// write returns the bytes to send for a write of p: deflated and flushed, or with FlushFinish +// the end of the stream (once; then p itself). +func (e *encoder) write(p []byte) ([]byte, error) { + if e.ended { + return p, nil + } + if e.shape.Flush != FlushFinish { + return e.flushData(p) + } + e.out.Reset() + if _, err := e.zw.Write(p); err != nil { + return nil, err + } + if err := e.zw.Close(); err != nil { + return nil, err + } + e.ended = true + return append(e.out.Bytes(), e.shape.AfterEnd...), nil +} + +// partial is zlib's Z_PARTIAL_FLUSH (deflate.c: FLUSH_BLOCK, then _tr_align): the data as one +// fixed-Huffman block of literals when there are any, then an empty fixed-Huffman block (3 +// header bits, the 7-bit end-of-block code), and only the whole bytes are sent. Since zlib +// 1.2.5.2 that is all: older versions added a second empty block when fewer than 9 bits of +// the last block's end code had been sent. +func (e *encoder) partial(p []byte) []byte { + w := &e.bw + if len(p) > 0 { + w.block(p) + } + w.block(nil) + return w.take() +} + +// bitWriter writes DEFLATE bits, the first one in the lowest bit of a byte (RFC 1951 3.1.1). +type bitWriter struct { + out []byte + acc uint32 + n uint // bits in acc, fewer than 8 after put +} + +func (w *bitWriter) put(v uint32, n uint) { + w.acc |= v << w.n + w.n += n + for w.n >= 8 { + w.out = append(w.out, byte(w.acc)) + w.acc >>= 8 + w.n -= 8 + } +} + +// code writes an n-bit Huffman code, which goes most significant bit first. +func (w *bitWriter) code(v uint32, n uint) { w.put(bits.Reverse32(v)>>(32-n), n) } + +// block writes a non-final fixed-Huffman block (BFINAL 0, BTYPE 01) of the literals p and the +// end-of-block code (RFC 1951 3.2.6: literals 0-143 are 8 bits from 00110000, 144-255 are 9 +// bits from 110010000, the end of block is 7 zero bits). +func (w *bitWriter) block(p []byte) { + w.put(2, 3) // BFINAL 0 in the lowest bit, BTYPE 01 above it + for _, b := range p { + if b < 144 { + w.code(0x30+uint32(b), 8) + } else { + w.code(0x190+uint32(b)-144, 9) + } + } + w.code(0, 7) +} + +// take returns the whole bytes written so far; the last bits stay. +func (w *bitWriter) take() []byte { + b := w.out + w.out = nil + return b +} diff --git a/fakeserver/compressshape_test.go b/fakeserver/compressshape_test.go new file mode 100644 index 0000000..3606361 --- /dev/null +++ b/fakeserver/compressshape_test.go @@ -0,0 +1,510 @@ +package fakeserver + +import ( + "bufio" + "bytes" + "compress/flate" + "io" + "net" + "strings" + "sync" + "testing" + "time" + + "github.com/go-while/nntp-testkit/internal/inflate" +) + +// Tests of CompressShape: the compressed output stream as the client's raw socket sees it. + +const ( + okLine = "206 Compression active\r\n" + marker = "\x00\x00\xff\xff" // the sync flush marker +) + +// recConn records every write to the socket (the pieces of the 206 line and of the compressed +// stream), so a test can tell what was one write without relying on read boundaries. +type recConn struct { + net.Conn + mu sync.Mutex + w [][]byte +} + +func (r *recConn) Write(p []byte) (int, error) { + // recorded before the write, so a client that has read the bytes finds them in writes() + r.mu.Lock() + r.w = append(r.w, bytes.Clone(p)) + i := len(r.w) - 1 + r.mu.Unlock() + n, err := r.Conn.Write(p) + if n < len(p) { + r.mu.Lock() + r.w[i] = r.w[i][:n] + r.mu.Unlock() + } + return n, err +} + +func (r *recConn) writes() [][]byte { + r.mu.Lock() + defer r.mu.Unlock() + return append([][]byte(nil), r.w...) +} + +// shapeServer is a Store with Compress and shape; DATE answers dateReply. The recConn sees +// the writes from the greeting on that follow the handler's start, so w[0] is the 206 write. +func shapeServer(t *testing.T, shape CompressShape) (*Server, *recConn) { + t.Helper() + rec := &recConn{} + h := StoreHandler(testStore(), StoreOptions{Compress: true, CompressShape: shape}) + s := New(t, Options{Deadline: 10 * time.Second, + Faults: Faults{Override: map[string][]string{"DATE": {"111 12345678901234"}}}, + Handler: func(c *Conn) { + rec.Conn = c.conn + c.conn = rec + h(c) + }}) + return s, rec +} + +// zClient is a raw client of a compressed connection: its commands are deflated and sync +// flushed in one write. +type zClient struct { + t *testing.T + c net.Conn + br *bufio.Reader // reads c behind the 206 line; use it once the test reads the compressed stream + zw *flate.Writer + b bytes.Buffer +} + +// zDial connects, sends COMPRESS DEFLATE and reads the 206 line byte by byte, so that bytes +// behind it stay in the socket. +func zDial(t *testing.T, s *Server) *zClient { + t.Helper() + c := bytesDial(t, s) + rawSend(t, c, "COMPRESS DEFLATE\r\n") + var line []byte + one := make([]byte, 1) + for !bytes.HasSuffix(line, []byte("\r\n")) { + if _, err := io.ReadFull(c, one); err != nil { + t.Fatalf("reading the 206: %v (got %q)", err, line) + } + line = append(line, one[0]) + } + if string(line) != okLine { + t.Fatalf("COMPRESS: %q", line) + } + z := &zClient{t: t, c: c, br: bufio.NewReader(c)} + z.zw, _ = flate.NewWriter(&z.b, flate.DefaultCompression) + return z +} + +func (z *zClient) send(cmd string) { + z.t.Helper() + z.b.Reset() + _, _ = z.zw.Write([]byte(cmd + "\r\n")) + _ = z.zw.Flush() + if _, err := z.c.Write(z.b.Bytes()); err != nil { + z.t.Fatal(err) + } +} + +// refStream deflates replies the way the default shape does (one default-level flate writer, +// sync flush after each reply), to know the bytes a test expects on the wire. +type refStream struct { + b bytes.Buffer + zw *flate.Writer +} + +func newRef() *refStream { + r := &refStream{} + r.zw, _ = flate.NewWriter(&r.b, flate.DefaultCompression) + return r +} + +func (r *refStream) flush(p string) []byte { + r.b.Reset() + _, _ = r.zw.Write([]byte(p)) + _ = r.zw.Flush() + return bytes.Clone(r.b.Bytes()) +} + +// plain reads the compressed stream of c with the kit's inflater, which passes on the bytes as +// soon as they can be decoded (so it also reads partial-flush streams), and returns a reader +// of the plaintext. +func plain(t *testing.T, z *zClient) *bufio.Reader { + t.Helper() + pr, pw := io.Pipe() + t.Cleanup(func() { _ = pr.Close() }) + go func() { + _ = pw.CloseWithError(inflate.Run(z.br, func(b []byte) error { + _, err := pw.Write(b) + return err + })) + }() + return bufio.NewReader(pr) +} + +func wantPlainLine(t *testing.T, br *bufio.Reader, c net.Conn, want string) { + t.Helper() + _ = c.SetReadDeadline(time.Now().Add(10 * time.Second)) + line, err := br.ReadString('\n') + if err != nil || line != want { + t.Fatalf("reply %q, %v; want %q", line, err, want) + } +} + +// The default shape: one socket write per reply, a sync flush at its end. +func TestCompressShapeDefault(t *testing.T) { + s, rec := shapeServer(t, CompressShape{}) + z := zDial(t, s) + br := plain(t, z) + z.send("DATE") + wantPlainLine(t, br, z.c, dateReply) + z.send("DATE") + wantPlainLine(t, br, z.c, dateReply) + w := rec.writes() + ref := newRef() + if len(w) != 3 || string(w[0]) != okLine || !bytes.Equal(w[1], ref.flush(dateReply)) || !bytes.Equal(w[2], ref.flush(dateReply)) { + t.Fatalf("writes %q", w) + } + if !bytes.HasSuffix(w[1], []byte(marker)) { + t.Errorf("no sync marker at the end of %x", w[1]) + } +} + +func TestCompressShapeSplitPieces(t *testing.T) { + const gap = 120 * time.Millisecond + ref := newRef() + first, second := ref.flush(dateReply), ref.flush(dateReply) + n := len(first) + for _, tc := range []struct { + name string + split Split + cuts []int + }{ + {"offsets from start and end", Split{At: []int{2, -3}, Gap: gap}, []int{2, n - 3}}, + {"inside the marker", Split{At: []int{-1, -2, -3}, Gap: gap}, []int{n - 3, n - 2, n - 1}}, + {"size", Split{Size: 7, Gap: gap}, nil}, + } { + t.Run(tc.name, func(t *testing.T) { + cuts := tc.cuts + if cuts == nil { + for k := tc.split.Size; k < n; k += tc.split.Size { + cuts = append(cuts, k) + } + } + var pieces []string + prev := 0 + for _, k := range append(append([]int{}, cuts...), n) { + pieces = append(pieces, string(first[prev:k])) + prev = k + } + s, rec := shapeServer(t, CompressShape{Split: tc.split}) + z := zDial(t, s) + start := time.Now() + z.send("DATE") + readPieces(t, z.c, start, pieces, gaps(len(pieces), gap)) + // the pieces are exactly the socket writes of the server + w := rec.writes() + if len(w) != 1+len(pieces) { + t.Fatalf("%d writes, want the 206 and %d pieces", len(w), len(pieces)) + } + for i, p := range pieces { + if string(w[1+i]) != p { + t.Errorf("write %d = %x, want %x", i, w[1+i], p) + } + } + // the stream goes on: the next reply is deflated after the first as the reference does + start = time.Now() + z.send("DATE") + readPieces(t, z.c, start, []string{string(second)}, []time.Duration{0}) + }) + } +} + +// A client inflater (compress/flate's reader) must cope with the stream arriving byte by byte. +func TestCompressShapeSplitBytewise(t *testing.T) { + s, _ := shapeServer(t, CompressShape{Split: Split{Size: 1}}) + z := zDial(t, s) + zr := flate.NewReader(z.br) + for range 3 { + z.send("DATE") + buf := make([]byte, len(dateReply)) + _ = z.c.SetReadDeadline(time.Now().Add(10 * time.Second)) + if _, err := io.ReadFull(zr, buf); err != nil || string(buf) != dateReply { + t.Fatalf("reply %q, %v", buf, err) + } + } +} + +func TestCompressShapeEmptyFlushWith206(t *testing.T) { + for _, tc := range []struct { + name string + shape CompressShape + empty string + }{ + {"sync", CompressShape{EmptyFlushWith206: true}, "\x00" + marker}, + {"partial", CompressShape{Flush: FlushPartial, EmptyFlushWith206: true}, "\x02"}, + } { + t.Run(tc.name, func(t *testing.T) { + s, rec := shapeServer(t, tc.shape) + z := zDial(t, s) + // the bytes behind the 206 line are already here: nothing was sent after the line + _ = z.c.SetReadDeadline(time.Now().Add(5 * time.Second)) + if got, err := z.br.Peek(len(tc.empty)); err != nil || string(got) != tc.empty { + t.Fatalf("bytes after the 206: %x, %v; want %x", got, err, tc.empty) + } + if w := rec.writes(); len(w) != 1 || string(w[0]) != okLine+tc.empty { + t.Fatalf("the 206 and the empty flush are not one write: %q", w) + } + // the empty block is part of the stream: replies follow it + br := plain(t, z) + z.send("DATE") + wantPlainLine(t, br, z.c, dateReply) + }) + } +} + +// Without the option nothing follows the 206 line. +func TestCompressShapeNoEmptyFlush(t *testing.T) { + s, rec := shapeServer(t, CompressShape{}) + z := zDial(t, s) + if !readQuiet(t, z.c, 200*time.Millisecond) { + t.Fatal("bytes after the 206") + } + if w := rec.writes(); len(w) != 1 || string(w[0]) != okLine { + t.Fatalf("writes %q", w) + } +} + +// With206 pipelines a deflated payload in the write of the 206 line (the client reads it +// before it has sent a command). +func TestCompressShapeWith206(t *testing.T) { + s, rec := shapeServer(t, CompressShape{With206: []byte("hello\r\n")}) + z := zDial(t, s) + br := plain(t, z) + wantPlainLine(t, br, z.c, "hello\r\n") + z.send("DATE") + wantPlainLine(t, br, z.c, dateReply) + w := rec.writes() + if string(w[0][:len(okLine)]) != okLine || len(w[0]) == len(okLine) { + t.Fatalf("first write %q", w[0]) + } +} + +// Partial flush as zlib's: replies decode, promptly, and the stream has no sync marker. +func TestCompressShapePartialFlush(t *testing.T) { + s, rec := shapeServer(t, CompressShape{Flush: FlushPartial}) + z := zDial(t, s) + br := plain(t, z) + for range 3 { + z.send("DATE") + wantPlainLine(t, br, z.c, dateReply) // arrives although the last bits of the flush block are not sent + } + z.send("HELP") + wantPlainLine(t, br, z.c, "100 Help text follows\r\n") + var stream []byte + for _, p := range rec.writes()[1:] { + stream = append(stream, p...) + } + if bytes.Contains(stream, []byte(marker)) { + t.Errorf("sync marker in a partial-flush stream: %x", stream) + } +} + +// The partial-flush encoder against compress/flate's reader, which has no partial flush of its +// own but decodes the stream once its last block is completed. +func TestPartialEncoderDecodes(t *testing.T) { + e, err := newEncoder(CompressShape{Flush: FlushPartial}) + if err != nil { + t.Fatal(err) + } + var plainText, stream []byte + all := make([]byte, 256) // every byte value: the 8-bit and the 9-bit codes + for i := range all { + all[i] = byte(i) + } + for _, p := range [][]byte{[]byte("211 6 1 6 compress.test\r\n"), nil, all, []byte("."), {0xff, 0xff, 0x00}, bytes.Repeat([]byte("xyz\xf0"), 5000)} { + b, _ := e.flushData(p) + stream = append(stream, b...) + plainText = append(plainText, p...) + } + if bytes.Contains(stream, []byte(marker)) { + t.Error("sync marker in the stream") + } + // complete the stream: the held bits are the zero tail of the empty block's end code; a + // final stored block header (BFINAL 1, BTYPE 00) follows them at once, then padding to a + // byte and the empty block's LEN and NLEN + unfinished := len(stream) + e.bw.put(1, 3) + if e.bw.n > 0 { + e.bw.put(0, 8-e.bw.n) + } + stream = append(append(stream, e.bw.take()...), 0x00, 0x00, 0xff, 0xff) + got, err := io.ReadAll(flate.NewReader(bytes.NewReader(stream))) + if err != nil || !bytes.Equal(got, plainText) { + t.Fatalf("decoded %d bytes, %v; want %d", len(got), err, len(plainText)) + } + // the kit's inflater decodes the unfinished stream as it is + var out []byte + _ = inflate.Run(bufio.NewReader(bytes.NewReader(stream[:unfinished])), func(b []byte) error { out = append(out, b...); return nil }) + if !bytes.Equal(out, plainText) { + t.Errorf("kit inflater: %d bytes, want %d", len(out), len(plainText)) + } +} + +// The first sync flush of an empty stream and the first partial flush have the bytes the +// tests above expect (a change of compress/flate would show here first). +func TestEmptyFlushBytes(t *testing.T) { + e, _ := newEncoder(CompressShape{}) + if b, _ := e.flushData(nil); string(b) != "\x00"+marker { + t.Errorf("sync empty flush %x", b) + } + e, _ = newEncoder(CompressShape{Flush: FlushPartial}) + if b, _ := e.flushData(nil); string(b) != "\x02" { + t.Errorf("partial empty flush %x", b) + } +} + +func TestCompressShapeFinish(t *testing.T) { + for _, after := range []string{"", "junk\r\nmore"} { + t.Run("after="+after, func(t *testing.T) { + s, rec := shapeServer(t, CompressShape{Flush: FlushFinish, AfterEnd: []byte(after)}) + z := zDial(t, s) + z.send("DATE") + br := z.br // a ByteReader: flate reads no byte past the final block + _ = z.c.SetReadDeadline(time.Now().Add(10 * time.Second)) + got, err := io.ReadAll(flate.NewReader(br)) + if err != nil || string(got) != dateReply { + t.Fatalf("stream %q, %v", got, err) + } + rest := make([]byte, len(after)) + if _, err := io.ReadFull(br, rest); err != nil || string(rest) != after { + t.Fatalf("after the end %q, %v", rest, err) + } + if w := rec.writes(); len(w) != 2 || !bytes.HasSuffix(w[1], []byte(after)) { + t.Fatalf("the final block and the trailing bytes are not one write: %q", w) + } + // the client's input is still inflated, the server's output is plain from here on + z.send("DATE") + line, err := br.ReadString('\n') + if err != nil || line != dateReply { + t.Fatalf("plain reply %q, %v", line, err) + } + }) + } +} + +// The shape also applies to a Peer and to a Script step. +func TestCompressShapePeerAndScript(t *testing.T) { + shape := CompressShape{Flush: FlushPartial, EmptyFlushWith206: true, Split: Split{Size: 2}} + t.Run("peer", func(t *testing.T) { + s := New(t, Options{Handler: PeerHandler(&PeerOptions{Compress: true, CompressShape: shape})}) + z := zDial(t, s) + if _, err := z.br.Peek(1); err != nil { // the empty block + t.Fatal(err) + } + br := plain(t, z) + z.send("MODE STREAM") + line, err := br.ReadString('\n') + if err != nil || !strings.HasPrefix(line, "203 ") { + t.Fatalf("%q, %v", line, err) + } + }) + t.Run("script", func(t *testing.T) { + s := New(t, Options{Script: []Step{ + {Expect: "COMPRESS DEFLATE", Reply: []string{"206 Compression active"}, Compress: true, CompressShape: shape}, + {Expect: "DATE", Reply: []string{"111 20260102030405"}}, + }}) + z := zDial(t, s) + if _, err := z.br.Peek(1); err != nil { + t.Fatal(err) + } + br := plain(t, z) + z.send("DATE") + wantPlainLine(t, br, z.c, "111 20260102030405\r\n") + }) +} + +// StartCompressShaped from a Handler that writes its own 206. +func TestStartCompressShaped(t *testing.T) { + s := New(t, Options{Handler: func(c *Conn) { + if _, err := c.ReadLine(); err != nil { + return + } + _ = c.WriteLines("206 Compression active") + if c.StartCompressShaped(CompressShape{Flush: FlushPartial}) != nil || c.StartCompressShaped(CompressShape{}) == nil { + return + } + _ = c.WriteLines("111 20260102030405") + c.drain() + }}) + z := zDial(t, s) + br := plain(t, z) + wantPlainLine(t, br, z.c, "111 20260102030405\r\n") +} + +// The pipelined tail is armed only when compression starts: a Script step without Compress and +// a Peer without Compress (a forced 206) stay plain. +func TestCompressShapeNotArmedWithoutCompress(t *testing.T) { + shape := CompressShape{EmptyFlushWith206: true, With206: []byte("x")} + check := func(t *testing.T, s *Server, cmd, wantPrefix string) { + t.Helper() + c := bytesDial(t, s) + rawSend(t, c, cmd+"\r\n") + br := bufio.NewReader(c) + line, err := br.ReadString('\n') + if err != nil || !strings.HasPrefix(line, "206 ") { + t.Fatalf("%q, %v", line, err) + } + if br.Buffered() != 0 || !readQuiet(t, c, 200*time.Millisecond) { + t.Fatal("bytes after the 206 of a connection that stays plain") + } + rawSend(t, c, "MODE STREAM\r\n") + _ = c.SetReadDeadline(time.Now().Add(10 * time.Second)) + if line, err = br.ReadString('\n'); err != nil || !strings.HasPrefix(line, wantPrefix) { + t.Fatalf("plain command after the 206: %q, %v", line, err) + } + } + t.Run("script", func(t *testing.T) { + s := New(t, Options{Script: []Step{ + {Expect: "COMPRESS DEFLATE", Reply: []string{"206 ok"}, CompressShape: shape}, + {Expect: "MODE STREAM", Reply: []string{"203 ok"}}, + }}) + check(t, s, "COMPRESS DEFLATE", "203 ") + }) + t.Run("peer", func(t *testing.T) { + s := New(t, Options{Handler: PeerHandler(&PeerOptions{CompressShape: shape}), + Faults: Faults{Override: map[string][]string{"COMPRESS": {"206 ok"}}}}) + check(t, s, "COMPRESS DEFLATE", "203 ") + }) +} + +// A write deadline that the caller set (writeKind does for its pauses) survives a Write with a +// Split gap; without one, the deadline set for the pauses is cleared again. +type dlConn struct { + net.Conn + dl []time.Time +} + +func (c *dlConn) Write(p []byte) (int, error) { return len(p), nil } +func (c *dlConn) SetWriteDeadline(t time.Time) error { c.dl = append(c.dl, t); return nil } + +func TestCompressSplitKeepsCallerDeadline(t *testing.T) { + for _, set := range []bool{true, false} { + sc := &dlConn{} + enc, _ := newEncoder(CompressShape{Split: Split{Size: 3, Gap: time.Millisecond}}) + d := &deflateConn{Conn: sc, enc: enc, sleep: func(time.Duration) bool { return true }, deadline: time.Hour} + var want time.Time + if set { + want = time.Now().Add(time.Minute) + _ = d.SetWriteDeadline(want) + } + if _, err := d.Write([]byte("hello\r\n")); err != nil { + t.Fatal(err) + } + if last := sc.dl[len(sc.dl)-1]; !last.Equal(want) { + t.Errorf("caller deadline set=%v: after Write the deadline is %v, want %v", set, last, want) + } + } +} diff --git a/fakeserver/conn.go b/fakeserver/conn.go index d99de45..97b3c43 100644 --- a/fakeserver/conn.go +++ b/fakeserver/conn.go @@ -36,6 +36,11 @@ type Conn struct { cmd string // command of the last client line read with ReadLine (the key of the byte-level faults) faulted bool // a fault closes the connection now (Faults.Abort decides how) + + shape CompressShape // the shape of the stream compressAfter starts (the handler's CompressShape) + shapeOn bool // Store/Peer: COMPRESS is offered, so a 206 to it starts compression + shapeStep bool // Script: the step being served starts compression, whatever its command + enc *encoder // made when the 206 carried pipelined bytes; startCompress takes it over } // ID returns the connection's number in accept order, starting at 1 (connections rejected by @@ -318,6 +323,7 @@ func (c *Conn) writeKind(line string, kind replyKind, lines []string) bool { if len(lines) > 0 { c.last = lines[0] } + tail := c.pipeline206((cmd == "COMPRESS" && c.shapeOn) || c.shapeStep, c.last) // empty flush etc. in the 206's write endless := normal && f.Endless[cmd] if endless && len(lines) > 0 && lines[len(lines)-1] == "." { lines = lines[:len(lines)-1] @@ -343,11 +349,15 @@ func (c *Conn) writeKind(line string, kind replyKind, lines []string) bool { if i > 0 && (time.Now().After(end) || !c.sleep(pause)) { return c.cutIf(w) } - if !w.write(joinCRLF([]string{l})) { + b := joinCRLF([]string{l}) + if i == len(lines)-1 { + b = append(b, tail...) + } + if !w.write(b) { return c.cutIf(w) } } - } else if len(lines) > 0 && !w.write(joinCRLF(lines)) { + } else if len(lines) > 0 && !w.write(append(joinCRLF(lines), tail...)) { return c.cutIf(w) } if endless { diff --git a/fakeserver/doc.go b/fakeserver/doc.go index c26d5f6..8921654 100644 --- a/fakeserver/doc.go +++ b/fakeserver/doc.go @@ -25,11 +25,12 @@ // // NoGreeting; then Handler when set, else Store when set, else Script (with NoGreeting and // // no Handler the Script/Store is not run: lines are only recorded). // -// type Step struct { Expect string; ReadDot bool; Delay time.Duration; Reply []string; Hang, CloseAfter, Compress bool } +// type Step struct { Expect string; ReadDot bool; Delay time.Duration; Reply []string; Hang, CloseAfter, Compress bool; CompressShape CompressShape } // // Semantics as go-pugleaf's cliFakeStep: Expect is an anchored regexp ("" = any line); a // // non-matching line gets "500 fakeserver: unexpected " and is recorded in Mismatches(); // // after the last step the connection keeps reading and recording until EOF. Compress -// // switches the connection to COMPRESS DEFLATE after the step's reply (Conn.StartCompress). +// // switches the connection to COMPRESS DEFLATE after the step's reply (Conn.StartCompress), +// // CompressShape shapes its output stream. // // type Server struct { ... } // func New(tb testing.TB, o Options) *Server // listens, serves, tb.Cleanup(Close); fails tb on listen error @@ -59,6 +60,7 @@ // func (c *Conn) ReadDotBlockRaw() ([]string, []byte, error) // the same, also the exact bytes of the block (not the "." line) // func (c *Conn) Raw() net.Conn // the TLS conn when TLS, else the TCP conn; after StartCompress the compressing conn on top // func (c *Conn) StartCompress() error // RFC 8054 COMPRESS DEFLATE from now on, both directions; call it after writing the 206; error when already on +// func (c *Conn) StartCompressShaped(s CompressShape) error // StartCompress with the output stream shaped, see CompressShape below // func (c *Conn) StartCompressOutputOnly() error // broken COMPRESS: deflate only the server's output, read the client's bytes uncompressed // func (c *Conn) Compressed() bool // func (c *Conn) Server() *Server @@ -93,6 +95,7 @@ // OnArticleRaw func(msgID string, raw []byte) // right after OnArticle: the article's exact stuffed bytes, no terminator (article.ParseDotBlock) // Compress bool // offer COMPRESS DEFLATE as StoreOptions.Compress; AUTHINFO answers 502 once active // CompressOutputOnly bool // offer it, but after the 206 compress only the output (Conn.StartCompressOutputOnly) +// CompressShape CompressShape // shape the compressed output stream, see below // HoldGreeting <-chan struct{} // greeting waits for close/receive; needs Options.NoGreeting (NewPeer sets it); the hold ends at Deadline (60 s default) and then the connection is dropped without a greeting, as on Close; the conn counts as open // GreetingDelay time.Duration // sleep before the greeting (after HoldGreeting), capped at Deadline; needs Options.NoGreeting (NewPeer sets it) // BeforeReply func(c *Conn, cmd, line, status string) // before every reply (IHAVE: 335 and the final one); may block, e.g. c.Wait(ch) @@ -167,6 +170,7 @@ // OverviewFields func(conn int, group string, fields []string) []string // rewrite the fields of every OVER line (overview out of step with the articles) // Compress bool // offer COMPRESS DEFLATE (RFC 8054), see below // CompressOutputOnly bool // offer it broken: after the 206 only the output is compressed, the client's input is read uncompressed +// CompressShape CompressShape // shape the compressed output stream, see below // CompressBeforeAuth bool // with User/Pass: list and take COMPRESS before AUTHINFO (INN's nnrpd); other commands still 480, AUTHINFO 502 once compressed // StartTLS *StartTLSOptions // offer STARTTLS, see below // } @@ -269,4 +273,19 @@ // client tests: the 206 and everything after it is compressed, but the client's compressed // commands are read as they are, so a client that follows the RFC gets no answer or "500" // answers to binary lines. +// +// CompressShape (StoreOptions, PeerOptions, Step, Conn.StartCompressShaped) shapes the output +// stream for client tests; the zero value is the behaviour above: +// +// type CompressShape struct { +// Flush CompressFlush // FlushSync (default); FlushPartial: zlib's Z_PARTIAL_FLUSH like INN's nnrpd (an empty fixed-Huffman block, no 00 00 FF FF marker, the last 0-7 bits unsent); FlushFinish: the first write after the 206 ends the stream, later writes are plain +// Split Split // cut the compressed bytes of each write into socket writes: At (negative = from the end, -1..-3 inside the sync marker), Size, Gap (bounded by Deadline) +// EmptyFlushWith206 bool // the 206 line and an empty flush block in one socket write (Store, Peer, Script; not a Handler's own 206) +// With206 []byte // deflated and flushed into the 206's socket write (compressed payload pipelined behind the 206) +// AfterEnd []byte // with FlushFinish: bytes after the final block, in its socket write +// } +// +// FlushPartial writes literals in fixed-Huffman blocks (no matches) with a small bit writer, +// since compress/flate cannot do a partial flush; compress/flate's reader cannot read such a +// stream before the end either (it holds data until a stored block), the kit's inflater can. package fakeserver diff --git a/fakeserver/peer.go b/fakeserver/peer.go index 7c68a16..fbe3a40 100644 --- a/fakeserver/peer.go +++ b/fakeserver/peer.go @@ -30,6 +30,9 @@ type PeerOptions struct { // peer's output and reads the client's input uncompressed (Conn.StartCompressOutputOnly): // a broken peer, for client tests. CompressOutputOnly bool + // CompressShape shapes the compressed output stream after the 206 (flush kind, split + // writes, an empty flush with the 206, a final block); the zero value is the default. + CompressShape CompressShape // Reject are the articles the peer refuses once it has read them, as a server whose // filter refuses the content does: CHECK still answers 238 (the id alone is wanted), @@ -130,6 +133,7 @@ func PeerHandler(o *PeerOptions) func(*Conn) { dropReply = "400 Service temporarily unavailable" } return func(c *Conn) { + c.shape, c.shapeOn = o.CompressShape, offerCompress if !o.greet(c) { return } diff --git a/fakeserver/server.go b/fakeserver/server.go index c8a1574..913d2f9 100644 --- a/fakeserver/server.go +++ b/fakeserver/server.go @@ -68,6 +68,9 @@ type Step struct { Hang bool // after the match (and ReadDot, Delay, Reply), never reply again; held until Close CloseAfter bool // close the connection after the reply Compress bool // after the reply (e.g. "206 Compression active"), compress both directions (Conn.StartCompress) + // CompressShape shapes the compressed output stream of a Compress step (see CompressShape); + // EmptyFlushWith206 and With206 go with the step's reply when that is a 206. + CompressShape CompressShape } // StoreOptions configures the store handler (Options.StoreOpts, StoreHandler). @@ -137,6 +140,13 @@ type StoreOptions struct { // inflating it (Conn.StartCompressOutputOnly): a broken server, for client tests. CompressOutputOnly bool + // CompressShape shapes the compressed output stream after the 206: the flush (sync, zlib's + // partial flush like INN's nnrpd, or a final block with trailing bytes), the compressed + // bytes cut into several socket writes, an empty flush in the packet of the 206 line. The + // zero value is the default: every write sync-flushed and sent in one socket write. It + // applies to Compress and CompressOutputOnly (see CompressShape). + CompressShape CompressShape + // CompressBeforeAuth makes a server with User or Pass set list "COMPRESS DEFLATE" before // authentication and take COMPRESS then, as INN's nnrpd does (COMPRESS is not a reading // command): the other commands but CAPABILITIES, MODE, HELP, AUTHINFO, DATE and QUIT still @@ -647,10 +657,13 @@ func (s *Server) runScript(c *Conn) { if st.Delay > 0 && !c.sleep(st.Delay) { return } - if !c.writeReply(line, st.Reply...) { + c.shape, c.shapeStep = st.CompressShape, st.Compress + ok := c.writeReply(line, st.Reply...) + c.shapeStep = false + if !ok { return } - if st.Compress && !c.Compressed() && c.StartCompress() != nil { + if st.Compress && !c.Compressed() && c.startCompress(false, st.CompressShape) != nil { return } if st.Hang { diff --git a/fakeserver/storehandler.go b/fakeserver/storehandler.go index 59fc3ff..a44e5b8 100644 --- a/fakeserver/storehandler.go +++ b/fakeserver/storehandler.go @@ -91,6 +91,7 @@ func StoreHandler(st *Store, o StoreOptions) func(*Conn) { seq := new(atomic.Int64) // generated POST Message-IDs, shared by the handler's connections postMu := new(sync.Mutex) // serialises filing POSTs across the handler's connections return func(c *Conn) { + c.shape, c.shapeOn = o.CompressShape, o.Compress || o.CompressOutputOnly h := &storeConn{st: st, o: o, c: c, seq: seq, postMu: postMu, authed: o.User == "" && o.Pass == ""} h.serve() }