csv_to_oscal_cd

trestle.tasks.csv_to_oscal_cd

OSCAL transformation tasks.

Attributes

ADJUSTED_RISK_RATING = 'Adjusted_Risk_Rating' module-attribute

CHECK_DESCRIPTION = 'Check_Description' module-attribute

CHECK_ID = 'Check_Id' module-attribute

COMPONENT_DESCRIPTION = 'Component_Description' module-attribute

COMPONENT_TITLE = 'Component_Title' module-attribute

COMPONENT_TYPE = 'Component_Type' module-attribute

CONTROL_ID_LIST = 'Control_Id_List' module-attribute

FETCHER_DESCRIPTION = 'Fetcher_Description' module-attribute

FETCHER_ID = 'Fetcher_Id' module-attribute

HEADER_DECORATION_CHAR = '$' module-attribute

NAMESPACE = 'Namespace' module-attribute

ORIGINAL_RISK_RATING = 'Original_Risk_Rating' module-attribute

PARAMETER = 'Parameter' module-attribute

PARAMETER_DESCRIPTION = f'{PARAMETER}_Description' module-attribute

PARAMETER_ID = f'{PARAMETER}_Id' module-attribute

PARAMETER_VALUE_ALTERNATIVES = f'{PARAMETER}_Value_Alternatives' module-attribute

PARAMETER_VALUE_DEFAULT = f'{PARAMETER}_Value_Default' module-attribute

PROFILE_DESCRIPTION = 'Profile_Description' module-attribute

PROFILE_SOURCE = 'Profile_Source' module-attribute

RISK_ADJUSTMENT = 'Risk_Adjustment' module-attribute

RULE_DESCRIPTION = 'Rule_Description' module-attribute

RULE_ID = 'Rule_Id' module-attribute

Row = Iterator[List[str]] module-attribute

TARGET_COMPONENT = 'Target_Component' module-attribute

logger = logging.getLogger(__name__) module-attribute

prefix_rule_set = 'rule_set_' module-attribute

validation = 'validation' module-attribute

Classes

CsvColumn

CsvColumn.

Source code in trestle/tasks/csv_to_oscal_cd.py

class CsvColumn():
    """CsvColumn."""

    _columns_required = [
        f'{COMPONENT_TITLE}',
        f'{COMPONENT_DESCRIPTION}',
        f'{COMPONENT_TYPE}',
        f'{RULE_ID}',
        f'{RULE_DESCRIPTION}',
        f'{PROFILE_SOURCE}',
        f'{PROFILE_DESCRIPTION}',
        f'{CONTROL_ID_LIST}',
        f'{NAMESPACE}',
    ]

    _columns_required_validation = [
        f'{COMPONENT_TITLE}',
        f'{COMPONENT_DESCRIPTION}',
        f'{COMPONENT_TYPE}',
        f'{RULE_ID}',
        f'{NAMESPACE}',
        f'{CHECK_ID}',
        f'{CHECK_DESCRIPTION}',
    ]

    _columns_optional = [
        f'{CHECK_ID}',
        f'{CHECK_DESCRIPTION}',
        f'{TARGET_COMPONENT}',
        f'{ORIGINAL_RISK_RATING}',
        f'{ADJUSTED_RISK_RATING}',
        f'{RISK_ADJUSTMENT}',
    ]

    _columns_parameter = [
        f'{PARAMETER_ID}',
        f'{PARAMETER_DESCRIPTION}',
        f'{PARAMETER_VALUE_ALTERNATIVES}',
        f'{PARAMETER_VALUE_DEFAULT}',
    ]

    _columns_ordered = _columns_required + _columns_parameter + _columns_optional

    @staticmethod
    def get_order(column_name: str) -> int:
        """Get order for column_name."""
        rval = sys.maxsize
        if column_name in CsvColumn._columns_ordered:
            rval = CsvColumn._columns_ordered.index(column_name)
        return rval

    @staticmethod
    def is_column_name_required(name: str) -> bool:
        """Is column name required."""
        return name in (CsvColumn._columns_required + CsvColumn._columns_required_validation)

    @staticmethod
    def is_column_name_optional(name: str) -> bool:
        """Is column name optional."""
        return name in (CsvColumn._columns_optional)

    @staticmethod
    def is_column_name_parameter(name: str) -> bool:
        """Is column name parameter."""
        for cname in CsvColumn._columns_parameter:
            if name.startswith(cname):
                return True
        return False

    @staticmethod
    def get_required_column_names() -> List[str]:
        """Get required column names."""
        rval = []
        rval += CsvColumn._columns_required
        return rval

    @staticmethod
    def get_optional_column_names() -> List[str]:
        """Get optional column names."""
        rval = []
        rval += CsvColumn._columns_optional
        return rval

    @staticmethod
    def get_parameter_column_names() -> List[str]:
        """Get parameter column names."""
        rval = []
        rval += CsvColumn._columns_parameters
        return rval

    @staticmethod
    def get_required_column_names_validation() -> List[str]:
        """Get required column names validation."""
        rval = []
        rval += CsvColumn._columns_required_validation
        return rval

    _rule_property_column_names = [
        f'{RULE_ID}',
        f'{RULE_DESCRIPTION}',
        f'{PARAMETER_ID}',
        f'{PARAMETER_DESCRIPTION}',
        f'{PARAMETER_VALUE_ALTERNATIVES}',
        f'{CHECK_ID}',
        f'{CHECK_DESCRIPTION}',
        f'{TARGET_COMPONENT}',
        f'{ORIGINAL_RISK_RATING}',
        f'{ADJUSTED_RISK_RATING}',
        f'{RISK_ADJUSTMENT}',
    ]

    @staticmethod
    def get_rule_property_column_names() -> List[str]:
        """Get rule property column names."""
        return CsvColumn._rule_property_column_names

    # columns required which do not become properties
    _columns_required_filtered = [
        f'{COMPONENT_TITLE}',
        f'{COMPONENT_DESCRIPTION}',
        f'{COMPONENT_TYPE}',
        f'{PROFILE_SOURCE}',
        f'{PROFILE_DESCRIPTION}',
        f'{CONTROL_ID_LIST}',
        f'{NAMESPACE}',
    ]

    # optional columns which do not become properties, initially
    _columns_optional_filtered = []

    _columns_filtered = _columns_required_filtered + _columns_optional_filtered

    @staticmethod
    def get_filtered_required_column_names() -> List[str]:
        """Get filtered required column names."""
        rval = []
        for column_name in CsvColumn.get_required_column_names():
            if column_name not in CsvColumn._columns_filtered:
                rval.append(column_name)
        return rval

    @staticmethod
    def get_filtered_optional_column_names() -> List[str]:
        """Get filtered optional column names."""
        rval = []
        for column_name in CsvColumn.get_optional_column_names():
            if column_name not in CsvColumn._columns_filtered:
                rval.append(column_name)
        return rval

    _check_property_column_names = [
        f'{RULE_ID}',
        f'{CHECK_ID}',
        f'{CHECK_DESCRIPTION}',
        f'{TARGET_COMPONENT}',
    ]

    @staticmethod
    def get_check_property_column_names() -> List[str]:
        """Get check property column names."""
        return CsvColumn._check_property_column_names

    # optional columns which do become properties, afterwards
    _columns_parameters = [
        f'{PARAMETER_ID}',
        f'{PARAMETER_DESCRIPTION}',
        f'{PARAMETER_VALUE_ALTERNATIVES}',
    ]

    # optional columns which require Param_Id be present in the row
    _columns_parameters_dependent = [
        f'{PARAMETER_DESCRIPTION}',
        f'{PARAMETER_VALUE_ALTERNATIVES}',
        f'{PARAMETER_VALUE_DEFAULT}',
    ]

