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