rafia_stpa/stpa/

constraints.rs

// *****************************************************************************
// Copyright (c) 2025 Codethink
//
// This program and the accompanying materials are made available under the
// terms of the Eclipse Public License 2.0 which is available at
// http://www.eclipse.org/legal/epl-2.0.
//
// This Source Code may also be made available under the following Secondary
// Licenses when the conditions for such availability set forth in the Eclipse
// Public License, v. 2.0 are satisfied: GNU General Public License, version 2
// with the GNU Classpath Exception which is
// available at https://www.gnu.org/software/classpath/license.html.
//
// SPDX-License-Identifier: EPL-2.0 OR GPL-2.0 WITH Classpath-exception-2.0
// *****************************************************************************
use std::{
    collections::HashSet,
    ffi::OsStr,
    fmt::{self, Display},
    fs::{self, remove_file, File, OpenOptions},
    io::{self, Write},
    option::Option,
    path::Path,
    str::FromStr,
    sync::Arc,
};

#[cfg(feature = "clap")]
use clap::ValueEnum;
use csv;
#[cfg(feature = "pyo3")]
use pyo3::prelude::*;
use serde::{Deserialize, Serialize};
use serde_yaml;
use strum::EnumIter;
use thiserror::Error;

#[cfg(feature = "pyo3")]
use crate::trudag::{
    create_link, create_node, find_root, remove_node, rooted_relative_path, set_node,
    set_node_links, RootError, TrudagError,
};
use crate::{
    define_default_link_type_impls,
    stpa::{LoadError, CYCLE_DETECTED_MSG},
};

use super::{CausalScenarioLink, ConstraintLink, HazardLink, StpaData, StpaProject};

#[derive(Debug, Clone)]
#[cfg_attr(feature = "clap", derive(ValueEnum))]
pub enum RefType {
    File,
    StpaDescribe,
}

impl Display for RefType {
    fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result {
        match self {
            RefType::File => write!(f, "file"),
            RefType::StpaDescribe => write!(f, "stpa-describe"),
        }
    }
}

impl FromStr for RefType {
    type Err = ();

    fn from_str(value: &str) -> Result<Self, Self::Err> {
        match value.to_lowercase().as_str() {
            "file" => Ok(RefType::File),
            "stpa-describe" => Ok(RefType::StpaDescribe),
            _ => Err(()),
        }
    }
}

#[derive(Clone, Hash, Eq, PartialEq, Ord, PartialOrd, Debug, Serialize, Deserialize)]
pub struct UcaLink(Arc<str>);

#[derive(Clone, Hash, Eq, PartialEq, Ord, PartialOrd, Debug, Serialize, Deserialize)]
pub struct CsLink(Arc<str>);

#[derive(Clone, Hash, Eq, PartialEq, Ord, PartialOrd, Debug, Serialize, Deserialize)]
pub struct TsfLink(Arc<str>);

define_default_link_type_impls!(TsfLink);
define_default_link_type_impls!(UcaLink);
define_default_link_type_impls!(CsLink);

#[derive(Debug, Serialize, Deserialize)]
pub struct ConstraintCsv {
    #[serde(rename = "Constraint Id")]
    pub id: String,
    #[serde(rename = "Description")]
    pub description: String,
    #[serde(rename = "Constraint Type")]
    pub constraint_type: String,
    #[serde(rename = "Link to Constraint(s)")]
    pub link_to_contraints: String,
    #[serde(rename = "Link to Hazard(s)")]
    pub link_to_hazards: String,
    #[serde(rename = "Links to UCA")]
    pub link_to_uca: String,
    #[serde(rename = "Links to CS")]
    pub link_to_cs: String,
    #[serde(rename = "Links to TSF")]
    pub link_to_tsf: String,
}

