nist_sync: split ControlCatalogs by NIST family

utils/nist_sync/export_to_gemara.py

@@ -4,6 +4,9 @@
 
 Reads product-specific NIST 800-53 control files and produces per product:
   - control_catalog.yaml  (ControlCatalog: NIST controls → XCCDF rule IDs)
+  - by_group/nist-800-53-rev5-{product}-{family}-catalog.yaml
+                           (same ControlCatalog, split one file per NIST
+                           family — mirrors complytime-policies' layout)
   - rules_mapping.yaml    (MappingDocument: traceability between layers)
   - products/{product}/profiles/nist_800_53.profile  (XCCDF tailoring base)
 
@@ -48,6 +51,7 @@
 from gemara.mapping import GemaraMappingBuilder
 from gemara.policy import product_full_name
 from gemara.schema import validate_catalog, validate_guidance, validate_mapping
+from gemara.yaml_output import configure_yaml
 
 
 DEFAULT_PRODUCTS = ["rhel8", "rhel9", "rhel10", "fedora"]
@@ -185,11 +189,7 @@ def load_policy(product: str, repo_root: Path) -> ssg.controls.Policy:
 
 
 def _yaml_instance() -> YAML:
-    yaml = YAML()
-    yaml.default_flow_style = False
-    yaml.allow_unicode = True
-    yaml.width = 120
-    return yaml
+    return configure_yaml(YAML())
 
 
 def write_yaml(data: Any, path: Path) -> None:
@@ -315,6 +315,16 @@ def export_product(
     if verbose:
         print(f"  Wrote {catalog_path}")
 
+    # Write one ControlCatalog per NIST family (AC, AU, SI, ...) alongside the
+    # combined one, matching complytime-policies' split-by-group layout.
+    by_group_dir = product_dir / "by_group"
+    by_group_dir.mkdir(parents=True, exist_ok=True)
+    for fam_id, fam_catalog in builder.build_by_group().items():
+        fam_path = by_group_dir / f"nist-800-53-rev5-{product}-{fam_id}-catalog.yaml"
+        write_yaml(fam_catalog, fam_path)
+    if verbose:
+        print(f"  Wrote {by_group_dir}/ ({len(list(by_group_dir.glob('*.yaml')))} family catalogs)")
+
     # Generate the XCCDF tailoring base profile (not committed — see .gitignore)
     _write_xccdf_profile(product, repo_root, verbose)
 

utils/nist_sync/gemara/catalog.py

@@ -226,3 +226,41 @@ def build(self) -> Dict[str, Any]:
             "groups": self._groups(),
             "controls": controls,
         }
+
+    def build_by_group(self) -> Dict[str, Dict[str, Any]]:
+        """Return one ControlCatalog dict per NIST 800-53 family/group.
+
+        Mirrors the layout complytime-policies splits the combined catalog
+        into (one file per family, e.g. AC, AU, SI) for a reduced line count
+        and to keep each catalog scoped to a single 'groups' entry.
+        """
+        by_group: Dict[str, Dict[str, Any]] = {}
+        for fam_id, fam_title in NIST_FAMILIES.items():
+            fam_controls = [
+                self._build_control(ctrl)
+                for ctrl in self.policy.controls
+                if _extract_family(ctrl.id) == fam_id
+            ]
+            if not fam_controls:
+                continue
+            by_group[fam_id] = {
+                "metadata": self._metadata_for_group(fam_id),
+                "title": f"{self.policy.title} - {fam_title}",
+                "groups": [{
+                    "id": fam_id,
+                    "title": fam_title,
+                    "description": f"NIST 800-53 {fam_id.upper()} family: {fam_title}",
+                }],
+                "controls": fam_controls,
+            }
+        return by_group
+
+    def _metadata_for_group(self, fam_id: str) -> Dict[str, Any]:
+        meta = self._metadata()
+        meta["id"] = f"nist-800-53-rev5-{self.product}-{fam_id}"
+        meta["description"] = (
+            f"NIST Special Publication 800-53 Revision 5 controls for "
+            f"{self.product.upper()}, generated from ComplianceAsCode "
+            f"({fam_id.upper()} family)"
+        )
+        return meta

utils/nist_sync/generate_complyctl_bundle.py

@@ -51,6 +51,7 @@
 
 sys.path.insert(0, str(_SCRIPT_DIR))
 from gemara.policy import RuleEntry, extract_rules_from_catalog, generate_policy  # noqa: E402
+from gemara.yaml_output import configure_yaml  # noqa: E402
 
 # OCI media types for complyctl v1.0.0-alpha.0 (go-gemara v0.0.1 split-layer format)
 _MEDIA_TYPE_POLICY = "application/vnd.gemara.policy.v1+yaml"
@@ -63,11 +64,7 @@ def _now_iso() -> str:
 
 
 def _yaml() -> YAML:
-    y = YAML()
-    y.default_flow_style = False
-    y.allow_unicode = True
-    y.width = 120
-    return y
+    return configure_yaml(YAML())
 
 
 def load_yaml(path: Path) -> Any:

utils/nist_sync/generate_policies_pr.py

@@ -10,7 +10,9 @@
     │   └── nist-800-53-rev5-{product}.yaml        # bundle manifest
     ├── governance/
     │   ├── catalogs/
-    │   │   └── nist-800-53-rev5-{product}-catalog.yaml
+    │   │   ├── nist-800-53-rev5-{product}-catalog.yaml   # combined, used by the Policy
+    │   │   └── nist-800-53/{product}/nist-800-53-{product}/
+    │   │       └── nist-800-53-rev5-{product}-{family}-catalog.yaml  # split by NIST family
     │   ├── guidance/
     │   │   └── nist-800-53-rev5-guidance.yaml      # shared across products
     │   └── policies/
@@ -49,16 +51,13 @@
 sys.path.insert(0, str(_SCRIPT_DIR))
 from gemara.policy import extract_rules_from_catalog, generate_policy  # noqa: E402
 from gemara.schema import validate_policy  # noqa: E402
+from gemara.yaml_output import configure_yaml  # noqa: E402
 
 _GUIDANCE_ID = "nist-800-53-rev5-guidance"
 
 
 def _yaml() -> YAML:
-    y = YAML()
-    y.default_flow_style = False
-    y.allow_unicode = True
-    y.width = 120
-    return y
+    return configure_yaml(YAML())
 
 
 def load_yaml(path: Path) -> Any:
@@ -125,6 +124,15 @@ def stage_product(product: str, gemara_dir: Path, output_dir: Path, has_guidance
     shutil.copy2(catalog_path, catalog_dest)
     print(f"  [{product.upper()}] Catalog:  {catalog_dest.relative_to(output_dir)}")
 
+    by_group_dir = gemara_dir / product / "by_group"
+    if by_group_dir.is_dir():
+        split_dir = catalogs_dir / "nist-800-53" / product / f"nist-800-53-{product}"
+        split_dir.mkdir(parents=True, exist_ok=True)
+        fam_files = sorted(by_group_dir.glob("*.yaml"))
+        for fam_file in fam_files:
+            shutil.copy2(fam_file, split_dir / fam_file.name)
+        print(f"  [{product.upper()}] Split catalogs: {split_dir.relative_to(output_dir)}/ ({len(fam_files)} files)")
+
     policy_dest = policies_dir / f"{pid}-policy.yaml"
     dump_yaml(policy, policy_dest)
     print(f"  [{product.upper()}] Policy:   {policy_dest.relative_to(output_dir)}")

utils/nist_sync/test_gemara_export.py

@@ -15,6 +15,7 @@
 """
 
 import argparse
+import io
 import sys
 from pathlib import Path
 from typing import Any, Dict, List
@@ -39,6 +40,7 @@
 sys.path.insert(0, str(_SCRIPT_DIR))
 from gemara.policy import extract_rules_from_catalog, generate_policy  # noqa: E402
 from gemara.schema import validate_policy  # noqa: E402
+from gemara.yaml_output import configure_yaml  # noqa: E402
 
 
 def load_yaml(path: Path) -> Any:
@@ -77,6 +79,30 @@ def check(self, condition: bool, ok_msg: str, fail_msg: str) -> None:
             self.fail(fail_msg)
 
 
+def test_yaml_serialization(result: TestResult) -> None:
+    """Keep generated YAML compatible with the shared yamllint defaults."""
+    yaml = configure_yaml(YAML())
+    data = {
+        "layers": ["governance/catalog.yaml"],
+        "metadata": {"groups": [{"id": "ac"}]},
+        "description": "word " * 200,
+    }
+    output = io.StringIO()
+    yaml.dump(data, output)
+    text = output.getvalue()
+
+    result.check(
+        "layers:\n  - governance/catalog.yaml\n" in text,
+        "nested sequences use yamllint-compatible indentation",
+        "nested sequence is not indented beneath its mapping key",
+    )
+    result.check(
+        not any(line.endswith((" ", "\t")) for line in text.splitlines()),
+        "generated YAML has no trailing whitespace",
+        "generated YAML contains trailing whitespace",
+    )
+
+
 # ---------------------------------------------------------------------------
 # Test suites
 # ---------------------------------------------------------------------------
@@ -688,6 +714,7 @@ def main() -> None:
     print("Policy validation (product-independent)")
     print(f"{'='*60}")
     validation_result = TestResult()
+    test_yaml_serialization(validation_result)
     test_validate_policy_catches_errors(validation_result)
     test_policy_parameters_from_variables(validation_result)
     all_passed += len(validation_result.passed)