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"""
Обычное наблюдение — координата, записанная во время сканирования.
Handshake — для этого AP в базе есть перехваченный handshake.
Пароль найден — для этого AP в базе есть credential.
Расчётная область — статистическая оценка положения AP по истории наблюдений. Это не гарантированная зона покрытия и не точная координата устройства.
""" 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""" {esc(format_datetime(item["observed_at"]))} {esc(item["session_id"])} {esc(coordinates)} {esc(item["rssi"] if item["rssi"] is not None else "—")} {esc(format_speed(item["speed"]))} {esc(item["channel"] if item["channel"] is not None else "—")} {esc(item["frequency"] if item["frequency"] is not None else "—")} """ ) return f"""
{''.join(rows)}
Дата Сессия Координаты RSSI Скорость Канал Частота
""" 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( """
Запрошенная точка доступа отсутствует в базе данных.
""" ) 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"""
← Все Access Points
""" page += card( "Основная информация", f"""
BSSID
{esc(ap["bssid"])}
ESSID
{esc(ap["essid"] or "—")}
Vendor
{esc(ap["vendor"] or "—")}
Security
{esc(security_text(ap))}
Channel
{esc(ap["channel"] if ap["channel"] is not None else "—")}
Frequency
{esc(ap["frequency"] if ap["frequency"] is not None else "—")}
First seen
{esc(format_datetime(ap["first_seen"]))}
Last seen
{esc(format_datetime(ap["last_seen"]))}
""" ) page += card( "История наблюдений", f"""
Наблюдений
{statistics["observations"]}
Сессий
{statistics["sessions"]}
С координатами
{statistics["coordinates"]}
RSSI min
{format_number(statistics["rssi_min"])}
RSSI avg
{format_number(statistics["rssi_avg"])}
RSSI max
{format_number(statistics["rssi_max"])}
Средняя скорость
{format_speed(statistics["speed_avg"])}
Максимальная скорость
{format_speed(statistics["speed_max"])}
""" ) 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""" {esc(credential["password"])} {esc(credential["source"] or "—")} {esc(verified)} {esc(format_datetime(credential["created_at"]))} #{esc(credential["handshake_id"])} """ ) credentials_html = "" if credential_rows: credentials_html = f"""
{''.join(credential_rows)}
Password Source Verified Created Handshake
""" else: credentials_html = """
Связанных credentials нет.
""" page += card( "Связанные данные", f"""
Статус: {esc(status_text)}
{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"""
Оценочный центр
{esc(estimated_coordinates)}
Размер эллипса
{esc(ellipse_size)}
Ориентация
{esc(ellipse_direction)}
Использовано координат
{len(useful_observations)}
Важно: расчётная точка является статистической оценкой по истории наблюдений. Она не означает, что AP физически находится непосредственно в этой координате. RSSI используется только как дополнительный относительный вес. Расчётная область показывает неопределённость оценки, а не гарантированную дальность действия Wi-Fi.
{esc(zone_message)}
""" ) password_text = "" if credentials: password = credentials[0]["password"] if password: password_text = ( f" · Password: {esc(password)}" ) page += card( "Карта истории", f"""
📡 {esc(ap["essid"] or "Access Point")} · AP {esc(ap["bssid"])} {password_text}
{render_map( geojson, context, ap["bssid"], )} """ ) page += card( "Полная история наблюдений", f"""
Здесь отображаются все записи из access_point_observations. Записи без координат сохраняются в истории, но не участвуют в пространственном расчёте.
{render_observation_table(observations_raw)} """ ) page += page_end() return page