### SSH Server with Single Authorized Keys File and Certificates

Source: https://github.com/ronf/asyncssh/blob/develop/docs/index.md

This example uses a single 'authorized_keys' file for all users and leverages certificate principals for authentication. It simplifies server setup by allowing the use of the `listen()` function instead of `create_server()`.

```python3
import asyncio, asyncssh, sys

def handle_client(process: asyncssh.SSHServerProcess) -> None:
    username = process.get_extra_info('username')
    process.stdout.write(f'Welcome to my SSH server, {username}!
')
    process.exit(0)

async def start_server() -> None:
    await asyncssh.listen('', 8022, server_host_keys=['ssh_host_key'],
                          authorized_client_keys='ssh_user_ca',
                          process_factory=handle_client)

loop = asyncio.new_event_loop()

```

--------------------------------

### Install AsyncSSH

Source: https://github.com/ronf/asyncssh/blob/develop/README.rst

Use this command to install the latest stable version of AsyncSSH.

```bash
pip install asyncssh
```

--------------------------------

### Start AsyncSSH Server with SCP Support

Source: https://github.com/ronf/asyncssh/blob/develop/docs/index.md

This snippet demonstrates how to start an AsyncSSH server that supports both SFTP and SCP. Ensure `sftp_factory=True` is set alongside `allow_scp=True` to use the same server instance for both protocols.

```python3
import asyncio, asyncssh, sys

async def start_server() -> None:
    await asyncssh.listen('', 8022, server_host_keys=['ssh_host_key'],
                          authorized_client_keys='ssh_user_ca',
                          sftp_factory=True, allow_scp=True)

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()
```

--------------------------------

### AsyncSSH Server with Line Editing Example

Source: https://github.com/ronf/asyncssh/blob/develop/docs/index.md

This example demonstrates how to set up an AsyncSSH server that utilizes line editing. It shows how to disable echo for sensitive input and switch between line and character input modes.

```python3
import asyncio, asyncssh, sys
from typing import cast

async def handle_client(process: asyncssh.SSHServerProcess):
    channel = cast(asyncssh.SSHLineEditorChannel, process.channel)

    username = process.get_extra_info('username')
    process.stdout.write(f'Welcome to my SSH server, {username}!

')

    channel.set_echo(False)
    process.stdout.write('Tell me a secret: ')
    secret = await process.stdin.readline()

    channel.set_line_mode(False)
    process.stdout.write('
Your secret is safe with me! ')
    await process.stdin.read(1)

    process.stdout.write('
')
    process.exit(0)

async def start_server() -> None:
    await asyncssh.listen('', 8022, server_host_keys=['ssh_host_key'],
                          authorized_client_keys='ssh_user_ca',
                          process_factory=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()
```

--------------------------------

### Start Default SFTP Server

Source: https://github.com/ronf/asyncssh/blob/develop/docs/index.md

Starts an SFTP server with default behavior on port 8022. Ensure 'ssh_host_key' and 'ssh_user_ca' are available.

```python3
import asyncio, asyncssh, sys

async def start_server() -> None:
    await asyncssh.listen('', 8022, server_host_keys=['ssh_host_key'],
                          authorized_client_keys='ssh_user_ca',
                          sftp_factory=True)

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()

```

--------------------------------

### Get Terminal Info and Handle Resizes with AsyncSSH

Source: https://github.com/ronf/asyncssh/blob/develop/docs/index.md

This example retrieves the client's terminal type and window size, and continuously listens for terminal size change events. It requires an asyncio event loop and AsyncSSH library.

