Network automation

import os
from netmiko import ConnectHandler

device = {
    "device_type": "cisco_ios",           # cisco_xe is an alias; cisco_nxos, arista_eos, juniper_junos...
    "host": "sw-01.example.com",
    "username": os.environ["NET_USER"],
    "password": os.environ["NET_PASS"],
    "secret": os.environ.get("NET_ENABLE", ""),
    "conn_timeout": 10,                   # TCP connect timeout, seconds
}

with ConnectHandler(**device) as conn:
    conn.enable()                                                    # needs "secret" if not already privileged
    raw = conn.send_command("show ip interface brief")               # str
    rows = conn.send_command("show ip interface brief", use_textfsm=True)   # list[dict], or str if no template matched

out = conn.send_command("copy running-config startup-config", expect_string=r"\[startup-config\]\?")
out += conn.send_command("\n", expect_string=r"#")

conn.send_config_from_file("acl.cfg")                                # config mode, one line at a time, exits
conn.send_config_set(cmds, cmd_verify=False)                         # faster on slow devices; skips echo checks, so errors are not caught
conn.send_multiline(["copy tftp: flash:", "192.0.2.40", "c9300.bin", "\n"])   # answer a chain of prompts
conn.send_command("show run", use_genie=True)                        # pyATS/Genie parser instead of TextFSM (pip install pyats genie)
conn.send_command("show ip route", use_ttp=True, ttp_template="route.ttp")    # TTP templates
conn.write_channel("show version\n"); time.sleep(1); conn.read_channel()   # raw, for pathological prompts
conn.find_prompt()                                                   # current prompt; changes when hostname changes
conn.disconnect()

from netmiko import file_transfer
file_transfer(conn, source_file="c9300-universalk9.17.12.04.SPA.bin", dest_file="c9300.bin",
              file_system="flash:", direction="put", overwrite_file=False)   # SCP; needs ip scp server enable

from netmiko import SSHDetect
guesser = SSHDetect(**{**device, "device_type": "autodetect"})
device["device_type"] = guesser.autodetect()                          # cisco_ios, cisco_nxos, arista_eos...

# Connect through a jump host with an SSH config file
device["ssh_config_file"] = "~/.ssh/config"                          # ProxyJump and per-host keys honoured

# config.yaml
inventory:
  plugin: SimpleInventory
  options:
    host_file: inventory/hosts.yaml
    group_file: inventory/groups.yaml
runner:
  plugin: threaded
  options:
    num_workers: 10

from nornir import InitNornir
from nornir_netmiko.tasks import netmiko_send_command
from nornir_utils.plugins.functions import print_result

nr = InitNornir(config_file="config.yaml")
result = nr.filter(site="syd").run(
    task=netmiko_send_command, command_string="show version", use_textfsm=True
)
print_result(result)
print(result.failed_hosts)          # dict of host -> MultiResult for hosts that raised

# inventory/hosts.yaml
sw-01.example.com:
  hostname: sw-01.example.com
  platform: ios
  groups: [syd-access]
  data: { site: syd, role: access }
# inventory/groups.yaml
syd-access:
  groups: [ios]
ios:
  platform: ios
  connection_options:
    netmiko: { extras: { secret: "" } }      # per-plugin options; credentials come from defaults.yaml or env

from nornir.core.filter import F
from nornir_napalm.plugins.tasks import napalm_get, napalm_configure
from nornir_utils.plugins.tasks.files import write_file

core = nr.filter(F(site="syd") & F(role="core") & ~F(platform="nxos"))   # boolean filters on host data
facts = core.run(task=napalm_get, getters=["facts", "interfaces_ip"])
for host, multi in facts.items():
    print(host, multi[0].result["facts"]["os_version"])

def backup(task):                                                  # a task that calls other tasks
    cfg = task.run(task=napalm_get, getters=["config"]).result["config"]["running"]
    task.run(task=write_file, filename=f"backups/{task.host}.cfg", content=cfg)

core.run(task=backup)
r = core.run(task=napalm_configure, configuration="ntp server 192.0.2.123", dry_run=True)   # diff only
print({h: res[0].diff for h, res in r.items()})

import os
from napalm import get_network_driver

driver = get_network_driver("ios")
with driver("sw-01.example.com", os.environ["NET_USER"], os.environ["NET_PASS"]) as dev:
    facts = dev.get_facts()
    interfaces = dev.get_interfaces()
    neighbours = dev.get_lldp_neighbors()
    arp = dev.get_arp_table()
    bgp = dev.get_bgp_neighbors()                    # {"global": {"peers": {ip: {"is_up": ..., "address_family": {...}}}}}
    env = dev.get_environment()                      # fans, power, temperature, cpu, memory
    counters = dev.get_interfaces_counters()         # tx/rx errors and discards per interface
    running = dev.get_config(retrieve="running", sanitized=True)["running"]   # secrets masked
    out = dev.cli(["show ip route summary", "show clock"])   # {command: output} for anything without a getter
    ok = dev.ping("192.0.2.1", source="192.0.2.2", count=5)["success"]["packet_loss"] == 0

# validate.yml
- get_facts:
    os_version: "17.12"        # substring match
- get_bgp_neighbors:
    global:
      peers:
        203.0.113.1:
          is_up: true
          address_family:
            ipv4:
              received_prefixes: { _mode: ">=", value: 10 }   # comparison operators on numbers
- get_interfaces:
    GigabitEthernet1/0/1:
      is_up: true
      is_enabled: true

report = dev.compliance_report("validate.yml")
print(report["complies"], {k: v for k, v in report.items() if isinstance(v, dict) and not v.get("complies", True)})

from ntc_templates.parse import parse_output

rows = parse_output(platform="cisco_ios", command="show ip interface brief", data=raw)
# [{'interface': 'GigabitEthernet1/0/1', 'ip_address': '198.51.100.2', 'status': 'up', 'proto': 'up'}, ...]

Value INTERFACE (\S+)
Value IP_ADDRESS (\S+)
Value STATUS (up|down|administratively down|deleted)
Value PROTO (up|down)

Start
  ^${INTERFACE}\s+${IP_ADDRESS}\s+\w+\s+\w+\s+${STATUS}\s+${PROTO} -> Record

import json
data = json.loads(conn.send_command("show ip route | json"))    # NX-OS

with driver("sw-01.example.com", user, password) as dev:
    dev.load_merge_candidate(filename="vlan.cfg")
    diff = dev.compare_config()
    if diff and approved(diff):
        dev.commit_config(revert_in=300)      # confirmed commit: reverts in 5 minutes unless confirmed
        verify(dev)                           # your own post-change checks
        dev.confirm_commit()
    else:
        dev.discard_config()

with ConnectHandler(**device) as conn:
    conn.enable()
    out = conn.send_config_set([
        "interface GigabitEthernet1/0/2",
        "description uplink to core",
        "switchport mode trunk",
    ])
    conn.save_config()               # write memory; without it the change is lost on reload

# group_vars/ios.yml
ansible_connection: ansible.netcommon.network_cli
ansible_network_os: cisco.ios.ios
ansible_become: true
ansible_become_method: enable

- name: Configure access VLANs
  hosts: ios
  gather_facts: false
  tasks:
    - name: Ensure VLANs exist
      cisco.ios.ios_vlans:
        config:
          - vlan_id: 20
            name: servers
        state: merged

    - name: Set uplink description
      cisco.ios.ios_config:
        parents: interface GigabitEthernet1/0/24
        lines:
          - description uplink to core
        backup: true
        save_when: modified

ansible-playbook vlans.yml --check --diff --limit sw-01   # show what would change, change nothing

