Merge pull request #11549 from Honny1/thin-ds-cli

Thin DS: Command Line Interface

CMakeLists.txt

@@ -13,6 +13,10 @@ if(SSG_LOG)
     set(LOG_LEVEL "DEBUG")
 endif()
 
+if(NOT SSG_THIN_DS)
+    set(SSG_THIN_DS_RULE_ID "off")
+endif()
+
 project(scap-security-guide NONE)
 
 list(APPEND CMAKE_MODULE_PATH "${PROJECT_SOURCE_DIR}/cmake")
@@ -275,6 +279,7 @@ message(STATUS "Separate SCAP files: ${SSG_SEPARATE_SCAP_FILES_ENABLED}")
 message(STATUS "Ansible Playbooks: ${SSG_ANSIBLE_PLAYBOOKS_ENABLED}")
 message(STATUS "Ansible Playbooks Per Rule: ${SSG_ANSIBLE_PLAYBOOKS_PER_RULE_ENABLED}")
 message(STATUS "Bash scripts: ${SSG_BASH_SCRIPTS_ENABLED}")
+message(STATUS "Thin data streams: ${SSG_THIN_DS}")
 if(SSG_JINJA2_CACHE_ENABLED)
     message(STATUS "jinja2 cache: enabled")
     message(STATUS "jinja2 cache dir: ${SSG_JINJA2_CACHE_DIR}")

build-scripts/build_xccdf.py

@@ -5,6 +5,8 @@
 import argparse
 import os
 import os.path
+from collections import namedtuple
+
 
 import ssg.build_yaml
 import ssg.utils
@@ -14,6 +16,9 @@
 import ssg.products
 
 
+Paths_ = namedtuple("Paths_", ["xccdf", "oval", "ocil", "build_ovals_dir"])
+
+
 def parse_args():
     parser = argparse.ArgumentParser(
         description="Converts SCAP Security Guide YAML benchmark data "
@@ -49,11 +54,75 @@ def parse_args():
         dest="build_ovals_dir",
         help="Directory to store OVAL document for each rule.",
     )
-    parser.add_argument("--resolved-base",
-                        help="To which directory to put processed rule/group/value YAMLs.")
+    parser.add_argument(
+        "--resolved-base",
+        help="To which directory to put processed rule/group/value YAMLs."
+    )
+    parser.add_argument(
+        "--thin-ds-components-dir",
+        help="Directory to store XCCDF, OVAL, OCIL, for thin data stream. (off: to disable)"
+        "e.g.: ~/scap-security-guide/build/rhel7/thin_ds_component/"
+        "Fake profiles are used to create thin DS. Components are generated for each profile.",
+    )
     return parser.parse_args()
 
 
+def link_oval(xccdftree, checks, output_file_name, build_ovals_dir):
+    translator = ssg.id_translate.IDTranslator("ssg")
+    oval_linker = ssg.build_renumber.OVALFileLinker(
+        translator, xccdftree, checks, output_file_name
+    )
+    oval_linker.build_ovals_dir = build_ovals_dir
+    oval_linker.link()
+    oval_linker.save_linked_tree()
+    oval_linker.link_xccdf()
+
+
+def link_ocil(xccdftree, checks, output_file_name, ocil):
+    translator = ssg.id_translate.IDTranslator("ssg")
+    ocil_linker = ssg.build_renumber.OCILFileLinker(
+        translator, xccdftree, checks, output_file_name
+    )
+    ocil_linker.link(ocil)
+    ocil_linker.save_linked_tree()
+    ocil_linker.link_xccdf()
+
+
+def link_benchmark(loader, xccdftree, args, benchmark=None):
+    checks = xccdftree.findall(".//{%s}check" % ssg.constants.XCCDF12_NS)
+
+    link_oval(xccdftree, checks, args.oval, args.build_ovals_dir)
+
+    ocil = loader.export_ocil_to_xml(benchmark)
+    if ocil is not None:
+        link_ocil(xccdftree, checks, args.ocil, ocil)
+
+    ssg.xml.ElementTree.ElementTree(xccdftree).write(args.xccdf)
+
+
+def get_path(path, file_name):
+    return os.path.join(path, file_name)
+
+
+def link_benchmark_per_profile(loader, args):
+    if not os.path.exists(args.thin_ds_components_dir):
+        os.makedirs(args.thin_ds_components_dir)
+
+    loader.off_ocil = True
+
+    for id_, benchmark in loader.get_benchmark_by_profile():
+        xccdftree = benchmark.to_xml_element(loader.env_yaml)
+        p = Paths_(
+            xccdf=get_path(args.thin_ds_components_dir, "xccdf_{}.xml".format(id_)),
+            oval=get_path(args.thin_ds_components_dir, "oval_{}.xml".format(id_)),
+            ocil=get_path(args.thin_ds_components_dir, "ocil_{}.xml".format(id_)),
+            build_ovals_dir=args.build_ovals_dir
+        )
+        link_benchmark(loader, xccdftree, p, benchmark)
+
+    loader.off_ocil = False
+
+
 def main():
     args = parse_args()
 
