projects/rscriptd/listener.cpp

   1 /*
   2  *    This program is free software; you can redistribute it and/or modify
   3  *    it under the terms of the GNU General Public License as published by
   4  *    the Free Software Foundation; either version 2 of the License, or
   5  *    (at your option) any later version.
   6  *
   7  *    This program is distributed in the hope that it will be useful,
   8  *    but WITHOUT ANY WARRANTY; without even the implied warranty of
   9  *    MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
  10  *    GNU General Public License for more details.
  11  *
  12  *    You should have received a copy of the GNU General Public License
  13  *    along with this program; if not, write to the Free Software
  14  *    Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA  02111-1307  USA
  15  */
  16
  17 /*
  18  *    Author : Boris Mikhailenko <stg34@stargazer.dp.ua>
  19  *    Author : Maxim Mamontov <faust@stargazer.dp.ua>
  20  */
  21
  22 #include "listener.h"
  23
  24 #include "stg/scriptexecuter.h"
  25 #include "stg/common.h"
  26 #include "stg/const.h"
  27
  28 #include <sstream>
  29 #include <algorithm>
  30 #include <chrono>
  31 #include <csignal>
  32 #include <cerrno>
  33 #include <ctime>
  34 #include <cstring>
  35
  36 #include <arpa/inet.h>
  37 #include <sys/uio.h> // readv
  38 #include <sys/types.h> // for historical versions of BSD
  39 #include <sys/socket.h>
  40 #include <netinet/in.h>
  41 #include <unistd.h>
  42
  43 void InitEncrypt(BLOWFISH_CTX * ctx, const std::string & password);
  44
  45 //-----------------------------------------------------------------------------
  46 LISTENER::LISTENER()
  47     : WriteServLog(STG::Logger::get()),
  48       port(0),
  49       receiverStopped(true),
  50       processorStopped(true),
  51       userTimeout(0),
  52       listenSocket(0),
  53       version("rscriptd listener v.1.2")
  54 {
  55 }
  56 //-----------------------------------------------------------------------------
  57 void LISTENER::SetPassword(const std::string & p)
  58 {
  59 password = p;
  60 printfd(__FILE__, "Encryption initiated with password \'%s\'\n", password.c_str());
  61 InitEncrypt(&ctxS, password);
  62 }
  63 //-----------------------------------------------------------------------------
  64 bool LISTENER::Start()
  65 {
  66 printfd(__FILE__, "LISTENER::Start()\n");
  67
  68 if (PrepareNet())
  69     {
  70     return true;
  71     }
  72
  73 if (!m_receiverThread.joinable())
  74     m_receiverThread = std::jthread([this](auto token){ Run(std::move(token)); });
  75
  76 if (!m_processorThread.joinable())
  77     m_processorThread = std::jthread([this](auto token){ RunProcessor(std::move(token)); });
  78
  79 errorStr = "";
  80
  81 return false;
  82 }
  83 //-----------------------------------------------------------------------------
  84 bool LISTENER::Stop()
  85 {
  86 m_receiverThread.request_stop();
  87 m_processorThread.request_stop();
  88
  89 printfd(__FILE__, "LISTENER::Stop()\n");
  90
  91 std::this_thread::sleep_for(std::chrono::milliseconds(500));
  92
  93 if (!processorStopped)
  94     {
  95     //5 seconds to thread stops itself
  96     for (int i = 0; i < 25 && !processorStopped; i++)
  97         std::this_thread::sleep_for(std::chrono::milliseconds(200));
  98
  99     //after 5 seconds waiting thread still running. now killing it
 100     if (!processorStopped)
 101         {
 102         //TODO pthread_cancel()
 103         m_processorThread.detach();
 104         printfd(__FILE__, "LISTENER killed Timeouter\n");
 105         }
 106     }
 107
 108 if (!receiverStopped)
 109     {
 110     //5 seconds to thread stops itself
 111     for (int i = 0; i < 25 && !receiverStopped; i++)
 112         std::this_thread::sleep_for(std::chrono::milliseconds(200));
 113
 114     //after 5 seconds waiting thread still running. now killing it
 115     if (!receiverStopped)
 116         {
 117         //TODO pthread_cancel()
 118         m_receiverThread.detach();
 119         printfd(__FILE__, "LISTENER killed Run\n");
 120         }
 121     }
 122
 123 if (receiverStopped)
 124     m_receiverThread.join();
 125 if (processorStopped)
 126     m_processorThread.join();
 127
 128 FinalizeNet();
 129
 130 for (const auto& user : users)
 131     Disconnect(user);
 132
 133 printfd(__FILE__, "LISTENER::Stoped successfully.\n");
 134
 135 return false;
 136 }
 137 //-----------------------------------------------------------------------------
 138 void LISTENER::Run(std::stop_token token)
 139 {
 140 receiverStopped = false;
 141
 142 while (!token.stop_requested())
 143     RecvPacket(token);
 144
 145 receiverStopped = true;
 146 }
 147 //-----------------------------------------------------------------------------
 148 void LISTENER::RunProcessor(std::stop_token token)
 149 {
 150 processorStopped = false;
 151
 152 while (!token.stop_requested())
 153 {
 154     std::this_thread::sleep_for(std::chrono::milliseconds(500));
 155     if (!pending.empty())
 156         ProcessPending();
 157     ProcessTimeouts();
 158 }
 159
 160 processorStopped = true;
 161 }
 162 //-----------------------------------------------------------------------------
 163 bool LISTENER::PrepareNet()
 164 {
 165 listenSocket = socket(AF_INET, SOCK_DGRAM, 0);
 166
 167 if (listenSocket < 0)
 168     {
 169     errorStr = "Cannot create socket.";
 170     return true;
 171     }
 172
 173 struct sockaddr_in listenAddr;
 174 listenAddr.sin_family = AF_INET;
 175 listenAddr.sin_port = htons(port);
 176 listenAddr.sin_addr.s_addr = inet_addr("0.0.0.0");
 177
 178 if (bind(listenSocket, reinterpret_cast<sockaddr*>(&listenAddr), sizeof(listenAddr)) < 0)
 179     {
 180     errorStr = "LISTENER: Bind failed.";
 181     return true;
 182     }
 183
 184 printfd(__FILE__, "LISTENER::PrepareNet() >>>> Start successfull.\n");
 185
 186 return false;
 187 }
 188 //-----------------------------------------------------------------------------
 189 bool LISTENER::FinalizeNet()
 190 {
 191 close(listenSocket);
 192
 193 return false;
 194 }
 195 //-----------------------------------------------------------------------------
 196 bool LISTENER::RecvPacket(const std::stop_token& token)
 197 {
 198 struct iovec iov[2];
 199
 200 char buffer[RS_MAX_PACKET_LEN];
 201 STG::RS::PACKET_HEADER packetHead;
 202
 203 iov[0].iov_base = reinterpret_cast<char *>(&packetHead);
 204 iov[0].iov_len = sizeof(packetHead);
 205 iov[1].iov_base = buffer;
 206 iov[1].iov_len = sizeof(buffer) - sizeof(packetHead);
 207
 208 size_t dataLen = 0;
 209 while (dataLen < sizeof(buffer))
 210     {
 211     if (!WaitPackets(listenSocket))
 212         {
 213         if (token.stop_requested())
 214             return false;
 215         continue;
 216         }
 217     int portion = readv(listenSocket, iov, 2);
 218     if (portion < 0)
 219         {
 220         return true;
 221         }
 222     dataLen += portion;
 223     }
 224
 225 if (CheckHeader(packetHead))
 226     {
 227     printfd(__FILE__, "Invalid packet or incorrect protocol version!\n");
 228     return true;
 229     }
 230
 231 std::string userLogin(reinterpret_cast<const char*>(packetHead.login));
 232 PendingData pd;
 233 pd.data.login = userLogin;
 234 pd.data.ip = ntohl(packetHead.ip);
 235 pd.data.id = ntohl(packetHead.id);
 236
 237 if (packetHead.packetType == RS_ALIVE_PACKET)
 238     {
 239     pd.type = PendingData::ALIVE;
 240     }
 241 else if (packetHead.packetType == RS_CONNECT_PACKET)
 242     {
 243     pd.type = PendingData::CONNECT;
 244     if (GetParams(buffer, pd.data))
 245         {
 246         return true;
 247         }
 248     }
 249 else if (packetHead.packetType == RS_DISCONNECT_PACKET)
 250     {
 251     pd.type = PendingData::DISCONNECT;
 252     if (GetParams(buffer, pd.data))
 253         {
 254         return true;
 255         }
 256     }
 257 else
 258     return true;
 259
 260 std::lock_guard lock(m_mutex);
 261 pending.push_back(pd);
 262
 263 return false;
 264 }
 265 //-----------------------------------------------------------------------------
 266 bool LISTENER::GetParams(char * buffer, UserData & data)
 267 {
 268 STG::RS::PACKET_TAIL packetTail;
 269
 270 DecryptString(&packetTail, buffer, sizeof(packetTail), &ctxS);
 271
 272 if (strncmp(reinterpret_cast<const char*>(packetTail.magic), RS_ID, RS_MAGIC_LEN))
 273     {
 274     printfd(__FILE__, "Invalid crypto magic\n");
 275     return true;
 276     }
 277
 278 std::ostringstream params;
 279 params << "\"" << data.login << "\" "
 280        << inet_ntostring(data.ip) << " "
 281        << data.id << " "
 282        << reinterpret_cast<const char*>(packetTail.params);
 283
 284 data.params = params.str();
 285
 286 return false;
 287 }
 288 //-----------------------------------------------------------------------------
 289 void LISTENER::ProcessPending()
 290 {
 291 auto it = pending.begin();
 292 size_t count = 0;
 293 printfd(__FILE__, "Pending: %d\n", pending.size());
 294 while (it != pending.end() && count < 256)
 295     {
 296     auto uit = std::lower_bound(users.begin(), users.end(), it->data.login);
 297     if (it->type == PendingData::CONNECT)
 298         {
 299         printfd(__FILE__, "Connect packet\n");
 300         if (uit == users.end() || uit->data.login != it->data.login)
 301             {
 302             printfd(__FILE__, "Connect new user '%s'\n", it->data.login.c_str());
 303             // Add new user
 304             Connect(*it);
 305             users.insert(uit, AliveData(it->data));
 306             }
 307         else
 308             {
 309             printfd(__FILE__, "Update existing user '%s'\n", it->data.login.c_str());
 310             // Update already existing user
 311             time(&uit->lastAlive);
 312             uit->data.params = it->data.params;
 313             }
 314         }
 315     else if (it->type == PendingData::ALIVE)
 316         {
 317         printfd(__FILE__, "Alive packet\n");
 318         if (uit != users.end() && uit->data.login == it->data.login)
 319             {
 320             printfd(__FILE__, "Alive user '%s'\n", it->data.login.c_str());
 321             // Update existing user
 322             time(&uit->lastAlive);
 323             }
 324         else
 325             {
 326             printfd(__FILE__, "Alive user '%s' is not found\n", it->data.login.c_str());
 327             }
 328         }
 329     else if (it->type == PendingData::DISCONNECT)
 330         {
 331         printfd(__FILE__, "Disconnect packet\n");
 332         if (uit != users.end() && uit->data.login == it->data.login.c_str())
 333             {
 334             printfd(__FILE__, "Disconnect user '%s'\n", it->data.login.c_str());
 335             // Disconnect existing user
 336             uit->data.params = it->data.params;
 337             Disconnect(*uit);
 338             users.erase(uit);
 339             }
 340         else
 341             {
 342             printfd(__FILE__, "Cannot find user '%s' for disconnect\n", it->data.login.c_str());
 343             }
 344         }
 345     else
 346         {
 347         printfd(__FILE__, "Unknown packet type\n");
 348         }
 349     ++it;
 350     ++count;
 351     }
 352 std::lock_guard lock(m_mutex);
 353 pending.erase(pending.begin(), it);
 354 }
 355 //-----------------------------------------------------------------------------
 356 void LISTENER::ProcessTimeouts()
 357 {
 358 const auto now = time(nullptr);
 359 const auto it = std::stable_partition(users.begin(), users.end(), [this, now](const auto& data){ return difftime(now, data.lastAlive) < userTimeout; });
 360
 361 if (it != users.end())
 362     {
 363     printfd(__FILE__, "Total users: %d, users to disconnect: %d\n", users.size(), std::distance(it, users.end()));
 364
 365     std::for_each(it, users.end(), [this](const auto& user){ Disconnect(user);});
 366
 367     users.erase(it, users.end());
 368     }
 369 }
 370 //-----------------------------------------------------------------------------
 371 bool LISTENER::Connect(const PendingData & pd) const
 372 {
 373 printfd(__FILE__, "Connect %s\n", pd.data.login.c_str());
 374 if (access(scriptOnConnect.c_str(), X_OK) == 0)
 375     {
 376     if (ScriptExec((scriptOnConnect + " " + pd.data.params).c_str()))
 377         {
 378         WriteServLog("Script %s cannot be executed for an unknown reason.", scriptOnConnect.c_str());
 379         return true;
 380         }
 381     }
 382 else
 383     {
 384     WriteServLog("Script %s cannot be executed. File not found.", scriptOnConnect.c_str());
 385     return true;
 386     }
 387 return false;
 388 }
 389 //-----------------------------------------------------------------------------
 390 bool LISTENER::Disconnect(const AliveData& ad) const
 391 {
 392 printfd(__FILE__, "Disconnect %s\n", ad.data.login.c_str());
 393 if (access(scriptOnDisconnect.c_str(), X_OK) == 0)
 394     {
 395     if (ScriptExec((scriptOnDisconnect + " " + ad.data.params).c_str()))
 396         {
 397         WriteServLog("Script %s cannot be executed for an unknown reson.", scriptOnDisconnect.c_str());
 398         return true;
 399         }
 400     }
 401 else
 402     {
 403     WriteServLog("Script %s cannot be executed. File not found.", scriptOnDisconnect.c_str());
 404     return true;
 405     }
 406 return false;
 407 }
 408 //-----------------------------------------------------------------------------
 409 bool LISTENER::CheckHeader(const STG::RS::PACKET_HEADER & header) const
 410 {
 411 if (strncmp(reinterpret_cast<const char*>(header.magic), RS_ID, RS_MAGIC_LEN))
 412     return true;
 413 if (strncmp(reinterpret_cast<const char*>(header.protoVer), "02", RS_PROTO_VER_LEN))
 414     return true;
 415 return false;
 416 }
 417 //-----------------------------------------------------------------------------
 418 void InitEncrypt(BLOWFISH_CTX * ctx, const std::string & password)
 419 {
 420 char keyL[PASSWD_LEN];
 421 memset(keyL, 0, PASSWD_LEN);
 422 strncpy(keyL, password.c_str(), PASSWD_LEN);
 423 Blowfish_Init(ctx, keyL, PASSWD_LEN);
 424 }
 425 //-----------------------------------------------------------------------------