marketing-shibata50/cmux
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

Reapply "Merge pull request #239 from manaflow-ai/issue-151-ssh-remote-port-proxying"

Browse source 
This reverts commit f7cbbad434.
This commit is contained in:
Lawrence Chen 2026-03-12 15:54:26 -07:00
parent 294217eb39
commit 19b59cae37
60 changed files with 17139 additions and 1249 deletions
Download patch file Download diff file Expand all files Collapse all files

82
daemon/remote/README.md Normal file
View file

@ -0,0 +1,82 @@
# cmuxd-remote (Go)
Go remote daemon for `cmux ssh` bootstrap, capability negotiation, and remote proxy RPC. It is not in the terminal keystroke hot path.
## Commands
1. `cmuxd-remote version`
2. `cmuxd-remote serve --stdio`
3. `cmuxd-remote cli <command> [args...]` — relay cmux commands to the local app over the reverse SSH forward
When invoked as `cmux` (via wrapper/symlink installed during bootstrap), the binary auto-dispatches to the `cli` subcommand. This is busybox-style argv[0] detection.
## RPC methods (newline-delimited JSON over stdio)
1. `hello`
2. `ping`
3. `proxy.open`
4. `proxy.close`
5. `proxy.write`
6. `proxy.read`
7. `session.open`
8. `session.close`
9. `session.attach`
10. `session.resize`
11. `session.detach`
12. `session.status`
Current integration in cmux:
1. `workspace.remote.configure` now bootstraps this binary over SSH when missing.
2. Client sends `hello` before enabling remote proxy transport.
3. Local workspace proxy broker serves SOCKS5 + HTTP CONNECT and tunnels stream traffic through `proxy.*` RPC over `serve --stdio`.
4. Daemon status/capabilities are exposed in `workspace.remote.status -> remote.daemon` (including `session.resize.min`).
`workspace.remote.configure` contract notes:
1. `port` / `local_proxy_port` accept integer values and numeric strings; explicit `null` clears each field.
2. Out-of-range values and invalid types return `invalid_params`.
3. `local_proxy_port` is an internal deterministic test hook used by bind-conflict regressions.
4. SSH option precedence checks are case-insensitive; user overrides for `StrictHostKeyChecking` and control-socket keys prevent default injection.
## Distribution
Release and nightly builds publish prebuilt `cmuxd-remote` binaries on GitHub Releases for:
1. `darwin/arm64`
2. `darwin/amd64`
3. `linux/arm64`
4. `linux/amd64`
The app embeds a compact manifest in `Info.plist` with:
1. exact release asset URLs
2. pinned SHA-256 digests
3. release tag and checksums asset URL
Release and nightly apps download and cache the matching binary locally, verify its SHA-256, then upload it to the remote host if needed. Dev builds can opt into a local `go build` fallback with `CMUX_REMOTE_DAEMON_ALLOW_LOCAL_BUILD=1`.
To inspect what a given app build trusts, run:
1. `cmux remote-daemon-status`
2. `cmux remote-daemon-status --os linux --arch amd64`
The command prints the exact release asset URL, expected SHA-256, local cache status, and a copy-pasteable `gh attestation verify` command for the selected platform.
## CLI relay
The `cli` subcommand (or `cmux` wrapper/symlink) connects to the local cmux app through an SSH reverse forward and relays commands. It supports both v1 text protocol and v2 JSON-RPC commands.
Socket discovery order:
1. `--socket <path>` flag
2. `CMUX_SOCKET_PATH` environment variable
3. `~/.cmux/socket_addr` file (written by the app after the reverse relay establishes)
For TCP addresses, the CLI retries for up to 15 seconds on connection refused, re-reading `~/.cmux/socket_addr` on each attempt to pick up updated relay ports.
Authenticated relay details:
1. Each SSH workspace gets its own relay ID and relay token.
2. The app runs a local loopback relay server that requires an HMAC-SHA256 challenge-response before forwarding a command to the real local Unix socket.
3. The remote shell never gets direct access to the local app socket. It only gets the reverse-forwarded relay port plus `~/.cmux/relay/<port>.auth`, which is written with `0600` permissions and removed when the relay stops.
Integration additions for the relay path:
1. Bootstrap installs `~/.cmux/bin/cmux` wrapper and keeps a default daemon target (`~/.cmux/bin/cmuxd-remote-current`).
2. A background `ssh -N -R` process reverse-forwards a TCP port to the authenticated local relay server. The relay address is written to `~/.cmux/socket_addr` on the remote.
3. Relay startup writes `~/.cmux/relay/<port>.daemon_path` so the wrapper can route each shell to the correct daemon binary when multiple local cmux instances or versions coexist.
4. Relay startup writes `~/.cmux/relay/<port>.auth` with the relay ID and token needed for HMAC authentication.

721
daemon/remote/cmd/cmuxd-remote/cli.go Normal file
View file

