Skip to content

Commit

Permalink
Add fix for trying to call get_addresses() when tunneling
Browse files Browse the repository at this point in the history
This commit adds support for calling the new SSHAcceptor method
get_addresses() on tunneled SSL listeners, and includes some other
minor cleanup from the last commit.
  • Loading branch information
ronf committed Oct 29, 2023
1 parent 71abf74 commit 4d8f2a4
Show file tree
Hide file tree
Showing 4 changed files with 23 additions and 15 deletions.
26 changes: 14 additions & 12 deletions asyncssh/connection.py
Original file line number Diff line number Diff line change
Expand Up @@ -694,36 +694,38 @@ def __getattr__(self, name: str) -> Any:
return getattr(self._server, name)

def get_addresses(self) -> List[Tuple]:
"""Get socket addresses being listened on
"""Return socket addresses being listened on
This method returns the IP addresses and ports being
listened on. It returns tuples of the form returned by
This method returns the socket addresses 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
:returns: A list of socket addresses being listened on
"""

return [sock.getsockname() for sock in self.sockets]
if hasattr(self._server, 'get_addresses'):
return self._server.get_addresses()
else:
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.
This method returns the port number being listened on.
If it is listening on multiple sockets with different port
numbers, this function will return `0`. In that case,
:meth:`get_addresses` can be used to retrieve the full
list of listening addresses and ports.
:returns: The port number being listened on
:returns: The port number being listened on, if there's only one
"""

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())
Expand Down
5 changes: 5 additions & 0 deletions asyncssh/listener.py
Original file line number Diff line number Diff line change
Expand Up @@ -184,6 +184,11 @@ def process_connection(self, orig_host: str, orig_port: int) -> \

return chan, self._session_factory(orig_host, orig_port)

def get_addresses(self) -> List[Tuple]:
"""Return the socket addresses being listened on"""

return [(self._listen_host, self._listen_port)]

def get_port(self) -> int:
"""Return the port number being listened on"""

Expand Down
1 change: 0 additions & 1 deletion tests/test_connection.py
Original file line number Diff line number Diff line change
Expand Up @@ -2278,7 +2278,6 @@ async def test_ssh_listen_context_manager(self):
async with self.listen() as server:
listen_port = server.get_port()


async with asyncssh.connect('127.0.0.1', listen_port,
known_hosts=(['skey.pub'], [], [])):
pass
6 changes: 4 additions & 2 deletions tests/test_forward.py
Original file line number Diff line number Diff line change
Expand Up @@ -422,8 +422,10 @@ async def test_ssh_listen_tunnel(self):

async with self.connect() as conn:
async with conn.listen_ssh(port=0, server_factory=Server,
server_host_keys=['skey']) as server2:
listen_port = server2.get_port()
server_host_keys=['skey']) as server:
listen_port = server.get_port()

self.assertEqual(server.get_addresses(), [('', listen_port)])

async with asyncssh.connect('127.0.0.1', listen_port,
known_hosts=(['skey.pub'], [], [])):
Expand Down

0 comments on commit 4d8f2a4

Please sign in to comment.