diff --git a/utils/nist_sync/export_to_gemara.py b/utils/nist_sync/export_to_gemara.py index 0d34e0a0f395..cf9150597521 100644 --- a/utils/nist_sync/export_to_gemara.py +++ b/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) diff --git a/utils/nist_sync/gemara/catalog.py b/utils/nist_sync/gemara/catalog.py index 0a0c33d09680..2309c5ff435e 100644 --- a/utils/nist_sync/gemara/catalog.py +++ b/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 diff --git a/utils/nist_sync/generate_complyctl_bundle.py b/utils/nist_sync/generate_complyctl_bundle.py index 0f05e49177bf..119c25b267d4 100644 --- a/utils/nist_sync/generate_complyctl_bundle.py +++ b/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: diff --git a/utils/nist_sync/generate_policies_pr.py b/utils/nist_sync/generate_policies_pr.py index dbfb797cc61b..67bbff440434 100644 --- a/utils/nist_sync/generate_policies_pr.py +++ b/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)}") diff --git a/utils/nist_sync/test_gemara_export.py b/utils/nist_sync/test_gemara_export.py index 4aadb6e8e80a..35b545d4f541 100644 --- a/utils/nist_sync/test_gemara_export.py +++ b/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)