oscap_report.scap_results_parser.data_structures

Complexity

Total Complexity

53

Size/Duplication

Total Lines	329
Duplicated Lines	0 %

Test Coverage

Coverage

80.87%

Importance

Changes

0

15 Methods

Rating	Name	Duplication	Size	Complexity
D	OvalNode._get_result_counts()	0	21	13
A	Remediation.get_type()	0	10	1
A	OvalTest.as_dict()	0	6	1
A	OvalNode.as_json()	0	2	1
A	OvalNode.log_oval_tree()	0	13	5
A	OvalObject.as_dict()	0	6	1
A	Rule.as_dict()	0	19	1
C	OvalNode.evaluate_tree()	0	22	10
A	Report.as_dict()	0	19	1
B	Report.get_severity_of_failed_rules_stats()	0	18	6
A	OvalNode.as_dict()	0	21	2
A	Group.as_dict()	0	17	2
B	Report.get_rule_results_stats()	0	25	6
A	Remediation.as_dict()	0	8	1
A	Report.get_failed_rules()	0	2	2

import json
import logging
from collections import Counter
from dataclasses import dataclass

from .exceptions import MissingProcessableRules
from .oval_result_eval import (EMPTY_RESULT, SHORT_RESULT_TO_FULL_RESULT,
                               OvalResult)


@dataclass
class Report:  # pylint: disable=R0902
    title: str = ""
    identity: str = ""
    profile_name: str = ""
    platform: str = ""
    target: str = ""
    cpe_platforms: str = ""
    scanner: str = ""
    scanner_version: str = ""
    benchmark_url: str = ""
    benchmark_id: str = ""
    benchmark_version: str = ""
    start_time: str = ""
    end_time: str = ""
    test_system: str = ""
    score: float = 0.0
    score_max: float = 0.0
    rules: dict = None

    def as_dict(self):
        return {
            "title": self.title,
            "profile_name": self.profile_name,
            "platform": self.platform,
            "target": self.target,
            "identit": self.identity,
            "cpe_platforms": self.cpe_platforms,
            "scanner": self.scanner,
            "scanner_version": self.scanner_version,
            "benchmark_url": self.benchmark_url,
            "benchmark_id": self.benchmark_id,
            "benchmark_version": self.benchmark_version,
            "start_time": self.start_time,
            "end_time": self.end_time,
            "test_system": self.test_system,
            "score": self.score,
            "score_max": self.score_max,
            "rules": self.rules,
        }

    def get_rule_results_stats(self):
        results_stats = {
            "fail": len(list(
                filter(lambda rule: rule.result.lower() == "fail", self.rules.values()))),
            "pass": len(list(
                filter(lambda rule: rule.result.lower() == "pass", self.rules.values()))),
            "unknown_error": len(list(
                filter(lambda rule:
                       rule.result.lower() in ("error", "unknown"), self.rules.values()))),
        }
        not_ignored_rules = len(list(
            filter(
                lambda rule: rule.result.lower() not in ("notselected", "notapplicable"),
                self.rules.values()
            )))
        if not_ignored_rules == 0:
            not_ignored_rules = 1
            logging.warning("There are no applicable or selected rules.")
        percent_per_rule = 100 / not_ignored_rules
        results_stats["other"] = not_ignored_rules - results_stats["fail"] - results_stats['pass']
        results_stats["fail_percent"] = results_stats["fail"] * percent_per_rule
        results_stats["pass_percent"] = results_stats["pass"] * percent_per_rule
        results_stats["other_percent"] = results_stats["other"] * percent_per_rule
        results_stats["sum_of_rules"] = not_ignored_rules
        return results_stats

    def get_severity_of_failed_rules_stats(self):
        failed_rules = self.get_failed_rules()
        count_of_failed_rules = len(failed_rules)
        if count_of_failed_rules == 0:
            raise MissingProcessableRules("There are no failed rules!")
        percent_per_rule = 100 / count_of_failed_rules
        severity_stats = {
            "low": sum(map(lambda rule: rule.severity.lower() == "low", failed_rules)),
            "medium": sum(map(lambda rule: rule.severity.lower() == "medium", failed_rules)),
            "high": sum(map(lambda rule: rule.severity.lower() == "high", failed_rules)),
            "unknown": sum(map(lambda rule: rule.severity.lower() == "unknown", failed_rules)),
        }
        severity_stats["low_percent"] = severity_stats["low"] * percent_per_rule
        severity_stats["medium_percent"] = severity_stats["medium"] * percent_per_rule
        severity_stats["high_percent"] = severity_stats["high"] * percent_per_rule
        severity_stats["unknown_percent"] = severity_stats["unknown"] * percent_per_rule
        severity_stats["sum_of_failed_rules"] = len(failed_rules)
        return severity_stats

    def get_failed_rules(self):
        return list(filter(lambda rule: rule.result.lower() == "fail", self.rules.values()))


@dataclass
class OvalObject():
    object_id: str = ""
    flag: str = ""
    object_type: str = ""
    object_data: dict = None

    def as_dict(self):
        return {
            "object_id": self.object_id,
            "flag": self.flag,
            "object_type": self.object_type,
            "object_data": self.object_data,
        }


@dataclass
class OvalTest():
    test_id: str = ""
    test_type: str = ""
    comment: str = ""
    oval_object: OvalObject = None

    def as_dict(self):
        return {
            "test_id": self.test_id,
            "test_type": self.test_type,
            "comment": self.comment,
            "oval_object": self.oval_object.as_dict(),
        }


