Capture postgres extended protocol queries in audit log

api/constants/constants.go

@@ -91,6 +91,12 @@ const (
 
 	// DarwinOS is the GOOS constant for Apple macOS/darwin.
 	DarwinOS = "darwin"
+
+	// UseOfClosedNetworkConnection is a special string some parts of
+	// go standard lib are using that is the only way to identify some errors
+	//
+	// TODO(r0mant): See if we can use net.ErrClosed and errors.Is() instead.
+	UseOfClosedNetworkConnection = "use of closed network connection"
 )
 
 // SecondFactorType is the type of 2FA authentication.

api/types/events/events.proto

@@ -1171,6 +1171,9 @@ message DatabaseSessionQuery {
         [ (gogoproto.nullable) = false, (gogoproto.embed) = true, (gogoproto.jsontag) = "" ];
     // DatabaseQuery is the executed query string.
     string DatabaseQuery = 5 [ (gogoproto.jsontag) = "db_query" ];
+    // DatabaseQueryParameters are the query parameters for prepared statements.
+    repeated string DatabaseQueryParameters = 6
+        [ (gogoproto.jsontag) = "db_query_parameters,omitempty" ];
 }
 
 // DatabaseSessionEnd is emitted when a user ends the database session.

api/utils/sshutils/chconn.go

@@ -22,6 +22,8 @@ import (
 	"sync"
 	"time"
 
+	"github.com/gravitational/teleport/api/constants"
+
 	"github.com/gravitational/trace"
 	"golang.org/x/crypto/ssh"
 )
