feat: use oscal_read and oscal_write for component definition update

Author: qduanmu

trestlebot/tasks/sync_cac_content_task.py

@@ -3,6 +3,7 @@
 
 """Trestle Bot Sync CaC Content Tasks"""
 
+import datetime
 import json
 import logging
 import os
@@ -33,7 +34,6 @@
     RuleInfo,
     RulesTransformer,
     get_component_info,
-    update_component_definition,
 )
 
 
@@ -190,17 +190,25 @@ def _create_or_update_compdef(self, compdef_type: str = "service") -> None:
         cd_json = cd_dir / "component-definition.json"
         if cd_json.exists():
             logger.info(f"The component definition for {self.product} exists.")
-            with open(cd_json, "r", encoding="utf-8") as f:
-                data = json.load(f)
-                components = data["component-definition"]["components"]
-            for index, component in enumerate(components):
-                if component.get("title") == oscal_component.title:
-                    # The update should be skipped if no content changes
-                    logger.info(f"Update props of component {product_name}")
-                    data["component-definition"]["components"][index][
-                        "props"
-                    ] = oscal_component.props
-                    update_component_definition(cd_json)
+            compdef = ComponentDefinition.oscal_read(cd_json)
+            updated = False
+            for index, component in enumerate(compdef.components):
+                if component.title == oscal_component.title:
+                    if component.props != oscal_component.props:
+                        compdef.components[index].props = oscal_component.props
+                        updated = True
+                        break
+            if updated:
+                logger.info(f"Update component definition: {cd_json}")
+                compdef.metadata.version = str(
+                    "{:.1f}".format(float(compdef.metadata.version) + 0.1)
+                )
+                compdef.metadata.last_modified = (
+                    datetime.datetime.now(datetime.timezone.utc)
+                    .replace(microsecond=0)
+                    .isoformat()
+                )
+                compdef.oscal_write(cd_json)
         else:
             logger.info(f"Creating component definition for product {self.product}")
             cd_dir.mkdir(exist_ok=True, parents=True)

trestlebot/transformers/cac_transformer.py

@@ -3,13 +3,11 @@
 
 """Transform rules from existing Compliance as Code locations into OSCAL properties."""
 
-import datetime
 import json
 import logging
 import os
 import re
 from html.parser import HTMLParser
-from pathlib import Path
 from typing import Any, Dict, List, Optional, Tuple
 
 import ssg.build_yaml
@@ -43,20 +41,6 @@ def get_component_info(product_name: str, cac_path: str) -> Tuple[str, str]:
         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)
-
-
 def add_prop(name: str, value: str, remarks: Optional[str] = None) -> Property:
     """Add a property to a set of rule properties."""
     prop = generate_sample_model(Property)