feat(agent): use multistream and yamux muxer instead of HTTP server

This change replaces the legacy per-connection HTTP/revdial
reverse-tunnel flow with a multiplexed, per-stream protocol (v2) based
on HashiCorp's yamux and multistream-select.

Yamux sessions allow a single agent websocket to host many logical
streams without repeating the HTTP reverse-listen handshake for each
session. Multistream-select provides explicit per-stream protocol
negotiation, making each logical stream's intent (SSH open/close, HTTP
proxy) explicit and versionable. The new approach reduces handshake
overhead and makes future protocol extensions easier and safer.

Over time the legacy reverse-tunnel implementation (per-request
HTTP/revdial) has become a maintenance burden and a performance limiter:
each logical session required a full HTTP handshake and a separate
reverse listener, which increased latency and duplicated logic across
server components. This change replaces that model with a single
multiplexed transport per agent using HashiCorp's yamux, and introduces
explicit per-stream protocol negotiation via multistream-select.

Practically, when an agent connects we now bind its websocket to a yamux
session (Manager.Bind) and keep that session alive with a light ping
loop. Individual logical operations—opening an SSH session, closing it,
or proxying HTTP—are created as yamux streams. Each stream negotiates
its intent using multistream identifiers (`/ssh/open/1.0.0`,
`/ssh/close/1.0.0`, `/http/proxy/1.0.0`) and then exchanges a small JSON
envelope for any parameters. The dialer package centralizes version
handling and prepares connections for callers via DialTo(ctx, tenant,
uid, target), while Target implementations encapsulate the
version-specific bootstrap (legacy HTTP v1 vs yamux+multistream v2).

We kept backward compatibility in mind: the Manager still recognizes
existing revdial-based connections and will return a v1 connection when
appropriate. That lets the system run both v1 and v2 agents concurrently
and supports staged rollouts. I also removed the duplicated tunnel
wiring and replaced it with a focused HTTP sidecar (http) that exposes
both the legacy endpoints and the new v2 bind endpoint; server and
session code now call into dialer.Dialer and Target helpers rather than
performing raw HTTP handshakes.

Author: henrybarreto

agent/agent.go