@ -0,0 +1,721 @@
package main
import (
"bufio"
"crypto/hmac"
"crypto/rand"
"crypto/sha256"
"encoding/hex"
"encoding/json"
"errors"
"fmt"
"io"
"net"
"os"
"path/filepath"
"strings"
"time"
)
type relayAuthState struct {
RelayID string `json:"relay_id"`
RelayToken string `json:"relay_token"`
}
// protocolVersion indicates whether a command uses the v1 text or v2 JSON-RPC protocol.
type protocolVersion int
const (
protoV1 protocolVersion = iota
protoV2
)
// commandSpec describes a single CLI command and how to relay it.
type commandSpec struct {
name string // CLI command name (e.g. "ping", "new-window")
proto protocolVersion // v1 text or v2 JSON-RPC
v1Cmd string // v1: literal command string sent over the socket
v2Method string // v2: JSON-RPC method name
// flagKeys lists parameter keys this command accepts.
// They are extracted from --key flags and added to params.
flagKeys []string
// noParams means the command takes no parameters at all.
noParams bool
}
var commands = []commandSpec{
// V1 text protocol commands
{name: "ping", proto: protoV1, v1Cmd: "ping", noParams: true},
{name: "new-window", proto: protoV1, v1Cmd: "new_window", noParams: true},
{name: "current-window", proto: protoV1, v1Cmd: "current_window", noParams: true},
{name: "close-window", proto: protoV1, v1Cmd: "close_window", flagKeys: []string{"window"}},
{name: "focus-window", proto: protoV1, v1Cmd: "focus_window", flagKeys: []string{"window"}},
{name: "list-windows", proto: protoV1, v1Cmd: "list_windows", noParams: true},
// V2 JSON-RPC commands
{name: "capabilities", proto: protoV2, v2Method: "system.capabilities", noParams: true},
{name: "list-workspaces", proto: protoV2, v2Method: "workspace.list", noParams: true},
{name: "new-workspace", proto: protoV2, v2Method: "workspace.create", flagKeys: []string{"command", "working-directory", "name"}},
{name: "close-workspace", proto: protoV2, v2Method: "workspace.close", flagKeys: []string{"workspace"}},
{name: "select-workspace", proto: protoV2, v2Method: "workspace.select", flagKeys: []string{"workspace"}},
{name: "current-workspace", proto: protoV2, v2Method: "workspace.current", noParams: true},
{name: "list-panels", proto: protoV2, v2Method: "panel.list", flagKeys: []string{"workspace"}},
{name: "focus-panel", proto: protoV2, v2Method: "panel.focus", flagKeys: []string{"panel", "workspace"}},
{name: "list-panes", proto: protoV2, v2Method: "pane.list", flagKeys: []string{"workspace"}},
{name: "list-pane-surfaces", proto: protoV2, v2Method: "pane.surfaces", flagKeys: []string{"pane"}},
{name: "new-pane", proto: protoV2, v2Method: "pane.create", flagKeys: []string{"workspace"}},
{name: "new-surface", proto: protoV2, v2Method: "surface.create", flagKeys: []string{"workspace", "pane"}},
{name: "new-split", proto: protoV2, v2Method: "surface.split", flagKeys: []string{"surface", "direction"}},
{name: "close-surface", proto: protoV2, v2Method: "surface.close", flagKeys: []string{"surface"}},
{name: "send", proto: protoV2, v2Method: "surface.send_text", flagKeys: []string{"surface", "text"}},
{name: "send-key", proto: protoV2, v2Method: "surface.send_key", flagKeys: []string{"surface", "key"}},
{name: "notify", proto: protoV2, v2Method: "notification.create", flagKeys: []string{"title", "body", "workspace"}},
{name: "refresh-surfaces", proto: protoV2, v2Method: "surface.refresh", noParams: true},
}
var commandIndex map[string]*commandSpec
func init() {
commandIndex = make(map[string]*commandSpec, len(commands))
for i := range commands {
commandIndex[commands[i].name] = &commands[i]
}
}
// runCLI is the entry point for the "cli" subcommand (or busybox "cmux" invocation).
func runCLI(args []string) int {
socketPath := os.Getenv("CMUX_SOCKET_PATH")
// Parse global flags
var jsonOutput bool
var remaining []string
for i := 0; i < len(args); i++ {
switch args[i] {
case "--socket":
if i+1 >= len(args) {
fmt.Fprintln(os.Stderr, "cmux: --socket requires a path")
return 2
}
socketPath = args[i+1]
i++
case "--json":
jsonOutput = true
case "--help", "-h":
cliUsage()
return 0
default:
remaining = append(remaining, args[i:]...)
goto doneFlags
}
}
doneFlags:
if len(remaining) == 0 {
cliUsage()
return 2
}
cmdName := remaining[0]
cmdArgs := remaining[1:]
if cmdName == "help" {
cliUsage()
return 0
}
// refreshAddr is set when the address came from socket_addr file (not env/flag),
// allowing retry loops to pick up updated relay ports.
var refreshAddr func() string
if socketPath == "" {
socketPath = readSocketAddrFile()
refreshAddr = readSocketAddrFile
}
if socketPath == "" {
fmt.Fprintln(os.Stderr, "cmux: CMUX_SOCKET_PATH not set and --socket not provided")
return 1
}
// Special case: "rpc" passthrough
if cmdName == "rpc" {
return runRPC(socketPath, cmdArgs, jsonOutput, refreshAddr)
}
// Browser subcommand delegation
if cmdName == "browser" {
return runBrowserRelay(socketPath, cmdArgs, jsonOutput, refreshAddr)
}
spec, ok := commandIndex[cmdName]
if !ok {
fmt.Fprintf(os.Stderr, "cmux: unknown command %q\n", cmdName)
return 2
}
switch spec.proto {
case protoV1:
return execV1(socketPath, spec, cmdArgs, refreshAddr)
case protoV2:
return execV2(socketPath, spec, cmdArgs, jsonOutput, refreshAddr)
default:
fmt.Fprintf(os.Stderr, "cmux: internal error: unknown protocol for %q\n", cmdName)
return 1
}
}
// execV1 sends a v1 text command over the socket.
func execV1(socketPath string, spec *commandSpec, args []string, refreshAddr func() string) int {
cmd := spec.v1Cmd
if !spec.noParams {
parsed := parseFlags(args, spec.flagKeys)
for _, key := range spec.flagKeys {
if val, ok := parsed.flags[key]; ok {
cmd += " " + val
}
}
}
resp, err := socketRoundTrip(socketPath, cmd, refreshAddr)
if err != nil {
fmt.Fprintf(os.Stderr, "cmux: %v\n", err)
return 1
}
fmt.Print(resp)
if !strings.HasSuffix(resp, "\n") {
fmt.Println()
}
return 0
}
// execV2 sends a v2 JSON-RPC request over the socket.
func execV2(socketPath string, spec *commandSpec, args []string, jsonOutput bool, refreshAddr func() string) int {
params := make(map[string]any)
if !spec.noParams {
parsed := parseFlags(args, spec.flagKeys)
// Map flag keys to JSON param keys (e.g. "workspace" → "workspace_id" where appropriate)
for _, key := range spec.flagKeys {
if val, ok := parsed.flags[key]; ok {
paramKey := flagToParamKey(key)
params[paramKey] = val
}
}
// First positional arg is used as initial_command if --command wasn't given
if _, ok := params["initial_command"]; !ok && len(parsed.positional) > 0 {
params["initial_command"] = parsed.positional[0]
}
// Fall back to env vars for common IDs
if _, ok := params["workspace_id"]; !ok {
if envWs := os.Getenv("CMUX_WORKSPACE_ID"); envWs != "" {
params["workspace_id"] = envWs
}
}
if _, ok := params["surface_id"]; !ok {
if envSf := os.Getenv("CMUX_SURFACE_ID"); envSf != "" {
params["surface_id"] = envSf
}
}
}
resp, err := socketRoundTripV2(socketPath, spec.v2Method, params, refreshAddr)
if err != nil {
fmt.Fprintf(os.Stderr, "cmux: %v\n", err)
return 1
}
if jsonOutput {
fmt.Println(resp)
} else {
fmt.Println(defaultRelayOutput(resp))
}
return 0
}
// runRPC sends an arbitrary JSON-RPC method with optional JSON params.
func runRPC(socketPath string, args []string, jsonOutput bool, refreshAddr func() string) int {
if len(args) == 0 {
fmt.Fprintln(os.Stderr, "cmux rpc: requires a method name")
return 2
}
method := args[0]
var params map[string]any
if len(args) > 1 {
if err := json.Unmarshal([]byte(args[1]), &params); err != nil {
fmt.Fprintf(os.Stderr, "cmux rpc: invalid JSON params: %v\n", err)
return 2
}
}
resp, err := socketRoundTripV2(socketPath, method, params, refreshAddr)
if err != nil {
fmt.Fprintf(os.Stderr, "cmux: %v\n", err)
return 1
}
fmt.Println(resp)
return 0
}
// runBrowserRelay handles "cmux browser <subcommand>" by mapping to browser.* v2 methods.
func runBrowserRelay(socketPath string, args []string, jsonOutput bool, refreshAddr func() string) int {
if len(args) == 0 {
fmt.Fprintln(os.Stderr, "cmux browser: requires a subcommand (open, navigate, back, forward, reload, get-url)")
return 2
}
sub := args[0]
subArgs := args[1:]
var method string
var flagKeys []string
switch sub {
case "open", "open-split", "new":
method = "browser.open"
flagKeys = []string{"url", "workspace", "surface"}
case "navigate":
method = "browser.navigate"
flagKeys = []string{"url", "surface"}
case "back":
method = "browser.back"
flagKeys = []string{"surface"}
case "forward":
method = "browser.forward"
flagKeys = []string{"surface"}
case "reload":
method = "browser.reload"
flagKeys = []string{"surface"}
case "get-url":
method = "browser.get_url"
flagKeys = []string{"surface"}
default:
fmt.Fprintf(os.Stderr, "cmux browser: unknown subcommand %q\n", sub)
return 2
}
params := make(map[string]any)
parsed := parseFlags(subArgs, flagKeys)
for _, key := range flagKeys {
if val, ok := parsed.flags[key]; ok {
paramKey := flagToParamKey(key)
params[paramKey] = val
}
}
resp, err := socketRoundTripV2(socketPath, method, params, refreshAddr)
if err != nil {
fmt.Fprintf(os.Stderr, "cmux: %v\n", err)
return 1
}
if jsonOutput {
fmt.Println(resp)
} else {
fmt.Println(defaultRelayOutput(resp))
}
return 0
}
func defaultRelayOutput(resp string) string {
var result any
if err := json.Unmarshal([]byte(resp), &result); err != nil {
trimmed := strings.TrimSpace(resp)
if trimmed == "" {
return "OK"
}
return trimmed
}
if relayResultIsEmpty(result) {
return "OK"
}
switch typed := result.(type) {
case string:
return typed
default:
encoded, err := json.MarshalIndent(typed, "", " ")
if err != nil {
return "OK"
}
return string(encoded)
}
}
func relayResultIsEmpty(result any) bool {
switch typed := result.(type) {
case nil:
return true
case map[string]any:
return len(typed) == 0
case []any:
return len(typed) == 0
case string:
return typed == ""
default:
return false
}
}
// flagToParamKey maps a CLI flag name to its JSON-RPC param key.
func flagToParamKey(key string) string {
switch key {
case "workspace":
return "workspace_id"
case "surface":
return "surface_id"
case "panel":
return "panel_id"
case "pane":
return "pane_id"
case "window":
return "window_id"
case "command":
return "initial_command"
case "name":
return "title"
case "working-directory":
return "working_directory"
default:
return key
}
}
// parsedFlags holds the results of flag parsing.
type parsedFlags struct {
flags map[string]string // --key value pairs
positional []string // non-flag arguments
}
// parseFlags extracts --key value pairs from args for the given allowed keys.
// Non-flag arguments are collected in positional.
func parseFlags(args []string, keys []string) parsedFlags {
allowed := make(map[string]bool, len(keys))
for _, k := range keys {
allowed[k] = true
}
result := parsedFlags{flags: make(map[string]string)}
for i := 0; i < len(args); i++ {
if !strings.HasPrefix(args[i], "--") {
result.positional = append(result.positional, args[i])
continue
}
key := strings.TrimPrefix(args[i], "--")
if !allowed[key] {
continue
}
if i+1 < len(args) {
result.flags[key] = args[i+1]
i++
}
}
return result
}
// readSocketAddrFile reads the socket address from ~/.cmux/socket_addr as a fallback
// when CMUX_SOCKET_PATH is not set. Written by the cmux app after the relay establishes.
func readSocketAddrFile() string {
home, err := os.UserHomeDir()
if err != nil {
return ""
}
data, err := os.ReadFile(filepath.Join(home, ".cmux", "socket_addr"))
if err != nil {
return ""
}
return strings.TrimSpace(string(data))
}
func readRelayAuthFile(socketPath string) *relayAuthState {
if strings.Contains(socketPath, ":") && !strings.HasPrefix(socketPath, "/") {
_, port, err := net.SplitHostPort(socketPath)
if err != nil || port == "" {
return nil
}
home, err := os.UserHomeDir()
if err != nil {
return nil
}
data, err := os.ReadFile(filepath.Join(home, ".cmux", "relay", port+".auth"))
if err != nil {
return nil
}
var state relayAuthState
if err := json.Unmarshal(data, &state); err != nil {
return nil
}
if state.RelayID == "" || state.RelayToken == "" {
return nil
}
return &state
}
return nil
}
func currentRelayAuth(socketPath string) *relayAuthState {
relayID := strings.TrimSpace(os.Getenv("CMUX_RELAY_ID"))
relayToken := strings.TrimSpace(os.Getenv("CMUX_RELAY_TOKEN"))
if relayID != "" && relayToken != "" {
return &relayAuthState{RelayID: relayID, RelayToken: relayToken}
}
return readRelayAuthFile(socketPath)
}
// dialSocket connects to the cmux socket. If addr contains a colon and doesn't
// start with '/', it's treated as a TCP address (host:port); otherwise Unix socket.
// For TCP connections, it retries briefly to allow the SSH reverse forward to establish.
// refreshAddr, if non-nil, is called on each retry to pick up updated socket_addr files.
func dialSocket(addr string, refreshAddr func() string) (net.Conn, error) {
if strings.Contains(addr, ":") && !strings.HasPrefix(addr, "/") {
conn, err := dialTCPRetry(addr, 15*time.Second, refreshAddr)
if err != nil {
return nil, err
}
if auth := currentRelayAuth(addr); auth != nil {
if err := authenticateRelayConn(conn, auth); err != nil {
conn.Close()
return nil, err
}
}
return conn, nil
}
return net.Dial("unix", addr)
}
// dialTCPRetry attempts a TCP connection, retrying on "connection refused" for up to timeout.
// This handles the case where the SSH reverse relay hasn't finished establishing yet.
// If refreshAddr is non-nil, it's called on each retry to pick up updated addresses
// (e.g. when socket_addr is rewritten by a new relay process).
func dialTCPRetry(addr string, timeout time.Duration, refreshAddr func() string) (net.Conn, error) {
deadline := time.Now().Add(timeout)
interval := 250 * time.Millisecond
printed := false
for {
conn, err := net.DialTimeout("tcp", addr, 2*time.Second)
if err == nil {
return conn, nil
}
if time.Now().After(deadline) {
return nil, err
}
// Only retry on connection refused (relay not ready yet)
if !isConnectionRefused(err) {
return nil, err
}
if !printed {
fmt.Fprintf(os.Stderr, "cmux: waiting for relay on %s...\n", addr)
printed = true
}
time.Sleep(interval)
// Re-read socket_addr in case the relay port has changed
if refreshAddr != nil {
if newAddr := refreshAddr(); newAddr != "" && newAddr != addr {
addr = newAddr
fmt.Fprintf(os.Stderr, "cmux: relay address updated to %s\n", addr)
}
}
}
}
func isConnectionRefused(err error) bool {
if opErr, ok := err.(*net.OpError); ok {
return strings.Contains(opErr.Err.Error(), "connection refused")
}
return strings.Contains(err.Error(), "connection refused")
}
func authenticateRelayConn(conn net.Conn, auth *relayAuthState) error {
reader := bufio.NewReader(conn)
_ = conn.SetDeadline(time.Now().Add(5 * time.Second))
var challenge struct {
Protocol string `json:"protocol"`
Version int `json:"version"`
RelayID string `json:"relay_id"`
Nonce string `json:"nonce"`
}
line, err := reader.ReadString('\n')
if err != nil {
return fmt.Errorf("failed to read relay auth challenge: %w", err)
}
if err := json.Unmarshal([]byte(line), &challenge); err != nil {
return fmt.Errorf("invalid relay auth challenge")
}
if challenge.Protocol != "cmux-relay-auth" || challenge.Version != 1 || challenge.RelayID != auth.RelayID || challenge.Nonce == "" {
return fmt.Errorf("relay auth challenge mismatch")
}
tokenBytes, err := hex.DecodeString(auth.RelayToken)
if err != nil {
return fmt.Errorf("invalid relay auth token")
}
mac := computeRelayMAC(tokenBytes, auth.RelayID, challenge.Nonce, challenge.Version)
payload, err := json.Marshal(map[string]any{
"relay_id": auth.RelayID,
"mac": hex.EncodeToString(mac),
})
if err != nil {
return fmt.Errorf("failed to encode relay auth response: %w", err)
}
if _, err := conn.Write(append(payload, '\n')); err != nil {
return fmt.Errorf("failed to send relay auth response: %w", err)
}
line, err = reader.ReadString('\n')
if err != nil {
return fmt.Errorf("failed to read relay auth result: %w", err)
}
var result struct {
OK bool `json:"ok"`
}
if err := json.Unmarshal([]byte(line), &result); err != nil {
return fmt.Errorf("invalid relay auth result")
}
if !result.OK {
return fmt.Errorf("relay auth rejected")
}
_ = conn.SetDeadline(time.Time{})
return nil
}
func computeRelayMAC(token []byte, relayID, nonce string, version int) []byte {
mac := hmac.New(sha256.New, token)
_, _ = io.WriteString(mac, fmt.Sprintf("relay_id=%s\nnonce=%s\nversion=%d", relayID, nonce, version))
return mac.Sum(nil)
}
// socketRoundTrip sends a raw text line and reads a raw text response (v1).
func socketRoundTrip(socketPath, command string, refreshAddr func() string) (string, error) {
conn, err := dialSocket(socketPath, refreshAddr)
if err != nil {
return "", fmt.Errorf("failed to connect to %s: %w", socketPath, err)
}
defer conn.Close()
if _, err := fmt.Fprintf(conn, "%s\n", command); err != nil {
return "", fmt.Errorf("failed to send command: %w", err)
}
// V1 handlers may return multiple lines (e.g. list_windows). Read until
// the stream goes idle briefly after seeing at least one newline.
reader := bufio.NewReader(conn)
var response strings.Builder
sawNewline := false
for {
readTimeout := 15 * time.Second
if sawNewline {
readTimeout = 120 * time.Millisecond
}
_ = conn.SetReadDeadline(time.Now().Add(readTimeout))
chunk, err := reader.ReadString('\n')
if chunk != "" {
response.WriteString(chunk)
if strings.Contains(chunk, "\n") {
sawNewline = true
}
}
if err != nil {
if netErr, ok := err.(net.Error); ok && netErr.Timeout() {
if sawNewline {
break
}
return "", fmt.Errorf("failed to read response: timeout waiting for response")
}
if errors.Is(err, io.EOF) {
break
}
return "", fmt.Errorf("failed to read response: %w", err)
}
}
return strings.TrimRight(response.String(), "\n"), nil
}
// socketRoundTripV2 sends a JSON-RPC request and returns the result JSON.
func socketRoundTripV2(socketPath, method string, params map[string]any, refreshAddr func() string) (string, error) {
conn, err := dialSocket(socketPath, refreshAddr)
if err != nil {
return "", fmt.Errorf("failed to connect to %s: %w", socketPath, err)
}
defer conn.Close()
id := randomHex(8)
req := map[string]any{
"id": id,
"method": method,
}
if params != nil {
req["params"] = params
} else {
req["params"] = map[string]any{}
}
payload, err := json.Marshal(req)
if err != nil {
return "", fmt.Errorf("failed to marshal request: %w", err)
}
if _, err := conn.Write(append(payload, '\n')); err != nil {
return "", fmt.Errorf("failed to send request: %w", err)
}
reader := bufio.NewReader(conn)
line, err := reader.ReadString('\n')
if err != nil {
return "", fmt.Errorf("failed to read response: %w", err)
}
// Parse the response to check for errors
var resp map[string]any
if err := json.Unmarshal([]byte(line), &resp); err != nil {
return strings.TrimRight(line, "\n"), nil
}
if ok, _ := resp["ok"].(bool); !ok {
if errObj, _ := resp["error"].(map[string]any); errObj != nil {
code, _ := errObj["code"].(string)
msg, _ := errObj["message"].(string)
return "", fmt.Errorf("server error [%s]: %s", code, msg)
}
return "", fmt.Errorf("server returned error response")
}
// Return the result portion as JSON
if result, ok := resp["result"]; ok {
resultJSON, err := json.Marshal(result)
if err != nil {
return "", fmt.Errorf("failed to marshal result: %w", err)
}
return string(resultJSON), nil
}
return "{}", nil
}
func randomHex(n int) string {
b := make([]byte, n)
_, _ = rand.Read(b)
return hex.EncodeToString(b)
}
func cliUsage() {
fmt.Fprintln(os.Stderr, "Usage: cmux [--socket <path>] [--json] <command> [args...]")
fmt.Fprintln(os.Stderr, "")
fmt.Fprintln(os.Stderr, "Commands:")
fmt.Fprintln(os.Stderr, " ping Check connectivity")
fmt.Fprintln(os.Stderr, " capabilities List server capabilities")
fmt.Fprintln(os.Stderr, " list-workspaces List all workspaces")
fmt.Fprintln(os.Stderr, " new-window Create a new window")
fmt.Fprintln(os.Stderr, " new-workspace Create a new workspace")
fmt.Fprintln(os.Stderr, " new-surface Create a new surface")
fmt.Fprintln(os.Stderr, " new-split Split an existing surface")
fmt.Fprintln(os.Stderr, " close-surface Close a surface")
fmt.Fprintln(os.Stderr, " close-workspace Close a workspace")
fmt.Fprintln(os.Stderr, " select-workspace Select a workspace")
fmt.Fprintln(os.Stderr, " send Send text to a surface")
fmt.Fprintln(os.Stderr, " send-key Send a key to a surface")
fmt.Fprintln(os.Stderr, " notify Create a notification")
fmt.Fprintln(os.Stderr, " browser <sub> Browser commands (open, navigate, back, forward, reload, get-url)")
fmt.Fprintln(os.Stderr, " rpc <method> [json-params] Send arbitrary JSON-RPC")
}

