ssg.playbook_builder.PlaybookBuilder.get_benchmark_variables() - Code Metrics - Inspection of "Merge pull request #5927 from jan-cerny/issue5874" - ComplianceAsCode/content - Measure and Improve Code Quality continuously with Scrutinizer

Passed

Push — master ( 65b335...62b6dd )

by Matěj

created 2020-07-21 09:31 UTC

PlaybookBuilder.get_benchmark_variables() A

↳ Parent: ssg.playbook_builder

Complexity

Conditions

Size

Total Lines	11
Code Lines	5

Duplication

Lines	0
Ratio	0 %

Code Coverage

Tests	5
CRAP Score	2

Importance

Changes

Metric	Value
cc	2
eloc	5
nop	1
dl	0
loc	11
ccs	5
cts	5
cp	1
crap	2
rs	10
c	0
b	0
f	0

#!/usr/bin/python3

import os
import re
import sys

from collections import OrderedDict

import ssg.rules
import ssg.utils
import ssg.yaml
import ssg.build_yaml
import ssg.build_remediations

COMMENTS_TO_PARSE = ["strategy", "complexity", "disruption"]


class PlaybookBuilder():
    def __init__(self, product_yaml_path, input_dir, output_dir, rules_dir):
        self.input_dir = input_dir
        self.output_dir = output_dir
        self.rules_dir = rules_dir
        product_yaml = ssg.yaml.open_raw(product_yaml_path)
        product_dir = os.path.dirname(product_yaml_path)
        relative_guide_dir = ssg.utils.required_key(product_yaml,
                                                    "benchmark_root")
        self.guide_dir = os.path.abspath(os.path.join(product_dir,
                                                      relative_guide_dir))
        relative_profiles_dir = ssg.utils.required_key(product_yaml,
                                                       "profiles_root")
        self.profiles_dir = os.path.abspath(
            os.path.join(product_dir, relative_profiles_dir))
        additional_content_directories = product_yaml.get("additional_content_directories", [])
        self.add_content_dirs = [os.path.abspath(os.path.join(product_dir, rd)) for rd in additional_content_directories]

    def choose_variable_value(self, var_id, variables, refinements):
        """
        Determine value of variable based on profile refinements.
        """
        if refinements and var_id in refinements:
            selector = refinements[var_id]
        else:
            selector = "default"
        try:
            options = variables[var_id]
        except KeyError:
            raise ValueError("Variable '%s' doesn't exist." % var_id)
        try:
            value = options[selector]
        except KeyError:
            if len(options.keys()) == 1:
                # We will assume that if there is just one option that it could
                # be selected as a default option.
                value = list(options.values())[0]
            elif selector == "default":
                # If no 'default' selector is present in the XCCDF value,
                # choose the first (consistent with oscap behavior)
                value = list(options.values())[0]
            else:
                raise ValueError(
                    "Selector '%s' doesn't exist in variable '%s'. "
                    "Available selectors: %s." %
                    (selector, var_id, ", ".join(options.keys()))
                )
        return value

    def get_data_from_snippet(self, snippet_yaml, variables, refinements):
        """
        Extracts and resolves tasks and variables from Ansible snippet.
        """
        xccdf_var_pattern = re.compile(r"\(xccdf-var\s+(\S+)\)")
        tasks = []
        values = dict()
        for item in snippet_yaml:
            if isinstance(item, str):
                match = xccdf_var_pattern.match(item)
                if match:
                    var_id = match.group(1)
                    value = self.choose_variable_value(var_id, variables,
                                                       refinements)
                    values[var_id] = value
                else:
                    raise ValueError("Found unknown item '%s'" % item)
            else:
                tasks.append(item)
        return tasks, values

    def get_benchmark_variables(self):
        """
        Get all variables, their selectors and values used in a given
        benchmark. Returns a dictionary where keys are variable IDs and
        values are dictionaries where keys are selectors and values are
        variable values.
        """
        variables = dict()
        for cur_dir in [self.guide_dir] + self.add_content_dirs:
            variables.update(self._get_rules_variables(cur_dir))
        return variables

    def _get_rules_variables(self, base_dir):
        for dirpath, dirnames, filenames in os.walk(base_dir):
            for filename in filenames:
                root, ext = os.path.splitext(filename)
                if ext == ".var":
                    full_path = os.path.join(dirpath, filename)
                    xccdf_value = ssg.build_yaml.Value.from_yaml(full_path)
                    # Make sure that selectors and values are strings
                    options = dict()
                    for k, v in xccdf_value.options.items():
                        options[str(k)] = str(v)

                    yield (xccdf_value.id_, options,)

    def _find_rule_title(self, rule_id):
        rule_path = os.path.join(self.rules_dir, rule_id + ".yml")
        rule_yaml = ssg.yaml.open_raw(rule_path)
        return rule_yaml["title"]

    def create_playbook(self, snippet_path, rule_id, variables,
                        refinements, output_dir):
        """
        Creates a Playbook from Ansible snippet for the given rule specified
        by rule ID, fills in the profile values and saves it into output_dir.
        """

        with open(snippet_path, "r") as snippet_file:
            snippet_str = snippet_file.read()
        fix = ssg.build_remediations.split_remediation_content_and_metadata(snippet_str)
        snippet_yaml = ssg.yaml.ordered_load(fix.contents)

        play_tasks, play_vars = self.get_data_from_snippet(
            snippet_yaml, variables, refinements
        )

        if len(play_tasks) == 0:
            raise ValueError(
                "Ansible remediation for rule '%s' in '%s' "
                "doesn't contain any task." %
                (rule_id, snippet_path)
            )

        tags = set()
        for task in play_tasks:
            tags |= set(task.pop("tags", []))

        play = OrderedDict()
        play["name"] = self._find_rule_title(rule_id)
        play["hosts"] = "@@HOSTS@@"
        play["become"] = True
        if len(play_vars) > 0:
            play["vars"] = play_vars
        if len(tags) > 0:
            play["tags"] = sorted(list(tags))
        play["tasks"] = play_tasks

        playbook = [play]
        playbook_path = os.path.join(output_dir, rule_id + ".yml")
        with open(playbook_path, "w") as playbook_file:
            # write remediation metadata (complexity, strategy, etc.) first
            for k, v in fix.config.items():
                playbook_file.write("# %s = %s\n" % (k, v))
            ssg.yaml.ordered_dump(
                playbook, playbook_file, default_flow_style=False
            )

    def open_profile(self, profile_path):
        """
        Opens and parses profile at the given profile_path.
        """
        if not os.path.isfile(profile_path):
            raise RuntimeError("'%s' is not a file!\n" % profile_path)
        profile_id, ext = os.path.splitext(os.path.basename(profile_path))
        if ext != ".profile":
            raise RuntimeError(
                "Encountered file '%s' while looking for profiles, "
                "extension '%s' is unknown. Skipping..\n"
                % (profile_path, ext)
            )

        profile = ssg.build_yaml.ResolvableProfile.from_yaml(profile_path)
        if not profile:
            raise RuntimeError("Could not parse profile %s.\n" % profile_path)
        return profile

    def create_playbooks_for_all_rules_in_profile(self, profile, variables):
        """
        Creates a Playbook for each rule selected in a profile from tasks
        extracted from snippets. Created Playbooks are parametrized by
        variables according to profile selection. Playbooks are written into
        a new subdirectory in output_dir.
        """
        profile_rules = profile.get_rule_selectors()
        profile_refines = profile.get_variable_selectors()

        profile_playbooks_dir = os.path.join(self.output_dir, profile.id_)
        os.makedirs(profile_playbooks_dir)

        for rule_id in profile_rules:
            snippet_path = os.path.join(self.input_dir, rule_id + ".yml")
            if os.path.exists(snippet_path):
                self.create_playbook(
                    snippet_path, rule_id, variables,
                    profile_refines, profile_playbooks_dir
                )

    def create_playbook_for_single_rule(self, profile, rule_id, variables):
        """
        Creates a Playbook for given rule specified by a rule_id. Created
        Playbooks are parametrized by variables according to profile selection.
        Playbooks are written into a new subdirectory in output_dir.
        """
        profile_rules = profile.get_rule_selectors()
        profile_refines = profile.get_variable_selectors()
        profile_playbooks_dir = os.path.join(self.output_dir, profile.id_)
        os.makedirs(profile_playbooks_dir)
        snippet_path = os.path.join(self.input_dir, rule_id + ".yml")
        if rule_id in profile_rules:
            self.create_playbook(
                snippet_path, rule_id, variables,
                profile_refines, profile_playbooks_dir
            )
        else:
            raise ValueError("Rule '%s' isn't part of profile '%s'" %
                             (rule_id, profile.id_))

    def create_playbooks_for_all_rules(self, variables):
        profile_playbooks_dir = os.path.join(self.output_dir, "all")
        os.makedirs(profile_playbooks_dir)
        for rule in os.listdir(self.rules_dir):
            rule_id, _ = os.path.splitext(rule)
            snippet_path = os.path.join(self.input_dir, rule_id + ".yml")
            if not os.path.exists(snippet_path):
                continue
            self.create_playbook(
                snippet_path, rule_id, variables,
                None, profile_playbooks_dir
            )

    def build(self, profile_id=None, rule_id=None):
        """
        Creates Playbooks for a specified profile.
        If profile is not given, creates playbooks for all profiles
        in the product.
        If the rule_id is not given, Playbooks are created for every rule.
        """
        variables = self.get_benchmark_variables()
        profiles = {}
        for profile_file in os.listdir(self.profiles_dir):
            profile_path = os.path.join(self.profiles_dir, profile_file)
            try:
                profile = self.open_profile(profile_path)
            except ssg.yaml.DocumentationNotComplete as e:
                msg = "Skipping incomplete profile {0}. To include incomplete " + \
                    "profiles, build in debug mode.\n"
                sys.stderr.write(msg.format(profile_path))
                continue
            except RuntimeError as e:
                sys.stderr.write(str(e))
                continue
            profiles[profile.id_] = profile

        if profile_id:
            to_process = [profile_id]
        else:
            to_process = list(profiles.keys())

        for p in to_process:
            profile = profiles[p]
            profile.resolve(profiles)
            if rule_id:
                self.create_playbook_for_single_rule(profile, rule_id,
                                                     variables)
            else:
                self.create_playbooks_for_all_rules_in_profile(
                    profile, variables)

        if not profile_id:
            # build playbooks for virtual '(all)' profile
            # this virtual profile contains all rules in the product
            self.create_playbooks_for_all_rules(variables)


