sys/net/nanocoap: implement observe

This adds the new `nanocoap_server_observe` module that implements the
server side of the CoAP Observe option. It does require cooperation
from the resource handler to work, though.

Co-Authored-By: mguetschow <mikolai.guetschow@tu-dresden.de>
Co-authored-by: benpicco <benpicco@googlemail.com>

examples/nanocoap_server/Makefile

@@ -20,7 +20,7 @@ USEMODULE += gnrc_icmpv6_echo
 USEMODULE += nanocoap_sock
 USEMODULE += nanocoap_resources
 
-USEMODULE += xtimer
+USEMODULE += ztimer_msec
 
 # include this for nicely formatting the returned internal value
 USEMODULE += fmt
@@ -48,10 +48,12 @@ HIGH_MEMORY_BOARDS := native native64 same54-xpro mcb2388
 
 ifneq (,$(filter $(BOARD),$(HIGH_MEMORY_BOARDS)))
   # enable separate response
-  USEMODULE += nanocoap_server_separate
   USEMODULE += event_callback
+  USEMODULE += event_periodic
   USEMODULE += event_thread
   USEMODULE += event_timeout_ztimer
+  USEMODULE += nanocoap_server_observe
+  USEMODULE += nanocoap_server_separate
 
   # enable fileserver
   USEMODULE += nanocoap_fileserver

examples/nanocoap_server/coap_handler.c

@@ -11,13 +11,13 @@
 #include <string.h>
 
 #include "event/callback.h"
-#include "event/timeout.h"
+#include "event/periodic.h"
 #include "event/thread.h"
+#include "event/timeout.h"
 #include "fmt.h"
 #include "net/nanocoap.h"
 #include "net/nanocoap_sock.h"
 #include "hashes/sha256.h"
-#include "kernel_defines.h"
 
 /* internal value that can be read/written via CoAP */
 static uint8_t internal_value = 0;
@@ -59,7 +59,7 @@ static ssize_t _riot_block2_handler(coap_pkt_t *pkt, uint8_t *buf, size_t len, c
 
     bufpos += coap_put_option_ct(bufpos, 0, COAP_FORMAT_TEXT);
     bufpos += coap_opt_put_block2(bufpos, COAP_OPT_CONTENT_FORMAT, &slicer, 1);
-    *bufpos++ = 0xff;
+    *bufpos++ = COAP_PAYLOAD_MARKER;
 
     /* Add actual content */
     bufpos += coap_blockwise_put_bytes(&slicer, bufpos, block2_intro, sizeof(block2_intro)-1);
@@ -196,7 +196,7 @@ static void _send_response(void *ctx)
 
     puts("_separate_handler(): send delayed response");
     nanocoap_server_send_separate(ctx, COAP_CODE_CONTENT, COAP_TYPE_NON,
-                                response, sizeof(response));
+                                  response, sizeof(response));
 }
 
 static ssize_t _separate_handler(coap_pkt_t *pkt, uint8_t *buf, size_t len, coap_request_ctx_t *context)
@@ -234,6 +234,107 @@ NANOCOAP_RESOURCE(separate) {
 };
 #endif /* MODULE_EVENT_THREAD */
 
+#ifdef MODULE_NANOCOAP_SERVER_OBSERVE
+static ssize_t _time_handler(coap_pkt_t *pkt, uint8_t *buf, size_t len, coap_request_ctx_t *context)
+{
+    uint32_t obs;
+    bool registered = false;
+    if (coap_opt_get_uint(pkt, COAP_OPT_OBSERVE, &obs)) {
+        /* No (valid) observe option present */
+        obs = UINT32_MAX;
+    }
+
+    uint32_t now = ztimer_now(ZTIMER_MSEC);
+
+    switch (obs) {
+    case 0:
+        /* register */
+        if (nanocoap_register_observer(context, pkt) == 0) {
+            registered = true;
+        }
+        break;
+    case 1:
+        /* unregister */
+        nanocoap_unregister_observer(context, pkt);
+        break;
+    default:
+        /* No (valid) observe option present --> ignore observe and handle
+         * as regular GET */
+        break;
+    }
+
+    const size_t estimated_data_len =
+        4       /* Max Observe Option size */
+        + 1     /* payload marker */
+        + 10    /* strlen("4294967295"), 4294967295 == UINT32_MAX */
+        + 1;    /* '\n' */
+    ssize_t hdr_len = coap_build_reply(pkt, COAP_CODE_CONTENT, buf, len, estimated_data_len);
+
+    if (hdr_len < 0) {
+        /* we undo any potential registration if we cannot reply */
+        nanocoap_unregister_observer(context, pkt);
+        return len;
+    }
+
+    if (hdr_len == 0) {
+        /* no response required, probably because of no-response option matching
+         * the response class */
+        return 0;
+    }
+
+    /* coap_build_reply() is a bit goofy: It returns the size of the written
+     * header + `estiamted_data_len`, so we have to subtract it again to obtain
+     * the size of data written. */
+    uint8_t *pos = buf + hdr_len - estimated_data_len;
+
+    if (registered) {
+        uint16_t last_opt = 0;
+        pos += coap_opt_put_observe(pos, last_opt, now);
+    }
+    *pos++ = COAP_PAYLOAD_MARKER;
+    pos += fmt_u32_dec((void *)pos, now);
+    *pos++ = '\n';
+
+    return (uintptr_t)pos - (uintptr_t)buf;
+}
+
+NANOCOAP_RESOURCE(time) {
+    .path = "/time", .methods = COAP_GET, .handler = _time_handler,
+};
+
+static void _notify_observer_handler(event_t *ev)
+{
+    (void)ev;
+    uint32_t now = ztimer_now(ZTIMER_MSEC);
+    uint8_t buf[32];
+    uint8_t *pos = buf;
+    uint16_t last_opt = 0;
+    pos += coap_opt_put_observe(pos, last_opt, now);
+    *pos++ = COAP_PAYLOAD_MARKER;
+    pos += fmt_u32_dec((void *)pos, now);
+    *pos++ = '\n';
+    iolist_t data = {
+        .iol_base = buf,
+        .iol_len = (uintptr_t)pos - (uintptr_t)buf,
+    };
+
+    /* `NANOCOAP_RESOURCE(time)` expends to XFA magic adding an entry named
+     * `coap_resource_time`. */
+    nanocoap_notify_observers(&coap_resource_time, &data);
+}
+
+void setup_observe_event(void)
+{
+    static event_t ev = {
+        .handler = _notify_observer_handler
+    };
+    static event_periodic_t pev;
+
+    event_periodic_init(&pev, ZTIMER_MSEC, EVENT_PRIO_MEDIUM, &ev);
+    event_periodic_start(&pev, MS_PER_SEC);
+}
+#endif /* MODULE_NANOCOAP_SERVER_OBSERVE */
+
 /* we can also include the fileserver module */
 #ifdef MODULE_NANOCOAP_FILESERVER
 #include "net/nanocoap/fileserver.h"

examples/nanocoap_server/main.c

@@ -20,13 +20,15 @@
 #include <stdio.h>
 
 #include "net/nanocoap_sock.h"
-#include "xtimer.h"
+#include "ztimer.h"
 
 #define COAP_INBUF_SIZE (256U)
 
 #define MAIN_QUEUE_SIZE     (8)
 static msg_t _main_msg_queue[MAIN_QUEUE_SIZE];
 