Functions

get_check_property_column_names() staticmethod

Get check property column names.

Source code in trestle/tasks/csv_to_oscal_cd.py

@staticmethod
def get_check_property_column_names() -> List[str]:
    """Get check property column names."""
    return CsvColumn._check_property_column_names

get_filtered_optional_column_names() staticmethod

Get filtered optional column names.

Source code in trestle/tasks/csv_to_oscal_cd.py

@staticmethod
def get_filtered_optional_column_names() -> List[str]:
    """Get filtered optional column names."""
    rval = []
    for column_name in CsvColumn.get_optional_column_names():
        if column_name not in CsvColumn._columns_filtered:
            rval.append(column_name)
    return rval

get_filtered_required_column_names() staticmethod

Get filtered required column names.

Source code in trestle/tasks/csv_to_oscal_cd.py

@staticmethod
def get_filtered_required_column_names() -> List[str]:
    """Get filtered required column names."""
    rval = []
    for column_name in CsvColumn.get_required_column_names():
        if column_name not in CsvColumn._columns_filtered:
            rval.append(column_name)
    return rval

get_optional_column_names() staticmethod

Get optional column names.

Source code in trestle/tasks/csv_to_oscal_cd.py

@staticmethod
def get_optional_column_names() -> List[str]:
    """Get optional column names."""
    rval = []
    rval += CsvColumn._columns_optional
    return rval

get_order(column_name) staticmethod

Get order for column_name.

Source code in trestle/tasks/csv_to_oscal_cd.py

@staticmethod
def get_order(column_name: str) -> int:
    """Get order for column_name."""
    rval = sys.maxsize
    if column_name in CsvColumn._columns_ordered:
        rval = CsvColumn._columns_ordered.index(column_name)
    return rval

get_parameter_column_names() staticmethod

Get parameter column names.

Source code in trestle/tasks/csv_to_oscal_cd.py

@staticmethod
def get_parameter_column_names() -> List[str]:
    """Get parameter column names."""
    rval = []
    rval += CsvColumn._columns_parameters
    return rval

get_required_column_names() staticmethod

Get required column names.

Source code in trestle/tasks/csv_to_oscal_cd.py

@staticmethod
def get_required_column_names() -> List[str]:
    """Get required column names."""
    rval = []
    rval += CsvColumn._columns_required
    return rval

get_required_column_names_validation() staticmethod

Get required column names validation.

Source code in trestle/tasks/csv_to_oscal_cd.py

@staticmethod
def get_required_column_names_validation() -> List[str]:
    """Get required column names validation."""
    rval = []
    rval += CsvColumn._columns_required_validation
    return rval

get_rule_property_column_names() staticmethod

Get rule property column names.

Source code in trestle/tasks/csv_to_oscal_cd.py

@staticmethod
def get_rule_property_column_names() -> List[str]:
    """Get rule property column names."""
    return CsvColumn._rule_property_column_names

is_column_name_optional(name) staticmethod

Is column name optional.

Source code in trestle/tasks/csv_to_oscal_cd.py

@staticmethod
def is_column_name_optional(name: str) -> bool:
    """Is column name optional."""
    return name in (CsvColumn._columns_optional)

is_column_name_parameter(name) staticmethod

Is column name parameter.

Source code in trestle/tasks/csv_to_oscal_cd.py

@staticmethod
def is_column_name_parameter(name: str) -> bool:
    """Is column name parameter."""
    for cname in CsvColumn._columns_parameter:
        if name.startswith(cname):
            return True
    return False

is_column_name_required(name) staticmethod

Is column name required.

Source code in trestle/tasks/csv_to_oscal_cd.py

@staticmethod
def is_column_name_required(name: str) -> bool:
    """Is column name required."""
    return name in (CsvColumn._columns_required + CsvColumn._columns_required_validation)

CsvToOscalComponentDefinition

Bases: TaskBase

Task to create OSCAL ComponentDefinition json.

Attributes:

Name	Type	Description
name		Name of the task.

Source code in trestle/tasks/csv_to_oscal_cd.py

