Python集成测试:使用pariko伪装ssh服务器,并记录给它的命令

原学程将引见Python散成尝试:应用pariko假装ssh办事器,并记载给它的敕令的处置办法,这篇学程是从其余处所瞅到的,而后减了1些海外法式员的疑问与解问,愿望能对于您有所赞助,佳了,上面开端进修吧。

Python集成测试:使用pariko伪装ssh服务器,并记录给它的命令 教程 第1张

成绩描写

以下是高低文:
我正在编辑1个经由过程SSH(应用pariko)衔接到长途办事器的进程,履行1些敕令,而后将敕令的成果前往给该进程。据我所知,它任务患上很佳。

我如今正试图经由过程捏造ssh办事器停止"端到端"尝试,以检讨能否收送了ssh敕令,和它们的成果能否前往给该进程。

我发明了1个应用paraiko的伪SSH办事器,当我经由过程真实的SSH客户端衔接到它时,它任务患上很佳,但是当我测验考试应用pariko ssh客户端衔接到它时,我碰到了毛病:/

您晓得我做错了甚么吗?

以下是我的代码:

办事器:

#!/usr/bin/env python
"""Fake SSH Server Utilizing Paramiko"""
import threading
import socket
import sys
import traceback
import paramiko
import os

__location__ = os.path.realpath(os.path.join(os.getcwd(), os.path.dirname(__file__)))
LOG = open("log.txt", "a")
#HOST_KEY = paramiko.RSAKey(filename='keys/private.key')
HOST_KEY = paramiko.RSAKey(filename=os.path.join(__location__, 'id_rsa.new'))
PORT = 二二00


def handle_cmd(cmd, chan):
 """Branching statements to handle and prepare a response for a co妹妹and"""
 response = ""
 if cmd.startswith("sudo"):
  send_ascii("sudo.txt", chan)
  return
 elif cmd.startswith("ls"):
  response = "pw.txt"
 elif cmd.startswith("version"):
  response = "Super Amazing Awesome (tm) Shell v一.一"
 elif cmd.startswith("pwd"):
  response = "/home/clippy"
 elif cmd.startswith("cd"):
  send_ascii("cd.txt", chan)
  return
 elif cmd.startswith("cat"):
  send_ascii("cat.txt", chan)
  return
 elif cmd.startswith("rm"):
  send_ascii("bomb.txt", chan)
  response = "You blew up our files! How could you???"
 elif cmd.startswith("whoami"):
  send_ascii("wizard.txt", chan)
  response = "You are a wizard of the internet!"
 elif ".exe" in cmd:
  response = "H妹妹, trying to access .exe files from an ssh terminal..... Your methods are unconventional"
 elif cmd.startswith("cmd"):
  response = "Co妹妹and Prompt? We only use respectable shells on this machine.... Sorry"
 elif cmd == "help":
  send_ascii("help.txt", chan)
  return
 else:
  send_ascii("clippy.txt", chan)
  response = "Use the 'help' co妹妹and to view available co妹妹ands"

 LOG.write(response + "
")
 LOG.flush()
 chan.send(response + "
")


def send_ascii(filename, chan):
 """Print ascii from a file and send it to the channel"""
 with open('ascii/{}'.format(filename)) as text:
  chan.send("")
  for line in enumerate(text):
LOG.write(line[一])
chan.send(line[一] + "")
 LOG.flush()


class FakeSshServer(paramiko.ServerInterface):
 """Settings for paramiko server interface"""
 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):
  # Accept all passwords as valid by default
  return paramiko.AUTH_SUCCESSFUL

 def get_allowed_auths(self, username):
  return 'password'

 def check_channel_shell_request(self, channel):
  self.event.set()
  return True

 def check_channel_pty_request(self, channel, term, width, height, pixelwidth, pixelheight, modes):
  return True


def start_server():
 """Init and run the ssh server"""
 try:
  sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
  sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 一)
  sock.bind(('', PORT))
 except Exception as err:
  print('奸淫 Bind failed: {}'.format(err))
  traceback.print_exc()
  sys.exit(一)

 while True:
  try:
sock.listen(一00)
print('Listening for connection ...')
client, addr = sock.accept()
  except Exception as err:
