Examples¶
In examples are used configuration files. config.yaml:
device_list: credentials.yaml
credentials.yaml
- device_type: cisco_asa
host: 1.1.1.1
password: ****
username: ****
- device_type: cisco_ios
host: 2.2.2.2
password: ****
username: ****
- device_type: fujitsu_switch
host: 3.3.3.3
password: ****
username: ****
- device_type: hp_comware
host: 4.4.4.4
password: ****
username: ****
- device_type: hp_comware_limited
host: 5.5.5.5
password: ****
username: ****
cmdline_password: '512900'
- device_type: arista_eos
host: 6.6.6.6
password: ****
username: ****
- device_type: juniper_junos
host: 7.7.7.7
password: ****
username: ****
ASA example¶
import asyncio
import logging
import yaml
import netdev
config_path = 'config.yaml'
logging.basicConfig(level=logging.INFO)
netdev.logger.setLevel(logging.DEBUG)
async def task(param):
async with netdev.create(**param) as asa:
# Testing sending simple command
out = await asa.send_command('show run')
print(out)
# Testing interactive dialog
out = await asa.send_command("copy r scp:", pattern=r'\[running-config\]\?', strip_command=False)
out += await asa.send_command("\n", pattern=r'\[\]\?', strip_command=False)
out += await asa.send_command("\n", strip_command=False)
print(out)
async def run():
config = yaml.safe_load(open(config_path, 'r'))
devices = yaml.safe_load(open(config['device_list'], 'r'))
tasks = [task(dev) for dev in devices if dev['device_type'] == 'cisco_asa']
await asyncio.wait(tasks)
loop = asyncio.get_event_loop()
loop.run_until_complete(run())
NX-OS example¶
import asyncio
import logging
import yaml
import netdev
config_path = 'config.yaml'
logging.basicConfig(level=logging.INFO)
netdev.logger.setLevel(logging.DEBUG)
async def task(param):
async with netdev.create(**param) as nxos:
# Testing sending simple command
out = await nxos.send_command('show run', strip_command=True)
print(out)
async def run():
config = yaml.safe_load(open(config_path, 'r'))
devices = yaml.safe_load(open(config['device_list'], 'r'))
tasks = [task(dev) for dev in devices if dev['device_type'] == 'cisco_nxos']
await asyncio.wait(tasks)
loop = asyncio.get_event_loop()
loop.run_until_complete(run())
IOS example¶
import asyncio
import logging
import yaml
import netdev
config_path = 'config.yaml'
logging.basicConfig(level=logging.INFO)
netdev.logger.setLevel(logging.DEBUG)
async def task(param):
async with netdev.create(**param) as ios:
# Testing sending simple command
out = await ios.send_command("show ver")
print(out)
# Testing sending configuration set
commands = ["line console 0", "exit"]
out = await ios.send_config_set(commands)
print(out)
# Testing sending simple command with long output
out = await ios.send_command("show run")
print(out)
# Testing interactive dialog
out = await ios.send_command("conf", pattern=r'\[terminal\]\?', strip_command=False)
out += await ios.send_command("term", strip_command=False)
out += await ios.send_command("exit", strip_command=False, strip_prompt=False)
print(out)
async def run():
config = yaml.safe_load(open(config_path, 'r'))
devices = yaml.safe_load(open(config['device_list'], 'r'))
tasks = [task(dev) for dev in devices if dev['device_type'] == 'cisco_ios']
await asyncio.wait(tasks)
loop = asyncio.get_event_loop()
loop.run_until_complete(run())
IOS XR example¶
import asyncio
import logging
import yaml
import netdev
config_path = 'config.yaml'
logging.basicConfig(level=logging.INFO)
netdev.logger.setLevel(logging.DEBUG)
async def task(param):
async with netdev.create(**param) as iosxr:
# Testing sending simple command
out = await iosxr.send_command("show ver")
print(out)
# Testing sending simple command with long output
out = await iosxr.send_command("show run")
print(out)
# Testing sending configuration set
commands = ["interface loopback 0", "description TEST LOOPBACK"]
out = await iosxr.send_config_set(commands, with_commit=True)
print(out)
# Testing failed commit
commands = ["interface GigabitEthernet 0/0/0/0", "service-policy input 1"]
out = await iosxr.send_config_set(commands, with_commit=False)
print(out)
try:
commands = ["interface GigabitEthernet 0/0/0/0", "service-policy input 2"]
await iosxr.send_config_set(commands)
except netdev.CommitError:
print("Commit Error")
async def run():
config = yaml.safe_load(open(config_path, 'r'))
devices = yaml.safe_load(open(config['device_list'], 'r'))
tasks = [task(dev) for dev in devices if dev['device_type'] == 'cisco_ios_xr']
await asyncio.gather(*tasks)
loop = asyncio.get_event_loop()
loop.run_until_complete(run())
IOS SG3XX example¶
#!/usr/bin/env python3
# Import Python library
import asyncio, netdev
# Coroutine used for the tasks
async def task(param):
# Create an object for the devices and open SSH connections
async with netdev.create(**param) as ios:
# Testing sending simple command
# Command to send
cmd = "show clock"
# Sending command
output = await ios.send_command(cmd)
# Display the output
print(output)
# Main coroutine
async def main():
# Parameters of the network device
my_device = { 'username' : 'LOGIN',
'password' : 'PASSWORD',
'host': 'IP_ADDRESS',
'device_type': 'cisco_sg3xx',
}
# List of devices
devices = [my_device]
# List of tasks
my_tasks = [task(dev) for dev in devices]
# Starting the coroutine of the tasks
await asyncio.wait(my_tasks)
# Main function call
if __name__ == '__main__':
# Run the main coroutine
asyncio.run(main())
'''
Result:
********************************************************************************
.14:07:35 J Aug 28 2020
Time source is sntp
Time from Browser is disabled
********************************************************************************
'''
Fujitsu example¶
import asyncio
import logging
import yaml
import netdev
config_path = 'config.yaml'
logging.basicConfig(level=logging.INFO)
netdev.logger.setLevel(logging.DEBUG)
async def task(param):
async with netdev.create(**param) as fuj:
# Testing sending configuration set
out = await fuj.send_config_set(['vlan database', 'exit'])
print(out)
# Testing sending simple command
out = await fuj.send_command('show ver')
print(out)
async def run():
config = yaml.safe_load(open(config_path, 'r'))
devices = yaml.safe_load(open(config['device_list'], 'r'))
tasks = [task(dev) for dev in devices if dev['device_type'] == 'fujitsu_switch']
await asyncio.wait(tasks)
loop = asyncio.get_event_loop()
loop.run_until_complete(run())
Comware example¶
import asyncio
import logging
import yaml
import netdev
config_path = 'config.yaml'
logging.basicConfig(level=logging.INFO)
netdev.logger.setLevel(logging.DEBUG)
async def task(param):
async with netdev.create(**param) as hp:
# Testing sending simple command
out = await hp.send_command('display ver')
print(out)
# Testing sending configuration set
commands = ["Vlan 1", "quit"]
out = await hp.send_config_set(commands)
print(out)
# Testing sending simple command with long output
out = await hp.send_command('display cur')
print(out)
async def run():
config = yaml.safe_load(open(config_path, 'r'))
devices = yaml.safe_load(open(config['device_list'], 'r'))
tasks = [task(dev) for dev in devices if dev['device_type'] == 'hp_comware']
await asyncio.wait(tasks)
loop = asyncio.get_event_loop()
loop.run_until_complete(run())
Comware Limited example¶
import asyncio
import logging
import yaml
import netdev
config_path = 'config.yaml'
logging.basicConfig(level=logging.INFO)
netdev.logger.setLevel(logging.DEBUG)
async def task(param):
async with netdev.create(**param) as hp:
# Testing sending simple command
out = await hp.send_command('display ver')
print(out)
# Testing sending configuration set
commands = ["Vlan 1", "quit"]
out = await hp.send_config_set(commands)
print(out)
# Testing sending simple command with long output
out = await hp.send_command('display cur')
print(out)
async def run():
config = yaml.safe_load(open(config_path, 'r'))
devices = yaml.safe_load(open(config['device_list'], 'r'))
tasks = [task(dev) for dev in devices if dev['device_type'] == 'hp_comware_limited']
await asyncio.wait(tasks)
loop = asyncio.get_event_loop()
loop.run_until_complete(run())
Mikrotik RouterOS example¶
import asyncio
import logging
import yaml
import netdev
config_path = 'config.yaml'
logging.basicConfig(level=logging.INFO)
netdev.logger.setLevel(logging.DEBUG)
async def task(param):
async with netdev.create(**param) as routeros:
# Testing sending simple command
commands = ['/ip address', 'print', '/']
for cmd in commands:
print(await routeros.send_command(cmd))
# Testing sending configuration set
out = await routeros.send_config_set(commands)
print(out)
async def run():
config = yaml.safe_load(open(config_path, 'r'))
devices = yaml.safe_load(open(config['device_list'], 'r'))
tasks = [task(dev) for dev in devices if dev['device_type'] == 'mikrotik_routeros']
await asyncio.wait(tasks)
loop = asyncio.get_event_loop()
loop.run_until_complete(run())
Arista EOS example¶
import asyncio
import logging
import yaml
import netdev
config_path = 'config.yaml'
logging.basicConfig(level=logging.INFO)
netdev.logger.setLevel(logging.DEBUG)
async def task(param):
async with netdev.create(**param) as arista:
# Testing sending simple command
out = await arista.send_command('show run', strip_command=True)
print(out)
# Testing sending configuration set
commands = ["vlan 1", "exit"]
out = await arista.send_config_set(commands)
print(out)
async def run():
config = yaml.safe_load(open(config_path, 'r'))
devices = yaml.safe_load(open(config['device_list'], 'r'))
tasks = [task(dev) for dev in devices if dev['device_type'] == 'arista_eos']
await asyncio.wait(tasks)
loop = asyncio.get_event_loop()
loop.run_until_complete(run())
Aruba AOS 6.X example¶
import asyncio
import logging
import yaml
import netdev
config_path = 'config.yaml'
logging.basicConfig(level=logging.INFO)
netdev.logger.setLevel(logging.DEBUG)
async def task(param):
async with netdev.create(**param) as ios:
# Testing sending simple command
out = await ios.send_command("show ver")
print(out)
# Testing sending configuration set
commands = ["interface loopback", "exit"]
out = await ios.send_config_set(commands)
print(out)
# Testing sending simple command with long output
out = await ios.send_command("show run")
print(out)
async def run():
config = yaml.safe_load(open(config_path, 'r'))
devices = yaml.safe_load(open(config['device_list'], 'r'))
tasks = [task(dev) for dev in devices if dev['device_type'] == 'aruba_aos_6']
await asyncio.wait(tasks)
loop = asyncio.get_event_loop()
loop.run_until_complete(run())
Juniper JunOS example¶
import asyncio
import logging
import yaml
import netdev
config_path = 'config.yaml'
logging.basicConfig(level=logging.INFO)
netdev.logger.setLevel(logging.DEBUG)
async def task(param):
async with netdev.create(**param) as junos:
# Testing sending simple command
out = await junos.send_command("show version")
print(out)
# Testing sending configuration set
commands = ["edit system", "edit login"]
out = await junos.send_config_set(commands, with_commit=True)
print(out)
# Testing sending simple command with long output
out = await junos.send_command("show config")
print(out)
# Testing interactive dialog
commands = ["set system login message 123", "delete system login message 123"]
out = await junos.send_config_set(commands, with_commit=False, exit_config_mode=False)
out += await junos.send_command("exit", pattern=r'Exit with uncommitted changes\?', strip_command=False)
out += await junos.send_command("no", strip_command=False)
out += await junos.send_command("rollback 0", strip_command=False)
out += await junos.send_command("exit configuration-mode", strip_command=False)
print(out)
async def run():
config = yaml.safe_load(open(config_path, 'r'))
devices = yaml.safe_load(open(config['device_list'], 'r'))
tasks = [task(dev) for dev in devices if dev['device_type'] == 'juniper_junos']
await asyncio.wait(tasks)
loop = asyncio.get_event_loop()
loop.run_until_complete(run())
Terminal example¶
import asyncio
import logging
import yaml
import netdev
config_path = 'config.yaml'
logging.basicConfig(level=logging.INFO)
netdev.logger.setLevel(logging.DEBUG)
async def task(param):
async with netdev.create(**param) as term:
# Testing sending simple command
out = await term.send_command('ls -al', strip_command=True)
print(out)
out = await term.send_command('pwd', strip_command=True)
print(out)
out = await term.send_command('echo test', strip_command=True)
print(out)
async def run():
config = yaml.safe_load(open(config_path, 'r'))
devices = yaml.safe_load(open(config['device_list'], 'r'))
tasks = [task(dev) for dev in devices if dev['device_type'] == 'terminal']
await asyncio.wait(tasks)
loop = asyncio.get_event_loop()
loop.run_until_complete(run())
Alcatel AOS example¶
#!/usr/bin/env python3
# Import Python library
import asyncio, netdev
# Coroutine used for the tasks
async def task(param):
# Create an object for the devices and open SSH connections
async with netdev.create(**param) as device:
# Testing sending simple command
# Command to send
cmd = "show system"
# Sending command
output = await device.send_command(cmd)
# Display the output
print(output)
# Display separator
print("*" * 80)
# Testing sending configuration set
# Commands to send
commands = ["vlan 3000", "no vlan 3001"]
# Sending command
output = await device.send_config_set(commands)
# Display the output
print(output)
# Main coroutine
async def main():
# Parameters of the network device
my_device = { 'username' : 'LOGIN',
'password' : 'PASSWORD',
'host': 'IP_ADDRESS',
'device_type': 'alcatel_aos',
}
# List of devices
devices = [my_device]
# List of tasks
my_tasks = [task(dev) for dev in devices]
# Starting the coroutine of the tasks
await asyncio.wait(my_tasks)
# Main function call
if __name__ == '__main__':
# Run the main coroutine
asyncio.run(main())
'''
Result:
********************************************************************************
System:
Description: Alcatel-Lucent Enterprise OS6860E-48 8.4.1.141.R03 GA, December 07, 2017.,
Object ID: 1.3.6.1.4.1.6486.801.1.1.2.1.11.1.7,
Up Time: 5 days 5 hours 3 minutes and 56 seconds,
Contact: Alcatel-Lucent, http://enterprise.alcatel-lucent.com,
Name: switch01,
Location: Somewhere nearby,
Services: 78,
Date & Time: SAT AUG 29 2020 18:48:53 (CEST)
Flash Space:
Primary CMM:
Available (bytes): 933896192,
Comments : None
********************************************************************************
vlan 3000
switch01 ==> no vlan 3001
switch01 ==>
********************************************************************************
'''