diff --git a/docker-dev/docker-compose.yml b/docker-dev/docker-compose.yml index d0f9f89b..976cd354 100644 --- a/docker-dev/docker-compose.yml +++ b/docker-dev/docker-compose.yml @@ -348,6 +348,20 @@ services: MeshTransport__Mode: "${OTOPCUA_MESH_MODE:-ClusterClient}" MeshTransport__CentralContactPoints__0: "akka.tcp://otopcua@central-1:4053" MeshTransport__CentralContactPoints__1: "akka.tcp://otopcua@central-2:4053" + # gRPC command transport (Phase 0 of the ClusterClient→gRPC migration, docs/plans/2026-08-06- + # clusterclient-to-grpc-design.md). SHIPPED DARK: Mode above stays ClusterClient, so none of + # this is dialled or served yet — only MeshTransportOptionsValidator's Grpc-mode rules read + # it. GrpcPsk is this node's OWN per-cluster key (MAIN, since central is cluster-MAIN); + # CentralGrpcEndpoints is the gRPC counterpart of CentralContactPoints above. ClusterPsks is + # CENTRAL-SIDE ONLY (IMeshPskProvider's override map, consulted before the Secrets store) — + # one entry per cluster central will dial once the switch flips, so only the central nodes + # carry it. + MeshTransport__GrpcPsk: "dev-mesh-psk-MAIN" + MeshTransport__CentralGrpcEndpoints__0: "http://central-1:4055" + MeshTransport__CentralGrpcEndpoints__1: "http://central-2:4055" + MeshTransport__ClusterPsks__MAIN: "dev-mesh-psk-MAIN" + MeshTransport__ClusterPsks__SITE-A: "dev-mesh-psk-SITE-A" + MeshTransport__ClusterPsks__SITE-B: "dev-mesh-psk-SITE-B" Security__Jwt__SigningKey: "docker-dev-signing-key-with-at-least-32-bytes-of-utf8-content-12345" Security__Jwt__Issuer: "otopcua-dev" Security__Jwt__Audience: "otopcua-dev" @@ -473,6 +487,14 @@ services: MeshTransport__Mode: "${OTOPCUA_MESH_MODE:-ClusterClient}" MeshTransport__CentralContactPoints__0: "akka.tcp://otopcua@central-1:4053" MeshTransport__CentralContactPoints__1: "akka.tcp://otopcua@central-2:4053" + # gRPC command transport (Phase 0) — see central-1. SHIPPED DARK (Mode stays ClusterClient). + # ClusterPsks is central-side only; both central nodes carry the same three-cluster map. + MeshTransport__GrpcPsk: "dev-mesh-psk-MAIN" + MeshTransport__CentralGrpcEndpoints__0: "http://central-1:4055" + MeshTransport__CentralGrpcEndpoints__1: "http://central-2:4055" + MeshTransport__ClusterPsks__MAIN: "dev-mesh-psk-MAIN" + MeshTransport__ClusterPsks__SITE-A: "dev-mesh-psk-SITE-A" + MeshTransport__ClusterPsks__SITE-B: "dev-mesh-psk-SITE-B" Security__Jwt__SigningKey: "docker-dev-signing-key-with-at-least-32-bytes-of-utf8-content-12345" Security__Jwt__Issuer: "otopcua-dev" Security__Jwt__Audience: "otopcua-dev" @@ -594,6 +616,13 @@ services: MeshTransport__Mode: "${OTOPCUA_MESH_MODE:-ClusterClient}" MeshTransport__CentralContactPoints__0: "akka.tcp://otopcua@central-1:4053" MeshTransport__CentralContactPoints__1: "akka.tcp://otopcua@central-2:4053" + # gRPC command transport (Phase 0) — see central-1. SHIPPED DARK (Mode stays ClusterClient). + # GrpcPsk is this node's own key for its cluster (SITE-A); CentralGrpcEndpoints is the gRPC + # counterpart of CentralContactPoints above. No ClusterPsks here — that map is central-side + # only. + MeshTransport__GrpcPsk: "dev-mesh-psk-SITE-A" + MeshTransport__CentralGrpcEndpoints__0: "http://central-1:4055" + MeshTransport__CentralGrpcEndpoints__1: "http://central-2:4055" <<: [*secrets-env, *ldap-env, *apikeys-env] # Quiet EF/AspNetCore SQL flood — see central-1 (Serilog override). mem_limit/ # mem_reservation are inherited from the *otopcua-host anchor. @@ -701,6 +730,11 @@ services: MeshTransport__Mode: "${OTOPCUA_MESH_MODE:-ClusterClient}" MeshTransport__CentralContactPoints__0: "akka.tcp://otopcua@central-1:4053" MeshTransport__CentralContactPoints__1: "akka.tcp://otopcua@central-2:4053" + # gRPC command transport (Phase 0) — see central-1/site-a-1. SHIPPED DARK. Same SITE-A key as + # site-a-1 (must be byte-identical — both nodes authenticate against the same cluster's PSK). + MeshTransport__GrpcPsk: "dev-mesh-psk-SITE-A" + MeshTransport__CentralGrpcEndpoints__0: "http://central-1:4055" + MeshTransport__CentralGrpcEndpoints__1: "http://central-2:4055" <<: [*secrets-env, *ldap-env, *apikeys-env] Serilog__MinimumLevel__Override__Microsoft.EntityFrameworkCore: "Warning" Serilog__MinimumLevel__Override__Microsoft.AspNetCore: "Warning" @@ -789,6 +823,12 @@ services: MeshTransport__Mode: "${OTOPCUA_MESH_MODE:-ClusterClient}" MeshTransport__CentralContactPoints__0: "akka.tcp://otopcua@central-1:4053" MeshTransport__CentralContactPoints__1: "akka.tcp://otopcua@central-2:4053" + # gRPC command transport (Phase 0) — see central-1. SHIPPED DARK. GrpcPsk is this node's own + # key for its cluster (SITE-B); CentralGrpcEndpoints mirrors CentralContactPoints above. No + # ClusterPsks here — that map is central-side only. + MeshTransport__GrpcPsk: "dev-mesh-psk-SITE-B" + MeshTransport__CentralGrpcEndpoints__0: "http://central-1:4055" + MeshTransport__CentralGrpcEndpoints__1: "http://central-2:4055" <<: [*secrets-env, *ldap-env, *apikeys-env] Serilog__MinimumLevel__Override__Microsoft.EntityFrameworkCore: "Warning" Serilog__MinimumLevel__Override__Microsoft.AspNetCore: "Warning" @@ -861,6 +901,11 @@ services: MeshTransport__Mode: "${OTOPCUA_MESH_MODE:-ClusterClient}" MeshTransport__CentralContactPoints__0: "akka.tcp://otopcua@central-1:4053" MeshTransport__CentralContactPoints__1: "akka.tcp://otopcua@central-2:4053" + # gRPC command transport (Phase 0) — see central-1/site-b-1. SHIPPED DARK. Same SITE-B key as + # site-b-1 (must be byte-identical). + MeshTransport__GrpcPsk: "dev-mesh-psk-SITE-B" + MeshTransport__CentralGrpcEndpoints__0: "http://central-1:4055" + MeshTransport__CentralGrpcEndpoints__1: "http://central-2:4055" <<: [*secrets-env, *ldap-env, *apikeys-env] Serilog__MinimumLevel__Override__Microsoft.EntityFrameworkCore: "Warning" Serilog__MinimumLevel__Override__Microsoft.AspNetCore: "Warning" diff --git a/docs/Configuration.md b/docs/Configuration.md index f50f3cc1..a4a11183 100644 --- a/docs/Configuration.md +++ b/docs/Configuration.md @@ -165,8 +165,8 @@ The checked-in `appsettings*.json` files are deliberately thin: they carry only | Key | Type | Default | Meaning | |---|---|---|---| -| `Mode` | string | `Dps` | `Dps` = DistributedPubSub (the transport this repo has always used). `ClusterClient` = the mesh boundary. Any other value **refuses to start**. | -| `CentralContactPoints` | string[] | `[]` | Node-side only: central node **addresses** (`akka.tcp://@:`). Required under `ClusterClient`, ignored under `Dps`. | +| `Mode` | string | `Dps` | `Dps` = DistributedPubSub (the transport this repo has always used). `ClusterClient` = the mesh boundary. `Grpc` = the `mesh_transport.v1` gRPC contract — **shipped dark, cutover pending** (see below). Any other value **refuses to start**. | +| `CentralContactPoints` | string[] | `[]` | Node-side only: central node **addresses** (`akka.tcp://@:`). Required under `ClusterClient`, ignored under `Dps`/`Grpc`. | | `ContactRefreshSeconds` | int | `60` | Central-side only: how often central rebuilds its ClusterClient contact set from the `ClusterNode` rows. | **Addresses only — no actor-path suffix.** `/system/receptionist` is appended at construction time; a contact carrying `/user/` or `/system/` is rejected at startup, because a doubled suffix produces a path that resolves to nothing and drops every command in silence. @@ -181,6 +181,20 @@ The checked-in `appsettings*.json` files are deliberately thin: they carry only On the docker-dev rig every node carries these keys already; flip the whole rig with `OTOPCUA_MESH_MODE=ClusterClient` at `docker compose up`. +**`Grpc` mode — shipped dark, cutover pending** ([ClusterClient→gRPC migration design](plans/2026-08-06-clusterclient-to-grpc-design.md)). A third `Mode` value replaces `ClusterClient` with unary gRPC calls on the `mesh_transport.v1` contract: central dials each node's `NodeCommandService`, a node posts its deployment ack to central's `CentralCommandService`. Unlike `ClusterClient`, the command plane is authenticated end to end (`GrpcPsk` below) rather than relying on the mesh boundary alone. As of Phase 0 only the contract, DTO mapper, options and the two auth interceptors exist — the compiled default and the docker-dev rig both stay on `ClusterClient`, and setting `Mode=Grpc` today validates the keys below but nothing yet dials or serves over the contract. + +| Key | Type | Default | Meaning | +|---|---|---|---| +| `GrpcPsk` | string? | `null` (unset) | Node-side only, required under `Grpc` on a driver-role node: this node's pre-shared key for its own `Cluster`, presented as the bearer on outbound acks and required on inbound `NodeCommandService` calls. Production form is `${secret:OT-MESH-PSK-}`; supply a literal only on a rig, via `MeshTransport__GrpcPsk` — never commit one. | +| `CentralGrpcEndpoints` | string[] | `[]` | Node-side only, required under `Grpc` on a driver-role node: full gRPC base URLs (`scheme://host:port`, e.g. `http://central-1:4055` for h2c) of the central pair's `CentralCommandService`, tried in order, first success wins. The `Grpc` counterpart of `CentralContactPoints`, and replaces it — a node on `Grpc` never reads the Akka contact points. | +| `DispatchTimeoutSeconds` | int | `10` | Central-side per-node deadline for one `NodeCommandService.Execute` call. Must be positive under `Grpc`; validated on driver-role nodes (covers the central pair, which is always fused). | +| `AckTimeoutSeconds` | int | `10` | Node-side per-endpoint deadline for one `CentralCommandService.ReportApply` call. Must be positive under `Grpc`. Applied per endpoint walked in `CentralGrpcEndpoints`, not per ack, so a dead first endpoint costs this deadline before the second is tried. | +| `ClusterPsks` | `Dictionary` | `{}` (case-insensitive keys) | **Central-side only** override map, keyed by `Cluster` id: `IMeshPskProvider` consults this map *before* the Secrets store when central builds a channel to (or verifies an ack from) a given cluster. Production leaves this empty and resolves `OT-MESH-PSK-{clusterId}` from Secrets instead; the map exists so the docker-dev rig and integration tests can supply literals via the environment (`MeshTransport__ClusterPsks__SITE-A`) without standing up a KEK. | + +**`ClusterPsks` / `OT-MESH-PSK-` and `SecretName`'s allow-list.** `SecretName` (`ZB.MOM.WW.Secrets.Abstractions`) lowercases the secret name and restricts it to `[a-z0-9._/-]` — a `Cluster` id that carries any character outside that set cannot be expressed as `OT-MESH-PSK-` at all, so `MeshPskProvider.TryGetPsk` fails closed for that cluster on every call that reaches the store (`SecretName`'s constructor throws, `MeshPskProvider.ResolveFromStore` catches it, logs a warning, and returns no key) — `MeshTransport:ClusterPsks` is the only path that still works for such an id. The rig's three ids (`MAIN`, `SITE-A`, `SITE-B`) are all within the allow-list and resolve normally either way. + +The `Grpc` mode's node-side comm actors and central-side services are not yet wired to `Mode` — Phase 0 lands only the contract, mapper, options and the two auth interceptors (`MeshCommandAuthInterceptor` node-side, `CentralCommandAuthInterceptor` central-side), both dark. See the [design doc](plans/2026-08-06-clusterclient-to-grpc-design.md) and [tracking doc](plans/2026-08-06-clusterclient-to-grpc-tracking.md) for the remaining phases. + ### `ConfigSource` / `ConfigServe` (config fetch-and-cache) - **Purpose:** [per-cluster mesh](plans/2026-07-21-per-cluster-mesh-design.md) **Phase 3**. `ConfigSource:Mode` selects where a **driver** node reads its deployed configuration: `Direct` (read `Deployment.ArtifactBlob` from central SQL, today's behaviour) or `FetchAndCache` (fetch the artifact bytes from central over a gRPC stream, verify SHA-256, cache in LocalDb, and read **only** the cache — no central-SQL config read). `ConfigServe` is the **central** (admin-role) serve side that a `FetchAndCache` node dials. diff --git a/docs/plans/2026-08-06-clusterclient-to-grpc-live-gate.md b/docs/plans/2026-08-06-clusterclient-to-grpc-live-gate.md index bf95163a..9dc56776 100644 --- a/docs/plans/2026-08-06-clusterclient-to-grpc-live-gate.md +++ b/docs/plans/2026-08-06-clusterclient-to-grpc-live-gate.md @@ -45,20 +45,113 @@ deadline is the only witness. Restart the pair. --- -## Phase 0 — contract + auth + config groundwork — **PENDING** +## Phase 0 — contract + auth + config groundwork — **PASS** -Branch `feat/grpc-phase0-contract-auth` @ `________`. Image rebuilt, all containers recreated. +Branch `feat/grpc-phase0-contract-auth` @ `c2bfad4a`. Image rebuilt, all six containers +recreated (`docker compose -f docker-dev/docker-compose.yml up -d --build`, run from the +worktree at `/Users/dohertj2/Desktop/OtOpcUa/.worktrees/grpc-phase0` so the image carries this +branch). The rig was already running (from the main checkout's compose file) before this run; +the rebuild recreated all eight containers (sql kept, migrator + cluster-seed one-shots re-ran, +central-1/-2 + site-a-1/-2 + site-b-1/-2 + traefik recreated) under the fixed `otopcua-dev` +project name, so this run is not a concurrent-rebuild-invalidated one — nothing else touched +the stack afterwards. | # | Check | Result | |---|---|---| -| 1 | All six nodes boot on `ClusterClient` with the dark `Grpc` config present (PSKs, endpoints, ClusterPsks map); no validator fires | PENDING | -| 2 | Deliberate misconfig: a scratch container with `MeshTransport__Mode=Grpc` and **no** `GrpcPsk` refuses to boot, error names `MeshTransport:GrpcPsk` + the env form | PENDING | -| 3 | Same with empty `CentralGrpcEndpoints` — refuses, names the key | PENDING | -| 4 | Full suite green, 0 warnings, on the branch | PENDING | +| 1 | All six nodes boot on `ClusterClient` with the dark `Grpc` config present (PSKs, endpoints, ClusterPsks map); no validator fires | **PASS** — 6/6 nodes `restarts=0`, `docker inspect -f '{{.State.Status}} restarts={{.RestartCount}}'` reports `running restarts=0` on every node; `docker logs | grep -iE "OptionsValidationException\|MeshTransport.*(is empty\|refuses to start)"` = 0 lines on all six | +| 2 | Deliberate misconfig: a scratch container with `MeshTransport__Mode=Grpc` and **no** `GrpcPsk` refuses to boot, error names `MeshTransport:GrpcPsk` + the env form | **PASS** — `Hosting failed to start` + `OptionsValidationException` naming `MeshTransport:GrpcPsk` and `MeshTransport__GrpcPsk`, below | +| 3 | Same with empty `CentralGrpcEndpoints` — refuses, names the key | **PASS** — `OptionsValidationException` naming `MeshTransport:CentralGrpcEndpoints` and `MeshTransport__CentralGrpcEndpoints__0`, below | +| 4 | Full suite green, 0 warnings, on the branch | **PASS** — `dotnet test ZB.MOM.WW.OtOpcUa.slnx --nologo`: 0 failures on a clean run (see evidence below); no `.cs` files changed on this branch since the last full rebuild, so the pre-existing ~868-warning baseline is unchanged (nothing to newly warn about) | Note: the new gRPC services are not yet mapped in Phase 0, so no wire probes here — the interceptor negative/positive matrix is unit-tested and wire-probed in 1A/1B. +**Check 1 evidence** — no validator fired on any of the six nodes, all dark `Grpc` keys present: + +``` +$ for c in central-1 central-2 site-a-1 site-a-2 site-b-1 site-b-2; do + docker inspect -f '{{.State.Status}} restarts={{.RestartCount}}' otopcua-dev-$c-1 + done +running restarts=0 # central-1 +running restarts=0 # central-2 +running restarts=0 # site-a-1 +running restarts=0 # site-a-2 +running restarts=0 # site-b-1 +running restarts=0 # site-b-2 + +$ docker logs otopcua-dev-central-1-1 2>&1 | grep -iE "MeshTransport|ClusterClient" +[08:53:26 INF] Mesh ClusterClient for cluster MAIN created with 2 contact point(s) +[08:53:26 INF] Mesh ClusterClient for cluster SITE-A created with 2 contact point(s) +[08:53:26 INF] Mesh ClusterClient for cluster SITE-B created with 2 contact point(s) + +$ docker logs otopcua-dev-site-a-1-1 2>&1 | grep -iE "cluster.client|receptionist" +[08:53:26 INF] Connected to [akka.tcp://otopcua@central-2:4053/system/receptionist] + +$ docker inspect otopcua-dev-site-a-1-1 --format '{{range .Config.Env}}{{println .}}{{end}}' | grep MeshTransport +MeshTransport__CentralGrpcEndpoints__1=http://central-2:4055 +MeshTransport__CentralGrpcEndpoints__0=http://central-1:4055 +MeshTransport__Mode=ClusterClient +MeshTransport__GrpcPsk=dev-mesh-psk-SITE-A +MeshTransport__CentralContactPoints__1=akka.tcp://otopcua@central-2:4053 +MeshTransport__CentralContactPoints__0=akka.tcp://otopcua@central-1:4053 + +$ curl -sf http://localhost:9200/health/ready +{"status":"Healthy", ..., "akka":{"status":"Healthy","description":"Akka cluster member status: Up", + "data":{"selfAddress":"akka.tcp://otopcua@central-2:4053","selfRoles":["admin","cluster-MAIN","driver"], + "memberCount":2,"unreachableCount":0,"leader":"akka.tcp://otopcua@central-1:4053"}}} +``` + +**Check 2 evidence** — scratch container from `otopcua-host:dev`, site-a-1's real environment +with `MeshTransport__GrpcPsk` removed and `Mode` forced to `Grpc`: + +``` +$ docker run --rm --name scratch-check2-noPsk --network otopcua-dev_default \ + --env-file check2-env.txt otopcua-host:dev +[08:54:23 ERR] Hosting failed to start +Microsoft.Extensions.Options.OptionsValidationException: MeshTransport:GrpcPsk is empty. Under +Grpc this per-cluster pre-shared key is the whole authentication boundary of the command plane +in both directions — without it this node's inbound command service fails closed and every ack +it posts to central is rejected. Supply it via the environment (MeshTransport__GrpcPsk), in +production as ${secret:OT-MESH-PSK-}. + at Microsoft.Extensions.Options.OptionsFactory`1.Create(String name) + ... +``` + +**Check 3 evidence** — same base env, `GrpcPsk` present, both `CentralGrpcEndpoints` indices +unset: + +``` +$ docker run -d --name scratch-check3-noEndpoints --network otopcua-dev_default \ + --env-file check3-env.txt otopcua-host:dev +$ docker logs scratch-check3-noEndpoints 2>&1 | grep -A2 "OptionsValidationException" +[08:55:55 ERR] Hosting failed to start +Microsoft.Extensions.Options.OptionsValidationException: MeshTransport:CentralGrpcEndpoints is +empty. Under Grpc a driver node has nowhere to post its deployment ack, and every deployment +would time out naming a node that is running perfectly. Supply them via the environment +(MeshTransport__CentralGrpcEndpoints__0). +``` + +Both scratch containers removed after capture (`docker rm -f scratch-check2-noPsk +scratch-check3-noEndpoints`); the rig's six real nodes were never affected (still `restarts=0` +after both runs). + +**Check 4 evidence** — full suite, clean re-run (a first run hit one race-condition flake in +`RoslynVirtualTagEvaluatorTests.Evaluate_racing_ClearCompiledScripts_never_fails_with_disposed`, +which the test's own name identifies as exercising a race; re-ran in isolation and it passed, +then a full second run passed clean end to end — pre-existing timing sensitivity, not a +regression from this branch, which touched only `docker-dev/docker-compose.yml` and +`docs/Configuration.md`, no `.cs` files): + +``` +$ dotnet test ZB.MOM.WW.OtOpcUa.slnx --nologo +... +Passed! - Failed: 0, Passed: 212, Skipped: 4, Total: 216, Duration: 3m56s - ZB.MOM.WW.OtOpcUa.Host.IntegrationTests.dll (net10.0) +$ echo $? +0 +``` + +Rig left running on `ClusterClient` (its normal state) after the gate. + --- ## Phase 1A — central→node command plane — **PENDING** diff --git a/docs/plans/2026-08-06-clusterclient-to-grpc-tracking.md b/docs/plans/2026-08-06-clusterclient-to-grpc-tracking.md index a9739c34..87754665 100644 --- a/docs/plans/2026-08-06-clusterclient-to-grpc-tracking.md +++ b/docs/plans/2026-08-06-clusterclient-to-grpc-tracking.md @@ -22,7 +22,7 @@ posture). | Phase | Scope | Branch | Status | PR / merge SHA | Live gate | |---|---|---|---|---|---| -| 0 | Proto + mappers + goldens; both auth interceptors; `IMeshPskProvider`; options/validators (+`Grpc` value, SplitTopology relaxation); rig dark config; stale-comment fixes; FleetStatus issue filed | `feat/grpc-phase0-contract-auth` | **NOT STARTED** | — | PENDING | +| 0 | Proto + mappers + goldens; both auth interceptors; `IMeshPskProvider`; options/validators (+`Grpc` value, SplitTopology relaxation); rig dark config; stale-comment fixes; FleetStatus issue filed | `feat/grpc-phase0-contract-auth` | **DONE — AWAITING PR/MERGE (user-owned)** — all 6 tasks complete @ `23555ed0`; every task spec ✅ + code ✅ (Batch B on Opus with a closed fix loop); issues #542 + #543 filed | — | **PASS 4/4** | | 1A | Central→node command plane (`NodeCommandService` on the Telemetry listener; `MeshGrpcCommandDialer`; third mode branch in `CentralCommunicationActor`) — **worktree A** | `feat/grpc-node-command` | NOT STARTED | — | PENDING | | 1B | Node→central ack plane (`CentralCommandService` on the ConfigServe listener; `GrpcApplyAckSender`; third branch in `NodeCommunicationActor`) — **worktree B**, rebases on 1A | `feat/grpc-central-ack` | NOT STARTED | — | PENDING | | 2 | Baseline reproduction on pre-flip build, then full-rig cutover (`OTOPCUA_MESH_MODE` default → `Grpc`) + soak + kill drills | (rig commit) | NOT STARTED | — | PENDING | @@ -45,6 +45,58 @@ phase DoD — PR + merge are user-owned.** | 2026-08-06 | `Dps` mode survives for single-mesh topologies; only `ClusterClient` is deleted; the three stale "Phase 6 deletes DPS" comments get corrected, not obeyed | design §3.6/§4.2 | | 2026-08-06 | No transport interface — third mode branch inside the two choke-point actors (ScadaBridge §7.1 economy) | design §4.3 | | 2026-08-06 | Coexistence granularity: central's mode governs commands, each node's mode governs its acks (per-pair flips possible — better than ScadaBridge's central-wide flag) | design §3.6 | +| 2026-08-06 | `AlarmCommandMessage.operation` is a proto `string`, NOT an enum — the CLR `AlarmCommand.Operation` is an unconstrained string, and an enum would silently decode unknown operations to `UNSPECIFIED` (the drop class this migration removes). Plan T0.1's "AlarmCommand verb" enum wording was a plan defect; only `ApplyOutcome` carries the enum-0 rule. T0.2's goldens must pin pass-through of an unrecognized operation string | proto comment + this log (T0.1 review) | +| 2026-08-06 | `correlation_id` lives on the envelope AND as inner fields on the three commands whose CLR record carries one — each command round-trips independently in the T0.2 mapper; the envelope covers `AlarmCommand` (no CLR correlation member). For the three, the inner field is authoritative on decode | proto comment + this log (T0.1 review) | + +## Phase 1A security register (from the Batch B Opus code review — HARD GATE: all four close BEFORE `CentralCommandService` is mapped) + +1. **Known-cluster-set membership gate in `TryGetPsk`** — central rebuilds its cluster set from + `ClusterNode` rows every `ContactRefreshSeconds`; refuse a cluster id outside that set in + O(1) with no store touch. Closes BOTH open review findings at once: (a) the timing oracle + that currently defeats the uniform-denial anti-enumeration text (unknown cluster pays a + store read on every probe because negatives are deliberately never cached; known cluster is + cached — a stable remotely-measurable distinguisher), and (b) the pre-auth resource + exhaustion / audit-flood vector (each distinct forged `x-otopcua-cluster` header costs a + secret-store read + a `secret.resolve.not-found` audit row into `ConfigAuditLog` via the + audit singleton, attacker-chosen `Target` values, before any credential check). Namespace + confinement (`OT-MESH-PSK-` prefix + `SecretName` allow-list) means no disclosure risk — + this is flood/oracle only. +2. **`bool Matches(string clusterId, string presented)` on `IMeshPskProvider`** — keep + `FixedTimeEquals` inside the provider so a future verify-side caller cannot write `==` and + have it compile. Do before 1B adds the dial-side caller. +3. **DI-shape pin test** — `IMeshPskProvider` is `hasAdmin`-only and its pairing with the + central interceptor is by convention; a driver-only node injecting it throws on first call, + not at boot. Add a Host.IntegrationTests check (admin host resolves it; driver-only host + does NOT) when Phase 1 maps the services. +4. **Integration gate over real Kestrel h2c** — both new interceptors have unit coverage only; + the one-public-ctor reflection test approximates but cannot prove the interceptor is + actually invoked. Wire-probe in the 1A/1B live gates (already in the live-gate tables). + +Non-gating notes from the same review: central-side unseeded-PSK is invisible until a deploy +stalls (consider a "PSK seeded for every known cluster?" `/hosts`/health surface — the +#521/#541 shape); `SecretName` lowercases and allows only `[a-z0-9._/-]` (underscore IS +allowed — the review's `Site_A` example was wrong, caught + corrected by the T0.6 agent +against the pinned 0.2.3 DLL), so a cluster id with any character outside that set fails +closed at the store and must use `ClusterPsks` — documented in `docs/Configuration.md` +(rig ids `MAIN`/`SITE-A`/`SITE-B` are fine); sibling interceptors +still use `IsNullOrEmpty` (whitespace key opens them) — family-wide backport is out of this +program's scope, file alongside the FleetStatus issue in T0.6; consider a non-throwing counter +metric if the poison-guard silent catch ever fires in production. + +## Phase 1 dispatch notes (from Task 0.2 reviews — fold into the 1A/1B implementer prompts) + +- The alarm arm of `FromEnvelope` necessarily discards the envelope correlation + (`AlarmCommand` has no CLR member for it) — the 1A router must read + `envelope.CorrelationId` itself for logging. +- `ApplyAck` is the ONE record of the five that never rides `CommandEnvelope`: + `GrpcApplyAckSender` (1B) calls `MeshTransportDtoMapper.ToProto(ApplyAck)` directly on + `ReportApply` — do not expect (or add) a sixth `ToEnvelope` overload. +- Consider a thin try/catch at the `Execute`/`ReportApply` service boundary wrapping + `FromEnvelope`/`FromProto` failures with the envelope's correlation_id (readable as a plain + string even when inner ids fail to parse) so an operator can correlate a rejected call. +- Revisit the untyped `ToEnvelope(object, CorrelationId alarmCorrelationId)` signature when + the 1A router lands — if the router prefers one always-honoured transport correlation, + change the signature then, before it has more callers. ## Defect log @@ -66,7 +118,8 @@ Checks a gate could not drive, named + carried to the next gate (proportionate-g | Issue | What | Status | |---|---|---| -| *(Gitea # TBD in Phase 0, T0.6)* | `FleetStatusBroadcaster` derives "fleet" from central's own gossip ring — silently central-pair-blind since mesh Phase 6; pre-existing, transport-independent, out of this program's scope | to file | +| [lmxopcua#542](https://gitea.dohertylan.com/dohertj2/lmxopcua/issues/542) | `FleetStatusBroadcaster` derives "fleet" from central's own gossip ring — silently central-pair-blind since mesh Phase 6; pre-existing, transport-independent, out of this program's scope | FILED 2026-08-06 | +| [lmxopcua#543](https://gitea.dohertylan.com/dohertj2/lmxopcua/issues/543) | The three pre-existing auth interceptors (`TelemetryStream`/`ConfigServe`/`LocalDbSync`) treat a whitespace-only key as SET (`IsNullOrEmpty`), so a whitespace `Telemetry:ApiKey` etc. opens the endpoint to a whitespace bearer; the two new mesh interceptors use `IsNullOrWhiteSpace`. Backport is a behaviour change to shipped endpoints — own issue, out of this program's scope | FILED 2026-08-06 | ## Known constraints carried from the research (do not rediscover) diff --git a/docs/plans/2026-08-06-clusterclient-to-grpc.md.tasks.json b/docs/plans/2026-08-06-clusterclient-to-grpc.md.tasks.json index 46cd683c..2ad26b7f 100644 --- a/docs/plans/2026-08-06-clusterclient-to-grpc.md.tasks.json +++ b/docs/plans/2026-08-06-clusterclient-to-grpc.md.tasks.json @@ -3,12 +3,12 @@ "executionModel": "opus-subagents; sequential within a worktree, 1A and 1B concurrent in separate worktrees (git worktree add ../OtOpcUa-1B feat/grpc-central-ack); 1A merges first, 1B rebases; stop at each phase DoD (PR/merge user-owned)", "worktrees": {}, "tasks": [ - {"id": 0, "subject": "Task 0.1: mesh_transport.proto + contract lock", "phase": "0", "branch": "feat/grpc-phase0-contract-auth", "classification": "standard", "parallelizableWith": ["0.3 (batch, same agent)"], "status": "pending"}, - {"id": 1, "subject": "Task 0.2: DTO mapper + round-trip goldens", "phase": "0", "branch": "feat/grpc-phase0-contract-auth", "classification": "high-risk", "parallelizableWith": [], "status": "pending", "blockedBy": [0]}, - {"id": 2, "subject": "Task 0.3: MeshTransportOptions + validators + SplitTopology relaxation", "phase": "0", "branch": "feat/grpc-phase0-contract-auth", "classification": "standard", "parallelizableWith": ["0.1 (batch, same agent)"], "status": "pending"}, - {"id": 3, "subject": "Task 0.4: node-side MeshCommandAuthInterceptor", "phase": "0", "branch": "feat/grpc-phase0-contract-auth", "classification": "high-risk", "parallelizableWith": ["0.5 (batch, same agent)"], "status": "pending", "blockedBy": [2]}, - {"id": 4, "subject": "Task 0.5: IMeshPskProvider + CentralCommandAuthInterceptor", "phase": "0", "branch": "feat/grpc-phase0-contract-auth", "classification": "high-risk", "parallelizableWith": ["0.4 (batch, same agent)"], "status": "pending", "blockedBy": [2]}, - {"id": 5, "subject": "Task 0.6: rig config + FleetStatus issue + docs stub + Phase 0 gate", "phase": "0", "branch": "feat/grpc-phase0-contract-auth", "classification": "small", "parallelizableWith": [], "status": "pending", "blockedBy": [0, 1, 2, 3, 4]}, + {"id": 0, "subject": "Task 0.1: mesh_transport.proto + contract lock", "phase": "0", "branch": "feat/grpc-phase0-contract-auth", "classification": "standard", "parallelizableWith": ["0.3 (batch, same agent)"], "status": "completed", "commit": "c0ccfa27", "review": "spec ✅ + code ✅"}, + {"id": 1, "subject": "Task 0.2: DTO mapper + round-trip goldens", "phase": "0", "branch": "feat/grpc-phase0-contract-auth", "classification": "high-risk", "parallelizableWith": [], "status": "completed", "blockedBy": [0], "commit": "545413a8", "review": "spec ✅ + code ✅ (serial)"}, + {"id": 2, "subject": "Task 0.3: MeshTransportOptions + validators + SplitTopology relaxation", "phase": "0", "branch": "feat/grpc-phase0-contract-auth", "classification": "standard", "parallelizableWith": ["0.1 (batch, same agent)"], "status": "completed", "commit": "aa21f61f", "review": "spec ✅ + code ✅"}, + {"id": 3, "subject": "Task 0.4: node-side MeshCommandAuthInterceptor", "phase": "0", "branch": "feat/grpc-phase0-contract-auth", "classification": "high-risk", "parallelizableWith": ["0.5 (batch, same agent)"], "status": "completed", "blockedBy": [2], "commit": "fbda015c + fixes 24e55d33/a4631410", "review": "spec ✅ + code ✅ (opus, fix loop closed)"}, + {"id": 4, "subject": "Task 0.5: IMeshPskProvider + CentralCommandAuthInterceptor", "phase": "0", "branch": "feat/grpc-phase0-contract-auth", "classification": "high-risk", "parallelizableWith": ["0.4 (batch, same agent)"], "status": "completed", "blockedBy": [2], "commit": "68a416e4 + fixes 24e55d33/a4631410", "review": "spec ✅ + code ✅ (opus, fix loop closed); 4 items gated on Phase 1A — see tracking §Phase 1A security register"}, + {"id": 5, "subject": "Task 0.6: rig config + FleetStatus issue + docs stub + Phase 0 gate", "phase": "0", "branch": "feat/grpc-phase0-contract-auth", "classification": "small", "parallelizableWith": [], "status": "completed", "blockedBy": [0, 1, 2, 3, 4], "commit": "23555ed0", "review": "code ✅", "gate": "PASS 4/4; issues #542 #543 filed"}, {"id": 6, "subject": "Task 1A.1: NodeCommandGrpcService on the telemetry listener", "phase": "1A", "branch": "feat/grpc-node-command", "classification": "standard", "parallelizableWith": ["1B.1 (other worktree)"], "status": "pending", "blockedBy": [5]}, {"id": 7, "subject": "Task 1A.2: MeshGrpcCommandDialer + third branch in CentralCommunicationActor", "phase": "1A", "branch": "feat/grpc-node-command", "classification": "high-risk", "parallelizableWith": ["1B.2 (other worktree)"], "status": "pending", "blockedBy": [6]}, {"id": 8, "subject": "Task 1A.3: 1A live gate (central pair flipped, rig-only, reverted)", "phase": "1A", "branch": "feat/grpc-node-command", "classification": "small", "parallelizableWith": [], "status": "pending", "blockedBy": [7]}, diff --git a/src/Core/ZB.MOM.WW.OtOpcUa.Cluster/MeshTransportOptions.cs b/src/Core/ZB.MOM.WW.OtOpcUa.Cluster/MeshTransportOptions.cs index d3043e0e..a52c515d 100644 --- a/src/Core/ZB.MOM.WW.OtOpcUa.Cluster/MeshTransportOptions.cs +++ b/src/Core/ZB.MOM.WW.OtOpcUa.Cluster/MeshTransportOptions.cs @@ -6,11 +6,18 @@ namespace ZB.MOM.WW.OtOpcUa.Cluster; /// /// /// -/// is a deliberate dark switch, not a permanent knob: both -/// transports live in the tree for one phase so the docker-dev rig can run the same -/// deployment over each and compare, and so a bad Phase 2 rolls back with a config change -/// rather than a revert. Phase 6 deletes the DistributedPubSub branch along with the single -/// mesh that made it viable. +/// is a deliberate dark switch: the transports live in the tree +/// side by side so the docker-dev rig can run the same deployment over each and compare, and +/// so a bad rollout rolls back with a config change rather than a revert. +/// +/// +/// The DistributedPubSub branch is retained, not scheduled for deletion. An earlier +/// comment here promised Phase 6 would delete it along with the single mesh; that was +/// deliberately reversed — see docs/Configuration.md § "The compiled transport-mode +/// defaults were deliberately NOT flipped". DPS remains the correct and default transport for +/// a single-mesh topology, which every integration test and every profile that leaves this +/// key unset still runs. What is being replaced is , by +/// . /// /// /// Discovery is deliberately asymmetric, matching the sister project. Central @@ -32,9 +39,19 @@ public sealed class MeshTransportOptions public const string ModeClusterClient = "ClusterClient"; /// - /// (default) or . Any other value fails - /// the host at startup rather than silently falling back — a transport that is neither of - /// these delivers nothing, and the symptom would be commands that vanish. + /// Commands cross the mesh boundary as unary gRPC calls on the mesh_transport.v1 + /// contract — central dials each node's NodeCommandService, a node posts its + /// deployment ack to central's CentralCommandService. The replacement for + /// ; unlike it, the command plane is authenticated + /// (). + /// + public const string ModeGrpc = "Grpc"; + + /// + /// (default), or + /// . Any other value fails the host at startup rather than silently + /// falling back — a transport that is none of these delivers nothing, and the symptom would + /// be commands that vanish. /// public string Mode { get; set; } = ModeDps; @@ -58,4 +75,91 @@ public sealed class MeshTransportOptions /// process — migration, direct SQL, a second admin node — is picked up without a restart. /// public int ContactRefreshSeconds { get; set; } = 60; + + /// + /// MeshTransport:CentralGrpcEndpoints — node-side only, and only under + /// : the gRPC base addresses of the central pair's + /// CentralCommandService, e.g. http://central-1:4055. Full URLs + /// (scheme://host:port), not bare host:port — http:// is h2c. + /// + /// + /// Tried in order, stopping at the first success, which is what preserves the + /// receptionist's one-ack-per-node-per-dispatch semantic: either central node routes the ack + /// to the same ConfigPublishCoordinator singleton proxy, so which one answers does not + /// matter, but two answering would double-count. The gRPC counterpart of + /// , and it replaces it — a node running + /// never reads the Akka contact points. + /// + public string[] CentralGrpcEndpoints { get; set; } = []; + + /// + /// MeshTransport:GrpcPsk — this node's pre-shared key for the command plane, one per + /// application Cluster. Under it is both the bearer this node + /// presents when posting an ack to central and the bearer it requires on inbound + /// NodeCommandService calls. + /// + /// + /// + /// Production form is a secret reference — ${secret:OT-MESH-PSK-<clusterId>}, + /// e.g. ${secret:OT-MESH-PSK-SITE-A} — resolved pre-host by the existing + /// SecretReferenceExpander, so it works on a driver-only node with no ConfigDb. + /// Supply a literal only on a rig, via the environment + /// (MeshTransport__GrpcPsk); never commit one. + /// + /// + /// Per-Cluster, not fleet-wide, on purpose. The three existing shared-bearer gRPC + /// channels are read/telemetry planes; this one carries RestartDriver, + /// AlarmCommand and deployment dispatch, so a single fleet key would let one + /// compromised site node command every other site and forge acks fleet-wide. + /// + /// + /// Nullable because unset must be distinguishable from blank at the boundary that + /// fails closed: the serve-side interceptor rejects every call when no key is + /// configured rather than defaulting to an empty comparison that any empty bearer would + /// satisfy. + /// + /// + public string? GrpcPsk { get; set; } + + /// + /// MeshTransport:DispatchTimeoutSeconds — central-side deadline, in seconds, for one + /// NodeCommandService.Execute call to one node. Must be greater than 0 under + /// . + /// + /// + /// Bounded per node because a fan-out dispatch dials every enabled node in the cluster: an + /// unbounded call to one unreachable node would stall the commands queued behind it. Expiry + /// is logged and dropped — the transport is fire-and-forget and layers no retry, so the + /// deploy path's 2-minute apply deadline stays the only recovery mechanism. + /// Validation of this key is scoped to driver-role nodes; that covers the central pair only + /// because central nodes are always fused (admin,driver,cluster-*). A pure-admin + /// central node would leave this unvalidated on the node that uses it. + /// + public int DispatchTimeoutSeconds { get; set; } = 10; + + /// + /// MeshTransport:AckTimeoutSeconds — node-side deadline, in seconds, for one + /// CentralCommandService.ReportApply call to one central endpoint. Must be greater + /// than 0 under . + /// + /// + /// Applied per endpoint, not per ack: the node walks in + /// order, so a dead first endpoint costs this deadline before the second is tried. + /// + public int AckTimeoutSeconds { get; set; } = 10; + + /// + /// MeshTransport:ClusterPsks:{clusterId} — central-side override map of per-cluster + /// PSKs, consulted before the Secrets store when central builds a channel to (or + /// verifies an ack from) a given Cluster. Keyed case-insensitively. + /// + /// + /// Production leaves this empty and the runtime provider resolves + /// OT-MESH-PSK-{clusterId} from the Secrets store instead; the map exists so the + /// docker-dev rig and integration tests can supply literals through the environment + /// (MeshTransport__ClusterPsks__SITE-A) without standing up a KEK. Central needs a map + /// rather than a single key because its cluster set is DB-dynamic — one key per site pair it + /// talks to — whereas a node holds exactly its own (). + /// + public Dictionary ClusterPsks { get; set; } = new(StringComparer.OrdinalIgnoreCase); } diff --git a/src/Core/ZB.MOM.WW.OtOpcUa.Cluster/MeshTransportOptionsValidator.cs b/src/Core/ZB.MOM.WW.OtOpcUa.Cluster/MeshTransportOptionsValidator.cs index afcd8779..ad47fc88 100644 --- a/src/Core/ZB.MOM.WW.OtOpcUa.Cluster/MeshTransportOptionsValidator.cs +++ b/src/Core/ZB.MOM.WW.OtOpcUa.Cluster/MeshTransportOptionsValidator.cs @@ -1,3 +1,4 @@ +using Microsoft.Extensions.Configuration; using Microsoft.Extensions.Options; namespace ZB.MOM.WW.OtOpcUa.Cluster; @@ -15,6 +16,22 @@ namespace ZB.MOM.WW.OtOpcUa.Cluster; /// public sealed class MeshTransportOptionsValidator : IValidateOptions { + private readonly IConfiguration _configuration; + + /// + /// DI-constructed by AddValidatedOptions (a plain AddSingleton), so a + /// constructor dependency is safe here. — not + /// IClusterRoleInfo — is the source of this node's roles: IClusterRoleInfo's + /// implementation needs the ActorSystem, which does not exist yet at + /// ValidateOnStart time. Same shape as and + /// . + /// + /// The application configuration, read for Cluster:Roles. + public MeshTransportOptionsValidator(IConfiguration configuration) + { + _configuration = configuration; + } + /// public ValidateOptionsResult Validate(string? name, MeshTransportOptions options) { @@ -25,13 +42,14 @@ public sealed class MeshTransportOptionsValidator : IValidateOptions()); + var isDriver = Array.IndexOf(roles, RoleParser.Driver) >= 0; + + if (isGrpc && isDriver) + { + if (string.IsNullOrWhiteSpace(options.GrpcPsk)) + { + errors.Add( + $"{MeshTransportOptions.SectionName}:{nameof(MeshTransportOptions.GrpcPsk)} is " + + $"empty. Under {MeshTransportOptions.ModeGrpc} this per-cluster pre-shared key is " + + "the whole authentication boundary of the command plane in both directions — " + + "without it this node's inbound command service fails closed and every ack it " + + "posts to central is rejected. Supply it via the environment " + + $"({MeshTransportOptions.SectionName}__{nameof(MeshTransportOptions.GrpcPsk)}), in " + + "production as ${secret:OT-MESH-PSK-}."); + } + + if (options.CentralGrpcEndpoints.Length == 0) + { + errors.Add( + $"{MeshTransportOptions.SectionName}:" + + $"{nameof(MeshTransportOptions.CentralGrpcEndpoints)} is empty. Under " + + $"{MeshTransportOptions.ModeGrpc} a driver node has nowhere to post its " + + "deployment ack, and every deployment would time out naming a node that is " + + "running perfectly. Supply them via the environment " + + $"({MeshTransportOptions.SectionName}__" + + $"{nameof(MeshTransportOptions.CentralGrpcEndpoints)}__0)."); + } + + foreach (var endpoint in options.CentralGrpcEndpoints) + { + if (!endpoint.StartsWith("http://", StringComparison.OrdinalIgnoreCase) + && !endpoint.StartsWith("https://", StringComparison.OrdinalIgnoreCase)) + { + errors.Add( + $"Central gRPC endpoint '{endpoint}' must start with 'http://' (h2c) or " + + "'https://' — it is a gRPC base address (scheme://host:port), not a bare " + + "host:port and not an Akka remoting address."); + } + } + + // Both deadlines are per-call and there is no retry behind them, so a non-positive value + // is not "wait forever" — it is a call that expires before it is made, i.e. a command + // plane that silently delivers nothing. + if (options.DispatchTimeoutSeconds <= 0) + { + errors.Add( + $"{MeshTransportOptions.SectionName}:" + + $"{nameof(MeshTransportOptions.DispatchTimeoutSeconds)} must be greater than 0 " + + $"(was {options.DispatchTimeoutSeconds})."); + } + + if (options.AckTimeoutSeconds <= 0) + { + errors.Add( + $"{MeshTransportOptions.SectionName}:" + + $"{nameof(MeshTransportOptions.AckTimeoutSeconds)} must be greater than 0 " + + $"(was {options.AckTimeoutSeconds})."); + } + } + return errors.Count == 0 ? ValidateOptionsResult.Success : ValidateOptionsResult.Fail(string.Join(" ", errors)); diff --git a/src/Core/ZB.MOM.WW.OtOpcUa.Cluster/SplitTopologyTransportValidator.cs b/src/Core/ZB.MOM.WW.OtOpcUa.Cluster/SplitTopologyTransportValidator.cs index e9a80a95..5a5bb597 100644 --- a/src/Core/ZB.MOM.WW.OtOpcUa.Cluster/SplitTopologyTransportValidator.cs +++ b/src/Core/ZB.MOM.WW.OtOpcUa.Cluster/SplitTopologyTransportValidator.cs @@ -98,14 +98,22 @@ public sealed class SplitTopologyTransportValidator : IValidateOptions +/// The one translation between the five cross-mesh command records and their +/// mesh_transport.v1 wire shapes (ClusterClient→gRPC migration Phase 0). Both directions live +/// here so a field can only be added to one side without the other by making this file +/// visibly asymmetric. +/// +/// +/// +/// Nothing here guesses. The transport it replaces — an Akka ClusterClient receptionist — +/// dropped an undeliverable command silently, and the entire point of the migration is to stop +/// doing that. So every decode that cannot produce a faithful record throws: an unset oneof, an +/// APPLY_OUTCOME_UNSPECIFIED ack, an enum value from a newer peer. A loud failure on one +/// command is recoverable; a deployment reported applied because an absent enum field decoded to +/// 0 is not. +/// +/// +/// Presence, not emptiness. The two nullable members that cross this contract — +/// and — are +/// optional string on the wire, so null maps to an absent field and "" to a +/// present-but-empty one. Collapsing them would report an operator's deliberately blank comment +/// as "no comment supplied", and a blank failure reason as "no reason recorded". +/// uses a message field, whose own absence carries the +/// null. +/// +/// +/// Correlation is written twice and read once. Encode stamps the envelope's +/// correlation_id and the command's own from the same source. Decode takes the command's +/// own for the three records that declare one — that member is domain state the record must +/// carry back verbatim — and leaves the envelope's to the transport, which is the only place +/// 's correlation can live at all since the record has no member for it. +/// +/// +public static class MeshTransportDtoMapper +{ + // ---------------------------------------------------------------- DispatchDeployment + + /// Projects onto its wire form. + /// The domain dispatch command. + /// The wire . + public static DispatchDeploymentCommand ToProto(DispatchDeployment message) + { + ArgumentNullException.ThrowIfNull(message); + + return new DispatchDeploymentCommand + { + DeploymentId = message.DeploymentId.ToString(), + RevisionHash = message.RevisionHash.Value ?? "", + CorrelationId = message.CorrelationId.ToString(), + }; + } + + /// Rebuilds the domain record from its wire form. + /// The wire dispatch command. + /// The domain . + /// An id field is not a hyphenless GUID. + /// The revision hash is absent or blank. + public static DispatchDeployment FromProto(DispatchDeploymentCommand proto) + { + ArgumentNullException.ThrowIfNull(proto); + + return new DispatchDeployment( + DeploymentId.Parse(proto.DeploymentId), + RevisionHash.Parse(proto.RevisionHash), + CorrelationId.Parse(proto.CorrelationId)); + } + + // ---------------------------------------------------------------- RestartDriver + + /// Projects onto its wire form. + /// The domain restart command. + /// The wire . + public static RestartDriverCommand ToProto(RestartDriver message) + { + ArgumentNullException.ThrowIfNull(message); + + // Required (non-optional) proto string setters throw on null and the record's nullable-ref + // annotations are compile-time only, so a null reaching here is already a producer bug. Coalesce + // rather than throw out of the mapper: a command with a blank field is diagnosable downstream, + // whereas a throw here is the silent drop this transport exists to remove. + return new RestartDriverCommand + { + ClusterId = message.ClusterId ?? "", + DriverInstanceId = message.DriverInstanceId ?? "", + ActorByUserName = message.ActorByUserName ?? "", + CorrelationId = message.CorrelationId.ToString("N"), + }; + } + + /// Rebuilds the domain record from its wire form. + /// The wire restart command. + /// The domain . + /// The correlation id is not a hyphenless GUID. + public static RestartDriver FromProto(RestartDriverCommand proto) + { + ArgumentNullException.ThrowIfNull(proto); + + return new RestartDriver( + proto.ClusterId, + proto.DriverInstanceId, + proto.ActorByUserName, + Guid.ParseExact(proto.CorrelationId, "N")); + } + + // ---------------------------------------------------------------- ReconnectDriver + + /// Projects onto its wire form. + /// The domain reconnect command. + /// The wire . + public static ReconnectDriverCommand ToProto(ReconnectDriver message) + { + ArgumentNullException.ThrowIfNull(message); + + return new ReconnectDriverCommand + { + ClusterId = message.ClusterId ?? "", + DriverInstanceId = message.DriverInstanceId ?? "", + ActorByUserName = message.ActorByUserName ?? "", + CorrelationId = message.CorrelationId.ToString("N"), + }; + } + + /// Rebuilds the domain record from its wire form. + /// The wire reconnect command. + /// The domain . + /// The correlation id is not a hyphenless GUID. + public static ReconnectDriver FromProto(ReconnectDriverCommand proto) + { + ArgumentNullException.ThrowIfNull(proto); + + return new ReconnectDriver( + proto.ClusterId, + proto.DriverInstanceId, + proto.ActorByUserName, + Guid.ParseExact(proto.CorrelationId, "N")); + } + + // ---------------------------------------------------------------- AlarmCommand + + /// Projects onto its wire form. + /// The domain alarm command. + /// The wire . + /// + /// is copied verbatim. It is an unconstrained string in the + /// record and stays one on the wire, so an operation this build does not recognise reaches the + /// far side intact instead of collapsing onto an enum's UNSPECIFIED and being dropped. + /// + public static AlarmCommandMessage ToProto(AlarmCommand message) + { + ArgumentNullException.ThrowIfNull(message); + + var proto = new AlarmCommandMessage + { + AlarmId = message.AlarmId ?? "", + Operation = message.Operation ?? "", + User = message.User ?? "", + }; + + if (message.Comment is not null) + proto.Comment = message.Comment; + if (message.UnshelveAtUtc is { } unshelveAt) + proto.UnshelveAtUtc = ToUtcTimestamp(unshelveAt); + + return proto; + } + + /// Rebuilds the domain record from its wire form. + /// The wire alarm command. + /// The domain . + public static AlarmCommand FromProto(AlarmCommandMessage proto) + { + ArgumentNullException.ThrowIfNull(proto); + + return new AlarmCommand( + proto.AlarmId, + proto.Operation, + proto.User, + proto.HasComment ? proto.Comment : null, + proto.UnshelveAtUtc?.ToDateTime()); + } + + // ---------------------------------------------------------------- ApplyAck + + /// Projects onto its wire form. + /// The domain per-node apply acknowledgment. + /// The wire . + /// + /// is not a declared . + /// + public static ApplyAckMessage ToProto(ApplyAck message) + { + ArgumentNullException.ThrowIfNull(message); + + var proto = new ApplyAckMessage + { + DeploymentId = message.DeploymentId.ToString(), + NodeId = message.NodeId.Value ?? "", + Outcome = message.Outcome switch + { + ApplyAckOutcome.Applied => ApplyOutcome.Applied, + ApplyAckOutcome.Failed => ApplyOutcome.Failed, + _ => throw new ArgumentOutOfRangeException( + nameof(message), message.Outcome, "Unknown ApplyAckOutcome"), + }, + CorrelationId = message.CorrelationId.ToString(), + }; + + if (message.FailureReason is not null) + proto.FailureReason = message.FailureReason; + + return proto; + } + + /// Rebuilds the domain record from its wire form. + /// The wire apply acknowledgment. + /// The domain . + /// + /// The outcome is APPLY_OUTCOME_UNSPECIFIED or a value this build does not know. + /// + /// An id field is not a hyphenless GUID. + /// The node id is absent or blank. + public static ApplyAck FromProto(ApplyAckMessage proto) + { + ArgumentNullException.ThrowIfNull(proto); + + // proto3 decodes an absent enum field as 0, and the deploy path seals a revision on Applied. + // ApplyAckOutcome has no Unknown member to land on, so an unspecified or unrecognised outcome + // must refuse rather than pick one of the two real answers — either choice would be a lie about + // what the node said, and the "Applied" lie seals a deployment nobody applied. + var outcome = proto.Outcome switch + { + ApplyOutcome.Applied => ApplyAckOutcome.Applied, + ApplyOutcome.Failed => ApplyAckOutcome.Failed, + _ => throw new InvalidOperationException( + $"ApplyAckMessage carries outcome '{proto.Outcome}', which has no ApplyAckOutcome " + + "equivalent. Refusing to decode rather than report an unsaid deployment result."), + }; + + return new ApplyAck( + DeploymentId.Parse(proto.DeploymentId), + NodeId.Parse(proto.NodeId), + outcome, + proto.HasFailureReason ? proto.FailureReason : null, + CorrelationId.Parse(proto.CorrelationId)); + } + + // ---------------------------------------------------------------- Envelope + + /// Wraps a dispatch command in its transport envelope. + /// The domain dispatch command. + /// The envelope, correlated from the record's own CorrelationId. + public static CommandEnvelope ToEnvelope(DispatchDeployment message) + { + ArgumentNullException.ThrowIfNull(message); + + return new CommandEnvelope + { + CorrelationId = message.CorrelationId.ToString(), + DispatchDeployment = ToProto(message), + }; + } + + /// Wraps a restart command in its transport envelope. + /// The domain restart command. + /// The envelope, correlated from the record's own CorrelationId. + public static CommandEnvelope ToEnvelope(RestartDriver message) + { + ArgumentNullException.ThrowIfNull(message); + + return new CommandEnvelope + { + CorrelationId = message.CorrelationId.ToString("N"), + RestartDriver = ToProto(message), + }; + } + + /// Wraps a reconnect command in its transport envelope. + /// The domain reconnect command. + /// The envelope, correlated from the record's own CorrelationId. + public static CommandEnvelope ToEnvelope(ReconnectDriver message) + { + ArgumentNullException.ThrowIfNull(message); + + return new CommandEnvelope + { + CorrelationId = message.CorrelationId.ToString("N"), + ReconnectDriver = ToProto(message), + }; + } + + /// Wraps an alarm command in its transport envelope. + /// The domain alarm command. + /// + /// Transport correlation. Explicit because is the one record with no + /// correlation member of its own, so the envelope is the only place it can be carried. + /// + /// The envelope. + public static CommandEnvelope ToEnvelope(AlarmCommand message, CorrelationId correlationId) + { + ArgumentNullException.ThrowIfNull(message); + + return new CommandEnvelope + { + CorrelationId = correlationId.ToString(), + AlarmCommand = ToProto(message), + }; + } + + /// + /// Untyped dispatch for the transport router, which holds its payload as + /// (an ). + /// + /// One of the four central→node command records. + /// + /// Consulted only when is an . The + /// other three carry their own correlation and it is authoritative for them, so passing a + /// different value here cannot make the envelope disagree with the command inside it. + /// + /// The envelope produced by the matching typed overload. + /// + /// is not one of the four handled command records — the transport + /// would otherwise carry a payload the far side cannot decode. + /// + public static CommandEnvelope ToEnvelope(object message, CorrelationId alarmCorrelationId) + { + ArgumentNullException.ThrowIfNull(message); + + return message switch + { + DispatchDeployment m => ToEnvelope(m), + RestartDriver m => ToEnvelope(m), + ReconnectDriver m => ToEnvelope(m), + AlarmCommand m => ToEnvelope(m, alarmCorrelationId), + _ => throw new ArgumentOutOfRangeException( + nameof(message), message.GetType().FullName, "Not a cross-mesh command record"), + }; + } + + /// Unwraps a transport envelope into the domain record it carries. + /// The received envelope. + /// + /// One of , , + /// or , boxed — the node-side handlers + /// match on the concrete type exactly as they do under the DPS transport. + /// + /// + /// The oneof is unset or names a case this build does not handle. Kept unreachable by + /// MeshTransportProtoContract and its contract-lock test; it throws anyway because the + /// alternative is a command that crosses the mesh and disappears. + /// + public static object FromEnvelope(CommandEnvelope envelope) + { + ArgumentNullException.ThrowIfNull(envelope); + + return envelope.CommandCase switch + { + CommandEnvelope.CommandOneofCase.DispatchDeployment => FromProto(envelope.DispatchDeployment), + CommandEnvelope.CommandOneofCase.RestartDriver => FromProto(envelope.RestartDriver), + CommandEnvelope.CommandOneofCase.ReconnectDriver => FromProto(envelope.ReconnectDriver), + CommandEnvelope.CommandOneofCase.AlarmCommand => FromProto(envelope.AlarmCommand), + _ => throw new InvalidOperationException( + $"CommandEnvelope (correlation '{envelope.CorrelationId}') carries oneof case " + + $"'{envelope.CommandCase}', which this build does not handle."), + }; + } + + /// + /// Converts a domain to a proto without ever + /// throwing on Kind: Utc is used as-is, Local is converted, and Unspecified is assumed + /// already-UTC (the producer contract for every …Utc member here). + /// + private static Timestamp ToUtcTimestamp(DateTime dt) + { + var utc = dt.Kind switch + { + DateTimeKind.Utc => dt, + DateTimeKind.Local => dt.ToUniversalTime(), + _ => DateTime.SpecifyKind(dt, DateTimeKind.Utc), + }; + + return Timestamp.FromDateTime(utc); + } +} diff --git a/src/Core/ZB.MOM.WW.OtOpcUa.Commons/Protos/MeshTransportProtoContract.cs b/src/Core/ZB.MOM.WW.OtOpcUa.Commons/Protos/MeshTransportProtoContract.cs new file mode 100644 index 00000000..d49ca315 --- /dev/null +++ b/src/Core/ZB.MOM.WW.OtOpcUa.Commons/Protos/MeshTransportProtoContract.cs @@ -0,0 +1,33 @@ +using ZB.MOM.WW.OtOpcUa.Commons.Protos.MeshTransport.V1; + +namespace ZB.MOM.WW.OtOpcUa.Commons.Protos; + +/// +/// The single source of truth for which variants +/// the cross-mesh command transport actually handles. Both the contract-lock test (which proves +/// this list stays in lock-step with the generated oneof) and the oneof↔domain mapper reference +/// this array, so adding a fifth command to mesh_transport.proto forces a matching entry +/// here or the build/test goes red. +/// +/// +/// Mirrors , and for a sharper reason: a telemetry variant +/// nobody handles loses an observation, whereas a command variant nobody handles loses a +/// deployment, a driver restart or an alarm acknowledgement — silently, since the sender's ack +/// only ever meant "received". +/// +public static class MeshTransportProtoContract +{ + /// + /// Exactly the four cross-mesh command oneof cases mirrored from the domain records: + /// DispatchDeployment / RestartDriver / ReconnectDriver / AlarmCommand. The fifth message on + /// this contract, ApplyAckMessage, travels the opposite direction on its own RPC and is + /// not a oneof member, so it is deliberately absent from this list. + /// + public static readonly CommandEnvelope.CommandOneofCase[] HandledCases = + [ + CommandEnvelope.CommandOneofCase.DispatchDeployment, + CommandEnvelope.CommandOneofCase.RestartDriver, + CommandEnvelope.CommandOneofCase.ReconnectDriver, + CommandEnvelope.CommandOneofCase.AlarmCommand, + ]; +} diff --git a/src/Core/ZB.MOM.WW.OtOpcUa.Commons/Protos/mesh_transport.proto b/src/Core/ZB.MOM.WW.OtOpcUa.Commons/Protos/mesh_transport.proto new file mode 100644 index 00000000..d91c52f8 --- /dev/null +++ b/src/Core/ZB.MOM.WW.OtOpcUa.Commons/Protos/mesh_transport.proto @@ -0,0 +1,118 @@ +syntax = "proto3"; + +package mesh_transport.v1; + +option csharp_namespace = "ZB.MOM.WW.OtOpcUa.Commons.Protos.MeshTransport.V1"; + +import "google/protobuf/timestamp.proto"; + +// ClusterClient→gRPC migration Phase 0: the cross-mesh command plane. The five messages that cross +// an application-Cluster boundary today ride an Akka ClusterClient receptionist; this contract is +// the replacement carrier. Four are central→node commands, one is the node→central deployment ack. +// +// The C# namespace deliberately says "MeshTransport", never "MeshCommand" — the latter is already +// the CLR envelope record ZB.MOM.WW.OtOpcUa.Commons.Messages.Mesh.MeshCommand, and a generated type +// of the same name would collide on every file that touches both sides of the mapper. +// +// Additive-only field evolution: never renumber or reuse a tag; a pre-field peer must read a new +// field's proto3 default correctly (which is why nullable domain members are modelled with `optional` +// for explicit presence, and why every enum reserves 0 as UNSPECIFIED). The oneof is locked by +// MeshTransportProtoContract + its test in the Commons test project. + +// Central → node. Delivered to the node's NodeCommunicationActor, which re-emits the decoded message +// on the node's EventStream. ONE-WAY semantics, preserved verbatim from the ClusterClient transport: +// the ack means "received and re-emitted", nothing more — not applied, not acknowledged by any +// subscriber. No retry is layered on a failure; a failed Execute is logged per node and dropped, +// which is an honest upgrade over the receptionist's silent drop rather than a semantic change. +service NodeCommandService { + rpc Execute (CommandEnvelope) returns (ExecuteAck); +} + +// Node → central. Delivered to central's CentralCommunicationActor, which forwards to the +// ConfigPublishCoordinator singleton proxy — so an ack landing on either node of the central pair +// routes identically, exactly as the receptionist made arrival-node-agnostic. ONE-WAY semantics: the +// ack means "handed to the coordinator", not "the deployment is complete". +service CentralCommandService { + rpc ReportApply (ApplyAckMessage) returns (ReportAck); +} + +message CommandEnvelope { + // Transport-level correlation, carried for every command including AlarmCommandMessage, which is + // the one CLR record with no correlation member of its own. The other three ALSO carry their own + // domain correlation field-for-field below; the duplication is deliberate, so each command stays + // independently round-trippable through the mapper and the envelope stays loggable on its own. + string correlation_id = 1; + oneof command { + DispatchDeploymentCommand dispatch_deployment = 2; // <- Messages.Deploy.DispatchDeployment + RestartDriverCommand restart_driver = 3; // <- Messages.Admin.RestartDriver + ReconnectDriverCommand reconnect_driver = 4; // <- Messages.Admin.ReconnectDriver + AlarmCommandMessage alarm_command = 5; // <- OpcUa.AlarmCommand + } +} + +// Deliberately empty. A non-OK gRPC status is the ONLY failure signal on this call — putting a +// success flag in the body would invite a caller to read OK+false as a distinct outcome the +// fire-and-forget contract does not have. Extensible later without a version bump. +message ExecuteAck { +} + +// Deliberately empty, for the same reason as ExecuteAck. +message ReportAck { +} + +// Mirrors ZB.MOM.WW.OtOpcUa.Commons.Messages.Deploy.DispatchDeployment. +message DispatchDeploymentCommand { + string deployment_id = 1; // DeploymentId (Guid) in its "N" hex form + string revision_hash = 2; // RevisionHash — lowercase 64-char SHA-256 hex + string correlation_id = 3; // CorrelationId (Guid) in its "N" hex form +} + +// Mirrors ZB.MOM.WW.OtOpcUa.Commons.Messages.Admin.RestartDriver. +message RestartDriverCommand { + string cluster_id = 1; + string driver_instance_id = 2; + string actor_by_user_name = 3; + string correlation_id = 4; // Guid in the record; "N" hex form +} + +// Mirrors ZB.MOM.WW.OtOpcUa.Commons.Messages.Admin.ReconnectDriver. +message ReconnectDriverCommand { + string cluster_id = 1; + string driver_instance_id = 2; + string actor_by_user_name = 3; + string correlation_id = 4; // Guid in the record; "N" hex form +} + +// Mirrors ZB.MOM.WW.OtOpcUa.Commons.OpcUa.AlarmCommand. +message AlarmCommandMessage { + string alarm_id = 1; + // The Part 9 operation (Acknowledge / Confirm / OneShotShelve / TimedShelve / Unshelve / Enable / + // Disable / AddComment). A `string` in the CLR record, and kept as one here — the same call the + // telemetry contract makes for transition_kind / state / level. Promoting it to a proto enum would + // put a lossy translation in the mapper for a set the CLR side does not constrain: an operation the + // enum did not know would decode to UNSPECIFIED and the command would be dropped, which is the + // exact silent-drop class this migration exists to remove. + string operation = 2; + string user = 3; + optional string comment = 4; // nullable in the record — presence distinguishes null from "" + google.protobuf.Timestamp unshelve_at_utc = 5; // DateTime? — absent Timestamp encodes null +} + +// Mirrors ZB.MOM.WW.OtOpcUa.Commons.Messages.Deploy.ApplyAck. +message ApplyAckMessage { + string deployment_id = 1; // DeploymentId (Guid) in its "N" hex form + string node_id = 2; // NodeId — a logical host name, already a string + ApplyOutcome outcome = 3; + optional string failure_reason = 4; // nullable in the record — presence distinguishes null from "" + string correlation_id = 5; // CorrelationId (Guid) in its "N" hex form +} + +// Mirrors ZB.MOM.WW.OtOpcUa.Commons.Messages.Deploy.ApplyAckOutcome, shifted by one so 0 stays +// meaningless. proto3 decodes an absent enum field as 0; if 0 were Applied, an ack that lost the +// field would report a deployment applied when the node said nothing at all — and the deploy path +// seals on exactly that signal. +enum ApplyOutcome { + APPLY_OUTCOME_UNSPECIFIED = 0; + APPLY_OUTCOME_APPLIED = 1; + APPLY_OUTCOME_FAILED = 2; +} diff --git a/src/Core/ZB.MOM.WW.OtOpcUa.Commons/ZB.MOM.WW.OtOpcUa.Commons.csproj b/src/Core/ZB.MOM.WW.OtOpcUa.Commons/ZB.MOM.WW.OtOpcUa.Commons.csproj index 5e55544f..bd560b43 100644 --- a/src/Core/ZB.MOM.WW.OtOpcUa.Commons/ZB.MOM.WW.OtOpcUa.Commons.csproj +++ b/src/Core/ZB.MOM.WW.OtOpcUa.Commons/ZB.MOM.WW.OtOpcUa.Commons.csproj @@ -28,6 +28,7 @@ + diff --git a/src/Server/ZB.MOM.WW.OtOpcUa.ControlPlane/AdminOperations/AdminOperationsActor.cs b/src/Server/ZB.MOM.WW.OtOpcUa.ControlPlane/AdminOperations/AdminOperationsActor.cs index 4b000bf4..9a7de9d6 100644 --- a/src/Server/ZB.MOM.WW.OtOpcUa.ControlPlane/AdminOperations/AdminOperationsActor.cs +++ b/src/Server/ZB.MOM.WW.OtOpcUa.ControlPlane/AdminOperations/AdminOperationsActor.cs @@ -88,8 +88,12 @@ public sealed class AdminOperationsActor : ReceiveActor /// /// /// The fallback keeps ~12 existing tests (and any node wired before per-cluster mesh Phase 2) - /// on exactly the pre-Phase-2 behaviour. It is not a permanent seam: Phase 6 deletes it along - /// with the DistributedPubSub branch and the single mesh. + /// on exactly the pre-Phase-2 behaviour. The DistributedPubSub branch behind it is + /// retained, not scheduled for deletion — an earlier note here promised Phase 6 would + /// remove it, which was deliberately reversed (docs/Configuration.md § "The compiled + /// transport-mode defaults were deliberately NOT flipped"): DPS is still the correct and + /// default transport for a single-mesh topology. What is being replaced is the ClusterClient + /// branch, by gRPC. /// /// The legacy DistributedPubSub topic. /// The command. diff --git a/src/Server/ZB.MOM.WW.OtOpcUa.ControlPlane/Communication/IMeshPskProvider.cs b/src/Server/ZB.MOM.WW.OtOpcUa.ControlPlane/Communication/IMeshPskProvider.cs new file mode 100644 index 00000000..61d37f74 --- /dev/null +++ b/src/Server/ZB.MOM.WW.OtOpcUa.ControlPlane/Communication/IMeshPskProvider.cs @@ -0,0 +1,244 @@ +using System.Collections.Concurrent; +using System.Diagnostics.CodeAnalysis; +using Microsoft.Extensions.Logging; +using Microsoft.Extensions.Options; +using ZB.MOM.WW.OtOpcUa.Cluster; +using ZB.MOM.WW.Secrets.Abstractions; + +namespace ZB.MOM.WW.OtOpcUa.ControlPlane.Communication; + +/// +/// Central's per-cluster pre-shared key lookup for the gRPC command plane. A node holds exactly +/// one key (MeshTransport:GrpcPsk, its own cluster's); central talks to every cluster, so +/// it needs one key per Cluster — both to authenticate itself when dialling a node and to +/// verify an inbound deployment ack. +/// +/// +/// +/// Fail-closed by contract. Every failure — unknown cluster, absent secret, blank +/// value, a store that throws — is reported as "no key" (false), and the caller +/// denies the inbound call or skips the outbound dial. There is no arm on which a caller may +/// proceed unauthenticated; a missing key must cost delivery, not authentication. +/// +/// +/// The Invalidate hooks exist because central's cluster set is DB-dynamic. They are +/// deliberately unconsumed at Phase 0 — Phase 1A wires them to the contact refresh, which is +/// the point at which central learns a cluster's membership (and so possibly its key) has +/// changed. +/// +/// +/// Invalidate is not a revocation barrier. It drops the cached entry; it does +/// not wait for, cancel, or supersede a resolve already in flight. A lookup that entered the +/// store before the call can still publish and return the pre-rotation key to its caller, and +/// a lookup racing the eviction can re-cache the old value it had already read. Key rotation +/// is therefore eventually consistent by seconds, not atomic — treat Invalidate as +/// "stop serving this from memory", never as "this key is now dead everywhere". +/// +/// +public interface IMeshPskProvider +{ + /// Gets the pre-shared key for one application Cluster. + /// The application Cluster id, matched case-insensitively. + /// The key, or null when none could be established. + /// true when a usable key was found; otherwise false. + bool TryGetPsk(string clusterId, [NotNullWhen(true)] out string? psk); + + /// Drops any cached key for one cluster, so the next lookup re-reads the store. + /// The application Cluster id, matched case-insensitively. + void Invalidate(string clusterId); + + /// Drops every cached key. + void InvalidateAll(); +} + +/// +public sealed class MeshPskProvider : IMeshPskProvider +{ + /// + /// Prefix of the secret holding a cluster's key — the full name is + /// OT-MESH-PSK-{clusterId}. The same name a node's + /// ${secret:OT-MESH-PSK-<clusterId>} reference expands, so central and the node + /// read one row rather than two conventions that drift. + /// + public const string SecretNamePrefix = "OT-MESH-PSK-"; + + /// + /// Ceiling on one store read. Not a tuning knob — a backstop: the read runs under the + /// publication lock, so a store that never answers would park the + /// calling gRPC thread and every other caller queued behind it for that cluster. + /// Expiry fails closed like any other fault. + /// + private static readonly TimeSpan StoreReadTimeout = TimeSpan.FromSeconds(3); + + private readonly IOptions _options; + private readonly ISecretResolver _secretResolver; + private readonly ILogger _logger; + + /// + /// Cached store resolutions, keyed by cluster id. Holds only values that came from the + /// Secrets store: the config map is re-read on every lookup (it is an in-memory dictionary), + /// which is what keeps the override unconditional rather than merely first-past-the-post. + /// + /// + /// under so a + /// burst of concurrent inbound acks for one cluster resolves the secret once instead of + /// stampeding the store. Negatives are never retained — see . + /// + private readonly ConcurrentDictionary> _cache = + new(StringComparer.OrdinalIgnoreCase); + + /// Creates the provider. + /// Mesh transport options; ClusterPsks is the override map. + /// The shared secret store read seam. + /// Logger for lookup-failure diagnostics. + public MeshPskProvider( + IOptions options, + ISecretResolver secretResolver, + ILogger logger) + { + ArgumentNullException.ThrowIfNull(options); + ArgumentNullException.ThrowIfNull(secretResolver); + ArgumentNullException.ThrowIfNull(logger); + + _options = options; + _secretResolver = secretResolver; + _logger = logger; + } + + /// + public bool TryGetPsk(string clusterId, [NotNullWhen(true)] out string? psk) + { + psk = null; + + if (string.IsNullOrWhiteSpace(clusterId)) + { + _logger.LogWarning( + "Mesh PSK lookup was asked for a blank cluster id; refusing rather than resolving a " + + "key that would belong to no cluster."); + return false; + } + + // Config map first, on every call. Blank counts as absent: an empty + // MeshTransport__ClusterPsks__ is an unset environment variable, not an operator + // deciding this cluster has an empty key. + if (_options.Value.ClusterPsks.TryGetValue(clusterId, out var configured) + && !string.IsNullOrWhiteSpace(configured)) + { + psk = configured; + return true; + } + + var entry = _cache.GetOrAdd( + clusterId, + static (id, self) => new Lazy( + () => self.ResolveFromStore(id), LazyThreadSafetyMode.ExecutionAndPublication), + this); + + string? resolved; + try + { + resolved = entry.Value; + } + catch + { + // ResolveFromStore swallows everything the STORE can throw — but it logs while doing so, + // and a logger provider that throws escapes the factory. Lazy under + // ExecutionAndPublication caches that exception and rethrows it on every later read, so + // without this guard one bad log call would deny the cluster for the life of the process + // (nothing calls Invalidate until Phase 1A). Evict instance-keyed, same as the negative + // path below, and fail closed. + // + // Deliberately silent: the most likely reason we are here is that logging is what threw, + // so logging the failure is the one action with a real chance of throwing again — out of + // an authentication gate, turning a denial into a 500. + _cache.TryRemove(new KeyValuePair>(clusterId, entry)); + return false; + } + + if (string.IsNullOrWhiteSpace(resolved)) + { + // Never retain a negative. A store that was briefly unreachable would otherwise stay + // "no key" for this cluster until something called Invalidate — and at Phase 0 nothing + // does, so a transient fault would present as a permanent, silent command outage. + _cache.TryRemove(new KeyValuePair>(clusterId, entry)); + return false; + } + + psk = resolved; + return true; + } + + /// + public void Invalidate(string clusterId) + { + if (string.IsNullOrWhiteSpace(clusterId)) + return; + + _cache.TryRemove(clusterId, out _); + } + + /// + public void InvalidateAll() => _cache.Clear(); + + /// + /// Reads OT-MESH-PSK-{clusterId} from the Secrets store, converting every failure mode + /// into null. + /// + /// The application Cluster id. + /// The stored key, or null when it could not be established. + /// + /// + /// Blocking on the async resolver is deliberate. The consumer is a gRPC server + /// interceptor, whose Authorize is synchronous in all four sibling interceptors, + /// and making this seam async would push that shape onto every one of them. What makes + /// the block acceptable is that it is bounded () and + /// happens at most once per cluster per process — every later lookup is a dictionary hit. + /// The bound does not assume anything about what backs the store: this code cannot see + /// whether the resolver reads a local file or crosses a network, so it refuses to wait + /// forever on either. + /// + /// + /// The catch is broad on purpose: this is an authentication boundary, and the set of + /// exceptions a secret store can raise (missing KEK, corrupt row, IO) is neither closed + /// nor interesting here. Anything other than a usable key is "no key". + /// + /// + private string? ResolveFromStore(string clusterId) + { + SecretName name; + try + { + name = new SecretName(SecretNamePrefix + clusterId); + } + catch (ArgumentException ex) + { + // Cluster ids are operator-authored and may carry characters SecretName's allow-list + // rejects. Log the id — it is not sensitive — and never the key. + _logger.LogWarning( + ex, + "Cluster {ClusterId} cannot be expressed as a secret name, so its mesh PSK cannot be " + + "resolved from the store. Supply it through MeshTransport:ClusterPsks instead, or " + + "rename the cluster to [a-z0-9._/-].", + clusterId); + return null; + } + + using var cts = new CancellationTokenSource(StoreReadTimeout); + try + { + return _secretResolver.GetAsync(name, cts.Token).GetAwaiter().GetResult(); + } + catch (Exception ex) + { + // Covers the timeout too — a cancelled read arrives here as OperationCanceledException + // and is treated as every other fault is: no key. + _logger.LogWarning( + ex, + "Failed to resolve the mesh PSK for cluster {ClusterId} from secret {SecretName}. " + + "Commands to that cluster and acks from it are refused until it resolves.", + clusterId, + name.Value); + return null; + } + } +} diff --git a/src/Server/ZB.MOM.WW.OtOpcUa.ControlPlane/Coordinators/ConfigPublishCoordinator.cs b/src/Server/ZB.MOM.WW.OtOpcUa.ControlPlane/Coordinators/ConfigPublishCoordinator.cs index 86787656..f71c4c74 100644 --- a/src/Server/ZB.MOM.WW.OtOpcUa.ControlPlane/Coordinators/ConfigPublishCoordinator.cs +++ b/src/Server/ZB.MOM.WW.OtOpcUa.ControlPlane/Coordinators/ConfigPublishCoordinator.cs @@ -68,8 +68,11 @@ public sealed class ConfigPublishCoordinator : ReceiveActor, IWithTimers /// /// /// Null in tests and on any node wired before Phase 2, where the fallback reproduces the - /// pre-Phase-2 behaviour exactly. The fallback is not a permanent seam — Phase 6 deletes it - /// along with the DistributedPubSub branch and the single mesh. + /// pre-Phase-2 behaviour exactly. The DistributedPubSub branch behind it is retained, + /// not scheduled for deletion — an earlier note here promised Phase 6 would remove it, which + /// was deliberately reversed (docs/Configuration.md § "The compiled transport-mode + /// defaults were deliberately NOT flipped"): DPS is still the correct and default transport + /// for a single-mesh topology. What is being replaced is the ClusterClient branch, by gRPC. /// private readonly IActorRef? _meshRouter; private readonly ILoggingAdapter _log = Context.GetLogger(); diff --git a/src/Server/ZB.MOM.WW.OtOpcUa.ControlPlane/ZB.MOM.WW.OtOpcUa.ControlPlane.csproj b/src/Server/ZB.MOM.WW.OtOpcUa.ControlPlane/ZB.MOM.WW.OtOpcUa.ControlPlane.csproj index 89d2622c..0ffdd314 100644 --- a/src/Server/ZB.MOM.WW.OtOpcUa.ControlPlane/ZB.MOM.WW.OtOpcUa.ControlPlane.csproj +++ b/src/Server/ZB.MOM.WW.OtOpcUa.ControlPlane/ZB.MOM.WW.OtOpcUa.ControlPlane.csproj @@ -19,6 +19,10 @@ Google.Protobuf + Grpc.Core.Api flow transitively from Commons (the generated client). --> + +