Python覚書集 – 第1回 telnetlibを使ってみた

1.はじめに


現在参画している自動化案件ではPythonを使っています。
ついては良く使われているライブラリやネットでも情報が薄いもの辺りをターゲットとして情報発信していきたいと思います。

2.telnetlibとは


今回はtelnetlibと言う標準で組み込まれているライブラリについて紹介したいと思います。
資格取得者はお馴染みの本家サイトにもバッチリ情報公開されています。

■21.19. telnetlib — Telnet クライアント

https://docs.python.jp/3/library/telnetlib.html?highlight=telnetlib#module-telnetlib 現場での使われ方としては、下記のような人が行っていた処理手順を自動化(Script化)するのに使っています。
①Cisico機器や対向サーバにtelnet接続
②コマンド実行(設定の更新など)
③コマンドの結果が正しいか検証 従来のメジャー(身近)な方法だと大きく2つあると思います。
(1)TeraTermのマクロ(ttl)を使う
(2)Shellを使う ここでPython(telnetlib)使うメリットあるの?と疑問を持った方がいると思います。
一部個人的な見解もありますが、以下のようなメリットがあると考えています。
– プラットフォームを選ばない
Javaみたいにとはいきませんし、Cygwinとか組み合わせ考えたら別ですが。
– telnet専用のライブラリだけあって検証などもし易い設計になっている。(例外もあるし)
– 言語仕様が充実
複雑な検証をするにもライブラリが充実しているし、Python自体がちょっとしたツールレベルのものでないですし。
(TeraTermのマクロは言語仕様が貧弱過ぎて検証や保守するには止めておいた方が良いです by 宮坂) じゃあデメリットはないのか?と言う方もいると思います。まぁ、、無いと言えば嘘になります;
– 本家サイトに記載のあるサンプルだと動かない(遠い目)
– ネット情報が少ない(英語でさえも本家同様のものしかみたことないです) 辺りかと思います。
これが今回どんな感じで使っているのか?を展開しようとした背景です。

3.検証って何ぞや?


ご存知の方も多いかも知れませんが、自動化における検証と言うのは
いわゆる手順書に載っている確認内容の部分のことになります。
これをRPAならぬRBA(ランブック自動化:Run Book Automation)と言います。 ■IT運用管理プロセスを自動化し、ビジネス変化の速さに備えるRBA
https://it.impressbm.co.jp/articles/-/9034 人が見れば一目瞭然なこともシステム的に行うとなると結構大変です。
例えば現状の手順書が『ls -l』の結果に対して「A、B、Cのファイルが存在していてパーミッションは640であること」
となっているとして、目視、読み合わせであれば検証できますが、プログラム的にとなると複数行の結果を解析する必要があります。
⇒なので期待値も書かれていて、システム的にも検証しやすい手順書であれば良いですが、
そうでない場合、期待値の確認して、検証仕様を新たに詰める必要があります。

★↓↓↓そこでtelnetがいよいよ登場です。↓↓↓

4.サンプルソース


お待ちかねのサンプルソースです。今回は説明は省略します。
時間のある方はどんな動きになるのか、どんな結果になるのか想像したり実際に動かしてみてください。(telnetサービスが必要です。)
[Telnetlib_Sample.py]
#! /usr/bin/env python3
# -*- coding: utf-8 -*-
'''
Telnetテスト
'''
import telnetlib
import time
import traceback
 
RETURN_OK = 0
RETURN_NG = 1
 