```python3
import asyncio, asyncssh, sys

async def handle_client(process: asyncssh.SSHServerProcess) -> None:
    width, height, pixwidth, pixheight = process.term_size

    process.stdout.write(f'Terminal type: {process.term_type}, ' 
                         f'size: {width}x{height}')
    if pixwidth and pixheight:
        process.stdout.write(f' ({pixwidth}x{pixheight} pixels)')
    process.stdout.write('\nTry resizing your window!\n')

    while not process.stdin.at_eof():
        try:
            await process.stdin.read()
        except asyncssh.TerminalSizeChanged as exc:
            process.stdout.write(f'New window size: {exc.width}x{exc.height}')
            if exc.pixwidth and exc.pixheight:
                process.stdout.write(f' ({exc.pixwidth}'
                                     f'x{exc.pixheight} pixels)')
            process.stdout.write('\n')

async def start_server() -> None:
    await asyncssh.listen('', 8022, server_host_keys=['ssh_host_key'],
                          authorized_client_keys='ssh_user_ca',
                          process_factory=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()

```

--------------------------------

### Starting Server Listeners

Source: https://github.com/ronf/asyncssh/blob/develop/docs/api.md

Methods to set up listeners on remote TCP ports or UNIX domain sockets.

```APIDOC
## Starting Server Listeners

### Description
Methods to set up listeners on remote TCP ports or UNIX domain sockets and receive `SSHReader` and `SSHWriter` objects in a callback when new connections are opened.

### Methods
- `start_server(host, port, session_factory)`
- `start_unix_server(path, session_factory)`

### Parameters
- `host` (str): The host to listen on.
- `port` (int): The port to listen on.
- `path` (str): The UNIX domain socket path to listen on.
- `session_factory`: Factory for session objects.

### Response
- `SSHReader`, `SSHWriter`: Objects for I/O on new connections.
```

--------------------------------

### Simple SSH Client Example

Source: https://github.com/ronf/asyncssh/blob/develop/docs/index.md

Connects to localhost, lists files in a directory, and handles potential process errors. Uses the SSHClientConnection as a context manager for automatic closure.

```python3
import asyncio, asyncssh, sys

async def run_client() -> None:
    async with asyncssh.connect('localhost') as conn:
        try:
            result = await conn.run('ls abc', check=True)
        except asyncssh.ProcessError as exc:
            print(exc.stderr, end='')
            print(f'Process exited with status {exc.exit_status}',
                  file=sys.stderr)
        else:
            print(result.stdout, end='')

try:
    asyncio.run(run_client())
except (OSError, asyncssh.Error) as exc:
    sys.exit('SSH connection failed: ' + str(exc))
```

--------------------------------

### Install AsyncSSH with Optional Extras on Windows

Source: https://github.com/ronf/asyncssh/blob/develop/README.rst

Install AsyncSSH along with specific optional dependencies for Windows. This command installs support for bcrypt, fido2, ifaddr, pkcs11, pyOpenSSL, and pywin32.

```bash
pip install 'asyncssh[bcrypt,fido2,ifaddr,pkcs11,pyOpenSSL,pywin32]'
```

--------------------------------

### Start SFTP Server with Custom Path Remapping

Source: https://github.com/ronf/asyncssh/blob/develop/docs/index.md

Starts an SFTP server that remaps user paths to their individual directories under /tmp/sftp/. Requires 'ssh_host_key' and 'ssh_user_ca'.

```python3
import asyncio, asyncssh, os, sys

class MySFTPServer(asyncssh.SFTPServer):
    def __init__(self, chan: asyncssh.SSHServerChannel):
        root = '/tmp/sftp/' + chan.get_extra_info('username')
        os.makedirs(root, exist_ok=True)
        super().__init__(chan, chroot=root)

async def start_server() -> None:
    await asyncssh.listen('', 8022, server_host_keys=['ssh_host_key'],
                          authorized_client_keys='ssh_user_ca',
                          sftp_factory=MySFTPServer)

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()

```

--------------------------------

### Install AsyncSSH with Optional Extras on UNIX

Source: https://github.com/ronf/asyncssh/blob/develop/README.rst

Install AsyncSSH along with specific optional dependencies for UNIX-like systems. This command installs support for bcrypt, fido2, gssapi, ifaddr, pkcs11, and pyOpenSSL.

```bash
pip install 'asyncssh[bcrypt,fido2,gssapi,ifaddr,pkcs11,pyOpenSSL]'
```

--------------------------------

### Install AsyncSSH Development Branch

Source: https://github.com/ronf/asyncssh/blob/develop/README.rst

