From 6bdf31cb2688cedd3c26706cabbc798c8e3e4dda Mon Sep 17 00:00:00 2001 From: rattatwinko Date: Fri, 28 Nov 2025 13:03:13 +0100 Subject: [PATCH] local server, useful for local things. Start a IRC Server on your LAN --- src/LocalServer.py | 213 ++++++++++++++++++++++++++++++++ src/main.py | 297 +++++++++++++++++++++++++++++++++++++++++++++ 2 files changed, 510 insertions(+) create mode 100644 src/LocalServer.py diff --git a/src/LocalServer.py b/src/LocalServer.py new file mode 100644 index 0000000..ac2542b --- /dev/null +++ b/src/LocalServer.py @@ -0,0 +1,213 @@ +import ipaddress +import logging +import socket +import threading +from typing import Callable, Iterable, List, Optional + +from irc import server as irc_server + +logger = logging.getLogger(__name__) + + +def _default_log_callback(message: str, color=None, bold: bool = False): + """Fallback logger when UI callback is not available.""" + logger.info(message) + + +class LocalOnlyIRCServer(irc_server.IRCServer): + """IRC server that only accepts connections from local/LAN addresses.""" + + def __init__( + self, + server_address, + request_handler_class, + allowed_networks: Iterable[ipaddress.IPv4Network], + blocked_callback: Optional[Callable[[str], None]] = None, + ): + self.allowed_networks = list(allowed_networks) + self.blocked_callback = blocked_callback + super().__init__(server_address, request_handler_class) + + def verify_request(self, request, client_address): + try: + ip = ipaddress.ip_address(client_address[0]) + for network in self.allowed_networks: + if ip in network: + return super().verify_request(request, client_address) + except ValueError: + logger.warning("Rejected malformed IP address: %s", client_address[0]) + return False + + if self.blocked_callback: + self.blocked_callback(client_address[0]) + logger.warning("Rejected non-LAN connection from %s", client_address[0]) + return False + + +class LocalServerManager: + """Manages the background IRC server lifecycle.""" + + DEFAULT_CHANNELS = ["#lobby"] + DEFAULT_ALLOWED_NETWORKS = [ + ipaddress.ip_network("127.0.0.0/8"), # Loopback + ipaddress.ip_network("10.0.0.0/8"), # RFC1918 + ipaddress.ip_network("172.16.0.0/12"), # RFC1918 + ipaddress.ip_network("192.168.0.0/16"), # RFC1918 + ipaddress.ip_network("169.254.0.0/16"), # Link-local + ] + + def __init__( + self, + log_callback: Callable[[str, Optional[object], bool], None] = _default_log_callback, + listen_host: str = "0.0.0.0", + listen_port: int = 6667, + ): + self.log_callback = log_callback or _default_log_callback + self.listen_host = listen_host + self.listen_port = listen_port + self.allowed_networks = list(self.DEFAULT_ALLOWED_NETWORKS) + self._channels = list(self.DEFAULT_CHANNELS) + + self._server: Optional[LocalOnlyIRCServer] = None + self._thread: Optional[threading.Thread] = None + self._lock = threading.RLock() + self._running = threading.Event() + self._ready = threading.Event() + self._error: Optional[Exception] = None + + # Public API --------------------------------------------------------- + def start(self, timeout: float = 5.0): + """Start the background IRC server.""" + with self._lock: + if self._running.is_set(): + raise RuntimeError("Local IRC server is already running.") + + self._running.set() + self._ready.clear() + self._error = None + + self._thread = threading.Thread( + target=self._serve_forever, + name="Local-IRC-Server", + daemon=True, + ) + self._thread.start() + + if not self._ready.wait(timeout): + self._running.clear() + raise TimeoutError("Local IRC server failed to start in time.") + + if self._error: + raise self._error + + def stop(self, timeout: float = 5.0): + """Stop the IRC server if it is running.""" + with self._lock: + if not self._running.is_set(): + return + + server = self._server + thread = self._thread + + if server: + server.shutdown() + server.server_close() + + if thread: + thread.join(timeout=timeout) + + with self._lock: + self._server = None + self._thread = None + self._running.clear() + self._ready.clear() + + def is_running(self) -> bool: + return self._running.is_set() + + def set_listen_host(self, host: str): + """Update the bind address. Local server must be stopped.""" + with self._lock: + if self._running.is_set(): + raise RuntimeError("Stop the server before changing the interface.") + self.listen_host = host + self._log(f"Local server interface set to {host}.") + + def get_channels(self) -> List[str]: + with self._lock: + return list(self._channels) + + def set_channels(self, channels: Iterable[str]): + cleaned = self._sanitize_channels(channels) + with self._lock: + self._channels = cleaned or list(self.DEFAULT_CHANNELS) + + if self.is_running(): + self._log( + "Channel list updated. Restart local server to apply changes.", + bold=True, + ) + + # Internal helpers --------------------------------------------------- + def _serve_forever(self): + try: + server = LocalOnlyIRCServer( + (self.listen_host, self.listen_port), + irc_server.IRCClient, + self.allowed_networks, + blocked_callback=lambda ip: self._log( + f"Blocked connection attempt from {ip}", bold=False + ), + ) + server.servername = socket.gethostname() or "wxirc-local" + + with self._lock: + self._server = server + seed_channels = self._channels or list(self.DEFAULT_CHANNELS) + for channel in seed_channels: + server.channels.setdefault(channel, irc_server.IRCChannel(channel)) + + self._log( + f"Local IRC server listening on {self.listen_host}:{self.listen_port}", + bold=True, + ) + self._ready.set() + server.serve_forever() + except Exception as exc: + logger.exception("Local IRC server failed: %s", exc) + self._error = exc + self._ready.set() + self._log(f"Local server error: {exc}", bold=True) + finally: + self._running.clear() + self._log("Local IRC server stopped.") + + def _sanitize_channels(self, channels: Iterable[str]) -> List[str]: + unique = [] + seen = set() + for channel in channels: + if not channel: + continue + name = channel.strip() + if not name.startswith("#"): + name = f"#{name}" + if not self._is_valid_channel(name): + continue + if name.lower() not in seen: + unique.append(name) + seen.add(name.lower()) + return unique + + @staticmethod + def _is_valid_channel(name: str) -> bool: + if len(name) < 2: + return False + allowed = "abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ0123456789-_#" + return all(ch in allowed for ch in name) + + def _log(self, message: str, color=None, bold: bool = False): + try: + self.log_callback(message, color, bold) + except Exception: + logger.info(message) + diff --git a/src/main.py b/src/main.py index 1b5a09e..8228315 100644 --- a/src/main.py +++ b/src/main.py @@ -17,6 +17,7 @@ from IRCPanel import IRCPanel from AboutDialog import AboutDialog from NotesDialog import NotesDialog from ScanWizard import ScanWizardDialog +from LocalServer import LocalServerManager # Set up logging logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s') @@ -39,6 +40,207 @@ class UIUpdate: self.args = args self.kwargs = kwargs + +class ManageChannelsDialog(wx.Dialog): + """Simple dialog for curating the local server channel allowlist.""" + + def __init__(self, parent, channels): + super().__init__(parent, title="Manage Local Channels", size=(360, 420)) + try: + self.SetIcon(parent.GetIcon()) + except Exception: + pass + panel = wx.Panel(self) + panel.SetBackgroundColour(parent.theme["window_bg"]) + + main_sizer = wx.BoxSizer(wx.VERTICAL) + info = wx.StaticText( + panel, + label="Channels are shared with anyone on your LAN who joins the built-in server.", + ) + info.Wrap(320) + main_sizer.Add(info, 0, wx.ALL | wx.EXPAND, 8) + + self.list_box = wx.ListBox(panel) + for channel in channels: + self.list_box.Append(channel) + main_sizer.Add(self.list_box, 1, wx.ALL | wx.EXPAND, 8) + + input_sizer = wx.BoxSizer(wx.HORIZONTAL) + self.channel_input = wx.TextCtrl(panel, style=wx.TE_PROCESS_ENTER) + self.channel_input.SetHint("#channel-name") + add_btn = wx.Button(panel, label="Add") + input_sizer.Add(self.channel_input, 1, wx.RIGHT, 4) + input_sizer.Add(add_btn, 0) + main_sizer.Add(input_sizer, 0, wx.EXPAND | wx.LEFT | wx.RIGHT, 8) + + remove_btn = wx.Button(panel, label="Remove Selected") + reset_btn = wx.Button(panel, label="Reset to #lobby") + btn_sizer = wx.BoxSizer(wx.HORIZONTAL) + btn_sizer.Add(remove_btn, 1, wx.RIGHT, 4) + btn_sizer.Add(reset_btn, 1) + main_sizer.Add(btn_sizer, 0, wx.EXPAND | wx.LEFT | wx.RIGHT | wx.TOP, 8) + + button_bar = wx.StdDialogButtonSizer() + ok_btn = wx.Button(panel, wx.ID_OK) + cancel_btn = wx.Button(panel, wx.ID_CANCEL) + button_bar.AddButton(ok_btn) + button_bar.AddButton(cancel_btn) + button_bar.Realize() + main_sizer.Add(button_bar, 0, wx.ALL | wx.EXPAND, 8) + + panel.SetSizer(main_sizer) + + # Bindings + add_btn.Bind(wx.EVT_BUTTON, self.on_add) + remove_btn.Bind(wx.EVT_BUTTON, self.on_remove) + reset_btn.Bind(wx.EVT_BUTTON, self.on_reset) + self.channel_input.Bind(wx.EVT_TEXT_ENTER, self.on_add) + ok_btn.Bind(wx.EVT_BUTTON, self.on_ok) + + def get_channels(self): + return [self.list_box.GetString(i) for i in range(self.list_box.GetCount())] + + def on_add(self, event): + value = self.channel_input.GetValue().strip() + if not value: + return + if not value.startswith('#'): + value = f"#{value}" + if not self._is_valid_channel(value): + wx.MessageBox( + "Channel names may contain letters, numbers, -, and _ only.", + "Invalid Channel", + wx.OK | wx.ICON_WARNING, + ) + return + if self.list_box.FindString(value) != wx.NOT_FOUND: + wx.MessageBox("Channel already exists.", "Duplicate Channel", wx.OK | wx.ICON_INFORMATION) + return + self.list_box.Append(value) + self.channel_input.Clear() + + def on_remove(self, event): + selection = self.list_box.GetSelection() + if selection != wx.NOT_FOUND: + self.list_box.Delete(selection) + + def on_reset(self, event): + self.list_box.Clear() + self.list_box.Append("#lobby") + + def on_ok(self, event): + if self.list_box.GetCount() == 0: + wx.MessageBox("Add at least one channel before saving.", "No Channels", wx.OK | wx.ICON_INFORMATION) + return + event.Skip() + + @staticmethod + def _is_valid_channel(name): + if len(name) < 2: + return False + allowed = "abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ0123456789-_#" + return all(ch in allowed for ch in name) + + +class InterfaceSelectDialog(wx.Dialog): + """Dialog that lets the user pick which local interface the server should bind to.""" + + def __init__(self, parent, current_host="127.0.0.1"): + super().__init__(parent, title="Select Network Interface", size=(420, 380)) + try: + self.SetIcon(parent.GetIcon()) + except Exception: + pass + + self.selected_host = current_host + panel = wx.Panel(self) + panel.SetBackgroundColour(parent.theme["window_bg"]) + main_sizer = wx.BoxSizer(wx.VERTICAL) + + info = wx.StaticText( + panel, + label="Choose the Network interface where your server should run:\n" + "You are EXPOSING a Server to your LOCAL Network, this may give away who you are!\n", + ) + info.Wrap(380) + main_sizer.Add(info, 0, wx.ALL | wx.EXPAND, 10) + + self.interface_list = wx.ListCtrl(panel, style=wx.LC_REPORT | wx.BORDER_SUNKEN) + self.interface_list.InsertColumn(0, "Interface", width=180) + self.interface_list.InsertColumn(1, "Address", width=180) + + self.interfaces = self._gather_interfaces() + current_index = 0 + for idx, entry in enumerate[tuple[str, str]](self.interfaces): + name, address = entry + self.interface_list.InsertItem(idx, name) + self.interface_list.SetItem(idx, 1, address) + if address == current_host: + current_index = idx + self.interface_list.Select(current_index) + self.interface_list.EnsureVisible(current_index) + if self.interfaces: + self.selected_host = self.interfaces[current_index][1] + self.interface_list.Bind(wx.EVT_LIST_ITEM_SELECTED, self.on_select) + self.interface_list.Bind(wx.EVT_LIST_ITEM_ACTIVATED, self.on_activate) + main_sizer.Add(self.interface_list, 1, wx.ALL | wx.EXPAND, 10) + + button_bar = wx.StdDialogButtonSizer() + ok_btn = wx.Button(panel, wx.ID_OK) + cancel_btn = wx.Button(panel, wx.ID_CANCEL) + button_bar.AddButton(ok_btn) + button_bar.AddButton(cancel_btn) + button_bar.Realize() + main_sizer.Add(button_bar, 0, wx.ALL | wx.EXPAND, 10) + + ok_btn.Bind(wx.EVT_BUTTON, self.on_ok) + + panel.SetSizer(main_sizer) + + def on_select(self, event): + index = event.GetIndex() + _, address = self.interfaces[index] + self.selected_host = address + + def _gather_interfaces(self): + entries = [ + ("Loopback only", "127.0.0.1"), + ("All interfaces", "0.0.0.0"), + ] + try: + import psutil + + seen = {addr for _, addr in entries} + for name, addrs in psutil.net_if_addrs().items(): + for addr in addrs: + if addr.family == socket.AF_INET and addr.address not in seen: + label = f"{name}" + entries.append((label, addr.address)) + seen.add(addr.address) + except Exception as e: + logger.warning(f"Unable to enumerate network interfaces: {e}") + + return entries + + def get_selected_host(self): + return self.selected_host + + def on_activate(self, event): + self.on_select(event) + self.EndModal(wx.ID_OK) + + def on_ok(self, event): + index = self.interface_list.GetFirstSelected() + if index == -1 and self.interfaces: + wx.MessageBox("Select an interface before starting the server.", "No Interface Selected", wx.OK | wx.ICON_INFORMATION) + return + if index != -1: + _, address = self.interfaces[index] + self.selected_host = address + event.Skip() + + class IRCFrame(wx.Frame): def __init__(self): super().__init__(None, title="wxIRC", size=(1200, 700)) @@ -71,6 +273,12 @@ class IRCFrame(wx.Frame): self.timestamps = True self.notes_data = defaultdict(dict) + self.server_menu_items = {} + self.local_bind_host = "127.0.0.1" + self.local_server_manager = LocalServerManager( + log_callback=self.log_local_server, + listen_host=self.local_bind_host, + ) # User color mapping - darker colors for white theme self.user_colors = {} @@ -435,10 +643,27 @@ class IRCFrame(wx.Frame): self.Bind(wx.EVT_MENU, self.on_menu_help, id=207) self.Bind(wx.EVT_MENU, self.on_scan_local_network, id=210) + # IRC Server menu + server_menu = wx.Menu() + start_item = server_menu.Append(401, "Start Local Server") + stop_item = server_menu.Append(402, "Stop Local Server") + stop_item.Enable(False) + server_menu.AppendSeparator() + manage_item = server_menu.Append(403, "Manage Channels...") + self.server_menu_items = { + "start": start_item, + "stop": stop_item, + "manage": manage_item, + } + self.Bind(wx.EVT_MENU, self.on_start_local_server, id=401) + self.Bind(wx.EVT_MENU, self.on_stop_local_server, id=402) + self.Bind(wx.EVT_MENU, self.on_manage_local_channels, id=403) + menubar.Append(file_menu, "&File") menubar.Append(edit_menu, "&Edit") menubar.Append(channel_menu, "&Channel") menubar.Append(tools_menu, "&Tools") + menubar.Append(server_menu, "&IRC Server") self.SetMenuBar(menubar) except Exception as e: @@ -903,6 +1128,16 @@ Available commands: self.channels["SERVER"].add_system_message(f"{self.get_timestamp()}{message}", color, bold) except Exception as e: logger.error(f"Error logging server message: {e}") + + def log_local_server(self, message, color=None, bold=False): + """Bridge LocalServerManager log output into the Server tab.""" + display_color = color or wx.Colour(34, 139, 34) # Forest green + self.safe_ui_update( + self.log_server, + f"[Local Server] {message}", + display_color, + bold, + ) def log_channel_message(self, channel, username, message, is_action=False, is_system=False): """Log a message to a channel with username coloring""" @@ -972,6 +1207,65 @@ Available commands: return False # Menu handlers + def update_server_menu_state(self): + try: + if not self.server_menu_items: + return + running = self.local_server_manager.is_running() + self.server_menu_items["start"].Enable(not running) + self.server_menu_items["stop"].Enable(running) + except Exception as e: + logger.error(f"Error updating server menu state: {e}") + + def on_start_local_server(self, event): + try: + dlg = InterfaceSelectDialog(self, self.local_bind_host) + result = dlg.ShowModal() + if result != wx.ID_OK: + dlg.Destroy() + return + selected_host = dlg.get_selected_host() + dlg.Destroy() + + if selected_host != self.local_bind_host: + self.local_server_manager.set_listen_host(selected_host) + self.local_bind_host = selected_host + + self.local_server_manager.start() + self.update_server_menu_state() + self.SetStatusText(f"Local IRC server running on {self.local_bind_host}:6667 (LAN only)") + self.log_local_server(f"Server is online and listening on {self.local_bind_host}:6667.") + except Exception as e: + logger.error(f"Failed to start local server: {e}") + self.log_local_server(f"Failed to start: {e}", wx.Colour(255, 0, 0), bold=True) + wx.MessageBox(f"Could not start local IRC server:\n{e}", "Server Error", wx.OK | wx.ICON_ERROR) + self.update_server_menu_state() + + def on_stop_local_server(self, event): + try: + self.local_server_manager.stop() + self.update_server_menu_state() + self.SetStatusText("Local IRC server stopped") + self.log_local_server("Server stopped.") + except Exception as e: + logger.error(f"Failed to stop local server: {e}") + self.log_local_server(f"Failed to stop: {e}", wx.Colour(255, 0, 0), bold=True) + + def on_manage_local_channels(self, event): + try: + dlg = ManageChannelsDialog(self, self.local_server_manager.get_channels()) + if dlg.ShowModal() == wx.ID_OK: + channels = dlg.get_channels() + self.local_server_manager.set_channels(channels) + self.log_local_server( + f"Manage Channels updated: {', '.join(channels)}", + wx.Colour(0, 100, 0), + ) + dlg.Destroy() + except Exception as e: + logger.error(f"Error managing local channels: {e}") + wx.MessageBox(f"Unable to open Manage Channels: {e}", "Error", wx.OK | wx.ICON_ERROR) + def on_menu_join(self, event): try: dlg = wx.TextEntryDialog(self, "Enter channel name:", "Join Channel") @@ -1368,6 +1662,9 @@ COMMANDS (type /help in chat for full list): # Stop UI timer first if self.ui_timer and self.ui_timer.IsRunning(): self.ui_timer.Stop() + + if self.local_server_manager and self.local_server_manager.is_running(): + self.local_server_manager.stop() # Notes data will be lost when app closes (RAM only) # User can save to file if they want persistence