fix(worker): spec the tunnel's routing table, which was the SSRF boundary

Review follow-up. One blocking finding and seven others.

The blocking one first, and it is this project's recurring shape: the untested
path. loopbackService is the function whose comment calls the discarded host
"the security property this function exists for", and nothing tested it. The
reviewer replaced its body with a dial of whatever the frontend named, no port
range, and all 131 specs passed. Every spec installed the permissive test
dialler, so the real routing table was exercised nowhere.

It now has specs, and the property is stated as reachability rather than as a
property of the code: a listener on 127.0.0.2 that only the frontend's target
names must NOT be reached. Plus the port-range table, fixedService, loopbackAddr,
tunnelEndpoint, and the table itself, which moved out of Run into tunnelServices
so it can be built without starting a worker. One spec drives a real stream
through that table over the wire, so the routing rules are exercised end to end
at least once rather than only in isolation. The reviewer's mutation now reddens
ten specs, and six narrower ones redden between two and four each, so no spec is
riding on another.

The shape changed too, not only the coverage. The dial address is built from a
loopbackHost constant and strconv.Itoa of a validated int, so nothing derived
from the wire reaches DialContext at all: restoring the hole takes ADDING a data
flow, not deleting a check.

And a taxonomy fix found while specifying it. A port outside this worker's
allocator range was reported as unavailable, which tells a frontend to retry
something that can never work. It is a bad request now, and a backend that is
merely not listening yet stays unavailable, which is the retryable one.

Agent nodes no longer get a tunnel credential. Nothing dials into an agent
worker, so a tunnel replaces nothing for it and no client would open one, and
the gate is at the mint site rather than in the handler: with no credential
minted the hash stays empty and the existing empty-hash refusal covers it, so
enforcement is structural.

Two comments and one doc paragraph said an anonymous registrant gets a "working"
credential. With auto-approve off the node is pending and the credential is
inert, which is the distinction this same change argues three files away to
justify minting for pending nodes at all.

A refusal reason over the frame limit was cut on a byte boundary and could split
a rune. It cuts on a rune boundary now, and the code survives truncation, which
is what keeps a refusal classifiable.

Also: the pending-node spec asserted only that a credential was non-empty, so a
credential derived from the shared token passed it; it now pins per-node-ness the
way the headline spec does. The tunnel handler's citations into nodes.go were
stale before this branch landed, having been written against a file the same
commit was editing, and are by function name now. The static-NATS path says
plainly that an externally forced rotation locks it out until restart, and where
that gets fixed. tunnelproto gained direct specs, including that a read failure
is never reported as a refusal.

Signed-off-by: Ettore Di Giacinto <mudler@localai.io>
Assisted-by: Claude Opus 5 [claude-code]

core/http/app.go

