1
|
1 |
|
import logging |
2
|
1 |
|
import re |
3
|
1 |
|
import string |
4
|
|
|
|
5
|
1 |
|
from time import time |
6
|
1 |
|
from flask import Blueprint, render_template, request, redirect, flash, Response, url_for |
7
|
1 |
|
from sqlalchemy.exc import IntegrityError |
|
|
|
|
8
|
|
|
|
9
|
1 |
|
from spike.model import db |
10
|
1 |
|
from spike.model.naxsi_rules import NaxsiRules |
11
|
1 |
|
from spike.model.naxsi_rulesets import NaxsiRuleSets |
12
|
1 |
|
from spike.model import naxsi_mz, naxsi_score |
13
|
|
|
|
14
|
1 |
|
rules = Blueprint('rules', __name__) |
15
|
|
|
|
16
|
|
|
|
17
|
1 |
|
@rules.route("/") |
18
|
|
|
def index(): |
19
|
1 |
|
_rules = NaxsiRules.query.order_by(NaxsiRules.sid.desc()).all() |
|
|
|
|
20
|
1 |
|
if not _rules: |
21
|
|
|
flash("No rules found, please create one", "success") |
22
|
|
|
return redirect(url_for("rules.new")) |
23
|
1 |
|
return render_template("rules/index.html", rules=_rules) |
24
|
|
|
|
25
|
|
|
|
26
|
1 |
|
@rules.route("/plain/<int:sid>", methods=["GET"]) |
27
|
|
|
def plain(sid): |
28
|
1 |
|
_rule = NaxsiRules.query.filter(NaxsiRules.sid == sid).first() |
|
|
|
|
29
|
1 |
|
if not _rule: |
30
|
|
|
flash("No rules found, please create one", "error") |
31
|
|
|
return redirect(url_for("rules.new")) |
32
|
1 |
|
return Response(_rule.fullstr(), mimetype='text/plain') |
33
|
|
|
|
34
|
|
|
|
35
|
1 |
|
@rules.route("/view/<int:sid>", methods=["GET"]) |
36
|
|
|
def view(sid): |
37
|
1 |
|
_rule = NaxsiRules.query.filter(NaxsiRules.sid == sid).first() |
|
|
|
|
38
|
1 |
|
if _rule is None: |
39
|
1 |
|
flash("no rules found, please create one", "error") |
40
|
1 |
|
return redirect(url_for("rules.index")) |
41
|
1 |
|
return render_template("rules/view.html", rule=_rule, rtext=_rule) |
42
|
|
|
|
43
|
|
|
|
44
|
1 |
|
@rules.route("/search/", methods=["GET"]) |
45
|
|
|
def search(): |
46
|
1 |
|
terms = request.args.get('s', '') |
47
|
|
|
|
48
|
1 |
|
if len(terms) < 2: |
49
|
1 |
|
return redirect(url_for("rules.index")) |
50
|
|
|
|
51
|
|
|
# No fancy injections |
52
|
1 |
|
whitelist = set(string.ascii_letters + string.digits + ':-_ ') |
53
|
1 |
|
filtered = ''.join(filter(whitelist.__contains__, terms)) |
54
|
|
|
|
55
|
1 |
|
if filtered.isdigit(): # get rule by id |
56
|
1 |
|
_rules = db.session.query(NaxsiRules).filter(NaxsiRules.sid == int(filtered)) |
57
|
|
|
else: |
58
|
1 |
|
cve = re.search('cve:\d{4}-\d{4,}', filtered, re.IGNORECASE) # search by CVE |
|
|
|
|
59
|
|
|
|
60
|
1 |
|
expression = '%' + filtered + '%' |
61
|
1 |
|
_rules = db.session.query(NaxsiRules).filter( |
62
|
|
|
db.or_( |
63
|
|
|
NaxsiRules.msg.like(expression), |
64
|
|
|
NaxsiRules.rmks.like(expression), |
65
|
|
|
NaxsiRules.detection.like(expression) |
66
|
|
|
) |
67
|
|
|
) |
68
|
1 |
|
if cve: |
69
|
1 |
|
_rules.filter(NaxsiRules.msg.like('%' + cve.group() + '%')) |
70
|
1 |
|
_rules = _rules.order_by(NaxsiRules.sid.desc()).all() |
71
|
1 |
|
return render_template("rules/index.html", rules=_rules, selection="Search: %s" % filtered, lsearch=terms) |
72
|
|
|
|
73
|
|
|
|
74
|
1 |
|
@rules.route("/new", methods=["GET", "POST"]) |
75
|
|
|
def new(): |
76
|
1 |
|
latest_sid = NaxsiRules.query.order_by(NaxsiRules.sid.desc()).first() |
|
|
|
|
77
|
1 |
|
if latest_sid is None: |
78
|
|
|
sid = 200001 |
79
|
|
|
else: |
80
|
1 |
|
sid = latest_sid.sid + 1 |
81
|
|
|
|
82
|
1 |
|
if request.method == "GET": |
83
|
1 |
|
_rulesets = NaxsiRuleSets.query.all() |
|
|
|
|
84
|
1 |
|
return render_template("rules/new.html", mz=naxsi_mz, rulesets=_rulesets, score=naxsi_score, latestn=sid) |
85
|
|
|
|
86
|
|
|
# create new rule |
87
|
1 |
|
logging.debug('Posted new request: %s', request.form) |
88
|
1 |
|
mz = "|".join(filter(len, request.form.getlist("mz") + request.form.getlist("custom_mz_val"))) |
89
|
|
|
|
90
|
1 |
|
score = request.form.get("score", "") |
91
|
1 |
|
score += ':' |
92
|
1 |
|
score += request.form.get("score_%s" % request.form.get("score", ""), "") |
93
|
|
|
|
94
|
1 |
|
nrule = NaxsiRules(request.form.get("msg", ""), request.form.get("detection", ""), mz, score, sid, |
95
|
|
|
request.form.get("ruleset", ""), request.form.get("rmks", ""), "1", |
96
|
|
|
request.form.get("negative", "") == 'checked', int(time())) |
97
|
|
|
|
98
|
1 |
|
errors, warnings = nrule.validate() |
99
|
|
|
|
100
|
1 |
|
if errors: |
101
|
1 |
|
for error in errors: |
102
|
1 |
|
flash(error, category='error') |
103
|
1 |
|
return redirect(url_for("rules.new")) |
104
|
1 |
|
elif warnings: |
105
|
1 |
|
for warning in warnings: |
106
|
1 |
|
flash(warning, category='warnings') |
107
|
|
|
|
108
|
1 |
|
db.session.add(nrule) |
109
|
1 |
|
db.session.commit() |
110
|
|
|
|
111
|
1 |
|
return redirect("/rules/edit/%s" % sid) |
112
|
|
|
|
113
|
|
|
|
114
|
1 |
|
@rules.route("/test/<int:sid>", methods=["GET", "POST"]) |
115
|
|
|
def test(sid): |
116
|
|
|
_rule = NaxsiRules.query.filter(NaxsiRules.sid == sid).first() |
|
|
|
|
117
|
|
|
if _rule is None: |
118
|
|
|
flash("no rules found, please create one", "error") |
119
|
|
|
return redirect(url_for("rules.index")) |
120
|
|
|
return render_template("rules/test.html", rule=_rule, rtext=_rule) |
121
|
|
|
|
122
|
|
|
|
123
|
1 |
|
@rules.route("/edit/<int:sid>", methods=["GET", "POST"]) |
124
|
|
|
def edit(sid): |
125
|
1 |
|
rinfo = NaxsiRules.query.filter(NaxsiRules.sid == sid).first() |
|
|
|
|
126
|
1 |
|
if not rinfo: |
127
|
1 |
|
return redirect(url_for("rules.index")) |
128
|
|
|
|
129
|
1 |
|
_rulesets = NaxsiRuleSets.query.all() |
|
|
|
|
130
|
1 |
|
rruleset = NaxsiRuleSets.query.filter(NaxsiRuleSets.name == rinfo.ruleset).first() |
|
|
|
|
131
|
1 |
|
custom_mz = "" |
132
|
1 |
|
mz_check = rinfo.mz |
133
|
1 |
|
if re.search(r"^\$[A-Z]+:(.*)\|[A-Z]+", mz_check): |
134
|
|
|
custom_mz = mz_check |
135
|
|
|
rinfo.mz = "custom" |
136
|
1 |
|
return render_template("rules/edit.html", mz=naxsi_mz, rulesets=_rulesets, score=naxsi_score, rules_info=rinfo, |
137
|
|
|
rule_ruleset=rruleset, custom_mz=custom_mz) |
138
|
|
|
|
139
|
|
|
|
140
|
1 |
|
@rules.route("/save/<int:sid>", methods=["POST"]) |
141
|
|
|
def save(sid): |
142
|
1 |
|
mz = "|".join(filter(len, request.form.getlist("mz") + request.form.getlist("custom_mz_val"))) |
143
|
1 |
|
score = "{}:{}".format(request.form.get("score", ""), request.form.get("score_%s" % request.form.get("score", ""))) |
144
|
1 |
|
nrule = NaxsiRules.query.filter(NaxsiRules.sid == sid).first() |
|
|
|
|
145
|
1 |
|
nrule.msg = request.form.get("msg", "") |
146
|
1 |
|
nrule.detection = request.form.get("detection", "") |
147
|
1 |
|
nrule.mz = mz |
148
|
1 |
|
nrule.score = score |
149
|
1 |
|
nrule.ruleset = request.form.get("ruleset", "") |
150
|
1 |
|
nrule.rmks = request.form.get("rmks", "") |
151
|
1 |
|
nrule.active = request.form.get("active", "") |
152
|
1 |
|
nrule.negative = request.form.get("negative", "") == 'checked' |
153
|
1 |
|
nrule.timestamp = int(time()) |
154
|
1 |
|
errors, warnings = nrule.validate() |
155
|
|
|
|
156
|
1 |
|
if errors: |
157
|
1 |
|
flash(",".join(errors), 'error') |
158
|
1 |
View Code Duplication |
return redirect("/rules/edit/%s" % sid) |
|
|
|
|
159
|
1 |
|
elif warnings: |
160
|
1 |
|
flash(",".join(warnings), 'warning') |
161
|
|
|
|
162
|
1 |
|
db.session.add(nrule) |
163
|
1 |
|
db.session.commit() |
164
|
|
|
|
165
|
1 |
|
return redirect("/rules/edit/%s" % sid) |
166
|
|
|
|
167
|
|
|
|
168
|
1 |
|
@rules.route("/del/<int:sid>", methods=["GET"]) |
169
|
1 |
|
def del_sid(sid=''): |
170
|
1 |
|
nrule = NaxsiRules.query.filter(NaxsiRules.sid == sid).first() |
|
|
|
|
171
|
1 |
|
if not nrule: |
172
|
|
|
return redirect(url_for("rules.index")) |
173
|
|
|
|
174
|
1 |
|
db.session.delete(nrule) |
175
|
1 |
|
db.session.commit() |
176
|
|
|
|
177
|
1 |
|
flash("Successfully deleted %s : %s" % (sid, nrule.msg), "success") |
178
|
1 |
|
return redirect(url_for("rules.index")) |
179
|
|
|
|
180
|
|
|
|
181
|
1 |
|
@rules.route("/deact/<int:sid>", methods=["GET"]) |
182
|
|
|
def deact(sid): |
183
|
1 |
|
nrule = NaxsiRules.query.filter(NaxsiRules.sid == sid).first() |
|
|
|
|
184
|
1 |
|
if nrule is None: |
185
|
1 |
|
return redirect(url_for("rules.index")) |
186
|
|
|
|
187
|
1 |
|
fm = 'deactivate' if nrule.active else 'reactivate' |
188
|
1 |
|
nrule.active = not nrule.active |
189
|
|
|
|
190
|
1 |
|
db.session.add(nrule) |
191
|
1 |
|
db.session.commit() |
192
|
|
|
|
193
|
1 |
|
flash("Successfully deactivated %s %sd : %s" % (fm, sid, nrule.msg), "success") |
194
|
1 |
|
_rulesets = NaxsiRuleSets.query.all() |
|
|
|
|
195
|
1 |
|
return render_template("rules/edit.html", mz=naxsi_mz, rulesets=_rulesets, score=naxsi_score, rules_info=nrule) |
196
|
|
|
|
197
|
|
|
|
198
|
1 |
|
@rules.route("/import", methods=["POST", "GET"]) |
199
|
|
|
def import_rules(): |
200
|
|
|
if request.method == "GET": |
201
|
|
|
return render_template("rules/import.html", rulesets=NaxsiRuleSets.query.all()) |
|
|
|
|
202
|
|
|
|
203
|
|
|
ruleset = request.form.get("ruleset", "") |
204
|
|
|
upfile = request.files.get('file', '') |
205
|
|
|
|
206
|
|
|
if not ruleset or not upfile: |
207
|
|
|
flash("Missing rule file and/or ruleset name.", 'error') |
208
|
|
|
return redirect(url_for("rules.new")) |
209
|
|
|
|
210
|
|
|
raw = upfile.stream.getvalue() |
211
|
|
|
|
212
|
|
|
success_imports = 0 |
213
|
|
|
potential_imports = 0 |
214
|
|
|
for potential_rule in raw.split('\n'): |
215
|
|
|
potential_rule = potential_rule.strip() |
216
|
|
|
if not potential_rule or potential_rule.startswith("#"): # Save ourselves some time by not trying to import comments |
|
|
|
|
217
|
|
|
continue |
218
|
|
|
potential_imports += 1 |
219
|
|
|
_rule = NaxsiRules(ruleset=ruleset, active=1) |
220
|
|
|
errors, warnings, rule = _rule.parse_rule(potential_rule) |
|
|
|
|
221
|
|
|
|
222
|
|
|
if errors: |
223
|
|
|
flash("Fail to parse %s: %s" % (potential_rule, ', '.join(errors)), 'error') |
224
|
|
|
continue |
225
|
|
|
else: |
226
|
|
|
db.session.add(rule) |
227
|
|
|
try: |
228
|
|
|
db.session.commit() |
229
|
|
|
success_imports += 1 |
230
|
|
|
except IntegrityError: |
231
|
|
|
db.session.rollback() |
232
|
|
|
flash("Rule %s has not an unique ID" % rule['sid']) |
233
|
|
|
flash("Imported %d rules out of %d lines in ruleset %s" % (success_imports, potential_imports, ruleset)) |
234
|
|
|
|
235
|
|
|
return render_template("rules/import.html", rulesets=NaxsiRuleSets.query.all()) |
|
|
|
|
236
|
|
|
|
This can be caused by one of the following:
1. Missing Dependencies
This error could indicate a configuration issue of Pylint. Make sure that your libraries are available by adding the necessary commands.
2. Missing __init__.py files
This error could also result from missing
__init__.py
files in your module folders. Make sure that you place one file in each sub-folder.