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 <arpa/inet.h>
  23 #include <sys/uio.h> // readv
  24 #include <sys/types.h> // for historical versions of BSD
  25 #include <sys/socket.h>
  26 #include <netinet/in.h>
  27 #include <unistd.h>
  28
  29 #include <csignal>
  30 #include <ctime>
  31 #include <cstring>
  32 #include <sstream>
  33 #include <algorithm>
  34
  35 #include "listener.h"
  36 #include "script_executer.h"
  37 #include "stg_locker.h"
  38 #include "common.h"
  39
  40 //-----------------------------------------------------------------------------
  41 LISTENER::LISTENER()
  42     : WriteServLog(GetStgLogger()),
  43       port(0),
  44       running(false),
  45       receiverStopped(true),
  46       processorStopped(true),
  47       userTimeout(0),
  48       listenSocket(0)
  49 {
  50 version = "rscriptd listener v.1.2";
  51
  52 pthread_mutex_init(&mutex, NULL);
  53 }
  54 //-----------------------------------------------------------------------------
  55 void LISTENER::SetPassword(const string & p)
  56 {
  57 password = p;
  58 printfd(__FILE__, "Encryption initiated with password \'%s\'\n", password.c_str());
  59 InitEncrypt(&ctxS, password);
  60 }
  61 //-----------------------------------------------------------------------------
  62 bool LISTENER::Start()
  63 {
  64 printfd(__FILE__, "LISTENER::Start()\n");
  65 running = true;
  66
  67 if (PrepareNet())
  68     {
  69     return true;
  70     }
  71
  72 if (receiverStopped)
  73     {
  74     if (pthread_create(&receiverThread, NULL, Run, this))
  75         {
  76         errorStr = "Cannot create thread.";
  77         return true;
  78         }
  79     }
  80
  81 if (processorStopped)
  82     {
  83     if (pthread_create(&processorThread, NULL, RunProcessor, this))
  84         {
  85         errorStr = "Cannot create thread.";
  86         return true;
  87         }
  88     }
  89
  90 errorStr = "";
  91
  92 return false;
  93 }
  94 //-----------------------------------------------------------------------------
  95 bool LISTENER::Stop()
  96 {
  97 running = false;
  98
  99 printfd(__FILE__, "LISTENER::Stop()\n");
 100
 101 usleep(500000);
 102
 103 if (!processorStopped)
 104     {
 105     //5 seconds to thread stops itself
 106     for (int i = 0; i < 25 && !processorStopped; i++)
 107         {
 108         usleep(200000);
 109         }
 110
 111     //after 5 seconds waiting thread still running. now killing it
 112     if (!processorStopped)
 113         {
 114         //TODO pthread_cancel()
 115         if (pthread_kill(processorThread, SIGINT))
 116             {
 117             errorStr = "Cannot kill thread.";
 118             return true;
 119             }
 120         printfd(__FILE__, "LISTENER killed Timeouter\n");
 121         }
 122     }
 123
 124 if (!receiverStopped)
 125     {
 126     //5 seconds to thread stops itself
 127     for (int i = 0; i < 25 && !receiverStopped; i++)
 128         {
 129         usleep(200000);
 130         }
 131
 132     //after 5 seconds waiting thread still running. now killing it
 133     if (!receiverStopped)
 134         {
 135         //TODO pthread_cancel()
 136         if (pthread_kill(receiverThread, SIGINT))
 137             {
 138             errorStr = "Cannot kill thread.";
 139             return true;
 140             }
 141         printfd(__FILE__, "LISTENER killed Run\n");
 142         }
 143     }
 144
 145 pthread_join(receiverThread, NULL);
 146 pthread_join(processorThread, NULL);
 147
 148 pthread_mutex_destroy(&mutex);
 149
 150 FinalizeNet();
 151
 152 std::for_each(users.begin(), users.end(), DisconnectUser(*this));
 153
 154 printfd(__FILE__, "LISTENER::Stoped successfully.\n");
 155
 156 return false;
 157 }
 158 //-----------------------------------------------------------------------------
 159 void * LISTENER::Run(void * d)
 160 {
 161 LISTENER * ia = static_cast<LISTENER *>(d);
 162
 163 ia->Runner();
 164
 165 return NULL;
 166 }
 167 //-----------------------------------------------------------------------------
 168 void LISTENER::Runner()
 169 {
 170 receiverStopped = false;
 171
 172 while (running)
 173     {
 174     RecvPacket();
 175     }
 176
 177 receiverStopped = true;
 178 }
 179 //-----------------------------------------------------------------------------
 180 void * LISTENER::RunProcessor(void * d)
 181 {
 182 LISTENER * ia = static_cast<LISTENER *>(d);
 183
 184 ia->ProcessorRunner();
 185
 186 return NULL;
 187 }
 188 //-----------------------------------------------------------------------------
 189 void LISTENER::ProcessorRunner()
 190 {
 191 processorStopped = false;
 192
 193 while (running)
 194     {
 195     usleep(500000);
 196     if (!pending.empty())
 197         ProcessPending();
 198     ProcessTimeouts();
 199     }
 200
 201 processorStopped = true;
 202 }
 203 //-----------------------------------------------------------------------------
 204 bool LISTENER::PrepareNet()
 205 {
 206 listenSocket = socket(AF_INET, SOCK_DGRAM, 0);
 207
 208 if (listenSocket < 0)
 209     {
 210     errorStr = "Cannot create socket.";
 211     return true;
 212     }
 213
 214 printfd(__FILE__, "Port: %d\n", port);
 215
 216 struct sockaddr_in listenAddr;
 217 listenAddr.sin_family = AF_INET;
 218 listenAddr.sin_port = htons(port);
 219 listenAddr.sin_addr.s_addr = inet_addr("0.0.0.0");
 220
 221 if (bind(listenSocket, (struct sockaddr*)&listenAddr, sizeof(listenAddr)) < 0)
 222     {
 223     errorStr = "LISTENER: Bind failed.";
 224     return true;
 225     }
 226
 227 printfd(__FILE__, "LISTENER::PrepareNet() >>>> Start successfull.\n");
 228
 229 return false;
 230 }
 231 //-----------------------------------------------------------------------------
 232 bool LISTENER::FinalizeNet()
 233 {
 234 close(listenSocket);
 235
 236 return false;
 237 }
 238 //-----------------------------------------------------------------------------
 239 bool LISTENER::RecvPacket()
 240 {
 241 struct iovec iov[2];
 242
 243 char buffer[RS_MAX_PACKET_LEN];
 244 RS_PACKET_HEADER packetHead;
 245
 246 iov[0].iov_base = reinterpret_cast<char *>(&packetHead);
 247 iov[0].iov_len = sizeof(packetHead);
 248 iov[1].iov_base = buffer;
 249 iov[1].iov_len = sizeof(buffer);
 250
 251 size_t dataLen = 0;
 252 while (dataLen < sizeof(buffer))
 253     {
 254     if (!WaitPackets(listenSocket))
 255         {
 256         if (!running)
 257             return false;
 258         continue;
 259         }
 260     int portion = readv(listenSocket, iov, 2);
 261     if (portion < 0)
 262         {
 263         return true;
 264         }
 265     dataLen += portion;
 266     }
 267
 268 if (CheckHeader(packetHead))
 269     {
 270     printfd(__FILE__, "Invalid packet or incorrect protocol version!\n");
 271     return true;
 272     }
 273
 274 std::string userLogin((char *)packetHead.login);
 275 PendingData data;
 276 data.login = userLogin;
 277 data.ip = ntohl(packetHead.ip);
 278 data.id = ntohl(packetHead.id);
 279
 280 if (packetHead.packetType == RS_ALIVE_PACKET)
 281     {
 282     data.type = PendingData::ALIVE;
 283     }
 284 else if (packetHead.packetType == RS_CONNECT_PACKET)
 285     {
 286     data.type = PendingData::CONNECT;
 287     if (GetParams(buffer, data))
 288         {
 289         return true;
 290         }
 291     }
 292 else if (packetHead.packetType == RS_DISCONNECT_PACKET)
 293     {
 294     data.type = PendingData::DISCONNECT;
 295     if (GetParams(buffer, data))
 296         {
 297         return true;
 298         }
 299     }
 300
 301 STG_LOCKER lock(&mutex, __FILE__, __LINE__);
 302 pending.push_back(data);
 303
 304 return false;
 305 }
 306 //-----------------------------------------------------------------------------
 307 bool LISTENER::GetParams(char * buffer, UserData & data)
 308 {
 309 RS_PACKET_TAIL packetTail;
 310
 311 Decrypt(&ctxS, (char *)&packetTail, buffer, sizeof(packetTail) / 8);
 312
 313 if (strncmp((char *)packetTail.magic, RS_ID, RS_MAGIC_LEN))
 314     {
 315     printfd(__FILE__, "Invalid crypto magic\n");
 316     return true;
 317     }
 318
 319 std::stringstream params;
 320 params << data.login << " "
 321        << inet_ntostring(data.ip) << " "
 322        << ntohl(data.id) << " "
 323        << (char *)packetTail.params;
 324
 325 data.params = params.str();
 326
 327 return false;
 328 }
 329 //-----------------------------------------------------------------------------
 330 void LISTENER::ProcessPending()
 331 {
 332 printfd(__FILE__, "Pending data size: %d\n", pending.size());
 333 std::list<PendingData>::iterator it(pending.begin());
 334 while (it != pending.end())
 335     {
 336     std::vector<AliveData>::iterator uit(
 337             std::lower_bound(
 338                 users.begin(),
 339                 users.end(),
 340                 it->login)
 341             );
 342     if (it->type == PendingData::CONNECT)
 343         {
 344         if (uit == users.end() || uit->login != it->login)
 345             {
 346             // Add new user
 347             Connect(*it);
 348             users.insert(uit, AliveData(static_cast<UserData>(*it)));
 349             }
 350         else if (uit->login == it->login)
 351             {
 352             // Update already existing user
 353             time(&uit->lastAlive);
 354             uit->params = it->params;
 355             }
 356         }
 357     else if (it->type == PendingData::ALIVE)
 358         {
 359         if (uit != users.end() && uit->login == it->login)
 360             {
 361             // Update existing user
 362             time(&uit->lastAlive);
 363             }
 364         }
 365     else if (it->type == PendingData::DISCONNECT)
 366         {
 367         if (uit != users.end() && uit->login == it->login.c_str())
 368             {
 369             // Disconnect existing user
 370             Disconnect(*uit);
 371             users.erase(uit);
 372             }
 373         }
 374
 375     STG_LOCKER lock(&mutex, __FILE__, __LINE__);
 376     pending.erase(it++);
 377     }
 378 }
 379 //-----------------------------------------------------------------------------
 380 void LISTENER::ProcessTimeouts()
 381 {
 382 const std::vector<AliveData>::iterator it(
 383         std::stable_partition(
 384             users.begin(),
 385             users.end(),
 386             IsNotTimedOut(userTimeout)
 387         )
 388     );
 389
 390 if (it != users.end())
 391     {
 392     printfd(__FILE__, "Total users: %d, users to disconnect: %d\n", users.size(), std::distance(it, users.end()));
 393
 394     std::for_each(
 395             it,
 396             users.end(),
 397             DisconnectUser(*this)
 398         );
 399
 400     users.erase(it, users.end());
 401     }
 402 }
 403 //-----------------------------------------------------------------------------
 404 bool LISTENER::Connect(const UserData & data) const
 405 {
 406 printfd(__FILE__, "Connect %s\n", data.login.c_str());
 407 if (access(scriptOnConnect.c_str(), X_OK) == 0)
 408     {
 409     if (ScriptExec(scriptOnConnect + " " + data.params))
 410         {
 411         WriteServLog("Script %s cannot be executed for an unknown reason.", scriptOnConnect.c_str());
 412         return true;
 413         }
 414     }
 415 else
 416     {
 417     WriteServLog("Script %s cannot be executed. File not found.", scriptOnConnect.c_str());
 418     return true;
 419     }
 420 return false;
 421 }
 422 //-----------------------------------------------------------------------------
 423 bool LISTENER::Disconnect(const UserData & data) const
 424 {
 425 printfd(__FILE__, "Disconnect %s\n", data.login.c_str());
 426 if (access(scriptOnDisconnect.c_str(), X_OK) == 0)
 427     {
 428     if (ScriptExec(scriptOnDisconnect + " " + data.params))
 429         {
 430         WriteServLog("Script %s cannot be executed for an unknown reson.", scriptOnDisconnect.c_str());
 431         return true;
 432         }
 433     }
 434 else
 435     {
 436     WriteServLog("Script %s cannot be executed. File not found.", scriptOnDisconnect.c_str());
 437     return true;
 438     }
 439 return false;
 440 }
 441 //-----------------------------------------------------------------------------
 442 void LISTENER::InitEncrypt(BLOWFISH_CTX * ctx, const string & password)
 443 {
 444 unsigned char keyL[PASSWD_LEN];
 445 memset(keyL, 0, PASSWD_LEN);
 446 strncpy((char *)keyL, password.c_str(), PASSWD_LEN);
 447 Blowfish_Init(ctx, keyL, PASSWD_LEN);
 448 }
 449 //-----------------------------------------------------------------------------
 450 void LISTENER::Decrypt(BLOWFISH_CTX * ctx, char * dst, const char * src, int len8)
 451 {
 452 if (dst != src)
 453     memcpy(dst, src, len8 * 8);
 454
 455 for (int i = 0; i < len8; i++)
 456     Blowfish_Decrypt(ctx, (uint32_t *)(dst + i * 8), (uint32_t *)(dst + i * 8 + 4));
 457 }
 458 //-----------------------------------------------------------------------------
 459 bool LISTENER::CheckHeader(const RS_PACKET_HEADER & header) const
 460 {
 461 if (strncmp((char *)header.magic, RS_ID, RS_MAGIC_LEN))
 462     {
 463     return true;
 464     }
 465 if (strncmp((char *)header.protoVer, "02", RS_PROTO_VER_LEN))
 466     {
 467     return true;
 468     }
 469 return false;
 470 }
 471 //-----------------------------------------------------------------------------
 472 bool LISTENER::WaitPackets(int sd) const
 473 {
 474 fd_set rfds;
 475 FD_ZERO(&rfds);
 476 FD_SET(sd, &rfds);
 477
 478 struct timeval tv;
 479 tv.tv_sec = 0;
 480 tv.tv_usec = 500000;
 481
 482 int res = select(sd + 1, &rfds, NULL, NULL, &tv);
 483 if (res == -1) // Error
 484     {
 485     if (errno != EINTR)
 486         {
 487         printfd(__FILE__, "Error on select: '%s'\n", strerror(errno));
 488         }
 489     return false;
 490     }
 491
 492 if (res == 0) // Timeout
 493     {
 494     return false;
 495     }
 496
 497 return true;
 498 }
 499 //-----------------------------------------------------------------------------