@@ -609,8 +609,17 @@ func API(application *application.Application) (*echo.Echo, error) {
 			// credential at registration whether or not one is configured. What
 			// is missing is the gate in FRONT of that. With no registration
 			// token, RegisterNodeEndpoint validates nothing, so anyone who can
-			// reach this frontend can register a node and be handed a working
-			// tunnel credential for it.
+			// reach this frontend can register a node and be issued a tunnel
+			// credential for it.
+			//
+			// How far that gets them depends on the OTHER knob. With
+			// auto-approve on, the node is healthy at once and the credential
+			// works immediately. With it off, the node is pending, and the
+			// tunnel route refuses a pending node on every dial, so the
+			// credential is inert until an admin approves it and approval is
+			// the real gate. Worth stating precisely, because the same commit
+			// argues exactly this distinction three files away to justify
+			// minting for pending nodes at all.
 			//
 			// This warning replaced one that said the opposite, that tunnels
 			// would refuse every dial without this token. That was true while

core/http/endpoints/cluster/connect.go

@@ -130,8 +130,13 @@ func ConnectHandler(registry *nodes.NodeRegistry, tunnels *clustersvc.TunnelRegi
 		// Only StatusPending is refused. The rest of /api/node/ self-service
 		// gates on nothing at all, but the two places that hand a node something
 		// DURABLE both refuse a pending one: the agent worker's API key
-		// (core/http/endpoints/localai/nodes.go:224) and its NATS credential
-		// (nodes.go:293). A tunnel is that kind of grant, not a heartbeat: it is
+		// (provisionAgentWorkerKey, guarded at its call site in
+		// core/http/endpoints/localai/nodes.go) and its NATS credential
+		// (attachNatsJWT in the same file). Cited by NAME, not by line: the
+		// previous version of this comment cited line numbers into a file this
+		// same commit was editing, and both were stale before it landed.
+		//
+		// A tunnel is that kind of grant, not a heartbeat: it is
 		// a standing pipe into the worker recorded in node_connections and
 		// relayed to by every other replica. Draining and unhealthy nodes keep
 		// their tunnels on purpose; draining means finish what you have, and a

core/http/endpoints/localai/nodes.go

@@ -306,6 +306,21 @@ func ApproveNodeEndpoint(registry *nodes.NodeRegistry, authDB *gorm.DB, hmacSecr
 // path in core/services/worker/worker.go does), because approval alone does not
 // prompt a re-registration and nothing else can hand it the secret.
 //
+// Only BACKEND nodes get one, and that is a decision rather than an oversight.
+// An agent worker serves no gRPC backends and no file staging; nothing dials
+// into it at all, so a tunnel replaces nothing for it and there is no client on
+// the agent side that would ever open one. Minting anyway would hand out a
+// working credential for a pipe nobody drives, which is surface without a
+// feature, and it would contradict every comment in this change that says
+// "backend workers, the ones that tunnel".
+//
+// The gate lives HERE and not in ConnectHandler, which never looks at NodeType.
+// It does not need to: an agent node's tunnel credential is never minted, so
+// its TunnelTokenHash stays empty and the handler's empty-hash branch refuses
+// it like any other node without one. Enforcement is therefore structural. The
+// day agent workers want a tunnel, relaxing this condition is the whole change,
+// and it has to be a deliberate one.
+//
 // A failure to mint or to store is logged and the response goes out without the
 // token. Registration is what gets a worker into the cluster at all, and
 // failing it over a credential the worker does not need until it tunnels would
@@ -313,7 +328,7 @@ func ApproveNodeEndpoint(registry *nodes.NodeRegistry, authDB *gorm.DB, hmacSecr
 // tunnel_token, reports that it has no credential, and retries at its next
 // registration.
 func attachTunnelToken(ctx context.Context, response map[string]any, registry *nodes.NodeRegistry, node *nodes.BackendNode) {
-	if node == nil {
+	if node == nil || node.NodeType != nodes.NodeTypeBackend {
 		return
 	}
 	// crypto/rand.Text: at least 128 bits of randomness, no error to handle and

core/http/endpoints/localai/nodes_test.go

@@ -154,15 +154,49 @@ var _ = Describe("Node HTTP handlers", func() {
 		})
 
 		It("issues a tunnel credential to a node still awaiting approval", func() {
-			resp := register(`{"name":"worker-pending","address":"10.0.0.5:50051"}`, "", false)
-			Expect(resp["status"]).To(Equal(nodes.StatusPending))
 			// Deliberately unlike the agent API key and the NATS JWT, which are
 			// both withheld from a pending node. Those work the moment they are
 			// issued; this one does not, because the tunnel endpoint re-reads
 			// the node's status on every dial and refuses a pending node. A
 			// worker that registers exactly once would otherwise never receive
 			// one, since approval alone prompts no re-registration.
-			Expect(resp["tunnel_token"]).ToNot(BeEmpty())
+			first := register(`{"name":"worker-pending","address":"10.0.0.5:50051","token":"shared"}`, "shared", false)
+			Expect(first["status"]).To(Equal(nodes.StatusPending))
+			plaintext, _ := first["tunnel_token"].(string)
+			Expect(plaintext).ToNot(BeEmpty())
+
+			// Non-empty alone does not pin per-node-ness, and a review's
+			// variant of the "derived from the shared token" mutation stayed
+			// green on exactly that gap. A pending node's credential has to be
+			// as unpredictable and as per-node as an approved one's, since it
+			// becomes live the moment an admin approves.
+			Expect(plaintext).ToNot(Equal("shared"))
+			node, err := registry.Get(context.Background(), first["id"].(string))
+			Expect(err).ToNot(HaveOccurred())
+			Expect(node.TunnelTokenHash).To(Equal(hashOf(plaintext)))
+			Expect(node.TunnelTokenHash).ToNot(Equal(node.TokenHash))
+
+			second := register(`{"name":"worker-pending-2","address":"10.0.0.5:50052","token":"shared"}`, "shared", false)
+			Expect(second["status"]).To(Equal(nodes.StatusPending))
+			Expect(second["tunnel_token"]).ToNot(Equal(plaintext))
+		})
+
+		It("does not issue a tunnel credential to an agent node", func() {
+			// An agent worker serves no gRPC backends and no file staging;
+			// nothing dials into it, so a tunnel replaces nothing for it and no
+			// client on its side would open one. Minting anyway would be
+			// credential surface with no feature behind it.
+			//
+			// Enforcement is structural rather than a second check: with no
+			// credential minted, the node's hash stays empty and the tunnel
+			// route refuses it like any other node without one.
+			resp := register(`{"name":"agent-1","node_type":"agent"}`, "", true)
+			Expect(resp["node_type"]).To(Equal(nodes.NodeTypeAgent))
+			Expect(resp).ToNot(HaveKey("tunnel_token"))
+
+			node, err := registry.Get(context.Background(), resp["id"].(string))
+			Expect(err).ToNot(HaveOccurred())
+			Expect(node.TunnelTokenHash).To(BeEmpty())
 		})
 
 		It("returns nats_jwt when account seed is configured", func() {

core/services/cluster/tunnelproto.go

@@ -8,6 +8,7 @@ import (
 	"fmt"
 	"io"
 	"strings"
+	"unicode/utf8"
 )
 
 // The framing every stream on a worker tunnel opens with.
@@ -154,10 +155,7 @@ func WriteStreamRefusal(w io.Writer, reason error) error {
 		}, reason.Error())
 	}
 	frame := replyPrefixRefused + code + streamRequestSeparator + text
-	if len(frame) > maxTunnelFrame {
-		frame = frame[:maxTunnelFrame]
-	}
-	return writeFrame(w, frame)
+	return writeFrame(w, truncateRunes(frame, maxTunnelFrame))
 }
 
 // ReadStreamReply reads the worker's answer. nil means the stream is now
@@ -196,6 +194,30 @@ func ReadStreamReply(r io.Reader) error {
 	}
 }
 
+// truncateRunes cuts s to at most limit BYTES, on a rune boundary.
+//
+// A plain slice would cut mid-rune and put a lone continuation byte on the
+// wire. Nothing breaks: the frame is length-prefixed so the framing survives,
+// and the reader's string() tolerates invalid UTF-8. What it costs is the
+// far side's log line ending in a replacement character, and a refusal reason
+// exists to be read by a person, so it should not arrive damaged.
+//
+// The code that reaches this is always short; only a cause from a local service
+// can be long enough to matter.
+func truncateRunes(s string, limit int) string {
+	if len(s) <= limit {
+		return s
+	}
+	cut := limit
+	// utf8.RuneStart finds the first byte of a rune. Walking back from the
+	// limit lands on the start of the rune that would have been split, and at
+	// most 3 steps are needed since a UTF-8 rune is at most 4 bytes.
+	for cut > 0 && !utf8.RuneStart(s[cut]) {
+		cut--
+	}
+	return s[:cut]
+}
+
 // writeFrame writes one length-prefixed frame in a single Write.
 //
 // One Write, not two: the underlying stream is a yamux stream whose writes

core/services/cluster/tunnelproto_test.go

@@ -0,0 +1,186 @@
+package cluster_test
+
+import (
+	"bytes"
+	"encoding/binary"
+	"io"
+	"strings"
+	"unicode/utf8"
+
+	. "github.com/onsi/ginkgo/v2"
+	. "github.com/onsi/gomega"
+
+	"github.com/mudler/LocalAI/core/services/cluster"
+)
+
+var _ = Describe("Worker tunnel stream framing", func() {
+	Describe("the request frame", func() {
+		DescribeTable("round-trips a tag and a target",
+			func(tag, target string) {
+				var buf bytes.Buffer
+				Expect(cluster.WriteStreamRequest(&buf, tag, target)).To(Succeed())
+				gotTag, gotTarget, err := cluster.ReadStreamRequest(&buf)
+				Expect(err).ToNot(HaveOccurred())
+				Expect(gotTag).To(Equal(tag))
+				Expect(gotTarget).To(Equal(target))
+			},
+			Entry("a tag and an address", cluster.StreamTagGRPC, "127.0.0.1:50051"),
+			Entry("a tag with no target", cluster.StreamTagHTTP, ""),
+			// The split is on the FIRST separator, so a target containing one
+			// must survive intact.
+			Entry("a target containing a space", cluster.StreamTagGRPC, "a b c"),
+		)
+
+		It("consumes exactly the frame and not one byte of what follows", func() {
+			// Load-bearing: the stream is handed to gRPC or net/http right
+			// after this, and a reader that over-read would eat the start of
+			// their conversation.
+			var buf bytes.Buffer
+			Expect(cluster.WriteStreamRequest(&buf, cluster.StreamTagGRPC, "127.0.0.1:1")).To(Succeed())
+			buf.WriteString("PRI * HTTP/2.0")
+
+			_, _, err := cluster.ReadStreamRequest(&buf)
+			Expect(err).ToNot(HaveOccurred())
+			rest, err := io.ReadAll(&buf)
+			Expect(err).ToNot(HaveOccurred())
+			Expect(string(rest)).To(Equal("PRI * HTTP/2.0"))
+		})
+
+		DescribeTable("refuses a tag it could not encode unambiguously",
+			func(tag string) {
+				var buf bytes.Buffer
+				Expect(cluster.WriteStreamRequest(&buf, tag, "x")).ToNot(Succeed())
+				Expect(buf.Len()).To(BeZero(), "a refused request must not put a partial frame on the wire")
+			},
+			Entry("empty", ""),
+			// A tag with a space would silently move part of itself into the
+			// target, so it is refused at the writer rather than a round trip
+			// later.
+			Entry("containing a space", "grpc stream"),
+		)
+
+		It("refuses a frame that declares more than the limit without allocating it", func() {
+			var hdr [2]byte
+			binary.BigEndian.PutUint16(hdr[:], 65535)
+			_, _, err := cluster.ReadStreamRequest(bytes.NewReader(hdr[:]))
+			Expect(err).To(HaveOccurred())
+			Expect(err.Error()).To(ContainSubstring("over the"))
+		})
+
+		It("reports a truncated frame as a truncated read, not as a refusal", func() {
+			// ReadStreamRequest must never produce ErrStreamRequestInvalid:
+			// that sentinel is what a worker SENDS, and a reader producing it
+			// would leave a caller unable to tell "the peer refused me" from
+			// "I could not read the peer".
+			var hdr [2]byte
+			binary.BigEndian.PutUint16(hdr[:], 10)
+			_, _, err := cluster.ReadStreamRequest(bytes.NewReader(append(hdr[:], 'a')))
+			Expect(err).To(MatchError(io.ErrUnexpectedEOF))
+			Expect(err).ToNot(MatchError(cluster.ErrStreamRequestInvalid))
+		})
+	})
+
+	Describe("the reply frame", func() {
+		It("reads an acceptance as nil", func() {
+			var buf bytes.Buffer
+			Expect(cluster.WriteStreamAccepted(&buf)).To(Succeed())
+			Expect(cluster.ReadStreamReply(&buf)).To(Succeed())
+		})
+
+		DescribeTable("keeps the three refusals apart",
+			func(sent error, others []error) {
+				var buf bytes.Buffer
+				Expect(cluster.WriteStreamRefusal(&buf, sent)).To(Succeed())
+				got := cluster.ReadStreamReply(&buf)
+				Expect(got).To(MatchError(sent))
+				// The whole point. A caller gives up on an unknown tag, retries
+				// an unavailable target, and reports a bad request as its own
+				// bug; collapsing any pair makes one of those wrong.
+				for _, other := range others {
+					Expect(got).ToNot(MatchError(other))
+				}
+			},
+			Entry("unknown tag", cluster.ErrStreamTagUnknown,
+				[]error{cluster.ErrStreamTargetUnavailable, cluster.ErrStreamRequestInvalid}),
+			Entry("unavailable target", cluster.ErrStreamTargetUnavailable,
+				[]error{cluster.ErrStreamTagUnknown, cluster.ErrStreamRequestInvalid}),
+			Entry("invalid request", cluster.ErrStreamRequestInvalid,
+				[]error{cluster.ErrStreamTagUnknown, cluster.ErrStreamTargetUnavailable}),
+		)
+
+		It("carries the reason text to the far side", func() {
+			var buf bytes.Buffer
+			Expect(cluster.WriteStreamRefusal(&buf, wrapReason(cluster.ErrStreamTagUnknown, "no-such-tag"))).To(Succeed())
+			Expect(cluster.ReadStreamReply(&buf).Error()).To(ContainSubstring("no-such-tag"))
+		})
+
+		It("reports an unrecognised code as itself, not as the nearest known one", func() {
+			// A code from a newer worker. Mapping it onto a known sentinel
+			// would make a frontend retry forever against a refusal that means
+			// something else entirely.
+			var buf bytes.Buffer
+			writeRawFrame(&buf, "err teapot short and stout")
+			got := cluster.ReadStreamReply(&buf)
+			Expect(got).To(HaveOccurred())
+			Expect(got.Error()).To(ContainSubstring("teapot"))
+			Expect(got).ToNot(MatchError(cluster.ErrStreamTagUnknown))
+			Expect(got).ToNot(MatchError(cluster.ErrStreamTargetUnavailable))
+			Expect(got).ToNot(MatchError(cluster.ErrStreamRequestInvalid))
+		})
+
+		It("reports a failure to READ the reply as itself, never as a refusal", func() {
+			// A refusal proves the worker is connected and said no. A read
+			// failure means the tunnel broke. A caller that treated the second
+			// as the first would report a dead link as a policy decision.
+			got := cluster.ReadStreamReply(bytes.NewReader(nil))
+			Expect(got).To(MatchError(io.EOF))
+			Expect(got).ToNot(MatchError(cluster.ErrStreamTagUnknown))
+			Expect(got).ToNot(MatchError(cluster.ErrStreamTargetUnavailable))
+			Expect(got).ToNot(MatchError(cluster.ErrStreamRequestInvalid))
+		})
+
+		It("truncates an over-long reason on a rune boundary, keeping it decodable", func() {
+			// Two-byte runes so a byte-boundary cut lands mid-rune for half of
+			// all lengths; the padding tunes the frame to land exactly there.
+			reason := wrapReason(cluster.ErrStreamTargetUnavailable, strings.Repeat("é", 2000))
+			var buf bytes.Buffer
+			Expect(cluster.WriteStreamRefusal(&buf, reason)).To(Succeed())
+
+			got := cluster.ReadStreamReply(&buf)
+			Expect(got).To(MatchError(cluster.ErrStreamTargetUnavailable))
+			Expect(utf8.ValidString(got.Error())).To(BeTrue(),
+				"the truncated reason reached the far side with a split rune in it")
+		})
+
+		It("still reports the code when the reason is truncated away", func() {
+			// The code must survive truncation: a refusal a frontend cannot
+			// classify is indistinguishable from a worker that hung up.
+			reason := wrapReason(cluster.ErrStreamTagUnknown, strings.Repeat("x", 4000))
+			var buf bytes.Buffer
+			Expect(cluster.WriteStreamRefusal(&buf, reason)).To(Succeed())
+			Expect(cluster.ReadStreamReply(&buf)).To(MatchError(cluster.ErrStreamTagUnknown))
+		})
+	})
+})
+
+// wrapReason builds the shape the worker sends: a sentinel with a cause.
+func wrapReason(sentinel error, text string) error {
+	return &reasonErr{sentinel: sentinel, text: text}
+}
+
+type reasonErr struct {
+	sentinel error
+	text     string
+}
+
+func (e *reasonErr) Error() string { return e.sentinel.Error() + ": " + e.text }
+func (e *reasonErr) Unwrap() error { return e.sentinel }
+
+// writeRawFrame puts a payload on the wire without going through the encoder,
+// so a spec can present a frame the encoder would never produce.
+func writeRawFrame(buf *bytes.Buffer, payload string) {
+	var hdr [2]byte
+	binary.BigEndian.PutUint16(hdr[:], uint16(len(payload)))
+	buf.Write(hdr[:])
+	buf.WriteString(payload)
+}

core/services/worker/tunnel.go

@@ -371,9 +371,7 @@ func (t *Tunnel) accept(ctx context.Context, stream net.Conn) (net.Conn, bool) {
 
 	local, err := svc(ctx, target)
 	if err != nil {
-		// An INFRASTRUCTURE failure, which a frontend may retry, and which must
-		// never be reported as the unknown tag above.
-		t.refuse(stream, fmt.Errorf("%w: %v", cluster.ErrStreamTargetUnavailable, err))
+		t.refuse(stream, classifyServiceFailure(err))
 		return nil, false
 	}
 
@@ -389,6 +387,26 @@ func (t *Tunnel) accept(ctx context.Context, stream net.Conn) (net.Conn, bool) {
 	return local, true
 }
 
+// classifyServiceFailure decides which refusal a local service's error is.
+//
+// A service that has ALREADY classified its own failure keeps that
+// classification. loopbackService does, and the distinction is not cosmetic: a
+// target outside this worker's backend port range is a request this worker will
+// never serve, while a backend that is not listening yet is a condition that
+// clears on its own. Reporting the first as the second tells a frontend to
+// retry something that can never work; reporting the second as the first makes
+// it give up on a backend that is merely starting.
+//
+// Anything unclassified is infrastructure, because that is what an unadorned
+// dial failure is, and it must never become the unknown-tag refusal: a tag this
+// worker serves does not stop being served because one dial failed.
+func classifyServiceFailure(err error) error {
+	if errors.Is(err, cluster.ErrStreamRequestInvalid) {
+		return err
+	}
+	return fmt.Errorf("%w: %v", cluster.ErrStreamTargetUnavailable, err)
+}
+
 // refuse reports why a stream will not be served and then ENDS it.
 //
 // The close is the part that matters and it is not optional. A worker that says
@@ -567,22 +585,59 @@ func tunnelEndpoint(frontendURL, nodeID string) (string, error) {
 //
 // The port range is the one the worker's own port allocator hands to backend
 // processes, so a stream cannot be pointed at some unrelated service that
-// happens to be listening on this host.
+// happens to be listening on this host. It is only as tight as the allocator's
+// range, which by default runs to 65535; a deployment that wants it narrow sets
+// LOCALAI_GRPC_MAX_PORT, which narrows both at once.
+//
+// Note the SHAPE, not only the checks. Nothing derived from the wire reaches
+// the dialler: the address is built from the loopbackHost constant and from
+// strconv.Itoa of an int this function validated, so `target` itself has no
+// path to DialContext at all. Relaxing this into an arbitrary-host dialler
+// therefore takes ADDING a data flow rather than deleting a check, which is the
+// difference between a guard and a property. It has specs either way; the shape
+// is what stops a plausible refactor from quietly restoring the hole.
 func loopbackService(minPort, maxPort int) LocalService {
 	return func(ctx context.Context, target string) (net.Conn, error) {
 		_, portStr, err := net.SplitHostPort(target)
 		if err != nil {
-			return nil, fmt.Errorf("routing a tunnel stream: %q is not a host:port: %w", target, err)
+			return nil, fmt.Errorf("%w: routing a tunnel stream: %q is not a host:port: %v",
+				cluster.ErrStreamRequestInvalid, target, err)
 		}
 		port, err := strconv.Atoi(portStr)
 		if err != nil {
-			return nil, fmt.Errorf("routing a tunnel stream: %q has no numeric port: %w", target, err)
+			return nil, fmt.Errorf("%w: routing a tunnel stream: %q has no numeric port: %v",
+				cluster.ErrStreamRequestInvalid, target, err)
 		}
 		if port < minPort || port > maxPort {
-			return nil, fmt.Errorf("routing a tunnel stream: port %d is outside this worker's backend range [%d, %d]", port, minPort, maxPort)
+			// Invalid rather than unavailable: no retry can bring a port
+			// outside this worker's own allocator range into it.
+			return nil, fmt.Errorf("%w: routing a tunnel stream: port %d is outside this worker's backend range [%d, %d]",
+				cluster.ErrStreamRequestInvalid, port, minPort, maxPort)
 		}
 		var d net.Dialer
-		return d.DialContext(ctx, "tcp", net.JoinHostPort("127.0.0.1", portStr))
+		return d.DialContext(ctx, "tcp", net.JoinHostPort(loopbackHost, strconv.Itoa(port)))
+	}
+}
+
+// loopbackHost is the only host any tunnel stream is ever dialled on. It is a
+// constant so that "the worker dials itself and nothing else" is a fact about
+// the code rather than a claim about its inputs.
+const loopbackHost = "127.0.0.1"
+
+// tunnelServices builds the routing table the worker installs on its tunnel.
+//
+// It exists as its own function so the table can be specced. The table is the
+// security boundary of this whole feature, and building it inline in Run left
+// it reachable only by starting a worker, which meant it was covered by nothing
+// and an arbitrary-host regression passed the entire suite.
+func tunnelServices(cfg *Config, httpBindAddr string) map[string]LocalService {
+	basePort := cfg.effectiveBasePort()
+	return map[string]LocalService{
+		// The frontend names a backend process by its port; the worker decides
+		// that only its own loopback, and only within its own backend port
+		// range, is reachable through it.
+		cluster.StreamTagGRPC: loopbackService(basePort, cfg.effectiveMaxPort(basePort)),
+		cluster.StreamTagHTTP: fixedService(loopbackAddr(httpBindAddr)),
 	}
 }
 
@@ -611,7 +666,7 @@ func loopbackAddr(bindAddr string) string {
 		return bindAddr
 	}
 	if host == "" || host == "0.0.0.0" || host == "::" || host == "[::]" {
-		return net.JoinHostPort("127.0.0.1", port)
+		return net.JoinHostPort(loopbackHost, port)
 	}
 	return bindAddr
 }

core/services/worker/tunnel_test.go

@@ -8,6 +8,7 @@ import (
 	"net"
 	"net/http"
 	"net/http/httptest"
+	"strconv"
 	"strings"
 	"sync/atomic"
 	"time"
@@ -213,6 +214,66 @@ var _ = Describe("Worker tunnel client", func() {
 			Eventually(read, "10s").Should(Receive(BeNil()))
 			Expect(string(buf)).To(Equal("ping"))
 		})
+
+		It("routes through the worker's OWN table, ignoring the host the frontend names", func() {
+			// Every other spec in this file installs dialLocalTCP, which dials
+			// whatever it is handed. This one installs tunnelServices, the
+			// table Run installs, so the wire path is exercised against the
+			// real routing rules at least once.
+			backend := echoListenerOn("127.0.0.1:0")
+			DeferCleanup(func() { _ = backend.Close() })
+			port := portOf(backend)
+
+			frontend = newFakeFrontend(false)
+			start(func(c *TunnelConfig) {
+				c.Services = tunnelServices(&Config{
+					ServeAddr:   fmt.Sprintf("0.0.0.0:%d", port),
+					GRPCMaxPort: port,
+				}, "0.0.0.0:1")
+			})
+
+			stream, err := session().OpenStream(ctx)
+			Expect(err).ToNot(HaveOccurred())
+			// A host that is not this machine, and a port that is.
+			Expect(cluster.WriteStreamRequest(stream, cluster.StreamTagGRPC,
+				fmt.Sprintf("attacker.invalid:%d", port))).To(Succeed())
+
+			reply := awaitErr(func() error { return cluster.ReadStreamReply(stream) })
+			Eventually(reply, "10s").Should(Receive(BeNil()))
+
+			_, err = stream.Write([]byte("loopback"))
+			Expect(err).ToNot(HaveOccurred())
+			buf := make([]byte, len("loopback"))
+			read := awaitErr(func() error {
+				_, err := io.ReadFull(stream, buf)
+				return err
+			})
+			Eventually(read, "10s").Should(Receive(BeNil()))
+			Expect(string(buf)).To(Equal("loopback"))
+		})
+
+		It("refuses a port outside its range as a bad request, over the wire", func() {
+			frontend = newFakeFrontend(false)
+			start(func(c *TunnelConfig) {
+				c.Services = tunnelServices(&Config{
+					ServeAddr:   "0.0.0.0:50051",
+					GRPCMaxPort: 50051,
+				}, "0.0.0.0:50050")
+			})
+
+			stream, err := session().OpenStream(ctx)
+			Expect(err).ToNot(HaveOccurred())
+			Expect(cluster.WriteStreamRequest(stream, cluster.StreamTagGRPC, "127.0.0.1:22")).To(Succeed())
+
+			reply := awaitErr(func() error { return cluster.ReadStreamReply(stream) })
+			var got error
+			Eventually(reply, "10s").Should(Receive(&got))
+			// Three refusals, three meanings. A frontend retries unavailable
+			// and gives up on this one.
+			Expect(got).To(MatchError(cluster.ErrStreamRequestInvalid))
+			Expect(got).ToNot(MatchError(cluster.ErrStreamTargetUnavailable))
+			Expect(got).ToNot(MatchError(cluster.ErrStreamTagUnknown))
+		})
 	})
 
 	Describe("refusing a stream it cannot serve", func() {
@@ -507,3 +568,221 @@ var _ = Describe("Worker tunnel client", func() {
 		})
 	})
 })
+
+// echoListenerOn is echoListener bound to a specific address, so a spec can put
+// a listener somewhere the worker must NOT reach.
+func echoListenerOn(addr string) net.Listener {
+	ln, err := net.Listen("tcp", addr)
+	Expect(err).ToNot(HaveOccurred())
+	go func() {
+		for {
+			conn, err := ln.Accept()
+			if err != nil {
+				return
+			}
+			go func() {
+				defer func() { _ = conn.Close() }()
+				_, _ = io.Copy(conn, conn)
+			}()
+		}
+	}()
+	return ln
+}
+
+// portOf returns the port a listener bound to.
+func portOf(ln net.Listener) int {
+	_, portStr, err := net.SplitHostPort(ln.Addr().String())
+	Expect(err).ToNot(HaveOccurred())
+	port, err := strconv.Atoi(portStr)
+	Expect(err).ToNot(HaveOccurred())
+	return port
+}
+
+// The routing table is the security boundary of the whole tunnel, and until now
+// nothing exercised it: every spec above installs dialLocalTCP, which is exactly
+// the permissive dialler loopbackService exists to prevent. A review turned
+// loopbackService into an arbitrary-host dialler and all 131 specs passed.
+var _ = Describe("Worker tunnel local services", func() {
+	var ctx context.Context
+
+	BeforeEach(func() { ctx = context.Background() })
+
+	Describe("loopbackService", func() {
+		It("reaches a loopback listener whose port is in range", func() {
+			ln := echoListenerOn("127.0.0.1:0")
+			DeferCleanup(func() { _ = ln.Close() })
+			port := portOf(ln)
+
+			conn, err := loopbackService(port, port)(ctx, ln.Addr().String())
+			Expect(err).ToNot(HaveOccurred())
+			DeferCleanup(func() { _ = conn.Close() })
+
+			_, err = conn.Write([]byte("hi"))
+			Expect(err).ToNot(HaveOccurred())
+			buf := make([]byte, 2)
+			_, err = io.ReadFull(conn, buf)
+			Expect(err).ToNot(HaveOccurred())
+			Expect(string(buf)).To(Equal("hi"))
+		})
+
+		It("ignores the host the frontend names and dials loopback anyway", func() {
+			ln := echoListenerOn("127.0.0.1:0")
+			DeferCleanup(func() { _ = ln.Close() })
+			port := portOf(ln)
+
+			// A host that is emphatically not this machine. If it were honoured
+			// the dial would fail or, far worse, succeed against something else.
+			conn, err := loopbackService(port, port)(ctx, fmt.Sprintf("attacker.invalid:%d", port))
+			Expect(err).ToNot(HaveOccurred())
+			DeferCleanup(func() { _ = conn.Close() })
+			Expect(conn.RemoteAddr().String()).To(Equal(ln.Addr().String()))
+		})
+
+		It("does not reach a listener on another local address the frontend names", func() {
+			// The SSRF proof, stated as a reachability fact rather than as a
+			// property of the code. The only listener is on 127.0.0.2; nothing
+			// is on 127.0.0.1 at that port. A service that honoured the named
+			// host would connect; one that dials loopback cannot.
+			victim, err := net.Listen("tcp", "127.0.0.2:0")
+			if err != nil {
+				Skip("this host cannot bind a second loopback address: " + err.Error())
+			}
+			DeferCleanup(func() { _ = victim.Close() })
+			port := portOf(victim)
+
+			conn, err := loopbackService(port, port)(ctx, victim.Addr().String())
+			if err == nil {
+				_ = conn.Close()
+				Fail("the worker reached a host the frontend named, so a stream can steer it off loopback")
+			}
+			Expect(err).To(HaveOccurred())
+		})
+
+		DescribeTable("refuses a target it will not route",
+			func(target string, minPort, maxPort int) {
+				_, err := loopbackService(minPort, maxPort)(ctx, target)
+				Expect(err).To(HaveOccurred())
+				// Invalid, not unavailable. No retry brings a port outside this
+				// worker's own allocator range into it, and telling a frontend
+				// to retry forever is how a refusal becomes a hang.
+				Expect(err).To(MatchError(cluster.ErrStreamRequestInvalid))
+				Expect(err).ToNot(MatchError(cluster.ErrStreamTargetUnavailable))
+			},
+			Entry("a port below the range", "127.0.0.1:50050", 50051, 50060),
+			Entry("a port above the range", "127.0.0.1:50061", 50051, 50060),
+			Entry("a non-numeric port", "127.0.0.1:http", 50051, 50060),
+			Entry("no port at all", "127.0.0.1", 50051, 50060),
+			Entry("an empty target", "", 50051, 50060),
+		)
+
+		It("reports a backend that is not listening as unavailable, which a frontend may retry", func() {
+			// The other half of the taxonomy: a port IN range with nothing on
+			// it is a backend that has not started yet, not a bad request.
+			ln := echoListenerOn("127.0.0.1:0")
+			port := portOf(ln)
+			Expect(ln.Close()).To(Succeed())
+
+			_, err := loopbackService(port, port)(ctx, ln.Addr().String())
+			Expect(err).To(HaveOccurred())
+			Expect(classifyServiceFailure(err)).To(MatchError(cluster.ErrStreamTargetUnavailable))
+			Expect(classifyServiceFailure(err)).ToNot(MatchError(cluster.ErrStreamRequestInvalid))
+		})
+	})
+
+	Describe("fixedService", func() {
+		It("reaches its own address whatever the frontend names", func() {
+			ln := echoListenerOn("127.0.0.1:0")
+			DeferCleanup(func() { _ = ln.Close() })
+
+			conn, err := fixedService(ln.Addr().String())(ctx, "attacker.invalid:9")
+			Expect(err).ToNot(HaveOccurred())
+			DeferCleanup(func() { _ = conn.Close() })
+			Expect(conn.RemoteAddr().String()).To(Equal(ln.Addr().String()))
+		})
+	})
+
+	DescribeTable("loopbackAddr rewrites a bind address into a dialable one",
+		func(bind, want string) {
+			Expect(loopbackAddr(bind)).To(Equal(want))
+		},
+		// Dialling 0.0.0.0 only accidentally reaches localhost, and not on
+		// every platform, so the wildcard is replaced rather than dialled.
+		Entry("IPv4 wildcard", "0.0.0.0:8080", "127.0.0.1:8080"),
+		Entry("IPv6 wildcard", "[::]:8080", "127.0.0.1:8080"),
+		Entry("no host", ":8080", "127.0.0.1:8080"),
+		Entry("an explicit host is left alone", "10.0.0.9:8080", "10.0.0.9:8080"),
+		Entry("an explicit loopback is left alone", "127.0.0.1:8080", "127.0.0.1:8080"),
+		Entry("something that is not host:port passes through", "not-an-address", "not-an-address"),
+	)
+
+	Describe("tunnelServices", func() {
+		// The table Run installs. Built by its own function precisely so this
+		// can be asserted without starting a worker.
+		It("serves exactly the two tags the frontend may name", func() {
+			cfg := &Config{ServeAddr: "0.0.0.0:50051"}
+			Expect(tunnelServices(cfg, "0.0.0.0:50050")).To(HaveLen(2))
+			Expect(tunnelServices(cfg, "0.0.0.0:50050")).To(HaveKey(cluster.StreamTagGRPC))
+			Expect(tunnelServices(cfg, "0.0.0.0:50050")).To(HaveKey(cluster.StreamTagHTTP))
+		})
+
+		It("bounds the gRPC service by THIS worker's configured port range", func() {
+			cfg := &Config{ServeAddr: "0.0.0.0:50051", GRPCMaxPort: 50052}
+			svc := tunnelServices(cfg, "0.0.0.0:50050")[cluster.StreamTagGRPC]
+
+			// The HTTP server's own port sits one below the base port, so a
+			// gRPC-tagged stream cannot be steered onto it.
+			_, err := svc(ctx, "127.0.0.1:50050")
+			Expect(err).To(MatchError(cluster.ErrStreamRequestInvalid))
+			_, err = svc(ctx, "127.0.0.1:50053")
+			Expect(err).To(MatchError(cluster.ErrStreamRequestInvalid))
+		})
+
+		// Pins that the HTTP service reaches the address Run configures. It
+		// does NOT pin the wildcard rewrite: on Linux dialling 0.0.0.0 reaches
+		// loopback anyway, so this spec stays green with loopbackAddr disabled.
+		// The loopbackAddr table above is what holds that, and it exists
+		// because the accident is not portable.
+		It("points the HTTP service at the worker's own server", func() {
+			ln := echoListenerOn("127.0.0.1:0")
+			DeferCleanup(func() { _ = ln.Close() })
+
+			cfg := &Config{ServeAddr: "0.0.0.0:50051"}
+			svc := tunnelServices(cfg, fmt.Sprintf("0.0.0.0:%d", portOf(ln)))[cluster.StreamTagHTTP]
+
+			conn, err := svc(ctx, "ignored:1")
+			Expect(err).ToNot(HaveOccurred())
+			DeferCleanup(func() { _ = conn.Close() })
+			Expect(conn.RemoteAddr().String()).To(Equal(ln.Addr().String()))
+		})
+	})
+
+	DescribeTable("tunnelEndpoint builds the URL the worker dials",
+		func(frontendURL, nodeID, want string) {
+			got, err := tunnelEndpoint(frontendURL, nodeID)
+			Expect(err).ToNot(HaveOccurred())
+			Expect(got).To(Equal(want))
+		},
+		Entry("http becomes ws", "http://frontend:8080", "n1", "ws://frontend:8080/api/cluster/connect?id=n1"),
+		Entry("https becomes wss", "https://frontend", "n1", "wss://frontend/api/cluster/connect?id=n1"),
+		Entry("ws passes through", "ws://frontend:8080", "n1", "ws://frontend:8080/api/cluster/connect?id=n1"),
+		Entry("wss passes through", "wss://frontend", "n1", "wss://frontend/api/cluster/connect?id=n1"),
+		// A frontend behind a path prefix keeps it: the path is appended, not
+		// assigned, exactly as the registration client builds its URLs.
+		Entry("a path prefix is kept", "https://host/localai", "n1", "wss://host/localai/api/cluster/connect?id=n1"),
+		Entry("a trailing slash is not doubled", "https://host/localai/", "n1", "wss://host/localai/api/cluster/connect?id=n1"),
+		Entry("the node id is escaped", "http://h", "a b&c", "ws://h/api/cluster/connect?id=a+b%26c"),
+	)
+
+	DescribeTable("tunnelEndpoint refuses a frontend URL it cannot dial",
+		func(frontendURL string) {
+			_, err := tunnelEndpoint(frontendURL, "n1")
+			Expect(err).To(HaveOccurred())
+		},
+		Entry("empty", ""),
+		// Refused rather than coerced: a worker silently dialling a scheme
+		// nobody configured is worse than one that says it cannot start.
+		Entry("a scheme that is not HTTP", "ftp://frontend"),
+		Entry("a bare host with no scheme", "frontend:8080/x"),
+		Entry("no host", "http://"),
+	)
+})

core/services/worker/worker.go

@@ -16,7 +16,6 @@ import (
 	"github.com/mudler/LocalAI/core/config"
 	"github.com/mudler/LocalAI/core/gallery"
 
-	"github.com/mudler/LocalAI/core/services/cluster"
 	"github.com/mudler/LocalAI/core/services/messaging"
 	"github.com/mudler/LocalAI/core/services/nodes"
 	grpc "github.com/mudler/LocalAI/pkg/grpc"
@@ -108,10 +107,20 @@ func Run(ctx *cliContext.Context, cfg *Config) error {
 		}
 		nodeID = res.ID
 		// This path registers exactly once and never again, so the credential
-		// it holds cannot go stale by rotation from its own side. It can still
-		// be superseded from OUTSIDE, by a second worker registering under the
-		// same node name, and there is nothing this worker can do about that
-		// but log the 401 and keep retrying.
+		// it holds cannot go stale by rotation from its own side.
+		//
+		// It CAN be superseded from outside: Register upserts by NAME, so a
+		// second worker registering under this node's name rotates the row's
+		// credential, and this worker then fails every tunnel dial with 401 for
+		// the life of the process. It logs that once per backoff and never
+		// recovers on its own; a restart fixes it, because startup re-registers
+		// unconditionally.
+		//
+		// Deliberately NOT fixed here. Re-registering after repeated tunnel
+		// 401s is a decision about the worker's lifecycle, and it belongs with
+		// the change that removes this worker's inbound listeners, when a
+		// worker that cannot tunnel is a worker that cannot be reached at all.
+		// Today it can still be reached at the addresses it advertises.
 		staticTunnelToken := res.TunnelToken
 		tunnelToken = func() string { return staticTunnelToken }
 		connectNats = func() (*messaging.Client, error) {
@@ -191,19 +200,15 @@ func Run(ctx *cliContext.Context, cfg *Config) error {
 	// frontend URL or this node's identity is unusable, and a worker that
 	// silently ran without its tunnel would look healthy while being
 	// unreachable to everything that dials through it.
-	tunnelBasePort := cfg.effectiveBasePort()
 	if cfg.WorkerTunnel {
 		tunnel, terr := StartTunnel(shutdownCtx, TunnelConfig{
 			FrontendURL: cfg.RegisterTo,
 			NodeID:      nodeID,
 			Token:       tunnelToken,
-			Services: map[string]LocalService{
-				// The frontend names a backend process by its port; the worker
-				// decides that only its own loopback, and only within its own
-				// backend port range, is reachable through it.
-				cluster.StreamTagGRPC: loopbackService(tunnelBasePort, cfg.effectiveMaxPort(tunnelBasePort)),
-				cluster.StreamTagHTTP: fixedService(loopbackAddr(httpAddr)),
-			},
+			// Built by tunnelServices rather than inline, so the routing
+			// table, which is this feature's security boundary, is reachable
+			// from a spec without starting a worker.
+			Services: tunnelServices(cfg, httpAddr),
 		})
 		if terr != nil {
 			nodes.ShutdownFileTransferServer(httpServer)
@@ -252,7 +257,7 @@ func Run(ctx *cliContext.Context, cfg *Config) error {
 	}()
 
 	// Process supervisor — manages multiple backend gRPC processes on different ports
-	basePort := tunnelBasePort
+	basePort := cfg.effectiveBasePort()
 	// Buffered so NATS stop handler can send without blocking
 	sigCh := make(chan os.Signal, 1)
 	signal.Notify(sigCh, syscall.SIGINT, syscall.SIGTERM)

docs/content/features/distributed-mode.md

@@ -116,7 +116,9 @@ A worker that presents a credential belonging to no node, or names a node ID the
 
 Unlike the agent worker's API key and its NATS credential, a tunnel credential **is** issued to a node still awaiting approval. It is inert until then: the tunnel route re-reads the node's status on every dial and refuses a pending one. Withholding it would instead strand workers that register exactly once, since approval on its own prompts no re-registration.
 
-A tunnel credential does not replace `LOCALAI_REGISTRATION_TOKEN`. Without one, node registration itself is unauthenticated, so anyone who can reach the frontend can register a worker and be issued a working tunnel credential for it. LocalAI warns about that at startup.
+A tunnel credential does not replace `LOCALAI_REGISTRATION_TOKEN`. Without one, node registration itself is unauthenticated, so anyone who can reach the frontend can register a worker and be issued a tunnel credential for it. How far that gets them depends on auto-approve: with auto-approve on the node is healthy at once and the credential works immediately; with it off the node is pending and the credential is inert until an admin approves, so approval is the real gate. LocalAI warns about the missing token at startup.
+
+Only **backend** nodes are issued one. An agent worker has no inbound surface for the tunnel to replace and no client for it, so minting one would widen the credential surface for nothing; its row keeps an empty tunnel credential and the tunnel route refuses it like any other node without one.
 
 The tunnel lands on exactly one frontend replica, and that replica records itself as the owner of the worker's connection in the `node_connections` table. When the socket dies the claim is dropped with it. If the replica stalls long enough for its peers to reap it, it re-claims the tunnels it still holds on a live session as soon as it re-registers, skipping any whose socket has already gone. That re-claim needs the replica to have an advertised address: without one it never had an instance row to begin with, and its tunnels are usable only by the replica holding them.
 
@@ -135,7 +137,9 @@ The worker holds the tunnel with one goroutine: it dials, serves the frontend's
 | `grpc` | a backend process on this worker | the port; the host is discarded and only `127.0.0.1` is dialled, within the worker's own backend port range |
 | `http` | the worker's own file-transfer and backend-log server | ignored; there is one such server and only the worker knows where it bound |
 
-A stream naming a tag the worker does not serve, or a local service it could not reach, is refused with a reason and the stream is **ended** rather than left open. Those two refusals are distinct on the wire on purpose: a frontend gives up on the first and may retry the second. One bad stream never affects the others or the session.
+The `grpc` row is the security boundary of the tunnel, and it is worth being explicit about it. A tunnel terminates inside the worker process, so a stream arriving on it can reach anything the worker can reach; if the frontend could name the host, whoever holds the frontend end could make every worker in the fleet dial arbitrary addresses on its private network. The worker therefore builds the dial address from a constant `127.0.0.1` and a port it has validated, and the string from the wire never reaches the dialler at all. The port range is the one the worker's own allocator hands to backend processes, which by default runs to 65535; setting `LOCALAI_GRPC_MAX_PORT` narrows the allocator and this range together, and a worker with a known backend count should set it.
+
+A stream naming a tag the worker does not serve, a target outside that port range, or a local service it could not reach, is refused with a reason and the stream is **ended** rather than left open. Those refusals are distinct on the wire on purpose: an unknown tag and an out-of-range target are requests this worker will never serve, while an unreachable local service is a backend that has not started yet. A frontend gives up on the first two and may retry the third. One bad stream never affects the others or the session.
 
 Reconnects use exponential backoff with jitter: the interval doubles from 500ms up to a ceiling of 30 seconds, and each wait is drawn between half of that interval and all of it, so no worker ever spins and a fleet that lost the same replica does not come back in lockstep. The interval returns to its floor only after a session that lasted at least 30 seconds. That last part is what stops a rolling frontend restart, where every dial succeeds and then dies moments later, from turning a fleet of workers into a retry storm against the first replica back up. A worker that is refused (`401`, `403`) keeps retrying on the same schedule rather than exiting: a re-registration or an admin approval fixes both without restarting it.