Handle reverse DNS failure in host-based authentication

This commit improves the handling of host lookups in host-based
authentication. If reverse DNS fails when trying to resolve the
hostname, the IP address will be used instead. Known host checking
was already matching either the host or the address before this, but
the reverse DNS lookup error caused the authentication to fail
before it got that far.

A similar change was made here on host name resolution for host-based
auth on the server side.

Thanks go to GitHub user xBiggs for suggesting this change.

asyncssh/connection.py

@@ -3565,11 +3565,14 @@ async def host_based_auth_requested(self) -> \
         client_host = self._options.client_host
 
         if client_host is None:
-            sockname = self.get_extra_info('sockname')
+            sockname = cast(SockAddr, self.get_extra_info('sockname'))
 
             if sockname:
-                client_host, _ = await self._loop.getnameinfo(
-                    cast(SockAddr, sockname), socket.NI_NUMERICSERV)
+                try:
+                    client_host, _ = await self._loop.getnameinfo(
+                        sockname, socket.NI_NUMERICSERV)
+                except socket.gaierror:
+                    client_host = sockname[0]
             else:
                 client_host = ''
 
@@ -5875,9 +5878,13 @@ async def validate_host_based_auth(self, username: str, key_data: bytes,
         if self._trust_client_host:
             resolved_host = client_host
         else:
-            resolved_host, _ = await self._loop.getnameinfo(
-                cast(SockAddr, self.get_extra_info('peername')),
-                socket.NI_NUMERICSERV)
+            peername = cast(SockAddr, self.get_extra_info('peername'))
+
+            try:
+                resolved_host, _ = await self._loop.getnameinfo(
+                    peername, socket.NI_NUMERICSERV)
+            except socket.gaierror:
+                resolved_host = peername[0]
 
             if resolved_host != client_host:
                 self.logger.info('Client host mismatch: received %s, '
@@ -5931,11 +5938,10 @@ async def _validate_openssh_certificate(
 
         self._key_options = options
 
-        if self.get_key_option('principals'):
-            username = ''
+        cert_user = None if self.get_key_option('principals') else username
 
         try:
-            cert.validate(CERT_TYPE_USER, username)
+            cert.validate(CERT_TYPE_USER, cert_user)
         except ValueError:
             return None
 

asyncssh/public_key.py

@@ -1767,7 +1767,7 @@ def _decode_options(options: bytes, decoders: _OpenSSHCertDecoders,
 
         return result
 
-    def validate(self, cert_type: int, principal: str) -> None:
+    def validate(self, cert_type: int, principal: Optional[str]) -> None:
         """Validate an OpenSSH certificate"""
 
         if self._cert_type != cert_type:
@@ -1781,7 +1781,8 @@ def validate(self, cert_type: int, principal: str) -> None:
         if now >= self._valid_before:
             raise ValueError('Certificate expired')
 
-        if principal and self.principals and principal not in self.principals:
+        if principal is not None and self.principals and \
+                principal not in self.principals:
             raise ValueError('Certificate principal mismatch')
 
 

tests/test_connection_auth.py

@@ -36,7 +36,8 @@
 
 from .keysign_stub import create_subprocess_exec_stub
 from .server import Server, ServerTestCase
-from .util import asynctest, gss_available, patch_getnameinfo, patch_gss
+from .util import asynctest, gss_available, patch_getnameinfo
+from .util import patch_getnameinfo_error, patch_gss
 from .util import make_certificate, nc_available, x509_available
 
 
@@ -860,6 +861,26 @@ async def test_disabled_trivial_client_host_auth(self):
                                disable_trivial_auth=True)
 
 
+class _TestHostBasedAuthNoRDNS(ServerTestCase):
+    """Unit tests for host-based authentication with no reverse DNS"""
+
+    @classmethod
+    async def start_server(cls):
+        """Start an SSH server which supports host-based authentication"""
+
+        return await cls.create_server(
+            _HostBasedServer, known_client_hosts='known_hosts')
+
+    @patch_getnameinfo_error
+    @asynctest
+    async def test_client_host_auth_no_rdns(self):
+        """Test connecting with host-based authentication with no RDNS"""
+
+        async with self.connect(username='user', client_host_keys='skey',
+                                client_username='user'):
+            pass
+
+
 @patch_getnameinfo
 class _TestCallbackHostBasedAuth(ServerTestCase):
     """Unit tests for host-based authentication using callback"""

tests/util.py

@@ -25,6 +25,7 @@
 import functools
 import os
 import shutil
+import socket
 import subprocess
 import sys
 import tempfile
@@ -106,6 +107,19 @@ def getnameinfo(sockaddr, flags):
     return patch('socket.getnameinfo', getnameinfo)(cls)
 
 
+def patch_getnameinfo_error(cls):
+    """Decorator for patching socket.getnameinfo to raise an error"""
+
+    def getnameinfo_error(sockaddr, flags):
+        """Mock failure of reverse DNS lookup of client address"""
+
+        # pylint: disable=unused-argument
+
+        raise socket.gaierror()
+
+    return patch('socket.getnameinfo', getnameinfo_error)(cls)
+
+
 def patch_extra_kex(cls):
     """Decorator for skipping extra kex algs"""