@@ -111,7 +113,16 @@ func (c *ChConn) RemoteAddr() net.Addr {
 
 // Read reads from the channel.
 func (c *ChConn) Read(data []byte) (int, error) {
-	return c.reader.Read(data)
+	n, err := c.reader.Read(data)
+	// A lot of code relies on "use of closed network connection" error to
+	// gracefully handle terminated connections so convert the closed pipe
+	// error to it.
+	if err != nil && err == io.ErrClosedPipe {
+		return n, trace.ConnectionProblem(err, constants.UseOfClosedNetworkConnection)
+	}
+	// Do not wrap the error to avoid masking the underlying error such as
+	// timeout error which is returned when read deadline is exceeded.
+	return n, err
 }
 
 // SetDeadline sets a connection deadline.

constants.go

@@ -26,25 +26,26 @@ import (
 // The following constants have been moved to /api/constants/constants.go, and are now
 // imported here for backwards compatibility. DELETE IN 7.0.0
 const (
-	Local                      = constants.Local
-	OIDC                       = constants.OIDC
-	SAML                       = constants.SAML
-	Github                     = constants.Github
-	HumanDateFormatSeconds     = constants.HumanDateFormatSeconds
-	DefaultImplicitRole        = constants.DefaultImplicitRole
-	APIDomain                  = constants.APIDomain
-	CertificateFormatStandard  = constants.CertificateFormatStandard
-	DurationNever              = constants.DurationNever
-	EnhancedRecordingMinKernel = constants.EnhancedRecordingMinKernel
-	EnhancedRecordingCommand   = constants.EnhancedRecordingCommand
-	EnhancedRecordingDisk      = constants.EnhancedRecordingDisk
-	EnhancedRecordingNetwork   = constants.EnhancedRecordingNetwork
-	KeepAliveNode              = constants.KeepAliveNode
-	KeepAliveApp               = constants.KeepAliveApp
-	KeepAliveDatabase          = constants.KeepAliveDatabase
-	WindowsOS                  = constants.WindowsOS
-	LinuxOS                    = constants.LinuxOS
-	DarwinOS                   = constants.DarwinOS
+	Local                        = constants.Local
+	OIDC                         = constants.OIDC
+	SAML                         = constants.SAML
+	Github                       = constants.Github
+	HumanDateFormatSeconds       = constants.HumanDateFormatSeconds
+	DefaultImplicitRole          = constants.DefaultImplicitRole
+	APIDomain                    = constants.APIDomain
+	CertificateFormatStandard    = constants.CertificateFormatStandard
+	DurationNever                = constants.DurationNever
+	EnhancedRecordingMinKernel   = constants.EnhancedRecordingMinKernel
+	EnhancedRecordingCommand     = constants.EnhancedRecordingCommand
+	EnhancedRecordingDisk        = constants.EnhancedRecordingDisk
+	EnhancedRecordingNetwork     = constants.EnhancedRecordingNetwork
+	KeepAliveNode                = constants.KeepAliveNode
+	KeepAliveApp                 = constants.KeepAliveApp
+	KeepAliveDatabase            = constants.KeepAliveDatabase
+	WindowsOS                    = constants.WindowsOS
+	LinuxOS                      = constants.LinuxOS
+	DarwinOS                     = constants.DarwinOS
+	UseOfClosedNetworkConnection = constants.UseOfClosedNetworkConnection
 )
 
 // WebAPIVersion is a current webapi version
@@ -647,10 +648,6 @@ const (
 )
 
 const (
-	// UseOfClosedNetworkConnection is a special string some parts of
-	// go standard lib are using that is the only way to identify some errors
-	UseOfClosedNetworkConnection = "use of closed network connection"
-
 	// NodeIsAmbiguous serves as an identifying error string indicating that
 	// the proxy subsystem found multiple nodes matching the specified hostname.
 	NodeIsAmbiguous = "err-node-is-ambiguous"

lib/events/api.go

@@ -355,11 +355,14 @@ const (
 	// AppSessionRequestEvent is an HTTP request and response.
 	AppSessionRequestEvent = "app.session.request"
 
-	// DatabaseSessionStartEvent indicates the start of a database session.
+	// DatabaseSessionStartEvent is emitted when a database client attempts
+	// to connect to a database.
 	DatabaseSessionStartEvent = "db.session.start"
-	// DatabaseSessionEndEvent indicates the end of a database session.
+	// DatabaseSessionEndEvent is emitted when a database client disconnects
+	// from a database.
 	DatabaseSessionEndEvent = "db.session.end"
-	// DatabaseSessionQueryEvent indicates a database query execution.
+	// DatabaseSessionQueryEvent is emitted when a database client executes
+	// a query.
 	DatabaseSessionQueryEvent = "db.session.query"
 
 	// SessionRejectedReasonMaxConnections indicates that a session.rejected event

lib/events/filesessions/filestream.go

@@ -119,12 +119,12 @@ func (h *Handler) CompleteUpload(ctx context.Context, upload events.StreamUpload
 		return trace.Wrap(err)
 	}
 	defer func() {
-		if err := f.Close(); err != nil {
-			h.WithError(err).Errorf("Failed to close file %q.", uploadPath)
-		}
 		if err := utils.FSUnlock(f); err != nil {
 			h.WithError(err).Errorf("Failed to unlock filesystem lock.")
 		}
+		if err := f.Close(); err != nil {
+			h.WithError(err).Errorf("Failed to close file %q.", uploadPath)
+		}
 	}()
 
 	files := make([]*os.File, 0, len(parts))

lib/srv/db/db_test.go → lib/srv/db/access_test.go

@@ -1,5 +1,5 @@
 /*
-Copyright 2020 Gravitational, Inc.
+Copyright 2020-2021 Gravitational, Inc.
 
 Licensed under the Apache License, Version 2.0 (the "License");
 you may not use this file except in compliance with the License.
@@ -39,6 +39,7 @@ import (
 	"github.com/jackc/pgconn"
 	"github.com/jonboulle/clockwork"
 	"github.com/pborman/uuid"
+	"github.com/siddontang/go-mysql/client"
 	"github.com/stretchr/testify/require"
 )
 
@@ -53,19 +54,7 @@ func TestPostgresAccess(t *testing.T) {
 	ctx := context.Background()
 	testCtx := setupTestContext(ctx, t)
 	t.Cleanup(func() { testCtx.Close() })
-
-	// Start multiplexer.
-	go testCtx.mux.Serve()
-	// Start fake Postgres server.
-	go testCtx.postgresServer.Serve()
-	// Start database proxy server.
-	go testCtx.proxyServer.Serve(testCtx.mux.DB())
-	// Start database service server.
-	go func() {
-		for conn := range testCtx.proxyConn {
-			testCtx.server.HandleConnection(conn)
-		}
-	}()
+	go testCtx.startHandlingPostgresConnections()
 
 	tests := []struct {
 		desc         string
@@ -149,19 +138,7 @@ func TestPostgresAccess(t *testing.T) {
 			require.NoError(t, err)
 
 			// Try to connect to the database as this user.
-			pgConn, err := postgres.MakeTestClient(ctx, common.TestClientConfig{
-				AuthClient: testCtx.authClient,
-				AuthServer: testCtx.authServer,
-				Address:    testCtx.mux.DB().Addr().String(),
-				Cluster:    testCtx.clusterName,
-				Username:   test.user,
-				RouteToDatabase: tlsca.RouteToDatabase{
-					ServiceName: "postgres-test",
-					Protocol:    defaults.ProtocolPostgres,
-					Username:    test.dbUser,
-					Database:    test.dbName,
-				},
-			})
+			pgConn, err := testCtx.postgresClient(ctx, test.user, test.dbUser, test.dbName)
 			if test.err != "" {
 				require.Error(t, err)
 				require.Contains(t, err.Error(), test.err)
@@ -188,17 +165,7 @@ func TestMySQLAccess(t *testing.T) {
 	ctx := context.Background()
 	testCtx := setupTestContext(ctx, t)
 	t.Cleanup(func() { testCtx.Close() })
-
-	// Start test MySQL server.
-	go testCtx.mysqlServer.Serve()
-	// Start MySQL proxy server.
-	go testCtx.proxyServer.ServeMySQL(testCtx.mysqlListener)
-	// Start database service server.
-	go func() {
-		for conn := range testCtx.proxyConn {
-			testCtx.server.HandleConnection(conn)
-		}
-	}()
+	go testCtx.startHandlingMySQLConnections()
 
 	tests := []struct {
 		// desc is the test case description.
@@ -257,18 +224,7 @@ func TestMySQLAccess(t *testing.T) {
 			require.NoError(t, err)
 
 			// Try to connect to the database as this user.
-			mysqlConn, err := mysql.MakeTestClient(common.TestClientConfig{
-				AuthClient: testCtx.authClient,
-				AuthServer: testCtx.authServer,
-				Address:    testCtx.mysqlListener.Addr().String(),
-				Cluster:    testCtx.clusterName,
-				Username:   test.user,
-				RouteToDatabase: tlsca.RouteToDatabase{
-					ServiceName: "mysql-test",
-					Protocol:    defaults.ProtocolPostgres,
-					Username:    test.dbUser,
-				},
-			})
+			mysqlConn, err := testCtx.mysqlClient(test.user, test.dbUser)
 			if test.err != "" {
 				require.Error(t, err)
 				require.Contains(t, err.Error(), test.err)
@@ -309,19 +265,7 @@ func TestDatabaseAccessDisabled(t *testing.T) {
 	ctx := context.Background()
 	testCtx := setupTestContext(ctx, t)
 	t.Cleanup(func() { testCtx.Close() })
-
-	// Start multiplexer.
-	go testCtx.mux.Serve()
-	// Start fake Postgres server.
-	go testCtx.postgresServer.Serve()
-	// Start database proxy server.
-	go testCtx.proxyServer.Serve(testCtx.mux.DB())
-	// Start database service server.
-	go func() {
-		for conn := range testCtx.proxyConn {
-			testCtx.server.HandleConnection(conn)
-		}
-	}()
+	go testCtx.startHandlingPostgresConnections()
 
 	userName := "alice"
 	roleName := "admin"
@@ -338,19 +282,7 @@ func TestDatabaseAccessDisabled(t *testing.T) {
 	require.NoError(t, err)
 
 	// Try to connect to the database as this user.
-	_, err = postgres.MakeTestClient(ctx, common.TestClientConfig{
-		AuthClient: testCtx.authClient,
-		AuthServer: testCtx.authServer,
-		Address:    testCtx.mux.DB().Addr().String(),
-		Cluster:    testCtx.clusterName,
-		Username:   userName,
-		RouteToDatabase: tlsca.RouteToDatabase{
-			ServiceName: "postgres-test",
-			Protocol:    defaults.ProtocolPostgres,
-			Username:    dbUser,
-			Database:    dbName,
-		},
-	})
+	_, err = testCtx.postgresClient(ctx, userName, dbUser, dbName)
 	require.Error(t, err)
 	require.Contains(t, err.Error(), "this Teleport cluster doesn't support database access")
 }
@@ -369,6 +301,66 @@ type testContext struct {
 	server           *Server
 	postgresDBServer types.DatabaseServer
 	mysqlDBServer    types.DatabaseServer
+	emitter          *testEmitter
+}
+
+func (c *testContext) startHandlingPostgresConnections() {
+	// Start multiplexer.
+	go c.mux.Serve()
+	// Start fake Postgres server.
+	go c.postgresServer.Serve()
+	// Start database proxy server.
+	go c.proxyServer.Serve(c.mux.DB())
+	// Start handling Postgres connection on the database server.
+	for conn := range c.proxyConn {
+		c.server.HandleConnection(conn)
+	}
+}
+
+// postgresClient connects to test Postgres through database access as a
+// specified Teleport user and database account.
+func (c *testContext) postgresClient(ctx context.Context, teleportUser, dbUser, dbName string) (*pgconn.PgConn, error) {
+	return postgres.MakeTestClient(ctx, common.TestClientConfig{
+		AuthClient: c.authClient,
+		AuthServer: c.authServer,
+		Address:    c.mux.DB().Addr().String(),
+		Cluster:    c.clusterName,
+		Username:   teleportUser,
+		RouteToDatabase: tlsca.RouteToDatabase{
+			ServiceName: c.postgresDBServer.GetName(),
+			Protocol:    defaults.ProtocolPostgres,
+			Username:    dbUser,
+			Database:    dbName,
+		},
+	})
+}
+
+func (c *testContext) startHandlingMySQLConnections() {
+	// Start test MySQL server.
+	go c.mysqlServer.Serve()
+	// Start MySQL proxy server.
+	go c.proxyServer.ServeMySQL(c.mysqlListener)
+	// Start handling MySQL connections on the database server.
+	for conn := range c.proxyConn {
+		c.server.HandleConnection(conn)
+	}
+}
+
+// mysqlClient connects to test MySQL through database access as a specified
+// Teleport user and database account.
+func (c *testContext) mysqlClient(teleportUser, dbUser string) (*client.Conn, error) {
+	return mysql.MakeTestClient(common.TestClientConfig{
+		AuthClient: c.authClient,
+		AuthServer: c.authServer,
+		Address:    c.mysqlListener.Addr().String(),
+		Cluster:    c.clusterName,
+		Username:   teleportUser,
+		RouteToDatabase: tlsca.RouteToDatabase{
+			ServiceName: c.mysqlDBServer.GetName(),
+			Protocol:    defaults.ProtocolMySQL,
+			Username:    dbUser,
+		},
+	})
 }
 
 // Close closes all resources associated with the test context.
@@ -487,6 +479,9 @@ func setupTestContext(ctx context.Context, t *testing.T) *testContext {
 	})
 	require.NoError(t, err)
 
+	// Create test audit events emitter.
+	emitter := newTestEmitter()
+
 	// Create database service server.
 	server, err := New(ctx, Config{
 		Clock:         clockwork.NewFakeClockAt(time.Now()),
@@ -498,6 +493,13 @@ func setupTestContext(ctx context.Context, t *testing.T) *testContext {
 		Servers:       []types.DatabaseServer{postgresDBServer, mysqlDBServer},
 		TLSConfig:     tlsConfig,
 		GetRotation:   func(teleport.Role) (*types.Rotation, error) { return &types.Rotation{}, nil },
+		NewAudit: func(common.AuditConfig) (common.Audit, error) {
+			// Use the same audit logger implementation but substitute the
+			// underlying emitter so events can be tracked in tests.
+			return common.NewAudit(common.AuditConfig{
+				Emitter: emitter,
+			})
+		},
 	})
 	require.NoError(t, err)
 
@@ -515,6 +517,7 @@ func setupTestContext(ctx context.Context, t *testing.T) *testContext {
 		tlsServer:        tlsServer,
 		authServer:       tlsServer.Auth(),
 		authClient:       dbAuthClient,
+		emitter:          emitter,
 	}
 }