feat: populate cac content product name as component title for CPLYTM-380 and CPLYTM-381

pyproject.toml

@@ -29,6 +29,7 @@ github3-py = "^4.0.1"
 python-gitlab = "^4.2.0"
 ruamel-yaml = "^0.18.5"
 pydantic = "^2.0.0"
+ssg = {git = "https://github.com/ComplianceasCode/content"}
 
 [tool.poetry.group.dev]
 optional = true
@@ -106,3 +107,8 @@ ignore_missing_imports = true
 [[tool.mypy.overrides]]
 module = "ruamel"
 ignore_missing_imports = true
+
+[[tool.mypy.overrides]]
+module = "ssg.products"
+ignore_missing_imports = true
+

tests/data/content/products/ocp4/product.yml

@@ -0,0 +1,132 @@
+product: ocp4
+full_name: Red Hat OpenShift Container Platform 4
+type: platform
+
+benchmark_id: OCP-4
+benchmark_root: "../../applications"
+
+profiles_root: "./profiles"
+
+pkg_system: "rpm"
+
+init_system: "systemd"
+
+reference_uris:
+  cis: 'https://www.cisecurity.org/benchmark/kubernetes/'
+  stigid: 'https://public.cyber.mil/stigs/downloads/?_dl_facet_stigs=container-platform'
+
+cpes_root: "../../shared/applicability"
+cpes:
+  - ocp4:
+      name: "cpe:/a:redhat:openshift_container_platform:4.1"
+      title: "Red Hat OpenShift Container Platform 4"
+      check_id: installed_app_is_ocp4
+
+  - ocp4-node:
+      name: "cpe:/o:redhat:openshift_container_platform_node:4"
+      title: "Red Hat OpenShift Container Platform 4 Node"
+      check_id: installed_app_is_ocp4_node
+
+  - ocp4-node-on-ovn:
+      name: "cpe:/a:redhat:openshift_container_platform_node_on_ovn:4"
+      title: "Red Hat OpenShift Container Platform 4 Node on OVN"
+      check_id: installed_app_is_ocp4_node_on_openshift-ovn
+
+  - ocp4-node-on-sdn:
+      name: "cpe:/a:redhat:openshift_container_platform_node_on_sdn:4"
+      title: "Red Hat OpenShift Container Platform 4 Node on SDN"
+      check_id: installed_app_is_ocp4_node_on_openshift-sdn
+
+  - ocp4.6:
+      name: "cpe:/a:redhat:openshift_container_platform:4.6"
+      title: "Red Hat OpenShift Container Platform 4.6"
+      check_id: installed_app_is_ocp4_6
+
+  - ocp4.7:
+      name: "cpe:/a:redhat:openshift_container_platform:4.7"
+      title: "Red Hat OpenShift Container Platform 4.7"
+      check_id: installed_app_is_ocp4_7
+
+  - ocp4.8:
+      name: "cpe:/a:redhat:openshift_container_platform:4.8"
+      title: "Red Hat OpenShift Container Platform 4.8"
+      check_id: installed_app_is_ocp4_8
+
+  - ocp4.9:
+      name: "cpe:/a:redhat:openshift_container_platform:4.9"
+      title: "Red Hat OpenShift Container Platform 4.9"
+      check_id: installed_app_is_ocp4_9
+
+  - ocp4.10:
+      name: "cpe:/a:redhat:openshift_container_platform:4.10"
+      title: "Red Hat OpenShift Container Platform 4.10"
+      check_id: installed_app_is_ocp4_10
+
+  - ocp4.11:
+      name: "cpe:/a:redhat:openshift_container_platform:4.11"
+      title: "Red Hat OpenShift Container Platform 4.11"
+      check_id: installed_app_is_ocp4_11
+
+  - ocp4.12:
+      name: "cpe:/a:redhat:openshift_container_platform:4.12"
+      title: "Red Hat OpenShift Container Platform 4.12"
+      check_id: installed_app_is_ocp4_12
+
+  - ocp4.13:
+      name: "cpe:/a:redhat:openshift_container_platform:4.13"
+      title: "Red Hat OpenShift Container Platform 4.13"
+      check_id: installed_app_is_ocp4_13
+
+  - ocp4.14:
+      name: "cpe:/a:redhat:openshift_container_platform:4.14"
+      title: "Red Hat OpenShift Container Platform 4.14"
+      check_id: installed_app_is_ocp4_14
+
+  - ocp4.15:
+      name: "cpe:/a:redhat:openshift_container_platform:4.15"
+      title: "Red Hat OpenShift Container Platform 4.15"
+      check_id: installed_app_is_ocp4_15
+
+  - ocp4.16:
+      name: "cpe:/a:redhat:openshift_container_platform:4.16"
+      title: "Red Hat OpenShift Container Platform 4.16"
+      check_id: installed_app_is_ocp4_16
+
+  - ocp4.17:
+      name: "cpe:/a:redhat:openshift_container_platform:4.17"
+      title: "Red Hat OpenShift Container Platform 4.17"
+      check_id: installed_app_is_ocp4_17
+
+  - ocp4.18:
+      name: "cpe:/a:redhat:openshift_container_platform:4.18"
+      title: "Red Hat OpenShift Container Platform 4.18"
+      check_id: installed_app_is_ocp4_18
+
+  - ocp4-on-aws:
+      name: "cpe:/a:redhat:openshift_container_platform_on_aws:4"
+      title: "Red Hat OpenShift Container Platform 4 on AWS"
+      check_id: installed_app_is_ocp4_on_aws
+
+  - ocp4-on-azure:
+      name: "cpe:/a:redhat:openshift_container_platform_on_azure:4"
+      title: "Red Hat OpenShift Container Platform 4 on Azure"
+      check_id: installed_app_is_ocp4_on_azure
+
+  - ocp4-on-gcp:
+      name: "cpe:/a:redhat:openshift_container_platform_on_gcp:4"
+      title: "Red Hat OpenShift Container Platform 4 on GCP"
+      check_id: installed_app_is_ocp4_on_gcp
+
+  - ocp4-on-ovn:
+      name: "cpe:/a:redhat:openshift_container_platform_on_ovn:4"
+      title: "Red Hat OpenShift Container Platform 4 on OVN"
+      check_id: installed_app_is_ocp4_on_openshiftovn
+
+  - ocp4-on-sdn:
+      name: "cpe:/a:redhat:openshift_container_platform_on_sdn:4"
+      title: "Red Hat OpenShift Container Platform 4 on SDN"
+      check_id: installed_app_is_ocp4_on_openshiftsdn
+
+
+# Requirement string, see: https://setuptools.readthedocs.io/en/latest/pkg_resources.html#requirements-parsing
+# requires: "openscap>=1.3.4"

tests/trestlebot/cli/test_sync_cac_content_cmd.py

@@ -8,10 +8,16 @@
 from click.testing import CliRunner
 from git import Repo
 
+from tests.testutils import setup_for_catalog, setup_for_profile
 from trestlebot.cli.commands.sync_cac_content import sync_cac_content_cmd
 
 
 test_product = "ocp4"
+cac_content_test_data = pathlib.Path("tests/data/content").resolve()
+test_prof_path = pathlib.Path("tests/data/json/").resolve()
+test_prof = "simplified_nist_profile"
+test_cat = "simplified_nist_catalog"
+test_comp_path = f"component-definitions/{test_product}/component-definition.json"
 
 
 def test_missing_required_option(tmp_repo: Tuple[str, Repo]) -> None:
@@ -37,3 +43,44 @@ def test_missing_required_option(tmp_repo: Tuple[str, Repo]) -> None:
         ],
     )
     assert result.exit_code == 2
+
+
+def test_sync_product_name(tmp_repo: Tuple[str, Repo]) -> None:
+    """Tests sync Cac content product name to OSCAL component title ."""
+    repo_dir, _ = tmp_repo
+    repo_path = pathlib.Path(repo_dir)
+    setup_for_catalog(repo_path, test_cat, "catalog")
+    setup_for_profile(repo_path, test_prof, "profile")
+
+    runner = CliRunner()
+    result = runner.invoke(
+        sync_cac_content_cmd,
+        [
+            "--product",
+            test_product,
+            "--repo-path",
+            str(repo_path.resolve()),
+            "--cac-content-root",
+            cac_content_test_data,
+            "--cac-profile",
+            "cac-profile",
+            "--oscal-profile",
+            test_prof,
+            "--committer-email",
+            "[email protected]",
+            "--committer-name",
+            "test name",
+            "--branch",
+            "test",
+            "--dry-run",
+        ],
+    )
+    # Check the CLI sync-cac-content is successful
+    assert result.exit_code == 0
+    # Check if the component definition is created
+    component_definition = repo_path.joinpath(test_comp_path)
+    assert component_definition.exists()
+    # Check if it populates the product name as the component title
+    with open(component_definition, "r", encoding="utf-8") as file:
+        content = file.read()
+    assert '"title": "ocp4"' in content

