From 03e27c4672f266974650bd4a20f9241eb458df40 Mon Sep 17 00:00:00 2001 From: Cameron Will Date: Mon, 18 May 2026 09:36:07 -0400 Subject: [PATCH 1/2] feat: Add repeater admin view with login, CLI, and password storage Adds a full repeater administration feature: login/logout to repeaters (admin or guest), send CLI commands with a terminal-style console, and optionally persist repeater passwords in the local database. Co-Authored-By: Claude Opus 4.6 --- src/meshcore_console/core/enums.py | 5 + src/meshcore_console/core/models.py | 11 + src/meshcore_console/core/services.py | 22 +- src/meshcore_console/meshcore/client.py | 85 ++- src/meshcore_console/meshcore/db.py | 8 + src/meshcore_console/meshcore/operations.py | 138 +++- .../meshcore/repeater_store.py | 40 + src/meshcore_console/meshcore/session.py | 27 +- src/meshcore_console/mock/client.py | 81 +- src/meshcore_console/mock/data.py | 52 ++ src/meshcore_console/ui_gtk/layout.py | 5 + src/meshcore_console/ui_gtk/resources/app.css | 65 ++ src/meshcore_console/ui_gtk/views/admin.py | 695 ++++++++++++++++++ .../ui_gtk/windows/main_window.py | 8 +- tests/integration/test_repeater_admin.py | 81 ++ tests/unit/test_db.py | 1 + tests/unit/test_repeater_store.py | 50 ++ 17 files changed, 1355 insertions(+), 19 deletions(-) create mode 100644 src/meshcore_console/meshcore/repeater_store.py create mode 100644 src/meshcore_console/ui_gtk/views/admin.py create mode 100644 tests/integration/test_repeater_admin.py create mode 100644 tests/unit/test_repeater_store.py diff --git a/src/meshcore_console/core/enums.py b/src/meshcore_console/core/enums.py index 0cbcabe..c10f840 100644 --- a/src/meshcore_console/core/enums.py +++ b/src/meshcore_console/core/enums.py @@ -66,6 +66,11 @@ class EventType(StrEnum): # Telemetry TELEMETRY_RECEIVED = "telemetry_received" + # Repeater admin + REPEATER_LOGIN = "repeater_login" + REPEATER_LOGOUT = "repeater_logout" + REPEATER_COMMAND_RESPONSE = "repeater_command_response" + # EventService events (mesh.* naming) MESH_CONTACT_NEW = "mesh.contact.new" MESH_CHANNEL_MESSAGE_NEW = "mesh.channel.message.new" diff --git a/src/meshcore_console/core/models.py b/src/meshcore_console/core/models.py index d841912..e77296b 100644 --- a/src/meshcore_console/core/models.py +++ b/src/meshcore_console/core/models.py @@ -42,6 +42,17 @@ class Message: rssi: int | None = None +@dataclass(slots=True) +class RepeaterLoginState: + peer_name: str + is_admin: bool = False + is_guest: bool = False + keep_alive_interval: int = 0 + acl_permissions: int = 0 + firmware_ver_level: int | None = None + logged_in_at: datetime = field(default_factory=lambda: datetime.now(UTC)) + + @dataclass(slots=True) class Channel: channel_id: str diff --git a/src/meshcore_console/core/services.py b/src/meshcore_console/core/services.py index 0763d30..89dd6c3 100644 --- a/src/meshcore_console/core/services.py +++ b/src/meshcore_console/core/services.py @@ -2,7 +2,7 @@ from typing import TYPE_CHECKING, Callable, Protocol -from meshcore_console.core.models import Channel, DeviceStatus, Message, Peer +from meshcore_console.core.models import Channel, DeviceStatus, Message, Peer, RepeaterLoginState from meshcore_console.core.types import MeshEventDict, SendResultDict if TYPE_CHECKING: @@ -96,3 +96,23 @@ def request_telemetry(self, peer_name: str) -> dict: def get_self_public_key(self) -> str | None: """Return this node's public key as a hex string, or None if unavailable.""" ... + + # -- Repeater admin -------------------------------------------------------- + + def login_to_repeater( + self, peer_name: str, password: str, *, save_password: bool = False + ) -> dict: ... + + def guest_login_to_repeater(self, peer_name: str) -> dict: ... + + def logout_from_repeater(self, peer_name: str) -> None: ... + + def send_repeater_command(self, peer_name: str, command: str) -> dict: ... + + def get_repeater_login_state(self, peer_name: str) -> RepeaterLoginState | None: ... + + def list_logged_in_repeaters(self) -> list[str]: ... + + def get_saved_repeater_password(self, peer_name: str) -> str | None: ... + + def delete_saved_repeater_password(self, peer_name: str) -> None: ... diff --git a/src/meshcore_console/meshcore/client.py b/src/meshcore_console/meshcore/client.py index e9ad23d..500a8c7 100644 --- a/src/meshcore_console/meshcore/client.py +++ b/src/meshcore_console/meshcore/client.py @@ -9,7 +9,7 @@ from uuid import uuid4 from meshcore_console.core.enums import EventType, PayloadType -from meshcore_console.core.models import Channel, DeviceStatus, Message, Peer +from meshcore_console.core.models import Channel, DeviceStatus, Message, Peer, RepeaterLoginState from meshcore_console.core.radio import rssi_to_signal_percent from meshcore_console.core.services import MeshcoreService from meshcore_console.core.types import MeshEventDict, SendResultDict @@ -19,6 +19,7 @@ from meshcore_console.meshcore.logging_setup import install_radio_error_handler from meshcore_console.meshcore.packet_codec import repair_utf8 from meshcore_console.meshcore.packet_store import PacketStore +from meshcore_console.meshcore.repeater_store import RepeaterPasswordStore from meshcore_console.meshcore.session import PyMCCoreSession from meshcore_console.meshcore.settings import MeshcoreSettings from meshcore_console.meshcore.settings_store import SettingsStore @@ -69,6 +70,8 @@ def __init__( self._peer_store = peer_store or PeerStore(self._db) self._channel_store = channel_store or UIChannelStore(self._db) self._gps_provider = gps_provider or create_gps_provider() + self._repeater_password_store = RepeaterPasswordStore(self._db) + self._repeater_sessions: dict[str, RepeaterLoginState] = {} # Load persisted state self._messages: list[Message] = self._message_store.get_all() self._channels: dict[str, Channel] = self._channel_store.get_all() @@ -193,6 +196,7 @@ def disconnect(self) -> None: self._session = self._new_session() self._shutdown_loop() self._connected = False + self._repeater_sessions.clear() self._gps_provider.stop() self._append_event( { @@ -845,6 +849,85 @@ def request_telemetry(self, peer_name: str) -> dict: ) return result # type: ignore[return-value] + # -- Repeater admin -------------------------------------------------------- + + def login_to_repeater( + self, peer_name: str, password: str, *, save_password: bool = False + ) -> dict: + if not self._connected: + self.connect() + result = self._run_async( + self._session.send_repeater_login(peer_name, password), + timeout=15.0, + ) + if result.get("success"): + state = RepeaterLoginState( + peer_name=peer_name, + is_admin=result.get("is_admin", False), + is_guest=password == "", + keep_alive_interval=result.get("keep_alive_interval", 0), + acl_permissions=result.get("acl_permissions", 0), + firmware_ver_level=result.get("firmware_ver_level"), + ) + self._repeater_sessions[peer_name] = state + if save_password: + self._repeater_password_store.save_password(peer_name, password) + self._append_event( + { + "type": EventType.REPEATER_LOGIN, + "data": { + "peer_name": peer_name, + "success": result.get("success"), + "is_admin": result.get("is_admin", False), + }, + } + ) + return result + + def guest_login_to_repeater(self, peer_name: str) -> dict: + return self.login_to_repeater(peer_name, password="") + + def logout_from_repeater(self, peer_name: str) -> None: + if self._connected: + try: + self._run_async(self._session.send_repeater_logout(peer_name), timeout=5.0) + except (RuntimeError, TimeoutError): + pass + self._repeater_sessions.pop(peer_name, None) + self._append_event({"type": EventType.REPEATER_LOGOUT, "data": {"peer_name": peer_name}}) + + def send_repeater_command(self, peer_name: str, command: str) -> dict: + if not self._connected: + self.connect() + result = self._run_async( + self._session.send_repeater_cli(peer_name, command), + timeout=20.0, + ) + self._append_event( + { + "type": EventType.REPEATER_COMMAND_RESPONSE, + "data": { + "peer_name": peer_name, + "command": command, + "success": result.get("success"), + "response_text": result.get("response_text"), + }, + } + ) + return result + + def get_repeater_login_state(self, peer_name: str) -> RepeaterLoginState | None: + return self._repeater_sessions.get(peer_name) + + def list_logged_in_repeaters(self) -> list[str]: + return list(self._repeater_sessions.keys()) + + def get_saved_repeater_password(self, peer_name: str) -> str | None: + return self._repeater_password_store.get_password(peer_name) + + def delete_saved_repeater_password(self, peer_name: str) -> None: + self._repeater_password_store.delete_password(peer_name) + def _get_local_telemetry(self) -> dict: """Provide local telemetry data for inbound requests.""" pos = self._resolve_position(use_settings=self._settings.allow_telemetry) diff --git a/src/meshcore_console/meshcore/db.py b/src/meshcore_console/meshcore/db.py index 404d6b3..629fe61 100644 --- a/src/meshcore_console/meshcore/db.py +++ b/src/meshcore_console/meshcore/db.py @@ -82,6 +82,14 @@ # Backfill: channels with a peer_name are DM channels "UPDATE channels SET kind = 'dm' WHERE peer_name IS NOT NULL", ), + # v5 -> v6: add repeater_passwords table + ( + """CREATE TABLE IF NOT EXISTS repeater_passwords ( + peer_name TEXT PRIMARY KEY, + password TEXT NOT NULL, + created_at TEXT NOT NULL + )""", + ), ] diff --git a/src/meshcore_console/meshcore/operations.py b/src/meshcore_console/meshcore/operations.py index fc96d05..3c3b879 100644 --- a/src/meshcore_console/meshcore/operations.py +++ b/src/meshcore_console/meshcore/operations.py @@ -11,17 +11,23 @@ SendResultDict, ) +TXT_TYPE_CLI_DATA = 1 -async def send_text(*, node: MeshNodeProtocol, peer_name: str, message: str) -> dict: - """Send a direct text message to a peer via PacketBuilder.""" - from pymc_core.protocol.packet_builder import PacketBuilder +def _resolve_contact(node: MeshNodeProtocol, peer_name: str) -> object: contacts = node.contacts - contact = None if contacts is not None: contact = contacts.get_by_name(peer_name) - if contact is None: - raise RuntimeError(f"Contact '{peer_name}' not found") + if contact is not None: + return contact + raise RuntimeError(f"Contact '{peer_name}' not found") + + +async def send_text(*, node: MeshNodeProtocol, peer_name: str, message: str) -> dict: + """Send a direct text message to a peer via PacketBuilder.""" + from pymc_core.protocol.packet_builder import PacketBuilder + + contact = _resolve_contact(node, peer_name) pkt, ack_crc = PacketBuilder.create_text_message( contact=contact, @@ -61,12 +67,7 @@ async def request_telemetry( """Request telemetry data from a remote peer via PacketBuilder.""" from pymc_core.protocol.packet_builder import PacketBuilder - contacts = node.contacts - contact = None - if contacts is not None: - contact = contacts.get_by_name(contact_name) - if contact is None: - raise RuntimeError(f"Contact '{contact_name}' not found") + contact = _resolve_contact(node, contact_name) pkt, _ts = PacketBuilder.create_telem_request( contact=contact, @@ -125,6 +126,119 @@ def _on_response(success: bool, text: str, parsed: dict) -> None: } +async def send_login( + *, + node: MeshNodeProtocol, + peer_name: str, + password: str, + timeout: float = 10.0, +) -> dict: + """Send a login request to a repeater and wait for the response.""" + from pymc_core.protocol.packet_builder import PacketBuilder + + contact = _resolve_contact(node, peer_name) + + login_handler = node.dispatcher.login_response_handler + dest_hash = bytes.fromhex(contact.public_key)[0] + login_handler.store_login_password(dest_hash, password) + + login_event = asyncio.Event() + login_result: dict = {"success": False, "data": {}} + + def _on_login(success: bool, data: dict) -> None: + login_result["success"] = success + login_result["data"] = data + login_event.set() + + login_handler.set_login_callback(_on_login) + try: + pkt = PacketBuilder.create_login_packet( + contact=contact, + local_identity=node.identity, + password=password, + ) + await node.dispatcher.send_packet(pkt, wait_for_ack=False) + await asyncio.wait_for(login_event.wait(), timeout=timeout) + except asyncio.TimeoutError: + return {"success": False, "reason": "Login response timeout"} + finally: + login_handler.set_login_callback(None) + login_handler.clear_login_password(dest_hash) + + data = login_result["data"] + return { + "success": login_result["success"], + "repeater": peer_name, + "is_admin": data.get("is_admin", False), + "keep_alive_interval": data.get("keep_alive_interval", 0), + "acl_permissions": data.get("reserved", data.get("permissions", 0)), + "firmware_ver_level": data.get("firmware_ver_level"), + "reason": "Login successful" if login_result["success"] else "Login failed", + } + + +async def send_logout(*, node: MeshNodeProtocol, peer_name: str) -> dict: + """Send a logout/disconnect to a repeater.""" + from pymc_core.protocol.packet_builder import PacketBuilder + + contact = _resolve_contact(node, peer_name) + + pkt, _crc = PacketBuilder.create_logout_packet( + contact=contact, + local_identity=node.identity, + ) + await node.dispatcher.send_packet(pkt, wait_for_ack=False) + return {"success": True, "repeater": peer_name} + + +async def send_repeater_command( + *, + node: MeshNodeProtocol, + peer_name: str, + command: str, + timeout: float = 15.0, +) -> dict: + """Send a CLI command to a repeater and wait for the response.""" + from pymc_core.protocol.packet_builder import PacketBuilder + + contact = _resolve_contact(node, peer_name) + + text_handler = node.dispatcher.text_message_handler + response_event = asyncio.Event() + response_data: dict = {"text": None} + + def _on_response(message_text: str, sender_contact: object) -> None: + response_data["text"] = message_text + response_event.set() + + text_handler.set_command_response_callback(_on_response) + try: + pkt, _crc = PacketBuilder.create_text_message( + contact=contact, + local_identity=node.identity, + message=command, + txt_type=TXT_TYPE_CLI_DATA, + ) + await node.dispatcher.send_packet(pkt, wait_for_ack=False) + await asyncio.wait_for(response_event.wait(), timeout=timeout) + return { + "success": True, + "repeater": peer_name, + "command": command, + "response_text": response_data["text"], + } + except asyncio.TimeoutError: + return { + "success": False, + "repeater": peer_name, + "command": command, + "response_text": None, + "reason": "Command response timeout", + } + finally: + text_handler.set_command_response_callback(None) + + async def send_advert( *, node: MeshNodeProtocol, diff --git a/src/meshcore_console/meshcore/repeater_store.py b/src/meshcore_console/meshcore/repeater_store.py new file mode 100644 index 0000000..c271641 --- /dev/null +++ b/src/meshcore_console/meshcore/repeater_store.py @@ -0,0 +1,40 @@ +"""Persistent storage for saved repeater admin passwords.""" + +from __future__ import annotations + +import base64 +import sqlite3 +from datetime import UTC, datetime + + +class RepeaterPasswordStore: + def __init__(self, conn: sqlite3.Connection) -> None: + self._conn = conn + + def save_password(self, peer_name: str, password: str) -> None: + encoded = base64.b64encode(password.encode("utf-8")).decode("ascii") + self._conn.execute( + "INSERT OR REPLACE INTO repeater_passwords (peer_name, password, created_at) " + "VALUES (?, ?, ?)", + (peer_name, encoded, datetime.now(UTC).isoformat()), + ) + self._conn.commit() + + def get_password(self, peer_name: str) -> str | None: + row = self._conn.execute( + "SELECT password FROM repeater_passwords WHERE peer_name = ?", + (peer_name,), + ).fetchone() + if row is None: + return None + return base64.b64decode(row[0]).decode("utf-8") + + def delete_password(self, peer_name: str) -> None: + self._conn.execute("DELETE FROM repeater_passwords WHERE peer_name = ?", (peer_name,)) + self._conn.commit() + + def list_saved(self) -> list[str]: + rows = self._conn.execute( + "SELECT peer_name FROM repeater_passwords ORDER BY peer_name" + ).fetchall() + return [r[0] for r in rows] diff --git a/src/meshcore_console/meshcore/session.py b/src/meshcore_console/meshcore/session.py index 2445e10..467d4a3 100644 --- a/src/meshcore_console/meshcore/session.py +++ b/src/meshcore_console/meshcore/session.py @@ -43,7 +43,15 @@ from .contact_book import ContactBook from .db import open_db from .event_bridge import attach_dispatcher_callbacks, attach_event_service_subscriber -from .operations import request_telemetry, send_advert, send_group_text, send_text +from .operations import ( + request_telemetry, + send_advert, + send_group_text, + send_login, + send_logout, + send_repeater_command, + send_text, +) from .runtime import create_mesh_node, create_radio, import_pymc_core @@ -645,6 +653,23 @@ async def send_telemetry_request( timeout=timeout, ) + async def send_repeater_login(self, peer_name: str, password: str) -> dict: + if self._node is None: + raise RuntimeError("Session is not started.") + return await send_login(node=self._node, peer_name=peer_name, password=password) + + async def send_repeater_logout(self, peer_name: str) -> dict: + if self._node is None: + raise RuntimeError("Session is not started.") + return await send_logout(node=self._node, peer_name=peer_name) + + async def send_repeater_cli(self, peer_name: str, command: str, timeout: float = 15.0) -> dict: + if self._node is None: + raise RuntimeError("Session is not started.") + return await send_repeater_command( + node=self._node, peer_name=peer_name, command=command, timeout=timeout + ) + async def listen_events(self) -> AsyncIterator[MeshEventDict]: if self._node is None: raise RuntimeError("Session is not started.") diff --git a/src/meshcore_console/mock/client.py b/src/meshcore_console/mock/client.py index 48b1d1a..9ab8008 100644 --- a/src/meshcore_console/mock/client.py +++ b/src/meshcore_console/mock/client.py @@ -6,13 +6,14 @@ from typing import Callable from uuid import uuid4 -from meshcore_console.core.enums import PayloadType -from meshcore_console.core.models import Channel, DeviceStatus, Message, Peer +from meshcore_console.core.enums import EventType, PayloadType +from meshcore_console.core.models import Channel, DeviceStatus, Message, Peer, RepeaterLoginState from meshcore_console.core.services import MeshcoreService from meshcore_console.meshcore.config import runtime_config_from_settings from meshcore_console.meshcore.settings import MeshcoreSettings from .data import ( + MOCK_CLI_RESPONSES, create_mock_boot_events, create_mock_channels, create_mock_messages, @@ -35,6 +36,9 @@ def __init__(self, node_name: str = "uconsole-node") -> None: self._event_buffer: list[dict] = [] self._event_history: list[dict] = [] + self._repeater_sessions: dict[str, RepeaterLoginState] = {} + self._saved_passwords: dict[str, str] = {} + # Initialize mock state self._channels = create_mock_channels() self._peers = create_mock_peers() @@ -280,6 +284,79 @@ def get_self_public_key(self) -> str | None: """Return a mock public key for testing.""" return "6b547fd13630e0f7a6b167df23b9876543210abcdef0123456789abcdef0a619" + # -- Repeater admin (mock) ------------------------------------------------- + + def login_to_repeater( + self, peer_name: str, password: str, *, save_password: bool = False + ) -> dict: + is_admin = password != "" + state = RepeaterLoginState( + peer_name=peer_name, + is_admin=is_admin, + is_guest=not is_admin, + keep_alive_interval=300, + acl_permissions=0x02 if is_admin else 0x01, + firmware_ver_level=12, + ) + self._repeater_sessions[peer_name] = state + if save_password and password: + self._saved_passwords[peer_name] = password + self._append_event( + { + "type": EventType.REPEATER_LOGIN, + "data": {"peer_name": peer_name, "success": True, "is_admin": is_admin}, + } + ) + return { + "success": True, + "repeater": peer_name, + "is_admin": is_admin, + "keep_alive_interval": 300, + "acl_permissions": 0x02 if is_admin else 0x01, + "firmware_ver_level": 12, + "reason": "Login successful", + } + + def guest_login_to_repeater(self, peer_name: str) -> dict: + return self.login_to_repeater(peer_name, password="") + + def logout_from_repeater(self, peer_name: str) -> None: + self._repeater_sessions.pop(peer_name, None) + self._append_event({"type": EventType.REPEATER_LOGOUT, "data": {"peer_name": peer_name}}) + + def send_repeater_command(self, peer_name: str, command: str) -> dict: + cmd_key = command.strip().split()[0].lower() if command.strip() else "" + response_text = MOCK_CLI_RESPONSES.get(cmd_key, f"OK: {command}") + self._append_event( + { + "type": EventType.REPEATER_COMMAND_RESPONSE, + "data": { + "peer_name": peer_name, + "command": command, + "success": True, + "response_text": response_text, + }, + } + ) + return { + "success": True, + "repeater": peer_name, + "command": command, + "response_text": response_text, + } + + def get_repeater_login_state(self, peer_name: str) -> RepeaterLoginState | None: + return self._repeater_sessions.get(peer_name) + + def list_logged_in_repeaters(self) -> list[str]: + return list(self._repeater_sessions.keys()) + + def get_saved_repeater_password(self, peer_name: str) -> str | None: + return self._saved_passwords.get(peer_name) + + def delete_saved_repeater_password(self, peer_name: str) -> None: + self._saved_passwords.pop(peer_name, None) + def _process_event_for_messages(self, event: dict) -> None: """Convert incoming packet events into messages and channels.""" data = event.get("data") diff --git a/src/meshcore_console/mock/data.py b/src/meshcore_console/mock/data.py index 1531932..a3da2bc 100644 --- a/src/meshcore_console/mock/data.py +++ b/src/meshcore_console/mock/data.py @@ -37,6 +37,48 @@ ] +MOCK_CLI_RESPONSES: dict[str, str] = { + "status": ( + "Uptime: 3d 14h 22m\n" + "Connected clients: 7\n" + "Packets forwarded: 12,847\n" + "Frequency: 910.525 MHz\n" + "TX Power: 20 dBm\n" + "SF: 10 BW: 250 kHz CR: 4/5" + ), + "ver": "MeshCore Repeater v1.12.3\nBoard: Heltec V3\nBuild: 2025-04-10", + "neighbors": ( + "Neighbors (3):\n" + " Alice -67 dBm SNR 8.5 direct\n" + " Bob -82 dBm SNR 3.2 2 hops\n" + " Charlie -91 dBm SNR -1.0 3 hops" + ), + "help": ( + "Available commands:\n" + " status Show repeater status\n" + " ver Firmware version\n" + " neighbors Connected peers\n" + " reboot Reboot repeater\n" + " advert Send advertisement\n" + " clock Show device time\n" + " set Configure parameters\n" + " get Query parameters\n" + " log View packet log" + ), + "clock": "Device time: 2025-04-15 14:32:07 UTC\nUptime: 3d 14h 22m", + "advert": "Advertisement sent (flood)", + "reboot": "Rebooting in 3 seconds...", + "log": ( + "Recent packets (last 5):\n" + " 14:31:52 ADVERT Alice -67 dBm\n" + " 14:31:48 GRP_TXT #test -73 dBm\n" + " 14:31:45 TXT_MSG Bob -82 dBm\n" + " 14:31:40 ACK Charlie -91 dBm\n" + " 14:31:38 ADVERT Diana -78 dBm" + ), +} + + def create_mock_channels() -> dict[str, Channel]: """Create mock channels for testing.""" return { @@ -68,6 +110,16 @@ def create_mock_peers() -> dict[str, Peer]: latitude=37.8044, longitude=-122.2712, location_updated=now, + is_favorite=True, + ), + "\U0001f4e1 Hilltop C": Peer( + peer_id="relay-003", + display_name="\U0001f4e1 Hilltop C", + signal_quality=44, + is_repeater=True, + latitude=37.7575, + longitude=-122.4376, + location_updated=now, ), "\U0001f43b Alice": Peer( peer_id="peer-alice", diff --git a/src/meshcore_console/ui_gtk/layout.py b/src/meshcore_console/ui_gtk/layout.py index dd2331a..863265e 100644 --- a/src/meshcore_console/ui_gtk/layout.py +++ b/src/meshcore_console/ui_gtk/layout.py @@ -49,6 +49,11 @@ def map_details_width(self) -> int: def status_card_width(self) -> int: return int(self.content_width * 0.15) + # -- Admin view ----------------------------------------------------------- + @property + def admin_list_width(self) -> int: + return int(self.content_width * 0.22) + @property def detail_block_wrap_chars(self) -> int: return max(16, int(self.content_width * 0.019)) diff --git a/src/meshcore_console/ui_gtk/resources/app.css b/src/meshcore_console/ui_gtk/resources/app.css index 2b3a7d3..f5f6a42 100644 --- a/src/meshcore_console/ui_gtk/resources/app.css +++ b/src/meshcore_console/ui_gtk/resources/app.css @@ -730,3 +730,68 @@ row:focus-visible { font-size: 16px; color: @mc_text_muted; } + +/* -- Admin / Repeater CLI ------------------------------------------------ */ + +.cli-console { + background: shade(@mc_bg, 0.9); + border-radius: 8px; + padding: 8px; +} + +.cli-console label { + font-family: monospace; + font-size: 13px; +} + +.cli-command { + color: @mc_accent; +} + +.cli-response { + color: @mc_text; +} + +.cli-error { + color: @mc_danger; +} + +.cli-entry { + font-family: monospace; + font-size: 13px; +} + +.login-status-admin { + color: @mc_accent; + font-weight: 700; +} + +.login-status-guest { + color: @mc_warn; + font-weight: 700; +} + +.login-status-offline { + color: @mc_text_muted; +} + +.admin-repeater-row { + padding: 8px 12px; +} + +.admin-login-dot { + color: @mc_accent; + font-size: 10px; +} + +.admin-login-dot-off { + color: @mc_text_muted; + font-size: 10px; +} + +.quick-cmd-btn { + font-family: monospace; + font-size: 12px; + padding: 2px 8px; + min-height: 24px; +} diff --git a/src/meshcore_console/ui_gtk/views/admin.py b/src/meshcore_console/ui_gtk/views/admin.py new file mode 100644 index 0000000..6a95e18 --- /dev/null +++ b/src/meshcore_console/ui_gtk/views/admin.py @@ -0,0 +1,695 @@ +from __future__ import annotations + +import logging +import threading +from typing import TYPE_CHECKING + +import gi + +gi.require_version("Gtk", "4.0") + +from gi.repository import Gdk, GLib, Gtk, Pango + +from meshcore_console.core.models import Peer +from meshcore_console.core.radio import format_rssi, format_snr +from meshcore_console.core.services import MeshcoreService +from meshcore_console.core.time import to_local +from meshcore_console.ui_gtk.helpers import clear_children +from meshcore_console.ui_gtk.layout import Layout +from meshcore_console.ui_gtk.state import UiEventStore +from meshcore_console.ui_gtk.widgets import DetailRow + +if TYPE_CHECKING: + from meshcore_console.core.models import RepeaterLoginState + +logger = logging.getLogger(__name__) + +_MAX_HISTORY = 50 + + +class AdminView(Gtk.Box): + def __init__(self, service: MeshcoreService, event_store: UiEventStore, layout: Layout) -> None: + super().__init__(orientation=Gtk.Orientation.HORIZONTAL, spacing=8) + self._service = service + self._event_store = event_store + self._event_cursor = 0 + self._selected_repeater: Peer | None = None + self._command_history: list[str] = [] + self._history_index = -1 + self._busy = False + self._refreshing_list = False + + # -- Left: repeater list ----------------------------------------------- + list_col = Gtk.Box(orientation=Gtk.Orientation.VERTICAL, spacing=0) + list_col.add_css_class("panel-card") + list_col.set_size_request(layout.admin_list_width, -1) + + header = Gtk.Label(label="Repeaters") + header.add_css_class("panel-title") + header.set_halign(Gtk.Align.START) + header.set_margin_start(12) + header.set_margin_top(10) + header.set_margin_bottom(8) + list_col.append(header) + + scroll = Gtk.ScrolledWindow.new() + scroll.set_vexpand(True) + scroll.set_policy(Gtk.PolicyType.NEVER, Gtk.PolicyType.AUTOMATIC) + + self._repeater_list = Gtk.ListBox.new() + self._repeater_list.set_selection_mode(Gtk.SelectionMode.SINGLE) + self._repeater_list.connect("row-selected", self._on_repeater_selected) + scroll.set_child(self._repeater_list) + list_col.append(scroll) + self.append(list_col) + + # -- Right: panel with notebook ---------------------------------------- + panel = Gtk.Box(orientation=Gtk.Orientation.VERTICAL, spacing=0) + panel.add_css_class("panel-card") + panel.set_hexpand(True) + + self._panel_stack = Gtk.Stack.new() + self._panel_stack.set_transition_type(Gtk.StackTransitionType.CROSSFADE) + self._panel_stack.set_transition_duration(150) + + # Empty state + empty = Gtk.Box(orientation=Gtk.Orientation.VERTICAL, spacing=8) + empty.set_valign(Gtk.Align.CENTER) + empty.set_halign(Gtk.Align.CENTER) + empty_label = Gtk.Label(label="Select a repeater") + empty_label.add_css_class("panel-muted") + empty.append(empty_label) + self._panel_stack.add_named(empty, "empty") + + # Admin panel with title + notebook + admin_outer = Gtk.Box(orientation=Gtk.Orientation.VERTICAL, spacing=0) + + self._admin_title = Gtk.Label(label="") + self._admin_title.add_css_class("panel-title") + self._admin_title.set_halign(Gtk.Align.START) + self._admin_title.set_margin_start(12) + self._admin_title.set_margin_top(10) + self._admin_title.set_margin_bottom(4) + admin_outer.append(self._admin_title) + + # Login bar (always visible at top of admin panel) + self._login_box = self._build_login_bar() + admin_outer.append(self._login_box) + + sep = Gtk.Separator.new(Gtk.Orientation.HORIZONTAL) + sep.set_margin_start(12) + sep.set_margin_end(12) + admin_outer.append(sep) + + # Notebook with Status and Console tabs + self._notebook = Gtk.Notebook.new() + self._notebook.set_vexpand(True) + self._notebook.set_margin_start(4) + self._notebook.set_margin_end(4) + self._notebook.set_margin_bottom(4) + + # Status tab + self._status_scroll = Gtk.ScrolledWindow.new() + self._status_scroll.set_vexpand(True) + self._status_scroll.set_policy(Gtk.PolicyType.NEVER, Gtk.PolicyType.AUTOMATIC) + self._status_content = Gtk.Box(orientation=Gtk.Orientation.VERTICAL, spacing=8) + self._status_content.set_margin_start(12) + self._status_content.set_margin_end(12) + self._status_content.set_margin_top(8) + self._status_content.set_margin_bottom(8) + self._status_scroll.set_child(self._status_content) + self._notebook.append_page(self._status_scroll, Gtk.Label(label="Status")) + + # Console tab + console_page = self._build_console_tab() + self._notebook.append_page(console_page, Gtk.Label(label="Console")) + + admin_outer.append(self._notebook) + self._panel_stack.add_named(admin_outer, "admin") + + self._panel_stack.set_visible_child_name("empty") + panel.append(self._panel_stack) + self.append(panel) + + # Wire events + event_store.connect("events-available", self._on_events) + GLib.idle_add(self._refresh_list) + + # -- Build login bar ------------------------------------------------------- + + def _build_login_bar(self) -> Gtk.Box: + box = Gtk.Box(orientation=Gtk.Orientation.VERTICAL, spacing=6) + box.set_margin_start(12) + box.set_margin_end(12) + box.set_margin_bottom(8) + + # Status line + self._status_label = Gtk.Label(label="Not logged in") + self._status_label.add_css_class("login-status-offline") + self._status_label.set_halign(Gtk.Align.START) + box.append(self._status_label) + + # Password row + pw_row = Gtk.Box(orientation=Gtk.Orientation.HORIZONTAL, spacing=6) + self._password_entry = Gtk.PasswordEntry.new() + self._password_entry.set_property("placeholder-text", "Password") + self._password_entry.set_show_peek_icon(True) + self._password_entry.set_hexpand(True) + self._password_entry.connect("activate", self._on_login_activate) + pw_row.append(self._password_entry) + + self._login_btn = Gtk.Button.new_with_label("Login") + self._login_btn.connect("clicked", self._on_login_clicked) + pw_row.append(self._login_btn) + + self._guest_btn = Gtk.Button.new_with_label("Guest") + self._guest_btn.connect("clicked", self._on_guest_clicked) + pw_row.append(self._guest_btn) + box.append(pw_row) + + # Save password / forget / logout row + save_row = Gtk.Box(orientation=Gtk.Orientation.HORIZONTAL, spacing=8) + self._save_check = Gtk.CheckButton.new_with_label("Save password") + save_row.append(self._save_check) + + self._forget_btn = Gtk.Button.new_with_label("Forget saved") + self._forget_btn.connect("clicked", self._on_forget_clicked) + self._forget_btn.set_visible(False) + save_row.append(self._forget_btn) + + self._logout_btn = Gtk.Button.new_with_label("Logout") + self._logout_btn.connect("clicked", self._on_logout_clicked) + self._logout_btn.set_sensitive(False) + save_row.append(self._logout_btn) + box.append(save_row) + + return box + + # -- Build console tab ----------------------------------------------------- + + def _build_console_tab(self) -> Gtk.Box: + box = Gtk.Box(orientation=Gtk.Orientation.VERTICAL, spacing=0) + + # Quick command buttons + quick_row = Gtk.Box(orientation=Gtk.Orientation.HORIZONTAL, spacing=4) + quick_row.set_margin_start(8) + quick_row.set_margin_end(8) + quick_row.set_margin_top(8) + quick_row.set_margin_bottom(4) + for cmd in ("status", "ver", "neighbors", "help", "log"): + btn = Gtk.Button.new_with_label(cmd) + btn.add_css_class("quick-cmd-btn") + btn.connect("clicked", self._on_quick_cmd, cmd) + quick_row.append(btn) + self._quick_row = quick_row + box.append(quick_row) + + # Console output + console_scroll = Gtk.ScrolledWindow.new() + console_scroll.set_vexpand(True) + console_scroll.set_policy(Gtk.PolicyType.NEVER, Gtk.PolicyType.AUTOMATIC) + console_scroll.set_margin_start(8) + console_scroll.set_margin_end(8) + + self._console_box = Gtk.Box(orientation=Gtk.Orientation.VERTICAL, spacing=2) + self._console_box.add_css_class("cli-console") + console_scroll.set_child(self._console_box) + self._console_scroll = console_scroll + box.append(console_scroll) + + # Command entry + entry_row = Gtk.Box(orientation=Gtk.Orientation.HORIZONTAL, spacing=6) + entry_row.set_margin_start(8) + entry_row.set_margin_end(8) + entry_row.set_margin_top(4) + entry_row.set_margin_bottom(8) + + self._cmd_entry = Gtk.Entry.new() + self._cmd_entry.add_css_class("cli-entry") + self._cmd_entry.set_placeholder_text("Type a command...") + self._cmd_entry.set_hexpand(True) + self._cmd_entry.connect("activate", self._on_cmd_activate) + + key_ctrl = Gtk.EventControllerKey.new() + key_ctrl.connect("key-pressed", self._on_cmd_key) + self._cmd_entry.add_controller(key_ctrl) + + entry_row.append(self._cmd_entry) + + self._send_btn = Gtk.Button.new_with_label("Send") + self._send_btn.connect("clicked", self._on_send_clicked) + entry_row.append(self._send_btn) + box.append(entry_row) + + self._set_cli_sensitive(False) + return box + + # -- Build status tab content ---------------------------------------------- + + def _populate_status(self, peer: Peer) -> None: + clear_children(self._status_content) + + # Signal section + self._add_section_header("Signal") + + if peer.rssi is not None: + self._status_content.append(DetailRow("RSSI:", format_rssi(peer.rssi))) + if peer.snr is not None: + self._status_content.append(DetailRow("SNR:", format_snr(peer.snr))) + if peer.signal_quality is not None: + self._status_content.append(DetailRow("Quality:", f"{peer.signal_quality}%")) + + if peer.rssi is None and peer.snr is None and peer.signal_quality is None: + no_signal = Gtk.Label(label="No signal data available") + no_signal.add_css_class("panel-muted") + no_signal.set_halign(Gtk.Align.START) + self._status_content.append(no_signal) + + # Location section + self._add_section_header("Location") + + if peer.latitude is not None and peer.longitude is not None: + lat_dir = "N" if peer.latitude >= 0 else "S" + lon_dir = "E" if peer.longitude >= 0 else "W" + coords = f"{abs(peer.latitude):.5f}° {lat_dir}, {abs(peer.longitude):.5f}° {lon_dir}" + self._status_content.append(DetailRow("Coordinates:", coords)) + + if peer.location_updated: + loc_time = to_local(peer.location_updated).strftime("%b %d at %H:%M") + self._status_content.append(DetailRow("Updated:", loc_time)) + else: + no_loc = Gtk.Label(label="No location reported") + no_loc.add_css_class("panel-muted") + no_loc.set_halign(Gtk.Align.START) + self._status_content.append(no_loc) + + # Activity section + self._add_section_header("Activity") + + if peer.last_advert_time: + time_str = to_local(peer.last_advert_time).strftime("%b %d, %Y at %I:%M %p") + self._status_content.append(DetailRow("Last Seen:", time_str)) + else: + self._status_content.append(DetailRow("Last Seen:", "Unknown")) + + # Login state section + state = self._service.get_repeater_login_state(peer.display_name) + self._add_section_header("Login") + + if state is not None: + role = "Admin" if state.is_admin else "Guest" + self._status_content.append(DetailRow("Role:", role)) + if state.firmware_ver_level is not None: + self._status_content.append(DetailRow("Firmware:", f"v{state.firmware_ver_level}")) + if state.keep_alive_interval: + self._status_content.append( + DetailRow("Keep-alive:", f"{state.keep_alive_interval}s") + ) + else: + not_logged = Gtk.Label(label="Not logged in") + not_logged.add_css_class("panel-muted") + not_logged.set_halign(Gtk.Align.START) + self._status_content.append(not_logged) + + # Public key section + if peer.public_key: + self._add_section_header("Public Key") + chunks = [peer.public_key[i : i + 4] for i in range(0, len(peer.public_key), 4)] + key_label = Gtk.Label(label=" ".join(chunks)) + key_label.add_css_class("analyzer-raw") + key_label.set_halign(Gtk.Align.START) + key_label.set_wrap(True) + key_label.set_wrap_mode(Pango.WrapMode.CHAR) + key_label.set_selectable(True) + self._status_content.append(key_label) + + def _add_section_header(self, title: str) -> None: + header = Gtk.Label(label=title) + header.add_css_class("message-detail-header") + header.set_halign(Gtk.Align.START) + header.set_margin_top(12) + self._status_content.append(header) + + # -- List management ------------------------------------------------------- + + def _refresh_list(self) -> bool: + peers = self._service.list_peers() + repeaters = sorted( + [p for p in peers if p.is_repeater], + key=lambda p: (not p.is_favorite, p.display_name), + ) + logged_in = set(self._service.list_logged_in_repeaters()) + + # Suppress selection handler during rebuild + self._refreshing_list = True + + child = self._repeater_list.get_first_child() + while child is not None: + nxt = child.get_next_sibling() + self._repeater_list.remove(child) + child = nxt + + for rp in repeaters: + row = self._make_repeater_row(rp, rp.display_name in logged_in) + self._repeater_list.append(row) + + # Re-select if still valid, and update the peer object + if self._selected_repeater: + for i, rp in enumerate(repeaters): + if rp.display_name == self._selected_repeater.display_name: + self._selected_repeater = rp + row_widget = self._repeater_list.get_row_at_index(i) + if row_widget: + self._repeater_list.select_row(row_widget) + break + + self._refreshing_list = False + return False + + def _make_repeater_row(self, peer: Peer, logged_in: bool) -> Gtk.ListBoxRow: + row = Gtk.ListBoxRow.new() + row._peer = peer # type: ignore[attr-defined] + hbox = Gtk.Box(orientation=Gtk.Orientation.HORIZONTAL, spacing=8) + hbox.add_css_class("admin-repeater-row") + + dot = Gtk.Label(label="●") + dot.add_css_class("admin-login-dot" if logged_in else "admin-login-dot-off") + hbox.append(dot) + + name = Gtk.Label(label=peer.display_name) + name.set_halign(Gtk.Align.START) + name.set_hexpand(True) + name.set_ellipsize(Pango.EllipsizeMode.END) + name.set_max_width_chars(20) + hbox.append(name) + + if peer.is_favorite: + star = Gtk.Image.new_from_icon_name("starred-symbolic") + star.add_css_class("panel-muted") + hbox.append(star) + + if peer.rssi is not None: + rssi = Gtk.Label(label=f"{peer.rssi} dBm") + rssi.add_css_class("panel-muted") + hbox.append(rssi) + + row.set_child(hbox) + return row + + # -- Selection ------------------------------------------------------------- + + def _on_repeater_selected(self, _listbox: Gtk.ListBox, row: Gtk.ListBoxRow | None) -> None: + if self._refreshing_list: + return + if row is None: + self._selected_repeater = None + self._panel_stack.set_visible_child_name("empty") + return + peer: Peer = row._peer # type: ignore[attr-defined] + self._selected_repeater = peer + self._show_admin_panel(peer) + + def _show_admin_panel(self, peer: Peer) -> None: + self._admin_title.set_text(peer.display_name) + self._panel_stack.set_visible_child_name("admin") + + # Update login state UI + state = self._service.get_repeater_login_state(peer.display_name) + self._update_login_ui(state) + + # Pre-fill saved password + saved = self._service.get_saved_repeater_password(peer.display_name) + if saved: + self._password_entry.set_text(saved) + self._forget_btn.set_visible(True) + self._save_check.set_active(True) + else: + self._password_entry.set_text("") + self._forget_btn.set_visible(False) + self._save_check.set_active(False) + + # Populate status tab + self._populate_status(peer) + + # Clear console for new repeater + clear_children(self._console_box) + + def _update_login_ui(self, state: RepeaterLoginState | None) -> None: + if state is None: + self._status_label.set_text("Not logged in") + self._status_label.remove_css_class("login-status-admin") + self._status_label.remove_css_class("login-status-guest") + self._status_label.add_css_class("login-status-offline") + self._login_btn.set_sensitive(True) + self._guest_btn.set_sensitive(True) + self._password_entry.set_sensitive(True) + self._logout_btn.set_sensitive(False) + self._set_cli_sensitive(False) + else: + if state.is_admin: + self._status_label.set_text("Logged in (admin)") + self._status_label.remove_css_class("login-status-offline") + self._status_label.remove_css_class("login-status-guest") + self._status_label.add_css_class("login-status-admin") + else: + self._status_label.set_text("Logged in (guest)") + self._status_label.remove_css_class("login-status-offline") + self._status_label.remove_css_class("login-status-admin") + self._status_label.add_css_class("login-status-guest") + if state.firmware_ver_level is not None: + self._status_label.set_text( + self._status_label.get_text() + f" FW: v{state.firmware_ver_level}" + ) + self._login_btn.set_sensitive(False) + self._guest_btn.set_sensitive(False) + self._password_entry.set_sensitive(False) + self._logout_btn.set_sensitive(True) + self._set_cli_sensitive(True) + + def _set_cli_sensitive(self, sensitive: bool) -> None: + self._cmd_entry.set_sensitive(sensitive) + self._send_btn.set_sensitive(sensitive) + child = self._quick_row.get_first_child() + while child is not None: + child.set_sensitive(sensitive) + child = child.get_next_sibling() + + # -- Login actions --------------------------------------------------------- + + def _on_login_activate(self, _entry: Gtk.PasswordEntry) -> None: + self._do_login(self._password_entry.get_text()) + + def _on_login_clicked(self, _btn: Gtk.Button) -> None: + self._do_login(self._password_entry.get_text()) + + def _on_guest_clicked(self, _btn: Gtk.Button) -> None: + self._do_login("") + + def _do_login(self, password: str) -> None: + if self._selected_repeater is None or self._busy: + return + peer_name = self._selected_repeater.display_name + save = self._save_check.get_active() + self._busy = True + self._login_btn.set_sensitive(False) + self._guest_btn.set_sensitive(False) + self._status_label.set_text("Logging in...") + + def work() -> None: + error: str | None = None + result: dict = {} + try: + result = self._service.login_to_repeater(peer_name, password, save_password=save) + except Exception as exc: + error = str(exc) or type(exc).__name__ + + def done() -> bool: + self._busy = False + if error: + self._append_console_error(f"Login failed: {error}") + self._login_btn.set_sensitive(True) + self._guest_btn.set_sensitive(True) + self._update_login_ui(None) + elif not result.get("success"): + reason = result.get("reason", "Login failed") + self._append_console_error(reason) + self._login_btn.set_sensitive(True) + self._guest_btn.set_sensitive(True) + self._update_login_ui(None) + else: + state = self._service.get_repeater_login_state(peer_name) + self._update_login_ui(state) + self._append_console_response("Login successful") + # Refresh status tab to show login info + if self._selected_repeater: + self._populate_status(self._selected_repeater) + self._refresh_list() + return False + + GLib.idle_add(done) + + threading.Thread(target=work, daemon=True).start() + + def _on_logout_clicked(self, _btn: Gtk.Button) -> None: + if self._selected_repeater is None: + return + peer_name = self._selected_repeater.display_name + self._service.logout_from_repeater(peer_name) + self._update_login_ui(None) + self._append_console_response("Logged out") + if self._selected_repeater: + self._populate_status(self._selected_repeater) + self._refresh_list() + + def _on_forget_clicked(self, _btn: Gtk.Button) -> None: + if self._selected_repeater is None: + return + self._service.delete_saved_repeater_password(self._selected_repeater.display_name) + self._password_entry.set_text("") + self._forget_btn.set_visible(False) + self._save_check.set_active(False) + + # -- CLI actions ----------------------------------------------------------- + + def _on_quick_cmd(self, _btn: Gtk.Button, command: str) -> None: + self._send_command(command) + + def _on_cmd_activate(self, _entry: Gtk.Entry) -> None: + text = self._cmd_entry.get_text().strip() + if text: + self._send_command(text) + + def _on_send_clicked(self, _btn: Gtk.Button) -> None: + text = self._cmd_entry.get_text().strip() + if text: + self._send_command(text) + + def _on_cmd_key( + self, + _ctrl: Gtk.EventControllerKey, + keyval: int, + _keycode: int, + _state: Gdk.ModifierType, + ) -> bool: + if keyval == Gdk.KEY_Up: + if self._command_history and self._history_index < len(self._command_history) - 1: + self._history_index += 1 + self._cmd_entry.set_text(self._command_history[-(self._history_index + 1)]) + self._cmd_entry.set_position(-1) + return True + if keyval == Gdk.KEY_Down: + if self._history_index > 0: + self._history_index -= 1 + self._cmd_entry.set_text(self._command_history[-(self._history_index + 1)]) + self._cmd_entry.set_position(-1) + elif self._history_index == 0: + self._history_index = -1 + self._cmd_entry.set_text("") + return True + return False + + def _send_command(self, command: str) -> None: + if self._selected_repeater is None or self._busy: + return + peer_name = self._selected_repeater.display_name + + # History + self._command_history.append(command) + if len(self._command_history) > _MAX_HISTORY: + self._command_history = self._command_history[-_MAX_HISTORY:] + self._history_index = -1 + + self._cmd_entry.set_text("") + self._append_console_command(command) + self._busy = True + self._send_btn.set_sensitive(False) + + def work() -> None: + error: str | None = None + result: dict = {} + try: + result = self._service.send_repeater_command(peer_name, command) + except Exception as exc: + error = str(exc) or type(exc).__name__ + + def done() -> bool: + self._busy = False + self._send_btn.set_sensitive(True) + if error: + self._append_console_error(f"Error: {error}") + elif not result.get("success"): + reason = result.get("reason", "No response") + self._append_console_error(reason) + else: + text = result.get("response_text", "") + if text: + self._append_console_response(text) + else: + self._append_console_response("(no response)") + return False + + GLib.idle_add(done) + + threading.Thread(target=work, daemon=True).start() + + # -- Console output -------------------------------------------------------- + + def _append_console_command(self, text: str) -> None: + label = Gtk.Label(label=f"> {text}") + label.add_css_class("cli-command") + label.set_halign(Gtk.Align.START) + label.set_wrap(True) + label.set_wrap_mode(Pango.WrapMode.CHAR) + label.set_selectable(True) + self._console_box.append(label) + self._scroll_console() + + def _append_console_response(self, text: str) -> None: + label = Gtk.Label(label=text) + label.add_css_class("cli-response") + label.set_halign(Gtk.Align.START) + label.set_wrap(True) + label.set_wrap_mode(Pango.WrapMode.CHAR) + label.set_selectable(True) + self._console_box.append(label) + self._scroll_console() + + def _append_console_error(self, text: str) -> None: + label = Gtk.Label(label=text) + label.add_css_class("cli-error") + label.set_halign(Gtk.Align.START) + label.set_wrap(True) + label.set_wrap_mode(Pango.WrapMode.CHAR) + label.set_selectable(True) + self._console_box.append(label) + self._scroll_console() + + def _scroll_console(self) -> None: + def _do_scroll() -> bool: + adj = self._console_scroll.get_vadjustment() + adj.set_value(adj.get_upper()) + return False + + GLib.idle_add(_do_scroll) + + # -- Events ---------------------------------------------------------------- + + def _on_events(self, _store: object) -> None: + self._event_cursor, events = self._event_store.since(self._event_cursor, limit=200) + needs_refresh = False + for event in events: + etype = event.get("type", "") + if etype in ( + "repeater_login", + "repeater_logout", + "advert_received", + "contact_received", + ): + needs_refresh = True + if needs_refresh: + self._refresh_list() + + # -- Public API for MainWindow.navigate_to --------------------------------- + + def get_default_focus(self) -> Gtk.Widget | None: + return self._repeater_list diff --git a/src/meshcore_console/ui_gtk/windows/main_window.py b/src/meshcore_console/ui_gtk/windows/main_window.py index a373647..7fce100 100644 --- a/src/meshcore_console/ui_gtk/windows/main_window.py +++ b/src/meshcore_console/ui_gtk/windows/main_window.py @@ -65,6 +65,7 @@ def _build_main_ui(self) -> bool: from meshcore_console.ui_gtk.views.map import MapView from meshcore_console.ui_gtk.views.messages import MessagesView from meshcore_console.ui_gtk.views.peers import PeersView + from meshcore_console.ui_gtk.views.admin import AdminView from meshcore_console.ui_gtk.views.settings import SettingsView header_bar = self._build_header_bar() @@ -81,6 +82,7 @@ def _build_main_ui(self) -> bool: self._stack.add_named(PeersView(self._service, self._event_store, layout), "peers") self._stack.add_named(MessagesView(self._service, self._event_store, layout), "messages") self._stack.add_named(MapView(self._service, self._event_store, layout), "map") + self._stack.add_named(AdminView(self._service, self._event_store, layout), "admin") self._stack.add_named(SettingsView(self._service), "settings") self._stack.set_visible_child_name("analyzer") @@ -129,6 +131,7 @@ def _build_header_bar(self) -> Adw.HeaderBar: ("Peers", "peers"), ("Channels", "messages"), ("Map", "map"), + ("Admin", "admin"), ]: btn = Gtk.ToggleButton.new_with_label(label) btn.add_css_class("nav-button") @@ -416,7 +419,7 @@ def _debug_geometry_tick(self) -> bool: visible = self._stack.get_visible_child_name() print(f"[ui-geom] window={win_w}x{win_h} visible={visible}") print(f"[ui-geom] stack={self._stack.get_width()}x{self._stack.get_height()}") - for name in ("analyzer", "peers", "messages", "map", "settings"): + for name in ("analyzer", "peers", "messages", "map", "admin", "settings"): child = self._stack.get_child_by_name(name) if child is not None: min_w, nat_w, _min_b, _nat_b = child.measure(Gtk.Orientation.HORIZONTAL, -1) @@ -725,7 +728,8 @@ def _on_key_pressed( Gdk.KEY_2: "peers", Gdk.KEY_3: "messages", Gdk.KEY_4: "map", - Gdk.KEY_5: "settings", + Gdk.KEY_5: "admin", + Gdk.KEY_6: "settings", } page = mapping.get(keyval) if page is None: diff --git a/tests/integration/test_repeater_admin.py b/tests/integration/test_repeater_admin.py new file mode 100644 index 0000000..d6add7d --- /dev/null +++ b/tests/integration/test_repeater_admin.py @@ -0,0 +1,81 @@ +from meshcore_console.mock import MockMeshcoreClient + + +def test_admin_login() -> None: + client = MockMeshcoreClient() + result = client.login_to_repeater("repeater-1", "admin123") + + assert result["success"] is True + assert result["is_admin"] is True + + state = client.get_repeater_login_state("repeater-1") + assert state is not None + assert state.is_admin is True + assert state.is_guest is False + + +def test_guest_login() -> None: + client = MockMeshcoreClient() + result = client.guest_login_to_repeater("repeater-1") + + assert result["success"] is True + assert result["is_admin"] is False + + state = client.get_repeater_login_state("repeater-1") + assert state is not None + assert state.is_admin is False + assert state.is_guest is True + + +def test_logout() -> None: + client = MockMeshcoreClient() + client.login_to_repeater("repeater-1", "admin123") + client.logout_from_repeater("repeater-1") + + assert client.get_repeater_login_state("repeater-1") is None + assert client.list_logged_in_repeaters() == [] + + +def test_list_logged_in_repeaters() -> None: + client = MockMeshcoreClient() + client.login_to_repeater("rpt-a", "pw") + client.login_to_repeater("rpt-b", "pw") + + logged_in = client.list_logged_in_repeaters() + assert sorted(logged_in) == ["rpt-a", "rpt-b"] + + +def test_send_command() -> None: + client = MockMeshcoreClient() + client.login_to_repeater("repeater-1", "admin123") + result = client.send_repeater_command("repeater-1", "status") + + assert result["success"] is True + assert "response_text" in result + assert len(result["response_text"]) > 0 + + +def test_save_and_get_password() -> None: + client = MockMeshcoreClient() + client.login_to_repeater("repeater-1", "s3cret", save_password=True) + + assert client.get_saved_repeater_password("repeater-1") == "s3cret" + + +def test_delete_saved_password() -> None: + client = MockMeshcoreClient() + client.login_to_repeater("repeater-1", "s3cret", save_password=True) + client.delete_saved_repeater_password("repeater-1") + + assert client.get_saved_repeater_password("repeater-1") is None + + +def test_login_emits_event() -> None: + client = MockMeshcoreClient() + client.login_to_repeater("repeater-1", "admin123") + events = client.poll_events() + + login_events = [e for e in events if e.get("type") == "repeater_login"] + assert len(login_events) >= 1 + assert login_events[-1]["data"]["peer_name"] == "repeater-1" + assert login_events[-1]["data"]["is_admin"] is True diff --git a/tests/unit/test_db.py b/tests/unit/test_db.py index fe38090..0220572 100644 --- a/tests/unit/test_db.py +++ b/tests/unit/test_db.py @@ -48,6 +48,7 @@ def conn(tmp_path): "peers", "messages", "packets", + "repeater_passwords", } EXPECTED_COLUMNS = { diff --git a/tests/unit/test_repeater_store.py b/tests/unit/test_repeater_store.py new file mode 100644 index 0000000..6efc331 --- /dev/null +++ b/tests/unit/test_repeater_store.py @@ -0,0 +1,50 @@ +from meshcore_console.meshcore.db import open_db +from meshcore_console.meshcore.repeater_store import RepeaterPasswordStore + + +def test_save_and_get_password(tmp_path) -> None: + conn = open_db(str(tmp_path / "test.db")) + store = RepeaterPasswordStore(conn) + + store.save_password("repeater-1", "s3cret") + assert store.get_password("repeater-1") == "s3cret" + conn.close() + + +def test_get_missing_password(tmp_path) -> None: + conn = open_db(str(tmp_path / "test.db")) + store = RepeaterPasswordStore(conn) + + assert store.get_password("nonexistent") is None + conn.close() + + +def test_delete_password(tmp_path) -> None: + conn = open_db(str(tmp_path / "test.db")) + store = RepeaterPasswordStore(conn) + + store.save_password("repeater-1", "s3cret") + store.delete_password("repeater-1") + assert store.get_password("repeater-1") is None + conn.close() + + +def test_overwrite_password(tmp_path) -> None: + conn = open_db(str(tmp_path / "test.db")) + store = RepeaterPasswordStore(conn) + + store.save_password("repeater-1", "old") + store.save_password("repeater-1", "new") + assert store.get_password("repeater-1") == "new" + conn.close() + + +def test_list_saved(tmp_path) -> None: + conn = open_db(str(tmp_path / "test.db")) + store = RepeaterPasswordStore(conn) + + assert store.list_saved() == [] + store.save_password("bravo", "pw1") + store.save_password("alpha", "pw2") + assert store.list_saved() == ["alpha", "bravo"] + conn.close() From 5ec1bb549d89e9d5ee7e046f04baa547d980467d Mon Sep 17 00:00:00 2001 From: Cameron Will Date: Mon, 18 May 2026 11:53:05 -0400 Subject: [PATCH 2/2] fix: Address review comments on repeater admin PR - Filter command responses by originating repeater public key - Move logout to background thread to avoid blocking GTK main loop - Guard login/command completion handlers against stale selection - Constrain console label width to prevent GTK4 window blowout Co-Authored-By: Claude Opus 4.6 --- src/meshcore_console/meshcore/operations.py | 5 ++ src/meshcore_console/ui_gtk/views/admin.py | 97 +++++++++++++-------- 2 files changed, 66 insertions(+), 36 deletions(-) diff --git a/src/meshcore_console/meshcore/operations.py b/src/meshcore_console/meshcore/operations.py index 3c3b879..6dcd9e7 100644 --- a/src/meshcore_console/meshcore/operations.py +++ b/src/meshcore_console/meshcore/operations.py @@ -207,7 +207,12 @@ async def send_repeater_command( response_event = asyncio.Event() response_data: dict = {"text": None} + expected_key = contact.public_key + def _on_response(message_text: str, sender_contact: object) -> None: + sender_key = getattr(sender_contact, "public_key", None) + if sender_key is not None and sender_key != expected_key: + return response_data["text"] = message_text response_event.set() diff --git a/src/meshcore_console/ui_gtk/views/admin.py b/src/meshcore_console/ui_gtk/views/admin.py index 6a95e18..ee70a91 100644 --- a/src/meshcore_console/ui_gtk/views/admin.py +++ b/src/meshcore_console/ui_gtk/views/admin.py @@ -505,24 +505,29 @@ def work() -> None: def done() -> bool: self._busy = False + still_selected = ( + self._selected_repeater is not None + and self._selected_repeater.display_name == peer_name + ) if error: self._append_console_error(f"Login failed: {error}") - self._login_btn.set_sensitive(True) - self._guest_btn.set_sensitive(True) - self._update_login_ui(None) + if still_selected: + self._login_btn.set_sensitive(True) + self._guest_btn.set_sensitive(True) + self._update_login_ui(None) elif not result.get("success"): reason = result.get("reason", "Login failed") self._append_console_error(reason) - self._login_btn.set_sensitive(True) - self._guest_btn.set_sensitive(True) - self._update_login_ui(None) + if still_selected: + self._login_btn.set_sensitive(True) + self._guest_btn.set_sensitive(True) + self._update_login_ui(None) else: - state = self._service.get_repeater_login_state(peer_name) - self._update_login_ui(state) - self._append_console_response("Login successful") - # Refresh status tab to show login info - if self._selected_repeater: + if still_selected and self._selected_repeater is not None: + state = self._service.get_repeater_login_state(peer_name) + self._update_login_ui(state) self._populate_status(self._selected_repeater) + self._append_console_response("Login successful") self._refresh_list() return False @@ -531,15 +536,35 @@ def done() -> bool: threading.Thread(target=work, daemon=True).start() def _on_logout_clicked(self, _btn: Gtk.Button) -> None: - if self._selected_repeater is None: + if self._selected_repeater is None or self._busy: return peer_name = self._selected_repeater.display_name - self._service.logout_from_repeater(peer_name) - self._update_login_ui(None) - self._append_console_response("Logged out") - if self._selected_repeater: - self._populate_status(self._selected_repeater) - self._refresh_list() + self._busy = True + self._logout_btn.set_sensitive(False) + + def work() -> None: + error: str | None = None + try: + self._service.logout_from_repeater(peer_name) + except Exception as exc: + error = str(exc) or type(exc).__name__ + + def done() -> bool: + self._busy = False + if error: + self._append_console_error(f"Logout failed: {error}") + self._logout_btn.set_sensitive(True) + return False + if self._selected_repeater and self._selected_repeater.display_name == peer_name: + self._update_login_ui(None) + self._populate_status(self._selected_repeater) + self._append_console_response("Logged out") + self._refresh_list() + return False + + GLib.idle_add(done) + + threading.Thread(target=work, daemon=True).start() def _on_forget_clicked(self, _btn: Gtk.Button) -> None: if self._selected_repeater is None: @@ -615,6 +640,12 @@ def work() -> None: def done() -> bool: self._busy = False self._send_btn.set_sensitive(True) + still_selected = ( + self._selected_repeater is not None + and self._selected_repeater.display_name == peer_name + ) + if not still_selected: + return False if error: self._append_console_error(f"Error: {error}") elif not result.get("success"): @@ -634,34 +665,28 @@ def done() -> bool: # -- Console output -------------------------------------------------------- - def _append_console_command(self, text: str) -> None: - label = Gtk.Label(label=f"> {text}") - label.add_css_class("cli-command") + def _make_console_label(self, text: str, css_class: str) -> Gtk.Label: + label = Gtk.Label(label=text) + label.add_css_class(css_class) label.set_halign(Gtk.Align.START) + label.set_xalign(0.0) label.set_wrap(True) label.set_wrap_mode(Pango.WrapMode.CHAR) label.set_selectable(True) - self._console_box.append(label) + label.set_size_request(10, -1) + label.set_hexpand(True) + return label + + def _append_console_command(self, text: str) -> None: + self._console_box.append(self._make_console_label(f"> {text}", "cli-command")) self._scroll_console() def _append_console_response(self, text: str) -> None: - label = Gtk.Label(label=text) - label.add_css_class("cli-response") - label.set_halign(Gtk.Align.START) - label.set_wrap(True) - label.set_wrap_mode(Pango.WrapMode.CHAR) - label.set_selectable(True) - self._console_box.append(label) + self._console_box.append(self._make_console_label(text, "cli-response")) self._scroll_console() def _append_console_error(self, text: str) -> None: - label = Gtk.Label(label=text) - label.add_css_class("cli-error") - label.set_halign(Gtk.Align.START) - label.set_wrap(True) - label.set_wrap_mode(Pango.WrapMode.CHAR) - label.set_selectable(True) - self._console_box.append(label) + self._console_box.append(self._make_console_label(text, "cli-error")) self._scroll_console() def _scroll_console(self) -> None: