diff --git a/ZB.MOM.WW.OtOpcUa.slnx b/ZB.MOM.WW.OtOpcUa.slnx index 4e577d11..c5a50698 100644 --- a/ZB.MOM.WW.OtOpcUa.slnx +++ b/ZB.MOM.WW.OtOpcUa.slnx @@ -49,6 +49,7 @@ + diff --git a/docs/drivers/Mqtt.md b/docs/drivers/Mqtt.md index 25c54975..6d1b921e 100644 --- a/docs/drivers/Mqtt.md +++ b/docs/drivers/Mqtt.md @@ -24,20 +24,24 @@ for the design. |---|---| | `Driver.Mqtt.Contracts` | Config records, enums, `MqttTagDefinition` + the pure `MqttTagDefinitionFactory`, and the one shared `MqttJson.Options`. **Deliberately transport-free** — no MQTTnet reference | | `Driver.Mqtt` | Runtime: `MqttDriver`, `MqttConnection` (connect + reconnect supervisor), `MqttSubscriptionManager`, `LastValueCache`, factory registration, `MqttDriverProbe` | +| `Driver.Mqtt.Transport` | `MqttClientOptionsFactory` — TLS posture, PEM CA-pin chain validator, credentials, protocol mapping. Connection **policy** only; nothing here connects | | `Driver.Mqtt.Browser` | `MqttDriverBrowser` + `MqttBrowseSession` — the AdminUI address picker's `#`-observation session | -> `.Browser` references `.Driver` (not just `.Contracts`) — a **documented, deliberate exception** -> so browse can reuse `MqttConnection.BuildClientOptions` (TLS / CA-pin / credentials / protocol -> mapping). Its cost is pulling `Core` into the AdminUI graph. The csproj carries a boxed warning; -> do not copy the pattern into a new driver's browser. The clean fix — extract the pure -> options builder down into `.Contracts` — is tracked as -> [#512](https://gitea.dohertylan.com/dohertj2/lmxopcua/issues/512). +> **`.Browser` depends on the shared leaves only** — `.Contracts` + `.Transport`, never the runtime +> `.Driver`. It used to reference `.Driver` under a documented exception; #512 removed it via two +> extractions (the client-options builder to `.Transport`, and the Sparkplug protocol layer — +> `SparkplugTopic`, `SparkplugCodec`, `RebirthRequester` — to `.Contracts`, beside the Tahu wire +> schema they encode). `MqttBrowserLayeringTests` guards the graph in both directions. +> +> There must stay **exactly one copy** of the TLS/CA-pin logic. A second copy would not break the +> build — it would silently diverge, browse quietly not honouring a pinned CA while the driver does. ## Capability Surface ```csharp public sealed class MqttDriver - : IDriver, ITagDiscovery, ISubscribable, IReadable, IHostConnectivityProbe, IRediscoverable, IAsyncDisposable, IDisposable + : IDriver, ITagDiscovery, ISubscribable, IReadable, IWritable, IHostConnectivityProbe, IRediscoverable, + IAsyncDisposable, IDisposable ``` | Capability | Entry point | Notes | @@ -46,12 +50,15 @@ public sealed class MqttDriver | `ITagDiscovery` | `DiscoverAsync` | Replays **only the authored `rawTags`** into an `Mqtt` folder. `SupportsOnlineDiscovery => false`. `RediscoverPolicy` is `Once` in Plain and **`UntilStable` in Sparkplug** (half of what a Sparkplug tag needs — its datatype — is only knowable from a birth) | | `ISubscribable` | `SubscribeAsync` / `OnDataChange` | One SUBSCRIBE per distinct authored topic. **`publishingInterval` is ignored** — MQTT is push-based; this driver neither polls nor throttles | | `IReadable` | `ReadAsync` | Serves the last received/retained value from `LastValueCache`; never throws. A never-seen tag reads `BadWaitingForInitialData` | +| `IWritable` | `WriteAsync` | Publishes to the tag's **`writeTopic`** (Plain) or as an **NCMD/DCMD** command (Sparkplug). One status per request; a tag with no write destination is refused `BadNotWritable`. See [Write-through](#write-through) | | `IHostConnectivityProbe` | `GetHostStatuses` | 1-second poll of `MqttConnection.State`; `HostName` is `"{host}:{port}"` | | `IRediscoverable` | `OnRediscoveryNeeded` | Nothing raises it in Plain mode. **Sparkplug raises it** when a birth for an *authored* scope carries a metric-name set different from the last one seen for that scope — both conditions are anti-storm, see [Rediscovery](#rediscovery-sparkplug) | -**Not implemented:** `IWritable`, `IPerCallHostResolver`, `IAlarmSource`, `IHistoryProvider`. -Materialized nodes are `SecurityClassification.ViewOnly`, `IsAlarm: false`, `IsHistorized: false` — -*"MQTT ingest is one-way … ViewOnly rather than an Operate node whose writes would silently vanish."* +**Not implemented:** `IPerCallHostResolver`, `IAlarmSource`, `IHistoryProvider`. +Materialized nodes are `IsAlarm: false`, `IsHistorized: false`. A node is +`SecurityClassification.Operate` when its tag declares a write destination and +`ViewOnly` when it does not — the browse-time answer comes from the same encoder the write path uses, +so a node can never advertise writability it would then refuse. Driver-type string: **`Mqtt`** (`DriverTypeNames.Mqtt`). @@ -274,6 +281,81 @@ nothing forever, so `OnReconnectedAsync` **throws** when zero filters are grante down rather than swallowing it. A SUBACK failure degrades just the affected RawPaths to `BadCommunicationError`. +## Write-through + +`IWritable` landed with [#508](https://gitea.dohertylan.com/dohertj2/lmxopcua/issues/508). Before it, +every MQTT node materialized read-only regardless of what the operator authored. + +| Mode | Destination | Payload | +|---|---|---| +| **Plain** | the tag's `writeTopic` | the value as UTF-8 text (`Raw`/`Scalar`) or a bare JSON literal (`Json`) | +| **Sparkplug B** | `spBv1.0/{group}/DCMD/{node}/{device}` for a device metric, `spBv1.0/{group}/NCMD/{node}` for an edge-node metric | a Sparkplug `Payload` carrying the metric **by name**, with an explicit `Datatype` and no `seq` | + +Both publish at **QoS 0, `retain: false`**. Retain is the one that matters and is deliberately not +configurable: a retained command would be replayed by the broker to the device on *every* future +subscribe — including the device's own reconnect — so one stale write would be re-applied forever. + +### `writeTopic` is separate from `topic`, and its absence is meaningful + +A tag commonly reads `…/state` and writes `…/set`, so the destination is its own authored field. A +Plain tag with **no** `writeTopic` is refused `BadNotWritable` — it is not published anywhere, and the +node is discovered `ViewOnly`. + +> This is the whole point of the design. The one outcome worse than read-only nodes is a node that +> *looks* writable, accepts a write, returns `Good`, and publishes nothing. Publishing back onto the +> tag's own read topic would be worse still: it would echo straight back and read as a confirmed +> device value. + +```jsonc +{ + "topic": "factory/oven/temp/state", + "writeTopic": "factory/oven/temp/set", // omit for a read-only tag + "payloadFormat": "Scalar", + "dataType": "Float64" +} +``` + +Sparkplug needs no extra key — a command is addressed by the binding tuple the tag already carries. + +### `Good` means published, not applied + +MQTT has no application-level write acknowledgement, so a successful PUBLISH is the strongest truth +available at this seam — the same optimistic-Good posture Galaxy's fire-and-forget write takes. Device +truth arrives afterwards as the echoed value on the read topic and corrects the node through ordinary +ingest (write-outcome self-correction, [#5](https://gitea.dohertylan.com/dohertj2/lmxopcua/issues/5)). + +**Inherited consequence:** this driver can never surface a failed *device* write, exactly as Galaxy +cannot. A publish that itself fails **is** reported `Bad` — the optimism covers what the device did +with the command, never whether it was sent. + +### Status codes + +| Situation | Status | +|---|---| +| Published | `Good` | +| Reference is not an authored tag | `BadNodeIdUnknown` | +| No write destination, or a data type with no command encoding | `BadNotWritable` | +| Value not representable in the tag's declared type | `BadTypeMismatch` | +| Publish failed, or no live broker connection | `BadCommunicationError` | + +One status per request, in request order; one bad request never fails the rest of the batch. + +### `WriteIdempotent` stays default-off + +MQTT gives no write ack and a Sparkplug command is not necessarily idempotent (a pulse, a recipe step, +a counter increment), so the resilience pipeline must not replay one. Operators opt in per tag. + +### Two things to know + +- **`jsonPath` is not applied on write.** It is a read-side extraction from the *read* topic's + document; the write topic is a different topic whose payload shape nothing in the tag config + describes. A `Json` write emits a bare JSON literal. Reconstructing a wrapped document from an + extraction path would be guesswork. +- **NCMD is a shared channel.** The driver publishes its own rebirth requests + (`Node Control/Rebirth`) on `spBv1.0/{group}/NCMD/{node}` — the same topic an edge-node metric write + uses. Anything consuming that topic must match on the metric name. The live gate caught exactly this: + a rebirth NCMD arrived before the write command. DCMD has no such contention. + ## Browse `MqttDriverBrowser` opens a **read-only observation session**: it subscribes at QoS 0 to `#` (or @@ -384,14 +466,13 @@ identical words. See [TestConnectProbes.md](TestConnectProbes.md). Every row is tracked. None is a silent failure — each either refuses in words or logs. +**Closed 2026-07-31:** write-through ([#508](https://gitea.dohertylan.com/dohertj2/lmxopcua/issues/508), live-gated), the slashed-metric browse-commit +([#509](https://gitea.dohertylan.com/dohertj2/lmxopcua/issues/509)), the stale rebirth affordance ([#510](https://gitea.dohertylan.com/dohertj2/lmxopcua/issues/510)), the `.Browser` layering exception +([#512](https://gitea.dohertylan.com/dohertj2/lmxopcua/issues/512)), the stray Sparkplug `payloadFormat` ([#513](https://gitea.dohertylan.com/dohertj2/lmxopcua/issues/513)) and rebirth on a +not-yet-birthed plant ([#514](https://gitea.dohertylan.com/dohertj2/lmxopcua/issues/514)). + | Gap | Issue | Effect | Notes | |---|:---:|---|---| -| **Write-through (`IWritable`)** | [#508](https://gitea.dohertylan.com/dohertj2/lmxopcua/issues/508) | Every MQTT node is read-only | Deferred to P3 — NCMD/DCMD + plain publish, optimistic-Good + self-correct on echo | -| **Rediscovery is inert** | [#507](https://gitea.dohertylan.com/dohertj2/lmxopcua/issues/507) | A DBIRTH introducing a metric does not change the OPC UA tree; redeploy to pick it up | v3 **platform** gap, not MQTT-specific — nothing subscribes to `OnRediscoveryNeeded` and `DriverHostActor.HandleDiscoveredNodes` hard-returns. Also affects Galaxy / TwinCAT / AbCip. Driver-side decision is log-observable | -| **Rebirth needs a non-empty tree** | [#514](https://gitea.dohertylan.com/dohertj2/lmxopcua/issues/514) | `Request rebirth…` cannot be armed on a plant that has not birthed since the window opened | See [Refresh](#refresh). Session side already supports a bare `{group}/{edgeNode}` scope | -| **`_canRebirth` is captured at browse-open** | [#510](https://gitea.dohertylan.com/dohertj2/lmxopcua/issues/510) | A session reaped for idleness (2 min) still draws the button, and using it errors | The error is shown, not swallowed; reopen the browser | +| **Rediscovery is advisory, not a rebuild** | [#507](https://gitea.dohertylan.com/dohertj2/lmxopcua/issues/507) (closed, superseded) | A DBIRTH introducing a metric does not change the served OPC UA tree; an operator must re-browse the device and commit | **No longer "inert".** `DriverInstanceActor` now subscribes to `OnRediscoveryNeeded` and the raise rides the driver-health snapshot to the AdminUI `/hosts` page as a **re-browse chip**. It deliberately does not inject nodes: v3 authors raw tags through the `/raw` browse-commit flow, and grafting nodes at runtime would materialise ones nobody approved and no deployment artifact records. #507's injection path was deleted rather than revived | | **Primary-host STATE publishing** | [#511](https://gitea.dohertylan.com/dohertj2/lmxopcua/issues/511) | `ActAsPrimaryHost` does nothing | Logs a startup warning rather than being silently inert — see the `Sparkplug` sub-object table | -| **A metric whose name contains `/` cannot be browse-committed** | [#509](https://gitea.dohertylan.com/dohertj2/lmxopcua/issues/509) | The commit is refused **whole**, in words (`Row N: Name must not contain '/'`) | The derived **tag Name** is a RawPath segment and may not contain `/`, while `metricName` legitimately may. This blocks the spec-mandatory `Node Control/Rebirth`. **Workaround: Manual entry** — author the tag with a `/`-free name and type the slashed `metricName` into the Sparkplug editor, which accepts it. Nothing is silently mis-bound | | **`DataSet` / `Template` / `PropertySet` metrics** | [#515](https://gitea.dohertylan.com/dohertj2/lmxopcua/issues/515) | Skipped with a warning | v1-unsupported; never coerced into a guessed type | -| **The tag editor writes `payloadFormat` into a Sparkplug blob** | [#513](https://gitea.dohertylan.com/dohertj2/lmxopcua/issues/513) | Cosmetic only | `payloadFormat` is Plain-only; the factory routes on the Sparkplug keys and ignores it. Noise in the blob, not a mis-binding | -| **`.Browser` references `.Driver`, not just `.Contracts`** | [#512](https://gitea.dohertylan.com/dohertj2/lmxopcua/issues/512) | A documented layering exception | Taken to reuse `MqttConnection.BuildClientOptions` rather than duplicate TLS/CA-pin logic — see the note at the top of this doc | diff --git a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt.Browser/MqttDriverBrowser.cs b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt.Browser/MqttDriverBrowser.cs index 911a87aa..df8335e0 100644 --- a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt.Browser/MqttDriverBrowser.cs +++ b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt.Browser/MqttDriverBrowser.cs @@ -23,12 +23,12 @@ namespace ZB.MOM.WW.OtOpcUa.Driver.Mqtt.Browser; /// sanctioned publish, and it is never reached by OpenAsync or by any browse call. /// /// -/// Layering note. Unlike every other bespoke browser here, this project references the -/// runtime .Driver project (to reuse MqttConnection.BuildClientOptions rather -/// than fork the TLS / CA-pin path). That is a known, deliberate exception with a documented -/// cost and a documented clean fix — see the comment on that ProjectReference in -/// ZB.MOM.WW.OtOpcUa.Driver.Mqtt.Browser.csproj before copying this project as a -/// template. +/// Layering note. This project depends on the shared leaves only — .Contracts for +/// the Sparkplug wire schema and codec, .Transport for +/// — and NOT on the runtime .Driver project, which +/// it used to reference under a documented exception (Gitea #512). Reuse of the TLS / CA-pin path +/// is what that exception bought, and it is preserved: there is still exactly one copy, now in +/// .Transport. Do not fork it. Pinned by MqttBrowserLayeringTests. /// /// public sealed class MqttDriverBrowser : IDriverBrowser @@ -73,7 +73,7 @@ public sealed class MqttDriverBrowser : IDriverBrowser var filter = BuildRootFilter(opts); var suffix = BuildBrowseClientIdSuffix(); - var clientOptions = MqttConnection.BuildClientOptions(ToBrowseOptions(opts), suffix, _logger); + var clientOptions = MqttClientOptionsFactory.BuildClientOptions(ToBrowseOptions(opts), suffix, _logger); using var openCts = CancellationTokenSource.CreateLinkedTokenSource(cancellationToken); openCts.CancelAfter(OpenBudget(opts)); diff --git a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt.Browser/ZB.MOM.WW.OtOpcUa.Driver.Mqtt.Browser.csproj b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt.Browser/ZB.MOM.WW.OtOpcUa.Driver.Mqtt.Browser.csproj index e32fa07f..44cb9c98 100644 --- a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt.Browser/ZB.MOM.WW.OtOpcUa.Driver.Mqtt.Browser.csproj +++ b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt.Browser/ZB.MOM.WW.OtOpcUa.Driver.Mqtt.Browser.csproj @@ -14,34 +14,36 @@ - + diff --git a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt.Contracts/MqttTagConfigKeys.cs b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt.Contracts/MqttTagConfigKeys.cs index 4a11784b..a40e6ead 100644 --- a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt.Contracts/MqttTagConfigKeys.cs +++ b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt.Contracts/MqttTagConfigKeys.cs @@ -26,6 +26,15 @@ public static class MqttTagConfigKeys /// The concrete MQTT topic a Plain-mode tag subscribes to. public const string Topic = "topic"; + /// + /// The concrete MQTT topic a Plain-mode tag PUBLISHES to when written (Gitea #508). Deliberately + /// separate from : a tag legitimately reads .../state and writes + /// .../set, and publishing a command onto the tag's own read topic would echo back and + /// read as a confirmed device value. Absent ⇒ the tag is not writable, and a write is refused + /// rather than silently dropped. + /// + public const string WriteTopic = "writeTopic"; + /// The Sparkplug group id — the first segment of the tag's binding tuple. public const string GroupId = "groupId"; diff --git a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt.Contracts/MqttTagDefinition.cs b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt.Contracts/MqttTagDefinition.cs index 15c3c3eb..6bd86ecd 100644 --- a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt.Contracts/MqttTagDefinition.cs +++ b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt.Contracts/MqttTagDefinition.cs @@ -60,6 +60,12 @@ namespace ZB.MOM.WW.OtOpcUa.Driver.Mqtt; /// default. Both factories set it from the key's presence so it /// means the same thing in either mode; nothing on the Plain ingest path reads it. /// +/// +/// The concrete MQTT topic a Plain-mode write publishes to, or when the +/// tag declares no write destination — in which case a write is refused with BadNotWritable +/// rather than silently dropped (Gitea #508). Unused in Sparkplug mode, where the command address is +/// the binding tuple itself (an NCMD/DCMD topic derived from group/edgeNode/device). +/// /// The Sparkplug group id; for a Plain-mode tag. /// The Sparkplug edge-node id; for a Plain-mode tag. /// @@ -82,4 +88,5 @@ public sealed record MqttTagDefinition( string? GroupId = null, string? EdgeNodeId = null, string? DeviceId = null, - string? MetricName = null); + string? MetricName = null, + string? WriteTopic = null); diff --git a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt.Contracts/MqttTagDefinitionFactory.cs b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt.Contracts/MqttTagDefinitionFactory.cs index ecdcf121..abceea13 100644 --- a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt.Contracts/MqttTagDefinitionFactory.cs +++ b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt.Contracts/MqttTagDefinitionFactory.cs @@ -104,7 +104,10 @@ public static class MqttTagDefinitionFactory DataType: dataType, Qos: qos, RetainSeed: ReadBoolOrDefault(root, "retainSeed", defaultValue: true), - DataTypeAuthored: dataTypeAuthored); + DataTypeAuthored: dataTypeAuthored, + // Optional, and absence is meaningful: no writeTopic ⇒ not writable (Gitea #508). + // Blank normalises to null so "" and absent cannot mean different things. + WriteTopic: NullIfBlank(ReadString(root, MqttTagConfigKeys.WriteTopic))); return true; } catch (JsonException) { return false; } @@ -190,7 +193,10 @@ public static class MqttTagDefinitionFactory GroupId: groupId, EdgeNodeId: edgeNodeId, DeviceId: deviceId, - MetricName: metricName); + MetricName: metricName, + // Carried through, but unread on the Sparkplug write path: a Sparkplug command is + // addressed by the binding tuple, not by a topic the operator typed. + WriteTopic: NullIfBlank(ReadString(root, MqttTagConfigKeys.WriteTopic))); return true; } catch (JsonException) { return false; } @@ -283,6 +289,17 @@ public static class MqttTagDefinitionFactory ? e.GetString() ?? "" : ""; + /// + /// Normalises a blank string to , for the optional fields where absence is + /// itself meaningful — writeTopic, where "absent" means "not writable" (Gitea #508). Without + /// this, "writeTopic": "" and an absent key would take different code paths and an empty + /// string could reach the transport as a publish topic. + /// + /// The raw read value. + /// The trimmed value, or when blank. + private static string? NullIfBlank(string value) + => string.IsNullOrWhiteSpace(value) ? null : value.Trim(); + private static bool ReadBoolOrDefault(JsonElement o, string name, bool defaultValue) => o.TryGetProperty(name, out var e) && e.ValueKind is JsonValueKind.True or JsonValueKind.False ? e.GetBoolean() diff --git a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt.Contracts/MqttWriteEncoder.cs b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt.Contracts/MqttWriteEncoder.cs new file mode 100644 index 00000000..45ee2981 --- /dev/null +++ b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt.Contracts/MqttWriteEncoder.cs @@ -0,0 +1,304 @@ +using System.Globalization; +using System.Text; +using System.Text.Json; +using Google.Protobuf; +using Org.Eclipse.Tahu.Protobuf; +using ZB.MOM.WW.OtOpcUa.Core.Abstractions; +using ZB.MOM.WW.OtOpcUa.Driver.Mqtt.Sparkplug; +using TahuDataType = Org.Eclipse.Tahu.Protobuf.DataType; + +namespace ZB.MOM.WW.OtOpcUa.Driver.Mqtt; + +/// +/// Why a write could not be encoded. Maps 1:1 onto the status the driver returns, so the reason a +/// write was refused is decided here — once, purely — rather than at the transport. +/// +public enum MqttWriteEncodeError +{ + /// Encoding succeeded. + None = 0, + + /// + /// The tag declares no write destination: a Plain tag with no writeTopic. Deliberately a + /// hard refusal — see 's remarks on why a silent no-op would be + /// worse than the honest read-only behaviour this replaced. + /// + NoWriteDestination, + + /// The value is null, or cannot be represented in the tag's declared data type. + ValueNotRepresentable, + + /// The tag's declared data type has no Sparkplug command encoding. + UnsupportedDataType, +} + +/// +/// Encodes an outbound MQTT write — the payload bytes and the topic to publish them on — for both +/// ingest shapes (Gitea #508). Pure: no I/O, no transport, no clock beyond the Sparkplug command +/// timestamp, so the whole encode contract is unit-testable without a broker. +/// +/// +/// +/// A missing write destination is a refusal, never a no-op. Before #508 the driver +/// implemented no IWritable at all, so every MQTT node materialised read-only — honest, if +/// limited. The one outcome that would be strictly worse is a node that looks writable, +/// accepts a write, returns Good and publishes nothing. So a Plain tag with no writeTopic +/// yields and the driver surfaces +/// BadNotWritable. +/// +/// +/// The write topic is deliberately separate from the read topic. A tag legitimately reads +/// .../state and writes .../set, so the destination is its own authored field rather +/// than being inferred from . Publishing a command back onto +/// the tag's own read topic would, on a broker with no device behind it, echo straight back and +/// read as a confirmed device value. +/// +/// +/// Sparkplug writes are commands, not data. A metric under a device encodes as a DCMD, one +/// owned by the edge node itself as an NCMD, addressed by the tag's stable +/// (group, edgeNode, device?, metric) tuple — never by the per-birth alias, which may be +/// reused for a different metric across a rebirth. No seq is stamped, for the reason +/// documents at length: the sequence stream belongs to the messages +/// an edge node itself publishes, and a strict node validating a host-supplied one against its own +/// counter would have grounds to reject the command. +/// +/// +/// jsonPath is not applied on write. It is a read-side extraction from the read topic's +/// document, and the write topic is a different topic whose payload shape nothing in the tag config +/// describes. A write therefore emits the value as a bare JSON +/// literal. Reconstructing a wrapped document from an extraction path would be guesswork, and +/// guessing a payload shape is how a write lands somewhere nobody authored. +/// +/// +public static class MqttWriteEncoder +{ + /// QoS for an outbound write. See the remarks on . + public const int WriteQos = 0; + + /// + /// Encodes one write for the tag under the driver's . + /// + /// The driver's ingest shape — chosen by driver config, never sniffed from the tag. + /// The resolved tag definition. + /// The value to write. + /// The topic + payload to publish, when this returns . + /// The encode outcome. + /// + /// QoS 0, retain . Retain is the one that matters: a retained + /// command would be replayed by the broker to the device on every future subscribe — including + /// the device's own reconnect — so a single stale write would be re-applied forever. This is the + /// same hazard avoids for NCMD, and the reason neither is + /// configurable. + /// + public static MqttWriteEncodeError TryEncode( + MqttMode mode, + MqttTagDefinition def, + object? value, + out MqttEncodedWrite encoded) + { + ArgumentNullException.ThrowIfNull(def); + encoded = default; + + return mode == MqttMode.SparkplugB + ? TryEncodeSparkplug(def, value, out encoded) + : TryEncodePlain(def, value, out encoded); + } + + /// Encodes a Plain-mode write: the value's text/JSON rendering onto the tag's write topic. + /// The resolved tag definition. + /// The value to write. + /// The topic + payload on success. + /// The encode outcome. + private static MqttWriteEncodeError TryEncodePlain( + MqttTagDefinition def, + object? value, + out MqttEncodedWrite encoded) + { + encoded = default; + + if (string.IsNullOrWhiteSpace(def.WriteTopic)) + { + return MqttWriteEncodeError.NoWriteDestination; + } + + if (value is null) + { + // No MQTT payload means "the value is null" — a zero-length payload is a legitimate + // clear-the-retained-message instruction, which is a different operation entirely. + return MqttWriteEncodeError.ValueNotRepresentable; + } + + string text; + if (def.PayloadFormat == MqttPayloadFormat.Json) + { + try + { + text = JsonSerializer.Serialize(value); + } + catch (NotSupportedException) + { + return MqttWriteEncodeError.ValueNotRepresentable; + } + } + else + { + // Raw and Scalar are both "the payload's UTF-8 text" on the read side; the write side + // mirrors that. Invariant culture, or a comma decimal separator on a European host would + // publish a number the device cannot parse. + text = value is IFormattable f + ? f.ToString(null, CultureInfo.InvariantCulture) + : value.ToString() ?? ""; + } + + encoded = new MqttEncodedWrite(def.WriteTopic!.Trim(), Encoding.UTF8.GetBytes(text)); + return MqttWriteEncodeError.None; + } + + /// Encodes a Sparkplug-mode write as an NCMD (node metric) or DCMD (device metric). + /// The resolved tag definition. + /// The value to write. + /// The topic + payload on success. + /// The encode outcome. + private static MqttWriteEncodeError TryEncodeSparkplug( + MqttTagDefinition def, + object? value, + out MqttEncodedWrite encoded) + { + encoded = default; + + if (def.GroupId is not { Length: > 0 } groupId + || def.EdgeNodeId is not { Length: > 0 } edgeNodeId + || def.MetricName is not { Length: > 0 } metricName) + { + // A Sparkplug tag's command address IS its binding tuple, so an incomplete tuple is the + // Sparkplug equivalent of a missing write topic rather than a different kind of failure. + return MqttWriteEncodeError.NoWriteDestination; + } + + if (value is null) + { + return MqttWriteEncodeError.ValueNotRepresentable; + } + + var metric = new Payload.Types.Metric { Name = metricName }; + var error = ApplyValue(metric, def.DataType, value); + if (error != MqttWriteEncodeError.None) + { + return error; + } + + var timestampMs = (ulong)DateTimeOffset.UtcNow.ToUnixTimeMilliseconds(); + metric.Timestamp = timestampMs; + + var payload = new Payload { Timestamp = timestampMs }; + payload.Metrics.Add(metric); + + // A device metric is commanded with DCMD addressed to that device; an edge-node metric with + // NCMD. Sending a device metric as NCMD would address the node, which does not own it. + var isDevice = def.DeviceId is { Length: > 0 }; + var topic = SparkplugTopic.Format( + groupId, + isDevice ? SparkplugMessageType.DCMD : SparkplugMessageType.NCMD, + edgeNodeId, + isDevice ? def.DeviceId : null); + + encoded = new MqttEncodedWrite(topic, payload.ToByteArray()); + return MqttWriteEncodeError.None; + } + + /// + /// Sets the metric's typed value field and its Datatype from the tag's declared type. + /// + /// The metric being built. + /// The tag's declared driver data type. + /// The value to convert. + /// The encode outcome. + /// + /// Datatype is set explicitly rather than left for the node to infer, for the same reason + /// sets it: a node that treats a missing or ambiguous field as + /// malformed drops the command silently, and a silently-dropped write is the failure mode this + /// whole feature must not have. + /// + private static MqttWriteEncodeError ApplyValue( + Payload.Types.Metric metric, + DriverDataType dataType, + object value) + { + try + { + switch (dataType) + { + case DriverDataType.Boolean: + metric.Datatype = (uint)TahuDataType.Boolean; + metric.BooleanValue = Convert.ToBoolean(value, CultureInfo.InvariantCulture); + return MqttWriteEncodeError.None; + + case DriverDataType.Int16: + metric.Datatype = (uint)TahuDataType.Int16; + metric.IntValue = unchecked((uint)Convert.ToInt16(value, CultureInfo.InvariantCulture)); + return MqttWriteEncodeError.None; + + case DriverDataType.Int32: + metric.Datatype = (uint)TahuDataType.Int32; + metric.IntValue = unchecked((uint)Convert.ToInt32(value, CultureInfo.InvariantCulture)); + return MqttWriteEncodeError.None; + + case DriverDataType.Int64: + metric.Datatype = (uint)TahuDataType.Int64; + metric.LongValue = unchecked((ulong)Convert.ToInt64(value, CultureInfo.InvariantCulture)); + return MqttWriteEncodeError.None; + + case DriverDataType.UInt16: + metric.Datatype = (uint)TahuDataType.Uint16; + metric.IntValue = Convert.ToUInt16(value, CultureInfo.InvariantCulture); + return MqttWriteEncodeError.None; + + case DriverDataType.UInt32: + metric.Datatype = (uint)TahuDataType.Uint32; + metric.IntValue = Convert.ToUInt32(value, CultureInfo.InvariantCulture); + return MqttWriteEncodeError.None; + + case DriverDataType.UInt64: + metric.Datatype = (uint)TahuDataType.Uint64; + metric.LongValue = Convert.ToUInt64(value, CultureInfo.InvariantCulture); + return MqttWriteEncodeError.None; + + case DriverDataType.Float32: + metric.Datatype = (uint)TahuDataType.Float; + metric.FloatValue = Convert.ToSingle(value, CultureInfo.InvariantCulture); + return MqttWriteEncodeError.None; + + case DriverDataType.Float64: + metric.Datatype = (uint)TahuDataType.Double; + metric.DoubleValue = Convert.ToDouble(value, CultureInfo.InvariantCulture); + return MqttWriteEncodeError.None; + + case DriverDataType.String: + metric.Datatype = (uint)TahuDataType.String; + metric.StringValue = value as string ?? Convert.ToString(value, CultureInfo.InvariantCulture) ?? ""; + return MqttWriteEncodeError.None; + + case DriverDataType.DateTime: + metric.Datatype = (uint)TahuDataType.DateTime; + metric.LongValue = (ulong)new DateTimeOffset( + Convert.ToDateTime(value, CultureInfo.InvariantCulture).ToUniversalTime()) + .ToUnixTimeMilliseconds(); + return MqttWriteEncodeError.None; + + default: + // DriverDataType.Reference and anything added later: refuse rather than guess a + // wire representation. An unsupported type must not become a String on the wire. + return MqttWriteEncodeError.UnsupportedDataType; + } + } + catch (Exception ex) when (ex is InvalidCastException or FormatException or OverflowException) + { + return MqttWriteEncodeError.ValueNotRepresentable; + } + } +} + +/// One encoded outbound write: where it goes and what it carries. +/// The concrete topic to publish on. +/// The encoded message body. +public readonly record struct MqttEncodedWrite(string Topic, byte[] Payload); diff --git a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt/Sparkplug/RebirthRequester.cs b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt.Contracts/Sparkplug/RebirthRequester.cs similarity index 100% rename from src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt/Sparkplug/RebirthRequester.cs rename to src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt.Contracts/Sparkplug/RebirthRequester.cs diff --git a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt/Sparkplug/SparkplugCodec.cs b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt.Contracts/Sparkplug/SparkplugCodec.cs similarity index 100% rename from src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt/Sparkplug/SparkplugCodec.cs rename to src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt.Contracts/Sparkplug/SparkplugCodec.cs diff --git a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt/Sparkplug/SparkplugTopic.cs b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt.Contracts/Sparkplug/SparkplugTopic.cs similarity index 100% rename from src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt/Sparkplug/SparkplugTopic.cs rename to src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt.Contracts/Sparkplug/SparkplugTopic.cs diff --git a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt.Transport/MqttClientOptionsFactory.cs b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt.Transport/MqttClientOptionsFactory.cs new file mode 100644 index 00000000..4e4e9278 --- /dev/null +++ b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt.Transport/MqttClientOptionsFactory.cs @@ -0,0 +1,244 @@ +using System.Net.Security; +using System.Security.Cryptography; +using System.Security.Cryptography.X509Certificates; +using Microsoft.Extensions.Logging; +using MQTTnet; + +namespace ZB.MOM.WW.OtOpcUa.Driver.Mqtt; + +/// +/// Assembles the MQTTnet client options for every broker connection this repo opens — the runtime +/// driver's, the browse session's and the Test-connect probe's alike. +/// +/// +/// +/// Why this is its own type in its own assembly (Gitea #512). It owns the security posture of +/// an MQTT connection: whether TLS is on, whether an untrusted certificate is tolerated, and — when a +/// PEM CA is pinned — the custom chain validator that replaces the OS trust anchor. Both the runtime +/// driver and the browse session must connect with identical semantics, and the failure mode +/// of a second copy is not a crash but a silent divergence: browse quietly not honouring a pinned CA +/// while the driver does. Previously this lived on MqttConnection and .Browser took a +/// documented layering exception to reach it; the exception is gone, and this is the one copy. +/// +/// +/// Pure. No I/O, no network, nothing read from disk — so it is safe to call on a +/// config-validation path. A configured is read +/// lazily by the returned validation callback at handshake time, not here. +/// +/// +public static class MqttClientOptionsFactory +{ + /// + /// Assembles the MQTTnet client options from . Pure — it performs + /// no I/O and touches no network, so it is safe to call on a config-validation path. A + /// configured is read lazily by the + /// returned certificate-validation callback at handshake time, not here. + /// + /// Broker connection settings. + /// + /// Appended to , so a browse/probe session can share + /// the configured identity without colliding with the driver's own client id. When both are + /// empty MQTTnet generates a random client id. + /// + /// Optional logger for the deferred CA-pin load; never receives credentials. + public static MqttClientOptions BuildClientOptions( + MqttDriverOptions options, + string? clientIdSuffix, + ILogger? logger = null) + { + ArgumentNullException.ThrowIfNull(options); + + var builder = new MqttClientOptionsBuilder() + .WithTcpServer(options.Host, options.Port) + .WithProtocolVersion(MapProtocolVersion(options.ProtocolVersion)) + .WithCleanSession(options.CleanSession) + .WithKeepAlivePeriod(TimeSpan.FromSeconds(options.KeepAliveSeconds)) + .WithTimeout(TimeSpan.FromSeconds(options.ConnectTimeoutSeconds)); + + var clientId = string.Concat(options.ClientId ?? string.Empty, clientIdSuffix ?? string.Empty); + if (!string.IsNullOrWhiteSpace(clientId)) + { + builder = builder.WithClientId(clientId); + } + + // An empty username means "connect anonymously"; MQTTnet would otherwise send an empty + // CONNECT username, which some brokers treat as a failed auth rather than as anonymous. + if (!string.IsNullOrEmpty(options.Username)) + { + builder = builder.WithCredentials(options.Username, options.Password ?? string.Empty); + } + + builder = builder.WithTlsOptions(tls => ConfigureTls(tls, options, logger)); + + return builder.Build(); + } + + private static void ConfigureTls(MqttClientTlsOptionsBuilder tls, MqttDriverOptions options, ILogger? logger) + { + if (!options.UseTls) + { + tls.UseTls(false); + return; + } + + tls.UseTls(true) + // Explicit SNI / hostname-verification target. Left unset, name validation depends on + // how MQTTnet infers the host from the endpoint — pin it to the configured host. + .WithTargetHost(options.Host); + + if (options.AllowUntrustedServerCertificate) + { + // Dev / on-prem escape hatch, off unless explicitly enabled — mirrors the + // ServerHistorian TLS knobs. This is the ONLY branch that accepts a bad certificate. + tls.WithAllowUntrustedCertificates(true) + .WithIgnoreCertificateChainErrors(true) + .WithCertificateValidationHandler(static _ => true); + return; + } + + if (string.IsNullOrWhiteSpace(options.CaCertificatePath)) + { + // No pin: MQTTnet's default handler validates against the OS trust store. + return; + } + + tls.WithCertificateValidationHandler(CreatePinnedCaValidator(options.CaCertificatePath, logger)); + } + + /// + /// Builds a certificate-validation callback that pins the broker chain to the PEM CA at + /// . The file is loaded lazily on first validation + /// (i.e. during the TLS handshake) so that this — and therefore + /// — stays free of I/O. Any failure to load or to chain + /// up to the pinned CA rejects the certificate: fail closed. + /// + private static Func CreatePinnedCaValidator( + string caCertificatePath, + ILogger? logger) + { + var trustedRoots = new Lazy( + () => LoadPemCa(caCertificatePath, logger), + LazyThreadSafetyMode.ExecutionAndPublication); + + return args => ValidateAgainstPinnedCa(args, trustedRoots.Value, caCertificatePath, logger); + } + + private static X509Certificate2Collection? LoadPemCa(string caCertificatePath, ILogger? logger) + { + try + { + var collection = new X509Certificate2Collection(); + collection.ImportFromPemFile(caCertificatePath); + if (collection.Count != 0) + { + return collection; + } + + logger?.LogError("MQTT CA pin '{CaCertificatePath}' contains no certificates; broker TLS will be rejected.", caCertificatePath); + } + catch (Exception ex) + { + logger?.LogError(ex, "MQTT CA pin '{CaCertificatePath}' could not be loaded; broker TLS will be rejected.", caCertificatePath); + } + + return null; + } + + private static bool ValidateAgainstPinnedCa( + MqttClientCertificateValidationEventArgs args, + X509Certificate2Collection? trustedRoots, + string caCertificatePath, + ILogger? logger) + { + if (trustedRoots is null || trustedRoots.Count == 0) + { + // Unreadable pin ⇒ no trust anchor ⇒ refuse. Never silently degrade to the OS store. + return false; + } + + // The pin replaces the trust anchor only. Anything else the platform flagged — a hostname + // mismatch, an absent certificate — remains fatal. + if ((args.SslPolicyErrors & ~SslPolicyErrors.RemoteCertificateChainErrors) != SslPolicyErrors.None) + { + logger?.LogError("MQTT broker certificate rejected: {SslPolicyErrors}.", args.SslPolicyErrors); + return false; + } + + if (args.Certificate is null) + { + return false; + } + + var presented = args.Certificate as X509Certificate2; + X509Certificate2? owned = null; + if (presented is null) + { + owned = X509CertificateLoader.LoadCertificate(args.Certificate.Export(X509ContentType.Cert)); + presented = owned; + } + + try + { + using var chain = new X509Chain(); + chain.ChainPolicy.TrustMode = X509ChainTrustMode.CustomRootTrust; + chain.ChainPolicy.CustomTrustStore.AddRange(trustedRoots); + + // Revocation is deliberately not checked: the pin exists precisely for self-issued + // dev / on-prem CAs, which typically publish no CRL or OCSP responder, so checking + // would fail every such chain. The pin itself is the revocation story — remove the CA. + chain.ChainPolicy.RevocationMode = X509RevocationMode.NoCheck; + + // Seed the intermediates the broker presented during the handshake. A leaf signed by + // an intermediate that chains up to the pinned root is a common broker topology, and + // that intermediate arrives on the wire rather than being installed locally — without + // it in the candidate pool a legitimate trust path simply cannot be built. + // CustomRootTrust still means only `trustedRoots` may terminate the chain, so a rogue + // self-signed cert smuggled in here cannot become an anchor. + // Both sources are read: platforms differ in whether the peer-supplied intermediates + // land in the incoming chain's elements, its ExtraStore, or both. + if (args.Chain is not null) + { + foreach (var element in args.Chain.ChainElements) + { + chain.ChainPolicy.ExtraStore.Add(element.Certificate); + } + + chain.ChainPolicy.ExtraStore.AddRange(args.Chain.ChainPolicy.ExtraStore); + } + + if (chain.Build(presented)) + { + return true; + } + + logger?.LogError( + "MQTT broker certificate does not chain to the pinned CA '{CaCertificatePath}': {ChainStatus}.", + caCertificatePath, + string.Join(", ", chain.ChainStatus.Select(s => s.Status))); + return false; + } + catch (CryptographicException ex) + { + // X509Chain.Build throws on a malformed certificate or an unusable policy. An exception + // escaping a TLS validation callback surfaces as an opaque handshake crash, so classify + // it here and refuse: an unverifiable certificate is a rejected certificate. + logger?.LogError( + ex, + "MQTT broker certificate could not be validated against the pinned CA '{CaCertificatePath}'; rejecting.", + caCertificatePath); + return false; + } + finally + { + owned?.Dispose(); + } + } + + private static MQTTnet.Formatter.MqttProtocolVersion MapProtocolVersion(MqttProtocolVersion version) + => version switch + { + MqttProtocolVersion.V311 => MQTTnet.Formatter.MqttProtocolVersion.V311, + MqttProtocolVersion.V500 => MQTTnet.Formatter.MqttProtocolVersion.V500, + _ => throw new ArgumentOutOfRangeException(nameof(version), version, "Unsupported MQTT protocol version."), + }; +} diff --git a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt.Transport/ZB.MOM.WW.OtOpcUa.Driver.Mqtt.Transport.csproj b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt.Transport/ZB.MOM.WW.OtOpcUa.Driver.Mqtt.Transport.csproj new file mode 100644 index 00000000..9d2b4e39 --- /dev/null +++ b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt.Transport/ZB.MOM.WW.OtOpcUa.Driver.Mqtt.Transport.csproj @@ -0,0 +1,40 @@ + + + + net10.0 + enable + enable + latest + true + ZB.MOM.WW.OtOpcUa.Driver.Mqtt + ZB.MOM.WW.OtOpcUa.Driver.Mqtt.Transport + + + + + + + + + + + + + diff --git a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt/MqttConnection.cs b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt/MqttConnection.cs index 1733fefc..6582de2e 100644 --- a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt/MqttConnection.cs +++ b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt/MqttConnection.cs @@ -568,7 +568,7 @@ public sealed class MqttConnection : IAsyncDisposable, IMqttSubscribeTransport, MqttClientOptions clientOptions; try { - clientOptions = BuildClientOptions(_options, clientIdSuffix: null, _logger); + clientOptions = MqttClientOptionsFactory.BuildClientOptions(_options, clientIdSuffix: null, _logger); } catch (Exception ex) { @@ -1345,217 +1345,4 @@ public sealed class MqttConnection : IAsyncDisposable, IMqttSubscribeTransport, } } - /// - /// Assembles the MQTTnet client options from . Pure — it performs - /// no I/O and touches no network, so it is safe to call on a config-validation path. A - /// configured is read lazily by the - /// returned certificate-validation callback at handshake time, not here. - /// - /// Broker connection settings. - /// - /// Appended to , so a browse/probe session can share - /// the configured identity without colliding with the driver's own client id. When both are - /// empty MQTTnet generates a random client id. - /// - /// Optional logger for the deferred CA-pin load; never receives credentials. - public static MqttClientOptions BuildClientOptions( - MqttDriverOptions options, - string? clientIdSuffix, - ILogger? logger = null) - { - ArgumentNullException.ThrowIfNull(options); - - var builder = new MqttClientOptionsBuilder() - .WithTcpServer(options.Host, options.Port) - .WithProtocolVersion(MapProtocolVersion(options.ProtocolVersion)) - .WithCleanSession(options.CleanSession) - .WithKeepAlivePeriod(TimeSpan.FromSeconds(options.KeepAliveSeconds)) - .WithTimeout(TimeSpan.FromSeconds(options.ConnectTimeoutSeconds)); - - var clientId = string.Concat(options.ClientId ?? string.Empty, clientIdSuffix ?? string.Empty); - if (!string.IsNullOrWhiteSpace(clientId)) - { - builder = builder.WithClientId(clientId); - } - - // An empty username means "connect anonymously"; MQTTnet would otherwise send an empty - // CONNECT username, which some brokers treat as a failed auth rather than as anonymous. - if (!string.IsNullOrEmpty(options.Username)) - { - builder = builder.WithCredentials(options.Username, options.Password ?? string.Empty); - } - - builder = builder.WithTlsOptions(tls => ConfigureTls(tls, options, logger)); - - return builder.Build(); - } - - private static void ConfigureTls(MqttClientTlsOptionsBuilder tls, MqttDriverOptions options, ILogger? logger) - { - if (!options.UseTls) - { - tls.UseTls(false); - return; - } - - tls.UseTls(true) - // Explicit SNI / hostname-verification target. Left unset, name validation depends on - // how MQTTnet infers the host from the endpoint — pin it to the configured host. - .WithTargetHost(options.Host); - - if (options.AllowUntrustedServerCertificate) - { - // Dev / on-prem escape hatch, off unless explicitly enabled — mirrors the - // ServerHistorian TLS knobs. This is the ONLY branch that accepts a bad certificate. - tls.WithAllowUntrustedCertificates(true) - .WithIgnoreCertificateChainErrors(true) - .WithCertificateValidationHandler(static _ => true); - return; - } - - if (string.IsNullOrWhiteSpace(options.CaCertificatePath)) - { - // No pin: MQTTnet's default handler validates against the OS trust store. - return; - } - - tls.WithCertificateValidationHandler(CreatePinnedCaValidator(options.CaCertificatePath, logger)); - } - - /// - /// Builds a certificate-validation callback that pins the broker chain to the PEM CA at - /// . The file is loaded lazily on first validation - /// (i.e. during the TLS handshake) so that this — and therefore - /// — stays free of I/O. Any failure to load or to chain - /// up to the pinned CA rejects the certificate: fail closed. - /// - private static Func CreatePinnedCaValidator( - string caCertificatePath, - ILogger? logger) - { - var trustedRoots = new Lazy( - () => LoadPemCa(caCertificatePath, logger), - LazyThreadSafetyMode.ExecutionAndPublication); - - return args => ValidateAgainstPinnedCa(args, trustedRoots.Value, caCertificatePath, logger); - } - - private static X509Certificate2Collection? LoadPemCa(string caCertificatePath, ILogger? logger) - { - try - { - var collection = new X509Certificate2Collection(); - collection.ImportFromPemFile(caCertificatePath); - if (collection.Count != 0) - { - return collection; - } - - logger?.LogError("MQTT CA pin '{CaCertificatePath}' contains no certificates; broker TLS will be rejected.", caCertificatePath); - } - catch (Exception ex) - { - logger?.LogError(ex, "MQTT CA pin '{CaCertificatePath}' could not be loaded; broker TLS will be rejected.", caCertificatePath); - } - - return null; - } - - private static bool ValidateAgainstPinnedCa( - MqttClientCertificateValidationEventArgs args, - X509Certificate2Collection? trustedRoots, - string caCertificatePath, - ILogger? logger) - { - if (trustedRoots is null || trustedRoots.Count == 0) - { - // Unreadable pin ⇒ no trust anchor ⇒ refuse. Never silently degrade to the OS store. - return false; - } - - // The pin replaces the trust anchor only. Anything else the platform flagged — a hostname - // mismatch, an absent certificate — remains fatal. - if ((args.SslPolicyErrors & ~SslPolicyErrors.RemoteCertificateChainErrors) != SslPolicyErrors.None) - { - logger?.LogError("MQTT broker certificate rejected: {SslPolicyErrors}.", args.SslPolicyErrors); - return false; - } - - if (args.Certificate is null) - { - return false; - } - - var presented = args.Certificate as X509Certificate2; - X509Certificate2? owned = null; - if (presented is null) - { - owned = X509CertificateLoader.LoadCertificate(args.Certificate.Export(X509ContentType.Cert)); - presented = owned; - } - - try - { - using var chain = new X509Chain(); - chain.ChainPolicy.TrustMode = X509ChainTrustMode.CustomRootTrust; - chain.ChainPolicy.CustomTrustStore.AddRange(trustedRoots); - - // Revocation is deliberately not checked: the pin exists precisely for self-issued - // dev / on-prem CAs, which typically publish no CRL or OCSP responder, so checking - // would fail every such chain. The pin itself is the revocation story — remove the CA. - chain.ChainPolicy.RevocationMode = X509RevocationMode.NoCheck; - - // Seed the intermediates the broker presented during the handshake. A leaf signed by - // an intermediate that chains up to the pinned root is a common broker topology, and - // that intermediate arrives on the wire rather than being installed locally — without - // it in the candidate pool a legitimate trust path simply cannot be built. - // CustomRootTrust still means only `trustedRoots` may terminate the chain, so a rogue - // self-signed cert smuggled in here cannot become an anchor. - // Both sources are read: platforms differ in whether the peer-supplied intermediates - // land in the incoming chain's elements, its ExtraStore, or both. - if (args.Chain is not null) - { - foreach (var element in args.Chain.ChainElements) - { - chain.ChainPolicy.ExtraStore.Add(element.Certificate); - } - - chain.ChainPolicy.ExtraStore.AddRange(args.Chain.ChainPolicy.ExtraStore); - } - - if (chain.Build(presented)) - { - return true; - } - - logger?.LogError( - "MQTT broker certificate does not chain to the pinned CA '{CaCertificatePath}': {ChainStatus}.", - caCertificatePath, - string.Join(", ", chain.ChainStatus.Select(s => s.Status))); - return false; - } - catch (CryptographicException ex) - { - // X509Chain.Build throws on a malformed certificate or an unusable policy. An exception - // escaping a TLS validation callback surfaces as an opaque handshake crash, so classify - // it here and refuse: an unverifiable certificate is a rejected certificate. - logger?.LogError( - ex, - "MQTT broker certificate could not be validated against the pinned CA '{CaCertificatePath}'; rejecting.", - caCertificatePath); - return false; - } - finally - { - owned?.Dispose(); - } - } - - private static MQTTnet.Formatter.MqttProtocolVersion MapProtocolVersion(MqttProtocolVersion version) - => version switch - { - MqttProtocolVersion.V311 => MQTTnet.Formatter.MqttProtocolVersion.V311, - MqttProtocolVersion.V500 => MQTTnet.Formatter.MqttProtocolVersion.V500, - _ => throw new ArgumentOutOfRangeException(nameof(version), version, "Unsupported MQTT protocol version."), - }; } diff --git a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt/MqttDriver.cs b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt/MqttDriver.cs index a6814efb..4f1e4d54 100644 --- a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt/MqttDriver.cs +++ b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt/MqttDriver.cs @@ -63,8 +63,31 @@ namespace ZB.MOM.WW.OtOpcUa.Driver.Mqtt; /// /// public sealed class MqttDriver - : IDriver, ITagDiscovery, ISubscribable, IReadable, IHostConnectivityProbe, IRediscoverable, IAsyncDisposable, IDisposable + : IDriver, ITagDiscovery, ISubscribable, IReadable, IWritable, IHostConnectivityProbe, IRediscoverable, + IAsyncDisposable, IDisposable { + // ---- Write status codes (Gitea #508) ---- + // + // NAMED constants, never inline literals: StatusCodeParityTests checks each against the pinned SDK + // by reflection and can only see a named const (Gitea #497/#519). The names must match + // Opc.Ua.StatusCodes members exactly. + + /// The reference is not an authored tag of this driver's live ingest path. + private const uint StatusBadNodeIdUnknown = 0x80340000u; + + /// + /// The tag declares no write destination — a Plain tag with no writeTopic, or a Sparkplug + /// tag with an incomplete binding tuple. The honest answer to "can I write this?", and the whole + /// reason the encoder refuses rather than no-opping. + /// + private const uint StatusBadNotWritable = 0x803B0000u; + + /// The supplied value cannot be represented in the tag's declared data type. + private const uint StatusBadTypeMismatch = 0x80740000u; + + /// The publish failed, or the driver has no live broker connection to publish through. + private const uint StatusBadCommunicationError = 0x80050000u; + /// /// Rough per-tag driver-attributable cost: the parsed (five /// strings — RawPath, topic, JSONPath and the record header) plus its cached @@ -480,11 +503,16 @@ public sealed class MqttDriver DriverDataType: ResolveDiscoveredDataType(def), IsArray: false, ArrayDim: null, - // MQTT ingest is one-way in P1 — this driver implements no IWritable leg, so every - // node is ViewOnly rather than an Operate node whose writes would silently vanish. - SecurityClass: SecurityClassification.ViewOnly, + // Gitea #508: MQTT now has a write leg, so a tag that declares a write destination is + // an Operate node. One that does not stays ViewOnly — the point of the distinction is + // that a node must never look writable while its writes would go nowhere. + SecurityClass: CanWrite(def) ? SecurityClassification.Operate : SecurityClassification.ViewOnly, IsHistorized: false, IsAlarm: false, + // Default-off, deliberately: MQTT gives no application-level write acknowledgement, and + // a Sparkplug command is not necessarily idempotent (a pulse, a recipe step, a counter + // increment). The resilience pipeline must not replay one by default; an operator opts + // in per tag via Tag.WriteIdempotent. WriteIdempotent: false)); } @@ -585,6 +613,142 @@ public sealed class MqttDriver return Task.FromResult>(results); } + // ---- IWritable (Gitea #508) ---- + + /// + /// Whether declares somewhere for a write to go: a writeTopic in + /// Plain mode, or a complete command tuple in Sparkplug mode. + /// + /// The resolved tag definition. + /// when a write to this tag can be addressed. + /// + /// Asked of the SAME encoder the write path uses, so the browse-time answer and the runtime + /// answer cannot drift — a node advertised Operate that then refuses every write would be the + /// silent-no-op failure in a new costume. The value is irrelevant here, so a probe value of the + /// tag's own type is encoded and thrown away. + /// + private bool CanWrite(MqttTagDefinition def) + => MqttWriteEncoder.TryEncode(_options.Mode, def, ProbeValueFor(def.DataType), out _) + != MqttWriteEncodeError.NoWriteDestination; + + /// A never-published sentinel value of the given type, for the probe. + /// The tag's declared data type. + /// A representable value, or a string for types with no natural zero. + private static object ProbeValueFor(DriverDataType dataType) => dataType switch + { + DriverDataType.Boolean => false, + DriverDataType.DateTime => DateTime.UnixEpoch, + DriverDataType.String or DriverDataType.Reference => "", + _ => 0, + }; + + /// + /// Publishes a batch of writes: one MQTT message per request, to the tag's write topic (Plain) + /// or as an NCMD/DCMD command (Sparkplug B). + /// + /// The requested writes. + /// Cancellation for the batch. + /// One result per request, in request order. + /// + /// + /// Never throws, and one bad request never fails the batch. Each slot carries its own + /// status, matching : failing the whole call would let a single stale + /// reference reject every other write in the same OPC UA request. + /// + /// + /// Good means "published", not "applied". MQTT has no application-level write + /// acknowledgement, so a successful PUBLISH is the strongest truth available at this seam — + /// the same optimistic-Good posture Galaxy's fire-and-forget write takes. Device-level truth + /// arrives afterwards, as the echoed value on the tag's read topic, and corrects the node + /// through the ordinary ingest path. A write whose PUBLISH itself fails is reported Bad: the + /// optimism covers what the device did with the command, never whether it was sent. + /// + /// + /// Consequence, inherited knowingly (see #5, write-outcome self-correction): because the + /// transport cannot report device-level rejection, this driver can never surface a failed + /// device write — exactly the property Galaxy has. A device that silently ignores a + /// command leaves the node reading whatever it last published, which is the honest picture; a + /// device that has no echo topic at all leaves the operator with no confirmation, which is a + /// property of the plant's topic design rather than of this code. + /// + /// + public async Task> WriteAsync( + IReadOnlyList writes, + CancellationToken cancellationToken) + { + ArgumentNullException.ThrowIfNull(writes); + + var results = new WriteResult[writes.Count]; + var connection = _connection; + + for (var i = 0; i < writes.Count; i++) + { + var w = writes[i]; + + if (!TryResolve(w.FullReference, out var def)) + { + results[i] = new WriteResult(StatusBadNodeIdUnknown); + continue; + } + + var error = MqttWriteEncoder.TryEncode(_options.Mode, def, w.Value, out var encoded); + if (error != MqttWriteEncodeError.None) + { + results[i] = new WriteResult(MapEncodeError(error)); + continue; + } + + if (connection is null) + { + // Not yet connected, or torn down mid-rebuild. Reported rather than queued: a write + // buffered now and flushed at an unbounded later moment is the replay hazard + // WriteIdempotent exists to prevent, decided by the driver instead of the operator. + results[i] = new WriteResult(StatusBadCommunicationError); + continue; + } + + try + { + await connection + .PublishAsync(encoded.Topic, encoded.Payload, MqttWriteEncoder.WriteQos, retain: false, cancellationToken) + .ConfigureAwait(false); + results[i] = new WriteResult(0u); + } + catch (OperationCanceledException) when (cancellationToken.IsCancellationRequested) + { + // Caller-cancelled: fill the remainder rather than throwing, so the batch shape holds. + for (var j = i; j < writes.Count; j++) + { + results[j] = new WriteResult(StatusBadCommunicationError); + } + + return results; + } + catch (Exception ex) + { + _logger?.LogWarning( + ex, + "MQTT driver '{DriverId}': write to '{Reference}' could not be published on '{Topic}'.", + _driverInstanceId, + w.FullReference, + encoded.Topic); + results[i] = new WriteResult(StatusBadCommunicationError); + } + } + + return results; + } + + /// Maps an encode refusal to the status an OPC UA client sees. + /// The encode outcome. + /// The OPC UA status code. + private static uint MapEncodeError(MqttWriteEncodeError error) => error switch + { + MqttWriteEncodeError.NoWriteDestination => StatusBadNotWritable, + MqttWriteEncodeError.UnsupportedDataType => StatusBadNotWritable, + _ => StatusBadTypeMismatch, + }; + // ---- IHostConnectivityProbe ---- /// The single broker this driver instance talks to, as host:port. @@ -939,7 +1103,7 @@ public sealed class MqttDriver // ONLINE half without the Will is strictly worse than shipping neither: a crashed node // would leave a retained ONLINE asserting forever that a dead host is alive, which is the // exact state the mechanism exists to prevent. The Will belongs in - // MqttConnection.BuildClientOptions; until it is there this flag stays inert and said so. + // MqttClientOptionsFactory.BuildClientOptions; until it is there this flag stays inert and said so. _logger?.LogWarning( "MQTT driver '{DriverId}': Sparkplug actAsPrimaryHost is set but primary-host STATE " + "publishing is not implemented — this driver observes STATE and never publishes one. " diff --git a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt/MqttDriverProbe.cs b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt/MqttDriverProbe.cs index d14886a4..e71666da 100644 --- a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt/MqttDriverProbe.cs +++ b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt/MqttDriverProbe.cs @@ -17,7 +17,7 @@ namespace ZB.MOM.WW.OtOpcUa.Driver.Mqtt; /// /// /// -/// Reuses for the connect (TLS, +/// Reuses for the connect (TLS, /// CA-pin, credentials, protocol-version mapping) rather than hand-rolling a second /// connect path — a duplicate would be a security divergence. A distinct client-id suffix /// (see ) keeps a probe from colliding with the driver's @@ -101,7 +101,7 @@ public sealed class MqttDriverProbe : IDriverProbe MqttClientOptions clientOptions; try { - clientOptions = MqttConnection.BuildClientOptions( + clientOptions = MqttClientOptionsFactory.BuildClientOptions( probeOptions, ProbeClientIdPrefix + Guid.NewGuid().ToString("N")[..8], NullLogger.Instance); diff --git a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt/Sparkplug/SparkplugIngestor.cs b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt/Sparkplug/SparkplugIngestor.cs index cb14d71f..a5c294ef 100644 --- a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt/Sparkplug/SparkplugIngestor.cs +++ b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt/Sparkplug/SparkplugIngestor.cs @@ -1038,7 +1038,7 @@ public sealed class SparkplugIngestor /// while their host is offline. Shipping the ONLINE half without the Will is strictly worse /// than shipping neither: an OtOpcUa node that crashes would leave a retained ONLINE asserting /// forever that a dead host is alive, which is precisely the state the mechanism exists to - /// prevent. The Will has to be set inside MqttConnection.BuildClientOptions — a + /// prevent. The Will has to be set inside MqttClientOptionsFactory.BuildClientOptions — a /// different file, a different task — so this task implements the observation half and /// warns when the flag is set. See the design doc §7/§8. /// diff --git a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt/ZB.MOM.WW.OtOpcUa.Driver.Mqtt.csproj b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt/ZB.MOM.WW.OtOpcUa.Driver.Mqtt.csproj index 55ffef0b..79edc634 100644 --- a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt/ZB.MOM.WW.OtOpcUa.Driver.Mqtt.csproj +++ b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt/ZB.MOM.WW.OtOpcUa.Driver.Mqtt.csproj @@ -14,6 +14,7 @@ + diff --git a/src/Server/ZB.MOM.WW.OtOpcUa.AdminUI/Browsing/SparkplugRebirthScope.cs b/src/Server/ZB.MOM.WW.OtOpcUa.AdminUI/Browsing/SparkplugRebirthScope.cs new file mode 100644 index 00000000..10c9f965 --- /dev/null +++ b/src/Server/ZB.MOM.WW.OtOpcUa.AdminUI/Browsing/SparkplugRebirthScope.cs @@ -0,0 +1,80 @@ +namespace ZB.MOM.WW.OtOpcUa.AdminUI.Browsing; + +/// +/// Validates an operator-typed Sparkplug rebirth scope for the raw browse modal (Gitea #514). +/// +/// +/// +/// Why a typed scope exists at all. The Sparkplug browse tree is birth-driven, and +/// births are never retained. On a quiet plant — every edge node birthed long before the browse +/// window opened, and is now publishing only NDATA/DDATA — the tree comes up empty. The +/// Request-rebirth scope was derived solely from a tree selection, so the one action that would +/// populate the tree could not be armed: the action needed the tree, and the tree needed the action. +/// +/// +/// This is advisory, not the gate. MqttBrowseSession.ResolveRebirthTargets re-resolves +/// the scope and refuses an unusable one before any message leaves, and the +/// DriverOperator policy is enforced server-side in BrowserSessionService. This exists +/// so the modal does not let an operator arm a two-click confirm that cannot succeed. Being the +/// looser of the two is therefore safe; being the stricter one would block a scope the session +/// accepts, which is the failure mode the repo's authoring surfaces avoid by convention. +/// +/// +/// A group-only scope is rejected on purpose. The session resolves a bare group by enumerating +/// the edge nodes it has observed — of which an empty tree has none — so it could not help the +/// case this input exists for, and on a populated tree it would fan one click across the whole group. +/// The tree-selection path still offers a group scope, with its own confirm-time cap and warning. +/// +/// +public static class SparkplugRebirthScope +{ + /// MQTT wildcards. A rebirth addresses one edge node exactly; a wildcard cannot name one. + private static readonly char[] Wildcards = ['+', '#']; + + /// + /// Describes why is not a usable {group}/{edgeNode} rebirth scope. + /// + /// The operator-typed scope, untrimmed; is treated as blank. + /// The operator-facing error text, or when the scope is usable. + public static string? Describe(string? raw) + { + var scope = (raw ?? "").Trim(); + if (scope.Length == 0) { return "Enter the edge node as 'group/edgeNode'."; } + + if (scope.IndexOfAny(Wildcards) >= 0) + { + return "An MQTT wildcard (+ or #) cannot address an edge node; name one exactly."; + } + + var segments = scope.Split('/'); + if (segments.Length != 2 || Array.Exists(segments, string.IsNullOrWhiteSpace)) + { + return "Use exactly 'group/edgeNode' — a group on its own would ask every edge node in it to " + + "republish, which this window cannot enumerate while the tree is empty."; + } + + return null; + } + + /// True when is a usable rebirth scope. + /// The operator-typed scope. + /// when finds no fault. + public static bool IsUsable(string? raw) => Describe(raw) is null; + + /// + /// Splits a scope already accepted by into its two segments. + /// + /// A scope for which is . + /// The trimmed (group, edgeNode) pair. + /// The scope is not usable. + public static (string GroupId, string EdgeNodeId) Split(string? raw) + { + if (Describe(raw) is { } error) + { + throw new ArgumentException($"'{raw}' is not a usable rebirth scope: {error}", nameof(raw)); + } + + var segments = raw!.Trim().Split('/'); + return (segments[0].Trim(), segments[1].Trim()); + } +} diff --git a/src/Server/ZB.MOM.WW.OtOpcUa.AdminUI/Components/Shared/Raw/RawBrowseModal.razor b/src/Server/ZB.MOM.WW.OtOpcUa.AdminUI/Components/Shared/Raw/RawBrowseModal.razor index 144a112f..1e57c8c3 100644 --- a/src/Server/ZB.MOM.WW.OtOpcUa.AdminUI/Components/Shared/Raw/RawBrowseModal.razor +++ b/src/Server/ZB.MOM.WW.OtOpcUa.AdminUI/Components/Shared/Raw/RawBrowseModal.razor @@ -126,9 +126,34 @@ @if (_rebirthTarget is null) { + @* Gitea #514: the tree is BIRTH-driven, and births are never retained. On a + quiet plant — every edge node birthed long before this window opened and is + now publishing only NDATA/DDATA — the tree is empty, so a tree-selected + scope can never be armed and the one action that would fill the tree is + unreachable. Naming an edge node directly breaks that cycle. Deliberately + an EDGE NODE rather than a group broadcast: a group rebirth is a + self-inflicted broker storm on a large plant. *@
- Click a group, edge node, device or metric in the tree above to choose the scope. + Click a group, edge node, device or metric in the tree above to choose the + scope — or, if the tree is empty because every edge node birthed + before this window opened, name one directly:
+
+ Edge node + + +
+ @if (!string.IsNullOrWhiteSpace(_manualScope) && ManualScopeError is { } manualErr) + { +
@manualErr
+ } } else { @@ -179,6 +204,17 @@ } } + else if (_rebirthExpired && _canOperate) + { + @* Gitea #510: the affordance is retired rather than left to fail on click. It is + drawn from a capability probed at browse-open, and sessions are reaped for + idleness — so a modal left open past the reap used to offer an action whose + token no longer resolved. Say what happened instead of silently vanishing. *@ +
+ This browse session expired while idle, so a rebirth can no longer be requested + through it. Close and re-open the browser to request one. +
+ }
@_selected.Count tag@(_selected.Count == 1 ? "" : "s") selected. @@ -276,6 +312,18 @@ private string? _rebirthOutcome; private string? _rebirthError; + /// + /// Set when a re-probe finds the session's rebirth capability gone (#510), so the retired + /// affordance can be explained rather than silently disappearing. + /// + private bool _rebirthExpired; + + /// + /// Operator-typed {group}/{edgeNode} rebirth scope, for the birth-driven-tree deadlock + /// (#514). Seeded with the driver's configured group so only the edge node has to be typed. + /// + private string _manualScope = ""; + private readonly Dictionary _selected = new(StringComparer.Ordinal); private readonly HashSet _selectedIds = new(StringComparer.Ordinal); @@ -316,6 +364,8 @@ _disabledReason = null; ResetRebirthState(); _canRebirth = false; + _rebirthExpired = false; + _manualScope = ""; _resolving = true; StateHasChanged(); @@ -373,8 +423,16 @@ { _token = result.Token; // Capability, not authorization: a plain-MQTT window (and every non-MQTT browser) has no - // re-announce action at all, so the affordance must not be drawn for it. + // re-announce action at all, so the affordance must not be drawn for it. Re-probed per + // click as well — see RebirthStillAvailable (#510). _canRebirth = BrowserService.CanRequestRebirth(_token); + _rebirthExpired = false; + // Only the edge node should have to be typed (#514): the group is the driver's entire + // subscription scope, so it is already in the config this browser was opened with. + if (_canRebirth && _manualScope.Length == 0 && TryReadConfiguredGroupId() is { } gid) + { + _manualScope = gid + "/"; + } } else { _openError = result.Message; } } @@ -502,9 +560,64 @@ { _rebirthOutcome = null; _rebirthError = null; + // #510 — the capability was probed once at browse-open; re-probe before offering to act on it. + if (!RebirthStillAvailable()) { return; } _rebirthConfirming = true; } + /// + /// Re-probes the open session's rebirth capability, retiring the affordance when it has gone + /// (Gitea #510). + /// + /// when a rebirth can still be requested through this session. + /// + /// CanRequestRebirth is a cheap, non-throwing, no-I/O probe that is false for a reaped or + /// unknown token, so re-running it per click costs nothing. It is a capability probe, not + /// an authorization check — the DriverOperator gate stays server-side on + /// RequestRebirthAsync and fails closed there, and _canOperate remains the + /// defence-in-depth half of the render condition. + /// + private bool RebirthStillAvailable() + { + if (BrowserService.CanRequestRebirth(_token)) { return true; } + + _canRebirth = false; + _rebirthExpired = true; + ResetRebirthState(); + return false; + } + + /// + /// Describes why is not a usable edge-node scope, or null when + /// it is. Mirrors the session's own {group}/{edgeNode} rule (Gitea #514). + /// + /// + /// Advisory only: MqttBrowseSession.ResolveRebirthTargets re-validates and refuses an + /// unresolvable scope before publishing anything, so this exists to keep the operator from + /// arming a confirm step that cannot succeed — not as the gate. A group-only scope is rejected on + /// purpose: the session would fan out to every observed edge node, and on a tree that is empty + /// (the case this input exists for) there are none to fan to. + /// + private string? ManualScopeError => SparkplugRebirthScope.Describe(_manualScope); + + /// Adopts the typed {group}/{edgeNode} as the rebirth scope (Gitea #514). + private void ApplyManualScope() + { + if (!RebirthStillAvailable()) { return; } + if (!SparkplugRebirthScope.IsUsable(_manualScope)) { return; } + + var (groupId, edgeNodeId) = SparkplugRebirthScope.Split(_manualScope); + _rebirthTarget = new RebirthTarget( + $"{groupId}/{edgeNodeId}", + $"edge node {edgeNodeId}", + $"Edge node '{edgeNodeId}' in group '{groupId}' republishes its whole metric set. " + + "This node was named directly, so it need not have appeared in the tree.", + IsGroup: false); + _rebirthConfirming = false; + _rebirthOutcome = null; + _rebirthError = null; + } + /// Disarms the confirm step without publishing anything. private void CancelRebirth() => _rebirthConfirming = false; @@ -516,6 +629,9 @@ private async Task ConfirmRebirthAsync() { if (_rebirthTarget is null) { return; } + // #510 again, at the last moment before publishing: the reap can land between arming the confirm + // step and clicking it, and this is the one path that reaches a live broker. + if (!RebirthStillAvailable()) { return; } _rebirthBusy = true; _rebirthOutcome = null; _rebirthError = null; @@ -558,6 +674,34 @@ ResetRebirthState(); } + /// + /// Reads the driver's configured Sparkplug group id out of the merged config, or null when + /// it is absent or the blob is unreadable. Purely a convenience seed for + /// — never a gate, so any failure is simply "no seed". + /// + /// The configured group id, or null. + private string? TryReadConfiguredGroupId() + { + try + { + // Case-insensitive: the merged blob is assembled from the driver + device configs and this + // is a display convenience, not a binding — it must not depend on which one won. + var root = System.Text.Json.Nodes.JsonNode.Parse(_mergedConfig)?.AsObject(); + var sparkplug = root? + .FirstOrDefault(p => string.Equals(p.Key, "Sparkplug", StringComparison.OrdinalIgnoreCase)) + .Value?.AsObject(); + var groupId = sparkplug? + .FirstOrDefault(p => string.Equals(p.Key, "GroupId", StringComparison.OrdinalIgnoreCase)) + .Value?.GetValue(); + + return string.IsNullOrWhiteSpace(groupId) ? null : groupId.Trim(); + } + catch + { + return null; + } + } + /// Clears every rebirth-panel field (modal re-open, browser close). private void ResetRebirthState() { @@ -586,12 +730,22 @@ .ToList(); if (_commitErrors.Count > 0) { return; } - var rows = _selected.Values + var selected = _selected.Values.ToList(); + var rows = selected .Select(s => RawBrowseCommitMapper.MapLeaf( _driverType, s.NodeId, s.Name, s.DriverDataType, DefaultDataType, _groupPrefix, s.FolderPath, _createGroups, s.AddressFields)) .ToList(); + // A leaf name is coerced into a legal RawPath segment (#509), and that coercion is not + // injective — 'A/B' and 'A_B' both land on 'A_B'. Refuse the colliding SET by name here: + // letting it through would silently merge two distinct metrics into one tag bound to + // whichever address survived, which is worse than the outright refusal #509 replaced. + _commitErrors = RawBrowseCommitMapper + .DescribeNameCollisions(selected.Select(s => s.Name).ToList(), rows) + .ToList(); + if (_commitErrors.Count > 0) { return; } + var outcome = await Svc.ImportTagsAsync(_effectiveDeviceId, rows); if (outcome.Errors.Count > 0) { @@ -613,6 +767,8 @@ var t = _token; _token = Guid.Empty; _canRebirth = false; + _rebirthExpired = false; + _manualScope = ""; ResetRebirthState(); Visible = false; _open = false; diff --git a/src/Server/ZB.MOM.WW.OtOpcUa.AdminUI/Components/Shared/Uns/TagEditors/MqttTagConfigEditor.razor b/src/Server/ZB.MOM.WW.OtOpcUa.AdminUI/Components/Shared/Uns/TagEditors/MqttTagConfigEditor.razor index 456c1626..39ecb1e0 100644 --- a/src/Server/ZB.MOM.WW.OtOpcUa.AdminUI/Components/Shared/Uns/TagEditors/MqttTagConfigEditor.razor +++ b/src/Server/ZB.MOM.WW.OtOpcUa.AdminUI/Components/Shared/Uns/TagEditors/MqttTagConfigEditor.razor @@ -30,6 +30,18 @@ @onchange="@(e => Update(() => _m.Topic = e.Value?.ToString() ?? ""))" />
Concrete topic — MQTT wildcards (+ / #) are not valid for a tag.
+
+ + +
+ Where a write is published. Deliberately separate from the read topic — a tag commonly + reads …/state and writes …/set. Leave blank for a + read-only tag: a write is then refused (BadNotWritable) rather than published + nowhere. +
+