fix(sftp): resolve connection poisoning with enterprise SFTP servers

bindings/sftp/client.go

@@ -16,7 +16,9 @@ package sftp
 import (
 	"errors"
 	"fmt"
+	"io"
 	"os"
+	"strings"
 	"sync"
 	"sync/atomic"
 
@@ -98,39 +100,55 @@ func (c *Client) list(path string) ([]os.FileInfo, error) {
 	return fi, nil
 }
 
-func (c *Client) create(path string) (*sftpClient.File, string, error) {
+func (c *Client) create(data []byte, path string) (string, error) {
 	dir, fileName := sftpClient.Split(path)
 
-	var file *sftpClient.File
-
 	createFn := func() error {
-		cErr := c.sftpClient.MkdirAll(dir)
-		if cErr != nil {
-			return cErr
+		// Only create directory if it doesn't already exist
+		// This prevents "not a directory" errors on strict SFTP servers like Axway MFT
+		if dir != "" && dir != "." && dir != "/" {
+			_, statErr := c.sftpClient.Stat(dir)
+			if statErr != nil {
+				// Directory doesn't exist, create it
+				if mkdirErr := c.sftpClient.MkdirAll(dir); mkdirErr != nil {
+					return fmt.Errorf("error create dir %s: %w", dir, mkdirErr)
+				}
+			}
 		}
 
-		file, cErr = c.sftpClient.Create(path)
+		file, cErr := c.sftpClient.Create(path)
 		if cErr != nil {
-			return cErr
+			return fmt.Errorf("error create file %s: %w", path, cErr)
+		}
+		defer file.Close()
+
+		_, wErr := file.Write(data)
+		if wErr != nil {
+			return fmt.Errorf("error write file %s: %w", path, wErr)
 		}
 
 		return nil
 	}
 
 	rErr := c.withReconnection(createFn)
 	if rErr != nil {
-		return nil, "", rErr
+		return "", rErr
 	}
 
-	return file, fileName, nil
+	return fileName, nil
 }
 
-func (c *Client) get(path string) (*sftpClient.File, error) {
-	var f *sftpClient.File
+func (c *Client) get(path string) ([]byte, error) {
+	var data []byte
 
 	fn := func() error {
-		var err error
-		f, err = c.sftpClient.Open(path)
+		f, err := c.sftpClient.Open(path)
+		if err != nil {
+			return err
+		}
+		defer f.Close()
+
+		data, err = io.ReadAll(f)
 		return err
 	}
 
@@ -139,7 +157,7 @@ func (c *Client) get(path string) (*sftpClient.File, error) {
 		return nil, err
 	}
 
-	return f, nil
+	return data, nil
 }
 
 func (c *Client) delete(path string) error {
@@ -184,8 +202,8 @@ func (c *Client) withReconnection(fn func() error) error {
 }
 
 func (c *Client) do(fn func() error) error {
-	c.lock.RLock()
-	defer c.lock.RUnlock()
+	c.lock.Lock()
+	defer c.lock.Unlock()
 	return fn()
 }
 
@@ -244,10 +262,33 @@ func (c *Client) shouldReconnect(err error) bool {
 		return false
 	}
 
-	// SFTP status errors that are logical, not connectivity (avoid reconnect)
+	// Check for StatusError using errors.As - if it's a StatusError,
+	// it's an SFTP protocol error, not a connection issue
+	var statusErr *sftpClient.StatusError
+	if errors.As(err, &statusErr) {
+		// Any StatusError is a protocol-level response from the server,
+		// meaning the connection is working fine
+		return false
+	}
+
+	// Check sentinel errors for SFTP logical errors
+	// These errors indicate application-level issues, not connection problems
 	if errors.Is(err, sftpClient.ErrSSHFxPermissionDenied) ||
 		errors.Is(err, sftpClient.ErrSSHFxNoSuchFile) ||
-		errors.Is(err, sftpClient.ErrSSHFxOpUnsupported) {
+		errors.Is(err, sftpClient.ErrSSHFxOpUnsupported) ||
+		errors.Is(err, sftpClient.ErrSSHFxFailure) ||
+		errors.Is(err, sftpClient.ErrSSHFxBadMessage) ||
+		errors.Is(err, sftpClient.ErrSSHFxEOF) {
+		return false
+	}
+
+	// Fallback: string matching for wrapped errors that may not implement Is/As
+	errStr := strings.ToLower(err.Error())
+	if strings.Contains(errStr, "permission denied") ||
+		strings.Contains(errStr, "no such file") ||
+		strings.Contains(errStr, "not a directory") ||
+		strings.Contains(errStr, "file exists") ||
+		strings.Contains(errStr, "bad message") {
 		return false
 	}
 

bindings/sftp/client_test.go

@@ -0,0 +1,128 @@
+/*
+Copyright 2025 The Dapr Authors
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+    http://www.apache.org/licenses/LICENSE-2.0
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package sftp
+
+import (
+	"errors"
+	"fmt"
+	"testing"
+
+	sftpClient "github.com/pkg/sftp"
+	"github.com/stretchr/testify/assert"
+)
+
+func TestShouldReconnect(t *testing.T) {
+	c := &Client{}
+
+	t.Run("nil error should not reconnect", func(t *testing.T) {
+		assert.False(t, c.shouldReconnect(nil))
+	})
+
+	t.Run("StatusError should not reconnect", func(t *testing.T) {
+		// StatusError with permission denied code (3)
+		statusErr := &sftpClient.StatusError{
+			Code: 3, // SSH_FX_PERMISSION_DENIED
+		}
+		assert.False(t, c.shouldReconnect(statusErr))
+	})
+
+	t.Run("wrapped StatusError should not reconnect", func(t *testing.T) {
+		// StatusError with no such file code (2)
+		statusErr := &sftpClient.StatusError{
+			Code: 2, // SSH_FX_NO_SUCH_FILE
+		}
+		wrappedErr := fmt.Errorf("operation failed: %w", statusErr)
+		assert.False(t, c.shouldReconnect(wrappedErr))
+	})
+
+	t.Run("ErrSSHFxPermissionDenied should not reconnect", func(t *testing.T) {
+		assert.False(t, c.shouldReconnect(sftpClient.ErrSSHFxPermissionDenied))
+	})
+
+	t.Run("ErrSSHFxNoSuchFile should not reconnect", func(t *testing.T) {
+		assert.False(t, c.shouldReconnect(sftpClient.ErrSSHFxNoSuchFile))
+	})
+
+	t.Run("ErrSSHFxOpUnsupported should not reconnect", func(t *testing.T) {
+		assert.False(t, c.shouldReconnect(sftpClient.ErrSSHFxOpUnsupported))
+	})
+
+	t.Run("ErrSSHFxFailure should not reconnect", func(t *testing.T) {
+		assert.False(t, c.shouldReconnect(sftpClient.ErrSSHFxFailure))
+	})
+
+	t.Run("ErrSSHFxBadMessage should not reconnect", func(t *testing.T) {
+		assert.False(t, c.shouldReconnect(sftpClient.ErrSSHFxBadMessage))
+	})
+
+	t.Run("ErrSSHFxEOF should not reconnect", func(t *testing.T) {
+		assert.False(t, c.shouldReconnect(sftpClient.ErrSSHFxEOF))
+	})
+
+	t.Run("permission denied string should not reconnect", func(t *testing.T) {
+		err := errors.New("sftp: permission denied")
+		assert.False(t, c.shouldReconnect(err))
+	})
+
+	t.Run("no such file string should not reconnect", func(t *testing.T) {
+		err := errors.New("sftp: no such file or directory")
+		assert.False(t, c.shouldReconnect(err))
+	})
+
+	t.Run("not a directory string should not reconnect", func(t *testing.T) {
+		err := errors.New("mkdir upload/: not a directory")
+		assert.False(t, c.shouldReconnect(err))
+	})
+
+	t.Run("file exists string should not reconnect", func(t *testing.T) {
+		err := errors.New("sftp: file exists error")
+		assert.False(t, c.shouldReconnect(err))
+	})
+
+	t.Run("bad message string should not reconnect", func(t *testing.T) {
+		err := errors.New("sftp: bad message")
+		assert.False(t, c.shouldReconnect(err))
+	})
+
+	t.Run("connection reset should reconnect", func(t *testing.T) {
+		err := errors.New("connection reset by peer")
+		assert.True(t, c.shouldReconnect(err))
+	})
+
+	t.Run("EOF should reconnect", func(t *testing.T) {
+		err := errors.New("EOF")
+		assert.True(t, c.shouldReconnect(err))
+	})
+
+	t.Run("broken pipe should reconnect", func(t *testing.T) {
+		err := errors.New("write: broken pipe")
+		assert.True(t, c.shouldReconnect(err))
+	})
+
+	t.Run("connection refused should reconnect", func(t *testing.T) {
+		err := errors.New("connection refused")
+		assert.True(t, c.shouldReconnect(err))
+	})
+
+	t.Run("timeout should reconnect", func(t *testing.T) {
+		err := errors.New("i/o timeout")
+		assert.True(t, c.shouldReconnect(err))
+	})
+
+	t.Run("unknown error should reconnect", func(t *testing.T) {
+		err := errors.New("unknown transport error")
+		assert.True(t, c.shouldReconnect(err))
+	})
+}
+

bindings/sftp/sftp.go

@@ -18,7 +18,6 @@ import (
 	"encoding/json"
 	"errors"
 	"fmt"
-	"io"
 	"reflect"
 
 	sftpClient "github.com/pkg/sftp"
@@ -167,23 +166,9 @@ func (sftp *Sftp) create(_ context.Context, req *bindings.InvokeRequest) (*bindi
 		return nil, fmt.Errorf("sftp binding error: %w", err)
 	}
 
-	c := sftp.c
-
-	file, fileName, err := c.create(path)
-	if err != nil {
-		return nil, fmt.Errorf("sftp binding error: error create file %s: %w", path, err)
-	}
-
-	defer func() {
-		closeErr := file.Close()
-		if closeErr != nil {
-			sftp.logger.Errorf("sftp binding error: error close file: %w", closeErr)
-		}
-	}()
-
-	_, err = file.Write(req.Data)
+	fileName, err := sftp.c.create(req.Data, path)
 	if err != nil {
-		return nil, fmt.Errorf("sftp binding error: error write file: %w", err)
+		return nil, fmt.Errorf("sftp binding error: %w", err)
 	}
 
 	jsonResponse, err := json.Marshal(createResponse{
@@ -249,20 +234,13 @@ func (sftp *Sftp) get(_ context.Context, req *bindings.InvokeRequest) (*bindings
 		return nil, fmt.Errorf("sftp binding error: %w", err)
 	}
 
-	c := sftp.c
-
-	file, err := c.get(path)
-	if err != nil {
-		return nil, fmt.Errorf("sftp binding error: error open file %s: %w", path, err)
-	}
-
-	b, err := io.ReadAll(file)
+	data, err := sftp.c.get(path)
 	if err != nil {
-		return nil, fmt.Errorf("sftp binding error: error read file %s: %w", path, err)
+		return nil, fmt.Errorf("sftp binding error: error reading file %s: %w", path, err)
 	}
 
 	return &bindings.InvokeResponse{
-		Data: b,
+		Data: data,
 	}, nil
 }