Pramiko/Netomiko などデバイス接続を簡単にするライブラリはありますが、取得した情報の取り扱いは結構大変だと思います。それに対して、今更ながらNAPALMがあると聞いたので調べました。

この記事のcodeはsourjp/lab-networkにアップロードしました。

NAPALM?

NAPALM (Network Automation and Programmability Abstraction Layer with Multivendor support) is a Python library that implements a set of functions to interact with
different router vendor devices using a unified API.

NAPALM は必要とされることが多い情報取得をメソッドとして用意しており、マルチベンダー間の差分を吸収します。しかも、その結果を dict に格納して返してくれます。つまり取得した情報がパース済みなので自動化が捗りそうです。

サポートデバイス

詳細はここにリスト化されています。抜粋すると次のようです。

Arista: EOS
Juniper: JUNOS
Cisco: IOS-XR, NX-OS, NX-OS SSH, IOS

面白いのがBackend Libraryで、各装置への接続はベンダに適した(?)ライブラリを選択しています。Junos なら PyEZ を選択し、特にない場合は Netmiko で接続をラップしているようです。

基本的な使い方

仕組みは次の通りで簡単です。

connection を作成
コマンドを送信、もしくは実装ずみのメソッドを呼び出す

Netmiko などと同じ雰囲気ですね。とりあえず vSRX を使いながらサンプルを見てみましょう。

from napalm import get_network_driver
import pprint

vsrx1_params = {
    "hostname": "localhost",
    "username": "root",
    "password": "Juniper",
    "optional_args": {"port": "2201"}
}

if __name__ == "__main__":
    junos_driver = get_network_driver("junos")
    conn = junos_driver(**vsrx1_params)
    conn.open()
    arps = conn.get_arp_table()
    for arp in arps:
        print(type(arp))
        pprint.pprint(arp, indent=2)
    conn.close()

特筆すべきことは Netmiko はベンダごとのコマンドを書いて投入しますが、NAPALM はベンダごとの違いを吸収するメソッドが用意されています。ここのGettersSupportMatrixを見てみると、機器/IF/Route/BGP の状態は用意されているようです。MPLS や VRF など複雑になると流石に対応していませんが、基本的な情報はカバーできるのは便利に感じます。

上記のサンプルから出力された結果は次の通りで、内容が同じなことはもちろんですが、dict型で格納されていることが大変便利です。

$ python main.py
<class 'dict'>
{ 'age': 498.0,
  'interface': 'ge-0/0/0.0',
  'ip': '10.0.2.2',
  'mac': '52:54:00:12:35:02'}
<class 'dict'>
{ 'age': 751.0,
  'interface': 'ge-0/0/0.0',
  'ip': '10.0.2.3',
  'mac': '52:54:00:12:35:03'}
<class 'dict'>
{ 'age': 1253.0,
  'interface': 'ge-0/0/1.0',
  'ip': '192.168.0.2',
  'mac': '08:00:27:B4:29:98'}

root@vsrx1> show arp expiration-time
MAC Address       Address         Name                      Interface           Flags    TTE
52:54:00:12:35:02 10.0.2.2        10.0.2.2                  ge-0/0/0.0          none 452
52:54:00:12:35:03 10.0.2.3        10.0.2.3                  ge-0/0/0.0          none 1096
08:00:27:b4:29:98 192.168.0.2     192.168.0.2               ge-0/0/1.0          none 158
Total entries: 3

ちなみにopen() と close() を自動でしてくれるので、python 的にはwithで書くのがスマートかと思います。

if __name__ == "__main__":
    junos_driver = get_network_driver("junos")
    with junos_driver(**vsrx1_params) as conn:
        arps = conn.get_arp_table()
        for arp in arps:
            print(type(arp))
            pprint.pprint(arp, indent=2)

より詳細な使い方

雰囲気は伝わったかと思うのでより具体的にみていきましょう。

状態の確認

いくつか試してみます。

get_bgp_neighbors()

BGP neighbor の情報を取得する場合です。

