diff --git a/quic/masque/masque_client_bin.cc b/quic/masque/masque_client_bin.cc
index 8e09a12..afd991a 100644
--- a/quic/masque/masque_client_bin.cc
+++ b/quic/masque/masque_client_bin.cc
@@ -29,6 +29,12 @@
                               false,
                               "If true, don't verify the server certificate.");
 
+DEFINE_QUIC_COMMAND_LINE_FLAG(std::string,
+                              masque_mode,
+                              "",
+                              "Allows setting MASQUE mode, valid values are "
+                              "open and legacy. Defaults to open.");
+
 namespace quic {
 
 namespace {
@@ -54,7 +60,8 @@
     masque_url = QuicUrl(absl::StrCat("https://", urls[0]), "https");
   }
   if (masque_url.host().empty()) {
-    QUIC_LOG(ERROR) << "Failed to parse MASQUE server address " << urls[0];
+    std::cerr << "Failed to parse MASQUE server address \"" << urls[0] << "\""
+              << std::endl;
     return 1;
   }
   std::unique_ptr<ProofVerifier> proof_verifier;
@@ -63,15 +70,23 @@
   } else {
     proof_verifier = CreateDefaultProofVerifier(masque_url.host());
   }
-  std::unique_ptr<MasqueEpollClient> masque_client =
-      MasqueEpollClient::Create(masque_url.host(), masque_url.port(),
-                                &epoll_server, std::move(proof_verifier));
+  MasqueMode masque_mode = MasqueMode::kOpen;
+  std::string mode_string = GetQuicFlag(FLAGS_masque_mode);
+  if (mode_string == "legacy") {
+    masque_mode = MasqueMode::kLegacy;
+  } else if (!mode_string.empty() && mode_string != "open") {
+    std::cerr << "Invalid masque_mode \"" << mode_string << "\"" << std::endl;
+    return 1;
+  }
+  std::unique_ptr<MasqueEpollClient> masque_client = MasqueEpollClient::Create(
+      masque_url.host(), masque_url.port(), masque_mode, &epoll_server,
+      std::move(proof_verifier));
   if (masque_client == nullptr) {
     return 1;
   }
 
   std::cerr << "MASQUE is connected " << masque_client->connection_id()