class CsvToOscalComponentDefinition(TaskBase):
    """
    Task to create OSCAL ComponentDefinition json.

    Attributes:
        name: Name of the task.
    """

    name = 'csv-to-oscal-cd'

    def __init__(self, config_object: Optional[configparser.SectionProxy]) -> None:
        """
        Initialize trestle task csv-to-oscal-cd.

        Args:
            config_object: Config section associated with the task.
        """
        super().__init__(config_object)

    def print_info(self) -> None:
        """Print the help string."""
        name = self.name
        oscal_name = 'component_definition'
        # help note identifiers
        note01 = '1'
        note02 = '2'
        note03 = '3'
        note04 = '4'
        note05 = '5'
        # help generation
        logger.info(f'Help information for {name} task.')
        logger.info('')
        logger.info(f'Purpose: From csv produce OSCAL {oscal_name} file.')
        logger.info('')
        logger.info('')
        logger.info(f'Configuration flags sit under [task.{name}]:')
        text1 = '  title                = '
        text2 = '(required) the component definition title.'
        logger.info(text1 + text2)
        text1 = '  version              = '
        text2 = '(required) the component definition version.'
        logger.info(text1 + text2)
        text1 = '  csv-file             = '
        text2 = '(required) the path of the csv file.'
        text3 = ' [1st row are column headings; 2nd row are column descriptions; 3rd row and beyond is data]'
        logger.info(text1 + text2 + text3)
        text1 = '  required columns:      '
        for text2 in CsvColumn.get_required_column_names():
            if text2 in [f'{RULE_DESCRIPTION}', f'{PROFILE_SOURCE}', f'{PROFILE_DESCRIPTION}', f'{CONTROL_ID_LIST}']:
                text2 += f' (see note {note01})'
            logger.info(text1 + '$$' + text2)
            text1 = '                         '
        text1 = '  optional columns:      '
        for text2 in CsvColumn.get_optional_column_names():
            if text2 in [f'{ORIGINAL_RISK_RATING}', f'{ADJUSTED_RISK_RATING}', f'{RISK_ADJUSTMENT}']:
                text2 += f' (see note {note01})'
            elif text2 in [f'{TARGET_COMPONENT}']:
                text2 += f' (see note {note03})'
            else:
                text2 += f' (see note {note02})'
            logger.info(text1 + '$' + text2)
            text1 = '                         '
        for text2 in CsvColumn.get_parameter_column_names():
            text2 += f' (see notes {note01}, {note05})'
            logger.info(text1 + '$' + text2)
            text1 = '                         '
        text1 = '  comment columns:       '
        text2 = f'Informational (see note {note04})'
        logger.info(text1 + '#' + text2)
        text1 = '  output-dir           = '
        text2 = '(required) the path of the output directory for synthesized OSCAL .json files.'
        logger.info(text1 + text2)
        text1 = '  component-definition = '
        text2 = '(optional) the path of the existing component-definition OSCAL .json file.'
        logger.info(text1 + text2)
        text1 = '  class.column-name    = '
        text2 = f'(optional) the class to associate with the specified column name, e.g. class.{RULE_ID} = scc_class'
        logger.info(text1 + text2)
        text1 = '  output-overwrite     = '
        text2 = '(optional) true [default] or false; replace existing output when true.'
        logger.info(text1 + text2)
        text1 = '  validate-controls    = '
        text2 = '(optional) on, warn, or off [default]; validate controls exist in resolved profile.'
        logger.info(text1 + text2)
        # Notes
        text1 = ''
        text2 = ''
        logger.info(text1 + text2)
        text1 = 'Notes: '
        text2 = f'[{note01}] column is ignored for validation component type'
        logger.info(text1 + text2)
        text1 = '       '
        text2 = f'[{note02}] column is required for validation component type'
        logger.info(text1 + text2)
        text1 = '       '
        text2 = f'[{note03}] column is optional for validation component type'
        text3 = f', but may be needed to prevent {RULE_ID} collisions'
        logger.info(text1 + text2 + text3)
        text1 = '       '
        text2 = f'[{note04}] column name starting with # causes column to be ignored'
        logger.info(text1 + text2)
        text1 = '       '
        text2 = f'[{note05}] additional parameters are specified by adding a common suffix per set'
        text3 = f', for example: {PARAMETER_ID}_1, {PARAMETER_DESCRIPTION}_1, ...{PARAMETER_ID}_2...'
        logger.info(text1 + text2 + text3)

    def configure(self) -> bool:
        """Configure."""
        self._timestamp = datetime.datetime.now(datetime.timezone.utc).replace(microsecond=0).isoformat()
        # config verbosity
        self._quiet = self._config.get('quiet', False)
        self._verbose = not self._quiet
        # title
        self._title = self._config.get('title')
        if self._title is None:
            logger.warning('config missing "title"')
            return False
        # version
        self._version = self._config.get('version')
        if self._version is None:
            logger.warning('config missing "version"')
            return False
        # config csv
        self._csv_file = self._config.get('csv-file')
        if self._csv_file is None:
            logger.warning('config missing "csv-file"')
            return False
        self._csv_path = pathlib.Path(self._csv_file)
        if not self._csv_path.exists():
            logger.warning('"csv-file" not found')
            return False
        # announce csv
        if self._verbose:
            logger.info(f'input: {self._csv_file}')
        # config cd
        self._cd_path = None
        self._cd_file = self._config.get('component-definition')
        if self._cd_file is not None:
            self._cd_path = pathlib.Path(self._cd_file)
            if not self._cd_path.exists():
                logger.warning('"component-definition" not found')
                return False
        # workspace
        self._workspace = os.getcwd()
        # validate_controls
        self._validate_controls = self._config.get('validate-controls', 'off')
        return True

    def get_class(self, name: str) -> str:
        """Get class value for specified name from config."""
        key = f'class.{name}'
        return self._config.get(key)

    def simulate(self) -> TaskOutcome:
        """Provide a simulated outcome."""
        return TaskOutcome('simulated-success')

    def execute(self) -> TaskOutcome:
        """Provide an executed outcome."""
        try:
            return self._execute()
        except Exception:
            logger.error(traceback.format_exc())
            return TaskOutcome('failure')

    def _execute(self) -> TaskOutcome:
        """Execute path core."""
        if not self.configure():
            return TaskOutcome('failure')
        # config output
        odir = self._config.get('output-dir')
        opth = pathlib.Path(odir)
        self._overwrite = self._config.getboolean('output-overwrite', True)
        # insure output dir exists
        opth.mkdir(exist_ok=True, parents=True)
        # calculate output file name & check writability
        oname = 'component-definition.json'
        ofile = opth / oname
        if not self._overwrite and pathlib.Path(ofile).exists():
            logger.warning(f'output: {ofile} already exists')
            return TaskOutcome('failure')
        # fetch existing component-definition, if any
        self._cd_mgr = _CdMgr(self._cd_path, self._title, self._timestamp, self._version)
        # fetch csv
        self._csv_mgr = _CsvMgr(self._csv_path)
        # create resolved profile -> catalog helper
        profile_list = self._csv_mgr.get_profile_list()
        self._resolved_profile_catalog_helper = _ResolvedProfileCatalogHelper(profile_list, self._workspace)
        self._unresolved_controls = []
        # calculate deletion, addition & modification rule lists
        rules = self._calculate_rules()
        # calculate deletion, addition & modification set-parameter lists
        set_params = self._calculate_set_params(rules[2])
        # calculate deletion, addition & modification control mapping lists
        control_mappings = self._calculate_control_mappings(rules[2])
        # rule set manager
        self._rule_set_id_mgr = _RuleSetIdMgr(self._cd_mgr.get_max_rule_set_number(), len(rules[1]))
        # rule additions, deletions & modifications (by row)
        self.rules_del(rules[0])
        self.rules_add(rules[1])
        self.rules_mod(rules[2])
        # set-parameters additions, deletions & modifications (by row)
        self.set_params_del(set_params[0])
        self.set_params_add(set_params[1])
        self.set_params_mod(set_params[2])
        # control mappings additions, deletions & modifications (by row)
        self.control_mappings_del(control_mappings[0])
        self.control_mappings_add(control_mappings[1])
        # note: control mappings mod is currently not possible
        # note: add/del user columns not currently supported
        if len(self._unresolved_controls) > 0:
            text = f'Unresolved controls: {self._unresolved_controls}'
            if self._validate_controls == 'warn':
                logger.warning(text)
            elif self._validate_controls == 'on':
                raise RuntimeError(text)
        # prepare new/revised component definition
        component_definition = self._cd_mgr.get_component_definition()
        # write OSCAL ComponentDefinition to file
        if self._verbose:
            logger.info(f'output: {ofile}')
        component_definition.oscal_write(pathlib.Path(ofile))
        return TaskOutcome('success')

    def _calculate_rules(self) -> tuple:
        """Calculate rules add, delete, modify."""
        cd_rules = self._cd_mgr.get_rule_keys()
        csv_rules = self._csv_mgr.get_rule_keys()
        del_rules = []
        add_rules = []
        mod_rules = []
        for key in cd_rules:
            if key in csv_rules:
                continue
            else:
                del_rules.append(key)
                logger.debug(f'rules del: {key}')
        for key in csv_rules:
            if key in cd_rules:
                mod_rules.append(key)
                logger.debug(f'rules mod: {key}')
            else:
                add_rules.append(key)
                logger.debug(f'rules add: {key}')
        return (del_rules, add_rules, mod_rules)

    def _calculate_set_params(self, mod_rules: List) -> tuple:
        """Calculate set parameters add, delete, modify."""
        cd_set_params = self._cd_mgr.get_set_params_keys()
        csv_set_params = self._csv_mgr.get_set_params_keys()
        del_set_params = []
        add_set_params = []
        mod_set_params = []
        for key in cd_set_params:
            rule_key = synthesize_rule_key(key[0], key[1], key[2], None, None)
            if rule_key not in mod_rules:
                continue
            if key in csv_set_params:
                continue
            else:
                del_set_params.append(key)
                logger.debug(f'params del: {key}')
        for key in csv_set_params:
            rule_key = synthesize_rule_key(key[0], key[1], key[2], None, None)
            if rule_key not in mod_rules:
                continue
            if key in cd_set_params:
                mod_set_params.append(key)
                logger.debug(f'params mod: {key}')
            else:
                add_set_params.append(key)
                logger.debug(f'params add: {key}')
        return (del_set_params, add_set_params, mod_set_params)

    def _calculate_control_mappings(self, mod_rules: List) -> tuple:
        """Calculate control mappings add, delete, modify."""
        cd_controls = self._cd_mgr.get_control_keys()
        csv_controls = self._csv_mgr.get_control_keys()
        del_control_mappings = []
        add_control_mappings = []
        mod_control_mappings = []
        for key in cd_controls:
            rule_key = synthesize_rule_key(key[0], key[1], key[2], None, None)
            if rule_key not in mod_rules:
                continue
            if key in csv_controls:
                continue
            else:
                del_control_mappings.append(key)
                logger.debug(f'ctl-maps del: {key}')
        for key in csv_controls:
            rule_key = synthesize_rule_key(key[0], key[1], key[2], None, None)
            if rule_key not in mod_rules:
                continue
            if key in cd_controls:
                mod_control_mappings.append(key)
                logger.debug(f'ctl-maps mod: {key}')
            else:
                add_control_mappings.append(key)
                logger.debug(f'ctl-maps add: {key}')
        return (del_control_mappings, add_control_mappings, mod_control_mappings)

    def _get_namespace(self, rule_key: tuple) -> str:
        """Get namespace."""
        return self._csv_mgr.get_value(rule_key, NAMESPACE).strip()

    def _get_prop_name(self, column_name: str) -> str:
        """Get property name."""
        return column_name.lstrip('$')

    def rules_del(self, del_rules: List[str]) -> None:
        """Delete rules."""
        for tokens in del_rules:
            component_title = tokens[0]
            component_type = tokens[1]
            rule_id = tokens[2]
            description = ''
            # component
            component = self._cd_mgr.get_component(component_title, component_type, description)
            # props
            component.props = self._delete_rule_props(component, rule_id)

    def _delete_rule_props(self, component: DefinedComponent, rule_id: str) -> List[Property]:
        """Delete rule props."""
        props = []
        rule_set = _RuleSetHelper.get_rule_set(component.props, rule_id)
        for prop in component.props:
            if prop.remarks != rule_set:
                props.append(prop)
            elif prop.name in self._csv_mgr.get_parameter_id_column_names():
                self._delete_rule_set_parameter(component, prop.value)
            elif prop.name == RULE_ID:
                self._delete_rule_implemented_requirement(component, prop.value)
        return props

    def _control_implementation_generator(
        self, control_implementations: List[ControlImplementation]
    ) -> Iterator[ControlImplementation]:
        """Control implementation generator."""
        if control_implementations:
            for control_implementation in control_implementations:
                yield control_implementation

    def _set_parameter_generator(self, set_parameters: List[SetParameter]) -> Iterator[SetParameter]:
        """Set parameter generator."""
        if set_parameters:
            for set_parameter in set_parameters:
                yield set_parameter

    def _implemented_requirement_generator(
        self, implemented_requirements: List[ImplementedRequirement]
    ) -> Iterator[ImplementedRequirement]:
        """Implemented-requirement generator."""
        if implemented_requirements:
            for implemented_requirement in implemented_requirements:
                yield implemented_requirement

    def _delete_rule_set_parameter(self, component: DefinedComponent, parameter_id: str) -> None:
        """Delete rule set-parameter."""
        control_implementations = component.control_implementations
        for control_implementation in self._control_implementation_generator(control_implementations):
            if control_implementation.set_parameters:
                set_parameters = control_implementation.set_parameters
                control_implementation.set_parameters = []
                for set_parameter in set_parameters:
                    if set_parameter.param_id != parameter_id:
                        _OscalHelper.add_set_parameter(control_implementation.set_parameters, set_parameter)
                if not len(control_implementation.set_parameters):
                    control_implementation.set_parameters = None

    def _delete_rule_implemented_requirement(self, component: DefinedComponent, rule_id: str) -> None:
        """Delete rule implemented_requirement."""
        control_implementations = component.control_implementations
        component.control_implementations = []
        for control_implementation in self._control_implementation_generator(control_implementations):
            if control_implementation.implemented_requirements:
                implemented_requirements = control_implementation.implemented_requirements
                control_implementation.implemented_requirements = []
                for implemented_requirement in implemented_requirements:
                    self._delete_ir_props(implemented_requirement, rule_id)
                    self._delete_ir_statements(implemented_requirement, rule_id)
                    if len(as_list(implemented_requirement.props)) or len(as_list(implemented_requirement.statements)):
                        control_implementation.implemented_requirements.append(implemented_requirement)
            if len(as_list(control_implementation.implemented_requirements)):
                component.control_implementations.append(control_implementation)

    def _delete_ir_statements(self, implemented_requirement: ImplementedRequirement, rule_id: str) -> None:
        """Delete implemented-requirement statements."""
        if implemented_requirement.statements:
            statements = implemented_requirement.statements
            implemented_requirement.statements = []
            for statement in statements:
                statement.props = self._delete_props(statement.props, rule_id)
                if not len(statement.props):
                    statement.props = None
                if statement.props:
                    implemented_requirement.statements.append(statement)
            if not len(implemented_requirement.statements):
                implemented_requirement.statements = None

    def _delete_ir_props(self, implemented_requirement: ImplementedRequirement, rule_id: str) -> None:
        """Delete implemented-requirement props."""
        if implemented_requirement.props:
            implemented_requirement.props = self._delete_props(implemented_requirement.props, rule_id)
            if not len(implemented_requirement.props):
                implemented_requirement.props = None

    def _delete_props(self, props: List[Property], rule_id: str) -> List[property]:
        """Delete props."""
        rval = []
        if props:
            for prop in props:
                if prop.name == RULE_ID and prop.value == rule_id:
                    continue
                rval.append(prop)
        return rval

    def rules_add(self, add_rules: List[str]) -> None:
        """Add rules."""
        for rule_key in add_rules:
            component_title = self._csv_mgr.get_value(rule_key, COMPONENT_TITLE)
            component_type = self._csv_mgr.get_value(rule_key, COMPONENT_TYPE)
            component_description = self._csv_mgr.get_value(rule_key, COMPONENT_DESCRIPTION)
            # component
            component = self._cd_mgr.get_component(component_title, component_type, component_description)
            # props
            component.props = as_list(component.props)
            component.props = component.props + self._create_rule_props(rule_key)
            # additional props, when not validation component
            if not self._is_validation(rule_key):
                # control implementation
                source = self._csv_mgr.get_value(rule_key, PROFILE_SOURCE)
                description = self._csv_mgr.get_value(rule_key, PROFILE_DESCRIPTION)
                control_implementation = self._get_control_implementation(component, source, description)
                # set-parameters
                set_parameters = self._create_set_parameters(rule_key)
                if set_parameters:
                    control_implementation.set_parameters = as_list(control_implementation.set_parameters)
                    _OscalHelper.add_set_parameters(control_implementation.set_parameters, set_parameters)
                # control-mappings
                control_mappings = self._csv_mgr.get_value(rule_key, CONTROL_ID_LIST).split()
                self._add_rule_prop(control_implementation, control_mappings, rule_key)

    def _is_validation(self, rule_key: tuple) -> bool:
        """Check for validation component."""
        component_type = self._csv_mgr.get_value(rule_key, COMPONENT_TYPE)
        return is_validation(component_type)

    def _add_rule_prop(
        self, control_implementation: ControlImplementation, control_mappings: List[str], rule_key: tuple
    ) -> None:
        """Add rule prop."""
        namespace = self._get_namespace(rule_key)
        for control_mapping in control_mappings:
            control_id = derive_control_id(control_mapping)
            implemented_requirement = self._get_implemented_requirement(control_implementation, control_id)
            # create rule implementation (as property)
            name = RULE_ID
            prop = Property(
                name=name,
                value=self._csv_mgr.get_value(rule_key, name),
                ns=namespace,
                class_=self.get_class(name),
            )
            part_id = derive_part_id(control_mapping)
            if part_id is None:
                implemented_requirement.props = as_list(implemented_requirement.props)
                implemented_requirement.props.append(prop)
            else:
                statement = self._get_statement(implemented_requirement, part_id)
                statement.props.append(prop)

    def _create_rule_props(self, rule_key: tuple) -> List[Property]:
        """Create rule props."""
        rule_set = self._rule_set_id_mgr.get_next_rule_set_id()
        row_number = self._csv_mgr.get_row_number(rule_key)
        rule_set_mgr = _RuleSetMgr(row_number, rule_set)
        namespace = self._get_namespace(rule_key)
        if self._is_validation(rule_key):
            column_names = CsvColumn.get_check_property_column_names()
        else:
            column_names = CsvColumn.get_rule_property_column_names()
        # req'd & optional props
        for column_name in column_names:
            prop_name = self._get_prop_name(column_name)
            prop_value = self._csv_mgr.get_value(rule_key, column_name).strip()
            rule_set_mgr.add_prop(prop_name, prop_value, namespace, self.get_class(prop_name))
        if not self._is_validation(rule_key):
            # parameter columns
            column_names = self._csv_mgr.get_parameter_column_names()
            for column_name in column_names:
                prop_name = self._get_prop_name(column_name)
                prop_value = self._csv_mgr.get_value(rule_key, column_name).strip()
                rule_set_mgr.add_prop(prop_name, prop_value, namespace, self.get_class(prop_name))
        # user props
        column_names = self._csv_mgr.get_user_column_names()
        for column_name in column_names:
            if column_name.startswith('#'):
                continue
            prop_name = self._get_prop_name(column_name)
            prop_value = self._csv_mgr.get_value(rule_key, column_name).strip()
            rule_set_mgr.add_prop(prop_name, prop_value, namespace, self.get_class(prop_name))
        return rule_set_mgr.get_props()

    def _get_control_implementation(
        self, component: DefinedComponent, source: str, description: str
    ) -> ControlImplementation:
        """Find or create control implementation."""
        component.control_implementations = as_list(component.control_implementations)
        for control_implementation in component.control_implementations:
            if control_implementation.source == source and control_implementation.description == description:
                return control_implementation
        control_implementation = ControlImplementation(
            uuid=str(uuid.uuid4()), source=source, description=description, implemented_requirements=[]
        )
        component.control_implementations.append(control_implementation)
        return control_implementation

    def _str_to_list(self, value: str) -> List[str]:
        """Transform string to list."""
        rval = []
        if ',' in value:
            values = value.split(',')
            # remove leading/trailing whitespace
            for v in values:
                rval.append(v.strip())
        else:
            rval.append(value)
        return rval

    def _create_set_parameters(self, rule_key: tuple) -> List[SetParameter]:
        """Create set parameters."""
        set_parameters = []
        for parameter_id_column_name in self._csv_mgr.get_parameter_id_column_names():
            suffix = parameter_id_column_name.replace(PARAMETER_ID, '')
            parameter_value_default_column_name = f'{PARAMETER_VALUE_DEFAULT}{suffix}'
            name = self._csv_mgr.get_value(rule_key, parameter_id_column_name)
            value = self._csv_mgr.get_value(rule_key, parameter_value_default_column_name)
            if name and value:
                try:
                    values = self._str_to_list(value)
                    set_parameter = SetParameter(
                        param_id=name,
                        values=values,
                    )
                    set_parameters.append(set_parameter)
                except Exception:
                    row_number = self._csv_mgr.get_row_number(rule_key)
                    text = (
                        f'row {row_number}: "{name}" is invalid for column {parameter_id_column_name} '
                        f'and/or "{",".join(values)}" is invalid for column {parameter_value_default_column_name}'
                    )
                    raise RuntimeError(text)
            elif name:
                row_number = self._csv_mgr.get_row_number(rule_key)
                text = f'row "{row_number}" missing value for "{parameter_value_default_column_name}"'
                logger.debug(text)
        return set_parameters

    def _get_implemented_requirement(
        self, control_implementation: ControlImplementation, control_id: str
    ) -> ImplementedRequirement:
        """Find or create implemented requirement."""
        if self._validate_controls != 'off':
            if not self._resolved_profile_catalog_helper.validate(control_id):
                if control_id not in self._unresolved_controls:
                    self._unresolved_controls.append(control_id)
        for implemented_requirement in control_implementation.implemented_requirements:
            if implemented_requirement.control_id == control_id:
                return implemented_requirement
        implemented_requirement = ImplementedRequirement(
            uuid=str(uuid.uuid4()),
            control_id=control_id,
            description='',
        )
        control_implementation.implemented_requirements.append(implemented_requirement)
        return implemented_requirement

    def _get_statement(self, implemented_requirement: ImplementedRequirement, part_id: str) -> Statement:
        """Find or create statement."""
        implemented_requirement.statements = as_list(implemented_requirement.statements)
        for statement in implemented_requirement.statements:
            if statement.statement_id == part_id:
                return statement
        statement = Statement(
            uuid=str(uuid.uuid4()),
            statement_id=part_id,
            description='',
            props=[],
        )
        implemented_requirement.statements.append(statement)
        return statement

    def rules_mod(self, mod_rules: List[str]) -> None:
        """Modify rules."""
        for rule_key in mod_rules:
            component_title = self._csv_mgr.get_value(rule_key, COMPONENT_TITLE)
            component_type = self._csv_mgr.get_value(rule_key, COMPONENT_TYPE)
            component_description = self._csv_mgr.get_value(rule_key, COMPONENT_DESCRIPTION)
            # component
            component = self._cd_mgr.get_component(component_title, component_type, component_description)
            # props
            component.props = self._modify_rule_props(component, rule_key)

    def _modify_rule_props(self, component: DefinedComponent, rule_key: tuple) -> List[Property]:
        """Modify rule props."""
        rule_id = self._csv_mgr.get_value(rule_key, RULE_ID)
        rule_set = _RuleSetHelper.get_rule_set(component.props, rule_id)
        rule_ns = self._csv_mgr.get_value(rule_key, NAMESPACE)
        column_names = CsvColumn.get_filtered_required_column_names() + CsvColumn.get_filtered_optional_column_names()
        # req'd & optional props
        for column_name in column_names:
            column_value = self._csv_mgr.get_value(rule_key, column_name).strip()
            class_ = self.get_class(column_name)
            self._cd_mgr.update_rule_definition(component, rule_set, column_name, column_value, rule_ns, class_)
        # parameter columns
        column_names = self._csv_mgr.get_parameter_column_names()
        for column_name in column_names:
            column_value = self._csv_mgr.get_value(rule_key, column_name).strip()
            class_ = self.get_class(column_name)
            self._cd_mgr.update_rule_definition(component, rule_set, column_name, column_value, rule_ns, class_)
        # user props
        column_names = self._csv_mgr.get_user_column_names()
        for column_name in column_names:
            column_value = self._csv_mgr.get_value(rule_key, column_name).strip()
            self._cd_mgr.update_rule_definition(component, rule_set, column_name, column_value, rule_ns, class_)
        return component.props

    def set_params_del(self, del_set_params: List[str]) -> None:
        """Set parameters delete."""
        for tokens in del_set_params:
            component_title = tokens[0]
            component_type = tokens[1]
            source = tokens[3]
            description = tokens[4]
            param_id = tokens[5]
            # ctl impl
            control_implementation = self._cd_mgr.find_control_implementation(
                component_title, component_type, source, description
            )
            if control_implementation:
                set_parameters = control_implementation.set_parameters
                control_implementation.set_parameters = []
                for set_parameter in self._set_parameter_generator(set_parameters):
                    if set_parameter.param_id == param_id:
                        continue
                    _OscalHelper.add_set_parameter(control_implementation.set_parameters, set_parameter)
                if control_implementation.set_parameters == []:
                    control_implementation.set_parameters = None

    def set_params_add(self, add_set_params: List[str]) -> None:
        """Set parameters add."""
        for tokens in add_set_params:
            component_title = tokens[0]
            component_type = tokens[1]
            rule_id = tokens[2]
            source = tokens[3]
            description = tokens[4]
            param_id = tokens[5]
            # ctl impl
            control_implementation = self._cd_mgr.find_control_implementation(
                component_title, component_type, source, description
            )
            control_implementation.set_parameters = as_list(control_implementation.set_parameters)
            # add
            rule_key = synthesize_rule_key(component_title, component_type, rule_id, None, None)
            values = [self._csv_mgr.get_default_value_by_id(rule_key, param_id)]
            set_parameter = SetParameter(
                param_id=param_id,
                values=values,
            )
            _OscalHelper.add_set_parameter(control_implementation.set_parameters, set_parameter)

    def set_params_mod(self, mod_set_params: List[str]) -> None:
        """Set parameters modify."""
        for tokens in mod_set_params:
            component_title = tokens[0]
            component_type = tokens[1]
            rule_id = tokens[2]
            source = tokens[3]
            description = tokens[4]
            param_id = tokens[5]
            # ctl impl
            control_implementation = self._cd_mgr.find_control_implementation(
                component_title, component_type, source, description
            )
            if control_implementation:
                set_parameters = control_implementation.set_parameters
                for set_parameter in self._set_parameter_generator(set_parameters):
                    if set_parameter.param_id != param_id:
                        continue
                    rule_key = synthesize_rule_key(component_title, component_type, rule_id, None, None)
                    values = [self._csv_mgr.get_default_value_by_id(rule_key, param_id)]
                    replacement = SetParameter(
                        param_id=param_id,
                        values=values,
                    )
                    if set_parameter.values == replacement.values:
                        continue
                    logger.debug(f'params-mod: {rule_id} {param_id} {set_parameter.values} -> {replacement.values}')
                    set_parameter.values = replacement.values

    def _control_mappings_generator(self, control_mappings: List[str]) -> Iterator[List[str]]:
        """Control mappings generator."""
        for tokens in control_mappings:
            component_title = tokens[0]
            component_type = tokens[1]
            source = tokens[3]
            description = tokens[4]
            control_implementation = self._cd_mgr.find_control_implementation(
                component_title, component_type, source, description
            )
            if control_implementation:
                yield tokens

    def control_mappings_del(self, del_control_mappings: List[str]) -> None:
        """Control mappings delete."""
        for tokens in self._control_mappings_generator(del_control_mappings):
            component_title = tokens[0]
            component_type = tokens[1]
            rule_id = tokens[2]
            source = tokens[3]
            description = tokens[4]
            smt_id = tokens[5]
            # ctl-id
            control_id = derive_control_id(smt_id)
            control_implementation = self._cd_mgr.find_control_implementation(
                component_title, component_type, source, description
            )
            implemented_requirements = control_implementation.implemented_requirements
            control_implementation.implemented_requirements = []
            for implemented_requirement in self._implemented_requirement_generator(implemented_requirements):
                if implemented_requirement.control_id == control_id:
                    implemented_requirement.statements = _OscalHelper.remove_rule_statement(
                        implemented_requirement.statements, rule_id, smt_id
                    )
                    implemented_requirement.props = _OscalHelper.remove_rule(implemented_requirement.props, rule_id)
                    if len(as_list(implemented_requirement.props)) or len(as_list(implemented_requirement.statements)):
                        control_implementation.implemented_requirements.append(implemented_requirement)
                else:
                    control_implementation.implemented_requirements.append(implemented_requirement)

    def control_mappings_add(self, add_control_mappings: List[str]) -> None:
        """Control mappings add."""
        for tokens in self._control_mappings_generator(add_control_mappings):
            component_title = tokens[0]
            component_type = tokens[1]
            rule_id = tokens[2]
            source = tokens[3]
            description = tokens[4]
            smt_id = tokens[5]
            # ctl-id
            control_id = derive_control_id(smt_id)
            control_implementation = self._cd_mgr.find_control_implementation(
                component_title, component_type, source, description
            )
            implemented_requirement = self._get_implemented_requirement(control_implementation, control_id)
            # namespace
            rule_key = synthesize_rule_key(tokens[0], tokens[1], tokens[2], None, None)
            ns = self._get_namespace(rule_key)
            # create rule implementation (as property)
            name = RULE_ID
            prop = Property(
                name=name,
                value=rule_id,
                ns=ns,
                class_=self.get_class(name),
            )
            if smt_id == control_id:
                implemented_requirement.props = as_list(implemented_requirement.props)
                implemented_requirement.props.append(prop)
            else:
                statement = self._get_statement(implemented_requirement, smt_id)
                statement.props.append(prop)

