Normalize backend config validation

internal/config/backend_view.go

@@ -0,0 +1,50 @@
+package config
+
+type backendView struct {
+	Backend   string
+	Host      string
+	User      string
+	Port      int
+	Path      string
+	Endpoint  string
+	Bucket    string
+	Prefix    string
+	Region    string
+	ForcePath *bool
+	Creds     Credentials
+	SSH       SSH
+}
+
+func backendViewFromSource(source Backend) backendView {
+	return backendView{
+		Backend:   source.Backend,
+		Host:      source.Host,
+		User:      source.User,
+		Port:      source.Port,
+		Path:      source.Path,
+		Endpoint:  source.Endpoint,
+		Bucket:    source.Bucket,
+		Prefix:    source.Prefix,
+		Region:    source.Region,
+		ForcePath: source.ForcePath,
+		Creds:     source.Creds,
+		SSH:       source.SSH,
+	}
+}
+
+func backendViewFromDestination(destination Destination) backendView {
+	return backendView{
+		Backend:   destination.Backend,
+		Host:      destination.Host,
+		User:      destination.User,
+		Port:      destination.Port,
+		Path:      destination.Path,
+		Endpoint:  destination.Endpoint,
+		Bucket:    destination.Bucket,
+		Prefix:    destination.Prefix,
+		Region:    destination.Region,
+		ForcePath: destination.ForcePath,
+		Creds:     destination.Creds,
+		SSH:       destination.SSH,
+	}
+}

internal/config/backend_view_test.go