@dataclass
class OvalNode:  # pylint: disable=R0902
    node_id: str
    node_type: str
    value: str
    negation: bool = False
    comment: str = ""
    tag: str = ""
    children: list = None
    test_info: OvalTest = None

    def as_dict(self):
        if not self.children:
            return {
                'node_id': self.node_id,
                'node_type': self.node_type,
                'value': self.value,
                'negation': self.negation,
                'comment': self.comment,
                'tag': self.tag,
                'test_info': self.test_info.as_dict(),
                'children': None
            }
        return {
            'node_id': self.node_id,
            'node_type': self.node_type,
            'value': self.value,
            'negation': self.negation,
            'comment': self.comment,
            'tag': self.tag,
            'test_info': None,
            'children': [child.as_dict() for child in self.children]
        }

    def as_json(self):
        return json.dumps(self.as_dict())

    def log_oval_tree(self, level=0):
        out = ""
        negation_str = ""
        if self.negation:
            negation_str = "not "
        if self.node_type != "value":
            out = "  " * level + self.node_type + " = " + negation_str + self.value
        else:
            out = "  " * level + self.node_id + " = " + negation_str + self.value
        logging.info(out)
        if self.children is not None:
            for child in self.children:
                child.log_oval_tree(level + 1)

    def _get_result_counts(self):
        result = Counter(EMPTY_RESULT)
        for child in self.children:
            if child.value == "true" and not child.negation:
                result["number_of_true"] += 1
            elif child.value == "true" and child.negation:
                result["number_of_false"] += 1
            elif child.value == "false" and not child.negation:
                result["number_of_false"] += 1
            elif child.value == "false" and child.negation:
                result["number_of_true"] += 1
            elif child.value == "not evaluated":
                result["number_of_noteval"] += 1
            elif child.value == "not applicable":
                result["number_of_notappl"] += 1
            else:
                if child.node_type != "value":
                    result["number_of_" + str(child.evaluate_tree())] += 1

The variable str does not seem to be defined.

                else:
                    result["number_of_" + child.value] += 1
        return result

    def evaluate_tree(self):
        results_counts = self._get_result_counts()
        oval_result = OvalResult(**results_counts)
        out_result = None
        if oval_result.is_notapp_result():
            out_result = "notappl"
        else:
            if self.node_type.lower() == "or":
                out_result = oval_result.eval_operator_or()
            elif self.node_type.lower() == "and":
                out_result = oval_result.eval_operator_and()
            elif self.node_type.lower() == "one":
                out_result = oval_result.eval_operator_one()
            elif self.node_type.lower() == "xor":
                out_result = oval_result.eval_operator_xor()

        if out_result == "true" and self.negation:
            out_result = "false"
        elif out_result == "false" and self.negation:
            out_result = "true"

        return SHORT_RESULT_TO_FULL_RESULT.get(out_result, out_result)


@dataclass
class Rule:  # pylint: disable=R0902
    rule_id: str = ""
    title: str = ""
    result: str = ""
    multi_check: bool = False
    time: str = ""
    severity: str = ""
    identifiers: list = None
    references: list = None
    description: str = ""
    rationale: str = ""
    warnings: list = None
    platforms: list = None
    oval_definition_id: str = ""
    message: str = ""
    remediations: list = None
    oval_tree: OvalNode = None
    cpe_tree: OvalNode = None

    def as_dict(self):
        return {
            "rule_id": self.rule_id,
            "title": self.title,
            "result": self.result,
            "multi_check": self.multi_check,
            "time": self.time,
            "severity": self.severity,
            "identifiers": self.identifiers,
            "references": self.references,
            "description": self.description,
            "rationale": self.rationale,
            "warnings": self.warnings,
            "platforms": self.platforms,
            "oval_definition_id": self.oval_definition_id,
            "message": self.message,
            "remediations": self.remediations,
            "oval_tree": self.oval_tree.as_dict(),
            "cpe_tree": self.cpe_tree.as_dict(),
        }


@dataclass
class Remediation:
    remediation_id: str = ""
    system: str = ""
    complexity: str = ""
    disruption: str = ""
    strategy: str = ""
    fix: str = ""

    def as_dict(self):
        return {
            "remediation_id": self.remediation_id,
            "system": self.system,
            "complexity": self.complexity,
            "disruption": self.disruption,
            "strategy": self.strategy,
            "fix": self.fix,
        }

    def get_type(self):
        script_types = {
            "urn:xccdf:fix:script:sh": "Shell script",
            "urn:xccdf:fix:script:ansible": "Ansible snippet",
            "urn:xccdf:fix:script:puppet": "Puppet snippet",
            "urn:redhat:anaconda:pre": "Anaconda snippet",
            "urn:xccdf:fix:script:kubernetes": "Kubernetes snippet",
            "urn:redhat:osbuild:blueprint": "OSBuild Blueprint snippet",
        }
        return script_types.get(self.system, "script")


@dataclass
class Group:
    group_id: str = ""
    title: str = ""
    description: str = ""
    platforms: list = None
    rules_ids: list = None
    sub_groups: list = None

    def as_dict(self):
        if not self.sub_groups:
            return {
                "group_id": self.group_id,
                "title": self.title,
                "description": self.description,
                "platforms": self.platforms,
                "rules_ids": self.rules_ids,
                "sub_groups": None,
            }
        return {
            "group_id": self.group_id,
            "title": self.title,
            "description": self.description,
            "platforms": self.platforms,
            "rules_ids": self.rules_ids,
            "sub_groups": [sub_group.as_dict() for sub_group in self.sub_groups],
        }