feat(plc): DAST the provisioned WebVisu + enumerate exposed Modbus points (#183)

Completes the provision-and-test loop's dynamic coverage (sub-tasks 3 + 4):

- provision_and_test now returns a ProvisionOutcome { ics findings, DAST run }.
  After the Modbus probe it runs a bounded, best-effort DAST scan against the
  provisioned web endpoint (independently timed out so it can't consume the whole
  instance lifetime), and the orchestrator persists the DAST scan run + findings
  linked to the scan run. Kept as a plain data return so the whole run is
  portable to a remote execution backend. On the OpenPLC substrate the web
  endpoint is OpenPLC's own UI (fidelity caveat documented); the CODESYS-runtime
  follow-up raises this to a real WebVisu.

- ICS Modbus probe now enumerates the exposed process surface (read-only): a Read
  Coils and a Read Holding Registers of the first block. Coils and holding
  registers are read/write process points, so an exposed block is an
  unauthenticated *write* surface — reported as `ics-modbus-exposed-points`
  (High). Read-only to detect (we never write), so it is safe on the live probe
  too, not just the provisioned instance.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>

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

@@ -103,6 +103,39 @@ async fn modbus_findings(host: &str, port: u16, repo_id: &str, budget: Duration)
         );
         findings.push(f);
     }
+
+    // Exposed process points: coils / holding registers that a read enumerated
+    // and that, over unauthenticated Modbus/TCP, are also writable. This is the
+    // concrete attack surface behind the exposure — the live variables an
+    // attacker can overwrite. (Read-only to detect: we never write.)
+    let coils = probe.coils_readable.unwrap_or(0);
+    let registers = probe.holding_registers_readable.unwrap_or(0);
+    if coils > 0 || registers > 0 {
+        let fp = dedup::compute_fingerprint(&[repo_id, "ics-modbus-exposed-points", &target]);
+        let mut f = Finding::new(
+            repo_id.to_string(),
+            fp,
+            "ics-probe".to_string(),
+            ScanType::IcsProbe,
+            "Writable process points exposed over unauthenticated Modbus/TCP".to_string(),
+            format!(
+                "Reading the device at {target} enumerated {coils} coil(s) and {registers} \
+                 holding register(s). Coils and holding registers are read/write process points \
+                 in Modbus, so any host that can reach this port can not only read but overwrite \
+                 live process state (force coils, change setpoints) without authentication."
+            ),
+            Severity::High,
+        );
+        f.rule_id = Some("ics-modbus-exposed-points".to_string());
+        f.cwe = Some("CWE-306".to_string());
+        f.remediation = Some(
+            "Segment the Modbus/TCP port to a trusted control network; where the device \
+             supports it use Modbus/TLS or an authenticating protocol gateway; restrict which \
+             function codes and register ranges are reachable from outside the control zone."
+                .to_string(),
+        );
+        findings.push(f);
+    }
     findings
 }
 

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

@@ -21,6 +21,13 @@ pub struct ModbusProbe {
     pub speaks_modbus: bool,
     /// Device identity, if disclosed via Read Device Identification (FC 43 / 14).
     pub device: Option<DeviceId>,
+    /// Coils returned by a Read Coils of the first block, if that address range
+    /// exists. Coils are read/write process bits, so an exposed block is an
+    /// unauthenticated write surface on the live process.
+    pub coils_readable: Option<u16>,
+    /// Holding registers returned by a Read Holding Registers of the first block,
+    /// if that range exists. Holding registers are read/write process words.
+    pub holding_registers_readable: Option<u16>,
 }
 
 /// Vendor / product / revision from Read Device Identification.
