feat(deflate): Adding support for transfer compression (#461)

Add MODE Z (deflate) transfer compression support per RFC-compliant FTP extensions.

Key features:
- Implement deflate compression for data transfers using compress/flate
- Add TransferFinalizer interface to properly terminate compressed streams
- Advertise MODE Z in FEAT response for client discovery
- Disallow REST (resume) in deflate mode (streaming format limitation)
- Handle closed connection errors gracefully

Implementation details:
- deflateReadWriter wraps connections with compression/decompression
- TransferMode enum (Stream=0, Deflate=1) tracks current mode per client
- MODE command accepts S (stream) and Z (deflate) parameters
- Proper BFINAL block written on stream close for valid decompression

Test coverage:
- Upload and download with deflate compression
- Mode switching between stream and deflate
- Error paths for invalid compression levels
- Unit tests for helper functions

Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>

Author: fclairamb

client_handler.go

@@ -2,6 +2,7 @@ package ftpserver
 
 import (
 	"bufio"
+	"compress/flate"
 	"errors"
 	"fmt"
 	"io"
@@ -33,6 +34,17 @@ const (
 	TransferTypeBinary
 )
 
+// TransferMode is the enumerable that represents the transfer mode (stream, block, compressed, deflate)
+type TransferMode int8
+
+// Transfer modes
+const (
+	// TransferModeStream is the standard uncompressed transfer mode
+	TransferModeStream TransferMode = iota
+	// TransferModeDeflate is the compressed transfer mode using deflate algorithm
+	TransferModeDeflate
+)
+
 // DataChannel is the enumerable that represents the data channel (active or passive)
 type DataChannel int8
 
@@ -103,6 +115,7 @@ type clientHandler struct {
 	debug               bool            // Show debugging info on the server side
 	transferTLS         bool            // Use TLS for transfer connection
 	controlTLS          bool            // Use TLS for control connection
+	transferMode        TransferMode    // Transfer mode (stream, deflate)
 	isTransferOpen      bool            // indicate if the transfer connection is opened
 	isTransferAborted   bool            // indicate if the transfer was aborted
 	connClosed          bool            // indicates if the connection has been commanded to close
@@ -660,7 +673,7 @@ func (c *clientHandler) GetTranferInfo() string {
 	return c.transfer.GetInfo()
 }
 
-func (c *clientHandler) TransferOpen(info string) (net.Conn, error) {
+func (c *clientHandler) TransferOpen(info string) (io.ReadWriter, error) {
 	c.transferMu.Lock()
 	defer c.transferMu.Unlock()
 
@@ -696,6 +709,17 @@ func (c *clientHandler) TransferOpen(info string) (net.Conn, error) {
 		return nil, err
 	}
 
+	var transferStream io.ReadWriter = conn
+
+	if c.transferMode == TransferModeDeflate {
+		transferStream, err = newDeflateTransfer(transferStream, c.server.settings.DeflateCompressionLevel)
+		if err != nil {
+			c.writeMessage(StatusActionNotTaken, fmt.Sprintf("Could not switch to deflate mode: %v", err))
+
+			return nil, fmt.Errorf("could not switch to deflate mode: %w", err)
+		}
+	}
+
 	c.isTransferOpen = true
 	c.transfer.SetInfo(info)
 
@@ -708,18 +732,56 @@ func (c *clientHandler) TransferOpen(info string) (net.Conn, error) {
 			"localAddr", conn.LocalAddr().String())
 	}
 
-	return conn, nil
+	return transferStream, nil
+}
+
+// Flusher is the interface that wraps the basic Flush method.
+type Flusher interface {
+	Flush() error
+}
+
+// TransferFinalizer is the interface for transfer streams that need explicit
+// finalization (e.g., deflate streams need to write end-of-stream markers).
+// This is distinct from closing the underlying connection.
+// We use FinalizeTransfer() instead of Close() to distinguish from io.Closer,
+// since net.Conn already implements io.Closer and we don't want to accidentally
+// close the underlying connection.
+type TransferFinalizer interface {
+	FinalizeTransfer() error
 }
 
-func (c *clientHandler) TransferClose(err error) {
+func (c *clientHandler) TransferClose(transfer io.ReadWriter, err error) {
 	c.transferMu.Lock()
 	defer c.transferMu.Unlock()
 
+	// Check if this is a transfer stream that needs explicit finalization (e.g., deflate).
+	// TransferFinalizer.FinalizeTransfer() for deflate writes the end-of-stream marker AND flushes,
+	// so we don't need to call Flush() separately.
+	if finalizer, ok := transfer.(TransferFinalizer); ok {
+		if errFinalize := finalizer.FinalizeTransfer(); errFinalize != nil {
+			c.logger.Warn(
+				"Error finalizing transfer stream",
+				"err", errFinalize,
+			)
+		}
+	} else if flush, ok := transfer.(Flusher); ok {
+		// Only flush if NOT a TransferCloser (deflate's Close already flushes)
+		if errFlush := flush.Flush(); errFlush != nil {
+			c.logger.Warn(
+				"Error flushing transfer connection",
+				"err", errFlush,
+			)
+		}
+	}
+
+	// Finally close the underlying connection.
+	// "Use of closed network connection" is normal in FTP - the client can close
+	// the connection when done, and we should treat this as success.
 	errClose := c.closeTransfer()
-	if errClose != nil {
+	if errClose != nil && !isClosedConnError(errClose) {
 		c.logger.Warn(
 			"Problem closing transfer connection",
-			"err", err,
+			"err", errClose,
 		)
 	}
 
@@ -730,6 +792,11 @@ func (c *clientHandler) TransferClose(err error) {
 		return
 	}
 
+	// Treat "connection already closed" as success - it means transfer completed
+	if isClosedConnError(errClose) {
+		errClose = nil
+	}
+
 	switch {
 	case err == nil && errClose == nil:
 		c.writeMessage(StatusClosingDataConn, "Closing transfer connection")
@@ -740,6 +807,19 @@ func (c *clientHandler) TransferClose(err error) {
 	}
 }
 
+// isClosedConnError checks if the error indicates the connection is already closed.
+// This is normal FTP behavior - the client can close the connection when done.
+func isClosedConnError(err error) bool {
+	if err == nil {
+		return false
+	}
+
+	errStr := err.Error()
+
+	return strings.Contains(errStr, "use of closed network connection") ||
+		strings.Contains(errStr, "connection reset by peer")
+}
+
 func (c *clientHandler) checkDataConnectionRequirement(dataConnIP net.IP, channelType DataChannel) error {
 	var requirement DataConnectionRequirement
 
@@ -821,3 +901,57 @@ func getMessageLines(message string) []string {
 
 	return lines
 }
+
+// Compile-time checks that deflateReadWriter implements required interfaces
+var (
+	_ Flusher           = (*deflateReadWriter)(nil)
+	_ TransferFinalizer = (*deflateReadWriter)(nil)
+)
+
+type deflateReadWriter struct {
+	reader io.ReadCloser // flate.NewReader returns io.ReadCloser
+	writer *flate.Writer
+}
+
+func (d *deflateReadWriter) Read(p []byte) (int, error) {
+	return d.reader.Read(p)
+}
+
+func (d *deflateReadWriter) Write(p []byte) (int, error) {
+	return d.writer.Write(p)
+}
+
+// Flush flushes buffered data to the underlying writer.
+func (d *deflateReadWriter) Flush() error {
+	return d.writer.Flush()
+}
+
+// FinalizeTransfer finalizes the deflate stream by writing the BFINAL block (end-of-stream marker).
+// This does NOT close the underlying connection - it only finalizes the deflate stream.
+func (d *deflateReadWriter) FinalizeTransfer() error {
+	// Close the writer to write the BFINAL block
+	if err := d.writer.Close(); err != nil {
+		return fmt.Errorf("error closing deflate writer: %w", err)
+	}
+
+	// Close the reader to release resources
+	if err := d.reader.Close(); err != nil {
+		return fmt.Errorf("error closing deflate reader: %w", err)
+	}
+
+	return nil
+}
+
+func newDeflateTransfer(conn io.ReadWriter, level int) (*deflateReadWriter, error) {
+	writer, err := flate.NewWriter(conn, level)
+	if err != nil {
+		return nil, fmt.Errorf("could not create deflate writer: %w", err)
+	}
+
+	reader := flate.NewReader(conn)
+
+	return &deflateReadWriter{
+		reader: reader,
+		writer: writer,
+	}, nil
+}

client_handler_test.go

@@ -1,6 +1,7 @@
 package ftpserver
 
 import (
+	"errors"
 	"fmt"
 	"net"
 	"slices"
@@ -490,3 +491,87 @@ func TestExtraData(t *testing.T) {
 		require.Equal(t, k, extra)
 	}
 }
+
+var (
+	errClosedConn    = net.ErrClosed
+	errConnReset     = errors.New("connection reset by peer")
+	errOther         = errors.New("some other error")
+	errWrappedClosed = fmt.Errorf("failed: %w", errClosedConn)
+)
+
+func TestIsClosedConnError(t *testing.T) {
+	t.Parallel()
+
+	tests := []struct {
+		name     string
+		err      error
+		expected bool
+	}{
+		{name: "nil error", err: nil, expected: false},
+		{name: "net.ErrClosed", err: errClosedConn, expected: true},
+		{name: "connection reset by peer", err: errConnReset, expected: true},
+		{name: "wrapped closed connection error", err: errWrappedClosed, expected: true},
+		{name: "other error", err: errOther, expected: false},
+	}
+
+	for _, tc := range tests {
+		t.Run(tc.name, func(t *testing.T) {
+			t.Parallel()
+			result := isClosedConnError(tc.err)
+			assert.Equal(t, tc.expected, result)
+		})
+	}
+}
+
+func TestDeflateReadWriterFlush(t *testing.T) {
+	t.Parallel()
+
+	// Create a buffer to write to
+	buf := &mockReadWriter{}
+
+	// Create deflate transfer
+	deflate, err := newDeflateTransfer(buf, 5)
+	require.NoError(t, err)
+
+	// Write some data
+	data := []byte("test data for deflate")
+	n, err := deflate.Write(data)
+	require.NoError(t, err)
+	require.Equal(t, len(data), n)
+
+	// Test Flush
+	err = deflate.Flush()
+	require.NoError(t, err)
+
+	// Verify data was written to buffer
+	require.Positive(t, buf.writeCount)
+}
+
+func TestNewDeflateTransferInvalidLevel(t *testing.T) {
+	t.Parallel()
+
+	buf := &mockReadWriter{}
+
+	// Test with invalid compression level (valid range is -2 to 9)
+	_, err := newDeflateTransfer(buf, 100)
+	require.Error(t, err)
+	require.Contains(t, err.Error(), "could not create deflate writer")
+}
+
+// mockReadWriter is a simple mock for testing
+type mockReadWriter struct {
+	writeCount int
+	readCount  int
+}
+
+func (m *mockReadWriter) Write(p []byte) (int, error) {
+	m.writeCount++
+
+	return len(p), nil
+}
+
+func (m *mockReadWriter) Read(_ []byte) (int, error) {
+	m.readCount++
+
+	return 0, nil
+}

driver.go

@@ -323,6 +323,7 @@ type Settings struct {
 	DisableSTAT              bool             // Disable Server STATUS, STAT on files and directories will still work
 	DisableSYST              bool             // Disable SYST
 	EnableCOMB               bool             // Enable COMB support
+	DeflateCompressionLevel  int              // Deflate compression level (1-9)
 	DefaultTransferType      TransferType     // Transfer type to use if the client don't send the TYPE command
 	// ActiveConnectionsCheck defines the security requirements for active connections
 	ActiveConnectionsCheck DataConnectionRequirement

driver_test.go

@@ -55,12 +55,12 @@ func (driver *TestServerDriver) Init() {
 	}
 
 	{
-		dir, _ := os.MkdirTemp("", "example")
-		if err := os.MkdirAll(dir, 0o750); err != nil {
+		driver.serverDir, _ = os.MkdirTemp("", "example")
+		if err := os.MkdirAll(driver.serverDir, 0o750); err != nil {
 			panic(err)
 		}
 
-		driver.fs = afero.NewBasePathFs(afero.NewOsFs(), dir)
+		driver.fs = afero.NewBasePathFs(afero.NewOsFs(), driver.serverDir)
 	}
 }
 
@@ -124,6 +124,7 @@ type authUserProvider func(user, pass string) (ClientDriver, error)
 type TestServerDriver struct {
 	Clients              []ClientContext
 	errPassiveListener   error
+	serverDir            string
 	fs                   afero.Fs
 	clientMU             sync.Mutex
 	AuthProvider         authUserProvider

errors_test.go

@@ -24,7 +24,7 @@ func TestCustomErrorsCode(t *testing.T) {
 func TestTransferCloseStorageExceeded(t *testing.T) {
 	buf := bytes.Buffer{}
 	h := clientHandler{writer: bufio.NewWriter(&buf)}
-	h.TransferClose(ErrStorageExceeded)
+	h.TransferClose(nil, ErrStorageExceeded)
 	require.Equal(t, "552 Issue during transfer: storage limit exceeded\r\n", buf.String())
 }
 

handle_dirs.go

@@ -191,7 +191,7 @@ func (c *clientHandler) handleLIST(param string) error {
 	if files, _, err := c.getFileList(param, true); err == nil || errors.Is(err, io.EOF) {
 		if tr, errTr := c.TransferOpen(info); errTr == nil {
 			err = c.dirTransferLIST(tr, files)
-			c.TransferClose(err)
+			c.TransferClose(tr, err)
 
 			return nil
 		}
@@ -210,7 +210,7 @@ func (c *clientHandler) handleNLST(param string) error {
 	if files, parentDir, err := c.getFileList(param, true); err == nil || errors.Is(err, io.EOF) {
 		if tr, errTrOpen := c.TransferOpen(info); errTrOpen == nil {
 			err = c.dirTransferNLST(tr, files, parentDir)
-			c.TransferClose(err)
+			c.TransferClose(tr, err)
 
 			return nil
 		}
@@ -257,7 +257,7 @@ func (c *clientHandler) handleMLSD(param string) error {
 	if files, _, err := c.getFileList(param, false); err == nil || errors.Is(err, io.EOF) {
 		if tr, errTr := c.TransferOpen(info); errTr == nil {
 			err = c.dirTransferMLSD(tr, files)
-			c.TransferClose(err)
+			c.TransferClose(tr, err)
 
 			return nil
 		}