Line data    Source code

       1             : /*
       2             :  * Copyright (c) 2013 Juniper Networks, Inc. All rights reserved.
       3             :  * Copyright Contributors to the OpenSDN project
       4             :  */
       5             : 
       6             : #include "base/regex.h"
       7             : #include "xmpp/xmpp_session.h"
       8             : 
       9             : #include "xmpp/xmpp_connection.h"
      10             : #include "xmpp/xmpp_log.h"
      11             : #include "xmpp/xmpp_proto.h"
      12             : #include "xmpp/xmpp_server.h"
      13             : #include "xmpp/xmpp_state_machine.h"
      14             : 
      15             : #include "sandesh/sandesh_trace.h"
      16             : #include "sandesh/xmpp_trace_sandesh_types.h"
      17             : 
      18             : using namespace std;
      19             : using contrail::regex;
      20             : using contrail::regex_match;
      21             : using contrail::regex_search;
      22             : 
      23             : using boost::asio::mutable_buffer;
      24             : 
      25             : const regex XmppSession::patt_(rXMPP_MESSAGE);
      26             : const regex XmppSession::stream_patt_(rXMPP_STREAM_START);
      27             : const regex XmppSession::stream_res_end_(rXMPP_STREAM_END);
      28             : const regex XmppSession::whitespace_(sXMPP_WHITESPACE);
      29             : const regex XmppSession::stream_features_patt_(rXMPP_STREAM_FEATURES);
      30             : const regex XmppSession::starttls_patt_(rXMPP_STREAM_STARTTLS);
      31             : const regex XmppSession::proceed_patt_(rXMPP_STREAM_PROCEED);
      32             : const regex XmppSession::end_patt_(rXMPP_STREAM_STANZA_END);
      33             : 
      34       14313 : XmppSession::XmppSession(XmppConnectionManager *manager, SslSocket *socket,
      35       14313 :     bool async_ready)
      36             :     : SslSession(manager, socket, async_ready),
      37       14313 :       manager_(manager),
      38       14313 :       connection_(NULL),
      39       14313 :       tag_known_(0),
      40       14313 :       task_instance_(-1),
      41       14313 :       stats_(XmppStanza::RESERVED_STANZA, XmppSession::StatsPair(0, 0)),
      42       14313 :       keepalive_probes_(kSessionKeepaliveProbes) {
      43       14313 :     buf_.reserve(kMaxMessageSize);
      44       14312 :     offset_ = buf_.begin();
      45       14313 :     stream_open_matched_ = false;
      46       14313 : }
      47             : 
      48       28625 : XmppSession::~XmppSession() {
      49       14313 :     set_observer(NULL);
      50       14312 :     connection_ = NULL;
      51       28624 : }
      52             : 
      53       14392 : void XmppSession::SetConnection(XmppConnection *connection) {
      54       14392 :     assert(connection);
      55       14392 :     connection_ = connection;
      56       14392 :     task_instance_ = connection_->GetTaskInstance();
      57       14392 : }
      58             : 
      59             : //
      60             : // Dissociate the connection from the this XmppSession.
      61             : // Do not invalidate the task_instance since it can be used to spawn an
      62             : // io::ReaderTask while this method is being executed.
      63             : //
      64       33518 : void XmppSession::ClearConnection() {
      65       33518 :     connection_ = NULL;
      66       33518 : }
      67             : 
      68             : //
      69             : // Concurrency: called in the context of bgp::Config task.
      70             : //
      71             : // Process write ready callback.
      72             : //
      73           1 : void XmppSession::ProcessWriteReady() {
      74           1 :     if (!connection_)
      75           0 :         return;
      76           1 :     connection_->WriteReady();
      77             : }
      78             : 
      79             : //
      80             : // Concurrency: called in the context of io thread.
      81             : //
      82             : // Handle write ready callback.
      83             : //
      84             : // Enqueue session to the XmppConnectionManager. The session is added to a
      85             : // WorkQueue gets processed in the context of bgp::Config task. Doing this
      86             : // ensures that we don't access the XmppConnection while the XmppConnection
      87             : // is trying to clear our back pointer to it.
      88             : //
      89             : // We can ignore any errors since the StateMachine will get informed of the
      90             : // TcpSession close independently and react to it.
      91             : //
      92           1 : void XmppSession::WriteReady(const boost::system::error_code &error) {
      93           1 :     if (error)
      94           0 :         return;
      95           1 :     manager_->EnqueueSession(this);
      96             : }
      97             : 
      98          30 : XmppSession::StatsPair XmppSession::Stats(unsigned int type) const {
      99          30 :     assert (type < (unsigned int)XmppStanza::RESERVED_STANZA);
     100          30 :     return stats_[type];
     101             : }
     102             : 
     103     4152396 : void XmppSession::IncStats(unsigned int type, uint64_t bytes) {
     104     4152396 :     assert (type < (unsigned int)XmppStanza::RESERVED_STANZA);
     105     4152396 :     stats_[type].first++;
     106     4152401 :     stats_[type].second += bytes;
     107     4152404 : }
     108             : 
     109        6588 : boost::system::error_code XmppSession::EnableTcpKeepalive(int hold_time) {
     110        6588 :     char *keepalive_time_str = getenv("TCP_KEEPALIVE_SECONDS");
     111        6588 :     if (keepalive_time_str) {
     112           0 :         hold_time = strtoul(keepalive_time_str, NULL, 0) * 3;
     113           0 :         if (!hold_time)
     114           0 :             return boost::system::error_code();
     115             :     }
     116             : 
     117        6588 :     if (hold_time <= 9) {
     118           0 :         hold_time = 9; // min hold-time in secs.
     119             :     }
     120        6588 :     hold_time = ((hold_time > 18)? hold_time/2 : hold_time);
     121        6588 :     keepalive_idle_time_ = hold_time/3;
     122        6588 :     keepalive_interval_ =
     123        6588 :         ((hold_time - keepalive_idle_time_)/keepalive_probes_);
     124        6588 :     tcp_user_timeout_ = (hold_time * 1000); // msec
     125             : 
     126        6588 :     return (SetSocketKeepaliveOptions(keepalive_idle_time_,
     127             :                                       keepalive_interval_,
     128             :                                       keepalive_probes_,
     129        6588 :                                       tcp_user_timeout_));
     130             : }
     131             : 
     132     1708963 : regex XmppSession::tag_to_pattern(const char *tag) {
     133     1708963 :     std::string token("</");
     134     1708957 :     token += ++tag;
     135     1708956 :     token += "[\\s\\t\\r\\n]*>";
     136             : 
     137     3417927 :     regex exp(token.c_str());
     138     3417926 :     return exp;
     139     1708961 : }
     140             : 
     141     1164320 : void XmppSession::SetBuf(const std::string &str) {
     142     1164320 :     if (buf_.empty()) {
     143     1112029 :         ReplaceBuf(str);
     144             :     } else {
     145       52295 :         int pos = offset_ - buf_.begin();
     146       52295 :         buf_ += str;
     147       52295 :         offset_ = buf_.begin() + pos;
     148             :     }
     149     1164317 : }
     150             : 
     151     4294327 : void XmppSession::ReplaceBuf(const std::string &str) {
     152     4294327 :     buf_ = str;
     153     4294327 :     buf_.reserve(kMaxMessageSize+8);
     154     4294329 :     offset_ = buf_.begin();
     155     4294329 : }
     156             : 
     157     4152438 : bool XmppSession::LeftOver() const {
     158     4152438 :     if (buf_.empty())
     159           0 :         return false;
     160     4152439 :     return (buf_.end() != offset_);
     161             : }
     162             : 
     163             : // Match a pattern in the buffer. Partially matched string is
     164             : // kept in buf_ for use in conjucntion with next buffer read.
     165     3403303 : int XmppSession::MatchRegex(const regex &patt) {
     166             : 
     167     3403303 :     std::string::const_iterator end = buf_.end();
     168             : 
     169     3403300 :     if (regex_search(offset_, end, res_, patt,
     170     3403288 :                      boost::match_default | boost::match_partial) == 0) {
     171       46247 :         return -1;
     172             :     }
     173     3357041 :     if(res_[0].matched == false) {
     174             :         // partial match
     175        6060 :         offset_ = res_[0].first;
     176        6060 :         return 1;
     177             :     } else {
     178     3350974 :         begin_tag_ = string(res_[0].first, res_[0].second);
     179     3350976 :         offset_ = res_[0].second;
     180     3350972 :         return 0;
     181             :     }
     182             : }
     183             : 
     184     4346607 : bool XmppSession::Match(Buffer buffer, int *result, bool NewBuf) {
     185     4346607 :     const XmppConnection *connection = this->Connection();
     186             : 
     187     4346605 :     if (connection == NULL) {
     188           0 :         return true;
     189             :     }
     190             : 
     191     4346605 :     xmsm::XmState state = connection->GetStateMcState();
     192             :     xmsm::XmOpenConfirmState oc_state =
     193     4346602 :         connection->GetStateMcOpenConfirmState();
     194             : 
     195     4346599 :     if (NewBuf) {
     196     1164304 :         const uint8_t *cp = BufferData(buffer);
     197             :         // TODO Avoid this copy
     198     1164298 :         std::string str(cp, cp + BufferSize(buffer));
     199     1164321 :         XmppSession::SetBuf(str);
     200     1164313 :     }
     201             : 
     202     4346612 :     int m = -1;
     203     4346612 :     *result = 0;
     204             :     do {
     205     6022112 :         if (!tag_known_) {
     206             :             // check for whitespaces
     207     4294531 :             size_t pos = buf_.find_first_not_of(sXMPP_VALIDWS);
     208     4294532 :             if (pos != 0) {
     209     2618821 :                 if (pos == string::npos) pos = buf_.size();
     210     2618822 :                 offset_ = buf_.begin() + pos;
     211     2618844 :                 return false;
     212             :             }
     213             :         }
     214             : 
     215     3403292 :         if (state == xmsm::ACTIVE || state == xmsm::IDLE) {
     216       13037 :             m = MatchRegex(tag_known_ ? stream_res_end_:stream_patt_);
     217     3390255 :         } else if (state == xmsm::CONNECT || state == xmsm::OPENSENT) {
     218             :             // Note, these are client only states
     219       14718 :             if (!stream_open_matched_) {
     220       12566 :                 m = MatchRegex(tag_known_ ? stream_res_end_:stream_patt_);
     221       12566 :                 if ((m == 0) && (tag_known_)) {
     222        6283 :                     stream_open_matched_ = true;
     223             :                 }
     224             :             } else {
     225        2152 :                 m = MatchRegex(tag_known_ ? tag_to_pattern(begin_tag_.c_str()):
     226             :                                             stream_features_patt_);
     227             :             }
     228     3375537 :         } else if ((state == xmsm::OPENCONFIRM) && !(IsSslDisabled())) {
     229       12398 :             if (connection->IsClient()) {
     230        6578 :                 if (oc_state == xmsm::OPENCONFIRM_FEATURE_NEGOTIATION) {
     231        2910 :                     m = MatchRegex(tag_known_ ? end_patt_: proceed_patt_);
     232        2910 :                     if ((m == 0) && (tag_known_)) {
     233             :                         // set the flag, as we do not want OnRead function to
     234             :                         // read any more data from basic socket.
     235        1455 :                         SetSslHandShakeInProgress(true);
     236             :                     }
     237        3668 :                 } else if (oc_state == xmsm::OPENCONFIRM_FEATURE_SUCCESS) {
     238        2910 :                     m = MatchRegex(tag_known_ ? stream_res_end_:stream_patt_);
     239             :                 } else {
     240         758 :                     m = MatchRegex(tag_known_ ? tag_to_pattern(begin_tag_.c_str()):
     241             :                                                 stream_features_patt_);
     242             :                 }
     243             :             } else {
     244        5819 :                 if (oc_state == xmsm::OPENCONFIRM_FEATURE_SUCCESS) {
     245        2910 :                     m = MatchRegex(tag_known_ ? stream_res_end_:stream_patt_);
     246             :                 } else {
     247        2909 :                     m = MatchRegex(tag_known_ ? end_patt_:starttls_patt_);
     248        2909 :                     if ((m == 0) && (tag_known_)) {
     249        1454 :                         SetSslHandShakeInProgress(true);
     250             :                     }
     251             :                 }
     252             :             }
     253     3363139 :         } else if (state == xmsm::OPENCONFIRM || state == xmsm::ESTABLISHED) {
     254     3363144 :             m = MatchRegex(tag_known_ ? tag_to_pattern(begin_tag_.c_str()):patt_);
     255             :         }
     256             : 
     257     3403293 :         if (m == 0) { // full match
     258     3350990 :             *result = 0;
     259     3350990 :             tag_known_ ^= 1;
     260     3350990 :             if (!tag_known_) {
     261             :                 // Found well formed xml
     262     1675490 :                 return false;
     263             :             }
     264       52303 :         } else if (m == -1) { // no match
     265       46245 :             return true;
     266             :         } else {
     267        6058 :             return true; // partial. read more
     268             :         }
     269     1675500 :     } while (true);
     270             : 
     271             :     return true;
     272             : }
     273             : 
     274             : // Read the socket stream and send messages to the connection object.
     275             : // The buffer is copied to local string for regex match.
     276             : // TODO Code need to change st Match() is done on buffer itself.
     277     1164378 : void XmppSession::OnRead(Buffer buffer) {
     278     1164378 :     if (this->Connection() == NULL || !connection_) {
     279             :         // Connection is deleted. Session is being deleted as well
     280             :         // Drop the packet.
     281          76 :         ReleaseBuffer(buffer);
     282          76 :         return;
     283             :     }
     284             : 
     285     1164303 :     if (connection_->disable_read()) {
     286           0 :         ReleaseBuffer(buffer);
     287             : 
     288             :         // Reset the hold timer as we did receive some thing from the peer
     289           0 :         connection_->state_machine()->StartHoldTimer();
     290           0 :         return;
     291             :     }
     292             : 
     293     1164305 :     int result = 0;
     294     1164305 :     bool more = Match(buffer, &result, true);
     295             :     do {
     296     4346633 :         if (more == false) {
     297     4294330 :             if (result < 0) {
     298             :                 // TODO generate error, close connection.
     299      141883 :                 break;
     300             :             }
     301             : 
     302             :             // We got good match. Process the message
     303     4294330 :             std::string::const_iterator st = buf_.begin();
     304     4294323 :             std::string xml = string(st, offset_);
     305             :             // Ensure we have not reached the end
     306     4294308 :             if (buf_.begin() == offset_) { // xml.size() == 0
     307      141892 :                 buf_.clear();
     308      141892 :                 break;
     309             :             }
     310             : 
     311     4152422 :             connection_->ReceiveMsg(this, xml);
     312             : 
     313     4294333 :         } else {
     314             :             // Read more data. Either we have partial match
     315             :             // or no match but in this state we need to keep
     316             :             // reading data.
     317       52303 :             break;
     318             :         }
     319             : 
     320     4152443 :         if (LeftOver()) {
     321     3182309 :             std::string::const_iterator st = buf_.end();
     322     3182308 :             ReplaceBuf(string(offset_, st));
     323     3182308 :             more = Match(buffer, &result, false);
     324             :         } else {
     325             :             // No more data in the Buffer
     326      970128 :             buf_.clear();
     327      970134 :             break;
     328             :         }
     329     3182307 :     } while (true);
     330             : 
     331     1164320 :     ReleaseBuffer(buffer);
     332     1164328 :     return;
     333             : }