1		#!/usr/bin/python3
2
3	2	import os
4	2	import re
5	2	import sys
6
7	2	from collections import OrderedDict
8
9	2	import ssg.rules
10	2	import ssg.utils
11	2	import ssg.yaml
12	2	import ssg.build_yaml
13	2	import ssg.build_remediations
14
15	2	COMMENTS_TO_PARSE = ["strategy", "complexity", "disruption"]
16
17
18	2	class PlaybookBuilder():
19	2	def __init__(self, product_yaml_path, input_dir, output_dir, rules_dir):
20	2	self.input_dir = input_dir
21	2	self.output_dir = output_dir
22	2	self.rules_dir = rules_dir
23	2	product_yaml = ssg.yaml.open_raw(product_yaml_path)
24	2	product_dir = os.path.dirname(product_yaml_path)
25	2	relative_guide_dir = ssg.utils.required_key(product_yaml,
26		"benchmark_root")
27	2	self.guide_dir = os.path.abspath(os.path.join(product_dir,
28		relative_guide_dir))
29	2	relative_profiles_dir = ssg.utils.required_key(product_yaml,
30		"profiles_root")
31	2	self.profiles_dir = os.path.abspath(
32		os.path.join(product_dir, relative_profiles_dir))
33	2	additional_content_directories = product_yaml.get("additional_content_directories", [])
34	2	self.add_content_dirs = [os.path.abspath(os.path.join(product_dir, rd)) for rd in additional_content_directories]
35
36	2	def choose_variable_value(self, var_id, variables, refinements):
37		"""
38		Determine value of variable based on profile refinements.
39		"""
40	2	if refinements and var_id in refinements:
41	2	selector = refinements[var_id]
42		else:
43		selector = "default"
44	2	try:
45	2	options = variables[var_id]
46		except KeyError:
47		raise ValueError("Variable '%s' doesn't exist." % var_id)
48	2	try:
49	2	value = options[selector]
50		except KeyError:
51		if len(options.keys()) == 1:
52		# We will assume that if there is just one option that it could
53		# be selected as a default option.
54		value = list(options.values())[0]
55		elif selector == "default":
56		# If no 'default' selector is present in the XCCDF value,
57		# choose the first (consistent with oscap behavior)
58		value = list(options.values())[0]
59		else:
60		raise ValueError(
61		"Selector '%s' doesn't exist in variable '%s'. "
62		"Available selectors: %s." %
63		(selector, var_id, ", ".join(options.keys()))
64		)
65	2	return value
66
67	2	def get_data_from_snippet(self, snippet_yaml, variables, refinements):
68		"""
69		Extracts and resolves tasks and variables from Ansible snippet.
70		"""
71	2	xccdf_var_pattern = re.compile(r"\(xccdf-var\s+(\S+)\)")
72	2	tasks = []
73	2	values = dict()
74	2	for item in snippet_yaml:
75	2	if isinstance(item, str):
76	2	match = xccdf_var_pattern.match(item)
77	2	if match:
78	2	var_id = match.group(1)
79	2	value = self.choose_variable_value(var_id, variables,
80		refinements)
81	2	values[var_id] = value
82		else:
83		raise ValueError("Found unknown item '%s'" % item)
84		else:
85	2	tasks.append(item)
86	2	return tasks, values
87
88	2	def get_benchmark_variables(self):
89		"""
90		Get all variables, their selectors and values used in a given
91		benchmark. Returns a dictionary where keys are variable IDs and
92		values are dictionaries where keys are selectors and values are
93		variable values.
94		"""
95	2	variables = dict()
96	2	for cur_dir in [self.guide_dir] + self.add_content_dirs:
97	2	variables.update(self._get_rules_variables(cur_dir))
98	2	return variables
99
100	2	def _get_rules_variables(self, base_dir):
101	2	for dirpath, dirnames, filenames in os.walk(base_dir):
102	2	for filename in filenames:
103	2	root, ext = os.path.splitext(filename)
104	2	if ext == ".var":
105	2	full_path = os.path.join(dirpath, filename)
106	2	xccdf_value = ssg.build_yaml.Value.from_yaml(full_path)
107		# Make sure that selectors and values are strings
108	2	options = dict()
109	2	for k, v in xccdf_value.options.items():
110	2	options[str(k)] = str(v)
		0 ignored issues – show Comprehensibility Best Practice introduced 2020-03-27 21:24 UTC by Report Bug Copy Issue Report The variable `str` does not seem to be defined. Loading history...