@@ -9,8 +9,10 @@ require (
 	github.com/docker/docker v28.5.2+incompatible
 	github.com/gliderlabs/ssh v0.3.5
 	github.com/go-playground/assert/v2 v2.2.0
-	github.com/labstack/echo/v4 v4.13.4
+	github.com/gorilla/websocket v1.5.0
+	github.com/hashicorp/yamux v0.1.2
 	github.com/mattn/go-shellwords v1.0.12
+	github.com/multiformats/go-multistream v0.6.1
 	github.com/openwall/yescrypt-go v1.0.0
 	github.com/pkg/errors v0.9.1
 	github.com/pkg/sftp v1.13.10
@@ -22,14 +24,6 @@ require (
 	golang.org/x/sys v0.38.0
 )
 
-require (
-	github.com/labstack/gommon v0.4.2 // indirect
-	github.com/mattn/go-colorable v0.1.14 // indirect
-	github.com/mattn/go-isatty v0.0.20 // indirect
-	github.com/valyala/bytebufferpool v1.0.0 // indirect
-	github.com/valyala/fasttemplate v1.2.2 // indirect
-)
-
 require (
 	github.com/Azure/go-ansiterm v0.0.0-20230124172434-306776ec8161 // indirect
 	github.com/Microsoft/go-winio v0.6.2 // indirect
@@ -48,12 +42,12 @@ require (
 	github.com/go-playground/validator/v10 v10.11.2 // indirect
 	github.com/go-resty/resty/v2 v2.7.0 // indirect
 	github.com/google/uuid v1.6.0 // indirect
-	github.com/gorilla/websocket v1.5.0 // indirect
 	github.com/inconshreveable/mousetrap v1.1.0 // indirect
 	github.com/kr/fs v0.1.0 // indirect
 	github.com/leodido/go-urn v1.2.2 // indirect
 	github.com/moby/docker-image-spec v1.3.1 // indirect
 	github.com/moby/sys/atomicwriter v0.1.0 // indirect
+	github.com/multiformats/go-varint v0.0.6 // indirect
 	github.com/opencontainers/go-digest v1.0.0 // indirect
 	github.com/opencontainers/image-spec v1.1.0 // indirect
 	github.com/pmezard/go-difflib v1.0.0 // indirect

agent/go.mod

@@ -55,6 +55,8 @@ github.com/gorilla/websocket v1.5.0 h1:PPwGk2jz7EePpoHN/+ClbZu8SPxiqlu12wZP/3sWm
 github.com/gorilla/websocket v1.5.0/go.mod h1:YR8l580nyteQvAITg2hZ9XVh4b55+EU/adAjf1fMHhE=
 github.com/grpc-ecosystem/grpc-gateway/v2 v2.19.1 h1:/c3QmbOGMGTOumP2iT/rCwB7b0QDGLKzqOmktBjT+Is=
 github.com/grpc-ecosystem/grpc-gateway/v2 v2.19.1/go.mod h1:5SN9VR2LTsRFsrEC6FHgRbTWrTHu6tqPeKxEQv15giM=
+github.com/hashicorp/yamux v0.1.2 h1:XtB8kyFOyHXYVFnwT5C3+Bdo8gArse7j2AQ0DA0Uey8=
+github.com/hashicorp/yamux v0.1.2/go.mod h1:C+zze2n6e/7wshOZep2A70/aQU6QBRWJO/G6FT1wIns=
 github.com/inconshreveable/mousetrap v1.1.0 h1:wN+x4NVGpMsO7ErUn/mUI3vEoE6Jt13X2s0bqwp9tc8=
 github.com/inconshreveable/mousetrap v1.1.0/go.mod h1:vpF70FUmC8bwa3OWnCshd2FqLfsEA9PFc4w1p2J65bw=
 github.com/jarcoal/httpmock v1.3.1 h1:iUx3whfZWVf3jT01hQTO/Eo5sAYtB2/rqaUuOtpInww=
@@ -65,16 +67,8 @@ github.com/kr/pretty v0.3.1 h1:flRD4NNwYAUpkphVc1HcthR4KEIFJ65n8Mw5qdRn3LE=
 github.com/kr/pretty v0.3.1/go.mod h1:hoEshYVHaxMs3cyo3Yncou5ZscifuDolrwPKZanG3xk=
 github.com/kr/text v0.2.0 h1:5Nx0Ya0ZqY2ygV366QzturHI13Jq95ApcVaJBhpS+AY=
 github.com/kr/text v0.2.0/go.mod h1:eLer722TekiGuMkidMxC/pM04lWEeraHUUmBw8l2grE=
-github.com/labstack/echo/v4 v4.13.4 h1:oTZZW+T3s9gAu5L8vmzihV7/lkXGZuITzTQkTEhcXEA=
-github.com/labstack/echo/v4 v4.13.4/go.mod h1:g63b33BZ5vZzcIUF8AtRH40DrTlXnx4UMC8rBdndmjQ=
-github.com/labstack/gommon v0.4.2 h1:F8qTUNXgG1+6WQmqoUWnz8WiEU60mXVVw0P4ht1WRA0=
-github.com/labstack/gommon v0.4.2/go.mod h1:QlUFxVM+SNXhDL/Z7YhocGIBYOiwB0mXm1+1bAPHPyU=
 github.com/leodido/go-urn v1.2.2 h1:7z68G0FCGvDk646jz1AelTYNYWrTNm0bEcFAo147wt4=
 github.com/leodido/go-urn v1.2.2/go.mod h1:kUaIbLZWttglzwNuG0pgsh5vuV6u2YcGBYz1hIPjtOQ=
-github.com/mattn/go-colorable v0.1.14 h1:9A9LHSqF/7dyVVX6g0U9cwm9pG3kP9gSzcuIPHPsaIE=
-github.com/mattn/go-colorable v0.1.14/go.mod h1:6LmQG8QLFO4G5z1gPvYEzlUgJ2wF+stgPZH1UqBm1s8=
-github.com/mattn/go-isatty v0.0.20 h1:xfD0iDuEKnDkl03q4limB+vH+GxLEtL/jb4xVJSWWEY=
-github.com/mattn/go-isatty v0.0.20/go.mod h1:W+V8PltTTMOvKvAeJH7IuucS94S2C6jfK/D7dTCTo3Y=
 github.com/mattn/go-shellwords v1.0.12 h1:M2zGm7EW6UQJvDeQxo4T51eKPurbeFbe8WtebGE2xrk=
 github.com/mattn/go-shellwords v1.0.12/go.mod h1:EZzvwXDESEeg03EKmM+RmDnNOPKG4lLtQsUlTZDWQ8Y=
 github.com/moby/docker-image-spec v1.3.1 h1:jMKff3w6PgbfSa69GfNg+zN/XLhfXJGnEx3Nl2EsFP0=
@@ -87,6 +81,10 @@ github.com/moby/term v0.5.0 h1:xt8Q1nalod/v7BqbG21f8mQPqH+xAaC9C3N3wfWbVP0=
 github.com/moby/term v0.5.0/go.mod h1:8FzsFHVUBGZdbDsJw/ot+X+d5HLUbvklYLJ9uGfcI3Y=
 github.com/morikuni/aec v1.0.0 h1:nP9CBfwrvYnBRgY6qfDQkygYDmYwOilePFkwzv4dU8A=
 github.com/morikuni/aec v1.0.0/go.mod h1:BbKIizmSmc5MMPqRYbxO4ZU0S0+P200+tUnFx7PXmsc=
+github.com/multiformats/go-multistream v0.6.1 h1:4aoX5v6T+yWmc2raBHsTvzmFhOI8WVOer28DeBBEYdQ=
+github.com/multiformats/go-multistream v0.6.1/go.mod h1:ksQf6kqHAb6zIsyw7Zm+gAuVo57Qbq84E27YlYqavqw=
+github.com/multiformats/go-varint v0.0.6 h1:gk85QWKxh3TazbLxED/NlDVv8+q+ReFJk7Y2W/KhfNY=
+github.com/multiformats/go-varint v0.0.6/go.mod h1:3Ls8CIEsrijN6+B7PbrXRPxHRPuXSrVKRY101jdMZYE=
 github.com/opencontainers/go-digest v1.0.0 h1:apOUWs51W5PlhuyGyz9FCeeBIOUDA/6nW8Oi/yOhh5U=
 github.com/opencontainers/go-digest v1.0.0/go.mod h1:0JzlMkj0TRzQZfJkVvzbP0HBR3IKzErnv2BNG4W4MAM=
 github.com/opencontainers/image-spec v1.1.0 h1:8SG7/vwALn54lVB/0yZ/MMwhFrPYtpEHQb2IpWsCzug=
@@ -124,10 +122,6 @@ github.com/stretchr/testify v1.8.0/go.mod h1:yNjHg4UonilssWZ8iaSj1OCr/vHnekPRkoO
 github.com/stretchr/testify v1.8.2/go.mod h1:w2LPCIKwWwSfY2zedu0+kehJoqGctiVI29o6fzry7u4=
 github.com/stretchr/testify v1.11.1 h1:7s2iGBzp5EwR7/aIZr8ao5+dra3wiQyKjjFuvgVKu7U=
 github.com/stretchr/testify v1.11.1/go.mod h1:wZwfW3scLgRK+23gO65QZefKpKQRnfz6sD981Nm4B6U=
-github.com/valyala/bytebufferpool v1.0.0 h1:GqA5TC/0021Y/b9FG4Oi9Mr3q7XYx6KllzawFIhcdPw=
-github.com/valyala/bytebufferpool v1.0.0/go.mod h1:6bBcMArwyJ5K/AmCkWv1jt77kVWyCJ6HpOuEn7z0Csc=
-github.com/valyala/fasttemplate v1.2.2 h1:lxLXG0uE3Qnshl9QyaK6XJxMXlQZELvChBOCmQD0Loo=
-github.com/valyala/fasttemplate v1.2.2/go.mod h1:KHLXt3tVN2HBp8eijSv/kGJopbvo7S+qRAEEKiv+SiQ=
 go.opentelemetry.io/auto/sdk v1.1.0 h1:cH53jehLUN6UFLY71z+NDOiNJqDdPRaXzTel0sJySYA=
 go.opentelemetry.io/auto/sdk v1.1.0/go.mod h1:3wSPjt5PWp2RhlCcmmOial7AvC4DQqZb7a7wCow3W8A=
 go.opentelemetry.io/contrib/instrumentation/net/http/otelhttp v0.51.0 h1:Xs2Ncz0gNihqu9iosIZ5SkBbWo5T8JhhLJFMQL1qmLI=
@@ -161,7 +155,6 @@ golang.org/x/sys v0.0.0-20210616094352-59db8d763f22/go.mod h1:oPkhp1MJrh7nUepCBc
 golang.org/x/sys v0.0.0-20220715151400-c0bba94af5f8/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
 golang.org/x/sys v0.0.0-20220728004956-3c1f35247d10/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
 golang.org/x/sys v0.0.0-20220825204002-c680a09ffe64/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
-golang.org/x/sys v0.6.0/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
 golang.org/x/sys v0.38.0 h1:3yZWxaJjBmCWXqhN1qh02AkOnCQ1poK6oF+a7xWL6Gc=
 golang.org/x/sys v0.38.0/go.mod h1:OgkHotnGiDImocRcuBABYBEXf8A9a87e/uXjp9XT3ks=
 golang.org/x/term v0.0.0-20201126162022-7de9c90e9dd1/go.mod h1:bj7SfCRtBDWHUb9snDiAeCFNEtKQo2Wmx5Cou7ajbmo=

agent/go.sum

@@ -0,0 +1,79 @@
+package tunnel
+
+import (
+	"context"
+	"encoding/json"
+	"errors"
+	"io"
+	"time"
+
+	log "github.com/sirupsen/logrus"
+)
+
+type Context struct {
+	ctx context.Context
+
+	encoder *json.Encoder
+	decoder *json.Decoder
+}
+
+func (c Context) Deadline() (deadline time.Time, ok bool) {
+	return c.ctx.Deadline()
+}
+
+func (c Context) Done() <-chan struct{} {
+	return c.ctx.Done()
+}
+
+func (c Context) Err() error {
+	return c.ctx.Err()
+}
+
+func (c Context) Value(key any) any {
+	return c.ctx.Value(key)
+}
+
+func (c Context) Status(status string) error {
+	if err := c.encoder.Encode(map[string]string{"status": status}); err != nil {
+		log.WithError(err).Error("failed to send status response")
+
+		return errors.Join(errors.New("failed to send status response"), err)
+	}
+
+	return nil
+}
+
+func (c Context) Error(err error) error {
+	if err := c.encoder.Encode(map[string]string{"error": err.Error()}); err != nil {
+		log.WithError(err).Error("failed to send error response")
+
+		return errors.Join(errors.New("failed to send error response"), err)
+	}
+
+	return nil
+}
+
+type Headers map[string]string
+
+func (c Context) Headers() (Headers, error) {
+	// TODO: cache the headers after the first call.
+	var header Headers
+
+	if err := c.decoder.Decode(&header); err != nil {
+		log.WithError(err).Error("failed to decode the header")
+
+		return nil, err
+	}
+
+	return header, nil
+}
+
+func NewContext(ctx context.Context, rwc io.ReadWriteCloser) Context {
+	return Context{
+		ctx:     ctx,
+		encoder: json.NewEncoder(rwc),
+		decoder: json.NewDecoder(rwc),
+	}
+}
+
+type Handler func(ctx Context, rwc io.ReadWriteCloser) error

agent/pkg/tunnel/context.go

@@ -2,102 +2,85 @@ package tunnel
 
 import (
 	"context"
+	"errors"
+	"io"
 	"net"
-	"net/http"
+	"os"
+	"time"
 
-	"github.com/labstack/echo/v4"
-	"github.com/shellhub-io/shellhub/pkg/revdial"
+	"github.com/gorilla/websocket"
+	"github.com/hashicorp/yamux"
+	"github.com/multiformats/go-multistream"
+	log "github.com/sirupsen/logrus"
 )
 
 type Tunnel struct {
-	router           *echo.Echo
-	srv              *http.Server
-	HTTPProxyHandler func(e echo.Context) error
-	SSHHandler       func(e echo.Context) error
-	SSHCloseHandler  func(e echo.Context) error
+	mux *multistream.MultistreamMuxer[string]
 }
 
-type Builder struct {
-	tunnel *Tunnel
-}
-
-func NewBuilder() *Builder {
-	return &Builder{
-		tunnel: NewTunnel(),
+func NewTunnel() *Tunnel {
+	return &Tunnel{
+		mux: multistream.NewMultistreamMuxer[string](),
 	}
 }
 
-func (t *Builder) WithHTTPProxyHandler(handler func(e echo.Context) error) *Builder {
-	t.tunnel.HTTPProxyHandler = handler
+func (t *Tunnel) Handle(protocol string, handler Handler) {
+	t.mux.AddHandler(protocol, func(protocol string, rwc io.ReadWriteCloser) error {
+		log.WithField("protocol", protocol).Debug("handling connection")
+		defer log.WithField("protocol", protocol).Debug("handling connection closed")
 
-	return t
+		// TODO: Should we receive a context from outside?
+		return handler(NewContext(context.TODO(), rwc), rwc)
+	})
 }
 
-func (t *Builder) WithSSHHandler(handler func(e echo.Context) error) *Builder {
-	t.tunnel.SSHHandler = handler
+// ErrTunnelDisconnect is returned when the tunnel connection is closed.
+var ErrTunnelDisconnect = errors.New("tunnel disconnected")
+
+func (t *Tunnel) Listen(conn net.Conn) error {
+	session, err := yamux.Server(conn, &yamux.Config{
+		AcceptBacklog:          256,
+		EnableKeepAlive:        true,
+		KeepAliveInterval:      35 * time.Second,
+		ConnectionWriteTimeout: 15 * time.Second,
+		MaxStreamWindowSize:    256 * 1024,
+		StreamCloseTimeout:     5 * time.Minute,
+		StreamOpenTimeout:      75 * time.Second,
+		LogOutput:              os.Stderr,
+	})
+	if err != nil {
+		log.WithError(err).Error("failed to create muxed session")
 
-	return t
-}
+		return err
+	}
 
-func (t *Builder) WithSSHCloseHandler(handler func(e echo.Context) error) *Builder {
-	t.tunnel.SSHCloseHandler = handler
+	for {
+		stream, err := session.Accept()
+		if err != nil {
+			defer session.Close()
 
-	return t
-}
+			log.WithError(err).Trace("failed to accept stream")
 
-func (t *Builder) Build() *Tunnel {
-	return t.tunnel
-}
+			switch {
+			case websocket.IsCloseError(err, websocket.CloseAbnormalClosure):
+				return errors.Join(ErrTunnelDisconnect, err)
+			}
 
-func NewTunnel() *Tunnel {
-	e := echo.New()
-
-	t := &Tunnel{
-		router: e,
-		srv: &http.Server{ //nolint:gosec
-			Handler: e,
-			ConnContext: func(ctx context.Context, c net.Conn) context.Context {
-				return context.WithValue(ctx, "http-conn", c) //nolint:revive
-			},
-		},
-		SSHHandler: func(_ echo.Context) error {
-			panic("ConnHandler can not be nil")
-		},
-		SSHCloseHandler: func(_ echo.Context) error {
-			panic("CloseHandler can not be nil")
-		},
-		HTTPProxyHandler: func(_ echo.Context) error {
-			panic("ProxyHandler can not be nil")
-		},
-	}
-	e.GET("/ssh/:id", func(e echo.Context) error {
-		return t.SSHHandler(e)
-	})
-	e.GET("/ssh/close/:id", func(e echo.Context) error {
-		return t.SSHCloseHandler(e)
-	})
-	e.CONNECT("/http/proxy/:addr", func(e echo.Context) error {
-		// NOTE: The CONNECT HTTP method requests that a proxy establish a HTTP tunnel to this server, and if
-		// successful, blindly forward data in both directions until the tunnel is closed.
-		//
-		// https://en.wikipedia.org/wiki/HTTP_tunnel
-		// https://developer.mozilla.org/en-US/docs/Web/HTTP/Methods/CONNECT
-		return t.HTTPProxyHandler(e)
-	})
+			return err
+		}
 
-	return t
-}
+		log.Trace("new stream accepted")
 
-// Listen to reverse listener.
-func (t *Tunnel) Listen(l *revdial.Listener) error {
-	return t.srv.Serve(l)
-}
+		go func() {
+			log.Trace("handling stream")
 
-// Close closes the tunnel.
-func (t *Tunnel) Close() error {
-	if err := t.router.Close(); err != nil {
-		return err
-	}
+			if err := t.mux.Handle(stream); err != nil {
+				log.WithError(err).Trace("failed to handle stream")
 
-	return t.srv.Close()
+				_ = stream.Close()
+			}
+
+			log.Trace("stream handled")
+		}()
+	}
 }

agent/pkg/tunnel/tunnel.go

@@ -389,6 +389,29 @@ server {
         proxy_redirect off;
     }
 
+    location /connection {
+        set $upstream ssh:8080;
+
+        auth_request /auth;
+        auth_request_set $tenant_id $upstream_http_x_tenant_id;
+        auth_request_set $device_uid $upstream_http_x_device_uid;
+        proxy_pass http://$upstream;
+        proxy_set_header Connection $connection_upgrade;
+        proxy_set_header Host $host;
+        proxy_set_header Upgrade $http_upgrade;
+        {{ if $cfg.EnableProxyProtocol -}}
+        proxy_set_header X-Real-IP $proxy_protocol_addr;
+        {{ else -}}
+        proxy_set_header X-Real-IP $x_real_ip;
+        {{ end -}}
+        proxy_set_header X-Device-UID $device_uid;
+        proxy_set_header X-Tenant-ID $tenant_id;
+        proxy_set_header X-Request-ID $request_id;
+        proxy_http_version 1.1;
+        proxy_cache_bypass $http_upgrade;
+        proxy_redirect off;
+    }
+
     location /ssh/auth {
         set $upstream api:8080;
 

gateway/nginx/conf.d/shellhub.conf

@@ -28,7 +28,12 @@ type publicAPI interface {
 	Endpoints() (*models.Endpoints, error)
 	AuthDevice(req *models.DeviceAuthRequest) (*models.DeviceAuthResponse, error)
 	AuthPublicKey(req *models.PublicKeyAuthRequest, token string) (*models.PublicKeyAuthResponse, error)
+	// NewReverseListener creates a new reverse listener to be used by the Agent to connect to ShellHub's SSH server.
+	//
+	//Deprecated: Use Connect instead.
 	NewReverseListener(ctx context.Context, token string, connPath string) (*revdial.Listener, error)
+	// Connect creates a new connection to be used by the Agent to connect to ShellHub's SSH server.
+	Connect(ctx context.Context, token string, path string) (net.Conn, error)
 }
 
 //go:generate mockery --name=Client --filename=client.go