-
Notifications
You must be signed in to change notification settings - Fork 1.7k
/
forward.go
161 lines (144 loc) · 5.18 KB
/
forward.go
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
package events
import (
"context"
"io"
"time"
"github.com/gravitational/teleport/lib/session"
"github.com/gravitational/trace"
)
// ForwarderConfig forwards session log events
// to the auth server, and writes the session playback to disk
type ForwarderConfig struct {
// SessionID is a session id to write
SessionID session.ID
// ServerID is a serverID data directory
ServerID string
// DataDir is a data directory
DataDir string
// RecordSessions is a sessions recording setting
RecordSessions bool
// Namespace is a namespace of the session
Namespace string
// ForwardTo is the audit log to forward non-print events to
ForwardTo IAuditLog
}
// CheckAndSetDefaults checks and sets default values
func (s *ForwarderConfig) CheckAndSetDefaults() error {
if s.ForwardTo == nil {
return trace.BadParameter("missing parameter bucket")
}
if s.DataDir == "" {
return trace.BadParameter("missing data dir")
}
return nil
}
// NewForwarder returns a new instance of session forwarder
func NewForwarder(cfg ForwarderConfig) (*Forwarder, error) {
if err := cfg.CheckAndSetDefaults(); err != nil {
return nil, trace.Wrap(err)
}
diskLogger, err := NewDiskSessionLogger(DiskSessionLoggerConfig{
SessionID: cfg.SessionID,
DataDir: cfg.DataDir,
RecordSessions: cfg.RecordSessions,
Namespace: cfg.Namespace,
ServerID: cfg.ServerID,
})
if err != nil {
return nil, trace.Wrap(err)
}
return &Forwarder{
ForwarderConfig: cfg,
sessionLogger: diskLogger,
}, nil
}
// ForwarderConfig forwards session log events
// to the auth server, and writes the session playback to disk
type Forwarder struct {
ForwarderConfig
sessionLogger *DiskSessionLogger
}
// Closer releases connection and resources associated with log if any
func (l *Forwarder) Close() error {
return l.sessionLogger.Finalize()
}
// EmitAuditEvent emits audit event
func (l *Forwarder) EmitAuditEvent(eventType string, fields EventFields) error {
return l.ForwardTo.EmitAuditEvent(eventType, fields)
}
// PostSessionSlice sends chunks of recorded session to the event log
func (l *Forwarder) PostSessionSlice(slice SessionSlice) error {
err := l.sessionLogger.PostSessionSlice(slice)
if err != nil {
return trace.Wrap(err)
}
// filter out chunks with session print events,
// as this logger forwards only audit events to the auth server
var chunks []*SessionChunk
for _, chunk := range slice.Chunks {
if chunk.EventType != SessionPrintEvent {
chunks = append(chunks, chunk)
}
if chunk.EventType == SessionEndEvent {
if err := l.sessionLogger.Finalize(); err != nil {
return trace.Wrap(err)
}
}
}
// no chunks to post (all chunks are print events)
if len(chunks) == 0 {
return nil
}
slice.Chunks = chunks
slice.Version = V3
err = l.ForwardTo.PostSessionSlice(slice)
return err
}
// UploadSessionRecording uploads session recording to the audit server
func (l *Forwarder) UploadSessionRecording(r SessionRecording) error {
return l.ForwardTo.UploadSessionRecording(r)
}
// PostSessionChunk returns a writer which SSH nodes use to submit
// their live sessions into the session log
func (l *Forwarder) PostSessionChunk(namespace string, sid session.ID, reader io.Reader) error {
return l.ForwardTo.PostSessionChunk(namespace, sid, reader)
}
// GetSessionChunk returns a reader which can be used to read a byte stream
// of a recorded session starting from 'offsetBytes' (pass 0 to start from the
// beginning) up to maxBytes bytes.
//
// If maxBytes > MaxChunkBytes, it gets rounded down to MaxChunkBytes
func (l *Forwarder) GetSessionChunk(namespace string, sid session.ID, offsetBytes, maxBytes int) ([]byte, error) {
return l.ForwardTo.GetSessionChunk(namespace, sid, offsetBytes, maxBytes)
}
// Returns all events that happen during a session sorted by time
// (oldest first).
//
// after tells to use only return events after a specified cursor Id
//
// This function is usually used in conjunction with GetSessionReader to
// replay recorded session streams.
func (l *Forwarder) GetSessionEvents(namespace string, sid session.ID, after int, includePrintEvents bool) ([]EventFields, error) {
return l.ForwardTo.GetSessionEvents(namespace, sid, after, includePrintEvents)
}
// SearchEvents is a flexible way to find The format of a query string
// depends on the implementing backend. A recommended format is urlencoded
// (good enough for Lucene/Solr)
//
// Pagination is also defined via backend-specific query format.
//
// The only mandatory requirement is a date range (UTC). Results must always
// show up sorted by date (newest first)
func (l *Forwarder) SearchEvents(fromUTC, toUTC time.Time, query string, limit int) ([]EventFields, error) {
return l.ForwardTo.SearchEvents(fromUTC, toUTC, query, limit)
}
// SearchSessionEvents returns session related events only. This is used to
// find completed session.
func (l *Forwarder) SearchSessionEvents(fromUTC time.Time, toUTC time.Time, limit int) ([]EventFields, error) {
return l.ForwardTo.SearchSessionEvents(fromUTC, toUTC, limit)
}
// WaitForDelivery waits for resources to be released and outstanding requests to
// complete after calling Close method
func (l *Forwarder) WaitForDelivery(ctx context.Context) error {
return l.ForwardTo.WaitForDelivery(ctx)
}