Joseph Doherty
438484fdca
v2-ci / unit-tests (push) Has been cancelled
Probe 'failover <primary> <backup> <mode> <dir>' drives ScadaBridge's own DataConnectionActor (production primary/backup machine, real DataConnectionFactory, default DataConnectionOptions, FailoverRetryCount 3) against two servers; run.py loses the primary (SIGKILL or SIGSTOP) once the probe reports a steady primary, and the probe requires Bad quality on loss, "Connected to backup", the backup's own value, a fresh ConditionRefresh replaying the native condition, and a write via the backup. rig-failover.sh runs the same probe against the local docker-dev central pair. Matrix legs gain a UNS subscription case and a Server-object alarm subscription on the reconnected session whose ConditionRefresh must replay the active condition (11 cases). run.py gains --versions, --suites, --failover-modes and --failover-loss. Claude-Session: https://claude.ai/code/session_01GT67ACmpMR7b9LGRPLijMi
198 lines
10 KiB
Python
198 lines
10 KiB
Python
#!/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)
|
|
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 --<version>-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 = {"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)
|
|
receipt = {
|
|
"artifacts": {name: {"path": str(path), "sha256": hashlib.sha256(path.read_bytes()).hexdigest()}
|
|
for name, path in artifacts.items()},
|
|
"legs": [],
|
|
}
|
|
# ScadaBridge#121: secure legs provision the consumer identity first, then exchange PUBLIC certs
|
|
# both ways. Each secure mode also runs two negative controls that must fail closed:
|
|
# untrusted-client - the server's trusted store does not hold the consumer certificate
|
|
# 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 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"]))
|
|
|
|
|
|
if __name__ == "__main__":
|
|
raise SystemExit(main())
|