Install the latest development version of AsyncSSH directly from its GitHub repository.

```bash
pip install git+https://github.com/ronf/asyncssh@develop
```

--------------------------------

### Start SFTP Client and Download File

Source: https://github.com/ronf/asyncssh/blob/develop/docs/index.md

Initiates an SFTP client connection and downloads a specified file. Handles potential OS or AsyncSSH errors.

```python
import asyncio, asyncssh, sys

async def run_client() -> None:
    async with asyncssh.connect('localhost') as conn:
        async with conn.start_sftp_client() as sftp:
            await sftp.get('example.txt')

try:
    asyncio.run(run_client())
except (OSError, asyncssh.Error) as exc:
    sys.exit('SFTP operation failed: ' + str(exc))
```

--------------------------------

### Run Multiple Clients in Parallel

Source: https://github.com/ronf/asyncssh/blob/develop/docs/index.md

This example shows how to execute commands on multiple hosts concurrently using asyncio.gather and process their results. It handles both successful and failed connections/commands. Requires asyncio and asyncssh.

```python3
import asyncio, asyncssh

async def run_client(host, command: str) -> asyncssh.SSHCompletedProcess:
    async with asyncssh.connect(host) as conn:
        return await conn.run(command)

async def run_multiple_clients() -> None:
    # Put your lists of hosts here
    hosts = 5 * ['localhost']

    tasks = (run_client(host, 'ls abc') for host in hosts)
    results = await asyncio.gather(*tasks, return_exceptions=True)

    for i, result in enumerate(results):
        task = f'Task {i+1} to host {hosts[i]}'

        if isinstance(result, Exception):
            print(f'{task} failed: {result}')
        elif result.exit_status != 0:
            print(f'{task} exited with status {result.exit_status}:')
            print(result.stderr, end='')
        else:
            print(f'{task} succeeded:')
            print(result.stdout, end='')

        print(75*'-')

asyncio.run(run_multiple_clients())
```

--------------------------------

### AsyncSSH Server with Callback-Based Input Handling

Source: https://github.com/ronf/asyncssh/blob/develop/docs/index.md

This example implements an SSH server using custom SSHServer and SSHServerSession subclasses, handling client input via callbacks. It processes numbers and calculates a total, similar to the direct process example but using an event-driven approach.

```python
import asyncio, asyncssh, sys

class MySSHServerSession(asyncssh.SSHServerSession):
    def __init__(self):
        self._input = ''
        self._total = 0

    def connection_made(self, chan: asyncssh.SSHServerChannel):
        self._chan = chan

    def shell_requested(self) -> bool:
        return True

    def session_started(self) -> None:
        self._chan.write('Enter numbers one per line, or EOF when done:\n')

    def data_received(self, data: str, datatype: asyncssh.DataType) -> None:
        self._input += data

        lines = self._input.split('\n')
        for line in lines[:-1]:
            try:
                if line:
                    self._total += int(line)
            except ValueError:
                self._chan.write_stderr(f'Invalid number: {line}\n')

        self._input = lines[-1]

    def eof_received(self) -> bool:
        self._chan.write(f'Total = {self._total}\n')
        self._chan.exit(0)
        return False

    def break_received(self, msec: int) -> bool:
        return self.eof_received()

    def soft_eof_received(self) -> None:
        self.eof_received()

class MySSHServer(asyncssh.SSHServer):
    def session_requested(self) -> asyncssh.SSHServerSession:
        return MySSHServerSession()

async def start_server() -> None:
    await asyncssh.create_server(MySSHServer, '', 8022,
                                 server_host_keys=['ssh_host_key'],
                                 authorized_client_keys='ssh_user_ca')

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()
```

--------------------------------

### Forward TCP Connections with Streams API

Source: https://github.com/ronf/asyncssh/blob/develop/docs/index.md

This example demonstrates using the streams API with `start_server` to listen on a port and forward connections. It provides a simple handler that reads data and writes it back to the sender. This approach is an alternative to the channel-based method.

