[extension/azureauth] Add support to be used in receivers (open-telemetry#39033)

<!--Ex. Fixing a bug - Describe the bug and how this fixes the issue.
Ex. Adding a feature - Explain what this achieves.-->
#### Description

For the extension to have support to be used in receivers, it needs to
implement:

- Interface `extensionauth.Server` so that it can check if requests have
valid authentication.
- Interface `azcore.TokenCredential` so the receivers can get a token if
necessary (it will be for all azure receivers).


<!-- Issue number (e.g. open-telemetry#1234) or full URL to issue, if applicable. -->
#### Link to tracking issue
Fixes open-telemetry#39012.

<!--Describe what testing was performed and which tests were added.-->
#### Testing

I have added unit tests.

<!--Describe the documentation added.-->
#### Documentation

Code comments and README file updated.

Author: constanca-m

.chloggen/azureauth-add-receiver-support.yaml

@@ -0,0 +1,27 @@
+# Use this changelog template to create an entry for release notes.
+
+# One of 'breaking', 'deprecation', 'new_component', 'enhancement', 'bug_fix'
+change_type: enhancement
+
+# The name of the component, or a single word describing the area of concern, (e.g. filelogreceiver)
+component: azureauthextension
+
+# A brief description of the change.  Surround your text with quotes ("") if it needs to start with a backtick (`).
+note: Implement extensionauth.Server and azcore.TokenCredential.
+
+# Mandatory: One or more tracking issues related to the change. You can use the PR number here if no issue exists.
+issues: [39012]
+
+# (Optional) One or more lines of additional information to render under the primary note.
+# These lines will be padded with 2 spaces and then inserted directly into the document.
+# Use pipe (|) for multiline entries.
+subtext:
+
+# If your change doesn't affect end users or the exported elements of any package,
+# you should instead start your pull request title with [chore] or use the "Skip Changelog" label.
+# Optional: The change log or logs in which this entry should be included.
+# e.g. '[user]' or '[user, api]'
+# Include 'user' if the change is relevant to end users.
+# Include 'api' if there is a change to a library API.
+# Default: '[user]'
+change_logs: []

extension/azureauthextension/README.md

@@ -10,7 +10,9 @@
 [development]: https://github.com/open-telemetry/opentelemetry-collector/blob/main/docs/component-stability.md#development
 <!-- end autogenerated section -->
 
-This extension implements both `extensionauth.Client` and `extensionauth.Server`, so it can be used in both exporters and receivers.
+This extension implements both `extensionauth.HTTPClient` and `extensionauth.Server`, so it can be used in both exporters and receivers.
+
+Additionally, the extension also implements `azcore.TokenCredential` so that Azure components can get the token by running the function `GetToken`. If the component supports HTTP client, then this should not be necessary, as the token will be placed in the authorization header.
 
 It supports 4 different types of authentication:
 - Managed identity for Azure resources

extension/azureauthextension/benchmark_test.go

@@ -0,0 +1,32 @@
+// Copyright The OpenTelemetry Authors
+// SPDX-License-Identifier: Apache-2.0
+
+package azureauthextension
+
+import (
+	"testing"
+	"time"
+
+	"github.com/Azure/azure-sdk-for-go/sdk/azcore"
+)
+
+func BenchmarkGetCurrentToken(b *testing.B) {
+	auth := &authenticator{}
+	auth.token.Store(azcore.AccessToken{Token: "test", ExpiresOn: time.Now().Add(time.Hour)})
+
+	b.ResetTimer()
+	for i := 0; i < b.N; i++ {
+		_, _ = auth.getCurrentToken()
+	}
+}
+
+func BenchmarkGetCurrentTokenParallel(b *testing.B) {
+	auth := &authenticator{}
+	auth.token.Store(azcore.AccessToken{Token: "test", ExpiresOn: time.Now().Add(time.Hour)})
+
+	b.RunParallel(func(pb *testing.PB) {
+		for pb.Next() {
+			_, _ = auth.getCurrentToken()
+		}
+	})
+}

extension/azureauthextension/config.go

@@ -24,6 +24,7 @@ var (
 	errEmptyFederatedTokenFile = errors.New(`empty "federated_token_file" field`)
 	errEmptyAuthentication     = fmt.Errorf("authentication configuration is empty, please choose one of %s", validOptions)
 	errEmptyScope              = errors.New(`the "scope" field for the token permissions is empty`)
+	errMutuallyExclusiveAuth   = errors.New(`"client_secret" and "client_certificate_path" are mutually exclusive`)
 )
 
 type Config struct {
@@ -89,6 +90,8 @@ func (cfg *ServicePrincipal) Validate() error {
 	}
 	if cfg.ClientCertificatePath == "" && cfg.ClientSecret == "" {
 		errs = append(errs, errEmptyClientCredential)
+	} else if cfg.ClientCertificatePath != "" && cfg.ClientSecret != "" {
+		errs = append(errs, errMutuallyExclusiveAuth)
 	}
 
 	if len(errs) > 0 {

extension/azureauthextension/config_test.go

@@ -61,6 +61,10 @@ func TestLoadConfig(t *testing.T) {
 				},
 			},
 		},
+		{
+			id:          component.NewIDWithName(metadata.Type, "service_principal_mutually_exclusive"),
+			expectedErr: fmt.Sprintf("%s: %s", "service_principal", errMutuallyExclusiveAuth.Error()),
+		},
 		{
 			id:          component.NewIDWithName(metadata.Type, "service_principal_empty_client_id"),
 			expectedErr: fmt.Sprintf("%s: %s", "service_principal", errEmptyClientID.Error()),

extension/azureauthextension/extension.go

@@ -5,59 +5,270 @@ package azureauthextension // import "github.com/open-telemetry/opentelemetry-co
 
 import (
 	"context"
+	"crypto"
+	"crypto/x509"
+	"errors"
+	"fmt"
 	"net/http"
+	"os"
+	"strings"
+	"sync/atomic"
+	"time"
 
 	"github.com/Azure/azure-sdk-for-go/sdk/azcore"
+	"github.com/Azure/azure-sdk-for-go/sdk/azcore/policy"
+	"github.com/Azure/azure-sdk-for-go/sdk/azidentity"
 	"go.opentelemetry.io/collector/component"
 	"go.opentelemetry.io/collector/extension"
 	"go.opentelemetry.io/collector/extension/extensionauth"
 	"go.uber.org/zap"
 )
 
+const (
+	scheme              = "Bearer"
+	authorizationHeader = "Authorization"
+)
+
+var (
+	errEmptyAuthorizationHeader      = errors.New("empty authorization header")
+	errMissingAuthorizationHeader    = errors.New("missing authorization header")
+	errUnexpectedAuthorizationFormat = errors.New(`unexpected authorization value format, expected to be of format "Bearer <token>"`)
+	errUnexpectedToken               = errors.New("received token does not match the expected one")
+	errUnavailableToken              = errors.New("azure authenticator has no access to token")
+)
+
 type authenticator struct {
+	scope      string
 	credential azcore.TokenCredential
-	logger     *zap.Logger
+
+	stopCh chan int
+	token  atomic.Value
+
+	logger *zap.Logger
 }
 
 var (
 	_ extension.Extension      = (*authenticator)(nil)
 	_ extensionauth.HTTPClient = (*authenticator)(nil)
 	_ extensionauth.Server     = (*authenticator)(nil)
+	_ azcore.TokenCredential   = (*authenticator)(nil)
 )
 
-func newAzureAuthenticator(_ *Config, logger *zap.Logger) *authenticator {
+func newAzureAuthenticator(cfg *Config, logger *zap.Logger) (*authenticator, error) {
 	var credential azcore.TokenCredential
-	// TODO
-	// if cfg.UseDefault {
-	// }
-	// if cfg.Workload != nil {
-	// }
-	// if cfg.Managed != nil {
-	// }
-	// if cfg.ServicePrincipal != nil {
-	// }
+	var err error
+	failMsg := "failed to create authenticator using"
+
+	if cfg.UseDefault {
+		if credential, err = azidentity.NewDefaultAzureCredential(nil); err != nil {
+			return nil, fmt.Errorf("%s default identity: %w", failMsg, err)
+		}
+	}
+
+	if cfg.Workload != nil {
+		if credential, err = azidentity.NewWorkloadIdentityCredential(&azidentity.WorkloadIdentityCredentialOptions{
+			ClientID:      cfg.Workload.ClientID,
+			TenantID:      cfg.Workload.TenantID,
+			TokenFilePath: cfg.Workload.FederatedTokenFile,
+		}); err != nil {
+			return nil, fmt.Errorf("%s workload identity: %w", failMsg, err)
+		}
+	}
+
+	if cfg.Managed != nil {
+		clientID := cfg.Managed.ClientID
+		var options *azidentity.ManagedIdentityCredentialOptions
+		if clientID != "" {
+			options = &azidentity.ManagedIdentityCredentialOptions{
+				ID: azidentity.ClientID(clientID),
+			}
+		}
+		if credential, err = azidentity.NewManagedIdentityCredential(options); err != nil {
+			return nil, fmt.Errorf("%s managed identity: %w", failMsg, err)
+		}
+	}
+
+	if cfg.ServicePrincipal != nil {
+		if cfg.ServicePrincipal.ClientCertificatePath != "" {
+			cert, privateKey, errParse := getCertificateAndKey(cfg.ServicePrincipal.ClientCertificatePath)
+			if errParse != nil {
+				return nil, fmt.Errorf("%s service principal with certificate: %w", failMsg, errParse)
+			}
+
+			if credential, err = azidentity.NewClientCertificateCredential(
+				cfg.ServicePrincipal.TenantID,
+				cfg.ServicePrincipal.ClientID,
+				[]*x509.Certificate{cert},
+				privateKey,
+				nil,
+			); err != nil {
+				return nil, fmt.Errorf("%s service principal with certificate: %w", failMsg, err)
+			}
+		}
+		if cfg.ServicePrincipal.ClientSecret != "" {
+			if credential, err = azidentity.NewClientSecretCredential(
+				cfg.ServicePrincipal.TenantID,
+				cfg.ServicePrincipal.ClientID,
+				cfg.ServicePrincipal.ClientSecret,
+				nil,
+			); err != nil {
+				return nil, fmt.Errorf("%s service principal with secret: %w", failMsg, err)
+			}
+		}
+	}
+
 	return &authenticator{
+		scope:      cfg.Scope,
 		credential: credential,
+		stopCh:     make(chan int, 1),
+		token:      atomic.Value{},
 		logger:     logger,
+	}, nil
+}
+
+// getCertificateAndKey from the file
+func getCertificateAndKey(filename string) (*x509.Certificate, crypto.PrivateKey, error) {
+	data, err := os.ReadFile(filename)
+	if err != nil {
+		return nil, nil, fmt.Errorf("could not read the certificate file: %w", err)
+	}
+
+	certs, privateKey, err := azidentity.ParseCertificates(data, nil)
+	if err != nil {
+		return nil, nil, fmt.Errorf("failed to parse certificates: %w", err)
 	}
+
+	return certs[0], privateKey, nil
 }
 
-func (a authenticator) Start(_ context.Context, _ component.Host) error {
-	// TODO
+func (a *authenticator) Start(ctx context.Context, _ component.Host) error {
+	go a.trackToken(ctx)
 	return nil
 }
 
-func (a authenticator) Shutdown(_ context.Context) error {
-	// TODO
+// updateToken makes a request to get a new token
+// if the authenticator does not have a token or
+// it has expired.
+func (a *authenticator) updateToken(ctx context.Context) (time.Time, error) {
+	if a.credential == nil {
+		return time.Time{}, errors.New("authenticator does not have credentials configured")
+	}
+	token, err := a.credential.GetToken(ctx, policy.TokenRequestOptions{
+		Scopes: []string{a.scope},
+	})
+	if err != nil {
+		// TODO Handle retries
+		return time.Time{}, fmt.Errorf("azure authenticator failed to get token: %w", err)
+	}
+	a.token.Store(token)
+	return token.ExpiresOn, nil
+}
+
+// trackToken runs on the background to refresh
+// the token if it expires
+func (a *authenticator) trackToken(ctx context.Context) {
+	expiresOn, err := a.updateToken(ctx)
+	if err != nil {
+		// TODO Handle retries
+		a.logger.Error("failed to update the token", zap.Error(err))
+		return
+	}
+
+	getRefresh := func(expiresOn time.Time) time.Duration {
+		// Refresh at 95% token lifetime
+		return time.Until(expiresOn) * 95 / 100
+	}
+
+	t := time.NewTicker(getRefresh(expiresOn))
+	defer t.Stop()
+
+	for {
+		select {
+		case <-ctx.Done():
+			a.logger.Info(
+				"azure authenticator no longer refreshing the token",
+				zap.String("reason", "context done"),
+			)
+			return
+		case <-a.stopCh:
+			a.logger.Info(
+				"azure authenticator no longer refreshing the token",
+				zap.String("reason", "received stop signal"),
+			)
+			close(a.stopCh)
+			return
+		case <-t.C:
+			expiresOn, err = a.updateToken(ctx)
+			if err != nil {
+				// TODO Handle retries
+				a.logger.Error("failed to update the token", zap.Error(err))
+				a.stopCh <- 1
+			} else {
+				t.Reset(getRefresh(expiresOn))
+			}
+		}
+	}
+}
+
+func (a *authenticator) Shutdown(_ context.Context) error {
+	select {
+	case a.stopCh <- 1:
+	default: // already stopped
+	}
 	return nil
 }
 
-func (a authenticator) Authenticate(ctx context.Context, _ map[string][]string) (context.Context, error) {
-	// TODO
+func (a *authenticator) getCurrentToken() (azcore.AccessToken, error) {
+	token := a.token.Load()
+	if token == nil {
+		return azcore.AccessToken{}, errUnavailableToken
+	}
+
+	return token.(azcore.AccessToken), nil
+}
+
+// GetToken returns an access token with a
+// valid token for authorization
+func (a *authenticator) GetToken(_ context.Context, _ policy.TokenRequestOptions) (azcore.AccessToken, error) {
+	return a.getCurrentToken()
+}
+
+// Authenticate adds an Authorization header
+// with the bearer token
+func (a *authenticator) Authenticate(ctx context.Context, headers map[string][]string) (context.Context, error) {
+	auth, ok := headers[authorizationHeader]
+	if !ok {
+		auth, ok = headers[strings.ToLower(authorizationHeader)]
+	}
+	if !ok {
+		return ctx, errMissingAuthorizationHeader
+	}
+	if len(auth) == 0 {
+		return ctx, errEmptyAuthorizationHeader
+	}
+
+	firstAuth := strings.Split(auth[0], " ")
+	if len(firstAuth) != 2 {
+		return ctx, errUnexpectedAuthorizationFormat
+	}
+	if firstAuth[0] != scheme {
+		return ctx, fmt.Errorf("expected %q scheme, got %q", scheme, firstAuth[0])
+	}
+
+	currentToken, err := a.getCurrentToken()
+	if err != nil {
+		return ctx, err
+	}
+
+	if firstAuth[1] != currentToken.Token {
+		return ctx, errUnexpectedToken
+	}
+
 	return ctx, nil
 }
 
-func (a authenticator) RoundTripper(_ http.RoundTripper) (http.RoundTripper, error) {
+func (a *authenticator) RoundTripper(_ http.RoundTripper) (http.RoundTripper, error) {
 	// TODO
+	// See request header: https://learn.microsoft.com/en-us/rest/api/azure/#request-header
 	return nil, nil
 }