diff --git a/services/lmstudio-proxy/proxy.go b/services/lmstudio-proxy/proxy.go index 6e3ed88df..8ea829b13 100644 --- a/services/lmstudio-proxy/proxy.go +++ b/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() { diff --git a/services/lmstudio-proxy/unpinned_peer_test.go b/services/lmstudio-proxy/unpinned_peer_test.go new file mode 100644 index 000000000..26e180241 --- /dev/null +++ b/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) + } +} diff --git a/services/nvpair-engine-manager/README.md b/services/nvpair-engine-manager/README.md index dc6681069..d3fd4fbba 100644 --- a/services/nvpair-engine-manager/README.md +++ b/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`, diff --git a/services/nvpair-engine-manager/remoteclient.go b/services/nvpair-engine-manager/remoteclient.go index 9f72ae174..681b0d319 100644 --- a/services/nvpair-engine-manager/remoteclient.go +++ b/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) } diff --git a/services/nvpair-engine-manager/remoteclient_bounds_test.go b/services/nvpair-engine-manager/remoteclient_bounds_test.go new file mode 100644 index 000000000..75fad9d13 --- /dev/null +++ b/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) + } +} diff --git a/services/nvpair-errors/README.md b/services/nvpair-errors/README.md index 4cd3d9883..6b10eac24 100644 --- a/services/nvpair-errors/README.md +++ b/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, diff --git a/services/nvpair-errors/httpserver.go b/services/nvpair-errors/httpserver.go index d4d46ac50..f92da9617 100644 --- a/services/nvpair-errors/httpserver.go +++ b/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 diff --git a/services/nvpair-errors/ingest_bounds_test.go b/services/nvpair-errors/ingest_bounds_test.go new file mode 100644 index 000000000..a7dd95479 --- /dev/null +++ b/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) + } + }) + } +} diff --git a/services/nvpair-manual-nodes/README.md b/services/nvpair-manual-nodes/README.md index 682e2a9df..9e5cdbf40 100644 --- a/services/nvpair-manual-nodes/README.md +++ b/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 diff --git a/services/nvpair-manual-nodes/manager.go b/services/nvpair-manual-nodes/manager.go index c6b1a23cd..7ccf25e43 100644 --- a/services/nvpair-manual-nodes/manager.go +++ b/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, diff --git a/services/nvpair-manual-nodes/response_bounds_test.go b/services/nvpair-manual-nodes/response_bounds_test.go new file mode 100644 index 000000000..8e87b0f7f --- /dev/null +++ b/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) + } + }) + } +} diff --git a/services/nvpair-node-scanner/README.md b/services/nvpair-node-scanner/README.md index 497d7d4dd..ea83b6dd7 100644 --- a/services/nvpair-node-scanner/README.md +++ b/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 diff --git a/services/nvpair-node-scanner/daemon.go b/services/nvpair-node-scanner/daemon.go index efadb3aff..117cf76d8 100644 --- a/services/nvpair-node-scanner/daemon.go +++ b/services/nvpair-node-scanner/daemon.go @@ -246,6 +246,15 @@ const engineNativeRefreshConcurrency = 16 // what the scanner is willing to parse. const engineNativeResponseLimit = 1 << 20 +// nodeInfoResponseLimit and modelsResponseLimit cap how much of a node-info +// or model-inventory answer is parsed. Both arrive from peers over the +// network; a value that extends past its cap fails enrichment and leaves the +// last-good cache intact. +const ( + nodeInfoResponseLimit = 1 << 20 + modelsResponseLimit = 16 << 20 +) + // telemetryRefreshConcurrency limits simultaneous node-info GETs across every // telemetry tick. const telemetryRefreshConcurrency = 16 @@ -1386,7 +1395,7 @@ func (d *daemon) fetchNodeInfoWithin(ctx context.Context, ip string, port int) ( return NodeInfoResponse{}, false } var info NodeInfoResponse - if err := json.NewDecoder(resp.Body).Decode(&info); err != nil { + if err := json.NewDecoder(io.LimitReader(resp.Body, nodeInfoResponseLimit)).Decode(&info); err != nil { return NodeInfoResponse{}, false } return info, true @@ -1489,7 +1498,7 @@ func (d *daemon) fetchModels(ip string, port int, clusterUUID string) ([]string, ModelsByEngine map[string][]string `json:"modelsByEngine"` LoadedByEngine map[string][]string `json:"loadedByEngine"` } - if err := json.NewDecoder(resp.Body).Decode(&body); err != nil { + if err := json.NewDecoder(io.LimitReader(resp.Body, modelsResponseLimit)).Decode(&body); err != nil { return nil, nil, nil, false } return body.Models, body.ModelsByEngine, body.LoadedByEngine, true diff --git a/services/nvpair-node-scanner/response_bounds_test.go b/services/nvpair-node-scanner/response_bounds_test.go new file mode 100644 index 000000000..bedf7862a --- /dev/null +++ b/services/nvpair-node-scanner/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 peer's node-info is parsed only up to nodeInfoResponseLimit. One whose JSON +// value extends past it fails enrichment rather than being buffered. +func TestFetchNodeInfoRejectsAnOversizedAnswer(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) + } + d := &daemon{http: srv.Client()} + if _, ok := d.fetchNodeInfo(host, port); ok != tc.want { + t.Fatalf("fetchNodeInfo ok = %v, want %v", ok, tc.want) + } + }) + } +} diff --git a/services/ollama-proxy/proxy.go b/services/ollama-proxy/proxy.go index c814fc5fb..48046fdc9 100644 --- a/services/ollama-proxy/proxy.go +++ b/services/ollama-proxy/proxy.go @@ -971,11 +971,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 { @@ -985,6 +985,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() { diff --git a/services/ollama-proxy/unpinned_peer_test.go b/services/ollama-proxy/unpinned_peer_test.go new file mode 100644 index 000000000..26e180241 --- /dev/null +++ b/services/ollama-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) + } +} diff --git a/services/omlx-proxy/proxy.go b/services/omlx-proxy/proxy.go index e0f28bd3c..d7d3f92b0 100644 --- a/services/omlx-proxy/proxy.go +++ b/services/omlx-proxy/proxy.go @@ -864,11 +864,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 { @@ -878,6 +878,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() { diff --git a/services/omlx-proxy/unpinned_peer_test.go b/services/omlx-proxy/unpinned_peer_test.go new file mode 100644 index 000000000..26e180241 --- /dev/null +++ b/services/omlx-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) + } +} diff --git a/services/versions.json b/services/versions.json index 68e393bba..f85176275 100644 --- a/services/versions.json +++ b/services/versions.json @@ -3,19 +3,19 @@ "product": "2.6.0", "installer": "2.6.0", "components": { - "ollama-proxy": "1.4.0", - "lmstudio-proxy": "1.4.0", + "ollama-proxy": "1.4.1", + "lmstudio-proxy": "1.4.1", "nvpair-node-info": "1.3.0", - "nvpair-node-scanner": "1.2.0", - "nvpair-manual-nodes": "1.2.1", + "nvpair-node-scanner": "1.2.1", + "nvpair-manual-nodes": "1.2.2", "nvpair-workload-manager": "1.3.0", - "nvpair-errors": "1.1.0", + "nvpair-errors": "1.1.1", "nvpair-node-settings": "3.1.0", "nvpair-ui-broker": "2.4.1", "nvpair-engine-manager": "1.4.0", "nvpair-cluster-manager": "2.1.1", "nvpair-job-scheduler": "1.1.0", "nvpair-tui": "1.2.0", - "omlx-proxy": "1.4.1" + "omlx-proxy": "1.4.2" } }