-            << std::endl;
+            << " in " << masque_mode << " mode" << std::endl;
 
   for (size_t i = 1; i < urls.size(); ++i) {
     if (!tools::SendEncapsulatedMasqueRequest(
diff --git a/quic/masque/masque_client_session.cc b/quic/masque/masque_client_session.cc
index eeedccf..5663c4a 100644
--- a/quic/masque/masque_client_session.cc
+++ b/quic/masque/masque_client_session.cc
@@ -3,10 +3,14 @@
 // found in the LICENSE file.
 
 #include "quic/masque/masque_client_session.h"
+#include "absl/algorithm/container.h"
+#include "quic/core/quic_data_reader.h"
+#include "common/platform/api/quiche_text_utils.h"
 
 namespace quic {
 
 MasqueClientSession::MasqueClientSession(
+    MasqueMode masque_mode,
     const QuicConfig& config,
     const ParsedQuicVersionVector& supported_versions,
     QuicConnection* connection,
@@ -20,36 +24,65 @@
                             server_id,
                             crypto_config,
                             push_promise_index),
+      masque_mode_(masque_mode),
       owner_(owner),
       compression_engine_(this) {}
 
 void MasqueClientSession::OnMessageReceived(absl::string_view message) {
   QUIC_DVLOG(1) << "Received DATAGRAM frame of length " << message.length();
+  if (masque_mode_ == MasqueMode::kLegacy) {
+    QuicConnectionId client_connection_id, server_connection_id;
+    QuicSocketAddress target_server_address;
+    std::vector<char> packet;
+    bool version_present;
+    if (!compression_engine_.DecompressDatagram(
+            message, &client_connection_id, &server_connection_id,
+            &target_server_address, &packet, &version_present)) {
+      return;
+    }
 
-  QuicConnectionId client_connection_id, server_connection_id;
-  QuicSocketAddress server_address;
-  std::vector<char> packet;
-  bool version_present;
-  if (!compression_engine_.DecompressDatagram(
-          message, &client_connection_id, &server_connection_id,
-          &server_address, &packet, &version_present)) {
+    auto connection_id_registration =
+        client_connection_id_registrations_.find(client_connection_id);
+    if (connection_id_registration ==
+        client_connection_id_registrations_.end()) {
+      QUIC_DLOG(ERROR) << "MasqueClientSession failed to dispatch "
+                       << client_connection_id;
+      return;
+    }
+    EncapsulatedClientSession* encapsulated_client_session =
+        connection_id_registration->second;
+    encapsulated_client_session->ProcessPacket(
+        absl::string_view(packet.data(), packet.size()), target_server_address);
+
+    QUIC_DVLOG(1) << "Sent " << packet.size() << " bytes to connection for "
+                  << client_connection_id;
     return;
   }
-
-  auto connection_id_registration =
-      client_connection_id_registrations_.find(client_connection_id);
-  if (connection_id_registration == client_connection_id_registrations_.end()) {
-    QUIC_DLOG(ERROR) << "MasqueClientSession failed to dispatch "
-                     << client_connection_id;
+  QuicDataReader reader(message);
+  QuicDatagramFlowId flow_id;
+  if (!reader.ReadVarInt62(&flow_id)) {
+    QUIC_DLOG(ERROR) << "Failed to parse flow_id";
+    return;
+  }
+  auto it =
+      absl::c_find_if(connect_udp_client_states_,
+                      [flow_id](const ConnectUdpClientState& connect_udp) {
+                        return connect_udp.flow_id() == flow_id;
+                      });
+  if (it == connect_udp_client_states_.end()) {
+    QUIC_DLOG(ERROR) << "Received unknown flow_id " << flow_id;
     return;
   }
   EncapsulatedClientSession* encapsulated_client_session =
-      connection_id_registration->second;
-  encapsulated_client_session->ProcessPacket(
-      absl::string_view(packet.data(), packet.size()), server_address);
+      it->encapsulated_client_session();
+  QuicSocketAddress target_server_address = it->target_server_address();
+  QUICHE_DCHECK_NE(encapsulated_client_session, nullptr);
+  QUICHE_DCHECK(target_server_address.IsInitialized());
+  absl::string_view packet = reader.ReadRemainingPayload();
+  encapsulated_client_session->ProcessPacket(packet, target_server_address);
 
-  QUIC_DVLOG(1) << "Sent " << packet.size() << " bytes to connection for "
-                << client_connection_id;
+  QUIC_DVLOG(1) << "Sent " << packet.size()
+                << " bytes to connection for flow_id " << flow_id;
 }
 
 void MasqueClientSession::OnMessageAcked(QuicMessageId message_id,
@@ -61,12 +94,87 @@
   QUIC_DVLOG(1) << "We believe DATAGRAM frame " << message_id << " was lost";
 }
 
-void MasqueClientSession::SendPacket(QuicConnectionId client_connection_id,
-                                     QuicConnectionId server_connection_id,
-                                     absl::string_view packet,
-                                     const QuicSocketAddress& server_address) {
-  compression_engine_.CompressAndSendPacket(
-      packet, client_connection_id, server_connection_id, server_address);
+const MasqueClientSession::ConnectUdpClientState*
+MasqueClientSession::GetOrCreateConnectUdpClientState(
+    const QuicSocketAddress& target_server_address,
+    EncapsulatedClientSession* encapsulated_client_session) {
+  for (const ConnectUdpClientState& client_state : connect_udp_client_states_) {
+    if (client_state.target_server_address() == target_server_address &&
+        client_state.encapsulated_client_session() ==
+            encapsulated_client_session) {
+      // Found existing CONNECT-UDP request.
+      return &client_state;
+    }
+  }
+  // No CONNECT-UDP request found, create a new one.
+  QuicSpdyClientStream* stream = CreateOutgoingBidirectionalStream();
+  if (stream == nullptr) {
+    // Stream flow control limits prevented us from opening a new stream.
+    QUIC_DLOG(ERROR) << "Failed to open CONNECT-UDP stream";
+    return nullptr;
+  }
+
+  QuicDatagramFlowId flow_id = compression_engine_.GetNextFlowId();
+
+  // Send the request.
+  spdy::Http2HeaderBlock headers;
+  headers[":method"] = "CONNECT-UDP";
+  headers[":scheme"] = "masque";
+  headers[":path"] = "/";
+  headers[":authority"] = target_server_address.ToString();
+  headers["datagram-flow-id"] = absl::StrCat(flow_id);
+  size_t bytes_sent =
+      stream->SendRequest(std::move(headers), /*body=*/"", /*fin=*/false);
+  if (bytes_sent == 0) {
+    QUIC_DLOG(ERROR) << "Failed to send CONNECT-UDP request";
+    return nullptr;
+  }
+
+  connect_udp_client_states_.push_back(ConnectUdpClientState(
+      stream, encapsulated_client_session, flow_id, target_server_address));
+  return &connect_udp_client_states_.back();
+}
+
+void MasqueClientSession::SendPacket(
+    QuicConnectionId client_connection_id,
+    QuicConnectionId server_connection_id,
+    absl::string_view packet,
+    const QuicSocketAddress& target_server_address,
+    EncapsulatedClientSession* encapsulated_client_session) {
+  if (masque_mode_ == MasqueMode::kLegacy) {
+    compression_engine_.CompressAndSendPacket(packet, client_connection_id,
+                                              server_connection_id,
+                                              target_server_address);
+    return;
+  }
+  const ConnectUdpClientState* connect_udp = GetOrCreateConnectUdpClientState(
+      target_server_address, encapsulated_client_session);
+  if (connect_udp == nullptr) {
+    QUIC_DLOG(ERROR) << "Failed to create CONNECT-UDP request";
+    return;
+  }
+
+  QuicDatagramFlowId flow_id = connect_udp->flow_id();
+  size_t slice_length =
+      QuicDataWriter::GetVarInt62Len(flow_id) + packet.length();
+  QuicUniqueBufferPtr buffer = MakeUniqueBuffer(
+      connection()->helper()->GetStreamSendBufferAllocator(), slice_length);
+  QuicDataWriter writer(slice_length, buffer.get());
+  if (!writer.WriteVarInt62(flow_id)) {
+    QUIC_BUG << "Failed to write flow_id";
+    return;
+  }
+  if (!writer.WriteBytes(packet.data(), packet.length())) {
+    QUIC_BUG << "Failed to write packet";
+    return;
+  }
+
+  QuicMemSlice slice(std::move(buffer), slice_length);
+  MessageResult message_result = SendMessage(QuicMemSliceSpan(&slice));
+
+  QUIC_DVLOG(1) << "Sent packet to " << target_server_address
+                << " compressed with flow ID " << flow_id
+                << " and got message result " << message_result;
 }
 
 void MasqueClientSession::RegisterConnectionId(
@@ -84,14 +192,65 @@
 }
 
 void MasqueClientSession::UnregisterConnectionId(
-    QuicConnectionId client_connection_id) {
+    QuicConnectionId client_connection_id,
+    EncapsulatedClientSession* encapsulated_client_session) {
   QUIC_DLOG(INFO) << "Unregistering " << client_connection_id;
-  if (client_connection_id_registrations_.find(client_connection_id) !=
-      client_connection_id_registrations_.end()) {
-    client_connection_id_registrations_.erase(client_connection_id);
-    owner_->UnregisterClientConnectionId(client_connection_id);
-    compression_engine_.UnregisterClientConnectionId(client_connection_id);
+  if (masque_mode_ == MasqueMode::kLegacy) {
+    if (client_connection_id_registrations_.find(client_connection_id) !=
+        client_connection_id_registrations_.end()) {
+      client_connection_id_registrations_.erase(client_connection_id);
+      owner_->UnregisterClientConnectionId(client_connection_id);
+      compression_engine_.UnregisterClientConnectionId(client_connection_id);
+    }
+    return;
   }
+
+  for (auto it = connect_udp_client_states_.begin();
+       it != connect_udp_client_states_.end();) {
+    if (it->encapsulated_client_session() == encapsulated_client_session) {
+      QUIC_DLOG(INFO) << "Removing state for flow_id " << it->flow_id();
+      auto* stream = it->stream();
+      it = connect_udp_client_states_.erase(it);
+      if (!stream->write_side_closed()) {
+        stream->Reset(QUIC_STREAM_CANCELLED);
+      }
+    } else {
+      ++it;
+    }
+  }
+}
+
+void MasqueClientSession::OnConnectionClosed(
+    const QuicConnectionCloseFrame& frame,
+    ConnectionCloseSource source) {
+  QuicSpdyClientSession::OnConnectionClosed(frame, source);
+  // Close all encapsulated sessions.
+  for (auto client_state : connect_udp_client_states_) {
+    client_state.encapsulated_client_session()->CloseConnection(
+        QUIC_CONNECTION_CANCELLED, "Underlying MASQUE connection was closed",
+        ConnectionCloseBehavior::SILENT_CLOSE);
+  }
+}
+
+void MasqueClientSession::OnStreamClosed(QuicStreamId stream_id) {
+  for (auto it = connect_udp_client_states_.begin();
+       it != connect_udp_client_states_.end();) {
+    if (it->stream()->id() == stream_id) {
+      QUIC_DLOG(INFO) << "Stream " << stream_id
+                      << " was closed, removing state for flow_id "
+                      << it->flow_id();
+      auto* encapsulated_client_session = it->encapsulated_client_session();
+      it = connect_udp_client_states_.erase(it);
+      encapsulated_client_session->CloseConnection(
+          QUIC_CONNECTION_CANCELLED,
+          "Underlying MASQUE CONNECT-UDP stream was closed",
+          ConnectionCloseBehavior::SILENT_CLOSE);
+    } else {
+      ++it;
+    }
+  }
+
+  QuicSpdyClientSession::OnStreamClosed(stream_id);
 }
 
 }  // namespace quic
diff --git a/quic/masque/masque_client_session.h b/quic/masque/masque_client_session.h
index 0eab5b2..f849c61 100644
--- a/quic/masque/masque_client_session.h
+++ b/quic/masque/masque_client_session.h
@@ -9,6 +9,7 @@
 #include "absl/strings/string_view.h"
 #include "quic/core/http/quic_spdy_client_session.h"
 #include "quic/masque/masque_compression_engine.h"
+#include "quic/masque/masque_utils.h"
 #include "quic/platform/api/quic_export.h"
 #include "quic/platform/api/quic_socket_address.h"
 
@@ -39,14 +40,21 @@
 
     // Process packet that was just decapsulated.
     virtual void ProcessPacket(absl::string_view packet,
-                               QuicSocketAddress server_address) = 0;
+                               QuicSocketAddress target_server_address) = 0;
+
+    // Close the encapsulated connection.
+    virtual void CloseConnection(
+        QuicErrorCode error,
+        const std::string& details,
+        ConnectionCloseBehavior connection_close_behavior) = 0;
   };
 
   // Takes ownership of |connection|, but not of |crypto_config| or
   // |push_promise_index| or |owner|. All pointers must be non-null. Caller
   // must ensure that |push_promise_index| and |owner| stay valid for the
   // lifetime of the newly created MasqueClientSession.
-  MasqueClientSession(const QuicConfig& config,
+  MasqueClientSession(MasqueMode masque_mode,
+                      const QuicConfig& config,
                       const ParsedQuicVersionVector& supported_versions,
                       QuicConnection* connection,
                       const QuicServerId& server_id,
@@ -60,17 +68,19 @@
 
   // From QuicSession.
   void OnMessageReceived(absl::string_view message) override;
-
   void OnMessageAcked(QuicMessageId message_id,
                       QuicTime receive_timestamp) override;
-
   void OnMessageLost(QuicMessageId message_id) override;
+  void OnConnectionClosed(const QuicConnectionCloseFrame& frame,
+                          ConnectionCloseSource source) override;
+  void OnStreamClosed(QuicStreamId stream_id) override;
 
   // Send encapsulated packet.
   void SendPacket(QuicConnectionId client_connection_id,
                   QuicConnectionId server_connection_id,
                   absl::string_view packet,
-                  const QuicSocketAddress& server_address);
+                  const QuicSocketAddress& target_server_address,
+                  EncapsulatedClientSession* encapsulated_client_session);
 
   // Register encapsulated client. This allows clients that are encapsulated
   // within this MASQUE session to indicate they own a given client connection
@@ -84,9 +94,48 @@
 
   // Unregister encapsulated client. |client_connection_id| must match a
   // value previously passed to RegisterConnectionId.
-  void UnregisterConnectionId(QuicConnectionId client_connection_id);
+  void UnregisterConnectionId(
+      QuicConnectionId client_connection_id,
+      EncapsulatedClientSession* encapsulated_client_session);
 
  private:
+  // State that the MasqueClientSession keeps for each CONNECT-UDP request.
+  class QUIC_NO_EXPORT ConnectUdpClientState {
+   public:
+    // |stream| and |encapsulated_client_session| must be valid for the lifetime
+    // of the ConnectUdpClientState.
+    explicit ConnectUdpClientState(
+        QuicSpdyClientStream* stream,
+        EncapsulatedClientSession* encapsulated_client_session,
+        QuicDatagramFlowId flow_id,
+        const QuicSocketAddress& target_server_address)
+        : stream_(stream),
+          encapsulated_client_session_(encapsulated_client_session),
+          flow_id_(flow_id),
+          target_server_address_(target_server_address) {}
+
+    QuicSpdyClientStream* stream() const { return stream_; }
+    EncapsulatedClientSession* encapsulated_client_session() const {
+      return encapsulated_client_session_;
+    }
+    QuicDatagramFlowId flow_id() const { return flow_id_; }
+    const QuicSocketAddress& target_server_address() const {
+      return target_server_address_;
+    }
+
+   private:
+    QuicSpdyClientStream* stream_;                            // Unowned.
+    EncapsulatedClientSession* encapsulated_client_session_;  // Unowned.
+    QuicDatagramFlowId flow_id_;
+    QuicSocketAddress target_server_address_;
+  };
+
+  const ConnectUdpClientState* GetOrCreateConnectUdpClientState(
+      const QuicSocketAddress& target_server_address,
+      EncapsulatedClientSession* encapsulated_client_session);
+
+  MasqueMode masque_mode_;
+  std::list<ConnectUdpClientState> connect_udp_client_states_;
   absl::flat_hash_map<QuicConnectionId,
                       EncapsulatedClientSession*,
                       QuicConnectionIdHash>
diff --git a/quic/masque/masque_compression_engine.h b/quic/masque/masque_compression_engine.h
index 96c5948..16b4549 100644
--- a/quic/masque/masque_compression_engine.h
+++ b/quic/masque/masque_compression_engine.h
@@ -70,6 +70,9 @@
   // compression table.
   void UnregisterClientConnectionId(QuicConnectionId client_connection_id);
 
+  // Generates a new datagram flow ID.
+  QuicDatagramFlowId GetNextFlowId();
+
  private:
   struct QUIC_NO_EXPORT MasqueCompressionContext {
     QuicConnectionId client_connection_id;
@@ -78,9 +81,6 @@
     bool validated = false;
   };
 
-  // Generates a new datagram flow ID.
-  QuicDatagramFlowId GetNextFlowId();
-
   // Finds or creates a new compression context to use during compression.
   // |client_connection_id_present| and |server_connection_id_present| indicate
   // whether the corresponding connection ID is present in the current packet.
diff --git a/quic/masque/masque_dispatcher.cc b/quic/masque/masque_dispatcher.cc
index 6edca62..03eb8f8 100644
--- a/quic/masque/masque_dispatcher.cc
+++ b/quic/masque/masque_dispatcher.cc
@@ -8,9 +8,11 @@
 namespace quic {
 
 MasqueDispatcher::MasqueDispatcher(
+    MasqueMode masque_mode,
     const QuicConfig* config,
     const QuicCryptoServerConfig* crypto_config,
     QuicVersionManager* version_manager,
+    QuicEpollServer* epoll_server,
     std::unique_ptr<QuicConnectionHelperInterface> helper,
     std::unique_ptr<QuicCryptoServerStreamBase::Helper> session_helper,
     std::unique_ptr<QuicAlarmFactory> alarm_factory,
@@ -24,6 +26,8 @@
                            std::move(alarm_factory),
                            masque_server_backend,
                            expected_server_connection_id_length),
+      masque_mode_(masque_mode),
+      epoll_server_(epoll_server),
       masque_server_backend_(masque_server_backend) {}
 
 std::unique_ptr<QuicSession> MasqueDispatcher::CreateQuicSession(
@@ -40,9 +44,9 @@
                          ParsedQuicVersionVector{version});
 
   auto session = std::make_unique<MasqueServerSession>(
-      config(), GetSupportedVersions(), connection, this, this,
-      session_helper(), crypto_config(), compressed_certs_cache(),
-      masque_server_backend_);
+      masque_mode_, config(), GetSupportedVersions(), connection, this, this,
+      epoll_server_, session_helper(), crypto_config(),
+      compressed_certs_cache(), masque_server_backend_);
   session->Initialize();
   return session;
 }
diff --git a/quic/masque/masque_dispatcher.h b/quic/masque/masque_dispatcher.h
index 2e0186c..ce371b5 100644
--- a/quic/masque/masque_dispatcher.h
+++ b/quic/masque/masque_dispatcher.h
@@ -8,6 +8,8 @@
 #include "absl/container/flat_hash_map.h"
 #include "quic/masque/masque_server_backend.h"
 #include "quic/masque/masque_server_session.h"
+#include "quic/masque/masque_utils.h"
+#include "quic/platform/api/quic_epoll.h"
 #include "quic/platform/api/quic_export.h"
 #include "quic/tools/quic_simple_dispatcher.h"
 
@@ -19,9 +21,11 @@
                                         public MasqueServerSession::Visitor {
  public:
   explicit MasqueDispatcher(
+      MasqueMode masque_mode,
       const QuicConfig* config,
       const QuicCryptoServerConfig* crypto_config,
       QuicVersionManager* version_manager,
+      QuicEpollServer* epoll_server,
       std::unique_ptr<QuicConnectionHelperInterface> helper,
       std::unique_ptr<QuicCryptoServerStreamBase::Helper> session_helper,
       std::unique_ptr<QuicAlarmFactory> alarm_factory,
@@ -51,6 +55,8 @@
       QuicConnectionId client_connection_id) override;
 
  private:
+  MasqueMode masque_mode_;
+  QuicEpollServer* epoll_server_;               // Unowned.
   MasqueServerBackend* masque_server_backend_;  // Unowned.
   // Mapping from client connection IDs to server sessions, allows routing
   // incoming packets to the right MASQUE connection.
diff --git a/quic/masque/masque_encapsulated_client_session.cc b/quic/masque/masque_encapsulated_client_session.cc
index 983a094..3d557c3 100644
--- a/quic/masque/masque_encapsulated_client_session.cc
+++ b/quic/masque/masque_encapsulated_client_session.cc
@@ -31,11 +31,19 @@
                                  received_packet);
 }
 
+void MasqueEncapsulatedClientSession::CloseConnection(
+    QuicErrorCode error,
+    const std::string& details,
+    ConnectionCloseBehavior connection_close_behavior) {
+  connection()->CloseConnection(error, details, connection_close_behavior);
+}
+
 void MasqueEncapsulatedClientSession::OnConnectionClosed(
-    const QuicConnectionCloseFrame& /*frame*/,
-    ConnectionCloseSource /*source*/) {
+    const QuicConnectionCloseFrame& frame,
+    ConnectionCloseSource source) {
+  QuicSpdyClientSession::OnConnectionClosed(frame, source);
   masque_client_session_->UnregisterConnectionId(
-      connection()->client_connection_id());
+      connection()->client_connection_id(), this);
 }
 
 }  // namespace quic
diff --git a/quic/masque/masque_encapsulated_client_session.h b/quic/masque/masque_encapsulated_client_session.h
index 570ccdb..0952159 100644
--- a/quic/masque/masque_encapsulated_client_session.h
+++ b/quic/masque/masque_encapsulated_client_session.h
@@ -44,6 +44,10 @@
   // From MasqueClientSession::EncapsulatedClientSession.
   void ProcessPacket(absl::string_view packet,
                      QuicSocketAddress server_address) override;
+  void CloseConnection(
+      QuicErrorCode error,
+      const std::string& details,
+      ConnectionCloseBehavior connection_close_behavior) override;
 
   // From QuicSession.
   void OnConnectionClosed(const QuicConnectionCloseFrame& frame,
diff --git a/quic/masque/masque_encapsulated_epoll_client.cc b/quic/masque/masque_encapsulated_epoll_client.cc
index 8fb938d..8891e4a 100644
--- a/quic/masque/masque_encapsulated_epoll_client.cc
+++ b/quic/masque/masque_encapsulated_epoll_client.cc
@@ -30,8 +30,8 @@
     absl::string_view packet(buffer, buf_len);
     client_->masque_client()->masque_client_session()->SendPacket(
         client_->session()->connection()->client_connection_id(),
-        client_->session()->connection()->connection_id(), packet,
-        peer_address);
+        client_->session()->connection()->connection_id(), packet, peer_address,
+        client_->masque_encapsulated_client_session());
     return WriteResult(WRITE_STATUS_OK, buf_len);
   }
 
@@ -94,7 +94,7 @@
 
 MasqueEncapsulatedEpollClient::~MasqueEncapsulatedEpollClient() {
   masque_client_->masque_client_session()->UnregisterConnectionId(
-      client_connection_id_);
+      client_connection_id_, masque_encapsulated_client_session());
 }
 
 std::unique_ptr<QuicSession>
