Quelle WebSocketChannel.cpp Sprache: C

/* -*- 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 <algorithm>

#include "WebSocketChannel.h"

#include "WebSocketConnectionBase.h"
#include "WebSocketFrame.h"
#include "WebSocketLog.h"
#include "mozilla/Atomics.h"
#include "mozilla/Attributes.h"
#include "mozilla/Base64.h"
#include "mozilla/Components.h"
#include "mozilla/EndianUtils.h"
#include "mozilla/MathAlgorithms.h"
#include "mozilla/ScopeExit.h"
#include "mozilla/StaticMutex.h"
#include "mozilla/StaticPrefs_privacy.h"
#include "mozilla/Telemetry.h"
#include "mozilla/TimeStamp.h"
#include "mozilla/Utf8.h"
#include "mozilla/net/WebSocketEventService.h"
#include "nsCRT.h"
#include "nsCharSeparatedTokenizer.h"
#include "nsComponentManagerUtils.h"
#include "nsError.h"
#include "nsIAsyncVerifyRedirectCallback.h"
#include "nsICancelable.h"
#include "nsIChannel.h"
#include "nsIClassOfService.h"
#include "nsICryptoHash.h"
#include "nsIDNSRecord.h"
#include "nsIDNSService.h"
#include "nsIDashboardEventNotifier.h"
#include "nsIEventTarget.h"
#include "nsIHttpChannel.h"
#include "nsIIOService.h"
#include "nsINSSErrorsService.h"
#include "nsINetworkLinkService.h"
#include "nsINode.h"
#include "nsIObserverService.h"
#include "nsIPrefBranch.h"
#include "nsIProtocolHandler.h"
#include "nsIProtocolProxyService.h"
#include "nsIProxiedChannel.h"
#include "nsIProxyInfo.h"
#include "nsIRandomGenerator.h"
#include "nsIRunnable.h"
#include "nsISocketTransport.h"
#include "nsITLSSocketControl.h"
#include "nsITransportProvider.h"
#include "nsITransportSecurityInfo.h"
#include "nsIURI.h"
#include "nsIURIMutator.h"
#include "nsNetCID.h"
#include "nsNetUtil.h"
#include "nsProxyRelease.h"
#include "nsServiceManagerUtils.h"
#include "nsSocketTransportService2.h"
#include "nsStringStream.h"
#include "nsThreadUtils.h"
#include "plbase64.h"
#include "prmem.h"
#include "prnetdb.h"
#include "zlib.h"

// 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::net {

NS_IMPL_ISUPPORTS(WebSocketChannel, nsIWebSocketChannel, nsIHttpUpgradeListener,
                  nsIRequestObserver, nsIStreamListener, nsIProtocolHandler,
                  nsIInputStreamCallback, nsIOutputStreamCallback,
                  nsITimerCallback, nsIDNSListener, nsIProtocolProxyCallback,
                  nsIInterfaceRequestor, nsIChannelEventSink,
                  nsIThreadRetargetableRequest, nsIObserver, nsINamed)

// 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/path/port.
class FailDelay {
public:
  FailDelay(nsCString address, nsCString path, int32_t port)
      : mAddress(std::move(address)), mPath(std::move(path)), 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, path=%s, port=%d: incremented delay "
         "to "
         "%" PRIu32,
         mAddress.get(), mPath.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)
  nsCString mPath;
  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;
    prefService = mozilla::components::Preferences::Service();
    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, nsCString& path, int32_t port) {
    if (mDelaysDisabled) return;

    UniquePtr<FailDelay> record(new FailDelay(address, path, 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, nsCString& path, 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->mPath.Equals(path) &&
          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;
      }
      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->mPath, 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;
          MutexAutoLock lock(ws->mMutex);
          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, nsCString& path, 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->mPath.Equals(path) &&
           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 hostFound = (sManager->IndexOf(ws->mAddress, ws->mOriginSuffix) >= 0);

    uint32_t failIndex = 0;
    FailDelay* fail = sManager->mFailures.Lookup(ws->mAddress, ws->mPath,
                                                 ws->mPort, &failIndex);
    bool existingFail = fail != nullptr;

    // Always add ourselves to queue, even if we'll connect immediately
    UniquePtr<nsOpenConn> newdata(
        new nsOpenConn(ws->mAddress, ws->mOriginSuffix, existingFail, ws));

    // If a connection has not previously failed then prioritize it over
    // connections that have
    if (existingFail) {
      sManager->mQueue.AppendElement(std::move(newdata));
    } else {
      uint32_t insertionIndex = sManager->IndexOfFirstFailure();
      MOZ_ASSERT(insertionIndex <= sManager->mQueue.Length(),
                 "Insertion index outside bounds");
      sManager->mQueue.InsertElementAt(insertionIndex, std::move(newdata));
    }

    if (hostFound) {
      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->mPath,
                               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, aChannel->mOriginSuffix);
  }

  // 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->mPath, 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, %s, %d completed"
               " [this=%p]",
               aChannel->mAddress.get(), aChannel->mPath.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, %s, %d failed: [this=%p]",
             aChannel->mAddress.get(), aChannel->mPath.get(),
             (int)aChannel->mPort, aChannel));
        sManager->mFailures.Add(aChannel->mAddress, aChannel->mPath,
                                aChannel->mPort);
      }
    }

    if (NS_IsMainThread()) {
      ContinueOnStopSession(aChannel, aReason);
    } else {
      NS_DispatchToMainThread(NS_NewRunnableFunction(
          "nsWSAdmissionManager::ContinueOnStopSession",
          [channel = RefPtr{aChannel}, reason = aReason]() {
            StaticMutexAutoLock lock(sLock);
            if (!sManager) {
              return;
            }

            nsWSAdmissionManager::ContinueOnStopSession(channel, reason);
          }));
    }
  }

  static void ContinueOnStopSession(WebSocketChannel* aChannel,
                                    nsresult aReason) {
    sLock.AssertCurrentThreadOwns();
    MOZ_ASSERT(NS_IsMainThread(), "not main thread");

    if (!aChannel->mConnecting) {
      return;
    }

    // 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.
#ifdef DEBUG
    {
      MutexAutoLock lock(aChannel->mMutex);
      MOZ_ASSERT(
          NS_FAILED(aReason) || aChannel->mScriptCloseCode == CLOSE_GOING_AWAY,
          "websocket closed while connecting w/o failing?");
    }
#endif
    Unused << aReason;

    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, aChannel->mOriginSuffix);
    }
  }

  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, nsCString& originSuffix, bool failed,
               WebSocketChannel* channel)
        : mAddress(addr),
          mOriginSuffix(originSuffix),
          mFailed(failed),
          mChannel(channel) {
      MOZ_COUNT_CTOR(nsOpenConn);
    }
    MOZ_COUNTED_DTOR(nsOpenConn)

    nsCString mAddress;
    nsCString mOriginSuffix;
    bool mFailed = false;
    RefPtr<WebSocketChannel> mChannel;
  };

  void ConnectNext(nsCString& hostName, nsCString& originSuffix) {
    MOZ_ASSERT(NS_IsMainThread(), "not main thread");

    int32_t index = IndexOf(hostName, originSuffix);
    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& aAddress, nsCString& aOriginSuffix) {
    for (uint32_t i = 0; i < mQueue.Length(); i++) {
      bool isPartitioned = StaticPrefs::privacy_partition_network_state() ||
                           StaticPrefs::privacy_firstparty_isolate();
      if (aAddress == (mQueue[i])->mAddress &&
          (!isPartitioned || aOriginSuffix == (mQueue[i])->mOriginSuffix)) {
        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;
  }

  // Returns the index of the first entry that failed, or else the last entry if
  // none found
  uint32_t IndexOfFirstFailure() {
    for (uint32_t i = 0; i < mQueue.Length(); i++) {
      if (mQueue[i]->mFailed) return i;
    }
    return mQueue.Length();
  }

  // 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 MOZ_GUARDED_BY(sLock);
  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 Runnable {
public:
  CallAcknowledge(WebSocketChannel* aChannel, uint32_t aSize)
      : Runnable("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) {
        if (!_retval.Append(reinterpret_cast<char*>(mBuffer), inflated,
                            fallible)) {
          return NS_ERROR_OUT_OF_MEMORY;
        }
      }

      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]{0};
};

//-----------------------------------------------------------------------------
// 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=%zd, "
           "original=%zd]",
           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),
      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),
      mCurrentOutSent(0),
      mHdrOutToSend(0),
      mHdrOut(nullptr),
      mCompressorMutex("WebSocketChannel::mCompressorMutex"),
      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 = mozilla::components::Dashboard::Service(&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::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 (!mIOThread) {
        // there has not been an asyncopen yet on the object and then we need
        // no ping.
        LOG(("WebSocket: early object, no ping needed"));
      } else {
        mIOThread->Dispatch(
            NewRunnableMethod("net::WebSocketChannel::OnNetworkChanged", this,
                              &WebSocketChannel::OnNetworkChanged),
            NS_DISPATCH_NORMAL);
      }
    }
  }

  return NS_OK;
}

nsresult WebSocketChannel::OnNetworkChanged() {
  if (!mDataStarted) {
    LOG(("WebSocket: data not started yet, no ping needed"));
    return NS_OK;
  }

  MOZ_ASSERT(mIOThread->IsOnCurrentThread(), "not on right 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(); }

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();
  }
}

// MainThread
void WebSocketChannel::BeginOpenInternal() {
  LOG(("WebSocketChannel::BeginOpenInternal() %p\n", this));
  MOZ_ASSERT(NS_IsMainThread(), "not main thread");

  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 = localChannel->AsyncOpen(this);

  if (NS_FAILED(rv)) {
    LOG(("WebSocketChannel::BeginOpenInternal: cannot async open\n"));
    AbortSession(NS_ERROR_WEBSOCKET_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(mIOThread->IsOnCurrentThread(), "not on right 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
      MutexAutoLock lock(mCompressorMutex);
      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) {
      if (mListenerMT) {
        nsCString utf8Data;
        {
          MutexAutoLock lock(mCompressorMutex);
          bool isDeflated =
              mPMCECompressor && mPMCECompressor->IsMessageDeflated();
          LOG(("WebSocketChannel:: %stext frame received\n",
               isDeflated ? "deflated " : ""));

          if (isDeflated) {
            rv = mPMCECompressor->Inflate(payload, payloadLength, utf8Data);
            if (NS_FAILED(rv)) {
              return rv;
            }
            LOG(
                ("WebSocketChannel:: message successfully inflated "
                 "[origLength=%d, newLength=%zd]\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());
        }

        if (nsCOMPtr<nsIEventTarget> target = GetTargetThread()) {
          target->Dispatch(new CallOnMessageAvailable(this, utf8Data, -1),
                           NS_DISPATCH_NORMAL);
        } else {
          return NS_ERROR_UNEXPECTED;
        }
        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) {
          if (nsCOMPtr<nsIEventTarget> target = GetTargetThread()) {
            target->Dispatch(new CallOnServerClose(this, mServerCloseCode,
                                                   mServerCloseReason),
                             NS_DISPATCH_NORMAL);
          } else {
            return NS_ERROR_UNEXPECTED;
          }
        }

        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) {
      if (mListenerMT) {
        nsCString binaryData;
        {
          MutexAutoLock lock(mCompressorMutex);
          bool isDeflated =
              mPMCECompressor && mPMCECompressor->IsMessageDeflated();
          LOG(("WebSocketChannel:: %sbinary frame received\n",
               isDeflated ? "deflated " : ""));

          if (isDeflated) {
            rv = mPMCECompressor->Inflate(payload, payloadLength, binaryData);
            if (NS_FAILED(rv)) {
              return rv;
            }
            LOG(
                ("WebSocketChannel:: message successfully inflated "
                 "[origLength=%d, newLength=%zd]\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());
        }

        if (nsCOMPtr<nsIEventTarget> target = GetTargetThread()) {
          target->Dispatch(
              new CallOnMessageAvailable(this, binaryData, binaryData.Length()),
              NS_DISPATCH_NORMAL);
        } else {
          return NS_ERROR_UNEXPECTED;
        }
        // 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(mIOThread->IsOnCurrentThread(), "not on right thread");

  LOG(
      ("WebSocketChannel::EnqueueOutgoingMessage %p "
       "queueing msg %p [type=%s len=%d]\n",
       this, aMsg, msgNames[aMsg->GetMsgType()], aMsg->Length()));

  aQueue.Push(aMsg);
  if (mSocketOut) {
    OnOutputStreamReady(mSocketOut);
  } else {
    DoEnqueueOutgoingMessage();
  }
}

uint16_t WebSocketChannel::ResultToCloseCode(nsresult resultCode) {
  if (NS_SUCCEEDED(resultCode)) return CLOSE_NORMAL;

  switch (resultCode) {
    case NS_ERROR_FILE_TOO_BIG:
--> --------------------

--> maximum size reached

--> --------------------

quality96%

¤ Dauer der Verarbeitung: 0.32 Sekunden (vorverarbeitet) ¤

Wurzel

Suchen

Beweissystem der NASA

Beweissystem Isabelle

NIST Cobol Testsuite

Cephes Mathematical Library

Wiener Entwicklungsmethode

Haftungshinweis

Die Informationen auf dieser Webseite wurden nach bestem Wissen sorgfältig zusammengestellt. Es wird jedoch weder Vollständigkeit, noch Richtigkeit, noch Qualität der bereit gestellten Informationen zugesichert.

Bemerkung:

Die farbliche Syntaxdarstellung ist noch experimentell.