emit session upload event

This commit is contained in:
Jane Quintero
2021-02-19 17:29:11 -08:00
committed by Russell Jones
parent 0f2ff3b557
commit 9b26bfe821
23 changed files with 890 additions and 314 deletions
File diff suppressed because it is too large Load Diff
+17
View File
@@ -1219,6 +1219,7 @@ message OneOf {
events.DatabaseSessionStart DatabaseSessionStart = 40;
events.DatabaseSessionEnd DatabaseSessionEnd = 41;
events.DatabaseSessionQuery DatabaseSessionQuery = 42;
events.SessionUpload SessionUpload = 43;
}
}
@@ -1232,3 +1233,19 @@ message StreamStatus {
google.protobuf.Timestamp LastUploadTime = 3
[ (gogoproto.stdtime) = true, (gogoproto.nullable) = false ];
}
// SessionUpload is a session upload
message SessionUpload {
// Metadata is a common event metadata
Metadata Metadata = 1
[ (gogoproto.nullable) = false, (gogoproto.embed) = true, (gogoproto.jsontag) = "" ];
// SessionMetadata is a common event session metadata
SessionMetadata SessionMetadata = 2
[ (gogoproto.nullable) = false, (gogoproto.embed) = true, (gogoproto.jsontag) = "" ];
// ID is a unique event identifier
string UID = 4 [ (gogoproto.jsontag) = "uid,omitempty" ];
// URL is where the url the session event data upload is at
string SessionURL = 5 [ (gogoproto.jsontag) = "url" ];
}
+6
View File
@@ -201,6 +201,10 @@ func ToOneOf(in AuditEvent) (*OneOf, error) {
out.Event = &OneOf_DatabaseSessionQuery{
DatabaseSessionQuery: e,
}
case *SessionUpload:
out.Event = &OneOf_SessionUpload{
SessionUpload: e,
}
default:
return nil, trace.BadParameter("event type %T is not supported", in)
}
@@ -293,6 +297,8 @@ func FromOneOf(in OneOf) (AuditEvent, error) {
return e, nil
} else if e := in.GetDatabaseSessionQuery(); e != nil {
return e, nil
} else if e := in.GetSessionUpload(); e != nil {
return e, nil
} else {
if in.Event == nil {
return nil, trace.BadParameter("failed to parse event, session record is corrupted")
+3
View File
@@ -214,6 +214,9 @@ func (w *Wrapper) CreateAuditStream(ctx context.Context, sid session.ID) (events
return w.NoCache.CreateAuditStream(ctx, sid)
}
func (w *Wrapper) GetUploadMetadata(sid session.ID) events.UploadMetadata {
return events.UploadMetadata{}
}
// Close closes all associated resources
func (w *Wrapper) Close() error {
err := w.NoCache.Close()
+2
View File
@@ -56,6 +56,8 @@ type APIConfig struct {
// KeepAliveCount specifies amount of missed keep alives
// to wait for until declaring connection as broken
KeepAliveCount int
// metadataGetter gets upload metadata
MetadataGetter events.UploadMetadataGetter
}
// CheckAndSetDefaults checks and sets default values
+5
View File
@@ -1923,6 +1923,11 @@ func (a *Server) ResumeAuditStream(ctx context.Context, sid session.ID, uploadID
return streamer.ResumeAuditStream(ctx, sid, uploadID)
}
// GetUploadMetadata gets the session upload metadata
func (a *Server) GetUploadMetadata(sid session.ID) events.UploadMetadata {
return events.UploadMetadata{}
}
// modeStreamer creates streamer based on the event mode
func (a *Server) modeStreamer() (events.Streamer, error) {
clusterConfig, err := a.GetClusterConfig()
+4
View File
@@ -2150,6 +2150,10 @@ func (c *Client) ResumeAuditStream(ctx context.Context, sid session.ID, uploadID
func (c *Client) CreateAuditStream(ctx context.Context, sid session.ID) (events.Stream, error) {
return c.APIClient.CreateAuditStream(ctx, string(sid))
}
// GetUploadMetadata gets session upload metadata
func (c *Client) GetUploadMetadata(sid session.ID) events.UploadMetadata {
return events.UploadMetadata{}
}
// WebService implements features used by Web UI clients
type WebService interface {
+26
View File
@@ -108,6 +108,7 @@ func (g *GRPCServer) CreateAuditStream(stream proto.AuthService_CreateAuditStrea
}
var eventStream events.Stream
var sessionID session.ID
g.Debugf("CreateAuditStream connection from %v.", auth.User.GetName())
streamStart := time.Now()
processed := int64(0)
@@ -150,6 +151,7 @@ func (g *GRPCServer) CreateAuditStream(stream proto.AuthService_CreateAuditStrea
if err != nil {
return trace.Wrap(err)
}
sessionID = session.ID(create.SessionID)
g.Debugf("Created stream: %v.", err)
go forwardEvents(eventStream)
defer closeStream(eventStream)
@@ -171,6 +173,30 @@ func (g *GRPCServer) CreateAuditStream(stream proto.AuthService_CreateAuditStrea
// do not use stream context to give the auth server finish the upload
// even if the stream's context is cancelled
err := eventStream.Complete(auth.CloseContext())
if err != nil {
return trail.ToGRPC(err)
}
h, err := auth.GetClusterName()
if err != nil {
return trail.ToGRPC(err)
}
sessionData := g.APIConfig.MetadataGetter.GetUploadMetadata(sessionID)
session := &events.SessionUpload{
Metadata: events.Metadata{
Type: events.SessionUploadEvent,
Code: events.SessionUploadCode,
Index: events.SessionUploadIndex,
ClusterName: h.GetClusterName(),
},
SessionMetadata: events.SessionMetadata{
SessionID: string(sessionData.SessionID),
},
SessionURL: sessionData.URL,
}
if err := g.Emitter.EmitAuditEvent(auth.CloseContext(), session); err != nil {
return trail.ToGRPC(err)
}
g.Debugf("Completed stream: %v.", err)
if err != nil {
return trail.ToGRPC(err)
+23 -2
View File
@@ -439,6 +439,8 @@ type Streamer interface {
// ResumeAuditStream resumes the stream for session upload that
// has not been completed yet.
ResumeAuditStream(ctx context.Context, sid session.ID, uploadID string) (Stream, error)
UploadMetadataGetter
}
// StreamPart represents uploaded stream part
@@ -489,6 +491,22 @@ type MultipartUploader interface {
// ListUploads lists uploads that have been initiated but not completed with
// earlier uploads returned first
ListUploads(ctx context.Context) ([]StreamUpload, error)
// GetUploadMetadata gets the upload metadata
GetUploadMetadata(sessionID session.ID) UploadMetadata
}
// UploadMetadata contains data about the session upload
type UploadMetadata struct {
// URL is the url at which the session recording is located
// it is free-form and uploader-specific
URL string
// SessionID is the event session ID
SessionID session.ID
}
// UploadMetadataGetter gets the metadata for session upload
type UploadMetadataGetter interface {
GetUploadMetadata(sid session.ID) UploadMetadata
}
// StreamWriter implements io.Writer to be plugged into the multi-writer
@@ -516,6 +534,9 @@ type IAuditLog interface {
// DELETE IN: 5.0.0
EmitAuditEventLegacy(Event, EventFields) error
// EmitAuditEvent emits audit event
EmitAuditEvent(context.Context, AuditEvent) error
// DELETE IN: 2.7.0
// This method is no longer necessary as nodes and proxies >= 2.7.0
// use UploadSessionRecording method.
@@ -602,7 +623,7 @@ func (f EventFields) GetString(key string) string {
return v
}
// GetString returns an int representation of a logged field
// GetInt returns an int representation of a logged field
func (f EventFields) GetInt(key string) int {
val, found := f[key]
if !found {
@@ -618,7 +639,7 @@ func (f EventFields) GetInt(key string) int {
return v
}
// GetString returns an int representation of a logged field
// GetTime returns an int representation of a logged field
func (f EventFields) GetTime(key string) time.Time {
val, found := f[key]
if !found {
+12 -4
View File
@@ -35,7 +35,6 @@ import (
"github.com/gravitational/teleport/lib/defaults"
"github.com/gravitational/teleport/lib/session"
"github.com/gravitational/teleport/lib/utils"
"github.com/gravitational/trace"
"github.com/jonboulle/clockwork"
"github.com/prometheus/client_golang/prometheus"
@@ -174,7 +173,7 @@ type AuditLogConfig struct {
// UploadHandler is a pluggable external upload handler,
// used to fetch sessions from external sources
UploadHandler UploadHandler
UploadHandler MultipartHandler
// ExternalLog is a pluggable external log service
ExternalLog IAuditLog
@@ -233,7 +232,7 @@ func (a *AuditLogConfig) CheckAndSetDefaults() error {
return nil
}
// Creates and returns a new Audit Log object whish will store its logfiles in
// NewAuditLog creates and returns a new Audit Log object whish will store its logfiles in
// a given directory. Session recording can be disabled by setting
// recordSessions to false.
func NewAuditLog(cfg AuditLogConfig) (*AuditLog, error) {
@@ -957,7 +956,16 @@ func (l *AuditLog) fetchSessionEvents(fileName string, afterN int) ([]EventField
// EmitAuditEvent adds a new event to the local file log
func (l *AuditLog) EmitAuditEvent(ctx context.Context, event AuditEvent) error {
err := l.localLog.EmitAuditEvent(ctx, event)
// If an external logger has been set, use it as the emitter, otherwise
// fallback to the local disk based emitter.
var emitAuditEvent func(ctx context.Context, event AuditEvent) error
if l.ExternalLog != nil {
emitAuditEvent = l.ExternalLog.EmitAuditEvent
} else {
emitAuditEvent = l.localLog.EmitAuditEvent
}
err := emitAuditEvent(ctx, event)
if err != nil {
auditFailedEmit.Inc()
return trace.Wrap(err)
+2 -1
View File
@@ -457,7 +457,8 @@ func (a *AuditWriter) closeStream(stream Stream) {
func (a *AuditWriter) completeStream(stream Stream) {
ctx, cancel := context.WithTimeout(a.cfg.Context, defaults.NetworkBackoffDuration)
defer cancel()
if err := stream.Complete(ctx); err != nil {
err := stream.Complete(ctx)
if err != nil {
a.log.WithError(err).Warning("Failed to complete stream.")
}
}
+18 -1
View File
@@ -22,7 +22,6 @@ import (
"github.com/gravitational/teleport"
"github.com/gravitational/teleport/lib/defaults"
"github.com/gravitational/trace"
"github.com/jonboulle/clockwork"
log "github.com/sirupsen/logrus"
@@ -30,6 +29,8 @@ import (
// UploadCompleterConfig specifies configuration for the uploader
type UploadCompleterConfig struct {
// AuditLog is used for storing logs
AuditLog IAuditLog
// Uploader allows the completer to list and complete uploads
Uploader MultipartUploader
// GracePeriod is the period after which uploads are considered
@@ -137,6 +138,22 @@ func (u *UploadCompleter) CheckUploads(ctx context.Context) error {
}
u.log.Debugf("Completed upload %v.", upload)
completed++
uploadData := u.cfg.Uploader.GetUploadMetadata(upload.SessionID)
session := &SessionUpload{
Metadata: Metadata{
Type: SessionUploadEvent,
Code: SessionUploadCode,
Index: SessionUploadIndex,
},
SessionMetadata: SessionMetadata{
SessionID: string(uploadData.SessionID),
},
SessionURL: uploadData.URL,
}
err = u.cfg.AuditLog.EmitAuditEvent(ctx, session)
if err != nil {
return trace.Wrap(err)
}
}
if completed > 0 {
u.log.Debugf("Found %v active uploads, completed %v.", len(uploads), completed)
+3
View File
@@ -61,3 +61,6 @@ func (d *DiscardAuditLog) SearchSessionEvents(fromUTC time.Time, toUTC time.Time
func (d *DiscardAuditLog) UploadSessionRecording(SessionRecording) error {
return nil
}
func (d *DiscardAuditLog) EmitAuditEvent(ctx context.Context, event AuditEvent) error {
return nil
}
+24
View File
@@ -251,6 +251,11 @@ func (*DiscardEmitter) ResumeAuditStream(ctx context.Context, sid session.ID, up
return &DiscardStream{}, nil
}
// GetUploadMetadata does nothing
func (d *DiscardEmitter) GetUploadMetadata(sid session.ID) UploadMetadata {
return UploadMetadata{}
}
// NewWriterEmitter returns a new instance of emitter writing to writer
func NewWriterEmitter(w io.WriteCloser) *WriterEmitter {
return &WriterEmitter{
@@ -411,6 +416,10 @@ func (s *CheckingStreamer) ResumeAuditStream(ctx context.Context, sid session.ID
}, nil
}
// GetUploadMetadata gets session upload metadata
func (s *CheckingStreamer) GetUploadMetadata(sid session.ID) UploadMetadata {
return UploadMetadata{SessionID: sid}
}
// CheckAndSetDefaults checks and sets default values
func (w *CheckingStreamerConfig) CheckAndSetDefaults() error {
if w.Inner == nil {
@@ -497,6 +506,11 @@ func (t *TeeStreamer) ResumeAuditStream(ctx context.Context, sid session.ID, upl
return &TeeStream{stream: stream, emitter: t.Emitter}, nil
}
// GetUploadMetadata gets session upload metadata
func (t *TeeStreamer) GetUploadMetadata(sid session.ID) UploadMetadata {
return t.streamer.GetUploadMetadata(sid)
}
// TeeStreamer creates streams that forwards non print events
// to emitter
type TeeStreamer struct {
@@ -627,6 +641,11 @@ func (s *CallbackStreamer) ResumeAuditStream(ctx context.Context, sid session.ID
}, nil
}
// GetUploadMetadata gets session upload metadata
func (s *CallbackStreamer) GetUploadMetadata(sid session.ID) UploadMetadata {
return UploadMetadata{SessionID: sid}
}
// CallbackStream call
type CallbackStream struct {
stream Stream
@@ -709,6 +728,11 @@ func (s *ReportingStreamer) ResumeAuditStream(ctx context.Context, sid session.I
}, nil
}
// GetUploadMetadata gets session upload metadata
func (s *ReportingStreamer) GetUploadMetadata(sid session.ID) UploadMetadata {
return s.streamer.GetUploadMetadata(sid)
}
// ReportingStream reports status of uploads to the events channel
type ReportingStream struct {
Stream
+9 -2
View File
@@ -56,6 +56,8 @@ type UploaderConfig struct {
EventsC chan events.UploadEvent
// Component is used for logging purposes
Component string
// AuditLog is used for storing logs
AuditLog events.IAuditLog
}
// CheckAndSetDefaults checks and sets default values of UploaderConfig
@@ -66,6 +68,9 @@ func (cfg *UploaderConfig) CheckAndSetDefaults() error {
if cfg.ScanDir == "" {
return trace.BadParameter("missing parameter ScanDir")
}
if cfg.AuditLog == nil {
return trace.BadParameter("missing parameter AuditLog")
}
if cfg.ConcurrentUploads <= 0 {
cfg.ConcurrentUploads = defaults.UploaderConcurrentUploads
}
@@ -113,6 +118,7 @@ func NewUploader(cfg UploaderConfig) (*Uploader, error) {
}),
cancel: cancel,
ctx: ctx,
auditLog: cfg.AuditLog,
semaphore: make(chan struct{}, cfg.ConcurrentUploads),
eventsCh: make(chan events.UploadEvent, cfg.ConcurrentUploads),
}
@@ -142,6 +148,7 @@ type Uploader struct {
cancel context.CancelFunc
ctx context.Context
eventsCh chan events.UploadEvent
auditLog events.IAuditLog
}
func (u *Uploader) writeSessionError(sessionID session.ID, err error) error {
@@ -430,11 +437,11 @@ func (u *Uploader) startUpload(fileName string) error {
})
return
}
u.log.WithFields(log.Fields{"duration": time.Since(start), "session-id": sessionID}).Debugf("Session upload completed.")
u.emitEvent(events.UploadEvent{
SessionID: string(upload.sessionID),
Created: u.cfg.Clock.Now().UTC(),
})
}()
return nil
}
@@ -514,7 +521,7 @@ func (u *Uploader) upload(up *upload) error {
}
}
if err := stream.Complete(u.ctx); err != nil {
if err = stream.Complete(u.ctx); err != nil {
u.log.WithError(err).Error("Failed to complete upload.")
return trace.Wrap(err)
}
+8
View File
@@ -249,6 +249,14 @@ func (h *Handler) ListUploads(ctx context.Context) ([]events.StreamUpload, error
return uploads, nil
}
// GetUploadMetadata gets the metadata for session upload
func (h *Handler) GetUploadMetadata(s session.ID) events.UploadMetadata {
return events.UploadMetadata{
URL: fmt.Sprintf("%v://%v/%v", teleport.SchemeFile, h.uploadsPath(), string(s)),
SessionID: s,
}
}
func (h *Handler) uploadsPath() string {
return filepath.Join(h.Directory, uploadsDir)
}
+9
View File
@@ -27,6 +27,7 @@ import (
"strings"
"time"
"github.com/gravitational/teleport"
"github.com/gravitational/teleport/lib/events"
"github.com/gravitational/teleport/lib/session"
@@ -275,6 +276,14 @@ func (h *Handler) ListUploads(ctx context.Context) ([]events.StreamUpload, error
return uploads, nil
}
// GetUploadMetadata gets the metadata for session upload
func (h *Handler) GetUploadMetadata(s session.ID) events.UploadMetadata {
return events.UploadMetadata{
URL: fmt.Sprintf("%v://%v/%v", teleport.SchemeGCS, h.path(s), string(s)),
SessionID: s,
}
}
const (
// uploadsKey is a key that holds all upload-related objects
uploadsKey = "uploads"
+4
View File
@@ -142,6 +142,10 @@ func (r *ForwardRecorder) GetAuditLog() IAuditLog {
return r.AuditLog
}
func (r *Forwarder) EmitAuditEvent(ctx context.Context, event AuditEvent) error{
return nil
}
// Write takes a chunk and writes it into the audit log
func (r *ForwardRecorder) Write(data []byte) (int, error) {
// we are copying buffer to prevent data corruption:
+10
View File
@@ -18,12 +18,14 @@ package s3sessions
import (
"context"
"fmt"
"io"
"sort"
"strings"
"time"
"github.com/aws/aws-sdk-go/service/s3/s3manager"
"github.com/gravitational/teleport"
"github.com/gravitational/teleport/lib/defaults"
"github.com/gravitational/teleport/lib/events"
"github.com/gravitational/teleport/lib/session"
@@ -184,3 +186,11 @@ func (h *Handler) ListUploads(ctx context.Context) ([]events.StreamUpload, error
})
return uploads, nil
}
// GetUploadMetadata gets the metadata for session upload
func (h *Handler) GetUploadMetadata(sessionID session.ID) events.UploadMetadata {
return events.UploadMetadata{
URL: fmt.Sprintf("%v://%v/%v", teleport.SchemeS3, h.Bucket, sessionID),
SessionID: sessionID,
}
}
+8
View File
@@ -160,6 +160,10 @@ func (s *ProtoStreamer) ResumeAuditStream(ctx context.Context, sid session.ID, u
CompletedParts: parts,
})
}
// GetUploadMetadata gets the session upload metadata
func (s *ProtoStreamer) GetUploadMetadata(sid session.ID) UploadMetadata {
return s.cfg.Uploader.GetUploadMetadata(sid)
}
// ProtoStreamConfig configures proto stream
type ProtoStreamConfig struct {
@@ -1275,3 +1279,7 @@ func (m *MemoryUploader) Download(ctx context.Context, sessionID session.ID, wri
}
return nil
}
func (m *MemoryUploader) GetUploadMetadata(sid session.ID) UploadMetadata {
return UploadMetadata{}
}
+2
View File
@@ -60,6 +60,7 @@ type (
SessionStart = events.SessionStart
SessionJoin = events.SessionJoin
SessionPrint = events.SessionPrint
SessionUpload = events.SessionUpload
SessionReject = events.SessionReject
Resize = events.Resize
SessionEnd = events.SessionEnd
@@ -106,6 +107,7 @@ type (
OneOf_UserDelete = events.OneOf_UserDelete //nolint
OneOf_UserPasswordChange = events.OneOf_UserPasswordChange //nolint
OneOf_SessionStart = events.OneOf_SessionStart //nolint
OneOf_SessionUpload = events.OneOf_SessionUpload //nolint
OneOf_SessionJoin = events.OneOf_SessionJoin //nolint
OneOf_SessionPrint = events.OneOf_SessionPrint //nolint
OneOf_SessionReject = events.OneOf_SessionReject //nolint
+4 -1
View File
@@ -1041,7 +1041,7 @@ func (process *TeleportProcess) initAuthService() error {
}
auditConfig := cfg.Auth.ClusterConfig.GetAuditConfig()
uploadHandler, err := initUploadHandler(
uploadHandler, err = initUploadHandler(
auditConfig, filepath.Join(cfg.DataDir, teleport.LogsDir))
if err != nil {
if !trace.IsNotFound(err) {
@@ -1099,6 +1099,7 @@ func (process *TeleportProcess) initAuthService() error {
uploadCompleter, err = events.NewUploadCompleter(events.UploadCompleterConfig{
Uploader: uploadHandler,
Component: teleport.ComponentAuth,
AuditLog: process.auditLog,
})
if err != nil {
return trace.Wrap(err)
@@ -1179,6 +1180,7 @@ func (process *TeleportProcess) initAuthService() error {
Authorizer: authorizer,
AuditLog: process.auditLog,
Emitter: checkingEmitter,
MetadataGetter: streamer,
}
var authCache auth.Cache
@@ -1932,6 +1934,7 @@ func (process *TeleportProcess) initUploaderService(accessPoint auth.AccessPoint
fileUploader, err := filesessions.NewUploader(filesessions.UploaderConfig{
ScanDir: filepath.Join(streamingDir...),
Streamer: accessPoint,
AuditLog: auditLog,
EventsC: process.Config.UploadEventsC,
})
if err != nil {
+8
View File
@@ -502,6 +502,10 @@ func (f *fakeServer) GetBPF() bpf.BPF {
return &bpf.NOP{}
}
func (f *fakeServer) GetUploadMetadata(sid rsession.ID) events.UploadMetadata {
return events.UploadMetadata{}
}
// fakeLog is used in tests to obtain the last event emit to the Audit Log.
type fakeLog struct {
}
@@ -510,6 +514,10 @@ func (a *fakeLog) EmitAuditEventLegacy(e events.Event, f events.EventFields) err
return trace.NotImplemented("not implemented")
}
func (a *fakeLog) EmitAuditEvent(ctx context.Context, e events.AuditEvent) error {
return trace.NotImplemented("not implemented")
}
func (a *fakeLog) PostSessionSlice(s events.SessionSlice) error {
return trace.NotImplemented("not implemented")
}