src/inspsocket.cpp

   1 /*
   2  * InspIRCd -- Internet Relay Chat Daemon
   3  *
   4  *   Copyright (C) 2020 Matt Schatz <genius3000@g3k.solutions>
   5  *   Copyright (C) 2019 linuxdaemon <linuxdaemon.irc@gmail.com>
   6  *   Copyright (C) 2018 Dylan Frank <b00mx0r@aureus.pw>
   7  *   Copyright (C) 2013-2016 Attila Molnar <attilamolnar@hush.com>
   8  *   Copyright (C) 2013, 2017-2020 Sadie Powell <sadie@witchery.services>
   9  *   Copyright (C) 2013 Adam <Adam@anope.org>
  10  *   Copyright (C) 2012 Robby <robby@chatbelgie.be>
  11  *   Copyright (C) 2009-2010 Daniel De Graaf <danieldg@inspircd.org>
  12  *   Copyright (C) 2007-2008 Robin Burchell <robin+git@viroteck.net>
  13  *   Copyright (C) 2007 John Brooks <special@inspircd.org>
  14  *   Copyright (C) 2007 Dennis Friis <peavey@inspircd.org>
  15  *   Copyright (C) 2007 Craig Edwards <brain@inspircd.org>
  16  *
  17  * This file is part of InspIRCd.  InspIRCd is free software: you can
  18  * redistribute it and/or modify it under the terms of the GNU General Public
  19  * License as published by the Free Software Foundation, version 2.
  20  *
  21  * This program is distributed in the hope that it will be useful, but WITHOUT
  22  * ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or FITNESS
  23  * FOR A PARTICULAR PURPOSE.  See the GNU General Public License for more
  24  * details.
  25  *
  26  * You should have received a copy of the GNU General Public License
  27  * along with this program.  If not, see <http://www.gnu.org/licenses/>.
  28  */
  29
  30
  31 #include "inspircd.h"
  32 #include "iohook.h"
  33
  34 static IOHook* GetNextHook(IOHook* hook)
  35 {
  36         IOHookMiddle* const iohm = IOHookMiddle::ToMiddleHook(hook);
  37         if (iohm)
  38                 return iohm->GetNextHook();
  39         return NULL;
  40 }
  41
  42 BufferedSocket::BufferedSocket()
  43 {
  44         Timeout = NULL;
  45         state = I_ERROR;
  46 }
  47
  48 BufferedSocket::BufferedSocket(int newfd)
  49 {
  50         Timeout = NULL;
  51         this->fd = newfd;
  52         this->state = I_CONNECTED;
  53         if (HasFd())
  54                 SocketEngine::AddFd(this, FD_WANT_FAST_READ | FD_WANT_EDGE_WRITE);
  55 }
  56
  57 void BufferedSocket::DoConnect(const irc::sockets::sockaddrs& dest, const irc::sockets::sockaddrs& bind, unsigned int maxtime)
  58 {
  59         BufferedSocketError err = BeginConnect(dest, bind, maxtime);
  60         if (err != I_ERR_NONE)
  61         {
  62                 state = I_ERROR;
  63                 SetError(SocketEngine::LastError());
  64                 OnError(err);
  65         }
  66 }
  67
  68 BufferedSocketError BufferedSocket::BeginConnect(const irc::sockets::sockaddrs& dest, const irc::sockets::sockaddrs& bind, unsigned int timeout)
  69 {
  70         if (!HasFd())
  71                 fd = socket(dest.family(), SOCK_STREAM, 0);
  72
  73         if (!HasFd())
  74                 return I_ERR_SOCKET;
  75
  76         if (bind.family() != 0)
  77         {
  78                 if (SocketEngine::Bind(fd, bind) < 0)
  79                         return I_ERR_BIND;
  80         }
  81
  82         SocketEngine::NonBlocking(fd);
  83
  84         if (SocketEngine::Connect(this, dest) == -1)
  85         {
  86                 if (errno != EINPROGRESS)
  87                         return I_ERR_CONNECT;
  88         }
  89
  90         this->state = I_CONNECTING;
  91
  92         if (!SocketEngine::AddFd(this, FD_WANT_NO_READ | FD_WANT_SINGLE_WRITE | FD_WRITE_WILL_BLOCK))
  93                 return I_ERR_NOMOREFDS;
  94
  95         this->Timeout = new SocketTimeout(this->GetFd(), this, timeout);
  96         ServerInstance->Timers.AddTimer(this->Timeout);
  97
  98         ServerInstance->Logs->Log("SOCKET", LOG_DEBUG, "BufferedSocket::DoConnect success");
  99         return I_ERR_NONE;
 100 }
 101
 102 void StreamSocket::Close()
 103 {
 104         if (closing)
 105                 return;
 106
 107         closing = true;
 108         if (HasFd())
 109         {
 110                 // final chance, dump as much of the sendq as we can
 111                 DoWrite();
 112
 113                 IOHook* hook = GetIOHook();
 114                 DelIOHook();
 115                 while (hook)
 116                 {
 117                         hook->OnStreamSocketClose(this);
 118                         IOHook* const nexthook = GetNextHook(hook);
 119                         delete hook;
 120                         hook = nexthook;
 121                 }
 122                 SocketEngine::Shutdown(this, 2);
 123                 SocketEngine::Close(this);
 124         }
 125 }
 126
 127 void StreamSocket::Close(bool writeblock)
 128 {
 129         if (getSendQSize() != 0 && writeblock)
 130                 closeonempty = true;
 131         else
 132                 Close();
 133 }
 134
 135 CullResult StreamSocket::cull()
 136 {
 137         Close();
 138         return EventHandler::cull();
 139 }
 140
 141 bool StreamSocket::GetNextLine(std::string& line, char delim)
 142 {
 143         std::string::size_type i = recvq.find(delim);
 144         if (i == std::string::npos)
 145                 return false;
 146         line.assign(recvq, 0, i);
 147         recvq.erase(0, i + 1);
 148         return true;
 149 }
 150
 151 int StreamSocket::HookChainRead(IOHook* hook, std::string& rq)
 152 {
 153         if (!hook)
 154                 return ReadToRecvQ(rq);
 155
 156         IOHookMiddle* const iohm = IOHookMiddle::ToMiddleHook(hook);
 157         if (iohm)
 158         {
 159                 // Call the next hook to put data into the recvq of the current hook
 160                 const int ret = HookChainRead(iohm->GetNextHook(), iohm->GetRecvQ());
 161                 if (ret <= 0)
 162                         return ret;
 163         }
 164         return hook->OnStreamSocketRead(this, rq);
 165 }
 166
 167 void StreamSocket::DoRead()
 168 {
 169         const std::string::size_type prevrecvqsize = recvq.size();
 170
 171         const int result = HookChainRead(GetIOHook(), recvq);
 172         if (result < 0)
 173         {
 174                 SetError("Read Error"); // will not overwrite a better error message
 175                 return;
 176         }
 177
 178         if (recvq.size() > prevrecvqsize)
 179                 OnDataReady();
 180 }
 181
 182 int StreamSocket::ReadToRecvQ(std::string& rq)
 183 {
 184                 char* ReadBuffer = ServerInstance->GetReadBuffer();
 185                 int n = SocketEngine::Recv(this, ReadBuffer, ServerInstance->Config->NetBufferSize, 0);
 186                 if (n == ServerInstance->Config->NetBufferSize)
 187                 {
 188                         SocketEngine::ChangeEventMask(this, FD_WANT_FAST_READ | FD_ADD_TRIAL_READ);
 189                         rq.append(ReadBuffer, n);
 190                 }
 191                 else if (n > 0)
 192                 {
 193                         SocketEngine::ChangeEventMask(this, FD_WANT_FAST_READ);
 194                         rq.append(ReadBuffer, n);
 195                 }
 196                 else if (n == 0)
 197                 {
 198                         error = "Connection closed";
 199                         SocketEngine::ChangeEventMask(this, FD_WANT_NO_READ | FD_WANT_NO_WRITE);
 200                         return -1;
 201                 }
 202                 else if (SocketEngine::IgnoreError())
 203                 {
 204                         SocketEngine::ChangeEventMask(this, FD_WANT_FAST_READ | FD_READ_WILL_BLOCK);
 205                         return 0;
 206                 }
 207                 else if (errno == EINTR)
 208                 {
 209                         SocketEngine::ChangeEventMask(this, FD_WANT_FAST_READ | FD_ADD_TRIAL_READ);
 210                         return 0;
 211                 }
 212                 else
 213                 {
 214                         error = SocketEngine::LastError();
 215                         SocketEngine::ChangeEventMask(this, FD_WANT_NO_READ | FD_WANT_NO_WRITE);
 216                         return -1;
 217                 }
 218         return n;
 219 }
 220
 221 /* Don't try to prepare huge blobs of data to send to a blocked socket */
 222 static const int MYIOV_MAX = IOV_MAX < 128 ? IOV_MAX : 128;
 223
 224 void StreamSocket::DoWrite()
 225 {
 226         if (getSendQSize() == 0)
 227         {
 228                 if (closeonempty)
 229                         Close();
 230
 231                 return;
 232         }
 233         if (!error.empty() || !HasFd())
 234         {
 235                 ServerInstance->Logs->Log("SOCKET", LOG_DEBUG, "DoWrite on errored or closed socket");
 236                 return;
 237         }
 238
 239         SendQueue* psendq = &sendq;
 240         IOHook* hook = GetIOHook();
 241         while (hook)
 242         {
 243                 int rv = hook->OnStreamSocketWrite(this, *psendq);
 244                 psendq = NULL;
 245
 246                 // rv == 0 means the socket has blocked. Stop trying to send data.
 247                 // IOHook has requested unblock notification from the socketengine.
 248                 if (rv == 0)
 249                         break;
 250
 251                 if (rv < 0)
 252                 {
 253                         SetError("Write Error"); // will not overwrite a better error message
 254                         break;
 255                 }
 256
 257                 IOHookMiddle* const iohm = IOHookMiddle::ToMiddleHook(hook);
 258                 hook = NULL;
 259                 if (iohm)
 260                 {
 261                         psendq = &iohm->GetSendQ();
 262                         hook = iohm->GetNextHook();
 263                 }
 264         }
 265
 266         if (psendq)
 267                 FlushSendQ(*psendq);
 268
 269         if (getSendQSize() == 0 && closeonempty)
 270                 Close();
 271 }
 272
 273 void StreamSocket::FlushSendQ(SendQueue& sq)
 274 {
 275                 // don't even try if we are known to be blocking
 276                 if (GetEventMask() & FD_WRITE_WILL_BLOCK)
 277                         return;
 278                 // start out optimistic - we won't need to write any more
 279                 int eventChange = FD_WANT_EDGE_WRITE;
 280                 while (error.empty() && !sq.empty() && eventChange == FD_WANT_EDGE_WRITE)
 281                 {
 282                         // Prepare a writev() call to write all buffers efficiently
 283                         int bufcount = sq.size();
 284
 285                         // cap the number of buffers at MYIOV_MAX
 286                         if (bufcount > MYIOV_MAX)
 287                         {
 288                                 bufcount = MYIOV_MAX;
 289                         }
 290
 291                         int rv_max = 0;
 292                         int rv;
 293                         {
 294                                 SocketEngine::IOVector iovecs[MYIOV_MAX];
 295                                 size_t j = 0;
 296                                 for (SendQueue::const_iterator i = sq.begin(), end = i+bufcount; i != end; ++i, j++)
 297                                 {
 298                                         const SendQueue::Element& elem = *i;
 299                                         iovecs[j].iov_base = const_cast<char*>(elem.data());
 300                                         iovecs[j].iov_len = elem.length();
 301                                         rv_max += iovecs[j].iov_len;
 302                                 }
 303                                 rv = SocketEngine::WriteV(this, iovecs, bufcount);
 304                         }
 305
 306                         if (rv == (int)sq.bytes())
 307                         {
 308                                 // it's our lucky day, everything got written out. Fast cleanup.
 309                                 // This won't ever happen if the number of buffers got capped.
 310                                 sq.clear();
 311                         }
 312                         else if (rv > 0)
 313                         {
 314                                 // Partial write. Clean out strings from the sendq
 315                                 if (rv < rv_max)
 316                                 {
 317                                         // it's going to block now
 318                                         eventChange = FD_WANT_FAST_WRITE | FD_WRITE_WILL_BLOCK;
 319                                 }
 320                                 while (rv > 0 && !sq.empty())
 321                                 {
 322                                         const SendQueue::Element& front = sq.front();
 323                                         if (front.length() <= (size_t)rv)
 324                                         {
 325                                                 // this string got fully written out
 326                                                 rv -= front.length();
 327                                                 sq.pop_front();
 328                                         }
 329                                         else
 330                                         {
 331                                                 // stopped in the middle of this string
 332                                                 sq.erase_front(rv);
 333                                                 rv = 0;
 334                                         }
 335                                 }
 336                         }
 337                         else if (rv == 0)
 338                         {
 339                                 error = "Connection closed";
 340                         }
 341                         else if (SocketEngine::IgnoreError())
 342                         {
 343                                 eventChange = FD_WANT_FAST_WRITE | FD_WRITE_WILL_BLOCK;
 344                         }
 345                         else if (errno == EINTR)
 346                         {
 347                                 // restart interrupted syscall
 348                                 errno = 0;
 349                         }
 350                         else
 351                         {
 352                                 error = SocketEngine::LastError();
 353                         }
 354                 }
 355                 if (!error.empty())
 356                 {
 357                         // error - kill all events
 358                         SocketEngine::ChangeEventMask(this, FD_WANT_NO_READ | FD_WANT_NO_WRITE);
 359                 }
 360                 else
 361                 {
 362                         SocketEngine::ChangeEventMask(this, eventChange);
 363                 }
 364 }
 365
 366 bool StreamSocket::OnSetEndPoint(const irc::sockets::sockaddrs& local, const irc::sockets::sockaddrs& remote)
 367 {
 368         return false;
 369 }
 370
 371 void StreamSocket::WriteData(const std::string &data)
 372 {
 373         if (!HasFd())
 374         {
 375                 ServerInstance->Logs->Log("SOCKET", LOG_DEBUG, "Attempt to write data to dead socket: %s",
 376                         data.c_str());
 377                 return;
 378         }
 379
 380         /* Append the data to the back of the queue ready for writing */
 381         sendq.push_back(data);
 382
 383         SocketEngine::ChangeEventMask(this, FD_ADD_TRIAL_WRITE);
 384 }
 385
 386 bool SocketTimeout::Tick(time_t)
 387 {
 388         ServerInstance->Logs->Log("SOCKET", LOG_DEBUG, "SocketTimeout::Tick");
 389
 390         if (SocketEngine::GetRef(this->sfd) != this->sock)
 391         {
 392                 delete this;
 393                 return false;
 394         }
 395
 396         if (this->sock->state == I_CONNECTING)
 397         {
 398                 // for connecting sockets, the timeout can occur
 399                 // which causes termination of the connection after
 400                 // the given number of seconds without a successful
 401                 // connection.
 402                 this->sock->OnTimeout();
 403                 this->sock->OnError(I_ERR_TIMEOUT);
 404                 this->sock->state = I_ERROR;
 405
 406                 ServerInstance->GlobalCulls.AddItem(sock);
 407         }
 408
 409         this->sock->Timeout = NULL;
 410         delete this;
 411         return false;
 412 }
 413
 414 void BufferedSocket::OnConnected() { }
 415 void BufferedSocket::OnTimeout() { return; }
 416
 417 void BufferedSocket::OnEventHandlerWrite()
 418 {
 419         if (state == I_CONNECTING)
 420         {
 421                 state = I_CONNECTED;
 422                 this->OnConnected();
 423                 if (!GetIOHook())
 424                         SocketEngine::ChangeEventMask(this, FD_WANT_FAST_READ | FD_WANT_EDGE_WRITE);
 425         }
 426         this->StreamSocket::OnEventHandlerWrite();
 427 }
 428
 429 BufferedSocket::~BufferedSocket()
 430 {
 431         this->Close();
 432         // The timer is removed from the TimerManager in Timer::~Timer()
 433         delete Timeout;
 434 }
 435
 436 void StreamSocket::OnEventHandlerError(int errornum)
 437 {
 438         if (!error.empty())
 439                 return;
 440
 441         if (errornum == 0)
 442                 SetError("Connection closed");
 443         else
 444                 SetError(SocketEngine::GetError(errornum));
 445
 446         BufferedSocketError errcode = I_ERR_OTHER;
 447         switch (errornum)
 448         {
 449                 case ETIMEDOUT:
 450                         errcode = I_ERR_TIMEOUT;
 451                         break;
 452                 case ECONNREFUSED:
 453                 case 0:
 454                         errcode = I_ERR_CONNECT;
 455                         break;
 456                 case EADDRINUSE:
 457                         errcode = I_ERR_BIND;
 458                         break;
 459                 case EPIPE:
 460                 case EIO:
 461                         errcode = I_ERR_WRITE;
 462                         break;
 463         }
 464
 465         // Log and call OnError()
 466         CheckError(errcode);
 467 }
 468
 469 void StreamSocket::OnEventHandlerRead()
 470 {
 471         if (!error.empty())
 472                 return;
 473
 474         try
 475         {
 476                 DoRead();
 477         }
 478         catch (CoreException& ex)
 479         {
 480                 ServerInstance->Logs->Log("SOCKET", LOG_DEFAULT, "Caught exception in socket processing on FD %d - '%s'", fd, ex.GetReason().c_str());
 481                 SetError(ex.GetReason());
 482         }
 483         CheckError(I_ERR_OTHER);
 484 }
 485
 486 void StreamSocket::OnEventHandlerWrite()
 487 {
 488         if (!error.empty())
 489                 return;
 490
 491         DoWrite();
 492         CheckError(I_ERR_OTHER);
 493 }
 494
 495 void StreamSocket::CheckError(BufferedSocketError errcode)
 496 {
 497         if (!error.empty())
 498         {
 499                 ServerInstance->Logs->Log("SOCKET", LOG_DEBUG, "Error on FD %d - '%s'", fd, error.c_str());
 500                 OnError(errcode);
 501         }
 502 }
 503
 504 IOHook* StreamSocket::GetModHook(Module* mod) const
 505 {
 506         for (IOHook* curr = GetIOHook(); curr; curr = GetNextHook(curr))
 507         {
 508                 if (curr->prov->creator == mod)
 509                         return curr;
 510         }
 511         return NULL;
 512 }
 513
 514 IOHook* StreamSocket::GetLastHook() const
 515 {
 516         IOHook* curr = GetIOHook();
 517         IOHook* last = curr;
 518
 519         for (; curr; curr = GetNextHook(curr))
 520                 last = curr;
 521
 522         return last;
 523 }
 524
 525
 526 void StreamSocket::AddIOHook(IOHook* newhook)
 527 {
 528         IOHook* curr = GetIOHook();
 529         if (!curr)
 530         {
 531                 iohook = newhook;
 532                 return;
 533         }
 534
 535         IOHookMiddle* lasthook;
 536         while (curr)
 537         {
 538                 lasthook = IOHookMiddle::ToMiddleHook(curr);
 539                 if (!lasthook)
 540                         return;
 541                 curr = lasthook->GetNextHook();
 542         }
 543
 544         lasthook->SetNextHook(newhook);
 545 }
 546
 547 size_t StreamSocket::getSendQSize() const
 548 {
 549         size_t ret = sendq.bytes();
 550         IOHook* curr = GetIOHook();
 551         while (curr)
 552         {
 553                 const IOHookMiddle* const iohm = IOHookMiddle::ToMiddleHook(curr);
 554                 if (!iohm)
 555                         break;
 556
 557                 ret += iohm->GetSendQ().bytes();
 558                 curr = iohm->GetNextHook();
 559         }
 560         return ret;
 561 }
 562
 563 void StreamSocket::SwapInternals(StreamSocket& other)
 564 {
 565         if (type != other.type)
 566                 return;
 567
 568         EventHandler::SwapInternals(other);
 569         std::swap(closeonempty, other.closeonempty);
 570         std::swap(closing, other.closing);
 571         std::swap(error, other.error);
 572         std::swap(iohook, other.iohook);
 573         std::swap(recvq, other.recvq);
 574         std::swap(sendq, other.sendq);
 575 }