feat: shared PTY terminal (M12 Tier 3) + readable form controls
release / release (push) Successful in 14s

- feat(share): host-consented interactive terminal over the relay. The host shares
  a real PTY shell (core/pty_session.py); the guest renders it with pyte and sends
  keystrokes (gui/terminal_widget.py) — vim/top/tab-completion/Ctrl-C work. Runs as
  the host's user (never root). The host reads along live and can type too, e.g. a
  sudo password, which stays local and is never sent to the guest. Off by default.
  Guest also pulls inventory on join (req_full).
- fix(gui): style all form controls (QLineEdit/QPlainTextEdit/spin boxes/combo/
  terminals) dark-on-light-text — Fusion defaulted them to unreadable light-on-light.
- replaces the command/response shell with the full PTY; adds pyte to the gui extra.

Verified end-to-end against the deployed relay (guest keystroke ran on host PTY).

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
This commit is contained in:
2026-05-21 20:16:22 +02:00
parent 67d4c1cb99
commit 2f6cab72c4
9 changed files with 368 additions and 55 deletions
+11
View File
@@ -5,6 +5,17 @@ All notable changes to RigDoctor are recorded here. Format follows
(`MAJOR.MINOR.PATCH`, pre-1.0). `__version__` and `pyproject.toml` must match the git
release tag (so the auto-updater, D18, can compare versions).
## [0.7.0] - 2026-05-21
### Added
- **Shared terminal (M12, Tier 3)**: when the host enables it, the session shares a real **PTY**
shell — the guest gets an interactive terminal (vim, top, tab-completion, Ctrl-C) running on
the host as the host's user. The host **reads along** live and can type too, e.g. a `sudo`
password — which stays local and is never sent to the guest. Off by default, host-consented.
The guest also pulls the host's inventory on join.
### Fixed
- **Input contrast**: all form controls (text fields, spin boxes, combo boxes, terminals) now
use the dark theme with readable text (Fusion defaulted them to light-on-light).
## [0.6.0] - 2026-05-21
### Added
- **Session sharing over the relay (M12)**: a **Share** tab — *Start shared session* (host)
+2 -2
View File
@@ -60,8 +60,8 @@ Escalating ladder, built in order:
it in RigDoctor. One-way, safest.
- [x] Tier 2: live read-only view — `rigdoctor share serve` (stdlib HTTP, token-gated:
sensors + health + inventory). Remote = user-chosen tunnel; GUI controls still to add.
- [ ] Tier 3: gated interactive terminal (wrap tmate/sshx; read-only default, read-write on
explicit consent), with session audit log.
- [x] Tier 3: host-consented interactive terminal — a real PTY shell shared over the relay
(own `pty`, pyte-rendered guest), off by default; host reads along + can type (sudo).
> **Out of scope:** stress/repro module (D7); multi-distro support and packaging beyond
> Ubuntu/apt + `.deb` (D15) — a thin seam is kept but not built out.
+2 -2
View File
@@ -4,7 +4,7 @@ build-backend = "setuptools.build_meta"
[project]
name = "rigdoctor"
version = "0.6.0"
version = "0.7.0"
description = "Modular hardware monitoring & crash diagnostics for Linux gamers."
readme = "README.md"
requires-python = ">=3.11"
@@ -13,7 +13,7 @@ requires-python = ">=3.11"
dependencies = []
[project.optional-dependencies]
gui = ["PySide6"]
gui = ["PySide6", "pyte"]
[project.scripts]
rigdoctor = "rigdoctor.cli:main"
+1 -1
View File
@@ -1,3 +1,3 @@
"""RigDoctor — modular hardware monitoring & crash diagnostics for Linux gamers."""
__version__ = "0.6.0"
__version__ = "0.7.0"
+59
View File
@@ -0,0 +1,59 @@
"""A pseudo-terminal running the host's shell (M12, Tier 3 — host side).
Spawns the user's login shell in a real PTY so interactive programs work over a shared
session: vim, top, tab-completion, colours, Ctrl-C, and `sudo` (which prompts inside the
PTY — the host types that password locally, so it's never sent to the guest). Runs as the
host's own user — never elevated. Linux-only (uses `pty`/`termios`).
"""
from __future__ import annotations
import fcntl
import os
import pty
import signal
import struct
import termios
class PtySession:
def __init__(self, rows: int = 24, cols: int = 80):
self.pid, self.master_fd = pty.fork()
if self.pid == 0: # child: become the shell
os.environ["TERM"] = "xterm-256color"
shell = os.environ.get("SHELL", "/bin/bash")
try:
os.execvp(shell, [shell])
finally:
os._exit(1)
os.set_blocking(self.master_fd, False)
self.set_size(rows, cols)
def set_size(self, rows: int, cols: int) -> None:
try:
fcntl.ioctl(self.master_fd, termios.TIOCSWINSZ, struct.pack("HHHH", rows, cols, 0, 0))
except OSError:
pass
def write(self, data: bytes) -> None:
try:
os.write(self.master_fd, data)
except OSError:
pass
def read(self, size: int = 65536) -> bytes:
try:
return os.read(self.master_fd, size)
except (BlockingIOError, OSError):
return b""
def close(self) -> None:
try:
os.close(self.master_fd)
except OSError:
pass
try:
os.kill(self.pid, signal.SIGHUP)
os.waitpid(self.pid, os.WNOHANG)
except (OSError, ChildProcessError, ProcessLookupError):
pass
+170 -50
View File
@@ -1,12 +1,20 @@
"""Share page (M12, Tier 2 over the relay): host or join a read-only shared session."""
"""Share page (M12): host or join a shared session over the relay.
Guest sees the host's live sensors + health + inventory (read-only). If the host enables it,
a full **PTY terminal** is shared: the guest types and the commands run on the host (as the
host's user), the host reads along, and the host can type too — e.g. a sudo password, which
stays local and is never sent to the guest.
"""
from __future__ import annotations
import base64
import json
from PySide6.QtCore import Qt, QTimer, QUrl
from PySide6.QtCore import Qt, QSocketNotifier, QTimer, QUrl
from PySide6.QtWebSockets import QWebSocket
from PySide6.QtWidgets import (
QCheckBox,
QFrame,
QHBoxLayout,
QLabel,
@@ -19,14 +27,20 @@ from PySide6.QtWidgets import (
from ..config import load_config, load_token
from ..core import share
from ..core.pty_session import PtySession
from ..core.sampler import Sampler
from ..core.sources import available_sources
from .terminal_widget import TerminalView
def _relay_url() -> str:
return load_config().get("relay_url", "wss://rigdoctor.jesseyvanofferen.com").rstrip("/")
def _b64(data: bytes) -> str:
return base64.b64encode(data).decode("ascii")
def _card(title: str) -> tuple[QFrame, QVBoxLayout]:
card = QFrame()
card.setObjectName("Card")
@@ -46,6 +60,8 @@ class SharePage(QWidget):
self._sampler = Sampler(available_sources())
self._host_ws: QWebSocket | None = None
self._guest_ws: QWebSocket | None = None
self._pty: PtySession | None = None
self._pty_notifier: QSocketNotifier | None = None
self._last_report = None
self._last_inv = None
self._timer = QTimer(self)
@@ -54,18 +70,22 @@ class SharePage(QWidget):
root = QVBoxLayout(self)
root.setContentsMargins(20, 18, 20, 18)
root.setSpacing(16)
root.setSpacing(14)
title = QLabel("Share")
title.setObjectName("PageTitle")
root.addWidget(title)
root.addWidget(self._build_host())
root.addWidget(self._build_guest(), 1)
# Host
host_card, hv = _card("Start a shared session")
# ------------------------------------------------------------------ host
def _build_host(self) -> QFrame:
card, v = _card("Start a shared session")
self._host_status = QLabel("Let someone with an account view your machine, read-only.")
self._host_status.setObjectName("Muted")
self._host_status.setWordWrap(True)
hv.addWidget(self._host_status)
hrow = QHBoxLayout()
v.addWidget(self._host_status)
row = QHBoxLayout()
self._start_btn = QPushButton("Start shared session")
self._start_btn.setObjectName("PrimaryButton")
self._start_btn.clicked.connect(self._start_host)
@@ -75,45 +95,25 @@ class SharePage(QWidget):
self._code_label = QLabel("")
self._code_label.setStyleSheet("font-weight:700; font-size:18px; color:#38bdf8; background:transparent;")
self._code_label.setTextInteractionFlags(Qt.TextInteractionFlag.TextSelectableByMouse)
hrow.addWidget(self._start_btn)
hrow.addWidget(self._stop_btn)
hrow.addSpacing(12)
hrow.addWidget(self._code_label)
hrow.addStretch(1)
hv.addLayout(hrow)
root.addWidget(host_card)
row.addWidget(self._start_btn)
row.addWidget(self._stop_btn)
row.addSpacing(12)
row.addWidget(self._code_label)
row.addStretch(1)
v.addLayout(row)
# Guest
guest_card, gv = _card("Join a shared session")
grow = QHBoxLayout()
self._code_input = QLineEdit()
self._code_input.setPlaceholderText("Enter share code")
self._code_input.setMaxLength(6)
self._code_input.setFixedWidth(160)
self._join_btn = QPushButton("Join")
self._join_btn.setObjectName("PrimaryButton")
self._join_btn.clicked.connect(self._join)
self._leave_btn = QPushButton("Leave")
self._leave_btn.setEnabled(False)
self._leave_btn.clicked.connect(self._leave)
grow.addWidget(self._code_input)
grow.addWidget(self._join_btn)
grow.addWidget(self._leave_btn)
grow.addStretch(1)
gv.addLayout(grow)
self._guest_status = QLabel("")
self._guest_status.setObjectName("Muted")
gv.addWidget(self._guest_status)
root.addWidget(guest_card)
self._allow_term = QCheckBox("Allow remote terminal — the guest runs commands as your user (you read along; you can type too, e.g. a sudo password)")
self._allow_term.setStyleSheet("color:#fb923c; background:transparent;")
self._allow_term.toggled.connect(self._toggle_terminal)
v.addWidget(self._allow_term)
self._view = QTextEdit()
self._view.setObjectName("Report")
self._view.setReadOnly(True)
self._view.setVisible(False)
root.addWidget(self._view, 1)
root.addStretch(0)
self._host_term = TerminalView()
self._host_term.keys.connect(lambda b: self._pty.write(b) if self._pty else None)
self._host_term.resized.connect(lambda r, c: self._pty.set_size(r, c) if self._pty else None)
self._host_term.setVisible(False)
v.addWidget(self._host_term)
return card
# --- host ---------------------------------------------------------------
def _start_host(self) -> None:
if not load_token():
self._host_status.setText("Set a Gitea access token in Setup → Account access first.")
@@ -135,13 +135,66 @@ class SharePage(QWidget):
if data.get("error"):
self._host_status.setText(f"Rejected: {data['error']}")
return
if "code" in data:
if "code" in data: # relay handshake
self._code_label.setText(data["code"])
self._host_status.setText(f"Sharing as {data.get('user', '?')} — give this code to whoever should view your machine.")
self._stop_btn.setEnabled(True)
self._host_ws.sendTextMessage(share.host_full_frame(self._sampler))
self._send_terminal_state()
if self._allow_term.isChecked():
self._start_pty()
self._timer.start()
# guest input also arrives here; ignored (read-only session)
return
kind = data.get("type") # frames forwarded from a guest
if kind == "req_full":
self._host_ws.sendTextMessage(share.host_full_frame(self._sampler))
elif kind == "pty_in" and self._pty:
self._pty.write(base64.b64decode(data["data"]))
elif kind == "pty_resize" and self._pty:
self._pty.set_size(int(data["rows"]), int(data["cols"]))
def _toggle_terminal(self, on: bool) -> None:
if on and self._host_ws and self._code_label.text():
self._start_pty()
elif not on:
self._stop_pty()
self._send_terminal_state()
def _send_terminal_state(self) -> None:
if self._host_ws and self._code_label.text():
self._host_ws.sendTextMessage(json.dumps({"type": "terminal", "enabled": self._allow_term.isChecked()}))
def _start_pty(self) -> None:
if self._pty:
return
rows, cols = self._host_term.grid()
self._pty = PtySession(rows=rows, cols=cols)
self._pty_notifier = QSocketNotifier(self._pty.master_fd, QSocketNotifier.Type.Read, self)
self._pty_notifier.activated.connect(self._on_pty_output)
self._host_term.reset()
self._host_term.setVisible(True)
def _on_pty_output(self) -> None:
if not self._pty:
return
data = self._pty.read()
if not data: # shell exited / EOF
self._stop_pty()
self._send_terminal_state()
self._allow_term.setChecked(False)
return
self._host_term.feed(data)
if self._host_ws:
self._host_ws.sendTextMessage(json.dumps({"type": "pty", "data": _b64(data)}))
def _stop_pty(self) -> None:
if self._pty_notifier:
self._pty_notifier.setEnabled(False)
self._pty_notifier = None
if self._pty:
self._pty.close()
self._pty = None
self._host_term.setVisible(False)
def _stream(self) -> None:
if self._host_ws:
@@ -149,6 +202,7 @@ class SharePage(QWidget):
def _stop_host(self) -> None:
self._timer.stop()
self._stop_pty()
if self._host_ws:
self._host_ws.close()
self._host_ws = None
@@ -159,13 +213,54 @@ class SharePage(QWidget):
def _host_closed(self) -> None:
self._timer.stop()
self._stop_pty()
self._start_btn.setEnabled(True)
self._stop_btn.setEnabled(False)
if self._code_label.text():
self._code_label.setText("")
self._host_status.setText("Disconnected from the relay.")
# --- guest --------------------------------------------------------------
# ----------------------------------------------------------------- guest
def _build_guest(self) -> QFrame:
card, v = _card("Join a shared session")
row = QHBoxLayout()
self._code_input = QLineEdit()
self._code_input.setPlaceholderText("Enter share code")
self._code_input.setMaxLength(6)
self._code_input.setFixedWidth(160)
self._join_btn = QPushButton("Join")
self._join_btn.setObjectName("PrimaryButton")
self._join_btn.clicked.connect(self._join)
self._leave_btn = QPushButton("Leave")
self._leave_btn.setEnabled(False)
self._leave_btn.clicked.connect(self._leave)
row.addWidget(self._code_input)
row.addWidget(self._join_btn)
row.addWidget(self._leave_btn)
row.addStretch(1)
v.addLayout(row)
self._guest_status = QLabel("")
self._guest_status.setObjectName("Muted")
v.addWidget(self._guest_status)
self._view = QTextEdit()
self._view.setObjectName("Report")
self._view.setReadOnly(True)
self._view.setVisible(False)
self._view.setMinimumHeight(200)
v.addWidget(self._view)
self._term_label = QLabel("")
self._term_label.setObjectName("Muted")
self._term_label.setVisible(False)
v.addWidget(self._term_label)
self._guest_term = TerminalView()
self._guest_term.keys.connect(self._guest_key)
self._guest_term.resized.connect(self._guest_resize)
self._guest_term.setVisible(False)
v.addWidget(self._guest_term)
return card
def _join(self) -> None:
code = self._code_input.text().strip().upper()
if not load_token():
@@ -195,19 +290,43 @@ class SharePage(QWidget):
self._guest_status.setText(f"Viewing {data.get('host', '?')}'s machine — read-only.")
self._leave_btn.setEnabled(True)
self._view.setVisible(True)
self._guest_ws.sendTextMessage(json.dumps({"type": "req_full"}))
return
kind = data.get("type")
if kind == "full":
self._last_report = data.get("report")
self._last_inv = data.get("inventory")
if kind in ("full", "snapshot"):
if kind == "full":
self._last_report = data.get("report")
self._last_inv = data.get("inventory")
self._view.setHtml(share.guest_html(data.get("snapshot"), self._last_report, self._last_inv))
elif kind == "terminal":
self._set_terminal_visible(bool(data.get("enabled")))
elif kind == "pty":
self._guest_term.feed(base64.b64decode(data["data"]))
def _set_terminal_visible(self, enabled: bool) -> None:
self._term_label.setVisible(True)
self._term_label.setText("Terminal enabled by host — your keystrokes run on their machine. Click here and type."
if enabled else "Terminal not enabled by the host.")
self._guest_term.setVisible(enabled)
if enabled:
self._guest_term.reset()
self._guest_resize(*self._guest_term.grid())
self._guest_term.setFocus()
def _guest_key(self, data: bytes) -> None:
if self._guest_ws:
self._guest_ws.sendTextMessage(json.dumps({"type": "pty_in", "data": _b64(data)}))
def _guest_resize(self, rows: int, cols: int) -> None:
if self._guest_ws:
self._guest_ws.sendTextMessage(json.dumps({"type": "pty_resize", "rows": rows, "cols": cols}))
def _leave(self) -> None:
if self._guest_ws:
self._guest_ws.close()
self._guest_ws = None
self._view.setVisible(False)
for w in (self._view, self._term_label, self._guest_term):
w.setVisible(False)
self._leave_btn.setEnabled(False)
self._join_btn.setEnabled(True)
self._guest_status.setText("Left the session.")
@@ -220,6 +339,7 @@ class SharePage(QWidget):
def shutdown(self) -> None:
self._timer.stop()
self._stop_pty()
for ws in (self._host_ws, self._guest_ws):
if ws:
ws.close()
+84
View File
@@ -0,0 +1,84 @@
"""A minimal terminal view: renders PTY output via pyte and emits keystrokes (M12, Tier 3).
Used by both sides of a shared session — the host (mirrors its local PTY, can also type, e.g.
a sudo password) and the guest (renders the streamed PTY, sends keystrokes). Monochrome for
now; cursor addressing / layout (vim, top) work via pyte.
"""
from __future__ import annotations
import pyte
from PySide6.QtCore import Qt, Signal
from PySide6.QtGui import QFontDatabase, QFontMetrics
from PySide6.QtWidgets import QPlainTextEdit
class TerminalView(QPlainTextEdit):
keys = Signal(bytes) # user keystrokes -> bytes for the PTY
resized = Signal(int, int) # rows, cols
def __init__(self, rows: int = 24, cols: int = 80):
super().__init__()
self.setLineWrapMode(QPlainTextEdit.LineWrapMode.NoWrap)
self.setFont(QFontDatabase.systemFont(QFontDatabase.SystemFont.FixedFont))
self.setUndoRedoEnabled(False)
self.setMinimumHeight(260)
self._rows, self._cols = rows, cols
self._screen = pyte.Screen(cols, rows)
self._stream = pyte.ByteStream(self._screen)
def grid(self) -> tuple[int, int]:
return self._rows, self._cols
def feed(self, data: bytes) -> None:
self._stream.feed(data)
self._render()
def reset(self) -> None:
self._screen.reset()
self._render()
def _render(self) -> None:
bar = self.verticalScrollBar().value()
self.setPlainText("\n".join(self._screen.display))
self.verticalScrollBar().setValue(bar)
def resizeEvent(self, event): # noqa: N802 (Qt override)
super().resizeEvent(event)
fm = QFontMetrics(self.font())
cw = max(1, fm.horizontalAdvance("M"))
ch = max(1, fm.height())
cols = max(20, self.viewport().width() // cw)
rows = max(6, self.viewport().height() // ch)
if (rows, cols) != (self._rows, self._cols):
self._rows, self._cols = rows, cols
self._screen.resize(rows, cols)
self._render()
self.resized.emit(rows, cols)
def keyPressEvent(self, event): # noqa: N802 (Qt override)
data = self._translate(event)
if data:
self.keys.emit(data)
event.accept() # display comes from PTY output, not local editing
@staticmethod
def _translate(event) -> bytes:
key = event.key()
mod = event.modifiers()
k = Qt.Key
if mod & Qt.KeyboardModifier.ControlModifier and k.Key_A.value <= key <= k.Key_Z.value:
return bytes([key - k.Key_A.value + 1]) # Ctrl-A..Ctrl-Z
special = {
k.Key_Return.value: b"\r", k.Key_Enter.value: b"\r",
k.Key_Backspace.value: b"\x7f", k.Key_Tab.value: b"\t",
k.Key_Escape.value: b"\x1b",
k.Key_Up.value: b"\x1b[A", k.Key_Down.value: b"\x1b[B",
k.Key_Right.value: b"\x1b[C", k.Key_Left.value: b"\x1b[D",
k.Key_Home.value: b"\x1b[H", k.Key_End.value: b"\x1b[F",
k.Key_Delete.value: b"\x1b[3~", k.Key_PageUp.value: b"\x1b[5~", k.Key_PageDown.value: b"\x1b[6~",
}
if key in special:
return special[key]
text = event.text()
return text.encode("utf-8") if text else b""
+12
View File
@@ -14,6 +14,7 @@ CARD_BORDER = "#2a2f39"
TRACK = "#2a2f39"
TEXT = "#e6e8eb"
MUTED = "#8b929c"
INPUT_BG = "#0d0f13" # form-control background (must stay dark — see contrast rule)
ACCENT = "#38bdf8"
COLD = "#7dd3fc" # icey-blue
@@ -138,4 +139,15 @@ QCheckBox::indicator:checked {{
QDialog {{ background: {BG}; }}
QMessageBox {{ background: {CARD}; }}
QDialog QLabel, QMessageBox QLabel {{ color: {TEXT}; background: transparent; }}
/* Form controls: keep dark bg + light text (Fusion defaults to light-on-light here). */
QLineEdit, QPlainTextEdit, QAbstractSpinBox, QComboBox {{
background: {INPUT_BG}; color: {TEXT};
border: 1px solid {CARD_BORDER}; border-radius: 6px; padding: 5px 8px;
selection-background-color: {ACCENT}; selection-color: #06222e;
}}
QLineEdit:focus, QPlainTextEdit:focus, QAbstractSpinBox:focus, QComboBox:focus {{
border: 1px solid {ACCENT};
}}
QLineEdit:disabled, QPlainTextEdit:disabled, QAbstractSpinBox:disabled {{ color: {MUTED}; }}
"""
+27
View File
@@ -0,0 +1,27 @@
"""Tests for the host PTY session (M12 Tier 3)."""
import time
import unittest
from rigdoctor.core.pty_session import PtySession
class PtySessionTests(unittest.TestCase):
def test_runs_command_and_reads_output(self):
pty = PtySession(rows=24, cols=80)
try:
time.sleep(0.4)
pty.read() # drain the shell prompt
pty.write(b"echo PTY_MARKER_42\n")
deadline = time.time() + 3
buf = ""
while time.time() < deadline and "PTY_MARKER_42" not in buf:
time.sleep(0.1)
buf += pty.read().decode(errors="replace")
self.assertIn("PTY_MARKER_42", buf)
finally:
pty.close()
if __name__ == "__main__":
unittest.main()