import logging from typing import Dict, List, Optional from librouteros import connect from librouteros.exceptions import TrapError, FatalError logger = logging.getLogger(__name__) class MikroTikManager: """MikroTik RouterOS API Manager""" def __init__(self): self.connections = {} def _get_connection(self, ip: str, username: str, password: str, port: int = 8728): """Get or create connection to MikroTik device""" key = f"{ip}:{port}" if key not in self.connections: try: api = connect( host=ip, username=username, password=password, port=port, timeout=10 ) self.connections[key] = api logger.info(f"Connected to MikroTik {ip}") except Exception as e: logger.error(f"Failed to connect to MikroTik {ip}: {e}") raise return self.connections[key] def close_connection(self, ip: str, port: int = 8728): """Close connection to MikroTik device""" key = f"{ip}:{port}" if key in self.connections: try: self.connections[key].close() del self.connections[key] except: pass def get_system_resource(self, ip: str, username: str, password: str) -> Dict: """Get system resources (CPU, Memory, etc.)""" try: api = self._get_connection(ip, username, password) resource = api.path('/system/resource') data = list(resource)[0] return { 'platform': data.get('platform', 'Unknown'), 'board_name': data.get('board-name', 'Unknown'), 'version': data.get('version', 'Unknown'), 'cpu': data.get('cpu', 'Unknown'), 'cpu_count': int(data.get('cpu-count', 0)), 'cpu_load': int(data.get('cpu-load', 0)), 'uptime': data.get('uptime', 'Unknown'), 'free_memory': int(data.get('free-memory', 0)), 'total_memory': int(data.get('total-memory', 0)), 'free_hdd_space': int(data.get('free-hdd-space', 0)), 'total_hdd_space': int(data.get('total-hdd-space', 0)), } except Exception as e: logger.error(f"Failed to get MikroTik system resource: {e}") raise def get_interfaces(self, ip: str, username: str, password: str) -> List[Dict]: """Get all interfaces""" try: api = self._get_connection(ip, username, password) interfaces_path = api.path('/interface') interfaces = [] for interface in interfaces_path: interfaces.append({ 'id': interface.get('.id', ''), 'name': interface.get('name', ''), 'type': interface.get('type', ''), 'mtu': int(interface.get('mtu', 0)), 'running': interface.get('running', 'false') == 'true', 'disabled': interface.get('disabled', 'false') == 'true', 'comment': interface.get('comment', ''), }) return interfaces except Exception as e: logger.error(f"Failed to get MikroTik interfaces: {e}") raise def get_ip_addresses(self, ip: str, username: str, password: str) -> List[Dict]: """Get IP addresses""" try: api = self._get_connection(ip, username, password) addresses_path = api.path('/ip/address') addresses = [] for addr in addresses_path: addresses.append({ 'id': addr.get('.id', ''), 'address': addr.get('address', ''), 'network': addr.get('network', ''), 'interface': addr.get('interface', ''), 'disabled': addr.get('disabled', 'false') == 'true', }) return addresses except Exception as e: logger.error(f"Failed to get MikroTik IP addresses: {e}") raise def get_dhcp_server_leases(self, ip: str, username: str, password: str) -> List[Dict]: """Get DHCP server leases""" try: api = self._get_connection(ip, username, password) leases_path = api.path('/ip/dhcp-server/lease') leases = [] for lease in leases_path: leases.append({ 'id': lease.get('.id', ''), 'address': lease.get('address', ''), 'mac_address': lease.get('mac-address', ''), 'client_id': lease.get('client-id', ''), 'hostname': lease.get('host-name', ''), 'server': lease.get('server', ''), 'status': lease.get('status', ''), 'expires_after': lease.get('expires-after', ''), }) return leases except Exception as e: logger.error(f"Failed to get DHCP leases: {e}") raise def get_pppoe_active(self, ip: str, username: str, password: str) -> List[Dict]: """Get active PPPoE sessions""" try: api = self._get_connection(ip, username, password) pppoe_path = api.path('/ppp/active') sessions = [] for session in pppoe_path: sessions.append({ 'id': session.get('.id', ''), 'name': session.get('name', ''), 'service': session.get('service', ''), 'caller_id': session.get('caller-id', ''), 'address': session.get('address', ''), 'uptime': session.get('uptime', ''), 'encoding': session.get('encoding', ''), }) return sessions except Exception as e: logger.error(f"Failed to get PPPoE sessions: {e}") raise def get_queue_simple(self, ip: str, username: str, password: str) -> List[Dict]: """Get simple queues""" try: api = self._get_connection(ip, username, password) queues_path = api.path('/queue/simple') queues = [] for queue in queues_path: queues.append({ 'id': queue.get('.id', ''), 'name': queue.get('name', ''), 'target': queue.get('target', ''), 'max_limit': queue.get('max-limit', ''), 'disabled': queue.get('disabled', 'false') == 'true', 'bytes_in': queue.get('bytes', '0/0').split('/')[0], 'bytes_out': queue.get('bytes', '0/0').split('/')[1] if '/' in queue.get('bytes', '') else '0', }) return queues except Exception as e: logger.error(f"Failed to get queues: {e}") raise def get_firewall_filter(self, ip: str, username: str, password: str) -> List[Dict]: """Get firewall filter rules""" try: api = self._get_connection(ip, username, password) rules_path = api.path('/ip/firewall/filter') rules = [] for rule in rules_path: rules.append({ 'id': rule.get('.id', ''), 'chain': rule.get('chain', ''), 'action': rule.get('action', ''), 'protocol': rule.get('protocol', ''), 'src_address': rule.get('src-address', ''), 'dst_address': rule.get('dst-address', ''), 'comment': rule.get('comment', ''), 'disabled': rule.get('disabled', 'false') == 'true', 'bytes': int(rule.get('bytes', 0)), 'packets': int(rule.get('packets', 0)), }) return rules except Exception as e: logger.error(f"Failed to get firewall rules: {e}") raise def create_backup(self, ip: str, username: str, password: str, name: Optional[str] = None) -> Dict: """Create device backup""" try: import time from datetime import datetime api = self._get_connection(ip, username, password) backup_name = name or f"backup_{datetime.now().strftime('%Y%m%d_%H%M%S')}" # Create backup system_path = api.path('/system/backup/save') system_path('name', backup_name) # Wait for backup to complete time.sleep(2) # Get backup file info files_path = api.path('/file') for file in files_path: if file.get('name', '').startswith(backup_name): return { 'name': file.get('name', ''), 'size': int(file.get('size', 0)), 'creation_time': file.get('creation-time', ''), 'success': True } return {'success': False, 'error': 'Backup file not found'} except Exception as e: logger.error(f"Failed to create MikroTik backup: {e}") raise def execute_command(self, ip: str, username: str, password: str, command: str) -> List[Dict]: """Execute custom command""" try: api = self._get_connection(ip, username, password) # Parse command path path = command.strip('/') result_path = api.path(f'/{path}') results = list(result_path) return results except Exception as e: logger.error(f"Failed to execute command: {e}") raise # Global instance mikrotik_manager = MikroTikManager()