src/manager/main/socket-manager.cpp

   1 /*
   2  *  Copyright (c) 2000 - 2013 Samsung Electronics Co., Ltd All Rights Reserved
   3  *
   4  *  Contact: Bumjin Im <bj.im@samsung.com>
   5  *
   6  *  Licensed under the Apache License, Version 2.0 (the "License");
   7  *  you may not use this file except in compliance with the License.
   8  *  You may obtain a copy of the License at
   9  *
  10  *      http://www.apache.org/licenses/LICENSE-2.0
  11  *
  12  *  Unless required by applicable law or agreed to in writing, software
  13  *  distributed under the License is distributed on an "AS IS" BASIS,
  14  *  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  15  *  See the License for the specific language governing permissions and
  16  *  limitations under the License
  17  */
  18 /*
  19  * @file        socket-manager.cpp
  20  * @author      Bartlomiej Grzelewski (b.grzelewski@samsung.com)
  21  * @version     1.0
  22  * @brief       Implementation of SocketManager.
  23  */
  24
  25 #include <set>
  26
  27 #include <signal.h>
  28 #include <sys/select.h>
  29 #include <sys/signalfd.h>
  30 #include <sys/types.h>
  31 #include <sys/socket.h>
  32 #include <sys/smack.h>
  33 #include <sys/un.h>
  34 #include <sys/stat.h>
  35 #include <unistd.h>
  36 #include <fcntl.h>
  37 #include <signal.h>
  38 #include <errno.h>
  39 #include <time.h>
  40
  41 #include <systemd/sd-daemon.h>
  42
  43 #include <dpl/log/log.h>
  44 #include <dpl/assert.h>
  45
  46 #include <smack-check.h>
  47 #include <socket-manager.h>
  48
  49 namespace {
  50
  51 const time_t SOCKET_TIMEOUT = 20;
  52
  53 } // namespace anonymous
  54
  55 namespace CKM {
  56
  57 struct DummyService : public GenericSocketService {
  58     ServiceDescriptionVector GetServiceDescription() {
  59         return ServiceDescriptionVector();
  60     }
  61     void Event(const AcceptEvent &event) { (void)event; }
  62     void Event(const WriteEvent &event) { (void)event; }
  63     void Event(const ReadEvent &event) { (void)event; }
  64     void Event(const CloseEvent &event) { (void)event; }
  65 };
  66
  67 struct SignalService : public GenericSocketService {
  68     int GetDescriptor() {
  69         LogInfo("set up");
  70         sigset_t mask;
  71         sigemptyset(&mask);
  72         sigaddset(&mask, SIGTERM);
  73         if (-1 == pthread_sigmask(SIG_BLOCK, &mask, NULL))
  74             return -1;
  75         return signalfd(-1, &mask, 0);
  76     }
  77
  78     ServiceDescriptionVector GetServiceDescription() {
  79         return ServiceDescriptionVector();
  80     }
  81
  82     void Event(const AcceptEvent &event) { (void)event; } // not supported
  83     void Event(const WriteEvent &event) { (void)event; }  // not supported
  84     void Event(const CloseEvent &event) { (void)event; }  // not supported
  85
  86     void Event(const ReadEvent &event) {
  87         LogDebug("Get signal information");
  88
  89         if(sizeof(struct signalfd_siginfo) != event.rawBuffer.size()) {
  90             LogError("Wrong size of signalfd_siginfo struct. Expected: "
  91                 << sizeof(signalfd_siginfo) << " Get: "
  92                 << event.rawBuffer.size());
  93             return;
  94         }
  95
  96         signalfd_siginfo *siginfo = (signalfd_siginfo*)(&(event.rawBuffer[0]));
  97
  98         if (siginfo->ssi_signo == SIGTERM) {
  99             LogInfo("Got signal: SIGTERM");
 100             static_cast<SocketManager*>(m_serviceManager)->MainLoopStop();
 101             return;
 102         }
 103
 104         LogInfo("This should not happend. Got signal: " << siginfo->ssi_signo);
 105     }
 106 };
 107
 108 SocketManager::SocketDescription&
 109 SocketManager::CreateDefaultReadSocketDescription(int sock, bool timeout)
 110 {
 111     if ((int)m_socketDescriptionVector.size() <= sock)
 112         m_socketDescriptionVector.resize(sock+20);
 113
 114     auto &desc = m_socketDescriptionVector[sock];
 115     desc.isListen = false;
 116     desc.isOpen = true;
 117     desc.interfaceID = 0;
 118     desc.service = NULL;
 119     desc.counter = ++m_counter;
 120
 121     if (timeout) {
 122         desc.timeout = time(NULL) + SOCKET_TIMEOUT;
 123         if (false == desc.isTimeout) {
 124             Timeout tm;
 125             tm.time = desc.timeout;
 126             tm.sock = sock;
 127             m_timeoutQueue.push(tm);
 128         }
 129     }
 130
 131     desc.isTimeout = timeout;
 132
 133     FD_SET(sock, &m_readSet);
 134     m_maxDesc = sock > m_maxDesc ? sock : m_maxDesc;
 135     return desc;
 136 }
 137
 138 SocketManager::SocketManager()
 139   : m_counter(0)
 140 {
 141     FD_ZERO(&m_readSet);
 142     FD_ZERO(&m_writeSet);
 143     if (-1 == pipe(m_notifyMe)) {
 144         int err = errno;
 145         ThrowMsg(Exception::InitFailed, "Error in pipe: " << strerror(err));
 146     }
 147     LogInfo("Pipe: Read desc: " << m_notifyMe[0] << " Write desc: " << m_notifyMe[1]);
 148
 149     auto &desc = CreateDefaultReadSocketDescription(m_notifyMe[0], false);
 150     desc.service = new DummyService;
 151
 152     // std::thread bases on pthread so this should work fine
 153     sigset_t set;
 154     sigemptyset(&set);
 155     sigaddset(&set, SIGPIPE);
 156     pthread_sigmask(SIG_BLOCK, &set, NULL);
 157
 158     // add support for TERM signal (passed from systemd)
 159     auto *signalService = new SignalService;
 160     signalService->SetSocketManager(this);
 161     int filefd = signalService->GetDescriptor();
 162     if (-1 == filefd) {
 163         LogError("Error in SignalService.GetDescriptor()");
 164         delete signalService;
 165     } else {
 166         auto &desc2 = CreateDefaultReadSocketDescription(filefd, false);
 167         desc2.service = signalService;
 168         LogInfo("SignalService mounted on " << filefd << " descriptor");
 169     }
 170 }
 171
 172 SocketManager::~SocketManager() {
 173     std::set<GenericSocketService*> serviceMap;
 174
 175     // Find all services. Set is used to remove duplicates.
 176     // In this implementation, services are not able to react in any way.
 177     for (size_t i=0; i < m_socketDescriptionVector.size(); ++i)
 178         if (m_socketDescriptionVector[i].isOpen)
 179             serviceMap.insert(m_socketDescriptionVector[i].service);
 180
 181     // Time to destroy all services.
 182     for(auto it = serviceMap.begin(); it != serviceMap.end(); ++it) {
 183         LogDebug("delete " << (void*)(*it));
 184         delete *it;
 185     }
 186
 187     for (size_t i = 0; i < m_socketDescriptionVector.size(); ++i)
 188         if (m_socketDescriptionVector[i].isOpen)
 189             close(i);
 190
 191     // All socket except one were closed. Now pipe input must be closed.
 192     close(m_notifyMe[1]);
 193 }
 194
 195 void SocketManager::ReadyForAccept(int sock) {
 196     struct sockaddr_un clientAddr;
 197     unsigned int clientLen = sizeof(clientAddr);
 198     int client = accept4(sock, (struct sockaddr*) &clientAddr, &clientLen, SOCK_NONBLOCK);
 199 //    LogInfo("Accept on sock: " << sock << " Socket opended: " << client);
 200     if (-1 == client) {
 201         int err = errno;
 202         LogDebug("Error in accept: " << strerror(err));
 203         return;
 204     }
 205
 206     auto &desc = CreateDefaultReadSocketDescription(client, true);
 207     desc.interfaceID = m_socketDescriptionVector[sock].interfaceID;
 208     desc.service = m_socketDescriptionVector[sock].service;
 209
 210     GenericSocketService::AcceptEvent event;
 211     event.connectionID.sock = client;
 212     event.connectionID.counter = desc.counter;
 213     event.interfaceID = desc.interfaceID;
 214     desc.service->Event(event);
 215 }
 216
 217 void SocketManager::ReadyForRead(int sock) {
 218     if (m_socketDescriptionVector[sock].isListen) {
 219         ReadyForAccept(sock);
 220         return;
 221     }
 222
 223     GenericSocketService::ReadEvent event;
 224     event.connectionID.sock = sock;
 225     event.connectionID.counter = m_socketDescriptionVector[sock].counter;
 226     event.rawBuffer.resize(4096);
 227
 228     auto &desc = m_socketDescriptionVector[sock];
 229     desc.timeout = time(NULL) + SOCKET_TIMEOUT;
 230
 231     ssize_t size = read(sock, &event.rawBuffer[0], 4096);
 232
 233     if (size == 0) {
 234         CloseSocket(sock);
 235     } else if (size >= 0) {
 236         event.rawBuffer.resize(size);
 237         desc.service->Event(event);
 238     } else if (size == -1) {
 239         int err = errno;
 240         switch(err) {
 241             case EAGAIN:
 242             case EINTR:
 243                 break;
 244             default:
 245                 LogDebug("Reading sock error: " << strerror(err));
 246                 CloseSocket(sock);
 247         }
 248     }
 249 }
 250
 251 void SocketManager::ReadyForWriteBuffer(int sock) {
 252     auto &desc = m_socketDescriptionVector[sock];
 253     size_t size = desc.rawBuffer.size();
 254     ssize_t result = write(sock, &desc.rawBuffer[0], size);
 255     if (result == -1) {
 256         int err = errno;
 257         switch(err) {
 258         case EAGAIN:
 259         case EINTR:
 260             // select will trigger write once again, nothing to do
 261             break;
 262         case EPIPE:
 263         default:
 264             LogDebug("Error during write: " << strerror(err));
 265             CloseSocket(sock);
 266             break;
 267         }
 268         return; // We do not want to propagate error to next layer
 269     }
 270
 271     desc.rawBuffer.erase(desc.rawBuffer.begin(), desc.rawBuffer.begin()+result);
 272
 273     desc.timeout = time(NULL) + SOCKET_TIMEOUT;
 274
 275     if (desc.rawBuffer.empty())
 276         FD_CLR(sock, &m_writeSet);
 277
 278     GenericSocketService::WriteEvent event;
 279     event.connectionID.sock = sock;
 280     event.connectionID.counter = desc.counter;
 281     event.size = result;
 282     event.left = desc.rawBuffer.size();
 283
 284     desc.service->Event(event);
 285 }
 286
 287 void SocketManager::ReadyForWrite(int sock) {
 288         ReadyForWriteBuffer(sock);
 289 }
 290
 291 void SocketManager::MainLoop() {
 292     // remove evironment values passed by systemd
 293     // uncomment it after removing old security-server code
 294     // sd_listen_fds(1);
 295
 296     // Daemon is ready to work.
 297     sd_notify(0, "READY=1");
 298
 299     m_working = true;
 300     while(m_working) {
 301         fd_set readSet = m_readSet;
 302         fd_set writeSet = m_writeSet;
 303
 304         timeval localTempTimeout;
 305         timeval *ptrTimeout = &localTempTimeout;
 306
 307         // I need to extract timeout from priority_queue.
 308         // Timeout in priority_queue may be deprecated.
 309         // I need to find some actual one.
 310         while(!m_timeoutQueue.empty()) {
 311             auto &top = m_timeoutQueue.top();
 312             auto &desc = m_socketDescriptionVector[top.sock];
 313
 314             if (top.time == desc.timeout) {
 315                 // This timeout matches timeout from socket.
 316                 // It can be used.
 317                 break;
 318             } else {
 319                 // This socket was used after timeout in priority queue was set up.
 320                 // We need to update timeout and find some useable one.
 321                 Timeout tm = { desc.timeout , top.sock};
 322                 m_timeoutQueue.pop();
 323                 m_timeoutQueue.push(tm);
 324             }
 325         }
 326
 327         if (m_timeoutQueue.empty()) {
 328             LogDebug("No usaable timeout found.");
 329             ptrTimeout = NULL; // select will wait without timeout
 330         } else {
 331             time_t currentTime = time(NULL);
 332             auto &pqTimeout = m_timeoutQueue.top();
 333
 334             // 0 means that select won't block and socket will be closed ;-)
 335             ptrTimeout->tv_sec =
 336               currentTime < pqTimeout.time ? pqTimeout.time - currentTime : 0;
 337             ptrTimeout->tv_usec = 0;
 338 //            LogDebug("Set up timeout: " << (int)ptrTimeout->tv_sec
 339 //                << " seconds. Socket: " << pqTimeout.sock);
 340         }
 341
 342         int ret = select(m_maxDesc+1, &readSet, &writeSet, NULL, ptrTimeout);
 343
 344         if (0 == ret) { // timeout
 345             Assert(!m_timeoutQueue.empty());
 346
 347             Timeout pqTimeout = m_timeoutQueue.top();
 348             m_timeoutQueue.pop();
 349
 350             auto &desc = m_socketDescriptionVector[pqTimeout.sock];
 351
 352             if (!desc.isTimeout || !desc.isOpen) {
 353                 // Connection was closed. Timeout is useless...
 354                 desc.isTimeout = false;
 355                 continue;
 356             }
 357
 358             if (pqTimeout.time < desc.timeout) {
 359                 // Is it possible?
 360                 // This socket was used after timeout. We need to update timeout.
 361                 pqTimeout.time = desc.timeout;
 362                 m_timeoutQueue.push(pqTimeout);
 363                 continue;
 364             }
 365
 366             // timeout from m_timeoutQueue matches with socket.timeout
 367             // and connection is open. Time to close it!
 368             // Putting new timeout in queue here is pointless.
 369             desc.isTimeout = false;
 370             CloseSocket(pqTimeout.sock);
 371
 372             // All done. Now we should process next select ;-)
 373             continue;
 374         }
 375
 376         if (-1 == ret) {
 377             switch(errno) {
 378             case EINTR:
 379                 LogDebug("EINTR in select");
 380                 break;
 381             default:
 382                 int err = errno;
 383                 LogError("Error in select: " << strerror(err));
 384                 return;
 385             }
 386             continue;
 387         }
 388         for(int i = 0; i<m_maxDesc+1 && ret; ++i) {
 389             if (FD_ISSET(i, &readSet)) {
 390                 ReadyForRead(i);
 391                 --ret;
 392             }
 393             if (FD_ISSET(i, &writeSet)) {
 394                 ReadyForWrite(i);
 395                 --ret;
 396             }
 397         }
 398         ProcessQueue();
 399     }
 400 }
 401
 402 void SocketManager::MainLoopStop()
 403 {
 404     m_working = false;
 405     NotifyMe();
 406 }
 407
 408 int SocketManager::GetSocketFromSystemD(
 409     const GenericSocketService::ServiceDescription &desc)
 410 {
 411     int fd;
 412
 413     // TODO optimalization - do it once in object constructor
 414     //                       and remember all information path->sockfd
 415     int n = sd_listen_fds(0);
 416
 417     LogInfo("sd_listen_fds returns: " << n);
 418
 419     if (n < 0) {
 420         LogError("Error in sd_listend_fds");
 421         ThrowMsg(Exception::InitFailed, "Error in sd_listend_fds");
 422     }
 423
 424     for(fd = SD_LISTEN_FDS_START; fd < SD_LISTEN_FDS_START+n; ++fd) {
 425         if (0 < sd_is_socket_unix(fd, SOCK_STREAM, 1,
 426                                   desc.serviceHandlerPath.c_str(), 0))
 427         {
 428             LogInfo("Useable socket " << desc.serviceHandlerPath <<
 429                 " was passed by SystemD under descriptor " << fd);
 430             return fd;
 431         }
 432     }
 433     LogInfo("No useable sockets were passed by systemd.");
 434     return -1;
 435 }
 436
 437 int SocketManager::CreateDomainSocketHelp(
 438     const GenericSocketService::ServiceDescription &desc)
 439 {
 440     int sockfd;
 441
 442     if (-1 == (sockfd = socket(AF_UNIX, SOCK_STREAM, 0))) {
 443         int err = errno;
 444         LogError("Error in socket: " << strerror(err));
 445         ThrowMsg(Exception::InitFailed, "Error in socket: " << strerror(err));
 446     }
 447
 448     if (smack_check()) {
 449         LogInfo("Set up smack label: " << desc.smackLabel);
 450
 451         if (0 != smack_fsetlabel(sockfd, desc.smackLabel.c_str(), SMACK_LABEL_IPIN)) {
 452             LogError("Error in smack_fsetlabel");
 453             ThrowMsg(Exception::InitFailed, "Error in smack_fsetlabel");
 454         }
 455     } else {
 456         LogInfo("No smack on platform. Socket won't be securied with smack label!");
 457     }
 458
 459     int flags;
 460     if (-1 == (flags = fcntl(sockfd, F_GETFL, 0)))
 461         flags = 0;
 462
 463     if (-1 == fcntl(sockfd, F_SETFL, flags | O_NONBLOCK)) {
 464         int err = errno;
 465         close(sockfd);
 466         LogError("Error in fcntl: " << strerror(err));
 467         ThrowMsg(Exception::InitFailed, "Error in fcntl: " << strerror(err));
 468     }
 469
 470     sockaddr_un serverAddress;
 471     memset(&serverAddress, 0, sizeof(serverAddress));
 472     serverAddress.sun_family = AF_UNIX;
 473     strcpy(serverAddress.sun_path, desc.serviceHandlerPath.c_str());
 474     unlink(serverAddress.sun_path);
 475
 476     mode_t originalUmask;
 477     originalUmask = umask(0);
 478
 479     if (-1 == bind(sockfd, (struct sockaddr*)&serverAddress, sizeof(serverAddress))) {
 480         int err = errno;
 481         close(sockfd);
 482         LogError("Error in bind: " << strerror(err));
 483         ThrowMsg(Exception::InitFailed, "Error in bind: " << strerror(err));
 484     }
 485
 486     umask(originalUmask);
 487
 488     if (-1 == listen(sockfd, 5)) {
 489         int err = errno;
 490         close(sockfd);
 491         LogError("Error in listen: " << strerror(err));
 492         ThrowMsg(Exception::InitFailed, "Error in listen: " << strerror(err));
 493     }
 494
 495     return sockfd;
 496 }
 497
 498 void SocketManager::CreateDomainSocket(
 499     GenericSocketService *service,
 500     const GenericSocketService::ServiceDescription &desc)
 501 {
 502     int sockfd = GetSocketFromSystemD(desc);
 503     if (-1 == sockfd)
 504         sockfd = CreateDomainSocketHelp(desc);
 505
 506     auto &description = CreateDefaultReadSocketDescription(sockfd, false);
 507
 508     description.isListen = true;
 509     description.interfaceID = desc.interfaceID;
 510     description.service = service;
 511
 512     LogDebug("Listen on socket: " << sockfd <<
 513         " Handler: " << desc.serviceHandlerPath.c_str());
 514 }
 515
 516 void SocketManager::RegisterSocketService(GenericSocketService *service) {
 517     service->SetSocketManager(this);
 518     auto serviceVector = service->GetServiceDescription();
 519     Try {
 520         for (auto iter = serviceVector.begin(); iter != serviceVector.end(); ++iter)
 521             CreateDomainSocket(service, *iter);
 522     } Catch (Exception::Base) {
 523         for (int i =0; i < (int)m_socketDescriptionVector.size(); ++i)
 524         {
 525             auto &desc = m_socketDescriptionVector[i];
 526             if (desc.service == service && desc.isOpen) {
 527                 close(i);
 528                 desc.isOpen = false;
 529             }
 530         }
 531         ReThrow(Exception::Base);
 532     }
 533 }
 534
 535 void SocketManager::Close(ConnectionID connectionID) {
 536     {
 537         std::lock_guard<std::mutex> ulock(m_eventQueueMutex);
 538         m_closeQueue.push(connectionID);
 539     }
 540     NotifyMe();
 541 }
 542
 543 void SocketManager::Write(ConnectionID connectionID, const RawBuffer &rawBuffer) {
 544     WriteBuffer buffer;
 545     buffer.connectionID = connectionID;
 546     buffer.rawBuffer = rawBuffer;
 547     {
 548         std::lock_guard<std::mutex> ulock(m_eventQueueMutex);
 549         m_writeBufferQueue.push(buffer);
 550     }
 551     NotifyMe();
 552 }
 553
 554 void SocketManager::NotifyMe() {
 555     TEMP_FAILURE_RETRY(write(m_notifyMe[1], "You have message ;-)", 1));
 556 }
 557
 558 void SocketManager::ProcessQueue() {
 559     WriteBuffer buffer;
 560     {
 561         std::lock_guard<std::mutex> ulock(m_eventQueueMutex);
 562         while (!m_writeBufferQueue.empty()) {
 563             buffer = m_writeBufferQueue.front();
 564             m_writeBufferQueue.pop();
 565
 566             auto &desc = m_socketDescriptionVector[buffer.connectionID.sock];
 567
 568             if (!desc.isOpen) {
 569                 LogDebug("Received packet for write but connection is closed. Packet ignored!");
 570                 continue;
 571             }
 572
 573             if (desc.counter != buffer.connectionID.counter)
 574             {
 575                 LogDebug("Received packet for write but counter is broken. Packet ignored!");
 576                 continue;
 577             }
 578
 579             std::copy(
 580                 buffer.rawBuffer.begin(),
 581                 buffer.rawBuffer.end(),
 582                 std::back_inserter(desc.rawBuffer));
 583
 584             FD_SET(buffer.connectionID.sock, &m_writeSet);
 585         }
 586
 587     }
 588
 589     while (1) {
 590         ConnectionID connection;
 591         {
 592             std::lock_guard<std::mutex> ulock(m_eventQueueMutex);
 593             if (m_closeQueue.empty())
 594                 return;
 595             connection = m_closeQueue.front();
 596             m_closeQueue.pop();
 597         }
 598
 599         if (!m_socketDescriptionVector[connection.sock].isOpen)
 600             continue;
 601
 602         if (connection.counter != m_socketDescriptionVector[connection.sock].counter)
 603             continue;
 604
 605         CloseSocket(connection.sock);
 606     }
 607 }
 608
 609 void SocketManager::CloseSocket(int sock) {
 610 //    LogInfo("Closing socket: " << sock);
 611     auto &desc = m_socketDescriptionVector[sock];
 612
 613     if (!(desc.isOpen)) {
 614         // This may happend when some information was waiting for write to the
 615         // socket and in the same time socket was closed by the client.
 616         LogError("Socket " << sock << " is not open. Nothing to do!");
 617         return;
 618     }
 619
 620     GenericSocketService::CloseEvent event;
 621     event.connectionID.sock = sock;
 622     event.connectionID.counter = desc.counter;
 623     auto service = desc.service;
 624
 625     desc.isOpen = false;
 626     desc.service = NULL;
 627     desc.interfaceID = -1;
 628     desc.rawBuffer.clear();
 629
 630     if (service)
 631         service->Event(event);
 632     else
 633         LogError("Critical! Service is NULL! This should never happend!");
 634
 635     TEMP_FAILURE_RETRY(close(sock));
 636     FD_CLR(sock, &m_readSet);
 637     FD_CLR(sock, &m_writeSet);
 638 }
 639
 640 } // namespace CKM