pyexpose.base package

Submodules

pyexpose.base.connector module

class pyexpose.base.connector.ExposeConnector[source]

Bases: ABC, Generic[ExposeSessionVar]

A factory class that represents a connection to a tunneling servers.

connect() → AsyncIterator[ExposeSessionVar][source]

Connects to the server, using the default settings.

Returns:

ExposeSession that wraps around asyncssh.SSHClientConnection

Return type:

AsyncIterator[ExposeSession]

Use specific providers from pyexpose.providers.localhost_run.LocalRunConnector() or pyexpose.providers.serveo.ServeoConnector()
from pyexpose.providers.serveo import ServeoConnector
connector = ServeoConnector()
async with connector.connect() as session:
    async with session.tunnel(8080) as tunnel:
        print("serveo.net exposed ip is " + tunnel.ip + " port is " + tunnel.port)
connect_with_connection(connection: SSHClientConnection) → AsyncIterator[ExposeSessionVar][source]

Connects to the server using an existing connection.

Parameters:

connection (asyncssh.SSHClientConnection) – The connection to use.

Returns:

ExposeSession that wraps around asyncssh.SSHClientConnection

Return type:

AsyncIterator[ExposeSession]

This is how to setup custom ssh tunneling service by taking example of https://localhost.run/, Requiring SSH connection to be done like ssh -R 80:localhost:8080 nokey@localhost.run
import asyncssh
async with asyncssh.connect('localhost.run', username="nokey", password="", known_hosts=None) as conn:
    async with connector.connect_with_connection(conn) as session:
        session.tunnel(<args>)
        ...
abstract get_factory() → Callable[[], ExposeSessionVar][source]

Returns a factory function that creates a new ExposeSessionVar instance.

Returns:

A factory function that creates a new ExposeSessionVar instance.

Return type:

Callable[[], ExposeSessionVar]

abstract open_connection() → AsyncIterator[SSHClientConnection][source]

Opens a connection to the server, Override this for creating new connectors

Returns:

asyncssh.SSHClientConnection

Return type:

AsyncIterator[asyncssh.SSHClientConnection]

pyexpose.base.logger module

pyexpose.base.logger.enable_logging()[source]

Enable logging for debugging purposes.

pyexpose.base.session module

class pyexpose.base.session.ExposeSession(session: SSHClientConnection, stdin: SSHWriter, stdout: SSHReader, stderr: SSHReader)[source]

Bases: ABC

Wrapper around asyncssh.SSHClientConnection that provides a context manager and several utilities for interacting with the server, and tunnels.

Variables:

session (asyncssh.SSHClientConnection) – The underlying asyncssh.SSHClientConnection that is used to communicate with the server.

abstract async parse_ip_port() → Tuple[str, int][source]

Parses the ip address and port from the server.

Returns:

The ip address and port.

Return type:

Tuple[str, int]

tunnel(local_port: int, local_host: str = 'localhost', remote_port: int = 80, remote_host: str = '') → AsyncIterator[ExposeTunnel][source]

Creates a tunnel between the local and remote ports, this is equipvalent to running the command ssh -R <remote_host>:<remote_port>:<local_host>:<local_port> <server>, which essentially forwards all the traffic from the remote_host:remote_port to the our local_host:local_port.

Parameters:

  • local_port (int) – The local port to forward traffic from. eg. 8080

  • local_host (str, optional) – The local host to forward traffic from. eg. localhost

  • remote_port (int, optional) – The remote port to forward traffic from to. eg. 80 means <server>:80 -> local_ip:local_port

  • remote_host (str, optional) – The remote host to forward traffic from to. eg. <server>, or a custom domain. Defaults to ‘’ (any).

Returns:

The listener that is listening for connections.

Return type:

AsyncIterator[ExposeTunnel]

Raises:

asyncssh.ChannelListenError if the listener can’t be opened or asyncio.TimeoutError if <server> doesn’t respond with the ip address and port.

# Simple HTTP expose
async with session.tunnel(8080) as tunnel:

# Expose HTTPS (remote:443 -> localhost:8443)
async with session.tunnel(8443, remote_port=443) as tunnel:

# Tunnel TCP (remote:1234 -> localhost:2345)
async with session.tunnel(2345, remote_port=1234) as tunnel:

# Use custom subdomain/domain (if provider supports)
async with session.tunnel(8080, remote_host="myreservedaddr.serveo.net") as tunnel:

# Tunnel to something other than localhost, ie remote:443 -> notlocal:8080
async with session.tunnel(8080, remote_port=443, local_host="notlocal") as tunnel:

pyexpose.base.tunnel module

class pyexpose.base.tunnel.ExposeTunnel(ip: str, port: int, listener: SSHListener)[source]

Bases: object

This class represents a tunnel that is created between the local and remote ports, wrapper around asyncssh.SSHListener containing the ip address and port that the server is listening on.

Variables:

  • ip (str) – The ip address that the server is listening on.

  • port (int) – The port that the server is listening on.

  • listener (asyncssh.SSHListener) – The underlying asyncssh.SSHListener that is listening for connections.

Module contents