Validate Distributor endpoints before publication

This commit is contained in:
2026-08-13 02:44:21 +00:00
parent 04b8358965
commit 0b57d99a97
9 changed files with 155 additions and 16 deletions

View File

@@ -120,7 +120,7 @@ Distributor notification is disabled by default. Its fields are:
| Field | Default | Rules when notification is enabled | | Field | Default | Rules when notification is enabled |
| --- | --- | --- | | --- | --- | --- |
| `enabled` | `false` | Activates Distributor notification validation. | | `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. | | `token_env` | `DISTRIBUTOR_UPLOAD_TOKEN` | Must name a valid environment variable. |
| `timeout` | `30s` | Must be greater than zero. | | `timeout` | `30s` | Must be greater than zero. |
| `failure_policy` | `error` | Must be `error`. | | `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`. The upload token is read from the environment variable named by `token_env`.
Use `secrets.directory` when a file-backed secret is appropriate. 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 When notification is enabled, each rendered single-report pipeline ID, bundle
ID, and idempotency key must contain at least one non-whitespace character. ID, and idempotency key must contain at least one non-whitespace character.

View File

@@ -8,8 +8,10 @@ and [operations guide](../../operations.md).
## Upload Admission ## Upload Admission
Weatherreporter uses an absolute HTTP(S) endpoint as a base URL. The client Weatherreporter uses an absolute HTTP(S) endpoint with a host as a base URL.
posts a gzip-compressed source bundle to: 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 ```text
POST /v1/pipelines/<pipeline_id>/upload POST /v1/pipelines/<pipeline_id>/upload

View File

@@ -6,10 +6,11 @@ attempt and calls `UploadFiles`, followed by `Status` for the accepted run.
## Client And Upload ## Client And Upload
The adapter constructs the client with the configured endpoint, bearer token, The adapter constructs the client with the prevalidated HTTP(S) endpoint,
and an HTTP client whose timeout is the configured Distributor timeout. It bearer token, and an HTTP client whose timeout is the configured Distributor
passes no custom retry options, so the pinned client's defaults apply: three timeout. The endpoint may include a path prefix but never userinfo, a query, or
attempts, 100 ms base delay, and one-second maximum delay. 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: For each notification, Weatherreporter calls `UploadFiles` with:

View File

@@ -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. top-level notification result contains the delivery diagnostic.
For a single report, Distributor notification follows the atomic output write. For a single report, Distributor notification follows the atomic output write.
Enabled notification templates are validated before report processing, including Enabled notification configuration, including the HTTP(S) endpoint and
the requirement that each rendered identity contains a non-whitespace character. templates, is validated before report processing. A malformed endpoint does not
See the [configuration reference](config.md) for pipeline, bundle, 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. idempotency-key, and per-report path templates.
## Comparison Bundles ## Comparison Bundles

View File

@@ -250,6 +250,9 @@ func GenerateDetailed(ctx context.Context, req GenerateRequest) (*ReportResult,
return nil, err return nil, err
} }
result := initialReportResult(req, resolved, PromptInspectionResult{}) 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) outputPath, err := resolveReportOutputPath(req.WorkingDir, req.OutputPath, req.Config.Output.Directory, resolved)
if err != nil { if err != nil {
return result, err 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 { if _, err := report.BatchForCommandName(string(req.Batch)); err != nil {
return nil, err return nil, err
} }
if err := preflightDistributorNotification(req.Config); err != nil {
return nil, err
}
outputDir, err := resolveOutputDirWithConfigured(req.WorkingDir, req.OutputDir, req.Config.Output.Directory) outputDir, err := resolveOutputDirWithConfigured(req.WorkingDir, req.OutputDir, req.Config.Output.Directory)
if err != nil { if err != nil {
return nil, err return nil, err
@@ -512,6 +518,16 @@ func reportNotifier(cfg config.Config, notifier Notifier) (Notifier, bool) {
}, true }, 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) { 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)) values, err := distributorTemplateValuesForReport(cfg, resolved, runID, filepath.Base(outputPath))
if err != nil { if err != nil {

View File

@@ -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) { func TestRunBatchDetailedUsesDefaultAndConfiguredOutputDirectories(t *testing.T) {
tests := []struct { tests := []struct {
name string name string

View File

@@ -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) { func TestGenerateDetailedReturnsResolvedResultWhenCollectionFails(t *testing.T) {
cfg := config.Defaults() cfg := config.Defaults()
cfg.WeatherAPI.Timezone, cfg.Location.ID = "America/Chicago", "home" cfg.WeatherAPI.Timezone, cfg.Location.ID = "America/Chicago", "home"

View File

@@ -1311,9 +1311,10 @@ func TestDisabledDistributorNotifyAcceptsMalformedBatchTemplates(t *testing.T) {
func TestEnabledDistributorNotifyValidation(t *testing.T) { func TestEnabledDistributorNotifyValidation(t *testing.T) {
tests := []struct { tests := []struct {
name string name string
mutate func(*Config) mutate func(*Config)
wantErr string wantErr string
wantAbsent string
}{ }{
{ {
name: "Endpoint", name: "Endpoint",
@@ -1322,6 +1323,35 @@ func TestEnabledDistributorNotifyValidation(t *testing.T) {
}, },
wantErr: "notify.distributor.endpoint", 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", name: "TokenEnvEmpty",
mutate: func(cfg *Config) { mutate: func(cfg *Config) {
@@ -1422,6 +1452,27 @@ func TestEnabledDistributorNotifyValidation(t *testing.T) {
if !strings.Contains(err.Error(), tt.wantErr) { if !strings.Contains(err.Error(), tt.wantErr) {
t.Fatalf("error = %q, want %q", 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)
}
}) })
} }
} }

View File

@@ -119,9 +119,8 @@ func validateDistributorNotify(cfg DistributorNotifyConfig) error {
return nil return nil
} }
parsed, err := url.Parse(cfg.Endpoint) if err := ValidateDistributorEndpoint(cfg.Endpoint); err != nil {
if err != nil || parsed.Scheme == "" || parsed.Host == "" { return fmt.Errorf("notify.distributor.endpoint %w", err)
return fmt.Errorf("notify.distributor.endpoint must be an absolute URL when enabled")
} }
if cfg.TokenEnv == "" { if cfg.TokenEnv == "" {
return fmt.Errorf("notify.distributor.token_env is required when enabled") return fmt.Errorf("notify.distributor.token_env is required when enabled")
@@ -172,6 +171,25 @@ func validateDistributorNotify(cfg DistributorNotifyConfig) error {
return nil 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 { func validateDistributorBatchNotify(cfg DistributorBatchNotifyConfig) error {
if !cfg.Enabled { if !cfg.Enabled {
return nil return nil