Merge b2e621cafc into 9049c8476a

* Include raw hostkey bytes in debug output

* Recognize LANcom LCOS software and support extraction of ssh-ed448 key type

LANcom router devices appear to be primarily used in Germany (see [1]
for examples on the public Internet), and they appear to support the
`ssh-ed448` key type which is documented in [2], but which has never
been supported by any as-yet-released version of OpenSSH.

[1] https://www.shodan.io/search?query=ssh+%22ed448%22
[2] https://datatracker.ietf.org/doc/html/rfc8709#name-public-key-format

README.md

@@ -216,6 +216,8 @@ For convenience, a web front-end on top of the command-line tool is available at
 ### v3.3.0-dev (???)
  - Added built-in policies for Ubuntu 24.04 LTS server and client, and OpenSSH 9.8.
  - Added IPv6 support for DHEat and connection rate tests.
+ - Added TCP port information to JSON policy scan results; credit [Fabian Malte Kopp](https://github.com/dreizehnutters).
+ - Added LANcom LCOS server recognition and Ed448 key extraction; credit [Daniel Lenski](https://github.com/dlenskiSB).
  - 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).
 

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/hostkeytest.py

@@ -52,6 +52,9 @@ class HostKeyTest:
 
         'ssh-ed25519':                      {'cert': False, 'variable_key_len': False},
         'ssh-ed25519-cert-v01@openssh.com': {'cert': True, 'variable_key_len': False},
+
+        'ssh-ed448':                      {'cert': False, 'variable_key_len': False},
+        # 'ssh-ed448-cert-v01@openssh.com': {'cert': True,  'variable_key_len': False},
     }
 
     TWO2K_MODULUS_WARNING = '2048-bit modulus only provides 112-bits of symmetric strength'
@@ -158,6 +161,7 @@ class HostKeyTest:
                 ca_key_type = kex_group.get_ca_type()
                 ca_modulus_size = kex_group.get_ca_size()
                 out.d("Hostkey type: [%s]; hostkey size: %u; CA type: [%s]; CA modulus size: %u" % (host_key_type, hostkey_modulus_size, ca_key_type, ca_modulus_size), write_now=True)
+                out.d("Raw hostkey bytes (%d): [%s]" % (len(raw_hostkey_bytes), raw_hostkey_bytes.hex()), write_now=True)
 
                 # Record all the host key info.
                 server_kex.set_host_key(host_key_type, raw_hostkey_bytes, hostkey_modulus_size, ca_key_type, ca_modulus_size)

src/ssh_audit/kexdh.py

@@ -134,6 +134,9 @@ class KexDH:  # pragma: nocover
         if self.__hostkey_type == 'ssh-ed25519':
             self.out.d("%s has a fixed host key modulus of 32." % self.__hostkey_type)
             self.__hostkey_n_len = 32
+        elif self.__hostkey_type == 'ssh-ed448':
+            self.out.d("%s has a fixed host key modulus of 57." % self.__hostkey_type)
+            self.__hostkey_n_len = 57
         else:
             # Here is the modulus size & actual modulus of the host key public key.
             hostkey_n, self.__hostkey_n_len, ptr = KexDH.__get_bytes(hostkey, ptr)

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/software.py

@@ -224,4 +224,8 @@ class Software:
         mx = re.match(r'^PuTTY_Release_(.*)', software)
         if mx:
             return cls(None, Product.PuTTY, mx.group(1), None, None)
+        mx = re.match(r'^lancom(.*)', software)
+        if mx:
+            v, p = 'LANcom', 'LCOS sshd'
+            return cls(v, p, mx.group(1), None, None)
         return None

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')
@@ -735,7 +735,7 @@ def evaluate_policy(out: OutputBuffer, aconf: AuditConf, banner: Optional['Banne
 
     passed, error_struct, error_str = aconf.policy.evaluate(banner, kex)
     if aconf.json:
-        json_struct = {'host': aconf.host, 'policy': aconf.policy.get_name_and_version(), 'passed': passed, 'errors': error_struct}
+        json_struct = {'host': aconf.host, 'port': aconf.port, 'policy': aconf.policy.get_name_and_version(), 'passed': passed, 'errors': error_struct}
         out.info(json.dumps(json_struct, indent=4 if aconf.json_print_indent else None, sort_keys=True))
     else:
         spacing = ''
@@ -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))
@@ -1318,7 +1318,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:

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)

test/docker/expected_results/openssh_5.6p1_custom_policy_test1.json

@@ -2,5 +2,6 @@
     "errors": [],
     "host": "localhost",
     "passed": true,
-    "policy": "Docker policy: test1 (version 1)"
+    "policy": "Docker policy: test1 (version 1)",
+    "port": 2222
 }

test/docker/expected_results/openssh_5.6p1_custom_policy_test10.json

@@ -27,5 +27,6 @@
     ],
     "host": "localhost",
     "passed": false,
-    "policy": "Docker poliicy: test10 (version 1)"
+    "policy": "Docker poliicy: test10 (version 1)",
+    "port": 2222
 }

test/docker/expected_results/openssh_5.6p1_custom_policy_test2.json

@@ -19,5 +19,6 @@
     ],
     "host": "localhost",
     "passed": false,
-    "policy": "Docker policy: test2 (version 1)"
+    "policy": "Docker policy: test2 (version 1)",
+    "port": 2222
 }

test/docker/expected_results/openssh_5.6p1_custom_policy_test3.json

@@ -18,5 +18,6 @@
     ],
     "host": "localhost",
     "passed": false,
-    "policy": "Docker policy: test3 (version 1)"
+    "policy": "Docker policy: test3 (version 1)",
+    "port": 2222
 }

test/docker/expected_results/openssh_5.6p1_custom_policy_test4.json

@@ -28,5 +28,6 @@
     ],
     "host": "localhost",
     "passed": false,
-    "policy": "Docker policy: test4 (version 1)"
+    "policy": "Docker policy: test4 (version 1)",
+    "port": 2222
 }

test/docker/expected_results/openssh_5.6p1_custom_policy_test5.json

@@ -27,5 +27,6 @@
     ],
     "host": "localhost",
     "passed": false,
-    "policy": "Docker policy: test5 (version 1)"
+    "policy": "Docker policy: test5 (version 1)",
+    "port": 2222
 }

test/docker/expected_results/openssh_5.6p1_custom_policy_test7.json

@@ -2,5 +2,6 @@
     "errors": [],
     "host": "localhost",
     "passed": true,
-    "policy": "Docker poliicy: test7 (version 1)"
+    "policy": "Docker poliicy: test7 (version 1)",
+    "port": 2222
 }

test/docker/expected_results/openssh_5.6p1_custom_policy_test8.json

@@ -15,5 +15,6 @@
     ],
     "host": "localhost",
     "passed": false,
-    "policy": "Docker poliicy: test8 (version 1)"
+    "policy": "Docker poliicy: test8 (version 1)",
+    "port": 2222
 }

test/docker/expected_results/openssh_5.6p1_custom_policy_test9.json

@@ -15,5 +15,6 @@
     ],
     "host": "localhost",
     "passed": false,
-    "policy": "Docker poliicy: test9 (version 1)"
+    "policy": "Docker poliicy: test9 (version 1)",
+    "port": 2222
 }

test/docker/expected_results/openssh_8.0p1_builtin_policy_test1.json

@@ -39,5 +39,6 @@
     ],
     "host": "localhost",
     "passed": false,
-    "policy": "Hardened OpenSSH Server v8.0 (version 4)"
+    "policy": "Hardened OpenSSH Server v8.0 (version 4)",
+    "port": 2222
 }

test/docker/expected_results/openssh_8.0p1_builtin_policy_test2.json

@@ -62,5 +62,6 @@
     ],
     "host": "localhost",
     "passed": false,
-    "policy": "Hardened OpenSSH Server v8.0 (version 4)"
+    "policy": "Hardened OpenSSH Server v8.0 (version 4)",
+    "port": 2222
 }

test/docker/expected_results/openssh_8.0p1_custom_policy_test11.json

@@ -2,5 +2,6 @@
     "errors": [],
     "host": "localhost",
     "passed": true,
-    "policy": "Docker policy: test11 (version 1)"
+    "policy": "Docker policy: test11 (version 1)",
+    "port": 2222
 }

test/docker/expected_results/openssh_8.0p1_custom_policy_test12.json

@@ -39,5 +39,6 @@
     ],
     "host": "localhost",
     "passed": false,
-    "policy": "Docker policy: test12 (version 1)"
+    "policy": "Docker policy: test12 (version 1)",
+    "port": 2222
 }

test/docker/expected_results/openssh_8.0p1_custom_policy_test13.json

@@ -2,5 +2,6 @@
     "errors": [],
     "host": "localhost",
     "passed": true,
-    "policy": "Docker policy: test13 (version 1)"
+    "policy": "Docker policy: test13 (version 1)",
+    "port": 2222
 }

test/docker/expected_results/openssh_8.0p1_custom_policy_test14.json

@@ -15,5 +15,6 @@
     ],
     "host": "localhost",
     "passed": false,
-    "policy": "Docker policy: test14 (version 1)"
+    "policy": "Docker policy: test14 (version 1)",
+    "port": 2222
 }

test/docker/expected_results/openssh_8.0p1_custom_policy_test15.json

@@ -2,5 +2,6 @@
     "errors": [],
     "host": "localhost",
     "passed": true,
-    "policy": "Docker policy: test15 (version 1)"
+    "policy": "Docker policy: test15 (version 1)",
+    "port": 2222
 }

test/docker/expected_results/openssh_8.0p1_custom_policy_test16.json

@@ -82,5 +82,6 @@
     ],
     "host": "localhost",
     "passed": false,
-    "policy": "Docker policy: test16 (version 1)"
+    "policy": "Docker policy: test16 (version 1)",
+    "port": 2222
 }

test/docker/expected_results/openssh_8.0p1_custom_policy_test17.json

@@ -2,5 +2,6 @@
     "errors": [],
     "host": "localhost",
     "passed": true,
-    "policy": "Docker policy: test17 (version 1)"
+    "policy": "Docker policy: test17 (version 1)",
+    "port": 2222
 }

test/docker/expected_results/openssh_8.0p1_custom_policy_test6.json

@@ -2,5 +2,6 @@
     "errors": [],
     "host": "localhost",
     "passed": true,
-    "policy": "Docker policy: test6 (version 1)"
+    "policy": "Docker policy: test6 (version 1)",
+    "port": 2222
 }