Attributes

name = 'csv-to-oscal-cd' class-attribute instance-attribute

Functions

__init__(config_object)

Initialize trestle task csv-to-oscal-cd.

Parameters:

Name	Type	Description	Default
config_object	Optional[SectionProxy]	Config section associated with the task.	required

Source code in trestle/tasks/csv_to_oscal_cd.py

def __init__(self, config_object: Optional[configparser.SectionProxy]) -> None:
    """
    Initialize trestle task csv-to-oscal-cd.

    Args:
        config_object: Config section associated with the task.
    """
    super().__init__(config_object)

configure()

Configure.

Source code in trestle/tasks/csv_to_oscal_cd.py

def configure(self) -> bool:
    """Configure."""
    self._timestamp = datetime.datetime.now(datetime.timezone.utc).replace(microsecond=0).isoformat()
    # config verbosity
    self._quiet = self._config.get('quiet', False)
    self._verbose = not self._quiet
    # title
    self._title = self._config.get('title')
    if self._title is None:
        logger.warning('config missing "title"')
        return False
    # version
    self._version = self._config.get('version')
    if self._version is None:
        logger.warning('config missing "version"')
        return False
    # config csv
    self._csv_file = self._config.get('csv-file')
    if self._csv_file is None:
        logger.warning('config missing "csv-file"')
        return False
    self._csv_path = pathlib.Path(self._csv_file)
    if not self._csv_path.exists():
        logger.warning('"csv-file" not found')
        return False
    # announce csv
    if self._verbose:
        logger.info(f'input: {self._csv_file}')
    # config cd
    self._cd_path = None
    self._cd_file = self._config.get('component-definition')
    if self._cd_file is not None:
        self._cd_path = pathlib.Path(self._cd_file)
        if not self._cd_path.exists():
            logger.warning('"component-definition" not found')
            return False
    # workspace
    self._workspace = os.getcwd()
    # validate_controls
    self._validate_controls = self._config.get('validate-controls', 'off')
    return True