@@ -0,0 +1,187 @@
+package config
+
+import "testing"
+
+func TestBackendViewsPreserveEquivalentStorageFields(t *testing.T) {
+	forcePathStyle := false
+	source := Backend{
+		Backend:   BackendS3,
+		Host:      "storage.example.com",
+		User:      "reports",
+		Port:      2222,
+		Path:      "/reports",
+		Endpoint:  "https://s3.example.com",
+		Bucket:    "source",
+		Prefix:    "incoming",
+		Region:    "us-west-2",
+		ForcePath: &forcePathStyle,
+		Creds: Credentials{
+			AccessKeyIDEnv:     "ACCESS_KEY_ID",
+			SecretAccessKeyEnv: "SECRET_ACCESS_KEY",
+		},
+		SSH: SSH{
+			KeyFile:       "/home/reports/.ssh/id_ed25519",
+			KnownHosts:    "/home/reports/.ssh/known_hosts",
+			HostKeyPolicy: HostKeyPolicyStrict,
+		},
+	}
+	destination := Destination{
+		Backend:   source.Backend,
+		Host:      source.Host,
+		User:      source.User,
+		Port:      source.Port,
+		Path:      source.Path,
+		Endpoint:  source.Endpoint,
+		Bucket:    source.Bucket,
+		Prefix:    source.Prefix,
+		Region:    source.Region,
+		ForcePath: source.ForcePath,
+		Creds:     source.Creds,
+		SSH:       source.SSH,
+	}
+
+	sourceView := backendViewFromSource(source)
+	destinationView := backendViewFromDestination(destination)
+
+	if sourceView != destinationView {
+		t.Fatalf("source view = %#v, destination view = %#v, want equivalent storage fields", sourceView, destinationView)
+	}
+}
+
+func TestBackendViewValidationKeepsHTTPUploadSourceOnly(t *testing.T) {
+	cfg := Config{Pipelines: []Pipeline{{
+		ID: "reports",
+		Source: Backend{
+			Backend: BackendHTTPUpload,
+			Upload:  HTTPUpload{TokenEnv: "UPLOAD_TOKEN"},
+		},
+		Destinations: []Destination{{
+			ID:      "archive",
+			Backend: BackendHTTPUpload,
+		}},
+	}}}
+	ApplyDefaults(&cfg)
+
+	err := Validate(cfg)
+	if err == nil {
+		t.Fatal("Validate() error = nil, want destination http_upload error")
+	}
+	if got, want := err.Error(), "pipelines[0].destinations[0].backend http_upload is only supported for sources"; got != want {
+		t.Fatalf("Validate() error = %q, want %q", got, want)
+	}
+}
+
+func TestBackendViewValidationAppliesStorageRulesToSourcesAndDestinations(t *testing.T) {
+	forcePathStyle := false
+	tests := []struct {
+		name        string
+		source      Backend
+		destination Destination
+		wantErr     bool
+	}{
+		{
+			name: "local valid",
+			source: Backend{
+				Backend: BackendLocal,
+				Path:    "/source",
+			},
+			destination: Destination{
+				Backend: BackendLocal,
+				Path:    "/destination",
+			},
+		},
+		{
+			name: "local missing path",
+			source: Backend{
+				Backend: BackendLocal,
+			},
+			destination: Destination{
+				Backend: BackendLocal,
+			},
+			wantErr: true,
+		},
+		{
+			name: "ssh valid",
+			source: Backend{
+				Backend: BackendSSH,
+				Host:    "source.example.com",
+				Port:    22,
+				Path:    "/source",
+				SSH:     SSH{HostKeyPolicy: HostKeyPolicyAcceptNew},
+			},
+			destination: Destination{
+				Backend: BackendSSH,
+				Host:    "destination.example.com",
+				Port:    22,
+				Path:    "/destination",
+				SSH:     SSH{HostKeyPolicy: HostKeyPolicyAcceptNew},
+			},
+		},
+		{
+			name: "ssh missing host",
+			source: Backend{
+				Backend: BackendSSH,
+				Port:    22,
+				Path:    "/source",
+				SSH:     SSH{HostKeyPolicy: HostKeyPolicyAcceptNew},
+			},
+			destination: Destination{
+				Backend: BackendSSH,
+				Port:    22,
+				Path:    "/destination",
+				SSH:     SSH{HostKeyPolicy: HostKeyPolicyAcceptNew},
+			},
+			wantErr: true,
+		},
+		{
+			name: "s3 valid",
+			source: Backend{
+				Backend:   BackendS3,
+				Endpoint:  "https://s3.example.com",
+				Bucket:    "source",
+				Prefix:    "incoming",
+				Region:    DefaultS3Region,
+				ForcePath: &forcePathStyle,
+			},
+			destination: Destination{
+				Backend:   BackendS3,
+				Endpoint:  "https://s3.example.com",
+				Bucket:    "destination",
+				Prefix:    "archive",
+				Region:    DefaultS3Region,
+				ForcePath: &forcePathStyle,
+			},
+		},
+		{
+			name: "s3 partial credentials",
+			source: Backend{
+				Backend:  BackendS3,
+				Endpoint: "https://s3.example.com",
+				Bucket:   "source",
+				Region:   DefaultS3Region,
+				Creds:    Credentials{AccessKeyIDEnv: "ACCESS_KEY_ID"},
+			},
+			destination: Destination{
+				Backend:  BackendS3,
+				Endpoint: "https://s3.example.com",
+				Bucket:   "destination",
+				Region:   DefaultS3Region,
+				Creds:    Credentials{AccessKeyIDEnv: "ACCESS_KEY_ID"},
+			},
+			wantErr: true,
+		},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			sourceErrors := validateBackend(nil, "source", backendViewFromSource(tt.source))
+			destinationErrors := validateBackend(nil, "destination", backendViewFromDestination(tt.destination))
+			if got := len(sourceErrors) > 0; got != tt.wantErr {
+				t.Fatalf("source validation errors = %#v, wantErr %t", sourceErrors, tt.wantErr)
+			}
+			if got := len(destinationErrors) > 0; got != tt.wantErr {
+				t.Fatalf("destination validation errors = %#v, wantErr %t", destinationErrors, tt.wantErr)
+			}
+		})
+	}
+}

internal/config/defaults.go

