src/inspsocket.cpp

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