c_meyer/KasperskyLab-TinyCheck
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

change hardcoded paths and introduce -f flag for calling analysis.py from frontende to skip device.json in report generation

Browse Source
This commit is contained in:
Janik Besendorf 2021-10-12 11:07:24 +02:00
parent 48539a13df
commit c170ae2fb2
5 changed files with 90 additions and 62 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

116
analysis/analysis.py
View File

@ -16,60 +16,82 @@ import os
containing a capture.pcap file.
"""
if __name__ == "__main__":
if len(sys.argv) == 2:
capture_directory = sys.argv[1]
if os.path.isdir(capture_directory):
manager = Manager()
alerts = manager.dict()
def analyze(capture_directory,frontend=False):
if os.path.isdir(capture_directory):
def zeekengine(alerts):
zeek = ZeekEngine(capture_directory)
zeek.start_zeek()
alerts["zeek"] = zeek.retrieve_alerts()
manager = Manager()
alerts = manager.dict()
# whitelist.json writing.
with open(os.path.join(capture_directory, "assets/whitelist.json"), "w") as f:
f.write(json.dumps(zeek.retrieve_whitelist(),
indent=4, separators=(',', ': ')))
def zeekengine(alerts):
zeek = ZeekEngine(capture_directory)
zeek.start_zeek()
alerts["zeek"] = zeek.retrieve_alerts()
# conns.json writing.
with open(os.path.join(capture_directory, "assets/conns.json"), "w") as f:
f.write(json.dumps(zeek.retrieve_conns(),
indent=4, separators=(',', ': ')))
if not os.path.isdir(os.path.join(capture_directory, "assets")):
os.mkdir(os.path.join(capture_directory, "assets"))
# whitelist.json writing.
with open(os.path.join(capture_directory, "assets/whitelist.json"), "w") as f:
f.write(json.dumps(zeek.retrieve_whitelist(),
indent=4, separators=(',', ': ')))
def snortengine(alerts):
suricata = SuricataEngine(capture_directory)
suricata.start_suricata()
alerts["suricata"] = suricata.get_alerts()
# conns.json writing.
with open(os.path.join(capture_directory, "assets/conns.json"), "w") as f:
f.write(json.dumps(zeek.retrieve_conns(),
indent=4, separators=(',', ': ')))
# Start the engines.
p1 = Process(target=zeekengine, args=(alerts,))
p2 = Process(target=snortengine, args=(alerts,))
p1.start()
p2.start()
def snortengine(alerts):
suricata = SuricataEngine(capture_directory)
suricata.start_suricata()
alerts["suricata"] = suricata.get_alerts()
# Wait to their end.
p1.join()
p2.join()
# Start the engines.
p1 = Process(target=zeekengine, args=(alerts,))
p2 = Process(target=snortengine, args=(alerts,))
p1.start()
p2.start()
# Some formating and alerts.json writing.
with open(os.path.join(capture_directory, "assets/alerts.json"), "w") as f:
report = {"high": [], "moderate": [], "low": []}
for alert in (alerts["zeek"] + alerts["suricata"]):
if alert["level"] == "High":
report["high"].append(alert)
if alert["level"] == "Moderate":
report["moderate"].append(alert)
if alert["level"] == "Low":
report["low"].append(alert)
f.write(json.dumps(report, indent=4, separators=(',', ': ')))
# Wait to their end.
p1.join()
p2.join()
# Some formating and alerts.json writing.
with open(os.path.join(capture_directory, "assets/alerts.json"), "w") as f:
report = {"high": [], "moderate": [], "low": []}
for alert in (alerts["zeek"] + alerts["suricata"]):
if alert["level"] == "High":
report["high"].append(alert)
if alert["level"] == "Moderate":
report["moderate"].append(alert)
if alert["level"] == "Low":
report["low"].append(alert)
f.write(json.dumps(report, indent=4, separators=(',', ': ')))
# Generate the report
report = Report(capture_directory,frontend)
report.generate_report()
# Generate the report
report = Report(capture_directory)
report.generate_report()
else:
print("The directory doesn't exist.")
else:
print("Please specify a capture directory in argument.")
print("The directory doesn't exist.")
def usage():
print("""Usage: python analysis.py [capture_directory]
where [capture_directory] is a directory containing a capture.pcap file
analysis.py -f starts the analysis in frontend mode intended to be called by the TinyCheck frontend.""")
if __name__ == "__main__":
if len(sys.argv) == 2: #called manually without frontend
analyze(sys.argv[1], False)
elif len(sys.argv) == 3:
if(sys.argv[1]) == "-f": #frontend mode
analyze(sys.argv[2], True)
else:
usage()
else:
usage()

27
analysis/classes/report.py
View File

@ -13,7 +13,7 @@ from utils import get_config
class Report(object):
def __init__(self, capture_directory):
def __init__(self, capture_directory, frontend):
self.capture_directory = capture_directory
self.alerts = self.read_json(os.path.join(
capture_directory, "assets/alerts.json"))
@ -21,10 +21,13 @@ class Report(object):
capture_directory, "assets/whitelist.json"))
self.conns = self.read_json(os.path.join(
capture_directory, "assets/conns.json"))
self.device = self.read_json(os.path.join(
capture_directory, "assets/device.json"))
self.capinfos = self.read_json(os.path.join(
capture_directory, "assets/capinfos.json"))
self.device = None
self.capinfos = None
if frontend:
self.device = self.read_json(os.path.join(
capture_directory, "assets/device.json"))
self.capinfos = self.read_json(os.path.join(
capture_directory, "assets/capinfos.json"))
try:
with open(os.path.join(self.capture_directory, "capture.pcap"), "rb") as f:
self.capture_sha1 = hashlib.sha1(f.read()).hexdigest()
@ -204,16 +207,18 @@ class Report(object):
"""
header = "<div class=\"header\">"
header += "<div class=\"logo\"></div>"
header += "<p><br /><strong>{}: {}</strong><br />".format(self.template["device_name"],
if self.device is not None:
header += "<p><br /><strong>{}: {}</strong><br />".format(self.template["device_name"],
self.device["name"])
header += "{}: {}<br />".format(self.template["device_mac"],
header += "{}: {}<br />".format(self.template["device_mac"],
self.device["mac_address"])
header += "{} {}<br />".format(self.template["report_generated_on"],
datetime.now().strftime("%d/%m/%Y - %H:%M:%S"))
header += "{}: {}s<br />".format(self.template["capture_duration"],
self.capinfos["Capture duration"])
header += "{}: {}<br />".format(self.template["packets_number"],
self.capinfos["Number of packets"])
if self.capinfos is not None:
header += "{}: {}s<br />".format(self.template["capture_duration"],
self.capinfos["Capture duration"])
header += "{}: {}<br />".format(self.template["packets_number"],
self.capinfos["Number of packets"])
header += "{}: {}<br />".format(
self.template["capture_sha1"], self.capture_sha1)
header += "</p>"

5
analysis/classes/zeekengine.py
View File

@ -236,6 +236,7 @@ class ZeekEngine(object):
pass
try: # Domain history check.
whois_record = whois.whois(c["resolution"])
creation_date = whois_record.creation_date if type(
whois_record.creation_date) is not list else whois_record.creation_date[0]
@ -247,6 +248,7 @@ class ZeekEngine(object):
"host": c["resolution"],
"level": "Moderate",
"id": "ACT-02"})
except:
pass
@ -443,11 +445,10 @@ class ZeekEngine(object):
"""
Start zeek and check the logs.
"""
sp.Popen("cd {} && /opt/zeek/bin/zeek -Cr capture.pcap protocols/ssl/validate-certs".format(
sp.Popen("cd {} && zeek -Cr capture.pcap protocols/ssl/validate-certs".format(
self.working_dir), shell=True).wait()
sp.Popen("cd {} && mv *.log assets/".format(self.working_dir),
shell=True).wait()
self.fill_dns(self.working_dir + "/assets/")
self.netflow_check(self.working_dir + "/assets/")
self.ssl_check(self.working_dir + "/assets/")

2
analysis/utils.py
View File

@ -10,7 +10,7 @@ import os
from functools import reduce
# I'm not going to use an ORM for that.
parent = "/".join(sys.path[0].split("/")[:-1])
parent = os.path.split(os.path.dirname(os.path.abspath(sys.argv[0])))[0]
conn = sqlite3.connect(os.path.join(parent, "tinycheck.sqlite3"))
cursor = conn.cursor()

2
server/frontend/app/classes/analysis.py
View File

@ -24,7 +24,7 @@ class Analysis(object):
if self.token is not None:
parent = "/".join(sys.path[0].split("/")[:-2])
sp.Popen(
[sys.executable, "{}/analysis/analysis.py".format(parent), "/tmp/{}".format(self.token)])
[sys.executable, "{}/analysis/analysis.py".format(parent), "-f", "/tmp/{}".format(self.token)])
return {"status": True,
"message": "Analysis started",
"token": self.token}