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

//! 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;
pub mod sbom;

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"
        );
    }

    /// Graphical logic must be analysed too: an FBD POU (blocks + in/out
    /// variables) is translated to synthetic ST, so the same rules fire on the
    /// cleartext Modbus block, the hardcoded HMI password and the safety write.
    #[test]
    fn fbd_graphical_body_is_analysed() {
        let all = analyze_tree(&demo_dir(), "demo-target");
        let fbd: Vec<_> = all
            .iter()
            .filter(|f| {
                f.file_path
                    .as_deref()
                    .is_some_and(|p| p.ends_with("pump_fbd.xml"))
            })
            .collect();
        assert!(
            !fbd.is_empty(),
            "pump_fbd.xml (FBD) should produce findings"
        );

        let rules: HashSet<&str> = fbd.iter().filter_map(|f| f.rule_id.as_deref()).collect();
        for r in [
            "plc-insecure-comm",          // Modbus_TCP_Master(AUTH := FALSE)
            "plc-insecure-protocol-port", // PORT := 502
            "plc-hardcoded-credential",   // HmiPassword := 'admin123'
            "plc-safety-bypass",          // Safety_Enable := FALSE
        ] {
            assert!(
                rules.contains(r),
                "expected rule {r} from FBD; got {rules:?}"
            );
        }
    }
}