# SPDX-FileCopyrightText: Christian Amsüss and the aiocoap contributors # # SPDX-License-Identifier: MIT """This module provides interface base classes to various aiocoap software components, especially with respect to request and response handling. It describes `abstract base classes`_ for messages, endpoints etc. It is *completely unrelated* to the concept of "network interfaces". .. _`abstract base classes`: https://docs.python.org/3/library/abc""" from __future__ import annotations import abc import asyncio import warnings from aiocoap.pipe import Pipe from aiocoap.numbers.constants import MAX_REGULAR_BLOCK_SIZE_EXP from typing import Optional, Callable class MessageInterface(metaclass=abc.ABCMeta): """A MessageInterface is an object that can exchange addressed messages over unreliable transports. Implementations send and receive messages with message type and message ID, and are driven by a Context that deals with retransmission. Usually, an MessageInterface refers to something like a local socket, and send messages to different remote endpoints depending on the message's addresses. Just as well, a MessageInterface can be useful for one single address only, or use various local addresses depending on the remote address. """ @abc.abstractmethod async def shutdown(self): """Deactivate the complete transport, usually irrevertably. When the coroutine returns, the object must have made sure that it can be destructed by means of ref-counting or a garbage collector run.""" @abc.abstractmethod def send(self, message): """Send a given :class:`Message` object""" @abc.abstractmethod async def determine_remote(self, message): """Return a value suitable for the message's remote property based on its .opt.uri_host or .unresolved_remote. May return None, which indicates that the MessageInterface can not transport the message (typically because it is of the wrong scheme).""" class EndpointAddress(metaclass=abc.ABCMeta): """An address that is suitable for routing through the application to a remote endpoint. Depending on the MessageInterface implementation used, an EndpointAddress property of a message can mean the message is exchanged "with [2001:db8::2:1]:5683, while my local address was [2001:db8:1::1]:5683" (typical of UDP6), "over the connected , wherever that's connected to" (simple6 or TCP) or "with participant 0x01 of the OSCAP key 0x..., routed over ". EndpointAddresses are only constructed by MessageInterface objects, either for incoming messages or when populating a message's .remote in :meth:`MessageInterface.determine_remote`. There is no requirement that those address are always identical for a given address. However, incoming addresses must be hashable and hash-compare identically to requests from the same context. The "same context", for the purpose of EndpointAddresses, means that the message must be eligible for request/response, blockwise (de)composition and observations. (For example, in a DTLS context, the hash must change between epochs due to RFC7252 Section 9.1.2). So far, it is required that hash-identical objects also compare the same. That requirement might go away in future to allow equality to reflect finer details that are not hashed. (The only property that is currently known not to be hashed is the local address in UDP6, because that is *unknown* in initially sent packages, and thus disregarded for comparison but needed to round-trip through responses.) """ @property @abc.abstractmethod def hostinfo(self): """The authority component of URIs that this endpoint represents when request are sent to it Note that the presence of a hostinfo does not necessarily mean that globally meaningful or even syntactically valid URI can be constructed out of it; use the :attr:`.uri` property for this.""" @property @abc.abstractmethod def hostinfo_local(self): """The authority component of URIs that this endpoint represents when requests are sent from it. As with :attr:`.hostinfo`, this does not necessarily produce sufficient input for a URI; use :attr:`.uri_local` instead.""" @property def uri(self): """Deprecated alias for uri_base""" return self.uri_base @property @abc.abstractmethod def uri_base(self): """The base URI for the peer (typically scheme plus .hostinfo). This raises :class:`.error.AnonymousHost` when executed on an address whose peer coordinates can not be expressed meaningfully in a URI.""" @property @abc.abstractmethod def uri_base_local(self): """The base URI for the local side of this remote. This raises :class:`.error.AnonymousHost` when executed on an address whose local coordinates can not be expressed meaningfully in a URI.""" @property @abc.abstractmethod def is_multicast(self): """True if the remote address is a multicast address, otherwise false.""" @property @abc.abstractmethod def is_multicast_locally(self): """True if the local address is a multicast address, otherwise false.""" @property @abc.abstractmethod def scheme(Self): """The that is used with addresses of this kind This is usually a class property. It is applicable to both sides of the communication. (Should there ever be a scheme that addresses the participants differently, a scheme_local will be added.)""" @property def maximum_block_size_exp(self) -> int: """The maximum negotiated block size that can be sent to this remote.""" return MAX_REGULAR_BLOCK_SIZE_EXP # Giving some slack so that barely-larger messages (like OSCORE typically # are) don't get fragmented -- but still for migration to maximum message # size so we don't have to guess any more how much may be option and how # much payload @property def maximum_payload_size(self) -> int: """The maximum payload size that can be sent to this remote. Only relevant if maximum_block_size_exp is 7. This will be removed in favor of a maximum message size when the block handlers can get serialization length predictions from the remote.""" return 1124 def as_response_address(self): """Address to be assigned to a response to messages that arrived with this message This can (and does, by default) return self, but gives the protocol the opportunity to react to create a modified copy to deal with variations from multicast. """ return self @property def authenticated_claims(self): """Iterable of objects representing any claims (e.g. an identity, or generally objects that can be used to authorize particular accesses) that were authenticated for this remote. This is experimental and may be changed without notice. Its primary use is on the server side; there, a request handler (or resource decorator) can use the claims to decide whether the client is authorized for a particular request. Use on the client side is planned as a requirement on a request, although (especially on side-effect free non-confidential requests) it can also be used in response processing.""" # "no claims" is a good default return () @property @abc.abstractmethod def blockwise_key(self): """A hashable (ideally, immutable) value that is only the same for remotes from which blocks may be combined. (With all current transports that means that the network addresses need to be in there, and the identity of the security context). It does *not* just hinge on the identity of the address object, as a first block may come in an OSCORE group request and follow-ups may come in pairwise requests. (And there might be allowed relaxations on the transport under OSCORE, but that'd need further discussion).""" # FIXME: should this behave like something that keeps the address # alive? Conversely, if the address gets deleted, can this reach the # block keys and make their stuff vanish from the caches? # # FIXME: what do security mechanisms best put here? Currently it's a # wild mix of keys (OSCORE -- only thing guaranteed to never be reused; # DTLS client because it's available) and claims (DTLS server, because # it's available and if the claims set matches it can't be that wrong # either can it?) class MessageManager(metaclass=abc.ABCMeta): """The interface an entity that drives a MessageInterface provides towards the MessageInterface for callbacks and object acquisition.""" @abc.abstractmethod def dispatch_message(self, message): """Callback to be invoked with an incoming message""" @abc.abstractmethod def dispatch_error(self, error: Exception, remote): """Callback to be invoked when the operating system indicated an error condition from a particular remote.""" @property @abc.abstractmethod def client_credentials(self): """A CredentialsMap that transports should consult when trying to establish a security context""" class TokenInterface(metaclass=abc.ABCMeta): @abc.abstractmethod def send_message( self, message, messageerror_monitor ) -> Optional[Callable[[], None]]: """Send a message. If it returns a a callable, the caller is asked to call in case it no longer needs the message sent, and to dispose of if it doesn't intend to any more. messageerror_monitor is a function that will be called at most once by the token interface: When the underlying layer is indicating that this concrete message could not be processed. This is typically the case for RSTs on from the message layer, and used to cancel observations. Errors that are not likely to be specific to a message (like retransmission timeouts, or ICMP errors) are reported through dispatch_error instead. (While the information which concrete message triggered that might be available, it is not likely to be relevant). Currently, it is up to the TokenInterface to unset the no_response option in response messages, and to possibly not send them.""" @abc.abstractmethod async def fill_or_recognize_remote(self, message): """Return True if the message is recognized to already have a .remote managedy by this TokenInterface, or return True and set a .remote on message if it should (by its unresolved remote or Uri-* options) be routed through this TokenInterface, or return False otherwise.""" class TokenManager(metaclass=abc.ABCMeta): # to be described in full; at least there is a dispatch_error in analogy to MessageManager's pass class RequestInterface(metaclass=abc.ABCMeta): @abc.abstractmethod async def fill_or_recognize_remote(self, message): pass @abc.abstractmethod def request(self, request: Pipe): pass class RequestProvider(metaclass=abc.ABCMeta): """ .. automethod:: request .. (which we have to list here manually because the private override in the method is needed for the repeated signature in Context) """ @abc.abstractmethod def request(self, request_message, handle_blockwise=True): """Create and act on a :class:`Request` object that will be handled according to the provider's implementation. Note that the request is not necessarily sent on the wire immediately; it may (but, depend on the transport does not necessarily) rely on the response to be waited for. If handle_blockwise is True (the default), the request provider will split the request and/or collect the response parts automatically. The block size indicated by the remote is used, and can be decreased by setting the message's :attr:`.remote.maximum_block_size_exp ` property. Note that by being a property of the remote, this may affect other block-wise operations on the same remote -- this should be desirable behavior. :meta private: (not actually private, just hiding from automodule due to being grouped with the important functions) """ class Request(metaclass=abc.ABCMeta): """A CoAP request, initiated by sending a message. Typically, this is not instantiated directly, but generated by a :meth:`RequestProvider.request` method.""" response = """A future that is present from the creation of the object and \ fulfilled with the response message. When legitimate errors occur, this becomes an aiocoap.Error. (Eg. on any kind of network failure, encryption trouble, or protocol violations). Any other kind of exception raised from this is a bug in aiocoap, and should better stop the whole application. """ class Resource(metaclass=abc.ABCMeta): """Interface that is expected by a :class:`.protocol.Context` to be present on the serversite, which renders all requests to that context.""" def __init__(self): super().__init__() # FIXME: These keep addresses alive, and thus possibly transports. # Going through the shutdown dance per resource seems extraneous. # Options are to accept addresses staying around (making sure they # don't keep their transports alive, if that's a good idea), to hash # them, or to make them weak. from .blockwise import Block1Spool, Block2Cache self._block1 = Block1Spool() self._block2 = Block2Cache() @abc.abstractmethod async def render(self, request): """Return a message that can be sent back to the requester. This does not need to set any low-level message options like remote, token or message type; it does however need to set a response code. A response returned may carry a no_response option (which is actually specified to apply to requests only); the underlying transports will decide based on that and its code whether to actually transmit the response.""" @abc.abstractmethod async def needs_blockwise_assembly(self, request): """Indicator whether aiocoap should assemble request blocks to a single request and extract the requested blocks from a complete-resource answer (True), or whether the resource will do that by itself (False).""" async def _render_to_pipe(self, pipe: Pipe) -> None: if not hasattr(self, "_block1"): warnings.warn( "No attribute _block1 found on instance of " f"{type(self).__name__}, make sure its __init__ code " "properly calls super()!", DeprecationWarning, ) from .blockwise import Block1Spool, Block2Cache self._block1 = Block1Spool() self._block2 = Block2Cache() req = pipe.request if await self.needs_blockwise_assembly(req): req = self._block1.feed_and_take(req) # Note that unless the lambda gets called, we're not fully # accessing req any more -- we're just looking at its block2 # option, and the blockwise key extracted earlier. res = await self._block2.extract_or_insert(req, lambda: self.render(req)) res.opt.block1 = req.opt.block1 else: res = await self.render(req) pipe.add_response(res, is_last=True) async def render_to_pipe(self, pipe: Pipe) -> None: """Create any number of responses (as indicated by the request) into the request stream. This method is provided by the base Resource classes; if it is overridden, then :meth:`~interfaces.Resource.render`, :meth:`needs_blockwise_assembly` and :meth:`~.interfaces.ObservableResource.add_observation` are not used any more. (They still need to be implemented to comply with the interface definition, which is yet to be updated).""" warnings.warn( "Request interface is changing: Resources should " "implement render_to_pipe or inherit from " "resource.Resource which implements that based on any " "provided render methods", DeprecationWarning, ) if isinstance(self, ObservableResource): # While the above deprecation is used, a resource previously # inheriting from (X, ObservableResource) with X inheriting from # Resource might find itself using this method. When migrating over # to inheriting from resource.Resource, this error will become # apparent and this can die with the rest of this workaround. return await ObservableResource._render_to_pipe(self, pipe) await self._render_to_pipe(pipe) class ObservableResource(Resource, metaclass=abc.ABCMeta): """Interface the :class:`.protocol.ServerObservation` uses to negotiate whether an observation can be established based on a request. This adds only functionality for registering and unregistering observations; the notification contents will be retrieved from the resource using the regular :meth:`~.Resource.render` method from crafted (fake) requests. """ @abc.abstractmethod async def add_observation(self, request, serverobservation): """Before the incoming request is sent to :meth:`~.Resource.render`, the :meth:`.add_observation` method is called. If the resource chooses to accept the observation, it has to call the `serverobservation.accept(cb)` with a callback that will be called when the observation ends. After accepting, the ObservableResource should call `serverobservation.trigger()` whenever it changes its state; the ServerObservation will then initiate notifications by having the request rendered again.""" async def _render_to_pipe(self, pipe: Pipe) -> None: from .protocol import ServerObservation # If block2:>0 comes along, we'd just ignore the observe if pipe.request.opt.observe != 0: return await Resource._render_to_pipe(self, pipe) # If block1 happens here, we can probably just not support it for the # time being. (Given that block1 + observe is untested and thus does # not work so far anyway). servobs = ServerObservation() await self.add_observation(pipe.request, servobs) try: first_response = await self.render(pipe.request) if ( not servobs._accepted or servobs._early_deregister or not first_response.code.is_successful() ): pipe.add_response(first_response, is_last=True) return # FIXME: observation numbers should actually not be per # asyncio.task, but per (remote, token). if a client renews an # observation (possibly with a new ETag or whatever is deemed # legal), the new observation events should still carry larger # numbers. (if they did not, the client might be tempted to discard # them). first_response.opt.observe = next_observation_number = 0 # If block2 were to happen here, we'd store the full response # here, and pick out block2:0. pipe.add_response(first_response, is_last=False) while True: await servobs._trigger # if you wonder why the lines around this are not just `response = # await servobs._trigger`, have a look at the 'double' tests in # test_observe.py: A later triggering could have replaced # servobs._trigger in the meantime. response = servobs._trigger.result() servobs._trigger = asyncio.get_running_loop().create_future() if response is None: response = await self.render(pipe.request) # If block2 were to happen here, we'd store the full response # here, and pick out block2:0. is_last = servobs._late_deregister or not response.code.is_successful() if not is_last: next_observation_number += 1 response.opt.observe = next_observation_number pipe.add_response(response, is_last=is_last) if is_last: return finally: servobs._cancellation_callback() async def render_to_pipe(self, request: Pipe) -> None: warnings.warn( "Request interface is changing: Resources should " "implement render_to_pipe or inherit from " "resource.Resource which implements that based on any " "provided render methods", DeprecationWarning, ) await self._render_to_pipe(request)