tcp: pair connection with the tls peer

api/oc_client_api.c

@@ -967,8 +967,9 @@ oc_close_session(const oc_endpoint_t *endpoint)
 #endif /* OC_SECURITY */
   } else if (endpoint->flags & TCP) {
 #ifdef OC_TCP
-    if (oc_connectivity_end_session_v1(endpoint, false)) {
-      oc_handle_session(endpoint, OC_SESSION_DISCONNECTED);
+    oc_endpoint_t session_endpoint;
+    while (oc_connectivity_end_session_v1(endpoint, false, &session_endpoint)) {
+      oc_handle_session(&session_endpoint, OC_SESSION_DISCONNECTED);
     }
 #endif /* OC_TCP */
   }

api/oc_endpoint.c

@@ -675,6 +675,18 @@ oc_endpoint_compare_address(const oc_endpoint_t *ep1, const oc_endpoint_t *ep2)
   return -1;
 }
 
+#ifdef OC_TCP
+static int
+oc_endpoint_compare_session_ids(const oc_endpoint_t *ep1,
+                                const oc_endpoint_t *ep2)
+{
+  if (ep1->session_id == 0 || ep2->session_id == 0) {
+    return 0; // session_id == 0 means any
+  }
+  return ep1->session_id == ep2->session_id ? 0 : -1;
+}
+#endif
+
 int
 oc_endpoint_compare(const oc_endpoint_t *ep1, const oc_endpoint_t *ep2)
 {
@@ -690,19 +702,37 @@ oc_endpoint_compare(const oc_endpoint_t *ep1, const oc_endpoint_t *ep2)
   if (ep1->flags & IPV6) {
     if (memcmp(ep1->addr.ipv6.address, ep2->addr.ipv6.address, 16) == 0 &&
         ep1->addr.ipv6.port == ep2->addr.ipv6.port) {
+#ifdef OC_TCP
+      return oc_endpoint_compare_session_ids(ep1, ep2);
+#else  /* OC_TCP */
       return 0;
+#endif /* !OC_TCP */
     }
     return -1;
   }
 #ifdef OC_IPV4
   else if (ep1->flags & IPV4) {
     if (memcmp(ep1->addr.ipv4.address, ep2->addr.ipv4.address, 4) == 0 &&
         ep1->addr.ipv4.port == ep2->addr.ipv4.port) {
+#ifdef OC_TCP
+      return oc_endpoint_compare_session_ids(ep1, ep2);
+#else  /* OC_TCP */
       return 0;
+#endif /* !OC_TCP */
     }
     return -1;
   }
 #endif /* OC_IPV4 */
+
+#ifdef OC_TCP
+  else if (ep1->flags & TCP) {
+    if (memcmp(ep1->addr.ipv6.address, ep2->addr.ipv6.address, 16) == 0 &&
+        ep1->addr.ipv6.port == ep2->addr.ipv6.port) {
+      return 0;
+    }
+    return -1;
+  }
+#endif
   // TODO: Add support for other endpoint types
   return -1;
 }

api/oc_tcp.c

@@ -24,13 +24,23 @@
 #include "oc_endpoint.h"
 #include "port/oc_connectivity.h"
 #include "oc_tcp_internal.h"
+#include "util/oc_atomic.h"
 #ifdef OC_SECURITY
 #include <mbedtls/ssl.h>
 #ifdef OC_OSCORE
 #include "messaging/coap/oscore_internal.h"
 #endif /* OC_OSCORE */
 #endif /* OC_SECURITY */
 
+static OC_ATOMIC_UINT32_T g_tcp_session_id = 0;
+
+uint32_t
+oc_tcp_get_new_session_id(void)
+{
+  uint32_t v = OC_ATOMIC_INCREMENT32(g_tcp_session_id);
+  return (v == 0) ? OC_ATOMIC_INCREMENT32(g_tcp_session_id) : v;
+}
+
 #ifdef OC_HAS_FEATURE_TCP_ASYNC_CONNECT
 
 #include "port/oc_allocator_internal.h"

api/oc_tcp_internal.h

@@ -23,6 +23,7 @@
 
 #ifdef OC_TCP
 
