fixup! tcp: pair connection with the tls peer

port/android/tcpadapter.c

@@ -28,6 +28,7 @@
 #include "oc_endpoint.h"
 #include "oc_session_events.h"
 #include "port/oc_assert.h"
+#include "port/oc_connectivity_internal.h"
 #include "port/oc_log_internal.h"
 #include "port/oc_tcp_socket_internal.h"
 #include "tcpadapter.h"
@@ -228,18 +229,16 @@ add_new_session(int sock, ip_context_t *dev, oc_endpoint_t *endpoint,
     return -1;
   }
 
-  endpoint->interface_index = (unsigned)if_index;
-
   session->dev = dev;
+  endpoint->interface_index = (unsigned)if_index;
+  if (session_id == 0) {
+    session_id = oc_tcp_get_new_session_id();
+  }
+  endpoint->session_id = session_id;
   memcpy(&session->endpoint, endpoint, sizeof(oc_endpoint_t));
   session->endpoint.next = NULL;
   session->sock = sock;
   session->csm_state = state;
-  if (session_id == 0) {
-    session->endpoint.session_id = oc_tcp_get_new_session_id();
-  } else {
-    session->endpoint.session_id = session_id;
-  }
 
   oc_list_add(session_list, session);
 
@@ -315,6 +314,22 @@ find_session_by_endpoint(const oc_endpoint_t *endpoint)
   return session;
 }
 
+static tcp_session_t *
+find_session_by_id(uint32_t session_id)
+{
+  tcp_session_t *session = oc_list_head(session_list);
+  while (session != NULL && session->endpoint.session_id != session_id) {
+    session = session->next;
+  }
+
+  if (!session) {
+    OC_DBG("could not find ongoing TCP session for session id %d", session_id);
+    return NULL;
+  }
+  OC_DBG("found TCP session for session id %d", session_id);
+  return session;
+}
+
 static tcp_session_t *
 get_ready_to_read_session(fd_set *setfds)
 {
@@ -796,6 +811,15 @@ oc_tcp_connection_state(const oc_endpoint_t *endpoint)
   return -1;
 }
 
+int
+oc_tcp_session_state(uint32_t session_id)
+{
+  if (find_session_by_id(session_id) != NULL) {
+    return OC_TCP_SOCKET_STATE_CONNECTED;
+  }
+  return -1;
+}
+
 tcp_csm_state_t
 oc_tcp_get_csm_state(const oc_endpoint_t *endpoint)
 {

port/esp32/adapter/src/tcpadapter.c

@@ -28,6 +28,7 @@
 #include "oc_endpoint.h"
 #include "oc_session_events.h"
 #include "port/oc_assert.h"
+#include "port/oc_connectivity_internal.h"
 #include "port/oc_log_internal.h"
 #include "port/oc_tcp_socket_internal.h"
 #include "tcpadapter.h"
@@ -198,18 +199,16 @@ add_new_session(int sock, ip_context_t *dev, oc_endpoint_t *endpoint,
     return -1;
   }
 
-  endpoint->interface_index = (unsigned)if_index;
-
   session->dev = dev;
+  endpoint->interface_index = (unsigned)if_index;
+  if (session_id == 0) {
+    session_id = oc_tcp_get_new_session_id();
+  }
+  endpoint->session_id = session_id;
   memcpy(&session->endpoint, endpoint, sizeof(oc_endpoint_t));
   session->endpoint.next = NULL;
   session->sock = sock;
   session->csm_state = state;
-  if (session_id == 0) {
-    session->endpoint.session_id = oc_tcp_get_new_session_id();
-  } else {
-    session->endpoint.session_id = session_id;
-  }
 
   oc_list_add(session_list, session);
 
@@ -286,6 +285,22 @@ find_session_by_endpoint(const oc_endpoint_t *endpoint)
   return session;
 }
 
+static tcp_session_t *
+find_session_by_id(uint32_t session_id)
+{
+  tcp_session_t *session = oc_list_head(session_list);
+  while (session != NULL && session->endpoint.session_id != session_id) {
+    session = session->next;
+  }
+
+  if (!session) {
+    OC_DBG("could not find ongoing TCP session for session id %d", session_id);
+    return NULL;
+  }
+  OC_DBG("found TCP session for session id %d", session_id);
+  return session;
+}
+
 static tcp_session_t *
 get_ready_to_read_session(fd_set *setfds)
 {
@@ -756,6 +771,15 @@ oc_tcp_connection_state(const oc_endpoint_t *endpoint)
   return -1;
 }
 
