diff --git a/suricata/update/category.py b/suricata/update/category.py new file mode 100644 index 00000000..8bf74544 --- /dev/null +++ b/suricata/update/category.py @@ -0,0 +1,157 @@ +# Copyright (C) 2017-2019 Open Information Security Foundation +# Copyright (c) 2020 Michael Schem +# +# You can copy, redistribute or modify this Program under the terms of +# the GNU General Public License version 2 as published by the Free +# Software Foundation. +# +# 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 +# version 2 along with this program; if not, write to the Free Software +# Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA +# 02110-1301, USA. + +""" Module for parsing categories.txt files. + +Parsing is done using regular expressions and the job of the module is +to do its best at parsing out fields of interest from the categories file +rather than perform a sanity check. + +""" + +from __future__ import print_function + +from suricata.update import fileparser + +import io +import re +import logging + +logger = logging.getLogger(__name__) + +# Compile a re pattern for basic iprep directive matching +category_pattern = re.compile(r"^(?P#)*[\s#]*" + r"(?P\d+)," + r"(?P\w+)," + r"(?P.*$)") + +class Category(dict): + """ Class representing an iprep category + + The category class also acts like a dictionary. + + Dictionary fields: + + - **enabled**: True if the category is enabled (uncommented), false is + disabled (commented) + - **id**: The maximum value for the category id is hard coded at 60 + currently (Suricata 5.0.3). + - **short_name**: The shortname that refers to the category. + - **description**: A description of the category. + + :param enabled: Optional parameter to set the enabled state of the category + + """ + + def __init__(self, enabled=None): + dict.__init__(self) + self["enabled"] = enabled + self["id"] = None + self["short_name"] = None + self["description"] = None + + def __getattr__(self, name): + return self[name] + + @property + def id(self): + """ The ID of the category. + + :returns: An int ID of the category + :rtype: int + """ + return int(self["id"]) + + @property + def idstr(self): + """Return the gid and sid of the rule as a string formatted like: + '[id]'""" + return "[%s]" % str(self.id) + + def __str__(self): + """ The string representation of the category. + + If the category is disabled it will be returned as commented out. + """ + return self.format() + + def format(self): + return "{0}{1},{2},{3}".format(u"" if self["enabled"] else u"# ", + self['id'], + self['short_name'], + self['description']) + + +def parse(buf, group=None): + """ Parse a single Iprep category from a string buffer. + + :param buf: A string buffer containing a single Iprep category. + + :returns: An instance of a :py:class:`.Category` representing the parsed Iprep category + """ + + if type(buf) == type(b""): + buf = buf.decode("utf-8") + buf = buf.strip() + + m = category_pattern.match(buf) + if not m: + return None + + if m.group("enabled") == "#": + enabled = False + else: + enabled = True + + # header = m.group("header").strip() + + category = Category(enabled=enabled) + + category["id"] = int(m.group("id").strip()) + + if not 0 < category["id"] < 60: + logging.error("Category id of {0}, not valid. Id is required to be between 0 and 60.".format(category["id"])) + return None + + category["short_name"] = m.group("short_name").strip() + + category["description"] = m.group("description").strip() + + return category + + +def parse_fileobj(fileobj, group=None): + """ Parse multiple ipreps from a file like object. + + Note: At this time ipreps must exist on one line. + + :param fileobj: A file like object to parse rules from. + + :returns: A list of :py:class:`.Iprep` instances, one for each rule parsed + """ + return fileparser.parse_fileobj(fileobj, parse, group) + +def parse_file(filename, group=None): + """ Parse multiple ipreps from the provided filename. + + :param filename: Name of file to parse ipreps from + + :returns: A list of :py:class:`.Iprep` instances, one for each iprep parsed + """ + with io.open(filename, encoding="utf-8") as fileobj: + return parse_fileobj(fileobj, group) + diff --git a/suricata/update/commands/addsource.py b/suricata/update/commands/addsource.py index a87095c0..14b1fe9e 100644 --- a/suricata/update/commands/addsource.py +++ b/suricata/update/commands/addsource.py @@ -37,6 +37,8 @@ def register(parser): help="Additional HTTP header to add to requests") parser.add_argument("--no-checksum", action="store_false", help="Skips downloading the checksum URL") + parser.add_argument("--iprep", action="store_true", + help="Identifies source as an IPRep Source") parser.set_defaults(func=add_source) @@ -67,6 +69,8 @@ def add_source(): header = args.http_header if args.http_header else None + is_iprep = args.iprep + source_config = sources.SourceConfiguration( - name, header=header, url=url, checksum=checksum) + name, header=header, url=url, checksum=checksum, is_iprep=is_iprep) sources.save_source_config(source_config) diff --git a/suricata/update/config.py b/suricata/update/config.py index 0aafc96f..cbf26975 100644 --- a/suricata/update/config.py +++ b/suricata/update/config.py @@ -49,6 +49,7 @@ DROP_CONF_KEY = "drop-conf" LOCAL_CONF_KEY = "local" OUTPUT_KEY = "output" +IPREP_OUTPUT_KEY = "iprep" DIST_RULE_DIRECTORY_KEY = "dist-rule-directory" if has_defaults: @@ -83,6 +84,7 @@ # The default file patterns to ignore. "ignore": [ "*deleted.rules", + ".*" ], } @@ -136,6 +138,12 @@ def get_output_dir(): return _config[OUTPUT_KEY] return os.path.join(get_state_dir(), "rules") +def get_iprep_dir(): + """Get the rule output directory.""" + if IPREP_OUTPUT_KEY in _config: + return _config[IPREP_OUTPUT_KEY] + return os.path.join(get_state_dir(), "iprep") + def args(): """Return sthe parsed argument object.""" return _args diff --git a/suricata/update/fileparser.py b/suricata/update/fileparser.py new file mode 100644 index 00000000..8c864267 --- /dev/null +++ b/suricata/update/fileparser.py @@ -0,0 +1,77 @@ +# Copyright (C) 2017-2019 Open Information Security Foundation +# Copyright (c) 2020 Michael Schem +# +# You can copy, redistribute or modify this Program under the terms of +# the GNU General Public License version 2 as published by the Free +# Software Foundation. +# +# 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 +# version 2 along with this program; if not, write to the Free Software +# Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA +# 02110-1301, USA. + +""" Module for files with either rules, ipreps, or category files. + +Parse funcitons + +""" + +from __future__ import print_function + +import sys +import re +import logging +import io + + +logger = logging.getLogger(__name__) + + +def parse_fileobj(fileobj, parse, group=None): + """ Parse multiple line based items from a file like object. + + Note: At this point items must exist on one line + + :param fileobj: A file like object to parse items from. + :param parse: A function used to parse items. + + :returns: A list of :py:class:`.Rule`, :py:class:`.Iprep`, or :py:class:`.Category` + instances depending on what parsing function is passed in. + """ + items = [] + buf = "" + for line in fileobj: + try: + if type(line) == type(b""): + line = line.decode() + except: + pass + if line.rstrip().endswith("\\"): + buf = "%s%s " % (buf, line.rstrip()[0:-1]) + continue + buf = buf + line + try: + item = parse(buf, group) + if item: + items.append(item) + except Exception as err: + logger.error("Failed to parse: %s: %s", buf.rstrip(), err) + buf = "" + return items + + +def parse_file(filename, group=None): + """ Parse multiple rules from the provided filename. + + :param filename: Name of file to parse ipreps from + + :returns: A list of .rules files or :py:class:`.Iprep` instances + for .list files, one for each rule parsed + """ + with io.open(filename, encoding="utf-8") as fileobj: + return parse_fileobj(fileobj, group) diff --git a/suricata/update/iprep.py b/suricata/update/iprep.py new file mode 100644 index 00000000..604e9312 --- /dev/null +++ b/suricata/update/iprep.py @@ -0,0 +1,183 @@ +# Copyright (C) 2017-2019 Open Information Security Foundation +# Copyright (c) 2011 Jason Ish +# +# You can copy, redistribute or modify this Program under the terms of +# the GNU General Public License version 2 as published by the Free +# Software Foundation. +# +# 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 +# version 2 along with this program; if not, write to the Free Software +# Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA +# 02110-1301, USA. + +""" Module for parsing Snort-like rules. + +Parsing is done using regular expressions and the job of this module +is to do its best at parsing out fields of interest from the rule +rather than perform a sanity check. + +The methods that parse multiple rules for a provided input +(parse_file, parse_fileobj) return a list of rules instead of dict +keyed by ID as its not the job of this module to detect or deal with +duplicate signature IDs. +""" + +from __future__ import print_function + +from suricata.update import fileparser + +import sys +import re +import logging +import io + +logger = logging.getLogger(__name__) + +# Compile a re pattern for basic iprep directive matching +iprep_pattern = re.compile(r"^(?P#)*[\s#]*" + r"(?P" + # r"(?P
[^()]+)" + r"(?P\d+\.\d+\.\d+\.\d+)," + r"(?P\d+)," + r"(?P\d+)" + r"$)") + +ip_pattern = re.compile(r"^(25[0-5]|2[0-4][0-9]|[0-1]?[0-9][0-9]?)\.(" + r"25[0-5]|2[0-4][0-9]|[0-1]?[0-9][0-9]?)\.(" + r"25[0-5]|2[0-4][0-9]|[0-1]?[0-9][0-9]?)\.(" + r"25[0-5]|2[0-4][0-9]|[0-1]?[0-9][0-9]?)$") + +class Iprep(dict): + """ Class representing an iprep directive + + The iprep directive is a class that also acts like a dictionary. + + Dictionary fields: + + - **group**: The group the rule belongs to, typically the filename. + - **enabled**: True if rule is enabled (uncommented), False is + disabled (commented) + - **ip**: The ip address of the directive. + - **category**: The IP is an IPv4 address in the quad-dotted notation or an IPv6 address. Both IP types support + networks in CIDR notation. + - **reputation_score**: The reputation score is the confidence that this IP is in the specified category, + represented by a number between 1 and 127 (0 means no data). + + :param enabled: Optional parameter to set the enabled state of the rule + :param group: Optional parameter to set the group (filename) of the rule + + """ + + def __init__(self, enabled=None, group=None): + dict.__init__(self) + self["enabled"] = enabled + self['ip'] = None + self['category'] = None + self['reputation_score'] = None + self["group"] = group + + def __getattr__(self, name): + return self[name] + + @property + def id(self): + """ The ID of the directive. + + :returns: A tuple (ip, category, reputation_score) representing the ID of the directive + :rtype: A tuple of 2 ints and one str + """ + return (str(self.ip), int(self.category), int(self.reputation_score)) + + @property + def idstr(self): + """Return the ip, category, and reputation score of the iprep directive as a string formatted like: + '[IP:CAT:REP]'""" + return "[%s:%s:%s]" % (str(self.ip), str(self.category), str(self.reputation_score)) + + def __str__(self): + """ The string representation of the directive. + + If the directive is disabled it will be returned as commented out. + """ + return self.format() + + def format(self): + return "{0}{1},{2},{3}".format(u"" if self["enabled"] else u"# ", + self['ip'], + self['category'], + self['reputation_score']) + + +class BadIprepError(Exception): + """Raises exception when an invalid Iprep is created""" + + +def parse(buf, group=None): + """ Parse a single iprep directive from a string buffer. + + :param buf: A string buffer containing a single iprep derective + + :returns: An instance of a :py:class:`.Iprep` representing the parsed iprep directive + """ + + if type(buf) == type(b""): + buf = buf.decode("utf-8") + buf = buf.strip() + + m = iprep_pattern.match(buf) + if not m: + return None + + if m.group("enabled") == "#": + enabled = False + else: + enabled = True + + # header = m.group("header").strip() + + iprep = Iprep(enabled=enabled, group=group) + + iprep['ip'] = m.group('ip') + + if not ip_pattern.search(iprep['ip']): + logging.error("Invalid iprep IP address. {0}".format(iprep['ip'])) + raise BadIprepError + + iprep['category'] = int(m.group('category')) + + iprep['reputation_score'] = int(m.group('reputation_score')) + + if not 0 <= iprep['reputation_score'] <= 127: + raise BadIprepError("Invalid reputation score of {0}".format(iprep['reputation_score'])) + + iprep["raw"] = m.group("raw").strip() + + return iprep + + +def parse_fileobj(fileobj, group=None): + """ Parse multiple ipreps from a file like object. + + Note: At this time ipreps must exist on one line. + + :param fileobj: A file like object to parse rules from. + + :returns: A list of :py:class:`.Iprep` instances, one for each rule parsed + """ + return fileparser.parse_fileobj(fileobj, parse, group) + + +def parse_file(filename, group=None): + """ Parse multiple ipreps from the provided filename. + + :param filename: Name of file to parse ipreps from + + :returns: A list of :py:class:`.Iprep` instances, one for each iprep parsed + """ + with io.open(filename, encoding="utf-8") as fileobj: + return parse_fileobj(fileobj, group) diff --git a/suricata/update/main.py b/suricata/update/main.py index 41cf0cb6..c3aaa0e6 100644 --- a/suricata/update/main.py +++ b/suricata/update/main.py @@ -59,6 +59,8 @@ notes, parsers, rule as rule_mod, + iprep as iprep_mod, + category as category_mod, sources, util, matchers as matchers_mod @@ -93,6 +95,8 @@ # The default filename to use for the output rule file. This is a # single file concatenating all input rule files together. DEFAULT_OUTPUT_RULE_FILENAME = "suricata.rules" +DEFAULT_OUTPUT_IPREP_FILENAME = "reputation.list" +DEFAULT_OUTPUT_CATEGORIES_FILENAME = "categories.txt" INDEX_EXPIRATION_TIME = 60 * 60 * 24 * 14 @@ -454,7 +458,6 @@ def handle_filehash_files(rule, dep_files, fhash): logger.error("%s file %s was not found" % (fhash, filehash_fname)) def write_merged(filename, rulemap, dep_files): - if not args.quiet: # List of rule IDs that have been added. added = [] @@ -476,11 +479,11 @@ def write_merged(filename, rulemap, dep_files): if not key in oldset: added.append(key) - enabled = len([rule for rule in rulemap.values() if rule.enabled]) + enabled = len([rule for rule in map.values() if rule.enabled]) logger.info("Writing rules to %s: total: %d; enabled: %d; " "added: %d; removed %d; modified: %d" % ( filename, - len(rulemap), + len(map), enabled, len(added), len(removed), @@ -607,6 +610,26 @@ def build_rule_map(rules): return rulemap +def build_iprep_map(ipreps): + """ Turn a list of ipreps into a mapping of ipreps """ + iprepmap = {} + + for iprep in ipreps: + if iprep.id not in iprepmap: + iprepmap[iprep.id] = iprep + + return iprepmap + +def build_categories_map(categories): + """ Turn a list of categories into a mapping of categories """ + categorymap = {} + + for category in categories: + if category.id not in categorymap: + categorymap[category.id] = category + + return categorymap + def dump_sample_configs(): for filename in configs.filenames: @@ -1109,9 +1132,10 @@ def _main(): except subprocess.CalledProcessError: return 1 - # Disable rule that are for app-layers that are not enabled. if suriconf: + reputation_files = [] for key in suriconf.keys(): + # Disable rule that are for app-layers that are not enabled. m = re.match("app-layer\.protocols\.([^\.]+)\.enabled", key) if m: proto = m.group(1) @@ -1126,6 +1150,10 @@ def _main(): if not "RUST" in suriconf.build_info["features"]: logger.info("Disabling rules for protocol {}".format(proto)) disable_matchers.append(matchers_mod.ProtoRuleMatcher(proto)) + # get reputaiton files + m = re.match("reputation-files\.(?P\d+)", key) + if m: + reputation_files.append(suriconf.conf.get(m.group())) # Check that the cache directory exists and is writable. if not os.path.exists(config.get_cache_dir()): @@ -1142,23 +1170,39 @@ def _main(): load_dist_rules(files) rules = [] + ipreps = [] + categories = [] classification_files = [] dep_files = {} for entry in sorted(files, key = lambda e: e.filename): + # Classification.config file if "classification.config" in entry.filename: classification_files.append((entry.filename, entry.content)) continue + # *.rules files if not entry.filename.endswith(".rules"): dep_files.update({entry.filename: entry.content}) continue + # Ignore files if ignore_file(config.get("ignore"), entry.filename): logger.info("Ignoring file {}".format(entry.filename)) continue + # IPREP Categories file + if entry.filename.endswith(DEFAULT_OUTPUT_CATEGORIES_FILENAME): + categories += category_mod.parse_fileobj(io.BytesIO(files[entry.filename]), entry.filename) + continue + # IPREP List files + if entry.filename.endswith(".list"): + ipreps += iprep_mod.parse_fileobj(io.BytesIO(files[entry.filename]), entry.filename) + continue logger.debug("Parsing {}".format(entry.filename)) rules += rule_mod.parse_fileobj(io.BytesIO(entry.content), entry.filename) rulemap = build_rule_map(rules) + categorymap = build_categories_map(categories) + iprepmap = build_iprep_map(ipreps) logger.info("Loaded %d rules." % (len(rules))) + logger.info("Loaded %d iprep directives." % (len(ipreps))) # Counts of user enabled and modified rules. enable_count = 0 @@ -1214,12 +1258,17 @@ def _main(): # Check that output directory exists, creating it if needed. check_output_directory(config.get_output_dir()) + check_output_directory(config.get_iprep_dir()) # Check that output directory is writable. if not os.access(config.get_output_dir(), os.W_OK): logger.error( "Output directory is not writable: %s", config.get_output_dir()) return 1 + if not os.access(config.get_iprep_dir(), os.W_OK): + logger.error( + "Output directory is not writable: %s", config.get_output_dir()) + return 1 # Backup the output directory. logger.info("Backing up current rules.") @@ -1231,8 +1280,19 @@ def _main(): # The default, write out a merged file. output_filename = os.path.join( config.get_output_dir(), DEFAULT_OUTPUT_RULE_FILENAME) + iprep_output_filename = os.path.join( + config.get_iprep_dir(), DEFAULT_OUTPUT_IPREP_FILENAME) + categories_output_filename = os.path.join( + config.get_iprep_dir(), DEFAULT_OUTPUT_CATEGORIES_FILENAME) + file_tracker.add(output_filename) + file_tracker.add(iprep_output_filename) + file_tracker.add(categories_output_filename) + write_merged(os.path.join(output_filename), rulemap, dep_files) + write_merged(os.path.join(iprep_output_filename), iprepmap) + write_merged(os.path.join(categories_output_filename), categorymap) + else: for filename in files: file_tracker.add( diff --git a/suricata/update/rule.py b/suricata/update/rule.py index 42c673e9..d4b0bd7a 100644 --- a/suricata/update/rule.py +++ b/suricata/update/rule.py @@ -29,6 +29,8 @@ from __future__ import print_function +from suricata.update import fileparser + import sys import re import logging @@ -43,6 +45,7 @@ r"\((?P.*)\)" r"$)") + # Rule actions we expect to see. actions = ( "alert", "log", "pass", "activate", "dynamic", "drop", "reject", "sdrop") @@ -52,6 +55,7 @@ class NoEndOfOptionError(Exception): missing.""" pass + class Rule(dict): """Class representing a rule. @@ -302,6 +306,7 @@ def parse(buf, group=None): return rule + def parse_fileobj(fileobj, group=None): """ Parse multiple rules from a file like object. @@ -311,26 +316,7 @@ def parse_fileobj(fileobj, group=None): :returns: A list of :py:class:`.Rule` instances, one for each rule parsed """ - rules = [] - buf = "" - for line in fileobj: - try: - if type(line) == type(b""): - line = line.decode() - except: - pass - if line.rstrip().endswith("\\"): - buf = "%s%s " % (buf, line.rstrip()[0:-1]) - continue - buf = buf + line - try: - rule = parse(buf, group) - if rule: - rules.append(rule) - except Exception as err: - logger.error("Failed to parse rule: %s: %s", buf.rstrip(), err) - buf = "" - return rules + return fileparser.parse_fileobj(fileobj, parse, group) def parse_file(filename, group=None): """ Parse multiple rules from the provided filename. diff --git a/suricata/update/sources.py b/suricata/update/sources.py index 651a4d60..c8a00268 100644 --- a/suricata/update/sources.py +++ b/suricata/update/sources.py @@ -87,12 +87,13 @@ def save_source_config(source_config): class SourceConfiguration: def __init__(self, name, header=None, url=None, - params={}, checksum=True): + params={}, checksum=True, is_iprep=False): self.name = name self.url = url self.params = params self.header = header self.checksum = checksum + self.is_iprep = is_iprep def dict(self): d = { @@ -106,6 +107,8 @@ def dict(self): d["http-header"] = self.header if self.checksum: d["checksum"] = self.checksum + if self.is_iprep: + d["is_iprep"] = self.is_iprep return d class Index: diff --git a/suricata/update/test.py b/suricata/update/test.py new file mode 100644 index 00000000..acde6040 --- /dev/null +++ b/suricata/update/test.py @@ -0,0 +1,3 @@ +from main import ThresholdProcessor + +ThresholdProcessor.replace() \ No newline at end of file diff --git a/tests/test_category.py b/tests/test_category.py new file mode 100644 index 00000000..60a77ee7 --- /dev/null +++ b/tests/test_category.py @@ -0,0 +1,44 @@ +# Copyright (C) 2017 Open Information Security Foundation +# Copyright (c) 2011-2020 Michael Schem +# +# You can copy, redistribute or modify this Program under the terms of +# the GNU General Public License version 2 as published by the Free +# Software Foundation. +# +# 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 +# version 2 along with this program; if not, write to the Free Software +# Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA +# 02110-1301, USA. + +from __future__ import print_function + +import unittest + +import suricata.update.category + +class CategoryTestCase(unittest.TestCase): + + def test_parse1(self): + category = suricata.update.category.parse("3,IntelData,Threat Intelligence Data") + self.assertEqual(category.enabled, True) + self.assertEqual(category.id, 3) + self.assertEqual(category.short_name, "IntelData") + self.assertEqual(category.description, "Threat Intelligence Data") + self.assertEqual(category.idstr, "[3]") + self.assertEqual(str(category), "3,IntelData,Threat Intelligence Data") + + def test_bad_id(self): + category = suricata.update.category.parse("61,IntelData,Threat Intelligence Data") + self.assertEqual(category, None) + + def test_disabled_category(self): + category = suricata.update.category.parse("# 3,IntelData,Threat Intelligence Data") + self.assertEqual(category.enabled, False) + + category = suricata.update.category.parse("#3,IntelData,Threat Intelligence Data") + self.assertEqual(category.enabled, False) diff --git a/tests/test_iprep.py b/tests/test_iprep.py new file mode 100644 index 00000000..437618f5 --- /dev/null +++ b/tests/test_iprep.py @@ -0,0 +1,77 @@ +# Copyright (C) 2017 Open Information Security Foundation +# Copyright (c) 2011-2013 Michael Schem +# +# You can copy, redistribute or modify this Program under the terms of +# the GNU General Public License version 2 as published by the Free +# Software Foundation. +# +# 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 +# version 2 along with this program; if not, write to the Free Software +# Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA +# 02110-1301, USA. + +from __future__ import print_function + +import sys +import unittest +import io +import tempfile + +import suricata.update.iprep + + +class RuleTestCase(unittest.TestCase): + + def test_parse_iprep_parse(self): + """ Test parsing an iprep directive.""" + iprep_string = u"52.0.161.90,3,1" + iprep = suricata.update.iprep.parse(iprep_string) + self.assertIsNotNone(iprep) + self.assertTrue(iprep['raw'] == "52.0.161.90,3,1") + self.assertTrue(iprep['ip'] == "52.0.161.90") + self.assertTrue(iprep['category'] == 3) + self.assertTrue(iprep['reputation_score'] == 1) + self.assertTrue(str(iprep) == iprep_string) + + def test_parse_disabled_iprep_parse(self): + """ Test parsing a disabled iprep directive.""" + iprep_string = u"# 52.0.161.90,3,127" + iprep = suricata.update.iprep.parse(iprep_string) + self.assertIsNotNone(iprep) + self.assertTrue(iprep['raw'] == "52.0.161.90,3,127") + self.assertTrue(iprep['ip'] == "52.0.161.90") + self.assertTrue(iprep['category'] == 3) + self.assertTrue(iprep['reputation_score'] == 127) + self.assertTrue(str(iprep) == iprep_string) + + def test_parse_bad_iprep_score(self): + """ Test parsing a iprep directive with a bad reputation_score """ + iprep_string = u"52.0.161.90,3,150" + self.assertRaises( + suricata.update.iprep.BadIprepError, + suricata.update.iprep.parse, iprep_string) + + def test_parse_bad_ip_addresss(self): + """ Test parsing of a iprep with a bad IP Address """ + bad_iprep_string = u"52.0.161.300,3,150" + self.assertRaises( + suricata.update.iprep.BadIprepError, + suricata.update.iprep.parse, bad_iprep_string + ) + + def test_parse_fileobj(self): + """ Test parsing a file like object containing ipreps """ + ipreps_buf = u"""52.0.161.90,3,125\n + 52.0.161.91,3,126\n + 52.0.161.92,3,127\n + """ + fileobj = io.StringIO() + fileobj.write(u"%s\n" % ipreps_buf) + fileobj.seek(0) + ipreps = suricata.update.iprep.parse_fileobj(fileobj,suricata.update.iprep.parse) + self.assertEqual(len(ipreps), 3) diff --git a/tests/test_main.py b/tests/test_main.py index 1425cd19..44f10bb2 100644 --- a/tests/test_main.py +++ b/tests/test_main.py @@ -22,6 +22,7 @@ import unittest import suricata.update.rule +import suricata.update.iprep from suricata.update import main import suricata.update.extract from suricata.update import matchers as matchers_mod @@ -119,6 +120,7 @@ def test_replace(self): class ModifyRuleFilterTestCase(unittest.TestCase): rule_string = """alert http $EXTERNAL_NET any -> $HOME_NET any (msg:"ET MALWARE Windows executable sent when remote host claims to send an image 2"; flow: established,from_server; content:"|0d 0a|Content-Type|3a| image/jpeg|0d 0a 0d 0a|MZ"; fast_pattern:12,20; classtype:trojan-activity; sid:2020757; rev:2;)""" + iprep_string = """52.0.161.90,3,120""" def test_id_match(self): rule0 = suricata.update.rule.parse(self.rule_string) diff --git a/tests/test_matchers.py b/tests/test_matchers.py index 8113c534..c54ac242 100644 --- a/tests/test_matchers.py +++ b/tests/test_matchers.py @@ -21,6 +21,7 @@ import unittest import suricata.update.rule +import suricata.update.iprep from suricata.update import main import suricata.update.extract from suricata.update import matchers as matchers_mod @@ -28,6 +29,7 @@ class GroupMatcherTestCase(unittest.TestCase): rule_string = """alert http $EXTERNAL_NET any -> $HOME_NET any (msg:"ET MALWARE Windows executable sent when remote host claims to send an image 2"; flow: established,from_server; content:"|0d 0a|Content-Type|3a| image/jpeg|0d 0a 0d 0a|MZ"; fast_pattern:12,20; classtype:trojan-activity; sid:2020757; rev:2;)""" + iprep_string = """52.0.161.90,3,120""" def test_match(self): rule = suricata.update.rule.parse(self.rule_string, "rules/malware.rules") @@ -42,9 +44,23 @@ def test_match(self): matcher.__class__, matchers_mod.GroupMatcher) self.assertTrue(matcher.match(rule)) + def test_iprep_match(self): + iprep = suricata.update.iprep.parse(self.iprep_string, "rules/test.list") + matcher = matchers_mod.parse_rule_match("group: test.list") + self.assertEqual( + matcher.__class__, matchers_mod.GroupMatcher) + self.assertTrue(matcher.match(iprep)) + + # Test match of just the group basename. + matcher = matchers_mod.parse_rule_match("group: test") + self.assertEqual( + matcher.__class__, matchers_mod.GroupMatcher) + self.assertTrue(matcher.match(iprep)) + class FilenameMatcherTestCase(unittest.TestCase): rule_string = """alert http $EXTERNAL_NET any -> $HOME_NET any (msg:"ET MALWARE Windows executable sent when remote host claims to send an image 2"; flow: established,from_server; content:"|0d 0a|Content-Type|3a| image/jpeg|0d 0a 0d 0a|MZ"; fast_pattern:12,20; classtype:trojan-activity; sid:2020757; rev:2;)""" + iprep_string = """52.0.161.90,3,120""" def test_match(self): rule = suricata.update.rule.parse(self.rule_string, "rules/trojan.rules") @@ -53,6 +69,13 @@ def test_match(self): matcher.__class__, matchers_mod.FilenameMatcher) self.assertTrue(matcher.match(rule)) + def test_iprep_match(self): + iprep = suricata.update.iprep.parse(self.iprep_string, "rules/test.list") + matcher = matchers_mod.parse_rule_match("filename: */test.list") + self.assertEqual( + matcher.__class__, matchers_mod.FilenameMatcher) + self.assertTrue(matcher.match(iprep)) + class LoadMatchersTestCase(unittest.TestCase): def test_trailing_comment(self): diff --git a/tests/test_rule.py b/tests/test_rule.py index c5808d82..ac9191b6 100644 --- a/tests/test_rule.py +++ b/tests/test_rule.py @@ -24,6 +24,7 @@ import suricata.update.rule import suricata.update.main +import suricata.update.fileparser class RuleTestCase(unittest.TestCase): @@ -79,7 +80,7 @@ def test_parse_fileobj(self): for i in range(2): fileobj.write(u"%s\n" % rule_buf) fileobj.seek(0) - rules = suricata.update.rule.parse_fileobj(fileobj) + rules = suricata.update.fileparser.parse_fileobj(fileobj, suricata.update.rule.parse) self.assertEqual(2, len(rules)) def test_parse_file(self): @@ -90,11 +91,11 @@ def test_parse_file(self): for i in range(2): tmp.write(("%s\n" % rule_buf).encode()) tmp.flush() - rules = suricata.update.rule.parse_file(tmp.name) + rules = suricata.update.fileparser.parse_file(tmp.name, suricata.update.rule.parse) self.assertEqual(2, len(rules)) def test_parse_file_with_unicode(self): - rules = suricata.update.rule.parse_file("./tests/rule-with-unicode.rules") + rules = suricata.update.fileparser.parse_file("./tests/rule-with-unicode.rules", suricata.update.rule.parse) def test_parse_decoder_rule(self): rule_string = """alert ( msg:"DECODE_NOT_IPV4_DGRAM"; sid:1; gid:116; rev:1; metadata:rule-type decode; classtype:protocol-command-decode;)""" @@ -106,7 +107,7 @@ def test_multiline_rule(self): alert dnp3 any any -> any any (msg:"SURICATA DNP3 Request flood detected"; \ app-layer-event:dnp3.flooded; sid:2200104; rev:1;) """ - rules = suricata.update.rule.parse_fileobj(io.StringIO(rule_string)) + rules = suricata.update.fileparser.parse_fileobj(io.StringIO(rule_string), suricata.update.rule.parse) self.assertEqual(len(rules), 1) def test_parse_nomsg(self):