Add --users-directory arguement to server CLIs

Author: vxgmichel

gambaterm/console.py

@@ -122,8 +122,8 @@ def pop_console_arguments(
     ) -> Callable[[], Console]:
         romfile: Path = namespace.romfile
         input_file: Path | None = namespace.input_file
+        save_directory: Path | None = namespace.save_directory
         force_gameboy: bool = namespace.__dict__.pop("force_gameboy")
-        save_directory: Path | None = namespace.__dict__.pop("save_directory")
         # Save directory defaults to the rom file directory (unless we read the input from a file)
         if input_file is None and save_directory is None:
             save_directory = romfile.parent

gambaterm/main.py

@@ -29,6 +29,7 @@ class AppConfig:
     speed: float
     skip_inputs: int
     cpr_sync: bool
+    save_directory: Path | None
 
 
 @dataclass
@@ -39,13 +40,6 @@ class LocalAppConfig(AppConfig):
 
 def add_base_arguments(parser: argparse.ArgumentParser) -> None:
     parser.add_argument("romfile", metavar="ROM", type=Path, help="Path to a rom file")
-    parser.add_argument(
-        "--save-directory",
-        "--sd",
-        type=Path,
-        default=None,
-        help="Path to the save directory (default to the ROM directory)",
-    )
 
 
 def add_input_file_arguments(parser: argparse.ArgumentParser) -> None:
@@ -119,6 +113,13 @@ def add_local_only_arguments(parser: argparse.ArgumentParser) -> None:
         type=Path,
         help="Record inputs into a file",
     )
+    parser.add_argument(
+        "--save-directory",
+        "--sd",
+        type=Path,
+        default=None,
+        help="Path to the save directory (default to the ROM directory)",
+    )
 
 
 def main(

gambaterm/remote_terminal.py

@@ -5,6 +5,7 @@
 from __future__ import annotations
 
 import codecs
+import hashlib
 import contextlib
 from typing import IO, Generator
 
@@ -84,3 +85,14 @@ def _height_and_width(self) -> WINSZ:
     def update_size(self, rows: int, columns: int) -> None:
         self._rows = rows
         self._columns = columns
+
+
+def user_directory_name(username: str | None) -> str:
+    """Hash the username into a safe directory name.
+
+    :param username: telnet/ssh-negotiated username, or ``None``
+    :returns: hex digest suitable for use as a directory name
+    """
+    if username is None:
+        return "_anonymous"
+    return hashlib.sha256(username.encode("utf-8")).hexdigest()[:16]

gambaterm/ssh.py

@@ -3,7 +3,6 @@
 import os
 import time
 import hmac
-import hashlib
 import asyncio
 import argparse
 import traceback
@@ -36,7 +35,7 @@
 )
 from .console import Console, GameboyColor
 
-from .remote_terminal import RemoteTerminal
+from .remote_terminal import RemoteTerminal, user_directory_name
 from .ssh_app_session import process_to_terminal
 
 
@@ -105,6 +104,7 @@ async def ssh_process_handler(process: SSHServerProcess[str]) -> int:
     console_cls: type[Console] = process.get_extra_info("console_cls")
     namespace: argparse.Namespace = process.get_extra_info("namespace")
     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")
     display = process.channel.get_x11_display()
     command = process.channel.get_command()
@@ -126,15 +126,15 @@ async def ssh_process_handler(process: SSHServerProcess[str]) -> int:
         )
 
     # Manage save directory — hash username to prevent path traversal
-    if "save_directory" in namespace.__dict__:
-        if getattr(namespace, "input_file", False):
-            setattr(namespace, "save_directory", None)
-        else:
-            safe_name = hashlib.sha256(username.encode("utf-8")).hexdigest()[:16]
-            save_directory = Path("ssh_save") / safe_name
-            save_directory.mkdir(parents=True, exist_ok=True)
-            (save_directory / "username").write_text(username)
-            setattr(namespace, "save_directory", save_directory)
+    namespace.save_directory = (
+        None
+        if getattr(namespace, "input_file", None)
+        else users_directory / user_directory_name(username)
+    )
+
+    if namespace.save_directory is not None:
+        namespace.save_directory.mkdir(parents=True, exist_ok=True)
+        (namespace.save_directory / "username").write_text(username)
 
     # Pop console arguments and extract configuration
     console_callback = console_cls.pop_console_arguments(namespace)
@@ -292,11 +292,13 @@ def __init__(
         console_cls: type[Console],
         namespace: argparse.Namespace,
         command_parser: CommandParser,
+        users_directory: Path,
         executor: ThreadPoolExecutor,
     ):
         self._gambaterm_console_cls = console_cls
         self._gambaterm_namespace = namespace
         self._gambaterm_command_parser = command_parser
+        self._gambaterm_users_directory = users_directory
         self._gambaterm_executor = executor
         self._gambaterm_authentication = authentication
 
@@ -305,6 +307,7 @@ def connection_made(self, conn: asyncssh.SSHServerConnection) -> None:
         conn.set_extra_info(executor=self._gambaterm_executor)
         conn.set_extra_info(namespace=self._gambaterm_namespace)
         conn.set_extra_info(command_parser=self._gambaterm_command_parser)
+        conn.set_extra_info(users_directory=self._gambaterm_users_directory)
 
     def begin_auth(self, username: str) -> bool:
         return not isinstance(self._gambaterm_authentication, NoAuthentication)
@@ -334,6 +337,7 @@ async def run_ssh_server(
     console_cls: type[Console],
     namespace: argparse.Namespace,
     command_parser: CommandParser,
+    users_directory: Path,
     executor: ThreadPoolExecutor,
 ) -> AsyncIterator[SSHAcceptor]:
     # Gambaterm configuration
