Support SSH auth for SFTP event source

api/jsonschema/schema.json

@@ -2690,6 +2690,10 @@
           "description": "PollIntervalDuration the interval at which to poll the SFTP server defaults to 10 seconds",
           "type": "string"
         },
+        "sshKeySecret": {
+          "$ref": "#/definitions/io.k8s.api.core.v1.SecretKeySelector",
+          "description": "SSHKeySecret refers to the secret that contains SSH key"
+        },
         "username": {
           "$ref": "#/definitions/io.k8s.api.core.v1.SecretKeySelector",
           "description": "Username required for authentication if any."

eventsources/sources/sftp/start.go

@@ -21,6 +21,7 @@ import (
 	"encoding/json"
 	"fmt"
 	"io/fs"
+	"os"
 	"regexp"
 	"strings"
 	"time"
@@ -73,19 +74,46 @@ func (el *EventListener) StartListening(ctx context.Context, dispatch func([]byt
 	if err != nil {
 		return fmt.Errorf("username not found, %w", err)
 	}
-	password, err := common.GetSecretFromVolume(el.SFTPEventSource.Password)
-	if err != nil {
-		return fmt.Errorf("password not found, %w", err)
-	}
 	address, err := common.GetSecretFromVolume(el.SFTPEventSource.Address)
 	if err != nil {
 		return fmt.Errorf("address not found, %w", err)
 	}
 
+	var authMethod ssh.AuthMethod
+	var hostKeyCallback ssh.HostKeyCallback
+
+	if el.SFTPEventSource.SSHKeySecret != nil {
+		sshKeyPath, err := common.GetSecretVolumePath(el.SFTPEventSource.SSHKeySecret)
+		if err != nil {
+			return fmt.Errorf("failed to get SSH key from mounted volume, %w", err)
+		}
+		sshKey, err := os.ReadFile(sshKeyPath)
+		if err != nil {
+			return fmt.Errorf("failed to read ssh key file. err: %+v", err)
+		}
+		signer, err := ssh.ParsePrivateKey(sshKey)
+		if err != nil {
+			return fmt.Errorf("failed to parse private ssh key. err: %+v", err)
+		}
+		publicKey, err := ssh.ParsePublicKey(sshKey)
+		if err != nil {
+			return fmt.Errorf("failed to parse public ssh key. err: %+v", err)
+		}
+		authMethod = ssh.PublicKeys(signer)
+		hostKeyCallback = ssh.FixedHostKey(publicKey)
+	} else {
+		password, err := common.GetSecretFromVolume(el.SFTPEventSource.Password)
+		if err != nil {
+			return fmt.Errorf("password not found, %w", err)
+		}
+		authMethod = ssh.Password(password)
+		hostKeyCallback = ssh.InsecureIgnoreHostKey()
+	}
+
 	sftpConfig := &ssh.ClientConfig{
 		User:            username,
-		Auth:            []ssh.AuthMethod{ssh.Password(password)},
-		HostKeyCallback: ssh.InsecureIgnoreHostKey(), // TODO: enable host key callback
+		Auth:            []ssh.AuthMethod{authMethod},
+		HostKeyCallback: hostKeyCallback,
 	}
 
 	var sshClient *ssh.Client

pkg/apis/eventsource/v1alpha1/types.go

@@ -303,18 +303,20 @@ type SFTPEventSource struct {
 	Username *corev1.SecretKeySelector `json:"username,omitempty" protobuf:"bytes,3,opt,name=username"`
 	// Password required for authentication if any.
 	Password *corev1.SecretKeySelector `json:"password,omitempty" protobuf:"bytes,4,opt,name=password"`
+	// SSHKeySecret refers to the secret that contains SSH key
+	SSHKeySecret *corev1.SecretKeySelector `json:"sshKeySecret,omitempty" protobuf:"bytes,5,opt,name=sshKeySecret"`
 	// Address sftp address.
-	Address *corev1.SecretKeySelector `json:"address,omitempty" protobuf:"bytes,5,opt,name=address"`
+	Address *corev1.SecretKeySelector `json:"address,omitempty" protobuf:"bytes,6,opt,name=address"`
 	// Metadata holds the user defined metadata which will passed along the event payload.
 	// +optional
-	Metadata map[string]string `json:"metadata,omitempty" protobuf:"bytes,6,rep,name=metadata"`
+	Metadata map[string]string `json:"metadata,omitempty" protobuf:"bytes,7,rep,name=metadata"`
 	// Filter
 	// +optional
-	Filter *EventSourceFilter `json:"filter,omitempty" protobuf:"bytes,7,opt,name=filter"`
+	Filter *EventSourceFilter `json:"filter,omitempty" protobuf:"bytes,8,opt,name=filter"`
 	// PollIntervalDuration the interval at which to poll the SFTP server
 	// defaults to 10 seconds
 	// +optional
-	PollIntervalDuration string `json:"pollIntervalDuration" protobuf:"varint,8,opt,name=pollIntervalDuration"`
+	PollIntervalDuration string `json:"pollIntervalDuration" protobuf:"varint,9,opt,name=pollIntervalDuration"`
 }
 
 // ResourceEventType is the type of event for the K8s resource mutation