Beholder CSA Authentication

pkg/beholder/auth.go

@@ -0,0 +1,23 @@
+package beholder
+
+import "fmt"
+
+// authHeaderKey is the name of the header that the node authenticator will use to send the auth token
+var authHeaderKey = "X-Beholder-Node-Auth-Token"
+
+// authHeaderVersion is the version of the auth header format
+var authHeaderVersion = "1"
+
+// BuildAuthHeaders creates the auth header value to be included on requests.
+// The current format for the header is:
+//
+// <version>:<public_key_hex>:<signature_hex>
+//
+// where the byte value of <public_key_hex> is what's being signed
+func BuildAuthHeaders(signer func([]byte) []byte, pubKey []byte) map[string]string {
+	messageBytes := pubKey
+	signature := signer(messageBytes)
+	headerValue := fmt.Sprintf("%s:%x:%x", authHeaderVersion, messageBytes, signature)
+
+	return map[string]string{authHeaderKey: headerValue}
+}

pkg/beholder/auth_test.go

@@ -0,0 +1,20 @@
+package beholder
+
+import (
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+)
+
+func TestBuildAuthHeaders(t *testing.T) {
+	mockPubKey := []byte("test-public-key")
+	mockSigner := func(data []byte) []byte {
+		return append(data, []byte("__test-signature")...)
+	}
+
+	expectedHeaders := map[string]string{
+		"X-Beholder-Node-Auth-Token": "1:746573742d7075626c69632d6b6579:746573742d7075626c69632d6b65795f5f746573742d7369676e6174757265",
+	}
+
+	assert.Equal(t, expectedHeaders, BuildAuthHeaders(mockSigner, mockPubKey))
+}

pkg/beholder/client.go

