feat(ics): dynamic Modbus/TCP probe for PLC/SPS devices (#172)

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

@@ -0,0 +1,138 @@
+//! Dynamic ICS (industrial control system) probing for PLC/SPS targets.
+//!
+//! 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.
+
+pub mod modbus;
+
+use std::time::Duration;
+
+use compliance_core::models::{Finding, ScanType, Severity};
+
+use crate::pipeline::dedup;
+
+/// Default Modbus/TCP port.
+const MODBUS_PORT: u16 = 502;
+
+/// Probe a PLC/SPS device's industrial-protocol surface and return findings.
+/// `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 mut findings = Vec::new();
+    if !probe.speaks_modbus {
+        // Not reachable, or the port does not speak Modbus — nothing to report.
+        return findings;
+    }
+    let target = format!("{host}:{port}");
+
+    // Reachable Modbus/TCP = unauthenticated, cleartext control access by design.
+    let fp = dedup::compute_fingerprint(&[repo_id, "ics-modbus-exposed", &target]);
+    let mut f = Finding::new(
+        repo_id.to_string(),
+        fp,
+        "ics-probe".to_string(),
+        ScanType::IcsProbe,
+        "Modbus/TCP control interface exposed without authentication".to_string(),
+        format!(
+            "The device at {target} answers Modbus/TCP requests. Modbus/TCP has no \
+             authentication or encryption in the protocol, so any host that can reach this \
+             port can read and write process variables (coils/registers) and disrupt the \
+             controlled process."
+        ),
+        Severity::Critical,
+    );
+    f.rule_id = Some("ics-modbus-exposed".to_string());
+    f.cwe = Some("CWE-306".to_string());
+    f.remediation = Some(
+        "Restrict the Modbus/TCP port to a trusted control network (segmentation / \
+         firewall / VPN), never expose it to IT or the internet, and prefer an authenticated \
+         transport (e.g. Modbus/TLS) or a secure protocol gateway where available."
+            .to_string(),
+    );
+    findings.push(f);
+
+    if let Some(dev) = &probe.device {
+        let details = [
+            dev.vendor.as_deref(),
+            dev.product.as_deref(),
+            dev.revision.as_deref(),
+        ]
+        .into_iter()
+        .flatten()
+        .collect::<Vec<_>>()
+        .join(" / ");
+        let fp = dedup::compute_fingerprint(&[repo_id, "ics-device-disclosure", &target]);
+        let mut f = Finding::new(
+            repo_id.to_string(),
+            fp,
+            "ics-probe".to_string(),
+            ScanType::IcsProbe,
+            "PLC device identity disclosed over Modbus".to_string(),
+            format!(
+                "The device at {target} discloses its identity via Modbus Read Device \
+                 Identification: {details}. This aids fingerprinting and targeting of \
+                 known-vulnerable firmware/runtime versions."
+            ),
+            Severity::Low,
+        );
+        f.rule_id = Some("ics-device-disclosure".to_string());
+        f.cwe = Some("CWE-200".to_string());
+        f.remediation = Some(
+            "Limit network reach to the device; Modbus device identification cannot be \
+             disabled, so exposure is bounded by network segmentation."
+                .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.
+fn parse_endpoint(endpoint: &str) -> (String, u16) {
+    let s = endpoint.trim();
+    let (scheme, rest) = match s.split_once("://") {
+        Some((sch, r)) => (Some(sch.to_ascii_lowercase()), r),
+        None => (None, s),
+    };
+    let hostport = rest.split(['/', '?']).next().unwrap_or(rest);
+    let (host, port) = match hostport.rsplit_once(':') {
+        Some((h, p)) => (h.to_string(), p.parse::<u16>().ok()),
+        None => (hostport.to_string(), None),
+    };
+    let port = match (scheme.as_deref(), port) {
+        // Explicit Modbus port, or a bare host:port the user chose.
+        (Some("modbus"), Some(p)) | (None, Some(p)) => p,
+        // An http(s)/WebVisu URL (or no port): Modbus is on its own port.
+        _ => MODBUS_PORT,
+    };
+    (host, port)
+}
+
+#[cfg(test)]
+mod tests {
+    use super::parse_endpoint;
+
+    #[test]
+    fn endpoint_parsing_picks_the_modbus_port() {
+        assert_eq!(parse_endpoint("10.0.0.5"), ("10.0.0.5".into(), 502));
+        assert_eq!(parse_endpoint("10.0.0.5:1502"), ("10.0.0.5".into(), 1502));
+        assert_eq!(
+            parse_endpoint("modbus://plc.local:5020"),
+            ("plc.local".into(), 5020)
+        );
+        // A WebVisu URL: the http port is ignored; Modbus is on 502.
+        assert_eq!(
+            parse_endpoint("http://plc.local:8080/webvisu"),
+            ("plc.local".into(), 502)
+        );
+        assert_eq!(
+            parse_endpoint("https://plc.local/"),
+            ("plc.local".into(), 502)
+        );
+    }
+}

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

@@ -0,0 +1,205 @@
+//! Minimal Modbus/TCP client for dynamic ICS probing.
+//!
+//! Modbus/TCP (port 502) has no authentication or encryption in the protocol, so
+//! an endpoint that answers requests is, by design, open to any host that can
+//! reach it. The probe only *reads* — a Read Holding Registers request and a Read
+//! Device Identification request — and never writes to the live process.
+
+use std::time::Duration;
+
+use tokio::io::{AsyncReadExt, AsyncWriteExt};
+use tokio::net::TcpStream;
+use tokio::time::timeout;
+
+/// Outcome of probing a Modbus/TCP endpoint.
+#[derive(Debug, Default, PartialEq, Eq)]
+pub struct ModbusProbe {
+    /// A TCP connection to the port was established.
+    pub reachable: bool,
+    /// The endpoint answered a Modbus request (a normal reply or a Modbus
+    /// exception) — i.e. it speaks Modbus, unauthenticated.
+    pub speaks_modbus: bool,
+    /// Device identity, if disclosed via Read Device Identification (FC 43 / 14).
+    pub device: Option<DeviceId>,
+}
+
+/// Vendor / product / revision from Read Device Identification.
+#[derive(Debug, Default, PartialEq, Eq)]
+pub struct DeviceId {
+    pub vendor: Option<String>,
+    pub product: Option<String>,
+    pub revision: Option<String>,
+}
+
+/// Probe a Modbus/TCP endpoint. Read-only: issues a Read Holding Registers and a
+/// Read Device Identification request; never writes to the device.
+pub async fn probe(host: &str, port: u16, budget: Duration) -> ModbusProbe {
+    let mut out = ModbusProbe::default();
+    let Ok(Ok(mut stream)) = timeout(budget, TcpStream::connect((host, port))).await else {
+        return out; // unreachable
+    };
+    out.reachable = true;
+
+    // Read Holding Registers (FC 0x03), unit 1, addr 0, qty 1 — a benign read.
+    let rhr = [0x03u8, 0x00, 0x00, 0x00, 0x01];
+    if let Some(resp) = txn(&mut stream, 1, &rhr, budget).await {
+        // A normal reply (0x03) or an exception (0x83) both prove it speaks Modbus.
+        if matches!(resp.first(), Some(0x03) | Some(0x83)) {
+            out.speaks_modbus = true;
+        }
+    }
+
+    // Read Device Identification (FC 0x2B / MEI 0x0E), basic (0x01), object 0.
+    let rdi = [0x2Bu8, 0x0E, 0x01, 0x00];
+    if let Some(resp) = txn(&mut stream, 1, &rdi, budget).await {
+        if resp.first() == Some(&0x2B) {
+            out.speaks_modbus = true;
+            out.device = parse_device_id(&resp);
+        }
+    }
+    out
+}
+
+/// Send one Modbus PDU and return the response PDU (function code + data), or
+/// `None` on timeout / malformed reply.
+async fn txn(stream: &mut TcpStream, unit: u8, pdu: &[u8], budget: Duration) -> Option<Vec<u8>> {
+    // MBAP header: transaction id (2) + protocol id (2) = 0 + length (2) + unit (1),
+    // then the PDU. `length` counts the unit byte plus the PDU.
+    let len = (pdu.len() + 1) as u16;
+    let mut frame = Vec::with_capacity(7 + pdu.len());
+    frame.extend_from_slice(&[0x00, 0x01]); // transaction id
+    frame.extend_from_slice(&[0x00, 0x00]); // protocol id
+    frame.extend_from_slice(&len.to_be_bytes());
+    frame.push(unit);
+    frame.extend_from_slice(pdu);
+    timeout(budget, stream.write_all(&frame)).await.ok()?.ok()?;
+
+    let mut hdr = [0u8; 7];
+    timeout(budget, stream.read_exact(&mut hdr))
+        .await
+        .ok()?
+        .ok()?;
+    // Reject non-Modbus replies (protocol id must be 0).
+    if hdr[2] != 0 || hdr[3] != 0 {
+        return None;
+    }
+    let plen = u16::from_be_bytes([hdr[4], hdr[5]]) as usize;
+    if !(2..=260).contains(&plen) {
+        return None;
+    }
+    let mut body = vec![0u8; plen - 1]; // minus the unit id already in hdr[6]
+    timeout(budget, stream.read_exact(&mut body))
+        .await
+        .ok()?
+        .ok()?;
+    Some(body)
+}
+
+/// Parse vendor / product / revision from a Read Device Identification PDU:
+/// `[0x2B, 0x0E, readDevIdCode, conformity, moreFollows, nextObjId, numObjects,
+/// (objId, len, bytes…)…]`.
+fn parse_device_id(pdu: &[u8]) -> Option<DeviceId> {
+    if pdu.len() < 7 {
+        return None;
+    }
+    let num = pdu[6] as usize;
+    let mut i = 7;
+    let mut dev = DeviceId::default();
+    for _ in 0..num {
+        if i + 2 > pdu.len() {
+            break;
+        }
+        let id = pdu[i];
+        let l = pdu[i + 1] as usize;
+        i += 2;
+        if i + l > pdu.len() {
+            break;
+        }
+        let val = String::from_utf8_lossy(&pdu[i..i + l]).trim().to_string();
+        i += l;
+        match id {
+            0x00 => dev.vendor = Some(val),
+            0x01 => dev.product = Some(val),
+            0x02 => dev.revision = Some(val),
+            _ => {}
+        }
+    }
+    if dev == DeviceId::default() {
+        None
+    } else {
+        Some(dev)
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use tokio::net::TcpListener;
+
+    /// A one-shot mock Modbus/TCP server that answers a Read Holding Registers
+    /// request and a Read Device Identification request on one connection.
+    async fn mock_server(with_device: bool) -> 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");
+            loop {
+                let mut hdr = [0u8; 7];
+                if sock.read_exact(&mut hdr).await.is_err() {
+                    break;
+                }
+                let plen = u16::from_be_bytes([hdr[4], hdr[5]]) as usize;
+                let mut pdu = vec![0u8; plen - 1];
+                if sock.read_exact(&mut pdu).await.is_err() {
+                    break;
+                }
+                let reply_pdu: Vec<u8> = match pdu.first() {
+                    Some(0x03) => vec![0x03, 0x02, 0x00, 0x00], // 1 register = 0
+                    Some(0x2B) if with_device => vec![
+                        0x2B, 0x0E, 0x01, 0x81, 0x00, 0x00, 0x02, // 2 objects
+                        0x00, 0x04, b'A', b'C', b'M', b'E', // vendor
+                        0x01, 0x03, b'P', b'L', b'C', // product
+                    ],
+                    _ => vec![pdu[0] | 0x80, 0x01], // exception
+                };
+                let len = (reply_pdu.len() + 1) as u16;
+                let mut frame = vec![hdr[0], hdr[1], 0x00, 0x00];
+                frame.extend_from_slice(&len.to_be_bytes());
+                frame.push(hdr[6]);
+                frame.extend_from_slice(&reply_pdu);
+                if sock.write_all(&frame).await.is_err() {
+                    break;
+                }
+            }
+        });
+        addr
+    }
+
+    #[tokio::test]
+    async fn probe_detects_a_modbus_endpoint_and_reads_device_id() {
+        let addr = mock_server(true).await;
+        let p = probe(&addr.ip().to_string(), addr.port(), Duration::from_secs(2)).await;
+        assert!(p.reachable && p.speaks_modbus);
+        let dev = p.device.expect("device id");
+        assert_eq!(dev.vendor.as_deref(), Some("ACME"));
+        assert_eq!(dev.product.as_deref(), Some("PLC"));
+    }
+
+    #[tokio::test]
+    async fn probe_reports_unreachable_for_a_closed_port() {
+        // 127.0.0.1:1 is (almost certainly) closed.
+        let p = probe("127.0.0.1", 1, Duration::from_millis(500)).await;
+        assert!(!p.reachable && !p.speaks_modbus);
+    }
+
+    #[test]
+    fn parses_device_identification_objects() {
+        let pdu = [
+            0x2B, 0x0E, 0x01, 0x81, 0x00, 0x00, 0x01, // 1 object
+            0x02, 0x05, b'v', b'1', b'.', b'2', b'3', // revision
+        ];
+        let dev = parse_device_id(&pdu).expect("device");
+        assert_eq!(dev.revision.as_deref(), Some("v1.23"));
+        assert!(dev.vendor.is_none());
+    }
+}

compliance-agent/src/pipeline/mod.rs

@@ -5,6 +5,7 @@ pub mod firmware_sbom;
 pub mod git;
 pub mod gitleaks;
 mod graph_build;
+pub mod ics;
 mod issue_creation;
 pub mod lint;
 pub mod orchestrator;

compliance-agent/src/pipeline/orchestrator.rs

@@ -455,8 +455,18 @@ impl PipelineOrchestrator {
         // SAST pipeline over it. A PLC device is reachable, so DAST still runs
         // against a WebVisu / exposed endpoint when one is provisioned.
         let mut new_count = 0u32;
-        if plan.has(ScanType::PlcControlLogic) {
+        let plc = plan.has(ScanType::PlcControlLogic);
+        let ics = plan.has(ScanType::IcsProbe);
+        if plc {
             new_count += self.run_plc_scan(target, &target_id, scan_run_id).await?;
+        }
+        if ics {
+            new_count += self.run_ics_probe(target, &target_id, scan_run_id).await?;
+        }
+        if plc || ics {
+            // PLC/SPS device: also DAST against a WebVisu / exposed endpoint. The
+            // control-logic scan already consumed the code artifact, so the SAST
+            // pipeline is not re-run.
             self.update_phase(scan_run_id, "dast_scanning").await;
             self.maybe_trigger_dast(&target_id, scan_run_id).await;
             return Ok(new_count);
@@ -565,6 +575,46 @@ impl PipelineOrchestrator {
         Ok(new_count)
     }
 
+    /// Probe a running PLC/SPS device over industrial protocols (Modbus/TCP, …)
+    /// and persist findings for exposed / unauthenticated control access. The
+    /// probe is read-only; it targets the Modbus port of the target's live URL.
+    async fn run_ics_probe(
+        &self,
+        target: &OnboardedTarget,
+        target_id: &str,
+        scan_run_id: &str,
+    ) -> Result<u32, AgentError> {
+        self.update_phase(scan_run_id, "ics_probe").await;
+        let Some(endpoint) = target.live_url().map(|a| a.source_ref.clone()) else {
+            tracing::warn!(target_id, "ICS probe: no live URL");
+            return Ok(0);
+        };
+        // Short per-request budget so an unreachable device doesn't stall the scan.
+        let budget = std::time::Duration::from_secs(5);
+        let findings = crate::pipeline::ics::probe_target(&endpoint, target_id, budget).await;
+        tracing::info!(
+            target_id,
+            endpoint = %endpoint,
+            found = findings.len(),
+            "ICS probe complete"
+        );
+        let mut new_count = 0u32;
+        for mut finding in findings {
+            finding.scan_run_id = Some(scan_run_id.to_string());
+            if self
+                .db
+                .findings()
+                .find_one(doc! { "fingerprint": &finding.fingerprint })
+                .await?
+                .is_none()
+            {
+                self.db.findings().insert_one(&finding).await?;
+                new_count += 1;
+            }
+        }
+        Ok(new_count)
+    }
+
     /// Store a control-application SBOM (CODESYS libraries + runtime) for a target
     /// and match it against known CVEs. Scoped to `package_manager = "codesys"` so
     /// it refreshes on re-scan and coexists with any firmware/source SBOM. The

compliance-agent/src/pipeline/plan.rs

@@ -106,6 +106,7 @@ fn phase_for(scan: ScanType) -> ScanPhase {
         ScanType::PlcControlLogic => ScanPhase::PlcAnalysis,
         ScanType::MobileStatic => ScanPhase::MobileStatic,
         ScanType::ContainerScan => ScanPhase::ContainerScan,
+        ScanType::IcsProbe => ScanPhase::IcsProbe,
     }
 }
 

compliance-core/src/models/scan.rs

@@ -24,6 +24,9 @@ pub enum ScanType {
     MobileStatic,
     /// Static analysis of a container image.
     ContainerScan,
+    /// Dynamic probing of a running PLC/SPS device over industrial protocols
+    /// (Modbus/TCP, OPC UA, …) for exposed/unauthenticated control access.
+    IcsProbe,
 }
 
 impl std::fmt::Display for ScanType {
@@ -43,6 +46,7 @@ impl std::fmt::Display for ScanType {
             Self::PlcControlLogic => write!(f, "plc_control_logic"),
             Self::MobileStatic => write!(f, "mobile_static"),
             Self::ContainerScan => write!(f, "container_scan"),
+            Self::IcsProbe => write!(f, "ics_probe"),
         }
     }
 }
@@ -76,6 +80,7 @@ pub enum ScanPhase {
     LlmTriage,
     IssueCreation,
     DastScanning,
+    IcsProbe,
     Completed,
 }
 

compliance-core/src/scan_matrix.rs

@@ -251,6 +251,12 @@ pub fn rules_for(target_type: TargetType) -> Vec<ScanRule> {
                     "Dynamic scan of the running device (WebVisu / exposed services)",
                     RunningUrl,
                 ),
+                ScanRule::new(
+                    ScanType::IcsProbe,
+                    false,
+                    "Probe the running device over industrial protocols (Modbus/TCP, …)",
+                    RunningUrl,
+                ),
             ]
         }
     }

docs/guide/plc.md

@@ -76,3 +76,22 @@ guard-aware), cleartext/insecure communication (CWE-319), insecure protocol port
 The **SBOM** view lists the CODESYS libraries (`pkg:codesys/<name>@<version>`) and
 the runtime; matching runtime components (e.g. the `Cmp*` / `3SLicense` libraries)
 surface real CODESYS advisories as CVE alerts.
+
+## Dynamic testing — ICS protocol probe
+
+Beyond the static analysis, Certifai can **probe the running device** over
+industrial protocols. Attach a **Live URL** artifact (the device host / WebVisu
+URL) to the PLC/SPS target and enable the **ICS Probe** scan.
+
+The probe is **read-only** — it never writes to the live process. It currently
+speaks **Modbus/TCP** (port 502): it confirms whether the device answers
+unauthenticated Modbus requests and reads its device identity (vendor / product /
+revision). Because Modbus/TCP has no authentication or encryption in the protocol,
+a reachable endpoint that answers is reported as an exposed control interface
+(CWE-306). OPC UA and EtherNet/IP probes are planned.
+
+::: warning
+The ICS probe connects to the live device. It is **opt-in** (off by default) and
+should only be run against targets you are authorized to test. It performs reads
+only, never writes.
+:::