From 4b748c2e5387c380e525ced2a6ce3d87bab0f69d Mon Sep 17 00:00:00 2001 From: Eric Rakestraw Date: Thu, 13 Aug 2026 02:55:25 +0000 Subject: [PATCH] Bound Distributor response diagnostics --- docs/integrations/distributor/api.md | 5 + docs/integrations/distributor/pkg-upload.md | 8 +- docs/internal/distributor-adapter.md | 22 +- docs/operations.md | 3 + internal/adapters/distributor/client.go | 86 ++++- .../adapters/distributor/client_http_test.go | 310 ++++++++++++++++++ internal/adapters/distributor/client_test.go | 12 +- internal/app/app.go | 39 +-- internal/app/batch_notification.go | 13 +- internal/app/distributor_notification_test.go | 45 +++ 10 files changed, 497 insertions(+), 46 deletions(-) create mode 100644 internal/adapters/distributor/client_http_test.go create mode 100644 internal/app/distributor_notification_test.go diff --git a/docs/integrations/distributor/api.md b/docs/integrations/distributor/api.md index 143b0a3..88babb6 100644 --- a/docs/integrations/distributor/api.md +++ b/docs/integrations/distributor/api.md @@ -30,6 +30,11 @@ at least one source-file mapping, before calling Distributor. It reads the beare the configured environment variable and redacts that value from errors. Request construction and timeout handling belong to the [Distributor adapter](../../internal/distributor-adapter.md). +Weatherreporter reads at most 1 MiB from each Distributor response. An +oversized response fails notification with a stable local diagnostic. Normal +Weatherreporter results retain upload and status identity but do not repeat +Distributor response bodies, status reports, or remote error text. + ## Idempotency Distributor scopes idempotency to the token, pipeline ID, and key. Keys must be diff --git a/docs/integrations/distributor/pkg-upload.md b/docs/integrations/distributor/pkg-upload.md index 7ce8670..379cc02 100644 --- a/docs/integrations/distributor/pkg-upload.md +++ b/docs/integrations/distributor/pkg-upload.md @@ -11,6 +11,9 @@ bearer token, and an HTTP client whose timeout is the configured Distributor timeout. The endpoint may include a path prefix but never userinfo, a query, or a fragment. It passes no custom retry options, so the pinned client's defaults apply: three attempts, 100 ms base delay, and one-second maximum delay. +The adapter bounds every response to 1 MiB before handing it to the pinned +client. A response above that boundary is rejected as a local overflow rather +than decoding or retaining a prefix. For each notification, Weatherreporter calls `UploadFiles` with: @@ -39,8 +42,9 @@ adapter translates it to its own conflict error without exposing the token. The adapter then calls `Status` for the accepted run. A terminal `failed` status is a notification failure. A status lookup failure or a timeout before a terminal status remains attached to the otherwise accepted upload as diagnostic -status information. Polling cadence, final failure handling, and redaction are -internal behavior documented in the +status information. Normal diagnostics use local status classifications; they +do not expose remote response text or the status report. Polling cadence, final +failure handling, and redaction are internal behavior documented in the [Distributor adapter](../../internal/distributor-adapter.md) and [application orchestration](../../internal/app-orchestration.md). diff --git a/docs/internal/distributor-adapter.md b/docs/internal/distributor-adapter.md index b43ea15..cfe8dbb 100644 --- a/docs/internal/distributor-adapter.md +++ b/docs/internal/distributor-adapter.md @@ -13,7 +13,9 @@ the token, an optional timeout, and an injectable upstream-client factory. `New` validates its configuration before creating the adapter. For each upload, the adapter reads the token from the configured environment variable and builds the upstream client with that endpoint, token, and an HTTP client whose timeout -matches the local positive timeout. +matches the local positive timeout. Its transport reads at most 1 MiB from any +Distributor response before the pinned client decodes it; an oversized response +is a distinct local failure and does not trigger an extra upload attempt. The upstream client is an implementation dependency, not a source of application configuration: retry ownership, pipeline selection, path @@ -43,20 +45,26 @@ persist notification artifacts. An accepted upload is followed by one status request. When a timeout is configured, a nonterminal result is polled until `succeeded` or `failed`, or until the context ends. The translated `UploadResult` contains the run ID, -status, and `RunStatus`, including pipeline ID, lifecycle timestamps, report, -and remote error details. +status, and `RunStatus`, including pipeline ID and lifecycle timestamps. +Remote response bodies, status reports, and remote error text are not retained +in normal results. HTTP failures retain a local typed status-code and +retryability classification; conflicts retain the local idempotency-conflict +type. Status lookup or polling errors are preserved in `UploadResult.StatusError` so -the caller can report an accepted-but-unconfirmed delivery. A terminal failed -run returns that result and an error. Upload failures return no result. Upstream +the caller can report an accepted-but-unconfirmed delivery, using a bounded +repository-owned diagnostic rather than remote text. A terminal failed run +returns that result and an error. Upload failures return no result. Upstream idempotency conflicts become the local `IdempotencyConflictError`, which adds endpoint, pipeline, bundle, idempotency, and file-path context while redacting the token. ## Verification -Focused tests cover configuration validation, request mapping, timeouts and -polling, status translation, conflict handling, and token redaction: +Focused tests cover configuration validation, request mapping, response size +boundaries, safe diagnostics, timeouts and polling, status translation, conflict +handling, and token redaction. A local HTTP server exercises the production +upload and status boundary: ```sh go test ./internal/adapters/distributor diff --git a/docs/operations.md b/docs/operations.md index 6c1cb6e..cd0f9e7 100644 --- a/docs/operations.md +++ b/docs/operations.md @@ -78,6 +78,9 @@ files remain at their selected destinations. A batch notification failure also leaves all successfully published report files in place. Distributor source files are those operator-owned Markdown outputs; rendered bundle paths and delivery status appear in the result, not in a local notification receipt. +Remote Distributor response text is not included in command output. Instead, +notification failures use stable local diagnostics while retaining the upload +and status identities needed to investigate delivery with Distributor. Report counters count report items only. A batch notification failure therefore returns a failed batch status even when all report counters show success; the top-level notification result contains the delivery diagnostic. diff --git a/internal/adapters/distributor/client.go b/internal/adapters/distributor/client.go index af362f1..751e631 100644 --- a/internal/adapters/distributor/client.go +++ b/internal/adapters/distributor/client.go @@ -2,10 +2,12 @@ package distributor import ( + "bytes" "context" "encoding/json" "errors" "fmt" + "io" "net/http" "os" "strings" @@ -107,6 +109,10 @@ type runStatus struct { const statusPollInterval = 250 * time.Millisecond +const maxDistributorResponseBytes int64 = 1 << 20 + +var errDistributorResponseTooLarge = fmt.Errorf("distributor response exceeds the %d-byte limit", maxDistributorResponseBytes) + func New(cfg config.DistributorNotifyConfig) *Client { return newClient(cfg, newDistributorUploadClient) } @@ -202,6 +208,7 @@ func (c *Client) Upload(ctx context.Context, req UploadRequest) (UploadResult, e UploadStatus: result.Status, } status, statusErr := waitForRunStatus(runCtx, uploadClient, result.RunID, c.Timeout > 0) + status = sanitizeRunStatus(status) if status.RunID != "" || status.Status != "" { uploadResult.RunStatus = &RunStatus{ RunID: status.RunID, @@ -218,7 +225,7 @@ func (c *Client) Upload(ctx context.Context, req UploadRequest) (UploadResult, e } } if statusErr != nil { - uploadResult.StatusError = redactTokenString(statusErr.Error(), token) + uploadResult.StatusError = safeDistributorDiagnostic(statusErr, token).Error() return uploadResult, nil } if status.Status == "failed" { @@ -261,10 +268,52 @@ type distributorUploadClient struct { client *distributorupload.Client } +type boundedResponseTransport struct { + base http.RoundTripper + limit int64 +} + +func (t boundedResponseTransport) RoundTrip(req *http.Request) (*http.Response, error) { + base := t.base + if base == nil { + base = http.DefaultTransport + } + response, err := base.RoundTrip(req) + if err != nil { + return nil, err + } + defer response.Body.Close() + + data, err := io.ReadAll(io.LimitReader(response.Body, t.limit+1)) + if err != nil { + return nil, err + } + if int64(len(data)) > t.limit { + return nil, errDistributorResponseTooLarge + } + response.Body = io.NopCloser(bytes.NewReader(data)) + response.ContentLength = int64(len(data)) + return response, nil +} + +type RemoteResponseError struct { + StatusCode int + Retryable bool +} + +func (e *RemoteResponseError) Error() string { + if e == nil || e.StatusCode == 0 { + return "distributor request failed" + } + return fmt.Sprintf("distributor request failed with HTTP status %d", e.StatusCode) +} + func newDistributorUploadClient(endpoint, token string, timeout time.Duration) (uploadClient, error) { - httpClient := (*http.Client)(nil) + httpClient := &http.Client{ + Transport: boundedResponseTransport{base: http.DefaultTransport, limit: maxDistributorResponseBytes}, + } if timeout > 0 { - httpClient = &http.Client{Timeout: timeout} + httpClient.Timeout = timeout } client, err := distributorupload.NewClient(distributorupload.ClientOptions{ Endpoint: endpoint, @@ -306,7 +355,7 @@ func (c distributorUploadClient) Status(ctx context.Context, runID string) (runS if err != nil { return runStatus{}, err } - return runStatus{ + return sanitizeRunStatus(runStatus{ RunID: status.RunID, PipelineID: status.PipelineID, Status: status.Status, @@ -315,7 +364,7 @@ func (c distributorUploadClient) Status(ctx context.Context, runID string) (runS FinishedAt: status.FinishedAt, Report: append(json.RawMessage(nil), status.Report...), Error: status.Error, - }, nil + }), nil } type uploadErrorContext struct { @@ -331,7 +380,7 @@ type uploadErrorContext struct { func wrapUploadError(err error, ctx uploadErrorContext) error { var conflict *distributorupload.IdempotencyConflictError isConflict := errors.As(err, &conflict) - err = redactToken(err, ctx.Token) + err = safeDistributorDiagnostic(err, ctx.Token) if isConflict { return &IdempotencyConflictError{ Err: fmt.Errorf("upload distributor bundle %q to pipeline %q at endpoint %q with idempotency key %q from sources %q as bundle paths %q: idempotency conflict: %w", ctx.BundleID, ctx.PipelineID, ctx.Endpoint, ctx.IdempotencyKey, ctx.SourcePaths, ctx.BundlePaths, err), @@ -340,6 +389,31 @@ func wrapUploadError(err error, ctx uploadErrorContext) error { return fmt.Errorf("upload distributor bundle %q to pipeline %q at endpoint %q with idempotency key %q from sources %q as bundle paths %q: %w", ctx.BundleID, ctx.PipelineID, ctx.Endpoint, ctx.IdempotencyKey, ctx.SourcePaths, ctx.BundlePaths, err) } +func safeDistributorDiagnostic(err error, token string) error { + if err == nil { + return nil + } + if errors.Is(err, errDistributorResponseTooLarge) { + return errDistributorResponseTooLarge + } + if errors.Is(err, context.Canceled) || errors.Is(err, context.DeadlineExceeded) { + return redactToken(err, token) + } + var httpErr *distributorupload.HTTPError + if errors.As(err, &httpErr) { + return &RemoteResponseError{StatusCode: httpErr.StatusCode, Retryable: httpErr.Retryable} + } + return errors.New("distributor request failed") +} + +func sanitizeRunStatus(status runStatus) runStatus { + status.Report = nil + if status.Error != "" { + status.Error = "distributor reported a failed run" + } + return status +} + func uploadSourcePaths(files []UploadFile) []string { paths := make([]string, 0, len(files)) for _, file := range files { diff --git a/internal/adapters/distributor/client_http_test.go b/internal/adapters/distributor/client_http_test.go new file mode 100644 index 0000000..ab2c1ca --- /dev/null +++ b/internal/adapters/distributor/client_http_test.go @@ -0,0 +1,310 @@ +package distributor + +import ( + "archive/tar" + "compress/gzip" + "context" + "errors" + "fmt" + "io" + "net/http" + "net/http/httptest" + "os" + "path/filepath" + "strings" + "testing" + "time" + + "gitea.maximumdirect.net/eric/weatherreporter/internal/config" +) + +const oversizedRemoteDiagnostic = "REMOTE-DIAGNOSTIC" + +func TestUploadUsesProductionHTTPBoundary(t *testing.T) { + const token = "test-upload-token" + var uploadCalls, statusCalls int + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + switch { + case r.Method == http.MethodPost && r.URL.Path == "/prefix/v1/pipelines/weather/upload": + uploadCalls++ + if got := r.Header.Get("Authorization"); got != "Bearer "+token { + t.Fatalf("authorization = %q", got) + } + if got := r.Header.Get("Idempotency-Key"); got != "bundle-key" { + t.Fatalf("idempotency key = %q", got) + } + if got := r.Header.Get("Content-Type"); got != "application/gzip" { + t.Fatalf("content type = %q", got) + } + verifyUploadedArchive(t, r.Body, "daily/report.md", "report body") + w.Header().Set("Content-Type", "application/json") + w.WriteHeader(http.StatusAccepted) + _, _ = io.WriteString(w, `{"run_id":"run-123","status":"accepted"}`) + case r.Method == http.MethodGet && r.URL.Path == "/prefix/runs/run-123": + statusCalls++ + if got := r.Header.Get("Authorization"); got != "Bearer "+token { + t.Fatalf("authorization = %q", got) + } + w.Header().Set("Content-Type", "application/json") + _, _ = io.WriteString(w, `{"run_id":"run-123","pipeline_id":"weather","status":"succeeded","report":{"detail":"REMOTE-DETAIL"}}`) + default: + t.Fatalf("unexpected request %s %s", r.Method, r.URL.Path) + } + })) + defer server.Close() + + client := productionClient(t, server.URL+"/prefix", token) + result, err := client.Upload(context.Background(), productionUploadRequest(t)) + if err != nil || uploadCalls != 1 || statusCalls != 1 || result.RunID != "run-123" || result.Status != "succeeded" || result.UploadStatus != "accepted" || result.RunStatus == nil || result.RunStatus.PipelineID != "weather" || len(result.RunStatus.Report) != 0 { + t.Fatalf("result/error/calls = %#v/%v/%d/%d", result, err, uploadCalls, statusCalls) + } +} + +func TestUploadClassifiesRemoteHTTPDiagnostics(t *testing.T) { + const token = "test-upload-token" + const remote = oversizedRemoteDiagnostic + for _, tt := range []struct { + name string + handle func(http.ResponseWriter, *http.Request) + check func(t *testing.T, result UploadResult, err error) + }{ + { + name: "upload failure", + handle: func(w http.ResponseWriter, r *http.Request) { + if r.Method != http.MethodPost { + t.Fatalf("method = %s", r.Method) + } + w.WriteHeader(http.StatusBadRequest) + _, _ = io.WriteString(w, `{"error":"REMOTE-DIAGNOSTIC","retryable":true}`) + }, + check: func(t *testing.T, _ UploadResult, err error) { + t.Helper() + var remoteErr *RemoteResponseError + if err == nil || !errors.As(err, &remoteErr) || remoteErr.StatusCode != http.StatusBadRequest || !remoteErr.Retryable { + t.Fatalf("error = %T %v", err, err) + } + }, + }, + { + name: "status failure", + handle: func(w http.ResponseWriter, r *http.Request) { + if r.Method == http.MethodPost { + w.WriteHeader(http.StatusAccepted) + _, _ = io.WriteString(w, `{"run_id":"run-123","status":"accepted"}`) + return + } + w.WriteHeader(http.StatusInternalServerError) + _, _ = io.WriteString(w, remote) + }, + check: func(t *testing.T, result UploadResult, err error) { + t.Helper() + if err != nil || result.Status != "accepted" || result.StatusError != "distributor request failed with HTTP status 500" { + t.Fatalf("result/error = %#v/%v", result, err) + } + }, + }, + { + name: "failed run", + handle: func(w http.ResponseWriter, r *http.Request) { + if r.Method == http.MethodPost { + w.WriteHeader(http.StatusAccepted) + _, _ = io.WriteString(w, `{"run_id":"run-123","status":"accepted"}`) + return + } + _, _ = io.WriteString(w, `{"run_id":"run-123","status":"failed","error":"REMOTE-DIAGNOSTIC","report":{"detail":"REMOTE-DIAGNOSTIC"}}`) + }, + check: func(t *testing.T, result UploadResult, err error) { + t.Helper() + if err == nil || result.Status != "failed" || result.RunStatus == nil || result.RunStatus.Error != "distributor reported a failed run" || len(result.RunStatus.Report) != 0 { + t.Fatalf("result/error = %#v/%v", result, err) + } + }, + }, + } { + t.Run(tt.name, func(t *testing.T) { + server := httptest.NewServer(http.HandlerFunc(tt.handle)) + defer server.Close() + result, err := productionClient(t, server.URL, token).Upload(context.Background(), productionUploadRequest(t)) + tt.check(t, result, err) + for _, value := range []string{fmt.Sprint(result), fmt.Sprint(err)} { + if strings.Contains(value, remote) || strings.Contains(value, token) { + t.Fatalf("normal diagnostic leaked remote value: %q", value) + } + } + }) + } +} + +func TestUploadBoundsHTTPResponses(t *testing.T) { + for _, tt := range []struct { + name string + response func(size int) string + statusCode int + statusBody func(size int) string + check func(t *testing.T, result UploadResult, err error, overflow bool) + }{ + { + name: "accepted response", + response: func(size int) string { + return paddedJSON(t, `{"run_id":"run-123","status":"accepted","detail":"REMOTE-DIAGNOSTIC"}`, size) + }, + statusBody: func(_ int) string { + return `{"run_id":"run-123","status":"succeeded"}` + }, + check: func(t *testing.T, result UploadResult, err error, overflow bool) { + t.Helper() + if overflow { + if !errors.Is(err, errDistributorResponseTooLarge) || result.RunID != "" { + t.Fatalf("overflow result/error = %#v/%v", result, err) + } + return + } + if err != nil || result.Status != "succeeded" { + t.Fatalf("bounded result/error = %#v/%v", result, err) + } + }, + }, + { + name: "status report", + response: func(_ int) string { + return `{"run_id":"run-123","status":"accepted"}` + }, + statusBody: func(size int) string { return statusReportBody(t, size) }, + check: func(t *testing.T, result UploadResult, err error, overflow bool) { + t.Helper() + if overflow { + if err != nil || result.Status != "accepted" || result.StatusError != errDistributorResponseTooLarge.Error() { + t.Fatalf("overflow result/error = %#v/%v", result, err) + } + return + } + if err != nil || result.Status != "succeeded" || result.RunStatus == nil || len(result.RunStatus.Report) != 0 { + t.Fatalf("bounded result/error = %#v/%v", result, err) + } + }, + }, + { + name: "error response", + response: func(size int) string { return repeatedToLength(oversizedRemoteDiagnostic, size) }, + statusCode: http.StatusBadRequest, + check: func(t *testing.T, result UploadResult, err error, overflow bool) { + t.Helper() + if overflow { + if !errors.Is(err, errDistributorResponseTooLarge) || result.RunID != "" { + t.Fatalf("overflow result/error = %#v/%v", result, err) + } + return + } + var remoteErr *RemoteResponseError + if !errors.As(err, &remoteErr) || remoteErr.StatusCode != http.StatusBadRequest { + t.Fatalf("bounded result/error = %#v/%v", result, err) + } + }, + }, + } { + for _, overflow := range []bool{false, true} { + t.Run(tt.name+"/"+map[bool]string{false: "limit", true: "over-limit"}[overflow], func(t *testing.T) { + size := int(maxDistributorResponseBytes) + if overflow { + size++ + } + var uploadCalls int + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + if r.Method == http.MethodPost { + uploadCalls++ + statusCode := tt.statusCode + if statusCode == 0 { + statusCode = http.StatusAccepted + } + w.WriteHeader(statusCode) + _, _ = io.WriteString(w, tt.response(size)) + return + } + _, _ = io.WriteString(w, tt.statusBody(size)) + })) + defer server.Close() + + result, err := productionClient(t, server.URL, "test-upload-token").Upload(context.Background(), productionUploadRequest(t)) + tt.check(t, result, err, overflow) + if strings.Contains(fmt.Sprint(result), oversizedRemoteDiagnostic) || strings.Contains(fmt.Sprint(err), oversizedRemoteDiagnostic) { + t.Fatalf("result/error leaked oversized response detail: %#v/%v", result, err) + } + if uploadCalls != 1 { + t.Fatalf("upload calls = %d, want one", uploadCalls) + } + }) + } + } +} + +func productionClient(t *testing.T, endpoint, token string) *Client { + t.Helper() + cfg := config.Defaults().Notify.Distributor + cfg.Endpoint = endpoint + cfg.Timeout = 0 + t.Setenv(cfg.TokenEnv, token) + return New(cfg) +} + +func productionUploadRequest(t *testing.T) UploadRequest { + t.Helper() + path := filepath.Join(t.TempDir(), "report.md") + if err := os.WriteFile(path, []byte("report body"), 0o600); err != nil { + t.Fatal(err) + } + return UploadRequest{ + PipelineID: "weather", BundleID: "bundle", IdempotencyKey: "bundle-key", + Files: []UploadFile{{SourcePath: path, BundlePath: "daily/report.md"}}, + CreatedAt: time.Date(2026, 6, 7, 12, 0, 0, 0, time.UTC), + } +} + +func verifyUploadedArchive(t *testing.T, body io.Reader, wantPath, wantContents string) { + t.Helper() + reader, err := gzip.NewReader(body) + if err != nil { + t.Fatal(err) + } + defer reader.Close() + archive := tar.NewReader(reader) + for { + header, err := archive.Next() + if errors.Is(err, io.EOF) { + break + } + if err != nil { + t.Fatal(err) + } + if header.Name != wantPath { + continue + } + contents, err := io.ReadAll(archive) + if err != nil || string(contents) != wantContents { + t.Fatalf("archive file contents/error = %q/%v", contents, err) + } + return + } + t.Fatalf("archive did not contain %q", wantPath) +} + +func paddedJSON(t *testing.T, value string, size int) string { + t.Helper() + if len(value) > size { + t.Fatalf("JSON length = %d, exceeds requested size %d", len(value), size) + } + return value + strings.Repeat(" ", size-len(value)) +} + +func statusReportBody(t *testing.T, size int) string { + t.Helper() + const prefix = `{"run_id":"run-123","pipeline_id":"weather","status":"succeeded","report":"` + const suffix = `"}` + if len(prefix)+len(suffix) > size { + t.Fatalf("status response exceeds requested size %d", size) + } + return prefix + repeatedToLength(oversizedRemoteDiagnostic, size-len(prefix)-len(suffix)) + suffix +} + +func repeatedToLength(value string, size int) string { + return strings.Repeat(value, size/len(value)+1)[:size] +} diff --git a/internal/adapters/distributor/client_test.go b/internal/adapters/distributor/client_test.go index d646d0e..fba6dca 100644 --- a/internal/adapters/distributor/client_test.go +++ b/internal/adapters/distributor/client_test.go @@ -45,8 +45,8 @@ func TestUploadUsesConfiguredClientAndFiles(t *testing.T) { if result.RunID != "run-123" || result.Status != "succeeded" || result.UploadStatus != "accepted" { t.Fatalf("result = %#v, want accepted run", result) } - if result.RunStatus == nil || result.RunStatus.PipelineID != "reports" || !strings.Contains(string(result.RunStatus.Report), "replace_older") { - t.Fatalf("RunStatus = %#v, want parsed run report", result.RunStatus) + if result.RunStatus == nil || result.RunStatus.PipelineID != "reports" || len(result.RunStatus.Report) != 0 { + t.Fatalf("RunStatus = %#v, want safe status details", result.RunStatus) } if factory.endpoint != cfg.Endpoint { t.Fatalf("factory endpoint = %q, want %q", factory.endpoint, cfg.Endpoint) @@ -247,8 +247,8 @@ func TestUploadPollsUntilTerminalStatus(t *testing.T) { if err != nil { t.Fatalf("Upload() error = %v", err) } - if result.Status != "succeeded" || result.RunStatus == nil || !strings.Contains(string(result.RunStatus.Report), "replace_older") { - t.Fatalf("result = %#v, want terminal succeeded status with run report", result) + if result.Status != "succeeded" || result.RunStatus == nil || len(result.RunStatus.Report) != 0 { + t.Fatalf("result = %#v, want terminal succeeded status without remote report", result) } if factory.client.statusCalls != 2 { t.Fatalf("status calls = %d, want 2", factory.client.statusCalls) @@ -298,8 +298,8 @@ func TestUploadFailsWhenDistributorRunFailed(t *testing.T) { if err == nil { t.Fatal("Upload() error = nil, want failed distributor run error") } - if result.RunStatus == nil || result.RunStatus.Status != "failed" || !strings.Contains(string(result.RunStatus.Report), "failed") { - t.Fatalf("result = %#v, want failed run status report", result) + if result.RunStatus == nil || result.RunStatus.Status != "failed" || len(result.RunStatus.Report) != 0 || result.RunStatus.Error != "distributor reported a failed run" { + t.Fatalf("result = %#v, want safe failed run status", result) } if strings.Contains(err.Error(), "secret-token") || strings.Contains(result.RunStatus.Error, "secret-token") { t.Fatalf("error/result leaked token: err=%q result=%#v", err.Error(), result) diff --git a/internal/app/app.go b/internal/app/app.go index d033655..5cc202d 100644 --- a/internal/app/app.go +++ b/internal/app/app.go @@ -650,25 +650,7 @@ func (n distributorNotifier) Notify(ctx context.Context, req NotificationRequest Files: distributorUploadFiles(req.ReportPath, req.BundlePaths), CreatedAt: req.CreatedAt, }) - notification := &NotificationResult{ - PipelineID: req.PipelineID, - BundleID: req.BundleID, - IdempotencyKey: req.IdempotencyKey, - RunID: result.RunID, - Status: result.Status, - UploadStatus: result.UploadStatus, - StatusError: result.StatusError, - } - if result.RunStatus != nil { - if result.RunStatus.PipelineID != "" { - notification.PipelineID = result.RunStatus.PipelineID - } - notification.AcceptedAt = result.RunStatus.AcceptedAt - notification.StartedAt = result.RunStatus.StartedAt - notification.FinishedAt = result.RunStatus.FinishedAt - notification.Report = append([]byte(nil), result.RunStatus.Report...) - notification.Error = result.RunStatus.Error - } + notification := notificationResultFromUpload(req.PipelineID, req.BundleID, req.IdempotencyKey, result) if err != nil { return notification, err } @@ -692,7 +674,7 @@ func notificationResultFromUpload(pipelineID string, bundleID string, idempotenc RunID: result.RunID, Status: result.Status, UploadStatus: result.UploadStatus, - StatusError: result.StatusError, + StatusError: safeDistributorStatusError(result.StatusError), } if result.RunStatus != nil { if result.RunStatus.PipelineID != "" { @@ -701,12 +683,25 @@ func notificationResultFromUpload(pipelineID string, bundleID string, idempotenc notification.AcceptedAt = result.RunStatus.AcceptedAt notification.StartedAt = result.RunStatus.StartedAt notification.FinishedAt = result.RunStatus.FinishedAt - notification.Report = append([]byte(nil), result.RunStatus.Report...) - notification.Error = result.RunStatus.Error + notification.Error = safeDistributorRunError(result.RunStatus.Error) } return notification } +func safeDistributorStatusError(value string) string { + if value == "" { + return "" + } + return "distributor status could not be confirmed" +} + +func safeDistributorRunError(value string) string { + if value == "" { + return "" + } + return "distributor reported a failed run" +} + func distributorUploadFiles(sourcePath string, bundlePaths []string) []distributoradapter.UploadFile { files := make([]distributoradapter.UploadFile, 0, len(bundlePaths)) for _, bundlePath := range bundlePaths { diff --git a/internal/app/batch_notification.go b/internal/app/batch_notification.go index bc7403e..861e502 100644 --- a/internal/app/batch_notification.go +++ b/internal/app/batch_notification.go @@ -81,7 +81,7 @@ func notifyBatch(ctx context.Context, cfg config.Config, batch BatchKind, runID batchResult := batchNotificationResult(req, notification) if wrappedErr != nil { batchResult.Status = "failed" - batchResult.Error = wrappedErr.Error() + batchResult.Error = safeDistributorNotificationFailure(wrappedErr) return batchResult } return batchResult @@ -233,7 +233,7 @@ func batchNotificationResult(req batchNotificationRequest, result *NotificationR notification.IdempotencyKey = result.IdempotencyKey } if result.Error != "" { - notification.Error = result.Error + notification.Error = safeDistributorRunError(result.Error) } } if notification.Status == "" { @@ -246,11 +246,18 @@ func failedBatchNotificationResult(req batchNotificationRequest, err error) *Bat notification := batchNotificationResult(req, nil) notification.Status = "failed" if err != nil { - notification.Error = err.Error() + notification.Error = safeDistributorNotificationFailure(err) } return notification } +func safeDistributorNotificationFailure(err error) string { + if err == nil { + return "" + } + return "distributor notification failed" +} + func renderBatchNotificationIdentity(cfg config.Config, batch BatchKind, runID string, startedAt time.Time) (batchNotificationIdentity, error) { values, err := batchNotificationTemplateValues(cfg, batch, runID, startedAt) if err != nil { diff --git a/internal/app/distributor_notification_test.go b/internal/app/distributor_notification_test.go new file mode 100644 index 0000000..f52bf95 --- /dev/null +++ b/internal/app/distributor_notification_test.go @@ -0,0 +1,45 @@ +package app + +import ( + "errors" + "strings" + "testing" + + distributoradapter "gitea.maximumdirect.net/eric/weatherreporter/internal/adapters/distributor" +) + +func TestNotificationResultFromUploadExcludesRemoteResponseDetails(t *testing.T) { + const remote = "REMOTE-DIAGNOSTIC" + notification := notificationResultFromUpload("weather", "bundle", "key", distributoradapter.UploadResult{ + RunID: "run-123", Status: "failed", UploadStatus: "accepted", StatusError: remote, + RunStatus: &distributoradapter.RunStatus{PipelineID: "weather", Status: "failed", Report: []byte(`{"detail":"REMOTE-DIAGNOSTIC"}`), Error: remote}, + }) + if notification == nil || notification.StatusError != "distributor status could not be confirmed" || notification.Error != "distributor reported a failed run" || len(notification.Report) != 0 { + t.Fatalf("notification = %#v", notification) + } + if strings.Contains(notification.StatusError, remote) || strings.Contains(notification.Error, remote) { + t.Fatalf("notification includes remote detail: %#v", notification) + } +} + +func TestBatchNotificationResultExcludesRemoteResponseDetails(t *testing.T) { + const remote = "REMOTE-DIAGNOSTIC" + notification := batchNotificationResult(batchNotificationRequest{PipelineID: "weather", BundleID: "bundle", IdempotencyKey: "key"}, &NotificationResult{Status: "failed", Error: remote}) + if notification == nil || notification.Error != "distributor reported a failed run" { + t.Fatalf("notification = %#v", notification) + } + if strings.Contains(notification.Error, remote) { + t.Fatalf("notification includes remote detail: %#v", notification) + } +} + +func TestFailedBatchNotificationResultExcludesRemoteResponseDetails(t *testing.T) { + const remote = "REMOTE-DIAGNOSTIC" + notification := failedBatchNotificationResult(batchNotificationRequest{PipelineID: "weather", BundleID: "bundle", IdempotencyKey: "key"}, errors.New(remote)) + if notification == nil || notification.Error != "distributor notification failed" { + t.Fatalf("notification = %#v", notification) + } + if strings.Contains(notification.Error, remote) { + t.Fatalf("notification includes remote detail: %#v", notification) + } +}