// This can represent both the file and the stpa reference
// We could just have it as a HashMap<String, String> so this
// struct enforces that only valid keys can from ether can be used
// but it does not stop some keys from one and other keys from the other
// we hope to only support the stpa reference in the future so instead
// of a more complex enum in the short term, the optional fields are used.
// The field usage is as follows:
// * trudag_type is always required and should be `file` or `stpa`.
// * path must be used for file.
// * sha can be used for file.
// * project and item must be used for item.
// One we only need to support stpa/describe references we will remove the
// options, and have cocreate fields
#[derive(Debug, Serialize, Deserialize)]
struct TrudagRef {
    #[serde(rename = "type")]
    trudag_type: String,
    #[serde(skip_serializing_if = "Option::is_none")]
    path: Option<String>,
    #[serde(skip_serializing_if = "Option::is_none")]
    project: Option<String>,
    #[serde(skip_serializing_if = "Option::is_none")]
    sha: Option<String>,
    #[serde(skip_serializing_if = "Option::is_none")]
    item: Option<String>,
}

#[derive(Debug, Serialize, Deserialize)]
struct TrudagYaml {
    references: Vec<TrudagRef>,
    level: String,
    normative: bool,
}

fn to_trudag_id(input: &str) -> String {
    input.replace(".", "_").replace("-", "_").to_uppercase()
}

#[derive(Debug, PartialEq, Clone, Serialize, Deserialize)]
pub struct Constraint {
    pub id: String,
    pub description: String,
    pub constraint_type: ConstraintType,
    pub link_to_constraints: Vec<ConstraintLink>,
    pub link_to_hazards: Vec<HazardLink>,
    pub link_to_uca: Vec<UcaLink>,
    pub link_to_cs: Vec<CausalScenarioLink>,
    pub link_to_tsf: Vec<String>,
}

impl Constraint {
    pub fn describe(
        &self,
        stpa_data: &StpaData,
        currently_expanding: &mut Vec<String>,
    ) -> Result<serde_json::Value, String> {
        // check we aren't currently expanding this constraint
        if currently_expanding.contains(&self.id) {
            return Ok(serde_json::json!({
                "constraint": {
                    "id": self.id,
                    "message": CYCLE_DETECTED_MSG,
                }
            }));
        } else {
            currently_expanding.push(self.id.clone());
        }

        let mut obj = serde_json::json!({"constraint": self});

        let constraints = self
            .link_to_constraints
            .iter()
            .map(|constraint| {
                stpa_data.describe_inner(constraint.0.as_ref(), currently_expanding).map_err(|e| {
                    format!(
                        "Failed while resolving link to constraint id {constraint:?} of constraint {:?}: {e}",
                        self.id
                    )
                })
            })
            .collect::<Result<Vec<_>, _>>()?;
        obj["constraint"]["link_to_constraints"] = serde_json::json!(constraints);

        let hazards = self
            .link_to_hazards
            .iter()
            .map(|hazard| {
                stpa_data
                    .describe_inner(hazard.0.as_ref(), currently_expanding)
                    .map_err(|e| {
                        format!(
                            "Failed while resolving hazard id {hazard:?} of constraint {:?}: {e}",
                            self.id
                        )
                    })
            })
            .collect::<Result<Vec<_>, _>>()?;
        obj["constraint"]["link_to_hazards"] = serde_json::json!(hazards);

        let ucas = self
            .link_to_uca
            .iter()
            .map(|uca| {
                stpa_data
                    .describe_inner(uca.0.as_ref(), currently_expanding)
                    .map_err(|e| {
                        format!(
                            "Failed while resolving UCA id {uca:?} of constraint {:?}: {e}",
                            self.id
                        )
                    })
            })
            .collect::<Result<Vec<_>, _>>()?;
        obj["constraint"]["link_to_uca"] = serde_json::json!(ucas);

        let cs = self
            .link_to_cs
            .iter()
            .map(|cs| {
                stpa_data
                    .describe_inner(cs.0.as_ref(), currently_expanding)
                    .map_err(|e| {
                        format!(
                            "Failed while resolving CS id {cs:?} of constraint {:?}: {e}",
                            self.id
                        )
                    })
            })
            .collect::<Result<Vec<_>, _>>()?;
        obj["constraint"]["link_to_cs"] = serde_json::json!(cs);

        let popped = currently_expanding.pop();
        debug_assert_eq!(popped.as_ref(), Some(&self.id));
        Ok(obj)
    }

    /// Get the file name that this Constraint will be saved in
    pub fn trudag_filename(&self, group: &str) -> String {
        format!("{}.md", self.trudag_name(group))
    }