696
daemon/remote/cmd/cmuxd-remote/cli_test.go Normal file
View file

@ -0,0 +1,696 @@
package main
import (
"bufio"
"crypto/hmac"
"crypto/sha256"
"encoding/hex"
"encoding/json"
"fmt"
"io"
"net"
"os"
"path/filepath"
"strings"
"testing"
"time"
)
func captureStdout(t *testing.T, fn func()) string {
t.Helper()
original := os.Stdout
reader, writer, err := os.Pipe()
if err != nil {
t.Fatalf("pipe stdout: %v", err)
}
os.Stdout = writer
defer func() {
os.Stdout = original
}()
fn()
if err := writer.Close(); err != nil {
t.Fatalf("close stdout writer: %v", err)
}
output, err := io.ReadAll(reader)
if err != nil {
t.Fatalf("read stdout: %v", err)
}
if err := reader.Close(); err != nil {
t.Fatalf("close stdout reader: %v", err)
}
return string(output)
}
// startMockSocket creates a Unix socket that accepts one connection,
// reads a line, and responds with the given canned response.
func startMockSocket(t *testing.T, response string) string {
t.Helper()
dir := t.TempDir()
sockPath := filepath.Join(dir, "cmux.sock")
ln, err := net.Listen("unix", sockPath)
if err != nil {
t.Fatalf("failed to listen: %v", err)
}
t.Cleanup(func() { ln.Close() })
go func() {
for {
conn, err := ln.Accept()
if err != nil {
return
}
buf := make([]byte, 4096)
n, _ := conn.Read(buf)
_ = n // consume request
conn.Write([]byte(response + "\n"))
conn.Close()
}
}()
return sockPath
}
// startMockV2Socket creates a Unix socket that echoes the received request's method
// back as a successful JSON-RPC response with the method name in the result.
func startMockV2Socket(t *testing.T) string {
t.Helper()
dir := t.TempDir()
sockPath := filepath.Join(dir, "cmux.sock")
ln, err := net.Listen("unix", sockPath)
if err != nil {
t.Fatalf("failed to listen: %v", err)
}
t.Cleanup(func() { ln.Close() })
go func() {
for {
conn, err := ln.Accept()
if err != nil {
return
}
buf := make([]byte, 4096)
n, _ := conn.Read(buf)
if n > 0 {
var req map[string]any
if err := json.Unmarshal(buf[:n], &req); err == nil {
resp := map[string]any{
"id": req["id"],
"ok": true,
"result": map[string]any{"method": req["method"], "params": req["params"]},
}
payload, _ := json.Marshal(resp)
conn.Write(append(payload, '\n'))
} else {
conn.Write([]byte(`{"ok":false,"error":{"code":"parse","message":"bad json"}}` + "\n"))
}
}
conn.Close()
}
}()
return sockPath
}
func startMockV2TCPSocketWithResult(t *testing.T, result any) string {
t.Helper()
ln, err := net.Listen("tcp", "127.0.0.1:0")
if err != nil {
t.Fatalf("failed to listen on TCP: %v", err)
}
t.Cleanup(func() { ln.Close() })
go func() {
for {
conn, err := ln.Accept()
if err != nil {
return
}
go func(conn net.Conn) {
defer conn.Close()
buf := make([]byte, 4096)
n, _ := conn.Read(buf)
if n == 0 {
return
}
var req map[string]any
if err := json.Unmarshal(buf[:n], &req); err != nil {
_, _ = conn.Write([]byte(`{"ok":false,"error":{"code":"parse","message":"bad json"}}` + "\n"))
return
}
resp := map[string]any{
"id": req["id"],
"ok": true,
"result": result,
}
payload, _ := json.Marshal(resp)
_, _ = conn.Write(append(payload, '\n'))
}(conn)
}
}()
return ln.Addr().String()
}
// startMockTCPSocket creates a TCP listener that responds with a canned response.
func startMockTCPSocket(t *testing.T, response string) string {
t.Helper()
ln, err := net.Listen("tcp", "127.0.0.1:0")
if err != nil {
t.Fatalf("failed to listen on TCP: %v", err)
}
t.Cleanup(func() { ln.Close() })
go func() {
for {
conn, err := ln.Accept()
if err != nil {
return
}
buf := make([]byte, 4096)
n, _ := conn.Read(buf)
_ = n
conn.Write([]byte(response + "\n"))
conn.Close()
}
}()
return ln.Addr().String()
}
func startMockAuthenticatedTCPSocket(t *testing.T, relayID, relayToken, response string) string {
t.Helper()
relayTokenBytes := mustHex(t, relayToken)
ln, err := net.Listen("tcp", "127.0.0.1:0")
if err != nil {
t.Fatalf("failed to listen on TCP: %v", err)
}
t.Cleanup(func() { ln.Close() })
go func() {
for {
conn, err := ln.Accept()
if err != nil {
return
}
go func(conn net.Conn) {
defer conn.Close()
nonce := "testnonce"
challenge, _ := json.Marshal(map[string]any{
"protocol": "cmux-relay-auth",
"version": 1,
"relay_id": relayID,
"nonce": nonce,
})
_, _ = conn.Write(append(challenge, '\n'))
reader := bufio.NewReader(conn)
line, err := reader.ReadString('\n')
if err != nil {
return
}
var authResp map[string]any
if err := json.Unmarshal([]byte(line), &authResp); err != nil {
_, _ = conn.Write([]byte(`{"ok":false}` + "\n"))
return
}
macHex, _ := authResp["mac"].(string)
receivedMAC, err := hex.DecodeString(macHex)
if err != nil {
_, _ = conn.Write([]byte(`{"ok":false}` + "\n"))
return
}
h := hmac.New(sha256.New, relayTokenBytes)
_, _ = io.WriteString(h, fmt.Sprintf("relay_id=%s\nnonce=%s\nversion=%d", relayID, nonce, 1))
expectedMAC := h.Sum(nil)
if !hmac.Equal(receivedMAC, expectedMAC) {
_, _ = conn.Write([]byte(`{"ok":false}` + "\n"))
return
}
_, _ = conn.Write([]byte(`{"ok":true}` + "\n"))
buf := make([]byte, 4096)
n, _ := conn.Read(buf)
_, _ = conn.Write([]byte(response))
if n > 0 && !strings.HasSuffix(response, "\n") {
_, _ = conn.Write([]byte("\n"))
}
}(conn)
}
}()
return ln.Addr().String()
}
func mustHex(t *testing.T, value string) []byte {
t.Helper()
data, err := hex.DecodeString(value)
if err != nil {
t.Fatalf("decode hex: %v", err)
}
return data
}
func TestDialTCPRetrySuccess(t *testing.T) {
// Get a free port, then close the listener so connection is refused initially.
ln, err := net.Listen("tcp", "127.0.0.1:0")
if err != nil {
t.Fatalf("listen: %v", err)
}
addr := ln.Addr().String()
ln.Close()
// Start a listener after a delay so the retry logic finds it.
go func() {
time.Sleep(400 * time.Millisecond)
ln2, err := net.Listen("tcp", addr)
if err != nil {
return
}
defer ln2.Close()
conn, err := ln2.Accept()
if err != nil {
return
}
conn.Close()
}()
conn, err := dialTCPRetry(addr, 3*time.Second, nil)
if err != nil {
t.Fatalf("dialTCPRetry should succeed after retry, got: %v", err)
}
conn.Close()
}
func TestDialTCPRetryTimeout(t *testing.T) {
// Get a free port and close it — nothing will ever listen.
ln, err := net.Listen("tcp", "127.0.0.1:0")
if err != nil {
t.Fatalf("listen: %v", err)
}
addr := ln.Addr().String()
ln.Close()
start := time.Now()
_, err = dialTCPRetry(addr, 600*time.Millisecond, nil)
elapsed := time.Since(start)
if err == nil {
t.Fatal("dialTCPRetry should fail when nothing is listening")
}
if elapsed < 500*time.Millisecond {
t.Fatalf("should have retried for ~600ms, only took %v", elapsed)
}
}
func TestCLIPingV1(t *testing.T) {
sockPath := startMockSocket(t, "pong")
code := runCLI([]string{"--socket", sockPath, "ping"})
if code != 0 {
t.Fatalf("ping should return 0, got %d", code)
}
}
func TestCLIPingV1OverTCP(t *testing.T) {
addr := startMockTCPSocket(t, "pong")
code := runCLI([]string{"--socket", addr, "ping"})
if code != 0 {
t.Fatalf("ping over TCP should return 0, got %d", code)
}
}
func TestCLIPingV1OverAuthenticatedTCPWithEnv(t *testing.T) {
relayID := "relay-1"
relayToken := strings.Repeat("a1", 32)
addr := startMockAuthenticatedTCPSocket(t, relayID, relayToken, "pong")
t.Setenv("CMUX_RELAY_ID", relayID)
t.Setenv("CMUX_RELAY_TOKEN", relayToken)
code := runCLI([]string{"--socket", addr, "ping"})
if code != 0 {
t.Fatalf("ping over authenticated TCP should return 0, got %d", code)
}
}
func TestCLIPingV1OverAuthenticatedTCPWithRelayFile(t *testing.T) {
relayID := "relay-2"
relayToken := strings.Repeat("b2", 32)
addr := startMockAuthenticatedTCPSocket(t, relayID, relayToken, "pong")
_, port, err := net.SplitHostPort(addr)
if err != nil {
t.Fatalf("split host port: %v", err)
}
home := t.TempDir()
t.Setenv("HOME", home)
t.Setenv("CMUX_RELAY_ID", "")
t.Setenv("CMUX_RELAY_TOKEN", "")
relayDir := filepath.Join(home, ".cmux", "relay")
if err := os.MkdirAll(relayDir, 0o700); err != nil {
t.Fatalf("mkdir relay dir: %v", err)
}
authPayload, _ := json.Marshal(relayAuthState{RelayID: relayID, RelayToken: relayToken})
if err := os.WriteFile(filepath.Join(relayDir, port+".auth"), authPayload, 0o600); err != nil {
t.Fatalf("write auth file: %v", err)
}
code := runCLI([]string{"--socket", addr, "ping"})
if code != 0 {
t.Fatalf("ping over authenticated TCP file relay should return 0, got %d", code)
}
}
func TestDialSocketDetection(t *testing.T) {
// Unix socket paths should attempt Unix dial
for _, path := range []string{"/tmp/cmux-nonexistent-test-99999.sock", "/var/run/cmux-nonexistent.sock"} {
conn, err := dialSocket(path, nil)
if conn != nil {
conn.Close()
}
// We expect a connection error (not found), not a panic
if err == nil {
t.Fatalf("dialSocket(%q) should fail for non-existent path", path)
}
}
// TCP addresses should attempt TCP dial
ln, err := net.Listen("tcp", "127.0.0.1:0")
if err != nil {
t.Fatalf("listen: %v", err)
}
defer ln.Close()
go func() {
conn, _ := ln.Accept()
if conn != nil {
conn.Close()
}
}()
conn, err := dialSocket(ln.Addr().String(), nil)
if err != nil {
t.Fatalf("dialSocket(%q) should succeed for TCP: %v", ln.Addr().String(), err)
}
conn.Close()
}
func TestCLINewWindowV1(t *testing.T) {
sockPath := startMockSocket(t, "OK window_id=abc123")
code := runCLI([]string{"--socket", sockPath, "new-window"})
if code != 0 {
t.Fatalf("new-window should return 0, got %d", code)
}
}
func TestSocketRoundTripReadsFullMultilineV1Response(t *testing.T) {
addr := startMockTCPSocket(t, "window:alpha\nwindow:beta\nwindow:gamma")
resp, err := socketRoundTrip(addr, "list_windows", nil)
if err != nil {
t.Fatalf("socketRoundTrip should succeed, got error: %v", err)
}
want := "window:alpha\nwindow:beta\nwindow:gamma"
if resp != want {
t.Fatalf("socketRoundTrip truncated v1 response: got %q want %q", resp, want)
}
}
func TestCLICloseWindowV1(t *testing.T) {
// Verify that the flag value is appended to the v1 command
dir := t.TempDir()
sockPath := filepath.Join(dir, "cmux.sock")
var received string
ln, err := net.Listen("unix", sockPath)
if err != nil {
t.Fatalf("listen: %v", err)
}
t.Cleanup(func() { ln.Close() })
go func() {
conn, err := ln.Accept()
if err != nil {
return
}
buf := make([]byte, 4096)
n, _ := conn.Read(buf)
received = strings.TrimSpace(string(buf[:n]))
conn.Write([]byte("OK\n"))
conn.Close()
}()
code := runCLI([]string{"--socket", sockPath, "close-window", "--window", "win-42"})
if code != 0 {
t.Fatalf("close-window should return 0, got %d", code)
}
if received != "close_window win-42" {
t.Fatalf("expected 'close_window win-42', got %q", received)
}
}
func TestCLIListWorkspacesV2(t *testing.T) {
sockPath := startMockV2Socket(t)
code := runCLI([]string{"--socket", sockPath, "--json", "list-workspaces"})
if code != 0 {
t.Fatalf("list-workspaces should return 0, got %d", code)
}
}
func TestCLIListWorkspacesV2DefaultOutputShowsResult(t *testing.T) {
sockPath := startMockV2TCPSocketWithResult(t, map[string]any{"method": "workspace.list", "params": map[string]any{}})
output := captureStdout(t, func() {
code := runCLI([]string{"--socket", sockPath, "list-workspaces"})
if code != 0 {
t.Fatalf("list-workspaces should return 0, got %d", code)
}
})
if !strings.Contains(output, "\"method\": \"workspace.list\"") {
t.Fatalf("expected default output to include result payload, got %q", output)
}
}
func TestCLINotifyDefaultOutputPrintsOKForEmptyResult(t *testing.T) {
sockPath := startMockV2TCPSocketWithResult(t, map[string]any{})
output := captureStdout(t, func() {
code := runCLI([]string{"--socket", sockPath, "notify", "--body", "hi"})
if code != 0 {
t.Fatalf("notify should return 0, got %d", code)
}
})
if strings.TrimSpace(output) != "OK" {
t.Fatalf("expected empty-result command to print OK, got %q", output)
}
}
func TestCLIRPCPassthrough(t *testing.T) {
sockPath := startMockV2Socket(t)
code := runCLI([]string{"--socket", sockPath, "rpc", "system.capabilities"})
if code != 0 {
t.Fatalf("rpc should return 0, got %d", code)
}
}
func TestCLIRPCWithParams(t *testing.T) {
sockPath := startMockV2Socket(t)
code := runCLI([]string{"--socket", sockPath, "rpc", "workspace.create", `{"title":"test"}`})
if code != 0 {
t.Fatalf("rpc with params should return 0, got %d", code)
}
}
func TestCLIUnknownCommand(t *testing.T) {
code := runCLI([]string{"--socket", "/dev/null", "does-not-exist"})
if code != 2 {
t.Fatalf("unknown command should return 2, got %d", code)
}
}
func TestCLINoSocket(t *testing.T) {
// Without CMUX_SOCKET_PATH set, should fail
os.Unsetenv("CMUX_SOCKET_PATH")
code := runCLI([]string{"ping"})
if code != 1 {
t.Fatalf("missing socket should return 1, got %d", code)
}
}
func TestCLISocketEnvVar(t *testing.T) {
sockPath := startMockSocket(t, "pong")
os.Setenv("CMUX_SOCKET_PATH", sockPath)
defer os.Unsetenv("CMUX_SOCKET_PATH")
code := runCLI([]string{"ping"})
if code != 0 {
t.Fatalf("ping with env socket should return 0, got %d", code)
}
}
func TestCLIV2FlagMapping(t *testing.T) {
// Verify that --workspace gets mapped to workspace_id in params
dir := t.TempDir()
sockPath := filepath.Join(dir, "cmux.sock")
var receivedParams map[string]any
ln, err := net.Listen("unix", sockPath)
if err != nil {
t.Fatalf("listen: %v", err)
}
t.Cleanup(func() { ln.Close() })
go func() {
conn, err := ln.Accept()
if err != nil {
return
}
buf := make([]byte, 4096)
n, _ := conn.Read(buf)
var req map[string]any
json.Unmarshal(buf[:n], &req)
receivedParams, _ = req["params"].(map[string]any)
resp := map[string]any{"id": req["id"], "ok": true, "result": map[string]any{}}
payload, _ := json.Marshal(resp)
conn.Write(append(payload, '\n'))
conn.Close()
}()
code := runCLI([]string{"--socket", sockPath, "--json", "close-workspace", "--workspace", "ws-abc"})
if code != 0 {
t.Fatalf("close-workspace should return 0, got %d", code)
}
if receivedParams["workspace_id"] != "ws-abc" {
t.Fatalf("expected workspace_id=ws-abc, got %v", receivedParams)
}
}
func TestBusyboxArgv0Detection(t *testing.T) {
// Verify that when argv[0] base is "cmux", we enter CLI mode
base := filepath.Base("cmux")
if base != "cmux" {
t.Fatalf("expected base 'cmux', got %q", base)
}
base2 := filepath.Base("/home/user/.cmux/bin/cmux")
if base2 != "cmux" {
t.Fatalf("expected base 'cmux', got %q", base2)
}
base3 := filepath.Base("cmuxd-remote")
if base3 == "cmux" {
t.Fatalf("cmuxd-remote should not match cmux")
}
}
func TestCLIBrowserSubcommand(t *testing.T) {
sockPath := startMockV2Socket(t)
code := runCLI([]string{"--socket", sockPath, "--json", "browser", "open", "--url", "https://example.com"})
if code != 0 {
t.Fatalf("browser open should return 0, got %d", code)
}
}
func TestCLINoArgs(t *testing.T) {
code := runCLI([]string{})
if code != 2 {
t.Fatalf("no args should return 2, got %d", code)
}
}
func TestCLIHelpFlag(t *testing.T) {
code := runCLI([]string{"--help"})
if code != 0 {
t.Fatalf("--help should return 0, got %d", code)
}
}
func TestCLIHelpCommand(t *testing.T) {
code := runCLI([]string{"help"})
if code != 0 {
t.Fatalf("help should return 0, got %d", code)
}
}
func TestFlagToParamKey(t *testing.T) {
tests := []struct {
input, expected string
}{
{"workspace", "workspace_id"},
{"surface", "surface_id"},
{"panel", "panel_id"},
{"pane", "pane_id"},
{"window", "window_id"},
{"command", "initial_command"},
{"name", "title"},
{"working-directory", "working_directory"},
{"title", "title"},
{"url", "url"},
{"direction", "direction"},
}
for _, tc := range tests {
got := flagToParamKey(tc.input)
if got != tc.expected {
t.Errorf("flagToParamKey(%q) = %q, want %q", tc.input, got, tc.expected)
}
}
}
func TestParseFlags(t *testing.T) {
args := []string{"positional-cmd", "--workspace", "ws-1", "--surface", "sf-2", "--unknown", "val"}
result := parseFlags(args, []string{"workspace", "surface"})
if result.flags["workspace"] != "ws-1" {
t.Errorf("expected workspace=ws-1, got %q", result.flags["workspace"])
}
if result.flags["surface"] != "sf-2" {
t.Errorf("expected surface=sf-2, got %q", result.flags["surface"])
}
if _, ok := result.flags["unknown"]; ok {
t.Errorf("unknown flag should not be parsed")
}
if len(result.positional) == 0 || result.positional[0] != "positional-cmd" {
t.Errorf("expected first positional=positional-cmd, got %v", result.positional)
}
}
func TestCLIEnvVarDefaults(t *testing.T) {
// Test that CMUX_WORKSPACE_ID and CMUX_SURFACE_ID are used as defaults
dir := t.TempDir()
sockPath := filepath.Join(dir, "cmux.sock")
var receivedParams map[string]any
ln, err := net.Listen("unix", sockPath)
if err != nil {
t.Fatalf("listen: %v", err)
}
t.Cleanup(func() { ln.Close() })
go func() {
conn, err := ln.Accept()
if err != nil {
return
}
buf := make([]byte, 4096)
n, _ := conn.Read(buf)
var req map[string]any
json.Unmarshal(buf[:n], &req)
receivedParams, _ = req["params"].(map[string]any)
resp := map[string]any{"id": req["id"], "ok": true, "result": map[string]any{}}
payload, _ := json.Marshal(resp)
conn.Write(append(payload, '\n'))
conn.Close()
}()
os.Setenv("CMUX_WORKSPACE_ID", "env-ws-id")
os.Setenv("CMUX_SURFACE_ID", "env-sf-id")
defer os.Unsetenv("CMUX_WORKSPACE_ID")
defer os.Unsetenv("CMUX_SURFACE_ID")
code := runCLI([]string{"--socket", sockPath, "--json", "close-surface"})
if code != 0 {
t.Fatalf("close-surface should return 0, got %d", code)
}
if receivedParams["workspace_id"] != "env-ws-id" {
t.Errorf("expected workspace_id from env, got %v", receivedParams["workspace_id"])
}
if receivedParams["surface_id"] != "env-sf-id" {
t.Errorf("expected surface_id from env, got %v", receivedParams["surface_id"])
}
}

