dxcluster.py
Rama main · réplica de lectura del repositorio original.
"""
dxcluster.py — DX Cluster client (Telnet).
Opens a TCP connection to a DX Cluster server, sends the login and
password (if required), then parses incoming lines in the DX de format:
DX de <spotter>: <freq> <call> <comment> <utc>
Each logged-in user can have their own connection with their own settings
(address, port, login, password) — managed per-user in ClusterManager.
Spots are broadcast via a callback (WS to the specific user) in this format:
{"type": "dx_spot", "freq_hz": int, "call": str, "spotter": str,
"comment": str, "utc": str, "ts": epoch_seconds, "band": str, "mode": str}
NOTE: the "message" field of a dx_status broadcast IS displayed directly
in the UI for the error/disconnected states (public/js/dxcluster.js's
updateStatusBadge, not routed through I18n) - keep any text assigned to
it in Polish; only comments/docstrings/print() in this file are in scope
for the backend English translation pass.
"""
import asyncio
import ipaddress
import re
import socket
import time
from typing import Callable, Optional
try:
from config import DATA as _DATA_DIR
except Exception:
from pathlib import Path as _Path
_DATA_DIR = _Path(__file__).parent
# Dziennik polaczenia DX Clustera - zgloszone na zywo: "hamctrl nie ma
# zaimplementowanych logow", wiec przy "ciagle sie rozlancza i laczy" nie
# bylo jak sprawdzic co faktycznie sie dzieje (produkcyjny EXE dziala bez
# okna konsoli - print() ginie bez sladu). Osobny, maly plik tekstowy w
# tym samym katalogu co config.DATA (czyli %APPDATA%\HAMCTRL na
# Windowsie), zeby bylo co wkleic/przeczytac po fakcie zamiast zgadywac.
_LOG_FILE = _DATA_DIR / "dxcluster.log"
_LOG_MAX_BYTES = 1_000_000 # ~1MB - obcinamy zamiast rosnac w nieskonczonosc
def _log_line(msg: str):
try:
ts = time.strftime('%Y-%m-%d %H:%M:%S')
line = f"[{ts}] {msg}\n"
if _LOG_FILE.exists() and _LOG_FILE.stat().st_size > _LOG_MAX_BYTES:
# Prosta rotacja: zostaw tylko ostatnia polowe pliku
try:
data = _LOG_FILE.read_text(encoding='utf-8', errors='ignore')
_LOG_FILE.write_text(data[len(data) // 2:], encoding='utf-8')
except Exception:
pass
with open(_LOG_FILE, 'a', encoding='utf-8') as f:
f.write(line)
except Exception:
pass # logowanie nigdy nie moze wywrocic polaczenia
async def _resolve_and_check_safe(host: str, port: int) -> tuple[bool, str]:
"""
SSRF guard (added 2026-09-02, found via a live permission audit): ANY
logged-in user - operator or viewer, no admin gate on this endpoint -
can set their OWN DX cluster host/port via /api/dxcluster/config, and
/api/dxcluster/connect then makes THE SERVER open a raw TCP connection
to it and send a "login" line. Without this check that's a working
SSRF primitive: point it at 127.0.0.1:<any local port>, an internal
LAN address, or another machine's admin interface, and the server
will connect and speak to it on the user's behalf. Resolves the
HOSTNAME to its actual IP (not just string-matching "127.0.0.1") so a
DNS name that merely points at an internal address is caught too, and
checks every address a name resolves to, not just the first.
"""
if not host:
return False, "Brak adresu hosta"
try:
loop = asyncio.get_event_loop()
infos = await asyncio.wait_for(
loop.getaddrinfo(host, port, type=socket.SOCK_STREAM), timeout=5.0)
except Exception as e:
return False, f"Nie mozna rozwiazac adresu {host!r}: {e}"
for info in infos:
ip_str = info[4][0]
try:
ip = ipaddress.ip_address(ip_str)
except ValueError:
continue
if (ip.is_private or ip.is_loopback or ip.is_link_local
or ip.is_reserved or ip.is_multicast or ip.is_unspecified):
return False, (f"Adres {host!r} ({ip_str}) wskazuje na siec "
f"prywatna/lokalna - zablokowane")
return True, ""
# Band prefixes from the IARU table (to place a spot on the waterfall).
# 160m/60m/6m DELIBERATELY NARROWER (real PL/EU allocation) - the same
# values as webapp.py::_BAND_RANGES (the table also used by the TX block).
# This used to be a separate, independently-maintained copy that drifted
# out of sync - the SAME bug already fixed once in webapp.py, just in a
# different file.
_BAND_RANGES = [
('160m', 1810000, 2000000),
('80m', 3500000, 3800000),
('60m', 5351500, 5366500),
('40m', 7000000, 7200000),
('30m', 10100000, 10150000),
('20m', 14000000, 14350000),
('17m', 18068000, 18168000),
('15m', 21000000, 21450000),
('12m', 24890000, 24990000),
('10m', 28000000, 29700000),
('6m', 50000000, 52000000),
('4m', 70000000, 70500000),
('2m', 144000000, 146000000),
('70cm', 430000000, 440000000),
]
def _get_band(freq_hz: int) -> str:
for name, lo, hi in _BAND_RANGES:
if lo <= freq_hz <= hi:
return name
return '?'
def _guess_mode(freq_hz: int, comment: str) -> str:
"""
Spot mode heuristic: comment -> exact digital mode frequencies ->
IARU R1 bandplan.
Order matters:
1. The comment (as SEPARATE WORDS \\b...\\b, so 'OK1CW' doesn't give CW)
2. EXACT FT8/FT4/MSK144 frequencies with +-2 kHz tolerance.
This must come BEFORE the bandplan, since e.g. 2m FT8 (144.174)
falls within the 144.150-144.400 phone segment and the bandplan
would incorrectly return SSB.
3. Bandplan (CW/DIGI/SSB/FM segments) - HF, VHF, UHF, microwave.
"""
c = (comment or '').upper()
# ── 1. Mode stated directly in the comment ──────────────────────────────
for pat, mode in (
(r'\bFT8\b', 'FT8'),
(r'\bFT4\b', 'FT4'),
(r'\bMSK144\b', 'MSK144'),
(r'\bJS8\b', 'DIGI'),
(r'\bQ65\b', 'DIGI'),
(r'\bJT(?:65|9|6M)\b', 'DIGI'),
(r'\bRTTY\b', 'RTTY'),
(r'\bPSK(?:31|63)?\b', 'PSK'),
(r'\bWSPR\b', 'DIGI'),
(r'\bSSB\b', 'SSB'),
(r'\b(?:LSB|USB)\b', 'SSB'),
(r'\bCW\b', 'CW'),
(r'\bFM\b', 'FM'),
(r'\bAM\b', 'AM'),
):
if re.search(pat, c):
return mode
khz = freq_hz / 1000.0
# ── 2. Exact digital mode frequencies (+-2 kHz tolerance) ────────────────
# An FT8 signal has ~50 Hz carriers within a 3 kHz band, spots vary by a
# few hundred Hz, hence the tolerance. MUST come before the bandplan
# (see the docstring).
FT8_KHZ = [
1840, 3573, 5357, 7074, 10136, 14074, 18100, 21074, 24915, 28074,
50313, 50323, # 6m (dwie czestotliwosci)
70100, 70154, # 4m
144174, # 2m
222065, # 1.25m
432174, # 70cm
1296174, # 23cm
2320174, # 13cm
3400174, 5760174, 10368174, 24048174, # 9cm/6cm/3cm/1.2cm
]
FT4_KHZ = [
3575, 7047.5, 10140, 14080, 18104, 21140, 24919, 28180,
50318, # 6m
144170, # 2m
]
MSK144_KHZ = [50260, 70230, 144360, 432360] # meteor scatter
def _near(target_list, tol=2.0):
return any(abs(khz - t) <= tol for t in target_list)
if _near(FT8_KHZ): return 'FT8'
if _near(FT4_KHZ): return 'FT4'
if _near(MSK144_KHZ): return 'MSK144'
# WSPR (narrow, 1 kHz tolerance)
WSPR_KHZ = [1836.6, 3568.6, 7038.6, 10138.7, 14095.6, 18104.6,
21094.6, 24924.6, 28124.6, 50293, 144489]
if _near(WSPR_KHZ, tol=1.0): return 'DIGI'
# ── 3. Satellites — BEFORE the bandplan, since FM segments would mask them ──
# 2m: 145.800-146.000, 70cm: 435.000-438.000 (IARU satellite segments)
if 145800 <= khz <= 146000 or 435000 <= khz <= 438000:
return 'SAT'
# ── 4. IARU Region 1 bandplan (kHz) ──────────────────────────────────────
BANDPLAN = [
# ── HF ──
(1810, 1838, 'CW'), (1838, 1843, 'DIGI'), (1843, 2000, 'SSB'),
(3500, 3570, 'CW'), (3570, 3600, 'DIGI'), (3600, 3800, 'SSB'),
(5250, 5450, 'SSB'), # 60m (channelized, USB)
(7000, 7040, 'CW'), (7040, 7060, 'DIGI'), (7060, 7200, 'SSB'),
(10100, 10130, 'CW'), (10130, 10150, 'DIGI'), # 30m - no phone
(14000, 14070, 'CW'), (14070, 14099, 'DIGI'), (14101, 14350, 'SSB'),
(18068, 18095, 'CW'), (18095, 18109, 'DIGI'), (18111, 18168, 'SSB'),
(21000, 21070, 'CW'), (21070, 21150, 'DIGI'), (21151, 21450, 'SSB'),
(24890, 24915, 'CW'), (24915, 24929, 'DIGI'), (24931, 24990, 'SSB'),
(28000, 28070, 'CW'), (28070, 28190, 'DIGI'), (28225, 29000, 'SSB'),
(29000, 29700, 'FM'),
# ── 6m ──
(50000, 50100, 'CW'), (50100, 50300, 'SSB'),
(50300, 50400, 'DIGI'), (50400, 52000, 'SSB'),
# ── 4m (70 MHz) — SSB from 70.200, narrower DIGI ──
(69900, 70100, 'CW'), (70100, 70200, 'DIGI'),
(70200, 70300, 'SSB'), (70300, 70500, 'FM'),
# ── 2m ──
(144000, 144150, 'CW'), (144150, 144400, 'SSB'),
(144400, 144500, 'DIGI'), (144500, 145800, 'FM'),
(146000, 148000, 'FM'), # region 2
# ── 1.25m (222 MHz, region 2) ──
(222000, 222150, 'CW'), (222150, 222250, 'SSB'), (222250, 225000, 'FM'),
# ── 70cm ──
(430000, 432000, 'FM'), (432000, 432100, 'CW'),
(432100, 432400, 'SSB'), (432400, 432500, 'DIGI'),
(432500, 435000, 'FM'), (438000, 440000, 'FM'),
# ── 23cm ──
(1296000, 1296150, 'CW'), (1296150, 1296400, 'SSB'),
(1296400, 1296600, 'DIGI'), (1296600, 1300000, 'FM'),
# ── 13cm ──
(2320000, 2320150, 'CW'), (2320150, 2320400, 'SSB'), (2320400, 2450000, 'FM'),
# ── 9cm ──
(3400000, 3400150, 'CW'), (3400150, 3400400, 'SSB'),
# ── 6cm ──
(5760000, 5760150, 'CW'), (5760150, 5760400, 'SSB'),
# ── 3cm ──
(10368000, 10368150, 'CW'), (10368150, 10368400, 'SSB'),
# ── 1.2cm ──
(24048000, 24048150, 'CW'), (24048150, 24048400, 'SSB'),
]
for lo, hi, mode in BANDPLAN:
if lo <= khz < hi:
return mode
return '?'
# Regex for the "DX de <spotter>: <freq_khz> <call> <comment> <utc>" line
# Example: "DX de SP3ABC-#: 14074.0 DX1CALL FT8 -12 dB 1234Z"
_DX_RE = re.compile(
r'^DX\s+de\s+([\w\-#/]+):?\s+([\d.]+)\s+([\w/\-]+)\s+(.*?)(\d{4}Z)?\s*$',
re.IGNORECASE
)
# FIX (2026-10-21, live report: spot comments showed "1755Z JO82" /
# "1755Z00" glued on, or just the time when there was no comment): _DX_RE
# above only recognises the time when it's the LAST thing on the line.
# Many clusters append more after it (the spotter's locator "1755Z JO82",
# stray characters, BEL \x07 control chars), so the time silently ended up
# inside the comment. This finds the time token at the tail regardless of
# what follows it and cuts it (plus everything after) off the comment.
_DX_TAIL_TIME_RE = re.compile(r'(?<!\d)(\d{4})Z\S*(?:\s+\S+)?\s*$', re.IGNORECASE)
_CTRL_CHARS_RE = re.compile(r'[\x00-\x1f\x7f]')
class DXClusterClient:
"""A single connection to a DX Cluster server for one user."""
IDLE_PROBE_S = 300.0 # cisza po ktorej wysylamy sonde (pusta linie)
PROBE_WAIT_S = 120.0 # ile czekamy na odpowiedz na sonde
def __init__(self, host: str, port: int, login: str, password: str = "",
on_spot: Optional[Callable[[dict], None]] = None,
on_status: Optional[Callable[[str, str], None]] = None):
self.host = host
self.port = port
self.login = login
self.password = password
self.on_spot = on_spot
self.on_status = on_status # (status, message) - "connecting" / "connected" / "disconnected" / "error"
self._reader: Optional[asyncio.StreamReader] = None
self._writer: Optional[asyncio.StreamWriter] = None
self._task: Optional[asyncio.Task] = None
self._connected = False
self._should_run = False
self._reconnect_delay = 5.0
self._max_reconnect_delay = 60.0
def is_connected(self) -> bool:
return self._connected
async def _status(self, status: str, msg: str = ""):
_log_line(f"{self.login}@{self.host}:{self.port} -> {status}" + (f" ({msg})" if msg else ""))
if self.on_status:
try:
res = self.on_status(status, msg)
if asyncio.iscoroutine(res): await res
except Exception as e:
print(f"[dx] on_status callback error: {e}")
async def connect(self):
"""Start the connection with background auto-reconnect."""
if self._task and not self._task.done():
return
self._should_run = True
self._task = asyncio.create_task(self._run_loop())
async def disconnect(self):
"""Stop the connection."""
self._should_run = False
if self._writer:
try:
self._writer.close()
await self._writer.wait_closed()
except Exception:
pass
self._writer = None
self._reader = None
self._connected = False
if self._task:
self._task.cancel()
try:
await self._task
except (asyncio.CancelledError, Exception):
pass
self._task = None
await self._status("disconnected", "")
async def _run_loop(self):
"""Connection loop with auto-reconnect on disconnect."""
delay = self._reconnect_delay
while self._should_run:
try:
await self._status("connecting", f"{self.host}:{self.port}")
self._reader, self._writer = await asyncio.wait_for(
asyncio.open_connection(self.host, self.port),
timeout=10.0
)
self._connected = True
delay = self._reconnect_delay # reset the delay after success
await self._status("connected", "")
try:
sock = self._writer.get_extra_info('socket')
if sock is not None:
sock.setsockopt(socket.SOL_SOCKET, socket.SO_KEEPALIVE, 1)
except Exception:
pass
# Login sequence
await self._do_login()
# Receive lines
# Wykrywanie martwego polaczenia (zgloszone na zywo: admin w
# nocy "wylaczyl sie" i wrocil dopiero po recznym
# rozlacz/polacz). readline() bez limitu czasu wisial w
# nieskonczonosc na polotwartym TCP (serwer/router zerwal
# polaczenie bez FIN) - _connected zostawal True, wiec ani
# petla reconnect, ani auto_connect przy logowaniu nic nie
# robily. Teraz: po IDLE_PROBE_S ciszy wysylamy pusta linie
# (DXSpider odpowiada promptem), a jesli przez PROBE_WAIT_S
# dalej nic nie przyjdzie - zrywamy i laczymy od nowa.
probe_sent = False
while self._should_run:
try:
line = await asyncio.wait_for(
self._reader.readline(),
timeout=self.PROBE_WAIT_S if probe_sent else self.IDLE_PROBE_S)
except asyncio.TimeoutError:
if probe_sent:
_log_line(f"{self.login}@{self.host}:{self.port} brak odpowiedzi na sonde "
f"- polaczenie martwe, reconnect")
raise ConnectionError("brak danych z serwera - polaczenie martwe")
_log_line(f"{self.login}@{self.host}:{self.port} cisza {self.IDLE_PROBE_S:.0f}s - wysylam sonde")
self._writer.write(b"\r\n")
await self._writer.drain()
probe_sent = True
continue
probe_sent = False
if not line:
break # server closed
try:
text = line.decode('utf-8', errors='ignore').strip()
except Exception:
continue
if text:
await self._process_line(text)
except asyncio.CancelledError:
raise
except asyncio.TimeoutError:
# NOTE: kept in Polish - this message IS shown directly in
# the UI status badge (public/js/dxcluster.js's
# updateStatusBadge), not routed through I18n.
await self._status("error", "Timeout polaczenia")
except Exception as e:
await self._status("error", str(e))
# Cleanup and wait before reconnecting
self._connected = False
if self._writer:
try: self._writer.close()
except: pass
self._writer = None
if not self._should_run:
break
# NOTE: kept in Polish - see the note above on the "error" status message.
await self._status("disconnected", f"reconnect za {delay:.0f}s")
try:
await asyncio.sleep(delay)
except asyncio.CancelledError:
break
delay = min(delay * 1.5, self._max_reconnect_delay)
async def _do_login(self):
"""Send the login and password. The server may prompt with 'login:',
'Please enter your call:', 'password:' etc. We use a simple strategy:
send the call, wait 500ms, then send the password if there is one."""
if not self._writer:
return
# Basic strategy — wait for the prompt, then send the login
await asyncio.sleep(0.5)
try:
self._writer.write((self.login + "\r\n").encode('ascii'))
await self._writer.drain()
_log_line(f"{self.login}@{self.host}:{self.port} sent login")
except Exception as e:
print(f"[dx] error sending login: {e}")
_log_line(f"{self.login}@{self.host}:{self.port} ERROR sending login: {e}")
return
# If there's a password, wait briefly and send it
if self.password:
await asyncio.sleep(1.0)
try:
self._writer.write((self.password + "\r\n").encode('ascii'))
await self._writer.drain()
_log_line(f"{self.login}@{self.host}:{self.port} sent password ({len(self.password)} chars)")
except Exception as e:
print(f"[dx] error sending password: {e}")
_log_line(f"{self.login}@{self.host}:{self.port} ERROR sending password: {e}")
async def _process_line(self, text: str):
"""Parse a line and (if it's a spot) call the callback."""
text = _CTRL_CHARS_RE.sub('', text).rstrip()
m = _DX_RE.match(text)
if not m:
# Non-spot lines (banner, prompts, sysop announcements, kick
# messages like "Reconnected as X ... this instance is
# disconnected") - exactly the lines that explain an
# unexpected disconnect, so these go to the log (spot lines
# themselves don't - too frequent, and not diagnostically
# useful for connection-stability issues).
_log_line(f"{self.login}@{self.host}:{self.port} << {text}")
return
spotter = m.group(1)
freq_str = m.group(2)
call = m.group(3)
comment = (m.group(4) or '').strip()
utc = m.group(5) or ''
if not utc:
_t = _DX_TAIL_TIME_RE.search(comment)
if _t:
utc = _t.group(1) + 'Z'
comment = comment[:_t.start()].strip()
# Freq is usually in kHz with a fraction (14074.0)
try:
freq_hz = int(round(float(freq_str) * 1000))
except ValueError:
return
spot = {
"type": "dx_spot",
"freq_hz": freq_hz,
"call": call.upper(),
"spotter": spotter.upper(),
"comment": comment[:60], # capped at 60 characters
"utc": utc,
"ts": time.time(),
"band": _get_band(freq_hz),
"mode": _guess_mode(freq_hz, comment),
}
if self.on_spot:
try:
res = self.on_spot(spot)
if asyncio.iscoroutine(res): await res
except Exception as e:
print(f"[dx] on_spot callback error: {e}")
async def send_command(self, cmd: str):
"""Send a command to the server (e.g. 'sh/dx 20m FT8', 'set/qth', 'q')."""
if not self._writer or not self._connected:
return False
try:
self._writer.write((cmd + "\r\n").encode('ascii'))
await self._writer.drain()
return True
except Exception as e:
print(f"[dx] send_command error: {e}")
return False
class ClusterManager:
"""Manages DX Cluster connections for all users.
Each user has at most 1 connection. The broadcast callback delivers
messages to the right user's WebSocket."""
def __init__(self, on_broadcast: Callable[[str, dict], None]):
"""on_broadcast(user_id, message) — sends over WS to a specific user."""
self.on_broadcast = on_broadcast
self._clients: dict[str, DXClusterClient] = {}
# Cache of the last N spots per user - returned when a tab is reopened.
# 300 (bylo 100): na zapchanym pasmie zalew spotow cyfrowych (FT8)
# wypychal z tego bufora spoty CW/SSB w czasie znacznie krotszym niz
# ich docelowy czas zycia na fladze wodospadu (SPOT_FLAG_TTL_S w
# waterfall.js, 3 min) - zgloszone na zywo jako "flagi z dx clustra
# pala sie zbyt krotko".
self._spot_history: dict[str, list[dict]] = {}
self._max_history = 300
def get_client(self, user_id: str) -> Optional[DXClusterClient]:
return self._clients.get(user_id)
def connected_user_ids(self) -> list[str]:
"""user_id's with a live (connect() called) client - used by
webapp.py's idle watchdog to find candidates for auto-disconnect."""
return list(self._clients.keys())
def get_history(self, user_id: str) -> list[dict]:
return list(self._spot_history.get(user_id, []))
def get_merged_history(self) -> list[dict]:
"""Union of every connected login's spot history - the admin's
always-on one AND any other user's own, if they also connect one.
Matches the live broadcast (webapp.py::_dx_broadcast now sends every
dx_spot to all online users, not just the one whose connection
received it) so a fresh page load/reconnect sees the same picture
the live feed would build up. Raw/undeduped on purpose - the client
(dxcluster.js::_addSpot) replays this through the same de-dup path
used for live spots, so there's only one place that knows about the
call+frequency tolerance."""
merged: list[dict] = []
for hist in self._spot_history.values():
merged.extend(hist)
merged.sort(key=lambda s: s.get("ts", 0))
return merged[-self._max_history:]
async def connect_user(self, user_id: str, host: str, port: int,
login: str, password: str = ""):
"""Create/reset the connection for a user."""
# Close the previous connection if one exists
old = self._clients.pop(user_id, None)
if old:
await old.disconnect()
async def _on_spot(spot):
# Cache in the history
hist = self._spot_history.setdefault(user_id, [])
hist.append(spot)
if len(hist) > self._max_history:
del hist[:len(hist) - self._max_history]
# Broadcast to the user
res = self.on_broadcast(user_id, spot)
if asyncio.iscoroutine(res): await res
async def _on_status(status, msg):
res = self.on_broadcast(user_id, {
"type": "dx_status", "status": status, "message": msg,
})
if asyncio.iscoroutine(res): await res
# SSRF guard - see _resolve_and_check_safe's docstring. Checked
# here (not just in webapp.py's /api/dxcluster/config save) so
# EVERY path that actually opens the connection is covered,
# including the auto_connect-on-login path in ws_handler.
_safe, _why = await _resolve_and_check_safe(host, port)
if not _safe:
await _on_status("error", _why)
return None
client = DXClusterClient(host, port, login, password,
on_spot=_on_spot, on_status=_on_status)
self._clients[user_id] = client
await client.connect()
return client
async def disconnect_user(self, user_id: str):
client = self._clients.pop(user_id, None)
if client:
await client.disconnect()
async def send_command(self, user_id: str, cmd: str) -> bool:
client = self._clients.get(user_id)
if not client: return False
return await client.send_command(cmd)
async def shutdown_all(self):
for user_id in list(self._clients.keys()):
await self.disconnect_user(user_id)