diff --git a/docs/cli.md b/docs/cli.md index d8ea8ef..7ec1a56 100644 --- a/docs/cli.md +++ b/docs/cli.md @@ -14,6 +14,7 @@ This discovers the example source bundle and publishes source files to `workspac distributor [--help] distributor version [--format text|json] distributor run [--config ] [--dry-run] [--force] [--format text|json] +distributor serve [--config ] distributor validate [--format text|json] distributor validate --config --pipeline [--bundle ] [--format text|json] distributor inspect [--format text|json] @@ -23,11 +24,12 @@ distributor manifest create --id [options] - `version`: prints the application name and version. Development builds print `distributor dev`. - `run`: loads a YAML config, discovers source bundles, plans each configured destination, writes selected outputs unless `--dry-run` is set, and prints a final status summary. +- `serve`: loads a YAML config, resolves HTTP upload bearer tokens, and runs the HTTP upload API. - `validate`: validates a local source bundle directory, a local source bundle tree, or one configured pipeline source. - `inspect`: validates source bundles and prints normalized bundle metadata for a local path or one configured pipeline source. - `manifest create`: creates `manifest.json` for a local source bundle directory. -`validate` and `inspect` have two mutually exclusive modes: a local path shortcut, or configured source mode with `--config --pipeline `. Configured source mode opens only the selected pipeline source and supports configured `local`, `ssh`, and `s3` sources. It does not open destinations. `run` executes configured sources and destinations. +`validate` and `inspect` have two mutually exclusive modes: a local path shortcut, or configured source mode with `--config --pipeline `. Configured source mode opens only the selected pipeline source and supports configured `local`, `ssh`, and `s3` sources. It does not open destinations. `run` executes configured `local`, `ssh`, and `s3` sources and destinations. `serve` executes configured `http_upload` sources through the upload API and normal destination fan-out. ## Flag reference @@ -49,6 +51,10 @@ Output-producing subcommands: - `--dry-run`: load config, discover bundles, inspect destination state, print planned actions and final status, and do not write output files, destination state, or SSH `known_hosts` entries. - `--force`: allow explicit destructive replacement for supported conflict cases in this run only. +`serve` flags: + +- `--config `: config file to load. If omitted, `serve` uses `/usr/local/etc/distributor/config.yml`. + `validate` and `inspect` configured source flags: - `--config `: config file to load for source validation or inspection. Required in configured source mode. @@ -127,6 +133,39 @@ Publish the local HTML example: go run ./cmd/distributor run --config examples/local-html.yml ``` +Start the HTTP upload API: + +```sh +go run ./cmd/distributor serve --config +``` + +Upload an archive to the configured `http_upload` pipeline associated with a bearer token: + +```sh +curl -X POST http://127.0.0.1:8080/upload \ + -H "Authorization: Bearer $DISTRIBUTOR_UPLOAD_TOKEN" \ + -H "Content-Type: application/gzip" \ + --data-binary @bundle.tar.gz +``` + +The upload response is accepted asynchronously: + +```json +{"run_id":"reports.20260603T120000Z.abcdef12","status":"accepted"} +``` + +Check upload status: + +```sh +curl http://127.0.0.1:8080/runs/ +``` + +Check server readiness: + +```sh +curl http://127.0.0.1:8080/healthz +``` + Preview local fan-out publication: ```sh diff --git a/docs/config.md b/docs/config.md index 4b799f9..d8633b8 100644 --- a/docs/config.md +++ b/docs/config.md @@ -2,15 +2,18 @@ ## Config File Location -`distributor run --config ` loads the YAML config at the provided path. +`distributor run --config ` and `distributor serve --config ` load +the YAML config at the provided path. -If `--config` is omitted, `run` uses: +If `--config` is omitted, both commands use: ```text /usr/local/etc/distributor/config.yml ``` -Config parsing rejects unknown YAML fields. The executable `run` backends are `local`, `ssh`, and `s3`. The schema also accepts `http_upload` as a source-only ingestion backend configuration. +Config parsing rejects unknown YAML fields. The executable `run` backends are +`local`, `ssh`, and `s3`. The `serve` command executes `http_upload` sources +through the HTTP upload API and normal destination fan-out. ## Minimal Local Config @@ -62,7 +65,9 @@ pipelines: ## HTTP Upload Source Configuration -HTTP upload sources are configured as pipeline sources only. They are not valid destination backends. +HTTP upload sources are configured as pipeline sources only. They are not valid +destination backends. `distributor serve` maps each configured upload token to +exactly one `http_upload` pipeline. ```yaml server: @@ -92,6 +97,50 @@ pipelines: `source.max_upload_size` is optional. When omitted, it defaults to `server.http.max_upload_size`. +The server resolves each `token_env` through the real process environment and +the configured `secrets.directory` resolver. Startup fails if any configured +upload token is missing, empty, or resolves to the same value as another upload +pipeline. Token values are not read from YAML and are not printed in API +responses. + +## HTTP Upload API + +`distributor serve` binds to `server.http.bind`, which defaults to +`127.0.0.1:8080`. + +Routes: + +- `GET /healthz`: returns readiness status after config and upload tokens load. +- `POST /upload`: accepts one tar or tar.gz source bundle archive. +- `GET /runs/`: returns an in-memory upload status record, or `404` if the run id is unknown or expired. + +`POST /upload` authenticates with: + +```text +Authorization: Bearer +``` + +The token selects the configured `http_upload` pipeline. Producers do not send a +pipeline id. Requests with a submitted `pipeline` or `pipeline_id` query value +are rejected. + +Accepted upload content types: + +- `application/x-tar` +- `application/gzip` +- `application/x-gzip` + +Accepted uploads return: + +```json +{"run_id":"","status":"accepted"} +``` + +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 +uploads. + ## HTML Publication To publish generated sidecar HTML from Markdown files: diff --git a/docs/internal/app.md b/docs/internal/app.md index 534bfba..3e788ba 100644 --- a/docs/internal/app.md +++ b/docs/internal/app.md @@ -34,6 +34,11 @@ same destination fan-out path as normal runs. source. They share source backend construction with run workflows and never open destination backends. +`Serve` is the CLI-facing HTTP upload server entrypoint. It loads config, +loads the configured secrets directory, resolves upload bearer tokens for +configured `http_upload` sources, creates an `UploadCoordinator`, binds +`server.http.bind`, and serves the upload API until its context is cancelled. + ## Run Reports `RunReport` is the structured result model for run workflows. It includes @@ -116,6 +121,32 @@ Completed records retain the final run report or error text until their committed staged bundle directories. The coordinator is memory-only and does not persist queue state, status records, or run reports. +## HTTP Upload Server + +The HTTP upload server is app-layer transport wiring around +`UploadCoordinator`. It owns request authentication, route dispatch, HTTP status +mapping, and JSON response projection. Bundle staging and publication remain in +the coordinator and staged-source run path. + +Server startup resolves every configured `http_upload` source `token_env` +through the config-owned environment resolver after `secrets.directory` has +been loaded. Startup fails when a token is missing, empty, or duplicates another +upload pipeline token. Error messages identify environment variable names and +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. +- `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. + ## Coordination `PipelineRunCoordinator` wraps `RunPipeline` with in-memory admission control. @@ -171,6 +202,8 @@ Run helpers are grouped by responsibility: - `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_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. - `transforms.go`: app-level transform registry wiring. - `source_select.go`: configured-source selection shared by `validate` and `inspect`. diff --git a/docs/internal/config.md b/docs/internal/config.md index 80d7828..0e747ef 100644 --- a/docs/internal/config.md +++ b/docs/internal/config.md @@ -14,7 +14,8 @@ Input is a YAML file containing optional `server`, optional `secrets`, and requi Known-field checking rejects misspelled or unknown YAML keys before defaults and validation run. -`LoadFile` does not read secret files. `Run` loads the configured secrets directory after config validation and before backend construction. +`LoadFile` does not read secret files. App entrypoints load the configured +secrets directory after config validation and before credential-consuming work. ## Defaults @@ -58,13 +59,13 @@ Destination links are optional. When a `links` block is present, `base_url` is r ## Executable support boundary -Config validation accepts `local`, `ssh`, `s3`, and source-only `http_upload` backend shapes. Runtime execution opens `local`, `ssh`, and `s3` through `internal/app`. +Config validation accepts `local`, `ssh`, `s3`, and source-only `http_upload` backend shapes. Runtime `run`, `validate`, and `inspect` workflows open `local`, `ssh`, and `s3` through `internal/app`. Runtime `serve` workflows execute `http_upload` sources through the app upload coordinator and HTTP server. SSH config uses structured fields: `host`, optional `user`, optional `port`, `path`, optional `ssh_key_file`, optional `known_hosts`, and optional `host_key_policy`. `host_key_policy` accepts YAML booleans and strings and normalizes `true`/`strict`, `accept-new`, and `false`/`off`. S3 config requires `endpoint` and `bucket`, normalizes optional `prefix`, defaults `region` to `us-east-1`, and defaults omitted `force_path_style` to `true` while preserving explicit `false`. -HTTP upload config is source-only. Config owns its YAML shape, defaulting, size and duration parsing, and validation. The config package does not resolve `token_env`, authenticate requests, stage uploads, or execute HTTP upload sources. +HTTP upload config is source-only. Config owns its YAML shape, defaulting, size and duration parsing, and validation. The config package does not authenticate requests, stage uploads, or execute HTTP upload sources. The app layer resolves `token_env` through the config-owned environment resolver before starting the HTTP server. ## Secrets and credential resolution diff --git a/docs/troubleshooting.md b/docs/troubleshooting.md index 84e6d38..d94f8e2 100644 --- a/docs/troubleshooting.md +++ b/docs/troubleshooting.md @@ -26,7 +26,10 @@ Safe fix: compare the file to the reference in [configuration](config.md) and re ## `validate config ... backend ... is unsupported` -Likely cause: a source or destination uses a backend name other than `local`, `ssh`, or `s3`. +Likely cause: a source or destination uses an unsupported backend name, or a +command is trying to execute a backend that is valid only for another workflow. +`run`, `validate`, and `inspect` execute `local`, `ssh`, and `s3` sources. +`serve` executes `http_upload` sources. Diagnostic: @@ -34,7 +37,127 @@ Diagnostic: rg -n "backend:" ``` -Safe fix: use `backend: local`, `backend: ssh`, or `backend: s3` for executable workflows. +Safe fix: use `backend: local`, `backend: ssh`, or `backend: s3` for normal +source and destination workflows. Use `backend: http_upload` only for sources +handled by `distributor serve`. + +## `bind HTTP server ... address already in use` + +Likely cause: another process is already listening on `server.http.bind`. + +Diagnostic: + +```sh +ss -ltnp | rg '' +``` + +Safe fix: stop the conflicting process or configure a different +`server.http.bind` value. The default bind address is `127.0.0.1:8080`. + +## `upload token environment variable ... is not set` + +Likely cause: a configured `http_upload` source references `token_env`, but the +variable is absent from both the real process environment and +`secrets.directory`. + +Diagnostic: + +```sh +env | cut -d= -f1 | rg '^$' +ls -l / +``` + +Safe fix: set the real environment variable or create a readable +secrets-directory file with the same name. Do not place literal token values in +YAML. + +## `upload token environment variables ... resolve to the same value` + +Likely cause: two configured `http_upload` pipelines resolve to the same bearer +token value. + +Diagnostic: + +```sh +rg -n 'token_env:' +``` + +Safe fix: assign a distinct non-empty token value to each `http_upload` +pipeline. Distributor does not print the duplicate token value. + +## `POST /upload` returns `401` + +Likely cause: the request is missing `Authorization: Bearer ` or the +token does not match any configured `http_upload` pipeline. + +Diagnostic: + +```sh +curl -i -X POST http://127.0.0.1:8080/upload \ + -H "Authorization: Bearer $DISTRIBUTOR_UPLOAD_TOKEN" \ + -H "Content-Type: application/x-tar" \ + --data-binary @bundle.tar +``` + +Safe fix: use the token value resolved by the configured `token_env`. Do not +include token values in logs or tickets. + +## `POST /upload` returns `413` + +Likely cause: the request body exceeds the selected pipeline's +`source.max_upload_size` or the default `server.http.max_upload_size`. + +Diagnostic: + +```sh +ls -lh bundle.tar bundle.tar.gz +rg -n 'max_upload_size:' +``` + +Safe fix: upload a smaller archive, remove unnecessary files from the source +bundle, or raise the configured upload size limit. + +## `POST /upload` returns `415` + +Likely cause: the upload uses an unsupported content type. The server accepts +uncompressed tar and gzip-compressed tar archives only. + +Diagnostic: + +```sh +file bundle.tar.gz +``` + +Safe fix: send `Content-Type: application/x-tar`, `application/gzip`, or +`application/x-gzip`, matching the archive format. + +## `POST /upload` returns `503` + +Likely cause: the in-memory upload queue is full. + +Diagnostic: + +```sh +rg -n 'queue_size|max_concurrency' +``` + +Safe fix: retry after active uploads finish, or increase `server.http.queue_size` +for the deployment. + +## `GET /runs/` returns `404` + +Likely cause: the run id is wrong, the process restarted, or the completed +status record expired after `server.http.retention`. + +Diagnostic: + +```sh +curl -i http://127.0.0.1:8080/runs/ +rg -n 'retention:' +``` + +Safe fix: use the exact `run_id` returned by `POST /upload`. If status retention +is too short for operators, increase `server.http.retention`. ## `--format: format must be text or json` diff --git a/internal/app/serve.go b/internal/app/serve.go new file mode 100644 index 0000000..4b0f7bb --- /dev/null +++ b/internal/app/serve.go @@ -0,0 +1,62 @@ +package app + +import ( + "context" + "errors" + "fmt" + "net" + "net/http" + + "gitea.maximumdirect.net/eric/distributor/internal/config" +) + +type ServeOptions struct { + ConfigPath string +} + +func Serve(ctx context.Context, options ServeOptions) error { + if ctx == nil { + ctx = context.Background() + } + if err := ctx.Err(); err != nil { + return err + } + + configPath := options.ConfigPath + if configPath == "" { + configPath = config.DefaultConfigPath + } + cfg, err := config.LoadFile(configPath) + if err != nil { + return err + } + secretLoad, err := config.LoadSecretEnvironment(cfg.Secrets.Directory, nil) + if err != nil { + return err + } + + handler, err := newUploadHTTPHandler(ctx, cfg, secretLoad.Environment) + if err != nil { + return err + } + listener, err := net.Listen("tcp", cfg.Server.HTTP.Bind) + if err != nil { + return fmt.Errorf("bind HTTP server %q: %w", cfg.Server.HTTP.Bind, err) + } + defer listener.Close() + + server := &http.Server{Handler: handler} + shutdownDone := make(chan struct{}) + go func() { + defer close(shutdownDone) + <-ctx.Done() + _ = server.Shutdown(context.Background()) + }() + + err = server.Serve(listener) + if errors.Is(err, http.ErrServerClosed) { + <-shutdownDone + return nil + } + return err +} diff --git a/internal/app/upload_coordinator.go b/internal/app/upload_coordinator.go index 33a53a6..a47bad7 100644 --- a/internal/app/upload_coordinator.go +++ b/internal/app/upload_coordinator.go @@ -201,6 +201,13 @@ func (coordinator *UploadCoordinator) Expire() []UploadRunRecord { return coordinator.expireLocked(coordinator.now().UTC()) } +func (coordinator *UploadCoordinator) CanAccept() bool { + coordinator.mu.Lock() + defer coordinator.mu.Unlock() + coordinator.expireLocked(coordinator.now().UTC()) + return len(coordinator.pending) < coordinator.queueSize +} + func (coordinator *UploadCoordinator) QueueDepth() int { coordinator.mu.Lock() defer coordinator.mu.Unlock() diff --git a/internal/app/upload_http.go b/internal/app/upload_http.go new file mode 100644 index 0000000..5153344 --- /dev/null +++ b/internal/app/upload_http.go @@ -0,0 +1,209 @@ +package app + +import ( + "bytes" + "context" + "encoding/json" + "errors" + "fmt" + "io" + "mime" + "net/http" + "strings" + + "gitea.maximumdirect.net/eric/distributor/internal/config" + "gitea.maximumdirect.net/eric/distributor/internal/ingest" +) + +type uploadCoordinator interface { + CanAccept() bool + Submit(context.Context, UploadRequest) (UploadRunRecord, error) + Status(UploadRunID) (UploadRunRecord, bool) +} + +type uploadHTTPHandler struct { + coordinator uploadCoordinator + tokens map[string]string + limits map[string]int64 +} + +type uploadAcceptedResponse struct { + RunID UploadRunID `json:"run_id"` + Status UploadStatus `json:"status"` +} + +type httpErrorResponse struct { + Error string `json:"error"` +} + +func newUploadHTTPHandler(ctx context.Context, cfg config.Config, environment config.Environment) (http.Handler, error) { + config.ApplyDefaults(&cfg) + tokens, limits, 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) { + tokens := make(map[string]string) + limits := make(map[string]int64) + for _, pipeline := range cfg.Pipelines { + if pipeline.Source.Backend != config.BackendHTTPUpload { + continue + } + 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) + } + if token == "" { + return nil, 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) + } + tokens[token] = pipeline.ID + limits[pipeline.ID] = int64(*pipeline.Source.Upload.MaxUploadSize) + } + return tokens, limits, nil +} + +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": + handler.handleUpload(w, r) + case r.Method == http.MethodGet && strings.HasPrefix(r.URL.Path, "/runs/"): + handler.handleRunStatus(w, r) + default: + writeHTTPError(w, http.StatusNotFound, "not found") + } +} + +func (handler uploadHTTPHandler) handleHealth(w http.ResponseWriter) { + writeJSON(w, http.StatusOK, map[string]string{"status": "ok"}) +} + +func (handler uploadHTTPHandler) handleUpload(w http.ResponseWriter, r *http.Request) { + if r.URL.Query().Has("pipeline") || r.URL.Query().Has("pipeline_id") { + writeHTTPError(w, http.StatusBadRequest, "pipeline id is not accepted") + return + } + pipelineID, ok := handler.authenticate(r.Header.Get("Authorization")) + if !ok { + writeHTTPError(w, http.StatusUnauthorized, "unauthorized") + return + } + contentType := r.Header.Get("Content-Type") + if !supportedUploadContentType(contentType) { + writeHTTPError(w, http.StatusUnsupportedMediaType, "unsupported content type") + return + } + if !handler.coordinator.CanAccept() { + 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), + }) + if err != nil { + writeUploadSubmitError(w, err) + return + } + writeJSON(w, http.StatusAccepted, uploadAcceptedResponse{ + RunID: record.ID, + Status: UploadStatusAccepted, + }) +} + +func (handler uploadHTTPHandler) handleRunStatus(w http.ResponseWriter, r *http.Request) { + rawRunID := strings.TrimPrefix(r.URL.Path, "/runs/") + if rawRunID == "" || strings.Contains(rawRunID, "/") { + writeHTTPError(w, http.StatusNotFound, "not found") + return + } + record, ok := handler.coordinator.Status(UploadRunID(rawRunID)) + if !ok { + writeHTTPError(w, http.StatusNotFound, "run not found") + return + } + writeJSON(w, http.StatusOK, record) +} + +func (handler uploadHTTPHandler) authenticate(header string) (string, bool) { + const prefix = "Bearer " + if !strings.HasPrefix(header, prefix) { + return "", false + } + token := strings.TrimSpace(strings.TrimPrefix(header, prefix)) + if token == "" { + return "", false + } + pipelineID, ok := handler.tokens[token] + 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): + writeHTTPError(w, http.StatusServiceUnavailable, "upload queue is full") + case errors.Is(err, ingest.ErrUploadTooLarge): + writeHTTPError(w, http.StatusRequestEntityTooLarge, "upload exceeds maximum size") + case errors.Is(err, ingest.ErrUnsupportedContentType): + writeHTTPError(w, http.StatusUnsupportedMediaType, "unsupported content type") + default: + writeHTTPError(w, http.StatusBadRequest, "upload rejected") + } +} + +func writeHTTPError(w http.ResponseWriter, status int, message string) { + writeJSON(w, status, httpErrorResponse{Error: message}) +} + +func writeJSON(w http.ResponseWriter, status int, value any) { + w.Header().Set("Content-Type", "application/json") + w.WriteHeader(status) + _ = json.NewEncoder(w).Encode(value) +} diff --git a/internal/app/upload_http_test.go b/internal/app/upload_http_test.go new file mode 100644 index 0000000..323df4c --- /dev/null +++ b/internal/app/upload_http_test.go @@ -0,0 +1,331 @@ +package app + +import ( + "context" + "encoding/json" + "errors" + "io" + "net/http" + "net/http/httptest" + "strings" + "testing" + "time" + + "gitea.maximumdirect.net/eric/distributor/internal/config" +) + +type fakeUploadCoordinator struct { + canAccept bool + submit func(context.Context, UploadRequest) (UploadRunRecord, error) + status func(UploadRunID) (UploadRunRecord, bool) +} + +func (fake fakeUploadCoordinator) CanAccept() bool { + return fake.canAccept +} + +func (fake fakeUploadCoordinator) Submit(ctx context.Context, request UploadRequest) (UploadRunRecord, error) { + if fake.submit == nil { + return UploadRunRecord{}, errors.New("unexpected submit") + } + return fake.submit(ctx, request) +} + +func (fake fakeUploadCoordinator) Status(runID UploadRunID) (UploadRunRecord, bool) { + if fake.status == nil { + return UploadRunRecord{}, false + } + return fake.status(runID) +} + +func TestResolveUploadTokensFailsForMissingAndDuplicateTokens(t *testing.T) { + cfg := uploadHTTPTestConfig() + + _, _, err := resolveUploadTokens(cfg, config.NewEnvironment(nil, func(string) (string, bool) { + return "", false + })) + if err == nil || !strings.Contains(err.Error(), "UPLOAD_TOKEN") { + t.Fatalf("resolveUploadTokens() error = %v, want missing UPLOAD_TOKEN", err) + } + + cfg.Pipelines = append(cfg.Pipelines, config.Pipeline{ + ID: "weekly", + Source: config.Backend{ + Backend: config.BackendHTTPUpload, + Upload: config.HTTPUpload{TokenEnv: "OTHER_UPLOAD_TOKEN"}, + }, + Destinations: cfg.Pipelines[0].Destinations, + }) + config.ApplyDefaults(&cfg) + secret := "super-secret-token" + _, _, err = resolveUploadTokens(cfg, uploadHTTPTestEnvironment(map[string]string{ + "UPLOAD_TOKEN": secret, + "OTHER_UPLOAD_TOKEN": secret, + })) + if err == nil { + t.Fatal("resolveUploadTokens() error = nil, want duplicate token error") + } + if strings.Contains(err.Error(), secret) { + t.Fatalf("duplicate token error exposed secret value: %q", err) + } +} + +func TestNewUploadHTTPHandlerAcceptsDefaultedConfig(t *testing.T) { + cfg := uploadHTTPTestConfig() + cfg.Server.HTTP.Bind = "" + cfg.Server.HTTP.StagingRoot = "" + cfg.Server.HTTP.MaxUploadSize = nil + cfg.Server.HTTP.QueueSize = 0 + cfg.Server.HTTP.MaxConcurrency = 0 + cfg.Server.HTTP.Retention = nil + cfg.Pipelines[0].Source.Upload.StagingPath = "" + cfg.Pipelines[0].Source.Upload.MaxUploadSize = nil + + handler, err := newUploadHTTPHandler(context.Background(), cfg, uploadHTTPTestEnvironment(map[string]string{ + "UPLOAD_TOKEN": "secret", + })) + if err != nil { + t.Fatalf("newUploadHTTPHandler() error = %v", err) + } + if handler == nil { + t.Fatal("newUploadHTTPHandler() = nil") + } +} + +func TestUploadHTTPHandlerAuthenticatesAndAcceptsUpload(t *testing.T) { + var submitted UploadRequest + handler := uploadHTTPHandler{ + coordinator: fakeUploadCoordinator{ + canAccept: true, + submit: func(_ context.Context, request UploadRequest) (UploadRunRecord, error) { + submitted = request + body, err := io.ReadAll(request.Body) + if err != nil { + t.Fatalf("read submitted body: %v", err) + } + if string(body) != "archive" { + t.Fatalf("submitted body = %q, want archive", body) + } + return UploadRunRecord{ID: "reports.20260603T120000Z.abcdef12", Status: UploadStatusAccepted}, nil + }, + }, + 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")) + request.Header.Set("Authorization", "Bearer valid-token") + request.Header.Set("Content-Type", "application/x-tar") + + handler.ServeHTTP(recorder, request) + + if recorder.Code != http.StatusAccepted { + t.Fatalf("status = %d, want %d; body = %q", recorder.Code, http.StatusAccepted, recorder.Body.String()) + } + if submitted.PipelineID != "reports" { + t.Fatalf("submitted pipeline = %q, want reports", submitted.PipelineID) + } + var response uploadAcceptedResponse + if err := json.Unmarshal(recorder.Body.Bytes(), &response); err != nil { + t.Fatalf("decode response: %v", err) + } + if response.RunID != "reports.20260603T120000Z.abcdef12" || response.Status != UploadStatusAccepted { + t.Fatalf("response = %#v, want accepted run id", response) + } + if strings.Contains(recorder.Body.String(), "valid-token") { + t.Fatalf("response exposed token: %q", recorder.Body.String()) + } +} + +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"} { + recorder := httptest.NewRecorder() + request := httptest.NewRequest(http.MethodPost, "/upload", strings.NewReader("archive")) + request.Header.Set("Authorization", authHeader) + request.Header.Set("Content-Type", "application/x-tar") + + handler.ServeHTTP(recorder, request) + + if recorder.Code != http.StatusUnauthorized { + t.Fatalf("auth %q status = %d, want %d", authHeader, recorder.Code, http.StatusUnauthorized) + } + if strings.Contains(recorder.Body.String(), "valid-token") || strings.Contains(recorder.Body.String(), "wrong-token") { + t.Fatalf("unauthorized response exposed token: %q", recorder.Body.String()) + } + } +} + +func TestUploadHTTPHandlerRejectsUnsupportedOversizedFullQueueAndPipelineID(t *testing.T) { + tests := []struct { + name string + canAccept bool + url string + contentType string + body io.Reader + wantStatus int + }{ + { + name: "unsupported content type", + canAccept: true, + url: "/upload", + contentType: "application/zip", + 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, + url: "/upload", + contentType: "application/x-tar", + body: &countingReader{reader: strings.NewReader("archive")}, + wantStatus: http.StatusServiceUnavailable, + }, + { + name: "submitted pipeline id", + canAccept: true, + 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) { + handler := uploadHTTPHandler{ + coordinator: fakeUploadCoordinator{ + canAccept: tt.canAccept, + submit: func(context.Context, UploadRequest) (UploadRunRecord, error) { + t.Fatal("Submit should not be called") + return UploadRunRecord{}, nil + }, + }, + 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) + request.Header.Set("Authorization", "Bearer valid-token") + request.Header.Set("Content-Type", tt.contentType) + + handler.ServeHTTP(recorder, request) + + if recorder.Code != tt.wantStatus { + t.Fatalf("status = %d, want %d; body = %q", recorder.Code, tt.wantStatus, recorder.Body.String()) + } + if reader, ok := tt.body.(*countingReader); ok && reader.reads != 0 { + t.Fatalf("full queue read body %d time(s), want zero", reader.reads) + } + }) + } +} + +func TestUploadHTTPHandlerRunStatusAndHealth(t *testing.T) { + finishedAt := time.Date(2026, 6, 3, 12, 0, 0, 0, time.UTC) + handler := uploadHTTPHandler{ + coordinator: fakeUploadCoordinator{ + canAccept: true, + status: func(runID UploadRunID) (UploadRunRecord, bool) { + if runID != "reports.20260603T120000Z.abcdef12" { + return UploadRunRecord{}, false + } + return UploadRunRecord{ + ID: runID, + PipelineID: "reports", + Status: UploadStatusSucceeded, + FinishedAt: &finishedAt, + }, true + }, + }, + tokens: map[string]string{"valid-token": "reports"}, + limits: map[string]int64{"reports": 1024}, + } + + recorder := httptest.NewRecorder() + handler.ServeHTTP(recorder, httptest.NewRequest(http.MethodGet, "/healthz", nil)) + if recorder.Code != http.StatusOK { + t.Fatalf("health status = %d, want %d", recorder.Code, http.StatusOK) + } + + recorder = httptest.NewRecorder() + handler.ServeHTTP(recorder, httptest.NewRequest(http.MethodGet, "/runs/reports.20260603T120000Z.abcdef12", nil)) + if recorder.Code != http.StatusOK { + t.Fatalf("run status = %d, want %d; body = %q", recorder.Code, http.StatusOK, recorder.Body.String()) + } + var record UploadRunRecord + if err := json.Unmarshal(recorder.Body.Bytes(), &record); err != nil { + t.Fatalf("decode run status: %v", err) + } + if record.ID != "reports.20260603T120000Z.abcdef12" || record.Status != UploadStatusSucceeded { + t.Fatalf("record = %#v, want succeeded run status", record) + } + + recorder = httptest.NewRecorder() + handler.ServeHTTP(recorder, httptest.NewRequest(http.MethodGet, "/runs/unknown", nil)) + if recorder.Code != http.StatusNotFound { + t.Fatalf("unknown run status = %d, want %d", recorder.Code, http.StatusNotFound) + } +} + +type countingReader struct { + reader io.Reader + reads int +} + +func (reader *countingReader) Read(data []byte) (int, error) { + reader.reads++ + return reader.reader.Read(data) +} + +func uploadHTTPTestConfig() config.Config { + size := config.ByteSize(1024) + retention := config.Duration(24 * time.Hour) + cfg := config.Config{ + Server: config.Server{HTTP: config.HTTPServer{ + Bind: config.DefaultHTTPBind, + StagingRoot: "/tmp/distributor-test", + MaxUploadSize: &size, + QueueSize: 2, + MaxConcurrency: 1, + Retention: &retention, + }}, + Pipelines: []config.Pipeline{{ + ID: "reports", + Source: config.Backend{ + Backend: config.BackendHTTPUpload, + Upload: config.HTTPUpload{ + TokenEnv: "UPLOAD_TOKEN", + StagingPath: "/tmp/distributor-test/reports", + MaxUploadSize: &size, + }, + }, + Destinations: []config.Destination{{ + ID: "local", + Backend: config.BackendLocal, + Path: "/tmp/distributor-output", + Publish: &config.PublishPolicy{Source: true}, + }}, + }}, + } + config.ApplyDefaults(&cfg) + return cfg +} + +func uploadHTTPTestEnvironment(values map[string]string) config.Environment { + return config.NewEnvironment(values, func(string) (string, bool) { + return "", false + }) +} diff --git a/internal/cli/root.go b/internal/cli/root.go index 569112e..efa938f 100644 --- a/internal/cli/root.go +++ b/internal/cli/root.go @@ -29,6 +29,8 @@ func Execute(ctx context.Context, args []string, stdout, stderr io.Writer) int { return versionCommand(ctx, args[1:], stdout, stderr) case "run": return runCommand(ctx, args[1:], stdout, stderr) + case "serve": + return serveCommand(ctx, args[1:], stdout, stderr) case "validate": return validateCommand(ctx, args[1:], stdout, stderr) case "inspect": @@ -51,6 +53,7 @@ Usage: Commands: version Print version information run Run configured distribution pipelines + serve Run the HTTP upload server validate Validate a source bundle or bundle tree inspect Inspect bundles or distributor state manifest Create source bundle manifests diff --git a/internal/cli/root_test.go b/internal/cli/root_test.go index 98c2157..29c7e0e 100644 --- a/internal/cli/root_test.go +++ b/internal/cli/root_test.go @@ -11,6 +11,7 @@ import ( "strings" "testing" + "gitea.maximumdirect.net/eric/distributor/internal/app" "gitea.maximumdirect.net/eric/distributor/internal/storage" "gitea.maximumdirect.net/eric/distributor/internal/testutil" producerbundle "gitea.maximumdirect.net/eric/distributor/pkg/bundle" @@ -92,6 +93,34 @@ func TestExecuteVersionJSON(t *testing.T) { } } +func TestExecuteServeParsesConfig(t *testing.T) { + originalServeApp := serveApp + defer func() { + serveApp = originalServeApp + }() + var gotOptions app.ServeOptions + serveApp = func(_ context.Context, options app.ServeOptions) error { + gotOptions = options + return nil + } + var stdout, stderr bytes.Buffer + + code := Execute(context.Background(), []string{"serve", "--config", "config.yml"}, &stdout, &stderr) + + if code != exitOK { + t.Fatalf("exit code = %d, want %d; stderr = %q", code, exitOK, stderr.String()) + } + if gotOptions.ConfigPath != "config.yml" { + t.Fatalf("ConfigPath = %q, want config.yml", gotOptions.ConfigPath) + } + if stdout.Len() != 0 { + t.Fatalf("stdout = %q, want empty", stdout.String()) + } + if stderr.Len() != 0 { + t.Fatalf("stderr = %q, want empty", stderr.String()) + } +} + func TestExecuteRejectsInvalidFormat(t *testing.T) { var stdout, stderr bytes.Buffer diff --git a/internal/cli/serve.go b/internal/cli/serve.go new file mode 100644 index 0000000..b04713d --- /dev/null +++ b/internal/cli/serve.go @@ -0,0 +1,46 @@ +package cli + +import ( + "context" + "flag" + "fmt" + "io" + + "gitea.maximumdirect.net/eric/distributor/internal/app" +) + +var serveApp = app.Serve + +func serveCommand(ctx context.Context, args []string, stdout, stderr io.Writer) int { + if hasHelp(args) { + printServeHelp(stdout) + return exitOK + } + + flags := flag.NewFlagSet("serve", flag.ContinueOnError) + flags.SetOutput(stderr) + configPath := flags.String("config", "", "path to config file") + if err := flags.Parse(args); err != nil { + return exitUsage + } + if rejectPositionalArgs(stderr, "serve", flags.Args()) { + return exitUsage + } + + if err := serveApp(ctx, app.ServeOptions{ConfigPath: *configPath}); err != nil { + return fail(stderr, err) + } + return exitOK +} + +func printServeHelp(w io.Writer) { + fmt.Fprint(w, `Usage: + distributor serve --config + +Options: + --config Path to config file + +Serve loads configured HTTP upload sources, resolves upload tokens through the +configured secret environment, and starts the HTTP upload API. +`) +}