From 4c7d12c453bd626c448906c91808653c0301842d Mon Sep 17 00:00:00 2001 From: Vincent Michel Date: Tue, 28 Jul 2026 20:16:14 +0200 Subject: [PATCH 1/2] Implement --robot-check for ssh --- gambaterm/ssh.py | 96 +++++++++++++++++++++++++++++++++++ tests/_frontend_ssh_server.py | 1 + 2 files changed, 97 insertions(+) diff --git a/gambaterm/ssh.py b/gambaterm/ssh.py index 036a1ab..db95dfc 100644 --- a/gambaterm/ssh.py +++ b/gambaterm/ssh.py @@ -1,6 +1,7 @@ from __future__ import annotations import os +import re import time import hmac import asyncio @@ -17,6 +18,7 @@ import structlog from asyncssh import ( SFTPServerFactory, + SSHReader, SSHServerConnection, SSHServerProcess, SSHAcceptor, @@ -59,6 +61,71 @@ [str, argparse.Namespace, Writer], argparse.Namespace ] +CPR_PATTERN = re.compile(r"\x1b\[(\d+);(\d+)R") + + +async def _read_cpr_response( + reader: SSHReader[str], +) -> tuple[int, int] | None: + buf = "" + while True: + try: + data = await reader.read(1) + except UnicodeDecodeError: + return None + if not data: + return None + buf += data + if buf.endswith("R"): + match = CPR_PATTERN.search(buf) + if match: + return (int(match.group(1)), int(match.group(2))) + + +async def get_cursor_position( + process: SSHServerProcess[str], + timeout: float = 1.0, +) -> tuple[int, int] | None: + # Send Device Status Report request + process.stdout.write("\x1b[6n") + await process.stdout.drain() + + # Read response: ESC [ row ; col R + try: + return await asyncio.wait_for(_read_cpr_response(process.stdin), timeout) + except asyncio.TimeoutError: + return None + + +async def do_robot_check( + process: SSHServerProcess[str], + timeout: float = 1.0, +) -> tuple[bool, float]: + start1 = time.perf_counter() + pos1 = await get_cursor_position(process, timeout) + if pos1 is None: + return False, 0.0 + delta1 = time.perf_counter() - start1 + + # Write test character + process.stdout.write(" ") + await process.stdout.drain() + + start2 = time.perf_counter() + pos2 = await get_cursor_position(process, timeout) + if pos2 is None: + return False, 0.0 + delta2 = time.perf_counter() - start2 + + # Clear the test character + process.stdout.write("\b") + await process.stdout.drain() + + _, x1 = pos1 + _, x2 = pos2 + delta = (delta1 + delta2) / 2 + return x2 - x1 == 1, delta + class InputSource(Enum): INPUT_FILE = auto() @@ -105,6 +172,7 @@ async def ssh_process_handler(process: SSHServerProcess[str]) -> int: command_parser: CommandParser = process.get_extra_info("command_parser") users_directory: Path = process.get_extra_info("users_directory") executor: ThreadPoolExecutor = process.get_extra_info("executor") + robot_check: bool = process.get_extra_info("robot_check") display = process.channel.get_x11_display() command = process.channel.get_command() terminal_type = process.get_terminal_type() @@ -136,11 +204,26 @@ async def ssh_process_handler(process: SSHServerProcess[str]) -> int: "Please use a terminal to access the interactive interface.", "Use `-t` to force pseudo-terminal allocation if a command is provided.", sep="\r\n", + end="\r\n", file=process.stdout, ) session_logger.warning("User did not use an interactive terminal") return 1 + # Robot check + if robot_check: + session_logger.info("Perform robot check") + passed, round_trip = await do_robot_check(process) + if not passed: + print( + "Your terminal does not seem to support cursor postion request (CPR).", + end="\r\n", + file=process.stdout, + ) + session_logger.warning("Terminal did not pass the robot check") + return 1 + session_logger.info("Robot check passed", round_trip=round_trip) + return await process_to_terminal( process, executor, @@ -329,6 +412,7 @@ class GambatermSSHServer(SSHServer): def __init__( self, authentication: AuthenticationMethod, + robot_check: bool, console_cls: type[Console], namespace: argparse.Namespace, command_parser: CommandParser, @@ -346,6 +430,7 @@ def __init__( self._gambaterm_users_directory = users_directory self._gambaterm_executor = executor self._gambaterm_authentication = authentication + self._gambaterm_robot_check = robot_check self._gambaterm_active_connections = active_connections self._gambaterm_active_sessions: set[GambatermSSHServerProcess] = set() self._gambaterm_frontend = frontend @@ -358,6 +443,7 @@ def connection_made(self, conn: SSHServerConnection) -> None: conn.set_extra_info(command_parser=self._gambaterm_command_parser) conn.set_extra_info(users_directory=self._gambaterm_users_directory) conn.set_extra_info(frontend=self._gambaterm_frontend) + conn.set_extra_info(robot_check=self._gambaterm_robot_check) self._gambaterm_active_connections[self] = conn def connection_lost(self, exc: Exception | None) -> None: @@ -404,6 +490,7 @@ async def run_ssh_server( bind: str, port: int, authentication: AuthenticationMethod, + robot_check: bool, console_cls: type[Console], namespace: argparse.Namespace, command_parser: CommandParser, @@ -466,6 +553,7 @@ async def run_ssh_server( server = await asyncssh.create_server( lambda: GambatermSSHServer( authentication, + robot_check, console_cls, namespace, command_parser, @@ -561,6 +649,12 @@ def main( action="store_true", help="Disable authentication altogether (no password nor public key required)", ) + parser.add_argument( + "--robot-check", + action="store_true", + default=False, + help="reject bots by checking if client responds to cursor position requests", + ) parser.add_argument( "--users-directory", type=Path, @@ -574,6 +668,7 @@ def main( port: int = namespace.__dict__.pop("port") password: str = namespace.__dict__.pop("password") no_auth: bool = namespace.__dict__.pop("no_auth") + robot_check: bool = namespace.__dict__.pop("robot_check") users_directory: Path = namespace.__dict__.pop("users_directory") # Determine authentication method @@ -612,6 +707,7 @@ async def async_main() -> None: bind, port, authentication, + robot_check, console_cls, namespace, command_parser, diff --git a/tests/_frontend_ssh_server.py b/tests/_frontend_ssh_server.py index 2a71fba..a008675 100644 --- a/tests/_frontend_ssh_server.py +++ b/tests/_frontend_ssh_server.py @@ -43,6 +43,7 @@ async def main() -> None: bind="127.0.0.1", port=8022, authentication=gambaterm_ssh.NoAuthentication(), + robot_check=False, console_cls=GameboyColor, namespace=namespace, command_parser=lambda cmd, ns, write: ns, From c19264c7af5e6cdf3e118b0082dd1518f34bff6c Mon Sep 17 00:00:00 2001 From: Vincent Michel Date: Tue, 28 Jul 2026 20:22:42 +0200 Subject: [PATCH 2/2] Bump mypy and flake8 --- .pre-commit-config.yaml | 4 ++-- gambaterm/run.py | 2 +- 2 files changed, 3 insertions(+), 3 deletions(-) diff --git a/.pre-commit-config.yaml b/.pre-commit-config.yaml index 77d98af..9d62fcd 100644 --- a/.pre-commit-config.yaml +++ b/.pre-commit-config.yaml @@ -9,13 +9,13 @@ repos: hooks: - id: flake8 - repo: https://github.com/pre-commit/pre-commit-hooks - rev: "v4.5.0" + rev: "v6.0.0" hooks: - id: mixed-line-ending - id: trailing-whitespace exclude: \.patch$ - repo: https://github.com/pre-commit/mirrors-mypy - rev: "v1.7.0" + rev: "v2.3.0" hooks: - id: mypy additional_dependencies: [ diff --git a/gambaterm/run.py b/gambaterm/run.py index b7d0e69..51265c3 100644 --- a/gambaterm/run.py +++ b/gambaterm/run.py @@ -36,7 +36,7 @@ def get_ref(width: int, height: int, console: Console) -> tuple[int, int]: return refx, refy -def write_frame(term: Terminal, frame_data: bytes) -> None: +def write_frame(term: Terminal, frame_data: bytes | bytearray) -> None: # Fix code page issue on windows: # `sys.stdout.buffer.raw` is a `WindowsConsoleIO` that always support UTF-8 # regardless of the configured codepage