trestlebot/cli/commands/sync_cac_content.py

@@ -8,6 +8,7 @@
 import click
 
 from trestlebot.cli.options.common import common_options, git_options, handle_exceptions
+from trestlebot.tasks.authored.compdef import AuthoredComponentDefinition
 
 
 logger = logging.getLogger(__name__)
@@ -53,8 +54,25 @@
 @handle_exceptions
 def sync_cac_content_cmd(ctx: click.Context, **kwargs: Any) -> None:
     """Transform CaC content to OSCAL component definition."""
-
     # Steps:
     # 1. Check options, logger errors if any and exit.
-    # 2. Create a new task to run the data transformation.
-    # 3. Initialize a Trestlebot object and run the task(s).
+    # 2. Initial product component definition with product name
+    # 3. Create a new task to run the data transformation.
+    # 4. Initialize a Trestlebot object and run the task(s).
+
+    # pre_tasks: List[TaskBase] = []
+
+    product = kwargs["product"]
+    cac_content_root = kwargs["cac_content_root"]
+    component_definition_type = kwargs.get("component_definition_type", "service")
+    working_dir = kwargs["repo_path"]
+
+    authored_comp: AuthoredComponentDefinition = AuthoredComponentDefinition(
+        trestle_root=working_dir,
+    )
+    authored_comp.create_update_cac_compdef(
+        comp_type=component_definition_type,
+        product=product,
+        cac_content_root=cac_content_root,
+        working_dir=working_dir,
+    )

trestlebot/tasks/authored/compdef.py

@@ -4,6 +4,8 @@
 
 """Trestle Bot functions for component definition authoring"""
 
+import json
+import logging
 import os
 import pathlib
 from typing import Callable, List, Optional
@@ -13,14 +15,20 @@
 from trestle.common.err import TrestleError
 from trestle.common.model_utils import ModelUtils
 from trestle.core.catalog.catalog_interface import CatalogInterface
+from trestle.core.generators import generate_sample_model
 from trestle.core.profile_resolver import ProfileResolver
 from trestle.core.repository import AgileAuthoring
+from trestle.oscal.component import ComponentDefinition, DefinedComponent
 
 from trestlebot.const import RULE_PREFIX, RULES_VIEW_DIR, YAML_EXTENSION
 from trestlebot.tasks.authored.base_authored import (
     AuthoredObjectBase,
     AuthoredObjectException,
 )