if __name__ == "__main__":
    junos_driver = get_network_driver("junos")

    with junos_driver(**vsrx1_params) as conn:
        bgp = conn.get_bgp_neighbors()
        print(type(bgp))
        pprint.pprint(bgp)

必要そうな情報が取れていますね。

root@vsrx1> show bgp summary
Groups: 1 Peers: 2 Down peers: 0
Table          Tot Paths  Act Paths Suppressed    History Damp State    Pending
inet.0                 2          0          0          0          0          0
Peer                     AS      InPkt     OutPkt    OutQ   Flaps Last Up/Dwn State|#Active/Received/Accepted/Damped...
192.168.1.2           65002        251        253       0       0     1:52:11 0/1/1/0              0/0/0/0
192.168.2.2           65002        250        251       0       0     1:52:07 0/1/1/0              0/0/0/0

$ python main.py
<class 'dict'>
{'global': {'peers': {'192.168.1.2': {'address_family': {'ipv4': {'accepted_prefixes': 1,
                                                                  'received_prefixes': 1,
                                                                  'sent_prefixes': 1},
                                                         'ipv6': {'accepted_prefixes': -1,
                                                                  'received_prefixes': -1,
                                                                  'sent_prefixes': -1}},
                                      'description': '',
                                      'is_enabled': True,
                                      'is_up': True,
                                      'local_as': 65001,
                                      'remote_as': 65002,
                                      'remote_id': '10.0.0.2',
                                      'uptime': 6699},
                      '192.168.2.2': {'address_family': {'ipv4': {'accepted_prefixes': 1,
                                                                  'received_prefixes': 1,
                                                                  'sent_prefixes': 1},
                                                         'ipv6': {'accepted_prefixes': -1,
                                                                  'received_prefixes': -1,
                                                                  'sent_prefixes': -1}},
                                      'description': '',
                                      'is_enabled': True,
                                      'is_up': True,
                                      'local_as': 65001,
                                      'remote_as': 65002,
                                      'remote_id': '10.0.0.2',
                                      'uptime': 6695}},
            'router_id': '10.0.0.1'}}

get_bgp_neighbors()

切り分けで interface あたりの MAC アドレスを知りたい場合があるかもしれません。

if __name__ == "__main__":
    junos_driver = get_network_driver("junos")

    with junos_driver(**vsrx1_params) as conn:
        interfaces = conn.get_interfaces()
        for k, v in interfaces.items():
            if k.startswith("ge-"):
                print(k, v.get("mac_address"))

dict で取得できるのは次のように簡単に情報を絞り込めるので便利です。

$ python main.py
ge-0/0/0 08:00:27:AE:F4:51
ge-0/0/1 08:00:27:14:98:E9
ge-0/0/2 08:00:27:65:07:98
ge-0/0/3 08:00:27:5D:86:04
ge-0/0/0.0 08:00:27:AE:F4:51
ge-0/0/1.0 08:00:27:14:98:E9
ge-0/0/2.0 08:00:27:65:07:98
ge-0/0/3.0 08:00:27:5D:86:04

traceroute()

使う機会があるかわかりませんが、想定通りの経路かどうか正常性確認のためにバッチ処理させるとかあるでしょうか？
neighbor 接続の設定前に生きているかの確認ならping()がシンプルかもしれません。

if __name__ == "__main__":
    junos_driver = get_network_driver("junos")

    with junos_driver(**vsrx1_params) as conn:
        resp = conn.traceroute(destination="10.0.0.2")
        pprint.pprint(resp)

