feat(sinks): enable sinks with token auth.

maestro/config/authentication_builder.go

@@ -4,6 +4,7 @@ import (
 	"github.com/orb-community/orb/maestro/password"
 	"github.com/orb-community/orb/pkg/types"
 	"github.com/orb-community/orb/sinks/authentication_type/basicauth"
+	"github.com/orb-community/orb/sinks/authentication_type/bearertokenauth"
 )
 
 const AuthenticationKey = "authentication"
@@ -20,7 +21,12 @@ func GetAuthService(authType string, service password.EncryptionService) AuthBui
 		return &BasicAuthBuilder{
 			encryptionService: service,
 		}
+	case bearertokenauth.AuthType:
+		return &BearerTokenAuthBuilder{
+			encryptionService: service,
+		}
 	}
+
 	return nil
 }
 
@@ -65,3 +71,50 @@ func (b *BasicAuthBuilder) EncodeAuth(config types.Metadata) (types.Metadata, er
 	config[AuthenticationKey] = authcfg
 	return config, nil
 }
+
+type BearerTokenAuthBuilder struct {
+	encryptionService password.EncryptionService
+}
+
+func (b *BearerTokenAuthBuilder) GetExtensionsFromMetadata(c types.Metadata) (Extensions, string) {
+	authcfg := c.GetSubMetadata(AuthenticationKey)
+	scheme := authcfg["scheme"].(string)
+	token := authcfg["token"].(string)
+
+	return Extensions{
+		BearerAuth: &BearerTokenAuthExtension{
+			Scheme: scheme,
+			Token:  token,
+		},
+	}, "bearertokenauth/withscheme"
+}
+
+func (b *BearerTokenAuthBuilder) DecodeAuth(config types.Metadata) (types.Metadata, error) {
+	authCfg := config.GetSubMetadata(AuthenticationKey)
+	token := authCfg["token"].(string)
+
+	decodedToken, err := b.encryptionService.DecodePassword(token)
+	if err != nil {
+		return nil, err
+	}
+
+	authCfg["token"] = decodedToken
+	config[AuthenticationKey] = authCfg
+
+	return config, nil
+}
+
+func (b *BearerTokenAuthBuilder) EncodeAuth(config types.Metadata) (types.Metadata, error) {
+	authcfg := config.GetSubMetadata(AuthenticationKey)
+	token := authcfg["token"].(string)
+
+	encodedToken, err := b.encryptionService.EncodePassword(token)
+	if err != nil {
+		return nil, err
+	}
+
+	authcfg["token"] = encodedToken
+	config[AuthenticationKey] = authcfg
+
+	return config, nil
+}

maestro/config/config_builder_test.go