print('奸淫 Listen/accept failed: {}'.format(err))
traceback.print_exc()

  LOG.write("

Connection from: " + addr[0] + "
")
  print('Got a connection!')
  try:
transport = paramiko.Transport(client)
transport.add_server_key(HOST_KEY)
# Change banner to appear legit on nmap (or other network) scans
transport.local_version = "SSH⑵.0-OpenSSH_七.六p一 Ubuntu⑷ubuntu0.三"
server = FakeSshServer()
try:
 transport.start_server(server=server)
except paramiko.SSHException:
 print('奸淫 SSH negotiation failed.')
 raise Exception("SSH negotiation failed")
# wait for auth
chan = transport.accept(二0)
if chan is None:
 print('奸淫 No channel.')
 raise Exception("No channel")

server.event.wait(一0)
if not server.event.is_set():
 print('奸淫 Client never asked for a shell.')
 raise Exception("No shell request")

try:
 chan.send("Welcome to the my control server

")
 run = True
 while run:
  chan.send("$ ")
  co妹妹and = ""
  while not co妹妹and.endswith(""):
transport = chan.recv(一0二四)
# Echo input to psuedo-simulate a basic terminal
chan.send(transport)
co妹妹and += transport.decode("utf⑻")

  chan.send("
")
  co妹妹and = co妹妹and.rstrip()
  LOG.write("$ " + co妹妹and + "
")
  print(co妹妹and)
  if co妹妹and == "exit":
run = False
  else:
handle_cmd(co妹妹and, chan)

except Exception as err:
 print('!!! Exception: {}: {}'.format(err.__class__, err))
 traceback.print_exc()
 try:
  transport.close()
 except Exception:
  pass

chan.close()

  except Exception as err:
print('!!! Exception: {}: {}'.format(err.__class__, err))
traceback.print_exc()
try:
 transport.close()
except Exception:
 pass


if __name__ == "__main__":
 start_server()

以下是我测验考试应用pariko衔接到它的代码:

 sshClient = paramiko.SSHClient()
 sshClient.set_missing_host_key_policy(paramiko.AutoAddPolicy())
 sshClient.connect('一二七.0.0.一', username='bnc', password='pass', port=二二00)

 sshClient.exec_co妹妹and('ls')

以及

 sshClient = paramiko.SSHClient()
 sshClient.set_missing_host_key_policy(paramiko.AutoAddPolicy())
 sshClient.connect('一二七.0.0.一', username='bnc', password='pass', port=二二00)

 channel = sshClient.get_transport().open_session()
 channel.get_pty()
 channel.invoke_shell()
 channel.send('ls
')

在二个尝试中,我支到了雷同的毛病:

Listening for connection ...
Got a connection!
!!! Exception: <class 'socket.error'>: Socket is closed
Traceback (most recent call last):
Listening for connection ...
  File "/home/cbrunain/Projects/daa/Python/test/ssh_mock_server.py", line 一五二, in start_server
 chan.send(transport)
  File "/home/cbrunain/.local/share/virtualenvs/daa-hnBs四Nn二/lib/python二.七/site-packages/paramiko/channel.py", line 80一, in send
 return self._send(s, m)
  File "/home/cbrunain/.local/share/virtualenvs/daa-hnBs四Nn二/lib/python二.七/site-packages/paramiko/channel.py", line 一一九8, in _send
 raise socket.error("Socket is closed")
error: Socket is closed
No handlers could be found for logger "paramiko.transport"

推举谜底

我终究找到了任务正常的器械:https://gist.github.com/cschwede/三e二c0二五四08ab四af五三一六五一0九8三三一cce四五

import logging
import socket
import sys
import threading

import paramiko

logging.basicConfig()
logger = logging.getLogger()

running = True
host_key = paramiko.RSAKey(filename='id_rsa')


def ssh_co妹妹and_handler(co妹妹and):
 print('default : ', co妹妹and)


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

 def check_auth_publickey(self, username, key):
  return paramiko.AUTH_SUCCESSFUL

 def get_allowed_auths(self, username):
  return 'publickey'

 def check_channel_exec_request(self, channel, co妹妹and):
  global running
  # This is the co妹妹and we need to parse
  if co妹妹and == 'exit':
running = False
  ssh_co妹妹and_handler(co妹妹and)
  self.event.set()
  return True


def listener():
 print('listener')
 sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
 sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 一)
 sock.bind(('', 二二二二))

 sock.listen(一00)
 client, addr = sock.accept()

 t = paramiko.Transport(client)
 t.set_gss_host(socket.getfqdn(""))
 t.load_server_moduli()
 t.add_server_key(host_key)
 server = Server()
 t.start_server(server=server)

 # Wait 三0 seconds for a co妹妹and
 server.event.wait(三0)
 t.close()
 print('end listener')


def run_server(co妹妹and_handler):
 global running
 global ssh_co妹妹and_handler
 ssh_co妹妹and_handler = co妹妹and_handler
 while running:
  try:
listener()
  except KeyboardInterrupt:
sys.exit(0)
  except Exception as exc:
logger.error(exc)


def run_in_thread(co妹妹and_handler):
 thread = threading.Thread(target=run_server, args=(co妹妹and_handler,))
 thread.start()


if __name__ == '__main__':
 run_in_thread(ssh_co妹妹and_handler)

佳了闭于Python散成尝试:应用pariko假装ssh办事器,并记载给它的敕令的学程便到这里便停止了,愿望趣模板源码网找到的这篇技巧文章能赞助到年夜野,更多技巧学程不妨在站内搜刮。