```python
import asyncio, asyncssh, sys

async def handle_connection(reader, writer):
    while not reader.at_eof():
        data = await reader.read(8192)
        writer.write(data)

    writer.close()

def connection_requested(orig_host, orig_port):
    print(f'Connection received from {orig_host}, port {orig_port}')
    return handle_connection

async def run_client():
    async with asyncssh.connect('localhost') as conn:
        server = await conn.start_server(connection_requested, '', 8888,
                                         encoding='utf-8')
        await server.wait_closed()

try:
    asyncio.run(run_client())
except (OSError, asyncssh.Error) as exc:
    sys.exit('SSH connection failed: ' + str(exc))

```

--------------------------------

### AsyncSSH Reverse Direction Client Example

Source: https://github.com/ronf/asyncssh/blob/develop/docs/index.md

This client initiates a reverse SSH connection and runs arbitrary shell commands received from the server. Ensure `client_host_key` and `trusted_server_keys` are configured.

```python
import asyncio, asyncssh, sys
from asyncio.subprocess import PIPE

async def handle_request(process: asyncssh.SSHServerProcess) -> None:
    """Run a command on the client, piping I/O over an SSH session"""

    assert process.command is not None

    local_proc = await asyncio.create_subprocess_shell(
        process.command, stdin=PIPE, stdout=PIPE, stderr=PIPE)

    await process.redirect(stdin=local_proc.stdin, stdout=local_proc.stdout,
                           stderr=local_proc.stderr)

    process.exit(await local_proc.wait())
    await process.wait_closed()

async def run_reverse_client() -> None:
    """Make an outbound connection and then become an SSH server on it"""

    conn = await asyncssh.connect_reverse(
        'localhost', 8022, server_host_keys=['client_host_key'],
        authorized_client_keys='trusted_server_keys',
        process_factory=handle_request, encoding=None)

    await conn.wait_closed()

try:
    asyncio.run(run_reverse_client())
except (OSError, asyncssh.Error) as exc:
    sys.exit('Reverse SSH connection failed: ' + str(exc))
```

--------------------------------

### Echo Server using SSHTCPSession

Source: https://github.com/ronf/asyncssh/blob/develop/docs/index.md

This example shows a server that echoes data received on port 7 using the SSHTCPSession class. It raises an error for any other port.

```python3
import asyncio, asyncssh, sys

class MySSHTCPSession(asyncssh.SSHTCPSession):
    def connection_made(self, chan: asyncssh.SSHTCPChannel) -> None:
        self._chan = chan

    def data_received(self, data: bytes, datatype: asyncssh.DataType) -> None:
        self._chan.write(data)

class MySSHServer(asyncssh.SSHServer):
    def connection_requested(self, dest_host: str, dest_port: int,
                             orig_host: str, orig_port: int) -> \
            asyncssh.SSHTCPSession:
        if dest_port == 7:
            return MySSHTCPSession()
        else:
            raise asyncssh.ChannelOpenError(
                asyncssh.OPEN_ADMINISTRATIVELY_PROHIBITED,
                'Only echo connections allowed')

async def start_server() -> None:
    await asyncssh.create_server(MySSHServer, '', 8022,
                                 server_host_keys=['ssh_host_key'],
                                 authorized_client_keys='ssh_user_ca')

loop = asyncio.new_event_loop()

try:
    loop.run_until_complete(start_server())
except (OSError, asyncssh.Error) as exc:
    sys.exit('SSH server failed: ' + str(exc))

loop.run_forever()

```

--------------------------------

### AsyncSSH Reverse Direction Server Example

Source: https://github.com/ronf/asyncssh/blob/develop/docs/index.md

This server accepts inbound reverse SSH connections and then acts as an SSH client, running commands on the connected client. Ensure `server_key` and `trusted_client_host_keys` are configured.

