Telnet 類圖(文本描述)
+---------------------------------------+
| Telnet |
+---------------------------------------+
| - host: str | # 目標主機
| - port: int | # 目標端口(默認23)
| - timeout: float | # 超時時間
| - sock: socket.socket | # 底層Socket連接
| - rawq: bytes | # 原始接收緩沖區
| - irawq: int | # 緩沖區索引
| - cookedq: bytes | # 處理后的數據緩沖區
| - option_callback: Callable | # 選項協商回調函數
+---------------------------------------+
| + __init__(host=None, port=0, timeout)| # 初始化對象(可選連接)
| + open(host, port, timeout) | # 建立Telnet連接
| + close() | # 關閉連接
| + read_until(match, timeout=None) | # 讀取直到匹配指定字節串
| + read_all() | # 讀取所有數據直到EOF
| + write(buffer) | # 發送字節流數據
| + interact() | # 進入交互模式
| - process_rawq() | # 處理原始數據(解析IAC命令)
| - _process_do(option) | # 處理服務端DO請求
| - _process_dont(option) | # 處理服務端DONT請求
| - set_option_negotiation_callback(cb) | # 設置選項協商回調
+---------------------------------------+
核心關系說明
-
依賴關系:
socket
模塊:Telnet
類通過sock
屬性依賴底層 Socket 連接。select
模塊:interact()
方法使用select
監聽輸入輸出。
-
數據流:
- 接收數據:
sock.recv()
→rawq
→process_rawq()
→cookedq
→read_until()
/read_all()
。 - 發送數據:
write()
→sock.sendall()
。
- 接收數據:
-
協議處理:
- IAC 命令解析:
process_rawq()
處理rawq
中的 IAC 序列(如0xFF
),調用_process_do
等方法響應選項協商。 - 回調擴展:通過
option_callback
支持自定義選項處理邏輯。
- IAC 命令解析:
關鍵方法詳解
1. read_until(match, timeout)
- 輸入:目標字節串
match
(如b"#"
),超時時間timeout
。 - 輸出:從
cookedq
中提取的字節流,直到匹配match
。 - 流程:
- 循環填充
rawq
(調用fill_rawq()
)。 - 處理
rawq
中的協議命令(調用process_rawq()
)。 - 檢查
cookedq
是否包含match
,若超時則拋出socket.timeout
。
- 循環填充
2. write(buffer)
- 輸入:字節流
buffer
(若為字符串需用戶自行編碼)。 - 流程:直接調用
sock.sendall(buffer)
發送數據。 - 注意:默認不處理編碼,需用戶確保字節流符合設備要求(如
buffer.encode("gbk")
)。
3. interact()
- 功能:實現用戶與遠程設備的實時交互。
- 流程:
- 監聽
stdin
和sock
的可讀事件(使用select.select()
)。 - 將用戶輸入發送到設備,設備響應輸出到終端。
- 按
Ctrl+]
退出交互模式。
- 監聽
示例調用流程
tn = Telnet(host="192.168.1.1") # 初始化并連接
tn.read_until(b"Username:") # 讀取直到用戶名提示
tn.write(b"admin\n") # 發送用戶名
tn.read_until(b"Password:") # 讀取直到密碼提示
tn.write(b"password123\n") # 發送密碼
tn.interact() # 進入交互模式
tn.close() # 關閉連接
擴展設計
-
自定義選項處理:
def custom_callback(tn, command, option):if command == DO and option == ECHO:tn.sock.sendall(IAC + WONT + ECHO) # 拒絕回顯選項tn = Telnet() tn.set_option_negotiation_callback(custom_callback)
-
替代方案:對安全性要求高的場景,建議使用
paramiko
(SSH 協議庫)。
通過此圖可快速掌握 telnetlib
的核心結構和數據流,便于調試或二次開發。