File size: 2,004 Bytes
debd19e | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 | import socket
import paramiko
# Define host and port
host_key = paramiko.RSAKey(filename='test_rsa.key')
host, port = 'localhost', 2222
# Create SSH server class
class SSHServer(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 == 'test' and password == 'password':
return paramiko.AUTH_SUCCESSFUL
return paramiko.AUTH_FAILED
# Main function to start the server
def start_server():
server = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
server.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
server.bind((host, port))
server.listen(100)
print(f'[*] Listening for connection on {host}:{port}...')
while True:
client, addr = server.accept()
print(f'[*] Accepted connection from {addr[0]}:{addr[1]}')
try:
transport = paramiko.Transport(client)
transport.add_server_key(host_key)
server = SSHServer()
transport.start_server(server=server)
chan = transport.accept(20)
if chan is None:
print('*** No channel.')
continue
print(f'[*] Authenticated!')
chan.send('Welcome to my SSH server!')
while True:
command = chan.recv(1024)
if not command:
break
print(f'[*] Received command: {command.decode()}')
chan.send(command)
transport.close()
except Exception as e:
print(f'[-] Caught exception: {e.__class__.__name__}: {e}')
try:
transport.close()
except:
pass
if __name__ == '__main__':
start_server()
|