ssg.controls.Status.from_control_info() - Code Metrics - Inspection of "Merge pull request #7518 from matejak/python2_fix_..." - ComplianceAsCode/content - Measure and Improve Code Quality continuously with Scrutinizer

Test Failed

Push — master ( ad184d...ba033d )

by Milan

created 2021-09-08 09:48 UTC

ssg.controls.Status.from_control_info() A

↳ Parent: ssg.controls

Complexity

Conditions

Size

Total Lines	24
Code Lines	20

Duplication

Lines	0
Ratio	0 %

Code Coverage

Tests	3
CRAP Score	4.679

Importance

Changes

Metric	Value
cc	3
eloc	20
nop	3
dl	0
loc	24
ccs	3
cts	7
cp	0.4286
crap	4.679
rs	9.4
c	0
b	0
f	0

import collections
import logging
import os
import copy
from glob import glob

import ssg.build_yaml
import ssg.yaml
import ssg.utils

from ssg.rules import get_rule_path_by_id


class InvalidStatus(Exception):
    pass

class Status():
    def __init__(self, status):
        self.status = status

    @classmethod
    def from_control_info(cls, ctrl, status):
        if status is None:
            return "pending"

        valid_statuses = [
            "pending",
            "not applicable",
            "inherently met",
            "documentation",
            "planned",
            "partial",
            "supported",
            "automated",
        ]

        if status not in valid_statuses:
            raise InvalidStatus(
                    "The given status '{given}' in the control '{control}' "
                    "was invalid. Please use one of "
                    "the following: {valid}".format(given=status,
                                                    control=ctrl,
                                                    valid=valid_statuses))
        return status

    def __str__(self):
        return self.status

    def __eq__(self, other):
        if isinstance(other, Status):
            return self.status == other.status
        elif isinstance(other, str):
            return self.status == other
        return False


class Control():
    def __init__(self):
        self.id = None
        self.rules = []
        self.variables = {}
        self.levels = []
        self.notes = ""
        self.title = ""
        self.description = ""
        self.automated = ""
        self.status = None

    @classmethod
    def from_control_dict(cls, control_dict, env_yaml=None, default_level=["default"]):
        control = cls()
        control.id = ssg.utils.required_key(control_dict, "id")
        control.title = control_dict.get("title")
        control.description = control_dict.get("description")
        control.status = Status.from_control_info(control.id, control_dict.get("status", None))
        control.automated = control_dict.get("automated", "no")
        if control.status == "automated":
            control.automated = "yes"
        if control.automated not in ["yes", "no", "partially"]:
            msg = (
                "Invalid value '%s' of automated key in control "
                "%s '%s'. Can be only 'yes', 'no', 'partially'."
                % (control.automated,  control.id, control.title))
            raise ValueError(msg)
        control.levels = control_dict.get("levels", default_level)
        control.notes = control_dict.get("notes", "")
        selections = control_dict.get("rules", [])

        product = None
        product_dir = None
        benchmark_root = None
        if env_yaml:
            product = env_yaml.get('product', None)
            product_dir = env_yaml.get('product_dir', None)
            benchmark_root = env_yaml.get('benchmark_root', None)
            content_dir = os.path.join(product_dir, benchmark_root)

        for item in selections:
            if "=" in item:
                varname, value = item.split("=", 1)
                control.variables[varname] = value
            else:
                # Check if rule is applicable to product, i.e.: prodtype has product id
                if product is None:
                    # The product was not specified, simply add the rule
                    control.rules.append(item)
                else:
                    rule_yaml = get_rule_path_by_id(content_dir, item)

                    if rule_yaml is None:
                        # item not found in benchmark_root
                        continue
                    rule = ssg.build_yaml.Rule.from_yaml(rule_yaml, env_yaml)
                    if rule.prodtype == "all" or product in rule.prodtype:
                        control.rules.append(item)
                    else:
                        logging.info("Rule {item} doesn't apply to {product}".format(
                            item=item,
                            product=product))

        control.related_rules = control_dict.get("related_rules", [])
        control.note = control_dict.get("note")
        return control


