diff --git a/client/internal/metrics/infra/ingest/main.go b/client/internal/metrics/infra/ingest/main.go index cda5315d2..a5ada0ca2 100644 --- a/client/internal/metrics/infra/ingest/main.go +++ b/client/internal/metrics/infra/ingest/main.go @@ -22,6 +22,7 @@ const ( maxDurationSeconds = 86400.0 // reject any duration field > 24 hours peerIDLength = 16 // truncated SHA-256: 8 bytes = 16 hex chars maxTagValueLength = 64 // reject tag values longer than this + peerIDTag = "peer_id" ) type measurementSpec struct { @@ -137,7 +138,8 @@ func handleIngest(client *http.Client, influxURL, influxToken string) http.Handl return } - if err := validatePeerIDFormat(r); err != nil { + peerID := r.Header.Get("X-Peer-ID") + if err := validatePeerIDFormat(peerID); err != nil { http.Error(w, err.Error(), http.StatusBadRequest) return } @@ -152,7 +154,7 @@ func handleIngest(client *http.Client, influxURL, influxToken string) http.Handl return } - validated, err := validateLineProtocol(body) + validated, err := validateLineProtocol(body, peerID) if err != nil { log.Printf("WARN validation failed from %s: %v", r.RemoteAddr, err) http.Error(w, err.Error(), http.StatusBadRequest) @@ -192,8 +194,7 @@ func forwardToInflux(w http.ResponseWriter, r *http.Request, client *http.Client // unauthenticated so that peers of self-hosted deployments, for which no shared // trust anchor exists, can report obfuscated telemetry. The check exists to bound // tag cardinality, so a malformed value is a bad request rather than an auth failure. -func validatePeerIDFormat(r *http.Request) error { - peerID := r.Header.Get("X-Peer-ID") +func validatePeerIDFormat(peerID string) error { if peerID == "" { return fmt.Errorf("missing X-Peer-ID header") } @@ -224,7 +225,7 @@ func readBody(r *http.Request) ([]byte, error) { // validateLineProtocol parses InfluxDB line protocol lines, // whitelists measurements and fields, and checks value bounds. -func validateLineProtocol(body []byte) ([]byte, error) { +func validateLineProtocol(body []byte, peerID string) ([]byte, error) { lines := strings.Split(strings.TrimSpace(string(body)), "\n") var valid []string @@ -234,7 +235,7 @@ func validateLineProtocol(body []byte) ([]byte, error) { continue } - if err := validateLine(line); err != nil { + if err := validateLine(line, peerID); err != nil { return nil, err } @@ -248,7 +249,7 @@ func validateLineProtocol(body []byte) ([]byte, error) { return []byte(strings.Join(valid, "\n") + "\n"), nil } -func validateLine(line string) error { +func validateLine(line, peerID string) error { // line protocol: measurement,tag=val,tag=val field=val,field=val timestamp parts := strings.SplitN(line, " ", 3) if len(parts) < 2 { @@ -266,7 +267,7 @@ func validateLine(line string) error { // Validate tags (everything after measurement name in parts[0]) for _, tagPair := range measurementAndTags[1:] { - if err := validateTag(tagPair, measurement, spec.allowedTags); err != nil { + if err := validateTag(tagPair, measurement, peerID, spec.allowedTags); err != nil { return err } } @@ -281,7 +282,7 @@ func validateLine(line string) error { return nil } -func validateTag(pair, measurement string, allowedTags map[string]bool) error { +func validateTag(pair, measurement, peerID string, allowedTags map[string]bool) error { kv := strings.SplitN(pair, "=", 2) if len(kv) != 2 { return fmt.Errorf("invalid tag: %q", pair) @@ -296,6 +297,10 @@ func validateTag(pair, measurement string, allowedTags map[string]bool) error { return fmt.Errorf("tag value too long for %q: %d > %d", tagName, len(kv[1]), maxTagValueLength) } + if tagName == peerIDTag && kv[1] != peerID { + return fmt.Errorf("peer_id tag does not match X-Peer-ID header") + } + return nil } diff --git a/client/internal/metrics/infra/ingest/main_test.go b/client/internal/metrics/infra/ingest/main_test.go index 526f127cb..5998809b6 100644 --- a/client/internal/metrics/infra/ingest/main_test.go +++ b/client/internal/metrics/infra/ingest/main_test.go @@ -1,7 +1,6 @@ package main import ( - "net/http" "strings" "testing" @@ -11,57 +10,57 @@ import ( func TestValidateLine_ValidPeerConnection(t *testing.T) { line := `netbird_peer_connection,deployment_type=cloud,connection_type=ice,attempt_type=initial,version=1.0.0,os=linux,arch=amd64,peer_id=abcdef0123456789,connection_pair_id=pair1234 signaling_to_connection_seconds=1.5,connection_to_wg_handshake_seconds=0.5,total_seconds=2 1234567890` - assert.NoError(t, validateLine(line)) + assert.NoError(t, validateLine(line, "abcdef0123456789")) } func TestValidateLine_ValidSync(t *testing.T) { line := `netbird_sync,deployment_type=selfhosted,version=2.0.0,os=darwin,arch=arm64,peer_id=abcdef0123456789 duration_seconds=1.5 1234567890` - assert.NoError(t, validateLine(line)) + assert.NoError(t, validateLine(line, "abcdef0123456789")) } func TestValidateLine_ValidLogin(t *testing.T) { line := `netbird_login,deployment_type=cloud,result=success,version=1.0.0,os=linux,arch=amd64,peer_id=abcdef0123456789 duration_seconds=3.2 1234567890` - assert.NoError(t, validateLine(line)) + assert.NoError(t, validateLine(line, "abcdef0123456789")) } func TestValidateLine_UnknownMeasurement(t *testing.T) { line := `unknown_metric,foo=bar value=1 1234567890` - err := validateLine(line) + err := validateLine(line, "abc") require.Error(t, err) assert.Contains(t, err.Error(), "unknown measurement") } func TestValidateLine_UnknownTag(t *testing.T) { line := `netbird_sync,deployment_type=cloud,evil_tag=injected,version=1.0.0,os=linux,arch=amd64,peer_id=abc duration_seconds=1.5 1234567890` - err := validateLine(line) + err := validateLine(line, "abc") require.Error(t, err) assert.Contains(t, err.Error(), "unknown tag") } func TestValidateLine_UnknownField(t *testing.T) { line := `netbird_sync,deployment_type=cloud,version=1.0.0,os=linux,arch=amd64,peer_id=abc injected_field=1 1234567890` - err := validateLine(line) + err := validateLine(line, "abc") require.Error(t, err) assert.Contains(t, err.Error(), "unknown field") } func TestValidateLine_NegativeValue(t *testing.T) { line := `netbird_sync,deployment_type=cloud,version=1.0.0,os=linux,arch=amd64,peer_id=abc duration_seconds=-1.5 1234567890` - err := validateLine(line) + err := validateLine(line, "abc") require.Error(t, err) assert.Contains(t, err.Error(), "negative") } func TestValidateLine_DurationTooLarge(t *testing.T) { line := `netbird_sync,deployment_type=cloud,version=1.0.0,os=linux,arch=amd64,peer_id=abc duration_seconds=100000 1234567890` - err := validateLine(line) + err := validateLine(line, "abc") require.Error(t, err) assert.Contains(t, err.Error(), "too large") } func TestValidateLine_TotalSecondsTooLarge(t *testing.T) { line := `netbird_peer_connection,deployment_type=cloud,connection_type=ice,attempt_type=initial,version=1.0.0,os=linux,arch=amd64,peer_id=abc,connection_pair_id=pair total_seconds=100000 1234567890` - err := validateLine(line) + err := validateLine(line, "abc") require.Error(t, err) assert.Contains(t, err.Error(), "too large") } @@ -69,7 +68,7 @@ func TestValidateLine_TotalSecondsTooLarge(t *testing.T) { func TestValidateLine_TagValueTooLong(t *testing.T) { longTag := strings.Repeat("a", maxTagValueLength+1) line := `netbird_sync,deployment_type=` + longTag + `,version=1.0.0,os=linux,arch=amd64,peer_id=abc duration_seconds=1.5 1234567890` - err := validateLine(line) + err := validateLine(line, "abc") require.Error(t, err) assert.Contains(t, err.Error(), "tag value too long") } @@ -79,7 +78,7 @@ func TestValidateLineProtocol_MultipleLines(t *testing.T) { "netbird_sync,deployment_type=cloud,version=1.0.0,os=linux,arch=amd64,peer_id=abc duration_seconds=1.5 1234567890\n" + "netbird_login,deployment_type=cloud,result=success,version=1.0.0,os=linux,arch=amd64,peer_id=abc duration_seconds=2.0 1234567890\n", ) - validated, err := validateLineProtocol(body) + validated, err := validateLineProtocol(body, "abc") require.NoError(t, err) assert.Contains(t, string(validated), "netbird_sync") assert.Contains(t, string(validated), "netbird_login") @@ -90,7 +89,7 @@ func TestValidateLineProtocol_RejectsOnBadLine(t *testing.T) { "netbird_sync,deployment_type=cloud,version=1.0.0,os=linux,arch=amd64,peer_id=abc duration_seconds=1.5 1234567890\n" + "evil_metric,foo=bar value=1 1234567890\n", ) - _, err := validateLineProtocol(body) + _, err := validateLineProtocol(body, "abc") require.Error(t, err) } @@ -109,11 +108,7 @@ func TestValidatePeerIDFormat(t *testing.T) { for _, tt := range tests { t.Run(tt.name, func(t *testing.T) { - r, _ := http.NewRequest(http.MethodPost, "/", nil) - if tt.peerID != "" { - r.Header.Set("X-Peer-ID", tt.peerID) - } - err := validatePeerIDFormat(r) + err := validatePeerIDFormat(tt.peerID) if tt.wantErr { require.Error(t, err) } else { @@ -122,3 +117,20 @@ func TestValidatePeerIDFormat(t *testing.T) { }) } } + +func TestValidateLine_PeerIDTagMismatchesHeader(t *testing.T) { + line := `netbird_sync,deployment_type=cloud,version=1.0.0,os=linux,arch=amd64,peer_id=deadbeefdeadbeef duration_seconds=1.5 1234567890` + err := validateLine(line, "abcdef0123456789") + require.Error(t, err) + assert.Contains(t, err.Error(), "does not match") +} + +func TestValidateLineProtocol_RejectsForgedPeerIDOnSecondLine(t *testing.T) { + body := []byte( + "netbird_sync,deployment_type=cloud,version=1.0.0,os=linux,arch=amd64,peer_id=abcdef0123456789 duration_seconds=1.5 1234567890\n" + + "netbird_sync,deployment_type=cloud,version=1.0.0,os=linux,arch=amd64,peer_id=deadbeefdeadbeef duration_seconds=2.0 1234567890\n", + ) + _, err := validateLineProtocol(body, "abcdef0123456789") + require.Error(t, err) + assert.Contains(t, err.Error(), "does not match") +}