fix(worker): make the tunnel credential's node-type gate actually structural

Re-review follow-up, three items. Two are the overclaiming-comment class again,
and the first is that class with a real defect underneath it.

attachTunnelToken said "enforcement is therefore structural": an ineligible node
never gets a credential, so its hash stays empty and the tunnel route's
empty-hash branch does the refusing. That was true for a node that had always
been an agent and false for one that had not. Register upserts by NAME, so a
backend node re-registering as an agent keeps its ID, and Register's struct
Updates zero-skips the credential column while writing the new node_type. The
early return left the credential the node earned as a backend sitting on a row
that is now an agent, and ConnectHandler never looks at node_type.

Fixed by making the claim true rather than by softening it, because the mint-site
gate was chosen precisely on the grounds that it was structural: an ineligible
node now has its column CLEARED, unconditionally, so the invariant does not
depend on what the row happened to contain. A spec pins it and was red before the
change. Same shape as the Register-upserts-by-name hazard already carried
forward: a name is not an identity.

Second, loopbackHost claimed to be the only host any tunnel stream is ever
dialled on. It is not: fixedService dials whatever Run built it from, which is
this worker's own LOCALAI_HTTP_ADDR, and loopbackAddr rewrites only a wildcard
bind, so an operator who binds the file-transfer server to a routable address
gets a routable dial. The property that matters is narrower and is what the
comment says now: the frontend cannot STEER the dial. The grpc tag builds its
address from a constant and a validated port with nothing from the wire reaching
the dialler, and the http tag ignores its target entirely. Worth stating exactly
rather than summarising, because the argument about what a stream can reach rests
on knowing which hosts are reachable, and an overstatement at that site is what
would let someone conclude the constant alone is doing the work.

Third, a spec named "without allocating it" measured no allocation. It now
asserts the mechanism the defence actually rests on, that the reader consumes the
two length bytes and not one byte of the body, through a counting reader. The
input carries a body on purpose: against input that ends after the header the
assertion would pass with the limit check deleted.

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

core/http/endpoints/localai/nodes.go

