%%% @doc Client-side CONNECT-IP session (RFC 9484). %%% %%% One `gen_statem' per tunnel, transport-generic: the same module %%% drives both HTTP/3 (`quic_h3', QUIC DATAGRAM frames) and HTTP/2 %%% (`erlang_h2', RFC 9297 DATAGRAM-type capsules on the stream body) %%% by dispatching on a `transport :: h3 | h2' field — following the %%% architecture already used by `masque_tcp_client_session'. %%% %%% Owner-side events (the process that called `masque:connect/3'): %%% %%% -module(masque_ip_client_session). -behaviour(gen_statem). -export([start_link/3, start/3, stop/1, info/1]). -export([send_ip_packet/2, recv/2, set_mode/2]). -export([request_addresses/2, assign_addresses/2, advertise_routes/2, ip_info/1]). -export([send_capsule/3]). -export([init/1, callback_mode/0, terminate/3, code_change/4]). -export([connecting/3, open/3, closing/3]). -include("masque.hrl"). -include("masque_ip.hrl"). -dialyzer({nowarn_function, [do_connect/2, verify_h3_peer_settings/1, verify_h2_peer_settings/1, request_headers/1, build_authority/2, is_ipv6_literal/1]}). -record(data, { owner :: pid(), owner_ref :: reference(), proxy_host :: binary(), proxy_port :: inet:port_number(), template :: masque_uri_template:template(), target :: masque_uri_ip:ip_target(), ipproto :: masque_uri_ip:ip_ipproto(), transport :: h3 | h2, mtu :: 1280..65535, conn :: pid() | undefined, stream_id :: non_neg_integer() | undefined, handshake_from :: gen_statem:from() | undefined, timeout_ref :: reference() | undefined, mode :: message | queue, rx_buf :: queue:queue(binary()), rx_waiters :: queue:queue({gen_statem:from(), reference()}), cap_buf = <<>> :: binary(), max_cap :: pos_integer(), %% Nonzero IDs the client has sent in ADDRESS_REQUEST and not yet %% answered locally (used by assign_addresses/2 validation). peer_pending = #{} :: #{pos_integer() => true}, %% Highest nonzero Request ID allocated for our own outbound %% ADDRESS_REQUEST (monotonic). next_req_id = 1 :: pos_integer(), %% Most recent server-advertised state. assigned = [] :: [masque_ip_capsule:address_entry()], routes = [] :: [masque_ip_capsule:route_entry()], %% Extra request headers prepended to the CONNECT-IP request. extra_headers = [] :: [{binary(), binary()}], %% When set, the conn is owned by a `masque_upstream_owner'; %% teardown releases the stream back to the pool instead of %% closing the conn. pool_owner :: pid() | undefined }). %%==================================================================== %% API %%==================================================================== start_link(Target, Opts, Owner) -> gen_statem:start_link(?MODULE, {Target, Opts, Owner}, []). start(Target, Opts, Owner) -> gen_statem:start(?MODULE, {Target, Opts, Owner}, []). stop(Pid) -> gen_statem:call(Pid, stop, 5000). info(Pid) -> gen_statem:call(Pid, info, 1000). %% @doc Send a full IP packet through the tunnel. -spec send_ip_packet(pid(), binary()) -> ok | {error, term()}. send_ip_packet(Pid, Packet) when is_binary(Packet) -> gen_statem:call(Pid, {send_ip_packet, Packet}). %% @doc Block for the next inbound IP packet when in `queue' mode. recv(Pid, Timeout) -> gen_statem:call(Pid, {recv, Timeout}, Timeout + 500). set_mode(Pid, Mode) when Mode =:= message; Mode =:= queue -> gen_statem:call(Pid, {set_mode, Mode}). %% @doc Send an ADDRESS_REQUEST capsule asking the server to assign %% addresses matching the given prefixes. Returns the list of %% allocated Request IDs (strictly nonzero). -spec request_addresses(pid(), [{4 | 6, inet:ip_address(), non_neg_integer()}]) -> {ok, [pos_integer()]} | {error, term()}. request_addresses(Pid, Prefixes) -> gen_statem:call(Pid, {request_addresses, Prefixes}). %% @doc Send an ADDRESS_ASSIGN capsule. Nonzero Request IDs must %% match an outstanding peer ADDRESS_REQUEST (tracked in the %% session's pending set); ID 0 is unprompted and always accepted. -spec assign_addresses(pid(), [masque_ip_capsule:address_entry()]) -> ok | {error, term()}. assign_addresses(Pid, Assignments) -> gen_statem:call(Pid, {assign_addresses, Assignments}). %% @doc Send a ROUTE_ADVERTISEMENT capsule. -spec advertise_routes(pid(), [masque_ip_capsule:route_entry()]) -> ok | {error, term()}. advertise_routes(Pid, Routes) -> gen_statem:call(Pid, {advertise_routes, Routes}). -spec ip_info(pid()) -> map(). ip_info(Pid) -> gen_statem:call(Pid, ip_info, 1000). send_capsule(Pid, Type, Value) -> gen_statem:call(Pid, {send_capsule, Type, Value}). %%==================================================================== %% gen_statem %%==================================================================== callback_mode() -> state_functions. init({{Target, IPProto}, Opts, Owner}) -> process_flag(trap_exit, true), {ProxyHost, ProxyPort} = maps:get(proxy, Opts), MRef = erlang:monitor(process, Owner), Mode = maps:get(mode, Opts, message), Transport = maps:get(transport, Opts, h3), MaxCap = maps:get(max_capsule_size, Opts, ?MASQUE_DEFAULT_MAX_CAPSULE_SIZE), Mtu = maps:get(mtu, Opts, 1500), Template = build_template(Opts, ProxyHost, ProxyPort), Data = #data{ owner = Owner, owner_ref = MRef, proxy_host = to_bin(ProxyHost), proxy_port = ProxyPort, template = Template, target = Target, ipproto = IPProto, transport = Transport, mtu = Mtu, mode = Mode, rx_buf = queue:new(), rx_waiters = queue:new(), max_cap = MaxCap, extra_headers = sanitise_extra_headers( maps:get(request_headers, Opts, [])), pool_owner = maps:get(pool_owner, Opts, undefined) }, {ok, connecting, Data, [{next_event, internal, {do_handshake, Opts}}]}. build_template(Opts, ProxyHost, ProxyPort) -> case maps:find(uri_template, Opts) of {ok, Raw} -> case masque_uri_ip:parse_client_template(Raw) of {ok, T} -> T; {error, Err} -> erlang:error({bad_template, Err}) end; error -> Authority = build_authority(to_bin(ProxyHost), ProxyPort), Raw = <<"https://", Authority/binary, ?MASQUE_DEFAULT_IP_URI_PATH_PATTERN/binary>>, {ok, T} = masque_uri_ip:parse_client_template(Raw), T end. %%==================================================================== %% States %%==================================================================== connecting(internal, {do_handshake, Opts}, Data) -> case do_connect(Data, Opts) of {ok, Conn, StreamId} -> Timeout = maps:get(timeout, Opts, 5000), TRef = erlang:start_timer(Timeout, self(), handshake_timeout), {keep_state, Data#data{conn = Conn, stream_id = StreamId, timeout_ref = TRef}}; {error, Reason} -> {stop, {handshake_failed, Reason}} end; connecting({call, From}, handshake_await, Data) -> {keep_state, Data#data{handshake_from = From}}; connecting({call, From}, {set_owner, NewOwner}, Data) -> {keep_state, swap_owner(NewOwner, Data), [{reply, From, ok}]}; connecting(info, {Tag, _Conn, closed}, Data) when Tag =:= h2 -> reply_handshake(Data, {error, peer_closed}), {stop, peer_closed}; connecting(info, {Tag, _Conn, {response, StreamId, Status, Headers}}, #data{stream_id = StreamId} = Data) when Tag =:= quic_h3; Tag =:= h2 -> cancel_timer(Data#data.timeout_ref), case Status of S when S >= 200, S < 300 -> case validate_response(Headers) of ok -> case check_datagram_mtu(Data) of ok -> reply_handshake(Data, ok), {next_state, open, Data#data{timeout_ref = undefined, handshake_from = undefined}}; {error, _} = MtuErr -> reply_handshake(Data, MtuErr), {stop, element(2, MtuErr)} end; {error, _} = Err -> reply_handshake(Data, Err), {stop, element(2, Err)} end; _ -> reply_handshake(Data, {error, {handshake_rejected, Status}}), {stop, {handshake_rejected, Status}} end; connecting(info, {timeout, TRef, handshake_timeout}, #data{timeout_ref = TRef} = Data) -> reply_handshake(Data, {error, handshake_timeout}), {stop, handshake_timeout}; connecting(info, {'DOWN', Ref, process, _, _}, #data{owner_ref = Ref}) -> {stop, owner_gone}; connecting(info, _Msg, Data) -> {keep_state, Data}; connecting({call, From}, info, Data) -> {keep_state, Data, [{reply, From, session_info(Data, connecting)}]}; connecting({call, From}, stop, Data) -> {stop_and_reply, normal, [{reply, From, ok}], Data}; connecting({call, From}, _Other, Data) -> {keep_state, Data, [{reply, From, {error, not_ready}}]}. open({call, From}, info, Data) -> {keep_state, Data, [{reply, From, session_info(Data, open)}]}; open({call, From}, ip_info, Data) -> {keep_state, Data, [{reply, From, #{assigned => Data#data.assigned, routes => Data#data.routes, mtu => Data#data.mtu, transport => Data#data.transport}}]}; open({call, From}, {send_ip_packet, Pkt}, Data) -> PktSz = byte_size(Pkt), case PktSz > Data#data.mtu of true -> {keep_state, Data, [{reply, From, {error, {packet_too_large, PktSz, Data#data.mtu}}}]}; false -> Reply = transport_send_datagram(Data, ?MASQUE_CONTEXT_ID_IP, Pkt), {keep_state, Data, [{reply, From, Reply}]} end; open({call, From}, {request_addresses, Prefixes}, Data) -> handle_request_addresses(From, Prefixes, Data); open({call, From}, {assign_addresses, Assignments}, Data) -> handle_assign_addresses(From, Assignments, Data); open({call, From}, {advertise_routes, Routes}, Data) -> handle_advertise_routes(From, Routes, Data); open({call, From}, {recv, Timeout}, Data) -> handle_recv_call(From, Timeout, Data); open({call, From}, {set_mode, Mode}, Data) -> {keep_state, Data#data{mode = Mode}, [{reply, From, ok}]}; open({call, From}, {set_owner, NewOwner}, Data) -> {keep_state, swap_owner(NewOwner, Data), [{reply, From, ok}]}; open({call, From}, {send_capsule, Type, Value}, Data) -> Enc = iolist_to_binary(masque_capsule:encode(Type, Value)), Reply = transport_send_data(Data, Enc, false), {keep_state, Data, [{reply, From, Reply}]}; open({call, From}, stop, Data) -> {next_state, closing, Data, [{reply, From, ok}, {next_event, internal, do_close}]}; %% H3 datagram arrival. open(info, {quic_h3, _Conn, {datagram, StreamId, Payload}}, #data{stream_id = StreamId} = Data) -> {keep_state, handle_inbound_datagram(Payload, Data)}; %% H3 or H2 stream body (carries RFC 9297 capsules — and H2 datagrams). open(info, {Tag, _Conn, {data, StreamId, Bytes, Fin}}, #data{stream_id = StreamId, cap_buf = Buf, max_cap = Max} = Data) when Tag =:= quic_h3; Tag =:= h2 -> New = <>, case byte_size(New) > Max of true -> client_stream_abort(capsule_buffer_overflow, Data); false -> drain_capsules(New, Fin, Data) end; open(info, {Tag, _Conn, {stream_reset, StreamId, _ErrorCode}}, #data{stream_id = StreamId} = Data) when Tag =:= quic_h3; Tag =:= h2 -> _ = notify_owner_closed(peer_reset, Data), {stop, peer_reset, Data}; open(info, {h2, _Conn, closed}, Data) -> _ = notify_owner_closed(peer_closed, Data), {stop, peer_closed, Data}; open(info, {timeout, TRef, {recv_timeout, From}}, Data) -> {keep_state, drop_waiter(TRef, From, Data)}; open(info, {'DOWN', Ref, process, _, _}, #data{owner_ref = Ref} = Data) -> {next_state, closing, Data, [{next_event, internal, do_close}]}; open(info, _Msg, Data) -> {keep_state, Data}. closing(internal, do_close, Data) -> _ = try transport_send_data(Data, <<>>, true) of ok -> ok; _ -> try transport_cancel(Data) catch _:_ -> ok end catch _:_ -> try transport_cancel(Data) catch _:_ -> ok end end, _ = session_teardown(Data), {stop, normal, Data}; closing(_Event, _Msg, Data) -> {keep_state, Data}. terminate(_Reason, _State, #data{conn = undefined} = D) -> cancel_all_waiters(D); terminate(_Reason, _State, Data) -> cancel_all_waiters(Data), _ = session_teardown(Data), ok. %% Close path abstraction: release the pooled stream back to the %% owner, or shut down the owned transport connection. session_teardown(#data{pool_owner = Pool, stream_id = StreamId}) when is_pid(Pool), is_integer(StreamId) -> masque_upstream_owner:release_stream(Pool, StreamId); session_teardown(#data{pool_owner = Pool}) when is_pid(Pool) -> ok; session_teardown(#data{conn = Conn} = Data) when is_pid(Conn) -> _ = (try transport_close(Data) catch _:_ -> ok end), ok; session_teardown(_) -> ok. code_change(_OldVsn, State, Data, _Extra) -> {ok, State, Data}. %%==================================================================== %% Transport dispatch %%==================================================================== do_connect(#data{pool_owner = PoolOwner} = Data, _Opts) when is_pid(PoolOwner) -> ReqHeaders = request_headers(Data), ReqOpts = case Data#data.transport of h3 -> #{end_stream => false}; h2 -> #{protocol => ?MASQUE_CONNECT_IP_PROTOCOL} end, case masque_upstream_owner:acquire_stream( PoolOwner, ReqHeaders, self(), ReqOpts) of {ok, StreamId, Conn} -> {ok, Conn, StreamId}; {error, _} = Err -> Err end; do_connect(#data{transport = h3} = Data, Opts) -> ConnOpts = maps:with([verify, cacerts], Opts), ConnOpts1 = ConnOpts#{ sync => true, settings => #{enable_connect_protocol => 1, h3_datagram => 1}, h3_datagram_enabled => true, quic_opts => #{ alpn => maps:get(alpn, Opts, [<<"h3">>]), max_datagram_frame_size => 65535 } }, case quic_h3:connect(Data#data.proxy_host, Data#data.proxy_port, ConnOpts1) of {ok, Conn} -> case verify_h3_peer_settings(Conn) of ok -> ReqHeaders = request_headers(Data), case quic_h3:request(Conn, ReqHeaders, #{end_stream => false}) of {ok, StreamId} -> {ok, Conn, StreamId}; {error, R} -> quic_h3:close(Conn), {error, {request, R}} end; {error, _} = Err -> quic_h3:close(Conn), Err end; {error, Reason} -> {error, {connect, Reason}} end; do_connect(#data{transport = h2} = Data, Opts) -> SSLOpts = [{server_name_indication, binary_to_list(Data#data.proxy_host)} | maps:get(ssl_opts, Opts, [])], ConnOpts = #{ transport => ssl, ssl_opts => SSLOpts, sync => true, verify => maps:get(verify, Opts, verify_none), timeout => maps:get(timeout, Opts, 5000), settings => #{enable_connect_protocol => 1} }, case h2:connect(Data#data.proxy_host, Data#data.proxy_port, ConnOpts) of {ok, Conn} -> case verify_h2_peer_settings(Conn) of ok -> ReqHeaders = request_headers(Data), case h2:request(Conn, ReqHeaders, #{protocol => ?MASQUE_CONNECT_IP_PROTOCOL}) of {ok, StreamId} -> {ok, Conn, StreamId}; {error, R} -> h2:close(Conn), {error, {request, R}} end; {error, _} = Err -> h2:close(Conn), Err end; {error, Reason} -> {error, {connect, Reason}} end. verify_h3_peer_settings(Conn) -> case quic_h3:get_peer_settings(Conn) of undefined -> ok; Settings when is_map(Settings) -> ECP = maps:get(enable_connect_protocol, Settings, 0), H3D = maps:get(h3_datagram, Settings, 0), if ECP =/= 1 -> {error, no_extended_connect}; H3D =/= 1 -> {error, no_h3_datagram}; true -> ok end end. verify_h2_peer_settings(Conn) -> Settings = h2:get_peer_settings(Conn), case maps:get(enable_connect_protocol, Settings, false) of true -> ok; 1 -> ok; _ -> {error, no_extended_connect} end. transport_send_datagram(#data{transport = h3, conn = C, stream_id = S}, Ctx, Payload) -> Enc = masque_datagram:encode(Ctx, Payload), quic_h3:send_datagram(C, S, Enc); transport_send_datagram(#data{transport = h2, conn = C, stream_id = S}, Ctx, Payload) -> Inner = iolist_to_binary(masque_datagram:encode(Ctx, Payload)), Cap = iolist_to_binary(h2_capsule:encode(datagram, Inner)), h2:send_data(C, S, Cap, false). transport_send_data(#data{transport = h3, conn = C, stream_id = S}, Bytes, Fin) -> quic_h3:send_data(C, S, Bytes, Fin); transport_send_data(#data{transport = h2, conn = C, stream_id = S}, Bytes, Fin) -> h2:send_data(C, S, Bytes, Fin). transport_cancel(#data{transport = h3, conn = C, stream_id = S}) -> quic_h3:cancel(C, S, ?MASQUE_H3_MESSAGE_ERROR); transport_cancel(#data{transport = h2, conn = C, stream_id = S}) -> h2:cancel(C, S, protocol_error). transport_close(#data{transport = h3, conn = C}) -> quic_h3:close(C); transport_close(#data{transport = h2, conn = C}) -> h2:close(C). request_headers(#data{template = T, target = Target, ipproto = IPProto, extra_headers = Extra}) -> Url = masque_uri_ip:expand(T, #{target => Target, ipproto => IPProto}), %% Split the synthesised URI into :authority and :path. {Scheme, Authority, Path} = split_url(Url), Base = [ {<<":method">>, <<"CONNECT">>}, {<<":protocol">>, ?MASQUE_CONNECT_IP_PROTOCOL}, {<<":scheme">>, Scheme}, {<<":authority">>, Authority}, {<<":path">>, Path}, {<<"capsule-protocol">>, <<"?1">>} ], Base ++ Extra. sanitise_extra_headers(List) when is_list(List) -> Reserved = [<<":method">>, <<":scheme">>, <<":authority">>, <<":path">>, <<":protocol">>, <<"capsule-protocol">>], [{K, V} || {K, V} <- List, is_binary(K), is_binary(V), not lists:member(K, Reserved)]. split_url(<<"https://", Rest/binary>>) -> {Auth, Path} = split_authority(Rest), {<<"https">>, Auth, Path}; split_url(<<"http://", Rest/binary>>) -> {Auth, Path} = split_authority(Rest), {<<"http">>, Auth, Path}. split_authority(Bin) -> case binary:match(Bin, <<"/">>) of {Pos, 1} -> <> = Bin, {A, P}; nomatch -> {Bin, <<"/">>} end. %%==================================================================== %% Inbound handlers %%==================================================================== handle_inbound_datagram(Payload, Data) -> case masque_datagram:decode(Payload) of {ok, {?MASQUE_CONTEXT_ID_IP, IPPkt}} -> deliver_packet(IPPkt, Data); {ok, {_OtherCtx, _}} -> %% Unknown context IDs are silently dropped per RFC 9484 §6. Data; {error, _} -> Data end. drain_capsules(Buf, Fin, Data) -> drain_capsules_1(Buf, Fin, Data). drain_capsules_1(Buf, Fin, Data) -> case decode_one_capsule(Data, Buf) of {ok, {Type, Inner}, Rest} -> case deliver_capsule(Type, Inner, Data) of {abort, Reason} -> client_stream_abort(Reason, Data); Data2 -> drain_capsules_1(Rest, Fin, Data2#data{cap_buf = <<>>}) end; {more, _} when Fin, Buf =/= <<>> -> client_stream_abort(truncated_capsule, Data); {more, _} -> {keep_state, Data#data{cap_buf = Buf}}; {error, _} -> client_stream_abort(malformed_capsule, Data) end. %% On H3 we use masque_capsule (which delegates to quic_h3_capsule); %% on H2 we use h2_capsule so the `datagram` type is resolved natively. decode_one_capsule(#data{transport = h2}, Buf) -> case h2_capsule:decode(Buf) of {ok, {datagram, Inner}, Rest} -> {ok, {datagram, Inner}, Rest}; {ok, {Type, Inner}, Rest} when is_integer(Type) -> {ok, {Type, Inner}, Rest}; Other -> Other end; decode_one_capsule(#data{transport = h3}, Buf) -> case masque_capsule:decode(Buf) of {ok, {Type, Inner, Rest}} -> {ok, {Type, Inner}, Rest}; Other -> Other end. deliver_capsule(datagram, Inner, Data) -> %% Only reached on H2. handle_inbound_datagram(Inner, Data); deliver_capsule(?MASQUE_CAPSULE_ADDRESS_ASSIGN, Inner, #data{owner = Owner} = Data) -> case masque_ip_capsule:decode_address_assign(Inner) of {ok, Entries} -> Owner ! {masque_address_assign, self(), Entries}, Data#data{assigned = Entries}; {error, _} -> {abort, malformed_capsule} end; deliver_capsule(?MASQUE_CAPSULE_ADDRESS_REQUEST, Inner, #data{owner = Owner, peer_pending = Pend} = Data) -> case masque_ip_capsule:decode_address_request(Inner) of {ok, Entries} -> Owner ! {masque_address_request, self(), Entries}, Pend1 = lists:foldl( fun(R, Acc) -> Id = element(2, R), %% #ip_prefix_request.request_id Acc#{Id => true} end, Pend, Entries), Data#data{peer_pending = Pend1}; {error, _} -> {abort, malformed_capsule} end; deliver_capsule(?MASQUE_CAPSULE_ROUTE_ADVERTISEMENT, Inner, #data{owner = Owner} = Data) -> case masque_ip_capsule:decode_route_advertisement(Inner) of {ok, Entries} -> Owner ! {masque_route_advertisement, self(), Entries}, Data#data{routes = Entries}; {error, _} -> {abort, malformed_capsule} end; deliver_capsule(Type, Inner, #data{owner = Owner} = Data) when is_integer(Type) -> Owner ! {masque_capsule, self(), Type, Inner}, Data. %%==================================================================== %% Outbound control-plane %%==================================================================== handle_request_addresses(From, Prefixes, Data) -> case build_request_entries(Prefixes, Data) of {ok, Entries, Data1} -> Body = masque_ip_capsule:encode_address_request(Entries), Cap = iolist_to_binary( masque_capsule:encode( ?MASQUE_CAPSULE_ADDRESS_REQUEST, Body)), case transport_send_data(Data1, Cap, false) of ok -> Ids = [R#ip_prefix_request.request_id || R <- Entries], {keep_state, Data1, [{reply, From, {ok, Ids}}]}; {error, _} = Err -> {keep_state, Data, [{reply, From, Err}]} end; {error, _} = Err -> {keep_state, Data, [{reply, From, Err}]} end. build_request_entries(Prefixes, Data) -> try {Rev, Next} = lists:foldl( fun({V, Addr, Pfx}, {Acc, N}) -> ok = check_prefix(V, Addr, Pfx), R = #ip_prefix_request{ request_id = N, version = V, address = Addr, prefix_len = Pfx}, {[R | Acc], N + 1} end, {[], Data#data.next_req_id}, Prefixes), {ok, lists:reverse(Rev), Data#data{next_req_id = Next}} catch throw:Err -> {error, Err}; error:_ -> {error, bad_prefix} end. handle_assign_addresses(From, Assignments, #data{peer_pending = Pend} = Data) -> case validate_assignments(Assignments, Pend) of {ok, Pend1} -> Body = masque_ip_capsule:encode_address_assign(Assignments), Cap = iolist_to_binary( masque_capsule:encode( ?MASQUE_CAPSULE_ADDRESS_ASSIGN, Body)), case transport_send_data(Data, Cap, false) of ok -> {keep_state, Data#data{peer_pending = Pend1}, [{reply, From, ok}]}; {error, _} = Err -> {keep_state, Data, [{reply, From, Err}]} end; {error, _} = Err -> {keep_state, Data, [{reply, From, Err}]} end. validate_assignments(Assignments, Pend) -> try Pend1 = lists:foldl( fun(#ip_assignment{request_id = 0, version = V, address = A, prefix_len = P}, Acc) -> ok = check_prefix(V, A, P), Acc; (#ip_assignment{request_id = Id, version = V, address = A, prefix_len = P}, Acc) -> ok = check_prefix(V, A, P), case maps:is_key(Id, Acc) of true -> maps:remove(Id, Acc); false -> throw({no_such_pending_request, Id}) end end, Pend, Assignments), {ok, Pend1} catch throw:Err -> {error, Err}; error:_ -> {error, bad_prefix} end. handle_advertise_routes(From, Routes, Data) -> try masque_ip_capsule:encode_route_advertisement(Routes) of Body -> Cap = iolist_to_binary( masque_capsule:encode( ?MASQUE_CAPSULE_ROUTE_ADVERTISEMENT, Body)), case transport_send_data(Data, Cap, false) of ok -> {keep_state, Data, [{reply, From, ok}]}; {error, _} = Err -> {keep_state, Data, [{reply, From, Err}]} end catch error:Reason -> {keep_state, Data, [{reply, From, {error, Reason}}]} end. check_prefix(4, {A,B,C,D}, P) when P >= 0, P =< 32, A >= 0, A =< 255, B >= 0, B =< 255, C >= 0, C =< 255, D >= 0, D =< 255 -> ok; check_prefix(6, Addr, P) when P >= 0, P =< 128, tuple_size(Addr) =:= 8 -> true = lists:all(fun(X) -> is_integer(X) andalso X >= 0 andalso X =< 16#FFFF end, tuple_to_list(Addr)), ok; check_prefix(V, _, P) -> throw({bad_prefix_length, P, V}). %%==================================================================== %% RFC 9484 §8 — negotiated datagram size check %%==================================================================== %% An IPv6 CONNECT-IP tunnel requires the HTTP datagram channel to %% carry a full 1280-byte IPv6 packet (plus the 1-byte context-id %% varint). On H3 we check the effective max datagram size after %% the 2xx response; on H2 there is no per-datagram size limit %% (capsules bound only by max_capsule_size) so the check is a %% no-op. Returns `{error, {mtu_too_low, Got, Min}}` on failure. check_datagram_mtu(#data{transport = h3, conn = C, stream_id = S}) -> case quic_h3:max_datagram_size(C, S) of N when is_integer(N), N >= ?MASQUE_IPV6_MIN_MTU + 1 -> ok; N when is_integer(N), N > 0 -> {error, {mtu_too_low, N, ?MASQUE_IPV6_MIN_MTU}}; _ -> %% 0 or undefined — datagrams unavailable; this was %% already caught in verify_h3_peer_settings. ok end; check_datagram_mtu(#data{transport = h2}) -> ok. %%==================================================================== %% Response validation %%==================================================================== %% RFC 9297 §3.4 forbids `content-length` / `content-type` on a %% capsule-protocol response. RFC 9484 §4 mandates %% `capsule-protocol: ?1` echoed in the 2xx response. validate_response(Headers) -> HasCL = header_present(<<"content-length">>, Headers), HasCT = header_present(<<"content-type">>, Headers), case header_value(<<"capsule-protocol">>, Headers) of <<"?1">> when not HasCL, not HasCT -> ok; <<"?1">> -> {error, malformed_response}; _ -> {error, capsule_protocol_missing} end. header_present(Name, Headers) -> lists:keyfind(Name, 1, Headers) =/= false. header_value(Name, Headers) -> case lists:keyfind(Name, 1, Headers) of {_, V} -> V; false -> undefined end. %%==================================================================== %% Misc %%==================================================================== client_stream_abort(Reason, #data{pool_owner = Pool, stream_id = StreamId} = Data) -> case is_pid(Pool) of true -> masque_upstream_owner:release_stream(Pool, StreamId); false -> _ = (try transport_cancel(Data) catch _:_ -> ok end), ok end, _ = notify_owner_closed(Reason, Data), {stop, Reason, Data}. notify_owner_closed(Reason, #data{owner = Owner, mode = message}) -> Owner ! {masque_closed, self(), Reason}; notify_owner_closed(_, _) -> ok. reply_handshake(#data{handshake_from = undefined}, _Reply) -> ok; reply_handshake(#data{handshake_from = From}, Reply) -> gen_statem:reply(From, Reply). session_info(#data{target = T, ipproto = P, proxy_host = PH, proxy_port = PP, transport = Transport}, State) -> #{state => State, protocol => ip, transport => Transport, proxy => {PH, PP}, target => T, ipproto => P}. cancel_timer(undefined) -> ok; cancel_timer(Ref) -> _ = erlang:cancel_timer(Ref), ok. swap_owner(NewOwner, #data{owner_ref = OldRef} = Data) -> _ = erlang:demonitor(OldRef, [flush]), NewRef = erlang:monitor(process, NewOwner), Data#data{owner = NewOwner, owner_ref = NewRef}. handle_recv_call(From, Timeout, #data{rx_buf = Buf} = Data) -> case queue:out(Buf) of {{value, Bytes}, Buf2} -> {keep_state, Data#data{rx_buf = Buf2}, [{reply, From, {ok, Bytes}}]}; {empty, _} -> TRef = erlang:start_timer(Timeout, self(), {recv_timeout, From}), {keep_state, Data#data{rx_waiters = queue:in({From, TRef}, Data#data.rx_waiters)}} end. deliver_packet(Pkt, #data{mode = message, owner = Owner} = Data) -> Owner ! {masque_ip_packet, self(), Pkt}, Data; deliver_packet(Pkt, #data{mode = queue, rx_waiters = Ws, rx_buf = Buf} = Data) -> case queue:out(Ws) of {{value, {From, TRef}}, Ws2} -> _ = erlang:cancel_timer(TRef), gen_statem:reply(From, {ok, Pkt}), Data#data{rx_waiters = Ws2}; {empty, _} -> case queue:len(Buf) < 1000 of true -> Data#data{rx_buf = queue:in(Pkt, Buf)}; false -> Data end end. drop_waiter(TRef, From, #data{rx_waiters = Ws} = Data) -> Ws2 = queue:filter( fun({F, T}) when F =:= From, T =:= TRef -> gen_statem:reply(F, {error, timeout}), false; (_) -> true end, Ws), Data#data{rx_waiters = Ws2}. cancel_all_waiters(#data{rx_waiters = Ws}) -> _ = queue:fold(fun({From, TRef}, _) -> _ = erlang:cancel_timer(TRef), gen_statem:reply(From, {error, closed}), ok end, ok, Ws), ok. to_bin(X) when is_binary(X) -> X; to_bin(X) when is_list(X) -> list_to_binary(X); to_bin(X) when is_atom(X) -> atom_to_binary(X, utf8). build_authority(Host, Port) -> HostPart = case is_ipv6_literal(Host) of true -> <<"[", Host/binary, "]">>; false -> Host end, iolist_to_binary([HostPart, ":", integer_to_binary(Port)]). is_ipv6_literal(Host) -> case inet:parse_address(binary_to_list(Host)) of {ok, {_, _, _, _, _, _, _, _}} -> true; _ -> false end.