@@ -3,10 +3,12 @@ package config
 import (
 	"context"
 	"fmt"
+	"testing"
+
+	"go.uber.org/zap"
+
 	"github.com/orb-community/orb/maestro/password"
 	"github.com/orb-community/orb/pkg/types"
-	"go.uber.org/zap"
-	"testing"
 )
 
 func TestReturnConfigYamlFromSink(t *testing.T) {
@@ -97,6 +99,30 @@ func TestReturnConfigYamlFromSink(t *testing.T) {
 			want:    `---\nreceivers:\n  kafka:\n    brokers:\n    - kafka:9092\n    topic: otlp_metrics-sink-id-22\n    protocol_version: 2.0.0\nextensions:\n  pprof:\n    endpoint: 0.0.0.0:1888\n  basicauth/exporter:\n    client_auth:\n      username: otlp-user\n      password: dbpass\nexporters:\n  otlphttp:\n    endpoint: https://acme.com/otlphttp/push\n    auth:\n      authenticator: basicauth/exporter\nservice:\n  extensions:\n  - pprof\n  - basicauth/exporter\n  pipelines:\n    metrics:\n      receivers:\n      - kafka\n      exporters:\n      - otlphttp\n`,
 			wantErr: false,
 		},
+		{
+			name: "otlp, token auth",
+			args: args{
+				in0:            context.Background(),
+				kafkaUrlConfig: "kafka:9092",
+				sink: &DeploymentRequest{
+					SinkID:  "sink-id-22",
+					OwnerID: "22",
+					Backend: "otlphttp",
+					Config: types.Metadata{
+						"exporter": types.Metadata{
+							"endpoint": "https://acme.com/otlphttp/push",
+						},
+						"authentication": types.Metadata{
+							"type":   "bearertokenauth",
+							"scheme": "Api-Token",
+							"token":  "abcdefg",
+						},
+					},
+				},
+			},
+			want:    `---\nreceivers:\n  kafka:\n    brokers:\n    - kafka:9092\n    topic: otlp_metrics-sink-id-22\n    protocol_version: 2.0.0\nextensions:\n  pprof:\n    endpoint: 0.0.0.0:1888\n  bearertokenauth/withscheme:\n    scheme: Api-Token\n    token: abcdefg\nexporters:\n  otlphttp:\n    endpoint: https://acme.com/otlphttp/push\n    auth:\n      authenticator: bearertokenauth/withscheme\nservice:\n  extensions:\n  - pprof\n  - bearertokenauth/withscheme\n  pipelines:\n    metrics:\n      receivers:\n      - kafka\n      exporters:\n      - otlphttp\n`,
+			wantErr: false,
+		},
 	}
 	for _, tt := range tests {
 		logger := zap.NewNop()

maestro/config/types.go

@@ -2,8 +2,9 @@ package config
 
 import (
 	"database/sql/driver"
-	"github.com/orb-community/orb/pkg/types"
 	"time"
+
+	"github.com/orb-community/orb/pkg/types"
 )
 
 type SinkData struct {
@@ -82,8 +83,8 @@ type Extensions struct {
 	PProf                *PProfExtension       `json:"pprof,omitempty" yaml:"pprof,omitempty" :"p_prof"`
 	ZPages               *ZPagesExtension      `json:"zpages,omitempty" yaml:"zpages,omitempty" :"z_pages"`
 	// Exporters Authentication
-	BasicAuth *BasicAuthenticationExtension `json:"basicauth/exporter,omitempty" yaml:"basicauth/exporter,omitempty" :"basic_auth"`
-	//BearerAuth *BearerAuthExtension          `json:"bearerauth/exporter,omitempty" yaml:"bearerauth/exporter,omitempty" :"bearer_auth"`
+	BasicAuth  *BasicAuthenticationExtension `json:"basicauth/exporter,omitempty" yaml:"basicauth/exporter,omitempty" :"basic_auth"`
+	BearerAuth *BearerTokenAuthExtension     `json:"bearertokenauth/withscheme,omitempty" yaml:"bearertokenauth/withscheme,omitempty"`
 }
 
 type HealthCheckExtension struct {
@@ -115,10 +116,9 @@ type BasicAuthenticationExtension struct {
 	ClientAuth *ClientAuth `json:"client_auth" yaml:"client_auth"`
 }
 
-type BearerAuthExtension struct {
-	BearerAuth *struct {
-		Token string `json:"token" yaml:"token"`
-	} `json:"client_auth" yaml:"client_auth"`
+type BearerTokenAuthExtension struct {
+	Scheme string `json:"scheme" yaml:"scheme"`
+	Token  string `json:"token" yaml:"token"`
 }
 
 type Exporters struct {

pkg/errors/types.go

@@ -41,13 +41,19 @@ var (
 
 	// ErrAuthTypeNotFound indicates that authentication type field was not found on the authentication field
 	ErrAuthTypeNotFound = New("malformed entity specification: authentication type field is expected on configuration field")
-	
+
 	// ErrInvalidAuthType indicates invalid authentication type
 	ErrInvalidAuthType = New("malformed entity specification. type key on authentication field is invalid")
 
 	// ErrPasswordNotFound indicates that password key was not found
 	ErrPasswordNotFound = New("malformed entity specification. password key is expected on authentication field")
 
+	// ErrSchemeNotFound indicates that token key was not found
+	ErrSchemeNotFound = New("malformed entity specification. scheme key is expected on authentication field")
+
+	// ErrTokendNotFound indicates that token key was not found
+	ErrTokenNotFound = New("malformed entity specification. token key is expected on authentication field")
+
 	// ErrEndPointNotFound indicates that endpoint field was not found on exporter field for otlp backend
 	ErrEndpointNotFound = New("malformed entity specification. endpoint field is expected on exporter field")
 
@@ -57,6 +63,12 @@ var (
 	// ErrInvalidPasswordType indicates invalid password key on authentication field
 	ErrInvalidPasswordType = New("malformed entity specification. password key on authentication field is invalid")
 
+	// ErrInvalidSchemeType indicates invalid scheme key on authentication field
+	ErrInvalidSchemeType = New("malformed entity specification. scheme key on authentication field is invalid")
+
+	// ErrInvalidTokenType indicates invalid token key on authentication field
+	ErrInvalidTokenType = New("malformed entity specification. token key on authentication field is invalid")
+
 	// ErrInvalidUsernameType indicates invalid username key on authentication field
 	ErrInvalidUsernameType = New("malformed entity specification. username key on authentication field is invalid")
 

sinks/api/http/transport.go

@@ -236,7 +236,6 @@ func encodeError(_ context.Context, err error, w http.ResponseWriter) {
 		case errors.Contains(errorVal, errors.ErrUnsupportedContentType):
 			w.WriteHeader(http.StatusUnsupportedMediaType)
 
-
 		case errors.Contains(errorVal, errors.ErrInvalidEndpoint):
 			w.WriteHeader(http.StatusBadRequest)
 		case errors.Contains(errorVal, errors.ErrEndpointNotFound):
@@ -251,8 +250,16 @@ func encodeError(_ context.Context, err error, w http.ResponseWriter) {
 			w.WriteHeader(http.StatusBadRequest)
 		case errors.Contains(errorVal, errors.ErrInvalidPasswordType):
 			w.WriteHeader(http.StatusBadRequest)
+		case errors.Contains(errorVal, errors.ErrInvalidTokenType):
+			w.WriteHeader(http.StatusBadRequest)
+		case errors.Contains(errorVal, errors.ErrTokenNotFound):
+			w.WriteHeader(http.StatusBadRequest)
+		case errors.Contains(errorVal, errors.ErrInvalidSchemeType):
+			w.WriteHeader(http.StatusBadRequest)
+		case errors.Contains(errorVal, errors.ErrSchemeNotFound):
+			w.WriteHeader(http.StatusBadRequest)
 		case errors.Contains(errorVal, errors.ErrInvalidAuthType):
-			w.WriteHeader(http.StatusBadRequest)	
+			w.WriteHeader(http.StatusBadRequest)
 		case errors.Contains(errorVal, errors.ErrRemoteHostNotFound):
 			w.WriteHeader(http.StatusBadRequest)
 		case errors.Contains(errorVal, errors.ErrAuthFieldNotFound):
@@ -262,7 +269,7 @@ func encodeError(_ context.Context, err error, w http.ResponseWriter) {
 		case errors.Contains(errorVal, errors.ErrExporterFieldNotFound):
 			w.WriteHeader(http.StatusBadRequest)
 		case errors.Contains(errorVal, errors.ErrInvalidBackend):
-			w.WriteHeader(http.StatusBadRequest)	
+			w.WriteHeader(http.StatusBadRequest)
 		case errors.Contains(errorVal, errors.ErrEntityNameNotFound):
 			w.WriteHeader(http.StatusBadRequest)
 		case errors.Contains(errorVal, errors.ErrMalformedEntity):