Network.cpp

00001 /**************************************************************************************
00002  * Copyright (c) 2016, Tomoaki Yamaguchi
00003  *
00004  * All rights reserved. This program and the accompanying materials
00005  * are made available under the terms of the Eclipse Public License v1.0
00006  * and Eclipse Distribution License v1.0 which accompany this distribution.
00007  *
00008  * The Eclipse Public License is available at
00009  *    http://www.eclipse.org/legal/epl-v10.html
00010  * and the Eclipse Distribution License is available at
00011  *   http://www.eclipse.org/org/documents/edl-v10.php.
00012  *
00013  * Contributors:
00014  *    Tomoaki Yamaguchi - initial API and implementation and/or initial documentation
00015  **************************************************************************************/
00016 
00017 #include <string.h>
00018 #include <openssl/ssl.h>
00019 #include <openssl/err.h>
00020 #include <openssl/rand.h>
00021 #include <unistd.h>
00022 #include <fcntl.h>
00023 #include <sys/socket.h>
00024 #include <regex>
00025 
00026 #include "Network.h"
00027 #include "MQTTSNGWDefines.h"
00028 #include "MQTTSNGWProcess.h"
00029 
00030 using namespace std;
00031 using namespace MQTTSNGW;
00032 
00033 #define SOCKET_MAXCONNECTIONS  5
00034 char* currentDateTime();
00035 
00036 /*========================================
00037  Class TCPStack
00038  =======================================*/
00039 TCPStack::TCPStack()
00040 {
00041     _addrinfo = 0;
00042     _sockfd = 0;
00043 }
00044 
00045 TCPStack::~TCPStack()
00046 {
00047     if (_addrinfo)
00048     {
00049         freeaddrinfo(_addrinfo);
00050     }
00051 }
00052 
00053 bool TCPStack::isValid()
00054 {
00055     return (_sockfd > 0);
00056 }
00057 
00058 void TCPStack::close()
00059 {
00060     _mutex.lock();
00061     if (_sockfd > 0)
00062     {
00063         ::close(_sockfd);
00064         _sockfd = 0;
00065         if (_addrinfo)
00066         {
00067             freeaddrinfo(_addrinfo);
00068             _addrinfo = 0;
00069         }
00070     }
00071     _mutex.unlock();
00072 
00073 }
00074 
00075 bool TCPStack::bind(const char* service)
00076 {
00077     if (isValid())
00078     {
00079         return false;
00080     }
00081     addrinfo hints;
00082     memset(&hints, 0, sizeof(addrinfo));
00083     hints.ai_family = AF_INET;
00084     hints.ai_socktype = SOCK_STREAM;
00085     hints.ai_flags = AI_PASSIVE;
00086 
00087     if (_addrinfo)
00088     {
00089         freeaddrinfo(_addrinfo);
00090     }
00091     int err = getaddrinfo(0, service, &hints, &_addrinfo);
00092     if (err)
00093     {
00094         WRITELOG("\n%s   \x1b[0m\x1b[31merror:\x1b[0m\x1b[37mgetaddrinfo(): %s\n", currentDateTime(),
00095                 gai_strerror(err));
00096         return false;
00097     }
00098 
00099     _sockfd = socket(_addrinfo->ai_family, _addrinfo->ai_socktype, _addrinfo->ai_protocol);
00100     if (_sockfd < 0)
00101     {
00102         return false;
00103     }
00104     int on = 1;
00105     if (setsockopt(_sockfd, SOL_SOCKET, SO_REUSEADDR, (const char*) &on, sizeof(on)) == -1)
00106     {
00107         return false;
00108     }
00109 
00110     if (::bind(_sockfd, _addrinfo->ai_addr, _addrinfo->ai_addrlen) < 0)
00111     {
00112         return false;
00113     }
00114     return true;
00115 }
00116 
00117 bool TCPStack::listen()
00118 {
00119     if (!isValid())
00120     {
00121         return false;
00122     }
00123     int listen_return = ::listen(_sockfd, SOCKET_MAXCONNECTIONS);
00124     if (listen_return == -1)
00125     {
00126         return false;
00127     }
00128     return true;
00129 }
00130 
00131 bool TCPStack::accept(TCPStack& new_socket)
00132 {
00133     sockaddr_storage sa;
00134     socklen_t len = sizeof(sa);
00135     new_socket._sockfd = ::accept(_sockfd, (struct sockaddr*) &sa, &len);
00136     if (new_socket._sockfd <= 0)
00137     {
00138         return false;
00139     }
00140     else
00141     {
00142         return true;
00143     }
00144 }
00145 
00146 int TCPStack::send(const uint8_t* buf, int length)
00147 {
00148     return ::send(_sockfd, buf, length, MSG_NOSIGNAL);
00149 }
00150 
00151 int TCPStack::recv(uint8_t* buf, int len)
00152 {
00153     return ::recv(_sockfd, buf, len, 0);
00154 }
00155 
00156 bool TCPStack::connect(const char* host, const char* service)
00157 {
00158     if (isValid())
00159     {
00160         return true;
00161     }
00162     addrinfo hints;
00163     memset(&hints, 0, sizeof(addrinfo));
00164     hints.ai_family = AF_INET;
00165     hints.ai_socktype = SOCK_STREAM;
00166     if (_addrinfo)
00167     {
00168         freeaddrinfo(_addrinfo);
00169     }
00170 
00171     int err = getaddrinfo(host, service, &hints, &_addrinfo);
00172     if (err)
00173     {
00174         WRITELOG("\n%s   \x1b[0m\x1b[31merror:\x1b[0m\x1b[37mgetaddrinfo(): %s\n", currentDateTime(),
00175                 gai_strerror(err));
00176         return false;
00177     }
00178 
00179     int sockfd = socket(_addrinfo->ai_family, _addrinfo->ai_socktype, _addrinfo->ai_protocol);
00180 
00181     if (sockfd < 0)
00182     {
00183         return false;
00184     }
00185     int on = 1;
00186 
00187     if (setsockopt(sockfd, SOL_SOCKET, SO_REUSEADDR, (const char*) &on, sizeof(on)) == -1)
00188     {
00189         return false;
00190     }
00191 
00192     if (::connect(sockfd, _addrinfo->ai_addr, _addrinfo->ai_addrlen) < 0)
00193     {
00194         DEBUGLOG("Can not connect the socket. Check the PortNo! \n");
00195         ::close(sockfd);
00196         return false;
00197     }
00198 
00199     _sockfd = sockfd;
00200     return true;
00201 }
00202 
00203 void TCPStack::setNonBlocking(const bool b)
00204 {
00205     int opts;
00206 
00207     opts = fcntl(_sockfd, F_GETFL);
00208 
00209     if (opts < 0)
00210     {
00211         return;
00212     }
00213 
00214     if (b)
00215     {
00216         opts = (opts | O_NONBLOCK);
00217     }
00218     else
00219     {
00220         opts = (opts & ~O_NONBLOCK);
00221     }
00222     fcntl(_sockfd, F_SETFL, opts);
00223 }
00224 
00225 int TCPStack::getSock()
00226 {
00227     return _sockfd;
00228 }
00229 
00230 /*========================================
00231  Class Network
00232  =======================================*/
00233 int Network::_numOfInstance = 0;
00234 SSL_CTX* Network::_ctx = 0;
00235 SSL_SESSION* Network::_session = 0;
00236 
00237 Network::Network(bool secure) :
00238         TCPStack()
00239 {
00240     _ssl = 0;
00241     _secureFlg = secure;
00242     _busy = false;
00243     _sslValid = false;
00244 }
00245 
00246 Network::~Network()
00247 {
00248     close();
00249 }
00250 
00251 bool Network::connect(const char* host, const char* port)
00252 {
00253     bool rc = false;
00254     _mutex.lock();
00255     if (_secureFlg)
00256     {
00257         goto exit;
00258     }
00259 
00260     if (getSock() == 0)
00261     {
00262         if (!TCPStack::connect(host, port))
00263         {
00264             goto exit;
00265         }
00266     }
00267     rc = true;
00268 exit:
00269     _mutex.unlock();
00270     return rc;
00271 }
00272 
00273 bool Network::connect(const char* host, const char* port, const char* caPath, const char* caFile, const char* certkey, const char* prvkey)
00274 {
00275     char errmsg[256];
00276     char peer_CN[256];
00277     bool rc;
00278 
00279     _mutex.lock();
00280     try
00281     {
00282         if (!_secureFlg)
00283         {
00284             WRITELOG("TLS is not required.\n");
00285             throw false;
00286         }
00287 
00288         if (_ctx == 0)
00289         {
00290             SSL_load_error_strings();
00291             SSL_library_init();
00292 
00293 #if ( OPENSSL_VERSION_NUMBER >= 0x10100000L )
00294             _ctx = SSL_CTX_new(TLS_client_method());
00295 #elif ( OPENSSL_VERSION_NUMBER >= 0x10001000L )
00296             _ctx = SSL_CTX_new(TLSv1_client_method());
00297 #else
00298             _ctx = SSL_CTX_new(SSLv23_client_method());
00299 #endif
00300 
00301             if (_ctx == 0)
00302             {
00303                 ERR_error_string_n(ERR_get_error(), errmsg, sizeof(errmsg));
00304                 WRITELOG("SSL_CTX_new() %s\n", errmsg);
00305                 throw false;
00306             }
00307 
00308 
00309             if (!SSL_CTX_load_verify_locations(_ctx, caFile, caPath))
00310             {
00311                 ERR_error_string_n(ERR_get_error(), errmsg, sizeof(errmsg));
00312                 WRITELOG("SSL_CTX_load_verify_locations() %s\n", errmsg);
00313                 throw false;
00314             }
00315 
00316             if ( certkey )
00317             {
00318                 if ( SSL_CTX_use_certificate_file(_ctx, certkey, SSL_FILETYPE_PEM) != 1 )
00319                 {
00320                     ERR_error_string_n(ERR_get_error(), errmsg, sizeof(errmsg));
00321                     WRITELOG("SSL_CTX_use_certificate_file() %s %s\n", certkey, errmsg);
00322                     throw false;
00323                 }
00324             }
00325             if ( prvkey )
00326             {
00327                 if ( SSL_CTX_use_PrivateKey_file(_ctx, prvkey, SSL_FILETYPE_PEM) != 1 )
00328                 {
00329                     ERR_error_string_n(ERR_get_error(), errmsg, sizeof(errmsg));
00330                     WRITELOG("SSL_use_PrivateKey_file() %s %s\n", prvkey, errmsg);
00331                     throw false;
00332                 }
00333             }
00334         }
00335 
00336         if (! TCPStack::isValid())
00337         {
00338             if ( !TCPStack::connect(host, port) )
00339             {
00340                 throw false;
00341             }
00342         }
00343 
00344         _ssl = SSL_new(_ctx);
00345         if (_ssl == 0)
00346         {
00347             ERR_error_string_n(ERR_get_error(), errmsg, sizeof(errmsg));
00348             WRITELOG("SSL_new()  %s\n", errmsg);
00349             throw false;
00350         }
00351 
00352         if (!SSL_set_fd(_ssl, TCPStack::getSock()))
00353         {
00354             ERR_error_string_n(ERR_get_error(), errmsg, sizeof(errmsg));
00355             WRITELOG("SSL_set_fd()  %s\n", errmsg);
00356             SSL_free(_ssl);
00357             _ssl = 0;
00358             throw false;
00359         }
00360 
00361         if (_session)
00362         {
00363             SSL_set_session(_ssl, _session);
00364         }
00365 
00366         if (SSL_connect(_ssl) != 1)
00367         {
00368             ERR_error_string_n(ERR_get_error(), errmsg, sizeof(errmsg));
00369             WRITELOG("SSL_connect() %s\n", errmsg);
00370             SSL_free(_ssl);
00371             _ssl = 0;
00372             throw false;
00373         }
00374 
00375         int result;
00376         if ( (result = SSL_get_verify_result(_ssl)) != X509_V_OK)
00377         {
00378             WRITELOG("SSL_get_verify_result() error: %s.\n", X509_verify_cert_error_string(result));
00379             SSL_free(_ssl);
00380             _ssl = 0;
00381             throw false;
00382         }
00383 
00384         X509* peer = SSL_get_peer_certificate(_ssl);
00385         X509_NAME_get_text_by_NID(X509_get_subject_name(peer), NID_commonName, peer_CN, 256);
00386         char* pos = peer_CN;
00387         if ( *pos == '*')
00388         {
00389             while (*host++ != '.');
00390             pos += 2;
00391         }
00392         if ( strcmp(host, pos))
00393         {
00394             WRITELOG("SSL_get_peer_certificate() error: Broker %s dosen't match the host name %s\n", peer_CN, host);
00395             SSL_free(_ssl);
00396             _ssl = 0;
00397             throw false;
00398         }
00399 
00400         if (_session == 0)
00401         {
00402             _session = SSL_get1_session(_ssl);
00403         }
00404         _numOfInstance++;
00405         _sslValid = true;
00406         rc = true;
00407     }
00408     catch (bool x)
00409     {
00410         rc = x;
00411     }
00412     _mutex.unlock();
00413     return rc;
00414 }
00415 
00416 int Network::send(const uint8_t* buf, uint16_t length)
00417 {
00418     char errmsg[256];
00419     fd_set rset;
00420     fd_set wset;
00421     bool writeBlockedOnRead = false;
00422     int bpos = 0;
00423 
00424     if (!_secureFlg)
00425     {
00426         return TCPStack::send(buf, length);
00427     }
00428     else
00429     {
00430         _mutex.lock();
00431 
00432         if ( !_ssl )
00433         {
00434             _mutex.unlock();
00435             return -1;
00436         }
00437         _busy = true;
00438 
00439         while (true)
00440         {
00441             FD_ZERO(&rset);
00442             FD_ZERO(&wset);
00443             FD_SET(getSock(), &rset);
00444             FD_SET(getSock(), &wset);
00445 
00446             int activity = select(getSock() + 1, &rset, &wset, 0, 0);
00447             if (activity > 0)
00448             {
00449                 if (FD_ISSET(getSock(), &wset) || (writeBlockedOnRead  && FD_ISSET(getSock(), &rset)))
00450                 {
00451 
00452                     writeBlockedOnRead = false;
00453                     int r = SSL_write(_ssl, buf + bpos, length);
00454 
00455                     switch (SSL_get_error(_ssl, r))
00456                     {
00457                     case SSL_ERROR_NONE:
00458                         length -= r;
00459                         bpos += r;
00460                         if (length == 0)
00461                         {
00462                             _busy = false;
00463                             _mutex.unlock();
00464                             return bpos;
00465                         }
00466                         break;
00467                     case SSL_ERROR_WANT_WRITE:
00468                         break;
00469                     case SSL_ERROR_WANT_READ:
00470                         writeBlockedOnRead = true;
00471                         break;
00472                     default:
00473                         ERR_error_string_n(ERR_get_error(), errmsg, sizeof(errmsg));
00474                         WRITELOG("TLSStack::send() default %s\n", errmsg);
00475                         _busy = false;
00476                         _mutex.unlock();
00477                         return -1;
00478                     }
00479                 }
00480             }
00481         }
00482     }
00483 }
00484 
00485 int Network::recv(uint8_t* buf, uint16_t len)
00486 {
00487     char errmsg[256];
00488     bool writeBlockedOnRead = false;
00489     bool readBlockedOnWrite = false;
00490     bool readBlocked = false;
00491     int rlen = 0;
00492     int bpos = 0;
00493     fd_set rset;
00494     fd_set wset;
00495 
00496     if (!_secureFlg)
00497     {
00498         return TCPStack::recv(buf, len);
00499     }
00500 
00501     if (_busy)
00502     {
00503         return 0;
00504     }
00505     _mutex.lock();
00506 
00507     if ( !_ssl )
00508     {
00509         _mutex.unlock();
00510         return 0;
00511     }
00512 
00513     _busy = true;
00514 loop:
00515     do
00516     {
00517         readBlockedOnWrite = false;
00518         readBlocked = false;
00519 
00520         rlen = SSL_read(_ssl, buf + bpos, len - bpos);
00521 
00522         switch (SSL_get_error(_ssl, rlen))
00523         {
00524         case SSL_ERROR_NONE:
00525             _busy = false;
00526             _mutex.unlock();
00527             return rlen + bpos;
00528             break;
00529         case SSL_ERROR_ZERO_RETURN:
00530             SSL_shutdown(_ssl);
00531             _ssl = 0;
00532             _numOfInstance--;
00533             //TCPStack::close();
00534             _busy = false;
00535             _mutex.unlock();
00536             return -1;
00537             break;
00538         case SSL_ERROR_WANT_READ:
00539             readBlocked = true;
00540             break;
00541         case SSL_ERROR_WANT_WRITE:
00542             readBlockedOnWrite = true;
00543             break;
00544         default:
00545             ERR_error_string_n(ERR_get_error(), errmsg, sizeof(errmsg));
00546             WRITELOG("Network::recv() %s\n", errmsg);
00547             _busy = false;
00548             _mutex.unlock();
00549             return -1;
00550         }
00551     } while (SSL_pending(_ssl) && !readBlocked);
00552 
00553     bpos += rlen;
00554     while (true)
00555     {
00556         FD_ZERO(&rset);
00557         FD_ZERO(&wset);
00558         FD_SET(getSock(), &rset);
00559         FD_SET(getSock(), &wset);
00560 
00561         int activity = select(getSock() + 1, &rset, &wset, 0, 0);
00562         if (activity > 0)
00563         {
00564             if ((FD_ISSET(getSock(),&rset) && !writeBlockedOnRead)
00565                     || (readBlockedOnWrite && FD_ISSET(getSock(), &wset)))
00566             {
00567                 goto loop;
00568             }
00569         }
00570         else
00571         {
00572             ERR_error_string_n(ERR_get_error(), errmsg, sizeof(errmsg));
00573             WRITELOG("TLSStack::recv() select %s\n", errmsg);
00574             _busy = false;
00575             _mutex.unlock();
00576             return -1;
00577         }
00578     }
00579 }
00580 
00581 void Network::close(void)
00582 {
00583     _mutex.lock();
00584     if (_secureFlg)
00585     {
00586         if (_ssl)
00587         {
00588             SSL_shutdown(_ssl);
00589             SSL_free(_ssl);
00590             _numOfInstance--;
00591             _ssl = 0;
00592             _sslValid = false;
00593             _busy = false;
00594         }
00595         if (_session && _numOfInstance == 0)
00596         {
00597             SSL_SESSION_free(_session);
00598             _session = 0;
00599         }
00600         if (_ctx && _numOfInstance == 0)
00601         {
00602             SSL_CTX_free(_ctx);
00603             _ctx = 0;
00604             ERR_free_strings();
00605         }
00606     }
00607     TCPStack::close();
00608     _mutex.unlock();
00609 }
00610 
00611 bool Network::isValid()
00612 {
00613     if ( TCPStack::isValid() )
00614     {
00615         if (_secureFlg)
00616         {
00617             if (_sslValid && !_busy)
00618             {
00619                 return true;
00620             }
00621         }
00622         else
00623         {
00624             return true;
00625         }
00626     }
00627     return false;
00628 }
00629 
00630 int Network::getSock()
00631 {
00632     return TCPStack::getSock();
00633 }
00634 
00635 bool Network::isSecure()
00636 {
00637     return _secureFlg;
00638 }
00639