c_meyer/SpyGuard
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

First commit!

Browse Source
This commit is contained in:
sda
2022-11-06 15:51:33 +01:00
parent 283cf9630f
commit 64daa44e9f
225 changed files with 94329 additions and 1 deletions
Download Patch File Download Diff File Expand all files Collapse all files

78
server/frontend/app/classes/analysis.py Executable file
View File

@ -0,0 +1,78 @@
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
import subprocess as sp
import json
import sys
import re
import os
class Analysis(object):
def __init__(self, token):
self.token = token if re.match(r"[A-F0-9]{8}", token) else None
def start(self) -> dict:
"""Start the analysis of the captured communication by lauching
analysis.py with the capture token as a paramater.
Returns:
dict: operation status
"""
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)])
return {"status": True,
"message": "Analysis started",
"token": self.token}
else:
return {"status": False,
"message": "Bad token provided",
"token": "null"}
def get_report(self) -> dict:
"""Generate a small json report of the analysis
containing the alerts and the device properties.
Returns:
dict: alerts, pcap and device info.
"""
device, alerts, pcap = {}, {}, {}
# Getting device configuration.
if os.path.isfile("/tmp/{}/assets/device.json".format(self.token)):
with open("/tmp/{}/assets/device.json".format(self.token), "r") as f:
device = json.load(f)
# Getting pcap infos.
if os.path.isfile("/tmp/{}/assets/capinfos.json".format(self.token)):
with open("/tmp/{}/assets/capinfos.json".format(self.token), "r") as f:
pcap = json.load(f)
# Getting alerts configuration.
if os.path.isfile("/tmp/{}/assets/alerts.json".format(self.token)):
with open("/tmp/{}/assets/alerts.json".format(self.token), "r") as f:
alerts = json.load(f)
# Getting detection methods.
if os.path.isfile("/tmp/{}/assets/detection_methods.json".format(self.token)):
with open("/tmp/{}/assets/detection_methods.json".format(self.token), "r") as f:
methods = json.load(f)
# Getting records.
if os.path.isfile("/tmp/{}/assets/records.json".format(self.token)):
with open("/tmp/{}/assets/records.json".format(self.token), "r") as f:
records = json.load(f)
if device != {} and alerts != {}:
return {"alerts": alerts,
"device": device,
"methods": methods,
"pcap": pcap,
"records": records}
else:
return {"message": "No report yet"}

161
server/frontend/app/classes/capture.py Executable file
View File