@@ -315,11 +315,21 @@ func ApproveNodeEndpoint(registry *nodes.NodeRegistry, authDB *gorm.DB, hmacSecr
 // "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.
+// It does not need to, PROVIDED an ineligible node ends up with no credential
+// rather than merely being handed no new one, because the handler's empty-hash
+// branch is what does the refusing. So this CLEARS the column instead of
+// returning early, and the difference is not theoretical: Register upserts by
+// NAME, so a backend node re-registering as an agent keeps its ID, and
+// Register's struct Updates zero-skips TunnelTokenHash while writing the new
+// node_type. An early return left a live credential on a row that had become an
+// agent. Clearing is what makes "enforcement is structural" true.
+//
+// It clears unconditionally rather than only when something is there, so the
+// invariant holds without depending on what the row happened to contain. The
+// cost is one UPDATE per agent registration.
+//
+// The day agent workers want a tunnel, relaxing the eligibility 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
@@ -328,7 +338,17 @@ 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 || node.NodeType != nodes.NodeTypeBackend {
+	if node == nil {
+		return
+	}
+	if node.NodeType != nodes.NodeTypeBackend {
+		// Cleared, not skipped. SetTunnelTokenHash writes the single column
+		// directly rather than through a struct update, so unlike Register it
+		// can write an empty value; see its doc.
+		if err := registry.SetTunnelTokenHash(ctx, node.ID, ""); err != nil {
+			xlog.Error("Failed to clear the tunnel credential of a node that is not a backend worker",
+				"node", node.Name, "type", node.NodeType, "error", err)
+		}
 		return
 	}
 	// crypto/rand.Text: at least 128 bits of randomness, no error to handle and

core/http/endpoints/localai/nodes_test.go

@@ -199,6 +199,34 @@ var _ = Describe("Node HTTP handlers", func() {
 			Expect(node.TunnelTokenHash).To(BeEmpty())
 		})
 
+		It("clears a tunnel credential when a node stops being a backend node", func() {
+			// Register upserts BY NAME, so a node can change node_type in place.
+			// Skipping the mint on the way through leaves the credential the
+			// node earned as a backend sitting on a row that is now an agent:
+			// Register's struct Updates zero-skips the column while writing the
+			// new node_type, so nothing else clears it. ConnectHandler never
+			// looks at node_type, so that stale hash is a usable tunnel
+			// credential for a node type that is not supposed to hold one.
+			//
+			// This is the same shape as the Register-upserts-by-name hazard
+			// already carried forward: a name is not an identity.
+			backend := register(`{"name":"shifty","address":"10.0.0.7:50051"}`, "", true)
+			Expect(backend["tunnel_token"]).ToNot(BeEmpty())
+
+			agent := register(`{"name":"shifty","node_type":"agent"}`, "", true)
+			Expect(agent["id"]).To(Equal(backend["id"]), "re-registration must keep the node identity")
+			Expect(agent["node_type"]).To(Equal(nodes.NodeTypeAgent))
+			Expect(agent).ToNot(HaveKey("tunnel_token"))
+
+			node, err := registry.Get(context.Background(), backend["id"].(string))
+			Expect(err).ToNot(HaveOccurred())
+			// The claim the gate makes is that an ineligible node HAS no
+			// credential, not merely that it was not handed a new one. Only
+			// then is the empty-hash refusal in ConnectHandler the enforcement.
+			Expect(node.TunnelTokenHash).To(BeEmpty(),
+				"the node kept the credential it earned as a backend, so the mint-site gate is not structural")
+		})
+
 		It("returns nats_jwt when account seed is configured", func() {
 			akp, err := nkeys.CreateAccount()
 			Expect(err).ToNot(HaveOccurred())

core/services/cluster/tunnelproto_test.go

@@ -59,12 +59,25 @@ var _ = Describe("Worker tunnel stream framing", func() {
 			Entry("containing a space", "grpc stream"),
 		)
 
-		It("refuses a frame that declares more than the limit without allocating it", func() {
+		It("refuses an over-long declared length after reading only the header", func() {
+			// The name used to say "without allocating it" and the spec
+			// measured nothing of the sort. What is actually checkable, and is
+			// the mechanism the defence rests on, is that the reader STOPS: it
+			// consumes the two length bytes and not one byte of the body, so a
+			// peer cannot make it allocate or read on demand.
+			//
+			// The body is present in the input on purpose. With an input that
+			// ends after the header, a reader that went on to read the body
+			// would still consume nothing more, and this assertion would pass
+			// with the limit check deleted.
 			var hdr [2]byte
 			binary.BigEndian.PutUint16(hdr[:], 65535)
-			_, _, err := cluster.ReadStreamRequest(bytes.NewReader(hdr[:]))
+			src := &countingReader{r: bytes.NewReader(append(hdr[:], make([]byte, 4096)...))}
+
+			_, _, err := cluster.ReadStreamRequest(src)
 			Expect(err).To(HaveOccurred())
 			Expect(err.Error()).To(ContainSubstring("over the"))
+			Expect(src.n).To(Equal(2), "the reader consumed part of a frame it had already refused")
 		})
 
 		It("reports a truncated frame as a truncated read, not as a refusal", func() {
@@ -176,6 +189,19 @@ type reasonErr struct {
 func (e *reasonErr) Error() string { return e.sentinel.Error() + ": " + e.text }
 func (e *reasonErr) Unwrap() error { return e.sentinel }
 
+// countingReader records how many bytes were actually consumed, so a spec can
+// assert where a reader stopped rather than only what it returned.
+type countingReader struct {
+	r io.Reader
+	n int
+}
+
+func (c *countingReader) Read(p []byte) (int, error) {
+	n, err := c.r.Read(p)
+	c.n += n
+	return n, err
+}
+
 // 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) {

core/services/worker/tunnel.go

@@ -619,9 +619,28 @@ func loopbackService(minPort, maxPort int) LocalService {
 	}
 }
 
-// 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.
+// loopbackHost is the host every stream the FRONTEND CAN STEER is dialled on.
+//
+// It is a constant so that "a stream cannot choose where the worker dials" is a
+// fact about the code rather than a claim about its inputs: the grpc tag builds
+// its address from this and a port it validated, and nothing derived from the
+// wire reaches the dialler.
+//
+// It is NOT the only host this file ever dials, and the difference is worth
+// stating exactly rather than summarising, because the whole argument about
+// what a stream can reach rests on knowing which hosts are reachable, and an
+// overstatement here is what would let a future reader conclude the constant
+// alone is doing the work.
+//
+// fixedService dials whatever address it was constructed with. Run constructs
+// it from this worker's own LOCALAI_HTTP_ADDR, which an operator may set to a
+// routable address; loopbackAddr only rewrites a WILDCARD bind, and leaves an
+// explicit host alone on purpose, because a server bound to one address is not
+// reachable on another. So the http tag can dial a non-loopback host. That host
+// is one the OPERATOR configured for this worker's own server, never one a
+// stream names: fixedService ignores its target entirely. The property the
+// design needs is that the frontend cannot steer the dial, and that holds for
+// both tags.
 const loopbackHost = "127.0.0.1"
 
 // tunnelServices builds the routing table the worker installs on its tunnel.