telemetry: validate reports on the collect endpoint (#10401)

* telemetry: validate reports on the collect endpoint

/api/collect is anonymous, so reports can't be authenticated, but a
real master can't produce a non-UUID topology_id, a version outside
N.NN(-enterprise), an unknown GOOS/GOARCH, or absurd counts — reject
those to keep casual junk out of the collected data, and cap the
request body at 4 KB.

* telemetry: integration test fixtures pass collect validation

The test's topology id and version were exactly the junk shapes the
new validation rejects; use a UUID and a plain version number.

telemetry/server/api/handlers.go

@@ -34,8 +34,8 @@ func (h *Handler) CollectTelemetry(w http.ResponseWriter, r *http.Request) {
 		return
 	}
 
-	// Read protobuf request
-	body, err := io.ReadAll(r.Body)
+	// Read protobuf request; real reports are well under 1 KB
+	body, err := io.ReadAll(http.MaxBytesReader(w, r.Body, maxRequestBytes))
 	if err != nil {
 		http.Error(w, "Failed to read request body", http.StatusBadRequest)
 		return
@@ -53,9 +53,8 @@ func (h *Handler) CollectTelemetry(w http.ResponseWriter, r *http.Request) {
 		return
 	}
 
-	// Validate required fields
-	if data.TopologyId == "" || data.Version == "" || data.Os == "" {
-		http.Error(w, "Missing required fields", http.StatusBadRequest)
+	if err := validateTelemetryData(data); err != nil {
+		http.Error(w, "Invalid telemetry data: "+err.Error(), http.StatusBadRequest)
 		return
 	}
 

telemetry/server/api/handlers_test.go

@@ -0,0 +1,98 @@
+package api
+
+import (
+	"bytes"
+	"net/http"
+	"net/http/httptest"
+	"testing"
+
+	"github.com/seaweedfs/seaweedfs/telemetry/proto"
+	"github.com/seaweedfs/seaweedfs/telemetry/server/storage"
+	protobuf "google.golang.org/protobuf/proto"
+)
+
+func validReport() *proto.TelemetryData {
+	return &proto.TelemetryData{
+		TopologyId:        "38422678-6a0d-4482-aa33-65b90010ac47",
+		Version:           "4.40",
+		Os:                "linux/amd64",
+		VolumeServerCount: 5,
+		TotalDiskBytes:    123456789,
+		TotalVolumeCount:  42,
+		FilerCount:        2,
+		BrokerCount:       1,
+	}
+}
+
+func postCollect(t *testing.T, h *Handler, body []byte, contentType string) *httptest.ResponseRecorder {
+	t.Helper()
+	req := httptest.NewRequest(http.MethodPost, "/api/collect", bytes.NewReader(body))
+	req.Header.Set("Content-Type", contentType)
+	w := httptest.NewRecorder()
+	h.CollectTelemetry(w, req)
+	return w
+}
+
+func marshalReport(t *testing.T, data *proto.TelemetryData) []byte {
+	t.Helper()
+	body, err := protobuf.Marshal(&proto.TelemetryRequest{Data: data})
+	if err != nil {
+		t.Fatalf("marshal: %v", err)
+	}
+	return body
+}
+
+func TestCollectTelemetryValidation(t *testing.T) {
+	// promauto registers on the global registry: one storage per test binary.
+	h := NewHandler(storage.NewPrometheusStorage())
+
+	t.Run("valid report accepted", func(t *testing.T) {
+		if w := postCollect(t, h, marshalReport(t, validReport()), "application/x-protobuf"); w.Code != http.StatusOK {
+			t.Errorf("got %d, want 200: %s", w.Code, w.Body.String())
+		}
+	})
+
+	t.Run("enterprise version accepted", func(t *testing.T) {
+		data := validReport()
+		data.Version = "4.40-enterprise"
+		if w := postCollect(t, h, marshalReport(t, data), "application/x-protobuf"); w.Code != http.StatusOK {
+			t.Errorf("got %d, want 200: %s", w.Code, w.Body.String())
+		}
+	})
+
+	rejected := []struct {
+		name   string
+		mutate func(*proto.TelemetryData)
+	}{
+		{"non-UUID topology_id", func(d *proto.TelemetryData) { d.TopologyId = "claude-diagnostic-probe" }},
+		{"empty topology_id", func(d *proto.TelemetryData) { d.TopologyId = "" }},
+		{"junk version", func(d *proto.TelemetryData) { d.Version = "probe" }},
+		{"version with suffix", func(d *proto.TelemetryData) { d.Version = "4.40-nightly" }},
+		{"junk os", func(d *proto.TelemetryData) { d.Os = "probe/probe" }},
+		{"os without slash", func(d *proto.TelemetryData) { d.Os = "linux" }},
+		{"negative count", func(d *proto.TelemetryData) { d.VolumeServerCount = -1 }},
+		{"absurd server count", func(d *proto.TelemetryData) { d.VolumeServerCount = 1_000_000 }},
+		{"absurd disk size", func(d *proto.TelemetryData) { d.TotalDiskBytes = 1 << 62 }},
+	}
+	for _, tc := range rejected {
+		t.Run(tc.name+" rejected", func(t *testing.T) {
+			data := validReport()
+			tc.mutate(data)
+			if w := postCollect(t, h, marshalReport(t, data), "application/x-protobuf"); w.Code != http.StatusBadRequest {
+				t.Errorf("got %d, want 400: %s", w.Code, w.Body.String())
+			}
+		})
+	}
+
+	t.Run("wrong content type rejected", func(t *testing.T) {
+		if w := postCollect(t, h, marshalReport(t, validReport()), "application/json"); w.Code != http.StatusUnsupportedMediaType {
+			t.Errorf("got %d, want 415", w.Code)
+		}
+	})
+
+	t.Run("oversized body rejected", func(t *testing.T) {
+		if w := postCollect(t, h, make([]byte, maxRequestBytes+1), "application/x-protobuf"); w.Code != http.StatusBadRequest {
+			t.Errorf("got %d, want 400", w.Code)
+		}
+	})
+}

telemetry/server/api/validate.go

@@ -0,0 +1,58 @@
+package api
+
+import (
+	"fmt"
+	"regexp"
+	"strings"
+
+	"github.com/seaweedfs/seaweedfs/telemetry/proto"
+)
+
+// The collect endpoint is anonymous, so reports can't be authenticated —
+// these checks only reject values a real SeaweedFS master can't produce,
+// to keep casual junk out of the collected data.
+
+const (
+	maxRequestBytes = 4096
+	maxServerCount  = 100_000       // volume servers / filers / brokers per cluster
+	maxVolumeCount  = 100_000_000   // volumes per cluster
+	maxDiskBytes    = uint64(1) << 60 // 1 EiB
+)
+
+var (
+	// TopologyId is a raft-generated UUID.
+	uuidRe = regexp.MustCompile(`^[0-9a-fA-F]{8}-[0-9a-fA-F]{4}-[0-9a-fA-F]{4}-[0-9a-fA-F]{4}-[0-9a-fA-F]{12}$`)
+	// Version is VERSION_NUMBER, e.g. "4.40", plus "-enterprise" for enterprise builds.
+	versionRe = regexp.MustCompile(`^\d+\.\d+(-enterprise)?$`)
+
+	validGoos = map[string]bool{
+		"linux": true, "darwin": true, "windows": true,
+		"freebsd": true, "openbsd": true, "netbsd": true,
+	}
+	validGoarch = map[string]bool{
+		"amd64": true, "arm64": true, "arm": true, "386": true,
+		"riscv64": true, "ppc64le": true, "s390x": true,
+		"mips64": true, "mips64le": true, "loong64": true,
+	}
+)
+
+func validateTelemetryData(data *proto.TelemetryData) error {
+	if !uuidRe.MatchString(data.TopologyId) {
+		return fmt.Errorf("topology_id is not a UUID")
+	}
+	if !versionRe.MatchString(data.Version) {
+		return fmt.Errorf("unrecognized version")
+	}
+	goos, goarch, ok := strings.Cut(data.Os, "/")
+	if !ok || !validGoos[goos] || !validGoarch[goarch] {
+		return fmt.Errorf("unrecognized os")
+	}
+	if data.VolumeServerCount < 0 || data.VolumeServerCount > maxServerCount ||
+		data.FilerCount < 0 || data.FilerCount > maxServerCount ||
+		data.BrokerCount < 0 || data.BrokerCount > maxServerCount ||
+		data.TotalVolumeCount < 0 || data.TotalVolumeCount > maxVolumeCount ||
+		data.TotalDiskBytes > maxDiskBytes {
+		return fmt.Errorf("values out of range")
+	}
+	return nil
+}

telemetry/test/integration.go

@@ -157,8 +157,8 @@ func waitForServer(url string, timeout time.Duration) bool {
 func testProtobufMarshaling() error {
 	// Test protobuf marshaling/unmarshaling
 	testData := &proto.TelemetryData{
-		TopologyId:        "test-cluster-12345",
-		Version:           "test-3.45",
+		TopologyId:        "00000000-0000-4000-8000-000000012345",
+		Version:           "3.45",
 		Os:                "linux/amd64",
 		VolumeServerCount: 2,
 		TotalDiskBytes:    1000000,
@@ -199,11 +199,11 @@ func testProtobufMarshaling() error {
 func testTelemetryClient() error {
 	// Create telemetry client
 	client := telemetry.NewClient(serverURL+"/api/collect", true)
-	client.SetTopologyId("test-topology-12345")
+	client.SetTopologyId("00000000-0000-4000-8000-000000054321")
 
 	// Create test data using protobuf format
 	testData := &proto.TelemetryData{
-		Version:           "test-3.45",
+		Version:           "3.45",
 		Os:                "linux/amd64",
 		VolumeServerCount: 3,
 		TotalDiskBytes:    1073741824, // 1GB