diff --git a/telemetry/server/api/handlers.go b/telemetry/server/api/handlers.go index b78e8c451..b3faf234d 100644 --- a/telemetry/server/api/handlers.go +++ b/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 } diff --git a/telemetry/server/api/handlers_test.go b/telemetry/server/api/handlers_test.go new file mode 100644 index 000000000..d6c10bf1b --- /dev/null +++ b/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) + } + }) +} diff --git a/telemetry/server/api/validate.go b/telemetry/server/api/validate.go new file mode 100644 index 000000000..8df581626 --- /dev/null +++ b/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 +} diff --git a/telemetry/test/integration.go b/telemetry/test/integration.go index f75a3ae89..cddb13a75 100644 --- a/telemetry/test/integration.go +++ b/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