202 lines
7.8 KiB
Python
202 lines
7.8 KiB
Python
"""Outbound RDP/VNC sessions to other machines, driven from Home Assistant.
|
|
|
|
The opposite direction from wayvnc: wayvnc is the *inbound* channel that lets a human
|
|
control this thin client, this module is the thin client connecting *out* to a laptop
|
|
or desktop elsewhere on the LAN. One client (Remmina) covers both protocols.
|
|
|
|
Per the security note in mqtt_discovery.py, the payload from HA only ever selects a
|
|
name out of the target list parsed from rdp-vnc.json. Host, port and protocol come from
|
|
that file; a payload that does not match a known name is dropped. The generated
|
|
.remmina profile is written to a path built from the target's index in that file, so
|
|
even a target name full of slashes could not escape the profile directory.
|
|
|
|
Credentials: the username (and Windows domain, if any) are read from a 0600 env file
|
|
that is never committed. The password deliberately is not — see
|
|
configs/remote-desktop/remote-desktop-credentials.env.example for why.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import logging
|
|
import os
|
|
import re
|
|
import subprocess
|
|
from dataclasses import dataclass
|
|
|
|
from .runtime_state import STATE_DIR, TEMPLATE_DIR, ensure_runtime_copy, load_json
|
|
from .sway_control import WS_WEB
|
|
|
|
log = logging.getLogger(__name__)
|
|
|
|
CONFIG_FILENAME = "rdp-vnc.json"
|
|
CREDENTIALS_PATH = os.path.join(TEMPLATE_DIR, "remote-desktop-credentials.env")
|
|
PROFILE_DIR = os.path.join(STATE_DIR, "remmina")
|
|
|
|
PROTOCOLS = {"rdp": "RDP", "vnc": "VNC"}
|
|
DEFAULT_PORTS = {"rdp": 3389, "vnc": 5900}
|
|
|
|
REMMINA_APP_ID = "org.remmina.Remmina"
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class Target:
|
|
name: str
|
|
host: str
|
|
port: int
|
|
protocol: str
|
|
credentials_ref: str
|
|
profile_path: str
|
|
|
|
|
|
def _read_credentials() -> dict[str, str]:
|
|
values: dict[str, str] = {}
|
|
try:
|
|
with open(CREDENTIALS_PATH, encoding="utf-8") as handle:
|
|
for line in handle:
|
|
line = line.strip()
|
|
if not line or line.startswith("#") or "=" not in line:
|
|
continue
|
|
key, _, value = line.partition("=")
|
|
values[key.strip().upper()] = value.strip().strip('"').strip("'")
|
|
except OSError:
|
|
log.info(
|
|
"%s not present; remote-desktop profiles will be generated without a username",
|
|
CREDENTIALS_PATH,
|
|
)
|
|
return values
|
|
|
|
|
|
class RemoteDesktop:
|
|
def __init__(self, sway):
|
|
self.sway = sway
|
|
self.config_path = ensure_runtime_copy(CONFIG_FILENAME)
|
|
self.targets: dict[str, Target] = {}
|
|
self.current: str | None = None
|
|
self.load()
|
|
|
|
def load(self) -> None:
|
|
config = load_json(self.config_path)
|
|
entries = config.get("targets")
|
|
if not isinstance(entries, list):
|
|
log.warning("%s has no 'targets' list; no remote-desktop entities", self.config_path)
|
|
entries = []
|
|
|
|
credentials = _read_credentials()
|
|
targets: dict[str, Target] = {}
|
|
|
|
for index, entry in enumerate(entries):
|
|
if not isinstance(entry, dict):
|
|
continue
|
|
name = str(entry.get("name") or "").strip()
|
|
host = str(entry.get("host") or "").strip()
|
|
protocol = str(entry.get("protocol") or "").strip().lower()
|
|
|
|
if not name or not host:
|
|
log.warning("skipping remote-desktop target %s: needs a name and a host", index)
|
|
continue
|
|
if protocol not in PROTOCOLS:
|
|
log.warning(
|
|
"skipping remote-desktop target %r: protocol must be one of %s",
|
|
name,
|
|
", ".join(sorted(PROTOCOLS)),
|
|
)
|
|
continue
|
|
if name in targets:
|
|
log.warning("skipping duplicate remote-desktop target %r", name)
|
|
continue
|
|
|
|
try:
|
|
port = int(entry.get("port") or DEFAULT_PORTS[protocol])
|
|
except (TypeError, ValueError):
|
|
port = DEFAULT_PORTS[protocol]
|
|
|
|
target = Target(
|
|
name=name,
|
|
host=host,
|
|
port=port,
|
|
protocol=protocol,
|
|
credentials_ref=str(entry.get("credentials_ref") or "").strip(),
|
|
# Indexed, not named: the filename must not be derived from a string a
|
|
# human typed into a config file, and the index is already unique.
|
|
profile_path=os.path.join(PROFILE_DIR, f"target-{index}.remmina"),
|
|
)
|
|
targets[name] = target
|
|
self._write_profile(target, credentials)
|
|
|
|
self.targets = targets
|
|
if self.current not in self.targets:
|
|
self.current = next(iter(self.targets), None)
|
|
log.info("loaded %d remote-desktop target(s)", len(self.targets))
|
|
|
|
def options(self) -> list[str]:
|
|
return list(self.targets)
|
|
|
|
def _write_profile(self, target: Target, credentials: dict[str, str]) -> None:
|
|
prefix = re.sub(r"[^A-Z0-9]", "_", target.credentials_ref.upper())
|
|
username = credentials.get(f"{prefix}_USERNAME", "") if prefix else ""
|
|
domain = credentials.get(f"{prefix}_DOMAIN", "") if prefix else ""
|
|
|
|
# No `password=` key. Remmina asks once and keeps it in its own store; writing
|
|
# one here would mean a cleartext credential for another machine sitting on an
|
|
# unattended kiosk.
|
|
profile = "\n".join(
|
|
(
|
|
"[remmina]",
|
|
f"name={target.name}",
|
|
f"protocol={PROTOCOLS[target.protocol]}",
|
|
f"server={target.host}:{target.port}",
|
|
f"username={username}",
|
|
f"domain={domain}",
|
|
"group=SmartestHome",
|
|
"window_maximize=1",
|
|
"viewmode=1",
|
|
"scale=1",
|
|
"disableclipboard=0",
|
|
"",
|
|
)
|
|
)
|
|
|
|
try:
|
|
os.makedirs(PROFILE_DIR, exist_ok=True)
|
|
with open(target.profile_path, "w", encoding="utf-8") as handle:
|
|
handle.write(profile)
|
|
os.chmod(target.profile_path, 0o600)
|
|
except OSError as exc:
|
|
log.warning("could not write %s: %s", target.profile_path, exc)
|
|
|
|
# --- entity surface -----------------------------------------------------
|
|
def select(self, payload: str) -> str | None:
|
|
name = payload.strip()
|
|
if name not in self.targets:
|
|
log.warning("ignoring unknown remote-desktop target %r", name)
|
|
return self.current
|
|
self.current = name
|
|
log.info("remote-desktop target set to %s", name)
|
|
return self.current
|
|
|
|
def connect(self, _payload: str = "") -> None:
|
|
if self.current is None:
|
|
log.warning("no remote-desktop target selected; nothing to connect to")
|
|
return
|
|
target = self.targets[self.current]
|
|
log.info("connecting to %s (%s %s:%s)", target.name, target.protocol, target.host, target.port)
|
|
# No process_pattern: Remmina is single-instance, and `remmina -c` against a
|
|
# running instance opens the connection there. Re-running it is how you switch
|
|
# target, so "already running, just focus" would be wrong.
|
|
self.sway.launch_app(["remmina", "-c", target.profile_path], workspace=WS_WEB)
|
|
self.sway.focus_window(f'app_id="{REMMINA_APP_ID}"')
|
|
|
|
def disconnect(self, _payload: str = "") -> None:
|
|
# A fixed argv, not a pattern built from anything inbound. -x matches the exact
|
|
# process name so a stray "remmina" substring elsewhere cannot be caught.
|
|
log.info("disconnecting remote desktop")
|
|
try:
|
|
subprocess.run(
|
|
["pkill", "-u", str(os.getuid()), "-x", "remmina"],
|
|
capture_output=True,
|
|
timeout=5,
|
|
check=False,
|
|
)
|
|
except (OSError, subprocess.TimeoutExpired) as exc:
|
|
log.warning("could not stop remmina: %s", exc)
|