Git
/
ssh-audit
mirror of https://github.com/jtesta/ssh-audit.git
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Merge b2e621cafc into 46ec4e3edc

This commit is contained in:
FlyingPhishy 2024-05-01 22:02:59 +00:00 committed by GitHub
parent 46ec4e3edc b2e621cafc
commit f577809c10
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
4 changed files with 25 additions and 15 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
src/ssh_audit/gextest.py
View File

@ -110,7 +110,7 @@ class GEXTest:
# before continuing to issue reconnects.
modulus_size_returned, reconnect_failed = GEXTest._send_init(out, s, kex_group, kex, gex_alg, bits_min, bits_pref, bits_max)
if reconnect_failed:
out.fail('Reconnect failed.')
out.error('Reconnect failed.')
return exitcodes.FAILURE
if modulus_size_returned > 0:

8
src/ssh_audit/outputbuffer.py
View File

@ -54,6 +54,14 @@ class OutputBuffer:
self.__is_color_supported = ('colorama' in sys.modules) or (os.name == 'posix')
self.line_ended = True
def error(self, msg, line_ended=True):
"""
Writes an error message to stderr.
"""
end = '' if line_ended else '\n'
sys.stderr.write(f'{msg}{end}')
sys.stderr.flush()
def _print(self, level: str, s: str = '', line_ended: bool = True) -> None:
'''Saves output to buffer (if in buffered mode), or immediately prints to stdout otherwise.'''

16
src/ssh_audit/ssh_audit.py
View File

@ -88,7 +88,7 @@ def usage(uout: OutputBuffer, err: Optional[str] = None) -> None:
p = os.path.basename(sys.argv[0])
uout.head('# {} {}, https://github.com/jtesta/ssh-audit\n'.format(p, VERSION))
if err is not None and len(err) > 0:
uout.fail(err + '\n')
uout.error(err + '\n')
retval = exitcodes.UNKNOWN_ERROR
uout.info('usage: {0} [options] <host>\n'.format(p))
uout.info(' -h, --help print this help')
@ -833,7 +833,7 @@ def list_policies(out: OutputBuffer, verbose: bool) -> None:
out.sep()
if len(server_policy_names) == 0 and len(client_policy_names) == 0:
out.fail("Error: no built-in policies found!")
out.error("Error: no built-in policies found!")
else:
out.info("\nHint: Use -P and provide the full name of a policy to run a policy scan with.\n")
out.info("Hint: Use -L -v to also see the change log for each policy.\n")
@ -1049,19 +1049,19 @@ def process_commandline(out: OutputBuffer, args: List[str], usage_cb: Callable[.
try:
aconf.policy = Policy(policy_file=aconf.policy_file, json_output=aconf.json)
except Exception as e:
out.fail("Error while loading policy file: %s: %s" % (str(e), traceback.format_exc()))
out.error("Error while loading policy file: %s: %s" % (str(e), traceback.format_exc()))
out.write()
sys.exit(exitcodes.UNKNOWN_ERROR)
# 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():
out.fail("Error: client audit selected, but server policy provided.")
out.error("Error: client audit selected, but server policy provided.")
out.write()
sys.exit(exitcodes.UNKNOWN_ERROR)
# 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:
out.fail("Error: server audit selected, but client policy provided.")
out.error("Error: server audit selected, but client policy provided.")
out.write()
sys.exit(exitcodes.UNKNOWN_ERROR)
@ -1260,7 +1260,7 @@ def audit(out: OutputBuffer, aconf: AuditConf, sshv: Optional[int] = None, print
err = s.connect()
if err is not None:
out.fail(err)
out.error(err)
# If we're running against multiple targets, return a connection error to the calling worker thread. Otherwise, write the error message to the console and exit.
if len(aconf.target_list) > 0:
@ -1308,7 +1308,7 @@ def audit(out: OutputBuffer, aconf: AuditConf, sshv: Optional[int] = None, print
err = fmt.format(err_pair[0], err_pair[1], packet_type)
if err is not None:
output(out, aconf, banner, header)
out.fail(err)
out.error(err)
return exitcodes.CONNECTION_ERROR
if sshv == 1:
program_retval = output(out, aconf, banner, header, pkm=SSH1_PublicKeyMessage.parse(payload))
@ -1316,7 +1316,7 @@ def audit(out: OutputBuffer, aconf: AuditConf, sshv: Optional[int] = None, print
try:
kex = SSH2_Kex.parse(out, payload)
except Exception:
out.fail("Failed to parse server's kex. Stack trace:\n%s" % str(traceback.format_exc()))
out.error("Failed to parse server's kex. Stack trace:\n%s" % str(traceback.format_exc()))
return exitcodes.CONNECTION_ERROR
if aconf.dheat is not None:

14
src/ssh_audit/ssh_socket.py
View File

@ -108,7 +108,8 @@ class SSH_Socket(ReadBuf, WriteBuf):
s.listen()
self.__sock_map[s.fileno()] = s
except Exception as e:
print("Warning: failed to listen on any IPv4 interfaces: %s" % str(e))
self.__outputbuffer.error("Warning: failed to listen on any IPv4 interfaces: %s" % str(e))
sys.exit(exitcodes.CONNECTION_ERROR)
try:
# Socket to listen on all IPv6 addresses.
@ -119,11 +120,12 @@ class SSH_Socket(ReadBuf, WriteBuf):
s.listen()
self.__sock_map[s.fileno()] = s
except Exception as e:
print("Warning: failed to listen on any IPv6 interfaces: %s" % str(e))
self.__outputbuffer.error("Warning: failed to listen on any IPv6 interfaces: %s" % str(e))
sys.exit(exitcodes.CONNECTION_ERROR)
# If we failed to listen on any interfaces, terminate.
if len(self.__sock_map.keys()) == 0:
print("Error: failed to listen on any IPv4 and IPv6 interfaces!")
self.__outputbuffer.error("Error: failed to listen on any IPv4 and IPv6 interfaces!")
sys.exit(exitcodes.CONNECTION_ERROR)
# Wait for an incoming connection. If a timeout was explicitly
@ -141,7 +143,7 @@ class SSH_Socket(ReadBuf, WriteBuf):
break
if self.__timeout_set and time_elapsed >= self.__timeout:
print("Timeout elapsed. Terminating...")
self.__outputbuffer.error("Timeout elapsed. Terminating...")
sys.exit(exitcodes.CONNECTION_ERROR)
# Accept the connection.
@ -275,7 +277,7 @@ class SSH_Socket(ReadBuf, WriteBuf):
payload_length = packet_length - padding_length - 1
check_size = 4 + 1 + payload_length + padding_length
if check_size % self.__block_size != 0:
self.__outputbuffer.fail('[exception] invalid ssh packet (block size)').write()
self.__outputbuffer.error('[exception] invalid ssh packet (block size)').write()
sys.exit(exitcodes.CONNECTION_ERROR)
self.ensure_read(payload_length)
if sshv == 1:
@ -290,7 +292,7 @@ class SSH_Socket(ReadBuf, WriteBuf):
if sshv == 1:
rcrc = SSH1.crc32(padding + payload)
if crc != rcrc:
self.__outputbuffer.fail('[exception] packet checksum CRC32 mismatch.').write()
self.__outputbuffer.error('[exception] packet checksum CRC32 mismatch.').write()
sys.exit(exitcodes.CONNECTION_ERROR)
else:
self.ensure_read(padding_length)