defmodule HTTP.HTTP2 do @moduledoc false import Bitwise alias HTTP.Headers alias HTTP.HTTP2.Frame alias HTTP.HTTP2.HPACK alias HTTP.Request @connection_preface "PRI * HTTP/2.0\r\n\r\nSM\r\n\r\n" @client_stream_id 1 @initial_window_size 65_535 @initial_max_frame_size 16_384 @max_window_size 2_147_483_647 @max_max_frame_size 16_777_215 @flag_end_stream 0x1 @flag_ack 0x1 @flag_end_headers 0x4 @flag_padded 0x8 @flag_priority 0x20 defstruct method: :get, buffer: <<>>, hpack: HPACK.new_decoder(), continuation: nil, status: nil, connection_receive_window: @initial_window_size, stream_receive_window: @initial_window_size, connection_send_window: @initial_window_size, stream_send_window: @initial_window_size, peer_initial_window_size: @initial_window_size, max_frame_size: @initial_max_frame_size, pending_body: "", outbound: [], done?: false @type event :: {:headers, non_neg_integer(), Headers.t()} | {:body, binary()} | :done @type t :: %__MODULE__{} @spec new(atom()) :: t() def new(method), do: %__MODULE__{method: method} def connection_preface, do: @connection_preface def serialize_request(%Request{} = request) do {conn, wire_request} = prepare_request(new(request.method), request) if conn.pending_body != "" do raise ArgumentError, "HTTP/2 request body exceeds the initial flow-control window; use prepare_request/2" end wire_request end @spec prepare_request(t(), Request.t()) :: {t(), iodata()} def prepare_request(%__MODULE__{} = conn, %Request{} = request) do if Request.streaming_body?(request) do raise ArgumentError, "HTTP/2 streaming request bodies are not supported" end {headers, body} = request |> request_headers() |> Request.put_body_headers(request) body = IO.iodata_to_binary(body) header_block = request |> pseudo_headers() |> Kernel.++(regular_headers(headers)) encoded_headers = HPACK.encode_headers(header_block) header_flags = @flag_end_headers ||| if(body == "", do: @flag_end_stream, else: 0) conn = %{conn | pending_body: body} {conn, outbound} = conn |> flush_pending_body() |> take_outbound() {conn, [ @connection_preface, Frame.encode(:settings, 0, 0, ""), Frame.encode(:headers, header_flags, @client_stream_id, encoded_headers), outbound ]} end @spec stream(t(), binary()) :: {:ok, t(), [event()]} | {:error, term()} def stream(%__MODULE__{} = conn, data) when is_binary(data) do conn |> append_buffer(data) |> parse([]) end @spec close(t()) :: {:ok, t(), [event()]} | {:error, term()} def close(%__MODULE__{done?: true} = conn), do: {:ok, conn, []} def close(_conn), do: {:error, :closed} @spec take_outbound(t()) :: {t(), iodata()} def take_outbound(%__MODULE__{outbound: outbound} = conn) do {%{conn | outbound: []}, Enum.reverse(outbound)} end defp append_buffer(%__MODULE__{buffer: buffer} = conn, data) do %{conn | buffer: buffer <> data} end defp parse(%__MODULE__{} = conn, events) do case Frame.decode(conn.buffer) do :more -> {:ok, conn, Enum.reverse(events)} {:ok, frame, rest} -> with {:ok, conn, frame_events} <- handle_frame(%{conn | buffer: rest}, frame) do parse(conn, append_events(events, frame_events)) end end end defp append_events(events, frame_events) do Enum.reduce(frame_events, events, fn event, acc -> [event | acc] end) end defp handle_frame(%__MODULE__{continuation: continuation}, %Frame{type: type}) when not is_nil(continuation) and type != :continuation do {:error, :expected_continuation} end defp handle_frame(conn, %Frame{type: :settings, stream_id: 0, flags: flags, payload: payload}) do cond do Frame.flag?(flags, @flag_ack) and payload != "" -> {:error, :invalid_settings_ack} Frame.flag?(flags, @flag_ack) -> {:ok, conn, []} rem(byte_size(payload), 6) != 0 -> {:error, :invalid_settings_frame} true -> with {:ok, conn} <- apply_settings(conn, payload) do conn = conn |> enqueue(Frame.encode(:settings, @flag_ack, 0, "")) |> flush_pending_body() {:ok, conn, []} end end end defp handle_frame(_conn, %Frame{type: :settings}), do: {:error, :invalid_settings_stream} defp handle_frame(conn, %Frame{type: :headers, stream_id: @client_stream_id} = frame) do with {:ok, fragment} <- headers_fragment(frame) do end_stream? = Frame.flag?(frame.flags, @flag_end_stream) if Frame.flag?(frame.flags, @flag_end_headers) do decode_response_headers(conn, fragment, end_stream?) else continuation = %{ stream_id: frame.stream_id, fragments: [fragment], end_stream?: end_stream? } {:ok, %{conn | continuation: continuation}, []} end end end defp handle_frame(_conn, %Frame{type: :headers}), do: {:error, :invalid_headers_stream} defp handle_frame( %__MODULE__{continuation: %{stream_id: stream_id} = continuation} = conn, %Frame{type: :continuation, stream_id: stream_id} = frame ) do fragments = [frame.payload | continuation.fragments] if Frame.flag?(frame.flags, @flag_end_headers) do header_block = fragments |> Enum.reverse() |> IO.iodata_to_binary() conn |> Map.put(:continuation, nil) |> decode_response_headers(header_block, continuation.end_stream?) else {:ok, %{conn | continuation: %{continuation | fragments: fragments}}, []} end end defp handle_frame(_conn, %Frame{type: :continuation}), do: {:error, :unexpected_continuation} defp handle_frame(conn, %Frame{type: :data, stream_id: @client_stream_id} = frame) do with {:ok, data, flow_controlled_size} <- data_payload(frame), {:ok, conn} <- consume_receive_window(conn, flow_controlled_size) do end_stream? = Frame.flag?(frame.flags, @flag_end_stream) forbidden? = response_body_forbidden?(conn) conn = if end_stream?, do: %{conn | done?: true}, else: conn events = [] |> maybe_body_event(data, forbidden?) |> maybe_done_event(end_stream?) {:ok, conn, events} end end defp handle_frame(_conn, %Frame{type: :data}), do: {:error, :invalid_data_stream} defp handle_frame(_conn, %Frame{ type: :rst_stream, stream_id: @client_stream_id, payload: <> }) do {:error, {:stream_reset, error_code(code)}} end defp handle_frame(_conn, %Frame{type: :rst_stream, stream_id: @client_stream_id}) do {:error, :invalid_rst_stream} end defp handle_frame(conn, %Frame{type: :rst_stream}), do: {:ok, conn, []} defp handle_frame(%__MODULE__{done?: true} = conn, %Frame{type: :goaway}), do: {:ok, conn, []} defp handle_frame(_conn, %Frame{ type: :goaway, stream_id: 0, payload: <<_reserved::1, last_stream_id::31, code::32, debug::binary>> }) when last_stream_id < @client_stream_id do {:error, {:goaway, error_code(code), debug}} end defp handle_frame(conn, %Frame{ type: :goaway, stream_id: 0, payload: <<_reserved::1, _last_stream_id::31, _code::32, _debug::binary>> }) do {:ok, conn, []} end defp handle_frame(_conn, %Frame{type: :goaway}), do: {:error, :invalid_goaway} defp handle_frame(conn, %Frame{type: :ping, stream_id: 0, flags: flags, payload: payload}) when byte_size(payload) == 8 do if Frame.flag?(flags, @flag_ack) do {:ok, conn, []} else {:ok, enqueue(conn, Frame.encode(:ping, @flag_ack, 0, payload)), []} end end defp handle_frame(_conn, %Frame{type: :ping}), do: {:error, :invalid_ping} defp handle_frame(_conn, %Frame{type: :window_update, payload: <<_reserved::1, 0::31>>}) do {:error, :invalid_window_update_increment} end defp handle_frame(conn, %Frame{ type: :window_update, stream_id: stream_id, payload: <<_reserved::1, increment::31>> }) do with {:ok, conn} <- update_send_window(conn, stream_id, increment) do {:ok, flush_pending_body(conn), []} end end defp handle_frame(_conn, %Frame{type: :window_update}), do: {:error, :invalid_window_update} defp handle_frame(conn, %Frame{}), do: {:ok, conn, []} defp decode_response_headers(%__MODULE__{} = conn, header_block, end_stream?) do with {:ok, hpack, headers} <- HPACK.decode(conn.hpack, header_block), {:ok, status, regular_headers} <- response_headers(headers) do conn = %{conn | hpack: hpack} if status in 100..199 do {:ok, conn, []} else headers = Headers.new(regular_headers) done? = end_stream? or HTTP.HTTP1.body_forbidden?(conn.method, status) conn = %{conn | status: status, done?: done?} events = [{:headers, status, headers}] |> maybe_done_event(done?) {:ok, conn, events} end end end defp response_headers(headers) do status = Enum.find_value(headers, fn {name, value} -> if name == ":status", do: value end) regular_headers = Enum.reject(headers, fn {name, _value} -> String.starts_with?(name, ":") end) with true <- is_binary(status), {status, ""} <- Integer.parse(status) do {:ok, status, regular_headers} else _ -> {:error, :invalid_response_headers} end end defp headers_fragment(%Frame{flags: flags, payload: payload}) do with {:ok, payload, pad_length} <- unpad_payload(payload, Frame.flag?(flags, @flag_padded)), {:ok, payload} <- drop_priority(payload, Frame.flag?(flags, @flag_priority)) do take_padding(payload, pad_length) end end defp data_payload(%Frame{flags: flags, payload: payload}) do with {:ok, payload, pad_length} <- unpad_payload(payload, Frame.flag?(flags, @flag_padded)) do with {:ok, data} <- take_padding(payload, pad_length) do {:ok, data, byte_size(payload) + if(Frame.flag?(flags, @flag_padded), do: 1, else: 0)} end end end defp consume_receive_window(conn, 0), do: {:ok, conn} defp consume_receive_window( %__MODULE__{ connection_receive_window: connection_window, stream_receive_window: stream_window } = conn, size ) do if size > connection_window or size > stream_window do {:error, :flow_control_error} else conn = conn |> Map.update!(:connection_receive_window, &(&1 - size)) |> Map.update!(:stream_receive_window, &(&1 - size)) |> restore_receive_window(size) {:ok, conn} end end defp restore_receive_window(%__MODULE__{} = conn, size) do conn |> Map.update!(:connection_receive_window, &(&1 + size)) |> Map.update!(:stream_receive_window, &(&1 + size)) |> enqueue(window_update_frame(0, size)) |> enqueue(window_update_frame(@client_stream_id, size)) end defp window_update_frame(stream_id, increment) do Frame.encode(:window_update, 0, stream_id, <<0::1, increment::31>>) end defp update_send_window(%__MODULE__{} = conn, 0, increment) do add_window(conn, :connection_send_window, increment) end defp update_send_window(%__MODULE__{} = conn, @client_stream_id, increment) do add_window(conn, :stream_send_window, increment) end defp update_send_window(%__MODULE__{} = conn, _stream_id, _increment), do: {:ok, conn} defp add_window(%__MODULE__{} = conn, key, increment) do window = Map.fetch!(conn, key) if window + increment > @max_window_size do {:error, :flow_control_error} else {:ok, Map.put(conn, key, window + increment)} end end defp apply_settings(%__MODULE__{} = conn, payload) do payload |> parse_settings([]) |> case do {:ok, settings} -> Enum.reduce_while(settings, {:ok, conn}, &apply_setting/2) {:error, _reason} = error -> error end end defp parse_settings(<<>>, settings), do: {:ok, Enum.reverse(settings)} defp parse_settings(<>, settings) do parse_settings(rest, [{id, value} | settings]) end defp parse_settings(_payload, _settings), do: {:error, :invalid_settings_frame} defp apply_setting({0x4, value}, {:ok, conn}) when value <= @max_window_size do delta = value - conn.peer_initial_window_size {:cont, {:ok, %{ conn | peer_initial_window_size: value, stream_send_window: conn.stream_send_window + delta }}} end defp apply_setting({0x4, _value}, {:ok, _conn}), do: {:halt, {:error, :flow_control_error}} defp apply_setting({0x5, value}, {:ok, conn}) when value in @initial_max_frame_size..@max_max_frame_size do {:cont, {:ok, %{conn | max_frame_size: value}}} end defp apply_setting({0x5, _value}, {:ok, _conn}), do: {:halt, {:error, :protocol_error}} defp apply_setting({_id, _value}, {:ok, conn}), do: {:cont, {:ok, conn}} defp unpad_payload(<>, true), do: {:ok, rest, pad_length} defp unpad_payload(<<>>, true), do: {:error, :invalid_padding} defp unpad_payload(payload, false), do: {:ok, payload, 0} defp take_padding(payload, 0), do: {:ok, payload} defp take_padding(payload, pad_length) when byte_size(payload) >= pad_length do body_length = byte_size(payload) - pad_length <> = payload {:ok, body} end defp take_padding(_payload, _pad_length), do: {:error, :invalid_padding} defp drop_priority(<<_exclusive::1, _dependency::31, _weight::8, rest::binary>>, true) do {:ok, rest} end defp drop_priority(_payload, true), do: {:error, :invalid_priority} defp drop_priority(payload, false), do: {:ok, payload} defp maybe_body_event(events, _data, true), do: events defp maybe_body_event(events, "", false), do: events defp maybe_body_event(events, data, false), do: events ++ [{:body, data}] defp maybe_done_event(events, false), do: events defp maybe_done_event(events, true), do: events ++ [:done] defp response_body_forbidden?(%__MODULE__{status: nil}), do: false defp response_body_forbidden?(%__MODULE__{method: method, status: status}) do HTTP.HTTP1.body_forbidden?(method, status) end defp enqueue(%__MODULE__{outbound: outbound} = conn, iodata) do %{conn | outbound: [iodata | outbound]} end defp flush_pending_body(%__MODULE__{pending_body: ""} = conn), do: conn defp flush_pending_body(%__MODULE__{} = conn) do writable = min(conn.connection_send_window, conn.stream_send_window) if writable <= 0 do conn else chunk_size = min(min(writable, conn.max_frame_size), byte_size(conn.pending_body)) <> = conn.pending_body flags = if rest == "", do: @flag_end_stream, else: 0 conn |> Map.update!(:connection_send_window, &(&1 - chunk_size)) |> Map.update!(:stream_send_window, &(&1 - chunk_size)) |> Map.put(:pending_body, rest) |> enqueue(Frame.encode(:data, flags, @client_stream_id, chunk)) |> flush_pending_body() end end defp pseudo_headers(%Request{} = request) do [ {":method", Request.method_token(request.method)}, {":scheme", request.url.scheme || "http"}, {":authority", Request.authority(request.url)}, {":path", Request.origin_form(request.url)} ] end defp request_headers(%Request{} = request) do request.headers |> Request.reject_unsupported_request_framing!() |> Headers.set_default("User-Agent", Headers.user_agent()) end defp regular_headers(%Headers{} = headers) do headers.headers |> Enum.reduce([], fn {name, value}, acc -> name = String.downcase(to_string(name)) value = to_string(value) if request_header_allowed?(name, value) do [{valid_header_name!(name), valid_header_value!(value)} | acc] else acc end end) |> Enum.reverse() end defp request_header_allowed?(name, _value) when name in [ "connection", "host", "keep-alive", "proxy-connection", "transfer-encoding", "upgrade" ] do false end defp request_header_allowed?("te", value), do: String.downcase(String.trim(value)) == "trailers" defp request_header_allowed?(_name, _value), do: true defp valid_header_name!(name) do if valid_token?(name) do name else raise ArgumentError, "invalid HTTP header name: #{inspect(name)}" end end defp valid_header_value!(value) do if safe_header_value?(value) do value else raise ArgumentError, "invalid HTTP header value for wire serialization" end end defp valid_token?(value) when is_binary(value) do value != "" and Enum.all?(:binary.bin_to_list(value), &token_char?/1) end defp token_char?(char) when char in ?0..?9, do: true defp token_char?(char) when char in ?a..?z, do: true defp token_char?(char) when char in ~c"!#$%&'*+-.^_`|~", do: true defp token_char?(_char), do: false defp safe_header_value?(value) do value |> :binary.bin_to_list() |> Enum.all?(fn char -> char == ?\t or (char >= 32 and char != 127) end) end defp error_code(0x0), do: :no_error defp error_code(0x1), do: :protocol_error defp error_code(0x2), do: :internal_error defp error_code(0x3), do: :flow_control_error defp error_code(0x4), do: :settings_timeout defp error_code(0x5), do: :stream_closed defp error_code(0x6), do: :frame_size_error defp error_code(0x7), do: :refused_stream defp error_code(0x8), do: :cancel defp error_code(0x9), do: :compression_error defp error_code(0xA), do: :connect_error defp error_code(0xB), do: :enhance_your_calm defp error_code(0xC), do: :inadequate_security defp error_code(0xD), do: :http_1_1_required defp error_code(code), do: {:unknown_error, code} end