From b81ba7dd06c3dc0c10d741ea7187f4ae2aeeddd6 Mon Sep 17 00:00:00 2001 From: rcourtman Date: Fri, 24 Jul 2026 00:07:19 +0100 Subject: [PATCH] Migrate TrueNAS to supported JSON-RPC transport --- .../v6/internal/SOURCE_OF_TRUTH.md | 14 +- .../v6/internal/subsystems/api-contracts.md | 8 +- .../v6/internal/subsystems/monitoring.md | 61 +- .../v6/internal/subsystems/registry.json | 4 +- .../src/api/__tests__/truenas.test.ts | 24 +- frontend-modern/src/api/truenas.ts | 37 + .../CredentialSlots/TrueNASCredentialSlot.tsx | 72 +- .../useTrueNASSettingsPanelState.test.tsx | 6 +- .../Settings/useTrueNASSettingsPanelState.ts | 4 +- internal/api/platform_mock_connections.go | 15 +- internal/api/truenas_handlers.go | 6 +- internal/api/truenas_handlers_test.go | 18 + internal/monitoring/truenas_poller.go | 36 +- internal/monitoring/truenas_poller_test.go | 32 + internal/truenas/client.go | 701 ++++++++------- internal/truenas/client_test.go | 99 ++- internal/truenas/provider.go | 25 + internal/truenas/transport.go | 556 ++++++++++++ internal/truenas/transport_test.go | 798 ++++++++++++++++++ 19 files changed, 2145 insertions(+), 371 deletions(-) create mode 100644 internal/truenas/transport.go create mode 100644 internal/truenas/transport_test.go diff --git a/docs/release-control/v6/internal/SOURCE_OF_TRUTH.md b/docs/release-control/v6/internal/SOURCE_OF_TRUTH.md index 096bca66f..49f74808f 100644 --- a/docs/release-control/v6/internal/SOURCE_OF_TRUTH.md +++ b/docs/release-control/v6/internal/SOURCE_OF_TRUTH.md @@ -615,8 +615,13 @@ claim, broad same-shape TrueNAS iteration is not default work above this declared floor. 1. Architecture boundary: - TrueNAS is API-first. The unified agent may augment a TrueNAS system later, - but it is not required for bootstrap or baseline support. TrueNAS must + TrueNAS is API-first. SCALE 25.04 and later uses the supported versioned + JSON-RPC 2.0 WebSocket API at `/api/current`; REST is a connection-local, + version-gated compatibility boundary only for SCALE releases before 25.04 + and TrueNAS CORE/FreeNAS. Authentication, permission, TLS, protocol, and + method errors on the current endpoint fail closed and never downgrade to + REST. The unified agent may augment a TrueNAS system later, but it is not + required for bootstrap or baseline support. TrueNAS must project into the canonical `agent`, `vm`, `app-container`, `storage`, `physical-disk`, and recovery contracts. TrueNAS app rows may carry a TrueNAS-native app facet sourced from `app.query` / `active_workloads`, and @@ -627,7 +632,10 @@ declared floor. Supported now through the shared platform-connections flow and `/api/truenas/connections`. Operators can add, test, edit, retest, and delete stored TrueNAS connections without re-entering masked secrets on - ordinary edits. `truenas_disabled` is only an explicit server opt-out, not + ordinary edits. Current API-key setup includes the key owner's username for + the supported `auth.login_ex` flow; username-less stored keys remain an + upgrade bridge only while the appliance still exposes its deprecated login + method. `truenas_disabled` is only an explicit server opt-out, not the baseline state. Out of scope for this floor: a separate TrueNAS-first wizard, agent-required bootstrap, or disabled-by-default launch posture. 3. Infrastructure visibility: diff --git a/docs/release-control/v6/internal/subsystems/api-contracts.md b/docs/release-control/v6/internal/subsystems/api-contracts.md index 14bad75e8..d1087533b 100644 --- a/docs/release-control/v6/internal/subsystems/api-contracts.md +++ b/docs/release-control/v6/internal/subsystems/api-contracts.md @@ -5359,7 +5359,13 @@ workspace can render real API-backed status and handoff context without inventing a settings-local shadow fetch path. `frontend-modern/src/api/truenas.ts` owns the browser normalizer for that observed summary and must preserve those native inventory facets instead of collapsing them into a generic app or -storage count. Zero-value legacy +storage count. The same redacted list payload carries an optional secret-free +`transport` object with `mode`, `endpoint`, `tls`, `connected`, +`authMechanism`, `applianceVersion`, `legacyReason`, `reconnects`, +`lastError`, and `lastConnectedAt`. Settings must identify current +`jsonrpc-websocket`, version-gated `legacy-rest`, and not-yet-negotiated states +without exposing a credential or rebuilding transport state in the browser. +Zero-value legacy `pollIntervalSeconds` config must normalize back to the canonical 60-second default at this same boundary instead of leaking ambiguous `0` values to the frontend. diff --git a/docs/release-control/v6/internal/subsystems/monitoring.md b/docs/release-control/v6/internal/subsystems/monitoring.md index d3e1a7aff..dd0364ba3 100644 --- a/docs/release-control/v6/internal/subsystems/monitoring.md +++ b/docs/release-control/v6/internal/subsystems/monitoring.md @@ -222,6 +222,7 @@ node-local Agent evidence. 45. `pkg/agents/kubernetes/report.go` 46. `internal/monitoring/temperature.go` 47. `internal/truenas/client.go` +47a. `internal/truenas/transport.go` 48. `internal/truenas/disk_health.go` 49. `internal/truenas/provider.go` 50. `internal/models/ceph_cluster_identity.go` @@ -434,8 +435,9 @@ node-local Agent evidence. hard-coded cluster name. 16. Add or change TrueNAS supplemental inventory only through the native TrueNAS provider path and unified-resource projection. TrueNAS apps are - API-owned application records: `app.query` is the preferred live inventory - source, with legacy REST allowed only as a compatibility fallback. The + API-owned application records: `app.query` is the live inventory source on + the negotiated current transport, with legacy REST allowed only for a + connection proven to run a release that predates the versioned API. The provider may preserve Docker-compatible runtime metadata for shared container tooling, but it must also publish the native app identity, state, version, update availability, workload containers, ports, images, volumes, @@ -463,10 +465,31 @@ node-local Agent evidence. `internal/truenas/contract_test.go`. The canonical-ID migration semantics for rows minted under the retired hostname-keyed derivation live in the unified-resources contract (record-declared succession, item 27). - TrueNAS storage and alert inventory follow native query methods first: + Every live TrueNAS client owns one explicit, immutable transport decision. + SCALE 25.04 and later must use JSON-RPC 2.0 over a TLS WebSocket at + `/api/current`; authentication, authorization, TLS, protocol, or method + failures on that endpoint are authoritative and must never wake the + deprecated REST bridge. Only an unsupported-endpoint WebSocket handshake + may trigger a REST `/system/info` version probe, and REST may then be + selected only for recognized SCALE releases before 25.04 or TrueNAS + CORE/FreeNAS. Unknown or current versions fail closed. The decision and + persistent socket belong to one configured client, so reconnects or + legacy negotiation for one appliance cannot alter another appliance. + Current API-key authentication uses `auth.login_ex` with the key owner's + username and `API_KEY_PLAIN`; password authentication uses + `PASSWORD_PLAIN`. Username-less stored API keys may use the deprecated + login method only as an upgrade bridge, with explicit remediation when a + release removes that method. Read calls may reconnect with bounded backoff + and replay once after a transport failure. Mutating app calls must never + replay after dispatch because their outcome is ambiguous. Event reads must + retain the `core.subscribe` ID, call `core.unsubscribe` before reusing the + session, and discard a socket after a terminal stream read deadline. + Connection summaries expose only secret-free transport mode, endpoint, + TLS, authentication mechanism, appliance version, reconnect count, and + last-error timing diagnostics. + TrueNAS storage and alert inventory follow the negotiated transport: pools use `pool.query`, datasets use `pool.dataset.query`, disks use - `disk.query`, and alerts use `alert.list`, with legacy REST allowed only - as compatibility fallback. Inventory readers must only consume fields the + `disk.query`, and alerts use `alert.list`. Inventory readers must only consume fields the API actually serves on every supported TrueNAS line (CORE 13 REST-only included): `pool.dataset.query` carries no `mounted` field, so a listed dataset counts as mounted unless `locked` or an explicit value says @@ -474,8 +497,9 @@ node-local Agent evidence. `extra.pools` join cannot cross the REST bridge, so per-disk pool membership and ZFS member state derive from the vdev topology that `pool.query` attaches unconditionally; and `disk.temperatures` is a - parameterized method the REST bridge only serves as POST, tried after - native JSON-RPC reporting. Missing disk telemetry is reported as + parameterized method the legacy REST bridge only serves as POST, while + current releases use native JSON-RPC reporting without per-method REST + fallback. Missing disk telemetry is reported as unknown, never as a failure signal. Regression coverage: `internal/truenas/client_api_shapes_test.go`. Unhealthy pool state from `pool.query` must emit a provider-native `zfs_pool_state` incident on @@ -501,9 +525,11 @@ node-local Agent evidence. `sharing.nfs.query` publishes native `TrueNASData.Share` on canonical `network-share` resources parented to the owning dataset or pool when the API/path supplies that evidence. TrueNAS protection inventory follows the - same native-query rule: ZFS snapshots prefer `zfs.resource.snapshot.query` - with older `pool.snapshot.query`/REST compatibility fallback, and - replication tasks prefer `replication.query`. + same native-query rule: current connections prefer + `zfs.resource.snapshot.query`, with `pool.snapshot.query` allowed only as a + same-transport method-name compatibility path; version-gated legacy + connections use REST. Replication tasks use `replication.query` on current + connections. TrueNAS system services are also native appliance inventory: `service.query` is the preferred source for service name, boot enablement, runtime state, and process IDs. Pulse must publish that data through `TrueNASData.Services` @@ -1682,9 +1708,9 @@ machine ID, capacity, and poll-health path instead of failing connection tests or background refreshes during JSON decoding. That same monitoring boundary now also owns live TrueNAS disk temperatures. `internal/truenas/client.go` and `internal/truenas/provider.go` must ingest -`disk.temperatures` from the TrueNAS API, fall back to modern -`reporting.get_data` `disktemp` when the dedicated endpoint is unavailable, and -project those readings into the canonical physical-disk model and risk path +legacy `disk.temperatures` from the REST API or `reporting.get_data` `disktemp` +from the current JSON-RPC transport, and project those readings into the +canonical physical-disk model and risk path instead of leaving temperature telemetry agent-only or adding a TrueNAS-local presentation shim. That same monitoring boundary also owns SMART-backed TrueNAS disk risk @@ -1701,10 +1727,11 @@ null, empty, missing, unknown, or unavailable SMART telemetry to canonical native failure states such as `FAULTED`, `FAILED`, `OFFLINE`, `REMOVED`, and `UNAVAIL` must continue to produce canonical disk-health risk. That same boundary owns boot-pool and replication-target storage posture. -`internal/truenas/client.go` must collect `boot.get_state` with the legacy REST -bridge as compatibility fallback, merge it only within the current configured -connection, and use its vdev leaves to enrich boot-disk pool membership and -native ZFS state. `internal/monitoring/truenas_poller.go` must correlate +`internal/truenas/client.go` must collect `boot.get_state` through the +connection's negotiated transport, use REST only on a version-gated legacy +connection, merge it only within the current configured connection, and use +its vdev leaves to enrich boot-disk pool membership and native ZFS state. +`internal/monitoring/truenas_poller.go` must correlate `replication.query` intent across providers within the same organization using local/PULL ownership or a unique configured/observed target-host match for remote PUSH tasks. The resulting `SET`/`REQUIRE` receive-side read-only posture diff --git a/docs/release-control/v6/internal/subsystems/registry.json b/docs/release-control/v6/internal/subsystems/registry.json index c6c2a7d1c..590c0cd3a 100644 --- a/docs/release-control/v6/internal/subsystems/registry.json +++ b/docs/release-control/v6/internal/subsystems/registry.json @@ -5255,6 +5255,7 @@ "internal/truenas/disk_health.go", "internal/truenas/fixtures.go", "internal/truenas/provider.go", + "internal/truenas/transport.go", "internal/truenas/types.go" ], "allow_same_subsystem_tests": false, @@ -5264,7 +5265,8 @@ "internal/monitoring/monitor_polling_test.go", "internal/truenas/client_test.go", "internal/truenas/contract_test.go", - "internal/truenas/provider_test.go" + "internal/truenas/provider_test.go", + "internal/truenas/transport_test.go" ] }, { diff --git a/frontend-modern/src/api/__tests__/truenas.test.ts b/frontend-modern/src/api/__tests__/truenas.test.ts index 217194ff4..0e1ee4d1d 100644 --- a/frontend-modern/src/api/__tests__/truenas.test.ts +++ b/frontend-modern/src/api/__tests__/truenas.test.ts @@ -19,6 +19,7 @@ describe('TrueNASAPI', () => { host: ' truenas.local ', port: 443, apiKey: ' ******** ', + username: ' pulse-readonly ', useHttps: true, insecureSkipVerify: false, enabled: true, @@ -40,6 +41,15 @@ describe('TrueNASAPI', () => { disks: 8, recoveryArtifacts: 18, }, + transport: { + mode: 'jsonrpc-websocket', + endpoint: ' wss://truenas.local/api/current ', + tls: true, + connected: true, + authMechanism: ' api-key-plain ', + applianceVersion: ' TrueNAS-SCALE-25.10.4 ', + reconnects: 1, + }, }, ]); @@ -50,7 +60,7 @@ describe('TrueNASAPI', () => { host: 'truenas.local', port: 443, apiKey: '********', - username: undefined, + username: 'pulse-readonly', password: undefined, useHttps: true, insecureSkipVerify: false, @@ -77,6 +87,18 @@ describe('TrueNASAPI', () => { disks: 8, recoveryArtifacts: 18, }, + transport: { + mode: 'jsonrpc-websocket', + endpoint: 'wss://truenas.local/api/current', + tls: true, + connected: true, + authMechanism: 'api-key-plain', + applianceVersion: 'TrueNAS-SCALE-25.10.4', + legacyReason: undefined, + reconnects: 1, + lastError: undefined, + lastConnectedAt: undefined, + }, monitorDatasets: false, monitorPools: false, monitorReplication: false, diff --git a/frontend-modern/src/api/truenas.ts b/frontend-modern/src/api/truenas.ts index dbfe8e172..d08e07978 100644 --- a/frontend-modern/src/api/truenas.ts +++ b/frontend-modern/src/api/truenas.ts @@ -21,6 +21,7 @@ type RawTrueNASConnection = Partial; type RawTrueNASConnectionPollError = Partial; type RawTrueNASConnectionPoll = Partial; type RawTrueNASConnectionObservedSummary = Partial; +type RawTrueNASConnectionTransportStatus = Partial; export interface TrueNASConnectionPollError { at?: string; @@ -50,6 +51,19 @@ export interface TrueNASConnectionObservedSummary { recoveryArtifacts: number; } +export interface TrueNASConnectionTransportStatus { + mode: 'negotiating' | 'jsonrpc-websocket' | 'legacy-rest'; + endpoint?: string; + tls: boolean; + connected: boolean; + authMechanism?: string; + applianceVersion?: string; + legacyReason?: string; + reconnects?: number; + lastError?: string; + lastConnectedAt?: string; +} + export interface TrueNASConnection { id: string; name: string; @@ -68,6 +82,7 @@ export interface TrueNASConnection { monitorReplication: boolean; poll?: TrueNASConnectionPollStatus; observed?: TrueNASConnectionObservedSummary; + transport?: TrueNASConnectionTransportStatus; } export interface TrueNASConnectionInput { @@ -134,6 +149,27 @@ const normalizeTrueNASConnectionObservedSummary = ( }; }; +const normalizeTrueNASConnectionTransportStatus = ( + transport: RawTrueNASConnectionTransportStatus | undefined, +): TrueNASConnectionTransportStatus | undefined => { + if (!transport || typeof transport !== 'object') return undefined; + const rawMode = optionalTrimmedString(transport.mode); + const mode = + rawMode === 'jsonrpc-websocket' || rawMode === 'legacy-rest' ? rawMode : 'negotiating'; + return { + mode, + endpoint: optionalTrimmedString(transport.endpoint), + tls: strictBoolean(transport.tls), + connected: strictBoolean(transport.connected), + authMechanism: optionalTrimmedString(transport.authMechanism), + applianceVersion: optionalTrimmedString(transport.applianceVersion), + legacyReason: optionalTrimmedString(transport.legacyReason), + reconnects: finiteNumberOrUndefined(transport.reconnects), + lastError: optionalTrimmedString(transport.lastError), + lastConnectedAt: optionalTrimmedString(transport.lastConnectedAt), + }; +}; + const normalizeTrueNASConnection = (connection: RawTrueNASConnection): TrueNASConnection => ({ id: trimmedString(connection.id), name: optionalTrimmedString(connection.name) ?? '', @@ -152,6 +188,7 @@ const normalizeTrueNASConnection = (connection: RawTrueNASConnection): TrueNASCo monitorReplication: strictBoolean(connection.monitorReplication), poll: normalizeTrueNASConnectionPoll(connection.poll), observed: normalizeTrueNASConnectionObservedSummary(connection.observed), + transport: normalizeTrueNASConnectionTransportStatus(connection.transport), }); const serializeTrueNASConnectionInput = (input: TrueNASConnectionInput) => ({ diff --git a/frontend-modern/src/components/Settings/ConnectionEditor/CredentialSlots/TrueNASCredentialSlot.tsx b/frontend-modern/src/components/Settings/ConnectionEditor/CredentialSlots/TrueNASCredentialSlot.tsx index 74f5e279b..d09fd61c0 100644 --- a/frontend-modern/src/components/Settings/ConnectionEditor/CredentialSlots/TrueNASCredentialSlot.tsx +++ b/frontend-modern/src/components/Settings/ConnectionEditor/CredentialSlots/TrueNASCredentialSlot.tsx @@ -28,6 +28,22 @@ export interface TrueNASCredentialSlotProps { deleteError?: string | null; } +const describeTransport = (transport: NonNullable): string => { + let description: string; + if (transport.mode === 'jsonrpc-websocket') { + description = `Transport: JSON-RPC WebSocket${transport.tls ? ' over TLS' : ''}; authentication: ${transport.authMechanism || 'session'}.`; + } else if (transport.mode === 'legacy-rest') { + description = + `Transport: legacy REST (${transport.applianceVersion || 'legacy TrueNAS'}). ${transport.legacyReason || ''}`.trim(); + } else { + description = 'Transport negotiation has not completed yet.'; + } + if (transport.lastError) { + description += ` Last error: ${transport.lastError}`; + } + return description; +}; + export const TrueNASCredentialSlot: Component = (props) => { let primed = false; @@ -132,21 +148,36 @@ export const TrueNASCredentialSlot: Component = (pro - +
+ + +
@@ -236,6 +267,17 @@ export const TrueNASCredentialSlot: Component = (pro + + {(transport) => ( + + )} + +
Collection scope
diff --git a/frontend-modern/src/components/Settings/__tests__/useTrueNASSettingsPanelState.test.tsx b/frontend-modern/src/components/Settings/__tests__/useTrueNASSettingsPanelState.test.tsx index 54a3a2260..ea676f5f0 100644 --- a/frontend-modern/src/components/Settings/__tests__/useTrueNASSettingsPanelState.test.tsx +++ b/frontend-modern/src/components/Settings/__tests__/useTrueNASSettingsPanelState.test.tsx @@ -80,6 +80,7 @@ describe('useTrueNASSettingsPanelState', () => { name: 'tower', host: 'truenas.local', apiKey: '********', + username: 'pulse-readonly', pollIntervalSeconds: 90, useHttps: true, insecureSkipVerify: false, @@ -91,6 +92,7 @@ describe('useTrueNASSettingsPanelState', () => { name: 'tower', host: 'truenas.local', apiKey: '********', + username: 'pulse-readonly', pollIntervalSeconds: 90, useHttps: true, insecureSkipVerify: false, @@ -105,6 +107,7 @@ describe('useTrueNASSettingsPanelState', () => { name: 'tower', host: 'truenas.local', apiKey: '********', + username: 'pulse-readonly', pollIntervalSeconds: 90, useHttps: true, insecureSkipVerify: false, @@ -128,6 +131,7 @@ describe('useTrueNASSettingsPanelState', () => { host: 'truenas.local', apiKey: '********', pollIntervalSeconds: 90, + username: 'pulse-readonly', }), ); expect(TrueNASAPI.updateConnection).toHaveBeenCalledWith( @@ -136,7 +140,7 @@ describe('useTrueNASSettingsPanelState', () => { host: 'truenas.local', apiKey: '********', pollIntervalSeconds: 90, - username: '', + username: 'pulse-readonly', password: '', }), ); diff --git a/frontend-modern/src/components/Settings/useTrueNASSettingsPanelState.ts b/frontend-modern/src/components/Settings/useTrueNASSettingsPanelState.ts index 2d15e7659..76337788d 100644 --- a/frontend-modern/src/components/Settings/useTrueNASSettingsPanelState.ts +++ b/frontend-modern/src/components/Settings/useTrueNASSettingsPanelState.ts @@ -73,7 +73,7 @@ const buildFormStateFromConnection = ( pollIntervalSeconds: String(connection.pollIntervalSeconds ?? 60), authMode, apiKey: '', - username: authMode === 'userpass' ? connection.username || '' : '', + username: connection.username || '', password: '', useHttps: connection.useHttps, insecureSkipVerify: connection.insecureSkipVerify, @@ -163,7 +163,7 @@ const buildConnectionInput = (form: TrueNASConnectionFormState): TrueNASConnecti if (form.authMode === 'apiKey') { input.apiKey = form.apiKey.trim() || (form.hasStoredApiKey ? REDACTED_SECRET : ''); - input.username = ''; + input.username = form.username.trim(); input.password = ''; } else { input.apiKey = ''; diff --git a/internal/api/platform_mock_connections.go b/internal/api/platform_mock_connections.go index abeeb38e5..57ce512f2 100644 --- a/internal/api/platform_mock_connections.go +++ b/internal/api/platform_mock_connections.go @@ -6,6 +6,7 @@ import ( "github.com/rcourtman/pulse-go-rewrite/internal/config" "github.com/rcourtman/pulse-go-rewrite/internal/mock" "github.com/rcourtman/pulse-go-rewrite/internal/monitoring" + "github.com/rcourtman/pulse-go-rewrite/internal/truenas" ) func mockTrueNASConnectionResponses() []trueNASConnectionResponse { @@ -43,6 +44,15 @@ func mockTrueNASConnectionResponses() []trueNASConnectionResponse { LastSuccessAt: mockPlatformTimePointer(fixture.CollectedAt), }, Observed: observed, + Transport: &truenas.TransportStatus{ + Mode: truenas.TransportJSONRPC, + Endpoint: "wss://truenas-main/api/current", + TLS: true, + Connected: true, + AuthMechanism: "api-key-plain", + ApplianceVersion: "TrueNAS-SCALE-25.10.4", + LastConnectedAt: mockPlatformTimePointer(fixture.CollectedAt), + }, }} } @@ -105,8 +115,9 @@ func mockTrueNASLedgerInputs() ([]config.TrueNASInstance, map[string]monitoring. for _, response := range responses { instances = append(instances, response.TrueNASInstance) summaries[response.TrueNASInstance.ID] = monitoring.TrueNASConnectionSummary{ - Poll: response.Poll, - Observed: response.Observed, + Poll: response.Poll, + Observed: response.Observed, + Transport: response.Transport, } } return instances, summaries diff --git a/internal/api/truenas_handlers.go b/internal/api/truenas_handlers.go index a812014f0..62ad0dc00 100644 --- a/internal/api/truenas_handlers.go +++ b/internal/api/truenas_handlers.go @@ -28,8 +28,9 @@ type TrueNASHandlers struct { type trueNASConnectionResponse struct { config.TrueNASInstance - Poll *monitoring.TrueNASConnectionPollStatus `json:"poll,omitempty"` - Observed *monitoring.TrueNASConnectionObservedSummary `json:"observed,omitempty"` + Poll *monitoring.TrueNASConnectionPollStatus `json:"poll,omitempty"` + Observed *monitoring.TrueNASConnectionObservedSummary `json:"observed,omitempty"` + Transport *truenas.TransportStatus `json:"transport,omitempty"` } type trueNASClient interface { @@ -126,6 +127,7 @@ func (h *TrueNASHandlers) HandleList(w http.ResponseWriter, r *http.Request) { if summary, ok := summaries[strings.TrimSpace(item.ID)]; ok { response.Poll = summary.Poll response.Observed = summary.Observed + response.Transport = summary.Transport } redacted = append(redacted, response) } diff --git a/internal/api/truenas_handlers_test.go b/internal/api/truenas_handlers_test.go index 2ed06332e..d94af1cc7 100644 --- a/internal/api/truenas_handlers_test.go +++ b/internal/api/truenas_handlers_test.go @@ -439,6 +439,12 @@ func TestTrueNASHandlers_HandleList_ReturnsMockConnectionsInMockMode(t *testing. if listed[0].Observed == nil || listed[0].Observed.Systems != 1 || listed[0].Observed.StoragePools == 0 { t.Fatalf("expected populated mock observed summary, got %+v", listed[0].Observed) } + if listed[0].Transport == nil || + listed[0].Transport.Mode != truenas.TransportJSONRPC || + !listed[0].Transport.Connected || + !listed[0].Transport.TLS { + t.Fatalf("expected current mock transport diagnostics, got %+v", listed[0].Transport) + } } func TestTrueNASHandlers_HandleList_IncludesPollAndObservedSummary(t *testing.T) { @@ -520,6 +526,18 @@ func TestTrueNASHandlers_HandleList_IncludesPollAndObservedSummary(t *testing.T) if listed[0].Observed.StoragePools != 1 || listed[0].Observed.Datasets != 1 || listed[0].Observed.Disks != 1 { t.Fatalf("unexpected observed counts: %+v", listed[0].Observed) } + if listed[0].Transport == nil || + listed[0].Transport.Mode != truenas.TransportLegacyREST || + listed[0].Transport.ApplianceVersion != "TrueNAS-SCALE-24.10.2" { + t.Fatalf("unexpected transport diagnostics: %+v", listed[0].Transport) + } + encodedTransport, err := json.Marshal(listed[0].Transport) + if err != nil { + t.Fatalf("encode transport diagnostics: %v", err) + } + if strings.Contains(string(encodedTransport), connection.APIKey) { + t.Fatalf("transport diagnostics exposed API key: %s", encodedTransport) + } } func TestTrueNASHandlers_HandleDelete_RemovesAndHandlesUnknownID(t *testing.T) { diff --git a/internal/monitoring/truenas_poller.go b/internal/monitoring/truenas_poller.go index 14e95abf9..80b59b2c2 100644 --- a/internal/monitoring/truenas_poller.go +++ b/internal/monitoring/truenas_poller.go @@ -62,8 +62,9 @@ type TrueNASConnectionObservedSummary struct { // TrueNASConnectionSummary merges poll health with the most recent discovered // platform contribution for one configured TrueNAS connection. type TrueNASConnectionSummary struct { - Poll *TrueNASConnectionPollStatus `json:"poll,omitempty"` - Observed *TrueNASConnectionObservedSummary `json:"observed,omitempty"` + Poll *TrueNASConnectionPollStatus `json:"poll,omitempty"` + Observed *TrueNASConnectionObservedSummary `json:"observed,omitempty"` + Transport *truenas.TransportStatus `json:"transport,omitempty"` } type trueNASConnectionRuntimeStatus struct { @@ -390,6 +391,10 @@ func (p *TrueNASPoller) ConnectionSummaries(orgID string, instances []config.Tru if p != nil { p.mu.Lock() status := p.cloneRuntimeStatusLocked(orgID, connID) + var provider *truenas.Provider + if byConnection := p.providersByOrg[orgID]; byConnection != nil { + provider = byConnection[connID] + } p.mu.Unlock() if status != nil { summary.Poll.LastAttemptAt = cloneTimePointer(status.LastAttemptAt) @@ -398,6 +403,10 @@ func (p *TrueNASPoller) ConnectionSummaries(orgID string, instances []config.Tru summary.Poll.LastError = cloneTrueNASConnectionPollError(status.LastError) summary.Observed = cloneTrueNASObservedSummary(status.Observed) } + if provider != nil { + transport := provider.TransportStatus() + summary.Transport = &transport + } } summaries[connID] = summary @@ -1484,6 +1493,29 @@ func classifyTrueNASError(err error, connectionID string) *internalerrors.Monito errType = internalerrors.ErrorTypeAPI } } else { + var authErr *truenas.RPCAuthError + if errors.As(err, &authErr) { + errType = internalerrors.ErrorTypeAuth + retryable = false + } + var handshakeErr *truenas.RPCHandshakeError + if errors.As(err, &handshakeErr) && + (handshakeErr.StatusCode == http.StatusUnauthorized || handshakeErr.StatusCode == http.StatusForbidden) { + errType = internalerrors.ErrorTypeAuth + retryable = false + } + var rpcErr *truenas.RPCError + if errors.As(err, &rpcErr) { + message := strings.ToLower(rpcErr.Message + " " + rpcErr.Reason) + if strings.HasPrefix(rpcErr.Method, "auth.") || + strings.EqualFold(rpcErr.Errname, "EACCES") || + strings.EqualFold(rpcErr.Errname, "EPERM") || + strings.Contains(message, "not authorized") || + strings.Contains(message, "permission") { + errType = internalerrors.ErrorTypeAuth + retryable = false + } + } // Transport-level errors: timeout takes precedence over generic connection failures. var urlErr *url.Error if (errors.As(err, &urlErr) && urlErr.Timeout()) || errors.Is(err, context.DeadlineExceeded) { diff --git a/internal/monitoring/truenas_poller_test.go b/internal/monitoring/truenas_poller_test.go index 76df5eac7..c5b254b5a 100644 --- a/internal/monitoring/truenas_poller_test.go +++ b/internal/monitoring/truenas_poller_test.go @@ -310,6 +310,11 @@ func TestTrueNASPollerConnectionSummariesExposeObservedCounts(t *testing.T) { if summary.Observed.Systems != 1 || summary.Observed.StoragePools != 1 || summary.Observed.Datasets != 1 || summary.Observed.Disks != 1 { t.Fatalf("unexpected observed counts: %+v", summary.Observed) } + if summary.Transport == nil || + summary.Transport.Mode != truenas.TransportLegacyREST || + !summary.Transport.Connected { + t.Fatalf("unexpected connection-local transport summary: %+v", summary.Transport) + } } func TestTrueNASObservedSummaryIncludesNativeRuntimeAndSharingFacets(t *testing.T) { @@ -1678,6 +1683,33 @@ func TestClassifyTrueNASError(t *testing.T) { expectedType: "auth", expectedRetry: false, }, + { + name: "websocket handshake 403 classifies as auth", + err: &truenas.RPCHandshakeError{StatusCode: 403, Err: fmt.Errorf("forbidden")}, + expectedType: "auth", + expectedRetry: false, + }, + { + name: "JSON-RPC login response classifies as auth", + err: &truenas.RPCAuthError{ + Mechanism: "api-key-plain", + ResponseType: "AUTH_ERR", + }, + expectedType: "auth", + expectedRetry: false, + }, + { + name: "JSON-RPC permission error classifies as auth", + err: &truenas.RPCError{ + Code: -32001, + Method: "pool.query", + Message: "Method call error", + Reason: "Not authorized", + Errname: "EACCES", + }, + expectedType: "auth", + expectedRetry: false, + }, { name: "context.DeadlineExceeded classifies as timeout", err: context.DeadlineExceeded, diff --git a/internal/truenas/client.go b/internal/truenas/client.go index 9a868b21b..8fca4913c 100644 --- a/internal/truenas/client.go +++ b/internal/truenas/client.go @@ -16,6 +16,7 @@ import ( "net/url" "strconv" "strings" + "sync" "time" "github.com/gorilla/websocket" @@ -37,7 +38,7 @@ const defaultAppLogIdleWait = 250 * time.Millisecond const maxAppLogTailLines = 500 -// ClientConfig configures the TrueNAS REST API client. +// ClientConfig configures the TrueNAS API client. type ClientConfig struct { Host string Port int @@ -50,15 +51,30 @@ type ClientConfig struct { Timeout time.Duration } -// Client is a thin HTTP wrapper around the TrueNAS REST API v2.0. +// Client owns one connection-local TrueNAS transport decision. Supported +// current SCALE releases use JSON-RPC 2.0 over WebSocket; REST is retained +// only as a version-gated boundary for legacy SCALE and CORE appliances. type Client struct { config ClientConfig httpClient *http.Client baseURL string rpcURL string + + rpcMu sync.Mutex + rpc *trueNASRPCClient + mode TransportMode + closed bool + reconnect int + + statusMu sync.RWMutex + status TransportStatus + + // Tests exercise protocol behavior with httptest's plaintext websocket. + // Production clients never set this escape hatch. + allowInsecureRPC bool } -// APIError represents an HTTP-level error from the TrueNAS REST API. +// APIError represents an HTTP-level error from the legacy TrueNAS REST API. type APIError struct { StatusCode int Method string @@ -70,7 +86,7 @@ func (e *APIError) Error() string { return fmt.Sprintf("truenas request %s %s failed: status=%d body=%q", e.Method, e.Path, e.StatusCode, e.Body) } -// NewClient creates a new TrueNAS REST API client. +// NewClient creates a new TrueNAS API client. func NewClient(config ClientConfig) (*Client, error) { host := strings.TrimSpace(config.Host) if host == "" { @@ -117,7 +133,7 @@ func NewClient(config ClientConfig) (*Client, error) { wsScheme = "wss" } - return &Client{ + client := &Client{ config: config, httpClient: &http.Client{ Timeout: timeout, @@ -125,7 +141,14 @@ func NewClient(config ClientConfig) (*Client, error) { }, baseURL: fmt.Sprintf("%s://%s/api/v2.0", scheme, hostPort), rpcURL: fmt.Sprintf("%s://%s/api/current", wsScheme, hostPort), - }, nil + mode: TransportUnknown, + } + client.status = TransportStatus{ + Mode: TransportUnknown, + Endpoint: client.rpcURL, + TLS: useHTTPS, + } + return client, nil } // TestConnection validates that the endpoint is reachable and authenticated. @@ -136,23 +159,48 @@ func (c *Client) TestConnection(ctx context.Context) error { return nil } -// Close releases idle HTTP transport connections held by the client. +// Close releases the persistent websocket session and idle HTTP transport +// connections held by the client. func (c *Client) Close() { - if c == nil || c.httpClient == nil || c.httpClient.Transport == nil { + if c == nil { return } - if transport, ok := c.httpClient.Transport.(interface{ CloseIdleConnections() }); ok { - transport.CloseIdleConnections() + c.rpcMu.Lock() + c.closed = true + c.closeRPCLocked() + c.rpcMu.Unlock() + c.updateTransportStatus(func(status *TransportStatus) { + status.Connected = false + }) + if c.httpClient != nil && c.httpClient.Transport != nil { + if transport, ok := c.httpClient.Transport.(interface{ CloseIdleConnections() }); ok { + transport.CloseIdleConnections() + } } } // GetSystemInfo returns high-level system metadata. func (c *Client) GetSystemInfo(ctx context.Context) (*SystemInfo, error) { var response systemInfoResponse - if err := c.getJSON(ctx, http.MethodGet, "/system/info", &response); err != nil { + mode, err := c.ensureTransport(ctx) + if err != nil { return nil, err } + if mode == TransportLegacyREST { + err = c.getJSON(ctx, http.MethodGet, "/system/info", &response) + } else { + err = c.callRPC(ctx, "system.info", []any{}, &response) + } + if err != nil { + return nil, err + } + c.updateTransportStatus(func(status *TransportStatus) { + status.ApplianceVersion = strings.TrimSpace(response.Version) + }) + return systemInfoFromResponse(response), nil +} +func systemInfoFromResponse(response systemInfoResponse) *SystemInfo { // MachineID is the raw DMI serial only. It must never fall back to the // reported hostname: hostname is not a machine identity, and the old // fallback gave two serial-less systems that report the same hostname @@ -178,7 +226,7 @@ func (c *Client) GetSystemInfo(ctx context.Context) (*SystemInfo, error) { MachineID: machineID, CPUCount: cpuCount, MemoryTotalBytes: response.Physmem, - }, nil + } } // GetSystemTelemetry retrieves live system telemetry from the modern TrueNAS @@ -186,31 +234,21 @@ func (c *Client) GetSystemInfo(ctx context.Context) (*SystemInfo, error) { // transport or endpoint failure as "telemetry unavailable" rather than a // system identity failure. func (c *Client) GetSystemTelemetry(ctx context.Context) (*SystemInfo, error) { - conn, err := c.dialRPC(ctx) - if err != nil { - return nil, err - } - defer func() { _ = conn.Close() }() - - rpc := trueNASRPCClient{ - conn: conn, - nextID: 1, - } - if err := rpc.authenticate(ctx, c.config); err != nil { - return nil, err - } - - temperatures, err := rpc.getSystemTemperatures(ctx) - if err != nil { - temperatures = nil - } - - subscriptionName := fmt.Sprintf("reporting.realtime:{\"interval\":%d}", defaultRealtimeIntervalSeconds) - if err := rpc.call(ctx, "core.subscribe", []any{subscriptionName}, nil); err != nil { - return nil, err - } - - telemetry, err := rpc.readSystemTelemetryEvent(ctx, defaultRealtimeIntervalSeconds) + var telemetry *SystemInfo + var temperatures map[string]float64 + err := c.withRPC(ctx, func(rpc *trueNASRPCClient) error { + temperatures, _ = rpc.getSystemTemperatures(ctx) + subscriptionName := fmt.Sprintf("reporting.realtime:{\"interval\":%d}", defaultRealtimeIntervalSeconds) + subscriptionID, err := rpc.subscribe(ctx, subscriptionName) + if err != nil { + return err + } + telemetry, err = rpc.readSystemTelemetryEvent(ctx, defaultRealtimeIntervalSeconds) + if err != nil { + return discardRPCSessionForStreamError(err) + } + return rpc.unsubscribe(ctx, subscriptionID) + }) if err != nil { return nil, err } @@ -223,34 +261,25 @@ func (c *Client) GetSystemTelemetry(ctx context.Context) (*SystemInfo, error) { // GetSystemMetricHistory retrieves historical system metrics from the native // TrueNAS reporting API for the canonical host-chart fallback path. func (c *Client) GetSystemMetricHistory(ctx context.Context, duration time.Duration) (*SystemMetricHistory, error) { - conn, err := c.dialRPC(ctx) - if err != nil { - return nil, err - } - defer func() { _ = conn.Close() }() - - rpc := trueNASRPCClient{ - conn: conn, - nextID: 1, - } - if err := rpc.authenticate(ctx, c.config); err != nil { - return nil, err - } - - return rpc.getSystemMetricHistory(ctx, duration) + var history *SystemMetricHistory + err := c.withRPC(ctx, func(rpc *trueNASRPCClient) error { + var err error + history, err = rpc.getSystemMetricHistory(ctx, duration) + return err + }) + return history, err } // GetPools returns storage pools. func (c *Client) GetPools(ctx context.Context) ([]Pool, error) { - pools, err := c.getPoolsRPC(ctx) - if err == nil { - return pools, nil + legacy, err := c.useLegacyREST(ctx) + if err != nil { + return nil, err } - restPools, restErr := c.getPoolsREST(ctx) - if restErr != nil { - return nil, fmt.Errorf("fetch truenas pools via rpc and rest: rpc=%w rest=%v", err, restErr) + if legacy { + return c.getPoolsREST(ctx) } - return restPools, nil + return c.getPoolsRPC(ctx) } func (c *Client) getPoolsRPC(ctx context.Context) ([]Pool, error) { @@ -283,17 +312,23 @@ func (c *Client) getPoolsRPC(ctx context.Context) ([]Pool, error) { // separate from pool.query on supported CORE and SCALE releases. func (c *Client) GetBootPool(ctx context.Context) (*Pool, error) { var response map[string]any - rpcErr := c.callRPC(ctx, "boot.get_state", []any{}, &response) - if rpcErr == nil { + legacy, err := c.useLegacyREST(ctx) + if err != nil { + return nil, err + } + if !legacy { + if err := c.callRPC(ctx, "boot.get_state", []any{}, &response); err != nil { + return nil, err + } if pool, ok := parseBootPoolState(response); ok { return &pool, nil } - rpcErr = fmt.Errorf("boot.get_state returned no boot pool identity") + return nil, fmt.Errorf("boot.get_state returned no boot pool identity") } response = nil - if restErr := c.getJSON(ctx, http.MethodGet, "/boot/get_state", &response); restErr != nil { - return nil, fmt.Errorf("fetch truenas boot pool via rpc and rest: rpc=%w rest=%v", rpcErr, restErr) + if err := c.getJSON(ctx, http.MethodGet, "/boot/get_state", &response); err != nil { + return nil, err } if pool, ok := parseBootPoolState(response); ok { return &pool, nil @@ -520,15 +555,14 @@ func (c *Client) getPoolsREST(ctx context.Context) ([]Pool, error) { // GetDatasets returns datasets and normalized capacity/read-only fields. func (c *Client) GetDatasets(ctx context.Context) ([]Dataset, error) { - datasets, err := c.getDatasetsRPC(ctx) - if err == nil { - return datasets, nil + legacy, err := c.useLegacyREST(ctx) + if err != nil { + return nil, err } - restDatasets, restErr := c.getDatasetsREST(ctx) - if restErr != nil { - return nil, fmt.Errorf("fetch truenas datasets via rpc and rest: rpc=%w rest=%v", err, restErr) + if legacy { + return c.getDatasetsREST(ctx) } - return restDatasets, nil + return c.getDatasetsRPC(ctx) } func (c *Client) getDatasetsRPC(ctx context.Context) ([]Dataset, error) { @@ -630,15 +664,14 @@ func (c *Client) getDatasetsREST(ctx context.Context) ([]Dataset, error) { // GetDisks returns the system disk inventory. func (c *Client) GetDisks(ctx context.Context) ([]Disk, error) { - disks, err := c.getDisksRPC(ctx) - if err == nil { - return disks, nil + legacy, err := c.useLegacyREST(ctx) + if err != nil { + return nil, err } - restDisks, restErr := c.getDisksREST(ctx) - if restErr != nil { - return nil, fmt.Errorf("fetch truenas disks via rpc and rest: rpc=%w rest=%v", err, restErr) + if legacy { + return c.getDisksREST(ctx) } - return restDisks, nil + return c.getDisksRPC(ctx) } func (c *Client) getDisksRPC(ctx context.Context) ([]Disk, error) { @@ -656,6 +689,9 @@ func (c *Client) getDisksREST(ctx context.Context) ([]Disk, error) { if err := c.getJSON(ctx, http.MethodGet, "/disk", &response); err != nil { return nil, err } + if len(response) == 0 { + return nil, nil + } identifiers := diskReportingIdentifiers(response) temperatures, err := c.getDiskTemperaturesWithFallback(ctx, identifiers) if err != nil { @@ -706,6 +742,9 @@ func (c *Client) getDisksREST(ctx context.Context) ([]Disk, error) { } func (c *Client) disksFromMaps(ctx context.Context, response []map[string]any) ([]Disk, error) { + if len(response) == 0 { + return nil, nil + } identifiers := diskReportingIdentifiersFromMaps(response) temperatures, err := c.getDiskTemperaturesWithFallback(ctx, identifiers) if err != nil { @@ -819,20 +858,20 @@ func (c *Client) GetDiskTemperatureHistory(ctx context.Context, identifiers []st return nil, fmt.Errorf("truenas disk temperature history requires disk identifiers") } - conn, err := c.dialRPC(ctx) - if err != nil { - return nil, err - } - defer func() { _ = conn.Close() }() - - rpc := &trueNASRPCClient{conn: conn, nextID: 1} - if err := rpc.authenticate(ctx, c.config); err != nil { - return nil, err - } - return rpc.getDiskTemperatureHistory(ctx, identifiers, duration) + var history map[string][]TimeSeriesPoint + err := c.withRPC(ctx, func(rpc *trueNASRPCClient) error { + var err error + history, err = rpc.getDiskTemperatureHistory(ctx, identifiers, duration) + return err + }) + return history, err } func (c *Client) getDiskTemperaturesWithFallback(ctx context.Context, identifiers []string) (map[string]int, error) { + legacy, err := c.useLegacyREST(ctx) + if err != nil { + return nil, err + } if len(identifiers) == 0 { reportingIdentifiers, err := c.listDiskReportingIdentifiers(ctx) if err == nil { @@ -840,11 +879,11 @@ func (c *Client) getDiskTemperaturesWithFallback(ctx context.Context, identifier } } - // Native JSON-RPC reporting first: it is the supported surface on - // SCALE 25.04+ and avoids waking the deprecated REST bridge (#1550). - reportingTemperatures, reportingErr := c.getDiskTemperaturesFromReporting(ctx, identifiers) - if reportingErr == nil && len(reportingTemperatures) > 0 { - return reportingTemperatures, nil + if !legacy { + // Native JSON-RPC reporting is the only supported path on SCALE + // 25.04+. An unavailable reporting method is best-effort telemetry; + // it must never wake the deprecated REST bridge. + return c.getDiskTemperaturesFromReporting(ctx, identifiers) } // disk.temperatures takes parameters, so REST v2.0 serves it as POST @@ -862,9 +901,6 @@ func (c *Client) getDiskTemperaturesWithFallback(ctx context.Context, identifier } if restErr != nil { - if reportingErr != nil { - return nil, fmt.Errorf("fetch truenas disk temperatures via reporting and rest: reporting=%w rest=%v", reportingErr, restErr) - } return nil, restErr } @@ -872,6 +908,17 @@ func (c *Client) getDiskTemperaturesWithFallback(ctx context.Context, identifier } func (c *Client) listDiskReportingIdentifiers(ctx context.Context) ([]string, error) { + legacy, err := c.useLegacyREST(ctx) + if err != nil { + return nil, err + } + if !legacy { + var response []map[string]any + if err := c.callRPC(ctx, "disk.query", []any{[]any{}, map[string]any{}}, &response); err != nil { + return nil, err + } + return diskReportingIdentifiersFromMaps(response), nil + } var response []diskResponse if err := c.getJSON(ctx, http.MethodGet, "/disk", &response); err != nil { return nil, err @@ -913,17 +960,13 @@ func (c *Client) getDiskTemperaturesFromReporting(ctx context.Context, identifie return nil, fmt.Errorf("truenas reporting disk temperature fallback requires disk identifiers") } - conn, err := c.dialRPC(ctx) - if err != nil { - return nil, err - } - defer func() { _ = conn.Close() }() - - rpc := &trueNASRPCClient{conn: conn, nextID: 1} - if err := rpc.authenticate(ctx, c.config); err != nil { - return nil, err - } - return rpc.getDiskTemperatures(ctx, identifiers) + var temperatures map[string]int + err := c.withRPC(ctx, func(rpc *trueNASRPCClient) error { + var err error + temperatures, err = rpc.getDiskTemperatures(ctx, identifiers) + return err + }) + return temperatures, err } func (c *Client) getDiskTemperatureAggregates(ctx context.Context, identifiers []string, windowDays int) (map[string]DiskTemperatureAggregate, error) { @@ -932,30 +975,25 @@ func (c *Client) getDiskTemperatureAggregates(ctx context.Context, identifiers [ return nil, fmt.Errorf("truenas disk temperature aggregates require disk identifiers") } - conn, err := c.dialRPC(ctx) - if err != nil { - return nil, err - } - defer func() { _ = conn.Close() }() - - rpc := &trueNASRPCClient{conn: conn, nextID: 1} - if err := rpc.authenticate(ctx, c.config); err != nil { - return nil, err - } - return rpc.getDiskTemperatureAggregates(ctx, identifiers, windowDays) + var aggregates map[string]DiskTemperatureAggregate + err := c.withRPC(ctx, func(rpc *trueNASRPCClient) error { + var err error + aggregates, err = rpc.getDiskTemperatureAggregates(ctx, identifiers, windowDays) + return err + }) + return aggregates, err } // GetAlerts returns active and dismissed TrueNAS alerts. func (c *Client) GetAlerts(ctx context.Context) ([]Alert, error) { - alerts, err := c.getAlertsRPC(ctx) - if err == nil { - return alerts, nil + legacy, err := c.useLegacyREST(ctx) + if err != nil { + return nil, err } - restAlerts, restErr := c.getAlertsREST(ctx) - if restErr != nil { - return nil, fmt.Errorf("fetch truenas alerts via rpc and rest: rpc=%w rest=%v", err, restErr) + if legacy { + return c.getAlertsREST(ctx) } - return restAlerts, nil + return c.getAlertsRPC(ctx) } func (c *Client) getAlertsRPC(ctx context.Context) ([]Alert, error) { @@ -1019,15 +1057,14 @@ func (c *Client) getAlertsREST(ctx context.Context) ([]Alert, error) { // GetServices returns the native TrueNAS system service inventory. func (c *Client) GetServices(ctx context.Context) ([]Service, error) { - services, err := c.getServicesRPC(ctx) - if err == nil { - return services, nil + legacy, err := c.useLegacyREST(ctx) + if err != nil { + return nil, err } - restServices, restErr := c.getServicesREST(ctx) - if restErr != nil { - return nil, fmt.Errorf("fetch truenas services via rpc and rest: rpc=%w rest=%v", err, restErr) + if legacy { + return c.getServicesREST(ctx) } - return restServices, nil + return c.getServicesRPC(ctx) } func (c *Client) getServicesRPC(ctx context.Context) ([]Service, error) { @@ -1068,37 +1105,22 @@ func parseServices(response []map[string]any) []Service { } // GetVMs returns the best-effort native TrueNAS VM inventory. TrueNAS 25.04+ -// documents vm.query on the JSON-RPC API, with the legacy REST endpoint kept -// as a compatibility fallback for existing client tests and older deployments. +// uses vm.query on JSON-RPC; recognized pre-25.04 SCALE and CORE releases use +// the connection's negotiated legacy REST transport. func (c *Client) GetVMs(ctx context.Context) ([]VirtualMachine, error) { - vms, err := c.getVMsRPC(ctx) - if err == nil { - return vms, nil - } - restVMs, restErr := c.getVMsREST(ctx) - if restErr != nil { - return nil, fmt.Errorf("fetch truenas vms via rpc and rest: rpc=%w rest=%v", err, restErr) - } - return restVMs, nil -} - -func (c *Client) getVMsRPC(ctx context.Context) ([]VirtualMachine, error) { - conn, err := c.dialRPC(ctx) + legacy, err := c.useLegacyREST(ctx) if err != nil { return nil, err } - defer func() { _ = conn.Close() }() - - rpc := trueNASRPCClient{ - conn: conn, - nextID: 1, - } - if err := rpc.authenticate(ctx, c.config); err != nil { - return nil, err + if legacy { + return c.getVMsREST(ctx) } + return c.getVMsRPC(ctx) +} +func (c *Client) getVMsRPC(ctx context.Context) ([]VirtualMachine, error) { var response []map[string]any - if err := rpc.call(ctx, "vm.query", []any{[]any{}, map[string]any{}}, &response); err != nil { + if err := c.queryRPC(ctx, "vm.query", &response); err != nil { return nil, err } return parseVirtualMachines(response), nil @@ -1113,16 +1135,21 @@ func (c *Client) getVMsREST(ctx context.Context) ([]VirtualMachine, error) { } // GetNetworkShares returns the best-effort native TrueNAS SMB/NFS sharing -// inventory. TrueNAS exposes the modern API as JSON-RPC query methods; the -// legacy REST endpoints are kept as compatibility fallbacks for older SCALE -// deployments and tests. +// inventory. Modern connections use JSON-RPC query methods; recognized legacy +// connections remain on REST for the lifetime of that client. func (c *Client) GetNetworkShares(ctx context.Context) ([]NetworkShare, error) { + legacy, err := c.useLegacyREST(ctx) + if err != nil { + return nil, err + } var shares []NetworkShare var errors []error - smb, err := c.getNetworkSharesRPC(ctx, "sharing.smb.query", "SMB") - if err != nil { + var smb []NetworkShare + if legacy { smb, err = c.getNetworkSharesREST(ctx, "/sharing/smb", "SMB") + } else { + smb, err = c.getNetworkSharesRPC(ctx, "sharing.smb.query", "SMB") } if err != nil { errors = append(errors, fmt.Errorf("smb: %w", err)) @@ -1130,9 +1157,11 @@ func (c *Client) GetNetworkShares(ctx context.Context) ([]NetworkShare, error) { shares = append(shares, smb...) } - nfs, err := c.getNetworkSharesRPC(ctx, "sharing.nfs.query", "NFS") - if err != nil { + var nfs []NetworkShare + if legacy { nfs, err = c.getNetworkSharesREST(ctx, "/sharing/nfs", "NFS") + } else { + nfs, err = c.getNetworkSharesRPC(ctx, "sharing.nfs.query", "NFS") } if err != nil { errors = append(errors, fmt.Errorf("nfs: %w", err)) @@ -1166,37 +1195,18 @@ func (c *Client) queryRPC(ctx context.Context, method string, result any) error return c.callRPC(ctx, method, []any{[]any{}, map[string]any{}}, result) } -func (c *Client) callRPC(ctx context.Context, method string, params any, result any) error { - conn, err := c.dialRPC(ctx) - if err != nil { - return err - } - defer func() { _ = conn.Close() }() - - rpc := trueNASRPCClient{ - conn: conn, - nextID: 1, - } - if err := rpc.authenticate(ctx, c.config); err != nil { - return err - } - return rpc.call(ctx, method, params, result) -} - // GetApps returns the best-effort TrueNAS app inventory as canonical workload -// candidates. TrueNAS 25.04+ documents app.query on the JSON-RPC API, with -// the legacy REST endpoint kept as a compatibility fallback for older -// deployments and existing tests. +// candidates. TrueNAS 25.04+ uses app.query on JSON-RPC; recognized legacy +// connections remain on their negotiated REST transport. func (c *Client) GetApps(ctx context.Context) ([]App, error) { - apps, err := c.getAppsRPC(ctx) - if err == nil { - return apps, nil + legacy, err := c.useLegacyREST(ctx) + if err != nil { + return nil, err } - restApps, restErr := c.getAppsREST(ctx) - if restErr != nil { - return nil, fmt.Errorf("fetch truenas apps via rpc and rest: rpc=%w rest=%v", err, restErr) + if legacy { + return c.getAppsREST(ctx) } - return restApps, nil + return c.getAppsRPC(ctx) } func (c *Client) getAppsRPC(ctx context.Context) ([]App, error) { @@ -1277,26 +1287,20 @@ func (c *Client) parseAppsWithStats(ctx context.Context, response []map[string]a // transport or endpoint failure as "stats unavailable" rather than inventory // failure. func (c *Client) GetAppStats(ctx context.Context) (map[string]AppStats, error) { - conn, err := c.dialRPC(ctx) - if err != nil { - return nil, err - } - defer func() { _ = conn.Close() }() - - rpc := trueNASRPCClient{ - conn: conn, - nextID: 1, - } - if err := rpc.authenticate(ctx, c.config); err != nil { - return nil, err - } - - subscriptionName := fmt.Sprintf("app.stats:{\"interval\":%d}", defaultAppStatsIntervalSeconds) - if err := rpc.call(ctx, "core.subscribe", []any{subscriptionName}, nil); err != nil { - return nil, err - } - - return rpc.readAppStatsEvent(ctx, defaultAppStatsIntervalSeconds) + var stats map[string]AppStats + err := c.withRPC(ctx, func(rpc *trueNASRPCClient) error { + subscriptionName := fmt.Sprintf("app.stats:{\"interval\":%d}", defaultAppStatsIntervalSeconds) + subscriptionID, err := rpc.subscribe(ctx, subscriptionName) + if err != nil { + return err + } + stats, err = rpc.readAppStatsEvent(ctx, defaultAppStatsIntervalSeconds) + if err != nil { + return discardRPCSessionForStreamError(err) + } + return rpc.unsubscribe(ctx, subscriptionID) + }) + return stats, err } // StartApp requests that TrueNAS start the named app through the canonical @@ -1329,20 +1333,6 @@ func (c *Client) GetAppLogs(ctx context.Context, appName, containerID string, ta tailLines = maxAppLogTailLines } - conn, err := c.dialRPC(ctx) - if err != nil { - return nil, err - } - defer func() { _ = conn.Close() }() - - rpc := trueNASRPCClient{ - conn: conn, - nextID: 1, - } - if err := rpc.authenticate(ctx, c.config); err != nil { - return nil, err - } - subscriptionArgs := map[string]any{ "app_name": appName, "container_id": containerID, @@ -1353,11 +1343,23 @@ func (c *Client) GetAppLogs(ctx context.Context, appName, containerID string, ta return nil, fmt.Errorf("marshal truenas app log subscription: %w", err) } subscriptionName := fmt.Sprintf("app.container_log_follow:%s", string(subscriptionJSON)) - if err := rpc.call(ctx, "core.subscribe", []any{subscriptionName}, nil); err != nil { - return nil, err - } - - return rpc.readAppLogEvents(ctx, tailLines) + var lines []AppLogLine + err = c.withRPC(ctx, func(rpc *trueNASRPCClient) error { + subscriptionID, err := rpc.subscribe(ctx, subscriptionName) + if err != nil { + return err + } + var reusable bool + lines, reusable, err = rpc.readAppLogEvents(ctx, tailLines) + if err != nil { + return discardRPCSessionForStreamError(err) + } + if !reusable { + return errRPCStreamSessionConsumed + } + return rpc.unsubscribe(ctx, subscriptionID) + }) + return lines, err } func (c *Client) executeAppAction(ctx context.Context, method, appID string) error { @@ -1366,49 +1368,38 @@ func (c *Client) executeAppAction(ctx context.Context, method, appID string) err return fmt.Errorf("truenas app id is required") } - conn, err := c.dialRPC(ctx) - if err != nil { - return err - } - defer func() { _ = conn.Close() }() - - rpc := trueNASRPCClient{ - conn: conn, - nextID: 1, - } - if err := rpc.authenticate(ctx, c.config); err != nil { - return err - } - if err := rpc.call(ctx, method, []any{appID}, nil); err != nil { + if err := c.callRPCAction(ctx, method, []any{appID}, nil); err != nil { return err } return nil } // GetZFSSnapshots returns a best-effort list of ZFS snapshots. Modern TrueNAS -// exposes snapshots through JSON-RPC; legacy REST remains a compatibility -// fallback for older deployments. +// uses JSON-RPC; recognized legacy connections remain on REST. func (c *Client) GetZFSSnapshots(ctx context.Context) ([]ZFSSnapshot, error) { - snapshots, err := c.getZFSSnapshotsRPC(ctx) - if err == nil { - return snapshots, nil + legacy, err := c.useLegacyREST(ctx) + if err != nil { + return nil, err } - restSnapshots, restErr := c.getZFSSnapshotsREST(ctx) - if restErr != nil { - return nil, fmt.Errorf("fetch truenas zfs snapshots via rpc and rest: rpc=%w rest=%v", err, restErr) + if legacy { + return c.getZFSSnapshotsREST(ctx) } - return restSnapshots, nil + return c.getZFSSnapshotsRPC(ctx) } func (c *Client) getZFSSnapshotsRPC(ctx context.Context) ([]ZFSSnapshot, error) { var response []map[string]any - if err := c.callRPC(ctx, "zfs.resource.snapshot.query", []any{map[string]any{ + err := c.callRPC(ctx, "zfs.resource.snapshot.query", []any{map[string]any{ "paths": []string{}, "recursive": true, "properties": []string{"creation", "used", "referenced"}, - }}, &response); err == nil { + }}, &response) + if err == nil { return parseZFSSnapshots(response), nil } + if !isMethodUnavailable(err) { + return nil, err + } if err := c.queryRPC(ctx, "pool.snapshot.query", &response); err != nil { return nil, err @@ -1489,15 +1480,14 @@ func parseZFSSnapshots(response []map[string]any) []ZFSSnapshot { // GetReplicationTasks returns a best-effort list of replication tasks including last-run state. func (c *Client) GetReplicationTasks(ctx context.Context) ([]ReplicationTask, error) { - tasks, err := c.getReplicationTasksRPC(ctx) - if err == nil { - return tasks, nil + legacy, err := c.useLegacyREST(ctx) + if err != nil { + return nil, err } - restTasks, restErr := c.getReplicationTasksREST(ctx) - if restErr != nil { - return nil, fmt.Errorf("fetch truenas replication tasks via rpc and rest: rpc=%w rest=%v", err, restErr) + if legacy { + return c.getReplicationTasksREST(ctx) } - return restTasks, nil + return c.getReplicationTasksRPC(ctx) } func (c *Client) getReplicationTasksRPC(ctx context.Context) ([]ReplicationTask, error) { @@ -1819,6 +1809,29 @@ type trueNASRPCClient struct { nextID int64 } +func (c *trueNASRPCClient) subscribe(ctx context.Context, event string) (string, error) { + var subscriptionID string + if err := c.call(ctx, "core.subscribe", []any{event}, &subscriptionID); err != nil { + return "", err + } + subscriptionID = strings.TrimSpace(subscriptionID) + if subscriptionID == "" { + return "", &discardRPCSessionError{err: fmt.Errorf("truenas rpc core.subscribe returned an empty subscription id")} + } + return subscriptionID, nil +} + +func (c *trueNASRPCClient) unsubscribe(ctx context.Context, subscriptionID string) error { + subscriptionID = strings.TrimSpace(subscriptionID) + if subscriptionID == "" { + return fmt.Errorf("truenas rpc subscription id is required") + } + if err := c.call(ctx, "core.unsubscribe", []any{subscriptionID}, nil); err != nil { + return &discardRPCSessionError{err: fmt.Errorf("unsubscribe %q: %w", subscriptionID, err)} + } + return nil +} + type trueNASRPCRequest struct { JSONRPC string `json:"jsonrpc"` ID int64 `json:"id"` @@ -1836,8 +1849,9 @@ type trueNASRPCResponse struct { } type trueNASRPCError struct { - Code int `json:"code"` - Message string `json:"message"` + Code int `json:"code"` + Message string `json:"message"` + Data json.RawMessage `json:"data,omitempty"` } type trueNASCollectionUpdate struct { @@ -1894,7 +1908,6 @@ func (c *Client) dialRPC(ctx context.Context) (*websocket.Conn, error) { if c == nil { return nil, fmt.Errorf("truenas client is nil") } - dialer := websocket.Dialer{ Proxy: http.ProxyFromEnvironment, } @@ -1914,28 +1927,88 @@ func (c *Client) dialRPC(ctx context.Context) (*websocket.Conn, error) { conn, response, err := dialer.DialContext(ctx, c.rpcURL, nil) if err != nil { + statusCode := 0 if response != nil { - return nil, fmt.Errorf("dial truenas rpc websocket: status=%d: %w", response.StatusCode, err) + statusCode = response.StatusCode } - return nil, fmt.Errorf("dial truenas rpc websocket: %w", err) + return nil, &RPCHandshakeError{StatusCode: statusCode, Err: err} } return conn, nil } -func (c *trueNASRPCClient) authenticate(ctx context.Context, config ClientConfig) error { +func (c *trueNASRPCClient) authenticate(ctx context.Context, config ClientConfig) (string, error) { if apiKey := strings.TrimSpace(config.APIKey); apiKey != "" { - return c.call(ctx, "auth.login_with_api_key", []any{apiKey}, nil) + if username := strings.TrimSpace(config.Username); username != "" { + var response struct { + ResponseType string `json:"response_type"` + } + err := c.call(ctx, "auth.login_ex", []any{map[string]any{ + "mechanism": "API_KEY_PLAIN", + "username": username, + "api_key": apiKey, + "login_options": map[string]any{ + "user_info": false, + }, + }}, &response) + if err != nil { + return "", err + } + if !strings.EqualFold(strings.TrimSpace(response.ResponseType), "SUCCESS") { + return "", &RPCAuthError{ + Mechanism: "api-key-plain", + ResponseType: strings.TrimSpace(response.ResponseType), + } + } + return "api-key-plain", nil + } + + var authenticated bool + if err := c.call(ctx, "auth.login_with_api_key", []any{apiKey}, &authenticated); err != nil { + if isMethodUnavailable(err) { + return "", fmt.Errorf("truenas rpc API-key authentication on this release requires the API key owner username; edit this connection and add it: %w", err) + } + return "", err + } + if !authenticated { + return "", &RPCAuthError{Mechanism: "legacy-api-key"} + } + return "legacy-api-key", nil } if config.Username != "" || config.Password != "" { - return c.call(ctx, "auth.login", []any{config.Username, config.Password}, nil) + var response struct { + ResponseType string `json:"response_type"` + } + err := c.call(ctx, "auth.login_ex", []any{map[string]any{ + "mechanism": "PASSWORD_PLAIN", + "username": config.Username, + "password": config.Password, + "login_options": map[string]any{ + "user_info": false, + }, + }}, &response) + if err != nil { + return "", err + } + if !strings.EqualFold(strings.TrimSpace(response.ResponseType), "SUCCESS") { + return "", &RPCAuthError{ + Mechanism: "password-plain", + ResponseType: strings.TrimSpace(response.ResponseType), + } + } + return "password-plain", nil + } + return "", &RPCAuthError{ + Mechanism: "configuration", + ResponseType: "credentials required", } - return fmt.Errorf("truenas rpc authentication requires api key or username/password") } func (c *trueNASRPCClient) call(ctx context.Context, method string, params any, result any) error { if c == nil || c.conn == nil { return fmt.Errorf("truenas rpc connection is nil") } + stopContext := c.armContext(ctx) + defer stopContext() request := trueNASRPCRequest{ JSONRPC: "2.0", @@ -1949,7 +2022,7 @@ func (c *trueNASRPCClient) call(ctx context.Context, method string, params any, _ = c.conn.SetWriteDeadline(deadline) } if err := c.conn.WriteJSON(request); err != nil { - return fmt.Errorf("write truenas rpc %s request: %w", method, err) + return &RPCTransportError{Method: method, Phase: "write", Err: err} } for { @@ -1959,7 +2032,7 @@ func (c *trueNASRPCClient) call(ctx context.Context, method string, params any, var message trueNASRPCResponse if err := c.conn.ReadJSON(&message); err != nil { - return fmt.Errorf("read truenas rpc %s response: %w", method, err) + return &RPCTransportError{Method: method, Phase: "read", Err: err} } if message.Method != "" { continue @@ -1968,7 +2041,7 @@ func (c *trueNASRPCClient) call(ctx context.Context, method string, params any, continue } if message.Error != nil { - return fmt.Errorf("truenas rpc %s failed: code=%d message=%q", method, message.Error.Code, strings.TrimSpace(message.Error.Message)) + return rpcErrorFromWire(method, message.Error) } if result == nil || len(message.Result) == 0 || string(message.Result) == "null" { return nil @@ -1984,6 +2057,8 @@ func (c *trueNASRPCClient) readAppStatsEvent(ctx context.Context, intervalSecond if c == nil || c.conn == nil { return nil, fmt.Errorf("truenas rpc connection is nil") } + stopContext := c.armContext(ctx) + defer stopContext() for { if deadline, ok := ctx.Deadline(); ok { @@ -1992,11 +2067,11 @@ func (c *trueNASRPCClient) readAppStatsEvent(ctx context.Context, intervalSecond var message trueNASRPCResponse if err := c.conn.ReadJSON(&message); err != nil { - return nil, fmt.Errorf("read truenas rpc app.stats notification: %w", err) + return nil, &RPCTransportError{Method: "app.stats", Phase: "read", Err: err} } if message.Method == "" { if message.Error != nil { - return nil, fmt.Errorf("truenas rpc app.stats failed: code=%d message=%q", message.Error.Code, strings.TrimSpace(message.Error.Message)) + return nil, rpcErrorFromWire("app.stats", message.Error) } continue } @@ -2046,10 +2121,12 @@ func (c *trueNASRPCClient) readAppStatsEvent(ctx context.Context, intervalSecond } } -func (c *trueNASRPCClient) readAppLogEvents(ctx context.Context, tailLines int) ([]AppLogLine, error) { +func (c *trueNASRPCClient) readAppLogEvents(ctx context.Context, tailLines int) ([]AppLogLine, bool, error) { if c == nil || c.conn == nil { - return nil, fmt.Errorf("truenas rpc connection is nil") + return nil, false, fmt.Errorf("truenas rpc connection is nil") } + stopContext := c.armContext(ctx) + defer stopContext() if tailLines <= 0 { tailLines = 100 } @@ -2074,13 +2151,16 @@ func (c *trueNASRPCClient) readAppLogEvents(ctx context.Context, tailLines int) var message trueNASRPCResponse if err := c.conn.ReadJSON(&message); err != nil { if isTimeoutError(err) { - return trimAppLogLines(lines, tailLines), nil + // Gorilla WebSocket documents a timed-out read as terminal for + // the connection. Preserve the collected log data, then make + // the caller discard this stream session instead of reusing it. + return trimAppLogLines(lines, tailLines), false, nil } - return nil, fmt.Errorf("read truenas rpc app.container_log_follow notification: %w", err) + return nil, false, &RPCTransportError{Method: "app.container_log_follow", Phase: "read", Err: err} } if message.Method == "" { if message.Error != nil { - return nil, fmt.Errorf("truenas rpc app.container_log_follow failed: code=%d message=%q", message.Error.Code, strings.TrimSpace(message.Error.Message)) + return nil, false, rpcErrorFromWire("app.container_log_follow", message.Error) } continue } @@ -2097,7 +2177,7 @@ func (c *trueNASRPCClient) readAppLogEvents(ctx context.Context, tailLines int) if len(appended) > len(lines) { lines = appended if len(lines) >= tailLines { - return trimAppLogLines(lines, tailLines), nil + return trimAppLogLines(lines, tailLines), true, nil } idleDeadline = time.Now().Add(defaultAppLogIdleWait) } @@ -2106,13 +2186,13 @@ func (c *trueNASRPCClient) readAppLogEvents(ctx context.Context, tailLines int) var raw any if err := json.Unmarshal(message.Params, &raw); err != nil { - return nil, fmt.Errorf("decode truenas app.container_log_follow notification: %w", err) + return nil, false, fmt.Errorf("decode truenas app.container_log_follow notification: %w", err) } appended := appendAppLogLines(lines, raw) if len(appended) > len(lines) { lines = appended if len(lines) >= tailLines { - return trimAppLogLines(lines, tailLines), nil + return trimAppLogLines(lines, tailLines), true, nil } idleDeadline = time.Now().Add(defaultAppLogIdleWait) } @@ -2180,6 +2260,53 @@ func isTimeoutError(err error) bool { return errors.As(err, &netErr) && netErr.Timeout() } +// armContext makes websocket I/O observe cancellation even when a caller +// supplies a cancellable context without a deadline. The cleanup waits for the +// cancellation watcher before clearing deadlines, preventing a completed call +// from poisoning the next serialized exchange on this connection. +func (c *trueNASRPCClient) armContext(ctx context.Context) func() { + if c == nil || c.conn == nil { + return func() {} + } + if ctx == nil { + ctx = context.Background() + } + if deadline, ok := ctx.Deadline(); ok { + _ = c.conn.SetReadDeadline(deadline) + _ = c.conn.SetWriteDeadline(deadline) + } else { + _ = c.conn.SetReadDeadline(time.Time{}) + _ = c.conn.SetWriteDeadline(time.Time{}) + } + + cancelled := ctx.Done() + if cancelled == nil { + return func() { + _ = c.conn.SetReadDeadline(time.Time{}) + _ = c.conn.SetWriteDeadline(time.Time{}) + } + } + + stop := make(chan struct{}) + stopped := make(chan struct{}) + go func() { + defer close(stopped) + select { + case <-cancelled: + now := time.Now() + _ = c.conn.SetReadDeadline(now) + _ = c.conn.SetWriteDeadline(now) + case <-stop: + } + }() + return func() { + close(stop) + <-stopped + _ = c.conn.SetReadDeadline(time.Time{}) + _ = c.conn.SetWriteDeadline(time.Time{}) + } +} + func (c *trueNASRPCClient) getSystemMetricHistory(ctx context.Context, duration time.Duration) (*SystemMetricHistory, error) { if c == nil || c.conn == nil { return nil, fmt.Errorf("truenas rpc connection is nil") @@ -2354,6 +2481,8 @@ func (c *trueNASRPCClient) readSystemTelemetryEvent(ctx context.Context, interva if c == nil || c.conn == nil { return nil, fmt.Errorf("truenas rpc connection is nil") } + stopContext := c.armContext(ctx) + defer stopContext() for { if deadline, ok := ctx.Deadline(); ok { @@ -2362,11 +2491,11 @@ func (c *trueNASRPCClient) readSystemTelemetryEvent(ctx context.Context, interva var message trueNASRPCResponse if err := c.conn.ReadJSON(&message); err != nil { - return nil, fmt.Errorf("read truenas rpc reporting.realtime notification: %w", err) + return nil, &RPCTransportError{Method: "reporting.realtime", Phase: "read", Err: err} } if message.Method == "" { if message.Error != nil { - return nil, fmt.Errorf("truenas rpc reporting.realtime failed: code=%d message=%q", message.Error.Code, strings.TrimSpace(message.Error.Message)) + return nil, rpcErrorFromWire("reporting.realtime", message.Error) } continue } diff --git a/internal/truenas/client_test.go b/internal/truenas/client_test.go index aca1a5aa8..a81816d32 100644 --- a/internal/truenas/client_test.go +++ b/internal/truenas/client_test.go @@ -152,8 +152,11 @@ func TestClientGetters(t *testing.T) { func TestClientAuthHeaderAPIKey(t *testing.T) { server := newMockServer(t, map[string]apiResponse{ - "/api/v2.0/system/info": {body: `{"hostname":"nas","version":"v","buildtime":"b","uptime_seconds":1}`}, + "/api/v2.0/system/info": {body: `{"hostname":"nas","version":"TrueNAS-SCALE-24.10.2","buildtime":"b","uptime_seconds":1}`}, }, func(t *testing.T, request *http.Request) { + if request.URL.Path == "/api/current" { + return + } if got := request.Header.Get("Authorization"); got != "Bearer test-key" { t.Fatalf("expected bearer auth header, got %q", got) } @@ -168,8 +171,11 @@ func TestClientAuthHeaderAPIKey(t *testing.T) { func TestClientAuthHeaderBasic(t *testing.T) { server := newMockServer(t, map[string]apiResponse{ - "/api/v2.0/system/info": {body: `{"hostname":"nas","version":"v","buildtime":"b","uptime_seconds":1}`}, + "/api/v2.0/system/info": {body: `{"hostname":"nas","version":"TrueNAS-SCALE-24.10.2","buildtime":"b","uptime_seconds":1}`}, }, func(t *testing.T, request *http.Request) { + if request.URL.Path == "/api/current" { + return + } username, password, ok := request.BasicAuth() if !ok { t.Fatalf("expected basic auth") @@ -187,36 +193,22 @@ func TestClientAuthHeaderBasic(t *testing.T) { } func TestGetSystemInfoAcceptsStructuredBuildTime(t *testing.T) { - server := newMockServer(t, map[string]apiResponse{ - "/api/v2.0/system/info": { - body: `{"hostname":"nas","version":"TrueNAS-SCALE-25.10.3.1","buildtime":{"$date":"2026-05-14T18:24:01+02:00"},"uptime_seconds":1}`, - }, - }, nil) - t.Cleanup(server.Close) - - client := mustClientForServer(t, server.URL, ClientConfig{APIKey: "test-key"}) - system, err := client.GetSystemInfo(context.Background()) - if err != nil { - t.Fatalf("GetSystemInfo() error = %v", err) + var response systemInfoResponse + if err := json.Unmarshal([]byte(`{"hostname":"nas","version":"TrueNAS-SCALE-25.10.3.1","buildtime":{"$date":"2026-05-14T18:24:01+02:00"},"uptime_seconds":1}`), &response); err != nil { + t.Fatalf("Unmarshal() error = %v", err) } + system := systemInfoFromResponse(response) if system.Build != "2026-05-14T18:24:01+02:00" { t.Fatalf("Build = %q, want structured buildtime date", system.Build) } } func TestGetSystemInfoAcceptsFractionalUptimeSeconds(t *testing.T) { - server := newMockServer(t, map[string]apiResponse{ - "/api/v2.0/system/info": { - body: `{"hostname":"nas","version":"TrueNAS-SCALE-25.10.3.1","buildtime":{"$date":"2026-05-14T18:24:01+02:00"},"uptime_seconds":360144.629139547}`, - }, - }, nil) - t.Cleanup(server.Close) - - client := mustClientForServer(t, server.URL, ClientConfig{APIKey: "test-key"}) - system, err := client.GetSystemInfo(context.Background()) - if err != nil { - t.Fatalf("GetSystemInfo() error = %v", err) + var response systemInfoResponse + if err := json.Unmarshal([]byte(`{"hostname":"nas","version":"TrueNAS-SCALE-25.10.3.1","buildtime":{"$date":"2026-05-14T18:24:01+02:00"},"uptime_seconds":360144.629139547}`), &response); err != nil { + t.Fatalf("Unmarshal() error = %v", err) } + system := systemInfoFromResponse(response) if system.UptimeSeconds != 360144 { t.Fatalf("UptimeSeconds = %d, want truncated fractional uptime", system.UptimeSeconds) } @@ -224,7 +216,7 @@ func TestGetSystemInfoAcceptsFractionalUptimeSeconds(t *testing.T) { func TestTestConnectionSuccessAndFailure(t *testing.T) { successServer := newMockServer(t, map[string]apiResponse{ - "/api/v2.0/system/info": {body: `{"hostname":"nas","version":"v","buildtime":"b","uptime_seconds":1}`}, + "/api/v2.0/system/info": {body: `{"hostname":"nas","version":"TrueNAS-SCALE-24.10.2","buildtime":"b","uptime_seconds":1}`}, }, nil) t.Cleanup(successServer.Close) @@ -667,6 +659,7 @@ func TestGetAppsEnrichesStatsFromRPC(t *testing.T) { }, }, }) + expectRPCUnsubscribe(t, conn, "sub-1") }) t.Cleanup(server.Close) @@ -813,24 +806,22 @@ func TestGetReplicationTasksUsesNativeReplicationQueryShape(t *testing.T) { } func TestStartAndStopAppUseRPCMethods(t *testing.T) { - var rpcCalls int + var rpcSessions int server := newMockServerWithRPC(t, defaultAPIResponses(), nil, func(t *testing.T, conn *websocket.Conn) { - rpcCalls++ + rpcSessions++ authReq := readRPCRequest(t, conn) if authReq.Method != "auth.login_with_api_key" { t.Fatalf("expected api-key auth method, got %q", authReq.Method) } writeRPCResult(t, conn, authReq.ID, true) - actionReq := readRPCRequest(t, conn) - expectedMethod := "app.start" - if rpcCalls == 2 { - expectedMethod = "app.stop" + for _, expectedMethod := range []string{"app.start", "app.stop"} { + actionReq := readRPCRequest(t, conn) + if actionReq.Method != expectedMethod { + t.Fatalf("expected %s, got %q", expectedMethod, actionReq.Method) + } + writeRPCResult(t, conn, actionReq.ID, true) } - if actionReq.Method != expectedMethod { - t.Fatalf("expected %s, got %q", expectedMethod, actionReq.Method) - } - writeRPCResult(t, conn, actionReq.ID, true) }) t.Cleanup(server.Close) @@ -841,8 +832,8 @@ func TestStartAndStopAppUseRPCMethods(t *testing.T) { if err := client.StopApp(context.Background(), "nextcloud"); err != nil { t.Fatalf("StopApp() error = %v", err) } - if rpcCalls != 2 { - t.Fatalf("expected two RPC app-action sessions, got %d", rpcCalls) + if rpcSessions != 1 { + t.Fatalf("expected one persistent RPC app-action session, got %d", rpcSessions) } } @@ -884,6 +875,7 @@ func TestGetAppLogsUsesRPCSubscription(t *testing.T) { "timestamp": "2026-03-29T18:01:00Z", }, }) + expectRPCUnsubscribe(t, conn, "sub-logs") time.Sleep(defaultAppLogIdleWait + 100*time.Millisecond) }) t.Cleanup(server.Close) @@ -967,6 +959,7 @@ func TestGetSystemTelemetryFromRPC(t *testing.T) { }, }, }) + expectRPCUnsubscribe(t, conn, "sub-1") }) t.Cleanup(server.Close) @@ -1130,6 +1123,7 @@ func TestGetSystemTelemetryIgnoresUnavailableTemperatureRPC(t *testing.T) { "cpu": map[string]any{"usage": 41}, }, }) + expectRPCUnsubscribe(t, conn, "sub-1") }) t.Cleanup(server.Close) @@ -1199,6 +1193,7 @@ func TestGetDisksToleratesUnavailableTemperatureEndpoint(t *testing.T) { } func TestGetDiskTemperaturesFallsBackToReportingRPC(t *testing.T) { + connectionCount := 0 server := newMockServerWithRPC(t, map[string]apiResponse{ "/api/v2.0/disk": { body: `[{"identifier":"{disk-1}","name":"sda","serial":"SER-A","size":1000000,"model":"Seagate","type":"HDD","pool":"tank","bus":"SATA","rotationrate":7200,"status":"ONLINE"}]`, @@ -1214,7 +1209,15 @@ func TestGetDiskTemperaturesFallsBackToReportingRPC(t *testing.T) { } writeRPCResult(t, conn, authReq.ID, true) + connectionCount++ temperatureReq := readRPCRequest(t, conn) + if connectionCount == 1 { + if temperatureReq.Method != "disk.query" { + t.Fatalf("expected disk.query, got %q", temperatureReq.Method) + } + writeRPCResult(t, conn, temperatureReq.ID, defaultRoutePayloadMaps(t, "/api/v2.0/disk")[:1]) + return + } if temperatureReq.Method != "reporting.get_data" { t.Fatalf("expected reporting.get_data, got %q", temperatureReq.Method) } @@ -1613,7 +1616,7 @@ func TestClientTLSFingerprintPinning(t *testing.T) { return } writer.Header().Set("Content-Type", "application/json") - _, _ = writer.Write([]byte(`{"hostname":"nas","version":"v","buildtime":"b","uptime_seconds":1}`)) + _, _ = writer.Write([]byte(`{"hostname":"nas","version":"TrueNAS-SCALE-24.10.2","buildtime":"b","uptime_seconds":1}`)) }) tlsServer := httptest.NewTLSServer(handler) @@ -1741,6 +1744,11 @@ func newMockServer(t *testing.T, responses map[string]apiResponse, assertRequest response, ok := responses[request.URL.Path] if !ok { + if request.URL.Path == "/api/v2.0/system/info" { + writer.Header().Set("Content-Type", "application/json") + _, _ = writer.Write([]byte(`{"hostname":"legacy-test","version":"TrueNAS-SCALE-24.10.2","buildtime":"24.10.2","uptime_seconds":1}`)) + return + } http.NotFound(writer, request) return } @@ -1857,6 +1865,20 @@ func writeRPCNotification(t *testing.T, conn *websocket.Conn, method string, par } } +func expectRPCUnsubscribe(t *testing.T, conn *websocket.Conn, subscriptionID string) { + t.Helper() + + request := readRPCRequest(t, conn) + if request.Method != "core.unsubscribe" { + t.Fatalf("expected core.unsubscribe, got %q", request.Method) + } + params, ok := request.Params.([]any) + if !ok || len(params) != 1 || params[0] != subscriptionID { + t.Fatalf("core.unsubscribe params = %#v, want [%q]", request.Params, subscriptionID) + } + writeRPCResult(t, conn, request.ID, nil) +} + func mustClientForServer(t *testing.T, serverURL string, config ClientConfig) *Client { t.Helper() @@ -1877,6 +1899,7 @@ func mustClientForServer(t *testing.T, serverURL string, config ClientConfig) *C if err != nil { t.Fatalf("NewClient() error = %v", err) } + client.allowInsecureRPC = true return client } diff --git a/internal/truenas/provider.go b/internal/truenas/provider.go index e8e16bbe1..dd8633f2b 100644 --- a/internal/truenas/provider.go +++ b/internal/truenas/provider.go @@ -72,6 +72,10 @@ type systemMetricHistoryFetcher interface { SystemMetricHistory(ctx context.Context, duration time.Duration) (*SystemMetricHistory, error) } +type transportStatusFetcher interface { + TransportStatus() TransportStatus +} + // APIFetcher loads snapshots from the live TrueNAS API client. type APIFetcher struct { Client *Client @@ -128,6 +132,15 @@ func (f *APIFetcher) SystemMetricHistory(ctx context.Context, duration time.Dura return f.Client.GetSystemMetricHistory(ctx, duration) } +// TransportStatus returns the underlying client's non-secret connection-local +// transport diagnostics. +func (f *APIFetcher) TransportStatus() TransportStatus { + if f == nil || f.Client == nil { + return TransportStatus{Mode: TransportUnknown} + } + return f.Client.TransportStatus() +} + // FixtureFetcher loads snapshots from static fixture data. type FixtureFetcher struct { Snapshot FixtureSnapshot @@ -210,6 +223,18 @@ func (p *Provider) Refresh(ctx context.Context) error { return nil } +// TransportStatus returns non-secret diagnostics when the provider is backed +// by the live API client. Fixture providers remain in the negotiating state. +func (p *Provider) TransportStatus() TransportStatus { + if p == nil { + return TransportStatus{Mode: TransportUnknown} + } + if source, ok := p.fetcher.(transportStatusFetcher); ok { + return source.TransportStatus() + } + return TransportStatus{Mode: TransportUnknown} +} + // ControlApp executes a native start/stop action against a TrueNAS app and // refreshes the cached snapshot so downstream readers observe canonical state. func (p *Provider) ControlApp(ctx context.Context, appID, action string) (*FixtureSnapshot, error) { diff --git a/internal/truenas/transport.go b/internal/truenas/transport.go new file mode 100644 index 000000000..afdc412e7 --- /dev/null +++ b/internal/truenas/transport.go @@ -0,0 +1,556 @@ +package truenas + +import ( + "context" + "encoding/json" + "errors" + "fmt" + "net/http" + "strconv" + "strings" + "time" +) + +var errRPCStreamSessionConsumed = errors.New("truenas rpc stream session cannot be reused") + +type discardRPCSessionError struct { + err error +} + +func (e *discardRPCSessionError) Error() string { + return fmt.Sprintf("discard truenas rpc session: %v", e.err) +} + +func (e *discardRPCSessionError) Unwrap() error { return e.err } + +func discardRPCSessionForStreamError(err error) error { + if err == nil { + return nil + } + var transportErr *RPCTransportError + if errors.As(err, &transportErr) { + return err + } + return &discardRPCSessionError{err: err} +} + +// TransportMode is the connection-local API transport selected for an +// appliance. A client never changes modes after successful negotiation. +type TransportMode string + +const ( + TransportUnknown TransportMode = "negotiating" + TransportJSONRPC TransportMode = "jsonrpc-websocket" + TransportLegacyREST TransportMode = "legacy-rest" +) + +// TransportStatus is a non-secret diagnostic projection of transport and +// authentication state for one appliance. +type TransportStatus struct { + Mode TransportMode `json:"mode"` + Endpoint string `json:"endpoint,omitempty"` + TLS bool `json:"tls"` + Connected bool `json:"connected"` + AuthMechanism string `json:"authMechanism,omitempty"` + ApplianceVersion string `json:"applianceVersion,omitempty"` + LegacyReason string `json:"legacyReason,omitempty"` + Reconnects int `json:"reconnects,omitempty"` + LastError string `json:"lastError,omitempty"` + LastConnectedAt *time.Time `json:"lastConnectedAt,omitempty"` +} + +// RPCError is a structured JSON-RPC method error. Data is intentionally +// reduced to non-secret middleware diagnostics. +type RPCError struct { + Code int + Method string + Message string + Reason string + Errname string +} + +func (e *RPCError) Error() string { + if e == nil { + return "truenas rpc error" + } + detail := strings.TrimSpace(e.Reason) + if detail == "" { + detail = strings.TrimSpace(e.Message) + } + if e.Errname != "" { + return fmt.Sprintf("truenas rpc %s failed: code=%d errname=%s message=%q", e.Method, e.Code, e.Errname, detail) + } + return fmt.Sprintf("truenas rpc %s failed: code=%d message=%q", e.Method, e.Code, detail) +} + +// RPCTransportError distinguishes a failed websocket exchange from an +// authoritative middleware method error. +type RPCTransportError struct { + Method string + Phase string + Err error +} + +func (e *RPCTransportError) Error() string { + return fmt.Sprintf("%s truenas rpc %s %s: %v", e.Phase, e.Method, transportPhaseNoun(e.Phase), e.Err) +} + +func (e *RPCTransportError) Unwrap() error { return e.Err } + +func transportPhaseNoun(phase string) string { + if phase == "write" { + return "request" + } + return "response" +} + +type RPCHandshakeError struct { + StatusCode int + Err error +} + +func (e *RPCHandshakeError) Error() string { + if e.StatusCode > 0 { + return fmt.Sprintf("dial truenas rpc websocket failed: status=%d: %v", e.StatusCode, e.Err) + } + return fmt.Sprintf("dial truenas rpc websocket failed: %v", e.Err) +} + +func (e *RPCHandshakeError) Unwrap() error { return e.Err } + +// RPCAuthError is an authoritative authentication refusal returned after the +// websocket handshake. It is distinct from a retryable transport failure and +// intentionally carries no credential material. +type RPCAuthError struct { + Mechanism string + ResponseType string +} + +func (e *RPCAuthError) Error() string { + if e == nil { + return "truenas rpc authentication failed" + } + if strings.TrimSpace(e.ResponseType) == "" { + return fmt.Sprintf("truenas rpc %s authentication failed", e.Mechanism) + } + return fmt.Sprintf("truenas rpc %s authentication failed: response_type=%q", e.Mechanism, e.ResponseType) +} + +// TransportStatus returns a secret-free snapshot. It never waits for an +// in-flight API call. +func (c *Client) TransportStatus() TransportStatus { + if c == nil { + return TransportStatus{Mode: TransportUnknown} + } + c.statusMu.RLock() + defer c.statusMu.RUnlock() + status := c.status + if status.LastConnectedAt != nil { + value := *status.LastConnectedAt + status.LastConnectedAt = &value + } + return status +} + +func (c *Client) updateTransportStatus(update func(*TransportStatus)) { + if c == nil { + return + } + c.statusMu.Lock() + update(&c.status) + c.statusMu.Unlock() +} + +func (c *Client) ensureTransport(ctx context.Context) (TransportMode, error) { + if c == nil { + return TransportUnknown, fmt.Errorf("truenas client is nil") + } + c.rpcMu.Lock() + defer c.rpcMu.Unlock() + return c.ensureTransportLocked(ctx) +} + +func (c *Client) useLegacyREST(ctx context.Context) (bool, error) { + mode, err := c.ensureTransport(ctx) + return mode == TransportLegacyREST, err +} + +func (c *Client) ensureTransportLocked(ctx context.Context) (TransportMode, error) { + if c.closed { + return TransportUnknown, fmt.Errorf("truenas client is closed") + } + if c.mode == TransportLegacyREST { + return c.mode, nil + } + if c.mode == TransportJSONRPC && c.rpc != nil { + return c.mode, nil + } + + reconnecting := c.mode == TransportJSONRPC + rpc, authMechanism, err := c.openAuthenticatedRPC(ctx) + if err == nil { + now := time.Now().UTC() + c.rpc = rpc + c.mode = TransportJSONRPC + c.updateTransportStatus(func(status *TransportStatus) { + status.Mode = TransportJSONRPC + status.Endpoint = c.rpcURL + status.Connected = true + status.AuthMechanism = authMechanism + status.LegacyReason = "" + status.LastError = "" + status.LastConnectedAt = &now + if reconnecting { + status.Reconnects++ + } + }) + return c.mode, nil + } + + if reconnecting { + lockedErr := fmt.Errorf("truenas JSON-RPC reconnect failed; transport is immutable for this client and REST downgrade is disabled: %w", err) + c.recordTransportError(lockedErr) + return TransportJSONRPC, lockedErr + } + + if !isUnsupportedWebSocketEndpoint(err) { + c.recordTransportError(err) + return TransportUnknown, err + } + + var response systemInfoResponse + if restErr := c.getJSON(ctx, http.MethodGet, "/system/info", &response); restErr != nil { + joined := fmt.Errorf("truenas JSON-RPC endpoint is unavailable and legacy version probe failed: websocket=%w rest=%v", err, restErr) + c.recordTransportError(joined) + return TransportUnknown, joined + } + if !supportsLegacyREST(response.Version) { + version := strings.TrimSpace(response.Version) + if version == "" { + version = "unknown" + } + rejected := fmt.Errorf("truenas %s requires JSON-RPC WebSocket at /api/current; refusing REST downgrade after websocket endpoint failure", version) + c.recordTransportError(rejected) + return TransportUnknown, rejected + } + + c.mode = TransportLegacyREST + version := strings.TrimSpace(response.Version) + c.updateTransportStatus(func(status *TransportStatus) { + status.Mode = TransportLegacyREST + status.Endpoint = c.baseURL + status.Connected = true + status.AuthMechanism = legacyRESTAuthMechanism(c.config) + status.ApplianceVersion = version + status.LegacyReason = "JSON-RPC /api/current is unavailable on this recognized legacy TrueNAS release" + status.LastError = "" + }) + return c.mode, nil +} + +func (c *Client) openAuthenticatedRPC(ctx context.Context) (*trueNASRPCClient, string, error) { + conn, err := c.dialRPC(ctx) + if err != nil { + return nil, "", err + } + if !c.config.UseHTTPS && !c.allowInsecureRPC { + _ = conn.Close() + return nil, "", fmt.Errorf("truenas JSON-RPC credentials require TLS; configure an https endpoint (certificate verification can be pinned or explicitly disabled)") + } + rpc := &trueNASRPCClient{conn: conn, nextID: 1} + authMechanism, err := rpc.authenticate(ctx, c.config) + if err != nil { + _ = conn.Close() + return nil, "", err + } + return rpc, authMechanism, nil +} + +func (c *Client) callRPC(ctx context.Context, method string, params any, result any) error { + return c.callRPCWithRetry(ctx, method, params, result, true) +} + +func (c *Client) withRPC(ctx context.Context, operation func(*trueNASRPCClient) error) error { + if c == nil { + return fmt.Errorf("truenas client is nil") + } + c.rpcMu.Lock() + defer c.rpcMu.Unlock() + + mode, err := c.ensureTransportLocked(ctx) + if err != nil { + return err + } + if mode != TransportJSONRPC || c.rpc == nil { + return fmt.Errorf("truenas JSON-RPC operation is unavailable over negotiated transport %s", mode) + } + if err := operation(c.rpc); err != nil { + if errors.Is(err, errRPCStreamSessionConsumed) { + c.closeRPCLocked() + c.reconnect = 0 + return nil + } + var discardErr *discardRPCSessionError + if errors.As(err, &discardErr) { + c.closeRPCLocked() + c.recordTransportError(err) + return err + } + var transportErr *RPCTransportError + if !errors.As(err, &transportErr) { + c.recordRPCOperationError(err) + return err + } + c.closeRPCLocked() + c.recordTransportError(err) + if err := c.waitReconnectBackoff(ctx); err != nil { + return err + } + rpc, authMechanism, reconnectErr := c.openAuthenticatedRPC(ctx) + if reconnectErr != nil { + c.recordTransportError(reconnectErr) + return reconnectErr + } + c.rpc = rpc + c.reconnect++ + now := time.Now().UTC() + c.updateTransportStatus(func(status *TransportStatus) { + status.Connected = true + status.AuthMechanism = authMechanism + status.Reconnects++ + status.LastConnectedAt = &now + status.LastError = "" + }) + if err := operation(c.rpc); err != nil { + var secondTransportErr *RPCTransportError + if errors.As(err, &secondTransportErr) { + c.closeRPCLocked() + c.recordTransportError(err) + } else { + c.recordRPCOperationError(err) + } + return err + } + } + c.reconnect = 0 + return nil +} + +func (c *Client) callRPCAction(ctx context.Context, method string, params any, result any) error { + return c.callRPCWithRetry(ctx, method, params, result, false) +} + +func (c *Client) callRPCWithRetry(ctx context.Context, method string, params any, result any, retryRead bool) error { + if c == nil { + return fmt.Errorf("truenas client is nil") + } + c.rpcMu.Lock() + defer c.rpcMu.Unlock() + + mode, err := c.ensureTransportLocked(ctx) + if err != nil { + return err + } + if mode != TransportJSONRPC || c.rpc == nil { + return fmt.Errorf("truenas rpc %s is unavailable over negotiated transport %s", method, mode) + } + + err = c.rpc.call(ctx, method, params, result) + if err == nil { + c.reconnect = 0 + c.updateTransportStatus(func(status *TransportStatus) { + status.Connected = true + status.LastError = "" + }) + return nil + } + var transportErr *RPCTransportError + if !errors.As(err, &transportErr) { + c.recordRPCOperationError(err) + return err + } + + c.closeRPCLocked() + c.recordTransportError(err) + if !retryRead { + return fmt.Errorf("truenas action %s transport failed after dispatch; outcome is unknown and Pulse will not replay it: %w", method, err) + } + + if err := c.waitReconnectBackoff(ctx); err != nil { + return err + } + rpc, authMechanism, reconnectErr := c.openAuthenticatedRPC(ctx) + if reconnectErr != nil { + c.recordTransportError(reconnectErr) + return reconnectErr + } + c.rpc = rpc + c.mode = TransportJSONRPC + c.reconnect++ + now := time.Now().UTC() + c.updateTransportStatus(func(status *TransportStatus) { + status.Connected = true + status.AuthMechanism = authMechanism + status.Reconnects++ + status.LastConnectedAt = &now + status.LastError = "" + }) + if err := c.rpc.call(ctx, method, params, result); err != nil { + var secondTransportErr *RPCTransportError + if errors.As(err, &secondTransportErr) { + c.closeRPCLocked() + c.recordTransportError(err) + } else { + c.recordRPCOperationError(err) + } + return err + } + c.reconnect = 0 + c.updateTransportStatus(func(status *TransportStatus) { + status.Connected = true + status.LastError = "" + }) + return nil +} + +func (c *Client) waitReconnectBackoff(ctx context.Context) error { + attempt := c.reconnect + if attempt > 4 { + attempt = 4 + } + delay := 100 * time.Millisecond * time.Duration(1< '9' + }) { + if len(field) != 2 && len(field) != 4 { + continue + } + major, err := strconv.Atoi(field) + if err != nil || major < 20 || major > 99 { + continue + } + rest := version[strings.Index(version, field)+len(field):] + for _, minorField := range strings.FieldsFunc(rest, func(r rune) bool { + return r < '0' || r > '9' + }) { + minor, err := strconv.Atoi(minorField) + if err == nil { + return major, minor, true + } + } + } + return 0, 0, false +} + +func legacyRESTAuthMechanism(config ClientConfig) string { + if strings.TrimSpace(config.APIKey) != "" { + return "rest-bearer-api-key" + } + return "rest-basic-password" +} + +func rpcErrorFromWire(method string, wire *trueNASRPCError) *RPCError { + rpcErr := &RPCError{Method: method} + if wire == nil { + return rpcErr + } + rpcErr.Code = wire.Code + rpcErr.Message = strings.TrimSpace(wire.Message) + if len(wire.Data) > 0 && string(wire.Data) != "null" { + var data map[string]any + if json.Unmarshal(wire.Data, &data) == nil { + rpcErr.Reason = strings.TrimSpace(readStringAny(data, "reason", "message")) + rpcErr.Errname = strings.TrimSpace(readStringAny(data, "errname", "name")) + } + } + return rpcErr +} + +func isMethodUnavailable(err error) bool { + var rpcErr *RPCError + if !errors.As(err, &rpcErr) { + return false + } + message := strings.ToLower(rpcErr.Message + " " + rpcErr.Reason) + return rpcErr.Code == -32601 || strings.Contains(message, "method not found") || strings.Contains(message, "not found") +} diff --git a/internal/truenas/transport_test.go b/internal/truenas/transport_test.go new file mode 100644 index 000000000..4b2f57cdb --- /dev/null +++ b/internal/truenas/transport_test.go @@ -0,0 +1,798 @@ +package truenas + +import ( + "context" + "encoding/json" + "errors" + "net/http" + "net/http/httptest" + "strings" + "sync" + "sync/atomic" + "testing" + + "github.com/gorilla/websocket" +) + +type protocolFixtureReply struct { + result any + err *trueNASRPCError + notifications []protocolFixtureNotification + close bool +} + +type protocolFixtureNotification struct { + method string + params any +} + +type protocolFixture struct { + server *httptest.Server + sessions atomic.Int32 + restRequests atomic.Int32 +} + +func newProtocolFixture( + t *testing.T, + handle func(session int, request trueNASRPCRequest) protocolFixtureReply, + handleREST http.HandlerFunc, +) *protocolFixture { + t.Helper() + + fixture := &protocolFixture{} + upgrader := websocket.Upgrader{} + fixture.server = httptest.NewTLSServer(http.HandlerFunc(func(writer http.ResponseWriter, request *http.Request) { + if request.URL.Path != "/api/current" || !websocket.IsWebSocketUpgrade(request) { + fixture.restRequests.Add(1) + if handleREST == nil { + http.NotFound(writer, request) + return + } + handleREST(writer, request) + return + } + + conn, err := upgrader.Upgrade(writer, request, nil) + if err != nil { + t.Errorf("upgrade protocol fixture websocket: %v", err) + return + } + defer func() { _ = conn.Close() }() + session := int(fixture.sessions.Add(1)) + + for { + var rpcRequest trueNASRPCRequest + if err := conn.ReadJSON(&rpcRequest); err != nil { + return + } + reply := handle(session, rpcRequest) + if reply.close { + return + } + response := trueNASRPCResponse{ + JSONRPC: "2.0", + ID: rpcRequest.ID, + Error: reply.err, + } + if reply.err == nil { + raw, err := json.Marshal(reply.result) + if err != nil { + t.Errorf("marshal protocol fixture result: %v", err) + return + } + response.Result = raw + } + if err := conn.WriteJSON(response); err != nil { + return + } + for _, notification := range reply.notifications { + params, err := json.Marshal(notification.params) + if err != nil { + t.Errorf("marshal protocol fixture notification: %v", err) + return + } + if err := conn.WriteJSON(trueNASRPCResponse{ + JSONRPC: "2.0", + Method: notification.method, + Params: params, + }); err != nil { + return + } + } + } + })) + t.Cleanup(fixture.server.Close) + return fixture +} + +func protocolFixtureClient(t *testing.T, serverURL string, config ClientConfig) *Client { + t.Helper() + config.InsecureSkipVerify = true + client := mustClientForServer(t, serverURL, config) + t.Cleanup(client.Close) + return client +} + +func TestJSONRPCTransportUsesLoginExForReadonlyAPIKey(t *testing.T) { + var sawLogin atomic.Bool + fixture := newProtocolFixture(t, func(_ int, request trueNASRPCRequest) protocolFixtureReply { + switch request.Method { + case "auth.login_ex": + params, ok := request.Params.([]any) + if !ok || len(params) != 1 { + t.Errorf("auth.login_ex params = %#v", request.Params) + return protocolFixtureReply{close: true} + } + login, ok := params[0].(map[string]any) + if !ok { + t.Errorf("auth.login_ex login payload = %#v", params[0]) + return protocolFixtureReply{close: true} + } + if login["mechanism"] != "API_KEY_PLAIN" || login["username"] != "pulse-readonly" || login["api_key"] != "readonly-key" { + t.Errorf("auth.login_ex login payload = %#v", login) + } + sawLogin.Store(true) + return protocolFixtureReply{result: map[string]any{"response_type": "SUCCESS", "user_info": nil}} + case "system.info": + return protocolFixtureReply{result: map[string]any{ + "hostname": "nas-readonly", + "version": "TrueNAS-SCALE-25.10.4", + "system_serial": "READONLY-1", + "uptime_seconds": 42, + "physical_cores": 8, + "physmem": 16 * 1024 * 1024 * 1024, + }} + default: + t.Errorf("unexpected rpc method %q", request.Method) + return protocolFixtureReply{close: true} + } + }, nil) + + client := protocolFixtureClient(t, fixture.server.URL, ClientConfig{ + APIKey: "readonly-key", + Username: "pulse-readonly", + }) + system, err := client.GetSystemInfo(context.Background()) + if err != nil { + t.Fatalf("GetSystemInfo() error = %v", err) + } + if !sawLogin.Load() || system.Hostname != "nas-readonly" || system.Version != "TrueNAS-SCALE-25.10.4" { + t.Fatalf("unexpected readonly system result: %+v", system) + } + if fixture.restRequests.Load() != 0 { + t.Fatalf("modern JSON-RPC session made %d REST requests", fixture.restRequests.Load()) + } + status := client.TransportStatus() + if status.Mode != TransportJSONRPC || !status.Connected || status.AuthMechanism != "api-key-plain" || !status.TLS { + t.Fatalf("unexpected transport status: %+v", status) + } +} + +func TestJSONRPCLegacyAPIKeyLoginRemovalRequestsOwnerUsername(t *testing.T) { + fixture := newProtocolFixture(t, func(_ int, request trueNASRPCRequest) protocolFixtureReply { + if request.Method != "auth.login_with_api_key" { + t.Errorf("unexpected rpc method %q", request.Method) + return protocolFixtureReply{close: true} + } + return protocolFixtureReply{err: &trueNASRPCError{ + Code: -32601, + Message: "Method not found", + }} + }, nil) + + client := protocolFixtureClient(t, fixture.server.URL, ClientConfig{APIKey: "upgrade-key"}) + _, err := client.GetSystemInfo(context.Background()) + if err == nil || !strings.Contains(err.Error(), "requires the API key owner username") { + t.Fatalf("GetSystemInfo() error = %v, want upgrade guidance", err) + } + if fixture.restRequests.Load() != 0 { + t.Fatalf("removed login method made %d REST downgrade requests", fixture.restRequests.Load()) + } +} + +func TestJSONRPCPermissionFailureDoesNotDowngradeToREST(t *testing.T) { + fixture := newProtocolFixture(t, func(_ int, request trueNASRPCRequest) protocolFixtureReply { + switch request.Method { + case "auth.login_ex": + return protocolFixtureReply{result: map[string]any{"response_type": "SUCCESS", "user_info": nil}} + case "system.info": + data, _ := json.Marshal(map[string]any{ + "reason": "Not authorized", + "errname": "EACCES", + }) + return protocolFixtureReply{err: &trueNASRPCError{ + Code: -32001, + Message: "Method call error", + Data: data, + }} + default: + return protocolFixtureReply{close: true} + } + }, func(writer http.ResponseWriter, _ *http.Request) { + t.Error("REST must not be called after an authoritative JSON-RPC permission error") + http.Error(writer, "unexpected", http.StatusInternalServerError) + }) + + client := protocolFixtureClient(t, fixture.server.URL, ClientConfig{ + APIKey: "readonly-key", + Username: "pulse-readonly", + }) + _, err := client.GetSystemInfo(context.Background()) + if err == nil || !strings.Contains(err.Error(), "EACCES") { + t.Fatalf("GetSystemInfo() error = %v, want structured permission failure", err) + } + if fixture.restRequests.Load() != 0 { + t.Fatalf("permission failure made %d REST requests", fixture.restRequests.Load()) + } + status := client.TransportStatus() + if status.Mode != TransportJSONRPC || !status.Connected { + t.Fatalf("permission failure changed transport authority: %+v", status) + } +} + +func TestJSONRPCReadonlyActionDenialIsAuthoritativeAndNotRetried(t *testing.T) { + var actionCalls atomic.Int32 + fixture := newProtocolFixture(t, func(_ int, request trueNASRPCRequest) protocolFixtureReply { + switch request.Method { + case "auth.login_ex": + return protocolFixtureReply{result: map[string]any{"response_type": "SUCCESS", "user_info": nil}} + case "app.start": + actionCalls.Add(1) + data, _ := json.Marshal(map[string]any{ + "reason": "Not authorized", + "errname": "EACCES", + }) + return protocolFixtureReply{err: &trueNASRPCError{ + Code: -32001, + Message: "Method call error", + Data: data, + }} + default: + return protocolFixtureReply{close: true} + } + }, func(writer http.ResponseWriter, _ *http.Request) { + t.Error("REST must not be called after an authoritative action denial") + http.Error(writer, "unexpected", http.StatusInternalServerError) + }) + + client := protocolFixtureClient(t, fixture.server.URL, ClientConfig{ + APIKey: "readonly-key", + Username: "pulse-readonly", + }) + err := client.StartApp(context.Background(), "nextcloud") + var rpcErr *RPCError + if !errors.As(err, &rpcErr) || rpcErr.Errname != "EACCES" || rpcErr.Method != "app.start" { + t.Fatalf("StartApp() error = %v, want structured app.start permission denial", err) + } + if actionCalls.Load() != 1 || fixture.sessions.Load() != 1 || fixture.restRequests.Load() != 0 { + t.Fatalf("actionCalls=%d sessions=%d restRequests=%d, denial was retried or downgraded", + actionCalls.Load(), fixture.sessions.Load(), fixture.restRequests.Load()) + } +} + +func TestTransportNegotiationAllowsOnlyRecognizedLegacyREST(t *testing.T) { + t.Run("legacy scale", func(t *testing.T) { + server := httptest.NewTLSServer(http.HandlerFunc(func(writer http.ResponseWriter, request *http.Request) { + switch request.URL.Path { + case "/api/current": + http.NotFound(writer, request) + case "/api/v2.0/system/info": + _, _ = writer.Write([]byte(`{"hostname":"legacy","version":"TrueNAS-SCALE-24.10.2","uptime_seconds":1}`)) + case "/api/v2.0/pool": + _, _ = writer.Write([]byte(`[{"id":1,"name":"tank","status":"ONLINE"}]`)) + default: + http.NotFound(writer, request) + } + })) + t.Cleanup(server.Close) + + client := protocolFixtureClient(t, server.URL, ClientConfig{APIKey: "legacy-key"}) + pools, err := client.GetPools(context.Background()) + if err != nil { + t.Fatalf("GetPools() legacy error = %v", err) + } + if len(pools) != 1 || pools[0].Name != "tank" { + t.Fatalf("unexpected legacy pools: %+v", pools) + } + if status := client.TransportStatus(); status.Mode != TransportLegacyREST || status.ApplianceVersion != "TrueNAS-SCALE-24.10.2" { + t.Fatalf("unexpected legacy status: %+v", status) + } + }) + + t.Run("legacy core", func(t *testing.T) { + server := httptest.NewTLSServer(http.HandlerFunc(func(writer http.ResponseWriter, request *http.Request) { + switch request.URL.Path { + case "/api/current": + http.NotFound(writer, request) + case "/api/v2.0/system/info": + _, _ = writer.Write([]byte(`{"hostname":"core","version":"TrueNAS-13.0-U6.4","uptime_seconds":1}`)) + case "/api/v2.0/pool": + _, _ = writer.Write([]byte(`[{"id":1,"name":"core-tank","status":"ONLINE"}]`)) + default: + http.NotFound(writer, request) + } + })) + t.Cleanup(server.Close) + + client := protocolFixtureClient(t, server.URL, ClientConfig{APIKey: "legacy-core-key"}) + pools, err := client.GetPools(context.Background()) + if err != nil { + t.Fatalf("GetPools() CORE error = %v", err) + } + if len(pools) != 1 || pools[0].Name != "core-tank" { + t.Fatalf("unexpected CORE pools: %+v", pools) + } + if status := client.TransportStatus(); status.Mode != TransportLegacyREST || + status.ApplianceVersion != "TrueNAS-13.0-U6.4" { + t.Fatalf("unexpected CORE status: %+v", status) + } + }) + + t.Run("current scale fail closed", func(t *testing.T) { + var poolRESTCalls atomic.Int32 + server := httptest.NewTLSServer(http.HandlerFunc(func(writer http.ResponseWriter, request *http.Request) { + switch request.URL.Path { + case "/api/current": + http.NotFound(writer, request) + case "/api/v2.0/system/info": + _, _ = writer.Write([]byte(`{"hostname":"current","version":"TrueNAS-SCALE-25.10.4","uptime_seconds":1}`)) + case "/api/v2.0/pool": + poolRESTCalls.Add(1) + _, _ = writer.Write([]byte(`[]`)) + default: + http.NotFound(writer, request) + } + })) + t.Cleanup(server.Close) + + client := protocolFixtureClient(t, server.URL, ClientConfig{APIKey: "current-key"}) + _, err := client.GetPools(context.Background()) + if err == nil || !strings.Contains(err.Error(), "refusing REST downgrade") { + t.Fatalf("GetPools() error = %v, want fail-closed downgrade error", err) + } + if poolRESTCalls.Load() != 0 { + t.Fatalf("current SCALE used pool REST %d times", poolRESTCalls.Load()) + } + }) +} + +func TestJSONRPCCredentialsRequireTLSWithoutSendingAuthentication(t *testing.T) { + var rpcRequests atomic.Int32 + var restRequests atomic.Int32 + upgrader := websocket.Upgrader{} + server := httptest.NewServer(http.HandlerFunc(func(writer http.ResponseWriter, request *http.Request) { + if request.URL.Path != "/api/current" || !websocket.IsWebSocketUpgrade(request) { + restRequests.Add(1) + http.NotFound(writer, request) + return + } + conn, err := upgrader.Upgrade(writer, request, nil) + if err != nil { + t.Errorf("upgrade plaintext websocket: %v", err) + return + } + defer func() { _ = conn.Close() }() + var rpcRequest trueNASRPCRequest + if conn.ReadJSON(&rpcRequest) == nil { + rpcRequests.Add(1) + } + })) + t.Cleanup(server.Close) + + client := mustClientForServer(t, server.URL, ClientConfig{ + APIKey: "must-not-cross-plaintext", + Username: "pulse-readonly", + }) + client.allowInsecureRPC = false + t.Cleanup(client.Close) + _, err := client.GetPools(context.Background()) + if err == nil || !strings.Contains(err.Error(), "credentials require TLS") { + t.Fatalf("GetPools() error = %v, want TLS requirement", err) + } + if rpcRequests.Load() != 0 || restRequests.Load() != 0 { + t.Fatalf("plaintext negotiation sent rpc=%d REST=%d requests after websocket upgrade", + rpcRequests.Load(), restRequests.Load()) + } +} + +func TestTransportNegotiationIsIsolatedAcrossAppliances(t *testing.T) { + modern := newProtocolFixture(t, func(_ int, request trueNASRPCRequest) protocolFixtureReply { + switch request.Method { + case "auth.login_ex": + return protocolFixtureReply{result: map[string]any{"response_type": "SUCCESS", "user_info": nil}} + case "pool.query": + return protocolFixtureReply{result: []map[string]any{{"id": 1, "name": "modern", "status": "ONLINE"}}} + default: + return protocolFixtureReply{close: true} + } + }, nil) + legacyServer := httptest.NewTLSServer(http.HandlerFunc(func(writer http.ResponseWriter, request *http.Request) { + switch request.URL.Path { + case "/api/current": + http.NotFound(writer, request) + case "/api/v2.0/system/info": + _, _ = writer.Write([]byte(`{"hostname":"legacy","version":"TrueNAS-SCALE-24.10.2"}`)) + case "/api/v2.0/pool": + _, _ = writer.Write([]byte(`[{"id":2,"name":"legacy","status":"ONLINE"}]`)) + default: + http.NotFound(writer, request) + } + })) + t.Cleanup(legacyServer.Close) + + modernClient := protocolFixtureClient(t, modern.server.URL, ClientConfig{ + APIKey: "modern-key", + Username: "pulse-readonly", + }) + legacyClient := protocolFixtureClient(t, legacyServer.URL, ClientConfig{APIKey: "legacy-key"}) + + modernPools, modernErr := modernClient.GetPools(context.Background()) + legacyPools, legacyErr := legacyClient.GetPools(context.Background()) + if modernErr != nil || legacyErr != nil { + t.Fatalf("multi-appliance polls failed: modern=%v legacy=%v", modernErr, legacyErr) + } + if len(modernPools) != 1 || modernPools[0].Name != "modern" || + len(legacyPools) != 1 || legacyPools[0].Name != "legacy" { + t.Fatalf("unexpected isolated inventories: modern=%+v legacy=%+v", modernPools, legacyPools) + } + modernStatus := modernClient.TransportStatus() + legacyStatus := legacyClient.TransportStatus() + if modernStatus.Mode != TransportJSONRPC || modernStatus.Endpoint == legacyStatus.Endpoint || + legacyStatus.Mode != TransportLegacyREST { + t.Fatalf("transport decisions leaked across clients: modern=%+v legacy=%+v", modernStatus, legacyStatus) + } +} + +func TestNegotiatedJSONRPCTransportNeverRenegotiatesToREST(t *testing.T) { + var websocketAttempts atomic.Int32 + var restRequests atomic.Int32 + upgrader := websocket.Upgrader{} + server := httptest.NewTLSServer(http.HandlerFunc(func(writer http.ResponseWriter, request *http.Request) { + switch request.URL.Path { + case "/api/current": + if websocketAttempts.Add(1) > 1 { + http.NotFound(writer, request) + return + } + conn, err := upgrader.Upgrade(writer, request, nil) + if err != nil { + t.Errorf("upgrade initial websocket: %v", err) + return + } + defer func() { _ = conn.Close() }() + for { + var rpcRequest trueNASRPCRequest + if err := conn.ReadJSON(&rpcRequest); err != nil { + return + } + var result any + switch rpcRequest.Method { + case "auth.login_with_api_key": + result = true + case "pool.query": + result = []map[string]any{{"id": 1, "name": "modern", "status": "ONLINE"}} + default: + t.Errorf("unexpected rpc method %q", rpcRequest.Method) + return + } + raw, err := json.Marshal(result) + if err != nil { + t.Errorf("marshal rpc result: %v", err) + return + } + if err := conn.WriteJSON(trueNASRPCResponse{ + JSONRPC: "2.0", + ID: rpcRequest.ID, + Result: raw, + }); err != nil { + return + } + } + case "/api/v2.0/system/info", "/api/v2.0/pool": + restRequests.Add(1) + _, _ = writer.Write([]byte(`{"version":"TrueNAS-SCALE-24.10.2"}`)) + default: + http.NotFound(writer, request) + } + })) + t.Cleanup(server.Close) + + client := protocolFixtureClient(t, server.URL, ClientConfig{APIKey: "modern-key"}) + if _, err := client.GetPools(context.Background()); err != nil { + t.Fatalf("initial GetPools() error = %v", err) + } + + client.rpcMu.Lock() + client.closeRPCLocked() + client.rpcMu.Unlock() + + _, err := client.GetPools(context.Background()) + if err == nil || !strings.Contains(err.Error(), "transport is immutable") { + t.Fatalf("GetPools() reconnect error = %v, want immutable transport failure", err) + } + if restRequests.Load() != 0 { + t.Fatalf("negotiated JSON-RPC transport made %d REST downgrade requests", restRequests.Load()) + } + status := client.TransportStatus() + if status.Mode != TransportJSONRPC || status.Connected { + t.Fatalf("immutable reconnect changed transport authority: %+v", status) + } +} + +func TestJSONRPCReadReconnectsButActionIsNeverReplayed(t *testing.T) { + t.Run("read reconnect", func(t *testing.T) { + var poolCalls atomic.Int32 + fixture := newProtocolFixture(t, func(session int, request trueNASRPCRequest) protocolFixtureReply { + if request.Method == "auth.login_with_api_key" { + return protocolFixtureReply{result: true} + } + if request.Method != "pool.query" { + return protocolFixtureReply{close: true} + } + call := poolCalls.Add(1) + if session == 1 && call == 2 { + return protocolFixtureReply{close: true} + } + return protocolFixtureReply{result: []map[string]any{{"id": 1, "name": "tank", "status": "ONLINE"}}} + }, nil) + + client := protocolFixtureClient(t, fixture.server.URL, ClientConfig{APIKey: "legacy-auth-key"}) + if _, err := client.GetPools(context.Background()); err != nil { + t.Fatalf("first GetPools() error = %v", err) + } + if _, err := client.GetPools(context.Background()); err != nil { + t.Fatalf("reconnected GetPools() error = %v", err) + } + if fixture.sessions.Load() != 2 || poolCalls.Load() != 3 { + t.Fatalf("sessions=%d poolCalls=%d, want one replay on a new session", fixture.sessions.Load(), poolCalls.Load()) + } + if status := client.TransportStatus(); status.Reconnects != 1 || status.Mode != TransportJSONRPC { + t.Fatalf("unexpected reconnect status: %+v", status) + } + }) + + t.Run("action outcome unknown", func(t *testing.T) { + var actionCalls atomic.Int32 + fixture := newProtocolFixture(t, func(_ int, request trueNASRPCRequest) protocolFixtureReply { + if request.Method == "auth.login_with_api_key" { + return protocolFixtureReply{result: true} + } + if request.Method == "app.start" { + actionCalls.Add(1) + return protocolFixtureReply{close: true} + } + return protocolFixtureReply{close: true} + }, nil) + + client := protocolFixtureClient(t, fixture.server.URL, ClientConfig{APIKey: "app-write-key"}) + err := client.StartApp(context.Background(), "nextcloud") + if err == nil || !strings.Contains(err.Error(), "outcome is unknown") { + t.Fatalf("StartApp() error = %v, want non-replay outcome", err) + } + if actionCalls.Load() != 1 || fixture.sessions.Load() != 1 { + t.Fatalf("actionCalls=%d sessions=%d, action was replayed", actionCalls.Load(), fixture.sessions.Load()) + } + }) +} + +func TestJSONRPCQueryReturnsCompleteUnpaginatedInventory(t *testing.T) { + poolFixture := make([]map[string]any, 0, 225) + for i := 0; i < 225; i++ { + poolFixture = append(poolFixture, map[string]any{"id": i + 1, "name": "pool-" + strings.Repeat("x", i%4), "status": "ONLINE"}) + } + fixture := newProtocolFixture(t, func(_ int, request trueNASRPCRequest) protocolFixtureReply { + if request.Method == "auth.login_with_api_key" { + return protocolFixtureReply{result: true} + } + if request.Method != "pool.query" { + return protocolFixtureReply{close: true} + } + params, ok := request.Params.([]any) + if !ok || len(params) != 2 { + t.Errorf("pool.query params = %#v", request.Params) + } else if options, ok := params[1].(map[string]any); !ok || len(options) != 0 { + t.Errorf("pool.query options must request the authoritative complete set, got %#v", params[1]) + } + return protocolFixtureReply{result: poolFixture} + }, nil) + + client := protocolFixtureClient(t, fixture.server.URL, ClientConfig{APIKey: "readonly-key"}) + pools, err := client.GetPools(context.Background()) + if err != nil { + t.Fatalf("GetPools() error = %v", err) + } + if len(pools) != len(poolFixture) { + t.Fatalf("GetPools() returned %d pools, want %d", len(pools), len(poolFixture)) + } +} + +func TestJSONRPCFetchSnapshotKeepsAllModernInventoryOnOneTransport(t *testing.T) { + methodCalls := make(map[string]int) + var methodCallsMu sync.Mutex + fixture := newProtocolFixture(t, func(_ int, request trueNASRPCRequest) protocolFixtureReply { + methodCallsMu.Lock() + methodCalls[request.Method]++ + methodCallsMu.Unlock() + + switch request.Method { + case "auth.login_ex": + return protocolFixtureReply{result: map[string]any{"response_type": "SUCCESS", "user_info": nil}} + case "system.info": + return protocolFixtureReply{result: map[string]any{ + "hostname": "inventory-nas", + "version": "TrueNAS-SCALE-25.10.4", + "system_serial": "INVENTORY-1", + "uptime_seconds": 60, + "physical_cores": 4, + "physmem": 8 * 1024 * 1024 * 1024, + }} + case "reporting.get_data": + return protocolFixtureReply{result: []any{}} + case "core.subscribe": + params, _ := request.Params.([]any) + event, _ := params[0].(string) + if strings.HasPrefix(event, "reporting.realtime:") { + return protocolFixtureReply{ + result: "sub-realtime", + notifications: []protocolFixtureNotification{{ + method: "collection_update", + params: map[string]any{ + "collection": event, + "fields": map[string]any{"cpu": map[string]any{"usage": 12}}, + }, + }}, + } + } + if strings.HasPrefix(event, "app.stats:") { + return protocolFixtureReply{ + result: "sub-app-stats", + notifications: []protocolFixtureNotification{{ + method: "collection_update", + params: map[string]any{ + "collection": event, + "fields": []any{}, + }, + }}, + } + } + t.Errorf("unexpected subscription event %q", event) + return protocolFixtureReply{close: true} + case "core.unsubscribe": + return protocolFixtureReply{result: nil} + case "pool.query", "pool.dataset.query", "disk.query", "alert.list", + "service.query", "app.query", "vm.query", "sharing.smb.query", + "sharing.nfs.query", "zfs.resource.snapshot.query", "replication.query": + return protocolFixtureReply{result: []any{}} + case "boot.get_state": + return protocolFixtureReply{result: map[string]any{ + "id": "boot-pool", "name": "boot-pool", "status": "ONLINE", + }} + default: + t.Errorf("unexpected rpc method %q", request.Method) + return protocolFixtureReply{close: true} + } + }, nil) + + client := protocolFixtureClient(t, fixture.server.URL, ClientConfig{ + APIKey: "readonly-key", + Username: "pulse-readonly", + }) + snapshot, err := client.FetchSnapshot(context.Background()) + if err != nil { + t.Fatalf("FetchSnapshot() error = %v", err) + } + if snapshot.System.Hostname != "inventory-nas" || snapshot.System.CPUPercent != 12 { + t.Fatalf("unexpected snapshot system: %+v", snapshot.System) + } + + for _, method := range []string{ + "system.info", + "pool.query", + "boot.get_state", + "pool.dataset.query", + "disk.query", + "alert.list", + "service.query", + "app.query", + "vm.query", + "sharing.smb.query", + "sharing.nfs.query", + "zfs.resource.snapshot.query", + "replication.query", + } { + methodCallsMu.Lock() + count := methodCalls[method] + methodCallsMu.Unlock() + if count != 1 { + t.Errorf("%s calls = %d, want exactly one", method, count) + } + } + methodCallsMu.Lock() + subscribeCalls := methodCalls["core.subscribe"] + unsubscribeCalls := methodCalls["core.unsubscribe"] + methodCallsMu.Unlock() + if subscribeCalls != 2 || unsubscribeCalls != 2 { + t.Fatalf("subscription lifecycle calls: subscribe=%d unsubscribe=%d, want 2/2", subscribeCalls, unsubscribeCalls) + } + if fixture.sessions.Load() != 1 || fixture.restRequests.Load() != 0 { + t.Fatalf("snapshot sessions=%d REST=%d, want one modern session and no REST", + fixture.sessions.Load(), fixture.restRequests.Load()) + } +} + +func TestJSONRPCMalformedSubscriptionEventDiscardsSession(t *testing.T) { + fixture := newProtocolFixture(t, func(session int, request trueNASRPCRequest) protocolFixtureReply { + switch request.Method { + case "auth.login_with_api_key": + return protocolFixtureReply{result: true} + case "core.subscribe": + return protocolFixtureReply{ + result: "sub-malformed", + notifications: []protocolFixtureNotification{{ + method: "collection_update", + params: map[string]any{ + "collection": "app.stats:{\"interval\":2}", + "fields": "not-an-array", + }, + }}, + } + case "pool.query": + if session != 2 { + t.Errorf("pool.query reused malformed subscription session %d", session) + } + return protocolFixtureReply{result: []map[string]any{{"id": 1, "name": "tank", "status": "ONLINE"}}} + default: + t.Errorf("unexpected rpc method %q", request.Method) + return protocolFixtureReply{close: true} + } + }, nil) + + client := protocolFixtureClient(t, fixture.server.URL, ClientConfig{APIKey: "readonly-key"}) + if _, err := client.GetAppStats(context.Background()); err == nil || + !strings.Contains(err.Error(), "decode truenas app.stats notification") { + t.Fatalf("GetAppStats() error = %v, want malformed event error", err) + } + pools, err := client.GetPools(context.Background()) + if err != nil { + t.Fatalf("GetPools() after discarded stream error = %v", err) + } + if len(pools) != 1 || pools[0].Name != "tank" || fixture.sessions.Load() != 2 { + t.Fatalf("unexpected post-discard state: pools=%+v sessions=%d", pools, fixture.sessions.Load()) + } + if status := client.TransportStatus(); status.Reconnects != 1 { + t.Fatalf("discarded stream reconnect was not observable: %+v", status) + } +} + +func TestTransportDiagnosticsAreConnectionLocalAndSecretFree(t *testing.T) { + fixture := newProtocolFixture(t, func(_ int, request trueNASRPCRequest) protocolFixtureReply { + if request.Method == "auth.login_with_api_key" { + data, _ := json.Marshal(map[string]any{ + "reason": "invalid credential never-expose-this-key", + "errname": "EINVAL", + }) + return protocolFixtureReply{err: &trueNASRPCError{ + Code: -32001, + Message: "Method call error", + Data: data, + }} + } + return protocolFixtureReply{close: true} + }, nil) + client := protocolFixtureClient(t, fixture.server.URL, ClientConfig{APIKey: "never-expose-this-key"}) + + if _, err := client.GetPools(context.Background()); err == nil { + t.Fatal("expected authentication failure") + } + status := client.TransportStatus() + encoded, err := json.Marshal(status) + if err != nil { + t.Fatalf("Marshal(status) error = %v", err) + } + if strings.Contains(string(encoded), "never-expose-this-key") { + t.Fatalf("transport status exposed API key: %s", encoded) + } + if status.Mode != TransportUnknown { + t.Fatalf("failed authentication selected a transport: %+v", status) + } +}