path: root/peekaboo/ruleset/rules.py

###############################################################################
#                                                                             #
# Peekaboo Extended Email Attachment Behavior Observation Owl                 #
#                                                                             #
# ruleset/                                                                    #
#         rules.py                                                            #
###############################################################################
#                                                                             #
# Copyright (C) 2016-2019  science + computing ag                             #
#                                                                             #
# This program is free software: you can redistribute it and/or modify        #
# it under the terms of the GNU General Public License as published by        #
# the Free Software Foundation, either version 3 of the License, or (at       #
# your option) any later version.                                             #
#                                                                             #
# This program is distributed in the hope that it will be useful, but         #
# WITHOUT ANY WARRANTY; without even the implied warranty of                  #
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU           #
# General Public License for more details.                                    #
#                                                                             #
# You should have received a copy of the GNU General Public License           #
# along with this program.  If not, see <http://www.gnu.org/licenses/>.       #
#                                                                             #
###############################################################################

""" Classes implementing the Ruleset """


import re
import logging
from peekaboo.ruleset import Result, RuleResult
from peekaboo.exceptions import PeekabooAnalysisDeferred, \
        CuckooSubmitFailedException, PeekabooRulesetConfigError


logger = logging.getLogger(__name__)


class Rule(object):
    """ This is the base class for all rules. It provides common infrastructure
    such as resources that can be used by the rules (configuration, database
    connection) or helper functions. """
    rule_name = 'unimplemented'

    def __init__(self, config=None, db_con=None):
        """ Initialize common configuration and resources """
        self.db_con = db_con
        self.config = config

        # initialise and validate configuration
        self.get_config()

    def result(self, result, reason, further_analysis):
        """ Construct a RuleResult for returning to the engine. """
        return RuleResult(self.rule_name, result=result, reason=reason,
                          further_analysis=further_analysis)

    def evaluate(self, sample):
        """ Evaluate a rule against a sample.

        @param sample: The sample to evaluate.
        @returns: RuleResult containing verdict, reason, source of this
                  assessment (i.e. the rule's name) and whether to continue
                  analysis or not.
        """
        raise NotImplementedError

    def get_config(self):
        """ Extract this rule's configuration out of the ruleset configuration
        object given at creation. To be overridden by child classes if they
        have configuration options. """
        # pass

    # the following getters are somewhat boilerplate but unavoidable for now.
    # They serve the purpose of keeping config access specifics out of rules for
    # the sake of readablility.
    def get_config_value(self, getter, option, *args, **kwargs):
        """ Get a configuation value for this rule from the ruleset
        configuration. Getter routine and option name to be provided by caller.
        The rule's name is always used as configuration section name.

        @param getter: getter routine to use
        @type getter: getter method of PeekabooConfigParser
        @param option: name of option to read
        @type option: string
        @param args, kwargs: additional arguments passed to the getter routine,
                             such as fallback.

        @returns: configuration value read from config
        """
        # additional common logic to go here
        return getter(self.rule_name, option, *args, **kwargs)

    def get_config_int(self, option, default=None):
        """ Get an integer from the ruleset configuration. See get_config_value
        for parameters. """
        return self.get_config_value(
            self.config.getint, option, fallback=default)

    def get_config_float(self, option, default=None):
        """ Get a float from the ruleset configuration. See get_config_value
        for parameters. """
        return self.get_config_value(
            self.config.getfloat, option, fallback=default)

    def get_config_list(self, option, default=None):
        """ Get a list from the ruleset configuration. See get_config_value
        for parameters. """
        return self.get_config_value(
            self.config.getlist, option, fallback=default)

    def get_config_relist(self, option, default=None):
        """ Get a list of compiled regular expressions from the ruleset. See
        get_config_value for parameters. """
        return self.get_config_value(
            self.config.getrelist, option, fallback=default)


class KnownRule(Rule):
    """ A rule determining if a sample is known by looking at the database for
    a previous record of an identical sample sample. """
    rule_name = 'known'

    def evaluate(self, sample):
        """ Try to get information about the sample from the database. Return
        the old result and reason if found and advise the engine to stop
        processing. """
        sample_info = self.db_con.sample_info_fetch(sample)
        if sample_info:
            return self.result(sample_info.result, sample_info.reason, False)

        return self.result(Result.unknown,
                           _("File is not yet known to the system"),
                           True)


class FileLargerThanRule(Rule):
    """ A rule determining by file size whether a sample can be harmful at all.
    """
    rule_name = 'file_larger_than'

    def get_config(self):
        self.size_threshold = self.get_config_int('bytes', 5)

    def evaluate(self, sample):
        """ Evaluate whether the sample is larger than a certain threshold.
        Advise the engine to stop processing if the size is below the
        threshold. """
        try:
            sample_size = sample.file_size
        except OSError as oserr:
            return self.result(
                Result.failed,
                _("Failure to determine sample file size: %s") % oserr,
                False)

        if sample_size > self.size_threshold:
            return self.result(Result.unknown,
                               _("File has more than %d bytes")
                               % self.size_threshold,
                               True)

        return self.result(
            Result.ignored,
            _("File is only %d bytes long") % sample_size,
            False)


