From 033b2e50152d4ef64f5a27524cbfdf352f98b843 Mon Sep 17 00:00:00 2001 From: Eric Rakestraw Date: Mon, 8 Jun 2026 04:27:49 +0000 Subject: [PATCH] Add upload token config validation --- examples/http-upload-local.yml | 7 +- internal/app/run_test.go | 12 +- internal/app/serve_test.go | 14 +- internal/app/upload_coordinator_test.go | 7 +- internal/app/upload_http.go | 19 ++- internal/app/upload_http_integration_test.go | 6 +- internal/app/upload_http_test.go | 12 +- internal/config/backend_view_test.go | 5 +- internal/config/config.go | 14 +- internal/config/load_test.go | 149 +++++++++++++++++-- internal/config/validate.go | 73 ++++++++- 11 files changed, 278 insertions(+), 40 deletions(-) diff --git a/examples/http-upload-local.yml b/examples/http-upload-local.yml index eca21cc..a315e38 100644 --- a/examples/http-upload-local.yml +++ b/examples/http-upload-local.yml @@ -9,11 +9,15 @@ server: queue_size: 16 max_concurrency: 1 retention: 24h +upload_tokens: + - id: example-uploader + token_env: DISTRIBUTOR_EXAMPLE_UPLOAD_TOKEN + allow_pipelines: + - example-http-upload pipelines: - id: example-http-upload source: backend: http_upload - token_env: DISTRIBUTOR_EXAMPLE_UPLOAD_TOKEN destinations: - id: local-archive backend: local @@ -21,4 +25,3 @@ pipelines: publish: source: true html: false - diff --git a/internal/app/run_test.go b/internal/app/run_test.go index 1be3d1d..fec7241 100644 --- a/internal/app/run_test.go +++ b/internal/app/run_test.go @@ -292,7 +292,6 @@ func TestRunPipelineWithLocalSourcePublishesToRegisteredDestinationBackends(t *t ID: "reports", Source: config.Backend{ Backend: config.BackendHTTPUpload, - Upload: config.HTTPUpload{TokenEnv: "UPLOAD_TOKEN"}, }, Destinations: []config.Destination{ { @@ -309,6 +308,11 @@ func TestRunPipelineWithLocalSourcePublishesToRegisteredDestinationBackends(t *t }, }, }}, + UploadTokens: []config.UploadToken{{ + ID: "reporter", + TokenEnv: "UPLOAD_TOKEN", + AllowPipelines: []string{"reports"}, + }}, } config.ApplyDefaults(&cfg) provider := fakeBackendFactoryProvider(t, map[string]storage.Backend{ @@ -1674,11 +1678,15 @@ func writeFanoutConfig(t *testing.T, sourceRoot, firstDestination, secondDestina func writeUploadPipelineConfig(t *testing.T, destinationRoot string) string { t.Helper() return writeConfigFile(t, ` +upload_tokens: + - id: reporter + token_env: UPLOAD_TOKEN + allow_pipelines: + - reports pipelines: - id: reports source: backend: http_upload - token_env: UPLOAD_TOKEN destinations: - id: archive backend: local diff --git a/internal/app/serve_test.go b/internal/app/serve_test.go index 0e40e12..e8a79d8 100644 --- a/internal/app/serve_test.go +++ b/internal/app/serve_test.go @@ -70,14 +70,24 @@ func writeServeUploadConfig(t *testing.T, tokenEnvs []string) string { server: http: bind: 127.0.0.1:0 -pipelines: +upload_tokens: ` for index, tokenEnv := range tokenEnvs { body += ` + - id: reporter-` + string(rune('a'+index)) + ` + token_env: ` + tokenEnv + ` + allow_pipelines: + - reports-` + string(rune('a'+index)) + ` +` + } + body += ` +pipelines: +` + for index := range tokenEnvs { + body += ` - id: reports-` + string(rune('a'+index)) + ` source: backend: http_upload - token_env: ` + tokenEnv + ` destinations: - id: archive backend: local diff --git a/internal/app/upload_coordinator_test.go b/internal/app/upload_coordinator_test.go index f0720e0..6060616 100644 --- a/internal/app/upload_coordinator_test.go +++ b/internal/app/upload_coordinator_test.go @@ -564,11 +564,11 @@ func uploadCoordinatorConfig(t *testing.T, opts uploadCoordinatorConfigOptions) }}, } for _, pipelineID := range opts.pipelineIDs { + tokenEnv := strings.ToUpper(strings.ReplaceAll(pipelineID, "-", "_")) + "_TOKEN" cfg.Pipelines = append(cfg.Pipelines, config.Pipeline{ ID: pipelineID, Source: config.Backend{ Backend: config.BackendHTTPUpload, - Upload: config.HTTPUpload{TokenEnv: strings.ToUpper(strings.ReplaceAll(pipelineID, "-", "_")) + "_TOKEN"}, }, Destinations: []config.Destination{{ ID: "archive", @@ -576,6 +576,11 @@ func uploadCoordinatorConfig(t *testing.T, opts uploadCoordinatorConfigOptions) Path: t.TempDir(), }}, }) + cfg.UploadTokens = append(cfg.UploadTokens, config.UploadToken{ + ID: pipelineID + "-reporter", + TokenEnv: tokenEnv, + AllowPipelines: []string{pipelineID}, + }) } return cfg } diff --git a/internal/app/upload_http.go b/internal/app/upload_http.go index 4b0be46..46aa0ce 100644 --- a/internal/app/upload_http.go +++ b/internal/app/upload_http.go @@ -49,22 +49,21 @@ func newUploadHTTPHandler(ctx context.Context, cfg config.Config, environment co func resolveUploadTokens(cfg config.Config, environment config.Environment) (map[string]string, error) { tokens := make(map[string]string) - for _, pipeline := range cfg.Pipelines { - if pipeline.Source.Backend != config.BackendHTTPUpload { - continue - } - tokenName := pipeline.Source.Upload.TokenEnv - token, ok := environment.Lookup(tokenName) + for _, uploadToken := range cfg.UploadTokens { + token, ok := environment.Lookup(uploadToken.TokenEnv) if !ok { - return nil, fmt.Errorf("upload token environment variable %s is not set", tokenName) + return nil, fmt.Errorf("upload token environment variable %s is not set", uploadToken.TokenEnv) } if token == "" { - return nil, fmt.Errorf("upload token environment variable %s is empty", tokenName) + return nil, fmt.Errorf("upload token environment variable %s is empty", uploadToken.TokenEnv) + } + if len(uploadToken.AllowPipelines) != 1 { + return nil, fmt.Errorf("upload token %s must allow exactly one pipeline for legacy upload routing", uploadToken.ID) } if existing, exists := tokens[token]; exists { - return nil, fmt.Errorf("upload token environment variables for pipelines %s and %s resolve to the same value", existing, pipeline.ID) + return nil, fmt.Errorf("upload token environment variables for pipelines %s and %s resolve to the same value", existing, uploadToken.AllowPipelines[0]) } - tokens[token] = pipeline.ID + tokens[token] = uploadToken.AllowPipelines[0] } return tokens, nil } diff --git a/internal/app/upload_http_integration_test.go b/internal/app/upload_http_integration_test.go index 675010d..4bebde5 100644 --- a/internal/app/upload_http_integration_test.go +++ b/internal/app/upload_http_integration_test.go @@ -311,7 +311,6 @@ func httpUploadIntegrationConfig(t *testing.T, pipelines []httpUploadPipelineSpe Source: config.Backend{ Backend: config.BackendHTTPUpload, Upload: config.HTTPUpload{ - TokenEnv: spec.tokenEnv, StagingPath: spec.stagingPath, MaxUploadSize: &size, }, @@ -326,6 +325,11 @@ func httpUploadIntegrationConfig(t *testing.T, pipelines []httpUploadPipelineSpe }) } cfg.Pipelines = append(cfg.Pipelines, pipeline) + cfg.UploadTokens = append(cfg.UploadTokens, config.UploadToken{ + ID: spec.id + "-reporter", + TokenEnv: spec.tokenEnv, + AllowPipelines: []string{spec.id}, + }) } config.ApplyDefaults(&cfg) return cfg diff --git a/internal/app/upload_http_test.go b/internal/app/upload_http_test.go index 1ae6852..8bd6a24 100644 --- a/internal/app/upload_http_test.go +++ b/internal/app/upload_http_test.go @@ -53,10 +53,14 @@ func TestResolveUploadTokensFailsForMissingAndDuplicateTokens(t *testing.T) { ID: "weekly", Source: config.Backend{ Backend: config.BackendHTTPUpload, - Upload: config.HTTPUpload{TokenEnv: "OTHER_UPLOAD_TOKEN"}, }, Destinations: cfg.Pipelines[0].Destinations, }) + cfg.UploadTokens = append(cfg.UploadTokens, config.UploadToken{ + ID: "weekly-reporter", + TokenEnv: "OTHER_UPLOAD_TOKEN", + AllowPipelines: []string{"weekly"}, + }) config.ApplyDefaults(&cfg) secret := "super-secret-token" _, err = resolveUploadTokens(cfg, uploadHTTPTestEnvironment(map[string]string{ @@ -362,7 +366,6 @@ func uploadHTTPTestConfig() config.Config { Source: config.Backend{ Backend: config.BackendHTTPUpload, Upload: config.HTTPUpload{ - TokenEnv: "UPLOAD_TOKEN", StagingPath: "/tmp/distributor-test/reports", MaxUploadSize: &size, }, @@ -374,6 +377,11 @@ func uploadHTTPTestConfig() config.Config { Publish: &config.PublishPolicy{Source: true}, }}, }}, + UploadTokens: []config.UploadToken{{ + ID: "reporter", + TokenEnv: "UPLOAD_TOKEN", + AllowPipelines: []string{"reports"}, + }}, } config.ApplyDefaults(&cfg) return cfg diff --git a/internal/config/backend_view_test.go b/internal/config/backend_view_test.go index 915cf25..8d396f0 100644 --- a/internal/config/backend_view_test.go +++ b/internal/config/backend_view_test.go @@ -53,12 +53,15 @@ func TestBackendViewValidationKeepsHTTPUploadSourceOnly(t *testing.T) { ID: "reports", Source: Backend{ Backend: BackendHTTPUpload, - Upload: HTTPUpload{TokenEnv: "UPLOAD_TOKEN"}, }, Destinations: []Destination{{ ID: "archive", Backend: BackendHTTPUpload, }}, + }}, UploadTokens: []UploadToken{{ + ID: "reporter", + TokenEnv: "UPLOAD_TOKEN", + AllowPipelines: []string{"reports"}, }}} ApplyDefaults(&cfg) diff --git a/internal/config/config.go b/internal/config/config.go index c4e02fe..047af4a 100644 --- a/internal/config/config.go +++ b/internal/config/config.go @@ -1,9 +1,10 @@ package config type Config struct { - Server Server `yaml:"server"` - Secrets Secrets `yaml:"secrets"` - Pipelines []Pipeline `yaml:"pipelines"` + Server Server `yaml:"server"` + Secrets Secrets `yaml:"secrets"` + UploadTokens []UploadToken `yaml:"upload_tokens"` + Pipelines []Pipeline `yaml:"pipelines"` } type Server struct { @@ -23,6 +24,12 @@ type Secrets struct { Directory string `yaml:"directory"` } +type UploadToken struct { + ID string `yaml:"id"` + TokenEnv string `yaml:"token_env"` + AllowPipelines []string `yaml:"allow_pipelines"` +} + type Pipeline struct { ID string `yaml:"id"` Source Backend `yaml:"source"` @@ -68,7 +75,6 @@ type Backend struct { } type HTTPUpload struct { - TokenEnv string `yaml:"token_env"` StagingPath string `yaml:"staging_path"` MaxUploadSize *ByteSize `yaml:"max_upload_size"` } diff --git a/internal/config/load_test.go b/internal/config/load_test.go index f9dc3ba..674eddc 100644 --- a/internal/config/load_test.go +++ b/internal/config/load_test.go @@ -256,13 +256,17 @@ pipelines: - id: weather-daily source: backend: http_upload - token_env: WEATHER_DAILY_UPLOAD_TOKEN staging_path: /srv/distributor/staging/weather-daily max_upload_size: 32MB destinations: - id: archive backend: local path: /archive +upload_tokens: + - id: weather-reporter + token_env: WEATHER_DAILY_UPLOAD_TOKEN + allow_pipelines: + - weather-daily `) server := cfg.Server.HTTP @@ -289,9 +293,6 @@ pipelines: if got, want := source.Backend, BackendHTTPUpload; got != want { t.Fatalf("source.backend = %q, want %q", got, want) } - if got, want := source.Upload.TokenEnv, "WEATHER_DAILY_UPLOAD_TOKEN"; got != want { - t.Fatalf("source.token_env = %q, want %q", got, want) - } if got, want := source.Upload.StagingPath, "/srv/distributor/staging/weather-daily"; got != want { t.Fatalf("source.staging_path = %q, want %q", got, want) } @@ -309,11 +310,15 @@ pipelines: - id: weather-daily source: backend: http_upload - token_env: WEATHER_DAILY_UPLOAD_TOKEN destinations: - id: archive backend: local path: /archive +upload_tokens: + - id: weather-reporter + token_env: WEATHER_DAILY_UPLOAD_TOKEN + allow_pipelines: + - weather-daily `) source := cfg.Pipelines[0].Source @@ -325,6 +330,120 @@ pipelines: } } +func TestLoadFileAcceptsHTTPUploadTokens(t *testing.T) { + tests := map[string]string{ + "valid multi pipeline token": ` +pipelines: + - id: weather-daily + source: + backend: http_upload + destinations: + - id: archive + backend: local + path: /archive/weather + - id: calendar-daily + source: + backend: http_upload + destinations: + - id: archive + backend: local + path: /archive/calendar +upload_tokens: + - id: reporter + token_env: REPORTER_UPLOAD_TOKEN + allow_pipelines: + - weather-daily + - calendar-daily +`, + "multiple tokens for one pipeline": ` +pipelines: + - id: reports + source: + backend: http_upload + destinations: + - id: archive + backend: local + path: /archive +upload_tokens: + - id: reporter-a + token_env: REPORTER_A_UPLOAD_TOKEN + allow_pipelines: + - reports + - id: reporter-b + token_env: REPORTER_B_UPLOAD_TOKEN + allow_pipelines: + - reports +`, + } + for name, body := range tests { + t.Run(name, func(t *testing.T) { + loadConfig(t, body) + }) + } +} + +func TestLoadFileRejectsInvalidUploadTokens(t *testing.T) { + tests := map[string]struct { + body string + want string + }{ + "missing token list": { + body: `pipelines: [{id: reports, source: {backend: http_upload}, destinations: [{id: archive, backend: local, path: /archive}]}]`, + want: "upload_tokens is required", + }, + "duplicate token ids": { + body: `upload_tokens: [{id: reporter, token_env: ONE_UPLOAD_TOKEN, allow_pipelines: [reports]}, {id: reporter, token_env: TWO_UPLOAD_TOKEN, allow_pipelines: [reports]}] +pipelines: [{id: reports, source: {backend: http_upload}, destinations: [{id: archive, backend: local, path: /archive}]}]`, + want: "upload token id reporter is duplicated", + }, + "duplicate allowlist entries": { + body: `upload_tokens: [{id: reporter, token_env: UPLOAD_TOKEN, allow_pipelines: [reports, reports]}] +pipelines: [{id: reports, source: {backend: http_upload}, destinations: [{id: archive, backend: local, path: /archive}]}]`, + want: "allow_pipelines contains duplicate pipeline id reports", + }, + "unknown allowed pipeline id": { + body: `upload_tokens: [{id: reporter, token_env: UPLOAD_TOKEN, allow_pipelines: [missing]}] +pipelines: [{id: reports, source: {backend: http_upload}, destinations: [{id: archive, backend: local, path: /archive}]}]`, + want: "references unknown pipeline missing", + }, + "non upload allowed pipeline id": { + body: `upload_tokens: [{id: reporter, token_env: UPLOAD_TOKEN, allow_pipelines: [reports]}, {id: uploader, token_env: OTHER_UPLOAD_TOKEN, allow_pipelines: [upload]}] +pipelines: [{id: reports, source: {backend: local, path: /source}, destinations: [{id: archive, backend: local, path: /archive}]}, {id: upload, source: {backend: http_upload}, destinations: [{id: archive, backend: local, path: /archive-upload}]}]`, + want: "references non-http_upload pipeline reports", + }, + "upload pipeline not allowed": { + body: `upload_tokens: [{id: reporter, token_env: UPLOAD_TOKEN, allow_pipelines: [reports]}] +pipelines: [{id: reports, source: {backend: http_upload}, destinations: [{id: archive, backend: local, path: /archive}]}, {id: other, source: {backend: http_upload}, destinations: [{id: archive, backend: local, path: /archive-other}]}]`, + want: "http_upload pipeline other is not allowed by any upload token", + }, + "missing token id": { + body: `upload_tokens: [{token_env: UPLOAD_TOKEN, allow_pipelines: [reports]}] +pipelines: [{id: reports, source: {backend: http_upload}, destinations: [{id: archive, backend: local, path: /archive}]}]`, + want: "upload_tokens[0].id is required", + }, + "invalid token id": { + body: `upload_tokens: [{id: ".reporter", token_env: UPLOAD_TOKEN, allow_pipelines: [reports]}] +pipelines: [{id: reports, source: {backend: http_upload}, destinations: [{id: archive, backend: local, path: /archive}]}]`, + want: "upload_tokens[0].id must be a slug-like identifier", + }, + "missing token env": { + body: `upload_tokens: [{id: reporter, allow_pipelines: [reports]}] +pipelines: [{id: reports, source: {backend: http_upload}, destinations: [{id: archive, backend: local, path: /archive}]}]`, + want: "upload_tokens[0].token_env is required", + }, + "missing allowlist": { + body: `upload_tokens: [{id: reporter, token_env: UPLOAD_TOKEN}] +pipelines: [{id: reports, source: {backend: http_upload}, destinations: [{id: archive, backend: local, path: /archive}]}]`, + want: "upload_tokens[0].allow_pipelines is required", + }, + } + for name, tt := range tests { + t.Run(name, func(t *testing.T) { + assertLoadError(t, tt.body, tt.want) + }) + } +} + func TestLoadFileValidBackendConfigs(t *testing.T) { tests := map[string]string{ "local": ` @@ -515,16 +634,22 @@ func TestLoadFileRejectsInvalidS3Config(t *testing.T) { func TestLoadFileRejectsInvalidHTTPUploadConfig(t *testing.T) { tests := map[string]string{ - "server size": `server: {http: {max_upload_size: 20XB}}`, - "source size": `pipelines: [{id: reports, source: {backend: http_upload, token_env: UPLOAD_TOKEN, max_upload_size: 20XB}, destinations: [{id: archive, backend: local, path: /archive}]}]`, - "zero source size": `pipelines: [{id: reports, source: {backend: http_upload, token_env: UPLOAD_TOKEN, max_upload_size: 0B}, destinations: [{id: archive, backend: local, path: /archive}]}]`, + "server size": `server: {http: {max_upload_size: 20XB}}`, + "source size": `upload_tokens: [{id: reporter, token_env: UPLOAD_TOKEN, allow_pipelines: [reports]}] +pipelines: [{id: reports, source: {backend: http_upload, max_upload_size: 20XB}, destinations: [{id: archive, backend: local, path: /archive}]}]`, + "zero source size": `upload_tokens: [{id: reporter, token_env: UPLOAD_TOKEN, allow_pipelines: [reports]}] +pipelines: [{id: reports, source: {backend: http_upload, max_upload_size: 0B}, destinations: [{id: archive, backend: local, path: /archive}]}]`, "server duration": `server: {http: {retention: forever}}`, "zero server duration": `server: {http: {retention: 0s}}`, - "missing token env": `pipelines: [{id: reports, source: {backend: http_upload}, destinations: [{id: archive, backend: local, path: /archive}]}]`, + "missing upload tokens": `pipelines: [{id: reports, source: {backend: http_upload}, destinations: [{id: archive, backend: local, path: /archive}]}]`, "destination http upload": `pipelines: [{id: reports, source: {backend: local, path: /source}, destinations: [{id: ingest, backend: http_upload}]}]`, - "literal token": `pipelines: [{id: reports, source: {backend: http_upload, token: secret, token_env: UPLOAD_TOKEN}, destinations: [{id: archive, backend: local, path: /archive}]}]`, - "unknown server field": `server: {http: {surprise: true}}`, - "unknown source field": `pipelines: [{id: reports, source: {backend: http_upload, token_env: UPLOAD_TOKEN, surprise: true}, destinations: [{id: archive, backend: local, path: /archive}]}]`, + "literal token": `upload_tokens: [{id: reporter, token: secret, token_env: UPLOAD_TOKEN, allow_pipelines: [reports]}] +pipelines: [{id: reports, source: {backend: http_upload}, destinations: [{id: archive, backend: local, path: /archive}]}]`, + "unknown server field": `server: {http: {surprise: true}}`, + "legacy source token env": `upload_tokens: [{id: reporter, token_env: UPLOAD_TOKEN, allow_pipelines: [reports]}] +pipelines: [{id: reports, source: {backend: http_upload, token_env: UPLOAD_TOKEN}, destinations: [{id: archive, backend: local, path: /archive}]}]`, + "unknown source field": `upload_tokens: [{id: reporter, token_env: UPLOAD_TOKEN, allow_pipelines: [reports]}] +pipelines: [{id: reports, source: {backend: http_upload, surprise: true}, destinations: [{id: archive, backend: local, path: /archive}]}]`, } for name, body := range tests { t.Run(name, func(t *testing.T) { diff --git a/internal/config/validate.go b/internal/config/validate.go index 102f099..2234bbe 100644 --- a/internal/config/validate.go +++ b/internal/config/validate.go @@ -29,6 +29,7 @@ func Validate(cfg Config) error { } pipelineIDs := make(map[string]struct{}, len(cfg.Pipelines)) + uploadPipelineIDs := make(map[string]struct{}) for pipelineIndex, pipeline := range cfg.Pipelines { pipelineContext := fmt.Sprintf("pipelines[%d]", pipelineIndex) if pipeline.ID == "" { @@ -42,6 +43,9 @@ func Validate(cfg Config) error { } errs = validateSourceBackend(errs, pipelineContext+".source", pipeline.Source) + if pipeline.Source.Backend == BackendHTTPUpload && pipeline.ID != "" { + uploadPipelineIDs[pipeline.ID] = struct{}{} + } errs = validateValidationPolicy(errs, pipelineContext+".validation", pipeline.Validation) if len(pipeline.Destinations) == 0 { errs = append(errs, pipelineContext+".destinations is required") @@ -68,6 +72,8 @@ func Validate(cfg Config) error { } } + errs = validateUploadTokens(errs, cfg.UploadTokens, pipelineIDs, uploadPipelineIDs) + if len(errs) > 0 { return errs } @@ -112,9 +118,6 @@ func validateDestinationBackend(errs ValidationErrors, context string, destinati } func validateHTTPUploadSource(errs ValidationErrors, context string, upload HTTPUpload) ValidationErrors { - if upload.TokenEnv == "" { - errs = append(errs, context+".token_env is required for http_upload backend") - } if upload.StagingPath == "" { errs = append(errs, context+".staging_path is required for http_upload backend") } @@ -124,6 +127,70 @@ func validateHTTPUploadSource(errs ValidationErrors, context string, upload HTTP return errs } +func validateUploadTokens(errs ValidationErrors, tokens []UploadToken, pipelineIDs, uploadPipelineIDs map[string]struct{}) ValidationErrors { + if len(uploadPipelineIDs) == 0 { + if len(tokens) > 0 { + errs = append(errs, "upload_tokens must reference configured http_upload pipelines") + } + return errs + } + if len(tokens) == 0 { + return append(errs, "upload_tokens is required when any pipeline source backend is http_upload") + } + + tokenIDs := make(map[string]struct{}, len(tokens)) + allowedUploadPipelineIDs := make(map[string]struct{}, len(uploadPipelineIDs)) + for tokenIndex, token := range tokens { + context := fmt.Sprintf("upload_tokens[%d]", tokenIndex) + if token.ID == "" { + errs = append(errs, context+".id is required") + } else if !idPattern.MatchString(token.ID) { + errs = append(errs, context+".id must be a slug-like identifier") + } else if _, exists := tokenIDs[token.ID]; exists { + errs = append(errs, "upload token id "+token.ID+" is duplicated") + } else { + tokenIDs[token.ID] = struct{}{} + } + + if token.TokenEnv == "" { + errs = append(errs, context+".token_env is required") + } + if len(token.AllowPipelines) == 0 { + errs = append(errs, context+".allow_pipelines is required") + } + + seenAllowed := make(map[string]struct{}, len(token.AllowPipelines)) + for allowIndex, pipelineID := range token.AllowPipelines { + allowContext := fmt.Sprintf("%s.allow_pipelines[%d]", context, allowIndex) + if pipelineID == "" { + errs = append(errs, allowContext+" is required") + continue + } + if _, exists := seenAllowed[pipelineID]; exists { + errs = append(errs, context+".allow_pipelines contains duplicate pipeline id "+pipelineID) + continue + } + seenAllowed[pipelineID] = struct{}{} + if _, exists := pipelineIDs[pipelineID]; !exists { + errs = append(errs, allowContext+" references unknown pipeline "+pipelineID) + continue + } + if _, exists := uploadPipelineIDs[pipelineID]; !exists { + errs = append(errs, allowContext+" references non-http_upload pipeline "+pipelineID) + continue + } + allowedUploadPipelineIDs[pipelineID] = struct{}{} + } + } + + for pipelineID := range uploadPipelineIDs { + if _, exists := allowedUploadPipelineIDs[pipelineID]; !exists { + errs = append(errs, "http_upload pipeline "+pipelineID+" is not allowed by any upload token") + } + } + return errs +} + func validateBackend(errs ValidationErrors, context string, backend backendView) ValidationErrors { switch backend.Backend { case "":