compliance-scanner-agent/compliance-agent/src/classify/firmware.rs

//! Firmware classification via tramiton.
//!
//! tramiton is the company's firmware build/repro engine; we do not re-implement
//! its detection. We depend on `tramiton-core` directly (same-company IP) and run
//! its provider analysis in-process behind a [`FirmwareDetector`] port, mapping
//! tramiton's `BuildPlan` onto a [`TargetType`]. A deterministic
//! [`MockFirmwareDetector`] backs the tests so CI unit tests need neither the
//! tramiton sources nor a real firmware tree.

use std::path::Path;

use compliance_core::error::CoreError;
use compliance_core::models::{DetectedFact, TargetType};
use compliance_core::traits::ClassifierVerdict;

/// A minimal firmware-detection summary, mapped from tramiton's `BuildPlan`.
/// Kept small and tramiton-independent so the classifier and the test mock don't
/// need to construct a full tramiton plan.
#[derive(Debug, Clone, Default)]
pub struct FirmwareDetection {
    /// The detecting provider (e.g. `zephyr`, `cmake`, `source-archaeology`).
    pub provider: String,
    /// Detection confidence: `low` | `medium` | `high`.
    pub confidence: String,
    /// Build-system label (e.g. `Zephyr`, `ESP-IDF`, `CMake`).
    pub build_system: String,
    /// Framework, when known (`zephyr`, `esp-idf`, `bare-metal`, ...).
    pub framework: Option<String>,
    /// Target board / MCU / arch.
    pub target: FirmwareTarget,
    /// Unresolved gaps in the plan.
    pub gaps: Vec<String>,
}

/// The detected firmware target (board / MCU / arch).
#[derive(Debug, Clone, Default)]
pub struct FirmwareTarget {
    /// Board name.
    pub board: Option<String>,
    /// MCU part.
    pub mcu: Option<String>,
    /// Architecture.
    pub arch: Option<String>,
}

/// A source of tramiton firmware detection.
#[allow(async_fn_in_trait)]
pub trait FirmwareDetector: Send + Sync {
    /// Run detection over a path, returning a firmware detection if tramiton
    /// could form a build plan.
    async fn detect(&self, path: &Path) -> Result<Option<FirmwareDetection>, CoreError>;
}

/// Uses `tramiton-core` in-process. The analysis is blocking (filesystem walk),
/// so it runs on a blocking thread to avoid stalling the async runtime. A path
/// with no recognizable build system yields `Ok(None)`.
pub struct TramitonNative;

impl FirmwareDetector for TramitonNative {
    async fn detect(&self, path: &Path) -> Result<Option<FirmwareDetection>, CoreError> {
        let path = path.to_path_buf();
        let plan = tokio::task::spawn_blocking(move || {
            let repo = tramiton_core::Repo::new(&path);
            tramiton_core::provider::analyze(&repo)
        })
        .await
        .map_err(|e| CoreError::Other(format!("tramiton detect task join error: {e}")))?
        .map_err(|e| CoreError::Other(format!("tramiton analyze error: {e}")))?;
        Ok(plan.map(|bp| detection_from_build_plan(&bp)))
    }
}

/// Map tramiton's `BuildPlan` onto our minimal detection summary.
fn detection_from_build_plan(bp: &tramiton_core::BuildPlan) -> FirmwareDetection {
    FirmwareDetection {
        provider: bp.provider.clone(),
        confidence: bp.confidence.to_string(),
        build_system: bp.build_system.label().to_string(),
        framework: bp.framework.clone(),
        target: FirmwareTarget {
            board: bp.target.board.clone(),
            mcu: bp.target.mcu.clone(),
            arch: bp.target.arch.clone(),
        },
        gaps: bp.gaps.clone(),
    }
}

