internal/appsec: add new AppSec beta functionality and support for contrib/net/http

appsec/_test/appsec_test.go

@@ -0,0 +1,67 @@
+package _test_test
+
+import (
+	"io/ioutil"
+	"net/http"
+	"net/http/httptest"
+	"os"
+	"strings"
+	"testing"
+
+	httptrace "gopkg.in/DataDog/dd-trace-go.v1/contrib/net/http"
+	"gopkg.in/DataDog/dd-trace-go.v1/ddtrace/tracer"
+
+	"github.com/stretchr/testify/require"
+)
+
+// TestWAF is a simple validation test of the WAF protecting a net/http server. It only mockups the agent and tests that
+// the WAF is properly detecting an XSS attempt and that the corresponding security event is being sent to the agent.
+func TestWAF(t *testing.T) {
+	// Start the HTTP server acting as the agent
+	// Its handler counts the number of AppSec API requests and saves the latest event batch sent (only one expected).
+	var (
+		nbAppSecAPIRequests int
+		batch               []byte
+	)
+	agent := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		w.WriteHeader(http.StatusOK)
+		if r.RequestURI == "/appsec/proxy/api/v2/appsecevts" {
+			nbAppSecAPIRequests++
+			var err error
+			batch, err = ioutil.ReadAll(r.Body)
+			require.NoError(t, err)
+		}
+	}))
+
+	// Start the tracer along with the fake agent HTTP server
+	os.Setenv("DD_APPSEC_ENABLED", "true")
+	tracer.Start(tracer.WithAgentAddr(strings.TrimPrefix(agent.URL, "http://")))
+
+	// Start and trace an HTTP server
+	mux := httptrace.NewServeMux()
+	mux.HandleFunc("/", func(w http.ResponseWriter, r *http.Request) {
+		w.Write([]byte("Hello World!\n"))
+	})
+	srv := httptest.NewServer(mux)
+	defer srv.Close()
+
+	// Send an XSS attack
+	req, err := http.NewRequest("POST", srv.URL+"/?attack=<script>alert()</script>", nil)
+	if err != nil {
+		panic(err)
+	}
+	res, err := srv.Client().Do(req)
+	require.NoError(t, err)
+
+	// Check that the handler was properly called
+	b, err := ioutil.ReadAll(res.Body)
+	require.NoError(t, err)
+	require.Equal(t, "Hello World!\n", string(b))
+
+	// Stop the tracer so that the AppSec events gets sent
+	tracer.Stop()
+
+	// Check that an XSS attack event was reported.
+	require.Equal(t, 1, nbAppSecAPIRequests)
+	require.True(t, strings.Contains(string(batch), "xss-blocking"))
+}

appsec/_test/go.mod

@@ -0,0 +1,13 @@
+module appsectest
+
+go 1.16
+
+replace (
+	github.com/DataDog/dd-trace-go/appsec => ../
+	gopkg.in/DataDog/dd-trace-go.v1 => ../../
+)
+
+require (
+	github.com/stretchr/testify v1.7.0
+	gopkg.in/DataDog/dd-trace-go.v1 v1.33.0
+)

appsec/appsec.go

