Source code

Revision control

Other Tools

/* -*- Mode: C++; tab-width: 8; indent-tabs-mode: nil; c-basic-offset: 2 -*- */
/* vim: set sw=2 ts=8 et tw=80 : */
/* This Source Code Form is subject to the terms of the Mozilla Public
* License, v. 2.0. If a copy of the MPL was not distributed with this
* file, You can obtain one at http://mozilla.org/MPL/2.0/. */
#include "WebSocketFrame.h"
#include "WebSocketLog.h"
#include "WebSocketChannel.h"
#include "mozilla/Atomics.h"
#include "mozilla/Attributes.h"
#include "mozilla/EndianUtils.h"
#include "mozilla/MathAlgorithms.h"
#include "mozilla/Utf8.h"
#include "mozilla/net/WebSocketEventService.h"
#include "nsIURI.h"
#include "nsIChannel.h"
#include "nsICryptoHash.h"
#include "nsIRunnable.h"
#include "nsIPrefBranch.h"
#include "nsICancelable.h"
#include "nsIClassOfService.h"
#include "nsIDNSRecord.h"
#include "nsIDNSService.h"
#include "nsIIOService.h"
#include "nsIProtocolProxyService.h"
#include "nsIProxyInfo.h"
#include "nsIProxiedChannel.h"
#include "nsIAsyncVerifyRedirectCallback.h"
#include "nsIDashboardEventNotifier.h"
#include "nsIEventTarget.h"
#include "nsIHttpChannel.h"
#include "nsIProtocolHandler.h"
#include "nsIRandomGenerator.h"
#include "nsISocketTransport.h"
#include "nsThreadUtils.h"
#include "nsINetworkLinkService.h"
#include "nsIObserverService.h"
#include "nsCharSeparatedTokenizer.h"
#include "nsNetCID.h"
#include "nsServiceManagerUtils.h"
#include "nsCRT.h"
#include "nsThreadUtils.h"
#include "nsError.h"
#include "mozilla/Base64.h"
#include "nsStringStream.h"
#include "nsAlgorithm.h"
#include "nsProxyRelease.h"
#include "nsNetUtil.h"
#include "nsINode.h"
#include "mozilla/StaticMutex.h"
#include "mozilla/Telemetry.h"
#include "mozilla/TimeStamp.h"
#include "nsSocketTransportService2.h"
#include "nsINSSErrorsService.h"
#include "nsWebSocketConnection.h"
#include "plbase64.h"
#include "prmem.h"
#include "prnetdb.h"
#include "zlib.h"
#include <algorithm>
// rather than slurp up all of nsIWebSocket.idl, which lives outside necko, just
// dupe one constant we need from it
#define CLOSE_GOING_AWAY 1001
using namespace mozilla;
using namespace mozilla::net;
namespace mozilla {
namespace net {
NS_IMPL_ISUPPORTS(WebSocketChannel, nsIWebSocketChannel, nsIHttpUpgradeListener,
nsIRequestObserver, nsIStreamListener, nsIProtocolHandler,
nsITimerCallback, nsIDNSListener, nsIProtocolProxyCallback,
nsIInterfaceRequestor, nsIChannelEventSink,
nsIThreadRetargetableRequest, nsIObserver, nsINamed,
nsIWebSocketConnectionListener)
// We implement RFC 6455, which uses Sec-WebSocket-Version: 13 on the wire.
#define SEC_WEBSOCKET_VERSION "13"
/*
* About SSL unsigned certificates
*
* wss will not work to a host using an unsigned certificate unless there
* is already an exception (i.e. it cannot popup a dialog asking for
* a security exception). This is similar to how an inlined img will
* fail without a dialog if fails for the same reason. This should not
* be a problem in practice as it is expected the websocket javascript
* is served from the same host as the websocket server (or of course,
* a valid cert could just be provided).
*
*/
// some helper classes
//-----------------------------------------------------------------------------
// FailDelayManager
//
// Stores entries (searchable by {host, port}) of connections that have recently
// failed, so we can do delay of reconnects per RFC 6455 Section 7.2.3
//-----------------------------------------------------------------------------
// Initial reconnect delay is randomly chosen between 200-400 ms.
// This is a gentler backoff than the 0-5 seconds the spec offhandedly suggests.
const uint32_t kWSReconnectInitialBaseDelay = 200;
const uint32_t kWSReconnectInitialRandomDelay = 200;
// Base lifetime (in ms) of a FailDelay: kept longer if more failures occur
const uint32_t kWSReconnectBaseLifeTime = 60 * 1000;
// Maximum reconnect delay (in ms)
const uint32_t kWSReconnectMaxDelay = 60 * 1000;
// hold record of failed connections, and calculates needed delay for reconnects
// to same host/port.
class FailDelay {
public:
FailDelay(nsCString address, int32_t port)
: mAddress(std::move(address)), mPort(port) {
mLastFailure = TimeStamp::Now();
mNextDelay = kWSReconnectInitialBaseDelay +
(rand() % kWSReconnectInitialRandomDelay);
}
// Called to update settings when connection fails again.
void FailedAgain() {
mLastFailure = TimeStamp::Now();
// We use a truncated exponential backoff as suggested by RFC 6455,
// but multiply by 1.5 instead of 2 to be more gradual.
mNextDelay = static_cast<uint32_t>(
std::min<double>(kWSReconnectMaxDelay, mNextDelay * 1.5));
LOG(
("WebSocket: FailedAgain: host=%s, port=%d: incremented delay to "
"%" PRIu32,
mAddress.get(), mPort, mNextDelay));
}
// returns 0 if there is no need to delay (i.e. delay interval is over)
uint32_t RemainingDelay(TimeStamp rightNow) {
TimeDuration dur = rightNow - mLastFailure;
uint32_t sinceFail = (uint32_t)dur.ToMilliseconds();
if (sinceFail > mNextDelay) return 0;
return mNextDelay - sinceFail;
}
bool IsExpired(TimeStamp rightNow) {
return (mLastFailure + TimeDuration::FromMilliseconds(
kWSReconnectBaseLifeTime + mNextDelay)) <=
rightNow;
}
nsCString mAddress; // IP address (or hostname if using proxy)
int32_t mPort;
private:
TimeStamp mLastFailure; // Time of last failed attempt
// mLastFailure + mNextDelay is the soonest we'll allow a reconnect
uint32_t mNextDelay; // milliseconds
};
class FailDelayManager {
public:
FailDelayManager() {
MOZ_COUNT_CTOR(FailDelayManager);
mDelaysDisabled = false;
nsCOMPtr<nsIPrefBranch> prefService =
do_GetService(NS_PREFSERVICE_CONTRACTID);
if (!prefService) {
return;
}
bool boolpref = true;
nsresult rv;
rv = prefService->GetBoolPref("network.websocket.delay-failed-reconnects",
&boolpref);
if (NS_SUCCEEDED(rv) && !boolpref) {
mDelaysDisabled = true;
}
}
~FailDelayManager() { MOZ_COUNT_DTOR(FailDelayManager); }
void Add(nsCString& address, int32_t port) {
if (mDelaysDisabled) return;
UniquePtr<FailDelay> record(new FailDelay(address, port));
mEntries.AppendElement(std::move(record));
}
// Element returned may not be valid after next main thread event: don't keep
// pointer to it around
FailDelay* Lookup(nsCString& address, int32_t port,
uint32_t* outIndex = nullptr) {
if (mDelaysDisabled) return nullptr;
FailDelay* result = nullptr;
TimeStamp rightNow = TimeStamp::Now();
// We also remove expired entries during search: iterate from end to make
// indexing simpler
for (int32_t i = mEntries.Length() - 1; i >= 0; --i) {
FailDelay* fail = mEntries[i].get();
if (fail->mAddress.Equals(address) && fail->mPort == port) {
if (outIndex) *outIndex = i;
result = fail;
// break here: removing more entries would mess up *outIndex.
// Any remaining expired entries will be deleted next time Lookup
// finds nothing, which is the most common case anyway.
break;
} else if (fail->IsExpired(rightNow)) {
mEntries.RemoveElementAt(i);
}
}
return result;
}
// returns true if channel connects immediately, or false if it's delayed
void DelayOrBegin(WebSocketChannel* ws) {
if (!mDelaysDisabled) {
uint32_t failIndex = 0;
FailDelay* fail = Lookup(ws->mAddress, ws->mPort, &failIndex);
if (fail) {
TimeStamp rightNow = TimeStamp::Now();
uint32_t remainingDelay = fail->RemainingDelay(rightNow);
if (remainingDelay) {
// reconnecting within delay interval: delay by remaining time
nsresult rv;
rv = NS_NewTimerWithCallback(getter_AddRefs(ws->mReconnectDelayTimer),
ws, remainingDelay,
nsITimer::TYPE_ONE_SHOT);
if (NS_SUCCEEDED(rv)) {
LOG(
("WebSocket: delaying websocket [this=%p] by %lu ms, changing"
" state to CONNECTING_DELAYED",
ws, (unsigned long)remainingDelay));
ws->mConnecting = CONNECTING_DELAYED;
return;
}
// if timer fails (which is very unlikely), drop down to BeginOpen
// call
} else if (fail->IsExpired(rightNow)) {
mEntries.RemoveElementAt(failIndex);
}
}
}
// Delays disabled, or no previous failure, or we're reconnecting after
// scheduled delay interval has passed: connect.
ws->BeginOpen(true);
}
// Remove() also deletes all expired entries as it iterates: better for
// battery life than using a periodic timer.
void Remove(nsCString& address, int32_t port) {
TimeStamp rightNow = TimeStamp::Now();
// iterate from end, to make deletion indexing easier
for (int32_t i = mEntries.Length() - 1; i >= 0; --i) {
FailDelay* entry = mEntries[i].get();
if ((entry->mAddress.Equals(address) && entry->mPort == port) ||
entry->IsExpired(rightNow)) {
mEntries.RemoveElementAt(i);
}
}
}
private:
nsTArray<UniquePtr<FailDelay>> mEntries;
bool mDelaysDisabled;
};
//-----------------------------------------------------------------------------
// nsWSAdmissionManager
//
// 1) Ensures that only one websocket at a time is CONNECTING to a given IP
// address (or hostname, if using proxy), per RFC 6455 Section 4.1.
// 2) Delays reconnects to IP/host after connection failure, per Section 7.2.3
//-----------------------------------------------------------------------------
class nsWSAdmissionManager {
public:
static void Init() {
StaticMutexAutoLock lock(sLock);
if (!sManager) {
sManager = new nsWSAdmissionManager();
}
}
static void Shutdown() {
StaticMutexAutoLock lock(sLock);
delete sManager;
sManager = nullptr;
}
// Determine if we will open connection immediately (returns true), or
// delay/queue the connection (returns false)
static void ConditionallyConnect(WebSocketChannel* ws) {
LOG(("Websocket: ConditionallyConnect: [this=%p]", ws));
MOZ_ASSERT(NS_IsMainThread(), "not main thread");
MOZ_ASSERT(ws->mConnecting == NOT_CONNECTING, "opening state");
StaticMutexAutoLock lock(sLock);
if (!sManager) {
return;
}
// If there is already another WS channel connecting to this IP address,
// defer BeginOpen and mark as waiting in queue.
bool found = (sManager->IndexOf(ws->mAddress) >= 0);
// Always add ourselves to queue, even if we'll connect immediately
UniquePtr<nsOpenConn> newdata(new nsOpenConn(ws->mAddress, ws));
sManager->mQueue.AppendElement(std::move(newdata));
if (found) {
LOG(
("Websocket: some other channel is connecting, changing state to "
"CONNECTING_QUEUED"));
ws->mConnecting = CONNECTING_QUEUED;
} else {
sManager->mFailures.DelayOrBegin(ws);
}
}
static void OnConnected(WebSocketChannel* aChannel) {
LOG(("Websocket: OnConnected: [this=%p]", aChannel));
MOZ_ASSERT(NS_IsMainThread(), "not main thread");
MOZ_ASSERT(aChannel->mConnecting == CONNECTING_IN_PROGRESS,
"Channel completed connect, but not connecting?");
StaticMutexAutoLock lock(sLock);
if (!sManager) {
return;
}
LOG(("Websocket: changing state to NOT_CONNECTING"));
aChannel->mConnecting = NOT_CONNECTING;
// Remove from queue
sManager->RemoveFromQueue(aChannel);
// Connection succeeded, so stop keeping track of any previous failures
sManager->mFailures.Remove(aChannel->mAddress, aChannel->mPort);
// Check for queued connections to same host.
// Note: still need to check for failures, since next websocket with same
// host may have different port
sManager->ConnectNext(aChannel->mAddress);
}
// Called every time a websocket channel ends its session (including going
// away w/o ever successfully creating a connection)
static void OnStopSession(WebSocketChannel* aChannel, nsresult aReason) {
LOG(("Websocket: OnStopSession: [this=%p, reason=0x%08" PRIx32 "]",
aChannel, static_cast<uint32_t>(aReason)));
StaticMutexAutoLock lock(sLock);
if (!sManager) {
return;
}
if (NS_FAILED(aReason)) {
// Have we seen this failure before?
FailDelay* knownFailure =
sManager->mFailures.Lookup(aChannel->mAddress, aChannel->mPort);
if (knownFailure) {
if (aReason == NS_ERROR_NOT_CONNECTED) {
// Don't count close() before connection as a network error
LOG(
("Websocket close() before connection to %s, %d completed"
" [this=%p]",
aChannel->mAddress.get(), (int)aChannel->mPort, aChannel));
} else {
// repeated failure to connect: increase delay for next connection
knownFailure->FailedAgain();
}
} else {
// new connection failure: record it.
LOG(("WebSocket: connection to %s, %d failed: [this=%p]",
aChannel->mAddress.get(), (int)aChannel->mPort, aChannel));
sManager->mFailures.Add(aChannel->mAddress, aChannel->mPort);
}
}
if (aChannel->mConnecting) {
MOZ_ASSERT(NS_IsMainThread(), "not main thread");
// Only way a connecting channel may get here w/o failing is if it was
// closed with GOING_AWAY (1001) because of navigation, tab close, etc.
MOZ_ASSERT(
NS_FAILED(aReason) || aChannel->mScriptCloseCode == CLOSE_GOING_AWAY,
"websocket closed while connecting w/o failing?");
sManager->RemoveFromQueue(aChannel);
bool wasNotQueued = (aChannel->mConnecting != CONNECTING_QUEUED);
LOG(("Websocket: changing state to NOT_CONNECTING"));
aChannel->mConnecting = NOT_CONNECTING;
if (wasNotQueued) {
sManager->ConnectNext(aChannel->mAddress);
}
}
}
static void IncrementSessionCount() {
StaticMutexAutoLock lock(sLock);
if (!sManager) {
return;
}
sManager->mSessionCount++;
}
static void DecrementSessionCount() {
StaticMutexAutoLock lock(sLock);
if (!sManager) {
return;
}
sManager->mSessionCount--;
}
static void GetSessionCount(int32_t& aSessionCount) {
StaticMutexAutoLock lock(sLock);
if (!sManager) {
return;
}
aSessionCount = sManager->mSessionCount;
}
private:
nsWSAdmissionManager() : mSessionCount(0) {
MOZ_COUNT_CTOR(nsWSAdmissionManager);
}
~nsWSAdmissionManager() { MOZ_COUNT_DTOR(nsWSAdmissionManager); }
class nsOpenConn {
public:
nsOpenConn(nsCString& addr, WebSocketChannel* channel)
: mAddress(addr), mChannel(channel) {
MOZ_COUNT_CTOR(nsOpenConn);
}
MOZ_COUNTED_DTOR(nsOpenConn)
nsCString mAddress;
WebSocketChannel* mChannel;
};
void ConnectNext(nsCString& hostName) {
MOZ_ASSERT(NS_IsMainThread(), "not main thread");
int32_t index = IndexOf(hostName);
if (index >= 0) {
WebSocketChannel* chan = mQueue[index]->mChannel;
MOZ_ASSERT(chan->mConnecting == CONNECTING_QUEUED,
"transaction not queued but in queue");
LOG(("WebSocket: ConnectNext: found channel [this=%p] in queue", chan));
mFailures.DelayOrBegin(chan);
}
}
void RemoveFromQueue(WebSocketChannel* aChannel) {
LOG(("Websocket: RemoveFromQueue: [this=%p]", aChannel));
int32_t index = IndexOf(aChannel);
MOZ_ASSERT(index >= 0, "connection to remove not in queue");
if (index >= 0) {
mQueue.RemoveElementAt(index);
}
}
int32_t IndexOf(nsCString& aStr) {
for (uint32_t i = 0; i < mQueue.Length(); i++)
if (aStr == (mQueue[i])->mAddress) return i;
return -1;
}
int32_t IndexOf(WebSocketChannel* aChannel) {
for (uint32_t i = 0; i < mQueue.Length(); i++)
if (aChannel == (mQueue[i])->mChannel) return i;
return -1;
}
// SessionCount might be decremented from the main or the socket
// thread, so manage it with atomic counters
Atomic<int32_t> mSessionCount;
// Queue for websockets that have not completed connecting yet.
// The first nsOpenConn with a given address will be either be
// CONNECTING_IN_PROGRESS or CONNECTING_DELAYED. Later ones with the same
// hostname must be CONNECTING_QUEUED.
//
// We could hash hostnames instead of using a single big vector here, but the
// dataset is expected to be small.
nsTArray<UniquePtr<nsOpenConn>> mQueue;
FailDelayManager mFailures;
static nsWSAdmissionManager* sManager;
static StaticMutex sLock;
};
nsWSAdmissionManager* nsWSAdmissionManager::sManager;
StaticMutex nsWSAdmissionManager::sLock;
//-----------------------------------------------------------------------------
// CallOnMessageAvailable
//-----------------------------------------------------------------------------
class CallOnMessageAvailable final : public Runnable {
public:
CallOnMessageAvailable(WebSocketChannel* aChannel, nsACString& aData,
int32_t aLen)
: Runnable("net::CallOnMessageAvailable"),
mChannel(aChannel),
mListenerMT(aChannel->mListenerMT),
mData(aData),
mLen(aLen) {}
NS_IMETHOD Run() override {
MOZ_ASSERT(mChannel->IsOnTargetThread());
if (mListenerMT) {
nsresult rv;
if (mLen < 0) {
rv = mListenerMT->mListener->OnMessageAvailable(mListenerMT->mContext,
mData);
} else {
rv = mListenerMT->mListener->OnBinaryMessageAvailable(
mListenerMT->mContext, mData);
}
if (NS_FAILED(rv)) {
LOG(
("OnMessageAvailable or OnBinaryMessageAvailable "
"failed with 0x%08" PRIx32,
static_cast<uint32_t>(rv)));
}
}
return NS_OK;
}
private:
~CallOnMessageAvailable() = default;
RefPtr<WebSocketChannel> mChannel;
RefPtr<BaseWebSocketChannel::ListenerAndContextContainer> mListenerMT;
nsCString mData;
int32_t mLen;
};
//-----------------------------------------------------------------------------
// CallOnStop
//-----------------------------------------------------------------------------
class CallOnStop final : public Runnable {
public:
CallOnStop(WebSocketChannel* aChannel, nsresult aReason)
: Runnable("net::CallOnStop"),
mChannel(aChannel),
mListenerMT(mChannel->mListenerMT),
mReason(aReason) {}
NS_IMETHOD Run() override {
MOZ_ASSERT(mChannel->IsOnTargetThread());
if (mListenerMT) {
nsresult rv =
mListenerMT->mListener->OnStop(mListenerMT->mContext, mReason);
if (NS_FAILED(rv)) {
LOG(
("WebSocketChannel::CallOnStop "
"OnStop failed (%08" PRIx32 ")\n",
static_cast<uint32_t>(rv)));
}
mChannel->mListenerMT = nullptr;
}
return NS_OK;
}
private:
~CallOnStop() = default;
RefPtr<WebSocketChannel> mChannel;
RefPtr<BaseWebSocketChannel::ListenerAndContextContainer> mListenerMT;
nsresult mReason;
};
//-----------------------------------------------------------------------------
// CallOnServerClose
//-----------------------------------------------------------------------------
class CallOnServerClose final : public Runnable {
public:
CallOnServerClose(WebSocketChannel* aChannel, uint16_t aCode,
nsACString& aReason)
: Runnable("net::CallOnServerClose"),
mChannel(aChannel),
mListenerMT(mChannel->mListenerMT),
mCode(aCode),
mReason(aReason) {}
NS_IMETHOD Run() override {
MOZ_ASSERT(mChannel->IsOnTargetThread());
if (mListenerMT) {
nsresult rv = mListenerMT->mListener->OnServerClose(mListenerMT->mContext,
mCode, mReason);
if (NS_FAILED(rv)) {
LOG(
("WebSocketChannel::CallOnServerClose "
"OnServerClose failed (%08" PRIx32 ")\n",
static_cast<uint32_t>(rv)));
}
}
return NS_OK;
}
private:
~CallOnServerClose() = default;
RefPtr<WebSocketChannel> mChannel;
RefPtr<BaseWebSocketChannel::ListenerAndContextContainer> mListenerMT;
uint16_t mCode;
nsCString mReason;
};
//-----------------------------------------------------------------------------
// CallAcknowledge
//-----------------------------------------------------------------------------
class CallAcknowledge final : public CancelableRunnable {
public:
CallAcknowledge(WebSocketChannel* aChannel, uint32_t aSize)
: CancelableRunnable("net::CallAcknowledge"),
mChannel(aChannel),
mListenerMT(mChannel->mListenerMT),
mSize(aSize) {}
NS_IMETHOD Run() override {
MOZ_ASSERT(mChannel->IsOnTargetThread());
LOG(("WebSocketChannel::CallAcknowledge: Size %u\n", mSize));
if (mListenerMT) {
nsresult rv =
mListenerMT->mListener->OnAcknowledge(mListenerMT->mContext, mSize);
if (NS_FAILED(rv)) {
LOG(("WebSocketChannel::CallAcknowledge: Acknowledge failed (%08" PRIx32
")\n",
static_cast<uint32_t>(rv)));
}
}
return NS_OK;
}
private:
~CallAcknowledge() = default;
RefPtr<WebSocketChannel> mChannel;
RefPtr<BaseWebSocketChannel::ListenerAndContextContainer> mListenerMT;
uint32_t mSize;
};
//-----------------------------------------------------------------------------
// CallOnTransportAvailable
//-----------------------------------------------------------------------------
class CallOnTransportAvailable final : public Runnable {
public:
CallOnTransportAvailable(WebSocketChannel* aChannel,
nsISocketTransport* aTransport,
nsIAsyncInputStream* aSocketIn,
nsIAsyncOutputStream* aSocketOut)
: Runnable("net::CallOnTransportAvailble"),
mChannel(aChannel),
mTransport(aTransport),
mSocketIn(aSocketIn),
mSocketOut(aSocketOut) {}
NS_IMETHOD Run() override {
LOG(("WebSocketChannel::CallOnTransportAvailable %p\n", this));
return mChannel->OnTransportAvailable(mTransport, mSocketIn, mSocketOut);
}
private:
~CallOnTransportAvailable() = default;
RefPtr<WebSocketChannel> mChannel;
nsCOMPtr<nsISocketTransport> mTransport;
nsCOMPtr<nsIAsyncInputStream> mSocketIn;
nsCOMPtr<nsIAsyncOutputStream> mSocketOut;
};
//-----------------------------------------------------------------------------
// PMCECompression
//-----------------------------------------------------------------------------
class PMCECompression {
public:
PMCECompression(bool aNoContextTakeover, int32_t aLocalMaxWindowBits,
int32_t aRemoteMaxWindowBits)
: mActive(false),
mNoContextTakeover(aNoContextTakeover),
mResetDeflater(false),
mMessageDeflated(false) {
this->mDeflater.next_in = nullptr;
this->mDeflater.avail_in = 0;
this->mDeflater.total_in = 0;
this->mDeflater.next_out = nullptr;
this->mDeflater.avail_out = 0;
this->mDeflater.total_out = 0;
this->mDeflater.msg = nullptr;
this->mDeflater.state = nullptr;
this->mDeflater.data_type = 0;
this->mDeflater.adler = 0;
this->mDeflater.reserved = 0;
this->mInflater.next_in = nullptr;
this->mInflater.avail_in = 0;
this->mInflater.total_in = 0;
this->mInflater.next_out = nullptr;
this->mInflater.avail_out = 0;
this->mInflater.total_out = 0;
this->mInflater.msg = nullptr;
this->mInflater.state = nullptr;
this->mInflater.data_type = 0;
this->mInflater.adler = 0;
this->mInflater.reserved = 0;
MOZ_COUNT_CTOR(PMCECompression);
mDeflater.zalloc = mInflater.zalloc = Z_NULL;
mDeflater.zfree = mInflater.zfree = Z_NULL;
mDeflater.opaque = mInflater.opaque = Z_NULL;
if (deflateInit2(&mDeflater, Z_DEFAULT_COMPRESSION, Z_DEFLATED,
-aLocalMaxWindowBits, 8, Z_DEFAULT_STRATEGY) == Z_OK) {
if (inflateInit2(&mInflater, -aRemoteMaxWindowBits) == Z_OK) {
mActive = true;
} else {
deflateEnd(&mDeflater);
}
}
}
~PMCECompression() {
MOZ_COUNT_DTOR(PMCECompression);
if (mActive) {
inflateEnd(&mInflater);
deflateEnd(&mDeflater);
}
}
bool Active() { return mActive; }
void SetMessageDeflated() {
MOZ_ASSERT(!mMessageDeflated);
mMessageDeflated = true;
}
bool IsMessageDeflated() { return mMessageDeflated; }
bool UsingContextTakeover() { return !mNoContextTakeover; }
nsresult Deflate(uint8_t* data, uint32_t dataLen, nsACString& _retval) {
if (mResetDeflater || mNoContextTakeover) {
if (deflateReset(&mDeflater) != Z_OK) {
return NS_ERROR_UNEXPECTED;
}
mResetDeflater = false;
}
mDeflater.avail_out = kBufferLen;
mDeflater.next_out = mBuffer;
mDeflater.avail_in = dataLen;
mDeflater.next_in = data;
while (true) {
int zerr = deflate(&mDeflater, Z_SYNC_FLUSH);
if (zerr != Z_OK) {
mResetDeflater = true;
return NS_ERROR_UNEXPECTED;
}
uint32_t deflated = kBufferLen - mDeflater.avail_out;
if (deflated > 0) {
_retval.Append(reinterpret_cast<char*>(mBuffer), deflated);
}
mDeflater.avail_out = kBufferLen;
mDeflater.next_out = mBuffer;
if (mDeflater.avail_in > 0) {
continue; // There is still some data to deflate
}
if (deflated == kBufferLen) {
continue; // There was not enough space in the buffer
}
break;
}
if (_retval.Length() < 4) {
MOZ_ASSERT(false, "Expected trailing not found in deflated data!");
mResetDeflater = true;
return NS_ERROR_UNEXPECTED;
}
_retval.Truncate(_retval.Length() - 4);
return NS_OK;
}
nsresult Inflate(uint8_t* data, uint32_t dataLen, nsACString& _retval) {
mMessageDeflated = false;
Bytef trailingData[] = {0x00, 0x00, 0xFF, 0xFF};
bool trailingDataUsed = false;
mInflater.avail_out = kBufferLen;
mInflater.next_out = mBuffer;
mInflater.avail_in = dataLen;
mInflater.next_in = data;
while (true) {
int zerr = inflate(&mInflater, Z_NO_FLUSH);
if (zerr == Z_STREAM_END) {
Bytef* saveNextIn = mInflater.next_in;
uint32_t saveAvailIn = mInflater.avail_in;
Bytef* saveNextOut = mInflater.next_out;
uint32_t saveAvailOut = mInflater.avail_out;
inflateReset(&mInflater);
mInflater.next_in = saveNextIn;
mInflater.avail_in = saveAvailIn;
mInflater.next_out = saveNextOut;
mInflater.avail_out = saveAvailOut;
} else if (zerr != Z_OK && zerr != Z_BUF_ERROR) {
return NS_ERROR_INVALID_CONTENT_ENCODING;
}
uint32_t inflated = kBufferLen - mInflater.avail_out;
if (inflated > 0) {
_retval.Append(reinterpret_cast<char*>(mBuffer), inflated);
}
mInflater.avail_out = kBufferLen;
mInflater.next_out = mBuffer;
if (mInflater.avail_in > 0) {
continue; // There is still some data to inflate
}
if (inflated == kBufferLen) {
continue; // There was not enough space in the buffer
}
if (!trailingDataUsed) {
trailingDataUsed = true;
mInflater.avail_in = sizeof(trailingData);
mInflater.next_in = trailingData;
continue;
}
return NS_OK;
}
}
private:
bool mActive;
bool mNoContextTakeover;
bool mResetDeflater;
bool mMessageDeflated;
z_stream mDeflater;
z_stream mInflater;
const static uint32_t kBufferLen = 4096;
uint8_t mBuffer[kBufferLen];
};
//-----------------------------------------------------------------------------
// OutboundMessage
//-----------------------------------------------------------------------------
enum WsMsgType {
kMsgTypeString = 0,
kMsgTypeBinaryString,
kMsgTypeStream,
kMsgTypePing,
kMsgTypePong,
kMsgTypeFin
};
static const char* msgNames[] = {"text", "binaryString", "binaryStream",
"ping", "pong", "close"};
class OutboundMessage {
public:
OutboundMessage(WsMsgType type, const nsACString& str)
: mMsg(mozilla::AsVariant(pString(str))),
mMsgType(type),
mDeflated(false) {
MOZ_COUNT_CTOR(OutboundMessage);
}
OutboundMessage(nsIInputStream* stream, uint32_t length)
: mMsg(mozilla::AsVariant(StreamWithLength(stream, length))),
mMsgType(kMsgTypeStream),
mDeflated(false) {
MOZ_COUNT_CTOR(OutboundMessage);
}
~OutboundMessage() {
MOZ_COUNT_DTOR(OutboundMessage);
switch (mMsgType) {
case kMsgTypeString:
case kMsgTypeBinaryString:
case kMsgTypePing:
case kMsgTypePong:
break;
case kMsgTypeStream:
// for now this only gets hit if msg deleted w/o being sent
if (mMsg.as<StreamWithLength>().mStream) {
mMsg.as<StreamWithLength>().mStream->Close();
}
break;
case kMsgTypeFin:
break; // do-nothing: avoid compiler warning
}
}
WsMsgType GetMsgType() const { return mMsgType; }
int32_t Length() {
if (mMsg.is<pString>()) {
return mMsg.as<pString>().mValue.Length();
}
return mMsg.as<StreamWithLength>().mLength;
}
int32_t OrigLength() {
if (mMsg.is<pString>()) {
pString& ref = mMsg.as<pString>();
return mDeflated ? ref.mOrigValue.Length() : ref.mValue.Length();
}
return mMsg.as<StreamWithLength>().mLength;
}
uint8_t* BeginWriting() {
MOZ_ASSERT(mMsgType != kMsgTypeStream,
"Stream should have been converted to string by now");
if (!mMsg.as<pString>().mValue.IsVoid()) {
return (uint8_t*)mMsg.as<pString>().mValue.BeginWriting();
}
return nullptr;
}
uint8_t* BeginReading() {
MOZ_ASSERT(mMsgType != kMsgTypeStream,
"Stream should have been converted to string by now");
if (!mMsg.as<pString>().mValue.IsVoid()) {
return (uint8_t*)mMsg.as<pString>().mValue.BeginReading();
}
return nullptr;
}
uint8_t* BeginOrigReading() {
MOZ_ASSERT(mMsgType != kMsgTypeStream,
"Stream should have been converted to string by now");
if (!mDeflated) return BeginReading();
if (!mMsg.as<pString>().mOrigValue.IsVoid()) {
return (uint8_t*)mMsg.as<pString>().mOrigValue.BeginReading();
}
return nullptr;
}
nsresult ConvertStreamToString() {
MOZ_ASSERT(mMsgType == kMsgTypeStream, "Not a stream!");
nsAutoCString temp;
{
StreamWithLength& ref = mMsg.as<StreamWithLength>();
nsresult rv = NS_ReadInputStreamToString(ref.mStream, temp, ref.mLength);
NS_ENSURE_SUCCESS(rv, rv);
if (temp.Length() != ref.mLength) {
return NS_ERROR_UNEXPECTED;
}
ref.mStream->Close();
}
mMsg = mozilla::AsVariant(pString(temp));
mMsgType = kMsgTypeBinaryString;
return NS_OK;
}
bool DeflatePayload(PMCECompression* aCompressor) {
MOZ_ASSERT(mMsgType != kMsgTypeStream,
"Stream should have been converted to string by now");
MOZ_ASSERT(!mDeflated);
nsresult rv;
pString& ref = mMsg.as<pString>();
if (ref.mValue.Length() == 0) {
// Empty message
return false;
}
nsAutoCString temp;
rv = aCompressor->Deflate(BeginReading(), ref.mValue.Length(), temp);
if (NS_FAILED(rv)) {
LOG(
("WebSocketChannel::OutboundMessage: Deflating payload failed "
"[rv=0x%08" PRIx32 "]\n",
static_cast<uint32_t>(rv)));
return false;
}
if (!aCompressor->UsingContextTakeover() &&
temp.Length() > ref.mValue.Length()) {
// When "<local>_no_context_takeover" was negotiated, do not send deflated
// payload if it's larger that the original one. OTOH, it makes sense
// to send the larger deflated payload when the sliding window is not
// reset between messages because if we would skip some deflated block
// we would need to empty the sliding window which could affect the
// compression of the subsequent messages.
LOG(
("WebSocketChannel::OutboundMessage: Not deflating message since the "
"deflated payload is larger than the original one [deflated=%d, "
"original=%d]",
temp.Length(), ref.mValue.Length()));
return false;
}
mDeflated = true;
mMsg.as<pString>().mOrigValue = mMsg.as<pString>().mValue;
mMsg.as<pString>().mValue = temp;
return true;
}
private:
struct pString {
nsCString mValue;
nsCString mOrigValue;
explicit pString(const nsACString& value)
: mValue(value), mOrigValue(VoidCString()) {}
};
struct StreamWithLength {
nsCOMPtr<nsIInputStream> mStream;
uint32_t mLength;
explicit StreamWithLength(nsIInputStream* stream, uint32_t Length)
: mStream(stream), mLength(Length) {}
};
mozilla::Variant<pString, StreamWithLength> mMsg;
WsMsgType mMsgType;
bool mDeflated;
};
//-----------------------------------------------------------------------------
// OutboundEnqueuer
//-----------------------------------------------------------------------------
class OutboundEnqueuer final : public Runnable {
public:
OutboundEnqueuer(WebSocketChannel* aChannel, OutboundMessage* aMsg)
: Runnable("OutboundEnquerer"), mChannel(aChannel), mMessage(aMsg) {}
NS_IMETHOD Run() override {
mChannel->EnqueueOutgoingMessage(mChannel->mOutgoingMessages, mMessage);
return NS_OK;
}
private:
~OutboundEnqueuer() = default;
RefPtr<WebSocketChannel> mChannel;
OutboundMessage* mMessage;
};
//-----------------------------------------------------------------------------
// WebSocketChannel
//-----------------------------------------------------------------------------
WebSocketChannel::WebSocketChannel()
: mPort(0),
mCloseTimeout(20000),
mOpenTimeout(20000),
mConnecting(NOT_CONNECTING),
mMaxConcurrentConnections(200),
mInnerWindowID(0),
mGotUpgradeOK(0),
mRecvdHttpUpgradeTransport(0),
mAutoFollowRedirects(0),
mAllowPMCE(1),
mPingOutstanding(0),
mReleaseOnTransmit(0),
mDataStarted(false),
mRequestedClose(false),
mClientClosed(false),
mServerClosed(false),
mStopped(false),
mCalledOnStop(false),
mTCPClosed(false),
mOpenedHttpChannel(false),
mIncrementedSessionCount(false),
mDecrementedSessionCount(false),
mMaxMessageSize(INT32_MAX),
mStopOnClose(NS_OK),
mServerCloseCode(CLOSE_ABNORMAL),
mScriptCloseCode(0),
mFragmentOpcode(nsIWebSocketFrame::OPCODE_CONTINUATION),
mFragmentAccumulator(0),
mBuffered(0),
mBufferSize(kIncomingBufferInitialSize),
mCurrentOut(nullptr),
mHdrOutSize(0),
mHdrOut(nullptr),
mDynamicOutputSize(0),
mDynamicOutput(nullptr),
mPrivateBrowsing(false),
mConnectionLogService(nullptr),
mMutex("WebSocketChannel::mMutex") {
MOZ_ASSERT(NS_IsMainThread(), "not main thread");
LOG(("WebSocketChannel::WebSocketChannel() %p\n", this));
nsWSAdmissionManager::Init();
mFramePtr = mBuffer = static_cast<uint8_t*>(moz_xmalloc(mBufferSize));
nsresult rv;
mConnectionLogService =
do_GetService("@mozilla.org/network/dashboard;1", &rv);
if (NS_FAILED(rv)) LOG(("Failed to initiate dashboard service."));
mService = WebSocketEventService::GetOrCreate();
}
WebSocketChannel::~WebSocketChannel() {
LOG(("WebSocketChannel::~WebSocketChannel() %p\n", this));
if (mWasOpened) {
MOZ_ASSERT(mCalledOnStop, "WebSocket was opened but OnStop was not called");
MOZ_ASSERT(mStopped, "WebSocket was opened but never stopped");
}
MOZ_ASSERT(!mCancelable, "DNS/Proxy Request still alive at destruction");
MOZ_ASSERT(!mConnecting, "Should not be connecting in destructor");
free(mBuffer);
free(mDynamicOutput);
delete mCurrentOut;
while ((mCurrentOut = mOutgoingPingMessages.PopFront())) {
delete mCurrentOut;
}
while ((mCurrentOut = mOutgoingPongMessages.PopFront())) {
delete mCurrentOut;
}
while ((mCurrentOut = mOutgoingMessages.PopFront())) {
delete mCurrentOut;
}
mListenerMT = nullptr;
NS_ReleaseOnMainThread("WebSocketChannel::mLoadGroup", mLoadGroup.forget());
NS_ReleaseOnMainThread("WebSocketChannel::mLoadInfo", mLoadInfo.forget());
NS_ReleaseOnMainThread("WebSocketChannel::mService", mService.forget());
}
NS_IMETHODIMP
WebSocketChannel::Observe(nsISupports* subject, const char* topic,
const char16_t* data) {
LOG(("WebSocketChannel::Observe [topic=\"%s\"]\n", topic));
if (strcmp(topic, NS_NETWORK_LINK_TOPIC) == 0) {
nsCString converted = NS_ConvertUTF16toUTF8(data);
const char* state = converted.get();
if (strcmp(state, NS_NETWORK_LINK_DATA_CHANGED) == 0) {
LOG(("WebSocket: received network CHANGED event"));
if (!mSocketThread) {
// there has not been an asyncopen yet on the object and then we need
// no ping.
LOG(("WebSocket: early object, no ping needed"));
} else {
// Next we check mDataStarted, which we need to do on mTargetThread.
if (!IsOnTargetThread()) {
mTargetThread->Dispatch(
NewRunnableMethod("net::WebSocketChannel::OnNetworkChanged", this,
&WebSocketChannel::OnNetworkChanged),
NS_DISPATCH_NORMAL);
} else {
nsresult rv = OnNetworkChanged();
if (NS_FAILED(rv)) {
LOG(("WebSocket: OnNetworkChanged failed (%08" PRIx32 ")",
static_cast<uint32_t>(rv)));
}
}
}
}
}
return NS_OK;
}
nsresult WebSocketChannel::OnNetworkChanged() {
if (IsOnTargetThread()) {
LOG(("WebSocketChannel::OnNetworkChanged() - on target thread %p", this));
if (!mDataStarted) {
LOG(("WebSocket: data not started yet, no ping needed"));
return NS_OK;
}
return mSocketThread->Dispatch(
NewRunnableMethod("net::WebSocketChannel::OnNetworkChanged", this,
&WebSocketChannel::OnNetworkChanged),
NS_DISPATCH_NORMAL);
}
MOZ_ASSERT(OnSocketThread(), "not on socket thread");
LOG(("WebSocketChannel::OnNetworkChanged() - on socket thread %p", this));
if (mPingOutstanding) {
// If there's an outstanding ping that's expected to get a pong back
// we let that do its thing.
LOG(("WebSocket: pong already pending"));
return NS_OK;
}
if (mPingForced) {
// avoid more than one
LOG(("WebSocket: forced ping timer already fired"));
return NS_OK;
}
LOG(("nsWebSocketChannel:: Generating Ping as network changed\n"));
if (!mPingTimer) {
// The ping timer is only conditionally running already. If it wasn't
// already created do it here.
mPingTimer = NS_NewTimer();
if (!mPingTimer) {
LOG(("WebSocket: unable to create ping timer!"));
NS_WARNING("unable to create ping timer!");
return NS_ERROR_OUT_OF_MEMORY;
}
}
// Trigger the ping timeout asap to fire off a new ping. Wait just
// a little bit to better avoid multi-triggers.
mPingForced = true;
mPingTimer->InitWithCallback(this, 200, nsITimer::TYPE_ONE_SHOT);
return NS_OK;
}
void WebSocketChannel::Shutdown() { nsWSAdmissionManager::Shutdown(); }
bool WebSocketChannel::IsOnTargetThread() {
MOZ_ASSERT(mTargetThread);
bool isOnTargetThread = false;
nsresult rv = mTargetThread->IsOnCurrentThread(&isOnTargetThread);
MOZ_ASSERT(NS_SUCCEEDED(rv));
return NS_FAILED(rv) ? false : isOnTargetThread;
}
void WebSocketChannel::GetEffectiveURL(nsAString& aEffectiveURL) const {
aEffectiveURL = mEffectiveURL;
}
bool WebSocketChannel::IsEncrypted() const { return mEncrypted; }
void WebSocketChannel::BeginOpen(bool aCalledFromAdmissionManager) {
MOZ_ASSERT(NS_IsMainThread(), "not main thread");
LOG(("WebSocketChannel::BeginOpen() %p\n", this));
// Important that we set CONNECTING_IN_PROGRESS before any call to
// AbortSession here: ensures that any remaining queued connection(s) are
// scheduled in OnStopSession
LOG(("Websocket: changing state to CONNECTING_IN_PROGRESS"));
mConnecting = CONNECTING_IN_PROGRESS;
if (aCalledFromAdmissionManager) {
// When called from nsWSAdmissionManager post an event to avoid potential
// re-entering of nsWSAdmissionManager and its lock.
NS_DispatchToMainThread(
NewRunnableMethod("net::WebSocketChannel::BeginOpenInternal", this,
&WebSocketChannel::BeginOpenInternal),
NS_DISPATCH_NORMAL);
} else {
BeginOpenInternal();
}
}
void WebSocketChannel::BeginOpenInternal() {
LOG(("WebSocketChannel::BeginOpenInternal() %p\n", this));
nsresult rv;
if (mRedirectCallback) {
LOG(("WebSocketChannel::BeginOpenInternal: Resuming Redirect\n"));
rv = mRedirectCallback->OnRedirectVerifyCallback(NS_OK);
mRedirectCallback = nullptr;
return;
}
nsCOMPtr<nsIChannel> localChannel = do_QueryInterface(mChannel, &rv);
if (NS_FAILED(rv)) {
LOG(("WebSocketChannel::BeginOpenInternal: cannot async open\n"));
AbortSession(NS_ERROR_UNEXPECTED);
return;
}
rv = NS_MaybeOpenChannelUsingAsyncOpen(localChannel, this);
if (NS_FAILED(rv)) {
LOG(("WebSocketChannel::BeginOpenInternal: cannot async open\n"));
AbortSession(NS_ERROR_CONNECTION_REFUSED);
return;
}
mOpenedHttpChannel = true;
rv = NS_NewTimerWithCallback(getter_AddRefs(mOpenTimer), this, mOpenTimeout,
nsITimer::TYPE_ONE_SHOT);
if (NS_FAILED(rv)) {
LOG(
("WebSocketChannel::BeginOpenInternal: cannot initialize open "
"timer\n"));
AbortSession(NS_ERROR_UNEXPECTED);
return;
}
}
bool WebSocketChannel::IsPersistentFramePtr() {
return (mFramePtr >= mBuffer && mFramePtr < mBuffer + mBufferSize);
}
// Extends the internal buffer by count and returns the total
// amount of data available for read
//
// Accumulated fragment size is passed in instead of using the member
// variable beacuse when transitioning from the stack to the persistent
// read buffer we want to explicitly include them in the buffer instead
// of as already existing data.
bool WebSocketChannel::UpdateReadBuffer(uint8_t* buffer, uint32_t count,
uint32_t accumulatedFragments,
uint32_t* available) {
LOG(("WebSocketChannel::UpdateReadBuffer() %p [%p %u]\n", this, buffer,
count));
if (!mBuffered) mFramePtr = mBuffer;
MOZ_ASSERT(IsPersistentFramePtr(), "update read buffer bad mFramePtr");
MOZ_ASSERT(mFramePtr - accumulatedFragments >= mBuffer,
"reserved FramePtr bad");
if (mBuffered + count <= mBufferSize) {
// append to existing buffer
LOG(("WebSocketChannel: update read buffer absorbed %u\n", count));
} else if (mBuffered + count - (mFramePtr - accumulatedFragments - mBuffer) <=
mBufferSize) {
// make room in existing buffer by shifting unused data to start
mBuffered -= (mFramePtr - mBuffer - accumulatedFragments);
LOG(("WebSocketChannel: update read buffer shifted %u\n", mBuffered));
::memmove(mBuffer, mFramePtr - accumulatedFragments, mBuffered);
mFramePtr = mBuffer + accumulatedFragments;
} else {
// existing buffer is not sufficient, extend it
mBufferSize += count + 8192 + mBufferSize / 3;
LOG(("WebSocketChannel: update read buffer extended to %u\n", mBufferSize));
uint8_t* old = mBuffer;
mBuffer = (uint8_t*)realloc(mBuffer, mBufferSize);
if (!mBuffer) {
mBuffer = old;
return false;
}
mFramePtr = mBuffer + (mFramePtr - old);
}
::memcpy(mBuffer + mBuffered, buffer, count);
mBuffered += count;
if (available) *available = mBuffered - (mFramePtr - mBuffer);
return true;
}
nsresult WebSocketChannel::ProcessInput(uint8_t* buffer, uint32_t count) {
LOG(("WebSocketChannel::ProcessInput %p [%d %d]\n", this, count, mBuffered));
MOZ_ASSERT(OnSocketThread(), "not on socket thread");
nsresult rv;
// The purpose of ping/pong is to actively probe the peer so that an
// unreachable peer is not mistaken for a period of idleness. This
// implementation accepts any application level read activity as a sign of
// life, it does not necessarily have to be a pong.
ResetPingTimer();
uint32_t avail;
if (!mBuffered) {
// Most of the time we can process right off the stack buffer without
// having to accumulate anything
mFramePtr = buffer;
avail = count;
} else {
if (!UpdateReadBuffer(buffer, count, mFragmentAccumulator, &avail)) {
return NS_ERROR_FILE_TOO_BIG;
}
}
uint8_t* payload;
uint32_t totalAvail = avail;
while (avail >= 2) {
int64_t payloadLength64 = mFramePtr[1] & kPayloadLengthBitsMask;
uint8_t finBit = mFramePtr[0] & kFinalFragBit;
uint8_t rsvBits = mFramePtr[0] & kRsvBitsMask;
uint8_t rsvBit1 = mFramePtr[0] & kRsv1Bit;
uint8_t rsvBit2 = mFramePtr[0] & kRsv2Bit;
uint8_t rsvBit3 = mFramePtr[0] & kRsv3Bit;
uint8_t opcode = mFramePtr[0] & kOpcodeBitsMask;
uint8_t maskBit = mFramePtr[1] & kMaskBit;
uint32_t mask = 0;
uint32_t framingLength = 2;
if (maskBit) framingLength += 4;
if (payloadLength64 < 126) {
if (avail < framingLength) break;
} else if (payloadLength64 == 126) {
// 16 bit length field
framingLength += 2;
if (avail < framingLength) break;
payloadLength64 = mFramePtr[2] << 8 | mFramePtr[3];
if (payloadLength64 < 126) {
// Section 5.2 says that the minimal number of bytes MUST
// be used to encode the length in all cases
LOG(("WebSocketChannel:: non-minimal-encoded payload length"));
return NS_ERROR_ILLEGAL_VALUE;
}
} else {
// 64 bit length
framingLength += 8;
if (avail < framingLength) break;
if (mFramePtr[2] & 0x80) {
// Section 4.2 says that the most significant bit MUST be
// 0. (i.e. this is really a 63 bit value)
LOG(("WebSocketChannel:: high bit of 64 bit length set"));
return NS_ERROR_ILLEGAL_VALUE;
}
// copy this in case it is unaligned
payloadLength64 = NetworkEndian::readInt64(mFramePtr + 2);
if (payloadLength64 <= 0xffff) {
// Section 5.2 says that the minimal number of bytes MUST
// be used to encode the length in all cases
LOG(("WebSocketChannel:: non-minimal-encoded payload length"));
return NS_ERROR_ILLEGAL_VALUE;
}
}
payload = mFramePtr + framingLength;
avail -= framingLength;
LOG(("WebSocketChannel::ProcessInput: payload %" PRId64 " avail %" PRIu32
"\n",
payloadLength64, avail));
CheckedInt<int64_t> payloadLengthChecked(payloadLength64);
payloadLengthChecked += mFragmentAccumulator;
if (!payloadLengthChecked.isValid() ||
payloadLengthChecked.value() > mMaxMessageSize) {
return NS_ERROR_FILE_TOO_BIG;
}
uint32_t payloadLength = static_cast<uint32_t>(payloadLength64);
if (avail < payloadLength) break;
LOG(("WebSocketChannel::ProcessInput: Frame accumulated - opcode %d\n",
opcode));
if (!maskBit && mIsServerSide) {
LOG(
("WebSocketChannel::ProcessInput: unmasked frame received "
"from client\n"));
return NS_ERROR_ILLEGAL_VALUE;
}
if (maskBit) {
if (!mIsServerSide) {
// The server should not be allowed to send masked frames to clients.
// But we've been allowing it for some time, so this should be
// deprecated with care.
LOG(("WebSocketChannel:: Client RECEIVING masked frame."));
}
mask = NetworkEndian::readUint32(payload - 4);
}
if (mask) {
ApplyMask(mask, payload, payloadLength);
} else if (mIsServerSide) {
LOG(
("WebSocketChannel::ProcessInput: masked frame with mask 0 received"
"from client\n"));
return NS_ERROR_ILLEGAL_VALUE;
}
// Control codes are required to have the fin bit set
if (!finBit && (opcode & kControlFrameMask)) {
LOG(("WebSocketChannel:: fragmented control frame code %d\n", opcode));
return NS_ERROR_ILLEGAL_VALUE;
}
if (rsvBits) {
// PMCE sets RSV1 bit in the first fragment when the non-control frame
// is deflated
if (mPMCECompressor && rsvBits == kRsv1Bit && mFragmentAccumulator == 0 &&
!(opcode & kControlFrameMask)) {
mPMCECompressor->SetMessageDeflated();
LOG(("WebSocketChannel::ProcessInput: received deflated frame\n"));
} else {
LOG(("WebSocketChannel::ProcessInput: unexpected reserved bits %x\n",
rsvBits));
return NS_ERROR_ILLEGAL_VALUE;
}
}
if (!finBit || opcode == nsIWebSocketFrame::OPCODE_CONTINUATION) {
// This is part of a fragment response
// Only the first frame has a non zero op code: Make sure we don't see a
// first frame while some old fragments are open
if ((mFragmentAccumulator != 0) &&
(opcode != nsIWebSocketFrame::OPCODE_CONTINUATION)) {
LOG(("WebSocketChannel:: nested fragments\n"));
return NS_ERROR_ILLEGAL_VALUE;
}
LOG(("WebSocketChannel:: Accumulating Fragment %" PRIu32 "\n",
payloadLength));
if (opcode == nsIWebSocketFrame::OPCODE_CONTINUATION) {
// Make sure this continuation fragment isn't the first fragment
if (mFragmentOpcode == nsIWebSocketFrame::OPCODE_CONTINUATION) {
LOG(("WebSocketHeandler:: continuation code in first fragment\n"));
return NS_ERROR_ILLEGAL_VALUE;
}
// For frag > 1 move the data body back on top of the headers
// so we have contiguous stream of data
MOZ_ASSERT(mFramePtr + framingLength == payload,
"payload offset from frameptr wrong");
::memmove(mFramePtr, payload, avail);
payload = mFramePtr;
if (mBuffered) mBuffered -= framingLength;
} else {
mFragmentOpcode = opcode;
}
if (finBit) {
LOG(("WebSocketChannel:: Finalizing Fragment\n"));
payload -= mFragmentAccumulator;
payloadLength += mFragmentAccumulator;
avail += mFragmentAccumulator;
mFragmentAccumulator = 0;
opcode = mFragmentOpcode;
// reset to detect if next message illegally starts with continuation
mFragmentOpcode = nsIWebSocketFrame::OPCODE_CONTINUATION;
} else {
opcode = nsIWebSocketFrame::OPCODE_CONTINUATION;
mFragmentAccumulator += payloadLength;
}
} else if (mFragmentAccumulator != 0 && !(opcode & kControlFrameMask)) {
// This frame is not part of a fragment sequence but we
// have an open fragment.. it must be a control code or else
// we have a problem
LOG(("WebSocketChannel:: illegal fragment sequence\n"));
return NS_ERROR_ILLEGAL_VALUE;
}
if (mServerClosed) {
LOG(("WebSocketChannel:: ignoring read frame code %d after close\n",
opcode));
// nop
} else if (mStopped) {
LOG(("WebSocketChannel:: ignoring read frame code %d after completion\n",
opcode));
} else if (opcode == nsIWebSocketFrame::OPCODE_TEXT) {
bool isDeflated = mPMCECompressor && mPMCECompressor->IsMessageDeflated();
LOG(("WebSocketChannel:: %stext frame received\n",
isDeflated ? "deflated " : ""));
if (mListenerMT) {
nsCString utf8Data;
if (isDeflated) {
rv = mPMCECompressor->Inflate(payload, payloadLength, utf8Data);
if (NS_FAILED(rv)) {
return rv;
}
LOG(
("WebSocketChannel:: message successfully inflated "
"[origLength=%d, newLength=%d]\n",
payloadLength, utf8Data.Length()));
} else {
if (!utf8Data.Assign((const char*)payload, payloadLength,
mozilla::fallible)) {
return NS_ERROR_OUT_OF_MEMORY;
}
}
// Section 8.1 says to fail connection if invalid utf-8 in text message
if (!IsUtf8(utf8Data)) {
LOG(("WebSocketChannel:: text frame invalid utf-8\n"));
return NS_ERROR_CANNOT_CONVERT_DATA;
}
RefPtr<WebSocketFrame> frame = mService->CreateFrameIfNeeded(
finBit, rsvBit1, rsvBit2, rsvBit3, opcode, maskBit, mask, utf8Data);
if (frame) {
mService->FrameReceived(mSerial, mInnerWindowID, frame.forget());
}
mTargetThread->Dispatch(new CallOnMessageAvailable(this, utf8Data, -1),
NS_DISPATCH_NORMAL);
if (mConnectionLogService && !mPrivateBrowsing) {
mConnectionLogService->NewMsgReceived(mHost, mSerial, count);
LOG(("Added new msg received for %s", mHost.get()));
}
}
} else if (opcode & kControlFrameMask) {
// control frames
if (payloadLength > 125) {
LOG(("WebSocketChannel:: bad control frame code %d length %d\n", opcode,
payloadLength));
return NS_ERROR_ILLEGAL_VALUE;
}
RefPtr<WebSocketFrame> frame = mService->CreateFrameIfNeeded(
finBit, rsvBit1, rsvBit2, rsvBit3, opcode, maskBit, mask, payload,
payloadLength);
if (opcode == nsIWebSocketFrame::OPCODE_CLOSE) {
LOG(("WebSocketChannel:: close received\n"));
mServerClosed = true;
mServerCloseCode = CLOSE_NO_STATUS;
if (payloadLength >= 2) {
mServerCloseCode = NetworkEndian::readUint16(payload);
LOG(("WebSocketChannel:: close recvd code %u\n", mServerCloseCode));
uint16_t msglen = static_cast<uint16_t>(payloadLength - 2);
if (msglen > 0) {
mServerCloseReason.SetLength(msglen);
memcpy(mServerCloseReason.BeginWriting(), (const char*)payload + 2,
msglen);
// section 8.1 says to replace received non utf-8 sequences
// (which are non-conformant to send) with u+fffd,
// but secteam feels that silently rewriting messages is
// inappropriate - so we will fail the connection instead.
if (!IsUtf8(mServerCloseReason)) {
LOG(("WebSocketChannel:: close frame invalid utf-8\n"));
return NS_ERROR_CANNOT_CONVERT_DATA;
}
LOG(("WebSocketChannel:: close msg %s\n",
mServerCloseReason.get()));
}
}
if (mCloseTimer) {
mCloseTimer->Cancel();
mCloseTimer = nullptr;
}
if (frame) {
// We send the frame immediately becuase we want to have it dispatched
// before the CallOnServerClose.
mService->FrameReceived(mSerial, mInnerWindowID, frame.forget());
frame = nullptr;
}
if (mListenerMT) {
mTargetThread->Dispatch(
new CallOnServerClose(this, mServerCloseCode, mServerCloseReason),
NS_DISPATCH_NORMAL);
}
if (mClientClosed) ReleaseSession();
} else if (opcode == nsIWebSocketFrame::OPCODE_PING) {
LOG(("WebSocketChannel:: ping received\n"));
GeneratePong(payload, payloadLength);
} else if (opcode == nsIWebSocketFrame::OPCODE_PONG) {
// opcode OPCODE_PONG: the mere act of receiving the packet is all we
// need to do for the pong to trigger the activity timers
LOG(("WebSocketChannel:: pong received\n"));
} else {
/* unknown control frame opcode */
LOG(("WebSocketChannel:: unknown control op code %d\n", opcode));
return NS_ERROR_ILLEGAL_VALUE;
}
if (mFragmentAccumulator) {
// Remove the control frame from the stream so we have a contiguous
// data buffer of reassembled fragments
LOG(("WebSocketChannel:: Removing Control From Read buffer\n"));
MOZ_ASSERT(mFramePtr + framingLength == payload,
"payload offset from frameptr wrong");
::memmove(mFramePtr, payload + payloadLength, avail - payloadLength);
payload = mFramePtr;
avail -= payloadLength;
if (mBuffered) mBuffered -= framingLength + payloadLength;
payloadLength = 0;
}
if (frame) {
mService->FrameReceived(mSerial, mInnerWindowID, frame.forget());
}
} else if (opcode == nsIWebSocketFrame::OPCODE_BINARY) {
bool isDeflated = mPMCECompressor && mPMCECompressor->IsMessageDeflated();
LOG(("WebSocketChannel:: %sbinary frame received\n",
isDeflated ? "deflated " : ""));
if (mListenerMT) {
nsCString binaryData;
if (isDeflated) {
rv = mPMCECompressor->Inflate(payload, payloadLength, binaryData);
if (NS_FAILED(rv)) {
return rv;
}
LOG(
("WebSocketChannel:: message successfully inflated "
"[origLength=%d, newLength=%d]\n",
payloadLength, binaryData.Length()));
} else {
if (!binaryData.Assign((const char*)payload, payloadLength,
mozilla::fallible)) {
return NS_ERROR_OUT_OF_MEMORY;
}
}
RefPtr<WebSocketFrame> frame =
mService->CreateFrameIfNeeded(finBit, rsvBit1, rsvBit2, rsvBit3,
opcode, maskBit, mask, binaryData);
if (frame) {
mService->FrameReceived(mSerial, mInnerWindowID, frame.forget());
}
mTargetThread->Dispatch(
new CallOnMessageAvailable(this, binaryData, binaryData.Length()),
NS_DISPATCH_NORMAL);
// To add the header to 'Networking Dashboard' log
if (mConnectionLogService && !mPrivateBrowsing) {
mConnectionLogService->NewMsgReceived(mHost, mSerial, count);
LOG(("Added new received msg for %s", mHost.get()));
}
}
} else if (opcode != nsIWebSocketFrame::OPCODE_CONTINUATION) {
/* unknown opcode */
LOG(("WebSocketChannel:: unknown op code %d\n", opcode));
return NS_ERROR_ILLEGAL_VALUE;
}
mFramePtr = payload + payloadLength;
avail -= payloadLength;
totalAvail = avail;
}
// Adjust the stateful buffer. If we were operating off the stack and
// now have a partial message then transition to the buffer, or if
// we were working off the buffer but no longer have any active state
// then transition to the stack
if (!IsPersistentFramePtr()) {
mBuffered = 0;
if (mFragmentAccumulator) {
LOG(("WebSocketChannel:: Setup Buffer due to fragment"));
if (!UpdateReadBuffer(mFramePtr - mFragmentAccumulator,
totalAvail + mFragmentAccumulator, 0, nullptr)) {
return NS_ERROR_FILE_TOO_BIG;
}
// UpdateReadBuffer will reset the frameptr to the beginning
// of new saved state, so we need to skip past processed framgents
mFramePtr += mFragmentAccumulator;
} else if (totalAvail) {
LOG(("WebSocketChannel:: Setup Buffer due to partial frame"));
if (!UpdateReadBuffer(mFramePtr, totalAvail, 0, nullptr)) {
return NS_ERROR_FILE_TOO_BIG;
}
}
} else if (!mFragmentAccumulator && !totalAvail) {
// If we were working off a saved buffer state and there is no partial
// frame or fragment in process, then revert to stack behavior
LOG(("WebSocketChannel:: Internal buffering not needed anymore"));
mBuffered = 0;
// release memory if we've been processing a large message
if (mBufferSize > kIncomingBufferStableSize) {
mBufferSize = kIncomingBufferStableSize;
free(mBuffer);
mBuffer = (uint8_t*)moz_xmalloc(mBufferSize);
}
}
return NS_OK;
}
/* static */
void WebSocketChannel::ApplyMask(uint32_t mask, uint8_t* data, uint64_t len) {
if (!data || len == 0) return;
// Optimally we want to apply the mask 32 bits at a time,
// but the buffer might not be alligned. So we first deal with
// 0 to 3 bytes of preamble individually
while (len && (reinterpret_cast<uintptr_t>(data) & 3)) {
*data ^= mask >> 24;
mask = RotateLeft(mask, 8);
data++;
len--;
}
// perform mask on full words of data
uint32_t* iData = (uint32_t*)data;
uint32_t* end = iData + (len / 4);
NetworkEndian::writeUint32(&mask, mask);
for (; iData < end; iData++) *iData ^= mask;
mask = NetworkEndian::readUint32(&mask);
data = (uint8_t*)iData;
len = len % 4;
// There maybe up to 3 trailing bytes that need to be dealt with
// individually
while (len) {
*data ^= mask >> 24;
mask = RotateLeft(mask, 8);
data++;
len--;
}
}
void WebSocketChannel::GeneratePing() {
nsAutoCString buf;
buf.AssignLiteral("PING");
EnqueueOutgoingMessage(mOutgoingPingMessages,
new OutboundMessage(kMsgTypePing, buf));
}
void WebSocketChannel::GeneratePong(uint8_t* payload, uint32_t len) {
nsAutoCString buf;
buf.SetLength(len);
if (buf.Length() < len) {
LOG(("WebSocketChannel::GeneratePong Allocation Failure\n"));
return;
}
memcpy(buf.BeginWriting(), payload, len);
EnqueueOutgoingMessage(mOutgoingPongMessages,
new OutboundMessage(kMsgTypePong, buf));
}
void WebSocketChannel::EnqueueOutgoingMessage(nsDeque<OutboundMessage>& aQueue,
OutboundMessage* aMsg) {
MOZ_ASSERT(OnSocketThread(), "not on socket thread");
LOG(
("WebSocketChannel::EnqueueOutgoingMessage %p "
"queueing msg %p [type=%s len=%d]\n",
this, aMsg, msgNames[aMsg->GetMsgType()], aMsg->Length()));
aQueue.Push(aMsg);
DoEnqueueOutgoingMessage();
}
uint16_t WebSocketChannel::ResultToCloseCode(nsresult resultCode) {
if (NS_SUCCEEDED(resultCode)) return CLOSE_NORMAL;
switch (resultCode) {
case NS_ERROR_FILE_TOO_BIG:
case NS_ERROR_OUT_OF_MEMORY:
return CLOSE_TOO_LARGE;
case NS_ERROR_CANNOT_CONVERT_DATA:
return CLOSE_INVALID_PAYLOAD;
case NS_ERROR_UNEXPECTED:
return CLOSE_INTERNAL_ERROR;
default:
return CLOSE_PROTOCOL_ERROR;
}
}
void WebSocketChannel::PrimeNewOutgoingMessage() {
LOG(("WebSocketChannel::PrimeNewOutgoingMessage() %p\n", this));
MOZ_ASSERT(OnSocketThread(), "not on socket thread");
MOZ_ASSERT(!mCurrentOut, "Current message in progress");
nsresult rv = NS_OK;
mCurrentOut = mOutgoingPongMessages.PopFront();
if (mCurrentOut) {
MOZ_ASSERT(mCurrentOut->GetMsgType() == kMsgTypePong, "Not pong message!");
} else {
mCurrentOut = mOutgoingPingMessages.PopFront();
if (mCurrentOut)