#!/usr/bin/env python3.7 # # Copyright (c) 2016-2024 by Ron Frederick and others. # # This program and the accompanying materials are made available under # the terms of the Eclipse Public License v2.0 which accompanies this # distribution and is available at: # # http://www.eclipse.org/legal/epl-2.0/ # # This program may also be made available under the following secondary # licenses when the conditions for such availability set forth in the # Eclipse Public License v2.0 are satisfied: # # GNU General Public License, Version 2.0, or any later versions of # that license # # SPDX-License-Identifier: EPL-2.0 OR GPL-2.0-or-later # # Contributors: # Ron Frederick - initial implementation, API, and documentation # To run this program, the file ``ssh_host_key`` must exist with an SSH # private key in it to use as a server host key. An SSH host certificate # can optionally be provided in the file ``ssh_host_key-cert.pub``. # # The file ``ssh_user_ca`` must exist with a cert-authority entry of # the certificate authority which can sign valid client certificates. import asyncio, asyncssh, sys from typing import List, cast class ChatClient: _clients: List['ChatClient'] = [] def __init__(self, process: asyncssh.SSHServerProcess): self._process = process @classmethod async def handle_client(cls, process: asyncssh.SSHServerProcess): await cls(process).run() async def readline(self) -> str: return cast(str, self._process.stdin.readline()) def write(self, msg: str) -> None: self._process.stdout.write(msg) def broadcast(self, msg: str) -> None: for client in self._clients: if client != self: client.write(msg) def begin_auth(self, username: str) -> bool: # If the user's password is the empty string, no auth is required #return False return True # passwords.get(username) != b'' def password_auth_supported(self) -> bool: return True def validate_password(self, username: str, password: str) -> bool: #if username not in passwords: # return False #pw = passwords[username] #if not password and not pw: # return True return True #return bcrypt.checkpw(password.encode('utf-8'), pw) async def run(self) -> None: self.write('Welcome to chat!\n\n') self.write('Enter your name: ') name = (await self.readline()).rstrip('\n') self.write(f'\n{len(self._clients)} other users are connected.\n\n') self._clients.append(self) self.broadcast(f'*** {name} has entered chat ***\n') try: async for line in self._process.stdin: self.broadcast(f'{name}: {line}') except asyncssh.BreakReceived: pass self.broadcast(f'*** {name} has left chat ***\n') self._clients.remove(self) async def start_server() -> None: await asyncssh.listen('', 2235, server_host_keys=['ssh_host_key'], process_factory=ChatClient.handle_client) loop = asyncio.new_event_loop() try: loop.run_until_complete(start_server()) except (OSError, asyncssh.Error) as exc: sys.exit('Error starting server: ' + str(exc)) loop.run_forever()