```python
import asyncio, asyncssh, sys

async def run_commands(conn: asyncssh.SSHClientConnection) -> None:
    """Run a series of commands on the client which connected to us"""

    commands = ('ls', 'sleep 30 && date', 'sleep 5 && cat /proc/cpuinfo')

    async with conn:
        tasks = [conn.run(cmd) for cmd in commands]

        for task in asyncio.as_completed(tasks):
            result = await task
            print('Command:', result.command)
            print('Return code:', result.returncode)
            print('Stdout:')
            print(result.stdout, end='')
            print('Stderr:')
            print(result.stderr, end='')
            print(75*'-')

async def start_reverse_server() -> None:
    """Accept inbound connections and then become an SSH client on them"""

    await asyncssh.listen_reverse(port=8022, client_keys=['server_key'],
                                  known_hosts='trusted_client_host_keys',
                                  acceptor=run_commands)

loop = asyncio.new_event_loop()

try:
    loop.run_until_complete(start_reverse_server())
except (OSError, asyncssh.Error) as exc:
    sys.exit('Error starting server: ' + str(exc))

loop.run_forever()
```

--------------------------------

### Simplified SSH Client with Callbacks

Source: https://github.com/ronf/asyncssh/blob/develop/docs/index.md

Use this simplified version when you only need to customize the SSH session callbacks and not the client itself. It connects and runs a command.

```python3
import asyncio, asyncssh, sys
from typing import Optional

class MySSHClientSession(asyncssh.SSHClientSession):
    def data_received(self, data: str, datatype: asyncssh.DataType) -> None:
        print(data, end='')

    def connection_lost(self, exc: Optional[Exception]) -> None:
        if exc:
            print('SSH session error: ' + str(exc), file=sys.stderr)

async def run_client() -> None:
    async with asyncssh.connect('localhost') as conn:
        chan, session = await conn.create_session(MySSHClientSession, 'ls abc')
        await chan.wait_closed()

try:
    asyncio.run(run_client())
except (OSError, asyncssh.Error) as exc:
    sys.exit('SSH connection failed: ' + str(exc))
```

--------------------------------

### Run SSH Command with AsyncSSH Client

Source: https://github.com/ronf/asyncssh/blob/develop/README.rst

This snippet demonstrates how to connect to 'localhost' using AsyncSSH, run an 'echo' command, and print the output. It includes basic error handling for connection failures.

```python
import asyncio, asyncssh, sys

async def run_client():
    async with asyncssh.connect('localhost') as conn:
        result = await conn.run('echo "Hello!"', check=True)
        print(result.stdout, end='')

try:
    asyncio.get_event_loop().run_until_complete(run_client())
except (OSError, asyncssh.Error) as exc:
    sys.exit('SSH connection failed: ' + str(exc))
```

--------------------------------

### Send Interactive Input to Remote Process with AsyncSSH

Source: https://github.com/ronf/asyncssh/blob/develop/docs/index.md

Use `create_process` to start a remote process and interact with its stdin and stdout. This is useful for programs that require user input, like 'bc'.

```python3
import asyncio, asyncssh, sys

async def run_client() -> None:
    async with asyncssh.connect('localhost') as conn:
        async with conn.create_process('bc') as process:
            for op in ['2+2', '1*2*3*4', '2^32']:
                process.stdin.write(op + '\n')
                result = await process.stdout.readline()
                print(op, '=', result, end='')

try:
    asyncio.run(run_client())
except (OSError, asyncssh.Error) as exc:
    sys.exit('SSH connection failed: ' + str(exc))
```

--------------------------------

### Simple AsyncSSH Server with Input Handling

Source: https://github.com/ronf/asyncssh/blob/develop/docs/index.md

This server reads numbers from client input, calculates their sum, and displays the total upon receiving EOF. It uses async/await syntax for handling client processes.

```python
import asyncio, asyncssh, sys

async def handle_client(process: asyncssh.SSHServerProcess) -> None:
    process.stdout.write('Enter numbers one per line, or EOF when done:\n')

    total = 0

    try:
        async for line in process.stdin:
            line = line.rstrip('\n')
            if line:
                try:
                    total += int(line)
                except ValueError:
                    process.stderr.write(f'Invalid number: {line}\n')
    except asyncssh.BreakReceived:
        pass

    process.stdout.write(f'Total = {total}\n')
    process.exit(0)

async def start_server() -> None:
    await asyncssh.listen('', 8022, server_host_keys=['ssh_host_key'],
                          authorized_client_keys='ssh_user_ca',
                          process_factory=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()
```