class Level():
    def __init__(self):
        self.id = None
        self.inherits_from = None

    @classmethod
    def from_level_dict(cls, level_dict):
        level = cls()
        level.id = ssg.utils.required_key(level_dict, "id")
        level.inherits_from = level_dict.get("inherits_from")
        return level

class Policy():
    def __init__(self, filepath, env_yaml=None):
        self.id = None
        self.env_yaml = env_yaml
        self.filepath = filepath
        self.controls = []
        self.controls_by_id = dict()
        self.levels = []
        self.levels_by_id = dict()
        self.title = ""
        self.source = ""

    def _parse_controls_tree(self, tree):
        default_level = ["default"]
        if self.levels:
            default_level = [self.levels[0].id]

        for node in tree:
            try:
                control = Control.from_control_dict(
                    node, self.env_yaml, default_level=default_level)
            except Exception as exc:
                msg = (
                    "Unable to parse controls from {filename}: {error}"
                    .format(filename=self.filepath, error=str(exc)))
                raise RuntimeError(msg)
            if "controls" in node:
                for sc in self._parse_controls_tree(node["controls"]):
                    yield sc
                    control.rules.extend(sc.rules)
                    control.variables.update(sc.variables)
                    control.related_rules.extend(sc.related_rules)
            yield control

    def load(self):
        yaml_contents = ssg.yaml.open_and_expand(self.filepath, self.env_yaml)
        self.id = ssg.utils.required_key(yaml_contents, "id")
        self.title = ssg.utils.required_key(yaml_contents, "title")
        self.source = yaml_contents.get("source", "")

        level_list = yaml_contents.get("levels", [])
        for lv in level_list:
            level = Level.from_level_dict(lv)
            self.levels.append(level)
            self.levels_by_id[level.id] = level

        controls_tree = ssg.utils.required_key(yaml_contents, "controls")
        for c in self._parse_controls_tree(controls_tree):
            self.controls.append(c)
            self.controls_by_id[c.id] = c

    def get_control(self, control_id):
        try:
            c = self.controls_by_id[control_id]
            return c
        except KeyError:
            msg = "%s not found in policy %s" % (
                control_id, self.id
            )
            raise ValueError(msg)

    def get_level(self, level_id):
        try:
            lv = self.levels_by_id[level_id]
            return lv
        except KeyError:
            msg = "Level %s not found in policy %s" % (
                level_id, self.id
            )
            raise ValueError(msg)

    def get_level_with_ancestors_sequence(self, level_id):
        # use OrderedDict for Python2 compatibility instead of ordered set
        levels = collections.OrderedDict()
        level = self.get_level(level_id)
        levels[level] = ""
        if level.inherits_from:
            for lv in level.inherits_from:
                eligible_levels = [l for l in self.get_level_with_ancestors_sequence(lv) if l not in levels.keys()]
                for l in eligible_levels:
                    levels[l] = ""
        return list(levels.keys())


class ControlsManager():
    def __init__(self, controls_dir, env_yaml=None):
        self.controls_dir = os.path.abspath(controls_dir)
        self.env_yaml = env_yaml
        self.policies = {}

    def load(self):
        if not os.path.exists(self.controls_dir):
            return
        for filename in sorted(glob(os.path.join(self.controls_dir, "*.yml"))):
            logging.info("Found file %s" % (filename))
            filepath = os.path.join(self.controls_dir, filename)
            policy = Policy(filepath, self.env_yaml)
            policy.load()
            self.policies[policy.id] = policy

    def get_control(self, policy_id, control_id):
        try:
            policy = self.policies[policy_id]
        except KeyError:
            msg = "policy '%s' doesn't exist" % (policy_id)
            raise ValueError(msg)
        control = policy.get_control(control_id)
        return control

    def _get_policy(self, policy_id):
        try:
            policy = self.policies[policy_id]
        except KeyError:
            msg = "policy '%s' doesn't exist" % (policy_id)
            raise ValueError(msg)
        return policy

    def get_all_controls_of_level(self, policy_id, level_id):
        policy = self._get_policy(policy_id)
        levels = policy.get_level_with_ancestors_sequence(level_id)
        all_policy_controls = self.get_all_controls(policy_id)
        eligible_controls = []
        already_defined_variables = set()
        # we will go level by level, from top to bottom
        # this is done to enable overriding of variables by higher levels
        for lv in levels:
            for control in all_policy_controls:
                if lv.id not in control.levels:
                    continue

                variables = set(control.variables.keys())

                variables_to_remove = variables.intersection(already_defined_variables)
                already_defined_variables.update(variables)

                new_c = self._get_control_without_variables(variables_to_remove, control)
                eligible_controls.append(new_c)

        return eligible_controls

    @staticmethod
    def _get_control_without_variables(variables_to_remove, control):
        if not variables_to_remove:
            return control

        new_c = copy.deepcopy(control)
        for var in variables_to_remove:
            del new_c.variables[var]
        return new_c

    def get_all_controls(self, policy_id):
        policy = self._get_policy(policy_id)
        return policy.controls_by_id.values()


1	2	import collections
2	2	import logging
3	2	import os
4	2	import copy
5	2	from glob import glob
6
7	2	import ssg.build_yaml
8	2	import ssg.yaml
9	2	import ssg.utils
10
11	2	from ssg.rules import get_rule_path_by_id
12
13
14	2	class InvalidStatus(Exception):
15	2	pass
16
17	2	class Status():
18	2	def __init__(self, status):
19		self.status = status
20
21	2	@classmethod
22		def from_control_info(cls, ctrl, status):
23	2	if status is None:
24	2	return "pending"
25
26		valid_statuses = [
27		"pending",
28		"not applicable",
29		"inherently met",
30		"documentation",
31		"planned",
32		"partial",
33		"supported",
34		"automated",
35		]
36
37		if status not in valid_statuses:
38		raise InvalidStatus(
39		"The given status '{given}' in the control '{control}' "
40		"was invalid. Please use one of "
41		"the following: {valid}".format(given=status,
42		control=ctrl,
43		valid=valid_statuses))
44		return status
45
46	2	def __str__(self):
47		return self.status
48
49	2	def __eq__(self, other):
50		if isinstance(other, Status):
51		return self.status == other.status
52		elif isinstance(other, str):
53		return self.status == other
54		return False
55
56
57	2	class Control():
58	2	def __init__(self):
59	2	self.id = None
60	2	self.rules = []
61	2	self.variables = {}
62	2	self.levels = []
63	2	self.notes = ""
64	2	self.title = ""
65	2	self.description = ""
66	2	self.automated = ""
67	2	self.status = None
68
69	2	@classmethod
70	2	def from_control_dict(cls, control_dict, env_yaml=None, default_level=["default"]):
71	2	control = cls()
72	2	control.id = ssg.utils.required_key(control_dict, "id")
73	2	control.title = control_dict.get("title")
74	2	control.description = control_dict.get("description")
75	2	control.status = Status.from_control_info(control.id, control_dict.get("status", None))
76	2	control.automated = control_dict.get("automated", "no")
77	2	if control.status == "automated":
78		control.automated = "yes"
79	2	if control.automated not in ["yes", "no", "partially"]:
80		msg = (
81		"Invalid value '%s' of automated key in control "
82		"%s '%s'. Can be only 'yes', 'no', 'partially'."
83		% (control.automated, control.id, control.title))
84		raise ValueError(msg)
85	2	control.levels = control_dict.get("levels", default_level)
86	2	control.notes = control_dict.get("notes", "")
87	2	selections = control_dict.get("rules", [])
88
89	2	product = None
90	2	product_dir = None
91	2	benchmark_root = None
92	2	if env_yaml:
93		product = env_yaml.get('product', None)
94		product_dir = env_yaml.get('product_dir', None)
95		benchmark_root = env_yaml.get('benchmark_root', None)
96		content_dir = os.path.join(product_dir, benchmark_root)
97
98	2	for item in selections:
99	2	if "=" in item:
100	2	varname, value = item.split("=", 1)
101	2	control.variables[varname] = value
102		else:
103		# Check if rule is applicable to product, i.e.: prodtype has product id
104	2	if product is None:
105		# The product was not specified, simply add the rule
106	2	control.rules.append(item)
107		else:
108		rule_yaml = get_rule_path_by_id(content_dir, item)
		0 ignored issues – show introduced 2021-04-01 12:48 UTC by Report Bug Copy Issue Report The variable `content_dir` does not seem to be defined in case `env_yaml` on line `92` is `False`. Are you sure this can never be the case? Loading history...
