import html
import json
from reports.database import query_all, query_one
from reports.template import (
ReportContext,
card,
page_begin,
page_end,
page_header,
)
from reports.utils import (
format_datetime as format_report_datetime,
)
def esc(value):
if value is None:
return "—"
return html.escape(
str(value),
quote=True,
)
def format_datetime(value):
return format_report_datetime(
value,
"%d.%m.%y-%H.%M.%S"
)
def format_popup_datetime(value):
return format_report_datetime(
value,
"%d-%m-%Y-%H-%M-%S"
)
def load_report_interval(conn):
row = query_one(
conn,
"""
SELECT
MIN(start_time) AS first_time,
MAX(
COALESCE(
end_time,
start_time
)
) AS last_time
FROM capture_sessions
""",
)
if not row:
return (
None,
None,
)
return (
row["first_time"],
row["last_time"],
)
def load_access_points(conn):
rows = query_all(
conn,
"""
WITH ranked_observations AS (
SELECT
obs.access_point_id,
obs.latitude,
obs.longitude,
obs.observed_at,
obs.rssi,
obs.channel,
ROW_NUMBER() OVER (
PARTITION BY obs.access_point_id
ORDER BY
obs.observed_at DESC,
obs.id DESC
) AS rn
FROM access_point_observations AS obs
WHERE
obs.latitude IS NOT NULL
AND obs.longitude IS NOT NULL
AND obs.latitude != 0
AND obs.longitude != 0
)
SELECT
ap.id,
ap.bssid,
ap.essid,
ap.vendor,
ap.has_handshake,
ap.has_pmkid,
ap.is_cracked,
obs.latitude,
obs.longitude,
obs.observed_at,
obs.rssi,
obs.channel,
(
SELECT c.password
FROM credentials AS c
WHERE c.access_point_id = ap.id
ORDER BY
c.created_at DESC,
c.id DESC
LIMIT 1
) AS password
FROM access_points AS ap
JOIN ranked_observations AS obs
ON obs.access_point_id = ap.id
AND obs.rn = 1
ORDER BY
obs.observed_at DESC,
ap.bssid
""",
)
return rows
def build_marker_js(observations):
ap_features = []
handshake_features = []
pmkid_features = []
password_features = []
for row in observations:
latitude = row["latitude"]
longitude = row["longitude"]
if (
latitude is None
or longitude is None
):
continue
ssid = (
row["essid"]
if row["essid"] is not None
and str(row["essid"]) != ""
else "—"
)
vendor = (
row["vendor"]
if row["vendor"] is not None
and str(row["vendor"]) != ""
else "—"
)
channel = (
row["channel"]
if row["channel"] is not None
else "—"
)
rssi = (
f"{row['rssi']} dBm"
if row["rssi"] is not None
else "—"
)
password = (
row["password"]
if row["password"] is not None
and str(row["password"]) != ""
else "—"
)
popup = (
f"{esc(ssid)}"
f"
MAC: {esc(row['bssid'])}"
f"
Vendor: {esc(vendor)}"
f"
Channel: {esc(channel)}"
f"
RSSI: {esc(rssi)}"
f"
Password: {esc(password)}"
f"
Date: "
f"{esc(format_popup_datetime(row['observed_at']))}"
f'
'
f"Подробнее"
)
feature = {
"type": "Feature",
"geometry": {
"type": "Point",
"coordinates": [
float(longitude),
float(latitude),
],
},
"properties": {
"popup": popup,
},
}
ap_features.append(feature)
if row["has_handshake"]:
handshake_features.append(
feature
)
if row["has_pmkid"]:
pmkid_features.append(
feature
)
if row["is_cracked"] or row["password"]:
password_features.append(
feature
)
return (
"const apFeatures = "
+ json.dumps(
ap_features,
ensure_ascii=False,
)
+ ";\n"
"const handshakeFeatures = "
+ json.dumps(
handshake_features,
ensure_ascii=False,
)
+ ";\n"
"const pmkidFeatures = "
+ json.dumps(
pmkid_features,
ensure_ascii=False,
)
+ ";\n"
"const passwordFeatures = "
+ json.dumps(
password_features,
ensure_ascii=False,
)
+ ";\n"
)
def render_access_points_map(
observations,
context,
):
marker_js = build_marker_js(
observations
)
maplibre_css_url = context.asset_url(
"maplibre/maplibre-gl.css"
)
maplibre_js_url = context.asset_url(
"maplibre/maplibre-gl.js"
)
map_html = """
"""
map_html = map_html.replace(
"__MAPLIBRE_CSS_URL__",
esc(maplibre_css_url),
)
map_html = map_html.replace(
"__MAPLIBRE_JS_URL__",
esc(maplibre_js_url),
)
map_html = map_html.replace(
"__MARKER_JS__",
marker_js,
)
return map_html
def render_access_points_page(
conn,
context=None,
):
if context is None:
context = ReportContext(
mode="server"
)
first_time, last_time = (
load_report_interval(conn)
)
observations = load_access_points(
conn
)
html = page_begin(
"📡 Access Points",
active="access_points",
context=context,
)
html += f"""
Time interval:
{esc(format_datetime(first_time))}
—
{esc(format_datetime(last_time))}
"""
html += card(
"""
🗺 Capture Map
""",
render_access_points_map(
observations,
context,
),
)
html += page_end()
return html