@@ -31,8 +38,13 @@ pub struct DeviceId {
     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.
+/// How many coils / holding registers to request when enumerating the exposed
+/// process surface. Read-only: a normal reply means the block exists and is,
+/// over unauthenticated Modbus/TCP, also writable.
+const ENUM_QTY: u16 = 16;
+
+/// Probe a Modbus/TCP endpoint. Read-only: issues Read Holding Registers, Read
+/// Coils, and Read Device Identification requests; 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 {
@@ -40,13 +52,28 @@ pub async fn probe(host: &str, port: u16, budget: Duration) -> ModbusProbe {
     };
     out.reachable = true;
 
-    // Read Holding Registers (FC 0x03), unit 1, addr 0, qty 1 — a benign read.
-    let rhr = [0x03u8, 0x00, 0x00, 0x00, 0x01];
+    // Read Holding Registers (FC 0x03), unit 1, addr 0 — a benign read that also
+    // enumerates the exposed register block.
+    let rhr = [0x03u8, 0x00, 0x00, (ENUM_QTY >> 8) as u8, ENUM_QTY as u8];
     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;
         }
+        if resp.first() == Some(&0x03) {
+            out.holding_registers_readable = Some(register_count_from_reply(&resp));
+        }
+    }
+
+    // Read Coils (FC 0x01), addr 0 — enumerates the exposed coil (bit) block.
+    let rc = [0x01u8, 0x00, 0x00, (ENUM_QTY >> 8) as u8, ENUM_QTY as u8];
+    if let Some(resp) = txn(&mut stream, 1, &rc, budget).await {
+        if matches!(resp.first(), Some(0x01) | Some(0x81)) {
+            out.speaks_modbus = true;
+        }
+        if resp.first() == Some(&0x01) {
+            out.coils_readable = Some(coil_count_from_reply(&resp));
+        }
     }
 
     // Read Device Identification (FC 0x2B / MEI 0x0E), basic (0x01), object 0.
@@ -60,6 +87,17 @@ pub async fn probe(host: &str, port: u16, budget: Duration) -> ModbusProbe {
     out
 }
 
+/// Coils reported by a Read Coils reply `[0x01, byte_count, data…]` (8 per byte).
+fn coil_count_from_reply(pdu: &[u8]) -> u16 {
+    pdu.get(1).map(|&b| u16::from(b) * 8).unwrap_or(0)
+}
+
+/// Registers reported by a Read Holding Registers reply `[0x03, byte_count,
+/// data…]` (2 bytes per register).
+fn register_count_from_reply(pdu: &[u8]) -> u16 {
+    pdu.get(1).map(|&b| u16::from(b) / 2).unwrap_or(0)
+}
+
 /// 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>> {
