webapp.py
Rama main · réplica de lectura del repositorio original.
#!/usr/bin/env python3
"""
webapp.py — WebSocket Hub + application (HTTP/WS routing, API).
"""
import re, time, json, asyncio, struct
import numpy as np
import aiohttp
import aiohttp.web as web
# ══════════════════════════════════════════════════════════════════════════════
# PERFORMANCE OPTIMIZATIONS (async sending, fast JSON)
# ══════════════════════════════════════════════════════════════════════════════
# orjson - if installed, 5-10x faster JSON than stdlib. Falls back to stdlib.
# Can be installed with: py -m pip install orjson
try:
import orjson
def _fast_json_bytes(msg):
"""Encode a message to UTF-8 bytes. Much faster than json.dumps."""
return orjson.dumps(msg, option=orjson.OPT_SERIALIZE_NUMPY)
_JSON_BACKEND = "orjson"
except ImportError:
def _fast_json_bytes(msg):
return json.dumps(msg, ensure_ascii=False).encode('utf-8')
_JSON_BACKEND = "stdlib"
# winloop/uvloop - a faster event loop than plain asyncio.
# Windows: winloop, Linux/Mac: uvloop. Install with:
# py -m pip install winloop (Windows)
# pip install uvloop (Linux/Mac)
def _install_fast_loop():
import sys
try:
if sys.platform == 'win32':
import winloop
winloop.install()
return "winloop"
else:
import uvloop
uvloop.install()
return "uvloop"
except ImportError:
return "asyncio-stdlib"
_LOOP_BACKEND = _install_fast_loop()
print(f"[perf] JSON={_JSON_BACKEND}, event_loop={_LOOP_BACKEND}")
# ══════════════════════════════════════════════════════════════════════════════
# Static file cache — keeps JS/CSS/HTML files in RAM (pre-compressed).
# Every GET for a static file gets bytes directly from the Map, no disk read.
#
# Files are cached lazily (on first request) and **unchanged** for the
# server's lifetime - restarting Python purges the cache. Fine for dev-time
# use: you restarted the server so they get the new build.
#
# Compression: gzip on-the-fly on first cache, bytes stored once.
# Brotli would be better but needs installing; gzip is in stdlib.
# ══════════════════════════════════════════════════════════════════════════════
import gzip
# Map[str_path] -> (mtime, bytes_raw, bytes_gz | None, mime, etag)
_STATIC_CACHE: dict = {}
_STATIC_CACHE_LOCK = None # asyncio.Lock created when the loop starts
# MIME types worth gzipping (text-based). Binary ones (png, opus) are
# already compressed so gzip gains nothing but adds overhead.
_GZIP_MIMES = {
"text/html", "text/css", "text/plain", "text/javascript",
"application/javascript", "application/json", "application/xml",
"image/svg+xml",
}
def _cache_static_file(fpath, mime: str) -> tuple:
"""Read a file from disk, pre-compress if text-based, return a cache entry."""
import hashlib
raw = fpath.read_bytes()
mtime = fpath.stat().st_mtime
# ETag = content hash (strong) - the client can send If-None-Match
etag = '"' + hashlib.md5(raw).hexdigest()[:16] + '"'
# gzip if text-based and > 1KB (smaller files aren't worth the overhead)
gz = None
if mime.split(";")[0].strip() in _GZIP_MIMES and len(raw) > 1024:
try:
gz = gzip.compress(raw, compresslevel=6) # 6 = speed/ratio balance
# Only keep the gzip version if it's actually smaller (already-compressed files have ~0.95 ratio)
if len(gz) >= len(raw) * 0.95:
gz = None
except Exception:
gz = None
return (mtime, raw, gz, mime, etag)
# Semantic version (MAJOR.MINOR.PATCH) - the public-facing version users see
# in the installer and in the GitHub release tag (vX.Y.Z), and what the
# update-check below compares against releases/latest. NOT the same thing as
# the "[build] webapp.py wersja BUILD-YYYY-MM-DD-..." line printed at
# startup, which is an internal per-commit diagnostic marker used to confirm
# a rebuilt EXE actually picked up the latest source - that one changes on
# nearly every commit, this one only on an actual release. Keep this in sync
# with VERSION (repo root) and #define AppVersion in HAMCTRL-installer.iss -
# all three are bumped together at release time, not per-commit.
SERVER_VERSION = "2.0.17"
# Internal per-commit diagnostic marker (see the comment above). Defined
# once here and printed both at server startup (server.py::amain, so it's
# visible without needing to transmit anything first) and at first FT8/FT4
# TX encode below - the two used to be out of sync (only the TX-time print
# existed despite the comment above claiming "printed at startup").
BUILD_MARKER = "BUILD-2026-11-02-LIGHT-MODE-COLORS"
# ── Update check ──────────────────────────────────────────────────────────────
# HAMCTRL checks GitHub Releases for a newer version and shows the admin a
# notice (it never downloads or installs anything — the admin fetches the new
# installer manually). Set this to your repo once it exists, e.g.
# "YOURCALL/hamctrl". While empty, the update check is simply skipped.
GITHUB_REPO = "SQ3MZM/HAMCTRL"
from config import (CALLSIGN, LOCATOR, PORT, HAMLIB_MODELS, SCOPE_MODELS,
CIV_NATIVE_MODELS,
MIME, PUBLIC, CFG_F, USR_F, ADMIN_PW, FIRST_RUN, VERBOSE, DATA,
WEBRTC_UDP_PORTS, WEBRTC_UPNP)
from auth import (jwt_sign, jwt_verify, hash_pw, hash_pw_secure,
verify_pw, needs_rehash)
from data import get_cfg, get_users, load_json, save_json, DEFAULT_MACROS
from crypto_secrets import encrypt_secret, decrypt_secret
import qso_db
import callbook
from audio import enumerate_audio_devices, auto_detect_radio_audio
from audio_stream import AudioStream
try:
from webrtc_audio import WebRTCAudioReceiver, install_udp_port_range
from webrtc_rx_audio import WebRTCAudioSender
_WEBRTC = True
_udp_range = install_udp_port_range(WEBRTC_UDP_PORTS)
if _udp_range:
print(f"[webrtc] audio UDP ports {_udp_range[0]}-{_udp_range[1]}"
+ (" - otwieram na routerze przez UPnP" if WEBRTC_UPNP
else " - przekieruj ten zakres UDP na routerze na ten komputer"), flush=True)
except Exception as e:
print(f"[webrtc] unavailable: {e}")
_WEBRTC = False
from wsjtx_udp import WsjtxUdpServer
from ft8_all_log import Ft8AllLog
from tunnel_manager import TunnelManager
from rotator import Rotator
from rigcat import RigCAT
from civ import CivRig
from com_bridge_ws import ComBridgeWs
import ft8_encoder
import ft4_encoder
try:
from dxcluster import ClusterManager
_DXCLUSTER_OK = True
except Exception as e:
print(f"[dxcluster] module unavailable: {e}")
_DXCLUSTER_OK = False
try:
from relay_controller import RelayController, list_serial_ports, MAX_PULSE_S, RELAY_COUNT
_RELAY_OK = True
except Exception as e:
print(f"[relay] module unavailable: {e}")
_RELAY_OK = False
import ft8_rx_decoder
try:
from deepcw_engine import deepcw_engine
except Exception as _e:
deepcw_engine = None
print(f"[deepcw] engine unavailable: {_e}", flush=True)
import ft4_rx_decoder
import waterfall
import qso_engine
from rigs.features import (FEATURES, effective_features, features_for_admin,
default_enabled_features, effective_dynamic,
dynamic_for_admin)
# Phone (not tablet) User-Agent heuristic for the / auto-routing to
# mobile.html — the same "Mobi" token approach used by most server-side
# mobile redirects (deliberately excludes iPad/Android tablets, which
# have enough screen for the desktop UI's scale-to-fit).
_MOBILE_UA_RE = re.compile(r'Mobi|iPhone|iPod|Android.*Mobile|Windows Phone|BlackBerry|Opera Mini|IEMobile')
# Operator chat (session-only, see App._chat_history) - ring buffer size
# and the per-message length cap (matches the input's maxlength in the HTML).
_CHAT_HISTORY_MAX = 150
_CHAT_MSG_MAX_LEN = 500
# PTT timing log - zgloszone na zywo (SDR we Wloszech): "wcisniecie PTT nie
# odrazu transmituje, jest zwloka". Istniejacy pomiar czasu wymiany CI-V przy
# PTT byl juz w kodzie, ale TYLKO za flaga VERBOSE i tylko do print() - w
# spakowanym EXE (bez okna konsoli) nikt tego nie widzi. Osobny maly plik
# tekstowy, zawsze wlaczony (nie tylko VERBOSE), ten sam wzorzec co
# dxcluster.log - zeby po fakcie bylo co wkleic zamiast zgadywac czy to
# siec, CI-V, czy potok audio.
_PTT_LOG_FILE = DATA / "ptt_timing.log"
_PTT_LOG_MAX_BYTES = 500_000
def _ptt_log_line(msg: str):
try:
ts = time.strftime('%Y-%m-%d %H:%M:%S')
line = f"[{ts}] {msg}\n"
if _PTT_LOG_FILE.exists() and _PTT_LOG_FILE.stat().st_size > _PTT_LOG_MAX_BYTES:
try:
data = _PTT_LOG_FILE.read_text(encoding='utf-8', errors='ignore')
_PTT_LOG_FILE.write_text(data[len(data) // 2:], encoding='utf-8')
except Exception:
pass
with open(_PTT_LOG_FILE, 'a', encoding='utf-8') as f:
f.write(line)
except Exception:
pass # logging must never break PTT
# DIAGNOSTIC (2026-10-20, live report: "user nie raz to wystepuje tylko co
# jakies logowanie widzi duzo za duzo trybow" - checkboxes in KONFIGURACJA
# show only 4 modes enabled, but on SOME logins the regular user's mode
# grid shows ~13). Intermittent, not reproduced from static code review
# alone (the save->self.cfg->GET->S.modes path reads correctly on paper) -
# same reasoning as ptt_timing.log/dxcluster.log above: log every GET
# /api/config/modes response instead of guessing, so the NEXT occurrence
# can be compared against what KONFIGURACJA actually has checked at that
# moment.
_MODES_LOG_FILE = DATA / "modes_config.log"
_MODES_LOG_MAX_BYTES = 200_000
def _modes_log_line(msg: str):
try:
ts = time.strftime('%Y-%m-%d %H:%M:%S')
line = f"[{ts}] {msg}\n"
if _MODES_LOG_FILE.exists() and _MODES_LOG_FILE.stat().st_size > _MODES_LOG_MAX_BYTES:
try:
data = _MODES_LOG_FILE.read_text(encoding='utf-8', errors='ignore')
_MODES_LOG_FILE.write_text(data[len(data) // 2:], encoding='utf-8')
except Exception:
pass
with open(_MODES_LOG_FILE, 'a', encoding='utf-8') as f:
f.write(line)
except Exception:
pass
# ══════════════════════════════════════════════════════════════════════════════
# WEBSOCKET HUB (aiohttp)
# ══════════════════════════════════════════════════════════════════════════════
# Message type -> channel mapping. Anything not listed here goes to
# 'control' (the default). This allows transparent routing without
# modifying hundreds of existing hub.broadcast() calls — just set the
# channels per the table below and broadcast fills in the channel itself.
_MSG_TYPE_TO_CHANNEL = {
# CI-V waterfall (Radio panel) - high volume, only for the Radio tab
'scope_frame': 'scope',
'scope_reset': 'scope',
# FT8/FT4/WSJT-X - high volume of decodes and waterfall, only for the FT8 tab
'ft8_waterfall': 'ft8',
'wsjtx_decode': 'ft8',
'wsjtx_clear': 'ft8',
'wsjtx_status': 'ft8',
'wsjtx_call_activity': 'ft8',
'wsjtx_qso_logged': 'ft8',
'wsjtx_tx_start': 'ft8',
'wsjtx_tx_stop': 'ft8',
'ft8_tx_freq': 'ft8',
'ft8_tx_status': 'ft8',
'ft8_tx_error': 'ft8',
'ft8_tx_halted': 'ft8',
'ft8_tx_period': 'ft8',
'ft8_tx_started': 'ft8',
'ft8_period_set': 'ft8',
'ft8_rx_status': 'ft8',
'ft8_rx_freq': 'ft8',
'ft8_split_status': 'ft8',
'ft8_decode_mode': 'ft8',
'auto_seq_status': 'ft8',
'auto_qso_status': 'ft8',
'auto_qso_error': 'ft8',
'auto_qso_complete': 'ft8',
'tune_status': 'ft8',
'hound_status': 'ft8',
'hound_step': 'ft8',
# The CW decoder works independently of the FT8 tab — the window can
# be opened from the keyer panel while operating CW. On the 'ft8'
# channel the text only reached clients subscribed to FT8: the backend
# logged decodes, but the CW window stayed silent because the
# recipient list was empty. Every client has 'control', and this is
# only a few characters per second — zero cost.
'deepcw_text': 'control',
# DX Cluster - only for the DXCluster tab
'dx_spot': 'dxcluster',
'dx_status': 'dxcluster',
}
def _channel_for_msg(msg: dict) -> str:
"""Pick a channel based on the message type. Default = 'control'."""
return _MSG_TYPE_TO_CHANNEL.get(msg.get('type', ''), 'control')
class WSHub:
"""Broadcast hub with channel subscriptions.
Clients subscribe only to the channels they care about (e.g. the Log
tab doesn't need scope_frame or ft8_waterfall). Broadcast filters
recipients by the channel parameter — an unsubscribed client doesn't
get the message.
Channels:
- 'control': default - everything about radio and system state
(freq, mode, ptt, radio_lock, chat, presence, toasts)
- 'scope': scope_frame from CI-V (Radio panel waterfall)
- 'ft8': ft8_waterfall, wsjtx_decode, auto_seq_status, tune_status etc.
- 'dxcluster': dx_spot, dx_status
The 'audio' channel is NOT here - audio goes through a separate
WebSocket path (audio_stream.py) with its own subscriber mechanism.
"""
def __init__(self):
self._clients: set = set()
# dict: ws -> set of channels the client subscribed to
# Default on add() is {'control'} - everyone gets radio control.
self._subs: dict = {}
self._lock = asyncio.Lock()
self._loop = None
self._last_ft8_diag = 0.0
def set_loop(self, loop):
self._loop = loop
async def add(self, ws):
async with self._lock:
self._clients.add(ws)
# Default: ALL channels - the client will drop the ones it
# doesn't want via subscribe {mode:'set'}. This resolves a
# race condition: a new client gets wsjtx_decode EVEN IF it
# hasn't yet sent subscribe {channels:['ft8']} before the
# first broadcast. Channel filtering still wins the same way
# — a client on the Log tab sends subscribe
# {channels:['control']} a moment later and stops getting
# scope_frame/ft8_waterfall.
self._subs[ws] = {'control', 'scope', 'ft8', 'dxcluster'}
async def remove(self, ws):
async with self._lock:
self._clients.discard(ws)
self._subs.pop(ws, None)
async def subscribe(self, ws, channels):
"""Client adds channels to its subscription (doesn't replace it)."""
async with self._lock:
if ws not in self._subs:
self._subs[ws] = set()
self._subs[ws].update(channels)
async def unsubscribe(self, ws, channels):
"""Client removes channels from its subscription. 'control' always stays."""
async with self._lock:
if ws not in self._subs:
return
self._subs[ws].difference_update(channels)
self._subs[ws].add('control') # can never be dropped
async def set_channels(self, ws, channels):
"""Set the full channel set (used when switching tabs)."""
async with self._lock:
self._subs[ws] = set(channels) | {'control'}
async def broadcast(self, msg: dict, skip=None, channel: str = None):
"""Parallel broadcast to a channel's subscribers.
If channel=None, the channel is auto-picked based on msg['type']
via _channel_for_msg() — so existing code calling hub.broadcast(msg)
doesn't need to be modified.
OPTIMIZATIONS:
1. orjson instead of json.dumps (5-10x faster for large objects)
2. Encode once to a string (not N times for N clients)
3. asyncio.gather instead of sequential await (parallel send)
4. Subscription filtering - a client only gets the channels it wants
5. Removed prints from the hot path
NOTE: we use send_str (not send_bytes) because the frontend
distinguishes:
- Binary (ArrayBuffer) -> Opus audio frame
- Text -> JSON control message
"""
if not self._clients:
return
# Auto-route the channel if not given explicitly
if channel is None:
channel = _channel_for_msg(msg)
# Snapshot clients without the lock + filter by channel
clients = [
ws for ws in self._clients
if ws is not skip and channel in self._subs.get(ws, set())
]
if channel == "ft8" and msg.get("type") in ("wsjtx_decode", "ft8_waterfall"):
# Diagnostic for the "mobile FT8 tab gets no decodes/waterfall"
# report - confirms whether these messages actually have any
# 'ft8'-subscribed recipient at all. Throttled (once/5s) since
# ft8_waterfall streams several times a second during decoding.
_now = time.time()
if _now - self._last_ft8_diag > 5.0:
self._last_ft8_diag = _now
print(f"[ws] broadcast type={msg.get('type')} channel=ft8 "
f"recipients={len(clients)}/{len(self._clients)}", flush=True)
if not clients:
return
try:
if _JSON_BACKEND == "orjson":
data = _fast_json_bytes(msg).decode('utf-8')
else:
data = json.dumps(msg, ensure_ascii=False)
except Exception:
data = json.dumps(msg, ensure_ascii=False, default=str)
# EPHEMERAL channels (scope, audio, ft8_waterfall) — data arrives
# many times per second, so a single dropped frame doesn't matter.
# When a client reads slowly, a TLS send blocks the loop waiting
# for the buffer to drain (backpressure — found via looplag:
# sslproto _do_write). We give it a short timeout: not keeping up
# = skip it for THIS frame, instead of freezing everyone. Important
# channels (control, chat) are sent normally.
_ephemeral = channel in ("scope", "audio", "ft8_waterfall")
async def _send_one(ws):
if _ephemeral:
# Slow client: data piles up in the TLS buffer, the next
# write blocks the loop (backpressure). Short timeout — not
# keeping up within 0.25s = skip this ephemeral frame for
# it (scope/audio, the next one will come). The buffer is
# capped by writer_limit when the WS is created (see
# ws_handler), so send_str quickly signals congestion
# instead of growing without bound.
try:
await asyncio.wait_for(ws.send_str(data), timeout=0.25)
except asyncio.TimeoutError:
return None
else:
await ws.send_str(data)
results = await asyncio.gather(
*[_send_one(ws) for ws in clients],
return_exceptions=True
)
dead = [ws for ws, res in zip(clients, results) if isinstance(res, Exception)]
if dead:
async with self._lock:
for ws in dead:
self._clients.discard(ws)
self._subs.pop(ws, None)
def broadcast_sync(self, msg: dict, channel: str = None):
"""Broadcast from a non-async thread (thread-safe).
The channel is auto-routed based on msg['type'] if not given."""
if self._loop and self._loop.is_running():
asyncio.run_coroutine_threadsafe(
self.broadcast(msg, channel=channel), self._loop
)
# ══════════════════════════════════════════════════════════════════════════════
# APPLICATION
# ══════════════════════════════════════════════════════════════════════════════
def _adif_field(name: str, value) -> str:
"""Build an ADIF field: <name:length>value. Empty -> skipped."""
if value is None:
return ""
s = str(value).strip()
if not s:
return ""
return f"<{name}:{len(s)}>{s}"
def qso_to_adif(qso: dict) -> str:
"""
Convert a QSO (a dict from our database) into the ADIF string required
by Cloudlog/WaveLog.
The Cloudlog API /index.php/api/qso expects:
{"key":..., "station_profile_id":..., "type":"adif", "string":"<call:5>...<eor>"}
It does NOT accept JSON fields (call, band, mode...) — that was a bug:
QSOs weren't reaching Cloudlog despite an HTTP 200 response.
Frequency: ADIF requires MHz (e.g. 14.074). Our database holds Hz or
MHz depending on the source (WSJT-X vs manual entry) - we normalize it.
"""
freq_raw = str(qso.get("freq", "")).strip()
freq_mhz = ""
if freq_raw:
try:
f = float(freq_raw)
if f > 1_000_000: # Hz
freq_mhz = f"{f / 1_000_000:.6f}".rstrip("0").rstrip(".")
elif f > 1000: # kHz
freq_mhz = f"{f / 1000:.6f}".rstrip("0").rstrip(".")
else: # already MHz
freq_mhz = f"{f:.6f}".rstrip("0").rstrip(".")
except (ValueError, TypeError):
freq_mhz = ""
parts = [
_adif_field("call", qso.get("call", "")),
_adif_field("qso_date", qso.get("qso_date", "")),
_adif_field("time_on", qso.get("time_on", "")),
_adif_field("time_off", qso.get("time_off", "")),
_adif_field("band", qso.get("band", "")),
_adif_field("mode", qso.get("mode", "")),
_adif_field("freq", freq_mhz),
_adif_field("rst_sent", qso.get("rst_sent", "")),
_adif_field("rst_rcvd", qso.get("rst_rcvd", "")),
_adif_field("gridsquare", qso.get("gridsquare", "")),
_adif_field("station_callsign", qso.get("my_call", "")),
_adif_field("operator", qso.get("my_call", "")),
_adif_field("my_gridsquare", qso.get("my_gridsquare", "")),
_adif_field("tx_pwr", qso.get("power", "")),
_adif_field("comment", qso.get("comment", "")),
_adif_field("prop_mode", qso.get("prop_mode", "")),
_adif_field("sat_name", qso.get("sat_name", "")),
_adif_field("sat_mode", qso.get("sat_mode", "")),
_adif_field("freq_rx", qso.get("freq_rx", "")),
_adif_field("band_rx", qso.get("band_rx", "")),
_adif_field("name", qso.get("name", "")),
_adif_field("qth", qso.get("qth", "")),
_adif_field("dxcc", qso.get("dxcc", "")),
_adif_field("country", qso.get("country", "")),
_adif_field("cont", qso.get("cont", "")),
_adif_field("cqz", qso.get("cqz", "")),
_adif_field("ituz", qso.get("ituz", "")),
_adif_field("state", qso.get("state", "")),
_adif_field("iota", qso.get("iota", "")),
_adif_field("qsl_sent", qso.get("qsl_sent", "")),
_adif_field("qsl_rcvd", qso.get("qsl_rcvd", "")),
_adif_field("lotw_qsl_sent", qso.get("lotw_qsl_sent", "")),
_adif_field("lotw_qsl_rcvd", qso.get("lotw_qsl_rcvd", "")),
_adif_field("lotw_qslsdate", qso.get("lotw_qslsdate", "")),
_adif_field("lotw_qslrdate", qso.get("lotw_qslrdate", "")),
_adif_field("eqsl_qsl_sent", qso.get("eqsl_qsl_sent", "")),
_adif_field("eqsl_qsl_rcvd", qso.get("eqsl_qsl_rcvd", "")),
_adif_field("pota_ref", qso.get("pota_ref", "")),
_adif_field("sota_ref", qso.get("sota_ref", "")),
_adif_field("wwff_ref", qso.get("wwff_ref", "")),
]
return "".join(p for p in parts if p) + "<eor>"
class App:
def __init__(self):
self.hub = WSHub()
self.cfg = get_cfg()
# Pick the backend right away based on the saved model in
# config.json, so we don't needlessly create a RigCAT (rigctld) at
# startup for a model that's going to go through CivRig
# (CIV_NATIVE_MODELS) anyway — avoids a serial-port conflict between
# the two backends.
_rigs = self.cfg.get("rigs") or [{}]
_saved_model = str(_rigs[0].get("model", ""))
if _saved_model in CIV_NATIVE_MODELS:
self.rig = CivRig(self.cfg, self._rig_bcast, log=print)
print(f"[rig] start backend -> direct CI-V (saved model {_saved_model})")
else:
self.rig = RigCAT()
self.rotators: list[Rotator] = []
self.users = get_users()
self._migrate_plaintext_secrets()
self.audio = AudioStream()
self.audio.cfg = self.cfg.get("audio", {})
# Radio audio card auto-detection — if the config has NO RX/TX
# devices set (or user_audio_auto=True), detect the card
# automatically by name. Recognizes IC-7300, IC-705, FT-991, etc.
self._audio_auto = self.cfg.setdefault("audio_auto_detect", True)
if self._audio_auto:
try:
detection = auto_detect_radio_audio()
if detection["detected"]:
if "audio" not in self.cfg: self.cfg["audio"] = {}
# Only overwrite if the user hasn't set it manually
if not self.cfg["audio"].get("rxDevice") and detection["rx"]:
self.cfg["audio"]["rxDevice"] = detection["rx"]
if not self.cfg["audio"].get("txDevice") and detection["tx"]:
self.cfg["audio"]["txDevice"] = detection["tx"]
self.audio.cfg = self.cfg["audio"]
print(f"[audio] auto-detect: {detection['pattern']} -> RX={detection['rx']!r}, TX={detection['tx']!r}")
self._audio_detection = detection
else:
print("[audio] auto-detect: radio card not detected, use manual configuration")
self._audio_detection = detection
except Exception as e:
print(f"[audio] auto-detect error: {e}")
self._audio_detection = {"detected": False, "rx": None, "tx": None,
"pattern": None, "all_rx": [], "all_tx": []}
else:
self._audio_detection = {"detected": False, "rx": None, "tx": None,
"pattern": None, "all_rx": [], "all_tx": []}
self.rust_audio = None # set by server.py when ham_audio.exe is available
self._rig_power_on = True # radio power state — updated by power_toggle
self._ft8_tx_abort = False
# Operator chat - session-only by design (no persistence layer),
# a simple in-memory ring buffer capped at _CHAT_HISTORY_MAX
# messages, see "chat_send" below. Cleared on every server restart.
self._chat_history: list[dict] = []
self._ft8_rx_enabled = False
# Who started WSJT-X (uid) — needed for auto-stop on disconnect or
# when the radio is released. Doesn't matter for multiple users
# since only one can decode at a time (the backend is global), but
# we need to know WHO in order to know when to stop.
self._ft8_rx_owner_uid: str | None = None
self._autoqso_uid: str | None = None # the operator who started the current CQ/auto-QSO
# Handoff item E (2026-10-20): candidates collected from the CURRENT
# decode batch when several stations reply to our CQ at once, plus
# the pending decision task that picks among them after a short
# collection window (see _decide_best_caller / qso_engine.pick_best_caller).
self._cq_candidates: dict = {}
self._cq_decision_task = None
self._last_auto_tx_key = None
self._last_auto_tx_action = None # dict of the last sent auto-QSO message, for retransmission
self._pre_pcm_cache = None # (call_to, call_de, report, pcm_bytes, duration) # auto-TX dedup: (call_to, report_or_grid)
self._tx_watchdog_task = None # asyncio Task for auto-off PTT
self._tune_stop = False # aborts the Tune tone
# COM Bridge WS - for Windows EXE clients that create a virtual COM.
# Clients connect over WSS to /ws/com-bridge, get a service->COM
# mapping, and forward bytes in both directions. Used by CW Skimmer,
# Logger32, HRD for remote access to the CI-V IC-7300 (and Yaesu/
# Kenwood in the future).
# Note: self.rig may be RigCAT() when CI-V is disabled - then the
# bridge has nothing to offer and clients see service_status civ=false.
civ_for_bridge = self.rig if isinstance(self.rig, CivRig) else None
# can_write: a user may WRITE to the radio through the COM Bridge
# only when they hold radio_lock or are an admin. Without this a
# user without the lock could tune the frequency via CW Skimmer/HRD,
# bypassing the UI locks.
def _com_can_write(uid: str) -> bool:
"""
May this user WRITE to the radio through the COM Bridge
(change freq/mode/PTT from CW Skimmer/HRD)?
Note: a viewer is already REJECTED at the WS connection stage
(com_bridge_ws_handler), so only admin and operators are
handled here.
Write access is granted to:
- admin (always)
- an operator HOLDING radio_lock
Not granted to:
- an operator WITHOUT the lock (view only)
- anyone without an account
"""
if not uid:
return False
u = next((x for x in self.users if x.get('id') == uid), None)
if not u:
return False
if u.get('role') == 'admin':
return True
# Operator - only when holding radio_lock
# (a viewer won't reach here - rejected at connection)
return self.radio_lock.get('user_id') == uid
self.com_bridge_ws = ComBridgeWs(civ_rig=civ_for_bridge, hub=self.hub,
log=print, can_write=_com_can_write)
# ── DX Cluster manager (per-user telnet connections) ──────────────────
self.dxcluster = None
self._server_start_time = time.time() # for computing uptime in /api/status
if _DXCLUSTER_OK:
async def _dx_broadcast(user_id: str, msg: dict):
# dx_spot: broadcast to EVERYONE online, not just the user
# whose cluster login received it (decided 2026-10-18) - this
# is one shared radio/waterfall, so the admin's always-on
# connection (see the idle watchdog's admin exemption) is
# meant to feed spot flags to every operator, not just admin.
# dx_status stays private to the connecting user - it's
# feedback about THEIR OWN connect attempt (e.g. "login
# failed"), broadcasting that to everyone would just be noise
# about a connection they have nothing to do with.
is_spot = msg.get("type") == "dx_spot"
for ws, info in list(self.online_users.items()):
if is_spot or info.get("user_id") == user_id:
try:
await ws.send_json(msg)
except Exception:
pass
self.dxcluster = ClusterManager(on_broadcast=_dx_broadcast)
# ── Relay controller (Arduino SP5IOU SDR220 emulator) ─────────────────
self.relay: RelayController | None = None
if _RELAY_OK:
rcfg = self.cfg.get("relay", {})
if rcfg.get("enabled") and rcfg.get("port"):
try:
self.relay = RelayController(
port=rcfg["port"],
baudrate=int(rcfg.get("baudrate", 9600)),
)
asyncio.ensure_future(self._relay_connect_task())
except Exception as e:
print(f"[relay] init error: {e}")
# ── Scope/waterfall state and TX marker ──────────────────────────────────
self._ft8_tx_freq_hz = 1000.0 # current target transmit frequency
# NOTE: NOT 1500Hz — that's a documented "sweet spot"/reference
# point of the IC-7300's internal digital processing in USB-D mode,
# which produces a fixed notch (~185Hz wide) at exactly that
# frequency in the RX audio path, independent of NB/NR/Notch/AGC/
# PTT. Confirmed on two independent recordings (Audacity, no code
# of ours involved) — present in USB-D, absent in plain USB. 1000Hz
# is a safe distance from that point.
self._ft8_tx_locked = False # OLD mechanism — to be removed, kept
# temporarily for future use in a "TX Hound" feature (planned
# layer, separate session). NO LONGER used by the current code.
self._ft8_tx_frozen = False # NEW mechanism: TX frozen in place,
# while RX automatically jumps to the frequency of every received
# message addressed to US (call_to == CALLSIGN) — lets the operator
# "freeze" TX and track the station currently transmitting to them,
# without manually moving the RX marker.
self._ft8_split_enabled = False # split mode: TX only above _ft8_split_min_hz
self._ft8_split_min_hz = 1200.0
self._ft8_rx_freq_hz = 1000.0 # independent RX marker (Rx Frequency panel) —
# starts at the same value as TX, but can be moved completely
# independently (e.g. listening on a different frequency than the current TX)
# ── Fake Split (Rig Split) ───────────────────────────────────────────
# Moves the VFO BEFORE transmitting so the TX audio lands near the
# center of the SSB filter (~1500Hz), instead of near its edge —
# at the edges the filter attenuates the signal (power drops,
# ALC/splatter). The VFO is restored after transmitting. The logic
# (the invariant on-air-freq = dial+audio) comes from
# fake_split_prototype.py. The enabled state is remembered in the config.
self._fake_split_enabled = bool(self.cfg.get("ft8", {}).get("fakeSplit", False))
self._fake_split_state = None # dict {dial_hz, audio_hz} to restore after TX, or None
# ── QSO automation (full FT8 automation) ────────────────────────────
self._qso_engine = qso_engine.QsoEngine(my_call=CALLSIGN, my_grid=LOCATOR)
self._auto_seq_enabled = True # ALWAYS active (there's no longer a
# UI toggle); clicking a macro is a manual override
# FT8 safety timer (WSJT-X's "Tx Watchdog") - False means the
# operator has NOT recently confirmed presence (click/TX macro),
# and the automation MUST stop reacting to callers (see
# _process_auto_qso) until an "ft8_timer_confirm" arrives from the
# frontend (FT8Timer.confirm() in wsjtx.js).
self._ft8_operator_present = True
self._auto_cq_text = None # the CQ text we're currently
# "transmitting" (used to recognize that a received reply concerns
# OUR CQ, not some unrelated message addressed to us outside a QSO context)
# Message scheduled to be sent in the NEXT available 15s window,
# generated by the automation (on_decode) — a single "slot", since
# at any moment there can be only one pending automatic transmission.
self._auto_pending_tx = None # dict {call_to, call_de, report_or_grid, r_flag} or None
# ── Periodic CQ calling ────────────────────────────────
# When the user calls CQ, we don't transmit just once - we repeat
# every full period (2 windows) until someone answers or the
# user/timer stops it. Without this, CQ only went out once.
self._cq_calling = False # whether we're actively calling CQ in a loop
self._cq_call_de = None # our call (for the repeating CQ)
self._cq_report = None # our grid (CQ CALL GRID)
self._cq_task = None # the CQ-repeat loop task
# Mutex protecting against PARALLEL execution of _ft8_tx_sequence —
# without this, if the decoder returns multiple messages in one
# window and more than one triggers an automatic reply (or the
# automation coincides in time with a manual TX), two parallel
# PTT/audio-feed calls would collide with each other. EVERY call to
# _ft8_tx_sequence (manual and automatic) must go through this lock.
self._ft8_tx_lock = asyncio.Lock()
self._ft8_tx_period = 1 # 1 = xx:00/xx:30 windows (default), 2 = xx:15/xx:45
self._qso_period_locked = False # True once a period is locked in for the current QSO
# Counter for "which automatic TX action is CURRENTLY the most
# recently scheduled one" — see the comment where it's used in
# _send_auto_tx/_ft8_tx_sequence_inner (stale-TX-guard).
self._autoqso_tx_seq = 0
self._ft8_decode_mode = "FT8" # "FT8" or "FT4" — which encoder/timing to use
self.webrtc = WebRTCAudioReceiver(
on_pcm_frame=self.audio.feed_tx_pcm,
on_track_started=self._webrtc_tx_start,
on_track_ended=self._webrtc_tx_stop,
) if _WEBRTC else None
# RX audio over WebRTC - one sender per connected client, keyed by
# their ws (see webrtc_rx_audio.py). The RX audio path (2026-08-24
# onward), replacing the ham_audio.exe WS/Opus path client-side.
# Unlike self.webrtc (TX, exclusive - one transmitter at a time),
# RX listening isn't exclusive, so this is a dict, not a single
# shared instance.
self._webrtc_rx_senders = {}
self.wsjtx = WsjtxUdpServer(self.hub.broadcast)
self.tunnel = TunnelManager(self.hub)
self._caps_cache = {}
# ── Radio Lock — one operator at a time ────────────────────────────────
# Who currently "holds" the radio (can tune, key PTT, change mode).
# Other users see the UI in read-only mode.
# timeout_min: minutes of inactivity after which the radio is auto-released.
# last_activity: time of the active operator's last action (for auto-release).
self.radio_lock: dict = {
"user_id": None,
"username": None,
"callsign": None,
"locked_at": None,
"last_activity": None,
"timeout_min": self.cfg.get("radio_lock_timeout", 20),
}
# Pending requests: {user_id: {"username", "callsign", "requested_at"}}
self.radio_requests: dict = {}
# Online users: {ws: {"user_id", "username", "callsign", "role", "joined_at"}}
self.online_users: dict = {}
# Login rate limiting (brute-force protection) - see the _login_* methods
self._login_fails_ip: dict = {} # {ip: {fails: [ts], blocked_until: ts}}
self._login_fails_user: dict = {} # {username: {...}}
# Background loops under SUPERVISION - if they crash, the supervisor
# restarts them (with growing backoff). Without this, one
# unexpected error kills the feature until the server is manually restarted.
self._supervise(lambda: self._radio_lock_watchdog(), "radio_lock_watchdog")
# Device watchdog: checks the radio/rotators/relays every 1h and
# reconnects stuck COM ports (only when the radio isn't
# transmitting and no one holds the lock - so as not to interrupt a QSO).
self._supervise(lambda: self._device_watchdog(), "device_watchdog")
# Event-loop lag detector — DIAGNOSTICS for latency spikes.
# Measures whether asyncio is freezing up. When the loop stalls
# longer than a threshold (something synchronous is blocking it),
# it prints this to the log along with the block duration. Points
# at the culprit behind RTT/audio spikes at low CPU usage.
self._supervise(lambda: self._loop_lag_monitor(), "loop_lag_monitor")
# Scope pump — sends the freshest waterfall frame at a steady
# 15fps FROM THE ASYNCIO LOOP (not from the reader thread). The
# reader only writes _latest_scope; this pump reads and broadcasts
# it. This way the radio's scope stream does NOT synchronize with
# the loop on every frame — which used to block asyncio for 100-500ms.
self._supervise(lambda: self._scope_pump(), "scope_pump")
# Telemetry pump — same decoupling as the scope pump above, applied
# to freq/mode/freqB/split/smeter/txmeter (2026-08-25). These used
# to call civ.bcast() -> broadcast_sync -> run_coroutine_threadsafe
# directly from _reader_loop/_poller_loop (real OS threads) on every
# single change - up to a few dozen cross-thread loop wakeups/sec
# during PTT (4 TX meters every poll cycle), competing with
# WebRTC's own RTP send timing on the same loop. civ.py now only
# writes the freshest message per metric into rig._telemetry_pending;
# this pump drains and broadcasts them at a steady rate instead.
self._supervise(lambda: self._telemetry_pump(), "telemetry_pump")
# Update check: once at startup, in the background, admin-only notice.
# Never blocks startup and stays silent on any network error.
self._supervise(lambda: self._update_check_loop(), "update_check")
# DX Cluster idle watchdog - a cluster connection lives on the
# backend regardless of open browser tabs (by design, so spots
# keep flowing in the background - see ClusterManager.get_history).
# Without a limit that connection lives FOREVER once started - live
# report 2026-10-17 ("ciagle sie rozlancza i laczy") traced to
# exactly this: a second HAMCTRL install (closed browser, backend
# connection never torn down) fighting a fresh login with the SAME
# cluster callsign for the one session DXSpider allows per call.
self._supervise(lambda: self._dxcluster_idle_watchdog(), "dxcluster_idle_watchdog")
asyncio.ensure_future(self._dxcluster_autostart_admins())
# Dziennik ALL.TXT (jak w WSJT-X): wszystko odebrane i nadane w
# FT8/FT4, osobno dla kazdego operatora - patrz ft8_all_log.py.
self._all_log = Ft8AllLog(DATA)
# Porty UDP audio WebRTC otwierane na routerze przez UPnP i odnawiane
# (upnp_ports.py) - bez recznego przekierowania na routerze.
if _WEBRTC and _udp_range and WEBRTC_UPNP:
from upnp_ports import UpnpPortMapper
self._upnp = UpnpPortMapper(*_udp_range)
self._supervise(lambda: self._upnp.refresh_loop(), "upnp")
async def _dxcluster_autostart_admins(self):
"""FIX (2026-10-21, live report: "dx cluster w adminie mial chodzic
caly czas ... rano nie dzialalo dopoki ... nie zalogowalem sie jako
admin"): auto_connect used to fire ONLY when the user logged in
(the WS connect handler), so after every server restart (new EXE
install, reboot) the admin's cluster stayed down until an admin
opened the page. The admin connection is the one meant to run 24/7
(see _dxcluster_idle_watchdog's admin exemption), so start it at
server startup too. Non-admins stay login-triggered on purpose -
their connections get idle-disconnected after 1h anyway."""
if not self.dxcluster:
return
await asyncio.sleep(5) # let the rest of startup settle first
for u in list(self.users):
if u.get("role") != "admin":
continue
dx_cfg = u.get("dxcluster", {}) or {}
if not (dx_cfg.get("auto_connect") and dx_cfg.get("host") and dx_cfg.get("login")):
continue
uid = u.get("id")
existing = self.dxcluster.get_client(uid)
if existing and existing.is_connected():
continue
try:
print(f"[dxcluster] startup auto-connect for admin {u.get('username')!r} "
f"-> {dx_cfg['host']}:{dx_cfg.get('port', 7300)}", flush=True)
await self.dxcluster.connect_user(
uid, dx_cfg["host"], int(dx_cfg.get("port", 7300)),
dx_cfg["login"], decrypt_secret(dx_cfg.get("password", "")))
except Exception as e:
print(f"[dxcluster] startup auto-connect failed for "
f"{u.get('username')!r}: {e}", flush=True)
async def _dxcluster_idle_watchdog(self):
"""
Every CHECK_INTERVAL_S, disconnect any non-admin user's DX Cluster
connection if none of their WS clients has had the DX Cluster or
Radio/WSJT-X tab open (subscribed to the 'dxcluster' or 'scope'
channel - see WSHub) for IDLE_TIMEOUT_S. "Open" is checked fresh
each tick - idle_since only starts counting once a check finds
them not engaged, so this is a full IDLE_TIMEOUT_S of continuous
idleness, not just one bad sample.
Admin is exempt on purpose: their cluster connection is the one
meant to keep collecting spots for the shared Radio-tab waterfall
flags even when nobody has a tab open at all.
"""
if not self.dxcluster:
return
CHECK_INTERVAL_S = 300.0 # 5 min
IDLE_TIMEOUT_S = 3600.0 # 1h
idle_since: dict[str, float] = {}
while True:
try:
await asyncio.sleep(CHECK_INTERVAL_S)
now = time.time()
connected = self.dxcluster.connected_user_ids()
for user_id in connected:
u = self.find_user_by_id(user_id)
if u and u.get("role") == "admin":
idle_since.pop(user_id, None)
continue
engaged = False
for ws, info in list(self.online_users.items()):
if info.get("user_id") != user_id:
continue
subs = self.hub._subs.get(ws, set())
if 'dxcluster' in subs or 'scope' in subs:
engaged = True
break
if engaged:
idle_since.pop(user_id, None)
continue
since = idle_since.setdefault(user_id, now)
idle_s = now - since
if idle_s >= IDLE_TIMEOUT_S:
print(f"[dx-watchdog] user {user_id} idle {int(idle_s)}s "
f"(no DX Cluster/Radio tab open) - disconnecting cluster", flush=True)
await self.dxcluster.disconnect_user(user_id)
idle_since.pop(user_id, None)
# Forget idle timers for users no longer connected at all
for uid in list(idle_since.keys()):
if uid not in connected:
idle_since.pop(uid, None)
except asyncio.CancelledError:
raise
except Exception as e:
print(f"[dx-watchdog] error: {e}", flush=True)
async def _telemetry_pump(self):
"""Reads the freshest freq/mode/freqB/split/smeter/txmeter messages
coalesced by civ.py and sends them at a steady ~20Hz. Decouples
that telemetry (reader/poller threads) from the event loop, same
reasoning as _scope_pump."""
while True:
await asyncio.sleep(0.05)
rig = self.rig
if not rig:
continue
lock = getattr(rig, "_telemetry_lock", None)
if lock is None:
continue
with lock:
pending = rig._telemetry_pending
if not pending:
continue
rig._telemetry_pending = {}
for msg in pending.values():
try:
await self.hub.broadcast(msg)
except Exception:
pass
async def _scope_pump(self):
"""Reads the freshest scope frame from the radio and sends it every
~66ms (15fps). Decouples the scope stream (reader thread) from the event loop."""
while True:
await asyncio.sleep(0.066)
rig = self.rig
if not rig:
continue
frame = getattr(rig, "_latest_scope", None)
if frame is not None:
rig._latest_scope = None # skonsumowane
try:
await self.hub.broadcast(frame, channel="scope")
except Exception:
pass
async def _loop_lag_monitor(self):
"""Uruchamia WATEK-obserwator, ktory wykrywa blokade petli i zrzuca stos
glownego watku W TRAKCIE blokady (nie po). Petla asyncio aktualizuje co
chwile znacznik czasu; watek sprawdza go niezaleznie (ma wlasny rdzen /
wywlaszczenie systemowe) i gdy znacznik sie nie odswieza = petla stoi.
DIAGNOSTIC — off by default. Developer tool; a released club install
should not print [looplag]. Enable only when chasing an event-loop stall
by setting HAM_LOOPLAG=1. We skip the whole watchdog unless turned on."""
import os as _os
if _os.environ.get("HAM_LOOPLAG", "") not in ("1", "true", "TRUE", "yes"):
# Diagnostic disabled. Don't return — the supervisor treats a return
# as a crash and would restart us every few seconds. Just idle here.
while True:
await asyncio.sleep(3600)
import time as _t
import sys as _sys
import traceback as _tb
import threading as _threading
self._loop_heartbeat = _t.monotonic()
_main_id = _threading.main_thread().ident
_state = {"dumped_at": 0.0}
# Startup grace: first seconds bring one-off init blocks (scipy import
# from bundle, launching ham_audio.exe, hamlib, RSA) — harmless, would
# only spam the log. Only steady-state stalls matter.
_start_ts = _t.monotonic()
_grace_s = 15.0
# Innermost frames that mean the loop is IDLE (sleeping in the OS
# selector waiting for events), NOT blocked by our code. A "lag" whose
# deepest frame is one of these is a false alarm — the loop simply had
# nothing to do — so we skip it. Real blocks have our code or a slow
# library (ssl, psutil, subprocess, numpy...) at the bottom of the stack.
_IDLE_MARKERS = ("_poll", "select", "run_forever", "_run_once",
"_loop_self_reading")
def _watchdog():
while True:
_t.sleep(0.02)
if _t.monotonic() - _start_ts < _grace_s:
continue # startup grace — ignore one-off init blocks
_hb = getattr(self, "_loop_heartbeat", 0.0)
_behind = (_t.monotonic() - _hb) * 1000.0
# loop hasn't refreshed heartbeat for >120ms = stalled right now
if _behind > 120 and (_hb - _state["dumped_at"]) > 0.001:
_state["dumped_at"] = _hb
try:
_f = _sys._current_frames().get(_main_id)
if _f is None:
continue
_stack = _tb.format_stack(_f)
# Look at the deepest frame; if it's idle polling, skip.
_deepest = _stack[-1] if _stack else ""
if any(_m in _deepest for _m in _IDLE_MARKERS):
continue # healthy idle wait, not a block
print(f"[looplag] Loop BLOCKED ~{_behind:.0f}ms "
f"(a real block, not idleness):", flush=True)
for _line in _stack[-5:]:
print(" " + _line.strip().replace("\n", " | "),
flush=True)
except Exception as _e:
print(f" (dump failed: {_e})", flush=True)
_th = _threading.Thread(target=_watchdog, daemon=True, name="loop-watchdog")
_th.start()
# Petla odswieza heartbeat tak czesto jak moze — gdy stoi, watek to widzi
while True:
self._loop_heartbeat = _t.monotonic()
await asyncio.sleep(0.02)
async def _update_check_loop(self):
"""Supervised wrapper: run the update check once at startup, then idle.
We must NOT return — the supervisor treats a returning task as a crash
and would restart it every few seconds. So do the one-shot check, then
sleep forever."""
try:
await self._do_update_check()
except Exception:
pass # never let the check crash the supervised task
while True:
await asyncio.sleep(3600)
async def _do_update_check(self):
"""Startup path: let the server finish coming up before touching the
network, then run the actual check (_run_update_check). Separate from
_run_update_check so the admin's manual "check now" button
(/api/update/check) can run the check immediately, without waiting
through this same 8s delay every time."""
self._update_info = None # {latest, current, url} once a newer one is found
await asyncio.sleep(8.0)
await self._run_update_check()
async def _run_update_check(self):
"""Check GitHub Releases once for a newer HAMCTRL version and store a
notice for the admin. Never downloads or installs anything — the admin
fetches the new installer manually. Fully optional and silent:
- Skipped if GITHUB_REPO is unset, or the admin turned off update
checks (cfg 'updateCheck' = False).
- Any network/parse error is swallowed; it must never disrupt startup
or spam the log. This is the product 'phoning home', so it is a
single best-effort request, off the hot path, that fails quietly.
"""
if not GITHUB_REPO:
return # no repo configured yet — nothing to check against
if self.cfg.get("updateCheck", True) is False:
return # admin disabled update checks
api_url = f"https://api.github.com/repos/{GITHUB_REPO}/releases/latest"
def _fetch_latest():
# Runs in a thread — urllib is blocking. Short timeout so a slow or
# unreachable GitHub never holds anything up.
import urllib.request, json as _json
req = urllib.request.Request(api_url, headers={
"Accept": "application/vnd.github+json",
"User-Agent": f"HAMCTRL/{SERVER_VERSION}",
})
with urllib.request.urlopen(req, timeout=8) as r:
data = _json.loads(r.read().decode("utf-8"))
tag = (data.get("tag_name") or "").lstrip("vV")
html = data.get("html_url") or f"https://github.com/{GITHUB_REPO}/releases"
return tag, html
try:
latest, url = await asyncio.to_thread(_fetch_latest)
except Exception:
return # offline, rate-limited, no release yet — stay silent
if latest and self._version_is_newer(latest, SERVER_VERSION):
self._update_info = {"latest": latest, "current": SERVER_VERSION,
"url": url}
print(f"[update] Newer version available: {latest} "
f"(you have {SERVER_VERSION}) — {url}", flush=True)
# Email the admin — but only ONCE per version. The check runs on
# every startup, so without this guard a restart would re-send the
# same notice. We remember the last version we emailed about in cfg;
# a new version resets it. Opt-in: only if updateEmail is enabled and
# SMTP is configured.
already = self.cfg.get("updateEmailedVersion", "")
# FIX (reported live 2026-08-26: "widze ze jest nowa wersja ale
# zaden mail nie przyszedl" — an update WAS detected, but the
# admin had zero visibility into WHY the email silently never
# went out). self._update_email_diag now records exactly what
# happened on the LAST attempt (or non-attempt), surfaced via
# /api/update/config -> AdminUpdate._render, instead of only a
# server-console print nobody watching the UI ever sees.
if not self.cfg.get("updateEmail", False):
self._update_email_diag = ("Powiadomienie mailem jest wylaczone "
"(wlacz przelacznik powyzej).")
elif already == latest:
self._update_email_diag = f"Mail o wersji {latest} zostal juz wyslany wczesniej."
else:
sent, diag = await self._email_admins_update(latest, url)
self._update_email_diag = diag
if sent:
self.cfg["updateEmailedVersion"] = latest
save_json(CFG_F, self.cfg)
async def _email_admins_update(self, latest: str, url: str) -> tuple:
"""Email every admin who has an address about a new version.
Returns (any_sent: bool, diagnostic: str) — the diagnostic explains
exactly what happened (no admin has an email set, SMTP error(s), or
who it was actually sent to), so the admin panel can show WHY an
expected email never arrived instead of silence. Best-effort:
exceptions are caught and folded into the diagnostic, never raised."""
# Language: the server-wide install-time language (self.cfg["lang"],
# seeded from the installer's language choice - see data.py's
# get_cfg and /api/config/lang), NOT a per-admin preference (there
# is no such stored setting - language in this app is a per-browser
# localStorage choice, invisible to the server). Requested live
# 2026-08-26: "musi byc po eng lub po polsku w zaleznosci od wersji
# HAMCTRL" - a PL-installed HAMCTRL should email in Polish, an
# EN-installed one in English.
if self.cfg.get("lang", "pl") == "en":
subject = f"HAMCTRL — new version {latest} available"
body = (
f"A new version of HAMCTRL is available: {latest}\n"
f"Your current version: {SERVER_VERSION}\n\n"
f"Download the new installer from the release page:\n {url}\n\n"
f"The installer will update your existing installation, keeping "
f"your data (accounts, configuration, QSO log).\n\n"
f"You can turn this notification off in the admin settings.\n\n"
f"73 de HAMCTRL\n"
)
else:
subject = f"HAMCTRL — dostepna nowa wersja {latest}"
body = (
f"Dostepna jest nowa wersja HAMCTRL: {latest}\n"
f"Twoja obecna wersja: {SERVER_VERSION}\n\n"
f"Pobierz nowy instalator ze strony wydania:\n {url}\n\n"
f"Instalator zaktualizuje istniejaca instalacje, zachowujac dane "
f"(konta, konfiguracje, dziennik QSO).\n\n"
f"To powiadomienie mozna wylaczyc w ustawieniach administratora.\n\n"
f"73 de HAMCTRL\n"
)
any_sent = False
admins_with_email = []
sent_to = []
errors = []
for u in self.users:
if u.get("role") != "admin":
continue
email_addr = u.get("email", "")
if not email_addr:
continue
admins_with_email.append(email_addr)
try:
ok, err = await self._send_email(email_addr, subject, body)
if ok:
any_sent = True
sent_to.append(email_addr)
else:
errors.append(f"{email_addr}: {err}")
except Exception as e:
errors.append(f"{email_addr}: {e}")
if not admins_with_email:
diag = ("Zaden administrator nie ma ustawionego adresu e-mail "
"(PROFIL -> E-MAIL -> ZAPISZ PROFIL).")
elif sent_to and not errors:
diag = f"Wyslano do: {', '.join(sent_to)}."
elif sent_to and errors:
diag = f"Wyslano do: {', '.join(sent_to)}. Bledy: {'; '.join(errors)}."
else:
diag = f"Nie wyslano do nikogo. Bledy: {'; '.join(errors)}."
return any_sent, diag
@staticmethod
def _version_is_newer(a: str, b: str) -> bool:
"""True if version string `a` is newer than `b`. Compares dotted numeric
parts (1.10 > 1.9); a non-numeric suffix like '-dev' sorts as older so a
clean release always beats a dev build of the same number."""
def _parts(v):
v = v.strip().lstrip("vV")
dev = 1
if "-" in v:
v, _suf = v.split("-", 1)
dev = 0 # '1.0-dev' < '1.0'
nums = []
for p in v.split("."):
try:
nums.append(int(p))
except ValueError:
nums.append(0)
return nums, dev
pa, da = _parts(a)
pb, db = _parts(b)
# pad to equal length
n = max(len(pa), len(pb))
pa += [0] * (n - len(pa))
pb += [0] * (n - len(pb))
if pa != pb:
return pa > pb
return da > db
def find_user(self, username: str) -> dict | None:
return next((u for u in self.users if u["username"].lower() == username.lower()), None)
def find_user_by_id(self, uid: str) -> dict | None:
return next((u for u in self.users if u["id"] == uid), None)
def _migrate_plaintext_secrets(self):
"""One-time startup pass: encrypt any CloudLog/QRZ/HamQTH/SMTP
credentials still stored as plaintext from before encryption at
rest was added (see crypto_secrets.py). Idempotent - encrypt_secret()
is a no-op on values already carrying the enc1: prefix, so this is
safe to run on every startup.
SMTP was found live 2026-08-26 (while diagnosing a silent
update-notification email failure) to have been missed entirely by
this pass since it was written - the password lives in self.cfg
(config.json), not per-user in self.users like the others, so it
needs its own separate encrypt-and-save step."""
smtp_pw = self.cfg.get("smtp", {}).get("password", "")
if smtp_pw:
enc = encrypt_secret(smtp_pw)
if enc != smtp_pw:
self.cfg["smtp"]["password"] = enc
save_json(CFG_F, self.cfg)
print("[secrets] encrypted SMTP password in config.json", flush=True)
changed = False
for u in self.users:
cl = u.get("cloudlog")
if cl:
for key in ("apiKeyQso", "apiKeyRadio"):
val = cl.get(key, "")
if val:
enc = encrypt_secret(val)
if enc != val:
cl[key] = enc
changed = True
cb = u.get("callbook")
if cb:
for key in ("qrzPassword", "hamqthPassword"):
val = cb.get(key, "")
if val:
enc = encrypt_secret(val)
if enc != val:
cb[key] = enc
changed = True
dx = u.get("dxcluster")
if dx:
val = dx.get("password", "")
if val:
enc = encrypt_secret(val)
if enc != val:
dx["password"] = enc
changed = True
if changed:
save_json(USR_F, self.users)
print("[secrets] encrypted CloudLog/QRZ/HamQTH/DX-Cluster credentials in users.json", flush=True)
# Mirrors admin.js's DEFAULT_PERMS (the set a NEW user gets, and what
# the admin-panel checkboxes fall back to for a user object that has
# no "permissions" dict at all) - used below ONLY for a key that is
# entirely ABSENT from a user's stored permissions, never for a key
# explicitly set to false.
_DEFAULT_PERMS = {
"ptt": True, "rfPower": True, "rfGain": True, "mode": True,
"band": True, "freq": True, "cw": True,
"rotator": True, "log": True, "settings": False, "admin": False,
}
def _has_perm(self, uid: str, role: str, key: str) -> bool:
"""Admin always has access. Otherwise check the granular permission
(permissions[key] in users.json, set from the user-edit form).
NOTE: the JWT only carries id/role/username/pw_ver (see jwt_sign in
/api/auth/login) - it does NOT carry permissions, so the full,
current user record must be re-read from self.users instead of
relying on the decoded token (which could be stale after an admin
changes permissions without the user logging in again).
FIX (2026-09-03, found live right after the granular-permission
audit shipped: "zwykly user nie widzi buttona splitu"): a key that
is entirely ABSENT from a user's stored permissions dict used to
be treated identically to one explicitly set to False - denied.
That's correct for a key that was ALREADY being enforced (an
admin who wanted it denied would have unchecked it, which DOES
save an explicit False), but wrong for a key like "split" that
this session just started enforcing for the first time: it had
no real effect before today (the SPLIT button was gated by
"freq" instead), so most existing accounts never got an explicit
permissions.split value saved at all - "never set" here means
"nobody ever made a decision about this", not "denied". Now: a
key present in the dict (True OR False) is trusted as-is: an
absent key falls back to _DEFAULT_PERMS, the same default a
brand-new user gets (matches admin.js's own `u?.permissions ||
DEFAULT_PERMS` fallback intent, just applied per-key instead of
only when the whole dict is missing)."""
if role == "admin":
return True
u = self.find_user_by_id(uid)
perms = (u or {}).get("permissions", {})
if key in perms:
return bool(perms[key])
# Fallback only covers the keys DEFAULT_PERMS actually defines
# (the ones with this migration concern) - an unlisted key (e.g.
# relay_N, which correctly defaults to False/not-granted unless
# explicitly given - see /api/relay/action's own check) stays
# denied by default, not silently opened.
return self._DEFAULT_PERMS.get(key, False)
def find_user_by_email(self, email: str) -> dict | None:
return next((u for u in self.users
if (u.get("email") or "").lower() == email.lower()), None)
# ── Radio Lock helpers ────────────────────────────────────────────────────
def _radio_lock_state(self) -> dict:
"""Serializable radio-lock state (for WS broadcast and the API)."""
return {
"type": "radio_lock_state",
"locked": self.radio_lock["user_id"] is not None,
"user_id": self.radio_lock["user_id"],
"username": self.radio_lock["username"],
"callsign": self.radio_lock["callsign"],
"locked_at": self.radio_lock["locked_at"],
"timeout_min": self.radio_lock["timeout_min"],
"requests": [
{"user_id": uid,
"username": r["username"],
"callsign": r["callsign"],
"requested_at": r["requested_at"]}
for uid, r in self.radio_requests.items()
],
}
def _online_users_state(self) -> list:
"""List of currently connected users."""
return [
{"user_id": v["user_id"],
"username": v["username"],
"callsign": v["callsign"],
"role": v["role"],
"has_lock": v["user_id"] == self.radio_lock["user_id"]}
for v in self.online_users.values()
]
def _user_has_lock(self, user_id: str) -> bool:
return self.radio_lock["user_id"] == user_id
def _lock_radio(self, user: dict):
"""Take over the radio for a user."""
self.radio_lock.update({
"user_id": user["id"],
"username": user["username"],
"callsign": user.get("callsign", user["username"]),
"locked_at": time.time(),
"last_activity": time.time(),
})
# Remove this user's request from the queue (if there was one)
self.radio_requests.pop(user["id"], None)
def _release_radio(self):
"""Release the radio."""
# Whoever is releasing/losing the radio - also stop WSJT-X if they
# owned it. Called WITHOUT await because _release_radio is sync
# (also used from the watchdog). The actual stop happens asynchronously.
prev_owner = self.radio_lock.get("user_id")
if prev_owner and prev_owner == self._ft8_rx_owner_uid and self._ft8_rx_enabled:
asyncio.ensure_future(self._stop_wsjtx_auto("oddanie radia"))
self.radio_lock.update({
"user_id": None, "username": None,
"callsign": None, "locked_at": None, "last_activity": None,
})
async def _stop_wsjtx_auto(self, reason: str):
"""Automatically stop WSJT-X for the given reason.
Called when:
- The WSJT-X owner disconnected (WS close)
- The owner released the radio (manually or via the idle watchdog)
- The owner lost the radio via an admin FORCE
- The owner logged out
Ensures decoding doesn't keep running "in the background" when no
one is controlling the radio, and doesn't block the audio TX
pipeline for other users."""
if not self._ft8_rx_enabled:
return
print(f"[ft8rx] AUTO-STOP: {reason} (owner={self._ft8_rx_owner_uid})", flush=True)
self._ft8_rx_enabled = False
self._ft8_rx_owner_uid = None
if self.rust_audio:
try:
await self.rust_audio.ft8_enable_rx(False, self._ft8_decode_mode)
except Exception as e:
print(f"[ft8rx] auto-stop rust_audio error: {e}")
# Broadcast so every UI knows WSJT-X has been stopped
await self.hub.broadcast({"type": "ft8_rx_status", "enabled": False})
await self.hub.broadcast({"type": "wsjtx_status", "running": False,
"decoding": False, "transmit": False})
# Informational toast (does it go on the 'ft8' channel
# automatically? No - this is control-level info for everyone)
await self.hub.broadcast({"type": "toast",
"msg": f"🛑 WSJT-X zatrzymany automatycznie ({reason})",
"level": "warn"})
def touch_activity(self, user_id: str):
"""Update the active operator's last-activity timestamp."""
# DIAGNOSTIC (2026-10-21, live report: "ja klikalem ... to nie bylo
# zanotowane przez serwer" - the operator insists they clicked to
# select a station, but still got auto-released as idle) - logs
# every call so we can see whether this ever gets called at all for
# that click (a frontend path that never sends a WS message would
# mean nothing reaches here), and if it does, whether
# _user_has_lock actually matched (a stale/mismatched user_id would
# silently no-op here without this).
_held_by = self.radio_lock.get("user_id")
if self._user_has_lock(user_id):
self.radio_lock["last_activity"] = time.time()
print(f"[radio_lock] touch_activity: OK for uid={user_id!r} "
f"(held_by={_held_by!r})")
else:
print(f"[radio_lock] touch_activity: NO-OP for uid={user_id!r} "
f"- lock held by {_held_by!r} instead")
# NOTE: _start_tx_watchdog used to be DEFINED TWICE in this same class
# (Python takes the second one - see the definition near
# _feature_allowed below) - the first one, the only one that
# broadcast tx_watchdog_start/countdown/cancel, was therefore
# COMPLETELY dead, unreachable code. The frontend never listened for
# these message types anyway (the #wj-tx-watchdog element in
# index.html was equally dead), so it was removed with no loss of
# functionality - the real PTT watchdog is the second definition,
# works independently, and is left unchanged.
async def _start_tune(self, duration_s: float, tone_hz: float = 1500.0):
"""Transmit a steady tone for duration_s seconds, for ATU tuning.
Generates a PCM sine wave at tone_hz (default 1500Hz - the middle
of the SSB band), 16-bit, mono, 12000 Hz sample rate. Sent through
the existing audio TX pipeline. PTT is turned on directly via
CI-V, turned off when finished. Actively checks self._tune_stop.
"""
import math
try:
self._tune_stop = False
print(f"[tune] START {tone_hz}Hz for {duration_s}s")
await self.hub.broadcast({"type": "tune_status", "active": True,
"duration": duration_s, "tone": tone_hz})
# Turn on PTT
self.rig.ptt = True
if not self.rig.sim:
try: await self.rig.set_ptt(True)
except Exception as e: print(f"[tune] set_ptt error: {e}")
await self.hub.broadcast({"type": "ptt", "ptt": True})
# Generate the tone PCM and feed it through the audio stream
sample_rate = 12000
chunk_size = 4800 # 400ms per chunk
total_samples = int(duration_s * sample_rate)
samples_sent = 0
phase = 0.0
two_pi_over_sr = 2.0 * math.pi * tone_hz / sample_rate
vol = min(float(self.cfg.get("audio", {}).get("txVolume", 1.0)), 8.0)
amp_scale = 0.6 * vol
# bulk_tx=True here (mirrors the FT8/FT4 TX sequence) keeps this
# synthesized tuning tone on the flat-multiplier path in
# _webrtc_playback_loop instead of the SSB mic AGC path (added
# 2026-09-04, see txGainModeSsb) - AGC is for the live
# microphone signal only, a constant-amplitude tone has nothing
# for it to normalize and would just override the intended
# amp_scale above with its own target level.
self.audio.bulk_tx = True
while samples_sent < total_samples and not self._tune_stop:
chunk = bytearray()
for _ in range(chunk_size):
if samples_sent >= total_samples: break
v = int(math.sin(phase) * 32767 * amp_scale)
v = max(-32767, min(32767, v))
chunk += v.to_bytes(2, 'little', signed=True)
phase += two_pi_over_sr
if phase > math.pi * 2: phase -= math.pi * 2
samples_sent += 1
if hasattr(self.audio, 'feed_tx_pcm'):
try: self.audio.feed_tx_pcm(bytes(chunk))
except Exception as e: print(f"[tune] feed error: {e}")
await asyncio.sleep(chunk_size / sample_rate * 0.9)
# Turn off PTT
self.rig.ptt = False
if not self.rig.sim:
try: await self.rig.set_ptt(False)
except Exception as e: print(f"[tune] set_ptt(False) error: {e}")
await self.hub.broadcast({"type": "ptt", "ptt": False})
await self.hub.broadcast({"type": "tune_status", "active": False})
print(f"[tune] STOP (stop_flag={self._tune_stop})")
except Exception as e:
print(f"[tune] error: {e}")
self.rig.ptt = False
if not self.rig.sim:
try: await self.rig.set_ptt(False)
except: pass
await self.hub.broadcast({"type": "ptt", "ptt": False})
await self.hub.broadcast({"type": "tune_status", "active": False})
finally:
self.audio.bulk_tx = False # restore anti-lag / SSB-mic path for whatever transmits next
async def _relay_connect_task(self):
"""Connect to the Arduino relay controller in the background."""
if self.relay:
await self.relay.connect()
async def _relay_reconnect_task(self):
"""Disconnect and reconnect with the new configuration."""
if self.relay:
await self.relay.disconnect()
self.relay = None
rcfg = self.cfg.get("relay", {})
if _RELAY_OK and rcfg.get("enabled") and rcfg.get("port"):
try:
self.relay = RelayController(
port=rcfg["port"],
baudrate=int(rcfg.get("baudrate", 9600)),
)
await self.relay.connect()
except Exception as e:
print(f"[relay] reconnect error: {e}")
def _supervise(self, coro_factory, name: str, restart_delay: float = 5.0):
"""
Run a background loop under supervision: if, despite internal
try/except blocks, the task crashes (an unexpected exception, a
bug), the supervisor RESTARTS it after restart_delay seconds.
Without this, a single unexpected error kills the feature (e.g.
the device watchdog or FT8 decoding) until the server is manually restarted.
coro_factory: a zero-argument function that returns a coroutine
(not a coroutine itself!), e.g. lambda: self._device_watchdog()
"""
async def _runner():
_fails = 0
while True:
try:
await coro_factory()
# The loop ended normally (it shouldn't) - restart it
print(f"[supervisor] '{name}' ended on its own - restarting "
f"in {restart_delay}s", flush=True)
except asyncio.CancelledError:
print(f"[supervisor] '{name}' cancelled (shutting down)", flush=True)
raise
except Exception as e:
_fails += 1
print(f"[supervisor] '{name}' CRASHED ({_fails}x): "
f"{type(e).__name__}: {e}", flush=True)
import traceback as _tb
_tb.print_exc()
# Growing backoff on repeated failures (5s, 10s, 20s,
# ... max 300s) to avoid a crash loop
_delay = min(restart_delay * (2 ** min(_fails - 1, 6)), 300.0)
print(f"[supervisor] '{name}' restarting in {_delay:.0f}s",
flush=True)
await asyncio.sleep(_delay)
continue
await asyncio.sleep(restart_delay)
return asyncio.ensure_future(_runner())
async def _device_watchdog(self):
"""
Every 1h checks whether devices (radio, rotators, relays) respond
on their COM port. If a port is stuck / a device doesn't respond -
disconnects and reconnects (soft reset).
SAFETY: doesn't touch anything when:
- the radio is transmitting (PTT active) - we don't interrupt a transmission
- someone holds radio_lock - we don't interrupt someone else's QSO
In that case the check is deferred to the next cycle.
The radio already has its own _reconnect_loop in civ.py (auto-
recovers from SIM once the port frees up), so for the radio we
only log its state and try to wake it up if it's in SIM despite an
available port.
"""
INTERVAL_S = 3600.0 # 1h
# First check 5 min after startup (give time for initialization)
await asyncio.sleep(300.0)
while True:
try:
# ── Safety conditions ────────────────────────────────────
ptt_active = bool(getattr(self.rig, "ptt", False))
lock_held = self.radio_lock.get("user_id") is not None
if ptt_active or lock_held:
reason = "PTT active" if ptt_active else "operator holding the radio"
print(f"[watchdog] skipping device check ({reason})",
flush=True)
await asyncio.sleep(INTERVAL_S)
continue
print("[watchdog] checking devices...", flush=True)
await self._watchdog_check_radio()
await self._watchdog_check_rotators()
await self._watchdog_check_relay()
except asyncio.CancelledError:
raise
except Exception as e:
print(f"[watchdog] general error: {e}", flush=True)
await asyncio.sleep(INTERVAL_S)
async def _watchdog_check_radio(self):
"""
Radio: check whether CI-V responds. The radio has its own
reconnect_loop in civ.py so here we only do diagnostics + force a
retry if it's in SIM.
"""
try:
is_sim = bool(getattr(self.rig, "sim", False))
has_port = getattr(self.rig, "_ser", None) is not None
if is_sim:
print("[watchdog] radio in SIMULATION - civ reconnect_loop "
"will try to recover the port automatically", flush=True)
return
if not has_port:
print("[watchdog] radio: no open port", flush=True)
return
# Health check: ask for the frequency (command 0x03).
# If the radio doesn't respond, get_freq returns None/raises.
ok = False
try:
if hasattr(self.rig, "get_freq"):
# get_freq is async (CivRig) - called directly
f = await asyncio.wait_for(self.rig.get_freq(), timeout=5.0)
ok = f is not None and f > 0
except Exception:
ok = False
if ok:
print("[watchdog] radio OK (responds on CI-V)", flush=True)
else:
print("[watchdog] radio NOT RESPONDING - trying reconnect",
flush=True)
# Close the port - civ's _reconnect_loop will reopen it
try:
if hasattr(self.rig, "_ser") and self.rig._ser:
self.rig._ser.close()
self.rig._ser = None
if hasattr(self.rig, "_start_sim"):
self.rig._start_sim() # this kicks off _reconnect_loop
print("[watchdog] radio: port closed, reconnect_loop "
"will try to recover", flush=True)
except Exception as e:
print(f"[watchdog] radio reconnect error: {e}", flush=True)
except Exception as e:
print(f"[watchdog] radio check error: {e}", flush=True)
async def _watchdog_check_rotators(self):
"""Rotators: check whether they respond to STATUS, reconnect if not."""
rotators = getattr(self, "rotators", None)
if not rotators:
return
for rot in rotators:
try:
if getattr(rot, "sim", False):
# In simulation - try connecting again (the port may have come back)
print(f"[watchdog] rotator {rot.name}: in simulation, "
f"trying to connect", flush=True)
await asyncio.to_thread(rot.connect)
continue
if not getattr(rot, "connected", False):
print(f"[watchdog] rotator {rot.name}: disconnected, "
f"trying to connect", flush=True)
await asyncio.to_thread(rot.connect)
continue
# Health check: read the position (a real query to the port)
pos = None
try:
pos = await asyncio.wait_for(
asyncio.to_thread(rot._read_pos, 3.0), timeout=5.0)
except Exception:
pos = None
if pos is not None:
print(f"[watchdog] rotator {rot.name} OK (az={pos:.0f}°)",
flush=True)
else:
print(f"[watchdog] rotator {rot.name} NOT RESPONDING - "
f"reconnecting", flush=True)
try:
await asyncio.to_thread(rot.close)
except Exception:
pass
await asyncio.sleep(1.0)
ok = await asyncio.to_thread(rot.connect)
print(f"[watchdog] rotator {rot.name} reconnect: "
f"{'OK' if ok else 'FAILED (simulation)'}", flush=True)
except Exception as e:
print(f"[watchdog] rotator {getattr(rot,'name','?')} error: {e}",
flush=True)
async def _watchdog_check_relay(self):
"""Relays: check whether the Arduino responds to RPK, reconnect if not."""
rcfg = self.cfg.get("relay", {})
if not rcfg.get("enabled") or not rcfg.get("port"):
return
relay = getattr(self, "relay", None)
if not relay:
print("[watchdog] relay: no instance, trying to connect", flush=True)
await self._relay_reconnect_task()
return
try:
if not relay.is_connected():
print("[watchdog] relay: disconnected, reconnecting", flush=True)
await self._relay_reconnect_task()
return
# Health check: read the states (a real RPK query to the Arduino)
ok = False
try:
states = await asyncio.wait_for(
relay.read_all_states(), timeout=5.0)
ok = states is not None
except Exception:
ok = False
if ok:
print("[watchdog] relay OK (Arduino responds)", flush=True)
else:
print("[watchdog] relay NOT RESPONDING - reconnecting", flush=True)
await self._relay_reconnect_task()
if self.relay and self.relay.is_connected():
print("[watchdog] relay reconnect OK", flush=True)
else:
print("[watchdog] relay reconnect FAILED", flush=True)
except Exception as e:
print(f"[watchdog] relay check error: {e}", flush=True)
async def _radio_lock_watchdog(self):
"""
Every 30s checks whether the active operator has exceeded the idle
timeout. If so — automatically releases the radio and broadcasts the state.
"""
await asyncio.sleep(5) # give it a moment to initialize
while True:
await asyncio.sleep(30)
try:
uid = self.radio_lock["user_id"]
if not uid:
continue
last = self.radio_lock["last_activity"] or self.radio_lock["locked_at"] or 0
timeout_s = self.radio_lock["timeout_min"] * 60
if time.time() - last > timeout_s:
uname = self.radio_lock["username"]
self._release_radio()
await self.hub.broadcast({
"type": "radio_lock_state",
**self._radio_lock_state(),
})
await self.hub.broadcast({
"type": "toast",
"message": f"Radio zwolnione automatycznie ({uname} — przekroczono czas bezczynnosci {self.radio_lock['timeout_min']} min)",
})
print(f"[radio_lock] Auto-release: {uname} exceeded the {self.radio_lock['timeout_min']} min timeout")
except Exception as e:
print(f"[radio_lock] watchdog error: {e}")
# ── SMTP — password reset ────────────────────────────────────────────────────
async def _send_email(self, to_email: str, subject: str, body: str):
"""Generic SMTP send (subject + plain-text body) using the same SMTP
config as password reset. Returns (True, None) or (False, error).
Used for admin update notices as well as password resets."""
import smtplib, email.message
smtp_cfg = self.cfg.get("smtp", {})
if not smtp_cfg.get("host") or not smtp_cfg.get("user") or not smtp_cfg.get("password"):
return False, "SMTP niekompletny (host/user/haslo)"
msg = email.message.EmailMessage()
msg["Subject"] = subject
msg["From"] = smtp_cfg.get("from") or smtp_cfg["user"]
msg["To"] = to_email
msg.set_content(body)
host = smtp_cfg["host"]; port = int(smtp_cfg.get("port", 587))
use_tls = smtp_cfg.get("use_tls", True)
user = smtp_cfg["user"]; password = decrypt_secret(smtp_cfg["password"])
def _send():
if use_tls:
with smtplib.SMTP(host, port, timeout=15) as s:
s.ehlo(); s.starttls(); s.ehlo(); s.login(user, password)
s.send_message(msg)
else:
with smtplib.SMTP_SSL(host, port, timeout=15) as s:
s.login(user, password); s.send_message(msg)
try:
await asyncio.to_thread(_send)
print(f"[smtp] Email sent to {to_email}: {subject}")
return True, None
except Exception as e:
print(f"[smtp] Send error to {to_email}: {e}")
return False, str(e)
async def _send_reset_email(self, to_email: str, username: str, token: str, base_url: str):
"""
Send an email via SMTP. Returns (True, None) or (False, error_str).
"""
import smtplib, email.message
smtp_cfg = self.cfg.get("smtp", {})
if not smtp_cfg.get("host"):
return False, "Brak hosta SMTP w konfiguracji"
if not smtp_cfg.get("user"):
return False, "Brak uzytkownika SMTP w konfiguracji"
if not smtp_cfg.get("password"):
return False, "Brak hasla SMTP — zapisz konfiguracje SMTP z haslem"
reset_url = f"{base_url.rstrip('/')}/reset.html?token={token}"
msg = email.message.EmailMessage()
msg["Subject"] = "Ham Radio Control — reset hasla"
msg["From"] = smtp_cfg.get("from") or smtp_cfg["user"]
msg["To"] = to_email
msg.set_content(
f"Czesc {username},\n\n"
f"Prosba o reset hasla dla konta Ham Radio Control Server.\n\n"
f"Kliknij ponizszy link (wazny przez 1 godzine):\n\n"
f" {reset_url}\n\n"
f"Jesli nie prosiles o reset hasla — zignoruj ten email.\n\n"
f"73 de Ham Radio Control Server\n"
)
host = smtp_cfg["host"]
port = int(smtp_cfg.get("port", 587))
use_tls = smtp_cfg.get("use_tls", True)
user = smtp_cfg["user"]
password = decrypt_secret(smtp_cfg["password"])
def _send():
if use_tls:
with smtplib.SMTP(host, port, timeout=15) as s:
s.ehlo()
s.starttls()
s.ehlo()
s.login(user, password)
s.send_message(msg)
else:
with smtplib.SMTP_SSL(host, port, timeout=15) as s:
s.login(user, password)
s.send_message(msg)
try:
await asyncio.to_thread(_send)
print(f"[smtp] Email sent to {to_email}")
return True, None
except smtplib.SMTPAuthenticationError as e:
err = f"Blad autoryzacji SMTP ({e.smtp_code}): nieprawidlowy login lub haslo. Dla Gmail uzyj 'App Password'."
print(f"[smtp] {err}")
return False, err
except smtplib.SMTPConnectError as e:
err = f"Nie mozna polaczyc z {host}:{port} — sprawdz host i port."
print(f"[smtp] {err}: {e}")
return False, err
except smtplib.SMTPException as e:
err = f"Blad SMTP: {e}"
print(f"[smtp] {err}")
return False, err
except TimeoutError:
err = f"Timeout polaczenia z {host}:{port} — sprawdz host/port/firewall."
print(f"[smtp] {err}")
return False, err
except Exception as e:
err = f"Blad wysylania: {type(e).__name__}: {e}"
print(f"[smtp] {err}")
return False, err
def get_rot(self, rid: int) -> Rotator | None:
return next((r for r in self.rotators if r.id == rid), None)
def init_rotators(self):
for r in self.rotators: r.close()
self.rotators = []
for rc in self.cfg.get("rotators", []):
if not rc.get("enabled", True): continue
rot = Rotator(rc, self.hub.broadcast_sync)
rot.connect()
self.rotators.append(rot)
n = len(self.rotators)
print(f"[rotator] {n} active" if n else "[rotator] none configured")
def _check_pw_ver(self, payload: dict | None) -> dict | None:
"""
Check whether the token has the current password version (pw_ver).
A password change increments pw_ver in users.json, so old tokens
(with a lower version) stop working - this invalidates sessions on
all devices.
Returns the payload if OK, or None if the token is stale/invalid.
Tokens issued before pw_ver was introduced (missing field) are
accepted only when the user also has no pw_ver (i.e. never changed
their password).
"""
if not payload:
return None
uid = payload.get("id") or payload.get("user_id")
if not uid:
return None
u = next((x for x in self.users if x.get("id") == uid), None)
if not u:
return None # user deleted
token_ver = int(payload.get("pw_ver", 0))
user_ver = int(u.get("pw_ver", 0))
if token_ver != user_ver:
print(f"[auth] token rejected: pw_ver={token_ver} != {user_ver} "
f"(user {u.get('username')!r} changed their password)", flush=True)
return None
return payload
# ── Login rate limiting (brute-force protection) ──────────────────
# The server is on the internet (duckdns) - bots scan and try to crack
# passwords. We limit login attempts per IP and per username.
#
# Rules:
# - max 5 failed attempts from one IP within 5 minutes -> 15 min block
# - max 5 failed attempts for one username within 5 min -> 15 min block
# (protects a specific account even when the attack comes from many IPs)
# - a successful login clears the counter for that IP and user
# Data is kept in memory (reset on restart) - enough for a club server,
# no need for Redis.
_LOGIN_MAX_FAILS = 5 # how many failed attempts before a block
_LOGIN_WINDOW_S = 300.0 # attempt-counting window (5 min)
_LOGIN_BLOCK_S = 900.0 # how long the block lasts (15 min)
def _rl_key_state(self, store: dict, key: str) -> dict:
st = store.get(key)
if st is None:
st = {"fails": [], "blocked_until": 0.0}
store[key] = st
return st
def _login_is_blocked(self, ip: str, username: str) -> tuple[bool, int]:
"""Is login blocked? Returns (blocked, seconds_left)."""
now = time.time()
for store, key in ((self._login_fails_ip, ip),
(self._login_fails_user, username)):
if not key:
continue
st = store.get(key)
if st and st["blocked_until"] > now:
return True, int(st["blocked_until"] - now)
return False, 0
def _login_record_fail(self, ip: str, username: str):
"""Record a failed attempt. If the limit is exceeded - block."""
now = time.time()
for store, key, label in ((self._login_fails_ip, ip, "IP"),
(self._login_fails_user, username, "user")):
if not key:
continue
st = self._rl_key_state(store, key)
# Remove old attempts outside the window
st["fails"] = [t for t in st["fails"] if now - t < self._LOGIN_WINDOW_S]
st["fails"].append(now)
if len(st["fails"]) >= self._LOGIN_MAX_FAILS:
st["blocked_until"] = now + self._LOGIN_BLOCK_S
st["fails"] = []
print(f"[auth] BLOCKING {label}={key!r} for "
f"{int(self._LOGIN_BLOCK_S/60)} min "
f"(too many failed login attempts)", flush=True)
def _login_record_success(self, ip: str, username: str):
"""Successful login - clear the counters."""
self._login_fails_ip.pop(ip, None)
self._login_fails_user.pop(username, None)
def _login_cleanup(self):
"""
Remove old entries so the dicts don't grow without bound.
STABILITY: under attack (millions of different usernames), just
removing "empty" entries isn't enough - all of them have fresh
attempts. So we additionally remove entries whose attempt window
has already passed, and if the dict is still too big - hard-trim
it to the most recent ones.
"""
now = time.time()
HARD_CAP = 500
for store in (self._login_fails_ip, self._login_fails_user):
# 1. Remove entries with no active block and no attempts in the window
dead = []
for k, st in store.items():
st["fails"] = [t for t in st["fails"]
if now - t < self._LOGIN_WINDOW_S]
if st["blocked_until"] < now and not st["fails"]:
dead.append(k)
for k in dead:
store.pop(k, None)
# 2. Hard limit - if still too many (an attack), keep only
# active blocks + the most recent attempts
if len(store) > HARD_CAP:
blocked = {k: st for k, st in store.items()
if st["blocked_until"] > now}
rest = [(k, st) for k, st in store.items()
if st["blocked_until"] <= now]
# Most recent attempts at the end of the fails list
rest.sort(key=lambda kv: (kv[1]["fails"][-1]
if kv[1]["fails"] else 0),
reverse=True)
keep = dict(rest[:max(0, HARD_CAP - len(blocked))])
keep.update(blocked) # blocks always stay
store.clear()
store.update(keep)
print(f"[auth] cleanup: trimmed to {len(store)} entries "
f"(possible brute-force attack)", flush=True)
# ── API ───────────────────────────────────────────────────────────────────
async def api(self, method: str, path: str, body: dict,
user: dict | None, query: dict, client_ip: str = "") -> tuple[int, any]:
p = path
if p == "/api/auth/login" and method == "POST":
un = body.get("username", "").lower()
pw = body.get("password", "")
# Rate limiting - check whether the IP/user is blocked
blocked, secs_left = self._login_is_blocked(client_ip, un)
if blocked:
mins = max(1, secs_left // 60)
print(f"[auth] login rejected (blocked) ip={client_ip} user={un!r}",
flush=True)
return 429, {"error": f"Za duzo nieudanych prob. "
f"Sprobuj ponownie za {mins} min."}
u = self.find_user(un)
if not u or not u.get("active") or not verify_pw(pw, u.get("password", "")):
self._login_record_fail(client_ip, un)
# Sporadic cleanup of old entries. We check BOTH dicts -
# an attacker could send millions of different usernames to
# exhaust memory (_login_fails_user would grow unbounded).
if (len(self._login_fails_ip) > 200
or len(self._login_fails_user) > 200):
self._login_cleanup()
print(f"[auth] failed login ip={client_ip} user={un!r}", flush=True)
return 401, {"error": "Błędne dane logowania"}
self._login_record_success(client_ip, un)
# Transparent upgrade: if this account still has a legacy SHA-256
# hash, now that we have the correct plaintext, re-hash it with
# scrypt and persist. Users migrate silently on their next login.
if needs_rehash(u.get("password", "")):
u["password"] = hash_pw_secure(pw)
try:
save_json(USR_F, self.users)
except Exception:
pass
# pw_ver = password version. A password change increments it in
# users.json, invalidating all old tokens (they no longer match the new version).
token = jwt_sign({"id": u["id"], "role": u["role"], "username": u["username"],
"pw_ver": int(u.get("pw_ver", 0))})
# Force a change of the default password: when an admin logs in
# still using the working default password (Admin1234!), the
# frontend shows a password-change screen. CRITICAL for the
# product - every club starts with the same default password,
# so it MUST be changed on first login.
must_change = (u.get("role") == "admin"
and verify_pw(ADMIN_PW, u.get("password", ""))
and not u.get("pw_changed"))
return 200, {"ok": True, "token": token, "role": u["role"],
"username": u["username"],
"callsign": CALLSIGN, "locator": LOCATOR,
"must_change_password": must_change,
"first_run": FIRST_RUN}
# ── Password reset: step 1 — send an email with a token ──────────────────
if p == "/api/auth/reset-request" and method == "POST":
email_or_user = body.get("email", "").strip().lower()
# Look up by email or username
u = (self.find_user_by_email(email_or_user) or
self.find_user(email_or_user))
# Always respond "ok" — don't reveal whether the user exists
if u and u.get("email"):
from auth import make_reset_token
token = make_reset_token(u["id"], u["username"], u["email"])
base_url = body.get("base_url", f"http://localhost:{PORT}")
ok, err = await self._send_reset_email(u["email"], u["username"], token, base_url)
if not ok:
# Do NOT log the token itself — the log is a place a token
# could leak from, and it grants account takeover. Log only
# that delivery failed so the admin knows to check SMTP.
print(f"[auth] SMTP error sending reset for "
f"{u['username']!r}: {err} (token NOT written to the log)")
return 200, {"ok": True, "message": "Jesli konto istnieje i ma email — link zostal wyslany"}
# ── Password reset: step 2 — set the new password from the token ────────
if p == "/api/auth/reset-confirm" and method == "POST":
token = body.get("token", "")
new_pw = body.get("password", "")
if len(new_pw) < 6:
return 400, {"error": "Haslo musi miec minimum 6 znakow"}
from auth import consume_reset_token
entry = consume_reset_token(token)
if not entry:
return 400, {"error": "Link resetowania jest nieprawid³owy lub wygas³ (wazny 1h)"}
u = self.find_user_by_id(entry["user_id"])
if not u:
return 400, {"error": "Uzytkownik nie istnieje"}
u["password"] = hash_pw_secure(new_pw)
u["pw_ver"] = int(u.get("pw_ver", 0)) + 1 # invalidate old tokens
save_json(USR_F, self.users)
print(f"[auth] Password reset for: {u['username']} "
f"(pw_ver={u['pw_ver']}, old sessions invalidated)")
return 200, {"ok": True, "message": "Haslo zostalo zmienione — mozesz sie zalogowac"}
# ── Password reset by an admin (no email) ─────────────────────────────
if p == "/api/auth/admin-reset" and method == "POST":
if not user: return 401, {"error": "Wymagane logowanie"}
if user.get("role") != "admin": return 403, {"error": "Tylko admin"}
target_id = body.get("user_id")
new_pw = body.get("password", "")
if len(new_pw) < 6:
return 400, {"error": "Haslo musi miec minimum 6 znakow"}
u = self.find_user_by_id(target_id)
if not u: return 404, {"error": "Uzytkownik nie istnieje"}
u["password"] = hash_pw_secure(new_pw)
u["pw_ver"] = int(u.get("pw_ver", 0)) + 1 # invalidate the user's old tokens
save_json(USR_F, self.users)
print(f"[auth] Admin reset the password for {u['username']!r} "
f"(pw_ver={u['pw_ver']}, old sessions invalidated)", flush=True)
return 200, {"ok": True}
if p == "/api/config/lang" and method == "GET":
# Public (no auth) - deliberately placed BEFORE the "if not
# user" gate right below. Read by an inline script at the very
# top of index.html, which runs for EVERY visitor including a
# fresh browser that has never logged in yet (before the
# client-side redirect to login.html even happens) - so this
# can't require a token the visitor doesn't have. Lets i18n.js
# pick the server-wide default language (set at install time -
# see HAMCTRL-installer.iss / data.py::get_cfg) instead of
# always starting in Polish. A per-browser choice saved in
# localStorage always overrides this once someone actually
# toggles the language via the UI.
return 200, {"lang": self.cfg.get("lang", "pl")}
if not user:
return 401, {"error": "Wymagane logowanie"}
role = user.get("role", "viewer")
uid = user.get("id", "")
# FIX: same promotion as ws_handler (see the comment there) - the
# JWT's role is frozen at login time, so granting the "admin"
# granular permission afterward only satisfied _has_perm() checks,
# not the many plain "role == 'admin'" checks throughout this file.
# A user with the Admin-panel permission should BE an admin, not a
# partial one.
if role != "admin":
_u_obj_perm = self.find_user_by_id(uid) or {}
if _u_obj_perm.get("permissions", {}).get("admin"):
role = "admin"
if p == "/api/auth/change-password" and method == "POST":
old_pw = body.get("old_password", "")
new_pw = body.get("new_password", "")
u_obj = self.find_user_by_id(uid)
if not u_obj: return 404, {"error": "Brak konta"}
if not verify_pw(old_pw, u_obj.get("password", "")):
return 403, {"error": "Złe stare hasło"}
if len(new_pw) < 8:
return 400, {"error": "Hasło min. 8 znaków"}
u_obj["password"] = hash_pw_secure(new_pw)
# Mark that the password was changed from the default - disables
# the forced password-change prompt on login (must_change_password).
u_obj["pw_changed"] = True
# Increment the password version - invalidates ALL old JWT
# tokens (on other devices too). The user has to log in again
# everywhere. Important when the password is changed because it leaked.
u_obj["pw_ver"] = int(u_obj.get("pw_ver", 0)) + 1
save_json(USR_F, self.users)
print(f"[auth] password changed for {u_obj.get('username')!r}, "
f"pw_ver={u_obj['pw_ver']} (old tokens invalidated)", flush=True)
# Generate a new token for the current session (so the user doesn't get logged out)
new_token = jwt_sign({"id": u_obj["id"], "role": u_obj["role"],
"username": u_obj["username"],
"pw_ver": u_obj["pw_ver"]})
return 200, {"ok": True, "token": new_token}
if p == "/api/user/profile" and method == "POST":
u_obj = self.find_user_by_id(uid)
if not u_obj: return 404, {"error": "Brak konta"}
# Field validation (defense against XSS - this data ends up in
# HTML in the admin panel, the online-users list, etc.)
if "callsign" in body:
_cs = str(body.get("callsign", "")).strip().upper()
if _cs and not re.match(r'^[A-Z0-9/]{1,16}$', _cs):
return 400, {"error": "Znak wywolawczy: tylko litery, cyfry i /"}
u_obj["callsign"] = _cs
if "locator" in body:
_loc = str(body.get("locator", "")).strip().upper()
if _loc and not re.match(r'^[A-Z0-9]{1,8}$', _loc):
return 400, {"error": "Lokator: tylko litery i cyfry"}
u_obj["locator"] = _loc
if "name" in body:
_nm = re.sub(r'[<>"\']', '', str(body.get("name", "")).strip())[:64]
u_obj["name"] = _nm
if "email" in body:
u_obj["email"] = str(body.get("email", "")).strip()[:128]
save_json(USR_F, self.users)
return 200, {"ok": True}
if p == "/api/user/tx_eq" and method == "GET":
u_obj = self.find_user_by_id(uid)
if not u_obj: return 404, {"error": "Brak konta"}
return 200, {"ok": True, "tx_eq": u_obj.get("tx_eq", None)}
if p == "/api/user/tx_eq" and method == "POST":
u_obj = self.find_user_by_id(uid)
if not u_obj: return 404, {"error": "Brak konta"}
# Validate the structure: preset (str) and bands (dict with 5 bands)
preset = body.get("preset", "default")
bands = body.get("bands", {})
if not isinstance(preset, str) or not isinstance(bands, dict):
return 400, {"error": "Nieprawidlowa struktura"}
# Validate the band values (int/float, range -20..+15)
allowed_bands = {"bass", "mud", "clarity", "punch", "air"}
clean_bands = {}
for k, v in bands.items():
if k in allowed_bands and isinstance(v, (int, float)):
clean_bands[k] = max(-20, min(15, float(v)))
u_obj["tx_eq"] = {"preset": preset, "bands": clean_bands}
save_json(USR_F, self.users)
return 200, {"ok": True}
# System-wide COM port listing — used by every settings dropdown that
# picks a serial port (radio, CW DTR/RTS keyer, rotator), so the
# operator selects from what Windows actually sees instead of typing
# a port name by hand (source of confusion/typos, esp. with the many
# COM ports the CI-V<->CW<->rotator setup can involve at once).
# Shares the same 30s cache pattern as the relay port list below —
# list_serial_ports() can be slow on Windows.
if p == "/api/system/serial-ports" and method == "GET":
if not self._has_perm(uid, role, "settings"):
return 403, {"error": "Brak uprawnien (ustawienia serwera)"}
now = time.time()
if not hasattr(self, '_serial_ports_cache') or now - getattr(self, '_serial_ports_cache_time', 0) > 30:
try:
self._serial_ports_cache = list_serial_ports() if _RELAY_OK else []
except Exception as _pe:
print(f"[system] serial-ports enumeration failed: {_pe}")
self._serial_ports_cache = []
self._serial_ports_cache_time = now
return 200, {"ok": True, "ports": self._serial_ports_cache}
# ── Relay controller (Arduino SP5IOU) ─────────────────────────────────
if p == "/api/relay/config" and method == "GET":
# Config + list of ports to choose from (admin only)
if role != "admin":
return 403, {"error": "Tylko admin"}
rcfg = self.cfg.get("relay", {})
# Cache the port list for 30s - list_serial_ports() can be slow on Windows
now = time.time()
if not hasattr(self, '_relay_ports_cache') or now - getattr(self, '_relay_ports_cache_time', 0) > 30:
self._relay_ports_cache = list_serial_ports() if _RELAY_OK else []
self._relay_ports_cache_time = now
return 200, {
"ok": True,
"config": rcfg,
"ports": self._relay_ports_cache,
"connected": bool(self.relay and self.relay.is_connected()),
"states": self.relay.get_states() if self.relay else [False] * 8,
"max_pulse_s": MAX_PULSE_S if _RELAY_OK else 10.0,
}
if p == "/api/relay/config" and method == "POST":
if role != "admin":
return 403, {"error": "Tylko admin"}
enabled = bool(body.get("enabled", False))
port = str(body.get("port", "")).strip()
baudrate = int(body.get("baudrate", 9600))
relays = body.get("relays", [])
if not isinstance(relays, list):
return 400, {"error": "relays musi byc lista"}
# Validate each relay
clean_relays = []
for i, r in enumerate(relays[:8]):
if not isinstance(r, dict): continue
mode = r.get("mode", "manual")
if mode not in ("manual", "momentary"):
mode = "manual"
clean_relays.append({
"id": i,
"name": str(r.get("name", f"REL{i}"))[:30],
"mode": mode,
"pulse_s": max(0.1, min(10.0, float(r.get("pulse_s", 1.0)))),
"visible": bool(r.get("visible", True)),
})
self.cfg["relay"] = {
"enabled": enabled,
"port": port,
"baudrate": baudrate,
"relays": clean_relays,
}
save_json(CFG_F, self.cfg)
# Restart the Arduino connection
asyncio.ensure_future(self._relay_reconnect_task())
return 200, {"ok": True}
if p == "/api/relay/state" and method == "GET":
# Available to all logged-in users - state + relays visible per user
rcfg = self.cfg.get("relay", {})
configured = rcfg.get("relays", [])
u_obj = self.find_user_by_id(uid) or {}
u_perms = u_obj.get("permissions", {})
# Admin sees all of them, a regular user only the ones granted via permissions[relay_N]
visible_relays = []
states = self.relay.get_states() if self.relay else [False] * 8
for r in configured:
rid = r.get("id", 0)
perm_key = f"relay_{rid}"
allowed = role == "admin" or bool(u_perms.get(perm_key, False))
if allowed and r.get("visible", True):
visible_relays.append({
**r,
"state": states[rid] if 0 <= rid < 8 else False,
})
return 200, {
"ok": True,
"connected": bool(self.relay and self.relay.is_connected()),
"relays": visible_relays,
}
if p == "/api/relay/action" and method == "POST":
# User clicks a relay button
if not self.relay or not self.relay.is_connected():
return 503, {"error": "Kontroler przekaznikow niepodlaczony"}
relay_id = int(body.get("id", -1))
if not 0 <= relay_id < 8:
return 400, {"error": "Nieprawidlowy id przekaznika"}
# Check permissions
u_obj = self.find_user_by_id(uid) or {}
u_perms = u_obj.get("permissions", {})
perm_key = f"relay_{relay_id}"
if role != "admin" and not u_perms.get(perm_key, False):
return 403, {"error": "Brak dostepu do tego przekaznika"}
# Find this relay's configuration
rcfg = self.cfg.get("relay", {})
relay_conf = None
for r in rcfg.get("relays", []):
if r.get("id") == relay_id:
relay_conf = r
break
if not relay_conf:
return 404, {"error": "Przekaznik nieskonfigurowany"}
mode = relay_conf.get("mode", "manual")
if mode == "momentary":
duration = float(relay_conf.get("pulse_s", 1.0))
asyncio.ensure_future(self.relay.pulse(relay_id, duration))
await self.hub.broadcast({"type": "relay_state", "id": relay_id, "state": True})
# Broadcast the off state after the pulse (simplified - delayed)
async def _notify_off():
await asyncio.sleep(duration + 0.1)
await self.hub.broadcast({"type": "relay_state", "id": relay_id, "state": False})
asyncio.ensure_future(_notify_off())
return 200, {"ok": True, "mode": "momentary", "duration_s": duration}
else: # manual
await self.relay.toggle(relay_id)
state = self.relay.get_state(relay_id)
await self.hub.broadcast({"type": "relay_state", "id": relay_id, "state": state})
return 200, {"ok": True, "mode": "manual", "state": state}
# ── DX Cluster ─────────────────────────────────────────────────────────
if p == "/api/dxcluster/config" and method == "GET":
u_obj = self.find_user_by_id(uid)
if not u_obj: return 404, {"error": "Brak konta"}
cfg = u_obj.get("dxcluster", {})
# Don't return the password in plaintext (just a flag whether it's set)
return 200, {
"ok": True,
"config": {
"host": cfg.get("host", ""),
"port": cfg.get("port", 7300),
"login": cfg.get("login", ""),
"has_password": bool(cfg.get("password", "")),
"auto_connect": bool(cfg.get("auto_connect", False)),
},
"connected": bool(self.dxcluster and self.dxcluster.get_client(uid) and self.dxcluster.get_client(uid).is_connected()),
# Merged across every connected login (see ClusterManager.
# get_merged_history), not just this user's own - matches the
# live dx_spot broadcast, which is now global too.
"history": self.dxcluster.get_merged_history() if self.dxcluster else [],
}
if p == "/api/dxcluster/config" and method == "POST":
u_obj = self.find_user_by_id(uid)
if not u_obj: return 404, {"error": "Brak konta"}
host = str(body.get("host", "")).strip()
try:
port = int(body.get("port", 7300))
except (TypeError, ValueError):
port = 7300
login = str(body.get("login", "")).strip()
password = body.get("password", None)
auto_connect = bool(body.get("auto_connect", False))
existing = u_obj.get("dxcluster", {})
# Keep the old (already-encrypted) password if a new one wasn't
# given (null = no change); a new password is encrypted before saving.
if password is None:
password = existing.get("password", "")
else:
password = encrypt_secret(password)
u_obj["dxcluster"] = {
"host": host,
"port": port,
"login": login,
"password": password,
"auto_connect": auto_connect,
}
save_json(USR_F, self.users)
return 200, {"ok": True}
if p == "/api/dxcluster/connect" and method == "POST":
if not self.dxcluster:
return 503, {"error": "DX Cluster niedostepny"}
u_obj = self.find_user_by_id(uid)
if not u_obj: return 404, {"error": "Brak konta"}
cfg = u_obj.get("dxcluster", {})
if not cfg.get("host") or not cfg.get("login"):
return 400, {"error": "Skonfiguruj adres serwera i login"}
asyncio.ensure_future(self.dxcluster.connect_user(
uid, cfg["host"], int(cfg.get("port", 7300)),
cfg["login"], decrypt_secret(cfg.get("password", ""))
))
return 200, {"ok": True}
if p == "/api/dxcluster/disconnect" and method == "POST":
if not self.dxcluster:
return 200, {"ok": True}
asyncio.ensure_future(self.dxcluster.disconnect_user(uid))
return 200, {"ok": True}
if p == "/api/dxcluster/command" and method == "POST":
if not self.dxcluster:
return 503, {"error": "DX Cluster niedostepny"}
cmd = str(body.get("cmd", "")).strip()
if not cmd: return 400, {"error": "Brak komendy"}
ok = await self.dxcluster.send_command(uid, cmd)
return 200 if ok else 503, {"ok": ok}
if p == "/api/dxcluster/spot" and method == "POST":
# Send a spot to the DX cluster.
# Command: DX <freq_khz> <call> <comment>
# The cluster signs the spot with the user's own login
# (per-user connection), so the spot goes out under the
# callsign of whoever sent it.
if not self.dxcluster:
return 503, {"ok": False, "error": "DX Cluster niedostępny"}
call = str(body.get("call", "")).strip().upper()
freq_hz = body.get("freq_hz", 0)
comment = str(body.get("comment", "")).strip()
# ── Validation (a spot goes out to the WHOLE WORLD - don't send junk) ────
if not call or not re.match(r'^[A-Z0-9/]{3,16}$', call):
return 400, {"ok": False,
"error": "Nieprawidłowy znak (litery, cyfry, /)"}
try:
freq_hz = int(float(freq_hz))
except (ValueError, TypeError):
return 400, {"ok": False, "error": "Nieprawidłowa częstotliwość"}
# Reasonable range: 1.8 MHz - 1300 MHz
if not (1_800_000 <= freq_hz <= 1_300_000_000):
return 400, {"ok": False,
"error": "Częstotliwość poza zakresem (1.8 MHz – 1.3 GHz)"}
# Comment: max 30 characters (most clusters' limit), no control
# characters that could break the telnet protocol
comment = re.sub(r'[\r\n\t]', ' ', comment)[:30].strip()
# The cluster expects the frequency in kHz (e.g. 14074.0 for 14.074 MHz).
# Given with one decimal digit - that's the convention.
freq_khz = freq_hz / 1000.0
khz_str = f"{freq_khz:.1f}".rstrip("0").rstrip(".")
cmd = f"DX {khz_str} {call}"
if comment:
cmd += f" {comment}"
ok = await self.dxcluster.send_command(uid, cmd)
if ok:
u = self.find_user_by_id(uid) or {}
print(f"[dx] spot sent by {u.get('callsign') or u.get('username')}: "
f"{cmd}", flush=True)
return 200, {"ok": True, "sent": cmd}
return 503, {"ok": False,
"error": "Brak połączenia z klastrem — połącz się najpierw"}
if p == "/api/dxcluster/history" and method == "GET":
if not self.dxcluster:
return 200, {"ok": True, "history": []}
return 200, {"ok": True, "history": self.dxcluster.get_merged_history()}
# ── Server status + diagnostics (admin) ─────────────────────────────
if p == "/api/health" and method == "GET":
# Lightweight status for the Settings tab ("System status"
# panel). Public - this is just basic server health, no
# sensitive data. Fields match what settings.js's loadStatus()
# expects. (The 'node' field is kept for frontend compatibility
# - a leftover from a Node.js template, we report the Python version.)
import platform as _plat
try:
_clients = len(getattr(self.hub, "_clients", set()))
except Exception:
_clients = 0
return 200, {
"ok": True,
"uptime": round(time.time() - self._server_start_time),
"node": f"Python {_plat.python_version()}",
"platform": _plat.system(),
"hamlib": (not bool(getattr(self.rig, "sim", True))),
"audio": bool(getattr(self.audio, "rx_active", False)),
"listeners": _clients,
}
if p == "/api/status/perf" and method == "GET":
# Detailed performance diagnostics - CPU, threads, loops, clients.
# Helps assess whether the server keeps up with many users.
if role != "admin":
return 403, {"error": "Tylko admin"}
import threading as _th
import platform as _plat
out = {
"ok": True,
"uptime_s": round(time.time() - self._server_start_time, 1),
"python": _plat.python_version(),
"platform": _plat.system(),
}
# ── Event loop (asyncio) ──────────────────────────────────────────
try:
_loop = asyncio.get_running_loop()
_tasks = [t for t in asyncio.all_tasks(_loop) if not t.done()]
out["event_loop"] = {
"backend": _JSON_BACKEND, # orjson / stdlib
"impl": type(_loop).__name__, # uvloop/winloop = faster
"active_tasks": len(_tasks),
# Names of the background loops (for diagnosing what's running)
"task_names": sorted({
(t.get_coro().__qualname__.split('.')[-1]
if hasattr(t, 'get_coro') and t.get_coro() else '?')
for t in _tasks
})[:20],
}
except Exception as e:
out["event_loop"] = {"error": str(e)}
# ── Python threads ─────────────────────────────────────────────────
try:
threads = _th.enumerate()
out["threads"] = {
"count": len(threads),
"names": sorted(t.name for t in threads)[:20],
}
except Exception as e:
out["threads"] = {"error": str(e)}
# ── CPU / RAM / processes (psutil) ──────────────────────────────────
# psutil on Windows blocks even with interval=None: cpu_times /
# process_iter read per-process info from the OS synchronously
# (looplag stack: cpu_percent -> _proc_info, and process_iter over
# every system process). Since this endpoint is admin-only and polled
# on a timer, run the entire psutil gather in a thread so it never
# freezes the event loop (which was causing the audio hitches).
def _gather_psutil():
_r = {}
try:
import psutil as _ps
_proc = _ps.Process()
with _proc.oneshot():
_cpu_proc = _proc.cpu_percent(interval=None)
_mem = _proc.memory_info()
_nthreads = _proc.num_threads()
_r["cpu"] = {
"cores_physical": _ps.cpu_count(logical=False),
"cores_logical": _ps.cpu_count(logical=True),
"system_percent": _ps.cpu_percent(interval=None),
"per_core_percent": _ps.cpu_percent(interval=None, percpu=True),
"python_process_percent": _cpu_proc,
"python_threads": _nthreads,
}
_vm = _ps.virtual_memory()
_r["memory"] = {
"system_percent": _vm.percent,
"system_total_mb": round(_vm.total / 1048576),
"python_rss_mb": round(_mem.rss / 1048576, 1),
}
_rust = None
for pr in _ps.process_iter(["name", "cpu_percent", "memory_info"]):
try:
if (pr.info["name"] or "").lower().startswith("ham_audio"):
_rust = {
"pid": pr.pid,
"cpu_percent": pr.cpu_percent(interval=None),
"rss_mb": round(pr.info["memory_info"].rss / 1048576, 1),
}
break
except Exception:
continue
_r["rust_audio_process"] = _rust or {"running": False}
except ImportError:
_r["cpu"] = {"error": "psutil niedostepny"}
return _r
try:
out.update(await asyncio.to_thread(_gather_psutil))
except Exception as _pe:
out["cpu"] = {"error": str(_pe)}
# ── WS clients + channel subscriptions ──────────────────────────────
try:
_clients = getattr(self.hub, "_clients", set())
_subs = getattr(self.hub, "_subs", {})
# Count how many clients subscribe to each channel
_chan_count = {}
for _ws, _chans in _subs.items():
for ch in (_chans or []):
_chan_count[ch] = _chan_count.get(ch, 0) + 1
out["websocket"] = {
"clients": len(_clients),
"online_users": len(getattr(self, "online_users", {})),
"channel_subscribers": _chan_count,
}
except Exception as e:
out["websocket"] = {"error": str(e)}
# ── Hot-path state (what's loading the CPU) ────────────────────────────────
out["workload"] = {
"ft8_rx_enabled": bool(getattr(self, "_ft8_rx_enabled", False)),
"ft8_decode_mode": getattr(self, "_ft8_decode_mode", None),
"audio_rx_active": bool(getattr(self.audio, "rx_active", False)),
"rust_audio_connected": bool(getattr(self, "rust_audio", None)),
"rig_sim": bool(getattr(self.rig, "sim", True)),
"ptt": bool(getattr(self.rig, "ptt", False)),
"cq_calling": bool(getattr(self, "_cq_calling", False)),
"radio_lock_held": self.radio_lock.get("user_id") is not None,
"com_bridge_clients": len(getattr(
getattr(self, "com_bridge_ws", None), "_clients", {}) or {}),
"rotators": len(getattr(self, "rotators", [])),
"relay_connected": bool(getattr(self, "relay", None)
and self.relay.is_connected()),
}
return 200, out
if p == "/api/status" and method == "GET":
if role != "admin":
return 403, {"error": "Tylko admin"}
import platform as _plat
uptime_s = time.time() - self._server_start_time
# CI-V / rig
is_civ = isinstance(self.rig, CivRig)
# Active radio from the rigs[] list (model/port/speed live
# here, not in the flat cfg). The first entry = active radio.
_active_rig = (self.cfg.get("rigs") or [{}])[0]
rig_connected = False
rig_sim = True
try:
rig_sim = bool(getattr(self.rig, "sim", True))
rig_connected = is_civ and not rig_sim and getattr(self.rig, "_ser", None) is not None
except Exception:
pass
# Audio backend
audio_backend = "rust" if self.rust_audio else ("ffmpeg" if getattr(self.audio, "_ffmpeg", None) else "python")
# CPU/RAM via psutil (if available)
cpu_pct = None
ram_pct = None
ram_used_mb = None
try:
import psutil as _ps
cpu_pct = _ps.cpu_percent(interval=None) # non-blocking (see above)
vm = _ps.virtual_memory()
ram_pct = vm.percent
ram_used_mb = round(vm.used / (1024*1024))
except Exception:
pass
return 200, {
"ok": True,
"uptime_s": round(uptime_s),
"version": SERVER_VERSION,
# Update notice — only surfaced to admins (operators can't act on
# it). None unless a newer release was found on GitHub.
"update": (getattr(self, "_update_info", None)
if role == "admin" else None),
"python": _plat.python_version(),
"platform": _plat.system(),
"online_count": len(self.online_users),
"rig": {
"backend": "civ" if is_civ else "hamlib",
"connected": rig_connected,
"sim": rig_sim,
# Model/port live in cfg["rigs"][0] (the radio list),
# not in the flat cfg["model"]. Read from the active
# radio, falling back to the old fields and the value
# actually used by the driver (self.rig.port/speed).
"model": (_active_rig.get("model")
or self.cfg.get("model") or "?"),
"port": (getattr(self.rig, "port", None)
or _active_rig.get("port")
or self.cfg.get("civ_port")
or self.cfg.get("port") or "?"),
"speed": (getattr(self.rig, "speed", None)
or _active_rig.get("speed") or "?"),
"freq": getattr(self.rig, "freq", 0),
},
"audio": {
"backend": audio_backend,
"rust": bool(self.rust_audio),
},
"dxcluster": {
"available": bool(self.dxcluster),
},
"relay": {
"available": _RELAY_OK,
"connected": bool(self.relay and self.relay.is_connected()),
},
"system": {
"cpu_pct": cpu_pct,
"ram_pct": ram_pct,
"ram_used_mb": ram_used_mb,
},
}
# CI-V connection test — sends a test freq read
if p == "/api/status/test_civ" and method == "POST":
if role != "admin":
return 403, {"error": "Tylko admin"}
if not isinstance(self.rig, CivRig):
return 200, {"ok": False, "message": "Radio nie jest w trybie CI-V (Hamlib/sim)"}
try:
freq = getattr(self.rig, "freq", 0)
sim = getattr(self.rig, "sim", True)
if sim:
return 200, {"ok": False, "message": "Radio w trybie symulacji — brak fizycznego CI-V"}
return 200, {"ok": True, "message": f"CI-V odpowiada, freq={freq/1e6:.3f} MHz"}
except Exception as e:
return 200, {"ok": False, "message": f"Blad CI-V: {e}"}
# ── Config backup / restore (admin) ────────────────────────────
if p == "/api/backup" and method == "GET":
if role != "admin":
return 403, {"error": "Tylko admin"}
# Return config.json + users.json as one downloadable JSON
# (user passwords included — this is a backup, the admin has access)
import datetime as _dt
return 200, {
"ok": True,
"backup_version": 1,
"created": _dt.datetime.utcnow().isoformat() + "Z",
"server_version": SERVER_VERSION,
"config": self.cfg,
"users": self.users,
}
if p == "/api/restore" and method == "POST":
if role != "admin":
return 403, {"error": "Tylko admin"}
data = body.get("backup", body) # accept {backup:{...}} or directly
if not isinstance(data, dict):
return 400, {"error": "Nieprawidlowy format backupu"}
restored = []
if "config" in data and isinstance(data["config"], dict):
self.cfg = data["config"]
save_json(CFG_F, self.cfg)
restored.append("config")
if "users" in data and isinstance(data["users"], (list, dict)):
self.users = data["users"]
save_json(USR_F, self.users)
restored.append("users")
if not restored:
return 400, {"error": "Backup nie zawiera config ani users"}
return 200, {"ok": True, "restored": restored,
"message": "Przywrocono. Zrestartuj serwer aby zastosowac wszystkie zmiany."}
if p == "/api/auth/me" and method == "GET":
u_obj = self.find_user_by_id(uid)
return 200, {"ok": True, "user": {
"id": uid,
"username": user.get("username", ""),
"role": role,
"callsign": (u_obj or {}).get("callsign") or user.get("username", ""),
"name": (u_obj or {}).get("name", ""),
"locator": (u_obj or {}).get("locator") or LOCATOR,
"email": (u_obj or {}).get("email", ""),
"active": (u_obj or {}).get("active", True),
"permissions": (u_obj or {}).get("permissions", {}),
}}
if p == "/api/radio/state" and method == "GET":
return 200, {**self._radio_lock_state(), "online": self._online_users_state()}
if p == "/api/radio/lock" and method == "POST":
"""Take over the radio (if free, or if you're an admin)."""
# FIX (2026-09-03): a viewer could acquire radio_lock - nothing
# here checked role at all. Beyond being semantically wrong
# ("widz" = read-only), this let a viewer satisfy
# _can_control_radio() and reach any WS handler that relies on
# it without ALSO being in _CONTROL_TYPES's role-based
# hard-block (ft8_tune, ft8_qsy) - see the fix in
# _can_control_radio for the full story.
if role == "viewer":
return 403, {"error": "Widz nie moze przejac radia"}
u_obj = self.find_user_by_id(uid)
if not u_obj: return 403, {"error": "Brak konta"}
if self.radio_lock["user_id"] and self.radio_lock["user_id"] != uid and role != "admin":
holder = self.radio_lock["username"]
return 409, {"error": f"Radio zajete przez {holder}. Popros o zwolnienie."}
self._lock_radio({"id": uid, "username": user.get("username",""),
"callsign": u_obj.get("callsign", user.get("username",""))})
await self.hub.broadcast({**self._radio_lock_state(), "online": self._online_users_state()})
return 200, {"ok": True}
if p == "/api/radio/release" and method == "POST":
"""Release the radio (active operator or admin only)."""
if self.radio_lock["user_id"] != uid and role != "admin":
return 403, {"error": "Nie masz aktywnej blokady radia"}
released_by = self.radio_lock["username"] or user.get("username", "")
self._release_radio()
# Clear pending requests (same as force-release below) — once
# the radio is free, everyone sees "TAKE TRX" directly anyway,
# and stale requests would only leave the requester's "REQUEST
# TRX" button permanently disabled (hasReq would never clear,
# since _lock_radio() removes the entry ONLY for the user who
# actually takes the radio, not for everyone who requested it).
self.radio_requests.clear()
await self.hub.broadcast({**self._radio_lock_state(), "online": self._online_users_state()})
await self.hub.broadcast({"type": "toast",
"message": f"✓ Radio zwolnione przez {released_by}"})
return 200, {"ok": True}
if p == "/api/radio/request" and method == "POST":
"""Send a radio request to the active operator."""
if self._user_has_lock(uid):
return 400, {"error": "Juz masz radio"}
if not self.radio_lock["user_id"]:
# Radio free — take it immediately
u_obj = self.find_user_by_id(uid) or {}
self._lock_radio({"id": uid, "username": user.get("username",""),
"callsign": u_obj.get("callsign", user.get("username",""))})
await self.hub.broadcast({**self._radio_lock_state(), "online": self._online_users_state()})
return 200, {"ok": True, "granted": True}
u_obj = self.find_user_by_id(uid) or {}
callsign = u_obj.get("callsign", user.get("username",""))
self.radio_requests[uid] = {
"username": user.get("username",""),
"callsign": callsign,
"requested_at": time.time(),
}
# Notify the active operator
await self.hub.broadcast({
"type": "radio_request_received",
"from_uid": uid,
"from_cs": callsign,
"message": f"{callsign} prosi o dostep do radia",
})
await self.hub.broadcast({**self._radio_lock_state(), "online": self._online_users_state()})
return 200, {"ok": True, "granted": False, "message": "Prosba wyslana"}
if p == "/api/radio/cancel-request" and method == "POST":
"""Withdraw a radio request."""
self.radio_requests.pop(uid, None)
await self.hub.broadcast({**self._radio_lock_state(), "online": self._online_users_state()})
return 200, {"ok": True}
if p == "/api/radio/reject-request" and method == "POST":
"""The active operator (or admin) rejects someone else's radio request.
The frontend (the "REJECT" button in _showRequestToast,
index.html) used to ONLY remove the toast locally and never
call any API — the request stayed in self.radio_requests
FOREVER (nothing removed it except taking/releasing the radio
or an admin force-release), so the requester's "REQUEST TRX"
button stayed permanently disabled (_renderOpPanel:
hasReq==True -> disabled)."""
target_uid = str(body.get("uid", ""))
if not target_uid:
return 400, {"error": "Brak uid"}
if role != "admin" and not self._user_has_lock(uid):
return 403, {"error": "Tylko aktywny operator lub admin moze odrzucic prosbe"}
req = self.radio_requests.pop(target_uid, None)
if req is None:
return 200, {"ok": True} # already stale (e.g. withdrawn in the meantime)
u_obj = self.find_user_by_id(uid) or {}
by_callsign = u_obj.get("callsign", user.get("username", ""))
await self.hub.broadcast({**self._radio_lock_state(), "online": self._online_users_state()})
await self.hub.broadcast({
"type": "radio_request_rejected",
"to_uid": target_uid,
"by": by_callsign,
})
return 200, {"ok": True}
if p == "/api/radio/force-release" and method == "POST":
"""Admin: force-release the radio."""
if role != "admin": return 403, {"error": "Tylko admin"}
holder = self.radio_lock["username"] or "?"
self._release_radio()
self.radio_requests.clear()
await self.hub.broadcast({**self._radio_lock_state(), "online": self._online_users_state()})
await self.hub.broadcast({"type": "toast",
"message": f"Admin wymusil zwolnienie radia (bylo: {holder})"})
return 200, {"ok": True}
if p == "/api/radio/timeout" and method == "POST":
"""Admin: change the idle timeout."""
if role != "admin": return 403, {"error": "Tylko admin"}
minutes = int(body.get("minutes", 20))
if not 1 <= minutes <= 480: return 400, {"error": "Zakres: 1-480 min"}
self.radio_lock["timeout_min"] = minutes
self.cfg["radio_lock_timeout"] = minutes
save_json(CFG_F, self.cfg)
await self.hub.broadcast({**self._radio_lock_state(), "online": self._online_users_state()})
return 200, {"ok": True}
# ── SMTP config (admin) ───────────────────────────────────────────────
if p == "/api/smtp/config" and method == "GET":
if role != "admin": return 403, {"error": "Tylko admin"}
smtp = self.cfg.get("smtp", {})
# Don't send the SMTP password to the frontend — just **** if it's set
return 200, {"host": smtp.get("host",""), "port": smtp.get("port", 587),
"user": smtp.get("user",""), "from": smtp.get("from",""),
"use_tls": smtp.get("use_tls", True),
"has_password": bool(smtp.get("password"))}
if p == "/api/smtp/config" and method == "POST":
if role != "admin": return 403, {"error": "Tylko admin"}
smtp = self.cfg.setdefault("smtp", {})
smtp.update({
"host": body.get("host", smtp.get("host","")),
"port": int(body.get("port", smtp.get("port", 587))),
"user": body.get("user", smtp.get("user","")),
"from": body.get("from", smtp.get("from","")),
"use_tls": bool(body.get("use_tls", smtp.get("use_tls", True))),
})
if body.get("password"): # Only overwrite the password if a new one was given
smtp["password"] = encrypt_secret(body["password"])
save_json(CFG_F, self.cfg)
return 200, {"ok": True}
if p == "/api/smtp/test" and method == "POST":
if role != "admin": return 403, {"error": "Tylko admin"}
u_obj = self.find_user_by_id(uid) or {}
to_email = body.get("email") or u_obj.get("email", "")
if not to_email:
return 400, {"error": "Podaj adres email docelowy w polu 'EMAIL TESTOWY — WYŚLIJ NA'"}
base_url = body.get("base_url", f"http://localhost:{PORT}")
ok, err = await self._send_reset_email(to_email, "test", "TEST_TOKEN_XXXX", base_url)
if ok:
return 200, {"ok": True, "message": f"✓ Email testowy wysłany do {to_email}"}
return 200, {"ok": False, "message": f"✗ Błąd: {err}"}
return 200, {"user": udata, "ok": True}
if p == "/api/update/config" and method == "GET":
if role != "admin": return 403, {"error": "Tylko admin"}
return 200, {
"enabled": self.cfg.get("updateCheck", True),
"email": self.cfg.get("updateEmail", False),
"current": SERVER_VERSION,
"repo_configured": bool(GITHUB_REPO),
"update": getattr(self, "_update_info", None),
"email_diag": getattr(self, "_update_email_diag", None),
}
if p == "/api/update/config" and method == "POST":
if role != "admin": return 403, {"error": "Tylko admin"}
if "enabled" in body:
self.cfg["updateCheck"] = bool(body["enabled"])
if "email" in body:
self.cfg["updateEmail"] = bool(body["email"])
save_json(CFG_F, self.cfg)
return 200, {"ok": True,
"enabled": self.cfg.get("updateCheck", True),
"email": self.cfg.get("updateEmail", False)}
if p == "/api/update/check" and method == "POST":
# Manual "check now" - the automatic check only ever runs once,
# 8s after server startup (_do_update_check), so without this an
# admin would have to restart the whole server just to see
# whether a newer release showed up since then.
if role != "admin": return 403, {"error": "Tylko admin"}
if not GITHUB_REPO:
return 200, {"ok": False, "error": "GITHUB_REPO nie skonfigurowany"}
await self._run_update_check()
return 200, {"ok": True, "current": SERVER_VERSION,
"update": getattr(self, "_update_info", None),
"email_diag": getattr(self, "_update_email_diag", None)}
if p == "/api/config" and method == "GET":
return 200, {"callsign": CALLSIGN, "locator": LOCATOR, "port": PORT,
"rigs": self.cfg.get("rigs", []),
"rotators": self.cfg.get("rotators", []),
"cwMacros": self.cfg.get("cwMacros", DEFAULT_MACROS),
"models": HAMLIB_MODELS,
"modes": ["USB","LSB","AM","FM","CW","RTTY","PKTUSB","PKTLSB"],
"enabledBands": self.cfg.get("enabledBands", []),
"enabledModes": self.cfg.get("enabledModes", [])}
if p == "/api/config/bands" and method == "GET":
# Bands come from the connected radio's PROFILE, not a shared
# list. This way an IC-9700 shows only 2m/70cm/23cm (no HF),
# while an IC-7300 shows HF+6m+4m (no 2m). The admin picks the
# profile via the radio model; when there's no profile,
# get_civ_profile gives a safe fallback.
try:
from rigs import get_civ_profile
_model = str(self.cfg.get("model") or self.cfg.get("rig_model") or "3073")
_profile = get_civ_profile(_model)
_pbands = _profile.get("bands", {})
except Exception:
_pbands = {}
if _pbands:
# profile format: name -> (min, max, def); the UI wants a min/max/def dict
all_bands = {name: {'min': lo, 'max': hi, 'def': df}
for name, (lo, hi, df) in _pbands.items()}
else:
# Fallback (in case the profile has no bands) — the full list as before
all_bands = {
'160m': {'min':1810000, 'max':2000000, 'def':1850000},
'80m': {'min':3500000, 'max':3800000, 'def':3650000},
'60m': {'min':5351500, 'max':5366500, 'def':5357000},
'40m': {'min':7000000, 'max':7200000, 'def':7100000},
'30m': {'min':10100000, 'max':10150000, 'def':10125000},
'20m': {'min':14000000, 'max':14350000, 'def':14200000},
'17m': {'min':18068000, 'max':18168000, 'def':18100000},
'15m': {'min':21000000, 'max':21450000, 'def':21200000},
'12m': {'min':24890000, 'max':24990000, 'def':24930000},
'10m': {'min':28000000, 'max':29700000, 'def':28400000},
'6m': {'min':50000000, 'max':52000000, 'def':50150000},
'4m': {'min':70000000, 'max':70500000, 'def':70150000},
'2m': {'min':144000000,'max':146000000,'def':144300000},
'70cm': {'min':430000000,'max':440000000,'def':432100000},
}
enabled = self.cfg.get("enabledBands", list(all_bands.keys()))
# filter out bands outside the profile (in case the config remembers older ones)
enabled = [b for b in enabled if b in all_bands]
return 200, {"allBands": all_bands, "enabledBands": enabled}
if p == "/api/config/station" and method == "GET":
# STATION locator — where the antenna is physically located.
# Used to compute the rotor azimuth toward a correspondent (the
# same for every user, since there's only one antenna). Defaults
# from .env STATION_LOCATOR.
return 200, {
"stationLocator": (self.cfg.get("stationLocator")
or LOCATOR or "").strip().upper(),
"callsign": CALLSIGN,
}
if p == "/api/config/station" and method == "POST":
if not self._has_perm(uid, role, "settings"): return 403, {"error": "Brak uprawnien (ustawienia serwera)"}
_loc = str(body.get("stationLocator", "")).strip().upper()
if _loc and not re.match(r"^[A-R]{2}\d{2}([A-X]{2})?$", _loc):
return 400, {"error": "Zly format lokatora (np. JO72 lub JO72AB)"}
self.cfg["stationLocator"] = _loc
save_json(CFG_F, self.cfg)
await self.hub.broadcast({"type": "init_patch",
"stationLocator": _loc})
print(f"[config] Station locator set: {_loc}", flush=True)
return 200, {"ok": True, "stationLocator": _loc}
if p == "/api/config/bands" and method == "POST":
if not self._has_perm(uid, role, "settings"): return 403, {"error": "Brak uprawnien (ustawienia serwera)"}
self.cfg["enabledBands"] = body.get("enabledBands", [])
save_json(CFG_F, self.cfg)
# Broadcast to clients so they refresh the band grid
await self.hub.broadcast({
"type": "config_update",
"enabledBands": self.cfg["enabledBands"],
})
return 200, {"ok": True}
if p == "/api/config/modes" and method == "GET":
all_modes = ["USB","LSB","AM","FM","CW","CW-R","RTTY","RTTY-R",
"USB-D","LSB-D","PSK","PSK-R","PKTUSB","PKTLSB",
"WFM","DV"]
_has_key = "enabledModes" in self.cfg
enabled = self.cfg.get("enabledModes", ["USB","LSB","AM","FM","CW","RTTY","PKTUSB","PKTLSB"])
# Modes with filters — the admin can assign a preferred filter
mode_filters = self.cfg.get("modeFilters", {})
_modes_log_line(f"GET /api/config/modes uid={uid} role={role} "
f"cfg_id={id(self.cfg)} has_key={_has_key} "
f"returned_enabledModes={enabled}")
return 200, {"allModes": all_modes, "enabledModes": enabled, "modeFilters": mode_filters}
if p == "/api/config/modes" and method == "POST":
if role != "admin": return 403, {"error": "Tylko admin"}
if "enabledModes" in body:
self.cfg["enabledModes"] = body["enabledModes"]
if "modeFilters" in body:
self.cfg["modeFilters"] = body["modeFilters"]
save_json(CFG_F, self.cfg)
_modes_log_line(f"POST /api/config/modes uid={uid} cfg_id={id(self.cfg)} "
f"saved_enabledModes={self.cfg.get('enabledModes')}")
await self.hub.broadcast({
"type": "config_update",
"enabledModes": self.cfg.get("enabledModes", []),
"modeFilters": self.cfg.get("modeFilters", {}),
})
return 200, {"ok": True}
if p == "/api/hamlib/status" and method == "GET":
mgr = getattr(self, 'hamlib', None)
if mgr:
return 200, {"servers": mgr.status()}
# Fallback — return the config without state (the server hasn't started yet)
cfg_servers = self.cfg.get('hamlibServers', [
{"port":4532,"enabled":True, "label":"Radio 1 — główne (WSJT-X)"},
{"port":4533,"enabled":False,"label":"Radio 2 — log/skimmer"},
{"port":4534,"enabled":False,"label":"Radio 3 — zapasowy"},
])
return 200, {"servers": [
{**s, "slot":i, "running":False, "clients":0}
for i, s in enumerate(cfg_servers)
]}
if p == "/api/hamlib/config" and method == "POST":
if not self._has_perm(uid, role, "settings"): return 403, {"error": "Brak uprawnien (ustawienia serwera)"}
servers = body.get("servers", [])
# Validate ports
ports = [s.get("port", 4532+i) for i,s in enumerate(servers)]
if len(set(ports)) != len(ports):
return 400, {"error": "Porty muszą być unikalne"}
for port in ports:
if not (1024 <= port <= 65535):
return 400, {"error": f"Port {port} poza zakresem 1024-65535"}
self.cfg["hamlibServers"] = servers
save_json(CFG_F, self.cfg)
# Restart if the manager is active
mgr = getattr(self, 'hamlib', None)
if mgr:
try:
await mgr.restart()
return 200, {"ok": True, "message": "✓ Konfiguracja zapisana i serwery zrestartowane"}
except Exception as e:
return 200, {"ok": True, "message": f"✓ Zapisano — restart wymagany ({e})"}
return 200, {"ok": True, "message": "✓ Konfiguracja zapisana — zrestartuj serwer aby zastosować"}
m = re.match(r"^/api/config/rig/(\d+)$", p)
if m and method == "POST":
if not self._has_perm(uid, role, "settings"): return 403, {"error": "Brak uprawnien (ustawienia serwera)"}
rid = m.group(1)
if not self.cfg.get("rigs"):
self.cfg["rigs"] = []
rigs = self.cfg["rigs"]
existing = next((r for r in rigs if str(r.get("id")) == str(rid)), None)
if existing:
existing.update(body)
else:
rigs.append({**body, "id": rid})
# Sync audio devices from the rig into the global audio config
if "audio" not in self.cfg:
self.cfg["audio"] = {}
if body.get("audioRx") is not None:
self.cfg["audio"]["rxDevice"] = body["audioRx"]
if body.get("audioTx") is not None:
self.cfg["audio"]["txDevice"] = body["audioTx"]
save_json(CFG_F, self.cfg)
return 200, {"ok": True}
if p == "/api/rig/status" and method == "GET":
return 200, {"freq": self.rig.freq, "mode": self.rig.mode,
"bandwidth": self.rig.bw, "ptt": self.rig.ptt,
"split": self.rig.split, "freqB": self.rig.freq_b,
"smeter": self.rig.s_meter,
"connected": self.rig.connected, "sim": self.rig.sim}
if p == "/api/rig/connect" and method == "POST":
if role != "admin": return 403, {"error": "Tylko admin"}
# Pick the backend by model: any model with a civ_profiles.py
# entry (CIV_NATIVE_MODELS - scope-capable OR an older Icom like
# the IC-746 with just a verified command set) -> direct CI-V
# (CivRig), everything else -> rigctld (RigCAT). Swaps app.rig
# on the fly if needed.
sel_model = str((body or {}).get("model") or "").strip()
want_civ = sel_model in CIV_NATIVE_MODELS
is_civ = isinstance(self.rig, CivRig)
if want_civ != is_civ and sel_model:
try: self.rig.close()
except Exception as _ce: print(f"[rig] closing the old backend: {_ce}")
# Give Windows a moment to actually release the serial port
# (terminating the rigctld process isn't instant)
await asyncio.sleep(0.5)
self.rig = CivRig(self.cfg, self._rig_bcast, log=print) if want_civ else RigCAT()
print(f"[rig] backend -> {'direct CI-V' if want_civ else 'rigctld'} (model {sel_model})")
# IMPORTANT: ComBridgeWs held a reference to the OLD CivRig.
# After creating the new one, the listener has to be
# rewired, otherwise COM Bridge clients (CW Skimmer/HRD)
# get no radio responses.
if hasattr(self, 'com_bridge_ws') and self.com_bridge_ws:
new_civ = self.rig if isinstance(self.rig, CivRig) else None
self.com_bridge_ws.attach_civ_rig(new_civ)
print(f"[rig] com_bridge_ws rewired to new CivRig={new_civ is not None}")
# Use values from the form (model/port/speed/civ) if given
ok = await self.rig.connect(self.cfg, override=body or None)
# Save the connection parameters to config.json — so after a
# server restart, App.__init__ picks the right backend
# (CivRig/RigCAT) and connect() connects to the same radio/port at startup.
if body:
rig_id = str(body.get("rigId") or body.get("id") or "1")
if not self.cfg.get("rigs"):
self.cfg["rigs"] = []
rigs = self.cfg["rigs"]
existing = next((r for r in rigs if str(r.get("id")) == rig_id), None)
persist_keys = {"model", "port", "speed", "civ", "civAddr", "name"}
persist = {k: v for k, v in body.items() if k in persist_keys and v}
if existing:
existing.update(persist)
else:
rigs.append({"id": rig_id, **persist})
save_json(CFG_F, self.cfg)
print(f"[rig] saved config for rig={rig_id}: {persist}")
# After connecting, broadcast the full radio state so the panel
# immediately shows the current freq/mode/S-meter (rig_poll only sends deltas).
try:
if ok:
self.rig.s_meter = await self.rig.get_smeter()
await self.hub.broadcast({
"type": "telemetry",
"freq": self.rig.freq, "freqB": self.rig.freq_b,
"mode": self.rig.mode, "bandwidth": self.rig.bw,
"filterNum": self.rig.filter_num,
"preamp": self.rig.preamp, "attenuator": self.rig.attenuator,
"tuner": self.rig.tuner,
"tuner": self.rig.tuner,
"sMeter": self.rig.s_meter, "ptt": self.rig.ptt,
"split": self.rig.split,
"connected": self.rig.connected, "sim": self.rig.sim,
})
except Exception as _be:
print(f"[rig] broadcast after connect: {_be}")
await self._refresh_caps_cache()
return 200, {"ok": ok, "sim": self.rig.sim,
"message": self.rig.last_msg,
"error": self.rig.last_err,
"rigctldPath": self.rig.rigctld_path,
"rigctldFound": self.rig.rigctld_found,
"freq": self.rig.freq, "mode": self.rig.mode,
"bandwidth": self.rig.bw}
# ── Radio features panel (capabilities + admin whitelist) ──────────────
if p == "/api/rig/features" and method == "GET":
# FIX (2026-09-03, live report: "user widzi wylaczone przyciski
# typu BK-IN"), REVISED same day: the first version of this fix
# made ONLY role=="admin" ever get the admin-shaped response
# (every detected dynamic action/slider, including DISABLED
# ones, each tagged with its enabled state - meant for the
# config table). That broke a real, intended feature: the user
# confirmed "settings" is a DELIBERATE admin-delegated grant -
# an operator with it is meant to fully use KONFIGURACJA
# (including FUNKCJE RADIA), same as an admin would, not a
# narrower "can see their own audio device" permission.
#
# The actual bug was narrower: radiofunctions.js's regular
# RADIO-tab panel (the LIVE control surface) and admin.js's
# KONFIGURACJA->FUNKCJE RADIA config table both call this same
# endpoint, and the response shape alone tells the frontend
# which one it's rendering into - so whichever shape a
# settings-holder received leaked into BOTH places at once.
# Fixed by making the caller's INTENT explicit instead:
# admin.js's config-page fetch now passes ?view=admin (see
# loadRigFeatures() in admin.js) - only THAT gets the full
# shape for a settings-holding non-admin. radiofunctions.js's
# plain fetch (no param) never does, so the live RADIO-tab
# panel is now always the filtered view unless the caller is
# a real admin (who reasonably sees/can-toggle everything
# everywhere anyway, same as before this whole investigation).
view_param = query.get("view", "")
if isinstance(view_param, list):
view_param = view_param[0] if view_param else ""
wants_admin_view = (view_param == "admin")
_admin_view = "admin" if (role == "admin" or
(wants_admin_view and self._has_perm(uid, role, "settings"))) else role
return 200, await self._get_rig_features(_admin_view)
if p == "/api/rig/features" and method == "POST":
if not self._has_perm(uid, role, "settings"): return 403, {"error": "Brak uprawnien (ustawienia serwera)"}
return 200, await self._set_rig_features(body or {})
# ── Advanced radio settings — modulation input, AGC mode, VOX
# fine-tuning, NB depth/width, TX delay (see rigs/civ_profiles.py's
# _IC7300_ADV for the full list). Most entries are "settings"-only
# (admin/delegated-admin, infrastructure-level - a deliberate
# 2026-10-15 decision, see the long comment on /api/rig/features GET
# above). A spec entry can opt into gating by the per-rig FEATURES
# whitelist instead via "feature" (see AGC_MODE in civ_profiles.py,
# "feature": "agc") - ONE KONFIGURACJA -> FUNKCJE RADIA switch then
# exposes it to every regular operator, same as tx_power/tuner/
# autotune already work (_feature_allowed, not _has_perm). Falls
# back to the "settings" permission when a spec has no "feature"
# field, so every pre-existing entry keeps today's behavior.
if p == "/api/rig/adv" and method == "GET":
adv_spec = self.rig.profile.get("adv", {})
values, visible_spec = {}, {}
for key, spec in adv_spec.items():
feat = spec.get("feature")
allowed = self._feature_allowed(feat, role) if feat else self._has_perm(uid, role, "settings")
if not allowed:
continue
visible_spec[key] = spec
values[key] = await self.rig.get_adv(key)
return 200, {"ok": True, "values": values, "spec": visible_spec}
if p == "/api/rig/adv" and method == "POST":
key = str(body.get("key", ""))
spec = self.rig.profile.get("adv", {}).get(key.upper())
if spec is None:
return 400, {"error": f"Nieznane ustawienie: {key}"}
feat = spec.get("feature")
allowed = self._feature_allowed(feat, role) if feat else self._has_perm(uid, role, "settings")
if not allowed:
return 403, {"error": "Brak uprawnien (ustawienia serwera)"}
ok = await self.rig.set_adv(key, body.get("value"))
if not ok:
return 200, {"ok": False, "error": "Radio odrzucilo komende (NG) lub nie odpowiedzialo"}
return 200, {"ok": True}
# REMOVED (2026-10-18): /api/rig/rit GET/POST (RIT/ΔTX, CI-V 0x21).
# User decision after live testing: Hz-set gave NG even after fixing
# the BCD byte order to match this file's proven frequency
# convention (freq_to_bcd/bcd_to_freq) - root cause not found, not
# worth continuing to chase. Fully removed end-to-end (civ.py's
# get_rit/_read_rit/set_rit_hz/set_rit_enabled/set_dtx_enabled,
# advradio.js's rit* functions, the #rit-row UI, the "rit" entry in
# rigs/features.py's FEATURES and _BASE_CAPS_CIV) - "zeby po nim
# slad nie zostal, nie potrzebujemy tego".
# ── CW Keyer (DTR/RTS keying - CAT/cmd-17 retired 2026-08-26) ───────
# Frontend (cw.js) calls:
# POST /api/cw/send {text, vars} -> sends text as CW (DTR/RTS)
# POST /api/cw/stop -> aborts sending, key up
# GET /api/cw/status -> {method, dtrPort, dtrLine, usbKeying}
# Settings UI (settings.js Settings._cwKeyerSave) calls:
# POST /api/cw/dtr-port {port, line} -> the ONE save action: sets
# cwMethod/cwDtrPort/cwDtrLine, calls configure_keyer, derives
# and sends the matching CI-V USB Keying value (1A 05 00 97),
# and turns BK-IN on - see the handler below for why this is
# all one call instead of three.
# /api/cw/method and /api/cw/usb-keying still exist standalone
# (harmless if called directly) but the UI no longer calls them
# separately - doing so used to require the two to be kept in sync
# by hand.
if p == "/api/cw/macros" and method == "GET":
u_obj = self.find_user_by_id(uid) or {}
macros = u_obj.get("cwMacros") or self.cfg.get("cwMacros", DEFAULT_MACROS)
return 200, {"macros": macros}
if p == "/api/cw/macros" and method == "POST":
macros = body.get("macros", [])
u_obj = self.find_user_by_id(uid)
if not u_obj: return 404, {"error": "Brak konta"}
u_obj["cwMacros"] = macros
save_json(USR_F, self.users)
return 200, {"ok": True}
if p == "/api/cw/send" and method == "POST":
print(f"[cw] /api/cw/send: method={self.cfg.get('cwMethod','dtr')!r} text={body.get('text','')!r} sim={self.rig.sim} ser={getattr(self.rig, '_ser', None) is not None}", flush=True)
if role == "viewer":
return 403, {"ok": False, "error": "Brak uprawnien (CW)"}
# FIX (2026-09-03, same audit as rotator/webrtc_offer): the
# granular per-operator "cw" permission (users.json
# permissions.cw) was only ever enforced by hiding the CW tile
# in the frontend (data-perm="cw") - a direct POST here worked
# for ANY non-viewer operator holding the lock, even one the
# admin explicitly denied CW access to.
if not self._has_perm(uid, role, "cw"):
return 403, {"ok": False, "error": "Brak uprawnien (CW)"}
if role != "admin" and not self._user_has_lock(uid):
holder = self.radio_lock["callsign"] or self.radio_lock["username"] or ""
msg_err = f"CW zablokowany — radio ma {holder}" if holder else "CW zablokowany — najpierw przejmij radio"
return 403, {"ok": False, "error": msg_err}
# Reject overlap: a CW transmission holds PTT for its whole duration.
# Firing a second macro before the first finishes made both texts
# overlap in the rig's CW buffer (garbled / truncated output). One
# transmission at a time — the operator waits for it to finish.
if getattr(self, "_cw_tx_busy", False):
return 200, {"ok": False, "error": "CW zajete — trwa nadawanie, poczekaj"}
if not self._is_band_allowed():
return 403, {"ok": False, "error": "TX zablokowany — pasmo niedozwolone przez admina"}
cross, band_a, band_b = await self._is_split_cross_band()
if cross:
return 403, {"ok": False, "error": f"CW TX zablokowany — split cross-band ({band_a} RX / {band_b} TX). Wylacz split lub ustaw VFO-B na to samo pasmo."}
text = str(body.get("text", ""))
cw_vars = body.get("vars", {}) or {}
for key, val in cw_vars.items():
text = text.replace("{" + key.upper() + "}", str(val))
text = text.replace("{" + key + "}", str(val))
text = text.strip()
if not text:
return 200, {"ok": False, "error": "Pusty tekst"}
# DTR/RTS is the only CW keying method - see cw_empty_ptt saga:
# cmd-17 CAT keying was retired entirely (2026-08-26) once
# DTR/RTS+BK-IN was live-confirmed reliable. DTR/RTS covers both
# modern radios (IC-7300+: internal CI-V "USB Keying", see
# set_usb_keying) and older non-USB radios (external transistor
# keyer physically wired into the KEY jack from the DTR line -
# the same mechanism, just without the CI-V 00 97 step, since
# there the wiring itself does the job).
method_cw = self.cfg.get("cwMethod", "dtr")
if method_cw not in ("dtr", "rts"):
method_cw = "dtr"
wpm = int(self.rig.level_values.get("KEYSPD", 18) or 18)
print(f"[cw] method {method_cw!r}, port={self.cfg.get('cwDtrPort','') or '(same as CI-V)'}", flush=True)
self._cw_tx_busy = True
# cw_tx_start drives the live sidetone (public/js/cw_sidetone.js)
# - the keyer thread shapes real dits/dahs on the DTR/RTS line,
# the browser can't observe that directly, so it synthesizes a
# same-content, same-WPM tone locally instead.
await self.hub.broadcast({"type": "cw_tx_start", "text": text, "wpm": wpm})
await self.hub.broadcast({"type": "cw_sending", "text": text,
"method": method_cw, "wpm": wpm})
try:
try:
# Approximate PTT indicator for connected clients - the
# real PTT ON happens moments later inside the
# background keyer thread (after its own ACK wait).
await self.hub.broadcast({"type": "ptt", "ptt": True})
await self.rig.send_cw_dtr_rts(text, wpm)
except Exception as e:
await self.hub.broadcast({"type": "ptt", "ptt": False})
return 200, {"ok": False, "error": str(e)}
finally:
self._cw_tx_busy = False
# send_cw_dtr_rts() is fire-and-forget (spawns a background
# keyer thread and returns immediately) - there's no other
# signal for when that thread actually finishes keying, so this
# timer is the only way to know. Use the SAME exact Morse timing
# as the rig (civ._cw_text_duration_s) instead of a flat guess.
try:
duration_s = self.rig._cw_text_duration_s(text, wpm) + 0.15
except Exception:
dit_ms = 1200.0 / max(5, wpm)
duration_s = len(text) * 10 * dit_ms / 1000.0
duration_s = min(60.0, max(0.3, duration_s))
async def _cw_done_after(delay):
await asyncio.sleep(delay)
await self.hub.broadcast({"type": "ptt", "ptt": False})
await self.hub.broadcast({"type": "cw_done"})
asyncio.create_task(_cw_done_after(duration_s))
return 200, {"ok": True}
if p == "/api/cw/stop" and method == "POST":
if role == "viewer":
return 403, {"ok": False, "error": "Brak uprawnien (CW)"}
if not self.rig.sim:
try:
await self.rig.stop_cw_dtr_rts()
except Exception as e:
return 200, {"ok": False, "error": str(e)}
await self.hub.broadcast({"type": "cw_stopped"})
return 200, {"ok": True}
if p == "/api/cw/status" and method == "GET":
return 200, {
"method": self.cfg.get("cwMethod", "dtr"),
"dtrPort": self.cfg.get("cwDtrPort", ""),
"dtrLine": self.cfg.get("cwDtrLine", "DTR"),
"usbKeying": self.cfg.get("cwUsbKeying", 0),
"wpm": self.rig.level_values.get("KEYSPD", 18),
"capabilities": {"dtr": True, "rts": True},
}
if p == "/api/cw/method" and method == "POST":
if role != "admin":
return 403, {"error": "Tylko admin"}
m = body.get("method", "dtr")
if m not in ("dtr", "rts"):
return 400, {"error": "Nieznana metoda"}
self.cfg["cwMethod"] = m
save_json(CFG_F, self.cfg)
port = self.cfg.get("cwDtrPort", "")
self.rig.configure_keyer(port, m.upper())
# BK-IN must be ON for the DTR/RTS key line to actually trigger
# TX (live-confirmed 2026-08-26).
if hasattr(self.rig, "set_bk_in") and not self.rig.sim:
try:
await asyncio.to_thread(self.rig.set_bk_in, True)
except Exception as e:
print(f"[cw] set_bk_in error: {e}", flush=True)
return 200, {"ok": True, "method": m}
if p == "/api/cw/usb-keying" and method == "POST":
if role != "admin":
return 403, {"error": "Tylko admin"}
try:
value = int(body.get("value", 0))
except (TypeError, ValueError):
return 400, {"ok": False, "error": "Nieprawidlowa wartosc"}
if value not in (0, 1, 2, 3, 4):
return 400, {"ok": False, "error": "value musi byc 0-4"}
if not hasattr(self.rig, "set_usb_keying"):
return 500, {"ok": False, "error": "set_usb_keying niedostepne w tym profilu"}
try:
# Blocking (up to 0.4s, waits for radio ACK/NG) - off the event loop thread.
ok = await asyncio.get_running_loop().run_in_executor(
None, self.rig.set_usb_keying, value)
except Exception as e:
return 500, {"ok": False, "error": str(e)}
if ok:
self.cfg["cwUsbKeying"] = value
save_json(CFG_F, self.cfg)
return 200, {"ok": True, "value": value}
return 200, {"ok": False, "error": "Radio nie potwierdzilo ustawienia USB Keying (zobacz log [civ] USB Keying)"}
if p == "/api/cw/dtr-port" and method == "POST":
if role != "admin":
return 403, {"error": "Tylko admin"}
port = body.get("port", "").strip()
line = body.get("line", "DTR").upper()
if line not in ("DTR", "RTS"):
return 400, {"error": "Linia musi byc DTR lub RTS"}
self.cfg["cwMethod"] = line.lower() # 'dtr' or 'rts'
self.cfg["cwDtrPort"] = port
self.cfg["cwDtrLine"] = line
save_json(CFG_F, self.cfg)
try:
self.rig.configure_keyer(port, line)
except Exception as e:
return 200, {"ok": False, "error": str(e)}
# Derive and send the matching CI-V USB Keying value (1A 05 00
# 97) automatically instead of making the admin pick USB(A)/
# USB(B) by hand - compare the entered port against the REAL
# CI-V port (self.rig.port), case-insensitively, empty=same.
# FIX (2026-08-26): this used to be guessed on the FRONTEND
# from "is the PORT field empty" alone - broke for this user,
# who types the SAME port CI-V already uses (not leaving it
# blank), which looked like a genuinely separate second port
# and picked USB(B) - a value the radio correctly rejects,
# surfaced as "Radio nie potwierdzilo ustawienia USB Keying"
# even though DTR itself keys fine via the same-port fallback.
civ_port = (getattr(self.rig, "port", "") or "").strip().upper()
same_as_civ = (not port) or (port.strip().upper() == civ_port)
usb_value = (1 if line == "DTR" else 2) + (0 if same_as_civ else 2)
usb_ok = True
if hasattr(self.rig, "set_usb_keying") and not self.rig.sim:
try:
usb_ok = await asyncio.get_running_loop().run_in_executor(
None, self.rig.set_usb_keying, usb_value)
except Exception as e:
usb_ok = False
print(f"[cw] set_usb_keying error: {e}", flush=True)
if usb_ok:
self.cfg["cwUsbKeying"] = usb_value
save_json(CFG_F, self.cfg)
# BK-IN must be ON for the DTR/RTS key line to actually trigger
# TX (live-confirmed 2026-08-26).
if hasattr(self.rig, "set_bk_in") and not self.rig.sim:
try:
await asyncio.to_thread(self.rig.set_bk_in, True)
except Exception as e:
print(f"[cw] set_bk_in error: {e}", flush=True)
msg = f"CW Keyer {line} skonfigurowany: {port or 'port CI-V'}"
if not usb_ok:
msg += " — UWAGA: radio NIE potwierdzilo USB Keying, kluczowanie moze nie dzialac"
return 200, {"ok": True, "message": msg, "usbKeyingOk": usb_ok, "usbKeyingValue": usb_value}
if p == "/api/rotator" and method == "GET":
return 200, [r.state() for r in self.rotators]
m = re.match(r"^/api/rotator/(\d+)/position$", p)
if m and method == "POST":
# Without this check, ANY logged-in user (even a viewer) could
# turn the antenna via a direct API call, regardless of whether
# the START/STOP buttons were grayed out in the UI (CSS
# .radio-readonly). Same pattern as /api/cw/send: a viewer is
# always blocked, everyone else must hold radio_lock (unless admin).
if role == "viewer":
return 403, {"error": "Brak uprawnien (rotator)"}
# FIX (2026-09-03): the granular per-operator "rotator"
# permission (users.json permissions.rotator, admin-editable
# per user) was ONLY ever enforced by hiding the tile in the
# frontend (index.html's data-perm="rotator") - a direct POST
# here worked regardless, for ANY non-viewer operator, even
# one the admin explicitly denied rotator access. Found while
# changing that tile to stay visible-but-disabled instead of
# hidden (a UI-only hide is not a permission boundary).
if not self._has_perm(uid, role, "rotator"):
return 403, {"error": "Brak uprawnien (rotator)"}
if role != "admin" and not self._user_has_lock(uid):
holder = self.radio_lock["callsign"] or self.radio_lock["username"] or ""
msg_err = f"Rotator zablokowany — radio ma {holder}" if holder else "Rotator zablokowany — najpierw przejmij radio"
return 403, {"error": msg_err}
rot = self.get_rot(int(m.group(1)))
if not rot: return 404, {"error": "Rotator nie znaleziony"}
rot.go_to(float(body.get("az", 0)))
return 200, {"ok": True}
m = re.match(r"^/api/rotator/(\d+)/stop$", p)
if m and method == "POST":
# Same as /position above — see the comment there (both the
# viewer block and the granular "rotator" permission check).
if role == "viewer":
return 403, {"error": "Brak uprawnien (rotator)"}
if not self._has_perm(uid, role, "rotator"):
return 403, {"error": "Brak uprawnien (rotator)"}
if role != "admin" and not self._user_has_lock(uid):
holder = self.radio_lock["callsign"] or self.radio_lock["username"] or ""
msg_err = f"Rotator zablokowany — radio ma {holder}" if holder else "Rotator zablokowany — najpierw przejmij radio"
return 403, {"error": msg_err}
rot = self.get_rot(int(m.group(1)))
if not rot: return 404, {"error": "Rotator nie znaleziony"}
# FIX (reported live 2026-08-24): stop() now reports whether the
# write to the port actually succeeded (see the comment on
# Rotator.stop in rotator.py) - surface that here instead of
# always claiming success, so the operator gets a real error
# instead of a false "stopped" when the port write failed.
if not rot.stop():
return 200, {"ok": False, "error": "Nie udało się wysłać STOP do rotora (port nie odpowiada)"}
return 200, {"ok": True}
m = re.match(r"^/api/rotator/(\d+)/test$", p)
if m and method == "POST":
rot = self.get_rot(int(m.group(1)))
if not rot:
return 200, {"testOk": False, "testMsg": "Rotator nie znaleziony"}
# REAL connection test: try connecting and reading the position
# from the COM port. The previous version only returned the
# stored state (and in the wrong fields — hence 'undefined' in
# the UI). Now we actively poll the rotor to confirm the port responds.
driver = getattr(rot, "model", "") or getattr(rot, "driver_type", "") or "?"
_was_sim_before = getattr(rot, "sim", False)
try:
# Try connecting if not already connected. NOTE: the
# rotor's connect() does NOT raise on error - it silently
# sets sim=True and returns False. So we must check the
# RESULT, otherwise a dead port would masquerade as a
# successful "simulation".
if not getattr(rot, "connected", False) and not _was_sim_before:
try:
ok = await asyncio.wait_for(
asyncio.to_thread(rot.connect), timeout=6.0)
except asyncio.TimeoutError:
return 200, {"testOk": False, "driverType": driver,
"testMsg": "Timeout przy otwieraniu portu (6s)"}
except Exception as _ce:
return 200, {"testOk": False, "driverType": driver,
"testMsg": f"Nie moge otworzyc portu: {_ce}"}
# connect() returned False and the rotor fell back to sim => the port is dead
if not ok and getattr(rot, "sim", False) and not _was_sim_before:
_err = getattr(rot, "last_err", "") or \
f"Port {getattr(rot, 'port', '?')} nie odpowiada"
return 200, {"testOk": False, "driverType": driver,
"testMsg": f"Połączenie nieudane: {_err}"}
# Read the position from the port — this confirms the
# rotor is responding. _read_pos blocks on the serial port,
# so it's run in a thread with a timeout (like the
# watchdog), so a dead port doesn't hang the server.
try:
az = await asyncio.wait_for(
asyncio.to_thread(rot._read_pos, 3.0), timeout=5.0)
except asyncio.TimeoutError:
return 200, {"testOk": False, "driverType": driver,
"testMsg": "Port nie odpowiada (timeout 5s)"}
except Exception as _re:
return 200, {"testOk": False, "driverType": driver,
"testMsg": f"Brak odpowiedzi z portu: {_re}"}
if az is None:
az = getattr(rot, "az", None)
# Real simulation (configured from the start), not an emergency fallback
if _was_sim_before and getattr(rot, "sim", False):
return 200, {"testOk": True, "driverType": "SYMULACJA",
"testPos": {"az": round(float(az or 0), 1)},
"testMsg": "Tryb symulacji (brak fizycznego portu)"}
return 200, {"testOk": True, "driverType": driver,
"testPos": {"az": round(float(az or 0), 1)},
"testMsg": "Port odpowiada"}
except Exception as e:
return 200, {"testOk": False, "driverType": driver,
"testMsg": str(e)}
if p == "/api/rotator/config" and method == "POST":
if not self._has_perm(uid, role, "settings"): return 403, {"error": "Brak uprawnien (ustawienia serwera)"}
self.cfg["rotators"] = body.get("rotators", [])
save_json(CFG_F, self.cfg)
self.init_rotators()
return 200, {"ok": True, "count": len(self.rotators)}
if p == "/api/users" and method == "GET":
if role != "admin": return 403, {"error": "Tylko admin"}
# Return the full set of editable profile fields - otherwise
# the admin doesn't see the current values when opening the
# user editor, and their changes "reset" fields to empty.
# The password (hash) isn't returned, for security.
return 200, [{"id": u["id"], "username": u["username"],
"callsign": u.get("callsign", u["username"]),
"name": u.get("name", ""),
"email": u.get("email", ""),
"locator": u.get("locator", ""),
"role": u["role"],
"active": u.get("active", True),
"permissions": u.get("permissions", {}),
}
for u in self.users]
if p == "/api/users" and method == "POST":
if role != "admin": return 403, {"error": "Tylko admin"}
# Validation - username and callsign only allow safe characters
# (no < > " etc.), so HTML/JS can't be injected (defense
# against XSS at the source, independent of escaping in the UI).
_un = body.get("username", "").strip()
if not _un or len(_un) > 32 or not re.match(r'^[A-Za-z0-9_.\-]+$', _un):
return 400, {"error": "Nazwa uzytkownika: 1-32 znakow, tylko litery, cyfry, . _ -"}
_cs = body.get("callsign", "").strip().upper()
if _cs and not re.match(r'^[A-Z0-9/]{1,16}$', _cs):
return 400, {"error": "Znak wywolawczy: tylko litery, cyfry i / (max 16)"}
_loc = body.get("locator", "").strip().upper()
if _loc and not re.match(r'^[A-Z0-9]{1,8}$', _loc):
return 400, {"error": "Lokator: tylko litery i cyfry (max 8)"}
# name and email - truncate length and strip HTML-unsafe characters
_nm = body.get("name", "").strip()[:64]
_nm = re.sub(r'[<>"\']', '', _nm)
_role = body.get("role", "viewer")
if _role not in ("admin", "operator", "viewer"):
return 400, {"error": "Nieprawidlowa rola (admin/operator/viewer)"}
new_u = {"id": str(int(time.time()*1000)),
"username": _un,
"password": hash_pw_secure(body.get("password", "changeme")),
"role": _role, "active": True,
"name": _nm,
"callsign": _cs,
"locator": _loc,
"email": body.get("email", "").strip()[:128],
"permissions": body.get("permissions", {}),
}
self.users.append(new_u)
save_json(USR_F, self.users)
return 200, {"ok": True, "id": new_u["id"]}
m = re.match(r"^/api/users/(.+)$", p)
if m and method in ("PUT", "PATCH", "DELETE"):
if role != "admin": return 403, {"error": "Tylko admin"}
target_uid = m.group(1)
# Nothing used to protect against deleting/demoting/
# deactivating the LAST active admin - such a request would
# succeed and leave the whole system with no one able to
# manage anything (recovery would require manually editing
# users.json on the server disk). The frontend hides the
# DELETE button for your OWN account (see renderUsers in
# admin.js), but that's UI only - nothing stopped deleting
# ANOTHER admin if it was the only one, or demoting/disabling your own account.
def _other_active_admins():
return sum(1 for _u in self.users
if _u["id"] != target_uid and _u.get("role") == "admin"
and _u.get("active", True))
if method == "DELETE":
_target = self.find_user_by_id(target_uid)
if (_target and _target.get("role") == "admin"
and _target.get("active", True) and _other_active_admins() == 0):
return 400, {"error": "Nie można usunąć ostatniego aktywnego admina"}
self.users = [u for u in self.users if u["id"] != target_uid]
save_json(USR_F, self.users)
return 200, {"ok": True}
u = self.find_user_by_id(target_uid)
if not u: return 404, {"error": "Nie znaleziono"}
_is_last_admin = u.get("role") == "admin" and u.get("active", True) and _other_active_admins() == 0
# Full set of editable fields - kept in sync with the user
# profile (the user edits via /api/user/profile, the admin via
# here - both write to the same fields in users.json).
if "password" in body and body["password"]:
u["password"] = hash_pw_secure(body["password"])
if "role" in body:
_new_role = body["role"]
if _new_role not in ("admin", "operator", "viewer"):
return 400, {"error": "Nieprawidlowa rola (admin/operator/viewer)"}
if _is_last_admin and _new_role != "admin":
return 400, {"error": "Nie można odebrać roli admina ostatniemu aktywnemu adminowi"}
u["role"] = _new_role
if "active" in body:
_new_active = bool(body["active"])
if _is_last_admin and not _new_active:
return 400, {"error": "Nie można dezaktywować ostatniego aktywnego admina"}
u["active"] = _new_active
if "name" in body: u["name"] = (body["name"] or "").strip()
if "callsign" in body: u["callsign"] = (body["callsign"] or "").strip().upper()
if "locator" in body: u["locator"] = (body["locator"] or "").strip().upper()
if "email" in body: u["email"] = (body["email"] or "").strip()
if "permissions" in body: u["permissions"] = body["permissions"]
save_json(USR_F, self.users)
return 200, {"ok": True}
if p.startswith("/api/scope"):
# Scope only available in direct CI-V mode AND on a model whose
# profile actually claims scope hardware - CIV_NATIVE_MODELS now
# also includes CI-V radios with NO scope (e.g. IC-746), so
# isinstance(CivRig) alone would let scope endpoints "succeed"
# into a radio that never sends 0x27 0x00 data at all.
_civ_scope = (isinstance(self.rig, CivRig)
and self.rig.profile.get("capabilities", {}).get("scope", False))
if not _civ_scope:
return 200, {"ok": False, "sim": True, "running": False,
"message": "Scope dostępny tylko dla radia ze scope (IC-7300/7610/705...) "
"w trybie bezpośrednim CI-V. Wybierz taki model i połącz."}
if p.endswith("/stop"):
return 200, self.rig.scope_stop()
if p.endswith("/span") and method == "POST":
# Change the waterfall span (the IC-7300 supports 2.5/5/10/25/50/100/250/500 kHz)
try:
span_hz = int(body.get("span_hz", 25000))
except (TypeError, ValueError):
return 400, {"ok": False, "error": "Nieprawidlowa wartosc span_hz"}
if hasattr(self.rig, 'set_scope_span'):
try:
# set_scope_span waits (blocking, up to 0.4s) for
# the radio's own ACK/NG - run off the event loop
# thread, same as every other rig set_* call.
ok = await asyncio.get_running_loop().run_in_executor(
None, self.rig.set_scope_span, span_hz)
if ok:
return 200, {"ok": True, "span_hz": span_hz}
return 400, {"ok": False, "error": f"Radio nie potwierdzil zmiany spanu na {span_hz} Hz "
f"(zobacz log [civ] scope span)"}
except Exception as e:
return 500, {"ok": False, "error": str(e)}
return 500, {"ok": False, "error": "set_scope_span niedostepne w tym profilu"}
if p.endswith("/mode") and method == "POST":
# Center/Fixed - czy okno wodospadu "podaza" za VFO (Center)
# czy zostaje na stalym zakresie czestotliwosci (Fixed).
try:
mode_code = int(body.get("mode", 0))
except (TypeError, ValueError):
return 400, {"ok": False, "error": "Nieprawidlowa wartosc mode"}
if mode_code not in (0, 1):
return 400, {"ok": False, "error": "mode musi byc 0 (Center) lub 1 (Fixed)"}
if hasattr(self.rig, 'set_scope_mode'):
try:
ok = await asyncio.get_running_loop().run_in_executor(
None, self.rig.set_scope_mode, mode_code)
if ok:
return 200, {"ok": True, "mode": mode_code}
return 400, {"ok": False, "error": f"Radio nie potwierdzilo zmiany trybu wodospadu na {mode_code} "
f"(zobacz log [civ] scope mode)"}
except Exception as e:
return 500, {"ok": False, "error": str(e)}
return 500, {"ok": False, "error": "set_scope_mode niedostepne w tym profilu"}
return 200, self.rig.scope_start()
if p == "/api/tunnel/cleanup" and method == "POST":
if role != "admin": return 403, {"error": "Tylko admin"}
result = await self.tunnel.cleanup()
return 200, result
if p == "/api/tunnel/status" and method == "GET":
return 200, self.tunnel.get_status()
if p == "/api/tunnel/config" and method == "GET":
# get_config() returns the WHOLE config, including the
# Cloudflare Tunnel token and the DuckDNS token in plaintext
# (see TunnelManager._cfg / save_config in tunnel_manager.py -
# no redaction here like e.g. /api/dxcluster/config, which only
# sends has_password). The whole INTERNET tab is
# data-perm="admin" in index.html, but that only HIDES the tab
# button in the UI - the real enforcement MUST be here. Without
# this gate, any logged-in viewer/operator could read these
# tokens via fetch('/api/tunnel/config') in the browser console
# (full control over the Cloudflare tunnel / ability to change
# the DuckDNS DNS record).
if role != "admin": return 403, {"error": "Tylko admin"}
return 200, self.tunnel.get_config()
if p == "/api/tunnel/config" and method == "POST":
if role != "admin": return 403, {"error": "Tylko admin"}
self.tunnel.save_config(body)
return 200, {"ok": True}
if p == "/api/tunnel/start" and method == "POST":
if role != "admin": return 403, {"error": "Tylko admin"}
asyncio.create_task(self.tunnel.start(
mode=body.get("mode"),
token=body.get("token"),
hostname=body.get("hostname"),
))
return 200, {"ok": True}
if p == "/api/tunnel/stop" and method == "POST":
if role != "admin": return 403, {"error": "Tylko admin"}
asyncio.create_task(self.tunnel.stop())
return 200, {"ok": True}
if p == "/api/tunnel/check" and method == "GET":
# check_available runs several subprocess.run calls (sc query,
# cloudflared --version) — all blocking. The frontend polls this
# status on a timer, so run it in a thread to avoid freezing the
# event loop (looplag stack pointed at _is_service_installed).
return 200, await asyncio.to_thread(self.tunnel.check_available)
if p == "/api/tunnel/gen-cert" and method == "POST":
if role != "admin": return 403, {"error": "Tylko admin"}
asyncio.create_task(self.tunnel.gen_cert_task())
return 200, {"ok": True}
if p == "/api/wsjtx/start" and method == "POST":
port = int(body.get("port", body.get("udpPort", 2237)))
ok = await self.wsjtx.start(port=port)
if ok:
await self.hub.broadcast({"type": "wsjtx_status", "running": True,
"text": f"UDP monitor aktywny (port {port})"})
return 200, {"ok": ok, "status": self.wsjtx.get_status()}
if p == "/api/ft8/halt" and method == "POST":
self._ft8_tx_abort = True
self._stop_cq_calling() # stop periodic CQ
# See the identical block/comment in WS "ft8_tx_stop" above —
# the frontend's haltTx() sends BOTH (WS + this REST call) on
# every HALT click, so the same fix has to be here too.
if self._qso_engine.is_active():
print(f"[autoqso] HALT: aborting QSO with {self._qso_engine.partner_call}")
self._qso_engine.abort_qso()
self._qso_period_locked = False
# Invalidate ANY already-scheduled (in-flight, waiting for
# its window) automatic transmission to this partner -
# without this the stale-TX-guard in _ft8_tx_sequence_inner
# had no way to know this was a REAL abort, not just "a
# newer action was scheduled" (see the comment at
# _autoqso_tx_seq in _send_auto_tx).
self._autoqso_tx_seq += 1
await self.hub.broadcast({"type": "auto_qso_status",
"state": self._qso_engine.state,
"partner": None})
if not self.rig.sim:
try: await self.rig.set_ptt(False)
except: pass
self.rig.ptt = False
await self.hub.broadcast({"type": "ptt", "ptt": False})
await self.hub.broadcast({"type": "ft8_tx_halted"})
return 200, {"ok": True}
if p == "/api/wsjtx/stop" and method == "POST":
await self.wsjtx.stop()
await self.hub.broadcast({"type": "wsjtx_status", "running": False,
"text": "UDP monitor zatrzymany"})
return 200, {"ok": True}
if p == "/api/wsjtx/status" and method == "GET":
return 200, self.wsjtx.get_status()
if p == "/api/audio/enumerate":
rust = getattr(self, 'rust_audio', None)
if rust and rust._connected:
try:
devices = await rust.list_devices()
if isinstance(devices, list):
rx = [d["name"] for d in devices if isinstance(d, dict) and d.get("is_input")]
tx = [d["name"] for d in devices if isinstance(d, dict) and not d.get("is_input")]
return 200, {"devices":{"rx":rx,"tx":tx},"source":"rust"}
except Exception as e:
print(f"[audio] Rust enumerate error: {e}", flush=True)
try: dev = enumerate_audio_devices()
except Exception: dev = {"rx":[],"tx":[],"source":"error"}
return 200, {"devices":{"rx":dev.get("rx",[]),"tx":dev.get("tx",[])},"source":dev.get("source","none")}
if p == "/api/audio/status" and method == "GET":
rust = getattr(self, 'rust_audio', None)
# txVolume is held by Python (config.json), NOT Rust. The Rust
# status doesn't know this value -> the UI reading status from
# Rust showed a stale/default one ("the backend shows something
# else"). So we ALWAYS inject txVolume (and the devices) from
# the Python config into the returned status.
_py_audio = self.cfg.get("audio", {})
_py_txvol = float(_py_audio.get("txVolume", 4.0))
_py_txvol_ssb = float(_py_audio.get("txVolumeSsb", 1.0))
_py_gain_mode_ssb = _py_audio.get("txGainModeSsb", "manual")
# Devices ALWAYS come from the Python config (not from Rust/not
# from self.audio) — Python is the source of truth for saved
# settings, in BOTH branches below. FIX: the fallback branch
# (Rust unavailable/error, e.g. right after saving a device
# while Rust is restarting the RX stream) had no "tx_device"
# at all, and took "rx_device" from self.audio.rx_device — an
# attribute of Python's OWN, UNRELATED capture (the CW
# decoder), not from the saved configuration. On refresh at
# that moment the UI got empty/wrong devices and visually
# "forgot" the selection despite the correct value being saved
# in config.json.
_rxd = _py_audio.get("rxDevice", "")
_txd = _py_audio.get("txDevice", "")
if rust and rust._connected:
try:
status = await rust.get_status()
if isinstance(status, dict) and "error" not in status:
if _rxd: status["rx_device"] = _rxd
if _txd: status["tx_device"] = _txd
status["txVolume"] = _py_txvol # overwrite with the value from config.json (FT8/FT4)
status["txVolumeSsb"] = _py_txvol_ssb # separate multiplier for the SSB microphone
status["txGainModeSsb"] = _py_gain_mode_ssb
return 200, status
except Exception as e:
print(f"[audio] Rust status error: {e}", flush=True)
_st = self.audio.get_status()
if isinstance(_st, dict):
if _rxd: _st["rx_device"] = _rxd
if _txd: _st["tx_device"] = _txd
_st["txVolume"] = _py_txvol
_st["txVolumeSsb"] = _py_txvol_ssb
_st["txGainModeSsb"] = _py_gain_mode_ssb
return 200, _st
if p == "/api/audio/detect" and method == "GET":
# Return the radio-card auto-detect status.
# Doesn't trigger a new detection - this is the cache from init.
if not self._has_perm(uid, role, "settings"): return 403, {"error": "Brak uprawnien (ustawienia serwera)"}
return 200, {
"ok": True,
"auto_enabled": self._audio_auto,
"detection": self._audio_detection,
"current": self.cfg.get("audio", {}),
}
if p == "/api/audio/detect" and method == "POST":
# Force a fresh detection (e.g. after plugging in the radio after the server started)
if not self._has_perm(uid, role, "settings"): return 403, {"error": "Brak uprawnien (ustawienia serwera)"}
try:
detection = auto_detect_radio_audio()
self._audio_detection = detection
if detection["detected"]:
if "audio" not in self.cfg: self.cfg["audio"] = {}
if detection["rx"]: self.cfg["audio"]["rxDevice"] = detection["rx"]
if detection["tx"]: self.cfg["audio"]["txDevice"] = detection["tx"]
self.audio.cfg = self.cfg["audio"]
save_json(CFG_F, self.cfg)
return 200, {"ok": True, "detection": detection}
except Exception as e:
return 500, {"error": str(e)}
# ── CI-V TCP Bridge (for legacy software via com0com) ─────────────────
# ── COM Bridge WS - per-user COM port configuration ────────────────
if p == "/api/com/config" and method == "GET":
# Return the current user's port configuration
# The JWT payload uses 'id' (not 'user_id') - consistent with the rest of the code
if not user:
return 401, {'error': 'not authenticated'}
uid = user.get('id') or user.get('user_id') # fallback for old tokens
u = next((x for x in self.users if x.get('id') == uid), None)
if not u:
return 404, {'error': 'user not found'}
ports = u.get('com_bridge', {}).get('ports', [])
return 200, {'ok': True, 'ports': ports}
if p == "/api/com/config" and method == "POST":
# Save the current user's port configuration
if not user:
return 401, {'error': 'not authenticated'}
uid = user.get('id') or user.get('user_id')
u = next((x for x in self.users if x.get('id') == uid), None)
if not u:
return 404, {'error': 'user not found'}
ports = body.get('ports', [])
if not isinstance(ports, list):
return 400, {'error': 'ports musi byc lista'}
if len(ports) > 16:
return 400, {'error': 'max 16 portow'}
# Validate each port
clean = []
for entry in ports:
if not isinstance(entry, dict):
continue
service = entry.get('service', '')
if service not in self.com_bridge_ws.SERVICES:
continue
clean.append({
'service': service,
'baud': int(entry.get('baud', 19200)),
'parity': str(entry.get('parity', 'N'))[:1],
'bits': int(entry.get('bits', 8)),
'stop': int(entry.get('stop', 1)),
})
if 'com_bridge' not in u:
u['com_bridge'] = {}
u['com_bridge']['ports'] = clean
save_json(USR_F, self.users)
return 200, {'ok': True, 'ports': clean}
if p == "/api/com/stats" and method == "GET":
# Stats of connected clients (for the admin)
if role != "admin":
return 403, {'error': 'tylko admin'}
return 200, self.com_bridge_ws.get_stats()
# ── Debug endpoint for diagnosing subscriptions (available to everyone)
if p == "/api/debug/subscriptions" and method == "GET":
clients_info = []
for cli_ws, channels in self.hub._subs.items():
info = self.online_users.get(cli_ws, {})
clients_info.append({
"user_id": info.get("user_id", "?"),
"username": info.get("username", "?"),
"callsign": info.get("callsign", ""),
"role": info.get("role", "?"),
"channels": sorted(channels),
})
my_uid = user.get("id", "") if user else ""
my_role = role if user else "?" # already promoted for the "admin" permission - see the fix near line 1793
return 200, {
"ok": True,
"my_uid": my_uid,
"my_role": my_role,
"ft8_rx_enabled": self._ft8_rx_enabled,
"ft8_rx_owner_uid": self._ft8_rx_owner_uid,
"clients": clients_info,
}
# ── CloudLog / WaveLog API — PER-USER settings ──────────────
# Every user has their own API keys, URL, and station — saved in
# the user's profile (users.json), not in the global cfg.
if p == "/api/cloudlog/config" and method == "GET":
if not user:
return 401, {"error": "Brak autoryzacji"}
u = self.find_user_by_id(user["id"])
cl = dict((u or {}).get("cloudlog", {}))
for key in ("apiKeyQso", "apiKeyRadio"):
if cl.get(key):
cl[key] = decrypt_secret(cl[key])
return 200, cl
if p == "/api/cloudlog/config" and method == "POST":
if not user:
return 401, {"error": "Brak autoryzacji"}
u = self.find_user_by_id(user["id"])
if not u:
return 404, {"error": "Uzytkownik nie istnieje"}
u["cloudlog"] = {
"url": body.get("url", "").strip(),
"apiKeyQso": encrypt_secret(body.get("apiKeyQso", "").strip()),
"apiKeyRadio": encrypt_secret(body.get("apiKeyRadio", "").strip()),
"stationId": int(body.get("stationId", 1)),
"liveEnabled": bool(body.get("liveEnabled", False)),
}
save_json(USR_F, self.users)
return 200, {"ok": True}
if p == "/api/cloudlog/test" and method == "POST":
if not user:
return 401, {"error": "Brak autoryzacji"}
import aiohttp as _aiohttp
url = body.get("url", "").rstrip("/")
api_key = body.get("apiKeyQso", "")
if not url or not api_key:
return 400, {"ok": False, "error": "Podaj adres i API Key"}
# If the user typed an address with /index.php - don't duplicate it
base = url[:-10].rstrip("/") if url.endswith("/index.php") else url
try:
# Cloudlog: API key in the URL, station_info endpoint.
# (user_info with an X-Auth-Key header DOESN'T EXIST -> 404)
# Returns the user's list of station profiles.
async with _aiohttp.ClientSession() as sess:
async with sess.get(
f"{base}/index.php/api/station_info/{api_key}",
timeout=_aiohttp.ClientTimeout(total=8)
) as resp:
txt = await resp.text()
if resp.status == 200:
try:
data = await resp.json(content_type=None)
except Exception:
return 200, {"ok": False,
"error": "Odpowiedz nie jest JSON — "
"sprawdz adres Cloudloga"}
# We expect a list of station profiles
if isinstance(data, list) and data:
stations = [
{"id": s.get("station_id"),
"name": s.get("station_profile_name", ""),
"callsign": s.get("station_callsign", ""),
"active": bool(s.get("station_active"))}
for s in data
]
names = ", ".join(
f"#{s['id']} {s['callsign'] or s['name']}"
for s in stations[:4])
return 200, {"ok": True,
"message": f"Połączono. Profile stacji: {names}",
"stations": stations}
if isinstance(data, dict) and data.get("status") == "failed":
return 200, {"ok": False,
"error": f"Cloudlog: {data.get('reason','błędny klucz')}"}
return 200, {"ok": False,
"error": "Brak profili stacji — "
"utwórz Station Location w Cloudlogu"}
elif resp.status in (401, 403):
return 200, {"ok": False, "error": "Błędny API Key"}
elif resp.status == 404:
return 200, {"ok": False,
"error": "404 — sprawdź adres (bez /index.php na końcu), "
"np. https://log.example.com"}
else:
return 200, {"ok": False,
"error": f"HTTP {resp.status}: {txt[:60]}"}
except Exception as e:
return 200, {"ok": False, "error": str(e)[:80]}
if p == "/api/cloudlog/radio" and method == "POST":
# Send the current frequency and mode to CloudLog/WaveLog
# NOTE: without this, the endpoint took 'url' directly from the
# body and POSTed it externally with NO authorization at all —
# the server acting as an open proxy (SSRF) for anyone who hits
# the server's address, even without a HAMCTRL account.
# /config and /test already had this check, it was missing here.
if not user:
return 401, {"error": "Brak autoryzacji"}
import aiohttp as _aiohttp
url = body.get("url", "").rstrip("/")
api_key = body.get("apiKey", "")
freq = int(body.get("freq", 0))
mode = body.get("mode", "USB")
station = int(body.get("stationId", 1))
if not url or not api_key:
return 400, {"ok": False, "error": "Brak konfiguracji"}
base = url[:-10].rstrip("/") if url.endswith("/index.php") else url
try:
# Format per the Cloudlog API/Radio documentation:
# {key, radio, frequency (Hz), mode, timestamp "YYYY/MM/DD HH:MM"}
# station_id is NOT a field of this endpoint (removed).
payload = {
"key": api_key,
"radio": "Ham Radio CTRL",
"frequency": str(freq),
"mode": mode,
"timestamp": __import__("datetime").datetime.utcnow().strftime("%Y/%m/%d %H:%M"),
}
async with _aiohttp.ClientSession() as sess:
async with sess.post(
f"{base}/index.php/api/radio",
json=payload,
headers={"Content-Type": "application/json"},
timeout=_aiohttp.ClientTimeout(total=5)
) as resp:
return 200, {"ok": resp.status in (200, 201)}
except Exception as e:
return 200, {"ok": False, "error": str(e)[:80]}
if p == "/api/cloudlog/qso" and method == "POST":
# Send a QSO to CloudLog/WaveLog (ADIF - see qso_to_adif)
# Same missing-authorization issue as /api/cloudlog/radio above — added.
if not user:
return 401, {"error": "Brak autoryzacji"}
import aiohttp as _aiohttp
url = body.get("url", "").rstrip("/")
api_key = body.get("apiKey", "")
station = body.get("stationId", 1)
qso = body.get("qso", {})
if not url or not api_key or not qso:
return 400, {"ok": False, "error": "Brak danych QSO lub konfiguracji"}
base = url[:-10].rstrip("/") if url.endswith("/index.php") else url
try:
payload = {
"key": api_key,
"station_profile_id": str(station),
"type": "adif",
"string": qso_to_adif(qso),
}
async with _aiohttp.ClientSession() as sess:
async with sess.post(
f"{base}/index.php/api/qso",
json=payload,
headers={"Content-Type": "application/json",
"Accept": "application/json"},
timeout=_aiohttp.ClientTimeout(total=8)
) as resp:
txt = await resp.text()
if resp.status in (200, 201):
try:
rdata = await resp.json(content_type=None)
except Exception:
rdata = {}
if isinstance(rdata, dict) and rdata.get("status") == "failed":
return 200, {"ok": False,
"error": f"Cloudlog: {rdata.get('reason','?')}"}
return 200, {"ok": True}
else:
return 200, {"ok": False, "error": f"HTTP {resp.status}: {txt[:80]}"}
except Exception as e:
return 200, {"ok": False, "error": str(e)[:80]}
# ── Callbook (QRZ.com / HamQTH.com) — callsign lookup ──────────
# Each user has their own login credentials (see callbook.py). Done
# server-side: neither service has CORS, and we don't want passwords in JS.
if p == "/api/callbook/config" and method == "GET":
if not user: return 401, {"error": "Brak autoryzacji"}
u = self.find_user_by_id(user["id"])
cfg = dict((u or {}).get("callbook", {}))
for key in ("qrzPassword", "hamqthPassword"):
if cfg.get(key):
cfg[key] = decrypt_secret(cfg[key])
return 200, cfg
if p == "/api/callbook/config" and method == "POST":
if not user: return 401, {"error": "Brak autoryzacji"}
u = self.find_user_by_id(user["id"])
if not u: return 404, {"error": "Uzytkownik nie istnieje"}
u["callbook"] = {
"qrzUsername": body.get("qrzUsername", "").strip(),
"qrzPassword": encrypt_secret(body.get("qrzPassword", "").strip()),
"hamqthUsername": body.get("hamqthUsername", "").strip(),
"hamqthPassword": encrypt_secret(body.get("hamqthPassword", "").strip()),
}
save_json(USR_F, self.users)
return 200, {"ok": True}
if p == "/api/callbook/test" and method == "POST":
if not user: return 401, {"error": "Brak autoryzacji"}
service = body.get("service", "")
username = body.get("username", "")
password = body.get("password", "")
if service not in ("qrz", "hamqth") or not username or not password:
return 400, {"ok": False, "error": "Brak danych"}
res = await callbook.test_connection(service, username, password, uid)
if not res.get("ok"):
res["error"] = res.get("error") or "Blad logowania - sprawdz dane"
return 200, res
if p == "/api/callbook/lookup" and method == "GET":
if not user: return 401, {"error": "Brak autoryzacji"}
call = query.get("call", "")
if isinstance(call, list): call = call[0] if call else ""
if not call:
return 400, {"ok": False, "error": "Brak znaku"}
u = self.find_user_by_id(uid) or {}
cb = u.get("callbook", {})
qrz_creds = ((cb.get("qrzUsername"), decrypt_secret(cb.get("qrzPassword")))
if cb.get("qrzUsername") and cb.get("qrzPassword") else None)
hamqth_creds = ((cb.get("hamqthUsername"), decrypt_secret(cb.get("hamqthPassword")))
if cb.get("hamqthUsername") and cb.get("hamqthPassword") else None)
if not qrz_creds and not hamqth_creds:
return 200, {"ok": False, "error": "Skonfiguruj QRZ.com lub HamQTH w USTAWIENIACH"}
res = await callbook.lookup(call, uid, qrz_creds, hamqth_creds)
if not res:
return 200, {"ok": False, "error": "Nie znaleziono znaku"}
return 200, {"ok": True, **res}
# ── FT8 Timer API ────────────────────────────────────────────────────────
if p == "/api/ft8timer/global" and method == "GET":
dur = self.cfg.get("ft8_safety_timer", 6)
return 200, {"duration_min": dur, "active": True}
if p == "/api/ft8timer/global" and method == "POST":
if role != "admin": return 403, {"error": "Tylko admin"}
dur = max(1, min(60, int(body.get("duration_min", 6))))
self.cfg["ft8_safety_timer"] = dur
save_json(CFG_F, self.cfg)
return 200, {"ok": True, "duration_min": dur}
# ── QSO Log API ──────────────────────────────────────────────────────────
is_admin_user = (role == "admin")
if p == "/api/ft8/alltxt" and method == "GET":
# Pobranie dziennika ALL.TXT (ft8_all_log.py) - wlasnego; admin
# moze podac ?user_id= innego operatora.
target = uid
if is_admin_user and query.get("user_id"):
q = query["user_id"]
target = q[0] if isinstance(q, list) else q
u_t = self.find_user_by_id(target) or {}
call = (u_t.get("callsign") or str(target)).replace("/", "_")
return "export", {
"body": self._all_log.read(target),
"content_type": "text/plain", # bez "; charset" - aiohttp go tu odrzuca (500)
"filename": f"ALL_{call}_{time.strftime('%Y%m%d')}.TXT",
}
if p == "/api/qsolog/worked_before" and method == "GET":
# Check whether a call has already been worked. Optional: band/mode.
call = query.get("call", "")
band = query.get("band", None)
mode = query.get("mode", None)
if isinstance(call, list): call = call[0] if call else ""
if isinstance(band, list): band = band[0] if band else None
if isinstance(mode, list): mode = mode[0] if mode else None
result = qso_db.worked_before(uid, call, band, mode)
return 200, {"ok": True, **result}
if p == "/api/qsolog/all" and method == "DELETE":
# FIX (2026-09-03): granular "log" permission (users.json
# permissions.log, labelled "Log QSO - zapis") was never
# actually checked anywhere - the QSO log TAB's data-perm="view"
# just means "visible to every authenticated user, no per-user
# gate" (it is NOT a real, admin-manageable permission - see
# auth.js's _REAL_PERM_KEYS), so unchecking "log" for an operator
# did nothing at all. Applies to every write below
# (add/edit/delete/bulk-import), not just this one.
if not self._has_perm(uid, role, "log"):
return 403, {"error": "Brak uprawnien (log QSO)"}
target_uid = query.get("user_id") or uid
if isinstance(target_uid, list): target_uid = target_uid[0]
# Admin can delete any user's log, a regular user only their own
if target_uid != uid and not is_admin_user:
return 403, {"error": "Brak uprawnień"}
# Deleting thousands of QSOs can take a while - don't block the event loop
count = await asyncio.to_thread(qso_db.delete_all, target_uid)
return 200, {"ok": True, "count": count}
if p == "/api/debug/qso_users" and method == "GET":
if not is_admin_user: return 403, {}
conn = qso_db._get_conn()
rows = conn.execute("SELECT user_id, COUNT(*) as cnt FROM qso GROUP BY user_id").fetchall()
return 200, {"users": [{"user_id": r[0], "count": r[1]} for r in rows]}
if p == "/api/qsolog" and method == "GET":
try:
# MultiDict — take the first values
filters = {k: (v[0] if isinstance(v, list) else v)
for k, v in query.items()}
filter_uid = filters.pop("user_id", None) or None
# The list can return thousands of QSOs - offload to a thread so as not to block
result = await asyncio.to_thread(
qso_db.list_qsos, uid, is_admin=is_admin_user,
filter_uid=filter_uid, **filters)
return 200, result
except Exception as e:
import traceback; traceback.print_exc()
return 500, {"error": str(e)}
if p == "/api/qsolog/bulk" and method == "POST":
# Bulk import of many QSOs at once
if not self._has_perm(uid, role, "log"):
return 403, {"ok": False, "error": "Brak uprawnien (log QSO)"}
qsos_list = body.get("qsos", [])
if not qsos_list:
return 400, {"ok": False, "error": "Brak QSO"}
try:
# STABILITY: an ADIF import with thousands of QSOs takes
# seconds. Calling it synchronously would FREEZE the event
# loop - every user would stall, WebSockets would time out.
# Run it in a thread.
result = await asyncio.to_thread(
qso_db.add_qsos_bulk, uid, qsos_list)
return 200, {"ok": True, "inserted": result["inserted"],
"skipped": result["skipped"],
"duplicates": result.get("duplicates", 0)}
except Exception as e:
import traceback; traceback.print_exc()
return 500, {"ok": False, "error": str(e)}
if p == "/api/qsolog" and method == "POST":
# Add a new QSO
if not self._has_perm(uid, role, "log"):
return 403, {"ok": False, "error": "Brak uprawnien (log QSO)"}
try:
body["source"] = body.get("source", "manual")
# FIX: my_call (-> ADIF OPERATOR/STATION_CALLSIGN) used to
# come ONLY from the frontend (S?.callsign, i.e.
# window.AppState.callsign, set async from /api/auth/me).
# Reported live: QSOs logged via the quick-log widget
# reached CloudLog with NO operator/station callsign at
# all, even though the user's profile definitely had one
# set — confirmed by exporting the QSO to ADIF and seeing
# no STATION_CALLSIGN/OPERATOR tag. Rather than chase the
# exact frontend timing (some page/reconnect path left
# AppState.callsign empty while window.CurrentUser was
# already populated - the two aren't kept in sync the same
# way), the backend now falls back to the AUTHENTICATED
# user's own profile callsign whenever the request didn't
# supply one - this is the actual source of truth anyway.
if not (body.get("my_call") or "").strip():
_u_obj = self.find_user_by_id(uid)
body["my_call"] = ((_u_obj or {}).get("callsign")
or user.get("username", ""))
qso = qso_db.add_qso(uid, body)
# Push to CloudLog if configured
asyncio.ensure_future(self._cloudlog_push_qso(uid, qso))
# Live-update the automation mini-log (WSJT-X page) - the
# same broadcast as on auto-save from the automation (see
# "QSO SAVED to the log" in _process_auto_qso) - one
# message type for both save paths (manual and automatic).
await self.hub.broadcast({"type": "qso_logged", "qso": qso})
return 200, {"ok": True, "id": qso["id"]}
except ValueError as e:
return 400, {"ok": False, "error": str(e)}
except Exception as e:
import traceback; traceback.print_exc()
return 500, {"ok": False, "error": str(e)}
if p.startswith("/api/deepcw/capture") and method == "POST":
# Diagnostics: record a dozen or so seconds of audio exactly as
# it's fed to the model (after resampling and the filter).
if deepcw_engine is None:
return 503, {"error": "Silnik DeepCW niedostepny"}
_sec = float(body.get("seconds", 15))
_path = deepcw_engine.start_capture(_sec)
return 200, {"ok": True, "path": _path, "seconds": _sec}
if p.startswith("/api/deepcw/capture_file") and method == "GET":
# Download the recorded file for listening back
try:
import pathlib as _pl
_f = _pl.Path(deepcw_engine._cap_path)
if not _f.exists():
return 404, {"error": "Brak nagrania — najpierw uruchom zapis"}
return "export", {
"body": _f.read_bytes(),
"content_type": "audio/wav",
"filename": "deepcw_capture.wav",
}
except Exception as e:
return 500, {"error": str(e)}
if p.startswith("/api/deepcw/download") and method == "POST":
# Download the ONNX model (~15 MB) from the repository — a
# network operation that writes to the data directory, so it
# requires the "server settings" permission (not just the admin role).
if not self._has_perm(uid, role, "settings"): return 403, {"error": "Brak uprawnien (ustawienia serwera)"}
if deepcw_engine is None:
return 503, {"error": "Silnik DeepCW niedostepny "
"(brak pakietu onnxruntime)"}
res = await deepcw_engine.download(broadcast_fn=self.hub.broadcast)
return (200 if res.get("ok") else 500), res
if p.startswith("/api/deepcw/engine_status") and method == "GET":
if deepcw_engine is None:
return 200, {"hasModel": False, "ready": False,
"error": "silnik niedostepny"}
_st = deepcw_engine.get_status()
# Whether the decoder is CURRENTLY running on the server (someone has
# a window open). Lets a freshly-opened/reloaded window sync to the
# real state instead of always assuming "off" — fixes the mismatch
# where the server kept decoding but the reopened UI showed nothing.
_st["running"] = getattr(self, "_cw_task", None) is not None
_st["viewers"] = len(getattr(self, "_cw_viewers", set()))
return 200, _st
if p.startswith("/api/deepcw/known_calls") and method == "GET":
# Pool of known calls for browser-side HIGHLIGHTING. Fed from
# FT8 decodes + the QSO log; the frontend adds its own DX
# cluster spots (each user has their own) and already-worked calls.
if deepcw_engine is None:
return 200, {"calls": []}
try:
# Add calls from the operator's log (one-off, cheap)
deepcw_engine.add_known_calls(
qso_db.worked_calls(uid, is_admin=is_admin_user))
except Exception:
pass
return 200, {"calls": sorted(deepcw_engine._known_calls)}
if p.startswith("/api/qsolog/calls") and method == "GET":
# Lightweight list of UNIQUE calls from the log — for marking
# worked stations in Band Activity. Replaces fetching the full
# QSO list (capped at 200 entries), which meant older contacts
# weren't marked as worked.
try:
# worked_calls does a SELECT DISTINCT over the whole QSO
# table — this was blocking the event loop for 100-150ms
# (found via the looplag detector: the stack pointed at
# qso_db.worked_calls). The query is offloaded to a thread
# so SQLite doesn't freeze asyncio (audio, pings).
calls = await asyncio.to_thread(
qso_db.worked_calls, uid, is_admin_user)
return 200, {"calls": calls}
except Exception as e:
return 500, {"error": str(e)}
if p.startswith("/api/qsolog/export") and method == "GET":
# Export ADIF/CSV/EDI
# NOTE: query arrives as {key: [value]} (lists). We must
# extract single values, otherwise fmt=['adi'] != 'adi' and the
# endpoint returned 400 "Nieznany format". Filters must also be
# strings, not lists (otherwise export_adif gets the wrong types).
def _first(v):
return v[0] if isinstance(v, list) else v
_raw = {k: _first(v) for k, v in query.items()}
fmt = _raw.pop("format", "adi")
# Pass only KNOWN filters into the export - otherwise an
# unexpected query param (token, cache-buster, user_id vs
# filter_uid, etc.) flows through **filters into list_qsos and
# can raise a TypeError -> 500.
_allowed = {"from", "to", "call", "band", "mode", "filter_uid", "ids"}
f = {k: v for k, v in _raw.items() if k in _allowed and v}
# The frontend sends 'user_id' - map it to 'filter_uid' which the backend knows
if _raw.get("user_id"):
f["filter_uid"] = _raw["user_id"]
try:
if fmt == "adi":
content_out = qso_db.export_adif(uid, is_admin=is_admin_user, **f)
ct = "application/octet-stream"
elif fmt == "csv":
content_out = qso_db.export_csv(uid, is_admin=is_admin_user, **f)
ct = "text/csv"
else:
return 400, {"error": "Nieznany format"}
# IMPORTANT: we do NOT return an aiohttp Response object
# here - http_handler does `status, result = await
# self.api(...)` and unpacks the return value as a tuple
# (status, result). A Response object isn't a tuple ->
# ValueError -> 500 (which is why EVERY export used to give
# a 500, while import worked because it returned a normal
# tuple). We return a special "export" type that
# http_handler handles with download headers.
return "export", {
"body": content_out.encode("utf-8"),
"content_type": ct,
"filename": f"qso_export.{fmt}",
}
except Exception as e:
import traceback
tb = traceback.format_exc()
print(f"[export] 500 ERROR fmt={fmt} filters={f}:\n{tb}", flush=True)
return 500, {"error": str(e)}
# /api/qsolog/<id>
qso_id_match = p.startswith("/api/qsolog/") and len(p) > len("/api/qsolog/")
if qso_id_match:
qso_id = p[len("/api/qsolog/"):]
if method == "GET":
qso = qso_db.get_qso(uid, qso_id, is_admin=is_admin_user)
if qso: return 200, qso
return 404, {"error": "QSO nie znalezione"}
if method == "PUT":
if not self._has_perm(uid, role, "log"):
return 403, {"ok": False, "error": "Brak uprawnien (log QSO)"}
try:
# Same fallback as POST /api/qsolog above - update_qso()
# unconditionally overwrites my_call with whatever the
# request sends, so an edit from a client with an empty
# AppState.callsign would silently WIPE an already-correct
# operator field instead of just leaving it as-is.
if not (body.get("my_call") or "").strip():
_u_obj = self.find_user_by_id(uid)
body["my_call"] = ((_u_obj or {}).get("callsign")
or user.get("username", ""))
ok = qso_db.update_qso(uid, qso_id, body, is_admin=is_admin_user)
return (200, {"ok": True}) if ok else (404, {"error": "QSO nie znalezione"})
except ValueError as e:
return 400, {"ok": False, "error": str(e)}
if method == "DELETE":
if not self._has_perm(uid, role, "log"):
return 403, {"ok": False, "error": "Brak uprawnien (log QSO)"}
ok = qso_db.delete_qso(uid, qso_id, is_admin=is_admin_user)
return (200, {"ok": True}) if ok else (404, {"error": "QSO nie znalezione"})
if p == "/api/audio/config" and method == "POST":
# Missing a permission check here was a bug — any logged-in
# user (even a viewer) could remotely change the whole shared
# station's audio device and TX Volume (a multiplier also used
# by the FT8/FT4 encoder, i.e. affecting the real signal on
# air). Now requires the same "server settings" permission as
# the rest of this tab.
if not self._has_perm(uid, role, "settings"): return 403, {"error": "Brak uprawnien (ustawienia serwera)"}
rx = body.get("rxDevice")
tx = body.get("txDevice")
br = body.get("bitrate", 24000)
tv = body.get("txVolume")
tv_ssb = body.get("txVolumeSsb")
gain_mode_ssb = body.get("txGainModeSsb")
if "audio" not in self.cfg: self.cfg["audio"] = {}
if rx is not None: self.cfg["audio"]["rxDevice"] = rx
if tx is not None: self.cfg["audio"]["txDevice"] = tx
if br: self.cfg["audio"]["bitrate"] = int(br)
if tv is not None:
self.cfg["audio"]["txVolume"] = float(tv)
self.audio.cfg = self.cfg["audio"]
if tv_ssb is not None:
self.cfg["audio"]["txVolumeSsb"] = float(tv_ssb)
self.audio.cfg = self.cfg["audio"]
if gain_mode_ssb in ("manual", "agc"):
self.cfg["audio"]["txGainModeSsb"] = gain_mode_ssb
self.audio.cfg = self.cfg["audio"]
save_json(CFG_F, self.cfg)
# When Rust is active — send the new devices to ham_audio
rust = getattr(self, 'rust_audio', None)
if rust and rust._connected:
# RESILIENCE TO DIFFERENT CARDS (a product used by many
# clubs): check whether the selected card EXISTS on the
# system. When the user picked a card that's now gone
# (USB unplugged) or a config from a different machine
# points at a nonexistent card - warn, but don't block
# (Rust falls back to default_input). Without this, FT8
# goes silent with no explanation when the configured card doesn't exist.
try:
_devs = await rust.list_devices()
_rx_names = [d.get("name","") for d in _devs
if isinstance(d, dict) and d.get("is_input")]
_tx_names = [d.get("name","") for d in _devs
if isinstance(d, dict) and not d.get("is_input")]
if rx and not any(rx in n or n in rx for n in _rx_names):
print(f"[audio] WARNING: RX card '{rx}' doesn't exist on the system "
f"(available: {_rx_names}). Rust will use the default.", flush=True)
await self.hub.broadcast({"type": "audio_warning",
"msg": f"Karta RX '{rx}' niedostepna - uzyto domyslnej"})
if tx and not any(tx in n or n in tx for n in _tx_names):
print(f"[audio] WARNING: TX card '{tx}' doesn't exist on the system "
f"(available: {_tx_names}). Rust will use the default.", flush=True)
except Exception as _e:
print(f"[audio] Can't verify the devices: {_e}", flush=True)
# HOT-SWAP the device WITHOUT restarting the ham_audio
# process (requires a Rust build with hot-swap support:
# audio.rs RX_DEVICE_GEN + main.rs SetRxDevice). Rust
# reloads just the RX stream on the fly (dropping the old
# one = clean WASAPI, no phantoms). We do NOT restart the
# process, so:
# - no phantom WASAPI devices (the main problem)
# - FT8 RX does NOT die (the process stays alive, the decoder keeps running)
# - the browser's Opus decoder does NOT lose sync (continuous stream)
# - no dead control connection (same process)
print(f"[audio] Hot-swap devices RX='{rx}' TX='{tx}' (no restart)", flush=True)
try:
if rx is not None:
await rust.set_rx_device(rx)
if tx is not None:
await rust.set_tx_device(tx)
if br:
await rust._send_ctrl({"cmd": "SetBitrate", "bps": int(br)})
print("[audio] Hot-swap OK - Rust reloaded the stream on the fly", flush=True)
# ALSO restart the independent Python capture (audio_stream.py's
# own PyAudio stream, running in parallel to Rust/cpal) if the RX
# device changed - this is the ONLY thing WebRTC RX audio
# (webrtc_rx_audio.py, subscribe_rx_pcm) reads from, completely
# separate from Rust. Previously only Rust got the new device
# here, so a hot-swap silently orphaned WebRTC RX on the OLD
# device (still "connecting"/"connected" at the WebRTC level -
# ICE/DTLS doesn't care there's no audio - just permanently
# silent), while FT8 decode/CI-V kept working fine since those
# go through Rust, not this. Reported live as "no audio in the
# new version" right after a device hot-swap.
if rx is not None:
try:
self.audio.start_rx(device=rx)
print(f"[audio] Python RX capture also restarted on '{rx}'", flush=True)
except Exception as _e2:
print(f"[audio] Python RX capture restart error: {_e2}", flush=True)
except Exception as _e:
print(f"[audio] Hot-swap error: {_e}", flush=True)
await self.hub.broadcast({"type": "audio_warning",
"msg": "Zmiana karty nie powiodla sie"})
return 200, {"ok": True, "cfg": self.cfg["audio"]}
if p == "/api/audio/rx/start" and method == "POST":
# Safety: make sure PTT is off before starting audio
# The IC-7300 can enter TX via DATA VOX when the PC opens the USB audio channel
if self.rig.ptt:
print("[audio] SAFETY: resetting PTT before starting RX audio")
self.rig.ptt = False
try: await self.rig.set_ptt(False)
except: pass
await self.hub.broadcast({"type": "ptt", "ptt": False})
dev = body.get("device") or self.cfg.get("audio",{}).get("rxDevice")
bitrate = body.get("bitrate", 24000)
ok = self.audio.start_rx(device=dev, bitrate=bitrate)
if ok: await self.hub.broadcast({"type":"audio_status","rx":True,"device":dev})
return 200, {"ok":ok,"status":self.audio.get_status()}
if p == "/api/audio/rx/stop" and method == "POST":
self.audio.stop_rx()
return 200, {"ok":True}
return 404, {"error": f"Nieznany endpoint: {method} {p}"}
# ── WebSocket handler (aiohttp) ────────────────────────────────────────────
async def ws_handler(self, request: web.Request) -> web.WebSocketResponse:
# WebSocket with permessage-deflate compression - compresses JSON
# at the WS frame level. Very good for JSON (3-8x smaller), cost: ~5% CPU.
# heartbeat=30 - ping/pong every 30s, detects dead connections.
# compress=9 - zlib max level (best ratio).
# max_msg_size=8MB - the 4MB default may be too small for waterfall history.
# NOTE: Opus audio (binary bytes) also goes over this same WS.
# Compressing already-compressed Opus gives a minimal gain (2-5%)
# but the cost is small too. The client decides on its own whether
# to decode (per the permessage-deflate spec).
# compress=False: WebSocket per-message compression runs synchronously
# on the event loop for EVERY frame (looplag stack: _send_compressed_
# frame_sync/compress_sync). Scope + audio + deepcw_text push many
# frames per second, so even level 1 caused audible audio hitches when
# the buffer drained. Opus and scope arrays are already compact, so the
# bandwidth saving was marginal — not worth blocking the loop. Off.
#
# SECURITY FIX (found live 2026-09-02 via an unauthenticated probe
# against the public duckdns deployment): auth used to be checked
# AFTER ws.prepare() below, with "no/invalid token" silently
# falling through to role="viewer" instead of being refused. That
# meant ANY anonymous internet client could open this socket with
# no token at all and immediately receive the full "init" dump
# (operator callsign, STATION locator, live freq/mode/PTT, rig
# audio device names, who else is online) plus every subsequent
# broadcast (S-meter, waterfall scope frames, rotator position,
# TX meters) via self.hub.add(ws) below - a live, continuous,
# unauthenticated surveillance feed of the station, confirmed by
# directly connecting with no query string at all. Regular WRITE
# actions were still correctly gated by role checks elsewhere, so
# this was a confidentiality leak, not a control-takeover - but a
# real one, and on the public internet, not a lab. Matches how
# com_bridge_ws_handler already does it correctly (403 before ever
# upgrading to a WebSocket) - the plain /ws endpoint just never got
# the same treatment. Auth is now checked and enforced BEFORE
# ws.prepare() - an invalid/missing token gets a plain HTTP 403,
# no WebSocket upgrade, no data of any kind sent.
token = request.rel_url.query.get("token", "")
user = self._check_pw_ver(jwt_verify(token)) if token else None
if not user:
return web.Response(status=403, text="Unauthorized - missing or invalid token")
role = user.get("role", "viewer")
uid = user.get("id", "")
# FIX: the JWT only ever carries the role from LOGIN time (see the
# comment at _has_perm) - granting the "admin" GRANULAR permission
# afterward (Admin panel tab, ADMIN.js "admin" key) made _has_perm()
# checks pass, but every one of the many plain "role == 'admin'"
# checks scattered through this file stayed blind to it, since they
# never looked at permissions at all. Reported live: "granting
# someone the admin-panel permission should make them a full admin,
# not a partial one". Re-reading the live user record here (once,
# per connection) and promoting role locally is far safer than
# hunting down every "== admin" check in this file individually.
u_obj = self.find_user_by_id(uid) or {}
if role != "admin" and u_obj.get("permissions", {}).get("admin"):
role = "admin"
_ws_kwargs = dict(heartbeat=30, compress=False, max_msg_size=8*1024*1024)
try:
ws = web.WebSocketResponse(writer_limit=128*1024, **_ws_kwargs)
except TypeError:
ws = web.WebSocketResponse(**_ws_kwargs)
await ws.prepare(request)
await self.hub.add(ws)
# Register the user as online
if user:
self.online_users[ws] = {
"user_id": uid,
"username": user.get("username", ""),
"callsign": u_obj.get("callsign", user.get("username", "")),
"locator": u_obj.get("locator", ""),
"role": role,
"joined_at": time.time(),
}
# Notify everyone about the new online user
_lock_state_join = self._radio_lock_state()
_lock_state_join.pop("type", None) # see the comment at init (ws_handler)
await self.hub.broadcast({
"type": "online_update",
"online": self._online_users_state(),
**_lock_state_join,
})
# Auto-connect the DX Cluster if the user has it enabled
if self.dxcluster:
dx_cfg = u_obj.get("dxcluster", {})
if dx_cfg.get("auto_connect") and dx_cfg.get("host") and dx_cfg.get("login"):
# Check whether it's not already connected
existing = self.dxcluster.get_client(uid)
if not existing or not existing.is_connected():
asyncio.ensure_future(self.dxcluster.connect_user(
uid, dx_cfg["host"], int(dx_cfg.get("port", 7300)),
dx_cfg["login"], decrypt_secret(dx_cfg.get("password", ""))
))
try:
_lock_state = self._radio_lock_state()
_lock_state.pop("type", None) # see the comment below — critical fix
await ws.send_str(json.dumps({
"type": "init",
"freq": self.rig.freq, "mode": self.rig.mode,
"bandwidth": self.rig.bw, "ptt": self.rig.ptt,
"filterNum": self.rig.filter_num,
"preamp": self.rig.preamp, "attenuator": self.rig.attenuator,
"split": self.rig.split, "freqB": self.rig.freq_b,
"vfo": getattr(self.rig, "vfo", "VFOA"),
"callsign": CALLSIGN, "locator": LOCATOR,
# STATION locator (STATION_LOCATOR from the admin config) —
# where the antenna physically stands. NOT the logged-in
# operator's locator: in remote operation each user is
# somewhere else, but the rotor turns in one place. The
# azimuth to a correspondent must be computed from the
# antenna, otherwise a user in a different location would
# get a bearing computed from the user's own location for
# an antenna standing somewhere else.
"stationLocator": (self.cfg.get("stationLocator")
or LOCATOR or "").strip().upper(),
# OPERATOR locator (from their account) — for FT8 reports/the QSO log
"operatorLocator": (self.online_users.get(ws, {}).get("locator") or "").strip().upper(),
"rotators": [r.state() for r in self.rotators],
"sim": self.rig.sim,
"connected": self.rig.connected,
"rigDisconnectedSince": getattr(self.rig, "_disconnected_since", None),
"rigPowerOn": self._rig_power_on,
"ovf": getattr(self.rig, "ovf", False),
"models": HAMLIB_MODELS,
"rigs": self.cfg.get("rigs", []),
"modes": ["USB","LSB","AM","FM","CW","CW-R","RTTY","RTTY-R","USB-D","LSB-D","PKTUSB","PKTLSB"],
"enabledBands": self.cfg.get("enabledBands", []),
"enabledModes": self.cfg.get("enabledModes", []),
"modeFilters": self.cfg.get("modeFilters", {}),
# CRITICAL: _radio_lock_state() returns its OWN key
# "type": "radio_lock_state". Unpacking (**) that dict
# AFTER "type": "init" in the same dict literal made
# PYTHON SILENTLY OVERWRITE "type" with "radio_lock_state"
# — the final message sent to the frontend NEVER had
# type=="init"! So the 'init' case in ws.js never ran,
# msg.modes never arrived, and the MODE panel was empty
# after every fresh connection/refresh. We remove the
# colliding "type" key from lock_state BEFORE unpacking
# (see _lock_state.pop("type") above).
**_lock_state,
"online": self._online_users_state(),
"my_uid": uid,
}))
# FT8/FT4 automation state (QSO engine) — without this a
# freshly connected client (new tab, page refresh, reconnect)
# didn't know the backend's current state and showed the
# default "IDLE" even though the automation was actually
# already running a QSO with someone else — the panel showed
# unrealistic info about what/who we're transmitting to, until
# the next decode arrived and triggered a broadcast.
await ws.send_str(json.dumps({
"type": "auto_seq_status",
"enabled": self._auto_seq_enabled,
"state": self._qso_engine.state,
"partner": self._qso_engine.partner_call,
# FIX (2026-09-05, reported live: "automat nie odpowiada
# wolajacym, dziala tylko manualnie"): _ft8_operator_present
# is set False by "ft8_timer_expired" (the Tx Watchdog) and
# only ever set back True by an explicit "ft8_timer_confirm"
# from a client. FT8Timer's "expired, waiting for
# confirmation" state is 100% local JS module state - if the
# WS reconnects (page reload, network blip, laptop
# sleep/wake) BEFORE the operator clicks confirm, the
# frontend comes back with a fresh, un-expired timer showing
# nothing wrong, while the backend silently ignores every
# automatic reply forever (manual ft8_start_auto_qso never
# checked this flag, so clicking a row kept working - hence
# "works only manually"). Included here so a freshly
# (re)connected client can immediately re-show the confirm
# prompt instead of hiding a state the backend never forgot.
"operatorPresent": self._ft8_operator_present,
}))
await ws.send_str(json.dumps({
"type": "ft8_fake_split_status",
"enabled": self._fake_split_enabled,
"targetHz": 1500,
}))
# FT8 RX on/off state - same reasoning as auto_seq_status above:
# a freshly (re)connected client used to default its own RX
# button/watchdog to "off" regardless of the real backend
# state (which persists across reconnects and can also be
# true because a DIFFERENT session turned it on, or false
# because the idle watchdog auto-stopped it - see
# _stop_wsjtx_auto). The broadcast this mirrors already
# existed for every OTHER connected client; this just also
# covers the client that is (re)connecting right now.
await ws.send_str(json.dumps({
"type": "ft8_rx_status",
"enabled": self._ft8_rx_enabled,
}))
# Chat history - session-only (self._chat_history is an
# in-memory ring buffer, see the chat_send handler below), so
# a freshly (re)connected client needs this snapshot to see
# anything said before they joined/reloaded - same pattern as
# the other init-time snapshots above.
await ws.send_str(json.dumps({
"type": "chat_init",
"history": self._chat_history,
}))
async for msg in ws:
if msg.type == aiohttp.WSMsgType.BINARY:
data = msg.data
if data and len(data) > 1 and data[0] == 0xA2:
if not self.audio.tx_active:
dev = self.cfg.get("audio", {}).get("txDevice")
self.audio.start_tx(device=dev)
print(f"[audio] TX auto-start dev={dev}")
await self.audio.feed_tx(data[1:])
self.touch_activity(uid)
elif data and len(data) > 5 and data[0] == 0xC1:
# DeepCW: [0xC1][src_rate 4B LE][PCM float32]
# The decoder only runs while the browser is
# SENDING audio (the CW panel is open) — closing
# the panel = zero CPU.
try:
import struct as _st
_sr = _st.unpack_from("<I", data, 1)[0]
_pcm = np.frombuffer(data[5:], dtype=np.float32)
if deepcw_engine is not None:
await deepcw_engine.feed(
_pcm, _sr, broadcast_fn=self.hub.broadcast)
except Exception as _e:
print(f"[deepcw] feed error: {_e}", flush=True)
elif msg.type == aiohttp.WSMsgType.TEXT:
try:
data = json.loads(msg.data)
t = data.get("type","")
print(f"[ws] TEXT t={t!r}", flush=True)
if t == "audio_start":
# Audio goes directly through the Rust WSS — Python doesn't proxy it
await ws.send_str(json.dumps({"type":"audio_ready","status":{"running":True,"rust":True}}))
elif t == "audio_stop":
pass # audio goes directly over the Rust WS, nothing for Python to release
else:
# Touch activity on any real operator action — an
# EXCLUDE list, not an include list. FIX: this
# used to be a hand-picked whitelist of ~12
# message types (freq/mode/ptt/rig_slider/...)
# that only covered plain radio-control actions.
# It never got extended when the FT8 panel grew
# its own message types (ft8_qsy, ft8_tx,
# ft8_start_auto_qso, ft8_toggle_auto_seq, chat_send,
# ...) — so a user actively running the FT8
# automation (CQ, answering callers, clicking
# bands) never touched last_activity at all.
# last_activity then stayed frozen at whatever
# _lock_radio() set it to on acquisition, so the
# idle watchdog (_radio_lock_watchdog) effectively
# measured "time since PRZEJMIJ TRX", not real
# idleness — exactly the reported symptom. Only
# exclude messages that fire on their own
# (automatic, not user-driven): 'ping' (client
# heartbeat, every 30s) and 'subscribe' (sent
# once on connect/reconnect).
if t not in ("ping", "subscribe"):
self.touch_activity(uid)
await self._ws_msg(data, ws, role)
except Exception as e:
print(f"[ws] {e}")
elif msg.type in (aiohttp.WSMsgType.ERROR, aiohttp.WSMsgType.CLOSE):
break
except Exception as e:
print(f"[ws] connection error: {e}", flush=True)
finally:
await self.hub.remove(ws)
# Stop TRACKING (but do NOT close) any WebRTC RX audio sender
# this client had open. The old code called _rx_sender.close()
# here, on the theory that a lost connection = the client is
# gone. WRONG: the control WS ('ws', this connection) and the
# RX WebRTC media connection are two SEPARATE peer connections
# - only signaling (webrtc_rx_start/answer/ice) goes over the
# control WS, the actual audio stream is its own independent
# WebRTC/UDP path once established. On a real remote link the
# control WS reconnects on its own periodically (ordinary
# network jitter, unrelated to whether audio itself is
# healthy) - closing RX audio every single time this happened
# was reported live as "audio never starts" (confirmed in the
# log: "close() called from ws_handler", not from an actual
# WebRTC failure - recv() had 1000+ successful calls with zero
# errors right before being killed). The old WS-based RX audio
# path never had this problem because it was ALWAYS a fully
# separate connection to ham_audio.exe, with no such coupling.
# A genuinely abandoned connection (the browser tab really
# closed, no reconnect ever comes) still gets cleaned up - its
# own WebRTCAudioSender.on_state_change transitions to 'failed'
# once WebRTC's own ICE consent-freshness checks detect the
# peer is truly gone, and closes itself independently of this
# bookkeeping dict.
self._webrtc_rx_senders.pop(ws, None)
# Remove from the online list and notify the others
self.online_users.pop(ws, None)
self.radio_requests.pop(uid, None)
# Check whether the user has another active WS (e.g. open in
# another tab/laptop+phone). If YES - don't stop WSJT-X or
# release the radio (since they're still online).
still_online = any(u.get("user_id") == uid for u in self.online_users.values())
# FIX (2026-10-22, live report + console screenshots: "serwer
# odebral mi posiadanie trx, zastopowal ft8 waterfall i
# dekodowanie"): this used to release the radio / stop FT8 the
# INSTANT the user's last WS dropped. The control WS drops and
# reconnects on its own on ordinary network jitter (see the
# WebRTC comment above - the screenshot showed exactly such a
# Disconnected -> Connected a second apart), so a 1-second blip
# took the radio away from an operator who never left. Now the
# release waits a grace period and is skipped if they're back.
if uid and not still_online and (
self.radio_lock["user_id"] == uid or uid == self._ft8_rx_owner_uid):
asyncio.ensure_future(self._release_after_disconnect_grace(uid))
# CW decoder viewer cleanup: a disconnect (logout, closed tab) must
# drop this client from the viewer set, exactly like closing the
# window. If it was the last viewer, stop the decoder so the server
# doesn't keep decoding (and printing to CMD) with nobody watching —
# this is the "server keeps decoding after logout" bug.
if hasattr(self, "_cw_viewers") and ws in self._cw_viewers:
self._cw_viewers.discard(ws)
if not self._cw_viewers:
_cwt = getattr(self, "_cw_task", None)
if _cwt:
_cwt.cancel()
self._cw_task = None
if self.audio:
self.audio.cw_rx_enabled = False
if deepcw_engine is not None:
deepcw_engine.reset()
print("[deepcw] decoder disabled (last viewer disconnected)",
flush=True)
_lock_state_disconnect = self._radio_lock_state()
_lock_state_disconnect.pop("type", None) # see the comment at init (ws_handler)
await self.hub.broadcast({
"type": "online_update",
"online": self._online_users_state(),
**_lock_state_disconnect,
})
return ws
async def _release_after_disconnect_grace(self, uid, grace_s: float = 60.0):
"""Release the radio / stop FT8 for a user whose LAST WS dropped -
but only if they haven't reconnected within grace_s (see the
comment at the call site in ws_handler's finally block)."""
await asyncio.sleep(grace_s)
if any(u.get("user_id") == uid for u in self.online_users.values()):
print(f"[radio_lock] uid={uid!r} reconnected within {grace_s:.0f}s "
f"- keeping the radio/FT8", flush=True)
return
if self.radio_lock["user_id"] == uid:
print(f"[radio_lock] uid={uid!r} gone for {grace_s:.0f}s - releasing the radio",
flush=True)
self._release_radio() # also stops WSJT-X if they were the owner
_st = self._radio_lock_state()
_st.pop("type", None)
await self.hub.broadcast({"type": "online_update",
"online": self._online_users_state(), **_st})
elif uid == self._ft8_rx_owner_uid and self._ft8_rx_enabled:
await self._stop_wsjtx_auto("rozłączenie sesji")
async def com_bridge_ws_handler(self, request: web.Request) -> web.WebSocketResponse:
"""
WebSocket endpoint /ws/com-bridge - for Windows EXE clients.
Authorization: Bearer token in the query string (?token=...) OR in
the Sec-WebSocket-Protocol header as 'bearer,<token>' (WS doesn't
allow a standard Authorization header).
After authorization, control is handed to ComBridgeWs.handle_client(),
which handles the protocol (hello/data/ping/etc).
"""
# Authorization - token from the query string
token = request.query.get('token', '')
if not token:
# Alternative: Sec-WebSocket-Protocol 'bearer,<token>'
protos = request.headers.get('Sec-WebSocket-Protocol', '').split(',')
protos = [p.strip() for p in protos]
if len(protos) >= 2 and protos[0] == 'bearer':
token = protos[1]
user = self._verify_token(token) if token else None
if not user:
return web.Response(status=403, text="Unauthorized - missing or invalid token")
# ── VIEWER HAS NO ACCESS TO THE BRIDGES ─────────────
# A viewer uses ONLY the web UI (where they can change freq with
# the lock, but not transmit). External CAT apps (CW Skimmer, HRD,
# Logger32) via the COM Bridge are reserved for operators and the
# admin. A viewer trying to connect gets refused - no ports are created for them.
_role = user.get('role', 'viewer')
if _role != 'admin':
_u_obj_bridge = self.find_user_by_id(user.get('id', '')) or {}
if _u_obj_bridge.get('permissions', {}).get('admin'):
_role = 'admin'
if _role == 'viewer':
print(f"[com_bridge_ws] REFUSED viewer "
f"{user.get('username','?')} (id={user.get('id','?')}) - "
f"bridges are for operators/admin only", flush=True)
return web.Response(
status=403,
text="COM Bridge access is for operators only. "
"As a viewer, use the web panel."
)
# Get this user's port config from users.json
# The JWT payload uses 'id' (consistent with the rest of the code)
user_id = user.get('id') or user.get('user_id', '')
user_full = next((u for u in self.users if u.get('id') == user_id), {})
# FIXED 2 CI-V PORTS FOR ALL USERS.
# Every user gets exactly 2 CI-V ports (for CW Skimmer + HRD, or
# any two CAT apps). There's no per-user configuration - the
# section in the web UI is informational only. CI-V addresses
# (E1, E2) are assigned automatically in build_assignments.
com_config = [
{'service': 'civ', 'baud': 19200},
{'service': 'civ', 'baud': 19200},
]
# compress=False because permessage-deflate corrupts small CI-V
# messages (10-byte frames). Traffic is small (~a few hundred
# B/s), compression is unnecessary and could break communication.
ws = web.WebSocketResponse(heartbeat=30, compress=False)
await ws.prepare(request)
# Delegate to ComBridgeWs - it holds the client state, protocol,
# rate limiting, validation.
try:
await self.com_bridge_ws.handle_client(ws, user, com_config)
except Exception as e:
print(f"[com_bridge_ws] handle_client error: {e}")
finally:
if not ws.closed:
await ws.close()
return ws
def _verify_token(self, token: str) -> dict:
"""
Verifies a JWT token and returns the user data, or None.
Used in com_bridge_ws_handler (since a plain WS has no Bearer header).
Returns a dict compatible with the rest of the code: {id, callsign, role}.
"""
if not token:
return None
try:
payload = self._check_pw_ver(jwt_verify(token))
if not payload:
return None
# The JWT payload uses 'id' (see webapp:1300 - jwt_sign({"id": u["id"], ...}))
uid = payload.get('id') or payload.get('user_id') or payload.get('sub')
u = next((x for x in self.users if x.get('id') == uid), None)
if not u:
return None
return {
'id': u.get('id'),
'user_id': u.get('id'), # alias for compatibility
'callsign': u.get('callsign', ''),
'role': u.get('role', 'user'),
}
except Exception as e:
print(f"[com_bridge_ws] token verification error: {e}")
return None
async def hamlib_ws_handler(self, request: web.Request) -> web.WebSocketResponse:
"""
WebSocket endpoint /hamlib — Hamlib TCP tunnel for remote WSJT-X.
The client (wsjtx_bridge.py) connects here instead of directly to port 4532.
Hamlib commands arrive as WS text, responses go back as WS text.
SECURITY FIX (found live 2026-09-02, same audit as ws_handler):
this endpoint had NO authentication at all - anyone on the internet
could open it with no token. HamlibSession._can_control() only
gates WRITE commands (SET_FREQ/SET_MODE/SET_PTT) behind "is
radio_lock currently held by ANYONE", not "is THIS connection the
actual lock holder" - so while the legitimate operator is actively
using the radio (the normal case, not an edge case), an anonymous
client here could inject real control commands and key the radio,
confirmed live: the handshake alone succeeds with zero auth
(deliberately did NOT send an actual Hamlib command in the live
test - the operator held the lock at the time, so it would have
actually gone through). Now requires the same ?token= a logged-in
session already has, and refuses viewer role - matching
com_bridge_ws_handler's existing, correct pattern for the other
external-CAT-software bridge.
"""
token = request.query.get('token', '')
user = self._verify_token(token) if token else None
if not user:
return web.Response(status=403, text="Unauthorized - missing or invalid token")
_role = user.get('role', 'viewer')
if _role != 'admin':
_u_obj_hl = self.find_user_by_id(user.get('id', '')) or {}
if _u_obj_hl.get('permissions', {}).get('admin'):
_role = 'admin'
if _role == 'viewer':
return web.Response(
status=403,
text="Hamlib bridge access is for operators/admin only. "
"As a viewer, use the web panel.")
ws = web.WebSocketResponse(heartbeat=30, compress=9)
await ws.prepare(request)
peer = request.remote
print(f"[hamlib-ws] WSJT-X bridge connected: {peer} (user={user.get('username','?')})")
from hamlib_server import HamlibSession
# Build a "virtual" reader/writer for HamlibSession
# using asyncio queues instead of TCP
cmd_queue = asyncio.Queue()
resp_queue = asyncio.Queue()
class WSReader:
async def readline(self):
line = await cmd_queue.get()
return (line + '\n').encode()
class WSWriter:
def get_extra_info(self, key):
return peer if key == 'peername' else None
def write(self, data): pass
async def drain(self): pass
def close(self): pass
# Run HamlibSession in a separate task
session = HamlibSession(self.rig, self.hub, WSReader(), WSWriter(),
"ws-bridge", app=self)
async def session_loop():
try:
async for msg in ws:
if msg.type == aiohttp.WSMsgType.TEXT:
cmd = msg.data.strip()
if not cmd:
continue
# Process the command and reply over WS
response = await session._handle(cmd)
await ws.send_str(response)
elif msg.type in (aiohttp.WSMsgType.ERROR, aiohttp.WSMsgType.CLOSE):
break
except Exception as e:
print(f"[hamlib-ws] Error: {e}")
finally:
print(f"[hamlib-ws] Disconnected: {peer}")
await session_loop()
return ws
def _current_rig_id(self) -> str:
"""ID of the currently connected radio in config.json['rigs'] (or '0' if none)."""
rigs = self.cfg.get("rigs") or []
model = str(getattr(self.rig, "model", ""))
for r in rigs:
if str(r.get("model", "")) == model:
return str(r.get("id", "0"))
return str(rigs[0].get("id", "0")) if rigs else "0"
def _get_enabled_features(self, rig_id: str) -> dict:
"""The admin whitelist for the given radio from config.json (or the default)."""
rigs = self.cfg.get("rigs") or []
for r in rigs:
if str(r.get("id", "")) == str(rig_id):
ef = r.get("enabledFeatures")
if ef:
return ef
return default_enabled_features()
def _get_enabled_dynamic(self, rig_id: str) -> dict:
"""The admin whitelist for dynamic actions/sliders (per dynamic_id)."""
rigs = self.cfg.get("rigs") or []
for r in rigs:
if str(r.get("id", "")) == str(rig_id):
return r.get("enabledDynamic") or {}
return {}
async def _get_rig_features(self, role: str) -> dict:
"""
GET /api/rig/features
- admin: full list (static features + dynamic actions/sliders)
with capabilities/enabled/effective (for the admin panel)
- viewer/operator: only effective (static) + allowed dynamic ones
"""
try:
capabilities = await self.rig.get_capabilities()
except Exception as e:
print(f"[features] get_capabilities error: {e}")
capabilities = {"actions": [], "sliders": [], "raw_caps": {}}
# Compatibility: if get_capabilities returned the old format (a flat bool dict)
if "raw_caps" not in capabilities and "actions" not in capabilities:
capabilities = {"actions": [], "sliders": [], "raw_caps": capabilities}
rig_id = self._current_rig_id()
enabled = self._get_enabled_features(rig_id)
enabled_dyn = self._get_enabled_dynamic(rig_id)
if role == "admin":
return {
"ok": True,
"rigId": rig_id,
"model": getattr(self.rig, "model", ""),
"sim": self.rig.sim,
"features": features_for_admin(capabilities, enabled),
"dynamic": dynamic_for_admin(capabilities, enabled_dyn),
}
else:
eff = effective_features(capabilities, enabled)
active = [
{"id": f["id"], "label": f["label"], "icon": f["icon"], "group": f["group"]}
for f in FEATURES if eff.get(f["id"])
]
dyn = effective_dynamic(capabilities, enabled_dyn)
return {"ok": True, "rigId": rig_id, "active": active,
"actions": dyn["actions"], "sliders": dyn["sliders"]}
async def _set_rig_features(self, body: dict) -> dict:
"""
POST /api/rig/features
body: {rigId?: str, enabledFeatures?: {feature_id: bool},
enabledDynamic?: {dynamic_id: bool}}
Admin-only — saves the whitelists to config.json['rigs'][i].
"""
rig_id = str(body.get("rigId") or self._current_rig_id())
new_enabled = body.get("enabledFeatures") or {}
new_enabled_dyn = body.get("enabledDynamic") or {}
# Validate the static ones — only known feature_ids
known_ids = {f["id"] for f in FEATURES}
new_enabled = {k: bool(v) for k, v in new_enabled.items() if k in known_ids}
# Dynamic: only validate the type (bool), ids are dynamic so we don't filter them
new_enabled_dyn = {k: bool(v) for k, v in new_enabled_dyn.items()}
if not self.cfg.get("rigs"):
self.cfg["rigs"] = []
rigs = self.cfg["rigs"]
found = False
for r in rigs:
if str(r.get("id", "")) == rig_id:
merged = dict(r.get("enabledFeatures") or default_enabled_features())
merged.update(new_enabled)
r["enabledFeatures"] = merged
merged_dyn = dict(r.get("enabledDynamic") or {})
merged_dyn.update(new_enabled_dyn)
r["enabledDynamic"] = merged_dyn
found = True
break
if not found and rigs:
merged = dict(rigs[0].get("enabledFeatures") or default_enabled_features())
merged.update(new_enabled)
rigs[0]["enabledFeatures"] = merged
merged_dyn = dict(rigs[0].get("enabledDynamic") or {})
merged_dyn.update(new_enabled_dyn)
rigs[0]["enabledDynamic"] = merged_dyn
rig_id = str(rigs[0].get("id", "0"))
save_json(CFG_F, self.cfg)
print(f"[features] admin saved enabledFeatures for rig={rig_id}: "
f"{new_enabled} | enabledDynamic: {new_enabled_dyn}")
# Broadcast the new effective set to all clients (the panel will refresh)
try:
capabilities = await self.rig.get_capabilities()
except Exception:
capabilities = {"actions": [], "sliders": [], "raw_caps": {}}
if "raw_caps" not in capabilities and "actions" not in capabilities:
capabilities = {"actions": [], "sliders": [], "raw_caps": capabilities}
enabled = self._get_enabled_features(rig_id)
enabled_dyn = self._get_enabled_dynamic(rig_id)
eff = effective_features(capabilities, enabled)
active = [
{"id": f["id"], "label": f["label"], "icon": f["icon"], "group": f["group"]}
for f in FEATURES if eff.get(f["id"])
]
dyn = effective_dynamic(capabilities, enabled_dyn)
await self.hub.broadcast({"type": "rig_features", "rigId": rig_id,
"active": active,
"actions": dyn["actions"], "sliders": dyn["sliders"]})
return {"ok": True, "rigId": rig_id,
"enabledFeatures": self._get_enabled_features(rig_id),
"enabledDynamic": self._get_enabled_dynamic(rig_id)}
async def _cloudlog_push_qso(self, user_id: str, qso: dict):
"""
Push a QSO to CloudLog/WaveLog if the user has the API configured.
IMPORTANT: the Cloudlog API accepts a QSO ONLY as an ADIF string in
the 'string' field (type='adif'). Sending JSON fields (call, band,
mode...) does NOT WORK - Cloudlog responds 200 but the QSO never
reaches the log. The profile field is named 'station_profile_id' (not 'station_id').
"""
try:
u = self.find_user_by_id(user_id)
if not u: return
cl = (u or {}).get("cloudlog", {})
url = cl.get("url", "").rstrip("/")
api_key = decrypt_secret(cl.get("apiKeyQso", ""))
station = cl.get("stationId", 1)
if not url or not api_key: return
# Don't duplicate /index.php if the user already typed it
base = url[:-10].rstrip("/") if url.endswith("/index.php") else url
import aiohttp as _aiohttp
adif = qso_to_adif(qso)
payload = {
"key": api_key,
"station_profile_id": str(station),
"type": "adif",
"string": adif,
}
async with _aiohttp.ClientSession() as sess:
async with sess.post(
f"{base}/index.php/api/qso", json=payload,
headers={"Content-Type": "application/json",
"Accept": "application/json"},
timeout=_aiohttp.ClientTimeout(total=8)
) as resp:
txt = await resp.text()
if resp.status in (200, 201):
# Cloudlog returns e.g. {"status":"created","adif_count":1}
# or {"status":"failed","reason":"..."}
try:
rdata = await resp.json(content_type=None)
except Exception:
rdata = {}
if isinstance(rdata, dict) and rdata.get("status") == "failed":
print(f"[cloudlog] {qso.get('call')}: rejected — "
f"{rdata.get('reason', '?')}", flush=True)
return
print(f"[cloudlog] {qso.get('call')} sent OK", flush=True)
# Save the Cloudlog ID if one was returned
cl_id = ""
if isinstance(rdata, dict):
cl_id = rdata.get("id") or rdata.get("qso_id") or ""
if cl_id:
try:
await asyncio.to_thread(
self._save_cloudlog_id, qso["id"], str(cl_id))
except Exception:
pass
else:
print(f"[cloudlog] {qso.get('call')}: HTTP {resp.status} "
f"— {txt[:100]}", flush=True)
except Exception as e:
print(f"[cloudlog] push QSO error: {e}", flush=True)
def _save_cloudlog_id(self, qso_id: str, cloudlog_id: str):
"""Save the Cloudlog ID against the QSO (sync - called via to_thread)."""
with qso_db._conn_lock:
conn = qso_db._get_conn()
conn.execute("UPDATE qso SET cloudlog_id=? WHERE id=?",
(cloudlog_id, qso_id))
conn.commit()
def _is_band_allowed(self) -> bool:
"""Check whether the radio's current frequency is within an allowed band.
Uses self._BAND_RANGES (the same table as _get_band_for_freq /
cross-band split) - this used to have a separate, independently
maintained dict that drifted apart from _BAND_RANGES for
160m/60m/6m (different boundaries in two places in the same file
for the same safety guard).
"""
freq = self.rig.freq
enabled = self.cfg.get("enabledBands", None)
if not enabled: # no configuration = all bands allowed
return True
for band in enabled:
lo, hi = self._BAND_RANGES.get(band, (0, 0))
if lo <= freq <= hi:
return True
return False
# Shared band table for all checks (avoids duplication) - 160m/60m/6m
# are deliberately NARROWER (the real PL/EU allocation) - this is the
# table used, among others, by the TX guard (_is_band_allowed), so it's
# safer to be conservative than to risk transmitting outside the real
# band allocation.
#
# MUST cover EVERY band from any radio profile in rigs/civ_profiles.py
# (the IC-9100/IC-9700 offer "23cm" in their BANDS) - otherwise the TX
# guard rejects a band as "not allowed" even though the admin
# explicitly enabled it and the radio physically has it. 23cm added
# with the same range as _BAND_23CM in civ_profiles.py (1240-1300 MHz).
# 13cm NOT added - no current radio profile offers it yet in "bands"
# (see _BAND_13CM in civ_profiles.py - defined, but unused), so adding
# it now would be guessing without any real hardware backing it.
_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),
'23cm': (1240000000,1300000000),
}
def _get_band_for_freq(self, hz: int) -> str | None:
"""Return the band name for a frequency, or None if outside all bands."""
for band, (lo, hi) in self._BAND_RANGES.items():
if lo <= hz <= hi:
return band
return None
async def _is_split_cross_band(self) -> tuple[bool, str, str]:
"""Check whether split is cross-band (VFO-A and VFO-B in different bands).
Returns (is_cross_band, band_a, band_b) — is_cross_band=True when
split is active and the bands differ. band_a/b are band names or 'poza pasmem'.
This guard protects against damaging the radio/antenna —
transmitting on a different band than the one the ATU tuned for
can damage the final amplifier or an antenna with the wrong impedance.
"""
# DIAGNOSTIC (2026-09-02): live-reported the VFO-B swap-read never
# ran despite split being on in the UI - this pins down exactly
# what this function sees at call time (split state, driver type,
# profile flag) instead of guessing further.
print(f"[rig] _is_split_cross_band: split={getattr(self.rig, 'split', None)!r} "
f"is_civ={isinstance(self.rig, CivRig)} "
f"has_vfo_b_read={self.rig.profile.get('has_vfo_b_read', 'N/A') if isinstance(self.rig, CivRig) else 'N/A (not CivRig)'}",
flush=True)
if not getattr(self.rig, 'split', False):
return False, '', ''
# Capture VFO-A's frequency BEFORE touching anything - FIX
# (live-seen 2026-09-02): this used to be read AFTER the VFO-B
# swap-read below, so on a has_vfo_b_read=False radio, self.rig.freq
# got read back showing VFO-B's frequency instead of VFO-A's
# (real report: freq_a=14331000 VFO-A / split target 21MHz, but the
# check logged freq_a=21231000 == freq_b - the swap's own read had
# already landed in self.rig.freq by the time this ran), making a
# genuine cross-band split look like "same band" and pass through
# unblocked. Freezing it here, before the swap ever starts, removes
# any chance of the swap's own traffic clobbering this value.
freq_a = self.rig.freq
# A radio that can't read VFO-B over CI-V at all (has_vfo_b_read=
# False - e.g. IC-746, command 0x25/0x26 is newer-generation-only
# and gets NG'd every time) leaves self.rig.freq_b frozen at
# whatever it was last (its init default, or stale) - comparing
# against THAT would be trusting a value never actually read from
# the radio. For these radios do a FRESH read right now, via a
# brief VFO-A/B swap (civ.py's get_freq_b_fresh - the same fallback
# Hamlib itself uses for pre-0x25 Icoms) instead of trusting the
# background-polled value. This only runs right before allowing a
# split TX, never continuously, and never while self.ptt is
# already True (civ.py's own guard).
if isinstance(self.rig, CivRig) and not self.rig.profile.get("has_vfo_b_read", True):
fresh_fb = None
if hasattr(self.rig, "get_freq_b_fresh"):
try:
fresh_fb = await self.rig.get_freq_b_fresh()
except Exception as e:
print(f"[rig] get_freq_b_fresh error: {e}", flush=True)
if fresh_fb is None:
# Radio didn't cooperate - we genuinely don't know band_b,
# so refuse to claim "same band" rather than guess.
return True, 'nieznane (radio nie odczytuje VFO-B)', 'nieznane (radio nie odczytuje VFO-B)'
self.rig.freq_b = fresh_fb
freq_b = getattr(self.rig, 'freq_b', freq_a)
band_a = self._get_band_for_freq(freq_a) or 'poza pasmem'
band_b = self._get_band_for_freq(freq_b) or 'poza pasmem'
print(f"[rig] _is_split_cross_band result: freq_a={freq_a} ({band_a}) "
f"freq_b={freq_b} ({band_b}) cross={band_a != band_b}", flush=True)
return (band_a != band_b), band_a, band_b
def _can_control_radio(self, ws, role: str) -> tuple[bool, str]: