ssg_test_suite.oscap.OscapRuleRunner.final() - Code Metrics - Inspection of "Fix remediation testing" - ComplianceAsCode/content - Measure and Improve Code Quality continuously with Scrutinizer

Passed

Pull Request — master (#3242)

by Marek

created 2018-09-03 11:21 UTC

ssg_test_suite.oscap.OscapRuleRunner.final() A

↳ Parent: Project

Complexity

Conditions

Size

Total Lines	4
Code Lines	2

Duplication

Lines	0
Ratio	0 %

Importance

Changes

Metric	Value
cc	1
eloc	2
nop	1
dl	0
loc	4
rs	10
c	0
b	0
f	0

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

import logging
import os.path
import re
import collections
import xml.etree.ElementTree
import json

from ssg_test_suite.log import LogHelper
from ssg_test_suite import test_env
from ssg_test_suite import common

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 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'] + common.IGNORE_KNOWN_HOSTS_OPTIONS + list(files) + [destination]
    if common.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'] + common.IGNORE_KNOWN_HOSTS_OPTIONS + [source, local_dir]
    if common.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 = common.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 = common.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 = common.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 = common.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, environment, profile, datastream, benchmark_id):
        self.environment = environment
        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_online_scanning_arguments(self):
        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.environment.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_online_scanning_arguments()
        self._generate_report_file()
        env = dict(SSH_ADDITIONAL_OPTIONS=" ".join(common.IGNORE_KNOWN_HOSTS_OPTIONS))
        env.update(os.environ)
        returncode = common.run_cmd_local(self.get_command, self.verbose_path, env=env)[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, environment, profile, datastream, benchmark_id,
            rule_id, script_name, dont_clean):
        super(RuleRunner, self).__init__(
            environment, 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_online_scanning_arguments()
        self._generate_report_file()
        self.command_options.extend(
            ['--rule', self.rule_id])
        returncode, self._oscap_output = self.environment.scan(
            self.command_options + self.command_operands, 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()

    def final(self):
        """ There is no need to run final scan again - result won't be different
        to what we already have in remediation step."""
        return True


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:
            self._test_target(target)
        except KeyboardInterrupt:
            logging.info("Terminating the test run due to keyboard interrupt.")
        except RuntimeError as exc:
            logging.error("Terminating due to error: {msg}.".format(msg=str(exc)))
        finally:
            self.finalize()

    def run_test_for_all_profiles(self, profiles, test_data=None):
        if len(profiles) > 1:
            with test_env.SavedState.create_from_environment(self.test_env, "prepared") as state:
                args_list = [(p, test_data) for p in profiles]
                state.map_on_top(self._run_test, args_list)
        elif profiles:
            self._run_test(profiles[0], test_data)

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

ComplianceAsCode / content

Pull Request — master (#3242)

ssg_test_suite.oscap.OscapRuleRunner.final() A

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like