+#include <stdint.h>
 #include "messaging/coap/constants.h"
 #include "port/oc_connectivity.h"
 #include "oc_endpoint.h"
@@ -34,6 +35,9 @@ extern "C" {
 #define OC_TCP_DEFAULT_RECEIVE_SIZE                                            \
   (COAP_TCP_DEFAULT_HEADER_LEN + COAP_TCP_MAX_EXTENDED_LENGTH_LEN)
 
+/** @brief Get new tcp session ID */
+uint32_t oc_tcp_get_new_session_id(void);
+
 #ifdef OC_HAS_FEATURE_TCP_ASYNC_CONNECT
 
 typedef struct oc_tcp_on_connect_event_s

include/oc_endpoint.h

@@ -98,14 +98,18 @@ typedef struct oc_endpoint_t
     oc_ipv4_addr_t ipv4; ///< ipv4 address
     oc_le_addr_t bt;     ///< blue tooth address
   } addr, addr_local;
-  unsigned interface_index; ///< interface index (valid intefaces are >0, 0
+  unsigned interface_index; ///< interface index (valid interfaces are >0, 0
                             ///< means no index or error)
   uint8_t priority;         ///< priority
   ocf_version_t version;    ///< ocf version
 #ifdef OC_OSCORE
   uint8_t piv[OSCORE_PIV_LEN];
   uint8_t piv_len;
 #endif /* OC_OSCORE */
+#ifdef OC_TCP
+  uint32_t session_id; ///< session id for pairing tls peer with tcp session - 0
+                       ///< means any
+#endif
 } oc_endpoint_t;
 
 #define oc_make_ipv4_endpoint(__name__, __flags__, __port__, ...)              \

port/android/ipadapter.c

@@ -1630,17 +1630,21 @@ oc_connectivity_shutdown(size_t device)
 void
 oc_connectivity_end_session(const oc_endpoint_t *endpoint)
 {
-  oc_connectivity_end_session_v1(endpoint, true);
+  while (oc_connectivity_end_session_v1(endpoint, true, NULL)) {
+    // no-op
+  }
 }
 
 bool
 oc_connectivity_end_session_v1(const oc_endpoint_t *endpoint,
-                               bool notify_session_end)
+                               bool notify_session_end,
+                               oc_endpoint_t *session_endpoint)
 {
   if (endpoint->flags & TCP) {
     ip_context_t *dev = get_ip_context_for_device(endpoint->device);
     if (dev) {
-      return oc_tcp_end_session(dev, endpoint, notify_session_end);
+      return oc_tcp_end_session(dev, endpoint, notify_session_end,
+                                session_endpoint);
     }
   }
   return false;

port/android/tcpadapter.c

@@ -235,6 +235,9 @@ add_new_session(int sock, ip_context_t *dev, oc_endpoint_t *endpoint,
   session->endpoint.next = NULL;
   session->sock = sock;
   session->csm_state = state;
+  if (session->endpoint.session_id == 0) {
+    session->endpoint.session_id = oc_tcp_get_new_session_id();
+  }
 
   oc_list_add(session_list, session);
 
@@ -463,11 +466,14 @@ oc_tcp_receive_message(ip_context_t *dev, fd_set *fds, oc_message_t *message)
 
 bool
 oc_tcp_end_session(ip_context_t *dev, const oc_endpoint_t *endpoint,
-                   bool notify_session_end)
+                   bool notify_session_end, oc_endpoint_t *session_endpoint)
 {
   pthread_mutex_lock(&dev->tcp.mutex);
   tcp_session_t *session = find_session_by_endpoint(endpoint);
   if (session) {
+    if (session_endpoint) {
+      memcpy(session_endpoint, &session->endpoint, sizeof(oc_endpoint_t));
+    }
     free_tcp_session(session, notify_session_end);
   }
   pthread_mutex_unlock(&dev->tcp.mutex);

port/android/tcpadapter.h

@@ -40,7 +40,8 @@ adapter_receive_state_t oc_tcp_receive_message(ip_context_t *dev, fd_set *fds,
                                                oc_message_t *message);
 
 bool oc_tcp_end_session(ip_context_t *dev, const oc_endpoint_t *endpoint,
-                        bool notify_session_end);
+                        bool notify_session_end,
+                        oc_endpoint_t *session_endpoint);
 
 #ifdef __cplusplus
 }

port/esp32/adapter/src/ipadapter.c

@@ -1618,17 +1618,21 @@ oc_connectivity_shutdown(size_t device)
 void
 oc_connectivity_end_session(const oc_endpoint_t *endpoint)
 {
-  oc_connectivity_end_session_v1(endpoint, true);
+  while (oc_connectivity_end_session_v1(endpoint, true, NULL)) {
+    // no-op
+  }
 }
 
 bool
 oc_connectivity_end_session_v1(const oc_endpoint_t *endpoint,
-                               bool notify_session_end)
+                               bool notify_session_end,
+                               oc_endpoint_t *session_endpoint)
 {
   if (endpoint->flags & TCP) {
     ip_context_t *dev = get_ip_context_for_device(endpoint->device);
     if (dev) {
-      return oc_tcp_end_session(dev, endpoint, notify_session_end);
+      return oc_tcp_end_session(dev, endpoint, notify_session_end,
+                                session_endpoint);
     }
   }
   return false;

port/esp32/adapter/src/tcpadapter.c

@@ -205,6 +205,9 @@ add_new_session(int sock, ip_context_t *dev, oc_endpoint_t *endpoint,
   session->endpoint.next = NULL;
   session->sock = sock;
   session->csm_state = state;
+  if (session->endpoint.session_id == 0) {
+    session->endpoint.session_id = oc_tcp_get_new_session_id();
+  }
 
   oc_list_add(session_list, session);
 
@@ -436,11 +439,14 @@ oc_tcp_receive_message(ip_context_t *dev, fd_set *fds, oc_message_t *message)
 
 bool
 oc_tcp_end_session(ip_context_t *dev, const oc_endpoint_t *endpoint,
-                   bool notify_session_end)
+                   bool notify_session_end, oc_endpoint_t *session_endpoint)
 {
   pthread_mutex_lock(&dev->tcp.mutex);
   tcp_session_t *session = find_session_by_endpoint(endpoint);
   if (session) {
+    if (session_endpoint) {
+      memcpy(session_endpoint, &session->endpoint, sizeof(oc_endpoint_t));
+    }
     free_tcp_session(session, notify_session_end);
   }
   pthread_mutex_unlock(&dev->tcp.mutex);

port/esp32/adapter/src/tcpadapter.h

@@ -40,7 +40,8 @@ adapter_receive_state_t oc_tcp_receive_message(ip_context_t *dev, fd_set *fds,
                                                oc_message_t *message);
 
 bool oc_tcp_end_session(ip_context_t *dev, const oc_endpoint_t *endpoint,
-                        bool notify_session_end);
+                        bool notify_session_end,
+                        oc_endpoint_t *session_endpoint);
 
 #ifdef __cplusplus
 }

port/linux/ipadapter.c

@@ -1579,18 +1579,21 @@ oc_connectivity_shutdown(size_t device)
 #ifdef OC_TCP
 bool
 oc_connectivity_end_session_v1(const oc_endpoint_t *endpoint,
-                               bool notify_session_end)
+                               bool notify_session_end,
+                               oc_endpoint_t *session_endpoint)
 {
   if ((endpoint->flags & TCP) != 0 &&
       oc_get_ip_context_for_device(endpoint->device) != NULL) {
-    return tcp_end_session(endpoint, notify_session_end);
+    return tcp_end_session(endpoint, notify_session_end, session_endpoint);
   }
   return false;
 }
 
 void
 oc_connectivity_end_session(const oc_endpoint_t *endpoint)
 {
-  oc_connectivity_end_session_v1(endpoint, true);
+  while (oc_connectivity_end_session_v1(endpoint, true, NULL)) {
+    // no-op
+  }
 }
 #endif /* OC_TCP */

port/linux/tcpsession.c

@@ -211,9 +211,24 @@ log_new_session(oc_endpoint_t *endpoint, int sock, bool is_connected)
     addr = oc_string(ep);
   }
   OC_DBG("new TCP session endpoint: %s, endpoint interface: %d, sock: %d, "
-         "connected: %d",
-         addr, endpoint->interface_index, sock, (int)is_connected);
+         "connected: %d, session_id: %u",
+         addr, endpoint->interface_index, sock, (int)is_connected,
+         (unsigned)endpoint->session_id);
 }
