diff --git a/scripts/validation/scadabridge-sdk-compatibility/Failover.cs b/scripts/validation/scadabridge-sdk-compatibility/Failover.cs new file mode 100644 index 000000000..ca153914e --- /dev/null +++ b/scripts/validation/scadabridge-sdk-compatibility/Failover.cs @@ -0,0 +1,225 @@ +using System.Collections.Concurrent; +using System.Diagnostics; +using System.Reflection; +using System.Text.Json; +using Akka.Actor; +using Microsoft.Extensions.Logging.Abstractions; +using Microsoft.Extensions.Options; +using ZB.MOM.WW.ScadaBridge.Commons.Interfaces.Protocol; +using ZB.MOM.WW.ScadaBridge.Commons.Interfaces.Services; +using ZB.MOM.WW.ScadaBridge.Commons.Messages.DataConnection; +using ZB.MOM.WW.ScadaBridge.Commons.Serialization; +using ZB.MOM.WW.ScadaBridge.Commons.Types.Enums; +using ZB.MOM.WW.ScadaBridge.Commons.Types.DataConnections; +using ZB.MOM.WW.ScadaBridge.DataConnectionLayer; +using ZB.MOM.WW.ScadaBridge.DataConnectionLayer.Actors; + +/// +/// ScadaBridge#125 redundancy leg. Drives ScadaBridge's own — the +/// production primary/backup state machine, the real , its production +/// defaults and FailoverRetryCount = 3 — against two servers. +/// Phase 1 proves a steady primary (tag values, write, Server-object alarm subscription + ConditionRefresh). +/// The probe then writes primary-ready and waits for primary-killed; whoever runs it (run.py, +/// or a person against the docker-dev pair) kills the primary abruptly in between. Phase 2 requires: Bad +/// quality pushed on loss, the actor reporting "Connected to backup", Good values from the backup, a fresh +/// ConditionRefresh on the backup session, and (optionally) a write through the backup. +/// +internal static class Failover +{ + const string Connection = "sdk2-pair"; + + static string Env(string name, string fallback) => + Environment.GetEnvironmentVariable(name) is { Length: > 0 } v ? v : fallback; + + static void Require(bool value, string reason) { if (!value) throw new InvalidOperationException(reason); } + + static bool Is(object? value, long expected) + { + try { return value is not null and not string && Convert.ToInt64(value) == expected; } + catch (Exception) { return false; } + } + + public static async Task RunAsync(string primary, string backup, string root, OpcUaGlobalOptions options, + Func configFor) + { + // Defaults fit the opcua-sdk-compatibility ProcessHost fixture; the docker-dev rig overrides them. + var tags = Env("SCADABRIDGE_PROBE_FAILOVER_TAGS", + "nsu=https://zb.com/otopcua/raw;s=Matrix/dev/Scalar,nsu=https://zb.com/otopcua/uns;s=Eq/Scalar").Split(','); + var writeTag = Env("SCADABRIDGE_PROBE_FAILOVER_WRITE_TAG", tags[0]); // "none" = no writes + var backupValue = Env("SCADABRIDGE_PROBE_FAILOVER_BACKUP_VALUE", "17"); // "none" = not checked + var alarmMessage = Env("SCADABRIDGE_PROBE_FAILOVER_ALARM_PREFIX", "matrix-"); // "none" = refresh only + var killWait = TimeSpan.FromSeconds(int.Parse(Env("SCADABRIDGE_PROBE_FAILOVER_KILL_WAIT_S", "90"))); + var failoverWait = TimeSpan.FromSeconds(int.Parse(Env("SCADABRIDGE_PROBE_FAILOVER_WAIT_S", "240"))); + + var cases = new List(); + var failures = 0; + var events = new ConcurrentQueue<(DateTimeOffset At, object Message)>(); + var health = DispatchProxy.Create(); + var calls = ((HealthRecorder)(object)health).Calls; + var dcOptions = new DataConnectionOptions(); // production defaults (ReconnectInterval 5 s, ...) + var factory = new DataConnectionFactory(NullLoggerFactory.Instance, Options.Create(options), null, Options.Create(dcOptions)); + var primaryConfig = OpcUaEndpointConfigSerializer.ToFlatDict(configFor(primary)); + var backupConfig = OpcUaEndpointConfigSerializer.ToFlatDict(configFor(backup)); + DateTimeOffset? killedAt = null, backupConnectedAt = null; + var timings = new Dictionary(); + + async Task Check(string name, Func action) + { + try { await action(); cases.Add(new { name, status = "passed" }); } + catch (Exception e) { failures++; cases.Add(new { name, status = "failed", error = e.ToString() }); } + } + async Task Until(Func found, TimeSpan limit, string what) + { + var stop = DateTime.UtcNow + limit; + while (DateTime.UtcNow < stop) + { + if (found() is { } at) return at; + await Task.Delay(50); + } + throw new TimeoutException($"{what}: not observed within {limit.TotalSeconds:F0} s"); + } + DateTimeOffset? Event(DateTimeOffset after, Func match) => + events.Where(e => e.At >= after && e.Message is T t && match(t)).Select(e => (DateTimeOffset?)e.At).FirstOrDefault(); + DateTimeOffset? Health(DateTimeOffset after, string label) => + calls.Where(c => c.At >= after && c.Method == nameof(ISiteHealthCollector.UpdateConnectionEndpoint) && Equals(c.Args.ElementAtOrDefault(1), label)) + .Select(c => (DateTimeOffset?)c.At).FirstOrDefault(); + bool GoodValue(TagValueUpdate u, string tag) => u.TagPath == tag && u.Quality == QualityCode.Good; + + using var system = ActorSystem.Create("probe", "akka.loglevel = INFO\nakka.stdout-loglevel = INFO"); + var instance = system.ActorOf(Props.Create(() => new Collector(events)), "instance"); + var adapter = factory.Create("OpcUa", primaryConfig); + var actor = system.ActorOf(Props.Create(() => new DataConnectionActor( + Connection, adapter, dcOptions, health, factory, "OpcUa", primaryConfig, backupConfig, 3, null)), Connection); + var started = DateTimeOffset.UtcNow; + actor.Tell(new SubscribeTagsRequest("tags", "probe-instance", Connection, tags, started), instance); + actor.Tell(new SubscribeAlarmsRequest("alarms", "probe-instance", Connection, "i=2253", null, started), instance); + + await Check("primary-connected", async () => await Until(() => Health(started, "Connected to primary"), TimeSpan.FromSeconds(60), "Connected to primary")); + await Check("primary-subscriptions-accepted", async () => + { + await Until(() => Event(started, r => r.Success), TimeSpan.FromSeconds(30), "SubscribeTagsResponse success"); + await Until(() => Event(started, r => r.Success), TimeSpan.FromSeconds(30), "SubscribeAlarmsResponse success"); + }); + await Check("primary-good-values-all-tags", async () => + { + foreach (var tag in tags) await Until(() => Event(started, u => GoodValue(u, tag)), TimeSpan.FromSeconds(30), "Good value for " + tag); + }); + await Check("primary-alarm-condition-refresh", async () => + { + await Until(() => Event(started, a => a.Transition.Kind == AlarmTransitionKind.SnapshotComplete), TimeSpan.FromSeconds(30), "SnapshotComplete on primary"); + if (alarmMessage != "none") + await Until(() => Event(started, a => a.Transition.Message.StartsWith(alarmMessage, StringComparison.Ordinal)), TimeSpan.FromSeconds(30), "native alarm on primary"); + }); + if (writeTag != "none") + await Check("primary-write", async () => + { + var at = DateTimeOffset.UtcNow; + var response = await actor.Ask(new WriteTagRequest("w1", Connection, writeTag, 412, at), TimeSpan.FromSeconds(30)); + Require(response.Success, response.ErrorMessage ?? "write refused"); + await Until(() => Event(at, u => GoodValue(u, writeTag) && Is(u.Value, 412)), TimeSpan.FromSeconds(30), "412 echoed by the primary"); + }); + + // Loss evidence is matched from the moment the probe hands the primary over to be killed, not from + // the recorded kill time: the consumer can see the socket close before the killer timestamps it. + var readyAt = DateTimeOffset.UtcNow; + if (failures == 0) + { + File.WriteAllText(Path.Combine(root, "primary-ready"), readyAt.ToString("O")); + Console.WriteLine("PRIMARY-READY " + Path.Combine(root, "primary-ready")); + var killedFile = Path.Combine(root, "primary-killed"); + await Check("primary-killed-by-runner", async () => + { + // The marker carries the kill time; a half-written marker is simply read again. + killedAt = await Until(() => File.Exists(killedFile) && DateTimeOffset.TryParse(File.ReadAllText(killedFile).Trim(), out var at) ? at : null, + killWait, "primary-killed marker"); + }); + } + if (killedAt is { } k) + { + await Check("loss-pushes-bad-quality", async () => + { + var at = await Until(() => Event(readyAt, u => u.Quality != QualityCode.Good) + ?? Event(readyAt, q => q.Quality != QualityCode.Good), failoverWait, "Bad quality after primary loss"); + timings["killToBadQualitySeconds"] = (at - k).TotalSeconds; + }); + await Check("failover-connected-to-backup", async () => + { + backupConnectedAt = await Until(() => Health(readyAt, "Connected to backup"), failoverWait, "Connected to backup"); + timings["killToBackupConnectedSeconds"] = (backupConnectedAt.Value - k).TotalSeconds; + }); + } + if (backupConnectedAt is { } b) + { + await Check("backup-good-values-all-tags", async () => + { + foreach (var tag in tags) + { + var at = await Until(() => Event(b, u => GoodValue(u, tag)), TimeSpan.FromSeconds(60), "Good value from backup for " + tag); + timings["killToGood:" + tag] = (at - killedAt!.Value).TotalSeconds; + } + // The fixture's backup never saw the primary's 412, so its own value proves the feed's source. + if (backupValue != "none" && writeTag != "none") + { + var first = events.Where(e => e.At >= b && e.Message is TagValueUpdate u && GoodValue(u, writeTag)).Select(e => (TagValueUpdate)e.Message).First(); + Require(Is(first.Value, long.Parse(backupValue)), $"First Good value after failover was {first.Value}, expected the backup's own {backupValue}"); + } + }); + await Check("backup-alarm-condition-refresh", async () => + { + var at = await Until(() => Event(b, a => a.Transition.Kind == AlarmTransitionKind.SnapshotComplete), TimeSpan.FromSeconds(60), "SnapshotComplete on backup"); + timings["killToBackupSnapshotCompleteSeconds"] = (at - killedAt!.Value).TotalSeconds; + if (alarmMessage != "none") + await Until(() => Event(b, a => a.Transition.Kind == AlarmTransitionKind.Snapshot + && a.Transition.Message.StartsWith(alarmMessage, StringComparison.Ordinal)), TimeSpan.FromSeconds(30), "native condition replayed by the backup's ConditionRefresh"); + }); + if (writeTag != "none") + await Check("backup-write", async () => + { + var at = DateTimeOffset.UtcNow; + var response = await actor.Ask(new WriteTagRequest("w2", Connection, writeTag, 913, at), TimeSpan.FromSeconds(30)); + Require(response.Success, response.ErrorMessage ?? "write refused"); + await Until(() => Event(at, u => GoodValue(u, writeTag) && Is(u.Value, 913)), TimeSpan.FromSeconds(30), "913 echoed by the backup"); + }); + } + + await system.Terminate(); + var sdkAssembly = typeof(Opc.Ua.NodeId).Assembly; + var result = new + { + leg = "failover", + consumer = typeof(DataConnectionActor).Assembly.GetName().Version?.ToString(), + sdk = FileVersionInfo.GetVersionInfo(sdkAssembly.Location).ProductVersion, + process = Environment.ProcessId, primary, backup, autoAccept = false, + failoverRetryCount = 3, reconnectIntervalSeconds = dcOptions.ReconnectInterval.TotalSeconds, + killedAt, backupConnectedAt, timings, cases, failures, + endpointLabels = calls.Where(c => c.Method == nameof(ISiteHealthCollector.UpdateConnectionEndpoint)) + .Select(c => new { at = c.At, label = c.Args.ElementAtOrDefault(1)?.ToString() }), + alarmTransitions = events.Where(e => e.Message is NativeAlarmTransitionUpdate) + .Select(e => new { at = e.At, kind = ((NativeAlarmTransitionUpdate)e.Message).Transition.Kind.ToString(), message = ((NativeAlarmTransitionUpdate)e.Message).Transition.Message, source = ((NativeAlarmTransitionUpdate)e.Message).Transition.SourceReference }), + alarmSourceUnavailable = events.Where(e => e.Message is NativeAlarmSourceUnavailable).Select(e => e.At), + values = events.Where(e => e.Message is TagValueUpdate).Select(e => new { at = e.At, tag = ((TagValueUpdate)e.Message).TagPath, value = ((TagValueUpdate)e.Message).Value?.ToString(), quality = ((TagValueUpdate)e.Message).Quality.ToString() }), + }; + File.WriteAllText(Path.Combine(root, "result.json"), JsonSerializer.Serialize(result, new JsonSerializerOptions { WriteIndented = true })); + Console.WriteLine($"Failover probe: {cases.Count - failures} passed, {failures} failed; {Path.Combine(root, "result.json")}"); + return failures == 0 && killedAt is not null ? 0 : 1; + } + + /// Stands in for the Instance Actor: records every message the connection actor sends it. + sealed class Collector : ReceiveActor + { + public Collector(ConcurrentQueue<(DateTimeOffset, object)> sink) => ReceiveAny(m => sink.Enqueue((DateTimeOffset.UtcNow, m))); + } +} + +/// Records every call (the actor's own endpoint/health edges). +public class HealthRecorder : DispatchProxy +{ + public ConcurrentQueue<(DateTimeOffset At, string Method, object?[] Args)> Calls { get; } = new(); + + protected override object? Invoke(MethodInfo? targetMethod, object?[]? args) + { + Calls.Enqueue((DateTimeOffset.UtcNow, targetMethod!.Name, args ?? [])); + var type = targetMethod.ReturnType; + return type == typeof(void) || !type.IsValueType ? null : Activator.CreateInstance(type); + } +} diff --git a/scripts/validation/scadabridge-sdk-compatibility/Program.cs b/scripts/validation/scadabridge-sdk-compatibility/Program.cs index 2d1c81a84..658d3660b 100644 --- a/scripts/validation/scadabridge-sdk-compatibility/Program.cs +++ b/scripts/validation/scadabridge-sdk-compatibility/Program.cs @@ -34,10 +34,13 @@ if (args.Length == 3 && args[0] == "prepare") Console.WriteLine($"Consumer identity {publicCert.Thumbprint} {publicCert.Subject}"); return 0; } -if (args.Length != 3) throw new ArgumentException("Usage: Probe | Probe prepare "); -var endpoint = args[0]; -var mode = Enum.Parse(args[1]); -var root = Path.GetFullPath(args[2]); +// ScadaBridge#125: "failover " drives the real DataConnectionActor +// primary/backup machine (Failover.cs); the runner kills the primary once the probe reports it is steady. +var failoverMode = args.Length == 5 && args[0] == "failover"; +if (!failoverMode && args.Length != 3) throw new ArgumentException("Usage: Probe | Probe failover | Probe prepare "); +var endpoint = failoverMode ? args[1] : args[0]; +var mode = Enum.Parse(failoverMode ? args[3] : args[1]); +var root = Path.GetFullPath(failoverMode ? args[4] : args[2]); if (Directory.Exists(root)) throw new IOException("Refusing to reuse an evidence/PKI directory."); Directory.CreateDirectory(root); static void Require(bool value, string reason) { if (!value) throw new InvalidOperationException(reason); } @@ -50,25 +53,32 @@ var pki = Environment.GetEnvironmentVariable("SCADABRIDGE_PROBE_PKI") is { Lengt var options = IdentityOptions(pki); var ownBefore = Directory.Exists(Path.Combine(pki, "own", "certs")) ? Directory.GetFiles(Path.Combine(pki, "own", "certs")).Order().ToArray() : []; // Populate only this probe's trust list; never change the consumer's existing stores. +// Several server certificate directories (a failover pair) are separated by Path.PathSeparator. var trustSource = Environment.GetEnvironmentVariable("SCADABRIDGE_PROBE_SERVER_CERTS"); if (!string.IsNullOrEmpty(trustSource)) { var destination = Path.Combine(options.TrustedPeerStorePath, "certs"); Directory.CreateDirectory(destination); - foreach (var cert in Directory.GetFiles(trustSource, "*.der")) File.Copy(cert, Path.Combine(destination, Path.GetFileName(cert))); + foreach (var source in trustSource.Split(Path.PathSeparator, StringSplitOptions.RemoveEmptyEntries)) + foreach (var cert in Directory.GetFiles(source, "*.der")) File.Copy(cert, Path.Combine(destination, Path.GetFileName(cert)), overwrite: true); } +// The rig variant (a deployed OtOpcUa pair, OpcUa:AllowAnonymous=true) sets SCADABRIDGE_PROBE_ANONYMOUS=1. +OpcUaEndpointConfig ConfigFor(string url) => new() +{ + EndpointUrl = url, SecurityMode = mode, AutoAcceptUntrustedCerts = false, + PublishingIntervalMs = 100, SamplingIntervalMs = 100, + UserIdentity = Environment.GetEnvironmentVariable("SCADABRIDGE_PROBE_ANONYMOUS") == "1" + ? new OpcUaUserIdentityConfig { TokenType = OpcUaUserTokenType.Anonymous } + : new OpcUaUserIdentityConfig + { + TokenType = OpcUaUserTokenType.UsernamePassword, Username = "apikey", + Password = Environment.GetEnvironmentVariable("OPCUA_MATRIX_PASSWORD") ?? throw new InvalidOperationException("Missing task credential"), + }, +}; +if (failoverMode) return await Failover.RunAsync(args[1], args[2], root, options, ConfigFor); var factory = new RealOpcUaClientFactory(options, NullLoggerFactory.Instance); await using var adapter = new OpcUaDataConnection(factory, NullLogger.Instance); -var config = new OpcUaEndpointConfig -{ - EndpointUrl = endpoint, SecurityMode = mode, AutoAcceptUntrustedCerts = false, - PublishingIntervalMs = 100, SamplingIntervalMs = 100, - UserIdentity = new OpcUaUserIdentityConfig - { - TokenType = OpcUaUserTokenType.UsernamePassword, Username = "apikey", - Password = Environment.GetEnvironmentVariable("OPCUA_MATRIX_PASSWORD") ?? throw new InvalidOperationException("Missing task credential"), - }, -}; +var config = ConfigFor(endpoint); async Task Check(string name, Func action) { try { await action(); cases.Add(new { name, status = "passed" }); } @@ -120,6 +130,16 @@ if (failures == 0) Require(actual.Quality.ToString() == "Good", "Quality was not Good"); await adapter.UnsubscribeAsync(handle, ct); }); + // ScadaBridge#125: subscriptions on the UNS NodeId too (Eq/Scalar is a separate UNS variable). + await Check("actual-adapter-monitor-uns", async () => + { + var values = new ConcurrentQueue(); + var handle = await adapter.SubscribeAsync(uns + "Scalar", (_, value) => values.Enqueue(value), ct); + Require((await adapter.WriteAsync(uns + "Scalar", 714, ct)).Success, "Monitored UNS write denied"); + await Wait(() => values.Any(x => x.Value is int i && i == 714)); + Require(values.Last(x => x.Value is int i && i == 714).Quality.ToString() == "Good", "UNS quality was not Good"); + await adapter.UnsubscribeAsync(handle, ct); + }); await Check("actual-adapter-alarm-filter-refresh", async () => { var alarms = new ConcurrentQueue(); @@ -133,6 +153,19 @@ if (failures == 0) await adapter.DisconnectAsync(ct); await adapter.ConnectAsync(OpcUaEndpointConfigSerializer.ToFlatDict(config), ct); Require((await adapter.ReadAsync(raw + "Scalar", ct)).Success, "Consumer failed after reconnect"); + }); + // ScadaBridge#125: the Server-object alarm subscription on the NEW session must get its own + // ConditionRefresh — the active native condition is REPLAYED (Snapshot kind) and the refresh completes. + await Check("actual-adapter-alarm-refresh-after-reconnect", async () => + { + var alarms = new ConcurrentQueue(); + var handle = await adapter.SubscribeAlarmsAsync("i=2253", null, alarms.Enqueue, ct); + await Wait(() => alarms.Any(x => x.Kind == AlarmTransitionKind.SnapshotComplete)); + var replayed = alarms.TakeWhile(x => x.Kind != AlarmTransitionKind.SnapshotComplete).ToArray(); + Require(replayed.Any(x => x.Kind == AlarmTransitionKind.Snapshot && x.SourceReference.Contains("Alarm", StringComparison.Ordinal)), + "ConditionRefresh on the reconnected session did not replay the active native condition; saw: " + + string.Join(", ", replayed.Select(x => $"{x.Kind}:{x.SourceReference}:{x.Message}"))); + await adapter.UnsubscribeAlarmsAsync(handle, ct); await adapter.DisconnectAsync(ct); }); } diff --git a/scripts/validation/scadabridge-sdk-compatibility/README.md b/scripts/validation/scadabridge-sdk-compatibility/README.md index 7122a4c07..de5568094 100644 --- a/scripts/validation/scadabridge-sdk-compatibility/README.md +++ b/scripts/validation/scadabridge-sdk-compatibility/README.md @@ -41,7 +41,10 @@ Never commit an entire output directory: it contains the fixture's private PKI. The consumed subset covers API-key activation, raw/UNS scalar and Int32[] roundtrips, durable hierarchical browse including the native alarm, data quality -and timestamps, alarm filter/ConditionRefresh and disconnect/reconnect. History +and timestamps on raw and UNS subscriptions, alarm filter/ConditionRefresh, +disconnect/reconnect, and (ScadaBridge#125) a Server-object alarm subscription on +the reconnected session whose ConditionRefresh must replay the active native +condition (11 cases per trusted leg). History and Part 9 acknowledgment are not exposed by the inspected consumer interfaces; the parent Shared/CLI matrix covers those application services instead. @@ -52,3 +55,44 @@ fixed in ScadaBridge `ff835447`/`36c64a6b`. Re-run 2026-11-12 with ScadaBridge ACL off; all trust negative controls failed closed. The runner retains the nonzero result and each actual failure; it does not loosen trust or change expected results to return success. No ScadaBridge production source or user PKI is modified. + +## Redundancy failover (ScadaBridge#125) + +`run.py` also runs a `failover` suite (`--suites matrix,failover`, both by +default; `--versions candidate` runs one application version only). Each leg +starts two independent servers and runs `Probe failover + `, which drives ScadaBridge's own `DataConnectionActor` (the +production primary/backup state machine, the real `DataConnectionFactory`, +production `DataConnectionOptions` defaults, `FailoverRetryCount = 3`). Once the +probe has a steady primary (Good values on a raw and a UNS tag, a write, the +Server-object alarm subscription with its ConditionRefresh) it writes +`primary-ready`; the runner then loses the primary and writes `primary-killed`. +The probe must then see Bad quality pushed, "Connected to backup", Good values +whose first value is the backup's own (proving the source), a fresh +ConditionRefresh replaying the native condition, and a write through the backup. +Two loss shapes run per security mode (`--failover-modes None,SignAndEncrypt`, +`--failover-loss kill,freeze`): + +- `kill` — SIGKILL; the OS closes the primary's sockets, so the client sees the loss at once. +- `freeze` — SIGSTOP; the sockets stay open and silent (hung host / dead link), so only the + consumer's session keep-alive can detect it. + +ScadaBridge does not read `ServiceLevel`/`ServerRedundancy`: its failover is +endpoint-level (primary/backup connection details), so two independent fixture +servers exercise exactly what it consumes. `rig-failover.sh ` +runs the same probe against the real local docker-dev pair (central-1 `:4840`, +central-2 `:4841`), `docker kill`s central-1 once the primary is steady and +starts it again afterwards. The rig admits anonymous sessions read-only and has +no active native alarm, so that variant checks values and ConditionRefresh +completion only. Local docker-dev only; never point it at a wonder box. + +Run 2026-11-12 (ScadaBridge#125) with ScadaBridge `94ae1860` (SDK client +1.5.378.106), ACL off, against the SDK 1.5.378.182 baseline (`83ed880e`) and +master `a28adba2` (SDK 2.0.1): matrix 11/11 in None, Sign and SignAndEncrypt on +both, all 8 trust controls failed closed (0x80130000 / 0x801A0000); failover 8/8 +legs (11/11 cases each). Loss-to-backup timings are the same on both versions: +`kill` about 20 s (three 5 s reconnect attempts, then the backup), `freeze` about +20 s to detect and 86–130 s to reach the backup, because each reconnect attempt +waits out a connect timeout against the silent primary. `rig-failover.sh` passed +9/9 against the docker-dev pair (image SDK 2.0.1), central-1 killed, about 20 s +to central-2. diff --git a/scripts/validation/scadabridge-sdk-compatibility/rig-failover.sh b/scripts/validation/scadabridge-sdk-compatibility/rig-failover.sh new file mode 100755 index 000000000..63070a78f --- /dev/null +++ b/scripts/validation/scadabridge-sdk-compatibility/rig-failover.sh @@ -0,0 +1,46 @@ +#!/usr/bin/env bash +# ScadaBridge#125 — the failover probe against the REAL docker-dev OtOpcUa pair (central-1 :4840 / +# central-2 :4841), not the in-process fixture. ScadaBridge's DataConnectionActor connects to central-1 +# as primary and central-2 as backup; once it reports a steady primary, central-1's container is killed +# (docker kill = abrupt loss) and the probe must land on central-2 with Good values and a fresh +# ConditionRefresh. central-1 is started again afterwards. LOCAL docker-dev only — never a wonder box. +# +# Usage: rig-failover.sh +# Anonymous (OpcUa:AllowAnonymous=true on the rig) cannot write there, so writes are off; the rig deploys +# no active native alarm, so the alarm check is "ConditionRefresh completes on each node" only. +set -euo pipefail +probe=$1; out=$2 +primary_container=${RIG_PRIMARY_CONTAINER:-otopcua-dev-central-1-1} +backup_container=${RIG_BACKUP_CONTAINER:-otopcua-dev-central-2-1} +mkdir -p "$out" +for c in "$primary_container" "$backup_container"; do + mkdir -p "$out/server-certs/$c" + docker cp "$c:/app/pki/own/certs/." "$out/server-certs/$c/" +done +export SCADABRIDGE_PROBE_ANONYMOUS=1 +export SCADABRIDGE_PROBE_SERVER_CERTS="$out/server-certs/$primary_container:$out/server-certs/$backup_container" +export SCADABRIDGE_PROBE_FAILOVER_TAGS=${RIG_TAGS:-"nsu=https://zb.com/otopcua/raw;s=int3b-mb-01/plc/HR100,nsu=https://zb.com/otopcua/uns;s=a/l/e1/HR5"} +export SCADABRIDGE_PROBE_FAILOVER_WRITE_TAG=none +export SCADABRIDGE_PROBE_FAILOVER_BACKUP_VALUE=none +export SCADABRIDGE_PROBE_FAILOVER_ALARM_PREFIX=none +dotnet "$probe" failover opc.tcp://localhost:${RIG_PRIMARY_PORT:-4840} opc.tcp://localhost:${RIG_BACKUP_PORT:-4841} None "$out/consumer" \ + > "$out/consumer.log" 2>&1 & +pid=$! +for _ in $(seq 1 1500); do + [[ -f "$out/consumer/primary-ready" ]] && break + kill -0 "$pid" 2>/dev/null || break + sleep 0.1 +done +status=0 +if [[ -f "$out/consumer/primary-ready" ]]; then + python3 -c 'from datetime import datetime, timezone; print(datetime.now(timezone.utc).isoformat())' > "$out/consumer/primary-killed.tmp" + docker kill "$primary_container" > /dev/null + mv "$out/consumer/primary-killed.tmp" "$out/consumer/primary-killed" + wait "$pid" || status=$? + docker start "$primary_container" > /dev/null +else + wait "$pid" || status=$? +fi +docker inspect "$primary_container" "$backup_container" --format '{{.Name}} {{.Config.Image}} {{.Image}} {{.State.Status}}' > "$out/containers.txt" +echo "probe exit $status; result $out/consumer/result.json" +exit "$status" diff --git a/scripts/validation/scadabridge-sdk-compatibility/run.py b/scripts/validation/scadabridge-sdk-compatibility/run.py index fc6cac63f..1de2e7e45 100644 --- a/scripts/validation/scadabridge-sdk-compatibility/run.py +++ b/scripts/validation/scadabridge-sdk-compatibility/run.py @@ -1,25 +1,169 @@ #!/usr/bin/env python3 """Run the real ScadaBridge consumer in separate processes against both application versions.""" import argparse +from datetime import datetime, timezone import hashlib import json import os from pathlib import Path import secrets +import signal import subprocess import time +SECURITY = {"None": "none", "Sign": "sign", "SignAndEncrypt": None} + + +def leg_env(mode): + env = os.environ.copy() + env["OPCUA_MATRIX_PASSWORD"] = "otua_matrix_" + secrets.token_urlsafe(32) + for name in ("SCADABRIDGE_PROBE_PKI", "SCADABRIDGE_PROBE_SERVER_CERTS", "OPCUA_MATRIX_SERVER_SECURITY"): + env.pop(name, None) + if SECURITY[mode]: + env["OPCUA_MATRIX_SERVER_SECURITY"] = SECURITY[mode] + return env + + +def prepare_identity(artifacts, env, leg, trusting_server_roots, result): + """Provision the consumer identity through ScadaBridge's own code; trust its PUBLIC cert on the given servers.""" + pki = leg / "consumer-pki" + der = leg / "consumer-identity.der" + with (leg / "prepare.log").open("w") as log: + prepared = subprocess.run(["dotnet", str(artifacts["probe"]), "prepare", str(pki), str(der)], + env=env, stdout=log, stderr=subprocess.STDOUT, timeout=120) + if prepared.returncode != 0: + raise RuntimeError(f"Consumer identity provisioning failed; inspect {leg / 'prepare.log'}") + env["SCADABRIDGE_PROBE_PKI"] = str(pki) + for server_root in trusting_server_roots: + trusted = server_root / "server-pki" / "trusted" / "certs" + trusted.mkdir(parents=True) + (trusted / "scadabridge-consumer.der").write_bytes(der.read_bytes()) + result["consumerIdentitySha256"] = hashlib.sha256(der.read_bytes()).hexdigest() + + +def start_server(host, server_root, env, log_path): + command = ["dotnet", str(host), "server", str(server_root)] + log = log_path.open("w") + server = subprocess.Popen(command, env=env, stdout=log, stderr=subprocess.STDOUT) + until = time.monotonic() + 40 + while not (server_root / "endpoint.txt").is_file(): + if server.poll() is not None or time.monotonic() >= until: + raise RuntimeError(f"Server failed to become ready; inspect {log_path}") + time.sleep(0.1) + return server, command, (server_root / "endpoint.txt").read_text().strip() + + +def stop_server(server, server_root): + if server.poll() is not None: + return server.returncode + server_root.mkdir(exist_ok=True) + (server_root / "stop").touch() + try: + server.wait(timeout=15) + except subprocess.TimeoutExpired: + server.terminate() + server.wait(timeout=10) + return server.returncode + + +def matrix_leg(version, mode, trust, artifacts, root): + leg = root / (f"{version}-{mode}" if trust == "trusted" else f"{version}-{mode}-{trust}") + leg.mkdir() + server_root = leg / "server" + env = leg_env(mode) + result = {"suite": "matrix", "server": version, "mode": mode, "trust": trust, + "expect": "pass" if trust == "trusted" else "fail"} + if mode != "None": + prepare_identity(artifacts, env, leg, [] if trust == "untrusted-client" else [server_root], result) + server, result["serverCommand"], endpoint = start_server(artifacts[f"{version}_host"], server_root, env, leg / "server.log") + try: + if trust != "untrusted-server": + env["SCADABRIDGE_PROBE_SERVER_CERTS"] = str(server_root / "server-pki" / "own" / "certs") + command = ["dotnet", str(artifacts["probe"]), endpoint, mode, str(leg / "consumer")] + with (leg / "consumer.log").open("w") as log: + probe = subprocess.run(command, env=env, stdout=log, stderr=subprocess.STDOUT, timeout=150) + result.update(command=command, exit=probe.returncode, result=str(leg / "consumer" / "result.json")) + finally: + result["serverExit"] = stop_server(server, server_root) + result["verdict"] = "passed" if result["serverExit"] == 0 and (result["exit"] == 0) == (result["expect"] == "pass") else "failed" + return result + + +def failover_leg(version, mode, loss, artifacts, root): + """ScadaBridge#125: two independent servers; the consumer's DataConnectionActor fails over when the primary is lost. + + loss=kill SIGKILL: the OS closes the primary's sockets, so the client sees the loss at once. + loss=freeze SIGSTOP: the primary's sockets stay open and silent (a hung host / dead link), so only the + consumer's session keep-alive can detect it. + """ + leg = root / f"{version}-failover-{mode}-{loss}" + leg.mkdir() + roots = {"primary": leg / "primary", "backup": leg / "backup"} + env = leg_env(mode) + result = {"suite": "failover", "server": version, "mode": mode, "loss": loss, "trust": "trusted", "expect": "pass"} + if mode != "None": + prepare_identity(artifacts, env, leg, list(roots.values()), result) + servers, endpoints = {}, {} + try: + for name, server_root in roots.items(): + servers[name], result[f"{name}Command"], endpoints[name] = start_server( + artifacts[f"{version}_host"], server_root, env, leg / f"{name}.log") + # Even a None-mode client validates the server certificate the endpoint hands it. + env["SCADABRIDGE_PROBE_SERVER_CERTS"] = os.pathsep.join( + str(r / "server-pki" / "own" / "certs") for r in roots.values()) + consumer = leg / "consumer" + command = ["dotnet", str(artifacts["probe"]), "failover", endpoints["primary"], endpoints["backup"], mode, str(consumer)] + result["command"] = command + with (leg / "consumer.log").open("w") as log: + probe = subprocess.Popen(command, env=env, stdout=log, stderr=subprocess.STDOUT) + until = time.monotonic() + 150 + while not (consumer / "primary-ready").is_file() and probe.poll() is None and time.monotonic() < until: + time.sleep(0.1) + if (consumer / "primary-ready").is_file(): + killed_at = datetime.now(timezone.utc).isoformat() # taken BEFORE the signal + if loss == "kill": + servers["primary"].kill() # abrupt loss: no graceful session close + servers["primary"].wait(timeout=10) + else: + os.kill(servers["primary"].pid, signal.SIGSTOP) + (consumer / "primary-killed").write_text(killed_at) + result["primaryKilledAt"] = killed_at + probe.wait(timeout=480) + result.update(exit=probe.returncode, result=str(consumer / "result.json")) + finally: + if loss == "freeze" and "primary" in servers and servers["primary"].poll() is None: + servers["primary"].kill() # a frozen process cannot honour the stop file + servers["primary"].wait(timeout=10) + for name, server in servers.items(): + result[f"{name}Exit"] = stop_server(server, roots[name]) + killed = "primaryKilledAt" in result + result["serverExit"] = result.get("backupExit") + result["verdict"] = "passed" if killed and result.get("exit") == 0 and result.get("backupExit") == 0 else "failed" + return result + def main(): parser = argparse.ArgumentParser(description=__doc__) - parser.add_argument("--baseline-host", type=Path, required=True) - parser.add_argument("--candidate-host", type=Path, required=True) + parser.add_argument("--baseline-host", type=Path) + parser.add_argument("--candidate-host", type=Path) parser.add_argument("--probe", type=Path, required=True) parser.add_argument("--results", type=Path, required=True) + parser.add_argument("--versions", default="baseline,candidate", + help="comma list of baseline/candidate (each needs its ---host)") + parser.add_argument("--suites", default="matrix,failover", help="comma list of matrix/failover") + parser.add_argument("--failover-modes", default="None,SignAndEncrypt") + parser.add_argument("--failover-loss", default="kill,freeze", help="comma list of kill/freeze") args = parser.parse_args() + versions = [v for v in args.versions.split(",") if v] + suites = [s for s in args.suites.split(",") if s] root = args.results.resolve() root.mkdir(parents=True, exist_ok=False) - artifacts = {name: path.resolve() for name, path in vars(args).items() if name != "results"} + artifacts = {"probe": args.probe.resolve()} + for version in versions: + host = getattr(args, f"{version}_host") + if host is None: + raise SystemExit(f"--{version}-host is required for version {version}") + artifacts[f"{version}_host"] = host.resolve() for path in artifacts.values(): if not path.is_file(): raise FileNotFoundError(path) @@ -34,69 +178,18 @@ def main(): # untrusted-server - the consumer's trusted store does not hold the server certificate legs = [(m, "trusted") for m in ("None", "Sign", "SignAndEncrypt")] legs += [(m, n) for m in ("Sign", "SignAndEncrypt") for n in ("untrusted-client", "untrusted-server")] - for version in ("baseline", "candidate"): - for mode, trust in legs: - leg = root / (f"{version}-{mode}" if trust == "trusted" else f"{version}-{mode}-{trust}") - leg.mkdir() - server_root = leg / "server" - env = os.environ.copy() - env["OPCUA_MATRIX_PASSWORD"] = "otua_matrix_" + secrets.token_urlsafe(32) - env.pop("SCADABRIDGE_PROBE_PKI", None) - env.pop("SCADABRIDGE_PROBE_SERVER_CERTS", None) - if mode == "None": - env["OPCUA_MATRIX_SERVER_SECURITY"] = "none" - elif mode == "Sign": - env["OPCUA_MATRIX_SERVER_SECURITY"] = "sign" - else: - env.pop("OPCUA_MATRIX_SERVER_SECURITY", None) - server_command = ["dotnet", str(artifacts[f"{version}_host"]), "server", str(server_root)] - result = {"server": version, "mode": mode, "trust": trust, "expect": "pass" if trust == "trusted" else "fail", - "serverCommand": server_command} - if mode != "None": - pki = leg / "consumer-pki" - der = leg / "consumer-identity.der" - with (leg / "prepare.log").open("w") as log: - prepared = subprocess.run(["dotnet", str(artifacts["probe"]), "prepare", str(pki), str(der)], - env=env, stdout=log, stderr=subprocess.STDOUT, timeout=120) - if prepared.returncode != 0: - raise RuntimeError(f"Consumer identity provisioning failed; inspect {leg / 'prepare.log'}") - env["SCADABRIDGE_PROBE_PKI"] = str(pki) - if trust != "untrusted-client": - trusted = server_root / "server-pki" / "trusted" / "certs" - trusted.mkdir(parents=True) - (trusted / "scadabridge-consumer.der").write_bytes(der.read_bytes()) - result["consumerIdentitySha256"] = hashlib.sha256(der.read_bytes()).hexdigest() - with (leg / "server.log").open("w") as server_log: - server = subprocess.Popen(server_command, env=env, stdout=server_log, stderr=subprocess.STDOUT) - try: - until = time.monotonic() + 40 - while not (server_root / "endpoint.txt").is_file(): - if server.poll() is not None or time.monotonic() >= until: - raise RuntimeError(f"Server failed to become ready; inspect {leg / 'server.log'}") - time.sleep(0.1) - endpoint = (server_root / "endpoint.txt").read_text().strip() - if trust != "untrusted-server": - env["SCADABRIDGE_PROBE_SERVER_CERTS"] = str(server_root / "server-pki" / "own" / "certs") - command = ["dotnet", str(artifacts["probe"]), endpoint, mode, str(leg / "consumer")] - with (leg / "consumer.log").open("w") as log: - probe = subprocess.run(command, env=env, stdout=log, stderr=subprocess.STDOUT, timeout=150) - result.update(command=command, exit=probe.returncode, - result=str(leg / "consumer" / "result.json")) - finally: - server_root.mkdir(exist_ok=True) - (server_root / "stop").touch() - try: - server.wait(timeout=15) - except subprocess.TimeoutExpired: - server.terminate() - server.wait(timeout=10) - result["serverExit"] = server.returncode - receipt["legs"].append(result) - (root / "receipt.json").write_text(json.dumps(receipt, indent=2)) - # A trusted leg passes only on exit 0; a negative control passes only when the consumer fails closed. - for leg in receipt["legs"]: - leg["verdict"] = "passed" if leg["serverExit"] == 0 and (leg["exit"] == 0) == (leg["expect"] == "pass") else "failed" - (root / "receipt.json").write_text(json.dumps(receipt, indent=2)) + for version in versions: + if "matrix" in suites: + for mode, trust in legs: + receipt["legs"].append(matrix_leg(version, mode, trust, artifacts, root)) + (root / "receipt.json").write_text(json.dumps(receipt, indent=2)) + if "failover" in suites: + for mode in args.failover_modes.split(","): + for loss in args.failover_loss.split(","): + receipt["legs"].append(failover_leg(version, mode, loss, artifacts, root)) + (root / "receipt.json").write_text(json.dumps(receipt, indent=2)) + # A trusted leg passes only on exit 0; a negative control passes only when the consumer fails closed; + # a failover leg passes only when the primary was really killed and the consumer passed on the backup. return int(any(leg["verdict"] != "passed" for leg in receipt["legs"]))