feat(pipeline): PLC/SPS control-logic security scanner (IEC 61131-3) (#162)

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

@@ -0,0 +1,200 @@
+//! PLC control-logic security scanner for IEC 61131-3 targets.
+//!
+//! Parses Structured Text (raw `.st`/`.scl`/`.exp` files and PLCopen-XML
+//! projects) into an AST and runs semantic control-logic security rules over it.
+//! Implements [`ScanType::PlcControlLogic`].
+
+pub mod ast;
+pub mod lexer;
+pub mod parser;
+pub mod plcopen;
+pub mod rules;
+
+use std::path::Path;
+
+use compliance_core::error::CoreError;
+use compliance_core::models::{Finding, ScanType};
+use compliance_core::traits::{ScanOutput, Scanner};
+
+use crate::pipeline::dedup;
+
+/// Scanner for `ScanType::PlcControlLogic`.
+pub struct PlcControlLogicScanner;
+
+impl Scanner for PlcControlLogicScanner {
+    fn name(&self) -> &str {
+        "plc-control-logic"
+    }
+
+    fn scan_type(&self) -> ScanType {
+        ScanType::PlcControlLogic
+    }
+
+    #[tracing::instrument(skip_all)]
+    async fn scan(&self, repo_path: &Path, repo_id: &str) -> Result<ScanOutput, CoreError> {
+        let findings = analyze_tree(repo_path, repo_id);
+        Ok(ScanOutput {
+            findings,
+            sbom_entries: Vec::new(),
+        })
+    }
+}
+
+/// Walk a PLC project tree and produce findings.
+pub(crate) fn analyze_tree(root: &Path, repo_id: &str) -> Vec<Finding> {
+    let mut findings = Vec::new();
+    for entry in walkdir::WalkDir::new(root)
+        .into_iter()
+        .filter_map(|e| e.ok())
+    {
+        if !entry.file_type().is_file() {
+            continue;
+        }
+        let path = entry.path();
+        let ext = path
+            .extension()
+            .and_then(|e| e.to_str())
+            .unwrap_or("")
+            .to_ascii_lowercase();
+        let is_st = matches!(ext.as_str(), "st" | "iecst" | "scl" | "exp" | "il");
+        let is_xml = matches!(ext.as_str(), "xml" | "plcopen" | "project");
+        if !is_st && !is_xml {
+            continue;
+        }
+        let Ok(content) = std::fs::read_to_string(path) else {
+            continue;
+        };
+        let pous = if is_xml {
+            plcopen::parse_plcopen(&content)
+        } else {
+            parser::parse(&content)
+        };
+        if pous.is_empty() {
+            continue;
+        }
+        let rel = path
+            .strip_prefix(root)
+            .unwrap_or(path)
+            .to_string_lossy()
+            .to_string();
+        for pou in &pous {
+            for hit in rules::analyze(pou) {
+                let line_s = hit.line.to_string();
+                let fingerprint =
+                    dedup::compute_fingerprint(&[repo_id, &rel, hit.rule_id, &pou.name, &line_s]);
+                let mut f = Finding::new(
+                    repo_id.to_string(),
+                    fingerprint,
+                    "plc-control-logic".to_string(),
+                    ScanType::PlcControlLogic,
+                    hit.title,
+                    hit.description,
+                    hit.severity,
+                );
+                f.file_path = Some(rel.clone());
+                f.line_number = Some(hit.line);
+                f.rule_id = Some(hit.rule_id.to_string());
+                f.cwe = hit.cwe.map(String::from);
+                f.remediation = Some(hit.remediation.to_string());
+                findings.push(f);
+            }
+        }
+    }
+    findings
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use std::collections::HashSet;
+    use std::path::PathBuf;
+
+    fn demo_dir() -> PathBuf {
+        PathBuf::from(env!("CARGO_MANIFEST_DIR"))
+            .parent()
+            .expect("workspace root")
+            .join("examples/plc-demo")
+    }
+
+    #[test]
+    fn scans_demo_project_end_to_end() {
+        let findings = analyze_tree(&demo_dir(), "demo-target");
+        assert!(!findings.is_empty(), "demo project should produce findings");
+
+        let rules: HashSet<&str> = findings
+            .iter()
+            .filter_map(|f| f.rule_id.as_deref())
+            .collect();
+        for r in [
+            "plc-hardcoded-credential",
+            "plc-default-password",
+            "plc-safety-bypass",
+            "plc-array-unchecked-index",
+            "plc-insecure-comm",
+            "plc-insecure-protocol-port",
+            "plc-unstructured-jump",
+            "plc-division-by-zero",
+        ] {
+            assert!(rules.contains(r), "expected rule {r}; got {rules:?}");
+        }
+
+        // Every finding is well-formed for storage.
+        for f in &findings {
+            assert_eq!(f.repo_id, "demo-target");
+            assert!(f.file_path.is_some(), "finding needs a file");
+            assert!(f.line_number.is_some(), "finding needs a line");
+        }
+
+        // The guarded division (IF ScaleFactor <> 0.0) must not be double-counted:
+        // exactly one division-by-zero (the unguarded MeasuredFlow divide).
+        let div0 = findings
+            .iter()
+            .filter(|f| f.rule_id.as_deref() == Some("plc-division-by-zero"))
+            .count();
+        assert_eq!(div0, 1, "only the unguarded division should be flagged");
+    }
+
+    /// The realistic OpenPLC-style traffic-light sample is mostly sound control
+    /// logic: the scanner must surface its few genuine defects and stay quiet on
+    /// the timed state machine and the guarded duty-cycle division.
+    #[test]
+    fn realistic_sample_flags_only_real_issues() {
+        let all = analyze_tree(&demo_dir(), "demo-target");
+        let tl: Vec<_> = all
+            .iter()
+            .filter(|f| {
+                f.file_path
+                    .as_deref()
+                    .is_some_and(|p| p.ends_with("traffic_light.st"))
+            })
+            .collect();
+        assert!(!tl.is_empty(), "traffic_light.st should produce findings");
+
+        let rules: HashSet<&str> = tl.iter().filter_map(|f| f.rule_id.as_deref()).collect();
+        // The three planted defects: hardcoded SCADA password, cleartext Modbus
+        // master (no auth), and a maintenance mode that drops the PedPermit.
+        for r in [
+            "plc-hardcoded-credential",
+            "plc-insecure-comm",
+            "plc-safety-bypass",
+        ] {
+            assert!(rules.contains(r), "expected rule {r}; got {rules:?}");
+        }
+        // Modbus/TCP on 502 is also an insecure-protocol port.
+        assert!(rules.contains("plc-insecure-protocol-port"));
+
+        // Low false positives: the guarded `IF LampCount <> 0` division and the
+        // JMP-free state machine must not trip anything.
+        assert_eq!(
+            tl.iter()
+                .filter(|f| f.rule_id.as_deref() == Some("plc-division-by-zero"))
+                .count(),
+            0,
+            "the guarded duty-cycle division must not be flagged"
+        );
+        assert!(
+            !rules.contains("plc-unstructured-jump"),
+            "the CASE state machine uses no JMP"
+        );
+    }
+}