691 lines
28 KiB
Python
691 lines
28 KiB
Python
#!/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("<b>📍 Current Location</b>")
|
|
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("<i>Not set</i>")
|
|
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("<b>Command Log</b>")
|
|
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"<b>{customer_name}</b> - {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"<small>{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 += "</small>"
|
|
|
|
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"<b>{customer_name}</b> - {location_name}"
|
|
)
|
|
self.location_extra_info.set_visible(False)
|
|
self.infrastructure_toggle.set_visible(False)
|
|
else:
|
|
self.location_details_label.set_markup("<i>Not set</i>")
|
|
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("<b>Network Configuration</b>")
|
|
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 = "🌐 <b>External:</b>" if i == 0 else "🌐 <b>Backup:</b>"
|
|
ext_label = Gtk.Label()
|
|
ext_label.set_markup(
|
|
f"<small>{label_text} {address}</small>")
|
|
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"<small>📡 <b>Network:</b> {network}</small>")
|
|
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("<b>Infrastructure</b>")
|
|
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} <b>{host.name}</b> ({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"<small>{host_text}</small>")
|
|
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} <i>{vm.name}</i> ({vm.ip_address})"
|
|
if vm_service_count > 0:
|
|
vm_text += f" • {vm_service_count} services"
|
|
|
|
vm_label = Gtk.Label()
|
|
vm_label.set_markup(f"<small>{vm_text}</small>")
|
|
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()
|