@@ -74,27 +143,12 @@ def main():
     loader.load_benchmark(benchmark_root)
 
     loader.add_fixes_to_rules()
-    xccdftree = loader.export_benchmark_to_xml()
-    ocil = loader.export_ocil_to_xml()
-
-    checks = xccdftree.findall(".//{%s}check" % ssg.constants.XCCDF12_NS)
 
-    translator = ssg.id_translate.IDTranslator("ssg")
+    if args.thin_ds_components_dir != "off":
+        link_benchmark_per_profile(loader, args)
 
-    oval_linker = ssg.build_renumber.OVALFileLinker(
-        translator, xccdftree, checks, args.oval)
-    oval_linker.build_ovals_dir = args.build_ovals_dir
-    oval_linker.link()
-    oval_linker.save_linked_tree()
-    oval_linker.link_xccdf()
-
-    ocil_linker = ssg.build_renumber.OCILFileLinker(
-        translator, xccdftree, checks, args.ocil)
-    ocil_linker.link(ocil)
-    ocil_linker.save_linked_tree()
-    ocil_linker.link_xccdf()
-
-    ssg.xml.ElementTree.ElementTree(xccdftree).write(args.xccdf)
+    xccdftree = loader.export_benchmark_to_xml()
+    link_benchmark(loader, xccdftree, args)
 
 
 if __name__ == "__main__":

build-scripts/compile_all.py

@@ -11,6 +11,8 @@
 import ssg.environment
 from ssg.build_cpe import ProductCPEs
 from ssg.constants import BENCHMARKS
+from ssg.entities.profile import ProfileWithInlinePolicies
+
 
 def create_parser():
     parser = argparse.ArgumentParser()
@@ -41,6 +43,14 @@ def create_parser():
     parser.add_argument(
         "--stig-references", help="DISA STIG Reference XCCDF file"
     )
+    parser.add_argument(
+        "--rule-id",
+        type=str,
+        help="Creates a profile with the specified rule and does not use other profiles."
+        "The profile ID is identical to the rule ID of the selected rule."
+        " If you want to process all rules in benchmark of the product, "
+        "you can use 'ALL_RULES'.",
+    )
     return parser
 
 
@@ -91,8 +101,12 @@ def get_all_resolved_profiles_by_id(
     return profiles_by_id
 
 
-def load_resolve_and_validate_profiles(env_yaml, profile_files, loader, controls_manager, product_cpes):
-    profiles_by_id = ssg.build_profile.make_name_to_profile_mapping(profile_files, env_yaml, product_cpes)
+def load_resolve_and_validate_profiles(
+        env_yaml, profile_files, loader, controls_manager, product_cpes
+        ):
+    profiles_by_id = ssg.build_profile.make_name_to_profile_mapping(
+        profile_files, env_yaml, product_cpes
+    )
 
     for p in profiles_by_id.values():
         p.resolve(profiles_by_id, loader.all_rules, controls_manager)
@@ -110,9 +124,11 @@ def save_everything(base_dir, loader, controls_manager, profiles):
         dump_compiled_profile(base_dir, p)
 
 
-def find_existing_rules(project_root):
+def find_existing_rules(project_root, relevant_benchmarks=None):
     rules = set()
-    for benchmark in BENCHMARKS:
+    if relevant_benchmarks is None:
+        relevant_benchmarks = BENCHMARKS
+    for benchmark in relevant_benchmarks:
         benchmark = os.path.join(project_root, benchmark)
         for dirpath, _, filenames in os.walk(benchmark):
             if "rule.yml" in filenames:
@@ -129,6 +145,40 @@ def add_stig_references(stig_reference_path, all_rules):
         rule.add_stig_references(stig_references)
 
 
+def get_relevant_benchmarks(env_yaml, product_yaml):
+    benchmark_paths = get_all_content_directories(env_yaml, product_yaml)
+    out = set()
+    for benchmark in BENCHMARKS:
+        for path in benchmark_paths:
+            if benchmark in os.path.normpath(path):
+                out.add(benchmark)
+    return out
+
+
+def get_minimal_profiles_by_id(rules, variables):
+    out = {}
+    for rule in rules:
+        data = {
+              'documentation_complete': True,
+              'variables': variables,
+              'selected': [rule],
+              'id_': rule,
+        }
+        profile = ProfileWithInlinePolicies.get_instance_from_full_dict(data)
+        out[profile.id_] = profile
+    return out
+
+
+def get_profiles_per_rule_by_id(rule_id, project_root_abspath, env_yaml, product_yaml):
+    relevant_benchmarks = get_relevant_benchmarks(env_yaml, product_yaml)
+    rules = find_existing_rules(project_root_abspath, relevant_benchmarks)
+    if rule_id not in rules and "ALL_RULES" not in rule_id:
+        raise Exception("Rule ID: {} not found!".format(rule_id))
+    if "ALL_RULES" not in rule_id:
+        rules = {rule_id}
+    return get_minimal_profiles_by_id(rules, {})
+
+
 def main():
     parser = create_parser()
     args = parser.parse_args()
@@ -168,8 +218,14 @@ def main():
 
     add_stig_references(args.stig_references, loader.all_rules.values())
 
-    profiles_by_id = get_all_resolved_profiles_by_id(
-        env_yaml, product_yaml, loader, product_cpes, controls_manager, controls_dir)
+    if args.rule_id is None or args.rule_id == "off":
+        profiles_by_id = get_all_resolved_profiles_by_id(
+            env_yaml, product_yaml, loader, product_cpes, controls_manager, controls_dir
+        )
+    else:
+        profiles_by_id = get_profiles_per_rule_by_id(
+            args.rule_id, project_root_abspath, env_yaml, product_yaml
+        )
 
     save_everything(
         args.resolved_base, loader, controls_manager, profiles_by_id.values())

build-scripts/compose_ds.py

@@ -4,6 +4,7 @@
 import os
 import sys
 import time
+import glob
 import xml.etree.ElementTree as ET
 
 from ssg.build_sce import collect_sce_checks
@@ -179,6 +180,13 @@ def parse_args():
     parser.add_argument(
         "--output-13", required=True,
         help="Output SCAP 1.3 source data stream file name")
+    parser.add_argument(
+        "--multiple-ds",
+        help="Directory where XCCDF, OVAL, OCIL files with lower case prefixes "
+        "xccdf, oval, ocil are stored to build multiple data streams. "
+        "Multiple streams are generated in the thin_ds subdirectory. (off: to disable) "
+        "e.g.: ~/scap-security-guide/build/rhel7/thin_ds_component/",
+    )
     return parser.parse_args()
 
 
@@ -195,6 +203,8 @@ def get_timestamp(file_name):
 def add_component(
         ds_collection, component_ref_parent, component_file_name,
         dependencies=None):
+    if not os.path.exists(component_file_name):
+        return
     component_id = "scap_%s_comp_%s" % (
         ID_NS, os.path.basename(component_file_name))
     component = ET.SubElement(
@@ -214,6 +224,9 @@ def add_component(
 
 
 def create_catalog(component_ref, dependencies):
+    dependencies = [dep for dep in dependencies if os.path.exists(dep)]
+    if len(dependencies) == 0:
+        return
     catalog = ET.SubElement(component_ref, "{%s}catalog" % cat_namespace)
     for dep in dependencies:
         uri = ET.SubElement(catalog, "{%s}uri" % cat_namespace)
@@ -250,6 +263,9 @@ def compose_ds(
         sce_check_files = collect_sce_checks(ds_collection)
         refdir = os.path.dirname(oval_file_name)
         embed_sce_checks_in_datastream(ds_collection, checklists, sce_check_files, refdir)
+
+    if hasattr(ET, "indent"):
+        ET.indent(ds_collection, space=" ", level=0)
     return ET.ElementTree(ds_collection)
 
 
@@ -264,12 +280,57 @@ def upgrade_ds_to_scap_13(ds):
     return ds
 
 
+def _store_ds(ds, output_13, output_12):
+    if output_12:
+        ds.write(output_12, xml_declaration=True)
+
+    ds_13 = upgrade_ds_to_scap_13(ds)
+    ds_13.write(output_13, xml_declaration=True)
+
+
+def append_id_to_file_name(path, id_):
+    return "{0}_{2}{1}".format(*os.path.splitext(path) + (id_,))
+
+
+def add_dir(path, dir):
+    return os.path.join(os.path.dirname(path), dir, os.path.basename(path))
+
+
+def _get_thin_ds_output_path(output, file_name):
+    return add_dir(
+        append_id_to_file_name(
+            output,
+            os.path.splitext(os.path.basename(file_name))[0]
+        ),
+        "thin_ds"
+    )
+
+
+def _compose_multiple_ds(args):
+
+    for xccdf in glob.glob("{}/xccdf*.xml".format(args.multiple_ds)):
+        oval = xccdf.replace("xccdf", "oval")
+        ocil = xccdf.replace("xccdf", "ocil")
+        ds = compose_ds(
+            xccdf, oval, ocil, args.cpe_dict, args.cpe_oval, args.enable_sce
+        )
+        output_13 = _get_thin_ds_output_path(args.output_13, xccdf.replace("xccdf_", ""))
+        output_12 = None
+
+        if not os.path.exists(os.path.dirname(output_13)):
+            os.makedirs(os.path.dirname(output_13))
+
+        _store_ds(ds, output_13, output_12)
+
+
 if __name__ == "__main__":
     args = parse_args()
     ssg.xml.register_namespaces()
+
+    if args.multiple_ds != "off":
+        _compose_multiple_ds(args)
+
     ds = compose_ds(
-        args.xccdf, args.oval, args.ocil, args.cpe_dict, args.cpe_oval, args.enable_sce)
-    if args.output_12:
-        ds.write(args.output_12)
-    ds_13 = upgrade_ds_to_scap_13(ds)
-    ds_13.write(args.output_13)
+        args.xccdf, args.oval, args.ocil, args.cpe_dict, args.cpe_oval, args.enable_sce
+    )
+    _store_ds(ds, args.output_13, args.output_12)