#!/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()