ssg.playbook_builder.PlaybookBuilder.__init__() - Code Metrics - Inspection of "Merge pull request #4936 from evgenyz/extra_rules_..." - ComplianceAsCode/content - Measure and Improve Code Quality continuously with Scrutinizer

Test Failed

Push — master ( 738e3d...d883b3 )

by Matěj

created 2019-10-31 12:41 UTC

ssg.playbook_builder.PlaybookBuilder.init() A

↳ Parent: ssg.playbook_builder

Complexity

Conditions

Size

Total Lines	16
Code Lines	16

Duplication

Lines	0
Ratio	0 %

Code Coverage

Tests	12
CRAP Score	1

Importance

Changes

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

ComplianceAsCode / content

Push — master ( 738e3d...d883b3 )

ssg.playbook_builder.PlaybookBuilder.__init__() A

Complexity

Size

Duplication

Code Coverage

Importance

Duplication Side-by-Side

Filter issues like

ssg.playbook_builder.PlaybookBuilder.init() A