1
|
|
|
from time import strftime, localtime |
2
|
|
|
import re |
3
|
|
|
|
4
|
|
|
from spike import create_app |
5
|
|
|
from spike.model import db |
6
|
|
|
from spike.model.naxsi_rules import NaxsiRules |
7
|
|
|
|
8
|
|
|
try: |
9
|
|
|
from urlparse import urlparse |
10
|
|
|
except ImportError: # python3 |
11
|
|
|
from urllib.parse import urlparse |
12
|
|
|
|
13
|
|
|
import unittest |
14
|
|
|
|
15
|
|
|
|
16
|
|
|
class FlaskrTestCase(unittest.TestCase): |
17
|
|
|
def setUp(self): |
18
|
|
|
app = create_app() |
19
|
|
|
db.init_app(app) |
20
|
|
|
app.config['TESTING'] = True |
21
|
|
|
self.app = app.test_client() |
22
|
|
|
self.created_rules = list() |
23
|
|
|
self.__create_rule() |
24
|
|
|
|
25
|
|
|
def tearDown(self): |
26
|
|
|
self.__delete_rule() |
27
|
|
|
|
28
|
|
|
def __create_rule(self): |
29
|
|
|
""" |
30
|
|
|
|
31
|
|
|
:return int: The id of the new rule |
32
|
|
|
""" |
33
|
|
|
current_sid = NaxsiRules.query.order_by(NaxsiRules.sid.desc()).first() |
34
|
|
|
current_sid = 1337 if current_sid is None else current_sid.sid + 1 |
35
|
|
|
|
36
|
|
|
db.session.add(NaxsiRules(u'POUET', 'str:test', u'BODY', u'$SQL:8', current_sid, u'WEB_APPS', |
37
|
|
|
u'f hqewifueiwf hueiwhf uiewh fiewh fhw', '1', True, 1457101045)) |
38
|
|
|
self.created_rules.append(current_sid) |
39
|
|
|
return int(current_sid) |
40
|
|
|
|
41
|
|
|
def __delete_rule(self, sid=None): |
42
|
|
|
if sid: |
43
|
|
|
db.session.delete(NaxsiRules.query.filter(sid == NaxsiRules.sid).first()) |
44
|
|
|
for sid in self.created_rules: |
45
|
|
|
_rule = NaxsiRules.query.filter(sid == NaxsiRules.sid).first() |
46
|
|
|
if _rule: |
47
|
|
|
db.session.delete(_rule) |
48
|
|
|
|
49
|
|
|
def test_index(self): |
50
|
|
|
rv = self.app.get('/', follow_redirects=True) |
51
|
|
|
self.assertEqual(rv.status_code, 200) |
52
|
|
|
self.assertIn('<title>SPIKE! - WAF Rules Builder</title>', rv.data) |
53
|
|
|
self.assertTrue(re.search(r'<h2>Naxsi - Rules \( \d+ \)</h2>', rv.data) is not None) |
54
|
|
|
|
55
|
|
|
def test_view(self): |
56
|
|
|
_rule = NaxsiRules.query.order_by(NaxsiRules.sid.desc()).first() |
57
|
|
|
rv = self.app.get('/rules/view/%d' % _rule.sid) |
58
|
|
|
self.assertEqual(rv.status_code, 200) |
59
|
|
|
|
60
|
|
|
_sid = _rule.sid + 1 |
61
|
|
|
rv = self.app.get('/rules/view/%d' % _sid) |
62
|
|
|
self.assertEqual(urlparse(rv.location).path, '/rules/') |
63
|
|
|
|
64
|
|
|
def test_new_rule(self): |
65
|
|
|
rv = self.app.get('/rules/new') |
66
|
|
|
self.assertEqual(rv.status_code, 200) |
67
|
|
|
|
68
|
|
|
data = { |
69
|
|
|
'msg': 'this is a test message', |
70
|
|
|
'detection': 'str:DETECTION', |
71
|
|
|
'mz': 'BODY', |
72
|
|
|
'custom_mz_val': '', |
73
|
|
|
'negative': 'checked', |
74
|
|
|
'score_$SQL': 8, |
75
|
|
|
'score': '$SQL', |
76
|
|
|
'rmks': 'this is a test remark', |
77
|
|
|
'ruleset': 'WEB_APPS' |
78
|
|
|
} |
79
|
|
|
rv = self.app.post('/rules/new', data=data, follow_redirects=True) |
80
|
|
|
|
81
|
|
|
_rule = NaxsiRules.query.order_by(NaxsiRules.sid.desc()).first() |
82
|
|
|
|
83
|
|
|
self.assertIn(('<li> - OK: created %d : %s</li>' % (_rule.sid, _rule.msg)), rv.data) |
84
|
|
|
self.assertEqual(_rule.msg, data['msg']) |
85
|
|
|
self.assertEqual(_rule.detection, data['detection']) |
86
|
|
|
self.assertEqual(_rule.mz, data['mz']) |
87
|
|
|
self.assertEqual(_rule.score, data['score'] + ':' + str(data['score_$SQL'])) |
88
|
|
|
self.assertEqual(_rule.rmks, data['rmks']) |
89
|
|
|
self.assertEqual(_rule.ruleset, data['ruleset']) |
90
|
|
|
db.session.delete(_rule) |
91
|
|
|
db.session.commit() |
92
|
|
|
|
93
|
|
|
def test_del_rule(self): |
94
|
|
|
_rule = NaxsiRules.query.order_by(NaxsiRules.sid.desc()).first() |
95
|
|
|
|
96
|
|
|
db.session.add(NaxsiRules(u'PIF', 'str:test', u'BODY', u'$SQL:8', _rule.sid + 1, u'WEB_APPS', |
97
|
|
|
u'f hqewifueiwf hueiwhf uiewh fiewh fhw', '1', True, 1457101045)) |
98
|
|
|
_sid = _rule.sid + 1 |
99
|
|
|
rv = self.app.get('/rules/del/%d' % _sid) |
100
|
|
|
self.assertEqual(rv.status_code, 302) |
101
|
|
|
|
102
|
|
|
_rule = NaxsiRules.query.order_by(NaxsiRules.sid.desc()).first() |
103
|
|
|
self.assertEqual(_rule.sid, _rule.sid) |
104
|
|
|
|
105
|
|
|
def test_plain_rule(self): |
106
|
|
|
_rule = NaxsiRules.query.order_by(NaxsiRules.sid.desc()).first() |
107
|
|
|
rv = self.app.get('/rules/plain/%d' % _rule.sid) |
108
|
|
|
self.assertEqual(rv.status_code, 200) |
109
|
|
|
rdate = strftime("%F - %H:%M", localtime(float(str(_rule.timestamp)))) |
110
|
|
|
rmks = "# ".join(_rule.rmks.strip().split("\n")) |
111
|
|
|
expected = """# |
112
|
|
|
# sid: %s | date: %s |
113
|
|
|
# |
114
|
|
|
# %s |
115
|
|
|
# |
116
|
|
|
%s""" % (_rule.sid, rdate, rmks, str(_rule)) |
117
|
|
|
self.assertEqual(expected, rv.data) |
118
|
|
|
|
119
|
|
|
def test_deact_rule(self): |
120
|
|
|
last_insert = NaxsiRules.query.order_by(NaxsiRules.sid.desc()).first().sid |
121
|
|
|
|
122
|
|
|
rv = self.app.get('/rules/deact/%d' % last_insert) # deactivate |
123
|
|
|
self.assertEqual(rv.status_code, 200) |
124
|
|
|
_rule = NaxsiRules.query.filter(NaxsiRules.sid == last_insert).first() |
125
|
|
|
self.assertEqual(_rule.active, 0) |
126
|
|
|
|
127
|
|
|
rv = self.app.get('/rules/deact/%d' % last_insert) # activate |
128
|
|
|
self.assertEqual(rv.status_code, 200) |
129
|
|
|
_rule = NaxsiRules.query.filter(NaxsiRules.sid == last_insert).first() |
130
|
|
|
self.assertEqual(_rule.active, 1) |
131
|
|
|
|
132
|
|
|
non_existent_sid = last_insert + 1 |
133
|
|
|
rv = self.app.get('/rules/deact/%d' % non_existent_sid) |
134
|
|
|
self.assertEqual(rv.status_code, 302) |
135
|
|
|
|
136
|
|
|
rv = self.app.get('/rules/deact/') |
137
|
|
|
self.assertEqual(rv.status_code, 404) |
138
|
|
|
|
139
|
|
|
def test_search_rule(self): |
140
|
|
|
rv = self.app.get('/rules/search/') |
141
|
|
|
self.assertEqual(rv.status_code, 302) |
142
|
|
|
|
143
|
|
|
rv = self.app.get('/rules/search/?s=a') |
144
|
|
|
self.assertEqual(rv.status_code, 302) |
145
|
|
|
|
146
|
|
|
rv = self.app.get('/rules/search/?s="OR 1=1;--') |
147
|
|
|
self.assertEqual(rv.status_code, 200) |
148
|
|
|
self.assertIn('<input type="text" name="s" size="20" value=""OR 1=1;--"', rv.data) |
149
|
|
|
self.assertIn('<p><strong>Search: OR 11--</strong></p>', rv.data) # filtered data |
150
|
|
|
|
151
|
|
|
rv = self.app.get('/rules/search/?s=1337') # get rule by id |
152
|
|
|
self.assertEqual(rv.status_code, 200) |
153
|
|
|
|
154
|
|
|
def test_edit_rule(self): |
155
|
|
|
non_existent_sid = NaxsiRules.query.order_by(NaxsiRules.sid.desc()).first().sid + 1 |
156
|
|
|
rv = self.app.get('/rules/edit/%d' % non_existent_sid) |
157
|
|
|
self.assertEqual(rv.status_code, 302) |
158
|
|
|
|
159
|
|
|
def test_parse_rule(self): |
160
|
|
|
rule_parser = NaxsiRules() |
161
|
|
|
rv = rule_parser.parse_rule("""MainRule "rx:select|union|update|delete|insert|table|from|ascii|hex|unhex|drop" "msg:sql keywords" "mz:BODY|URL|ARGS|$HEADERS_VAR:Cookie" "s:$SQL:4" id:1000;""") |
|
|
|
|
162
|
|
|
self.assertEqual(rv, True) |
163
|
|
|
|
164
|
|
|
self.assertEqual(rule_parser.warnings, ['Cookie in $HEADERS_VAR:Cookie is not lowercase. naxsi is case-insensitive', 'rule IDs below 10k are reserved (1000)']) |
|
|
|
|
165
|
|
|
rv = rule_parser.parse_rule("""BasicRule "rx:^ratata$" "mz:$URL:/foobar|$BODY_VAR_X:^tutu$" id:4200001 "s:$SQL:8";""") |
|
|
|
|
166
|
|
|
self.assertEqual(rv, False) |
167
|
|
|
self.assertEqual(rule_parser.error, ["You can't mix static $* with regex $*_X (set(['$BODY_VAR_X', '$URL']))", "parsing of element 'mz:$URL:/foobar|$BODY_VAR_X:^tutu$' failed."]) |
|
|
|
|
168
|
|
|
|
This check looks for lines that are too long. You can specify the maximum line length.