Поиск по MAC-адресу на коммутаторах Juniper
В локальной сети часто нужно узнать, на каком порту коммутатора находится определенный MAC-адрес устройства. Задача решается легко, если в сети несколько коммутаторов, но, когда их больше 30, всё значительно усложняется. Хочу поделиться небольшим скриптом на Python, который ищет нужный MAC-адрес в сети и возвращает имя и порт коммутатора, на котором зарегистрирован этот MAC.
Конструктивная критика приветствуется. Подробности под катом.
Если дизайн сети выполнен правильно, то есть корневой коммутатор CORE, к которому подключены коммутаторы распределения DS (Distribution Switch), а к ним, в свою очередь, коммутаторы уровня доступа AS (Access Switch). Это правило не всегда выполняется, коммутаторы доступа могут быть подключены последовательно. В любом случае, на порту вышестоящего коммутатора находятся все MAC-адреса устройств, подключенных к нижестоящему коммутатору.
Например, если интересующее нас устройство подключено к коммутатору AS3, то, начав поиск с CORE, мы найдем этот адрес на порту, ведущему к DS1. Зайдя на DS1, мы обнаружим этот MAC на порту, ведущему к AS2, зайдя на AS2, мы увидим, что он ведет нас к AS3, и только на AS3 мы найдем конкретный порт, к которому подключено интересующее нас устройство.
Делать это всё руками не хотелось, перебирать все коммутаторы в цикле и определять, где аплинк, а где нет тоже, поэтому родилось следующее решение, которым и хочу поделиться.
Немного теории.
Чтобы найти MAC 08:62:66: c7: b3:45 на коммутаторе Juniper, нужно выполнить следующую команду:
show ethernet-switching table | match 08:62:66:c7:b3:45
Если такой MAC есть, ответ будет следующим:
vlan151 08:62:66:c7:b3:45 D - xe-0/0/23.0
В последней колонке будет имя интерфейса коммутатора, на котором зарегистрирован MAC. Но как понять, куда ведет этот интерфейс? И тут на помощь приходят Interface Descriptions. Это строки в конфигурационном файле коммутатора, которые позволяют назначить текстовые метки интерфейсам.
Команда
show interfaces xe-0/0/23 descriptions
покажет следующее:
Interface Admin Link Description
xe-0/0/23 up up SW>DS1
При конфигурации мы указываем, что этот интерфейс ведет к нижестоящему коммутатору:
set interfaces xe-0/0/23 description SW>DS1
Реализация
Предлагаемый скрипт будет делать следующее:
- подключаться по SSH на корневой коммутатор;
- проверять, на каком интерфейсе находится передаваемый в параметрах MAC-адрес;
- проверять Description этого интерфейса;
- если интерфейс ведет к коммутатору, рекурсивно заходить на следующий по цепочке коммутатор.
#в этот список будут складываться результаты поиска
searchpass = []
#main функция принимает в качестве параметра MAC-адрес и вызывает функцию checkswitch, в которую в качестве параметра передает имя корневого коммутатора и MAC-адрес. Весь результат поиска складывается в список searchpass в формате json.
def main(argv):
mac_addr = argv[0]
checkswitch('CORE',mac_addr)
for switch in searchpass:
print (json.dumps(switch, ensure_ascii=False))
if __name__ == "__main__":
main(sys.argv[1:])
#функция рекурсивного поиска MAC-адреса
def checkswitch(hostname,mac_addr):
try:
#создаем пустой словарь возвращаемых значений, ключу host присваиваем имя коммутатора
returnvalue = {}
returnvalue['host']=hostname
#sendCommand подключается к заданному коммутатору по SSH и вызывает команду поиска MAC-адреса в таблице коммутации
answer = sendCommand(hostname,'show ethernet-switching table | match '+mac_addr)
#делим строку на колонки, в последней находится имя интерфейса
#vlan151 08:62:66:c7:b3:45 D - xe-0/0/23.0
if(answer!=0):
iface = answer.split()[4]
returnvalue['iface']=iface
#проверяем description интерфейса, нужно отрезать последние 2 символа .0 от имени и забрать последнюю строку
#xe-0/0/23 up up SW>DS01
answer = sendCommand(hostname,'show interfaces '+iface[:-2]+' descriptions | last 1 | no-more')
iface = answer.split()
#Если description на интерфейсе есть, записываем его в словарь и проверяем, начинается ли он с SW>. Если да, отрезаем эти 3 символа, берем имя следующего коммутатора и рекурсивно вызываем функцию снова.
if(len(iface)>2):
iface=iface[3]
returnvalue['description']=iface
else:
returnvalue['description']='none'
searchpass.append(returnvalue)
if (iface[:3]=='SW>'):
checkswitch(iface[3:],mac_addr)
else:
returnvalue['iface']='none'
searchpass.append(returnvalue)
except Exception as e:
print(e)
Таким образом, скрипт пройдет по всем коммутаторам сети, начиная с ядра, и попытается найти нужный MAC. Для успешной работы достаточно поддерживать в актуальном состоянии descriptions на интерфейсах, а топология может быть практически любой сложности.
Пример работы скрипта:
python findmac.py 00:17:fc:21:e8:f9
{"host": "CORE", "iface": "xe-0/0/23.0", "description": "SW>DS1"}
{"host": "DS1", "iface": "xe-0/0/11.0", "description": "SW>AS2"}
{"host": "AS2", "iface": "xe-1/0/1.0", "description": "SW>AS3"}
{"host": "AS3", "iface": "ge-0/0/26.0", "description": "none"}
Если MAC отсутствует, получим
{"host": "CORE", "iface": "none"}
Последняя строка — интересующий нас коммутатор и порт, но при этом мы можем отследить весь путь поиска.
Полный код — под спойлером, благодарю за внимание.
import paramiko
import time
import sys
import json
import threading
import logging
login = 'user1'
password = 'pass1234'
searchpass = []
port = 22
class LogPipe(threading.Thread):
def __init__(self, level):
threading.Thread.__init__(self)
self.daemon = False
self.level = level
self.fdRead, self.fdWrite = os.pipe()
self.pipeReader = os.fdopen(self.fdRead)
self.start()
def fileno(self):
return self.fdWrite
def run(self):
for line in iter(self.pipeReader.readline, ''):
logging.log(self.level, line.strip('\n'))
self.pipeReader.close()
def close(self):
os.close(self.fdWrite)
def execute_ssh_command(host, port, username, password, command):
try:
# Create the SSH client.
ssh = paramiko.SSHClient()
ssh.set_missing_host_key_policy(paramiko.AutoAddPolicy())
# Connect to the host.
ssh.connect(host, port, username, password, look_for_keys=False)
# Send the command (non-blocking)
stdin, stdout, stderr = ssh.exec_command(command)
# Wait for the command to terminate
while not stdout.channel.exit_status_ready() and not stdout.channel.recv_ready():
time.sleep(1)
stdoutstring = stdout.readlines()
stderrstring = stderr.readlines()
return stdoutstring, stderrstring
finally:
if ssh is not None:
# Close client connection.
ssh.close()
def sendCommand (hostname,command):
returnvalue = 0
logging.info('Host '+hostname+', command: '+command)
Try:
#add .mydomain for FQDN
(stdoutstring, stderrstring) = execute_ssh_command(hostname+'.mydomain', port, login, password, command+'\n')
if (len(stdoutstring)>0):
logging.info(stdoutstring[0])
if (len(stderrstring)>0):
logging.info(stderrstring[0])
except Exception as e:
return returnvalue
else:
returnvalue = stdoutstring[0]
finally:
return returnvalue
def checkswitch(hostname,mac_addr):
try:
returnvalue = {}
returnvalue['host']=hostname
answer = sendCommand(hostname,'show ethernet-switching table | match '+mac_addr)
if(answer!=0):
iface = answer.split()[4]
returnvalue['iface']=iface
#cut .0 prefix in the interface name
answer = sendCommand(hostname,'show interfaces '+iface[:-2]+' descriptions | last 1 | no-more')
iface = answer.split()
if(len(iface)>2):
iface=iface[3]
returnvalue['description']=iface
else:
returnvalue['description']='none'
searchpass.append(returnvalue)
if (iface[:3]=='SW>'):
checkswitch(iface[3:],mac_addr)
else:
returnvalue['iface']='none'
searchpass.append(returnvalue)
except Exception as e:
logging.info(e)
def main(argv):
mac_addr = argv[0]
#configure log
logging.basicConfig(filename='/var/log/findmac.log', level=logging.INFO, format='%(asctime)s %(message)s')
logging.info('Find MAC: '+mac_addr)
checkswitch('CORE',mac_addr)
for switch in searchpass:
print (json.dumps(switch, ensure_ascii=False))
if __name__ == "__main__":
main(sys.argv[1:])
