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