mirror of
https://github.com/pdxlocations/contact.git
synced 2026-07-28 04:23:00 +02:00
Enhance remote admin verification to establish authorized sessions and update related tests
This commit is contained in:
@@ -1187,7 +1187,7 @@ def handle_backtick(stdscr: curses.window) -> None:
|
||||
except RemoteAdminCancelled:
|
||||
logging.info("Remote admin request cancelled for %s", remote_node_num)
|
||||
remote_status(None)
|
||||
except SystemExit as exc:
|
||||
except (SystemExit, PermissionError) as exc:
|
||||
logging.warning("Remote admin was rejected for %s: %s", remote_node_num, exc)
|
||||
if wait_win is not None:
|
||||
wait_win.erase()
|
||||
|
||||
@@ -321,18 +321,22 @@ def _request_remote_channels_with_timeout(node: object, cancel_callback=None) ->
|
||||
|
||||
|
||||
def verify_remote_admin(node: object, status_callback=None, cancel_callback=None) -> None:
|
||||
"""Confirm a remote node accepts admin requests before opening its menu.
|
||||
"""Establish and verify an authorized remote-admin session.
|
||||
|
||||
A single device-config request provides an authorization check without
|
||||
eagerly downloading every radio and module setting. The successful
|
||||
response also seeds the first radio section the user may open.
|
||||
Config reads alone are not a sufficient authorization check. The
|
||||
Meshtastic admin-session exchange only provides a session passkey when
|
||||
the remote node accepts this client's administrative credentials.
|
||||
"""
|
||||
device_field = node.localConfig.DESCRIPTOR.fields_by_name.get("device")
|
||||
if device_field is None:
|
||||
raise RuntimeError("The remote node does not expose a device configuration section.")
|
||||
if status_callback:
|
||||
status_callback("Checking remote-admin permission…")
|
||||
_request_remote_with_timeout(node.requestConfig, device_field, cancel_callback=cancel_callback)
|
||||
_request_remote_with_timeout(node.ensureSessionKey, cancel_callback=cancel_callback)
|
||||
|
||||
try:
|
||||
remote_session = node.iface._getOrCreateByNum(node.nodeNum).get("adminSessionPassKey")
|
||||
except (AttributeError, TypeError):
|
||||
remote_session = None
|
||||
if not remote_session:
|
||||
raise PermissionError("The selected node did not authorize this admin request.")
|
||||
|
||||
|
||||
def _request_remote_section(
|
||||
|
||||
@@ -53,11 +53,11 @@ class ControlUiTests(unittest.TestCase):
|
||||
reconnect.assert_called_once_with(stdscr, interface)
|
||||
self.assertIs(result, new_interface)
|
||||
|
||||
def test_verify_remote_admin_requests_only_device_config(self) -> None:
|
||||
device_field = object()
|
||||
def test_verify_remote_admin_establishes_an_authorized_session(self) -> None:
|
||||
node = SimpleNamespace(
|
||||
localConfig=SimpleNamespace(DESCRIPTOR=SimpleNamespace(fields_by_name={"device": device_field})),
|
||||
requestConfig=mock.Mock(),
|
||||
nodeNum=123,
|
||||
ensureSessionKey=mock.Mock(),
|
||||
iface=SimpleNamespace(_getOrCreateByNum=mock.Mock(return_value={"adminSessionPassKey": b"key"})),
|
||||
)
|
||||
status_callback = mock.Mock()
|
||||
|
||||
@@ -65,7 +65,19 @@ class ControlUiTests(unittest.TestCase):
|
||||
control_ui.verify_remote_admin(node, status_callback=status_callback)
|
||||
|
||||
status_callback.assert_called_once_with("Checking remote-admin permission…")
|
||||
request.assert_called_once_with(node.requestConfig, device_field, cancel_callback=None)
|
||||
request.assert_called_once_with(node.ensureSessionKey, cancel_callback=None)
|
||||
node.iface._getOrCreateByNum.assert_called_once_with(123)
|
||||
|
||||
def test_verify_remote_admin_rejects_missing_session_key(self) -> None:
|
||||
node = SimpleNamespace(
|
||||
nodeNum=123,
|
||||
ensureSessionKey=mock.Mock(),
|
||||
iface=SimpleNamespace(_getOrCreateByNum=mock.Mock(return_value={})),
|
||||
)
|
||||
|
||||
with mock.patch.object(control_ui, "_request_remote_with_timeout"):
|
||||
with self.assertRaises(PermissionError):
|
||||
control_ui.verify_remote_admin(node)
|
||||
|
||||
def test_redraw_main_ui_after_reconnect_refreshes_channels_nodes_and_layout(self) -> None:
|
||||
stdscr = mock.Mock()
|
||||
|
||||
Reference in New Issue
Block a user