control_mappings_add(add_control_mappings)

Control mappings add.

Source code in trestle/tasks/csv_to_oscal_cd.py

def control_mappings_add(self, add_control_mappings: List[str]) -> None:
    """Control mappings add."""
    for tokens in self._control_mappings_generator(add_control_mappings):
        component_title = tokens[0]
        component_type = tokens[1]
        rule_id = tokens[2]
        source = tokens[3]
        description = tokens[4]
        smt_id = tokens[5]
        # ctl-id
        control_id = derive_control_id(smt_id)
        control_implementation = self._cd_mgr.find_control_implementation(
            component_title, component_type, source, description
        )
        implemented_requirement = self._get_implemented_requirement(control_implementation, control_id)
        # namespace
        rule_key = synthesize_rule_key(tokens[0], tokens[1], tokens[2], None, None)
        ns = self._get_namespace(rule_key)
        # create rule implementation (as property)
        name = RULE_ID
        prop = Property(
            name=name,
            value=rule_id,
            ns=ns,
            class_=self.get_class(name),
        )
        if smt_id == control_id:
            implemented_requirement.props = as_list(implemented_requirement.props)
            implemented_requirement.props.append(prop)
        else:
            statement = self._get_statement(implemented_requirement, smt_id)
            statement.props.append(prop)