- name: Facts, commands and resource modules
  hosts: ios
  gather_facts: false
  tasks:
    - name: Collect structured facts and the running config
      cisco.ios.ios_facts:
        gather_subset: [min, config]
        gather_network_resources: [interfaces, l2_interfaces, vlans]   # resource-module data models
      register: facts

    - name: Show commands, waiting for a condition
      cisco.ios.ios_command:
        commands:
          - show ip ospf neighbor
          - show ip bgp summary
        wait_for:
          - result[0] contains FULL              # retry until true or fail
        retries: 6
        interval: 10
      register: show

    - name: Interfaces as data, overriding what is there
      cisco.ios.ios_l2_interfaces:
        config:
          - name: GigabitEthernet1/0/2
            mode: access
            access: { vlan: 20 }
          - name: GigabitEthernet1/0/24
            mode: trunk
            trunk: { allowed_vlans: [10, 20, 30], native_vlan: 999 }
        state: replaced                          # only the listed interfaces are rewritten

    - name: What commands would the data become (no device contact)
      cisco.ios.ios_vlans:
        config: [{ vlan_id: 20, name: servers }]
        state: rendered
      register: rendered

    - name: Reach into a saved config and produce data
      cisco.ios.ios_interfaces:
        running_config: "{{ lookup('file', 'backups/sw-01.cfg') }}"
        state: parsed
      register: parsed

    - name: Free-form lines with a diff against the intended config
      cisco.ios.ios_config:
        src: templates/baseline.j2               # Jinja template rendered on the control node
        diff_against: intended
        intended_config: "{{ lookup('template', 'templates/baseline.j2') }}"
        diff_ignore_lines: ['^ntp clock-period', '^! Last configuration']
        backup: true
        backup_options: { dir_path: backups/, filename: "{{ inventory_hostname }}.cfg" }
        save_when: modified

# Replace the whole config from a rendered template, with a timed revert and post-checks
import difflib

with driver("sw-01.example.com", user, password) as dev:
    before = dev.get_config(retrieve="running")["running"]
    dev.load_replace_candidate(filename="rendered/sw-01.cfg")
    diff = dev.compare_config()
    if not diff:
        dev.discard_config(); print("no change"); raise SystemExit
    print(diff)
    dev.commit_config(revert_in=300)               # device reverts in 5 minutes unless confirmed
    checks = dev.compliance_report("validate.yml")
    if checks["complies"] and dev.ping("192.0.2.1")["success"]["packet_loss"] == 0:
        dev.confirm_commit()
    else:
        dev.rollback()                             # explicit, rather than waiting for the timer
        raise SystemExit("post-checks failed; rolled back")
    after = dev.get_config(retrieve="running")["running"]
    open("diffs/sw-01.diff", "w").writelines(difflib.unified_diff(before.splitlines(True), after.splitlines(True), "before", "after"))

# Ansible: back up, change with a diff, verify, and restore the backup on failure
- name: Change with rollback
  hosts: ios
  gather_facts: false
  tasks:
    - name: Backup
      cisco.ios.ios_config:
        backup: true
        backup_options: { dir_path: "backups/{{ inventory_hostname }}", filename: pre-change.cfg }
      check_mode: false

    - name: Copy the backup to flash for configure replace
      ansible.netcommon.net_put:
        src: "backups/{{ inventory_hostname }}/pre-change.cfg"
        dest: flash:pre-change.cfg
      check_mode: false

    - name: Apply
      block:
        - name: Push change
          cisco.ios.ios_config:
            src: templates/change.j2
            diff_against: running
        - name: Verify
          cisco.ios.ios_command:
            commands: [show ip ospf neighbor]
            wait_for: [result[0] contains FULL]
            retries: 6
            interval: 10
      rescue:
        - name: Roll back to the pre-change config
          cisco.ios.ios_command:
            commands:
              - command: configure replace flash:pre-change.cfg force
        - name: Fail the play after restoring
          ansible.builtin.fail:
            msg: "verification failed on {{ inventory_hostname }}; configuration restored"

from jinja2 import Environment, FileSystemLoader, StrictUndefined

env = Environment(loader=FileSystemLoader("templates"), trim_blocks=True, lstrip_blocks=True,
                  undefined=StrictUndefined)            # fail on a missing variable instead of rendering blank
cfg = env.get_template("switch.j2").render(hostname="sw-01", vlans=[10, 20, 30])

hostname {{ hostname }}
{% for vlan in vlans %}
vlan {{ vlan }}
 name VLAN{{ vlan }}
{% endfor %}

from concurrent.futures import ThreadPoolExecutor
from pathlib import Path

def collect(host: str) -> tuple[str, str]:
    with ConnectHandler(host=host, **base) as c:
        return host, c.send_command("show version")

