From 22aa999a7b56019b0bab80bfa7dd12dcd341b7df Mon Sep 17 00:00:00 2001 From: sawka Date: Sun, 12 Jun 2022 13:39:48 -0700 Subject: [PATCH] initial checkin of main-server --- cmd/main-server.go | 330 +++++++++++++++++++++++++++++++++++++++++ go.mod | 15 ++ go.sum | 12 ++ pkg/wsshell/wsshell.go | 131 ++++++++++++++++ 4 files changed, 488 insertions(+) create mode 100644 cmd/main-server.go create mode 100644 go.mod create mode 100644 go.sum create mode 100644 pkg/wsshell/wsshell.go diff --git a/cmd/main-server.go b/cmd/main-server.go new file mode 100644 index 00000000..cdc507d4 --- /dev/null +++ b/cmd/main-server.go @@ -0,0 +1,330 @@ +package main + +import ( + "encoding/json" + "fmt" + "io" + "net/http" + "os" + "os/exec" + "time" + + "github.com/fsnotify/fsnotify" + "github.com/gorilla/mux" + + "github.com/scripthaus-dev/sh2-runner/pkg/base" + "github.com/scripthaus-dev/sh2-runner/pkg/packet" + "github.com/scripthaus-dev/sh2-server/pkg/wsshell" +) + +const HttpReadTimeout = 5 * time.Second +const HttpWriteTimeout = 21 * time.Second +const HttpMaxHeaderBytes = 60000 +const HttpTimeoutDuration = 21 * time.Second + +var GlobalRunnerProc *RunnerProc + +type PtyTailWs struct { + Shell *wsshell.WSShell + SessionId string + CmdId string + Position string + Watcher *fsnotify.Watcher +} + +type RunnerProc struct { + Cmd *exec.Cmd + Input *packet.PacketSender + Output chan packet.PacketType +} + +func TailFile(tailWs *PtyTailWs) error { +outer: + for { + select { + case event, ok := <-tailWs.Watcher.Events: + if !ok { + break outer + } + if event.Op&fsnotify.Write == fsnotify.Write { + tailWs.Shell.WriteChan <- []byte("*") + } + + case _, ok := <-tailWs.Watcher.Errors: + if !ok { + break outer + } + + case <-tailWs.Shell.CloseChan: + break outer + } + } + return nil +} + +func HandleWs(w http.ResponseWriter, r *http.Request) { + shell, err := wsshell.StartWS(w, r) + if err != nil { + w.WriteHeader(500) + w.Write([]byte(fmt.Sprintf("cannot ugprade websocket: %v", err))) + return + } + defer shell.Conn.Close() + tailWs := &PtyTailWs{ + Shell: shell, + } + tailWs.Watcher, err = fsnotify.NewWatcher() + if err != nil { + fmt.Printf("Error creating watcher: %v\n", err) + return + } + defer tailWs.Watcher.Close() + go func() { + defer shell.Conn.Close() + TailFile(tailWs) + }() + for msg := range shell.ReadChan { + jmsg := map[string]interface{}{} + err = json.Unmarshal(msg, &jmsg) + if err != nil { + fmt.Printf("error unmarshalling ws message: %v\n", err) + break + } + sessionId, ok := jmsg["sessionid"].(string) + if !ok || sessionId == "" { + fmt.Printf("bad ws message, no sessionid\n") + break + } + cmdId, ok := jmsg["cmdid"].(string) + if !ok || cmdId == "" { + fmt.Printf("bad ws message, no cmdId\n") + break + } + if tailWs.SessionId != "" { + fmt.Printf("bad ws message, sessionid already set\n") + break + } + tailWs.SessionId = sessionId + tailWs.CmdId = cmdId + pathStr := GetPtyOutFile(sessionId, cmdId) + err = tailWs.Watcher.Add(pathStr) + if err != nil { + fmt.Printf("error adding watcher: %v\n", err) + break + } + } +} + +func GetPtyOutFile(sessionId string, cmdId string) string { + pathStr := fmt.Sprintf("/Users/mike/scripthaus/.sessions/%s/%s.ptyout", sessionId, cmdId) + return pathStr +} + +func GetPtyOut(w http.ResponseWriter, r *http.Request) { + w.Header().Set("Access-Control-Allow-Origin", r.Header.Get("Origin")) + w.Header().Set("Access-Control-Allow-Credentials", "true") + w.Header().Set("Vary", "Origin") + w.Header().Set("Cache-Control", "no-cache") + qvals := r.URL.Query() + sessionId := qvals.Get("sessionid") + cmdId := qvals.Get("cmdid") + if sessionId == "" || cmdId == "" { + w.WriteHeader(500) + w.Write([]byte(fmt.Sprintf("must specify sessionid and cmdid"))) + return + } + pathStr := GetPtyOutFile(sessionId, cmdId) + fd, err := os.Open(pathStr) + if err != nil { + w.WriteHeader(500) + w.Write([]byte(fmt.Sprintf("cannot open file '%s': %v", pathStr, err))) + return + } + w.WriteHeader(http.StatusOK) + io.Copy(w, fd) +} + +type runCommandParams struct { + SessionId string `json:"sessionid"` + Command string `json:"command"` +} + +func WriteJsonError(w http.ResponseWriter, errVal error) { + w.WriteHeader(500) + errMap := make(map[string]interface{}) + errMap["error"] = errVal.Error() + barr, _ := json.Marshal(errMap) + w.Write(barr) + return +} + +func WriteJsonSuccess(w http.ResponseWriter, data interface{}) { + rtnMap := make(map[string]interface{}) + rtnMap["success"] = true + if data != nil { + rtnMap["data"] = data + } + barr, err := json.Marshal(rtnMap) + if err != nil { + WriteJsonError(w, err) + return + } + w.WriteHeader(200) + w.Write(barr) + return +} + +func HandleRunCommand(w http.ResponseWriter, r *http.Request) { + w.Header().Set("Access-Control-Allow-Origin", r.Header.Get("Origin")) + w.Header().Set("Access-Control-Allow-Credentials", "true") + w.Header().Set("Access-Control-Allow-Methods", "POST, OPTIONS") + w.Header().Set("Vary", "Origin") + w.Header().Set("Cache-Control", "no-cache") + if r.Method == "GET" || r.Method == "OPTIONS" { + w.WriteHeader(200) + return + } + decoder := json.NewDecoder(r.Body) + var params runCommandParams + err := decoder.Decode(¶ms) + if err != nil { + WriteJsonError(w, fmt.Errorf("error decoding json: %w", err)) + return + } + fmt.Printf("RUN COMMAND sessionid[%s] cmd[%s]\n", params.SessionId, params.Command) + WriteJsonSuccess(w, nil) + return +} + +// /api/start-session +// returns: +// * userid +// * sessionid +// +// /api/ptyout (pos=[position]) - returns contents of ptyout file +// params: +// * sessionid +// * cmdid +// * pos +// returns: +// * stream of ptyout file (text, utf-8) +// +// POST /api/run-command +// params +// * userid +// * sessionid +// returns +// * cmdid +// +// /api/refresh-session +// params +// * sessionid +// * start -- can be negative +// * numlines +// returns +// * permissions (readonly, comment, command) +// * lines +// * lineid +// * ts +// * userid +// * linetype +// * text +// * cmdid + +// /ws +// ->watch-session: +// * sessionid +// ->watch: +// * sessionid +// * cmdid +// ->focus: +// * sessionid +// * cmdid +// ->input: +// * sessionid +// * cmdid +// * data +// ->signal: +// * sessionid +// * cmdid +// * data +// <-data: +// * sessionid +// * cmdid +// * pos +// * data +// <-session-data: +// * sessionid +// * line + +// session-doc +// timestamp | user | cmd-type | data +// cmd-type = comment +// cmd-type = command, commandid=ABC + +// how to know if command is still executing? is command done? + +// local -- .ptyout, .stdin +// remote -- transfer controller program +// controller-startcmd -- start command (with options) => returns cmdid +// controller-watchsession [sessionid] +// transfer [cmdid:pos] pairs. streams back anything new written to ptyout on stdout +// stdin-packet [cmdid:user:data] +// startcmd will figure out the correct +// + +func LaunchRunnerProc() (*RunnerProc, error) { + runnerPath, err := base.GetScRunnerPath() + if err != nil { + return nil, err + } + ecmd := exec.Command(runnerPath) + inputWriter, err := ecmd.StdinPipe() + if err != nil { + return nil, err + } + outputReader, err := ecmd.StdoutPipe() + if err != nil { + return nil, err + } + ecmd.Stderr = nil // /dev/null + ecmd.Start() + rtn := &RunnerProc{Cmd: ecmd} + rtn.Output = packet.PacketParser(outputReader) + rtn.Input = packet.MakePacketSender(inputWriter) + return rtn, nil +} + +func ProcessPackets(runner *RunnerProc) { + for pk := range runner.Output { + fmt.Printf("runner-packet: %v\n", pk) + } +} + +func main() { + runnerProc, err := LaunchRunnerProc() + if err != nil { + fmt.Printf("error launching runner-proc: %v\n", err) + return + } + GlobalRunnerProc = runnerProc + go ProcessPackets(runnerProc) + fmt.Printf("Started local runner pid[%d]\n", runnerProc.Cmd.Process.Pid) + gr := mux.NewRouter() + gr.HandleFunc("/api/ptyout", GetPtyOut) + gr.HandleFunc("/ws", HandleWs) + gr.HandleFunc("/api/run-command", HandleRunCommand).Methods("GET", "POST", "OPTIONS") + server := &http.Server{ + Addr: "localhost:8080", + ReadTimeout: HttpReadTimeout, + WriteTimeout: HttpWriteTimeout, + MaxHeaderBytes: HttpMaxHeaderBytes, + Handler: http.TimeoutHandler(gr, HttpTimeoutDuration, "Timeout"), + } + server.SetKeepAlivesEnabled(false) + fmt.Printf("Running on http://localhost:8080\n") + err = server.ListenAndServe() + if err != nil { + fmt.Printf("ERROR: %v\n", err) + } +} diff --git a/go.mod b/go.mod new file mode 100644 index 00000000..d806aa77 --- /dev/null +++ b/go.mod @@ -0,0 +1,15 @@ +module github.com/scripthaus-dev/sh2-server + +go 1.17 + +require ( + github.com/creack/pty v1.1.18 // indirect + github.com/fsnotify/fsnotify v1.5.4 // indirect + github.com/google/uuid v1.3.0 // indirect + github.com/gorilla/mux v1.8.0 // indirect + github.com/gorilla/websocket v1.5.0 // indirect + golang.org/x/sys v0.0.0-20220412211240-33da011f77ad // indirect + github.com/scripthaus-dev/sh2-runner v0.0.0 +) + +replace "github.com/scripthaus-dev/sh2-runner" v0.0.0 => /Users/mike/work/gopath/src/github.com/scripthaus-dev/sh2-runner/ diff --git a/go.sum b/go.sum new file mode 100644 index 00000000..a1b065d4 --- /dev/null +++ b/go.sum @@ -0,0 +1,12 @@ +github.com/creack/pty v1.1.18 h1:n56/Zwd5o6whRC5PMGretI4IdRLlmBXYNjScPaBgsbY= +github.com/creack/pty v1.1.18/go.mod h1:MOBLtS5ELjhRRrroQr9kyvTxUAFNvYEK993ew/Vr4O4= +github.com/fsnotify/fsnotify v1.5.4 h1:jRbGcIw6P2Meqdwuo0H1p6JVLbL5DHKAKlYndzMwVZI= +github.com/fsnotify/fsnotify v1.5.4/go.mod h1:OVB6XrOHzAwXMpEM7uPOzcehqUV2UqJxmVXmkdnm1bU= +github.com/google/uuid v1.3.0 h1:t6JiXgmwXMjEs8VusXIJk2BXHsn+wx8BZdTaoZ5fu7I= +github.com/google/uuid v1.3.0/go.mod h1:TIyPZe4MgqvfeYDBFedMoGGpEw/LqOeaOT+nhxU+yHo= +github.com/gorilla/mux v1.8.0 h1:i40aqfkR1h2SlN9hojwV5ZA91wcXFOvkdNIeFDP5koI= +github.com/gorilla/mux v1.8.0/go.mod h1:DVbg23sWSpFRCP0SfiEN6jmj59UnW/n46BH5rLB71So= +github.com/gorilla/websocket v1.5.0 h1:PPwGk2jz7EePpoHN/+ClbZu8SPxiqlu12wZP/3sWmnc= +github.com/gorilla/websocket v1.5.0/go.mod h1:YR8l580nyteQvAITg2hZ9XVh4b55+EU/adAjf1fMHhE= +golang.org/x/sys v0.0.0-20220412211240-33da011f77ad h1:ntjMns5wyP/fN65tdBD4g8J5w8n015+iIIs9rtjXkY0= +golang.org/x/sys v0.0.0-20220412211240-33da011f77ad/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg= diff --git a/pkg/wsshell/wsshell.go b/pkg/wsshell/wsshell.go new file mode 100644 index 00000000..c969b687 --- /dev/null +++ b/pkg/wsshell/wsshell.go @@ -0,0 +1,131 @@ +package wsshell + +import ( + "encoding/json" + "log" + "net/http" + "net/url" + "sync" + "time" + + "github.com/google/uuid" + "github.com/gorilla/websocket" +) + +var upgrader = websocket.Upgrader{ + ReadBufferSize: 4 * 1024, + WriteBufferSize: 4 * 1024, + HandshakeTimeout: 1 * time.Second, + CheckOrigin: func(r *http.Request) bool { return true }, +} + +type WSShell struct { + Conn *websocket.Conn + RemoteAddr string + ConnId string + Query url.Values + OpenTime time.Time + NumPings int + LastPing time.Time + LastRecv time.Time + Header http.Header + + CloseChan chan bool + WriteChan chan []byte + ReadChan chan []byte +} + +func (ws *WSShell) WritePump() { + writeWait := 2 * time.Second + pingPeriod := 2 * time.Second + ticker := time.NewTicker(pingPeriod) + defer func() { + ticker.Stop() + ws.Conn.Close() + }() + for { + select { + case <-ticker.C: + now := time.Now() + pingMessage := map[string]interface{}{"type": "ping", "stime": now.Unix()} + jsonVal, _ := json.Marshal(pingMessage) + _ = ws.Conn.SetWriteDeadline(time.Now().Add(writeWait)) // no error + err := ws.Conn.WriteMessage(websocket.TextMessage, jsonVal) + ws.NumPings++ + ws.LastPing = now + if err != nil { + log.Printf("WritePump %s err: %v\n", ws.RemoteAddr, err) + return + } + + case msgBytes := <-ws.WriteChan: + _ = ws.Conn.SetWriteDeadline(time.Now().Add(writeWait)) // no error + err := ws.Conn.WriteMessage(websocket.TextMessage, msgBytes) + if err != nil { + log.Printf("WritePump %s err: %v\n", ws.RemoteAddr, err) + return + } + } + } +} + +func (ws *WSShell) ReadPump() { + readWait := 5 * time.Second + defer func() { + ws.Conn.Close() + }() + ws.Conn.SetReadLimit(4096) + ws.Conn.SetReadDeadline(time.Now().Add(readWait)) + for { + _, message, err := ws.Conn.ReadMessage() + if err != nil { + log.Printf("ReadPump %s Err: %v\n", ws.RemoteAddr, err) + break + } + jmsg := map[string]interface{}{} + err = json.Unmarshal(message, &jmsg) + if err != nil { + log.Printf("Error unmarshalling json: %v\n", err) + break + } + ws.Conn.SetReadDeadline(time.Now().Add(readWait)) + ws.LastRecv = time.Now() + if str, ok := jmsg["type"].(string); ok && str == "pong" { + // nothing + continue + } + ws.ReadChan <- message + } + +} + +func StartWS(w http.ResponseWriter, r *http.Request) (*WSShell, error) { + conn, err := upgrader.Upgrade(w, r, nil) + if err != nil { + return nil, err + } + ws := WSShell{Conn: conn, ConnId: uuid.New().String(), OpenTime: time.Now()} + ws.CloseChan = make(chan bool) + ws.WriteChan = make(chan []byte, 10) + ws.ReadChan = make(chan []byte, 10) + ws.RemoteAddr = r.RemoteAddr + ws.Query = r.URL.Query() + ws.Header = r.Header + var wg sync.WaitGroup + wg.Add(1) + go func() { + defer wg.Done() + ws.WritePump() + }() + wg.Add(1) + go func() { + defer wg.Done() + ws.ReadPump() + }() + go func() { + wg.Wait() + close(ws.CloseChan) + close(ws.ReadChan) + }() + return &ws, nil +}