Merge pull request #10048 from jan-cerny/refactor_build_ovals

Refactor ssg.build_ovals module

build-scripts/combine_ovals.py

@@ -33,13 +33,18 @@ def parse_args():
         help="Directory to store intermediate built OVAL files."
     )
     p.add_argument("--output", type=argparse.FileType("wb"), required=True)
-    p.add_argument("ovaldirs", metavar="OVAL_DIR", nargs="+",
-                   help="Shared directory(ies) from which we will collect "
-                   "OVAL definitions to combine. Order matters, latter "
-                   "directories override former. These will be overwritten "
-                   "by OVALs in the product_yaml['guide'] directory (which "
-                   "in turn preference oval/{{{ product }}}.xml over "
-                   "oval/shared.xml for a given rule.")
+    p.add_argument(
+        "--include-benchmark", action="store_true",
+        help="Include OVAL checks from rule directories in the benchmark "
+        "directory tree which is specified by product.yml "
+        "in the `benchmark_root` key.")
+    p.add_argument(
+        "ovaldirs", metavar="OVAL_DIR", nargs="+",
+        help="Shared directory(ies) from which we will collect OVAL "
+        "definitions to combine. Order matters, latter directories override "
+        "former. If --include-benchmark is provided, these will be "
+        "overwritten by OVALs in the rule directory (which in turn preference "
+        "oval/{{{ product }}}.xml over oval/shared.xml for a given rule.")
 
     return p.parse_args()
 
@@ -57,12 +62,12 @@ def main():
         ssg.utils.required_key(env_yaml, "target_oval_version_str"),
         ssg.utils.required_key(env_yaml, "ssg_version"))
 
-    body = ssg.build_ovals.checks(
+    oval_builder = ssg.build_ovals.OVALBuilder(
         env_yaml,
         args.product_yaml,
-        ssg.utils.required_key(env_yaml, "target_oval_version_str"),
         args.ovaldirs,
         args.build_ovals_dir)
+    body = oval_builder.build_shorthand(args.include_benchmark)
 
     # parse new file(string) as an ssg.xml.ElementTree, so we can reorder elements
     # appropriately

cmake/SSGCommon.cmake

@@ -262,7 +262,7 @@ macro(ssg_build_oval_unlinked PRODUCT)
     set(OVAL_COMBINE_PATHS "${BUILD_CHECKS_DIR}/shared/oval" "${SSG_SHARED}/checks/oval" "${BUILD_CHECKS_DIR}/oval" "${CMAKE_CURRENT_SOURCE_DIR}/checks/oval")
     add_custom_command(
         OUTPUT "${CMAKE_CURRENT_BINARY_DIR}/oval-unlinked.xml"
-        COMMAND env "PYTHONPATH=$ENV{PYTHONPATH}" "${PYTHON_EXECUTABLE}" "${SSG_BUILD_SCRIPTS}/combine_ovals.py" --build-config-yaml "${CMAKE_BINARY_DIR}/build_config.yml" --product-yaml "${CMAKE_CURRENT_SOURCE_DIR}/product.yml" --output "${CMAKE_CURRENT_BINARY_DIR}/oval-unlinked.xml" --build-ovals-dir "${CMAKE_CURRENT_BINARY_DIR}/checks/oval" ${OVAL_COMBINE_PATHS}
+        COMMAND env "PYTHONPATH=$ENV{PYTHONPATH}" "${PYTHON_EXECUTABLE}" "${SSG_BUILD_SCRIPTS}/combine_ovals.py" --include-benchmark --build-config-yaml "${CMAKE_BINARY_DIR}/build_config.yml" --product-yaml "${CMAKE_CURRENT_SOURCE_DIR}/product.yml" --output "${CMAKE_CURRENT_BINARY_DIR}/oval-unlinked.xml" --build-ovals-dir "${CMAKE_CURRENT_BINARY_DIR}/checks/oval" ${OVAL_COMBINE_PATHS}
         COMMAND "${XMLLINT_EXECUTABLE}" --format --output "${CMAKE_CURRENT_BINARY_DIR}/oval-unlinked.xml" "${CMAKE_CURRENT_BINARY_DIR}/oval-unlinked.xml"
         DEPENDS generate-internal-templated-content-${PRODUCT}
         COMMENT "[${PRODUCT}-content] generating oval-unlinked.xml"

ssg/build_ovals.py

@@ -289,106 +289,168 @@ def _check_rule_id(oval_file_tree, rule_id):
     return False
 
 
-def checks(env_yaml, yaml_path, oval_version, oval_dirs, build_ovals_dir=None):
-    """
-    Concatenate all XML files in the oval directory, to create the document
-    body. Then concatenates this with all XML files in the guide directories,
-    preferring {{{ product }}}.xml to shared.xml.
-
-    oval_dirs: list of directory with oval files (later has higher priority)
+def _create_output_directory(directory_path=None):
+    if directory_path and not os.path.exists(directory_path):
+        os.makedirs(directory_path)
+
+
+def _list_full_paths(directory):
+    full_paths = [os.path.join(directory, x) for x in os.listdir(directory)]
+    return sorted(full_paths)
+
+
+class OVALBuilder:
+    def __init__(
+            self, env_yaml, product_yaml_path, shared_directories,
+            build_ovals_dir):
+        self.env_yaml = env_yaml
+        self.product_yaml_path = product_yaml_path
+        self.shared_directories = shared_directories
+        self.build_ovals_dir = build_ovals_dir
+        self.already_loaded = dict()
+        self.oval_version = utils.required_key(
+            env_yaml, "target_oval_version_str")
+        self.product = utils.required_key(env_yaml, "product")
+
+    def build_shorthand(self, include_benchmark):
+        _create_output_directory(self.build_ovals_dir)
+        all_checks = []
+        if include_benchmark:
+            all_checks += self._get_checks_from_benchmark()
+        all_checks += self._get_checks_from_shared_directories()
+        document_body = "".join(all_checks)
+        return document_body
+
+    def _get_checks_from_benchmark(self):
+        product_dir = os.path.dirname(self.product_yaml_path)
+        relative_guide_dir = utils.required_key(self.env_yaml, "benchmark_root")
+        guide_dir = os.path.abspath(
+            os.path.join(product_dir, relative_guide_dir))
+        additional_content_directories = self.env_yaml.get(
+            "additional_content_directories", [])
+        dirs_to_scan = [guide_dir]
+        for rd in additional_content_directories:
+            abspath = os.path.abspath(os.path.join(product_dir, rd))
+            dirs_to_scan.append(abspath)
+        rule_dirs = list(find_rule_dirs_in_paths(dirs_to_scan))
+        oval_checks = self._process_directories(rule_dirs, True)
+        return oval_checks
+
+    def _get_checks_from_shared_directories(self):
+        # earlier directory has higher priority
+        reversed_dirs = self.shared_directories[::-1]
+        oval_checks = self._process_directories(reversed_dirs, False)
+        return oval_checks
+
+    def _process_directories(self, directories, from_benchmark):
+        oval_checks = []
+        for directory in directories:
+            if not os.path.exists(directory):
+                continue
+            oval_checks += self._process_directory(directory, from_benchmark)
+        return oval_checks
 
-    Return: The document body
-    """
+    def _get_list_of_oval_files(self, directory, from_benchmark):
+        if from_benchmark:
+            oval_files = get_rule_dir_ovals(directory, self.product)
+        else:
+            oval_files = _list_full_paths(directory)
+        return oval_files
 
-    body = []
-    product = utils.required_key(env_yaml, "product")
-    included_checks_count = 0
-    reversed_dirs = oval_dirs[::-1]  # earlier directory has higher priority
-    already_loaded = dict()  # filename -> oval_version
-    local_env_yaml = dict()
-    local_env_yaml.update(env_yaml)
-
-    product_dir = os.path.dirname(yaml_path)
-    relative_guide_dir = utils.required_key(env_yaml, "benchmark_root")
-    guide_dir = os.path.abspath(os.path.join(product_dir, relative_guide_dir))
-    additional_content_directories = env_yaml.get("additional_content_directories", [])
-    add_content_dirs = [os.path.abspath(os.path.join(product_dir, rd)) for rd in additional_content_directories]
-
-    if build_ovals_dir:
-        # Create output directory if it doesn't yet exist.
-        if not os.path.exists(build_ovals_dir):
-            os.makedirs(build_ovals_dir)
-
-    for _dir_path in find_rule_dirs_in_paths([guide_dir] + add_content_dirs):
-        rule_id = get_rule_dir_id(_dir_path)
-
-        rule_path = os.path.join(_dir_path, "rule.yml")
+    def _process_directory(self, directory, from_benchmark):
         try:
-            rule = Rule.from_yaml(rule_path, env_yaml)
+            context = self._get_context(directory, from_benchmark)
         except DocumentationNotComplete:
-            # Happens on non-debug build when a rule is "documentation-incomplete"
-            continue
-        prodtypes = parse_prodtype(rule.prodtype)
-
-        local_env_yaml['rule_id'] = rule.id_
-        local_env_yaml['rule_title'] = rule.title
-        local_env_yaml['products'] = prodtypes # default is all
-
-        for _path in get_rule_dir_ovals(_dir_path, product):
-            # To be compatible with the later checks, use the rule_id
-            # (i.e., the value of _dir) to recreate the expected filename if
-            # this OVAL was in a rule directory.
-            filename = "%s.xml" % rule_id
-
-            xml_content = process_file_with_macros(_path, local_env_yaml)
-
-            if not _check_is_applicable_for_product(xml_content, product):
+            return []
+        oval_files = self._get_list_of_oval_files(directory, from_benchmark)
+        oval_checks = self._get_directory_oval_checks(
+            context, oval_files, from_benchmark)
+        return oval_checks
+
+    def _get_directory_oval_checks(self, context, oval_files, from_benchmark):
+        oval_checks = []
+        for file_path in oval_files:
+            xml_content = self._process_oval_file(
+                file_path, from_benchmark, context)
+            if xml_content is None:
                 continue
+            oval_checks.append(xml_content)
+        return oval_checks
 
-            if build_ovals_dir:
-                # store intermediate files
-                output_file_name = rule_id + ".xml"
-                output_filepath = os.path.join(build_ovals_dir, output_file_name)
-                with open(output_filepath, "w") as f:
-                    f.write(xml_content)
-
-            if _check_is_loaded(already_loaded, filename, oval_version):
-                continue
-            oval_file_tree = _create_oval_tree_from_string(xml_content)
-            if not _check_rule_id(oval_file_tree, rule_id,):
-                msg = "OVAL definition in '%s' doesn't match rule ID '%s'." % (
-                    _path, rule_id)
-                print(msg, file=sys.stderr)
-            if not _check_oval_version_from_oval(oval_file_tree, oval_version):
-                continue
-
-            body.append(xml_content)
-            included_checks_count += 1
-            already_loaded[filename] = oval_version
-
-    for oval_dir in reversed_dirs:
-        if not os.path.isdir(oval_dir):
-            continue
-        # sort the files to make output deterministic
-        for filename in sorted(os.listdir(oval_dir)):
-            if not filename.endswith(".xml"):
-                continue
-            oval_file_path = os.path.join(oval_dir, filename)
-            if "checks_from_templates" in oval_dir:
-                with open(oval_file_path, "r") as f:
-                    xml_content = f.read()
-            else:
-                xml_content = process_file_with_macros(oval_file_path, env_yaml)
+    def _read_oval_file(self, file_path, context, from_benchmark):
+        if from_benchmark or "checks_from_templates" not in file_path:
+            xml_content = process_file_with_macros(file_path, context)
+        else:
+            with open(file_path, "r") as f:
+                xml_content = f.read()
+        return xml_content
+
+    def _create_key(self, file_path, from_benchmark):
+        if from_benchmark:
+            rule_id = os.path.basename(
+                (os.path.dirname(os.path.dirname(file_path))))
+            oval_key = "%s.xml" % rule_id
+        else:
+            oval_key = os.path.basename(file_path)
+        return oval_key
+
+    def _process_oval_file(self, file_path, from_benchmark, context):
+        if not file_path.endswith(".xml"):
+            return None
+        oval_key = self._create_key(file_path, from_benchmark)
+        if _check_is_loaded(self.already_loaded, oval_key, self.oval_version):
+            return None
+        xml_content = self._read_oval_file(file_path, context, from_benchmark)
+        if not self._manage_oval_file_xml_content(
+                file_path, xml_content, from_benchmark):
+            return None
+        self.already_loaded[oval_key] = self.oval_version
+        return xml_content
+
+    def _manage_oval_file_xml_content(
+            self, file_path, xml_content, from_benchmark):
+        if not _check_is_applicable_for_product(xml_content, self.product):
+            return False
+        oval_file_tree = _create_oval_tree_from_string(xml_content)
+        if not _check_oval_version_from_oval(oval_file_tree, self.oval_version):
+            return False
+        if from_benchmark:
+            self._benchmark_specific_actions(
+                file_path, xml_content, oval_file_tree)
+        return True
 
-            if not _check_is_applicable_for_product(xml_content, product):
-                continue
-            if _check_is_loaded(already_loaded, filename, oval_version):
-                continue
-            oval_file_tree = _create_oval_tree_from_string(xml_content)
-            if not _check_oval_version_from_oval(oval_file_tree, oval_version):
-                continue
-            body.append(xml_content)
-            included_checks_count += 1
-            already_loaded[filename] = oval_version
+    def _benchmark_specific_actions(
+            self, file_path, xml_content, oval_file_tree):
+        rule_id = os.path.basename(
+            (os.path.dirname(os.path.dirname(file_path))))
+        self._store_intermediate_file(rule_id, xml_content)
+        if not _check_rule_id(oval_file_tree, rule_id):
+            msg = "OVAL definition in '%s' doesn't match rule ID '%s'." % (
+                file_path, rule_id)
+            print(msg, file=sys.stderr)
+
+    def _get_context(self, directory, from_benchmark):
+        if from_benchmark:
+            rule_path = os.path.join(directory, "rule.yml")
+            rule = Rule.from_yaml(rule_path, self.env_yaml)
+            context = self._create_local_env_yaml_for_rule(rule)
+        else:
+            context = self.env_yaml
+        return context
 
-    return "".join(body)
+    def _create_local_env_yaml_for_rule(self, rule):
+        local_env_yaml = dict()
+        local_env_yaml.update(self.env_yaml)
+        local_env_yaml['rule_id'] = rule.id_
+        local_env_yaml['rule_title'] = rule.title
+        prodtypes = parse_prodtype(rule.prodtype)
+        local_env_yaml['products'] = prodtypes  # default is all
+        return local_env_yaml
+
+    def _store_intermediate_file(self, rule_id, xml_content):
+        if not self.build_ovals_dir:
+            return
+        output_file_name = rule_id + ".xml"
+        output_filepath = os.path.join(self.build_ovals_dir, output_file_name)
+        with open(output_filepath, "w") as f:
+            f.write(xml_content)

tests/unit/ssg-module/test_build_ovals.py

@@ -0,0 +1,43 @@
+import os
+import pytest
+import tempfile
+import xml.etree.ElementTree as ET
+
+import ssg.build_ovals
+
+PROJECT_ROOT = os.path.join(os.path.dirname(__file__), "..", "..", "..", )
+DATADIR = os.path.abspath(
+    os.path.join(os.path.dirname(__file__), "test_build_ovals_data"))
+PRODUCT_YAML = os.path.join(DATADIR, "product.yml")
+SHARED_OVALS = os.path.join(DATADIR, "shared_ovals")
+BUILD_OVALS_DIR = tempfile.mkdtemp()
+
+shared_oval_1_def_tag = '<definition class="compliance" ' \
+    'id="tmux_conf_readable_by_others" version="1">'
+benchmark_oval_1_def_tag = '<definition class="compliance" ' \
+    'id="selinux_state" version="1">'
+
+
+def test_build_ovals():
+    env_yaml = {
+        "product": "rhel9",
+        "target_oval_version_str": "5.11",
+    }
+    obuilder = ssg.build_ovals.OVALBuilder(
+        env_yaml, PRODUCT_YAML, [SHARED_OVALS], BUILD_OVALS_DIR)
+    shorthand = obuilder.build_shorthand(include_benchmark=False)
+    assert shared_oval_1_def_tag in shorthand
+    assert benchmark_oval_1_def_tag not in shorthand
+
+
+def test_build_ovals_include_benchmark():
+    env_yaml = {
+        "benchmark_root": "./guide",
+        "product": "rhel9",
+        "target_oval_version_str": "5.11",
+    }
+    obuilder = ssg.build_ovals.OVALBuilder(
+        env_yaml, PRODUCT_YAML, [SHARED_OVALS], BUILD_OVALS_DIR)
+    shorthand = obuilder.build_shorthand(include_benchmark=True)
+    assert shared_oval_1_def_tag in shorthand
+    assert benchmark_oval_1_def_tag in shorthand

tests/unit/ssg-module/test_build_ovals_data/guide/selinux_state/oval/shared.xml

@@ -0,0 +1,27 @@
+<def-group>
+  <definition class="compliance" id="selinux_state" version="1">
+    {{{ oval_metadata("The SELinux state should be enforcing the local policy.") }}}
+    <criteria operator="AND">
+      <criterion comment="enforce is disabled" test_ref="test_etc_selinux_config" />
+    </criteria>
+  </definition>
+
+  <ind:textfilecontent54_test check="all" check_existence="all_exist"
+  comment="/selinux/enforce is 1" id="test_etc_selinux_config" version="1">
+    <ind:object object_ref="object_etc_selinux_config" />
+    <ind:state state_ref="state_etc_selinux_config" />
+  </ind:textfilecontent54_test>
+
+  <ind:textfilecontent54_object id="object_etc_selinux_config" version="1">
+    <ind:filepath>/etc/selinux/config</ind:filepath>
+    <ind:pattern operation="pattern match">^SELINUX=(.*)$</ind:pattern>
+    <ind:instance datatype="int">1</ind:instance>
+  </ind:textfilecontent54_object>
+
+  <ind:textfilecontent54_state id="state_etc_selinux_config" version="1">
+    <ind:subexpression datatype="string" operation="equals" var_check="all" var_ref="var_selinux_state" />
+  </ind:textfilecontent54_state>
+
+  <external_variable comment="external variable for selinux state"
+  datatype="string" id="var_selinux_state" version="1" />
+</def-group>