@ -0,0 +1,161 @@
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
import subprocess as sp
from app.utils import stop_monitoring, read_config, get_iocs, get_device_uuid
from app.classes.network import Network
from os import mkdir, path, chmod
import sys
import json
import random
class Capture(object):
def __init__(self):
self.random_choice_alphabet = "ABCDEF1234567890"
self.rules_file = "/tmp/rules"
self.generate_rule_file()
def start_capture(self) -> dict:
"""Start a dumpcap capture on the created AP interface and save
the generated pcap in a temporary directory under /tmp/.
Returns:
dict: Capture token and operation status.
"""
# Few context variable assignment
self.capture_token = "".join([random.choice(self.random_choice_alphabet) for i in range(8)])
self.capture_dir = "/tmp/{}/".format(self.capture_token)
self.assets_dir = "/tmp/{}/assets/".format(self.capture_token)
self.iface = read_config(("network", "in"))
self.pcap = self.capture_dir + "capture.pcap"
self.rules_file = "/tmp/rules"
# For packets monitoring
self.list_pkts = []
self.last_pkts = 0
# Make the capture and the assets directory
mkdir(self.capture_dir)
chmod(self.capture_dir, 0o777)
mkdir(self.assets_dir)
chmod(self.assets_dir, 0o777)
# Kill possible potential process
stop_monitoring()
# Writing the instance UUID for reporting.
with open("/tmp/{}/assets/instance.json".format(self.capture_token), "w") as f:
f.write(json.dumps({ "instance_uuid" : get_device_uuid().strip() }))
try:
sp.Popen(["dumpcap", "-n", "-i", self.iface, "-w", self.pcap])
sp.Popen(["suricata", "-c", "/etc/suricata/suricata.yaml", "-i", self.iface, "-l", self.assets_dir, "-S", self.rules_file])
return { "status": True,
"message": "Capture started",
"capture_token": self.capture_token }
except:
return { "status": False,
"message": f"Unexpected error: {sys.exc_info()[0]}"}
def get_capture_stats(self) -> dict:
""" Get some dirty capture statistics in order to have a sparkline
in the background of capture view.
Returns:
dict: dict containing stats associated to the capture
"""
with open("/sys/class/net/{}/statistics/tx_packets".format(self.iface)) as f:
tx_pkts = int(f.read())
with open("/sys/class/net/{}/statistics/rx_packets".format(self.iface)) as f:
rx_pkts = int(f.read())
if self.last_pkts == 0:
self.last_pkts = tx_pkts + rx_pkts
return {"status": True,
"packets": [0*400]}
else:
curr_pkts = (tx_pkts + rx_pkts) - self.last_pkts
self.last_pkts = tx_pkts + rx_pkts
self.list_pkts.append(curr_pkts)
return {"status": True,
"packets": self.beautify_stats(self.list_pkts)}
@staticmethod
def beautify_stats(data) -> list:
"""Add 0 at the end of the array if the len of the array is less
than max_len. Else, get the last 100 stats. This allows to
show a kind of "progressive chart" in the background for
the first packets.
Args:
data (list): list of integers
Returns:
list: list of integers
"""
max_len = 400
if len(data) >= max_len:
return data[-max_len:]
else:
return data + [1] * (max_len - len(data))
def stop_capture(self) -> dict:
"""Stop dumpcap & suricata if any instance present & ask create_capinfos.
Returns:
dict: operation status
"""
network = Network()
# We stop the monitoring and the associated hotspot.
if stop_monitoring():
if network.delete_hotspot():
self.create_capinfos()
return {"status": True,
"message": "Capture stopped"}
else:
return {"status": False,
"message": "No active hotspot"}
else:
return {"status": False,
"message": "No active capture"}
def create_capinfos(self) -> bool:
"""Creates a capinfo json file.
Returns:
bool: True if everything worked well.
"""
self.pcap = self.capture_dir + "capture.pcap"
infos = sp.Popen(["capinfos", self.pcap], stdout=sp.PIPE, stderr=sp.PIPE)
infos = infos.communicate()[0]
data = {}
for l in infos.decode().splitlines():
try:
l = l.split(": ") if ": " in l else l.split("= ")
if len(l[0]) and len(l[1]):
data[l[0].strip()] = l[1].strip()
except:
continue
with open("{}capinfos.json".format(self.assets_dir), 'w') as f:
json.dump(data, f)
return True
def generate_rule_file(self) -> bool:
"""Generate a suricata rules files.
Returns:
bool: operation status.
"""
rules = [r[0] for r in get_iocs("snort")]
try:
with open(self.rules_file, "w+") as f:
f.write("\n".join(rules))
return True
except:
return False

61
server/frontend/app/classes/device.py Executable file
View File

@ -0,0 +1,61 @@
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
from cmath import rect
import subprocess as sp
from app.utils import read_config
import json
import os
import re
class Device(object):
def __init__(self, token):
self.iface_in = read_config(("network", "in"))
self.token = token if re.match(r"[A-F0-9]{8}", token) else None
return None
def get(self) -> dict:
"""Get the device properties (such as Mac address, name, IP etc.)
By reading the device.json file if exists. Or reading the leases
files and writing the result into device.json.
Returns:
dict: device infos.
"""
if not os.path.isfile("/tmp/{}/assets/device.json".format(self.token)):
device = self.read_leases()
if device["status"] != False:
with open("/tmp/{}/assets/device.json".format(self.token), "w") as f:
f.write(json.dumps(device))
else:
with open("/tmp/{}/assets/device.json".format(self.token)) as f:
device = json.load(f)
return device
def read_leases(self) -> dict:
"""Get the first connected device to the generated
networks by using ARP.
Returns:
dict: connected device.
"""
sh = sp.Popen(["arp"], stdout=sp.PIPE, stderr=sp.PIPE)
sh = sh.communicate()
for line in sh[0].splitlines():
line = line.decode("utf8")
if self.iface_in in line:
rec = [x for x in line.split(" ") if x]
if rec[-1] == self.iface_in and rec[1] == "ether":
return {
"status": True,
"name": rec[2],
"ip_address": rec[0],
"mac_address": rec[2]
}
else:
return {"status": False,
"message": "Device not connected"}

