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

contrib/internal/httputil/trace.go

@@ -15,6 +15,9 @@ import (
 	"gopkg.in/DataDog/dd-trace-go.v1/ddtrace"
 	"gopkg.in/DataDog/dd-trace-go.v1/ddtrace/ext"
 	"gopkg.in/DataDog/dd-trace-go.v1/ddtrace/tracer"
+	httpinstr "gopkg.in/DataDog/dd-trace-go.v1/internal/appsec/instrumentation/http"
+	appsectypes "gopkg.in/DataDog/dd-trace-go.v1/internal/appsec/types"
+	"gopkg.in/DataDog/dd-trace-go.v1/internal/dyngo"
 )
 
 // TraceConfig defines the configuration for request tracing.
@@ -54,6 +57,34 @@ func TraceAndServe(h http.Handler, cfg *TraceConfig) {
 
 	cfg.ResponseWriter = wrapResponseWriter(cfg.ResponseWriter, span)
 
+	op := dyngo.StartOperation(
+		httpinstr.HandlerOperationArgs{
+			IsTLS:       cfg.Request.TLS != nil,
+			Method:      httpinstr.Method(cfg.Request.Method),
+			Host:        httpinstr.Host(cfg.Request.Host),
+			RequestURI:  httpinstr.RequestURI(cfg.Request.RequestURI),
+			RemoteAddr:  httpinstr.RemoteAddr(cfg.Request.RemoteAddr),
+			Headers:     httpinstr.Header(cfg.Request.Header),
+			UserAgent:   httpinstr.UserAgent(cfg.Request.UserAgent()),
+			QueryValues: httpinstr.QueryValues(cfg.Request.URL.Query()),
+		},
+	)
+	op.OnData(func(_ *dyngo.Operation, e *appsectypes.SecurityEvent) {
+		// Keep this trace due to the security event
+		span.SetTag(ext.SamplingPriority, ext.ManualKeep)
+		// Add context to the event
+		spanCtx := span.Context()
+		// Add the APM context to the event
+		e.AddContext(appsectypes.SpanContext{
+			TraceID: spanCtx.TraceID(),
+			SpanID:  spanCtx.SpanID(),
+		})
+	})
+	defer func() {
+		// TODO: get the status code out of the wrapped response write
+		op.Finish(httpinstr.HandlerOperationRes{})
+	}()
+
 	h.ServeHTTP(cfg.ResponseWriter, cfg.Request.WithContext(ctx))
 }
 

ddtrace/tracer/tracer.go

@@ -17,7 +17,9 @@ import (
 	"gopkg.in/DataDog/dd-trace-go.v1/ddtrace"
 	"gopkg.in/DataDog/dd-trace-go.v1/ddtrace/ext"
 	"gopkg.in/DataDog/dd-trace-go.v1/ddtrace/internal"
+	"gopkg.in/DataDog/dd-trace-go.v1/internal/appsec"
 	"gopkg.in/DataDog/dd-trace-go.v1/internal/log"
+	"gopkg.in/DataDog/dd-trace-go.v1/internal/version"
 )
 
 var _ ddtrace.Tracer = (*tracer)(nil)
