Analysis translation test

This commit is contained in:
Félix Aime
2021-02-04 11:33:33 +01:00
parent 2625186b72
commit 96d55c2eca
5 changed files with 360 additions and 126 deletions

View File

@ -2,10 +2,13 @@ import weasyprint
import os
import json
import hashlib
import re
import sys
from weasyprint import HTML
from pathlib import Path
from datetime import datetime
from utils import get_config
class Report(object):
@ -28,6 +31,14 @@ class Report(object):
except:
self.capture_sha1 = "N/A"
self.userlang = get_config(("frontend", "userlang"))
# Load template language
if not re.match("^[a-z]{2}$", self.userlang):
self.userlang = "en"
with open(os.path.join(os.path.dirname(os.path.realpath(sys.argv[0])), "lang/{}.json".format(self.userlang)))as f:
self.template = json.load(f)["report"]
def read_json(self, json_path):
"""
Read and convert a JSON file.
@ -59,21 +70,35 @@ class Report(object):
:return: str
"""
if len(self.alerts["high"]):
return "<div class=\"warning high\">Your device seems to be compromised as you have {} high alert(s).</div>".format(self.nb_translate(len(self.alerts["high"])))
msg = "<div class=\"warning high\">"
msg += self.template["high_msg"].format(
self.nb_translate(len(self.alerts["high"])))
msg += "</div>"
return msg
elif len(self.alerts["moderate"]):
return "<div class=\"warning moderate\">You have {} moderate alert(s), your device might be compromised. Please look at them carefully.</div>".format(self.nb_translate(len(self.alerts["moderate"])))
msg = "<div class=\"warning moderate\">"
msg += self.template["moderate_msg"].format(
self.nb_translate(len(self.alerts["moderate"])))
msg += "</div>"
return msg
elif len(self.alerts["low"]):
return "<div class=\"warning low\">You have only {} low alert(s), don't hesitate to check them.</div>".format(self.nb_translate(len(self.alerts["low"])))
msg = "<div class=\"warning low\">"
msg += self.template["low_msg"].format(
self.nb_translate(len(self.alerts["low"])))
msg += "</div>"
return msg
else:
return "<div class=\"warning low\">Everything looks fine, zero alerts. Don't hesitate to check the uncategorized communications, if any.</div>"
msg = "<div class=\"warning low\">"
msg += self.template["none_msg"]
msg += "</div>"
return msg
def nb_translate(self, nb):
"""
Translate a number in a string.
:return: str
"""
a = ["one", "two", "three", "four", "five",
"six", "seven", "height", "nine"]
a = self.template["numbers"]
return a[nb-1] if nb <= 9 else str(nb)
def generate_suspect_conns_block(self):
@ -85,17 +110,17 @@ class Report(object):
if not len([c for c in self.conns if c["alert_tiggered"] == True]):
return ""
title = "<h2>Suspect communications</h2>"
table = """<table>
<thead>
<tr>
<th>Protocol</th>
<th>Domain</th>
<th>Dst IP Address</th>
<th>Dst port</th>
</tr>
</thead>
<tbody>"""
title = "<h2>{}</h2>".format(self.template["suspect_title"])
table = "<table>"
table += " <thead>"
table += " <tr>"
table += " <th>{}</th>".format(self.template["protocol"])
table += " <th>{}</th>".format(self.template["domain"])
table += " <th>{}</th>".format(self.template["dst_ip"])
table += " <th>{}</th>".format(self.template["dst_port"])
table += " </tr>"
table += " </thead>"
table += "<tbody>"
for rec in self.conns:
if rec["alert_tiggered"] == True:
@ -117,17 +142,17 @@ class Report(object):
if not len([c for c in self.conns if c["alert_tiggered"] == False]):
return ""
title = "<h2>Uncategorized communications</h2>"
table = """<table>
<thead>
<tr>
<th>Protocol</th>
<th>Domain</th>
<th>Dst IP Address</th>
<th>Dst port</th>
</tr>
</thead>
<tbody>"""
title = "<h2>{}</h2>".format(self.template["uncat_title"])
table = "<table>"
table += " <thead>"
table += " <tr>"
table += " <th>{}</th>".format(self.template["protocol"])
table += " <th>{}</th>".format(self.template["domain"])
table += " <th>{}</th>".format(self.template["dst_ip"])
table += " <th>{}</th>".format(self.template["dst_port"])
table += " </tr>"
table += " </thead>"
table += "<tbody>"
for rec in self.conns:
if rec["alert_tiggered"] == False:
@ -149,17 +174,17 @@ class Report(object):
if not len(self.whitelist):
return ""
title = "<h2>Whitelisted communications</h2>"
table = """<table>
<thead>
<tr>
<th>Protocol</th>
<th>Domain</th>
<th>Dst IP Address</th>
<th>Dst port</th>
</tr>
</thead>
<tbody>"""
title = "<h2>{}</h2>".format(self.template["whitelist_title"])
table = "<table>"
table += " <thead>"
table += " <tr>"
table += " <th>{}</th>".format(self.template["protocol"])
table += " <th>{}</th>".format(self.template["domain"])
table += " <th>{}</th>".format(self.template["dst_ip"])
table += " <th>{}</th>".format(self.template["dst_port"])
table += " </tr>"
table += " </thead>"
table += "<tbody>"
for rec in sorted(self.whitelist, key=lambda k: k['resolution']):
table += "<tr>"
@ -179,17 +204,18 @@ class Report(object):
"""
header = "<div class=\"header\">"
header += "<div class=\"logo\"></div>"
header += "<p><br /><strong>Device name: {}</strong><br />".format(
self.device["name"])
header += "Device MAC address: {}<br />".format(
self.device["mac_address"])
header += "Report generated on {}<br />".format(
datetime.now().strftime("%d/%m/%Y at %H:%M:%S"))
header += "Capture duration: {}s<br />".format(
self.capinfos["Capture duration"])
header += "Number of packets: {}<br />".format(
self.capinfos["Number of packets"])
header += "Capture SHA1: {}<br />".format(self.capture_sha1)
header += "<p><br /><strong>{}: {}</strong><br />".format(self.template["device_name"],
self.device["name"])
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"])
header += "{}: {}<br />".format(
self.template["capture_sha1"], self.capture_sha1)
header += "</p>"
header += "</div>"
return header
@ -420,16 +446,16 @@ class Report(object):
}
@page {
@top-center {
content: "REPORT_HEADER - Page " counter(page) " of " counter(pages) ".";
content: "REPORT_HEADER - Page " counter(page) " / " counter(pages) ".";
font-size:12px;
color:#CCC;
}
@bottom-center {
content: "This report has been autogenerated by a Tinycheck device. For any question, bug report or feedback, please contact tinycheck@kaspersky.com.";
content: REPORT_FOOTER;
font-size:12px;
color:#CCC;
}
}
</style>
</head>
<body>""".replace("REPORT_HEADER", "Report for the capture {}".format(self.capture_sha1))
<body>""".replace("REPORT_HEADER", "{} {}".format(self.template["report_for_the_capture"], self.capture_sha1)).replace("REPORT_FOOTER", self.template["report_footer"])

View File

@ -1,7 +1,7 @@
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
from utils import get_iocs, get_apname, get_device
from utils import get_iocs, get_apname, get_device, get_config
import time
import os
import subprocess as sp
@ -20,6 +20,14 @@ class SuricataEngine():
self.rules = [r[0] for r in get_iocs(
"snort")] + self.generate_contextual_alerts()
self.userlang = get_config(("frontend", "user_lang"))
# Load template language
if not re.match("^[a-z]{2,3}$", self.userlang):
self.userlang = "en"
with open(os.path.join(os.path.dirname(os.path.realpath(sys.argv[0])), "lang/{}.json".format(self.userlang))) as f:
self.template = json.load(f)["alerts"]
def start_suricata(self):
"""
Launch suricata against the capture.pcap file.
@ -37,9 +45,8 @@ class SuricataEngine():
s = line.split("[**]")[1].strip()
m = re.search(
r"\[\d+\:(?P<sid>\d+)\:(?P<rev>\d+)\] (?P<title>[ -~]+)", s)
self.alerts.append({"title": "Suricata rule tiggered: {}".format(m.group('title')),
"description": """A network detection rule has been tiggered. It's likely that your device has been compromised
or contains a malicious application.""",
self.alerts.append({"title": self.template["SNORT-01"]["title"].format(m.group('title')),
"description": self.template["SNORT-01"]["description"],
"level": "High",
"id": "SNORT-01"})
# Remove fast.log
@ -48,7 +55,7 @@ class SuricataEngine():
def generate_rule_file(self):
"""
Generate the rules file passed to suricata.
:return: bool if operation succeed.
:return: bool if operation succeed.
"""
try:
with open(self.rules_file, "w+") as f:

View File

@ -10,6 +10,8 @@ import subprocess as sp
import json
import pydig
import os
import re
import sys
class ZeekEngine(object):
@ -24,10 +26,17 @@ class ZeekEngine(object):
self.files = []
self.whitelist = []
# Get analysis configuration
# Get analysis and userlang configuration
self.heuristics_analysis = get_config(("analysis", "heuristics"))
self.iocs_analysis = get_config(("analysis", "iocs"))
self.whitelist_analysis = get_config(("analysis", "whitelist"))
self.userlang = get_config(("frontend", "user_lang"))
# Load template language
if not re.match("^[a-z]{2,3}$", self.userlang):
self.userlang = "en"
with open(os.path.join(os.path.dirname(os.path.realpath(sys.argv[0])), "lang/{}.json".format(self.userlang))) as f:
self.template = json.load(f)["alerts"]
def fill_dns(self, dir):
"""
@ -99,47 +108,41 @@ class ZeekEngine(object):
# Check for UDP / ICMP (strange from a smartphone.)
if c["proto"] in ["UDP", "ICMP"]:
c["alert_tiggered"] = True
self.alerts.append({"title": "{} communication going outside the local network to {}.".format(c["proto"].upper(), c["resolution"]),
"description": "The {} protocol is commonly used in internal networks. Please, verify if the host {} leveraged other alerts which may ".format(c["proto"].upper(), c["resolution"])
+ "indicates a possible malicious behavior.",
"host": c["resolution"],
self.alerts.append({"title": self.template["PROTO-01"]["title"].format(c["proto"].upper(), c["resolution"]),
"description": self.template["PROTO-01"]["description"].format(c["proto"].upper(), c["resolution"]),
"host": c["resolution"],
"level": "Moderate",
"id": "PROTO-01"})
# Check for use of ports over 1024.
if c["port_dst"] >= max_ports:
c["alert_tiggered"] = True
self.alerts.append({"title": "{} connection to {} to a port over or equal to {}.".format(c["proto"].upper(), c["resolution"], max_ports),
"description": "{} connections have been seen to {} by using the port {}. The use of non-standard port can be sometimes associated to malicious activities. ".format(c["proto"].upper(), c["resolution"], c["port_dst"])
+ "We recommend to check if this host has a good reputation by looking on other alerts and search it on the internet.",
"host": c["resolution"],
self.alerts.append({"title": self.template["PROTO-02"]["title"].format(c["proto"].upper(), c["resolution"], max_ports),
"description": self.template["PROTO-02"]["description"].format(c["proto"].upper(), c["resolution"], c["port_dst"]),
"host": c["resolution"],
"level": "Low",
"id": "PROTO-02"})
# Check for use of HTTP.
if c["service"] == "http" and c["port_dst"] == http_default_port:
c["alert_tiggered"] = True
self.alerts.append({"title": "HTTP communications have been done to the host {}".format(c["resolution"]),
"description": "Your device exchanged with the host {} by using HTTP, an unencrypted protocol. ".format(c["resolution"])
+ "Even if this behavior is not malicious by itself, it is unusual to see HTTP communications issued from smartphone applications "
+ "running in the background. Please check the host reputation by searching it on the internet.",
"host": c["resolution"],
self.alerts.append({"title": self.template["PROTO-03"]["title"].format(c["resolution"]),
"description": self.template["PROTO-03"]["description"].format(c["resolution"]),
"host": c["resolution"],
"level": "Low",
"id": "PROTO-03"})
# Check for use of HTTP on a non standard port.
if c["service"] == "http" and c["port_dst"] != http_default_port:
c["alert_tiggered"] = True
self.alerts.append({"title": "HTTP communications have been seen to the host {} on a non standard port ({}).".format(c["resolution"], c["port_dst"]),
"description": "Your device exchanged with the host {} by using HTTP, an unencrypted protocol on the port {}. ".format(c["resolution"], c["port_dst"])
+ "This behavior is quite unusual. Please check the host reputation by searching it on the internet.",
"host": c["resolution"],
self.alerts.append({"title": self.template["PROTO-04"]["title"].format(c["resolution"], c["port_dst"]),
"description": self.template["PROTO-04"]["description"].format(c["resolution"], c["port_dst"]),
"host": c["resolution"],
"level": "Moderate",
"id": "PROTO-04"})
# Check for non-resolved IP address.
if c["ip_dst"] == c["resolution"]:
c["alert_tiggered"] = True
self.alerts.append({"title": "The server {} hasn't been resolved by any DNS query during the session".format(c["ip_dst"]),
"description": "It means that the server {} is likely not resolved by any domain name or the resolution has already been cached by ".format(c["ip_dst"])
+ "the device. If the host appears in other alerts, please check it.",
self.alerts.append({"title": self.template["PROTO-05"]["title"].format(c["ip_dst"]),
"description": self.template["PROTO-05"]["description"].format(c["ip_dst"]),
"host": c["ip_dst"],
"level": "Low",
"id": "PROTO-05"})
@ -159,10 +162,8 @@ class ZeekEngine(object):
for host in bl_hosts:
if c["ip_dst"] == host[0]:
c["alert_tiggered"] = True
self.alerts.append({"title": "A connection has been made to {} ({}) which is tagged as {}.".format(c["resolution"], c["ip_dst"], host[1].upper()),
"description": "The host {} has been explicitly blacklisted for malicious activities. Your device is likely compromised ".format(c["ip_dst"])
+ "and needs to be investigated more deeply by IT security professionals.",
self.alerts.append({"title": self.template["IOC-01"]["title"].format(c["resolution"], c["ip_dst"], host[1].upper()),
"description": self.template["IOC-01"]["description"].format(c["ip_dst"]),
"host": c["resolution"],
"level": "High",
"id": "IOC-01"})
@ -171,10 +172,8 @@ class ZeekEngine(object):
for cidr in bl_cidrs:
if IPAddress(c["ip_dst"]) in cidr[0]:
c["alert_tiggered"] = True
self.alerts.append({"title": "Communication to {} under the CIDR {} which is tagged as {}.".format(c["resolution"], cidr[0], cidr[1].upper()),
"description": "The server {} is hosted under a network which is known to host malicious activities. Even if this behavior is not malicious by itself, ".format(c["resolution"])
+ "you need to check if other alerts are also mentioning this host. If you have some doubts, please "
+ "search this host on the internet to see if its legit or not.",
self.alerts.append({"title": self.template["IOC-02"]["title"].format(c["resolution"], cidr[0], cidr[1].upper()),
"description": self.template["IOC-02"]["description"].format(c["resolution"]),
"host": c["resolution"],
"level": "Moderate",
"id": "IOC-02"})
@ -183,42 +182,37 @@ class ZeekEngine(object):
if c["resolution"].endswith(domain[0]):
if domain[1] != "tracker":
c["alert_tiggered"] = True
self.alerts.append({"title": "A DNS request have been done to {} which is tagged as {}.".format(c["resolution"], domain[1].upper()),
"description": "The domain name {} seen in the capture has been explicitly tagged as malicious. This indicates that ".format(c["resolution"])
+ "your device is likely compromised and needs to be investigated deeply.",
self.alerts.append({"title": self.template["IOC-03"]["title"].format(c["resolution"], domain[1].upper()),
"description": self.template["IOC-03"]["description"].format(c["resolution"]),
"host": c["resolution"],
"level": "High",
"id": "IOC-03"})
else:
c["alert_tiggered"] = True
self.alerts.append({"title": "A DNS request have been done to {} which is tagged as {}.".format(c["resolution"], domain[1].upper()),
"description": "The domain name {} seen in the capture has been explicitly tagged as a Tracker. This ".format(c["resolution"])
+ "indicates that one of the active apps is geo-tracking your moves.",
self.alerts.append({"title": self.template["IOC-04"]["title"].format(c["resolution"], domain[1].upper()),
"description": self.template["IOC-04"]["description"].format(c["resolution"]),
"host": c["resolution"],
"level": "Moderate",
"id": "IOC-03"})
"id": "IOC-04"})
# Check for blacklisted FreeDNS.
for domain in bl_freedns:
if c["resolution"].endswith("." + domain[0]):
c["alert_tiggered"] = True
self.alerts.append({"title": "A DNS request have been done to the domain {} which is a Free DNS.".format(c["resolution"]),
"description": "The domain name {} is using a Free DNS service. This kind of service is commonly used by cybercriminals ".format(c["resolution"])
+ "or state-sponsored threat actors during their operations. It is very suspicious that an application running in background use this kind of service, please investigate.",
self.alerts.append({"title": self.template["IOC-05"]["title"].format(c["resolution"]),
"description": self.template["IOC-05"]["description"].format(c["resolution"]),
"host": c["resolution"],
"level": "Moderate",
"id": "IOC-04"})
"id": "IOC-05"})
# Check for suspect tlds.
for tld in bl_tlds:
if c["resolution"].endswith(tld[0]):
c["alert_tiggered"] = True
self.alerts.append({"title": "A DNS request have been done to the domain {} which contains a suspect TLD.".format(c["resolution"]),
"description": "The domain name {} is using a suspect Top Level Domain ({}). Even not malicious, this non-generic TLD is used regularly by cybercrime ".format(c["resolution"], tld[0])
+ "or state-sponsored operations. Please check this domain by searching it on an internet search engine. If other alerts are related to this "
+ "host, please consider it as very suspicious.",
self.alerts.append({"title": self.template["IOC-06"]["title"].format(c["resolution"]),
"description": self.template["IOC-06"]["description"].format(c["resolution"], tld[0]),
"host": c["resolution"],
"level": "Low",
"id": "IOC-05"})
"id": "IOC-06"})
# Check for use of suspect nameservers.
try:
@ -230,13 +224,11 @@ class ZeekEngine(object):
for ns in bl_nameservers:
if name_servers[0].endswith(".{}.".format(ns[0])):
c["alert_tiggered"] = True
self.alerts.append({"title": "The domain {} is using a suspect nameserver ({}).".format(c["resolution"], name_servers[0]),
"description": "The domain name {} is using a nameserver that has been explicitly tagged to be associated to malicious activities. ".format(c["resolution"])
+ "Many cybercriminals and state-sponsored threat actors are using this kind of registrars because they allow cryptocurrencies and anonymous payments. It"
+ " is adviced to investigate on this domain and the associated running application by doing a forensic analysis of the phone.",
self.alerts.append({"title": self.template["IOC-07"]["title"].format(c["resolution"], name_servers[0]),
"description": self.template["IOC-07"]["description"].format(c["resolution"]),
"host": c["resolution"],
"level": "Moderate",
"id": "IOC-06"})
"id": "IOC-07"})
def files_check(self, dir):
"""
@ -268,12 +260,11 @@ class ZeekEngine(object):
if f["sha1"] == cert[0]:
host = self.resolve(f["ip_dst"])
c["alert_tiggered"] = True
self.alerts.append({"title": "A certificate associated to {} activities have been found in the communication to {}.".format(cert[1].upper(), host),
"description": "The certificate ({}) associated to {} has been explicitly tagged as malicious. This indicates that ".format(f["sha1"], host)
+ "your device is likely compromised and need a forensic analysis.",
self.alerts.append({"title": self.template["IOC-08"]["title"].format(cert[1].upper(), host),
"description": self.template["IOC-08"]["description"].format(f["sha1"], host),
"host": host,
"level": "High",
"id": "IOC-07"})
"id": "IOC-08"})
def ssl_check(self, dir):
"""
@ -306,30 +297,24 @@ class ZeekEngine(object):
# Check for non generic SSL port.
if cert["port"] not in ssl_default_ports:
c["alert_tiggered"] = True
self.alerts.append({"title": "SSL connection done on a non standard port ({}) to {}".format(cert["port"], host),
"description": "It is not common to see SSL connections issued from smartphones using non-standard ports. Even this can be totally legit,"
+ " we recommend to check the reputation of {}, by looking at its WHOIS record, the associated autonomus system, its creation date, and ".format(host)
+ " by searching it the internet.",
self.alerts.append({"title": self.template["SSL-01"]["title"].format(cert["port"], host),
"description": self.template["SSL-01"]["description"].format(host),
"host": host,
"level": "Moderate",
"id": "SSL-01"})
# Check Free SSL certificates.
if cert["issuer"] in free_issuers:
c["alert_tiggered"] = True
self.alerts.append({"title": "An SSL connection to {} is using a free certificate.".format(host),
"description": "Free certificates — such as Let's Encrypt — are wildly used by command and control servers associated to "
+ "malicious implants or phishing web pages. We recommend to check the host associated to this certificate, "
+ "by looking at the domain name, its creation date, or by checking its reputation on the internet.",
self.alerts.append({"title": self.template["SSL-02"]["title"].format(host),
"description": self.template["SSL-02"]["description"],
"host": host,
"level": "Moderate",
"id": "SSL-02"})
# Check for self-signed certificates.
if cert["validation_status"] == "self signed certificate in certificate chain":
c["alert_tiggered"] = True
self.alerts.append({"title": "The certificate associated to {} is self-signed.".format(host),
"description": "The use of self-signed certificates is a common thing for attacker infrastructure. We recommend to check the host {} ".format(host)
+ "which is associated to this certificate, by looking at the domain name (if any), its WHOIS record, its creation date, and "
+ " by checking its reputation on the internet.",
self.alerts.append({"title": self.template["SSL-03"]["title"].format(host),
"description": self.template["SSL-03"]["description"].format(host),
"host": host,
"level": "Moderate",
"id": "SSL-03"})
@ -349,8 +334,8 @@ class ZeekEngine(object):
for host, nb in hosts.items():
if nb >= get_config(("analysis", "max_alerts")):
self.alerts.append({"title": "Check alerts for {}".format(host),
"description": "Please, check the reputation of the host {}, this one seems to be malicious as it leveraged {} alerts during the session.".format(host, nb),
self.alerts.append({"title": self.template["ADV-01"]["title"].format(host),
"description": self.template["ADV-01"]["description"].format(host, nb),
"host": host,
"level": "High",
"id": "ADV-01"})