Merge in progress

Makefile

@@ -110,6 +110,15 @@ ifneq ("$(OS)", "windows")
 	zip -q -A $(BUILDDIR)/teleport
 endif
 
+#
+# make full-ent - Builds Teleport enterprise binaries
+#
+.PHONY:full-ent
+full-ent:
+ifneq ("$(OS)", "windows")
+	@if [ -f e/Makefile ]; then $(MAKE) -C e full; fi
+endif
+
 #
 # make clean - Removed all build artifacts.
 #

constants.go

@@ -51,6 +51,13 @@ const (
 	SSHSessionID = "SSH_SESSION_ID"
 )
 
+const (
+	// HTTPNextProtoTLS is the NPN/ALPN protocol negotiated during
+	// HTTP/1.1.'s TLS setup.
+	// https://www.iana.org/assignments/tls-extensiontype-values/tls-extensiontype-values.xhtml#alpn-protocol-ids
+	HTTPNextProtoTLS = "http/1.1"
+)
+
 const (
 	// HTTPSProxy is an environment variable pointing to a HTTPS proxy.
 	HTTPSProxy = "HTTPS_PROXY"

go.mod

@@ -87,7 +87,7 @@ require (
 	google.golang.org/api v0.10.0
 	google.golang.org/appengine v1.6.3 // indirect
 	google.golang.org/genproto v0.0.0-20190916214212-f660b8655731
-	google.golang.org/grpc v1.23.0
+	google.golang.org/grpc v1.23.1
 	gopkg.in/check.v1 v1.0.0-20190902080502-41f04d3bba15
 	gopkg.in/yaml.v2 v2.2.8
 	gotest.tools v2.2.0+incompatible // indirect

go.sum

@@ -471,6 +471,8 @@ google.golang.org/grpc v1.20.1/go.mod h1:10oTOabMzJvdu6/UiuZezV6QK5dSlG84ov/aaiq
 google.golang.org/grpc v1.21.1/go.mod h1:oYelfM1adQP15Ek0mdvEgi9Df8B9CZIaU1084ijfRaM=
 google.golang.org/grpc v1.23.0 h1:AzbTB6ux+okLTzP8Ru1Xs41C303zdcfEht7MQnYJt5A=
 google.golang.org/grpc v1.23.0/go.mod h1:Y5yQAOtifL1yxbo5wqy6BxZv8vAUGQwXBOALyacEbxg=
+google.golang.org/grpc v1.23.1 h1:q4XQuHFC6I28BKZpo6IYyb3mNO+l7lSOxRuYTCiDfXk=
+google.golang.org/grpc v1.23.1/go.mod h1:Y5yQAOtifL1yxbo5wqy6BxZv8vAUGQwXBOALyacEbxg=
 gopkg.in/alecthomas/kingpin.v2 v2.2.6/go.mod h1:FMv+mEhP44yOT+4EoQTLFTRgOQ1FBLkstjWtayDeSgw=
 gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0=
 gopkg.in/check.v1 v1.0.0-20180628173108-788fd7840127 h1:qIbj1fsPNlZgppZ+VLlY7N33q108Sa+fhmuc+sWQYwY=

lib/auth/api.go

@@ -1,5 +1,5 @@
 /*
-Copyright 2015 Gravitational, Inc.
+Copyright 2015-2020 Gravitational, Inc.
 
 Licensed under the Apache License, Version 2.0 (the "License");
 you may not use this file except in compliance with the License.
@@ -20,9 +20,11 @@ import (
 	"context"
 	"io"
 
-	"github.com/gravitational/trace"
-
+	"github.com/gravitational/teleport/lib/events"
 	"github.com/gravitational/teleport/lib/services"
+	"github.com/gravitational/teleport/lib/session"
+
+	"github.com/gravitational/trace"
 )
 
 // Announcer specifies interface responsible for announcing presence
@@ -102,6 +104,8 @@ type AccessPoint interface {
 	ReadAccessPoint
 	// Announcer adds methods used to announce presence
 	Announcer
+	// Streamer creates and manages audit streams
+	events.Streamer
 
 	// UpsertTunnelConnection upserts tunnel connection
 	UpsertTunnelConnection(conn services.TunnelConnection) error
@@ -159,6 +163,16 @@ type Wrapper struct {
 	Write AccessPoint
 }
 
+// ResumeAuditStream resumes existing audit stream
+func (w *Wrapper) ResumeAuditStream(ctx context.Context, sid session.ID, uploadID string) (events.Stream, error) {
+	return w.Write.ResumeAuditStream(ctx, sid, uploadID)
+}
+
+// CreateAuditStream creates new audit stream
+func (w *Wrapper) CreateAuditStream(ctx context.Context, sid session.ID) (events.Stream, error) {
+	return w.Write.CreateAuditStream(ctx, sid)
+}
+
 // Close closes all associated resources
 func (w *Wrapper) Close() error {
 	err := w.Write.Close()

lib/auth/apiserver.go

@@ -49,6 +49,23 @@ type APIConfig struct {
 	SessionService session.Service
 	AuditLog       events.IAuditLog
 	Authorizer     Authorizer
+	Emitter        events.Emitter
+	// KeepAlivePeriod defines period between keep alives
+	KeepAlivePeriod time.Duration
+	// KeepAliveCount specifies amount of missed keep alives
+	// to wait for until declaring connection as broken
+	KeepAliveCount int
+}
+
+// CheckAndSetDefaults checks and sets default values
+func (a *APIConfig) CheckAndSetDefaults() error {
+	if a.KeepAlivePeriod == 0 {
+		a.KeepAlivePeriod = defaults.ServerKeepAliveTTL
+	}
+	if a.KeepAliveCount == 0 {
+		a.KeepAliveCount = defaults.KeepAliveCountMax
+	}
+	return nil
 }
 
 // APIServer implements http API server for AuthServer interface
@@ -260,9 +277,7 @@ func (s *APIServer) withAuth(handler HandlerWithAuthFunc) httprouter.Handle {
 		}
 		auth := &AuthWithRoles{
 			authServer: s.AuthServer,
-			user:       authContext.User,
-			checker:    authContext.Checker,
-			identity:   authContext.Identity,
+			context:    *authContext,
 			sessions:   s.SessionService,
 			alog:       s.AuthServer.IAuditLog,
 		}
@@ -1869,9 +1884,9 @@ func (s *APIServer) emitAuditEvent(auth ClientI, w http.ResponseWriter, r *http.
 	// For backwards compatibility, check if the full event struct has
 	// been sent in the request or just the event type.
 	if req.Event.Name != "" {
-		err = auth.EmitAuditEvent(req.Event, req.Fields)
+		err = auth.EmitAuditEventLegacy(req.Event, req.Fields)
 	} else {
-		err = auth.EmitAuditEvent(events.Event{Name: req.Type}, req.Fields)
+		err = auth.EmitAuditEventLegacy(events.Event{Name: req.Type}, req.Fields)
 	}
 	if err != nil {
 		return nil, trace.Wrap(err)

lib/auth/auth.go

@@ -41,6 +41,7 @@ import (
 	"github.com/gravitational/teleport/lib/limiter"
 	"github.com/gravitational/teleport/lib/services"
 	"github.com/gravitational/teleport/lib/services/local"
+	"github.com/gravitational/teleport/lib/session"
 	"github.com/gravitational/teleport/lib/sshca"
 	"github.com/gravitational/teleport/lib/sshutils"
 	"github.com/gravitational/teleport/lib/tlsca"
@@ -89,6 +90,12 @@ func NewAuthServer(cfg *InitConfig, opts ...AuthServerOption) (*AuthServer, erro
 	if cfg.AuditLog == nil {
 		cfg.AuditLog = events.NewDiscardAuditLog()
 	}
+	if cfg.Emitter == nil {
+		cfg.Emitter = events.NewDiscardEmitter()
+	}
+	if cfg.Streamer == nil {
+		cfg.Streamer = events.NewDiscardEmitter()
+	}
 
 	limiter, err := limiter.NewConnectionsLimiter(limiter.LimiterConfig{
 		MaxConnections: defaults.LimiterMaxConcurrentSignatures,
@@ -109,6 +116,8 @@ func NewAuthServer(cfg *InitConfig, opts ...AuthServerOption) (*AuthServer, erro
 		caSigningAlg:    cfg.CASigningAlg,
 		cancelFunc:      cancelFunc,
 		closeCtx:        closeCtx,
+		emitter:         cfg.Emitter,
+		streamer:        cfg.Streamer,
 		AuthServices: AuthServices{
 			Trust:                cfg.Trust,
 			Presence:             cfg.Presence,
@@ -217,6 +226,13 @@ type AuthServer struct {
 	cache AuthCache
 
 	limiter *limiter.ConnectionsLimiter
+
+	// Emitter is events emitter, used to submit discrete events
+	emitter events.Emitter
+
+	// streamer is events sessionstreamer, used to create continuous
+	// session related streams
+	streamer events.Streamer
 }
 
 // SetCache sets cache used by auth server
@@ -554,9 +570,9 @@ func (s *AuthServer) generateUserCert(req certRequest) (*certs, error) {
 		Principals:       allowedLogins,
 		Usage:            req.usage,
 		RouteToCluster:   req.routeToCluster,
+		Traits:           req.traits,
 		KubernetesGroups: kubeGroups,
 		KubernetesUsers:  kubeUsers,
-		Traits:           req.traits,
 	}
 	subject, err := identity.Subject()
 	if err != nil {
@@ -643,7 +659,7 @@ func (s *AuthServer) WithUserLock(username string, authenticateFn func() error)
 
 // PreAuthenticatedSignIn is for 2-way authentication methods like U2F where the password is
 // already checked before issuing the second factor challenge
-func (s *AuthServer) PreAuthenticatedSignIn(user string, identity *tlsca.Identity) (services.WebSession, error) {
+func (s *AuthServer) PreAuthenticatedSignIn(user string, identity tlsca.Identity) (services.WebSession, error) {
 	roles, traits, err := services.ExtractFromIdentity(s, identity)
 	if err != nil {
 		return nil, trace.Wrap(err)
@@ -735,7 +751,7 @@ func (s *AuthServer) CheckU2FSignResponse(user string, response *u2f.SignRespons
 
 // ExtendWebSession creates a new web session for a user based on a valid previous sessionID,
 // method is used to renew the web session for a user
-func (s *AuthServer) ExtendWebSession(user string, prevSessionID string, identity *tlsca.Identity) (services.WebSession, error) {
+func (s *AuthServer) ExtendWebSession(user string, prevSessionID string, identity tlsca.Identity) (services.WebSession, error) {
 	prevSession, err := s.GetWebSession(user, prevSessionID)
 	if err != nil {
 		return nil, trace.Wrap(err)
@@ -837,10 +853,16 @@ func (a *AuthServer) GenerateToken(ctx context.Context, req GenerateTokenRequest
 	user := clientUsername(ctx)
 	for _, role := range req.Roles {
 		if role == teleport.RoleTrustedCluster {
-			if err := a.EmitAuditEvent(events.TrustedClusterTokenCreate, events.EventFields{
-				events.EventUser: user,
+			if err := a.emitter.EmitAuditEvent(ctx, &events.TrustedClusterTokenCreate{
+				Metadata: events.Metadata{
+					Type: events.TrustedClusterTokenCreateEvent,
+					Code: events.TrustedClusterTokenCreateCode,
+				},
+				UserMetadata: events.UserMetadata{
+					User: user,
+				},
 			}); err != nil {
-				log.Warnf("Failed to emit trusted cluster token create event: %v", err)
+				log.WithError(err).Warn("Failed to emit trusted cluster token create event.")
 			}
 		}
 	}
@@ -1397,11 +1419,20 @@ func (a *AuthServer) DeleteRole(ctx context.Context, name string) error {
 		return trace.Wrap(err)
 	}
 
-	if err := a.EmitAuditEvent(events.RoleDeleted, events.EventFields{
-		events.FieldName: name,
-		events.EventUser: clientUsername(ctx),
-	}); err != nil {
-		log.Warnf("Failed to emit role deleted event: %v", err)
+	err = a.emitter.EmitAuditEvent(a.closeCtx, &events.RoleDelete{
+		Metadata: events.Metadata{
+			Type: events.RoleDeletedEvent,
+			Code: events.RoleDeletedCode,
+		},
+		UserMetadata: events.UserMetadata{
+			User: clientUsername(ctx),
+		},
+		ResourceMetadata: events.ResourceMetadata{
+			Name: name,
+		},
+	})
+	if err != nil {
+		log.WithError(err).Warnf("Failed to emit role deleted event.")
 	}
 
 	return nil
@@ -1413,13 +1444,21 @@ func (a *AuthServer) upsertRole(ctx context.Context, role services.Role) error {
 		return trace.Wrap(err)
 	}
 
-	if err := a.EmitAuditEvent(events.RoleCreated, events.EventFields{
-		events.FieldName: role.GetName(),
-		events.EventUser: clientUsername(ctx),
-	}); err != nil {
-		log.Warnf("Failed to emit role created event: %v", err)
+	err := a.emitter.EmitAuditEvent(a.closeCtx, &events.RoleCreate{
+		Metadata: events.Metadata{
+			Type: events.RoleCreatedEvent,
+			Code: events.RoleCreatedCode,
+		},
+		UserMetadata: events.UserMetadata{
+			User: clientUsername(ctx),
+		},
+		ResourceMetadata: events.ResourceMetadata{
+			Name: role.GetName(),
+		},
+	})
+	if err != nil {
+		log.WithError(err).Warnf("Failed to emit role create event.")
 	}
-
 	return nil
 }
 
@@ -1451,11 +1490,17 @@ func (a *AuthServer) CreateAccessRequest(ctx context.Context, req services.Acces
 	if err := a.DynamicAccess.CreateAccessRequest(ctx, req); err != nil {
 		return trace.Wrap(err)
 	}
-	err = a.EmitAuditEvent(events.AccessRequestCreated, events.EventFields{
-		events.AccessRequestID:    req.GetName(),
-		events.EventUser:          req.GetUser(),
-		events.UserRoles:          req.GetRoles(),
-		events.AccessRequestState: req.GetState().String(),
+	err = a.emitter.EmitAuditEvent(a.closeCtx, &events.AccessRequestCreate{
+		Metadata: events.Metadata{
+			Type: events.AccessRequestCreateEvent,
+			Code: events.AccessRequestCreateCode,
+		},
+		UserMetadata: events.UserMetadata{
+			User: req.GetUser(),
+		},
+		Roles:        req.GetRoles(),
+		RequestID:    req.GetName(),
+		RequestState: req.GetState().String(),
 	})
 	return trace.Wrap(err)
 }
@@ -1464,15 +1509,24 @@ func (a *AuthServer) SetAccessRequestState(ctx context.Context, reqID string, st
 	if err := a.DynamicAccess.SetAccessRequestState(ctx, reqID, state); err != nil {
 		return trace.Wrap(err)
 	}
-	fields := events.EventFields{
-		events.AccessRequestID:    reqID,
-		events.AccessRequestState: state.String(),
-		events.UpdatedBy:          clientUsername(ctx),
+	event := &events.AccessRequestCreate{
+		Metadata: events.Metadata{
+			Type: events.AccessRequestUpdateEvent,
+			Code: events.AccessRequestUpdateCode,
+		},
+		ResourceMetadata: events.ResourceMetadata{
+			UpdatedBy: clientUsername(ctx),
+		},
+		RequestID:    reqID,
+		RequestState: state.String(),
 	}
 	if delegator := getDelegator(ctx); delegator != "" {
-		fields[events.AccessRequestDelegator] = delegator
+		event.Delegator = delegator
+	}
+	err := a.emitter.EmitAuditEvent(a.closeCtx, event)
+	if err != nil {
+		log.WithError(err).Warn("Failed to emit access request update event.")
 	}
-	err := a.EmitAuditEvent(events.AccessRequestUpdated, fields)
 	return trace.Wrap(err)
 }
 
@@ -1588,6 +1642,41 @@ func (a *AuthServer) GetAllTunnelConnections(opts ...services.MarshalOption) (co
 	return a.GetCache().GetAllTunnelConnections(opts...)
 }
 
+// CreateAuditStream creates audit event stream
+func (a *AuthServer) CreateAuditStream(ctx context.Context, sid session.ID) (events.Stream, error) {
+	streamer, err := a.modeStreamer()
+	if err != nil {
+		return nil, trace.Wrap(err)
+	}
+	return streamer.CreateAuditStream(ctx, sid)
+}
+
+// ResumeAuditStream resumes the stream that has been created
+func (a *AuthServer) ResumeAuditStream(ctx context.Context, sid session.ID, uploadID string) (events.Stream, error) {
+	streamer, err := a.modeStreamer()
+	if err != nil {
+		return nil, trace.Wrap(err)
+	}
+	return streamer.ResumeAuditStream(ctx, sid, uploadID)
+}
+
+// modeStreamer creates streamer based on the event mode
+func (a *AuthServer) modeStreamer() (events.Streamer, error) {
+	clusterConfig, err := a.GetClusterConfig()
+	if err != nil {
+		return nil, trace.Wrap(err)
+	}
+	mode := clusterConfig.GetSessionRecording()
+	// In sync mode, some events are fan-out to the events
+	// database and all the others are stored within a session
+	if services.IsRecordSync(mode) {
+		return events.NewTeeStreamer(a.streamer, a.emitter), nil
+	}
+	// in async mode, some events have been already submitted
+	// during the session, so submitting those events are not necessary
+	return a.streamer, nil
+}
+
 // authKeepAliver is a keep aliver using auth server directly
 type authKeepAliver struct {
 	sync.RWMutex