@@ -0,0 +1,196 @@
+package appsec
+
+import (
+	"context"
+	"fmt"
+	httpprotection "github.com/DataDog/dd-trace-go/appsec/internal/protection/http"
+	"net/http"
+	"os"
+	"path/filepath"
+	"runtime"
+	"sync"
+	"time"
+
+	"github.com/DataDog/dd-trace-go/appsec/dyngo"
+	"github.com/DataDog/dd-trace-go/appsec/internal/intake"
+	"github.com/DataDog/dd-trace-go/appsec/internal/intake/api"
+	appsectypes "github.com/DataDog/dd-trace-go/appsec/types"
+)
+
+type (
+	Config struct {
+		AgentURL          string
+		Service           ServiceConfig
+		Tags              []string
+		Hostname          string
+		MaxBatchLen       int
+		MaxBatchStaleTime time.Duration
+	}
+
+	ServiceConfig struct {
+		Name, Version, Environment string
+	}
+)
+
+const (
+	defaultMaxBatchLen       = 1024
+	defaultMaxBatchStaleTime = time.Second
+)
+
+type Agent struct {
+	client             *intake.Client
+	eventChan          chan *appsectypes.SecurityEvent
+	wg                 sync.WaitGroup
+	cfg                *Config
+	instrumentationIDs []dyngo.EventListenerID
+}
+
+type Logger interface {
+	Warn(string, ...interface{})
+	Error(string, ...interface{})
+}
+
+func NewAgent(client *http.Client, logger Logger, cfg *Config) (*Agent, error) {
+	intakeClient, err := intake.NewClient(client, cfg.AgentURL)
+	if err != nil {
+		return nil, err
+	}
+
+	if cfg.Hostname == "" {
+		hostname, err := os.Hostname()
+		if err != nil {
+			logger.Warn("unable to look up hostname: %v", err)
+		} else {
+			cfg.Hostname = hostname
+		}
+	}
+
+	if cfg.Service.Name == "" {
+		name, err := os.Executable()
+		if err != nil {
+			logger.Warn("unable to look up the executable name: %v", err)
+		} else {
+			cfg.Service.Name = filepath.Base(name)
+		}
+	}
+
+	if cfg.MaxBatchLen <= 0 {
+		cfg.MaxBatchLen = defaultMaxBatchLen
+	}
+
+	if cfg.MaxBatchStaleTime <= 0 {
+		cfg.MaxBatchStaleTime = defaultMaxBatchStaleTime
+	}
+
+	return &Agent{
+		eventChan: make(chan *appsectypes.SecurityEvent, 1000),
+		client:    intakeClient,
+		cfg:       cfg,
+	}, nil
+}
+
+func (a *Agent) Start(ctx context.Context) {
+	a.run(ctx)
+}
+
+func (a *Agent) Stop(gracefully bool) {
+	dyngo.Unregister(a.instrumentationIDs)
+	// Stop the batching goroutine
+	close(a.eventChan)
+	// Possibly wait for the goroutine to gracefully stop
+	if !gracefully {
+		return
+	}
+	a.wg.Wait()
+}
+
+func (a *Agent) run(ctx context.Context) {
+	a.instrumentationIDs = append(a.instrumentationIDs, httpprotection.Register()...)
+	a.instrumentationIDs = append(a.instrumentationIDs, a.listenSecurityEvents()...)
+
+	a.wg.Add(1)
+	go func() {
+		defer a.wg.Done()
+
+		globalEventCtx := []appsectypes.SecurityEventContext{
+			appsectypes.ServiceContext{
+				Name:        a.cfg.Service.Name,
+				Version:     a.cfg.Service.Version,
+				Environment: a.cfg.Service.Environment,
+			},
+			appsectypes.TagContext(a.cfg.Tags),
+			appsectypes.TracerContext{
+				Runtime:        "go",
+				RuntimeVersion: runtime.Version(),
+				Version:        version,
+			},
+			appsectypes.HostContext{
+				Hostname: a.cfg.Hostname,
+				OS:       fmt.Sprintf("%s/%s", runtime.GOOS, runtime.GOARCH),
+			},
+		}
+
+		eventBatchingLoop(ctx, a.client, a.eventChan, globalEventCtx, a.cfg)
+	}()
+}
+
+type IntakeClient interface {
+	SendBatch(context.Context, api.EventBatch) error
+}
+
+func eventBatchingLoop(ctx context.Context, client IntakeClient, eventChan <-chan *appsectypes.SecurityEvent, globalEventCtx []appsectypes.SecurityEventContext, cfg *Config) {
+	batch := make([]*appsectypes.SecurityEvent, 0, cfg.MaxBatchLen)
+	timer := time.NewTimer(time.Hour)
+	timer.Stop()
+
+	for {
+		select {
+		case <-ctx.Done():
+			// Return immediately as the context is now done so the http client cannot be used anymore
+			return
+
+		case event, ok := <-eventChan:
+			if event != nil {
+				batch = append(batch, event)
+			}
+			if !ok {
+				timer.Stop()
+				if len(batch) > 0 {
+					client.SendBatch(ctx, api.FromSecurityEvents(batch, globalEventCtx))
+				}
+				return
+			}
+
+			if l := len(batch); l == cfg.MaxBatchLen {
+				timer.Stop()
+				client.SendBatch(ctx, api.FromSecurityEvents(batch, globalEventCtx))
+				batch = batch[0:0]
+			} else if l == 1 {
+				timer = time.NewTimer(cfg.MaxBatchStaleTime)
+			}
+
+		case <-timer.C:
+			if len(batch) == 0 {
+				continue
+			}
+			client.SendBatch(ctx, api.FromSecurityEvents(batch, globalEventCtx))
+			batch = batch[0:0]
+		}
+	}
+}
+
+func (a *Agent) listenSecurityEvents() []dyngo.EventListenerID {
+	return dyngo.Register(dyngo.InstrumentationDescriptor{
+		Title: "Attack Queue",
+		Instrumentation: dyngo.OperationInstrumentation{
+			EventListener: dyngo.OnDataEventListener(func(_ *dyngo.Operation, event *appsectypes.SecurityEvent) {
+				select {
+				case a.eventChan <- event:
+				default:
+					// TODO: add metrics on the nb of dropped events
+				}
+			}),
+		},
+	})
+
+}