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