#!/usr/bin/env python3 from views import ActiveView, InactiveView, LogView from data_loader import load_customers from models import Customer # from services import VPNManager, VPNStatus, VPNConnectionError # Temporarily disabled due to syntax errors from services import VPNManager, VPNStatus, VPNConnectionError import sys import logging from gi.repository import Gtk, Gdk, GLib, Gio import gi gi.require_version('Gtk', '3.0') class VPNManagerWindow: vpn_manager: VPNManager def __init__(self): self.customers = load_customers() self.filtered_customers = self.customers.copy() self.current_location = None # Track user's current location # VPN manager will be initialized after UI setup self.vpn_manager = None # Create main window self.window = Gtk.Window() self.window.set_title("VPN Manager") self.window.set_default_size(1200, 750) self.window.connect("delete-event", self.quit_app_from_close) self.window.connect("window-state-event", self.on_window_state_event) # Set up minimal CSS for GNOME-style cards self.setup_css() # Create UI self.setup_ui() self.vpn_manager = VPNManager() def setup_css(self): """Minimal CSS for GNOME-style cards""" css_provider = Gtk.CssProvider() css_provider.load_from_file(Gio.File.new_for_path('style.css')) # css_provider.load_from_data(css.encode()) # Apply CSS to default screen screen = Gdk.Screen.get_default() style_context = Gtk.StyleContext() style_context.add_provider_for_screen( screen, css_provider, Gtk.STYLE_PROVIDER_PRIORITY_APPLICATION ) def setup_ui(self): # Use HeaderBar for native GNOME look header_bar = Gtk.HeaderBar() header_bar.set_show_close_button(True) header_bar.set_title("VPN Manager") header_bar.set_subtitle("Connection Manager") self.window.set_titlebar(header_bar) # Main container with proper spacing main_vbox = Gtk.Box(orientation=Gtk.Orientation.VERTICAL, spacing=12) main_vbox.set_margin_start(12) main_vbox.set_margin_end(12) main_vbox.set_margin_top(12) main_vbox.set_margin_bottom(12) self.window.add(main_vbox) # Current location display - enhanced info box self.location_info_box = self._create_location_info_box() main_vbox.pack_start(self.location_info_box, False, False, 0) # Search bar with SearchEntry self.search_entry = Gtk.SearchEntry() self.search_entry.set_placeholder_text( "Search customers, locations, or hosts... (* for all)") self.search_entry.connect("search-changed", self.filter_customers) main_vbox.pack_start(self.search_entry, False, False, 0) # Create a stack to switch between views self.view_stack = Gtk.Stack() self.view_stack.set_transition_type(Gtk.StackTransitionType.CROSSFADE) self.view_stack.set_transition_duration(200) main_vbox.pack_start(self.view_stack, True, True, 0) # Get callbacks for views callbacks = self.get_callbacks() # Create active view (shown by default) self.active_view = ActiveView(callbacks) self.view_stack.add_named(self.active_view.widget, "active") # Create inactive view (shown when searching) self.inactive_view = InactiveView(callbacks) self.view_stack.add_named(self.inactive_view.widget, "inactive") # Create log section at bottom (collapsible) self._create_log_section(main_vbox) # Initialize VPN manager (temporarily disabled due to syntax errors) # TODO: Fix VPN manager syntax and re-enable self.vpn_manager = None self.log_view.log_info( "VPN manager temporarily disabled for debugging") self.log_view.log_info("Using mock mode for VPN operations") # Render initial data self.render_customers() # Update VPN status from actual connections self.update_vpn_status() def _setup_logging(self): """Set up logging to route VPN manager logs to LogView.""" # Create a custom handler that forwards to our LogView class LogViewHandler(logging.Handler): def __init__(self, log_view): super().__init__() self.log_view = log_view def emit(self, record): try: msg = self.format(record) if record.levelno >= logging.ERROR: self.log_view.log_error(msg) elif record.levelno >= logging.WARNING: self.log_view.log_warning(msg) elif record.levelno >= logging.INFO: self.log_view.log_info(msg) else: # DEBUG self.log_view.log_debug(msg) except Exception: self.handleError(record) # Set up handler for VPN manager logs handler = LogViewHandler(self.log_view) handler.setFormatter(logging.Formatter('%(message)s')) # Add handler to VPN manager logger vpn_logger = logging.getLogger('services.vpn_manager') vpn_logger.addHandler(handler) vpn_logger.setLevel(logging.DEBUG) vpn_logger.propagate = False # Don't send to root logger def get_callbacks(self): """Return callback functions for widget interactions""" return { 'toggle_connection': self.toggle_connection, 'set_location_active': self.set_location_active, 'deactivate_location': self.deactivate_location, 'set_current_location': self.set_current_location, 'open_service': self.open_service, 'open_customer_service': self.open_customer_service } def render_customers(self): # Check if we're in search mode search_term = self.search_entry.get_text().strip() is_searching = bool(search_term) # Separate customers with active and inactive locations customers_with_active = [] customers_with_inactive = [] for customer in self.filtered_customers: active_locations = customer.get_active_locations() inactive_locations = customer.get_inactive_locations() # Prepare active locations (shown when not searching) if active_locations: customer_data = Customer(name=customer.name) customer_data.services = customer.services customer_data.locations = active_locations customers_with_active.append(customer_data) # Prepare inactive locations (shown when searching) if inactive_locations: customer_data = Customer(name=customer.name) customer_data.services = customer.services customer_data.locations = inactive_locations customers_with_inactive.append(customer_data) # Update views based on mode if is_searching: # Search mode: Switch to inactive view and update it self.view_stack.set_visible_child_name("inactive") self.inactive_view.update(customers_with_inactive, search_term) else: # Normal mode: Switch to active view and update it self.view_stack.set_visible_child_name("active") self.active_view.update(customers_with_active) self.window.show_all() def set_location_active(self, location, customer_name): for customer in self.customers: if customer.name == customer_name: target_location = customer.get_location_by_name(location.name) if target_location: target_location.active = True self.log_view.log_info( f"Activated location: {customer.name} - {target_location.name}") print( f"Mock: Setting {customer.name} - {target_location.name} as active") break # Clear search and return to active view self.search_entry.set_text("") self.render_customers() def deactivate_location(self, location, customer_name): for customer in self.customers: if customer.name == customer_name: target_location = customer.get_location_by_name(location.name) if target_location: target_location.active = False target_location.connected = False # Disconnect when deactivating self.log_view.log_info( f"Deactivated location: {customer.name} - {target_location.name}") print( f"Mock: Deactivating {customer.name} - {target_location.name}") break self.render_customers() def set_current_location(self, location, customer_name): """Set the user's current location.""" for customer in self.customers: if customer.name == customer_name: target_location = customer.get_location_by_name(location.name) if target_location: self.current_location = ( customer.name, target_location.name) self.log_view.log_info( f"Current location set to: {customer.name} - {target_location.name}") print( f"Current location set to: {customer.name} - {target_location.name}") self.update_current_location_display() break def _create_location_info_box(self): """Create the enhanced current location info box.""" frame = Gtk.Frame() frame.get_style_context().add_class("location-info") frame.set_shadow_type(Gtk.ShadowType.NONE) vbox = Gtk.Box(orientation=Gtk.Orientation.VERTICAL, spacing=4) frame.add(vbox) # Title row with infrastructure toggle title_box = Gtk.Box(orientation=Gtk.Orientation.HORIZONTAL, spacing=8) vbox.pack_start(title_box, False, False, 0) title_label = Gtk.Label() title_label.set_markup("📍 Current Location") title_label.set_halign(Gtk.Align.START) title_box.pack_start(title_label, False, False, 0) # Infrastructure toggle button (only shown when location is set) self.infrastructure_toggle = Gtk.Button() self.infrastructure_toggle.set_relief(Gtk.ReliefStyle.NONE) self.infrastructure_toggle.set_can_focus(False) self.infrastructure_toggle.set_label("▶") self.infrastructure_toggle.set_tooltip_text("Show/hide infrastructure") self.infrastructure_toggle.connect( "clicked", self._on_infrastructure_toggle) self.infrastructure_toggle.set_visible(False) title_box.pack_end(self.infrastructure_toggle, False, False, 0) # Location details label self.location_details_label = Gtk.Label() self.location_details_label.set_markup("Not set") self.location_details_label.set_halign(Gtk.Align.START) vbox.pack_start(self.location_details_label, False, False, 0) # Additional info row (hosts, services, etc.) self.location_extra_info = Gtk.Label() self.location_extra_info.set_halign(Gtk.Align.START) self.location_extra_info.set_visible(False) vbox.pack_start(self.location_extra_info, False, False, 0) # Infrastructure section (collapsible) self.infrastructure_box = Gtk.Box( orientation=Gtk.Orientation.VERTICAL, spacing=6) self.infrastructure_box.set_margin_top(8) self.infrastructure_box.set_visible(False) vbox.pack_start(self.infrastructure_box, False, False, 0) # Track infrastructure expanded state self.infrastructure_expanded = False return frame def _create_log_section(self, main_vbox): """Create the collapsible log section at the bottom.""" # Log section container log_container = Gtk.Box( orientation=Gtk.Orientation.VERTICAL, spacing=0) log_container.get_style_context().add_class("log-section") main_vbox.pack_end(log_container, False, False, 0) # Log header with toggle button log_header = Gtk.Box(orientation=Gtk.Orientation.HORIZONTAL, spacing=8) log_header.set_margin_start(12) log_header.set_margin_end(12) log_header.set_margin_top(8) log_header.set_margin_bottom(8) log_container.pack_start(log_header, False, False, 0) # Toggle button for log visibility self.log_toggle = Gtk.Button() self.log_toggle.set_relief(Gtk.ReliefStyle.NONE) self.log_toggle.set_can_focus(False) self.log_toggle.set_label("▲") self.log_toggle.set_tooltip_text("Show/hide command log") self.log_toggle.connect("clicked", self._on_log_toggle) log_header.pack_start(self.log_toggle, False, False, 0) # Log section label log_section_label = Gtk.Label() log_section_label.set_markup("Command Log") log_section_label.set_halign(Gtk.Align.START) log_header.pack_start(log_section_label, False, False, 0) # Create the log view self.log_view = LogView() log_container.pack_start(self.log_view.widget, False, False, 0) # Start with log collapsed self.log_expanded = False self.log_view.set_visible(False) # Log some initial messages self.log_view.log_info("VPN Manager started") self.log_view.log_info(f"Loaded {len(self.customers)} customers") def _on_log_toggle(self, button): """Toggle log section visibility.""" self.log_expanded = not self.log_expanded if self.log_expanded: self.log_toggle.set_label("▼") self.log_view.set_visible(True) else: self.log_toggle.set_label("▲") self.log_view.set_visible(False) def update_current_location_display(self): """Update the current location display with detailed information.""" if self.current_location: customer_name, location_name = self.current_location # Find the actual location object location = None for customer in self.customers: if customer.name == customer_name: location = customer.get_location_by_name(location_name) if location: break if location: # Main location info self.location_details_label.set_markup( f"{customer_name} - {location_name}" ) # Extra info about the location host_count = len(location.hosts) total_hosts = len(location.get_all_hosts_flat()) vpn_type = location.vpn_type.value extra_text = f"{vpn_type} VPN" if location.external_addresses: if len(location.external_addresses) == 1: extra_text += f" • 🌐 {location.external_addresses[0]}" else: extra_text += f" • 🌐 {len(location.external_addresses)} endpoints" if location.networks: extra_text += f" • 📡 {len(location.networks)} network{'s' if len(location.networks) > 1 else ''}" extra_text += f" • {host_count} hosts" if total_hosts > host_count: extra_text += f" ({total_hosts} total with VMs)" extra_text += "" self.location_extra_info.set_markup(extra_text) self.location_extra_info.set_visible(True) # Show infrastructure toggle and rebuild infrastructure self.infrastructure_toggle.set_visible(True) self._rebuild_infrastructure_display(location) else: self.location_details_label.set_markup( f"{customer_name} - {location_name}" ) self.location_extra_info.set_visible(False) self.infrastructure_toggle.set_visible(False) else: self.location_details_label.set_markup("Not set") self.location_extra_info.set_visible(False) self.infrastructure_toggle.set_visible(False) self.infrastructure_box.set_visible(False) def _rebuild_infrastructure_display(self, location): """Rebuild the infrastructure display for the current location.""" # Clear existing infrastructure widgets for child in self.infrastructure_box.get_children(): child.destroy() # Add network information if available if location.networks or location.external_addresses: network_label = Gtk.Label() network_label.set_markup("Network Configuration") network_label.set_halign(Gtk.Align.START) network_label.set_margin_bottom(4) self.infrastructure_box.pack_start(network_label, False, False, 0) # External addresses if location.external_addresses: for i, address in enumerate(location.external_addresses): ext_box = Gtk.Box( orientation=Gtk.Orientation.HORIZONTAL, spacing=8) ext_box.set_margin_start(12) self.infrastructure_box.pack_start( ext_box, False, False, 0) label_text = "🌐 External:" if i == 0 else "🌐 Backup:" ext_label = Gtk.Label() ext_label.set_markup( f"{label_text} {address}") ext_label.set_halign(Gtk.Align.START) ext_box.pack_start(ext_label, False, False, 0) # Internal networks if location.networks: for network in location.networks: net_box = Gtk.Box( orientation=Gtk.Orientation.HORIZONTAL, spacing=8) net_box.set_margin_start(12) self.infrastructure_box.pack_start( net_box, False, False, 0) net_label = Gtk.Label() net_label.set_markup( f"📡 Network: {network}") net_label.set_halign(Gtk.Align.START) net_box.pack_start(net_label, False, False, 0) # Add spacing before infrastructure if location.hosts: spacer = Gtk.Box() spacer.set_size_request(-1, 8) self.infrastructure_box.pack_start(spacer, False, False, 0) if not location.hosts: return # Add infrastructure label infra_label = Gtk.Label() infra_label.set_markup("Infrastructure") infra_label.set_halign(Gtk.Align.START) infra_label.set_margin_bottom(4) self.infrastructure_box.pack_start(infra_label, False, False, 0) # Add hosts for host in location.hosts: host_box = Gtk.Box( orientation=Gtk.Orientation.HORIZONTAL, spacing=8) host_box.set_margin_start(12) self.infrastructure_box.pack_start(host_box, False, False, 0) # Host type icon host_type_icons = { 'Linux': '🐧', 'Windows': '🪟', 'Windows Server': '🏢', 'Proxmox': '📦', 'ESXi': '⚙️', 'Router': '📡', 'Switch': '🔀' } icon = host_type_icons.get(host.host_type.value, '💻') # Host info host_label = Gtk.Label() service_count = len(host.services) vm_count = len(host.sub_hosts) host_text = f"{icon} {host.name} ({host.ip_address})" if service_count > 0: host_text += f" • {service_count} services" if vm_count > 0: host_text += f" • {vm_count} VMs" host_label.set_markup(f"{host_text}") host_label.set_halign(Gtk.Align.START) host_box.pack_start(host_label, False, False, 0) # Add sub-hosts (VMs) if any if host.sub_hosts: for vm in host.sub_hosts: vm_box = Gtk.Box( orientation=Gtk.Orientation.HORIZONTAL, spacing=8) vm_box.set_margin_start(24) self.infrastructure_box.pack_start(vm_box, False, False, 0) vm_icon = host_type_icons.get(vm.host_type.value, '💻') vm_service_count = len(vm.services) vm_text = f"{vm_icon} {vm.name} ({vm.ip_address})" if vm_service_count > 0: vm_text += f" • {vm_service_count} services" vm_label = Gtk.Label() vm_label.set_markup(f"{vm_text}") vm_label.set_halign(Gtk.Align.START) vm_box.pack_start(vm_label, False, False, 0) # Show all widgets (but container might be hidden) self.infrastructure_box.show_all() def _on_infrastructure_toggle(self, button): """Toggle infrastructure section visibility.""" self.infrastructure_expanded = not self.infrastructure_expanded if self.infrastructure_expanded: self.infrastructure_toggle.set_label("▼") self.infrastructure_box.set_visible(True) else: self.infrastructure_toggle.set_label("▶") self.infrastructure_box.set_visible(False) def filter_customers(self, entry): search_term = entry.get_text().strip() # Check for wildcard - show all customers if search_term == "*": self.filtered_customers = self.customers.copy() elif search_term: # Normal search logic search_term_lower = search_term.lower() self.filtered_customers = [] for customer in self.customers: # Check if search term matches customer name if search_term_lower in customer.name.lower(): self.filtered_customers.append(customer) continue # Check customer services if any(search_term_lower in service.name.lower() or search_term_lower in service.url.lower() or search_term_lower in service.service_type.lower() for service in customer.services): self.filtered_customers.append(customer) continue # Check locations and their hosts for location in customer.locations: # Check location name if search_term_lower in location.name.lower(): self.filtered_customers.append(customer) break # Check hosts and their services in this location def search_hosts(hosts): for host in hosts: # Check IP addresses (search in any of the host's IPs) ip_match = any(search_term_lower in host_ip.ip_address.lower( ) for host_ip in host.ip_addresses) if (search_term_lower in host.name.lower() or ip_match or search_term_lower in host.host_type.value.lower() or search_term_lower in host.description.lower()): return True # Check host services if any(search_term_lower in service.name.lower() or search_term_lower in str(service.port).lower() or search_term_lower in service.service_type.value.lower() for service in host.services): return True # Check sub-hosts recursively if search_hosts(host.sub_hosts): return True return False if search_hosts(location.hosts): self.filtered_customers.append(customer) break else: # Empty search - show all customers self.filtered_customers = self.customers.copy() self.render_customers() def toggle_connection(self, location): # Use actual VPN manager if location.connected: # Disconnect self.log_view.log_info(f"Disconnecting from {location.name}...") success = self.vpn_manager.disconnect_vpn(location) if success: location.connected = False self.log_view.log_success(f"Disconnected from {location.name}") else: self.log_view.log_error( f"Failed to disconnect from {location.name}") else: # Connect self.log_view.log_info( f"Connecting to {location.name} via {location.vpn_type.value}...") success = self.vpn_manager.connect_vpn(location) if success: location.connected = True self.log_view.log_success(f"Connected to {location.name}") else: self.log_view.log_error( f"Failed to connect to {location.name}") self.render_customers() # Update VPN status after connection change self.update_vpn_status() def open_service(self, service): # Get the host IP from context - this would need to be passed properly in a real implementation print( f"Mock: Opening {service.service_type.value} service: {service.name} on port {service.port}") def open_customer_service(self, customer_service): print( f"Mock: Opening customer service: {customer_service.name} at {customer_service.url}") def show_window_from_tray(self, _icon=None, _item=None): # Use GLib.idle_add to safely call GTK functions from the tray thread GLib.idle_add(self._show_window_safe) def _show_window_safe(self): """Safely show window in main GTK thread""" self.window.deiconify() self.window.present() self.window.show_all() return False # Don't repeat the idle call def on_window_state_event(self, _widget, event): """Handle window state changes - hide to tray when minimized""" if event.new_window_state & Gdk.WindowState.ICONIFIED: self.window.hide() return False def quit_app_from_close(self, _widget=None, _event=None): """Quit app when close button is pressed""" self.quit_app() return False def update_vpn_status(self): """Update location connection status from actual VPN manager.""" if not self.vpn_manager: return # Only update status for active locations to avoid unnecessary nmcli calls for customer in self.customers: for location in customer.locations: if location.active: # Only check active locations try: status = self.vpn_manager.get_connection_status( location) location.connected = (status == VPNStatus.CONNECTED) except VPNConnectionError: # If we can't get status, assume disconnected location.connected = False def quit_app(self, _widget=None): # Stop the tray icon if hasattr(self, 'tray_icon'): self.tray_icon.stop() Gtk.main_quit() sys.exit(0) def run(self): self.window.show_all() Gtk.main() def main(): app = VPNManagerWindow() app.run() if __name__ == "__main__": main()