From b2e621cafc0c387231ddfba99a6bfea66cbcb6a7 Mon Sep 17 00:00:00 2001 From: FlyingFish Date: Wed, 1 May 2024 23:00:25 +0100 Subject: [PATCH] Modified OutputBuffer to have an error function to output to stderr. Change .fail with errors to .error --- src/ssh_audit/gextest.py | 2 +- src/ssh_audit/outputbuffer.py | 8 ++++++++ src/ssh_audit/ssh_audit.py | 16 ++++++++-------- src/ssh_audit/ssh_socket.py | 14 ++++++++------ 4 files changed, 25 insertions(+), 15 deletions(-) diff --git a/src/ssh_audit/gextest.py b/src/ssh_audit/gextest.py index 3ce86d1..61c35bd 100644 --- a/src/ssh_audit/gextest.py +++ b/src/ssh_audit/gextest.py @@ -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: diff --git a/src/ssh_audit/outputbuffer.py b/src/ssh_audit/outputbuffer.py index 38723e3..0798310 100644 --- a/src/ssh_audit/outputbuffer.py +++ b/src/ssh_audit/outputbuffer.py @@ -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.''' diff --git a/src/ssh_audit/ssh_audit.py b/src/ssh_audit/ssh_audit.py index 4fff3ae..7df2e78 100755 --- a/src/ssh_audit/ssh_audit.py +++ b/src/ssh_audit/ssh_audit.py @@ -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] \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: diff --git a/src/ssh_audit/ssh_socket.py b/src/ssh_audit/ssh_socket.py index 4afaa57..8832c93 100644 --- a/src/ssh_audit/ssh_socket.py +++ b/src/ssh_audit/ssh_socket.py @@ -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)