Netmiko SSH Automation（Netmiko SSH自動化）
このスキルは、Netmikoを使用してネットワークデバイスに接続するPython自動化コードの作成・レビュー時に使用します。デフォルトのパスは読み取り専用に保ちます。設定変更には、個別の変更ウィンドウ、ピアレビュー、ロールバック計画が必要です。
使用するタイミング
- ルーター、スイッチ、ファイアウォール全体からの
showコマンド出力の収集 - インターフェース、ルーティング、設定エビデンスのための小規模な監査スクリプトの作成
- ネットワークSSHスクリプトへのタイムアウトと例外処理の追加
- テンプレートが存在する場合のTextFSMによるコマンド出力のパース
- 本番デバイスに影響を与える前の自動化のレビュー
安全デフォルト
- 読み取り専用の
send_command()収集から開始する - インベントリは小さく明示的に保ち、アドレス範囲全体をスイープしない
- 環境変数、ボールト、
getpassを使用し、認証情報をハードコードしない - 接続タイムアウトと読み取りタイムアウトを設定する
- 古いデバイスに過負荷がかからないよう、同時実行数を制限する
send_config_set()の前に明示的なオペレータフラグを要求する- 変更が確認・承認されるまで
save_config()を呼び出さない
読み取り専用接続パターン
import os
from getpass import getpass
from netmiko import ConnectHandler
from netmiko.exceptions import (
NetmikoAuthenticationException,
NetmikoTimeoutException,
ReadTimeout,
)
device = {
"device_type": "cisco_ios",
"host": "192.0.2.10",
"username": os.environ.get("NETMIKO_USERNAME") or input("Username: "),
"password": os.environ.get("NETMIKO_PASSWORD") or getpass("Password: "),
"secret": os.environ.get("NETMIKO_ENABLE_SECRET"),
"conn_timeout": 10,
"auth_timeout": 20,
"banner_timeout": 15,
"read_timeout_override": 30,
}
try:
with ConnectHandler(**device) as conn:
if device.get("secret") and not conn.check_enable_mode():
conn.enable()
output = conn.send_command("show ip interface brief", read_timeout=30)
print(output)
except NetmikoAuthenticationException:
print("認証に失敗しました")
except NetmikoTimeoutException:
print("SSH接続がタイムアウトしました")
except ReadTimeout:
print("コマンド読み取りがタイムアウトしました")
例ではドキュメント用のプレースホルダアドレスを使用してください。実際のインベントリは、無視リストに登録されたローカルファイルまたはシークレット管理システムに保管します。
バッチ収集
from concurrent.futures import ThreadPoolExecutor, as_completed
from typing import Any
def collect_show(device: dict[str, Any], command: str) -> dict[str, Any]:
host = device["host"]
try:
with ConnectHandler(**device) as conn:
output = conn.send_command(command, read_timeout=45)
return {"host": host, "ok": True, "output": output}
except (NetmikoAuthenticationException, NetmikoTimeoutException, ReadTimeout) as exc:
return {"host": host, "ok": False, "error": type(exc).__name__}
results = []
with ThreadPoolExecutor(max_workers=8) as pool:
futures = [pool.submit(collect_show, device, "show version") for device in devices]
for future in as_completed(futures):
results.append(future.result())
デバイス資産とAAAシステムがより高い接続ボリュームを処理できると分かっている場合を除き、max_workers は低く抑えてください。
構造化パース
Netmikoは、サポートされているコマンド出力のパースをTextFSM、TTP、Genieに依頼できます。パーサーの出力は最適化として扱い、唯一のエビデンスパスとしては扱わないでください。
with ConnectHandler(**device) as conn:
parsed = conn.send_command(
"show ip interface brief",
use_textfsm=True,
raise_parsing_error=False,
read_timeout=30,
)
if isinstance(parsed, str):
print("パーサーテンプレートが見つかりませんでした。レビュー用に生の出力を保存します")
else:
for row in parsed:
print(row)
パース結果がブロッキング判断の基準となる場合は、生のコマンド出力をパース結果と一緒に保存し、オペレーターが不一致を検査できるようにします。
ガード付き設定パターン
import os
commands = [
"interface GigabitEthernet0/1",
"description CHANGE-1234 UPLINK-TO-CORE",
]
apply_changes = os.environ.get("APPLY_NETWORK_CHANGES") == "1"
if not apply_changes:
print("ドライランです。候補コマンド:")
print("\n".join(commands))
else:
with ConnectHandler(**device) as conn:
conn.enable()
before = conn.send_command("show running-config interface GigabitEthernet0/1")
output = conn.send_config_set(commands)
after = conn.send_command("show running-config interface GigabitEthernet0/1")
print(before)
print(output)
print(after)
print("起動設定を保存する前に動作を確認してください。")
設定の保存は別途承認ステップです。本番環境では、ロールバックスニペットを含め、変更レコードに変更前後のエビデンスを記録してください。
レビューチェックリスト
- スクリプトは明示的なインベントリソースを特定しているか？
- 認証情報がソース、ログ、例外メッセージに含まれていないか？
conn_timeout、auth_timeout、コマンドread_timeoutが設定されているか？- バッチ全体を停止させることなく、デバイスごとに失敗が報告されているか？
- スクリプトは幅広いスキャンや無制限の同時実行を避けているか？
- 設定変更はドライランまたは明示的なオペレータフラグの後ろにあるか？
save_config()は初期プッシュから分離され、検証と関連付けられているか？
アンチパターン
- パスワード、イネーブルシークレット、秘密鍵をソースにハードコーディングする
- 設定コマンドをデフォルトのコードパスとして送信する
- レビュー済みインベントリではなくCIDR範囲に対して自動化を実行する
- サニタイズなしで完全なランニングコンフィグを共有システムにログ出力する
- パーサーの成功をデバイス状態が正しいことの証明として扱う
関連情報
- スキル:
cisco-ios-patterns - スキル:
network-config-validation - スキル:
network-interface-health