--------------------------------

### SSH Client Distinguishing Stdout/Stderr

Source: https://github.com/ronf/asyncssh/blob/develop/docs/index.md

This example shows how to differentiate between standard output and standard error streams within the SSH session callbacks. It prints stderr to the console's error stream.

```python3
import asyncio, asyncssh, sys
from typing import Optional

class MySSHClientSession(asyncssh.SSHClientSession):
    def data_received(self, data: str, datatype: asyncssh.DataType) -> None:
        if datatype == asyncssh.EXTENDED_DATA_STDERR:
            print(data, end='', file=sys.stderr)
        else:
            print(data, end='')

    def connection_lost(self, exc: Optional[Exception]) -> None:
        if exc:
            print('SSH session error: ' + str(exc), file=sys.stderr)

async def run_client() -> None:
    async with asyncssh.connect('localhost') as conn:
        chan, session = await conn.create_session(MySSHClientSession, 'ls abc')
        await chan.wait_closed()

try:
    asyncio.run(run_client())
except (OSError, asyncssh.Error) as exc:
    sys.exit('SSH connection failed: ' + str(exc))
```

--------------------------------

### Echo Server using Streams API

Source: https://github.com/ronf/asyncssh/blob/develop/docs/index.md

This example uses the streams API to handle direct TCP connections, echoing data back to the client. It returns a handler coroutine for new connections.

```python3
import asyncio, asyncssh, sys

async def handle_connection(reader: asyncssh.SSHReader,
                            writer: asyncssh.SSHWriter) -> None:
    while not reader.at_eof():
        data = await reader.read(8192)

        try:
            writer.write(data)
        except BrokenPipeError:
            break

    writer.close()

class MySSHServer(asyncssh.SSHServer):
    def connection_requested(self, dest_host: str, dest_port: int,
                             orig_host: str, orig_port: int) -> \
            asyncssh.SSHSocketSessionFactory:
        if dest_port == 7:
            return handle_connection
        else:
            raise asyncssh.ChannelOpenError(
                      asyncssh.OPEN_ADMINISTRATIVELY_PROHIBITED,
                      'Only echo connections allowed')

async def start_server() -> None:
    await asyncssh.create_server(MySSHServer, '', 8022,
                                 server_host_keys=['ssh_host_key'],
                                 authorized_client_keys='ssh_user_ca')

loop = asyncio.new_event_loop()

try:
    loop.run_until_complete(start_server())
except (OSError, asyncssh.Error) as exc:
    sys.exit('SSH server failed: ' + str(exc))

loop.run_forever()

```

--------------------------------

### Connecting with Custom Known Hosts

Source: https://github.com/ronf/asyncssh/blob/develop/docs/index.md

Establishes an SSH connection to localhost, specifying a custom file for server host key verification.

```python3
async with asyncssh.connect('localhost', known_hosts='my_known_hosts') as conn:
```

--------------------------------

### AsyncSSH Multi-Client Chat Service

Source: https://github.com/ronf/asyncssh/blob/develop/docs/index.md

Implement a basic chat service where multiple clients can connect simultaneously and exchange messages. This example uses a class to manage individual clients and broadcast messages.

```python
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)

    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('', 8022, server_host_keys=['ssh_host_key'],
                          authorized_client_keys='ssh_user_ca',
                          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()
```

--------------------------------

### Redirect Output to Local File with AsyncSSH

Source: https://github.com/ronf/asyncssh/blob/develop/docs/index.md

The `run` method can redirect the remote process's stdout to a local file using the `stdout` argument. This example redirects the output of 'tail -r' to '/tmp/stdout'.

```python3
import asyncio, asyncssh, sys

async def run_client() -> None:
    async with asyncssh.connect('localhost') as conn:
        await conn.run('tail -r', input='1\n2\n3\n', stdout='/tmp/stdout')

try:
    asyncio.run(run_client())
except (OSError, asyncssh.Error) as exc:
    sys.exit('SSH connection failed: ' + str(exc))
```

