compliance-scanner-agent/werkbank-exec/src/ics/opcua.rs

//! 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());
    }
}