Use ALPN to determine what sort of connection we've accepted

Motivation:

To determine whether we're dealing with gRPC or gRPC Web (i.e. HTTP/2 or
HTTP/1) the first bytes received on a new connection are parsed. When
TLS is enabled and ALPN is supported there's no need to do this: we
will be informed about the protocol we negotiated with our peer and
therefore how to configure the channel. We also never enforced that the
protocol negotiated via ALPN was used!

Modifications:

- Add a 'GRPCServerPipelineConfigurator' handler to supersede the
  'HTTPProtocolSwitcher'. The new handler will use either ALPN and fall
  back to parsing the incoming bytes.
- The parsing behavior is slightly different to that of
  'HTTPProtocolSwitcher' in that we only read off enough bytes for the
  HTTP/2 connection preface, or enough bytes to parse an HTTP/1 request
  line (rather than reading the whole buffer as a string and processing
  that).
- Added a server configuration option to disable ALPN being required.
- Added tests.

Result:

- Server pipeline configuration is driven by ALPN, falling back to
  parsing.
- ALPN is enforced on the server (partially resolving grpc#1042)

Author: glbrntt

Sources/GRPC/GRPCServerPipelineConfigurator.swift

@@ -0,0 +1,350 @@
+/*
+ * Copyright 2020, gRPC Authors All rights reserved.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+import Logging
+import NIO
+import NIOHTTP1
+import NIOHTTP2
+import NIOTLS
+
+/// Configures a server pipeline for gRPC with the appropriate handlers depending on the HTTP
+/// version used for transport.
+///
+/// If TLS is enabled then the handler listens for an 'TLSUserEvent.handshakeCompleted' event and
+/// configures the pipeline appropriately for the protocol negotiated via ALPN. If TLS is not
+/// configured then the HTTP version is determined by parsing the inbound byte stream.
+final class GRPCServerPipelineConfigurator: ChannelInboundHandler, RemovableChannelHandler {
+  internal typealias InboundIn = ByteBuffer
+  internal typealias InboundOut = ByteBuffer
+
+  /// The server configuration.
+  private let configuration: Server.Configuration
+
+  /// Reads which we're holding on to before the pipeline is configured.
+  private var bufferedReads = CircularBuffer<NIOAny>()
+
+  /// The current state.
+  private var state: State = .notConfigured
+
+  private enum State {
+    /// The pipeline isn't configured yet.
+    case notConfigured
+    /// We're configuring the pipeline.
+    case configuring
+  }
+
+  init(configuration: Server.Configuration) {
+    self.configuration = configuration
+  }
+
+  /// Makes a gRPC Server keepalive handler.
+  private func makeKeepaliveHandler() -> GRPCServerKeepaliveHandler {
+    return .init(configuration: self.configuration.connectionKeepalive)
+  }
+
+  /// Makes a gRPC idle handler for the server..
+  private func makeIdleHandler() -> GRPCIdleHandler {
+    return .init(
+      mode: .server,
+      logger: self.configuration.logger,
+      idleTimeout: self.configuration.connectionIdleTimeout
+    )
+  }
+
+  /// Makes an HTTP/2 handler.
+  private func makeHTTP2Handler() -> NIOHTTP2Handler {
+    return .init(mode: .server)
+  }
+
+  /// Makes an HTTP/2 multiplexer suitable handling gRPC requests.
+  private func makeHTTP2Multiplexer(for channel: Channel) -> HTTP2StreamMultiplexer {
+    var logger = self.configuration.logger
+
+    return .init(
+      mode: .server,
+      channel: channel,
+      targetWindowSize: self.configuration.httpTargetWindowSize
+    ) { stream in
+      stream.getOption(HTTP2StreamChannelOptions.streamID).map { streamID -> Logger in
+        logger[metadataKey: MetadataKey.h2StreamID] = "\(streamID)"
+        return logger
+      }.recover { _ in
+        logger[metadataKey: MetadataKey.h2StreamID] = "<unknown>"
+        return logger
+      }.flatMap { logger in
+        // TODO: provide user configuration for header normalization.
+        let handler = self.makeHTTP2ToRawGRPCHandler(normalizeHeaders: true, logger: logger)
+        return stream.pipeline.addHandler(handler)
+      }
+    }
+  }
+
+  /// Makes an HTTP/2 to raw gRPC server handler.
+  private func makeHTTP2ToRawGRPCHandler(
+    normalizeHeaders: Bool,
+    logger: Logger
+  ) -> HTTP2ToRawGRPCServerCodec {
+    return HTTP2ToRawGRPCServerCodec(
+      servicesByName: self.configuration.serviceProvidersByName,
+      encoding: self.configuration.messageEncoding,
+      errorDelegate: self.configuration.errorDelegate,
+      normalizeHeaders: normalizeHeaders,
+      logger: logger
+    )
+  }
+
+  /// The pipeline finished configuring.
+  private func configurationCompleted(result: Result<Void, Error>, context: ChannelHandlerContext) {
+    switch result {
+    case .success:
+      context.pipeline.removeHandler(context: context, promise: nil)
+    case let .failure(error):
+      self.errorCaught(context: context, error: error)
+    }
+  }
+
+  /// Configures the pipeline to handle gRPC requests on an HTTP/2 connection.
+  private func configureHTTP2(context: ChannelHandlerContext) {
+    // We're now configuring the pipeline.
+    self.state = .configuring
+
+    // We could use 'Channel.configureHTTP2Pipeline', but then we'd have to find the right handlers
+    // to then insert our keepalive and idle handlers between. We can just add everything together.
+    var handlers: [ChannelHandler] = []
+    handlers.reserveCapacity(4)
+    handlers.append(self.makeHTTP2Handler())
+    handlers.append(self.makeKeepaliveHandler())
+    handlers.append(self.makeIdleHandler())
+    handlers.append(self.makeHTTP2Multiplexer(for: context.channel))
+
+    // Now configure the pipeline with the handlers.
+    context.channel.pipeline.addHandlers(handlers).whenComplete { result in
+      self.configurationCompleted(result: result, context: context)
+    }
+  }
+
+  /// Configures the pipeline to handle gRPC-Web requests on an HTTP/1 connection.
+  private func configureHTTP1(context: ChannelHandlerContext) {
+    // We're now configuring the pipeline.
+    self.state = .configuring
+
+    context.pipeline.configureHTTPServerPipeline(withErrorHandling: true).flatMap {
+      context.pipeline.addHandlers([
+        WebCORSHandler(),
+        GRPCWebToHTTP2ServerCodec(scheme: self.configuration.tls == nil ? "http" : "https"),
+        // There's no need to normalize headers for HTTP/1.
+        self.makeHTTP2ToRawGRPCHandler(normalizeHeaders: false, logger: self.configuration.logger),
+      ])
+    }.whenComplete { result in
+      self.configurationCompleted(result: result, context: context)
+    }
+  }
+
+  /// Attempts to determine the HTTP version from the buffer and then configure the pipeline
+  /// appropriately. Closes the connection if the HTTP version could not be determined.
+  private func determineHTTPVersionAndConfigurePipeline(
+    usingBuffer buffer: ByteBuffer,
+    context: ChannelHandlerContext
+  ) {
+    if HTTPVersionParser.prefixedWithHTTP2ConnectionPreface(buffer) {
+      self.configureHTTP2(context: context)
+    } else if HTTPVersionParser.prefixedWithHTTP1RequestLine(buffer) {
+      self.configureHTTP1(context: context)
+    } else {
+      self.configuration.logger.error("Unable to determine http version, closing")
+      context.close(mode: .all, promise: nil)
+    }
+  }
+
+  /// Handles a 'TLSUserEvent.handshakeCompleted' event and configures the pipeline to handle gRPC
+  /// requests.
+  private func handleHandshakeCompletedEvent(
+    _ event: TLSUserEvent,
+    context: ChannelHandlerContext
+  ) {
+    switch event {
+    case let .handshakeCompleted(negotiatedProtocol):
+      self.configuration.logger.debug("TLS handshake completed", metadata: [
+        "alpn": "\(negotiatedProtocol ?? "nil")",
+      ])
+
+      switch negotiatedProtocol {
+      case let .some(negotiated):
+        if GRPCApplicationProtocolIdentifier.isHTTP2Like(negotiated) {
+          self.configureHTTP2(context: context)
+        } else if GRPCApplicationProtocolIdentifier.isHTTP1(negotiated) {
+          self.configureHTTP1(context: context)
+        } else {
+          self.configuration.logger.warning("Unsupported ALPN identifier '\(negotiated)', closing")
+          context.close(mode: .all, promise: nil)
+        }
+
+      case .none:
+        if self.configuration.tls?.requireALPN ?? true {
+          self.configuration.logger.warning("No ALPN protocol negotiated, closing'")
+          context.close(mode: .all, promise: nil)
+        } else {
+          self.configuration.logger.warning("No ALPN protocol negotiated'")
+        }
+      }
+
+    case .shutdownCompleted:
+      // We don't care about this here.
+      ()
+    }
+  }
+
+  // MARK: - Channel Handler
+
+  internal func errorCaught(context: ChannelHandlerContext, error: Error) {
+    if let delegate = self.configuration.errorDelegate {
+      let baseError: Error
+
+      if let errorWithContext = error as? GRPCError.WithContext {
+        baseError = errorWithContext.error
+      } else {
+        baseError = error
+      }
+
+      delegate.observeLibraryError(baseError)
+    }
+
+    context.close(mode: .all, promise: nil)
+  }
+
+  internal func userInboundEventTriggered(context: ChannelHandlerContext, event: Any) {
+    switch self.state {
+    case .notConfigured:
+      if let event = event as? TLSUserEvent {
+        self.handleHandshakeCompletedEvent(event, context: context)
+      }
+    case .configuring:
+      ()
+    }
+
+    context.fireUserInboundEventTriggered(event)
+  }
+
+  internal func channelRead(context: ChannelHandlerContext, data: NIOAny) {
+    self.bufferedReads.append(data)
+
+    switch self.state {
+    case .notConfigured:
+      let buffer = self.unwrapInboundIn(data)
+      self.determineHTTPVersionAndConfigurePipeline(usingBuffer: buffer, context: context)
+
+    case .configuring:
+      // Just buffer, that's okay.
+      ()
+    }
+
+    // Don't forward the reads: we'll do so when we have configured the pipeline.
+  }
+
+  internal func removeHandler(
+    context: ChannelHandlerContext,
+    removalToken: ChannelHandlerContext.RemovalToken
+  ) {
+    // Forward any buffered reads.
+    while let read = self.bufferedReads.popFirst() {
+      context.fireChannelRead(read)
+    }
+    context.leavePipeline(removalToken: removalToken)
+  }
+}
+
+// MARK: - HTTP Version Parser
+
+struct HTTPVersionParser {
+  /// HTTP/2 connection preface bytes. See RFC 7540 § 5.3.
+  private static let http2ClientMagic = [
+    UInt8(ascii: "P"),
+    UInt8(ascii: "R"),
+    UInt8(ascii: "I"),
+    UInt8(ascii: " "),
+    UInt8(ascii: "*"),
+    UInt8(ascii: " "),
+    UInt8(ascii: "H"),
+    UInt8(ascii: "T"),
+    UInt8(ascii: "T"),
+    UInt8(ascii: "P"),
+    UInt8(ascii: "/"),
+    UInt8(ascii: "2"),
+    UInt8(ascii: "."),
+    UInt8(ascii: "0"),
+    UInt8(ascii: "\r"),
+    UInt8(ascii: "\n"),
+    UInt8(ascii: "\r"),
+    UInt8(ascii: "\n"),
+    UInt8(ascii: "S"),
+    UInt8(ascii: "M"),
+    UInt8(ascii: "\r"),
+    UInt8(ascii: "\n"),
+    UInt8(ascii: "\r"),
+    UInt8(ascii: "\n"),
+  ]
+
+  /// Determines whether the bytes in the `ByteBuffer` are prefixed with the HTTP/2 client
+  /// connection preface.
+  static func prefixedWithHTTP2ConnectionPreface(_ buffer: ByteBuffer) -> Bool {
+    var buffer = buffer
+
+    guard let bytes = buffer.readBytes(length: HTTPVersionParser.http2ClientMagic.count) else {
+      return false
+    }
+
+    return bytes == HTTPVersionParser.http2ClientMagic
+  }
+
+  private static let http1_1 = [
+    UInt8(ascii: "H"),
+    UInt8(ascii: "T"),
+    UInt8(ascii: "T"),
+    UInt8(ascii: "P"),
+    UInt8(ascii: "/"),
+    UInt8(ascii: "1"),
+    UInt8(ascii: "."),
+    UInt8(ascii: "1"),
+  ]
+
+  /// Determines whether the bytes in the `ByteBuffer` are prefixed with an HTTP/1.1 request line.
+  static func prefixedWithHTTP1RequestLine(_ buffer: ByteBuffer) -> Bool {
+    var readableBytesView = buffer.readableBytesView
+
+    // From RFC 2616 § 5.1:
+    //   Request-Line = Method SP Request-URI SP HTTP-Version CRLF
+
+    // Read off the Method and Request-URI (and spaces).
+    guard readableBytesView.trimPrefix(to: UInt8(ascii: " ")) != nil,
+      readableBytesView.trimPrefix(to: UInt8(ascii: " ")) != nil else {
+      return false
+    }
+
+    // Read off the HTTP-Version and CR.
+    guard let versionView = readableBytesView.trimPrefix(to: UInt8(ascii: "\r")) else {
+      return false
+    }
+
+    // Check that the LF followed the CR.
+    guard readableBytesView.first == UInt8(ascii: "\n") else {
+      return false
+    }
+
+    // Now check the HTTP version.
+    var version = ByteBuffer(versionView)
+    let versionBytes = version.readBytes(length: version.readableBytes)!
+    return versionBytes == HTTPVersionParser.http1_1
+  }
+}

Sources/GRPC/GRPCServerRequestRoutingHandler.swift

@@ -81,10 +81,10 @@ struct CallPath {
 extension Collection where Self == Self.SubSequence, Self.Element: Equatable {
   /// Trims out the prefix up to `separator`, and returns it.
   /// Sets self to the subsequence after the separator, and returns the subsequence before the separator.
-  /// If self is emtpy returns `nil`
+  /// If self is empty returns `nil`
   /// - parameters:
   ///     - separator : The Element between the head which is returned and the rest which is left in self.
-  /// - returns: SubSequence containing everything between the beginnning and the first occurance of
+  /// - returns: SubSequence containing everything between the beginning and the first occurrence of
   /// `separator`.  If `separator` is not found this will be the entire Collection. If the collection is empty
   /// returns `nil`
   mutating func trimPrefix(to separator: Element) -> SubSequence? {