From 0d346dcdf5abb61bd3533b8ec52e43a2ab045e9f Mon Sep 17 00:00:00 2001 From: Eric Rakestraw Date: Thu, 4 Jun 2026 00:14:32 +0000 Subject: [PATCH] Stage uploads before accepting HTTP runs --- docs/cli.md | 3 +- docs/config.md | 5 +- docs/internal/app.md | 28 ++--- docs/internal/ingest.md | 3 + docs/operations.md | 11 +- docs/troubleshooting.md | 18 +++ internal/app/upload_coordinator.go | 112 ++++++++++--------- internal/app/upload_http.go | 57 ++-------- internal/app/upload_http_integration_test.go | 97 +++++++++++----- internal/app/upload_http_test.go | 52 ++++++--- internal/ingest/archive.go | 5 + internal/ingest/archive_test.go | 19 ++++ 12 files changed, 248 insertions(+), 162 deletions(-) diff --git a/docs/cli.md b/docs/cli.md index f206f05..504fbbf 100644 --- a/docs/cli.md +++ b/docs/cli.md @@ -149,7 +149,8 @@ curl -X POST http://127.0.0.1:8080/upload \ --data-binary @bundle.tar.gz ``` -The upload response is accepted asynchronously: +The upload response is returned after the archive is staged and validated; the +destination fan-out continues asynchronously: ```json {"run_id":"reports.20260603T120000Z.abcdef12","status":"accepted"} diff --git a/docs/config.md b/docs/config.md index 1c32d14..167a895 100644 --- a/docs/config.md +++ b/docs/config.md @@ -130,12 +130,15 @@ Accepted upload content types: - `application/gzip` - `application/x-gzip` -Accepted uploads return: +Accepted uploads return after the archive is staged and validated: ```json {"run_id":"","status":"accepted"} ``` +Malformed tar or gzip content and invalid staged bundles are rejected before a +run id is issued. + The run id can be queried through `GET /runs/` while the status record is retained in memory. Completed records expire after `server.http.retention`; expiration also removes committed staged bundle directories for completed diff --git a/docs/internal/app.md b/docs/internal/app.md index 3e788ba..268a5ba 100644 --- a/docs/internal/app.md +++ b/docs/internal/app.md @@ -87,10 +87,10 @@ prepared. ## Upload Coordination `UploadCoordinator` owns in-memory coordination for asynchronous upload -processing. It admits uploads for configured `http_upload` pipelines, generates -run IDs, tracks status records, stages accepted archives through -`internal/ingest`, and executes the selected pipeline through -`RunPipelineWithLocalSource`. +processing. It admits uploads for configured `http_upload` pipelines, reserves +queue capacity before request-body staging, stages and validates archives +through `internal/ingest`, tracks accepted status records, and executes the +selected pipeline through `RunPipelineWithLocalSource`. Upload run IDs use: @@ -111,10 +111,11 @@ The coordinator records these statuses: - `expired` Admission is bounded by `server.http.queue_size`. Full queues are rejected -before the upload body is staged. Execution is bounded by +before the upload body is read. Successfully reserved uploads are staged and +validated before an accepted run record is created. Execution is bounded by `server.http.max_concurrency`, and only one upload for a given pipeline may run -at a time. Later uploads for the same pipeline remain queued until the active -run finishes. +at a time. Later accepted uploads for the same pipeline remain queued until the +active run finishes. Completed records retain the final run report or error text until `server.http.retention` elapses. Expiration removes completed status records and @@ -137,15 +138,16 @@ pipeline ids, but not token values. Routes: - `GET /healthz`: returns `200` after config, secrets, tokens, coordinator, and route setup succeed. -- `POST /upload`: accepts authenticated tar and tar.gz archives and returns an accepted run id. +- `POST /upload`: stages and validates an authenticated tar or tar.gz archive, then returns an accepted run id. - `GET /runs/`: returns the current in-memory upload status record or `404`. The upload token maps to exactly one configured pipeline. Producers do not submit pipeline ids, and submitted `pipeline` or `pipeline_id` query values are -rejected. Full queues are rejected before the request body is read. Oversized -uploads, unsupported content types, invalid bearer tokens, full queues, and -unknown status records are mapped to stable HTTP status codes without returning -secret token values. +rejected. Full queues are rejected before the request body is read. Malformed +archives and invalid staged bundles are rejected before a run id is issued. +Oversized uploads, unsupported content types, invalid bearer tokens, full +queues, and unknown status records are mapped to stable HTTP status codes +without returning secret token values. ## Coordination @@ -201,7 +203,7 @@ Run helpers are grouped by responsibility: - `run_warnings.go`: secret and SSH warning records. - `run_notify.go`: notification event projection and action filtering. - `run_coordinator.go`: in-memory run admission, run IDs, status records, and duplicate-run errors. -- `upload_coordinator.go`: in-memory upload admission, queueing, status tracking, staging handoff, and staged-source execution. +- `upload_coordinator.go`: in-memory upload admission, queue reservation, staging handoff, status tracking, queueing, and staged-source execution. - `upload_http.go`: HTTP upload authentication, routes, JSON response projection, and HTTP error mapping. - `serve.go`: config/secrets loading and HTTP server startup. - `backends.go`: app-level backend factory wiring. diff --git a/docs/internal/ingest.md b/docs/internal/ingest.md index 5e86574..7720ee1 100644 --- a/docs/internal/ingest.md +++ b/docs/internal/ingest.md @@ -6,6 +6,9 @@ ## Archive staging +`ValidateContentType` owns accepted upload content-type policy for archive +staging callers. + `StageArchive` accepts one upload body, content type, pipeline staging path, run id, and explicit size and file-count limits. It writes the request body to temporary storage while enforcing the configured upload size limit, extracts the archive into temporary local storage, validates the extracted source bundle, and then commits the validated bundle to: ```text diff --git a/docs/operations.md b/docs/operations.md index 7b6c20a..c2e6324 100644 --- a/docs/operations.md +++ b/docs/operations.md @@ -71,9 +71,9 @@ go run ./cmd/distributor validate --config examples/local-publish.yml --pipeline ## HTTP Upload Workflow `distributor serve` runs the HTTP upload API for pipelines whose source backend -is `http_upload`. Each upload token maps to one configured pipeline, and each -accepted archive is staged, validated, and published through the same -destination fan-out path used by local source runs. +is `http_upload`. Each upload token maps to one configured pipeline. A valid +archive is staged and validated before a run id is returned, then published +through the same destination fan-out path used by local source runs. Minimal local HTTP upload configuration: @@ -119,7 +119,7 @@ curl -X POST http://127.0.0.1:8080/upload \ --data-binary @bundle.tar.gz ``` -Successful admission returns a run id: +Successful staging and admission returns a run id: ```json {"run_id":"reports.20260603T120000Z.abcdef12","status":"accepted"} @@ -136,6 +136,9 @@ or error details on failure. Status is memory-only and expires after `server.http.retention`; completed staged bundle directories are removed on expiry. Restarting the process clears upload status and queue state. +Malformed archives and invalid source bundles are rejected by `POST /upload` +before a run id is issued. + Use `GET /healthz` for readiness after config and tokens load: ```sh diff --git a/docs/troubleshooting.md b/docs/troubleshooting.md index d94f8e2..83b08e5 100644 --- a/docs/troubleshooting.md +++ b/docs/troubleshooting.md @@ -102,6 +102,24 @@ curl -i -X POST http://127.0.0.1:8080/upload \ Safe fix: use the token value resolved by the configured `token_env`. Do not include token values in logs or tickets. +## `POST /upload` returns `400` + +Likely cause: the archive content is malformed, the gzip body is invalid, the +tar body cannot be extracted safely, or the extracted source bundle fails +manifest and file validation. + +Diagnostic: + +```sh +tar -tf bundle.tar +tar -tzf bundle.tar.gz +go run ./cmd/distributor validate +``` + +Safe fix: rebuild the tar or tar.gz archive from one complete source bundle +root. The archive must contain exactly one root-level `manifest.json`, and every +manifest-listed file must exist as a regular file with matching size and digest. + ## `POST /upload` returns `413` Likely cause: the request body exceeds the selected pipeline's diff --git a/internal/app/upload_coordinator.go b/internal/app/upload_coordinator.go index a47bad7..e35431a 100644 --- a/internal/app/upload_coordinator.go +++ b/internal/app/upload_coordinator.go @@ -78,6 +78,7 @@ type UploadCoordinator struct { queueSize int maxConcurrency int runningCount int + reservedCount int activePipeline map[string]bool pending []*uploadJob records map[UploadRunID]UploadRunRecord @@ -88,9 +89,10 @@ type uploadStageFunc func(context.Context, ingest.StageOptions) (ingest.StagedBu type uploadRunFunc func(context.Context, config.Config, RunPipelineWithLocalSourceOptions) (RunReport, error) type uploadJob struct { - recordID UploadRunID - request UploadRequest - pipeline config.Pipeline + recordID UploadRunID + request UploadRequest + pipeline config.Pipeline + stagedRoot string } type uploadCoordinatorHooks struct { @@ -164,24 +166,49 @@ func (coordinator *UploadCoordinator) Submit(ctx context.Context, request Upload if err != nil { return UploadRunRecord{}, err } + if err := ingest.ValidateContentType(request.ContentType); err != nil { + return UploadRunRecord{}, err + } + + coordinator.mu.Lock() + coordinator.expireLocked(coordinator.now().UTC()) + if coordinator.queueFullLocked() { + coordinator.mu.Unlock() + return UploadRunRecord{}, UploadQueueFullError{QueueSize: coordinator.queueSize} + } + coordinator.reservedCount++ + coordinator.mu.Unlock() + + staged, err := coordinator.stage(ctx, ingest.StageOptions{ + Body: request.Body, + ContentType: request.ContentType, + PipelineStagingPath: pipeline.Source.Upload.StagingPath, + RunID: string(runID), + MaxUploadSize: int64(*pipeline.Source.Upload.MaxUploadSize), + MaxExtractedSize: int64(*pipeline.Source.Upload.MaxUploadSize), + MaxFileCount: uploadMaxFileCount(request.MaxFileCount), + }) + if err != nil { + coordinator.releaseReservation() + return UploadRunRecord{}, err + } coordinator.mu.Lock() defer coordinator.mu.Unlock() - coordinator.expireLocked(coordinator.now().UTC()) - if len(coordinator.pending) >= coordinator.queueSize { - return UploadRunRecord{}, UploadQueueFullError{QueueSize: coordinator.queueSize} - } + coordinator.reservedCount-- record := UploadRunRecord{ ID: runID, PipelineID: pipeline.ID, Status: UploadStatusAccepted, AcceptedAt: coordinator.now().UTC(), + StagedRoot: staged.Root, } coordinator.records[runID] = record coordinator.pending = append(coordinator.pending, &uploadJob{ - recordID: runID, - request: request, - pipeline: pipeline, + recordID: runID, + request: request, + pipeline: pipeline, + stagedRoot: staged.Root, }) coordinator.notify() return record, nil @@ -205,13 +232,13 @@ func (coordinator *UploadCoordinator) CanAccept() bool { coordinator.mu.Lock() defer coordinator.mu.Unlock() coordinator.expireLocked(coordinator.now().UTC()) - return len(coordinator.pending) < coordinator.queueSize + return !coordinator.queueFullLocked() } func (coordinator *UploadCoordinator) QueueDepth() int { coordinator.mu.Lock() defer coordinator.mu.Unlock() - return len(coordinator.pending) + return len(coordinator.pending) + coordinator.reservedCount } func (coordinator *UploadCoordinator) RunningCount() int { @@ -290,47 +317,30 @@ func (coordinator *UploadCoordinator) markPendingQueuedLocked() { } func (coordinator *UploadCoordinator) runJob(job *uploadJob) { - record := coordinator.currentRecord(job.recordID) - maxFileCount := job.request.MaxFileCount - if maxFileCount <= 0 { - maxFileCount = DefaultUploadMaxFileCount - } - staged, err := coordinator.stage(coordinator.ctx, ingest.StageOptions{ - Body: job.request.Body, - ContentType: job.request.ContentType, - PipelineStagingPath: job.pipeline.Source.Upload.StagingPath, - RunID: string(record.ID), - MaxUploadSize: int64(*job.pipeline.Source.Upload.MaxUploadSize), - MaxExtractedSize: int64(*job.pipeline.Source.Upload.MaxUploadSize), - MaxFileCount: maxFileCount, + report, err := coordinator.run(coordinator.ctx, coordinator.cfg, RunPipelineWithLocalSourceOptions{ + PipelineID: job.pipeline.ID, + SourceRoot: job.stagedRoot, + DryRun: job.request.DryRun, + Force: job.request.Force, }) - if err == nil { - coordinator.setStagedRoot(job.recordID, staged.Root) - var report RunReport - report, err = coordinator.run(coordinator.ctx, coordinator.cfg, RunPipelineWithLocalSourceOptions{ - PipelineID: job.pipeline.ID, - SourceRoot: staged.Root, - DryRun: job.request.DryRun, - Force: job.request.Force, - }) - coordinator.complete(job, &report, err) - return + coordinator.complete(job, &report, err) +} + +func (coordinator *UploadCoordinator) releaseReservation() { + coordinator.mu.Lock() + defer coordinator.mu.Unlock() + coordinator.reservedCount-- +} + +func (coordinator *UploadCoordinator) queueFullLocked() bool { + return len(coordinator.pending)+coordinator.reservedCount >= coordinator.queueSize +} + +func uploadMaxFileCount(value int) int { + if value > 0 { + return value } - coordinator.complete(job, nil, err) -} - -func (coordinator *UploadCoordinator) currentRecord(runID UploadRunID) UploadRunRecord { - coordinator.mu.Lock() - defer coordinator.mu.Unlock() - return coordinator.records[runID] -} - -func (coordinator *UploadCoordinator) setStagedRoot(runID UploadRunID, root string) { - coordinator.mu.Lock() - defer coordinator.mu.Unlock() - record := coordinator.records[runID] - record.StagedRoot = root - coordinator.records[runID] = record + return DefaultUploadMaxFileCount } func (coordinator *UploadCoordinator) complete(job *uploadJob, report *RunReport, runErr error) { diff --git a/internal/app/upload_http.go b/internal/app/upload_http.go index 5153344..c1246c5 100644 --- a/internal/app/upload_http.go +++ b/internal/app/upload_http.go @@ -1,13 +1,10 @@ package app import ( - "bytes" "context" "encoding/json" "errors" "fmt" - "io" - "mime" "net/http" "strings" @@ -24,7 +21,6 @@ type uploadCoordinator interface { type uploadHTTPHandler struct { coordinator uploadCoordinator tokens map[string]string - limits map[string]int64 } type uploadAcceptedResponse struct { @@ -38,20 +34,18 @@ type httpErrorResponse struct { func newUploadHTTPHandler(ctx context.Context, cfg config.Config, environment config.Environment) (http.Handler, error) { config.ApplyDefaults(&cfg) - tokens, limits, err := resolveUploadTokens(cfg, environment) + tokens, err := resolveUploadTokens(cfg, environment) if err != nil { return nil, err } return uploadHTTPHandler{ coordinator: NewUploadCoordinator(ctx, cfg), tokens: tokens, - limits: limits, }, nil } -func resolveUploadTokens(cfg config.Config, environment config.Environment) (map[string]string, map[string]int64, error) { +func resolveUploadTokens(cfg config.Config, environment config.Environment) (map[string]string, error) { tokens := make(map[string]string) - limits := make(map[string]int64) for _, pipeline := range cfg.Pipelines { if pipeline.Source.Backend != config.BackendHTTPUpload { continue @@ -59,18 +53,17 @@ func resolveUploadTokens(cfg config.Config, environment config.Environment) (map tokenName := pipeline.Source.Upload.TokenEnv token, ok := environment.Lookup(tokenName) if !ok { - return nil, nil, fmt.Errorf("upload token environment variable %s is not set", tokenName) + return nil, fmt.Errorf("upload token environment variable %s is not set", tokenName) } if token == "" { - return nil, nil, fmt.Errorf("upload token environment variable %s is empty", tokenName) + return nil, fmt.Errorf("upload token environment variable %s is empty", tokenName) } if existing, exists := tokens[token]; exists { - return nil, 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, pipeline.ID) } tokens[token] = pipeline.ID - limits[pipeline.ID] = int64(*pipeline.Source.Upload.MaxUploadSize) } - return tokens, limits, nil + return tokens, nil } func (handler uploadHTTPHandler) ServeHTTP(w http.ResponseWriter, r *http.Request) { @@ -101,7 +94,7 @@ func (handler uploadHTTPHandler) handleUpload(w http.ResponseWriter, r *http.Req return } contentType := r.Header.Get("Content-Type") - if !supportedUploadContentType(contentType) { + if err := ingest.ValidateContentType(contentType); err != nil { writeHTTPError(w, http.StatusUnsupportedMediaType, "unsupported content type") return } @@ -109,19 +102,10 @@ func (handler uploadHTTPHandler) handleUpload(w http.ResponseWriter, r *http.Req writeHTTPError(w, http.StatusServiceUnavailable, "upload queue is full") return } - body, err := readUploadBody(r.Body, handler.limits[pipelineID]) - if err != nil { - if errors.Is(err, ingest.ErrUploadTooLarge) { - writeHTTPError(w, http.StatusRequestEntityTooLarge, "upload exceeds maximum size") - return - } - writeHTTPError(w, http.StatusBadRequest, "read upload body failed") - return - } record, err := handler.coordinator.Submit(r.Context(), UploadRequest{ PipelineID: pipelineID, ContentType: contentType, - Body: bytes.NewReader(body), + Body: r.Body, }) if err != nil { writeUploadSubmitError(w, err) @@ -160,31 +144,6 @@ func (handler uploadHTTPHandler) authenticate(header string) (string, bool) { return pipelineID, ok } -func supportedUploadContentType(contentType string) bool { - mediaType, _, err := mime.ParseMediaType(contentType) - if err != nil { - mediaType = contentType - } - switch mediaType { - case ingest.ContentTypeTar, ingest.ContentTypeGzip, ingest.ContentTypeXGzip: - return true - default: - return false - } -} - -func readUploadBody(body io.Reader, maxSize int64) ([]byte, error) { - limited := &io.LimitedReader{R: body, N: maxSize + 1} - data, err := io.ReadAll(limited) - if err != nil { - return nil, err - } - if int64(len(data)) > maxSize { - return nil, ingest.ErrUploadTooLarge - } - return data, nil -} - func writeUploadSubmitError(w http.ResponseWriter, err error) { switch { case IsUploadQueueFull(err): diff --git a/internal/app/upload_http_integration_test.go b/internal/app/upload_http_integration_test.go index ca98e1d..bb91088 100644 --- a/internal/app/upload_http_integration_test.go +++ b/internal/app/upload_http_integration_test.go @@ -13,6 +13,7 @@ import ( "net/http/httptest" "os" "path/filepath" + "strings" "testing" "time" @@ -68,32 +69,65 @@ func TestHTTPUploadPublishesTarAndGzipFanout(t *testing.T) { } } -func TestHTTPUploadInvalidArchiveFailsWithoutPublishing(t *testing.T) { +func TestHTTPUploadInvalidArchiveIsRejectedWithoutRunID(t *testing.T) { destination := t.TempDir() - cfg := httpUploadIntegrationConfig(t, []httpUploadPipelineSpec{{ + coordinator := NewUploadCoordinator(context.Background(), httpUploadIntegrationConfig(t, []httpUploadPipelineSpec{{ id: "reports", tokenEnv: "REPORTS_TOKEN", stagingPath: filepath.Join(t.TempDir(), "reports"), destinations: []string{destination}, - }}, 4, 1) - handler, err := newUploadHTTPHandler(context.Background(), cfg, uploadHTTPTestEnvironment(map[string]string{ - "REPORTS_TOKEN": "reports-secret", - })) - if err != nil { - t.Fatalf("newUploadHTTPHandler() error = %v", err) + }}, 4, 1)) + handler := uploadHTTPHandler{ + coordinator: coordinator, + tokens: map[string]string{"reports-secret": "reports"}, } server := httptest.NewServer(handler) defer server.Close() - runID := submitHTTPUpload(t, server, "reports-secret", ingest.ContentTypeTar, []byte("not a tar archive")) - record := waitForHTTPUploadStatus(t, server, runID, UploadStatusFailed) + 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) + } + if strings.Contains(body, "run_id") || strings.Contains(body, "reports-secret") { + t.Fatalf("invalid archive response exposed run id or token: %s", body) + } + if got := coordinator.QueueDepth(); got != 0 { + t.Fatalf("queue depth = %d, want 0", got) + } + assertDirectoryEmpty(t, destination) +} - if record.Error == "" { - t.Fatal("failed status error is empty") +func TestHTTPUploadOversizedArchiveIsRejectedWithoutRunID(t *testing.T) { + destination := t.TempDir() + stagingPath := filepath.Join(t.TempDir(), "reports") + cfg := httpUploadIntegrationConfig(t, []httpUploadPipelineSpec{{ + id: "reports", + tokenEnv: "REPORTS_TOKEN", + stagingPath: stagingPath, + destinations: []string{destination}, + }}, 4, 1) + size := config.ByteSize(4) + cfg.Server.HTTP.MaxUploadSize = &size + cfg.Pipelines[0].Source.Upload.MaxUploadSize = &size + coordinator := NewUploadCoordinator(context.Background(), cfg) + handler := uploadHTTPHandler{ + coordinator: coordinator, + tokens: map[string]string{"reports-secret": "reports"}, } - if record.Report != nil { - t.Fatalf("failed staging report = %#v, want nil", record.Report) + 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) } + if strings.Contains(body, "run_id") || strings.Contains(body, "reports-secret") { + t.Fatalf("oversized response exposed run id or token: %s", body) + } + if got := coordinator.QueueDepth(); got != 0 { + t.Fatalf("queue depth = %d, want 0", got) + } + assertDirectoryEmpty(t, stagingPath) assertDirectoryEmpty(t, destination) } @@ -122,7 +156,6 @@ func TestHTTPUploadSamePipelineRequestsSerialize(t *testing.T) { handler := uploadHTTPHandler{ coordinator: coordinator, tokens: map[string]string{"reports-secret": "reports"}, - limits: map[string]int64{"reports": 1024}, } server := httptest.NewServer(handler) defer server.Close() @@ -177,10 +210,6 @@ func TestHTTPUploadDifferentPipelinesRunConcurrently(t *testing.T) { "one-secret": "reports-one", "two-secret": "reports-two", }, - limits: map[string]int64{ - "reports-one": 1024, - "reports-two": 1024, - }, } server := httptest.NewServer(handler) defer server.Close() @@ -247,6 +276,22 @@ 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) + if status != http.StatusAccepted { + 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 { + t.Fatalf("decode accepted response: %v", err) + } + if accepted.RunID == "" || accepted.Status != UploadStatusAccepted { + t.Fatalf("accepted response = %#v, want run id and accepted status", accepted) + } + return accepted.RunID +} + +func postHTTPUpload(t *testing.T, server *httptest.Server, token, contentType string, body []byte) (int, string) { t.Helper() request, err := http.NewRequest(http.MethodPost, server.URL+"/upload", bytes.NewReader(body)) if err != nil { @@ -259,17 +304,11 @@ func submitHTTPUpload(t *testing.T, server *httptest.Server, token, contentType t.Fatalf("POST /upload error = %v", err) } defer response.Body.Close() - if response.StatusCode != http.StatusAccepted { - t.Fatalf("POST /upload status = %d, want %d", response.StatusCode, http.StatusAccepted) + data, err := io.ReadAll(response.Body) + if err != nil { + t.Fatalf("read response body: %v", err) } - var accepted uploadAcceptedResponse - if err := json.NewDecoder(response.Body).Decode(&accepted); err != nil { - t.Fatalf("decode accepted response: %v", err) - } - if accepted.RunID == "" || accepted.Status != UploadStatusAccepted { - t.Fatalf("accepted response = %#v, want run id and accepted status", accepted) - } - return accepted.RunID + return response.StatusCode, string(data) } func waitForHTTPUploadStatus(t *testing.T, server *httptest.Server, runID UploadRunID, status UploadStatus) UploadRunRecord { diff --git a/internal/app/upload_http_test.go b/internal/app/upload_http_test.go index 323df4c..335b882 100644 --- a/internal/app/upload_http_test.go +++ b/internal/app/upload_http_test.go @@ -12,6 +12,7 @@ import ( "time" "gitea.maximumdirect.net/eric/distributor/internal/config" + "gitea.maximumdirect.net/eric/distributor/internal/ingest" ) type fakeUploadCoordinator struct { @@ -41,7 +42,7 @@ func (fake fakeUploadCoordinator) Status(runID UploadRunID) (UploadRunRecord, bo func TestResolveUploadTokensFailsForMissingAndDuplicateTokens(t *testing.T) { cfg := uploadHTTPTestConfig() - _, _, err := resolveUploadTokens(cfg, config.NewEnvironment(nil, func(string) (string, bool) { + _, err := resolveUploadTokens(cfg, config.NewEnvironment(nil, func(string) (string, bool) { return "", false })) if err == nil || !strings.Contains(err.Error(), "UPLOAD_TOKEN") { @@ -58,7 +59,7 @@ func TestResolveUploadTokensFailsForMissingAndDuplicateTokens(t *testing.T) { }) config.ApplyDefaults(&cfg) secret := "super-secret-token" - _, _, err = resolveUploadTokens(cfg, uploadHTTPTestEnvironment(map[string]string{ + _, err = resolveUploadTokens(cfg, uploadHTTPTestEnvironment(map[string]string{ "UPLOAD_TOKEN": secret, "OTHER_UPLOAD_TOKEN": secret, })) @@ -110,7 +111,6 @@ func TestUploadHTTPHandlerAuthenticatesAndAcceptsUpload(t *testing.T) { }, }, tokens: map[string]string{"valid-token": "reports"}, - limits: map[string]int64{"reports": 1024}, } recorder := httptest.NewRecorder() request := httptest.NewRequest(http.MethodPost, "/upload", strings.NewReader("archive")) @@ -141,7 +141,6 @@ func TestUploadHTTPHandlerRejectsUnauthorizedRequests(t *testing.T) { handler := uploadHTTPHandler{ coordinator: fakeUploadCoordinator{canAccept: true}, tokens: map[string]string{"valid-token": "reports"}, - limits: map[string]int64{"reports": 1024}, } for _, authHeader := range []string{"", "Bearer wrong-token"} { @@ -178,14 +177,6 @@ func TestUploadHTTPHandlerRejectsUnsupportedOversizedFullQueueAndPipelineID(t *t body: strings.NewReader("archive"), wantStatus: http.StatusUnsupportedMediaType, }, - { - name: "oversized", - canAccept: true, - url: "/upload", - contentType: "application/x-tar", - body: strings.NewReader("too-large"), - wantStatus: http.StatusRequestEntityTooLarge, - }, { name: "full queue", canAccept: false, @@ -214,7 +205,6 @@ func TestUploadHTTPHandlerRejectsUnsupportedOversizedFullQueueAndPipelineID(t *t }, }, tokens: map[string]string{"valid-token": "reports"}, - limits: map[string]int64{"reports": 4}, } recorder := httptest.NewRecorder() request := httptest.NewRequest(http.MethodPost, tt.url, tt.body) @@ -233,6 +223,41 @@ func TestUploadHTTPHandlerRejectsUnsupportedOversizedFullQueueAndPipelineID(t *t } } +func TestUploadHTTPHandlerMapsSubmitErrors(t *testing.T) { + tests := []struct { + name string + err error + wantStatus int + }{ + {name: "oversized", err: ingest.ErrUploadTooLarge, wantStatus: http.StatusRequestEntityTooLarge}, + {name: "unsupported", err: ingest.ErrUnsupportedContentType, wantStatus: http.StatusUnsupportedMediaType}, + {name: "malformed", err: errors.New("malformed archive"), wantStatus: http.StatusBadRequest}, + } + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + handler := uploadHTTPHandler{ + coordinator: fakeUploadCoordinator{ + canAccept: true, + submit: func(context.Context, UploadRequest) (UploadRunRecord, error) { + return UploadRunRecord{}, tt.err + }, + }, + tokens: map[string]string{"valid-token": "reports"}, + } + recorder := httptest.NewRecorder() + request := httptest.NewRequest(http.MethodPost, "/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 != tt.wantStatus { + t.Fatalf("status = %d, want %d; body = %q", recorder.Code, tt.wantStatus, recorder.Body.String()) + } + }) + } +} + func TestUploadHTTPHandlerRunStatusAndHealth(t *testing.T) { finishedAt := time.Date(2026, 6, 3, 12, 0, 0, 0, time.UTC) handler := uploadHTTPHandler{ @@ -251,7 +276,6 @@ func TestUploadHTTPHandlerRunStatusAndHealth(t *testing.T) { }, }, tokens: map[string]string{"valid-token": "reports"}, - limits: map[string]int64{"reports": 1024}, } recorder := httptest.NewRecorder() diff --git a/internal/ingest/archive.go b/internal/ingest/archive.go index c2dc9e5..7d15fb6 100644 --- a/internal/ingest/archive.go +++ b/internal/ingest/archive.go @@ -138,6 +138,11 @@ func validateRunID(value string) error { return nil } +func ValidateContentType(contentType string) error { + _, err := archiveFormat(contentType) + return err +} + type archiveKind int const ( diff --git a/internal/ingest/archive_test.go b/internal/ingest/archive_test.go index efd2f1d..c827762 100644 --- a/internal/ingest/archive_test.go +++ b/internal/ingest/archive_test.go @@ -47,6 +47,25 @@ func TestStageArchiveRejectsUnsupportedContentType(t *testing.T) { } } +func TestValidateContentType(t *testing.T) { + for _, contentType := range []string{ + ContentTypeTar, + ContentTypeGzip, + ContentTypeXGzip, + ContentTypeGzip + "; charset=binary", + } { + t.Run(contentType, func(t *testing.T) { + if err := ValidateContentType(contentType); err != nil { + t.Fatalf("ValidateContentType() error = %v", err) + } + }) + } + + if err := ValidateContentType("application/zip"); !errors.Is(err, ErrUnsupportedContentType) { + t.Fatalf("ValidateContentType() error = %v, want ErrUnsupportedContentType", err) + } +} + func TestStageArchiveEnforcesMaxUploadSize(t *testing.T) { archive := validArchive(t, false) err := stageArchiveError(t, archive, ContentTypeTar, func(opts *StageOptions) {