実践ネット自動化スクリプト集

現場で役立つ！Pythonによるネットワーク疎通確認スクリプト：Ping/Traceroute結果を自動解析

システムインフラの構築や運用において、ネットワークの疎通確認は基本中の基本です。手動でネットワーク機器にログインし、PingやTracerouteコマンドを実行するのは日常的な作業ですが、対象が増えたり、定期的な確認が必要になったりすると、大きな負担となります。

Pythonを使ったネットワーク自動化は、このような定型作業を効率化するための強力な手段となります。Pythonスキルをお持ちの皆様であれば、ネットワーク機器への直接的な操作経験が限定的であっても、既存のライブラリを活用することで、疎通確認の自動化を比較的容易に実現できます。

本記事では、Pythonを使ってネットワーク機器上でPingやTracerouteコマンドを実行し、その結果を自動的に解析するスクリプトの実装方法について解説します。

なぜネットワーク疎通確認を自動化するのか

• 作業負担の軽減: 手動でのコマンド実行や結果の確認は、対象機器が多いほど時間と労力がかかります。自動化により、これらの作業を削減できます。
• ヒューマンエラーの防止: 手動での入力ミスや確認漏れを防ぎ、作業の正確性を向上させます。
• 定期的な状態把握: 定期的に疎通確認を実行することで、ネットワーク経路の変化や遅延の悪化などを早期に検知できます。
• レポート作成の効率化: 自動的に取得・解析した結果を整形し、レポート作成プロセスを効率化できます。
• トラブルシューティングの迅速化: 問題発生時に複数の地点からの疎通状況をまとめて確認することで、原因特定の時間を短縮できます。

Pythonによる疎通確認自動化のステップ

Pythonでネットワーク機器上のコマンドを実行し、結果を解析するには、以下のステップが一般的です。

1. ネットワーク機器への接続: SSHなどのプロトコルを使用して機器に接続します。
2. コマンドの実行: 接続したセッションを通じて、PingやTracerouteコマンドを実行します。
3. 結果の取得: コマンドの標準出力を取得します。
4. 結果の解析: 取得した文字列から、必要な情報（成功/失敗、応答時間、経路など）を抽出します。
5. 結果の出力/処理: 解析した結果を整形して出力したり、他のシステムと連携させたりします。

これらのステップを実現するために、Pythonにはいくつかの便利なライブラリがあります。中でも、ネットワーク機器とのCLI操作に特化したnetmikoは、多くのベンダーやOSに対応しており、容易にSSH接続やコマンド実行が可能です。

netmikoを使ったPingコマンド実行と結果解析

ここでは、netmikoを使用してネットワーク機器に接続し、Pingコマンドを実行して結果を解析する基本的なスクリプト例を示します。

事前にnetmikoライブラリをインストールしておく必要があります。

pip install netmiko

以下のコードは、指定したIPアドレスの機器にSSH接続し、特定の宛先IPアドレスへのPingを実行、その結果から疎通の成否やパケットロス率を判定する例です。

import os
import re
from netmiko import ConnectHandler, NetmikoTimeoutException, NetmikoAuthenticationException

# 機器接続情報 (サンプル)
# 実際の運用では認証情報を安全に管理してください（環境変数、シークレットマネージャーなど）
DEVICE_IP = 'YOUR_DEVICE_IP' # 接続対象のネットワーク機器IP
USERNAME = 'YOUR_USERNAME'   # ユーザー名
PASSWORD = 'YOUR_PASSWORD'   # パスワード
DEVICE_TYPE = 'cisco_ios'    # 機器タイプ (例: cisco_ios, juniper_junosなど)

# Ping実行対象
TARGET_IP = '8.8.8.8' # Pingテストする宛先IPアドレス