$ python main.py
{'success': {1: {'probes': {1: {'host_name': '10.0.0.2',
                                'ip_address': '10.0.0.2',
                                'rtt': 20.16},
                            2: {'host_name': '10.0.0.2',
                                'ip_address': '10.0.0.2',
                                'rtt': 15.726},
                            3: {'host_name': '10.0.0.2',
                                'ip_address': '10.0.

cli()

メソッドで対応していない部分はcli()を利用します。これも dict 型で key(CMD):value(string)と返ってくるため、value を'で split すれば扱うことができそうです。

if __name__ == "__main__":
    junos_driver = get_network_driver("junos")

    CMD = ["show version and haiku",
           "show system uptime",
           "show system alarms"]

    with junos_driver(**vsrx1_params) as conn:
        resp = conn.cli(CMD)
        print(type(resp))
        pprint.pprint(resp)

$ python main.py
<class 'dict'>
{'show system alarms': '\n'
                       '1 alarms currently active\n'
                       'Alarm time               Class  Description\n'
                       '2020-12-13 09:40:40 UTC  Minor  Rescue configuration '
                       'is not set\n',
 'show system uptime': '\n'
                       'Current time: 2020-12-14 05:50:56 UTC\n'
                       'System booted: 2020-12-13 09:40:17 UTC (20:10:39 ago)\n'
                       'Protocols started: 2020-12-13 09:40:39 UTC (20:10:17 '
                       'ago)\n'
                       'Last configured: 2020-12-14 05:19:30 UTC (00:31:26 '
                       'ago) by root\n'
                       ' 5:50AM  up 20:11, 1 user, load averages: 0.00, 0.01, '
                       '0.00\n',
 'show version and haiku': '\n'
                           'Hostname: vsrx1\n'
                           'Model: firefly-perimeter\n'
                           'JUNOS Software Release [12.1X47-D15.4]\n'
                           '\n'
                           '\n'
                           '        IS-IS sleeps.\n'
                           '        BGP peers are quiet.\n'

エラーの場合

get_environment()を試すと次のようにエラーが出ました。vSRX が原因な気がしますが、マトリクス表で○ であってもテストは必要そうですね。

$ python main.py
NAPALM didn't catch this exception. Please, fill a bugfix on https://github.com/napalm-automation/napalm/issues
Don't forget to include this traceback.
Traceback (most recent call last):
...
  File "/Users/sourjp/go/src/github.com/sourjp/lab-network/.venv/lib/python3.7/site-packages/jnpr/junos/device.py", line 854, in execute
    raise EzErrors.RpcError(cmd=rpc_cmd_e, rsp=rsp, errs=ex)
jnpr.junos.exception.RpcError: RpcError(severity: error, bad_element: None, message: command is not valid on the firefly-perimeter)

状態の変更

状態の変更には以下のコマンドセットが用意されています。ここら辺はベンダごとのパッケージでオーバーライドして文法が合うように対応されます。JUNOSの場合は次の通りです。

CMD	説明
load_replace_candidate()	引数に filename or config を与えて`設定全体を入れ替え`。階層形式のみ対応。
load_merge_candidate()	引数に filename or config を与えて`設定差分を追加`。階層形式,set, xml, json に対応。
compare_config()	candidate と runnning の差分出力
commit_config()	反映
discard_config()	変更を中止
rollback()	変更を１世代前に戻す

上記のコマンドセットから、以下のようにプログラムを用意するのが良さそうです。

変更
差分を表示
[y/N]を選択
y なら変更、N なら中止

BGP neighbor を設定して、Neighbor Up の確認までを簡単に書いてみました。

if __name__ == "__main__":
    junos_driver = get_network_driver("junos")

    CMD = [
        "set protocols bgp group ebgp type external",
        "set protocols bgp group ebgp export export-route",
        "set protocols bgp group ebgp peer-as 65002",
        "set protocols bgp group ebgp neighbor 192.168.1.2",
        "set policy-options policy-statement export-route term 1 from route-filter 10.0.0.1/32 exact",
        "set policy-options policy-statement export-route term 1 then accept"]

    with junos_driver(**vsrx1_params) as conn:
        target = "192.168.1.2"

        print(f"Test connectivity to {target}")
        ping = conn.ping(target)
        if ping["success"]["packet_loss"] != 0:
            print(f"Couldn't reach to neigbor {target}")
        print(f"Found {target}")

        print("Start installing config.")
        for cmd in CMD:
            conn.load_merge_candidate(config=cmd)

        diff = conn.compare_config()
        print("Diff:\n" + diff)

        while True:
            choice = input("Would you like to commit these changes? [y/N}: ")
            if choice in ["y", "yes", "Y"]:
                conn.commit_config()
                print("Config has commited.")
                break
            elif choice in ["n", "No", "N"]:
                conn.discard_config()
                print("Abort.")
                break

        print("Check BGP neighbor.")
        count = 0
        while count < 6:
            time.sleep(2)
            try:
                bgp = conn.get_bgp_neighbors()
                info = bgp["global"]["peers"].get(target)
                if info["is_enabled"]:
                    print(f"BGP neighbor {target} has up.")
                    break
            except BaseException:
                raise ValueError(f"Couldn't find {target} in BGP neighbors.")
            count += 1

        if count >= 5:
            print("Couldn't confirm BGP neighbor up. Check it directly.")
        print("Done")

操作画面の出力です。

$ python load_merge_config.py
Test connectivity to 192.168.1.2
Found 192.168.1.2
Start installing config.
Diff:
[edit protocols]
+   bgp {
+       group ebgp {
+           type external;
+           export export-route;
+           peer-as 65002;
+           neighbor 192.168.1.2;
+       }
+   }
[edit]
+  policy-options {
+      policy-statement export-route {
+          term 1 {
+              from {
+                  route-filter 10.0.0.1/32 exact;
+              }
+              then accept;
+          }
+      }
+  }
Would you like to commit these changes? [y/N}: y
Check BGP neighbor up.
BGP neighbor 192.168.1.2 has up.
Done.

今までの情報を整理するだけで簡単にかけますね。Netmiko ですと、上のような BGP Neighbor Up を確認するためにパースするのも大変ですが、ライブラリに任せられるのは楽です。

設定の投入にあたっては、list ではなく string で与える必要があります。ソースコードを読むと、最初の文字をみてフォーマットを判断するのですが、strip()やsplit()は list 型に対応していません。

   def _detect_config_format(self, config):
       ...
        if config.strip().startswith("<"):
            return "xml"
        elif config.strip().split(" ")[0] in set_action_matches:
            return "set"
        elif self._is_json_format(config):
            return "json"
        return fmt

Validation

もう一つ興味深い機能が Validation で、YAML ファイルにその装置の状態がどうあるべきか？を宣言し、それ通りになっているかを確認します。詳細はここにあります。

使い方は NAPALM で用意しているメソッドの取得結果の期待値を YAML に設定し、それをcompliance_report()で読み込みます。

とりあえず BGP neighbor のチェックをしたサンプルです。

if __name__ == "__main__":
    junos_driver = get_network_driver("junos")

    with junos_driver(**vsrx1_params) as conn:
        report = conn.compliance_report("compliance.yaml")
        pprint.pprint(report, width=40, indent=1)

---
- get_bgp_neighbors:
    global:
      peers:
        _mode: strict
        192.168.1.2:
          is_enabled: true
          address_family:
            ipv4:
              sent_prefixes: ">=1"
              received_prefixes: ">=1"

- ping:
    _name: ping_neighbor
    _kwargs:
      destination: 192.168.1.2
      source: 192.168.1.1
    success:
      packet_loss: 0
    _mode: strict

出力結果が次の通りで、compliesがTrueになっていると YAML で定義した状態通りになっています。

$ python validation.py
{'complies': True,
 'get_bgp_neighbors': {'complies': True,
                       'extra': [],
                       'missing': [],
                       'present': {'global': {'complies': True,
                                              'nested': True}}},
 'ping_neighbor': {'complies': True,
                   'extra': [],
                   'missing': [],
                   'present': {'success': {'complies': True,
                                           'nested': True}}},
 'skipped': []}

もし validation に失敗すると次のように失敗した場所が確認できました。出力がとても長いので抜粋しています。

$ python validation.py {'complies': False,
 'get_bgp_neighbors': {'complies': False,
...
    'diff': {'complies': False,
                'extra': [],
                'missing': [],
                'present': {'received_prefixes': {'actual_value': 0,
                                                'complies': False,
                                                'expected_value': '>=1',
                                                'nested': False},
                            'sent_prefixes': {'complies': True,
                                            'nested': False}}},
...
 'ping_neighbor': {'complies': True,
                   'extra': [],
                   'missing': [],
                   'present': {'success': {'complies': True,
                                           'nested': True}}},
 'skipped': []}

監視と重なる部分があるかもしれませんが、デバイス当たりの状態を定義してそれを確認できるのは活用できるかもしれません。

CLI

NAPALM はコマンドも用意されています。そのため、デバッグや簡単に接続を試す時に利用できます。
詳細はここに書いてあります。

基本的な使い方は次の引数を与えます。

接続に必要な情報(--vendor, --user, --password, --optional_args)
命令(configure, call, validate)

また--debugを入れることで、接続性のデバッグが可能になります。

(lab-network) (base) sour:netmiko_lab sourjp$ napalm --debug --vendor junos --user root --password Juniper --optional_args 'port=2201' localhost call get_facts
2020-12-14 19:11:29,451 - napalm - DEBUG - Starting napalm's debugging tool
2020-12-14 19:11:29,451 - napalm - DEBUG - Gathering napalm packages
2020-12-14 19:11:29,452 - napalm - DEBUG - napalm==3.2.0
2020-12-14 19:11:29,452 - napalm - DEBUG - get_network_driver - Calling with args: ('junos',), {}
2020-12-14 19:11:29,452 - napalm - DEBUG - get_network_driver - Successful
2020-12-14 19:11:29,452 - napalm - DEBUG - __init__ - Calling with args: (<class 'napalm.junos.junos.JunOSDriver'>, 'localhost', 'root'), {'password': '*******', 'timeout': 60, 'optional_args': {'port': 2201}}
2020-12-14 19:11:29,456 - napalm - DEBUG - __init__ - Successful
2020-12-14 19:11:29,456 - napalm - DEBUG - pre_connection_tests - Calling with args: (<napalm.junos.junos.JunOSDriver object at 0x104d8c6d0>,), {}
2020-12-14 19:11:29,456 - napalm - DEBUG - open - Calling with args: (<napalm.junos.junos.JunOSDriver object at 0x104d8c6d0>,), {}
2020-12-14 19:11:30,096 - napalm - DEBUG - open - Successful
2020-12-14 19:11:30,097 - napalm - DEBUG - connection_tests - Calling with args: (<napalm.junos.junos.JunOSDriver object at 0x104d8c6d0>,), {}
2020-12-14 19:11:30,097 - napalm - DEBUG - get_facts - Calling with args: (<napalm.junos.junos.JunOSDriver object at 0x104d8c6d0>,), {}
2020-12-14 19:11:32,353 - napalm - DEBUG - Gathered facts:
{
    "vendor": "Juniper",
...
        "vlan"
    ]
}
2020-12-14 19:11:33,083 - napalm - DEBUG - method - Successful
2020-12-14 19:11:33,083 - napalm - DEBUG - close - Calling with args: (<napalm.junos.junos.JunOSDriver object at 0x104d8c6d0>,), {}
2020-12-14 19:11:33,213 - napalm - DEBUG - close - Successful
2020-12-14 19:11:33,213 - napalm - DEBUG - post_connection_tests - Calling with args: (<napalm.junos.junos.JunOSDriver object at 0x104d8c6d0>,), {}

まとめ

いかがでしょうか？マルチベンダーと言いつつ使い慣れた JUNOS ばかりで動作確認してしまいましたが、データの取得には使い勝手の良さを感じました。
ただ対応しているベンダの数は Netmiko と比べると少ないので、基本は NAPALM を使いつつ、足りない部分は Netmiko でカバーするのが良いかもしれません。

参考

続・マルチベンダルータ制御 API ライブラリ NAPALM を触ってみた