    /// Get the trudag nane of this constraint when exported to a trudag file
    ///
    /// For a file in a TSF graph `some/folder/GROUP-ID.md` the name becomes `GROUP-ID`
    pub fn trudag_name(&self, group: &str) -> String {
        format!("{group}-{}", self.trudag_id())
    }

    /// Get the trudag id of this constraint when exported to a trudag file
    ///
    /// For a file in a TSF graph `some/folder/GROUP-ID.md`
    pub fn trudag_id(&self) -> String {
        to_trudag_id(&self.id)
    }

    #[cfg(feature = "pyo3")]
    /// Create the contents of a trudag element based on this constraint
    ///
    /// This function can create items with ether basic file refrernces
    /// or describe based references.
    fn to_trudag(
        &self,
        root: &Path,
        stpa_project: &StpaProject,
        level: String,
        reftype: &RefType,
    ) -> String {
        let stpa_project_file =
            rooted_relative_path(root, stpa_project.project_file.as_ref().unwrap())
                .unwrap()
                .to_string_lossy()
                .to_string();
        let constraint_path = rooted_relative_path(
            root,
            &stpa_project
                .root
                .as_ref()
                .unwrap()
                .join(stpa_project.constraints.as_ref().unwrap()),
        )
        .unwrap()
        .to_string_lossy()
        .to_string();
        let references = match reftype {
            RefType::File => {
                let mut full_refs = vec![TrudagRef {
                    path: Some(constraint_path),
                    trudag_type: "file".to_string(),
                    sha: None,
                    item: None,
                    project: None,
                }];

                if !self.link_to_hazards.is_empty() {
                    let hazard_path = rooted_relative_path(
                        root,
                        &stpa_project
                            .root
                            .as_ref()
                            .unwrap()
                            .join(stpa_project.hazards.as_ref().unwrap()),
                    )
                    .unwrap()
                    .to_string_lossy()
                    .to_string();
                    full_refs.push(TrudagRef {
                        trudag_type: "file".to_string(),
                        path: Some(hazard_path),
                        sha: None,
                        item: None,
                        project: None,
                    })
                }
                if !self.link_to_uca.is_empty() {
                    let uca_path = rooted_relative_path(
                        root,
                        &stpa_project
                            .root
                            .as_ref()
                            .unwrap()
                            .join(stpa_project.uca.as_ref().unwrap()),
                    )
                    .unwrap()
                    .to_string_lossy()
                    .to_string();
                    full_refs.push(TrudagRef {
                        trudag_type: "file".to_string(),
                        path: Some(uca_path),
                        sha: None,
                        item: None,
                        project: None,
                    })
                }
                if !self.link_to_cs.is_empty() {
                    let scenario_path = rooted_relative_path(
                        root,
                        &stpa_project
                            .root
                            .as_ref()
                            .unwrap()
                            .join(stpa_project.scenarios.as_ref().unwrap()),
                    )
                    .unwrap()
                    .to_string_lossy()
                    .to_string();
                    full_refs.push(TrudagRef {
                        trudag_type: "file".to_string(),
                        path: Some(scenario_path),
                        sha: None,
                        item: None,
                        project: None,
                    })
                }
                full_refs
            }
            RefType::StpaDescribe => vec![TrudagRef {
                project: Some(stpa_project_file),
                path: None,
                trudag_type: "stpa-describe".to_string(),
                sha: None,
                item: Some(self.id.to_string()),
            }],
        };
        let meta = TrudagYaml {
            references,
            level,
            normative: true,
        };

        let meta = serde_yaml::to_string(&meta).unwrap();
        let mut out = String::new();
        out.push_str("---\n");
        out.push_str(&meta);

        out.push_str("---\n\n");
        out.push_str(&self.description);
        out.push('\n');

        out
    }
}

impl From<ConstraintCsv> for Constraint {
    fn from(csv: ConstraintCsv) -> Self {
        Constraint {
            id: csv.id,
            description: csv.description,
            constraint_type: ConstraintType::from(csv.constraint_type.as_str()),
            link_to_constraints: csv
                .link_to_contraints
                .split(",")
                .filter(|e| !e.is_empty())
                .map(|s| ConstraintLink::from(s.trim()))
                .collect(),
            link_to_hazards: csv
                .link_to_hazards
                .split(",")
                .filter(|e| !e.is_empty())
                .map(|s| HazardLink::from(s.trim()))
                .collect(),
            link_to_uca: csv
                .link_to_uca
                .split(",")
                .filter(|e| !e.is_empty())
                .map(|s| UcaLink::from(s.trim()))
                .collect(),
            link_to_cs: csv
                .link_to_cs
                .split(",")
                .filter(|e| !e.is_empty())
                .map(|s| CausalScenarioLink::from(s.trim()))
                .collect(),
            link_to_tsf: csv
                .link_to_tsf
                .split(",")
                .filter(|e| !e.is_empty())
                .map(|s| s.trim().to_owned())
                .collect(),
        }
    }
}