def run_ping_test(device_ip, username, password, device_type, target_ip):
    """
    ネットワーク機器上でPingコマンドを実行し、結果を解析する関数
    """
    print(f"[{device_ip}] 宛先 {target_ip} へのPingテストを開始します...")

    device = {
        'device_type': device_type,
        'host': device_ip,
        'username': username,
        'password': password,
        'secret': password, # enableパスワードが必要な場合
    }

    try:
        # NetmikoでSSH接続
        with ConnectHandler(**device) as net_connect:
            print(f"[{device_ip}] 接続に成功しました。")

            # enableモードに移行 (必要であれば)
            # net_connect.enable()

            # Pingコマンド実行
            # コマンドは機器のOSによって異なります。ここではCisco IOSを想定。
            command = f"ping {target_ip}"
            print(f"[{device_ip}] コマンド実行: '{command}'")
            output = net_connect.send_command(command)

            print(f"[{device_ip}] コマンド出力:\n---\n{output}\n---")

            # Ping結果の解析 (Cisco IOSの出力形式を想定)
            # 解析部分は機器やOSによって適宜変更が必要です
            loss_match = re.search(r'(\d+)\% packet loss', output)
            sent_match = re.search(r'Sent: (\d+)', output)
            received_match = re.search(r'Received: (\d+)', output)
            min_rtt_match = re.search(r'min/avg/max = ([\d\.]+)/([\d\.]+)/([\d\.]+) ms', output)

            result = {
                'device': device_ip,
                'target': target_ip,
                'status': 'Unknown',
                'sent': 0,
                'received': 0,
                'loss_percent': 100,
                'rtt_min_ms': None,
                'rtt_avg_ms': None,
                'rtt_max_ms': None,
                'raw_output': output.strip()
            }

            if loss_match:
                loss_percent = int(loss_match.group(1))
                result['loss_percent'] = loss_percent

                # パケットロス率で疎通成否を判定
                if loss_percent == 0:
                    result['status'] = 'Success'
                else:
                    result['status'] = 'Failure'
                    print(f"[{device_ip}] Ping失敗: {loss_percent}% packet loss")

            if sent_match:
                result['sent'] = int(sent_match.group(1))
            if received_match:
                result['received'] = int(received_match.group(1))

            if min_rtt_match:
                result['rtt_min_ms'] = float(min_rtt_match.group(1))
                result['rtt_avg_ms'] = float(min_rtt_match.group(2))
                result['rtt_max_ms'] = float(min_rtt_match.group(3))


            print(f"[{device_ip}] Pingテスト結果: ステータス={result['status']}, パケットロス={result['loss_percent']}%")
            return result

    except NetmikoTimeoutException:
        print(f"[{device_ip}] エラー: 接続タイムアウト")
        return {'device': device_ip, 'target': target_ip, 'status': 'Connection Timeout', 'raw_output': ''}
    except NetmikoAuthenticationException:
        print(f"[{device_ip}] エラー: 認証失敗")
        return {'device': device_ip, 'target': target_ip, 'status': 'Authentication Failure', 'raw_output': ''}
    except Exception as e:
        print(f"[{device_ip}] エラー: {e}")
        return {'device': device_ip, 'target': target_ip, 'status': f'Error: {e}', 'raw_output': ''}

if __name__ == "__main__":
    # 実際の機器情報に置き換えて実行してください
    # 複数機器に対して実行する場合は、リストなどで機器情報を管理します
    # Example usage:
    # device_info = {
    #     'device_ip': '192.168.1.1',
    #     'username': 'admin',
    #     'password': 'password',
    #     'device_type': 'cisco_ios'
    # }
    # target = '8.8.8.8'
    # ping_result = run_ping_test(**device_info, target_ip=target)
    # print("\n--- Ping Result ---")
    # print(ping_result)

    print("実行には、上記のサンプルコード中の接続情報と宛先IPを適切に設定してください。")
    print("また、mainブロック内のコメントアウトされた実行例を参考に、関数を呼び出してください。")

コード解説:

• ConnectHandlerクラスを使用して、指定された機器にSSH接続を試みます。
• send_command()メソッドは、機器に対してコマンドを送信し、その実行結果（標準出力）を文字列として返します。Pingコマンドは完了まで時間がかかることがあるため、send_commandはデフォルトでコマンドプロンプトが戻ってくるまで待機します。必要に応じてcmd_timeoutなどの引数を調整してください。
• 取得した出力文字列に対して、reモジュール（正規表現）を使ってパケットロス率やRTT（Round Trip Time）の情報を抽出しています。正規表現のパターンは、使用する機器のOSやバージョンによって出力形式が異なるため、実際の機器の出力を確認して調整する必要があります。
• 接続エラーや認証エラー、その他の例外が発生した場合も適切にハンドリングし、エラーメッセージを出力しています。

netmikoを使ったTracerouteコマンド実行と結果解析

Tracerouteも基本的な考え方はPingと同じですが、出力形式がより複雑になります。経路上の各ホップのIPアドレスや応答時間などを解析する必要があります。

Tracerouteの出力例（Cisco IOS）:

Type escape sequence to abort.
Tracing the route to 8.8.8.8

  1 192.168.1.254 4 msec 4 msec 4 msec
  2 10.0.0.1 8 msec 7 msec 8 msec
  3 203.0.113.1 12 msec 11 msec 12 msec
  ...