with ThreadPoolExecutor(max_workers=10) as pool:      # devices and TACACS/RADIUS servers cap concurrent sessions
    for host, out in pool.map(collect, hosts):
        Path(f"out/{host}.txt").write_text(out)

fping -a -g 192.0.2.0/24 2>/dev/null                       # addresses that answer ICMP
nmap -sn 192.0.2.0/24 -oG -                                # host discovery, greppable output
snmpwalk -v2c -c "$SNMP_COMMUNITY" sw-01.example.com IF-MIB::ifDescr   # interface list via SNMP

import logging
logging.basicConfig(filename="netmiko_debug.log", level=logging.DEBUG)   # library debug log

device["session_log"] = "session.log"      # full transcript of what was sent and received

# Reachability sweep
fping -a -g 192.0.2.0/24 2>/dev/null | tee reachable.txt

# Run one command on every device and keep the output
while read -r h; do echo "== $h"; ssh -o ConnectTimeout=5 "$h" 'show version | include uptime'; done < hosts.txt

# Diff a device's running config against the repository copy
ssh sw-01.example.com 'show running-config' | diff -u configs/sw-01.cfg - | head -40

# Parse saved command output into JSON
python3 -c 'import sys,json;from ntc_templates.parse import parse_output;print(json.dumps(parse_output(platform="cisco_ios",command=sys.argv[1],data=sys.stdin.read())))' "show ip interface brief" < out.txt

# Interfaces that are down but not administratively down
ssh sw-01.example.com 'show ip interface brief' | awk '$5=="down" && $6=="down"'

# Back up every device before a change window
while read -r h; do ssh "$h" 'show running-config' > "backups/$h-$(date +%F).cfg"; done < hosts.txt

# Find a MAC address across the fleet
while read -r h; do ssh "$h" 'show mac address-table | include 0011.2233' | sed "s/^/$h /"; done < hosts.txt

# Count routes before and after a change
ssh rtr-01.example.com 'show ip route summary | include Total'

# Normalise a config for diffing (strip volatile lines)
sed -E '/^(Building configuration|Current configuration|! Last configuration change|! NVRAM config last updated|ntp clock-period)/d' sw-01.cfg > sw-01.norm.cfg

