Introduce shared gRPC authorization interceptor (#5515) (#5517)

* Introduce shared gRPC authorization interceptor

Introduce a new gRPC authorization interceptor which
is used for both OTLP and Jaeger. This new interceptor
dispatches to method-specific handlers. All OTLP methods
use the "authorization" metadata, whereas Jaeger has
different approaches depending on the method: extract
process tags for PostSpans, and anonymous auth only for
GetSamplingStrategy.

* beater/otlp: revert metrics change

* tests/system: update Jaeger system test

The response error message has changed, update the
test to expect the new one.

* beater/jaeger: fix logger name

(cherry picked from commit 92335c5)

Co-authored-by: Andrew Wilkins <[email protected]>

beater/interceptors/authorization.go

@@ -0,0 +1,89 @@
+// Licensed to Elasticsearch B.V. under one or more contributor
+// license agreements. See the NOTICE file distributed with
+// this work for additional information regarding copyright
+// ownership. Elasticsearch B.V. licenses this file to you under
+// the Apache License, Version 2.0 (the "License"); you may
+// not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied.  See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+package interceptors
+
+import (
+	"context"
+
+	"google.golang.org/grpc"
+	"google.golang.org/grpc/codes"
+	"google.golang.org/grpc/metadata"
+	"google.golang.org/grpc/status"
+
+	"github.com/elastic/apm-server/beater/authorization"
+	"github.com/elastic/apm-server/beater/headers"
+)
+
+// MethodAuthorizationHandler is a function type for obtaining an Authorization
+// for a gRPC method call. This is used to authorize gRPC method calls by extracting
+// authentication tokens from incoming context metadata or from the request payload.
+type MethodAuthorizationHandler func(ctx context.Context, req interface{}) authorization.Authorization
+
+// Authorization returns a grpc.UnaryServerInterceptor that ensures method
+// calls are authorized before passing on to the next handler.
+//
+// Authorization is performed using a MethodAuthorizationHandler from the
+// combined map parameters, keyed on the full gRPC method name (info.FullMethod).
+// If there is no handler defined for the method, authorization fails.
+func Authorization(methodHandlers ...map[string]MethodAuthorizationHandler) grpc.UnaryServerInterceptor {
+	combinedMethodHandlers := make(map[string]MethodAuthorizationHandler)
+	for _, methodHandlers := range methodHandlers {
+		for method, handler := range methodHandlers {
+			combinedMethodHandlers[method] = handler
+		}
+	}
+	return func(
+		ctx context.Context,
+		req interface{},
+		info *grpc.UnaryServerInfo,
+		handler grpc.UnaryHandler,
+	) (interface{}, error) {
+		authHandler, ok := combinedMethodHandlers[info.FullMethod]
+		if !ok {
+			return nil, status.Errorf(codes.Unauthenticated, "no auth method defined for %q", info.FullMethod)
+		}
+		auth := authHandler(ctx, req)
+		authResult, err := auth.AuthorizedFor(ctx, authorization.Resource{})
+		if err != nil {
+			return nil, err
+		} else if !authResult.Authorized {
+			message := "unauthorized"
+			if authResult.Reason != "" {
+				message = authResult.Reason
+			}
+			return nil, status.Error(codes.Unauthenticated, message)
+		}
+		ctx = authorization.ContextWithAuthorization(ctx, auth)
+		return handler(ctx, req)
+	}
+}
+
+// MetadataMethodAuthorizationHandler returns a MethodAuthorizationHandler
+// that extracts authentication parameters from the "authorization" metadata in ctx,
+// calling authHandler.AuthorizedFor.
+func MetadataMethodAuthorizationHandler(authHandler *authorization.Handler) MethodAuthorizationHandler {
+	return func(ctx context.Context, req interface{}) authorization.Authorization {
+		var authHeader string
+		if md, ok := metadata.FromIncomingContext(ctx); ok {
+			if values := md.Get(headers.Authorization); len(values) > 0 {
+				authHeader = values[0]
+			}
+		}
+		return authHandler.AuthorizationFor(authorization.ParseAuthorizationHeader(authHeader))
+	}
+}

beater/interceptors/authorization_test.go

@@ -0,0 +1,141 @@
+// Licensed to Elasticsearch B.V. under one or more contributor
+// license agreements. See the NOTICE file distributed with
+// this work for additional information regarding copyright
+// ownership. Elasticsearch B.V. licenses this file to you under
+// the Apache License, Version 2.0 (the "License"); you may
+// not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied.  See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+package interceptors_test
+
+import (
+	"context"
+	"errors"
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+	"google.golang.org/grpc"
+	"google.golang.org/grpc/codes"
+	"google.golang.org/grpc/metadata"
+	"google.golang.org/grpc/status"
+
+	"github.com/elastic/apm-server/beater/authorization"
+	"github.com/elastic/apm-server/beater/config"
+	"github.com/elastic/apm-server/beater/interceptors"
+)
+
+func TestAuthorization(t *testing.T) {
+	type contextKey struct{}
+	origContext := context.WithValue(context.Background(), contextKey{}, 123)
+	origReq := "grpc_request"
+	origResp := "grpc_response"
+	origErr := errors.New("handler error")
+
+	authorizedResult := authorization.Result{Authorized: true}
+	anonymousResult := authorization.Result{Authorized: true, Anonymous: true}
+	unauthorizedResult := authorization.Result{Authorized: false, Reason: "no particular reason"}
+
+	authorized := authorizationFunc(func(ctx context.Context, _ authorization.Resource) (authorization.Result, error) {
+		assert.Equal(t, 123, ctx.Value(contextKey{}))
+		return authorizedResult, nil
+	})
+	anonymous := authorizationFunc(func(ctx context.Context, _ authorization.Resource) (authorization.Result, error) {
+		assert.Equal(t, 123, ctx.Value(contextKey{}))
+		return anonymousResult, nil
+	})
+	unauthorized := authorizationFunc(func(ctx context.Context, _ authorization.Resource) (authorization.Result, error) {
+		assert.Equal(t, 123, ctx.Value(contextKey{}))
+		return unauthorizedResult, nil
+	})
+	authError := authorizationFunc(func(ctx context.Context, _ authorization.Resource) (authorization.Result, error) {
+		assert.Equal(t, 123, ctx.Value(contextKey{}))
+		return authorization.Result{}, errors.New("error checking authorization")
+	})
+
+	makeMethodAuthorizationHandler := func(auth authorization.Authorization) interceptors.MethodAuthorizationHandler {
+		return func(ctx context.Context, req interface{}) authorization.Authorization {
+			require.Equal(t, origReq, req)
+			return auth
+		}
+	}
+
+	interceptor := interceptors.Authorization(
+		map[string]interceptors.MethodAuthorizationHandler{
+			"authorized": makeMethodAuthorizationHandler(authorized),
+			"anonymous":  makeMethodAuthorizationHandler(anonymous),
+		},
+		map[string]interceptors.MethodAuthorizationHandler{
+			"unauthorized": makeMethodAuthorizationHandler(unauthorized),
+			"authError":    makeMethodAuthorizationHandler(authError),
+		},
+	)
+
+	type test struct {
+		method       string
+		expectResult authorization.Result
+		expectResp   interface{}
+		expectErr    error
+	}
+	for _, test := range []test{{
+		method:       "authorized",
+		expectResult: authorizedResult,
+		expectResp:   origResp,
+		expectErr:    origErr,
+	}, {
+		method:       "anonymous",
+		expectResult: anonymousResult,
+		expectResp:   origResp,
+		expectErr:    origErr,
+	}, {
+		method:     "unauthorized",
+		expectResp: nil,
+		expectErr:  status.Error(codes.Unauthenticated, "no particular reason"),
+	}, {
+		method:     "authError",
+		expectResp: nil,
+		expectErr:  errors.New("error checking authorization"),
+	}} {
+		t.Run(test.method, func(t *testing.T) {
+			var authorizedForResult authorization.Result
+			var authorizedForErr error
+			next := func(ctx context.Context, req interface{}) (interface{}, error) {
+				authorizedForResult, authorizedForErr = authorization.AuthorizedFor(ctx, authorization.Resource{})
+				return origResp, origErr
+			}
+			resp, err := interceptor(origContext, origReq, &grpc.UnaryServerInfo{FullMethod: test.method}, next)
+			assert.Equal(t, test.expectErr, err)
+			assert.Equal(t, test.expectResp, resp)
+			assert.Equal(t, test.expectResult, authorizedForResult)
+			assert.NoError(t, authorizedForErr)
+		})
+	}
+}
+
+func TestMetadataMethodAuthorizationHandler(t *testing.T) {
+	authBuilder, _ := authorization.NewBuilder(config.AgentAuth{SecretToken: "abc123"})
+	authHandler := authBuilder.ForPrivilege(authorization.PrivilegeEventWrite.Action)
+	methodHandler := interceptors.MetadataMethodAuthorizationHandler(authHandler)
+
+	ctx := context.Background()
+	ctx = metadata.NewIncomingContext(ctx, metadata.Pairs("authorization", "Bearer abc123"))
+	auth := methodHandler(ctx, nil)
+	result, err := auth.AuthorizedFor(ctx, authorization.Resource{})
+	require.NoError(t, err)
+	assert.Equal(t, authorization.Result{Authorized: true}, result)
+}
+
+type authorizationFunc func(context.Context, authorization.Resource) (authorization.Result, error)
+
+func (f authorizationFunc) AuthorizedFor(ctx context.Context, resource authorization.Resource) (authorization.Result, error) {
+	return f(ctx, resource)
+}

beater/jaeger/common.go

@@ -21,18 +21,14 @@ import (
 	"context"
 
 	"github.com/jaegertracing/jaeger/model"
-	"github.com/pkg/errors"
 	"go.opentelemetry.io/collector/consumer"
 	trjaeger "go.opentelemetry.io/collector/translator/trace/jaeger"
 
 	"github.com/elastic/beats/v7/libbeat/monitoring"
 
-	"github.com/elastic/apm-server/beater/authorization"
 	"github.com/elastic/apm-server/beater/request"
 )
 
-var errNotAuthorized = errors.New("not authorized")
-
 type monitoringMap map[request.ResultID]*monitoring.Int
 
 func (m monitoringMap) inc(id request.ResultID) {
@@ -58,36 +54,3 @@ func consumeBatch(
 	traces := trjaeger.ProtoBatchToInternalTraces(batch)
 	return consumer.ConsumeTraces(ctx, traces)
 }
-
-type authFunc func(context.Context, model.Batch) (context.Context, error)
-
-func noAuth(ctx context.Context, _ model.Batch) (context.Context, error) {
-	return ctx, nil
-}
-
-func makeAuthFunc(authTag string, authHandler *authorization.Handler) authFunc {
-	return func(ctx context.Context, batch model.Batch) (context.Context, error) {
-		var kind, token string
-		for i, kv := range batch.Process.GetTags() {
-			if kv.Key != authTag {
-				continue
-			}
-			// Remove the auth tag.
-			batch.Process.Tags = append(batch.Process.Tags[:i], batch.Process.Tags[i+1:]...)
-			kind, token = authorization.ParseAuthorizationHeader(kv.VStr)
-			break
-		}
-		auth := authHandler.AuthorizationFor(kind, token)
-		result, err := auth.AuthorizedFor(ctx, authorization.Resource{})
-		if !result.Authorized {
-			if err != nil {
-				return nil, errors.Wrap(err, errNotAuthorized.Error())
-			}
-			// NOTE(axw) for now at least, we do not return result.Reason in the error message,
-			// as it refers to the "Authorization header" which is incorrect for Jaeger.
-			return nil, errNotAuthorized
-		}
-		ctx = authorization.ContextWithAuthorization(ctx, auth)
-		return ctx, nil
-	}
-}