turnasyncsocket.cxx

这是国外的resip协议栈

#include "TurnAsyncSocket.hxx"
#include "../AsyncSocketBase.hxx"
#include "ErrorCode.hxx"
#include <boost/bind.hpp>
#include <rutil/MD5Stream.hxx>
#include <rutil/WinLeakCheck.hxx>
#include <rutil/Logger.hxx>
#include "../ReTurnSubsystem.hxx"

#define RESIPROCATE_SUBSYSTEM ReTurnSubsystem::RETURN

using namespace std;
using namespace resip;

#define UDP_RT0 100  // RTO - Estimate of Roundtrip time - 100ms is recommened for fixed line transport - the initial value should be configurable
                     // Should also be calculation this on the fly
#define UDP_MAX_RETRANSMITS  7       // Defined by RFC3489-bis11
#define TCP_RESPONSE_TIME 7900       // Defined by RFC3489-bis11
#define UDP_FINAL_REQUEST_TIME (UDP_RT0 * 16)  // Defined by RFC3489-bis11

namespace reTurn {

// Initialize static members
unsigned int TurnAsyncSocket::UnspecifiedLifetime = 0xFFFFFFFF;
unsigned int TurnAsyncSocket::UnspecifiedBandwidth = 0xFFFFFFFF; 
unsigned short TurnAsyncSocket::UnspecifiedPort = 0;
asio::ip::address TurnAsyncSocket::UnspecifiedIpAddress = asio::ip::address::from_string("0.0.0.0");

TurnAsyncSocket::TurnAsyncSocket(asio::io_service& ioService, 
                                 AsyncSocketBase& asyncSocketBase,
                                 TurnAsyncSocketHandler* turnAsyncSocketHandler,
                                 const asio::ip::address& address, 
                                 unsigned short port,
                                 bool turnFraming) : 
   mIOService(ioService),
   mTurnAsyncSocketHandler(turnAsyncSocketHandler),
   mTurnFraming(turnFraming),
   mLocalBinding(StunTuple::None /* Set properly by sub class */, address, port),
   mHaveAllocation(false),
   mActiveDestination(0),
   mAsyncSocketBase(asyncSocketBase),
   mCloseAfterDestroyAllocationFinishes(false),
   mAllocationTimer(ioService)
{
}

TurnAsyncSocket::~TurnAsyncSocket()
{
   clearActiveRequestMap();
   cancelAllocationTimer();
   DebugLog(<< "TurnAsyncSocket::~TurnAsyncSocket destroyed!");
}

void
TurnAsyncSocket::disableTurnAsyncHandler()
{
   mTurnAsyncSocketHandler = 0;
}

void
TurnAsyncSocket::requestSharedSecret()
{
   mGuards.push(mAsyncSocketBase.shared_from_this());
   mIOService.post(boost::bind(&TurnAsyncSocket::doRequestSharedSecret, this));
}

void
TurnAsyncSocket::doRequestSharedSecret()
{
   GuardReleaser guardReleaser(mGuards);
   // Should we check here if TLS and deny?

   // Ensure Connected
   if(!mAsyncSocketBase.isConnected())
   {
      if(mTurnAsyncSocketHandler) mTurnAsyncSocketHandler->onSharedSecretFailure(getSocketDescriptor(), asio::error_code(reTurn::NotConnected, asio::error::misc_category));
   }
   else
   {
      // Form Shared Secret request
      StunMessage* request = createNewStunMessage(StunMessage::StunClassRequest, StunMessage::SharedSecretMethod);

      // Send the Request and start transaction timers
      sendStunMessage(request);
   }
}

void
TurnAsyncSocket::setUsernameAndPassword(const char* username, const char* password, bool shortTermAuth)
{
   mGuards.push(mAsyncSocketBase.shared_from_this());
   mIOService.post(boost::bind(&TurnAsyncSocket::doSetUsernameAndPassword, this, new Data(username), new Data(password), shortTermAuth));
}

void 
TurnAsyncSocket::doSetUsernameAndPassword(Data* username, Data* password, bool shortTermAuth)
{
   GuardReleaser guardReleaser(mGuards);
   mUsername = *username;
   mPassword = *password;
   if(shortTermAuth)
   {
      // If we are using short term auth, then use short term password as HMAC key
      mHmacKey = *password;
   }
   delete username;
   delete password;
}

void 
TurnAsyncSocket::bindRequest()
{
   mGuards.push(mAsyncSocketBase.shared_from_this());
   mIOService.post(boost::bind(&TurnAsyncSocket::doBindRequest, this));
}

void 
TurnAsyncSocket::doBindRequest()
{
   GuardReleaser guardReleaser(mGuards);
   // Ensure Connected
   if(!mAsyncSocketBase.isConnected())
   {
      if(mTurnAsyncSocketHandler) mTurnAsyncSocketHandler->onBindFailure(getSocketDescriptor(), asio::error_code(reTurn::NotConnected, asio::error::misc_category));
   }
   else
   {
      // Form Stun Bind request
      StunMessage* request = createNewStunMessage(StunMessage::StunClassRequest, StunMessage::BindMethod);
      sendStunMessage(request);
   }
}

void
TurnAsyncSocket::createAllocation(unsigned int lifetime,
                                  unsigned int bandwidth,
                                  unsigned short requestedPortProps, 
                                  unsigned short requestedPort,
                                  StunTuple::TransportType requestedTransportType, 
                                  const asio::ip::address &requestedIpAddress)
{
   mGuards.push(mAsyncSocketBase.shared_from_this());
   mIOService.post(boost::bind(&TurnAsyncSocket::doCreateAllocation, this, lifetime, 
                                                                           bandwidth, 
                                                                           requestedPortProps, 
                                                                           requestedPort, 
                                                                           requestedTransportType, 
                                                                           requestedIpAddress));
}

void
TurnAsyncSocket::doCreateAllocation(unsigned int lifetime,
                                    unsigned int bandwidth,
                                    unsigned short requestedPortProps, 
                                    unsigned short requestedPort,
                                    StunTuple::TransportType requestedTransportType, 
                                    const asio::ip::address &requestedIpAddress)
{
   GuardReleaser guardReleaser(mGuards);

   // Store Allocation Properties
   mRequestedTransportType = requestedTransportType;

   // Relay Transport Type is requested type or socket type
   if(mRequestedTransportType != StunTuple::None)
   {
      mRelayTransportType = mRequestedTransportType;
   }
   else
   {
      mRelayTransportType = mLocalBinding.getTransportType();
   }

   // Ensure Connected
   if(!mAsyncSocketBase.isConnected())
   {
      if(mTurnAsyncSocketHandler) mTurnAsyncSocketHandler->onAllocationFailure(getSocketDescriptor(), asio::error_code(reTurn::NotConnected, asio::error::misc_category));
      return;
   }

   if(mHaveAllocation)
   {
      if(mTurnAsyncSocketHandler) mTurnAsyncSocketHandler->onAllocationFailure(getSocketDescriptor(), asio::error_code(reTurn::AlreadyAllocated, asio::error::misc_category));
      return;
   }

   // Form Turn Allocate request
   StunMessage* request = createNewStunMessage(StunMessage::StunClassRequest, StunMessage::TurnAllocateMethod);
   if(lifetime != UnspecifiedLifetime)
   {
      request->mHasTurnLifetime = true;
      request->mTurnLifetime = lifetime;
   }
   if(bandwidth != UnspecifiedBandwidth)
   {
      request->mHasTurnBandwidth = true;
      request->mTurnBandwidth = bandwidth;
   }
   if(requestedTransportType != StunTuple::None && requestedTransportType != StunTuple::TLS)
   {      
      request->mHasTurnRequestedTransport = true;
      if(requestedTransportType == StunTuple::UDP)
      {
         request->mTurnRequestedTransport = StunMessage::RequestedTransportUdp;
      }
      else if(requestedTransportType == StunTuple::TCP &&
              mLocalBinding.getTransportType() != StunTuple::UDP)  // Ensure client is not requesting TCP over a UDP transport
      {
         request->mTurnRequestedTransport = StunMessage::RequestedTransportTcp;
      }
      else
      {
         if(mTurnAsyncSocketHandler) mTurnAsyncSocketHandler->onAllocationFailure(getSocketDescriptor(), asio::error_code(reTurn::InvalidRequestedTransport, asio::error::misc_category));
         delete request;
         return;
      }
   }
   if(requestedIpAddress != UnspecifiedIpAddress)
   {
      request->mHasTurnRequestedIp = true;
      StunTuple requestedIpTuple(StunTuple::None, requestedIpAddress, 0);
      StunMessage::setStunAtrAddressFromTuple(request->mTurnRequestedIp, requestedIpTuple);
   }
   if(requestedPortProps != StunMessage::PortPropsNone || requestedPort != UnspecifiedPort)
   {
      request->mHasTurnRequestedPortProps = true;
      request->mTurnRequestedPortProps.props = requestedPortProps;
      request->mTurnRequestedPortProps.port = requestedPort;
   }
      
   sendStunMessage(request);
}
   
void 
TurnAsyncSocket::refreshAllocation(unsigned int lifetime)
{
   mGuards.push(mAsyncSocketBase.shared_from_this());
   mIOService.post(boost::bind(&TurnAsyncSocket::doRefreshAllocation, this, lifetime));
}

void 
TurnAsyncSocket::doRefreshAllocation(unsigned int lifetime)
{
   GuardReleaser guardReleaser(mGuards);
   if(!mHaveAllocation)
   {
      if(mTurnAsyncSocketHandler) mTurnAsyncSocketHandler->onRefreshFailure(getSocketDescriptor(), asio::error_code(NoAllocation, asio::error::misc_category));
      if(mCloseAfterDestroyAllocationFinishes)
      {
         mHaveAllocation = false;
         actualClose();
      }
      return;
   }

   // Form Turn Refresh request
   StunMessage* request = createNewStunMessage(StunMessage::StunClassRequest, StunMessage::TurnRefreshMethod);
   if(lifetime != UnspecifiedLifetime)
   {
      request->mHasTurnLifetime = true;
      request->mTurnLifetime = lifetime;
   }
   //if(mRequestedBandwidth != UnspecifiedBandwidth)
   //{
   //   request.mHasTurnBandwidth = true;
   //   request.mTurnBandwidth = mRequestedBandwidth;
   //}

   sendStunMessage(request);
}

void 
TurnAsyncSocket::destroyAllocation()
{
   mGuards.push(mAsyncSocketBase.shared_from_this());
   mIOService.post(boost::bind(&TurnAsyncSocket::doDestroyAllocation, this));
}

void 
TurnAsyncSocket::doDestroyAllocation()
{
   doRefreshAllocation(0);
}

void
TurnAsyncSocket::setActiveDestination(const asio::ip::address& address, unsigned short port)
{
   mGuards.push(mAsyncSocketBase.shared_from_this());
   mIOService.post(boost::bind(&TurnAsyncSocket::doSetActiveDestination, this, address, port));
}

void
TurnAsyncSocket::doSetActiveDestination(const asio::ip::address& address, unsigned short port)
{
   GuardReleaser guardReleaser(mGuards);

   // Setup Remote Peer 
   StunTuple remoteTuple(mRelayTransportType, address, port);
   RemotePeer* remotePeer = mChannelManager.findRemotePeerByPeerAddress(remoteTuple);
   if(remotePeer)
   {
      mActiveDestination = remotePeer;
   }
   else
   {
      // No remote peer yet (ie. not data sent or received from remote peer) - so create one
      mActiveDestination = mChannelManager.createRemotePeer(remoteTuple, mChannelManager.getNextChannelNumber(), 0);
      assert(mActiveDestination);
   }
   DebugLog(<< "TurnAsyncSocket::doSetActiveDestination: Active Destination set to: " << remoteTuple);
   if(mTurnAsyncSocketHandler) mTurnAsyncSocketHandler->onSetActiveDestinationSuccess(getSocketDescriptor());
}

void
TurnAsyncSocket::clearActiveDestination()
{
   mGuards.push(mAsyncSocketBase.shared_from_this());
   mIOService.post(boost::bind(&TurnAsyncSocket::doClearActiveDestination, this));
}

void
TurnAsyncSocket::doClearActiveDestination()
{
   GuardReleaser guardReleaser(mGuards);

   // ensure there is an allocation
   if(!mHaveAllocation)
   {
      if(mTurnAsyncSocketHandler) mTurnAsyncSocketHandler->onClearActiveDestinationFailure(getSocketDescriptor(), asio::error_code(reTurn::NoAllocation, asio::error::misc_category));
      return;
   }

   mActiveDestination = 0;
   if(mTurnAsyncSocketHandler) mTurnAsyncSocketHandler->onClearActiveDestinationSuccess(getSocketDescriptor());
}

StunMessage* 
TurnAsyncSocket::createNewStunMessage(UInt16 stunclass, UInt16 method, bool addAuthInfo)
{
   StunMessage* msg = new StunMessage();
   msg->createHeader(stunclass, method);

   if(addAuthInfo && !mUsername.empty() && !mHmacKey.empty())
   {
      msg->mHasMessageIntegrity = true;
      msg->setUsername(mUsername.c_str()); 
      msg->mHmacKey = mHmacKey;
      if(!mRealm.empty())
      {
         msg->setRealm(mRealm.c_str());
      }
      if(!mNonce.empty())
      {
         msg->setNonce(mNonce.c_str());
      }
   }
   return msg;
}

void
TurnAsyncSocket::sendStunMessage(StunMessage* message, bool reTransmission)
{
#define REQUEST_BUFFER_SIZE 1024
   boost::shared_ptr<DataBuffer> buffer = AsyncSocketBase::allocateBuffer(REQUEST_BUFFER_SIZE);
   unsigned int bufferSize;
   if(mTurnFraming)  
   {
      bufferSize = message->stunEncodeFramedMessage((char*)buffer->data(), REQUEST_BUFFER_SIZE);
   }
   else
   {
      bufferSize = message->stunEncodeMessage((char*)buffer->data(), REQUEST_BUFFER_SIZE);
   }
   buffer->truncate(bufferSize);  // set size to real size

   if(!reTransmission)
   {
      // If message is a request, then start appropriate transaction and retranmission timers
      if(message->mClass == StunMessage::StunClassRequest)
      {
         boost::shared_ptr<RequestEntry> requestEntry(new RequestEntry(mIOService, this, message));
         mActiveRequestMap[message->mHeader.magicCookieAndTid] = requestEntry;
         requestEntry->startTimer();
      }
      else
      {
         delete message;
      }
   }

   send(buffer);
}

void 
TurnAsyncSocket::handleReceivedData(const asio::ip::address& address, unsigned short port, boost::shared_ptr<DataBuffer>& data)
{
   if(mTurnFraming)
   {
      if(data->size() > 4)
      {
         // Get Channel number
         unsigned short channelNumber;
         memcpy(&channelNumber, &(*data)[0], 2);
         channelNumber = ntohs(channelNumber);

         if(channelNumber == 0)
         {
            // Handle Stun Message
            StunMessage* stunMsg = new StunMessage(mLocalBinding, 
                                                   StunTuple(mLocalBinding.getTransportType(), mAsyncSocketBase.getConnectedAddress(), mAsyncSocketBase.getConnectedPort()), 
                                                   &(*data)[4], data->size()-4);