NaxsiRules.parse_rule() - Code Metrics - Inspection of "Improves a bit the readability of the code" - nbs-system/spike - Measure and Improve Code Quality continuously with Scrutinizer

Passed

Branch — master (7424d3)

by jvo

created 2016-03-17 13:22 UTC

NaxsiRules.parse_rule() F

↳ Parent: NaxsiRules

Complexity

Conditions

Size

Total Lines

Duplication

Lines	0
Ratio	0 %

Code Coverage

Tests	30
CRAP Score	14.1472

Metric	Value
cc	14
dl	0
loc	51
ccs	30
cts	33
cp	0.9091
crap	14.1472
rs	2.8289

How to fix Long Method Complexity

from time import strftime, localtime

from spike.model import db
from shlex import shlex


class NaxsiRules(db.Model):
    __bind_key__ = 'rules'
    __tablename__ = 'naxsi_rules'

    id = db.Column(db.Integer, primary_key=True)
    msg = db.Column(db.String(), nullable=False)
    detection = db.Column(db.String(1024), nullable=False)
    mz = db.Column(db.String(1024), nullable=False)
    score = db.Column(db.String(1024), nullable=False)
    sid = db.Column(db.Integer, nullable=False, unique=True)
    ruleset = db.Column(db.String(1024), nullable=False)
    rmks = db.Column(db.Text, nullable=True, server_default="")
    active = db.Column(db.Integer, nullable=False, server_default="1")
    negative = db.Column(db.Integer, nullable=False, server_default='0')
    timestamp = db.Column(db.Integer, nullable=False)

    mr_kw = ["MainRule", "BasicRule", "main_rule", "basic_rule"]
    static_mz = {"$ARGS_VAR", "$BODY_VAR", "$URL", "$HEADERS_VAR"}
    full_zones = {"ARGS", "BODY", "URL", "HEADERS", "FILE_EXT", "RAW_BODY"}
    rx_mz = {"$ARGS_VAR_X", "$BODY_VAR_X", "$URL_X", "$HEADERS_VAR_X"}
    sub_mz = list(static_mz) + list(full_zones) + list(rx_mz)

    def __init__(self, msg="", detection="", mz="", score="", sid=42000, ruleset="", rmks="", active=0, negative=0,
                 timestamp=0):
        self.msg = msg
        self.detection = detection
        self.mz = mz
        self.score = score
        self.sid = sid
        self.ruleset = ruleset
        self.rmks = rmks
        self.active = active
        self.negative = 1 if negative == 'checked' else 0
        self.timestamp = timestamp
        self.warnings = []
        self.error = []

    def fullstr(self):
        rdate = strftime("%F - %H:%M", localtime(float(str(self.timestamp))))
        rmks = "# ".join(self.rmks.strip().split("\n"))
        return "#\n# sid: {0} | date: {1}\n#\n# {2}\n#\n{3}".format(self.sid, rdate, rmks, self.__str__())

    def __str__(self):
        negate = 'negative' if self.negative == 1 else ''
        return 'MainRule {} "{}" "msg:{}" "mz:{}" "s:{}" id:{} ;'.format(
            negate, self.detection, self.msg, self.mz, self.score, self.sid)

    def explain(self):
        """ Return a string explaining the rule

         :return str: A textual explanation of the rule
         """
        translation = {'ARGS': 'argument', 'BODY': 'body', 'URL': 'url', 'HEADER': 'header'}
        explanation = 'The rule number <strong>%d</strong> is ' % self.sid
        if self.negative:
            explanation += '<strong>not</strong> '
        explanation += 'setting the '

        scores = []
        for score in self.score.split(','):
            scores.append('<strong>{0}</strong> to <strong>{1}</strong> '.format(*score.split(':', 1)))
        explanation += ', '.join(scores) + 'when it '
        if self.detection.startswith('str:'):
            explanation += 'finds the string <strong>{}</strong> '.format(self.detection[4:])
        else:
            explanation += 'matches the regexp <strong>{}</strong> '.format(self.detection[3:])

        zones = []
        for mz in self.mz.split('|'):
            if mz.startswith('$'):
                current_zone, arg = mz.split(":", 1)
                zone_name = "?"

                for translated_name in translation:  # translate zone names
                    if translated_name in current_zone:
                        zone_name = translation[translated_name]

                if "$URL" in current_zone:
                    regexp = "matching regex" if current_zone == "$URL_X" else ""
                    explanation += "on the URL {} '{}' ".format(regexp, arg)
                else:
                    regexp = "matching regex" if current_zone.endswith("_X") else ""
                    explanation += "in the var with name {} '{}' of {} ".format(regexp, arg, zone_name)
            else:
                zones.append('the <strong>{0}</strong>'.format(translation[mz]))
        return explanation

    def validate(self):
        self.__validate_matchzone(self.mz)
        self.__validate_id(self.sid)
        self.__validate_detection(self.detection)

        if not self.msg:
            self.warnings.append("Rule has no 'msg:'.")
        if not self.score:
            self.error.append("Rule has no score.")

    def __fail(self, msg):
        self.error.append(msg)
        return False

    # Bellow are parsers for specific parts of a rule

    def __validate_detection(self, p_str, label="", assign=False):
        p_str = label + p_str
        if not p_str.islower():
            self.warnings.append("detection {} is not lower-case. naxsi is case-insensitive".format(p_str))
        if p_str.startswith("str:") or p_str.startswith("rx:"):
            if assign is True:
                self.detection = p_str
        else:
            return self.__fail("detection {} is neither rx: or str:".format(p_str))
        return True

    def __validate_genericstr(self, p_str, label="", assign=False):
        if assign is False:
            return True
        if label == "s:":
            self.score = p_str
        elif label == "msg:":
            self.msg = p_str
        elif label == "negative":
            self.negative = 1
        elif label != "":
            return self.__fail("Unknown fragment {}".format(label+p_str))
        return True

    def __validate_matchzone(self, p_str, label="", assign=False):
        has_zone = False
        mz_state = set()
        for loc in p_str.split('|'):
            keyword, arg = loc, None
            if loc.startswith("$"):
                if loc.find(":") == -1:
                    return self.__fail("Missing 2nd part after ':' in {0}".format(loc))
                keyword, arg = loc.split(":")

            if keyword not in self.sub_mz:  # check if `keyword` is a valid keyword
                return self.__fail("'{0}' no a known sub-part of mz : {1}".format(keyword, self.sub_mz))

            mz_state.add(keyword)

            # verify that the rule doesn't attempt to target REGEX and STATIC _VAR/URL at the same time
            if len(self.rx_mz & mz_state) and len(self.static_mz & mz_state):
                return self.__fail("You can't mix static $* with regex $*_X ({})".format(', '.join(mz_state)))

            if arg and not arg.islower(): # just a gentle reminder
                self.warnings.append("{0} in {1} is not lowercase. naxsi is case-insensitive".format(arg, loc))

            # the rule targets an actual zone
            if keyword not in ["$URL", "$URL_X"] and keyword in (self.rx_mz | self.full_zones | self.static_mz):

                has_zone = True

        if has_zone is False:
            return self.__fail("The rule/whitelist doesn't target any zone.")

        if assign is True:
            self.mz = p_str

        return True

    def __validate_id(self, p_str, label="", assign=False):
        try:
            num = int(p_str)
            if num < 10000:
                self.warnings.append("rule IDs below 10k are reserved ({0})".format(num))
        except ValueError:
            self.error.append("id:{0} is not numeric".format(p_str))
            return False
        if assign is True:
            self.sid = num
        return True

    @staticmethod
    def splitter(s):

        lexer = shlex(s)
        lexer.whitespace_split = True
        return list(iter(lexer.get_token, ''))

    def parse_rule(self, full_str):
        """
        Parse and validate a full naxsi rule
        :param full_str: raw rule
        :return: [True|False, dict]
        """
        self.warnings = []
        self.error = []

        func_map = {"id:": self.__validate_id, "str:": self.__validate_detection,
                    "rx:": self.__validate_detection, "msg:": self.__validate_genericstr,
                    "mz:": self.__validate_matchzone, "negative": self.__validate_genericstr,
                    "s:": self.__validate_genericstr}
        ret = False
        split = self.splitter(full_str)  # parse string
        intersection = set(split).intersection(set(self.mr_kw))

        if not intersection:
            return self.__fail("No mainrule/basicrule keyword.")
        elif len(intersection) > 1:
            return self.__fail("Multiple mainrule/basicrule keywords.")

        split.remove(intersection.pop())  # remove the mainrule/basicrule keyword

        if ";" in split:
            split.remove(";")

        while split:  # iterate while there is data, as handlers can defer
            for keyword in split:
                orig_kw = keyword
                keyword = keyword.strip()

                if keyword.endswith(";"):  # remove semi-colons
                    keyword = keyword[:-1]
                if keyword.startswith(('"', "'")) and (keyword[0] == keyword[-1]):  # remove (double-)quotes
                    keyword = keyword[1:-1]

                for frag_kw in func_map:
                    ret = False
                    if keyword.startswith(frag_kw):
                        # parser funcs returns True/False
                        ret = func_map[frag_kw](keyword[len(frag_kw):], label=frag_kw, assign=True)
                        if ret is True:
                            split.remove(orig_kw)
                        else:
                            return self.__fail("parsing of element '{0}' failed.".format(keyword))
                        break

                if orig_kw in split and ret is not None:  # we have an item that wasn't successfully parsed
                    return False
        return True