@@ -93,6 +93,7 @@ func newGRPCClient(cfg Config, otlploggrpcNew otlploggrpcFactory) (*Client, erro
 	opts := []otlploggrpc.Option{
 		otlploggrpc.WithTLSCredentials(creds),
 		otlploggrpc.WithEndpoint(cfg.OtelExporterGRPCEndpoint),
+		otlploggrpc.WithHeaders(cfg.AuthHeaders),
 	}
 	if cfg.LogRetryConfig != nil {
 		// NOTE: By default, the retry is enabled in the OTel SDK
@@ -242,6 +243,21 @@ func newOtelResource(cfg Config) (resource *sdkresource.Resource, err error) {
 	if err != nil {
 		return nil, err
 	}
+
+	// Add csa public key resource attribute
+	csaPublicKeyHex := "not-configured"
+	if len(cfg.AuthPublicKeyHex) > 0 {
+		csaPublicKeyHex = cfg.AuthPublicKeyHex
+	}
+	csaPublicKeyAttr := attribute.String("csa_public_key", csaPublicKeyHex)
+	resource, err = sdkresource.Merge(
+		sdkresource.NewSchemaless(csaPublicKeyAttr),
+		resource,
+	)
+	if err != nil {
+		return nil, err
+	}
+
 	// Add custom resource attributes
 	resource, err = sdkresource.Merge(
 		sdkresource.NewSchemaless(cfg.ResourceAttributes...),
@@ -282,6 +298,7 @@ func newTracerProvider(config Config, resource *sdkresource.Resource, creds cred
 	exporterOpts := []otlptracegrpc.Option{
 		otlptracegrpc.WithTLSCredentials(creds),
 		otlptracegrpc.WithEndpoint(config.OtelExporterGRPCEndpoint),
+		otlptracegrpc.WithHeaders(config.AuthHeaders),
 	}
 	if config.TraceRetryConfig != nil {
 		// NOTE: By default, the retry is enabled in the OTel SDK
@@ -318,6 +335,7 @@ func newMeterProvider(config Config, resource *sdkresource.Resource, creds crede
 	opts := []otlpmetricgrpc.Option{
 		otlpmetricgrpc.WithTLSCredentials(creds),
 		otlpmetricgrpc.WithEndpoint(config.OtelExporterGRPCEndpoint),
+		otlpmetricgrpc.WithHeaders(config.AuthHeaders),
 	}
 	if config.MetricRetryConfig != nil {
 		// NOTE: By default, the retry is enabled in the OTel SDK

pkg/beholder/config.go

@@ -35,6 +35,10 @@ type Config struct {
 	LogBatchProcessor bool
 	// Retry config for shared log exporter, used by Emitter and Logger
 	LogRetryConfig *RetryConfig
+
+	// Auth
+	AuthPublicKeyHex string
+	AuthHeaders      map[string]string
 }
 
 type RetryConfig struct {

pkg/beholder/config_test.go

@@ -50,6 +50,6 @@ func ExampleConfig() {
 	}
 	fmt.Printf("%+v\n", *config.LogRetryConfig)
 	// Output:
-	// {InsecureConnection:true CACertFile: OtelExporterGRPCEndpoint:localhost:4317 OtelExporterHTTPEndpoint:localhost:4318 ResourceAttributes:[{Key:package_name Value:{vtype:4 numeric:0 stringly:beholder slice:<nil>}} {Key:sender Value:{vtype:4 numeric:0 stringly:beholderclient slice:<nil>}}] EmitterExportTimeout:1s EmitterBatchProcessor:true TraceSampleRatio:1 TraceBatchTimeout:1s TraceSpanExporter:<nil> TraceRetryConfig:<nil> MetricReaderInterval:1s MetricRetryConfig:<nil> LogExportTimeout:1s LogBatchProcessor:true LogRetryConfig:<nil>}
+	// {InsecureConnection:true CACertFile: OtelExporterGRPCEndpoint:localhost:4317 OtelExporterHTTPEndpoint:localhost:4318 ResourceAttributes:[{Key:package_name Value:{vtype:4 numeric:0 stringly:beholder slice:<nil>}} {Key:sender Value:{vtype:4 numeric:0 stringly:beholderclient slice:<nil>}}] EmitterExportTimeout:1s EmitterBatchProcessor:true TraceSampleRatio:1 TraceBatchTimeout:1s TraceSpanExporter:<nil> TraceRetryConfig:<nil> MetricReaderInterval:1s MetricRetryConfig:<nil> LogExportTimeout:1s LogBatchProcessor:true LogRetryConfig:<nil> AuthPublicKeyHex: AuthHeaders:map[]}
 	// {InitialInterval:5s MaxInterval:30s MaxElapsedTime:1m0s}
 }

pkg/beholder/httpclient.go

@@ -58,6 +58,7 @@ func newHTTPClient(cfg Config, otlploghttpNew otlploghttpFactory) (*Client, erro
 	opts := []otlploghttp.Option{
 		tlsConfigOption,
 		otlploghttp.WithEndpoint(cfg.OtelExporterHTTPEndpoint),
+		otlploghttp.WithHeaders(cfg.AuthHeaders),
 	}
 	if cfg.LogRetryConfig != nil {
 		// NOTE: By default, the retry is enabled in the OTel SDK
@@ -164,6 +165,7 @@ func newHTTPTracerProvider(config Config, resource *sdkresource.Resource, tlsCon
 	exporterOpts := []otlptracehttp.Option{
 		tlsConfigOption,
 		otlptracehttp.WithEndpoint(config.OtelExporterHTTPEndpoint),
+		otlptracehttp.WithHeaders(config.AuthHeaders),
 	}
 	if config.TraceRetryConfig != nil {
 		// NOTE: By default, the retry is enabled in the OTel SDK
@@ -205,6 +207,7 @@ func newHTTPMeterProvider(config Config, resource *sdkresource.Resource, tlsConf
 	opts := []otlpmetrichttp.Option{
 		tlsConfigOption,
 		otlpmetrichttp.WithEndpoint(config.OtelExporterHTTPEndpoint),
+		otlpmetrichttp.WithHeaders(config.AuthHeaders),
 	}
 	if config.MetricRetryConfig != nil {
 		// NOTE: By default, the retry is enabled in the OTel SDK

pkg/loop/config.go

@@ -26,6 +26,8 @@ const (
 	envTelemetryCACertFile       = "CL_TELEMETRY_CA_CERT_FILE"
 	envTelemetryAttribute        = "CL_TELEMETRY_ATTRIBUTE_"
 	envTelemetryTraceSampleRatio = "CL_TELEMETRY_TRACE_SAMPLE_RATIO"
+	envTelemetryAuthHeader       = "CL_TELEMETRY_AUTH_HEADER"
+	envTelemetryAuthPubKeyHex    = "CL_TELEMETRY_AUTH_PUB_KEY_HEX"
 )
 
 // EnvConfig is the configuration between the application and the LOOP executable. The values
@@ -47,6 +49,8 @@ type EnvConfig struct {
 	TelemetryCACertFile         string
 	TelemetryAttributes         OtelAttributes
 	TelemetryTraceSampleRatio   float64
+	TelemetryAuthHeaders        map[string]string
+	TelemetryAuthPubKeyHex      string
 }
 
 // AsCmdEnv returns a slice of environment variable key/value pairs for an exec.Cmd.
@@ -78,6 +82,11 @@ func (e *EnvConfig) AsCmdEnv() (env []string) {
 		add(envTelemetryAttribute+k, v)
 	}
 
+	for k, v := range e.TelemetryAuthHeaders {
+		add(envTelemetryAuthHeader+k, v)
+	}
+	add(envTelemetryAuthPubKeyHex, e.TelemetryAuthPubKeyHex)
+
 	return
 }
 
@@ -124,6 +133,8 @@ func (e *EnvConfig) parse() error {
 		e.TelemetryCACertFile = os.Getenv(envTelemetryCACertFile)
 		e.TelemetryAttributes = getMap(envTelemetryAttribute)
 		e.TelemetryTraceSampleRatio = getFloat64OrZero(envTelemetryTraceSampleRatio)
+		e.TelemetryAuthHeaders = getMap(envTelemetryAuthHeader)
+		e.TelemetryAuthPubKeyHex = os.Getenv(envTelemetryAuthPubKeyHex)
 	}
 	return nil
 }

pkg/loop/config_test.go

@@ -128,6 +128,8 @@ func TestEnvConfig_AsCmdEnv(t *testing.T) {
 		TelemetryCACertFile:         "foo/bar",
 		TelemetryAttributes:         OtelAttributes{"foo": "bar", "baz": "42"},
 		TelemetryTraceSampleRatio:   0.42,
+		TelemetryAuthHeaders:        map[string]string{"header-key": "header-value"},
+		TelemetryAuthPubKeyHex:      "pub-key-hex",
 	}
 	got := map[string]string{}
 	for _, kv := range envCfg.AsCmdEnv() {
@@ -152,6 +154,8 @@ func TestEnvConfig_AsCmdEnv(t *testing.T) {
 	assert.Equal(t, "0.42", got[envTelemetryTraceSampleRatio])
 	assert.Equal(t, "bar", got[envTelemetryAttribute+"foo"])
 	assert.Equal(t, "42", got[envTelemetryAttribute+"baz"])
+	assert.Equal(t, "header-value", got[envTelemetryAuthHeader+"header-key"])
+	assert.Equal(t, "pub-key-hex", got[envTelemetryAuthPubKeyHex])
 }
 
 func TestGetMap(t *testing.T) {

pkg/loop/server.go

@@ -90,12 +90,15 @@ func (s *Server) start() error {
 		if tracingConfig.Enabled {
 			attributes = tracingConfig.Attributes()
 		}
+
 		beholderCfg := beholder.Config{
 			InsecureConnection:       envCfg.TelemetryInsecureConnection,
 			CACertFile:               envCfg.TelemetryCACertFile,
 			OtelExporterGRPCEndpoint: envCfg.TelemetryEndpoint,
 			ResourceAttributes:       append(attributes, envCfg.TelemetryAttributes.AsStringAttributes()...),
 			TraceSampleRatio:         envCfg.TelemetryTraceSampleRatio,
+			AuthHeaders:              envCfg.TelemetryAuthHeaders,
+			AuthPublicKeyHex:         envCfg.TelemetryAuthPubKeyHex,
 		}
 
 		if tracingConfig.Enabled {