control_mappings_del(del_control_mappings)

Control mappings delete.

Source code in trestle/tasks/csv_to_oscal_cd.py

def control_mappings_del(self, del_control_mappings: List[str]) -> None:
    """Control mappings delete."""
    for tokens in self._control_mappings_generator(del_control_mappings):
        component_title = tokens[0]
        component_type = tokens[1]
        rule_id = tokens[2]
        source = tokens[3]
        description = tokens[4]
        smt_id = tokens[5]
        # ctl-id
        control_id = derive_control_id(smt_id)
        control_implementation = self._cd_mgr.find_control_implementation(
            component_title, component_type, source, description
        )
        implemented_requirements = control_implementation.implemented_requirements
        control_implementation.implemented_requirements = []
        for implemented_requirement in self._implemented_requirement_generator(implemented_requirements):
            if implemented_requirement.control_id == control_id:
                implemented_requirement.statements = _OscalHelper.remove_rule_statement(
                    implemented_requirement.statements, rule_id, smt_id
                )
                implemented_requirement.props = _OscalHelper.remove_rule(implemented_requirement.props, rule_id)
                if len(as_list(implemented_requirement.props)) or len(as_list(implemented_requirement.statements)):
                    control_implementation.implemented_requirements.append(implemented_requirement)
            else:
                control_implementation.implemented_requirements.append(implemented_requirement)

