Report sanitized tus upload creation errors

Extract tus and backend error codes from bounded response bodies to improve failure diagnostics without exposing sensitive messages. Add regression coverage for sanitization.

machinery/src/cloud/tus_client.go

@@ -10,6 +10,7 @@ import (
 	"net/url"
 	"os"
 	"path/filepath"
+	"regexp"
 	"sort"
 	"strconv"
 	"strings"
@@ -49,6 +50,12 @@ func resumableUploadsEnabled() bool {
 const tusDefaultChunkSize int64 = 8 << 20 // 8 MiB (>= S3 multipart minimum part size)
 
 const tusProgressBucketPercent int64 = 10
+const tusErrorBodyLimit int64 = 16 << 10
+
+var (
+	tusErrorCodePattern    = regexp.MustCompile(`\bERR_[A-Z0-9_]+\b`)
+	tusBackendErrorPattern = regexp.MustCompile(`\bapi error ([A-Za-z][A-Za-z0-9._-]{0,63}):`)
+)
 
 // tusChunkSize returns the number of bytes to send per PATCH request. It
 // defaults to tusDefaultChunkSize (8 MiB) and can be overridden with the
@@ -144,6 +151,7 @@ func runTusUpload(baseURL, metadata, fileName, label, slot string, setHeaders tu
 
 	sidecar := tusSidecarPath(fileName, slot)
 	uploadURL := loadTusResumeState(sidecar, baseURL)
+	uploadEstablished := uploadURL != ""
 
 	const maxAttempts = 4
 	restartedAfterComplete := false
@@ -181,6 +189,7 @@ func runTusUpload(baseURL, metadata, fileName, label, slot string, setHeaders tu
 				continue
 			}
 			uploadURL = created
+			uploadEstablished = true
 			saveTusResumeState(sidecar, tusResumeState{UploadURL: uploadURL, VaultURI: baseURL, Size: size})
 		}
 
@@ -290,7 +299,15 @@ func runTusUpload(baseURL, metadata, fileName, label, slot string, setHeaders tu
 	// report responded=false so the caller keeps the recording queued and retries
 	// later instead of consuming its retry budget and entering the long back-off
 	// timeout.
-	return false, lastStatus > 0, true, "resumable upload did not complete after retries", errors.New(label + ": resumable upload did not complete after retries")
+	supported = true
+	if !uploadEstablished && lastStatus >= http.StatusInternalServerError {
+		// A persistent server failure before creation means no upload URL was
+		// issued and no recording bytes were sent. Falling back to the legacy
+		// endpoint is therefore safe and preserves uploads while the server-side
+		// tus/provider path is unavailable.
+		supported = false
+	}
+	return false, lastStatus > 0, supported, "resumable upload did not complete after retries", errors.New(label + ": resumable upload did not complete after retries")
 }
 
 // uploadVaultResumable uploads a recording directly to a Kerberos Vault using
@@ -390,11 +407,17 @@ func tusCreate(client *http.Client, baseURL string, size int64, metadata string,
 	if err != nil {
 		return "", 0, err
 	}
-	io.Copy(io.Discard, resp.Body)
 
 	if resp.StatusCode != http.StatusCreated {
+		body, _ := io.ReadAll(io.LimitReader(resp.Body, tusErrorBodyLimit))
+		_, _ = io.Copy(io.Discard, resp.Body)
+		details := tusErrorDetails(body)
+		if details != "" {
+			return "", resp.StatusCode, fmt.Errorf("unexpected status creating upload: %s (%s)", resp.Status, details)
+		}
 		return "", resp.StatusCode, fmt.Errorf("unexpected status creating upload: %s", resp.Status)
 	}
+	_, _ = io.Copy(io.Discard, resp.Body)
 	location := resp.Header.Get("Location")
 	if location == "" {
 		return "", resp.StatusCode, errors.New("missing Location header in create response")
@@ -402,6 +425,17 @@ func tusCreate(client *http.Client, baseURL string, size int64, metadata string,
 	return resolveTusLocation(baseURL, location), resp.StatusCode, nil
 }
 
+func tusErrorDetails(body []byte) string {
+	var details []string
+	if code := tusErrorCodePattern.Find(body); len(code) != 0 {
+		details = append(details, string(code))
+	}
+	if match := tusBackendErrorPattern.FindSubmatch(body); len(match) == 2 {
+		details = append(details, "backend="+string(match[1]))
+	}
+	return strings.Join(details, ", ")
+}
+
 // tusHead performs the tus "offset" request (HEAD) and returns the current
 // server-side upload offset.
 func tusHead(client *http.Client, uploadURL string, setHeaders tusHeaderFunc) (int64, int, error) {

machinery/src/cloud/tus_client_test.go

@@ -46,6 +46,8 @@ type fakeTus struct {
 	// unsupported makes the creation endpoint return 404, simulating an older
 	// vault without a tus endpoint.
 	unsupported bool
+	// createStatus makes every creation request fail with the configured status.
+	createStatus int
 	// failFinalize causes the next N completing PATCH requests to return 502
 	// after storing the bytes, simulating a failed completion hook.
 	failFinalize int
@@ -134,6 +136,13 @@ func (s *fakeTus) ServeHTTP(w http.ResponseWriter, r *http.Request) {
 			w.WriteHeader(http.StatusNotFound)
 			return
 		}
+		if s.createStatus != 0 {
+			s.mu.Lock()
+			s.creates++
+			s.mu.Unlock()
+			w.WriteHeader(s.createStatus)
+			return
+		}
 		length, _ := strconv.ParseInt(r.Header.Get("Upload-Length"), 10, 64)
 		s.mu.Lock()
 		s.counter++
@@ -840,6 +849,135 @@ func TestUploadHubResumable_Unsupported(t *testing.T) {
 	}
 }
 
+func TestUploadHubResumableFallsBackAfterCreateServerErrors(t *testing.T) {
+	srv := newFakeTus()
+	srv.createStatus = http.StatusInternalServerError
+	ts := httptest.NewServer(srv)
+	defer ts.Close()
+
+	fileName := "f.mp4"
+	withRecording(t, fileName, []byte("hello"))
+	oldDelay := tusBackoffBaseDelay
+	tusBackoffBaseDelay = time.Millisecond
+	t.Cleanup(func() { tusBackoffBaseDelay = oldDelay })
+
+	uploaded, responded, supported, _, err := uploadHubResumable(testHubConfig(ts.URL), fileName, "test", "hub")
+	if uploaded {
+		t.Fatal("uploaded = true, want false")
+	}
+	if !responded {
+		t.Fatal("responded = false, want true")
+	}
+	if supported {
+		t.Fatal("supported = true, want false so the caller uses the legacy upload")
+	}
+	if err == nil {
+		t.Fatal("error = nil, want resumable create failure")
+	}
+	if got := srv.createCount(); got != 4 {
+		t.Fatalf("create attempts = %d, want 4", got)
+	}
+}
+
+func TestUploadKerberosHubFallsBackThroughHubAfterTusCreateServerErrors(t *testing.T) {
+	fileName := "f.mp4"
+	payload := []byte("recording")
+	withRecording(t, fileName, payload)
+	t.Setenv("AGENT_DISABLE_RESUMABLE_UPLOAD", "")
+
+	oldDelay := tusBackoffBaseDelay
+	tusBackoffBaseDelay = time.Millisecond
+	t.Cleanup(func() { tusBackoffBaseDelay = oldDelay })
+
+	var tusCreates, legacyHeads, legacyPosts int
+	server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		switch {
+		case r.Method == http.MethodPost && r.URL.Path == tusUploadPath:
+			tusCreates++
+			w.WriteHeader(http.StatusInternalServerError)
+		case r.Method == http.MethodHead && r.URL.Path == "/storage/upload":
+			legacyHeads++
+			w.WriteHeader(http.StatusOK)
+		case r.Method == http.MethodPost && r.URL.Path == "/storage/upload":
+			legacyPosts++
+			got, err := io.ReadAll(r.Body)
+			if err != nil {
+				t.Errorf("read legacy upload: %v", err)
+			}
+			if !bytes.Equal(got, payload) {
+				t.Errorf("legacy upload body = %q, want %q", got, payload)
+			}
+			w.WriteHeader(http.StatusOK)
+		default:
+			t.Errorf("unexpected request: %s %s", r.Method, r.URL.Path)
+			w.WriteHeader(http.StatusNotFound)
+		}
+	}))
+	defer server.Close()
+
+	config := testHubConfig(server.URL)
+	uploaded, responded, err := UploadKerberosHub(&models.Configuration{Config: *config}, fileName)
+	if err != nil {
+		t.Fatalf("UploadKerberosHub error: %v", err)
+	}
+	if !uploaded || !responded {
+		t.Fatalf("UploadKerberosHub = uploaded %v, responded %v; want true, true", uploaded, responded)
+	}
+	if tusCreates != 4 {
+		t.Errorf("tus create attempts = %d, want 4", tusCreates)
+	}
+	if legacyHeads != 1 {
+		t.Errorf("legacy Hub authorization requests = %d, want 1", legacyHeads)
+	}
+	if legacyPosts != 1 {
+		t.Errorf("legacy Hub upload requests = %d, want 1", legacyPosts)
+	}
+}
+
+func TestTusCreateReportsSanitizedServerErrorCodes(t *testing.T) {
+	const sensitiveMessage = "credential=do-not-log-this"
+	server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) {
+		w.WriteHeader(http.StatusInternalServerError)
+		_, _ = io.WriteString(w, "ERR_INTERNAL_SERVER_ERROR: operation error S3: CreateMultipartUpload, api error AccessDenied: "+sensitiveMessage)
+	}))
+	defer server.Close()
+
+	_, status, err := tusCreate(server.Client(), server.URL, 8, "", func(http.Header, string) {}, "f.mp4")
+	if status != http.StatusInternalServerError {
+		t.Fatalf("status = %d, want %d", status, http.StatusInternalServerError)
+	}
+	if err == nil {
+		t.Fatal("error = nil, want create failure")
+	}
+	message := err.Error()
+	for _, want := range []string{"ERR_INTERNAL_SERVER_ERROR", "backend=AccessDenied"} {
+		if !strings.Contains(message, want) {
+			t.Errorf("error %q does not contain %q", message, want)
+		}
+	}
+	if strings.Contains(message, sensitiveMessage) {
+		t.Fatalf("error exposed backend message: %q", message)
+	}
+}
+
+func TestUploadHubResumableDoesNotFallbackAfterCreateClientError(t *testing.T) {
+	srv := newFakeTus()
+	srv.createStatus = http.StatusBadRequest
+	ts := httptest.NewServer(srv)
+	defer ts.Close()
+
+	fileName := "f.mp4"
+	withRecording(t, fileName, []byte("hello"))
+	oldDelay := tusBackoffBaseDelay
+	tusBackoffBaseDelay = time.Millisecond
+	t.Cleanup(func() { tusBackoffBaseDelay = oldDelay })
+
+	_, _, supported, _, _ := uploadHubResumable(testHubConfig(ts.URL), fileName, "test", "hub")
+	if !supported {
+		t.Fatal("supported = false for a client rejection, which must not bypass authorization via legacy fallback")
+	}
+}
+
 func TestEncodeTusMetadata(t *testing.T) {
 	got := encodeTusMetadata(map[string]string{
 		"b":     "2",