109		if rule_yaml is None:
110		# item not found in benchmark_root
111		continue
112		rule = ssg.build_yaml.Rule.from_yaml(rule_yaml, env_yaml)
113		if rule.prodtype == "all" or product in rule.prodtype:
114		control.rules.append(item)
115		else:
116		logging.info("Rule {item} doesn't apply to {product}".format(
117		item=item,
118		product=product))
119
120	2	control.related_rules = control_dict.get("related_rules", [])
121	2	control.note = control_dict.get("note")
122	2	return control
123
124
125	2	class Level():
126	2	def __init__(self):
127	2	self.id = None
128	2	self.inherits_from = None
129
130	2	@classmethod
131		def from_level_dict(cls, level_dict):
132	2	level = cls()
133	2	level.id = ssg.utils.required_key(level_dict, "id")
134	2	level.inherits_from = level_dict.get("inherits_from")
135	2	return level
136
137	2	class Policy():
138	2	def __init__(self, filepath, env_yaml=None):
139	2	self.id = None
140	2	self.env_yaml = env_yaml
141	2	self.filepath = filepath
142	2	self.controls = []
143	2	self.controls_by_id = dict()
144	2	self.levels = []
145	2	self.levels_by_id = dict()
146	2	self.title = ""
147	2	self.source = ""
148
149	2	def _parse_controls_tree(self, tree):
150	2	default_level = ["default"]
151	2	if self.levels:
152	2	default_level = [self.levels[0].id]
153
154	2	for node in tree:
155	2	try:
156	2	control = Control.from_control_dict(
157		node, self.env_yaml, default_level=default_level)
158		except Exception as exc:
159		msg = (
160		"Unable to parse controls from {filename}: {error}"
161		.format(filename=self.filepath, error=str(exc)))
162		raise RuntimeError(msg)
163	2	if "controls" in node:
164	2	for sc in self._parse_controls_tree(node["controls"]):
165	2	yield sc
166	2	control.rules.extend(sc.rules)
167	2	control.variables.update(sc.variables)
168	2	control.related_rules.extend(sc.related_rules)
169	2	yield control
170
171	2	def load(self):
172	2	yaml_contents = ssg.yaml.open_and_expand(self.filepath, self.env_yaml)
173	2	self.id = ssg.utils.required_key(yaml_contents, "id")
174	2	self.title = ssg.utils.required_key(yaml_contents, "title")
175	2	self.source = yaml_contents.get("source", "")
176
177	2	level_list = yaml_contents.get("levels", [])
178	2	for lv in level_list:
179	2	level = Level.from_level_dict(lv)
180	2	self.levels.append(level)
181	2	self.levels_by_id[level.id] = level
182
183	2	controls_tree = ssg.utils.required_key(yaml_contents, "controls")
184	2	for c in self._parse_controls_tree(controls_tree):
185	2	self.controls.append(c)
186	2	self.controls_by_id[c.id] = c
187
188	2	def get_control(self, control_id):
189	2	try:
190	2	c = self.controls_by_id[control_id]
191	2	return c
192		except KeyError:
193		msg = "%s not found in policy %s" % (
194		control_id, self.id
195		)
196		raise ValueError(msg)
197
198	2	def get_level(self, level_id):
199	2	try:
200	2	lv = self.levels_by_id[level_id]
201	2	return lv
202		except KeyError:
203		msg = "Level %s not found in policy %s" % (
204		level_id, self.id
205		)
206		raise ValueError(msg)
207
208	2	def get_level_with_ancestors_sequence(self, level_id):
209		# use OrderedDict for Python2 compatibility instead of ordered set
210	2	levels = collections.OrderedDict()
211	2	level = self.get_level(level_id)
212	2	levels[level] = ""
213	2	if level.inherits_from:
214	2	for lv in level.inherits_from:
215	2	eligible_levels = [l for l in self.get_level_with_ancestors_sequence(lv) if l not in levels.keys()]
216	2	for l in eligible_levels:
217	2	levels[l] = ""
218	2	return list(levels.keys())
219
220
221	2	class ControlsManager():
222	2	def __init__(self, controls_dir, env_yaml=None):
223	2	self.controls_dir = os.path.abspath(controls_dir)
224	2	self.env_yaml = env_yaml
225	2	self.policies = {}
226
227	2	def load(self):
228	2	if not os.path.exists(self.controls_dir):
229		return
230	2	for filename in sorted(glob(os.path.join(self.controls_dir, "*.yml"))):
231	2	logging.info("Found file %s" % (filename))
232	2	filepath = os.path.join(self.controls_dir, filename)
233	2	policy = Policy(filepath, self.env_yaml)
234	2	policy.load()
235	2	self.policies[policy.id] = policy
236
237	2	def get_control(self, policy_id, control_id):
238	2	try:
239	2	policy = self.policies[policy_id]
240		except KeyError:
241		msg = "policy '%s' doesn't exist" % (policy_id)
242		raise ValueError(msg)
243	2	control = policy.get_control(control_id)
244	2	return control
245
246	2	def _get_policy(self, policy_id):
247	2	try:
248	2	policy = self.policies[policy_id]
249		except KeyError:
250		msg = "policy '%s' doesn't exist" % (policy_id)
251		raise ValueError(msg)
252	2	return policy
253
254	2	def get_all_controls_of_level(self, policy_id, level_id):
255	2	policy = self._get_policy(policy_id)
256	2	levels = policy.get_level_with_ancestors_sequence(level_id)
257	2	all_policy_controls = self.get_all_controls(policy_id)
258	2	eligible_controls = []
259	2	already_defined_variables = set()
260		# we will go level by level, from top to bottom
261		# this is done to enable overriding of variables by higher levels
262	2	for lv in levels:
263	2	for control in all_policy_controls:
264	2	if lv.id not in control.levels:
265	2	continue
266
267	2	variables = set(control.variables.keys())
268
269	2	variables_to_remove = variables.intersection(already_defined_variables)
270	2	already_defined_variables.update(variables)
271
272	2	new_c = self._get_control_without_variables(variables_to_remove, control)
273	2	eligible_controls.append(new_c)
274
275	2	return eligible_controls
276
277	2	@staticmethod
278		def _get_control_without_variables(variables_to_remove, control):
279	2	if not variables_to_remove:
280	2	return control
281
282	2	new_c = copy.deepcopy(control)
283	2	for var in variables_to_remove:
284	2	del new_c.variables[var]
285	2	return new_c
286
287	2	def get_all_controls(self, policy_id):
288	2	policy = self._get_policy(policy_id)
289		return policy.controls_by_id.values()
290

ComplianceAsCode / content

Push — master ( ad184d...ba033d )

ssg.controls.Status.from_control_info() A

Complexity

Size

Duplication

Code Coverage

Importance

Duplication Side-by-Side

Filter issues like