Das deutsche Python-Forum

Code: Alles auswählen

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

    def check_channel_request(self, kind, chanid):
        return OPEN_SUCCEEDED

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

    def check_channel_exec_request ( self, channel, command ):
        print ( f'....> check_channel_exec_request commmand: {command}' )
        return True 

class ConnHandlerThd(threading.Thread):
    def __init__(self, conn, addr, serverkeyfile ):
        threading.Thread.__init__(self)
        self._conn = conn
        self._addr = addr
        self._serverkeyfile = serverkeyfile

    def run(self):
        server_key = paramiko.RSAKey.from_private_key_file(self._serverkeyfile)
        name = server_key.get_name()

        transport = paramiko.Transport(self._conn)
        transport.add_server_key(server_key)

        transport.set_subsystem_handler( 'sftp', paramiko.SFTPServer, StubSFTPServer)

        server = StubServer()
        transport.start_server(server=server)

        channel = transport.accept()
        while transport.is_active():
            time.sleep(1)

def start_server(host, port, serverkeyfile, logfile=logfile, level=level, backlog=backlog ):
    paramiko_level = getattr(paramiko.common, level)
    paramiko.common.logging.basicConfig(level=paramiko_level)

    paramiko.util.log_to_file ( logfile, level = level )

    server_socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
    server_socket.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, True)
    server_socket.bind((host, port))
    server_socket.listen(backlog)

    while True:
        conn, addr = server_socket.accept()
        srv_thd = ConnHandlerThd(conn, addr, serverkeyfile )
        srv_thd.setDaemon(True)
        srv_thd.start()

start_server(host=host, port=port, serverkeyfile=keyfile, level=level, logfile=logfile, backlog=backlog)

Code: Alles auswählen

#!/usr/bin/env python3
import socket
import threading
import time

import paramiko

HOST = ...
PORT = ...
KEY_FILENAME = ...
LOG_FILENAME = ...
LEVEL = ...
BACKLOG = ...


class StubSFTPServer:
    ...


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

    def check_channel_request(self, kind, chanid):
        return paramiko.OPEN_SUCCEEDED

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

    def check_channel_exec_request(self, channel, command):
        print(f"....> check_channel_exec_request commmand: {command}")
        return True


def handle_connection(connection, server_key_filename):
    server_key = paramiko.RSAKey.from_private_key_file(server_key_filename)

    transport = paramiko.Transport(connection)
    transport.add_server_key(server_key)
    transport.set_subsystem_handler(
        "sftp", paramiko.SFTPServer, StubSFTPServer
    )
    transport.start_server(server=StubServer())
    transport.accept()

    while transport.is_active():
        time.sleep(1)


def start_server(host, port, server_key_filename, log_filename, level, backlog):
    paramiko_level = getattr(paramiko.common, level)
    paramiko.common.logging.basicConfig(level=paramiko_level)
    paramiko.util.log_to_file(log_filename, level=level)

    server_socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
    server_socket.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, True)
    server_socket.bind((host, port))
    server_socket.listen(backlog)

    while True:
        connection, _address = server_socket.accept()
        threading.Thread(
            target=handle_connection,
            args=[connection, server_key_filename],
            daemon=True,
        ).start()


def main():
    start_server(HOST, PORT, KEY_FILENAME, LOG_FILENAME, LEVEL, BACKLOG)


if __name__ == "__main__":
    main()