feat: improve port forwarding, remote executor setup, and CLI port config (#388)
This commit is contained in:
@@ -93,6 +93,7 @@ func provideGateway(
|
||||
gateway.SetLSPHandler(lifecycleMgr, userSvc, lspinstaller.NewRegistry(dataDir, log))
|
||||
gateway.SetVscodeProxy(lifecycleMgr)
|
||||
gateway.SetPortProxy(lifecycleMgr)
|
||||
gateway.SetPortTunnel(lifecycleMgr)
|
||||
}
|
||||
|
||||
orchestratorHandlers := orchestratorhandlers.NewHandlers(orchestratorSvc, log)
|
||||
@@ -149,7 +150,7 @@ func provideGateway(
|
||||
vscodeHandlers := agenthandlers.NewVscodeHandlers(lifecycleMgr, gateway.VscodeProxyHandler, log)
|
||||
vscodeHandlers.RegisterHandlers(gateway.Dispatcher)
|
||||
|
||||
portHandlers := agenthandlers.NewPortHandlers(lifecycleMgr, log)
|
||||
portHandlers := agenthandlers.NewPortHandlers(lifecycleMgr, gateway.TunnelManager, log)
|
||||
portHandlers.RegisterHandlers(gateway.Dispatcher)
|
||||
}
|
||||
|
||||
|
||||
@@ -10,16 +10,27 @@ import (
|
||||
"go.uber.org/zap"
|
||||
)
|
||||
|
||||
// PortHandlers provides WebSocket handlers for port listing operations.
|
||||
// TunnelController manages port tunnels.
|
||||
// Implemented by websocket.TunnelManager.
|
||||
type TunnelController interface {
|
||||
StartTunnel(sessionID string, port int, tunnelPort int) (int, error)
|
||||
StopTunnel(sessionID string, port int) error
|
||||
// ListTunnels returns []TunnelInfo-like structs with port/tunnel_port fields.
|
||||
ListTunnels(sessionID string) any
|
||||
}
|
||||
|
||||
// PortHandlers provides WebSocket handlers for port listing and tunnel operations.
|
||||
type PortHandlers struct {
|
||||
lifecycleMgr *lifecycle.Manager
|
||||
tunnelCtrl TunnelController
|
||||
logger *logger.Logger
|
||||
}
|
||||
|
||||
// NewPortHandlers creates a new PortHandlers instance.
|
||||
func NewPortHandlers(lifecycleMgr *lifecycle.Manager, log *logger.Logger) *PortHandlers {
|
||||
func NewPortHandlers(lifecycleMgr *lifecycle.Manager, tunnelCtrl TunnelController, log *logger.Logger) *PortHandlers {
|
||||
return &PortHandlers{
|
||||
lifecycleMgr: lifecycleMgr,
|
||||
tunnelCtrl: tunnelCtrl,
|
||||
logger: log.WithFields(zap.String("component", "port_handlers")),
|
||||
}
|
||||
}
|
||||
@@ -27,6 +38,12 @@ func NewPortHandlers(lifecycleMgr *lifecycle.Manager, log *logger.Logger) *PortH
|
||||
// RegisterHandlers registers port handlers with the WebSocket dispatcher.
|
||||
func (h *PortHandlers) RegisterHandlers(d *ws.Dispatcher) {
|
||||
d.RegisterFunc(ws.ActionPortList, h.wsPortList)
|
||||
|
||||
if h.tunnelCtrl != nil {
|
||||
d.RegisterFunc(ws.ActionPortTunnelStart, h.wsTunnelStart)
|
||||
d.RegisterFunc(ws.ActionPortTunnelStop, h.wsTunnelStop)
|
||||
d.RegisterFunc(ws.ActionPortTunnelList, h.wsTunnelList)
|
||||
}
|
||||
}
|
||||
|
||||
// portListRequest for port.list action.
|
||||
@@ -61,7 +78,95 @@ func (h *PortHandlers) wsPortList(ctx context.Context, msg *ws.Message) (*ws.Mes
|
||||
fmt.Sprintf("failed to list ports: %v", err), nil)
|
||||
}
|
||||
|
||||
return ws.NewResponse(msg.ID, msg.Action, map[string]interface{}{
|
||||
return ws.NewResponse(msg.ID, msg.Action, map[string]any{
|
||||
"ports": ports,
|
||||
})
|
||||
}
|
||||
|
||||
type tunnelStartRequest struct {
|
||||
SessionID string `json:"session_id"`
|
||||
Port int `json:"port"`
|
||||
TunnelPort int `json:"tunnel_port"`
|
||||
}
|
||||
|
||||
func (h *PortHandlers) wsTunnelStart(_ context.Context, msg *ws.Message) (*ws.Message, error) {
|
||||
var req tunnelStartRequest
|
||||
if err := msg.ParsePayload(&req); err != nil {
|
||||
return ws.NewError(msg.ID, msg.Action, ws.ErrorCodeBadRequest, "invalid request: "+err.Error(), nil)
|
||||
}
|
||||
|
||||
if req.SessionID == "" {
|
||||
return ws.NewError(msg.ID, msg.Action, ws.ErrorCodeValidation, "session_id is required", nil)
|
||||
}
|
||||
if req.Port < 1 || req.Port > 65535 {
|
||||
return ws.NewError(msg.ID, msg.Action, ws.ErrorCodeValidation, "port must be 1-65535", nil)
|
||||
}
|
||||
if req.TunnelPort != 0 && (req.TunnelPort < 1 || req.TunnelPort > 65535) {
|
||||
return ws.NewError(msg.ID, msg.Action, ws.ErrorCodeValidation, "tunnel_port must be 0 (random) or 1-65535", nil)
|
||||
}
|
||||
|
||||
tunnelPort, err := h.tunnelCtrl.StartTunnel(req.SessionID, req.Port, req.TunnelPort)
|
||||
if err != nil {
|
||||
h.logger.Error("failed to start tunnel",
|
||||
zap.Error(err),
|
||||
zap.String("session_id", req.SessionID),
|
||||
zap.Int("port", req.Port))
|
||||
return ws.NewError(msg.ID, msg.Action, ws.ErrorCodeInternalError,
|
||||
fmt.Sprintf("failed to start tunnel: %v", err), nil)
|
||||
}
|
||||
|
||||
return ws.NewResponse(msg.ID, msg.Action, map[string]any{
|
||||
"tunnel_port": tunnelPort,
|
||||
})
|
||||
}
|
||||
|
||||
type tunnelStopRequest struct {
|
||||
SessionID string `json:"session_id"`
|
||||
Port int `json:"port"`
|
||||
}
|
||||
|
||||
func (h *PortHandlers) wsTunnelStop(_ context.Context, msg *ws.Message) (*ws.Message, error) {
|
||||
var req tunnelStopRequest
|
||||
if err := msg.ParsePayload(&req); err != nil {
|
||||
return ws.NewError(msg.ID, msg.Action, ws.ErrorCodeBadRequest, "invalid request: "+err.Error(), nil)
|
||||
}
|
||||
|
||||
if req.SessionID == "" {
|
||||
return ws.NewError(msg.ID, msg.Action, ws.ErrorCodeValidation, "session_id is required", nil)
|
||||
}
|
||||
if req.Port < 1 || req.Port > 65535 {
|
||||
return ws.NewError(msg.ID, msg.Action, ws.ErrorCodeValidation, "port must be 1-65535", nil)
|
||||
}
|
||||
|
||||
if err := h.tunnelCtrl.StopTunnel(req.SessionID, req.Port); err != nil {
|
||||
h.logger.Error("failed to stop tunnel",
|
||||
zap.Error(err),
|
||||
zap.String("session_id", req.SessionID),
|
||||
zap.Int("port", req.Port))
|
||||
return ws.NewError(msg.ID, msg.Action, ws.ErrorCodeInternalError,
|
||||
fmt.Sprintf("failed to stop tunnel: %v", err), nil)
|
||||
}
|
||||
|
||||
return ws.NewResponse(msg.ID, msg.Action, map[string]any{})
|
||||
}
|
||||
|
||||
type tunnelListRequest struct {
|
||||
SessionID string `json:"session_id"`
|
||||
}
|
||||
|
||||
func (h *PortHandlers) wsTunnelList(_ context.Context, msg *ws.Message) (*ws.Message, error) {
|
||||
var req tunnelListRequest
|
||||
if err := msg.ParsePayload(&req); err != nil {
|
||||
return ws.NewError(msg.ID, msg.Action, ws.ErrorCodeBadRequest, "invalid request: "+err.Error(), nil)
|
||||
}
|
||||
|
||||
if req.SessionID == "" {
|
||||
return ws.NewError(msg.ID, msg.Action, ws.ErrorCodeValidation, "session_id is required", nil)
|
||||
}
|
||||
|
||||
tunnels := h.tunnelCtrl.ListTunnels(req.SessionID)
|
||||
|
||||
return ws.NewResponse(msg.ID, msg.Action, map[string]any{
|
||||
"tunnels": tunnels,
|
||||
})
|
||||
}
|
||||
|
||||
@@ -68,7 +68,7 @@ const defaultSpritesPrepareScript = `#!/bin/bash
|
||||
# Prepare Sprites.dev cloud sandbox
|
||||
#
|
||||
# Pre-installed tools (no need to install):
|
||||
# git, curl, wget, gh (GitHub CLI), node, pnpm (via corepack), python, go,
|
||||
# git, curl, wget, gh (GitHub CLI), node, python, go,
|
||||
# build-essential, openssh-client, ca-certificates
|
||||
|
||||
set -euo pipefail
|
||||
@@ -77,8 +77,17 @@ set -euo pipefail
|
||||
mkdir -p ~/.ssh
|
||||
ssh-keyscan -t ed25519 github.com gitlab.com bitbucket.org >> ~/.ssh/known_hosts 2>/dev/null
|
||||
|
||||
# ---- Enable pnpm via corepack ----
|
||||
corepack enable
|
||||
# ---- Configure git/gh for HTTPS auth (token-based, no SSH keys needed) ----
|
||||
# Rewrite SSH URLs to HTTPS so git clone git@github.com:... works via token auth
|
||||
git config --global url."https://github.com/".insteadOf "git@github.com:"
|
||||
git config --global url."https://github.com/".insteadOf "ssh://git@github.com/"
|
||||
# Register gh as git credential helper (provides GITHUB_TOKEN to git for HTTPS)
|
||||
gh auth setup-git 2>/dev/null || true
|
||||
# Ensure gh CLI itself uses HTTPS for gh repo clone
|
||||
gh config set git_protocol https --host github.com 2>/dev/null || true
|
||||
|
||||
# ---- Install pnpm globally ----
|
||||
npm install -g pnpm > /dev/null 2>&1
|
||||
|
||||
# ---- Git identity ----
|
||||
{{git.identity_setup}}
|
||||
|
||||
@@ -11,6 +11,7 @@ import (
|
||||
type ListeningPort struct {
|
||||
Port int `json:"port"`
|
||||
Address string `json:"address"`
|
||||
Process string `json:"process,omitempty"`
|
||||
}
|
||||
|
||||
// ListPorts returns all TCP ports currently listening inside the executor.
|
||||
|
||||
@@ -5,35 +5,65 @@ import (
|
||||
"context"
|
||||
"encoding/json"
|
||||
"fmt"
|
||||
"io"
|
||||
"net/http"
|
||||
"strings"
|
||||
"time"
|
||||
|
||||
"github.com/gorilla/websocket"
|
||||
)
|
||||
|
||||
// StartShellTerminal creates a new per-terminal shell session on agentctl.
|
||||
// Retries on transient connection errors (common with Sprites proxy tunnels).
|
||||
func (c *Client) StartShellTerminal(ctx context.Context, terminalID string, cols, rows int) error {
|
||||
body, _ := json.Marshal(map[string]any{
|
||||
payload, _ := json.Marshal(map[string]any{
|
||||
"terminal_id": terminalID,
|
||||
"cols": cols,
|
||||
"rows": rows,
|
||||
})
|
||||
|
||||
req, err := http.NewRequestWithContext(ctx, "POST", c.baseURL+"/api/v1/shell/terminal/start", bytes.NewReader(body))
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
req.Header.Set("Content-Type", "application/json")
|
||||
const maxAttempts = 3
|
||||
backoff := 200 * time.Millisecond
|
||||
var lastErr error
|
||||
|
||||
resp, err := c.httpClient.Do(req)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
defer func() { _ = resp.Body.Close() }()
|
||||
for attempt := 1; attempt <= maxAttempts; attempt++ {
|
||||
req, err := http.NewRequestWithContext(ctx, "POST", c.baseURL+"/api/v1/shell/terminal/start", bytes.NewReader(payload))
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
req.Header.Set("Content-Type", "application/json")
|
||||
|
||||
if resp.StatusCode != http.StatusOK {
|
||||
return fmt.Errorf("start shell terminal failed: %s", resp.Status)
|
||||
resp, err := c.httpClient.Do(req)
|
||||
if err != nil {
|
||||
lastErr = err
|
||||
if attempt < maxAttempts && isTransientConnError(err) {
|
||||
time.Sleep(backoff)
|
||||
backoff *= 2
|
||||
continue
|
||||
}
|
||||
return err
|
||||
}
|
||||
|
||||
status := resp.StatusCode
|
||||
_ = resp.Body.Close()
|
||||
|
||||
if status == http.StatusOK {
|
||||
return nil
|
||||
}
|
||||
return fmt.Errorf("start shell terminal failed: %d", status)
|
||||
}
|
||||
return nil
|
||||
return lastErr
|
||||
}
|
||||
|
||||
func isTransientConnError(err error) bool {
|
||||
if err == nil {
|
||||
return false
|
||||
}
|
||||
msg := err.Error()
|
||||
return strings.Contains(msg, "connection reset") ||
|
||||
strings.Contains(msg, "connection refused") ||
|
||||
err == io.EOF ||
|
||||
strings.Contains(msg, io.ErrUnexpectedEOF.Error())
|
||||
}
|
||||
|
||||
// StreamShellTerminal opens a binary WebSocket connection to a per-terminal shell.
|
||||
|
||||
@@ -9,6 +9,7 @@ import (
|
||||
"net/http"
|
||||
"os"
|
||||
"os/exec"
|
||||
"regexp"
|
||||
"runtime"
|
||||
"strconv"
|
||||
"strings"
|
||||
@@ -17,10 +18,13 @@ import (
|
||||
"github.com/gin-gonic/gin"
|
||||
)
|
||||
|
||||
const defaultListenAddr = "0.0.0.0"
|
||||
|
||||
// ListeningPort represents a TCP port with an active listener.
|
||||
type ListeningPort struct {
|
||||
Port int `json:"port"`
|
||||
Address string `json:"address"`
|
||||
Process string `json:"process,omitempty"`
|
||||
}
|
||||
|
||||
// handleListPorts returns all TCP ports currently listening inside the executor.
|
||||
@@ -35,33 +39,100 @@ func (s *Server) handleListPorts(c *gin.Context) {
|
||||
|
||||
func detectListeningPorts() ([]ListeningPort, error) {
|
||||
if runtime.GOOS == "linux" {
|
||||
ports, err := detectFromSS()
|
||||
if err == nil {
|
||||
return ports, nil
|
||||
}
|
||||
return detectFromProcNet()
|
||||
}
|
||||
return detectFromLsof()
|
||||
}
|
||||
|
||||
// detectFromProcNet parses /proc/net/tcp for LISTEN state (0A) sockets.
|
||||
func detectFromProcNet() ([]ListeningPort, error) {
|
||||
file, err := os.Open("/proc/net/tcp")
|
||||
// processNameRe extracts the first process name from ss users field.
|
||||
// Example: users:(("next-server (v1",pid=16186,fd=24)) → "next-server (v1"
|
||||
var processNameRe = regexp.MustCompile(`\(\("([^"]+)"`)
|
||||
|
||||
// detectFromSS uses ss -tlnp to detect listening ports with process names.
|
||||
// Covers both IPv4 and IPv6 sockets.
|
||||
func detectFromSS() ([]ListeningPort, error) {
|
||||
ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
|
||||
defer cancel()
|
||||
|
||||
out, err := exec.CommandContext(ctx, "ss", "-tlnpH").Output()
|
||||
if err != nil {
|
||||
return detectFromLsof() // fallback
|
||||
return nil, fmt.Errorf("ss failed: %w", err)
|
||||
}
|
||||
defer func() { _ = file.Close() }()
|
||||
|
||||
seen := make(map[int]bool)
|
||||
var ports []ListeningPort
|
||||
|
||||
scanner := bufio.NewScanner(file)
|
||||
scanner.Scan() // skip header
|
||||
for scanner.Scan() {
|
||||
port, addr, ok := parseProcNetLine(scanner.Text())
|
||||
if !ok || port < 1024 || seen[port] {
|
||||
for _, line := range strings.Split(string(out), "\n") {
|
||||
line = strings.TrimSpace(line)
|
||||
if line == "" {
|
||||
continue
|
||||
}
|
||||
|
||||
fields := strings.Fields(line)
|
||||
if len(fields) < 4 {
|
||||
continue
|
||||
}
|
||||
|
||||
// Field 3 is local address:port (e.g. "0.0.0.0:3000", "*:8080", "[::]:3000")
|
||||
localAddr := fields[3]
|
||||
host, portStr, splitErr := net.SplitHostPort(localAddr)
|
||||
if splitErr != nil {
|
||||
continue
|
||||
}
|
||||
port, convErr := strconv.Atoi(portStr)
|
||||
if convErr != nil || port < 1024 || seen[port] {
|
||||
continue
|
||||
}
|
||||
seen[port] = true
|
||||
ports = append(ports, ListeningPort{Port: port, Address: addr})
|
||||
|
||||
var process string
|
||||
// Process info is in the last field(s) — rejoin everything after peer address
|
||||
rest := strings.Join(fields[5:], " ")
|
||||
if match := processNameRe.FindStringSubmatch(rest); len(match) > 1 {
|
||||
process = match[1]
|
||||
}
|
||||
|
||||
if host == "*" || host == "[::]" {
|
||||
host = defaultListenAddr
|
||||
}
|
||||
|
||||
ports = append(ports, ListeningPort{Port: port, Address: host, Process: process})
|
||||
}
|
||||
return ports, scanner.Err()
|
||||
return ports, nil
|
||||
}
|
||||
|
||||
// detectFromProcNet parses /proc/net/tcp and /proc/net/tcp6 for LISTEN state (0A) sockets.
|
||||
func detectFromProcNet() ([]ListeningPort, error) {
|
||||
seen := make(map[int]bool)
|
||||
var ports []ListeningPort
|
||||
|
||||
for _, path := range []string{"/proc/net/tcp", "/proc/net/tcp6"} {
|
||||
file, err := os.Open(path)
|
||||
if err != nil {
|
||||
continue
|
||||
}
|
||||
|
||||
scanner := bufio.NewScanner(file)
|
||||
scanner.Scan() // skip header
|
||||
for scanner.Scan() {
|
||||
port, addr, ok := parseProcNetLine(scanner.Text())
|
||||
if !ok || port < 1024 || seen[port] {
|
||||
continue
|
||||
}
|
||||
seen[port] = true
|
||||
ports = append(ports, ListeningPort{Port: port, Address: addr})
|
||||
}
|
||||
_ = file.Close()
|
||||
}
|
||||
|
||||
if len(ports) == 0 {
|
||||
return detectFromLsof()
|
||||
}
|
||||
return ports, nil
|
||||
}
|
||||
|
||||
// parseProcNetLine extracts port and address from a /proc/net/tcp line.
|
||||
@@ -94,42 +165,51 @@ func parseProcNetLine(line string) (int, string, bool) {
|
||||
|
||||
func parseHexIP(hexIP string) string {
|
||||
b, err := hex.DecodeString(hexIP)
|
||||
if err != nil || len(b) != 4 {
|
||||
return "0.0.0.0"
|
||||
if err != nil || len(b) < 4 {
|
||||
return defaultListenAddr
|
||||
}
|
||||
// /proc/net/tcp stores in little-endian
|
||||
return fmt.Sprintf("%d.%d.%d.%d", b[3], b[2], b[1], b[0])
|
||||
if len(b) == 4 {
|
||||
// IPv4: /proc/net/tcp stores in little-endian
|
||||
return fmt.Sprintf("%d.%d.%d.%d", b[3], b[2], b[1], b[0])
|
||||
}
|
||||
// IPv6: treat all addresses as all-interfaces for port detection purposes.
|
||||
return defaultListenAddr
|
||||
}
|
||||
|
||||
// detectFromLsof uses lsof as a fallback for non-Linux systems.
|
||||
func detectFromLsof() ([]ListeningPort, error) {
|
||||
ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
|
||||
defer cancel()
|
||||
out, err := exec.CommandContext(ctx, "lsof", "-iTCP", "-sTCP:LISTEN", "-nP", "-F", "n").Output()
|
||||
out, err := exec.CommandContext(ctx, "lsof", "-iTCP", "-sTCP:LISTEN", "-nP", "-F", "nc").Output()
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("lsof failed: %w", err)
|
||||
}
|
||||
|
||||
seen := make(map[int]bool)
|
||||
var ports []ListeningPort
|
||||
var currentProcess string
|
||||
|
||||
for _, line := range strings.Split(string(out), "\n") {
|
||||
if !strings.HasPrefix(line, "n") {
|
||||
if len(line) == 0 {
|
||||
continue
|
||||
}
|
||||
// Format: n*:PORT or n127.0.0.1:PORT or n[::]:PORT
|
||||
addr := line[1:] // strip 'n' prefix
|
||||
_, portStr, err := net.SplitHostPort(addr)
|
||||
if err != nil {
|
||||
continue
|
||||
switch line[0] {
|
||||
case 'c':
|
||||
currentProcess = line[1:]
|
||||
case 'n':
|
||||
// Format: n*:PORT or n127.0.0.1:PORT or n[::]:PORT
|
||||
addr := line[1:]
|
||||
host, portStr, splitErr := net.SplitHostPort(addr)
|
||||
if splitErr != nil {
|
||||
continue
|
||||
}
|
||||
port, convErr := strconv.Atoi(portStr)
|
||||
if convErr != nil || port < 1024 || seen[port] {
|
||||
continue
|
||||
}
|
||||
seen[port] = true
|
||||
ports = append(ports, ListeningPort{Port: port, Address: host, Process: currentProcess})
|
||||
}
|
||||
port, err := strconv.Atoi(portStr)
|
||||
if err != nil || port < 1024 || seen[port] {
|
||||
continue
|
||||
}
|
||||
seen[port] = true
|
||||
host, _, _ := net.SplitHostPort(addr)
|
||||
ports = append(ports, ListeningPort{Port: port, Address: host})
|
||||
}
|
||||
return ports, nil
|
||||
}
|
||||
|
||||
@@ -89,7 +89,8 @@ func (s *Server) createPortProxy(port int) *httputil.ReverseProxy {
|
||||
r.SetURL(target)
|
||||
r.Out.URL.Path = r.In.URL.Path
|
||||
r.Out.URL.RawPath = ""
|
||||
r.Out.Header.Set("Origin", target.String())
|
||||
// Preserve original Host header for CORS/Origin validation in tunneled apps.
|
||||
r.Out.Host = r.In.Host
|
||||
if r.Out.Header.Get("Upgrade") != "" {
|
||||
r.Out.Header.Set("Connection", "Upgrade")
|
||||
}
|
||||
|
||||
@@ -0,0 +1,313 @@
|
||||
package websocket
|
||||
|
||||
import (
|
||||
"context"
|
||||
"errors"
|
||||
"fmt"
|
||||
"net"
|
||||
"net/http"
|
||||
"net/http/httputil"
|
||||
"net/url"
|
||||
"os"
|
||||
"strconv"
|
||||
"strings"
|
||||
"sync"
|
||||
"syscall"
|
||||
|
||||
"go.uber.org/zap"
|
||||
|
||||
"github.com/kandev/kandev/internal/agent/lifecycle"
|
||||
"github.com/kandev/kandev/internal/common/logger"
|
||||
)
|
||||
|
||||
// TunnelInfo describes an active tunnel binding.
|
||||
type TunnelInfo struct {
|
||||
Port int `json:"port"`
|
||||
TunnelPort int `json:"tunnel_port"`
|
||||
}
|
||||
|
||||
// tunnelEntry tracks a running tunnel HTTP server.
|
||||
type tunnelEntry struct {
|
||||
port int
|
||||
tunnelPort int
|
||||
server *http.Server
|
||||
ln net.Listener
|
||||
cancel context.CancelFunc
|
||||
}
|
||||
|
||||
// TunnelManager manages per-session port tunnels that bind dedicated host
|
||||
// ports and reverse-proxy traffic to agentctl's port-proxy endpoint.
|
||||
// Unlike the path-based PortProxyHandler, tunnels serve apps at the root
|
||||
// path so web apps that expect "/" work correctly.
|
||||
type TunnelManager struct {
|
||||
lifecycleMgr *lifecycle.Manager
|
||||
logger *logger.Logger
|
||||
|
||||
mu sync.Mutex
|
||||
tunnels map[string]*tunnelEntry // key: "sessionId:port"
|
||||
}
|
||||
|
||||
// NewTunnelManager creates a new TunnelManager.
|
||||
func NewTunnelManager(lifecycleMgr *lifecycle.Manager, log *logger.Logger) *TunnelManager {
|
||||
return &TunnelManager{
|
||||
lifecycleMgr: lifecycleMgr,
|
||||
logger: log.WithFields(zap.String("component", "port-tunnel")),
|
||||
tunnels: make(map[string]*tunnelEntry),
|
||||
}
|
||||
}
|
||||
|
||||
// StartTunnel creates a tunnel for the given session and port.
|
||||
// If tunnelPort > 0, it binds to that specific port; if 0, the OS picks one.
|
||||
// Returns the actual tunnel port.
|
||||
func (m *TunnelManager) StartTunnel(sessionID string, port int, tunnelPort int) (int, error) {
|
||||
cacheKey := sessionID + ":" + strconv.Itoa(port)
|
||||
|
||||
// Check for existing tunnel under lock to prevent duplicate creation.
|
||||
m.mu.Lock()
|
||||
if entry, ok := m.tunnels[cacheKey]; ok {
|
||||
tp := entry.tunnelPort
|
||||
m.mu.Unlock()
|
||||
return tp, nil
|
||||
}
|
||||
// Reserve the key with a nil entry to prevent concurrent creation.
|
||||
m.tunnels[cacheKey] = nil
|
||||
m.mu.Unlock()
|
||||
|
||||
target, ln, err := m.resolveAndBind(cacheKey, sessionID, tunnelPort)
|
||||
if err != nil {
|
||||
return 0, err
|
||||
}
|
||||
actualPort := ln.Addr().(*net.TCPAddr).Port
|
||||
|
||||
proxy := m.createTunnelProxy(cacheKey, target, port)
|
||||
ctx, cancel := context.WithCancel(context.Background())
|
||||
srv := &http.Server{Handler: proxy}
|
||||
|
||||
entry := &tunnelEntry{
|
||||
port: port,
|
||||
tunnelPort: actualPort,
|
||||
server: srv,
|
||||
ln: ln,
|
||||
cancel: cancel,
|
||||
}
|
||||
|
||||
m.mu.Lock()
|
||||
m.tunnels[cacheKey] = entry
|
||||
m.mu.Unlock()
|
||||
|
||||
m.serveTunnel(ctx, srv, ln, cacheKey, sessionID, port, actualPort)
|
||||
|
||||
m.logger.Info("tunnel started",
|
||||
zap.String("session_id", sessionID),
|
||||
zap.Int("port", port),
|
||||
zap.Int("tunnel_port", actualPort))
|
||||
|
||||
return actualPort, nil
|
||||
}
|
||||
|
||||
// resolveAndBind resolves the agentctl target URL for the session and binds a
|
||||
// local TCP listener for the tunnel. On failure it cleans up the placeholder
|
||||
// entry reserved by StartTunnel.
|
||||
func (m *TunnelManager) resolveAndBind(cacheKey, sessionID string, tunnelPort int) (*url.URL, net.Listener, error) {
|
||||
cleanup := func() {
|
||||
m.mu.Lock()
|
||||
if m.tunnels[cacheKey] == nil {
|
||||
delete(m.tunnels, cacheKey)
|
||||
}
|
||||
m.mu.Unlock()
|
||||
}
|
||||
|
||||
execution, ok := m.lifecycleMgr.GetExecutionBySessionID(sessionID)
|
||||
if !ok {
|
||||
cleanup()
|
||||
return nil, nil, fmt.Errorf("session not found or no active execution")
|
||||
}
|
||||
|
||||
agentctlClient := execution.GetAgentCtlClient()
|
||||
if agentctlClient == nil {
|
||||
cleanup()
|
||||
return nil, nil, fmt.Errorf("agentctl client not available")
|
||||
}
|
||||
|
||||
target, err := url.Parse(agentctlClient.BaseURL())
|
||||
if err != nil {
|
||||
cleanup()
|
||||
return nil, nil, fmt.Errorf("failed to parse agentctl URL: %w", err)
|
||||
}
|
||||
|
||||
ln, err := net.Listen("tcp", fmt.Sprintf(":%d", tunnelPort))
|
||||
if err != nil {
|
||||
cleanup()
|
||||
if isAddrInUse(err) {
|
||||
return nil, nil, fmt.Errorf("port %d is already in use, choose a different port", tunnelPort)
|
||||
}
|
||||
return nil, nil, fmt.Errorf("failed to bind tunnel port %d: %w", tunnelPort, err)
|
||||
}
|
||||
|
||||
return target, ln, nil
|
||||
}
|
||||
|
||||
// serveTunnel starts the tunnel HTTP server and its shutdown goroutine.
|
||||
func (m *TunnelManager) serveTunnel(ctx context.Context, srv *http.Server, ln net.Listener, cacheKey, sessionID string, port, tunnelPort int) {
|
||||
go func() {
|
||||
<-ctx.Done()
|
||||
_ = srv.Close()
|
||||
}()
|
||||
|
||||
go func() {
|
||||
if serveErr := srv.Serve(ln); serveErr != nil && serveErr != http.ErrServerClosed {
|
||||
m.logger.Error("tunnel server error",
|
||||
zap.String("session_id", sessionID),
|
||||
zap.Int("port", port),
|
||||
zap.Int("tunnel_port", tunnelPort),
|
||||
zap.Error(serveErr))
|
||||
m.removeTunnel(cacheKey)
|
||||
}
|
||||
}()
|
||||
}
|
||||
|
||||
// StopTunnel stops a tunnel for the given session and port.
|
||||
func (m *TunnelManager) StopTunnel(sessionID string, port int) error {
|
||||
cacheKey := sessionID + ":" + strconv.Itoa(port)
|
||||
|
||||
m.mu.Lock()
|
||||
entry, ok := m.tunnels[cacheKey]
|
||||
if !ok {
|
||||
m.mu.Unlock()
|
||||
return fmt.Errorf("no tunnel found for session %s port %d", sessionID, port)
|
||||
}
|
||||
delete(m.tunnels, cacheKey)
|
||||
m.mu.Unlock()
|
||||
|
||||
entry.cancel()
|
||||
_ = entry.ln.Close()
|
||||
|
||||
m.logger.Info("tunnel stopped",
|
||||
zap.String("session_id", sessionID),
|
||||
zap.Int("port", port),
|
||||
zap.Int("tunnel_port", entry.tunnelPort))
|
||||
|
||||
return nil
|
||||
}
|
||||
|
||||
// ListTunnels returns all active tunnels for a session.
|
||||
// Returns []TunnelInfo (satisfies TunnelController interface).
|
||||
func (m *TunnelManager) ListTunnels(sessionID string) any {
|
||||
m.mu.Lock()
|
||||
defer m.mu.Unlock()
|
||||
|
||||
prefix := sessionID + ":"
|
||||
tunnels := make([]TunnelInfo, 0)
|
||||
for key, entry := range m.tunnels {
|
||||
if strings.HasPrefix(key, prefix) {
|
||||
tunnels = append(tunnels, TunnelInfo{
|
||||
Port: entry.port,
|
||||
TunnelPort: entry.tunnelPort,
|
||||
})
|
||||
}
|
||||
}
|
||||
return tunnels
|
||||
}
|
||||
|
||||
// InvalidateSession stops all tunnels for a session.
|
||||
func (m *TunnelManager) InvalidateSession(sessionID string) {
|
||||
m.mu.Lock()
|
||||
prefix := sessionID + ":"
|
||||
var toStop []*tunnelEntry
|
||||
for key, entry := range m.tunnels {
|
||||
if strings.HasPrefix(key, prefix) {
|
||||
toStop = append(toStop, entry)
|
||||
delete(m.tunnels, key)
|
||||
}
|
||||
}
|
||||
m.mu.Unlock()
|
||||
|
||||
for _, entry := range toStop {
|
||||
entry.cancel()
|
||||
_ = entry.ln.Close()
|
||||
}
|
||||
|
||||
if len(toStop) > 0 {
|
||||
m.logger.Info("invalidated session tunnels",
|
||||
zap.String("session_id", sessionID),
|
||||
zap.Int("count", len(toStop)))
|
||||
}
|
||||
}
|
||||
|
||||
// Shutdown stops all active tunnels.
|
||||
func (m *TunnelManager) Shutdown() {
|
||||
m.mu.Lock()
|
||||
entries := make([]*tunnelEntry, 0, len(m.tunnels))
|
||||
for _, entry := range m.tunnels {
|
||||
entries = append(entries, entry)
|
||||
}
|
||||
m.tunnels = make(map[string]*tunnelEntry)
|
||||
m.mu.Unlock()
|
||||
|
||||
for _, entry := range entries {
|
||||
entry.cancel()
|
||||
_ = entry.ln.Close()
|
||||
}
|
||||
|
||||
if len(entries) > 0 {
|
||||
m.logger.Info("shutdown all tunnels", zap.Int("count", len(entries)))
|
||||
}
|
||||
}
|
||||
|
||||
func isAddrInUse(err error) bool {
|
||||
var opErr *net.OpError
|
||||
if errors.As(err, &opErr) {
|
||||
var sysErr *os.SyscallError
|
||||
if errors.As(opErr.Err, &sysErr) {
|
||||
return errors.Is(sysErr.Err, syscall.EADDRINUSE)
|
||||
}
|
||||
}
|
||||
return false
|
||||
}
|
||||
|
||||
func (m *TunnelManager) removeTunnel(cacheKey string) {
|
||||
m.mu.Lock()
|
||||
defer m.mu.Unlock()
|
||||
delete(m.tunnels, cacheKey)
|
||||
}
|
||||
|
||||
func (m *TunnelManager) createTunnelProxy(cacheKey string, target *url.URL, port int) *httputil.ReverseProxy {
|
||||
portStr := strconv.Itoa(port)
|
||||
|
||||
proxy := &httputil.ReverseProxy{}
|
||||
proxy.Rewrite = func(r *httputil.ProxyRequest) {
|
||||
r.SetURL(target)
|
||||
// Rewrite: /{path} → /api/v1/port-proxy/{port}/{path}
|
||||
incoming := r.In.URL.Path
|
||||
if incoming == "" {
|
||||
incoming = "/"
|
||||
}
|
||||
r.Out.URL.Path = "/api/v1/port-proxy/" + portStr + incoming
|
||||
r.Out.URL.RawPath = ""
|
||||
// Preserve original Host header for CORS/Origin validation.
|
||||
r.Out.Host = r.In.Host
|
||||
if r.Out.Header.Get("Upgrade") != "" {
|
||||
r.Out.Header.Set("Connection", "Upgrade")
|
||||
}
|
||||
}
|
||||
|
||||
proxy.ModifyResponse = func(resp *http.Response) error {
|
||||
if resp.StatusCode == http.StatusSwitchingProtocols {
|
||||
resp.Header.Set("Connection", "Upgrade")
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
proxy.ErrorHandler = func(w http.ResponseWriter, r *http.Request, err error) {
|
||||
m.logger.Error("tunnel proxy error",
|
||||
zap.String("cache_key", cacheKey),
|
||||
zap.String("request_path", r.URL.Path),
|
||||
zap.Error(err))
|
||||
http.Error(w, "tunnel proxy error", http.StatusBadGateway)
|
||||
}
|
||||
|
||||
// Flush immediately for SSE/streaming responses.
|
||||
proxy.FlushInterval = -1
|
||||
|
||||
return proxy
|
||||
}
|
||||
@@ -19,6 +19,7 @@ type Gateway struct {
|
||||
LSPHandler *LSPHandler
|
||||
VscodeProxyHandler *VscodeProxyHandler
|
||||
PortProxyHandler *PortProxyHandler
|
||||
TunnelManager *TunnelManager
|
||||
logger *logger.Logger
|
||||
}
|
||||
|
||||
@@ -60,6 +61,11 @@ func (g *Gateway) SetPortProxy(lifecycleMgr *lifecycle.Manager) {
|
||||
g.PortProxyHandler = NewPortProxyHandler(lifecycleMgr, g.logger)
|
||||
}
|
||||
|
||||
// SetPortTunnel enables dedicated port tunnel management.
|
||||
func (g *Gateway) SetPortTunnel(lifecycleMgr *lifecycle.Manager) {
|
||||
g.TunnelManager = NewTunnelManager(lifecycleMgr, g.logger)
|
||||
}
|
||||
|
||||
// SetupRoutes adds the WebSocket routes to the Gin engine
|
||||
func (g *Gateway) SetupRoutes(router *gin.Engine) {
|
||||
router.GET("/ws", g.Handler.HandleConnection)
|
||||
|
||||
@@ -1,8 +1,11 @@
|
||||
package repoclone
|
||||
|
||||
import (
|
||||
"context"
|
||||
"fmt"
|
||||
"os/exec"
|
||||
"strings"
|
||||
"time"
|
||||
)
|
||||
|
||||
// ProtocolSSH is the SSH git protocol.
|
||||
@@ -12,9 +15,17 @@ const ProtocolSSH = "ssh"
|
||||
const ProtocolHTTPS = "https"
|
||||
|
||||
// DetectGitProtocol returns the user's preferred git clone protocol.
|
||||
// Currently returns "ssh" unconditionally. In the future this may be
|
||||
// configurable or auto-detected from gh CLI / SSH config.
|
||||
// It checks the gh CLI config (`gh config get git_protocol`). If gh reports
|
||||
// "https", it returns ProtocolHTTPS. Otherwise it defaults to ProtocolSSH.
|
||||
func DetectGitProtocol() string {
|
||||
ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
|
||||
defer cancel()
|
||||
out, err := exec.CommandContext(ctx, "gh", "config", "get", "git_protocol").Output()
|
||||
if err == nil {
|
||||
if strings.TrimSpace(string(out)) == ProtocolHTTPS {
|
||||
return ProtocolHTTPS
|
||||
}
|
||||
}
|
||||
return ProtocolSSH
|
||||
}
|
||||
|
||||
|
||||
@@ -98,7 +98,7 @@ func TestProviderHost(t *testing.T) {
|
||||
|
||||
func TestDetectGitProtocol(t *testing.T) {
|
||||
got := DetectGitProtocol()
|
||||
if got != ProtocolSSH {
|
||||
t.Errorf("expected %q, got %q", ProtocolSSH, got)
|
||||
if got != ProtocolSSH && got != ProtocolHTTPS {
|
||||
t.Errorf("expected %q or %q, got %q", ProtocolSSH, ProtocolHTTPS, got)
|
||||
}
|
||||
}
|
||||
|
||||
@@ -261,7 +261,10 @@ const (
|
||||
ActionVscodeOpenFile = "vscode.openFile" // Open a file in code-server for a session
|
||||
|
||||
// Port actions
|
||||
ActionPortList = "port.list" // List listening ports on a remote executor
|
||||
ActionPortList = "port.list" // List listening ports on a remote executor
|
||||
ActionPortTunnelStart = "port.tunnel.start" // Start a port tunnel
|
||||
ActionPortTunnelStop = "port.tunnel.stop" // Stop a port tunnel
|
||||
ActionPortTunnelList = "port.tunnel.list" // List active port tunnels
|
||||
|
||||
// Secret actions
|
||||
ActionSecretList = "secrets.list"
|
||||
|
||||
+12
-4
@@ -44,8 +44,8 @@ Options:
|
||||
run Use release bundles (default).
|
||||
--dev Alias for "dev".
|
||||
--version Release tag to install (default: latest).
|
||||
--backend-port Override backend port.
|
||||
--web-port Override web port.
|
||||
--backend-port Override backend port (or KANDEV_BACKEND_PORT env var).
|
||||
--web-port Override web port (or KANDEV_WEB_PORT env var).
|
||||
--verbose, -v Show info logs from backend + web (start mode only).
|
||||
--debug Show debug logs + agent message dumps (start mode only).
|
||||
--help, -h Show help.
|
||||
@@ -107,6 +107,11 @@ function parseArgs(argv: string[]): CliOptions {
|
||||
return opts;
|
||||
}
|
||||
|
||||
function envPort(name: string): number | undefined {
|
||||
const val = process.env[name];
|
||||
return val ? Number(val) : undefined;
|
||||
}
|
||||
|
||||
function findRepoRoot(startDir: string): string | null {
|
||||
let current = path.resolve(startDir);
|
||||
while (true) {
|
||||
@@ -132,8 +137,11 @@ function findRepoRoot(startDir: string): string | null {
|
||||
|
||||
async function main(): Promise<void> {
|
||||
const raw = parseArgs(process.argv.slice(2));
|
||||
const backendPort = ensureValidPort(raw.backendPort, "backend port");
|
||||
const webPort = ensureValidPort(raw.webPort, "web port");
|
||||
const backendPort = ensureValidPort(
|
||||
raw.backendPort ?? envPort("KANDEV_BACKEND_PORT"),
|
||||
"backend port",
|
||||
);
|
||||
const webPort = ensureValidPort(raw.webPort ?? envPort("KANDEV_WEB_PORT"), "web port");
|
||||
|
||||
if (raw.command === "dev") {
|
||||
const repoRoot = findRepoRoot(process.cwd());
|
||||
|
||||
@@ -14,7 +14,7 @@ export const RANDOM_PORT_RETRIES = 10;
|
||||
|
||||
// Backend healthcheck timeout during startup.
|
||||
export const HEALTH_TIMEOUT_MS_RELEASE = 15000;
|
||||
export const HEALTH_TIMEOUT_MS_DEV = 60000;
|
||||
export const HEALTH_TIMEOUT_MS_DEV = 600000;
|
||||
|
||||
// Local user cache/data directories for release bundles and DB.
|
||||
export const CACHE_DIR = path.join(os.homedir(), ".kandev", "bin");
|
||||
|
||||
@@ -1,6 +1,6 @@
|
||||
"use client";
|
||||
|
||||
import { useState, useCallback } from "react";
|
||||
import { useState, useCallback, useEffect, useRef } from "react";
|
||||
import {
|
||||
IconNetwork,
|
||||
IconExternalLink,
|
||||
@@ -9,13 +9,17 @@ import {
|
||||
IconRefresh,
|
||||
IconPlus,
|
||||
IconLoader2,
|
||||
IconPlugConnected,
|
||||
IconPlugConnectedX,
|
||||
IconInfoCircle,
|
||||
} from "@tabler/icons-react";
|
||||
import { Button } from "@kandev/ui/button";
|
||||
import { Input } from "@kandev/ui/input";
|
||||
import { Badge } from "@kandev/ui/badge";
|
||||
import { Dialog, DialogContent, DialogHeader, DialogTitle, DialogTrigger } from "@kandev/ui/dialog";
|
||||
import { Tooltip, TooltipContent, TooltipTrigger } from "@kandev/ui/tooltip";
|
||||
import { listPorts, type ListeningPort } from "@/lib/api/domains/port-api";
|
||||
import { listPorts, listTunnels, type ListeningPort } from "@/lib/api/domains/port-api";
|
||||
import { useTunnelActions } from "./use-tunnel-actions";
|
||||
import { getBackendConfig } from "@/lib/config";
|
||||
import { toast } from "sonner";
|
||||
|
||||
@@ -24,76 +28,245 @@ function buildPortProxyUrl(sessionId: string, port: number): string {
|
||||
return `${backendUrl}/port-proxy/${sessionId}/${port}/`;
|
||||
}
|
||||
|
||||
function CopyUrlButton({ url }: { url: string }) {
|
||||
const [copied, setCopied] = useState(false);
|
||||
function buildTunnelUrl(tunnelPort: number): string {
|
||||
const backendUrl = getBackendConfig().apiBaseUrl;
|
||||
const { protocol, hostname } = new URL(backendUrl);
|
||||
return `${protocol}//${hostname}:${tunnelPort}/`;
|
||||
}
|
||||
|
||||
function InfoTip({ text }: { text: string }) {
|
||||
return (
|
||||
<Tooltip>
|
||||
<TooltipTrigger asChild>
|
||||
<IconInfoCircle className="h-3.5 w-3.5 text-muted-foreground/60 shrink-0 cursor-help" />
|
||||
</TooltipTrigger>
|
||||
<TooltipContent side="top" className="max-w-[240px] text-xs">
|
||||
{text}
|
||||
</TooltipContent>
|
||||
</Tooltip>
|
||||
);
|
||||
}
|
||||
|
||||
function UrlActions({ url }: { url: string }) {
|
||||
const [copied, setCopied] = useState(false);
|
||||
const handleCopy = useCallback(() => {
|
||||
navigator.clipboard?.writeText(url);
|
||||
setCopied(true);
|
||||
setTimeout(() => setCopied(false), 1500);
|
||||
}, [url]);
|
||||
|
||||
return (
|
||||
<div className="flex items-center gap-0.5">
|
||||
<Tooltip>
|
||||
<TooltipTrigger asChild>
|
||||
<Button
|
||||
size="sm"
|
||||
variant="ghost"
|
||||
className="cursor-pointer h-7 w-7 p-0"
|
||||
onClick={handleCopy}
|
||||
>
|
||||
{copied ? (
|
||||
<IconCheck className="h-3.5 w-3.5 text-green-500" />
|
||||
) : (
|
||||
<IconCopy className="h-3.5 w-3.5" />
|
||||
)}
|
||||
</Button>
|
||||
</TooltipTrigger>
|
||||
<TooltipContent>Copy URL</TooltipContent>
|
||||
</Tooltip>
|
||||
<Tooltip>
|
||||
<TooltipTrigger asChild>
|
||||
<Button size="sm" variant="ghost" className="cursor-pointer h-7 w-7 p-0" asChild>
|
||||
<a href={url} target="_blank" rel="noopener noreferrer">
|
||||
<IconExternalLink className="h-3.5 w-3.5" />
|
||||
</a>
|
||||
</Button>
|
||||
</TooltipTrigger>
|
||||
<TooltipContent>Open in new tab</TooltipContent>
|
||||
</Tooltip>
|
||||
</div>
|
||||
);
|
||||
}
|
||||
|
||||
function TunnelToggleButton({
|
||||
isTunnelActive,
|
||||
tunnelPending,
|
||||
onStop,
|
||||
onToggleForm,
|
||||
}: {
|
||||
isTunnelActive: boolean;
|
||||
tunnelPending?: boolean;
|
||||
onStop: () => void;
|
||||
onToggleForm: () => void;
|
||||
}) {
|
||||
const Icon = isTunnelActive ? IconPlugConnectedX : IconPlugConnected;
|
||||
return (
|
||||
<Tooltip>
|
||||
<TooltipTrigger asChild>
|
||||
<Button
|
||||
size="sm"
|
||||
variant="ghost"
|
||||
className="cursor-pointer h-7 w-7 p-0"
|
||||
onClick={handleCopy}
|
||||
className={`cursor-pointer h-7 w-7 p-0 ${isTunnelActive ? "text-destructive hover:text-destructive" : ""}`}
|
||||
onClick={isTunnelActive ? onStop : onToggleForm}
|
||||
disabled={tunnelPending}
|
||||
>
|
||||
{copied ? (
|
||||
<IconCheck className="h-3.5 w-3.5 text-green-500" />
|
||||
{tunnelPending ? (
|
||||
<IconLoader2 className="h-3.5 w-3.5 animate-spin" />
|
||||
) : (
|
||||
<IconCopy className="h-3.5 w-3.5" />
|
||||
<Icon className="h-3.5 w-3.5" />
|
||||
)}
|
||||
</Button>
|
||||
</TooltipTrigger>
|
||||
<TooltipContent>Copy URL</TooltipContent>
|
||||
<TooltipContent>{isTunnelActive ? "Stop tunnel" : "Start tunnel"}</TooltipContent>
|
||||
</Tooltip>
|
||||
);
|
||||
}
|
||||
|
||||
function PortUrlRow({
|
||||
label,
|
||||
tip,
|
||||
url,
|
||||
variant = "outline",
|
||||
}: {
|
||||
label: string;
|
||||
tip: string;
|
||||
url: string;
|
||||
variant?: "outline" | "default";
|
||||
}) {
|
||||
return (
|
||||
<div className="flex items-center gap-2 min-w-0">
|
||||
<Badge variant={variant} className="text-[10px] px-1.5 py-0 shrink-0">
|
||||
{label}
|
||||
</Badge>
|
||||
<InfoTip text={tip} />
|
||||
<span className="text-xs text-muted-foreground truncate min-w-0 flex-1">{url}</span>
|
||||
<div className="shrink-0">
|
||||
<UrlActions url={url} />
|
||||
</div>
|
||||
</div>
|
||||
);
|
||||
}
|
||||
|
||||
function PortUrlRows({ proxyUrl, tunnelUrl }: { proxyUrl: string; tunnelUrl: string | null }) {
|
||||
return (
|
||||
<div className="space-y-1 overflow-hidden">
|
||||
<PortUrlRow
|
||||
label="Proxy"
|
||||
tip="Path-based proxy. Works for APIs but may break web apps that expect to be served at /."
|
||||
url={proxyUrl}
|
||||
/>
|
||||
{tunnelUrl && (
|
||||
<PortUrlRow
|
||||
label="Tunnel"
|
||||
tip="Dedicated port tunnel. App is served at /, so assets and routing work correctly."
|
||||
url={tunnelUrl}
|
||||
variant="default"
|
||||
/>
|
||||
)}
|
||||
</div>
|
||||
);
|
||||
}
|
||||
|
||||
type PortRowProps = {
|
||||
port: number;
|
||||
address?: string;
|
||||
process?: string;
|
||||
sessionId: string;
|
||||
badge: "Detected" | "Manual";
|
||||
tunnelPort?: number;
|
||||
tunnelPending?: boolean;
|
||||
onTunnelStart: (port: number, requestedPort?: number) => void;
|
||||
onTunnelStop: (port: number) => void;
|
||||
};
|
||||
|
||||
function PortRow({ port, address, sessionId, badge }: PortRowProps) {
|
||||
function PortRow({
|
||||
port,
|
||||
address,
|
||||
process,
|
||||
sessionId,
|
||||
badge,
|
||||
tunnelPort,
|
||||
tunnelPending,
|
||||
onTunnelStart,
|
||||
onTunnelStop,
|
||||
}: PortRowProps) {
|
||||
const [showTunnelForm, setShowTunnelForm] = useState(false);
|
||||
const [tunnelPortInput, setTunnelPortInput] = useState("");
|
||||
const proxyUrl = buildPortProxyUrl(sessionId, port);
|
||||
const tunnelUrl = tunnelPort ? buildTunnelUrl(tunnelPort) : null;
|
||||
const isTunnelActive = !!tunnelPort;
|
||||
|
||||
const handleStartTunnel = useCallback(() => {
|
||||
const requestedPort = tunnelPortInput ? parseInt(tunnelPortInput, 10) : undefined;
|
||||
if (
|
||||
tunnelPortInput &&
|
||||
(isNaN(requestedPort!) || requestedPort! < 1 || requestedPort! > 65535)
|
||||
) {
|
||||
toast.error("Enter a valid port (1-65535) or leave blank for random");
|
||||
return;
|
||||
}
|
||||
onTunnelStart(port, requestedPort);
|
||||
setShowTunnelForm(false);
|
||||
setTunnelPortInput("");
|
||||
}, [port, tunnelPortInput, onTunnelStart]);
|
||||
|
||||
return (
|
||||
<div
|
||||
data-testid={`port-forward-row-${port}`}
|
||||
className="flex items-center justify-between gap-2 px-3 py-2 rounded-md bg-muted/40 hover:bg-muted/60 transition-colors"
|
||||
className="rounded-md bg-muted/40 hover:bg-muted/60 transition-colors px-3 py-2 space-y-1.5"
|
||||
>
|
||||
<div className="flex items-center gap-2 min-w-0">
|
||||
<span className="text-sm font-mono font-medium">{port}</span>
|
||||
{address && address !== "0.0.0.0" && address !== "*" && (
|
||||
<span className="text-xs text-muted-foreground">{address}</span>
|
||||
)}
|
||||
<Badge
|
||||
variant={badge === "Detected" ? "secondary" : "outline"}
|
||||
className="text-[10px] px-1.5 py-0"
|
||||
>
|
||||
{badge}
|
||||
</Badge>
|
||||
</div>
|
||||
<div className="flex items-center gap-0.5">
|
||||
<CopyUrlButton url={proxyUrl} />
|
||||
<Tooltip>
|
||||
<TooltipTrigger asChild>
|
||||
<Button size="sm" variant="ghost" className="cursor-pointer h-7 w-7 p-0" asChild>
|
||||
<a href={proxyUrl} target="_blank" rel="noopener noreferrer">
|
||||
<IconExternalLink className="h-3.5 w-3.5" />
|
||||
</a>
|
||||
</Button>
|
||||
</TooltipTrigger>
|
||||
<TooltipContent>Open in new tab</TooltipContent>
|
||||
</Tooltip>
|
||||
<div className="flex items-center justify-between gap-2">
|
||||
<div className="flex items-center gap-2 min-w-0">
|
||||
<span className="text-sm font-mono font-medium">{port}</span>
|
||||
{process && (
|
||||
<span className="text-xs text-muted-foreground truncate max-w-[120px]">{process}</span>
|
||||
)}
|
||||
{address && address !== "0.0.0.0" && address !== "*" && (
|
||||
<span className="text-xs text-muted-foreground">{address}</span>
|
||||
)}
|
||||
<Badge
|
||||
variant={badge === "Detected" ? "secondary" : "outline"}
|
||||
className="text-[10px] px-1.5 py-0"
|
||||
>
|
||||
{badge}
|
||||
</Badge>
|
||||
</div>
|
||||
<div className="flex items-center gap-0.5">
|
||||
<TunnelToggleButton
|
||||
isTunnelActive={isTunnelActive}
|
||||
tunnelPending={tunnelPending}
|
||||
onStop={() => onTunnelStop(port)}
|
||||
onToggleForm={() => setShowTunnelForm((v) => !v)}
|
||||
/>
|
||||
</div>
|
||||
</div>
|
||||
|
||||
{showTunnelForm && !isTunnelActive && (
|
||||
<div className="flex items-center gap-2">
|
||||
<Input
|
||||
type="number"
|
||||
placeholder="Random"
|
||||
value={tunnelPortInput}
|
||||
onChange={(e) => setTunnelPortInput(e.target.value)}
|
||||
onKeyDown={(e) => e.key === "Enter" && (e.preventDefault(), handleStartTunnel())}
|
||||
className="h-7 text-xs w-24"
|
||||
min={1}
|
||||
max={65535}
|
||||
/>
|
||||
<Button
|
||||
size="sm"
|
||||
variant="outline"
|
||||
className="cursor-pointer h-7 text-xs gap-1"
|
||||
onClick={handleStartTunnel}
|
||||
disabled={tunnelPending}
|
||||
>
|
||||
Start
|
||||
</Button>
|
||||
<InfoTip text="Specify a local port or leave blank for a random one. For Docker/K8s, use a port you've pre-exposed." />
|
||||
</div>
|
||||
)}
|
||||
|
||||
<PortUrlRows proxyUrl={proxyUrl} tunnelUrl={tunnelUrl} />
|
||||
</div>
|
||||
);
|
||||
}
|
||||
@@ -105,6 +278,10 @@ function PortListSection({
|
||||
loading,
|
||||
loaded,
|
||||
onRefresh,
|
||||
activeTunnels,
|
||||
pendingTunnels,
|
||||
onTunnelStart,
|
||||
onTunnelStop,
|
||||
}: {
|
||||
detectedPorts: ListeningPort[];
|
||||
manualPorts: number[];
|
||||
@@ -112,6 +289,10 @@ function PortListSection({
|
||||
loading: boolean;
|
||||
loaded: boolean;
|
||||
onRefresh: () => void;
|
||||
activeTunnels: Map<number, number>;
|
||||
pendingTunnels: Set<number>;
|
||||
onTunnelStart: (port: number, requestedPort?: number) => void;
|
||||
onTunnelStop: (port: number) => void;
|
||||
}) {
|
||||
const detectedPortNumbers = new Set(detectedPorts.map((p) => p.port));
|
||||
const uniqueManualPorts = manualPorts.filter((p) => !detectedPortNumbers.has(p));
|
||||
@@ -119,7 +300,10 @@ function PortListSection({
|
||||
return (
|
||||
<div className="space-y-2">
|
||||
<div className="flex items-center justify-between">
|
||||
<span className="text-sm font-medium">Listening Ports</span>
|
||||
<span className="text-sm font-medium flex items-center gap-1.5">
|
||||
Listening Ports
|
||||
<InfoTip text="TCP ports with active listeners inside the remote executor. Click refresh to re-scan." />
|
||||
</span>
|
||||
<Button
|
||||
size="sm"
|
||||
variant="ghost"
|
||||
@@ -151,12 +335,26 @@ function PortListSection({
|
||||
key={`d-${p.port}`}
|
||||
port={p.port}
|
||||
address={p.address}
|
||||
process={p.process}
|
||||
sessionId={sessionId}
|
||||
badge="Detected"
|
||||
tunnelPort={activeTunnels.get(p.port)}
|
||||
tunnelPending={pendingTunnels.has(p.port)}
|
||||
onTunnelStart={onTunnelStart}
|
||||
onTunnelStop={onTunnelStop}
|
||||
/>
|
||||
))}
|
||||
{uniqueManualPorts.map((port) => (
|
||||
<PortRow key={`m-${port}`} port={port} sessionId={sessionId} badge="Manual" />
|
||||
<PortRow
|
||||
key={`m-${port}`}
|
||||
port={port}
|
||||
sessionId={sessionId}
|
||||
badge="Manual"
|
||||
tunnelPort={activeTunnels.get(port)}
|
||||
tunnelPending={pendingTunnels.has(port)}
|
||||
onTunnelStart={onTunnelStart}
|
||||
onTunnelStop={onTunnelStop}
|
||||
/>
|
||||
))}
|
||||
</div>
|
||||
</div>
|
||||
@@ -178,7 +376,10 @@ function ManualPortInput({ onAdd }: { onAdd: (port: number) => void }) {
|
||||
|
||||
return (
|
||||
<div className="space-y-2">
|
||||
<span className="text-sm font-medium">Add Port Manually</span>
|
||||
<span className="text-sm font-medium flex items-center gap-1.5">
|
||||
Add Port Manually
|
||||
<InfoTip text="Add a port that isn't auto-detected. Useful for services not yet started." />
|
||||
</span>
|
||||
<div className="flex gap-2">
|
||||
<Input
|
||||
data-testid="port-forward-port-input"
|
||||
@@ -206,11 +407,28 @@ function ManualPortInput({ onAdd }: { onAdd: (port: number) => void }) {
|
||||
);
|
||||
}
|
||||
|
||||
function PortForwardDialogContent({ sessionId }: { sessionId: string }) {
|
||||
function PortForwardDialogContent({
|
||||
sessionId,
|
||||
activeTunnels,
|
||||
setActiveTunnels,
|
||||
}: {
|
||||
sessionId: string;
|
||||
activeTunnels: Map<number, number>;
|
||||
setActiveTunnels: (updater: (prev: Map<number, number>) => Map<number, number>) => void;
|
||||
}) {
|
||||
const [detectedPorts, setDetectedPorts] = useState<ListeningPort[]>([]);
|
||||
const [manualPorts, setManualPorts] = useState<number[]>([]);
|
||||
const [loading, setLoading] = useState(false);
|
||||
const [loaded, setLoaded] = useState(false);
|
||||
const { pendingTunnels, handleTunnelStart, handleTunnelStop } = useTunnelActions(
|
||||
sessionId,
|
||||
setActiveTunnels,
|
||||
);
|
||||
|
||||
// Use a ref so refresh doesn't depend on activeTunnels identity (Map reference
|
||||
// changes when listTunnels resolves, which would recreate the callback).
|
||||
const activeTunnelsRef = useRef(activeTunnels);
|
||||
activeTunnelsRef.current = activeTunnels;
|
||||
|
||||
const refresh = useCallback(async () => {
|
||||
setLoading(true);
|
||||
@@ -218,6 +436,17 @@ function PortForwardDialogContent({ sessionId }: { sessionId: string }) {
|
||||
const ports = await listPorts(sessionId);
|
||||
setDetectedPorts(ports);
|
||||
setLoaded(true);
|
||||
|
||||
const tunnels = activeTunnelsRef.current;
|
||||
const detectedSet = new Set(ports.map((p) => p.port));
|
||||
const tunnelOnlyPorts = [...tunnels.keys()].filter((p) => !detectedSet.has(p));
|
||||
if (tunnelOnlyPorts.length > 0) {
|
||||
setManualPorts((prev) => {
|
||||
const existing = new Set(prev);
|
||||
const toAdd = tunnelOnlyPorts.filter((p) => !existing.has(p));
|
||||
return toAdd.length > 0 ? [...prev, ...toAdd] : prev;
|
||||
});
|
||||
}
|
||||
} finally {
|
||||
setLoading(false);
|
||||
}
|
||||
@@ -237,7 +466,7 @@ function PortForwardDialogContent({ sessionId }: { sessionId: string }) {
|
||||
return (
|
||||
<DialogContent
|
||||
data-testid="port-forward-dialog"
|
||||
className="sm:max-w-md"
|
||||
className="sm:max-w-2xl overflow-hidden"
|
||||
onOpenAutoFocus={() => !loaded && refresh()}
|
||||
>
|
||||
<DialogHeader>
|
||||
@@ -246,7 +475,7 @@ function PortForwardDialogContent({ sessionId }: { sessionId: string }) {
|
||||
Port Forwarding
|
||||
</DialogTitle>
|
||||
</DialogHeader>
|
||||
<div className="space-y-4">
|
||||
<div className="space-y-4 min-w-0 max-h-[60vh] overflow-y-auto">
|
||||
<PortListSection
|
||||
detectedPorts={detectedPorts}
|
||||
manualPorts={manualPorts}
|
||||
@@ -254,6 +483,10 @@ function PortForwardDialogContent({ sessionId }: { sessionId: string }) {
|
||||
loading={loading}
|
||||
loaded={loaded}
|
||||
onRefresh={refresh}
|
||||
activeTunnels={activeTunnels}
|
||||
pendingTunnels={pendingTunnels}
|
||||
onTunnelStart={handleTunnelStart}
|
||||
onTunnelStop={handleTunnelStop}
|
||||
/>
|
||||
<ManualPortInput onAdd={handleAddManual} />
|
||||
</div>
|
||||
@@ -270,6 +503,23 @@ export function PortForwardButton({
|
||||
sessionId?: string | null;
|
||||
isAgentctlReady?: boolean;
|
||||
}) {
|
||||
const [activeTunnels, setActiveTunnelsRaw] = useState<Map<number, number>>(new Map());
|
||||
const hasActiveTunnels = activeTunnels.size > 0;
|
||||
|
||||
const setActiveTunnels = useCallback(
|
||||
(updater: (prev: Map<number, number>) => Map<number, number>) => {
|
||||
setActiveTunnelsRaw((prev) => updater(prev));
|
||||
},
|
||||
[],
|
||||
);
|
||||
|
||||
useEffect(() => {
|
||||
if (!sessionId || !isAgentctlReady) return;
|
||||
listTunnels(sessionId).then((tunnels) => {
|
||||
setActiveTunnelsRaw(new Map(tunnels.map((t) => [t.port, t.tunnel_port])));
|
||||
});
|
||||
}, [sessionId, isAgentctlReady]);
|
||||
|
||||
if (!isRemoteExecutor || !sessionId || !isAgentctlReady) return null;
|
||||
|
||||
return (
|
||||
@@ -280,16 +530,24 @@ export function PortForwardButton({
|
||||
<Button
|
||||
data-testid="port-forward-button"
|
||||
size="sm"
|
||||
variant="outline"
|
||||
variant={hasActiveTunnels ? "default" : "outline"}
|
||||
className="cursor-pointer px-2"
|
||||
>
|
||||
<IconNetwork className="h-4 w-4" />
|
||||
</Button>
|
||||
</DialogTrigger>
|
||||
</TooltipTrigger>
|
||||
<TooltipContent>Port Forwarding</TooltipContent>
|
||||
<TooltipContent>
|
||||
{hasActiveTunnels
|
||||
? `Port Forwarding (${activeTunnels.size} tunnel${activeTunnels.size > 1 ? "s" : ""} active)`
|
||||
: "Port Forwarding"}
|
||||
</TooltipContent>
|
||||
</Tooltip>
|
||||
<PortForwardDialogContent sessionId={sessionId} />
|
||||
<PortForwardDialogContent
|
||||
sessionId={sessionId}
|
||||
activeTunnels={activeTunnels}
|
||||
setActiveTunnels={setActiveTunnels}
|
||||
/>
|
||||
</Dialog>
|
||||
);
|
||||
}
|
||||
|
||||
@@ -0,0 +1,62 @@
|
||||
"use client";
|
||||
|
||||
import { useState, useCallback } from "react";
|
||||
import { startTunnel, stopTunnel } from "@/lib/api/domains/port-api";
|
||||
import { toast } from "sonner";
|
||||
|
||||
export function useTunnelActions(
|
||||
sessionId: string,
|
||||
setActiveTunnels: (updater: (prev: Map<number, number>) => Map<number, number>) => void,
|
||||
) {
|
||||
const [pendingTunnels, setPendingTunnels] = useState<Set<number>>(new Set());
|
||||
|
||||
const handleTunnelStart = useCallback(
|
||||
async (port: number, requestedPort?: number) => {
|
||||
setPendingTunnels((prev) => new Set(prev).add(port));
|
||||
try {
|
||||
const tunnelPort = await startTunnel(sessionId, port, requestedPort);
|
||||
setActiveTunnels((prev) => new Map(prev).set(port, tunnelPort));
|
||||
toast.success(`Tunnel started on port ${tunnelPort}`);
|
||||
} catch (err) {
|
||||
toast.error(
|
||||
`Failed to start tunnel: ${err instanceof Error ? err.message : "unknown error"}`,
|
||||
);
|
||||
} finally {
|
||||
setPendingTunnels((prev) => {
|
||||
const next = new Set(prev);
|
||||
next.delete(port);
|
||||
return next;
|
||||
});
|
||||
}
|
||||
},
|
||||
[sessionId, setActiveTunnels],
|
||||
);
|
||||
|
||||
const handleTunnelStop = useCallback(
|
||||
async (port: number) => {
|
||||
setPendingTunnels((prev) => new Set(prev).add(port));
|
||||
try {
|
||||
await stopTunnel(sessionId, port);
|
||||
setActiveTunnels((prev) => {
|
||||
const next = new Map(prev);
|
||||
next.delete(port);
|
||||
return next;
|
||||
});
|
||||
toast.success("Tunnel stopped");
|
||||
} catch (err) {
|
||||
toast.error(
|
||||
`Failed to stop tunnel: ${err instanceof Error ? err.message : "unknown error"}`,
|
||||
);
|
||||
} finally {
|
||||
setPendingTunnels((prev) => {
|
||||
const next = new Set(prev);
|
||||
next.delete(port);
|
||||
return next;
|
||||
});
|
||||
}
|
||||
},
|
||||
[sessionId, setActiveTunnels],
|
||||
);
|
||||
|
||||
return { pendingTunnels, handleTunnelStart, handleTunnelStop };
|
||||
}
|
||||
@@ -101,14 +101,16 @@ test.describe("Port Forward Dialog", () => {
|
||||
await expect(session.portForwardDialog).toBeVisible();
|
||||
});
|
||||
|
||||
test("refresh shows no listening ports message", async ({ testPage, apiClient, seedData }) => {
|
||||
test("auto-refresh loads port list on open", async ({ testPage, apiClient, seedData }) => {
|
||||
const { session } = await seedRemoteSession(testPage, apiClient, seedData, "Refresh Test");
|
||||
await session.portForwardButton.click();
|
||||
await expect(session.portForwardDialog).toBeVisible();
|
||||
// Dialog auto-refreshes on open; wait for the "no ports" message
|
||||
await expect(session.portForwardDialog.getByText("No listening ports detected")).toBeVisible({
|
||||
timeout: 10_000,
|
||||
});
|
||||
// Dialog auto-refreshes on open; wait for the placeholder to disappear,
|
||||
// meaning either ports were detected or the "no ports" message appeared.
|
||||
// The E2E host may have real listening ports, so we accept either outcome.
|
||||
const noPortsMessage = session.portForwardDialog.getByText("No listening ports detected");
|
||||
const detectedBadge = session.portForwardDialog.getByText("Detected").first();
|
||||
await expect(noPortsMessage.or(detectedBadge)).toBeVisible({ timeout: 10_000 });
|
||||
});
|
||||
|
||||
test("add manual port shows row with Manual badge", async ({ testPage, apiClient, seedData }) => {
|
||||
|
||||
@@ -1,6 +1,8 @@
|
||||
import { getWebSocketClient } from "@/lib/ws/connection";
|
||||
|
||||
export type ListeningPort = { port: number; address: string };
|
||||
export type ListeningPort = { port: number; address: string; process?: string };
|
||||
|
||||
export type TunnelInfo = { port: number; tunnel_port: number };
|
||||
|
||||
export async function listPorts(sessionId: string): Promise<ListeningPort[]> {
|
||||
const client = getWebSocketClient();
|
||||
@@ -15,3 +17,43 @@ export async function listPorts(sessionId: string): Promise<ListeningPort[]> {
|
||||
return [];
|
||||
}
|
||||
}
|
||||
|
||||
export async function startTunnel(
|
||||
sessionId: string,
|
||||
port: number,
|
||||
tunnelPort?: number,
|
||||
): Promise<number> {
|
||||
const client = getWebSocketClient();
|
||||
if (!client) throw new Error("WebSocket not connected");
|
||||
|
||||
const response = (await client.request("port.tunnel.start", {
|
||||
session_id: sessionId,
|
||||
port,
|
||||
tunnel_port: tunnelPort ?? 0,
|
||||
})) as { tunnel_port: number };
|
||||
return response.tunnel_port;
|
||||
}
|
||||
|
||||
export async function stopTunnel(sessionId: string, port: number): Promise<void> {
|
||||
const client = getWebSocketClient();
|
||||
if (!client) throw new Error("WebSocket not connected");
|
||||
|
||||
await client.request("port.tunnel.stop", {
|
||||
session_id: sessionId,
|
||||
port,
|
||||
});
|
||||
}
|
||||
|
||||
export async function listTunnels(sessionId: string): Promise<TunnelInfo[]> {
|
||||
const client = getWebSocketClient();
|
||||
if (!client) return [];
|
||||
|
||||
try {
|
||||
const response = (await client.request("port.tunnel.list", {
|
||||
session_id: sessionId,
|
||||
})) as { tunnels: TunnelInfo[] };
|
||||
return response.tunnels ?? [];
|
||||
} catch {
|
||||
return [];
|
||||
}
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user