Emit new event for DynamoDB requests via app access

lib/events/api.go

@@ -395,6 +395,10 @@ const (
 	// AppSessionRequestEvent is an HTTP request and response.
 	AppSessionRequestEvent = "app.session.request"
 
+	// AppSessionDynamoDBRequestEvent is emitted when DynamoDB client sends
+	// a request via app access session.
+	AppSessionDynamoDBRequestEvent = "app.session.dynamodb.request"
+
 	// DatabaseCreateEvent is emitted when a database resource is created.
 	DatabaseCreateEvent = "db.create"
 	// DatabaseUpdateEvent is emitted when a database resource is updated.

lib/events/codes.go

@@ -108,6 +108,8 @@ const (
 	AppSessionEndCode = "T2011I"
 	// SessionRecordingAccessCode is the session recording view data event code.
 	SessionRecordingAccessCode = "T2012I"
+	// AppSessionDynamoDBRequestCode is the application request/response code.
+	AppSessionDynamoDBRequestCode = "T2013I"
 
 	// AppCreateCode is the app.create event code.
 	AppCreateCode = "TAP03I"

lib/events/dynamic.go

@@ -151,6 +151,8 @@ func FromEventFields(fields EventFields) (events.AuditEvent, error) {
 		e = &events.AppSessionChunk{}
 	case AppSessionRequestEvent:
 		e = &events.AppSessionRequest{}
+	case AppSessionDynamoDBRequestEvent:
+		e = &events.AppSessionDynamoDBRequest{}
 	case AppCreateEvent:
 		e = &events.AppCreate{}
 	case AppUpdateEvent:

lib/service/service.go

@@ -4295,9 +4295,9 @@ func (process *TeleportProcess) initApps() {
 			return trace.Wrap(err)
 		}
 
-		ok := false
+		shouldSkipCleanup := false
 		defer func() {
-			if !ok {
+			if !shouldSkipCleanup {
 				warnOnErr(conn.Close(), log)
 			}
 		}()
@@ -4431,6 +4431,12 @@ func (process *TeleportProcess) initApps() {
 
 		proxyGetter := reversetunnel.NewConnectedProxyGetter()
 
+		defer func() {
+			if !shouldSkipCleanup {
+				warnOnErr(asyncEmitter.Close(), log)
+			}
+		}()
+
 		appServer, err := app.New(process.ExitContext(), &app.Config{
 			Clock:                process.Config.Clock,
 			DataDir:              process.Config.DataDir,
@@ -4456,7 +4462,7 @@ func (process *TeleportProcess) initApps() {
 		}
 
 		defer func() {
-			if !ok {
+			if !shouldSkipCleanup {
 				warnOnErr(appServer.Close(), log)
 			}
 		}()
@@ -4494,13 +4500,16 @@ func (process *TeleportProcess) initApps() {
 		log.Infof("All applications successfully started.")
 
 		// Cancel deferred cleanup actions, because we're going
-		// to regsiter an OnExit handler to take care of it
-		ok = true
+		// to register an OnExit handler to take care of it
+		shouldSkipCleanup = true
 
 		// Execute this when process is asked to exit.
 		process.OnExit("apps.stop", func(payload interface{}) {
 			log.Infof("Shutting down.")
 			warnOnErr(appServer.Close(), log)
+			if asyncEmitter != nil {
+				warnOnErr(asyncEmitter.Close(), log)
+			}
 			agentPool.Stop()
 			warnOnErr(asyncEmitter.Close(), log)
 			warnOnErr(conn.Close(), log)

lib/srv/app/aws/endpoints.go

@@ -124,6 +124,22 @@ func endpointsIDFromSigningName(signingName string) string {
 	return signingName
 }
 
+func isDynamoDBEndpoint(re *endpoints.ResolvedEndpoint) bool {
+	// Some clients may sign some services with upper case letters. We use all
+	// lower cases in our mapping.
+	signingName := strings.ToLower(re.SigningName)
+	_, ok := dynamoDBSigningNames[signingName]
+	return ok
+}
+
+// dynamoDBSigningNames is a set of signing names used for DynamoDB APIs.
+var dynamoDBSigningNames = map[string]struct{}{
+	// signing name for dynamodb and dynamodbstreams API.
+	"dynamodb": {},
+	// signing name for dynamodb accelerator API.
+	"dax": {},
+}
+
 // signingNameToEndpointsID is a map of AWS services' signing names to their
 // endpoints IDs.
 //

lib/srv/app/aws/handler.go

@@ -18,7 +18,7 @@ package aws
 
 import (
 	"bytes"
-	"context"
+	"io"
 	"net/http"
 	"net/url"
 
@@ -29,14 +29,12 @@ import (
 	"github.com/aws/aws-sdk-go/aws/endpoints"
 	awssession "github.com/aws/aws-sdk-go/aws/session"
 	"github.com/gravitational/oxy/forward"
-	"github.com/gravitational/oxy/utils"
+	oxyutils "github.com/gravitational/oxy/utils"
 	"github.com/gravitational/trace"
 	"github.com/jonboulle/clockwork"
 	"github.com/sirupsen/logrus"
 
-	apievents "github.com/gravitational/teleport/api/types/events"
 	"github.com/gravitational/teleport/lib/defaults"
-	"github.com/gravitational/teleport/lib/events"
 	"github.com/gravitational/teleport/lib/srv/app/common"
 	awsutils "github.com/gravitational/teleport/lib/utils/aws"
 )
@@ -52,7 +50,7 @@ func NewSigningService(config SigningServiceConfig) (*SigningService, error) {
 
 	fwd, err := forward.New(
 		forward.RoundTripper(svc),
-		forward.ErrorHandler(utils.ErrorHandlerFunc(svc.formatForwardResponseError)),
+		forward.ErrorHandler(oxyutils.ErrorHandlerFunc(svc.formatForwardResponseError)),
 		forward.PassHostHeader(true),
 	)
 	if err != nil {
@@ -74,8 +72,8 @@ type SigningService struct {
 
 // SigningServiceConfig is the SigningService configuration.
 type SigningServiceConfig struct {
-	// Client is an HTTP client instance used for HTTP calls.
-	Client *http.Client
+	// RoundTripper is an http.RoundTripper instance used for requests.
+	RoundTripper http.RoundTripper
 	// Log is the Logger.
 	Log logrus.FieldLogger
 	// Session is AWS session.
@@ -90,14 +88,12 @@ type SigningServiceConfig struct {
 
 // CheckAndSetDefaults validates the SigningServiceConfig config.
 func (s *SigningServiceConfig) CheckAndSetDefaults() error {
-	if s.Client == nil {
+	if s.RoundTripper == nil {
 		tr, err := defaults.Transport()
 		if err != nil {
 			return trace.Wrap(err)
 		}
-		s.Client = &http.Client{
-			Transport: tr,
-		}
+		s.RoundTripper = tr
 	}
 	if s.Clock == nil {
 		s.Clock = clockwork.NewRealClock()
@@ -137,6 +133,7 @@ func (s *SigningServiceConfig) CheckAndSetDefaults() error {
 //	 5. Sign HTTP request.
 //	 6. Forward the signed HTTP request to the AWS API.
 func (s *SigningService) RoundTrip(req *http.Request) (*http.Response, error) {
+	defer req.Body.Close()
 	sessionCtx, err := common.GetSessionContext(req)
 	if err != nil {
 		return nil, trace.Wrap(err)
@@ -145,46 +142,31 @@ func (s *SigningService) RoundTrip(req *http.Request) (*http.Response, error) {
 	if err != nil {
 		return nil, trace.Wrap(err)
 	}
-	signedReq, err := s.prepareSignedRequest(req, resolvedEndpoint, sessionCtx)
+	payload, err := awsutils.GetAndReplaceReqBody(req)
 	if err != nil {
 		return nil, trace.Wrap(err)
 	}
-	resp, err := s.Client.Do(signedReq)
+	signedReq, err := s.prepareSignedRequest(req, payload, resolvedEndpoint, sessionCtx)
 	if err != nil {
 		return nil, trace.Wrap(err)
 	}
-
-	if err := s.emitAuditEvent(req.Context(), signedReq, resp, sessionCtx, resolvedEndpoint); err != nil {
+	resp, err := s.RoundTripper.RoundTrip(signedReq)
+	if err != nil {
+		return nil, trace.Wrap(err)
+	}
+	// emit audit event with original request, but change the URL since we resolved and rewrote it.
+	signedReq.Body = io.NopCloser(bytes.NewReader(payload))
+	if isDynamoDBEndpoint(resolvedEndpoint) {
+		err = sessionCtx.Audit.OnDynamoDBRequest(req.Context(), sessionCtx, signedReq, resp, resolvedEndpoint)
+	} else {
+		err = sessionCtx.Audit.OnRequest(req.Context(), sessionCtx, signedReq, resp, resolvedEndpoint)
+	}
+	if err != nil {
 		s.Log.WithError(err).Warn("Failed to emit audit event.")
 	}
 	return resp, nil
 }
 
-// emitAuditEvent writes details of the AWS request to audit stream.
-func (s *SigningService) emitAuditEvent(ctx context.Context, req *http.Request, resp *http.Response, sessionCtx *common.SessionContext, endpoint *endpoints.ResolvedEndpoint) error {
-	event := &apievents.AppSessionRequest{
-		Metadata: apievents.Metadata{
-			Type: events.AppSessionRequestEvent,
-			Code: events.AppSessionRequestCode,
-		},
-		Method:     req.Method,
-		Path:       req.URL.Path,
-		RawQuery:   req.URL.RawQuery,
-		StatusCode: uint32(resp.StatusCode),
-		AppMetadata: apievents.AppMetadata{
-			AppURI:        sessionCtx.App.GetURI(),
-			AppPublicAddr: sessionCtx.App.GetPublicAddr(),
-			AppName:       sessionCtx.App.GetName(),
-		},
-		AWSRequestMetadata: apievents.AWSRequestMetadata{
-			AWSRegion:  endpoint.SigningRegion,
-			AWSService: endpoint.SigningName,
-			AWSHost:    req.Host,
-		},
-	}
-	return trace.Wrap(sessionCtx.Emitter.EmitAuditEvent(ctx, event))
-}
-
 func (s *SigningService) formatForwardResponseError(rw http.ResponseWriter, r *http.Request, err error) {
 	switch trace.Unwrap(err).(type) {
 	case *trace.BadParameterError:
@@ -201,15 +183,11 @@ func (s *SigningService) formatForwardResponseError(rw http.ResponseWriter, r *h
 
 // prepareSignedRequest creates a new HTTP request and rewrites the header from the original request and returns a new
 // HTTP request signed by STS AWS API.
-func (s *SigningService) prepareSignedRequest(r *http.Request, re *endpoints.ResolvedEndpoint, sessionCtx *common.SessionContext) (*http.Request, error) {
+func (s *SigningService) prepareSignedRequest(r *http.Request, payload []byte, re *endpoints.ResolvedEndpoint, sessionCtx *common.SessionContext) (*http.Request, error) {
 	url, err := urlForResolvedEndpoint(r, re)
 	if err != nil {
 		return nil, trace.Wrap(err)
 	}
-	payload, err := awsutils.GetAndReplaceReqBody(r)
-	if err != nil {
-		return nil, trace.Wrap(err)
-	}
 	reqCopy, err := http.NewRequest(r.Method, url, bytes.NewReader(payload))
 	if err != nil {
 		return nil, trace.Wrap(err)