diff --git a/core/http/endpoints/localai/nodes.go b/core/http/endpoints/localai/nodes.go index f674cb71a..0d3f3ebb3 100644 --- a/core/http/endpoints/localai/nodes.go +++ b/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 diff --git a/core/http/endpoints/localai/nodes_test.go b/core/http/endpoints/localai/nodes_test.go index 85a2600b6..dababff38 100644 --- a/core/http/endpoints/localai/nodes_test.go +++ b/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()) diff --git a/core/services/cluster/tunnelproto_test.go b/core/services/cluster/tunnelproto_test.go index 752fb9e9b..2664d9d50 100644 --- a/core/services/cluster/tunnelproto_test.go +++ b/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) { diff --git a/core/services/worker/tunnel.go b/core/services/worker/tunnel.go index 469bd47c8..0eabacb0f 100644 --- a/core/services/worker/tunnel.go +++ b/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.