Joseph Doherty
ea83fc58b2
Probe 'prepare' provisions the consumer identity via OpcUaApplicationIdentity.EnsureApplicationCertificateAsync and exports the public DER; runner installs it in the server trusted store before start. Adds Sign (OPCUA_MATRIX_SERVER_SECURITY=sign), untrusted-client / untrusted-server fail-closed controls per secure mode, and OPCUA_MATRIX_ACL=off. Autoaccept stays false everywhere.
105 lines
5.8 KiB
Python
105 lines
5.8 KiB
Python
#!/usr/bin/env python3
|
|
"""Run the real ScadaBridge consumer in separate processes against both application versions."""
|
|
import argparse
|
|
import hashlib
|
|
import json
|
|
import os
|
|
from pathlib import Path
|
|
import secrets
|
|
import subprocess
|
|
import time
|
|
|
|
|
|
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("--probe", type=Path, required=True)
|
|
parser.add_argument("--results", type=Path, required=True)
|
|
args = parser.parse_args()
|
|
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"}
|
|
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 ("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))
|
|
return int(any(leg["verdict"] != "passed" for leg in receipt["legs"]))
|
|
|
|
|
|
if __name__ == "__main__":
|
|
raise SystemExit(main())
|