From 7eed1a26ae19a3a762ef840475901eca9d8b9d3f Mon Sep 17 00:00:00 2001 From: Eric Rakestraw Date: Sun, 31 May 2026 14:33:17 -0500 Subject: [PATCH] Fixes and cleanup after implementation of the S3 and SSH roadmap --- docs/cli.md | 2 +- docs/config.md | 2 +- docs/operations.md | 4 +- docs/roadmap/implementation.md | 37 +---------------- internal/adapters/s3/backend.go | 7 ++-- internal/adapters/s3/backend_test.go | 57 ++++++++++++++++++++++++++ internal/adapters/ssh/hostkeys.go | 3 ++ internal/adapters/ssh/hostkeys_test.go | 51 +++++++++++++++++++++++ internal/adapters/ssh/options.go | 15 +++---- internal/app/backends.go | 36 ++++++++++++---- internal/app/backends_test.go | 55 +++++++++++++++++++++++++ internal/app/run.go | 1 + internal/app/run_test.go | 39 ++++++++++++++++++ 13 files changed, 250 insertions(+), 59 deletions(-) diff --git a/docs/cli.md b/docs/cli.md index 8e6a389..571ccd1 100644 --- a/docs/cli.md +++ b/docs/cli.md @@ -38,7 +38,7 @@ All subcommands: `run` flags: - `--config `: config file to load. If omitted, `run` uses `/usr/local/etc/distributor/config.yml`. -- `--dry-run`: load config, discover bundles, inspect destination state, print planned actions and final status, and do not write files. +- `--dry-run`: load config, discover bundles, inspect destination state, print planned actions and final status, and do not write output files, destination state, or SSH `known_hosts` entries. - `--force`: allow explicit destructive replacement for supported conflict cases in this run only. `run` does not accept positional arguments. `validate` and `inspect` accept at most one path; omitting the path returns a required-path error. diff --git a/docs/config.md b/docs/config.md index 19c5bba..50b2a59 100644 --- a/docs/config.md +++ b/docs/config.md @@ -134,7 +134,7 @@ Authentication uses SSH agent identities first when `SSH_AUTH_SOCK` is set, then Host key policies: - `strict`, `true`, and `"true"` require a matching known host key. -- `accept-new` accepts and persists a new host key, but fails if an existing key changed. +- `accept-new` accepts and persists a new host key, but fails if an existing key changed. During `run --dry-run`, new host keys are accepted only for the current connection and are not persisted. - `off`, `false`, and `"false"` disable host key checking and are insecure. `accept-new` and `strict` use `known_hosts` when configured. If omitted, distributor uses the current service user's default OpenSSH `known_hosts` path where practical. `accept-new` fails when it needs to persist a new host key and no writable `known_hosts` path is available. It does not create a missing parent `.ssh` directory. diff --git a/docs/operations.md b/docs/operations.md index dee345a..0efe75f 100644 --- a/docs/operations.md +++ b/docs/operations.md @@ -72,7 +72,7 @@ Do not edit `.distributor.json` by hand during normal operation. If it is missin ## Dry Runs -`--dry-run` loads and validates config, discovers source bundles, inspects destination state, plans outputs, and prints summary lines. It does not write output files or destination state. +`--dry-run` loads and validates config, discovers source bundles, inspects destination state, plans outputs, and prints summary lines. It does not write output files, destination state, or SSH `known_hosts` entries. Dry-run output is useful before publishing to confirm actions such as `publish_new`, `replace_older`, `force_replace`, `skip_same`, and `skip_destination_newer`. @@ -117,7 +117,7 @@ SSH execution uses SFTP over `golang.org/x/crypto/ssh` and `github.com/pkg/sftp` Configure `ssh_key_file`, an SSH agent, or both. Agent identities are attempted first, followed by the configured key file. YAML password authentication is not supported. -The default host key policy is `accept-new`. New host keys are written to `known_hosts` when the file path is writable. Changed host keys are fatal for both `strict` and `accept-new`. The `off` policy disables host key checking and `run` prints a warning when stdout is enabled. +The default host key policy is `accept-new`. New host keys are written to `known_hosts` when the file path is writable. During `--dry-run`, unknown host keys may be accepted for the current connection but are not written to `known_hosts`; a later non-dry-run may persist the same key. Changed host keys are fatal for both `strict` and `accept-new`. The `off` policy disables host key checking and `run` prints a warning when stdout is enabled. Recovery boundaries are the same as local storage: replacement deletes only managed output paths recorded in `.distributor.json` plus the state file, and failed writes are cleaned up where practical. Distributor never performs broad recursive remote deletion. diff --git a/docs/roadmap/implementation.md b/docs/roadmap/implementation.md index d213eb9..4aa509f 100644 --- a/docs/roadmap/implementation.md +++ b/docs/roadmap/implementation.md @@ -24,41 +24,6 @@ it is implemented. `distributor` is ready for routine use against producer pipelines using the implemented local, SSH/SFTP, and S3-compatible backends. -Implemented capabilities: - -- standard-library CLI commands for `version`, `run`, `validate`, and - `inspect`; -- YAML config loading, defaults, known-field rejection, and validation; -- local, SSH/SFTP, and S3-compatible source and destination backends; -- structured SSH/SFTP configuration with agent and key-file authentication, - known-hosts handling, and backend-rooted path confinement; -- S3-compatible configuration with endpoint, region, addressing mode, TLS - controls, credential environment references, streaming reads and writes, - listing, metadata, and managed deletion; -- top-level `secrets.directory` loading for credential values, with process - environment precedence and without mutating the process environment; -- backend-rooted storage interfaces with typed errors, safe logical paths, - traversal, collision checks, managed deletion, and explicit forced - replacement; -- source bundle discovery, manifest parsing, RFC3339 timestamp handling, - duplicate path checks, path safety checks, symlink rejection, per-file digest - validation, and bundle digest validation; -- destination `.distributor.json` state parsing, validation, output metadata, - and source comparison; -- publication of source files, Markdown sidecar HTML, or both; -- destination output collision detection before writes; -- managed replacement for older destination state; -- conservative conflict failures by default for unmanaged destination content - and conflicting managed destination content; -- explicit forced replacement with `run --force`, limited to the configured - destination bundle path; -- deterministic dry-run output and final run summaries; -- deterministic sequential fan-out with aggregated destination failures; -- cleanup of outputs written during failed publish attempts where practical; -- no-op notification hook after successful publish or replacement; -- current user, operator, internal, integration, and development documentation - for implemented behavior. - ## Active Roadmap There are no active implementation items in this roadmap. @@ -80,7 +45,7 @@ until a roadmap entry is selected and implemented: - compatibility parsing for legacy SSH URI config; - broad recursive destination deletion outside managed bundle paths; - concurrent fan-out publishing; -- resumable multipart S3 uploads; +- streaming, resumable, or multipart S3 uploads; - cloud-provider-specific IAM integration docs; - repository-managed packaging, release, and deployment automation. diff --git a/internal/adapters/s3/backend.go b/internal/adapters/s3/backend.go index 1ffa17a..b86a61d 100644 --- a/internal/adapters/s3/backend.go +++ b/internal/adapters/s3/backend.go @@ -208,9 +208,10 @@ func (b *Backend) Walk(ctx context.Context, logicalPrefix string, opts storage.W } else if err != nil { return err } - return nil - } - if !storage.IsNotFound(err) { + if opts.Limit > 0 && visited >= opts.Limit { + return nil + } + } else if !storage.IsNotFound(err) { return err } } diff --git a/internal/adapters/s3/backend_test.go b/internal/adapters/s3/backend_test.go index 9ed46b9..293ffe8 100644 --- a/internal/adapters/s3/backend_test.go +++ b/internal/adapters/s3/backend_test.go @@ -66,6 +66,63 @@ func TestStatRequiresExactObject(t *testing.T) { } } +func TestWalkIncludesExactObjectAndDescendants(t *testing.T) { + client := newFakeClient(map[string]string{ + "root/bundle": "marker", + "root/bundle/report.md": "report", + }) + backend := newTestBackend(t, "root", client) + entries, err := storage.List(context.Background(), backend, "bundle", storage.WalkOptions{Recursive: true}) + if err != nil { + t.Fatalf("List() error = %v", err) + } + paths := entryPaths(entries) + if got, want := paths, []string{"bundle", "bundle/report.md"}; !equalStrings(got, want) { + t.Fatalf("paths = %v, want %v", got, want) + } +} + +func TestWalkExactObjectHonorsLimitBeforeListingDescendants(t *testing.T) { + client := newFakeClient(map[string]string{ + "root/bundle": "marker", + "root/bundle/report.md": "report", + }) + backend := newTestBackend(t, "root", client) + var entries []storage.Entry + err := backend.Walk(context.Background(), "bundle", storage.WalkOptions{Recursive: true, Limit: 1}, func(entry storage.Entry) error { + entries = append(entries, entry) + return nil + }) + if err != nil { + t.Fatalf("Walk() error = %v", err) + } + paths := entryPaths(entries) + if got, want := paths, []string{"bundle"}; !equalStrings(got, want) { + t.Fatalf("paths = %v, want %v", got, want) + } + if len(client.tokens) != 0 { + t.Fatalf("list calls = %d, want none", len(client.tokens)) + } +} + +func TestHasAnyWithExactObjectStopsBeforeListingDescendants(t *testing.T) { + client := newFakeClient(map[string]string{ + "root/bundle": "marker", + "root/bundle/report.md": "report", + }) + backend := newTestBackend(t, "root", client) + found, err := backend.HasAny(context.Background(), "bundle") + if err != nil { + t.Fatalf("HasAny() error = %v", err) + } + if !found { + t.Fatal("HasAny() = false, want true") + } + if len(client.tokens) != 0 { + t.Fatalf("list calls = %d, want none", len(client.tokens)) + } +} + func TestWriteFromChecksOverwriteBeforePut(t *testing.T) { client := newFakeClient(map[string]string{"root/report.md": "old"}) backend := newTestBackend(t, "root", client) diff --git a/internal/adapters/ssh/hostkeys.go b/internal/adapters/ssh/hostkeys.go index 828a5a7..f927cbe 100644 --- a/internal/adapters/ssh/hostkeys.go +++ b/internal/adapters/ssh/hostkeys.go @@ -54,6 +54,9 @@ func acceptNewHostKeyCallback(options Options) (cryptossh.HostKeyCallback, error return fmt.Errorf("host key for %s has changed: %w", hostname, err) } } + if options.ReadOnlyKnownHosts { + return nil + } if options.KnownHosts == "" { return fmt.Errorf("host key for %s is unknown and no writable known_hosts path is available", hostname) } diff --git a/internal/adapters/ssh/hostkeys_test.go b/internal/adapters/ssh/hostkeys_test.go index 7ad756c..0a6a193 100644 --- a/internal/adapters/ssh/hostkeys_test.go +++ b/internal/adapters/ssh/hostkeys_test.go @@ -36,6 +36,25 @@ func TestAcceptNewHostKeyCallbackPersistsUnknownHost(t *testing.T) { } } +func TestAcceptNewHostKeyCallbackReadOnlyDoesNotPersistUnknownHost(t *testing.T) { + key := testPublicKey(t) + knownHosts := filepath.Join(t.TempDir(), "known_hosts") + callback, err := acceptNewHostKeyCallback(Options{ + KnownHosts: knownHosts, + ReadOnlyKnownHosts: true, + }) + if err != nil { + t.Fatalf("acceptNewHostKeyCallback() error = %v", err) + } + + if err := callback("example.com:22", &net.TCPAddr{IP: net.ParseIP("192.0.2.10"), Port: 22}, key); err != nil { + t.Fatalf("callback() error = %v", err) + } + if _, err := os.Stat(knownHosts); !os.IsNotExist(err) { + t.Fatalf("known_hosts stat error = %v, want not exist", err) + } +} + func TestAcceptNewHostKeyCallbackRejectsChangedHostKey(t *testing.T) { first := testPublicKey(t) second := testPublicKey(t) @@ -54,6 +73,27 @@ func TestAcceptNewHostKeyCallbackRejectsChangedHostKey(t *testing.T) { } } +func TestAcceptNewHostKeyCallbackReadOnlyRejectsChangedHostKey(t *testing.T) { + first := testPublicKey(t) + second := testPublicKey(t) + knownHosts := filepath.Join(t.TempDir(), "known_hosts") + if err := os.WriteFile(knownHosts, []byte(knownhosts.Line([]string{knownhosts.Normalize("example.com:22")}, first)+"\n"), 0o600); err != nil { + t.Fatalf("write known_hosts: %v", err) + } + callback, err := acceptNewHostKeyCallback(Options{ + KnownHosts: knownHosts, + ReadOnlyKnownHosts: true, + }) + if err != nil { + t.Fatalf("acceptNewHostKeyCallback() error = %v", err) + } + + err = callback("example.com:22", &net.TCPAddr{IP: net.ParseIP("192.0.2.10"), Port: 22}, second) + if err == nil || !strings.Contains(err.Error(), "has changed") { + t.Fatalf("callback() error = %v, want changed host key", err) + } +} + func TestAcceptNewHostKeyCallbackRequiresWritableKnownHostsForUnknownHost(t *testing.T) { callback, err := acceptNewHostKeyCallback(Options{}) if err != nil { @@ -66,6 +106,17 @@ func TestAcceptNewHostKeyCallbackRequiresWritableKnownHostsForUnknownHost(t *tes } } +func TestAcceptNewHostKeyCallbackReadOnlyAllowsMissingKnownHosts(t *testing.T) { + callback, err := acceptNewHostKeyCallback(Options{ReadOnlyKnownHosts: true}) + if err != nil { + t.Fatalf("acceptNewHostKeyCallback() error = %v", err) + } + + if err := callback("example.com:22", &net.TCPAddr{IP: net.ParseIP("192.0.2.10"), Port: 22}, testPublicKey(t)); err != nil { + t.Fatalf("callback() error = %v", err) + } +} + func TestStrictHostKeyCallbackRequiresKnownHosts(t *testing.T) { _, err := hostKeyCallback(Options{HostKeyPolicy: HostKeyPolicyStrict}) if err == nil || !strings.Contains(err.Error(), "known_hosts is required") { diff --git a/internal/adapters/ssh/options.go b/internal/adapters/ssh/options.go index c95a2c9..ef208c4 100644 --- a/internal/adapters/ssh/options.go +++ b/internal/adapters/ssh/options.go @@ -20,13 +20,14 @@ const ( type HostKeyPolicy string type Options struct { - Host string - User string - Port int - Root string - KeyFile string - KnownHosts string - HostKeyPolicy HostKeyPolicy + Host string + User string + Port int + Root string + KeyFile string + KnownHosts string + HostKeyPolicy HostKeyPolicy + ReadOnlyKnownHosts bool } func (o Options) normalized() (Options, error) { diff --git a/internal/app/backends.go b/internal/app/backends.go index 5ceb2ba..33d5a46 100644 --- a/internal/app/backends.go +++ b/internal/app/backends.go @@ -21,6 +21,7 @@ const ( sshKeyFileKey = "ssh_key_file" sshKnownHostsKey = "known_hosts" sshHostKeyPolicyKey = "host_key_policy" + sshReadOnlyHostsKey = "read_only_known_hosts" s3EndpointKey = "endpoint" s3BucketKey = "bucket" s3PrefixKey = "prefix" @@ -31,8 +32,9 @@ const ( ) type backendFactory struct { - registry *storage.Registry - environment config.Environment + registry *storage.Registry + environment config.Environment + readOnlyKnownHosts bool } func newBackendFactory() *backendFactory { @@ -52,14 +54,22 @@ func newBackendFactoryWithEnvironment(environment config.Environment) *backendFa if err != nil { return nil, fmt.Errorf("ssh port: %w", err) } + readOnlyKnownHosts := false + if raw := cfg[sshReadOnlyHostsKey]; raw != "" { + readOnlyKnownHosts, err = strconv.ParseBool(raw) + if err != nil { + return nil, fmt.Errorf("ssh read_only_known_hosts: %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]), + Host: cfg[sshHostKey], + User: cfg[sshUserKey], + Port: port, + Root: cfg[storagePathKey], + KeyFile: cfg[sshKeyFileKey], + KnownHosts: cfg[sshKnownHostsKey], + HostKeyPolicy: sshadapter.HostKeyPolicy(cfg[sshHostKeyPolicyKey]), + ReadOnlyKnownHosts: readOnlyKnownHosts, }) }) _ = registry.Register(config.BackendS3, func(ctx context.Context, cfg storage.OpenConfig) (storage.Backend, error) { @@ -117,6 +127,9 @@ func (f *backendFactory) sourceOpenConfig(source config.Backend) (storage.OpenCo return nil, err } } + if source.Backend == config.BackendSSH { + cfg[sshReadOnlyHostsKey] = strconv.FormatBool(f.readOnlyKnownHosts) + } return cfg, nil } @@ -127,6 +140,9 @@ func (f *backendFactory) destinationOpenConfig(destination config.Destination) ( return nil, err } } + if destination.Backend == config.BackendSSH { + cfg[sshReadOnlyHostsKey] = strconv.FormatBool(f.readOnlyKnownHosts) + } return cfg, nil } @@ -156,6 +172,7 @@ func sourceOpenConfig(source config.Backend) storage.OpenConfig { cfg[sshKeyFileKey] = source.SSH.KeyFile cfg[sshKnownHostsKey] = source.SSH.KnownHosts cfg[sshHostKeyPolicyKey] = string(source.SSH.HostKeyPolicy) + cfg[sshReadOnlyHostsKey] = "false" } return cfg } @@ -169,6 +186,7 @@ func destinationOpenConfig(destination config.Destination) storage.OpenConfig { cfg[sshKeyFileKey] = destination.SSH.KeyFile cfg[sshKnownHostsKey] = destination.SSH.KnownHosts cfg[sshHostKeyPolicyKey] = string(destination.SSH.HostKeyPolicy) + cfg[sshReadOnlyHostsKey] = "false" } return cfg } diff --git a/internal/app/backends_test.go b/internal/app/backends_test.go index 026e041..443aaf9 100644 --- a/internal/app/backends_test.go +++ b/internal/app/backends_test.go @@ -107,6 +107,61 @@ func TestBackendFactoryOpensSSHDestinationWithRegisteredOpener(t *testing.T) { } } +func TestBackendFactorySetsReadOnlyKnownHostsForDryRunSSH(t *testing.T) { + factory := &backendFactory{ + registry: storage.NewRegistry(), + readOnlyKnownHosts: true, + } + 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) + } + + _, err := factory.openDestination(context.Background(), config.Destination{ + Backend: config.BackendSSH, + Host: "destination.example.com", + User: "deploy", + Port: 22, + Path: "/archive", + SSH: config.SSH{HostKeyPolicy: config.HostKeyPolicyAcceptNew}, + }) + if err != nil { + t.Fatalf("openDestination() error = %v", err) + } + if got[sshReadOnlyHostsKey] != "true" { + t.Fatalf("open config %s = %q, want true", sshReadOnlyHostsKey, got[sshReadOnlyHostsKey]) + } +} + +func TestBackendFactoryUsesPersistentKnownHostsByDefault(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) + } + + _, err := factory.openDestination(context.Background(), config.Destination{ + Backend: config.BackendSSH, + Host: "destination.example.com", + User: "deploy", + Port: 22, + Path: "/archive", + SSH: config.SSH{HostKeyPolicy: config.HostKeyPolicyAcceptNew}, + }) + if err != nil { + t.Fatalf("openDestination() error = %v", err) + } + if got[sshReadOnlyHostsKey] != "false" { + t.Fatalf("open config %s = %q, want false", sshReadOnlyHostsKey, got[sshReadOnlyHostsKey]) + } +} + func TestBackendFactoryRejectsUnsupportedSource(t *testing.T) { factory := newBackendFactory() _, err := factory.openSource(context.Background(), config.Backend{ diff --git a/internal/app/run.go b/internal/app/run.go index 8698d55..9779977 100644 --- a/internal/app/run.go +++ b/internal/app/run.go @@ -61,6 +61,7 @@ func runConfigWithBackendFactory(ctx context.Context, cfg config.Config, options } } backends := provider(secretLoad.Environment) + backends.readOnlyKnownHosts = options.DryRun transforms := newTransformRegistry() if options.Stdout != nil { if _, err := fmt.Fprintf(options.Stdout, "Configured pipelines: %d\n", len(cfg.Pipelines)); err != nil { diff --git a/internal/app/run_test.go b/internal/app/run_test.go index ab690be..d77feb5 100644 --- a/internal/app/run_test.go +++ b/internal/app/run_test.go @@ -49,6 +49,45 @@ func TestRunDryRunPrintsConfigSummary(t *testing.T) { } } +func TestRunDryRunUsesReadOnlySSHKnownHosts(t *testing.T) { + sourceRoot := t.TempDir() + writeSourceBundle(t, sourceRoot, "", testBundleOptions{}) + cfg := config.Config{Pipelines: []config.Pipeline{{ + ID: "reports", + Source: config.Backend{Backend: config.BackendLocal, Path: sourceRoot}, + Destinations: []config.Destination{{ + ID: "archive", + Backend: config.BackendSSH, + Host: "destination.example.com", + Path: "/archive", + }}, + }}} + config.ApplyDefaults(&cfg) + var got storage.OpenConfig + provider := func(environment config.Environment) *backendFactory { + registry := storage.NewRegistry() + if err := registry.Register(config.BackendLocal, func(ctx context.Context, cfg storage.OpenConfig) (storage.Backend, error) { + return local.New(cfg[storagePathKey]) + }); err != nil { + t.Fatalf("register local backend: %v", err) + } + if err := 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 ssh backend: %v", err) + } + return &backendFactory{registry: registry, environment: environment} + } + + if err := runConfigWithBackendFactory(context.Background(), cfg, RunOptions{DryRun: true}, provider); err != nil { + t.Fatalf("runConfigWithBackendFactory() error = %v", err) + } + if got[sshReadOnlyHostsKey] != "true" { + t.Fatalf("open config %s = %q, want true", sshReadOnlyHostsKey, got[sshReadOnlyHostsKey]) + } +} + func TestRunLoadsSecretsBeforeOpeningBackends(t *testing.T) { sourceRoot := filepath.Join(t.TempDir(), "missing-source") destinationRoot := t.TempDir()