@@ -134,30 +134,24 @@ func duration(value Duration) *Duration {
 }
 
 func applyBackendDefaults(backend *Backend) {
-	if backend.Backend == BackendSSH {
+	applyStorageBackendDefaults(backend.Backend, &backend.Port, &backend.SSH, &backend.Region, &backend.Prefix, &backend.ForcePath)
-		if backend.Port == 0 {
-			backend.Port = 22
-		}
-		if backend.SSH.HostKeyPolicy == "" {
-			backend.SSH.HostKeyPolicy = HostKeyPolicyAcceptNew
-		}
-	}
-	if backend.Backend == BackendS3 {
-		applyS3Defaults(&backend.Region, &backend.Prefix, &backend.ForcePath)
-	}
 }
 
 func applyDestinationDefaults(destination *Destination) {
-	if destination.Backend == BackendSSH {
+	applyStorageBackendDefaults(destination.Backend, &destination.Port, &destination.SSH, &destination.Region, &destination.Prefix, &destination.ForcePath)
-		if destination.Port == 0 {
+}
-			destination.Port = 22
+
+func applyStorageBackendDefaults(backend string, port *int, ssh *SSH, region, prefix *string, forcePath **bool) {
+	if backend == BackendSSH {
+		if *port == 0 {
+			*port = 22
 		}
-		if destination.SSH.HostKeyPolicy == "" {
+		if ssh.HostKeyPolicy == "" {
-			destination.SSH.HostKeyPolicy = HostKeyPolicyAcceptNew
+			ssh.HostKeyPolicy = HostKeyPolicyAcceptNew
 		}
 	}
-	if destination.Backend == BackendS3 {
+	if backend == BackendS3 {
-		applyS3Defaults(&destination.Region, &destination.Prefix, &destination.ForcePath)
+		applyS3Defaults(region, prefix, forcePath)
 	}
 }
 

internal/config/validate.go

@@ -100,7 +100,7 @@ func validateSourceBackend(errs ValidationErrors, context string, backend Backen
 	if backend.Backend == BackendHTTPUpload {
 		return validateHTTPUploadSource(errs, context, backend.Upload)
 	}
-	return validateBackend(errs, context, backend.Backend, backend.Host, backend.Port, backend.Path, backend.Endpoint, backend.Bucket, backend.Prefix, backend.SSH.HostKeyPolicy, backend.Creds)
+	return validateBackend(errs, context, backendViewFromSource(backend))
 }
 
 func validateDestinationBackend(errs ValidationErrors, context string, destination Destination) ValidationErrors {
@@ -108,7 +108,7 @@ func validateDestinationBackend(errs ValidationErrors, context string, destinati
 		errs = append(errs, context+".backend "+BackendHTTPUpload+" is only supported for sources")
 		return errs
 	}
-	return validateBackend(errs, context, destination.Backend, destination.Host, destination.Port, destination.Path, destination.Endpoint, destination.Bucket, destination.Prefix, destination.SSH.HostKeyPolicy, destination.Creds)
+	return validateBackend(errs, context, backendViewFromDestination(destination))
 }
 
 func validateHTTPUploadSource(errs ValidationErrors, context string, upload HTTPUpload) ValidationErrors {
@@ -124,47 +124,47 @@ func validateHTTPUploadSource(errs ValidationErrors, context string, upload HTTP
 	return errs
 }
 
-func validateBackend(errs ValidationErrors, context, backend, host string, port int, path, endpoint, bucket, prefix string, hostKeyPolicy HostKeyPolicy, creds Credentials) ValidationErrors {
+func validateBackend(errs ValidationErrors, context string, backend backendView) ValidationErrors {
-	switch backend {
+	switch backend.Backend {
 	case "":
 		errs = append(errs, context+".backend is required")
 	case BackendLocal:
-		if path == "" {
+		if backend.Path == "" {
 			errs = append(errs, context+".path is required for local backend")
 		}
 	case BackendSSH:
-		if host == "" {
+		if backend.Host == "" {
 			errs = append(errs, context+".host is required for ssh backend")
 		}
-		if path == "" {
+		if backend.Path == "" {
 			errs = append(errs, context+".path is required for ssh backend")
 		}
-		if port < 0 || port > 65535 {
+		if backend.Port < 0 || backend.Port > 65535 {
 			errs = append(errs, context+".port must be between 1 and 65535")
 		}
-		if port == 0 {
+		if backend.Port == 0 {
 			errs = append(errs, context+".port is required for ssh backend after defaults are applied")
 		}
-		if hostKeyPolicy != "" {
+		if backend.SSH.HostKeyPolicy != "" {
-			if _, ok := NormalizeHostKeyPolicy(string(hostKeyPolicy)); !ok {
+			if _, ok := NormalizeHostKeyPolicy(string(backend.SSH.HostKeyPolicy)); !ok {
 				errs = append(errs, context+".host_key_policy must be strict, true, accept-new, off, or false")
 			}
 		}
 	case BackendS3:
-		if endpoint == "" {
+		if backend.Endpoint == "" {
 			errs = append(errs, context+".endpoint is required for s3 backend")
 		}
-		if bucket == "" {
+		if backend.Bucket == "" {
 			errs = append(errs, context+".bucket is required for s3 backend")
 		}
-		if err := ValidateS3Prefix(prefix); err != nil {
+		if err := ValidateS3Prefix(backend.Prefix); err != nil {
 			errs = append(errs, context+".prefix must be a clean relative slash-separated path")
 		}
-		if (creds.AccessKeyIDEnv == "") != (creds.SecretAccessKeyEnv == "") {
+		if (backend.Creds.AccessKeyIDEnv == "") != (backend.Creds.SecretAccessKeyEnv == "") {
 			errs = append(errs, context+".credentials.access_key_id_env and credentials.secret_access_key_env must be configured together")
 		}
 	default:
-		errs = append(errs, context+".backend "+backend+" is unsupported")
+		errs = append(errs, context+".backend "+backend.Backend+" is unsupported")
 	}
 	return errs
 }