181
server/frontend/app/classes/network.py Executable file
View File

@ -0,0 +1,181 @@
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
import subprocess as sp
import netifaces as ni
import requests
import re
import qrcode
import base64
import random
import requests
from app.utils import read_config
from io import BytesIO
class Network(object):
def __init__(self):
self.AP_SSID = False
self.AP_PASS = False
self.iface_out = read_config(("network", "out"))
self.iface_in = read_config(("network", "in"))
self.random_choice_alphabet = "abcdef1234567890"
def check_status(self) -> dict:
"""The method check_status check the IP addressing of the connected interface
and return its associated IP.
Returns:
dict: contains the network context.
"""
ctx = { "internet": self.check_internet() }
for iface in ni.interfaces():
if iface != self.iface_in and iface.startswith(("wl", "en", "et")):
addrs = ni.ifaddresses(iface)
try:
ctx["ip_out"] = addrs[ni.AF_INET][0]["addr"]
except:
ctx["ip_out"] = "Not connected"
return ctx
def wifi_list_networks(self) -> dict:
"""List the available wifi networks by using nmcli
Returns:
dict: list of available networks.
"""
networks = []
if self.iface_out.startswith("wl"):
sh = sp.Popen(["nmcli", "-f", "SSID,SIGNAL", "dev", "wifi", "list", "ifname", self.iface_out], stdout=sp.PIPE, stderr=sp.PIPE)
sh = sh.communicate()
for network in [n.decode("utf8") for n in sh[0].splitlines()][1:]:
name = network.strip()[:-3].strip()
signal = network.strip()[-3:].strip()
if name not in [n["name"] for n in networks] and name != "--":
networks.append({"name" : name, "signal" : int(signal) })
return { "networks": networks }
def wifi_setup(self, ssid, password) -> dict:
"""Connect to a WiFi network by using nmcli
Args:
ssid (str): Network SSID
password (str): Network password
Returns:
dict: operation status
"""
if len(password) >= 8 and len(ssid):
sh = sp.Popen(["nmcli", "dev", "wifi", "connect", ssid, "password", password, "ifname", self.iface_out], stdout=sp.PIPE, stderr=sp.PIPE)
sh = sh.communicate()
if re.match(".*[a-f0-9]{8}-[a-f0-9]{4}-[a-f0-9]{4}-[a-f0-9]{4}-[a-f0-9]{12}.*", sh[0].decode('utf8')):
return {"status": True,
"message": "Wifi connected"}
else:
return {"status": False,
"message": "Wifi not connected"}
else:
return {"status": False,
"message": "Empty SSID or/and password length less than 8 chars."}
def start_hotspot(self) -> dict:
"""Generates an Access Point by using nmcli and provide to
the GUI the associated ssid, password and qrcode.
Returns:
dict: hostpost description
"""
self.delete_hotspot()
try:
if read_config(("network", "tokenized_ssids")):
token = "".join([random.choice(self.random_choice_alphabet) for i in range(4)])
self.AP_SSID = random.choice(read_config(("network", "ssids"))) + "-" + token
else:
self.AP_SSID = random.choice(read_config(("network", "ssids")))
except:
token = "".join([random.choice(self.random_choice_alphabet) for i in range(4)])
self.AP_SSID = "wifi-" + token
self.AP_PASS = "".join([random.choice(self.random_choice_alphabet) for i in range(8)])
sp.Popen(["nmcli", "con", "add", "type", "wifi", "ifname", self.iface_in, "con-name", self.AP_SSID, "autoconnect", "yes", "ssid", self.AP_SSID]).wait()
sp.Popen(["nmcli", "con", "modify", self.AP_SSID, "802-11-wireless.mode", "ap", "802-11-wireless.band", "bg", "ipv4.method", "shared"]).wait()
sp.Popen(["nmcli", "con", "modify", self.AP_SSID, "wifi-sec.key-mgmt", "wpa-psk", "wifi-sec.psk", self.AP_PASS]).wait()
if self.launch_hotstop():
return {"status": True,
"message": "AP started",
"ssid": self.AP_SSID,
"password": self.AP_PASS,
"qrcode": self.generate_qr_code()}
else:
return {"status": False,
"message": "Error while creating AP."}
def generate_qr_code(self) -> str:
"""Returns a QRCode based on the SSID and the password.
Returns:
str: String representing the QRcode as data scheme.
"""
qrc = qrcode.make("WIFI:S:{};T:WPA;P:{};;".format(self.AP_SSID, self.AP_PASS))
buffered = BytesIO()
qrc.save(buffered, format="PNG")
return "data:image/png;base64,{}".format(base64.b64encode(buffered.getvalue()).decode("utf8"))
def launch_hotstop(self) -> bool:
"""This method enables the hotspot by asking nmcli to activate it,
then the result is checked against a regex in order to know if everything is good.
Returns:
bool: true if hotspot created.
"""
sh = sp.Popen(["nmcli", "con", "up", self.AP_SSID], stdout=sp.PIPE, stderr=sp.PIPE)
sh = sh.communicate()
return re.match(".*/ActiveConnection/[0-9]+.*", sh[0].decode("utf8"))
def check_internet(self) -> bool:
"""Check the internet link just with a small http request
to an URL present in the configuration
Returns:
bool: True if everything works.
"""
try:
url = read_config(("network", "internet_check"))
requests.get(url, timeout=10)
return True
except:
return False
def delete_hotspot(self) -> bool:
"""
Delete the previously created hotspot.
"""
sh = sp.Popen(["nmcli", "con", "show"], stdout=sp.PIPE, stderr=sp.PIPE)
for line in sh.communicate()[0].splitlines():
line = line.decode('utf8')
if self.iface_in in line:
ssids = re.search("^[a-zA-Z]+\-[0-9a-f]{4}", line)
if ssids:
sp.Popen(["nmcli", "con", "delete", ])
sh = sp.Popen(["nmcli", "con", "delete", ssids[0]], stdout=sp.PIPE, stderr=sp.PIPE)
sh = sh.communicate()
if re.match(".*[a-f0-9]{8}-[a-f0-9]{4}-[a-f0-9]{4}-[a-f0-9]{4}-[a-f0-9]{12}.*", sh[0].decode("utf8")):
return True
else:
return False