execute()

Provide an executed outcome.

Source code in trestle/tasks/csv_to_oscal_cd.py

def execute(self) -> TaskOutcome:
    """Provide an executed outcome."""
    try:
        return self._execute()
    except Exception:
        logger.error(traceback.format_exc())
        return TaskOutcome('failure')

get_class(name)

Get class value for specified name from config.

Source code in trestle/tasks/csv_to_oscal_cd.py

def get_class(self, name: str) -> str:
    """Get class value for specified name from config."""
    key = f'class.{name}'
    return self._config.get(key)

print_info()

Print the help string.

Source code in trestle/tasks/csv_to_oscal_cd.py

def print_info(self) -> None:
    """Print the help string."""
    name = self.name
    oscal_name = 'component_definition'
    # help note identifiers
    note01 = '1'
    note02 = '2'
    note03 = '3'
    note04 = '4'
    note05 = '5'
    # help generation
    logger.info(f'Help information for {name} task.')
    logger.info('')
    logger.info(f'Purpose: From csv produce OSCAL {oscal_name} file.')
    logger.info('')
    logger.info('')
    logger.info(f'Configuration flags sit under [task.{name}]:')
    text1 = '  title                = '
    text2 = '(required) the component definition title.'
    logger.info(text1 + text2)
    text1 = '  version              = '
    text2 = '(required) the component definition version.'
    logger.info(text1 + text2)
    text1 = '  csv-file             = '
    text2 = '(required) the path of the csv file.'
    text3 = ' [1st row are column headings; 2nd row are column descriptions; 3rd row and beyond is data]'
    logger.info(text1 + text2 + text3)
    text1 = '  required columns:      '
    for text2 in CsvColumn.get_required_column_names():
        if text2 in [f'{RULE_DESCRIPTION}', f'{PROFILE_SOURCE}', f'{PROFILE_DESCRIPTION}', f'{CONTROL_ID_LIST}']:
            text2 += f' (see note {note01})'
        logger.info(text1 + '$$' + text2)
        text1 = '                         '
    text1 = '  optional columns:      '
    for text2 in CsvColumn.get_optional_column_names():
        if text2 in [f'{ORIGINAL_RISK_RATING}', f'{ADJUSTED_RISK_RATING}', f'{RISK_ADJUSTMENT}']:
            text2 += f' (see note {note01})'
        elif text2 in [f'{TARGET_COMPONENT}']:
            text2 += f' (see note {note03})'
        else:
            text2 += f' (see note {note02})'
        logger.info(text1 + '$' + text2)
        text1 = '                         '
    for text2 in CsvColumn.get_parameter_column_names():
        text2 += f' (see notes {note01}, {note05})'
        logger.info(text1 + '$' + text2)
        text1 = '                         '
    text1 = '  comment columns:       '
    text2 = f'Informational (see note {note04})'
    logger.info(text1 + '#' + text2)
    text1 = '  output-dir           = '
    text2 = '(required) the path of the output directory for synthesized OSCAL .json files.'
    logger.info(text1 + text2)
    text1 = '  component-definition = '
    text2 = '(optional) the path of the existing component-definition OSCAL .json file.'
    logger.info(text1 + text2)
    text1 = '  class.column-name    = '
    text2 = f'(optional) the class to associate with the specified column name, e.g. class.{RULE_ID} = scc_class'
    logger.info(text1 + text2)
    text1 = '  output-overwrite     = '
    text2 = '(optional) true [default] or false; replace existing output when true.'
    logger.info(text1 + text2)
    text1 = '  validate-controls    = '
    text2 = '(optional) on, warn, or off [default]; validate controls exist in resolved profile.'
    logger.info(text1 + text2)
    # Notes
    text1 = ''
    text2 = ''
    logger.info(text1 + text2)
    text1 = 'Notes: '
    text2 = f'[{note01}] column is ignored for validation component type'
    logger.info(text1 + text2)
    text1 = '       '
    text2 = f'[{note02}] column is required for validation component type'
    logger.info(text1 + text2)
    text1 = '       '
    text2 = f'[{note03}] column is optional for validation component type'
    text3 = f', but may be needed to prevent {RULE_ID} collisions'
    logger.info(text1 + text2 + text3)
    text1 = '       '
    text2 = f'[{note04}] column name starting with # causes column to be ignored'
    logger.info(text1 + text2)
    text1 = '       '
    text2 = f'[{note05}] additional parameters are specified by adding a common suffix per set'
    text3 = f', for example: {PARAMETER_ID}_1, {PARAMETER_DESCRIPTION}_1, ...{PARAMETER_ID}_2...'
    logger.info(text1 + text2 + text3)

rules_add(add_rules)

Add rules.

Source code in trestle/tasks/csv_to_oscal_cd.py

def rules_add(self, add_rules: List[str]) -> None:
    """Add rules."""
    for rule_key in add_rules:
        component_title = self._csv_mgr.get_value(rule_key, COMPONENT_TITLE)
        component_type = self._csv_mgr.get_value(rule_key, COMPONENT_TYPE)
        component_description = self._csv_mgr.get_value(rule_key, COMPONENT_DESCRIPTION)
        # component
        component = self._cd_mgr.get_component(component_title, component_type, component_description)
        # props
        component.props = as_list(component.props)
        component.props = component.props + self._create_rule_props(rule_key)
        # additional props, when not validation component
        if not self._is_validation(rule_key):
            # control implementation
            source = self._csv_mgr.get_value(rule_key, PROFILE_SOURCE)
            description = self._csv_mgr.get_value(rule_key, PROFILE_DESCRIPTION)
            control_implementation = self._get_control_implementation(component, source, description)
            # set-parameters
            set_parameters = self._create_set_parameters(rule_key)
            if set_parameters:
                control_implementation.set_parameters = as_list(control_implementation.set_parameters)
                _OscalHelper.add_set_parameters(control_implementation.set_parameters, set_parameters)
            # control-mappings
            control_mappings = self._csv_mgr.get_value(rule_key, CONTROL_ID_LIST).split()
            self._add_rule_prop(control_implementation, control_mappings, rule_key)

