From 9d4694c6d8059c0c026936317c5e3ad3814dbeb4 Mon Sep 17 00:00:00 2001 From: Eric Rakestraw Date: Mon, 8 Jun 2026 04:32:20 +0000 Subject: [PATCH] Route uploads by pipeline path --- internal/app/upload_coordinator.go | 1 + internal/app/upload_http.go | 98 +++++++++--- internal/app/upload_http_integration_test.go | 67 +++++--- internal/app/upload_http_test.go | 154 ++++++++++++++++--- internal/config/validate.go | 10 +- 5 files changed, 261 insertions(+), 69 deletions(-) diff --git a/internal/app/upload_coordinator.go b/internal/app/upload_coordinator.go index 77326bc..6d90832 100644 --- a/internal/app/upload_coordinator.go +++ b/internal/app/upload_coordinator.go @@ -44,6 +44,7 @@ type UploadRunRecord struct { } type UploadRequest struct { + TokenID string PipelineID string ContentType string Body io.Reader diff --git a/internal/app/upload_http.go b/internal/app/upload_http.go index 46aa0ce..aff40da 100644 --- a/internal/app/upload_http.go +++ b/internal/app/upload_http.go @@ -19,8 +19,15 @@ type uploadCoordinator interface { } type uploadHTTPHandler struct { - coordinator uploadCoordinator - tokens map[string]string + coordinator uploadCoordinator + tokens map[string]resolvedUploadToken + uploadPipelines map[string]struct{} +} + +type resolvedUploadToken struct { + ID string + Value string + AllowedPipelines map[string]struct{} } type uploadAcceptedResponse struct { @@ -42,37 +49,57 @@ func newUploadHTTPHandler(ctx context.Context, cfg config.Config, environment co return nil, err } return uploadHTTPHandler{ - coordinator: NewUploadCoordinator(ctx, cfg), - tokens: tokens, + coordinator: NewUploadCoordinator(ctx, cfg), + tokens: tokens, + uploadPipelines: uploadPipelineSet(cfg), }, nil } -func resolveUploadTokens(cfg config.Config, environment config.Environment) (map[string]string, error) { - tokens := make(map[string]string) +func resolveUploadTokens(cfg config.Config, environment config.Environment) (map[string]resolvedUploadToken, error) { + tokens := make(map[string]resolvedUploadToken) 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", uploadToken.TokenEnv) + return nil, fmt.Errorf("upload token %s environment variable %s is not set", uploadToken.ID, uploadToken.TokenEnv) } if token == "" { - 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) + return nil, fmt.Errorf("upload token %s environment variable %s is empty", uploadToken.ID, uploadToken.TokenEnv) } 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, uploadToken.AllowPipelines[0]) + return nil, fmt.Errorf("upload token environment variables for tokens %s and %s resolve to the same value", existing.ID, uploadToken.ID) + } + tokens[token] = resolvedUploadToken{ + ID: uploadToken.ID, + Value: token, + AllowedPipelines: pipelineIDSet(uploadToken.AllowPipelines), } - tokens[token] = uploadToken.AllowPipelines[0] } return tokens, nil } +func uploadPipelineSet(cfg config.Config) map[string]struct{} { + pipelines := make(map[string]struct{}) + for _, pipeline := range cfg.Pipelines { + if pipeline.Source.Backend == config.BackendHTTPUpload { + pipelines[pipeline.ID] = struct{}{} + } + } + return pipelines +} + +func pipelineIDSet(ids []string) map[string]struct{} { + set := make(map[string]struct{}, len(ids)) + for _, id := range ids { + set[id] = struct{}{} + } + return set +} + func (handler uploadHTTPHandler) ServeHTTP(w http.ResponseWriter, r *http.Request) { switch { case r.Method == http.MethodGet && r.URL.Path == "/healthz": handler.handleHealth(w) - case r.Method == http.MethodPost && r.URL.Path == "/upload": + case r.Method == http.MethodPost && strings.HasPrefix(r.URL.Path, "/v1/pipelines/"): handler.handleUpload(w, r) case r.Method == http.MethodGet && strings.HasPrefix(r.URL.Path, "/runs/"): handler.handleRunStatus(w, r) @@ -90,11 +117,28 @@ func (handler uploadHTTPHandler) handleUpload(w http.ResponseWriter, r *http.Req writeHTTPError(w, http.StatusBadRequest, "pipeline id is not accepted") return } - pipelineID, ok := handler.authenticate(r.Header.Get("Authorization")) + pipelineID, ok := uploadPipelineIDFromPath(r.URL.Path) + if !ok { + writeHTTPError(w, http.StatusNotFound, "not found") + return + } + if !config.IsSlugLikeID(pipelineID) { + writeHTTPError(w, http.StatusBadRequest, "invalid pipeline id") + return + } + token, ok := handler.authenticate(r.Header.Get("Authorization")) if !ok { writeHTTPError(w, http.StatusUnauthorized, "unauthorized") return } + if _, ok := handler.uploadPipelines[pipelineID]; !ok { + writeHTTPError(w, http.StatusNotFound, "upload pipeline not found") + return + } + if _, ok := token.AllowedPipelines[pipelineID]; !ok { + writeHTTPError(w, http.StatusForbidden, "forbidden") + return + } contentType := r.Header.Get("Content-Type") if err := ingest.ValidateContentType(contentType); err != nil { writeHTTPError(w, http.StatusUnsupportedMediaType, "unsupported content type") @@ -106,6 +150,7 @@ func (handler uploadHTTPHandler) handleUpload(w http.ResponseWriter, r *http.Req return } record, err := handler.coordinator.Submit(r.Context(), UploadRequest{ + TokenID: token.ID, PipelineID: pipelineID, ContentType: contentType, Body: r.Body, @@ -121,6 +166,19 @@ func (handler uploadHTTPHandler) handleUpload(w http.ResponseWriter, r *http.Req }) } +func uploadPipelineIDFromPath(path string) (string, bool) { + const prefix = "/v1/pipelines/" + const suffix = "/upload" + if !strings.HasPrefix(path, prefix) || !strings.HasSuffix(path, suffix) { + return "", false + } + pipelineID := strings.TrimSuffix(strings.TrimPrefix(path, prefix), suffix) + if pipelineID == "" || strings.Contains(pipelineID, "/") { + return "", false + } + return pipelineID, true +} + func (handler uploadHTTPHandler) handleRunStatus(w http.ResponseWriter, r *http.Request) { rawRunID := strings.TrimPrefix(r.URL.Path, "/runs/") if rawRunID == "" || strings.Contains(rawRunID, "/") { @@ -135,17 +193,17 @@ func (handler uploadHTTPHandler) handleRunStatus(w http.ResponseWriter, r *http. writeJSON(w, http.StatusOK, record) } -func (handler uploadHTTPHandler) authenticate(header string) (string, bool) { +func (handler uploadHTTPHandler) authenticate(header string) (resolvedUploadToken, bool) { const prefix = "Bearer " if !strings.HasPrefix(header, prefix) { - return "", false + return resolvedUploadToken{}, false } token := strings.TrimSpace(strings.TrimPrefix(header, prefix)) if token == "" { - return "", false + return resolvedUploadToken{}, false } - pipelineID, ok := handler.tokens[token] - return pipelineID, ok + resolved, ok := handler.tokens[token] + return resolved, ok } func uploadIdempotencyKey(header http.Header) (string, error) { diff --git a/internal/app/upload_http_integration_test.go b/internal/app/upload_http_integration_test.go index 4bebde5..6e1c98c 100644 --- a/internal/app/upload_http_integration_test.go +++ b/internal/app/upload_http_integration_test.go @@ -78,15 +78,16 @@ func TestHTTPUploadInvalidArchiveIsRejectedWithoutRunID(t *testing.T) { destinations: []string{destination}, }}, 4, 1)) handler := uploadHTTPHandler{ - coordinator: coordinator, - tokens: map[string]string{"reports-secret": "reports"}, + coordinator: coordinator, + tokens: map[string]resolvedUploadToken{"reports-secret": uploadHTTPTestToken("reports-reporter", "reports-secret", "reports")}, + uploadPipelines: pipelineIDSet([]string{"reports"}), } server := httptest.NewServer(handler) defer server.Close() status, body := postHTTPUpload(t, server, "reports-secret", ingest.ContentTypeTar, []byte("not a tar archive")) if status != http.StatusBadRequest { - t.Fatalf("POST /upload status = %d, want %d; body = %s", status, http.StatusBadRequest, body) + t.Fatalf("POST upload status = %d, want %d; body = %s", status, http.StatusBadRequest, body) } if strings.Contains(body, "run_id") || strings.Contains(body, "reports-secret") { t.Fatalf("invalid archive response exposed run id or token: %s", body) @@ -106,8 +107,9 @@ func TestHTTPUploadIdempotencyReturnsOriginalRunForSameBundle(t *testing.T) { destinations: []string{destination}, }}, 4, 1)) handler := uploadHTTPHandler{ - coordinator: coordinator, - tokens: map[string]string{"reports-secret": "reports"}, + coordinator: coordinator, + tokens: map[string]resolvedUploadToken{"reports-secret": uploadHTTPTestToken("reports-reporter", "reports-secret", "reports")}, + uploadPipelines: pipelineIDSet([]string{"reports"}), } server := httptest.NewServer(handler) defer server.Close() @@ -133,8 +135,9 @@ func TestHTTPUploadIdempotencyReturnsConflictForDifferentBundle(t *testing.T) { destinations: []string{destination}, }}, 4, 1)) handler := uploadHTTPHandler{ - coordinator: coordinator, - tokens: map[string]string{"reports-secret": "reports"}, + coordinator: coordinator, + tokens: map[string]resolvedUploadToken{"reports-secret": uploadHTTPTestToken("reports-reporter", "reports-secret", "reports")}, + uploadPipelines: pipelineIDSet([]string{"reports"}), } server := httptest.NewServer(handler) defer server.Close() @@ -146,7 +149,7 @@ func TestHTTPUploadIdempotencyReturnsConflictForDifferentBundle(t *testing.T) { ID: "weather.daily.brentwood.2026-05-31", })) if status != http.StatusConflict { - t.Fatalf("POST /upload status = %d, want %d; body = %s", status, http.StatusConflict, body) + t.Fatalf("POST upload status = %d, want %d; body = %s", status, http.StatusConflict, body) } if strings.Contains(body, "reports-secret") { t.Fatalf("conflict response exposed token: %s", body) @@ -167,15 +170,16 @@ func TestHTTPUploadOversizedArchiveIsRejectedWithoutRunID(t *testing.T) { cfg.Pipelines[0].Source.Upload.MaxUploadSize = &size coordinator := NewUploadCoordinator(context.Background(), cfg) handler := uploadHTTPHandler{ - coordinator: coordinator, - tokens: map[string]string{"reports-secret": "reports"}, + coordinator: coordinator, + tokens: map[string]resolvedUploadToken{"reports-secret": uploadHTTPTestToken("reports-reporter", "reports-secret", "reports")}, + uploadPipelines: pipelineIDSet([]string{"reports"}), } server := httptest.NewServer(handler) defer server.Close() status, body := postHTTPUpload(t, server, "reports-secret", ingest.ContentTypeTar, bundleArchive(t, false, testutil.BundleOptions{})) if status != http.StatusRequestEntityTooLarge { - t.Fatalf("POST /upload status = %d, want %d; body = %s", status, http.StatusRequestEntityTooLarge, body) + t.Fatalf("POST upload status = %d, want %d; body = %s", status, http.StatusRequestEntityTooLarge, body) } if strings.Contains(body, "run_id") || strings.Contains(body, "reports-secret") { t.Fatalf("oversized response exposed run id or token: %s", body) @@ -210,8 +214,9 @@ func TestHTTPUploadSamePipelineRequestsSerialize(t *testing.T) { }, }) handler := uploadHTTPHandler{ - coordinator: coordinator, - tokens: map[string]string{"reports-secret": "reports"}, + coordinator: coordinator, + tokens: map[string]resolvedUploadToken{"reports-secret": uploadHTTPTestToken("reports-reporter", "reports-secret", "reports")}, + uploadPipelines: pipelineIDSet([]string{"reports"}), } server := httptest.NewServer(handler) defer server.Close() @@ -262,16 +267,17 @@ func TestHTTPUploadDifferentPipelinesRunConcurrently(t *testing.T) { }) handler := uploadHTTPHandler{ coordinator: coordinator, - tokens: map[string]string{ - "one-secret": "reports-one", - "two-secret": "reports-two", + tokens: map[string]resolvedUploadToken{ + "one-secret": uploadHTTPTestToken("reports-one-reporter", "one-secret", "reports-one"), + "two-secret": uploadHTTPTestToken("reports-two-reporter", "two-secret", "reports-two"), }, + uploadPipelines: pipelineIDSet([]string{"reports-one", "reports-two"}), } server := httptest.NewServer(handler) defer server.Close() - firstRunID := submitHTTPUpload(t, server, "one-secret", ingest.ContentTypeTar, []byte("first")) - secondRunID := submitHTTPUpload(t, server, "two-secret", ingest.ContentTypeTar, []byte("second")) + firstRunID := submitHTTPUploadToPipeline(t, server, "reports-one", "one-secret", ingest.ContentTypeTar, []byte("first")) + secondRunID := submitHTTPUploadToPipeline(t, server, "reports-two", "two-secret", ingest.ContentTypeTar, []byte("second")) waitForStartedPipelines(t, started, "reports-one", "reports-two") waitForHTTPUploadStatus(t, server, firstRunID, UploadStatusRunning) waitForHTTPUploadStatus(t, server, secondRunID, UploadStatusRunning) @@ -337,7 +343,12 @@ func httpUploadIntegrationConfig(t *testing.T, pipelines []httpUploadPipelineSpe func submitHTTPUpload(t *testing.T, server *httptest.Server, token, contentType string, body []byte) UploadRunID { t.Helper() - status, responseBody := postHTTPUpload(t, server, token, contentType, body) + return submitHTTPUploadToPipeline(t, server, "reports", token, contentType, body) +} + +func submitHTTPUploadToPipeline(t *testing.T, server *httptest.Server, pipelineID, token, contentType string, body []byte) UploadRunID { + t.Helper() + status, responseBody := postHTTPUploadToPipeline(t, server, pipelineID, token, contentType, body) return decodeAcceptedHTTPUpload(t, status, responseBody) } @@ -350,7 +361,7 @@ func submitHTTPUploadWithKey(t *testing.T, server *httptest.Server, token, conte func decodeAcceptedHTTPUpload(t *testing.T, status int, responseBody string) UploadRunID { t.Helper() if status != http.StatusAccepted { - t.Fatalf("POST /upload status = %d, want %d; body = %s", status, http.StatusAccepted, responseBody) + t.Fatalf("POST upload status = %d, want %d; body = %s", status, http.StatusAccepted, responseBody) } var accepted uploadAcceptedResponse if err := json.Unmarshal([]byte(responseBody), &accepted); err != nil { @@ -364,12 +375,22 @@ func decodeAcceptedHTTPUpload(t *testing.T, status int, responseBody string) Upl func postHTTPUpload(t *testing.T, server *httptest.Server, token, contentType string, body []byte) (int, string) { t.Helper() - return postHTTPUploadWithKey(t, server, token, contentType, "", body) + return postHTTPUploadToPipeline(t, server, "reports", token, contentType, body) +} + +func postHTTPUploadToPipeline(t *testing.T, server *httptest.Server, pipelineID, token, contentType string, body []byte) (int, string) { + t.Helper() + return postHTTPUploadWithKeyToPipeline(t, server, pipelineID, token, contentType, "", body) } func postHTTPUploadWithKey(t *testing.T, server *httptest.Server, token, contentType, key string, body []byte) (int, string) { t.Helper() - request, err := http.NewRequest(http.MethodPost, server.URL+"/upload", bytes.NewReader(body)) + return postHTTPUploadWithKeyToPipeline(t, server, "reports", token, contentType, key, body) +} + +func postHTTPUploadWithKeyToPipeline(t *testing.T, server *httptest.Server, pipelineID, token, contentType, key string, body []byte) (int, string) { + t.Helper() + request, err := http.NewRequest(http.MethodPost, server.URL+"/v1/pipelines/"+pipelineID+"/upload", bytes.NewReader(body)) if err != nil { t.Fatalf("NewRequest() error = %v", err) } @@ -380,7 +401,7 @@ func postHTTPUploadWithKey(t *testing.T, server *httptest.Server, token, content } response, err := server.Client().Do(request) if err != nil { - t.Fatalf("POST /upload error = %v", err) + t.Fatalf("POST upload error = %v", err) } defer response.Body.Close() data, err := io.ReadAll(response.Body) diff --git a/internal/app/upload_http_test.go b/internal/app/upload_http_test.go index 8bd6a24..a8a4518 100644 --- a/internal/app/upload_http_test.go +++ b/internal/app/upload_http_test.go @@ -75,6 +75,39 @@ func TestResolveUploadTokensFailsForMissingAndDuplicateTokens(t *testing.T) { } } +func TestResolveUploadTokensAllowsMultiplePipelines(t *testing.T) { + cfg := uploadHTTPTestConfig() + cfg.Pipelines = append(cfg.Pipelines, config.Pipeline{ + ID: "weekly", + Source: config.Backend{ + Backend: config.BackendHTTPUpload, + }, + Destinations: cfg.Pipelines[0].Destinations, + }) + cfg.UploadTokens[0].AllowPipelines = []string{"reports", "weekly"} + config.ApplyDefaults(&cfg) + + tokens, err := resolveUploadTokens(cfg, uploadHTTPTestEnvironment(map[string]string{ + "UPLOAD_TOKEN": "secret", + })) + if err != nil { + t.Fatalf("resolveUploadTokens() error = %v", err) + } + token, ok := tokens["secret"] + if !ok { + t.Fatal("resolved token missing") + } + if token.ID != "reporter" || token.Value != "secret" { + t.Fatalf("resolved token = %#v, want id and value", token) + } + if _, ok := token.AllowedPipelines["reports"]; !ok { + t.Fatalf("allowed pipelines = %#v, want reports", token.AllowedPipelines) + } + if _, ok := token.AllowedPipelines["weekly"]; !ok { + t.Fatalf("allowed pipelines = %#v, want weekly", token.AllowedPipelines) + } +} + func TestNewUploadHTTPHandlerAcceptsDefaultedConfig(t *testing.T) { cfg := uploadHTTPTestConfig() cfg.Server.HTTP.Bind = "" @@ -113,10 +146,11 @@ func TestUploadHTTPHandlerAuthenticatesAndAcceptsUpload(t *testing.T) { return UploadRunRecord{ID: "reports.20260603T120000Z.abcdef12", Status: UploadStatusAccepted}, nil }, }, - tokens: map[string]string{"valid-token": "reports"}, + tokens: map[string]resolvedUploadToken{"valid-token": uploadHTTPTestToken("reporter", "valid-token", "reports")}, + uploadPipelines: pipelineIDSet([]string{"reports"}), } recorder := httptest.NewRecorder() - request := httptest.NewRequest(http.MethodPost, "/upload", strings.NewReader("archive")) + request := httptest.NewRequest(http.MethodPost, "/v1/pipelines/reports/upload", strings.NewReader("archive")) request.Header.Set("Authorization", "Bearer valid-token") request.Header.Set("Content-Type", "application/x-tar") request.Header.Set("Idempotency-Key", "producer.retry:20260603") @@ -129,6 +163,9 @@ func TestUploadHTTPHandlerAuthenticatesAndAcceptsUpload(t *testing.T) { if submitted.PipelineID != "reports" { t.Fatalf("submitted pipeline = %q, want reports", submitted.PipelineID) } + if submitted.TokenID != "reporter" { + t.Fatalf("submitted token id = %q, want reporter", submitted.TokenID) + } if submitted.IdempotencyKey != "producer.retry:20260603" { t.Fatalf("submitted idempotency key = %q, want producer.retry:20260603", submitted.IdempotencyKey) } @@ -146,13 +183,14 @@ func TestUploadHTTPHandlerAuthenticatesAndAcceptsUpload(t *testing.T) { func TestUploadHTTPHandlerRejectsUnauthorizedRequests(t *testing.T) { handler := uploadHTTPHandler{ - coordinator: fakeUploadCoordinator{}, - tokens: map[string]string{"valid-token": "reports"}, + coordinator: fakeUploadCoordinator{}, + tokens: map[string]resolvedUploadToken{"valid-token": uploadHTTPTestToken("reporter", "valid-token", "reports")}, + uploadPipelines: pipelineIDSet([]string{"reports"}), } - for _, authHeader := range []string{"", "Bearer wrong-token"} { + for _, authHeader := range []string{"", "Basic valid-token", "Bearer", "Bearer wrong-token"} { recorder := httptest.NewRecorder() - request := httptest.NewRequest(http.MethodPost, "/upload", strings.NewReader("archive")) + request := httptest.NewRequest(http.MethodPost, "/v1/pipelines/reports/upload", strings.NewReader("archive")) request.Header.Set("Authorization", authHeader) request.Header.Set("Content-Type", "application/x-tar") @@ -167,7 +205,73 @@ func TestUploadHTTPHandlerRejectsUnauthorizedRequests(t *testing.T) { } } -func TestUploadHTTPHandlerRejectsUnsupportedContentTypeInvalidKeyAndPipelineID(t *testing.T) { +func TestUploadHTTPHandlerRejectsForbiddenPipeline(t *testing.T) { + handler := uploadHTTPHandler{ + coordinator: fakeUploadCoordinator{ + submit: func(context.Context, UploadRequest) (UploadRunRecord, error) { + t.Fatal("Submit should not be called") + return UploadRunRecord{}, nil + }, + }, + tokens: map[string]resolvedUploadToken{"valid-token": uploadHTTPTestToken("reporter", "valid-token", "reports")}, + uploadPipelines: pipelineIDSet([]string{"reports", "private"}), + } + recorder := httptest.NewRecorder() + request := httptest.NewRequest(http.MethodPost, "/v1/pipelines/private/upload", strings.NewReader("archive")) + request.Header.Set("Authorization", "Bearer valid-token") + request.Header.Set("Content-Type", "application/x-tar") + + handler.ServeHTTP(recorder, request) + + if recorder.Code != http.StatusForbidden { + t.Fatalf("status = %d, want %d; body = %q", recorder.Code, http.StatusForbidden, recorder.Body.String()) + } + if strings.Contains(recorder.Body.String(), "valid-token") { + t.Fatalf("forbidden response exposed token: %q", recorder.Body.String()) + } +} + +func TestUploadHTTPHandlerRejectsInvalidPathAndRemovedLegacyUpload(t *testing.T) { + handler := uploadHTTPHandler{ + coordinator: fakeUploadCoordinator{ + submit: func(context.Context, UploadRequest) (UploadRunRecord, error) { + t.Fatal("Submit should not be called") + return UploadRunRecord{}, nil + }, + }, + tokens: map[string]resolvedUploadToken{"valid-token": uploadHTTPTestToken("reporter", "valid-token", "reports")}, + uploadPipelines: pipelineIDSet([]string{"reports"}), + } + tests := []struct { + name string + url string + wantStatus int + }{ + {name: "legacy upload", url: "/upload", wantStatus: http.StatusNotFound}, + {name: "missing pipeline", url: "/v1/pipelines//upload", wantStatus: http.StatusNotFound}, + {name: "extra segment", url: "/v1/pipelines/reports/upload/extra", wantStatus: http.StatusNotFound}, + {name: "invalid pipeline id", url: "/v1/pipelines/.reports/upload", wantStatus: http.StatusBadRequest}, + {name: "pipeline query", url: "/v1/pipelines/reports/upload?pipeline=other", wantStatus: http.StatusBadRequest}, + {name: "pipeline id query", url: "/v1/pipelines/reports/upload?pipeline_id=other", wantStatus: http.StatusBadRequest}, + {name: "unknown upload pipeline", url: "/v1/pipelines/missing/upload", wantStatus: http.StatusNotFound}, + } + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + recorder := httptest.NewRecorder() + request := httptest.NewRequest(http.MethodPost, tt.url, strings.NewReader("archive")) + request.Header.Set("Authorization", "Bearer valid-token") + request.Header.Set("Content-Type", "application/x-tar") + + handler.ServeHTTP(recorder, request) + + if recorder.Code != tt.wantStatus { + t.Fatalf("status = %d, want %d; body = %q", recorder.Code, tt.wantStatus, recorder.Body.String()) + } + }) + } +} + +func TestUploadHTTPHandlerRejectsUnsupportedContentTypeAndInvalidKey(t *testing.T) { tests := []struct { name string url string @@ -178,14 +282,14 @@ func TestUploadHTTPHandlerRejectsUnsupportedContentTypeInvalidKeyAndPipelineID(t }{ { name: "unsupported content type", - url: "/upload", + url: "/v1/pipelines/reports/upload", contentType: "application/zip", body: strings.NewReader("archive"), wantStatus: http.StatusUnsupportedMediaType, }, { name: "invalid key syntax", - url: "/upload", + url: "/v1/pipelines/reports/upload", contentType: "application/x-tar", keyValues: []string{"bad key"}, body: strings.NewReader("archive"), @@ -193,7 +297,7 @@ func TestUploadHTTPHandlerRejectsUnsupportedContentTypeInvalidKeyAndPipelineID(t }, { name: "empty key", - url: "/upload", + url: "/v1/pipelines/reports/upload", contentType: "application/x-tar", keyValues: []string{""}, body: strings.NewReader("archive"), @@ -201,7 +305,7 @@ func TestUploadHTTPHandlerRejectsUnsupportedContentTypeInvalidKeyAndPipelineID(t }, { name: "too long key", - url: "/upload", + url: "/v1/pipelines/reports/upload", contentType: "application/x-tar", keyValues: []string{strings.Repeat("a", 129)}, body: strings.NewReader("archive"), @@ -209,19 +313,12 @@ func TestUploadHTTPHandlerRejectsUnsupportedContentTypeInvalidKeyAndPipelineID(t }, { name: "multiple keys", - url: "/upload", + url: "/v1/pipelines/reports/upload", contentType: "application/x-tar", keyValues: []string{"one", "two"}, body: strings.NewReader("archive"), wantStatus: http.StatusBadRequest, }, - { - name: "submitted pipeline id", - url: "/upload?pipeline_id=reports", - contentType: "application/x-tar", - body: strings.NewReader("archive"), - wantStatus: http.StatusBadRequest, - }, } for _, tt := range tests { t.Run(tt.name, func(t *testing.T) { @@ -232,7 +329,8 @@ func TestUploadHTTPHandlerRejectsUnsupportedContentTypeInvalidKeyAndPipelineID(t return UploadRunRecord{}, nil }, }, - tokens: map[string]string{"valid-token": "reports"}, + tokens: map[string]resolvedUploadToken{"valid-token": uploadHTTPTestToken("reporter", "valid-token", "reports")}, + uploadPipelines: pipelineIDSet([]string{"reports"}), } recorder := httptest.NewRecorder() request := httptest.NewRequest(http.MethodPost, tt.url, tt.body) @@ -274,10 +372,11 @@ func TestUploadHTTPHandlerMapsSubmitErrors(t *testing.T) { return UploadRunRecord{}, tt.err }, }, - tokens: map[string]string{"valid-token": "reports"}, + tokens: map[string]resolvedUploadToken{"valid-token": uploadHTTPTestToken("reporter", "valid-token", "reports")}, + uploadPipelines: pipelineIDSet([]string{"reports"}), } recorder := httptest.NewRecorder() - request := httptest.NewRequest(http.MethodPost, "/upload", strings.NewReader("archive")) + request := httptest.NewRequest(http.MethodPost, "/v1/pipelines/reports/upload", strings.NewReader("archive")) request.Header.Set("Authorization", "Bearer valid-token") request.Header.Set("Content-Type", "application/x-tar") @@ -310,7 +409,8 @@ func TestUploadHTTPHandlerRunStatusAndHealth(t *testing.T) { }, true }, }, - tokens: map[string]string{"valid-token": "reports"}, + tokens: map[string]resolvedUploadToken{"valid-token": uploadHTTPTestToken("reporter", "valid-token", "reports")}, + uploadPipelines: pipelineIDSet([]string{"reports"}), } recorder := httptest.NewRecorder() @@ -392,3 +492,11 @@ func uploadHTTPTestEnvironment(values map[string]string) config.Environment { return "", false }) } + +func uploadHTTPTestToken(id, value string, pipelines ...string) resolvedUploadToken { + return resolvedUploadToken{ + ID: id, + Value: value, + AllowedPipelines: pipelineIDSet(pipelines), + } +} diff --git a/internal/config/validate.go b/internal/config/validate.go index 2234bbe..52d912e 100644 --- a/internal/config/validate.go +++ b/internal/config/validate.go @@ -10,6 +10,10 @@ import ( var idPattern = regexp.MustCompile(`^[a-zA-Z0-9][a-zA-Z0-9._-]*$`) +func IsSlugLikeID(value string) bool { + return idPattern.MatchString(value) +} + type ValidationErrors []string func (e ValidationErrors) Error() string { @@ -34,7 +38,7 @@ func Validate(cfg Config) error { pipelineContext := fmt.Sprintf("pipelines[%d]", pipelineIndex) if pipeline.ID == "" { errs = append(errs, pipelineContext+".id is required") - } else if !idPattern.MatchString(pipeline.ID) { + } else if !IsSlugLikeID(pipeline.ID) { errs = append(errs, pipelineContext+".id must be a slug-like identifier") } else if _, exists := pipelineIDs[pipeline.ID]; exists { errs = append(errs, "pipeline id "+pipeline.ID+" is duplicated") @@ -56,7 +60,7 @@ func Validate(cfg Config) error { destinationContext := fmt.Sprintf("%s.destinations[%d]", pipelineContext, destinationIndex) if destination.ID == "" { errs = append(errs, destinationContext+".id is required") - } else if !idPattern.MatchString(destination.ID) { + } else if !IsSlugLikeID(destination.ID) { errs = append(errs, destinationContext+".id must be a slug-like identifier") } else if _, exists := destinationIDs[destination.ID]; exists { errs = append(errs, "destination id "+destination.ID+" is duplicated in pipeline "+pipeline.ID) @@ -144,7 +148,7 @@ func validateUploadTokens(errs ValidationErrors, tokens []UploadToken, pipelineI context := fmt.Sprintf("upload_tokens[%d]", tokenIndex) if token.ID == "" { errs = append(errs, context+".id is required") - } else if !idPattern.MatchString(token.ID) { + } else if !IsSlugLikeID(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")