mirror of
https://github.com/pdxlocations/contact.git
synced 2026-07-28 04:23:00 +02:00
Verify remote admin access before opening settings
This commit is contained in:
@@ -1,6 +1,6 @@
|
||||
import time
|
||||
|
||||
from typing import Any, Dict
|
||||
from typing import Any, Dict, Optional
|
||||
|
||||
import google.protobuf.json_format
|
||||
from meshtastic import BROADCAST_NUM
|
||||
@@ -168,7 +168,7 @@ def send_message(
|
||||
message: str,
|
||||
destination: int = BROADCAST_NUM,
|
||||
channel: int = 0,
|
||||
reply_id: int | None = None,
|
||||
reply_id: Optional[int] = None,
|
||||
reply_context: str = "",
|
||||
) -> None:
|
||||
"""
|
||||
|
||||
@@ -15,7 +15,7 @@ from contact.utilities.utils import (
|
||||
build_reply_prefix,
|
||||
)
|
||||
from contact.settings import settings_menu
|
||||
from contact.ui.control_ui import RemoteAdminCancelled
|
||||
from contact.ui.control_ui import RemoteAdminCancelled, verify_remote_admin
|
||||
from contact.message_handlers.tx_handler import send_message, send_traceroute
|
||||
from contact.utilities.utils import parse_protobuf
|
||||
from contact.ui.colors import get_color
|
||||
@@ -1031,12 +1031,19 @@ def handle_backtick(stdscr: curses.window) -> None:
|
||||
wait_win.timeout(-1)
|
||||
|
||||
try:
|
||||
remote_node = interface.getNode(remote_node_num, False)
|
||||
verify_remote_admin(
|
||||
remote_node,
|
||||
status_callback=remote_status,
|
||||
cancel_callback=remote_cancel_requested,
|
||||
)
|
||||
remote_status(None)
|
||||
settings_menu(
|
||||
stdscr,
|
||||
interface,
|
||||
# Settings requests configs explicitly; avoid getNode's
|
||||
# implicit channel request racing that setup.
|
||||
node=interface.getNode(remote_node_num, False),
|
||||
node=remote_node,
|
||||
remote=True,
|
||||
status_callback=remote_status,
|
||||
cancel_callback=remote_cancel_requested,
|
||||
@@ -1067,7 +1074,7 @@ def handle_backtick(stdscr: curses.window) -> None:
|
||||
handle_resize(stdscr, False)
|
||||
|
||||
|
||||
def show_remote_admin_wait(stdscr: curses.window, node_name: str) -> curses.window | None:
|
||||
def show_remote_admin_wait(stdscr: curses.window, node_name: str) -> Optional[curses.window]:
|
||||
"""Show progress while Meshtastic retrieves remote-admin settings."""
|
||||
message = f"Attempting remote admin of {node_name}, waiting for response..."
|
||||
try:
|
||||
@@ -1086,7 +1093,7 @@ def show_remote_admin_wait(stdscr: curses.window, node_name: str) -> curses.wind
|
||||
return None
|
||||
|
||||
|
||||
def show_connection_status(stdscr: curses.window, title: str, message: str) -> curses.window | None:
|
||||
def show_connection_status(stdscr: curses.window, title: str, message: str) -> Optional[curses.window]:
|
||||
"""Display a non-blocking connection-status dialog over the current UI."""
|
||||
try:
|
||||
height, width = stdscr.getmaxyx()
|
||||
@@ -1103,7 +1110,7 @@ def show_connection_status(stdscr: curses.window, title: str, message: str) -> c
|
||||
return None
|
||||
|
||||
|
||||
def update_remote_admin_wait(wait_win: curses.window | None, message: str) -> None:
|
||||
def update_remote_admin_wait(wait_win: Optional[curses.window], message: str) -> None:
|
||||
"""Append the current remote-admin request to the waiting overlay."""
|
||||
if wait_win is None:
|
||||
return
|
||||
|
||||
@@ -319,6 +319,21 @@ def _request_remote_channels_with_timeout(node: object, cancel_callback=None) ->
|
||||
time.sleep(0.1)
|
||||
|
||||
|
||||
def verify_remote_admin(node: object, status_callback=None, cancel_callback=None) -> None:
|
||||
"""Confirm a remote node accepts admin requests before opening its menu.
|
||||
|
||||
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.
|
||||
"""
|
||||
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)
|
||||
|
||||
|
||||
def _request_remote_section(
|
||||
node: object, menu_path: List[str], selected_option: str, status_callback=None, cancel_callback=None
|
||||
) -> bool:
|
||||
|
||||
@@ -53,14 +53,16 @@ class ContactUiTests(unittest.TestCase):
|
||||
with mock.patch.object(contact_ui, "handle_resize"):
|
||||
with mock.patch.object(contact_ui, "get_list_input", return_value="Remote admin: node"):
|
||||
with mock.patch.object(contact_ui, "get_name_from_database", return_value="node"):
|
||||
with mock.patch.object(contact_ui, "settings_menu", side_effect=SystemExit(1)):
|
||||
with mock.patch("contact.ui.dialog.dialog") as dialog:
|
||||
contact_ui.handle_backtick(stdscr)
|
||||
with mock.patch.object(contact_ui, "verify_remote_admin", side_effect=SystemExit(1)):
|
||||
with mock.patch.object(contact_ui, "settings_menu") as settings_menu:
|
||||
with mock.patch("contact.ui.dialog.dialog") as dialog:
|
||||
contact_ui.handle_backtick(stdscr)
|
||||
|
||||
dialog.assert_called_once_with(
|
||||
"Remote admin rejected",
|
||||
"The selected node did not authorize this admin request.",
|
||||
)
|
||||
settings_menu.assert_not_called()
|
||||
|
||||
def test_show_remote_admin_wait_draws_status(self) -> None:
|
||||
stdscr = mock.Mock()
|
||||
|
||||
@@ -53,6 +53,20 @@ 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()
|
||||
node = SimpleNamespace(
|
||||
localConfig=SimpleNamespace(DESCRIPTOR=SimpleNamespace(fields_by_name={"device": device_field})),
|
||||
requestConfig=mock.Mock(),
|
||||
)
|
||||
status_callback = mock.Mock()
|
||||
|
||||
with mock.patch.object(control_ui, "_request_remote_with_timeout") as request:
|
||||
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)
|
||||
|
||||
def test_redraw_main_ui_after_reconnect_refreshes_channels_nodes_and_layout(self) -> None:
|
||||
stdscr = mock.Mock()
|
||||
|
||||
|
||||
Reference in New Issue
Block a user