From ccc63937b67752d07b4e43f80cb5529df37e7141 Mon Sep 17 00:00:00 2001 From: Evan Simkowitz Date: Tue, 2 Apr 2024 15:46:32 -0700 Subject: [PATCH] Support running ephemeral commands (#543) * initial * save work, starting to add backend types * save work * save work * Add EphemeralWriteCloser * Command pipes thru, triggers infinite loop * save debugging * fix bad merge * save debug statements * fixing spaghetti * clean up code * got cwd override working * Add separate paths for stdout and stderr writers * fix stdout/stderr * env vars are now working * revert waveshell changes * Making EphemeralWriteCloser into a more generic BufferedPipe * formatting * comment * delete unused package * more naming changes * add package comment * add UsePty to EphemeralRunOpts * document UsePty * ensure only one downstream writer can read from the buffer * store pointer to syncs * remove inshellisense stuff for now * remove debugs * revert yarn * remove unnecessary debugs in main-server * more debugging removed * revert tsconfig --- src/models/clientsettingsview.ts | 1 - src/models/model.ts | 93 ++++++++++++ src/types/custom.d.ts | 20 +++ waveshell/pkg/packet/packet.go | 2 +- waveshell/pkg/shellenv/shellenv.go | 13 +- wavesrv/cmd/main-server.go | 174 +++++++++++++++++------ wavesrv/pkg/bufferedpipe/bufferedpipe.go | 172 ++++++++++++++++++++++ wavesrv/pkg/cmdrunner/cmdrunner.go | 41 +++--- wavesrv/pkg/cmdrunner/shparse.go | 1 + wavesrv/pkg/ephemeral/ephemeral.go | 28 ++++ wavesrv/pkg/remote/remote.go | 115 +++++++++++---- wavesrv/pkg/scpacket/scpacket.go | 34 +++-- wavesrv/pkg/sstore/sstore.go | 56 ++++---- 13 files changed, 616 insertions(+), 134 deletions(-) create mode 100644 wavesrv/pkg/bufferedpipe/bufferedpipe.go create mode 100644 wavesrv/pkg/ephemeral/ephemeral.go diff --git a/src/models/clientsettingsview.ts b/src/models/clientsettingsview.ts index 7c8c2f1f..ad177e7a 100644 --- a/src/models/clientsettingsview.ts +++ b/src/models/clientsettingsview.ts @@ -3,7 +3,6 @@ import * as mobx from "mobx"; import { Model } from "./model"; -import { checkKeyPressed, adaptFromReactOrNativeKeyEvent } from "@/util/keyutil"; class ClientSettingsViewModel { globalModel: Model; diff --git a/src/models/model.ts b/src/models/model.ts index 091f4ef7..29c4cdb1 100644 --- a/src/models/model.ts +++ b/src/models/model.ts @@ -36,6 +36,7 @@ import { GlobalCommandRunner } from "./global"; import { clearMonoFontCache, getMonoFontSize } from "@/util/textmeasure"; import type { TermWrap } from "@/plugins/terminal/term"; import * as util from "@/util/util"; +import { url } from "node:inspector"; type SWLinePtr = { line: LineType; @@ -1402,6 +1403,7 @@ class Model { kwargs: { ...kwargs }, uicontext: this.getUIContext(), interactive: interactive, + ephemeralopts: null, }; /** console.log( @@ -1415,6 +1417,97 @@ class Model { return this.submitCommandPacket(pk, interactive); } + getSingleEphemeralCommandOutput(url: URL): Promise { + return fetch(url, { method: "get", headers: this.getFetchHeaders() }) + .then((resp) => resp.text()) + .catch((err) => { + this.errorHandler("getting ephemeral command output", err, true); + return ""; + }); + } + + async getEphemeralCommandOutput( + ephemeralCommandResponse: EphemeralCommandResponsePacketType + ): Promise { + let stdout = ""; + let stderr = ""; + if (ephemeralCommandResponse.stdouturl) { + const url = new URL(this.getBaseHostPort() + ephemeralCommandResponse.stdouturl); + console.log("stdouturl", url); + stdout = await this.getSingleEphemeralCommandOutput(url); + } + if (ephemeralCommandResponse.stderrurl) { + const url = new URL(this.getBaseHostPort() + ephemeralCommandResponse.stderrurl); + console.log("stderrurl", url); + stderr = await this.getSingleEphemeralCommandOutput(url); + } + return { stdout: stdout, stderr: stderr }; + } + + submitEphemeralCommandPacket( + cmdPk: FeCmdPacketType, + interactive: boolean + ): Promise { + if (this.debugCmds > 0) { + console.log("[cmd]", cmdPacketString(cmdPk)); + if (this.debugCmds > 1) { + console.trace(); + } + } + // adding cmdStr for debugging only (easily filter run-command calls in the network tab of debugger) + const cmdStr = cmdPk.metacmd + (cmdPk.metasubcmd ? ":" + cmdPk.metasubcmd : ""); + const url = new URL(this.getBaseHostPort() + "/api/run-ephemeral-command?cmd=" + cmdStr); + const fetchHeaders = this.getFetchHeaders(); + const prtn = fetch(url, { + method: "post", + body: JSON.stringify(cmdPk), + headers: fetchHeaders, + }) + .then(async (resp) => { + const data = await handleJsonFetchResponse(url, resp); + if (data.success) { + return data.data as EphemeralCommandResponsePacketType; + } else { + console.log("error running ephemeral command", data); + return {}; + } + }) + .catch((err) => { + this.errorHandler("calling run-ephemeral-command", err, interactive); + return {}; + }); + return prtn; + } + + submitEphemeralCommand( + metaCmd: string, + metaSubCmd: string, + args: string[], + kwargs: Record, + interactive: boolean, + ephemeralopts?: EphemeralCmdOptsType + ): Promise { + const pk: FeCmdPacketType = { + type: "fecmd", + metacmd: metaCmd, + metasubcmd: metaSubCmd, + args: args, + kwargs: { ...kwargs }, + uicontext: this.getUIContext(), + interactive: interactive, + ephemeralopts: ephemeralopts, + }; + console.log( + "CMD", + pk.metacmd + (pk.metasubcmd != null ? ":" + pk.metasubcmd : ""), + pk.args, + pk.kwargs, + pk.interactive, + pk.ephemeralopts + ); + return this.submitEphemeralCommandPacket(pk, interactive); + } + submitChatInfoCommand(chatMsg: string, curLineStr: string, clear: boolean): Promise { const commandStr = "/chat " + chatMsg; const interactive = false; diff --git a/src/types/custom.d.ts b/src/types/custom.d.ts index 7ca11a6e..b7c0a6bd 100644 --- a/src/types/custom.d.ts +++ b/src/types/custom.d.ts @@ -1,3 +1,5 @@ +import { override } from "mobx"; + declare module "*.svg" { import * as React from "react"; export const ReactComponent: React.FunctionComponent & { title?: string }>; @@ -186,6 +188,13 @@ declare global { build: string; }; + type EphemeralCmdOptsType = { + overridecwd?: string; + timeoutms?: number; + expectsresponse: boolean; + env: { [k: string]: string }; + }; + type FeCmdPacketType = { type: string; metacmd: string; @@ -195,6 +204,7 @@ declare global { rawstr?: string; uicontext: UIContextType; interactive: boolean; + ephemeralopts?: EphemeralCmdOptsType; }; type FeInputPacketType = { @@ -793,6 +803,16 @@ declare global { error?: string; }; + type EphemeralCommandOutputType = { + stdout: string; + stderr: string; + }; + + type EphemeralCommandResponsePacketType = { + stdouturl?: string; + stderrurl?: string; + }; + type LineHeightChangeCallbackType = (lineNum: number, newHeight: number, oldHeight: number) => void; type OpenAIPacketType = { diff --git a/waveshell/pkg/packet/packet.go b/waveshell/pkg/packet/packet.go index ef3d0414..22e9828f 100644 --- a/waveshell/pkg/packet/packet.go +++ b/waveshell/pkg/packet/packet.go @@ -820,7 +820,7 @@ type RunPacketType struct { State *ShellState `json:"state,omitempty"` StatePtr *ShellStatePtr `json:"stateptr,omitempty"` // added in Wave v0.7.2 StateComplete bool `json:"statecomplete,omitempty"` // set to true if state is complete (the default env should not be set) - UsePty bool `json:"usepty,omitempty"` + UsePty bool `json:"usepty,omitempty"` // Set to true if the command should be run in a pty. This will write all output to the stdout file descriptor with PTY formatting. TermOpts *TermOpts `json:"termopts,omitempty"` Fds []RemoteFd `json:"fds,omitempty"` RunData []RunDataType `json:"rundata,omitempty"` diff --git a/waveshell/pkg/shellenv/shellenv.go b/waveshell/pkg/shellenv/shellenv.go index 6f10ca47..28970111 100644 --- a/waveshell/pkg/shellenv/shellenv.go +++ b/waveshell/pkg/shellenv/shellenv.go @@ -11,6 +11,7 @@ import ( "github.com/wavetermdev/waveterm/waveshell/pkg/packet" "github.com/wavetermdev/waveterm/waveshell/pkg/simpleexpand" "github.com/wavetermdev/waveterm/waveshell/pkg/utilfn" + "github.com/wavetermdev/waveterm/waveshell/pkg/wlog" ) const ( @@ -344,21 +345,21 @@ func ShellVarMapFromState(state *packet.ShellState) map[string]string { } func DumpVarMapFromState(state *packet.ShellState) { - fmt.Printf("DUMP-STATE-VARS:\n") + wlog.Logf("DUMP-STATE-VARS:\n") if state == nil { - fmt.Printf(" nil\n") + wlog.Logf(" nil\n") return } decls := VarDeclsFromState(state) for _, decl := range decls { - fmt.Printf(" %s %#v\n", decl.Name, decl) + wlog.Logf(" %s %#v\n", decl.Name, decl) } envMap := EnvMapFromState(state) - fmt.Printf("DUMP-STATE-ENV:\n") + wlog.Logf("DUMP-STATE-ENV:\n") for k, v := range envMap { - fmt.Printf(" %s=%s\n", k, v) + wlog.Logf(" %s=%s\n", k, v) } - fmt.Printf("\n\n") + wlog.Logf("\n\n") } func VarDeclsFromState(state *packet.ShellState) []*DeclareDeclType { diff --git a/wavesrv/cmd/main-server.go b/wavesrv/cmd/main-server.go index 9aab0704..41e37de7 100644 --- a/wavesrv/cmd/main-server.go +++ b/wavesrv/cmd/main-server.go @@ -35,13 +35,16 @@ import ( "github.com/wavetermdev/waveterm/waveshell/pkg/packet" "github.com/wavetermdev/waveterm/waveshell/pkg/server" "github.com/wavetermdev/waveterm/waveshell/pkg/wlog" + "github.com/wavetermdev/waveterm/wavesrv/pkg/bufferedpipe" "github.com/wavetermdev/waveterm/wavesrv/pkg/cmdrunner" + "github.com/wavetermdev/waveterm/wavesrv/pkg/ephemeral" "github.com/wavetermdev/waveterm/wavesrv/pkg/pcloud" "github.com/wavetermdev/waveterm/wavesrv/pkg/promptenc" "github.com/wavetermdev/waveterm/wavesrv/pkg/releasechecker" "github.com/wavetermdev/waveterm/wavesrv/pkg/remote" "github.com/wavetermdev/waveterm/wavesrv/pkg/rtnstate" "github.com/wavetermdev/waveterm/wavesrv/pkg/scbase" + "github.com/wavetermdev/waveterm/wavesrv/pkg/scbus" "github.com/wavetermdev/waveterm/wavesrv/pkg/scpacket" "github.com/wavetermdev/waveterm/wavesrv/pkg/scws" "github.com/wavetermdev/waveterm/wavesrv/pkg/sstore" @@ -98,6 +101,7 @@ const ( CacheControlHeaderNoCache = "no-cache" ContentTypeHeaderKey = "Content-Type" ContentTypeJson = "application/json" + ContentTypeText = "text/plain" ) func setWSState(state *scws.WSState) { @@ -255,30 +259,30 @@ func HandleRtnState(w http.ResponseWriter, r *http.Request) { } log.Printf("[error] in handlertnstate: %v\n", r) debug.PrintStack() - w.WriteHeader(500) + w.WriteHeader(http.StatusInternalServerError) w.Write([]byte(fmt.Sprintf(ErrorPanic, r))) }() qvals := r.URL.Query() screenId := qvals.Get("screenid") lineId := qvals.Get("lineid") if screenId == "" || lineId == "" { - w.WriteHeader(500) + w.WriteHeader(http.StatusInternalServerError) w.Write([]byte("must specify screenid and lineid")) return } if _, err := uuid.Parse(screenId); err != nil { - w.WriteHeader(500) + w.WriteHeader(http.StatusInternalServerError) w.Write([]byte(fmt.Sprintf(ErrorInvalidScreenId, err))) return } if _, err := uuid.Parse(lineId); err != nil { - w.WriteHeader(500) + w.WriteHeader(http.StatusInternalServerError) w.Write([]byte(fmt.Sprintf(ErrorInvalidLineId, err))) return } data, err := rtnstate.GetRtnStateDiff(r.Context(), screenId, lineId) if err != nil { - w.WriteHeader(500) + w.WriteHeader(http.StatusInternalServerError) w.Write([]byte(fmt.Sprintf("cannot get rtnstate diff: %v", err))) return } @@ -290,18 +294,18 @@ func HandleRemotePty(w http.ResponseWriter, r *http.Request) { qvals := r.URL.Query() remoteId := qvals.Get("remoteid") if remoteId == "" { - w.WriteHeader(500) + w.WriteHeader(http.StatusInternalServerError) w.Write([]byte("must specify remoteid")) return } if _, err := uuid.Parse(remoteId); err != nil { - w.WriteHeader(500) + w.WriteHeader(http.StatusInternalServerError) w.Write([]byte(fmt.Sprintf("invalid remoteid: %v", err))) return } realOffset, data, err := remote.ReadRemotePty(r.Context(), remoteId) if err != nil { - w.WriteHeader(500) + w.WriteHeader(http.StatusInternalServerError) w.Write([]byte(fmt.Sprintf("error reading ptyout file: %v", err))) return } @@ -315,17 +319,17 @@ func HandleGetPtyOut(w http.ResponseWriter, r *http.Request) { screenId := qvals.Get("screenid") lineId := qvals.Get("lineid") if screenId == "" || lineId == "" { - w.WriteHeader(500) + w.WriteHeader(http.StatusInternalServerError) w.Write([]byte("must specify screenid and lineid")) return } if _, err := uuid.Parse(screenId); err != nil { - w.WriteHeader(500) + w.WriteHeader(http.StatusInternalServerError) w.Write([]byte(fmt.Sprintf(ErrorInvalidScreenId, err))) return } if _, err := uuid.Parse(lineId); err != nil { - w.WriteHeader(500) + w.WriteHeader(http.StatusInternalServerError) w.Write([]byte(fmt.Sprintf(ErrorInvalidLineId, err))) return } @@ -335,7 +339,7 @@ func HandleGetPtyOut(w http.ResponseWriter, r *http.Request) { w.WriteHeader(http.StatusOK) return } - w.WriteHeader(500) + w.WriteHeader(http.StatusInternalServerError) w.Write([]byte(html.EscapeString(fmt.Sprintf("error reading ptyout file: %v", err)))) return } @@ -510,49 +514,49 @@ func HandleReadFile(w http.ResponseWriter, r *http.Request) { contentType = "application/octet-stream" } if screenId == "" || lineId == "" { - w.WriteHeader(500) + w.WriteHeader(http.StatusInternalServerError) w.Write([]byte("must specify sessionid, screenid, and lineid")) return } if path == "" { - w.WriteHeader(500) + w.WriteHeader(http.StatusInternalServerError) w.Write([]byte("must specify path")) return } if _, err := uuid.Parse(screenId); err != nil { - w.WriteHeader(500) + w.WriteHeader(http.StatusInternalServerError) w.Write([]byte(fmt.Sprintf(ErrorInvalidScreenId, err))) return } if _, err := uuid.Parse(lineId); err != nil { - w.WriteHeader(500) + w.WriteHeader(http.StatusInternalServerError) w.Write([]byte(fmt.Sprintf(ErrorInvalidLineId, err))) return } if !ContentTypeHeaderValidRe.MatchString(contentType) { - w.WriteHeader(500) + w.WriteHeader(http.StatusInternalServerError) w.Write([]byte("invalid mimetype specified")) return } _, cmd, err := sstore.GetLineCmdByLineId(r.Context(), screenId, lineId) if err != nil { - w.WriteHeader(500) + w.WriteHeader(http.StatusInternalServerError) w.Write([]byte(fmt.Sprintf("invalid lineid: %v", err))) return } if cmd == nil { - w.WriteHeader(500) + w.WriteHeader(http.StatusInternalServerError) w.Write([]byte("invalid line, no cmd")) return } if cmd.Remote.RemoteId == "" { - w.WriteHeader(500) + w.WriteHeader(http.StatusInternalServerError) w.Write([]byte("invalid line, no remote")) return } msh := remote.GetRemoteById(cmd.Remote.RemoteId) if msh == nil { - w.WriteHeader(500) + w.WriteHeader(http.StatusInternalServerError) w.Write([]byte("invalid line, cannot resolve remote")) return } @@ -572,25 +576,25 @@ func HandleReadFile(w http.ResponseWriter, r *http.Request) { } iter, err := msh.StreamFile(r.Context(), streamPk) if err != nil { - w.WriteHeader(500) + w.WriteHeader(http.StatusInternalServerError) w.Write([]byte(fmt.Sprintf("error trying to stream file: %v", err))) return } defer iter.Close() respIf, err := iter.Next(r.Context()) if err != nil { - w.WriteHeader(500) + w.WriteHeader(http.StatusInternalServerError) w.Write([]byte(fmt.Sprintf("error getting streamfile response: %v", err))) return } resp, ok := respIf.(*packet.StreamFileResponseType) if !ok { - w.WriteHeader(500) + w.WriteHeader(http.StatusInternalServerError) w.Write([]byte(fmt.Sprintf("bad response packet type: %T", respIf))) return } if resp.Error != "" { - w.WriteHeader(500) + w.WriteHeader(http.StatusInternalServerError) w.Write([]byte(fmt.Sprintf("error response: %s", resp.Error))) return } @@ -622,7 +626,7 @@ func HandleReadFile(w http.ResponseWriter, r *http.Request) { func WriteJsonError(w http.ResponseWriter, errVal error) { w.Header().Set(ContentTypeHeaderKey, ContentTypeJson) - w.WriteHeader(200) + w.WriteHeader(http.StatusOK) errMap := make(map[string]interface{}) errMap["error"] = errVal.Error() errorCode := base.GetErrorCode(errVal) @@ -645,7 +649,7 @@ func WriteJsonSuccess(w http.ResponseWriter, data interface{}) { WriteJsonError(w, err) return } - w.WriteHeader(200) + w.WriteHeader(http.StatusOK) w.Write(barr) } @@ -664,7 +668,7 @@ func HandleRunCommand(w http.ResponseWriter, r *http.Request) { var commandPk scpacket.FeCommandPacketType err := decoder.Decode(&commandPk) if err != nil { - WriteJsonError(w, fmt.Errorf("error decoding json: %w", err)) + WriteJsonError(w, fmt.Errorf(ErrorDecodingJson, err)) return } update, err := cmdrunner.HandleCommand(r.Context(), &commandPk) @@ -678,29 +682,111 @@ func HandleRunCommand(w http.ResponseWriter, r *http.Request) { WriteJsonSuccess(w, update) } +func HandleRunEphemeralCommand(w http.ResponseWriter, r *http.Request) { + defer func() { + r := recover() + if r == nil { + return + } + log.Printf("[error] in run-ephemeral-command: %v\n", r) + debug.PrintStack() + WriteJsonError(w, fmt.Errorf(ErrorPanic, r)) + }() + w.Header().Set(CacheControlHeaderKey, CacheControlHeaderNoCache) + decoder := json.NewDecoder(r.Body) + var commandPk scpacket.FeCommandPacketType + err := decoder.Decode(&commandPk) + if err != nil { + WriteJsonError(w, fmt.Errorf(ErrorDecodingJson, err)) + return + } + log.Printf("Running ephemeral command: %v\n", commandPk) + + if commandPk.EphemeralOpts == nil { + commandPk.EphemeralOpts = &ephemeral.EphemeralRunOpts{} + } + + if commandPk.EphemeralOpts.TimeoutMs == 0 { + commandPk.EphemeralOpts.TimeoutMs = ephemeral.DefaultEphemeralTimeoutMs + } + + // These need to be defined here so we can use the methods of the BufferedPipe that are not part of io.WriteCloser + var stdoutPipe, stderrPipe *bufferedpipe.BufferedPipe + + if commandPk.EphemeralOpts.ExpectsResponse { + // Create new buffered pipes for stdout and stderr + stdoutPipe = bufferedpipe.NewBufferedPipe(ephemeral.DefaultEphemeralTimeoutDuration) + commandPk.EphemeralOpts.StdoutWriter = stdoutPipe + stderrPipe = bufferedpipe.NewBufferedPipe(ephemeral.DefaultEphemeralTimeoutDuration) + commandPk.EphemeralOpts.StderrWriter = stderrPipe + } + + update, err := cmdrunner.HandleCommand(r.Context(), &commandPk) + if err != nil { + log.Printf("Error occurred while running ephemeral command: %v\n", err) + if commandPk.EphemeralOpts.ExpectsResponse { + log.Printf("Closing buffered pipes\n") + stdoutPipe.Close() + stderrPipe.Close() + } + WriteJsonError(w, err) + return + } + + resp := scpacket.EphemeralCommandResponsePacketType{} + + // No error occurred, so we can write the response to the client + if commandPk.EphemeralOpts.ExpectsResponse { + // If the client expects a response, we need to send the urls of the stdout and stderr outputs + stdoutUrl, err := stdoutPipe.GetOutputUrl() + if err != nil { + log.Printf("Error occurred while getting stdout url: %v\n", err) + WriteJsonError(w, err) + return + } + resp.StdoutUrl = stdoutUrl + stderrUrl, err := stderrPipe.GetOutputUrl() + if err != nil { + log.Printf("Error occurred while getting stderr url: %v\n", err) + WriteJsonError(w, err) + return + } + resp.StderrUrl = stderrUrl + } + + WriteJsonSuccess(w, resp) + + // With ephemeral commands, we can't send the update back directly, so we need to send it through the update bus + if update != nil { + log.Printf("Sending update to main update bus\n") + update.Clean() + scbus.MainUpdateBus.DoUpdate(update) + } +} + func CheckIsDir(dirHandler http.Handler, fileHandler http.Handler) http.Handler { return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { configPath := r.URL.Path configAbsPath, err := filepath.Abs(configPath) if err != nil { - w.WriteHeader(500) + w.WriteHeader(http.StatusInternalServerError) w.Write([]byte(fmt.Sprintf("error getting absolute path: %v", err))) return } configBaseDir := path.Join(scbase.GetWaveHomeDir(), "config") configFullPath := path.Join(scbase.GetWaveHomeDir(), configAbsPath) if !strings.HasPrefix(configFullPath, configBaseDir) { - w.WriteHeader(500) + w.WriteHeader(http.StatusInternalServerError) w.Write([]byte(fmt.Sprintf("error: path is not in config folder"))) return } fstat, err := os.Stat(configFullPath) if errors.Is(err, fs.ErrNotExist) { - w.WriteHeader(404) + w.WriteHeader(http.StatusNotFound) w.Write([]byte(fmt.Sprintf("file not found: %v", configAbsPath))) return } else if err != nil { - w.WriteHeader(500) + w.WriteHeader(http.StatusInternalServerError) w.Write([]byte(fmt.Sprintf("file stat err: %v", err))) return } @@ -717,12 +803,12 @@ func AuthKeyMiddleWare(next http.Handler) http.Handler { reqAuthKey := r.Header.Get("X-AuthKey") w.Header().Set(CacheControlHeaderKey, CacheControlHeaderNoCache) if reqAuthKey == "" { - w.WriteHeader(500) + w.WriteHeader(http.StatusInternalServerError) w.Write([]byte("no x-authkey header")) return } if reqAuthKey != scbase.WaveAuthKey { - w.WriteHeader(500) + w.WriteHeader(http.StatusInternalServerError) w.Write([]byte("x-authkey header is invalid")) return } @@ -737,19 +823,19 @@ func AuthKeyWrapAllowHmac(fn WebFnType) WebFnType { // try hmac qvals := r.URL.Query() if !qvals.Has("hmac") { - w.WriteHeader(500) + w.WriteHeader(http.StatusInternalServerError) w.Write([]byte("no x-authkey header")) return } hmacOk, err := promptenc.ValidateUrlHmac([]byte(scbase.WaveAuthKey), r.URL.Path, qvals) if err != nil || !hmacOk { - w.WriteHeader(500) + w.WriteHeader(http.StatusInternalServerError) w.Write([]byte(fmt.Sprintf("error validating hmac"))) return } // fallthrough (hmac is valid) } else if reqAuthKey != scbase.WaveAuthKey { - w.WriteHeader(500) + w.WriteHeader(http.StatusInternalServerError) w.Write([]byte("x-authkey header is invalid")) return } @@ -763,12 +849,12 @@ func AuthKeyWrap(fn WebFnType) WebFnType { return func(w http.ResponseWriter, r *http.Request) { reqAuthKey := r.Header.Get("X-AuthKey") if reqAuthKey == "" { - w.WriteHeader(500) + w.WriteHeader(http.StatusInternalServerError) w.Write([]byte("no x-authkey header")) return } if reqAuthKey != scbase.WaveAuthKey { - w.WriteHeader(500) + w.WriteHeader(http.StatusInternalServerError) w.Write([]byte("x-authkey header is invalid")) return } @@ -898,13 +984,13 @@ func configDirHandler(w http.ResponseWriter, r *http.Request) { configFullPath := path.Join(scbase.GetWaveHomeDir(), configPath) dirFile, err := os.Open(configFullPath) if err != nil { - w.WriteHeader(500) + w.WriteHeader(http.StatusInternalServerError) w.Write([]byte(fmt.Sprintf("error opening specified dir: %v", err))) return } entries, err := dirFile.Readdir(0) if err != nil { - w.WriteHeader(500) + w.WriteHeader(http.StatusInternalServerError) w.Write([]byte(fmt.Sprintf("error getting files: %v", err))) return } @@ -916,11 +1002,11 @@ func configDirHandler(w http.ResponseWriter, r *http.Request) { } dirListJson, err := json.Marshal(files) if err != nil { - w.WriteHeader(500) + w.WriteHeader(http.StatusInternalServerError) w.Write([]byte(fmt.Sprintf("json err: %v", err))) return } - w.WriteHeader(200) + w.WriteHeader(http.StatusOK) w.Header().Set("Content-Type", "application/json") w.Write(dirListJson) } @@ -1014,6 +1100,8 @@ func main() { gr.HandleFunc("/api/rtnstate", AuthKeyWrap(HandleRtnState)) gr.HandleFunc("/api/get-screen-lines", AuthKeyWrap(HandleGetScreenLines)) gr.HandleFunc("/api/run-command", AuthKeyWrap(HandleRunCommand)).Methods("POST") + gr.HandleFunc("/api/run-ephemeral-command", AuthKeyWrap(HandleRunEphemeralCommand)).Methods("POST") + gr.HandleFunc(bufferedpipe.BufferedPipeGetterUrl, AuthKeyWrapAllowHmac(bufferedpipe.HandleGetBufferedPipeOutput)) gr.HandleFunc("/api/get-client-data", AuthKeyWrap(HandleGetClientData)) gr.HandleFunc("/api/set-winsize", AuthKeyWrap(HandleSetWinSize)) gr.HandleFunc("/api/log-active-state", AuthKeyWrap(HandleLogActiveState)) diff --git a/wavesrv/pkg/bufferedpipe/bufferedpipe.go b/wavesrv/pkg/bufferedpipe/bufferedpipe.go new file mode 100644 index 00000000..3877c443 --- /dev/null +++ b/wavesrv/pkg/bufferedpipe/bufferedpipe.go @@ -0,0 +1,172 @@ +// Copyright 2023, Command Line Inc. +// SPDX-License-Identifier: Apache-2.0 + +// Provides a mechanism for writing data to a buffer and reading it later. The buffer is stored in a map of buffered pipes, which are removed after a certain amount of time. The output of a buffered pipe can be read by sending a GET request to a specific URL. +package bufferedpipe + +import ( + "bytes" + "io" + "log" + "net/http" + "net/url" + "sync" + "sync/atomic" + "time" + + "github.com/google/uuid" + "github.com/wavetermdev/waveterm/wavesrv/pkg/promptenc" + "github.com/wavetermdev/waveterm/wavesrv/pkg/scbase" +) + +const ( + BufferedPipeMapTTL = 30 * time.Second // The time-to-live for a buffered pipe in the map of buffered pipes. + BufferedPipeGetterUrl = "/api/buffered-pipe" // The URL for getting the output of a buffered pipe. +) + +// A pipe that allows for lazy writing to a downstream writer. Data written to the pipe is buffered until WriteTo is called. +type BufferedPipe struct { + Key string // a unique key for the pipe + buffer bytes.Buffer // buffer of data to be written to the downstream writer once it is ready + closed atomic.Bool // whether the pipe has been closed + bufferDataCond *sync.Cond // Condition variable to signal waiting writers that there is either data to write or the pipe has been closed + downstreamLock *sync.Mutex // Lock to ensure that only one goroutine can read from the buffer at a time +} + +// Create a new BufferedPipe with a timeout. The writer will be closed after the timeout +func NewBufferedPipe(timeout time.Duration) *BufferedPipe { + newPipe := &BufferedPipe{ + Key: uuid.New().String(), + buffer: bytes.Buffer{}, + closed: atomic.Bool{}, + bufferDataCond: &sync.Cond{L: &sync.Mutex{}}, + downstreamLock: &sync.Mutex{}, + } + SetBufferedPipe(newPipe) + time.AfterFunc(timeout, func() { + newPipe.Close() + }) + return newPipe +} + +// Get the URL for reading the output of the pipe. +func (pipe *BufferedPipe) GetOutputUrl() (string, error) { + qvals := make(url.Values) + qvals.Set("key", pipe.Key) + qvals.Set("nonce", uuid.New().String()) + hmacStr, err := promptenc.ComputeUrlHmac([]byte(scbase.WaveAuthKey), BufferedPipeGetterUrl, qvals) + if err != nil { + return "", err + } + + qvals.Set("hmac", hmacStr) + return BufferedPipeGetterUrl + "?" + qvals.Encode(), nil +} + +// Write data to the buffer. +func (pipe *BufferedPipe) Write(p []byte) (n int, err error) { + if pipe.closed.Load() { + return 0, io.ErrClosedPipe + } + + defer func() { + pipe.bufferDataCond.L.Unlock() + pipe.bufferDataCond.Broadcast() + }() + pipe.bufferDataCond.L.Lock() + + return pipe.buffer.Write(p) +} + +// Write all buffered data to a waiting writer and block, sending all subsequent data until the pipe is closed. Only one goroutine should call this method. +func (pipe *BufferedPipe) WriteTo(w io.Writer) (n int64, err error) { + // Lock the buffer to ensure that only one downstream writer can read from it at a time. + if !pipe.downstreamLock.TryLock() { + return 0, io.ErrClosedPipe + } + + defer func() { + pipe.bufferDataCond.L.Unlock() + pipe.downstreamLock.Unlock() + }() + pipe.bufferDataCond.L.Lock() + for { + n1, err := pipe.buffer.WriteTo(w) + if err != nil { + return n, err + } + n += n1 + + // Check if the pipe has been closed. If it has, we don't need to wait for more data. + if pipe.closed.Load() { + break + } + + // Wait for more data to be written to the buffer or for the pipe to be closed. + pipe.bufferDataCond.Wait() + } + return n, nil +} + +// Close the pipe. This will cause any blocking WriteTo calls to return. +func (pipe *BufferedPipe) Close() error { + defer pipe.bufferDataCond.Broadcast() + pipe.closed.Store(true) + return nil +} + +// Ensure that BufferedPipe implements the io.WriteCloser and io.WriterTo interfaces. +var _ io.WriteCloser = (*BufferedPipe)(nil) +var _ io.WriterTo = (*BufferedPipe)(nil) + +type BufferedPipeMap struct { + _map map[string]*BufferedPipe + lock sync.Mutex +} + +// A global map of registered buffered pipes. +var bufferedPipes = BufferedPipeMap{_map: make(map[string]*BufferedPipe)} + +// Get a buffered pipe from the map of buffered pipes, given a key. +func GetBufferedPipe(key string) (*BufferedPipe, bool) { + bufferedPipes.lock.Lock() + defer bufferedPipes.lock.Unlock() + + ewc, ok := bufferedPipes._map[key] + return ewc, ok +} + +// Set a buffered pipe in the map of buffered pipes. +func SetBufferedPipe(pipe *BufferedPipe) { + bufferedPipes.lock.Lock() + defer bufferedPipes.lock.Unlock() + key := pipe.Key + bufferedPipes._map[key] = pipe + + // Remove the buffered pipe after a certain amount of time + time.AfterFunc(BufferedPipeMapTTL, func() { + bufferedPipes.lock.Lock() + defer bufferedPipes.lock.Unlock() + pipe.Close() + log.Printf("removing buffered pipe %s", key) + delete(bufferedPipes._map, key) + }) +} + +// Handle a HTTP GET request to get the output of a buffered pipe, given a key. +func HandleGetBufferedPipeOutput(w http.ResponseWriter, r *http.Request) { + qvals := r.URL.Query() + key := qvals.Get("key") + pipe, ok := GetBufferedPipe(key) + if !ok { + http.Error(w, "buffered pipe not found", http.StatusNotFound) + return + } + + w.Header().Set("Content-Type", "text/plain") + _, err := pipe.WriteTo(w) + if err != nil { + http.Error(w, "error writing from buffer", http.StatusInternalServerError) + return + } +} diff --git a/wavesrv/pkg/cmdrunner/cmdrunner.go b/wavesrv/pkg/cmdrunner/cmdrunner.go index 48f4fab9..42cdca1f 100644 --- a/wavesrv/pkg/cmdrunner/cmdrunner.go +++ b/wavesrv/pkg/cmdrunner/cmdrunner.go @@ -37,6 +37,7 @@ import ( "github.com/wavetermdev/waveterm/wavesrv/pkg/bookmarks" "github.com/wavetermdev/waveterm/wavesrv/pkg/comp" "github.com/wavetermdev/waveterm/wavesrv/pkg/dbutil" + "github.com/wavetermdev/waveterm/wavesrv/pkg/ephemeral" "github.com/wavetermdev/waveterm/wavesrv/pkg/history" "github.com/wavetermdev/waveterm/wavesrv/pkg/pcloud" "github.com/wavetermdev/waveterm/wavesrv/pkg/promptenc" @@ -538,10 +539,10 @@ func SyncCommand(ctx context.Context, pk *scpacket.FeCommandPacketType) (scbus.U runPacket.Command = ":" runPacket.ReturnState = true rcOpts := remote.RunCommandOpts{ - SessionId: ids.SessionId, - ScreenId: ids.ScreenId, - RemotePtr: ids.Remote.RemotePtr, - Ephemeral: true, + SessionId: ids.SessionId, + ScreenId: ids.ScreenId, + RemotePtr: ids.Remote.RemotePtr, + EphemeralOpts: &ephemeral.EphemeralRunOpts{TimeoutMs: ephemeral.DefaultEphemeralTimeoutMs}, } _, callback, err := remote.RunCommand(ctx, rcOpts, runPacket) if callback != nil { @@ -620,6 +621,7 @@ func RunCommand(ctx context.Context, pk *scpacket.FeCommandPacketType) (scbus.Up newPk.RawStr = pk.RawStr newPk.UIContext = pk.UIContext newPk.Interactive = pk.Interactive + newPk.EphemeralOpts = pk.EphemeralOpts evalDepth := getEvalDepth(ctx) ctxWithDepth := context.WithValue(ctx, depthContextKey, evalDepth+1) return EvalCommand(ctxWithDepth, newPk) @@ -638,9 +640,10 @@ func RunCommand(ctx context.Context, pk *scpacket.FeCommandPacketType) (scbus.Up runPacket.Command = strings.TrimSpace(cmdStr) runPacket.ReturnState = resolveBool(pk.Kwargs["rtnstate"], isRtnStateCmd) rcOpts := remote.RunCommandOpts{ - SessionId: ids.SessionId, - ScreenId: ids.ScreenId, - RemotePtr: ids.Remote.RemotePtr, + SessionId: ids.SessionId, + ScreenId: ids.ScreenId, + RemotePtr: ids.Remote.RemotePtr, + EphemeralOpts: pk.EphemeralOpts, } cmd, callback, err := remote.RunCommand(ctx, rcOpts, runPacket) if callback != nil { @@ -657,15 +660,19 @@ func RunCommand(ctx context.Context, pk *scpacket.FeCommandPacketType) (scbus.Up if langArg != "" { lineState[sstore.LineState_Lang] = langArg } - update, err := addLineForCmd(ctx, "/run", true, ids, cmd, renderer, lineState) - if err != nil { - return nil, err + + // If we are running an ephemeral command, we don't want to add the line to the screen + if pk.EphemeralOpts == nil { + update, err := addLineForCmd(ctx, "/run", true, ids, cmd, renderer, lineState) + if err != nil { + return nil, err + } + update.AddUpdate(sstore.InteractiveUpdate(pk.Interactive)) + // this update is sent asynchronously for timing issues. the cmd update comes async as well + // so if we return this directly it sometimes gets evaluated first. by pushing it on the MainBus + // it ensures it happens after the command creation event. + scbus.MainUpdateBus.DoScreenUpdate(ids.ScreenId, update) } - update.AddUpdate(sstore.InteractiveUpdate(pk.Interactive)) - // this update is sent asynchronously for timing issues. the cmd update comes async as well - // so if we return this directly it sometimes gets evaluated first. by pushing it on the MainBus - // it ensures it happens after the command creation event. - scbus.MainUpdateBus.DoScreenUpdate(ids.ScreenId, update) return nil, nil } @@ -737,6 +744,7 @@ func EvalCommand(ctx context.Context, pk *scpacket.FeCommandPacketType) (scbus.U ctxWithHistory := context.WithValue(ctx, historyContextKey, &historyContext) var update scbus.UpdatePacket newPk, rtnErr := EvalMetaCommand(ctxWithHistory, pk) + if rtnErr == nil { update, rtnErr = HandleCommand(ctxWithHistory, newPk) } else { @@ -752,7 +760,8 @@ func EvalCommand(ctx context.Context, pk *scpacket.FeCommandPacketType) (scbus.U } var hasModelUpdate bool var modelUpdate *scbus.ModelUpdatePacketType - if update == nil { + if update == nil && newPk.EphemeralOpts == nil { + // We don't want to serve an update if we are processing an ephemeral command hasModelUpdate = true modelUpdate = scbus.MakeUpdatePacket() update = modelUpdate diff --git a/wavesrv/pkg/cmdrunner/shparse.go b/wavesrv/pkg/cmdrunner/shparse.go index b0011b9d..61aacfcc 100644 --- a/wavesrv/pkg/cmdrunner/shparse.go +++ b/wavesrv/pkg/cmdrunner/shparse.go @@ -373,6 +373,7 @@ func EvalMetaCommand(ctx context.Context, origPk *scpacket.FeCommandPacketType) rtnPk.UIContext = origPk.UIContext rtnPk.RawStr = origPk.RawStr rtnPk.Interactive = origPk.Interactive + rtnPk.EphemeralOpts = origPk.EphemeralOpts for key, val := range origPk.Kwargs { rtnPk.Kwargs[key] = val } diff --git a/wavesrv/pkg/ephemeral/ephemeral.go b/wavesrv/pkg/ephemeral/ephemeral.go new file mode 100644 index 00000000..4b3c11fe --- /dev/null +++ b/wavesrv/pkg/ephemeral/ephemeral.go @@ -0,0 +1,28 @@ +// Copyright 2024, Command Line Inc. +// SPDX-License-Identifier: Apache-2.0 + +// Manage additional options for ephemeral commands (commands that are not saved to the history). +package ephemeral + +import ( + "io" + "sync/atomic" + "time" +) + +const ( + DefaultEphemeralTimeoutMs = 5000 // The default timeout for ephemeral commands in milliseconds. + DefaultEphemeralTimeoutDuration = DefaultEphemeralTimeoutMs * time.Millisecond // The default timeout for ephemeral commands as a time.Duration. +) + +// Options specific to ephemeral commands (commands that are not saved to the history) +type EphemeralRunOpts struct { + Env map[string]string `json:"env,omitempty"` // Environment variables to set for the command. + OverrideCwd string `json:"overridecwd,omitempty"` // A directory to use as the current working directory. Defaults to the last set shell state. + UsePty bool `json:"usepty"` // If set, the command is run in a pseudo-terminal and all output will be written to the StdoutWriter. If not set, the command is run in a normal shell and the output is written to the StdoutWriter and StderrWriter. + TimeoutMs int64 `json:"timeoutms"` // The maximum time to wait for the command to complete. If the command does not complete within this time, it is killed. + ExpectsResponse bool `json:"expectsresponse"` // If set, the command is expected to return a response. If this is false, ResposeWriter is not set. + StdoutWriter io.WriteCloser `json:"-"` // A writer to receive the command's stdout. If not set, the command's output is discarded. (set by remote.go) + StderrWriter io.WriteCloser `json:"-"` // A writer to receive the command's stderr. If not set, the command's output is discarded. (set by remote.go) + Canceled atomic.Bool `json:"canceled,omitempty"` // If set, the command was canceled before it completed. +} diff --git a/wavesrv/pkg/remote/remote.go b/wavesrv/pkg/remote/remote.go index 7b3375ad..0328905c 100644 --- a/wavesrv/pkg/remote/remote.go +++ b/wavesrv/pkg/remote/remote.go @@ -19,7 +19,6 @@ import ( "strconv" "strings" "sync" - "sync/atomic" "syscall" "time" @@ -35,6 +34,7 @@ import ( "github.com/wavetermdev/waveterm/waveshell/pkg/shexec" "github.com/wavetermdev/waveterm/waveshell/pkg/statediff" "github.com/wavetermdev/waveterm/waveshell/pkg/utilfn" + "github.com/wavetermdev/waveterm/wavesrv/pkg/ephemeral" "github.com/wavetermdev/waveterm/wavesrv/pkg/scbase" "github.com/wavetermdev/waveterm/wavesrv/pkg/scbus" "github.com/wavetermdev/waveterm/wavesrv/pkg/scpacket" @@ -172,13 +172,12 @@ type CommandInputSink interface { } type RunCmdType struct { - CK base.CommandKey - SessionId string - ScreenId string - RemotePtr sstore.RemotePtrType - RunPacket *packet.RunPacketType - Ephemeral bool - EphCancled atomic.Bool // only for Ephemeral commands, if true, then the command result should be discarded + CK base.CommandKey + SessionId string + ScreenId string + RemotePtr sstore.RemotePtrType + RunPacket *packet.RunPacketType + EphemeralOpts *ephemeral.EphemeralRunOpts } type ReinitCommandSink struct { @@ -1892,8 +1891,8 @@ type RunCommandOpts struct { NoCreateCmdPtyFile bool // this command will not go into the DB, and will not have a ptyout file created - // forces special packet handling (sets RunCommandType.Ephemeral) - Ephemeral bool + // forces special packet handling (sets RunCommandType.EphemeralOpts) + EphemeralOpts *ephemeral.EphemeralRunOpts } // returns (CmdType, allow-updates-callback, err) @@ -1924,6 +1923,10 @@ func RunCommand(ctx context.Context, rcOpts RunCommandOpts, runPacket *packet.Ru return nil, nil, fmt.Errorf("runPacket.StatePtr should not be set, it is set in RunCommand") } + if rcOpts.EphemeralOpts != nil { + log.Printf("[info] running ephemeral command ck: %s\n", runPacket.CK) + } + // pending state command logic // if we are currently running a command that can change the state, we need to wait for it to finish if rcOpts.StatePtr == nil { @@ -1933,9 +1936,10 @@ func RunCommand(ctx context.Context, rcOpts RunCommandOpts, runPacket *packet.Ru } ok, existingRct := msh.testAndSetPendingStateCmd(screenId, remotePtr, newPSC) if !ok { - if existingRct.Ephemeral { + if rcOpts.EphemeralOpts != nil { // if the existing command is ephemeral, we cancel it and continue - existingRct.EphCancled.Store(true) + log.Printf("[warning] canceling existing ephemeral state cmd: %s\n", existingRct.CK) + rcOpts.EphemeralOpts.Canceled.Store(true) } else { line, _, err := sstore.GetLineCmdByLineId(ctx, screenId, existingRct.CK.GetCmdId()) return nil, nil, makePSCLineError(existingRct.CK, line, err) @@ -1960,15 +1964,37 @@ func RunCommand(ctx context.Context, rcOpts RunCommandOpts, runPacket *packet.Ru var err error statePtr, err = sstore.GetRemoteStatePtr(ctx, sessionId, screenId, remotePtr) if err != nil { + log.Printf("[error] RunCommand: cannot get remote state: %v\n", err) return nil, nil, fmt.Errorf("cannot run command: %w", err) } if statePtr == nil { + log.Printf("[error] RunCommand: no valid shell state found\n") return nil, nil, fmt.Errorf("cannot run command: no valid shell state found") } } // statePtr will not be nil runPacket.StatePtr = statePtr currentState, err := sstore.GetFullState(ctx, *statePtr) + + if rcOpts.EphemeralOpts != nil { + // Setting UsePty to false will ensure that the outputs get written to the correct file descriptors to extract stdout and stderr + runPacket.UsePty = rcOpts.EphemeralOpts.UsePty + + // Ephemeral commands can override the cwd without persisting it to the DB + if rcOpts.EphemeralOpts.OverrideCwd != "" { + currentState.Cwd = rcOpts.EphemeralOpts.OverrideCwd + } + + // Ephemeral commands can override the env without persisting it to the DB + if len(rcOpts.EphemeralOpts.Env) > 0 { + curEnvs := shellenv.DeclMapFromState(currentState) + for key, val := range rcOpts.EphemeralOpts.Env { + curEnvs[key] = &shellenv.DeclareDeclType{Name: key, Value: val, Args: "x"} + } + currentState.ShellVars = shellenv.SerializeDeclMap(curEnvs) + } + } + if err != nil || currentState == nil { return nil, nil, fmt.Errorf("cannot load current remote state: %w", err) } @@ -2032,21 +2058,21 @@ func RunCommand(ctx context.Context, rcOpts RunCommandOpts, runPacket *packet.Ru RunOut: nil, RtnState: runPacket.ReturnState, } - if !rcOpts.NoCreateCmdPtyFile && !rcOpts.Ephemeral { + if !rcOpts.NoCreateCmdPtyFile && rcOpts.EphemeralOpts == nil { err = sstore.CreateCmdPtyFile(ctx, cmd.ScreenId, cmd.LineId, cmd.TermOpts.MaxPtySize) if err != nil { // TODO the cmd is running, so this is a tricky error to handle return nil, nil, fmt.Errorf("cannot create local ptyout file for running command: %v", err) } } - msh.AddRunningCmd(&RunCmdType{ - CK: runPacket.CK, - SessionId: sessionId, - ScreenId: screenId, - RemotePtr: remotePtr, - RunPacket: runPacket, - Ephemeral: rcOpts.Ephemeral, - }) + runningCmdType := &RunCmdType{ + CK: runPacket.CK, + SessionId: sessionId, + ScreenId: screenId, + RemotePtr: remotePtr, + RunPacket: runPacket, + EphemeralOpts: rcOpts.EphemeralOpts} + msh.AddRunningCmd(runningCmdType) return cmd, func() { removeCmdWait(runPacket.CK) }, nil } @@ -2121,13 +2147,17 @@ func (msh *MShellProc) HandleFeInput(inputPk *scpacket.FeInputPacketType) error func (msh *MShellProc) AddRunningCmd(rct *RunCmdType) { msh.Lock.Lock() defer msh.Lock.Unlock() + if rct.EphemeralOpts != nil { + log.Printf("[info] adding ephemeral running command: %s\n", rct.CK) + } msh.RunningCmds[rct.RunPacket.CK] = rct } func (msh *MShellProc) GetRunningCmd(ck base.CommandKey) *RunCmdType { msh.Lock.Lock() defer msh.Lock.Unlock() - return msh.RunningCmds[ck] + rtn := msh.RunningCmds[ck] + return rtn } func (msh *MShellProc) RemoveRunningCmd(ck base.CommandKey) { @@ -2319,14 +2349,15 @@ func (msh *MShellProc) handleCmdDonePacket(rct *RunCmdType, donePk *packet.CmdDo } // this will remove from RunningCmds and from PendingStateCmds defer msh.RemoveRunningCmd(donePk.CK) - if rct.Ephemeral && rct.EphCancled.Load() { + if rct.EphemeralOpts != nil && rct.EphemeralOpts.Canceled.Load() { + log.Printf("cmddone %s (ephemeral canceled)\n", donePk.CK) // do nothing when an ephemeral command is canceled return } ctx, cancelFn := context.WithTimeout(context.Background(), 5*time.Second) defer cancelFn() update := scbus.MakeUpdatePacket() - if !rct.Ephemeral { + if rct.EphemeralOpts == nil { // only update DB for non-ephemeral commands err := sstore.UpdateCmdDoneInfo(ctx, update, donePk.CK, donePk, sstore.CmdStatusDone) if err != nil { @@ -2342,6 +2373,14 @@ func (msh *MShellProc) handleCmdDonePacket(rct *RunCmdType, donePk *packet.CmdDo update.AddUpdate(*screen) } } + + // Close the ephemeral response writer if it exists + if rct.EphemeralOpts != nil && rct.EphemeralOpts.ExpectsResponse { + log.Printf("closing ephemeral response writers\n") + defer rct.EphemeralOpts.StdoutWriter.Close() + defer rct.EphemeralOpts.StderrWriter.Close() + } + // ephemeral commands *do* update the remote state // not all commands get a final state (only RtnState commands have this returned) // so in those cases finalState will be nil @@ -2360,7 +2399,7 @@ func (msh *MShellProc) handleCmdDonePacket(rct *RunCmdType, donePk *packet.CmdDo update.AddUpdate(sstore.MakeSessionUpdateForRemote(rct.SessionId, newRI)) } // ephemeral commands *do not* update cmd state (there is no command) - if newRI != nil && !rct.Ephemeral { + if newRI != nil && rct.EphemeralOpts == nil { newRIStatePtr := packet.ShellStatePtr{BaseHash: newRI.StateBaseHash, DiffHashArr: newRI.StateDiffHashArr} err = sstore.UpdateCmdRtnState(ctx, donePk.CK, newRIStatePtr) if err != nil { @@ -2378,10 +2417,6 @@ func (msh *MShellProc) handleCmdFinalPacket(rct *RunCmdType, finalPk *packet.Cmd return } defer msh.RemoveRunningCmd(finalPk.CK) - if rct.Ephemeral { - // just remove the running command, but there is no DB state to update in this case - return - } rtnCmd, err := sstore.GetCmdByScreenId(context.Background(), finalPk.CK.GetGroupId(), finalPk.CK.GetCmdId()) if err != nil { log.Printf("error calling GetCmdById in handleCmdFinalPacket: %v\n", err) @@ -2420,21 +2455,41 @@ func (msh *MShellProc) ResetDataPos(ck base.CommandKey) { func (msh *MShellProc) handleDataPacket(rct *RunCmdType, dataPk *packet.DataPacketType, dataPosMap *utilfn.SyncMap[base.CommandKey, int64]) { if rct == nil { + log.Printf("error handling data packet: no running cmd found %s\n", dataPk.CK) ack := makeDataAckPacket(dataPk.CK, dataPk.FdNum, 0, fmt.Errorf("no running cmd found")) msh.ServerProc.Input.SendPacket(ack) return } realData, err := base64.StdEncoding.DecodeString(dataPk.Data64) if err != nil { + log.Printf("error decoding data packet: %v\n", err) ack := makeDataAckPacket(dataPk.CK, dataPk.FdNum, 0, err) msh.ServerProc.Input.SendPacket(ack) return } - if rct.Ephemeral { + if rct.EphemeralOpts != nil { + // Write to the response writer if it's set + if len(realData) > 0 && rct.EphemeralOpts.ExpectsResponse { + switch dataPk.FdNum { + case 1: + _, err := rct.EphemeralOpts.StdoutWriter.Write(realData) + if err != nil { + log.Printf("*error writing to ephemeral stdout writer: %v\n", err) + } + case 2: + _, err := rct.EphemeralOpts.StderrWriter.Write(realData) + if err != nil { + log.Printf("*error writing to ephemeral stderr writer: %v\n", err) + } + default: + log.Printf("error handling data packet: invalid fdnum %d\n", dataPk.FdNum) + } + } ack := makeDataAckPacket(dataPk.CK, dataPk.FdNum, len(realData), nil) msh.ServerProc.Input.SendPacket(ack) return } + var ack *packet.DataAckPacketType if len(realData) > 0 { dataPos := dataPosMap.Get(dataPk.CK) diff --git a/wavesrv/pkg/scpacket/scpacket.go b/wavesrv/pkg/scpacket/scpacket.go index 6e604643..df6a6423 100644 --- a/wavesrv/pkg/scpacket/scpacket.go +++ b/wavesrv/pkg/scpacket/scpacket.go @@ -15,9 +15,10 @@ import ( "github.com/wavetermdev/waveterm/waveshell/pkg/base" "github.com/wavetermdev/waveterm/waveshell/pkg/packet" "github.com/wavetermdev/waveterm/waveshell/pkg/utilfn" + "github.com/wavetermdev/waveterm/wavesrv/pkg/ephemeral" ) -var RemoteNameRe = regexp.MustCompile("^\\*?[a-zA-Z0-9_-]+$") +var RemoteNameRe = regexp.MustCompile(`^\*?[a-zA-Z0-9_-]+$`) type RemotePtrType struct { OwnerId string `json:"ownerid"` @@ -84,16 +85,18 @@ const WatchScreenPacketStr = "watchscreen" const FeInputPacketStr = "feinput" const RemoteInputPacketStr = "remoteinput" const CmdInputTextPacketStr = "cmdinputtext" +const EphemeralCommandResponsePacketStr = "ephemeralcommandresponse" type FeCommandPacketType struct { - Type string `json:"type"` - MetaCmd string `json:"metacmd"` - MetaSubCmd string `json:"metasubcmd,omitempty"` - Args []string `json:"args,omitempty"` - Kwargs map[string]string `json:"kwargs,omitempty"` - RawStr string `json:"rawstr,omitempty"` - UIContext *UIContextType `json:"uicontext,omitempty"` - Interactive bool `json:"interactive"` + Type string `json:"type"` + MetaCmd string `json:"metacmd"` + MetaSubCmd string `json:"metasubcmd,omitempty"` + Args []string `json:"args,omitempty"` + Kwargs map[string]string `json:"kwargs,omitempty"` + RawStr string `json:"rawstr,omitempty"` + UIContext *UIContextType `json:"uicontext,omitempty"` + Interactive bool `json:"interactive"` + EphemeralOpts *ephemeral.EphemeralRunOpts `json:"ephemeralopts,omitempty"` } func (pk *FeCommandPacketType) GetRawStr() string { @@ -121,6 +124,19 @@ func (pk *FeCommandPacketType) GetRawStr() string { return cmd + " " + strings.Join(args, " ") } +type EphemeralCommandResponsePacketType struct { + Type string `json:"type"` + StdoutUrl string `json:"stdouturl"` + StderrUrl string `json:"stderrurl"` + Error string `json:"error,omitempty"` +} + +func (*EphemeralCommandResponsePacketType) GetType() string { + return EphemeralCommandResponsePacketStr +} + +var _ PacketType = &EphemeralCommandResponsePacketType{} + type UIContextType struct { SessionId string `json:"sessionid"` ScreenId string `json:"screenid"` diff --git a/wavesrv/pkg/sstore/sstore.go b/wavesrv/pkg/sstore/sstore.go index 1c8ff9d7..2cc6a2a7 100644 --- a/wavesrv/pkg/sstore/sstore.go +++ b/wavesrv/pkg/sstore/sstore.go @@ -853,34 +853,6 @@ func (r *RemoteType) GetName() string { return r.RemoteCanonicalName } -type CmdType struct { - ScreenId string `json:"screenid"` - LineId string `json:"lineid"` - Remote RemotePtrType `json:"remote"` - CmdStr string `json:"cmdstr"` - RawCmdStr string `json:"rawcmdstr"` - FeState map[string]string `json:"festate"` - StatePtr packet.ShellStatePtr `json:"state"` - TermOpts TermOpts `json:"termopts"` - OrigTermOpts TermOpts `json:"origtermopts"` - Status string `json:"status"` - CmdPid int `json:"cmdpid"` - RemotePid int `json:"remotepid"` - RestartTs int64 `json:"restartts,omitempty"` - DoneTs int64 `json:"donets"` - ExitCode int `json:"exitcode"` - DurationMs int `json:"durationms"` - RunOut []packet.PacketType `json:"runout,omitempty"` - RtnState bool `json:"rtnstate,omitempty"` - RtnStatePtr packet.ShellStatePtr `json:"rtnstateptr,omitempty"` - Remove bool `json:"remove,omitempty"` // not persisted to DB - Restarted bool `json:"restarted,omitempty"` // not persisted to DB -} - -func (CmdType) GetType() string { - return "cmd" -} - func (r *RemoteType) ToMap() map[string]interface{} { rtn := make(map[string]interface{}) rtn["remoteid"] = r.RemoteId @@ -926,6 +898,34 @@ func (r *RemoteType) FromMap(m map[string]interface{}) bool { return true } +type CmdType struct { + ScreenId string `json:"screenid"` + LineId string `json:"lineid"` + Remote RemotePtrType `json:"remote"` + CmdStr string `json:"cmdstr"` + RawCmdStr string `json:"rawcmdstr"` + FeState map[string]string `json:"festate"` + StatePtr packet.ShellStatePtr `json:"state"` + TermOpts TermOpts `json:"termopts"` + OrigTermOpts TermOpts `json:"origtermopts"` + Status string `json:"status"` + CmdPid int `json:"cmdpid"` + RemotePid int `json:"remotepid"` + RestartTs int64 `json:"restartts,omitempty"` + DoneTs int64 `json:"donets"` + ExitCode int `json:"exitcode"` + DurationMs int `json:"durationms"` + RunOut []packet.PacketType `json:"runout,omitempty"` + RtnState bool `json:"rtnstate,omitempty"` + RtnStatePtr packet.ShellStatePtr `json:"rtnstateptr,omitempty"` + Remove bool `json:"remove,omitempty"` // not persisted to DB + Restarted bool `json:"restarted,omitempty"` // not persisted to DB +} + +func (CmdType) GetType() string { + return "cmd" +} + func (cmd *CmdType) ToMap() map[string]interface{} { rtn := make(map[string]interface{}) rtn["screenid"] = cmd.ScreenId