Repository: agronholm/anyio

Issue:
Should we implement an asyncio-like get_extra_info() method on streams?
We have stream wrappers such as `TLSStream` and `BufferedByteReceiveStream` which expose a different set of attributes than the streams they wrap. I imagine it will be quite inconvenient for developers that just want the remote address (for example) to manually traverse the chain of wrapped transports to find the information they want. Asyncio has a pretty convenient solution to this: [get_extra_info()](https://docs.python.org/3/library/asyncio-protocol.html?#asyncio.BaseTransport.get_extra_info). I think it would be really useful provide a similar method so developers don't have to care how many layers down the information resides.

Discussion:

Patch:
diff --git a/docs/api.rst b/docs/api.rst
index f994e96..f446dac 100644
--- a/docs/api.rst
+++ b/docs/api.rst
@@ -17,6 +17,13 @@ Asynchronous resources
 
 .. autoclass:: anyio.abc.AsyncResource
 
+Typed attributes
+----------------
+
+.. autofunction:: anyio.typed_attribute
+.. autoclass:: anyio.TypedAttributeSet
+.. autoclass:: anyio.TypedAttributeProvider
+
 Timeouts and cancellation
 -------------------------
 
@@ -85,6 +92,7 @@ Streams and stream wrappers
 .. autoclass:: anyio.streams.text.TextReceiveStream
 .. autoclass:: anyio.streams.text.TextSendStream
 .. autoclass:: anyio.streams.text.TextStream
+.. autoclass:: anyio.streams.tls.TLSAttribute
 .. autoclass:: anyio.streams.tls.TLSStream
 .. autoclass:: anyio.streams.tls.TLSListener
 
@@ -103,7 +111,7 @@ Sockets and networking
 .. autofunction:: anyio.wait_socket_readable
 .. autofunction:: anyio.wait_socket_writable
 
-.. autoclass:: anyio.abc.SocketProvider()
+.. autoclass:: anyio.abc.SocketAttribute
 .. autoclass:: anyio.abc.SocketStream()
 .. autoclass:: anyio.abc.SocketListener()
 .. autoclass:: anyio.abc.UDPSocket()
diff --git a/docs/index.rst b/docs/index.rst
index 5d08fad..e9e59e4 100644
--- a/docs/index.rst
+++ b/docs/index.rst
@@ -14,6 +14,7 @@ The manual
    cancellation
    synchronization
    streams
+   typedattrs
    networking
    threads
    subprocesses
diff --git a/docs/typedattrs.rst b/docs/typedattrs.rst
new file mode 100644
index 0000000..96923b6
--- /dev/null
+++ b/docs/typedattrs.rst
@@ -0,0 +1,63 @@
+Using typed attributes
+======================
+
+On AnyIO, streams and listeners can be layered on top of each other to provide extra functionality.
+But when you want to look up information from one of the layers down below, you might have to
+traverse the entire chain to find what you're looking for, which is highly inconvenient. To address
+this, AnyIO has a system of *typed attributes* where you can look for a specific attribute by its
+unique key. If a stream or listener wrapper does not have the attribute you're looking for, it will
+look it up in the wrapped instance, and that wrapper can look in its wrapped instance and so on,
+until the attribute is either found or the end of the chain is reached. This also lets wrappers
+override attributes from the wrapped objects when necessary.
+
+A common use case is finding the IP address of the remote side of a TCP connection when the
+stream may be either :class:`~anyio.abc.SocketStream` or :class:`~anyio.streams.tls.TLSStream`::
+
+    from anyio import connect_tcp, connect_tcp_with_tls
+
+
+    async def connect(host, port, tls: bool):
+        if tls:
+            stream = await connect_tcp(host, port)
+        else:
+            stream = await connect_tcp_with_tls(host, port)
+
+        print('Connected to', stream.extra(SocketAttribute.remote_address))
+
+Each typed attribute provider class should document the set of attributes it provides on its own.
+
+Defining your own typed attributes
+----------------------------------
+
+By convention, typed attributes are stored together in a container class with other attributes of
+the same category::
+
+    from anyio import TypedAttribute, TypedAttributeSet
+
+
+    class MyTypedAttribute:
+        string_valued_attribute = TypedAttribute[str]()
+        some_float_attribute = TypedAttribute[float]()
+
+To provide values for these attributes, implement the
+:meth:`~anyio.TypedAttributeProvider.extra_attributes` property in your class::
+
+    from anyio import TypedAttributeProvider
+
+
+    class MyAttributeProvider(TypedAttributeProvider):
+        def extra_attributes():
+            return {
+                MyTypedAttribute.string_valued_attribute: lambda: 'my attribute value',
+                MyTypedAttribute.some_float_attribute: lambda: 6.492
+            }
+
+If your class inherits from another typed attribute provider, make sure you include its attributes
+in the return value::
+
+    class AnotherAttributeProvider(MyAttributeProvider):
+        def extra_attributes():
+            return {
+                **super().extra_attributes,
+                MyTypedAttribute.string_valued_attribute: lambda: 'overridden attribute value'
+            }
diff --git a/docs/versionhistory.rst b/docs/versionhistory.rst
index e71132a..8514512 100644
--- a/docs/versionhistory.rst
+++ b/docs/versionhistory.rst
@@ -5,6 +5,9 @@ This library adheres to `Semantic Versioning 2.0 <http://semver.org/>`_.
 
 **UNRELEASED**
 
+- **BACKWARDS INCOMPATIBLE** Replaced individual stream/listener attributes with a system of
+  typed extra attributes which traverses the entire chain of wrapped streams/listeners to find the
+  attribute being looked for
 - Added the ``tls_standard_compatible`` argument to ``connect_tcp_with_tls()`` (this was mistakenly
   omitted when ``connect_tcp()`` was split in two)
 - Fixed Hypothesis support in the pytest plugin (it was not actually running the Hypothesis tests
diff --git a/src/anyio/__init__.py b/src/anyio/__init__.py
index d21e90d..21e8338 100644
--- a/src/anyio/__init__.py
+++ b/src/anyio/__init__.py
@@ -11,7 +11,8 @@ __all__ = ('run', 'sleep', 'current_time', 'get_all_backends', 'get_cancelled_ex
            'get_current_task', 'get_running_tasks', 'wait_all_tasks_blocked',
            'run_sync_in_worker_thread', 'run_async_from_thread',
            'current_default_worker_thread_limiter', 'create_blocking_portal',
-           'start_blocking_portal')
+           'start_blocking_portal', 'typed_attribute', 'TypedAttributeSet',
+           'TypedAttributeProvider')
 
 from ._core._eventloop import current_time, get_all_backends, get_cancelled_exc_class, run, sleep
 from ._core._exceptions import (
@@ -34,6 +35,7 @@ from ._core._testing import TaskInfo, get_current_task, get_running_tasks, wait_
 from ._core._threads import (
     create_blocking_portal, current_default_worker_thread_limiter, run_async_from_thread,
     run_sync_in_worker_thread, start_blocking_portal)
+from ._core._typedattr import TypedAttributeProvider, TypedAttributeSet, typed_attribute
 
 # Re-export imports so they look like they live directly in this package
 for key, value in list(locals().items()):
diff --git a/src/anyio/_backends/_asyncio.py b/src/anyio/_backends/_asyncio.py
index 73f1437..a20f35b 100644
--- a/src/anyio/_backends/_asyncio.py
+++ b/src/anyio/_backends/_asyncio.py
@@ -724,6 +724,10 @@ class SocketStream(abc.SocketStream):
         self._send_guard = ResourceGuard('writing to')
         self._closed = False
 
+    @property
+    def _raw_socket(self) -> socket.socket:
+        return self._transport.get_extra_info('socket')
+
     async def receive(self, max_bytes: int = 65536) -> bytes:
         with self._receive_guard:
             await checkpoint()
@@ -783,20 +787,16 @@ class SocketStream(abc.SocketStream):
             await asyncio.sleep(0)
             self._transport.abort()
 
-    @property
-    def raw_socket(self) -> socket.socket:
-        return self._transport.get_extra_info('socket')
-
 
 class SocketListener(abc.SocketListener):
     def __init__(self, raw_socket: socket.SocketType):
-        self._raw_socket = raw_socket
+        self.__raw_socket = raw_socket
         self._loop = cast(asyncio.BaseEventLoop, get_running_loop())
         self._accept_guard = ResourceGuard('accepting connections from')
 
     @property
-    def raw_socket(self) -> socket.socket:
-        return self._raw_socket
+    def _raw_socket(self) -> socket.socket:
+        return self.__raw_socket
 
     async def accept(self) -> abc.SocketStream:
         with self._accept_guard:
@@ -830,15 +830,15 @@ class UDPSocket(abc.UDPSocket):
         self._send_guard = ResourceGuard('writing to')
         self._closed = False
 
+    @property
+    def _raw_socket(self) -> SocketType:
+        return self._transport.get_extra_info('socket')
+
     async def aclose(self) -> None:
         if not self._transport.is_closing():
             self._closed = True
             self._transport.close()
 
-    @property
-    def raw_socket(self) -> socket.socket:
-        return self._transport.get_extra_info('socket')
-
     async def receive(self) -> Tuple[bytes, IPSockAddrType]:
         with self._receive_guard:
             await checkpoint()
@@ -876,15 +876,15 @@ class ConnectedUDPSocket(abc.ConnectedUDPSocket):
         self._send_guard = ResourceGuard('writing to')
         self._closed = False
 
+    @property
+    def _raw_socket(self) -> SocketType:
+        return self._transport.get_extra_info('socket')
+
     async def aclose(self) -> None:
         if not self._transport.is_closing():
             self._closed = True
             self._transport.close()
 
-    @property
-    def raw_socket(self) -> SocketType:
-        return self._transport.get_extra_info('socket')
-
     async def receive(self) -> bytes:
         with self._receive_guard:
             await checkpoint()
diff --git a/src/anyio/_backends/_curio.py b/src/anyio/_backends/_curio.py
index f38d041..52d2238 100644
--- a/src/anyio/_backends/_curio.py
+++ b/src/anyio/_backends/_curio.py
@@ -578,7 +578,7 @@ class _CurioSocketMixin(Generic[T_SockAddr]):
         self._closed = False
 
     @property
-    def raw_socket(self) -> socket.socket:
+    def _raw_socket(self) -> socket.socket:
         return self._curio_socket._socket
 
     async def aclose(self) -> None:
diff --git a/src/anyio/_backends/_trio.py b/src/anyio/_backends/_trio.py
index bfa1464..3ab7d13 100644
--- a/src/anyio/_backends/_trio.py
+++ b/src/anyio/_backends/_trio.py
@@ -21,9 +21,9 @@ from .._core._synchronization import ResourceGuard
 from ..abc.sockets import IPSockAddrType, UDPPacketType
 
 try:
-    import trio.lowlevel as trio_lowlevel
+    from trio import lowlevel as trio_lowlevel
 except ImportError:
-    import trio.hazmat as trio_lowlevel
+    from trio import hazmat as trio_lowlevel
     from trio.hazmat import wait_readable, wait_writable
 else:
     from trio.lowlevel import wait_readable, wait_writable
@@ -296,7 +296,7 @@ class _TrioSocketMixin(Generic[T_SockAddr]):
             raise BrokenResourceError
 
     @property
-    def raw_socket(self) -> socket.socket:
+    def _raw_socket(self) -> socket.socket:
         return self._trio_socket._sock
 
     async def aclose(self) -> None:
diff --git a/src/anyio/_core/_exceptions.py b/src/anyio/_core/_exceptions.py
index 261d3aa..735f32e 100644
--- a/src/anyio/_core/_exceptions.py
+++ b/src/anyio/_core/_exceptions.py
@@ -63,5 +63,12 @@ class IncompleteRead(Exception):
         super().__init__('The stream was closed before the read operation could be completed')
 
 
+class TypedAttributeLookupError(LookupError):
+    """
+    Raised by :meth:`~anyio.TypedAttributeProvider.extra` when the given typed attribute is not
+    found and no default value has been given.
+    """
+
+
 class WouldBlock(Exception):
     """Raised by ``X_nowait`` functions if ``X()`` would block."""
diff --git a/src/anyio/_core/_typedattr.py b/src/anyio/_core/_typedattr.py
new file mode 100644
index 0000000..4901e8f
--- /dev/null
+++ b/src/anyio/_core/_typedattr.py
@@ -0,0 +1,78 @@
+import sys
+from typing import Callable, Mapping, TypeVar, Union, overload
+
+from .._core._exceptions import TypedAttributeLookupError
+
+if sys.version_info >= (3, 8):
+    from typing import final
+else:
+    from typing_extensions import final
+
+T_Attr = TypeVar('T_Attr')
+T_Default = TypeVar('T_Default')
+undefined = object()
+
+
+def typed_attribute():
+    """Return a unique object, used to mark typed attributes."""
+    return object()
+
+
+class TypedAttributeSet:
+    """
+    Superclass for typed attribute collections.
+
+    Checks that every public attribute of every subclass has a type annotation.
+    """
+
+    def __init_subclass__(cls) -> None:
+        annotations = getattr(cls, '__annotations__', {})
+        for attrname in dir(cls):
+            if not attrname.startswith('_') and attrname not in annotations:
+                raise TypeError(f'Attribute {attrname!r} is missing its type annotation')
+
+        super().__init_subclass__()
+
+
+class TypedAttributeProvider:
+    """Base class for classes that wish to provide typed extra attributes."""
+
+    @property
+    def extra_attributes(self) -> Mapping[T_Attr, Callable[[], T_Attr]]:
+        """
+        A mapping of the extra attributes to callables that return the corresponding values.
+
+        If the provider wraps another provider, the attributes from that wrapper should also be
+        included in the returned mapping (but the wrapper may override the callables from the
+        wrapped instance).
+
+        """
+        return {}
+
+    @overload
+    def extra(self, attribute: T_Attr) -> T_Attr:
+        ...
+
+    @overload
+    def extra(self, attribute: T_Attr, default: T_Default) -> Union[T_Attr, T_Default]:
+        ...
+
+    @final
+    def extra(self, attribute, default=undefined):
+        """
+        extra(attribute, default=undefined)
+
+        Return the value of the given typed extra attribute.
+
+        :param TypedAttribute attribute: the attribute to look for
+        :param default: the value that should be returned if no value is found for the attribute
+        :raises TypedAttributeLookupError: if the search failed and no default value was given
+
+        """
+        try:
+            return self.extra_attributes[attribute]()
+        except KeyError:
+            if default is undefined:
+                raise TypedAttributeLookupError('Attribute not found') from None
+            else:
+                return default
diff --git a/src/anyio/abc/__init__.py b/src/anyio/abc/__init__.py
index 3e6edba..ba53e51 100644
--- a/src/anyio/abc/__init__.py
+++ b/src/anyio/abc/__init__.py
@@ -1,4 +1,4 @@
-__all__ = ('AsyncResource', 'SocketProvider', 'SocketStream', 'SocketListener', 'UDPSocket',
+__all__ = ('AsyncResource', 'SocketAttribute', 'SocketStream', 'SocketListener', 'UDPSocket',
            'ConnectedUDPSocket', 'UnreliableObjectReceiveStream', 'UnreliableObjectSendStream',
            'UnreliableObjectStream', 'ObjectReceiveStream', 'ObjectSendStream', 'ObjectStream',
            'ByteReceiveStream', 'ByteSendStream', 'ByteStream', 'AnyUnreliableByteReceiveStream',
@@ -8,7 +8,7 @@ __all__ = ('AsyncResource', 'SocketProvider', 'SocketStream', 'SocketListener',
            'BlockingPortal')
 
 from .resources import AsyncResource
-from .sockets import ConnectedUDPSocket, SocketListener, SocketProvider, SocketStream, UDPSocket
+from .sockets import ConnectedUDPSocket, SocketAttribute, SocketListener, SocketStream, UDPSocket
 from .streams import (
     AnyByteReceiveStream, AnyByteSendStream, AnyByteStream, AnyUnreliableByteReceiveStream,
     AnyUnreliableByteSendStream, AnyUnreliableByteStream, ByteReceiveStream, ByteSendStream,
diff --git a/src/anyio/abc/sockets.py b/src/anyio/abc/sockets.py
index 7f01b07..77c88c9 100644
--- a/src/anyio/abc/sockets.py
+++ b/src/anyio/abc/sockets.py
@@ -3,6 +3,7 @@ from ipaddress import IPv4Address, IPv6Address
 from socket import AddressFamily, SocketType
 from typing import Any, AsyncContextManager, Callable, Generic, Optional, Tuple, TypeVar, Union
 
+from .._core._typedattr import TypedAttributeProvider, TypedAttributeSet, typed_attribute
 from .streams import ByteStream, Listener, T_Stream, UnreliableObjectStream
 from .tasks import TaskGroup
 
@@ -22,74 +23,69 @@ class _NullAsyncContextManager:
         pass
 
 
-class SocketProvider(Generic[T_SockAddr]):
-    """Abstract base class for socket-based streams and listeners."""
+class SocketAttribute(TypedAttributeSet):
+    #: the address family of the underlying socket
+    family: AddressFamily = typed_attribute()
+    #: the local socket address of the underlying socket
+    local_address: SockAddrType = typed_attribute()
+    #: for IP addresses, the local port the underlying socket is bound to
+    local_port: int = typed_attribute()
+    #: the underlying stdlib socket object
+    raw_socket: SocketType = typed_attribute()
+    #: the remote address the underlying socket is connected to
+    remote_address: SockAddrType = typed_attribute()
+    #: for IP addresses, the remote port the underlying socket is connected to
+    remote_port: int = typed_attribute()
 
-    @property
-    @abstractmethod
-    def raw_socket(self) -> SocketType:
-        """
-        The underlying raw socket object.
-
-        .. warning:: This should only be used for advanced use cases, so only use this if you know
-            what you're doing. Otherwise you might break things in subtle ways.
-        """
-
-    def getsockopt(self, level, optname, *args):
-        """
-        Get a socket option from the underlying socket.
-
-        :return: the return value of :meth:`~socket.socket.getsockopt`
-
-        """
-        return self.raw_socket.getsockopt(level, optname, *args)
-
-    def setsockopt(self, level, optname, value, *args) -> None:
-        """
-        Set a socket option on the underlying socket.
-
-        This calls :meth:`~socket.socket.setsockopt` on the underlying socket.
-
-        """
-        self.raw_socket.setsockopt(level, optname, value, *args)
 
+class _SocketProvider(Generic[T_SockAddr], TypedAttributeProvider):
     @property
-    def family(self) -> AddressFamily:
-        """The address family of the underlying socket."""
-        return self.raw_socket.family
+    def extra_attributes(self):
+        from .._core._sockets import convert_ipv6_sockaddr as convert
+
+        attributes = {
+            SocketAttribute.family: lambda: self._raw_socket.family,
+            SocketAttribute.local_address: lambda: convert(self._raw_socket.getsockname()),
+            SocketAttribute.raw_socket: lambda: self._raw_socket
+        }
+        try:
+            peername = convert(self._raw_socket.getpeername())
+        except OSError:
+            peername = None
+
+        # Provide the remote address for connected sockets
+        if peername is not None:
+            attributes[SocketAttribute.remote_address] = lambda: peername
+
+        # Provide local and remote ports for IP based sockets
+        if self._raw_socket.family in (AddressFamily.AF_INET, AddressFamily.AF_INET6):
+            attributes[SocketAttribute.local_port] = lambda: self._raw_socket.getsockname()[1]
+            if peername is not None:
+                attributes[SocketAttribute.remote_port] = lambda: peername[1]
+
+        return attributes
 
     @property
-    def local_address(self) -> T_SockAddr:
-        """
-        The bound address of the underlying local socket.
-
-        For TCP streams, this is a tuple of (IP address, port).
-        For UNIX socket streams, this is the path to the socket.
-
-        """
-        from anyio._core._sockets import convert_ipv6_sockaddr
-        return convert_ipv6_sockaddr(self.raw_socket.getsockname())
-
+    @abstractmethod
+    def _raw_socket(self) -> SocketType:
+        pass
 
-class SocketStream(Generic[T_SockAddr], ByteStream, SocketProvider[T_SockAddr]):
-    """Transports bytes over a socket."""
 
-    @property
-    def remote_address(self) -> T_SockAddr:
-        """
-        The address this socket is connected to.
+class SocketStream(Generic[T_SockAddr], ByteStream, _SocketProvider[T_SockAddr]):
+    """
+    Transports bytes over a socket.
 
-        For TCP streams, this is a tuple of (IP address, port).
-        For UNIX socket streams, this is the path to the socket.
-
-        """
-        from anyio._core._sockets import convert_ipv6_sockaddr
-        return convert_ipv6_sockaddr(self.raw_socket.getpeername())
+    Supports all relevant extra attributes from :class:`~SocketAttribute`.
+    """
 
 
 class SocketListener(Generic[T_SockAddr], Listener[SocketStream[T_SockAddr]],
-                     SocketProvider[T_SockAddr]):
-    """Listens to incoming socket connections."""
+                     _SocketProvider[T_SockAddr]):
+    """
+    Listens to incoming socket connections.
+
+    Supports all relevant extra attributes from :class:`~SocketAttribute`.
+    """
 
     @abstractmethod
     async def accept(self) -> SocketStream[T_SockAddr]:
@@ -113,18 +109,20 @@ class SocketListener(Generic[T_SockAddr], Listener[SocketStream[T_SockAddr]],
                 await task_group.spawn(handler, stream)
 
 
-class UDPSocket(UnreliableObjectStream[UDPPacketType], SocketProvider[IPSockAddrType]):
-    """Represents an unconnected UDP socket."""
+class UDPSocket(UnreliableObjectStream[UDPPacketType], _SocketProvider[IPSockAddrType]):
+    """
+    Represents an unconnected UDP socket.
+
+    Supports all relevant extra attributes from :class:`~SocketAttribute`.
+    """
 
     async def sendto(self, data: bytes, host: str, port: int) -> None:
         return await self.send((data, (host, port)))
 
 
-class ConnectedUDPSocket(UnreliableObjectStream[bytes], SocketProvider[IPSockAddrType]):
-    """Represents an connected UDP socket."""
+class ConnectedUDPSocket(UnreliableObjectStream[bytes], _SocketProvider[IPSockAddrType]):
+    """
+    Represents an connected UDP socket.
 
-    @property
-    def remote_address(self) -> IPSockAddrType:
-        """The address this socket is connected to."""
-        from anyio._core._sockets import convert_ipv6_sockaddr
-        return convert_ipv6_sockaddr(self.raw_socket.getpeername())
+    Supports all relevant extra attributes from :class:`~SocketAttribute`.
+    """
diff --git a/src/anyio/abc/streams.py b/src/anyio/abc/streams.py
index 1a11ea5..f06fb70 100644
--- a/src/anyio/abc/streams.py
+++ b/src/anyio/abc/streams.py
@@ -2,6 +2,7 @@ from abc import abstractmethod
 from typing import Any, Callable, Generic, Optional, TypeVar, Union
 
 from .._core._exceptions import EndOfStream
+from .._core._typedattr import TypedAttributeProvider
 from .resources import AsyncResource
 from .tasks import TaskGroup
 
@@ -9,7 +10,7 @@ T_Item = TypeVar('T_Item')
 T_Stream = TypeVar('T_Stream')
 
 
-class UnreliableObjectReceiveStream(Generic[T_Item], AsyncResource):
+class UnreliableObjectReceiveStream(Generic[T_Item], AsyncResource, TypedAttributeProvider):
     """
     An interface for receiving objects.
 
@@ -42,7 +43,7 @@ class UnreliableObjectReceiveStream(Generic[T_Item], AsyncResource):
         """
 
 
-class UnreliableObjectSendStream(Generic[T_Item], AsyncResource):
+class UnreliableObjectSendStream(Generic[T_Item], AsyncResource, TypedAttributeProvider):
     """
     An interface for sending objects.
 
@@ -101,7 +102,7 @@ class ObjectStream(Generic[T_Item], ObjectReceiveStream[T_Item], ObjectSendStrea
         """
 
 
-class ByteReceiveStream(AsyncResource):
+class ByteReceiveStream(AsyncResource, TypedAttributeProvider):
     """
     An interface for receiving bytes from a single peer.
 
@@ -129,7 +130,7 @@ class ByteReceiveStream(AsyncResource):
         """
 
 
-class ByteSendStream(AsyncResource):
+class ByteSendStream(AsyncResource, TypedAttributeProvider):
     """An interface for sending bytes to a single peer."""
 
     @abstractmethod
@@ -168,7 +169,7 @@ AnyByteSendStream = Union[ObjectSendStream[bytes], ByteSendStream]
 AnyByteStream = Union[ObjectStream[bytes], ByteStream]
 
 
-class Listener(Generic[T_Stream], AsyncResource):
+class Listener(Generic[T_Stream], AsyncResource, TypedAttributeProvider):
     """An interface for objects that let you accept incoming connections."""
 
     @abstractmethod
diff --git a/src/anyio/streams/buffered.py b/src/anyio/streams/buffered.py
index 6d9c2e0..5c132cb 100644
--- a/src/anyio/streams/buffered.py
+++ b/src/anyio/streams/buffered.py
@@ -24,6 +24,10 @@ class BufferedByteReceiveStream(ByteReceiveStream):
         """The bytes currently in the buffer."""
         return bytes(self._buffer)
 
+    @property
+    def extra_attributes(self):
+        return self.receive_stream.extra_attributes
+
     async def receive(self, max_bytes: int = 65536) -> bytes:
         if self._closed:
             raise ClosedResourceError
diff --git a/src/anyio/streams/stapled.py b/src/anyio/streams/stapled.py
index e657308..c344d49 100644
--- a/src/anyio/streams/stapled.py
+++ b/src/anyio/streams/stapled.py
@@ -14,6 +14,9 @@ class StapledByteStream(ByteStream):
     """
     Combines two byte streams into a single, bidirectional byte stream.
 
+    Extra attributes will be provided from both streams, with the receive stream providing the
+    values in case of a conflict.
+
     :param ByteSendStream send_stream: the sending byte stream
     :param ByteReceiveStream receive_stream: the receiving byte stream
     """
@@ -34,12 +37,19 @@ class StapledByteStream(ByteStream):
         await self.send_stream.aclose()
         await self.receive_stream.aclose()
 
+    @property
+    def extra_attributes(self):
+        return dict(**self.send_stream.extra_attributes, **self.receive_stream.extra_attributes)
+
 
 @dataclass
 class StapledObjectStream(Generic[T_Item], ObjectStream[T_Item]):
     """
     Combines two object streams into a single, bidirectional object stream.
 
+    Extra attributes will be provided from both streams, with the receive stream providing the
+    values in case of a conflict.
+
     :param ObjectSendStream send_stream: the sending object stream
     :param ObjectReceiveStream receive_stream: the receiving object stream
     """
@@ -60,6 +70,10 @@ class StapledObjectStream(Generic[T_Item], ObjectStream[T_Item]):
         await self.send_stream.aclose()
         await self.receive_stream.aclose()
 
+    @property
+    def extra_attributes(self):
+        return dict(**self.send_stream.extra_attributes, **self.receive_stream.extra_attributes)
+
 
 @dataclass
 class MultiListener(Generic[T_Stream], Listener[T_Stream]):
@@ -69,6 +83,9 @@ class MultiListener(Generic[T_Stream], Listener[T_Stream]):
     Any MultiListeners in the given collection of listeners will have their listeners moved into
     this one.
 
+    Extra attributes are provided from each listener, with each successive listener overriding any
+    conflicting attributes from the previous one.
+
     :param listeners: listeners to serve
     :type listeners: Sequence[Listener[T_Stream]]
     """
@@ -98,3 +115,11 @@ class MultiListener(Generic[T_Stream], Listener[T_Stream]):
     async def aclose(self) -> None:
         for listener in self.listeners:
             await listener.aclose()
+
+    @property
+    def extra_attributes(self):
+        attributes = {}
+        for listener in self.listeners:
+            attributes.update(listener)
+
+        return attributes
diff --git a/src/anyio/streams/text.py b/src/anyio/streams/text.py
index 2b2079b..6c4c398 100644
--- a/src/anyio/streams/text.py
+++ b/src/anyio/streams/text.py
@@ -44,6 +44,10 @@ class TextReceiveStream(ObjectReceiveStream[str]):
         await self.transport_stream.aclose()
         self._decoder.reset()
 
+    @property
+    def extra_attributes(self):
+        return self.transport_stream.extra_attributes
+
 
 @dataclass
 class TextSendStream(ObjectSendStream[str]):
@@ -74,6 +78,10 @@ class TextSendStream(ObjectSendStream[str]):
     async def aclose(self) -> None:
         await self.transport_stream.aclose()
 
+    @property
+    def extra_attributes(self):
+        return self.transport_stream.extra_attributes
+
 
 @dataclass
 class TextStream(ObjectStream[str]):
@@ -81,6 +89,9 @@ class TextStream(ObjectStream[str]):
     A bidirectional stream that decodes bytes to strings on receive and encodes strings to bytes on
     send.
 
+    Extra attributes will be provided from both streams, with the receive stream providing the
+    values in case of a conflict.
+
     :param AnyByteStream transport_stream: any bytes-based stream
     :param str encoding: character encoding to use for encoding/decoding strings to/from bytes
         (defaults to ``utf-8``)
@@ -113,3 +124,7 @@ class TextStream(ObjectStream[str]):
     async def aclose(self) -> None:
         await self._send_stream.aclose()
         await self._receive_stream.aclose()
+
+    @property
+    def extra_attributes(self):
+        return dict(**self.send_stream.extra_attributes, **self.receive_stream.extra_attributes)
diff --git a/src/anyio/streams/tls.py b/src/anyio/streams/tls.py
index 971e838..dc734f3 100644
--- a/src/anyio/streams/tls.py
+++ b/src/anyio/streams/tls.py
@@ -1,30 +1,48 @@
 import ssl
-import sys
 from dataclasses import dataclass
 from functools import wraps
-from typing import Any, Callable, Dict, List, Optional, Tuple, TypeVar, Union, overload
+from typing import Any, Callable, Dict, List, Optional, Tuple, TypeVar, Union
 
 from .. import BrokenResourceError, EndOfStream
+from .._core._typedattr import TypedAttributeSet, typed_attribute
 from ..abc import AnyByteStream, ByteStream, Listener, TaskGroup
 
-if sys.version_info >= (3, 8):
-    from typing import Literal
-else:
-    from typing_extensions import Literal
-
 T_Retval = TypeVar('T_Retval')
 
 
+class TLSAttribute(TypedAttributeSet):
+    """Contains Transport Layer Security related attributes."""
+    #: the selected ALPN protocol
+    alpn_protocol: bool = typed_attribute()
+    #: the selected cipher
+    cipher: Tuple[str, str, int] = typed_attribute()
+    #: the peer certificate in dictionary form (see :meth:`ssl.SSLSocket.getpeercert` for more
+    #: information)
+    peer_certificate: Dict[str, Union[str, tuple]] = typed_attribute()
+    #: the peer certificate in binary form
+    peer_certificate_binary: bytes = typed_attribute()
+    #: ``True`` if this is the server side of the connection
+    server_side: bool = typed_attribute()
+    #: ciphers shared between both ends of the TLS connection
+    shared_ciphers: List[Tuple[str, str, int]] = typed_attribute()
+    #: the :class:`~ssl.SSLObject` used for encryption
+    ssl_object: ssl.SSLObject = typed_attribute()
+    #: ``True`` if this stream does (and expects) a closing TLS handshake when the stream is being
+    #: closed
+    standard_compatible: bool = typed_attribute()
+    #: the TLS protocol version (e.g. ``TLSv1.2``)
+    tls_version: str = typed_attribute()
+
+
 @dataclass
 class TLSStream(ByteStream):
     """
     A stream wrapper that encrypts all sent data and decrypts received data.
 
     This class has no public initializer; use :meth:`wrap` instead.
+    All extra attributes from :class:`~TLSAttribute` are supported.
 
     :var AnyByteStream transport_stream: the wrapped stream
-    :var bool standard_compatible: ``True`` if a closing handshake is expected from the peer and
-        done from this side, ``False`` if not
 
     """
     transport_stream: AnyByteStream
@@ -132,51 +150,27 @@ class TLSStream(ByteStream):
         await self._call_sslobject_method(self._ssl_object.write, item)
 
     async def send_eof(self) -> None:
-        version_tuple = tuple(int(part) for part in self.tls_version.split('.'))
+        tls_version = self.extra(TLSAttribute.tls_version)
+        version_tuple = tuple(int(part) for part in tls_version.split('.'))
         if version_tuple < (1, 3):
             raise NotImplementedError(f'send_eof() requires at least TLS v1.3; current session '
-                                      f'uses {self.tls_version}')
+                                      f'uses {tls_version}')
 
         raise NotImplementedError('send_eof() has not yet been implemented for TLS streams')
 
     @property
-    def alpn_protocol(self) -> Optional[str]:
-        return self._ssl_object.selected_alpn_protocol()  # type: ignore
-
-    def get_channel_binding(self, cb_type: str = 'tls-unique') -> bytes:
-        return self._ssl_object.get_channel_binding(cb_type)  # type: ignore
-
-    @property
-    def tls_version(self) -> str:
-        return self._ssl_object.version()  # type: ignore
-
-    @property
-    def cipher(self) -> Tuple[str, str, int]:
-        return self._ssl_object.cipher()  # type: ignore
-
-    @property
-    def shared_ciphers(self) -> List[Tuple[str, str, int]]:
-        return self._ssl_object.shared_ciphers()  # type: ignore
-
-    @property
-    def server_hostname(self) -> Optional[str]:
-        return self._ssl_object.server_hostname
-
-    @property
-    def server_side(self) -> bool:
-        return self._ssl_object.server_side
-
-    @overload
-    def getpeercert(self, binary_form: Literal[False] = False) -> Dict[str, Union[str, tuple]]:
-        ...
-
-    @overload
-    def getpeercert(self, binary_form: Literal[True]) -> bytes:
-        ...
-
-    def getpeercert(self, binary_form: bool = False) -> Union[Dict[str, Union[str, tuple]], bytes,
-                                                              None]:
-        return self._ssl_object.getpeercert(binary_form)
+    def extra_attributes(self):
+        return {
+            TLSAttribute.alpn_protocol: self._ssl_object.selected_alpn_protocol,
+            TLSAttribute.cipher: self._ssl_object.cipher,
+            TLSAttribute.peer_certificate: lambda: self._ssl_object.getpeercert(False),
+            TLSAttribute.peer_certificate_binary: lambda: self._ssl_object.getpeercert(True),
+            TLSAttribute.server_side: lambda: self._ssl_object.server_side,
+            TLSAttribute.shared_ciphers: lambda: self._ssl_object.shared_ciphers(),
+            TLSAttribute.standard_compatible: lambda: self.standard_compatible,
+            TLSAttribute.ssl_object: lambda: self._ssl_object,
+            TLSAttribute.tls_version: self._ssl_object.version
+        }
 
 
 @dataclass
@@ -185,6 +179,8 @@ class TLSListener(Listener[TLSStream]):
     A convenience listener that wraps another listener and auto-negotiates a TLS session on every
     accepted connection.
 
+    Supports only the :attr:`~TLSAttribute.standard_compatible` extra attribute.
+
     :param Listener listener: the listener to wrap
     :param ssl_context: the SSL context object
     :param standard_compatible: a flag passed through to :meth:`TLSStream.wrap`
@@ -206,3 +202,9 @@ class TLSListener(Listener[TLSStream]):
 
     async def aclose(self) -> None:
         await self.listener.aclose()
+
+    @property
+    def extra_attributes(self):
+        return {
+            TLSAttribute.standard_compatible: lambda: self.standard_compatible,
+        }


Test patch:
diff --git a/tests/streams/test_tls.py b/tests/streams/test_tls.py
index b555e51..c8b2f3f 100644
--- a/tests/streams/test_tls.py
+++ b/tests/streams/test_tls.py
@@ -6,7 +6,7 @@ from threading import Thread
 import pytest
 
 from anyio import BrokenResourceError, connect_tcp
-from anyio.streams.tls import TLSStream
+from anyio.streams.tls import TLSAttribute, TLSStream
 
 
 class TestTLSStream:
@@ -95,7 +95,7 @@ class TestTLSStream:
 
         server_thread.join()
         server_sock.close()
-        assert wrapper.alpn_protocol == 'dummy2'
+        assert wrapper.extra(TLSAttribute.alpn_protocol) == 'dummy2'
         assert server_alpn_protocol == b'dummy2'
 
     @pytest.mark.parametrize('server_compatible, client_compatible', [
diff --git a/tests/test_sockets.py b/tests/test_sockets.py
index ff614ae..7790d15 100644
--- a/tests/test_sockets.py
+++ b/tests/test_sockets.py
@@ -12,6 +12,7 @@ from anyio import (
     connect_unix, create_connected_udp_socket, create_event, create_task_group,
     create_tcp_listener, create_udp_socket, create_unix_listener, getaddrinfo, getnameinfo,
     move_on_after, wait_all_tasks_blocked)
+from anyio.abc.sockets import SocketAttribute
 from anyio.streams.stapled import MultiListener
 
 pytestmark = pytest.mark.anyio
@@ -134,10 +135,8 @@ class TestTCPStream:
 
     async def test_socket_options(self, family, server_addr):
         async with await connect_tcp(*server_addr) as stream:
-            assert stream.family == family
-            assert stream.getsockopt(socket.IPPROTO_TCP, socket.TCP_NODELAY) != 0
-            stream.setsockopt(socket.SOL_SOCKET, socket.SO_RCVBUF, 80000)
-            assert stream.getsockopt(socket.SOL_SOCKET, socket.SO_RCVBUF) in (80000, 160000)
+            raw_socket = stream.extra(SocketAttribute.raw_socket)
+            assert raw_socket.getsockopt(socket.IPPROTO_TCP, socket.TCP_NODELAY) != 0
 
     @pytest.mark.skipif(not socket.has_ipv6, reason='IPv6 is not available')
     @pytest.mark.parametrize('local_addr, expected_client_addr', [
@@ -298,9 +297,9 @@ class TestTCPListener:
     async def test_accept(self, family):
         async with await create_tcp_listener(local_host='localhost', family=family) as multi:
             for listener in multi.listeners:
-                client = socket.socket(listener.family)
+                client = socket.socket(listener.extra(SocketAttribute.family))
                 client.settimeout(1)
-                client.connect(listener.local_address)
+                client.connect(listener.extra(SocketAttribute.local_address))
                 stream = await listener.accept()
                 client.sendall(b'blah')
                 request = await stream.receive()
@@ -312,21 +311,25 @@ class TestTCPListener:
     async def test_socket_options(self, family):
         async with await create_tcp_listener(local_host='localhost', family=family) as multi:
             for listener in multi.listeners:
+                raw_socket = listener.extra(SocketAttribute.raw_socket)
                 if sys.platform == 'win32':
-                    assert listener.getsockopt(socket.SOL_SOCKET, socket.SO_EXCLUSIVEADDRUSE) != 0
+                    assert raw_socket.getsockopt(
+                        socket.SOL_SOCKET, socket.SO_EXCLUSIVEADDRUSE) != 0
                 else:
-                    assert listener.getsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR) != 0
+                    assert raw_socket.getsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR) != 0
 
-                listener.setsockopt(socket.SOL_SOCKET, socket.SO_RCVBUF, 80000)
-                assert listener.getsockopt(socket.SOL_SOCKET, socket.SO_RCVBUF) in (80000, 160000)
+                raw_socket.setsockopt(socket.SOL_SOCKET, socket.SO_RCVBUF, 80000)
+                assert raw_socket.getsockopt(
+                    socket.SOL_SOCKET, socket.SO_RCVBUF) in (80000, 160000)
 
-                client = socket.socket(listener.family)
+                client = socket.socket(raw_socket.family)
                 client.settimeout(1)
-                client.connect(listener.local_address)
+                client.connect(raw_socket.getsockname())
 
                 async with await listener.accept() as stream:
-                    assert stream.family == listener.family
-                    assert stream.getsockopt(socket.IPPROTO_TCP, socket.TCP_NODELAY) != 0
+                    raw_socket = stream.extra(SocketAttribute.raw_socket)
+                    assert raw_socket.family == listener.extra(SocketAttribute.family)
+                    assert raw_socket.getsockopt(socket.IPPROTO_TCP, socket.TCP_NODELAY) != 0
 
                 client.close()
 
@@ -336,11 +339,13 @@ class TestTCPListener:
         assert len(multi1.listeners) == 1
 
         multi2 = await create_tcp_listener(
-            local_host='localhost', local_port=multi1.listeners[0].local_address[1],
+            local_host='localhost',
+            local_port=multi1.listeners[0].extra(SocketAttribute.local_port),
             family=family, reuse_port=True)
         assert len(multi2.listeners) == 1
 
-        assert multi1.listeners[0].local_address == multi1.listeners[0].local_address
+        assert multi1.listeners[0].extra(SocketAttribute.local_address) == \
+               multi2.listeners[0].extra(SocketAttribute.local_address)
         await multi1.aclose()
         await multi2.aclose()
 
@@ -432,12 +437,6 @@ class TestUNIXStream:
         thread.join()
         assert chunks == [b'bl', b'ah']
 
-    async def test_socket_options(self, server_sock, socket_path):
-        async with await connect_unix(socket_path) as stream:
-            assert stream.family == socket.AddressFamily.AF_UNIX
-            stream.setsockopt(socket.SOL_SOCKET, socket.SO_RCVBUF, 80000)
-            assert stream.getsockopt(socket.SOL_SOCKET, socket.SO_RCVBUF) in (80000, 160000)
-
     async def test_concurrent_send(self, server_sock, socket_path):
         async def send_data():
             while True:
@@ -516,23 +515,25 @@ class TestUNIXListener:
 
     async def test_socket_options(self, socket_path):
         async with await create_unix_listener(socket_path) as listener:
-            assert listener.family == socket.AddressFamily.AF_UNIX
-            listener.setsockopt(socket.SOL_SOCKET, socket.SO_RCVBUF, 80000)
-            assert listener.getsockopt(socket.SOL_SOCKET, socket.SO_RCVBUF) in (80000, 160000)
+            listener_socket = listener.extra(SocketAttribute.raw_socket)
+            assert listener_socket.family == socket.AddressFamily.AF_UNIX
+            listener_socket.setsockopt(socket.SOL_SOCKET, socket.SO_RCVBUF, 80000)
+            assert listener_socket.getsockopt(
+                socket.SOL_SOCKET, socket.SO_RCVBUF) in (80000, 160000)
 
-            client = socket.socket(listener.family)
+            client = socket.socket(listener_socket.family)
             client.settimeout(1)
-            client.connect(listener.local_address)
+            client.connect(listener_socket.getsockname())
 
             async with await listener.accept() as stream:
-                assert stream.family == listener.family
+                assert stream.extra(SocketAttribute.family) == listener_socket.family
 
             client.close()
 
 
 async def test_multi_listener(tmp_path_factory):
     async def handle(stream):
-        client_addresses.append(stream.remote_address)
+        client_addresses.append(stream.extra(SocketAttribute.remote_address))
         await event.set()
         await stream.aclose()
 
@@ -548,12 +549,14 @@ async def test_multi_listener(tmp_path_factory):
             await tg.spawn(multi_listener.serve, handle)
             for listener in multi_listener.listeners:
                 event = create_event()
-                if sys.platform != 'win32' and listener.family == socket.AddressFamily.AF_UNIX:
-                    stream = await connect_unix(listener.local_address)
+                local_address = listener.extra(SocketAttribute.local_address)
+                if sys.platform != 'win32' and listener.extra(SocketAttribute.family) == \
+                        socket.AddressFamily.AF_UNIX:
+                    stream = await connect_unix(local_address)
                 else:
-                    stream = await connect_tcp(*listener.local_address)
+                    stream = await connect_tcp(*local_address)
 
-                expected_addresses.append(stream.local_address)
+                expected_addresses.append(stream.extra(SocketAttribute.local_address))
                 await event.wait()
                 await stream.aclose()
 
@@ -566,15 +569,16 @@ async def test_multi_listener(tmp_path_factory):
 class TestUDPSocket:
     async def test_send_receive(self, family):
         async with await create_udp_socket(local_host='localhost', family=family) as sock:
-            await sock.sendto(b'blah', *sock.local_address)
+            host, port = sock.extra(SocketAttribute.local_address)
+            await sock.sendto(b'blah', host, port)
             request, addr = await sock.receive()
             assert request == b'blah'
-            assert addr == sock.local_address
+            assert addr == sock.extra(SocketAttribute.local_address)
 
-            await sock.sendto(b'halb', *sock.local_address)
+            await sock.sendto(b'halb', host, port)
             response, addr = await sock.receive()
             assert response == b'halb'
-            assert addr == sock.local_address
+            assert addr == (host, port)
 
     async def test_iterate(self, family):
         async def serve():
@@ -582,31 +586,25 @@ class TestUDPSocket:
                 await server.send((packet[::-1], addr))
 
         async with await create_udp_socket(family=family, local_host='localhost') as server:
-            host, port = server.local_address
+            host, port = server.extra(SocketAttribute.local_address)
             async with await create_udp_socket(family=family, local_host='localhost') as client:
                 async with create_task_group() as tg:
                     await tg.spawn(serve)
                     await client.sendto(b'FOOBAR', host, port)
-                    assert await client.receive() == (b'RABOOF', server.local_address)
+                    assert await client.receive() == (b'RABOOF', (host, port))
                     await client.sendto(b'123456', host, port)
-                    assert await client.receive() == (b'654321', server.local_address)
+                    assert await client.receive() == (b'654321', (host, port))
                     await tg.cancel_scope.cancel()
 
-    async def test_socket_options(self, family):
-        async with await create_udp_socket(family=family, local_host='localhost') as udp:
-            assert udp.family == family
-            udp.setsockopt(socket.SOL_SOCKET, socket.SO_RCVBUF, 80000)
-            assert udp.getsockopt(socket.SOL_SOCKET, socket.SO_RCVBUF) in (80000, 160000)
-
     @pytest.mark.skipif(sys.platform == 'win32', reason='Not supported on Windows')
     async def test_reuse_port(self, family):
         async with await create_udp_socket(family=family, local_host='localhost',
                                            reuse_port=True) as udp:
-            port = udp.local_address[1]
+            port = udp.extra(SocketAttribute.local_port)
             assert port != 0
             async with await create_udp_socket(family=family, local_host='localhost',
                                                local_port=port, reuse_port=True) as udp2:
-                assert port == udp2.local_address[1]
+                assert port == udp2.extra(SocketAttribute.local_port)
 
     async def test_concurrent_receive(self):
         async with await create_udp_socket(family=socket.AF_INET, local_host='localhost') as udp:
@@ -640,7 +638,7 @@ class TestUDPSocket:
 
     async def test_send_after_close(self):
         udp = await create_udp_socket(family=socket.AF_INET, local_host='localhost')
-        host, port = udp.local_address
+        host, port = udp.extra(SocketAttribute.local_address)
         await udp.aclose()
         with pytest.raises(ClosedResourceError):
             await udp.sendto(b'foo', host, port)
@@ -666,34 +664,27 @@ class TestConnectedUDPSocket:
                 await udp2.send(packet[::-1])
 
         async with await create_udp_socket(family=family, local_host='localhost') as udp1:
-            host, port = udp1.local_address
+            host, port = udp1.extra(SocketAttribute.local_address)
             async with await create_connected_udp_socket(host, port) as udp2:
-                host, port = udp2.local_address
+                host, port = udp2.extra(SocketAttribute.local_address)
                 async with create_task_group() as tg:
                     await tg.spawn(serve)
                     await udp1.sendto(b'FOOBAR', host, port)
-                    assert await udp1.receive() == (b'RABOOF', udp2.local_address)
+                    assert await udp1.receive() == (b'RABOOF', (host, port))
                     await udp1.sendto(b'123456', host, port)
-                    assert await udp1.receive() == (b'654321', udp2.local_address)
+                    assert await udp1.receive() == (b'654321', (host, port))
                     await tg.cancel_scope.cancel()
 
-    async def test_socket_options(self, family):
-        async with await create_connected_udp_socket(
-                'localhost', 5000, family=family, local_host='localhost') as udp:
-            assert udp.family == family
-            udp.setsockopt(socket.SOL_SOCKET, socket.SO_RCVBUF, 80000)
-            assert udp.getsockopt(socket.SOL_SOCKET, socket.SO_RCVBUF) in (80000, 160000)
-
     @pytest.mark.skipif(sys.platform == 'win32', reason='Not supported on Windows')
     async def test_reuse_port(self, family):
         async with await create_connected_udp_socket(
                 'localhost', 6000, family=family, local_host='localhost', reuse_port=True) as udp:
-            port = udp.local_address[1]
+            port = udp.extra(SocketAttribute.local_port)
             assert port != 0
             async with await create_connected_udp_socket(
                     'localhost', 6001, family=family, local_host='localhost', local_port=port,
                     reuse_port=True) as udp2:
-                assert port == udp2.local_address[1]
+                assert port == udp2.extra(SocketAttribute.local_port)
 
     async def test_concurrent_receive(self):
         async with await create_connected_udp_socket(