+
+static void
+log_free_session(oc_endpoint_t *endpoint, int sock)
+{
+  oc_string64_t ep;
+  const char *addr = "";
+  if (oc_endpoint_to_string64(endpoint, &ep)) {
+    addr = oc_string(ep);
+  }
+  OC_DBG("free TCP session endpoint: %s, endpoint interface: %d, sock: %d, "
+         "session_id: %u",
+         addr, endpoint->interface_index, sock, (unsigned)endpoint->session_id);
+}
+
 #endif /* OC_DBG_IS_ENABLED */
 
 static tcp_session_t *
@@ -240,6 +255,9 @@ add_new_session_locked(int sock, ip_context_t *dev, oc_endpoint_t *endpoint,
   session->sock = sock;
   session->csm_state = state;
   session->notify_session_end = true;
+  if (session->endpoint.session_id == 0) {
+    session->endpoint.session_id = oc_tcp_get_new_session_id();
+  }
 
   oc_list_add(g_session_list, session);
 
@@ -311,8 +329,9 @@ free_session_locked(tcp_session_t *session, bool signal)
     signal_network_thread(&session->dev->tcp);
   }
   close(session->sock);
-
-  OC_DBG("free TCP session(%p, fd=%d)", (void *)session, session->sock);
+#if OC_DBG_IS_ENABLED
+  log_free_session(&session->endpoint, session->sock);
+#endif /* OC_DBG_IS_ENABLED */
   oc_memb_free(&g_tcp_session_s, session);
 }
 
@@ -713,12 +732,16 @@ free_waiting_session_async_locked(tcp_waiting_session_t *ws,
 #endif /* OC_HAS_FEATURE_TCP_ASYNC_CONNECT */
 
 bool
-tcp_end_session(const oc_endpoint_t *endpoint, bool notify_session_end)
+tcp_end_session(const oc_endpoint_t *endpoint, bool notify_session_end,
+                oc_endpoint_t *session_endpoint)
 {
   pthread_mutex_lock(&g_mutex);
   tcp_session_t *s = find_session_by_endpoint_locked(endpoint);
   if (s != NULL) {
     free_session_async_locked(s, notify_session_end);
+    if (session_endpoint) {
+      memcpy(session_endpoint, &s->endpoint, sizeof(oc_endpoint_t));
+    }
     pthread_mutex_unlock(&g_mutex);
     return true;
   }
@@ -727,6 +750,9 @@ tcp_end_session(const oc_endpoint_t *endpoint, bool notify_session_end)
   tcp_waiting_session_t *ws = find_waiting_session_by_endpoint_locked(endpoint);
   if (ws != NULL) {
     free_waiting_session_async_locked(ws, notify_session_end);
+    if (session_endpoint) {
+      memcpy(session_endpoint, &ws->endpoint, sizeof(oc_endpoint_t));
+    }
     pthread_mutex_unlock(&g_mutex);
     return true;
   }

port/linux/tcpsession.h

@@ -105,7 +105,8 @@ adapter_receive_state_t tcp_receive_message(ip_context_t *dev, fd_set *fds,
  * @brief Schedule the session associated with the endpoint to be stopped and
  * deallocated (if it exists).
  */
-bool tcp_end_session(const oc_endpoint_t *endpoint, bool notify_session_end);
+bool tcp_end_session(const oc_endpoint_t *endpoint, bool notify_session_end,
+                     oc_endpoint_t *session_endpoint);
 
 /**
  * @brief Handle data received on the signal pipe.

port/oc_connectivity_internal.h

@@ -104,10 +104,12 @@ void handle_session_event_callback(const oc_endpoint_t *endpoint,
  * @param endpoint the endpoint to close the session for
  * @param notify_session_end send the notification about the disconnection
  * session.
+ * @param session_endpoint the endpoint of the session with session id
  * @return bool true if the session will be closed
  */
 bool oc_connectivity_end_session_v1(const oc_endpoint_t *endpoint,
-                                    bool notify_session_end);
+                                    bool notify_session_end,
+                                    oc_endpoint_t *session_endpoint);
 
 #ifdef __cplusplus
 }