#11772 Correct deferred annotations (#11770)

src/twisted/conch/test/test_tap.py

@@ -5,7 +5,7 @@
 Tests for L{twisted.conch.tap}.
 """
 
-from typing import Any
+from typing import Any, Tuple, Union
 
 from twisted.application.internet import StreamServerEndpointService
 from twisted.cred import error
@@ -131,7 +131,7 @@ def test_authSuccess(self) -> Deferred[None]:
         correct = UsernamePassword(*self.usernamePassword)
         d = checker.requestAvatarId(correct)
 
-        def checkSuccess(username: bytes) -> None:
+        def checkSuccess(username: Union[bytes, Tuple[()]]) -> None:
             self.assertEqual(username, correct.username)
 
         return d.addCallback(checkSuccess)

src/twisted/conch/test/test_unix.py

@@ -6,8 +6,17 @@
 from zope.interface import implementer
 
 from twisted.conch.interfaces import IConchUser
-from twisted.cred.checkers import InMemoryUsernamePasswordDatabaseDontUse
-from twisted.cred.credentials import IUsernamePassword, UsernamePassword
+from twisted.cred.checkers import (
+    AllowAnonymousAccess,
+    InMemoryUsernamePasswordDatabaseDontUse,
+)
+from twisted.cred.credentials import (
+    Anonymous,
+    IAnonymous,
+    IUsernamePassword,
+    UsernamePassword,
+)
+from twisted.cred.error import LoginDenied
 from twisted.cred.portal import Portal
 from twisted.internet.interfaces import IReactorProcess
 from twisted.python.fakepwd import UserDatabase
@@ -123,3 +132,12 @@ def test_unixSSHRealm(self) -> None:
         self.assertIsInstance(avatar, UnixConchUser)
         assert isinstance(avatar, UnixConchUser)  # legibility for mypy
         self.assertEqual(avatar.getHomeDir(), home)
+
+    def test_unixSSHRefusesAnonymousLogins(self) -> None:
+        """
+        L{UnixSSHRealm} will refuse anonymous logins.
+        """
+        p = Portal(UnixSSHRealm(), [AllowAnonymousAccess()])
+        result = p.login(IAnonymous(Anonymous()), None, IConchUser)
+        loginDenied = self.failureResultOf(result)
+        self.assertIsInstance(loginDenied.value, LoginDenied)

src/twisted/conch/unix.py

@@ -35,6 +35,7 @@
     FXF_WRITE,
 )
 from twisted.cred import portal
+from twisted.cred.error import LoginDenied
 from twisted.internet.error import ProcessExitedAlready
 from twisted.internet.interfaces import IListeningPort
 from twisted.logger import Logger
@@ -51,10 +52,12 @@
 class UnixSSHRealm:
     def requestAvatar(
         self,
-        username: bytes,
+        username: bytes | Tuple[()],
         mind: object,
         *interfaces: portal._InterfaceItself,
     ) -> Tuple[portal._InterfaceItself, UnixConchUser, Callable[[], None]]:
+        if not isinstance(username, bytes):
+            raise LoginDenied("UNIX SSH realm does not authorize anonymous sessions.")
         user = UnixConchUser(username.decode())
         return interfaces[0], user, user.logout
 

src/twisted/cred/portal.py

@@ -39,7 +39,7 @@ class IRealm(Interface):
     """
 
     def requestAvatar(
-        avatarId: bytes, mind: object, *interfaces: _InterfaceItself
+        avatarId: Union[bytes, Tuple[()]], mind: object, *interfaces: _InterfaceItself
     ) -> Union[Deferred[_requestResult], _requestResult]:
         """
         Return avatar which provides one of the given interfaces.

src/twisted/internet/_resolver.py

@@ -49,7 +49,6 @@
 from twisted.internet.threads import deferToThreadPool
 from twisted.logger import Logger
 from twisted.python.compat import nativeString
-from twisted.python.failure import Failure
 
 if TYPE_CHECKING:
     from twisted.python.threadpool import ThreadPool
@@ -247,26 +246,25 @@ def resolveHostName(
 
         resolution = HostResolution(hostName)
         resolutionReceiver.resolutionBegan(resolution)
-        onAddress = self._simpleResolver.getHostByName(hostName)
-
-        def addressReceived(address: str) -> None:
-            resolutionReceiver.addressResolved(IPv4Address("TCP", address, portNumber))
-
-        def errorReceived(error: Failure) -> None:
-            if not error.check(DNSLookupError):
-                self._log.failure(
+        (
+            self._simpleResolver.getHostByName(hostName)
+            .addCallback(
+                lambda address: resolutionReceiver.addressResolved(
+                    IPv4Address("TCP", address, portNumber)
+                )
+            )
+            .addErrback(
+                lambda error: None
+                if error.check(DNSLookupError)
+                else self._log.failure(
                     "while looking up {name} with {resolver}",
                     error,
                     name=hostName,
                     resolver=self._simpleResolver,
                 )
-
-        onAddress.addCallbacks(addressReceived, errorReceived)
-
-        def finish(result: None) -> None:
-            resolutionReceiver.resolutionComplete()
-
-        onAddress.addCallback(finish)
+            )
+            .addCallback(lambda nothing: resolutionReceiver.resolutionComplete())
+        )
         return resolution