Протоколы SSH Telnet в Pyton. лаба15. Подключение к оборудованию
 Скачать 27.69 Kb.
|
|
Подключение к оборудованию Подключение к оборудованию по протоколам: SSH Telnet В Python есть несколько модулей, которые позволяют подключаться к оборудованию и выполнять команды. Используем модуль scrapli , который позволяет подключаться к сетевому оборудованию используя Telnet, SSH или NETCONF. Установка scrapli: pip install scrapli Три основные составляющие части scrapli: transport - это конкретный способ подключения к оборудованию channel - следующий уровень над транспортом, который отвечает за отправку команд, получение вывода и другими взаимодействиями с оборудованием driver - это интерфейс, который предоставляется пользователю для работы со scrapli. Тут есть как специфические драйверы типа IOSXEDriver, который понимает как взаимодействовать с оборудованием конкретного типа, так и базовый драйвер Driver, который предоставляет минимальный интерфейс для работы через SSH/Telnet. Доступные варианты транспорта: system - используется встроенный SSH клиент, подразумевается использование клиента на Linux/MacOS paramiko - модуль paramiko ssh2 - используется модуль ssh2-python (обертка вокруг C библиотеки libssh2) telnet - будет использоваться telnetlib asyncssh - модуль asyncssh asynctelnet - telnet клиент написанный с использованием asyncio Основные примеры будут с использованием транспорта system. В scrapli глобально есть два варианта подключения: используя общий класс Scrapli, который выбирает нужный driver по параметру platform или конкретный driver, например, IOSXEDriver. При этом параметры передаются те же самые и конкретному драйверу и Scrapli. Кроме этих вариантов, есть также общие (базовые) драйверы. Если в scrapli (или scrapli community) нет поддержки необходимой платформы, можно добавить платформу в scrapli community или использовать: Driver GenericDriver NetworkDriver Параметры подключения Основные параметры подключения: host - IP-адрес или имя хоста auth_username - имя пользователя auth_password - пароль auth_secondary - пароль на enable auth_strict_key - контролирует проверку SSH ключей сервера, а именно разрешать ли подключаться к серверам ключ которых не сохранен в ssh/known_hosts. False - разрешить подключение (по умолчанию значение True) platform - нужно указывать при использовании Scrapli transport - какой транспорт использовать transport_options - опции для конкретного транспорта При подключении без менеджера контекста, сначала надо передать параметры драйверу или Scrapli, а затем вызвать метод open: from scrapli import Scrapli r1 = { "host": "192.168.100.1", "auth_username": "cisco", "auth_password": "cisco", "auth_secondary": "cisco", "auth_strict_key": False, "platform": "cisco_iosxe" } In [2]: ssh = Scrapli(**r1) In [3]: ssh.open() После этого можно отправлять команды: In [4]: ssh.get_prompt() Out[4]: 'R1#' In [5]: ssh.close() При использовании менеджера контекста, open вызывать не надо: In [8]: with Scrapli(**r1) as ssh: ...: print(ssh.get_prompt()) ...: R1# Использование драйвера Подключение с использованием драйвера IOSXEDriver (технически подключение выполняется к Cisco IOS): In [11]: from scrapli.driver.core import IOSXEDriver In [12]: r1_driver = { ...: "host": "192.168.100.1", ...: "auth_username": "cisco", ...: "auth_password": "cisco", ...: "auth_secondary": "cisco", ...: "auth_strict_key": False, ...: } In [13]: with IOSXEDriver(**r1_driver) as ssh: ...: print(ssh.get_prompt()) ...: R1# Отправка команд В scrapli есть несколько методов для отправки команд/ Все эти методы возвращают объект Response, а не вывод команды в виде строки. In [15]: reply = ssh.send_command("sh clock") In [16]: reply Out[16]: Response(host='192.168.100.1',channel_input='sh clock',textfsm_platform='cisco_iosxe',genie_platform='iosxe',failed_when_contains=['% Ambiguous command', '% Incomplete command', '% Invalid input detected', '% Unknown command']) Получить вывод команды можно обратившись к атрибуту result: In [17]: reply.result Out[17]: '*17:31:54.232 UTC Wed Jan 25 2023' Атрибут raw_result содержит байтовую строку с полным выводом: In [18]: reply.raw_result Out[18]: b'\n*17:31:54.232 UTC Wed Jan 25 2023'\nR1#' Для команд, которые выполняются дольше обычных show, может быть необходимо знать время выполнения команды: In [18]: r = ssh.send_command("ping 10.1.1.1") In [19]: r.result Out[19]: 'Type escape sequence to abort.\nSending 5, 100-byte ICMP Echos to 10.1.1.1, timeout is 2 seconds:\n.....\nSuccess rate is 0 percent (0/5)' In [20]: r.elapsed_time Out[20]: 10.047594 In [21]: r.start_time Out[21]: datetime.datetime(2023, 4, 1, 7, 10, 56, 63697) In [22]: r.finish_time Out[22]: datetime.datetime(2023, 4, 1, 7, 11, 6, 111291) Атрибут channel_input возвращает команду, которая была отправлена на оборудование: In [23]: r.channel_input Out[23]: 'ping 10.1.1.1' Метод send_command позволяет отправить одну команду на устройство. Вызов метода send_command: In [15]: reply = ssh.send_command("sh clock") In [16]: reply Out[16]: Response(host='192.168.100.1',channel_input='sh clock',textfsm_platform='cisco_iosxe',genie_platform='iosxe',failed_when_contains=['% Ambiguous command', '% Incomplete command', '% Invalid input detected', '% Unknown command']) Параметр timeout_ops указывает сколько ждать выполнения команды: In [19]: ssh.send_command("ping 8.8.8.8", timeout_ops=20) Out[19]: Response Если команда не выполнилась за указанное время, сгенерируется исключение ScrapliTimeout (вывод сокращен): In [20]: ssh.send_command("ping 8.8.8.8", timeout_ops=2) --------------------------------------------------------------------------- ScrapliTimeout Traceback (most recent call last) ----> 1 ssh.send_command("ping 8.8.8.8", timeout_ops=2) Кроме получения обычного вывода команды, scrapli также позволяет получить структурированный вывод, например, с помощью метода textfsm_parse_output: In [21]: reply = ssh.send_command("sh ip int br") In [22]: reply.textfsm_parse_output() Out[22]: [{'intf': 'Ethernet0/0', 'ipaddr': '192.168.100.1', 'status': 'up', 'proto': 'up'}, {'intf': 'Ethernet0/1', 'ipaddr': '192.168.200.1', 'status': 'up', 'proto': 'up'}, {'intf': 'Ethernet0/2', 'ipaddr': 'unassigned', 'status': 'up', 'proto': 'up'}, {'intf': 'Ethernet0/3', 'ipaddr': '192.168.130.1', 'status': 'up', 'proto': 'up'}] Обнаружение ошибок Методы для отправки команд автоматически проверяют вывод на наличие ошибок. Для каждого вендора/типа оборудования это свои ошибки, плюс можно самостоятельно указать наличие каких строк в выводе будет считаться ошибкой. По умолчанию для IOSXEDriver ошибками будут считаться такие строки: In [21]: ssh.failed_when_contains Out[21]: ['% Ambiguous command', '% Incomplete command', '% Invalid input detected', '% Unknown command'] Атрибут failed у объекта Response возвращает True, если команда отработала с ошибкой и False, если без ошибки: In [23]: reply = ssh.send_command("sh clck") In [24]: reply.result Out[24]: " ^\n% Invalid input detected at '^' marker." In [25]: reply Out[25]: Response(host='192.168.100.1',channel_input='sh clck',textfsm_platform='cisco_iosxe',genie_platform='iosxe',failed_when_contains=['% Ambiguous command', '% Incomplete command', '% Invalid input detected', '% Unknown command']) In [26]: reply.failed Out[26]: True Метод send_config Метод send_config позволяет отправить одну команду конфигурационного режима. In [33]: r = ssh.send_config("username user1 password password1") In [34]: r.result Out[34]: '' Можно добавлять параметр strip_prompt=False и тогда в выводе появится приглашение: In [37]: r = ssh.send_config("username user1 password password1", strip_prompt=False) In [38]: r.result Out[38]: 'R1(config)#' Методы send_commands, send_configs Методы send_commands, send_configs отличаются от send_command, send_config тем, что могут отправлять несколько команд. Кроме того, эти методы возвращают не Response, а MultiResponse. In [44]: reply = ssh.send_commands(["sh clock", "sh ip int br"]) In [45]: reply Out[45]: [Response(host='192.168.100.1',channel_input='sh clock',textfsm_platform='cisco_iosxe',genie_platform='iosxe',failed_when_contains=['% Ambiguous command', '% Incomplete command', '% Invalid input detected', '% Unknown command']), Response(host='192.168.100.1',channel_input='sh ip int br',textfsm_platform='cisco_iosxe',genie_platform='iosxe',failed_when_contains=['% Ambiguous command', '% Incomplete command', '% Invalid input detected', '% Unknown command'])] In [46]: for r in reply: ...: print(r) ...: print(r.result) ...: Response *08:38:20.115 UTC Thu Apr 1 2021 Response Interface IP-Address OK? Method Status Protocol Ethernet0/0 192.168.100.1 YES NVRAM up up Ethernet0/1 192.168.200.1 YES NVRAM up up Ethernet0/2 unassigned YES NVRAM up up Ethernet0/3 192.168.130.1 YES NVRAM up up In [47]: reply.result Out[47]: 'sh clock\n*08:38:20.115 UTC Thu Apr 1 2021sh ip int br\nInterface IP-Address OK? Method Status Protocol\nEthernet0/0 192.168.100.1 YES NVRAM up up\nEthernet0/1 192.168.200.1 YES NVRAM up up\nEthernet0/2 unassigned YES NVRAM up up\nEthernet0/3 192.168.130.1 YES NVRAM up up' In [48]: reply[0] Out[48]: Response(host='192.168.100.1',channel_input='sh clock',textfsm_platform='cisco_iosxe',genie_platform='iosxe',failed_when_contains=['% Ambiguous command', '% Incomplete command', '% Invalid input detected', '% Unknown command']) In [49]: reply[1] Out[49]: Response(host='192.168.100.1',channel_input='sh ip int br',textfsm_platform='cisco_iosxe',genie_platform='iosxe',failed_when_contains=['% Ambiguous command', '% Incomplete command', '% Invalid input detected', '% Unknown command']) In [50]: reply[0].result Out[50]: '*08:38:20.115 UTC Thu Apr 1 2021' При отправке нескольких команд также очень удобно использовать параметр stop_on_failed. По умолчанию он равен False, поэтому выполняются все команды, но если указать stop_on_failed=True, после возникновения ошибки в какой-то команде, следующие команды не будут выполняться: In [59]: reply = ssh.send_commands(["ping 192.168.100.2", "sh clck", "sh ip int br"], stop_on_failed=True) In [60]: reply Out[60]: [Response(host='192.168.100.1',channel_input='ping 192.168.100.2',textfsm_platform='cisco_iosxe',genie_platform='iosxe',failed_when_contains=['% Ambiguous command', '% Incomplete command', '% Invalid input detected', '% Unknown command']), Response(host='192.168.100.1',channel_input='sh clck',textfsm_platform='cisco_iosxe',genie_platform='iosxe',failed_when_contains=['% Ambiguous command', '% Incomplete command', '% Invalid input detected', '% Unknown command'])] In [61]: print(reply) MultiResponse In [62]: reply.result Out[62]: "ping 192.168.100.2\nType escape sequence to abort.\nSending 5, 100-byte ICMP Echos to 192.168.100.2, timeout is 2 seconds:\n!!!!!\nSuccess rate is 100 percent (5/5), round-trip min/avg/max = 1/2/6 mssh clck\n ^\n% Invalid input detected at '^' marker." In [63]: for r in reply: ...: print(r) ...: print(r.result) ...: Response Type escape sequence to abort. Sending 5, 100-byte ICMP Echos to 192.168.100.2, timeout is 2 seconds: !!!!! Success rate is 100 percent (5/5), round-trip min/avg/max = 1/2/6 ms Response ^ % Invalid input detected at '^' marker. Подключение telnet Для подключения к оборудовани по Telnet надо указать transport равным telnet и обязательно указать параметр port равным 23 (или тому порту который используется у вас для подключения по Telnet): from scrapli.driver.core import IOSXEDriver from scrapli.exceptions import ScrapliException import socket r1 = { "host": "192.168.100.1", "auth_username": "cisco", "auth_password": "cisco2", "auth_secondary": "cisco", "auth_strict_key": False, "transport": "telnet", "port": 23, # обязательно указывать при подключении telnet } def send_show(device, show_command): try: with IOSXEDriver(**device) as ssh: reply = ssh.send_command(show_command) return reply.result except socket.timeout as error: print(error) except ScrapliException as error: print(error, device["host"]) if __name__ == "__main__": output = send_show(r1, "sh ip int br") print(output) Использования scrapli from scrapli.driver.core import IOSXEDriver from scrapli.exceptions import ScrapliException r1 = { "host": "192.168.100.1", "auth_username": "cisco", "auth_password": "cisco", "auth_secondary": "cisco", "auth_strict_key": False, "timeout_socket": 5, # timeout for establishing socket/initial connection "timeout_transport": 10, # timeout for ssh|telnet transport } def send_show(device, show_command): try: with IOSXEDriver(**device) as ssh: reply = ssh.send_command(show_command) return reply.result except ScrapliException as error: print(error, device["host"]) if __name__ == "__main__": output = send_show(r1, "sh ip int br") print(output) from pprint import pprint from scrapli import Scrapli r1 = { "host": "192.168.100.1", "auth_username": "cisco", "auth_password": "cisco", "auth_secondary": "cisco", "auth_strict_key": False, "platform": "cisco_iosxe", } def send_show(device, show_commands): if type(show_commands) == str: show_commands = [show_commands] cmd_dict = {} with Scrapli(**device) as ssh: for cmd in show_commands: reply = ssh.send_command(cmd) cmd_dict[cmd] = reply.result return cmd_dict if __name__ == "__main__": print("show".center(20, "#")) output = send_show(r1, ["sh ip int br", "sh ver | i uptime"]) pprint(output, width=120) from pprint import pprint from scrapli import Scrapli r1 = { "host": "192.168.100.1", "auth_username": "cisco", "auth_password": "cisco", "auth_secondary": "cisco", "auth_strict_key": False, "platform": "cisco_iosxe", } def send_cfg(device, cfg_commands, strict=False): output = "" if type(cfg_commands) == str: cfg_commands = [cfg_commands] with Scrapli(**device) as ssh: reply = ssh.send_configs(cfg_commands, stop_on_failed=strict) for cmd_reply in reply: if cmd_reply.failed: print(f"Привыполнениикомандывозниклаошибка:\n{reply.result}\n") output = reply.result return output if __name__ == "__main__": output_cfg = send_cfg( r1, ["interfacelo11", "ip address 11.1.1.1 255.255.255.255"], strict=True ) print(output_cfg) |