| |
| import asyncio |
| import asyncssh |
| import os |
| import sys |
|
|
| |
| class MySSHServerSession(asyncssh.SSHServerSession): |
| def connection_made(self, chan): |
| self.chan = chan |
|
|
| def session_started(self): |
| self.chan.write("Добро пожаловать в SSH-сервис на Hugging Face Spaces!\n") |
| self.chan.write("Введите команду (для выхода наберите 'exit'):\n") |
|
|
| def data_received(self, data, datatype): |
| |
| if data.strip() == 'exit': |
| self.chan.write("До свидания!\n") |
| self.chan.exit(0) |
| else: |
| |
| self.chan.write("Вы ввели: " + data) |
|
|
| def connection_lost(self, exc): |
| pass |
|
|
| |
| class MySSHServer(asyncssh.SSHServer): |
| def session_requested(self): |
| return MySSHServerSession() |
|
|
| |
| async def start_ssh_server(): |
| |
| port = int(os.environ.get('PORT', 2222)) |
|
|
| |
| |
| host_key_file = os.environ.get('HOST_KEY_FILE') |
| if host_key_file and os.path.exists(host_key_file): |
| host_keys = [host_key_file] |
| else: |
| print("Генерация временного ключа сервера...", file=sys.stderr) |
| host_keys = [asyncssh.generate_private_key('ssh-rsa')] |
|
|
| try: |
| server = await asyncssh.create_server( |
| MySSHServer, '', port, |
| server_host_keys=host_keys, |
| |
| password_authenticator=lambda username, password: True, |
| allow_pty=True |
| ) |
| except (OSError, asyncssh.Error) as exc: |
| sys.exit('Ошибка при запуске сервера: ' + str(exc)) |
| |
| print('SSH-сервер запущен на порту', port) |
| await server.wait_closed() |
|
|
| def main(): |
| try: |
| asyncio.run(start_ssh_server()) |
| except (OSError, asyncssh.Error) as exc: |
| sys.exit('Ошибка: ' + str(exc)) |
|
|
| if __name__ == '__main__': |
| main() |
|
|