| // Copyright 2024 The Pigweed Authors |
| // |
| // 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 |
| // |
| // https://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. |
| |
| #include "pw_grpc/connection.h" |
| |
| #include <cinttypes> |
| #include <cstring> |
| #include <string_view> |
| #include <type_traits> |
| |
| #include "pw_assert/check.h" |
| #include "pw_chrono/system_clock.h" |
| #include "pw_grpc_private/hpack.h" |
| #include "pw_log/log.h" |
| #include "pw_preprocessor/compiler.h" |
| #include "pw_status/try.h" |
| |
| namespace pw::grpc { |
| namespace internal { |
| |
| // RFC 9113 §6 |
| // Enum names left in naming style of RFC |
| enum class FrameType : uint8_t { |
| DATA = 0x00, |
| HEADERS = 0x01, |
| PRIORITY = 0x02, |
| RST_STREAM = 0x03, |
| SETTINGS = 0x04, |
| PUSH_PROMISE = 0x05, |
| PING = 0x06, |
| GOAWAY = 0x07, |
| WINDOW_UPDATE = 0x08, |
| CONTINUATION = 0x09, |
| }; |
| |
| // RFC 9113 §4.1 |
| constexpr size_t kFrameHeaderEncodedSize = 9; |
| struct FrameHeader { |
| uint32_t payload_length; |
| FrameType type; |
| uint8_t flags; |
| StreamId stream_id; |
| }; |
| |
| // RFC 9113 §7 |
| // Enum names left in naming style of RFC |
| enum class Http2Error : uint32_t { |
| NO_ERROR = 0x00, |
| PROTOCOL_ERROR = 0x01, |
| INTERNAL_ERROR = 0x02, |
| FLOW_CONTROL_ERROR = 0x03, |
| SETTINGS_TIMEOUT = 0x04, |
| STREAM_CLOSED = 0x05, |
| FRAME_SIZE_ERROR = 0x06, |
| REFUSED_STREAM = 0x07, |
| CANCEL = 0x08, |
| COMPRESSION_ERROR = 0x09, |
| CONNECT_ERROR = 0x0a, |
| ENHANCE_YOUR_CALM = 0x0b, |
| INADEQUATE_SECURITY = 0x0c, |
| HTTP_1_1_REQUIRED = 0x0d, |
| }; |
| |
| } // namespace internal |
| |
| namespace { |
| |
| using internal::FrameHeader; |
| using internal::FrameType; |
| using internal::Http2Error; |
| using internal::kMaxConcurrentStreams; |
| using internal::kMaxGrpcMessageSize; |
| |
| // RFC 9113 §3.4 |
| constexpr std::string_view kExpectedConnectionPrefaceLiteral( |
| "PRI * HTTP/2.0\r\n\r\nSM\r\n\r\n"); |
| |
| static_assert(kMaxMethodNameSize == kHpackMaxStringSize); |
| |
| enum { |
| FLAGS_ACK = 0x01, |
| FLAGS_END_STREAM = 0x01, |
| FLAGS_END_HEADERS = 0x04, |
| FLAGS_PADDED = 0x08, |
| FLAGS_PRIORITY = 0x20, |
| }; |
| |
| // RFC 9113 §6.5.2 |
| enum SettingType : uint16_t { |
| SETTINGS_HEADER_TABLE_SIZE = 0x01, |
| SETTINGS_ENABLE_PUSH = 0x02, |
| SETTINGS_MAX_CONCURRENT_STREAMS = 0x03, |
| SETTINGS_INITIAL_WINDOW_SIZE = 0x04, |
| SETTINGS_MAX_FRAME_SIZE = 0x05, |
| SETTINGS_MAX_HEADER_LIST_SIZE = 0x06, |
| }; |
| |
| Status ReadExactly(stream::Reader& reader, ByteSpan buffer) { |
| size_t bytes_read = 0; |
| while (bytes_read < buffer.size()) { |
| PW_TRY_ASSIGN(auto out, reader.Read(buffer.subspan(bytes_read))); |
| bytes_read += out.size(); |
| } |
| return OkStatus(); |
| } |
| |
| Result<FrameHeader> ReadFrameHeader(stream::Reader& reader) { |
| std::array<std::byte, internal::kFrameHeaderEncodedSize> buffer; |
| PW_TRY(ReadExactly(reader, buffer)); |
| |
| // RFC 9113 §4.1 |
| FrameHeader out; |
| ByteBuilder builder(as_writable_bytes(span{buffer})); |
| auto it = builder.begin(); |
| auto type_and_length = it.ReadUint32(endian::big); |
| out.payload_length = type_and_length >> 8; |
| out.type = static_cast<FrameType>(type_and_length & 0xff); |
| out.flags = it.ReadUint8(); |
| out.stream_id = it.ReadUint32(endian::big) & 0x7fffffff; |
| return out; |
| } |
| |
| template <typename T, std::enable_if_t<std::is_integral_v<T>, bool> = true> |
| constexpr T ToNetworkOrder(T value) { |
| return bytes::ConvertOrder(/*from=*/endian::native, |
| /*to=*/endian::big, |
| value); |
| } |
| |
| template <typename T, std::enable_if_t<std::is_enum_v<T>, bool> = true> |
| constexpr std::underlying_type_t<T> ToNetworkOrder(T value) { |
| return ToNetworkOrder(static_cast<std::underlying_type_t<T>>(value)); |
| } |
| |
| // Use this instead of FrameHeader when writing frames. |
| PW_PACKED(struct) WireFrameHeader { |
| WireFrameHeader(FrameHeader h) |
| : payload_length_and_type(ToNetworkOrder(h.payload_length << 8 | |
| static_cast<uint32_t>(h.type))), |
| flags(h.flags), |
| stream_id(ToNetworkOrder(h.stream_id)) {} |
| |
| uint32_t payload_length_and_type; |
| uint8_t flags; |
| uint32_t stream_id; |
| }; |
| |
| template <typename T> |
| ConstByteSpan AsBytes(T& object) { |
| return as_bytes(span<T, 1>{&object, 1}); |
| } |
| |
| // RFC 9113 §6.1 |
| Status SendData(SendQueue& send_queue, |
| StreamId stream_id, |
| ConstByteSpan payload1, |
| ConstByteSpan payload2) { |
| PW_LOG_DEBUG("Conn.Send DATA with id=%" PRIu32 " len1=%" PRIu32 |
| " len2=%" PRIu32, |
| stream_id, |
| static_cast<uint32_t>(payload1.size()), |
| static_cast<uint32_t>(payload2.size())); |
| WireFrameHeader frame(FrameHeader{ |
| .payload_length = |
| static_cast<uint32_t>(payload1.size() + payload2.size()), |
| .type = FrameType::DATA, |
| .flags = 0, |
| .stream_id = stream_id, |
| }); |
| std::array<ConstByteSpan, 3> data_vector = {AsBytes(frame)}; |
| size_t i = 1; |
| if (!payload1.empty()) { |
| data_vector[i++] = payload1; |
| } |
| if (!payload2.empty()) { |
| data_vector[i++] = payload2; |
| } |
| PW_TRY(send_queue.SendBytesVector(span{data_vector.data(), i})); |
| return OkStatus(); |
| } |
| |
| // RFC 9113 §6.2 |
| Status SendHeaders(SendQueue& send_queue, |
| StreamId stream_id, |
| ConstByteSpan payload1, |
| ConstByteSpan payload2, |
| bool end_stream) { |
| PW_LOG_DEBUG("Conn.Send HEADERS with id=%" PRIu32 " len1=%" PRIu32 |
| " len2=%" PRIu32 " end=%d", |
| stream_id, |
| static_cast<uint32_t>(payload1.size()), |
| static_cast<uint32_t>(payload2.size()), |
| end_stream); |
| WireFrameHeader frame(FrameHeader{ |
| .payload_length = |
| static_cast<uint32_t>(payload1.size() + payload2.size()), |
| .type = FrameType::HEADERS, |
| .flags = FLAGS_END_HEADERS, |
| .stream_id = stream_id, |
| }); |
| |
| if (end_stream) { |
| frame.flags |= FLAGS_END_STREAM; |
| } |
| |
| std::array<ConstByteSpan, 3> headers_vector = {AsBytes(frame)}; |
| size_t i = 1; |
| if (!payload1.empty()) { |
| headers_vector[i++] = payload1; |
| } |
| if (!payload2.empty()) { |
| headers_vector[i++] = payload2; |
| } |
| PW_TRY(send_queue.SendBytesVector(span{headers_vector.data(), i})); |
| |
| return OkStatus(); |
| } |
| |
| // RFC 9113 §6.4 |
| Status SendRstStream(SendQueue& send_queue, |
| StreamId stream_id, |
| Http2Error code) { |
| PW_PACKED(struct) RstStreamFrame { |
| WireFrameHeader header; |
| uint32_t error_code; |
| }; |
| RstStreamFrame frame{ |
| .header = WireFrameHeader(FrameHeader{ |
| .payload_length = 4, |
| .type = FrameType::RST_STREAM, |
| .flags = 0, |
| .stream_id = stream_id, |
| }), |
| .error_code = ToNetworkOrder(code), |
| }; |
| PW_TRY(send_queue.SendBytes(AsBytes(frame))); |
| |
| return OkStatus(); |
| } |
| |
| // RFC 9113 §6.9 |
| Status SendWindowUpdates(SendQueue& send_queue, |
| StreamId stream_id, |
| uint32_t increment) { |
| // It is illegal to send updates with increment=0. |
| if (increment == 0) { |
| return OkStatus(); |
| } |
| if (increment & 0x80000000) { |
| // Upper bit is reserved, error. |
| return Status::InvalidArgument(); |
| } |
| |
| PW_LOG_DEBUG("Conn.Send WINDOW_UPDATE frames with id=%" PRIu32 |
| " increment=%" PRIu32, |
| stream_id, |
| increment); |
| |
| PW_PACKED(struct) WindowUpdateFrame { |
| WireFrameHeader header; |
| uint32_t increment; |
| }; |
| WindowUpdateFrame frames[2] = { |
| { |
| .header = WireFrameHeader(FrameHeader{ |
| .payload_length = 4, |
| .type = FrameType::WINDOW_UPDATE, |
| .flags = 0, |
| .stream_id = 0, |
| }), |
| .increment = ToNetworkOrder(increment), |
| }, |
| { |
| .header = WireFrameHeader(FrameHeader{ |
| .payload_length = 4, |
| .type = FrameType::WINDOW_UPDATE, |
| .flags = 0, |
| .stream_id = stream_id, |
| }), |
| .increment = ToNetworkOrder(increment), |
| }, |
| }; |
| PW_TRY(send_queue.SendBytes(as_bytes(span{frames}))); |
| return OkStatus(); |
| } |
| |
| // RFC 9113 §6.5 |
| Status SendSettingsAck(SendQueue& send_queue) { |
| PW_LOG_DEBUG("Conn.Send SETTINGS ACK"); |
| WireFrameHeader frame(FrameHeader{ |
| .payload_length = 0, |
| .type = FrameType::SETTINGS, |
| .flags = FLAGS_ACK, |
| .stream_id = 0, |
| }); |
| PW_TRY(send_queue.SendBytes(AsBytes(frame))); |
| return OkStatus(); |
| } |
| |
| } // namespace |
| |
| Connection::Connection(stream::ReaderWriter& socket, |
| SendQueue& send_queue, |
| RequestCallbacks& callbacks, |
| allocator::Allocator* message_assembly_allocator) |
| : socket_(socket), |
| send_queue_(send_queue), |
| reader_(*this, callbacks), |
| writer_(*this) { |
| LockState()->message_assembly_allocator_ = message_assembly_allocator; |
| } |
| |
| Status Connection::Reader::ProcessFrame() { |
| if (!received_connection_preface_) { |
| return Status::FailedPrecondition(); |
| } |
| |
| PW_TRY_ASSIGN(auto frame, ReadFrameHeader(connection_.socket_.as_reader())); |
| switch (frame.type) { |
| // Frames that we handle. |
| case FrameType::DATA: |
| PW_TRY(ProcessDataFrame(frame)); |
| break; |
| case FrameType::HEADERS: |
| PW_TRY(ProcessHeadersFrame(frame)); |
| break; |
| case FrameType::PRIORITY: |
| PW_TRY(ProcessIgnoredFrame(frame)); |
| break; |
| case FrameType::RST_STREAM: |
| PW_TRY(ProcessRstStreamFrame(frame)); |
| break; |
| case FrameType::SETTINGS: |
| PW_TRY(ProcessSettingsFrame(frame, /*send_ack=*/true)); |
| break; |
| case FrameType::PING: |
| PW_TRY(ProcessPingFrame(frame)); |
| break; |
| case FrameType::WINDOW_UPDATE: |
| PW_TRY(ProcessWindowUpdateFrame(frame)); |
| break; |
| |
| // Frames that trigger an immediate connection close. |
| case FrameType::GOAWAY: |
| PW_LOG_ERROR("Client sent GOAWAY"); |
| // don't bother sending GOAWAY in response |
| return Status::Internal(); |
| case FrameType::PUSH_PROMISE: |
| PW_LOG_ERROR("Client sent PUSH_PROMISE"); |
| SendGoAway(Http2Error::PROTOCOL_ERROR); |
| return Status::Internal(); |
| case FrameType::CONTINUATION: |
| PW_LOG_ERROR("Client sent CONTINUATION: unsupported"); |
| SendGoAway(Http2Error::INTERNAL_ERROR); |
| return Status::Internal(); |
| } |
| |
| return OkStatus(); |
| } |
| |
| pw::Result<std::reference_wrapper<Connection::Stream>> |
| Connection::SharedState::LookupStream(StreamId id) { |
| for (size_t i = 0; i < streams.size(); i++) { |
| if (streams[i].id == id) { |
| return streams[i]; |
| } |
| } |
| return Status::NotFound(); |
| } |
| |
| Status Connection::Writer::SendResponseMessage(StreamId stream_id, |
| ConstByteSpan message) { |
| auto state = connection_.LockState(); |
| auto stream = state->LookupStream(stream_id); |
| if (!stream.ok()) { |
| return Status::NotFound(); |
| } |
| |
| if (message.size() > kMaxGrpcMessageSize) { |
| PW_LOG_WARN("Message %" PRIu32 " bytes on id=%" PRIu32 |
| " exceeds maximum message size", |
| static_cast<uint32_t>(message.size()), |
| stream_id); |
| return Status::InvalidArgument(); |
| } |
| |
| // This should block until there is enough send window. |
| if (static_cast<int32_t>(message.size()) > stream->get().send_window || |
| static_cast<int32_t>(message.size()) > state->connection_send_window) { |
| PW_LOG_WARN("Not enough window to send %" PRIu32 " bytes on id=%" PRIu32, |
| static_cast<uint32_t>(message.size()), |
| stream_id); |
| return Status::ResourceExhausted(); |
| } |
| |
| auto status = OkStatus(); |
| if (!stream->get().started_response) { |
| stream->get().started_response = true; |
| status = SendHeaders(connection_.send_queue_, |
| stream_id, |
| ResponseHeadersPayload(), |
| ConstByteSpan(), |
| /*end_stream=*/false); |
| } |
| if (status.ok()) { |
| // Write a Length-Prefixed-Message payload. |
| ByteBuffer<5> prefix; |
| prefix.PutUint8(0); |
| prefix.PutUint32(message.size(), endian::big); |
| status = SendData(connection_.send_queue_, stream_id, prefix, message); |
| } |
| if (!status.ok()) { |
| PW_LOG_WARN("Failed sending response message on id=%" PRIu32 " error=%d", |
| stream_id, |
| status.code()); |
| return Status::Unavailable(); |
| } |
| stream->get().send_window -= message.size(); |
| state->connection_send_window -= message.size(); |
| return OkStatus(); |
| } |
| |
| Status Connection::Writer::SendResponseComplete(StreamId stream_id, |
| Status response_code) { |
| auto state = connection_.LockState(); |
| auto stream = state->LookupStream(stream_id); |
| if (!stream.ok()) { |
| return Status::NotFound(); |
| } |
| |
| Status status; |
| if (!stream->get().started_response) { |
| // If the response has not started yet, we need to include the initial |
| // headers. |
| PW_LOG_DEBUG("Conn.SendResponseWithTrailers id=%" PRIu32 " code=%d", |
| stream_id, |
| response_code.code()); |
| status = SendHeaders(connection_.send_queue_, |
| stream_id, |
| ResponseHeadersPayload(), |
| ResponseTrailersPayload(response_code), |
| /*end_stream=*/true); |
| } else { |
| PW_LOG_DEBUG("Conn.SendTrailers id=%" PRIu32 " code=%d", |
| stream_id, |
| response_code.code()); |
| status = SendHeaders(connection_.send_queue_, |
| stream_id, |
| ConstByteSpan(), |
| ResponseTrailersPayload(response_code), |
| /*end_stream=*/true); |
| } |
| |
| if (!status.ok()) { |
| PW_LOG_WARN("Failed sending response complete on id=%" PRIu32 " error=%d", |
| stream_id, |
| status.code()); |
| return Status::Unavailable(); |
| } |
| |
| PW_LOG_DEBUG("Conn.CloseStream id=%" PRIu32, stream_id); |
| stream->get().Reset(); |
| |
| return OkStatus(); |
| } |
| |
| pw::Status Connection::Reader::CreateStream(StreamId id) { |
| auto state = connection_.LockState(); |
| for (size_t i = 0; i < state->streams.size(); i++) { |
| if (state->streams[i].id != 0) { |
| continue; |
| } |
| PW_LOG_DEBUG("Conn.CreateStream id=%" PRIu32 " at slot=%" PRIu32, |
| id, |
| static_cast<uint32_t>(i)); |
| state->streams[i].id = id; |
| state->streams[i].half_closed = false; |
| state->streams[i].started_response = false; |
| state->streams[i].send_window = initial_send_window_; |
| return OkStatus(); |
| } |
| PW_LOG_WARN("Conn.CreateStream id=%" PRIu32 " OUT OF SPACE", id); |
| return Status::ResourceExhausted(); |
| } |
| |
| void Connection::Reader::CloseStream(Connection::Stream& stream) { |
| StreamId id = stream.id; |
| PW_LOG_DEBUG("Conn.CloseStream id=%" PRIu32, id); |
| stream.Reset(); |
| callbacks_.OnCancel(id); |
| } |
| |
| // RFC 9113 §3.4 |
| Status Connection::Reader::ProcessConnectionPreface() { |
| if (received_connection_preface_) { |
| return OkStatus(); |
| } |
| |
| callbacks_.OnNewConnection(); |
| |
| // The preface starts with a literal string. |
| auto literal = span{payload_scratch_}.subspan( |
| 0, kExpectedConnectionPrefaceLiteral.size()); |
| |
| PW_TRY(ReadExactly(connection_.socket_.as_reader(), literal)); |
| if (std::memcmp(literal.data(), |
| kExpectedConnectionPrefaceLiteral.data(), |
| kExpectedConnectionPrefaceLiteral.size()) != 0) { |
| PW_LOG_ERROR("Invalid connection preface literal"); |
| return Status::Internal(); |
| } |
| |
| PW_LOG_DEBUG("Conn.Preface received literal"); |
| |
| // Client must send a SETTINGS frames. |
| PW_TRY_ASSIGN(auto client_frame, |
| ReadFrameHeader(connection_.socket_.as_reader())); |
| if (client_frame.type != FrameType::SETTINGS) { |
| PW_LOG_ERROR( |
| "Connection preface missing SETTINGS frame, found frame.type=%d", |
| static_cast<int>(client_frame.type)); |
| return Status::Internal(); |
| } |
| |
| // Don't send an ACK yet, we'll do that below. |
| PW_TRY(ProcessSettingsFrame(client_frame, /*send_ack=*/false)); |
| PW_LOG_DEBUG("Conn.Preface received SETTINGS"); |
| |
| // We must send a SETTINGS frame. |
| // RFC 9113 §6.5.2 |
| PW_PACKED(struct) Setting { |
| uint16_t id; |
| uint32_t value; |
| }; |
| PW_PACKED(struct) SettingsFrame { |
| WireFrameHeader header; |
| Setting settings[2]; |
| }; |
| SettingsFrame server_frame{ |
| .header = WireFrameHeader(FrameHeader{ |
| .payload_length = 12, |
| .type = FrameType::SETTINGS, |
| .flags = 0, |
| .stream_id = 0, |
| }), |
| .settings = |
| { |
| { |
| .id = ToNetworkOrder(SETTINGS_HEADER_TABLE_SIZE), |
| .value = ToNetworkOrder(kHpackDynamicHeaderTableSize), |
| }, |
| { |
| .id = ToNetworkOrder(SETTINGS_MAX_CONCURRENT_STREAMS), |
| .value = ToNetworkOrder(kMaxConcurrentStreams), |
| }, |
| }, |
| }; |
| PW_LOG_DEBUG("Conn.Send SETTINGS"); |
| PW_TRY(connection_.send_queue_.SendBytes(AsBytes(server_frame))); |
| |
| // We must ack the client's SETTINGS frame *after* sending our SETTINGS. |
| PW_TRY(SendSettingsAck(connection_.send_queue_)); |
| |
| received_connection_preface_ = true; |
| PW_LOG_DEBUG("Conn.Preface complete"); |
| return OkStatus(); |
| } |
| |
| // RFC 9113 §6.1 |
| Status Connection::Reader::ProcessDataFrame(const FrameHeader& frame) { |
| PW_LOG_DEBUG("Conn.Recv DATA id=%" PRIu32 " flags=0x%x len=%" PRIu32, |
| frame.stream_id, |
| frame.flags, |
| frame.payload_length); |
| |
| if (frame.stream_id == 0) { |
| // RFC 9113 §6.1: "If a DATA frame is received whose Stream Identifier field |
| // is 0x00, the recipient MUST respond with a connection error of type |
| // PROTOCOL_ERROR." |
| SendGoAway(Http2Error::PROTOCOL_ERROR); |
| return Status::Internal(); |
| } |
| |
| // From RFC 9113 §6.9: "A receiver that receives a flow-controlled frame MUST |
| // always account for its contribution against the connection flow-control |
| // window, unless the receiver treats this as a connection error. This is |
| // necessary even if the frame is in error. The sender counts the frame toward |
| // the flow-control window, but if the receiver does not, the flow-control |
| // window at the sender and receiver can become different." |
| // |
| // To simplify this, we send WINDOW_UPDATE frames eagerly. |
| // |
| // In the future we should do something less chatty. |
| PW_TRY(SendWindowUpdates( |
| connection_.send_queue_, frame.stream_id, frame.payload_length)); |
| |
| { |
| auto state = connection_.LockState(); |
| auto stream = state->LookupStream(frame.stream_id); |
| if (!stream.ok()) { |
| PW_LOG_DEBUG("Ignoring DATA on closed stream id=%" PRIu32, |
| frame.stream_id); |
| // Stream has been fully closed: silently ignore. |
| return OkStatus(); |
| } |
| |
| if (stream->get().half_closed) { |
| PW_LOG_ERROR("Recv DATA on half-closed stream id=%" PRIu32, |
| frame.stream_id); |
| // RFC 9113 §6.1: "If a DATA frame is received whose stream is not in the |
| // "open" or "half-closed (local)" state, the recipient MUST respond with |
| // a stream error of type STREAM_CLOSED." |
| PW_TRY(SendRstStreamAndClose(stream->get(), Http2Error::STREAM_CLOSED)); |
| return OkStatus(); |
| } |
| } |
| |
| PW_TRY_ASSIGN(auto payload, ReadFramePayload(frame)); |
| |
| // Drop padding. |
| if ((frame.flags & FLAGS_PADDED) != 0) { |
| uint32_t pad_length = static_cast<uint32_t>(payload[0]); |
| if (pad_length >= frame.payload_length) { |
| // RFC 9113 §6.1: "If the length of the padding is the length of the frame |
| // payload or greater, the recipient MUST treat this as a connection error |
| // of type PROTOCOL_ERROR." |
| SendGoAway(Http2Error::PROTOCOL_ERROR); |
| return Status::Internal(); |
| } |
| payload = payload.subspan(1, payload.size() - pad_length - 1); |
| } |
| |
| auto state = connection_.LockState(); |
| auto maybe_stream = state->LookupStream(frame.stream_id); |
| if (!maybe_stream.ok()) { |
| return OkStatus(); |
| } |
| Stream* stream = &maybe_stream->get(); |
| |
| // Parse repeated grpc Length-Prefix-Message. |
| // https://github.com/grpc/grpc/blob/v1.60.x/doc/PROTOCOL-HTTP2.md#requests |
| while (!payload.empty()) { |
| uint32_t message_length; |
| |
| // If we aren't reassembling a message, read the next length prefix. |
| if (!stream->assembly_buffer) { |
| size_t read = std::min(5 - static_cast<size_t>(stream->prefix_received), |
| payload.size()); |
| std::copy(payload.begin(), |
| payload.begin() + read, |
| stream->prefix_buffer.data() + stream->prefix_received); |
| stream->prefix_received += read; |
| payload = payload.subspan(read); |
| |
| // Read the length prefix. |
| if (stream->prefix_received < 5) { |
| continue; |
| } |
| stream->prefix_received = 0; |
| |
| ByteBuilder builder(stream->prefix_buffer); |
| auto it = builder.begin(); |
| auto message_compressed = it.ReadUint8(); |
| message_length = it.ReadUint32(endian::big); |
| if (message_compressed != 0) { |
| PW_LOG_ERROR("Unsupported: grpc message is compressed"); |
| PW_TRY(SendRstStreamAndClose(*stream, Http2Error::INTERNAL_ERROR)); |
| return OkStatus(); |
| } |
| |
| if (message_length > payload.size()) { |
| // gRPC message is split across DATA frames, must allocate buffer. |
| if (!state->message_assembly_allocator_) { |
| PW_LOG_ERROR( |
| "Unsupported: split grpc message without allocator provided"); |
| PW_TRY(SendRstStreamAndClose(*stream, Http2Error::INTERNAL_ERROR)); |
| return OkStatus(); |
| } |
| |
| stream->assembly_buffer = static_cast<std::byte*>( |
| state->message_assembly_allocator_->Allocate( |
| allocator::Layout(message_length))); |
| if (stream->assembly_buffer == nullptr) { |
| PW_LOG_ERROR("Partial message reassembly buffer allocation failed"); |
| PW_TRY(SendRstStreamAndClose(*stream, Http2Error::INTERNAL_ERROR)); |
| return OkStatus(); |
| } |
| stream->message_length = message_length; |
| stream->message_received = 0; |
| continue; |
| } |
| } |
| |
| pw::ByteSpan message; |
| |
| // Reading message payload. |
| if (stream->assembly_buffer != nullptr) { |
| uint32_t read = |
| std::min(stream->message_length - stream->message_received, |
| static_cast<uint32_t>(payload.size())); |
| std::copy(payload.begin(), |
| payload.begin() + read, |
| stream->assembly_buffer + stream->message_received); |
| payload = payload.subspan(read); |
| stream->message_received += read; |
| if (stream->message_received < stream->message_length) { |
| continue; |
| } |
| // Fully received message. |
| message = pw::span(stream->assembly_buffer, stream->message_length); |
| } else { |
| message = payload.subspan(0, message_length); |
| payload = payload.subspan(message_length); |
| } |
| |
| // Release state lock before callback, reacquire after. |
| connection_.UnlockState(std::move(state)); |
| const auto status = callbacks_.OnMessage(frame.stream_id, message); |
| state = connection_.LockState(); |
| auto maybe_stream = state->LookupStream(frame.stream_id); |
| if (!maybe_stream.ok()) { |
| return OkStatus(); |
| } |
| stream = &maybe_stream->get(); |
| |
| if (!status.ok()) { |
| PW_TRY(SendRstStreamAndClose(*stream, Http2Error::INTERNAL_ERROR)); |
| return OkStatus(); |
| } |
| |
| if (stream->assembly_buffer != nullptr) { |
| state->message_assembly_allocator_->Deallocate(stream->assembly_buffer); |
| stream->assembly_buffer = nullptr; |
| stream->message_length = 0; |
| stream->message_received = 0; |
| } |
| } |
| |
| // grpc requires every request stream to end with an empty DATA frame with |
| // FLAGS_END_STREAM. If a client sends FLAGS_END_STREAM with a non-empty |
| // payload, it's not specified how the server should respond. We choose to |
| // accept the payload before ending the stream. |
| // See: https://github.com/grpc/grpc/blob/v1.60.x/doc/PROTOCOL-HTTP2.md. |
| if ((frame.flags & FLAGS_END_STREAM) != 0) { |
| stream->half_closed = true; |
| connection_.UnlockState(std::move(state)); |
| callbacks_.OnHalfClose(frame.stream_id); |
| } |
| |
| return OkStatus(); |
| } |
| |
| // RFC 9113 §6.2 |
| Status Connection::Reader::ProcessHeadersFrame(const FrameHeader& frame) { |
| PW_LOG_DEBUG("Conn.Recv HEADERS id=%" PRIu32 " len=%" PRIu32, |
| frame.stream_id, |
| frame.payload_length); |
| |
| if (frame.stream_id == 0) { |
| // RFC 9113 §6.2: "If a HEADERS frame is received whose Stream Identifier |
| // field is 0x00, the recipient MUST respond with a connection error of type |
| // PROTOCOL_ERROR." |
| SendGoAway(Http2Error::PROTOCOL_ERROR); |
| return Status::Internal(); |
| } |
| if (frame.stream_id % 2 != 1 || frame.stream_id <= last_stream_id_) { |
| // RFC 9113 §5.1.1: "Streams initiated by a client MUST use odd-numbered |
| // stream identifiers ... The identifier of a newly established stream MUST |
| // be numerically greater than all streams that the initiating endpoint has |
| // opened ... An endpoint that receives an unexpected stream identifier MUST |
| // respond with a connection error of type PROTOCOL_ERROR." |
| SendGoAway(Http2Error::PROTOCOL_ERROR); |
| return Status::Internal(); |
| } |
| |
| last_stream_id_ = frame.stream_id; |
| |
| { |
| auto state = connection_.LockState(); |
| if (auto stream = state->LookupStream(frame.stream_id); stream.ok()) { |
| PW_LOG_DEBUG("Client sent HEADERS after the first stream message"); |
| // grpc requests cannot contain trailers. |
| // See: https://github.com/grpc/grpc/blob/v1.60.x/doc/PROTOCOL-HTTP2.md. |
| PW_TRY(SendRstStreamAndClose(stream->get(), Http2Error::PROTOCOL_ERROR)); |
| return OkStatus(); |
| } |
| } |
| |
| if ((frame.flags & FLAGS_END_STREAM) != 0) { |
| PW_LOG_DEBUG("Client sent HEADERS with END_STREAM"); |
| // grpc requests must send END_STREAM in an empty DATA frame. |
| // See: https://github.com/grpc/grpc/blob/v1.60.x/doc/PROTOCOL-HTTP2.md. |
| PW_TRY(SendRstStream( |
| connection_.send_queue_, frame.stream_id, Http2Error::PROTOCOL_ERROR)); |
| return OkStatus(); |
| } |
| if ((frame.flags & FLAGS_END_HEADERS) == 0) { |
| PW_LOG_ERROR("Client sent HEADERS frame without END_HEADERS: unsupported"); |
| SendGoAway(Http2Error::INTERNAL_ERROR); |
| return Status::Internal(); |
| } |
| |
| PW_TRY_ASSIGN(auto payload, ReadFramePayload(frame)); |
| |
| // Drop padding. |
| if ((frame.flags & FLAGS_PADDED) != 0) { |
| uint32_t pad_length = static_cast<uint32_t>(payload[0]); |
| if (pad_length >= frame.payload_length) { |
| // RFC 9113 §6.2: "If the length of the padding is the length of the frame |
| // payload or greater, the recipient MUST treat this as a connection error |
| // of type PROTOCOL_ERROR." |
| SendGoAway(Http2Error::PROTOCOL_ERROR); |
| return Status::Internal(); |
| } |
| payload = payload.subspan(1, payload.size() - pad_length - 1); |
| } |
| |
| // Drop priority fields. |
| if ((frame.flags & FLAGS_PRIORITY) != 0) { |
| payload = payload.subspan(5); |
| } |
| |
| PW_TRY_ASSIGN(auto method_name, HpackParseRequestHeaders(payload)); |
| if (!CreateStream(frame.stream_id).ok()) { |
| PW_LOG_WARN("Too many streams, rejecting id=%" PRIu32, frame.stream_id); |
| return SendRstStream( |
| connection_.send_queue_, frame.stream_id, Http2Error::REFUSED_STREAM); |
| } |
| |
| if (const auto status = callbacks_.OnNew(frame.stream_id, method_name); |
| !status.ok()) { |
| auto state = connection_.LockState(); |
| if (auto stream = state->LookupStream(frame.stream_id); stream.ok()) { |
| return SendRstStreamAndClose(stream->get(), Http2Error::INTERNAL_ERROR); |
| } |
| } |
| |
| return OkStatus(); |
| } |
| |
| // RFC 9113 §6.4 |
| Status Connection::Reader::ProcessRstStreamFrame(const FrameHeader& frame) { |
| PW_LOG_DEBUG("Conn.Recv RST_STREAM id=%" PRIu32 " len=%" PRIu32, |
| frame.stream_id, |
| frame.payload_length); |
| |
| if (frame.stream_id == 0) { |
| // RFC 9113 §6.4: "If a RST_STREAM frame is received with a stream |
| // identifier of 0x00, the recipient MUST treat this as a connection error |
| // of type PROTOCOL_ERROR". |
| SendGoAway(Http2Error::PROTOCOL_ERROR); |
| return Status::Internal(); |
| } |
| if (frame.stream_id > last_stream_id_) { |
| // RFC 9113 §6.4: "If a RST_STREAM frame identifying an idle stream is |
| // received, the recipient MUST treat this as a connection error of type |
| // PROTOCOL_ERROR." |
| SendGoAway(Http2Error::PROTOCOL_ERROR); |
| return Status::Internal(); |
| } |
| if (frame.payload_length != 4) { |
| // RFC 9113 §6.4: "A RST_STREAM frame with a length other than 4 octets MUST |
| // be treated as a connection error of type FRAME_SIZE_ERROR." |
| SendGoAway(Http2Error::FRAME_SIZE_ERROR); |
| return Status::Internal(); |
| } |
| |
| PW_TRY_ASSIGN(auto payload, ReadFramePayload(frame)); |
| ByteBuilder builder(payload); |
| auto error_code = builder.begin().ReadUint32(endian::big); |
| |
| PW_LOG_DEBUG("Conn.RstStream id=%" PRIu32 " error=%" PRIu32, |
| frame.stream_id, |
| error_code); |
| auto state = connection_.LockState(); |
| if (auto stream = state->LookupStream(frame.stream_id); stream.ok()) { |
| CloseStream(stream->get()); |
| } |
| return OkStatus(); |
| } |
| |
| // RFC 9113 §6.5 |
| Status Connection::Reader::ProcessSettingsFrame(const FrameHeader& frame, |
| bool send_ack) { |
| PW_LOG_DEBUG("Conn.Recv SETTINGS len=%" PRIu32 " flags=0x%x", |
| frame.payload_length, |
| frame.flags); |
| |
| if ((frame.flags & FLAGS_ACK) != 0) { |
| // RFC 9113 §6.5: "Receipt of a SETTINGS frame with the ACK flag set and a |
| // length field value other than 0 MUST be treated as a connection error of |
| // type FRAME_SIZE_ERROR." |
| if (frame.payload_length != 0) { |
| PW_LOG_ERROR("Invalid SETTINGS frame: has ACK with non-empty payload"); |
| SendGoAway(Http2Error::FRAME_SIZE_ERROR); |
| return Status::Internal(); |
| } |
| // Don't ACK an ACK. |
| send_ack = false; |
| } else { |
| // RFC 9113 §6.5: "A SETTINGS frame with a length other than a multiple of 6 |
| // octets MUST be treated as a connection error of type FRAME_SIZE_ERROR." |
| if (frame.payload_length % 6 != 0) { |
| PW_LOG_ERROR("Invalid SETTINGS frame: payload size invalid"); |
| SendGoAway(Http2Error::FRAME_SIZE_ERROR); |
| return Status::Internal(); |
| } |
| } |
| |
| if (frame.stream_id != 0) { |
| // RFC 9113 §6.5: "If an endpoint receives a SETTINGS frame whose Stream |
| // Identifier field is anything other than 0x00, the endpoint MUST respond |
| // with a connection error of type PROTOCOL_ERROR." |
| SendGoAway(Http2Error::PROTOCOL_ERROR); |
| return Status::Internal(); |
| } |
| |
| PW_TRY_ASSIGN(auto payload, ReadFramePayload(frame)); |
| |
| // RFC 9113 §6.5.2 |
| ByteBuilder builder(payload); |
| for (auto it = builder.begin(); it != builder.end();) { |
| auto id = it.ReadUint16(endian::big); |
| auto value = it.ReadUint32(endian::big); |
| PW_LOG_DEBUG("Applying SETTING id=%" PRIu16 " value=%" PRIu32, id, value); |
| switch (id) { |
| case SETTINGS_INITIAL_WINDOW_SIZE: { |
| // RFC 9113 §6.5.2: "Values above the maximum flow-control window size |
| // of 2^31-1 MUST be treated as a connection error of type |
| // FLOW_CONTROL_ERROR." |
| if ((value & (1 << 31)) != 0) { |
| SendGoAway(Http2Error::FLOW_CONTROL_ERROR); |
| return Status::Internal(); |
| } |
| // RFC 9113 §6.9.2: "When the value of SETTINGS_INITIAL_WINDOW_SIZE |
| // changes, a receiver MUST adjust the size of all stream flow-control |
| // windows that it maintains by the difference between the new value and |
| // the old value." |
| int32_t newval = static_cast<int32_t>(value); |
| int32_t delta = newval - initial_send_window_; |
| auto state = connection_.LockState(); |
| for (size_t i = 0; i < state->streams.size(); i++) { |
| if (state->streams[i].id == 0) { |
| continue; |
| } |
| if (PW_ADD_OVERFLOW(state->streams[i].send_window, |
| delta, |
| &state->streams[i].send_window)) { |
| SendGoAway(Http2Error::FLOW_CONTROL_ERROR); |
| return Status::Internal(); |
| } |
| } |
| initial_send_window_ = newval; |
| break; |
| } |
| case SETTINGS_MAX_FRAME_SIZE: |
| // RFC 9113 §6.5.2: "Values outside this range MUST be treated as a |
| // connection error of type PROTOCOL_ERROR". |
| if (value < 16384 || 16777215 < value) { |
| SendGoAway(Http2Error::PROTOCOL_ERROR); |
| return Status::Internal(); |
| } |
| // We never send frame payloads larger than 16384, so we don't need to |
| // track the client's preference. |
| break; |
| // Ignore these. |
| // SETTINGS_HEADER_TABLE_SIZE: our responses don't use the dynamic table |
| // SETTINGS_ENABLE_PUSH: we don't support push |
| // SETTINGS_MAX_CONCURRENT_STREAMS: we don't support push |
| // SETTINGS_MAX_HEADER_LIST_SIZE: we send very tiny response HEADERS |
| default: |
| break; |
| } |
| } |
| |
| if (send_ack) { |
| PW_TRY(SendSettingsAck(connection_.send_queue_)); |
| } |
| |
| return OkStatus(); |
| } |
| |
| // RFC 9113 §6.7 |
| Status Connection::Reader::ProcessPingFrame(const FrameHeader& frame) { |
| PW_LOG_DEBUG("Conn.Recv PING len=%" PRIu32, frame.payload_length); |
| |
| if (frame.stream_id != 0) { |
| // RFC 9113 §6.7: "If a PING frame is received with a Stream Identifier |
| // field value other than 0x00, the recipient MUST respond with a connection |
| // error of type PROTOCOL_ERROR." |
| SendGoAway(Http2Error::PROTOCOL_ERROR); |
| return Status::Internal(); |
| } |
| if (frame.payload_length != 8) { |
| // RFC 9113 §6.7: "Receipt of a PING frame with a length field value other |
| // than 8 MUST be treated as a connection error of type FRAME_SIZE_ERROR." |
| SendGoAway(Http2Error::FRAME_SIZE_ERROR); |
| return Status::Internal(); |
| } |
| |
| PW_TRY_ASSIGN(auto payload, ReadFramePayload(frame)); |
| |
| // Don't ACK an ACK. |
| if ((frame.flags & FLAGS_ACK) != 0) { |
| return OkStatus(); |
| } |
| |
| // Send an ACK. |
| PW_PACKED(struct) PingFrame { |
| WireFrameHeader header; |
| uint64_t opaque_data; |
| }; |
| ByteBuilder builder(payload); |
| PingFrame ack_frame = { |
| .header = WireFrameHeader(FrameHeader{ |
| .payload_length = 8, |
| .type = FrameType::PING, |
| .flags = FLAGS_ACK, |
| .stream_id = 0, |
| }), |
| // Since we're going to echo this, read as native endian so it gets echoed |
| // exactly as-is. |
| .opaque_data = builder.begin().ReadUint64(endian::native), |
| }; |
| PW_TRY(connection_.send_queue_.SendBytes(AsBytes(ack_frame))); |
| return OkStatus(); |
| } |
| |
| // RFC 9113 §6.9 |
| Status Connection::Reader::ProcessWindowUpdateFrame(const FrameHeader& frame) { |
| PW_LOG_DEBUG("Conn.Recv WINDOW_UPDATE id=%" PRIu32 " len=%" PRIu32, |
| frame.stream_id, |
| frame.payload_length); |
| |
| if (frame.payload_length != 4) { |
| // RFC 9113 §6.9: "A WINDOW_UPDATE frame with a length other than 4 octets |
| // MUST be treated as a connection error of type FRAME_SIZE_ERROR." |
| SendGoAway(Http2Error::FRAME_SIZE_ERROR); |
| return Status::Internal(); |
| } |
| |
| // Read window size increment. |
| PW_TRY_ASSIGN(auto payload, ReadFramePayload(frame)); |
| ByteBuilder builder(payload); |
| int32_t delta = static_cast<int32_t>(builder.begin().ReadUint32(endian::big) & |
| 0x7fffffff); |
| |
| auto state = connection_.LockState(); |
| auto stream = state->LookupStream(frame.stream_id); |
| |
| if (delta == 0) { |
| // RFC 9113 §6.9: "A receiver MUST treat a WINDOW_UPDATE frame with an |
| // increment of 0 as a stream error of type PROTOCOL_ERROR; errors on the |
| // connection flow-control window MUST be treated as a connection error." |
| if (frame.stream_id == 0) { |
| SendGoAway(Http2Error::PROTOCOL_ERROR); |
| return Status::Internal(); |
| } else { |
| if (!stream.ok()) { |
| // Already closed |
| return OkStatus(); |
| } |
| PW_TRY(SendRstStreamAndClose(stream->get(), Http2Error::PROTOCOL_ERROR)); |
| return OkStatus(); |
| } |
| } |
| |
| // RFC 9113 §6.9.1: "If a sender receives a WINDOW_UPDATE that causes a |
| // flow-control window to exceed 2^31-1 bytes, it MUST terminate either the |
| // stream or the connection, as appropriate ... with an error code of |
| // FLOW_CONTROL_ERROR" |
| if (frame.stream_id == 0) { |
| if (PW_ADD_OVERFLOW(state->connection_send_window, |
| delta, |
| &state->connection_send_window)) { |
| SendGoAway(Http2Error::FLOW_CONTROL_ERROR); |
| return Status::Internal(); |
| } |
| } else if (stream.ok()) { |
| if (PW_ADD_OVERFLOW( |
| stream->get().send_window, delta, &stream->get().send_window)) { |
| PW_TRY( |
| SendRstStreamAndClose(stream->get(), Http2Error::FLOW_CONTROL_ERROR)); |
| return OkStatus(); |
| } |
| } |
| |
| return OkStatus(); |
| } |
| |
| // Advance past the payload. |
| Status Connection::Reader::ProcessIgnoredFrame(const FrameHeader& frame) { |
| PW_TRY(ReadFramePayload(frame)); |
| return OkStatus(); |
| } |
| |
| Result<ByteSpan> Connection::Reader::ReadFramePayload( |
| const FrameHeader& frame) { |
| if (frame.payload_length == 0) { |
| return ByteSpan(); |
| } |
| if (frame.payload_length > payload_scratch_.size()) { |
| PW_LOG_ERROR("Frame type=%d payload too large: %" PRIu32 " > %" PRIu32, |
| static_cast<int>(frame.type), |
| frame.payload_length, |
| static_cast<uint32_t>(payload_scratch_.size())); |
| SendGoAway(Http2Error::FRAME_SIZE_ERROR); |
| return Status::Internal(); |
| } |
| auto payload = span{payload_scratch_}.subspan(0, frame.payload_length); |
| PW_TRY(ReadExactly(connection_.socket_.as_reader(), payload)); |
| return payload; |
| } |
| |
| // RFC 9113 §6.8 |
| void Connection::Reader::SendGoAway(Http2Error code) { |
| if (!received_connection_preface_) { |
| // RFC 9113 §3.4: "A GOAWAY frame MAY be omitted in this case, since an |
| // invalid preface indicates that the peer is not using HTTP/2." |
| return; |
| } |
| |
| // Close all open streams. |
| { |
| auto state = connection_.LockState(); |
| for (size_t i = 0; i < state->streams.size(); i++) { |
| if (state->streams[i].id != 0) { |
| CloseStream(state->streams[i]); |
| } |
| } |
| } |
| |
| PW_PACKED(struct) GoAwayFrame { |
| WireFrameHeader header; |
| uint32_t last_stream_id; |
| uint32_t error_code; |
| }; |
| GoAwayFrame frame{ |
| .header = WireFrameHeader(FrameHeader{ |
| .payload_length = 8, |
| .type = FrameType::GOAWAY, |
| .flags = 0, |
| .stream_id = 0, |
| }), |
| .last_stream_id = ToNetworkOrder(last_stream_id_), |
| .error_code = ToNetworkOrder(code), |
| }; |
| // Ignore errors since we're about to close the connection anyway. |
| connection_.send_queue_.SendBytes(AsBytes(frame)).IgnoreError(); |
| } |
| |
| // RFC 9113 §6.4 |
| Status Connection::Reader::SendRstStreamAndClose(Stream& stream, |
| Http2Error code) { |
| // Ignore errors as we are closing anyways. |
| SendRstStream(connection_.send_queue_, stream.id, code).IgnoreError(); |
| CloseStream(stream); |
| return OkStatus(); |
| } |
| |
| } // namespace pw::grpc |
