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

Fixes and cleanup after implementation of the S3 and SSH roadmap

Browse Source
This commit is contained in:
Eric Rakestraw
2026-05-31 14:33:17 -05:00
parent 1d71a151cc
commit 7eed1a26ae
13 changed files with 250 additions and 59 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
docs/cli.md
View File

@@ -38,7 +38,7 @@ All subcommands:
`run` flags:
- `--config <path>`: 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.

2
docs/config.md
View File

@@ -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.

4
docs/operations.md
View File

@@ -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.

37
docs/roadmap/implementation.md
View File

@@ -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.

7
internal/adapters/s3/backend.go
View File

@@ -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
}
}

57
internal/adapters/s3/backend_test.go
View File

@@ -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)

3
internal/adapters/ssh/hostkeys.go
View File

@@ -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)
}

51
internal/adapters/ssh/hostkeys_test.go
View File

@@ -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") {

15
internal/adapters/ssh/options.go
View File

@@ -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) {

36
internal/app/backends.go
View File

@@ -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
}

55
internal/app/backends_test.go
View File

@@ -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{

1
internal/app/run.go
View File

@@ -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 {

39
internal/app/run_test.go
View File

@@ -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()