+from trestlebot.transformers.cac_transformer import (
+    get_component_info,
+    update_component_definition,
+)
 from trestlebot.transformers.trestle_rule import (
     ComponentInfo,
     Control,
@@ -30,6 +38,9 @@
 from trestlebot.transformers.yaml_transformer import FromRulesYAMLTransformer
 
 
+logger = logging.getLogger(__name__)
+
+
 class FilterByProfile:
     """Filter controls by a profile."""
 
@@ -158,6 +169,61 @@ def create_new_default(
         )
         rules_view_builder.write_to_yaml(rule_dir)
 
+    def create_update_cac_compdef(
+        self,
+        comp_type: str,
+        product: str,
+        cac_content_root: str,
+        working_dir: str,
+    ) -> None:
+        """Create component definition for cac content
+
+        Args:
+            comp_description: Description of the component
+            comp_type: Type of the component
+            product: Product name for the component
+            cac_content_root: ComplianceAsCode repo path
+            working_dir: workplace repo path
+        """
+        # Initial component definition fields
+        component_definition = generate_sample_model(ComponentDefinition)
+        component_definition.metadata.title = f"Component definition for {product}"
+        component_definition.metadata.version = "1.0"
+        component_definition.components = list()
+        oscal_component = generate_sample_model(DefinedComponent)
+        product_name, full_name = get_component_info(product, cac_content_root)
+        oscal_component.title = product_name
+        oscal_component.description = full_name
+        oscal_component.type = comp_type
+
+        # Create all of the component properties for rules
+        # This part will be updated in CPLYTM-218
+        """
+        rules: List[RuleInfo] = self.rules_transformer.get_all_rules()
+        all_rule_properties: List[Property] = self.rules_transformer.transform(rules)
+        oscal_component.props = none_if_empty(all_rule_properties)
+        """
+        repo_path = pathlib.Path(working_dir)
+        out_path = repo_path.joinpath(f"{const.MODEL_DIR_COMPDEF}/{product}/")
+        oname = "component-definition.json"
+        ofile = out_path / oname
+        if ofile.exists():
+            logger.info(f"The component for product {product} exists.")
+            with open(ofile, "r", encoding="utf-8") as f:
+                data = json.load(f)
+            for component in data["component-definition"]["components"]:
+                if component.get("title") == oscal_component.title:
+                    logger.info("Update the exsisting component definition.")
+                    # Need to update props parts if the rules updated
+                    # Update the version and last modify time
+                    update_component_definition(ofile)
+        else:
+            logger.info(f"Creating component definition for product {product}")
+            out_path.mkdir(exist_ok=True, parents=True)
+            ofile = out_path / oname
+            component_definition.components.append(oscal_component)
+            component_definition.oscal_write(ofile)
+
 
 class RulesViewBuilder:
     """Write TrestleRule objects to YAML files in rules view."""

trestlebot/transformers/cac_transformer.py

@@ -0,0 +1,38 @@
+# SPDX-License-Identifier: Apache-2.0
+# Copyright (c) 2024 Red Hat, Inc.
+
+import datetime
+import json
+from pathlib import Path
+from typing import Tuple
+
+from ssg.products import load_product_yaml, product_yaml_path
+
+
+def get_component_info(product_name: str, cac_path: str) -> Tuple[str, str]:
+    """Get the product name from product yml file via the SSG library."""
+    if product_name and cac_path:
+        # Get the product yaml file path
+        product_yml_path = product_yaml_path(cac_path, product_name)
+        # Load the product data
+        product = load_product_yaml(product_yml_path)
+        # Return product name from product yml file
+        component_title = product._primary_data.get("product")
+        component_description = product._primary_data.get("full_name")
+        return (component_title, component_description)
+    else:
+        raise ValueError("component_title is empty or None")
+
+
+def update_component_definition(compdef_file: Path) -> None:
+    # Update the component definition version and modify time
+    with open(compdef_file, "r", encoding="utf-8") as f:
+        data = json.load(f)
+    current_version = data["component-definition"]["metadata"]["version"]
+    data["component-definition"]["metadata"]["version"] = str(
+        "{:.1f}".format(float(current_version) + 0.1)
+    )
+    current_time = datetime.datetime.now().isoformat()
+    data["component-definition"]["metadata"]["last-modified"] = current_time
+    with open(compdef_file, "w", encoding="utf-8") as f:
+        json.dump(data, f, ensure_ascii=False, indent=2)