2611 lines
50 KiB
Python
2611 lines
50 KiB
Python
import os
|
|
import sys
|
|
import signal
|
|
import subprocess
|
|
import threading
|
|
import logging
|
|
import secrets
|
|
import time
|
|
import tempfile
|
|
from collections import deque
|
|
from datetime import datetime, timezone
|
|
from pathlib import Path
|
|
|
|
|
|
PROJECT_ROOT = Path(__file__).resolve().parent.parent
|
|
|
|
sys.path.insert(
|
|
0,
|
|
str(PROJECT_ROOT / "src")
|
|
)
|
|
|
|
from flask import (
|
|
Flask,
|
|
Response,
|
|
jsonify,
|
|
redirect,
|
|
render_template,
|
|
request,
|
|
send_file,
|
|
send_from_directory,
|
|
url_for,
|
|
)
|
|
|
|
from reports.database import close as close_database
|
|
from reports.database import connect as connect_database
|
|
from reports.dashboard import (
|
|
generate_capture_sessions,
|
|
generate_dashboard,
|
|
)
|
|
from cracklab.server.web import cracklab_bp
|
|
from reports.search import (
|
|
build_hash_export,
|
|
build_search_data,
|
|
normalize_criteria,
|
|
render_search_page,
|
|
)
|
|
from reports.hashcat import (
|
|
build_all_hash_export,
|
|
build_export_filename,
|
|
import_hashcat_show,
|
|
)
|
|
from reports.hashcat_page import (
|
|
render_hashcat_page,
|
|
)
|
|
from reports.trips_page import render_trips_page
|
|
from reports.wifitrip import (
|
|
export_wifitrip,
|
|
import_wifitrip,
|
|
validate_wifitrip,
|
|
)
|
|
from reports.session_page import render_session_page
|
|
from reports.access_points import render_access_points_page
|
|
from reports.access_points_page import render_access_point_page
|
|
from reports.template import ReportContext
|
|
from reports.vendors import render_vendors_page
|
|
from vendor_service import VendorService
|
|
from sudo import SudoChecker
|
|
from config import Config
|
|
from gps_test import GPSTestService
|
|
|
|
app = Flask(
|
|
__name__,
|
|
template_folder=str(
|
|
PROJECT_ROOT / "gui" / "templates"
|
|
),
|
|
static_folder=str(
|
|
PROJECT_ROOT / "gui" / "static"
|
|
)
|
|
)
|
|
|
|
app.register_blueprint(
|
|
cracklab_bp
|
|
)
|
|
|
|
@app.before_request
|
|
def scanner_navigation_guard():
|
|
|
|
status = get_scanner_status()
|
|
|
|
if not status["running"]:
|
|
return None
|
|
|
|
if (
|
|
request.endpoint is not None
|
|
and request.endpoint.startswith("cracklab.")
|
|
):
|
|
return None
|
|
|
|
allowed_endpoints = {
|
|
"index",
|
|
"api_status",
|
|
"scanner_stop",
|
|
"api_gps_test",
|
|
"gps_test_stop",
|
|
"static"
|
|
}
|
|
|
|
if request.endpoint in allowed_endpoints:
|
|
return None
|
|
|
|
return redirect(
|
|
url_for("index")
|
|
)
|
|
|
|
|
|
scanner_process = None
|
|
|
|
scanner_health = "READY"
|
|
|
|
scanner_stopped_at = None
|
|
|
|
scanner_output = deque(
|
|
maxlen=500
|
|
)
|
|
|
|
scanner_lock = threading.Lock()
|
|
|
|
vendor_previews = {}
|
|
|
|
vendor_preview_lock = threading.Lock()
|
|
|
|
vendor_update_job = None
|
|
|
|
vendor_update_lock = threading.Lock()
|
|
|
|
gps_test_service = None
|
|
|
|
gps_test_lock = threading.Lock()
|
|
|
|
def check_sudo():
|
|
|
|
try:
|
|
|
|
result = subprocess.run(
|
|
[
|
|
"sudo",
|
|
"-n",
|
|
"-v"
|
|
],
|
|
stdin=subprocess.DEVNULL,
|
|
stdout=subprocess.DEVNULL,
|
|
stderr=subprocess.DEVNULL
|
|
)
|
|
|
|
except KeyboardInterrupt:
|
|
return False
|
|
|
|
except Exception:
|
|
return False
|
|
|
|
return result.returncode == 0
|
|
|
|
|
|
def sudo_keepalive():
|
|
|
|
checker = SudoChecker()
|
|
|
|
while True:
|
|
|
|
time.sleep(60)
|
|
|
|
checker.refresh()
|
|
|
|
def gps_test_output(message):
|
|
|
|
with scanner_lock:
|
|
|
|
scanner_output.append(
|
|
message
|
|
)
|
|
|
|
|
|
def get_gps_test_status():
|
|
|
|
with gps_test_lock:
|
|
|
|
service = gps_test_service
|
|
|
|
if service is None:
|
|
|
|
return {
|
|
"running": False,
|
|
"state": "READY",
|
|
"nmea_received": False,
|
|
"fix_received": False,
|
|
"nmea_count": 0,
|
|
"fix_count": 0,
|
|
"latitude": None,
|
|
"longitude": None,
|
|
"error": None
|
|
}
|
|
|
|
return {
|
|
"running": service.running,
|
|
"state": service.state,
|
|
"nmea_received": service.nmea_received,
|
|
"fix_received": service.fix_received,
|
|
"nmea_count": service.nmea_count,
|
|
"fix_count": service.fix_count,
|
|
"latitude": service.latitude,
|
|
"longitude": service.longitude,
|
|
"error": service.error
|
|
}
|
|
|
|
def get_scanner_status():
|
|
|
|
global scanner_health
|
|
global scanner_stopped_at
|
|
global scanner_process
|
|
|
|
with scanner_lock:
|
|
|
|
process = scanner_process
|
|
|
|
if (
|
|
scanner_health == "STOPPED"
|
|
and
|
|
scanner_stopped_at is not None
|
|
and
|
|
time.time() - scanner_stopped_at >= 5
|
|
):
|
|
|
|
scanner_health = "READY"
|
|
|
|
scanner_stopped_at = None
|
|
|
|
if process is None:
|
|
return {
|
|
"running": False,
|
|
"pid": None,
|
|
"health": scanner_health
|
|
}
|
|
|
|
return_code = process.poll()
|
|
|
|
if return_code is None:
|
|
return {
|
|
"running": True,
|
|
"pid": process.pid,
|
|
"health": scanner_health
|
|
}
|
|
|
|
scanner_process = None
|
|
scanner_health = "STOPPED"
|
|
scanner_stopped_at = time.time()
|
|
|
|
scanner_output.append(
|
|
f"[GUI] Scanner process exited with code {return_code}."
|
|
)
|
|
|
|
return {
|
|
"running": False,
|
|
"pid": process.pid,
|
|
"health": scanner_health,
|
|
"return_code": return_code
|
|
}
|
|
|
|
|
|
def read_scanner_output(process):
|
|
|
|
global scanner_health
|
|
|
|
try:
|
|
|
|
for line in iter(
|
|
process.stdout.readline,
|
|
""
|
|
):
|
|
|
|
line = line.rstrip()
|
|
|
|
if line.startswith(
|
|
"ScannerState:"
|
|
):
|
|
|
|
state = line.split(
|
|
":",
|
|
1
|
|
)[1].strip()
|
|
|
|
if state in {
|
|
"READY",
|
|
"ACTIVE",
|
|
"WARNING",
|
|
"STOPPED"
|
|
}:
|
|
|
|
if (
|
|
state != "STOPPED"
|
|
or
|
|
process.poll() is not None
|
|
):
|
|
|
|
with scanner_lock:
|
|
|
|
scanner_health = state
|
|
|
|
continue
|
|
|
|
if line:
|
|
|
|
with scanner_lock:
|
|
|
|
scanner_output.append(
|
|
line
|
|
)
|
|
|
|
except Exception as error:
|
|
|
|
with scanner_lock:
|
|
|
|
scanner_output.append(
|
|
f"[GUI] Output reader error: {error}"
|
|
)
|
|
|
|
finally:
|
|
|
|
try:
|
|
|
|
process.stdout.close()
|
|
|
|
except Exception:
|
|
|
|
pass
|
|
|
|
|
|
def get_wifi_interfaces():
|
|
|
|
try:
|
|
|
|
result = subprocess.run(
|
|
[
|
|
"sudo",
|
|
"-n",
|
|
"iw",
|
|
"dev"
|
|
],
|
|
|
|
cwd=PROJECT_ROOT,
|
|
|
|
capture_output=True,
|
|
|
|
text=True,
|
|
|
|
timeout=10
|
|
|
|
)
|
|
|
|
except Exception as error:
|
|
|
|
return [], str(error)
|
|
|
|
|
|
if result.returncode != 0:
|
|
|
|
error = (
|
|
result.stderr.strip()
|
|
or
|
|
f"iw exited with code {result.returncode}"
|
|
)
|
|
|
|
return [], error
|
|
|
|
|
|
interfaces = []
|
|
|
|
current = None
|
|
|
|
for line in result.stdout.splitlines():
|
|
|
|
line = line.strip()
|
|
|
|
if line.startswith("Interface"):
|
|
|
|
parts = line.split()
|
|
|
|
if len(parts) >= 2:
|
|
|
|
current = parts[1]
|
|
|
|
interfaces.append(current)
|
|
|
|
|
|
return interfaces, None
|
|
|
|
def get_wifi_channels(interface):
|
|
|
|
try:
|
|
|
|
result = subprocess.run(
|
|
[
|
|
"sudo",
|
|
"-n",
|
|
str(
|
|
PROJECT_ROOT
|
|
/ "bin"
|
|
/ "hcxdumptool"
|
|
),
|
|
"-I",
|
|
interface
|
|
],
|
|
|
|
cwd=PROJECT_ROOT,
|
|
|
|
capture_output=True,
|
|
|
|
text=True,
|
|
|
|
timeout=30
|
|
)
|
|
|
|
except Exception as error:
|
|
|
|
return {
|
|
"2.4": [],
|
|
"5": [],
|
|
"6": [],
|
|
"error": str(error)
|
|
}
|
|
|
|
|
|
output = (
|
|
result.stdout
|
|
+
|
|
"\n"
|
|
+
|
|
result.stderr
|
|
)
|
|
|
|
|
|
if result.returncode != 0:
|
|
|
|
error = (
|
|
result.stderr.strip()
|
|
or
|
|
result.stdout.strip()
|
|
or
|
|
f"hcxdumptool exited with code "
|
|
f"{result.returncode}"
|
|
)
|
|
|
|
return {
|
|
"2.4": [],
|
|
"5": [],
|
|
"6": [],
|
|
"error": error
|
|
}
|
|
|
|
|
|
import re
|
|
|
|
|
|
channels = {
|
|
"2.4": [],
|
|
"5": [],
|
|
"6": []
|
|
}
|
|
|
|
|
|
pattern = re.compile(
|
|
r"(\d{4,5})\s+"
|
|
r"\[\s*(\d+)\s*\]"
|
|
r"(?:\s+(disabled))?"
|
|
)
|
|
|
|
|
|
seen = {
|
|
"2.4": set(),
|
|
"5": set(),
|
|
"6": set()
|
|
}
|
|
|
|
|
|
for line in output.splitlines():
|
|
|
|
matches = pattern.findall(line)
|
|
|
|
|
|
for frequency, channel, disabled in matches:
|
|
|
|
frequency = int(frequency)
|
|
channel = int(channel)
|
|
|
|
|
|
if 2400 <= frequency < 2500:
|
|
|
|
band = "2.4"
|
|
suffix = "a"
|
|
|
|
elif 5000 <= frequency < 5925:
|
|
|
|
band = "5"
|
|
suffix = "b"
|
|
|
|
elif 5925 <= frequency <= 7125:
|
|
|
|
band = "6"
|
|
suffix = "c"
|
|
|
|
else:
|
|
|
|
continue
|
|
|
|
|
|
key = (
|
|
channel,
|
|
frequency
|
|
)
|
|
|
|
|
|
if key in seen[band]:
|
|
|
|
continue
|
|
|
|
|
|
seen[band].add(key)
|
|
|
|
|
|
channels[band].append(
|
|
{
|
|
"channel": channel,
|
|
"frequency": frequency,
|
|
"value": f"{channel}{suffix}",
|
|
"available": not bool(disabled)
|
|
}
|
|
)
|
|
|
|
|
|
for band in channels:
|
|
|
|
channels[band].sort(
|
|
key=lambda item: (
|
|
item["channel"],
|
|
item["frequency"]
|
|
)
|
|
)
|
|
|
|
|
|
return {
|
|
"2.4": channels["2.4"],
|
|
"5": channels["5"],
|
|
"6": channels["6"],
|
|
"error": None
|
|
}
|
|
|
|
def read_config():
|
|
|
|
import yaml
|
|
|
|
config_file = (
|
|
PROJECT_ROOT
|
|
/
|
|
"config"
|
|
/
|
|
"config.yaml"
|
|
)
|
|
|
|
try:
|
|
|
|
with open(
|
|
config_file,
|
|
"r"
|
|
) as file:
|
|
|
|
data = yaml.safe_load(file)
|
|
|
|
gps = data["gps"]
|
|
|
|
if "source" not in gps:
|
|
|
|
data["gps"] = {
|
|
"source": "gpsrelay",
|
|
"gpsrelay": {
|
|
"host": gps["host"],
|
|
"port": gps["port"],
|
|
},
|
|
"gnss": {
|
|
"handler": "direct",
|
|
"direct": {
|
|
"device":
|
|
"/dev/serial/by-id/"
|
|
"usb-u-blox_AG_-_www.u-blox.com_"
|
|
"u-blox_GNSS_receiver-if00",
|
|
},
|
|
"gpsd": {},
|
|
"modem": {
|
|
"device":
|
|
"/dev/serial/by-id/"
|
|
"usb-Quectel_EM160R-GL_718c89b2-if01-port0",
|
|
"at_device":
|
|
"/dev/serial/by-id/"
|
|
"usb-Quectel_EM160R-GL_718c89b2-if03-port0",
|
|
},
|
|
},
|
|
}
|
|
|
|
elif "gnss" in gps and "direct" not in gps["gnss"]:
|
|
|
|
old_gnss = gps["gnss"]
|
|
|
|
data["gps"]["gnss"] = {
|
|
"handler": old_gnss.get(
|
|
"handler",
|
|
"direct"
|
|
),
|
|
"direct": {
|
|
"device":
|
|
old_gnss.get(
|
|
"device",
|
|
"/dev/serial/by-id/"
|
|
"usb-u-blox_AG_-_www.u-blox.com_"
|
|
"u-blox_GNSS_receiver-if00"
|
|
)
|
|
},
|
|
"gpsd": {},
|
|
"modem": {
|
|
"device":
|
|
"/dev/serial/by-id/"
|
|
"usb-Quectel_EM160R-GL_718c89b2-if01-port0",
|
|
"at_device":
|
|
old_gnss.get(
|
|
"at_device",
|
|
"/dev/serial/by-id/"
|
|
"usb-Quectel_EM160R-GL_718c89b2-if03-port0"
|
|
),
|
|
},
|
|
}
|
|
|
|
return data
|
|
|
|
except Exception as error:
|
|
|
|
return {
|
|
"_error": str(error)
|
|
}
|
|
|
|
|
|
def save_config(data):
|
|
|
|
import re
|
|
import yaml
|
|
|
|
config_file = (
|
|
PROJECT_ROOT
|
|
/
|
|
"config"
|
|
/
|
|
"config.yaml"
|
|
)
|
|
|
|
with open(
|
|
config_file,
|
|
"r"
|
|
) as file:
|
|
|
|
text = file.read()
|
|
|
|
def yaml_string(value):
|
|
|
|
import json
|
|
|
|
return json.dumps(
|
|
str(value),
|
|
ensure_ascii=False
|
|
)
|
|
|
|
from zoneinfo import ZoneInfo
|
|
|
|
timezone_name = str(
|
|
data["timezone"]
|
|
).strip()
|
|
|
|
if not timezone_name:
|
|
raise RuntimeError(
|
|
"Timezone cannot be empty."
|
|
)
|
|
|
|
try:
|
|
ZoneInfo(timezone_name)
|
|
except Exception as error:
|
|
raise RuntimeError(
|
|
f"Invalid timezone: {timezone_name}"
|
|
) from error
|
|
|
|
values = {
|
|
"database": {
|
|
"file": yaml_string(
|
|
data["database"]["file"]
|
|
),
|
|
},
|
|
|
|
"vendor": {
|
|
"database": yaml_string(
|
|
data["vendor"]["database"]
|
|
),
|
|
},
|
|
|
|
"wifi": {
|
|
"enable_2_4ghz": (
|
|
"true"
|
|
if data["wifi"]["enable_2_4ghz"]
|
|
else
|
|
"false"
|
|
),
|
|
"channels_2_4ghz": yaml_string(
|
|
data["wifi"]["channels_2_4ghz"]
|
|
),
|
|
"enable_5ghz": (
|
|
"true"
|
|
if data["wifi"]["enable_5ghz"]
|
|
else
|
|
"false"
|
|
),
|
|
"channels_5ghz": yaml_string(
|
|
data["wifi"]["channels_5ghz"]
|
|
),
|
|
"enable_6ghz": (
|
|
"true"
|
|
if data["wifi"]["enable_6ghz"]
|
|
else
|
|
"false"
|
|
),
|
|
"channels_6ghz": yaml_string(
|
|
data["wifi"]["channels_6ghz"]
|
|
),
|
|
"filtermode": str(
|
|
data["wifi"]["filtermode"]
|
|
),
|
|
"extra_args": yaml_string(
|
|
data["wifi"]["extra_args"]
|
|
),
|
|
},
|
|
|
|
"captures": {
|
|
"directory": yaml_string(
|
|
data["captures"]["directory"]
|
|
),
|
|
},
|
|
|
|
"reports": {
|
|
"recent_sessions_limit": str(
|
|
data["reports"]["recent_sessions_limit"]
|
|
),
|
|
},
|
|
|
|
"logging": {
|
|
"level": yaml_string(
|
|
data["logging"]["level"]
|
|
),
|
|
},
|
|
|
|
"status": {
|
|
"interval": str(
|
|
data["status"]["interval"]
|
|
),
|
|
},
|
|
}
|
|
|
|
timezone_pattern = (
|
|
r"(^timezone:[ \t]*)"
|
|
r"([^#\n]*?)"
|
|
r"([ \t]*#.*)?$"
|
|
)
|
|
|
|
text, count = re.subn(
|
|
timezone_pattern,
|
|
lambda item: (
|
|
item.group(1)
|
|
+
|
|
yaml_string(
|
|
timezone_name
|
|
)
|
|
+
|
|
(
|
|
item.group(3)
|
|
if item.group(3)
|
|
else
|
|
""
|
|
)
|
|
),
|
|
text,
|
|
count=1,
|
|
flags=re.MULTILINE
|
|
)
|
|
|
|
if count != 1:
|
|
raise RuntimeError(
|
|
"Configuration key not found: timezone"
|
|
)
|
|
|
|
gps_pattern = (
|
|
r"(^gps:[ \t]*\n)"
|
|
r"(.*?)(?=^[A-Za-z_][A-Za-z0-9_]*:[ \t]*$|\Z)"
|
|
)
|
|
|
|
gps_match = re.search(
|
|
gps_pattern,
|
|
text,
|
|
flags=re.MULTILINE | re.DOTALL
|
|
)
|
|
|
|
if not gps_match:
|
|
|
|
raise RuntimeError(
|
|
"Configuration section not found: gps"
|
|
)
|
|
|
|
gps_text = (
|
|
" source: "
|
|
+ yaml_string(
|
|
data["gps"]["source"]
|
|
)
|
|
+ "\n"
|
|
" gpsrelay:\n"
|
|
" host: "
|
|
+ yaml_string(
|
|
data["gps"]["gpsrelay"]["host"]
|
|
)
|
|
+ "\n"
|
|
" port: "
|
|
+ str(
|
|
data["gps"]["gpsrelay"]["port"]
|
|
)
|
|
+ "\n"
|
|
" gnss:\n"
|
|
" handler: "
|
|
+ yaml_string(
|
|
data["gps"]["gnss"]["handler"]
|
|
)
|
|
+ "\n"
|
|
" direct:\n"
|
|
" device: "
|
|
+ yaml_string(
|
|
data["gps"]["gnss"]["direct"]["device"]
|
|
)
|
|
+ "\n"
|
|
" gpsd: {}\n"
|
|
" modem:\n"
|
|
" device: "
|
|
+ yaml_string(
|
|
data["gps"]["gnss"]["modem"]["device"]
|
|
)
|
|
+ "\n"
|
|
" at_device: "
|
|
+ yaml_string(
|
|
data["gps"]["gnss"]["modem"]["at_device"]
|
|
)
|
|
+ "\n"
|
|
)
|
|
|
|
text = (
|
|
text[:gps_match.start(2)]
|
|
+
|
|
gps_text
|
|
+
|
|
text[gps_match.end(2):]
|
|
)
|
|
|
|
for section, section_values in values.items():
|
|
|
|
section_pattern = (
|
|
r"(^"
|
|
+ re.escape(section)
|
|
+ r":[ \t]*\n)"
|
|
r"(.*?)(?=^[A-Za-z_][A-Za-z0-9_]*:[ \t]*$|\Z)"
|
|
)
|
|
|
|
match = re.search(
|
|
section_pattern,
|
|
text,
|
|
flags=re.MULTILINE | re.DOTALL
|
|
)
|
|
|
|
if not match:
|
|
|
|
raise RuntimeError(
|
|
f"Configuration section not found: {section}"
|
|
)
|
|
|
|
section_text = match.group(2)
|
|
|
|
for key, value in section_values.items():
|
|
|
|
pattern = (
|
|
r"(^[ \t]+"
|
|
+ re.escape(key)
|
|
+ r":[ \t]*)"
|
|
+ r"([^#\n]*?)"
|
|
+ r"([ \t]*#.*)?$"
|
|
)
|
|
|
|
section_text, count = re.subn(
|
|
pattern,
|
|
lambda item: (
|
|
item.group(1)
|
|
+
|
|
value
|
|
+
|
|
(
|
|
item.group(3)
|
|
if item.group(3)
|
|
else
|
|
""
|
|
)
|
|
),
|
|
section_text,
|
|
count=1,
|
|
flags=re.MULTILINE
|
|
)
|
|
|
|
if count != 1:
|
|
|
|
raise RuntimeError(
|
|
f"Configuration key not found: "
|
|
f"{section}.{key}"
|
|
)
|
|
|
|
text = (
|
|
text[:match.start(2)]
|
|
+
|
|
section_text
|
|
+
|
|
text[match.end(2):]
|
|
)
|
|
|
|
with open(
|
|
config_file,
|
|
"w"
|
|
) as file:
|
|
|
|
file.write(text)
|
|
|
|
def render_report(
|
|
render_function,
|
|
*args,
|
|
**kwargs
|
|
):
|
|
|
|
conn = connect_database()
|
|
|
|
try:
|
|
|
|
context = ReportContext(
|
|
mode="server"
|
|
)
|
|
|
|
html = render_function(
|
|
conn,
|
|
*args,
|
|
context=context,
|
|
**kwargs
|
|
)
|
|
|
|
return Response(
|
|
html,
|
|
mimetype="text/html"
|
|
)
|
|
|
|
finally:
|
|
close_database(conn)
|
|
|
|
@app.route("/reports/vendors/")
|
|
def reports_vendors():
|
|
|
|
config = read_config()
|
|
|
|
if "_error" in config:
|
|
|
|
return jsonify(
|
|
{
|
|
"error": config["_error"]
|
|
}
|
|
), 500
|
|
|
|
vendor_config = config.get(
|
|
"vendor",
|
|
{}
|
|
)
|
|
|
|
database_config = config.get(
|
|
"database",
|
|
{}
|
|
)
|
|
|
|
service = VendorService(
|
|
PROJECT_ROOT / vendor_config["database"],
|
|
PROJECT_ROOT / database_config["file"],
|
|
vendor_config["sources"],
|
|
vendor_config.get(
|
|
"mirrors",
|
|
{}
|
|
)
|
|
)
|
|
|
|
job_id = request.args.get(
|
|
"job_id"
|
|
)
|
|
|
|
job = None
|
|
preview = None
|
|
token = None
|
|
|
|
if job_id:
|
|
|
|
with vendor_update_lock:
|
|
|
|
if (
|
|
vendor_update_job is not None
|
|
and vendor_update_job["id"] == job_id
|
|
):
|
|
|
|
job = dict(
|
|
vendor_update_job
|
|
)
|
|
|
|
service = job["service"]
|
|
|
|
if job is not None:
|
|
|
|
token = job.get(
|
|
"preview_token"
|
|
)
|
|
|
|
if token:
|
|
|
|
with vendor_preview_lock:
|
|
|
|
preview = vendor_previews.get(
|
|
token
|
|
)
|
|
|
|
return render_report(
|
|
render_vendors_page,
|
|
service,
|
|
preview,
|
|
token,
|
|
job=job
|
|
)
|
|
|
|
def _run_vendor_update(
|
|
job_id,
|
|
service
|
|
):
|
|
|
|
global vendor_update_job
|
|
|
|
def progress_callback(
|
|
stage,
|
|
source=None,
|
|
message=None
|
|
):
|
|
|
|
with vendor_update_lock:
|
|
|
|
if (
|
|
vendor_update_job is None
|
|
or vendor_update_job["id"] != job_id
|
|
):
|
|
return
|
|
|
|
vendor_update_job["stage"] = stage
|
|
vendor_update_job["source"] = source
|
|
vendor_update_job["message"] = message
|
|
|
|
try:
|
|
|
|
preview = service.prepare_update(
|
|
progress_callback=progress_callback
|
|
)
|
|
|
|
token = secrets.token_urlsafe(
|
|
32
|
|
)
|
|
|
|
with vendor_preview_lock:
|
|
|
|
while len(vendor_previews) >= 5:
|
|
|
|
old_token, old_preview = (
|
|
next(
|
|
iter(
|
|
vendor_previews.items()
|
|
)
|
|
)
|
|
)
|
|
|
|
del vendor_previews[
|
|
old_token
|
|
]
|
|
|
|
service._cleanup_preview(
|
|
old_preview
|
|
)
|
|
|
|
vendor_previews[
|
|
token
|
|
] = preview
|
|
|
|
with vendor_update_lock:
|
|
|
|
if (
|
|
vendor_update_job is not None
|
|
and vendor_update_job["id"] == job_id
|
|
):
|
|
|
|
vendor_update_job["status"] = "ready"
|
|
vendor_update_job["stage"] = "ready"
|
|
vendor_update_job["source"] = None
|
|
vendor_update_job["message"] = (
|
|
"Обновление готово к применению"
|
|
)
|
|
vendor_update_job["preview_token"] = token
|
|
vendor_update_job["error"] = None
|
|
|
|
except Exception as error:
|
|
|
|
with vendor_update_lock:
|
|
|
|
if (
|
|
vendor_update_job is not None
|
|
and vendor_update_job["id"] == job_id
|
|
):
|
|
|
|
vendor_update_job["status"] = "error"
|
|
vendor_update_job["stage"] = "error"
|
|
vendor_update_job["message"] = None
|
|
vendor_update_job["error"] = str(
|
|
error
|
|
)
|
|
|
|
|
|
@app.route(
|
|
"/reports/vendors/update",
|
|
methods=["POST"]
|
|
)
|
|
def reports_vendors_update():
|
|
|
|
global vendor_update_job
|
|
|
|
config = read_config()
|
|
|
|
if "_error" in config:
|
|
|
|
return jsonify(
|
|
{
|
|
"error": config["_error"]
|
|
}
|
|
), 500
|
|
|
|
vendor_config = config.get(
|
|
"vendor",
|
|
{}
|
|
)
|
|
|
|
database_config = config.get(
|
|
"database",
|
|
{}
|
|
)
|
|
|
|
service = VendorService(
|
|
PROJECT_ROOT / vendor_config["database"],
|
|
PROJECT_ROOT / database_config["file"],
|
|
vendor_config["sources"],
|
|
vendor_config.get(
|
|
"mirrors",
|
|
{}
|
|
)
|
|
)
|
|
|
|
with vendor_update_lock:
|
|
|
|
if (
|
|
vendor_update_job is not None
|
|
and vendor_update_job["status"] == "running"
|
|
):
|
|
|
|
job = dict(
|
|
vendor_update_job
|
|
)
|
|
|
|
return render_report(
|
|
render_vendors_page,
|
|
job["service"],
|
|
job=job
|
|
)
|
|
|
|
job_id = secrets.token_urlsafe(
|
|
16
|
|
)
|
|
|
|
vendor_update_job = {
|
|
"id": job_id,
|
|
"status": "running",
|
|
"stage": "starting",
|
|
"source": None,
|
|
"message": "Подготовка обновления",
|
|
"error": None,
|
|
"preview_token": None,
|
|
"service": service
|
|
}
|
|
|
|
thread = threading.Thread(
|
|
target=_run_vendor_update,
|
|
args=(
|
|
job_id,
|
|
service
|
|
),
|
|
daemon=True
|
|
)
|
|
|
|
thread.start()
|
|
|
|
job = dict(
|
|
vendor_update_job
|
|
)
|
|
|
|
return render_report(
|
|
render_vendors_page,
|
|
service,
|
|
job=job
|
|
)
|
|
|
|
|
|
@app.route(
|
|
"/reports/vendors/update/status/<job_id>"
|
|
)
|
|
def reports_vendors_update_status(
|
|
job_id
|
|
):
|
|
|
|
with vendor_update_lock:
|
|
|
|
if (
|
|
vendor_update_job is None
|
|
or vendor_update_job["id"] != job_id
|
|
):
|
|
|
|
return jsonify(
|
|
{
|
|
"status": "missing",
|
|
"error":
|
|
"Vendor update job is missing."
|
|
}
|
|
), 404
|
|
|
|
job = dict(
|
|
vendor_update_job
|
|
)
|
|
|
|
return jsonify(
|
|
{
|
|
"status": job["status"],
|
|
"stage": job["stage"],
|
|
"source": job["source"],
|
|
"message": job["message"],
|
|
"error": job["error"],
|
|
"preview_ready":
|
|
job["preview_token"] is not None
|
|
}
|
|
)
|
|
|
|
@app.route(
|
|
"/reports/vendors/apply",
|
|
methods=["POST"]
|
|
)
|
|
def reports_vendors_apply():
|
|
|
|
token = request.form.get(
|
|
"token",
|
|
""
|
|
)
|
|
|
|
with vendor_preview_lock:
|
|
|
|
preview = vendor_previews.pop(
|
|
token,
|
|
None
|
|
)
|
|
|
|
if preview is None:
|
|
|
|
return jsonify(
|
|
{
|
|
"error":
|
|
"Vendor preview is missing or expired."
|
|
}
|
|
), 400
|
|
|
|
config = read_config()
|
|
|
|
if "_error" in config:
|
|
|
|
return jsonify(
|
|
{
|
|
"error": config["_error"]
|
|
}
|
|
), 500
|
|
|
|
vendor_config = config.get(
|
|
"vendor",
|
|
{}
|
|
)
|
|
|
|
database_config = config.get(
|
|
"database",
|
|
{}
|
|
)
|
|
|
|
service = VendorService(
|
|
PROJECT_ROOT / vendor_config["database"],
|
|
PROJECT_ROOT / database_config["file"],
|
|
vendor_config["sources"],
|
|
vendor_config.get(
|
|
"mirrors",
|
|
{}
|
|
)
|
|
)
|
|
|
|
selected = {
|
|
"unknown_to_vendor":
|
|
request.form.get(
|
|
"unknown_to_vendor"
|
|
) == "1",
|
|
"vendor_to_vendor":
|
|
request.form.get(
|
|
"vendor_to_vendor"
|
|
) == "1",
|
|
"vendor_to_unknown":
|
|
request.form.get(
|
|
"vendor_to_unknown"
|
|
) == "1"
|
|
}
|
|
|
|
try:
|
|
|
|
result = service.apply(
|
|
preview,
|
|
**selected
|
|
)
|
|
|
|
except Exception as error:
|
|
|
|
return render_report(
|
|
render_vendors_page,
|
|
service,
|
|
message={
|
|
"type": "error",
|
|
"text": str(error)
|
|
}
|
|
)
|
|
|
|
return render_report(
|
|
render_vendors_page,
|
|
service,
|
|
message={
|
|
"type": "success",
|
|
"text": (
|
|
"Update completed. "
|
|
f"OUI updated: "
|
|
f"{'yes' if result['oui_updated'] else 'no'}. "
|
|
f"WiFi AP records updated: "
|
|
f"{result['wifi_updated']}."
|
|
)
|
|
}
|
|
)
|
|
|
|
@app.route("/reports/")
|
|
def reports_dashboard():
|
|
|
|
return render_report(
|
|
generate_dashboard
|
|
)
|
|
|
|
|
|
@app.route("/reports/sessions/")
|
|
def reports_capture_sessions():
|
|
|
|
return render_report(
|
|
generate_capture_sessions
|
|
)
|
|
|
|
@app.route("/reports/trips/")
|
|
def reports_trips():
|
|
|
|
return render_report(
|
|
render_trips_page
|
|
)
|
|
|
|
|
|
@app.route("/reports/trips/export")
|
|
def reports_trips_export():
|
|
|
|
filename = (
|
|
"wifi-gps-mapper-"
|
|
f"{datetime.now(timezone.utc).strftime('%Y%m%d-%H%M%S')}.wifitrip"
|
|
)
|
|
|
|
temp_file = tempfile.NamedTemporaryFile(
|
|
prefix="wifitrip-",
|
|
suffix=".tmp",
|
|
delete=False,
|
|
)
|
|
|
|
temp_path = Path(
|
|
temp_file.name
|
|
)
|
|
|
|
temp_file.close()
|
|
|
|
output_path = (
|
|
temp_path.with_name(
|
|
filename
|
|
)
|
|
)
|
|
|
|
temp_path.unlink(
|
|
missing_ok=True
|
|
)
|
|
|
|
conn = connect_database()
|
|
|
|
try:
|
|
|
|
result = export_wifitrip(
|
|
conn,
|
|
output_path=output_path,
|
|
)
|
|
|
|
except Exception:
|
|
|
|
output_path.unlink(
|
|
missing_ok=True
|
|
)
|
|
|
|
raise
|
|
|
|
finally:
|
|
close_database(conn)
|
|
|
|
response = send_file(
|
|
output_path,
|
|
as_attachment=True,
|
|
download_name=output_path.name,
|
|
mimetype="application/zip",
|
|
)
|
|
|
|
response.headers[
|
|
"X-Wifitrip-Sessions"
|
|
] = str(result["sessions"])
|
|
|
|
response.headers[
|
|
"X-Wifitrip-Size"
|
|
] = str(result["size"])
|
|
|
|
response.headers[
|
|
"X-Wifitrip-Duration"
|
|
] = str(result["duration"])
|
|
|
|
response.call_on_close(
|
|
lambda: output_path.unlink(
|
|
missing_ok=True
|
|
)
|
|
)
|
|
|
|
return response
|
|
|
|
|
|
@app.route(
|
|
"/reports/trips/preflight",
|
|
methods=["POST"]
|
|
)
|
|
def reports_trips_preflight():
|
|
|
|
uploaded_file = request.files.get(
|
|
"wifitrip_file"
|
|
)
|
|
|
|
if uploaded_file is None:
|
|
return jsonify(
|
|
{
|
|
"valid": False,
|
|
"error":
|
|
"Wifitrip package is required."
|
|
}
|
|
), 400
|
|
|
|
temp_file = tempfile.NamedTemporaryFile(
|
|
suffix=".wifitrip",
|
|
delete=False,
|
|
)
|
|
|
|
package_path = Path(
|
|
temp_file.name
|
|
)
|
|
|
|
try:
|
|
|
|
uploaded_file.save(
|
|
temp_file
|
|
)
|
|
|
|
temp_file.close()
|
|
|
|
result = validate_wifitrip(
|
|
package_path
|
|
)
|
|
|
|
return jsonify(
|
|
result
|
|
)
|
|
|
|
except Exception as error:
|
|
|
|
return jsonify(
|
|
{
|
|
"valid": False,
|
|
"error": str(error),
|
|
}
|
|
), 400
|
|
|
|
finally:
|
|
|
|
if not temp_file.closed:
|
|
temp_file.close()
|
|
|
|
package_path.unlink(
|
|
missing_ok=True
|
|
)
|
|
|
|
|
|
@app.route(
|
|
"/reports/trips/import",
|
|
methods=["POST"]
|
|
)
|
|
def reports_trips_import():
|
|
|
|
uploaded_file = request.files.get(
|
|
"wifitrip_file"
|
|
)
|
|
|
|
if uploaded_file is None:
|
|
return render_report(
|
|
render_trips_page,
|
|
message={
|
|
"type": "error",
|
|
"text":
|
|
"Wifitrip package is required.",
|
|
}
|
|
), 400
|
|
|
|
temp_file = tempfile.NamedTemporaryFile(
|
|
suffix=".wifitrip",
|
|
delete=False,
|
|
)
|
|
|
|
package_path = Path(
|
|
temp_file.name
|
|
)
|
|
|
|
try:
|
|
|
|
uploaded_file.save(
|
|
temp_file
|
|
)
|
|
|
|
temp_file.close()
|
|
|
|
validation = validate_wifitrip(
|
|
package_path
|
|
)
|
|
|
|
if not validation.get("valid"):
|
|
|
|
return render_report(
|
|
render_trips_page,
|
|
validation=validation,
|
|
message={
|
|
"type": "error",
|
|
"text": (
|
|
"Import was rejected. "
|
|
"The package is invalid and "
|
|
"no sessions were imported."
|
|
),
|
|
}
|
|
), 400
|
|
|
|
conn = connect_database()
|
|
|
|
try:
|
|
|
|
result = import_wifitrip(
|
|
conn,
|
|
package_path,
|
|
)
|
|
|
|
finally:
|
|
close_database(conn)
|
|
|
|
if result["failed"]:
|
|
|
|
return render_report(
|
|
render_trips_page,
|
|
import_result=result,
|
|
validation=validation,
|
|
message={
|
|
"type": "error",
|
|
"text": (
|
|
"Import completed with "
|
|
"failed sessions."
|
|
),
|
|
}
|
|
), 400
|
|
|
|
return render_report(
|
|
render_trips_page,
|
|
import_result=result,
|
|
validation=validation,
|
|
message={
|
|
"type": "success",
|
|
"text": (
|
|
"Import completed successfully."
|
|
),
|
|
}
|
|
)
|
|
|
|
except Exception as error:
|
|
|
|
return render_report(
|
|
render_trips_page,
|
|
message={
|
|
"type": "error",
|
|
"text": (
|
|
"Import failed: "
|
|
f"{error}"
|
|
),
|
|
}
|
|
), 500
|
|
|
|
finally:
|
|
|
|
if not temp_file.closed:
|
|
temp_file.close()
|
|
|
|
package_path.unlink(
|
|
missing_ok=True
|
|
)
|
|
|
|
|
|
@app.route("/reports/session/<int:session_id>")
|
|
def reports_session(session_id):
|
|
|
|
return render_report(
|
|
render_session_page,
|
|
session_id
|
|
)
|
|
|
|
|
|
@app.route("/reports/access-points/")
|
|
def reports_access_points():
|
|
|
|
return render_report(
|
|
render_access_points_page
|
|
)
|
|
|
|
@app.route("/reports/access-points/<bssid>")
|
|
def reports_access_point(bssid):
|
|
|
|
return render_report(
|
|
render_access_point_page,
|
|
bssid
|
|
)
|
|
|
|
@app.route("/reports/hashcat/")
|
|
def reports_hashcat():
|
|
|
|
return render_report(
|
|
render_hashcat_page
|
|
)
|
|
|
|
|
|
@app.route("/reports/hashcat/export")
|
|
def reports_hashcat_export():
|
|
|
|
vendor = request.args.get(
|
|
"vendor",
|
|
""
|
|
).strip()
|
|
|
|
conn = connect_database()
|
|
|
|
try:
|
|
|
|
if vendor:
|
|
|
|
criteria = normalize_criteria(
|
|
vendor=vendor
|
|
)
|
|
|
|
if not criteria:
|
|
|
|
return jsonify(
|
|
{
|
|
"error":
|
|
"At least one search criterion "
|
|
"is required."
|
|
}
|
|
), 400
|
|
|
|
content = build_hash_export(
|
|
conn,
|
|
criteria
|
|
)
|
|
|
|
filename = build_export_filename(
|
|
"search",
|
|
vendor
|
|
)
|
|
|
|
else:
|
|
|
|
content = build_all_hash_export(
|
|
conn
|
|
)
|
|
|
|
filename = build_export_filename(
|
|
"all"
|
|
)
|
|
|
|
return Response(
|
|
content,
|
|
mimetype="text/plain",
|
|
headers={
|
|
"Content-Disposition":
|
|
f'attachment; filename="{filename}"'
|
|
}
|
|
)
|
|
|
|
except Exception as error:
|
|
|
|
return jsonify(
|
|
{
|
|
"error": str(error)
|
|
}
|
|
), 500
|
|
|
|
finally:
|
|
|
|
close_database(conn)
|
|
|
|
@app.route(
|
|
"/reports/hashcat/preflight",
|
|
methods=["POST"]
|
|
)
|
|
def reports_hashcat_preflight():
|
|
|
|
uploaded_file = request.files.get(
|
|
"hashcat_file"
|
|
)
|
|
|
|
if uploaded_file is None:
|
|
return jsonify(
|
|
{
|
|
"valid": False,
|
|
"error":
|
|
"Hashcat result file is required."
|
|
}
|
|
), 400
|
|
|
|
try:
|
|
content = uploaded_file.read().decode(
|
|
"utf-8"
|
|
)
|
|
except UnicodeDecodeError:
|
|
return jsonify(
|
|
{
|
|
"valid": False,
|
|
"error":
|
|
"Hashcat result file must be "
|
|
"valid UTF-8 text."
|
|
}
|
|
), 400
|
|
|
|
from reports.hashcat import (
|
|
validate_hashcat_show_file,
|
|
)
|
|
|
|
conn = connect_database()
|
|
|
|
try:
|
|
|
|
result = validate_hashcat_show_file(
|
|
conn,
|
|
content.splitlines(
|
|
keepends=True
|
|
)
|
|
)
|
|
|
|
result["valid"] = (
|
|
result["total_lines"] > 0
|
|
and result["matched_handshakes"] > 0
|
|
and result["invalid_lines"] == 0
|
|
)
|
|
|
|
return jsonify(
|
|
result
|
|
)
|
|
|
|
except Exception as error:
|
|
|
|
return jsonify(
|
|
{
|
|
"valid": False,
|
|
"error": str(error)
|
|
}
|
|
), 500
|
|
|
|
finally:
|
|
|
|
close_database(conn)
|
|
|
|
@app.route(
|
|
"/reports/hashcat/import",
|
|
methods=["POST"]
|
|
)
|
|
def reports_hashcat_import():
|
|
|
|
uploaded_file = request.files.get(
|
|
"hashcat_file"
|
|
)
|
|
|
|
if uploaded_file is None:
|
|
return jsonify(
|
|
{
|
|
"error":
|
|
"Hashcat result file is required."
|
|
}
|
|
), 400
|
|
|
|
try:
|
|
content = uploaded_file.read().decode(
|
|
"utf-8"
|
|
)
|
|
except UnicodeDecodeError:
|
|
return jsonify(
|
|
{
|
|
"error":
|
|
"Hashcat result file must be "
|
|
"valid UTF-8 text."
|
|
}
|
|
), 400
|
|
|
|
conn = connect_database()
|
|
|
|
try:
|
|
|
|
result = import_hashcat_show(
|
|
conn,
|
|
content.splitlines(
|
|
keepends=True
|
|
)
|
|
)
|
|
|
|
if result["errors"]:
|
|
|
|
html = render_hashcat_page(
|
|
conn,
|
|
context=ReportContext(
|
|
mode="server"
|
|
),
|
|
import_result=result,
|
|
message={
|
|
"type": "error",
|
|
"text": (
|
|
"Import was rejected. "
|
|
"No credentials were committed."
|
|
),
|
|
}
|
|
)
|
|
|
|
return Response(
|
|
html,
|
|
mimetype="text/html",
|
|
status=400
|
|
)
|
|
|
|
html = render_hashcat_page(
|
|
conn,
|
|
context=ReportContext(
|
|
mode="server"
|
|
),
|
|
import_result=result,
|
|
message={
|
|
"type": "success",
|
|
"text": (
|
|
"Import completed successfully."
|
|
),
|
|
}
|
|
)
|
|
|
|
return Response(
|
|
html,
|
|
mimetype="text/html"
|
|
)
|
|
|
|
except Exception as error:
|
|
|
|
return jsonify(
|
|
{
|
|
"error": str(error)
|
|
}
|
|
), 500
|
|
|
|
finally:
|
|
|
|
close_database(conn)
|
|
|
|
@app.route("/reports/search/")
|
|
def reports_search():
|
|
|
|
return render_report(
|
|
render_search_page
|
|
)
|
|
|
|
|
|
@app.route("/reports/search/data")
|
|
def reports_search_data():
|
|
|
|
criteria = normalize_criteria(
|
|
vendor=request.args.get(
|
|
"vendor",
|
|
""
|
|
)
|
|
)
|
|
|
|
if not criteria:
|
|
|
|
return jsonify(
|
|
{
|
|
"error":
|
|
"At least one search criterion is required."
|
|
}
|
|
), 400
|
|
|
|
conn = connect_database()
|
|
|
|
try:
|
|
|
|
data = build_search_data(
|
|
conn,
|
|
criteria
|
|
)
|
|
|
|
return jsonify(
|
|
data
|
|
)
|
|
|
|
except Exception as error:
|
|
|
|
return jsonify(
|
|
{
|
|
"error": str(error)
|
|
}
|
|
), 500
|
|
|
|
finally:
|
|
|
|
close_database(conn)
|
|
|
|
|
|
@app.route("/reports/search/export")
|
|
def reports_search_export():
|
|
|
|
criteria = normalize_criteria(
|
|
vendor=request.args.get(
|
|
"vendor",
|
|
""
|
|
)
|
|
)
|
|
|
|
if not criteria:
|
|
|
|
return jsonify(
|
|
{
|
|
"error":
|
|
"At least one search criterion is required."
|
|
}
|
|
), 400
|
|
|
|
conn = connect_database()
|
|
|
|
try:
|
|
|
|
content = build_hash_export(
|
|
conn,
|
|
criteria
|
|
)
|
|
|
|
filename = build_export_filename(
|
|
"search",
|
|
request.args.get(
|
|
"vendor",
|
|
""
|
|
)
|
|
)
|
|
|
|
return Response(
|
|
content,
|
|
mimetype="text/plain",
|
|
headers={
|
|
"Content-Disposition":
|
|
f'attachment; filename="{filename}"'
|
|
}
|
|
)
|
|
|
|
except Exception as error:
|
|
|
|
return jsonify(
|
|
{
|
|
"error": str(error)
|
|
}
|
|
), 500
|
|
|
|
finally:
|
|
|
|
close_database(conn)
|
|
|
|
|
|
@app.route("/reports/static/<path:filename>")
|
|
def reports_static(filename):
|
|
|
|
return send_from_directory(
|
|
PROJECT_ROOT / "reports" / "static",
|
|
filename
|
|
)
|
|
|
|
@app.route("/")
|
|
def index():
|
|
|
|
config = read_config()
|
|
|
|
interfaces, interface_error = (
|
|
get_wifi_interfaces()
|
|
)
|
|
|
|
status = get_scanner_status()
|
|
gps_test = get_gps_test_status()
|
|
|
|
with scanner_lock:
|
|
|
|
output = list(
|
|
scanner_output
|
|
)
|
|
|
|
return render_template(
|
|
"index.html",
|
|
config=config,
|
|
interfaces=interfaces,
|
|
interface_error=interface_error,
|
|
status=status,
|
|
gps_test=gps_test,
|
|
output=output
|
|
)
|
|
|
|
|
|
@app.route(
|
|
"/api/status"
|
|
)
|
|
def api_status():
|
|
|
|
status = get_scanner_status()
|
|
gps_test = get_gps_test_status()
|
|
|
|
with scanner_lock:
|
|
|
|
output = list(
|
|
scanner_output
|
|
)
|
|
|
|
return jsonify(
|
|
{
|
|
"status": status,
|
|
"gps_test": gps_test,
|
|
"output": output
|
|
}
|
|
)
|
|
|
|
@app.route(
|
|
"/gps-test/start",
|
|
methods=["POST"]
|
|
)
|
|
def gps_test_start():
|
|
|
|
global gps_test_service
|
|
|
|
if get_scanner_status()["running"]:
|
|
|
|
with scanner_lock:
|
|
|
|
scanner_output.append(
|
|
"[GUI] GPS test cannot start "
|
|
"while scanner is running."
|
|
)
|
|
|
|
return jsonify(
|
|
{
|
|
"ok": False,
|
|
"error":
|
|
"GPS test cannot start "
|
|
"while scanner is running."
|
|
}
|
|
), 409
|
|
|
|
with gps_test_lock:
|
|
|
|
if (
|
|
gps_test_service is not None
|
|
and
|
|
gps_test_service.running
|
|
):
|
|
|
|
return jsonify(
|
|
{
|
|
"ok": False,
|
|
"error":
|
|
"GPS test is already running."
|
|
}
|
|
), 409
|
|
|
|
try:
|
|
|
|
cfg = Config()
|
|
|
|
gps_test_service = GPSTestService(
|
|
cfg,
|
|
callback=gps_test_output
|
|
)
|
|
|
|
if not gps_test_service.start():
|
|
|
|
gps_test_service = None
|
|
|
|
return jsonify(
|
|
{
|
|
"ok": False,
|
|
"error":
|
|
"Failed to start GPS test."
|
|
}
|
|
), 500
|
|
|
|
except Exception as error:
|
|
|
|
gps_test_service = None
|
|
|
|
with scanner_lock:
|
|
|
|
scanner_output.append(
|
|
f"[GUI] Failed to start GPS test: "
|
|
f"{error}"
|
|
)
|
|
|
|
return jsonify(
|
|
{
|
|
"ok": False,
|
|
"error": str(error)
|
|
}
|
|
), 500
|
|
|
|
return jsonify(
|
|
{
|
|
"ok": True
|
|
}
|
|
)
|
|
|
|
|
|
@app.route(
|
|
"/gps-test/stop",
|
|
methods=["POST"]
|
|
)
|
|
def gps_test_stop():
|
|
|
|
with gps_test_lock:
|
|
|
|
service = gps_test_service
|
|
|
|
if (
|
|
service is None
|
|
or
|
|
not service.running
|
|
):
|
|
|
|
return jsonify(
|
|
{
|
|
"ok": False,
|
|
"error":
|
|
"GPS test is not running."
|
|
}
|
|
), 409
|
|
|
|
service.stop()
|
|
|
|
if (
|
|
service.thread is not None
|
|
and
|
|
service.thread is not threading.current_thread()
|
|
):
|
|
|
|
service.thread.join(
|
|
timeout=2
|
|
)
|
|
|
|
with scanner_lock:
|
|
|
|
scanner_output.append(
|
|
"[GUI] GPS test stop requested."
|
|
)
|
|
|
|
return jsonify(
|
|
{
|
|
"ok": True
|
|
}
|
|
)
|
|
|
|
|
|
@app.route(
|
|
"/api/gps-test"
|
|
)
|
|
def api_gps_test():
|
|
|
|
return jsonify(
|
|
get_gps_test_status()
|
|
)
|
|
|
|
@app.route(
|
|
"/api/wifi/channels"
|
|
)
|
|
def api_wifi_channels():
|
|
|
|
interface = request.args.get(
|
|
"interface",
|
|
""
|
|
).strip()
|
|
|
|
|
|
if not interface:
|
|
|
|
return jsonify(
|
|
{
|
|
"2.4": [],
|
|
"5": [],
|
|
"6": [],
|
|
"error": "WiFi interface was not selected."
|
|
}
|
|
), 400
|
|
|
|
|
|
interfaces, interface_error = (
|
|
get_wifi_interfaces()
|
|
)
|
|
|
|
|
|
if interface_error:
|
|
|
|
return jsonify(
|
|
{
|
|
"2.4": [],
|
|
"5": [],
|
|
"6": [],
|
|
"error": interface_error
|
|
}
|
|
), 500
|
|
|
|
|
|
if interface not in interfaces:
|
|
|
|
return jsonify(
|
|
{
|
|
"2.4": [],
|
|
"5": [],
|
|
"6": [],
|
|
"error": "Invalid WiFi interface."
|
|
}
|
|
), 400
|
|
|
|
|
|
return jsonify(
|
|
get_wifi_channels(interface)
|
|
)
|
|
|
|
@app.route(
|
|
"/config/save",
|
|
methods=["POST"]
|
|
)
|
|
def config_save():
|
|
|
|
if get_scanner_status()["running"]:
|
|
|
|
with scanner_lock:
|
|
|
|
scanner_output.append(
|
|
"[GUI] Configuration cannot be "
|
|
"saved while scanner is running."
|
|
)
|
|
|
|
return jsonify(
|
|
{
|
|
"ok": False,
|
|
"error":
|
|
"Configuration cannot be saved "
|
|
"while scanner is running."
|
|
}
|
|
), 409
|
|
|
|
if get_gps_test_status()["running"]:
|
|
|
|
with scanner_lock:
|
|
scanner_output.append(
|
|
"[GUI] Configuration cannot be "
|
|
"saved while GPS test is running."
|
|
)
|
|
|
|
return jsonify(
|
|
{
|
|
"ok": False,
|
|
"error":
|
|
"Configuration cannot be saved "
|
|
"while GPS test is running."
|
|
}
|
|
), 409
|
|
|
|
|
|
try:
|
|
|
|
data = request.get_json(
|
|
silent=True
|
|
)
|
|
|
|
if not data:
|
|
|
|
raise RuntimeError(
|
|
"No configuration data received."
|
|
)
|
|
|
|
save_config(data)
|
|
|
|
with scanner_lock:
|
|
|
|
scanner_output.append(
|
|
"[GUI] Configuration saved."
|
|
)
|
|
|
|
return jsonify(
|
|
{
|
|
"ok": True,
|
|
"message":
|
|
"Configuration saved successfully."
|
|
}
|
|
)
|
|
|
|
|
|
except Exception as error:
|
|
|
|
with scanner_lock:
|
|
|
|
scanner_output.append(
|
|
f"[GUI] Failed to save configuration: "
|
|
f"{error}"
|
|
)
|
|
|
|
return jsonify(
|
|
{
|
|
"ok": False,
|
|
"error": str(error)
|
|
}
|
|
), 500
|
|
|
|
|
|
@app.route(
|
|
"/scanner/start",
|
|
methods=["POST"]
|
|
)
|
|
def scanner_start():
|
|
|
|
global scanner_process
|
|
global scanner_health
|
|
global scanner_stopped_at
|
|
|
|
with scanner_lock:
|
|
|
|
if (
|
|
scanner_process is not None
|
|
and
|
|
scanner_process.poll() is None
|
|
):
|
|
|
|
return redirect(
|
|
url_for("index")
|
|
)
|
|
|
|
if get_gps_test_status()["running"]:
|
|
|
|
with scanner_lock:
|
|
|
|
scanner_output.append(
|
|
"[GUI] Scanner cannot start "
|
|
"while GPS test is running."
|
|
)
|
|
|
|
return redirect(
|
|
url_for("index")
|
|
)
|
|
|
|
interface = request.form.get(
|
|
"interface",
|
|
""
|
|
).strip()
|
|
|
|
|
|
if not interface:
|
|
|
|
with scanner_lock:
|
|
|
|
scanner_output.append(
|
|
"[GUI] WiFi interface was not selected."
|
|
)
|
|
|
|
return redirect(
|
|
url_for("index")
|
|
)
|
|
|
|
|
|
environment = os.environ.copy()
|
|
|
|
environment[
|
|
"WIFI_INTERFACE"
|
|
] = interface
|
|
|
|
|
|
command = [
|
|
sys.executable,
|
|
"src/main.py"
|
|
]
|
|
|
|
|
|
try:
|
|
|
|
process = subprocess.Popen(
|
|
|
|
command,
|
|
|
|
cwd=PROJECT_ROOT,
|
|
|
|
env=environment,
|
|
|
|
stdin=subprocess.DEVNULL,
|
|
|
|
stdout=subprocess.PIPE,
|
|
|
|
stderr=subprocess.STDOUT,
|
|
|
|
text=True,
|
|
|
|
bufsize=1,
|
|
|
|
process_group=0
|
|
|
|
)
|
|
|
|
except Exception as error:
|
|
|
|
with scanner_lock:
|
|
|
|
scanner_output.append(
|
|
f"[GUI] Failed to start scanner: {error}"
|
|
)
|
|
|
|
return redirect(
|
|
url_for("index")
|
|
)
|
|
|
|
|
|
with scanner_lock:
|
|
|
|
scanner_process = process
|
|
|
|
scanner_health = "ACTIVE"
|
|
|
|
scanner_stopped_at = None
|
|
|
|
scanner_output.clear()
|
|
|
|
scanner_output.append(
|
|
"[GUI] Scanner started."
|
|
)
|
|
|
|
scanner_output.append(
|
|
f"[GUI] PID: {process.pid}"
|
|
)
|
|
|
|
scanner_output.append(
|
|
f"[GUI] Interface: {interface}"
|
|
)
|
|
|
|
|
|
thread = threading.Thread(
|
|
target=read_scanner_output,
|
|
args=(process,),
|
|
daemon=True
|
|
)
|
|
|
|
thread.start()
|
|
|
|
|
|
return redirect(
|
|
url_for("index")
|
|
)
|
|
|
|
|
|
@app.route(
|
|
"/scanner/stop",
|
|
methods=["POST"]
|
|
)
|
|
def scanner_stop():
|
|
|
|
global scanner_health
|
|
|
|
with scanner_lock:
|
|
|
|
process = scanner_process
|
|
|
|
if (
|
|
process is None
|
|
or
|
|
process.poll() is not None
|
|
):
|
|
return redirect(
|
|
url_for("index")
|
|
)
|
|
|
|
scanner_health = "STOPPING"
|
|
|
|
scanner_output.append(
|
|
"[GUI] SIGINT sent to scanner."
|
|
)
|
|
|
|
try:
|
|
|
|
os.killpg(
|
|
process.pid,
|
|
signal.SIGINT
|
|
)
|
|
|
|
except Exception as error:
|
|
|
|
with scanner_lock:
|
|
|
|
scanner_health = "ACTIVE"
|
|
|
|
scanner_output.append(
|
|
f"[GUI] Failed to stop scanner: {error}"
|
|
)
|
|
|
|
return redirect(
|
|
url_for("index")
|
|
)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
|
|
print(
|
|
"Checking sudo privileges..."
|
|
)
|
|
|
|
if not check_sudo():
|
|
|
|
print(
|
|
"sudo authentication failed."
|
|
)
|
|
|
|
raise SystemExit(1)
|
|
|
|
|
|
print(
|
|
"sudo: OK"
|
|
)
|
|
|
|
|
|
thread = threading.Thread(
|
|
target=sudo_keepalive,
|
|
daemon=True
|
|
)
|
|
|
|
thread.start()
|
|
|
|
|
|
logging.getLogger("werkzeug").disabled = True
|
|
|
|
app.run(
|
|
host="127.0.0.1",
|
|
port=5000,
|
|
debug=False
|
|
) |