@@ -78,6 +80,7 @@ type tracer struct {
 	// rules for applying a sampling rate to spans that match the designated service
 	// or operation name.
 	rulesSampling *rulesSampler
+	appsec        *appsec.Agent
 }
 
 const (
@@ -113,6 +116,7 @@ func Start(opts ...StartOption) {
 	if t.config.HasFeature("discovery") {
 		t.loadAgentFeatures()
 	}
+	startAppSec(t)
 	internal.SetGlobalTracer(t)
 	if t.config.logStartup {
 		logStartup(t)
@@ -121,6 +125,9 @@ func Start(opts ...StartOption) {
 
 // Stop stops the started tracer. Subsequent calls are valid but become no-op.
 func Stop() {
+	if v, ok := internal.GetGlobalTracer().(*tracer); ok && v.appsec != nil {
+		v.appsec.Stop(true)
+	}
 	internal.SetGlobalTracer(&internal.NoopTracer{})
 	log.Flush()
 }
@@ -498,3 +505,44 @@ func (t *tracer) sample(span *span) {
 	}
 	t.prioritySampling.apply(span)
 }
+
+// Start AppSec when DD_APPSEC_ENABLED is true.
+func startAppSec(t *tracer) {
+	if appsecEnabled := os.Getenv("DD_APPSEC_ENABLED"); appsecEnabled == "" {
+		return
+	} else if enabled, err := strconv.ParseBool(appsecEnabled); err != nil {
+		log.Error("appsec: could not parse DD_APPSEC_ENABLED value `%s` as a boolean value", appsecEnabled)
+		return
+	} else if !enabled {
+		return
+	}
+
+	appsecAgent, err := appsec.NewAgent(
+		t.config.httpClient,
+		appsecLogger{},
+		&appsec.Config{
+			Version:  version.Tag,
+			AgentURL: fmt.Sprintf("http://%s/", t.config.agentAddr),
+			Service: appsec.ServiceConfig{
+				Name:        t.config.serviceName,
+				Version:     t.config.version,
+				Environment: t.config.env,
+			},
+		},
+	)
+	if err != nil {
+		log.Error("could not start appsec: %v", err)
+	}
+	appsecAgent.Start()
+	t.appsec = appsecAgent
+}
+
+type appsecLogger struct{}
+
+func (appsecLogger) Warn(fmt string, v ...interface{}) {
+	log.Warn(fmt, v...)
+}
+
+func (appsecLogger) Error(fmt string, v ...interface{}) {
+	log.Error(fmt, v...)
+}

go.mod

@@ -7,5 +7,8 @@ require (
 	github.com/DataDog/gostackparse v0.5.0
 	github.com/DataDog/sketches-go v1.0.0
 	github.com/google/pprof v0.0.0-20210423192551-a2663126120b
+	github.com/google/uuid v1.3.0
+	github.com/sqreen/go-libsqreen v0.7.2
+	github.com/stretchr/testify v1.7.0
 	github.com/tinylib/msgp v1.1.2
 )

go.sum

@@ -265,6 +265,8 @@ github.com/google/renameio v0.1.0/go.mod h1:KWCgfxg9yswjAJkECMjeO8J8rahYeXnNhOm4
 github.com/google/uuid v1.1.1/go.mod h1:TIyPZe4MgqvfeYDBFedMoGGpEw/LqOeaOT+nhxU+yHo=
 github.com/google/uuid v1.1.2 h1:EVhdT+1Kseyi1/pUmXKaFxYsDNy9RQYkMWRH68J/W7Y=
 github.com/google/uuid v1.1.2/go.mod h1:TIyPZe4MgqvfeYDBFedMoGGpEw/LqOeaOT+nhxU+yHo=
+github.com/google/uuid v1.3.0 h1:t6JiXgmwXMjEs8VusXIJk2BXHsn+wx8BZdTaoZ5fu7I=
+github.com/google/uuid v1.3.0/go.mod h1:TIyPZe4MgqvfeYDBFedMoGGpEw/LqOeaOT+nhxU+yHo=
 github.com/googleapis/gax-go/v2 v2.0.4/go.mod h1:0Wqv26UfaUD9n4G6kQubkQ+KchISgw+vpHVxEJEs9eg=
 github.com/googleapis/gax-go/v2 v2.0.5 h1:sjZBwGj9Jlw33ImPtvFviGYvseOtDM7hkSKB7+Tv3SM=
 github.com/googleapis/gax-go/v2 v2.0.5/go.mod h1:DWXyrwAJ9X0FpwwEdw+IPEYBICEFu5mhpdKc/us6bOk=
@@ -530,6 +532,9 @@ github.com/spf13/pflag v0.0.0-20170130214245-9ff6c6923cff/go.mod h1:DYY7MBk1bdzu
 github.com/spf13/pflag v1.0.3/go.mod h1:DYY7MBk1bdzusC3SYhjObp+wFpr4gzcvqqNjLnInEg4=
 github.com/spf13/pflag v1.0.5 h1:iy+VFUOCP1a+8yFto/drg2CJ5u0yRoB7fZw3DKv/JXA=
 github.com/spf13/pflag v1.0.5/go.mod h1:McXfInJRrz4CZXVZOBLb0bTZqETkiAhM9Iw0y3An2Bg=
+github.com/sqreen/go-libsqreen v0.7.2 h1:7W2i9EGQJyoy1/iuxyhg6PWuGnJkHX9T8880rUfWFsE=
+github.com/sqreen/go-libsqreen v0.7.2/go.mod h1:krFVmXmHM5SaWeED8jDb8KwrViK505KDBpYJ8IY2Ks8=
+github.com/stretchr/objx v0.1.0 h1:4G4v2dO3VZwixGIRoQ5Lfboy6nUhCyYzaqnIAPPhYs4=
 github.com/stretchr/objx v0.1.0/go.mod h1:HFkY916IF+rwdDfMAkV7OtwuqBVzrE8GR6GFx+wExME=
 github.com/stretchr/objx v0.1.1 h1:2vfRuCMp5sSVIDSqO8oNnWJq7mPa6KVP3iPIwFBuy8A=
 github.com/stretchr/objx v0.1.1/go.mod h1:HFkY916IF+rwdDfMAkV7OtwuqBVzrE8GR6GFx+wExME=

internal/appsec/appsec.go

@@ -0,0 +1,222 @@
+// 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 2016 Datadog, Inc.
+
+package appsec
+
+import (
+	"context"
+	"fmt"
+	"net/http"
+	"os"
+	"path/filepath"
+	"runtime"
+	"sync"
+	"time"
+
+	"gopkg.in/DataDog/dd-trace-go.v1/internal/appsec/internal/intake"
+	"gopkg.in/DataDog/dd-trace-go.v1/internal/appsec/internal/intake/api"
+	httpprotection "gopkg.in/DataDog/dd-trace-go.v1/internal/appsec/internal/protection/http"
+	appsectypes "gopkg.in/DataDog/dd-trace-go.v1/internal/appsec/types"
+	"gopkg.in/DataDog/dd-trace-go.v1/internal/dyngo"
+)
+
+type (
+	Config struct {
+		AgentURL          string
+		Service           ServiceConfig
+		Tags              []string
+		Hostname          string
+		MaxBatchLen       int
+		MaxBatchStaleTime time.Duration
+		Version           string
+	}
+
+	ServiceConfig struct {
+		Name, Version, Environment string
+	}
+)
+
+// Default batching configuration values.
+const (
+	defaultMaxBatchLen       = 1024
+	defaultMaxBatchStaleTime = time.Second
+)
+
+// Default timeout of intake requests.
+const defaultIntakeTimeout = 10 * time.Second
+
+type Agent struct {
+	client          *intake.Client
+	eventChan       chan *appsectypes.SecurityEvent
+	wg              sync.WaitGroup
+	cfg             *Config
+	unregisterInstr []dyngo.UnregisterFunc
+}
+
+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() {
+	a.run()
+}
+
+func (a *Agent) Stop(gracefully bool) {
+	for _, unregister := range a.unregisterInstr {
+		unregister()
+	}
+	// 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() {
+	a.unregisterInstr = append(a.unregisterInstr, httpprotection.Register(), 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:        a.cfg.Version,
+			},
+			appsectypes.HostContext{
+				Hostname: a.cfg.Hostname,
+				OS:       fmt.Sprintf("%s/%s", runtime.GOOS, runtime.GOARCH),
+			},
+		}
+
+		eventBatchingLoop(a.client, a.eventChan, globalEventCtx, a.cfg)
+	}()
+}
+
+type IntakeClient interface {
+	SendBatch(context.Context, api.EventBatch) error
+}
+
+func eventBatchingLoop(client IntakeClient, eventChan <-chan *appsectypes.SecurityEvent, globalEventCtx []appsectypes.SecurityEventContext, cfg *Config) {
+	// The batch of events
+	batch := make([]*appsectypes.SecurityEvent, 0, cfg.MaxBatchLen)
+
+	// Timer initialized to a first dummy time value to initialize it and so that we can immediately
+	// use its channel field in the following select statement.
+	timer := time.NewTimer(time.Hour)
+	timer.Stop()
+
+	// Helper function stopping the timer, sending the batch and resetting it.
+	sendBatch := func() {
+		if !timer.Stop() {
+			// Remove the time value from the channel in case the timer fired and so that we avoid
+			// sending the batch again in the next loop iteration due to a value in the timer
+			// channel.
+			select {
+			case <-timer.C:
+			default:
+			}
+		}
+		ctx, cancel := context.WithTimeout(context.Background(), defaultIntakeTimeout)
+		defer cancel()
+		client.SendBatch(ctx, api.FromSecurityEvents(batch, globalEventCtx))
+		batch = batch[0:0]
+	}
+
+	// Loop-select between the event channel or the stale timer (when enabled).
+	for {
+		select {
+		case event, ok := <-eventChan:
+			// Add the event to the batch.
+			// The event might be nil when closing the channel while it was empty.
+			if event != nil {
+				batch = append(batch, event)
+			}
+			if !ok {
+				// The event channel has been closed. Send the batch if it's not empty.
+				if len(batch) > 0 {
+					sendBatch()
+				}
+				return
+			}
+			// Send the batch when it's full or start the timer when this is the first value in
+			// the batch.
+			if l := len(batch); l == cfg.MaxBatchLen {
+				sendBatch()
+			} else if l == 1 {
+				timer.Reset(cfg.MaxBatchStaleTime)
+			}
+
+		case <-timer.C:
+			sendBatch()
+		}
+	}
+}
+
+func (a *Agent) listenSecurityEvents() dyngo.UnregisterFunc {
+	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
+				}
+			}),
+		},
+	})
+
+}