test: retry OPC UA host start on a port bind race (TOCTOU in AllocateFreePort)
The tests probe a free port (bind 0, release) and start the SDK server later; under a highly parallel run another test can take the port in between and the SDK fails with 'Failed to establish tcp listener sockets for Ipv4 and IPv6.'. OpcUaHostStart.StartAsync builds a fresh host + server on a fresh port per attempt (up to 5) when the failure is a bind failure; all 45 start sites in OpcUaServer.Tests, OpcUaServer.IntegrationTests and Runtime.Tests use it. Per-attempt GUID ApplicationUris are hoisted out of the host factory because a changed URI against a reused PKI store fails with 'Cannot access private key'. Claude-Session: https://claude.ai/code/session_01Ef2ozxS7QPanGT8uoATbTe
This commit is contained in:
Joseph Doherty
+130
-135
@@ -303,146 +303,141 @@ public sealed class AclSubscriptionEnforcementTests
|
||||
public static async Task<Rig> StartAsync(string name, CancellationToken ct)
|
||||
{
|
||||
var rig = new Rig(Path.Combine(Path.GetTempPath(), $"otopcua-{name}-{Guid.NewGuid():N}"));
|
||||
rig._port = AllocateFreePort();
|
||||
rig._server = new OtOpcUaSdkServer();
|
||||
rig._host = new OpcUaApplicationHost(
|
||||
new OpcUaApplicationHostOptions
|
||||
var applicationUri = $"urn:OtOpcUa.{name}.{Guid.NewGuid():N}";
|
||||
(rig._host, rig._server) = await OpcUaHostStart.StartAsync(
|
||||
port =>
|
||||
{
|
||||
ApplicationName = $"OtOpcUa.{name}",
|
||||
ApplicationUri = $"urn:OtOpcUa.{name}.{Guid.NewGuid():N}",
|
||||
OpcUaPort = rig._port,
|
||||
PublicHostname = "127.0.0.1",
|
||||
PkiStoreRoot = rig._pkiRoot,
|
||||
EnabledSecurityProfiles = [OpcUaSecurityProfile.None],
|
||||
AutoAcceptUntrustedClientCertificates = true,
|
||||
},
|
||||
NullLogger<OpcUaApplicationHost>.Instance);
|
||||
|
||||
await rig._host.StartAsync(rig._server, ct);
|
||||
return rig;
|
||||
}
|
||||
|
||||
/// <summary>Switches enforcement on with one grant, returning the state so a test can redeploy.</summary>
|
||||
public DataPlaneAuthorityState Enforce(
|
||||
NodeAcl grant, IReadOnlyDictionary<AclScopeRef, IReadOnlyList<NodeAclPath>>? scopes = null)
|
||||
{
|
||||
var state = new DataPlaneAuthorityState();
|
||||
state.Apply(Cluster, "rev-1", [], [grant], scopes ?? NoScopes);
|
||||
|
||||
// BOTH must be set: Authorizer decides, and DataPlaneAuthority is what carries the Changed
|
||||
// signal that reaches live subscriptions. Wiring only the first passes every test that makes a
|
||||
// request and none that revokes.
|
||||
Nm.DataPlaneAuthority = state;
|
||||
Nm.Authorizer = new DataPlaneAuthorizer(
|
||||
state, new TriePermissionEvaluator(state.Tries), enabled: true, NullLogger.Instance);
|
||||
return state;
|
||||
}
|
||||
|
||||
public void Materialise()
|
||||
{
|
||||
Nm.EnsureFolder("Plant", null, "Plant", AddressSpaceRealm.Raw);
|
||||
Nm.EnsureFolder("Plant/Modbus", "Plant", "Modbus", AddressSpaceRealm.Raw);
|
||||
Nm.EnsureFolder(Device, "Plant/Modbus", "dev1", AddressSpaceRealm.Raw);
|
||||
Nm.EnsureVariable(Tag, Device, "Speed", "Float", writable: false, AddressSpaceRealm.Raw);
|
||||
}
|
||||
|
||||
public void MaterialiseTwoDevices()
|
||||
{
|
||||
Materialise();
|
||||
Nm.EnsureFolder("Plant/Modbus/dev9", "Plant/Modbus", "dev9", AddressSpaceRealm.Raw);
|
||||
Nm.EnsureVariable(
|
||||
"Plant/Modbus/dev9/Torque", "Plant/Modbus/dev9", "Torque", "Float",
|
||||
writable: false, AddressSpaceRealm.Raw);
|
||||
}
|
||||
|
||||
public void Write(float value) =>
|
||||
Nm.WriteValue(Tag, value, OpcUaStatus.Good, DateTime.UtcNow, AddressSpaceRealm.Raw);
|
||||
|
||||
public ushort RawNamespace(ISession session)
|
||||
{
|
||||
var ns = (ushort)session.NamespaceUris.GetIndex(V3NodeIds.RawNamespaceUri);
|
||||
ns.ShouldBeGreaterThan((ushort)0);
|
||||
return ns;
|
||||
}
|
||||
|
||||
public NodeId TagNodeId(ISession session) => new(Tag, RawNamespace(session));
|
||||
|
||||
public async Task<(Subscription Subscription, List<DataValue> Received)> SubscribeAsync(
|
||||
ISession session, CancellationToken ct, bool expectFailure = false)
|
||||
{
|
||||
var received = new List<DataValue>();
|
||||
var gate = new object();
|
||||
|
||||
var subscription = new Subscription(session.DefaultSubscription) { PublishingInterval = 100 };
|
||||
session.AddSubscription(subscription);
|
||||
await subscription.CreateAsync(ct);
|
||||
|
||||
var item = new MonitoredItem(subscription.DefaultItem)
|
||||
rig._port = port;
|
||||
return new OpcUaApplicationHost(
|
||||
new OpcUaApplicationHostOptions
|
||||
{
|
||||
ApplicationName = $"OtOpcUa.{name}",
|
||||
ApplicationUri = applicationUri,
|
||||
OpcUaPort = port,
|
||||
PublicHostname = "127.0.0.1",
|
||||
PkiStoreRoot = rig._pkiRoot,
|
||||
EnabledSecurityProfiles = [OpcUaSecurityProfile.None],
|
||||
AutoAcceptUntrustedClientCertificates = true,
|
||||
},
|
||||
NullLogger<OpcUaApplicationHost>.Instance);
|
||||
},
|
||||
() => new OtOpcUaSdkServer(), ct);
|
||||
return rig;
|
||||
}
|
||||
|
||||
/// <summary>Switches enforcement on with one grant, returning the state so a test can redeploy.</summary>
|
||||
public DataPlaneAuthorityState Enforce(
|
||||
NodeAcl grant, IReadOnlyDictionary<AclScopeRef, IReadOnlyList<NodeAclPath>>? scopes = null)
|
||||
{
|
||||
StartNodeId = TagNodeId(session),
|
||||
AttributeId = Attributes.Value,
|
||||
SamplingInterval = 50,
|
||||
QueueSize = 10,
|
||||
};
|
||||
item.Notification += (mi, _) =>
|
||||
var state = new DataPlaneAuthorityState();
|
||||
state.Apply(Cluster, "rev-1", [], [grant], scopes ?? NoScopes);
|
||||
|
||||
// BOTH must be set: Authorizer decides, and DataPlaneAuthority is what carries the Changed
|
||||
// signal that reaches live subscriptions. Wiring only the first passes every test that makes a
|
||||
// request and none that revokes.
|
||||
Nm.DataPlaneAuthority = state;
|
||||
Nm.Authorizer = new DataPlaneAuthorizer(
|
||||
state, new TriePermissionEvaluator(state.Tries), enabled: true, NullLogger.Instance);
|
||||
return state;
|
||||
}
|
||||
|
||||
public void Materialise()
|
||||
{
|
||||
foreach (var v in mi.DequeueValues()) lock (gate) received.Add(v);
|
||||
};
|
||||
subscription.AddItem(item);
|
||||
await subscription.ApplyChangesAsync(ct);
|
||||
|
||||
if (!expectFailure) item.Status.Error.ShouldBeNull();
|
||||
|
||||
// The list is only ever appended under `gate` and read from the test thread between awaits;
|
||||
// a snapshot-on-read wrapper would hide ordering rather than help.
|
||||
return (subscription, received);
|
||||
}
|
||||
|
||||
public async Task<ISession> ConnectAsync(CancellationToken ct)
|
||||
{
|
||||
var appConfig = new ApplicationConfiguration
|
||||
Nm.EnsureFolder("Plant", null, "Plant", AddressSpaceRealm.Raw);
|
||||
Nm.EnsureFolder("Plant/Modbus", "Plant", "Modbus", AddressSpaceRealm.Raw);
|
||||
Nm.EnsureFolder(Device, "Plant/Modbus", "dev1", AddressSpaceRealm.Raw);
|
||||
Nm.EnsureVariable(Tag, Device, "Speed", "Float", writable: false, AddressSpaceRealm.Raw);
|
||||
}
|
||||
|
||||
public void MaterialiseTwoDevices()
|
||||
{
|
||||
ApplicationName = "OtOpcUa.AclClient",
|
||||
ApplicationUri = $"urn:OtOpcUa.AclClient.{Guid.NewGuid():N}",
|
||||
ApplicationType = ApplicationType.Client,
|
||||
SecurityConfiguration = TestClientSecurity.Build(TestClientSecurity.AllocatePkiRoot()),
|
||||
ClientConfiguration = new ClientConfiguration { DefaultSessionTimeout = 60_000 },
|
||||
};
|
||||
await appConfig.ValidateAsync(ApplicationType.Client, ct);
|
||||
appConfig.CertificateValidator.CertificateValidation += (_, e) => e.Accept = true;
|
||||
|
||||
var endpoint = await CoreClientUtils.SelectEndpointAsync(
|
||||
appConfig, $"opc.tcp://127.0.0.1:{_port}/OtOpcUa", false, DefaultTelemetry.Create(_ => { }), ct);
|
||||
|
||||
return await new DefaultSessionFactory(DefaultTelemetry.Create(_ => { })).CreateAsync(
|
||||
appConfig,
|
||||
new ConfiguredEndpoint(null, endpoint, EndpointConfiguration.Create(appConfig)),
|
||||
updateBeforeConnect: false,
|
||||
checkDomain: false,
|
||||
sessionName: "AclSubscriptionEnforcementTests",
|
||||
sessionTimeout: 60_000,
|
||||
identity: new UserIdentity(new AnonymousIdentityToken()),
|
||||
preferredLocales: null,
|
||||
ct: ct);
|
||||
}
|
||||
|
||||
public async ValueTask DisposeAsync()
|
||||
{
|
||||
await _host.DisposeAsync();
|
||||
if (Directory.Exists(_pkiRoot))
|
||||
Materialise();
|
||||
Nm.EnsureFolder("Plant/Modbus/dev9", "Plant/Modbus", "dev9", AddressSpaceRealm.Raw);
|
||||
Nm.EnsureVariable(
|
||||
"Plant/Modbus/dev9/Torque", "Plant/Modbus/dev9", "Torque", "Float",
|
||||
writable: false, AddressSpaceRealm.Raw);
|
||||
}
|
||||
|
||||
public void Write(float value) =>
|
||||
Nm.WriteValue(Tag, value, OpcUaStatus.Good, DateTime.UtcNow, AddressSpaceRealm.Raw);
|
||||
|
||||
public ushort RawNamespace(ISession session)
|
||||
{
|
||||
try { Directory.Delete(_pkiRoot, recursive: true); }
|
||||
catch (IOException) { /* best effort */ }
|
||||
var ns = (ushort)session.NamespaceUris.GetIndex(V3NodeIds.RawNamespaceUri);
|
||||
ns.ShouldBeGreaterThan((ushort)0);
|
||||
return ns;
|
||||
}
|
||||
|
||||
public NodeId TagNodeId(ISession session) => new(Tag, RawNamespace(session));
|
||||
|
||||
public async Task<(Subscription Subscription, List<DataValue> Received)> SubscribeAsync(
|
||||
ISession session, CancellationToken ct, bool expectFailure = false)
|
||||
{
|
||||
var received = new List<DataValue>();
|
||||
var gate = new object();
|
||||
|
||||
var subscription = new Subscription(session.DefaultSubscription) { PublishingInterval = 100 };
|
||||
session.AddSubscription(subscription);
|
||||
await subscription.CreateAsync(ct);
|
||||
|
||||
var item = new MonitoredItem(subscription.DefaultItem)
|
||||
{
|
||||
StartNodeId = TagNodeId(session),
|
||||
AttributeId = Attributes.Value,
|
||||
SamplingInterval = 50,
|
||||
QueueSize = 10,
|
||||
};
|
||||
item.Notification += (mi, _) =>
|
||||
{
|
||||
foreach (var v in mi.DequeueValues()) lock (gate) received.Add(v);
|
||||
};
|
||||
subscription.AddItem(item);
|
||||
await subscription.ApplyChangesAsync(ct);
|
||||
|
||||
if (!expectFailure) item.Status.Error.ShouldBeNull();
|
||||
|
||||
// The list is only ever appended under `gate` and read from the test thread between awaits;
|
||||
// a snapshot-on-read wrapper would hide ordering rather than help.
|
||||
return (subscription, received);
|
||||
}
|
||||
|
||||
public async Task<ISession> ConnectAsync(CancellationToken ct)
|
||||
{
|
||||
var appConfig = new ApplicationConfiguration
|
||||
{
|
||||
ApplicationName = "OtOpcUa.AclClient",
|
||||
ApplicationUri = $"urn:OtOpcUa.AclClient.{Guid.NewGuid():N}",
|
||||
ApplicationType = ApplicationType.Client,
|
||||
SecurityConfiguration = TestClientSecurity.Build(TestClientSecurity.AllocatePkiRoot()),
|
||||
ClientConfiguration = new ClientConfiguration { DefaultSessionTimeout = 60_000 },
|
||||
};
|
||||
await appConfig.ValidateAsync(ApplicationType.Client, ct);
|
||||
appConfig.CertificateValidator.CertificateValidation += (_, e) => e.Accept = true;
|
||||
|
||||
var endpoint = await CoreClientUtils.SelectEndpointAsync(
|
||||
appConfig, $"opc.tcp://127.0.0.1:{_port}/OtOpcUa", false, DefaultTelemetry.Create(_ => { }), ct);
|
||||
|
||||
return await new DefaultSessionFactory(DefaultTelemetry.Create(_ => { })).CreateAsync(
|
||||
appConfig,
|
||||
new ConfiguredEndpoint(null, endpoint, EndpointConfiguration.Create(appConfig)),
|
||||
updateBeforeConnect: false,
|
||||
checkDomain: false,
|
||||
sessionName: "AclSubscriptionEnforcementTests",
|
||||
sessionTimeout: 60_000,
|
||||
identity: new UserIdentity(new AnonymousIdentityToken()),
|
||||
preferredLocales: null,
|
||||
ct: ct);
|
||||
}
|
||||
|
||||
public async ValueTask DisposeAsync()
|
||||
{
|
||||
await _host.DisposeAsync();
|
||||
if (Directory.Exists(_pkiRoot))
|
||||
{
|
||||
try { Directory.Delete(_pkiRoot, recursive: true); }
|
||||
catch (IOException) { /* best effort */ }
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
private static int AllocateFreePort()
|
||||
{
|
||||
var listener = new TcpListener(IPAddress.Loopback, 0);
|
||||
listener.Start();
|
||||
var port = ((IPEndPoint)listener.LocalEndpoint).Port;
|
||||
listener.Stop();
|
||||
return port;
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user