src/transport/raw/TCP.cpp

   1 /*
   2  *
   3  *    Copyright (c) 2020-2021 Project CHIP Authors
   4  *    Copyright (c) 2013-2017 Nest Labs, Inc.
   5  *    All rights reserved.
   6  *
   7  *    Licensed under the Apache License, Version 2.0 (the "License");
   8  *    you may not use this file except in compliance with the License.
   9  *    You may obtain a copy of the License at
  10  *
  11  *        http://www.apache.org/licenses/LICENSE-2.0
  12  *
  13  *    Unless required by applicable law or agreed to in writing, software
  14  *    distributed under the License is distributed on an "AS IS" BASIS,
  15  *    WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  16  *    See the License for the specific language governing permissions and
  17  *    limitations under the License.
  18  */
  19
  20 /**
  21  *    @file
  22  *      This file implements the CHIP Transport object that maintains TCP connections
  23  *      to peers. Handles both establishing new connections and accepting peer connection
  24  *      requests.
  25  */
  26 #include <transport/raw/TCP.h>
  27
  28 #include <core/CHIPEncoding.h>
  29 #include <support/CodeUtils.h>
  30 #include <support/logging/CHIPLogging.h>
  31 #include <transport/raw/MessageHeader.h>
  32
  33 #include <inttypes.h>
  34
  35 namespace chip {
  36 namespace Transport {
  37 namespace {
  38
  39 using namespace chip::Encoding;
  40
  41 // Packets start with a 16-bit size
  42 constexpr size_t kPacketSizeBytes = 2;
  43
  44 // TODO: Actual limit may be lower (spec issue #2119)
  45 constexpr uint16_t kMaxMessageSize = static_cast<uint16_t>(System::PacketBuffer::kMaxSizeWithoutReserve - kPacketSizeBytes);
  46
  47 constexpr int kListenBacklogSize = 2;
  48
  49 } // namespace
  50
  51 TCPBase::~TCPBase()
  52 {
  53     if (mListenSocket != nullptr)
  54     {
  55         // endpoint is only non null if it is initialized and listening
  56         mListenSocket->Free();
  57         mListenSocket = nullptr;
  58     }
  59
  60     CloseActiveConnections();
  61
  62     for (size_t i = 0; i < mPendingPacketsSize; i++)
  63     {
  64         mPendingPackets[i].packetBuffer = nullptr;
  65     }
  66 }
  67
  68 void TCPBase::CloseActiveConnections()
  69 {
  70     for (size_t i = 0; i < mActiveConnectionsSize; i++)
  71     {
  72         if (mActiveConnections[i].InUse())
  73         {
  74             mActiveConnections[i].Free();
  75             mUsedEndPointCount--;
  76         }
  77     }
  78 }
  79
  80 CHIP_ERROR TCPBase::Init(TcpListenParameters & params)
  81 {
  82     CHIP_ERROR err = CHIP_NO_ERROR;
  83
  84     VerifyOrExit(mState == State::kNotReady, err = CHIP_ERROR_INCORRECT_STATE);
  85
  86 #if INET_CONFIG_ENABLE_TCP_ENDPOINT
  87     err = params.GetInetLayer()->NewTCPEndPoint(&mListenSocket);
  88 #else
  89     err = CHIP_SYSTEM_ERROR_NOT_SUPPORTED;
  90 #endif
  91     SuccessOrExit(err);
  92
  93     err = mListenSocket->Bind(params.GetAddressType(), Inet::IPAddress::Any, params.GetListenPort(), params.GetInterfaceId());
  94     SuccessOrExit(err);
  95
  96     err = mListenSocket->Listen(kListenBacklogSize);
  97     SuccessOrExit(err);
  98
  99     mListenSocket->AppState             = reinterpret_cast<void *>(this);
 100     mListenSocket->OnDataReceived       = OnTcpReceive;
 101     mListenSocket->OnConnectComplete    = OnConnectionComplete;
 102     mListenSocket->OnConnectionClosed   = OnConnectionClosed;
 103     mListenSocket->OnConnectionReceived = OnConnectionReceived;
 104     mListenSocket->OnAcceptError        = OnAcceptError;
 105     mEndpointType                       = params.GetAddressType();
 106
 107     mState = State::kInitialized;
 108
 109 exit:
 110     if (err != CHIP_NO_ERROR)
 111     {
 112         ChipLogError(Inet, "Failed to initialize TCP transport: %s", ErrorStr(err));
 113         if (mListenSocket)
 114         {
 115             mListenSocket->Free();
 116             mListenSocket = nullptr;
 117         }
 118     }
 119
 120     return err;
 121 }
 122
 123 TCPBase::ActiveConnectionState * TCPBase::FindActiveConnection(const PeerAddress & address)
 124 {
 125     if (address.GetTransportType() != Type::kTcp)
 126     {
 127         return nullptr;
 128     }
 129
 130     for (size_t i = 0; i < mActiveConnectionsSize; i++)
 131     {
 132         if (!mActiveConnections[i].InUse())
 133         {
 134             continue;
 135         }
 136         Inet::IPAddress addr;
 137         uint16_t port;
 138         mActiveConnections[i].mEndPoint->GetPeerInfo(&addr, &port);
 139
 140         if ((addr == address.GetIPAddress()) && (port == address.GetPort()))
 141         {
 142             return &mActiveConnections[i];
 143         }
 144     }
 145
 146     return nullptr;
 147 }
 148
 149 TCPBase::ActiveConnectionState * TCPBase::FindActiveConnection(const Inet::TCPEndPoint * endPoint)
 150 {
 151     for (size_t i = 0; i < mActiveConnectionsSize; i++)
 152     {
 153         if (mActiveConnections[i].mEndPoint == endPoint)
 154         {
 155             return &mActiveConnections[i];
 156         }
 157     }
 158     return nullptr;
 159 }
 160
 161 CHIP_ERROR TCPBase::SendMessage(const PacketHeader & header, const Transport::PeerAddress & address,
 162                                 System::PacketBufferHandle msgBuf)
 163 {
 164     // Sent buffer data format is:
 165     //    - packet size as a uint16_t (inludes size of header and actual data)
 166     //    - header
 167     //    - actual data
 168     const size_t prefixSize = header.EncodeSizeBytes() + kPacketSizeBytes;
 169
 170     VerifyOrReturnError(address.GetTransportType() == Type::kTcp, CHIP_ERROR_INVALID_ARGUMENT);
 171     VerifyOrReturnError(mState == State::kInitialized, CHIP_ERROR_INCORRECT_STATE);
 172     VerifyOrReturnError(prefixSize + msgBuf->DataLength() <= std::numeric_limits<uint16_t>::max(), CHIP_ERROR_INVALID_ARGUMENT);
 173
 174     // The check above about prefixSize + msgBuf->DataLength() means prefixSize
 175     // definitely fits in uint16_t.
 176     VerifyOrReturnError(msgBuf->EnsureReservedSize(static_cast<uint16_t>(prefixSize)), CHIP_ERROR_NO_MEMORY);
 177
 178     msgBuf->SetStart(msgBuf->Start() - prefixSize);
 179
 180     // Length is actual data, without considering the length bytes themselves
 181     VerifyOrReturnError(msgBuf->DataLength() >= kPacketSizeBytes, CHIP_ERROR_INTERNAL);
 182
 183     uint8_t * output = msgBuf->Start();
 184     LittleEndian::Write16(output, static_cast<uint16_t>(msgBuf->DataLength() - kPacketSizeBytes));
 185
 186     uint16_t actualEncodedHeaderSize;
 187     ReturnErrorOnFailure(header.Encode(output, msgBuf->DataLength(), &actualEncodedHeaderSize));
 188
 189     // header encoding has to match space that we allocated
 190     VerifyOrReturnError(prefixSize == actualEncodedHeaderSize + kPacketSizeBytes, CHIP_ERROR_INTERNAL);
 191
 192     // Reuse existing connection if one exists, otherwise a new one
 193     // will be established
 194     ActiveConnectionState * connection = FindActiveConnection(address);
 195
 196     if (connection != nullptr)
 197     {
 198         return connection->mEndPoint->Send(std::move(msgBuf));
 199     }
 200     else
 201     {
 202         return SendAfterConnect(address, std::move(msgBuf));
 203     }
 204 }
 205
 206 CHIP_ERROR TCPBase::SendAfterConnect(const PeerAddress & addr, System::PacketBufferHandle msg)
 207 {
 208     // This will initiate a connection to the specified peer
 209     CHIP_ERROR err               = CHIP_NO_ERROR;
 210     PendingPacket * packet       = nullptr;
 211     bool alreadyConnecting       = false;
 212     Inet::TCPEndPoint * endPoint = nullptr;
 213
 214     // Iterate through the ENTIRE array. If a pending packet for
 215     // the address already exists, this means a connection is pending and
 216     // does NOT need to be re-established.
 217     for (size_t i = 0; i < mPendingPacketsSize; i++)
 218     {
 219         if (mPendingPackets[i].packetBuffer.IsNull())
 220         {
 221             if (packet == nullptr)
 222             {
 223                 // found a slot to store the packet into
 224                 packet = mPendingPackets + i;
 225             }
 226         }
 227         else if (mPendingPackets[i].peerAddress == addr)
 228         {
 229             // same destination exists.
 230             alreadyConnecting = true;
 231
 232             // ensure packets are ORDERED
 233             if (packet != nullptr)
 234             {
 235                 packet->peerAddress  = addr;
 236                 packet->packetBuffer = std::move(mPendingPackets[i].packetBuffer);
 237                 packet               = mPendingPackets + i;
 238             }
 239         }
 240     }
 241
 242     VerifyOrExit(packet != nullptr, err = CHIP_ERROR_NO_MEMORY);
 243
 244     // If already connecting, buffer was just enqueued for more sending
 245     VerifyOrExit(!alreadyConnecting, err = CHIP_NO_ERROR);
 246
 247     // Ensures sufficient active connections size exist
 248     VerifyOrExit(mUsedEndPointCount < mActiveConnectionsSize, err = CHIP_ERROR_NO_MEMORY);
 249
 250 #if INET_CONFIG_ENABLE_TCP_ENDPOINT
 251     err = mListenSocket->Layer().NewTCPEndPoint(&endPoint);
 252 #else
 253     err = CHIP_SYSTEM_ERROR_NOT_SUPPORTED;
 254 #endif
 255     SuccessOrExit(err);
 256
 257     endPoint->AppState             = reinterpret_cast<void *>(this);
 258     endPoint->OnDataReceived       = OnTcpReceive;
 259     endPoint->OnConnectComplete    = OnConnectionComplete;
 260     endPoint->OnConnectionClosed   = OnConnectionClosed;
 261     endPoint->OnConnectionReceived = OnConnectionReceived;
 262     endPoint->OnAcceptError        = OnAcceptError;
 263     endPoint->OnPeerClose          = OnPeerClosed;
 264
 265     err = endPoint->Connect(addr.GetIPAddress(), addr.GetPort(), addr.GetInterface());
 266     SuccessOrExit(err);
 267
 268     // enqueue the packet once the connection succeeds
 269     packet->peerAddress  = addr;
 270     packet->packetBuffer = std::move(msg);
 271     mUsedEndPointCount++;
 272
 273 exit:
 274     if (err != CHIP_NO_ERROR)
 275     {
 276         if (endPoint != nullptr)
 277         {
 278             endPoint->Free();
 279         }
 280     }
 281     return err;
 282 }
 283
 284 CHIP_ERROR TCPBase::ProcessReceivedBuffer(Inet::TCPEndPoint * endPoint, const PeerAddress & peerAddress,
 285                                           System::PacketBufferHandle buffer)
 286 {
 287     ActiveConnectionState * state = FindActiveConnection(endPoint);
 288     VerifyOrReturnError(state != nullptr, CHIP_ERROR_INTERNAL);
 289     state->mReceived.AddToEnd(std::move(buffer));
 290
 291     while (!state->mReceived.IsNull())
 292     {
 293         uint8_t messageSizeBuf[kPacketSizeBytes];
 294         CHIP_ERROR err = state->mReceived->Read(messageSizeBuf);
 295         if (err == CHIP_ERROR_BUFFER_TOO_SMALL)
 296         {
 297             // We don't have enough data to read the message size. Wait until there's more.
 298             return CHIP_NO_ERROR;
 299         }
 300         else if (err != CHIP_NO_ERROR)
 301         {
 302             return err;
 303         }
 304         uint16_t messageSize = LittleEndian::Get16(messageSizeBuf);
 305         if (messageSize >= kMaxMessageSize)
 306         {
 307             // This message is too long for upper layers.
 308             return CHIP_ERROR_MESSAGE_TOO_LONG;
 309         }
 310         // The subtraction will not underflow because we successfully read kPacketSizeBytes.
 311         if (messageSize > (state->mReceived->TotalLength() - kPacketSizeBytes))
 312         {
 313             // We have not yet received the complete message.
 314             return CHIP_NO_ERROR;
 315         }
 316         state->mReceived.Consume(kPacketSizeBytes);
 317         ReturnErrorOnFailure(ProcessSingleMessage(peerAddress, state, messageSize));
 318     }
 319
 320     return CHIP_NO_ERROR;
 321 }
 322
 323 CHIP_ERROR TCPBase::ProcessSingleMessage(const PeerAddress & peerAddress, ActiveConnectionState * state, uint16_t messageSize)
 324 {
 325     // We enter with `state->mReceived` containing at least one full message, perhaps in a chain.
 326     // `state->mReceived->Start()` currently points to the message data.
 327     // On exit, `state->mReceived` will have had `messageSize` bytes consumed, no matter what.
 328     System::PacketBufferHandle message;
 329     if (state->mReceived->DataLength() == messageSize)
 330     {
 331         // In this case, the head packet buffer contains exactly the message.
 332         // This is common because typical messages fit in a network packet, and are delivered as such.
 333         // Peel off the head to pass upstream, which effectively consumes it from `state->mReceived`.
 334         message = state->mReceived.PopHead();
 335     }
 336     else
 337     {
 338         // The message is either longer or shorter than the head buffer.
 339         // In either case, copy the message to a fresh linear buffer to pass upstream. We always copy, rather than provide
 340         // a shared reference to the current buffer, in case upper layers manipulate the buffer in ways that would affect
 341         // our use, e.g. chaining it elsewhere or reusing space beyond the current message.
 342         message = System::PacketBufferHandle::New(messageSize, 0);
 343         if (message.IsNull())
 344         {
 345             return CHIP_ERROR_NO_MEMORY;
 346         }
 347         CHIP_ERROR err = state->mReceived->Read(message->Start(), messageSize);
 348         state->mReceived.Consume(messageSize);
 349         ReturnErrorOnFailure(err);
 350         message->SetDataLength(messageSize);
 351     }
 352
 353     PacketHeader header;
 354     ReturnErrorOnFailure(header.DecodeAndConsume(message));
 355
 356     HandleMessageReceived(header, peerAddress, std::move(message));
 357     return CHIP_NO_ERROR;
 358 }
 359
 360 INET_ERROR TCPBase::OnTcpReceive(Inet::TCPEndPoint * endPoint, System::PacketBufferHandle buffer)
 361 {
 362     Inet::IPAddress ipAddress;
 363     uint16_t port;
 364
 365     endPoint->GetPeerInfo(&ipAddress, &port);
 366     PeerAddress peerAddress = PeerAddress::TCP(ipAddress, port);
 367
 368     TCPBase * tcp  = reinterpret_cast<TCPBase *>(endPoint->AppState);
 369     CHIP_ERROR err = tcp->ProcessReceivedBuffer(endPoint, peerAddress, std::move(buffer));
 370
 371     if (err != CHIP_NO_ERROR)
 372     {
 373         // Connection could need to be closed at this point
 374         ChipLogError(Inet, "Failed to accept received TCP message: %s", ErrorStr(err));
 375         return INET_ERROR_UNEXPECTED_EVENT;
 376     }
 377     return INET_NO_ERROR;
 378 }
 379
 380 void TCPBase::OnConnectionComplete(Inet::TCPEndPoint * endPoint, INET_ERROR inetErr)
 381 {
 382     CHIP_ERROR err          = CHIP_NO_ERROR;
 383     bool foundPendingPacket = false;
 384     TCPBase * tcp           = reinterpret_cast<TCPBase *>(endPoint->AppState);
 385     Inet::IPAddress ipAddress;
 386     uint16_t port;
 387
 388     endPoint->GetPeerInfo(&ipAddress, &port);
 389     PeerAddress addr = PeerAddress::TCP(ipAddress, port);
 390
 391     // Send any pending packets
 392     for (size_t i = 0; i < tcp->mPendingPacketsSize; i++)
 393     {
 394         if ((tcp->mPendingPackets[i].peerAddress != addr) || (tcp->mPendingPackets[i].packetBuffer.IsNull()))
 395         {
 396             continue;
 397         }
 398         foundPendingPacket = true;
 399
 400         System::PacketBufferHandle buffer   = std::move(tcp->mPendingPackets[i].packetBuffer);
 401         tcp->mPendingPackets[i].peerAddress = PeerAddress::Uninitialized();
 402
 403         if ((inetErr == CHIP_NO_ERROR) && (err == CHIP_NO_ERROR))
 404         {
 405             err = endPoint->Send(std::move(buffer));
 406         }
 407     }
 408
 409     if (err == CHIP_NO_ERROR)
 410     {
 411         err = inetErr;
 412     }
 413
 414     if (!foundPendingPacket && (err == CHIP_NO_ERROR))
 415     {
 416         // Force a close: new connections are only expected when a
 417         // new buffer is being sent.
 418         ChipLogError(Inet, "Connection accepted without pending buffers");
 419         err = CHIP_ERROR_CONNECTION_CLOSED_UNEXPECTEDLY;
 420     }
 421
 422     // cleanup packets or mark as free
 423     if (err != CHIP_NO_ERROR)
 424     {
 425         ChipLogError(Inet, "Connection complete encountered an error: %s", ErrorStr(err));
 426         endPoint->Free();
 427         tcp->mUsedEndPointCount--;
 428     }
 429     else
 430     {
 431         bool connectionStored = false;
 432         for (size_t i = 0; i < tcp->mActiveConnectionsSize; i++)
 433         {
 434             if (!tcp->mActiveConnections[i].InUse())
 435             {
 436                 tcp->mActiveConnections[i].Init(endPoint);
 437                 connectionStored = true;
 438                 break;
 439             }
 440         }
 441
 442         // since we track end points counts, we always expect to store the
 443         // connection.
 444         if (!connectionStored)
 445         {
 446             endPoint->Free();
 447             ChipLogError(Inet, "Internal logic error: insufficient space to store active connection");
 448         }
 449     }
 450 }
 451
 452 void TCPBase::OnConnectionClosed(Inet::TCPEndPoint * endPoint, INET_ERROR err)
 453 {
 454     TCPBase * tcp = reinterpret_cast<TCPBase *>(endPoint->AppState);
 455
 456     ChipLogProgress(Inet, "Connection closed.");
 457
 458     for (size_t i = 0; i < tcp->mActiveConnectionsSize; i++)
 459     {
 460         if (tcp->mActiveConnections[i].mEndPoint == endPoint)
 461         {
 462             ChipLogProgress(Inet, "Freeing closed connection.");
 463             tcp->mActiveConnections[i].Free();
 464             tcp->mUsedEndPointCount--;
 465         }
 466     }
 467 }
 468
 469 void TCPBase::OnConnectionReceived(Inet::TCPEndPoint * listenEndPoint, Inet::TCPEndPoint * endPoint,
 470                                    const Inet::IPAddress & peerAddress, uint16_t peerPort)
 471 {
 472     TCPBase * tcp = reinterpret_cast<TCPBase *>(listenEndPoint->AppState);
 473
 474     if (tcp->mUsedEndPointCount < tcp->mActiveConnectionsSize)
 475     {
 476         // have space to use one more (even if considering pending connections)
 477         for (size_t i = 0; i < tcp->mActiveConnectionsSize; i++)
 478         {
 479             if (!tcp->mActiveConnections[i].InUse())
 480             {
 481                 tcp->mActiveConnections[i].Init(endPoint);
 482                 break;
 483             }
 484         }
 485
 486         endPoint->AppState             = listenEndPoint->AppState;
 487         endPoint->OnDataReceived       = OnTcpReceive;
 488         endPoint->OnConnectComplete    = OnConnectionComplete;
 489         endPoint->OnConnectionClosed   = OnConnectionClosed;
 490         endPoint->OnConnectionReceived = OnConnectionReceived;
 491         endPoint->OnAcceptError        = OnAcceptError;
 492         endPoint->OnPeerClose          = OnPeerClosed;
 493     }
 494     else
 495     {
 496         ChipLogError(Inet, "Insufficient connection space to accept new connections");
 497         endPoint->Free();
 498     }
 499 }
 500
 501 void TCPBase::OnAcceptError(Inet::TCPEndPoint * endPoint, INET_ERROR err)
 502 {
 503     ChipLogError(Inet, "Accept error: %s", ErrorStr(err));
 504 }
 505
 506 void TCPBase::Disconnect(const PeerAddress & address)
 507 {
 508     // Closes an existing connection
 509     for (size_t i = 0; i < mActiveConnectionsSize; i++)
 510     {
 511         if (mActiveConnections[i].InUse())
 512         {
 513             Inet::IPAddress ipAddress;
 514             uint16_t port;
 515
 516             mActiveConnections[i].mEndPoint->GetPeerInfo(&ipAddress, &port);
 517             if (address == PeerAddress::TCP(ipAddress, port))
 518             {
 519                 // NOTE: this leaves the socket in TIME_WAIT.
 520                 // Calling Abort() would clean it since SO_LINGER would be set to 0,
 521                 // however this seems not to be useful.
 522                 mActiveConnections[i].Free();
 523                 mUsedEndPointCount--;
 524             }
 525         }
 526     }
 527 }
 528
 529 void TCPBase::OnPeerClosed(Inet::TCPEndPoint * endPoint)
 530 {
 531     TCPBase * tcp = reinterpret_cast<TCPBase *>(endPoint->AppState);
 532
 533     for (size_t i = 0; i < tcp->mActiveConnectionsSize; i++)
 534     {
 535         if (tcp->mActiveConnections[i].mEndPoint == endPoint)
 536         {
 537             ChipLogProgress(Inet, "Freeing connection: connection closed by peer");
 538             tcp->mActiveConnections[i].Free();
 539             tcp->mUsedEndPointCount--;
 540         }
 541     }
 542 }
 543
 544 bool TCPBase::HasActiveConnections() const
 545 {
 546     for (size_t i = 0; i < mActiveConnectionsSize; i++)
 547     {
 548         if (mActiveConnections[i].InUse())
 549         {
 550             return true;
 551         }
 552     }
 553
 554     return false;
 555 }
 556
 557 } // namespace Transport
 558 } // namespace chip