commit 7bd3f755bef18a78c9dc0f3008bff5f5f6ae4878 Author: grandir66 Date: Thu Feb 5 22:09:40 2026 +0100 Initial commit: UnifiMultimirror port mirroring tool - TUI application (Textual) for interactive port selection - CLI application (Rich) as alternative interface - SSH manager for device connection - Config parser for running-config, interface status, port-channel - Script generator for Bash, Python, and command files - Support for multiple source ports and LAG interfaces Co-authored-by: Cursor diff --git a/.gitignore b/.gitignore new file mode 100644 index 0000000..9c6dd36 --- /dev/null +++ b/.gitignore @@ -0,0 +1,48 @@ +# Python +__pycache__/ +*.py[cod] +*$py.class +*.so +.Python +build/ +develop-eggs/ +dist/ +downloads/ +eggs/ +.eggs/ +lib/ +lib64/ +parts/ +sdist/ +var/ +wheels/ +*.egg-info/ +.installed.cfg +*.egg + +# Virtual environments +venv/ +ENV/ +env/ +ssh_env/ + +# IDE +.idea/ +.vscode/ +*.swp +*.swo +*~ + +# OS +.DS_Store +Thumbs.db + +# Project specific +.ssh_key +config.json +old_config.json +generated/ +!generated/.gitkeep + +# Logs +*.log diff --git a/README.md b/README.md new file mode 100644 index 0000000..1a0d84a --- /dev/null +++ b/README.md @@ -0,0 +1,154 @@ +# UnifiMultimirror + +Tool per la gestione del **port mirroring multi-porta** su switch Ubiquiti UniFi/EdgeSwitch. + +Il controller UniFi supporta nativamente solo una singola porta in mirroring. Questo tool permette di configurare **multiple porte source** via CLI SSH, superando questa limitazione. + +## Caratteristiche + +- **Interfaccia TUI** (Text User Interface) con navigazione da tastiera +- **Visualizzazione porte** con stato link, VLAN (PVID e Tagged), ruolo mirroring +- **Configurazione interattiva** con selezione checkbox delle porte source +- **Generazione script** riutilizzabili (Bash, Python, comandi puri) +- **Applicazione diretta** della configurazione via SSH +- Supporto per **porte fisiche** e **LAG** (Link Aggregation Groups) + +## Screenshot + +``` +┌──────────────────────────────────────────────────────────────────────────────┐ +│ │ Porta │ Descrizione │ Stato │ PVID │ Tagged │ Ruolo│ +├─────┼───────┼──────────────────────────┼───────┼──────┼───────────────┼──────┤ +│ ◀◀ │ 0/3 │ SFP_ 3 - MONITORING PORT │ UP │ 1 │ 2,4-7,15-16.. │ DEST │ +│ [X] │ 0/4 │ SFP_ 4 - SW CED 237 │ UP │ 1 │ 2,4-7,15-16.. │ SRC │ +│ [ ] │ 0/5 │ SFP_ 5 - SW 239 CLIENT │ UP │ 1 │ 2,4-7,15-16.. │ - │ +│ [X] │ 0/8 │ SFP_ 8 - SW CED 245 AP │ UP │ 1 │ 2,4-7,15-16.. │ SRC │ +└─────┴───────┴──────────────────────────┴───────┴──────┴───────────────┴──────┘ +``` + +## Requisiti + +- Python 3.8+ +- Switch Ubiquiti UniFi o EdgeSwitch con accesso SSH abilitato +- Credenziali admin + +## Installazione + +```bash +# Clona il repository +git clone https://github.com/tuousername/UnifiMultimirror.git +cd UnifiMultimirror + +# Installa le dipendenze +pip install -r requirements.txt +``` + +### Dipendenze + +- `paramiko` - Connessione SSH +- `cryptography` - Crittografia password +- `rich` - Formattazione output CLI +- `textual` - Interfaccia TUI + +## Utilizzo + +### Interfaccia TUI (consigliata) + +```bash +python tui_app.py +``` + +**Navigazione:** +- `↑/↓` - Muovi cursore +- `Spazio/Enter` - Seleziona/deseleziona porta source +- `D` - Imposta porta come destination +- `A` - Seleziona tutte le porte +- `N` - Deseleziona tutte +- `F10` - Conferma configurazione +- `ESC` - Annulla + +**Schermata principale:** +- `R` - Ricarica configurazione dal device +- `C` - Configura mirroring +- `G` - Genera script +- `F5` - Applica configurazione +- `Q` - Esci + +### Interfaccia CLI classica + +```bash +python main.py +``` + +Menu testuale con opzioni numerate. + +## Script Generati + +Quando selezioni "Genera Script", vengono creati 3 file nella cartella `generated/`: + +| File | Descrizione | +|------|-------------| +| `mirror_config_*.sh` | Script Bash (richiede `sshpass`) | +| `mirror_config_*.py` | Script Python standalone | +| `mirror_commands_*.txt` | Comandi puri per uso manuale | + +### Esempio script Bash + +```bash +./generated/mirror_config_20260205_143000.sh +# oppure con password come argomento +./generated/mirror_config_20260205_143000.sh "mypassword" +``` + +## Comandi CLI EdgeSwitch + +Il tool genera automaticamente i seguenti comandi: + +``` +cli +enable +configure +no monitor session 1 # Rimuove sessione esistente +monitor session 1 destination interface 0/3 # Porta monitor +monitor session 1 source interface 0/4 # Porta da monitorare +monitor session 1 source interface 0/8 # Altra porta source +... +monitor session 1 mode # Attiva sessione +exit +``` + +## Note Importanti + +- Le modifiche via CLI sono **temporanee** e potrebbero essere sovrascritte dal controller UniFi al prossimo provisioning +- Usa gli script generati per **riapplicare** la configurazione dopo un reboot o provisioning +- L'utente SSH deve essere `admin` + +## Struttura Progetto + +``` +UnifiMultimirror/ +├── tui_app.py # Interfaccia TUI (Textual) +├── main.py # Interfaccia CLI classica +├── ssh_manager.py # Gestione connessione SSH +├── config_parser.py # Parser configurazione switch +├── ui.py # UI per CLI classica (Rich) +├── script_generator.py # Generatore script +├── requirements.txt # Dipendenze Python +├── generated/ # Script generati +│ └── .gitkeep +└── README.md +``` + +## Compatibilità + +Testato su: +- USW-Pro-Aggregation (firmware 7.1.26) +- Altri switch UniFi/EdgeSwitch con CLI simile + +## Licenza + +MIT License + +## Autore + +Progetto sviluppato per semplificare la gestione del port mirroring multi-porta su reti Ubiquiti. diff --git a/config_parser.py b/config_parser.py new file mode 100644 index 0000000..d8cd50d --- /dev/null +++ b/config_parser.py @@ -0,0 +1,525 @@ +#!/usr/bin/env python3 +""" +Configuration Parser for Ubiquiti EdgeSwitch/UniFi switches. +Parses running-config, interface status, and port-channel outputs. +""" + +import re +from dataclasses import dataclass, field +from typing import Dict, List, Optional, Set, Tuple + + +@dataclass +class InterfaceConfig: + """Configuration for a single interface.""" + port_id: str + description: str = "" + link_state: str = "Unknown" # Up, Down, Unknown + speed: str = "" # 10G Full, 25G Full, etc. + media_type: str = "" # DAC, 10GBase-SR, etc. + pvid: int = 1 # Native/untagged VLAN + tagged_vlans: List[int] = field(default_factory=list) + untagged_vlans: List[int] = field(default_factory=list) + is_shutdown: bool = False + lag_member_of: str = "" # LAG ID if member (3/1, 3/2, etc.) + + +@dataclass +class LAGConfig: + """Configuration for a Link Aggregation Group.""" + lag_id: str # 3/1, 3/2, etc. + name: str = "" # ch1, ch2, etc. + link_state: str = "Down" # Up, Down + lag_type: str = "Dynamic" # Dynamic, Static + member_ports: List[str] = field(default_factory=list) + active_ports: List[str] = field(default_factory=list) + + +@dataclass +class MonitorSession: + """Port mirroring session configuration.""" + session_id: int + destination: str = "" + sources: List[str] = field(default_factory=list) + mode: str = "" + + +@dataclass +class SystemInfo: + """System information from the switch.""" + hostname: str = "" + description: str = "" + software_version: str = "" + uptime: str = "" + ip_address: str = "" + netmask: str = "" + gateway: str = "" + mgmt_vlan: int = 1 + + +@dataclass +class SwitchConfig: + """Complete switch configuration.""" + system_info: SystemInfo = field(default_factory=SystemInfo) + interfaces: Dict[str, InterfaceConfig] = field(default_factory=dict) + lags: Dict[str, LAGConfig] = field(default_factory=dict) + monitor_sessions: Dict[int, MonitorSession] = field(default_factory=dict) + vlans: List[int] = field(default_factory=list) + + +class ConfigParser: + """Parser for Ubiquiti switch configuration outputs.""" + + def __init__(self): + self.config = SwitchConfig() + + # ========================================================================= + # VLAN list parsing utilities + # ========================================================================= + + @staticmethod + def parse_vlan_list(vlan_str: str) -> List[int]: + """ + Parse VLAN list string like "2,4-7,15-16,20,40" into list of integers. + """ + vlans = [] + if not vlan_str: + return vlans + + parts = vlan_str.replace(" ", "").split(",") + for part in parts: + if "-" in part: + try: + start, end = part.split("-", 1) + vlans.extend(range(int(start), int(end) + 1)) + except ValueError: + continue + else: + try: + vlans.append(int(part)) + except ValueError: + continue + + return sorted(set(vlans)) + + @staticmethod + def format_vlan_list(vlans: List[int]) -> str: + """ + Format list of VLAN IDs into compact string like "2,4-7,15-16,20". + """ + if not vlans: + return "" + + vlans = sorted(set(vlans)) + result = [] + start = vlans[0] + end = vlans[0] + + for vlan in vlans[1:]: + if vlan == end + 1: + end = vlan + else: + if start == end: + result.append(str(start)) + else: + result.append(f"{start}-{end}") + start = end = vlan + + # Add last range + if start == end: + result.append(str(start)) + else: + result.append(f"{start}-{end}") + + return ",".join(result) + + # ========================================================================= + # Running config parser + # ========================================================================= + + def parse_running_config(self, config_text: str) -> None: + """Parse running-config output.""" + lines = config_text.split("\n") + + current_interface: Optional[str] = None + current_iface_config: Optional[InterfaceConfig] = None + + for line in lines: + line = line.rstrip() + stripped = line.strip() + + # Skip empty lines and prompts + if not stripped or stripped.startswith("(UBNT)"): + continue + + # System info from comments + if stripped.startswith("!System Description"): + match = re.search(r'"([^"]+)"', stripped) + if match: + self.config.system_info.description = match.group(1) + + elif stripped.startswith("!System Software Version"): + match = re.search(r'"([^"]+)"', stripped) + if match: + self.config.system_info.software_version = match.group(1) + + elif stripped.startswith("!System Up Time"): + match = re.search(r'"([^"]+)"', stripped) + if match: + self.config.system_info.uptime = match.group(1) + + # Network parameters + elif stripped.startswith("network parms"): + parts = stripped.split() + if len(parts) >= 4: + self.config.system_info.ip_address = parts[2] + self.config.system_info.netmask = parts[3] + if len(parts) >= 5: + self.config.system_info.gateway = parts[4] + + # Management VLAN + elif stripped.startswith("network mgmt_vlan"): + parts = stripped.split() + if len(parts) >= 3: + try: + self.config.system_info.mgmt_vlan = int(parts[2]) + except ValueError: + pass + + # Global VLAN database (only when NOT inside an interface block) + elif stripped.startswith("vlan ") and "database" not in stripped and current_interface is None: + # Global VLAN definition like "vlan 2,4-7,15-16,20,40,88,133,666" + match = re.match(r"vlan\s+([\d,\-]+)$", stripped) + if match: + self.config.vlans = self.parse_vlan_list(match.group(1)) + + # Hostname + elif stripped.startswith("snmp-server sysname"): + match = re.search(r'"([^"]+)"', stripped) + if match: + self.config.system_info.hostname = match.group(1) + + # Interface start + elif stripped.startswith("interface "): + # Save previous interface + if current_interface and current_iface_config: + self.config.interfaces[current_interface] = current_iface_config + + match = re.match(r"interface\s+(0/\d+|lag\s+\d+|3/\d+)", stripped) + if match: + current_interface = match.group(1).replace(" ", "") + current_iface_config = InterfaceConfig(port_id=current_interface) + else: + current_interface = None + current_iface_config = None + + # Interface configuration lines + elif current_iface_config: + if stripped.startswith("description"): + match = re.search(r"'([^']*)'", stripped) + if match: + current_iface_config.description = match.group(1) + + elif stripped.startswith("vlan pvid"): + parts = stripped.split() + if len(parts) >= 3: + try: + current_iface_config.pvid = int(parts[2]) + except ValueError: + pass + + elif stripped.startswith("vlan participation include"): + # vlan participation include shows VLANs the port participates in + # We store this separately - tagged_vlans comes from "vlan tagging" + vlan_str = stripped.replace("vlan participation include", "").strip() + vlans = self.parse_vlan_list(vlan_str) + # Store as participation VLANs (untagged + tagged combined) + current_iface_config.untagged_vlans = vlans + + elif stripped.startswith("vlan tagging"): + # vlan tagging specifies which VLANs are tagged (802.1Q) + # This is the authoritative source for tagged VLANs + vlan_str = stripped.replace("vlan tagging", "").strip() + current_iface_config.tagged_vlans = self.parse_vlan_list(vlan_str) + + elif stripped == "shutdown": + current_iface_config.is_shutdown = True + + elif stripped.startswith("addport"): + match = re.search(r"addport\s+(3/\d+)", stripped) + if match: + current_iface_config.lag_member_of = match.group(1) + + elif stripped == "exit": + # Save interface config + if current_interface and current_iface_config: + self.config.interfaces[current_interface] = current_iface_config + current_interface = None + current_iface_config = None + + # Monitor session + elif stripped.startswith("monitor session"): + match = re.match( + r"monitor session (\d+) (destination|source) interface (.+)", + stripped + ) + if match: + session_id = int(match.group(1)) + role = match.group(2) + interface = match.group(3).strip() + + if session_id not in self.config.monitor_sessions: + self.config.monitor_sessions[session_id] = MonitorSession( + session_id=session_id + ) + + session = self.config.monitor_sessions[session_id] + if role == "destination": + session.destination = interface + else: + session.sources.append(interface) + + elif "mode" in stripped: + # monitor session 1 mode + match = re.match(r"monitor session (\d+) mode", stripped) + if match: + session_id = int(match.group(1)) + if session_id in self.config.monitor_sessions: + self.config.monitor_sessions[session_id].mode = "enabled" + + # Save last interface if any + if current_interface and current_iface_config: + self.config.interfaces[current_interface] = current_iface_config + + # ========================================================================= + # Interface status parser + # ========================================================================= + + def parse_interfaces_status(self, status_text: str) -> None: + """Parse 'show interfaces status all' output.""" + lines = status_text.split("\n") + + # Skip header lines + in_data = False + + for line in lines: + # Detect start of data after separator line + if line.startswith("---------"): + in_data = True + continue + + if not in_data: + continue + + # Parse port line + # Format: Port Name State Mode Status Type Flow + # Example: 0/3 SFP_ 3 - MONITORING PORT Up Auto D 10G Full DAC Inactive + + if not line.strip(): + continue + + # Port ID is the first field + match = re.match(r"^(\d+/\d+)\s+", line) + if not match: + continue + + port_id = match.group(1) + rest = line[match.end():] + + # Parse fixed-width columns (approximate positions) + # Name: ~28 chars, State: ~8 chars, Mode: ~12 chars, Status: ~12 chars, Type: ~20 chars + + # Find Up or Down to locate state column + state_match = re.search(r"\s+(Up|Down)\s+", rest) + if state_match: + name = rest[:state_match.start()].strip() + state = state_match.group(1) + after_state = rest[state_match.end():] + + # Parse remaining fields + parts = after_state.split() + + # Mode + Status + Type + Flow + mode = "" + status = "" + media_type = "" + + if parts: + # Mode (Auto D, 10G Full, 25G Full, etc.) + if parts[0] in ("Auto", "10G", "25G", "1G", "100M"): + mode = parts[0] + if len(parts) > 1 and parts[1] in ("D", "Full", "Half"): + mode += " " + parts[1] + parts = parts[2:] + else: + parts = parts[1:] + + # Status (actual speed) + if parts and parts[0] in ("10G", "25G", "1G", "100M"): + status = parts[0] + if len(parts) > 1 and parts[1] == "Full": + status += " Full" + parts = parts[2:] + else: + parts = parts[1:] + + # Media type + if parts: + media_type = parts[0] + + # Update interface config + if port_id in self.config.interfaces: + iface = self.config.interfaces[port_id] + iface.link_state = state + iface.speed = status if status else mode + iface.media_type = media_type + if name and not iface.description: + iface.description = name + else: + # Create interface if not exists + self.config.interfaces[port_id] = InterfaceConfig( + port_id=port_id, + description=name, + link_state=state, + speed=status if status else mode, + media_type=media_type + ) + + # ========================================================================= + # Port-channel parser + # ========================================================================= + + def parse_port_channel_brief(self, pc_text: str) -> None: + """Parse 'show port-channel brief' output.""" + lines = pc_text.split("\n") + + in_data = False + + for line in lines: + # Detect start of data after separator line + if line.startswith("---------"): + in_data = True + continue + + if not in_data: + continue + + if not line.strip(): + continue + + # Format: LogicalIf Name Min State Trap Type MbrPorts ActivePorts + # Example: 3/1 ch1 1 Up Disabled Dynamic 0/31,0/32 0/31,0/32 + + match = re.match( + r"^(\d+/\d+)\s+(\S+)\s+(\d+)\s+(Up|Down)\s+(\S+)\s+(Dynamic|Static)\s*(.*)$", + line + ) + + if match: + lag_id = match.group(1) + name = match.group(2) + state = match.group(4) + lag_type = match.group(6) + ports_str = match.group(7).strip() + + # Parse member and active ports + member_ports = [] + active_ports = [] + + if ports_str: + parts = ports_str.split() + if len(parts) >= 1: + member_ports = [p.strip() for p in parts[0].split(",") if p.strip()] + if len(parts) >= 2: + active_ports = [p.strip() for p in parts[1].split(",") if p.strip()] + + self.config.lags[lag_id] = LAGConfig( + lag_id=lag_id, + name=name, + link_state=state, + lag_type=lag_type, + member_ports=member_ports, + active_ports=active_ports + ) + + # ========================================================================= + # Convenience methods + # ========================================================================= + + def get_physical_ports(self) -> List[str]: + """Get list of physical port IDs (0/x format).""" + return sorted( + [p for p in self.config.interfaces.keys() if p.startswith("0/")], + key=lambda x: int(x.split("/")[1]) + ) + + def get_lag_ports(self) -> List[str]: + """Get list of LAG port IDs (3/x format).""" + return sorted( + [p for p in self.config.lags.keys()], + key=lambda x: int(x.split("/")[1]) + ) + + def get_mirroring_destination(self, session_id: int = 1) -> Optional[str]: + """Get the destination port for a mirroring session.""" + if session_id in self.config.monitor_sessions: + return self.config.monitor_sessions[session_id].destination + return None + + def get_mirroring_sources(self, session_id: int = 1) -> List[str]: + """Get list of source ports for a mirroring session.""" + if session_id in self.config.monitor_sessions: + return self.config.monitor_sessions[session_id].sources + return [] + + def is_mirroring_source(self, port_id: str, session_id: int = 1) -> bool: + """Check if port is a mirroring source.""" + return port_id in self.get_mirroring_sources(session_id) + + def is_mirroring_destination(self, port_id: str, session_id: int = 1) -> bool: + """Check if port is the mirroring destination.""" + return port_id == self.get_mirroring_destination(session_id) + + +# ============================================================================= +# Testing +# ============================================================================= + +if __name__ == "__main__": + # Test with sample config + sample_config = """ +!System Description "USW-Pro-Aggregation, 7.1.26.15869, Linux 4.4.153" +!System Software Version "7.1.26.15869" +snmp-server sysname "SW-CED-AGG-243" +network parms 192.168.40.243 255.255.255.0 192.168.40.254 +network mgmt_vlan 40 +vlan database +vlan 2,4-7,15-16,20,40,88,133,666 +exit + +interface 0/3 +description 'SFP_ 3 - MONITORING PORT' +vlan participation include 2,4-7,15-16,20,40,88,133,666 +vlan tagging 2,4-7,15-16,20,40,88,133,666 +exit + +interface 0/4 +description 'SFP_ 4 - SW CED 237' +vlan participation include 2,4-7,15-16,20,40,88,133,666 +vlan tagging 2,4-7,15-16,20,40,88,133,666 +exit + +monitor session 1 destination interface 0/3 +monitor session 1 source interface 0/4 +monitor session 1 source interface 0/8 +monitor session 1 mode +""" + + parser = ConfigParser() + parser.parse_running_config(sample_config) + + print(f"Hostname: {parser.config.system_info.hostname}") + print(f"VLANs: {parser.config.vlans}") + print(f"Interfaces: {list(parser.config.interfaces.keys())}") + print(f"Monitor destination: {parser.get_mirroring_destination()}") + print(f"Monitor sources: {parser.get_mirroring_sources()}") diff --git a/main.py b/main.py new file mode 100644 index 0000000..c6c2bd4 --- /dev/null +++ b/main.py @@ -0,0 +1,466 @@ +#!/usr/bin/env python3 +""" +Ubiquiti Port Mirroring Manager + +A tool to configure port mirroring on Ubiquiti UniFi/EdgeSwitch devices via SSH. +Provides an interactive menu interface for viewing port status, configuring +mirroring, and generating reusable configuration scripts. + +Usage: + python main.py + +Requirements: + pip install paramiko rich cryptography +""" + +import sys +from pathlib import Path +from typing import Optional, List, Tuple + +from rich.console import Console +from rich.panel import Panel +from rich.prompt import Confirm, Prompt +from rich import box + +from ssh_manager import SSHManager, setup_config_interactive +from config_parser import ConfigParser +from ui import PortMirroringUI +from script_generator import ScriptGenerator + + +class PortMirroringManager: + """Main application controller.""" + + def __init__(self): + self.console = Console() + self.ssh_manager = SSHManager() + self.parser = ConfigParser() + self.ui = PortMirroringUI() + self.script_gen = ScriptGenerator() + + self.is_connected = False + self.config_loaded = False + + # Pending changes (not yet applied) + self.pending_destination: Optional[str] = None + self.pending_sources: List[str] = [] + self.has_pending_changes = False + + # ========================================================================= + # Initialization + # ========================================================================= + + def initialize(self) -> bool: + """Initialize the application - always ask for IP and password.""" + self.console.print() + self.console.print(Panel( + "[bold]UBIQUITI PORT MIRRORING MANAGER[/bold]\n\n" + "Gestione port mirroring per switch UniFi/EdgeSwitch", + style="bold white on blue", + box=box.DOUBLE + )) + self.console.print() + + # Always ask for connection details + return self._ask_connection_details() + + def _ask_connection_details(self) -> bool: + """Ask user for IP and password, then connect.""" + from getpass import getpass + + self.console.print("[bold]CONNESSIONE AL DISPOSITIVO[/bold]") + self.console.print() + + # Check if we have saved config + saved_ip = "" + if self.ssh_manager.config_exists(): + success, _ = self.ssh_manager.load_config() + if success: + saved_ip = self.ssh_manager.hostname + + # Ask for IP + if saved_ip: + ip_prompt = f"IP del dispositivo [{saved_ip}]" + ip_input = Prompt.ask(ip_prompt, default=saved_ip) + else: + ip_input = Prompt.ask("IP del dispositivo") + + if not ip_input or not ip_input.strip(): + self.console.print("[red]IP non valido[/red]") + return False + + ip_address = ip_input.strip() + + # Username is always admin + username = "admin" + self.console.print(f"[dim]Username: {username}[/dim]") + + # Ask for password + self.console.print() + password = getpass("Password: ") + + if not password: + self.console.print("[red]Password non valida[/red]") + return False + + # Save config + self.ssh_manager.create_config( + hostname=ip_address, + username=username, + password=password, + port=22 + ) + + # Reload config + success, msg = self.ssh_manager.load_config() + if not success: + self.console.print(f"[red]Errore: {msg}[/red]") + return False + + self.console.print() + + # Connect + return self.connect_to_device() + + def connect_to_device(self) -> bool: + """Connect to the switch and load configuration.""" + self.console.print(f"\n[dim]Connessione a {self.ssh_manager.hostname}...[/dim]") + + success, msg = self.ssh_manager.connect() + if not success: + self.console.print(f"[red]✗ {msg}[/red]") + return False + + self.is_connected = True + self.console.print(f"[green]✓ {msg}[/green]") + + # Load configuration + return self.reload_configuration() + + def reload_configuration(self) -> bool: + """Reload configuration from the switch.""" + if not self.is_connected: + self.console.print("[red]Non connesso al dispositivo[/red]") + return False + + self.console.print("[dim]Recupero configurazione...[/dim]") + + try: + # Get running config + running_config = self.ssh_manager.get_running_config() + self.parser.parse_running_config(running_config) + self.console.print("[green]✓[/green] Running configuration caricata") + + # Get interface status + iface_status = self.ssh_manager.get_interfaces_status() + self.parser.parse_interfaces_status(iface_status) + self.console.print("[green]✓[/green] Stato interfacce caricato") + + # Get port-channel info + pc_brief = self.ssh_manager.get_port_channel_brief() + self.parser.parse_port_channel_brief(pc_brief) + self.console.print("[green]✓[/green] Configurazione LAG caricata") + + # Update UI + self.ui.set_config(self.parser) + self.ui.set_device_info(self.ssh_manager.hostname, self.ssh_manager.username) + + self.config_loaded = True + self.console.print() + + # Show summary + dest = self.parser.get_mirroring_destination() + sources = self.parser.get_mirroring_sources() + + if dest: + self.console.print( + f"[dim]Mirroring attivo: Destination=[bold]{dest}[/bold], " + f"Sources={len(sources)} porte[/dim]" + ) + else: + self.console.print("[dim]Nessun mirroring configurato[/dim]") + + return True + + except Exception as e: + self.console.print(f"[red]Errore durante il caricamento: {e}[/red]") + return False + + def disconnect(self) -> None: + """Disconnect from the switch.""" + if self.is_connected: + self.ssh_manager.disconnect() + self.is_connected = False + self.console.print("[dim]Disconnesso dal dispositivo[/dim]") + + # ========================================================================= + # Main menu handlers + # ========================================================================= + + def run(self) -> None: + """Main application loop.""" + try: + if not self.initialize(): + self.console.print("\n[yellow]Inizializzazione fallita. Uscita.[/yellow]") + return + + while True: + choice = self.ui.show_main_menu() + + if choice == "0": + # Exit + if self.has_pending_changes: + if not Confirm.ask( + "[yellow]Ci sono modifiche non applicate. Uscire comunque?[/yellow]", + default=False + ): + continue + break + + elif choice == "1": + # View port status + self.ui.show_port_status() + + elif choice == "2": + # View mirroring status + self.ui.show_mirroring_status() + + elif choice == "3": + # Configure mirroring + self.handle_configure_mirroring() + + elif choice == "4": + # Generate script + self.handle_generate_script() + + elif choice == "5": + # Apply configuration + self.handle_apply_configuration() + + elif choice == "6": + # Reload configuration + if self.is_connected: + self.reload_configuration() + self.ui.wait_for_enter() + else: + self.console.print("[yellow]Non connesso. Riconnessione...[/yellow]") + if self.connect_to_device(): + self.ui.wait_for_enter() + + except KeyboardInterrupt: + self.console.print("\n\n[yellow]Interrotto dall'utente[/yellow]") + + finally: + self.disconnect() + self.console.print("\n[dim]Arrivederci![/dim]\n") + + # ========================================================================= + # Configure mirroring + # ========================================================================= + + def handle_configure_mirroring(self) -> None: + """Handle mirroring configuration.""" + if not self.config_loaded: + self.console.print("[red]Configurazione non caricata[/red]") + self.ui.wait_for_enter() + return + + result = self.ui.configure_mirroring() + + if result: + destination, sources = result + self.pending_destination = destination + self.pending_sources = sources + self.has_pending_changes = True + + self.ui.clear_screen() + self.ui.print_header("CONFIGURAZIONE COMPLETATA") + + # Show commands + commands = self.ui.generate_mirroring_commands(destination, sources) + self.ui.show_command_preview(destination, sources) + + self.console.print() + self.console.print(Panel( + "[bold green]Configurazione salvata in memoria[/bold green]\n\n" + "Usa le opzioni del menu per:\n" + " [4] Genera Script - Crea script riutilizzabile\n" + " [5] Applica Configurazione - Applica ora al device", + box=box.ROUNDED + )) + + self.ui.wait_for_enter() + + # ========================================================================= + # Generate script + # ========================================================================= + + def handle_generate_script(self) -> None: + """Handle script generation.""" + self.ui.clear_screen() + self.ui.print_header("GENERA SCRIPT DI CONFIGURAZIONE") + + # Check if we have pending changes or current config + if self.has_pending_changes: + destination = self.pending_destination + sources = self.pending_sources + self.console.print("[cyan]Usando configurazione in memoria (modifiche pending)[/cyan]\n") + elif self.config_loaded: + destination = self.parser.get_mirroring_destination() + sources = self.parser.get_mirroring_sources() + if not destination: + self.console.print("[yellow]Nessun mirroring configurato sul device.[/yellow]") + self.console.print("Usa prima [3] Configura Port Mirroring.\n") + self.ui.wait_for_enter() + return + self.console.print("[cyan]Usando configurazione attuale del device[/cyan]\n") + else: + self.console.print("[red]Nessuna configurazione disponibile[/red]") + self.ui.wait_for_enter() + return + + # Generate commands + commands = self.ui.generate_mirroring_commands(destination, sources) + + # Show preview + self.ui.show_command_preview(destination, sources) + self.console.print() + + if not Confirm.ask("Generare gli script?", default=True): + return + + # Generate all formats + try: + description = f"Mirroring: {destination} <- {len(sources)} sources" + + paths = self.script_gen.generate_all( + commands=commands, + device_ip=self.ssh_manager.hostname, + username=self.ssh_manager.username, + port=self.ssh_manager.port, + config=self.parser.config if self.config_loaded else None, + description=description + ) + + self.console.print() + self.console.print(Panel( + "[bold green]Script generati con successo![/bold green]\n\n" + f"[bold]Bash script:[/bold]\n {paths['bash']}\n\n" + f"[bold]Python script:[/bold]\n {paths['python']}\n\n" + f"[bold]File comandi:[/bold]\n {paths['commands']}", + title="FILE GENERATI", + box=box.ROUNDED + )) + + self.console.print() + self.console.print("[dim]Per eseguire lo script bash:[/dim]") + self.console.print(f" [bold]chmod +x {paths['bash'].name}[/bold]") + self.console.print(f" [bold]./{paths['bash'].name}[/bold]") + + except Exception as e: + self.console.print(f"[red]Errore durante la generazione: {e}[/red]") + + self.ui.wait_for_enter() + + # ========================================================================= + # Apply configuration + # ========================================================================= + + def handle_apply_configuration(self) -> None: + """Handle applying configuration to device.""" + self.ui.clear_screen() + self.ui.print_header("APPLICA CONFIGURAZIONE") + + if not self.is_connected: + self.console.print("[red]Non connesso al dispositivo[/red]") + if Confirm.ask("Vuoi riconnetterti?", default=True): + if not self.connect_to_device(): + self.ui.wait_for_enter() + return + else: + self.ui.wait_for_enter() + return + + # Check if we have pending changes + if not self.has_pending_changes: + self.console.print("[yellow]Nessuna modifica in sospeso.[/yellow]") + self.console.print("Usa prima [3] Configura Port Mirroring.\n") + self.ui.wait_for_enter() + return + + destination = self.pending_destination + sources = self.pending_sources + + # Generate and show commands + commands = self.ui.generate_mirroring_commands(destination, sources) + + self.console.print("[bold]Configurazione da applicare:[/bold]\n") + self.ui.show_command_preview(destination, sources) + + self.console.print() + self.console.print(Panel( + "[bold yellow]ATTENZIONE[/bold yellow]\n\n" + "Questa operazione modificherà la configurazione del port mirroring.\n" + "Le modifiche saranno attive immediatamente ma potrebbero essere\n" + "sovrascritte dal controller UniFi al prossimo provisioning.", + box=box.ROUNDED, + style="yellow" + )) + + self.console.print() + if not Confirm.ask("[bold]Applicare la configurazione?[/bold]", default=False): + self.console.print("[dim]Operazione annullata[/dim]") + self.ui.wait_for_enter() + return + + # Apply configuration + self.console.print() + self.console.print("[dim]Applicazione in corso...[/dim]") + + try: + success, output = self.ssh_manager.apply_mirroring_config(commands) + + if success: + self.console.print() + self.console.print(Panel( + "[bold green]Configurazione applicata con successo![/bold green]", + box=box.DOUBLE, + style="green" + )) + + # Clear pending changes + self.has_pending_changes = False + self.pending_destination = None + self.pending_sources = [] + + # Offer to reload + self.console.print() + if Confirm.ask("Ricaricare la configurazione per verificare?", default=True): + self.reload_configuration() + + else: + self.console.print() + self.console.print(Panel( + f"[bold red]Errore durante l'applicazione[/bold red]\n\n{output}", + box=box.ROUNDED, + style="red" + )) + + except Exception as e: + self.console.print(f"[red]Errore: {e}[/red]") + + self.ui.wait_for_enter() + + +# ============================================================================= +# Entry point +# ============================================================================= + +def main(): + """Application entry point.""" + app = PortMirroringManager() + app.run() + + +if __name__ == "__main__": + main() diff --git a/requirements.txt b/requirements.txt new file mode 100644 index 0000000..a6bd5d7 --- /dev/null +++ b/requirements.txt @@ -0,0 +1,4 @@ +paramiko>=3.0.0 +cryptography>=41.0.0 +rich>=13.0.0 +textual>=0.47.0 diff --git a/script_generator.py b/script_generator.py new file mode 100644 index 0000000..acc4de7 --- /dev/null +++ b/script_generator.py @@ -0,0 +1,497 @@ +#!/usr/bin/env python3 +""" +Script Generator for Ubiquiti Port Mirroring configuration. +Generates reusable bash scripts and command files. +""" + +import os +from datetime import datetime +from pathlib import Path +from typing import List, Optional + +from config_parser import SwitchConfig + + +class ScriptGenerator: + """Generates reusable configuration scripts.""" + + def __init__(self, output_dir: Optional[Path] = None): + self.script_dir = Path(__file__).parent + self.output_dir = output_dir or self.script_dir / "generated" + + # Create output directory if needed + self.output_dir.mkdir(parents=True, exist_ok=True) + + def generate_timestamp(self) -> str: + """Generate timestamp string for filenames.""" + return datetime.now().strftime("%Y%m%d_%H%M%S") + + def generate_human_timestamp(self) -> str: + """Generate human-readable timestamp for comments.""" + return datetime.now().strftime("%Y-%m-%d %H:%M:%S") + + # ========================================================================= + # Bash script generation + # ========================================================================= + + def generate_bash_script( + self, + commands: List[str], + device_ip: str, + username: str = "admin", + port: int = 22, + config: Optional[SwitchConfig] = None, + description: str = "" + ) -> Path: + """ + Generate a bash script to apply mirroring configuration. + + Args: + commands: List of configuration commands (without cli/enable/config prefixes) + device_ip: Switch IP address + username: SSH username + port: SSH port + config: Optional switch config for additional info + description: Optional description of what this script does + + Returns: + Path to generated script + """ + timestamp = self.generate_timestamp() + human_ts = self.generate_human_timestamp() + + filename = f"mirror_config_{timestamp}.sh" + filepath = self.output_dir / filename + + # Build device info + device_info = device_ip + if config and config.system_info.hostname: + device_info = f"{config.system_info.hostname} ({device_ip})" + + model_info = "" + if config and config.system_info.description: + model_info = config.system_info.description.split(",")[0] + + # Build commands block + cli_commands = [ + "cli", + "enable", + "configure", + *commands, + "exit", + "exit", + "exit" + ] + commands_block = "\n".join(cli_commands) + + # Build script content + script_content = f'''#!/bin/bash +# ============================================================ +# Ubiquiti Port Mirroring Configuration Script +# ============================================================ +# Generated: {human_ts} +# Device: {device_info} +{f"# Model: {model_info}" if model_info else ""} +{f"# Description: {description}" if description else ""} +# ============================================================ +# +# This script applies port mirroring configuration via SSH. +# +# REQUIREMENTS: +# - sshpass installed (brew install sshpass / apt install sshpass) +# - SSH access to the device +# +# USAGE: +# ./mirror_config_{timestamp}.sh # Will prompt for password +# ./mirror_config_{timestamp}.sh "password" # Password as argument +# +# ============================================================ + +set -e + +# Device configuration +DEVICE_IP="{device_ip}" +USERNAME="{username}" +SSH_PORT={port} + +# Colors for output +RED='\\033[0;31m' +GREEN='\\033[0;32m' +YELLOW='\\033[1;33m' +NC='\\033[0m' # No Color + +echo -e "${{YELLOW}}============================================================${{NC}}" +echo -e "${{YELLOW}} Ubiquiti Port Mirroring Configuration${{NC}}" +echo -e "${{YELLOW}}============================================================${{NC}}" +echo "" +echo "Device: $DEVICE_IP" +echo "User: $USERNAME" +echo "" + +# Get password +if [ -z "$1" ]; then + read -sp "Enter SSH password for $USERNAME@$DEVICE_IP: " PASSWORD + echo "" +else + PASSWORD="$1" +fi + +if [ -z "$PASSWORD" ]; then + echo -e "${{RED}}Error: Password is required${{NC}}" + exit 1 +fi + +# Check if sshpass is available +if ! command -v sshpass &> /dev/null; then + echo -e "${{RED}}Error: sshpass is not installed${{NC}}" + echo "Install with: brew install sshpass (macOS) or apt install sshpass (Linux)" + exit 1 +fi + +# Commands to execute +read -r -d '' COMMANDS << 'CMDEOF' || true +{commands_block} +CMDEOF + +echo -e "${{YELLOW}}Connecting to $DEVICE_IP...${{NC}}" +echo "" + +# Execute commands via SSH +if sshpass -p "$PASSWORD" ssh -o StrictHostKeyChecking=no -o UserKnownHostsFile=/dev/null \\ + -p $SSH_PORT $USERNAME@$DEVICE_IP "$COMMANDS" 2>/dev/null; then + echo "" + echo -e "${{GREEN}}============================================================${{NC}}" + echo -e "${{GREEN}} Configuration applied successfully!${{NC}}" + echo -e "${{GREEN}}============================================================${{NC}}" +else + echo "" + echo -e "${{RED}}============================================================${{NC}}" + echo -e "${{RED}} Error applying configuration${{NC}}" + echo -e "${{RED}}============================================================${{NC}}" + exit 1 +fi +''' + + # Write script + with open(filepath, 'w', encoding='utf-8') as f: + f.write(script_content) + + # Make executable + os.chmod(filepath, 0o755) + + return filepath + + # ========================================================================= + # Command file generation + # ========================================================================= + + def generate_command_file( + self, + commands: List[str], + device_ip: str, + config: Optional[SwitchConfig] = None, + description: str = "", + include_cli_prefix: bool = True + ) -> Path: + """ + Generate a text file with commands only. + + Args: + commands: List of configuration commands + device_ip: Switch IP address + config: Optional switch config for additional info + description: Optional description + include_cli_prefix: Include cli/enable/config prefix commands + + Returns: + Path to generated file + """ + timestamp = self.generate_timestamp() + human_ts = self.generate_human_timestamp() + + filename = f"mirror_commands_{timestamp}.txt" + filepath = self.output_dir / filename + + # Build device info + device_info = device_ip + if config and config.system_info.hostname: + device_info = f"{config.system_info.hostname} ({device_ip})" + + # Build content + lines = [ + "# ============================================================", + "# Ubiquiti Port Mirroring Commands", + "# ============================================================", + f"# Generated: {human_ts}", + f"# Device: {device_info}", + ] + + if description: + lines.append(f"# Description: {description}") + + lines.extend([ + "# ============================================================", + "#", + "# To apply these commands manually:", + "# 1. SSH to the device", + "# 2. Enter: cli", + "# 3. Enter: enable", + "# 4. Enter: configure", + "# 5. Execute the commands below", + "# 6. Enter: exit (3 times to logout)", + "#", + "# ============================================================", + "", + ]) + + if include_cli_prefix: + lines.extend([ + "# === CLI Access Commands ===", + "cli", + "enable", + "configure", + "", + "# === Mirroring Configuration ===", + ]) + + lines.extend(commands) + + if include_cli_prefix: + lines.extend([ + "", + "# === Exit Commands ===", + "exit", + "exit", + "exit", + ]) + + # Write file + with open(filepath, 'w', encoding='utf-8') as f: + f.write("\n".join(lines) + "\n") + + return filepath + + # ========================================================================= + # Python script generation (alternative to bash) + # ========================================================================= + + def generate_python_script( + self, + commands: List[str], + device_ip: str, + username: str = "admin", + port: int = 22, + config: Optional[SwitchConfig] = None, + description: str = "" + ) -> Path: + """ + Generate a Python script to apply mirroring configuration. + Uses paramiko, doesn't require sshpass. + + Returns: + Path to generated script + """ + timestamp = self.generate_timestamp() + human_ts = self.generate_human_timestamp() + + filename = f"mirror_config_{timestamp}.py" + filepath = self.output_dir / filename + + # Build device info + device_info = device_ip + if config and config.system_info.hostname: + device_info = f"{config.system_info.hostname} ({device_ip})" + + # Format commands as Python list + commands_str = ",\n ".join([f'"{cmd}"' for cmd in commands]) + + script_content = f'''#!/usr/bin/env python3 +""" +Ubiquiti Port Mirroring Configuration Script + +Generated: {human_ts} +Device: {device_info} +{f"Description: {description}" if description else ""} + +Requirements: pip install paramiko +Usage: python mirror_config_{timestamp}.py +""" + +import paramiko +import time +import sys +from getpass import getpass + + +# Configuration +DEVICE_IP = "{device_ip}" +USERNAME = "{username}" +SSH_PORT = {port} + +# Commands to execute +COMMANDS = [ + "cli", + "enable", + "configure", + {commands_str}, + "exit", + "exit", + "exit", +] + + +def apply_config(password: str) -> bool: + """Apply configuration to device.""" + try: + print(f"Connecting to {{DEVICE_IP}}...") + + ssh = paramiko.SSHClient() + ssh.set_missing_host_key_policy(paramiko.AutoAddPolicy()) + ssh.connect( + hostname=DEVICE_IP, + port=SSH_PORT, + username=USERNAME, + password=password, + timeout=10 + ) + + shell = ssh.invoke_shell() + time.sleep(2) + + # Clear buffer + if shell.recv_ready(): + shell.recv(65536) + + print("Connected. Applying configuration...") + print() + + for cmd in COMMANDS: + print(f" >>> {{cmd}}") + shell.send(cmd + "\\n") + time.sleep(1) + + if shell.recv_ready(): + response = shell.recv(4096).decode('utf-8', errors='ignore') + for line in response.strip().split("\\n")[:2]: + if line.strip() and not line.strip().startswith("("): + print(f" {{line.strip()}}") + + shell.close() + ssh.close() + + print() + print("=" * 50) + print(" Configuration applied successfully!") + print("=" * 50) + return True + + except paramiko.AuthenticationException: + print("ERROR: Authentication failed") + return False + except Exception as e: + print(f"ERROR: {{e}}") + return False + + +def main(): + print("=" * 50) + print(" Ubiquiti Port Mirroring Configuration") + print("=" * 50) + print() + print(f"Device: {{DEVICE_IP}}") + print(f"User: {{USERNAME}}") + print() + + if len(sys.argv) > 1: + password = sys.argv[1] + else: + password = getpass(f"Enter SSH password for {{USERNAME}}@{{DEVICE_IP}}: ") + + if not password: + print("Error: Password is required") + sys.exit(1) + + success = apply_config(password) + sys.exit(0 if success else 1) + + +if __name__ == "__main__": + main() +''' + + # Write script + with open(filepath, 'w', encoding='utf-8') as f: + f.write(script_content) + + # Make executable + os.chmod(filepath, 0o755) + + return filepath + + # ========================================================================= + # Generate all formats + # ========================================================================= + + def generate_all( + self, + commands: List[str], + device_ip: str, + username: str = "admin", + port: int = 22, + config: Optional[SwitchConfig] = None, + description: str = "" + ) -> dict: + """ + Generate all script formats. + + Returns: + Dict with paths: {'bash': Path, 'python': Path, 'commands': Path} + """ + bash_path = self.generate_bash_script( + commands, device_ip, username, port, config, description + ) + + python_path = self.generate_python_script( + commands, device_ip, username, port, config, description + ) + + cmd_path = self.generate_command_file( + commands, device_ip, config, description + ) + + return { + 'bash': bash_path, + 'python': python_path, + 'commands': cmd_path + } + + +# ============================================================================= +# Testing +# ============================================================================= + +if __name__ == "__main__": + generator = ScriptGenerator() + + # Test commands + commands = [ + "no monitor session 1", + "monitor session 1 destination interface 0/3", + "monitor session 1 source interface 0/4", + "monitor session 1 source interface 0/8", + "monitor session 1 source interface 0/17", + "monitor session 1 source interface 0/18", + "monitor session 1 mode", + ] + + # Generate all + paths = generator.generate_all( + commands=commands, + device_ip="192.168.40.243", + username="admin", + description="Port mirroring for network monitoring" + ) + + print("Generated files:") + for fmt, path in paths.items(): + print(f" {fmt}: {path}") diff --git a/ssh_manager.py b/ssh_manager.py new file mode 100644 index 0000000..d3283ca --- /dev/null +++ b/ssh_manager.py @@ -0,0 +1,399 @@ +#!/usr/bin/env python3 +""" +SSH Manager for Ubiquiti EdgeSwitch/UniFi switches. +Handles connection, command execution, and output capture. +""" + +import paramiko +import time +import os +import stat +import json +import base64 +from pathlib import Path +from typing import Optional, List, Tuple +from cryptography.fernet import Fernet +from getpass import getpass + + +class SSHManager: + """Manages SSH connections to Ubiquiti switches.""" + + def __init__(self, config_path: Optional[Path] = None): + self.script_dir = Path(__file__).parent + self.config_file = config_path or self.script_dir / "config.json" + self.key_file = self.script_dir / ".ssh_key" + + self.ssh_client: Optional[paramiko.SSHClient] = None + self.shell = None + self.hostname: str = "" + self.username: str = "" + self.port: int = 22 + self.timeout: int = 10 + + # Settings + self.command_delay: float = 1.0 + self.response_timeout: float = 3.0 + + # ========================================================================= + # Encryption methods + # ========================================================================= + + def _generate_key(self) -> bytes: + """Generate and save an encryption key.""" + key = Fernet.generate_key() + with open(self.key_file, 'wb') as f: + f.write(key) + os.chmod(self.key_file, stat.S_IRUSR | stat.S_IWUSR) + return key + + def _load_key(self) -> bytes: + """Load encryption key, generate if not exists.""" + try: + with open(self.key_file, 'rb') as f: + return f.read() + except FileNotFoundError: + return self._generate_key() + + def encrypt_password(self, password: str) -> str: + """Encrypt a password.""" + key = self._load_key() + f = Fernet(key) + encrypted = f.encrypt(password.encode()) + return base64.b64encode(encrypted).decode() + + def decrypt_password(self, encrypted_password: str) -> Optional[str]: + """Decrypt a password.""" + try: + key = self._load_key() + f = Fernet(key) + decoded = base64.b64decode(encrypted_password.encode()) + return f.decrypt(decoded).decode() + except Exception: + return None + + # ========================================================================= + # Configuration methods + # ========================================================================= + + def create_config(self, hostname: str, username: str, password: str, + port: int = 22) -> bool: + """Create configuration file with encrypted password.""" + encrypted_password = self.encrypt_password(password) + + config = { + "ssh": { + "hostname": hostname, + "username": username, + "password_encrypted": encrypted_password, + "port": port, + "timeout": 10 + }, + "settings": { + "command_delay": 1.0, + "response_timeout": 3.0 + } + } + + with open(self.config_file, 'w', encoding='utf-8') as f: + json.dump(config, f, indent=4, ensure_ascii=False) + os.chmod(self.config_file, stat.S_IRUSR | stat.S_IWUSR) + return True + + def load_config(self) -> Tuple[bool, str]: + """Load configuration from file. Returns (success, message).""" + try: + with open(self.config_file, 'r', encoding='utf-8') as f: + config = json.load(f) + + ssh_config = config['ssh'] + self.hostname = ssh_config['hostname'] + self.username = ssh_config['username'] + self.port = ssh_config.get('port', 22) + self.timeout = ssh_config.get('timeout', 10) + + # Decrypt password + if 'password_encrypted' in ssh_config: + password = self.decrypt_password(ssh_config['password_encrypted']) + if password is None: + return False, "Impossibile decrittare la password" + self._password = password + elif 'password' in ssh_config: + self._password = ssh_config['password'] + else: + return False, "Password non trovata nella configurazione" + + # Load settings + settings = config.get('settings', {}) + self.command_delay = settings.get('command_delay', 1.0) + self.response_timeout = settings.get('response_timeout', 3.0) + + return True, "Configurazione caricata" + + except FileNotFoundError: + return False, f"File configurazione non trovato: {self.config_file}" + except json.JSONDecodeError as e: + return False, f"Errore nel file di configurazione: {e}" + except KeyError as e: + return False, f"Chiave mancante nella configurazione: {e}" + + def config_exists(self) -> bool: + """Check if configuration file exists.""" + return self.config_file.exists() + + # ========================================================================= + # Connection methods + # ========================================================================= + + def connect(self) -> Tuple[bool, str]: + """ + Connect to the switch and enter CLI mode. + Returns (success, message). + """ + try: + self.ssh_client = paramiko.SSHClient() + self.ssh_client.set_missing_host_key_policy(paramiko.AutoAddPolicy()) + + self.ssh_client.connect( + hostname=self.hostname, + port=self.port, + username=self.username, + password=self._password, + timeout=self.timeout + ) + + # Open interactive shell + self.shell = self.ssh_client.invoke_shell() + time.sleep(2) + + # Clear initial buffer + if self.shell.recv_ready(): + self.shell.recv(65536) + + # Enter CLI mode + self._send_command("cli", wait_prompt=False) + time.sleep(1) + self._send_command("enable", wait_prompt=False) + time.sleep(0.5) + + return True, f"Connesso a {self.hostname}" + + except paramiko.AuthenticationException: + return False, "Autenticazione fallita. Controlla username e password." + except paramiko.SSHException as e: + return False, f"Errore SSH: {e}" + except Exception as e: + return False, f"Errore di connessione: {e}" + + def disconnect(self) -> None: + """Disconnect from the switch.""" + if self.shell: + try: + self._send_command("exit", wait_prompt=False) + time.sleep(0.3) + self._send_command("exit", wait_prompt=False) + except Exception: + pass + self.shell.close() + self.shell = None + + if self.ssh_client: + self.ssh_client.close() + self.ssh_client = None + + def is_connected(self) -> bool: + """Check if connected to switch.""" + return self.shell is not None and self.ssh_client is not None + + # ========================================================================= + # Command execution methods + # ========================================================================= + + def _send_command(self, command: str, wait_prompt: bool = True) -> str: + """ + Send a single command and return the response. + """ + if not self.shell: + raise RuntimeError("Non connesso allo switch") + + self.shell.send(command + "\n") + time.sleep(self.command_delay) + + response = "" + start_time = time.time() + + while time.time() - start_time < self.response_timeout: + if self.shell.recv_ready(): + data = self.shell.recv(65536).decode('utf-8', errors='ignore') + response += data + if not wait_prompt: + break + # Check for prompt + if response.strip().endswith(('#', '>', ')')): + break + time.sleep(0.1) + + return response + + def _send_command_long(self, command: str, timeout: float = 10.0) -> str: + """ + Send command expecting long output (like show running-config). + Handles --More-- prompts. + """ + if not self.shell: + raise RuntimeError("Non connesso allo switch") + + self.shell.send(command + "\n") + + response = "" + start_time = time.time() + last_data_time = time.time() + + while time.time() - start_time < timeout: + if self.shell.recv_ready(): + data = self.shell.recv(65536).decode('utf-8', errors='ignore') + response += data + last_data_time = time.time() + + # Handle --More-- prompt + if '--More--' in data or '--more--' in data.lower(): + self.shell.send(" ") # Space to continue + time.sleep(0.2) + continue + + else: + # No data available + if time.time() - last_data_time > 2.0: + # No new data for 2 seconds, assume complete + break + time.sleep(0.1) + + return response + + def execute_command(self, command: str) -> str: + """Execute a single command and return output.""" + return self._send_command(command) + + def execute_commands(self, commands: List[str]) -> List[str]: + """Execute multiple commands and return list of outputs.""" + results = [] + for cmd in commands: + result = self._send_command(cmd) + results.append(result) + return results + + # ========================================================================= + # High-level methods for switch data retrieval + # ========================================================================= + + def get_running_config(self) -> str: + """Retrieve running configuration from switch.""" + return self._send_command_long("show running-config", timeout=30.0) + + def get_interfaces_status(self) -> str: + """Retrieve interface status from switch.""" + return self._send_command_long("show interfaces status all", timeout=15.0) + + def get_port_channel_brief(self) -> str: + """Retrieve port-channel/LAG summary from switch.""" + return self._send_command_long("show port-channel brief", timeout=10.0) + + def enter_config_mode(self) -> bool: + """Enter configuration mode.""" + response = self._send_command("configure") + return "config" in response.lower() or "(Config)" in response + + def exit_config_mode(self) -> bool: + """Exit configuration mode.""" + self._send_command("exit", wait_prompt=False) + return True + + def apply_mirroring_config(self, commands: List[str]) -> Tuple[bool, str]: + """ + Apply mirroring configuration commands. + + Args: + commands: List of commands (without cli/enable/config prefixes) + + Returns: + (success, output) + """ + try: + # Enter config mode + self.enter_config_mode() + time.sleep(0.5) + + output_lines = [] + for cmd in commands: + result = self._send_command(cmd) + output_lines.append(f">>> {cmd}") + if result.strip(): + output_lines.append(result.strip()) + + # Exit config mode + self.exit_config_mode() + + return True, "\n".join(output_lines) + + except Exception as e: + return False, f"Errore durante l'applicazione: {e}" + + +# ============================================================================= +# Interactive configuration setup +# ============================================================================= + +def setup_config_interactive() -> Optional[SSHManager]: + """Interactive setup for SSH configuration.""" + print("\n=== CONFIGURAZIONE SSH ===\n") + + hostname = input("IP del dispositivo: ").strip() + if not hostname: + print("IP non valido") + return None + + username = input("Username [admin]: ").strip() or "admin" + password = getpass("Password: ").strip() + if not password: + print("Password non valida") + return None + + port_str = input("Porta SSH [22]: ").strip() or "22" + try: + port = int(port_str) + except ValueError: + port = 22 + + manager = SSHManager() + manager.create_config(hostname, username, password, port) + + print("\n✓ Configurazione salvata e crittografata") + return manager + + +if __name__ == "__main__": + # Test connection + manager = SSHManager() + + if not manager.config_exists(): + manager = setup_config_interactive() + if not manager: + exit(1) + + success, msg = manager.load_config() + if not success: + print(f"Errore: {msg}") + exit(1) + + print(f"Connessione a {manager.hostname}...") + success, msg = manager.connect() + + if success: + print(f"✓ {msg}") + print("\nRecupero configurazione...") + config = manager.get_running_config() + print(f"Ricevuti {len(config)} caratteri") + manager.disconnect() + print("✓ Disconnesso") + else: + print(f"✗ {msg}") diff --git a/tui_app.py b/tui_app.py new file mode 100644 index 0000000..1c66c0d --- /dev/null +++ b/tui_app.py @@ -0,0 +1,823 @@ +#!/usr/bin/env python3 +""" +Textual-based TUI Application for Ubiquiti Port Mirroring Manager. +Full-screen application with keyboard navigation and interactive selection. +""" + +from textual.app import App, ComposeResult +from textual.containers import Container, Horizontal, Vertical, ScrollableContainer +from textual.widgets import ( + Header, Footer, Static, Button, Label, Input, + DataTable, TabbedContent, TabPane, Checkbox, Rule +) +from textual.binding import Binding +from textual.screen import Screen, ModalScreen +from textual.message import Message +from textual import work +from textual import events +from rich.text import Text + +from typing import Optional, List, Set, Tuple +from pathlib import Path + +from ssh_manager import SSHManager +from config_parser import ConfigParser, SwitchConfig + + +# ============================================================================= +# Login Screen +# ============================================================================= + +class LoginScreen(ModalScreen): + """Modal screen for SSH login.""" + + BINDINGS = [ + Binding("escape", "cancel", "Annulla"), + ] + + CSS = """ + LoginScreen { + align: center middle; + } + + #login-container { + width: 60; + height: auto; + border: thick $accent; + background: $surface; + padding: 1 2; + } + + #login-title { + text-align: center; + text-style: bold; + margin-bottom: 1; + } + + .login-label { + margin-top: 1; + } + + .login-input { + margin-bottom: 1; + } + + #login-buttons { + margin-top: 1; + align: center middle; + } + + #login-buttons Button { + margin: 0 1; + } + + #login-error { + color: $error; + text-align: center; + margin-top: 1; + } + """ + + def __init__(self, saved_ip: str = ""): + super().__init__() + self.saved_ip = saved_ip + self.result: Optional[Tuple[str, str]] = None + + def compose(self) -> ComposeResult: + with Container(id="login-container"): + yield Label("CONNESSIONE AL DISPOSITIVO", id="login-title") + yield Rule() + yield Label("IP del dispositivo:", classes="login-label") + yield Input(value=self.saved_ip, placeholder="192.168.1.1", id="ip-input", classes="login-input") + yield Label("Password (utente: admin):", classes="login-label") + yield Input(placeholder="password", password=True, id="password-input", classes="login-input") + yield Label("", id="login-error") + with Horizontal(id="login-buttons"): + yield Button("Connetti", variant="primary", id="btn-connect") + yield Button("Esci", variant="default", id="btn-cancel") + + def on_button_pressed(self, event: Button.Pressed) -> None: + if event.button.id == "btn-connect": + ip = self.query_one("#ip-input", Input).value.strip() + password = self.query_one("#password-input", Input).value + + if not ip: + self.query_one("#login-error", Label).update("IP non valido") + return + if not password: + self.query_one("#login-error", Label).update("Password non valida") + return + + self.result = (ip, password) + self.dismiss(self.result) + + elif event.button.id == "btn-cancel": + self.dismiss(None) + + def on_input_submitted(self, event: Input.Submitted) -> None: + # Press Enter to submit + if event.input.id == "password-input": + self.query_one("#btn-connect", Button).press() + elif event.input.id == "ip-input": + self.query_one("#password-input", Input).focus() + + def action_cancel(self) -> None: + self.dismiss(None) + + +# ============================================================================= +# Port Selection Screen (for mirroring configuration) +# ============================================================================= + +class PortSelectionScreen(Screen): + """Full-screen port selection with keyboard navigation.""" + + BINDINGS = [ + Binding("up", "move_up", "Su"), + Binding("down", "move_down", "Giù"), + Binding("space", "toggle_select", "Seleziona"), + Binding("enter", "toggle_select", "Seleziona"), + Binding("a", "select_all", "Tutti"), + Binding("n", "select_none", "Nessuno"), + Binding("d", "set_destination", "Destination"), + Binding("escape", "cancel", "Annulla"), + Binding("f10", "confirm", "Conferma"), + ] + + CSS = """ + PortSelectionScreen { + layout: vertical; + } + + #port-table-container { + height: 1fr; + border: solid $primary; + margin: 1; + } + + #selection-info { + height: 5; + border: solid $accent; + margin: 0 1 1 1; + padding: 0 1; + } + + #info-destination { + color: $warning; + } + + #info-sources { + color: $success; + } + + .highlight-row { + background: $accent; + } + """ + + def __init__( + self, + ports: List[Tuple[str, str, str, int, str]], # (id, desc, state, pvid, tagged) + current_dest: Optional[str], + current_sources: Set[str] + ): + super().__init__() + self.ports = ports + self.destination = current_dest + self.sources = current_sources.copy() + self.cursor_row = 0 + self.result: Optional[Tuple[str, List[str]]] = None + + def compose(self) -> ComposeResult: + yield Header() + + with Container(id="port-table-container"): + table = DataTable(id="port-table", cursor_type="row") + yield table + + with Container(id="selection-info"): + yield Label("DESTINATION: (nessuna)", id="info-destination") + yield Label("SOURCES: 0 porte selezionate", id="info-sources") + yield Label("[D]=Destination [SPAZIO]=Toggle [A]=Tutti [N]=Nessuno [F10]=Conferma [ESC]=Annulla", id="info-help") + + yield Footer() + + def on_mount(self) -> None: + """Initialize the table.""" + table = self.query_one("#port-table", DataTable) + + # Add columns + table.add_column("", width=3, key="check") + table.add_column("Porta", width=8, key="port") + table.add_column("Descrizione", width=28, key="desc") + table.add_column("Stato", width=6, key="state") + table.add_column("PVID", width=5, key="pvid") + table.add_column("Tagged VLANs", width=24, key="tagged") + table.add_column("Ruolo", width=14, key="role") + + # Add rows + for port_id, desc, state, pvid, tagged in self.ports: + self._add_port_row(table, port_id, desc, state, pvid, tagged) + + # Update info + self._update_info() + + # Focus table + table.focus() + + def _add_port_row(self, table: DataTable, port_id: str, desc: str, state: str, pvid: int, tagged: str) -> None: + """Add a row to the table.""" + # Checkbox + if port_id == self.destination: + check = Text("◀◀", style="bold magenta") + role = Text("DESTINATION", style="bold magenta") + elif port_id in self.sources: + check = Text("[X]", style="bold cyan") + role = Text("SOURCE", style="bold cyan") + else: + check = Text("[ ]", style="dim") + role = Text("-", style="dim") + + # State + state_text = Text("UP", style="bold green") if state == "Up" else Text("DOWN", style="dim") + + table.add_row( + check, + port_id, + desc[:28], + state_text, + str(pvid), + tagged[:24] if tagged else "-", + role, + key=port_id + ) + + def _refresh_table(self) -> None: + """Refresh all rows in the table.""" + table = self.query_one("#port-table", DataTable) + + for port_id, desc, state, pvid, tagged in self.ports: + # Checkbox + if port_id == self.destination: + check = Text("◀◀", style="bold magenta") + role = Text("DESTINATION", style="bold magenta") + elif port_id in self.sources: + check = Text("[X]", style="bold cyan") + role = Text("SOURCE", style="bold cyan") + else: + check = Text("[ ]", style="dim") + role = Text("-", style="dim") + + state_text = Text("UP", style="bold green") if state == "Up" else Text("DOWN", style="dim") + + # Update row + table.update_cell(port_id, "check", check) + table.update_cell(port_id, "role", role) + + self._update_info() + + def _update_info(self) -> None: + """Update the info panel.""" + dest_label = self.query_one("#info-destination", Label) + src_label = self.query_one("#info-sources", Label) + + if self.destination: + # Find description + desc = "" + for p in self.ports: + if p[0] == self.destination: + desc = p[1] + break + dest_label.update(f"DESTINATION: {self.destination} ({desc})") + else: + dest_label.update("DESTINATION: (nessuna - premi D per impostare)") + + src_label.update(f"SOURCES: {len(self.sources)} porte selezionate") + + def action_move_up(self) -> None: + table = self.query_one("#port-table", DataTable) + table.action_cursor_up() + + def action_move_down(self) -> None: + table = self.query_one("#port-table", DataTable) + table.action_cursor_down() + + def action_toggle_select(self) -> None: + """Toggle selection of current row.""" + table = self.query_one("#port-table", DataTable) + + if table.cursor_row is None: + return + + # Get port_id from our ports list using cursor position + if table.cursor_row >= len(self.ports): + return + + port_id = self.ports[table.cursor_row][0] + + if port_id == self.destination: + # Can't select destination as source + self.notify("Non puoi selezionare la destination come source", severity="warning") + return + + if port_id in self.sources: + self.sources.remove(port_id) + else: + self.sources.add(port_id) + + self._refresh_table() + + def action_set_destination(self) -> None: + """Set current row as destination.""" + table = self.query_one("#port-table", DataTable) + + if table.cursor_row is None: + return + + # Get port_id from our ports list using cursor position + if table.cursor_row >= len(self.ports): + return + + port_id = self.ports[table.cursor_row][0] + + # Remove from sources if was selected + self.sources.discard(port_id) + self.destination = port_id + + self._refresh_table() + self.notify(f"Destination impostata: {port_id}", severity="information") + + def action_select_all(self) -> None: + """Select all ports as source (except destination).""" + for port_id, _, _, _, _ in self.ports: + if port_id != self.destination: + self.sources.add(port_id) + self._refresh_table() + + def action_select_none(self) -> None: + """Deselect all sources.""" + self.sources.clear() + self._refresh_table() + + def action_cancel(self) -> None: + """Cancel and return to main.""" + self.app.pop_screen() + + def action_confirm(self) -> None: + """Confirm selection.""" + if not self.destination: + self.notify("Devi selezionare una porta DESTINATION (premi D)", severity="error") + return + + if not self.sources: + self.notify("Devi selezionare almeno una porta SOURCE", severity="error") + return + + self.result = (self.destination, list(self.sources)) + self.dismiss(self.result) + + +# ============================================================================= +# Main Application +# ============================================================================= + +class PortMirroringApp(App): + """Main Textual application for Port Mirroring Manager.""" + + TITLE = "Ubiquiti Port Mirroring Manager" + + CSS = """ + Screen { + background: $surface; + } + + #main-container { + height: 1fr; + } + + #status-bar { + height: 3; + dock: bottom; + background: $primary-background; + padding: 0 1; + } + + #connection-status { + color: $success; + } + + .port-table { + height: 1fr; + } + + #tab-ports DataTable { + height: 1fr; + } + + #tab-mirroring { + padding: 1; + } + + #mirror-info { + height: auto; + border: solid $accent; + padding: 1; + margin-bottom: 1; + } + + #mirror-dest { + color: $warning; + text-style: bold; + margin-bottom: 1; + } + + #mirror-sources-title { + color: $success; + text-style: bold; + } + + #action-buttons { + height: auto; + margin-top: 1; + } + + #action-buttons Button { + margin: 0 1 0 0; + } + + .info-panel { + height: auto; + border: solid $primary; + padding: 1; + margin: 1; + } + + #device-info { + height: auto; + dock: top; + background: $primary; + color: $text; + padding: 0 1; + text-align: center; + } + """ + + BINDINGS = [ + Binding("q", "quit", "Esci"), + Binding("r", "reload", "Ricarica"), + Binding("c", "configure", "Configura"), + Binding("g", "generate", "Genera Script"), + Binding("f5", "apply", "Applica"), + ] + + def __init__(self): + super().__init__() + self.ssh_manager = SSHManager() + self.parser = ConfigParser() + self.is_connected = False + self.device_ip = "" + + # Pending configuration + self.pending_dest: Optional[str] = None + self.pending_sources: List[str] = [] + + def compose(self) -> ComposeResult: + yield Header() + yield Label("Non connesso", id="device-info") + + with TabbedContent(id="main-tabs"): + with TabPane("Stato Porte", id="tab-ports"): + yield DataTable(id="ports-table", cursor_type="row") + + with TabPane("Mirroring", id="tab-mirroring"): + with Container(id="mirror-info"): + yield Label("DESTINATION: -", id="mirror-dest") + yield Label("SOURCE PORTS:", id="mirror-sources-title") + yield DataTable(id="sources-table", cursor_type="row") + + with Horizontal(id="action-buttons"): + yield Button("Configura Mirroring", variant="primary", id="btn-configure") + yield Button("Genera Script", variant="default", id="btn-generate") + yield Button("Applica al Device", variant="warning", id="btn-apply") + + yield Footer() + + def on_mount(self) -> None: + """Called when app is mounted.""" + # Setup tables + self._setup_ports_table() + self._setup_sources_table() + + # Show login (use call_later to avoid blocking) + self.call_later(self._show_login_screen) + + def _setup_ports_table(self) -> None: + """Setup the ports table columns.""" + table = self.query_one("#ports-table", DataTable) + table.add_column("Porta", width=8, key="port") + table.add_column("Descrizione", width=28, key="desc") + table.add_column("Stato", width=6, key="state") + table.add_column("PVID", width=5, key="pvid") + table.add_column("Tagged VLANs", width=26, key="tagged") + table.add_column("Mirroring", width=14, key="mirror") + + def _setup_sources_table(self) -> None: + """Setup the sources table columns.""" + table = self.query_one("#sources-table", DataTable) + table.add_column("Porta", width=10, key="port") + table.add_column("Descrizione", width=32, key="desc") + table.add_column("Stato", width=8, key="state") + + def _show_login_screen(self) -> None: + """Show login screen.""" + saved_ip = "" + if self.ssh_manager.config_exists(): + success, _ = self.ssh_manager.load_config() + if success: + saved_ip = self.ssh_manager.hostname + + self.push_screen(LoginScreen(saved_ip), callback=self._on_login_result) + + def _on_login_result(self, result: Optional[Tuple[str, str]]) -> None: + """Handle login result.""" + if result is None: + self.exit() + return + + ip, password = result + self._do_connect(ip, password) + + @work(exclusive=True, thread=True) + def _do_connect(self, ip: str, password: str) -> None: + """Connect to device (runs in worker thread).""" + self.notify(f"Connessione a {ip}...", severity="information") + + # Save and load config + self.ssh_manager.create_config(ip, "admin", password, 22) + success, msg = self.ssh_manager.load_config() + + if not success: + self.notify(f"Errore: {msg}", severity="error") + self.call_from_thread(self._show_login_screen) + return + + # Connect + success, msg = self.ssh_manager.connect() + + if not success: + self.notify(f"Connessione fallita: {msg}", severity="error") + self.call_from_thread(self._show_login_screen) + return + + self.is_connected = True + self.device_ip = ip + + # Update header (must be called from main thread) + self.call_from_thread(self._update_device_info, f"Connesso a {ip} (admin)") + + self.notify("Connesso! Caricamento configurazione...", severity="information") + + # Load config + self._load_config_sync() + + def _update_device_info(self, text: str) -> None: + """Update device info label.""" + self.query_one("#device-info", Label).update(text) + + def _load_config_sync(self) -> None: + """Load configuration synchronously (called from worker thread).""" + if not self.is_connected: + self.notify("Non connesso", severity="error") + return + + try: + # Get running config + running_config = self.ssh_manager.get_running_config() + self.parser.parse_running_config(running_config) + + # Get interface status + iface_status = self.ssh_manager.get_interfaces_status() + self.parser.parse_interfaces_status(iface_status) + + # Get port-channel + pc_brief = self.ssh_manager.get_port_channel_brief() + self.parser.parse_port_channel_brief(pc_brief) + + # Update device info + hostname = self.parser.config.system_info.hostname or self.device_ip + self.call_from_thread(self._update_device_info, f"{hostname} ({self.device_ip})") + + # Refresh tables (must be called from main thread) + self.call_from_thread(self._refresh_ports_table) + self.call_from_thread(self._refresh_mirroring_info) + + self.notify("Configurazione caricata", severity="information") + + except Exception as e: + self.notify(f"Errore: {e}", severity="error") + + @work(exclusive=True, thread=True) + def _reload_config_worker(self) -> None: + """Reload configuration in worker thread.""" + self._load_config_sync() + + def _refresh_ports_table(self) -> None: + """Refresh the ports table.""" + table = self.query_one("#ports-table", DataTable) + table.clear() + + dest = self.parser.get_mirroring_destination() + sources = set(self.parser.get_mirroring_sources()) + + for port_id in self.parser.get_physical_ports(): + iface = self.parser.config.interfaces.get(port_id) + if not iface: + continue + + # State + state = Text("UP", style="bold green") if iface.link_state == "Up" else Text("DOWN", style="dim") + + # Tagged VLANs + tagged = ConfigParser.format_vlan_list(iface.tagged_vlans) + if len(tagged) > 26: + tagged = tagged[:23] + "..." + + # Mirroring + if port_id == dest: + mirror = Text("◀◀ DEST", style="bold magenta") + elif port_id in sources: + mirror = Text("▶▶ SRC", style="bold cyan") + else: + mirror = Text("-", style="dim") + + table.add_row( + port_id, + iface.description[:28] if iface.description else "", + state, + str(iface.pvid), + tagged or "-", + mirror, + key=port_id + ) + + def _refresh_mirroring_info(self) -> None: + """Refresh mirroring info panel.""" + dest = self.parser.get_mirroring_destination() + sources = self.parser.get_mirroring_sources() + + # Destination + dest_label = self.query_one("#mirror-dest", Label) + if dest: + iface = self.parser.config.interfaces.get(dest) + desc = iface.description if iface else "" + dest_label.update(f"DESTINATION: {dest} ({desc})") + else: + dest_label.update("DESTINATION: (non configurata)") + + # Sources table + table = self.query_one("#sources-table", DataTable) + table.clear() + + for src in sorted(sources, key=lambda x: (int(x.split("/")[0]), int(x.split("/")[1]))): + iface = self.parser.config.interfaces.get(src) + if iface: + state = "UP" if iface.link_state == "Up" else "DOWN" + desc = iface.description or "" + else: + state = "-" + desc = "" + + table.add_row(src, desc[:32], state, key=src) + + # Update title + self.query_one("#mirror-sources-title", Label).update(f"SOURCE PORTS ({len(sources)}):") + + def action_reload(self) -> None: + """Reload configuration.""" + self._reload_config_worker() + + def action_configure(self) -> None: + """Open port selection screen.""" + if not self.is_connected: + self.notify("Non connesso", severity="error") + return + + # Prepare port list + ports = [] + for port_id in self.parser.get_physical_ports(): + iface = self.parser.config.interfaces.get(port_id) + if iface: + tagged = ConfigParser.format_vlan_list(iface.tagged_vlans) + ports.append((port_id, iface.description or "", iface.link_state, iface.pvid, tagged)) + + # Add LAGs + for lag_id in self.parser.get_lag_ports(): + lag = self.parser.config.lags.get(lag_id) + if lag and (lag.member_ports or lag.link_state == "Up"): + desc = f"LAG {lag.name} ({','.join(lag.member_ports)})" + ports.append((lag_id, desc, lag.link_state, 1, "")) + + current_dest = self.parser.get_mirroring_destination() + current_sources = set(self.parser.get_mirroring_sources()) + + # Show selection screen with callback + self.push_screen( + PortSelectionScreen(ports, current_dest, current_sources), + callback=self._on_configure_result + ) + + def _on_configure_result(self, result: Optional[Tuple[str, List[str]]]) -> None: + """Handle configuration result.""" + if result: + dest, sources = result + self.pending_dest = dest + self.pending_sources = sources + self.notify(f"Configurazione salvata: {dest} <- {len(sources)} sources", severity="information") + + def action_generate(self) -> None: + """Generate configuration script.""" + from script_generator import ScriptGenerator + + dest = self.pending_dest or self.parser.get_mirroring_destination() + sources = self.pending_sources or self.parser.get_mirroring_sources() + + if not dest: + self.notify("Nessuna configurazione da generare", severity="warning") + return + + # Generate commands + commands = self._generate_commands(dest, sources) + + # Generate scripts + generator = ScriptGenerator() + paths = generator.generate_all( + commands=commands, + device_ip=self.device_ip, + username="admin", + config=self.parser.config, + description=f"Mirroring: {dest} <- {len(sources)} sources" + ) + + self.notify(f"Script generati in: {paths['bash'].parent}", severity="information") + + def action_apply(self) -> None: + """Apply configuration to device.""" + if not self.is_connected: + self.notify("Non connesso", severity="error") + return + + dest = self.pending_dest + sources = self.pending_sources + + if not dest or not sources: + self.notify("Nessuna configurazione da applicare. Usa prima 'Configura'", severity="warning") + return + + self._apply_config_worker(dest, sources) + + @work(exclusive=True, thread=True) + def _apply_config_worker(self, dest: str, sources: List[str]) -> None: + """Apply configuration in worker thread.""" + commands = self._generate_commands(dest, sources) + + self.notify("Applicazione in corso...", severity="information") + + success, output = self.ssh_manager.apply_mirroring_config(commands) + + if success: + self.notify("Configurazione applicata!", severity="information") + self.pending_dest = None + self.pending_sources = [] + self._load_config_sync() + else: + self.notify(f"Errore: {output}", severity="error") + + def _generate_commands(self, dest: str, sources: List[str]) -> List[str]: + """Generate mirroring commands.""" + commands = [f"no monitor session 1"] + commands.append(f"monitor session 1 destination interface {dest}") + + for src in sorted(sources, key=lambda x: (int(x.split("/")[0]), int(x.split("/")[1]))): + commands.append(f"monitor session 1 source interface {src}") + + commands.append("monitor session 1 mode") + return commands + + def action_quit(self) -> None: + """Quit application.""" + if self.is_connected: + self.ssh_manager.disconnect() + self.exit() + + +# ============================================================================= +# Entry Point +# ============================================================================= + +def main(): + app = PortMirroringApp() + app.run() + + +if __name__ == "__main__": + main() diff --git a/ui.py b/ui.py new file mode 100644 index 0000000..1ade7f4 --- /dev/null +++ b/ui.py @@ -0,0 +1,972 @@ +#!/usr/bin/env python3 +""" +Text-based User Interface for Ubiquiti Port Mirroring Manager. +Uses the 'rich' library for enhanced terminal output. +""" + +import os +import sys +from typing import Optional, List, Set, Tuple +from rich.console import Console +from rich.table import Table +from rich.panel import Panel +from rich.prompt import Prompt, Confirm +from rich.text import Text +from rich.style import Style +from rich import box + +from config_parser import ConfigParser, SwitchConfig, InterfaceConfig, LAGConfig + + +class PortMirroringUI: + """Text-based UI for Port Mirroring configuration.""" + + def __init__(self): + self.console = Console() + self.config: Optional[SwitchConfig] = None + self.parser: Optional[ConfigParser] = None + + # Current mirroring selection + self.selected_destination: Optional[str] = None + self.selected_sources: Set[str] = set() + + # Device info + self.device_ip: str = "" + self.device_user: str = "" + + def set_config(self, parser: ConfigParser) -> None: + """Set the configuration parser.""" + self.parser = parser + self.config = parser.config + + # Load current mirroring state + self.selected_destination = parser.get_mirroring_destination() + self.selected_sources = set(parser.get_mirroring_sources()) + + def set_device_info(self, ip: str, user: str) -> None: + """Set device connection info for display.""" + self.device_ip = ip + self.device_user = user + + def _get_port_sort_key(self, port_id: str) -> Tuple[int, int]: + """Get sort key for port ID (prefix, number).""" + if "/" in port_id: + parts = port_id.split("/") + try: + return (int(parts[0]), int(parts[1])) + except ValueError: + return (999, 0) + return (999, 0) + + # ========================================================================= + # Screen utilities + # ========================================================================= + + def clear_screen(self) -> None: + """Clear the terminal screen.""" + os.system('cls' if os.name == 'nt' else 'clear') + + def wait_for_enter(self, message: str = "Premi INVIO per continuare...") -> None: + """Wait for user to press Enter.""" + self.console.print(f"\n[dim]{message}[/dim]") + input() + + def print_header(self, title: str = "UBIQUITI PORT MIRRORING MANAGER") -> None: + """Print application header.""" + self.clear_screen() + + # Build header text + header_lines = [title] + + if self.config and self.config.system_info.hostname: + header_lines.append(f"Device: {self.config.system_info.hostname}") + elif self.device_ip: + header_lines.append(f"Device: {self.device_ip}") + + if self.config and self.config.system_info.description: + # Extract model from description + desc = self.config.system_info.description + if "," in desc: + model = desc.split(",")[0] + header_lines.append(f"Model: {model}") + + header_text = "\n".join(header_lines) + + self.console.print(Panel( + header_text, + style="bold white on blue", + box=box.DOUBLE + )) + self.console.print() + + # ========================================================================= + # Main menu + # ========================================================================= + + def show_main_menu(self) -> str: + """Display main menu and return selected option.""" + self.print_header() + + menu_items = [ + ("1", "Visualizza Stato Porte"), + ("2", "Visualizza Configurazione Mirroring"), + ("3", "Configura Port Mirroring"), + ("4", "Genera Script di Configurazione"), + ("5", "Applica Configurazione al Device"), + ("6", "Ricarica Configurazione dal Device"), + ("0", "Esci"), + ] + + table = Table(show_header=False, box=box.ROUNDED, padding=(0, 2)) + table.add_column("Opzione", style="bold cyan") + table.add_column("Descrizione") + + for opt, desc in menu_items: + table.add_row(f"[{opt}]", desc) + + self.console.print(table) + self.console.print() + + choice = Prompt.ask( + "Seleziona opzione", + choices=["0", "1", "2", "3", "4", "5", "6"], + default="1" + ) + + return choice + + # ========================================================================= + # Port status view + # ========================================================================= + + def show_port_status(self) -> None: + """Display port status with menu for full/compact view.""" + while True: + self.print_header("STATO PORTE SWITCH") + + if not self.config: + self.console.print("[red]Configurazione non caricata[/red]") + self.wait_for_enter() + return + + # Show options + self.console.print("[bold]Opzioni visualizzazione:[/bold]") + self.console.print(" [1] Vista completa (tutte le porte)") + self.console.print(" [2] Vista compatta (solo porte UP e in mirroring)") + self.console.print(" [0] Torna al menu principale") + self.console.print() + + choice = Prompt.ask("Seleziona", choices=["0", "1", "2"], default="1") + + if choice == "0": + return + elif choice == "1": + self._show_ports_full() + elif choice == "2": + self._show_ports_compact() + + def _show_ports_full(self) -> None: + """Display full port status table with all ports.""" + self.clear_screen() + self.print_header("STATO PORTE - VISTA COMPLETA") + + self._render_port_table(compact=False) + + # LAG table + if self.config.lags: + self.console.print() + self._show_lag_table() + + self.wait_for_enter() + + def _show_ports_compact(self) -> None: + """Display compact view with only UP ports and mirroring ports.""" + self.clear_screen() + self.print_header("STATO PORTE - VISTA COMPATTA") + + self._render_port_table(compact=True) + + # LAG table (only active) + if self.config.lags: + active_lags = {k: v for k, v in self.config.lags.items() + if v.link_state == "Up" or self.parser.is_mirroring_source(k)} + if active_lags: + self.console.print() + self._show_lag_table(only_active=True) + + self.wait_for_enter() + + def _render_port_table(self, compact: bool = False) -> None: + """Render the port status table.""" + if not self.config or not self.parser: + return + + # Create table + table = Table(box=box.ROUNDED, show_lines=False, padding=(0, 1)) + + table.add_column("Porta", style="bold", width=6, justify="center") + table.add_column("Descrizione", width=26) + table.add_column("Stato", width=5, justify="center") + table.add_column("PVID", width=5, justify="center") + table.add_column("Tagged VLANs", width=22) + table.add_column("Mirroring", width=16) + + # Get sorted physical ports + ports = self.parser.get_physical_ports() + + dest_port = self.parser.get_mirroring_destination() + source_ports = set(self.parser.get_mirroring_sources()) + + for port_id in ports: + iface = self.config.interfaces.get(port_id) + if not iface: + continue + + is_up = iface.link_state == "Up" + is_dest = port_id == dest_port + is_source = port_id in source_ports + + # Compact mode: skip ports that are DOWN and not in mirroring + if compact and not is_up and not is_dest and not is_source: + continue + + # State styling + if is_up: + state_text = Text("UP", style="bold green") + else: + state_text = Text("DOWN", style="dim") + + # VLAN PVID + pvid_text = str(iface.pvid) + + # Tagged VLANs - format as compact string + tagged_vlans = ConfigParser.format_vlan_list(iface.tagged_vlans) + if len(tagged_vlans) > 22: + tagged_vlans = tagged_vlans[:19] + "..." + if not tagged_vlans: + tagged_vlans = "-" + + # Mirroring status with clear visual + if is_dest: + mirror_text = Text("◀◀ DEST", style="bold magenta on black") + elif is_source: + mirror_text = Text("▶▶ SRC", style="bold cyan") + else: + mirror_text = Text("-", style="dim") + + # Row style based on mirroring + row_style = "" + if is_dest: + row_style = "on dark_magenta" + elif is_source: + row_style = "" + + table.add_row( + port_id, + (iface.description or "")[:26], + state_text, + pvid_text, + tagged_vlans, + mirror_text, + style=row_style + ) + + self.console.print(table) + + # Summary + self.console.print() + self._show_mirroring_summary() + + def _show_mirroring_summary(self) -> None: + """Show a clear mirroring summary box.""" + if not self.parser: + return + + dest = self.parser.get_mirroring_destination() + sources = self.parser.get_mirroring_sources() + + if not dest: + self.console.print(Panel( + "[yellow]Nessun port mirroring configurato[/yellow]", + title="MIRRORING", + box=box.ROUNDED, + width=60 + )) + return + + # Build summary + dest_iface = self.config.interfaces.get(dest) if self.config else None + dest_desc = dest_iface.description if dest_iface else "" + + lines = [ + f"[bold magenta]DESTINATION:[/bold magenta] {dest}" + + (f" ({dest_desc})" if dest_desc else ""), + "", + f"[bold cyan]SOURCES ({len(sources)}):[/bold cyan]" + ] + + # Group sources for display + source_list = sorted(sources, key=self._get_port_sort_key) + + # Show in columns (4 per row) + row = [] + for src in source_list: + row.append(src) + if len(row) == 6: + lines.append(" " + " ".join(row)) + row = [] + if row: + lines.append(" " + " ".join(row)) + + self.console.print(Panel( + "\n".join(lines), + title="MIRRORING ATTIVO", + box=box.DOUBLE, + width=60, + style="bold" + )) + + def _show_lag_table(self, only_active: bool = False) -> None: + """Display LAG status table.""" + if not self.config or not self.config.lags: + return + + table = Table( + title="Link Aggregation Groups (LAG)", + box=box.ROUNDED, + show_lines=False + ) + + table.add_column("LAG", style="bold", width=6) + table.add_column("Nome", width=6) + table.add_column("Stato", width=5, justify="center") + table.add_column("Porte Membri", width=14) + table.add_column("PVID", width=4, justify="center") + table.add_column("Tagged", width=18) + table.add_column("Mirroring", width=12) + + for lag_id in sorted(self.config.lags.keys(), key=lambda x: int(x.split("/")[1])): + lag = self.config.lags[lag_id] + + is_up = lag.link_state == "Up" + is_source = self.parser.is_mirroring_source(lag_id) if self.parser else False + + # Filter based on mode + if only_active and not is_up and not is_source: + continue + if not only_active and not lag.member_ports and not is_up: + continue + + # State styling + if is_up: + state_text = Text("UP", style="bold green") + else: + state_text = Text("DOWN", style="dim") + + # Mirroring status + if is_source: + mirror_text = Text("▶▶ SRC", style="bold cyan") + else: + mirror_text = Text("-", style="dim") + + # Get VLAN from LAG interface config + lag_iface = self.config.interfaces.get(f"lag{lag_id.split('/')[1]}") + if lag_iface: + pvid = str(lag_iface.pvid) + tagged = ConfigParser.format_vlan_list(lag_iface.tagged_vlans) + if len(tagged) > 18: + tagged = tagged[:15] + "..." + if not tagged: + tagged = "-" + else: + pvid = "1" + tagged = "-" + + table.add_row( + lag_id, + lag.name, + state_text, + ",".join(lag.member_ports) or "-", + pvid, + tagged, + mirror_text + ) + + self.console.print(table) + + # ========================================================================= + # Mirroring status view + # ========================================================================= + + def show_mirroring_status(self) -> None: + """Display current mirroring configuration in a clear format.""" + self.print_header("CONFIGURAZIONE MIRRORING ATTUALE") + + if not self.config: + self.console.print("[red]Configurazione non caricata[/red]") + self.wait_for_enter() + return + + if not self.config.monitor_sessions: + self.console.print(Panel( + "[yellow]Nessuna sessione di mirroring configurata[/yellow]", + title="Monitor Session", + box=box.ROUNDED + )) + self.wait_for_enter() + return + + for session_id, session in self.config.monitor_sessions.items(): + # ===== DESTINATION BOX ===== + if session.destination: + dest_iface = self.config.interfaces.get(session.destination) + dest_desc = dest_iface.description if dest_iface else "" + dest_state = "" + if dest_iface: + dest_state = "[green]UP[/green]" if dest_iface.link_state == "Up" else "[dim]DOWN[/dim]" + + dest_content = f"[bold]{session.destination}[/bold] {dest_state}\n{dest_desc}" + + self.console.print(Panel( + dest_content, + title="◀◀ DESTINATION (porta di monitoraggio)", + box=box.DOUBLE, + style="magenta", + width=60 + )) + else: + self.console.print(Panel( + "[yellow]Non configurata[/yellow]", + title="DESTINATION", + box=box.ROUNDED, + width=60 + )) + + self.console.print() + + # ===== SOURCES TABLE ===== + if session.sources: + table = Table( + title=f"▶▶ SOURCE PORTS ({len(session.sources)} porte monitorate)", + box=box.ROUNDED, + show_lines=False, + width=60 + ) + + table.add_column("Porta", style="bold cyan", width=8) + table.add_column("Stato", width=6, justify="center") + table.add_column("Descrizione", width=36) + + # Sort sources + sources_sorted = sorted(session.sources, key=self._get_port_sort_key) + + for src in sources_sorted: + src_iface = self.config.interfaces.get(src) + + if src_iface: + state = Text("UP", style="green") if src_iface.link_state == "Up" else Text("DOWN", style="dim") + desc = src_iface.description or "" + else: + # Check if it's a LAG + lag = self.config.lags.get(src) + if lag: + state = Text("UP", style="green") if lag.link_state == "Up" else Text("DOWN", style="dim") + desc = f"LAG {lag.name} ({','.join(lag.member_ports)})" + else: + state = Text("-", style="dim") + desc = "" + + table.add_row(src, state, desc[:36]) + + self.console.print(table) + else: + self.console.print("[yellow]Nessuna porta source configurata[/yellow]") + + self.wait_for_enter() + + # ========================================================================= + # Mirroring configuration with checkbox style + # ========================================================================= + + def configure_mirroring(self) -> Optional[Tuple[str, List[str]]]: + """ + Interactive mirroring configuration with checkbox-style selection. + Returns (destination, sources) or None if cancelled. + """ + if not self.config or not self.parser: + self.console.print("[red]Configurazione non caricata[/red]") + self.wait_for_enter() + return None + + # Initialize with current config + current_dest = self.parser.get_mirroring_destination() + current_sources = set(self.parser.get_mirroring_sources()) + + selected_dest = current_dest + selected_sources = current_sources.copy() + + # Get all available ports + all_ports = self._get_all_available_ports() + + while True: + self.clear_screen() + self.print_header("CONFIGURA PORT MIRRORING") + + # ===== DESTINATION SECTION ===== + self.console.print("[bold]STEP 1: PORTA DESTINATION[/bold] (riceve il traffico mirrorato)") + self.console.print() + + if selected_dest: + dest_iface = self.config.interfaces.get(selected_dest) + dest_desc = dest_iface.description if dest_iface else "" + self.console.print(Panel( + f"[bold magenta]{selected_dest}[/bold magenta] - {dest_desc}", + title="◀◀ DESTINATION SELEZIONATA", + box=box.ROUNDED, + width=60 + )) + else: + self.console.print(Panel( + "[yellow]Nessuna destination selezionata[/yellow]", + title="DESTINATION", + box=box.ROUNDED, + width=60 + )) + + self.console.print() + + # ===== SOURCES SECTION with checkboxes ===== + self.console.print("[bold]STEP 2: PORTE SOURCE[/bold] (traffico da monitorare)") + self.console.print("[dim]Seleziona/deseleziona inserendo il numero della porta[/dim]") + self.console.print() + + # Create checkbox table + self._show_checkbox_table(all_ports, selected_dest, selected_sources) + + self.console.print() + + # ===== MENU OPTIONS ===== + self.console.print("[bold]Comandi:[/bold]") + self.console.print(" [D] Cambia DESTINATION") + self.console.print(" [numero] Toggle porta source (es: 4 per 0/4)") + self.console.print(" [A] Seleziona TUTTE le porte come source") + self.console.print(" [N] Deseleziona tutte le porte source") + self.console.print(" [R] Range porte (es: 17-24)") + self.console.print(" [C] CONFERMA e salva") + self.console.print(" [X] Annulla ed esci") + self.console.print() + + choice = Prompt.ask("Comando").strip().upper() + + if choice == "X": + return None + + elif choice == "C": + # Confirm + if not selected_dest: + self.console.print("[red]Devi selezionare una porta DESTINATION[/red]") + self.wait_for_enter() + continue + + if not selected_sources: + self.console.print("[red]Devi selezionare almeno una porta SOURCE[/red]") + self.wait_for_enter() + continue + + # Show confirmation + return self._show_config_confirmation(selected_dest, list(selected_sources)) + + elif choice == "D": + # Change destination + new_dest = self._select_destination(all_ports, selected_dest) + if new_dest: + # Remove from sources if was selected + selected_sources.discard(new_dest) + selected_dest = new_dest + + elif choice == "A": + # Select all + for port_id, _, _ in all_ports: + if port_id != selected_dest: + selected_sources.add(port_id) + + elif choice == "N": + # Deselect all + selected_sources.clear() + + elif choice == "R": + # Range input + range_str = Prompt.ask("Inserisci range (es: 17-24 o 0/17-0/24)") + ports_in_range = self._parse_port_range(range_str) + for p in ports_in_range: + if p != selected_dest and p in {x[0] for x in all_ports}: + selected_sources.add(p) + + else: + # Try to parse as port number for toggle + port_id = self._normalize_port_id(choice) + if port_id and port_id in {p[0] for p in all_ports}: + if port_id == selected_dest: + self.console.print("[yellow]Non puoi selezionare la destination come source[/yellow]") + self.wait_for_enter() + elif port_id in selected_sources: + selected_sources.remove(port_id) + else: + selected_sources.add(port_id) + + def _get_all_available_ports(self) -> List[Tuple[str, str, str]]: + """Get all available ports for selection.""" + ports = [] + + # Physical ports + for port_id in self.parser.get_physical_ports(): + iface = self.config.interfaces.get(port_id) + if iface: + ports.append((port_id, iface.description or "", iface.link_state)) + + # LAG ports + for lag_id in self.parser.get_lag_ports(): + lag = self.config.lags.get(lag_id) + if lag and (lag.member_ports or lag.link_state == "Up"): + desc = f"LAG {lag.name} ({','.join(lag.member_ports)})" + ports.append((lag_id, desc, lag.link_state)) + + return ports + + def _show_checkbox_table( + self, + ports: List[Tuple[str, str, str]], + dest: Optional[str], + sources: Set[str] + ) -> None: + """Show ports with checkbox-style selection.""" + table = Table(box=box.ROUNDED, show_lines=False, padding=(0, 1)) + + table.add_column("", width=3, justify="center") # Checkbox + table.add_column("Porta", style="bold", width=6) + table.add_column("Descrizione", width=24) + table.add_column("Stato", width=5, justify="center") + table.add_column("PVID", width=4, justify="center") + table.add_column("Tagged", width=18) + table.add_column("Ruolo", width=12) + + for port_id, desc, state in ports: + # Checkbox + if port_id == dest: + checkbox = Text("◀◀", style="bold magenta") + role = Text("DESTINATION", style="bold magenta") + row_style = "on dark_magenta" + elif port_id in sources: + checkbox = Text("[X]", style="bold cyan") + role = Text("SOURCE", style="cyan") + row_style = "" + else: + checkbox = Text("[ ]", style="dim") + role = Text("-", style="dim") + row_style = "dim" if state == "Down" else "" + + # State + state_text = Text("UP", style="green") if state == "Up" else Text("DOWN", style="dim") + + # VLAN info + iface = self.config.interfaces.get(port_id) + if iface: + pvid = str(iface.pvid) + tagged = ConfigParser.format_vlan_list(iface.tagged_vlans) + if len(tagged) > 18: + tagged = tagged[:15] + "..." + if not tagged: + tagged = "-" + else: + pvid = "1" + tagged = "-" + + table.add_row( + checkbox, + port_id, + desc[:24], + state_text, + pvid, + tagged, + role, + style=row_style + ) + + self.console.print(table) + + # Stats + self.console.print() + self.console.print(f"[dim]Porte source selezionate: [bold]{len(sources)}[/bold][/dim]") + + def _select_destination( + self, + ports: List[Tuple[str, str, str]], + current: Optional[str] + ) -> Optional[str]: + """Select destination port.""" + self.clear_screen() + self.print_header("SELEZIONA DESTINATION") + + self.console.print("Seleziona la porta che riceverà il traffico mirrorato:") + self.console.print() + + table = Table(box=box.ROUNDED, show_lines=False) + table.add_column("Porta", style="bold", width=8) + table.add_column("Descrizione", width=32) + table.add_column("Stato", width=6) + + for port_id, desc, state in ports: + style = "bold magenta" if port_id == current else "" + state_text = "UP" if state == "Up" else "DOWN" + state_style = "green" if state == "Up" else "dim" + + table.add_row( + Text(port_id, style=style), + Text(desc[:32], style=style), + Text(state_text, style=state_style) + ) + + self.console.print(table) + self.console.print() + + choice = Prompt.ask( + "Inserisci porta (es: 0/3) o X per annullare", + default=current or "" + ) + + if choice.upper() == "X": + return None + + port_id = self._normalize_port_id(choice) + if port_id and port_id in {p[0] for p in ports}: + return port_id + + self.console.print(f"[red]Porta non valida: {choice}[/red]") + self.wait_for_enter() + return current + + def _show_config_confirmation( + self, + dest: str, + sources: List[str] + ) -> Optional[Tuple[str, List[str]]]: + """Show configuration confirmation screen.""" + self.clear_screen() + self.print_header("CONFERMA CONFIGURAZIONE") + + # Destination box + dest_iface = self.config.interfaces.get(dest) + dest_desc = dest_iface.description if dest_iface else "" + + self.console.print(Panel( + f"[bold]{dest}[/bold]\n{dest_desc}", + title="◀◀ DESTINATION", + box=box.DOUBLE, + style="magenta", + width=50 + )) + + self.console.print() + + # Sources table + table = Table( + title=f"▶▶ SOURCE PORTS ({len(sources)})", + box=box.ROUNDED, + width=50 + ) + table.add_column("Porta", style="cyan", width=8) + table.add_column("Descrizione", width=32) + + sources_sorted = sorted(sources, key=self._get_port_sort_key) + for src in sources_sorted: + src_iface = self.config.interfaces.get(src) + desc = src_iface.description if src_iface else "" + if not desc and src.startswith("3/"): + lag = self.config.lags.get(src) + if lag: + desc = f"LAG {lag.name}" + table.add_row(src, desc[:32]) + + self.console.print(table) + + self.console.print() + + # Commands preview + commands = self.generate_mirroring_commands(dest, sources) + self.console.print(Panel( + "\n".join(commands), + title="COMANDI DA ESEGUIRE", + box=box.ROUNDED, + style="dim" + )) + + self.console.print() + + if Confirm.ask("[bold]Confermare questa configurazione?[/bold]", default=True): + self.selected_destination = dest + self.selected_sources = set(sources) + return (dest, sources) + + return None + + def _normalize_port_id(self, port_str: str) -> Optional[str]: + """Normalize port ID input.""" + port_str = port_str.strip() + + # Handle formats: 0/3, 03, 3 (for physical), 3/1 (for LAG) + if "/" in port_str: + return port_str + + # Try to parse as number for physical port + try: + num = int(port_str) + return f"0/{num}" + except ValueError: + return None + + def _parse_port_range(self, range_str: str) -> List[str]: + """ + Parse port range string like "0/4,0/8,0/17-0/24,3/1". + Returns list of port IDs. + """ + ports = [] + + if not range_str: + return ports + + parts = range_str.replace(" ", "").split(",") + + for part in parts: + if "-" in part: + # Range: 0/17-0/24 or 17-24 + try: + start_str, end_str = part.split("-", 1) + + # Parse start + if "/" in start_str: + prefix, start_num = start_str.rsplit("/", 1) + start = int(start_num) + else: + prefix = "0" + start = int(start_str) + + # Parse end + if "/" in end_str: + _, end_num = end_str.rsplit("/", 1) + end = int(end_num) + else: + end = int(end_str) + + for i in range(start, end + 1): + ports.append(f"{prefix}/{i}") + + except ValueError: + continue + else: + # Single port + normalized = self._normalize_port_id(part) + if normalized: + ports.append(normalized) + + return ports + + # ========================================================================= + # Command preview + # ========================================================================= + + def show_command_preview(self, destination: str, sources: List[str], session_id: int = 1) -> List[str]: + """Show preview of commands to be executed.""" + commands = self.generate_mirroring_commands(destination, sources, session_id) + + self.console.print(Panel( + "\n".join(commands), + title="COMANDI DA ESEGUIRE", + box=box.ROUNDED, + style="dim" + )) + + return commands + + def generate_mirroring_commands( + self, + destination: str, + sources: List[str], + session_id: int = 1 + ) -> List[str]: + """Generate list of mirroring configuration commands.""" + commands = [] + + # Clear existing session + commands.append(f"no monitor session {session_id}") + + # Set destination + commands.append(f"monitor session {session_id} destination interface {destination}") + + # Add sources + for src in sorted(sources, key=lambda x: ( + 0 if x.startswith("0/") else 1, + int(x.split("/")[1]) if "/" in x else 0 + )): + commands.append(f"monitor session {session_id} source interface {src}") + + # Enable session + commands.append(f"monitor session {session_id} mode") + + return commands + + +# ============================================================================= +# Testing +# ============================================================================= + +if __name__ == "__main__": + ui = PortMirroringUI() + + # Test with mock data + from config_parser import ConfigParser + + sample_config = """ +!System Description "USW-Pro-Aggregation, 7.1.26.15869" +snmp-server sysname "SW-CED-AGG-243" +network parms 192.168.40.243 255.255.255.0 192.168.40.254 + +interface 0/1 +description 'SFP_ 1' +exit + +interface 0/2 +description 'SFP_ 2' +exit + +interface 0/3 +description 'SFP_ 3 - MONITORING PORT' +exit + +interface 0/4 +description 'SFP_ 4 - SW CED 237' +exit + +monitor session 1 destination interface 0/3 +monitor session 1 source interface 0/4 +monitor session 1 mode +""" + + sample_status = """ +Port Name State Mode Status Type +--------- ---------------------------- ------ ---------- ---------- ------------------ +0/1 SFP_ 1 Down Auto D Unknown +0/2 SFP_ 2 Down 10G Full Unknown +0/3 SFP_ 3 - MONITORING PORT Up Auto D 10G Full DAC +0/4 SFP_ 4 - SW CED 237 Up 10G Full 10G Full DAC +""" + + parser = ConfigParser() + parser.parse_running_config(sample_config) + parser.parse_interfaces_status(sample_status) + + ui.set_config(parser) + ui.set_device_info("192.168.40.243", "admin") + + # Show menu + choice = ui.show_main_menu() + print(f"Selected: {choice}")