Google Git
Sign in
chromium / git / chromium / trunk / . / chrome / browser / chromeos / web_socket_proxy.cc
blob: 0eecd1d885893e7b1a3d6626880ec6de004aadb0 [file] [log] [blame]
// Copyright (c) 2012 The Chromium Authors. All rights reserved.
// Use of this source code is governed by a BSD-style license that can be
// found in the LICENSE file.
#include "chrome/browser/chromeos/web_socket_proxy.h"
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <algorithm>
#include <limits>
#include <list>
#include <map>
#include <vector>
#include <errno.h>
#include <fcntl.h>
#include <netinet/in.h>
#include <signal.h>
#include <sys/socket.h>
#include <sys/types.h>
#include <sys/wait.h>
#include "base/base64.h"
#include "base/basictypes.h"
#include "base/bind.h"
#include "base/bind_helpers.h"
#include "base/lazy_instance.h"
#include "base/logging.h"
#include "base/memory/ref_counted.h"
#include "base/memory/scoped_ptr.h"
#include "base/memory/weak_ptr.h"
#include "base/message_loop.h"
#include "base/sequenced_task_runner_helpers.h"
#include "base/sha1.h"
#include "base/stl_util.h"
#include "base/string_number_conversions.h"
#include "base/string_util.h"
#include "base/sys_byteorder.h"
#include "chrome/browser/chromeos/web_socket_proxy_helper.h"
#include "chrome/browser/internal_auth.h"
#include "chrome/common/chrome_notification_types.h"
#include "chrome/common/url_constants.h"
#include "content/public/browser/browser_thread.h"
#include "content/public/browser/notification_details.h"
#include "content/public/browser/notification_service.h"
#include "content/public/browser/notification_types.h"
#include "content/public/common/url_constants.h"
#include "googleurl/src/gurl.h"
#include "googleurl/src/url_parse.h"
#include "net/base/address_list.h"
#include "net/base/cert_verifier.h"
#include "net/base/host_port_pair.h"
#include "net/base/io_buffer.h"
#include "net/base/net_errors.h"
#include "net/base/ssl_config_service.h"
#include "net/socket/client_socket_factory.h"
#include "net/socket/client_socket_handle.h"
#include "net/socket/ssl_client_socket.h"
#include "net/socket/stream_socket.h"
#include "third_party/libevent/evdns.h"
#include "third_party/libevent/event.h"
using content::BrowserThread;
namespace chromeos {
namespace {
const uint8 kCRLF[] = "\r\n";
const uint8 kCRLFCRLF[] = "\r\n\r\n";
// Not a constant but preprocessor definition for easy concatenation.
#define kProxyPath "/tcpproxy"
enum WebSocketStatusCode {
WS_CLOSE_NORMAL = 1000,
WS_CLOSE_GOING_AWAY = 1001,
WS_CLOSE_PROTOCOL_ERROR = 1002,
WS_CLOSE_UNACCEPTABLE_DATA = 1003,
WS_CLOSE_DESTINATION_ERROR = 4000,
WS_CLOSE_LIMIT_VIOLATION = 4001,
WS_CLOSE_RESOLUTION_FAILED = 4002,
WS_CLOSE_UNEXPECTED = 4003
};
enum WebSocketFrameOpcode {
WS_OPCODE_TEXT = 1,
WS_OPCODE_CLOSE = 8
};
// Fixed-size (one-byte) messages communicated via control pipe.
const char kControlMessageShutdown[] = { '.' };
const char kControlMessageNetworkChange[] = { ':' };
// Returns true on success.
bool SetNonBlock(int fd) {
int flags = fcntl(fd, F_GETFL, 0);
return flags >= 0 && fcntl(fd, F_SETFL, flags | O_NONBLOCK) == 0;
}
// Returns true on success.
bool IgnoreSigPipe() {
struct sigaction sa;
sa.sa_handler = SIG_IGN;
sa.sa_flags = 0;
if (sigemptyset(&sa.sa_mask) || sigaction(SIGPIPE, &sa, 0)) {
LOG(ERROR) << "WebSocketProxy: Failed to disable sigpipe";
return false;
}
return true;
}
uint64 ReadNetworkInteger(uint8* buf, int num_bytes) {
uint64 rv = 0;
DCHECK_GE(num_bytes, 0);
DCHECK_LE(num_bytes, 8);
while (num_bytes--) {
rv <<= 8;
rv += *buf++;
}
return rv;
}
void WriteNetworkInteger(int64 n, uint8* buf, int num_bytes) {
DCHECK_GE(num_bytes, 0);
DCHECK_LE(num_bytes, 8);
while (num_bytes--) {
buf[num_bytes] = n % (1 << 8);
n >>= 8;
}
}
typedef uint8 (*AsciiFilter)(uint8);
uint8 AsciiFilterVerbatim(uint8 c) {
return c;
}
uint8 AsciiFilterLower(uint8 c) {
return base::ToLowerASCII(c);
}
std::string FetchAsciiSnippet(uint8* begin, uint8* end, AsciiFilter filter) {
std::string rv;
for (; begin < end; ++begin) {
if (!isascii(*begin))
return rv;
rv += filter(*begin);
}
return rv;
}
std::string FetchExtensionIdFromOrigin(const std::string &origin) {
GURL url(origin);
if (url.SchemeIs(chrome::kExtensionScheme))
return url.host();
else
return std::string();
}
inline size_t strlen(const void* s) {
return ::strlen(static_cast<const char*>(s));
}
void SendNotification(int port) {
DCHECK(BrowserThread::CurrentlyOn(BrowserThread::UI));
content::NotificationService::current()->Notify(
chrome::NOTIFICATION_WEB_SOCKET_PROXY_STARTED,
content::NotificationService::AllSources(), content::Details<int>(&port));
}
class Conn;
// Websocket to TCP proxy server.
class Serv {
public:
Serv();
~Serv();
// Do not call it twice.
void Run();
// Terminates running server (should be called on a different thread).
void Shutdown();
// Called on network change. Reinitializes DNS resolution service.
void OnNetworkChange();
void ZapConn(Conn*);
void MarkConnImportance(Conn*, bool important);
Conn* GetFreshConn();
bool IsConnSane(Conn*);
void CloseAll();
static void OnConnect(int listening_sock, short event, void*);
static void OnControlRequest(int fd, short event, void*);
struct event_base* evbase() { return evbase_; }
// Libevent base.
struct event_base* evbase_;
// Socket to listen incoming websocket connections.
int listening_sock_;
// TODO(dilmah): remove this extra socket as soon as obsolete
// getPassportForTCP function is removed from webSocketProxyPrivate API.
// Additional socket to listen incoming connections on fixed port 10101.
int extra_listening_sock_;
// Used to communicate control requests: either shutdown request or network
// change notification.
int control_descriptor_[2];
// Flag whether shutdown has been requested.
bool shutdown_requested_;
// List of pending connections; We are trying to keep size of this list
// below kConnPoolLimit in LRU fashion.
typedef std::list<Conn*> ConnPool;
ConnPool conn_pool_;
// Reverse map to look up a connection in a conn_pool.
typedef std::map<Conn*, ConnPool::iterator> RevMap;
RevMap rev_map_;
scoped_ptr<struct event> connection_event_;
scoped_ptr<struct event> control_event_;
// TODO(dilmah): remove this extra event as soon as obsolete
// getPassportForTCP function is removed from webSocketProxyPrivate API.
scoped_ptr<struct event> extra_connection_event_;
DISALLOW_COPY_AND_ASSIGN(Serv);
};
// Connection (amalgamates both channels between proxy and javascript and
// between proxy and destination).
class Conn {
public:
enum Phase {
// Initial stage of connection.
PHASE_WAIT_HANDSHAKE,
PHASE_WAIT_DESTFRAME,
PHASE_WAIT_DESTCONNECT,
// Operational stage of connection.
PHASE_OUTSIDE_FRAME,
PHASE_INSIDE_FRAME_BASE64,
PHASE_INSIDE_FRAME_SKIP,
// Terminal stage of connection.
PHASE_SHUT, // Closing handshake was emitted, buffers may be pending.
PHASE_DEFUNCT // Connection was nuked.
};
// Channel structure (either proxy<->browser or proxy<->destination).
class Chan {
public:
explicit Chan(Conn* master)
: master_(master),
write_pending_(false),
read_bev_(NULL),
write_bev_(NULL),
read_fd_(-1),
write_fd_(-1) {
}
~Chan() {
Zap();
}
// Returns true on success.
bool Write(const void* data, size_t size) {
if (write_bev_ == NULL)
return false;
write_pending_ = true;
return (0 == bufferevent_write(write_bev_, data, size));
}
void Zap() {
if (read_bev_) {
bufferevent_disable(read_bev_, EV_READ);
bufferevent_free(read_bev_);
}
if (write_bev_ && write_bev_ != read_bev_) {
bufferevent_disable(write_bev_, EV_READ);
bufferevent_free(write_bev_);
}
read_bev_ = NULL;
write_bev_ = NULL;
if (write_fd_ && read_fd_ == write_fd_)
shutdown(write_fd_, SHUT_RDWR);
if (write_fd_ >= 0) {
close(write_fd_);
DCHECK_GE(read_fd_, 0);
}
if (read_fd_ && read_fd_ != write_fd_)
close(read_fd_);
read_fd_ = -1;
write_fd_ = -1;
write_pending_ = false;
master_->ConsiderSuicide();
}
void Shut() {
if (!write_pending_)
Zap();
}
int read_fd() const { return read_fd_; }
void set_read_fd(int fd) { read_fd_ = fd; }
int write_fd() const { return write_fd_; }
void set_write_fd(int fd) { write_fd_ = fd; }
bool write_pending() const { return write_pending_; }
void set_write_pending(bool pending) { write_pending_ = pending; }
struct bufferevent* read_bev() const { return read_bev_; }
void set_read_bev(struct bufferevent* bev) { read_bev_ = bev; }
struct bufferevent* write_bev() const { return write_bev_; }
void set_write_bev(struct bufferevent* bev) { write_bev_ = bev; }
private:
Conn* master_;
bool write_pending_; // Whether write buffer is not flushed yet.
struct bufferevent* read_bev_;
struct bufferevent* write_bev_;
// UNIX descriptors.
int read_fd_;
int write_fd_;
DISALLOW_COPY_AND_ASSIGN(Chan);
};
// Status of processing incoming data.
enum Status {
STATUS_OK,
STATUS_INCOMPLETE, // Not all required data is present in buffer yet.
STATUS_SKIP,
STATUS_ABORT // Data is invalid. We must shut connection.
};
// Unfortunately evdns callbacks are uncancellable,
// so potentially we can receive callback for a deleted Conn.
// Even worse, storage of deleted Conn may be reused
// for a new connection and new connection can receive callback
// destined for deleted Conn.
// EventKey is introduced in order to prevent that.
typedef void* EventKey;
typedef std::map<EventKey, Conn*> EventKeyMap;
explicit Conn(Serv* master);
~Conn();
static Conn* Get(EventKey evkey);
void Shut(int status, const void* reason);
void ConsiderSuicide();
Status ConsumeHeader(struct evbuffer*);
Status ConsumeDestframe(struct evbuffer*);
Status ConsumeFrameHeader(struct evbuffer*);
Status ProcessFrameData(struct evbuffer*);
// Return true on success.
bool EmitHandshake(Chan*);
bool EmitFrame(
Chan*, WebSocketFrameOpcode opcode, const void* buf, size_t size);
// Attempts to establish second connection (to remote TCP service).
// Returns true on success.
bool TryConnectDest(const struct sockaddr*, socklen_t);
// Return security origin associated with this connection.
const std::string& GetOrigin();
// Used as libevent callbacks.
static void OnDestConnectTimeout(int, short, EventKey);
static void OnPrimchanRead(struct bufferevent*, EventKey);
static void OnPrimchanWrite(struct bufferevent*, EventKey);
static void OnPrimchanError(struct bufferevent*, short what, EventKey);
static void OnDestResolutionIPv4(int result, char type, int count,
int ttl, void* addr_list, EventKey);
static void OnDestResolutionIPv6(int result, char type, int count,
int ttl, void* addr_list, EventKey);
static void OnDestchanRead(struct bufferevent*, EventKey);
static void OnDestchanWrite(struct bufferevent*, EventKey);
static void OnDestchanError(struct bufferevent*, short what, EventKey);
Chan& primchan() { return primchan_; }
EventKey evkey() const { return evkey_; }
private:
Serv* master_;
Phase phase_;
uint64 frame_bytes_remaining_;
uint8 frame_mask_[4];
int frame_mask_index_;
// We maintain two channels per Conn:
// primary channel is websocket connection.
Chan primchan_;
// Destination channel is a proxied connection.
Chan destchan_;
EventKey evkey_;
// Header fields supplied by client at initial websocket handshake.
std::map<std::string, std::string> header_fields_;
// Parameters requested via query component of GET resource.
std::map<std::string, std::string> requested_parameters_;
// Hostname and port of destination socket.
// Websocket client supplies them in first data frame (destframe).
std::string destname_;
int destport_;
// Preresolved |destname_| (empty if not pre-resolved).
std::string destaddr_;
// Whether TLS over TCP requested.
bool do_tls_;
// We try to DNS resolve hostname in both IPv4 and IPv6 domains.
// Track resolution failures here.
bool destresolution_ipv4_failed_;
bool destresolution_ipv6_failed_;
// Used to schedule a timeout for initial phase of connection.
scoped_ptr<struct event> destconnect_timeout_event_;
static base::LazyInstance<EventKeyMap>::Leaky evkey_map_;
static EventKey last_evkey_;
DISALLOW_COPY_AND_ASSIGN(Conn);
};
class SSLChan : public MessageLoopForIO::Watcher {
public:
static void Start(const net::AddressList& address_list,
const net::HostPortPair& host_port_pair,
int read_pipe,
int write_pipe) {
DCHECK(BrowserThread::CurrentlyOn(BrowserThread::IO));
SSLChan* ALLOW_UNUSED chan = new SSLChan(
address_list, host_port_pair, read_pipe, write_pipe);
}
private:
enum Phase {
PHASE_CONNECTING,
PHASE_RUNNING,
PHASE_CLOSING,
PHASE_CLOSED
};
class DerivedIOBufferWithSize : public net::IOBufferWithSize {
public:
DerivedIOBufferWithSize(net::IOBuffer* host, int size)
: IOBufferWithSize(host->data(), size), host_(host) {
DCHECK(host_);
DCHECK(host_->data());
}
protected:
virtual ~DerivedIOBufferWithSize() {
data_ = NULL; // We do not own memory, bypass base class destructor.
}
scoped_refptr<net::IOBuffer> host_;
};
// Provides queue of data represented as IOBuffers.
class IOBufferQueue {
public:
// We do not allocate all capacity at once but lazily in |buf_size_| chunks.
explicit IOBufferQueue(int capacity)
: buf_size_(1 + capacity / kNumBuffersLimit) {
}
// Obtains IOBuffer to add new data to back.
net::IOBufferWithSize* GetIOBufferToFill() {
if (back_ == NULL) {
if (storage_.size() >= kNumBuffersLimit)
return NULL;
storage_.push_back(new net::IOBufferWithSize(buf_size_));
back_ = new net::DrainableIOBuffer(storage_.back(), buf_size_);
}
return new DerivedIOBufferWithSize(
back_.get(), back_->BytesRemaining());
}
// Obtains IOBuffer with some data from front.
net::IOBufferWithSize* GetIOBufferToProcess() {
if (front_ == NULL) {
if (storage_.empty())
return NULL;
front_ = new net::DrainableIOBuffer(storage_.front(), buf_size_);
}
int front_capacity = (storage_.size() == 1 && back_) ?
back_->BytesConsumed() : buf_size_;
return new DerivedIOBufferWithSize(
front_.get(), front_capacity - front_->BytesConsumed());
}
// Records number of bytes as added to back.
void DidFill(int bytes) {
DCHECK(back_);
back_->DidConsume(bytes);
if (back_->BytesRemaining() == 0)
back_ = NULL;
}
// Pops number of bytes from front.
void DidProcess(int bytes) {
DCHECK(front_);
front_->DidConsume(bytes);
if (front_->BytesRemaining() == 0) {
storage_.pop_front();
front_ = NULL;
}
}
void Clear() {
front_ = NULL;
back_ = NULL;
storage_.clear();
}
private:
static const unsigned kNumBuffersLimit = 12;
const int buf_size_;
std::list< scoped_refptr<net::IOBufferWithSize> > storage_;
scoped_refptr<net::DrainableIOBuffer> front_;
scoped_refptr<net::DrainableIOBuffer> back_;
DISALLOW_COPY_AND_ASSIGN(IOBufferQueue);
};
SSLChan(const net::AddressList address_list,
const net::HostPortPair host_port_pair,
int read_pipe,
int write_pipe)
: phase_(PHASE_CONNECTING),
host_port_pair_(host_port_pair),
inbound_stream_(WebSocketProxy::kBufferLimit),
outbound_stream_(WebSocketProxy::kBufferLimit),
read_pipe_(read_pipe),
write_pipe_(write_pipe),
ALLOW_THIS_IN_INITIALIZER_LIST(weak_factory_(this)) {
if (!SetNonBlock(read_pipe_) || !SetNonBlock(write_pipe_)) {
Shut(net::ERR_UNEXPECTED);
return;
}
net::ClientSocketFactory* factory =
net::ClientSocketFactory::GetDefaultFactory();
socket_.reset(factory->CreateTransportClientSocket(
address_list, NULL, net::NetLog::Source()));
if (socket_ == NULL) {
Shut(net::ERR_FAILED);
return;
}
int result = socket_->Connect(base::Bind(&SSLChan::OnSocketConnect,
base::Unretained(this)));
if (result != net::ERR_IO_PENDING)
OnSocketConnect(result);
}
~SSLChan() {
phase_ = PHASE_CLOSED;
write_pipe_controller_.StopWatchingFileDescriptor();
read_pipe_controller_.StopWatchingFileDescriptor();
close(write_pipe_);
close(read_pipe_);
}
void Shut(int ALLOW_UNUSED net_error_code) {
if (phase_ != PHASE_CLOSED) {
phase_ = PHASE_CLOSING;
scoped_refptr<net::IOBufferWithSize> buf[] = {
outbound_stream_.GetIOBufferToProcess(),
inbound_stream_.GetIOBufferToProcess()
};
for (int i = arraysize(buf); i--;) {
if (buf[i] && buf[i]->size() > 0) {
MessageLoop::current()->PostTask(
FROM_HERE,
base::Bind(&SSLChan::Proceed, weak_factory_.GetWeakPtr()));
return;
}
}
phase_ = PHASE_CLOSED;
if (socket_ != NULL) {
socket_->Disconnect();
socket_.reset();
}
MessageLoop::current()->DeleteSoon(FROM_HERE, this);
}
}
void OnSocketConnect(int result) {
if (phase_ != PHASE_CONNECTING) {
NOTREACHED();
return;
}
if (result) {
Shut(result);
return;
}
net::ClientSocketHandle* handle = new net::ClientSocketHandle();
handle->set_socket(socket_.release());
net::ClientSocketFactory* factory =
net::ClientSocketFactory::GetDefaultFactory();
net::SSLClientSocketContext ssl_context;
if (!cert_verifier_.get())
cert_verifier_.reset(net::CertVerifier::CreateDefault());
ssl_context.cert_verifier = cert_verifier_.get();
socket_.reset(factory->CreateSSLClientSocket(
handle, host_port_pair_, ssl_config_, ssl_context));
if (!socket_.get()) {
LOG(WARNING) << "Failed to create an SSL client socket.";
OnSSLHandshakeCompleted(net::ERR_UNEXPECTED);
return;
}
result = socket_->Connect(base::Bind(&SSLChan::OnSSLHandshakeCompleted,
base::Unretained(this)));
if (result != net::ERR_IO_PENDING)
OnSSLHandshakeCompleted(result);
}
void OnSSLHandshakeCompleted(int result) {
if (result) {
Shut(result);
return;
}
is_socket_read_pending_ = false;
is_socket_write_pending_ = false;
is_read_pipe_blocked_ = false;
is_write_pipe_blocked_ = false;
MessageLoopForIO::current()->WatchFileDescriptor(
read_pipe_, false, MessageLoopForIO::WATCH_READ,
&read_pipe_controller_, this);
MessageLoopForIO::current()->WatchFileDescriptor(
write_pipe_, false, MessageLoopForIO::WATCH_WRITE,
&write_pipe_controller_, this);
phase_ = PHASE_RUNNING;
Proceed();
}
void OnSocketRead(int result) {
DCHECK(is_socket_read_pending_);
is_socket_read_pending_ = false;
if (result <= 0) {
Shut(result);
return;
}
inbound_stream_.DidFill(result);
Proceed();
}
void OnSocketWrite(int result) {
DCHECK(is_socket_write_pending_);
is_socket_write_pending_ = false;
if (result < 0) {
outbound_stream_.Clear();
Shut(result);
return;
}
outbound_stream_.DidProcess(result);
Proceed();
}
// MessageLoopForIO::Watcher overrides.
virtual void OnFileCanReadWithoutBlocking(int fd) OVERRIDE {
if (fd != read_pipe_) {
NOTREACHED();
return;
}
is_read_pipe_blocked_ = false;
Proceed();
}
virtual void OnFileCanWriteWithoutBlocking(int fd) OVERRIDE {
if (fd != write_pipe_) {
NOTREACHED();
return;
}
is_write_pipe_blocked_ = false;
Proceed();
}
private:
void Proceed() {
if (phase_ != PHASE_RUNNING && phase_ != PHASE_CLOSING)
return;
for (bool proceed = true; proceed;) {
proceed = false;
if (!is_read_pipe_blocked_ && phase_ == PHASE_RUNNING) {
scoped_refptr<net::IOBufferWithSize> buf =
outbound_stream_.GetIOBufferToFill();
if (buf && buf->size() > 0) {
int rv = read(read_pipe_, buf->data(), buf->size());
if (rv > 0) {
outbound_stream_.DidFill(rv);
proceed = true;
} else if (rv == -1 && errno == EAGAIN) {
is_read_pipe_blocked_ = true;
MessageLoopForIO::current()->WatchFileDescriptor(
read_pipe_, false, MessageLoopForIO::WATCH_READ,
&read_pipe_controller_, this);
} else if (rv == 0) {
Shut(0);
} else {
DCHECK_LT(rv, 0);
Shut(net::ERR_UNEXPECTED);
return;
}
}
}
if (!is_socket_read_pending_ && phase_ == PHASE_RUNNING) {
scoped_refptr<net::IOBufferWithSize> buf =
inbound_stream_.GetIOBufferToFill();
if (buf && buf->size() > 0) {
int rv = socket_->Read(
buf, buf->size(),
base::Bind(&SSLChan::OnSocketRead, base::Unretained(this)));
is_socket_read_pending_ = true;
if (rv != net::ERR_IO_PENDING) {
MessageLoop::current()->PostTask(
FROM_HERE, base::Bind(&SSLChan::OnSocketRead,
weak_factory_.GetWeakPtr(), rv));
}
}
}
if (!is_socket_write_pending_) {
scoped_refptr<net::IOBufferWithSize> buf =
outbound_stream_.GetIOBufferToProcess();
if (buf && buf->size() > 0) {
int rv = socket_->Write(
buf, buf->size(),
base::Bind(&SSLChan::OnSocketWrite, base::Unretained(this)));
is_socket_write_pending_ = true;
if (rv != net::ERR_IO_PENDING) {
MessageLoop::current()->PostTask(
FROM_HERE, base::Bind(&SSLChan::OnSocketWrite,
weak_factory_.GetWeakPtr(), rv));
}
} else if (phase_ == PHASE_CLOSING) {
Shut(0);
}
}
if (!is_write_pipe_blocked_) {
scoped_refptr<net::IOBufferWithSize> buf =
inbound_stream_.GetIOBufferToProcess();
if (buf && buf->size() > 0) {
int rv = write(write_pipe_, buf->data(), buf->size());
if (rv > 0) {
inbound_stream_.DidProcess(rv);
proceed = true;
} else if (rv == -1 && errno == EAGAIN) {
is_write_pipe_blocked_ = true;
MessageLoopForIO::current()->WatchFileDescriptor(
write_pipe_, false, MessageLoopForIO::WATCH_WRITE,
&write_pipe_controller_, this);
} else {
DCHECK_LE(rv, 0);
inbound_stream_.Clear();
Shut(net::ERR_UNEXPECTED);
return;
}
} else if (phase_ == PHASE_CLOSING) {
Shut(0);
}
}
}
}
Phase phase_;
scoped_ptr<net::StreamSocket> socket_;
net::HostPortPair host_port_pair_;
scoped_ptr<net::CertVerifier> cert_verifier_;
net::SSLConfig ssl_config_;
IOBufferQueue inbound_stream_;
IOBufferQueue outbound_stream_;
int read_pipe_;
int write_pipe_;
bool is_socket_read_pending_;
bool is_socket_write_pending_;
bool is_read_pipe_blocked_;
bool is_write_pipe_blocked_;
base::WeakPtrFactory<SSLChan> weak_factory_;
MessageLoopForIO::FileDescriptorWatcher read_pipe_controller_;
MessageLoopForIO::FileDescriptorWatcher write_pipe_controller_;
friend class base::DeleteHelper<SSLChan>;
DISALLOW_COPY_AND_ASSIGN(SSLChan);
};
Serv::Serv()
: evbase_(NULL),
listening_sock_(-1),
extra_listening_sock_(-1),
shutdown_requested_(false) {
control_descriptor_[0] = -1;
control_descriptor_[1] = -1;
}
Serv::~Serv() {
CloseAll();
}
void Serv::Run() {
if (evbase_ || shutdown_requested_)
return;
evbase_ = event_init();
if (!evbase_) {
LOG(ERROR) << "WebSocketProxy: Couldn't create libevent base";
return;
}
if (pipe(control_descriptor_) ||
!SetNonBlock(control_descriptor_[0]) ||
!SetNonBlock(control_descriptor_[1])) {
LOG(ERROR) << "WebSocketProxy: Failed to create control pipe";
return;
}
listening_sock_ = socket(AF_INET, SOCK_STREAM, 0);
if (listening_sock_ < 0) {
LOG(ERROR) << "WebSocketProxy: Failed to create socket";
return;
}
{
int on = 1;
setsockopt(listening_sock_, SOL_SOCKET, SO_REUSEADDR, &on, sizeof(on));
}
struct sockaddr_in addr;
memset(&addr, 0, sizeof(addr));
addr.sin_family = AF_INET;
// Let the OS allocate a port number.
addr.sin_port = base::HostToNet16(0);
addr.sin_addr.s_addr = base::HostToNet32(INADDR_LOOPBACK);
if (bind(listening_sock_,
reinterpret_cast<struct sockaddr*>(&addr),
sizeof(addr))) {
LOG(ERROR) << "WebSocketProxy: Failed to bind server socket";
return;
}
if (listen(listening_sock_, 12)) {
LOG(ERROR) << "WebSocketProxy: Failed to listen server socket";
return;
}
if (!SetNonBlock(listening_sock_)) {
LOG(ERROR) << "WebSocketProxy: Failed to go non block";
return;
}
connection_event_.reset(new struct event);
event_set(connection_event_.get(), listening_sock_, EV_READ | EV_PERSIST,
&OnConnect, this);
event_base_set(evbase_, connection_event_.get());
if (event_add(connection_event_.get(), NULL)) {
LOG(ERROR) << "WebSocketProxy: Failed to add listening event";
return;
}
{
// TODO(dilmah): remove this control block as soon as obsolete
// getPassportForTCP function is removed from webSocketProxyPrivate API.
// Following block adds extra listening socket on fixed port 10101.
extra_listening_sock_ = socket(AF_INET, SOCK_STREAM, 0);
if (extra_listening_sock_ < 0) {
LOG(ERROR) << "WebSocketProxy: Failed to create socket";
return;
}
{
int on = 1;
setsockopt(listening_sock_, SOL_SOCKET, SO_REUSEADDR, &on, sizeof(on));
}
const int kPort = 10101;
memset(&addr, 0, sizeof(addr));
addr.sin_family = AF_INET;
addr.sin_port = base::HostToNet16(kPort);
addr.sin_addr.s_addr = base::HostToNet32(INADDR_LOOPBACK);
if (bind(extra_listening_sock_,
reinterpret_cast<struct sockaddr*>(&addr),
sizeof(addr))) {
LOG(ERROR) << "WebSocketProxy: Failed to bind server socket";
return;
}
if (listen(extra_listening_sock_, 12)) {
LOG(ERROR) << "WebSocketProxy: Failed to listen server socket";
return;
}
if (!SetNonBlock(extra_listening_sock_)) {
LOG(ERROR) << "WebSocketProxy: Failed to go non block";
return;
}
extra_connection_event_.reset(new struct event);
event_set(extra_connection_event_.get(), extra_listening_sock_,
EV_READ | EV_PERSIST, &OnConnect, this);
event_base_set(evbase_, extra_connection_event_.get());
if (event_add(extra_connection_event_.get(), NULL)) {
LOG(ERROR) << "WebSocketProxy: Failed to add listening event";
return;
}
}
control_event_.reset(new struct event);
event_set(control_event_.get(), control_descriptor_[0], EV_READ | EV_PERSIST,
&OnControlRequest, this);
event_base_set(evbase_, control_event_.get());
if (event_add(control_event_.get(), NULL)) {
LOG(ERROR) << "WebSocketProxy: Failed to add control event";
return;
}
if (evdns_init())
LOG(WARNING) << "WebSocketProxy: Failed to initialize evDNS";
if (!IgnoreSigPipe()) {
LOG(ERROR) << "WebSocketProxy: Failed to ignore SIGPIPE";
return;
}
memset(&addr, 0, sizeof(addr));
socklen_t addr_len = sizeof(addr);
if (getsockname(
listening_sock_, reinterpret_cast<struct sockaddr*>(&addr), &addr_len)) {
LOG(ERROR) << "Failed to determine listening port";
return;
}
BrowserThread::PostTask(
BrowserThread::UI, FROM_HERE,
base::Bind(&SendNotification, base::NetToHost16(addr.sin_port)));
LOG(INFO) << "WebSocketProxy: Starting event dispatch loop.";
event_base_dispatch(evbase_);
if (shutdown_requested_)
LOG(INFO) << "WebSocketProxy: Event dispatch loop terminated upon request";
else
LOG(ERROR) << "WebSocketProxy: Event dispatch loop terminated unexpectedly";
CloseAll();
}
void Serv::Shutdown() {
int r ALLOW_UNUSED =
write(control_descriptor_[1], kControlMessageShutdown, 1);
}
void Serv::OnNetworkChange() {
int r ALLOW_UNUSED =
write(control_descriptor_[1], kControlMessageNetworkChange, 1);
}
void Serv::CloseAll() {
while (!conn_pool_.empty())
ZapConn(conn_pool_.back());
if (listening_sock_ >= 0) {
shutdown(listening_sock_, SHUT_RDWR);
close(listening_sock_);
}
for (int i = 0; i < 2; ++i) {
if (control_descriptor_[i] >= 0) {
control_descriptor_[i] = -1;
close(control_descriptor_[i]);
}
}
if (control_event_.get()) {
event_del(control_event_.get());
control_event_.reset();
}
if (extra_connection_event_.get()) {
event_del(extra_connection_event_.get());
extra_connection_event_.reset();
}
if (connection_event_.get()) {
event_del(connection_event_.get());
connection_event_.reset();
}
if (evbase_) {
event_base_free(evbase_);
evbase_ = NULL;
}
}
void Serv::ZapConn(Conn* cs) {
RevMap::iterator rit = rev_map_.find(cs);
if (rit != rev_map_.end()) {
conn_pool_.erase(rit->second);
rev_map_.erase(rit);
delete cs;
}
}
void Serv::MarkConnImportance(Conn* cs, bool important) {
if (conn_pool_.size() < WebSocketProxy::kConnPoolLimit / 4) {
// Fast common path.
return;
}
RevMap::iterator rit = rev_map_.find(cs);
if (rit != rev_map_.end()) {
ConnPool::iterator it = rit->second;
CHECK(*it == cs);
if (important && it == conn_pool_.begin()) {
// Already at the top. Shortcut.
return;
}
conn_pool_.erase(it);
}
if (important) {
conn_pool_.push_front(cs);
rev_map_[cs] = conn_pool_.begin();
} else {
conn_pool_.push_back(cs);
rev_map_[cs] = conn_pool_.end();
--rev_map_[cs];
}
}
Conn* Serv::GetFreshConn() {
if (conn_pool_.size() > WebSocketProxy::kConnPoolLimit) {
// Connections overflow. Shut those oldest not active.
ConnPool::iterator it = conn_pool_.end();
--it;
for (int i = conn_pool_.size() - WebSocketProxy::kConnPoolLimit; i-- > 0;) {
// Shut may invalidate an iterator; hence postdecrement.
(*it--)->Shut(WS_CLOSE_GOING_AWAY,
"Flood of new connections, getting rid of old ones");
}
if (conn_pool_.size() > WebSocketProxy::kConnPoolLimit + 12) {
// Connections overflow. Zap the oldest not active.
ZapConn(conn_pool_.back());
}
}
Conn* cs = new Conn(this);
conn_pool_.push_front(cs);
rev_map_[cs] = conn_pool_.begin();
return cs;
}
bool Serv::IsConnSane(Conn* cs) {
return rev_map_.find(cs) != rev_map_.end();
}
// static
void Serv::OnConnect(int listening_sock, short event, void* ctx) {
Serv* self = static_cast<Serv*>(ctx);
Conn* cs = self->GetFreshConn();
int sock = accept(listening_sock, NULL, NULL);
if (sock < 0 || !SetNonBlock(sock)) {
// Read readiness was triggered on listening socket
// yet we failed to accept a connection; definitely weird.
NOTREACHED();
self->ZapConn(cs);
return;
}
cs->primchan().set_read_fd(sock);
cs->primchan().set_write_fd(sock);
struct bufferevent* bev = bufferevent_new(
sock,
&Conn::OnPrimchanRead, &Conn::OnPrimchanWrite, &Conn::OnPrimchanError,
cs->evkey());
if (bev == NULL) {
self->ZapConn(cs);
return;
}
cs->primchan().set_read_bev(bev);
cs->primchan().set_write_bev(bev);
bufferevent_base_set(self->evbase_, bev);
bufferevent_setwatermark(bev, EV_READ, 0, WebSocketProxy::kBufferLimit);
if (bufferevent_enable(bev, EV_READ | EV_WRITE)) {
self->ZapConn(cs);
return;
}
}
// static
void Serv::OnControlRequest(int fd, short event, void* ctx) {
Serv* self = static_cast<Serv*>(ctx);
char c;
if (1 == read(fd, &c, 1) && c == *kControlMessageNetworkChange) {
// OnNetworkChange request.
evdns_clear_nameservers_and_suspend();
evdns_init();
evdns_resume();
} else if (c == *kControlMessageShutdown) {
self->shutdown_requested_ = true;
event_base_loopbreak(self->evbase_);
}
}
Conn::Conn(Serv* master)
: master_(master),
phase_(PHASE_WAIT_HANDSHAKE),
frame_bytes_remaining_(0),
frame_mask_index_(0),
primchan_(this),
destchan_(this),
do_tls_(false),
destresolution_ipv4_failed_(false),
destresolution_ipv6_failed_(false) {
while (evkey_map_.Get().find(last_evkey_) != evkey_map_.Get().end()) {
last_evkey_ = reinterpret_cast<EventKey>(reinterpret_cast<size_t>(
last_evkey_) + 1);
}
evkey_ = last_evkey_;
evkey_map_.Get()[evkey_] = this;
// Schedule timeout for initial phase of connection.
destconnect_timeout_event_.reset(new struct event);
evtimer_set(destconnect_timeout_event_.get(),
&OnDestConnectTimeout, evkey_);
event_base_set(master_->evbase(),
destconnect_timeout_event_.get());
struct timeval tv;
tv.tv_sec = 20;
tv.tv_usec = 0;
evtimer_add(destconnect_timeout_event_.get(), &tv);
}
Conn::~Conn() {
phase_ = PHASE_DEFUNCT;
event_del(destconnect_timeout_event_.get());
if (evkey_map_.Get()[evkey_] == this)
evkey_map_.Get().erase(evkey_);
else
NOTREACHED();
}
Conn* Conn::Get(EventKey evkey) {
EventKeyMap::iterator it = evkey_map_.Get().find(evkey);
if (it == evkey_map_.Get().end())
return NULL;
Conn* cs = it->second;
if (cs == NULL ||
cs->evkey_ != evkey ||
cs->master_ == NULL ||
cs->phase_ < 0 ||
cs->phase_ > PHASE_SHUT ||
!cs->master_->IsConnSane(cs)) {
return NULL;
}
return cs;
}
void Conn::Shut(int status, const void* reason) {
if (phase_ >= PHASE_SHUT)
return;
master_->MarkConnImportance(this, false);
std::vector<uint8> payload(2 + strlen(reason));
WriteNetworkInteger(status, vector_as_array(&payload), 2);
memcpy(vector_as_array(&payload) + 2, reason, strlen(reason));
EmitFrame(
&primchan_, WS_OPCODE_CLOSE, vector_as_array(&payload), payload.size());
primchan_.Shut();
destchan_.Shut();
phase_ = PHASE_SHUT;
}
void Conn::ConsiderSuicide() {
if (!primchan_.write_pending() && !destchan_.write_pending())
master_->ZapConn(this);
}
Conn::Status Conn::ConsumeHeader(struct evbuffer* evb) {
uint8* buf = EVBUFFER_DATA(evb);
size_t buf_size = EVBUFFER_LENGTH(evb);
static const uint8 kGetPrefix[] = "GET " kProxyPath;
static const uint8 kKeyValueDelimiter[] = ": ";
if (buf_size <= 0)
return STATUS_INCOMPLETE;
if (!buf)
return STATUS_ABORT;
if (!std::equal(buf, buf + std::min(buf_size, strlen(kGetPrefix)),
kGetPrefix)) {
// Data head does not match what is expected.
return STATUS_ABORT;
}
if (buf_size >= WebSocketProxy::kHeaderLimit)
return STATUS_ABORT;
uint8* buf_end = buf + buf_size;
// Handshake request must end with double CRLF.
uint8* term_pos = std::search(buf, buf_end, kCRLFCRLF,
kCRLFCRLF + strlen(kCRLFCRLF));
if (term_pos == buf_end)
return STATUS_INCOMPLETE;
term_pos += strlen(kCRLFCRLF);
// First line is "GET path?query protocol" line. If query is empty then we
// fall back to (obsolete) way of obtaining parameters from first websocket
// frame. Otherwise query contains all required parameters (host, port etc).
uint8* get_request_end = std::search(
buf, term_pos, kCRLF, kCRLF + strlen(kCRLF));
DCHECK(get_request_end != term_pos);
uint8* resource_end = std::find(
buf + strlen(kGetPrefix), get_request_end, ' ');
if (*resource_end != ' ')
return STATUS_ABORT;
if (resource_end != buf + strlen(kGetPrefix)) {
char* piece = reinterpret_cast<char*>(buf) + strlen(kGetPrefix) + 1;
url_parse::Component query(
0, resource_end - reinterpret_cast<uint8*>(piece));
for (url_parse::Component key, value;
url_parse::ExtractQueryKeyValue(piece, &query, &key, &value);) {
if (key.len > 0) {
requested_parameters_[std::string(piece + key.begin, key.len)] =
net::UnescapeURLComponent(std::string(piece + value.begin,
value.len), net::UnescapeRule::URL_SPECIAL_CHARS);
}
}
}
for (uint8* pos = get_request_end;;) {
pos += strlen(kCRLF);
if (term_pos - pos < static_cast<ptrdiff_t>(strlen(kCRLF)))
return STATUS_ABORT;
if (term_pos - pos == static_cast<ptrdiff_t>(strlen(kCRLF)))
break;
uint8* npos = std::search(pos, term_pos, kKeyValueDelimiter,
kKeyValueDelimiter + strlen(kKeyValueDelimiter));
if (npos == term_pos)
return STATUS_ABORT;
std::string key = FetchAsciiSnippet(pos, npos, AsciiFilterLower);
pos = std::search(npos += strlen(kKeyValueDelimiter), term_pos,
kCRLF, kCRLF + strlen(kCRLF));
if (pos == term_pos)
return STATUS_ABORT;
if (!key.empty()) {
header_fields_[key] = FetchAsciiSnippet(npos, pos,
key == "sec-websocket-key" ? AsciiFilterVerbatim : AsciiFilterLower);
}
}
// Values of Upgrade and Connection fields are hardcoded in the protocol.
if (header_fields_["upgrade"] != "websocket" ||
header_fields_["connection"] != "upgrade" ||
header_fields_["sec-websocket-key"].size() != 24) {
return STATUS_ABORT;
}
if (header_fields_["sec-websocket-version"] != "8" &&
header_fields_["sec-websocket-version"] != "13") {
return STATUS_ABORT;
}
// Normalize origin (e.g. leading slash).
GURL origin = GURL(GetOrigin()).GetOrigin();
if (!origin.is_valid())
return STATUS_ABORT;
if (!requested_parameters_.empty()) {
destname_ = requested_parameters_["hostname"];
int port;
if (!base::StringToInt(requested_parameters_["port"], &port) ||
port < 0 || port >= 1 << 16) {
return STATUS_ABORT;
}
destport_ = port;
destaddr_ = requested_parameters_["addr"];
do_tls_ = (requested_parameters_["tls"] == "true");
requested_parameters_["extension_id"] =
FetchExtensionIdFromOrigin(GetOrigin());
std::string passport(requested_parameters_["passport"]);
requested_parameters_.erase("passport");
if (!chrome::InternalAuthVerification::VerifyPassport(
passport, "web_socket_proxy", requested_parameters_)) {
return STATUS_ABORT;
}
}
evbuffer_drain(evb, term_pos - buf);
return STATUS_OK;
}
bool Conn::EmitHandshake(Chan* chan) {
std::vector<std::string> boilerplate;
boilerplate.push_back("HTTP/1.1 101 WebSocket Protocol Handshake");
boilerplate.push_back("Upgrade: websocket");
boilerplate.push_back("Connection: Upgrade");
{
// Take care of Accept field.
std::string word = header_fields_["sec-websocket-key"];
word += "258EAFA5-E914-47DA-95CA-C5AB0DC85B11";
std::string accept_token;
base::Base64Encode(base::SHA1HashString(word), &accept_token);
boilerplate.push_back("Sec-WebSocket-Accept: " + accept_token);
}
boilerplate.push_back("");
for (size_t i = 0; i < boilerplate.size(); ++i) {
if (!chan->Write(boilerplate[i].c_str(), boilerplate[i].size()) ||
!chan->Write(kCRLF, strlen(kCRLF))) {
return false;
}
}
return true;
}
bool Conn::EmitFrame(
Chan* chan, WebSocketFrameOpcode opcode, const void* buf, size_t size) {
uint8 header[10];
int header_size = 2;
DCHECK(chan);
DCHECK(opcode >= 0 && opcode < 16);
header[0] = opcode | 0x80; // FIN bit set.
if (size < 126) {
header[1] = size;
} else if (size < (1 << 16)) {
header[1] = 126;
WriteNetworkInteger(size, header + 2, 2);
header_size += 2;
} else {
header[1] = 127;
WriteNetworkInteger(size, header + 2, 8);
header_size += 8;
}
return chan->Write(header, header_size) && chan->Write(buf, size);
}
Conn::Status Conn::ConsumeDestframe(struct evbuffer* evb) {
if (!requested_parameters_.empty()) {
// Parameters were already provided (and verified) in query component of
// websocket URL.
return STATUS_OK;
}
if (frame_bytes_remaining_ == 0) {
Conn::Status rv = ConsumeFrameHeader(evb);
if (rv != STATUS_OK)
return rv;
if (frame_bytes_remaining_ == 0 ||
frame_bytes_remaining_ >= WebSocketProxy::kHeaderLimit) {
return STATUS_ABORT;
}
}
uint8* buf = EVBUFFER_DATA(evb);
size_t buf_size = EVBUFFER_LENGTH(evb);
if (buf_size < frame_bytes_remaining_)
return STATUS_INCOMPLETE;
for (size_t i = 0; i < buf_size; ++i) {
buf[i] ^= frame_mask_[frame_mask_index_];
frame_mask_index_ = (frame_mask_index_ + 1) % 4;
}
std::string passport;
if (!WebSocketProxyHelper::FetchPassportAddrNamePort(
buf, buf + frame_bytes_remaining_,
&passport, &destaddr_, &destname_, &destport_)) {
return STATUS_ABORT;
}
std::map<std::string, std::string> map;
map["hostname"] = destname_;
map["port"] = base::IntToString(destport_);
map["extension_id"] = FetchExtensionIdFromOrigin(GetOrigin());
if (!destaddr_.empty())
map["addr"] = destaddr_;
if (!chrome::InternalAuthVerification::VerifyPassport(
passport, "web_socket_proxy", map)) {
return STATUS_ABORT;
}
evbuffer_drain(evb, frame_bytes_remaining_);
frame_bytes_remaining_ = 0;
return STATUS_OK;
}
Conn::Status Conn::ConsumeFrameHeader(struct evbuffer* evb) {
uint8* buf = EVBUFFER_DATA(evb);
size_t buf_size = EVBUFFER_LENGTH(evb);
size_t header_size = 2;
if (buf_size < header_size)
return STATUS_INCOMPLETE;
if (buf[0] & 0x70) {
// We are not able to handle non-zero reserved bits.
NOTIMPLEMENTED();
return STATUS_ABORT;
}
bool fin_flag = buf[0] & 0x80;
if (!fin_flag) {
NOTIMPLEMENTED();
return STATUS_ABORT;
}
int opcode = buf[0] & 0x0f;
switch (opcode) {
case WS_OPCODE_TEXT:
break;
case WS_OPCODE_CLOSE:
return STATUS_ABORT;
default:
NOTIMPLEMENTED();
return STATUS_ABORT;
}
bool mask_flag = buf[1] & 0x80;
if (!mask_flag) {
// Client-to-server frames must be masked.
return STATUS_ABORT;
}
frame_bytes_remaining_ = buf[1] & 0x7f;
int extra_size = 0;
if (frame_bytes_remaining_ == 126)
extra_size = 2;
else if (frame_bytes_remaining_ == 127)
extra_size = 8;
if (buf_size < header_size + extra_size + sizeof(frame_mask_))
return STATUS_INCOMPLETE;
if (extra_size)
frame_bytes_remaining_ = ReadNetworkInteger(buf + header_size, extra_size);
header_size += extra_size;
memcpy(frame_mask_, buf + header_size, sizeof(frame_mask_));
header_size += sizeof(frame_mask_);
frame_mask_index_ = 0;
evbuffer_drain(evb, header_size);
return STATUS_OK;
}
Conn::Status Conn::ProcessFrameData(struct evbuffer* evb) {
uint8* buf = EVBUFFER_DATA(evb);
size_t buf_size = EVBUFFER_LENGTH(evb);
DCHECK_GE(frame_bytes_remaining_, 1u);
if (frame_bytes_remaining_ < buf_size)
buf_size = frame_bytes_remaining_;
// base64 is encoded in chunks of 4 bytes.
buf_size = buf_size / 4 * 4;
if (buf_size < 1)
return STATUS_INCOMPLETE;
switch (phase_) {
case PHASE_INSIDE_FRAME_BASE64: {
for (size_t i = 0; i < buf_size; ++i) {
buf[i] ^= frame_mask_[frame_mask_index_];
frame_mask_index_ = (frame_mask_index_ + 1) % 4;
}
std::string out_bytes;
base::Base64Decode(std::string(buf, buf + buf_size), &out_bytes);
evbuffer_drain(evb, buf_size);
DCHECK(destchan_.write_bev());
if (!destchan_.Write(out_bytes.c_str(), out_bytes.size()))
return STATUS_ABORT;
break;
}
case PHASE_INSIDE_FRAME_SKIP: {
evbuffer_drain(evb, buf_size);
break;
}
default: {
return STATUS_ABORT;
}
}
frame_bytes_remaining_ -= buf_size;
return frame_bytes_remaining_ ? STATUS_INCOMPLETE : STATUS_OK;
}
bool Conn::TryConnectDest(const struct sockaddr* addr, socklen_t addrlen) {
if (destchan_.read_fd() >= 0 || destchan_.read_bev() != NULL)
return false;
if (do_tls_) {
int fd[4];
if (pipe(fd) || pipe(fd + 2))
return false;
destchan_.set_read_fd(fd[0]);
destchan_.set_write_fd(fd[3]);
for (int i = arraysize(fd); i--;) {
if (!SetNonBlock(fd[i]))
return false;
}
destchan_.set_read_bev(bufferevent_new(
destchan_.read_fd(),
&OnDestchanRead, NULL, &OnDestchanError,
evkey_));
destchan_.set_write_bev(bufferevent_new(
destchan_.write_fd(),
NULL, &OnDestchanWrite, &OnDestchanError,
evkey_));
net::IPEndPoint endpoint;
if (!endpoint.FromSockAddr(addr, addrlen))
return false;
net::AddressList addrlist(endpoint);
net::HostPortPair host_port_pair(destname_, destport_);
BrowserThread::PostTask(
BrowserThread::IO, FROM_HERE, base::Bind(
&SSLChan::Start, addrlist, host_port_pair, fd[2], fd[1]));
} else {
int sock = socket(addr->sa_family, SOCK_STREAM, 0);
if (sock < 0)
return false;
destchan_.set_read_fd(sock);
destchan_.set_write_fd(sock);
if (!SetNonBlock(sock))
return false;
if (connect(sock, addr, addrlen)) {
if (errno != EINPROGRESS)
return false;
}
destchan_.set_read_bev(bufferevent_new(
sock,
&OnDestchanRead, &OnDestchanWrite, &OnDestchanError,
evkey_));
destchan_.set_write_bev(destchan_.read_bev());
}
if (destchan_.read_bev() == NULL || destchan_.write_bev() == NULL)
return false;
if (bufferevent_base_set(master_->evbase(), destchan_.read_bev()) ||
bufferevent_base_set(master_->evbase(), destchan_.write_bev())) {
return false;
}
bufferevent_setwatermark(
destchan_.read_bev(), EV_READ, 0, WebSocketProxy::kBufferLimit);
if (bufferevent_enable(destchan_.read_bev(), EV_READ) ||
bufferevent_enable(destchan_.write_bev(), EV_WRITE)) {
return false;
}
return true;
}
const std::string& Conn::GetOrigin() {
return header_fields_[header_fields_["sec-websocket-version"] == "8" ?
"sec-websocket-origin" : "origin"];
}
// static
void Conn::OnPrimchanRead(struct bufferevent* bev, EventKey evkey) {
Conn* cs = Conn::Get(evkey);
if (bev == NULL ||
cs == NULL ||
bev != cs->primchan_.read_bev()) {
NOTREACHED();
return;
}
if (EVBUFFER_LENGTH(EVBUFFER_INPUT(bev)) <= 0)
return;
cs->master_->MarkConnImportance(cs, true);
for (;;) {
switch (cs->phase_) {
case PHASE_WAIT_HANDSHAKE: {
switch (cs->ConsumeHeader(EVBUFFER_INPUT(bev))) {
case STATUS_OK: {
break;
}
case STATUS_INCOMPLETE: {
return;
}
case STATUS_ABORT:
default: {
cs->master_->ZapConn(cs);
return;
}
}
// Header consumed OK. Do respond.
if (!cs->EmitHandshake(&cs->primchan_)) {
cs->master_->ZapConn(cs);
return;
}
cs->phase_ = PHASE_WAIT_DESTFRAME;
}
case PHASE_WAIT_DESTFRAME: {
switch (cs->ConsumeDestframe(EVBUFFER_INPUT(bev))) {
case STATUS_OK: {
{
// Unfortunately libevent as of 1.4 does not look into /etc/hosts.
// There seems to be no easy API to perform only "local" part of
// getaddrinfo resolution. Hence this hack for "localhost".
if (cs->destname_ == "localhost")
cs->destname_ = "127.0.0.1";
}
if (cs->destaddr_.empty())
cs->destaddr_ = cs->destname_;
{
struct sockaddr_in sa;
memset(&sa, 0, sizeof(sa));
sa.sin_port = base::HostToNet16(cs->destport_);
if (inet_pton(sa.sin_family = AF_INET,
cs->destaddr_.c_str(),
&sa.sin_addr) == 1) {
// valid IPv4 address supplied.
if (cs->TryConnectDest((struct sockaddr*)&sa, sizeof(sa))) {
cs->phase_ = PHASE_WAIT_DESTCONNECT;
return;
}
}
}
{
if (cs->destaddr_.size() >= 2 &&
cs->destaddr_[0] == '[' &&
cs->destaddr_[cs->destaddr_.size() - 1] == ']') {
// Literal IPv6 address in brackets.
cs->destaddr_ =
cs->destaddr_.substr(1, cs->destaddr_.size() - 2);
}
struct sockaddr_in6 sa;
memset(&sa, 0, sizeof(sa));
sa.sin6_port = base::HostToNet16(cs->destport_);
if (inet_pton(sa.sin6_family = AF_INET6,
cs->destaddr_.c_str(),
&sa.sin6_addr) == 1) {
// valid IPv6 address supplied.
if (cs->TryConnectDest((struct sockaddr*)&sa, sizeof(sa))) {
cs->phase_ = PHASE_WAIT_DESTCONNECT;
return;
}
}
}
// Asynchronous DNS resolution.
if (evdns_count_nameservers() < 1) {
evdns_clear_nameservers_and_suspend();
evdns_init();
evdns_resume();
}
evdns_resolve_ipv4(cs->destname_.c_str(), 0,
&OnDestResolutionIPv4, evkey);
evdns_resolve_ipv6(cs->destname_.c_str(), 0,
&OnDestResolutionIPv6, evkey);
cs->phase_ = PHASE_WAIT_DESTCONNECT;
return;
}
case STATUS_INCOMPLETE: {
return;
}
case STATUS_ABORT:
default: {
cs->Shut(WS_CLOSE_DESTINATION_ERROR,
"Incorrect destination specification in first frame");
return;
}
}
}
case PHASE_WAIT_DESTCONNECT: {
if (EVBUFFER_LENGTH(EVBUFFER_INPUT(bev)) >=
WebSocketProxy::kBufferLimit) {
cs->Shut(WS_CLOSE_LIMIT_VIOLATION, "Read buffer overflow");
}
return;
}
case PHASE_OUTSIDE_FRAME: {
switch (cs->ConsumeFrameHeader(EVBUFFER_INPUT(bev))) {
case STATUS_OK: {
if (cs->frame_bytes_remaining_ % 4) {
// We expect base64 encoded data (encoded in 4-bytes chunks).
cs->Shut(WS_CLOSE_UNACCEPTABLE_DATA,
"Frame payload size is not multiple of 4");
return;
}
cs->phase_ = PHASE_INSIDE_FRAME_BASE64;
// Process remaining data if any.
break;
}
case STATUS_SKIP: {
cs->phase_ = PHASE_INSIDE_FRAME_SKIP;
// Process remaining data if any.
break;
}
case STATUS_INCOMPLETE: {
return;
}
case STATUS_ABORT:
default: {
cs->Shut(WS_CLOSE_PROTOCOL_ERROR, "Invalid frame header");
return;
}
}
break;
}
case PHASE_INSIDE_FRAME_BASE64:
case PHASE_INSIDE_FRAME_SKIP: {
switch (cs->ProcessFrameData(EVBUFFER_INPUT(bev))) {
case STATUS_OK: {
cs->phase_ = PHASE_OUTSIDE_FRAME;
// Handle remaining data if any.
break;
}
case STATUS_INCOMPLETE: {
return;
}
case STATUS_ABORT:
default: {
cs->Shut(WS_CLOSE_UNACCEPTABLE_DATA, "Invalid frame data");
return;
}
}
break;
}
case PHASE_SHUT: {
evbuffer_drain(EVBUFFER_INPUT(bev),
EVBUFFER_LENGTH(EVBUFFER_INPUT(bev)));
return;
}
case PHASE_DEFUNCT:
default: {
NOTREACHED();
cs->master_->ZapConn(cs);
return;
}
}
}
}
// static
void Conn::OnPrimchanWrite(struct bufferevent* bev, EventKey evkey) {
Conn* cs = Conn::Get(evkey);
if (bev == NULL ||
cs == NULL ||
bev != cs->primchan_.write_bev()) {
NOTREACHED();
return;
}
// Write callback is called when low watermark is reached, 0 by default.
cs->primchan_.set_write_pending(false);
if (cs->phase_ >= PHASE_SHUT) {
cs->master_->ZapConn(cs);
return;
}
if (cs->phase_ > PHASE_WAIT_DESTCONNECT)
OnDestchanRead(cs->destchan_.read_bev(), evkey);
if (cs->phase_ >= PHASE_SHUT)
cs->primchan_.Zap();
}
// static
void Conn::OnPrimchanError(struct bufferevent* bev,
short what, EventKey evkey) {
Conn* cs = Conn::Get(evkey);
if (bev == NULL ||
cs == NULL ||
(bev != cs->primchan_.read_bev() && bev != cs->primchan_.write_bev())) {
return;
}
cs->primchan_.set_write_pending(false);
if (cs->phase_ >= PHASE_SHUT)
cs->master_->ZapConn(cs);
else
cs->Shut(WS_CLOSE_NORMAL, "Error reported on websocket channel");
}
// static
void Conn::OnDestResolutionIPv4(int result, char type,
int count, int ttl,
void* addr_list, EventKey evkey) {
Conn* cs = Conn::Get(evkey);
if (cs == NULL)
return;
if (cs->phase_ != PHASE_WAIT_DESTCONNECT)
return;
if (result == DNS_ERR_NONE &&
count >= 1 &&
addr_list != NULL &&
type == DNS_IPv4_A) {
for (int i = 0; i < count; ++i) {
struct sockaddr_in sa;
memset(&sa, 0, sizeof(sa));
sa.sin_family = AF_INET;
sa.sin_port = base::HostToNet16(cs->destport_);
DCHECK(sizeof(sa.sin_addr) == sizeof(struct in_addr));
memcpy(&sa.sin_addr,
static_cast<struct in_addr*>(addr_list) + i,
sizeof(sa.sin_addr));
if (cs->TryConnectDest((struct sockaddr*)&sa, sizeof(sa)))
return;
}
}
cs->destresolution_ipv4_failed_ = true;
if (cs->destresolution_ipv4_failed_ && cs->destresolution_ipv6_failed_)
cs->Shut(WS_CLOSE_RESOLUTION_FAILED, "DNS resolution failed");
}
// static
void Conn::OnDestResolutionIPv6(int result, char type,
int count, int ttl,
void* addr_list, EventKey evkey) {
Conn* cs = Conn::Get(evkey);
if (cs == NULL)
return;
if (cs->phase_ != PHASE_WAIT_DESTCONNECT)
return;
if (result == DNS_ERR_NONE &&
count >= 1 &&
addr_list != NULL &&
type == DNS_IPv6_AAAA) {
for (int i = 0; i < count; ++i) {
struct sockaddr_in6 sa;
memset(&sa, 0, sizeof(sa));
sa.sin6_family = AF_INET6;
sa.sin6_port = base::HostToNet16(cs->destport_);
DCHECK(sizeof(sa.sin6_addr) == sizeof(struct in6_addr));
memcpy(&sa.sin6_addr,
static_cast<struct in6_addr*>(addr_list) + i,
sizeof(sa.sin6_addr));
if (cs->TryConnectDest((struct sockaddr*)&sa, sizeof(sa)))
return;
}
}
cs->destresolution_ipv6_failed_ = true;
if (cs->destresolution_ipv4_failed_ && cs->destresolution_ipv6_failed_)
cs->Shut(WS_CLOSE_RESOLUTION_FAILED, "DNS resolution failed");
}
// static
void Conn::OnDestConnectTimeout(int, short, EventKey evkey) {
Conn* cs = Conn::Get(evkey);
if (cs == NULL)
return;
if (cs->phase_ > PHASE_WAIT_DESTCONNECT)
return;
cs->Shut(WS_CLOSE_RESOLUTION_FAILED, "DNS resolution timeout");
}
// static
void Conn::OnDestchanRead(struct bufferevent* bev, EventKey evkey) {
Conn* cs = Conn::Get(evkey);
if (bev == NULL ||
cs == NULL ||
bev != cs->destchan_.read_bev()) {
NOTREACHED();
return;
}
if (EVBUFFER_LENGTH(EVBUFFER_INPUT(bev)) <= 0)
return;
if (cs->primchan_.write_bev() == NULL) {
cs->master_->ZapConn(cs);
return;
}
cs->master_->MarkConnImportance(cs, true);
std::string out_bytes;
base::Base64Encode(
std::string(
static_cast<const char*>(static_cast<void*>(
EVBUFFER_DATA(EVBUFFER_INPUT(bev)))),
EVBUFFER_LENGTH(EVBUFFER_INPUT(bev))),
&out_bytes);
evbuffer_drain(EVBUFFER_INPUT(bev), EVBUFFER_LENGTH(EVBUFFER_INPUT(bev)));
if (!cs->EmitFrame(&cs->primchan_, WS_OPCODE_TEXT,
out_bytes.c_str(), out_bytes.size())) {
cs->Shut(WS_CLOSE_UNEXPECTED, "Failed to write websocket frame");
}
}
// static
void Conn::OnDestchanWrite(struct bufferevent* bev, EventKey evkey) {
Conn* cs = Conn::Get(evkey);
if (bev == NULL ||
cs == NULL ||
bev != cs->destchan_.write_bev()) {
NOTREACHED();
return;
}
// Write callback is called when low watermark is reached, 0 by default.
cs->destchan_.set_write_pending(false);
if (cs->phase_ == PHASE_WAIT_DESTCONNECT)
cs->phase_ = PHASE_OUTSIDE_FRAME;
if (cs->phase_ < PHASE_SHUT)
OnPrimchanRead(cs->primchan_.read_bev(), evkey);
else
cs->destchan_.Zap();
}
// static
void Conn::OnDestchanError(struct bufferevent* bev,
short what, EventKey evkey) {
Conn* cs = Conn::Get(evkey);
if (bev == NULL ||
cs == NULL ||
(bev != cs->destchan_.read_bev() && bev != cs->destchan_.write_bev())) {
return;
}
cs->destchan_.set_write_pending(false);
if (cs->phase_ >= PHASE_SHUT)
cs->master_->ZapConn(cs);
else
cs->Shut(WS_CLOSE_DESTINATION_ERROR,
"Failure reported on destination channel");
}
// static
Conn::EventKey Conn::last_evkey_ = 0;
// static
base::LazyInstance<Conn::EventKeyMap>::Leaky
Conn::evkey_map_ = LAZY_INSTANCE_INITIALIZER;
} // namespace
WebSocketProxy::WebSocketProxy() : impl_(new Serv()) {
}
WebSocketProxy::~WebSocketProxy() {
delete static_cast<Serv*>(impl_);
impl_ = NULL;
}
void WebSocketProxy::Run() {
static_cast<Serv*>(impl_)->Run();
}
void WebSocketProxy::Shutdown() {
static_cast<Serv*>(impl_)->Shutdown();
}
void WebSocketProxy::OnNetworkChange() {
static_cast<Serv*>(impl_)->OnNetworkChange();
}
} // namespace chromeos