sendnrw/netbird
2
0
Fork 0
mirror of https://github.com/netbirdio/netbird.git synced 2026-09-05 22:41:30 +02:00
Code Issues Packages Projects Releases Wiki Activity

[client] Reject metrics ingest bodies whose peer_id tag disagrees with the header

Browse Source 
validateTag checked tag names against the per-measurement allowlist but only
bounded the value length, so the peer_id tag was free-form text up to 64 bytes
and could differ from the X-Peer-ID header the request was accepted with.

Tie the two together: the tag value must equal the header value. Since the
header is already checked to be 16 hex characters, this transitively constrains
the tag to the same shape. Every client sends the same value in both places
(metrics.go feeds agentInfo.peerID to both push.SetPeerID and the body tags),
so well-behaved clients are unaffected.

The mismatch is rejected rather than silently overwritten: rewriting the value
would re-serialize caller-controlled text back into line protocol and would
hide misbehaving senders instead of surfacing them. Rejection also matches the
other input validation failures in the same handler, which all return 400.

This narrows the value space of the peer_id tag but does not by itself bound
InfluxDB series cardinality: a sender that puts the same arbitrary 16 hex
characters in both the header and the body still passes. Limiting that needs a
per-source rate limit in front of the service.
This commit is contained in:
Zoltán Papp
2026-08-31 17:21:02 +02:00
parent cab326000e
commit 91d4f61284
2 changed files with 44 additions and 27 deletions
Download Patch File Download Diff File Expand all files Collapse all files

23
client/internal/metrics/infra/ingest/main.go
View File

@@ -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
}

48
client/internal/metrics/infra/ingest/main_test.go
View File

@@ -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")
}