以下のコードは、Tracerouteコマンドを実行し、各ホップの情報を抽出する例です。

import re
# Netmikoのインポート部分は上記のPingスクリプトを参照

# Traceroute実行対象
TARGET_IP_TRACEROUTE = '8.8.8.8'

def run_traceroute_test(device_ip, username, password, device_type, target_ip):
    """
    ネットワーク機器上でTracerouteコマンドを実行し、結果を解析する関数
    """
    print(f"[{device_ip}] 宛先 {target_ip} へのTracerouteテストを開始します...")

    device = {
        'device_type': device_type,
        'host': device_ip,
        'username': username,
        'password': password,
        'secret': password, # enableパスワードが必要な場合
    }

    try:
        with ConnectHandler(**device) as net_connect:
            print(f"[{device_ip}] 接続に成功しました。")

            # enableモードに移行 (必要であれば)
            # net_connect.enable()

            # Tracerouteコマンド実行
            # コマンドは機器のOSによって異なります。ここではCisco IOSを想定。
            command = f"traceroute {target_ip}"
            print(f"[{device_ip}] コマンド実行: '{command}'")
            # Tracerouteは完了まで時間がかかる可能性が高いため、timeoutを長めに設定することが推奨されます
            output = net_connect.send_command(command, cmd_timeout=120) # 例: 120秒タイムアウト

            print(f"[{device_ip}] コマンド出力:\n---\n{output}\n---")

            # Traceroute結果の解析 (Cisco IOSの出力形式を想定)
            # 各行がホップ情報を表すため、行ごとに処理します
            hops = []
            # 出力の中からホップ情報が記述されている行を抽出するための正規表現
            # 例: " 1 192.168.1.254 4 msec 4 msec 4 msec" のような行にマッチ
            # 注意: 出力形式は機器や設定により大きく異なる可能性があります
            hop_pattern = re.compile(r'^\s*(\d+)\s+([\w\.\*]+)\s+([\d\.\s]+)\s*msec') # シンプルなパターン例

            lines = output.splitlines()
            for line in lines:
                match = hop_pattern.match(line)
                if match:
                    hop_number = int(match.group(1))
                    ip_or_name = match.group(2)
                    rtt_values_str = match.group(3).strip()

                    # RTT値の解析 (複数の値がある場合がある)
                    rtt_values = [float(r) if r != '*' else None for r in rtt_values_str.split()]

                    hop_info = {
                        'hop': hop_number,
                        'target': ip_or_name,
                        'rtt_ms': rtt_values
                    }
                    hops.append(hop_info)

            if hops:
                print(f"[{device_ip}] Tracerouteテスト結果 ({len(hops)}ホップ):")
                for hop in hops:
                    print(f"  Hop {hop['hop']}: {hop['target']} (RTTs: {hop['rtt_ms']})")
                return {'device': device_ip, 'target': target_ip, 'hops': hops, 'raw_output': output.strip()}
            else:
                 print(f"[{device_ip}] Traceroute結果からホップ情報を解析できませんでした。")
                 return {'device': device_ip, 'target': target_ip, 'hops': [], 'raw_output': output.strip(), 'status': 'Parsing Error'}


    except NetmikoTimeoutException:
        print(f"[{device_ip}] エラー: 接続タイムアウト")
        return {'device': device_ip, 'target': target_ip, 'hops': [], 'raw_output': '', 'status': 'Connection Timeout'}
    except NetmikoAuthenticationException:
        print(f"[{device_ip}] エラー: 認証失敗")
        return {'device': device_ip, 'target': target_ip, 'hops': [], 'raw_output': '', 'status': 'Authentication Failure'}
    except Exception as e:
        print(f"[{device_ip}] エラー: {e}")
        return {'device': device_ip, 'target': target_ip, 'hops': [], 'raw_output': '', 'status': f'Error: {e}'}

# Tracerouteの実行例 (Pingと同様にコメントアウトを解除して実行)
# if __name__ == "__main__":
    # device_info = {
    #     'device_ip': '192.168.1.1',
    #     'username': 'admin',
    #     'password': 'password',
    #     'device_type': 'cisco_ios'
    # }
    # target = '8.8.8.8'
    # traceroute_result = run_traceroute_test(**device_info, target_ip=target)
    # print("\n--- Traceroute Result ---")
    # print(traceroute_result)

コード解説:

