/
terminalSesion.go
390 lines (344 loc) · 11.7 KB
/
terminalSesion.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
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
// Copyright 2017 The Kubernetes Authors.
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
// http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.
package terminal
import (
"crypto/rand"
"encoding/hex"
"encoding/json"
"fmt"
"github.com/devtron-labs/devtron/pkg/cluster"
"go.uber.org/zap"
"io"
"k8s.io/apimachinery/pkg/api/errors"
"log"
"net/http"
"sync"
"gopkg.in/igm/sockjs-go.v3/sockjs"
v1 "k8s.io/api/core/v1"
"k8s.io/client-go/kubernetes"
"k8s.io/client-go/kubernetes/scheme"
"k8s.io/client-go/rest"
"k8s.io/client-go/tools/remotecommand"
)
const END_OF_TRANSMISSION = "\u0004"
// PtyHandler is what remotecommand expects from a pty
type PtyHandler interface {
io.Reader
io.Writer
remotecommand.TerminalSizeQueue
}
// TerminalSession implements PtyHandler (using a SockJS connection)
type TerminalSession struct {
id string
bound chan error
sockJSSession sockjs.Session
sizeChan chan remotecommand.TerminalSize
doneChan chan struct{}
}
// TerminalMessage is the messaging protocol between ShellController and TerminalSession.
//
// OP DIRECTION FIELD(S) USED DESCRIPTION
// ---------------------------------------------------------------------
// bind fe->be SessionID Id sent back from TerminalResponse
// stdin fe->be Data Keystrokes/paste buffer
// resize fe->be Rows, Cols New terminal size
// stdout be->fe Data Output from the process
// toast be->fe Data OOB message to be shown to the user
type TerminalMessage struct {
Op, Data, SessionID string
Rows, Cols uint16
}
// TerminalSize handles pty->process resize events
// Called in a loop from remotecommand as long as the process is running
func (t TerminalSession) Next() *remotecommand.TerminalSize {
select {
case size := <-t.sizeChan:
return &size
case <-t.doneChan:
return nil
}
}
// Read handles pty->process messages (stdin, resize)
// Called in a loop from remotecommand as long as the process is running
func (t TerminalSession) Read(p []byte) (int, error) {
m, err := t.sockJSSession.Recv()
if err != nil {
// Send terminated signal to process to avoid resource leak
return copy(p, END_OF_TRANSMISSION), err
}
var msg TerminalMessage
if err := json.Unmarshal([]byte(m), &msg); err != nil {
return copy(p, END_OF_TRANSMISSION), err
}
switch msg.Op {
case "stdin":
return copy(p, msg.Data), nil
case "resize":
t.sizeChan <- remotecommand.TerminalSize{Width: msg.Cols, Height: msg.Rows}
return 0, nil
default:
return copy(p, END_OF_TRANSMISSION), fmt.Errorf("unknown message type '%s'", msg.Op)
}
}
// Write handles process->pty stdout
// Called from remotecommand whenever there is any output
func (t TerminalSession) Write(p []byte) (int, error) {
msg, err := json.Marshal(TerminalMessage{
Op: "stdout",
Data: string(p),
})
if err != nil {
return 0, err
}
if err = t.sockJSSession.Send(string(msg)); err != nil {
return 0, err
}
return len(p), nil
}
// Toast can be used to send the user any OOB messages
// hterm puts these in the center of the terminal
func (t TerminalSession) Toast(p string) error {
msg, err := json.Marshal(TerminalMessage{
Op: "toast",
Data: p,
})
if err != nil {
return err
}
if err = t.sockJSSession.Send(string(msg)); err != nil {
return err
}
return nil
}
// SessionMap stores a map of all TerminalSession objects and a lock to avoid concurrent conflict
type SessionMap struct {
Sessions map[string]TerminalSession
Lock sync.RWMutex
}
// Get return a given terminalSession by sessionId
func (sm *SessionMap) Get(sessionId string) TerminalSession {
sm.Lock.RLock()
defer sm.Lock.RUnlock()
return sm.Sessions[sessionId]
}
// Set store a TerminalSession to SessionMap
func (sm *SessionMap) Set(sessionId string, session TerminalSession) {
sm.Lock.Lock()
defer sm.Lock.Unlock()
sm.Sessions[sessionId] = session
}
// Close shuts down the SockJS connection and sends the status code and reason to the client
// Can happen if the process exits or if there is an error starting up the process
// For now the status code is unused and reason is shown to the user (unless "")
func (sm *SessionMap) Close(sessionId string, status uint32, reason string) {
sm.Lock.Lock()
defer sm.Lock.Unlock()
err := sm.Sessions[sessionId].sockJSSession.Close(status, reason)
if err != nil {
log.Println(err)
}
delete(sm.Sessions, sessionId)
}
var terminalSessions = SessionMap{Sessions: make(map[string]TerminalSession)}
// handleTerminalSession is Called by net/http for any new /api/sockjs connections
func handleTerminalSession(session sockjs.Session) {
var (
buf string
err error
msg TerminalMessage
terminalSession TerminalSession
)
if buf, err = session.Recv(); err != nil {
log.Printf("handleTerminalSession: can't Recv: %v", err)
return
}
if err = json.Unmarshal([]byte(buf), &msg); err != nil {
log.Printf("handleTerminalSession: can't UnMarshal (%v): %s", err, buf)
return
}
if msg.Op != "bind" {
log.Printf("handleTerminalSession: expected 'bind' message, got: %s", buf)
session.Close(http.StatusBadRequest, fmt.Sprintf("expected 'bind' message, got '%s'", buf))
return
}
if terminalSession = terminalSessions.Get(msg.SessionID); terminalSession.id == "" {
log.Printf("handleTerminalSession: can't find session '%s'", msg.SessionID)
session.Close(http.StatusGone, fmt.Sprintf("handleTerminalSession: can't find session '%s'", msg.SessionID))
return
}
terminalSession.sockJSSession = session
terminalSessions.Set(msg.SessionID, terminalSession)
terminalSession.bound <- nil
}
// CreateAttachHandler is called from main for /api/sockjs
func CreateAttachHandler(path string) http.Handler {
return sockjs.NewHandler(path, sockjs.DefaultOptions, handleTerminalSession)
}
// startProcess is called by handleAttach
// Executed cmd in the container specified in request and connects it up with the ptyHandler (a session)
func startProcess(k8sClient kubernetes.Interface, cfg *rest.Config,
cmd []string, ptyHandler PtyHandler, sessionRequest *TerminalSessionRequest) error {
namespace := sessionRequest.Namespace
podName := sessionRequest.PodName
containerName := sessionRequest.ContainerName
req := k8sClient.CoreV1().RESTClient().Post().
Resource("pods").
Name(podName).
Namespace(namespace).
SubResource("exec")
req.VersionedParams(&v1.PodExecOptions{
Container: containerName,
Command: cmd,
Stdin: true,
Stdout: true,
Stderr: true,
TTY: true,
}, scheme.ParameterCodec)
exec, err := remotecommand.NewSPDYExecutor(cfg, "POST", req.URL())
if err != nil {
return err
}
err = exec.Stream(remotecommand.StreamOptions{
Stdin: ptyHandler,
Stdout: ptyHandler,
Stderr: ptyHandler,
TerminalSizeQueue: ptyHandler,
Tty: true,
})
if err != nil {
return err
}
return nil
}
// genTerminalSessionId generates a random session ID string. The format is not really interesting.
// This ID is used to identify the session when the client opens the SockJS connection.
// Not the same as the SockJS session id! We can't use that as that is generated
// on the client side and we don't have it yet at this point.
func genTerminalSessionId() (string, error) {
bytes := make([]byte, 16)
if _, err := rand.Read(bytes); err != nil {
return "", err
}
id := make([]byte, hex.EncodedLen(len(bytes)))
hex.Encode(id, bytes)
return string(id), nil
}
// isValidShell checks if the Shell is an allowed one
func isValidShell(validShells []string, shell string) bool {
for _, validShell := range validShells {
if validShell == shell {
return true
}
}
return false
}
type TerminalSessionRequest struct {
Shell string
SessionId string
Namespace string
PodName string
ContainerName string
EnvironmentId int
AppId int
}
// WaitForTerminal is called from apihandler.handleAttach as a goroutine
// Waits for the SockJS connection to be opened by the client the session to be bound in handleTerminalSession
func WaitForTerminal(k8sClient kubernetes.Interface, cfg *rest.Config, request *TerminalSessionRequest) {
select {
case <-terminalSessions.Get(request.SessionId).bound:
close(terminalSessions.Get(request.SessionId).bound)
var err error
validShells := []string{"bash", "sh", "powershell", "cmd"}
if isValidShell(validShells, request.Shell) {
cmd := []string{request.Shell}
err = startProcess(k8sClient, cfg, cmd, terminalSessions.Get(request.SessionId), request)
} else {
// No Shell given or it was not valid: try some shells until one succeeds or all fail
// FIXME: if the first Shell fails then the first keyboard event is lost
for _, testShell := range validShells {
cmd := []string{testShell}
if err = startProcess(k8sClient, cfg, cmd, terminalSessions.Get(request.SessionId), request); err == nil {
break
}
}
}
if err != nil {
terminalSessions.Close(request.SessionId, 2, err.Error())
return
}
terminalSessions.Close(request.SessionId, 1, "Process exited")
}
}
type TerminalSessionHandler interface {
GetTerminalSession(req *TerminalSessionRequest) (statusCode int, message *TerminalMessage, err error)
}
type TerminalSessionHandlerImpl struct {
environmentService cluster.EnvironmentService
clusterService cluster.ClusterService
logger *zap.SugaredLogger
}
func NewTerminalSessionHandlerImpl(environmentService cluster.EnvironmentService, clusterService cluster.ClusterService,
logger *zap.SugaredLogger) *TerminalSessionHandlerImpl {
return &TerminalSessionHandlerImpl{
environmentService: environmentService,
clusterService: clusterService,
logger: logger,
}
}
func (impl *TerminalSessionHandlerImpl) GetTerminalSession(req *TerminalSessionRequest) (statusCode int, message *TerminalMessage, err error) {
sessionID, err := genTerminalSessionId()
if err != nil {
statusCode := http.StatusInternalServerError
statusError, ok := err.(*errors.StatusError)
if ok && statusError.Status().Code > 0 {
statusCode = int(statusError.Status().Code)
}
return statusCode, nil, err
}
req.SessionId = sessionID
terminalSessions.Set(sessionID, TerminalSession{
id: sessionID,
bound: make(chan error),
sizeChan: make(chan remotecommand.TerminalSize),
})
config, client, err := impl.getClientConfig(req.EnvironmentId)
if err != nil {
impl.logger.Errorw("error in fetching config", "err", err)
return http.StatusInternalServerError, nil, err
}
go WaitForTerminal(client, config, req)
return http.StatusOK, &TerminalMessage{SessionID: sessionID}, nil
}
func (impl *TerminalSessionHandlerImpl) getClientConfig(envId int) (*rest.Config, *kubernetes.Clientset, error) {
cluster, err := impl.environmentService.FindClusterByEnvId(envId)
if err != nil {
impl.logger.Errorw("error in fetching cluster detail", "envId", envId, "err", err)
return nil, nil, err
}
config, err := impl.clusterService.GetClusterConfig(cluster)
if err != nil {
impl.logger.Errorw("error in config", "err", err)
return nil, nil, err
}
cfg := &rest.Config{}
cfg.Host = config.Host
cfg.BearerToken = config.BearerToken
cfg.Insecure = true
clientSet, err := kubernetes.NewForConfig(cfg)
if err != nil {
impl.logger.Errorw("error in clientSet", "err", err)
return nil, nil, err
}
return cfg, clientSet, nil
}