@@ -389,7 +393,12 @@ async def run_ssh_server(
 
     server = await asyncssh.create_server(
         lambda: SSHServer(
-            authentication, console_cls, namespace, command_parser, executor
+            authentication,
+            console_cls,
+            namespace,
+            command_parser,
+            users_directory,
+            executor,
         ),
         bind,
         port,
@@ -465,13 +474,20 @@ def main(
         action="store_true",
         help="Disable authentication altogether (no password nor public key required)",
     )
+    parser.add_argument(
+        "--users-directory",
+        type=Path,
+        default=Path("users_save"),
+        help="Directory containing one save directory per user (default is ./users_save)",
+    )
 
     # Parse arguments
     namespace = parser.parse_args(parser_args)
     bind: str = namespace.__dict__.pop("bind")
     port: int = namespace.__dict__.pop("port")
     password: str = namespace.__dict__.pop("password")
     no_auth: bool = namespace.__dict__.pop("no_auth")
+    users_directory: Path = namespace.__dict__.pop("users_directory")
 
     # Determine authentication method
     if no_auth and password is None:
@@ -512,6 +528,7 @@ async def async_main() -> None:
                     console_cls,
                     namespace,
                     command_parser,
+                    users_directory,
                     executor,
                 ):
                     await asyncio.Future()

gambaterm/telnet.py

@@ -1,7 +1,6 @@
 from __future__ import annotations
 
 import time
-import hashlib
 import asyncio
 import argparse
 import traceback
@@ -40,19 +39,10 @@
     console_input_from_keyboard_protocol_context,
     is_kitty_keyboard_protocol_supported,
 )
-from .remote_terminal import RemoteTerminal
-from .telnet_app_session import telnet_to_terminal
-
-
-def _save_dir_name(username: str | None) -> str:
-    """Hash the username into a safe directory name.
-
-    :param username: telnet-negotiated username, or ``None``
-    :returns: hex digest suitable for use as a directory name
-    """
-    if username is None:
-        return "_anonymous"
-    return hashlib.sha256(username.encode("utf-8")).hexdigest()[:16]
+from .remote_terminal import RemoteTerminal, user_directory_name
+from .telnet_app_session import (
+    telnet_to_terminal,
+)
 
 
 def thread_target(
@@ -129,14 +119,21 @@ def make_telnet_shell(
     app_config: argparse.Namespace,
     console_cls: Type[Console],
     idle_timeout: float | None,
+    users_directory: Path,
     executor: ThreadPoolExecutor,
 ) -> ShellCallback:
     """Create a telnet shell callback with app_config and executor bound."""
 
     async def telnet_shell(reader: TelnetReader, writer: TelnetWriter) -> None:
         try:
             await _telnet_shell(
-                reader, writer, app_config, console_cls, idle_timeout, executor
+                reader,
+                writer,
+                app_config,
+                console_cls,
+                idle_timeout,
+                users_directory,
+                executor,
             )
         except (KeyboardInterrupt, EOFError):
             pass
@@ -243,6 +240,7 @@ async def _telnet_shell(
     app_config: argparse.Namespace,
     console_cls: type[Console],
     idle_timeout: float | None,
+    users_directory: Path,
     executor: ThreadPoolExecutor,
 ) -> int:
     peername = writer.get_extra_info("peername")
@@ -277,12 +275,12 @@ async def _telnet_shell(
     print(f"[Terminal Info] {peer_host}: ttype={terminal_type}, {cols}x{rows}")
 
     try:
-        # Copy namespace and set telnet-specific save directory
+        # Copy namespace and set save directory
         namespace = argparse.Namespace(**vars(app_config))
         save_directory = (
             None
             if getattr(namespace, "input_file", None)
-            else Path("telnet_save") / _save_dir_name(username)
+            else users_directory / user_directory_name(username)
         )
         namespace.save_directory = save_directory
         if save_directory is not None:
@@ -320,11 +318,18 @@ async def run_telnet_server(
     idle_timeout: float | None,
     console_cls: type[Console],
     namespace: argparse.Namespace,
+    users_directory: Path,
     executor: ThreadPoolExecutor,
 ) -> AsyncIterator[telnetlib3.Server]:
     import telnetlib3
 
-    shell = make_telnet_shell(namespace, console_cls, idle_timeout, executor)
+    shell = make_telnet_shell(
+        namespace,
+        console_cls,
+        idle_timeout,
+        users_directory,
+        executor,
+    )
 
     if robot_check or max_players > 0:
         from telnetlib3.guard_shells import ConnectionCounter, busy_shell
@@ -427,13 +432,20 @@ def main(
         default=None,
         help="Idle timeout in seconds (default is disabled)",
     )
+    parser.add_argument(
+        "--users-directory",
+        type=Path,
+        default=Path("users_save"),
+        help="Directory containing one save directory per user (default is ./users_save)",
+    )
 
     namespace = parser.parse_args(parser_args)
     bind: str = namespace.__dict__.pop("bind")
     port: int = namespace.__dict__.pop("port")
     robot_check: bool = namespace.__dict__.pop("robot_check")
     max_players: int = namespace.__dict__.pop("max_players")
     idle_timeout: float | None = namespace.__dict__.pop("idle_timeout")
+    users_directory: Path = namespace.__dict__.pop("users_directory")
 
     try:
         with ThreadPoolExecutor(max_workers=32) as executor:
@@ -447,6 +459,7 @@ async def async_main() -> None:
                     idle_timeout,
                     console_cls,
                     namespace,
+                    users_directory,
                     executor,
                 ):
                     await asyncio.Future()