#[cfg(feature = "pyo3")]
#[derive(Error, Debug)]
pub enum ToTrudagError {
    #[error("Could not find root")]
    Root(#[from] RootError),
    #[error("Project does not exist")]
    NoProject(io::Error, String),
    #[error("Could not create a link")]
    Create(TrudagError, String),
    #[error("Could not link a item")]
    Link(TrudagError, String, String),
    #[error("Could not remove a item")]
    Remove(TrudagError, String),
    #[error("Could not clear a item")]
    Clear(TrudagError, String),
    #[error("Could not save")]
    Save(#[from] std::io::Error),
    #[error("Python trudag error")]
    ErrorPython(#[from] PyErr),
}

#[derive(Debug, PartialEq, Clone, Serialize, Deserialize, Default)]
pub struct Constraints {
    pub constraints: Vec<Constraint>,
}

impl Constraints {
    pub fn from_yaml(text: &str) -> Result<Constraints, LoadError> {
        serde_yaml::from_str::<Constraints>(text).map_err(LoadError::Yaml)
    }
    pub fn to_yaml(&self) -> Result<String, LoadError> {
        serde_yaml::to_string(&self).map_err(LoadError::Yaml)
    }
    pub fn from_csv(filename: &Path) -> Result<Constraints, LoadError> {
        let mut constraints = vec![];
        let file = File::open(filename)?;
        for constraint in csv::ReaderBuilder::new()
            .from_reader(file)
            .deserialize::<ConstraintCsv>()
        {
            constraints.push(constraint?.into());
        }
        Ok(Constraints { constraints })
    }
    pub fn from_file(filename: &Path) -> Result<Constraints, LoadError> {
        match filename.extension().and_then(OsStr::to_str) {
            Some("csv") => Self::from_csv(filename),
            Some("yml") | Some("yaml") => {
                let text = fs::read_to_string(filename)?;
                Self::from_yaml(&text)
            }
            _ => Err(LoadError::NotSupportedFormat),
        }
    }
    #[cfg(feature = "pyo3")]
    pub fn to_trudag(
        &self,
        folder: &Path,
        stpa_project: &StpaProject,
        group: &str,
        reftype: RefType,
        set_links: bool,
    ) -> Result<(), ToTrudagError> {
        let folder = folder.canonicalize().map_err(|e| {
            ToTrudagError::NoProject(
                e,
                folder
                    .to_str()
                    .map(|str| str.to_owned())
                    .unwrap_or(format!("{folder:?}")),
            )
        })?;

        let trudag_root = find_root(&folder)?;

        let paths = fs::read_dir(&folder).unwrap();

        let trudag_graph = Python::with_gil(|py| {
            let trudag_dotstop_graph = py.import("trudag.dotstop.core.graph")?;
            let from_dir = trudag_dotstop_graph.getattr("build_trustable_graph")?;
            let trudag_graph = from_dir.call(
                (trudag_root.join(".dotstop.dot"), trudag_root.clone()),
                None,
            )?;
            let trudag_graph = trudag_graph.unbind();
            PyResult::Ok(trudag_graph)
        })?;

        // Remove any items we no longer use to avoid anything deleted in the stpa from
        // being left in the trustable report
        let mut reused_constraints = HashSet::new();
        for path in paths {
            let path = path.unwrap();
            let path = path.path();
            if path.is_file() {
                let constraint_file_name = path.file_name().unwrap().to_string_lossy();
                let constraint = path.file_stem().unwrap().to_string_lossy();
                if let Some(existing) = self.constraints.iter().find(|cmp_constraint| {
                    cmp_constraint.trudag_filename(group).as_str() == constraint_file_name
                }) {
                    reused_constraints.insert(existing.id.to_owned());
                } else {
                    remove_node(&trudag_root, &trudag_graph, &constraint)
                        .map_err(|e| ToTrudagError::Remove(e, constraint.to_string()))?;

                    remove_file(&path)?;
                    println!("Removed {path:?}");
                }
            }
        }

        for (index, constraint) in self.constraints.iter().enumerate() {
            if !reused_constraints.contains(&constraint.id) {
                create_node(
                    &trudag_root,
                    &trudag_graph,
                    group,
                    None,
                    &folder,
                    constraint,
                )
                .map_err(|e| ToTrudagError::Create(e, format!("{constraint:?}")))?;
            }
            let filename = folder.join(constraint.trudag_filename(group));
            let trudag = constraint.to_trudag(
                &trudag_root,
                stpa_project,
                format!("1.{}", index + 1),
                &reftype,
            );
            let mut f = OpenOptions::new()
                .write(true)
                .truncate(true)
                .open(filename)?;
            f.write_all(trudag.as_bytes())?;
            println!("Created {}", constraint.id)
        }

        let trudag_graph = Python::with_gil(|py| {
            let trudag_dotstop_graph = py.import("trudag.dotstop.core.graph")?;
            let from_dir = trudag_dotstop_graph.getattr("build_trustable_graph")?;
            let trudag_graph = from_dir.call(
                (trudag_root.join(".dotstop.dot"), trudag_root.clone()),
                None,
            )?;
            let trudag_graph = trudag_graph.unbind();
            PyResult::Ok(trudag_graph)
        })?;

        // We create and then link in two loops as we need to ensure that elements are created,
        // before we link to them.
        for constraint in &self.constraints {
            for tsf in &constraint.link_to_tsf {
                create_link(&trudag_root, &trudag_graph, group, constraint, tsf).map_err(|e| {
                    ToTrudagError::Link(e, format!("{constraint:?}"), tsf.to_string())
                })?;
            }
            for constraint_link in &constraint.link_to_constraints {
                create_link(
                    &trudag_root,
                    &trudag_graph,
                    group,
                    constraint,
                    &format!("{group}-{}", to_trudag_id(&constraint_link.0)),
                )
                .map_err(|e| {
                    ToTrudagError::Link(e, format!("{constraint:?}"), constraint_link.0.to_string())
                })?;
            }
            println!("Linked {}", constraint.id)
        }
        // Todo: Does this need to be done after the links or could it be done in the first loop
        for constraint in &self.constraints {
            set_node(&trudag_root, &trudag_graph, group, constraint)
                .map_err(|e| ToTrudagError::Clear(e, format!("{constraint:?}")))?;
            if set_links {
                set_node_links(&trudag_root, &trudag_graph, group, constraint)
                    .map_err(|e| ToTrudagError::Clear(e, format!("{constraint:?}")))?;
            }
            println!("Set {}", constraint.id)
        }
        Ok(())
    }
    pub fn find(&self, id: &str) -> Option<&Constraint> {
        self.constraints
            .iter()
            .find(|constraint| constraint.id == id)
    }
}

#[derive(Debug, Clone, Copy, Serialize, Deserialize, PartialEq, EnumIter)]
pub enum ConstraintType {
    SystemLevelConstraint,
    ControllerFunctionalConstraint,
    CausalScenarioConstraint,
    Missing,
}

impl fmt::Display for ConstraintType {
    fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result {
        match self {
            ConstraintType::SystemLevelConstraint => {
                write!(f, "System Level Constraint")
            }
            ConstraintType::ControllerFunctionalConstraint => {
                write!(f, "Constroller Functional Constartint")
            }
            ConstraintType::CausalScenarioConstraint => {
                write!(f, "Causal Scenario Constraint")
            }
            ConstraintType::Missing => {
                write!(f, "Constraint not definded or missing")
            }
        }
    }
}

impl From<&str> for ConstraintType {
    fn from(raw: &str) -> Self {
        match raw.trim() {
            "SLC" => ConstraintType::SystemLevelConstraint,
            "CFC" => ConstraintType::ControllerFunctionalConstraint,
            "CSC" => ConstraintType::CausalScenarioConstraint,
            _ => ConstraintType::Missing,
        }
    }
}