class Telnet_Test():
	"""
	telnet実行クラス
	"""
	def __init__(self, hostname, username, password, prompt,
		connect_timeout = 5, reply_timeout = 5, encode = "utf-8",
		interval_time = 5, try_maximum = 2):
	"""
	コンストラクタ
	hostname in str ホスト名(必須)
	username in str ユーザ名(必須)
	password in str パスワード(必須)
	prompt in str ログイン後初回プロンプト(必須)
	connect_timeout in int 接続待機時間(defaultは10)
	reply_timeout in int コマンド応答待機時間(defaultは5)
	encode in str エンコード(defaultはutf-8)
	interval_time in int リトライまでの待機時間(defaultは10)
	try_maximum in int 最大トライ回数(defaultは3)
	"""
 
	# 初期化
	self.hostname = hostname
	self.username = username
	self.password = password
	self.prompt = prompt
	self.connect_timeout = connect_timeout
	self.reply_timeout = reply_timeout
	self.encode = encode
	self.interval_time = interval_time
	self.try_maximum = try_maximum
 
	def __exec_command(self, command):
		"""
		telnet コマンド実行
		command in str コマンド
		"""
 
		# コマンド補正
		buf = command.encode(self.encode) + b"\n"
		# コマンド実行
		self.telnet.write(buf)
 
	def __login(self, recv_message, send_message, prompt, command):
		"""
		telnet ログイン共通処理
		recv_message in str 受信メッセージ
		send_message in str 送信メッセージ
		prompt in str プロンプト
		command in str コマンド
		"""

		# 待ちメッセージ
		print("%sプロンプト待ち 接続先 = %s, タイムアウト = %d"
			% (recv_message, self.hostname, self.reply_timeout))
		# プロンプト読み込み
		responce = self.telnet.read_until(prompt.encode(self.encode),
					self.reply_timeout)
		# 改行文字で分割
		line_list = responce.decode(self.encode).splitlines()
		# 受信メッセージ
		print("%sプロンプト受信 接続先 = %s, リプライ = "
		% (recv_message, self.hostname), end="")
 
		# ログイン不可の場合
		if not line_list[-1].endswith(prompt):
		raise EOFError
 
		if send_message != "":
		# 送信メッセージ
		print("%s送信 接続先 = %s, ユーザ名 = %s"
			% (send_message, self.hostname, self.username))
		# コマンド実行
		self.__exec_command(command)
 
	def __check(self):
		"""
		引数チェック
		"""
 
		# 戻り値初期化
		result_stat = RETURN_OK
		# 必須チェック対象引数
		check_argus1 = {"hostname":self.hostname, "username":self.username,
				"password":self.password, "prompt":self.prompt}
		# 数値チェック対象引数
		check_argus2 = {"connect_timeout":self.connect_timeout,
				"reply_timeout":self.reply_timeout,
				"interval_time":self.interval_time,
				"try_maximum":self.try_maximum}
		# 必須チェック
		for k, v in check_argus1.items():
			if len(v) == 0:
				print("引数 %s は必須項目です" % (k))
				result_stat = RETURN_NG
		# 数値チェック
		for k, v in check_argus2.items():
			if type(v) != int:
				print("引数 %s は数値です" % (k))
				result_stat = RETURN_NG
 
		# 戻り値
		return result_stat
 
	def exec(self, prompt, command_list):
		"""
		telnet実行
		prompt in str プロンプト
		command_list in list コマンドリスト
		戻り値 int 0はTrue、1はFalse
		"""

		# 引数チェック
		if self.__check() == RETURN_NG:
			# 処理中断
			return RETURN_NG

		# トライカウント
		try_count = 0
		# 戻り値初期化
		result_stat = RETURN_OK
		# telnetインスタンス生成
		self.telnet = telnetlib.Telnet()
 
		#接続処理
		print("TelnetTest3 Login " + "-" * 100)
		while True:
			try:
			# 接続
			print("telnet接続開始 接続先 = %s, タイムアウト = %d"
				% (self.hostname, self.connect_timeout))
			self.telnet.open(self.hostname, 23, self.connect_timeout)
			print("telnet接続成功 接続先 = " + self.hostname)
 
			# ログイン
			self.__login("telnetログイン", "telnetユーザ名",
				"login: ", self.username)
 
			# ユーザー
			self.__login("telnetパスワード", "telnetパスワード",
				"Password: ", self.password)
 
			# パスワード
			self.__login("コマンド", "", self.prompt, "")

			# 接続完了
			break

			except (OSError, EOFError, Exception) as exception:
 
				self.telnet.close()
				print(exception.args)
				print(traceback.format_exc())

				# リトライカウント加算
				try_count += 1
				# リトライカウントが3回以上の場合
				if try_count >= self.try_maximum:
					result_stat = RETURN_NG
					break
				time.sleep(self.interval_time)
 
		# 接続不可の場合
		if result_stat == RETURN_NG:
			# 処理中断
			return result_stat
 
		# コマンド処理
		print("TelnetTest3 Start " + "-" * 100)
 
		# 戻り値(行リストのリスト)初期設定
		replys = []
		errlist = []
 
		# プロンプトが空の場合
		if prompt == "":
			prompt = self.prompt
 
		# コマンドリスト分、繰り返す
		for command in command_list:
		try:
			# コマンド実行
			print("telnetコマンド送信 接続先 = %s, コマンド = %s"
				% (self.hostname, command))
			self.__exec_command(command)
 
			print("コマンドプロンプト待ち 接続先 = %s, "\
				"タイムアウト = %d" % (self.hostname, self.reply_timeout))
 
			responce = self.telnet.read_until(
				prompt.encode(self.encode), self.reply_timeout)
			# 改行文字で分割
			line_list = responce.decode(self.encode).splitlines()
			print("コマンドプロンプト受信 接続先 = %s, リプライ = "
				% (self.hostname), end="")
			print(line_list)
 
			# 受信文字に指定したプロンプトがある場合
			if line_list[-1].endswith(prompt):
				# 戻り値(行リスト)初期設定
				reply = []
 
				# コマンドリプライの有効範囲を抽出
				for index in range(len(line_list) - 1):
					reply.append(line_list[index])
 
				# コマンドエラーの場合
				if reply[1].find("コマンドが見つかりません") >= 0:
					errlist.append(command)
				else:
					# 行リストのリストに追加
					replys.append(reply)
			else:
				errlist.append(command)
 
		except (OSError, EOFError, Exception) as exception:
 
			print(exception.args)
			print(traceback.format_exc())
			result_stat = RETURN_NG
 
	# ログアウト
	self.__exec_command("exit")
	self.telnet.close()
 
	# 受信結果出力
	print("■□■ 受信結果(正常) ■□■")
	for reply in replys:
		print("□実行コマンド --> ", end="")
		for x in reply:
			print(x)
	print("■□■ 受信結果(エラー) ■□■")
	for command in errlist:
		print("受信できませんでした コマンド --> " + command)
 
	# 戻り値
	return result_stat
 
if __name__ == '__main__':
 
	# 初期化
	host_name = "XXX.XXX.XXX.XXX"
	username = "YAMADA"
	password = "TAROU"
	prompt = "[tarou@localhost ~]$ "
	# コマンドリスト
	command_list = []
	command_list.append("echo $?")
	command_list.append("ls -al")
	command_list.append('date "+%Y%m%d-%H%M%S"')
 
	# telnet 初期化
	telnet = Telnet_Test(host_name, username, password, prompt)
 
	# telnet 実行
	if telnet.exec(prompt, command_list) == RETURN_OK :
	print("TelnetTest3 OK " + "-" * 100)
 
	else:
		print("TelnetTest3 NG " + "-" * 100)

投稿者: エムシバ君

コメントを残す

メールアドレスが公開されることはありません。 * が付いている欄は必須項目です