77
server/frontend/app/classes/save.py Executable file
View File

@ -0,0 +1,77 @@
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
import io
import re
import shutil
from datetime import datetime
import psutil
import pyudev
from flask import jsonify, send_file
class Save():
def __init__(self):
self.mount_point = ""
return None
def usb_check(self) -> dict:
"""Check if an USB storage is connected or not.
Returns:
dict: contains the connection status.
"""
self.usb_devices = []
context = pyudev.Context()
removable = [device for device in context.list_devices(
subsystem='block', DEVTYPE='disk')]
for device in removable:
if "usb" in device.sys_path:
partitions = [device.device_node for device in context.list_devices(
subsystem='block', DEVTYPE='partition', parent=device)]
for p in psutil.disk_partitions():
if p.device in partitions:
self.mount_point = p.mountpoint
return jsonify({"status": True,
"message": "USB storage connected"})
self.mount_point = ""
return jsonify({"status": False,
"message": "USB storage not connected"})
def save_capture(self, token, method) -> any:
"""Save the capture to the USB device or push a ZIP
file to download.
Args:
token (str): capture token
method (str): method used to save
Returns:
dict: operation status OR Flask answer.
"""
if re.match(r"[A-F0-9]{8}", token):
try:
if method == "usb":
cd = datetime.now().strftime("%d%m%Y-%H%M")
if shutil.make_archive("{}/SpyGuard_{}".format(self.mount_point, cd), "zip", "/tmp/{}/".format(token)):
shutil.rmtree("/tmp/{}/".format(token))
return jsonify({"status": True,
"message": "Capture saved on the USB key"})
elif method == "url":
cd = datetime.now().strftime("%d%m%Y-%H%M")
if shutil.make_archive("/tmp/SpyGuard_{}".format(cd), "zip", "/tmp/{}/".format(token)):
shutil.rmtree("/tmp/{}/".format(token))
with open("/tmp/SpyGuard_{}.zip".format(cd), "rb") as f:
return send_file(
io.BytesIO(f.read()),
mimetype="application/octet-stream",
as_attachment=True,
attachment_filename="SpyGuard_{}.zip".format(cd))
except:
return jsonify({"status": False,
"message": "Error while saving capture"})
else:
return jsonify({"status": False,
"message": "Bad token value"})