• Tracerouteの出力は機器や設定、宛先までの経路によって行数が可変であり、各行のフォーマットもPingより複雑です。
• ここでは、出力文字列を改行で分割し、各行に対して正規表現でマッチングを試みています。
• 正規表現は、ホップ番号、IPアドレスまたはホスト名、応答時間（RTT）の情報を抽出するように設計されています。このパターンはあくまでCisco IOSの一般的な出力を想定した例であり、実際の環境に合わせて厳密に調整する必要があります。特に、応答時間部分の*や!などの表記、ホスト名解決の有無などで出力形式は大きく変わります。
• 抽出した各ホップの情報を辞書としてリストに格納しています。

実践的な考慮点

現場でこれらのスクリプトを利用する際には、さらに以下の点を考慮することが重要です。

• 認証情報の管理: コード中に直接パスワードを記述することは非常に危険です。環境変数、設定ファイル、専用のシークレット管理ツール（HashiCorp Vault, CyberArk, AWS Secrets Managerなど）を利用して、安全に認証情報を管理してください。
• 複数機器への対応: 実際の環境では、複数のネットワーク機器に対して同時に疎通確認を実行することが多いでしょう。機器のリストを管理し、ループ処理や、より高度なフレームワーク（Nornirなど）を利用して並列実行を検討します。
• エラーハンドリングの強化: 接続失敗、認証失敗、コマンド実行エラー、コマンド出力の予期しない形式など、様々なエラーケースを想定し、適切なログ出力や通知を行う必要があります。
• 出力形式: スクリプトの実行結果を人間が読みやすい形式（表形式など）や、他のツールで利用しやすい形式（CSV, JSONなど）で出力できるようにすると、活用の幅が広がります。
• 定期実行と監視連携: スクリプトをCron（Linux）やタスクスケジューラ（Windows）で定期実行し、その結果を監視システム（Zabbix, Prometheus+Alertmanagerなど）に連携させることで、ネットワークの状態変化を継続的に監視できます。
• コマンド出力の多様性: ネットワーク機器のベンダー（Cisco, Juniper, Arista, FortiNetなど）やOS、バージョンによって、PingやTracerouteコマンドの出力形式は大きく異なります。汎用的なスクリプトを目指す場合は、複数の出力形式に対応できる解析ロジックを実装するか、より抽象化されたライブラリ（NAPALMなど）の利用を検討します。ただし、NAPALMが全ての疎通確認機能を提供しているわけではないため、CLIコマンド実行と組み合わせる必要がある場合が多いです。
• APIの活用: 最新のネットワーク機器は、CLIだけでなくAPI（RESTConf, NETConf, ベンダー固有API）を提供しています。可能な場合は、CLI解析よりも構造化されたデータを取得できるAPIの利用を検討すると、解析処理が容易になり、より信頼性の高い自動化が実現できます。ただし、Ping/Tracerouteのような診断コマンドがAPIで直接提供されているかは、機器の仕様によります。

インフラ自動化における位置づけ

ネットワーク疎通確認の自動化スクリプトは、インフラ自動化のワークフローにおいて様々な場面で活用できます。

• プロビジョニング後の確認: IaCツール（Terraform, Ansibleなど）でネットワーク機器の設定変更や新しいVPC/VNetの作成を行った後に、期待通りに通信が可能になったかを確認するための自動テストとして組み込むことができます。
• デプロイメントパイプライン: アプリケーションのデプロイメントやインフラ変更を含むCI/CDパイプラインに、ネットワーク疎通確認のステップを組み込むことで、変更によって発生したネットワーク問題を早期に検知し、デプロイの失敗やロールバックを判断する基準とすることができます。
• 定期ヘルスチェック: 運用中のシステムに対して定期的に実行し、ネットワークの状態を監視するヘルスチェックの一環として利用できます。

まとめ

本記事では、Pythonとnetmikoライブラリを使用して、ネットワーク機器上でPingおよびTracerouteコマンドを実行し、その結果を自動的に解析する基本的なスクリプトの実装方法について解説しました。

ネットワーク機器への直接的な操作経験が限定的であっても、Pythonの豊富なライブラリと文字列処理、正規表現のスキルを組み合わせることで、現場で役立つ実用的なネットワーク自動化スクリプトを作成することが可能です。

ここで示したコード例は基本的なものですが、これを基に、認証情報の安全な管理、複数機器への対応、詳細なエラーハンドリング、そして結果の多様な出力形式への対応などを実装していくことで、より堅牢で実践的な自動化ツールへと発展させることができます。

ネットワーク自動化は、インフラ運用の効率化と品質向上に不可欠な要素です。ぜひPythonスキルを活かして、ネットワークの自動化に取り組んでみてください。