eric/distributor
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases 7 Wiki Activity

Add SSH SFTP backend support

Browse Source
This commit is contained in:
Eric Rakestraw
2026-05-31 16:53:37 +00:00
parent 1ad566264f
commit 84f77ec0d0
29 changed files with 1629 additions and 40 deletions
Download Patch File Download Diff File Expand all files Collapse all files

60
internal/app/backends.go
View File

@@ -3,14 +3,25 @@ package app
import (
"context"
"fmt"
"strconv"
"gitea.maximumdirect.net/eric/distributor/internal/adapters/local"
sshadapter "gitea.maximumdirect.net/eric/distributor/internal/adapters/ssh"
"gitea.maximumdirect.net/eric/distributor/internal/config"
"gitea.maximumdirect.net/eric/distributor/internal/storage"
)
const storagePathKey = "path"
const (
sshHostKey = "host"
sshUserKey = "user"
sshPortKey = "port"
sshKeyFileKey = "ssh_key_file"
sshKnownHostsKey = "known_hosts"
sshHostKeyPolicyKey = "host_key_policy"
)
type backendFactory struct {
registry *storage.Registry
}
@@ -23,23 +34,64 @@ func newBackendFactory() *backendFactory {
}
return local.New(cfg[storagePathKey])
})
_ = registry.Register(config.BackendSSH, func(ctx context.Context, cfg storage.OpenConfig) (storage.Backend, error) {
port, err := strconv.Atoi(cfg[sshPortKey])
if err != nil {
return nil, fmt.Errorf("ssh port: %w", err)
}
return sshadapter.New(ctx, sshadapter.Options{
Host: cfg[sshHostKey],
User: cfg[sshUserKey],
Port: port,
Root: cfg[storagePathKey],
KeyFile: cfg[sshKeyFileKey],
KnownHosts: cfg[sshKnownHostsKey],
HostKeyPolicy: sshadapter.HostKeyPolicy(cfg[sshHostKeyPolicyKey]),
})
})
return &backendFactory{registry: registry}
}
func (f *backendFactory) openSource(ctx context.Context, source config.Backend) (storage.Backend, error) {
if source.Backend != config.BackendLocal {
if source.Backend != config.BackendLocal && source.Backend != config.BackendSSH {
return nil, fmt.Errorf("source backend %s is not implemented for execution", source.Backend)
}
return f.registry.Open(ctx, source.Backend, storage.OpenConfig{storagePathKey: source.Path})
return f.registry.Open(ctx, source.Backend, sourceOpenConfig(source))
}
func (f *backendFactory) openDestination(ctx context.Context, destination config.Destination) (storage.Backend, error) {
if destination.Backend != config.BackendLocal {
if destination.Backend != config.BackendLocal && destination.Backend != config.BackendSSH {
return nil, fmt.Errorf("backend %s is not implemented for execution", destination.Backend)
}
return f.registry.Open(ctx, destination.Backend, storage.OpenConfig{storagePathKey: destination.Path})
return f.registry.Open(ctx, destination.Backend, destinationOpenConfig(destination))
}
func (f *backendFactory) openLocalPath(ctx context.Context, path string) (storage.Backend, error) {
return f.registry.Open(ctx, config.BackendLocal, storage.OpenConfig{storagePathKey: path})
}
func sourceOpenConfig(source config.Backend) storage.OpenConfig {
cfg := storage.OpenConfig{storagePathKey: source.Path}
if source.Backend == config.BackendSSH {
cfg[sshHostKey] = source.Host
cfg[sshUserKey] = source.User
cfg[sshPortKey] = strconv.Itoa(source.Port)
cfg[sshKeyFileKey] = source.SSH.KeyFile
cfg[sshKnownHostsKey] = source.SSH.KnownHosts
cfg[sshHostKeyPolicyKey] = string(source.SSH.HostKeyPolicy)
}
return cfg
}
func destinationOpenConfig(destination config.Destination) storage.OpenConfig {
cfg := storage.OpenConfig{storagePathKey: destination.Path}
if destination.Backend == config.BackendSSH {
cfg[sshHostKey] = destination.Host
cfg[sshUserKey] = destination.User
cfg[sshPortKey] = strconv.Itoa(destination.Port)
cfg[sshKeyFileKey] = destination.SSH.KeyFile
cfg[sshKnownHostsKey] = destination.SSH.KnownHosts
cfg[sshHostKeyPolicyKey] = string(destination.SSH.HostKeyPolicy)
}
return cfg
}

123
internal/app/backends_test.go
View File

@@ -6,6 +6,8 @@ import (
"testing"
"gitea.maximumdirect.net/eric/distributor/internal/config"
"gitea.maximumdirect.net/eric/distributor/internal/storage"
"gitea.maximumdirect.net/eric/distributor/internal/storage/fake"
)
func TestBackendFactoryOpensLocalSource(t *testing.T) {
@@ -47,14 +49,72 @@ func TestBackendFactoryOpensDirectLocalPath(t *testing.T) {
}
}
func TestBackendFactoryOpensSSHSourceWithRegisteredOpener(t *testing.T) {
factory := &backendFactory{registry: storage.NewRegistry()}
var got storage.OpenConfig
if err := factory.registry.Register(config.BackendSSH, func(ctx context.Context, cfg storage.OpenConfig) (storage.Backend, error) {
got = cfg
return fake.New(), nil
}); err != nil {
t.Fatalf("Register() error = %v", err)
}
backend, err := factory.openSource(context.Background(), config.Backend{
Backend: config.BackendSSH,
Host: "source.example.com",
User: "reports",
Port: 22,
Path: "/reports",
SSH: config.SSH{HostKeyPolicy: config.HostKeyPolicyAcceptNew},
})
if err != nil {
t.Fatalf("openSource() error = %v", err)
}
if backend == nil {
t.Fatal("openSource() backend = nil")
}
if got[sshHostKey] != "source.example.com" || got[storagePathKey] != "/reports" {
t.Fatalf("open config = %#v, want SSH source fields", got)
}
}
func TestBackendFactoryOpensSSHDestinationWithRegisteredOpener(t *testing.T) {
factory := &backendFactory{registry: storage.NewRegistry()}
var got storage.OpenConfig
if err := factory.registry.Register(config.BackendSSH, func(ctx context.Context, cfg storage.OpenConfig) (storage.Backend, error) {
got = cfg
return fake.New(), nil
}); err != nil {
t.Fatalf("Register() error = %v", err)
}
backend, err := factory.openDestination(context.Background(), config.Destination{
Backend: config.BackendSSH,
Host: "destination.example.com",
User: "deploy",
Port: 2222,
Path: "/archive",
SSH: config.SSH{HostKeyPolicy: config.HostKeyPolicyStrict},
})
if err != nil {
t.Fatalf("openDestination() error = %v", err)
}
if backend == nil {
t.Fatal("openDestination() backend = nil")
}
if got[sshHostKey] != "destination.example.com" || got[sshPortKey] != "2222" || got[sshHostKeyPolicyKey] != "strict" {
t.Fatalf("open config = %#v, want SSH destination fields", got)
}
}
func TestBackendFactoryRejectsUnsupportedSource(t *testing.T) {
factory := newBackendFactory()
_, err := factory.openSource(context.Background(), config.Backend{
Backend: config.BackendSSH,
URI: "ssh://reports@example.com:22",
Path: "/reports",
Backend: config.BackendS3,
Endpoint: "https://s3.example.com",
Bucket: "reports",
})
if err == nil || !strings.Contains(err.Error(), "source backend ssh is not implemented for execution") {
if err == nil || !strings.Contains(err.Error(), "source backend s3 is not implemented for execution") {
t.Fatalf("openSource() error = %v, want not implemented", err)
}
}
@@ -70,3 +130,58 @@ func TestBackendFactoryRejectsUnsupportedDestination(t *testing.T) {
t.Fatalf("openDestination() error = %v, want not implemented", err)
}
}
func TestBackendFactoryBuildsSSHSourceOpenConfig(t *testing.T) {
cfg := sourceOpenConfig(config.Backend{
Backend: config.BackendSSH,
Host: "source.example.com",
User: "reports",
Port: 2222,
Path: "/reports",
SSH: config.SSH{
KeyFile: "/home/reports/.ssh/id_ed25519",
KnownHosts: "/home/reports/.ssh/known_hosts",
HostKeyPolicy: config.HostKeyPolicyStrict,
},
})
assertOpenConfig(t, cfg, map[string]string{
storagePathKey: "/reports",
sshHostKey: "source.example.com",
sshUserKey: "reports",
sshPortKey: "2222",
sshKeyFileKey: "/home/reports/.ssh/id_ed25519",
sshKnownHostsKey: "/home/reports/.ssh/known_hosts",
sshHostKeyPolicyKey: "strict",
})
}
func TestBackendFactoryBuildsSSHDestinationOpenConfig(t *testing.T) {
cfg := destinationOpenConfig(config.Destination{
Backend: config.BackendSSH,
Host: "destination.example.com",
User: "deploy",
Port: 22,
Path: "/srv/archive",
SSH: config.SSH{
HostKeyPolicy: config.HostKeyPolicyAcceptNew,
},
})
assertOpenConfig(t, cfg, map[string]string{
storagePathKey: "/srv/archive",
sshHostKey: "destination.example.com",
sshUserKey: "deploy",
sshPortKey: "22",
sshHostKeyPolicyKey: "accept-new",
})
}
func assertOpenConfig(t *testing.T, got map[string]string, want map[string]string) {
t.Helper()
for key, wantValue := range want {
if gotValue := got[key]; gotValue != wantValue {
t.Fatalf("open config %s = %q, want %q", key, gotValue, wantValue)
}
}
}

47
internal/app/run.go
View File

@@ -52,16 +52,23 @@ func runConfig(ctx context.Context, cfg config.Config, options RunOptions) error
}
}
for _, pipeline := range cfg.Pipelines {
if options.Stdout != nil {
if err := writeSSHWarnings(options.Stdout, pipeline); err != nil {
return err
}
}
sourceBackend, err := backends.openSource(ctx, pipeline.Source)
if err != nil {
return fmt.Errorf("pipeline %s: %w", pipeline.ID, err)
}
bundles, err := bundle.Discover(ctx, sourceBackend, "")
if err != nil {
closeBackend(sourceBackend)
return fmt.Errorf("pipeline %s discover source bundles: %w", pipeline.ID, err)
}
if options.Stdout != nil {
if _, err := fmt.Fprintf(options.Stdout, "- pipeline=%s source=%s bundles=%d destinations=%s\n", pipeline.ID, pipeline.Source.Backend, len(bundles), destinationSummary(pipeline.Destinations)); err != nil {
closeBackend(sourceBackend)
return err
}
}
@@ -76,6 +83,13 @@ func runConfig(ctx context.Context, cfg config.Config, options RunOptions) error
}
continue
}
closeDestination := true
deferCloseDestination := func() {
if closeDestination {
closeBackend(destinationBackend)
closeDestination = false
}
}
req := publish.Request{
PipelineID: pipeline.ID,
DestinationID: destination.ID,
@@ -97,6 +111,7 @@ func runConfig(ctx context.Context, cfg config.Config, options RunOptions) error
writePlanLine(options.Stdout, plan, err)
}
if err != nil {
deferCloseDestination()
failures.add(pipeline.ID, destination.ID, storage.DisplayPath(sourceBundle.RootRelativePath), err)
summary.recordFailure()
continue
@@ -104,20 +119,24 @@ func runConfig(ctx context.Context, cfg config.Config, options RunOptions) error
summary.recordPlan(plan.Action)
if !options.DryRun {
if err := publish.Execute(ctx, req, plan); err != nil {
deferCloseDestination()
failures.add(pipeline.ID, destination.ID, storage.DisplayPath(sourceBundle.RootRelativePath), err)
summary.recordFailure()
continue
}
if shouldNotify(plan.Action) {
if err := notifier.Notify(ctx, notifyEvent(plan)); err != nil {
deferCloseDestination()
failures.add(pipeline.ID, destination.ID, storage.DisplayPath(sourceBundle.RootRelativePath), err)
summary.recordFailure()
continue
}
}
}
deferCloseDestination()
}
}
closeBackend(sourceBackend)
}
if options.Stdout != nil {
if _, err := fmt.Fprintln(options.Stdout, summary.Line()); err != nil {
@@ -130,6 +149,18 @@ func runConfig(ctx context.Context, cfg config.Config, options RunOptions) error
return nil
}
type closeableBackend interface {
Close() error
}
func closeBackend(backend storage.Backend) {
closeable, ok := backend.(closeableBackend)
if !ok {
return
}
_ = closeable.Close()
}
func writePlanLine(w io.Writer, plan publish.Plan, planErr error) {
if w == nil {
return
@@ -174,6 +205,22 @@ func destinationSummary(destinations []config.Destination) string {
return strings.Join(ids, ",")
}
func writeSSHWarnings(w io.Writer, pipeline config.Pipeline) error {
if pipeline.Source.Backend == config.BackendSSH && pipeline.Source.SSH.HostKeyPolicy == config.HostKeyPolicyOff {
if _, err := fmt.Fprintf(w, "Warning: pipeline=%s source host_key_policy=off disables SSH host key checking\n", pipeline.ID); err != nil {
return err
}
}
for _, destination := range pipeline.Destinations {
if destination.Backend == config.BackendSSH && destination.SSH.HostKeyPolicy == config.HostKeyPolicyOff {
if _, err := fmt.Fprintf(w, "Warning: pipeline=%s destination=%s host_key_policy=off disables SSH host key checking\n", pipeline.ID, destination.ID); err != nil {
return err
}
}
}
return nil
}
func shouldNotify(action publish.Action) bool {
return action == publish.ActionPublishNew || action == publish.ActionReplaceOlder
}

29
internal/app/run_test.go
View File

@@ -11,6 +11,7 @@ import (
"time"
"gitea.maximumdirect.net/eric/distributor/internal/bundle"
"gitea.maximumdirect.net/eric/distributor/internal/config"
"gitea.maximumdirect.net/eric/distributor/internal/notify"
"gitea.maximumdirect.net/eric/distributor/internal/state"
"gitea.maximumdirect.net/eric/distributor/internal/storage"
@@ -46,6 +47,34 @@ func TestRunDryRunPrintsConfigSummary(t *testing.T) {
}
}
func TestWriteSSHWarningsReportsInsecureHostKeyPolicy(t *testing.T) {
var stdout bytes.Buffer
err := writeSSHWarnings(&stdout, config.Pipeline{
ID: "reports",
Source: config.Backend{
Backend: config.BackendSSH,
SSH: config.SSH{HostKeyPolicy: config.HostKeyPolicyOff},
},
Destinations: []config.Destination{{
ID: "archive",
Backend: config.BackendSSH,
SSH: config.SSH{HostKeyPolicy: config.HostKeyPolicyOff},
}},
})
if err != nil {
t.Fatalf("writeSSHWarnings() error = %v", err)
}
output := stdout.String()
for _, want := range []string{
"pipeline=reports source host_key_policy=off disables SSH host key checking",
"pipeline=reports destination=archive host_key_policy=off disables SSH host key checking",
} {
if !strings.Contains(output, want) {
t.Fatalf("output = %q, want substring %q", output, want)
}
}
}
func TestRunPublishesNewLocalBundle(t *testing.T) {
sourceRoot := t.TempDir()
destinationRoot := t.TempDir()