summaryrefslogtreecommitdiffstats
path: root/examples/asyncio-ssh-python-embed.py
blob: 9bbad86f131a1c71c9cf50e04d0871fc0162401d (plain)
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
#!/usr/bin/env python
"""
Example of running the Python REPL through an SSH connection in an asyncio process.
This requires Python 3, asyncio and asyncssh.

Run this example and then SSH to localhost, port 8222.
"""

import asyncio
import logging

import asyncssh

from ptpython.contrib.asyncssh_repl import ReplSSHServerSession

logging.basicConfig()
logging.getLogger().setLevel(logging.INFO)


class MySSHServer(asyncssh.SSHServer):
    """
    Server without authentication, running `ReplSSHServerSession`.
    """

    def __init__(self, get_namespace):
        self.get_namespace = get_namespace

    def begin_auth(self, username):
        # No authentication.
        return False

    def session_requested(self):
        return ReplSSHServerSession(self.get_namespace)


async def main(port: int = 8222) -> None:
    """
    Example that starts the REPL through an SSH server.
    """
    # Namespace exposed in the REPL.
    environ = {"hello": "world"}

    # Start SSH server.
    def create_server() -> MySSHServer:
        return MySSHServer(lambda: environ)

    print("Listening on :%i" % port)
    print('To connect, do "ssh localhost -p %i"' % port)

    await asyncssh.create_server(
        create_server, "", port, server_host_keys=["/etc/ssh/ssh_host_dsa_key"]
    )
    await asyncio.Future()  # Wait forever.


if __name__ == "__main__":
    asyncio.run(main())