Refactor configauth to support multiple extension auth interfaces

Signed-off-by: Bogdan Drutu <[email protected]>

Author: bogdandrutu

config/configauth/configauth.go

@@ -10,6 +10,14 @@ import (
 	"context"
 	"errors"
 	"fmt"
+	"net/http"
+
+	"google.golang.org/grpc"
+	"google.golang.org/grpc/codes"
+	"google.golang.org/grpc/metadata"
+	"google.golang.org/grpc/status"
+
+	"go.opentelemetry.io/collector/internal/grpcutil"
 
 	"go.opentelemetry.io/collector/component"
 	"go.opentelemetry.io/collector/extension/extensionauth"
@@ -20,6 +28,7 @@ var (
 	errNotHTTPClient         = errors.New("requested authenticator is not a HTTP client authenticator")
 	errNotGRPCClient         = errors.New("requested authenticator is not a gRPC client authenticator")
 	errNotServer             = errors.New("requested authenticator is not a server authenticator")
+	errMetadataNotFound      = errors.New("no request metadata found")
 )
 
 // Authentication defines the auth settings for the receiver.
@@ -28,8 +37,48 @@ type Authentication struct {
 	AuthenticatorID component.ID `mapstructure:"authenticator,omitempty"`
 }
 
-// GetServerAuthenticator attempts to select the appropriate extensionauth.Server from the list of extensions,
-// based on the requested extension name. If an authenticator is not found, an error is returned.
+// GetGRPCServerOptions attempts to select the appropriate extensionauth.Server from the list of extensions,
+// based on the requested extension name and return the grpc.ServerOption to be used with the grpc.Server.
+// If an authenticator is not found, an error is returned.
+func (a Authentication) GetGRPCServerOptions(_ context.Context, extensions map[component.ID]component.Component) ([]grpc.ServerOption, error) {
+	ext, found := extensions[a.AuthenticatorID]
+	if !found {
+		return nil, fmt.Errorf("failed to resolve authenticator %q: %w", a.AuthenticatorID, errAuthenticatorNotFound)
+	}
+
+	eauth, ok := ext.(extensionauth.Server)
+	if !ok {
+		return nil, errNotServer
+	}
+
+	uInterceptor := func(ctx context.Context, req any, info *grpc.UnaryServerInfo, handler grpc.UnaryHandler) (resp any, err error) {
+		return authServerUnaryInterceptor(ctx, req, info, handler, eauth)
+	}
+	sInterceptors := func(srv any, ss grpc.ServerStream, info *grpc.StreamServerInfo, handler grpc.StreamHandler) error {
+		return authServerStreamInterceptor(srv, ss, info, handler, eauth)
+	}
+
+	return []grpc.ServerOption{grpc.ChainUnaryInterceptor(uInterceptor), grpc.ChainStreamInterceptor(sInterceptors)}, nil
+}
+
+// GetHTTPHandler attempts to select the appropriate extensionauth.Server from the list of extensions,
+// based on the requested extension name and return the http.Handler to be used with the http.Server.
+// If an authenticator is not found, an error is returned.
+func (a Authentication) GetHTTPHandler(_ context.Context, extensions map[component.ID]component.Component, next http.Handler, reqParams []string) (http.Handler, error) {
+	ext, found := extensions[a.AuthenticatorID]
+	if !found {
+		return nil, fmt.Errorf("failed to resolve authenticator %q: %w", a.AuthenticatorID, errAuthenticatorNotFound)
+	}
+
+	eauth, ok := ext.(extensionauth.Server)
+	if !ok {
+		return nil, errNotServer
+	}
+
+	return authInterceptor(next, eauth, reqParams), nil
+}
+
+// Deprecated: [v0.123.0] use GetGRPCServerOptions or GetHTTPServer.
 func (a Authentication) GetServerAuthenticator(_ context.Context, extensions map[component.ID]component.Component) (extensionauth.Server, error) {
 	if ext, found := extensions[a.AuthenticatorID]; found {
 		if server, ok := ext.(extensionauth.Server); ok {
@@ -41,9 +90,7 @@ func (a Authentication) GetServerAuthenticator(_ context.Context, extensions map
 	return nil, fmt.Errorf("failed to resolve authenticator %q: %w", a.AuthenticatorID, errAuthenticatorNotFound)
 }
 
-// GetHTTPClientAuthenticator attempts to select the appropriate extensionauth.Client from the list of extensions,
-// based on the component id of the extension. If an authenticator is not found, an error is returned.
-// This should be only used by HTTP clients.
+// Deprecated: [v0.123.0] use GetHTTPRoundTripper.
 func (a Authentication) GetHTTPClientAuthenticator(_ context.Context, extensions map[component.ID]component.Component) (extensionauth.HTTPClient, error) {
 	if ext, found := extensions[a.AuthenticatorID]; found {
 		if client, ok := ext.(extensionauth.HTTPClient); ok {
@@ -54,9 +101,25 @@ func (a Authentication) GetHTTPClientAuthenticator(_ context.Context, extensions
 	return nil, fmt.Errorf("failed to resolve authenticator %q: %w", a.AuthenticatorID, errAuthenticatorNotFound)
 }
 
-// GetGRPCClientAuthenticator attempts to select the appropriate extensionauth.Client from the list of extensions,
-// based on the component id of the extension. If an authenticator is not found, an error is returned.
-// This should be only used by gRPC clients.
+// GetHTTPRoundTripper attempts to select the appropriate extensionauth.Client from the list of extensions,
+// based on the component id of the extension and return the http.RoundTripper to be used with the http.Client.
+// If an authenticator is not found, an error is returned. This should be only used by HTTP clients.
+func (a Authentication) GetHTTPRoundTripper(_ context.Context, extensions map[component.ID]component.Component, base http.RoundTripper) (http.RoundTripper, error) {
+	ext, found := extensions[a.AuthenticatorID]
+	if !found {
+		return nil, fmt.Errorf("failed to resolve authenticator %q: %w", a.AuthenticatorID, errAuthenticatorNotFound)
+	}
+
+	// Currently only support `extensionauth.HTTPClient`.
+	client, ok := ext.(extensionauth.HTTPClient)
+	if !ok {
+		return nil, errNotHTTPClient
+	}
+
+	return client.RoundTripper(base)
+}
+
+// Deprecated: [v0.123.0] Use GetGRPCDialOptions.
 func (a Authentication) GetGRPCClientAuthenticator(_ context.Context, extensions map[component.ID]component.Component) (extensionauth.GRPCClient, error) {
 	if ext, found := extensions[a.AuthenticatorID]; found {
 		if client, ok := ext.(extensionauth.GRPCClient); ok {
@@ -66,3 +129,74 @@ func (a Authentication) GetGRPCClientAuthenticator(_ context.Context, extensions
 	}
 	return nil, fmt.Errorf("failed to resolve authenticator %q: %w", a.AuthenticatorID, errAuthenticatorNotFound)
 }
+
+// GetGRPCDialOptions attempts to select the appropriate extensionauth.Client from the list of extensions,
+// based on the component id of the extension and return the grpc.DialOptions to be used with grpc.ClientConn.
+// If an authenticator is not found, an error is returned. This should be only used by gRPC clients.
+func (a Authentication) GetGRPCDialOptions(_ context.Context, extensions map[component.ID]component.Component) ([]grpc.DialOption, error) {
+	ext, found := extensions[a.AuthenticatorID]
+	if !found {
+		return nil, fmt.Errorf("failed to resolve authenticator %q: %w", a.AuthenticatorID, errAuthenticatorNotFound)
+	}
+
+	// Currently only support `extensionauth.GRPCClient`.
+	client, ok := ext.(extensionauth.GRPCClient)
+	if !ok {
+		return nil, errNotGRPCClient
+	}
+
+	perRPCCredentials, err := client.PerRPCCredentials()
+	if err != nil {
+		return nil, err
+	}
+
+	return []grpc.DialOption{grpc.WithPerRPCCredentials(perRPCCredentials)}, nil
+}
+
+func authServerUnaryInterceptor(ctx context.Context, req any, _ *grpc.UnaryServerInfo, handler grpc.UnaryHandler, eauth extensionauth.Server) (any, error) {
+	headers, ok := metadata.FromIncomingContext(ctx)
+	if !ok {
+		return nil, errMetadataNotFound
+	}
+
+	ctx, err := eauth.Authenticate(ctx, headers)
+	if err != nil {
+		return nil, status.Error(codes.Unauthenticated, err.Error())
+	}
+
+	return handler(ctx, req)
+}
+
+func authServerStreamInterceptor(srv any, stream grpc.ServerStream, _ *grpc.StreamServerInfo, handler grpc.StreamHandler, eauth extensionauth.Server) error {
+	ctx := stream.Context()
+	headers, ok := metadata.FromIncomingContext(ctx)
+	if !ok {
+		return errMetadataNotFound
+	}
+
+	ctx, err := eauth.Authenticate(ctx, headers)
+	if err != nil {
+		return status.Error(codes.Unauthenticated, err.Error())
+	}
+
+	return handler(srv, grpcutil.WrapServerStream(ctx, stream))
+}
+
+func authInterceptor(next http.Handler, eauth extensionauth.Server, requestParams []string) http.Handler {
+	return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		sources := r.Header
+		query := r.URL.Query()
+		for _, param := range requestParams {
+			if val, ok := query[param]; ok {
+				sources[param] = val
+			}
+		}
+		ctx, err := eauth.Authenticate(r.Context(), sources)
+		if err != nil {
+			http.Error(w, http.StatusText(http.StatusUnauthorized), http.StatusUnauthorized)
+			return
+		}
+
+		next.ServeHTTP(w, r.WithContext(ctx))
+	})
+}