826 lines
34 KiB
Python
826 lines
34 KiB
Python
#!/usr/bin/env python3
|
||
"""NetDesk desktop GUI - renders the approved HTML design in a native
|
||
WebView2 window (pywebview), wired to the real engine backend."""
|
||
|
||
import base64
|
||
import json
|
||
import os
|
||
import secrets
|
||
import string
|
||
import subprocess
|
||
import sys
|
||
import threading
|
||
import time
|
||
import urllib.request
|
||
import urllib.error
|
||
import webbrowser
|
||
from pathlib import Path
|
||
|
||
IS_WIN = sys.platform == "win32"
|
||
IS_MAC = sys.platform == "darwin"
|
||
if IS_WIN:
|
||
import ctypes
|
||
HERE = Path(__file__).resolve().parent
|
||
if getattr(sys, "frozen", False):
|
||
HERE = Path(sys.executable).resolve().parent
|
||
sys.path.insert(0, HERE)
|
||
else:
|
||
sys.path.insert(0, str(HERE))
|
||
|
||
import netdesk_host as nd # engine module
|
||
|
||
CREATE_NO_WINDOW = 0x08000000 if IS_WIN else 0
|
||
NL = chr(10)
|
||
|
||
|
||
def _html_path():
|
||
if getattr(sys, "frozen", False):
|
||
me = getattr(sys, "_MEIPASS", os.path.dirname(sys.executable))
|
||
cand = os.path.join(me, "netdesk_gui.html")
|
||
if os.path.exists(cand):
|
||
return cand
|
||
return os.path.join(os.path.dirname(sys.executable), "netdesk_gui.html")
|
||
return str(HERE / "netdesk_gui.html")
|
||
|
||
|
||
def engine_exe():
|
||
if getattr(sys, "frozen", False):
|
||
return [sys.executable, "--engine"]
|
||
return [sys.executable, str(HERE / "netdesk_host.py"), "--engine"]
|
||
|
||
|
||
class Api:
|
||
"""Exposed to the HTML UI as window.pywebview.api.*"""
|
||
|
||
def __init__(self):
|
||
self.proc = None
|
||
self.cfg = nd.Config(nd.appdata_dir() / "config.json")
|
||
self.portable = getattr(sys, "frozen", False) and \
|
||
self.cfg.path.parent.resolve() == HERE
|
||
|
||
_svc_cache = {"v": None, "t": 0.0}
|
||
_stu_cache = {"v": None, "t": 0.0}
|
||
|
||
def _service_installed(self):
|
||
# NOTE: `schtasks /Query /TN NetDesk` returns "Access is denied" for a
|
||
# NON-elevated process when the task was created elevated - which made
|
||
# the toggle grey on every fresh app start. The COM API distinguishes
|
||
# not-found (0x80070002) from access-denied (0x80070005 = task EXISTS)
|
||
# by HRESULT, which is locale-independent.
|
||
if not IS_WIN:
|
||
return False
|
||
if time.time() - type(self)._svc_cache["t"] < 10:
|
||
return bool(type(self)._svc_cache["v"])
|
||
ps = ("$s=New-Object -ComObject Schedule.Service;$s.Connect();"
|
||
"try{$null=$s.GetFolder('\\').GetTask('NetDesk');Write-Output 1}"
|
||
"catch{if($_.Exception.HResult -eq -2147024891){Write-Output 1}"
|
||
"else{Write-Output 0}}")
|
||
try:
|
||
r = subprocess.run(["powershell", "-NoProfile", "-Command", ps],
|
||
capture_output=True, text=True, timeout=20,
|
||
creationflags=CREATE_NO_WINDOW)
|
||
v = (r.stdout or "").strip().endswith("1")
|
||
except Exception:
|
||
v = bool(type(self)._svc_cache["v"]) # keep last known on error
|
||
type(self)._svc_cache = {"v": v, "t": time.time()}
|
||
return v
|
||
|
||
def _startup_present(self):
|
||
if not IS_WIN:
|
||
return False
|
||
if time.time() - type(self)._stu_cache["t"] < 10:
|
||
return bool(type(self)._stu_cache["v"])
|
||
r = subprocess.run(["reg", "query",
|
||
r"HKCU\Software\Microsoft\Windows\CurrentVersion\Run",
|
||
"/v", "NetDesk"], capture_output=True,
|
||
creationflags=CREATE_NO_WINDOW).returncode
|
||
type(self)._stu_cache = {"v": r == 0, "t": time.time()}
|
||
return r == 0
|
||
|
||
def _engine_alive(self):
|
||
sf = self.cfg.path.parent / "gui_status.json"
|
||
try:
|
||
d = json.loads(sf.read_text("utf-8"))
|
||
return time.time() - d.get("ts", 0) < 12
|
||
except Exception:
|
||
return False
|
||
|
||
def log_err(self, msg):
|
||
try:
|
||
with open(self.cfg.path.parent / "gui_err.log", "a", encoding="utf-8") as f:
|
||
f.write(time.strftime("%H:%M:%S ") + str(msg) + NL)
|
||
except Exception:
|
||
pass
|
||
return {"ok": True}
|
||
|
||
# ---------------- account (sign in / sign up, binds this PC to the web app)
|
||
def _api_base(self):
|
||
url = self.cfg.data.get("server") or nd.DEFAULT_SERVER
|
||
url = url.replace("wss://", "https://").replace("ws://", "http://")
|
||
url = url.rsplit("/ws", 1)[0]
|
||
return url.rstrip("/")
|
||
|
||
def _api_post(self, path, payload):
|
||
data = json.dumps(payload).encode()
|
||
headers = {"Content-Type": "application/json"}
|
||
tok = self.cfg.data.get("account_token", "")
|
||
if tok:
|
||
headers["Authorization"] = "Bearer " + tok
|
||
req = urllib.request.Request(self._api_base() + path, data=data,
|
||
headers=headers, method="POST")
|
||
try:
|
||
with urllib.request.urlopen(req, timeout=20) as r:
|
||
return json.loads(r.read().decode() or "{}")
|
||
except urllib.error.HTTPError as e:
|
||
try:
|
||
return json.loads(e.read().decode() or "{}")
|
||
except Exception:
|
||
return {"error": f"server error {e.code}"}
|
||
except Exception as e:
|
||
return {"error": f"cannot reach server: {e}"}
|
||
|
||
def _account_result(self, r):
|
||
if r.get("ok") and r.get("token"):
|
||
self.cfg.data["account_token"] = r["token"]
|
||
self.cfg.data["account_email"] = r.get("email", "")
|
||
self.cfg.save() # engine picks it up at the next (re)register via reload()
|
||
return {"ok": True, "email": r.get("email", ""), "msg": r.get("msg", "")}
|
||
return {"ok": False, "error": r.get("error", "sign-in failed")}
|
||
|
||
def account_login(self, email, password):
|
||
email = (email or "").strip().lower()
|
||
if "@" not in email or len(password or "") < 1:
|
||
return {"ok": False, "error": "Enter your email and password."}
|
||
return self._account_result(self._api_post(
|
||
"/api/auth/login", {"email": email, "password": password, "app": True}))
|
||
|
||
def account_signup(self, email, password):
|
||
email = (email or "").strip().lower()
|
||
if "@" not in email:
|
||
return {"ok": False, "error": "Enter a valid email."}
|
||
if len(password or "") < 8:
|
||
return {"ok": False, "error": "Password needs 8+ characters."}
|
||
return self._account_result(self._api_post(
|
||
"/api/auth/signup", {"email": email, "password": password, "app": True}))
|
||
|
||
def account_logout(self):
|
||
tok = self.cfg.data.get("account_token", "")
|
||
if tok:
|
||
try:
|
||
self._api_post("/api/auth/logout", {})
|
||
except Exception:
|
||
pass
|
||
self.cfg.data.pop("account_token", None)
|
||
self.cfg.data.pop("account_email", None)
|
||
self.cfg.save()
|
||
return {"ok": True}
|
||
|
||
def get_state(self):
|
||
try:
|
||
return self._get_state_inner()
|
||
except Exception as e:
|
||
try:
|
||
self.log_err("get_state: " + repr(e))
|
||
except Exception:
|
||
pass
|
||
return {"id": self.cfg.id, "state": "error", "error": repr(e),
|
||
"portable": self.portable}
|
||
|
||
_last_online = 0.0
|
||
|
||
def _get_state_inner(self):
|
||
state, in_sess = "stopped", False
|
||
sf = self.cfg.path.parent / "gui_status.json"
|
||
try:
|
||
d = json.loads(sf.read_text("utf-8"))
|
||
if time.time() - d.get("ts", 0) < 12:
|
||
state, in_sess = d.get("state"), bool(d.get("in_session"))
|
||
except Exception:
|
||
pass
|
||
# Anti-flicker: a stale duplicate engine writes "error: already
|
||
# connected elsewhere" between the real engine's online heartbeats.
|
||
# That error literally means the ID IS being served by this machine,
|
||
# and any brief error/connecting blip within 12s of a confirmed
|
||
# online heartbeat is noise, not a real outage.
|
||
now = time.time()
|
||
cls = type(self)
|
||
alive = self._engine_alive() # some local engine heartbeated < 12s ago
|
||
# A stale duplicate engine writes "already connected elsewhere" between
|
||
# the real engine's online heartbeats. When an engine is genuinely
|
||
# heartbeating locally, that means the ID IS served here -> online.
|
||
if isinstance(state, str) and "already connected" in state and alive:
|
||
state = "online"
|
||
if state in ("online", "session"):
|
||
cls._last_online = now
|
||
elif (state != "stopped" or alive) and now - cls._last_online < 12:
|
||
# brief connecting/error blip within 12s of a confirmed online
|
||
# (e.g. slow-network reconnect) is noise, not a real outage
|
||
state = "online"
|
||
# Truly stopped only when OUR spawned engine exited AND no engine at all
|
||
# is heartbeating - never override a live/reconnecting engine's status.
|
||
if self.proc is not None and self.proc.poll() is not None \
|
||
and not alive and now - cls._last_online >= 12:
|
||
state = "stopped"
|
||
elif state is None:
|
||
state = "running" if alive else "stopped"
|
||
return {
|
||
"id": self.cfg.id,
|
||
"server": self.cfg.data.get("server") or nd.DEFAULT_SERVER,
|
||
"regKey": self.cfg.data.get("reg_key", ""),
|
||
"name": self.cfg.data.get("name", ""),
|
||
"password": self.cfg.data.get("pw_plain", ""),
|
||
"account": self.cfg.data.get("account_email", ""),
|
||
"state": state,
|
||
"inSession": in_sess,
|
||
"totp": bool(self.cfg.data.get("totp")),
|
||
"allowIps": self.cfg.data.get("allow_ips", ""),
|
||
"portable": self.portable,
|
||
"configPath": str(self.cfg.path),
|
||
"serviceInstalled": self._service_installed(),
|
||
"startup": self._startup_present(),
|
||
"privacy": bool(self.cfg.data.get("privacy_on_session")
|
||
and self.cfg.data.get("allow_display_power")),
|
||
"lockOnEnd": bool(self.cfg.data.get("lock_on_end")),
|
||
"closeToTray": bool(self.cfg.data.get("close_to_tray", True)),
|
||
"inTray": self._tray_running(),
|
||
"isMac": IS_MAC,
|
||
}
|
||
|
||
def copy_id(self):
|
||
if IS_WIN:
|
||
subprocess.run(["powershell", "-Command",
|
||
f"Set-Clipboard -Value '{self.cfg.id}'"],
|
||
capture_output=True, creationflags=CREATE_NO_WINDOW)
|
||
return {"ok": True}
|
||
|
||
def start_engine(self, server, regkey, name):
|
||
if self._engine_alive():
|
||
return {"ok": False, "error": "Engine is already running."}
|
||
server = nd.normalize_server(server) or (self.cfg.data.get("server") or nd.DEFAULT_SERVER)
|
||
self.cfg.data["server"] = server
|
||
self.cfg.data["server_locked_by_user"] = True
|
||
if regkey.strip():
|
||
self.cfg.data["reg_key"] = regkey.strip()
|
||
if name.strip():
|
||
self.cfg.data["name"] = name.strip()
|
||
self.cfg.save()
|
||
cmd = engine_exe() + ["--server", server]
|
||
regkey = regkey.strip() or self.cfg.data.get("reg_key", "")
|
||
if regkey:
|
||
cmd += ["--reg-key", regkey]
|
||
if name.strip():
|
||
cmd += ["--name", name.strip()]
|
||
flags = CREATE_NO_WINDOW if IS_WIN else 0
|
||
self.proc = subprocess.Popen(cmd, creationflags=flags,
|
||
stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL)
|
||
return {"ok": True}
|
||
|
||
def ensure_engine(self):
|
||
"""Auto-bring the host ONLINE on app launch (TeamViewer-style) unless
|
||
the user chose to stay offline. Safe: a fresh config always carries a
|
||
random password, so 'online' never means 'open to anyone'. No-op when
|
||
an engine (user or SYSTEM broker) is already serving this ID."""
|
||
try:
|
||
if self.cfg.data.get("start_offline"):
|
||
return {"ok": False, "offline": True}
|
||
if self._engine_alive():
|
||
return {"ok": True, "already": True}
|
||
server = self.cfg.data.get("server") or nd.DEFAULT_SERVER
|
||
cmd = engine_exe() + ["--server", server]
|
||
rk = self.cfg.data.get("reg_key", "")
|
||
if rk:
|
||
cmd += ["--reg-key", rk]
|
||
nm = self.cfg.data.get("name", "")
|
||
if nm:
|
||
cmd += ["--name", nm]
|
||
flags = CREATE_NO_WINDOW if IS_WIN else 0
|
||
self.proc = subprocess.Popen(cmd, creationflags=flags,
|
||
stdout=subprocess.DEVNULL,
|
||
stderr=subprocess.DEVNULL)
|
||
return {"ok": True, "started": True}
|
||
except Exception as e:
|
||
return {"ok": False, "error": str(e)}
|
||
|
||
def set_start_offline(self, on):
|
||
"""User preference: skip auto-going-online when the app launches."""
|
||
self.cfg.data["start_offline"] = bool(on)
|
||
self.cfg.save()
|
||
return {"ok": True}
|
||
|
||
def stop_engine(self):
|
||
if self.proc is not None and self.proc.poll() is None:
|
||
self.proc.terminate()
|
||
self.proc = None
|
||
return {"ok": True}
|
||
if IS_WIN and getattr(sys, "frozen", False):
|
||
subprocess.run(["powershell", "-Command",
|
||
"Get-CimInstance Win32_Process -Filter \"Name='NetDeskHost.exe'\" | "
|
||
"Where-Object { $_.CommandLine -like '*--engine*' } | "
|
||
"ForEach-Object { Stop-Process -Id $_.ProcessId -Force }"],
|
||
capture_output=True, creationflags=CREATE_NO_WINDOW)
|
||
return {"ok": True}
|
||
return {"ok": False, "error": "not running"}
|
||
|
||
def set_password(self, pw):
|
||
pw = (pw or "").strip()
|
||
if len(pw) < 8:
|
||
return {"ok": False, "error": "Use at least 8 characters."}
|
||
self.cfg.set_password(pw)
|
||
return {"ok": True, "msg": "Password updated - it applies to the next connection."}
|
||
|
||
def generate_password(self):
|
||
"""One-click random password, TeamViewer/RustDesk style. Saved and
|
||
shown; the engine picks it up at the next incoming connection."""
|
||
pw = nd.gen_password(10)
|
||
self.cfg.set_password(pw)
|
||
return {"ok": True, "password": pw}
|
||
|
||
def toggle_2fa(self):
|
||
if self.cfg.data.get("totp"):
|
||
self.cfg.data.pop("totp", None)
|
||
self.cfg.save()
|
||
return {"ok": True, "on": False}
|
||
secret = "".join(secrets.choice(string.ascii_uppercase + "234567") for _ in range(32))
|
||
self.cfg.set_totp(secret)
|
||
f = self.cfg.path.parent / "totp_secret.txt"
|
||
f.write_text("NetDesk 2FA secret (base32)\n\n" + secret + "\n", "utf-8")
|
||
return {"ok": True, "on": True, "secret": secret, "path": str(f)}
|
||
|
||
def totp_qr(self):
|
||
"""QR code (base64 PNG) for the current TOTP secret - scan it with
|
||
Google Authenticator / Authy / 1Password."""
|
||
secret = self.cfg.data.get("totp", "")
|
||
if not secret:
|
||
return {"ok": False, "error": "2FA is not enabled"}
|
||
label = f"NetDesk:{self.cfg.id}"
|
||
uri = (f"otpauth://totp/{urllib.parse.quote(label)}?secret={secret}"
|
||
f"&issuer=NetDesk&digits=6&period=30")
|
||
try:
|
||
import io
|
||
import qrcode
|
||
img = qrcode.make(uri, box_size=5, border=2)
|
||
buf = io.BytesIO()
|
||
img.save(buf, format="PNG")
|
||
b64 = base64.b64encode(buf.getvalue()).decode()
|
||
return {"ok": True, "png": "data:image/png;base64," + b64,
|
||
"secret": secret}
|
||
except Exception as e:
|
||
return {"ok": False, "error": str(e)[:120], "secret": secret}
|
||
|
||
def set_allow_ips(self, text):
|
||
"""Save the viewer IP/CIDR allowlist (empty = allow everyone)."""
|
||
import ipaddress
|
||
parts = str(text or "").replace(",", " ").split()
|
||
bad = []
|
||
for p in parts:
|
||
try:
|
||
ipaddress.ip_network(p, strict=False)
|
||
except ValueError:
|
||
bad.append(p)
|
||
if bad:
|
||
return {"ok": False, "error": "Not a valid IP/CIDR: " + ", ".join(bad[:3])}
|
||
self.cfg.data["allow_ips"] = " ".join(parts)
|
||
self.cfg.save()
|
||
return {"ok": True, "count": len(parts)}
|
||
|
||
def wake_device(self, dev_id):
|
||
"""Wake-on-LAN an offline fleet computer (relayed via the server)."""
|
||
tok = self.cfg.data.get("account_token", "")
|
||
if not tok:
|
||
return {"ok": False, "error": "not signed in"}
|
||
r = self._api_post("/api/wake", {"id": str(dev_id)})
|
||
if r.get("ok"):
|
||
return {"ok": True, "already_online": bool(r.get("already_online")),
|
||
"relayed": r.get("relayed", 0)}
|
||
return {"ok": False, "error": r.get("error", "wake failed")}
|
||
|
||
def export_logs(self):
|
||
"""Copy the session audit log to the Desktop for support/compliance."""
|
||
try:
|
||
src = self.cfg.path.parent / "host.log"
|
||
if not src.is_file():
|
||
return {"ok": False, "error": "no log file yet"}
|
||
dest_dir = Path.home() / "Desktop"
|
||
if not dest_dir.is_dir():
|
||
dest_dir = Path.home()
|
||
dest = dest_dir / time.strftime("netdesk-log-%Y%m%d-%H%M%S.txt")
|
||
dest.write_bytes(src.read_bytes())
|
||
return {"ok": True, "path": str(dest)}
|
||
except Exception as e:
|
||
return {"ok": False, "error": str(e)[:120]}
|
||
|
||
# ---------------- enterprise UI data feeds ----------------
|
||
def ping_relay(self):
|
||
"""Round-trip time to the configured relay's /health endpoint.
|
||
Retries once so a single dropped packet on a slow line never
|
||
flashes 'offline'."""
|
||
err = ""
|
||
for _ in range(2):
|
||
try:
|
||
t0 = time.time()
|
||
with urllib.request.urlopen(self._api_base() + "/health", timeout=10) as r:
|
||
r.read()
|
||
return {"ok": True, "ms": int((time.time() - t0) * 1000)}
|
||
except Exception as e:
|
||
err = str(e)[:120]
|
||
return {"ok": False, "error": err}
|
||
|
||
def list_devices(self):
|
||
"""This account's computers from the relay (for My Computers)."""
|
||
tok = self.cfg.data.get("account_token", "")
|
||
if not tok:
|
||
return {"ok": False, "error": "not signed in", "devices": []}
|
||
req = urllib.request.Request(self._api_base() + "/api/devices",
|
||
headers={"Authorization": "Bearer " + tok})
|
||
try:
|
||
with urllib.request.urlopen(req, timeout=10) as r:
|
||
d = json.loads(r.read().decode())
|
||
return {"ok": True, "devices": d.get("devices", [])}
|
||
except Exception as e:
|
||
return {"ok": False, "error": str(e)[:120], "devices": []}
|
||
|
||
def get_logs(self, n=150):
|
||
"""Tail of the local agent log for the Session Logs view."""
|
||
try:
|
||
p = self.cfg.path.parent / "host.log"
|
||
lines = p.read_text("utf-8", errors="replace").splitlines()[-int(n):]
|
||
return {"ok": True, "lines": lines}
|
||
except Exception as e:
|
||
return {"ok": False, "lines": [], "error": str(e)[:120]}
|
||
|
||
def open_partner(self, pid):
|
||
"""Outgoing connection: open the remote-control viewer in its own app
|
||
window (WebView2) - app-to-app like RustDesk/TeamViewer, one window per
|
||
partner. Falls back to the system browser only if the window can't open."""
|
||
digits = "".join(ch for ch in str(pid or "") if ch.isdigit())
|
||
if len(digits) != 9:
|
||
return {"ok": False, "error": "Enter the partner's 9-digit ID."}
|
||
url = self._api_base() + "/app?id=" + digits + "&embed=1"
|
||
try:
|
||
import webview
|
||
import ctypes
|
||
sw, sh = 1536, 864
|
||
try:
|
||
u = ctypes.windll.user32
|
||
try:
|
||
scale = u.GetDpiForSystem() / 96.0
|
||
except Exception:
|
||
scale = 1.0
|
||
sw = u.GetSystemMetrics(0) / (scale or 1.0)
|
||
sh = u.GetSystemMetrics(1) / (scale or 1.0)
|
||
except Exception:
|
||
pass
|
||
w, h = int(sw * 0.8), int(sh * 0.82)
|
||
title = "NetDesk – " + " ".join(digits[i:i + 3] for i in range(0, 9, 3))
|
||
# the viewer window has the normal OS frame (resizable, maximize)
|
||
webview.create_window(title, url, width=w, height=h, min_size=(720, 460),
|
||
x=max(0, int((sw - w) / 2)), y=max(0, int((sh - h) / 2)),
|
||
background_color="#05070a", text_select=False)
|
||
return {"ok": True, "app": True}
|
||
except Exception as e:
|
||
self.log_err("viewer window: %s" % e)
|
||
webbrowser.open(url)
|
||
return {"ok": True}
|
||
|
||
_maximized = False
|
||
|
||
def maximize_window(self):
|
||
"""Toggle maximize/restore (never fullscreen - that traps a frameless
|
||
window with no way out)."""
|
||
try:
|
||
import webview
|
||
for w in webview.windows:
|
||
try:
|
||
if type(self)._maximized:
|
||
w.restore()
|
||
type(self)._maximized = False
|
||
else:
|
||
w.maximize()
|
||
type(self)._maximized = True
|
||
except Exception:
|
||
pass
|
||
except Exception:
|
||
pass
|
||
return {"ok": True, "maximized": type(self)._maximized}
|
||
|
||
def fit_window(self, height):
|
||
"""Resize the window height to fit its content (width unchanged)."""
|
||
try:
|
||
import webview
|
||
h = max(560, min(int(height), 1400))
|
||
for w in webview.windows:
|
||
try:
|
||
w.resize(getattr(w, "width", 1024) or 1024, h)
|
||
except Exception:
|
||
pass
|
||
except Exception:
|
||
pass
|
||
return {"ok": True}
|
||
|
||
def set_switches(self, startup=None, privacy=None, lock=None):
|
||
if privacy is not None:
|
||
self.cfg.data["privacy_on_session"] = bool(privacy)
|
||
self.cfg.data["allow_display_power"] = bool(privacy)
|
||
if lock is not None:
|
||
self.cfg.data["lock_on_end"] = bool(lock)
|
||
self.cfg.save()
|
||
return {"ok": True}
|
||
|
||
def install_service(self):
|
||
if not IS_WIN:
|
||
return {"ok": False, "error": "Windows only"}
|
||
# hand over: stop the GUI-spawned engine; the service owns the connection
|
||
self.stop_engine()
|
||
server = self.cfg.data.get("server") or nd.DEFAULT_SERVER
|
||
# clear any stale result from a previous attempt so we read a fresh one
|
||
result_path = nd.programdata_dir() / "install_result.json"
|
||
try:
|
||
result_path.unlink()
|
||
except Exception:
|
||
pass
|
||
exe = sys.executable
|
||
params = f'--install --server "{server}"'
|
||
reg = self.cfg.data.get("reg_key", "")
|
||
if reg:
|
||
params += f' --reg-key "{reg}"'
|
||
if not getattr(sys, "frozen", False):
|
||
params = f'"{HERE / "netdesk_host.py"}" ' + params
|
||
try:
|
||
r = ctypes.windll.shell32.ShellExecuteW(None, "runas", exe, params, None, 1)
|
||
if r <= 32:
|
||
return {"ok": False, "error": "UAC was cancelled - accept the prompt to install."}
|
||
except Exception as e:
|
||
return {"ok": False, "error": str(e)}
|
||
# Wait up to 60 s: the elevated one-file exe must first self-extract
|
||
# (a few seconds) before it can create the task. Trust install_result.json
|
||
# (written by the elevated installer) as the definitive answer, and fall
|
||
# back to querying the scheduled task directly.
|
||
for _ in range(120):
|
||
try:
|
||
res = json.loads(result_path.read_text("utf-8"))
|
||
if res.get("ok"):
|
||
# force the status cache so the toggle turns green immediately
|
||
type(self)._svc_cache = {"v": True, "t": time.time()}
|
||
return {"ok": True,
|
||
"msg": "Service installed - reachable at boot, the logon screen "
|
||
"and UAC prompts. The engine handed over to the service."}
|
||
return {"ok": False, "error": res.get("error", "install failed")}
|
||
except Exception:
|
||
pass
|
||
if self._service_installed():
|
||
type(self)._svc_cache = {"v": True, "t": time.time()}
|
||
return {"ok": True,
|
||
"msg": "Service installed - reachable at boot, the logon screen "
|
||
"and UAC prompts."}
|
||
time.sleep(0.5)
|
||
return {"ok": False,
|
||
"error": "Install is taking longer than expected. If you accepted the UAC "
|
||
"prompt, the toggle should turn on shortly - reopen the app to check."}
|
||
|
||
def remove_service_full(self):
|
||
r = self.remove_service()
|
||
# also stop broker/agent processes from the removed service
|
||
if IS_WIN and getattr(sys, "frozen", False):
|
||
subprocess.run(["powershell", "-Command",
|
||
"Get-CimInstance Win32_Process -Filter \"Name='NetDeskHost.exe'\" | "
|
||
"Where-Object { $_.CommandLine -like '*--broker*' -or "
|
||
"$_.CommandLine -like '*--engine*' } | "
|
||
"ForEach-Object { Stop-Process -Id $_.ProcessId -Force }"],
|
||
capture_output=True, creationflags=CREATE_NO_WINDOW)
|
||
return r
|
||
|
||
def remove_service(self):
|
||
if not IS_WIN:
|
||
return {"ok": False}
|
||
subprocess.run(["schtasks", "/Delete", "/TN", "NetDesk", "/F"],
|
||
capture_output=True, creationflags=CREATE_NO_WINDOW)
|
||
return {"ok": True, "msg": "Service removed (if it existed)."}
|
||
|
||
def add_startup(self):
|
||
if getattr(sys, "frozen", False):
|
||
exe = Path(sys.executable).resolve()
|
||
else:
|
||
exe = HERE / "NetDeskHost.exe"
|
||
val = f'"{exe}" --engine --server "{self.cfg.data.get("server") or "wss://netdesk.space/ws"}"'
|
||
subprocess.run(["reg", "add", r"HKCU\Software\Microsoft\Windows\CurrentVersion\Run",
|
||
"/v", "NetDesk", "/t", "REG_SZ", "/d", val, "/f"],
|
||
capture_output=True, creationflags=CREATE_NO_WINDOW)
|
||
return {"ok": True}
|
||
|
||
def remove_startup(self):
|
||
if not IS_WIN:
|
||
return {"ok": False}
|
||
subprocess.run(["reg", "delete", r"HKCU\Software\Microsoft\Windows\CurrentVersion\Run",
|
||
"/v", "NetDesk", "/f"], capture_output=True,
|
||
creationflags=CREATE_NO_WINDOW)
|
||
return {"ok": True, "msg": "Startup entry removed."}
|
||
|
||
def open_viewer(self):
|
||
url = self.cfg.data.get("server") or "wss://netdesk.space/ws"
|
||
url = url.replace("wss://", "https://").replace("ws://", "http://")
|
||
url = url.rsplit("/ws", 1)[0] or "http://127.0.0.1:8443"
|
||
webbrowser.open(url)
|
||
return {"ok": True}
|
||
|
||
def close_request(self):
|
||
"""✕ button: hide to tray by default, or really quit if disabled."""
|
||
if self.cfg.data.get("close_to_tray", True):
|
||
self.hide_window()
|
||
return {"ok": True, "hidden": True}
|
||
return {"ok": True, "hidden": False, "quit": True}
|
||
|
||
def hide_window(self):
|
||
try:
|
||
import webview
|
||
for w in webview.windows:
|
||
w.hide()
|
||
except Exception:
|
||
pass
|
||
|
||
def show_window(self):
|
||
try:
|
||
import webview
|
||
for w in webview.windows:
|
||
w.show()
|
||
except Exception:
|
||
pass
|
||
|
||
def minimize(self):
|
||
try:
|
||
import webview
|
||
for w in webview.windows:
|
||
w.minimize()
|
||
except Exception:
|
||
pass
|
||
return {"ok": True}
|
||
|
||
def set_close_to_tray(self, on):
|
||
self.cfg.data["close_to_tray"] = bool(on)
|
||
self.cfg.save()
|
||
return {"ok": True}
|
||
|
||
def quit_all(self):
|
||
try:
|
||
self.stop_engine()
|
||
except Exception:
|
||
pass
|
||
try:
|
||
import webview
|
||
for w in webview.windows:
|
||
w.destroy()
|
||
except Exception:
|
||
pass
|
||
os._exit(0)
|
||
|
||
def _tray_running(self):
|
||
return getattr(self, "_tray", None) is not None and self._tray.visible
|
||
|
||
def open_logs(self):
|
||
p = str(self.cfg.path.parent)
|
||
if IS_WIN:
|
||
os.startfile(p) # noqa
|
||
elif IS_MAC:
|
||
subprocess.Popen(["open", p])
|
||
else:
|
||
subprocess.Popen(["xdg-open", p])
|
||
return {"ok": True}
|
||
|
||
|
||
def _tray_image():
|
||
from PIL import Image as _I
|
||
return _I.open(_io.BytesIO(base64.b64decode(ICON_B64))) if ICON_B64 else None
|
||
|
||
|
||
def _start_tray(api):
|
||
"""System tray icon with the NetDesk logo and control menu."""
|
||
try:
|
||
import pystray
|
||
img = _tray_image()
|
||
if img is None:
|
||
return
|
||
import webview
|
||
|
||
def show_w(api_ref, icon_ref, item):
|
||
api.show_window()
|
||
|
||
def quit_all(icon_ref, item):
|
||
api.quit_all()
|
||
|
||
def noop(icon_ref, item):
|
||
pass
|
||
|
||
menu = pystray.Menu(
|
||
pystray.MenuItem("Show NetDesk", lambda ic, it: (api.show_window()),
|
||
default=True),
|
||
pystray.Menu.SEPARATOR,
|
||
pystray.MenuItem("Start engine",
|
||
lambda ic, it: api.start_engine(
|
||
api.cfg.data.get("server", ""),
|
||
"", api.cfg.data.get("name", ""))),
|
||
pystray.MenuItem("Stop engine", lambda ic, it: api.stop_engine()),
|
||
pystray.Menu.SEPARATOR,
|
||
pystray.MenuItem("Open Web Viewer", lambda ic, it: api.open_viewer()),
|
||
pystray.MenuItem("Exit NetDesk", lambda ic, it: api.quit_all()),
|
||
)
|
||
icon = pystray.Icon("NetDesk", img, "NetDesk", menu)
|
||
api._tray = icon
|
||
icon.run_detached()
|
||
except Exception as e:
|
||
print("tray unavailable:", e)
|
||
|
||
|
||
def run():
|
||
import webview
|
||
api = Api()
|
||
# standard client size like RustDesk/TeamViewer - never starts maximized.
|
||
# Sized to ~62% x 75% of the desktop so it always feels like a windowed
|
||
# app; the page auto-zooms so the full layout fits (see fitZoom in HTML).
|
||
sw, sh = 1536, 864
|
||
try:
|
||
import ctypes
|
||
u = ctypes.windll.user32
|
||
try:
|
||
scale = u.GetDpiForSystem() / 96.0
|
||
except Exception:
|
||
scale = 1.0
|
||
sw = u.GetSystemMetrics(0) / (scale or 1.0)
|
||
sh = u.GetSystemMetrics(1) / (scale or 1.0)
|
||
except Exception:
|
||
pass
|
||
w = int(min(1060, max(880, sw * 0.62)))
|
||
h = int(min(760, max(600, sh * 0.75)))
|
||
win = webview.create_window("NetDesk", _html_path(), js_api=api,
|
||
width=w, height=h, min_size=(860, 590),
|
||
x=max(20, int((sw - w) / 2)), y=max(20, int((sh - h) / 2.4)),
|
||
maximized=False,
|
||
frameless=True, easy_drag=False,
|
||
background_color="#0b0f15")
|
||
|
||
def _on_closing():
|
||
# native ✕ hides to tray by default (RustDesk/TeamViewer behavior),
|
||
# or really quits when "Close to tray" is off
|
||
if api.cfg.data.get("close_to_tray", True):
|
||
api.hide_window()
|
||
return False # cancel the close - just hide
|
||
api.quit_all()
|
||
return True
|
||
try:
|
||
win.events.closing += _on_closing
|
||
except Exception:
|
||
pass
|
||
|
||
threading.Thread(target=_start_tray, args=(api,), daemon=True).start()
|
||
|
||
# Auto bring the host ONLINE on launch (TeamViewer-style), and on the very
|
||
# first run default "Start with Windows" ON so it reconnects at every login.
|
||
# Both are safe (a fresh config always carries a random password) and fully
|
||
# reversible from the toggles. Runs off-thread so the UI never blocks on it.
|
||
def _autostart():
|
||
try:
|
||
api.ensure_engine()
|
||
except Exception:
|
||
pass
|
||
try:
|
||
if IS_WIN and not api.cfg.data.get("autostart_initialized"):
|
||
api.add_startup()
|
||
api.cfg.data["autostart_initialized"] = True
|
||
api.cfg.save()
|
||
except Exception:
|
||
pass
|
||
threading.Thread(target=_autostart, daemon=True).start()
|
||
|
||
webview.start()
|
||
|
||
|
||
def selftest():
|
||
import webview
|
||
api = Api()
|
||
result = {"ok": False}
|
||
|
||
def _check():
|
||
time.sleep(2.2)
|
||
try:
|
||
st = api.get_state()
|
||
result["ok"] = isinstance(st, dict) and "id" in st
|
||
except Exception as e:
|
||
result["err"] = str(e)
|
||
for w in webview.windows:
|
||
try:
|
||
w.destroy()
|
||
except Exception:
|
||
pass
|
||
|
||
threading.Thread(target=_check, daemon=True).start()
|
||
webview.create_window("NetDesk selftest", _html_path(), js_api=api,
|
||
width=580, height=760, background_color="#0b0e13")
|
||
webview.start()
|
||
if result["ok"]:
|
||
print("GUI_SELFTEST_OK")
|
||
else:
|
||
print("GUI_SELFTEST_FAIL", result)
|
||
sys.exit(1)
|
||
|
||
|
||
if __name__ == "__main__":
|
||
run()
|