Skip to content

Commit

Permalink
Add get_addresses() and get_port() to SSHAcceptor class
Browse files Browse the repository at this point in the history
This commit adds functions to get listening socket information on
SSH listeners. New method get_addresses() gets a list of all of the
IP addresses and ports being listened on, and get_port() will return
just the listening port, but only when there's a single listening
socket or if all the sockets are listening on the same port. Otherwise,
a value of 0 is returned. Thanks go to Allison Karlitskaya for
suggesting this feature.
  • Loading branch information
ronf committed Oct 28, 2023
1 parent 5ca8487 commit 71abf74
Show file tree
Hide file tree
Showing 3 changed files with 55 additions and 5 deletions.
50 changes: 48 additions & 2 deletions asyncssh/connection.py
Original file line number Diff line number Diff line change
Expand Up @@ -693,13 +693,59 @@ async def __aexit__(self, _exc_type: Optional[Type[BaseException]],
def __getattr__(self, name: str) -> Any:
return getattr(self._server, name)

def get_addresses(self) -> List[Tuple]:
"""Get socket addresses being listened on
This method returns the IP addresses and ports being
listened on. It returns tuples of the form returned by
:meth:`socket.getsockname`. If the listener was created
using a hostname, the host's resolved IPs will be returned.
If the requested listening port was `0`, the selected
listening ports will be returned.
:returns: A list of IP addresses and ports being listened on
"""

return [sock.getsockname() for sock in self.sockets]

def get_port(self) -> int:
"""Return the port number being listened on
This method returns the port number this SSHAcceptor is
listening on. If it is listening on multiple sockets with
different port numbers, this function will return `0`. In
this case, :meth:`get_addresses` can be used to retrieve
the full list of listening addresses and ports.
:returns: The port number being listened on
"""

if hasattr(self._server, 'get_port'):
print(type(self._server))
return self._server.get_port()
else:
ports = set(addr[1] for addr in self.get_addresses())
return ports.pop() if len(ports) == 1 else 0

def close(self) -> None:
"""Close this SSH listener"""
"""Stop listening for new connections
This method can be called to stop listening for new
SSH connections. Existing connections will remain open.
"""

self._server.close()

async def wait_closed(self) -> None:
"""Wait for this SSH listener to close"""
"""Wait for this listener to close
This method is a coroutine which waits for this
listener to be closed.
"""

await self._server.wait_closed()

Expand Down
8 changes: 6 additions & 2 deletions docs/api.rst
Original file line number Diff line number Diff line change
Expand Up @@ -1038,9 +1038,13 @@ SSHAcceptor

.. autoclass:: SSHAcceptor()

====================== =
============================= =
.. automethod:: get_addresses
.. automethod:: get_port
.. automethod:: close
.. automethod:: wait_closed
.. automethod:: update
====================== =
============================= =


SSHListener
Expand Down
2 changes: 1 addition & 1 deletion tests/test_connection.py
Original file line number Diff line number Diff line change
Expand Up @@ -2276,7 +2276,7 @@ async def test_ssh_listen_context_manager(self):
"""Test using an SSH listener as a context manager"""

async with self.listen() as server:
listen_port = server.sockets[0].getsockname()[1]
listen_port = server.get_port()


async with asyncssh.connect('127.0.0.1', listen_port,
Expand Down

0 comments on commit 71abf74

Please sign in to comment.