rattatwinko/ircdvd
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases 1 Wiki Activity

local server, useful for local things. Start a IRC Server on your LAN

Browse Source
This commit is contained in:
rattatwinko
2025-11-28 13:03:13 +01:00
parent 8737720af6
commit 6bdf31cb26
2 changed files with 510 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files

213
src/LocalServer.py Normal file
View File

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

297
src/main.py
View File

@@ -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:
@@ -904,6 +1129,16 @@ Available commands:
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"""
try:
@@ -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")
@@ -1369,6 +1663,9 @@ COMMANDS (type /help in chat for full list):
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