111	2	yield (xccdf_value.id_, options,)
112
113	2	def _find_rule_title(self, rule_id):
114	2	rule_path = os.path.join(self.rules_dir, rule_id + ".yml")
115	2	rule_yaml = ssg.yaml.open_raw(rule_path)
116	2	return rule_yaml["title"]
117
118	2	def create_playbook(self, snippet_path, rule_id, variables,
119		refinements, output_dir):
120		"""
121		Creates a Playbook from Ansible snippet for the given rule specified
122		by rule ID, fills in the profile values and saves it into output_dir.
123		"""
124
125	2	with open(snippet_path, "r") as snippet_file:
126	2	snippet_str = snippet_file.read()
127	2	fix = ssg.build_remediations.split_remediation_content_and_metadata(snippet_str)
128	2	snippet_yaml = ssg.yaml.ordered_load(fix.contents)
129
130	2	play_tasks, play_vars = self.get_data_from_snippet(
131		snippet_yaml, variables, refinements
132		)
133
134	2	if len(play_tasks) == 0:
135		raise ValueError(
136		"Ansible remediation for rule '%s' in '%s' "
137		"doesn't contain any task." %
138		(rule_id, snippet_path)
139		)
140
141	2	tags = set()
142	2	for task in play_tasks:
143	2	tags \|= set(task.pop("tags", []))
144
145	2	play = OrderedDict()
146	2	play["name"] = self._find_rule_title(rule_id)
147	2	play["hosts"] = "@@HOSTS@@"
148	2	play["become"] = True
149	2	if len(play_vars) > 0:
150	2	play["vars"] = play_vars
151	2	if len(tags) > 0:
152	2	play["tags"] = sorted(list(tags))
153	2	play["tasks"] = play_tasks
154
155	2	playbook = [play]
156	2	playbook_path = os.path.join(output_dir, rule_id + ".yml")
157	2	with open(playbook_path, "w") as playbook_file:
158		# write remediation metadata (complexity, strategy, etc.) first
159	2	for k, v in fix.config.items():
160	2	playbook_file.write("# %s = %s\n" % (k, v))
161	2	ssg.yaml.ordered_dump(
162		playbook, playbook_file, default_flow_style=False
163		)
164
165	2	def open_profile(self, profile_path):
166		"""
167		Opens and parses profile at the given profile_path.
168		"""
169	2	if not os.path.isfile(profile_path):
170		raise RuntimeError("'%s' is not a file!\n" % profile_path)
171	2	profile_id, ext = os.path.splitext(os.path.basename(profile_path))
172	2	if ext != ".profile":
173		raise RuntimeError(
174		"Encountered file '%s' while looking for profiles, "
175		"extension '%s' is unknown. Skipping..\n"
176		% (profile_path, ext)
177		)
178
179	2	profile = ssg.build_yaml.ResolvableProfile.from_yaml(profile_path)
180	2	if not profile:
181		raise RuntimeError("Could not parse profile %s.\n" % profile_path)
182	2	return profile
183
184	2	def create_playbooks_for_all_rules_in_profile(self, profile, variables):
185		"""
186		Creates a Playbook for each rule selected in a profile from tasks
187		extracted from snippets. Created Playbooks are parametrized by
188		variables according to profile selection. Playbooks are written into
189		a new subdirectory in output_dir.
190		"""
191		profile_rules = profile.get_rule_selectors()
192		profile_refines = profile.get_variable_selectors()
193
194		profile_playbooks_dir = os.path.join(self.output_dir, profile.id_)
195		os.makedirs(profile_playbooks_dir)
196
197		for rule_id in profile_rules:
198		snippet_path = os.path.join(self.input_dir, rule_id + ".yml")
199		if os.path.exists(snippet_path):
200		self.create_playbook(
201		snippet_path, rule_id, variables,
202		profile_refines, profile_playbooks_dir
203		)
204
205	2	def create_playbook_for_single_rule(self, profile, rule_id, variables):
206		"""
207		Creates a Playbook for given rule specified by a rule_id. Created
208		Playbooks are parametrized by variables according to profile selection.
209		Playbooks are written into a new subdirectory in output_dir.
210		"""
211	2	profile_rules = profile.get_rule_selectors()
212	2	profile_refines = profile.get_variable_selectors()
213	2	profile_playbooks_dir = os.path.join(self.output_dir, profile.id_)
214	2	os.makedirs(profile_playbooks_dir)
215	2	snippet_path = os.path.join(self.input_dir, rule_id + ".yml")
216	2	if rule_id in profile_rules:
217	2	self.create_playbook(
218		snippet_path, rule_id, variables,
219		profile_refines, profile_playbooks_dir
220		)
221		else:
222		raise ValueError("Rule '%s' isn't part of profile '%s'" %
223		(rule_id, profile.id_))
224
225	2	def create_playbooks_for_all_rules(self, variables):
226		profile_playbooks_dir = os.path.join(self.output_dir, "all")
227		os.makedirs(profile_playbooks_dir)
228		for rule in os.listdir(self.rules_dir):
229		rule_id, _ = os.path.splitext(rule)
230		snippet_path = os.path.join(self.input_dir, rule_id + ".yml")
231		if not os.path.exists(snippet_path):
232		continue
233		self.create_playbook(
234		snippet_path, rule_id, variables,
235		None, profile_playbooks_dir
236		)
237
238	2	def build(self, profile_id=None, rule_id=None):
239		"""
240		Creates Playbooks for a specified profile.
241		If profile is not given, creates playbooks for all profiles
242		in the product.
243		If the rule_id is not given, Playbooks are created for every rule.
244		"""
245	2	variables = self.get_benchmark_variables()
246	2	profiles = {}
247	2	for profile_file in os.listdir(self.profiles_dir):
248	2	profile_path = os.path.join(self.profiles_dir, profile_file)
249	2	try:
250	2	profile = self.open_profile(profile_path)
251		except ssg.yaml.DocumentationNotComplete as e:
252		msg = "Skipping incomplete profile {0}. To include incomplete " + \
253		"profiles, build in debug mode.\n"
254		sys.stderr.write(msg.format(profile_path))
255		continue
256		except RuntimeError as e:
257		sys.stderr.write(str(e))
258		continue
259	2	profiles[profile.id_] = profile
260
261	2	if profile_id:
262	2	to_process = [profile_id]
263		else:
264		to_process = list(profiles.keys())
265
266	2	for p in to_process:
267	2	profile = profiles[p]
268	2	profile.resolve(profiles)
269	2	if rule_id:
270	2	self.create_playbook_for_single_rule(profile, rule_id,
271		variables)
272		else:
273		self.create_playbooks_for_all_rules_in_profile(
274		profile, variables)
275
276	2	if not profile_id:
277		# build playbooks for virtual '(all)' profile
278		# this virtual profile contains all rules in the product
279		self.create_playbooks_for_all_rules(variables)
280

ComplianceAsCode / content

Push — master ( 65b335...62b6dd )

PlaybookBuilder.get_benchmark_variables() A

Complexity

Size

Duplication

Code Coverage

Importance

Duplication Side-by-Side

Filter issues like