internal/appsec: implement WAF actions for http/grpc

contrib/google.golang.org/grpc/appsec.go

@@ -12,27 +12,34 @@ import (
 	"gopkg.in/DataDog/dd-trace-go.v1/ddtrace"
 	"gopkg.in/DataDog/dd-trace-go.v1/internal/appsec/dyngo/instrumentation"
 	"gopkg.in/DataDog/dd-trace-go.v1/internal/appsec/dyngo/instrumentation/grpcsec"
+	"gopkg.in/DataDog/dd-trace-go.v1/internal/appsec/dyngo/instrumentation/httpsec"
 
 	"golang.org/x/net/context"
 	"google.golang.org/grpc"
 	"google.golang.org/grpc/metadata"
 	"google.golang.org/grpc/peer"
+	"google.golang.org/grpc/status"
 )
 
 // UnaryHandler wrapper to use when AppSec is enabled to monitor its execution.
 func appsecUnaryHandlerMiddleware(span ddtrace.Span, handler grpc.UnaryHandler) grpc.UnaryHandler {
 	instrumentation.SetAppSecEnabledTags(span)
 	return func(ctx context.Context, req interface{}) (interface{}, error) {
 		md, _ := metadata.FromIncomingContext(ctx)
-		op := grpcsec.StartHandlerOperation(grpcsec.HandlerOperationArgs{Metadata: md}, nil)
+		ip := httpsec.IPFromHeaders(md, "")
+		op := grpcsec.StartHandlerOperation(grpcsec.HandlerOperationArgs{Metadata: md, ClientIP: ip}, nil)
 		defer func() {
 			events := op.Finish(grpcsec.HandlerOperationRes{})
 			instrumentation.SetTags(span, op.Tags())
 			if len(events) == 0 {
 				return
 			}
-			setAppSecTags(ctx, span, events)
+			setAppSecTags(ctx, span, ip, events)
 		}()
+		if op.BlockedCode != nil {
+			op.AddTag("appsec.blocked", true)
+			return nil, status.Errorf(*op.BlockedCode, "Request blocked")
+		}
 		defer grpcsec.StartReceiveOperation(grpcsec.ReceiveOperationArgs{}, op).Finish(grpcsec.ReceiveOperationRes{Message: req})
 		return handler(ctx, req)
 	}
@@ -43,15 +50,20 @@ func appsecStreamHandlerMiddleware(span ddtrace.Span, handler grpc.StreamHandler
 	instrumentation.SetAppSecEnabledTags(span)
 	return func(srv interface{}, stream grpc.ServerStream) error {
 		md, _ := metadata.FromIncomingContext(stream.Context())
-		op := grpcsec.StartHandlerOperation(grpcsec.HandlerOperationArgs{Metadata: md}, nil)
+		ip := httpsec.IPFromHeaders(md, "")
+		op := grpcsec.StartHandlerOperation(grpcsec.HandlerOperationArgs{Metadata: md, ClientIP: ip}, nil)
 		defer func() {
 			events := op.Finish(grpcsec.HandlerOperationRes{})
 			instrumentation.SetTags(span, op.Tags())
 			if len(events) == 0 {
 				return
 			}
-			setAppSecTags(stream.Context(), span, events)
+			setAppSecTags(stream.Context(), span, ip, events)
 		}()
+		if op.BlockedCode != nil {
+			op.AddTag("appsec.blocked", true)
+			return status.Error(*op.BlockedCode, "Request blocked")
+		}
 		return handler(srv, appsecServerStream{ServerStream: stream, handlerOperation: op})
 	}
 }
@@ -72,11 +84,11 @@ func (ss appsecServerStream) RecvMsg(m interface{}) error {
 }
 
 // Set the AppSec tags when security events were found.
-func setAppSecTags(ctx context.Context, span ddtrace.Span, events []json.RawMessage) {
+func setAppSecTags(ctx context.Context, span ddtrace.Span, clientIP instrumentation.NetaddrIP, events []json.RawMessage) {
 	md, _ := metadata.FromIncomingContext(ctx)
-	var addr net.Addr
+	var peerAddr net.Addr
 	if p, ok := peer.FromContext(ctx); ok {
-		addr = p.Addr
+		peerAddr = p.Addr
 	}
-	grpcsec.SetSecurityEventTags(span, events, addr, md)
+	grpcsec.SetSecurityEventTags(span, events, clientIP, peerAddr, md)
 }

contrib/google.golang.org/grpc/appsec_test.go

@@ -14,7 +14,9 @@ import (
 	"gopkg.in/DataDog/dd-trace-go.v1/internal/appsec"
 
 	"github.com/stretchr/testify/require"
+	"google.golang.org/grpc/codes"
 	"google.golang.org/grpc/metadata"
+	"google.golang.org/grpc/status"
 )
 
 func TestAppSec(t *testing.T) {
@@ -94,3 +96,90 @@ func TestAppSec(t *testing.T) {
 		require.True(t, strings.Contains(event, "ua0-600-55x")) // canary rule attack attempt
 	})
 }
+
+// Test that http blocking works by using custom rules/rules data
+func TestBlocking(t *testing.T) {
+	t.Setenv("DD_APPSEC_RULES", "../../../internal/appsec/testdata/blocking.json")
+	appsec.Start()
+	defer appsec.Stop()
+	if !appsec.Enabled() {
+		t.Skip("appsec disabled")
+	}
+
+	rig, err := newRig(false)
+	require.NoError(t, err)
+	defer rig.Close()
+
+	client := rig.client
+
+	t.Run("unary-block", func(t *testing.T) {
+		mt := mocktracer.Start()
+		defer mt.Stop()
+
+		// Send a XSS attack in the payload along with the canary value in the RPC metadata
+		ctx := metadata.NewOutgoingContext(context.Background(), metadata.Pairs("dd-canary", "dd-test-scanner-log", "x-client-ip", "1.2.3.4"))
+		reply, err := client.Ping(ctx, &FixtureRequest{Name: "<script>alert('xss');</script>"})
+
+		require.Nil(t, reply)
+		require.Equal(t, codes.Aborted, status.Code(err))
+
+		finished := mt.FinishedSpans()
+		require.Len(t, finished, 1)
+		// The request should have the attack attempts
+		event, _ := finished[0].Tag("_dd.appsec.json").(string)
+		require.NotNil(t, event)
+		require.True(t, strings.Contains(event, "blk-001-001"))
+	})
+
+	t.Run("unary-no-block", func(t *testing.T) {
+		mt := mocktracer.Start()
+		defer mt.Stop()
+
+		// Send a XSS attack in the payload along with the canary value in the RPC metadata
+		ctx := metadata.NewOutgoingContext(context.Background(), metadata.Pairs("dd-canary", "dd-test-scanner-log", "x-client-ip", "1.2.3.5"))
+		reply, err := client.Ping(ctx, &FixtureRequest{Name: "<script>alert('xss');</script>"})
+
+		require.Equal(t, "passed", reply.Message)
+		require.Equal(t, codes.OK, status.Code(err))
+	})
+
+	t.Run("stream-block", func(t *testing.T) {
+		mt := mocktracer.Start()
+		defer mt.Stop()
+
+		ctx := metadata.NewOutgoingContext(context.Background(), metadata.Pairs("dd-canary", "dd-test-scanner-log", "x-client-ip", "1.2.3.4"))
+		stream, err := client.StreamPing(ctx)
+		require.NoError(t, err)
+		reply, err := stream.Recv()
+
+		require.Equal(t, codes.Aborted, status.Code(err))
+		require.Nil(t, reply)
+
+		finished := mt.FinishedSpans()
+		require.Len(t, finished, 1)
+		// The request should have the attack attempts
+		event, _ := finished[0].Tag("_dd.appsec.json").(string)
+		require.NotNil(t, event)
+		require.True(t, strings.Contains(event, "blk-001-001"))
+	})
+
+	t.Run("stream-no-block", func(t *testing.T) {
+		mt := mocktracer.Start()
+		defer mt.Stop()
+
+		ctx := metadata.NewOutgoingContext(context.Background(), metadata.Pairs("dd-canary", "dd-test-scanner-log", "x-client-ip", "1.2.3.5"))
+		stream, err := client.StreamPing(ctx)
+		require.NoError(t, err)
+
+		// Send a XSS attack
+		err = stream.Send(&FixtureRequest{Name: "<script>alert('xss');</script>"})
+		require.NoError(t, err)
+		reply, err := stream.Recv()
+		require.Equal(t, codes.OK, status.Code(err))
+		require.Equal(t, "passed", reply.Message)
+
+		err = stream.CloseSend()
+		require.NoError(t, err)
+	})
+
+}

internal/appsec/dyngo/instrumentation/common.go

@@ -49,7 +49,7 @@ func (m *TagsHolder) Tags() map[string]interface{} {
 // See httpsec/http.go and grpcsec/grpc.go.
 type SecurityEventsHolder struct {
 	events []json.RawMessage
-	mu     sync.Mutex
+	mu     sync.RWMutex
 }
 
 // AddSecurityEvents adds the security events to the collected events list.
@@ -62,9 +62,18 @@ func (s *SecurityEventsHolder) AddSecurityEvents(events ...json.RawMessage) {
 
 // Events returns the list of stored events.
 func (s *SecurityEventsHolder) Events() []json.RawMessage {
+	s.mu.RLock()
+	defer s.mu.RUnlock()
 	return s.events
 }
 
+// ClearEvents clears the list of stored events
+func (s *SecurityEventsHolder) ClearEvents() {
+	s.mu.Lock()
+	defer s.mu.Unlock()
+	s.events = s.events[0:0]
+}
+
 // SetTags fills the span tags using the key/value pairs found in `tags`
 func SetTags(span TagSetter, tags map[string]interface{}) {
 	for k, v := range tags {
@@ -102,9 +111,11 @@ func SetEventSpanTags(span TagSetter, events []json.RawMessage) error {
 
 // Create the value of the security event tag.
 // TODO(Julio-Guerra): a future libddwaf version should return something
-//   avoiding us the following events concatenation logic which currently
-//   involves unserializing the top-level JSON arrays to concatenate them
-//   together.
+//
+//	avoiding us the following events concatenation logic which currently
+//	involves unserializing the top-level JSON arrays to concatenate them
+//	together.
+//
 // TODO(Julio-Guerra): avoid serializing the json in the request hot path
 func makeEventTagValue(events []json.RawMessage) (json.RawMessage, error) {
 	var v interface{}

internal/appsec/dyngo/instrumentation/grpcsec/actions.go

@@ -0,0 +1,69 @@
+// Unless explicitly stated otherwise all files in this repository are licensed
+// under the Apache License Version 2.0.
+// This product includes software developed at Datadog (https://www.datadoghq.com/).
+// Copyright 2022 Datadog, Inc.
+
+//go:build appsec
+// +build appsec
+
+package grpcsec
+
+import (
+	"sync"
+
+	"google.golang.org/grpc/codes"
+)
+
+// Action is used to identify any action kind
+type Action interface {
+	isAction()
+}
+
+// ActionsHandler handles WAF actions registration and execution
+type ActionsHandler struct {
+	mu      sync.RWMutex
+	actions map[string]Action
+}
+
+// NewActionsHandler returns an action handler holding the default ASM actions.
+// Currently, only the default "block" action is supported
+func NewActionsHandler() ActionsHandler {
+	// Register the default "block" action as specified in the blocking RFC
+	actions := map[string]Action{"block": &BlockRequestAction{Status: codes.Aborted}}
+
+	return ActionsHandler{
+		actions: actions,
+	}
+}
+
+// RegisterAction registers a specific action to the actions handler. If the action kind is unknown
+// the action will have no effect
+func (h *ActionsHandler) RegisterAction(id string, a Action) {
+	h.mu.Lock()
+	defer h.mu.Unlock()
+	h.actions[id] = a
+}
+
+// Apply executes the action identified by `id`
+func (h *ActionsHandler) Apply(id string, op *HandlerOperation) bool {
+	h.mu.RLock()
+	a, ok := h.actions[id]
+	h.mu.RUnlock()
+	if !ok {
+		return false
+	}
+	// Currently, only the "block_request" type is supported, so we only need to check for blockRequestParams
+	if p, ok := a.(*BlockRequestAction); ok {
+		op.BlockedCode = &p.Status
+		return true
+	}
+	return false
+}
+
+// BlockRequestAction is the struct used to perform the request blocking action
+type BlockRequestAction struct {
+	// Status is the return code to use when blocking the request
+	Status codes.Code
+}
+
+func (*BlockRequestAction) isAction() {}

internal/appsec/dyngo/instrumentation/grpcsec/grpc.go

@@ -15,6 +15,8 @@ import (
 
 	"gopkg.in/DataDog/dd-trace-go.v1/internal/appsec/dyngo"
 	"gopkg.in/DataDog/dd-trace-go.v1/internal/appsec/dyngo/instrumentation"
+
+	"google.golang.org/grpc/codes"
 )
 
 // Abstract gRPC server handler operation definitions. It is based on two
@@ -39,12 +41,14 @@ type (
 		dyngo.Operation
 		instrumentation.TagsHolder
 		instrumentation.SecurityEventsHolder
+		BlockedCode *codes.Code
 	}
 	// HandlerOperationArgs is the grpc handler arguments.
 	HandlerOperationArgs struct {
 		// Message received by the gRPC handler.
 		// Corresponds to the address `grpc.server.request.metadata`.
 		Metadata map[string][]string
+		ClientIP instrumentation.NetaddrIP
 	}
 	// HandlerOperationRes is the grpc handler results. Empty as of today.
 	HandlerOperationRes struct{}

internal/appsec/dyngo/instrumentation/grpcsec/tags.go

@@ -17,26 +17,31 @@ import (
 
 // SetSecurityEventTags sets the AppSec-specific span tags when a security event
 // occurred into the service entry span.
-func SetSecurityEventTags(span ddtrace.Span, events []json.RawMessage, addr net.Addr, md map[string][]string) {
-	if err := setSecurityEventTags(span, events, addr, md); err != nil {
+func SetSecurityEventTags(span ddtrace.Span, events []json.RawMessage, clientIP instrumentation.NetaddrIP, peerAddr net.Addr, md map[string][]string) {
+	if err := setSecurityEventTags(span, events, clientIP, peerAddr, md); err != nil {
 		log.Error("appsec: %v", err)
 	}
 }
 
-func setSecurityEventTags(span ddtrace.Span, events []json.RawMessage, addr net.Addr, md map[string][]string) error {
+func setSecurityEventTags(span ddtrace.Span, events []json.RawMessage, clientIP instrumentation.NetaddrIP, peerAddr net.Addr, md map[string][]string) error {
 	if err := instrumentation.SetEventSpanTags(span, events); err != nil {
 		return err
 	}
-	var ip string
-	switch actual := addr.(type) {
+	var peerIP string
+	switch actual := peerAddr.(type) {
 	case *net.UDPAddr:
-		ip = actual.IP.String()
+		peerIP = actual.IP.String()
 	case *net.TCPAddr:
-		ip = actual.IP.String()
+		peerIP = actual.IP.String()
 	}
-	if ip != "" {
-		span.SetTag("network.client.ip", ip)
+	if peerIP != "" {
+		span.SetTag("network.client.ip", peerIP)
 	}
+
+	if clientIP.IsValid() {
+		span.SetTag("http.client_ip", clientIP.String())
+	}
+
 	for h, v := range httpsec.NormalizeHTTPHeaders(md) {
 		span.SetTag("grpc.metadata."+h, v)
 	}