ssg_test_suite.oscap.Checker.finalize() - Code Metrics - Inspection of "SSG test suite support for Docker." - ComplianceAsCode/content - Measure and Improve Code Quality continuously with Scrutinizer

Passed

Pull Request — master (#3179)

by Matěj

created 2018-08-20 13:16 UTC

ssg_test_suite.oscap.Checker.finalize() A

↳ Parent: Project

Complexity

Conditions

Size

Total Lines	10
Code Lines	9

Duplication

Lines	0
Ratio	0 %

Importance

Changes

Metric	Value
cc	3
eloc	9
nop	1
dl	0
loc	10
rs	9.95
c	0
b	0
f	0

#!/usr/bin/env python2
from __future__ import print_function

import logging
import os.path
import re
import subprocess
import collections
import xml.etree.ElementTree
import json
from ssg_test_suite.log import LogHelper

logging.getLogger(__name__).addHandler(logging.NullHandler())

_CONTEXT_RETURN_CODES = {'pass': 0,
                         'fail': 2,
                         'error': 1,
                         'notapplicable': 0,
                         'fixed': 0}

_ANSIBLE_TEMPLATE = 'urn:xccdf:fix:script:ansible'
_BASH_TEMPLATE = 'urn:xccdf:fix:script:sh'
_XCCDF_NS = 'http://checklists.nist.gov/xccdf/1.2'


def analysis_to_serializable(analysis):
    result = dict(analysis)
    for key, value in analysis.items():
        if type(value) == set:
            result[key] = tuple(value)
    return result


def save_analysis_to_json(analysis, output_fname):
    analysis2 = analysis_to_serializable(analysis)
    with open(output_fname, "w") as f:
        json.dump(analysis2, f)


def triage_xml_results(fname):
    tree = xml.etree.ElementTree.parse(fname)
    all_xml_results = tree.findall(".//{%s}rule-result" % _XCCDF_NS)

    triaged = collections.defaultdict(set)
    for result in list(all_xml_results):
        idref = result.get("idref")
        status = result.find("{%s}result" % _XCCDF_NS).text
        triaged[status].add(idref)

    return triaged


def run_cmd_local(command, verbose_path):
    command_string = ' '.join(command)
    logging.debug('Running {}'.format(command_string))
    returncode, output = _run_cmd(command, verbose_path)
    return returncode, output


def run_cmd_remote(command_string, domain_ip, verbose_path):
    machine = 'root@{0}'.format(domain_ip)
    remote_cmd = ['ssh', machine, command_string]
    logging.debug('Running {}'.format(command_string))
    returncode, output = _run_cmd(remote_cmd, verbose_path)
    return returncode, output


def _run_cmd(command_list, verbose_path):
    returncode = 0
    output = b""
    try:
        with open(verbose_path, 'w') as verbose_file:
            output = subprocess.check_output(
                command_list, stderr=verbose_file)
    except subprocess.CalledProcessError as e:
        returncode = e.returncode
        output = e.output
    return returncode, output.decode('utf-8')


def send_files_remote(verbose_path, remote_dir, domain_ip, *files):
    """Upload files to VM."""
    # files is a list of absolute paths on the host
    success = True
    destination = 'root@{0}:{1}'.format(domain_ip, remote_dir)
    files_string = ' '.join(files)

    logging.debug('Uploading files {0} to {1}'.format(files_string,
                                                      destination))
    command = ['scp'] + list(files) + [destination]
    if run_cmd_local(command, verbose_path)[0] != 0:
        logging.error('Failed to upload files {0}'.format(files_string))
        success = False
    return success


def get_file_remote(verbose_path, local_dir, domain_ip, remote_path):
    """Download a file from VM."""
    # remote_path is an absolute path of a file on remote machine
    success = True
    source = 'root@{0}:{1}'.format(domain_ip, remote_path)
    logging.debug('Downloading file {0} to {1}'
                  .format(source, local_dir))
    command = ['scp', source, local_dir]
    if run_cmd_local(command, verbose_path)[0] != 0:
        logging.error('Failed to download file {0}'.format(remote_path))
        success = False
    return success


def find_result_id_in_output(output):
    match = re.search('result id.*$', output, re.IGNORECASE | re.MULTILINE)
    if match is None:
        return None
    # Return the right most word of the match which is the result id.
    return match.group(0).split()[-1]


def ansible_playbook_set_hosts(playbook):
    """Updates ansible playbok to apply to all hosts."""
    with open(playbook, 'r') as f:
        lines = f.readlines()
    lines.insert(1, ' - hosts: all\n')
    with open(playbook, 'w') as f:
        for line in lines:
            f.write(line)


def get_result_id_from_arf(arf_path, verbose_path):
    command = ['oscap', 'info', arf_path]
    command_string = ' '.join(command)
    returncode, output = run_cmd_local(command, verbose_path)
    if returncode != 0:
        raise RuntimeError('{0} returned {1} exit code'.
                           format(command_string, returncode))
    res_id = find_result_id_in_output(output)
    if res_id is None:
        raise RuntimeError('Failed to find result ID in {0}'
                           .format(arf_path))
    return res_id


def generate_fixes_remotely(formatting, verbose_path):
    command_base = ['oscap', 'xccdf', 'generate', 'fix']
    command_options = [
        '--benchmark-id', formatting['benchmark_id'],
        '--profile', formatting['profile'],
        '--template', formatting['output_template'],
        '--output', '/{output_file}'.format(** formatting),
    ]
    command_operands = ['/{arf_file}'.format(** formatting)]
    if 'result_id' in formatting:
        command_options.extend(['--result-id', formatting['result_id']])

    command_string = ' '.join(command_base + command_options + command_operands)
    rc, stdout = run_cmd_remote(command_string,
                                formatting['domain_ip'], verbose_path)
    if rc != 0:
        msg = ('Command {0} ended with return code {1} (expected 0).'
               .format(command_string, rc))
        raise RuntimeError(msg)


def run_stage_remediation_ansible(run_type, formatting, verbose_path):
    """
       Returns False on error, or True in case of successful bash scripts
       run."""
    formatting['output_template'] = _ANSIBLE_TEMPLATE
    send_arf_to_remote_machine_and_generate_remediations_there(
        run_type, formatting, verbose_path)
    if not get_file_remote(verbose_path, LogHelper.LOG_DIR,
                           formatting['domain_ip'],
                           '/' + formatting['output_file']):
        return False
    ansible_playbook_set_hosts(formatting['playbook'])
    command = (
        'ansible-playbook',  '-i', '{0},'.format(formatting['domain_ip']),
        '-u' 'root', formatting['playbook'])
    command_string = ' '.join(command)
    returncode, output = run_cmd_local(command, verbose_path)
    # Appends output of ansible-playbook to the verbose_path file.
    with open(verbose_path, 'a') as f:
        f.write('Stdout of "{}":'.format(command_string))
        f.write(output)
    if returncode != 0:
        msg = (
            'Ansible playbook remediation run has '
            'exited with return code {} instead of expected 0'
            .format(returncode))
        LogHelper.preload_log(logging.ERROR, msg, 'fail')
        return False
    return True


def run_stage_remediation_bash(run_type, formatting, verbose_path):
    """
       Returns False on error, or True in case of successful Ansible playbook
       run."""
    formatting['output_template'] = _BASH_TEMPLATE
    send_arf_to_remote_machine_and_generate_remediations_there(
        run_type, formatting, verbose_path)
    if not get_file_remote(verbose_path, LogHelper.LOG_DIR,
                           formatting['domain_ip'],
                           '/' + formatting['output_file']):
        return False

    command_string = '/bin/bash /{output_file}'.format(** formatting)
    returncode, output = run_cmd_remote(
        command_string, formatting['domain_ip'], verbose_path)
    # Appends output of script execution to the verbose_path file.
    with open(verbose_path, 'a') as f:
        f.write('Stdout of "{}":'.format(command_string))
        f.write(output)
    if returncode != 0:
        msg = (
            'Bash script remediation run has exited with return code {} '
            'instead of expected 0'.format(returncode))
        LogHelper.preload_log(logging.ERROR, msg, 'fail')
        return False
    return True


def send_arf_to_remote_machine_and_generate_remediations_there(
        run_type, formatting, verbose_path):
    if run_type == 'rule':
        try:
            res_id = get_result_id_from_arf(formatting['arf'], verbose_path)
        except Exception as exc:
            logging.error(str(exc))
            return False
        formatting['result_id'] = res_id

    if not send_files_remote(
            verbose_path, '/', formatting['domain_ip'], formatting['arf']):
        return False

    try:
        generate_fixes_remotely(formatting, verbose_path)
    except Exception as exc:
        logging.error(str(exc))
        return False


class GenericRunner(object):
    def __init__(self, domain_ip, profile, datastream, benchmark_id):
        self.domain_ip = domain_ip
        self.profile = profile
        self.datastream = datastream
        self.benchmark_id = benchmark_id

        self.arf_file = ''
        self.arf_path = ''
        self.verbose_path = ''
        self.report_path = ''
        self.results_path = ''
        self.stage = 'undefined'

        self.clean_files = False
        self._filenames_to_clean_afterwards = set()

        self.command_base = []
        self.command_options = []
        self.command_operands = []

    def _make_arf_path(self):
        self.arf_file = self._get_arf_file()
        self.arf_path = os.path.join(LogHelper.LOG_DIR, self.arf_file)

    def _get_arf_file(self):
        raise NotImplementedError()

    def _make_verbose_path(self):
        verbose_file = self._get_verbose_file()
        verbose_path = os.path.join(LogHelper.LOG_DIR, verbose_file)
        self.verbose_path = LogHelper.find_name(verbose_path, '.verbose.log')

    def _get_verbose_file(self):
        raise NotImplementedError()

    def _make_report_path(self):
        report_file = self._get_report_file()
        report_path = os.path.join(LogHelper.LOG_DIR, report_file)
        self.report_path = LogHelper.find_name(report_path, '.html')

    def _get_report_file(self):
        raise NotImplementedError()

    def _make_results_path(self):
        results_file = self._get_results_file()
        results_path = os.path.join(LogHelper.LOG_DIR, results_file)
        self.results_path = LogHelper.find_name(results_path, '.xml')

    def _get_results_file(self):
        raise NotImplementedError()

    def _generate_report_file(self):
        self.command_options.extend([
            '--report', self.report_path,
        ])
        self._filenames_to_clean_afterwards.add(self.report_path)

    def prepare_oscap_ssh_arguments(self):
        full_hostname = 'root@{}'.format(self.domain_ip)
        self.command_base.extend(
            ['oscap-ssh', full_hostname, '22', 'xccdf', 'eval'])
        self.command_options.extend([
            '--benchmark-id', self.benchmark_id,
            '--profile', self.profile,
            '--verbose', 'DEVEL',
            '--progress', '--oval-results',
        ])
        self.command_operands.append(self.datastream)

    def run_stage(self, stage):
        self.stage = stage

        self._make_verbose_path()
        self._make_report_path()
        self._make_arf_path()
        self._make_results_path()

        self.command_base = []
        self.command_options = []
        self.command_operands = []

        result = None
        if stage == 'initial':
            result = self.initial()
        elif stage == 'remediation':
            result = self.remediation()
        elif stage == 'final':
            result = self.final()
        else:
            raise RuntimeError('Unknown stage: {}.'.format(stage))

        if self.clean_files:
            for fname in tuple(self._filenames_to_clean_afterwards):
                try:
                    os.remove(fname)
                except OSError as exc:
                    logging.error(
                        "Failed to cleanup file '{0}'"
                        .format(fname))
                finally:
                    self._filenames_to_clean_afterwards.remove(fname)

        if result:
            LogHelper.log_preloaded('pass')
        else:
            LogHelper.log_preloaded('fail')
        return result

    @property
    def get_command(self):
        return self.command_base + self.command_options + self.command_operands

    def make_oscap_call(self):
        raise NotImplementedError()

    def initial(self):
        self.command_options += ['--results', self.results_path]
        result = self.make_oscap_call()
        return result

    def remediation(self):
        raise NotImplementedError()

    def final(self):
        self.command_options += ['--results', self.results_path]
        result = self.make_oscap_call()
        return result

    def analyze(self, stage):
        triaged_results = triage_xml_results(self.results_path)
        triaged_results["stage"] = stage
        triaged_results["runner"] = self.__class__.__name__
        return triaged_results

    def _get_formatting_dict_for_remediation(self):
        formatting = {
            'domain_ip': self.domain_ip,
            'profile': self.profile,
            'datastream': self.datastream,
            'benchmark_id': self.benchmark_id
        }
        formatting['arf'] = self.arf_path
        formatting['arf_file'] = self.arf_file
        return formatting


class ProfileRunner(GenericRunner):
    def _get_arf_file(self):
        return '{0}-initial-arf.xml'.format(self.profile)

    def _get_verbose_file(self):
        return '{0}-{1}'.format(self.profile, self.stage)

    def _get_report_file(self):
        return '{0}-{1}'.format(self.profile, self.stage)

    def _get_results_file(self):
        return '{0}-{1}-results'.format(self.profile, self.stage)

    def make_oscap_call(self):
        self.prepare_oscap_ssh_arguments()
        self._generate_report_file()
        returncode = run_cmd_local(self.get_command, self.verbose_path)[0]
        if returncode not in [0, 2]:
            logging.error(('Profile run should end with return code 0 or 2 '
                           'not "{0}" as it did!').format(returncode))
            return False
        return True


class RuleRunner(GenericRunner):
    def __init__(
            self, domain_ip, profile, datastream, benchmark_id,
            rule_id, script_name, dont_clean):
        super(RuleRunner, self).__init__(
            domain_ip, profile, datastream, benchmark_id,
        )

        self.rule_id = rule_id
        self.context = None
        self.script_name = script_name
        self.clean_files = not dont_clean

        self._oscap_output = ''

    def _get_arf_file(self):
        return '{0}-initial-arf.xml'.format(self.rule_id)

    def _get_verbose_file(self):
        return '{0}-{1}-{2}'.format(self.rule_id, self.script_name, self.stage)

    def _get_report_file(self):
        return '{0}-{1}-{2}'.format(self.rule_id, self.script_name, self.stage)

    def _get_results_file(self):
        return '{0}-{1}-{2}-results-{3}'.format(
            self.rule_id, self.script_name, self.profile, self.stage)

    def make_oscap_call(self):
        self.prepare_oscap_ssh_arguments()
        self._generate_report_file()
        self.command_options.extend(
            ['--rule', self.rule_id])

        returncode, self._oscap_output = run_cmd_local(self.get_command, self.verbose_path)

        expected_return_code = _CONTEXT_RETURN_CODES[self.context]

        if returncode != expected_return_code:
            msg = (
                'Scan has exited with return code {0}, '
                'instead of expected {1} during stage {2}'
                .format(returncode, expected_return_code, self.stage)
            )
            LogHelper.preload_log(logging.ERROR, msg, 'fail')
            return False
        return True

    def final(self):
        success = super(RuleRunner, self).final()
        success = success and self._analyze_output_of_oscap_call()

        return success

    def _analyze_output_of_oscap_call(self):
        local_success = True
        # check expected result
        actual_results = re.findall('{0}:(.*)$'.format(self.rule_id),
                                    self._oscap_output,
                                    re.MULTILINE)
        if actual_results:
            if self.context not in actual_results:
                LogHelper.preload_log(logging.ERROR,
                                      ('Rule result should have been '
                                       '"{0}", but is "{1}"!'
                                       ).format(self.context,
                                                ', '.join(actual_results)),
                                      'fail')
                local_success = False
        else:
            msg = (
                'Rule {0} has not been evaluated! Wrong profile selected?'
                .format(self.rule_id))
            LogHelper.preload_log(logging.ERROR, msg, 'fail')
            local_success = False
        return local_success

    def _get_formatting_dict_for_remediation(self):
        fmt = super(RuleRunner, self)._get_formatting_dict_for_remediation()
        fmt['rule_id'] = self.rule_id

        return fmt

    def run_stage_with_context(self, stage, context):
        self.context = context
        return self.run_stage(stage)


class OscapProfileRunner(ProfileRunner):
    def remediation(self):
        self.command_options += ['--remediate']
        return self.make_oscap_call()


class AnsibleProfileRunner(ProfileRunner):
    def initial(self):
        self.command_options += ['--results-arf', self.arf_path]
        return super(AnsibleProfileRunner, self).initial()

    def remediation(self):
        formatting = self._get_formatting_dict_for_remediation()
        formatting['output_file'] = '{0}.yml'.format(self.profile)
        formatting['playbook'] = os.path.join(LogHelper.LOG_DIR,
                                              formatting['output_file'])

        return run_stage_remediation_ansible('profile', formatting, self.verbose_path)


class BashProfileRunner(ProfileRunner):
    def initial(self):
        self.command_options += ['--results-arf', self.arf_path]
        return super(BashProfileRunner, self).initial()

    def remediation(self):
        formatting = self._get_formatting_dict_for_remediation()
        formatting['output_file'] = '{0}.sh'.format(self.profile)

        return run_stage_remediation_bash('profile', formatting, self.verbose_path)


class OscapRuleRunner(RuleRunner):
    def remediation(self):
        self.command_options += ['--remediate']
        return self.make_oscap_call()


class BashRuleRunner(RuleRunner):
    def initial(self):
        self.command_options += ['--results-arf', self.arf_path]
        return super(BashRuleRunner, self).initial()

    def remediation(self):
        formatting = self._get_formatting_dict_for_remediation()
        formatting['output_file'] = '{0}.sh'.format(self.rule_id)

        success = run_stage_remediation_bash('rule', formatting, self.verbose_path)
        return success


class AnsibleRuleRunner(RuleRunner):
    def initial(self):
        self.command_options += ['--results-arf', self.arf_path]
        return super(AnsibleRuleRunner, self).initial()

    def remediation(self):
        formatting = self._get_formatting_dict_for_remediation()
        formatting['output_file'] = '{0}.yml'.format(self.rule_id)
        formatting['playbook'] = os.path.join(LogHelper.LOG_DIR,
                                              formatting['output_file'])

        success = run_stage_remediation_ansible('rule', formatting, self.verbose_path)
        return success


class Checker(object):
    def __init__(self, test_env):
        self.test_env = test_env
        self.executed_tests = 0

        self.datastream = ""
        self.benchmark_id = ""
        self.remediate_using = ""

    def test_target(self, target):
        self.start()
        try:
            with self.test_env.in_layer('origin'):
                self._test_target(target)
        except KeyboardInterrupt:
            logging.info("Terminating the test run due to keyboard interrupt.")
        finally:
            self.finalize()

    def _test_by_profiles(self, profiles, ** run_test_args):
        if len(profiles) > 1:
            for profile in profiles:
                with self.test_env.in_layer('profile'):
                    self._run_test(profile, ** run_test_args)
        elif profiles:
            self._run_test(profiles[0], ** run_test_args)

    def _test_target(self, target):
        raise NotImplementedError()

    def start(self):
        self.executed_tests = 0

        try:
            self.test_env.start()
        except Exception as exc:
            msg = ("Failed to start test environment '{0}': {1}"
                   .format(self.test_env.name, str(exc)))
            raise RuntimeError(msg)

    def finalize(self):
        if not self.executed_tests:
            logging.error("Nothing has been tested!")

        try:
            self.test_env.finalize()
        except Exception as exc:
            msg = ("Failed to finalize test environment '{0}': {1}"
                   .format(self.test_env.name, str(exc)))
            raise RuntimeError(msg)


REMEDIATION_PROFILE_RUNNERS = {
    'oscap': OscapProfileRunner,
    'bash': BashProfileRunner,
    'ansible': AnsibleProfileRunner,
}


REMEDIATION_RULE_RUNNERS = {
    'oscap': OscapRuleRunner,
    'bash': BashRuleRunner,
    'ansible': AnsibleRuleRunner,
}


REMEDIATION_RUNNER_TO_REMEDIATION_MEANS = {
    'oscap': 'bash',
    'bash': 'bash',
    'ansible': 'ansible',
}


1			#!/usr/bin/env python2
2			from __future__ import print_function
3
4			import logging
5			import os.path
6			import re
7			import subprocess
8			import collections
9			import xml.etree.ElementTree
10			import json
11			from ssg_test_suite.log import LogHelper
12
13			logging.getLogger(__name__).addHandler(logging.NullHandler())
14
15			_CONTEXT_RETURN_CODES = {'pass': 0,
16			'fail': 2,
17			'error': 1,
18			'notapplicable': 0,
19			'fixed': 0}
20
21			_ANSIBLE_TEMPLATE = 'urn:xccdf:fix:script:ansible'
22			_BASH_TEMPLATE = 'urn:xccdf:fix:script:sh'
23			_XCCDF_NS = 'http://checklists.nist.gov/xccdf/1.2'
24
25
26			def analysis_to_serializable(analysis):
27			result = dict(analysis)
28			for key, value in analysis.items():
29			if type(value) == set:
30			result[key] = tuple(value)
31			return result
32
33
34			def save_analysis_to_json(analysis, output_fname):
35			analysis2 = analysis_to_serializable(analysis)
36			with open(output_fname, "w") as f:
37			json.dump(analysis2, f)
38
39
40			def triage_xml_results(fname):
41			tree = xml.etree.ElementTree.parse(fname)
42			all_xml_results = tree.findall(".//{%s}rule-result" % _XCCDF_NS)
43
44			triaged = collections.defaultdict(set)
45			for result in list(all_xml_results):
46			idref = result.get("idref")
47			status = result.find("{%s}result" % _XCCDF_NS).text
48			triaged[status].add(idref)
49
50			return triaged
51
52
53			def run_cmd_local(command, verbose_path):
54			command_string = ' '.join(command)
55			logging.debug('Running {}'.format(command_string))
56			returncode, output = _run_cmd(command, verbose_path)
57			return returncode, output
58
59
60			def run_cmd_remote(command_string, domain_ip, verbose_path):
61			machine = 'root@{0}'.format(domain_ip)
62			remote_cmd = ['ssh', machine, command_string]
63			logging.debug('Running {}'.format(command_string))
64			returncode, output = _run_cmd(remote_cmd, verbose_path)
65			return returncode, output
66
67
68			def _run_cmd(command_list, verbose_path):
69			returncode = 0
70			output = b""
71			try:
72			with open(verbose_path, 'w') as verbose_file:
73			output = subprocess.check_output(
74			command_list, stderr=verbose_file)
75			except subprocess.CalledProcessError as e:
76			returncode = e.returncode
77			output = e.output
78			return returncode, output.decode('utf-8')
79
80
81			def send_files_remote(verbose_path, remote_dir, domain_ip, *files):
82			"""Upload files to VM."""
83			# files is a list of absolute paths on the host
84			success = True
85			destination = 'root@{0}:{1}'.format(domain_ip, remote_dir)
86			files_string = ' '.join(files)
87
88			logging.debug('Uploading files {0} to {1}'.format(files_string,
89			destination))
90			command = ['scp'] + list(files) + [destination]
91			if run_cmd_local(command, verbose_path)[0] != 0:
92			logging.error('Failed to upload files {0}'.format(files_string))
93			success = False
94			return success
95
96
97			def get_file_remote(verbose_path, local_dir, domain_ip, remote_path):
98			"""Download a file from VM."""
99			# remote_path is an absolute path of a file on remote machine
100			success = True
101			source = 'root@{0}:{1}'.format(domain_ip, remote_path)
102			logging.debug('Downloading file {0} to {1}'
103			.format(source, local_dir))
104			command = ['scp', source, local_dir]
105			if run_cmd_local(command, verbose_path)[0] != 0:
106			logging.error('Failed to download file {0}'.format(remote_path))
107			success = False
108			return success
109
110
111			def find_result_id_in_output(output):
112			match = re.search('result id.*$', output, re.IGNORECASE \| re.MULTILINE)
113			if match is None:
114			return None
115			# Return the right most word of the match which is the result id.
116			return match.group(0).split()[-1]
117
118
119			def ansible_playbook_set_hosts(playbook):
120			"""Updates ansible playbok to apply to all hosts."""
121			with open(playbook, 'r') as f:
122			lines = f.readlines()
123			lines.insert(1, ' - hosts: all\n')
124			with open(playbook, 'w') as f:
125			for line in lines:
126			f.write(line)
127
128
129			def get_result_id_from_arf(arf_path, verbose_path):
130			command = ['oscap', 'info', arf_path]
131			command_string = ' '.join(command)
132			returncode, output = run_cmd_local(command, verbose_path)
133			if returncode != 0:
134			raise RuntimeError('{0} returned {1} exit code'.
135			format(command_string, returncode))
136			res_id = find_result_id_in_output(output)
137			if res_id is None:
138			raise RuntimeError('Failed to find result ID in {0}'
139			.format(arf_path))
140			return res_id
141
142
143			def generate_fixes_remotely(formatting, verbose_path):
144			command_base = ['oscap', 'xccdf', 'generate', 'fix']
145			command_options = [
146			'--benchmark-id', formatting['benchmark_id'],
147			'--profile', formatting['profile'],
148			'--template', formatting['output_template'],
149			'--output', '/{output_file}'.format(** formatting),
150			]
151			command_operands = ['/{arf_file}'.format(** formatting)]
152			if 'result_id' in formatting:
153			command_options.extend(['--result-id', formatting['result_id']])
154
155			command_string = ' '.join(command_base + command_options + command_operands)
156			rc, stdout = run_cmd_remote(command_string,
157			formatting['domain_ip'], verbose_path)
158			if rc != 0:
159			msg = ('Command {0} ended with return code {1} (expected 0).'
160			.format(command_string, rc))
161			raise RuntimeError(msg)
162
163
164			def run_stage_remediation_ansible(run_type, formatting, verbose_path):
165			"""
166			Returns False on error, or True in case of successful bash scripts
167			run."""
168			formatting['output_template'] = _ANSIBLE_TEMPLATE
169			send_arf_to_remote_machine_and_generate_remediations_there(
170			run_type, formatting, verbose_path)
171			if not get_file_remote(verbose_path, LogHelper.LOG_DIR,
172			formatting['domain_ip'],
173			'/' + formatting['output_file']):
174			return False
175			ansible_playbook_set_hosts(formatting['playbook'])
176			command = (
177			'ansible-playbook', '-i', '{0},'.format(formatting['domain_ip']),
178			'-u' 'root', formatting['playbook'])
179			command_string = ' '.join(command)
180			returncode, output = run_cmd_local(command, verbose_path)
181			# Appends output of ansible-playbook to the verbose_path file.
182			with open(verbose_path, 'a') as f:
183			f.write('Stdout of "{}":'.format(command_string))
184			f.write(output)
185			if returncode != 0:
186			msg = (
187			'Ansible playbook remediation run has '
188			'exited with return code {} instead of expected 0'
189			.format(returncode))
190			LogHelper.preload_log(logging.ERROR, msg, 'fail')
191			return False
192			return True
193
194
195			def run_stage_remediation_bash(run_type, formatting, verbose_path):
196			"""
197			Returns False on error, or True in case of successful Ansible playbook
198			run."""
199			formatting['output_template'] = _BASH_TEMPLATE
200			send_arf_to_remote_machine_and_generate_remediations_there(
201			run_type, formatting, verbose_path)
202			if not get_file_remote(verbose_path, LogHelper.LOG_DIR,
203			formatting['domain_ip'],
204			'/' + formatting['output_file']):
205			return False
206
207			command_string = '/bin/bash /{output_file}'.format(** formatting)
208			returncode, output = run_cmd_remote(
209			command_string, formatting['domain_ip'], verbose_path)
210			# Appends output of script execution to the verbose_path file.
211			with open(verbose_path, 'a') as f:
212			f.write('Stdout of "{}":'.format(command_string))
213			f.write(output)
214			if returncode != 0:
215			msg = (
216			'Bash script remediation run has exited with return code {} '
217			'instead of expected 0'.format(returncode))
218			LogHelper.preload_log(logging.ERROR, msg, 'fail')
219			return False
220			return True
221
222
223			def send_arf_to_remote_machine_and_generate_remediations_there(
224			run_type, formatting, verbose_path):
225			if run_type == 'rule':
226			try:
227			res_id = get_result_id_from_arf(formatting['arf'], verbose_path)
228			except Exception as exc:
229			logging.error(str(exc))
230			return False
231			formatting['result_id'] = res_id
232
233			if not send_files_remote(
234			verbose_path, '/', formatting['domain_ip'], formatting['arf']):
235			return False
236
237			try:
238			generate_fixes_remotely(formatting, verbose_path)
239			except Exception as exc:
240			logging.error(str(exc))
241			return False
242
243
244			class GenericRunner(object):
245			def __init__(self, domain_ip, profile, datastream, benchmark_id):
246			self.domain_ip = domain_ip
247			self.profile = profile
248			self.datastream = datastream
249			self.benchmark_id = benchmark_id
250
251			self.arf_file = ''
252			self.arf_path = ''
253			self.verbose_path = ''
254			self.report_path = ''
255			self.results_path = ''
256			self.stage = 'undefined'
257
258			self.clean_files = False
259			self._filenames_to_clean_afterwards = set()
260
261			self.command_base = []
262			self.command_options = []
263			self.command_operands = []
264
265			def _make_arf_path(self):
266			self.arf_file = self._get_arf_file()
267			self.arf_path = os.path.join(LogHelper.LOG_DIR, self.arf_file)
268
269			def _get_arf_file(self):
270			raise NotImplementedError()
271
272			def _make_verbose_path(self):
273			verbose_file = self._get_verbose_file()
274			verbose_path = os.path.join(LogHelper.LOG_DIR, verbose_file)
275			self.verbose_path = LogHelper.find_name(verbose_path, '.verbose.log')
276
277			def _get_verbose_file(self):
278			raise NotImplementedError()
279
280			def _make_report_path(self):
281			report_file = self._get_report_file()
282			report_path = os.path.join(LogHelper.LOG_DIR, report_file)
283			self.report_path = LogHelper.find_name(report_path, '.html')
284
285			def _get_report_file(self):
286			raise NotImplementedError()
287
288			def _make_results_path(self):
289			results_file = self._get_results_file()
290			results_path = os.path.join(LogHelper.LOG_DIR, results_file)
291			self.results_path = LogHelper.find_name(results_path, '.xml')
292
293			def _get_results_file(self):
294			raise NotImplementedError()
295
296			def _generate_report_file(self):
297			self.command_options.extend([
298			'--report', self.report_path,
299			])
300			self._filenames_to_clean_afterwards.add(self.report_path)
301
302			def prepare_oscap_ssh_arguments(self):
303			full_hostname = 'root@{}'.format(self.domain_ip)
304			self.command_base.extend(
305			['oscap-ssh', full_hostname, '22', 'xccdf', 'eval'])
306			self.command_options.extend([
307			'--benchmark-id', self.benchmark_id,
308			'--profile', self.profile,
309			'--verbose', 'DEVEL',
310			'--progress', '--oval-results',
311			])
312			self.command_operands.append(self.datastream)
313
314			def run_stage(self, stage):
315			self.stage = stage
316
317			self._make_verbose_path()
318			self._make_report_path()
319			self._make_arf_path()
320			self._make_results_path()
321
322			self.command_base = []
323			self.command_options = []
324			self.command_operands = []
325
326			result = None
327			if stage == 'initial':
328			result = self.initial()
329			elif stage == 'remediation':
330			result = self.remediation()
331			elif stage == 'final':
332			result = self.final()
333			else:
334			raise RuntimeError('Unknown stage: {}.'.format(stage))
335
336			if self.clean_files:
337			for fname in tuple(self._filenames_to_clean_afterwards):
338			try:
339			os.remove(fname)
340			except OSError as exc:
341			logging.error(
342			"Failed to cleanup file '{0}'"
343			.format(fname))
344			finally:
345			self._filenames_to_clean_afterwards.remove(fname)
346
347			if result:
348			LogHelper.log_preloaded('pass')
349			else:
350			LogHelper.log_preloaded('fail')
351			return result
352
353			@property
354			def get_command(self):
355			return self.command_base + self.command_options + self.command_operands
356
357			def make_oscap_call(self):
358			raise NotImplementedError()
359
360			def initial(self):
361			self.command_options += ['--results', self.results_path]
362			result = self.make_oscap_call()
363			return result
364
365			def remediation(self):
366			raise NotImplementedError()
367
368			def final(self):
369			self.command_options += ['--results', self.results_path]
370			result = self.make_oscap_call()
371			return result
372
373			def analyze(self, stage):
374			triaged_results = triage_xml_results(self.results_path)
375			triaged_results["stage"] = stage
376			triaged_results["runner"] = self.__class__.__name__
377			return triaged_results
378
379			def _get_formatting_dict_for_remediation(self):
380			formatting = {
381			'domain_ip': self.domain_ip,
382			'profile': self.profile,
383			'datastream': self.datastream,
384			'benchmark_id': self.benchmark_id
385			}
386			formatting['arf'] = self.arf_path
387			formatting['arf_file'] = self.arf_file
388			return formatting
389
390
391			class ProfileRunner(GenericRunner):
392			def _get_arf_file(self):
393			return '{0}-initial-arf.xml'.format(self.profile)
394
395			def _get_verbose_file(self):
396			return '{0}-{1}'.format(self.profile, self.stage)
397
398			def _get_report_file(self):
399			return '{0}-{1}'.format(self.profile, self.stage)
400
401			def _get_results_file(self):
402			return '{0}-{1}-results'.format(self.profile, self.stage)
403
404			def make_oscap_call(self):
405			self.prepare_oscap_ssh_arguments()
406			self._generate_report_file()
407			returncode = run_cmd_local(self.get_command, self.verbose_path)[0]
408			if returncode not in [0, 2]:
409			logging.error(('Profile run should end with return code 0 or 2 '
410			'not "{0}" as it did!').format(returncode))
411			return False
412			return True
413
414
415			class RuleRunner(GenericRunner):
416			def __init__(
417			self, domain_ip, profile, datastream, benchmark_id,
418			rule_id, script_name, dont_clean):
419			super(RuleRunner, self).__init__(
420			domain_ip, profile, datastream, benchmark_id,
421			)
422
423			self.rule_id = rule_id
424			self.context = None
425			self.script_name = script_name
426			self.clean_files = not dont_clean
427
428			self._oscap_output = ''
429
430			def _get_arf_file(self):
431			return '{0}-initial-arf.xml'.format(self.rule_id)
432
433			def _get_verbose_file(self):
434			return '{0}-{1}-{2}'.format(self.rule_id, self.script_name, self.stage)
435
436			def _get_report_file(self):
437			return '{0}-{1}-{2}'.format(self.rule_id, self.script_name, self.stage)
438
439			def _get_results_file(self):
440			return '{0}-{1}-{2}-results-{3}'.format(
441			self.rule_id, self.script_name, self.profile, self.stage)
442
443			def make_oscap_call(self):
444			self.prepare_oscap_ssh_arguments()
445			self._generate_report_file()
446			self.command_options.extend(
447			['--rule', self.rule_id])
448
449			returncode, self._oscap_output = run_cmd_local(self.get_command, self.verbose_path)
450
451			expected_return_code = _CONTEXT_RETURN_CODES[self.context]
452
453			if returncode != expected_return_code:
454			msg = (
455			'Scan has exited with return code {0}, '
456			'instead of expected {1} during stage {2}'
457			.format(returncode, expected_return_code, self.stage)
458			)
459			LogHelper.preload_log(logging.ERROR, msg, 'fail')
460			return False
461			return True
462
463			def final(self):
464			success = super(RuleRunner, self).final()
465			success = success and self._analyze_output_of_oscap_call()
466
467			return success
468
469			def _analyze_output_of_oscap_call(self):
470			local_success = True
471			# check expected result
472			actual_results = re.findall('{0}:(.*)$'.format(self.rule_id),
473			self._oscap_output,
474			re.MULTILINE)
475			if actual_results:
476			if self.context not in actual_results:
477			LogHelper.preload_log(logging.ERROR,
478			('Rule result should have been '
479			'"{0}", but is "{1}"!'
480			).format(self.context,
481			', '.join(actual_results)),
482			'fail')
483			local_success = False
484			else:
485			msg = (
486			'Rule {0} has not been evaluated! Wrong profile selected?'
487			.format(self.rule_id))
488			LogHelper.preload_log(logging.ERROR, msg, 'fail')
489			local_success = False
490			return local_success
491
492			def _get_formatting_dict_for_remediation(self):
493			fmt = super(RuleRunner, self)._get_formatting_dict_for_remediation()
494			fmt['rule_id'] = self.rule_id
495
496			return fmt
497
498			def run_stage_with_context(self, stage, context):
499			self.context = context
500			return self.run_stage(stage)
501
502
503			class OscapProfileRunner(ProfileRunner):
504			def remediation(self):
505			self.command_options += ['--remediate']
506			return self.make_oscap_call()
507
508
509			class AnsibleProfileRunner(ProfileRunner):
510			def initial(self):
511			self.command_options += ['--results-arf', self.arf_path]
512			return super(AnsibleProfileRunner, self).initial()
513
514			def remediation(self):
515			formatting = self._get_formatting_dict_for_remediation()
516			formatting['output_file'] = '{0}.yml'.format(self.profile)
517			formatting['playbook'] = os.path.join(LogHelper.LOG_DIR,
518			formatting['output_file'])
519
520			return run_stage_remediation_ansible('profile', formatting, self.verbose_path)
521
522
523			class BashProfileRunner(ProfileRunner):
524			def initial(self):
525			self.command_options += ['--results-arf', self.arf_path]
526			return super(BashProfileRunner, self).initial()
527
528			def remediation(self):
529			formatting = self._get_formatting_dict_for_remediation()
530			formatting['output_file'] = '{0}.sh'.format(self.profile)
531
532			return run_stage_remediation_bash('profile', formatting, self.verbose_path)
533
534
535			class OscapRuleRunner(RuleRunner):
536			def remediation(self):
537			self.command_options += ['--remediate']
538			return self.make_oscap_call()
539
540
541			class BashRuleRunner(RuleRunner):
542			def initial(self):
543			self.command_options += ['--results-arf', self.arf_path]
544			return super(BashRuleRunner, self).initial()
545
546			def remediation(self):
547			formatting = self._get_formatting_dict_for_remediation()
548			formatting['output_file'] = '{0}.sh'.format(self.rule_id)
549
550			success = run_stage_remediation_bash('rule', formatting, self.verbose_path)
551			return success
552
553
554			class AnsibleRuleRunner(RuleRunner):
555			def initial(self):
556			self.command_options += ['--results-arf', self.arf_path]
557			return super(AnsibleRuleRunner, self).initial()
558
559			def remediation(self):
560			formatting = self._get_formatting_dict_for_remediation()
561			formatting['output_file'] = '{0}.yml'.format(self.rule_id)
562			formatting['playbook'] = os.path.join(LogHelper.LOG_DIR,
563			formatting['output_file'])
564
565			success = run_stage_remediation_ansible('rule', formatting, self.verbose_path)
566			return success
567
568
569			class Checker(object):
570			def __init__(self, test_env):
571			self.test_env = test_env
572			self.executed_tests = 0
573
574			self.datastream = ""
575			self.benchmark_id = ""
576			self.remediate_using = ""
577
578			def test_target(self, target):
579			self.start()
580			try:
581			with self.test_env.in_layer('origin'):
582			self._test_target(target)
583			except KeyboardInterrupt:
584			logging.info("Terminating the test run due to keyboard interrupt.")
585			finally:
586			self.finalize()
587
588			def _test_by_profiles(self, profiles, ** run_test_args):
589			if len(profiles) > 1:
590			for profile in profiles:
591			with self.test_env.in_layer('profile'):
592			self._run_test(profile, ** run_test_args)
593			elif profiles:
594			self._run_test(profiles[0], ** run_test_args)
595
596			def _test_target(self, target):
597			raise NotImplementedError()
598
599			def start(self):
600			self.executed_tests = 0
601
602			try:
603			self.test_env.start()
604			except Exception as exc:
605			msg = ("Failed to start test environment '{0}': {1}"
606			.format(self.test_env.name, str(exc)))
607			raise RuntimeError(msg)
608
609			def finalize(self):
610			if not self.executed_tests:
611			logging.error("Nothing has been tested!")
612
613			try:
614			self.test_env.finalize()
615			except Exception as exc:
616			msg = ("Failed to finalize test environment '{0}': {1}"
617			.format(self.test_env.name, str(exc)))
618			raise RuntimeError(msg)
619
620
621			REMEDIATION_PROFILE_RUNNERS = {
622			'oscap': OscapProfileRunner,
623			'bash': BashProfileRunner,
624			'ansible': AnsibleProfileRunner,
625			}
626
627
628			REMEDIATION_RULE_RUNNERS = {
629			'oscap': OscapRuleRunner,
630			'bash': BashRuleRunner,
631			'ansible': AnsibleRuleRunner,
632			}
633
634
635			REMEDIATION_RUNNER_TO_REMEDIATION_MEANS = {
636			'oscap': 'bash',
637			'bash': 'bash',
638			'ansible': 'ansible',
639			}
640

ComplianceAsCode / content

Pull Request — master (#3179)

ssg_test_suite.oscap.Checker.finalize() A

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like