+int
+oc_tcp_session_state(uint32_t session_id)
+{
+  if (find_session_by_id(session_id) != NULL) {
+    return OC_TCP_SOCKET_STATE_CONNECTED;
+  }
+  return -1;
+}
+
 tcp_csm_state_t
 oc_tcp_get_csm_state(const oc_endpoint_t *endpoint)
 {

port/linux/tcpsession.c

@@ -205,6 +205,7 @@ get_interface_index(int sock)
 static void
 log_new_session(oc_endpoint_t *endpoint, int sock, bool is_connected)
 {
+  // GCOVR_EXCL_START
   oc_string64_t ep;
   const char *addr = "";
   if (oc_endpoint_to_string64(endpoint, &ep)) {
@@ -214,11 +215,13 @@ log_new_session(oc_endpoint_t *endpoint, int sock, bool is_connected)
          "connected: %d, session_id: %u",
          addr, endpoint->interface_index, sock, (int)is_connected,
          (unsigned)endpoint->session_id);
+  // GCOVR_EXCL_STOP
 }
 
 static void
 log_free_session(oc_endpoint_t *endpoint, int sock)
 {
+  // GCOVR_EXCL_START
   oc_string64_t ep;
   const char *addr = "";
   if (oc_endpoint_to_string64(endpoint, &ep)) {
@@ -227,6 +230,7 @@ log_free_session(oc_endpoint_t *endpoint, int sock)
   OC_DBG("free TCP session endpoint: %s, endpoint interface: %d, sock: %d, "
          "session_id: %u",
          addr, endpoint->interface_index, sock, (unsigned)endpoint->session_id);
+  // GCOVR_EXCL_STOP
 }
 
 #endif /* OC_DBG_IS_ENABLED */
@@ -250,16 +254,16 @@ add_new_session_locked(int sock, ip_context_t *dev, oc_endpoint_t *endpoint,
 
   session->dev = dev;
   endpoint->interface_index = (unsigned)if_index;
+  if (session_id == 0) {
+    session_id = oc_tcp_get_new_session_id();
+  }
+  endpoint->session_id = session_id;
   memcpy(&session->endpoint, endpoint, sizeof(oc_endpoint_t));
   session->endpoint.next = NULL;
+  session->endpoint.interface_index = (unsigned)if_index;
   session->sock = sock;
   session->csm_state = state;
   session->notify_session_end = true;
-  if (session_id == 0) {
-    session->endpoint.session_id = oc_tcp_get_new_session_id();
-  } else {
-    session->endpoint.session_id = session_id;
-  }
 
   oc_list_add(g_session_list, session);
 
@@ -304,8 +308,9 @@ accept_new_session_locked(ip_context_t *dev, int fd, fd_set *setfds,
 #endif /* !OC_IPV4 */
   }
 
-  if (add_new_session_locked(new_socket, dev, endpoint, /*session_id*/ 0,
-                             CSM_NONE) == NULL) {
+  tcp_session_t *session = add_new_session_locked(new_socket, dev, endpoint,
+                                                  /*session_id*/ 0, CSM_NONE);
+  if (session == NULL) {
     OC_ERR("could not record new TCP session");
     close(new_socket);
     return -1;
@@ -551,6 +556,16 @@ find_session_by_endpoint_locked(const oc_endpoint_t *endpoint)
   return session;
 }
 
+static tcp_session_t *
+find_session_by_id_locked(uint32_t session_id)
+{
+  tcp_session_t *session = oc_list_head(g_session_list);
+  while (session != NULL && session->endpoint.session_id != session_id) {
+    session = session->next;
+  }
+  return session;
+}
+
 #ifdef OC_HAS_FEATURE_TCP_ASYNC_CONNECT
 
 static tcp_session_t *
@@ -586,6 +601,17 @@ find_waiting_session_by_endpoint_locked(const oc_endpoint_t *endpoint)
   return ws;
 }
 
+static tcp_waiting_session_t *
+find_waiting_session_by_id_locked(uint32_t session_id)
+{
+  tcp_waiting_session_t *ws =
+    (tcp_waiting_session_t *)oc_list_head(g_waiting_session_list);
+  while (ws != NULL && ws->endpoint.session_id != session_id) {
+    ws = ws->next;
+  }
+  return ws;
+}
+
 static tcp_waiting_session_t *
 add_new_waiting_session_locked(int sock, ip_context_t *dev,
                                const oc_endpoint_t *endpoint,
@@ -879,6 +905,24 @@ oc_tcp_connection_state(const oc_endpoint_t *endpoint)
   return -1;
 }
 
+int
+oc_tcp_session_state(uint32_t session_id)
+{
+  pthread_mutex_lock(&g_mutex);
+  if (find_session_by_id_locked(session_id) != NULL) {
+    pthread_mutex_unlock(&g_mutex);
+    return OC_TCP_SOCKET_STATE_CONNECTED;
+  }
+#ifdef OC_HAS_FEATURE_TCP_ASYNC_CONNECT
+  if (find_waiting_session_by_id_locked(session_id) != NULL) {
+    pthread_mutex_unlock(&g_mutex);
+    return OC_TCP_SOCKET_STATE_CONNECTING;
+  }
+#endif /* OC_HAS_FEATURE_TCP_ASYNC_CONNECT */
+  pthread_mutex_unlock(&g_mutex);
+  return -1;
+}
+
 static int
 tcp_send_message(int sockfd, const oc_message_t *message)
 {
@@ -1370,6 +1414,9 @@ oc_tcp_connect(oc_endpoint_t *endpoint, on_tcp_connect_t on_tcp_connect,
 {
   oc_tcp_connect_result_t ret =
     oc_tcp_connect_to_endpoint(endpoint, on_tcp_connect, on_tcp_connect_data);
+  if (ret.session_id != 0) {
+    endpoint->session_id = ret.session_id;
+  }
   return ret.error != 0 ? ret.error : (int)ret.state;
 }
 

port/oc_connectivity_internal.h

@@ -130,6 +130,17 @@ void oc_tcp_set_connect_retry(uint8_t max_count, uint16_t timeout);
 
 #endif /* OC_HAS_FEATURE_TCP_ASYNC_CONNECT */
 
+/**
+ * @brief Get state of TCP connection for given session id.
+ *
+ * @param session_id session id
+ * @return OC_TCP_SOCKET_STATE_CONNECTED TCP connection exists and it is ongoing
+ * @return OC_TCP_SOCKET_STATE_CONNECTING TCP connection is waiting to be
+ * established
+ * @return -1 otherwise
+ */
+int oc_tcp_session_state(uint32_t session_id);
+
 #endif /* OC_TCP */
 
 #ifdef __cplusplus

port/unittest/connectivitytest.cpp

@@ -42,6 +42,8 @@
 #include <optional>
 #include <string>
 
+using namespace std::chrono_literals;
+
 static constexpr size_t kDeviceID = 0;
 
 class TestConnectivity : public testing::Test {
@@ -462,11 +464,14 @@ TEST_F(TestConnectivityWithServer, oc_tcp_update_csm_state_P)
     coap_serialize_message(&packet, msg->data, oc_message_buffer_size(msg));
 
   oc_send_buffer(msg);
+  ep.session_id = msg->endpoint.session_id;
   oc_message_unref(msg);
 #endif /* OC_HAS_FEATURE_TCP_ASYNC_CONNECT */
 
 #ifdef OC_TCP
   EXPECT_EQ(OC_TCP_SOCKET_STATE_CONNECTED, oc_tcp_connection_state(&ep));
+  ASSERT_NE(0, ep.session_id);
+  EXPECT_EQ(OC_TCP_SOCKET_STATE_CONNECTED, oc_tcp_session_state(ep.session_id));
 #endif /* OC_TCP */
 
   EXPECT_EQ(0, oc_tcp_update_csm_state(&ep, CSM_DONE));
@@ -526,7 +531,7 @@ TEST_F(TestConnectivityWithServer, oc_tcp_connect_timeout)
     return;
   }
 
-  EXPECT_EQ(OC_TCP_SOCKET_STATE_CONNECTING, ret);
+  ASSERT_EQ(OC_TCP_SOCKET_STATE_CONNECTING, ret);
 
   oc_message_t *msg = oc_allocate_message();
   memcpy(&msg->endpoint, &ep, sizeof(oc_endpoint_t));
@@ -544,6 +549,36 @@ TEST_F(TestConnectivityWithServer, oc_tcp_connect_timeout)
   restore_defaults();
 }
 
+TEST_F(TestConnectivityWithServer, oc_tcp_cleanup_waiting_session)
+{
+  auto addr = "coap+tcp://[::1]:12345";
+  oc_endpoint_t ep1 =
+    oc::endpoint::FromString(addr); // reachable address, but inactive port
+
+  oc_tcp_connect_result_t ret1 =
+    oc_tcp_connect_to_endpoint(&ep1, nullptr, nullptr);
+  ASSERT_EQ(0, ret1.error);
+  ASSERT_EQ(OC_TCP_SOCKET_STATE_CONNECTING, ret1.state);
+  ASSERT_NE(0, ret1.session_id);
+
+  // disconnect is asynchronous, we should be able to open a new session to
+  // the same endpoint
+  oc_close_session(&ep1);
+
+  oc_endpoint_t ep2 = oc::endpoint::FromString(addr);
+  oc_tcp_connect_result_t ret2 =
+    oc_tcp_connect_to_endpoint(&ep2, nullptr, nullptr);
+  ASSERT_EQ(0, ret2.error);
+  ASSERT_EQ(OC_TCP_SOCKET_STATE_CONNECTING, ret2.state);
+  ASSERT_NE(0, ret2.session_id);
+
+  oc::TestDevice::PoolEventsMsV1(20ms);
+
+  EXPECT_EQ(-1, oc_tcp_session_state(ret1.session_id));
+  EXPECT_EQ(OC_TCP_SOCKET_STATE_CONNECTING,
+            oc_tcp_session_state(ret2.session_id));
+}
+
 #endif /* __linux__ */
 
 TEST_F(TestConnectivityWithServer, oc_tcp_connect_repeat_fail)