From d89d2a0c04978f0129f969d89b08ab718d203795 Mon Sep 17 00:00:00 2001 From: Tyler Goodlet Date: Sun, 5 Oct 2025 12:32:53 -0400 Subject: [PATCH] Support per-`ib.vnc_addrs` vnc passwords Such that the `brokers.toml` can contain any of the following = dict|tuple styles, ```toml [ib.vnc_addrs] 4002 = {host = 'localhost', port = 5900, pw = 'doggy'} # host, port, pw 4002 = {host = 'localhost', port = 5900} # host, port, pw 4002 = ['localhost', 5900] # host, port, pw ``` With the first line demonstrating a vnc-server password (as normally set via a `.env` file in the `dockering/ib/` subdir) with the `pw =` field. This obviously removes the hardcoded `'doggy'` password from prior. Impl details in `.brokers.ib._util`: - pass the `ib.api.Client` down into `vnc_click_hack()` doing all config reading within and removing host, port unpacking in the callingn `data_reset_hack()`. - also pass the client `try_xdo_manual()` and comment (with plans to remove) the recently added localhost-only fallback section since we now have a fully working py vnc client again with `pyvnc` B) - in `vnc_click_hack()` match for all the possible config line styles and, * pass any `pw` field to `pyvncVNCConfig`, * continue matching host, port without password, * fallthrough to raising a val-err when neither ^ match. --- piker/brokers/ib/_util.py | 105 ++++++++++++++++++++++++-------------- 1 file changed, 67 insertions(+), 38 deletions(-) diff --git a/piker/brokers/ib/_util.py b/piker/brokers/ib/_util.py index cb3affc7..1cf50926 100644 --- a/piker/brokers/ib/_util.py +++ b/piker/brokers/ib/_util.py @@ -38,7 +38,6 @@ from piker.brokers._util import get_logger if TYPE_CHECKING: from .api import Client - from ib_insync import IB import i3ipc log = get_logger('piker.brokers.ib') @@ -62,7 +61,7 @@ no_setup_msg:str = ( def try_xdo_manual( - vnc_sockaddr: str, + client: Client, ): ''' Do the "manual" `xdo`-based screen switch + click @@ -79,6 +78,7 @@ def try_xdo_manual( _reset_tech = 'i3ipc_xdotool' return True except OSError: + vnc_sockaddr: str = client.conf.vnc_addrs log.exception( no_setup_msg.format(vnc_sockaddr=vnc_sockaddr) ) @@ -86,7 +86,6 @@ def try_xdo_manual( async def data_reset_hack( - # vnc_host: str, client: Client, reset_type: Literal['data', 'connection'], @@ -118,36 +117,24 @@ async def data_reset_hack( that need to be wrangle. ''' - ib_client: IB = client.ib - # look up any user defined vnc socket address mapped from # a particular API socket port. - api_port: str = str(ib_client.client.port) - vnc_host: str - vnc_port: int - vnc_sockaddr: tuple[str] | None = client.conf.get('vnc_addrs') - - if not vnc_sockaddr: + vnc_addrs: tuple[str]|None = client.conf.get('vnc_addrs') + if not vnc_addrs: log.warning( - no_setup_msg.format(vnc_sockaddr=vnc_sockaddr) + no_setup_msg.format(vnc_sockaddr=client.conf) + 'REQUIRES A `vnc_addrs: array` ENTRY' ) - vnc_host, vnc_port = vnc_sockaddr.get( - api_port, - ('localhost', 3003) - ) global _reset_tech - match _reset_tech: case 'vnc': try: await tractor.to_asyncio.run_task( partial( vnc_click_hack, - host=vnc_host, - port=vnc_port, + client=client, ) ) except ( @@ -158,29 +145,31 @@ async def data_reset_hack( import i3ipc # noqa (since a deps dynamic check) except ModuleNotFoundError: log.warning( - no_setup_msg.format(vnc_sockaddr=vnc_sockaddr) + no_setup_msg.format(vnc_sockaddr=client.conf) ) return False - if vnc_host not in { - 'localhost', - '127.0.0.1', - }: - focussed, matches = i3ipc_fin_wins_titled() - if not matches: - log.warning( - no_setup_msg.format(vnc_sockaddr=vnc_sockaddr) - ) - return False - else: - try_xdo_manual(vnc_sockaddr) + # XXX, Xorg only workaround.. + # TODO? remove now that we have `pyvnc`? + # if vnc_host not in { + # 'localhost', + # '127.0.0.1', + # }: + # focussed, matches = i3ipc_fin_wins_titled() + # if not matches: + # log.warning( + # no_setup_msg.format(vnc_sockaddr=vnc_sockaddr) + # ) + # return False + # else: + # try_xdo_manual(vnc_sockaddr) # localhost but no vnc-client or it borked.. else: - try_xdo_manual(vnc_sockaddr) + try_xdo_manual(client) case 'i3ipc_xdotool': - try_xdo_manual(vnc_sockaddr) + try_xdo_manual(client) # i3ipc_xdotool_manual_click_hack() case _ as tech: @@ -191,15 +180,55 @@ async def data_reset_hack( async def vnc_click_hack( - host: str, - port: int, - reset_type: str = 'data' + client: Client, + reset_type: str = 'data', + pw: str|None = None, + ) -> None: ''' Reset the data or network connection for the VNC attached ib-gateway using a (magic) keybinding combo. + A vnc-server password can be set either by an input `pw` param or + set in the client's config with the latter loaded from the user's + `brokers.toml` in a vnc-addrs-port-mapping section, + + .. code:: toml + + [ib.vnc_addrs] + 4002 = {host = 'localhost', port = 5900, pw = 'doggy'} + ''' + api_port: str = str(client.client.port) + conf: dict = client.conf + vnc_addrs: dict[int, tuple] = conf.get('vnc_addrs') + if not vnc_addrs: + return None + + addr_entry: dict|tuple = vnc_addrs.get( + api_port, + ('localhost', 5900) # a typical default + ) + if pw is None: + match addr_entry: + case ( + host, + port, + ): + pass + + case { + 'host': host, + 'port': port, + 'pw': pw + }: + pass + + case _: + raise ValueError( + f'Invalid `ib.vnc_addrs` entry ?\n' + f'{addr_entry!r}\n' + ) try: from pyvnc import ( AsyncVNCClient, @@ -226,7 +255,7 @@ async def vnc_click_hack( VNCConfig( host=host, port=port, - password='doggy', + password=pw, ) ) async with client: