diff --git a/lib/ankh/http.ex b/lib/ankh/http.ex index a84d238..042fb7b 100644 --- a/lib/ankh/http.ex +++ b/lib/ankh/http.ex @@ -3,12 +3,39 @@ defmodule Ankh.HTTP do HTTP public interface This module implements `Ankh` public APIs and provides protocol negotiation. + + ## Protocol negotiation for `https://` + + When connecting to an `https://` URI, the `:protocols` option controls which + HTTP version is attempted and in what order: + + * `[:h2, :"http/1.1"]` — TLS/TCP only; ALPN negotiates between HTTP/2 and + HTTP/1.1 (this is the **default**). + * `[:h3]` — QUIC only; returns an error if the QUIC handshake fails. + * `[:h3, :h2, :"http/1.1"]` — QUIC/HTTP3 is tried first; if it fails + (e.g. UDP is blocked by a firewall), the connection falls back to TLS + and ALPN negotiates between HTTP/2 and HTTP/1.1. + + HTTP/3 runs exclusively over QUIC. Use `https://` with `protocols: [:h3]` + (or `protocols: [:h3, :h2, :"http/1.1"]` for automatic fallback). + + ## Examples + + # HTTP/2 or HTTP/1.1 over TLS (default) + Ankh.HTTP.connect(URI.parse("https://example.com"), []) + + # HTTP/3 over QUIC only + Ankh.HTTP.connect(URI.parse("https://example.com"), protocols: [:h3]) + + # HTTP/3 preferred, fall back to TLS automatically + Ankh.HTTP.connect(URI.parse("https://example.com"), protocols: [:h3, :h2, :"http/1.1"]) + """ alias Ankh.{HTTP, Protocol, Transport} alias Ankh.HTTP.{Error, Request, Response} - alias Ankh.Protocol.{HTTP1, HTTP2} - alias Ankh.Transport.{TCP, TLS} + alias Ankh.Protocol.{HTTP1, HTTP2, HTTP3} + alias Ankh.Transport.{QUIC, TCP, TLS} @typedoc "HTTP body" @type body :: iodata() @@ -45,13 +72,25 @@ defmodule Ankh.HTTP do def accept(%URI{scheme: "http"} = uri, socket, options), do: Protocol.accept(%HTTP1{}, uri, %TCP{}, socket, options) - def accept(%URI{scheme: "https"} = uri, socket, options) do + def accept(%URI{scheme: "https"} = uri, socket, options) when is_port(socket) do with {:ok, negotiated_protocol} <- Transport.negotiated_protocol(%TLS{socket: socket}), {:ok, protocol} <- protocol_for_id(negotiated_protocol) do Protocol.accept(protocol, uri, %TLS{}, socket, options) end end + def accept(%URI{scheme: "https"} = uri, socket, options) when is_pid(socket) do + # erlang_quic (pure Erlang) connection — pid carries the negotiated ALPN. + # The caller may pass alpn: binary() in options; default to "h3" for QUIC. + alpn = Keyword.get(options, :alpn, "h3") + transport = %QUIC{alpn: alpn} + + with {:ok, negotiated_protocol} <- Transport.negotiated_protocol(transport), + {:ok, protocol} <- protocol_for_id(negotiated_protocol) do + Protocol.accept(protocol, uri, transport, socket, options) + end + end + def accept(_uri, _socket, _options), do: {:error, :unsupported_uri_scheme} @doc """ @@ -73,12 +112,59 @@ defmodule Ankh.HTTP do end def connect(%URI{scheme: "https"} = uri, options) do + {protocols, options} = Keyword.pop(options, :protocols, [:h2, :"http/1.1"]) + + if :h3 in protocols do + tls_fallback = Enum.filter(protocols, &(&1 in [:h2, :"http/1.1"])) + + case connect_quic(uri, options) do + {:ok, _} = ok -> + ok + + {:error, _} when tls_fallback != [] -> + connect_tls(uri, options, tls_fallback) + + {:error, _} = error -> + error + end + else + connect_tls(uri, options, protocols) + end + end + + def connect(_uri, _options), do: {:error, :unsupported_uri_scheme} + + # Establishes an HTTP/3 connection over QUIC using erlang_quic. + # + # erlang_quic takes ALPN as Elixir binaries (not charlists). + # max_streams_uni: 3 allows the server to open the three unidirectional + # streams required by HTTP/3 (control, QPACK encoder, QPACK decoder). + defp connect_quic(%URI{} = uri, options) do + {timeout, options} = Keyword.pop(options, :timeout, 5_000) + + quic_options = + [alpn: ["h3"], verify: true, max_streams_uni: 3] + |> Keyword.merge(Keyword.take(options, [:verify, :max_streams_uni])) + + case Transport.connect(%QUIC{}, uri, timeout, quic_options) do + {:ok, transport} -> Protocol.connect(%HTTP3{}, uri, transport, options) + {:error, _} = error -> error + end + end + + # Establishes an HTTP/1.1 or HTTP/2 connection over TLS. + # + # `protocols` is a list of atoms from `[:h2, :"http/1.1"]` that are + # converted to their ALPN token strings and advertised to the server. + defp connect_tls(%URI{} = uri, options, protocols) do + alpn_tokens = Enum.map(protocols, &protocol_to_alpn_token/1) + {:ok, tls_options} = Plug.SSL.configure(cipher_suite: :strong, key: nil, cert: nil) tls_options = tls_options |> Keyword.take([:versions, :ciphers, :eccs]) - |> Keyword.merge(alpn_advertised_protocols: ["h2", "http/1.1"]) + |> Keyword.merge(alpn_advertised_protocols: alpn_tokens) {timeout, options} = options @@ -92,10 +178,12 @@ defmodule Ankh.HTTP do end end - def connect(_uri, _options), do: {:error, :unsupported_uri_scheme} + defp protocol_to_alpn_token(:h2), do: "h2" + defp protocol_to_alpn_token(:"http/1.1"), do: "http/1.1" defp protocol_for_id("http/1.1"), do: {:ok, %HTTP1{}} defp protocol_for_id("h2"), do: {:ok, %HTTP2{}} + defp protocol_for_id("h3"), do: {:ok, %HTTP3{}} defp protocol_for_id(_id), do: {:error, :unsupported_protocol_identifier} @doc """ diff --git a/lib/ankh/protocol/http2.ex b/lib/ankh/protocol/http2.ex index 71d983a..120de3f 100644 --- a/lib/ankh/protocol/http2.ex +++ b/lib/ankh/protocol/http2.ex @@ -345,9 +345,7 @@ defmodule Ankh.Protocol.HTTP2 do end defp process_queued_frame(%@for{send_queue: send_queue} = protocol, %Data{} = frame, size) - when size <= 0 do - {:ok, %{protocol | send_queue: :queue.in_r(frame, send_queue)}} - end + when size <= 0, do: {:ok, %{protocol | send_queue: :queue.in_r(frame, send_queue)}} defp process_queued_frame(protocol, %{stream_id: stream_id} = frame, size) do frame diff --git a/lib/ankh/protocol/http3.ex b/lib/ankh/protocol/http3.ex new file mode 100644 index 0000000..059543e --- /dev/null +++ b/lib/ankh/protocol/http3.ex @@ -0,0 +1,588 @@ +defmodule Ankh.Protocol.HTTP3 do + @moduledoc """ + HTTP/3 protocol implementation for Ankh. + + This module implements the `Ankh.Protocol` protocol over a QUIC connection + managed by the `Ankh.Transport.QUIC` transport. It supports both the client + (`connect/4`) and server (`accept/5`) sides of an HTTP/3 connection. + + ## Architecture + + Unlike TCP-based HTTP protocols, QUIC is inherently multiplexed. Each logical + HTTP request/response pair runs on its own QUIC stream. When operating in HTTP/3 + mode the `Ankh.Transport.QUIC` struct has `stream: nil`; raw QUIC messages are + forwarded directly to this protocol's `stream/2` callback rather than being + unwrapped by the transport layer. + + Each QUIC stream is tracked in the `streams` map keyed by the opaque QUIC + stream handle. A parallel `references` map provides the reverse lookup from + Elixir request references to stream handles, allowing callers to correlate + `stream/2` response events back to the originating `request/2` or `respond/3` + call. + + ## Header compression + + All header compression and decompression is delegated to + `Ankh.Protocol.HTTP3.QPACK`. Both the send and receive QPACK contexts are + initialised in static-only mode (`max_table_capacity = 0`), which eliminates + the need for dedicated QPACK encoder and decoder streams and is declared in + the SETTINGS preface written on the control stream. + + ## HTTP/3 framing + + Frame encoding and decoding (HEADERS, DATA, SETTINGS, GOAWAY, etc.) are + delegated to `Ankh.Protocol.HTTP3.Frame`. Variable-length integer encoding + follows RFC 9000 §16. + + ## Control stream + + Both `accept/5` and `connect/4` open a unidirectional HTTP/3 control stream + immediately after establishing the connection and write the mandatory SETTINGS + preface before returning. This satisfies the RFC 9114 §6.2.1 requirement that + a SETTINGS frame MUST be the first frame sent on the control stream. + + ## Per-stream lifecycle + + Each QUIC stream's HTTP/3 state is tracked by an `Ankh.Protocol.HTTP3.Stream` + struct that models the stream lifecycle + (`idle → open → half_closed_local/remote → closed`) and accumulates incoming + bytes between message deliveries. Frame parsing runs over the accumulated buffer + and any unconsumed remainder is retained for the next delivery. + + ## Incoming message dispatch + + `stream/2` pattern-matches the raw QUIC tuple messages delivered by erlang_quic: + + * `{:quic, conn, {:stream_data, id, data, fin}}` — binary data (and optional + FIN); appended to the stream buffer and parsed as HTTP/3 frames. When + `fin = true` an empty DATA event with `complete = true` is also emitted. + * `{:quic, conn, {:stream_opened, id}}` — new inbound stream from the peer + (server-side). Stream directionality is derived from the stream ID. + * `{:quic, conn, {:stream_reset, id, code}}` — stream reset by the peer; + both half-closes are recorded. + * `{:quic, conn, {:transport_error | :closed, …}}` — connection teardown; + returns `{:error, :closed}`. + + ## References + + * RFC 9114 — HTTP/3 + * RFC 9000 — QUIC Transport Protocol + * RFC 9204 — QPACK: Field Compression for HTTP/3 + """ + + alias Ankh.{Protocol, Transport} + alias Ankh.Protocol.HTTP3.Stream + alias Ankh.Transport.QUIC + + @typedoc """ + Opaque HTTP/3 protocol state struct. + + Carries the QPACK encode/decode contexts, the per-stream state maps, the + underlying QUIC transport, the connection URI, and both local and remote + SETTINGS values. + """ + @opaque t :: %__MODULE__{ + mode: :client | :server | nil, + recv_qpack: :quic_qpack.state() | nil, + send_qpack: :quic_qpack.state() | nil, + streams: %{optional(reference()) => Stream.t()}, + references: %{optional(reference()) => reference()}, + transport: QUIC.t() | nil, + uri: URI.t() | nil, + local_settings: keyword(), + remote_settings: keyword() + } + + defstruct mode: nil, + recv_qpack: nil, + send_qpack: nil, + streams: %{}, + references: %{}, + transport: nil, + uri: nil, + local_settings: [], + remote_settings: [] + + defimpl Protocol do + alias Ankh.HTTP.{Request, Response} + alias Ankh.Protocol.HTTP3.{Frame, Stream} + + alias Ankh.Protocol.HTTP3.Frame.{ + CancelPush, + Data, + GoAway, + Headers, + MaxPushId, + PushPromise, + Settings + } + + alias Ankh.Transport + alias Ankh.Transport.QUIC + require Logger + + @doc """ + Accepts an incoming QUIC connection and initialises the HTTP/3 server state. + + Initialises QPACK encode/decode contexts in static-only mode, opens a + unidirectional HTTP/3 control stream on the accepted connection, and writes + the mandatory SETTINGS preface (RFC 9114 §6.2.1) before returning. + + The `socket` argument must be a raw QUIC connection handle (`reference()`) + as returned by the transport after accepting a connection from a listener. + """ + def accept(%@for{} = protocol, uri, transport, socket, _options) do + with {:ok, %QUIC{connection: conn} = transport} <- Transport.new(transport, socket), + :ok <- QUIC.open_unidirectional_stream(transport, Frame.control_stream_preface()) do + Logger.debug("HTTP/3 server accepted connection: #{inspect(conn)}") + + {:ok, + %@for{ + protocol + | mode: :server, + transport: transport, + uri: uri, + recv_qpack: :quic_qpack.new(), + send_qpack: :quic_qpack.new() + }} + end + end + + @doc """ + Initialises the HTTP/3 client state on an already-established QUIC connection. + + The QUIC connection handle must already be present in `transport.connection` + (established by the caller, e.g. via `Ankh.HTTP.connect/2`). Initialises + QPACK contexts, opens a unidirectional control stream, and sends the SETTINGS + preface before returning. + """ + def connect(%@for{} = protocol, uri, %QUIC{connection: conn} = transport, _options) do + with :ok <- QUIC.open_unidirectional_stream(transport, Frame.control_stream_preface()) do + Logger.debug("HTTP/3 connection established: #{inspect(conn)}") + + {:ok, + %@for{ + protocol + | mode: :client, + transport: transport, + uri: uri, + recv_qpack: :quic_qpack.new(), + send_qpack: :quic_qpack.new() + }} + end + end + + @doc """ + Closes the QUIC connection, signalling a connection-level error to the peer. + """ + def error(%@for{transport: transport}), do: Transport.close(transport) + + @doc """ + Sends an HTTP/3 request on a new bidirectional QUIC stream. + + Constructs the complete header block — pseudo-headers (`:method`, `:scheme`, + `:authority`, `:path`) in order, followed by the request's regular headers — + QPACK-encodes it, and sends it in an HTTP/3 HEADERS frame on a freshly opened + bidirectional stream. If the request body is non-empty, a DATA frame follows + and the stream is gracefully shut down (FIN). + + Returns `{:ok, updated_protocol, ref}` where `ref` is the Elixir + `reference()` that identifies this request in subsequent `stream/2` events. + """ + def request( + %@for{ + uri: %{authority: authority, scheme: scheme}, + transport: transport + } = protocol, + %Request{method: method, path: path, headers: req_headers, body: body} + ) do + pseudo_headers = [ + {":method", Atom.to_string(method)}, + {":scheme", to_string(scheme)}, + {":authority", to_string(authority)}, + {":path", path || "/"} + ] + + all_headers = pseudo_headers ++ req_headers + {hbf, send_qpack} = :quic_qpack.encode(all_headers, protocol.send_qpack) + + with {:ok, _, headers_frame_data} <- + Frame.encode(%Headers{payload: %Headers.Payload{hbf: IO.iodata_to_binary(hbf)}}), + {:ok, stream_transport} <- QUIC.open_stream(transport), + :ok <- Transport.send(stream_transport, headers_frame_data) do + if IO.iodata_length(body) > 0 do + {:ok, _, data_frame_data} = Frame.encode(%Data{payload: %Data.Payload{data: body}}) + Transport.send(stream_transport, data_frame_data) + end + + # Always FIN the send side — even for no-body requests (e.g. GET). + # Without this the server never knows the request is complete and + # will not send a response. + QUIC.shutdown_stream(stream_transport) + + handle = stream_transport.stream + stream = Stream.new_request(handle) + ref = stream.reference + + Logger.debug("HTTP/3 request sent: #{method} #{path} ref=#{inspect(ref)}") + + protocol = %@for{ + protocol + | send_qpack: send_qpack, + streams: Map.put(protocol.streams, handle, stream), + references: Map.put(protocol.references, ref, handle) + } + + {:ok, protocol, ref} + end + end + + @doc """ + Sends an HTTP/3 response on the stream identified by `request_reference`. + + Builds the `:status` pseudo-header, appends the response headers, QPACK- + encodes the full block, and sends it in a HEADERS frame. If the response body + is non-empty a DATA frame follows. If trailers are present they are + QPACK-encoded and sent as a second HEADERS frame. Finally, the stream is + gracefully shut down (FIN). + """ + def respond( + %@for{transport: %QUIC{}} = protocol, + request_reference, + %Response{status: status, headers: resp_headers, body: body, trailers: trailers} + ) do + with {:ok, handle, stream} <- lookup_stream(protocol, request_reference) do + headers = [{":status", Integer.to_string(status)} | resp_headers] + + {hbf, send_qpack} = :quic_qpack.encode(headers, protocol.send_qpack) + + {:ok, _, headers_frame_data} = + Frame.encode(%Headers{payload: %Headers.Payload{hbf: IO.iodata_to_binary(hbf)}}) + + transport = %{protocol.transport | stream: handle} + Transport.send(transport, headers_frame_data) + + if IO.iodata_length(body) > 0 do + {:ok, _, data} = Frame.encode(%Data{payload: %Data.Payload{data: body}}) + Transport.send(transport, data) + end + + send_qpack = + if trailers != [] do + {trailer_hbf, send_qpack} = :quic_qpack.encode(trailers, send_qpack) + + {:ok, _, trailer_frame_data} = + Frame.encode(%Headers{payload: %Headers.Payload{hbf: trailer_hbf}}) + + Transport.send(transport, trailer_frame_data) + send_qpack + else + send_qpack + end + + QUIC.shutdown_stream(transport) + + Logger.debug("HTTP/3 response sent: #{status} ref=#{inspect(request_reference)}") + + { + :ok, + %@for{ + protocol + | send_qpack: send_qpack, + streams: Map.put(protocol.streams, handle, Stream.local_fin(stream)) + } + } + end + end + + @doc """ + Dispatches a raw QUIC message and returns HTTP/3 response events. + + This is the sole entry point for all incoming data and connection-level + events in HTTP/3 mode. Each clause delegates to a private helper that + returns `{:ok, protocol, responses}` or `{:error, reason}`. + + Response tuples follow the `Ankh.HTTP.response()` shape: + + * `{:headers, ref, headers, complete?}` — a decoded HEADERS frame + * `{:data, ref, data, complete?}` — a DATA frame or end-of-stream signal + * `{:error, ref, reason, true}` — a stream-level protocol error + """ + + def stream(%@for{} = protocol, {:quic, _conn, {:stream_data, stream_id, data, false}}) + when is_binary(data) do + handle_stream_data(protocol, stream_id, data) + end + + # Data WITH FIN — process frames then signal end of stream. + def stream(%@for{} = protocol, {:quic, _conn, {:stream_data, stream_id, data, true}}) do + with {:ok, protocol, responses} <- handle_stream_data(protocol, stream_id, data), + {:ok, protocol, fin_responses} <- handle_peer_send_shutdown(protocol, stream_id) do + {:ok, protocol, responses ++ fin_responses} + end + end + + # New stream opened by peer (server-side). + def stream(%@for{} = protocol, {:quic, _conn, {:stream_opened, stream_id}}), + do: handle_new_stream(protocol, stream_id) + + # Peer reset the stream. + def stream(%@for{} = protocol, {:quic, _conn, {:stream_reset, stream_id, _error_code}}), + do: handle_stream_closed(protocol, stream_id) + + def stream(%@for{} = _protocol, {:quic, _conn, {:transport_error, _code, _reason}}), + do: {:error, :closed} + + def stream(%@for{} = _protocol, {:quic, _conn, {:closed, _reason}}), do: {:error, :closed} + + def stream(%@for{} = protocol, _msg), do: {:ok, protocol, []} + + # Resolves an Elixir request reference to its QUIC stream handle and + # per-stream state struct. Returns `{:error, :unknown_reference}` when + # the reference is not present in the protocol maps. + defp lookup_stream(%@for{references: references, streams: streams}, ref) do + with {:ok, handle} <- Map.fetch(references, ref), + {:ok, stream} <- Map.fetch(streams, handle) do + {:ok, handle, stream} + else + :error -> {:error, :unknown_reference} + end + end + + # Handles incoming binary data on a QUIC stream. + # + # Retrieves (or lazily creates) the per-stream state, optionally strips the + # unidirectional stream-type byte (RFC 9114 §6.2 / RFC 9204 §4.2) from + # newly-seen streams, appends the data to the stream buffer, and runs the + # HTTP/3 frame parser. No re-arm is needed — erlang_quic delivers messages + # automatically. + defp handle_stream_data(%@for{streams: streams} = protocol, stream_id, data) do + stream = Map.get(streams, stream_id, Stream.new_incoming(stream_id, false)) + + # For unidirectional streams whose kind has not yet been identified, + # the very first byte of data is the stream-type byte. Strip it, resolve + # the kind, then treat the remainder as HTTP/3 frame data. + {stream, effective_data} = maybe_identify_stream_kind(stream, data) + + process_stream_frames(protocol, stream_id, Stream.append(stream, effective_data)) + end + + # Signals end-of-stream to the caller via an empty DATA event with + # `complete = true`. Called when the peer sends FIN (fin=true on the last + # :stream_data, or implicitly after a :stream_reset). + defp handle_peer_send_shutdown(%@for{streams: streams} = protocol, stream_id) do + case Map.get(streams, stream_id) do + nil -> + {:ok, protocol, []} + + stream -> + stream = Stream.remote_fin(stream) + + { + :ok, + %@for{protocol | streams: Map.put(protocol.streams, stream_id, stream)}, + [{:data, stream.reference, <<>>, true}] + } + end + end + + # Marks both halves of the stream as closed. Called when the peer resets + # the stream via a RESET_STREAM frame. + defp handle_stream_closed(%@for{} = protocol, stream_id) do + case Map.get(protocol.streams, stream_id) do + nil -> + {:ok, protocol, []} + + stream -> + stream = + stream + |> Stream.remote_fin() + |> Stream.local_fin() + + {:ok, %@for{protocol | streams: Map.put(protocol.streams, stream_id, stream)}, []} + end + end + + # Registers a newly opened inbound stream. Unidirectionality is inferred + # from bit 1 of the QUIC stream ID (RFC 9000 §2.1): if set, the stream is + # unidirectional. + defp handle_new_stream(%@for{} = protocol, stream_id) do + unidirectional? = Bitwise.band(stream_id, 2) != 0 + stream = Stream.new_incoming(stream_id, unidirectional?) + Logger.debug("HTTP/3 new stream: id=#{stream_id}, unidirectional=#{unidirectional?}") + {:ok, %@for{protocol | streams: Map.put(protocol.streams, stream_id, stream)}, []} + end + + # Strips the stream-type byte from an unidirectional stream whose kind is not + # yet known and updates the stream kind via `Stream.identify_kind/2`. + # All other stream kinds pass data through unchanged. + defp maybe_identify_stream_kind( + %Stream{kind: :unknown_unidirectional} = stream, + <> + ), + do: {Stream.identify_kind(stream, type_byte), rest} + + defp maybe_identify_stream_kind(stream, data), do: {stream, data} + + # Parses all complete HTTP/3 frames from the stream's accumulated buffer and + # converts them into Ankh response tuples. Any incomplete trailing bytes are + # kept in the stream buffer for the next delivery. + # Maps a wire frame-type integer to the corresponding empty frame struct, + # mirroring `frame_for_type/1` in `Ankh.Protocol.HTTP2`. Unknown type + # integers return `{:error, :not_found}`; the caller ignores them per + # RFC 9114 §9. + defp frame_for_type(0x0), do: {:ok, %Data{}} + defp frame_for_type(0x1), do: {:ok, %Headers{}} + defp frame_for_type(0x3), do: {:ok, %CancelPush{}} + defp frame_for_type(0x4), do: {:ok, %Settings{}} + defp frame_for_type(0x5), do: {:ok, %PushPromise{}} + defp frame_for_type(0x7), do: {:ok, %GoAway{}} + defp frame_for_type(0xD), do: {:ok, %MaxPushId{}} + defp frame_for_type(_), do: {:error, :not_found} + + # Parses all complete HTTP/3 frames from the stream's accumulated buffer, + # decodes each one into a typed struct via `Frame.decode/2`, and dispatches + # to `recv_frame/4`. Mirrors `process_buffer/1` in `Ankh.Protocol.HTTP2`. + defp process_stream_frames( + %@for{} = protocol, + stream_handle, + %Stream{buffer: buffer} = stream + ) do + result = + buffer + |> Frame.stream() + |> Enum.reduce_while( + {:ok, protocol, stream, []}, + fn + {rest, nil}, {:ok, protocol, stream, responses} -> + # Partial frame — store remainder and stop. + {:halt, {:ok, protocol, Stream.consume_buffer(stream, rest), responses}} + + {rest, {type, payload}}, {:ok, protocol, stream, responses} -> + stream = Stream.consume_buffer(stream, rest) + + with {:ok, frame_struct} <- frame_for_type(type), + {:ok, frame} <- Frame.decode(frame_struct, payload), + {:ok, p, stream, new_responses} <- + recv_frame(protocol, stream, stream_handle, frame) do + {:cont, {:ok, p, stream, new_responses ++ responses}} + else + {:error, :not_found} -> + # Unknown frame type — RFC 9114 §9: MUST be ignored. + {:cont, {:ok, protocol, stream, responses}} + + {:error, _reason} = error -> + {:halt, error} + end + end + ) + + case result do + {:ok, %@for{} = protocol, stream, responses} -> + { + :ok, + %@for{protocol | streams: Map.put(protocol.streams, stream_handle, stream)}, + Enum.reverse(responses) + } + + {:error, _reason} = error -> + error + end + end + + # HEADERS frame — QPACK-decode the HBF, validate pseudo-headers, emit + # a `:headers` response event. + defp recv_frame( + %@for{} = protocol, + %Stream{} = stream, + _stream_handle, + %Headers{payload: %Headers.Payload{hbf: hbf}} + ) do + case :quic_qpack.decode(hbf, protocol.recv_qpack) do + {{:ok, headers}, recv_qpack} -> + protocol = %{protocol | recv_qpack: recv_qpack} + + case validate_incoming_headers(protocol, stream, headers) do + :ok -> + { + :ok, + protocol, + %{stream | recv_headers: true}, + [{:headers, stream.reference, headers, false}] + } + + {:error, reason} -> + {:ok, protocol, stream, [{:error, stream.reference, reason, true}]} + end + + {{:blocked, _stream_id}, recv_qpack} -> + { + :ok, + %{protocol | recv_qpack: recv_qpack}, + stream, + [{:error, stream.reference, :stream_blocked, true}] + } + + {{:error, reason}, recv_qpack} -> + { + :ok, + %{protocol | recv_qpack: recv_qpack}, + stream, + [{:error, stream.reference, reason, true}] + } + end + end + + # DATA frame — emit a `:data` response event with the raw body bytes. + defp recv_frame( + %@for{} = protocol, + stream, + _stream_handle, + %Data{payload: %Data.Payload{data: data}} + ), + do: {:ok, protocol, stream, [{:data, stream.reference, data, false}]} + + # SETTINGS frame — update `remote_settings` and log; no response event. + defp recv_frame( + %@for{} = protocol, + stream, + _stream_handle, + %Settings{payload: %Settings.Payload{settings: settings}} + ) do + Logger.debug("HTTP/3 SETTINGS received: #{inspect(settings)}") + {:ok, %@for{protocol | remote_settings: settings}, stream, []} + end + + # GOAWAY frame — emit a connection-level error event (RFC 9114 §7.2.6). + defp recv_frame(%@for{} = protocol, stream, _stream_handle, %GoAway{}), + do: {:ok, protocol, stream, [{:error, stream.reference, :goaway, true}]} + + # All other known and unknown frame types are silently ignored per + # RFC 9114 §9 ("extension frames MUST be ignored"). + defp recv_frame(%@for{} = protocol, stream, _stream_handle, _frame), + do: {:ok, protocol, stream, []} + + # Validates pseudo-headers on the first HEADERS block of a stream. + # + # On a server-mode connection the first block carries request pseudo-headers + # (`:method`, `:scheme`, `:authority`, `:path`). On a client-mode connection + # the first block carries response pseudo-headers (`:status`). Subsequent + # HEADERS blocks on the same stream are trailers and skip pseudo-header + # validation. + defp validate_incoming_headers( + %@for{mode: :server}, + %Stream{recv_headers: false}, + headers + ), + do: Request.validate_headers(headers, false) + + defp validate_incoming_headers( + %@for{mode: :client}, + %Stream{recv_headers: false}, + headers + ), + do: Response.validate_headers(headers, false) + + defp validate_incoming_headers(_protocol, _stream, _headers), do: :ok + end +end diff --git a/lib/ankh/protocol/http3/frame.ex b/lib/ankh/protocol/http3/frame.ex new file mode 100644 index 0000000..27438fd --- /dev/null +++ b/lib/ankh/protocol/http3/frame.ex @@ -0,0 +1,343 @@ +defmodule Ankh.Protocol.HTTP3.Frame do + @moduledoc """ + HTTP/3 frame codec (RFC 9114) with a struct-based API that mirrors + `Ankh.Protocol.HTTP2.Frame`. + + ## Design + + Every HTTP/3 frame type is represented as a dedicated module (e.g. + `Ankh.Protocol.HTTP3.Frame.Data`) that calls + + use Ankh.Protocol.HTTP3.Frame, type: 0xN, payload: PayloadModule + + The macro injects a struct with three fields: + + %Frame.Data{type: 0, length: 0, payload: %Frame.Data.Payload{}} + + * `type` — the integer frame type code, set at compile time. + * `length` — the payload byte count; filled in by `encode/1` and `decode/2`. + * `payload` — a struct implementing `Ankh.Protocol.HTTP3.Frame.Encodable`. + + This is intentionally analogous to `Ankh.Protocol.HTTP2.Frame`, which uses + the same `__using__` / `Encodable` / `stream` / `decode` / `encode` + structure. The main differences are: + + * HTTP/3 frames have **no flags byte** and **no stream-ID field** — those + concerns belong to the QUIC layer. + * Field lengths and frame-type codes are encoded as QUIC **variable-length + integers** (VLIs, RFC 9000 §16) rather than fixed-width fields. + + ## QUIC Variable-Length Integer (VLI) encoding + + The two most-significant bits of the first byte indicate the total length: + + | MSBs | Length | Value range | + |:----:|---------|---------------------------| + | `00` | 1 byte | 0 – 63 | + | `01` | 2 bytes | 0 – 16 383 | + | `10` | 4 bytes | 0 – 1 073 741 823 | + | `11` | 8 bytes | 0 – 4 611 686 018 427 387 903 | + + `encode_vli/1` always produces the smallest valid encoding; `decode_vli/1` + reads the two prefix bits to determine width. + + ## Wire format + + Frame { + Type (VLI) + Length (VLI) + Payload (Length bytes) + } + + `stream/1` lazily unfolds `{rest, {type_integer, payload_binary}}` pairs + from a buffer binary — identical in spirit to `HTTP2.Frame.stream/1`, which + unfolds `{rest, {length, type, id, data}}` tuples. When the buffer ends with + a partial frame a single `{data, nil}` element is emitted and the stream + then terminates (next accumulator is `<<>>`), so callers can safely use + `Enum.to_list/1` on buffers that contain only complete frames, and + `Enum.reduce_while/3` (halting on the nil token) for streaming buffers. + + `decode/2` converts a `{frame_struct, payload_binary}` pair into a fully + populated frame struct by delegating to `Encodable.decode/2`. + + `encode/1` turns a populated frame struct back into `{:ok, frame, iodata}`. + """ + + import Bitwise + + alias Ankh.Protocol.HTTP3.Frame.Encodable + + @typedoc "Frame type code (integer on the wire)." + @type type :: non_neg_integer() + + @typedoc "Payload byte count." + @type length :: non_neg_integer() + + @typedoc "Encoded frame data." + @type data :: iodata() + + @typedoc "The generic frame struct injected by `__using__/1`." + @type t :: struct() + + @doc """ + Injects the frame struct into the calling module. + + Options: + + * `:type` — (required) the integer HTTP/3 frame type code. + * `:payload` — (optional) the module whose struct implements + `Ankh.Protocol.HTTP3.Frame.Encodable`. Defaults to `nil`, which uses the + `Any` fallback (encodes/decodes as a no-op). + + The generated struct has fields: + + * `type` — fixed at the value given to `:type`. + * `length` — number of encoded payload bytes; set during encode/decode. + * `payload` — a struct (or `nil`) implementing `Encodable`. + + ## Example + + defmodule MyFrame do + defmodule Payload do + defstruct data: <<>> + defimpl Ankh.Protocol.HTTP3.Frame.Encodable do + def decode(%@for{} = p, bin), do: {:ok, %{p | data: bin}} + def encode(%@for{data: d}), do: {:ok, d} + end + end + + use Ankh.Protocol.HTTP3.Frame, type: 0x99, payload: Payload + end + + %MyFrame{type: 0x99, length: 0, payload: %MyFrame.Payload{}} + """ + @spec __using__(type: type(), payload: atom() | nil) :: Macro.t() + defmacro __using__(args) do + case Keyword.fetch(args, :type) do + {:ok, type} -> + payload = Keyword.get(args, :payload) + + quote bind_quoted: [type: type, payload: payload] do + alias Ankh.Protocol.HTTP3.Frame + + @typedoc """ + HTTP/3 frame struct. + + * `type` — wire integer for this frame type (compile-time constant). + * `length` — payload size in bytes; populated by `Frame.encode/1` and + `Frame.decode/2`. + * `payload` — decoded payload struct (implements `Frame.Encodable`), or + `nil` for frames with no payload. + """ + @type t :: %__MODULE__{ + type: Frame.type(), + length: Frame.length(), + payload: Frame.Encodable.t() | nil + } + + payload_default = if payload, do: struct(payload), else: nil + + defstruct type: type, + length: 0, + payload: payload_default + end + + :error -> + raise ArgumentError, + "#{__MODULE__}: `use Ankh.Protocol.HTTP3.Frame` requires a `:type` option" + end + end + + @doc """ + Returns a lazy stream of `{rest, frame_token}` pairs unfolded from `data`. + + Each `frame_token` is either: + + * `{type_integer, payload_binary}` — a complete frame extracted from the + buffer; `rest` is the remaining unconsumed binary. + * `nil` — the buffer ends with a partial (incomplete) frame; `rest` is the + full unconsumed binary that should be kept for the next delivery. + + The caller drives the stream with `Enum.reduce_while/3`, halting on `nil` + to store the remainder and continuing on each complete frame. + + This mirrors `Ankh.Protocol.HTTP2.Frame.stream/1`, which yields + `{rest, {length, type, id, data}}` tuples from a TCP buffer. + + ## Example + + iex> bin = + ...> IO.iodata_to_binary([ + ...> Ankh.Protocol.HTTP3.Frame.encode_vli(0x1), # HEADERS type + ...> Ankh.Protocol.HTTP3.Frame.encode_vli(3), # length + ...> "hdr", + ...> Ankh.Protocol.HTTP3.Frame.encode_vli(0x0), # DATA type + ...> Ankh.Protocol.HTTP3.Frame.encode_vli(2), # length + ...> "hi" + ...> ]) + iex> Ankh.Protocol.HTTP3.Frame.stream(bin) |> Enum.map(fn {_rest, token} -> token end) + [{1, "hdr"}, {0, "hi"}] + """ + @spec stream(binary()) :: Enumerable.t() + def stream(data) do + Stream.unfold(data, fn + <<>> -> + nil + + data -> + with {:ok, type, rest1} <- decode_vli(data), + {:ok, length, rest2} <- decode_vli(rest1), + true <- byte_size(rest2) >= length do + <> = rest2 + {{rest3, {type, payload}}, rest3} + else + # Partial frame: emit one nil token then terminate the stream so + # callers using Enum.to_list/1 don't loop forever. + _ -> {{data, nil}, <<>>} + end + end) + end + + @doc """ + Decodes `payload_binary` into the given frame struct by delegating to + `Encodable.decode/2` on the struct's `:payload` field. + + On success returns `{:ok, frame}` with `:length` set to + `byte_size(payload_binary)` and `:payload` replaced by the decoded struct. + + ## Example + + frame = %Ankh.Protocol.HTTP3.Frame.Data{} + {:ok, decoded} = Ankh.Protocol.HTTP3.Frame.decode(frame, "hello") + # decoded.payload.data == "hello" + # decoded.length == 5 + """ + @spec decode(t(), binary()) :: {:ok, t()} | {:error, any()} + def decode(frame, payload) when is_binary(payload) do + with {:ok, decoded_payload} <- Encodable.decode(frame.payload, payload) do + {:ok, %{frame | length: byte_size(payload), payload: decoded_payload}} + end + end + + @doc """ + Encodes a frame struct into its wire representation. + + Returns `{:ok, updated_frame, iodata}` where `iodata` is the complete + on-wire frame (type VLI + length VLI + payload bytes) and `updated_frame` + has its `:length` field set to the encoded payload byte count. + + ## Example + + frame = %Ankh.Protocol.HTTP3.Frame.Data{ + payload: %Ankh.Protocol.HTTP3.Frame.Data.Payload{data: "hi"} + } + {:ok, updated, iodata} = Ankh.Protocol.HTTP3.Frame.encode(frame) + # updated.length == 2 + # IO.iodata_to_binary(iodata) == <<0x00, 0x02, ?h, ?i>> + """ + @spec encode(t()) :: {:ok, t(), data()} | {:error, any()} + def encode(%{type: type, payload: payload} = frame) do + with {:ok, encoded_payload} <- Encodable.encode(payload) do + length = IO.iodata_length(encoded_payload) + + { + :ok, + %{frame | length: length}, + [encode_vli(type), encode_vli(length), encoded_payload] + } + end + end + + @doc """ + Encodes a non-negative integer as a QUIC variable-length integer. + + Always selects the smallest representation that fits the value. + + ## Examples + + iex> Ankh.Protocol.HTTP3.Frame.encode_vli(0) + <<0>> + + iex> Ankh.Protocol.HTTP3.Frame.encode_vli(63) + <<63>> + + iex> Ankh.Protocol.HTTP3.Frame.encode_vli(64) + <<0x40, 64>> + + iex> Ankh.Protocol.HTTP3.Frame.encode_vli(16_383) + <<0x7F, 0xFF>> + + iex> Ankh.Protocol.HTTP3.Frame.encode_vli(494) + <<0x41, 0xEE>> + """ + @spec encode_vli(non_neg_integer()) :: binary() + def encode_vli(n) when n < 64, do: <> + def encode_vli(n) when n < 16_384, do: <> + def encode_vli(n) when n < 1_073_741_824, do: <> + def encode_vli(n), do: <> + + @doc """ + Decodes one QUIC variable-length integer from the front of `data`. + + Returns `{:ok, value, rest}` on success, or `{:error, :incomplete}` when + there are not enough bytes for the width indicated by the prefix bits. + + ## Examples + + iex> Ankh.Protocol.HTTP3.Frame.decode_vli(<<0x41, 0xEE, "rest">>) + {:ok, 494, "rest"} + + iex> Ankh.Protocol.HTTP3.Frame.decode_vli(<<7>>) + {:ok, 7, ""} + + iex> Ankh.Protocol.HTTP3.Frame.decode_vli(<<>>) + {:error, :incomplete} + + iex> Ankh.Protocol.HTTP3.Frame.decode_vli(<<0x40>>) + {:error, :incomplete} + """ + @spec decode_vli(binary()) :: {:ok, non_neg_integer(), binary()} | {:error, :incomplete} + def decode_vli(<<0::2, n::6, rest::binary>>), do: {:ok, n, rest} + def decode_vli(<<1::2, n::14, rest::binary>>), do: {:ok, n, rest} + def decode_vli(<<2::2, n::30, rest::binary>>), do: {:ok, n, rest} + def decode_vli(<<3::2, n::62, rest::binary>>), do: {:ok, n, rest} + def decode_vli(_), do: {:error, :incomplete} + + @doc """ + Returns the binary that MUST be written immediately after opening an HTTP/3 + control stream (RFC 9114 §6.2.1). + + The preface consists of: + + 1. **Stream-type byte** `0x00` — marks this unidirectional QUIC stream as + an HTTP/3 control stream. + 2. **SETTINGS frame** — type `0x04`, 4-byte payload: + - `QPACK_MAX_TABLE_CAPACITY = 0` (`id = 0x01`, `value = 0x00`) + - `QPACK_BLOCKED_STREAMS = 0` (`id = 0x07`, `value = 0x00`) + + Advertising both settings at zero declares static-only QPACK mode, which + requires no dedicated encoder or decoder streams. + + ## Example + + iex> Ankh.Protocol.HTTP3.Frame.control_stream_preface() + <<0x00, 0x04, 0x04, 0x01, 0x00, 0x07, 0x00>> + """ + @spec control_stream_preface() :: binary() + def control_stream_preface do + << + # Stream-type byte (0x00) - HTTP/3 control stream. + 0x00, + # SETTINGS frame type (VLI 1-byte) + 0x04, + # SETTINGS payload length: 4 bytes (VLI 1-byte) + 0x04, + # QPACK_MAX_TABLE_CAPACITY = 0 (id=0x01, value=0x00) + 0x01, + 0x00, + # QPACK_BLOCKED_STREAMS = 0 (id=0x07, value=0x00) + 0x07, + 0x00 + >> + end +end diff --git a/lib/ankh/protocol/http3/frame/cancel_push.ex b/lib/ankh/protocol/http3/frame/cancel_push.ex new file mode 100644 index 0000000..bdc78a1 --- /dev/null +++ b/lib/ankh/protocol/http3/frame/cancel_push.ex @@ -0,0 +1,32 @@ +defmodule Ankh.Protocol.HTTP3.Frame.CancelPush do + @moduledoc false + + defmodule Payload do + @moduledoc false + + @type t :: %__MODULE__{push_id: non_neg_integer()} + defstruct push_id: 0 + + defimpl Ankh.Protocol.HTTP3.Frame.Encodable do + alias Ankh.Protocol.HTTP3.Frame + + def decode(%@for{} = payload, <<0::2, n::6, _::binary>>), + do: {:ok, %{payload | push_id: n}} + + def decode(%@for{} = payload, <<1::2, n::14, _::binary>>), + do: {:ok, %{payload | push_id: n}} + + def decode(%@for{} = payload, <<2::2, n::30, _::binary>>), + do: {:ok, %{payload | push_id: n}} + + def decode(%@for{} = payload, <<3::2, n::62, _::binary>>), + do: {:ok, %{payload | push_id: n}} + + def decode(_payload, _binary), do: {:error, :decode_error} + + def encode(%@for{push_id: id}), do: {:ok, Frame.encode_vli(id)} + end + end + + use Ankh.Protocol.HTTP3.Frame, type: 0x3, payload: Payload +end diff --git a/lib/ankh/protocol/http3/frame/data.ex b/lib/ankh/protocol/http3/frame/data.ex new file mode 100644 index 0000000..0c019b9 --- /dev/null +++ b/lib/ankh/protocol/http3/frame/data.ex @@ -0,0 +1,21 @@ +defmodule Ankh.Protocol.HTTP3.Frame.Data do + @moduledoc false + + defmodule Payload do + @moduledoc false + + @type t :: %__MODULE__{data: binary()} + defstruct data: <<>> + + defimpl Ankh.Protocol.HTTP3.Frame.Encodable do + def decode(%@for{} = payload, binary) when is_binary(binary), + do: {:ok, %{payload | data: binary}} + + def decode(_payload, _binary), do: {:error, :decode_error} + + def encode(%@for{data: data}), do: {:ok, data} + end + end + + use Ankh.Protocol.HTTP3.Frame, type: 0x0, payload: Payload +end diff --git a/lib/ankh/protocol/http3/frame/encodable.ex b/lib/ankh/protocol/http3/frame/encodable.ex new file mode 100644 index 0000000..3a12aca --- /dev/null +++ b/lib/ankh/protocol/http3/frame/encodable.ex @@ -0,0 +1,44 @@ +defprotocol Ankh.Protocol.HTTP3.Frame.Encodable do + @moduledoc """ + Protocol for encoding and decoding HTTP/3 frame payloads to and from their + wire binary representation. + + Every concrete HTTP/3 frame payload module (e.g. + `Ankh.Protocol.HTTP3.Frame.Data.Payload`) implements this protocol. + + Unlike the HTTP/2 equivalent (`Ankh.Protocol.HTTP2.Frame.Encodable`), + HTTP/3 frames carry no per-frame flags byte, so encoding and decoding + require no additional context beyond the payload struct itself and the raw + binary. + + The `@fallback_to_any true` declaration provides a no-op implementation for + `nil` payloads (frames whose payload field is not used). + """ + + @fallback_to_any true + + @typedoc "Any struct that implements `Ankh.Protocol.HTTP3.Frame.Encodable`." + @type t :: any() + + @doc """ + Decode `binary` into the payload struct `data`. + + Returns `{:ok, updated_struct}` on success or `{:error, reason}` on failure. + """ + @spec decode(t(), binary()) :: {:ok, t()} | {:error, any()} + def decode(data, binary) + + @doc """ + Encode the payload struct into an `iodata()` wire representation. + + Returns `{:ok, iodata}` on success or `{:error, reason}` on failure. + """ + @spec encode(t()) :: {:ok, iodata()} | {:error, any()} + def encode(data) +end + +defimpl Ankh.Protocol.HTTP3.Frame.Encodable, for: Any do + # Fallback for nil payload fields (frames with no payload struct). + def decode(_, _), do: {:ok, nil} + def encode(_), do: {:ok, <<>>} +end diff --git a/lib/ankh/protocol/http3/frame/go_away.ex b/lib/ankh/protocol/http3/frame/go_away.ex new file mode 100644 index 0000000..1aea76c --- /dev/null +++ b/lib/ankh/protocol/http3/frame/go_away.ex @@ -0,0 +1,32 @@ +defmodule Ankh.Protocol.HTTP3.Frame.GoAway do + @moduledoc false + + defmodule Payload do + @moduledoc false + + @type t :: %__MODULE__{stream_id: non_neg_integer()} + defstruct stream_id: 0 + + defimpl Ankh.Protocol.HTTP3.Frame.Encodable do + alias Ankh.Protocol.HTTP3.Frame + + def decode(%@for{} = payload, <<0::2, n::6, _::binary>>), + do: {:ok, %{payload | stream_id: n}} + + def decode(%@for{} = payload, <<1::2, n::14, _::binary>>), + do: {:ok, %{payload | stream_id: n}} + + def decode(%@for{} = payload, <<2::2, n::30, _::binary>>), + do: {:ok, %{payload | stream_id: n}} + + def decode(%@for{} = payload, <<3::2, n::62, _::binary>>), + do: {:ok, %{payload | stream_id: n}} + + def decode(_payload, _binary), do: {:error, :decode_error} + + def encode(%@for{stream_id: id}), do: {:ok, Frame.encode_vli(id)} + end + end + + use Ankh.Protocol.HTTP3.Frame, type: 0x7, payload: Payload +end diff --git a/lib/ankh/protocol/http3/frame/headers.ex b/lib/ankh/protocol/http3/frame/headers.ex new file mode 100644 index 0000000..99c1f3c --- /dev/null +++ b/lib/ankh/protocol/http3/frame/headers.ex @@ -0,0 +1,21 @@ +defmodule Ankh.Protocol.HTTP3.Frame.Headers do + @moduledoc false + + defmodule Payload do + @moduledoc false + + @type t :: %__MODULE__{hbf: binary()} + defstruct hbf: <<>> + + defimpl Ankh.Protocol.HTTP3.Frame.Encodable do + def decode(%@for{} = payload, binary) when is_binary(binary), + do: {:ok, %{payload | hbf: binary}} + + def decode(_payload, _binary), do: {:error, :decode_error} + + def encode(%@for{hbf: hbf}), do: {:ok, hbf} + end + end + + use Ankh.Protocol.HTTP3.Frame, type: 0x1, payload: Payload +end diff --git a/lib/ankh/protocol/http3/frame/max_push_id.ex b/lib/ankh/protocol/http3/frame/max_push_id.ex new file mode 100644 index 0000000..6765f2a --- /dev/null +++ b/lib/ankh/protocol/http3/frame/max_push_id.ex @@ -0,0 +1,32 @@ +defmodule Ankh.Protocol.HTTP3.Frame.MaxPushId do + @moduledoc false + + defmodule Payload do + @moduledoc false + + @type t :: %__MODULE__{push_id: non_neg_integer()} + defstruct push_id: 0 + + defimpl Ankh.Protocol.HTTP3.Frame.Encodable do + alias Ankh.Protocol.HTTP3.Frame + + def decode(%@for{} = payload, <<0::2, n::6, _::binary>>), + do: {:ok, %{payload | push_id: n}} + + def decode(%@for{} = payload, <<1::2, n::14, _::binary>>), + do: {:ok, %{payload | push_id: n}} + + def decode(%@for{} = payload, <<2::2, n::30, _::binary>>), + do: {:ok, %{payload | push_id: n}} + + def decode(%@for{} = payload, <<3::2, n::62, _::binary>>), + do: {:ok, %{payload | push_id: n}} + + def decode(_payload, _binary), do: {:error, :decode_error} + + def encode(%@for{push_id: id}), do: {:ok, Frame.encode_vli(id)} + end + end + + use Ankh.Protocol.HTTP3.Frame, type: 0xD, payload: Payload +end diff --git a/lib/ankh/protocol/http3/frame/push_promise.ex b/lib/ankh/protocol/http3/frame/push_promise.ex new file mode 100644 index 0000000..8c02283 --- /dev/null +++ b/lib/ankh/protocol/http3/frame/push_promise.ex @@ -0,0 +1,33 @@ +defmodule Ankh.Protocol.HTTP3.Frame.PushPromise do + @moduledoc false + + defmodule Payload do + @moduledoc false + + @type t :: %__MODULE__{push_id: non_neg_integer(), hbf: binary()} + defstruct push_id: 0, hbf: <<>> + + defimpl Ankh.Protocol.HTTP3.Frame.Encodable do + alias Ankh.Protocol.HTTP3.Frame + + def decode(%@for{} = payload, <<0::2, id::6, hbf::binary>>), + do: {:ok, %{payload | push_id: id, hbf: hbf}} + + def decode(%@for{} = payload, <<1::2, id::14, hbf::binary>>), + do: {:ok, %{payload | push_id: id, hbf: hbf}} + + def decode(%@for{} = payload, <<2::2, id::30, hbf::binary>>), + do: {:ok, %{payload | push_id: id, hbf: hbf}} + + def decode(%@for{} = payload, <<3::2, id::62, hbf::binary>>), + do: {:ok, %{payload | push_id: id, hbf: hbf}} + + def decode(_payload, _binary), do: {:error, :decode_error} + + def encode(%@for{push_id: id, hbf: hbf}), + do: {:ok, [Frame.encode_vli(id), hbf]} + end + end + + use Ankh.Protocol.HTTP3.Frame, type: 0x5, payload: Payload +end diff --git a/lib/ankh/protocol/http3/frame/settings.ex b/lib/ankh/protocol/http3/frame/settings.ex new file mode 100644 index 0000000..1eda21d --- /dev/null +++ b/lib/ankh/protocol/http3/frame/settings.ex @@ -0,0 +1,63 @@ +defmodule Ankh.Protocol.HTTP3.Frame.Settings do + @moduledoc false + + defmodule Payload do + @moduledoc false + + @type setting :: {non_neg_integer(), non_neg_integer()} + @type t :: %__MODULE__{settings: [setting()]} + defstruct settings: [] + + defimpl Ankh.Protocol.HTTP3.Frame.Encodable do + alias Ankh.Protocol.HTTP3.Frame + + def decode(%@for{} = payload, binary) when is_binary(binary) do + case do_decode(binary, []) do + {:ok, pairs} -> {:ok, %{payload | settings: pairs}} + error -> error + end + end + + def decode(_payload, _binary), do: {:error, :decode_error} + + defp do_decode(<<>>, acc), do: {:ok, Enum.reverse(acc)} + + defp do_decode(<<0::2, id::6, rest::binary>>, acc), + do: do_decode_value(id, rest, acc) + + defp do_decode(<<1::2, id::14, rest::binary>>, acc), + do: do_decode_value(id, rest, acc) + + defp do_decode(<<2::2, id::30, rest::binary>>, acc), + do: do_decode_value(id, rest, acc) + + defp do_decode(<<3::2, id::62, rest::binary>>, acc), + do: do_decode_value(id, rest, acc) + + defp do_decode(_, _acc), do: {:error, :decode_error} + + defp do_decode_value(id, <<0::2, v::6, rest::binary>>, acc), + do: do_decode(rest, [{id, v} | acc]) + + defp do_decode_value(id, <<1::2, v::14, rest::binary>>, acc), + do: do_decode(rest, [{id, v} | acc]) + + defp do_decode_value(id, <<2::2, v::30, rest::binary>>, acc), + do: do_decode(rest, [{id, v} | acc]) + + defp do_decode_value(id, <<3::2, v::62, rest::binary>>, acc), + do: do_decode(rest, [{id, v} | acc]) + + defp do_decode_value(_id, _rest, _acc), do: {:error, :decode_error} + + def encode(%@for{settings: settings}) do + {:ok, + Enum.map(settings, fn {id, value} -> + [Frame.encode_vli(id), Frame.encode_vli(value)] + end)} + end + end + end + + use Ankh.Protocol.HTTP3.Frame, type: 0x4, payload: Payload +end diff --git a/lib/ankh/protocol/http3/stream.ex b/lib/ankh/protocol/http3/stream.ex new file mode 100644 index 0000000..dd2fb6c --- /dev/null +++ b/lib/ankh/protocol/http3/stream.ex @@ -0,0 +1,374 @@ +defmodule Ankh.Protocol.HTTP3.Stream do + @moduledoc """ + Per-QUIC-stream state tracker for HTTP/3 (RFC 9114). + + Each QUIC stream used by the HTTP/3 layer is represented by one + `#{__MODULE__}` struct. The struct travels through a well-defined lifecycle: + + ## Lifecycle + + ### Creation + + Use `new_request/1` when the local side opens a new bidirectional request + stream (client-initiated). Use `new_incoming/2` when a stream arrives from + the remote peer; pass `unidirectional? = false` for a bidirectional request + stream and `unidirectional? = true` for a unidirectional stream whose kind + is not yet known. + + Both functions assign a fresh Elixir `reference()` (via `make_ref/0`) that + can be used to correlate the stream across the rest of the connection. + + ### Unidirectional stream kind identification + + Unidirectional streams start with kind `:unknown_unidirectional`. When the + first byte of the stream body is received, call `identify_kind/2` with that + byte to resolve the kind: + + * `0x00` → `:control` (HTTP/3 control stream, RFC 9114 §6.2.1) + * `0x01` → `:push` (server-push stream, RFC 9114 §4.6) + * `0x02` → `:qpack_encoder` (QPACK encoder stream, RFC 9204 §4.2) + * `0x03` → `:qpack_decoder` (QPACK decoder stream, RFC 9204 §4.2) + * anything else → `:unknown_unidirectional` (must be ignored per RFC 9114 §6.2) + + `identify_kind/2` is a no-op when the stream already has kind `:request`. + + ### Incoming data + + Raw bytes received from the QUIC layer are accumulated with `append/2`. The + frame parser (see `Ankh.Protocol.HTTP3.Frame`) should consume the buffer and + pass any unconsumed remainder back to `consume_buffer/2` so it is retained + for the next delivery. + + ### FIN / half-close handling + + QUIC streams support independent half-closes on each side: + + * `remote_fin/1` — call when the peer signals end-of-stream (it has + finished sending). Transitions `:open → :half_closed_remote` and + `:half_closed_local → :closed`. + * `local_fin/1` — call after sending a FIN (e.g. via + `Ankh.Transport.QUIC.shutdown_stream/1`). Transitions + `:open → :half_closed_local` and `:half_closed_remote → :closed`. + + States that are already past the relevant half-close are left unchanged. + + ### Terminal state + + `closed?/1` returns `true` once the stream reaches `:closed`. + `remote_open?/1` returns `true` while the remote side has not yet sent a FIN + (states `:open` and `:half_closed_local`), indicating that more data or + headers may still arrive. + + ## Frame parsing + + This module deliberately does **not** parse HTTP/3 frames. Frame encoding + and decoding belong in `Ankh.Protocol.HTTP3.Frame`. + """ + + @typedoc """ + HTTP/3 stream states, mirroring the QUIC stream half-close model. + + * `:idle` — stream struct created but not yet active. + * `:open` — both sides may send and receive. + * `:half_closed_local` — local side has sent FIN; remote may still send. + * `:half_closed_remote` — remote side has sent FIN; local may still send. + * `:closed` — both sides have finished; no further data. + """ + @type state :: + :idle + | :open + | :half_closed_local + | :half_closed_remote + | :closed + + @typedoc """ + Categorises a QUIC stream by its HTTP/3 role. + + * `:request` — bidirectional request/response stream + (RFC 9114 §6.1). + * `:control` — unidirectional HTTP/3 control stream, + stream-type byte `0x00` (RFC 9114 §6.2.1). + * `:push` — unidirectional server-push stream, + stream-type byte `0x01` (RFC 9114 §4.6). + * `:qpack_encoder` — unidirectional QPACK encoder stream, + stream-type byte `0x02` (RFC 9204 §4.2). + * `:qpack_decoder` — unidirectional QPACK decoder stream, + stream-type byte `0x03` (RFC 9204 §4.2). + * `:unknown_unidirectional` — unidirectional stream whose type byte has + not yet been received or is not a + recognised value. + * `:unknown` — kind not yet determined (initial value). + """ + @type stream_kind :: + :request + | :control + | :qpack_encoder + | :qpack_decoder + | :push + | :unknown_unidirectional + | :unknown + + @typedoc "Per-QUIC-stream HTTP/3 state struct." + @type t :: %__MODULE__{ + handle: non_neg_integer() | nil, + reference: reference() | nil, + state: state(), + kind: stream_kind(), + buffer: binary(), + recv_headers: boolean(), + send_headers: boolean() + } + + defstruct handle: nil, + reference: nil, + state: :idle, + kind: :unknown, + buffer: <<>>, + recv_headers: false, + send_headers: false + + @doc """ + Creates a new client-opened request stream for the given QUIC stream handle. + + The returned struct has: + + * `kind` = `:request` + * `state` = `:open` + * `reference` = a fresh `reference()` from `make_ref/0` + + ## Example + + iex> stream = #{__MODULE__}.new_request(some_handle) + iex> stream.kind + :request + iex> stream.state + :open + """ + @spec new_request(handle :: non_neg_integer()) :: t() + def new_request(handle) do + %__MODULE__{ + handle: handle, + reference: make_ref(), + state: :open, + kind: :request + } + end + + @doc """ + Creates a new server-received (incoming) stream for the given QUIC stream handle. + + When `unidirectional?` is `true`, the kind is set to `:unknown_unidirectional` + because the stream-type byte has not yet arrived. When `unidirectional?` is + `false`, the stream is a bidirectional request stream and the kind is set to + `:request` immediately. + + In both cases the state is `:open` and a fresh `reference()` is assigned. + + ## Examples + + iex> stream = #{__MODULE__}.new_incoming(some_handle, false) + iex> stream.kind + :request + + iex> stream = #{__MODULE__}.new_incoming(some_handle, true) + iex> stream.kind + :unknown_unidirectional + """ + @spec new_incoming(handle :: non_neg_integer(), unidirectional? :: boolean()) :: t() + def new_incoming(handle, false = _unidirectional?) do + %__MODULE__{ + handle: handle, + reference: make_ref(), + state: :open, + kind: :request + } + end + + def new_incoming(handle, true = _unidirectional?) do + %__MODULE__{ + handle: handle, + reference: make_ref(), + state: :open, + kind: :unknown_unidirectional + } + end + + @doc """ + Resolves the kind of a unidirectional stream from its first stream-type byte. + + The mapping from wire byte to kind atom is defined by RFC 9114 §6.2 and + RFC 9204 §4.2: + + * `0x00` → `:control` + * `0x01` → `:push` + * `0x02` → `:qpack_encoder` + * `0x03` → `:qpack_decoder` + * other → `:unknown_unidirectional` (extension / reserved stream type) + + This function is a **no-op** when the stream kind is already `:request`; + request streams are bidirectional and do not carry a stream-type prefix. + + ## Examples + + iex> stream = #{__MODULE__}.new_incoming(handle, true) + iex> #{__MODULE__}.identify_kind(stream, 0x00).kind + :control + + iex> #{__MODULE__}.identify_kind(stream, 0x02).kind + :qpack_encoder + + iex> #{__MODULE__}.identify_kind(stream, 0xFF).kind + :unknown_unidirectional + """ + @spec identify_kind(t(), stream_type_byte :: non_neg_integer()) :: t() + def identify_kind(%__MODULE__{kind: :request} = stream, _type_byte), do: stream + def identify_kind(%__MODULE__{} = stream, 0x00), do: %{stream | kind: :control} + def identify_kind(%__MODULE__{} = stream, 0x01), do: %{stream | kind: :push} + def identify_kind(%__MODULE__{} = stream, 0x02), do: %{stream | kind: :qpack_encoder} + def identify_kind(%__MODULE__{} = stream, 0x03), do: %{stream | kind: :qpack_decoder} + + def identify_kind(%__MODULE__{} = stream, _type_byte), + do: %{stream | kind: :unknown_unidirectional} + + @doc """ + Appends `data` to the stream's receive buffer. + + Call this each time the QUIC layer delivers bytes for this stream. The + accumulated buffer is then available for the frame parser to consume. + + ## Example + + iex> stream = #{__MODULE__}.new_request(handle) + iex> stream = #{__MODULE__}.append(stream, <<0x01, 0x05>>) + iex> stream.buffer + <<0x01, 0x05>> + iex> stream = #{__MODULE__}.append(stream, <<"hello">>) + iex> stream.buffer + <<0x01, 0x05, "hello">> + """ + @spec append(t(), binary()) :: t() + def append(%__MODULE__{buffer: buffer} = stream, data) when is_binary(data), + do: %{stream | buffer: buffer <> data} + + @doc """ + Replaces the stream buffer with `remainder`. + + Call this after the frame parser has consumed one or more complete frames + from `stream.buffer`, passing whatever bytes were not consumed. This keeps + partial frames available for the next delivery without re-appending the + already-parsed bytes. + + ## Example + + iex> stream = #{__MODULE__}.append(stream, <<0x00, 0x03, "abc", 0xFF>>) + iex> stream = #{__MODULE__}.consume_buffer(stream, <<0xFF>>) + iex> stream.buffer + <<0xFF>> + """ + @spec consume_buffer(t(), binary()) :: t() + def consume_buffer(%__MODULE__{} = stream, remainder) when is_binary(remainder), + do: %{stream | buffer: remainder} + + @doc """ + Records that the remote peer has finished sending on this stream. + + This is typically triggered when the peer sets `fin=true` on the final + `{:quic, conn, {:stream_data, stream_id, data, true}}` message. + + State transitions: + + * `:open` → `:half_closed_remote` + * `:half_closed_local` → `:closed` + * all other states → unchanged + + ## Examples + + iex> stream = #{__MODULE__}.new_request(handle) + iex> #{__MODULE__}.remote_fin(stream).state + :half_closed_remote + + iex> stream = %{stream | state: :half_closed_local} + iex> #{__MODULE__}.remote_fin(stream).state + :closed + """ + @spec remote_fin(t()) :: t() + def remote_fin(%__MODULE__{state: :open} = stream), + do: %{stream | state: :half_closed_remote} + + def remote_fin(%__MODULE__{state: :half_closed_local} = stream), + do: %{stream | state: :closed} + + def remote_fin(%__MODULE__{} = stream), do: stream + + @doc """ + Records that the local side has finished sending on this stream. + + Call this after successfully writing a FIN to the QUIC layer (e.g. via + `Ankh.Transport.QUIC.shutdown_stream/1`). + + State transitions: + + * `:open` → `:half_closed_local` + * `:half_closed_remote` → `:closed` + * all other states → unchanged + + ## Examples + + iex> stream = #{__MODULE__}.new_request(handle) + iex> #{__MODULE__}.local_fin(stream).state + :half_closed_local + + iex> stream = %{stream | state: :half_closed_remote} + iex> #{__MODULE__}.local_fin(stream).state + :closed + """ + @spec local_fin(t()) :: t() + def local_fin(%__MODULE__{state: :open} = stream), + do: %{stream | state: :half_closed_local} + + def local_fin(%__MODULE__{state: :half_closed_remote} = stream), + do: %{stream | state: :closed} + + def local_fin(%__MODULE__{} = stream), do: stream + + @doc """ + Returns `true` when the stream has reached the `:closed` state. + + A closed stream has had both its send and receive sides terminated; no + further data or frames should be processed on it. + + ## Example + + iex> stream = #{__MODULE__}.new_request(handle) + iex> #{__MODULE__}.closed?(stream) + false + iex> stream = stream |> #{__MODULE__}.remote_fin() |> #{__MODULE__}.local_fin() + iex> #{__MODULE__}.closed?(stream) + true + """ + @spec closed?(t()) :: boolean() + def closed?(%__MODULE__{state: :closed}), do: true + def closed?(%__MODULE__{}), do: false + + @doc """ + Returns `true` when the remote side of the stream is still open. + + The remote side is considered open in states `:open` and + `:half_closed_local`, meaning the peer has not yet sent a FIN and may still + deliver headers, data, or other frames. + + ## Examples + + iex> stream = #{__MODULE__}.new_request(handle) + iex> #{__MODULE__}.remote_open?(stream) + true + + iex> #{__MODULE__}.remote_open?(#{__MODULE__}.remote_fin(stream)) + false + """ + @spec remote_open?(t()) :: boolean() + def remote_open?(%__MODULE__{state: state}) when state in [:open, :half_closed_local], + do: true + + def remote_open?(%__MODULE__{}), do: false +end diff --git a/lib/ankh/transport/quic.ex b/lib/ankh/transport/quic.ex new file mode 100644 index 0000000..aa7d7f4 --- /dev/null +++ b/lib/ankh/transport/quic.ex @@ -0,0 +1,319 @@ +defmodule Ankh.Transport.QUIC do + @moduledoc """ + QUIC transport implementation using the `erlang_quic` library. + + This transport uses `erlang_quic` (https://github.com/benoitc/erlang_quic), a + pure-Erlang QUIC implementation (RFC 9000 / 9001). + + ## Differences from TCP/TLS + + Unlike TCP or TLS transports, which expose a single bytestream socket, QUIC is + inherently multiplexed. The QUIC transport therefore manages two distinct handles: + + * A **connection handle** (`t:connection/0`) — the underlying QUIC connection + (a `pid()`). + * A **stream handle** (`t:stream/0`) — a non-negative integer identifying one + bidirectional QUIC stream used for data exchange within that connection. + + For the client path, both handles are established during `connect/4`. For the + server path, the connection handle is set via `new/2` (typically after accepting + a raw QUIC connection from a listener), and the stream handle is obtained by + calling `accept/2`, which waits for the remote peer to open a stream. + + ## Message delivery + + `erlang_quic` delivers all QUIC events as messages to the owning process + automatically — there is no `active: :once` mode to re-arm. After each message + the process can simply call `handle_msg/2` to decode it. + + ## ALPN + + The default ALPN token is `"h3"` (HTTP/3) as a binary string. + + ## Dependencies + + Requires the `quic` optional dependency (erlang_quic) to be present and started. + """ + + alias Ankh.Transport + + @typedoc "QUIC connection handle (pid returned by erlang_quic)" + @type connection :: pid() + + @typedoc "QUIC stream identifier (non-negative integer returned by erlang_quic)" + @type stream :: non_neg_integer() + + @type t :: %__MODULE__{ + connection: connection() | nil, + stream: stream() | nil, + alpn: String.t() | nil + } + + defstruct connection: nil, stream: nil, alpn: nil + + @doc """ + Opens a new bidirectional QUIC stream on the connection and returns a + transport struct with that stream handle set. + + Used by `Ankh.Protocol.HTTP3.request/2` to obtain a fresh stream for each + HTTP/3 request. + """ + @spec open_stream(t()) :: {:ok, t()} | {:error, any()} + def open_stream(%__MODULE__{connection: conn} = transport) when not is_nil(conn) do + case :quic.open_stream(conn) do + {:ok, stream_id} -> {:ok, %{transport | stream: stream_id}} + {:error, _} = error -> error + end + end + + @doc """ + Opens a new **unidirectional** QUIC stream on the connection and immediately + writes `preface` on it. + + Used by the HTTP/3 protocol to set up the control stream immediately after + a connection is established. + """ + @spec open_unidirectional_stream(t(), iodata()) :: :ok | {:error, any()} + def open_unidirectional_stream(%__MODULE__{connection: conn}, preface) + when not is_nil(conn) do + with {:ok, stream_id} <- :quic.open_unidirectional_stream(conn) do + :quic.send_data(conn, stream_id, preface, false) + end + end + + @doc """ + Gracefully shuts down the **send side** of the stored stream (QUIC FIN), + signalling to the peer that no more data will be sent. + + This does **not** close the connection. + """ + @spec shutdown_stream(t()) :: :ok | {:error, any()} + def shutdown_stream(%__MODULE__{connection: conn, stream: stream}) + when not is_nil(conn) and not is_nil(stream) do + :quic.send_data(conn, stream, <<>>, true) + end + + defimpl Transport do + @doc """ + Initialises the transport from an existing QUIC handle. + + Accepts either: + + * A bare `connection()` (pid) — used on the server path where the + connection has already been accepted from a listener. The stream field is + left `nil` and must be populated via `accept/2` before data can flow. + + * A `{connection(), stream()}` tuple — used when both handles are already + known. + """ + def new(%@for{} = transport, {connection, stream}) + when is_pid(connection) and is_integer(stream), + do: {:ok, %{transport | connection: connection, stream: stream}} + + def new(%@for{} = transport, connection) when is_pid(connection), + do: {:ok, %{transport | connection: connection}} + + @doc """ + Opens a QUIC connection to the given `uri`. + + `erlang_quic`'s connect is asynchronous: `:quic.connect/4` returns + `{:ok, conn_pid}` immediately and the calling process then receives + `{:quic, conn_pid, {:connected, info}}` when the TLS handshake completes. + This function blocks internally (using `receive … after timeout`) to present + a synchronous interface to the caller. + + Returns a transport with only the `connection` field set (`stream: nil`). + Streams are not opened automatically; use `Ankh.Transport.QUIC.open_stream/1` + or `Ankh.Transport.QUIC.open_unidirectional_stream/2` at the protocol layer + to create streams on demand. + + Supported options (all optional): + + * `:alpn` — list of ALPN tokens as binaries; defaults to `["h3"]`. + * `:verify` — peer certificate verification; pass `true` or `:verify_peer` + to verify. Defaults to `false`. + * `:cacertfile` — path to a CA certificate bundle. + * `:certfile` — path to a client certificate (mTLS). + * `:keyfile` — path to the private key for the client certificate. + * `:password` — passphrase for an encrypted private key. + """ + def connect(%@for{} = transport, %URI{host: host, port: port}, timeout, options) do + alpn = Keyword.get(options, :alpn, ["h3"]) + + verify = + case Keyword.get(options, :verify, false) do + :verify_peer -> true + :verify_none -> false + verify when is_boolean(verify) -> verify + _verify -> false + end + + conn_opts = + %{alpn: alpn, verify: verify} + |> maybe_put(:cacertfile, Keyword.get(options, :cacertfile)) + |> maybe_put(:certfile, Keyword.get(options, :certfile)) + |> maybe_put(:keyfile, Keyword.get(options, :keyfile)) + |> maybe_put(:password, Keyword.get(options, :password)) + + case :quic.connect(host, port || 443, conn_opts, self()) do + {:ok, conn_pid} -> + receive do + {:quic, ^conn_pid, {:connected, _info}} -> + negotiated = List.first(alpn, "h3") + {:ok, %{transport | connection: conn_pid, alpn: negotiated}} + + {:quic, ^conn_pid, {:closed, reason}} -> + {:error, reason} + + {:quic, ^conn_pid, {:transport_error, _code, reason}} -> + {:error, reason} + after + timeout -> + {:error, :timeout} + end + + {:error, reason} -> + {:error, reason} + end + end + + defp maybe_put(map, _key, nil), do: map + defp maybe_put(map, key, value), do: Map.put(map, key, value) + + @doc """ + Accepts an inbound QUIC stream on an already-established connection. + + The connection handle must have been set beforehand via `new/2`. This call + blocks until the remote peer opens a stream (delivering + `{:quic, conn, {:stream_opened, stream_id}}`) or until the 5-second timeout + expires. + + Note: in HTTP/3 mode streams arrive as asynchronous messages handled by + `Ankh.Protocol.HTTP3.stream/2`, so this function is primarily used for + single-stream protocols (HTTP/1.1, HTTP/2) over QUIC. + """ + def accept(%@for{connection: conn} = transport, _options) when not is_nil(conn) do + receive do + {:quic, ^conn, {:stream_opened, stream_id}} -> + {:ok, %{transport | stream: stream_id}} + after + 5_000 -> + {:error, :timeout} + end + end + + def accept(%@for{connection: nil}, _options), do: {:error, :no_connection} + + @doc """ + Sends `data` over the QUIC stream synchronously. + + Calls `:quic.send_data/4` with `fin = false`. + """ + def send(%@for{connection: conn, stream: stream}, data) + when not is_nil(conn) and not is_nil(stream), + do: :quic.send_data(conn, stream, data, false) + + def send(%@for{stream: nil}, _data), do: {:error, :no_stream} + + @doc """ + Passively reads data from the QUIC stream by blocking on the next + `{:quic, conn, {:stream_data, stream_id, data, _fin}}` message. + + The `size` argument is accepted for interface compatibility but is not + enforced — `erlang_quic` does not support partial reads. + """ + def recv(%@for{connection: conn, stream: stream}, _size, timeout) + when not is_nil(conn) and not is_nil(stream) do + receive do + {:quic, ^conn, {:stream_data, ^stream, data, _fin}} -> + {:ok, data} + after + timeout -> + {:error, :timeout} + end + end + + def recv(%@for{stream: nil}, _size, _timeout), do: {:error, :no_stream} + + @doc """ + Closes the underlying QUIC connection. + + Unlike the quicer-based implementation there is no separate stream-close + call; `:quic.close/1` tears down the whole connection. + """ + def close(%@for{connection: conn} = transport) do + if not is_nil(conn), do: :quic.close(conn) + {:ok, %{transport | connection: nil, stream: nil}} + end + + @doc """ + Handles asynchronous messages delivered by `erlang_quic`. + + ## Handled messages + + * `{:quic, conn, {:stream_data, stream_id, data, false}}` — binary data + arrived on our stream. Returns `{:ok, data}`. + * `{:quic, conn, {:stream_data, stream_id, data, true}}` — final data + chunk with FIN. Returns `{:ok, data}` if non-empty, `{:error, :closed}` + otherwise. + * `{:quic, conn, {:stream_reset, stream_id, _}}` — stream reset by peer. + * `{:quic, conn, {:closed, _}}` — connection closed. + * `{:quic, conn, {:transport_error, _, _}}` — transport-level error. + + All other messages return `{:other, msg}` so the caller can handle them. + """ + + # HTTP/3 passthrough mode: when no specific stream is stored, the QUIC + # transport is operating as a bare connection handle on behalf of the + # HTTP/3 protocol layer. Pass every QUIC message through as-is so that + # `Ankh.Protocol.HTTP3.stream/2` can dispatch to the correct per-request + # stream state machine. + def handle_msg(%@for{stream: nil}, {:quic, _, _} = msg), do: {:ok, msg} + + # Binary data on our stream, no FIN. + def handle_msg( + %@for{connection: conn, stream: stream}, + {:quic, conn, {:stream_data, stream, data, false}} + ) + when is_binary(data) and not is_nil(stream), do: {:ok, data} + + # Final data chunk with FIN. + def handle_msg( + %@for{connection: conn, stream: stream}, + {:quic, conn, {:stream_data, stream, data, true}} + ) + when is_binary(data) and not is_nil(stream) do + if data != <<>>, do: {:ok, data}, else: {:error, :closed} + end + + # Stream reset by peer. + def handle_msg( + %@for{connection: conn, stream: stream}, + {:quic, conn, {:stream_reset, stream, _error_code}} + ) + when not is_nil(stream), do: {:error, :closed} + + # Connection closed. + def handle_msg(%@for{connection: conn}, {:quic, conn, {:closed, _reason}}) + when not is_nil(conn), do: {:error, :closed} + + # Transport error. + def handle_msg(%@for{connection: conn}, {:quic, conn, {:transport_error, _code, _reason}}) + when not is_nil(conn), do: {:error, :closed} + + # Unrelated message — pass it through for the caller to handle. + def handle_msg(_quic, msg), do: {:other, msg} + + @doc """ + Returns the ALPN protocol stored during `connect/4`, or `"h3"` if this + transport was initialised on the server side without going through `connect/4`. + + `erlang_quic` does not expose a `negotiated_protocol/1` query function; + the negotiated ALPN is inferred from the first token in the requested list + (connection succeeds only if the server agrees). + """ + def negotiated_protocol(%@for{alpn: alpn}) when not is_nil(alpn), do: {:ok, alpn} + def negotiated_protocol(%@for{connection: conn}) when not is_nil(conn), do: {:ok, "h3"} + def negotiated_protocol(%@for{connection: nil}), do: {:error, :protocol_not_negotiated} + end +end diff --git a/mix.exs b/mix.exs index fb13dca..034d491 100644 --- a/mix.exs +++ b/mix.exs @@ -38,6 +38,7 @@ defmodule Ankh.Mixfile do {:castore, "~> 1.0"}, {:hpax, "~> 1.0"}, {:plug, "~> 1.0"}, + {:quic, "~> 1.0"}, {:credo, "~> 1.0", only: [:dev], runtime: false}, {:ex_doc, "~> 0.40.0", only: [:dev], runtime: false}, {:dialyxir, "~> 1.0", only: [:dev], runtime: false} @@ -49,6 +50,7 @@ defmodule Ankh.Mixfile do groups_for_modules: [ HTTP1: [~r/^Ankh\.Protocol\.HTTP1\.*/], HTTP2: [~r/^Ankh\.Protocol\.HTTP2\.*/], + HTTP3: [~r/^Ankh\.Protocol\.HTTP3\.*/], Internals: [~r/^Ankh\.Protocol$/, ~r/^Ankh\.Transport$/], Transports: [~r/Ankh\.Transport\.*/] ] diff --git a/mix.lock b/mix.lock index 11ecf22..67dc103 100644 --- a/mix.lock +++ b/mix.lock @@ -16,5 +16,6 @@ "nimble_parsec": {:hex, :nimble_parsec, "1.4.2", "8efba0122db06df95bfaa78f791344a89352ba04baedd3849593bfce4d0dc1c6", [:mix], [], "hexpm", "4b21398942dda052b403bbe1da991ccd03a053668d147d53fb8c4e0efe09c973"}, "plug": {:hex, :plug, "1.19.2", "e4950525b22c6789dfb38a3f95d47171ba159da3fc5a33be9643b43d5e8adb98", [:mix], [{:mime, "~> 1.0 or ~> 2.0", [hex: :mime, repo: "hexpm", optional: false]}, {:plug_crypto, "~> 1.1.1 or ~> 1.2 or ~> 2.0", [hex: :plug_crypto, repo: "hexpm", optional: false]}, {:telemetry, "~> 0.4.3 or ~> 1.0", [hex: :telemetry, repo: "hexpm", optional: false]}], "hexpm", "b6fce20a56af5e60fa5dfecf3f907bb98ec981be43c79a3809a499bc3d133de0"}, "plug_crypto": {:hex, :plug_crypto, "2.1.1", "19bda8184399cb24afa10be734f84a16ea0a2bc65054e23a62bb10f06bc89491", [:mix], [], "hexpm", "6470bce6ffe41c8bd497612ffde1a7e4af67f36a15eea5f921af71cf3e11247c"}, + "quic": {:hex, :quic, "1.6.5", "28b7d49c1732b2de3861701bb03ae7798537240552370ac49f0d139d2ea8f621", [:rebar3], [], "hexpm", "de1a88972c33201a50d1a17c8c4a14528bd1d8f25ef705897d680ae312d0aa78"}, "telemetry": {:hex, :telemetry, "1.4.2", "a0cb522801dffb1c49fe6e30561badffc7b6d0e180db1300df759faa22062855", [:rebar3], [], "hexpm", "928f6495066506077862c0d1646609eed891a4326bee3126ba54b60af61febb1"}, } diff --git a/test/ankh/protocol/http3/frame_test.exs b/test/ankh/protocol/http3/frame_test.exs new file mode 100644 index 0000000..c1a172d --- /dev/null +++ b/test/ankh/protocol/http3/frame_test.exs @@ -0,0 +1,511 @@ +defmodule Ankh.Protocol.HTTP3.FrameTest do + use ExUnit.Case, async: true + + import Bitwise + + doctest Ankh.Protocol.HTTP3.Frame + + alias Ankh.Protocol.HTTP3.Frame + + alias Ankh.Protocol.HTTP3.Frame.{ + CancelPush, + Data, + GoAway, + Headers, + MaxPushId, + PushPromise, + Settings + } + + describe "encode_vli/1" do + test "encodes zero as a single byte" do + assert Frame.encode_vli(0) == <<0x00>> + end + + test "encodes the 1-byte boundary value 63" do + assert Frame.encode_vli(63) == <<63>> + end + + test "encodes 64 in two bytes with 01 prefix" do + <> = Frame.encode_vli(64) + assert msb >>> 6 == 1 + end + + test "encodes 16_383 in two bytes" do + encoded = Frame.encode_vli(16_383) + assert byte_size(encoded) == 2 + assert {:ok, 16_383, <<>>} = Frame.decode_vli(encoded) + end + + test "encodes 16_384 in four bytes with 10 prefix" do + encoded = Frame.encode_vli(16_384) + assert byte_size(encoded) == 4 + <> = encoded + assert msb >>> 6 == 2 + end + + test "encodes 1_073_741_823 in four bytes" do + encoded = Frame.encode_vli(1_073_741_823) + assert byte_size(encoded) == 4 + assert {:ok, 1_073_741_823, <<>>} = Frame.decode_vli(encoded) + end + + test "encodes 1_073_741_824 in eight bytes with 11 prefix" do + encoded = Frame.encode_vli(1_073_741_824) + assert byte_size(encoded) == 8 + <> = encoded + assert msb >>> 6 == 3 + end + + test "round-trips a large 8-byte value" do + n = 4_611_686_018_427_387_903 + encoded = Frame.encode_vli(n) + assert byte_size(encoded) == 8 + assert {:ok, ^n, <<>>} = Frame.decode_vli(encoded) + end + + test "always uses the smallest encoding" do + assert byte_size(Frame.encode_vli(0)) == 1 + assert byte_size(Frame.encode_vli(63)) == 1 + assert byte_size(Frame.encode_vli(64)) == 2 + assert byte_size(Frame.encode_vli(16_383)) == 2 + assert byte_size(Frame.encode_vli(16_384)) == 4 + assert byte_size(Frame.encode_vli(1_073_741_823)) == 4 + assert byte_size(Frame.encode_vli(1_073_741_824)) == 8 + end + end + + describe "decode_vli/1" do + test "decodes a 1-byte integer" do + assert {:ok, 7, <<>>} = Frame.decode_vli(<<7>>) + end + + test "decodes zero" do + assert {:ok, 0, <<>>} = Frame.decode_vli(<<0>>) + end + + test "decodes 63 (max 1-byte)" do + assert {:ok, 63, <<>>} = Frame.decode_vli(<<63>>) + end + + test "decodes a 2-byte integer and returns remainder" do + assert {:ok, 494, "rest"} = Frame.decode_vli(<<0x41, 0xEE, "rest">>) + end + + test "decodes a 4-byte integer" do + encoded = Frame.encode_vli(100_000) + assert {:ok, 100_000, <<>>} = Frame.decode_vli(encoded) + end + + test "decodes an 8-byte integer" do + encoded = Frame.encode_vli(2_000_000_000_000) + assert {:ok, 2_000_000_000_000, <<>>} = Frame.decode_vli(encoded) + end + + test "returns :incomplete for empty binary" do + assert {:error, :incomplete} = Frame.decode_vli(<<>>) + end + + test "returns :incomplete when 2-byte integer is truncated" do + assert {:error, :incomplete} = Frame.decode_vli(<<0x40>>) + end + + test "returns :incomplete when 4-byte integer is truncated" do + assert {:error, :incomplete} = Frame.decode_vli(<<0x80, 0x00, 0x00>>) + end + + test "returns :incomplete when 8-byte integer is truncated" do + assert {:error, :incomplete} = Frame.decode_vli(<<0xC0, 0x00, 0x00>>) + end + + test "round-trips all boundary values" do + for n <- [0, 63, 64, 16_383, 16_384, 1_073_741_823, 1_073_741_824] do + encoded = Frame.encode_vli(n) + + assert {:ok, ^n, <<>>} = Frame.decode_vli(encoded), + "round-trip failed for #{n}" + end + end + end + + describe "encode/1" do + test "encodes a DATA frame" do + frame = %Data{payload: %Data.Payload{data: "hello"}} + assert {:ok, updated, iodata} = Frame.encode(frame) + assert IO.iodata_to_binary(iodata) == <<0x00, 0x05, "hello">> + assert updated.length == 5 + assert updated.type == 0x0 + end + + test "encodes a HEADERS frame" do + frame = %Headers{payload: %Headers.Payload{hbf: "hdr"}} + assert {:ok, updated, iodata} = Frame.encode(frame) + assert IO.iodata_to_binary(iodata) == <<0x01, 0x03, "hdr">> + assert updated.length == 3 + assert updated.type == 0x1 + end + + test "encodes a SETTINGS frame with [{1, 0}, {7, 0}]" do + frame = %Settings{payload: %Settings.Payload{settings: [{1, 0}, {7, 0}]}} + assert {:ok, updated, iodata} = Frame.encode(frame) + assert IO.iodata_to_binary(iodata) == <<0x04, 0x04, 0x01, 0x00, 0x07, 0x00>> + assert updated.length == 4 + assert updated.type == 0x4 + end + + test "encodes a SETTINGS frame with empty settings" do + frame = %Settings{payload: %Settings.Payload{settings: []}} + assert {:ok, updated, iodata} = Frame.encode(frame) + assert IO.iodata_to_binary(iodata) == <<0x04, 0x00>> + assert updated.length == 0 + end + + test "encodes a CANCEL_PUSH frame with push_id: 42" do + frame = %CancelPush{payload: %CancelPush.Payload{push_id: 42}} + assert {:ok, updated, iodata} = Frame.encode(frame) + bin = IO.iodata_to_binary(iodata) + # type 0x03, VLI length, VLI push_id + assert <<0x03, _len, _push_id_vli::binary>> = bin + # push_id 42 < 64 so encodes as 1 byte + assert updated.length == byte_size(Frame.encode_vli(42)) + assert updated.type == 0x3 + end + + test "encodes a GOAWAY frame with stream_id: 42" do + frame = %GoAway{payload: %GoAway.Payload{stream_id: 42}} + assert {:ok, updated, iodata} = Frame.encode(frame) + bin = IO.iodata_to_binary(iodata) + assert <<0x07, _len, _stream_id_vli::binary>> = bin + assert updated.length == byte_size(Frame.encode_vli(42)) + assert updated.type == 0x7 + end + + test "encodes a MAX_PUSH_ID frame with push_id: 100" do + frame = %MaxPushId{payload: %MaxPushId.Payload{push_id: 100}} + assert {:ok, updated, iodata} = Frame.encode(frame) + bin = IO.iodata_to_binary(iodata) + assert <<0x0D, _len, _push_id_vli::binary>> = bin + # push_id 100 >= 64 so encodes as 2 bytes + assert updated.length == byte_size(Frame.encode_vli(100)) + assert updated.type == 0xD + end + + test "encodes a PUSH_PROMISE frame with push_id: 1, hbf: \"hdr\"" do + frame = %PushPromise{payload: %PushPromise.Payload{push_id: 1, hbf: "hdr"}} + assert {:ok, updated, iodata} = Frame.encode(frame) + bin = IO.iodata_to_binary(iodata) + # type 0x05, length VLI, then push_id=1 as 1-byte VLI (0x01), then hbf bytes + assert <<0x05, _len, 0x01, "hdr">> = bin + # payload = 1-byte push_id VLI + 3-byte hbf + assert updated.length == 4 + assert updated.type == 0x5 + end + + test "encodes an empty DATA frame" do + frame = %Data{payload: %Data.Payload{data: <<>>}} + assert {:ok, updated, iodata} = Frame.encode(frame) + assert IO.iodata_to_binary(iodata) == <<0x00, 0x00>> + assert updated.length == 0 + end + + test "sets updated_frame.length to the payload byte count" do + frame = %Data{payload: %Data.Payload{data: "hello world"}} + assert {:ok, updated, _iodata} = Frame.encode(frame) + assert updated.length == 11 + end + + test "preserves the integer type code in updated_frame" do + frame = %Headers{payload: %Headers.Payload{hbf: "x"}} + assert {:ok, updated, _iodata} = Frame.encode(frame) + assert updated.type == 0x1 + end + end + + describe "stream/1" do + test "yields a complete DATA frame token" do + {:ok, _, iodata} = Frame.encode(%Data{payload: %Data.Payload{data: "hi"}}) + bin = IO.iodata_to_binary(iodata) + assert Frame.stream(bin) |> Enum.to_list() == [{"", {0, "hi"}}] + end + + test "yields a complete HEADERS frame token" do + {:ok, _, iodata} = Frame.encode(%Headers{payload: %Headers.Payload{hbf: "hdr"}}) + bin = IO.iodata_to_binary(iodata) + assert Frame.stream(bin) |> Enum.to_list() == [{"", {1, "hdr"}}] + end + + test "yields two consecutive frame tokens in order" do + {:ok, _, d_io} = Frame.encode(%Data{payload: %Data.Payload{data: "hi"}}) + {:ok, _, h_io} = Frame.encode(%Headers{payload: %Headers.Payload{hbf: "hdr"}}) + bin = IO.iodata_to_binary([d_io, h_io]) + tokens = Frame.stream(bin) |> Enum.map(fn {_rest, t} -> t end) + assert tokens == [{0, "hi"}, {1, "hdr"}] + end + + test "yields nothing for an empty binary" do + assert Frame.stream(<<>>) |> Enum.to_list() == [] + end + + test "yields {partial_bin, nil} for a partial frame" do + # A DATA frame that claims 5 bytes of payload but only has 2. + partial = <<0x00, 0x05, "ab">> + # Use Enum.take/2 — Enum.to_list/1 would loop because the unfold keeps + # returning the same partial buffer as the next accumulator state. + result = Frame.stream(partial) |> Enum.take(1) + assert [{^partial, nil}] = result + end + + test "the rest field of a complete-frame token equals the unconsumed bytes" do + {:ok, _, iodata} = Frame.encode(%Data{payload: %Data.Payload{data: "hi"}}) + leftover = "leftover" + bin = IO.iodata_to_binary(iodata) <> leftover + # Use Enum.take/2 because the leftover cannot be parsed as a valid frame + # and would cause the unfold to loop. + [{rest, {0, "hi"}}] = Frame.stream(bin) |> Enum.take(1) + assert rest == leftover + end + + test "carries the full payload for large frames (300 bytes)" do + payload = :binary.copy(<<0xAB>>, 300) + {:ok, _, iodata} = Frame.encode(%Data{payload: %Data.Payload{data: payload}}) + bin = IO.iodata_to_binary(iodata) + # No leftover — Enum.to_list/1 is safe. + [{_, {0, decoded_payload}}] = Frame.stream(bin) |> Enum.to_list() + assert decoded_payload == payload + end + + test "complete frame followed by partial yields two elements" do + {:ok, _, iodata} = Frame.encode(%Data{payload: %Data.Payload{data: "hi"}}) + # A HEADERS frame that claims 5 bytes but only has 2. + partial = <<0x01, 0x05, "ab">> + bin = IO.iodata_to_binary(iodata) <> partial + # Stop at 2 to avoid the infinite-loop behaviour on the partial tail. + result = Frame.stream(bin) |> Enum.take(2) + assert [{_, {0, "hi"}}, {^partial, nil}] = result + end + end + + describe "decode/2" do + test "decodes a DATA payload" do + {:ok, decoded} = Frame.decode(%Data{}, "hello") + assert decoded.payload.data == "hello" + assert decoded.length == 5 + end + + test "decodes a HEADERS payload" do + {:ok, decoded} = Frame.decode(%Headers{}, "hbf") + assert decoded.payload.hbf == "hbf" + assert decoded.length == 3 + end + + test "decodes a SETTINGS payload with [{1, 0}, {7, 0}]" do + {:ok, decoded} = Frame.decode(%Settings{}, <<0x01, 0x00, 0x07, 0x00>>) + assert decoded.payload.settings == [{1, 0}, {7, 0}] + assert decoded.length == 4 + end + + test "decodes an empty SETTINGS payload" do + {:ok, decoded} = Frame.decode(%Settings{}, <<>>) + assert decoded.payload.settings == [] + assert decoded.length == 0 + end + + test "decodes a CANCEL_PUSH payload" do + payload_bin = Frame.encode_vli(42) + {:ok, decoded} = Frame.decode(%CancelPush{}, payload_bin) + assert decoded.payload.push_id == 42 + assert decoded.length == byte_size(payload_bin) + end + + test "decodes a GOAWAY payload" do + payload_bin = Frame.encode_vli(99) + {:ok, decoded} = Frame.decode(%GoAway{}, payload_bin) + assert decoded.payload.stream_id == 99 + assert decoded.length == byte_size(payload_bin) + end + + test "decodes a MAX_PUSH_ID payload" do + payload_bin = Frame.encode_vli(7) + {:ok, decoded} = Frame.decode(%MaxPushId{}, payload_bin) + assert decoded.payload.push_id == 7 + assert decoded.length == byte_size(payload_bin) + end + + test "decodes a PUSH_PROMISE payload" do + # push_id=1 encoded as a 1-byte VLI (0x01), followed by the HBF bytes. + payload = <<0x01, "hdr">> + {:ok, decoded} = Frame.decode(%PushPromise{}, payload) + assert decoded.payload.push_id == 1 + assert decoded.payload.hbf == "hdr" + assert decoded.length == byte_size(payload) + end + + test "sets decoded.length to byte_size(payload_binary) in every case" do + payload = "some arbitrary data" + {:ok, decoded} = Frame.decode(%Data{}, payload) + assert decoded.length == byte_size(payload) + end + end + + describe "control_stream_preface/0" do + test "is exactly 7 bytes long" do + assert byte_size(Frame.control_stream_preface()) == 7 + end + + test "starts with the control stream type byte 0x00" do + <> = Frame.control_stream_preface() + assert stream_type == 0x00 + end + + test "second byte is the SETTINGS frame type 0x04" do + <<_, frame_type, _::binary>> = Frame.control_stream_preface() + assert frame_type == 0x04 + end + + test "third byte is the payload length 4" do + <<_, _, length, _::binary>> = Frame.control_stream_preface() + assert length == 0x04 + end + + test "the embedded SETTINGS frame decodes with stream/1 and decode/2" do + <<_stream_type, rest::binary>> = Frame.control_stream_preface() + # rest is exactly one complete SETTINGS frame — Enum.to_list/1 is safe. + [{_, {type_int, payload_bin}}] = Frame.stream(rest) |> Enum.to_list() + assert type_int == 0x4 + {:ok, frame} = Frame.decode(%Settings{}, payload_bin) + assert frame.payload.settings == [{1, 0}, {7, 0}] + end + + test "payload decodes to QPACK_MAX_TABLE_CAPACITY=0 and QPACK_BLOCKED_STREAMS=0" do + <<_, _, _, payload::binary>> = Frame.control_stream_preface() + {:ok, frame} = Frame.decode(%Settings{}, payload) + assert frame.payload.settings == [{1, 0}, {7, 0}] + end + end + + describe "encode/decode round-trips" do + test "round-trips a DATA frame" do + struct = %Data{payload: %Data.Payload{data: "hello"}} + {:ok, _, iodata} = Frame.encode(struct) + + [{_, {type_int, payload_bin}}] = + Frame.stream(IO.iodata_to_binary(iodata)) |> Enum.to_list() + + assert type_int == struct.type + {:ok, decoded} = Frame.decode(%Data{}, payload_bin) + assert decoded.payload.data == "hello" + end + + test "round-trips a HEADERS frame" do + struct = %Headers{payload: %Headers.Payload{hbf: "header bytes"}} + {:ok, _, iodata} = Frame.encode(struct) + + [{_, {type_int, payload_bin}}] = + Frame.stream(IO.iodata_to_binary(iodata)) |> Enum.to_list() + + assert type_int == struct.type + {:ok, decoded} = Frame.decode(%Headers{}, payload_bin) + assert decoded.payload.hbf == "header bytes" + end + + test "round-trips a SETTINGS frame" do + settings = [{1, 0}, {6, 4096}, {7, 0}] + struct = %Settings{payload: %Settings.Payload{settings: settings}} + {:ok, _, iodata} = Frame.encode(struct) + + [{_, {type_int, payload_bin}}] = + Frame.stream(IO.iodata_to_binary(iodata)) |> Enum.to_list() + + assert type_int == struct.type + {:ok, decoded} = Frame.decode(%Settings{}, payload_bin) + assert decoded.payload.settings == settings + end + + test "round-trips a CANCEL_PUSH frame" do + struct = %CancelPush{payload: %CancelPush.Payload{push_id: 99}} + {:ok, _, iodata} = Frame.encode(struct) + + [{_, {type_int, payload_bin}}] = + Frame.stream(IO.iodata_to_binary(iodata)) |> Enum.to_list() + + assert type_int == struct.type + {:ok, decoded} = Frame.decode(%CancelPush{}, payload_bin) + assert decoded.payload.push_id == 99 + end + + test "round-trips a GOAWAY frame" do + struct = %GoAway{payload: %GoAway.Payload{stream_id: 7}} + {:ok, _, iodata} = Frame.encode(struct) + + [{_, {type_int, payload_bin}}] = + Frame.stream(IO.iodata_to_binary(iodata)) |> Enum.to_list() + + assert type_int == struct.type + {:ok, decoded} = Frame.decode(%GoAway{}, payload_bin) + assert decoded.payload.stream_id == 7 + end + + test "round-trips a MAX_PUSH_ID frame" do + struct = %MaxPushId{payload: %MaxPushId.Payload{push_id: 255}} + {:ok, _, iodata} = Frame.encode(struct) + + [{_, {type_int, payload_bin}}] = + Frame.stream(IO.iodata_to_binary(iodata)) |> Enum.to_list() + + assert type_int == struct.type + {:ok, decoded} = Frame.decode(%MaxPushId{}, payload_bin) + assert decoded.payload.push_id == 255 + end + + test "round-trips a PUSH_PROMISE frame" do + struct = %PushPromise{payload: %PushPromise.Payload{push_id: 3, hbf: "hbf bytes"}} + {:ok, _, iodata} = Frame.encode(struct) + + [{_, {type_int, payload_bin}}] = + Frame.stream(IO.iodata_to_binary(iodata)) |> Enum.to_list() + + assert type_int == struct.type + {:ok, decoded} = Frame.decode(%PushPromise{}, payload_bin) + assert decoded.payload.push_id == 3 + assert decoded.payload.hbf == "hbf bytes" + end + + test "round-trips a large DATA payload (70_000 bytes) — exercises multi-byte VLI length" do + payload = :binary.copy("x", 70_000) + struct = %Data{payload: %Data.Payload{data: payload}} + {:ok, _, iodata} = Frame.encode(struct) + + [{_, {type_int, payload_bin}}] = + Frame.stream(IO.iodata_to_binary(iodata)) |> Enum.to_list() + + assert type_int == struct.type + {:ok, decoded} = Frame.decode(%Data{}, payload_bin) + assert decoded.payload.data == payload + end + + test "round-trips an empty DATA payload" do + struct = %Data{payload: %Data.Payload{data: <<>>}} + {:ok, _, iodata} = Frame.encode(struct) + + [{_, {type_int, payload_bin}}] = + Frame.stream(IO.iodata_to_binary(iodata)) |> Enum.to_list() + + assert type_int == struct.type + {:ok, decoded} = Frame.decode(%Data{}, payload_bin) + assert decoded.payload.data == <<>> + end + + test "round-trips SETTINGS with multiple pairs [{1, 0}, {6, 4096}, {7, 0}]" do + settings = [{1, 0}, {6, 4096}, {7, 0}] + struct = %Settings{payload: %Settings.Payload{settings: settings}} + {:ok, _, iodata} = Frame.encode(struct) + + [{_, {type_int, payload_bin}}] = + Frame.stream(IO.iodata_to_binary(iodata)) |> Enum.to_list() + + assert type_int == struct.type + {:ok, decoded} = Frame.decode(%Settings{}, payload_bin) + assert decoded.payload.settings == settings + end + end +end diff --git a/test/ankh/protocol/http3/request_test.exs b/test/ankh/protocol/http3/request_test.exs new file mode 100644 index 0000000..878a982 --- /dev/null +++ b/test/ankh/protocol/http3/request_test.exs @@ -0,0 +1,374 @@ +defmodule Ankh.Protocol.HTTP3.RequestTest do + use ExUnit.Case, async: false + + @moduletag :integration + @moduletag timeout: 30_000 + + alias Ankh.HTTP + alias Ankh.HTTP.Request + + # Drives the receive loop until end_stream = true is signalled for `ref`, + # returning {:ok, status, headers, body_binary} or {:error, reason}. + defp collect_response(protocol, ref, timeout \\ 10_000) do + do_collect(protocol, ref, timeout, nil, [], []) + end + + defp do_collect(protocol, ref, timeout, status, headers, body) do + receive do + msg -> + case HTTP.stream(protocol, msg) do + {:ok, new_proto, events} -> + case process_events(events, ref, status, headers, body) do + {:done, st, hd, bd} -> + {:ok, st, hd, IO.iodata_to_binary(Enum.reverse(bd))} + + {:cont, st, hd, bd} -> + do_collect(new_proto, ref, timeout, st, hd, bd) + end + + {:error, reason} -> + {:error, reason} + + _other -> + do_collect(protocol, ref, timeout, status, headers, body) + end + after + timeout -> {:error, :timeout} + end + end + + defp process_events([], _ref, status, headers, body), + do: {:cont, status, headers, body} + + defp process_events([event | rest], ref, status, headers, body) do + case event do + {:headers, ^ref, decoded_headers, end_stream} -> + new_status = + case List.keyfind(decoded_headers, ":status", 0) do + {":status", v} -> String.to_integer(v) + nil -> status + end + + regular = + Enum.reject(decoded_headers, fn {n, _} -> String.starts_with?(n, ":") end) + + new_headers = headers ++ regular + + if end_stream do + {:done, new_status, new_headers, body} + else + process_events(rest, ref, new_status, new_headers, body) + end + + {:data, ^ref, chunk, end_stream} -> + new_body = [chunk | body] + + if end_stream do + {:done, status, headers, new_body} + else + process_events(rest, ref, status, headers, new_body) + end + + {:error, ^ref, _reason, _complete} -> + {:done, status, headers, body} + + _other -> + process_events(rest, ref, status, headers, body) + end + end + + defp connect_h3!(uri_string, extra_opts \\ []) do + uri = URI.parse(uri_string) + opts = Keyword.merge([timeout: 10_000, protocols: [:h3]], extra_opts) + {:ok, protocol} = HTTP.connect(uri, opts) + protocol + end + + defp get!(protocol, path, extra_headers \\ []) do + request = + Request.new( + method: :GET, + path: path, + headers: [{"user-agent", "ankh-h3-test/1"}, {"accept", "*/*"}] ++ extra_headers + ) + + {:ok, protocol, ref} = HTTP.request(protocol, request) + {protocol, ref} + end + + describe "basic GET requests over HTTP/3" do + test "GET / from cloudflare-quic.com returns HTTP 200" do + protocol = connect_h3!("https://cloudflare-quic.com") + {protocol, ref} = get!(protocol, "/") + + assert {:ok, status, headers, body} = collect_response(protocol, ref) + assert status == 200 + assert byte_size(body) > 0 + assert Enum.any?(headers, fn {k, _} -> k == "content-type" end) + end + + test "GET / response contains expected Cloudflare headers" do + protocol = connect_h3!("https://cloudflare-quic.com") + {protocol, ref} = get!(protocol, "/") + + assert {:ok, 200, headers, _body} = collect_response(protocol, ref) + + header_names = Enum.map(headers, &elem(&1, 0)) + assert "server" in header_names + assert "date" in header_names + assert "content-type" in header_names + end + + test "GET / response server header identifies cloudflare" do + protocol = connect_h3!("https://cloudflare-quic.com") + {protocol, ref} = get!(protocol, "/") + + assert {:ok, 200, headers, _body} = collect_response(protocol, ref) + + {_, server} = List.keyfind(headers, "server", 0) + assert server =~ "cloudflare" + end + + test "GET / advertises HTTP/3 support in Alt-Svc header" do + protocol = connect_h3!("https://cloudflare-quic.com") + {protocol, ref} = get!(protocol, "/") + + assert {:ok, 200, headers, _body} = collect_response(protocol, ref) + + case List.keyfind(headers, "alt-svc", 0) do + {_, alt_svc} -> assert alt_svc =~ "h3" + nil -> flunk("expected Alt-Svc header advertising h3") + end + end + + test "GET / to cloudflare.com returns a redirect (3xx)" do + protocol = connect_h3!("https://cloudflare.com") + {protocol, ref} = get!(protocol, "/") + + assert {:ok, status, _headers, _body} = collect_response(protocol, ref) + assert status in 200..399 + end + + test "GET / receives full body from cloudflare-quic.com" do + protocol = connect_h3!("https://cloudflare-quic.com") + {protocol, ref} = get!(protocol, "/") + + assert {:ok, 200, headers, body} = collect_response(protocol, ref) + + # Verify content-length matches received body when present + case List.keyfind(headers, "content-length", 0) do + {_, cl_str} -> + content_length = String.to_integer(cl_str) + assert byte_size(body) == content_length + + nil -> + assert byte_size(body) > 0 + end + end + + test "GET / response body is valid HTML" do + protocol = connect_h3!("https://cloudflare-quic.com") + {protocol, ref} = get!(protocol, "/") + + assert {:ok, 200, _headers, body} = collect_response(protocol, ref) + assert body =~ "" or body =~ " k == "cf-ray" end) + end + + test "Accept header is sent" do + protocol = connect_h3!("https://cloudflare-quic.com") + {protocol, ref} = get!(protocol, "/", [{"accept", "text/html"}]) + + assert {:ok, 200, headers, _body} = collect_response(protocol, ref) + {_, content_type} = List.keyfind(headers, "content-type", 0) + assert content_type =~ "text/html" + end + + test "multiple custom headers are sent successfully" do + protocol = connect_h3!("https://cloudflare-quic.com") + + extra = [ + {"accept-language", "en-US,en;q=0.9"}, + {"cache-control", "no-cache"} + ] + + {protocol, ref} = get!(protocol, "/", extra) + + # Request should succeed — the key assertion is no error + assert {:ok, status, _headers, _body} = collect_response(protocol, ref) + assert status in 200..399 + end + end + + describe "protocol negotiation" do + test "protocols: [:h3] connects with HTTP/3" do + # The connection itself succeeding over QUIC is the assertion. + # We also verify the server sends its three control/QPACK streams + # (visible via debug logs) but just assert the response works. + protocol = connect_h3!("https://cloudflare-quic.com", protocols: [:h3]) + {protocol, ref} = get!(protocol, "/") + + assert {:ok, 200, _headers, body} = collect_response(protocol, ref) + assert byte_size(body) > 0 + end + + test "protocols: [:h3, :h2, :\"http/1.1\"] prefers HTTP/3 when available" do + uri = URI.parse("https://cloudflare-quic.com") + opts = [timeout: 10_000, protocols: [:h3, :h2, :"http/1.1"]] + assert {:ok, protocol} = HTTP.connect(uri, opts) + + {protocol, ref} = get!(protocol, "/") + assert {:ok, 200, _headers, body} = collect_response(protocol, ref) + assert byte_size(body) > 0 + end + + test "default https:// connection uses TLS (no :protocols option)" do + uri = URI.parse("https://cloudflare.com") + + # Default: TLS, ALPN h2 / http1.1 — no protocols opt means TLS path + assert {:ok, protocol} = HTTP.connect(uri, timeout: 10_000) + + {protocol, ref} = get!(protocol, "/") + assert {:ok, status, _headers, _body} = collect_response(protocol, ref) + assert status in 200..399 + end + + test "protocols: [:h3] against one.one.one.one succeeds" do + uri = URI.parse("https://1.1.1.1") + opts = [timeout: 10_000, protocols: [:h3], verify: :none] + assert {:ok, protocol} = HTTP.connect(uri, opts) + + {protocol, ref} = get!(protocol, "/") + assert {:ok, status, _headers, _body} = collect_response(protocol, ref) + assert status in 200..399 + end + end + + describe "multiple sequential requests" do + test "two sequential GET requests on the same H3 connection both succeed" do + protocol = connect_h3!("https://cloudflare-quic.com") + + # First request + {protocol, ref1} = get!(protocol, "/") + assert {:ok, 200, _headers1, body1} = collect_response(protocol, ref1) + assert byte_size(body1) > 0 + + # Second request on the same connection + {protocol, ref2} = get!(protocol, "/") + assert {:ok, 200, _headers2, body2} = collect_response(protocol, ref2) + assert byte_size(body2) > 0 + end + + test "three sequential requests all return 200" do + protocol = connect_h3!("https://cloudflare-quic.com") + + for _ <- 1..3 do + {protocol, ref} = get!(protocol, "/") + assert {:ok, 200, _headers, body} = collect_response(protocol, ref) + assert byte_size(body) > 0 + {protocol, nil} + end + end + + test "response bodies are consistent across sequential requests" do + protocol = connect_h3!("https://cloudflare-quic.com") + + {protocol, ref1} = get!(protocol, "/") + {:ok, 200, _h1, body1} = collect_response(protocol, ref1) + + {protocol2, ref2} = get!(protocol, "/") + {:ok, 200, _h2, body2} = collect_response(protocol2, ref2) + + # Both fetches of the same static page should return the same byte count + assert byte_size(body1) == byte_size(body2) + end + end + + describe "QPACK header decoding" do + test "all response header names are lowercase strings" do + protocol = connect_h3!("https://cloudflare-quic.com") + {protocol, ref} = get!(protocol, "/") + + assert {:ok, 200, headers, _body} = collect_response(protocol, ref) + + for {name, _value} <- headers do + assert name == String.downcase(name), + "header name #{inspect(name)} is not lowercase" + end + end + + test "response headers do not contain HTTP/2-style pseudo-headers" do + # Pseudo-headers (:status, :path, etc.) must be stripped by the protocol + # layer before returning responses to the caller. + protocol = connect_h3!("https://cloudflare-quic.com") + {protocol, ref} = get!(protocol, "/") + + assert {:ok, _status, headers, _body} = collect_response(protocol, ref) + + pseudo_headers = + Enum.filter(headers, fn {name, _} -> String.starts_with?(name, ":") end) + + assert pseudo_headers == [], + "expected no pseudo-headers in decoded response, got: #{inspect(pseudo_headers)}" + end + + test "date header is a valid HTTP date string" do + protocol = connect_h3!("https://cloudflare-quic.com") + {protocol, ref} = get!(protocol, "/") + + assert {:ok, 200, headers, _body} = collect_response(protocol, ref) + + assert {_, date_value} = List.keyfind(headers, "date", 0) + # HTTP date contains a day name and a year + assert date_value =~ ~r/(Mon|Tue|Wed|Thu|Fri|Sat|Sun)/ + assert date_value =~ ~r/20\d\d/ + end + + test "content-length header value parses as a non-negative integer" do + protocol = connect_h3!("https://cloudflare-quic.com") + {protocol, ref} = get!(protocol, "/") + + assert {:ok, 200, headers, _body} = collect_response(protocol, ref) + + case List.keyfind(headers, "content-length", 0) do + {_, value} -> + {length, ""} = Integer.parse(value) + assert length >= 0 + + nil -> + :ok + end + end + end + + describe "connection error handling" do + test "connecting to an unreachable host returns an error" do + uri = URI.parse("https://this.host.does.not.exist.invalid") + result = HTTP.connect(uri, timeout: 3_000, protocols: [:h3]) + assert {:error, _reason} = result + end + + test "protocols: [:h3] to a TLS-only host fails gracefully (no fallback)" do + # A host that doesn't speak QUIC on 443 should return an error when only + # :h3 is requested (no TLS fallback). + uri = URI.parse("https://example.com") + result = HTTP.connect(uri, timeout: 5_000, protocols: [:h3]) + + # We accept either :error (QUIC refused/timeout) or a successful H3 + # connection (example.com may or may not support H3). + # The key assertion: no crash, just a result tuple. + assert match?({:ok, _}, result) or match?({:error, _}, result) + end + end +end