Fix Session.peer type annotation to support IPv6 and UNIX sockets

PR #522

Fixes #518

Author: cuu508

aiosmtpd/handlers.py

@@ -27,6 +27,7 @@
 from aiosmtpd import _get_or_new_eventloop
 from aiosmtpd.smtp import SMTP as SMTPServer
 from aiosmtpd.smtp import Envelope as SMTPEnvelope
+from aiosmtpd.smtp import PeerType
 from aiosmtpd.smtp import Session as SMTPSession
 
 T = TypeVar("T")
@@ -38,7 +39,7 @@
 log = logging.getLogger("mail.debug")
 
 
-def _format_peer(peer: str) -> str:
+def _format_peer(peer: PeerType) -> str:
     # This is a separate function mostly so the test suite can craft a
     # reproducible output.
     return "X-Peer: {!r}".format(peer)
@@ -76,7 +77,7 @@ def from_cli(cls: Type[T], parser: ArgumentParser, *args) -> T:
             parser.error("Debugging usage: [stdout|stderr]")
         return cls(stream)  # type: ignore[call-arg]
 
-    def _print_message_content(self, peer: str, data: Union[str, bytes]) -> None:
+    def _print_message_content(self, peer: PeerType, data: Union[str, bytes]) -> None:
         in_headers = True
         for line in data.splitlines():
             # Dump the RFC 2822 headers first.

aiosmtpd/smtp.py

@@ -65,6 +65,8 @@ class _DataState(enum.Enum):
 RT = TypeVar("RT")  # "ReturnType"
 DecoratorType = Callable[[Callable[..., RT]], Callable[..., RT]]
 
+# IPv4, IPv6, UNIX socket, not set
+PeerType = Union[tuple[str, int], tuple[str, int, int, int], str, None]
 
 # endregion
 
@@ -155,7 +157,7 @@ def __repr__(self) -> str:
 @public
 class Session:
     def __init__(self, loop: asyncio.AbstractEventLoop):
-        self.peer: Optional[str] = None
+        self.peer: PeerType = None
         self.ssl: Optional[dict[str, Any]] = None
         self.host_name: Optional[str] = None
         self.extended_smtp = False