Coverage for src/aiocoap/message.py: 0%
287 statements
« prev ^ index » next coverage.py v7.7.0, created at 2025-03-20 17:26 +0000
1# SPDX-FileCopyrightText: Christian Amsüss and the aiocoap contributors
2#
3# SPDX-License-Identifier: MIT
5from __future__ import annotations
7import ipaddress
8import urllib.parse
9import struct
10import copy
11import string
12from collections import namedtuple
14from . import error, optiontypes
15from .numbers.codes import Code, CHANGED
16from .numbers.types import Type
17from .numbers.constants import TransportTuning, MAX_REGULAR_BLOCK_SIZE_EXP
18from .options import Options
19from .util import hostportjoin, hostportsplit, Sentinel, quote_nonascii
20from .util.uri import quote_factory, unreserved, sub_delims
21from . import interfaces
23__all__ = ["Message", "NoResponse"]
25# FIXME there should be a proper inteface for this that does all the urllib
26# patching possibly required and works with pluggable transports. urls qualify
27# if they can be parsed into the Proxy-Scheme / Uri-* structure.
28coap_schemes = ["coap", "coaps", "coap+tcp", "coaps+tcp", "coap+ws", "coaps+ws"]
30# Monkey patch urllib to make URL joining available in CoAP
31# This is a workaround for <http://bugs.python.org/issue23759>.
32urllib.parse.uses_relative.extend(coap_schemes)
33urllib.parse.uses_netloc.extend(coap_schemes)
36class Message(object):
37 """CoAP Message with some handling metadata
39 This object's attributes provide access to the fields in a CoAP message and
40 can be directly manipulated.
42 * Some attributes are additional data that do not round-trip through
43 serialization and deserialization. They are marked as "non-roundtrippable".
44 * Some attributes that need to be filled for submission of the message can
45 be left empty by most applications, and will be taken care of by the
46 library. Those are marked as "managed".
48 The attributes are:
50 * :attr:`payload`: The payload (body) of the message as bytes.
51 * :attr:`mtype`: Message type (CON, ACK etc, see :mod:`.numbers.types`).
52 Managed unless set by the application.
53 * :attr:`code`: The code (either request or response code), see
54 :mod:`.numbers.codes`.
55 * :attr:`opt`: A container for the options, see :class:`.options.Options`.
57 * :attr:`mid`: The message ID. Managed by the :class:`.Context`.
58 * :attr:`token`: The message's token as bytes. Managed by the :class:`.Context`.
59 * :attr:`remote`: The socket address of the other side, managed by the
60 :class:`.protocol.Request` by resolving the ``.opt.uri_host`` or
61 ``unresolved_remote``, or by the stack by echoing the incoming
62 request's. Follows the :class:`.interfaces.EndpointAddress` interface.
63 Non-roundtrippable.
65 While a message has not been transmitted, the property is managed by the
66 :class:`.Message` itself using the :meth:`.set_request_uri()` or the
67 constructor `uri` argument.
68 * :attr:`transport_tuning`: Parameters used by one or more transports to
69 guide transmission. These are purely advisory hints; unknown properties
70 of that object are ignored, and transports consider them over built-in
71 constants on a best-effort basis.
73 Note that many attributes are mandatory if this is not None; it is
74 recommended that any objects passed in here are based on the
75 :class:`aiocoap.numbers.constants.TransportTuning` class.
77 * :attr:`request`: The request to which an incoming response message
78 belongs; only available at the client. Managed by the
79 :class:`.interfaces.RequestProvider` (typically a :class:`.Context`).
81 These properties are still available but deprecated:
83 * requested_*: Managed by the :class:`.protocol.Request` a response results
84 from, and filled with the request's URL data. Non-roundtrippable.
86 * unresolved_remote: ``host[:port]`` (strictly speaking; hostinfo as in a
87 URI) formatted string. If this attribute is set, it overrides
88 ``.RequestManageropt.uri_host`` (and ``-_port``) when it comes to filling the
89 ``remote`` in an outgoing request.
91 Use this when you want to send a request with a host name that would not
92 normally resolve to the destination address. (Typically, this is used for
93 proxying.)
95 Options can be given as further keyword arguments at message construction
96 time. This feature is experimental, as future message parameters could
97 collide with options.
100 The four messages involved in an exchange
101 -----------------------------------------
103 ::
105 Requester Responder
107 +-------------+ +-------------+
108 | request msg | ---- send request ---> | request msg |
109 +-------------+ +-------------+
110 |
111 processed into
112 |
113 v
114 +-------------+ +-------------+
115 | response m. | <--- send response --- | response m. |
116 +-------------+ +-------------+
119 The above shows the four message instances involved in communication
120 between an aiocoap client and server process. Boxes represent instances of
121 Message, and the messages on the same line represent a single CoAP as
122 passed around on the network. Still, they differ in some aspects:
124 * The requested URI will look different between requester and responder
125 if the requester uses a host name and does not send it in the message.
126 * If the request was sent via multicast, the response's requested URI
127 differs from the request URI because it has the responder's address
128 filled in. That address is not known at the responder's side yet, as
129 it is typically filled out by the network stack.
130 * It is yet unclear whether the response's URI should contain an IP
131 literal or a host name in the unicast case if the Uri-Host option was
132 not sent.
133 * Properties like Message ID and token will differ if a proxy was
134 involved.
135 * Some options or even the payload may differ if a proxy was involved.
136 """
138 def __init__(
139 self,
140 *,
141 mtype=None,
142 mid=None,
143 code=None,
144 payload=b"",
145 token=b"",
146 uri=None,
147 transport_tuning=None,
148 **kwargs,
149 ):
150 self.version = 1
151 if mtype is None:
152 # leave it unspecified for convenience, sending functions will know what to do
153 self.mtype = None
154 else:
155 self.mtype = Type(mtype)
156 self.mid = mid
157 if code is None:
158 # as above with mtype
159 self.code = None
160 else:
161 self.code = Code(code)
162 self.token = token
163 self.payload = payload
164 self.opt = Options()
166 self.remote = None
168 self.transport_tuning = transport_tuning or TransportTuning()
170 # deprecation error, should go away roughly after 0.2 release
171 if self.payload is None:
172 raise TypeError("Payload must not be None. Use empty string instead.")
174 if uri:
175 self.set_request_uri(uri)
177 for k, v in kwargs.items():
178 setattr(self.opt, k, v)
180 def __repr__(self):
181 return "<aiocoap.Message at %#x: %s %s (%s, %s) remote %s%s%s>" % (
182 id(self),
183 self.mtype if self.mtype is not None else "no mtype,",
184 self.code,
185 "MID %s" % self.mid if self.mid is not None else "no MID",
186 "token %s" % self.token.hex() if self.token else "empty token",
187 self.remote,
188 ", %s option(s)" % len(self.opt._options) if self.opt._options else "",
189 ", %s byte(s) payload" % len(self.payload) if self.payload else "",
190 )
192 def _repr_html_(self):
193 """An HTML representation for Jupyter and similar environments
195 While the precise format is not guaranteed, it will be usable through
196 tooltips and possibly fold-outs:
198 >>> from aiocoap import *
199 >>> msg = Message(code=GET, uri="coap://localhost/other/separate")
200 >>> html = msg._repr_html_()
201 >>> 'Message with code <abbr title="Request Code 0.01">GET</abbr>' in html
202 True
203 >>> '3 options</summary>' in html
204 True
205 """
206 import html
208 return f"""<details style="padding-left:1em"><summary style="margin-left:-1em;display:list-item;">Message with code {self.code._repr_html_() if self.code is not None else "None"}, remote {html.escape(str(self.remote))}</summary>
209 {self.opt._repr_html_()}{self.payload_html()}"""
211 def payload_html(self):
212 """An HTML representation of the payload
214 The precise format is not guaranteed, but generally it may involve
215 pretty-printing, syntax highlighting, and visible notes that content
216 was reflowed or absent.
218 The result may vary depending on the available modules (falling back go
219 plain HTML-escaped ``repr()`` of the payload).
220 """
221 import html
223 if not self.payload:
224 return "<p>No payload</p>"
225 else:
226 from . import defaults
228 if defaults.prettyprint_missing_modules():
229 return f"<code>{html.escape(repr(self.payload))}</code>"
230 else:
231 from .util.prettyprint import pretty_print, lexer_for_mime
233 (notes, mediatype, text) = pretty_print(self)
234 import pygments
235 from pygments.formatters import HtmlFormatter
237 try:
238 lexer = lexer_for_mime(mediatype)
239 text = pygments.highlight(text, lexer, HtmlFormatter())
240 except pygments.util.ClassNotFound:
241 text = html.escape(text)
242 return (
243 "<div>"
244 + "".join(
245 f'<p style="color:gray;font-size:small;">{html.escape(n)}</p>'
246 for n in notes
247 )
248 + f"<pre>{text}</pre>"
249 + "</div>"
250 )
252 def copy(self, **kwargs):
253 """Create a copy of the Message. kwargs are treated like the named
254 arguments in the constructor, and update the copy."""
255 # This is part of moving messages in an "immutable" direction; not
256 # necessarily hard immutable. Let's see where this goes.
258 new = type(self)(
259 mtype=kwargs.pop("mtype", self.mtype),
260 mid=kwargs.pop("mid", self.mid),
261 code=kwargs.pop("code", self.code),
262 payload=kwargs.pop("payload", self.payload),
263 token=kwargs.pop("token", self.token),
264 # Assuming these are not readily mutated, but rather passed
265 # around in a class-like fashion
266 transport_tuning=kwargs.pop("transport_tuning", self.transport_tuning),
267 )
268 new.remote = kwargs.pop("remote", self.remote)
269 new.opt = copy.deepcopy(self.opt)
271 if "uri" in kwargs:
272 new.set_request_uri(kwargs.pop("uri"))
274 for k, v in kwargs.items():
275 setattr(new.opt, k, v)
277 return new
279 @classmethod
280 def decode(cls, rawdata, remote=None):
281 """Create Message object from binary representation of message."""
282 try:
283 (vttkl, code, mid) = struct.unpack("!BBH", rawdata[:4])
284 except struct.error:
285 raise error.UnparsableMessage("Incoming message too short for CoAP")
286 version = (vttkl & 0xC0) >> 6
287 if version != 1:
288 raise error.UnparsableMessage("Fatal Error: Protocol Version must be 1")
289 mtype = (vttkl & 0x30) >> 4
290 token_length = vttkl & 0x0F
291 msg = Message(mtype=mtype, mid=mid, code=code)
292 msg.token = rawdata[4 : 4 + token_length]
293 msg.payload = msg.opt.decode(rawdata[4 + token_length :])
294 msg.remote = remote
295 return msg
297 def encode(self):
298 """Create binary representation of message from Message object."""
299 if self.code is None or self.mtype is None or self.mid is None:
300 raise TypeError(
301 "Fatal Error: Code, Message Type and Message ID must not be None."
302 )
303 rawdata = bytes(
304 [
305 (self.version << 6)
306 + ((self.mtype & 0x03) << 4)
307 + (len(self.token) & 0x0F)
308 ]
309 )
310 rawdata += struct.pack("!BH", self.code, self.mid)
311 rawdata += self.token
312 rawdata += self.opt.encode()
313 if len(self.payload) > 0:
314 rawdata += bytes([0xFF])
315 rawdata += self.payload
316 return rawdata
318 def get_cache_key(self, ignore_options=()):
319 """Generate a hashable and comparable object (currently a tuple) from
320 the message's code and all option values that are part of the cache key
321 and not in the optional list of ignore_options (which is the list of
322 option numbers that are not technically NoCacheKey but handled by the
323 application using this method).
325 >>> from aiocoap.numbers import GET
326 >>> m1 = Message(code=GET)
327 >>> m2 = Message(code=GET)
328 >>> m1.opt.uri_path = ('s', '1')
329 >>> m2.opt.uri_path = ('s', '1')
330 >>> m1.opt.size1 = 10 # the only no-cache-key option in the base spec
331 >>> m2.opt.size1 = 20
332 >>> m1.get_cache_key() == m2.get_cache_key()
333 True
334 >>> m2.opt.etag = b'000'
335 >>> m1.get_cache_key() == m2.get_cache_key()
336 False
337 >>> from aiocoap.numbers.optionnumbers import OptionNumber
338 >>> ignore = [OptionNumber.ETAG]
339 >>> m1.get_cache_key(ignore) == m2.get_cache_key(ignore)
340 True
341 """
343 options = []
345 for option in self.opt.option_list():
346 if option.number in ignore_options or (
347 option.number.is_safetoforward() and option.number.is_nocachekey()
348 ):
349 continue
350 options.append((option.number, option.value))
352 return (self.code, tuple(options))
354 #
355 # splitting and merging messages into and from message blocks
356 #
358 def _extract_block(self, number, size_exp, max_bert_size):
359 """Extract block from current message."""
360 if size_exp == 7:
361 start = number * 1024
362 size = 1024 * (max_bert_size // 1024)
363 else:
364 size = 2 ** (size_exp + 4)
365 start = number * size
367 if start >= len(self.payload):
368 raise error.BadRequest("Block request out of bounds")
370 end = start + size if start + size < len(self.payload) else len(self.payload)
371 more = True if end < len(self.payload) else False
373 payload = self.payload[start:end]
374 blockopt = (number, more, size_exp)
376 if self.code.is_request():
377 return self.copy(payload=payload, mid=None, block1=blockopt)
378 else:
379 return self.copy(payload=payload, mid=None, block2=blockopt)
381 def _append_request_block(self, next_block):
382 """Modify message by appending another block"""
383 if not self.code.is_request():
384 raise ValueError("_append_request_block only works on requests.")
386 block1 = next_block.opt.block1
387 if block1.more:
388 if len(next_block.payload) == block1.size:
389 pass
390 elif (
391 block1.size_exponent == 7 and len(next_block.payload) % block1.size == 0
392 ):
393 pass
394 else:
395 raise error.BadRequest("Payload size does not match Block1")
396 if block1.start == len(self.payload):
397 self.payload += next_block.payload
398 self.opt.block1 = block1
399 self.token = next_block.token
400 self.mid = next_block.mid
401 if not block1.more and next_block.opt.block2 is not None:
402 self.opt.block2 = next_block.opt.block2
403 else:
404 # possible extension point: allow messages with "gaps"; then
405 # ValueError would only be raised when trying to overwrite an
406 # existing part; it is doubtful though that the blockwise
407 # specification even condones such behavior.
408 raise ValueError()
410 def _append_response_block(self, next_block):
411 """Append next block to current response message.
412 Used when assembling incoming blockwise responses."""
413 if not self.code.is_response():
414 raise ValueError("_append_response_block only works on responses.")
416 block2 = next_block.opt.block2
417 if not block2.is_valid_for_payload_size(len(next_block.payload)):
418 raise error.UnexpectedBlock2("Payload size does not match Block2")
419 if block2.start != len(self.payload):
420 # Does not need to be implemented as long as the requesting code
421 # sequentially clocks out data
422 raise error.NotImplemented()
424 if next_block.opt.etag != self.opt.etag:
425 raise error.ResourceChanged()
427 self.payload += next_block.payload
428 self.opt.block2 = block2
429 self.token = next_block.token
430 self.mid = next_block.mid
432 def _generate_next_block2_request(self, response):
433 """Generate a sub-request for next response block.
435 This method is used by client after receiving blockwise response from
436 server with "more" flag set."""
438 # Note: response here is the assembled response, but (due to
439 # _append_response_block's workings) it carries the Block2 option of
440 # the last received block.
442 next_after_received = len(response.payload) // response.opt.block2.size
443 blockopt = optiontypes.BlockOption.BlockwiseTuple(
444 next_after_received, False, response.opt.block2.size_exponent
445 )
447 # has been checked in assembly, just making sure
448 assert blockopt.start == len(response.payload), (
449 "Unexpected state of preassembled message"
450 )
452 blockopt = blockopt.reduced_to(response.remote.maximum_block_size_exp)
454 return self.copy(
455 payload=b"",
456 mid=None,
457 token=None,
458 block2=blockopt,
459 block1=None,
460 observe=None,
461 )
463 def _generate_next_block1_response(self):
464 """Generate a response to acknowledge incoming request block.
466 This method is used by server after receiving blockwise request from
467 client with "more" flag set."""
468 response = Message(code=CHANGED, token=self.token)
469 response.remote = self.remote
470 if (
471 self.opt.block1.block_number == 0
472 and self.opt.block1.size_exponent
473 > self.transport_tuning.DEFAULT_BLOCK_SIZE_EXP
474 ):
475 new_size_exponent = self.transport_tuning.DEFAULT_BLOCK_SIZE_EXP
476 response.opt.block1 = (0, True, new_size_exponent)
477 else:
478 response.opt.block1 = (
479 self.opt.block1.block_number,
480 True,
481 self.opt.block1.size_exponent,
482 )
483 return response
485 #
486 # the message in the context of network and addresses
487 #
489 def get_request_uri(self, *, local_is_server=False):
490 """The absolute URI this message belongs to.
492 For requests, this is composed from the options (falling back to the
493 remote). For responses, this is largely taken from the original request
494 message (so far, that could have been trackecd by the requesting
495 application as well), but -- in case of a multicast request -- with the
496 host replaced by the responder's endpoint details.
498 This implements Section 6.5 of RFC7252.
500 By default, these values are only valid on the client. To determine a
501 message's request URI on the server, set the local_is_server argument
502 to True. Note that determining the request URI on the server is brittle
503 when behind a reverse proxy, may not be possible on all platforms, and
504 can only be applied to a request message in a renderer (for the
505 response message created by the renderer will only be populated when it
506 gets transmitted; simple manual copying of the request's remote to the
507 response will not magically make this work, for in the very case where
508 the request and response's URIs differ, that would not catch the
509 difference and still report the multicast address, while the actual
510 sending address will only be populated by the operating system later).
511 """
513 # maybe this function does not belong exactly *here*, but it belongs to
514 # the results of .request(message), which is currently a message itself.
516 if hasattr(self, "_original_request_uri"):
517 # During server-side processing, a message's options may be altered
518 # to the point where its options don't accurately reflect its URI
519 # any more. In that case, this is stored.
520 return self._original_request_uri
522 if self.code.is_response():
523 refmsg = self.request
525 if refmsg.remote.is_multicast:
526 if local_is_server:
527 multicast_netloc_override = self.remote.hostinfo_local
528 else:
529 multicast_netloc_override = self.remote.hostinfo
530 else:
531 multicast_netloc_override = None
532 else:
533 refmsg = self
534 multicast_netloc_override = None
536 proxyuri = refmsg.opt.proxy_uri
537 if proxyuri is not None:
538 return proxyuri
540 scheme = refmsg.opt.proxy_scheme or refmsg.remote.scheme
541 query = refmsg.opt.uri_query or ()
542 path = refmsg.opt.uri_path
544 if multicast_netloc_override is not None:
545 netloc = multicast_netloc_override
546 else:
547 if local_is_server:
548 netloc = refmsg.remote.hostinfo_local
549 else:
550 netloc = refmsg.remote.hostinfo
552 if refmsg.opt.uri_host is not None or refmsg.opt.uri_port is not None:
553 host, port = hostportsplit(netloc)
555 host = refmsg.opt.uri_host or host
556 port = refmsg.opt.uri_port or port
558 # FIXME: This sounds like it should be part of
559 # hpostportjoin/-split
560 escaped_host = quote_nonascii(host)
562 # FIXME: "If host is not valid reg-name / IP-literal / IPv4address,
563 # fail"
565 netloc = hostportjoin(escaped_host, port)
567 # FIXME this should follow coap section 6.5 more closely
568 query = "&".join(_quote_for_query(q) for q in query)
569 path = "".join("/" + _quote_for_path(p) for p in path) or "/"
571 fragment = None
572 params = "" # are they not there at all?
574 # Eases debugging, for when they raise from urunparse you won't know
575 # which of them it was
576 assert scheme is not None, "Remote has no scheme set"
577 assert netloc is not None, "Remote has no netloc set"
578 return urllib.parse.urlunparse((scheme, netloc, path, params, query, fragment))
580 def set_request_uri(self, uri, *, set_uri_host=True):
581 """Parse a given URI into the uri_* fields of the options.
583 The remote does not get set automatically; instead, the remote data is
584 stored in the uri_host and uri_port options. That is because name resolution
585 is coupled with network specifics the protocol will know better by the
586 time the message is sent. Whatever sends the message, be it the
587 protocol itself, a proxy wrapper or an alternative transport, will know
588 how to handle the information correctly.
590 When ``set_uri_host=False`` is passed, the host/port is stored in the
591 ``unresolved_remote`` message property instead of the uri_host option;
592 as a result, the unresolved host name is not sent on the wire, which
593 breaks virtual hosts but makes message sizes smaller.
595 This implements Section 6.4 of RFC7252.
597 This raises IncompleteUrlError if URI references are passed in (instead
598 of a full URI), and MalformedUrlError if the URI specification or the
599 library's expectations of URI shapes (eg. 'coap+tcp:no-slashes') are
600 violated.
601 """
603 try:
604 parsed = urllib.parse.urlparse(uri)
605 except ValueError as e:
606 raise error.MalformedUrlError from e
608 if parsed.fragment:
609 raise error.MalformedUrlError(
610 "Fragment identifiers can not be set on a request URI"
611 )
613 if not parsed.scheme:
614 raise error.IncompleteUrlError()
616 if parsed.scheme not in coap_schemes:
617 self.opt.proxy_uri = uri
618 return
620 if not parsed.hostname:
621 raise error.MalformedUrlError("CoAP URIs need a hostname")
623 if parsed.username or parsed.password:
624 raise error.MalformedUrlError("User name and password not supported.")
626 try:
627 if parsed.path not in ("", "/"):
628 # FIXME: This tolerates incomplete % sequences.
629 self.opt.uri_path = [
630 urllib.parse.unquote(x, errors="strict")
631 for x in parsed.path.split("/")[1:]
632 ]
633 else:
634 self.opt.uri_path = []
635 if parsed.query:
636 # FIXME: This tolerates incomplete % sequences.
637 self.opt.uri_query = [
638 urllib.parse.unquote(x, errors="strict")
639 for x in parsed.query.split("&")
640 ]
641 else:
642 self.opt.uri_query = []
643 except UnicodeError as e:
644 raise error.MalformedUrlError(
645 "Percent encoded strings in CoAP URIs need to be UTF-8 encoded"
646 ) from e
648 self.remote = UndecidedRemote(parsed.scheme, parsed.netloc)
650 try:
651 _ = parsed.port
652 except ValueError as e:
653 raise error.MalformedUrlError("Port must be numeric") from e
655 is_ip_literal = parsed.netloc.startswith("[") or (
656 parsed.hostname.count(".") == 3
657 and all(c in "0123456789." for c in parsed.hostname)
658 and all(int(x) <= 255 for x in parsed.hostname.split("."))
659 )
661 if set_uri_host and not is_ip_literal:
662 try:
663 self.opt.uri_host = urllib.parse.unquote(
664 parsed.hostname, errors="strict"
665 ).translate(_ascii_lowercase)
666 except UnicodeError as e:
667 raise error.MalformedUrlError(
668 "Percent encoded strings in CoAP URI hosts need to be UTF-8 encoded"
669 ) from e
671 # Deprecated accessors to moved functionality
673 @property
674 def unresolved_remote(self):
675 return self.remote.hostinfo
677 @unresolved_remote.setter
678 def unresolved_remote(self, value):
679 # should get a big fat deprecation warning
680 if value is None:
681 self.remote = UndecidedRemote("coap", None)
682 else:
683 self.remote = UndecidedRemote("coap", value)
685 @property
686 def requested_scheme(self):
687 if self.code.is_request():
688 return self.remote.scheme
689 else:
690 return self.request.requested_scheme
692 @requested_scheme.setter
693 def requested_scheme(self, value):
694 self.remote = UndecidedRemote(value, self.remote.hostinfo)
696 @property
697 def requested_proxy_uri(self):
698 return self.request.opt.proxy_uri
700 @property
701 def requested_hostinfo(self):
702 return self.request.opt.uri_host or self.request.unresolved_remote
704 @property
705 def requested_path(self):
706 return self.request.opt.uri_path
708 @property
709 def requested_query(self):
710 return self.request.opt.uri_query
713class UndecidedRemote(
714 namedtuple("_UndecidedRemote", ("scheme", "hostinfo")), interfaces.EndpointAddress
715):
716 """Remote that is set on messages that have not been sent through any any
717 transport.
719 It describes scheme, hostname and port that were set in
720 :meth:`.set_request_uri()` or when setting a URI per Message constructor.
722 * :attr:`scheme`: The scheme string
723 * :attr:`hostinfo`: The authority component of the URI, as it would occur
724 in the URI.
726 In order to produce URIs identical to those received in responses, and
727 because the underlying types should really be binary anyway, IP addresses
728 in the hostinfo are normalized:
730 >>> UndecidedRemote("coap+tcp", "[::0001]:1234")
731 UndecidedRemote(scheme='coap+tcp', hostinfo='[::1]:1234')
732 """
734 # This is settable per instance, for other transports to pick it up.
735 maximum_block_size_exp = MAX_REGULAR_BLOCK_SIZE_EXP
737 def __new__(cls, scheme, hostinfo):
738 if "[" in hostinfo:
739 (host, port) = hostportsplit(hostinfo)
740 ip = ipaddress.ip_address(host)
741 host = str(ip)
742 hostinfo = hostportjoin(host, port)
744 return super().__new__(cls, scheme, hostinfo)
746 @classmethod
747 def from_pathless_uri(cls, uri: str) -> UndecidedRemote:
748 """Create an UndecidedRemote for a given URI that has no query, path,
749 fragment or other components not expressed in an UndecidedRemote
751 >>> from aiocoap.message import UndecidedRemote
752 >>> UndecidedRemote.from_pathless_uri("coap://localhost")
753 UndecidedRemote(scheme='coap', hostinfo='localhost')
754 """
756 parsed = urllib.parse.urlparse(uri)
758 if parsed.username or parsed.password:
759 raise ValueError("User name and password not supported.")
761 if parsed.path not in ("", "/") or parsed.query or parsed.fragment:
762 raise ValueError(
763 "Paths and query and fragment can not be set on an UndecidedRemote"
764 )
766 return cls(parsed.scheme, parsed.netloc)
769_ascii_lowercase = str.maketrans(string.ascii_uppercase, string.ascii_lowercase)
771_quote_for_path = quote_factory(unreserved + sub_delims + ":@")
772_quote_for_query = quote_factory(
773 unreserved + "".join(c for c in sub_delims if c != "&") + ":@/?"
774)
776#: Result that can be returned from a render method instead of a Message when
777#: due to defaults (eg. multicast link-format queries) or explicit
778#: configuration (eg. the No-Response option), no response should be sent at
779#: all. Note that per RFC7967 section 2, an ACK is still sent to a CON
780#: request.
781#:
782#: Depercated; set the no_response option on a regular response instead (see
783#: :meth:`.interfaces.Resource.render` for details).
784NoResponse = Sentinel("NoResponse")