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,22 @@ 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")
 
+	// ErrUsernameNotFound indicates that username key was not found
+	ErrUsernameNotFound = New("malformed entity specification. username key is expected on authentication field")
+
 	// ErrPasswordNotFound indicates that password key was not found
 	ErrPasswordNotFound = New("malformed entity specification. password key is expected on authentication field")
 
+	// ErrSchemeNotFound indicates that scheme 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 +66,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/endpoint_test.go

@@ -857,7 +857,7 @@ func TestAuthenticationTypesEndpoints(t *testing.T) {
 				err = json.Unmarshal(body, &authResponse)
 				require.NoError(t, err, "must not error")
 				require.NotNil(t, authResponse, "response must not be nil")
-				require.Equal(t, 1, len(authResponse.AuthenticationTypes), "must contain basicauth for now")
+				require.Equal(t, 2, len(authResponse.AuthenticationTypes), "must contain basicauth and bearertokenauth")
 			},
 		},
 		"view authentication type basicauth": {

sinks/api/http/transport.go

@@ -236,13 +236,14 @@ 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):
 			w.WriteHeader(http.StatusBadRequest)
 		case errors.Contains(errorVal, errors.ErrBackendNotFound):
 			w.WriteHeader(http.StatusBadRequest)
+		case errors.Contains(errorVal, errors.ErrUsernameNotFound):
+			w.WriteHeader(http.StatusBadRequest)
 		case errors.Contains(errorVal, errors.ErrPasswordNotFound):
 			w.WriteHeader(http.StatusBadRequest)
 		case errors.Contains(errorVal, errors.ErrAuthTypeNotFound):
@@ -251,8 +252,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 +271,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):

sinks/authentication_type/basicauth/authentication.go

@@ -1,11 +1,14 @@
 package basicauth
 
 import (
+	"strings"
+
+	"gopkg.in/yaml.v3"
+
 	"github.com/orb-community/orb/pkg/errors"
 	"github.com/orb-community/orb/pkg/types"
 	"github.com/orb-community/orb/sinks/authentication_type"
 	"github.com/orb-community/orb/sinks/backend"
-	"gopkg.in/yaml.v3"
 )
 
 const (
@@ -35,8 +38,8 @@ var (
 const AuthType = "basicauth"
 
 type AuthConfig struct {
-	Username          string `json:"username" ,yaml:"username"`
-	Password          string `json:"password" ,yaml:"password"`
+	Username          *string `json:"username" ,yaml:"username"`
+	Password          *string `json:"password" ,yaml:"password"`
 	encryptionService authentication_type.PasswordService
 }
 
@@ -56,27 +59,32 @@ func (a *AuthConfig) GetFeatureConfig() []authentication_type.ConfigFeature {
 func (a *AuthConfig) ValidateConfiguration(inputFormat string, input interface{}) error {
 	switch inputFormat {
 	case "object":
+		if _, ok := input.(types.Metadata)[UsernameConfigFeature]; !ok {
+			return errors.Wrap(errors.ErrUsernameNotFound, errors.New("username field was not found"))
+		}
+
+		if _, ok := input.(types.Metadata)[PasswordConfigFeature]; !ok {
+			return errors.Wrap(errors.ErrPasswordNotFound, errors.New("password field was not found"))
+		}
+
 		for key, value := range input.(types.Metadata) {
-			if _, ok := value.(string); !ok {
-				if key == "password" {
-					return errors.Wrap(errors.ErrInvalidPasswordType, errors.New("invalid auth type for field: " + key))
-				}
-				if key == "type" {
-					return errors.Wrap(errors.ErrInvalidAuthType, errors.New("invalid auth type for field: " + key))
-				}
-				if key == "username" {
-					return errors.Wrap(errors.ErrInvalidUsernameType, errors.New("invalid auth type for field: " + key))
-				}
-			}
-			vs := value.(string)
 			if key == UsernameConfigFeature {
-				if len(vs) == 0 {
-					return errors.New("username cannot be empty")
+				if _, ok := value.(string); !ok {
+					return errors.Wrap(errors.ErrInvalidUsernameType, errors.New("invalid auth type for field: "+key))
+				}
+
+				if len(strings.Fields(value.(string))) == 0 {
+					return errors.Wrap(errors.ErrInvalidUsernameType, errors.New("invalid authentication username"))
 				}
 			}
+
 			if key == PasswordConfigFeature {
-				if len(vs) == 0 {
-					return errors.New("password cannot be empty")
+				if _, ok := value.(string); !ok {
+					return errors.Wrap(errors.ErrInvalidPasswordType, errors.New("invalid auth type for field: "+key))
+				}
+
+				if len(strings.Fields(value.(string))) == 0 {
+					return errors.Wrap(errors.ErrInvalidPasswordType, errors.New("invalid authentication password"))
 				}
 			}
 		}
@@ -85,12 +93,24 @@ func (a *AuthConfig) ValidateConfiguration(inputFormat string, input interface{}
 		if err != nil {
 			return err
 		}
-		if len(a.Username) == 0 {
-			return errors.New("username cannot be empty")
-		} else if len(a.Password) == 0 {
-			return errors.New("password cannot be empty")
+
+		if a.Username == nil {
+			return errors.Wrap(errors.ErrUsernameNotFound, errors.New("username field was not found"))
+		}
+
+		if len(strings.Fields(*a.Username)) == 0 {
+			return errors.Wrap(errors.ErrInvalidUsernameType, errors.New("invalid authentication username"))
+		}
+
+		if a.Password == nil {
+			return errors.Wrap(errors.ErrPasswordNotFound, errors.New("password field was not found"))
+		}
+
+		if len(strings.Fields(*a.Password)) == 0 {
+			return errors.Wrap(errors.ErrInvalidPasswordType, errors.New("invalid authentication password"))
 		}
 	}
+
 	return nil
 }