1034
daemon/remote/cmd/cmuxd-remote/main.go Normal file
View file

File diff suppressed because it is too large Load diff

531
daemon/remote/cmd/cmuxd-remote/main_test.go Normal file
View file

@ -0,0 +1,531 @@
package main
import (
"bytes"
"encoding/base64"
"encoding/json"
"io"
"math"
"net"
"strconv"
"strings"
"testing"
"time"
)
func TestRunVersion(t *testing.T) {
var out bytes.Buffer
code := run([]string{"version"}, strings.NewReader(""), &out, &bytes.Buffer{})
if code != 0 {
t.Fatalf("run version exit code = %d, want 0", code)
}
if strings.TrimSpace(out.String()) == "" {
t.Fatalf("version output should not be empty")
}
}
func TestRunStdioHelloAndPing(t *testing.T) {
input := strings.NewReader(
`{"id":1,"method":"hello","params":{}}` + "\n" +
`{"id":2,"method":"ping","params":{}}` + "\n",
)
var out bytes.Buffer
code := run([]string{"serve", "--stdio"}, input, &out, &bytes.Buffer{})
if code != 0 {
t.Fatalf("run serve exit code = %d, want 0", code)
}
lines := strings.Split(strings.TrimSpace(out.String()), "\n")
if len(lines) != 2 {
t.Fatalf("got %d response lines, want 2: %q", len(lines), out.String())
}
var first map[string]any
if err := json.Unmarshal([]byte(lines[0]), &first); err != nil {
t.Fatalf("failed to decode first response: %v", err)
}
if ok, _ := first["ok"].(bool); !ok {
t.Fatalf("first response should be ok=true: %v", first)
}
firstResult, _ := first["result"].(map[string]any)
if firstResult == nil {
t.Fatalf("first response missing result object: %v", first)
}
capabilities, _ := firstResult["capabilities"].([]any)
if len(capabilities) < 2 {
t.Fatalf("hello should return capabilities: %v", firstResult)
}
var second map[string]any
if err := json.Unmarshal([]byte(lines[1]), &second); err != nil {
t.Fatalf("failed to decode second response: %v", err)
}
if ok, _ := second["ok"].(bool); !ok {
t.Fatalf("second response should be ok=true: %v", second)
}
}
func TestRunStdioInvalidJSONAndUnknownMethod(t *testing.T) {
input := strings.NewReader(
`{"id":1,"method":"hello","params":{}` + "\n" +
`{"id":2,"method":"unknown","params":{}}` + "\n",
)
var out bytes.Buffer
code := run([]string{"serve", "--stdio"}, input, &out, &bytes.Buffer{})
if code != 0 {
t.Fatalf("run serve exit code = %d, want 0", code)
}
lines := strings.Split(strings.TrimSpace(out.String()), "\n")
if len(lines) != 2 {
t.Fatalf("got %d response lines, want 2: %q", len(lines), out.String())
}
var first map[string]any
if err := json.Unmarshal([]byte(lines[0]), &first); err != nil {
t.Fatalf("failed to decode first response: %v", err)
}
if ok, _ := first["ok"].(bool); ok {
t.Fatalf("first response should be ok=false for invalid JSON: %v", first)
}
firstError, _ := first["error"].(map[string]any)
if got := firstError["code"]; got != "invalid_request" {
t.Fatalf("invalid JSON should return invalid_request; got=%v payload=%v", got, first)
}
var second map[string]any
if err := json.Unmarshal([]byte(lines[1]), &second); err != nil {
t.Fatalf("failed to decode second response: %v", err)
}
if ok, _ := second["ok"].(bool); ok {
t.Fatalf("second response should be ok=false for unknown method: %v", second)
}
secondError, _ := second["error"].(map[string]any)
if got := secondError["code"]; got != "method_not_found" {
t.Fatalf("unknown method should return method_not_found; got=%v payload=%v", got, second)
}
}
func TestRunStdioSessionResizeFlow(t *testing.T) {
input := strings.NewReader(
`{"id":1,"method":"session.open","params":{"session_id":"sess-stdio"}}` + "\n" +
`{"id":2,"method":"session.attach","params":{"session_id":"sess-stdio","attachment_id":"a1","cols":120,"rows":40}}` + "\n" +
`{"id":3,"method":"session.attach","params":{"session_id":"sess-stdio","attachment_id":"a2","cols":90,"rows":30}}` + "\n" +
`{"id":4,"method":"session.status","params":{"session_id":"sess-stdio"}}` + "\n",
)
var out bytes.Buffer
code := run([]string{"serve", "--stdio"}, input, &out, &bytes.Buffer{})
if code != 0 {
t.Fatalf("run serve exit code = %d, want 0", code)
}
lines := strings.Split(strings.TrimSpace(out.String()), "\n")
if len(lines) != 4 {
t.Fatalf("got %d response lines, want 4: %q", len(lines), out.String())
}
var status map[string]any
if err := json.Unmarshal([]byte(lines[3]), &status); err != nil {
t.Fatalf("failed to decode status response: %v", err)
}
if ok, _ := status["ok"].(bool); !ok {
t.Fatalf("session.status should be ok=true: %v", status)
}
result, _ := status["result"].(map[string]any)
if result == nil {
t.Fatalf("session.status missing result object: %v", status)
}
effectiveCols, _ := result["effective_cols"].(float64)
effectiveRows, _ := result["effective_rows"].(float64)
if int(effectiveCols) != 90 || int(effectiveRows) != 30 {
t.Fatalf("session smallest-wins effective size mismatch: got=%vx%v payload=%v", effectiveCols, effectiveRows, result)
}
}
func TestProxyStreamRoundTrip(t *testing.T) {
listener, err := net.Listen("tcp", "127.0.0.1:0")
if err != nil {
t.Fatalf("listen failed: %v", err)
}
defer listener.Close()
done := make(chan struct{})
go func() {
defer close(done)
conn, acceptErr := listener.Accept()
if acceptErr != nil {
return
}
defer conn.Close()
buffer := make([]byte, 4)
if _, readErr := io.ReadFull(conn, buffer); readErr != nil {
return
}
if string(buffer) != "ping" {
return
}
_, _ = conn.Write([]byte("pong"))
}()
server := &rpcServer{
nextStreamID: 1,
nextSessionID: 1,
streams: map[string]net.Conn{},
sessions: map[string]*sessionState{},
}
defer server.closeAll()
port := listener.Addr().(*net.TCPAddr).Port
openResp := server.handleRequest(rpcRequest{
ID: 1,
Method: "proxy.open",
Params: map[string]any{
"host": "127.0.0.1",
"port": port,
"timeout_ms": 1000,
},
})
if !openResp.OK {
t.Fatalf("proxy.open failed: %+v", openResp)
}
openResult, _ := openResp.Result.(map[string]any)
streamID, _ := openResult["stream_id"].(string)
if streamID == "" {
t.Fatalf("proxy.open missing stream_id: %+v", openResp)
}
writeResp := server.handleRequest(rpcRequest{
ID: 2,
Method: "proxy.write",
Params: map[string]any{
"stream_id": streamID,
"data_base64": base64.StdEncoding.EncodeToString([]byte("ping")),
},
})
if !writeResp.OK {
t.Fatalf("proxy.write failed: %+v", writeResp)
}
readResp := server.handleRequest(rpcRequest{
ID: 3,
Method: "proxy.read",
Params: map[string]any{
"stream_id": streamID,
"max_bytes": 8,
"timeout_ms": 1000,
},
})
if !readResp.OK {
t.Fatalf("proxy.read failed: %+v", readResp)
}
readResult, _ := readResp.Result.(map[string]any)
dataBase64, _ := readResult["data_base64"].(string)
data, decodeErr := base64.StdEncoding.DecodeString(dataBase64)
if decodeErr != nil {
t.Fatalf("proxy.read returned invalid base64: %v", decodeErr)
}
if string(data) != "pong" {
t.Fatalf("proxy.read payload=%q, want %q", string(data), "pong")
}
closeResp := server.handleRequest(rpcRequest{
ID: 4,
Method: "proxy.close",
Params: map[string]any{
"stream_id": streamID,
},
})
if !closeResp.OK {
t.Fatalf("proxy.close failed: %+v", closeResp)
}
select {
case <-done:
case <-time.After(2 * time.Second):
t.Fatalf("proxy test server goroutine did not finish")
}
}
func TestGetIntParamRejectsFractionalFloat64(t *testing.T) {
params := map[string]any{
"port": 80.9,
"timeout_ms": 100.0,
}
if _, ok := getIntParam(params, "port"); ok {
t.Fatalf("fractional float64 should be rejected")
}
timeout, ok := getIntParam(params, "timeout_ms")
if !ok {
t.Fatalf("integral float64 should be accepted")
}
if timeout != 100 {
t.Fatalf("timeout_ms = %d, want 100", timeout)
}
}
func TestRunStdioOversizedFrameContinuesServing(t *testing.T) {
oversized := `{"id":1,"method":"ping","params":{"blob":"` + strings.Repeat("a", maxRPCFrameBytes) + `"}}`
input := strings.NewReader(oversized + "\n" + `{"id":2,"method":"ping","params":{}}` + "\n")
var out bytes.Buffer
code := run([]string{"serve", "--stdio"}, input, &out, &bytes.Buffer{})
if code != 0 {
t.Fatalf("run serve exit code = %d, want 0", code)
}
lines := strings.Split(strings.TrimSpace(out.String()), "\n")
if len(lines) != 2 {
t.Fatalf("got %d response lines, want 2: %q", len(lines), out.String())
}
var first map[string]any
if err := json.Unmarshal([]byte(lines[0]), &first); err != nil {
t.Fatalf("failed to decode first response: %v", err)
}
if ok, _ := first["ok"].(bool); ok {
t.Fatalf("first response should be oversized-frame error: %v", first)
}
firstError, _ := first["error"].(map[string]any)
if got := firstError["code"]; got != "invalid_request" {
t.Fatalf("oversized frame should return invalid_request; got=%v payload=%v", got, first)
}
var second map[string]any
if err := json.Unmarshal([]byte(lines[1]), &second); err != nil {
t.Fatalf("failed to decode second response: %v", err)
}
if ok, _ := second["ok"].(bool); !ok {
t.Fatalf("second response should still be handled after oversized frame: %v", second)
}
}
func TestProxyOpenInvalidParams(t *testing.T) {
server := &rpcServer{
nextStreamID: 1,
nextSessionID: 1,
streams: map[string]net.Conn{},
sessions: map[string]*sessionState{},
}
defer server.closeAll()
resp := server.handleRequest(rpcRequest{
ID: 1,
Method: "proxy.open",
Params: map[string]any{
"host": "127.0.0.1",
"port": strconv.Itoa(8080),
},
})
if resp.OK {
t.Fatalf("proxy.open with invalid port type should fail: %+v", resp)
}
errObj, _ := resp.Error, resp.Error
if errObj == nil || errObj.Code != "invalid_params" {
t.Fatalf("proxy.open invalid params should return invalid_params: %+v", resp)
}
}
func TestSessionResizeCoordinator(t *testing.T) {
server := &rpcServer{
nextStreamID: 1,
nextSessionID: 1,
streams: map[string]net.Conn{},
sessions: map[string]*sessionState{},
}
defer server.closeAll()
openResp := server.handleRequest(rpcRequest{
ID: 1,
Method: "session.open",
Params: map[string]any{
"session_id": "sess-rz",
},
})
if !openResp.OK {
t.Fatalf("session.open failed: %+v", openResp)
}
attachSmall := server.handleRequest(rpcRequest{
ID: 2,
Method: "session.attach",
Params: map[string]any{
"session_id": "sess-rz",
"attachment_id": "a-small",
"cols": 90,
"rows": 30,
},
})
assertEffectiveSize(t, attachSmall, 90, 30)
attachLarge := server.handleRequest(rpcRequest{
ID: 3,
Method: "session.attach",
Params: map[string]any{
"session_id": "sess-rz",
"attachment_id": "a-large",
"cols": 120,
"rows": 40,
},
})
assertEffectiveSize(t, attachLarge, 90, 30) // RZ-001: smallest wins
resizeLarge := server.handleRequest(rpcRequest{
ID: 4,
Method: "session.resize",
Params: map[string]any{
"session_id": "sess-rz",
"attachment_id": "a-large",
"cols": 200,
"rows": 60,
},
})
assertEffectiveSize(t, resizeLarge, 90, 30) // RZ-002: still bounded by smallest
detachSmall := server.handleRequest(rpcRequest{
ID: 5,
Method: "session.detach",
Params: map[string]any{
"session_id": "sess-rz",
"attachment_id": "a-small",
},
})
assertEffectiveSize(t, detachSmall, 200, 60) // RZ-003: expands to next smallest
detachLarge := server.handleRequest(rpcRequest{
ID: 6,
Method: "session.detach",
Params: map[string]any{
"session_id": "sess-rz",
"attachment_id": "a-large",
},
})
assertEffectiveSize(t, detachLarge, 200, 60) // no attachments: keep last-known size
assertAttachmentCount(t, detachLarge, 0)
reattach := server.handleRequest(rpcRequest{
ID: 7,
Method: "session.attach",
Params: map[string]any{
"session_id": "sess-rz",
"attachment_id": "a-reconnect",
"cols": 110,
"rows": 50,
},
})
assertEffectiveSize(t, reattach, 110, 50) // RZ-004: recompute from active attachments on reattach
}
func TestSessionInvalidParamsAndNotFound(t *testing.T) {
server := &rpcServer{
nextStreamID: 1,
nextSessionID: 1,
streams: map[string]net.Conn{},
sessions: map[string]*sessionState{},
}
defer server.closeAll()
missingSession := server.handleRequest(rpcRequest{
ID: 1,
Method: "session.attach",
Params: map[string]any{
"session_id": "missing",
"attachment_id": "a1",
"cols": 80,
"rows": 24,
},
})
if missingSession.OK || missingSession.Error == nil || missingSession.Error.Code != "not_found" {
t.Fatalf("session.attach on missing session should return not_found: %+v", missingSession)
}
badSize := server.handleRequest(rpcRequest{
ID: 2,
Method: "session.attach",
Params: map[string]any{
"session_id": "missing",
"attachment_id": "a1",
"cols": 0,
"rows": 24,
},
})
if badSize.OK || badSize.Error == nil || badSize.Error.Code != "invalid_params" {
t.Fatalf("session.attach with cols=0 should return invalid_params: %+v", badSize)
}
}
func assertEffectiveSize(t *testing.T, resp rpcResponse, wantCols, wantRows int) {
t.Helper()
if !resp.OK {
t.Fatalf("expected ok response, got error: %+v", resp)
}
result, ok := resp.Result.(map[string]any)
if !ok {
t.Fatalf("response missing result map: %+v", resp)
}
gotCols := asInt(t, result["effective_cols"], "effective_cols")
gotRows := asInt(t, result["effective_rows"], "effective_rows")
if gotCols != wantCols || gotRows != wantRows {
t.Fatalf("effective size = %dx%d, want %dx%d payload=%+v", gotCols, gotRows, wantCols, wantRows, result)
}
}
func assertAttachmentCount(t *testing.T, resp rpcResponse, want int) {
t.Helper()
if !resp.OK {
t.Fatalf("expected ok response, got error: %+v", resp)
}
result, ok := resp.Result.(map[string]any)
if !ok {
t.Fatalf("response missing result map: %+v", resp)
}
attachments, ok := result["attachments"].([]map[string]any)
if ok {
if len(attachments) != want {
t.Fatalf("attachments len = %d, want %d payload=%+v", len(attachments), want, result)
}
return
}
attachmentsAny, ok := result["attachments"].([]any)
if !ok {
t.Fatalf("attachments field has unexpected type (%T) payload=%+v", result["attachments"], result)
}
if len(attachmentsAny) != want {
t.Fatalf("attachments len = %d, want %d payload=%+v", len(attachmentsAny), want, result)
}
}
func asInt(t *testing.T, value any, field string) int {
t.Helper()
switch typed := value.(type) {
case int:
return typed
case int8:
return int(typed)
case int16:
return int(typed)
case int32:
return int(typed)
case int64:
return int(typed)
case uint:
return int(typed)
case uint8:
return int(typed)
case uint16:
return int(typed)
case uint32:
return int(typed)
case uint64:
return int(typed)
case float64:
if typed != math.Trunc(typed) {
t.Fatalf("%s should be integer-valued, got %v", field, typed)
}
return int(typed)
default:
t.Fatalf("%s has unexpected type %T (%v)", field, value, value)
return 0
}
}

3
daemon/remote/go.mod Normal file
View file

@ -0,0 +1,3 @@
module github.com/manaflow-ai/cmux/daemon/remote
go 1.22