feat: add parameter aggregation support for SSP (#1668)

Signed-off-by: Alejandro Jose Leiva Palomo <alejandro.leiva.palomo@ibm.com>

tests/data/json/profile_aggregation.json

@@ -0,0 +1,106 @@
+{
+  "profile": {
+    "uuid": "83be30e4-6c88-40e0-9927-286ff0e43d59",
+    "metadata": {
+      "title": "SI-7 profile",
+      "last-modified": "2024-06-24T19:35:43.973109+00:00",
+      "version": "0.0.1",
+      "oscal-version": "1.1.2"
+    },
+    "imports": [
+      {
+        "href": "https://raw.githubusercontent.com/usnistgov/oscal-content/main/nist.gov/SP800-53/rev5/json/NIST_SP-800-53_rev5_catalog.json",
+        "include-controls": [
+          {
+            "with-ids": [
+              "si-7"
+            ]
+          }
+        ]
+      }
+    ],
+    "modify": {
+      "set-parameters": [
+        {
+          "param-id": "si-07_odp.01",
+          "props": [
+            {
+              "name": "param-value-origin",
+              "value": "OCISO"
+            }
+          ],
+          "values": [
+            "software"
+          ]
+        },
+        {
+          "param-id": "si-07_odp.02",
+          "props": [
+            {
+              "name": "param-value-origin",
+              "value": "OCISO"
+            }
+          ],
+          "values": [
+            "firmware"
+          ]
+        },
+        {
+          "param-id": "si-07_odp.03",
+          "props": [
+            {
+              "name": "param-value-origin",
+              "value": "OCISO"
+            }
+          ],
+          "values": [
+            "information"
+          ]
+        },
+        {
+          "param-id": "si-07_odp.04",
+          "props": [
+            {
+              "name": "param-value-origin",
+              "value": "OCISO"
+            }
+          ],
+          "values": [
+            "notify the System Owner, ISSO, ISSM, and the Incident Response team"
+          ]
+        },
+        {
+          "param-id": "si-07_odp.05",
+          "props": [
+            {
+              "name": "param-value-origin",
+              "value": "OCISO"
+            }
+          ],
+          "values": [
+            "notify the System Owner, ISSO, ISSM, and the Incident Response team"
+          ]
+        },
+        {
+          "param-id": "si-07_odp.06",
+          "props": [
+            {
+              "name": "param-value-origin",
+              "value": "OCISO"
+            }
+          ],
+          "values": [
+            "notify the System Owner, ISSO, ISSM, and the Incident Response team"
+          ]
+        },
+        {
+          "param-id": "si-7_prm_1"
+        },
+        {
+          "param-id": "si-7_prm_2"
+        }
+      ],
+      "alters": []
+    }
+  }
+}

tests/trestle/core/commands/author/ssp_test.py

@@ -1303,3 +1303,75 @@ def test_ssp_generate_no_cds_include_all_parts(tmp_trestle_dir: pathlib.Path) ->
 
     node = tree.get_node_for_key('## Implementation for part a.')
     assert node.content.raw_text == part_a_text_no_comp
+
+
+def test_ssp_generate_aggregates_no_cds(tmp_trestle_dir: pathlib.Path) -> None:
+    """Test the ssp generator with no comp defs does aggregate values from aggregated parameters."""
+    args, _ = setup_for_ssp(tmp_trestle_dir, 'profile_aggregation', ssp_name)
+
+    args.compdefs = None
+    ssp_cmd = SSPGenerate()
+    assert ssp_cmd._run(args) == 0
+    md_dir = tmp_trestle_dir / ssp_name
+    si_7 = md_dir / 'si-7.md'
+    assert si_7.exists()
+
+    md_api = MarkdownAPI()
+    header, tree = md_api.processor.process_markdown(si_7)
+    si_7_odp_01 = header['x-trestle-set-params']['si-07_odp.01']
+    si_7_odp_01['ssp-values'] = ['changed value in the ssp markdown']
+
+    md_api.write_markdown_with_header(si_7, header, tree.content.raw_text)
+
+    # now assemble the edited controls into json ssp
+    ssp_assemble = SSPAssemble()
+    assemble_args = argparse.Namespace(
+        trestle_root=tmp_trestle_dir,
+        markdown=ssp_name,
+        output=ssp_name,
+        verbose=0,
+        regenerate=False,
+        version='',
+        name=None,
+        compdefs=None
+    )
+    assert ssp_assemble._run(assemble_args) == 0
+
+    # Verify the correct information is in the assembled ssp
+    ssp, _ = ModelUtils.load_model_for_class(tmp_trestle_dir, ssp_name, ossp.SystemSecurityPlan, FileContentType.JSON)
+    imp_reqs = ssp.control_implementation.implemented_requirements
+    si_7_imp_req = next((i_req for i_req in imp_reqs if i_req.control_id == 'si-7'), None)
+    si_07_odp_01 = next((param for param in si_7_imp_req.set_parameters if param.param_id == 'si-07_odp.01'), None)
+    changed_value_in_ssp = next(
+        (val for val in si_07_odp_01.values if val == 'changed value in the ssp markdown'), None
+    )
+    assert changed_value_in_ssp is not None
+
+    # regenerate the SSP again
+    ssp_cmd = SSPGenerate()
+    assert ssp_cmd._run(args) == 0
+    md_dir = tmp_trestle_dir / ssp_name
+    si_7 = md_dir / 'si-7.md'
+    assert si_7.exists()
+
+    md_api = MarkdownAPI()
+    header, tree = md_api.processor.process_markdown(si_7)
+    si_7_odp_01 = header['x-trestle-set-params']['si-07_odp.01']
+    assert 'changed value in the ssp markdown' in si_7_odp_01['ssp-values']
+
+
+def test_ssp_generate_aggregates_no_param_value_orig(tmp_trestle_dir: pathlib.Path) -> None:
+    """Test the ssp generator aggregate parameters have no parame-value-origin."""
+    args, _ = setup_for_ssp(tmp_trestle_dir, 'profile_aggregation', ssp_name)
+
+    args.compdefs = None
+    ssp_cmd = SSPGenerate()
+    assert ssp_cmd._run(args) == 0
+    md_dir = tmp_trestle_dir / ssp_name
+    si_7 = md_dir / 'si-7.md'
+    assert si_7.exists()
+
+    md_api = MarkdownAPI()
+    header, _ = md_api.processor.process_markdown(si_7)
+    si_7_prm_1 = header['x-trestle-set-params']['si-7_prm_1']
+    assert const.PARAM_VALUE_ORIGIN not in si_7_prm_1.keys()

trestle/core/catalog/catalog_writer.py

@@ -165,9 +165,11 @@ def _construct_set_parameters_dict(
                 # adds it to prof-param-value-origin
                 if prof_param_value_origin != '' and prof_param_value_origin is not None:
                     if context.purpose == ContextPurpose.PROFILE:
-                        new_dict[const.PROFILE_PARAM_VALUE_ORIGIN] = prof_param_value_origin
+                        if const.AGGREGATES not in [prop.name for prop in as_list(param.props)]:
+                            new_dict[const.PROFILE_PARAM_VALUE_ORIGIN] = prof_param_value_origin
                 else:
-                    new_dict[const.PROFILE_PARAM_VALUE_ORIGIN] = const.REPLACE_ME_PLACEHOLDER
+                    if const.AGGREGATES not in [prop.name for prop in as_list(param.props)]:
+                        new_dict[const.PROFILE_PARAM_VALUE_ORIGIN] = const.REPLACE_ME_PLACEHOLDER
                 # then insert the original, incoming values as values
                 if param_id in control_param_dict:
                     orig_param = control_param_dict[param_id]
@@ -180,6 +182,8 @@ def _construct_set_parameters_dict(
                         new_dict.pop(const.VALUES)
                     if new_dict[const.GUIDELINES] is None:
                         new_dict.pop(const.GUIDELINES)
+                    if const.AGGREGATES in [prop.name for prop in as_list(orig_param.props)]:
+                        new_dict.pop(const.PROFILE_PARAM_VALUE_ORIGIN)
             else:
                 # if the profile doesnt change this param at all, show it in the header with values
                 tmp_dict = ModelUtils.parameter_to_dict(param_dict, True)

trestle/core/control_interface.py

@@ -670,6 +670,24 @@ def _param_selection_as_str(param: common.Parameter, verbose: bool = False, brac
             return choices_str
         return ''
 
+    @staticmethod
+    def _param_as_aggregated_value(
+        param: common.Parameter,
+        param_dict: Dict[str, common.Parameter],
+        verbose: bool = False,
+        brackets: bool = False
+    ) -> str:
+        """Convert parameter aggregation to str."""
+        # review is an aggregated parameter
+        if const.AGGREGATES in [prop.name for prop in as_list(param.props)]:
+            aggregated_values = ''
+            for prop in as_list(param.props):
+                if prop.value not in param_dict:
+                    continue
+                aggregated_values += ', '.join(as_list(param_dict[prop.value].values)) + ', '
+            return aggregated_values[:-2]
+        return ''
+
     @staticmethod
     def _param_label_choices_as_str(param: common.Parameter, verbose: bool = False, brackets: bool = False) -> str:
         """Convert param label or choices to string, using choices if present."""
@@ -681,6 +699,7 @@ def _param_label_choices_as_str(param: common.Parameter, verbose: bool = False,
     @staticmethod
     def _param_values_assignment_str(
         param: common.Parameter,
+        param_dict: Dict[str, common.Parameter],
         value_assigned_prefix: Optional[str] = None,
         value_not_assigned_prefix: Optional[str] = None
     ) -> str:
@@ -692,6 +711,9 @@ def _param_values_assignment_str(
         # otherwise use param selection if present
         if not param_str:
             param_str = ControlInterface._param_selection_as_str(param, True, False)
+        # otherwise use param aggregated values if present
+        if not param_str:
+            param_str = ControlInterface._param_as_aggregated_value(param, param_dict, True, False)
         # finally use label and param_id as fallbacks
         if not param_str:
             param_str = param.label if param.label else param.id
@@ -722,7 +744,8 @@ def param_to_str(
         brackets: bool = False,
         params_format: Optional[str] = None,
         value_assigned_prefix: Optional[str] = None,
-        value_not_assigned_prefix: Optional[str] = None
+        value_not_assigned_prefix: Optional[str] = None,
+        param_dict: Dict[str, common.Parameter] = None
     ) -> Optional[str]:
         """
         Convert parameter to string based on best available representation.
@@ -755,7 +778,7 @@ def param_to_str(
                 param_str = ''
         elif param_rep == ParameterRep.ASSIGNMENT_FORM:
             param_str = ControlInterface._param_values_assignment_str(
-                param, value_assigned_prefix, value_not_assigned_prefix
+                param, param_dict, value_assigned_prefix, value_not_assigned_prefix
             )
             if not param_str:
                 param_str = ''
@@ -854,11 +877,20 @@ def _replace_params(
             elif param_dict[param_ids[i]] is not None:
                 param = param_dict[param_ids[i]]
                 param_str = ControlInterface.param_to_str(
-                    param, param_rep, False, False, params_format, value_assigned_prefix, value_not_assigned_prefix
+                    param,
+                    param_rep,
+                    False,
+                    False,
+                    params_format,
+                    value_assigned_prefix,
+                    value_not_assigned_prefix,
+                    param_dict
                 )
                 text = text.replace(staches[i], param_str, 1).strip()
                 if show_value_warnings and param_rep != ParameterRep.LABEL_OR_CHOICES and not param.values:
-                    logger.warning(f'Parameter {param_id} has no values and was referenced by prose.')
+                    # verifies the current parameter is not an aggregated parameter to throw a warning
+                    if const.AGGREGATES not in [prop.name for prop in as_list(param.props)]:
+                        logger.warning(f'Parameter {param_id} has no values and was referenced by prose.')
             elif show_value_warnings:
                 logger.warning(f'Control prose references param {param_ids[i]} with no specified value.')
         # there may be staches remaining that we can't replace if not in param_dict