diff --git a/quic/masque/masque_epoll_client.cc b/quic/masque/masque_epoll_client.cc
index 1d77e55..bf09da4 100644
--- a/quic/masque/masque_epoll_client.cc
+++ b/quic/masque/masque_epoll_client.cc
@@ -12,6 +12,7 @@
 MasqueEpollClient::MasqueEpollClient(
     QuicSocketAddress server_address,
     const QuicServerId& server_id,
+    MasqueMode masque_mode,
     QuicEpollServer* epoll_server,
     std::unique_ptr<ProofVerifier> proof_verifier,
     const std::string& authority)
@@ -20,6 +21,7 @@
                  MasqueSupportedVersions(),
                  epoll_server,
                  std::move(proof_verifier)),
+      masque_mode_(masque_mode),
       authority_(authority) {}
 
 std::unique_ptr<QuicSession> MasqueEpollClient::CreateQuicClientSession(
@@ -28,8 +30,8 @@
   QUIC_DLOG(INFO) << "Creating MASQUE session for "
                   << connection->connection_id();
   return std::make_unique<MasqueClientSession>(
-      *config(), supported_versions, connection, server_id(), crypto_config(),
-      push_promise_index(), this);
+      masque_mode_, *config(), supported_versions, connection, server_id(),
+      crypto_config(), push_promise_index(), this);
 }
 
 MasqueClientSession* MasqueEpollClient::masque_client_session() {
@@ -44,6 +46,7 @@
 std::unique_ptr<MasqueEpollClient> MasqueEpollClient::Create(
     const std::string& host,
     int port,
+    MasqueMode masque_mode,
     QuicEpollServer* epoll_server,
     std::unique_ptr<ProofVerifier> proof_verifier) {
   // Build the masque_client, and try to connect.
@@ -57,7 +60,7 @@
   // std::make_unique<MasqueEpollClient>(...) because the constructor for
   // MasqueEpollClient is private and therefore not accessible from make_unique.
   auto masque_client = QuicWrapUnique(new MasqueEpollClient(
-      addr, server_id, epoll_server, std::move(proof_verifier),
+      addr, server_id, masque_mode, epoll_server, std::move(proof_verifier),
       absl::StrCat(host, ":", port)));
 
   if (masque_client == nullptr) {
@@ -78,33 +81,35 @@
     return nullptr;
   }
 
-  std::string body = "foo";
+  if (masque_client->masque_mode() == MasqueMode::kLegacy) {
+    // Construct the legacy mode init request.
+    spdy::Http2HeaderBlock header_block;
+    header_block[":method"] = "POST";
+    header_block[":scheme"] = "https";
+    header_block[":authority"] = masque_client->authority_;
+    header_block[":path"] = "/.well-known/masque/init";
+    std::string body = "foo";
 
-  // Construct a GET or POST request for supplied URL.
-  spdy::Http2HeaderBlock header_block;
-  header_block[":method"] = "POST";
-  header_block[":scheme"] = "https";
-  header_block[":authority"] = masque_client->authority_;
-  header_block[":path"] = "/.well-known/masque/init";
+    // Make sure to store the response, for later output.
+    masque_client->set_store_response(true);
 
-  // Make sure to store the response, for later output.
-  masque_client->set_store_response(true);
+    // Send the MASQUE init command.
+    masque_client->SendRequestAndWaitForResponse(header_block, body,
+                                                 /*fin=*/true);
 
-  // Send the MASQUE init command.
-  masque_client->SendRequestAndWaitForResponse(header_block, body,
-                                               /*fin=*/true);
+    if (!masque_client->connected()) {
+      QUIC_LOG(ERROR)
+          << "MASQUE init request caused connection failure. Error: "
+          << QuicErrorCodeToString(masque_client->session()->error());
+      return nullptr;
+    }
 
-  if (!masque_client->connected()) {
-    QUIC_LOG(ERROR) << "MASQUE init request caused connection failure. Error: "
-                    << QuicErrorCodeToString(masque_client->session()->error());
-    return nullptr;
-  }
-
-  const int response_code = masque_client->latest_response_code();
-  if (response_code != 200) {
-    QUIC_LOG(ERROR) << "MASQUE init request failed with HTTP response code "
-                    << response_code;
-    return nullptr;
+    const int response_code = masque_client->latest_response_code();
+    if (response_code != 200) {
+      QUIC_LOG(ERROR) << "MASQUE init request failed with HTTP response code "
+                      << response_code;
+      return nullptr;
+    }
   }
   return masque_client;
 }
diff --git a/quic/masque/masque_epoll_client.h b/quic/masque/masque_epoll_client.h
index 345c18c..1db161f 100644
--- a/quic/masque/masque_epoll_client.h
+++ b/quic/masque/masque_epoll_client.h
@@ -6,6 +6,7 @@
 #define QUICHE_QUIC_MASQUE_MASQUE_EPOLL_CLIENT_H_
 
 #include "quic/masque/masque_client_session.h"
+#include "quic/masque/masque_utils.h"
 #include "quic/platform/api/quic_export.h"
 #include "quic/tools/quic_client.h"
 
@@ -19,6 +20,7 @@
   static std::unique_ptr<MasqueEpollClient> Create(
       const std::string& host,
       int port,
+      MasqueMode masque_mode,
       QuicEpollServer* epoll_server,
       std::unique_ptr<ProofVerifier> proof_verifier);
 
@@ -37,10 +39,13 @@
   void UnregisterClientConnectionId(
       QuicConnectionId client_connection_id) override;
 
+  MasqueMode masque_mode() const { return masque_mode_; }
+
  private:
   // Constructor is private, use Create() instead.
   MasqueEpollClient(QuicSocketAddress server_address,
                     const QuicServerId& server_id,
+                    MasqueMode masque_mode,
                     QuicEpollServer* epoll_server,
                     std::unique_ptr<ProofVerifier> proof_verifier,
                     const std::string& authority);
@@ -49,6 +54,7 @@
   MasqueEpollClient(const MasqueEpollClient&) = delete;
   MasqueEpollClient& operator=(const MasqueEpollClient&) = delete;
 
+  MasqueMode masque_mode_;
   std::string authority_;
 };
 
diff --git a/quic/masque/masque_epoll_server.cc b/quic/masque/masque_epoll_server.cc
index 34769a3..e8a0850 100644
--- a/quic/masque/masque_epoll_server.cc
+++ b/quic/masque/masque_epoll_server.cc
@@ -11,15 +11,18 @@
 
 namespace quic {
 
-MasqueEpollServer::MasqueEpollServer(MasqueServerBackend* masque_server_backend)
+MasqueEpollServer::MasqueEpollServer(MasqueMode masque_mode,
+                                     MasqueServerBackend* masque_server_backend)
     : QuicServer(CreateDefaultProofSource(),
                  masque_server_backend,
                  MasqueSupportedVersions()),
+      masque_mode_(masque_mode),
       masque_server_backend_(masque_server_backend) {}
 
 QuicDispatcher* MasqueEpollServer::CreateQuicDispatcher() {
   return new MasqueDispatcher(
-      &config(), &crypto_config(), version_manager(),
+      masque_mode_, &config(), &crypto_config(), version_manager(),
+      epoll_server(),
       std::make_unique<QuicEpollConnectionHelper>(epoll_server(),
                                                   QuicAllocator::BUFFER_POOL),
       std::make_unique<QuicSimpleCryptoServerStreamHelper>(),
diff --git a/quic/masque/masque_epoll_server.h b/quic/masque/masque_epoll_server.h
index 57eab6a..88161ee 100644
--- a/quic/masque/masque_epoll_server.h
+++ b/quic/masque/masque_epoll_server.h
@@ -6,6 +6,7 @@
 #define QUICHE_QUIC_MASQUE_MASQUE_EPOLL_SERVER_H_
 
 #include "quic/masque/masque_server_backend.h"
+#include "quic/masque/masque_utils.h"
 #include "quic/platform/api/quic_export.h"
 #include "quic/tools/quic_server.h"
 
@@ -14,7 +15,8 @@
 // QUIC server that implements MASQUE.
 class QUIC_NO_EXPORT MasqueEpollServer : public QuicServer {
  public:
-  explicit MasqueEpollServer(MasqueServerBackend* masque_server_backend);
+  explicit MasqueEpollServer(MasqueMode masque_mode,
+                             MasqueServerBackend* masque_server_backend);
 
   // Disallow copy and assign.
   MasqueEpollServer(const MasqueEpollServer&) = delete;
@@ -24,6 +26,7 @@
   QuicDispatcher* CreateQuicDispatcher() override;
 
  private:
+  MasqueMode masque_mode_;
   MasqueServerBackend* masque_server_backend_;  // Unowned.
 };
 
diff --git a/quic/masque/masque_server_backend.cc b/quic/masque/masque_server_backend.cc
index 1994819..230f7e7 100644
--- a/quic/masque/masque_server_backend.cc
+++ b/quic/masque/masque_server_backend.cc
@@ -19,9 +19,10 @@
 
 }  // namespace
 
-MasqueServerBackend::MasqueServerBackend(const std::string& server_authority,
+MasqueServerBackend::MasqueServerBackend(MasqueMode masque_mode,
+                                         const std::string& server_authority,
                                          const std::string& cache_directory)
-    : server_authority_(server_authority) {
+    : masque_mode_(masque_mode), server_authority_(server_authority) {
   if (!cache_directory.empty()) {
     QuicMemoryCacheBackend::InitializeBackend(cache_directory);
   }
@@ -43,16 +44,24 @@
   absl::string_view path = path_pair->second;
   absl::string_view scheme = scheme_pair->second;
   absl::string_view method = method_pair->second;
-  if (scheme != "https" || method != "POST" || request_body.empty()) {
-    // MASQUE requests MUST be a non-empty https POST.
-    return false;
-  }
+  std::string masque_path = "";
+  if (masque_mode_ == MasqueMode::kLegacy) {
+    if (scheme != "https" || method != "POST" || request_body.empty()) {
+      // MASQUE requests MUST be a non-empty https POST.
+      return false;
+    }
 
-  if (path.rfind("/.well-known/masque/", 0) != 0) {
-    // This request is not a MASQUE path.
-    return false;
+    if (path.rfind("/.well-known/masque/", 0) != 0) {
+      // This request is not a MASQUE path.
+      return false;
+    }
+    masque_path = path.substr(sizeof("/.well-known/masque/") - 1);
+  } else {
+    if (method != "CONNECT-UDP") {
+      // Unexpected method.
+      return false;
+    }
   }
-  std::string masque_path(path.substr(sizeof("/.well-known/masque/") - 1));
 
   if (!server_authority_.empty()) {
     auto authority_pair = request_headers.find(":authority");
diff --git a/quic/masque/masque_server_backend.h b/quic/masque/masque_server_backend.h
index 1809359..86b1e00 100644
--- a/quic/masque/masque_server_backend.h
+++ b/quic/masque/masque_server_backend.h
@@ -6,6 +6,7 @@
 #define QUICHE_QUIC_MASQUE_MASQUE_SERVER_BACKEND_H_
 
 #include "absl/container/flat_hash_map.h"
+#include "quic/masque/masque_utils.h"
 #include "quic/platform/api/quic_export.h"
 #include "quic/tools/quic_memory_cache_backend.h"
 
@@ -27,7 +28,8 @@
     virtual ~BackendClient() = default;
   };
 
-  explicit MasqueServerBackend(const std::string& server_authority,
+  explicit MasqueServerBackend(MasqueMode masque_mode,
+                               const std::string& server_authority,
                                const std::string& cache_directory);
 
   // Disallow copy and assign.
@@ -57,6 +59,7 @@
       const std::string& request_body,
       QuicSimpleServerBackend::RequestHandler* request_handler);
 
+  MasqueMode masque_mode_;
   std::string server_authority_;
   absl::flat_hash_map<std::string, std::unique_ptr<QuicBackendResponse>>
       active_response_map_;
diff --git a/quic/masque/masque_server_bin.cc b/quic/masque/masque_server_bin.cc
index b25bf83..a6b6225 100644
--- a/quic/masque/masque_server_bin.cc
+++ b/quic/masque/masque_server_bin.cc
@@ -35,6 +35,12 @@
     "Specifies the authority over which the server will accept MASQUE "
     "requests. Defaults to empty which allows all authorities.");
 
+DEFINE_QUIC_COMMAND_LINE_FLAG(std::string,
+                              masque_mode,
+                              "",
+                              "Allows setting MASQUE mode, valid values are "
+                              "open and legacy. Defaults to open.");
+
 int main(int argc, char* argv[]) {
   const char* usage = "Usage: masque_server [options]";
   std::vector<std::string> non_option_args =
@@ -44,17 +50,28 @@
     return 0;
   }
 
-  auto backend = std::make_unique<quic::MasqueServerBackend>(
-      GetQuicFlag(FLAGS_server_authority), GetQuicFlag(FLAGS_cache_dir));
+  quic::MasqueMode masque_mode = quic::MasqueMode::kOpen;
+  std::string mode_string = GetQuicFlag(FLAGS_masque_mode);
+  if (mode_string == "legacy") {
+    masque_mode = quic::MasqueMode::kLegacy;
+  } else if (!mode_string.empty() && mode_string != "open") {
+    std::cerr << "Invalid masque_mode \"" << mode_string << "\"" << std::endl;
+    return 1;
+  }
 
-  auto server = std::make_unique<quic::MasqueEpollServer>(backend.get());
+  auto backend = std::make_unique<quic::MasqueServerBackend>(
+      masque_mode, GetQuicFlag(FLAGS_server_authority),
+      GetQuicFlag(FLAGS_cache_dir));
+
+  auto server =
+      std::make_unique<quic::MasqueEpollServer>(masque_mode, backend.get());
 
   if (!server->CreateUDPSocketAndListen(quic::QuicSocketAddress(
           quic::QuicIpAddress::Any6(), GetQuicFlag(FLAGS_port)))) {
     return 1;
   }
 
-  std::cerr << "Started MASQUE server" << std::endl;
+  std::cerr << "Started " << masque_mode << " MASQUE server" << std::endl;
   server->HandleEventsForever();
   return 0;
 }
diff --git a/quic/masque/masque_server_session.cc b/quic/masque/masque_server_session.cc
index 228444a..de231ba 100644
--- a/quic/masque/masque_server_session.cc
+++ b/quic/masque/masque_server_session.cc
@@ -4,14 +4,79 @@
 
 #include "quic/masque/masque_server_session.h"
 
+#include <netdb.h>
+
+#include "absl/strings/str_cat.h"
+#include "quic/core/quic_data_reader.h"
+#include "quic/core/quic_udp_socket.h"
+#include "quic/tools/quic_url.h"
+#include "common/platform/api/quiche_text_utils.h"
+
 namespace quic {
 
+namespace {
+// RAII wrapper for QuicUdpSocketFd.
+class FdWrapper {
+ public:
+  // Takes ownership of |fd| and closes the file descriptor on destruction.
+  explicit FdWrapper(int address_family) {
+    QuicUdpSocketApi socket_api;
+    fd_ =
+        socket_api.Create(address_family,
+                          /*receive_buffer_size =*/kDefaultSocketReceiveBuffer,
+                          /*send_buffer_size =*/kDefaultSocketReceiveBuffer);
+  }
+
+  ~FdWrapper() {
+    if (fd_ == kQuicInvalidSocketFd) {
+      return;
+    }
+    QuicUdpSocketApi socket_api;
+    socket_api.Destroy(fd_);
+  }
+
+  // Hands ownership of the file descriptor to the caller.
+  QuicUdpSocketFd extract_fd() {
+    QuicUdpSocketFd fd = fd_;
+    fd_ = kQuicInvalidSocketFd;
+    return fd;
+  }
+
+  // Keeps ownership of the file descriptor.
+  QuicUdpSocketFd fd() { return fd_; }
+
+  // Disallow copy and move.
+  FdWrapper(const FdWrapper&) = delete;
+  FdWrapper(FdWrapper&&) = delete;
+  FdWrapper& operator=(const FdWrapper&) = delete;
+  FdWrapper& operator=(FdWrapper&&) = delete;
+
+ private:
+  QuicUdpSocketFd fd_;
+};
+
+std::unique_ptr<QuicBackendResponse> CreateBackendErrorResponse(
+    absl::string_view status,
+    absl::string_view body) {
+  spdy::Http2HeaderBlock response_headers;
+  response_headers[":status"] = status;
+  auto response = std::make_unique<QuicBackendResponse>();
+  response->set_response_type(QuicBackendResponse::REGULAR_RESPONSE);
+  response->set_headers(std::move(response_headers));
+  response->set_body(body);
+  return response;
+}
+
+}  // namespace
+
 MasqueServerSession::MasqueServerSession(
+    MasqueMode masque_mode,
     const QuicConfig& config,
     const ParsedQuicVersionVector& supported_versions,
     QuicConnection* connection,
     QuicSession::Visitor* visitor,
     Visitor* owner,
+    QuicEpollServer* epoll_server,
     QuicCryptoServerStreamBase::Helper* helper,
     const QuicCryptoServerConfig* crypto_config,
     QuicCompressedCertsCache* compressed_certs_cache,
@@ -26,41 +91,74 @@
                               masque_server_backend),
       masque_server_backend_(masque_server_backend),
       owner_(owner),
-      compression_engine_(this) {
+      epoll_server_(epoll_server),
+      compression_engine_(this),
+      masque_mode_(masque_mode) {
   masque_server_backend_->RegisterBackendClient(connection_id(), this);
 }
 
 void MasqueServerSession::OnMessageReceived(absl::string_view message) {
   QUIC_DVLOG(1) << "Received DATAGRAM frame of length " << message.length();
+  if (masque_mode_ == MasqueMode::kLegacy) {
+    QuicConnectionId client_connection_id, server_connection_id;
+    QuicSocketAddress target_server_address;
+    std::vector<char> packet;
+    bool version_present;
+    if (!compression_engine_.DecompressDatagram(
+            message, &client_connection_id, &server_connection_id,
+            &target_server_address, &packet, &version_present)) {
+      return;
+    }
 
-  QuicConnectionId client_connection_id, server_connection_id;
-  QuicSocketAddress server_address;
-  std::vector<char> packet;
-  bool version_present;
-  if (!compression_engine_.DecompressDatagram(
-          message, &client_connection_id, &server_connection_id,
-          &server_address, &packet, &version_present)) {
+    QUIC_DVLOG(1) << "Received packet of length " << packet.size() << " for "
+                  << target_server_address << " client "
+                  << client_connection_id;
+
+    if (version_present) {
+      if (client_connection_id.length() != kQuicDefaultConnectionIdLength) {
+        QUIC_DLOG(ERROR)
+            << "Dropping long header with invalid client_connection_id "
+            << client_connection_id;
+        return;
+      }
+      owner_->RegisterClientConnectionId(client_connection_id, this);
+    }
+
+    WriteResult write_result = connection()->writer()->WritePacket(
+        packet.data(), packet.size(), connection()->self_address().host(),
+        target_server_address, nullptr);
+    QUIC_DVLOG(1) << "Got " << write_result << " for " << packet.size()
+                  << " bytes to " << target_server_address;
+    return;
+  }
+  QUICHE_DCHECK_EQ(masque_mode_, MasqueMode::kOpen);
+  QuicDataReader reader(message);
+  QuicDatagramFlowId flow_id;
+  if (!reader.ReadVarInt62(&flow_id)) {
+    QUIC_DLOG(ERROR) << "Failed to read flow_id";
     return;
   }
 
-  QUIC_DVLOG(1) << "Received packet of length " << packet.size() << " for "
-                << server_address << " client " << client_connection_id;
-
-  if (version_present) {
-    if (client_connection_id.length() != kQuicDefaultConnectionIdLength) {
-      QUIC_DLOG(ERROR)
-          << "Dropping long header with invalid client_connection_id "
-          << client_connection_id;
-      return;
-    }
-    owner_->RegisterClientConnectionId(client_connection_id, this);
+  auto it =
+      absl::c_find_if(connect_udp_server_states_,
+                      [flow_id](const ConnectUdpServerState& connect_udp) {
+                        return connect_udp.flow_id() == flow_id;
+                      });
+  if (it == connect_udp_server_states_.end()) {
+    QUIC_DLOG(ERROR) << "Received unknown flow_id " << flow_id;
+    return;
   }
-
-  WriteResult write_result = connection()->writer()->WritePacket(
-      packet.data(), packet.size(), connection()->self_address().host(),
-      server_address, nullptr);
-  QUIC_DVLOG(1) << "Got " << write_result << " for " << packet.size()
-                << " bytes to " << server_address;
+  QuicSocketAddress target_server_address = it->target_server_address();
+  QUICHE_DCHECK(target_server_address.IsInitialized());
+  QuicUdpSocketFd fd = it->fd();
+  QUICHE_DCHECK_NE(fd, kQuicInvalidSocketFd);
+  absl::string_view packet = reader.ReadRemainingPayload();
+  QuicUdpSocketApi socket_api;
+  QuicUdpPacketInfo packet_info;
+  packet_info.SetPeerAddress(target_server_address);
+  WriteResult write_result =
+      socket_api.WritePacket(fd, packet.data(), packet.length(), packet_info);
+  QUIC_DVLOG(1) << "Wrote packet to server with result " << write_result;
 }
 
 void MasqueServerSession::OnMessageAcked(QuicMessageId message_id,
@@ -73,17 +171,123 @@
 }
 
 void MasqueServerSession::OnConnectionClosed(
-    const QuicConnectionCloseFrame& /*frame*/,
-    ConnectionCloseSource /*source*/) {
+    const QuicConnectionCloseFrame& frame,
+    ConnectionCloseSource source) {
+  QuicSimpleServerSession::OnConnectionClosed(frame, source);
   QUIC_DLOG(INFO) << "Closing connection for " << connection_id();
   masque_server_backend_->RemoveBackendClient(connection_id());
+  // Clearing this state will close all sockets.
+  connect_udp_server_states_.clear();
+}
+
+void MasqueServerSession::OnStreamClosed(QuicStreamId stream_id) {
+  connect_udp_server_states_.remove_if(
+      [stream_id](const ConnectUdpServerState& connect_udp) {
+        return connect_udp.stream_id() == stream_id;
+      });
+
+  QuicSimpleServerSession::OnStreamClosed(stream_id);
 }
 
 std::unique_ptr<QuicBackendResponse> MasqueServerSession::HandleMasqueRequest(
     const std::string& masque_path,
-    const spdy::Http2HeaderBlock& /*request_headers*/,
+    const spdy::Http2HeaderBlock& request_headers,
     const std::string& request_body,
-    QuicSimpleServerBackend::RequestHandler* /*request_handler*/) {
+    QuicSimpleServerBackend::RequestHandler* request_handler) {
+  if (masque_mode_ != MasqueMode::kLegacy) {
+    auto path_pair = request_headers.find(":path");
+    auto scheme_pair = request_headers.find(":scheme");
+    auto method_pair = request_headers.find(":method");
+    auto flow_id_pair = request_headers.find("datagram-flow-id");
+    auto authority_pair = request_headers.find(":authority");
+    if (path_pair == request_headers.end() ||
+        scheme_pair == request_headers.end() ||
+        method_pair == request_headers.end() ||
+        flow_id_pair == request_headers.end() ||
+        authority_pair == request_headers.end()) {
+      QUIC_DLOG(ERROR) << "MASQUE request is missing required headers";
+      return CreateBackendErrorResponse("400", "Missing required headers");
+    }
+    absl::string_view path = path_pair->second;
+    absl::string_view scheme = scheme_pair->second;
+    absl::string_view method = method_pair->second;
+    absl::string_view flow_id_str = flow_id_pair->second;
+    absl::string_view authority = authority_pair->second;
+    if (path.empty()) {
+      QUIC_DLOG(ERROR) << "MASQUE request with empty path";
+      return CreateBackendErrorResponse("400", "Empty path");
+    }
+    if (scheme.empty()) {
+      return CreateBackendErrorResponse("400", "Empty scheme");
+      return nullptr;
+    }
+    if (method != "CONNECT-UDP") {
+      QUIC_DLOG(ERROR) << "MASQUE request with bad method \"" << method << "\"";
+      return CreateBackendErrorResponse("400", "Bad method");
+    }
+    QuicDatagramFlowId flow_id;
+    if (!absl::SimpleAtoi(flow_id_str, &flow_id)) {
+      QUIC_DLOG(ERROR) << "MASQUE request with bad flow_id \"" << flow_id_str
+                       << "\"";
+      return CreateBackendErrorResponse("400", "Bad flow ID");
+    }
+    QuicUrl url(absl::StrCat("https://", authority));
+    if (!url.IsValid() || url.PathParamsQuery() != "/") {
+      QUIC_DLOG(ERROR) << "MASQUE request with bad authority \"" << authority
+                       << "\"";
+      return CreateBackendErrorResponse("400", "Bad authority");
+    }
+
+    std::string port = absl::StrCat(url.port());
+    addrinfo hint = {};
+    hint.ai_protocol = IPPROTO_UDP;
+
+    addrinfo* info_list = nullptr;
+    int result =
+        getaddrinfo(url.host().c_str(), port.c_str(), &hint, &info_list);
+    if (result != 0) {
+      QUIC_DLOG(ERROR) << "Failed to resolve " << authority << ": "
+                       << gai_strerror(result);
+      return CreateBackendErrorResponse("500", "DNS resolution failed");
+    }
+
+    QUICHE_CHECK_NE(info_list, nullptr);
+    std::unique_ptr<addrinfo, void (*)(addrinfo*)> info_list_owned(
+        info_list, freeaddrinfo);
+    QuicSocketAddress target_server_address(info_list->ai_addr,
+                                            info_list->ai_addrlen);
+    QUIC_DLOG(INFO) << "Got CONNECT_UDP request flow_id=" << flow_id
+                    << " target_server_address=\"" << target_server_address
+                    << "\"";
+
+    FdWrapper fd_wrapper(target_server_address.host().AddressFamilyToInt());
+    if (fd_wrapper.fd() == kQuicInvalidSocketFd) {
+      QUIC_DLOG(ERROR) << "Socket creation failed";
+      return CreateBackendErrorResponse("500", "Socket creation failed");
+    }
+    QuicSocketAddress any_v6_address(QuicIpAddress::Any6(), 0);
+    QuicUdpSocketApi socket_api;
+    if (!socket_api.Bind(fd_wrapper.fd(), any_v6_address)) {
+      QUIC_DLOG(ERROR) << "Socket bind failed";
+      return CreateBackendErrorResponse("500", "Socket bind failed");
+    }
+    epoll_server_->RegisterFDForRead(fd_wrapper.fd(), this);
+
+    connect_udp_server_states_.emplace_back(ConnectUdpServerState(
+        flow_id, request_handler->stream_id(), target_server_address,
+        fd_wrapper.extract_fd(), epoll_server_));
+
+    spdy::Http2HeaderBlock response_headers;
+    response_headers[":status"] = "200";
+    response_headers["datagram-flow-id"] = absl::StrCat(flow_id);
+    auto response = std::make_unique<QuicBackendResponse>();
+    response->set_response_type(QuicBackendResponse::INCOMPLETE_RESPONSE);
+    response->set_headers(std::move(response_headers));
+    response->set_body("");
+
+    return response;
+  }
+
   QUIC_DLOG(INFO) << "MasqueServerSession handling MASQUE request";
 
   if (masque_path == "init") {
@@ -120,9 +324,166 @@
 void MasqueServerSession::HandlePacketFromServer(
     const ReceivedPacketInfo& packet_info) {
   QUIC_DVLOG(1) << "MasqueServerSession received " << packet_info;
-  compression_engine_.CompressAndSendPacket(
-      packet_info.packet.AsStringPiece(), packet_info.destination_connection_id,
-      packet_info.source_connection_id, packet_info.peer_address);
+  if (masque_mode_ == MasqueMode::kLegacy) {
+    compression_engine_.CompressAndSendPacket(
+        packet_info.packet.AsStringPiece(),
+        packet_info.destination_connection_id, packet_info.source_connection_id,
+        packet_info.peer_address);
+    return;
+  }
+  QUIC_LOG(ERROR) << "Ignoring packet from server in " << masque_mode_
+                  << " mode";
+}
+
+void MasqueServerSession::OnRegistration(QuicEpollServer* /*eps*/,
+                                         QuicUdpSocketFd fd,
+                                         int event_mask) {
+  QUIC_DVLOG(1) << "OnRegistration " << fd << " event_mask " << event_mask;
+}
+
+void MasqueServerSession::OnModification(QuicUdpSocketFd fd, int event_mask) {
+  QUIC_DVLOG(1) << "OnModification " << fd << " event_mask " << event_mask;
+}
+
+void MasqueServerSession::OnEvent(QuicUdpSocketFd fd, QuicEpollEvent* event) {
+  if ((event->in_events & EPOLLIN) == 0) {
+    QUIC_DVLOG(1) << "Ignoring OnEvent fd " << fd << " event mask "
+                  << event->in_events;
+    return;
+  }
+  auto it = absl::c_find_if(connect_udp_server_states_,
+                            [fd](const ConnectUdpServerState& connect_udp) {
+                              return connect_udp.fd() == fd;
+                            });
+  if (it == connect_udp_server_states_.end()) {
+    QUIC_BUG << "Got unexpected event mask " << event->in_events
+             << " on unknown fd " << fd;
+    return;
+  }
+  QuicDatagramFlowId flow_id = it->flow_id();
+  QuicSocketAddress expected_target_server_address =
+      it->target_server_address();
+  QUICHE_DCHECK(expected_target_server_address.IsInitialized());
+  QUIC_DVLOG(1) << "Received readable event on fd " << fd << " (mask "
+                << event->in_events << ") flow_id " << flow_id << " server "
+                << expected_target_server_address;
+  QuicUdpSocketApi socket_api;
+  BitMask64 packet_info_interested(QuicUdpPacketInfoBit::PEER_ADDRESS);
+  char packet_buffer[kMaxIncomingPacketSize];
+  char control_buffer[kDefaultUdpPacketControlBufferSize];
+  while (true) {
+    QuicUdpSocketApi::ReadPacketResult read_result;
+    read_result.packet_buffer = {packet_buffer, sizeof(packet_buffer)};
+    read_result.control_buffer = {control_buffer, sizeof(control_buffer)};
+    socket_api.ReadPacket(fd, packet_info_interested, &read_result);
+    if (!read_result.ok) {
+      // Most likely there is nothing left to read, break out of read loop.
+      break;
+    }
+    if (!read_result.packet_info.HasValue(QuicUdpPacketInfoBit::PEER_ADDRESS)) {
+      QUIC_BUG << "Missing peer address when reading from fd " << fd;
+      continue;
+    }
+    if (read_result.packet_info.peer_address() !=
+        expected_target_server_address) {
+      QUIC_DLOG(ERROR) << "Ignoring UDP packet on fd " << fd
+                       << " from unexpected server address "
+                       << read_result.packet_info.peer_address()
+                       << " (expected " << expected_target_server_address
+                       << ")";
+      continue;
+    }
+    if (!connection()->connected()) {
+      QUIC_BUG << "Unexpected incoming UDP packet on fd " << fd << " from "
+               << expected_target_server_address
+               << " because MASQUE connection is closed";
+      return;
+    }
+    // The packet is valid, send it to the client in a DATAGRAM frame.
+    size_t slice_length = QuicDataWriter::GetVarInt62Len(flow_id) +
+                          read_result.packet_buffer.buffer_len;
+    QuicUniqueBufferPtr buffer = MakeUniqueBuffer(
+        connection()->helper()->GetStreamSendBufferAllocator(), slice_length);
+    QuicDataWriter writer(slice_length, buffer.get());
+    if (!writer.WriteVarInt62(flow_id)) {
+      QUIC_BUG << "Failed to write flow_id";
+      continue;
+    }
+    if (!writer.WriteBytes(read_result.packet_buffer.buffer,
+                           read_result.packet_buffer.buffer_len)) {
+      QUIC_BUG << "Failed to write packet";
+      continue;
+    }
+    QUICHE_DCHECK_EQ(writer.remaining(), 0u);
+    QuicMemSlice slice(std::move(buffer), slice_length);
+    MessageResult message_result = SendMessage(QuicMemSliceSpan(&slice));
+    QUIC_DVLOG(1) << "Sent UDP packet from target server of length "
+                  << read_result.packet_buffer.buffer_len << " with flow ID "
+                  << flow_id << " and got message result " << message_result;
+  }
+}
+
+void MasqueServerSession::OnUnregistration(QuicUdpSocketFd fd, bool replaced) {
+  QUIC_DVLOG(1) << "OnUnregistration " << fd << " " << (replaced ? "" : "!")
+                << " replaced";
+}
+
+void MasqueServerSession::OnShutdown(QuicEpollServer* /*eps*/,
+                                     QuicUdpSocketFd fd) {
+  QUIC_DVLOG(1) << "OnShutdown " << fd;
+}
+
+std::string MasqueServerSession::Name() const {
+  return std::string("MasqueServerSession-") + connection_id().ToString();
+}
+
+MasqueServerSession::ConnectUdpServerState::ConnectUdpServerState(
+    QuicDatagramFlowId flow_id,
+    QuicStreamId stream_id,
+    const QuicSocketAddress& target_server_address,
+    QuicUdpSocketFd fd,
+    QuicEpollServer* epoll_server)
+    : flow_id_(flow_id),
+      stream_id_(stream_id),
+      target_server_address_(target_server_address),
+      fd_(fd),
+      epoll_server_(epoll_server) {
+  QUICHE_DCHECK_NE(fd_, kQuicInvalidSocketFd);
+  QUICHE_DCHECK_NE(epoll_server_, nullptr);
+}
+
+MasqueServerSession::ConnectUdpServerState::~ConnectUdpServerState() {
+  if (fd_ == kQuicInvalidSocketFd) {
+    return;
+  }
+  QuicUdpSocketApi socket_api;
+  QUIC_DLOG(INFO) << "Closing fd " << fd_;
+  epoll_server_->UnregisterFD(fd_);
+  socket_api.Destroy(fd_);
+}
+
+MasqueServerSession::ConnectUdpServerState::ConnectUdpServerState(
+    MasqueServerSession::ConnectUdpServerState&& other) {
+  fd_ = kQuicInvalidSocketFd;
+  *this = std::move(other);
+}
+
+MasqueServerSession::ConnectUdpServerState&
+MasqueServerSession::ConnectUdpServerState::operator=(
+    MasqueServerSession::ConnectUdpServerState&& other) {
+  if (fd_ != kQuicInvalidSocketFd) {
+    QuicUdpSocketApi socket_api;
+    QUIC_DLOG(INFO) << "Closing fd " << fd_;
+    epoll_server_->UnregisterFD(fd_);
+    socket_api.Destroy(fd_);
+  }
+  flow_id_ = other.flow_id_;
+  stream_id_ = other.stream_id_;
+  target_server_address_ = other.target_server_address_;
+  fd_ = other.fd_;
+  epoll_server_ = other.epoll_server_;
+  other.fd_ = kQuicInvalidSocketFd;
+  return *this;
 }
 
 }  // namespace quic
diff --git a/quic/masque/masque_server_session.h b/quic/masque/masque_server_session.h
index 9e43b63..a73e641 100644
--- a/quic/masque/masque_server_session.h
+++ b/quic/masque/masque_server_session.h
@@ -5,8 +5,12 @@
 #ifndef QUICHE_QUIC_MASQUE_MASQUE_SERVER_SESSION_H_
 #define QUICHE_QUIC_MASQUE_MASQUE_SERVER_SESSION_H_
 
+#include "quic/core/quic_types.h"
+#include "quic/core/quic_udp_socket.h"
 #include "quic/masque/masque_compression_engine.h"
 #include "quic/masque/masque_server_backend.h"
+#include "quic/masque/masque_utils.h"
+#include "quic/platform/api/quic_epoll.h"
 #include "quic/platform/api/quic_export.h"
 #include "quic/tools/quic_simple_server_session.h"
 
@@ -15,7 +19,8 @@
 // QUIC server session for connection to MASQUE proxy.
 class QUIC_NO_EXPORT MasqueServerSession
     : public QuicSimpleServerSession,
-      public MasqueServerBackend::BackendClient {
+      public MasqueServerBackend::BackendClient,
+      public QuicEpollCallbackInterface {
  public:
   // Interface meant to be implemented by owner of this MasqueServerSession
   // instance.
@@ -33,11 +38,13 @@
   };
 
   explicit MasqueServerSession(
+      MasqueMode masque_mode,
       const QuicConfig& config,
       const ParsedQuicVersionVector& supported_versions,
       QuicConnection* connection,
       QuicSession::Visitor* visitor,
       Visitor* owner,
+      QuicEpollServer* epoll_server,
       QuicCryptoServerStreamBase::Helper* helper,
       const QuicCryptoServerConfig* crypto_config,
       QuicCompressedCertsCache* compressed_certs_cache,
@@ -54,6 +61,7 @@
   void OnMessageLost(QuicMessageId message_id) override;
   void OnConnectionClosed(const QuicConnectionCloseFrame& frame,
                           ConnectionCloseSource source) override;
+  void OnStreamClosed(QuicStreamId stream_id) override;
 
   // From MasqueServerBackend::BackendClient.
   std::unique_ptr<QuicBackendResponse> HandleMasqueRequest(
@@ -62,13 +70,61 @@
       const std::string& request_body,
       QuicSimpleServerBackend::RequestHandler* request_handler) override;
 
+  // From QuicEpollCallbackInterface.
+  void OnRegistration(QuicEpollServer* eps,
+                      QuicUdpSocketFd fd,
+                      int event_mask) override;
+  void OnModification(QuicUdpSocketFd fd, int event_mask) override;
+  void OnEvent(QuicUdpSocketFd fd, QuicEpollEvent* event) override;
+  void OnUnregistration(QuicUdpSocketFd fd, bool replaced) override;
+  void OnShutdown(QuicEpollServer* eps, QuicUdpSocketFd fd) override;
+  std::string Name() const override;
+
   // Handle packet for client, meant to be called by MasqueDispatcher.
   void HandlePacketFromServer(const ReceivedPacketInfo& packet_info);
 
  private:
+  // State that the MasqueServerSession keeps for each CONNECT-UDP request.
+  class QUIC_NO_EXPORT ConnectUdpServerState {
+   public:
+    // ConnectUdpServerState takes ownership of |fd|. It will unregister it
+    // from |epoll_server| and close the file descriptor when destructed.
+    explicit ConnectUdpServerState(
+        QuicDatagramFlowId flow_id,
+        QuicStreamId stream_id,
+        const QuicSocketAddress& target_server_address,
+        QuicUdpSocketFd fd,
+        QuicEpollServer* epoll_server);
+
+    ~ConnectUdpServerState();
+
+    // Disallow copy but allow move.
+    ConnectUdpServerState(const ConnectUdpServerState&) = delete;
+    ConnectUdpServerState(ConnectUdpServerState&&);
+    ConnectUdpServerState& operator=(const ConnectUdpServerState&) = delete;
+    ConnectUdpServerState& operator=(ConnectUdpServerState&&);
+
+    QuicDatagramFlowId flow_id() const { return flow_id_; }
+    QuicStreamId stream_id() const { return stream_id_; }
+    const QuicSocketAddress& target_server_address() const {
+      return target_server_address_;
+    }
+    QuicUdpSocketFd fd() const { return fd_; }
+
+   private:
+    QuicDatagramFlowId flow_id_;
+    QuicStreamId stream_id_;
+    QuicSocketAddress target_server_address_;
+    QuicUdpSocketFd fd_;             // Owned.
+    QuicEpollServer* epoll_server_;  // Unowned.
+  };
+
   MasqueServerBackend* masque_server_backend_;  // Unowned.
   Visitor* owner_;                              // Unowned.
+  QuicEpollServer* epoll_server_;               // Unowned.
   MasqueCompressionEngine compression_engine_;
+  MasqueMode masque_mode_;
+  std::list<ConnectUdpServerState> connect_udp_server_states_;
   bool masque_initialized_ = false;
 };
 
diff --git a/quic/masque/masque_utils.cc b/quic/masque/masque_utils.cc
index f4f77f5..17dec0d 100644
--- a/quic/masque/masque_utils.cc
+++ b/quic/masque/masque_utils.cc
@@ -27,4 +27,21 @@
   return config;
 }
 
+std::string MasqueModeToString(MasqueMode masque_mode) {
+  switch (masque_mode) {
+    case MasqueMode::kInvalid:
+      return "Invalid";
+    case MasqueMode::kLegacy:
+      return "Legacy";
+    case MasqueMode::kOpen:
+      return "Open";
+  }
+  return absl::StrCat("Unknown(", static_cast<int>(masque_mode), ")");
+}
+
+std::ostream& operator<<(std::ostream& os, const MasqueMode& masque_mode) {
+  os << MasqueModeToString(masque_mode);
+  return os;
+}
+
 }  // namespace quic
diff --git a/quic/masque/masque_utils.h b/quic/masque/masque_utils.h
index f151f55..8113047 100644
--- a/quic/masque/masque_utils.h
+++ b/quic/masque/masque_utils.h
@@ -18,7 +18,24 @@
 QUIC_NO_EXPORT QuicConfig MasqueEncapsulatedConfig();
 
 // Maximum packet size for encapsulated connections.
-const QuicByteCount kMasqueMaxEncapsulatedPacketSize = 1300;
+enum : QuicByteCount { kMasqueMaxEncapsulatedPacketSize = 1300 };
+
+// Mode that MASQUE is operating in.
+enum class MasqueMode : uint8_t {
+  kInvalid = 0,  // Should never be used.
+  kLegacy = 1,   // Legacy mode uses the legacy MASQUE protocol as documented in
+  // <https://tools.ietf.org/html/draft-schinazi-masque-protocol>. That version
+  // of MASQUE uses a custom application-protocol over HTTP/3, and also allows
+  // unauthenticated clients.
+  kOpen = 2,  // Open mode uses the MASQUE HTTP CONNECT-UDP method as documented
+  // in <https://tools.ietf.org/html/draft-ietf-masque-connect-udp>. This mode
+  // allows unauthenticated clients (a more restricted mode will be added to
+  // this enum at a later date).
+};
+
+QUIC_NO_EXPORT std::string MasqueModeToString(MasqueMode masque_mode);
+QUIC_NO_EXPORT std::ostream& operator<<(std::ostream& os,
+                                        const MasqueMode& masque_mode);
 
 }  // namespace quic