Recognize WebSockets over HTTP/3

Summary:
`HQStreamCodec` does not recognize an Extended CONNECT with `:protocol=websocket` as a WebSocket upgrade, so revproxy rejects it with `400 connect_method_disabled`. Mark downstream WebSocket CONNECTs as upgrades, matching `HTTP2Codec`, without a codec-specific enable flag. Both `HQSession` and `HQMultiCodec` use this behavior; D125389660 handles revproxy's per-VIP acceptance policy and `SETTINGS_ENABLE_CONNECT_PROTOCOL` advertisement.

Skip unknown frames, including GREASE and padding, on ordinary and WebSocket CONNECT streams. RFC 9114 section 4.4 requires rejecting known non-DATA frames; sections 7.2.8 and 9 require ignoring GREASE and unknown frame types. Known non-DATA frames still fail with `H3_FRAME_UNEXPECTED`.

Tests also pin the existing `QPACKCodec` conversion of an upgrade response from 101 to HTTP/3 status 200. The changed `xplat/proxygen/lib` mirrors are byte-identical.

Reviewed By: hanidamlaj

Differential Revision: D125389659

fbshipit-source-id: 808a4cea2036e5f235f318086215e44cf041b481

proxygen/lib/http/codec/HQStreamCodec.cpp

@@ -21,6 +21,29 @@ namespace proxygen::hq {
 using namespace folly;
 using namespace folly::io;
 
+namespace {
+bool isKnownFrameType(FrameType type) {
+  switch (type) {
+    case hq::FrameType::DATA:
+    case hq::FrameType::HEADERS:
+    case hq::FrameType::CANCEL_PUSH:
+    case hq::FrameType::SETTINGS:
+    case hq::FrameType::PUSH_PROMISE:
+    case hq::FrameType::GOAWAY:
+    case hq::FrameType::MAX_PUSH_ID:
+    case hq::FrameType::PRIORITY_UPDATE:
+    case hq::FrameType::PUSH_PRIORITY_UPDATE:
+    case hq::FrameType::FB_PRIORITY_UPDATE:
+    case hq::FrameType::FB_PUSH_PRIORITY_UPDATE:
+    case hq::FrameType::WEBTRANSPORT_BIDI:
+      return true;
+    case hq::FrameType::PADDING:
+    default:
+      return false;
+  }
+}
+} // namespace
+
 HQStreamCodec::HQStreamCodec(StreamID streamId,
                              TransportDirection direction,
                              QPACKCodec& headerCodec,
@@ -46,7 +69,7 @@ HQStreamCodec::~HQStreamCodec() {
 }
 
 ParseResult HQStreamCodec::checkFrameAllowed(FrameType type) {
-  if (isConnect_ && type != hq::FrameType::DATA) {
+  if (isConnect_ && type != hq::FrameType::DATA && isKnownFrameType(type)) {
     setParseErrorContext(
         "non-data-frame-on-connect",
         folly::to<std::string>("type=", getFrameTypeString(type)));
@@ -238,6 +261,10 @@ void HQStreamCodec::onHeadersComplete(HTTPHeaderSize decodedSize,
   if (transportDirection_ == TransportDirection::DOWNSTREAM &&
       msg->getMethod() == HTTPMethod::CONNECT) {
     isConnect_ = true;
+    const auto* protocol = msg->getUpgradeProtocol();
+    if (protocol && *protocol == headers::kWebsocketString) {
+      msg->setIngressWebsocketUpgrade();
+    }
   }
 
   if (acknowledge) {

proxygen/lib/http/codec/test/HQCodecTest.cpp

@@ -879,6 +879,107 @@ TEST_F(HQCodecTest, WebTransportProtocol) {
   EXPECT_EQ("webtransport", *callbacks_.msg->getUpgradeProtocol());
 }
 
+namespace {
+HTTPMessage getWebsocketRequest() {
+  HTTPMessage req = getGetRequest("/websocket");
+  req.setSecure(true);
+  req.setEgressWebsocketUpgrade();
+  return req;
+}
+} // namespace
+
+TEST_F(HQCodecTest, WebsocketUpgrade) {
+  upstreamCodec_->generateHeader(queue_, streamId_, getWebsocketRequest());
+  parse();
+
+  EXPECT_EQ(callbacks_.streamErrors, 0);
+  EXPECT_EQ(HTTPMethod::CONNECT, callbacks_.msg->getMethod());
+  EXPECT_TRUE(callbacks_.msg->isIngressWebsocketUpgrade());
+  EXPECT_EQ(headers::kWebsocketString, *callbacks_.msg->getUpgradeProtocol());
+}
+
+TEST_F(HQCodecTest, WebsocketResponseIs200) {
+  HTTPMessage resp;
+  resp.setStatusCode(101);
+  resp.setEgressWebsocketUpgrade();
+  downstreamCodec_->generateHeader(queue_, streamId_, resp);
+  parseUpstream();
+
+  EXPECT_EQ(callbacks_.streamErrors, 0);
+  EXPECT_EQ(callbacks_.msg->getStatusCode(), 200);
+}
+
+TEST_F(HQCodecTest, WebsocketMalformedRequest) {
+  const std::string kConnect{"CONNECT"};
+  const std::string kWebsocketPath{"/websocket"};
+  const std::string kSchemeHttps{"https"};
+  const std::string kHost{"www.foo.com"};
+  using compress::Header;
+  auto method = Header::makeHeaderForTest(headers::kMethod, kConnect);
+  auto protocol =
+      Header::makeHeaderForTest(headers::kProtocol, headers::kWebsocketString);
+  auto path = Header::makeHeaderForTest(headers::kPath, kWebsocketPath);
+  auto scheme = Header::makeHeaderForTest(headers::kScheme, kSchemeHttps);
+  auto authority = Header::makeHeaderForTest(headers::kAuthority, kHost);
+  std::vector<std::vector<Header>> malformedRequests{
+      {method, protocol, scheme, authority},
+      {method, protocol, path, authority},
+      {method, protocol, protocol, path, scheme, authority},
+  };
+
+  for (auto& reqHeaders : malformedRequests) {
+    hq::writeHeaders(queue_,
+                     qpackUpstream_.encode(reqHeaders, streamId_).stream);
+    HQStreamCodec downstreamCodec(
+        streamId_,
+        TransportDirection::DOWNSTREAM,
+        qpackDownstream_,
+        qpackDownEncoderWriteBuf_,
+        qpackDownDecoderWriteBuf_,
+        [] { return std::numeric_limits<uint64_t>::max(); },
+        ingressSettings_);
+    downstreamCodec.setCallback(&callbacks_);
+    downstreamCodec.onIngress(*queue_.front());
+    queue_.reset();
+  }
+
+  EXPECT_EQ(callbacks_.headersComplete, 0);
+  EXPECT_EQ(callbacks_.streamErrors, malformedRequests.size());
+  EXPECT_EQ(callbacks_.sessionErrors, 0);
+}
+
+TEST_F(HQCodecTest, WebsocketIgnoresUnknownFrames) {
+  upstreamCodec_->generateHeader(queue_, streamId_, getWebsocketRequest());
+  writeValidFrame(queue_, FrameType(*getGreaseId(0)));
+  upstreamCodec_->generatePadding(queue_, streamId_, 16);
+  writeValidFrame(queue_, FrameType::DATA);
+  parse();
+  EXPECT_EQ(callbacks_.bodyCalls, 1);
+  EXPECT_EQ(callbacks_.sessionErrors, 0);
+
+  writeValidFrame(queue_, FrameType::HEADERS);
+  parse();
+  EXPECT_EQ(callbacks_.sessionErrors, 1);
+  EXPECT_EQ(callbacks_.lastParseError->getHttp3ErrorCode(),
+            HTTP3::ErrorCode::HTTP_FRAME_UNEXPECTED);
+}
+
+TEST_F(HQCodecTest, ConnectIgnoresUnknownFrames) {
+  HTTPMessage req;
+  req.setMethod(HTTPMethod::CONNECT);
+  req.getHeaders().add(HTTP_HEADER_HOST, "myhost:1234");
+  upstreamCodec_->generateHeader(queue_, streamId_, req);
+  writeValidFrame(queue_, FrameType(*getGreaseId(0)));
+  upstreamCodec_->generatePadding(queue_, streamId_, 16);
+  writeValidFrame(queue_, FrameType::DATA);
+  parse();
+  EXPECT_EQ(callbacks_.headersComplete, 1);
+  EXPECT_FALSE(callbacks_.msg->isIngressWebsocketUpgrade());
+  EXPECT_EQ(callbacks_.bodyCalls, 1);
+  EXPECT_EQ(callbacks_.streamErrors, 0);
+  EXPECT_EQ(callbacks_.sessionErrors, 0);
+}
+
 TEST_F(HQCodecTest, PaddingIgnored) {
   std::string data("abcde");
   auto buf = folly::IOBuf::copyBuffer(data.data(), data.length());

proxygen/lib/http/codec/test/HQMultiCodecTest.cpp

@@ -175,6 +175,25 @@ TEST_F(HQMultiCodecTest, RemoveCodecWithBlockedTrailers) {
   EXPECT_EQ(callbacks.trailers, 0);
 }
 
+TEST_F(HQMultiCodecTest, WebsocketUpgrade) {
+  HQMultiCodec upstreamCodec{TransportDirection::UPSTREAM};
+  HTTPCodec::StreamID id{0};
+  upstreamCodec.addCodec(id);
+  auto req = getGetRequest("/websocket");
+  req.setSecure(true);
+  req.setEgressWebsocketUpgrade();
+  upstreamCodec.generateHeader(writeBuf_, id, req);
+
+  FakeHTTPCodecCallback callbacks;
+  callbacks.setSessionStreamId(kSessionStreamId);
+  codec_.addCodec(id);
+  codec_.setCallback(&callbacks);
+  EXPECT_TRUE(codec_.setCurrentStream(id));
+  codec_.onIngress(*writeBuf_.front());
+  EXPECT_EQ(callbacks.headersComplete, 1);
+  EXPECT_TRUE(callbacks.msg->isIngressWebsocketUpgrade());
+}
+
 TEST_F(HQMultiCodecTest, Generic) {
   EXPECT_TRUE(codec_.supportsParallelRequests());
   EXPECT_NE(codec_.getEgressSettings(), nullptr);

proxygen/lib/http/session/test/HQDownstreamSessionTest.cpp

@@ -2082,6 +2082,76 @@ TEST_P(HQDownstreamSessionTestDatagram, BufferedDatagramDroppedOnNon2xx) {
   hqSession_->dropConnection();
 }
 
+namespace {
+HTTPMessage getWebsocketRequest() {
+  auto req = getGetRequest("/websocket");
+  req.setSecure(true);
+  req.setEgressWebsocketUpgrade();
+  return req;
+}
+
+HTTPMessage getWebsocketResponse() {
+  HTTPMessage resp;
+  resp.setStatusCode(101);
+  resp.setEgressWebsocketUpgrade();
+  return resp;
+}
+} // namespace
+
+TEST_P(HQDownstreamSessionTest, WebsocketExtendedConnect) {
+  auto handler = addSimpleStrictHandler();
+  handler->expectHeaders([&](std::shared_ptr<HTTPMessage> msg) {
+    EXPECT_TRUE(msg->isIngressWebsocketUpgrade());
+    handler->txn_->sendHeaders(getWebsocketResponse());
+    handler->sendBody(10);
+  });
+  auto id = sendRequest(getWebsocketRequest(), /*eom=*/false);
+  flushRequestsAndLoopN(1);
+
+  handler->expectBody();
+  auto& request = getStream(id);
+  request.codec->generateBody(
+      request.buf, request.id, makeBuf(20), HTTPCodec::NoPadding, false);
+  flushRequestsAndLoopN(1);
+
+  handler->expectEOM();
+  request.readEOF = true;
+  flushRequestsAndLoopN(1);
+
+  handler->expectDetachTransaction();
+  handler->sendEOM();
+  flushRequestsAndLoop();
+
+  auto& stream = socketDriver_->streams_[id];
+  EXPECT_TRUE(stream.writeEOF);
+  FakeHTTPCodecCallback callbacks;
+  request.codec->setCallback(&callbacks);
+  request.codec->onIngress(*stream.writeBuf.front());
+  request.codec->onIngressEOF();
+  EXPECT_EQ(callbacks.msg->getStatusCode(), 200);
+  EXPECT_EQ(callbacks.bodyLength, 10);
+  EXPECT_EQ(callbacks.messageComplete, 1);
+  hqSession_->closeWhenIdle();
+}
+
+TEST_P(HQDownstreamSessionTest, WebsocketExtendedConnectCancel) {
+  auto handler = addSimpleStrictHandler();
+  handler->expectHeaders([&](std::shared_ptr<HTTPMessage> msg) {
+    EXPECT_TRUE(msg->isIngressWebsocketUpgrade());
+    handler->txn_->sendHeaders(getWebsocketResponse());
+  });
+  auto id = sendRequest(getWebsocketRequest(), /*eom=*/false);
+  flushRequestsAndLoopN(1);
+
+  handler->expectDetachTransaction();
+  handler->txn_->sendAbort(ErrorCode::CANCEL);
+  flushRequestsAndLoop();
+
+  EXPECT_EQ(socketDriver_->streams_[id].error,
+            folly::make_optional(HTTP3::ErrorCode::HTTP_REQUEST_CANCELLED));
+  hqSession_->closeWhenIdle();
+}
+
 // Just open a stream and send nothing
 TEST_P(HQDownstreamSessionTest, zeroBytes) {
   auto id = nextStreamId();