server.py
Rama main · réplica de lectura del repositorio original.
#!/usr/bin/env python3
"""server.py — Ham Radio Control Server — entry point"""
import os as _os
# Limit BLAS/numpy threads before importing numpy (see launcher.py) — so
# audio FFT doesn't grab every core and stall the audio.
for _v in ("OMP_NUM_THREADS", "OPENBLAS_NUM_THREADS", "MKL_NUM_THREADS",
"NUMEXPR_NUM_THREADS", "VECLIB_MAXIMUM_THREADS"):
_os.environ.setdefault(_v, "2")
import asyncio, sys, socket, subprocess, pathlib
def _ensure_deps():
missing = []
try: import aiohttp
except ImportError: missing.append("aiohttp>=3.9")
try: import serial
except ImportError: missing.append("pyserial>=3.5")
try: import cryptography
except ImportError: missing.append("cryptography")
try: import websockets
except ImportError: missing.append("websockets")
try: import opuslib
except ImportError:
try: import opus
except ImportError: missing.append("opuslib")
if missing:
print(f"[setup] Installing: {', '.join(missing)}")
subprocess.run(
[sys.executable,"-m","pip","install"]+missing+
["--quiet","--no-warn-script-location"],
check=True
)
# Audio — optional
try:
import opuslib, opuslib.api
print("[setup] opuslib OK")
except Exception:
subprocess.run(
[sys.executable,"-m","pip","install","opuslib",
"--quiet","--no-warn-script-location"],
capture_output=True, timeout=120
)
try:
import pyaudio
print("[setup] pyaudio OK")
except ImportError:
if sys.platform == "win32":
r = subprocess.run(
[sys.executable,"-m","pip","install","pyaudio",
"--only-binary",":all:","--quiet","--no-warn-script-location"],
capture_output=True, timeout=60
)
if r.returncode != 0:
try:
subprocess.run(
[sys.executable,"-m","pip","install","pipwin",
"--quiet","--no-warn-script-location"],
capture_output=True, timeout=60
)
subprocess.run(
[sys.executable,"-m","pipwin","install","pyaudio"],
capture_output=True, timeout=120
)
except Exception:
pass
else:
subprocess.run(
[sys.executable,"-m","pip","install","pyaudio",
"--quiet","--no-warn-script-location"],
capture_output=True, timeout=120
)
_ensure_deps()
# Imports after _ensure_deps (aiohttp must already be installed)
import aiohttp.web as web
from config import PORT, ADMIN_PW, HAMLIB_PORT, DATA, VERBOSE
import os as _os
# Which interface to listen on. Default 0.0.0.0 (all interfaces) so operators on
# the LAN and the Cloudflare tunnel can both reach the server — this is required
# for normal club use. A cautious admin who only wants remote access via the
# tunnel (and nothing exposed on the LAN) can set HAM_BIND_HOST=127.0.0.1, which
# limits listening to localhost; the tunnel still works because cloudflared
# connects to localhost. Documented for clubs so 0.0.0.0 is a known choice.
BIND_HOST = _os.environ.get("HAM_BIND_HOST", "0.0.0.0")
import webapp
from webapp import App
@web.middleware
async def _security_headers_middleware(request, handler):
"""
Adds standard hardening headers to every HTTP response. Found missing
entirely during a live security pass 2026-09-02 (curl against the
public duckdns deployment showed zero of these on any response).
Deliberately does NOT add script-src/style-src to the CSP below - the
frontend relies heavily on inline onclick=/style= attributes
throughout (201+838 in index.html, 48+30 in mobile.html, counted
2026-09-03), so locking those two directives down would mean
rewriting every single one to addEventListener/CSS classes first -
a real, dedicated refactor, not a header bolted on blind. The four
CSP directives actually shipped below don't touch inline
scripts/styles at all, so they carry none of that risk. These are
all safe defaults with no such risk:
- X-Content-Type-Options: stops the browser from guessing a
different content type than what the server declared (MIME-sniffing).
- X-Frame-Options: DENY - this control panel is never meant to be
embedded in another site's frame (clickjacking protection).
- Referrer-Policy: don't leak the full URL (which can carry a
session token in ?token=...) to a third-party site via the
Referer header if a link is ever clicked out.
- Strict-Transport-Security: the app is HTTPS-only already (see
launcher.py) - tells the browser to never even try plain HTTP for
this host.
- Content-Security-Policy (added 2026-09-03, the safe subset only):
object-src 'none' blocks <object>/<embed>/<applet> (this app uses
none); base-uri 'self' stops an injected <base> tag from
hijacking every relative URL/script src on the page (verified no
<base> tag is used legitimately); form-action 'self' stops a
compromised page from redirecting a form submit to an attacker
domain (verified no form here submits externally);
frame-ancestors 'none' is the modern equivalent of the
X-Frame-Options above. None of these four restrict script-src or
style-src, so none of them touch the inline onclick=/style=
attributes mentioned above - zero risk of breaking existing UI.
"""
resp = await handler(request)
resp.headers.setdefault("X-Content-Type-Options", "nosniff")
resp.headers.setdefault("X-Frame-Options", "DENY")
resp.headers.setdefault("Referrer-Policy", "no-referrer")
resp.headers.setdefault("Strict-Transport-Security", "max-age=31536000; includeSubDomains")
resp.headers.setdefault("Content-Security-Policy",
"object-src 'none'; base-uri 'self'; "
"form-action 'self'; frame-ancestors 'none'")
return resp
async def amain():
app = App()
loop = asyncio.get_running_loop()
app.hub.set_loop(loop)
# Audio RX is always enabled from server startup (previously required a
# manual button click in the UI, which has since been removed — see
# settings.js/index.html). Uses the saved device from config, the same
# way the old POST /api/audio/rx/start endpoint did.
try:
_rx_dev = app.cfg.get("audio", {}).get("rxDevice")
if app.audio.start_rx(device=_rx_dev):
print(f"[audio] RX auto-start OK (device={_rx_dev or 'default'})")
else:
print("[audio] RX auto-start FAILED — check the audio device")
except Exception as e:
print(f"[audio] RX auto-start error: {e}")
async def _initial_rig_connect():
await app.rig.connect(app.cfg)
await app._refresh_caps_cache()
# Auto-start the waterfall scope when a REAL radio is connected (not
# SIM). Without this, _enable_scope was never called -> the radio
# never sent 0x27 frames -> the waterfall only showed the
# simulation. The frontend has startScope() but nothing was calling
# it - hence the "waterfall stopped working" regression. The
# backend now enables the scope itself after connecting.
try:
# hasattr(scope_start) alone isn't enough - EVERY CivRig
# instance has the method regardless of model.
# CIV_NATIVE_MODELS (added 2026-09-02 for IC-746) covers CI-V
# radios with NO scope hardware at all - without this check,
# startup would issue 0x27 scope commands to a radio that
# doesn't understand them (live-seen: "scope requires 115200!"
# + "radio REJECTED (NG)" spam on an IC-746 at 19200bd, right
# before ALC/PWR/SWR meter reads started failing too - the same
# capabilities gate already added to webapp.py's
# _on_rig_reconnected() for the reconnect case, missed here for
# the initial-startup connect).
_has_scope = getattr(app.rig, "profile", {}).get("capabilities", {}).get("scope", False)
if not app.rig.sim and _has_scope and hasattr(app.rig, "scope_start"):
await asyncio.sleep(1.0) # give the radio a moment after connect
# scope_start does blocking writes to the port (time.sleep) —
# run it in a thread so it doesn't freeze the loop on startup (looplag).
await asyncio.to_thread(app.rig.scope_start)
print("[rig] scope auto-enabled after connecting to the radio", flush=True)
except Exception as e:
print(f"[rig] auto scope_start error: {e}", flush=True)
# Force the radio's own MON (CI-V 0x45, "Monitor sidetone TX") OFF
# on the initial connect too - mirrors the same defensive call in
# webapp.py's _on_rig_reconnected() (2026-09-04), which only covers
# later reconnects, not this separate first-connect-at-startup path.
# See that call site for the full rationale.
try:
if not app.rig.sim and hasattr(app.rig, "set_func"):
await app.rig.set_func("MON", False)
print("[rig] MON forced OFF on initial connect (defensive)", flush=True)
except Exception as e:
print(f"[rig] MON force-off on initial connect error: {e}", flush=True)
asyncio.create_task(_initial_rig_connect())
app.init_rotators()
async def rot_poll():
tick = 0
while True:
await asyncio.sleep(0.5)
tick += 1
for r in app.rotators:
if r.moving:
# While moving: broadcast every 0.5s
await app.hub.broadcast({"type": "rotator_update", "rotator": r.state()})
elif tick % 4 == 0:
# While stationary: poll the hardware for position (STATUS) every 2s, then broadcast
if not r.sim:
pos = await asyncio.to_thread(r._read_pos, 1.5)
if pos is not None:
r.az = pos
await app.hub.broadcast({"type": "rotator_update", "rotator": r.state()})
# Radio polling (only when real hardware is connected — skipped in sim)
async def rig_poll():
cnt = 0
while True:
await asyncio.sleep(0.25)
if app.rig.sim or not app.rig.connected:
continue
cnt += 1
# S-meter/mode/freq broadcast is already done by the reader in
# civ.py (_poller_loop sends it via self.bcast). This is just a
# console heartbeat.
# PREVIOUSLY there was extra polling here with a bug (get_smeter
# returns the cached self.s_meter, so 'abs(lvl - s_meter)' was
# always 0 - it never broadcast, and it also overwrote correct
# values). Fixed 2026-07-05.
# Console heartbeat every ~2s (freq/mode/S-meter) - only in
# VERBOSE mode (HAM_VERBOSE=1). Silent by default - doesn't flood the console.
if VERBOSE and cnt % 8 == 0:
print(f"[rig] {app.rig.freq/1e6:.6f} MHz {app.rig.mode} "
f"S-meter={app.rig.s_meter:.1f}", flush=True)
# Supervised background loops (_supervise restarts them if they crash).
# rot_poll/rig_poll have their own try/except in the loop, but the
# supervisor protects against an unforeseen crash of the whole task.
app._supervise(lambda: rot_poll(), "rot_poll")
app._supervise(lambda: rig_poll(), "rig_poll")
app.audio.set_loop(asyncio.get_running_loop())
app._supervise(lambda: app._ft8_rx_loop(), "ft8_rx_loop")
app._supervise(lambda: app._waterfall_loop(), "waterfall_loop")
asyncio.create_task(app.tunnel.autostart())
print(f"[audio] Stream ready | opus={app.audio.get_status()['opus_lib']}")
# ── WSJT-X UDP monitor — autostart ───────────────────────────────────────
# NOTE: WSJT-X itself uses port 2237 as its local endpoint.
# Our server must listen on a DIFFERENT port (default 2238).
# In WSJT-X: Settings → Reporting → UDP Server: localhost, Port: 2238
wsjtx_port = app.cfg.get("wsjtxUdpPort", 2238)
wsjtx_auto = app.cfg.get("wsjtxAutostart", True)
if wsjtx_auto:
ok = await app.wsjtx.start(port=wsjtx_port)
if ok:
print(f"[wsjtx] UDP monitor active on port {wsjtx_port}")
else:
print(f"[wsjtx] Autostart failed — port {wsjtx_port} in use")
# ── rigctld emulator (Hamlib NET rigctl) — port 4532 ─────────────────────
# NOTE: we do NOT run a real rigctld.exe alongside our own CivRig — both
# would try to open the same COM port (e.g. COM13 for the IC-7300),
# causing a conflict: rigctld gets a TCP connection but "write_block()
# failed" on every command, because the serial port is already held by
# our main server (civ.py).
#
# Instead our emulator (hamlib_server.py) talks to the SAME self.rig
# (CivRig) object already used by the rest of the server — without
# opening an extra serial port. This is the only conflict-free approach
# when one process is meant to control the radio.
try:
from hamlib_server import HamlibManager
app.hamlib = HamlibManager(app)
# FORCE a fresh config on EVERY startup — don't trust the saved
# config from cfg.json, which could have been set to enabled=False
# during earlier UI-panel testing and silently block every slot
# with no error log at all.
app.cfg['hamlibServers'] = [
{"port": 4532, "enabled": True, "label": "Radio 1 — WSJT-X"},
{"port": 4533, "enabled": False, "label": "Radio 2"},
{"port": 4534, "enabled": False, "label": "Radio 3"},
]
await app.hamlib.start_all()
_n = len(app.hamlib.servers)
if _n > 0:
_ports = [s.port for s in app.hamlib.servers]
print(f"[hamlib] {_n} server(s) active on ports: {_ports}", flush=True)
else:
print(f"[hamlib] WARNING: ZERO servers started!", flush=True)
except Exception as e:
import traceback
print(f"[hamlib] EXCEPTION on startup: {e}", flush=True)
traceback.print_exc()
app.hamlib = None
web_app = web.Application(middlewares=[_security_headers_middleware])
web_app.router.add_route("GET", "/ws", app.ws_handler)
web_app.router.add_route("GET", "/hamlib", app.hamlib_ws_handler)
web_app.router.add_route("GET", "/ws/com-bridge", app.com_bridge_ws_handler)
web_app.router.add_route("OPTIONS","/{path:.*}", app.http_handler)
web_app.router.add_route("GET", "/{path:.*}", app.http_handler)
web_app.router.add_route("POST", "/{path:.*}", app.http_handler)
web_app.router.add_route("PUT", "/{path:.*}", app.http_handler)
web_app.router.add_route("PATCH", "/{path:.*}", app.http_handler)
web_app.router.add_route("DELETE", "/{path:.*}", app.http_handler)
runner = web.AppRunner(web_app)
await runner.setup()
# ── Rust audio bridge (ham_audio.exe) ────────────────────────────────────
try:
from audio_rust_bridge import rust_audio
audio_cfg = app.cfg.get("audio", {})
started = await rust_audio.start(hub=app.hub, cfg=audio_cfg)
if started:
print("[server] Rust audio bridge active (low latency)", flush=True)
app.rust_audio = rust_audio # available but doesn't replace app.audio
else:
print("[server] ham_audio.exe unavailable — using PyAudio", flush=True)
app.rust_audio = None
except Exception as e:
print(f"[server] Rust bridge error: {e} — using PyAudio", flush=True)
app.rust_audio = None
# Start HTTP (always on PORT)
site_http = web.TCPSite(runner, BIND_HOST, PORT)
# DeepCW - optional. If the deepcw_model/deepcw_engine modules are
# missing, the server starts without the neural CW decoder.
try:
from deepcw_model import deepcw_manager
deepcw_manager.load_from_disk()
asyncio.ensure_future(deepcw_manager.auto_check_loop(app.hub.broadcast))
# Auto-install DeepCW dependencies if missing
async def _ensure_deepcw_deps():
import importlib, subprocess, sys
missing = []
for pkg in ('onnxruntime', 'numpy'):
if importlib.util.find_spec(pkg) is None:
missing.append(pkg)
if missing:
print(f"[deepcw] Installing missing packages: {missing}", flush=True)
subprocess.check_call([sys.executable, '-m', 'pip', 'install'] + missing)
print("[deepcw] Installation complete", flush=True)
from deepcw_engine import deepcw_engine
await asyncio.sleep(1)
ok = await deepcw_engine.load()
print(f"[deepcw] load result: {ok}", flush=True)
asyncio.ensure_future(_ensure_deepcw_deps())
except ImportError as e:
print(f"[server] DeepCW module unavailable ({e}) - server starting without the neural CW decoder", flush=True)
except Exception as e:
print(f"[server] DeepCW init error: {e} - server continuing", flush=True)
# ── SSL / HTTPS ───────────────────────────────────────────────────────────
# getUserMedia (TX microphone) requires HTTPS or localhost
# Check whether certificates exist
import ssl as _ssl
import pathlib as _pl
import json as _json
_ssl_ctx = None
# First check for a certificate from the tunnel config (Let's Encrypt)
_tunnel_cfg_path = DATA / "tunnel_config.json"
_letsencrypt_cert = None
_letsencrypt_key = None
if _tunnel_cfg_path.exists():
try:
_tcfg = _json.loads(_tunnel_cfg_path.read_text())
_cp = _pl.Path(_tcfg.get("certPath", ""))
_kp = _pl.Path(_tcfg.get("keyPath", ""))
if _cp.exists() and _kp.exists():
_letsencrypt_cert = _cp
_letsencrypt_key = _kp
print(f"[ssl] Let's Encrypt cert found: {_cp}")
except Exception as _e:
print(f"[ssl] Error reading tunnel_config.json: {_e}")
# cert.pem/key.pem are saved to DATA (the writable directory - next to
# the EXE or APPDATA when installed in Program Files). Without this,
# "Permission denied" occurs on a Program Files install (read-only).
_cert = _letsencrypt_cert or (DATA / "cert.pem")
_key = _letsencrypt_key or (DATA / "key.pem")
if _cert.exists() and _key.exists():
# Use the existing certificate (Let's Encrypt or self-signed)
try:
_ssl_ctx = _ssl.SSLContext(_ssl.PROTOCOL_TLS_SERVER)
_ssl_ctx.load_cert_chain(str(_cert), str(_key))
print(f"[ssl] Certificate loaded: {_cert}")
except Exception as e:
print(f"[ssl] Certificate error: {e} — starting HTTP")
_ssl_ctx = None
else:
# Auto-generate a self-signed cert (no install required)
try:
from cryptography import x509
from cryptography.x509.oid import NameOID
from cryptography.hazmat.primitives import hashes, serialization
from cryptography.hazmat.primitives.asymmetric import rsa
import datetime
print("[ssl] Generating a self-signed certificate...")
_privkey = rsa.generate_private_key(public_exponent=65537, key_size=2048)
_subject = x509.Name([
x509.NameAttribute(NameOID.COMMON_NAME, u"your-server.ddns.net"),
])
_cert_obj = (
x509.CertificateBuilder()
.subject_name(_subject)
.issuer_name(_subject)
.public_key(_privkey.public_key())
.serial_number(x509.random_serial_number())
.not_valid_before(datetime.datetime.now(datetime.timezone.utc))
.not_valid_after(datetime.datetime.now(datetime.timezone.utc) + datetime.timedelta(days=3650))
.add_extension(x509.SubjectAlternativeName([
x509.DNSName(u"your-server.ddns.net"),
x509.DNSName(u"localhost"),
]), critical=False)
.sign(_privkey, hashes.SHA256())
)
_cert.write_bytes(_cert_obj.public_bytes(serialization.Encoding.PEM))
_key.write_bytes(_privkey.private_bytes(
serialization.Encoding.PEM,
serialization.PrivateFormat.TraditionalOpenSSL,
serialization.NoEncryption()
))
_ssl_ctx = _ssl.SSLContext(_ssl.PROTOCOL_TLS_SERVER)
_ssl_ctx.load_cert_chain(str(_cert), str(_key))
print(f"[ssl] Self-signed cert generated — cert.pem + key.pem")
except ImportError:
print("[ssl] 'cryptography' missing — starting HTTP")
print("[ssl] To enable HTTPS: pip install cryptography")
_ssl_ctx = None
# Start HTTP (always on PORT)
site_http = web.TCPSite(runner, BIND_HOST, PORT)
await site_http.start()
# Start HTTPS on PORT+1 if we have a certificate
_https_port = None
if _ssl_ctx:
_https_port = PORT + 1
try:
site_https = web.TCPSite(runner, BIND_HOST, _https_port, ssl_context=_ssl_ctx)
await site_https.start()
print(f"[ssl] HTTPS available on port {_https_port}")
except Exception as e:
print(f"[ssl] HTTPS error: {e}")
_https_port = None
# ── Certificate hot-reload (for 24/7 operation) ──────────────────────────
# The server loads the cert at startup. When you renew it (--gen-cert,
# manually or via the installer's scheduled task), the new file lands
# on disk, but the running
# server still holds the old one in memory. This task checks every 6h
# whether the cert file changed and reloads it LIVE
# (load_cert_chain on the existing SSLContext) - new connections get the
# new cert, WITHOUT restarting the server. Essential for software
# running 24/7 - the cert renews itself with zero downtime.
if _ssl_ctx and _cert and _key:
async def _cert_reload_watcher():
import os as _os
try:
last_mtime = _os.path.getmtime(str(_cert))
except Exception:
last_mtime = 0
while True:
await asyncio.sleep(6 * 3600) # check every 6 hours
try:
# Prefer Let's Encrypt if a fresh one has appeared
cur_cert, cur_key = _cert, _key
m = _os.path.getmtime(str(cur_cert))
if m != last_mtime:
_ssl_ctx.load_cert_chain(str(cur_cert), str(cur_key))
last_mtime = m
print(f"[ssl] Certificate reloaded live (no restart): "
f"{cur_cert}", flush=True)
except Exception as _e:
print(f"[ssl] Certificate hot-reload failed: {_e}", flush=True)
asyncio.create_task(_cert_reload_watcher())
print("[ssl] Certificate hot-reload active (checking every 6h) - "
"renewal doesn't require a restart", flush=True)
try:
s = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
s.connect(("8.8.8.8", 80))
local_ip = s.getsockname()[0]
s.close()
except:
local_ip = "127.0.0.1"
# One-line startup self-test. Before this, the only way to know whether
# CI-V/audio actually came up was reading the full startup log on the
# server's own console - not something a remote operator can do without
# RDP/SSH. _initial_rig_connect() above is fire-and-forget (not awaited),
# so this waits a few seconds to give it (and a slow "radio was off,
# wakeup" retry) a real chance to finish before checking.
async def _startup_self_test():
await asyncio.sleep(4.0)
rig_ok = bool(app.rig.connected and not app.rig.sim)
audio_ok = bool(app.rust_audio and getattr(app.rust_audio, "_connected", False))
status = "OK" if (rig_ok and audio_ok) else "WARNING"
print(f"[selftest] {status}: CI-V {'connected' if rig_ok else 'NOT connected'}, "
f"audio bridge {'connected' if audio_ok else 'NOT connected (PyAudio fallback)'}",
flush=True)
asyncio.create_task(_startup_self_test())
print(f"[build] webapp.py wersja {webapp.BUILD_MARKER}", flush=True)
print(f"""
╔════════════════════════════════════════════════╗
║ Ham Radio Control v{webapp.SERVER_VERSION} [Python+aiohttp] ║
╚════════════════════════════════════════════════╝
► HTTP: http://localhost:{PORT}
► HTTP: http://{local_ip}:{PORT}
{f'► HTTPS: https://{local_ip}:{_https_port} ← USE THIS (TX microphone)' if _https_port else '► HTTPS: unavailable (pip install cryptography)'}
► Login: admin / {ADMIN_PW}
► WebSocket: ws://localhost:{PORT}/ws
{f'► WebSocket SSL: wss://localhost:{_https_port}/ws' if _https_port else ''}
► Python {sys.version.split()[0]} | {sys.platform}
{f'► NOTE: self-signed cert — the browser will show a warning, click Advanced -> Proceed' if _https_port and not ((DATA / 'cert.pem').stat().st_size > 5000 if (DATA / 'cert.pem').exists() else False) else ''}
""")
await asyncio.Event().wait()
def main():
# Silence the Windows WinError 10054 (ConnectionResetError) in asyncio —
# it shows up when the browser closes the connection before the server
# does (e.g. a page refresh, closing a tab). This is normal behavior,
# not an application bug — by default asyncio logs it as an exception,
# cluttering the console.
import logging
# NOTE (stability): we do NOT silence the whole asyncio logger to
# CRITICAL, since that would hide "Task exception was never retrieved" -
# a signal that a background loop died silently. Instead we filter out
# only the ConnectionResetError noise.
logging.getLogger("asyncio").setLevel(logging.WARNING)
def _silence_reset(loop, context):
exc = context.get("exception")
if isinstance(exc, ConnectionResetError):
return # ignore WinError 10054 (client dropped the connection)
# STABILITY: a background task crash (e.g. _ft8_rx_loop,
# _device_watchdog) must be LOUD - otherwise the feature silently
# stops working.
_msg = context.get("message", "")
_task = context.get("future") or context.get("task")
if exc is not None:
print(f"[asyncio] UNHANDLED EXCEPTION in task: "
f"{type(exc).__name__}: {exc}", flush=True)
if _task is not None:
print(f"[asyncio] task: {_task!r}", flush=True)
import traceback as _tb
_tb.print_exception(type(exc), exc, exc.__traceback__)
elif _msg:
print(f"[asyncio] {_msg}", flush=True)
loop.default_exception_handler(context)
async def _run():
loop = asyncio.get_event_loop()
loop.set_exception_handler(_silence_reset)
await amain()
try:
asyncio.run(_run())
except KeyboardInterrupt:
pass
if __name__ == "__main__":
main()