1	1	from time import strftime, localtime
2
3	1	from spike.model import db
4	1	from shlex import shlex
5
6
7	1	class NaxsiRules(db.Model):
8	1	__bind_key__ = 'rules'
9	1	__tablename__ = 'naxsi_rules'
10
11	1	id = db.Column(db.Integer, primary_key=True)
12	1	msg = db.Column(db.String(), nullable=False)
13	1	detection = db.Column(db.String(1024), nullable=False)
14	1	mz = db.Column(db.String(1024), nullable=False)
15	1	score = db.Column(db.String(1024), nullable=False)
16	1	sid = db.Column(db.Integer, nullable=False, unique=True)
17	1	ruleset = db.Column(db.String(1024), nullable=False)
18	1	rmks = db.Column(db.Text, nullable=True, server_default="")
19	1	active = db.Column(db.Integer, nullable=False, server_default="1")
20	1	negative = db.Column(db.Integer, nullable=False, server_default='0')
21	1	timestamp = db.Column(db.Integer, nullable=False)
22
23	1	mr_kw = ["MainRule", "BasicRule", "main_rule", "basic_rule"]
24	1	static_mz = {"$ARGS_VAR", "$BODY_VAR", "$URL", "$HEADERS_VAR"}
25	1	full_zones = {"ARGS", "BODY", "URL", "HEADERS", "FILE_EXT", "RAW_BODY"}
26	1	rx_mz = {"$ARGS_VAR_X", "$BODY_VAR_X", "$URL_X", "$HEADERS_VAR_X"}
27	1	sub_mz = list(static_mz) + list(full_zones) + list(rx_mz)
28
29	1	def __init__(self, msg="", detection="", mz="", score="", sid=42000, ruleset="", rmks="", active=0, negative=0,
30		timestamp=0):
31	1	self.msg = msg
32	1	self.detection = detection
33	1	self.mz = mz
34	1	self.score = score
35	1	self.sid = sid
36	1	self.ruleset = ruleset
37	1	self.rmks = rmks
38	1	self.active = active
39	1	self.negative = 1 if negative == 'checked' else 0
40	1	self.timestamp = timestamp
41	1	self.warnings = []
42	1	self.error = []
43
44	1	def fullstr(self):
45	1	rdate = strftime("%F - %H:%M", localtime(float(str(self.timestamp))))
46	1	rmks = "# ".join(self.rmks.strip().split("\n"))
47	1	return "#\n# sid: {0} \| date: {1}\n#\n# {2}\n#\n{3}".format(self.sid, rdate, rmks, self.__str__())
48
49	1	def __str__(self):
50	1	negate = 'negative' if self.negative == 1 else ''
51	1	return 'MainRule {} "{}" "msg:{}" "mz:{}" "s:{}" id:{} ;'.format(
52		negate, self.detection, self.msg, self.mz, self.score, self.sid)
53
54	1	def explain(self):
55		""" Return a string explaining the rule
56
57		:return str: A textual explanation of the rule
58		"""
59	1	translation = {'ARGS': 'argument', 'BODY': 'body', 'URL': 'url', 'HEADER': 'header'}
60	1	explanation = 'The rule number <strong>%d</strong> is ' % self.sid
61	1	if self.negative:
62		explanation += '<strong>not</strong> '
63	1	explanation += 'setting the '
64
65	1	scores = []
66	1	for score in self.score.split(','):
67	1	scores.append('<strong>{0}</strong> to <strong>{1}</strong> '.format(*score.split(':', 1)))
68	1	explanation += ', '.join(scores) + 'when it '
69	1	if self.detection.startswith('str:'):
70	1	explanation += 'finds the string <strong>{}</strong> '.format(self.detection[4:])
71		else:
72		explanation += 'matches the regexp <strong>{}</strong> '.format(self.detection[3:])
73
74	1	zones = []
75	1	for mz in self.mz.split('\|'):
76	1	if mz.startswith('$'):
77		current_zone, arg = mz.split(":", 1)
78		zone_name = "?"
79
80		for translated_name in translation: # translate zone names
81		if translated_name in current_zone:
82		zone_name = translation[translated_name]
83
84		if "$URL" in current_zone:
85		regexp = "matching regex" if current_zone == "$URL_X" else ""
86		explanation += "on the URL {} '{}' ".format(regexp, arg)
87		else:
88		regexp = "matching regex" if current_zone.endswith("_X") else ""
89		explanation += "in the var with name {} '{}' of {} ".format(regexp, arg, zone_name)
90		else:
91	1	zones.append('the <strong>{0}</strong>'.format(translation[mz]))
92	1	return explanation
93
94	1	def validate(self):
95	1	self.__validate_matchzone(self.mz)
96	1	self.__validate_id(self.sid)
97	1	self.__validate_detection(self.detection)
98
99	1	if not self.msg:
100		self.warnings.append("Rule has no 'msg:'.")
101	1	if not self.score:
102		self.error.append("Rule has no score.")
103
104	1	def __fail(self, msg):
105	1	self.error.append(msg)
106	1	return False
107
108		# Bellow are parsers for specific parts of a rule
109
110	1	def __validate_detection(self, p_str, label="", assign=False):
111	1	p_str = label + p_str
112	1	if not p_str.islower():
113	1	self.warnings.append("detection {} is not lower-case. naxsi is case-insensitive".format(p_str))
114	1	if p_str.startswith("str:") or p_str.startswith("rx:"):
115	1	if assign is True:
116	1	self.detection = p_str
117		else:
118	1	return self.__fail("detection {} is neither rx: or str:".format(p_str))
119	1	return True
120
121	1	def __validate_genericstr(self, p_str, label="", assign=False):
122	1	if assign is False:
123		return True
124	1	if label == "s:":
125	1	self.score = p_str
126	1	elif label == "msg:":
127	1	self.msg = p_str
128		elif label == "negative":
129		self.negative = 1
130		elif label != "":
131		return self.__fail("Unknown fragment {}".format(label+p_str))
132	1	return True
133
134	1	def __validate_matchzone(self, p_str, label="", assign=False):
135	1	has_zone = False
136	1	mz_state = set()
137	1	for loc in p_str.split('\|'):
138	1	keyword, arg = loc, None
139	1	if loc.startswith("$"):
140	1	if loc.find(":") == -1:
141		return self.__fail("Missing 2nd part after ':' in {0}".format(loc))
142	1	keyword, arg = loc.split(":")
143
144	1	if keyword not in self.sub_mz: # check if `keyword` is a valid keyword
145		return self.__fail("'{0}' no a known sub-part of mz : {1}".format(keyword, self.sub_mz))
146
147	1	mz_state.add(keyword)
148
149		# verify that the rule doesn't attempt to target REGEX and STATIC _VAR/URL at the same time
150	1	if len(self.rx_mz & mz_state) and len(self.static_mz & mz_state):
151	1	return self.__fail("You can't mix static $* with regex $*_X ({})".format(', '.join(mz_state)))
152
153	1	if arg and not arg.islower(): # just a gentle reminder
154	1	self.warnings.append("{0} in {1} is not lowercase. naxsi is case-insensitive".format(arg, loc))
155
156		# the rule targets an actual zone
157	1	if keyword not in ["$URL", "$URL_X"] and keyword in (self.rx_mz \| self.full_zones \| self.static_mz):
		0 ignored issues – show Unused Code Coding Style introduced 2016-03-15 16:16 UTC by Report Bug Copy Issue Report There is an unnecessary parenthesis after `in`. Loading history...
