1537 lines
38 KiB
Python
1537 lines
38 KiB
Python
import html
|
||
import json
|
||
import math
|
||
from collections import defaultdict
|
||
|
||
from reports.database import query_all, query_one
|
||
from reports.utils import (
|
||
format_datetime as format_report_datetime,
|
||
)
|
||
|
||
from reports.template import (
|
||
ReportContext,
|
||
card,
|
||
page_begin,
|
||
page_end,
|
||
page_header,
|
||
)
|
||
|
||
|
||
def esc(value):
|
||
return html.escape("" if value is None else str(value), quote=True)
|
||
|
||
|
||
def format_datetime(value):
|
||
return format_report_datetime(
|
||
value,
|
||
"%d.%m.%y-%H.%M.%S"
|
||
)
|
||
|
||
|
||
def format_number(value, digits=1):
|
||
if value is None:
|
||
return "—"
|
||
|
||
try:
|
||
return f"{float(value):.{digits}f}"
|
||
except (TypeError, ValueError):
|
||
return str(value)
|
||
|
||
|
||
def format_speed(value):
|
||
if value is None:
|
||
return "—"
|
||
|
||
try:
|
||
speed = float(value)
|
||
except (TypeError, ValueError):
|
||
return str(value)
|
||
|
||
if abs(speed) < 0.05:
|
||
return "0"
|
||
|
||
return f"{speed:.1f}"
|
||
|
||
|
||
def haversine_m(lat1, lon1, lat2, lon2):
|
||
radius = 6371000.0
|
||
|
||
phi1 = math.radians(lat1)
|
||
phi2 = math.radians(lat2)
|
||
dphi = math.radians(lat2 - lat1)
|
||
dlambda = math.radians(lon2 - lon1)
|
||
|
||
a = (
|
||
math.sin(dphi / 2.0) ** 2
|
||
+ math.cos(phi1)
|
||
* math.cos(phi2)
|
||
* math.sin(dlambda / 2.0) ** 2
|
||
)
|
||
|
||
return 2.0 * radius * math.asin(min(1.0, math.sqrt(a)))
|
||
|
||
|
||
def weighted_mean(values):
|
||
if not values:
|
||
return None
|
||
|
||
total_weight = sum(weight for _, weight in values)
|
||
|
||
if total_weight <= 0:
|
||
return sum(value for value, _ in values) / len(values)
|
||
|
||
return sum(value * weight for value, weight in values) / total_weight
|
||
|
||
|
||
def percentile(values, fraction):
|
||
if not values:
|
||
return None
|
||
|
||
ordered = sorted(values)
|
||
|
||
if len(ordered) == 1:
|
||
return ordered[0]
|
||
|
||
position = (len(ordered) - 1) * fraction
|
||
lower = int(math.floor(position))
|
||
upper = int(math.ceil(position))
|
||
|
||
if lower == upper:
|
||
return ordered[lower]
|
||
|
||
ratio = position - lower
|
||
return ordered[lower] + (ordered[upper] - ordered[lower]) * ratio
|
||
|
||
|
||
def observation_weight(rssi):
|
||
"""
|
||
RSSI is used only as a relative confidence weight.
|
||
|
||
We deliberately do not convert RSSI to a physical distance because
|
||
environmental attenuation, antenna characteristics, reflections and
|
||
transmitter power make such a conversion unreliable for this task.
|
||
"""
|
||
if rssi is None:
|
||
return 1.0
|
||
|
||
try:
|
||
value = float(rssi)
|
||
except (TypeError, ValueError):
|
||
return 1.0
|
||
|
||
# Keep the influence deliberately moderate.
|
||
#
|
||
# - -90 dBm -> 1
|
||
# - -70 dBm -> 2
|
||
# - -50 dBm -> 4
|
||
#
|
||
# This makes stronger observations useful without allowing a handful
|
||
# of unusually strong readings to dominate the whole history.
|
||
normalized = max(0.0, min(40.0, value + 90.0))
|
||
return 1.0 + normalized / 10.0
|
||
|
||
|
||
def valid_observations(rows):
|
||
result = []
|
||
|
||
for row in rows:
|
||
try:
|
||
latitude = float(row["latitude"])
|
||
longitude = float(row["longitude"])
|
||
except (TypeError, ValueError):
|
||
continue
|
||
|
||
if not (-90.0 <= latitude <= 90.0):
|
||
continue
|
||
|
||
if not (-180.0 <= longitude <= 180.0):
|
||
continue
|
||
|
||
if abs(latitude) < 1e-12 and abs(longitude) < 1e-12:
|
||
continue
|
||
|
||
result.append(
|
||
{
|
||
"id": row["id"],
|
||
"session_id": row["session_id"],
|
||
"observed_at": row["observed_at"],
|
||
"latitude": latitude,
|
||
"longitude": longitude,
|
||
"speed": row["speed"],
|
||
"rssi": row["rssi"],
|
||
"channel": row["channel"],
|
||
"frequency": row["frequency"],
|
||
"essid": row["essid"],
|
||
"encryption": row["encryption"],
|
||
"cipher": row["cipher"],
|
||
"akm": row["akm"],
|
||
"country": row["country"],
|
||
}
|
||
)
|
||
|
||
return result
|
||
|
||
def normalize_bursts(observations):
|
||
"""
|
||
Treat repeated observations at the same time and position as one
|
||
independent spatial observation.
|
||
|
||
RSSI values inside a burst are averaged so that short-lived RSSI
|
||
fluctuations do not give duplicate rows additional influence.
|
||
"""
|
||
if not observations:
|
||
return observations
|
||
|
||
groups = {}
|
||
|
||
for item in observations:
|
||
key = (
|
||
item["session_id"],
|
||
item["observed_at"],
|
||
item["latitude"],
|
||
item["longitude"],
|
||
)
|
||
|
||
group = groups.setdefault(
|
||
key,
|
||
{
|
||
"item": item,
|
||
"rssi_values": [],
|
||
},
|
||
)
|
||
|
||
if item["rssi"] is not None:
|
||
try:
|
||
group["rssi_values"].append(float(item["rssi"]))
|
||
except (TypeError, ValueError):
|
||
pass
|
||
|
||
result = []
|
||
|
||
for group in groups.values():
|
||
item = dict(group["item"])
|
||
rssi_values = group["rssi_values"]
|
||
|
||
if rssi_values:
|
||
item["rssi"] = sum(rssi_values) / len(rssi_values)
|
||
|
||
result.append(item)
|
||
|
||
return result
|
||
|
||
def local_projection(observations):
|
||
"""
|
||
Convert geographic coordinates to a local metre-based coordinate system.
|
||
|
||
x = east/west
|
||
y = north/south
|
||
"""
|
||
if not observations:
|
||
return None, []
|
||
|
||
center_lat = sum(item["latitude"] for item in observations) / len(observations)
|
||
center_lon = sum(item["longitude"] for item in observations) / len(observations)
|
||
|
||
cos_lat = math.cos(math.radians(center_lat))
|
||
|
||
if abs(cos_lat) < 1e-8:
|
||
cos_lat = 1e-8
|
||
|
||
projected = []
|
||
|
||
meters_per_degree = 111320.0
|
||
|
||
for item in observations:
|
||
x = (
|
||
(item["longitude"] - center_lon)
|
||
* meters_per_degree
|
||
* cos_lat
|
||
)
|
||
y = (item["latitude"] - center_lat) * meters_per_degree
|
||
|
||
projected.append((item, x, y))
|
||
|
||
return (center_lat, center_lon), projected
|
||
|
||
|
||
def weighted_center(observations):
|
||
if not observations:
|
||
return None
|
||
|
||
center, projected = local_projection(observations)
|
||
|
||
weighted_x = []
|
||
weighted_y = []
|
||
|
||
for item, x, y in projected:
|
||
weight = observation_weight(item["rssi"])
|
||
weighted_x.append((x, weight))
|
||
weighted_y.append((y, weight))
|
||
|
||
x = weighted_mean(weighted_x)
|
||
y = weighted_mean(weighted_y)
|
||
|
||
center_lat, center_lon = center
|
||
meters_per_degree = 111320.0
|
||
cos_lat = math.cos(math.radians(center_lat))
|
||
|
||
if abs(cos_lat) < 1e-8:
|
||
cos_lat = 1e-8
|
||
|
||
latitude = center_lat + y / meters_per_degree
|
||
longitude = center_lon + x / (meters_per_degree * cos_lat)
|
||
|
||
return {
|
||
"latitude": latitude,
|
||
"longitude": longitude,
|
||
"x": x,
|
||
"y": y,
|
||
}
|
||
|
||
|
||
def robust_observations(observations):
|
||
"""
|
||
Remove only strong spatial outliers.
|
||
|
||
The method intentionally has a conservative threshold. We do not want
|
||
to destroy normal driving tracks merely because the AP was observed
|
||
from several nearby positions.
|
||
"""
|
||
if len(observations) < 8:
|
||
return observations
|
||
|
||
center = weighted_center(observations)
|
||
|
||
if not center:
|
||
return observations
|
||
|
||
distances = []
|
||
|
||
for item in observations:
|
||
distance = haversine_m(
|
||
center["latitude"],
|
||
center["longitude"],
|
||
item["latitude"],
|
||
item["longitude"],
|
||
)
|
||
distances.append((item, distance))
|
||
|
||
numeric_distances = [distance for _, distance in distances]
|
||
|
||
median = percentile(numeric_distances, 0.50)
|
||
p90 = percentile(numeric_distances, 0.90)
|
||
|
||
if median is None or p90 is None:
|
||
return observations
|
||
|
||
# Robust scale estimate.
|
||
deviations = [abs(distance - median) for distance in numeric_distances]
|
||
mad = percentile(deviations, 0.50) or 0.0
|
||
|
||
robust_limit = median + max(3.0 * mad, 0.0)
|
||
|
||
# Never trim inside the central 90% envelope.
|
||
limit = max(robust_limit, p90)
|
||
|
||
# Avoid creating an artificially tiny area.
|
||
minimum_limit = max(75.0, median * 1.5)
|
||
limit = max(limit, minimum_limit)
|
||
|
||
filtered = [
|
||
item
|
||
for item, distance in distances
|
||
if distance <= limit
|
||
]
|
||
|
||
# If the filter became too aggressive, retain the original data.
|
||
if len(filtered) < max(5, int(len(observations) * 0.60)):
|
||
return observations
|
||
|
||
return filtered
|
||
|
||
|
||
def covariance_ellipse(observations):
|
||
"""
|
||
Calculate an oriented uncertainty ellipse in metres.
|
||
|
||
The ellipse describes the spatial dispersion of the useful observations,
|
||
not a guaranteed physical reception radius of the AP.
|
||
"""
|
||
if len(observations) < 2:
|
||
return None
|
||
|
||
center, projected = local_projection(observations)
|
||
|
||
weights = [
|
||
observation_weight(item["rssi"])
|
||
for item, _, _ in projected
|
||
]
|
||
|
||
total_weight = sum(weights)
|
||
|
||
if total_weight <= 0:
|
||
return None
|
||
|
||
mean_x = sum(x * weight for (_, x, _), weight in zip(projected, weights))
|
||
mean_y = sum(y * weight for (_, _, y), weight in zip(projected, weights))
|
||
|
||
mean_x /= total_weight
|
||
mean_y /= total_weight
|
||
|
||
covariance_xx = 0.0
|
||
covariance_xy = 0.0
|
||
covariance_yy = 0.0
|
||
|
||
for (_, x, y), weight in zip(projected, weights):
|
||
dx = x - mean_x
|
||
dy = y - mean_y
|
||
|
||
covariance_xx += weight * dx * dx
|
||
covariance_xy += weight * dx * dy
|
||
covariance_yy += weight * dy * dy
|
||
|
||
covariance_xx /= total_weight
|
||
covariance_xy /= total_weight
|
||
covariance_yy /= total_weight
|
||
|
||
trace = covariance_xx + covariance_yy
|
||
determinant = (
|
||
covariance_xx * covariance_yy
|
||
- covariance_xy * covariance_xy
|
||
)
|
||
|
||
determinant = max(0.0, determinant)
|
||
|
||
discriminant = max(
|
||
0.0,
|
||
trace * trace - 4.0 * determinant,
|
||
)
|
||
|
||
lambda1 = max(
|
||
0.0,
|
||
(trace + math.sqrt(discriminant)) / 2.0,
|
||
)
|
||
|
||
lambda2 = max(
|
||
0.0,
|
||
(trace - math.sqrt(discriminant)) / 2.0,
|
||
)
|
||
|
||
if lambda1 <= 0.0:
|
||
return None
|
||
|
||
if abs(covariance_xy) > 1e-9:
|
||
axis_x = lambda1 - covariance_yy
|
||
axis_y = covariance_xy
|
||
elif covariance_xx >= covariance_yy:
|
||
axis_x = 1.0
|
||
axis_y = 0.0
|
||
else:
|
||
axis_x = 0.0
|
||
axis_y = 1.0
|
||
|
||
axis_length = math.hypot(axis_x, axis_y)
|
||
|
||
if axis_length <= 1e-9:
|
||
return None
|
||
|
||
axis_x /= axis_length
|
||
axis_y /= axis_length
|
||
|
||
angle = math.degrees(math.atan2(axis_x, axis_y))
|
||
|
||
# Approximately 90% confidence ellipse for a 2D Gaussian.
|
||
scale = math.sqrt(4.605170186)
|
||
|
||
semi_major = math.sqrt(lambda1) * scale
|
||
semi_minor = math.sqrt(lambda2) * scale
|
||
|
||
# A tiny covariance can occur when the same point was observed many
|
||
# times. Keep the visualization visible but do not claim centimetre
|
||
# accuracy.
|
||
semi_major = max(25.0, semi_major)
|
||
semi_minor = max(20.0, semi_minor)
|
||
|
||
# Prevent visually misleading enormous ellipses.
|
||
semi_major = min(5000.0, semi_major)
|
||
semi_minor = min(5000.0, semi_minor)
|
||
|
||
center_lat, center_lon = center
|
||
meters_per_degree = 111320.0
|
||
cos_lat = math.cos(math.radians(center_lat))
|
||
|
||
if abs(cos_lat) < 1e-8:
|
||
cos_lat = 1e-8
|
||
|
||
latitude = center_lat + mean_y / meters_per_degree
|
||
longitude = center_lon + mean_x / (meters_per_degree * cos_lat)
|
||
|
||
return {
|
||
"latitude": latitude,
|
||
"longitude": longitude,
|
||
"semi_major": semi_major,
|
||
"semi_minor": semi_minor,
|
||
"angle": angle,
|
||
}
|
||
|
||
|
||
def ellipse_polygon(ellipse, points=64):
|
||
if not ellipse:
|
||
return []
|
||
|
||
lat0 = ellipse["latitude"]
|
||
lon0 = ellipse["longitude"]
|
||
|
||
angle = math.radians(ellipse["angle"])
|
||
cos_lat = math.cos(math.radians(lat0))
|
||
|
||
if abs(cos_lat) < 1e-8:
|
||
cos_lat = 1e-8
|
||
|
||
result = []
|
||
|
||
for index in range(points + 1):
|
||
theta = 2.0 * math.pi * index / points
|
||
|
||
local_x = ellipse["semi_major"] * math.cos(theta)
|
||
local_y = ellipse["semi_minor"] * math.sin(theta)
|
||
|
||
x = (
|
||
local_x * math.cos(angle)
|
||
+ local_y * math.sin(angle)
|
||
)
|
||
y = (
|
||
-local_x * math.sin(angle)
|
||
+ local_y * math.cos(angle)
|
||
)
|
||
|
||
latitude = lat0 + y / 111320.0
|
||
longitude = lon0 + x / (111320.0 * cos_lat)
|
||
|
||
result.append([longitude, latitude])
|
||
|
||
return result
|
||
|
||
|
||
def spatial_zones(observations):
|
||
"""
|
||
Find clearly separated observation groups.
|
||
|
||
This is intentionally conservative. A zone is reported only when there
|
||
are at least three observations in a group and the groups are separated
|
||
by roughly 600 metres.
|
||
"""
|
||
if len(observations) < 6:
|
||
return []
|
||
|
||
remaining = list(observations)
|
||
groups = []
|
||
|
||
while remaining:
|
||
seed = remaining.pop(0)
|
||
group = [seed]
|
||
changed = True
|
||
|
||
while changed:
|
||
changed = False
|
||
keep = []
|
||
|
||
for item in remaining:
|
||
close = False
|
||
|
||
for member in group:
|
||
if (
|
||
haversine_m(
|
||
item["latitude"],
|
||
item["longitude"],
|
||
member["latitude"],
|
||
member["longitude"],
|
||
)
|
||
<= 300.0
|
||
):
|
||
close = True
|
||
break
|
||
|
||
if close:
|
||
group.append(item)
|
||
changed = True
|
||
else:
|
||
keep.append(item)
|
||
|
||
remaining = keep
|
||
|
||
groups.append(group)
|
||
|
||
significant = [
|
||
group
|
||
for group in groups
|
||
if len(group) >= 3
|
||
]
|
||
|
||
if len(significant) <= 1:
|
||
return []
|
||
|
||
centers = []
|
||
|
||
for group in significant:
|
||
center = weighted_center(group)
|
||
|
||
if center:
|
||
centers.append((group, center))
|
||
|
||
separated = False
|
||
|
||
for index, (_, center_a) in enumerate(centers):
|
||
for _, center_b in centers[index + 1:]:
|
||
if (
|
||
haversine_m(
|
||
center_a["latitude"],
|
||
center_a["longitude"],
|
||
center_b["latitude"],
|
||
center_b["longitude"],
|
||
)
|
||
> 600.0
|
||
):
|
||
separated = True
|
||
break
|
||
|
||
if separated:
|
||
break
|
||
|
||
if not separated:
|
||
return []
|
||
|
||
return [
|
||
{
|
||
"observations": group,
|
||
"center": center,
|
||
}
|
||
for group, center in centers
|
||
]
|
||
|
||
|
||
def load_access_point(conn, bssid):
|
||
return query_one(
|
||
conn,
|
||
"""
|
||
SELECT
|
||
id,
|
||
bssid,
|
||
essid,
|
||
encryption,
|
||
cipher,
|
||
akm,
|
||
country,
|
||
channel,
|
||
frequency,
|
||
vendor,
|
||
first_seen,
|
||
last_seen,
|
||
times_seen,
|
||
last_rssi,
|
||
last_latitude,
|
||
last_longitude,
|
||
last_speed,
|
||
has_handshake,
|
||
has_pmkid,
|
||
is_cracked
|
||
FROM access_points
|
||
WHERE lower(bssid) = lower(?)
|
||
LIMIT 1
|
||
""",
|
||
(bssid,),
|
||
)
|
||
|
||
|
||
def load_observations(conn, access_point_id):
|
||
return query_all(
|
||
conn,
|
||
"""
|
||
SELECT
|
||
id,
|
||
session_id,
|
||
observed_at,
|
||
latitude,
|
||
longitude,
|
||
speed,
|
||
rssi,
|
||
channel,
|
||
frequency,
|
||
essid,
|
||
encryption,
|
||
cipher,
|
||
akm,
|
||
country
|
||
FROM access_point_observations
|
||
WHERE access_point_id = ?
|
||
ORDER BY observed_at, id
|
||
""",
|
||
(access_point_id,),
|
||
)
|
||
|
||
|
||
def load_credentials(conn, access_point_id):
|
||
return query_all(
|
||
conn,
|
||
"""
|
||
SELECT
|
||
c.password,
|
||
c.source,
|
||
c.created_at,
|
||
c.verified,
|
||
c.handshake_id
|
||
FROM credentials AS c
|
||
WHERE c.access_point_id = ?
|
||
ORDER BY c.created_at DESC, c.id DESC
|
||
""",
|
||
(access_point_id,),
|
||
)
|
||
|
||
|
||
def security_text(ap):
|
||
values = []
|
||
|
||
for key in ("encryption", "cipher", "akm"):
|
||
value = ap[key]
|
||
|
||
if value and value not in values:
|
||
values.append(str(value))
|
||
|
||
return " / ".join(values) if values else "—"
|
||
|
||
|
||
def calculate_statistics(observations):
|
||
rssis = []
|
||
|
||
speeds = []
|
||
|
||
for item in observations:
|
||
if item["rssi"] is not None:
|
||
try:
|
||
rssis.append(float(item["rssi"]))
|
||
except (TypeError, ValueError):
|
||
pass
|
||
|
||
if item["speed"] is not None:
|
||
try:
|
||
speeds.append(float(item["speed"]))
|
||
except (TypeError, ValueError):
|
||
pass
|
||
|
||
sessions = {
|
||
item["session_id"]
|
||
for item in observations
|
||
if item["session_id"] is not None
|
||
}
|
||
|
||
return {
|
||
"observations": len(observations),
|
||
"sessions": len(sessions),
|
||
"coordinates": len(valid_observations(observations)),
|
||
"rssi_min": min(rssis) if rssis else None,
|
||
"rssi_avg": sum(rssis) / len(rssis) if rssis else None,
|
||
"rssi_max": max(rssis) if rssis else None,
|
||
"speed_avg": sum(speeds) / len(speeds) if speeds else None,
|
||
"speed_max": max(speeds) if speeds else None,
|
||
}
|
||
|
||
|
||
def build_geojson(observations, ellipse=None, zones=None):
|
||
features = []
|
||
|
||
for item in observations:
|
||
try:
|
||
latitude = float(item["latitude"])
|
||
longitude = float(item["longitude"])
|
||
except (TypeError, ValueError):
|
||
continue
|
||
|
||
properties = {
|
||
"type": "observation",
|
||
"id": item["id"],
|
||
"session_id": item["session_id"],
|
||
"observed_at": format_datetime(item["observed_at"]),
|
||
"rssi": item["rssi"],
|
||
"speed": item["speed"],
|
||
"map_status": item.get("map_status", "normal"),
|
||
}
|
||
|
||
features.append(
|
||
{
|
||
"type": "Feature",
|
||
"geometry": {
|
||
"type": "Point",
|
||
"coordinates": [longitude, latitude],
|
||
},
|
||
"properties": properties,
|
||
}
|
||
)
|
||
|
||
if ellipse:
|
||
polygon = ellipse_polygon(ellipse)
|
||
|
||
if polygon:
|
||
features.append(
|
||
{
|
||
"type": "Feature",
|
||
"geometry": {
|
||
"type": "Polygon",
|
||
"coordinates": [polygon],
|
||
},
|
||
"properties": {
|
||
"type": "uncertainty",
|
||
},
|
||
}
|
||
)
|
||
|
||
if zones:
|
||
for index, zone in enumerate(zones, start=1):
|
||
center = zone["center"]
|
||
|
||
features.append(
|
||
{
|
||
"type": "Feature",
|
||
"geometry": {
|
||
"type": "Point",
|
||
"coordinates": [
|
||
center["longitude"],
|
||
center["latitude"],
|
||
],
|
||
},
|
||
"properties": {
|
||
"type": "zone",
|
||
"zone": index,
|
||
"observations": len(zone["observations"]),
|
||
},
|
||
}
|
||
)
|
||
|
||
return {
|
||
"type": "FeatureCollection",
|
||
"features": features,
|
||
}
|
||
|
||
|
||
def render_map(geojson, context, title):
|
||
maplibre_css = context.asset_url("maplibre/maplibre-gl.css")
|
||
maplibre_js = context.asset_url("maplibre/maplibre-gl.js")
|
||
|
||
geojson_json = json.dumps(
|
||
geojson,
|
||
ensure_ascii=False,
|
||
separators=(",", ":"),
|
||
)
|
||
|
||
return f"""
|
||
<link rel="stylesheet" href="{esc(maplibre_css)}">
|
||
|
||
<style>
|
||
.ap-history-map {{
|
||
width: 100%;
|
||
height: 620px;
|
||
border-radius: 10px;
|
||
overflow: hidden;
|
||
}}
|
||
|
||
.ap-map-legend {{
|
||
margin-top: 12px;
|
||
color: var(--text);
|
||
font-size: 13px;
|
||
line-height: 1.55;
|
||
}}
|
||
|
||
.ap-map-legend-row {{
|
||
display: flex;
|
||
align-items: center;
|
||
gap: 8px;
|
||
margin: 4px 0;
|
||
}}
|
||
|
||
.ap-map-legend strong {{
|
||
color: var(--text);
|
||
}}
|
||
|
||
.ap-map-legend-marker {{
|
||
display: inline-block;
|
||
flex: 0 0 auto;
|
||
width: 10px;
|
||
height: 10px;
|
||
border-radius: 50%;
|
||
border: 1px solid rgba(0, 0, 0, 0.25);
|
||
}}
|
||
|
||
.ap-map-legend-marker.normal {{
|
||
background: #2ca25f;
|
||
}}
|
||
|
||
.ap-map-legend-marker.handshake {{
|
||
background: #f0a000;
|
||
}}
|
||
|
||
.ap-map-legend-marker.cracked {{
|
||
background: #d33;
|
||
}}
|
||
|
||
.ap-map-legend-marker.uncertainty {{
|
||
width: 18px;
|
||
height: 10px;
|
||
border: 2px solid #3388ff;
|
||
border-radius: 50%;
|
||
background: rgba(51, 136, 255, 0.16);
|
||
}}
|
||
</style>
|
||
|
||
<div id="access-point-history-map" class="ap-history-map"></div>
|
||
|
||
<div class="ap-map-legend">
|
||
<div class="ap-map-legend-row">
|
||
<span class="ap-map-legend-marker normal"></span>
|
||
<span><strong>Обычное наблюдение</strong> — координата, записанная во время сканирования.</span>
|
||
</div>
|
||
<div class="ap-map-legend-row">
|
||
<span class="ap-map-legend-marker handshake"></span>
|
||
<span><strong>Handshake</strong> — для этого AP в базе есть перехваченный handshake.</span>
|
||
</div>
|
||
<div class="ap-map-legend-row">
|
||
<span class="ap-map-legend-marker cracked"></span>
|
||
<span><strong>Пароль найден</strong> — для этого AP в базе есть credential.</span>
|
||
</div>
|
||
<div class="ap-map-legend-row">
|
||
<span class="ap-map-legend-marker uncertainty"></span>
|
||
<span><strong>Расчётная область</strong> — статистическая оценка положения AP по истории наблюдений. Это не гарантированная зона покрытия и не точная координата устройства.</span>
|
||
</div>
|
||
</div>
|
||
|
||
<script src="{esc(maplibre_js)}"></script>
|
||
<script>
|
||
(function() {{
|
||
const observations = {geojson_json};
|
||
|
||
const map = new maplibregl.Map({{
|
||
container: "access-point-history-map",
|
||
style: "http://127.0.0.1:8080/styles/server-map/style.json",
|
||
center: [30.32, 59.93],
|
||
zoom: 5
|
||
}});
|
||
|
||
map.addControl(new maplibregl.NavigationControl(), "top-right");
|
||
map.addControl(new maplibregl.ScaleControl({{
|
||
maxWidth: 120,
|
||
unit: "metric"
|
||
}}));
|
||
|
||
map.on("load", function() {{
|
||
map.addSource("ap-history", {{
|
||
type: "geojson",
|
||
data: observations
|
||
}});
|
||
|
||
map.addLayer({{
|
||
id: "ap-uncertainty-fill",
|
||
type: "fill",
|
||
source: "ap-history",
|
||
filter: ["==", ["get", "type"], "uncertainty"],
|
||
paint: {{
|
||
"fill-color": "#3388ff",
|
||
"fill-opacity": 0.16
|
||
}}
|
||
}});
|
||
|
||
map.addLayer({{
|
||
id: "ap-uncertainty-line",
|
||
type: "line",
|
||
source: "ap-history",
|
||
filter: ["==", ["get", "type"], "uncertainty"],
|
||
paint: {{
|
||
"line-color": "#3388ff",
|
||
"line-width": 2,
|
||
"line-opacity": 0.55
|
||
}}
|
||
}});
|
||
|
||
map.addLayer({{
|
||
id: "ap-observations",
|
||
type: "circle",
|
||
source: "ap-history",
|
||
filter: ["==", ["get", "type"], "observation"],
|
||
paint: {{
|
||
"circle-radius": 5,
|
||
"circle-color": [
|
||
"match",
|
||
["get", "map_status"],
|
||
"cracked", "#d33",
|
||
"handshake", "#f0a000",
|
||
"#2ca25f"
|
||
],
|
||
"circle-opacity": 0.72,
|
||
"circle-stroke-width": 1,
|
||
"circle-stroke-color": "#fff"
|
||
}}
|
||
}});
|
||
|
||
map.addLayer({{
|
||
id: "ap-zones",
|
||
type: "circle",
|
||
source: "ap-history",
|
||
filter: ["==", ["get", "type"], "zone"],
|
||
paint: {{
|
||
"circle-radius": 8,
|
||
"circle-color": "#f0a000",
|
||
"circle-opacity": 0.9,
|
||
"circle-stroke-width": 2,
|
||
"circle-stroke-color": "#fff"
|
||
}}
|
||
}});
|
||
|
||
map.on("click", "ap-observations", function(e) {{
|
||
const p = e.features[0].properties;
|
||
|
||
new maplibregl.Popup()
|
||
.setLngLat(e.lngLat)
|
||
.setHTML(
|
||
"<strong>Наблюдение</strong><br>" +
|
||
"Дата: " + escapeHtml(p.observed_at) + "<br>" +
|
||
"Сессия: <a href=\\"/reports/session/" +
|
||
encodeURIComponent(p.session_id) + "\\">" +
|
||
escapeHtml(p.session_id) + "</a><br>" +
|
||
"RSSI: " + escapeHtml(p.rssi ?? "—") + "<br>" +
|
||
"Скорость: " + escapeHtml(p.speed ?? "—")
|
||
)
|
||
.addTo(map);
|
||
}});
|
||
|
||
map.on("click", "ap-zones", function(e) {{
|
||
const p = e.features[0].properties;
|
||
|
||
new maplibregl.Popup()
|
||
.setLngLat(e.lngLat)
|
||
.setHTML(
|
||
"<strong>Отдельная зона наблюдений</strong><br>" +
|
||
"Наблюдений: " +
|
||
escapeHtml(p.observations)
|
||
)
|
||
.addTo(map);
|
||
}});
|
||
|
||
map.on("mouseenter", "ap-observations", function() {{
|
||
map.getCanvas().style.cursor = "pointer";
|
||
}});
|
||
|
||
map.on("mouseleave", "ap-observations", function() {{
|
||
map.getCanvas().style.cursor = "";
|
||
}});
|
||
|
||
map.on("mouseenter", "ap-zones", function() {{
|
||
map.getCanvas().style.cursor = "pointer";
|
||
}});
|
||
|
||
map.on("mouseleave", "ap-zones", function() {{
|
||
map.getCanvas().style.cursor = "";
|
||
}});
|
||
|
||
map.on("click", "ap-zones", function(e) {{
|
||
const p = e.features[0].properties;
|
||
|
||
new maplibregl.Popup()
|
||
.setLngLat(e.lngLat)
|
||
.setHTML(
|
||
"<strong>Отдельная зона наблюдений</strong><br>" +
|
||
"Наблюдений: " +
|
||
escapeHtml(p.observations)
|
||
)
|
||
.addTo(map);
|
||
}});
|
||
|
||
map.on("mouseenter", "ap-zones", function() {{
|
||
map.getCanvas().style.cursor = "pointer";
|
||
}});
|
||
|
||
map.on("mouseleave", "ap-zones", function() {{
|
||
map.getCanvas().style.cursor = "";
|
||
}});
|
||
|
||
const coordinates = observations.features
|
||
.filter(function(feature) {{
|
||
return feature.geometry.type === "Point";
|
||
}})
|
||
.map(function(feature) {{
|
||
return feature.geometry.coordinates;
|
||
}});
|
||
|
||
if (coordinates.length === 1) {{
|
||
map.setCenter(coordinates[0]);
|
||
map.setZoom(16);
|
||
}} else if (coordinates.length > 1) {{
|
||
const bounds = coordinates.reduce(
|
||
function(bounds, coordinate) {{
|
||
return bounds.extend(coordinate);
|
||
}},
|
||
new maplibregl.LngLatBounds(
|
||
coordinates[0],
|
||
coordinates[0]
|
||
)
|
||
);
|
||
|
||
map.fitBounds(bounds, {{
|
||
padding: 70,
|
||
maxZoom: 16
|
||
}});
|
||
}}
|
||
}});
|
||
|
||
function escapeHtml(value) {{
|
||
return String(value ?? "")
|
||
.replace(/&/g, "&")
|
||
.replace(/</g, "<")
|
||
.replace(/>/g, ">")
|
||
.replace(/"/g, """)
|
||
.replace(/'/g, "'");
|
||
}}
|
||
}})();
|
||
</script>
|
||
"""
|
||
|
||
|
||
def render_observation_table(observations):
|
||
rows = []
|
||
|
||
for item in reversed(observations):
|
||
coordinates = "—"
|
||
|
||
if (
|
||
item["latitude"] is not None
|
||
and item["longitude"] is not None
|
||
):
|
||
coordinates = (
|
||
f"{float(item['latitude']):.6f}, "
|
||
f"{float(item['longitude']):.6f}"
|
||
)
|
||
|
||
rows.append(
|
||
f"""
|
||
<tr>
|
||
<td>{esc(format_datetime(item["observed_at"]))}</td>
|
||
<td>
|
||
<a href="/reports/session/{esc(item['session_id'])}">
|
||
{esc(item["session_id"])}
|
||
</a>
|
||
</td>
|
||
<td>{esc(coordinates)}</td>
|
||
<td>{esc(item["rssi"] if item["rssi"] is not None else "—")}</td>
|
||
<td>{esc(format_speed(item["speed"]))}</td>
|
||
<td>{esc(item["channel"] if item["channel"] is not None else "—")}</td>
|
||
<td>{esc(item["frequency"] if item["frequency"] is not None else "—")}</td>
|
||
</tr>
|
||
"""
|
||
)
|
||
|
||
return f"""
|
||
<div style="overflow-x:auto;">
|
||
<table class="report-table">
|
||
<thead>
|
||
<tr>
|
||
<th>Дата</th>
|
||
<th>Сессия</th>
|
||
<th>Координаты</th>
|
||
<th>RSSI</th>
|
||
<th>Скорость</th>
|
||
<th>Канал</th>
|
||
<th>Частота</th>
|
||
</tr>
|
||
</thead>
|
||
<tbody>
|
||
{''.join(rows)}
|
||
</tbody>
|
||
</table>
|
||
</div>
|
||
"""
|
||
|
||
|
||
def render_access_point_page(
|
||
conn,
|
||
bssid,
|
||
context=None,
|
||
):
|
||
if context is None:
|
||
context = ReportContext(mode="server")
|
||
|
||
ap = load_access_point(conn, bssid)
|
||
|
||
if not ap:
|
||
page = page_begin(
|
||
"Access Point",
|
||
active="access_points",
|
||
context=context,
|
||
)
|
||
|
||
page += page_header(
|
||
"Точка доступа не найдена",
|
||
"BSSID: " + esc(bssid),
|
||
)
|
||
|
||
page += card(
|
||
"""
|
||
<div style="padding: 10px 0;">
|
||
Запрошенная точка доступа отсутствует в базе данных.
|
||
</div>
|
||
"""
|
||
)
|
||
|
||
page += page_end()
|
||
return page
|
||
|
||
observations_raw = load_observations(conn, ap["id"])
|
||
observations = valid_observations(observations_raw)
|
||
credentials = load_credentials(conn, ap["id"])
|
||
|
||
if ap["is_cracked"]:
|
||
map_status = "cracked"
|
||
elif ap["has_handshake"]:
|
||
map_status = "handshake"
|
||
else:
|
||
map_status = "normal"
|
||
|
||
map_observations = []
|
||
|
||
for item in observations:
|
||
map_item = dict(item)
|
||
map_item["map_status"] = map_status
|
||
map_observations.append(map_item)
|
||
|
||
statistics = calculate_statistics(observations_raw)
|
||
|
||
burst_observations = normalize_bursts(observations)
|
||
useful_observations = robust_observations(burst_observations)
|
||
|
||
estimated_center = weighted_center(burst_observations)
|
||
ellipse = covariance_ellipse(burst_observations)
|
||
zones = spatial_zones(observations)
|
||
|
||
geojson = build_geojson(
|
||
map_observations,
|
||
ellipse=ellipse,
|
||
zones=zones,
|
||
)
|
||
|
||
page = page_begin(
|
||
f"AP {ap['bssid']}",
|
||
active="access_points",
|
||
context=context,
|
||
)
|
||
|
||
page += page_header(
|
||
f"📡 {esc(ap['essid'] or 'Access Point')}",
|
||
f"BSSID: {esc(ap['bssid'])}",
|
||
)
|
||
|
||
page += f"""
|
||
<div style="margin-bottom:18px;">
|
||
<a href="/reports/access-points/"
|
||
style="text-decoration:none;">
|
||
← Все Access Points
|
||
</a>
|
||
</div>
|
||
"""
|
||
|
||
page += card(
|
||
"Основная информация",
|
||
f"""
|
||
<div class="report-grid">
|
||
<div>
|
||
<strong>BSSID</strong><br>
|
||
{esc(ap["bssid"])}
|
||
</div>
|
||
|
||
<div>
|
||
<strong>ESSID</strong><br>
|
||
{esc(ap["essid"] or "—")}
|
||
</div>
|
||
|
||
<div>
|
||
<strong>Vendor</strong><br>
|
||
{esc(ap["vendor"] or "—")}
|
||
</div>
|
||
|
||
<div>
|
||
<strong>Security</strong><br>
|
||
{esc(security_text(ap))}
|
||
</div>
|
||
|
||
<div>
|
||
<strong>Channel</strong><br>
|
||
{esc(ap["channel"] if ap["channel"] is not None else "—")}
|
||
</div>
|
||
|
||
<div>
|
||
<strong>Frequency</strong><br>
|
||
{esc(ap["frequency"] if ap["frequency"] is not None else "—")}
|
||
</div>
|
||
|
||
<div>
|
||
<strong>First seen</strong><br>
|
||
{esc(format_datetime(ap["first_seen"]))}
|
||
</div>
|
||
|
||
<div>
|
||
<strong>Last seen</strong><br>
|
||
{esc(format_datetime(ap["last_seen"]))}
|
||
</div>
|
||
</div>
|
||
"""
|
||
)
|
||
|
||
page += card(
|
||
"История наблюдений",
|
||
f"""
|
||
<div class="report-grid">
|
||
<div>
|
||
<strong>Наблюдений</strong><br>
|
||
{statistics["observations"]}
|
||
</div>
|
||
|
||
<div>
|
||
<strong>Сессий</strong><br>
|
||
{statistics["sessions"]}
|
||
</div>
|
||
|
||
<div>
|
||
<strong>С координатами</strong><br>
|
||
{statistics["coordinates"]}
|
||
</div>
|
||
|
||
<div>
|
||
<strong>RSSI min</strong><br>
|
||
{format_number(statistics["rssi_min"])}
|
||
</div>
|
||
|
||
<div>
|
||
<strong>RSSI avg</strong><br>
|
||
{format_number(statistics["rssi_avg"])}
|
||
</div>
|
||
|
||
<div>
|
||
<strong>RSSI max</strong><br>
|
||
{format_number(statistics["rssi_max"])}
|
||
</div>
|
||
|
||
<div>
|
||
<strong>Средняя скорость</strong><br>
|
||
{format_speed(statistics["speed_avg"])}
|
||
</div>
|
||
|
||
<div>
|
||
<strong>Максимальная скорость</strong><br>
|
||
{format_speed(statistics["speed_max"])}
|
||
</div>
|
||
</div>
|
||
"""
|
||
)
|
||
|
||
status_parts = []
|
||
|
||
if ap["has_handshake"]:
|
||
status_parts.append("Handshake")
|
||
|
||
if ap["has_pmkid"]:
|
||
status_parts.append("PMKID")
|
||
|
||
if ap["is_cracked"]:
|
||
status_parts.append("Cracked")
|
||
|
||
if status_parts:
|
||
status_text = ", ".join(status_parts)
|
||
else:
|
||
status_text = "Нет связанных данных"
|
||
|
||
credential_rows = []
|
||
|
||
for credential in credentials:
|
||
verified = "Да" if credential["verified"] else "Нет"
|
||
|
||
credential_rows.append(
|
||
f"""
|
||
<tr>
|
||
<td>{esc(credential["password"])}</td>
|
||
<td>{esc(credential["source"] or "—")}</td>
|
||
<td>{esc(verified)}</td>
|
||
<td>{esc(format_datetime(credential["created_at"]))}</td>
|
||
<td>#{esc(credential["handshake_id"])}</td>
|
||
</tr>
|
||
"""
|
||
)
|
||
|
||
credentials_html = ""
|
||
|
||
if credential_rows:
|
||
credentials_html = f"""
|
||
<div style="overflow-x:auto;">
|
||
<table class="report-table">
|
||
<thead>
|
||
<tr>
|
||
<th>Password</th>
|
||
<th>Source</th>
|
||
<th>Verified</th>
|
||
<th>Created</th>
|
||
<th>Handshake</th>
|
||
</tr>
|
||
</thead>
|
||
<tbody>
|
||
{''.join(credential_rows)}
|
||
</tbody>
|
||
</table>
|
||
</div>
|
||
"""
|
||
else:
|
||
credentials_html = """
|
||
<div style="padding:8px 0;">
|
||
Связанных credentials нет.
|
||
</div>
|
||
"""
|
||
|
||
page += card(
|
||
"Связанные данные",
|
||
f"""
|
||
<div style="margin-bottom:14px;">
|
||
<strong>Статус:</strong> {esc(status_text)}
|
||
</div>
|
||
|
||
{credentials_html}
|
||
"""
|
||
)
|
||
|
||
if estimated_center:
|
||
estimated_coordinates = (
|
||
f"{estimated_center['latitude']:.6f}, "
|
||
f"{estimated_center['longitude']:.6f}"
|
||
)
|
||
else:
|
||
estimated_coordinates = "Недостаточно координат"
|
||
|
||
if ellipse:
|
||
ellipse_size = (
|
||
f"{ellipse['semi_major']:.0f} × "
|
||
f"{ellipse['semi_minor']:.0f} м"
|
||
)
|
||
|
||
ellipse_direction = f"{ellipse['angle']:.0f}°"
|
||
else:
|
||
ellipse_size = "Недостаточно данных"
|
||
ellipse_direction = "—"
|
||
|
||
if zones:
|
||
zone_message = (
|
||
f"История содержит {len(zones)} пространственно "
|
||
f"разделённых зоны наблюдений. Поэтому одна общая "
|
||
f"область неопределённости может быть misleading."
|
||
)
|
||
else:
|
||
zone_message = (
|
||
"Наблюдения не показывают достаточно выраженного "
|
||
"разделения на несколько независимых пространственных зон."
|
||
)
|
||
|
||
|
||
page += card(
|
||
"Расчёт положения",
|
||
f"""
|
||
<div class="report-grid">
|
||
<div>
|
||
<strong>Оценочный центр</strong><br>
|
||
{esc(estimated_coordinates)}
|
||
</div>
|
||
|
||
<div>
|
||
<strong>Размер эллипса</strong><br>
|
||
{esc(ellipse_size)}
|
||
</div>
|
||
|
||
<div>
|
||
<strong>Ориентация</strong><br>
|
||
{esc(ellipse_direction)}
|
||
</div>
|
||
|
||
<div>
|
||
<strong>Использовано координат</strong><br>
|
||
{len(useful_observations)}
|
||
</div>
|
||
</div>
|
||
|
||
<div style="
|
||
margin-top:15px;
|
||
padding:12px;
|
||
border-radius:8px;
|
||
background:rgba(128,128,128,0.08);
|
||
line-height:1.5;
|
||
">
|
||
<strong>Важно:</strong>
|
||
расчётная точка является статистической оценкой по истории
|
||
наблюдений. Она не означает, что AP физически находится
|
||
непосредственно в этой координате.
|
||
RSSI используется только как дополнительный относительный вес.
|
||
Расчётная область показывает неопределённость оценки, а не
|
||
гарантированную дальность действия Wi-Fi.
|
||
</div>
|
||
|
||
<div style="
|
||
margin-top:12px;
|
||
color:var(--muted-text, #777);
|
||
line-height:1.5;
|
||
">
|
||
{esc(zone_message)}
|
||
</div>
|
||
"""
|
||
)
|
||
|
||
password_text = ""
|
||
|
||
if credentials:
|
||
password = credentials[0]["password"]
|
||
|
||
if password:
|
||
password_text = (
|
||
f" · Password: {esc(password)}"
|
||
)
|
||
|
||
page += card(
|
||
"Карта истории",
|
||
f"""
|
||
<div style="
|
||
margin-bottom:12px;
|
||
font-size:15px;
|
||
font-weight:600;
|
||
">
|
||
📡 {esc(ap["essid"] or "Access Point")}
|
||
· AP {esc(ap["bssid"])}
|
||
{password_text}
|
||
</div>
|
||
|
||
{render_map(
|
||
geojson,
|
||
context,
|
||
ap["bssid"],
|
||
)}
|
||
"""
|
||
)
|
||
|
||
page += card(
|
||
"Полная история наблюдений",
|
||
f"""
|
||
<div style="
|
||
margin-bottom:12px;
|
||
color:var(--muted-text, #777);
|
||
line-height:1.5;
|
||
">
|
||
Здесь отображаются все записи из
|
||
<code>access_point_observations</code>.
|
||
Записи без координат сохраняются в истории, но не участвуют
|
||
в пространственном расчёте.
|
||
</div>
|
||
|
||
{render_observation_table(observations_raw)}
|
||
"""
|
||
)
|
||
|
||
page += page_end()
|
||
|
||
return page
|