lmxopcua/scripts/validation/scadabridge-sdk-compatibility/Program.cs

using System.Collections.Concurrent;
using System.Diagnostics;
using System.Text.Json;
using Microsoft.Extensions.Logging.Abstractions;
using ZB.MOM.WW.ScadaBridge.Commons.Interfaces.Protocol;
using ZB.MOM.WW.ScadaBridge.Commons.Serialization;
using ZB.MOM.WW.ScadaBridge.Commons.Types.Alarms;
using ZB.MOM.WW.ScadaBridge.Commons.Types.DataConnections;
using ZB.MOM.WW.ScadaBridge.Commons.Types.Enums;
using ZB.MOM.WW.ScadaBridge.DataConnectionLayer;
using ZB.MOM.WW.ScadaBridge.DataConnectionLayer.Adapters;

// ScadaBridge#121: the consumer now owns a configurable application identity. "prepare" provisions it
// through the consumer's own EnsureApplicationCertificateAsync (the same reuse/creation path a secured
// connect uses) and exports only the PUBLIC DER, which the runner places in the server's trusted store.
static OpcUaGlobalOptions IdentityOptions(string pki) => new()
{
    ApplicationName = "ScadaBridge-SDK2-compatibility",
    ApplicationUri = "urn:scadabridge-sdk2-compatibility:ScadaBridge-DCL",
    ApplicationCertificateSubject = "CN=ScadaBridge-SDK2-compatibility",
    OwnCertificateStorePath = Path.Combine(pki, "own"),
    AutoCreateApplicationCertificate = true,
    TrustedIssuerStorePath = Path.Combine(pki, "issuers"),
    TrustedPeerStorePath = Path.Combine(pki, "trusted"),
    RejectedCertificateStorePath = Path.Combine(pki, "rejected"),
};
if (args.Length == 3 && args[0] == "prepare")
{
    var pkiRoot = Path.GetFullPath(args[1]);
    if (Directory.Exists(pkiRoot)) throw new IOException("Refusing to reuse a PKI directory.");
    using var publicCert = await OpcUaApplicationIdentity.EnsureApplicationCertificateAsync(IdentityOptions(pkiRoot), NullLogger.Instance);
    Require(!publicCert.HasPrivateKey, "Exported identity must be public-only");
    File.WriteAllBytes(Path.GetFullPath(args[2]), publicCert.RawData);
    Console.WriteLine($"Consumer identity {publicCert.Thumbprint} {publicCert.Subject}");
    return 0;
}
// ScadaBridge#125: "failover <primary> <backup> <mode> <dir>" 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 <endpoint> <None|Sign|SignAndEncrypt> <new-result-directory> | Probe failover <primary> <backup> <mode> <new-result-directory> | Probe prepare <new-pki-directory> <public-der-out>");
var endpoint = failoverMode ? args[1] : args[0];
var mode = Enum.Parse<OpcUaSecurityMode>(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); }
using var deadline = new CancellationTokenSource(TimeSpan.FromMinutes(2));
var ct = deadline.Token;
var cases = new List<object>();
var failures = 0;
// A prepared identity (SCADABRIDGE_PROBE_PKI) is reused; otherwise a fresh task-local store.
var pki = Environment.GetEnvironmentVariable("SCADABRIDGE_PROBE_PKI") is { Length: > 0 } prepared ? Path.GetFullPath(prepared) : Path.Combine(root, "pki");
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 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<OpcUaDataConnection>.Instance);
var config = ConfigFor(endpoint);
async Task Check(string name, Func<Task> 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 Wait(Func<bool> predicate)
{
    for (var i = 0; i < 200; i++) { if (predicate()) return; await Task.Delay(50, ct); }
    Require(false, "No expected consumer notification within 10 seconds.");
}
const string raw = "nsu=https://zb.com/otopcua/raw;s=Matrix/dev/";
const string uns = "nsu=https://zb.com/otopcua/uns;s=Eq/";
await Check("actual-adapter-connect-api-key", () => adapter.ConnectAsync(OpcUaEndpointConfigSerializer.ToFlatDict(config), ct));
if (failures == 0)
{
    foreach (var prefix in new[] { raw, uns })
    {
        await Check($"scalar-round-trip-{prefix}", async () =>
        {
            var write = await adapter.WriteAsync(prefix + "Scalar", 412, ct);
            Require(write.Success, write.ErrorMessage ?? "Scalar write denied");
            var read = await adapter.ReadAsync(prefix + "Scalar", ct);
            Require(read.Success && read.Value?.Value is int i && i == 412, "Scalar value/type changed");
        });
        await Check($"array-round-trip-{prefix}", async () =>
        {
            var write = await adapter.WriteAsync(prefix + "Vector", new[] { 31, 32, 33 }, ct);
            Require(write.Success, write.ErrorMessage ?? "Array write denied");
            var read = await adapter.ReadAsync(prefix + "Vector", ct);
            Require(read.Success && read.Value?.Value is int[] a && a.SequenceEqual(new[] { 31, 32, 33 }), "Array value/type changed");
        });
    }
    await Check("actual-adapter-durable-browse", async () =>
    {
        var page = await adapter.BrowseChildrenAsync("nsu=https://zb.com/otopcua/raw;s=Matrix/dev", cancellationToken: ct);
        Require(page.Children.Count(x => x.DisplayName is "Scalar" or "Vector") == 2, "Missing durable raw variable children");
        Require(page.Children.Any(x => x.DisplayName == "Alarm"), "Missing native alarm under its raw parent");
        var projected = await adapter.BrowseChildrenAsync("nsu=https://zb.com/otopcua/uns;s=Eq", cancellationToken: ct);
        Require(projected.Children.Any(x => x.DisplayName == "Vector"), "Missing UNS variable child");
    });
    await Check("actual-adapter-monitor-quality-time", async () =>
    {
        var values = new ConcurrentQueue<TagValue>();
        var handle = await adapter.SubscribeAsync(raw + "Scalar", (_, value) => values.Enqueue(value), ct);
        Require((await adapter.WriteAsync(raw + "Scalar", 713, ct)).Success, "Monitored write denied");
        await Wait(() => values.Any(x => x.Value is int i && i == 713));
        var actual = values.Last(x => x.Value is int i && i == 713);
        Require(actual.Timestamp > new DateTimeOffset(2026, 10, 1, 0, 0, 0, TimeSpan.Zero), "Missing source timestamp");
        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<TagValue>();
        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<NativeAlarmTransition>();
        var handle = await adapter.SubscribeAlarmsAsync("i=2253", null, alarms.Enqueue, ct);
        await Wait(() => alarms.Any(x => x.Kind == AlarmTransitionKind.SnapshotComplete)
            && alarms.Any(x => x.Message == "matrix-alarm" && x.SourceReference.Contains("Alarm", StringComparison.Ordinal)));
        await adapter.UnsubscribeAlarmsAsync(handle, ct);
    });
    await Check("actual-adapter-disconnect-reconnect", async () =>
    {
        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<NativeAlarmTransition>();
        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);
    });
}
var sdkAssembly = typeof(Opc.Ua.NodeId).Assembly;
var result = new
{
    consumer = typeof(OpcUaDataConnection).Assembly.GetName().Version?.ToString(),
    sdk = FileVersionInfo.GetVersionInfo(sdkAssembly.Location).ProductVersion,
    process = Environment.ProcessId, mode = mode.ToString(), autoAccept = false,
    endpoint, cases, failures,
    applicationUri = options.ApplicationUri,
    ownCertificatesBefore = ownBefore.Select(Path.GetFileName),
    ownCertificatesAfter = Directory.Exists(Path.Combine(pki, "own", "certs")) ? Directory.GetFiles(Path.Combine(pki, "own", "certs")).Order().Select(Path.GetFileName).ToArray() : [],
    notImplementedByConsumer = new[] { "HistoryRead/CP", "Part9 acknowledgement methods" },
};
File.WriteAllText(Path.Combine(root, "result.json"), JsonSerializer.Serialize(result, new JsonSerializerOptions { WriteIndented = true }));
Console.WriteLine($"Consumer probe: {cases.Count - failures} passed, {failures} failed; {Path.Combine(root, "result.json")}");
return failures == 0 ? 0 : 1;