@@ -154,7 +192,8 @@ mod tests {
                     break;
                 }
                 let reply_pdu: Vec<u8> = match pdu.first() {
-                    Some(0x03) => vec![0x03, 0x02, 0x00, 0x00], // 1 register = 0
+                    Some(0x03) => vec![0x03, 0x02, 0x00, 0x00], // 1 register (byte_count 2)
+                    Some(0x01) => vec![0x01, 0x02, 0xFF, 0xFF], // 16 coils (byte_count 2)
                     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
@@ -185,6 +224,24 @@ mod tests {
         assert_eq!(dev.product.as_deref(), Some("PLC"));
     }
 
+    #[tokio::test]
+    async fn probe_enumerates_exposed_process_points() {
+        let addr = mock_server(false).await;
+        let p = probe(&addr.ip().to_string(), addr.port(), Duration::from_secs(2)).await;
+        assert!(p.speaks_modbus);
+        // The mock returns a 2-byte holding-register block (1 register) and a
+        // 2-byte coil block (16 coils).
+        assert_eq!(p.holding_registers_readable, Some(1));
+        assert_eq!(p.coils_readable, Some(16));
+    }
+
+    #[test]
+    fn reply_counts_decode_byte_counts() {
+        assert_eq!(register_count_from_reply(&[0x03, 0x08]), 4); // 8 bytes → 4 regs
+        assert_eq!(coil_count_from_reply(&[0x01, 0x03]), 24); // 3 bytes → 24 coils
+        assert_eq!(register_count_from_reply(&[0x03]), 0); // malformed → 0
+    }
+
     #[tokio::test]
     async fn probe_reports_unreachable_for_a_closed_port() {
         // 127.0.0.1:1 is (almost certainly) closed.

compliance-agent/src/pipeline/orchestrator.rs

@@ -595,7 +595,7 @@ impl PipelineOrchestrator {
         let http = crate::pipeline::plc::runtime::http_client()?;
         let provisioner =
             crate::pipeline::plc::runtime::DockerSoftPlc::new(self.config.plc_runtime.clone());
-        let findings = crate::pipeline::plc::runtime::provision_and_test(
+        let outcome = crate::pipeline::plc::runtime::provision_and_test(
             &provisioner,
             &http,
             &self.config.plc_runtime,
@@ -605,12 +605,13 @@ impl PipelineOrchestrator {
         .await?;
         tracing::info!(
             target_id,
-            found = findings.len(),
+            found = outcome.findings.len(),
+            dast = outcome.dast.is_some(),
             "provision-and-test complete"
         );
 
         let mut new_count = 0u32;
-        for mut finding in findings {
+        for mut finding in outcome.findings {
             finding.scan_run_id = Some(scan_run_id.to_string());
             if self
                 .db
@@ -623,6 +624,21 @@ impl PipelineOrchestrator {
                 new_count += 1;
             }
         }
+
+        // Persist the DAST scan of the provisioned web endpoint, linked to this
+        // scan run (mirrors `maybe_trigger_dast`).
+        if let Some(dast) = outcome.dast {
+            let mut scan_run = dast.scan_run;
+            scan_run.sast_scan_run_id = Some(scan_run_id.to_string());
+            if let Err(e) = self.db.dast_scan_runs().insert_one(&scan_run).await {
+                tracing::warn!(target_id, error = %e, "failed to store provisioned DAST scan run");
+            }
+            for finding in &dast.findings {
+                if let Err(e) = self.db.dast_findings().insert_one(finding).await {
+                    tracing::warn!(target_id, error = %e, "failed to store provisioned DAST finding");
+                }
+            }
+        }
         Ok(new_count)
     }
 

compliance-agent/src/pipeline/plc/runtime/mod.rs

@@ -20,6 +20,7 @@ use std::time::Duration;
 
 use secrecy::ExposeSecret;
 
+use compliance_core::models::dast::{DastFinding, DastScanRun, DastTarget, DastTargetType};
 use compliance_core::models::Finding;
 use compliance_core::PlcRuntimeConfig;
 
@@ -27,6 +28,27 @@ use crate::error::AgentError;
 
 pub use provision::{DockerSoftPlc, ProvisionedRuntime, SoftPlc};
 
+/// The result of a DAST scan against a provisioned web endpoint.
+#[derive(Debug)]
+pub struct DastRunResult {
+    /// The scan-run record (linked to the onboarded target).
+    pub scan_run: DastScanRun,
+    /// The DAST findings.
+    pub findings: Vec<DastFinding>,
+}
+
+/// Everything a provision-and-test run produced: the ICS-probe findings plus, if
+/// it ran, the DAST scan of the provisioned web endpoint. The caller persists
+/// both — keeping this a plain data return means the whole run is portable to a
+/// remote execution backend that just hands the results back.
+#[derive(Debug, Default)]
+pub struct ProvisionOutcome {
+    /// ICS-probe findings from the provisioned Modbus endpoint.
+    pub findings: Vec<Finding>,
+    /// DAST scan of the provisioned web endpoint, if it ran.
+    pub dast: Option<DastRunResult>,
+}
+
 /// A control-logic program ready to load into a soft-PLC: the source text plus a
 /// cosmetic file name (OpenPLC re-stores it under its own name).
 #[derive(Debug, Clone)]
@@ -129,7 +151,7 @@ pub async fn provision_and_test<P: SoftPlc>(
     cfg: &PlcRuntimeConfig,
     program: &PlcProgram,
     target_id: &str,
-) -> Result<Vec<Finding>, AgentError> {
+) -> Result<ProvisionOutcome, AgentError> {
     let handle = provisioner.provision(target_id).await?;
     tracing::info!(
         target_id,
@@ -159,19 +181,19 @@ pub async fn provision_and_test<P: SoftPlc>(
                 instance = %handle.name,
                 "provision-and-test hit the lifetime deadline; torn down"
             );
-            Ok(Vec::new())
+            Ok(ProvisionOutcome::default())
         }
     }
 }
 
-/// The load → start → probe body, run under the caller's deadline.
+/// The load → start → probe → DAST body, run under the caller's deadline.
 async fn run_dynamic_test(
     http: &reqwest::Client,
     cfg: &PlcRuntimeConfig,
     program: &PlcProgram,
     target_id: &str,
     handle: &ProvisionedRuntime,
-) -> Result<Vec<Finding>, AgentError> {
+) -> Result<ProvisionOutcome, AgentError> {
     let ready_budget = Duration::from_secs((cfg.max_lifetime_secs / 3).clamp(10, 60));
     openplc::wait_ready(http, &handle.webvisu_url, ready_budget).await?;
 
@@ -198,7 +220,54 @@ async fn run_dynamic_test(
         found = findings.len(),
         "provision-and-test probe complete"
     );
-    Ok(findings)
+
+    // DAST the provisioned web endpoint (independently bounded so it can't eat
+    // the whole lifetime). On the OpenPLC substrate this is OpenPLC's own web UI,
+    // not a customer HMI — the CODESYS-runtime follow-up raises the fidelity —
+    // but it proves the deploy→run→probe→DAST loop end to end.
+    let dast_budget = Duration::from_secs((cfg.max_lifetime_secs / 2).clamp(20, 120));
+    let dast = match tokio::time::timeout(dast_budget, run_webvisu_dast(handle, target_id)).await {
+        Ok(d) => d,
+        Err(_) => {
+            tracing::warn!(target_id, instance = %handle.name, "provision-and-test DAST timed out");
+            None
+        }
+    };
+
+    Ok(ProvisionOutcome { findings, dast })
+}
+
+/// Run a bounded DAST scan against the provisioned web endpoint and tag the
+/// results with our target id. Best-effort — a DAST failure never fails the run.
+async fn run_webvisu_dast(handle: &ProvisionedRuntime, target_id: &str) -> Option<DastRunResult> {
+    let mut dt = DastTarget::new(
+        "provisioned-webvisu".to_string(),
+        handle.webvisu_url.clone(),
+        DastTargetType::WebApp,
+    );
+    dt.repo_id = Some(target_id.to_string());
+    dt.max_crawl_depth = 2; // shallow — the instance is ephemeral
+
+    let orchestrator = compliance_dast::DastOrchestrator::new(100);
+    match orchestrator.run_scan(&dt, Vec::new()).await {
+        Ok((mut scan_run, mut findings)) => {
+            scan_run.target_id = target_id.to_string();
+            for f in &mut findings {
+                f.target_id = target_id.to_string();
+            }
+            tracing::info!(
+                target_id,
+                instance = %handle.name,
+                dast_findings = findings.len(),
+                "provision-and-test DAST complete"
+            );
+            Some(DastRunResult { scan_run, findings })
+        }
+        Err(e) => {
+            tracing::warn!(target_id, instance = %handle.name, error = %e, "provision-and-test DAST failed");
+            None
+        }
+    }
 }
 
 #[cfg(test)]
@@ -319,7 +388,8 @@ mod tests {
         let out = provision_and_test(&fake, &http, &short_cfg(), &prog, "t1")
             .await
             .expect("ok on deadline");
-        assert!(out.is_empty(), "deadline path yields no findings");
+        assert!(out.findings.is_empty(), "deadline path yields no findings");
+        assert!(out.dast.is_none(), "deadline path runs no DAST");
         assert_eq!(provisions.load(Ordering::SeqCst), 1);
         assert_eq!(teardowns.load(Ordering::SeqCst), 1, "teardown must run");
     }