refactor(werkbank): extract soft-PLC provisioning + ICS probe into werkbank-exec (WB-04a) (#208)

werkbank-exec/src/ics/opcua.rs

@@ -0,0 +1,131 @@
+//! Minimal OPC UA reachability probe.
+//!
+//! Speaks just the OPC UA Connection Protocol (UACP) handshake — a `HEL` (Hello)
+//! message, expecting an `ACK` (or `ERR`) reply — to confirm an OPC UA server is
+//! listening (default port 4840). It does **not** open a secure channel or make
+//! service calls; deep analysis of the server's SecurityPolicy / user-token
+//! policies (the common `None` + `Anonymous` misconfiguration) is a follow-on best
+//! done with a full OPC UA stack.
+
+use std::time::Duration;
+
+use tokio::io::{AsyncReadExt, AsyncWriteExt};
+use tokio::net::TcpStream;
+use tokio::time::timeout;
+
+/// Outcome of an OPC UA handshake probe.
+#[derive(Debug, Default, PartialEq, Eq)]
+pub struct OpcUaProbe {
+    /// A TCP connection to the port was established.
+    pub reachable: bool,
+    /// The endpoint replied to the UACP Hello (`ACK`) or rejected it (`ERR`) —
+    /// either way it speaks OPC UA.
+    pub is_opcua: bool,
+}
+
+/// Probe an OPC UA endpoint with a UACP Hello. Read-only handshake only.
+pub async fn probe(host: &str, port: u16, budget: Duration) -> OpcUaProbe {
+    let mut out = OpcUaProbe::default();
+    let Ok(Ok(mut stream)) = timeout(budget, TcpStream::connect((host, port))).await else {
+        return out;
+    };
+    out.reachable = true;
+
+    let hello = hello_message(&format!("opc.tcp://{host}:{port}"));
+    if timeout(budget, stream.write_all(&hello))
+        .await
+        .ok()
+        .and_then(Result::ok)
+        .is_none()
+    {
+        return out;
+    }
+
+    // Read the 3-byte message type of the reply: ACK (accepted) or ERR (rejected
+    // our hello) both prove the peer speaks the OPC UA connection protocol.
+    let mut mt = [0u8; 3];
+    if timeout(budget, stream.read_exact(&mut mt))
+        .await
+        .ok()
+        .and_then(Result::ok)
+        .is_none()
+    {
+        return out;
+    }
+    if &mt == b"ACK" || &mt == b"ERR" {
+        out.is_opcua = true;
+    }
+    out
+}
+
+/// Build a UACP `HEL` (Hello) message advertising our buffer sizes + endpoint URL.
+fn hello_message(endpoint_url: &str) -> Vec<u8> {
+    let url = endpoint_url.as_bytes();
+    let mut m = Vec::with_capacity(32 + url.len());
+    m.extend_from_slice(b"HELF");
+    m.extend_from_slice(&0u32.to_le_bytes()); // message size — patched below
+    m.extend_from_slice(&0u32.to_le_bytes()); // ProtocolVersion
+    m.extend_from_slice(&65536u32.to_le_bytes()); // ReceiveBufferSize
+    m.extend_from_slice(&65536u32.to_le_bytes()); // SendBufferSize
+    m.extend_from_slice(&0u32.to_le_bytes()); // MaxMessageSize (0 = no limit)
+    m.extend_from_slice(&0u32.to_le_bytes()); // MaxChunkCount
+    m.extend_from_slice(&(url.len() as i32).to_le_bytes()); // EndpointUrl length
+    m.extend_from_slice(url);
+    let size = m.len() as u32;
+    m[4..8].copy_from_slice(&size.to_le_bytes());
+    m
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use tokio::net::TcpListener;
+
+    /// A mock OPC UA server that reads the Hello and replies with an `ACK` frame.
+    async fn mock_server() -> std::net::SocketAddr {
+        let listener = TcpListener::bind("127.0.0.1:0").await.expect("bind");
+        let addr = listener.local_addr().expect("addr");
+        tokio::spawn(async move {
+            let (mut sock, _) = listener.accept().await.expect("accept");
+            // Read the Hello header (8 bytes) to learn the size, then drain it.
+            let mut hdr = [0u8; 8];
+            if sock.read_exact(&mut hdr).await.is_err() {
+                return;
+            }
+            let size = u32::from_le_bytes([hdr[4], hdr[5], hdr[6], hdr[7]]) as usize;
+            let mut rest = vec![0u8; size.saturating_sub(8)];
+            let _ = sock.read_exact(&mut rest).await;
+            // Reply: ACK + size + 5 u32 fields.
+            let mut ack = Vec::new();
+            ack.extend_from_slice(b"ACKF");
+            ack.extend_from_slice(&28u32.to_le_bytes());
+            for _ in 0..5 {
+                ack.extend_from_slice(&0u32.to_le_bytes());
+            }
+            let _ = sock.write_all(&ack).await;
+        });
+        addr
+    }
+
+    #[tokio::test]
+    async fn probe_detects_an_opcua_server() {
+        let addr = mock_server().await;
+        let p = probe(&addr.ip().to_string(), addr.port(), Duration::from_secs(2)).await;
+        assert!(p.reachable && p.is_opcua);
+    }
+
+    #[tokio::test]
+    async fn probe_reports_unreachable_for_a_closed_port() {
+        let p = probe("127.0.0.1", 1, Duration::from_millis(500)).await;
+        assert!(!p.reachable && !p.is_opcua);
+    }
+
+    #[test]
+    fn hello_message_is_well_formed() {
+        let m = hello_message("opc.tcp://h:4840");
+        assert_eq!(&m[0..4], b"HELF");
+        // The embedded size equals the actual length.
+        let size = u32::from_le_bytes([m[4], m[5], m[6], m[7]]) as usize;
+        assert_eq!(size, m.len());
+    }
+}