/// Map a firmware detection to a target type. Framework/build-system signals
/// distinguish RTOS from bare-metal from Yocto.
pub fn detection_to_target_type(det: &FirmwareDetection) -> TargetType {
    let framework = det.framework.as_deref().unwrap_or("").to_lowercase();
    let build_system = det.build_system.to_lowercase();
    let signal = format!("{framework} {build_system} {}", det.provider.to_lowercase());

    const RTOS: [&str; 6] = ["zephyr", "esp-idf", "freertos", "nuttx", "riot", "chibios"];
    if signal.contains("bitbake") || signal.contains("yocto") || signal.contains("openembedded") {
        TargetType::EmbeddedLinuxYocto
    } else if RTOS.iter().any(|k| signal.contains(k)) {
        TargetType::FirmwareRtos
    } else {
        TargetType::FirmwareBareMetal
    }
}

/// Map tramiton's confidence label to a `[0,1]` score.
fn confidence_score(label: &str) -> f32 {
    match label.to_lowercase().as_str() {
        "high" => 0.9,
        "medium" => 0.6,
        "low" => 0.3,
        _ => 0.4,
    }
}

/// Turn a firmware detection into a classifier verdict, carrying the MCU / board
/// / build-system as facts.
pub fn detection_to_verdict(det: &FirmwareDetection) -> ClassifierVerdict {
    let target_type = detection_to_target_type(det);
    let mut facts = vec![DetectedFact::new(
        "build_system",
        det.build_system.clone(),
        "tramiton",
    )];
    if let Some(fw) = &det.framework {
        facts.push(DetectedFact::new("framework", fw.clone(), "tramiton"));
    }
    if let Some(mcu) = &det.target.mcu {
        facts.push(DetectedFact::new("mcu", mcu.clone(), "tramiton"));
    }
    if let Some(board) = &det.target.board {
        facts.push(DetectedFact::new("board", board.clone(), "tramiton"));
    }
    if let Some(arch) = &det.target.arch {
        facts.push(DetectedFact::new("arch", arch.clone(), "tramiton"));
    }
    ClassifierVerdict {
        target_type,
        confidence: confidence_score(&det.confidence),
        facts,
        rationale: format!(
            "tramiton detected build system '{}'{}",
            det.build_system,
            det.framework
                .as_ref()
                .map(|f| format!(" (framework {f})"))
                .unwrap_or_default()
        ),
    }
}

/// A deterministic [`FirmwareDetector`] for tests — returns a preset detection.
pub struct MockFirmwareDetector {
    /// The detection to return (or `None` for "no detection").
    pub detection: Option<FirmwareDetection>,
}

impl FirmwareDetector for MockFirmwareDetector {
    async fn detect(&self, _path: &Path) -> Result<Option<FirmwareDetection>, CoreError> {
        Ok(self.detection.clone())
    }
}

#[cfg(test)]
#[allow(clippy::expect_used, clippy::unwrap_used)]
mod tests {
    use super::*;

    fn detection(build_system: &str, framework: Option<&str>) -> FirmwareDetection {
        FirmwareDetection {
            provider: build_system.to_string(),
            confidence: "high".to_string(),
            build_system: build_system.to_string(),
            framework: framework.map(|s| s.to_string()),
            target: FirmwareTarget {
                mcu: Some("stm32f429".to_string()),
                ..Default::default()
            },
            gaps: Vec::new(),
        }
    }

    #[test]
    fn zephyr_maps_to_rtos() {
        assert_eq!(
            detection_to_target_type(&detection("zephyr", Some("zephyr"))),
            TargetType::FirmwareRtos
        );
    }

    #[test]
    fn bare_cmake_maps_to_bare_metal() {
        assert_eq!(
            detection_to_target_type(&detection("cmake", Some("bare-metal"))),
            TargetType::FirmwareBareMetal
        );
    }

    #[test]
    fn bitbake_maps_to_yocto() {
        assert_eq!(
            detection_to_target_type(&detection("bitbake", None)),
            TargetType::EmbeddedLinuxYocto
        );
    }

    #[test]
    fn verdict_carries_mcu_fact_and_confidence() {
        let v = detection_to_verdict(&detection("esp-idf", Some("esp-idf")));
        assert_eq!(v.target_type, TargetType::FirmwareRtos);
        assert!((v.confidence - 0.9).abs() < f32::EPSILON);
        assert!(v
            .facts
            .iter()
            .any(|f| f.key == "mcu" && f.value == "stm32f429"));
    }
}