From 7dfcbef774e2c74c2d59a1ec0526096fcbc58f82 Mon Sep 17 00:00:00 2001 From: Aarnav Tale Date: Fri, 30 May 2025 19:35:47 -0400 Subject: [PATCH] feat: initial webssh tooling --- agent/cmd/hp_agent/hp_agent.go | 4 + agent/internal/hpagent/handler.go | 170 ++++++++++++++++++-------- agent/internal/sshutil/connect.go | 161 +++++++++++++++++++++++++ agent/internal/sshutil/encoder.go | 41 +++++++ agent/internal/sshutil/handler.go | 56 +++++++++ agent/internal/sshutil/sessions.go | 67 +++++++++++ agent/internal/util/logger.go | 37 ++++-- app/server/README.md | 4 + app/server/agent/dispatcher.ts | 36 ++++++ app/server/agent/ssh.ts | 185 +++++++++++++++++++++++++++++ app/server/index.ts | 24 +++- app/server/web/agent.ts | 14 ++- go.mod | 12 +- go.sum | 12 ++ package.json | 2 + pnpm-lock.yaml | 50 +++++--- 16 files changed, 792 insertions(+), 83 deletions(-) create mode 100644 agent/internal/sshutil/connect.go create mode 100644 agent/internal/sshutil/encoder.go create mode 100644 agent/internal/sshutil/handler.go create mode 100644 agent/internal/sshutil/sessions.go create mode 100644 app/server/agent/dispatcher.ts create mode 100644 app/server/agent/ssh.ts diff --git a/agent/cmd/hp_agent/hp_agent.go b/agent/cmd/hp_agent/hp_agent.go index 8212059..d4dcfe5 100644 --- a/agent/cmd/hp_agent/hp_agent.go +++ b/agent/cmd/hp_agent/hp_agent.go @@ -1,9 +1,12 @@ package main import ( + "os" + _ "github.com/joho/godotenv/autoload" "github.com/tale/headplane/agent/internal/config" "github.com/tale/headplane/agent/internal/hpagent" + "github.com/tale/headplane/agent/internal/sshutil" "github.com/tale/headplane/agent/internal/tsnet" "github.com/tale/headplane/agent/internal/util" ) @@ -31,5 +34,6 @@ func main() { ID: agent.ID, }) + sshutil.StartInputReader(os.NewFile(3, "sshin")) hpagent.FollowMaster(agent) } diff --git a/agent/internal/hpagent/handler.go b/agent/internal/hpagent/handler.go index 1228194..76868d6 100644 --- a/agent/internal/hpagent/handler.go +++ b/agent/internal/hpagent/handler.go @@ -2,13 +2,16 @@ package hpagent import ( "bufio" - "encoding/json" + "bytes" + // "encoding/json" "os" - "sync" + // "sync" + "github.com/fxamacker/cbor/v2" + "github.com/tale/headplane/agent/internal/sshutil" "github.com/tale/headplane/agent/internal/tsnet" "github.com/tale/headplane/agent/internal/util" - "tailscale.com/tailcfg" + // "tailscale.com/tailcfg" ) // Represents messages from the Headplane master @@ -16,6 +19,25 @@ type RecvMessage struct { NodeIDs []string } +type CborMessage struct { + Op string `cbor:"op"` + Payload cbor.RawMessage `cbor:"payload"` +} + +type SSHConnect struct { + SessionId string `cbor:"sessionId"` + Username string `cbor:"username"` + Hostname string `cbor:"hostname"` + Port int `cbor:"port"` +} + +type SSHMessage struct { + op string + username string + hostname string + Id string +} + type SendMessage struct { Type string Data any @@ -25,63 +47,111 @@ type SendMessage struct { func FollowMaster(agent *tsnet.TSAgent) { log := util.GetLogger() scanner := bufio.NewScanner(os.Stdin) + log.Info("Listening for messages from Headplane master on stdin") for scanner.Scan() { line := scanner.Bytes() + log.Info("Got bytes delimited by newline"); + + var msg CborMessage + decoder := cbor.NewDecoder(bytes.NewReader(line)) + err := decoder.Decode(&msg) - var msg RecvMessage - err := json.Unmarshal(line, &msg) if err != nil { - log.Error("Unable to unmarshal message: %s", err) - log.Debug("Full Error: %v", err) - continue + log.Error("Unable to decode message from master: %s", err) + continue; } - log.Debug("Recieved message from master: %v", line) - - if len(msg.NodeIDs) == 0 { - log.Debug("Message recieved had no node IDs") - log.Debug("Full message: %s", line) - continue + log.Debug("Received message from master: %s", msg) + var sshPayload SSHConnect + err = cbor.Unmarshal(msg.Payload, &sshPayload) + if err != nil { + log.Error("Unable to unmarshal SSH connect payload: %s", err) + continue; } - // Accumulate the results since we invoke via gofunc - results := make(map[string]*tailcfg.HostinfoView) - mu := sync.Mutex{} - wg := sync.WaitGroup{} - - for _, nodeID := range msg.NodeIDs { - wg.Add(1) - go func(nodeID string) { - defer wg.Done() - result, err := agent.GetStatusForPeer(nodeID) - if err != nil { - log.Error("Unable to get status for node %s: %s", nodeID, err) - return - } - - if result == nil { - log.Debug("No status for node %s", nodeID) - return - } - - mu.Lock() - results[nodeID] = result - mu.Unlock() - }(nodeID) - } - - wg.Wait() - - // Send the results back to the Headplane master - log.Debug("Sending status back to master: %v", results) - log.Msg(&SendMessage{ - Type: "status", - Data: results, + log.Info("Opening SSH PTY for session %s to %s@%s:%d", sshPayload.SessionId, sshPayload.Username, sshPayload.Hostname, sshPayload.Port) + sshutil.OpenSshPty(agent, sshutil.SshConnectParams{ + Hostname: sshPayload.Hostname, + Port: sshPayload.Port, + Username: sshPayload.Username, + Id: sshPayload.SessionId, }) } - if err := scanner.Err(); err != nil { - log.Fatal("Error reading from stdin: %s", err) - } + + + // var msg RecvMessage + // err := json.Unmarshal(line, &msg) + // if err != nil { + // var cborMsg CborMessage + // dec := cbor.NewDecoder(bytes.NewReader(line)) + // err := dec.Decode(&cborMsg) + + // if err == nil { + // log.Info("Unmarshalled CBOR message: %s", cborMsg) + // var sshPayload SSHConnect + // err = cbor.Unmarshal(cborMsg.Payload, &sshPayload) + // sshutil.OpenSshPty(agent, sshutil.SshConnectParams{ + // Hostname: sshPayload.Hostname, + // Port: sshPayload.Port, + // Username: sshPayload.Username, + // Id: sshPayload.SessionId, + // }) + + // return; + // } + + // log.Error("Unable to unmarshal message: %s", err) + // log.Debug("Full Error: %v", err) + // continue + // } + + // log.Debug("Recieved message from master: %v", line) + + // if len(msg.NodeIDs) == 0 { + // log.Debug("Message recieved had no node IDs") + // log.Debug("Full message: %s", line) + // continue + // } + + // // Accumulate the results since we invoke via gofunc + // results := make(map[string]*tailcfg.HostinfoView) + // mu := sync.Mutex{} + // wg := sync.WaitGroup{} + + // for _, nodeID := range msg.NodeIDs { + // wg.Add(1) + // go func(nodeID string) { + // defer wg.Done() + // result, err := agent.GetStatusForPeer(nodeID) + // if err != nil { + // log.Error("Unable to get status for node %s: %s", nodeID, err) + // return + // } + + // if result == nil { + // log.Debug("No status for node %s", nodeID) + // return + // } + + // mu.Lock() + // results[nodeID] = result + // mu.Unlock() + // }(nodeID) + // } + + // wg.Wait() + + // // Send the results back to the Headplane master + // log.Debug("Sending status back to master: %v", results) + // log.Msg(&SendMessage{ + // Type: "status", + // Data: results, + // }) + // } + + // if err := scanner.Err(); err != nil { + // log.Fatal("Error reading from stdin: %s", err) + // } } diff --git a/agent/internal/sshutil/connect.go b/agent/internal/sshutil/connect.go new file mode 100644 index 0000000..3e35aa6 --- /dev/null +++ b/agent/internal/sshutil/connect.go @@ -0,0 +1,161 @@ +package sshutil + +import ( + "context" + "errors" + "io" + "os" + "strconv" + "strings" + + "github.com/tale/headplane/agent/internal/tsnet" + "github.com/tale/headplane/agent/internal/util" + "golang.org/x/crypto/ssh" +) + +type SshConnectParams struct { + Hostname string + Port int + Username string + Id string +} + +func dialAndValidateTailscaleSSH(agent *tsnet.TSAgent, params SshConnectParams) (*ssh.Client, error) { + log := util.GetLogger() + + addr := strings.Join([]string{params.Hostname, ":", strconv.Itoa(params.Port)}, "") + + log.Debug("Attempting to dial %s via Tailscale SSH", addr) + conn, err := agent.Dial(context.Background(), "tcp", addr) + if err != nil { + log.Error("Failed to connect to Tailscale SSH: %s", err) + return nil, err + } + + log.Debug("Connected to Tailscale SSH at %s", addr) + config := &ssh.ClientConfig{ + User: params.Username, + HostKeyCallback: ssh.InsecureIgnoreHostKey(), + } + + clientConn, chans, reqs, err := ssh.NewClientConn(conn, addr, config) + if err != nil { + log.Error("Failed to create SSH client connection: %s", err) + conn.Close() + return nil, err + } + + client := ssh.NewClient(clientConn, chans, reqs) + sVer := string(client.ServerVersion()) + + if !strings.Contains(sVer, "Tailscale") { + log.Error("Connected to non-Tailscale SSH server: %s", sVer) + conn.Close() + return nil, errors.New("not a Tailscale SSH server") + } + + log.Info("Connected to SSH server running %s at %s", client.ServerVersion(), addr) + return client, nil +} + +func bindStdinToFd(sess *ssh.Session, fd int) error { + log := util.GetLogger() + + sshIn := os.NewFile(uintptr(fd), "sshInput") + if sshIn == nil { + log.Error("Failed to create file from stdin fd %d", fd) + return errors.New("failed to create file from stdin fd") + } + + stdin, err := sess.StdinPipe() + if err != nil { + log.Error("Failed to get stdin pipe: %s", err) + return err + } + + go io.Copy(stdin, sshIn) // From Node → SSH session + return nil +} + +func bindStdoutToFd(sess *ssh.Session, fd int) error { + log := util.GetLogger() + + sshOut := os.NewFile(uintptr(fd), "sshOutput") + if sshOut == nil { + log.Error("Failed to create file from stdout fd %d", fd) + return errors.New("failed to create file from stdout fd") + } + + stdout, err := sess.StdoutPipe() + if err != nil { + log.Error("Failed to get stdout pipe: %s", err) + return err + } + + go io.Copy(sshOut, stdout) // From SSH → Node + return nil +} + +func OpenSshPty(agent *tsnet.TSAgent, params SshConnectParams) (*ssh.Client, error) { + log := util.GetLogger() + + if agent == nil { + log.Error("Tailscale agent is nil") + return nil, errors.New("tailscale agent is nil") + } + + if params.Hostname == "" || params.Port <= 0 || params.Username == "" { + log.Error("Invalid SSH connection parameters: %+v", params) + return nil, errors.New("invalid SSH connection parameters") + } + + client, err := dialAndValidateTailscaleSSH(agent, params) + if err != nil { + log.Error("Failed to open SSH pty: %s", err) + return nil, err + } + + go func() { + sess, err := client.NewSession() + if err != nil { + log.Error("Failed to create new SSH session: %s", err) + client.Close() + } + + modes := ssh.TerminalModes{ + ssh.ECHO: 1, // enable echoing + ssh.TTY_OP_ISPEED: 14400, + ssh.TTY_OP_OSPEED: 14400, + } + + if err := sess.RequestPty("xterm-256color", 80, 40, modes); err != nil { + log.Error("Failed to request PTY: %s", err) + client.Close() + } + + ctx := addSession(params.Id, sess) + go func() { + for data := range ctx.InputCh { + if _, err := ctx.Stdin.Write(data); err != nil { + log.Error("Failed to write to SSH stdin: %s", err) + return + } + } + }(); + + if err := sess.Shell(); err != nil { + log.Error("Failed to start shell: %s", err) + client.Close() + } + + + log.Info("Successfully opened SSH pty for %s@%s:%d", params.Username, params.Hostname, params.Port) + go streamSSHOutput(params.Id, ctx.Stdout, os.NewFile(4, "sshOutput")) + sess.Wait(); + sess.Close(); + log.Info("SSH session %s closed (goSide)", params.Id) + RemoveSession(params.Id) + }() + + return client, nil +} diff --git a/agent/internal/sshutil/encoder.go b/agent/internal/sshutil/encoder.go new file mode 100644 index 0000000..66d307a --- /dev/null +++ b/agent/internal/sshutil/encoder.go @@ -0,0 +1,41 @@ +package sshutil + +import ( + "encoding/binary" + "fmt" +) + +func encodeFrame(id string, data []byte) ([]byte, error) { + sid := []byte(id) + if len(sid) > 255 { + return nil, fmt.Errorf("session ID too long") + } + + payloadLen := len(data) + buf := make([]byte, 1+len(sid)+4+payloadLen) + + buf[0] = byte(len(sid)) // SID length + copy(buf[1:], sid) // SID + binary.BigEndian.PutUint32(buf[1+len(sid):], uint32(payloadLen)) // Payload length + copy(buf[1+len(sid)+4:], data) // Payload + + return buf, nil +} + +func decodeFrame(buf []byte) (id string, payload []byte, ok bool) { + if len(buf) < 5 { + return "", nil, false + } + sidLen := int(buf[0]) + if len(buf) < 1+sidLen+4 { + return "", nil, false + } + id = string(buf[1 : 1+sidLen]) + payloadLen := int(binary.BigEndian.Uint32(buf[1+sidLen:])) + + if len(buf) < 1+sidLen+4+payloadLen { + return "", nil, false + } + payload = buf[1+sidLen+4 : 1+sidLen+4+payloadLen] + return id, payload, true +} diff --git a/agent/internal/sshutil/handler.go b/agent/internal/sshutil/handler.go new file mode 100644 index 0000000..db2722a --- /dev/null +++ b/agent/internal/sshutil/handler.go @@ -0,0 +1,56 @@ +package sshutil + +import ( + "io" + + "github.com/tale/headplane/agent/internal/util" +) + +func StartInputReader(fd3 io.Reader) { + log := util.GetLogger(); + + log.Info("Starting SSH fd3 input reader") + go func() { + buffer := make([]byte, 8192) + for { + n, err := fd3.Read(buffer) + if err != nil { + log.Error("fd3 read error: %v", err) + return + } + offset := 0 + for offset < n { + id, payload, ok := decodeFrame(buffer[offset:n]) + if !ok { + break // Wait for more data + } + offset += 1 + len(id) + 4 + len(payload) + + sessionsMu.RLock() + sess, ok := sessions[id] + sessionsMu.RUnlock() + if !ok { + log.Error("invalid session id: %s", id) + continue + } + _, err := sess.Stdin.Write(payload) + if err != nil { + log.Error("failed to write to session stdin: %v", err) + continue + } + } + } + }() +} + +func streamSSHOutput(id string, r io.Reader, fd4 io.Writer) { + buf := make([]byte, 1024) + for { + n, err := r.Read(buf) + if err != nil { + break + } + frame, _ := encodeFrame(id, buf[:n]) + fd4.Write(frame) + } +} diff --git a/agent/internal/sshutil/sessions.go b/agent/internal/sshutil/sessions.go new file mode 100644 index 0000000..7c07c39 --- /dev/null +++ b/agent/internal/sshutil/sessions.go @@ -0,0 +1,67 @@ +package sshutil + +import ( + "io" + "sync" + + "golang.org/x/crypto/ssh" +) + +type SessionContext struct { + ID string + Session *ssh.Session + Stdin io.WriteCloser + Stdout io.Reader + InputCh chan []byte +} + +var sessions = make(map[string]*SessionContext) +var sessionsMu sync.RWMutex + +func addSession(id string, session *ssh.Session) *SessionContext { + sessionsMu.Lock() + defer sessionsMu.Unlock() + + if _, exists := sessions[id]; exists { + return nil // Session with this ID already exists + } + + stdin, err := session.StdinPipe() + if err != nil { + return nil // Handle error appropriately in production code + } + + stdout, err := session.StdoutPipe() + if err != nil { + stdin.Close() // Close stdin if stdout pipe creation fails + return nil // Handle error appropriately in production code + } + + sessionContext := &SessionContext{ + ID: id, + Session: session, + Stdin: stdin, + Stdout: stdout, + } + + sessions[id] = sessionContext + return sessionContext +} + +func GetSession(id string) *SessionContext { + sessionsMu.RLock() + defer sessionsMu.RUnlock() + + return sessions[id] // Returns nil if session does not exist +} + +func RemoveSession(id string) { + sessionsMu.Lock() + defer sessionsMu.Unlock() + + if sessionContext, exists := sessions[id]; exists { + sessionContext.Stdin.Close() // Close the stdin pipe + sessionContext.Session.Close() // Close the SSH session + delete(sessions, id) // Remove from the map + } +} diff --git a/agent/internal/util/logger.go b/agent/internal/util/logger.go index 213b452..9754633 100644 --- a/agent/internal/util/logger.go +++ b/agent/internal/util/logger.go @@ -107,11 +107,34 @@ func (l *Logger) Msg(obj any) { } func escapeString(s string) string { - replacer := strings.NewReplacer( - `"`, `\"`, - `\`, `\\`, - "\n", `\n`, - "\t", `\t`, - ) - return replacer.Replace(s) + var b strings.Builder + b.Grow(len(s) + 16) // pre-grow to reduce reallocs + + for i := 0; i < len(s); i++ { + c := s[i] + switch c { + case '"': + b.WriteString(`\"`) + case '\\': + b.WriteString(`\\`) + case '\b': + b.WriteString(`\b`) + case '\f': + b.WriteString(`\f`) + case '\n': + b.WriteString(`\n`) + case '\r': + b.WriteString(`\r`) + case '\t': + b.WriteString(`\t`) + default: + if c < 0x20 { + // Control characters like 0x01, 0x07 (bell), etc. + fmt.Fprintf(&b, `\u%04x`, c) + } else { + b.WriteByte(c) + } + } + } + return b.String() } diff --git a/app/server/README.md b/app/server/README.md index 76d5b9b..ef23e17 100644 --- a/app/server/README.md +++ b/app/server/README.md @@ -7,6 +7,10 @@ many side-effects (in this case, importing a module may run code). ``` server ├── index.ts: Loads everything and starts the web server. +├── agent/ +│ ├── dispatcher.ts: Serializes commands for the agent control fd (stdin). +│ ├── ssh.ts: Manages & multiplexes the active web SSH connections +│ ├── env.ts: Checks the environment variables for custom overrides. ├── config/ │ ├── integration/ │ │ ├── abstract.ts: Defines the abstract class for integrations. diff --git a/app/server/agent/dispatcher.ts b/app/server/agent/dispatcher.ts new file mode 100644 index 0000000..c02d25c --- /dev/null +++ b/app/server/agent/dispatcher.ts @@ -0,0 +1,36 @@ +import type { Writable } from 'node:stream'; +import { decode, encode } from 'cbor2'; + +interface Command { + op: string; + payload: unknown; +} + +interface SSHConnectCommand extends Command { + op: 'ssh_conn'; + payload: { + sessionId: string; + username: string; + hostname: string; + port: number; + }; +} + +type AgentCommand = SSHConnectCommand; + +export async function dispatchCommand( + dispatcher: Writable, + command: AgentCommand, +) { + return new Promise((resolve, reject) => { + const encodedCommand = Buffer.concat([encode(command), Buffer.from('\n')]); + dispatcher.write(encodedCommand, (err) => { + console.log('Command dispatched:', command, err); + if (err) { + reject(err); + } else { + resolve(); + } + }); + }); +} diff --git a/app/server/agent/ssh.ts b/app/server/agent/ssh.ts new file mode 100644 index 0000000..0656aa9 --- /dev/null +++ b/app/server/agent/ssh.ts @@ -0,0 +1,185 @@ +import { ChildProcess } from 'node:child_process'; +import { randomUUID } from 'node:crypto'; +import type { Readable, Writable } from 'node:stream'; +import { Context } from 'hono'; +import { WSContext, WSEvents } from 'hono/ws'; +import log from '~/utils/log'; +import { dispatchCommand } from './dispatcher'; + +interface SSHConnection { + username: string; + hostname: string; + port: number; +} + +interface SSHSession { + connectionDetails: SSHConnection; + connected: boolean; + sessionId: string; + ws: WSContext; +} + +export function createSSHMultiplexer(proc: ChildProcess): SSHMultiplexer { + return new SSHMultiplexer(proc); +} + +export class SSHMultiplexer { + private connections: Map; + private child: ChildProcess; + + constructor(proc: ChildProcess) { + this.connections = new Map(); + this.child = proc; + + this.handleStdout(); + } + + // TODO: Determine if we want to allow multiple connections for the same + // target or attempt to reuse the existing connection (sounds stupid) + private async connect(conn: SSHConnection, ws: WSContext) { + const sessionId = randomUUID(); + const session: SSHSession = { + connectionDetails: conn, + connected: true, + sessionId, + ws, + }; + + log.info('agent', 'Dispatching SSH connection for %s', sessionId); + await dispatchCommand(this.child.stdin, { + op: 'ssh_conn', + payload: { + sessionId, + ...conn, + }, + }); + + this.connections.set(sessionId, session); + return sessionId; + } + + websocketHandler(c: Context): WSEvents { + return { + onOpen: async (_, ws) => { + const { username, hostname, port } = c.req.query(); + if (!username || !hostname || !port) { + ws.close(1008, 'Missing connection parameters'); + return; + } + + const conn: SSHConnection = { + username, + hostname, + port: Number.parseInt(port, 10), + }; + + try { + const sessionId = await this.connect(conn, ws); + ws.raw = sessionId; + ws.send(JSON.stringify({ status: 'connected', sessionId })); + } catch (error) { + ws.close(1011, `Connection failed: ${error.message}`); + } + }, + + onMessage: async (event, ws) => { + const sessionId = ws.raw; + if (!sessionId || !this.connections.has(sessionId)) { + ws.close(1008, 'Invalid session ID'); + return; + } + + const session = this.connections.get(sessionId); + if (!session || !session.connected) { + ws.close(1008, 'Session not connected'); + return; + } + + const encodedFrame = this.encodeFrame(sessionId, event.data); + this.child.stdio[3]?.write(encodedFrame); + }, + + onClose: (_, ws) => { + const sessionId = ws.raw; + if (sessionId && this.connections.has(sessionId)) { + const session = this.connections.get(sessionId); + if (session) { + session.connected = false; + this.connections.delete(sessionId); + } + } + }, + + onError: (event, ws) => { + const sessionId = ws.raw; + if (sessionId && this.connections.has(sessionId)) { + const session = this.connections.get(sessionId); + if (session) { + session.connected = false; + this.connections.delete(sessionId); + } + } + + log.error('agent', 'SSH WebSocket Error with %s', sessionId); + console.log(event); + }, + }; + } + + private encodeFrame(id: string, data: string | Buffer): Buffer { + const sid = Buffer.from(id, 'utf8'); + const payload = Buffer.isBuffer(data) ? data : Buffer.from(data, 'utf8'); + + // FIX: include +4 for the payload length + const frame = Buffer.alloc(1 + sid.length + 4 + payload.length); + frame.writeUint8(sid.length, 0); // 1 byte for sid length + sid.copy(frame, 1); // SID + frame.writeUint32BE(payload.length, 1 + sid.length); // 4 bytes for payload length + payload.copy(frame, 1 + sid.length + 4); // Payload + + return frame; + } + + private decodeFrame(frame: Buffer): { id: string; data: Buffer } | undefined { + if (frame.length < 5) return; + + const sidLength = frame.readUint8(0); + if (frame.length < 1 + sidLength + 4) return; + + const id = frame.slice(1, 1 + sidLength).toString('utf8'); + const payloadLength = frame.readUint32BE(1 + sidLength); + if (frame.length < 1 + sidLength + 4 + payloadLength) return; + + const data = frame.slice( + 1 + sidLength + 4, + 1 + sidLength + 4 + payloadLength, + ); + + return { id, data }; + } + + private handleStdout() { + const stdout = this.child.stdio[4]; + if (!stdout) { + return; + } + + stdout.on('data', (bytes) => { + console.log(Buffer.from(bytes).toString('utf8')); + const decoded = this.decodeFrame(bytes); + if (!decoded) { + return; + } + + const { id, data } = decoded; + console.log(id, data); + const session = this.connections.get(id); + if (!session || !session.connected) { + log.warn('agent', 'Received data for disconnected session %s', id); + return; + } + + session.ws.send(data); + }); + } +} diff --git a/app/server/index.ts b/app/server/index.ts index fc3b8b5..0bf31c4 100644 --- a/app/server/index.ts +++ b/app/server/index.ts @@ -28,6 +28,11 @@ const config = await loadConfig( }), ); +const agentManager = await loadAgentSocket( + config.integration?.agent, + config.headscale.url, +); + // We also use this file to load anything needed by the react router code. // These are usually per-request things that we need access to, like the // helper that can issue and revoke cookies. @@ -56,10 +61,7 @@ const appLoadContext = { config.headscale.tls_cert_path, ), - agents: await loadAgentSocket( - config.integration?.agent, - config.headscale.url, - ), + agents: agentManager, integration: await loadIntegration(config.integration), oidc: config.oidc ? await createOidcClient(config.oidc) : undefined, }; @@ -85,4 +87,18 @@ export default createHonoServer({ listeningListener(info) { log.info('server', 'Running on %s:%s', info.address, info.port); }, + + useWebSocket: true, + configure: (app, { upgradeWebSocket }) => { + if (agentManager === undefined) { + return; + } + + app.get( + '/_ssh_plexer', + upgradeWebSocket((c) => { + return agentManager.multiplexer!.websocketHandler(c); + }), + ); + }, }); diff --git a/app/server/web/agent.ts b/app/server/web/agent.ts index 4d446d4..a17e5c7 100644 --- a/app/server/web/agent.ts +++ b/app/server/web/agent.ts @@ -10,10 +10,12 @@ import { } from 'node:fs/promises'; import { exit } from 'node:process'; import { createInterface } from 'node:readline'; +import { Readable, Writable } from 'node:stream'; import { setTimeout } from 'node:timers/promises'; import { type } from 'arktype'; import { HostInfo } from '~/types'; import log from '~/utils/log'; +import { SSHMultiplexer, createSSHMultiplexer } from '../agent/ssh'; import type { HeadplaneConfig } from '../config/schema'; interface LogResponse { @@ -113,6 +115,7 @@ class AgentManager { >; private spawnProcess: ChildProcess | null; + multiplexer: SSHMultiplexer | null; private agentId: string | null; constructor( @@ -124,6 +127,7 @@ class AgentManager { this.config = config; this.headscaleUrl = headscaleUrl; this.spawnProcess = null; + this.multiplexer = null; this.agentId = null; this.startAgent(); @@ -184,7 +188,7 @@ class AgentManager { ); this.spawnProcess = spawn(this.config.executable_path, [], { detached: false, - stdio: ['pipe', 'pipe', 'pipe'], + stdio: ['pipe', 'pipe', 'pipe', 'pipe', 'pipe'], env: { HOME: process.env.HOME, HEADPLANE_EMBEDDED: 'true', @@ -210,6 +214,14 @@ class AgentManager { return; } + const sshInput = this.spawnProcess.stdio[3]; + const sshOutput = this.spawnProcess.stdio[4]; + + if (sshInput && sshOutput) { + log.info('agent', 'Using SSH multiplexer manager'); + this.multiplexer = createSSHMultiplexer(this.spawnProcess); + } + const rlStdout = createInterface({ input: this.spawnProcess.stdout, crlfDelay: Number.POSITIVE_INFINITY, diff --git a/go.mod b/go.mod index 6e32cbd..b6437af 100644 --- a/go.mod +++ b/go.mod @@ -33,7 +33,7 @@ require ( github.com/coreos/go-iptables v0.7.1-0.20240112124308-65c67c9f46e6 // indirect github.com/dblohm7/wingoes v0.0.0-20240119213807-a09d6be7affa // indirect github.com/digitalocean/go-smbios v0.0.0-20180907143718-390a4f403a8e // indirect - github.com/fxamacker/cbor/v2 v2.6.0 // indirect + github.com/fxamacker/cbor/v2 v2.8.0 // indirect github.com/gaissmai/bart v0.11.1 // indirect github.com/go-json-experiment/json v0.0.0-20231102232822-2e55bd4e08b0 // indirect github.com/go-ole/go-ole v1.3.0 // indirect @@ -76,14 +76,14 @@ require ( github.com/vishvananda/netns v0.0.4 // indirect github.com/x448/float16 v0.8.4 // indirect go4.org/netipx v0.0.0-20231129151722-fdeea329fbba // indirect - golang.org/x/crypto v0.25.0 // indirect + golang.org/x/crypto v0.38.0 // indirect golang.org/x/exp v0.0.0-20240119083558-1b970713d09a // indirect golang.org/x/mod v0.19.0 // indirect golang.org/x/net v0.27.0 // indirect - golang.org/x/sync v0.9.0 // indirect - golang.org/x/sys v0.27.0 // indirect - golang.org/x/term v0.22.0 // indirect - golang.org/x/text v0.16.0 // indirect + golang.org/x/sync v0.14.0 // indirect + golang.org/x/sys v0.33.0 // indirect + golang.org/x/term v0.32.0 // indirect + golang.org/x/text v0.25.0 // indirect golang.org/x/time v0.5.0 // indirect golang.org/x/tools v0.23.0 // indirect golang.zx2c4.com/wintun v0.0.0-20230126152724-0fa3db229ce2 // indirect diff --git a/go.sum b/go.sum index d08684a..56d6ef3 100644 --- a/go.sum +++ b/go.sum @@ -63,6 +63,8 @@ github.com/frankban/quicktest v1.14.6 h1:7Xjx+VpznH+oBnejlPUj8oUpdxnVs4f8XU8WnHk github.com/frankban/quicktest v1.14.6/go.mod h1:4ptaffx2x8+WTWXmUCuVU6aPUX1/Mz7zb5vbUoiM6w0= github.com/fxamacker/cbor/v2 v2.6.0 h1:sU6J2usfADwWlYDAFhZBQ6TnLFBHxgesMrQfQgk1tWA= github.com/fxamacker/cbor/v2 v2.6.0/go.mod h1:pxXPTn3joSm21Gbwsv0w9OSA2y1HFR9qXEeXQVeNoDQ= +github.com/fxamacker/cbor/v2 v2.8.0 h1:fFtUGXUzXPHTIUdne5+zzMPTfffl3RD5qYnkY40vtxU= +github.com/fxamacker/cbor/v2 v2.8.0/go.mod h1:vM4b+DJCtHn+zz7h3FFp/hDAI9WNWCsZj23V5ytsSxQ= github.com/gaissmai/bart v0.11.1 h1:5Uv5XwsaFBRo4E5VBcb9TzY8B7zxFf+U7isDxqOrRfc= github.com/gaissmai/bart v0.11.1/go.mod h1:KHeYECXQiBjTzQz/om2tqn3sZF1J7hw9m6z41ftj3fg= github.com/github/fakeca v0.1.0 h1:Km/MVOFvclqxPM9dZBC4+QE564nU4gz4iZ0D9pMw28I= @@ -194,6 +196,8 @@ go4.org/netipx v0.0.0-20231129151722-fdeea329fbba h1:0b9z3AuHCjxk0x/opv64kcgZLBs go4.org/netipx v0.0.0-20231129151722-fdeea329fbba/go.mod h1:PLyyIXexvUFg3Owu6p/WfdlivPbZJsZdgWZlrGope/Y= golang.org/x/crypto v0.25.0 h1:ypSNr+bnYL2YhwoMt2zPxHFmbAN1KZs/njMG3hxUp30= golang.org/x/crypto v0.25.0/go.mod h1:T+wALwcMOSE0kXgUAnPAHqTLW+XHgcELELW8VaDgm/M= +golang.org/x/crypto v0.38.0 h1:jt+WWG8IZlBnVbomuhg2Mdq0+BBQaHbtqHEFEigjUV8= +golang.org/x/crypto v0.38.0/go.mod h1:MvrbAqul58NNYPKnOra203SB9vpuZW0e+RRZV+Ggqjw= golang.org/x/exp v0.0.0-20240119083558-1b970713d09a h1:Q8/wZp0KX97QFTc2ywcOE0YRjZPVIx+MXInMzdvQqcA= golang.org/x/exp v0.0.0-20240119083558-1b970713d09a/go.mod h1:idGWGoKP1toJGkd5/ig9ZLuPcZBC3ewk7SzmH0uou08= golang.org/x/exp/typeparams v0.0.0-20240314144324-c7f7c6466f7f h1:phY1HzDcf18Aq9A8KkmRtY9WvOFIxN8wgfvy6Zm1DV8= @@ -207,6 +211,8 @@ golang.org/x/net v0.27.0/go.mod h1:dDi0PyhWNoiUOrAS8uXv/vnScO4wnHQO4mj9fn/RytE= golang.org/x/sync v0.0.0-20210220032951-036812b2e83c/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM= golang.org/x/sync v0.9.0 h1:fEo0HyrW1GIgZdpbhCRO0PkJajUS5H9IFUztCgEo2jQ= golang.org/x/sync v0.9.0/go.mod h1:Czt+wKu1gCyEFDUtn0jG5QVvpJ6rzVqr5aXyt9drQfk= +golang.org/x/sync v0.14.0 h1:woo0S4Yywslg6hp4eUFjTVOyKt0RookbpAHG4c1HmhQ= +golang.org/x/sync v0.14.0/go.mod h1:1dzgHSNfp02xaA81J2MS99Qcpr2w7fw1gpm99rleRqA= golang.org/x/sys v0.0.0-20200217220822-9197077df867/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs= golang.org/x/sys v0.0.0-20200728102440-3e129f6d46b1/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs= golang.org/x/sys v0.0.0-20220622161953-175b2fd9d664/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg= @@ -216,10 +222,16 @@ golang.org/x/sys v0.4.1-0.20230131160137-e7d7f63158de/go.mod h1:oPkhp1MJrh7nUepC golang.org/x/sys v0.6.0/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg= golang.org/x/sys v0.27.0 h1:wBqf8DvsY9Y/2P8gAfPDEYNuS30J4lPHJxXSb/nJZ+s= golang.org/x/sys v0.27.0/go.mod h1:/VUhepiaJMQUp4+oa/7Zr1D23ma6VTLIYjOOTFZPUcA= +golang.org/x/sys v0.33.0 h1:q3i8TbbEz+JRD9ywIRlyRAQbM0qF7hu24q3teo2hbuw= +golang.org/x/sys v0.33.0/go.mod h1:BJP2sWEmIv4KK5OTEluFJCKSidICx8ciO85XgH3Ak8k= golang.org/x/term v0.22.0 h1:BbsgPEJULsl2fV/AT3v15Mjva5yXKQDyKf+TbDz7QJk= golang.org/x/term v0.22.0/go.mod h1:F3qCibpT5AMpCRfhfT53vVJwhLtIVHhB9XDjfFvnMI4= +golang.org/x/term v0.32.0 h1:DR4lr0TjUs3epypdhTOkMmuF5CDFJ/8pOnbzMZPQ7bg= +golang.org/x/term v0.32.0/go.mod h1:uZG1FhGx848Sqfsq4/DlJr3xGGsYMu/L5GW4abiaEPQ= golang.org/x/text v0.16.0 h1:a94ExnEXNtEwYLGJSIUxnWoxoRz/ZcCsV63ROupILh4= golang.org/x/text v0.16.0/go.mod h1:GhwF1Be+LQoKShO3cGOHzqOgRrGaYc9AvblQOmPVHnI= +golang.org/x/text v0.25.0 h1:qVyWApTSYLk/drJRO5mDlNYskwQznZmkpV2c8q9zls4= +golang.org/x/text v0.25.0/go.mod h1:WEdwpYrmk1qmdHvhkSTNPm3app7v4rsT8F2UD6+VHIA= golang.org/x/time v0.5.0 h1:o7cqy6amK/52YcAKIPlM3a+Fpj35zvRj2TP+e1xFSfk= golang.org/x/time v0.5.0/go.mod h1:3BpzKBy/shNhVucY/MWOyx10tF3SFh9QdLuxbVysPQM= golang.org/x/tools v0.23.0 h1:SGsXPZ+2l4JsgaCKkx+FQ9YZ5XEtA1GZYuoDjenLjvg= diff --git a/package.json b/package.json index 2978657..7d9bd9c 100644 --- a/package.json +++ b/package.json @@ -29,6 +29,7 @@ "@uiw/codemirror-theme-xcode": "^4.23.12", "@uiw/react-codemirror": "^4.23.12", "arktype": "^2.1.20", + "cbor2": "^2.0.1", "clsx": "^2.1.1", "dotenv": "^16.5.0", "isbot": "^5.1.28", @@ -55,6 +56,7 @@ "@react-router/dev": "^7.6.1", "@tailwindcss/vite": "^4.1.8", "@types/websocket": "^1.0.10", + "hono": "^4.7.10", "lefthook": "^1.11.13", "postcss": "^8.5.4", "react-router-dom": "^7.6.1", diff --git a/pnpm-lock.yaml b/pnpm-lock.yaml index 3cd1df9..3546ee5 100644 --- a/pnpm-lock.yaml +++ b/pnpm-lock.yaml @@ -64,6 +64,9 @@ importers: arktype: specifier: ^2.1.20 version: 2.1.20 + cbor2: + specifier: ^2.0.1 + version: 2.0.1 clsx: specifier: ^2.1.1 version: 2.1.1 @@ -137,6 +140,9 @@ importers: '@types/websocket': specifier: ^1.0.10 version: 1.0.10 + hono: + specifier: ^4.7.10 + version: 4.7.10 lefthook: specifier: ^1.11.13 version: 1.11.13 @@ -406,6 +412,10 @@ packages: '@codemirror/view@6.36.8': resolution: {integrity: sha512-yoRo4f+FdnD01fFt4XpfpMCcCAo9QvZOtbrXExn4SqzH32YC6LgzqxfLZw/r6Ge65xyY03mK/UfUqrVw1gFiFg==} + '@cto.af/wtf8@0.0.2': + resolution: {integrity: sha512-ATm4UQiKrdm5GnU6BvIwUDN+LDEtt23zuzKFpnfDT59ULAd0aMYm/nSFzbSO02garLcXumRC13PzNfa7BsfvSg==} + engines: {node: '>=20'} + '@dnd-kit/accessibility@3.1.1': resolution: {integrity: sha512-2P+YgaXF+gRsIihwwY1gCsQSYnu9Zyj2py8kY5fFvUM1qm2WA2u639R6YNVfU4GWr+ZM5mqEsfHZZLoRONbemw==} peerDependencies: @@ -1692,6 +1702,10 @@ packages: caniuse-lite@1.0.30001718: resolution: {integrity: sha512-AflseV1ahcSunK53NfEs9gFWgOEmzr0f+kaMFA4xiLZlr9Hzt7HxcSpIFcnNCUkz6R6dWKa54rUz3HUmI3nVcw==} + cbor2@2.0.1: + resolution: {integrity: sha512-9bE8+tueGxONyxpttNKkAKKcGVtAPeoSJ64AjVTTjEuBOuRaeeP76EN9BbmQqkz1ZeTP0QPvksNBKwvEutIUzQ==} + engines: {node: '>=20'} + chokidar@4.0.3: resolution: {integrity: sha512-Qgzu8kfBvo+cA4962jnP1KkS6Dop5NS6g7R5LFYJr4b8Ub94PPQXUksCw9PvXoeXPRRddRNC5C1JQUR2SMGtnA==} engines: {node: '>= 14.16.0'} @@ -1924,8 +1938,8 @@ packages: resolution: {integrity: sha512-0hJU9SCPvmMzIBdZFqNPXWa6dqh7WdH0cII9y+CyS8rG3nL48Bclra9HmKhVVUHyPWNH5Y7xDwAB7bfgSjkUMQ==} engines: {node: '>= 0.4'} - hono@4.7.6: - resolution: {integrity: sha512-564rVzELU+9BRqqx5k8sT2NFwGD3I3Vifdb6P7CmM6FiarOSY+fDC+6B+k9wcCb86ReoayteZP2ki0cRLN1jbw==} + hono@4.7.10: + resolution: {integrity: sha512-QkACju9MiN59CKSY5JsGZCYmPZkA6sIW6OFCUp7qDjZu6S6KHtJHhAc9Uy9mV9F8PJ1/HQ3ybZF2yjCa/73fvQ==} engines: {node: '>=16.9.0'} hosted-git-info@6.1.3: @@ -3138,6 +3152,8 @@ snapshots: style-mod: 4.1.2 w3c-keyname: 2.2.8 + '@cto.af/wtf8@0.0.2': {} + '@dnd-kit/accessibility@3.1.1(react@19.1.0)': dependencies: react: 19.1.0 @@ -3275,23 +3291,23 @@ snapshots: dependencies: tslib: 2.8.1 - '@hono/node-server@1.14.0(hono@4.7.6)': + '@hono/node-server@1.14.0(hono@4.7.10)': dependencies: - hono: 4.7.6 + hono: 4.7.10 - '@hono/node-ws@1.1.1(@hono/node-server@1.14.0(hono@4.7.6))(bufferutil@4.0.9)(hono@4.7.6)(utf-8-validate@5.0.10)': + '@hono/node-ws@1.1.1(@hono/node-server@1.14.0(hono@4.7.10))(bufferutil@4.0.9)(hono@4.7.10)(utf-8-validate@5.0.10)': dependencies: - '@hono/node-server': 1.14.0(hono@4.7.6) - hono: 4.7.6 + '@hono/node-server': 1.14.0(hono@4.7.10) + hono: 4.7.10 ws: 8.18.1(bufferutil@4.0.9)(utf-8-validate@5.0.10) transitivePeerDependencies: - bufferutil - utf-8-validate - '@hono/vite-dev-server@0.19.0(hono@4.7.6)': + '@hono/vite-dev-server@0.19.0(hono@4.7.10)': dependencies: - '@hono/node-server': 1.14.0(hono@4.7.6) - hono: 4.7.6 + '@hono/node-server': 1.14.0(hono@4.7.10) + hono: 4.7.10 minimatch: 9.0.5 '@internationalized/date@3.8.1': @@ -4801,6 +4817,10 @@ snapshots: caniuse-lite@1.0.30001718: {} + cbor2@2.0.1: + dependencies: + '@cto.af/wtf8': 0.0.2 + chokidar@4.0.3: dependencies: readdirp: 4.1.2 @@ -5029,7 +5049,7 @@ snapshots: dependencies: function-bind: 1.1.2 - hono@4.7.6: {} + hono@4.7.10: {} hosted-git-info@6.1.3: dependencies: @@ -5423,12 +5443,12 @@ snapshots: react-router-hono-server@2.13.0(patch_hash=6549978df41006e07f1335bfe4ca86224ea36ed40d3f08dfef33143bad54005c)(@react-router/dev@7.6.1(@types/node@22.15.24)(jiti@2.4.2)(lightningcss@1.30.1)(react-router@7.6.1(react-dom@19.1.0(react@19.1.0))(react@19.1.0))(terser@5.39.0)(tsx@4.19.4)(typescript@5.8.3)(vite@6.3.5(@types/node@22.15.24)(jiti@2.4.2)(lightningcss@1.30.1)(terser@5.39.0)(tsx@4.19.4)(yaml@2.8.0))(yaml@2.8.0))(@types/react@19.1.6)(bufferutil@4.0.9)(react-router@7.6.1(react-dom@19.1.0(react@19.1.0))(react@19.1.0))(utf-8-validate@5.0.10)(vite@6.3.5(@types/node@22.15.24)(jiti@2.4.2)(lightningcss@1.30.1)(terser@5.39.0)(tsx@4.19.4)(yaml@2.8.0)): dependencies: '@drizzle-team/brocli': 0.11.0 - '@hono/node-server': 1.14.0(hono@4.7.6) - '@hono/node-ws': 1.1.1(@hono/node-server@1.14.0(hono@4.7.6))(bufferutil@4.0.9)(hono@4.7.6)(utf-8-validate@5.0.10) - '@hono/vite-dev-server': 0.19.0(hono@4.7.6) + '@hono/node-server': 1.14.0(hono@4.7.10) + '@hono/node-ws': 1.1.1(@hono/node-server@1.14.0(hono@4.7.10))(bufferutil@4.0.9)(hono@4.7.10)(utf-8-validate@5.0.10) + '@hono/vite-dev-server': 0.19.0(hono@4.7.10) '@react-router/dev': 7.6.1(@types/node@22.15.24)(jiti@2.4.2)(lightningcss@1.30.1)(react-router@7.6.1(react-dom@19.1.0(react@19.1.0))(react@19.1.0))(terser@5.39.0)(tsx@4.19.4)(typescript@5.8.3)(vite@6.3.5(@types/node@22.15.24)(jiti@2.4.2)(lightningcss@1.30.1)(terser@5.39.0)(tsx@4.19.4)(yaml@2.8.0))(yaml@2.8.0) '@types/react': 19.1.6 - hono: 4.7.6 + hono: 4.7.10 react-router: 7.6.1(react-dom@19.1.0(react@19.1.0))(react@19.1.0) vite: 6.3.5(@types/node@22.15.24)(jiti@2.4.2)(lightningcss@1.30.1)(terser@5.39.0)(tsx@4.19.4)(yaml@2.8.0) transitivePeerDependencies: