chore: Handle sftp reconnections

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

Author: javier-aliaga

bindings/sftp/proxy/proxy.go

@@ -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.Server.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/sftp.go

@@ -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 (
@@ -25,9 +38,9 @@ const (
 
 // Sftp is a binding for file operations on sftp server.
 type Sftp struct {
-	metadata   *sftpMetadata
-	logger     logger.Logger
-	sftpClient *sftpClient.Client
+	metadata *sftpMetadata
+	logger   logger.Logger
+	c        *Client
 }
 
 // sftpMetadata defines the sftp metadata.
@@ -115,19 +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 {
-		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.metadata = m
-	sftp.sftpClient = newSftpClient
-
 	return nil
 }
 
@@ -161,14 +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
 
-	err = sftp.sftpClient.MkdirAll(dir)
-	if err != nil {
-		return nil, fmt.Errorf("sftp binding error: error create dir %s: %w", dir, err)
-	}
-
-	file, err := sftp.sftpClient.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)
 	}
@@ -211,7 +212,9 @@ func (sftp *Sftp) list(_ context.Context, req *bindings.InvokeRequest) (*binding
 		return nil, fmt.Errorf("sftp binding error: %w", err)
 	}
 
-	files, err := sftp.sftpClient.ReadDir(path)
+	c := sftp.c
+
+	files, err := c.list(path)
 	if err != nil {
 		return nil, fmt.Errorf("sftp binding error: error read dir %s: %w", path, err)
 	}
@@ -246,7 +249,9 @@ func (sftp *Sftp) get(_ context.Context, req *bindings.InvokeRequest) (*bindings
 		return nil, fmt.Errorf("sftp binding error: %w", err)
 	}
 
-	file, err := sftp.sftpClient.Open(path)
+	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)
 	}
@@ -272,7 +277,9 @@ func (sftp *Sftp) delete(_ context.Context, req *bindings.InvokeRequest) (*bindi
 		return nil, fmt.Errorf("sftp binding error: %w", err)
 	}
 
-	err = sftp.sftpClient.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)
 	}
@@ -296,7 +303,7 @@ func (sftp *Sftp) Invoke(ctx context.Context, req *bindings.InvokeRequest) (*bin
 }
 
 func (sftp *Sftp) Close() error {
-	return sftp.sftpClient.Close()
+	return sftp.c.Close()
 }
 
 func (metadata sftpMetadata) getPath(requestMetadata map[string]string) (path string, err error) {