feat(ics): OPC UA reachability probe (#148)

Extends the ICS probe with OPC UA (port 4840), a common CODESYS runtime service.
A read-only UACP handshake (Hello → Ack/Err) confirms an OPC UA server is
listening and flags it for review — the finding notes the common insecure default
(SecurityPolicy None + Anonymous user token) that allows unauthenticated,
unencrypted access.

- pipeline::ics::opcua — minimal UACP Hello/Ack probe (no secure channel).
- probe_target now checks Modbus/TCP + OPC UA (no longer early-returns on
  no-Modbus); emits `ics-opcua-exposed` (Medium, CWE-319).

Deep SecurityPolicy / user-token analysis via a full OPC UA stack is a follow-on.
Unit-tested against an in-process mock OPC UA server. Tracker #167.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>

compliance-agent/src/pipeline/ics/mod.rs

@@ -3,9 +3,10 @@
 //! Where the control-logic scanner is static (over ST / PLCopen XML), this probes
 //! the *running* device over industrial protocols and reports exposed /
 //! unauthenticated control interfaces. It is read-only: it never writes to a live
-//! process. Modbus/TCP is implemented first; OPC UA / EtherNet-IP are follow-ons.
+//! process. Modbus/TCP and OPC UA are implemented; EtherNet-IP is a follow-on.
 
 pub mod modbus;
+pub mod opcua;
 
 use std::time::Duration;
 
@@ -13,14 +14,23 @@ use compliance_core::models::{Finding, ScanType, Severity};
 
 use crate::pipeline::dedup;
 
-/// Default Modbus/TCP port.
+/// Default Modbus/TCP and OPC UA ports (each on its own well-known port,
+/// independent of any WebVisu HTTP port).
 const MODBUS_PORT: u16 = 502;
+const OPCUA_PORT: u16 = 4840;
 
-/// Probe a PLC/SPS device's industrial-protocol surface and return findings.
-/// `endpoint` is the target's live-URL / host reference.
+/// Probe a PLC/SPS device's industrial-protocol surface (Modbus/TCP + OPC UA) and
+/// return findings. Read-only. `endpoint` is the target's live-URL / host reference.
 pub async fn probe_target(endpoint: &str, repo_id: &str, budget: Duration) -> Vec<Finding> {
-    let (host, port) = parse_endpoint(endpoint);
-    let probe = modbus::probe(&host, port, budget).await;
+    let (host, modbus_port) = parse_endpoint(endpoint);
+    let mut findings = modbus_findings(&host, modbus_port, repo_id, budget).await;
+    findings.extend(opcua_findings(&host, OPCUA_PORT, repo_id, budget).await);
+    findings
+}
+
+/// Findings from probing the Modbus/TCP surface.
+async fn modbus_findings(host: &str, port: u16, repo_id: &str, budget: Duration) -> Vec<Finding> {
+    let probe = modbus::probe(host, port, budget).await;
     let mut findings = Vec::new();
     if !probe.speaks_modbus {
         // Not reachable, or the port does not speak Modbus — nothing to report.
@@ -90,6 +100,43 @@ pub async fn probe_target(endpoint: &str, repo_id: &str, budget: Duration) -> Ve
     findings
 }
 
+/// Findings from probing the OPC UA surface (default port 4840). A reachability
+/// probe only: it flags an exposed OPC UA server for review of its security
+/// policy / authentication (deep SecurityPolicy analysis is a follow-on).
+async fn opcua_findings(host: &str, port: u16, repo_id: &str, budget: Duration) -> Vec<Finding> {
+    let probe = opcua::probe(host, port, budget).await;
+    let mut findings = Vec::new();
+    if !probe.is_opcua {
+        return findings;
+    }
+    let target = format!("{host}:{port}");
+    let fp = dedup::compute_fingerprint(&[repo_id, "ics-opcua-exposed", &target]);
+    let mut f = Finding::new(
+        repo_id.to_string(),
+        fp,
+        "ics-probe".to_string(),
+        ScanType::IcsProbe,
+        "OPC UA server exposed on the network".to_string(),
+        format!(
+            "An OPC UA server answers at {target}. Verify it enforces message security \
+             (a SecurityPolicy other than None) and rejects anonymous sessions — the common \
+             default of SecurityPolicy None + an Anonymous user token allows unauthenticated, \
+             unencrypted read/write of the server's address space."
+        ),
+        Severity::Medium,
+    );
+    f.rule_id = Some("ics-opcua-exposed".to_string());
+    f.cwe = Some("CWE-319".to_string());
+    f.remediation = Some(
+        "Restrict OPC UA (4840) to a trusted network; require a signed & encrypted \
+         SecurityPolicy (Basic256Sha256 or better) with certificate / username \
+         authentication, and disable the Anonymous user token."
+            .to_string(),
+    );
+    findings.push(f);
+    findings
+}
+
 /// Extract `(host, port)` from a target reference. Modbus lives on its own port
 /// (502 by default), independent of any HTTP/WebVisu URL, so unless the reference
 /// explicitly carries `modbus://host:port` or a bare `host:port`, we probe 502.

compliance-agent/src/pipeline/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());
+    }
+}