from netmiko import ConnectHandler import ipaddress, json, re class Switch_Praser: def __init__(self, ip:str, user:str, password:str, port:int, vrf:str, vendor:str, action:str, prefix:str): self.ip = ip self.user = user self.password = password self.port = port self.vrf = vrf self.vendor = vendor self.action = action self.prefix = prefix # If prefix exist in switch fib def get_prefix_exist(self): output = "" command = "" if self.vendor == "ciscoNxosSsh": device = { "device_type": "cisco_nxos", "host": self.ip, "username": self.user, "password": self.password, "port": self.port } switch = ConnectHandler(**device) if ipaddress.ip_network(self.prefix).version == 4: command = f"show ip route {self.prefix} vrf {self.vrf} | json" elif ipaddress.ip_network(self.prefix).version == 6: command = f"show ipv6 route {self.prefix} vrf {self.vrf} | json" output = switch.send_command(command) switch.disconnect() if "TABLE_prefix" in json.loads(output)["TABLE_vrf"]["ROW_vrf"]["TABLE_addrf"]["ROW_addrf"]: return True else: return False # JunOS under development def get_if_ipv6(self): output = "" command:str = "" address:list = [] if self.vendor == "ciscoNxosSsh": device = { "device_type": "cisco_nxos", "host": self.ip, "username": self.user, "password": self.password, "port": self.port } switch = ConnectHandler(**device) command = f"show ipv6 interface {self.port} vrf {self.vrf} | json" for line in json.loads(switch.send_command(command))["TABLE_intf"]["ROW_intf"]["TABLE_addr"]["ROW_addr"]: address.append(line["addr"]) switch.disconnect() return address # JunOS under development def add_route(self): command:list = [] if self.vendor == "ciscoNxosSsh": device = { "device_type": "cisco_nxos", "host": self.ip, "username": self.user, "password": self.password, "port": self.port } switch = ConnectHandler(**device) if ipaddress.ip_network(self.prefix).version == 4: command = [ f"vrf context {self.vrf}", f"ip route {self.prefix} {self.port}" ] elif ipaddress.ip_network(self.prefix).version == 6: if_addrs = self.get_if_ipv6() for line in if_addrs: if re.match(r'^\s*(?:(?:[0-9A-Fa-f]{1,4}:){7}[0-9A-Fa-f]{1,4}|(?:[0-9A-Fa-f]{1,4}:){1,7}:|(?:[0-9A-Fa-f]{1,4}:){1,6}:[0-9A-Fa-f]{1,4}|(?:[0-9A-Fa-f]{1,4}:){1,5}(?::[0-9A-Fa-f]{1,4}){1,2}|(?:[0-9A-Fa-f]{1,4}:){1,4}(?::[0-9A-Fa-f]{1,4}){1,3}|(?:[0-9A-Fa-f]{1,4}:){1,3}(?::[0-9A-Fa-f]{1,4}){1,4}|(?:[0-9A-Fa-f]{1,4}:){1,2}(?::[0-9A-Fa-f]{1,4}){1,5}|[0-9A-Fa-f]{1,4}:(?:(?::[0-9A-Fa-f]{1,4}){1,6})|:(?:(?::[0-9A-Fa-f]{1,4}){1,7}|:))(?:%[0-9A-Za-z]{1,})?\/127\s*$', line) != None: ptp = ipaddress.ip_network(line)[1] break command = [ f"vrf context {self.vrf}", f"ipv6 route {self.prefix} {ptp}" ] switch.send_config_set(command) switch.disconnect() def del_route(self): output = "" command:list = [] if self.vendor == "ciscoNxosSsh": device = { "device_type": "cisco_nxos", "host": self.ip, "username": self.user, "password": self.password, "port": self.port } switch = ConnectHandler(**device) if ipaddress.ip_network(self.prefix).version == 4: command = [ f"vrf context {self.vrf}", f"no ip route {self.prefix} {self.port}" ] elif ipaddress.ip_network(self.prefix).version == 6: output = switch.send_command(f"show ipv6 route {self.prefix} vrf {self.vrf} | json") if "TABLE_prefix" in json.loads(output)["TABLE_vrf"]["ROW_vrf"]["TABLE_addrf"]["ROW_addrf"]: next_hop = json.loads(output)["TABLE_vrf"]["ROW_vrf"]["TABLE_addrf"]["ROW_addrf"]["TABLE_prefix"]["ROW_prefix"]["TABLE_path"]["ROW_path"]["ipnexthop"] command = [ f"vrf context {self.vrf}", f"no ipv6 route {self.prefix} {next_hop}" ] switch.send_config_set(command) switch.disconnect()