rules_del(del_rules)

Delete rules.

Source code in trestle/tasks/csv_to_oscal_cd.py

def rules_del(self, del_rules: List[str]) -> None:
    """Delete rules."""
    for tokens in del_rules:
        component_title = tokens[0]
        component_type = tokens[1]
        rule_id = tokens[2]
        description = ''
        # component
        component = self._cd_mgr.get_component(component_title, component_type, description)
        # props
        component.props = self._delete_rule_props(component, rule_id)

rules_mod(mod_rules)

Modify rules.

Source code in trestle/tasks/csv_to_oscal_cd.py

def rules_mod(self, mod_rules: List[str]) -> None:
    """Modify rules."""
    for rule_key in mod_rules:
        component_title = self._csv_mgr.get_value(rule_key, COMPONENT_TITLE)
        component_type = self._csv_mgr.get_value(rule_key, COMPONENT_TYPE)
        component_description = self._csv_mgr.get_value(rule_key, COMPONENT_DESCRIPTION)
        # component
        component = self._cd_mgr.get_component(component_title, component_type, component_description)
        # props
        component.props = self._modify_rule_props(component, rule_key)

set_params_add(add_set_params)

Set parameters add.

Source code in trestle/tasks/csv_to_oscal_cd.py

def set_params_add(self, add_set_params: List[str]) -> None:
    """Set parameters add."""
    for tokens in add_set_params:
        component_title = tokens[0]
        component_type = tokens[1]
        rule_id = tokens[2]
        source = tokens[3]
        description = tokens[4]
        param_id = tokens[5]
        # ctl impl
        control_implementation = self._cd_mgr.find_control_implementation(
            component_title, component_type, source, description
        )
        control_implementation.set_parameters = as_list(control_implementation.set_parameters)
        # add
        rule_key = synthesize_rule_key(component_title, component_type, rule_id, None, None)
        values = [self._csv_mgr.get_default_value_by_id(rule_key, param_id)]
        set_parameter = SetParameter(
            param_id=param_id,
            values=values,
        )
        _OscalHelper.add_set_parameter(control_implementation.set_parameters, set_parameter)

set_params_del(del_set_params)

Set parameters delete.

Source code in trestle/tasks/csv_to_oscal_cd.py

def set_params_del(self, del_set_params: List[str]) -> None:
    """Set parameters delete."""
    for tokens in del_set_params:
        component_title = tokens[0]
        component_type = tokens[1]
        source = tokens[3]
        description = tokens[4]
        param_id = tokens[5]
        # ctl impl
        control_implementation = self._cd_mgr.find_control_implementation(
            component_title, component_type, source, description
        )
        if control_implementation:
            set_parameters = control_implementation.set_parameters
            control_implementation.set_parameters = []
            for set_parameter in self._set_parameter_generator(set_parameters):
                if set_parameter.param_id == param_id:
                    continue
                _OscalHelper.add_set_parameter(control_implementation.set_parameters, set_parameter)
            if control_implementation.set_parameters == []:
                control_implementation.set_parameters = None

set_params_mod(mod_set_params)

Set parameters modify.

Source code in trestle/tasks/csv_to_oscal_cd.py

def set_params_mod(self, mod_set_params: List[str]) -> None:
    """Set parameters modify."""
    for tokens in mod_set_params:
        component_title = tokens[0]
        component_type = tokens[1]
        rule_id = tokens[2]
        source = tokens[3]
        description = tokens[4]
        param_id = tokens[5]
        # ctl impl
        control_implementation = self._cd_mgr.find_control_implementation(
            component_title, component_type, source, description
        )
        if control_implementation:
            set_parameters = control_implementation.set_parameters
            for set_parameter in self._set_parameter_generator(set_parameters):
                if set_parameter.param_id != param_id:
                    continue
                rule_key = synthesize_rule_key(component_title, component_type, rule_id, None, None)
                values = [self._csv_mgr.get_default_value_by_id(rule_key, param_id)]
                replacement = SetParameter(
                    param_id=param_id,
                    values=values,
                )
                if set_parameter.values == replacement.values:
                    continue
                logger.debug(f'params-mod: {rule_id} {param_id} {set_parameter.values} -> {replacement.values}')
                set_parameter.values = replacement.values

simulate()

Provide a simulated outcome.

Source code in trestle/tasks/csv_to_oscal_cd.py

def simulate(self) -> TaskOutcome:
    """Provide a simulated outcome."""
    return TaskOutcome('simulated-success')

Functions

derive_control_id(control_mapping)

Derive control id.

Source code in trestle/tasks/csv_to_oscal_cd.py

def derive_control_id(control_mapping: str) -> str:
    """Derive control id."""
    rval = control_mapping.split('_smt')[0]
    return rval

derive_part_id(control_mapping)

Derive part id.

Source code in trestle/tasks/csv_to_oscal_cd.py

def derive_part_id(control_mapping: str) -> str:
    """Derive part id."""
    if '_smt.' in control_mapping:
        rval = control_mapping
    else:
        rval = None
    return rval

etype(target)

Get etype.

Source code in trestle/tasks/csv_to_oscal_cd.py

def etype(target: str) -> str:
    """Get etype."""
    if target:
        return 'invalid'
    else:
        return 'missing'

is_validation(component_type)

Check for validation component.

Source code in trestle/tasks/csv_to_oscal_cd.py

def is_validation(component_type: str) -> bool:
    """Check for validation component."""
    return component_type.lower().strip() == validation

row_property_builder(row, name, value, ns, class_, remarks)

Row property builder.

Source code in trestle/tasks/csv_to_oscal_cd.py

def row_property_builder(row: int, name: str, value, ns: str, class_: str, remarks: str) -> Property:
    """Row property builder."""
    # name
    try:
        Property(
            name=name,
            value='value',
        )
    except Exception:
        text = f'property for row: {row} name: {name} is {etype(name)}'
        raise RuntimeError(text)
    # value
    try:
        Property(
            name=name,
            value=value,
        )
    except Exception:
        text = f'property for row: {row} value: {value} is {etype(value)}'
        raise RuntimeError(text)
    # ns
    try:
        Property(
            name=name,
            value=value,
            ns=ns,
        )
    except Exception:
        text = f'property for row: {row} ns: {ns} is {etype(ns)}'
        raise RuntimeError(text)
    # class
    try:
        Property(
            name=name,
            value=value,
            class_=class_,
        )
    except Exception:
        text = f'property for row: {row} class: {class_} is {etype(class_)}'
        raise RuntimeError(text)
    # prop
    prop = Property(
        name=name,
        value=value,
        ns=ns,
        class_=class_,
        remarks=remarks,
    )
    return prop

synthesize_rule_key(component_title, component_type, rule_id, check_id, target_component)

Synthesize rule_key.

Source code in trestle/tasks/csv_to_oscal_cd.py

def synthesize_rule_key(
    component_title: str,
    component_type: str,
    rule_id: str,
    check_id: Union[str, None],
    target_component: Union[str, None]
) -> tuple:
    """Synthesize rule_key."""
    if is_validation(component_type):
        rval = (component_title, component_type, rule_id, check_id, target_component)
    else:
        rval = (component_title, component_type, rule_id, None, None)
    return rval

handler: python