Merge b2e621cafc into d7398baad7

README.md

@@ -221,6 +221,8 @@ For convenience, a web front-end on top of the command-line tool is available at
  - Fixed crash when running with `-P` and `-T` options simultaneously.
  - Fixed host key tests from only reporting a key type at most once despite multiple hosts supporting it; credit [Daniel Lenski](https://github.com/dlenskiSB).
  - Fixed DHEat connection rate testing on MacOS X and BSD platforms; credit [Drew Noel](https://github.com/drewmnoel) and [Michael Osipov](https://github.com/michael-o).
+ - Added 1 new cipher: `grasshopper-ctr128`.
+ - Added 2 new key exchanges: `mlkem768x25519-sha256`, `sntrup761x25519-sha512`.
 
 ### v3.2.0 (2024-04-22)
  - Added implementation of the DHEat denial-of-service attack (see `--dheat` option; [CVE-2002-20001](https://nvd.nist.gov/vuln/detail/CVE-2002-20001)).

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:

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.'''
 

src/ssh_audit/ssh2_kexdb.py

@@ -64,11 +64,13 @@ class SSH2_KexDB:  # pylint: disable=too-few-public-methods
     INFO_DEFAULT_OPENSSH_CIPHER = 'default cipher since OpenSSH 6.9'
     INFO_DEFAULT_OPENSSH_KEX_65_TO_73 = 'default key exchange from OpenSSH 6.5 to 7.3'
     INFO_DEFAULT_OPENSSH_KEX_74_TO_89 = 'default key exchange from OpenSSH 7.4 to 8.9'
-    INFO_DEFAULT_OPENSSH_KEX_90 = 'default key exchange since OpenSSH 9.0'
+    INFO_DEFAULT_OPENSSH_KEX_90_TO_98 = 'default key exchange from OpenSSH 9.0 to 9.8'
+    INFO_DEFAULT_OPENSSH_KEX_99 = 'default key exchange since OpenSSH 9.9'
     INFO_DEPRECATED_IN_OPENSSH88 = 'deprecated in OpenSSH 8.8: https://www.openssh.com/txt/release-8.8'
     INFO_DISABLED_IN_DBEAR67 = 'disabled in Dropbear SSH 2015.67'
     INFO_DISABLED_IN_OPENSSH70 = 'disabled in OpenSSH 7.0: https://www.openssh.com/txt/release-7.0'
     INFO_NEVER_IMPLEMENTED_IN_OPENSSH = 'despite the @openssh.com tag, this was never implemented in OpenSSH'
+    INFO_HYBRID_PQ_X25519_KEX = 'hybrid key exchange based on post-quantum resistant algorithm and proven conventional X25519 algorithm'
     INFO_REMOVED_IN_OPENSSH61 = 'removed since OpenSSH 6.1, removed from specification'
     INFO_REMOVED_IN_OPENSSH69 = 'removed in OpenSSH 6.9: https://www.openssh.com/txt/release-6.9'
     INFO_REMOVED_IN_OPENSSH70 = 'removed in OpenSSH 7.0: https://www.openssh.com/txt/release-7.0'
@@ -189,11 +191,13 @@ class SSH2_KexDB:  # pylint: disable=too-few-public-methods
             'kexguess2@matt.ucc.asn.au': [['d2013.57']],
             'm383-sha384@libassh.org': [[], [FAIL_UNPROVEN]],
             'm511-sha512@libassh.org': [[], [FAIL_UNPROVEN]],
+            'mlkem768x25519-sha256': [['9.9'], [], [], [INFO_HYBRID_PQ_X25519_KEX]],
             'rsa1024-sha1': [[], [FAIL_1024BIT_MODULUS, FAIL_SHA1]],
             'rsa2048-sha256': [[], [], [WARN_2048BIT_MODULUS]],
             'sm2kep-sha2-nistp256': [[], [FAIL_NSA_BACKDOORED_CURVE, FAIL_UNTRUSTED]],
             'sntrup4591761x25519-sha512@tinyssh.org': [['8.0', '8.4'], [], [WARN_EXPERIMENTAL], [INFO_WITHDRAWN_PQ_ALG]],
-            'sntrup761x25519-sha512@openssh.com': [['8.5'], [], [], [INFO_DEFAULT_OPENSSH_KEX_90]],
+            'sntrup761x25519-sha512': [['9.9'], [], [], [INFO_DEFAULT_OPENSSH_KEX_99, INFO_HYBRID_PQ_X25519_KEX]],
+            'sntrup761x25519-sha512@openssh.com': [['8.5'], [], [], [INFO_DEFAULT_OPENSSH_KEX_90_TO_98, INFO_HYBRID_PQ_X25519_KEX]],
             'x25519-kyber-512r3-sha256-d00@amazon.com': [[]],
             'x25519-kyber512-sha512@aws.amazon.com': [[]],
         },
@@ -346,6 +350,7 @@ class SSH2_KexDB:  # pylint: disable=too-few-public-methods
             'des-cbc-ssh1': [[], [FAIL_DES], [WARN_CIPHER_MODE, WARN_BLOCK_SIZE]],
             'des-cbc@ssh.com': [[], [FAIL_DES], [WARN_CIPHER_MODE, WARN_BLOCK_SIZE]],
             'des': [[], [FAIL_DES], [WARN_CIPHER_MODE, WARN_BLOCK_SIZE]],
+            'grasshopper-ctr128': [[], [FAIL_UNTRUSTED]],
             'idea-cbc': [[], [FAIL_IDEA], [WARN_CIPHER_MODE]],
             'idea-cfb': [[], [FAIL_IDEA], [WARN_CIPHER_MODE]],
             'idea-ctr': [[], [FAIL_IDEA]],

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] <host>\n'.format(p))
     uout.info('   -h,  --help             print this help')
@@ -723,7 +723,7 @@ def output(out: OutputBuffer, aconf: AuditConf, banner: Optional[Banner], header
         # Build & write the JSON struct.
         out.info(json.dumps(build_struct(aconf.host + ":" + str(aconf.port), banner, cves, kex=kex, client_host=client_host, software=software, algorithms=algs, algorithm_recommendation_suppress_list=algorithm_recommendation_suppress_list, additional_notes=additional_notes), indent=4 if aconf.json_print_indent else None, sort_keys=True))
     elif len(unknown_algorithms) > 0:  # If we encountered any unknown algorithms, ask the user to report them.
-        out.warn("\n\n!!! WARNING: unknown algorithm(s) found!: %s.  Please email the full output above to the maintainer (jtesta@positronsecurity.com), or create a Github issue at <https://github.com/jtesta/ssh-audit/issues>.\n" % ','.join(unknown_algorithms))
+        out.warn("\n\n!!! WARNING: unknown algorithm(s) found!: %s.  If this is the latest version of ssh-audit (see <https://github.com/jtesta/ssh-audit/releases>), please create a new Github issue at <https://github.com/jtesta/ssh-audit/issues> with the full output above.\n" % ','.join(unknown_algorithms))
 
     return program_retval
 
@@ -835,7 +835,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")
@@ -1051,19 +1051,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)
 
@@ -1262,7 +1262,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:
@@ -1310,7 +1310,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))
@@ -1319,7 +1319,7 @@ def audit(out: OutputBuffer, aconf: AuditConf, sshv: Optional[int] = None, print
             kex = SSH2_Kex.parse(out, payload)
             out.d(str(kex))
         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:

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)