pt/ssh-audit
1
0
Fork 0
mirror of https://github.com/jtesta/ssh-audit.git synced 2024-09-19 12:09:45 -04:00
Code Issues Packages Projects Releases Wiki Activity

Removed sys.exit from _resolve in ssh_socket.py (#187)

Browse Source 
Changed (and documented) _resolve so the application does not quit when a hostname cannot be resolved.

Adapted connect function to expect incoming exceptions from _resolve

This fixes issue #186
This commit is contained in:
Dani Cuesta 2023-06-20 15:21:06 +02:00 committed by GitHub
parent e99cb0b579
commit a74c3abdde
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
1 changed files with 20 additions and 18 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

38
src/ssh_audit/ssh_socket.py
View File

@ -75,24 +75,26 @@ class SSH_Socket(ReadBuf, WriteBuf):
self.client_port = None self.client_port = None
def _resolve(self) -> Iterable[Tuple[int, Tuple[Any, ...]]]: def _resolve(self) -> Iterable[Tuple[int, Tuple[Any, ...]]]:
"""Resolves a hostname into a list of IPs
Raises
------
socket.gaierror [Errno -2]
If the hostname cannot be resolved.
"""
# If __ip_version_preference has only one entry, then it means that ONLY that IP version should be used. # If __ip_version_preference has only one entry, then it means that ONLY that IP version should be used.
if len(self.__ip_version_preference) == 1: if len(self.__ip_version_preference) == 1:
family = socket.AF_INET if self.__ip_version_preference[0] == 4 else socket.AF_INET6 family = socket.AF_INET if self.__ip_version_preference[0] == 4 else socket.AF_INET6
else: else:
family = socket.AF_UNSPEC family = socket.AF_UNSPEC
try: stype = socket.SOCK_STREAM
stype = socket.SOCK_STREAM r = socket.getaddrinfo(self.__host, self.__port, family, stype)
r = socket.getaddrinfo(self.__host, self.__port, family, stype)
# If the user has a preference for using IPv4 over IPv6 (or vice-versa), then sort the list returned by getaddrinfo() so that the preferred address type comes first. # If the user has a preference for using IPv4 over IPv6 (or vice-versa), then sort the list returned by getaddrinfo() so that the preferred address type comes first.
if len(self.__ip_version_preference) == 2: if len(self.__ip_version_preference) == 2:
r = sorted(r, key=lambda x: x[0], reverse=(self.__ip_version_preference[0] == 6)) # pylint: disable=superfluous-parens r = sorted(r, key=lambda x: x[0], reverse=(self.__ip_version_preference[0] == 6)) # pylint: disable=superfluous-parens
for af, socktype, _proto, _canonname, addr in r: for af, socktype, _proto, _canonname, addr in r:
if socktype == socket.SOCK_STREAM: if socktype == socket.SOCK_STREAM:
yield af, addr yield af, addr
except socket.error as e:
self.__outputbuffer.fail('[exception] {}'.format(e)).write()
sys.exit(exitcodes.CONNECTION_ERROR)
# Listens on a server socket and accepts one connection (used for # Listens on a server socket and accepts one connection (used for
# auditing client connections). # auditing client connections).
@ -152,18 +154,18 @@ class SSH_Socket(ReadBuf, WriteBuf):
def connect(self) -> Optional[str]: def connect(self) -> Optional[str]:
'''Returns None on success, or an error string.''' '''Returns None on success, or an error string.'''
err = None err = None
for af, addr in self._resolve(): s = None
s = None try:
try: for af, addr in self._resolve():
s = socket.socket(af, socket.SOCK_STREAM) s = socket.socket(af, socket.SOCK_STREAM)
s.settimeout(self.__timeout) s.settimeout(self.__timeout)
self.__outputbuffer.d(("Connecting to %s:%d..." % ('[%s]' % addr[0] if Utils.is_ipv6_address(addr[0]) else addr[0], addr[1])), write_now=True) self.__outputbuffer.d(("Connecting to %s:%d..." % ('[%s]' % addr[0] if Utils.is_ipv6_address(addr[0]) else addr[0], addr[1])), write_now=True)
s.connect(addr) s.connect(addr)
self.__sock = s self.__sock = s
return None return None
except socket.error as e: except socket.error as e:
err = e err = e
self._close_socket(s) self._close_socket(s)
if err is None: if err is None:
errm = 'host {} has no DNS records'.format(self.__host) errm = 'host {} has no DNS records'.format(self.__host)
else: else: