Bound JSON reads from peers and fail closed when a peer's pin is gone

services/lmstudio-proxy/proxy.go

@@ -816,11 +816,11 @@ func (p *Proxy) peerHTTPTransport(peerUUID string) *http.Transport {
 		delete(p.peerTransports, peerUUID)
 	}
 	if p.mesh == nil {
-		return newProxyTransport(nil)
+		return unpinnedPeerTransport()
 	}
 	cfg, ok := p.mesh.ClientTLSConfig(peerUUID)
 	if !ok {
-		return newProxyTransport(nil)
+		return unpinnedPeerTransport()
 	}
 	tr := newProxyTransport(cfg)
 	if p.peerTransports == nil {
@@ -830,6 +830,21 @@ func (p *Proxy) peerHTTPTransport(peerUUID string) *http.Transport {
 	return tr
 }
 
+// errPeerUnpinned is the dial error of a fail-closed peer transport.
+var errPeerUnpinned = stderrors.New("peer is not a pinned cluster member")
+
+// unpinnedPeerTransport fails every dial. A peer candidate is resolved only
+// while its pin is held, but the pin can go between resolution and dial; the
+// peer is then refused outright rather than dialed with a transport that does
+// not pin its certificate.
+func unpinnedPeerTransport() *http.Transport {
+	return &http.Transport{
+		DialContext: func(context.Context, string, string) (net.Conn, error) {
+			return nil, errPeerUnpinned
+		},
+	}
+}
+
 // dropUnpinnedPeerTransports closes idle conns for peer Transports whose pins
 // are gone. Safe to call from the mesh Watch callback.
 func (p *Proxy) dropUnpinnedPeerTransports() {

services/lmstudio-proxy/unpinned_peer_test.go

@@ -0,0 +1,27 @@
+// SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved.
+// SPDX-License-Identifier: Apache-2.0
+
+package main
+
+import (
+	stderrors "errors"
+	"net/http"
+	"testing"
+)
+
+// A peer whose pin is gone is refused at dial time, never reached over a
+// transport that does not pin its certificate.
+func TestPeerTransportFailsClosedWithoutAPin(t *testing.T) {
+	p := &Proxy{}
+	req, err := http.NewRequest(http.MethodGet, "https://192.0.2.1:11434/api/tags", nil)
+	if err != nil {
+		t.Fatal(err)
+	}
+	resp, err := p.peerHTTPTransport("peer-without-pin").RoundTrip(req)
+	if resp != nil {
+		_ = resp.Body.Close()
+	}
+	if !stderrors.Is(err, errPeerUnpinned) {
+		t.Fatalf("err = %v, want errPeerUnpinned", err)
+	}
+}

services/nvpair-engine-manager/README.md

@@ -271,7 +271,9 @@ directory (fed by its own `discovery:subscribe{services:[ec]}` to the broker
 relay), dials that peer's `ec` surface over pin-based cluster mTLS, and — for
 `remote-install` / `remote-install-tool` / `remote-pull-model` — mints an `opId`,
 relays each streamed progress frame up as `engine:remote-progress`, and settles
-the request with the terminal result. They fail if this node isn't clustered or the target isn't a
+the request with the terminal result. Each NDJSON progress or terminal frame is
+limited to 1 MiB; the limit applies per frame so a long operation can report
+progress throughout its lifetime. They fail if this node isn't clustered or the target isn't a
 pinned cluster peer. See "Remote engine management" below.
 
 `engine:install`, `engine:start`, `engine:stop`, `engine:restart`,

services/nvpair-engine-manager/remoteclient.go

@@ -10,6 +10,7 @@ package main
 // server side enforces with a 403.
 
 import (
+	"bufio"
 	"bytes"
 	"context"
 	"encoding/json"
@@ -25,6 +26,8 @@ import (
 const (
 	remoteResponseHeaderTimeout      = 30 * time.Second
 	remoteReadyResponseHeaderTimeout = 11 * time.Minute
+	// remoteStreamFrameBytes caps one NDJSON progress or result frame.
+	remoteStreamFrameBytes = 1 << 20
 )
 
 // remoteFrameError preserves an optional JSON-RPC code from an ec error frame.
@@ -183,15 +186,19 @@ func (c *remoteClient) stream(ctx context.Context, path string, body any, onProg
 		return streamFrame{}, fmt.Errorf("remote %s: HTTP %d: %s", path, resp.StatusCode, strings.TrimSpace(string(data)))
 	}
 
-	dec := json.NewDecoder(resp.Body)
+	// The limit applies per frame, so a long operation can report progress
+	// for as long as it runs while no single frame can grow without bound.
+	scanner := bufio.NewScanner(resp.Body)
+	scanner.Buffer(make([]byte, 4096), remoteStreamFrameBytes)
 	var result streamFrame
 	haveResult := false
-	for {
+	for scanner.Scan() {
+		line := bytes.TrimSpace(scanner.Bytes())
+		if len(line) == 0 {
+			continue
+		}
 		var f streamFrame
-		if derr := dec.Decode(&f); derr != nil {
-			if derr == io.EOF {
-				break
-			}
+		if derr := json.Unmarshal(line, &f); derr != nil {
 			return streamFrame{}, fmt.Errorf("remote %s: decode frame: %w", path, derr)
 		}
 		switch f.Type {
@@ -206,6 +213,9 @@ func (c *remoteClient) stream(ctx context.Context, path string, body any, onProg
 			return streamFrame{}, &remoteFrameError{Code: f.Code, Msg: f.Message}
 		}
 	}
+	if err := scanner.Err(); err != nil {
+		return streamFrame{}, fmt.Errorf("remote %s: read frame: %w", path, err)
+	}
 	if !haveResult {
 		return streamFrame{}, fmt.Errorf("remote %s: stream ended without a result", path)
 	}

services/nvpair-engine-manager/remoteclient_bounds_test.go

@@ -0,0 +1,51 @@
+// SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved.
+// SPDX-License-Identifier: Apache-2.0
+
+package main
+
+import (
+	"context"
+	"fmt"
+	"net/http"
+	"net/http/httptest"
+	"strings"
+	"testing"
+)
+
+func streamingPeer(t *testing.T, body string) *remoteClient {
+	t.Helper()
+	srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) {
+		w.Header().Set("Content-Type", "application/x-ndjson")
+		_, _ = fmt.Fprint(w, body)
+	}))
+	t.Cleanup(srv.Close)
+	return &remoteClient{http: srv.Client(), base: srv.URL}
+}
+
+// A peer cannot make this node buffer an unbounded frame: one line past the
+// per-frame cap fails the operation.
+func TestRemoteStreamRejectsAnOversizedFrame(t *testing.T) {
+	huge := `{"type":"progress","message":"` + strings.Repeat("x", remoteStreamFrameBytes) + `"}` + "\n"
+	c := streamingPeer(t, huge+`{"type":"result"}`+"\n")
+	_, err := c.stream(context.Background(), "/v1/control/install", map[string]string{}, nil)
+	if err == nil || !strings.Contains(err.Error(), "read frame") {
+		t.Fatalf("err = %v, want a read-frame error", err)
+	}
+}
+
+// The cap is per frame, not per stream: a long operation reports progress for
+// as long as it runs.
+func TestRemoteStreamAcceptsManySmallFrames(t *testing.T) {
+	var body strings.Builder
+	frame := `{"type":"progress","message":"` + strings.Repeat("x", 4096) + `"}` + "\n"
+	for body.Len() < 2*remoteStreamFrameBytes {
+		body.WriteString(frame)
+	}
+	body.WriteString(`{"type":"result","stage":"done"}` + "\n")
+	progress := 0
+	c := streamingPeer(t, body.String())
+	result, err := c.stream(context.Background(), "/v1/control/install", map[string]string{}, func(streamFrame) { progress++ })
+	if err != nil || result.Stage != "done" || progress == 0 {
+		t.Fatalf("result = %+v, err = %v, progress = %d; want done after progress", result, err, progress)
+	}
+}

services/nvpair-errors/README.md

@@ -258,6 +258,9 @@ producer --errors:report--> nvpair-errors A --POST /v1/errors--> nvpair-errors B
 — return this node's local-origin snapshot (the same body we push;
 handy for manual verification).
 
+Ingest JSON decoding reads at most 1 MiB. A snapshot whose JSON value extends
+past that limit returns `400` without reconciling any errors.
+
 Both require **pin-based cluster mTLS**. The caller must present a
 certificate this node currently pins; anything else is refused with
 `403`. There is no plain-HTTP personality: this is a cluster data plane,

services/nvpair-errors/httpserver.go

@@ -23,6 +23,7 @@ import (
 	"crypto/tls"
 	"encoding/json"
 	"fmt"
+	"io"
 	"log/slog"
 	"net"
 	"net/http"
@@ -73,10 +74,14 @@ func newErrorsMux(mgr *Manager, mesh *clustertrust.Mesh) *http.ServeMux {
 // peer reconcile our own origin (the manager rejects nodeId ==
 // localNodeID as a no-op, but we reject empty here so the client gets
 // a clear signal rather than a silent accept).
+// maxSyncBodyBytes caps a peer's pushed snapshot. A snapshot whose JSON value
+// extends past it is a 400 without reconciling any errors.
+const maxSyncBodyBytes = 1 << 20
+
 func handleIngest(w http.ResponseWriter, r *http.Request, mgr *Manager) {
 	defer r.Body.Close()
 	var env errors.SyncEnvelope
-	if err := json.NewDecoder(r.Body).Decode(&env); err != nil {
+	if err := json.NewDecoder(io.LimitReader(r.Body, maxSyncBodyBytes)).Decode(&env); err != nil {
 		slog.Warn("ingest: bad request body", "err", err)
 		http.Error(w, "invalid JSON body", http.StatusBadRequest)
 		return

services/nvpair-errors/ingest_bounds_test.go

@@ -0,0 +1,35 @@
+// SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved.
+// SPDX-License-Identifier: Apache-2.0
+
+package main
+
+import (
+	"net/http"
+	"net/http/httptest"
+	"strings"
+	"testing"
+)
+
+// A peer's snapshot is parsed only up to maxSyncBodyBytes; one that extends past
+// it is refused rather than buffered.
+func TestIngestRejectsAnOversizedSnapshot(t *testing.T) {
+	m, _ := newTestManager(t)
+	for _, tc := range []struct {
+		name    string
+		message string
+		want    int
+	}{
+		{"within the limit", "disk full", http.StatusNoContent},
+		{"past the limit", strings.Repeat("x", maxSyncBodyBytes), http.StatusBadRequest},
+	} {
+		t.Run(tc.name, func(t *testing.T) {
+			body := `{"nodeId":"peer-a","errors":[{"id":"e1","message":"` + tc.message + `","timestamp":1}]}`
+			req := httptest.NewRequest(http.MethodPost, "/v1/errors", strings.NewReader(body))
+			rec := httptest.NewRecorder()
+			handleIngest(rec, req, m)
+			if rec.Code != tc.want {
+				t.Fatalf("status %d, want %d", rec.Code, tc.want)
+			}
+		})
+	}
+}

services/nvpair-manual-nodes/README.md

@@ -7,6 +7,10 @@ SPDX-License-Identifier: Apache-2.0
 
 A Go service for managing manually configured nodes on networks where mDNS discovery is unavailable. Accepts node addresses via JSON-RPC, probes each for Ollama, LM Studio, and node-info, and emits status events.
 
+Node-info JSON decoding reads at most 1 MiB, and model-list decoding at most
+16 MiB. A JSON value that extends past its limit follows the existing probe or
+model-parse failure path.
+
 ## Worker control health
 
 An id-bearing `worker/health` request (no params) returns `{"ok":true}` with

services/nvpair-manual-nodes/manager.go

@@ -57,6 +57,13 @@ const (
 	// statement about the expected size.
 	engineNativeResponseLimit = 1 << 20
 
+	// nodeInfoResponseLimit, modelsResponseLimit and probeErrorResponseLimit
+	// cap what a manual node's answers can make this process parse. A value
+	// that extends past its cap follows the probe's existing failure path.
+	nodeInfoResponseLimit   = 1 << 20
+	modelsResponseLimit     = 16 << 20
+	probeErrorResponseLimit = 64 << 10
+
 	// probeFailThreshold is the number of CONSECUTIVE probes (at
 	// probeInterval = 10s, so 30 s of unreachability) where neither
 	// Ollama nor node-info answered before we surface the node as
@@ -1046,7 +1053,7 @@ func (m *Manager) fetchEngineManagerModels(client *http.Client, scheme, addr str
 		return false, engineManagerModels{}
 	}
 	var res engineManagerModels
-	if err := json.NewDecoder(resp.Body).Decode(&res); err != nil {
+	if err := json.NewDecoder(io.LimitReader(resp.Body, modelsResponseLimit)).Decode(&res); err != nil {
 		slog.Debug("manual probe engine-manager decode failed",
 			"addr", addr, "port", port, "scheme", scheme, "err", err,
 			"duration_ms", time.Since(start).Milliseconds())
@@ -1176,7 +1183,7 @@ func (m *Manager) probeLMStudio(addr string, port int) (bool, []string) {
 			ID string `json:"id"`
 		} `json:"data"`
 	}
-	if err := json.NewDecoder(resp.Body).Decode(&result); err != nil {
+	if err := json.NewDecoder(io.LimitReader(resp.Body, modelsResponseLimit)).Decode(&result); err != nil {
 		// Reachable, but the model list didn't parse — still report it up.
 		slog.Debug("manual probe lmstudio up (models parse failed)",
 			"addr", addr, "port", port, "err", err)
@@ -1212,7 +1219,7 @@ func (m *Manager) probeOMLX(addr string, port int) (bool, []string) {
 		var errResp struct {
 			Code string `json:"code"`
 		}
-		if err := json.NewDecoder(resp.Body).Decode(&errResp); err == nil && errResp.Code == "loopback-only" {
+		if err := json.NewDecoder(io.LimitReader(resp.Body, probeErrorResponseLimit)).Decode(&errResp); err == nil && errResp.Code == "loopback-only" {
 			slog.Debug("manual probe omlx proxy identified via loopback-only 403", "addr", addr, "port", port)
 			return true, nil
 		}
@@ -1229,7 +1236,7 @@ func (m *Manager) probeOMLX(addr string, port int) (bool, []string) {
 			ID string `json:"id"`
 		} `json:"data"`
 	}
-	if err := json.NewDecoder(resp.Body).Decode(&result); err != nil {
+	if err := json.NewDecoder(io.LimitReader(resp.Body, modelsResponseLimit)).Decode(&result); err != nil {
 		slog.Debug("manual probe omlx up (models parse failed)",
 			"addr", addr, "port", port, "err", err)
 		return true, nil
@@ -1286,7 +1293,7 @@ func (m *Manager) fetchOllamaModels(addr string, port int) []string {
 			Name string `json:"name"`
 		} `json:"models"`
 	}
-	if err := json.NewDecoder(resp.Body).Decode(&result); err != nil {
+	if err := json.NewDecoder(io.LimitReader(resp.Body, modelsResponseLimit)).Decode(&result); err != nil {
 		return nil
 	}
 	names := make([]string, 0, len(result.Models))
@@ -1348,13 +1355,13 @@ func (m *Manager) probeNodeInfoWithPoll(ctx context.Context, client *http.Client
 	}
 
 	var result NodeInfoResponse
-	if err := json.NewDecoder(resp.Body).Decode(&result); err != nil {
+	if err := json.NewDecoder(io.LimitReader(resp.Body, nodeInfoResponseLimit)).Decode(&result); err != nil {
 		slog.Debug("manual probe node-info decode failed",
 			"addr", addr, "port", port, "scheme", scheme, "err", err,
 			"duration_ms", time.Since(start).Milliseconds())
 		return false, NodeInfoResponse{}
 	}
-	_, _ = io.Copy(io.Discard, resp.Body)
+	_, _ = io.Copy(io.Discard, io.LimitReader(resp.Body, probeErrorResponseLimit))
 	slog.Debug("manual probe node-info up",
 		"addr", addr, "port", port, "scheme", scheme, "gpus", len(result.GPUs),
 		"has_cpu", result.CPU != nil, "has_memory", result.Memory != nil,

services/nvpair-manual-nodes/response_bounds_test.go

@@ -0,0 +1,46 @@
+// SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved.
+// SPDX-License-Identifier: Apache-2.0
+
+package main
+
+import (
+	"fmt"
+	"net"
+	"net/http"
+	"net/http/httptest"
+	"strconv"
+	"strings"
+	"testing"
+)
+
+// A manual node's node-info is parsed only up to nodeInfoResponseLimit. One
+// whose JSON value extends past it follows the probe's failure path.
+func TestProbeNodeInfoRejectsAnOversizedAnswer(t *testing.T) {
+	for _, tc := range []struct {
+		name    string
+		padding int
+		want    bool
+	}{
+		{"within the limit", 16, true},
+		{"past the limit", nodeInfoResponseLimit, false},
+	} {
+		t.Run(tc.name, func(t *testing.T) {
+			srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) {
+				_, _ = fmt.Fprintf(w, `{"hostUuid":"peer-uuid","hostname":"%s"}`, strings.Repeat("x", tc.padding))
+			}))
+			defer srv.Close()
+			host, portStr, err := net.SplitHostPort(srv.Listener.Addr().String())
+			if err != nil {
+				t.Fatal(err)
+			}
+			port, err := strconv.Atoi(portStr)
+			if err != nil {
+				t.Fatal(err)
+			}
+			m, _, _ := newTestManager()
+			if up, _ := m.probeNodeInfo(srv.Client(), "http", host, port); up != tc.want {
+				t.Fatalf("probeNodeInfo up = %v, want %v", up, tc.want)
+			}
+		})
+	}
+}

services/nvpair-node-scanner/README.md

@@ -12,6 +12,9 @@ Concretely, the daemon:
 - **Advertises** this node's one `_nvpair-node._tcp` record, built from a registry of the local services its parent (the broker) registers — `ni`/`ol`/`lm`/`er`/`wl`/`cl`/`em`/`ec` ports plus the node identity (`uuid=`, `cluster-uuid=` when clustered, `ip=`). One record per node covers every service.
 - **Browses** `_nvpair-node._tcp` for every node on the LAN (including itself) and maintains a queryable directory keyed by host UUID.
 - **Enriches** each discovered node: GPU/CPU/memory inventory from its `ni` (node-info) port over plain HTTP, and the model list from its `em` (engine-manager) port (`GET /v1/models`) over cluster mTLS — a peer's model inventory is cluster data, so it is fetched only when this node holds a pin for the principal that peer advertises, while this node's own list is read over loopback in plaintext — the flat union, the per-engine breakdown (`modelsByEngine`), and the per-engine set of models loaded in memory (`loadedByEngine`), all carried onto the node's directory entry so remote cards can show loaded state. Each enrichment has a last-good cache so a transient fetch miss doesn't blank the node's card. Node-info enrichment can optionally be moved onto HTTPS (see the TLS flags below) — off by default and gated only on operator flags, never inferred per-node.
+  Node-info JSON decoding reads at most 1 MiB, and model inventory decoding reads
+  at most 16 MiB. A JSON value that extends past its limit fails enrichment and
+  leaves the last-good cache intact.
 - **Carries the engine's own view** of every node: the engine-native counters and rates the node's `em` (engine-manager) port publishes at `GET /v1/engine-stats`, read over the same transport as the model list (cluster mTLS for a pinned peer, loopback in plaintext for this node's own engine-manager) and folded onto the node's directory entry as `engineNative`. It runs on its own, slower cadence (a minute) than the model sweep, because the engine's counters move slowly, and it reuses the same per-peer client pool, so it costs no extra connections to a peer. Every sweep is a new observation, so the view's `observedAt` stamp advances even when the engine's counters do not move, and a client can compute staleness as `now - observedAt`. A node with no engine-native view is a normal state, not a failure: a peer this node holds no pin for, an engine with no admin API, and an unreachable engine all leave the field absent. A node that has never produced a view has none; a node whose fetch fails keeps the view it had (the last-good rule the model list follows too), and a view that is already stored survives the mDNS browse path replacing the record.
 - **Samples scheduling telemetry** from healthy nodes every two seconds with
   stable per-node jitter. Consecutive remote failures back off to 4, 8, 16, then