/* * libjingle * Copyright 2004--2005, Google Inc. * * Redistribution and use in source and binary forms, with or without * modification, are permitted provided that the following conditions are met: * * 1. Redistributions of source code must retain the above copyright notice, * this list of conditions and the following disclaimer. * 2. Redistributions in binary form must reproduce the above copyright notice, * this list of conditions and the following disclaimer in the documentation * and/or other materials provided with the distribution. * 3. The name of the author may not be used to endorse or promote products * derived from this software without specific prior written permission. * * THIS SOFTWARE IS PROVIDED BY THE AUTHOR ``AS IS'' AND ANY EXPRESS OR IMPLIED * WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE IMPLIED WARRANTIES OF * MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO * EVENT SHALL THE AUTHOR BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, * SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO, * PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR PROFITS; * OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, * WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR * OTHERWISE) ARISING IN ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF * ADVISED OF THE POSSIBILITY OF SUCH DAMAGE. */ #include "talk/base/common.h" #include "talk/p2p/base/transport.h" #include "talk/p2p/base/sessionmanager.h" #include "talk/p2p/base/transportchannelimpl.h" #include "talk/p2p/base/constants.h" #include "talk/xmllite/xmlelement.h" #include "talk/xmpp/constants.h" namespace { struct ChannelParams { std::string name; std::string session_type; cricket::TransportChannelImpl* channel; buzz::XmlElement* elem; ChannelParams() : channel(NULL), elem(NULL) {} }; typedef talk_base::TypedMessageData ChannelMessage; const int MSG_CREATECHANNEL = 1; const int MSG_DESTROYCHANNEL = 2; const int MSG_DESTROYALLCHANNELS = 3; const int MSG_CONNECTCHANNELS = 4; const int MSG_RESETCHANNELS = 5; const int MSG_ONSIGNALINGREADY = 6; const int MSG_FORWARDCHANNELMESSAGE = 7; const int MSG_READSTATE = 8; const int MSG_WRITESTATE = 9; const int MSG_REQUESTSIGNALING = 10; const int MSG_ONCHANNELMESSAGE = 11; const int MSG_CONNECTING = 12; } // namespace namespace cricket { Transport::Transport(SessionManager* session_manager, const std::string& name) : session_manager_(session_manager), name_(name), destroyed_(false), readable_(false), writable_(false), connect_requested_(false), allow_local_ips_(false) { ASSERT(session_manager_->signaling_thread()->IsCurrent()); } Transport::~Transport() { ASSERT(session_manager_->signaling_thread()->IsCurrent()); ASSERT(destroyed_); } TransportChannelImpl* Transport::CreateChannel(const std::string& name, const std::string &session_type) { ChannelParams params; params.name = name; params.session_type = session_type; ChannelMessage msg(¶ms); session_manager_->worker_thread()->Send(this, MSG_CREATECHANNEL, &msg); return msg.data()->channel; } TransportChannelImpl* Transport::CreateChannel_w(const std::string& name, const std::string &session_type) { ASSERT(session_manager_->worker_thread()->IsCurrent()); TransportChannelImpl* impl = CreateTransportChannel(name, session_type); impl->SignalReadableState.connect(this, &Transport::OnChannelReadableState); impl->SignalWritableState.connect(this, &Transport::OnChannelWritableState); impl->SignalRequestSignaling.connect( this, &Transport::OnChannelRequestSignaling); impl->SignalChannelMessage.connect(this, &Transport::OnChannelMessage); talk_base::CritScope cs(&crit_); ASSERT(channels_.find(name) == channels_.end()); channels_[name] = impl; destroyed_ = false; if (connect_requested_) { impl->Connect(); if (channels_.size() == 1) { // If this is the first channel, then indicate that we have started // connecting. session_manager_->signaling_thread()->Post(this, MSG_CONNECTING, NULL); } } return impl; } TransportChannelImpl* Transport::GetChannel(const std::string& name) { talk_base::CritScope cs(&crit_); ChannelMap::iterator iter = channels_.find(name); return (iter != channels_.end()) ? iter->second : NULL; } bool Transport::HasChannels() { talk_base::CritScope cs(&crit_); return !channels_.empty(); } void Transport::DestroyChannel(const std::string& name) { ChannelParams params; params.name = name; ChannelMessage msg(¶ms); session_manager_->worker_thread()->Send(this, MSG_DESTROYCHANNEL, &msg); } void Transport::DestroyChannel_w(const std::string& name) { ASSERT(session_manager_->worker_thread()->IsCurrent()); TransportChannelImpl* impl = NULL; { talk_base::CritScope cs(&crit_); ChannelMap::iterator iter = channels_.find(name); ASSERT(iter != channels_.end()); impl = iter->second; channels_.erase(iter); } if (connect_requested_ && channels_.empty()) { // We're not longer attempting to connect. session_manager_->signaling_thread()->Post(this, MSG_CONNECTING, NULL); } if (impl) DestroyTransportChannel(impl); } void Transport::ConnectChannels() { ASSERT(session_manager_->signaling_thread()->IsCurrent()); session_manager_->worker_thread()->Post(this, MSG_CONNECTCHANNELS, NULL); } void Transport::ConnectChannels_w() { ASSERT(session_manager_->worker_thread()->IsCurrent()); if (connect_requested_) return; connect_requested_ = true; session_manager_->signaling_thread()->Post(this, MSG_ONCHANNELMESSAGE, NULL); CallChannels_w(&TransportChannelImpl::Connect); if (!channels_.empty()) { session_manager_->signaling_thread()->Post(this, MSG_CONNECTING, NULL); } } void Transport::OnConnecting_s() { ASSERT(session_manager_->signaling_thread()->IsCurrent()); SignalConnecting(this); } void Transport::DestroyAllChannels() { ASSERT(session_manager_->signaling_thread()->IsCurrent()); session_manager_->worker_thread()->Send(this, MSG_DESTROYALLCHANNELS, NULL); destroyed_ = true; } void Transport::DestroyAllChannels_w() { ASSERT(session_manager_->worker_thread()->IsCurrent()); std::vector impls; { talk_base::CritScope cs(&crit_); for (ChannelMap::iterator iter = channels_.begin(); iter != channels_.end(); ++iter) { impls.push_back(iter->second); } channels_.clear(); } for (size_t i = 0; i < impls.size(); ++i) DestroyTransportChannel(impls[i]); } void Transport::ResetChannels() { ASSERT(session_manager_->signaling_thread()->IsCurrent()); session_manager_->worker_thread()->Post(this, MSG_RESETCHANNELS, NULL); } void Transport::ResetChannels_w() { ASSERT(session_manager_->worker_thread()->IsCurrent()); // We are no longer attempting to connect connect_requested_ = false; // Clear out the old messages, they aren't relevant talk_base::CritScope cs(&crit_); for (size_t i=0; isignaling_thread()->IsCurrent()); session_manager_->worker_thread()->Post(this, MSG_ONSIGNALINGREADY, NULL); // Notify the subclass. OnTransportSignalingReady(); } void Transport::CallChannels_w(TransportChannelFunc func) { ASSERT(session_manager_->worker_thread()->IsCurrent()); talk_base::CritScope cs(&crit_); for (ChannelMap::iterator iter = channels_.begin(); iter != channels_.end(); ++iter) { ((iter->second)->*func)(); } } void Transport::ForwardChannelMessage(const std::string& name, buzz::XmlElement* elem) { ASSERT(session_manager_->signaling_thread()->IsCurrent()); ASSERT(HasChannel(name)); ChannelParams* params = new ChannelParams(); params->name = name; params->elem = elem; ChannelMessage* msg = new ChannelMessage(params); session_manager_->worker_thread()->Post(this, MSG_FORWARDCHANNELMESSAGE, msg); } void Transport::ForwardChannelMessage_w(const std::string& name, buzz::XmlElement* elem) { ASSERT(session_manager_->worker_thread()->IsCurrent()); ChannelMap::iterator iter = channels_.find(name); // It's ok for a channel to go away while this message is in transit. if (iter != channels_.end()) { iter->second->OnChannelMessage(elem); } delete elem; } void Transport::OnChannelReadableState(TransportChannel* channel) { ASSERT(session_manager_->worker_thread()->IsCurrent()); session_manager_->signaling_thread()->Post(this, MSG_READSTATE, NULL); } void Transport::OnChannelReadableState_s() { ASSERT(session_manager_->signaling_thread()->IsCurrent()); bool readable = GetTransportState_s(true); if (readable_ != readable) { readable_ = readable; SignalReadableState(this); } } void Transport::OnChannelWritableState(TransportChannel* channel) { ASSERT(session_manager_->worker_thread()->IsCurrent()); session_manager_->signaling_thread()->Post(this, MSG_WRITESTATE, NULL); } void Transport::OnChannelWritableState_s() { ASSERT(session_manager_->signaling_thread()->IsCurrent()); bool writable = GetTransportState_s(false); if (writable_ != writable) { writable_ = writable; SignalWritableState(this); } } bool Transport::GetTransportState_s(bool read) { ASSERT(session_manager_->signaling_thread()->IsCurrent()); bool result = false; talk_base::CritScope cs(&crit_); for (ChannelMap::iterator iter = channels_.begin(); iter != channels_.end(); ++iter) { bool b = (read ? iter->second->readable() : iter->second->writable()); result = result || b; } return result; } void Transport::OnChannelRequestSignaling() { ASSERT(session_manager_->worker_thread()->IsCurrent()); session_manager_->signaling_thread()->Post(this, MSG_REQUESTSIGNALING, NULL); } void Transport::OnChannelRequestSignaling_s() { ASSERT(session_manager_->signaling_thread()->IsCurrent()); SignalRequestSignaling(this); } void Transport::OnChannelMessage(TransportChannelImpl* impl, buzz::XmlElement* elem) { ASSERT(session_manager_->worker_thread()->IsCurrent()); talk_base::CritScope cs(&crit_); messages_.push_back(elem); // We hold any messages until the client lets us connect. if (connect_requested_) { session_manager_->signaling_thread()->Post( this, MSG_ONCHANNELMESSAGE, NULL); } } void Transport::OnChannelMessage_s() { ASSERT(session_manager_->signaling_thread()->IsCurrent()); ASSERT(connect_requested_); std::vector msgs; { talk_base::CritScope cs(&crit_); msgs.swap(messages_); } if (!msgs.empty()) OnTransportChannelMessages(msgs); } void Transport::OnTransportChannelMessages( const std::vector& msgs) { std::vector elems; for (size_t i = 0; i < msgs.size(); ++i) { buzz::XmlElement* elem = new buzz::XmlElement(buzz::QName(name(), "transport")); elem->AddElement(msgs[i]); elems.push_back(elem); } SignalTransportMessage(this, elems); } void Transport::OnMessage(talk_base::Message* msg) { switch (msg->message_id) { case MSG_CREATECHANNEL: { ChannelParams* params = static_cast(msg->pdata)->data(); params->channel = CreateChannel_w(params->name, params->session_type); } break; case MSG_DESTROYCHANNEL: { ChannelParams* params = static_cast(msg->pdata)->data(); DestroyChannel_w(params->name); } break; case MSG_CONNECTCHANNELS: ConnectChannels_w(); break; case MSG_RESETCHANNELS: ResetChannels_w(); break; case MSG_DESTROYALLCHANNELS: DestroyAllChannels_w(); break; case MSG_ONSIGNALINGREADY: CallChannels_w(&TransportChannelImpl::OnSignalingReady); break; case MSG_FORWARDCHANNELMESSAGE: { ChannelParams* params = static_cast(msg->pdata)->data(); ForwardChannelMessage_w(params->name, params->elem); delete params; } break; case MSG_CONNECTING: OnConnecting_s(); break; case MSG_READSTATE: OnChannelReadableState_s(); break; case MSG_WRITESTATE: OnChannelWritableState_s(); break; case MSG_REQUESTSIGNALING: OnChannelRequestSignaling_s(); break; case MSG_ONCHANNELMESSAGE: OnChannelMessage_s(); break; } } bool Transport::BadRequest(const buzz::XmlElement* stanza, const std::string& text, const buzz::XmlElement* extra_info) { SignalTransportError(this, stanza, buzz::QN_STANZA_BAD_REQUEST, "modify", text, extra_info); return false; } bool Transport::ParseAddress(const buzz::XmlElement* stanza, const buzz::XmlElement* elem, talk_base::SocketAddress* address) { ASSERT(elem->HasAttr(QN_ADDRESS)); ASSERT(elem->HasAttr(QN_PORT)); // Record the parts of the address. address->SetIP(elem->Attr(QN_ADDRESS)); std::istringstream ist(elem->Attr(QN_PORT)); int port; ist >> port; address->SetPort(port); // No address zero. if (address->IsAny()) return BadRequest(stanza, "candidate has address of zero", NULL); // Always disallow addresses that refer to the local host. if (address->IsLocalIP() && !allow_local_ips_) return BadRequest(stanza, "candidate has local IP address", NULL); // Disallow all ports below 1024, except for 80 and 443 on public addresses. if (port < 1024) { if ((port != 80) && (port != 443)) return BadRequest(stanza, "candidate has port below 1024, but not 80 or 443", NULL); if (address->IsPrivateIP()) { return BadRequest(stanza, "candidate has port of 80 or 443 with private " "IP address", NULL); } } return true; } } // namespace cricket