ssg.playbook_builder.PlaybookBuilder.choose_variable_value() - Code Metrics - Inspection of "Merge pull request #4429 from cipherboy/oh-no-docu..." - ComplianceAsCode/content - Measure and Improve Code Quality continuously with Scrutinizer

Passed

Push — master ( 2e0e1e...e323ea )

by Marek

created 2019-06-20 14:58 UTC

PlaybookBuilder.choose_variable_value() B

↳ Parent: ssg.playbook_builder

Complexity

Conditions

Size

Total Lines	30
Code Lines	19

Duplication

Lines	0
Ratio	0 %

Code Coverage

Tests	8
CRAP Score	14.2702

Importance

Changes

Metric	Value
cc	7
eloc	19
nop	4
dl	0
loc	30
ccs	8
cts	17
cp	0.4706
crap	14.2702
rs	8
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))

    def get_profile_selections(self, profile):
        """
        Provides a tuple (rules, variables) where rules is a list of rules
        selected in the profile and variables is a dictionary of
        variables and their values in the profile. The method can handle
        profiles which extend other profiles.
        """
        rules = []
        variables = dict()
        if profile.extends:
            extended_profile_path = os.path.join(
                self.profiles_dir, profile.extends + ".profile")
            try:
                extended_profile = ssg.build_yaml.Profile.from_yaml(
                    extended_profile_path)
            except ssg.yaml.DocumentationNotComplete:
                sys.stderr.write("Skipping incomplete profile %s.\n" % extended_profile_path)
                return None
            if not profile:
                sys.stderr.write(
                    "Could not parse profile %s.\n" % extended_profile_path)
                return None
            rules, variables = self.get_profile_selections(extended_profile)
        rules.extend(profile.get_rule_selectors())
        variables.update(profile.get_variable_selectors())
        return rules, variables

    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 dirpath, dirnames, filenames in os.walk(self.guide_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)

                    variables[xccdf_value.id_] = options
        return variables

    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.Profile.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_refines = self.get_profile_selections(profile)

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

ComplianceAsCode / content

Push — master ( 2e0e1e...e323ea )

PlaybookBuilder.choose_variable_value() B

Complexity

Size

Duplication

Code Coverage

Importance

Duplication Side-by-Side

Filter issues like