+extern void setup_observe_event(void);
+
 int main(void)
 {
     puts("RIOT nanocoap example application");
@@ -35,7 +37,11 @@ int main(void)
     msg_init_queue(_main_msg_queue, MAIN_QUEUE_SIZE);
 
     puts("Waiting for address autoconfiguration...");
-    xtimer_sleep(3);
+    ztimer_sleep(ZTIMER_MSEC, 3 * MS_PER_SEC);
+
+    if (IS_USED(MODULE_NANOCOAP_SERVER_OBSERVE)) {
+        setup_observe_event();
+    }
 
     /* print network addresses */
     printf("{\"IPv6 addresses\": [\"");

sys/Makefile.dep

@@ -529,6 +529,10 @@ ifneq (,$(filter nanocoap_server_auto_init,$(USEMODULE)))
   USEMODULE += nanocoap_server
 endif
 
+ifneq (,$(filter nanocoap_server_observe,$(USEMODULE)))
+  USEMODULE += nanocoap_server_separate
+endif
+
 ifneq (,$(filter nanocoap_server_separate,$(USEMODULE)))
   USEMODULE += nanocoap_server
   USEMODULE += sock_aux_local

sys/include/net/coap.h

@@ -519,6 +519,7 @@ typedef enum {
  */
 #define COAP_OBS_REGISTER        (0)
 #define COAP_OBS_DEREGISTER      (1)
+#define COAP_OBS_MAX_VALUE_MASK  (0xffffff) /**< observe value is 24 bits */
 /** @} */
 
 /**

sys/include/net/nanocoap.h

@@ -1790,6 +1790,22 @@ static inline size_t coap_opt_put_block2_control(uint8_t *buf, uint16_t lastonum
                              (block->blknum << 4) | block->szx);
 }
 
+/**
+ * @brief   Insert an CoAP Observe Option into the buffer
+ *
+ * @param[out]  buf         Buffer to write to
+ * @param[in]   lastonum    last option number (must be < 6)
+ * @param[in]   obs         observe number to write
+ *
+ * @returns     amount of bytes written to @p buf
+ */
+static inline size_t coap_opt_put_observe(uint8_t *buf, uint16_t lastonum,
+                                          uint32_t obs)
+{
+    obs &= COAP_OBS_MAX_VALUE_MASK; /* trim obs down to 24 bit */
+    return coap_opt_put_uint(buf, lastonum, COAP_OPT_OBSERVE, obs);
+}
+
 /**
  * @brief   Encode the given string as multi-part option into buffer
  *

sys/include/net/nanocoap_sock.h

@@ -352,6 +352,99 @@ ssize_t nanocoap_server_build_separate(const nanocoap_server_response_ctx_t *ctx
 int nanocoap_server_sendv_separate(const nanocoap_server_response_ctx_t *ctx,
                                    const iolist_t *reply);
 
+/**
+ * @brief           Register an observer
+ * @param[in]       req_ctx         Request context belonging to @p req_pkt
+ * @param[in,out]   req_pkt         Request that contained the observe registration request
+ *
+ * @warning This depends on module `nanocoap_server_observe`
+ *
+ * @note    If the same endpoint already was registered on the same resource,
+ *          it will just update the token and keep the existing entry. This
+ *          way duplicate detection is not needed and we eagerly can reclaim
+ *          resources when a client lost state.
+ *
+ * @warning Preventing the same endpoint to registers more than once (using
+ *          different tokens) to the same resource deviates from RFC 7641.
+ *
+ * The deviation here is intentional. A server can receive a second registration
+ * from the same endpoint for the same resource for one of the following
+ * reasons:
+ *
+ * 1. Reaffirming the registration by using the same token again.
+ * 2. Losing state on the client side.
+ * 3. A malicious client trying to exhaust resources.
+ * 4. The same resource has different representations depending on the
+ *    request. (E.g. `/.well-known/core` can yield a wildly different response
+ *    depending on filters provided via URI-Query Options.)
+ *
+ * For case 1 updating the registration is matching what the spec mandates.
+ * For two the old registration will not be of value for the client, and
+ * overwriting it makes more efficient use of network bandwidth and RAM.
+ * For 3 the deviation forces the adversary to send observe requests from
+ * different ports to exhaust resources, which is a very minor improvement.
+ * For 4 the deviation is a problem. However, the observe API does not allow to
+ * send out different notification messages for the same resource anyway, so
+ * case 4 cannot occur here.
+ *
+ * @retval  0           Success
+ * @retval  -ENOMEM     Not enough resources to register another observer
+ * @retval  <0          Negative errno code indicating error
+ */
+int nanocoap_register_observer(const coap_request_ctx_t *req_ctx, coap_pkt_t *req_pkt);
+
+/**
+ * @brief   Unregister an observer
+ * @param   req_ctx         Request context belonging to @p req_pkt
+ * @param   req_pkt         Received request for unregistration
+ *
+ * @warning This depends on module `nanocoap_server_observe`
+ *
+ * @note    It is safe to call this multiple times, e.g. duplicate detection
+ *          is not needed for this.
+ */
+void nanocoap_unregister_observer(const coap_request_ctx_t *req_ctx,
+                                  const coap_pkt_t *req_pkt);
+
+/**
+ * @brief   Unregister a stale observation due to a reset message received
+ * @param[in]   ep      Endpoint to wipe from the observer list
+ * @param[in]   msg_id  Message ID of the notification send.
+ */
+void nanocoap_unregister_observer_due_to_reset(const sock_udp_ep_t *ep,
+                                               uint16_t msg_id);
+
+/**
+ * @brief   Notify all currently registered observers of the given resource
+ *
+ * @param[in]   res     Resource to send updates for
+ * @param[in]   iol     I/O list containing the CoAP Options, payload marker,
+ *                      and payload of the update to send up
+ *
+ * @pre     @p iol contains everything but the CoAP header needed to send out.
+ *          This will at least be a CoAP observe option, a payload marker,
+ *          and a payload
+ *
+ * @post    For each registered observer a CoAP packet header is generated and
+ *          the concatenation of that header and the provided list is sent
+ */
+void nanocoap_notify_observers(const coap_resource_t *res, const iolist_t *iol);
+
+/**
+ * @brief   Build and send notification to observers registered to a specific
+ *          resource.
+ *
+ * @note    Use @ref nanocoap_notify_observers for more control (such
+ *          as adding custom options) over the notification(s) to send.
+ *
+ * @param[in]   res         Resource to send updates for
+ * @param[in]   obs         24-bit number to add as observe option
+ * @param[in]   payload     Payload to send out
+ * @param[in]   payload_len Length of @p payload in bytes
+ */
+void nanocoap_notify_observers_simple(const coap_resource_t *res, uint32_t obs,
+                                      const void *payload, size_t payload_len);
+
 /**
  * @brief   Get next consecutive message ID for use when building a new
  *          CoAP request.

sys/net/application_layer/nanocoap/nanocoap.c

@@ -28,6 +28,7 @@
 
 #include "bitarithm.h"
 #include "net/nanocoap.h"
+#include "net/nanocoap_sock.h"
 
 #define ENABLE_DEBUG 0
 #include "debug.h"
@@ -494,6 +495,11 @@ ssize_t coap_handle_req(coap_pkt_t *pkt, uint8_t *resp_buf, unsigned resp_buf_le
 {
     assert(ctx);
 
+    if (IS_USED(MODULE_NANOCOAP_SERVER_OBSERVE) && (coap_get_type(pkt) == COAP_TYPE_RST)) {
+        nanocoap_unregister_observer_due_to_reset(coap_request_ctx_get_remote_udp(ctx),
+                                                  coap_get_id(pkt));
+    }
+
     if (coap_get_code_class(pkt) != COAP_REQ) {
         DEBUG("coap_handle_req(): not a request.\n");
         return -EBADMSG;

sys/net/application_layer/nanocoap/sock.c

@@ -25,6 +25,7 @@
 #include <string.h>
 #include <stdio.h>
 
+#include "container.h"
 #include "net/credman.h"
 #include "net/nanocoap.h"
 #include "net/nanocoap_sock.h"
@@ -48,6 +49,10 @@
 #  define CONFIG_NANOCOAP_DTLS_HANDSHAKE_BUF_SIZE (160)
 #endif
 
+#ifndef CONFIG_NANOCOAP_MAX_OBSERVERS
+#  define CONFIG_NANOCOAP_MAX_OBSERVERS 4
+#endif
+
 enum {
     STATE_REQUEST_SEND,     /**< request was just sent or will be sent again */
     STATE_STOP_RETRANSMIT,  /**< stop retransmissions due to a matching empty ACK */
@@ -64,6 +69,35 @@ typedef struct {
 #endif
 } _block_ctx_t;
 
+/**
+ * @brief   Structure to track the state of an observation
+ */
+typedef struct {
+    /**
+     * @brief   Context needed to build notifications (e.g. Token, endpoint
+     *          to send to)
+     *
+     * @details To safe ROM, we reuse the separate response code to also
+     *          send notifications, as the functionality is almost identical.
+     */
+    nanocoap_server_response_ctx_t response;
+    /**
+     * @brief   The resource the client has subscribed to
+     *
+     * @details This is `NULL` when the slot is free
+     */
+    const coap_resource_t *resource;
+    /**
+     * @brief   Message ID used in the last notification
+     */
+    uint16_t msg_id;
+} _observer_t;
+
+#if MODULE_NANOCOAP_SERVER_OBSERVE
+static _observer_t _observer_pool[CONFIG_NANOCOAP_MAX_OBSERVERS];
+static mutex_t _observer_pool_lock;
+#endif
+
 int nanocoap_sock_dtls_connect(nanocoap_sock_t *sock, sock_udp_ep_t *local,
                                const sock_udp_ep_t *remote, credman_tag_t tag)
 {
@@ -1150,7 +1184,13 @@ int nanocoap_server_sendv_separate(const nanocoap_server_response_ctx_t *ctx,
     if (!sock_udp_ep_is_multicast(&ctx->local)) {
         aux_out_ptr = &aux_out;
     }
-    return sock_udp_sendv_aux(NULL, reply, &ctx->remote, aux_out_ptr);
+    ssize_t retval = sock_udp_sendv_aux(NULL, reply, &ctx->remote, aux_out_ptr);
+
+    if (retval < 0) {
+        return retval;
+    }
+
+    return 0;
 }
 
 int nanocoap_server_send_separate(const nanocoap_server_response_ctx_t *ctx,
@@ -1184,3 +1224,120 @@ int nanocoap_server_send_separate(const nanocoap_server_response_ctx_t *ctx,
     return nanocoap_server_sendv_separate(ctx, &head);
 }
 #endif
+
+#if MODULE_NANOCOAP_SERVER_OBSERVE
+int nanocoap_register_observer(const coap_request_ctx_t *req_ctx, coap_pkt_t *req_pkt)
+{
+    mutex_lock(&_observer_pool_lock);
+
+    _observer_t *free = NULL;
+    const coap_resource_t *resource = req_ctx->resource;
+
+    for (size_t i = 0; i < CONFIG_NANOCOAP_MAX_OBSERVERS; i++) {
+        if (_observer_pool[i].resource == NULL) {
+            free = &_observer_pool[i];
+        }
+        if ((_observer_pool[i].resource == resource)
+                && sock_udp_ep_equal(&_observer_pool[i].response.remote,
+                                     coap_request_ctx_get_remote_udp(req_ctx)))
+        {
+            /* Deviation from the standard: Subscribing twice makes no
+             * sense with our CoAP implementation, so either this is a
+             * reaffirmation of an existing subscription (same token) or the
+             * client lost state (different token). We just update the
+             * subscription in either case */
+            DEBUG("nanocoap: observe slot %" PRIuSIZE " reused\n", i);
+            uint8_t tkl = coap_get_token_len(req_pkt);
+            _observer_pool[i].response.tkl = tkl;
+            memcpy(_observer_pool[i].response.token, coap_get_token(req_pkt), tkl);
+            mutex_unlock(&_observer_pool_lock);
+            return 0;
+        }
+    }
+
+    if (!free) {
+        DEBUG_PUTS("nanocoap: observe registration failed, no free slot");
+        mutex_unlock(&_observer_pool_lock);
+        return -ENOMEM;
+    }
+
+    int retval = nanocoap_server_prepare_separate(&free->response, req_pkt, req_ctx);
+    if (retval) {
+        DEBUG("nanocoap: observe registration failed: %d\n", retval);
+        mutex_unlock(&_observer_pool_lock);
+        return retval;
+    }
+    free->resource = req_ctx->resource;
+    free->msg_id = random_uint32();
+    mutex_unlock(&_observer_pool_lock);
+    DEBUG("nanocoap: new observe registration at slot %" PRIuSIZE "\n",
+          index_of(_observer_pool, free));
+    return 0;
+}
+
+void nanocoap_unregister_observer(const coap_request_ctx_t *req_ctx,
+                                  const coap_pkt_t *req_pkt)
+{
+    mutex_lock(&_observer_pool_lock);
+    for (size_t i = 0; i < CONFIG_NANOCOAP_MAX_OBSERVERS; i++) {
+        if ((_observer_pool[i].resource == req_ctx->resource)
+                && (_observer_pool[i].response.tkl == coap_get_token_len(req_pkt))
+                && !memcmp(_observer_pool[i].response.token, coap_get_token(req_pkt),
+                           _observer_pool[i].response.tkl)
+                && sock_udp_ep_equal(&_observer_pool[i].response.remote, coap_request_ctx_get_remote_udp(req_ctx))) {
+            DEBUG("nanocoap: observer at index %" PRIuSIZE " unregistered\n", i);
+            _observer_pool[i].resource = NULL;
+        }
+    }
+    mutex_unlock(&_observer_pool_lock);
+}
+
+void nanocoap_unregister_observer_due_to_reset(const sock_udp_ep_t *ep,
+                                               uint16_t msg_id)
+{
+    mutex_lock(&_observer_pool_lock);
+    for (size_t i = 0; i < CONFIG_NANOCOAP_MAX_OBSERVERS; i++) {
+        if ((_observer_pool[i].resource != NULL)
+                && (_observer_pool[i].msg_id == msg_id)
+                && sock_udp_ep_equal(&_observer_pool[i].response.remote, ep)) {
+            DEBUG("nanocoap: observer at index %" PRIuSIZE " unregistered due to RST\n", i);
+            _observer_pool[i].resource = NULL;
+            return;
+        }
+    }
+    mutex_unlock(&_observer_pool_lock);
+}
+
+void nanocoap_notify_observers(const coap_resource_t *res, const iolist_t *iol)
+{
+    mutex_lock(&_observer_pool_lock);
+    for (size_t i = 0; i < CONFIG_NANOCOAP_MAX_OBSERVERS; i++) {
+        if (_observer_pool[i].resource == res) {
+            uint8_t rbuf[sizeof(coap_hdr_t) + COAP_TOKEN_LENGTH_MAX + 1];
+
+            ssize_t hdr_len = nanocoap_server_build_separate(&_observer_pool[i].response, rbuf, sizeof(rbuf),
+                                                             COAP_CODE_CONTENT, COAP_TYPE_NON,
+                                                             ++_observer_pool[i].msg_id);
+            if (hdr_len < 0) {
+                /* no need to keep the observer in the pool, if we cannot
+                 * send anyway */
+                _observer_pool[i].resource = NULL;
+                continue;
+            }
+
+            const iolist_t msg = {
+                .iol_base = rbuf,
+                .iol_len = hdr_len,
+                .iol_next = (iolist_t *)iol
+            };
+
+            if (nanocoap_server_sendv_separate(&_observer_pool[i].response, &msg)) {
+                /* no need to keep the observer in the pool, if we cannot
+                 * send anyway */
+                _observer_pool[i].resource = NULL;
+            }
+        }
+    }
+    mutex_unlock(&_observer_pool_lock);
+}
+#endif