chore: Add sftp reconnections

Signed-off-by: Javier Aliaga <javier@diagrid.io>

Author: javier-aliaga

bindings/sftp/integration/README.md

@@ -0,0 +1,111 @@
+/*
+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"
+	"io"
+	"log"
+	"net"
+	"sync/atomic"
+)
+
+type Proxy struct {
+	ListenAddr        string
+	UpstreamAddr      string
+	Client            net.Conn
+	Server            net.Conn
+	ReconnectionCount atomic.Int32
+	Listener          net.Listener
+}
+
+func (p *Proxy) ListenAndServe() error {
+	ln, err := net.Listen("tcp", p.ListenAddr)
+	if err != nil {
+		log.Fatalf("listen: %v", err)
+	}
+	log.Printf("Proxy listening on %s -> %s", p.ListenAddr, p.UpstreamAddr)
+	p.Listener = ln
+
+	for {
+		client, err := ln.Accept()
+		if err != nil {
+			if errors.Is(err, net.ErrClosed) {
+				return nil
+			}
+			log.Printf("accept error: %v", err)
+			continue
+		}
+		go p.handle(client)
+	}
+}
+
+func (p *Proxy) handle(client net.Conn) {
+	defer client.Close()
+
+	// Connect to upstream SFTP server
+	server, err := net.Dial("tcp", p.UpstreamAddr)
+	if err != nil {
+		log.Printf("dial upstream: %v", err)
+		return
+	}
+	defer server.Close()
+
+	p.Client = client
+	p.Server = server
+	p.ReconnectionCount.Add(1)
+	errCh := make(chan error, 2)
+
+	// client -> server
+	go func() {
+		_, cErr := io.Copy(server, client)
+		errCh <- cErr
+	}()
+
+	// server -> client
+	go func() {
+		_, cErr := io.Copy(client, server)
+		errCh <- cErr
+	}()
+
+	// When either direction ends, close both ends
+	if err := <-errCh; err != nil && !isUsefullyClosed(err) {
+		log.Printf("proxy stream ended with error: %v", err)
+	}
+}
+
+func (p *Proxy) KillServerConn() error {
+	return p.Client.Close()
+}
+
+func (p *Proxy) Close() {
+	if p.Client != nil {
+		_ = p.Client.Close()
+	}
+
+	if p.Server != nil {
+		_ = p.Server.Close()
+	}
+
+	if p.Listener != nil {
+		_ = p.Listener.Close()
+	}
+
+	p.ReconnectionCount.Store(0)
+}
+
+// isUsefullyClosed filters common close conditions from logging noise
+func isUsefullyClosed(err error) bool {
+	return err == io.EOF || errors.Is(err, net.ErrClosed)
+}

bindings/sftp/integration/docker-compose.yaml

@@ -1,3 +1,16 @@
+/*
+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 (
@@ -7,7 +20,6 @@ import (
 	"fmt"
 	"io"
 	"reflect"
-	"sync"
 
 	sftpClient "github.com/pkg/sftp"
 	"golang.org/x/crypto/ssh"
@@ -26,30 +38,9 @@ const (
 
 // Sftp is a binding for file operations on sftp server.
 type Sftp struct {
-	metadata     *sftpMetadata
-	logger       logger.Logger
-	sftpClient   *sftpClient.Client
-	sshClient    *ssh.Client
-	clientConfig *ssh.ClientConfig
-	lock         sync.RWMutex
-}
-
-func (sftp *Sftp) Client() (*sftpClient.Client, error) {
-	sftp.lock.RLock()
-	current := sftp.sftpClient
-	sftp.lock.RUnlock()
-
-	if current != nil {
-		if _, err := current.Getwd(); err == nil {
-			return current, nil
-		}
-	}
-
-	err := sftp.handleReconnection()
-	if err != nil {
-		return nil, err
-	}
-	return sftp.sftpClient, nil
+	metadata *sftpMetadata
+	logger   logger.Logger
+	c        *Client
 }
 
 // sftpMetadata defines the sftp metadata.
@@ -137,21 +128,12 @@ func (sftp *Sftp) Init(_ context.Context, metadata bindings.Metadata) error {
 		HostKeyCallback: hostKeyCallback,
 	}
 
-	sshClient, err := ssh.Dial("tcp", m.Address, config)
-	if err != nil {
-		return fmt.Errorf("sftp binding error: error create ssh client: %w", err)
-	}
-
-	newSftpClient, err := sftpClient.NewClient(sshClient)
+	sftp.metadata = m
+	sftp.c, err = newClient(m.Address, config)
 	if err != nil {
-		_ = sshClient.Close()
-		return fmt.Errorf("sftp binding error: error create sftp client: %w", err)
+		return fmt.Errorf("sftp binding error: create sftp client error: %w", err)
 	}
 
-	sftp.clientConfig = config
-	sftp.metadata = m
-	sftp.sftpClient = newSftpClient
-
 	return nil
 }
 
@@ -185,19 +167,9 @@ func (sftp *Sftp) create(_ context.Context, req *bindings.InvokeRequest) (*bindi
 		return nil, fmt.Errorf("sftp binding error: %w", err)
 	}
 
-	dir, fileName := sftpClient.Split(path)
+	c := sftp.c
 
-	c, err := sftp.Client()
-	if err != nil {
-		return nil, fmt.Errorf("sftp binding error: error getting sftp client: %w", err)
-	}
-
-	err = c.MkdirAll(dir)
-	if err != nil {
-		return nil, fmt.Errorf("sftp binding error: error create dir %s: %w", dir, err)
-	}
-
-	file, err := c.Create(path)
+	file, fileName, err := c.create(path)
 	if err != nil {
 		return nil, fmt.Errorf("sftp binding error: error create file %s: %w", path, err)
 	}
@@ -240,12 +212,9 @@ func (sftp *Sftp) list(_ context.Context, req *bindings.InvokeRequest) (*binding
 		return nil, fmt.Errorf("sftp binding error: %w", err)
 	}
 
-	c, err := sftp.Client()
-	if err != nil {
-		return nil, fmt.Errorf("sftp binding error: error getting sftp client: %w", err)
-	}
+	c := sftp.c
 
-	files, err := c.ReadDir(path)
+	files, err := c.list(path)
 	if err != nil {
 		return nil, fmt.Errorf("sftp binding error: error read dir %s: %w", path, err)
 	}
@@ -280,12 +249,9 @@ func (sftp *Sftp) get(_ context.Context, req *bindings.InvokeRequest) (*bindings
 		return nil, fmt.Errorf("sftp binding error: %w", err)
 	}
 
-	c, err := sftp.Client()
-	if err != nil {
-		return nil, fmt.Errorf("sftp binding error: error getting sftp client: %w", err)
-	}
+	c := sftp.c
 
-	file, err := c.Open(path)
+	file, err := c.get(path)
 	if err != nil {
 		return nil, fmt.Errorf("sftp binding error: error open file %s: %w", path, err)
 	}
@@ -311,11 +277,9 @@ func (sftp *Sftp) delete(_ context.Context, req *bindings.InvokeRequest) (*bindi
 		return nil, fmt.Errorf("sftp binding error: %w", err)
 	}
 
-	c, err := sftp.Client()
-	if err != nil {
-		return nil, fmt.Errorf("sftp binding error: error getting sftp client: %w", err)
-	}
-	err = c.Remove(path)
+	c := sftp.c
+
+	err = c.delete(path)
 	if err != nil {
 		return nil, fmt.Errorf("sftp binding error: error remove file %s: %w", path, err)
 	}
@@ -339,9 +303,7 @@ func (sftp *Sftp) Invoke(ctx context.Context, req *bindings.InvokeRequest) (*bin
 }
 
 func (sftp *Sftp) Close() error {
-	sftp.lock.Lock()
-	defer sftp.lock.Unlock()
-	return sftp.sftpClient.Close()
+	return sftp.c.Close()
 }
 
 func (metadata sftpMetadata) getPath(requestMetadata map[string]string) (path string, err error) {
@@ -375,41 +337,3 @@ func (sftp *Sftp) GetComponentMetadata() (metadataInfo metadata.MetadataMap) {
 	metadata.GetMetadataInfoFromStructType(reflect.TypeOf(metadataStruct), &metadataInfo, metadata.BindingType)
 	return
 }
-
-func (sftp *Sftp) handleReconnection() error {
-	sftp.lock.Lock()
-	defer sftp.lock.Unlock()
-
-	// Re-check after acquiring the write lock
-	if sftp.sftpClient != nil {
-		if _, err := sftp.sftpClient.Getwd(); err == nil {
-			return nil
-		}
-		_ = sftp.sftpClient.Close()
-		sftp.sftpClient = nil
-	}
-	if sftp.sshClient != nil {
-		_ = sftp.sshClient.Close()
-		sftp.sshClient = nil
-	}
-
-	if sftp.metadata == nil || sftp.clientConfig == nil {
-		return fmt.Errorf("sftp binding error: client not initialized")
-	}
-
-	sshClient, err := ssh.Dial("tcp", sftp.metadata.Address, sftp.clientConfig)
-	if err != nil {
-		return fmt.Errorf("sftp binding error: error create ssh client: %w", err)
-	}
-
-	newSftpClient, err := sftpClient.NewClient(sshClient)
-	if err != nil {
-		_ = sshClient.Close()
-		return fmt.Errorf("sftp binding error: error create sftp client: %w", err)
-	}
-
-	sftp.sshClient = sshClient
-	sftp.sftpClient = newSftpClient
-
-	return nil
-}