# Diff every backup against the previous day's copy and list devices that changed
for f in backups/*-"$(date +%F)".cfg; do h=${f#backups/}; h=${h%-*}; diff -q "backups/$h-$(date -d yesterday +%F).cfg" "$f" >/dev/null || echo "$h changed"; done

# Facts from every device as JSON with NAPALM's CLI
napalm --user "$NET_USER" --password "$NET_PASS" --vendor ios sw-01.example.com call get_facts

# Diff a candidate config against the device without applying it
napalm --user "$NET_USER" --password "$NET_PASS" --vendor ios sw-01.example.com configure candidate.cfg --strategy replace --dry-run

# Validate a device against a state file, exit non-zero on failure
napalm --user "$NET_USER" --password "$NET_PASS" --vendor ios sw-01.example.com validate validate.yml

# Ansible: run one show command everywhere and print output per host
ansible ios -m cisco.ios.ios_command -a 'commands="show ip interface brief"' | sed -n '/stdout_lines/,/]/p'

# Ansible: gather structured facts for one host into JSON
ansible sw-01.example.com -m cisco.ios.ios_facts -a 'gather_subset=min gather_network_resources=interfaces' > facts.json

# Ansible: back up every device with the vendor-neutral module
ansible ios -m ansible.netcommon.cli_backup -a 'dir_path=backups/'

# Ansible: what a playbook would change, for one site
ansible-playbook change.yml --check --diff --limit site_syd

# Which ntc-templates exist for a platform
python3 -c 'import ntc_templates,os;p=os.path.join(os.path.dirname(ntc_templates.__file__),"templates");print("\n".join(sorted(f for f in os.listdir(p) if f.startswith("cisco_ios"))))'

# Test a TextFSM template against saved output
python3 -c 'import sys,textfsm;print(textfsm.TextFSM(open(sys.argv[1])).ParseTextToDicts(open(sys.argv[2]).read()))' template.textfsm out.txt

# Devices whose SSH host key changed since the last run (possible replacement or MITM)
while read -r h; do ssh-keyscan -t ed25519 -T 5 "$h" 2>/dev/null | ssh-keygen -lf - ; done < hosts.txt | sort > keys.new; diff keys.old keys.new

# SSH algorithm negotiation failure on old IOS: allow legacy kex for one connection
ssh -o KexAlgorithms=+diffie-hellman-group14-sha1 -o HostKeyAlgorithms=+ssh-rsa sw-old.example.com

# Software version distribution across the fleet
while read -r h; do ssh "$h" 'show version | include Version' | head -1 | sed "s/^/$h /"; done < hosts.txt | awk '{print $NF}' | sort | uniq -c

# Interfaces with errors across the fleet (raw counters, no parser)
while read -r h; do ssh "$h" 'show interfaces | include line protocol|input errors' | paste - - | awk -v h="$h" '$NF+0>0 || $(NF-6)+0>0 {print h, $1}'; done < hosts.txt

# LLDP neighbour list from every device, as edges for a topology
while read -r h; do ssh "$h" 'show lldp neighbors detail | include System Name|Port id|Local Intf' | paste - - - | sed "s/^/$h /"; done < hosts.txt

# Uptime under a day (devices that reloaded recently)
while read -r h; do ssh "$h" 'show version | include uptime' | grep -vE 'week|day' | sed "s/^/$h /"; done < hosts.txt

# Serial numbers for an inventory or RMA
while read -r h; do ssh "$h" 'show inventory | include SN:' | head -1 | sed "s/^/$h /"; done < hosts.txt

# Batfish: check reachability against a directory of configs (needs a running batfish container)
python3 -c 'from pybatfish.client.session import Session;bf=Session();bf.init_snapshot("configs/",name="s",overwrite=True);print(bf.q.reachability(pathConstraints={"startLocation":"sw-01"},headers={"dstIps":"192.0.2.10","applications":["ssh"]}).answer().frame())'

# Count of lines per config file, to spot a truncated backup
wc -l backups/*.cfg | sort -n | head

#!/usr/bin/env python3
"""Snapshot or compare network state with NAPALM.

Usage: netstate.py snapshot hosts.txt before/
       netstate.py compare before/ after/
Credentials from NET_USER and NET_PASS.
"""
import json
import os
import sys
from concurrent.futures import ThreadPoolExecutor
from pathlib import Path

from napalm import get_network_driver

GETTERS = ["get_facts", "get_interfaces", "get_lldp_neighbors", "get_bgp_neighbors", "get_arp_table"]

def snapshot(host: str, outdir: Path) -> str:
    driver = get_network_driver(os.environ.get("NET_DRIVER", "ios"))
    with driver(host, os.environ["NET_USER"], os.environ["NET_PASS"], optional_args={"secret": os.environ.get("NET_ENABLE", "")}) as dev:
        state = {g: getattr(dev, g)() for g in GETTERS}
        routes = dev.cli(["show ip route summary"])["show ip route summary"]
    state["route_summary"] = [l for l in routes.splitlines() if l.strip().startswith(("Total", "connected", "static", "ospf", "bgp"))]
    state["interfaces_up"] = sorted(i for i, d in state["get_interfaces"].items() if d["is_up"])
    state["bgp_up"] = sorted(p for p, d in state["get_bgp_neighbors"].get("global", {}).get("peers", {}).items() if d["is_up"])
    state["lldp"] = sorted(f"{i} -> {n['hostname']}:{n['port']}" for i, ns in state["get_lldp_neighbors"].items() for n in ns)
    (outdir / f"{host}.json").write_text(json.dumps(state, indent=1, sort_keys=True, default=str))
    return host

def compare(a: Path, b: Path) -> int:
    rc = 0
    for fa in sorted(a.glob("*.json")):
        fb = b / fa.name
        if not fb.exists():
            print(f"{fa.stem}: missing in {b}"); rc = 1; continue
        sa, sb = json.loads(fa.read_text()), json.loads(fb.read_text())
        for key in ("interfaces_up", "bgp_up", "lldp", "route_summary"):
            lost, gained = sorted(set(sa[key]) - set(sb[key])), sorted(set(sb[key]) - set(sa[key]))
            if lost or gained:
                rc = 1
                print(f"{fa.stem} {key}: -{lost} +{gained}")
    return rc

if sys.argv[1] == "snapshot":
    hosts = [h.strip() for h in open(sys.argv[2]) if h.strip() and not h.startswith("#")]
    out = Path(sys.argv[3]); out.mkdir(parents=True, exist_ok=True)
    with ThreadPoolExecutor(max_workers=10) as pool:
        for fut in [pool.submit(snapshot, h, out) for h in hosts]:
            try:
                print("ok", fut.result())
            except Exception as exc:
                print("FAILED", exc, file=sys.stderr)
else:
    sys.exit(compare(Path(sys.argv[2]), Path(sys.argv[3])))

#!/usr/bin/env python3
"""Render intended configs and report drift against live devices.

Usage: drift.py devices.yml templates/ [--apply]
devices.yml: a list of {host, platform, template, vars...}; templates/<template> is a Jinja file.
"""
import os
import sys

import yaml
from jinja2 import Environment, FileSystemLoader, StrictUndefined
from napalm import get_network_driver

devices = yaml.safe_load(open(sys.argv[1]))
env = Environment(loader=FileSystemLoader(sys.argv[2]), trim_blocks=True, lstrip_blocks=True, undefined=StrictUndefined)
apply = "--apply" in sys.argv
drift = 0
for d in devices:
    intended = env.get_template(d["template"]).render(**d)
    driver = get_network_driver(d.get("platform", "ios"))
    try:
        with driver(d["host"], os.environ["NET_USER"], os.environ["NET_PASS"]) as dev:
            dev.load_replace_candidate(config=intended)
            diff = dev.compare_config()
            if not diff:
                dev.discard_config(); print(f"{d['host']}: in sync"); continue
            drift += 1
            print(f"{d['host']}: DRIFT\n{diff}\n")
            if apply:
                dev.commit_config(revert_in=180)
                if dev.ping(d.get("check_ip", "192.0.2.1"))["success"]["packet_loss"] == 0:
                    dev.confirm_commit(); print(f"{d['host']}: applied")
                else:
                    dev.rollback(); print(f"{d['host']}: rolled back, ping failed")
            else:
                dev.discard_config()
    except Exception as exc:
        drift += 1
        print(f"{d['host']}: ERROR {exc}", file=sys.stderr)
sys.exit(1 if drift else 0)

#!/usr/bin/env python3
"""Run a show command on many devices and write parsed rows to CSV.

Usage: show2csv.py hosts.txt "show interfaces status" out.csv
"""
import csv
import os
import sys
from concurrent.futures import ThreadPoolExecutor

from netmiko import ConnectHandler

hosts_file, command, out = sys.argv[1:4]
hosts = [h.strip() for h in open(hosts_file) if h.strip() and not h.startswith("#")]

def run(host):
    dev = {"device_type": os.environ.get("NET_TYPE", "cisco_ios"), "host": host, "username": os.environ["NET_USER"],
           "password": os.environ["NET_PASS"], "secret": os.environ.get("NET_ENABLE", ""), "conn_timeout": 10, "read_timeout": 60}
    try:
        with ConnectHandler(**dev) as c:
            if dev["secret"]:
                c.enable()
            rows = c.send_command(command, use_textfsm=True)
        if not isinstance(rows, list):
            return host, None, "no template match"
        return host, rows, None
    except Exception as exc:
        return host, None, str(exc)

fields, records, failures = None, [], []
with ThreadPoolExecutor(max_workers=10) as pool:
    for host, rows, err in pool.map(run, hosts):
        if err:
            failures.append((host, err)); continue
        for r in rows:
            fields = fields or ["host", *r.keys()]
            records.append({"host": host, **r})
with open(out, "w", newline="") as f:
    w = csv.DictWriter(f, fieldnames=fields or ["host"])
    w.writeheader(); w.writerows(records)
print(f"{len(records)} rows from {len(hosts) - len(failures)} devices -> {out}")
for host, err in failures:
    print(f"FAILED {host}: {err}", file=sys.stderr)
sys.exit(1 if failures else 0)