--------------------------------

### Connecting with Client Keys

Source: https://github.com/ronf/asyncssh/blob/develop/docs/index.md

Connects to localhost using specified client SSH keys for authentication.

```python3
async with asyncssh.connect('localhost', client_keys=['my_ssh_key']) as conn:
```

--------------------------------

### Simple Password Authenticated SSH Server

Source: https://github.com/ronf/asyncssh/blob/develop/docs/index.md

This server listens on port 8022 and supports password authentication using bcrypt hashing. It prints a welcome message upon successful authentication and shell session start.

```python3
# 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``.

import asyncio, asyncssh, bcrypt, sys
from typing import Optional

passwords = {'guest': b'',                # guest account with no password
             'user123': bcrypt.hashpw(b'secretpw', bcrypt.gensalt()),
            }

def handle_client(process: asyncssh.SSHServerProcess) -> None:
    username = process.get_extra_info('username')
    process.stdout.write(f'Welcome to my SSH server, {username}!
')
    process.exit(0)

class MySSHServer(asyncssh.SSHServer):
    def connection_made(self, conn: asyncssh.SSHServerConnection) -> None:
        peername = conn.get_extra_info('peername')[0]
        print(f'SSH connection received from {peername}.')

    def connection_lost(self, exc: Optional[Exception]) -> None:
        if exc:
            print('SSH connection error: ' + str(exc), file=sys.stderr)
        else:
            print('SSH connection closed.')

    def begin_auth(self, username: str) -> bool:
        # If the user's password is the empty string, no auth is required
        return 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 bcrypt.checkpw(password.encode('utf-8'), pw)

async def start_server() -> None:
    await asyncssh.create_server(MySSHServer, '', 8022,
                                 server_host_keys=['ssh_host_key'],
                                 process_factory=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()

```

--------------------------------

### Custom SSH Client with Callbacks

Source: https://github.com/ronf/asyncssh/blob/develop/docs/index.md

Use this when you need to customize both the SSH client and session behavior with callbacks. It demonstrates handling connection events and receiving data.

```python3
import asyncio, asyncssh, sys
from typing import Optional

class MySSHClientSession(asyncssh.SSHClientSession):
    def data_received(self, data: str, datatype: asyncssh.DataType) -> None:
        print(data, end='')

    def connection_lost(self, exc: Optional[Exception]) -> None:
        if exc:
            print('SSH session error: ' + str(exc), file=sys.stderr)

class MySSHClient(asyncssh.SSHClient):
    def connection_made(self, conn: asyncssh.SSHClientConnection) -> None:
        print(f'Connection made to {conn.get_extra_info('peername')[0]}.')

    def auth_completed(self) -> None:
        print('Authentication successful.')

async def run_client() -> None:
    conn, client = await asyncssh.create_connection(MySSHClient, 'localhost')

    async with conn:
        chan, session = await conn.create_session(MySSHClientSession, 'ls abc')
        await chan.wait_closed()

try:
    asyncio.run(run_client())
except (OSError, asyncssh.Error) as exc:
    sys.exit('SSH connection failed: ' + str(exc))
```

--------------------------------

### Pipe Local Process Output to Remote Process with AsyncSSH

Source: https://github.com/ronf/asyncssh/blob/develop/docs/index.md

This example pipes the stdout of a local 'echo' command to a remote 'tail -r' process. It uses `subprocess.Popen` for the local process and `conn.run` with `stdin` argument for the remote process.

```python3
import asyncio, asyncssh, subprocess, sys

async def run_client() -> None:
    async with asyncssh.connect('localhost') as conn:
        local_proc = subprocess.Popen(r'echo "1\n2\n3"', shell=True,
                                      stdout=subprocess.PIPE)
        remote_result = await conn.run('tail -r', stdin=local_proc.stdout)
        print(remote_result.stdout, end='')

try:
    asyncio.run(run_client())
except (OSError, asyncssh.Error) as exc:
    sys.exit('SSH connection failed: ' + str(exc))
```