Git/ssh-audit
1
0
Fork 0
mirror of https://github.com/jtesta/ssh-audit.git synced 2024-11-24 19:41:40 +01:00
Code Issues Packages Projects Releases Wiki Activity

Updated program return values for various connection error instances and unknown errors.

Browse Source
This commit is contained in:
Joe Testa 2020-08-11 19:45:59 -04:00
parent c7ad1828d8
commit 6d84cfdc31
1 changed files with 12 additions and 10 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

22
ssh-audit.py
View File

@ -63,11 +63,13 @@ except ImportError: # pragma: nocover
def usage(err: Optional[str] = None) -> None: def usage(err: Optional[str] = None) -> None:
retval = PROGRAM_RETVAL_GOOD
uout = Output() uout = Output()
p = os.path.basename(sys.argv[0]) p = os.path.basename(sys.argv[0])
uout.head('# {} {}, https://github.com/jtesta/ssh-audit\n'.format(p, VERSION)) uout.head('# {} {}, https://github.com/jtesta/ssh-audit\n'.format(p, VERSION))
if err is not None and len(err) > 0: if err is not None and len(err) > 0:
uout.fail('\n' + err) uout.fail('\n' + err)
retval = PROGRAM_RETVAL_UNKNOWN_ERROR
uout.info('usage: {0} [options] <host>\n'.format(p)) uout.info('usage: {0} [options] <host>\n'.format(p))
uout.info(' -h, --help print this help') uout.info(' -h, --help print this help')
uout.info(' -1, --ssh1 force ssh version 1 only') uout.info(' -1, --ssh1 force ssh version 1 only')
@ -88,7 +90,7 @@ def usage(err: Optional[str] = None) -> None:
uout.info(' -T, --targets=<hosts.txt> a file containing a list of target hosts (one\n per line, format HOST[:PORT])') uout.info(' -T, --targets=<hosts.txt> a file containing a list of target hosts (one\n per line, format HOST[:PORT])')
uout.info(' -v, --verbose verbose output') uout.info(' -v, --verbose verbose output')
uout.sep() uout.sep()
sys.exit(1) sys.exit(retval)
# Validates policy files and performs policy testing # Validates policy files and performs policy testing
@ -631,17 +633,17 @@ class AuditConf:
aconf.policy = Policy(policy_file=aconf.policy_file) aconf.policy = Policy(policy_file=aconf.policy_file)
except Exception as e: except Exception as e:
print("Error while loading policy file: %s: %s" % (str(e), traceback.format_exc())) print("Error while loading policy file: %s: %s" % (str(e), traceback.format_exc()))
sys.exit(-1) sys.exit(PROGRAM_RETVAL_UNKNOWN_ERROR)
# If the user wants to do a client audit, but provided a server policy, terminate. # If the user wants to do a client audit, but provided a server policy, terminate.
if aconf.client_audit and aconf.policy.is_server_policy(): if aconf.client_audit and aconf.policy.is_server_policy():
print("Error: client audit selected, but server policy provided.") print("Error: client audit selected, but server policy provided.")
sys.exit(-1) sys.exit(PROGRAM_RETVAL_UNKNOWN_ERROR)
# If the user wants to do a server audit, but provided a client policy, terminate. # If the user wants to do a server audit, but provided a client policy, terminate.
if aconf.client_audit is False and aconf.policy.is_server_policy() is False: if aconf.client_audit is False and aconf.policy.is_server_policy() is False:
print("Error: server audit selected, but client policy provided.") print("Error: server audit selected, but client policy provided.")
sys.exit(-1) sys.exit(PROGRAM_RETVAL_UNKNOWN_ERROR)
return aconf return aconf
@ -2386,7 +2388,7 @@ class SSH: # pylint: disable=too-few-public-methods
yield af, addr yield af, addr
except socket.error as e: except socket.error as e:
out.fail('[exception] {}'.format(e)) out.fail('[exception] {}'.format(e))
sys.exit(1) sys.exit(PROGRAM_RETVAL_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).
@ -2416,7 +2418,7 @@ class SSH: # pylint: disable=too-few-public-methods
# If we failed to listen on any interfaces, terminate. # If we failed to listen on any interfaces, terminate.
if len(self.__sock_map.keys()) == 0: if len(self.__sock_map.keys()) == 0:
print("Error: failed to listen on any IPv4 and IPv6 interfaces!") print("Error: failed to listen on any IPv4 and IPv6 interfaces!")
sys.exit(-1) sys.exit(PROGRAM_RETVAL_CONNECTION_ERROR)
# Wait for an incoming connection. If a timeout was explicitly # Wait for an incoming connection. If a timeout was explicitly
# set by the user, terminate when it elapses. # set by the user, terminate when it elapses.
@ -2434,7 +2436,7 @@ class SSH: # pylint: disable=too-few-public-methods
if self.__timeout_set and time_elapsed >= self.__timeout: if self.__timeout_set and time_elapsed >= self.__timeout:
print("Timeout elapsed. Terminating...") print("Timeout elapsed. Terminating...")
sys.exit(-1) sys.exit(PROGRAM_RETVAL_CONNECTION_ERROR)
# Accept the connection. # Accept the connection.
c, addr = self.__sock_map[fds[0][0]].accept() c, addr = self.__sock_map[fds[0][0]].accept()
@ -2462,7 +2464,7 @@ class SSH: # pylint: disable=too-few-public-methods
errt = (self.__host, self.__port, err) errt = (self.__host, self.__port, err)
errm = 'cannot connect to {} port {}: {}'.format(*errt) errm = 'cannot connect to {} port {}: {}'.format(*errt)
out.fail('[exception] {}'.format(errm)) out.fail('[exception] {}'.format(errm))
sys.exit(1) sys.exit(PROGRAM_RETVAL_CONNECTION_ERROR)
def get_banner(self, sshv: int = 2) -> Tuple[Optional['SSH.Banner'], List[str], Optional[str]]: def get_banner(self, sshv: int = 2) -> Tuple[Optional['SSH.Banner'], List[str], Optional[str]]:
if self.__sock is None: if self.__sock is None:
@ -2553,7 +2555,7 @@ class SSH: # pylint: disable=too-few-public-methods
check_size = 4 + 1 + payload_length + padding_length check_size = 4 + 1 + payload_length + padding_length
if check_size % self.__block_size != 0: if check_size % self.__block_size != 0:
out.fail('[exception] invalid ssh packet (block size)') out.fail('[exception] invalid ssh packet (block size)')
sys.exit(1) sys.exit(PROGRAM_RETVAL_CONNECTION_ERROR)
self.ensure_read(payload_length) self.ensure_read(payload_length)
if sshv == 1: if sshv == 1:
payload = self.read(payload_length - 4) payload = self.read(payload_length - 4)
@ -2568,7 +2570,7 @@ class SSH: # pylint: disable=too-few-public-methods
rcrc = SSH1.crc32(padding + payload) rcrc = SSH1.crc32(padding + payload)
if crc != rcrc: if crc != rcrc:
out.fail('[exception] packet checksum CRC32 mismatch.') out.fail('[exception] packet checksum CRC32 mismatch.')
sys.exit(1) sys.exit(PROGRAM_RETVAL_CONNECTION_ERROR)
else: else:
self.ensure_read(padding_length) self.ensure_read(padding_length)
padding = self.read(padding_length) padding = self.read(padding_length)