158	1	has_zone = True
159
160	1	if has_zone is False:
161		return self.__fail("The rule/whitelist doesn't target any zone.")
162
163	1	if assign is True:
164	1	self.mz = p_str
165
166	1	return True
167
168	1	def __validate_id(self, p_str, label="", assign=False):
169	1	try:
170	1	num = int(p_str)
171	1	if num < 10000:
172	1	self.warnings.append("rule IDs below 10k are reserved ({0})".format(num))
173		except ValueError:
174		self.error.append("id:{0} is not numeric".format(p_str))
175		return False
176	1	if assign is True:
177	1	self.sid = num
178	1	return True
179
180	1	@staticmethod
181		def splitter(s):
		0 ignored issues – show Coding Style Naming introduced 2016-03-16 11:23 UTC by Report Bug Copy Issue Report The name `s` does not conform to the argument naming conventions (`[a-z_][a-z0-9_]{1,30}$`). This check looks for invalid names for a range of different identifiers. You can set regular expressions to which the identifiers must conform if the defaults do not match your requirements. If your project includes a Pylint configuration file, the settings contained in that file take precedence. To find out more about Pylint, please refer to their site. Loading history...
182	1	lexer = shlex(s)
183	1	lexer.whitespace_split = True
184	1	return list(iter(lexer.get_token, ''))
185
186	1	def parse_rule(self, full_str):
187		"""
188		Parse and validate a full naxsi rule
189		:param full_str: raw rule
190		:return: [True\|False, dict]
191		"""
192	1	self.warnings = []
193	1	self.error = []
194
195	1	func_map = {"id:": self.__validate_id, "str:": self.__validate_detection,
196		"rx:": self.__validate_detection, "msg:": self.__validate_genericstr,
197		"mz:": self.__validate_matchzone, "negative": self.__validate_genericstr,
198		"s:": self.__validate_genericstr}
199	1	ret = False
200	1	split = self.splitter(full_str) # parse string
201	1	intersection = set(split).intersection(set(self.mr_kw))
202
203	1	if not intersection:
204		return self.__fail("No mainrule/basicrule keyword.")
205	1	elif len(intersection) > 1:
206		return self.__fail("Multiple mainrule/basicrule keywords.")
207
208	1	split.remove(intersection.pop()) # remove the mainrule/basicrule keyword
209
210	1	if ";" in split:
211	1	split.remove(";")
212
213	1	while split: # iterate while there is data, as handlers can defer
214	1	for keyword in split:
215	1	orig_kw = keyword
216	1	keyword = keyword.strip()
217
218	1	if keyword.endswith(";"): # remove semi-colons
219	1	keyword = keyword[:-1]
220	1	if keyword.startswith(('"', "'")) and (keyword[0] == keyword[-1]): # remove (double-)quotes
221	1	keyword = keyword[1:-1]
222
223	1	for frag_kw in func_map:
224	1	ret = False
225	1	if keyword.startswith(frag_kw):
226		# parser funcs returns True/False
227	1	ret = func_map[frag_kw](keyword[len(frag_kw):], label=frag_kw, assign=True)
228	1	if ret is True:
229	1	split.remove(orig_kw)
230		else:
231	1	return self.__fail("parsing of element '{0}' failed.".format(keyword))
232	1	break
233
234	1	if orig_kw in split and ret is not None: # we have an item that wasn't successfully parsed
235		return False
236		return True
237

nbs-system / spike

Branch — master (7424d3)

NaxsiRules.parse_rule() F

Complexity

Size

Duplication

Code Coverage

How to fix Long Method Complexity

Long Method

Complexity

Duplication Side-by-Side

Filter issues like