class FileTypeOnWhitelistRule(Rule):
    """ A rule checking whether the known file type(s) of the sample are on a
    whitelist. """
    rule_name = 'file_type_on_whitelist'

    def get_config(self):
        whitelist = self.get_config_list('whitelist')
        if not whitelist:
            raise PeekabooRulesetConfigError(
                "Empty whitelist, check %s rule config." % self.rule_name)

        self.whitelist = set(whitelist)

    def evaluate(self, sample):
        """ Ignore the file only if *all* of its mime types are on the
        whitelist and we could determine at least one. """
        if sample.mimetypes and sample.mimetypes.issubset(self.whitelist):
            return self.result(Result.ignored,
                               _("File type is on whitelist"),
                               False)

        return self.result(Result.unknown,
                           _("File type is not on whitelist"),
                           True)


class FileTypeOnGreylistRule(Rule):
    """ A rule checking whether any of the sample's known file types are on a
    greylist, i.e. enabled for analysis. """
    rule_name = 'file_type_on_greylist'

    def get_config(self):
        greylist = self.get_config_list('greylist')
        if not greylist:
            raise PeekabooRulesetConfigError(
                "Empty greylist, check %s rule config." % self.rule_name)

        self.greylist = set(greylist)

    def evaluate(self, sample):
        """ Continue analysis if any of the sample's MIME types are on the
        greylist or in case we don't have one. """
        if not sample.mimetypes or sample.mimetypes.intersection(self.greylist):
            return self.result(Result.unknown,
                               _("File type is on the list of types to "
                                 "analyze"),
                               True)

        return self.result(Result.unknown,
                           _("File type is not on the list of types to "
                             "analyse (%s)") % sample.mimetypes,
                           False)


class OfficeMacroRule(Rule):
    """ A rule checking the sample for Office macros. """
    rule_name = 'office_macro'

    def evaluate(self, sample):
        """ Report the sample as bad if it contains a macro. """
        if sample.office_macros:
            return self.result(Result.bad,
                               _("The file contains an Office macro"),
                               False)

        return self.result(Result.unknown,
                           _("The file does not contain a recognizable "
                             "Office macro"),
                           True)


class CuckooRule(Rule):
    """ A common base class for rules that evaluate the Cuckoo report. """
    def evaluate(self, sample):
        """ If a report is present for the sample in question we call method
        evaluate_report() implemented by subclasses to evaluate it for
        findings. Otherwise we submit the sample to Cuckoo and raise
        PeekabooAnalysisDeferred to abort the current run of the ruleset
        until the report arrives. If submission to Cuckoo fails we will
        ourselves report the sample as failed.

        @param sample: The sample to evaluate.
        @raises PeekabooAnalysisDeferred: if the sample was submitted to Cuckoo
        @returns: RuleResult containing verdict.
        """
        report = sample.cuckoo_report
        if report is None:
            try:
                job_id = sample.submit_to_cuckoo()
            except CuckooSubmitFailedException as failed:
                logger.error("Submit to Cuckoo failed: %s", failed)
                # exception message intentionally not present in message
                # delivered back to client as to not disclose internal
                # information, should request user to contact admin instead
                return self.result(
                    Result.failed,
                    _("Behavioral analysis by Cuckoo has produced an error "
                      "and did not finish successfully"),
                    False)

            logger.info('Sample submitted to Cuckoo. Job ID: %s. '
                        'Sample: %s', job_id, sample)
            raise PeekabooAnalysisDeferred()

        # call report evaluation function if we get here
        return self.evaluate_report(report)

    def evaluate_report(self, report):
        """ Evaluate a Cuckoo report.

        @param report: The Cuckoo report.
        @returns: RuleResult containing verdict.
        """
        raise NotImplementedError


class CuckooEvilSigRule(CuckooRule):
    """ A rule evaluating the signatures from the Cuckoo report against a list
    of signatures considered bad. """
    rule_name = 'cuckoo_evil_sig'

    def get_config(self):
        # list all installed signatures
        # grep -o "description.*" -R . ~/cuckoo2.0/modules/signatures/
        self.bad_sigs = self.get_config_relist('signature')
        if not self.bad_sigs:
            raise PeekabooRulesetConfigError(
                "Empty bad signature list, check %s rule config." %
                self.rule_name)

    def evaluate_report(self, report):
        """ Evaluate the sample against signatures that if matched mark a
        sample as bad. """
        # look through matched signatures
        sigs = []
        for descr in report.signatures:
            logger.debug(descr['description'])
            sigs.append(descr['description'])

        # check if there is a "bad" signatures and return bad
        matched_bad_sigs = []
        for bad_sig in self.bad_sigs:
            # iterate over each sig individually to allow regexes to use
            # anchors such as ^ and $ and avoid mismatches, e.g. by ['foo',
            # 'bar'] being stringified to "['foo', 'bar']" and matching
            # /fo.*ar/.
            for sig in sigs:
                match = re.search(bad_sig, sig)
                if match:
                    matched_bad_sigs.append(sig)

        if not matched_bad_sigs:
            return self.result(Result.unknown,
                               _("No signature suggesting malware detected"),
                               True)

        matched = ''.ljust(8).join(["%s\n" % s for s in matched_bad_sigs])
        return self.result(Result.bad,
                           _("The following signatures have been recognized: "
                             "%s") % matched,
                           False)


class CuckooScoreRule(CuckooRule):
    """ A rule checking the score reported by Cuckoo against a configurable
    threshold. """
    rule_name = 'cuckoo_score'

    def get_config(self):
        self.score_threshold = self.get_config_float('higher_than', 4.0)

    def evaluate_report(self, report):
        """ Evaluate the score reported by Cuckoo against the threshold from
        the configuration and report sample as bad if above. """

        if report.score >= self.score_threshold:
            return self.result(