Python has become the de facto language for network automation due to its readability, extensive library ecosystem, and strong community support.
Why Python for Network Automation?
Advantages
- Readable syntax: Easy to learn and maintain
- Rich ecosystem: Libraries for every network vendor and protocol
- Cross-platform: Runs on Windows, Linux, macOS
- Strong community: Extensive documentation and support
- Integration: Works with APIs, databases, cloud services
Use Cases
- Configuration management across devices
- Network inventory and discovery
- Compliance auditing and reporting
- Troubleshooting and diagnostics
- Automated backups and restore
- Network monitoring and alerting
Essential Libraries
1. Netmiko
SSH-based device configuration and command execution.
Installation
pip install netmiko
Basic Usage
from netmiko import ConnectHandler
# Define device
device = {
'device_type': 'cisco_ios',
'host': '192.168.1.1',
'username': 'admin',
'password': 'password',
'secret': 'enable_password', # Optional
}
# Connect and execute commands
with ConnectHandler(**device) as net_connect:
net_connect.enable()
output = net_connect.send_command('show ip interface brief')
print(output)
Configuration Changes
config_commands = [
'interface GigabitEthernet0/1',
'description Uplink to Core',
'ip address 10.0.1.1 255.255.255.0',
'no shutdown'
]
with ConnectHandler(**device) as net_connect:
output = net_connect.send_config_set(config_commands)
print(output)
# Save configuration
net_connect.save_config()
Supported Platforms
- Cisco IOS, IOS-XE, IOS-XR, NX-OS
- Arista EOS
- Juniper Junos
- Palo Alto PAN-OS
- F5 Networks
- Many others (100+ device types)
2. NAPALM
Vendor-neutral API for network device interaction.
Installation
pip install napalm
Basic Usage
from napalm import get_network_driver
driver = get_network_driver('ios')
device = driver(
hostname='192.168.1.1',
username='admin',
password='password'
)
device.open()
# Get facts
facts = device.get_facts()
print(f"Hostname: {facts['hostname']}")
print(f"Model: {facts['model']}")
print(f"Uptime: {facts['uptime']}")
# Get interfaces
interfaces = device.get_interfaces()
for interface, data in interfaces.items():
print(f"{interface}: {data['is_up']}")
device.close()
Configuration Management with Rollback
# Load configuration (doesn't apply yet)
device.open()
device.load_merge_candidate(filename='config.txt')
# Compare changes
diff = device.compare_config()
print(diff)
# Apply if looks good
if input("Apply? (yes/no): ").lower() == 'yes':
device.commit_config()
else:
device.discard_config()
device.close()
Supported Operations
get_facts(),get_interfaces(),get_bgp_neighbors()get_arp_table(),get_mac_address_table()load_merge_candidate(),load_replace_candidate()compare_config(),commit_config(),rollback()
3. Nornir
Parallel execution framework for network automation.
Installation
pip install nornir nornir-netmiko nornir-napalm nornir-utils
Basic Setup
from nornir import InitNornir
from nornir_netmiko import netmiko_send_command
from nornir_utils.plugins.functions import print_result
# Initialize with inventory
nr = InitNornir(config_file="config.yaml")
# Run task on all devices in parallel
result = nr.run(
task=netmiko_send_command,
command_string="show version"
)
print_result(result)
Configuration File (config.yaml)
inventory:
plugin: SimpleInventory
options:
host_file: "inventory/hosts.yaml"
group_file: "inventory/groups.yaml"
runner:
plugin: threaded
options:
num_workers: 10
Inventory File (hosts.yaml)
core-switch-1:
hostname: 192.168.1.1
groups:
- cisco_switches
data:
site: datacenter-1
core-switch-2:
hostname: 192.168.1.2
groups:
- cisco_switches
data:
site: datacenter-2
Advanced Task
from nornir_netmiko import netmiko_send_config
def configure_interface(task):
"""Configure interface description based on inventory data"""
site = task.host.data.get('site', 'unknown')
config = [
'interface GigabitEthernet0/0',
f'description Link to {site}',
'no shutdown'
]
task.run(
task=netmiko_send_config,
config_commands=config
)
# Run on all devices
result = nr.run(task=configure_interface)
print_result(result)
4. Paramiko
Low-level SSH library (Netmiko is built on this).
Installation
pip install paramiko
Basic Usage
import paramiko
ssh = paramiko.SSHClient()
ssh.set_missing_host_key_policy(paramiko.AutoAddPolicy())
ssh.connect('192.168.1.1', username='admin', password='password')
stdin, stdout, stderr = ssh.exec_command('show ip route')
output = stdout.read().decode()
print(output)
ssh.close()
When to Use Paramiko
- Need more control than Netmiko provides
- Custom SSH interactions
- Non-standard device types
- File transfers over SFTP
5. PyEZ (for Juniper)
Juniper-specific Python library.
Installation
pip install junos-eznc
Basic Usage
from jnpr.junos import Device
dev = Device(host='192.168.1.1', user='admin', password='password')
dev.open()
# Get facts
print(dev.facts)
# Execute RPC
routes = dev.rpc.get_route_information()
print(routes)
dev.close()
6. pyATS/Genie (Cisco)
Cisco’s test automation framework with parsing capabilities.
Installation
pip install pyats genie
Basic Usage
from genie.testbed import load
testbed = load('testbed.yaml')
device = testbed.devices['router1']
device.connect()
# Parse output into structured data
output = device.parse('show ip interface brief')
for interface, data in output['interface'].items():
print(f"{interface}: {data['ip_address']}")
Common Automation Patterns
1. Device Backup
Automated configuration backups with version control.
from netmiko import ConnectHandler
from datetime import datetime
import os
def backup_device(device_info, backup_dir='backups'):
"""Backup device configuration"""
# Create backup directory if doesn't exist
os.makedirs(backup_dir, exist_ok=True)
# Connect and get config
with ConnectHandler(**device_info) as net_connect:
config = net_connect.send_command('show running-config')
# Generate filename with timestamp
hostname = device_info['host']
timestamp = datetime.now().strftime('%Y%m%d_%H%M%S')
filename = f"{backup_dir}/{hostname}_{timestamp}.cfg"
# Save to file
with open(filename, 'w') as f:
f.write(config)
print(f"Backed up {hostname} to {filename}")
return filename
# Usage
device = {
'device_type': 'cisco_ios',
'host': '192.168.1.1',
'username': 'admin',
'password': 'password',
}
backup_device(device)
2. Bulk Configuration Deployment
Deploy configurations to multiple devices.
from nornir import InitNornir
from nornir_netmiko import netmiko_send_config
from nornir_utils.plugins.functions import print_result
def deploy_ntp_config(task):
"""Deploy NTP configuration to all devices"""
ntp_servers = [
'ntp server 10.0.0.1',
'ntp server 10.0.0.2',
'ntp server 10.0.0.3'
]
task.run(
task=netmiko_send_config,
config_commands=ntp_servers
)
# Initialize and run
nr = InitNornir(config_file="config.yaml")
result = nr.run(task=deploy_ntp_config)
print_result(result)
# Check for failures
if result.failed:
print("Failed devices:")
for host, task_result in result.failed_hosts.items():
print(f" {host}: {task_result.exception}")
3. Network Inventory Collection
Gather device information for documentation.
from napalm import get_network_driver
import csv
def collect_inventory(devices):
"""Collect inventory data from devices"""
inventory = []
for device_info in devices:
driver = get_network_driver(device_info['driver'])
device = driver(
hostname=device_info['host'],
username=device_info['username'],
password=device_info['password']
)
try:
device.open()
facts = device.get_facts()
inventory.append({
'hostname': facts['hostname'],
'model': facts['model'],
'serial': facts['serial_number'],
'os_version': facts['os_version'],
'uptime': facts['uptime']
})
device.close()
except Exception as e:
print(f"Error collecting from {device_info['host']}: {e}")
return inventory
def save_to_csv(inventory, filename='inventory.csv'):
"""Save inventory to CSV file"""
keys = inventory[0].keys()
with open(filename, 'w', newline='') as f:
writer = csv.DictWriter(f, fieldnames=keys)
writer.writeheader()
writer.writerows(inventory)
# Usage
devices = [
{'driver': 'ios', 'host': '192.168.1.1', 'username': 'admin', 'password': 'pass'},
{'driver': 'ios', 'host': '192.168.1.2', 'username': 'admin', 'password': 'pass'},
]
inventory = collect_inventory(devices)
save_to_csv(inventory)
4. Compliance Checking
Audit devices against security baseline.
from netmiko import ConnectHandler
def check_compliance(device_info, baseline):
"""Check device compliance against baseline"""
results = {
'hostname': device_info['host'],
'compliant': True,
'violations': []
}
with ConnectHandler(**device_info) as net_connect:
config = net_connect.send_command('show running-config')
# Check required settings
for setting in baseline['required']:
if setting not in config:
results['compliant'] = False
results['violations'].append(f"Missing: {setting}")
# Check forbidden settings
for setting in baseline['forbidden']:
if setting in config:
results['compliant'] = False
results['violations'].append(f"Found forbidden: {setting}")
return results
# Define baseline
baseline = {
'required': [
'service password-encryption',
'logging buffered',
'ntp server',
],
'forbidden': [
'no service password-encryption',
'enable password', # Should use enable secret
]
}
# Check device
device = {
'device_type': 'cisco_ios',
'host': '192.168.1.1',
'username': 'admin',
'password': 'password',
}
result = check_compliance(device, baseline)
if not result['compliant']:
print(f"Violations found on {result['hostname']}:")
for violation in result['violations']:
print(f" - {violation}")
5. Automated Troubleshooting
Collect diagnostic information for troubleshooting.
from netmiko import ConnectHandler
from datetime import datetime
def troubleshoot_interface(device_info, interface):
"""Collect troubleshooting data for an interface"""
commands = [
f'show interface {interface}',
f'show ip interface {interface}',
f'show controllers {interface}',
f'show logging | include {interface}',
]
output = {}
with ConnectHandler(**device_info) as net_connect:
for command in commands:
output[command] = net_connect.send_command(command)
# Save to file
timestamp = datetime.now().strftime('%Y%m%d_%H%M%S')
filename = f"troubleshooting_{interface}_{timestamp}.txt"
with open(filename, 'w') as f:
for command, result in output.items():
f.write(f"\n{'='*60}\n")
f.write(f"Command: {command}\n")
f.write(f"{'='*60}\n")
f.write(result)
f.write("\n")
print(f"Troubleshooting data saved to {filename}")
return output
# Usage
device = {
'device_type': 'cisco_ios',
'host': '192.168.1.1',
'username': 'admin',
'password': 'password',
}
troubleshoot_interface(device, 'GigabitEthernet0/1')
Working with APIs
REST API Interaction
Many modern network devices offer REST APIs.
import requests
import json
class NetworkDevice:
"""Generic REST API wrapper for network devices"""
def __init__(self, host, username, password, verify_ssl=False):
self.base_url = f"https://{host}/api/v1"
self.auth = (username, password)
self.verify = verify_ssl
def get(self, endpoint):
"""GET request"""
url = f"{self.base_url}/{endpoint}"
response = requests.get(url, auth=self.auth, verify=self.verify)
response.raise_for_status()
return response.json()
def post(self, endpoint, data):
"""POST request"""
url = f"{self.base_url}/{endpoint}"
headers = {'Content-Type': 'application/json'}
response = requests.post(
url,
auth=self.auth,
headers=headers,
data=json.dumps(data),
verify=self.verify
)
response.raise_for_status()
return response.json()
# Usage
device = NetworkDevice('192.168.1.1', 'admin', 'password')
# Get interfaces
interfaces = device.get('interfaces')
for interface in interfaces:
print(f"{interface['name']}: {interface['status']}")
# Configure interface
config = {
'name': 'GigabitEthernet0/1',
'description': 'Uplink to Core',
'enabled': True
}
device.post('interfaces/configure', config)
NETCONF Example
Using ncclient for NETCONF operations.
from ncclient import manager
import xml.dom.minidom
# Connect
with manager.connect(
host='192.168.1.1',
port=830,
username='admin',
password='password',
hostkey_verify=False
) as m:
# Get running configuration
config = m.get_config(source='running')
# Pretty print XML
xml_str = xml.dom.minidom.parseString(str(config)).toprettyxml()
print(xml_str)
# Edit configuration
config_xml = """
<config>
<interface>
<name>GigabitEthernet0/0</name>
<description>Uplink</description>
</interface>
</config>
"""
m.edit_config(target='candidate', config=config_xml)
m.commit()
Error Handling Best Practices
Robust Connection Handling
from netmiko import ConnectHandler
from netmiko.exceptions import NetmikoTimeoutException, NetmikoAuthenticationException
import logging
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)
def safe_connect(device_info, max_retries=3):
"""Connect with retry logic and proper error handling"""
for attempt in range(max_retries):
try:
connection = ConnectHandler(**device_info)
logger.info(f"Connected to {device_info['host']}")
return connection
except NetmikoTimeoutException:
logger.error(f"Timeout connecting to {device_info['host']} (attempt {attempt + 1})")
if attempt == max_retries - 1:
raise
except NetmikoAuthenticationException:
logger.error(f"Authentication failed for {device_info['host']}")
raise # Don't retry auth failures
except Exception as e:
logger.error(f"Unexpected error: {e}")
if attempt == max_retries - 1:
raise
return None
# Usage
device = {
'device_type': 'cisco_ios',
'host': '192.168.1.1',
'username': 'admin',
'password': 'password',
}
try:
conn = safe_connect(device)
if conn:
output = conn.send_command('show version')
print(output)
conn.disconnect()
except Exception as e:
logger.error(f"Failed to connect: {e}")
Graceful Degradation
from nornir import InitNornir
from nornir_netmiko import netmiko_send_command
def collect_data_with_fallback(task):
"""Try primary command, fallback to alternative"""
commands = [
'show ip interface brief', # Cisco
'show interfaces terse', # Juniper
'show interface status' # Alternative
]
for cmd in commands:
try:
result = task.run(
task=netmiko_send_command,
command_string=cmd
)
if result.result:
return result
except Exception:
continue
raise Exception("All commands failed")
nr = InitNornir(config_file="config.yaml")
result = nr.run(task=collect_data_with_fallback)
Testing and Development
Using Mock Devices
import unittest
from unittest.mock import patch, MagicMock
class TestNetworkAutomation(unittest.TestCase):
@patch('netmiko.ConnectHandler')
def test_backup_device(self, mock_connect):
"""Test device backup function"""
# Setup mock
mock_device = MagicMock()
mock_device.send_command.return_value = "hostname test\n"
mock_connect.return_value.__enter__.return_value = mock_device
# Test function
device_info = {
'device_type': 'cisco_ios',
'host': '192.168.1.1',
'username': 'admin',
'password': 'password',
}
filename = backup_device(device_info)
# Assertions
mock_device.send_command.assert_called_with('show running-config')
self.assertTrue(filename.endswith('.cfg'))
if __name__ == '__main__':
unittest.main()
Security Best Practices
Credential Management
Never hardcode credentials!
import os
from getpass import getpass
# Option 1: Environment variables
username = os.environ.get('NET_USERNAME')
password = os.environ.get('NET_PASSWORD')
# Option 2: Prompt user
if not password:
password = getpass('Enter password: ')
# Option 3: Use secrets management
from keyring import get_password
password = get_password('network_automation', username)
# Option 4: Configuration file (encrypted)
import json
with open('credentials.json.enc') as f:
creds = json.load(f) # Decrypt first!
SSH Key Authentication
device = {
'device_type': 'cisco_ios',
'host': '192.168.1.1',
'username': 'admin',
'use_keys': True,
'key_file': '/home/user/.ssh/id_rsa',
}
Secrets Management with Vault
import hvac
# Connect to Vault
client = hvac.Client(url='http://vault.example.com:8200')
client.token = os.environ['VAULT_TOKEN']
# Read secret
secret = client.secrets.kv.v2.read_secret_version(path='network/credentials')
username = secret['data']['data']['username']
password = secret['data']['data']['password']
Performance Optimization
Parallel Execution with Threading
from concurrent.futures import ThreadPoolExecutor, as_completed
from netmiko import ConnectHandler
def backup_single_device(device_info):
"""Backup single device"""
with ConnectHandler(**device_info) as conn:
config = conn.send_command('show running-config')
return device_info['host'], config
def backup_multiple_devices(devices, max_workers=10):
"""Backup multiple devices in parallel"""
results = {}
with ThreadPoolExecutor(max_workers=max_workers) as executor:
future_to_device = {
executor.submit(backup_single_device, device): device
for device in devices
}
for future in as_completed(future_to_device):
device = future_to_device[future]
try:
hostname, config = future.result()
results[hostname] = config
print(f"✓ Backed up {hostname}")
except Exception as e:
print(f"✗ Failed {device['host']}: {e}")
return results
# Usage
devices = [
{'device_type': 'cisco_ios', 'host': '192.168.1.1', 'username': 'admin', 'password': 'pass'},
{'device_type': 'cisco_ios', 'host': '192.168.1.2', 'username': 'admin', 'password': 'pass'},
# ... more devices
]
results = backup_multiple_devices(devices, max_workers=20)
Production Deployment Tips
- Logging: Use Python’s logging module, not print()
- Configuration: Externalize all settings to config files
- Error handling: Catch and handle all exceptions gracefully
- Dry-run mode: Implement preview/check mode before applying changes
- Rollback: Always have a rollback plan for configuration changes
- Monitoring: Track script execution and alert on failures
- Version control: Keep scripts in Git with proper versioning
- Documentation: Document what scripts do and how to use them
- Testing: Test on lab devices before production
- Change control: Follow change management processes
Conclusion
Python provides powerful tools for network automation. Start with simple tasks like backups and gradually expand to more complex workflows. Always prioritize security, error handling, and maintainability over clever code.
The goal is reliable, repeatable network operations that reduce human error and free up time for higher-value work.