From 0b57d99a971f464ff4e8e1cd303200bbba3fe759 Mon Sep 17 00:00:00 2001 From: Eric Rakestraw Date: Thu, 13 Aug 2026 02:44:21 +0000 Subject: [PATCH] Validate Distributor endpoints before publication --- docs/config.md | 7 ++- docs/integrations/distributor/api.md | 6 ++- docs/integrations/distributor/pkg-upload.md | 9 ++-- docs/operations.md | 7 +-- internal/app/app.go | 16 ++++++ internal/app/batch_generation_test.go | 23 +++++++++ internal/app/generation_test.go | 22 ++++++++ internal/config/config_test.go | 57 +++++++++++++++++++-- internal/config/validate.go | 24 +++++++-- 9 files changed, 155 insertions(+), 16 deletions(-) diff --git a/docs/config.md b/docs/config.md index f4e5570..5c4088f 100644 --- a/docs/config.md +++ b/docs/config.md @@ -120,7 +120,7 @@ Distributor notification is disabled by default. Its fields are: | Field | Default | Rules when notification is enabled | | --- | --- | --- | | `enabled` | `false` | Activates Distributor notification validation. | -| `endpoint` | `https://distributor.example.com` | Must be an absolute URL. | +| `endpoint` | `https://distributor.example.com` | Must be an absolute HTTP(S) base URL with a host and no userinfo, query, or fragment. A path prefix is allowed. | | `token_env` | `DISTRIBUTOR_UPLOAD_TOKEN` | Must name a valid environment variable. | | `timeout` | `30s` | Must be greater than zero. | | `failure_policy` | `error` | Must be `error`. | @@ -135,6 +135,11 @@ Distributor notification is disabled by default. Its fields are: The upload token is read from the environment variable named by `token_env`. Use `secrets.directory` when a file-backed secret is appropriate. +When notification is enabled, Weatherreporter validates the Distributor endpoint +before prompt inspection, weather collection, or output publication. Use an +HTTP(S) base URL such as `https://distributor.example.com/archive`; do not put +credentials, a query string, or a fragment in the endpoint. + When notification is enabled, each rendered single-report pipeline ID, bundle ID, and idempotency key must contain at least one non-whitespace character. diff --git a/docs/integrations/distributor/api.md b/docs/integrations/distributor/api.md index 9dfcd2a..143b0a3 100644 --- a/docs/integrations/distributor/api.md +++ b/docs/integrations/distributor/api.md @@ -8,8 +8,10 @@ and [operations guide](../../operations.md). ## Upload Admission -Weatherreporter uses an absolute HTTP(S) endpoint as a base URL. The client -posts a gzip-compressed source bundle to: +Weatherreporter uses an absolute HTTP(S) endpoint with a host as a base URL. +It allows a path prefix but rejects userinfo, query strings, and fragments +before local report work begins. The client posts a gzip-compressed source +bundle to: ```text POST /v1/pipelines//upload diff --git a/docs/integrations/distributor/pkg-upload.md b/docs/integrations/distributor/pkg-upload.md index d5f2429..7ce8670 100644 --- a/docs/integrations/distributor/pkg-upload.md +++ b/docs/integrations/distributor/pkg-upload.md @@ -6,10 +6,11 @@ attempt and calls `UploadFiles`, followed by `Status` for the accepted run. ## Client And Upload -The adapter constructs the client with the configured endpoint, bearer token, -and an HTTP client whose timeout is the configured Distributor timeout. It -passes no custom retry options, so the pinned client's defaults apply: three -attempts, 100 ms base delay, and one-second maximum delay. +The adapter constructs the client with the prevalidated HTTP(S) endpoint, +bearer token, and an HTTP client whose timeout is the configured Distributor +timeout. The endpoint may include a path prefix but never userinfo, a query, or +a fragment. It passes no custom retry options, so the pinned client's defaults +apply: three attempts, 100 ms base delay, and one-second maximum delay. For each notification, Weatherreporter calls `UploadFiles` with: diff --git a/docs/operations.md b/docs/operations.md index e1528e3..6c1cb6e 100644 --- a/docs/operations.md +++ b/docs/operations.md @@ -83,9 +83,10 @@ returns a failed batch status even when all report counters show success; the top-level notification result contains the delivery diagnostic. For a single report, Distributor notification follows the atomic output write. -Enabled notification templates are validated before report processing, including -the requirement that each rendered identity contains a non-whitespace character. -See the [configuration reference](config.md) for pipeline, bundle, +Enabled notification configuration, including the HTTP(S) endpoint and +templates, is validated before report processing. A malformed endpoint does not +collect weather data, generate a report, publish output, or invoke Distributor. +See the [configuration reference](config.md) for endpoint, pipeline, bundle, idempotency-key, and per-report path templates. ## Comparison Bundles diff --git a/internal/app/app.go b/internal/app/app.go index 360b267..d033655 100644 --- a/internal/app/app.go +++ b/internal/app/app.go @@ -250,6 +250,9 @@ func GenerateDetailed(ctx context.Context, req GenerateRequest) (*ReportResult, return nil, err } result := initialReportResult(req, resolved, PromptInspectionResult{}) + if err := preflightDistributorNotification(req.Config); err != nil { + return result, err + } outputPath, err := resolveReportOutputPath(req.WorkingDir, req.OutputPath, req.Config.Output.Directory, resolved) if err != nil { return result, err @@ -302,6 +305,9 @@ func RunBatchDetailed(ctx context.Context, req BatchRequest) (*BatchResult, erro if _, err := report.BatchForCommandName(string(req.Batch)); err != nil { return nil, err } + if err := preflightDistributorNotification(req.Config); err != nil { + return nil, err + } outputDir, err := resolveOutputDirWithConfigured(req.WorkingDir, req.OutputDir, req.Config.Output.Directory) if err != nil { return nil, err @@ -512,6 +518,16 @@ func reportNotifier(cfg config.Config, notifier Notifier) (Notifier, bool) { }, true } +func preflightDistributorNotification(cfg config.Config) error { + if !cfg.Notify.Distributor.Enabled { + return nil + } + if err := config.ValidateDistributorEndpoint(cfg.Notify.Distributor.Endpoint); err != nil { + return fmt.Errorf("validate notify.distributor.endpoint: %w", err) + } + return nil +} + func buildNotificationRequest(cfg config.Config, resolved report.Resolved, outputPath, runID string, generatedAt time.Time) (NotificationRequest, error) { values, err := distributorTemplateValuesForReport(cfg, resolved, runID, filepath.Base(outputPath)) if err != nil { diff --git a/internal/app/batch_generation_test.go b/internal/app/batch_generation_test.go index 85d5fdf..517439b 100644 --- a/internal/app/batch_generation_test.go +++ b/internal/app/batch_generation_test.go @@ -61,6 +61,29 @@ func TestRunBatchDetailedNotifiesOnlyAfterAllOutputsExist(t *testing.T) { } } +func TestRunBatchDetailedRejectsUnsupportedDistributorEndpointBeforeWork(t *testing.T) { + outputDir := t.TempDir() + cfg := generationDistributorConfig() + cfg.Notify.Distributor.Endpoint = "ftp://distributor.example.test" + bundle := generationBundle(t) + collector := &generationCollector{bundle: &bundle} + executor := &generationExecutor{} + notifier := &generationNotifier{} + + result, err := RunBatchDetailed(context.Background(), BatchRequest{ + Config: cfg, Batch: BatchMorning, + Now: generationTime("2026-05-29T08:30:00-05:00"), WorkingDir: t.TempDir(), OutputDir: outputDir, + Collector: collector, Executor: executor, Notifier: notifier, + }) + if err == nil || result != nil || collector.called || executor.promptInspections != 0 || executor.called || notifier.calls != 0 || notifier.batchCalls != 0 { + t.Fatalf("RunBatchDetailed() result/error/collector/executor/notifier = %#v/%v/%t/%#v/%#v", result, err, collector.called, executor, notifier) + } + entries, readErr := os.ReadDir(outputDir) + if readErr != nil || len(entries) != 0 { + t.Fatalf("output directory entries/error = %v/%v", entries, readErr) + } +} + func TestRunBatchDetailedUsesDefaultAndConfiguredOutputDirectories(t *testing.T) { tests := []struct { name string diff --git a/internal/app/generation_test.go b/internal/app/generation_test.go index c2e6d90..487389c 100644 --- a/internal/app/generation_test.go +++ b/internal/app/generation_test.go @@ -311,6 +311,28 @@ func TestGenerateDetailedRejectsOverlongOutputBeforeWork(t *testing.T) { } } +func TestGenerateDetailedRejectsUnsupportedDistributorEndpointBeforeWork(t *testing.T) { + outputPath := filepath.Join(t.TempDir(), "daily.md") + cfg := generationDistributorConfig() + cfg.Notify.Distributor.Endpoint = "ftp://distributor.example.test" + bundle := generationBundle(t) + collector := &generationCollector{bundle: &bundle} + executor := &generationExecutor{} + notifier := &generationNotifier{} + + result, err := GenerateDetailed(context.Background(), GenerateRequest{ + Config: cfg, Report: ReportDaily, + Date: generationTime("2026-05-29T12:00:00-05:00"), Now: generationTime("2026-05-29T08:30:00-05:00"), + WorkingDir: t.TempDir(), OutputPath: outputPath, Collector: collector, Executor: executor, Notifier: notifier, + }) + if err == nil || result == nil || collector.called || executor.promptInspections != 0 || executor.called || notifier.calls != 0 { + t.Fatalf("GenerateDetailed() result/error/collector/executor/notifier = %#v/%v/%t/%#v/%#v", result, err, collector.called, executor, notifier) + } + if _, statErr := os.Stat(outputPath); !os.IsNotExist(statErr) { + t.Fatalf("output exists after endpoint preflight failure: %v", statErr) + } +} + func TestGenerateDetailedReturnsResolvedResultWhenCollectionFails(t *testing.T) { cfg := config.Defaults() cfg.WeatherAPI.Timezone, cfg.Location.ID = "America/Chicago", "home" diff --git a/internal/config/config_test.go b/internal/config/config_test.go index b5e65ae..289da85 100644 --- a/internal/config/config_test.go +++ b/internal/config/config_test.go @@ -1311,9 +1311,10 @@ func TestDisabledDistributorNotifyAcceptsMalformedBatchTemplates(t *testing.T) { func TestEnabledDistributorNotifyValidation(t *testing.T) { tests := []struct { - name string - mutate func(*Config) - wantErr string + name string + mutate func(*Config) + wantErr string + wantAbsent string }{ { name: "Endpoint", @@ -1322,6 +1323,35 @@ func TestEnabledDistributorNotifyValidation(t *testing.T) { }, wantErr: "notify.distributor.endpoint", }, + { + name: "UnsupportedEndpointScheme", + mutate: func(cfg *Config) { + cfg.Notify.Distributor.Endpoint = "ftp://distributor.example.test" + }, + wantErr: "notify.distributor.endpoint", + }, + { + name: "EndpointUserinfo", + mutate: func(cfg *Config) { + cfg.Notify.Distributor.Endpoint = "https://userinfo-secret@distributor.example.test" + }, + wantErr: "notify.distributor.endpoint", + wantAbsent: "userinfo-secret", + }, + { + name: "EndpointQuery", + mutate: func(cfg *Config) { + cfg.Notify.Distributor.Endpoint = "https://distributor.example.test?preview=1" + }, + wantErr: "notify.distributor.endpoint", + }, + { + name: "EndpointFragment", + mutate: func(cfg *Config) { + cfg.Notify.Distributor.Endpoint = "https://distributor.example.test#status" + }, + wantErr: "notify.distributor.endpoint", + }, { name: "TokenEnvEmpty", mutate: func(cfg *Config) { @@ -1422,6 +1452,27 @@ func TestEnabledDistributorNotifyValidation(t *testing.T) { if !strings.Contains(err.Error(), tt.wantErr) { t.Fatalf("error = %q, want %q", err.Error(), tt.wantErr) } + if tt.wantAbsent != "" && strings.Contains(err.Error(), tt.wantAbsent) { + t.Fatalf("error = %q, must not contain endpoint userinfo", err.Error()) + } + }) + } +} + +func TestEnabledDistributorNotifyAcceptsHTTPBasePaths(t *testing.T) { + for _, endpoint := range []string{ + "http://distributor.example.test/archive", + "https://distributor.example.test/archive/", + } { + t.Run(endpoint, func(t *testing.T) { + cfg := Defaults() + cfg.Notify.Distributor.Enabled = true + cfg.Notify.Distributor.Endpoint = endpoint + cfg.Notify.Distributor.PipelineIDTemplate = "weatherreporter.{artifact_group}" + + if err := Validate(cfg); err != nil { + t.Fatalf("Validate() error = %v", err) + } }) } } diff --git a/internal/config/validate.go b/internal/config/validate.go index 29f5dc2..33673d3 100644 --- a/internal/config/validate.go +++ b/internal/config/validate.go @@ -119,9 +119,8 @@ func validateDistributorNotify(cfg DistributorNotifyConfig) error { return nil } - parsed, err := url.Parse(cfg.Endpoint) - if err != nil || parsed.Scheme == "" || parsed.Host == "" { - return fmt.Errorf("notify.distributor.endpoint must be an absolute URL when enabled") + if err := ValidateDistributorEndpoint(cfg.Endpoint); err != nil { + return fmt.Errorf("notify.distributor.endpoint %w", err) } if cfg.TokenEnv == "" { return fmt.Errorf("notify.distributor.token_env is required when enabled") @@ -172,6 +171,25 @@ func validateDistributorNotify(cfg DistributorNotifyConfig) error { return nil } +// ValidateDistributorEndpoint verifies the endpoint grammar accepted by the +// pinned Distributor upload client. +func ValidateDistributorEndpoint(endpoint string) error { + parsed, err := url.Parse(endpoint) + if err != nil || parsed.Scheme == "" || parsed.Host == "" { + return fmt.Errorf("must be an absolute URL") + } + if parsed.Scheme != "http" && parsed.Scheme != "https" { + return fmt.Errorf("must use http or https") + } + if parsed.User != nil { + return fmt.Errorf("must not include userinfo") + } + if parsed.RawQuery != "" || parsed.Fragment != "" { + return fmt.Errorf("must not include query or fragment") + } + return nil +} + func validateDistributorBatchNotify(cfg DistributorBatchNotifyConfig) error { if !cfg.Enabled { return nil