Support per-`ib.vnc_addrs` vnc passwords

Such that the `brokers.toml` can contain any of the following
<port> = 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.
hist_backfill_fixes
Tyler Goodlet 2025-10-05 12:32:53 -04:00
parent cca2b14548
commit d89d2a0c04
1 changed files with 67 additions and 38 deletions

View File

@ -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: