diff --git a/.github/workflows/windows.yml b/.github/workflows/windows.yml index 072f703..3b0a9c3 100644 --- a/.github/workflows/windows.yml +++ b/.github/workflows/windows.yml @@ -113,6 +113,16 @@ jobs: timeout-minutes: 10 run: rebar3 ct + # A run killed by the step timeout prints no case level detail, so the + # suite that hung is known and the case inside it is not. + - name: The Common Test logs, which say which case + if: failure() + uses: actions/upload-artifact@v4 + with: + name: ct-logs + path: _build/test/logs + retention-days: 5 + - name: What the TLS server logged, if anything went wrong if: failure() shell: pwsh diff --git a/guides/configuration.md b/guides/configuration.md index 3fb0d61..77b7c43 100644 --- a/guides/configuration.md +++ b/guides/configuration.md @@ -44,10 +44,16 @@ for where that is and is not true. `inet` is the driver, and is one option away for a machine or a workload that disagrees. The two transports take the same `socket_options`, but the socket transport has -no driver behind it to hold the ones it does not use: `nodelay`, `keepalive`, -`recbuf` and `sndbuf` are set on the socket, `binary`, `{packet, raw}`, -`{active, false}` and `buffer` are accepted and ignored, and anything else is -refused at connect rather than silently dropped. +no driver behind it to hold the ones it does not use. `inet` and `inet6` choose +which address a hostname resolves to; `nodelay`, `keepalive`, `recbuf` and +`sndbuf` are set on the socket; `binary`, `{packet, raw}`, `{active, false}` and +`buffer` are accepted and ignored; anything else is refused at connect rather +than silently dropped. + +`inet6` means v6 or nothing. A host with both an A and a AAAA record resolves to +v4 unless you say otherwise, and a caller that asked for v6 gets an error rather +than a v4 connection, because a quiet fall back is the option accepted and not +honoured. ## A pool diff --git a/src/minato_conn.erl b/src/minato_conn.erl index 41d66e5..90fa1eb 100644 --- a/src/minato_conn.erl +++ b/src/minato_conn.erl @@ -156,6 +156,8 @@ the server issued is the whole of the authorisation. prefix = ~"" :: binary(), prepared = 0 :: non_neg_integer(), limit :: non_neg_integer(), + send_timeout = infinity :: timeout(), + send_timeout_close = false :: boolean(), opts :: opts() }). @@ -362,10 +364,31 @@ rather than four. """. -spec send(conn(), minato_protocol:frontend() | [minato_protocol:frontend()]) -> ok | {error, error()}. -send(#conn{transport = Transport, socket = Socket}, Messages) when is_list(Messages) -> - write(Transport, Socket, [minato_protocol:encode(Message) || Message <- Messages]); -send(#conn{transport = Transport, socket = Socket}, Message) -> - write(Transport, Socket, minato_protocol:encode(Message)). +send(#conn{} = Conn, Messages) when is_list(Messages) -> + written(Conn, [minato_protocol:encode(Message) || Message <- Messages]); +send(#conn{} = Conn, Message) -> + written(Conn, minato_protocol:encode(Message)). + +%% `send_timeout` is the driver's own option and the driver applies it itself. +%% The socket transport has no driver underneath to hold it, so the deadline is +%% passed to the send instead. A write that times out has left the server half a +%% message, so `send_timeout_close` is honoured here as well: the connection is +%% not usable again either way, and a caller that asked for it to be closed +%% should not have to notice that for itself. +-spec written(conn(), iodata()) -> ok | {error, error()}. +written(#conn{transport = minato_socket, send_timeout = Timeout} = Conn, Bytes) -> + timed(Conn, minato_socket:send(handle(Conn), Bytes, Timeout)); +written(#conn{transport = Transport, socket = Socket}, Bytes) -> + write(Transport, Socket, Bytes). + +-spec timed(conn(), ok | {error, term()}) -> ok | {error, error()}. +timed(_Conn, ok) -> + ok; +timed(#conn{send_timeout_close = true} = Conn, {error, timeout}) -> + ok = close(Conn), + {error, {socket, timeout}}; +timed(_Conn, {error, Reason}) -> + {error, {socket, Reason}}. -doc """ Hand the socket to another process. @@ -952,9 +975,31 @@ new(Transport, Socket, Opts) -> limit = maps:get(prepared_statements, Opts, ?DEFAULT_STATEMENTS), prefix = binary:encode_hex(crypto:strong_rand_bytes(8), lowercase), frame_opts = maps:get(frame_opts, Opts, #{}), + send_timeout = send_timeout(Opts), + send_timeout_close = option(send_timeout_close, Opts) =:= true, opts = maps:remove(password, Opts) }. +%% The driver reads these off the socket, so they arrive as socket options +%% rather than as connection options and are read here once instead of on every +%% send. A `getopt` per write would be a NIF call per write, which is the cost +%% this transport exists to avoid. +-spec send_timeout(opts()) -> timeout(). +send_timeout(Opts) -> + case option(send_timeout, Opts) of + Milliseconds when is_integer(Milliseconds), Milliseconds >= 0 -> Milliseconds; + _Absent -> infinity + end. + +-spec option(atom(), opts()) -> term(). +option(Name, Opts) -> + found(Name, maps:get(socket_options, Opts, [])). + +-spec found(atom(), [term()]) -> term(). +found(Name, [{Name, Value} | _Rest]) -> Value; +found(Name, [_Other | Rest]) -> found(Name, Rest); +found(_Name, []) -> undefined. + -spec start(conn(), opts()) -> {ok, conn()} | {error, error()}. start(Conn, Opts) -> User = maps:get(user, Opts), @@ -1187,7 +1232,20 @@ key(Option) -> -spec tls_options(inet:socket_address() | inet:hostname(), opts()) -> [ssl:tls_client_option()]. tls_options(Host, Opts) -> Given = maps:get(ssl_options, Opts, []), - Given ++ [Option || Option <- tls_defaults(Host), not given(Option, Given)]. + Given ++ [Option || Option <- tls_defaults(Host), not superseded(Option, Given)]. + +%% A default is replaced by the option that answers the same question, not +%% merely by one spelled the same way. `cacertfile` and `cacerts` are one +%% question - which authorities are trusted - so a caller pinning a private CA +%% by file has to take the public bundle out of the answer rather than be handed +%% both, where the wider set is the one that decides. +-spec superseded(ssl:tls_client_option(), [ssl:tls_client_option()]) -> boolean(). +superseded(Option, Given) -> + lists:any(fun(Candidate) -> question(Candidate) =:= question(Option) end, Given). + +-spec question(term()) -> term(). +question({cacertfile, _Authorities}) -> cacerts; +question(Option) -> key(Option). -spec tls_defaults(inet:socket_address() | inet:hostname()) -> [ssl:tls_client_option()]. tls_defaults(Host) -> diff --git a/src/minato_socket.erl b/src/minato_socket.erl index dadc7db..b584bb4 100644 --- a/src/minato_socket.erl +++ b/src/minato_socket.erl @@ -47,7 +47,7 @@ sends its `SSLRequest` through this module and hands the upgraded socket to including the certificate hash SCRAM binds to. """. --export([connect/4, send/2, recv/3, close/1, controlling_process/2, setopts/2]). +-export([connect/4, send/2, send/3, recv/3, close/1, controlling_process/2, setopts/2]). -export([activate/1, cancel/2]). %% eqWAlizer reads the eight words of an `inet:ip6_address()` as `integer()` and @@ -56,6 +56,11 @@ including the certificate hash SCRAM binds to. %% this. The v4 half of the same function checks. -eqwalizer({nowarn_function, target/2}). +%% `socket:send/3` answers `{ok, RestData}` where `RestData` is the tail of the +%% `iodata()` that went in, but its spec says `term()`, so eqWAlizer reads a +%% list tail as a bare list rather than as an iolist. +-eqwalizer({nowarn_function, sending/3}). + -doc """ A socket handle. @@ -79,22 +84,42 @@ where it hands over the bytes. Open a connection. Takes the option list `m:gen_tcp` would take, so a caller does not have to know -which transport it got. `binary`, `{mode, binary}`, `{active, false}`, -`{packet, raw}` and `{buffer, _}` are what this module always does or has no -equivalent of, and are accepted and ignored. `{nodelay, _}`, `{keepalive, _}`, -`{recbuf, _}` and `{sndbuf, _}` are set on the socket. Anything else is refused -rather than silently dropped: this is the default transport, and an option that -did something through the driver must not quietly stop doing it here. +which transport it got. + +`inet` and `inet6` choose the address family a hostname resolves to, as they do +for the driver. `{nodelay, _}`, `{keepalive, _}`, `{recbuf, _}` and +`{sndbuf, _}` are set on the socket. `{send_timeout, _}` and +`{send_timeout_close, _}` are read by `m:minato_conn` and applied to `send/3`, +because there is no driver here to hold them. `binary`, `{mode, binary}`, +`{active, false}`, `{packet, raw}` and `{buffer, _}` are what this module always +does or has no equivalent of, and are accepted and ignored. + +Anything else is refused rather than silently dropped: this is the default +transport, and an option that did something through the driver must not quietly +stop doing it here. That cuts both ways, so the list above is what real callers +pass rather than what a socket happens to be settable to - `inet` is neither, +and refusing it broke every deployment that names its address family. """. -spec connect( inet:socket_address() | inet:hostname(), inet:port_number(), [term()], timeout() ) -> {ok, socket()} | {error, term()}. connect(Host, Port, Opts, Timeout) -> - case address(Host, Port) of + case address(Host, Port, family(Opts)) of {ok, {Domain, Address}} -> opened(Domain, Address, Opts, Timeout); {error, Reason} -> {error, Reason} end. +-doc false. +-spec family([term()]) -> inet | inet6 | any. +%% `inet` and `inet6` are not options a socket is set to, which is why asking +%% `setopt/2` about them was the wrong question: they choose which address a +%% name resolves to, and the driver reads them before it connects. `any` is +%% both, in the order `resolved/3` tries them. +family([inet | _Rest]) -> inet; +family([inet6 | _Rest]) -> inet6; +family([_Other | Rest]) -> family(Rest); +family([]) -> any. + -doc """ Send every byte, or say why not. @@ -103,12 +128,60 @@ still answer with what is left rather than with `ok`, so what is left is sent. """. -spec send(socket(), iodata()) -> ok | {error, term()}. send(Socket, Data) -> - case socket:send(Socket, Data) of + send(Socket, Data, infinity). + +-doc """ +Send every byte, or say why not, within `Timeout`. + +The deadline is for the whole of it rather than for each attempt, because a +peer that accepts one byte per attempt would otherwise never time out. What is +left when the deadline passes is not sent and not retried: a `Query` the server +has half of is a connection that has to be given up on, which is what +`send_timeout` is for. +""". +-spec send(socket(), iodata(), timeout()) -> ok | {error, term()}. +send(Socket, Data, infinity) -> + sent(Socket, Data, infinity); +send(Socket, Data, Timeout) -> + sent(Socket, Data, erlang:monotonic_time(millisecond) + Timeout). + +-type written() :: ok | {ok, iodata()} | {error, term() | {term(), iodata()}}. + +%% `socket:send/3` answers more shapes than its spec narrows to, so the union +%% this module matches on is stated here rather than inferred. What comes back +%% as `RestData` is the tail of the `iodata()` that went in, which eqWAlizer +%% reads as a bare list because the spec says `term()`. +-spec sending(socket(), iodata(), timeout()) -> written(). +sending(Socket, Data, Timeout) -> + case socket:send(Socket, Data, Timeout) of ok -> ok; - {ok, Rest} -> send(Socket, Rest); + {ok, Rest} when is_binary(Rest); is_list(Rest) -> {ok, Rest}; {error, Reason} -> {error, Reason} end. +-spec sent(socket(), iodata(), integer() | infinity) -> ok | {error, term()}. +sent(Socket, Data, Deadline) -> + case remaining(Deadline) of + 0 -> {error, timeout}; + Left -> written(Socket, Data, Deadline, sending(Socket, Data, Left)) + end. + +-spec written(socket(), iodata(), integer() | infinity, written()) -> ok | {error, term()}. +written(_Socket, _Data, _Deadline, ok) -> ok; +written(Socket, _Data, Deadline, {ok, Rest}) -> sent(Socket, Rest, Deadline); +written(_Socket, _Data, _Deadline, {error, {timeout, _Rest}}) -> {error, timeout}; +written(_Socket, _Data, _Deadline, {error, {Reason, _Rest}}) -> {error, Reason}; +written(_Socket, _Data, _Deadline, {error, Reason}) -> {error, Reason}. + +-spec remaining(integer() | infinity) -> timeout(). +remaining(Deadline) when is_integer(Deadline) -> + case Deadline - erlang:monotonic_time(millisecond) of + Left when Left > 0 -> Left; + _Passed -> 0 + end; +remaining(infinity) -> + infinity. + -doc """ Read `Length` bytes, or whatever has arrived when `Length` is 0. @@ -218,6 +291,10 @@ setopt(_Socket, {mode, binary}) -> ok; setopt(_Socket, {packet, raw}) -> ok; setopt(_Socket, {packet, 0}) -> ok; setopt(_Socket, {buffer, _Driver_side}) -> ok; +setopt(_Socket, inet) -> ok; +setopt(_Socket, inet6) -> ok; +setopt(_Socket, {send_timeout, _Read_by_the_connection}) -> ok; +setopt(_Socket, {send_timeout_close, _Read_by_the_connection}) -> ok; setopt(_Socket, Option) -> {error, {unsupported_socket_option, Option}}. %%---------------------------------------------------------------------- @@ -270,29 +347,38 @@ failed(Socket, Reason) -> ok = close(Socket), {error, Reason}. --spec address(inet:socket_address() | inet:hostname(), inet:port_number()) -> +-spec address(inet:socket_address() | inet:hostname(), inet:port_number(), inet | inet6 | any) -> {ok, {socket:domain(), socket:sockaddr()}} | {error, term()}. -address({local, Path}, _Port) -> +address({local, Path}, _Port, _Family) -> {ok, {local, #{family => local, path => iolist_to_binary(Path)}}}; -address(Host, Port) when is_list(Host); is_atom(Host) -> - resolved(Host, Port); -address(Address, Port) -> +address(Host, Port, Family) when is_list(Host); is_atom(Host) -> + resolved(Host, Port, Family); +address(Address, Port, _Family) -> {ok, target(Address, Port)}. --spec resolved(inet:hostname(), inet:port_number()) -> - {ok, {socket:domain(), socket:sockaddr()}} | {error, term()}. -resolved(Host, Port) -> - case inet:getaddr(Host, inet) of - {ok, Address} -> {ok, target(Address, Port)}; - {error, _Not_a_v4_address} -> resolved6(Host, Port) - end. - --spec resolved6(inet:hostname(), inet:port_number()) -> +-spec resolved(inet:hostname(), inet:port_number(), inet | inet6 | any) -> {ok, {socket:domain(), socket:sockaddr()}} | {error, term()}. -resolved6(Host, Port) -> - case inet:getaddr(Host, inet6) of +%% A name that has both an A and a AAAA record resolves to the v4 address unless +%% the caller said otherwise, which is what the driver does. A caller that said +%% `inet6` gets v6 or an error, never a quiet v4: a host reachable both ways is +%% exactly where a silent fallback would hide the fact that the option did +%% nothing. +resolved(Host, Port, inet6) -> + resolved(Host, Port, inet6, fun(Reason) -> {error, Reason} end); +resolved(Host, Port, inet) -> + resolved(Host, Port, inet, fun(Reason) -> {error, Reason} end); +resolved(Host, Port, any) -> + resolved(Host, Port, inet, fun(_Not_a_v4_address) -> resolved(Host, Port, inet6) end). + +-type resolution() :: {ok, {socket:domain(), socket:sockaddr()}} | {error, term()}. + +-spec resolved( + inet:hostname(), inet:port_number(), inet | inet6, fun((term()) -> resolution()) +) -> resolution(). +resolved(Host, Port, Family, Otherwise) -> + case inet:getaddr(Host, Family) of {ok, Address} -> {ok, target(Address, Port)}; - {error, Reason} -> {error, Reason} + {error, Reason} -> Otherwise(Reason) end. -spec target(inet:ip_address(), inet:port_number()) -> diff --git a/test/minato_tls_SUITE.erl b/test/minato_tls_SUITE.erl index 4041119..34440b7 100644 --- a/test/minato_tls_SUITE.erl +++ b/test/minato_tls_SUITE.erl @@ -26,7 +26,7 @@ Skips when no TLS server answers. -include_lib("stdlib/include/assert.hrl"). --export([all/0, init_per_suite/1, end_per_suite/1]). +-export([suite/0, all/0, init_per_suite/1, end_per_suite/1]). -export([ a_verified_handshake_opens_a_working_session/1, the_server_reports_the_session_as_encrypted/1, @@ -38,11 +38,22 @@ Skips when no TLS server answers. the_exchange_is_bound_to_the_tls_session/1, channel_binding_can_be_required/1, a_required_binding_is_refused_without_tls/1, - the_certificate_hash_is_the_one_the_server_computes/1 + the_certificate_hash_is_the_one_the_server_computes/1, + a_ca_pinned_by_file_is_the_ca_that_decides/1 ]). -define(TIMEOUT, 15000). +%% Without one of these a case that blocks runs into Common Test's thirty minute +%% default, which is longer than CI gives the whole step, so a hang arrives as a +%% killed job with no case name and no stack. Every operation in this suite is +%% bounded by a five second connect or a ten second read, so a case that reaches +%% sixty seconds is stuck somewhere that has no deadline of its own, and the +%% timetrap's stack trace is the thing worth having. +-spec suite() -> [{timetrap, {seconds, pos_integer()}}]. +suite() -> + [{timetrap, {seconds, 60}}]. + -spec all() -> [atom()]. all() -> [ @@ -56,7 +67,8 @@ all() -> the_exchange_is_bound_to_the_tls_session, channel_binding_can_be_required, a_required_binding_is_refused_without_tls, - the_certificate_hash_is_the_one_the_server_computes + the_certificate_hash_is_the_one_the_server_computes, + a_ca_pinned_by_file_is_the_ca_that_decides ]. -spec init_per_suite([{atom(), term()}]) -> [{atom(), term()}] | {skip, term()}. @@ -98,6 +110,17 @@ a_hostname_the_certificate_does_not_cover_is_refused(_Config) -> Elsewhere = (verified())#{host => "127.0.0.1", ssl_options => [{cacerts, cacerts()}]}, ?assertMatch({error, {tls, _Reason}}, minato_conn:connect(Elsewhere)). +%% `cacertfile` and `cacerts` answer the same question, so a caller who pins a +%% private CA by file must not also be handed the public bundle: two answers to +%% "which authorities are trusted" is the wider one deciding. The test CA is not +%% in any public bundle, so a handshake that succeeds is one this file carried. +-spec a_ca_pinned_by_file_is_the_ca_that_decides([{atom(), term()}]) -> ok. +a_ca_pinned_by_file_is_the_ca_that_decides(_Config) -> + Pinned = (verified())#{ssl_options => [{cacertfile, certificate("ca.crt")}]}, + {ok, Conn} = minato_conn:connect(Pinned), + {ok, #{rows := [{1}]}, Ready} = minato_query:query(Conn, ~"SELECT $1::int4", [1]), + ok = minato_conn:close(Ready). + -spec verification_can_be_turned_off_deliberately([{atom(), term()}]) -> ok. verification_can_be_turned_off_deliberately(_Config) -> Unverified = (verified())#{host => "127.0.0.1", ssl_options => [{verify, verify_none}]}, diff --git a/test/minato_transport_SUITE.erl b/test/minato_transport_SUITE.erl index 198f560..0b582a2 100644 --- a/test/minato_transport_SUITE.erl +++ b/test/minato_transport_SUITE.erl @@ -31,6 +31,9 @@ Skips when no server answers. a_transaction_commits/1, a_result_set_spanning_many_reads_arrives_whole/1, a_socket_option_the_driver_takes_is_taken_here/1, + an_address_family_is_chosen_not_refused/1, + an_address_family_is_not_quietly_swapped/1, + a_send_to_a_peer_that_stopped_reading_gives_up/1, an_option_this_transport_cannot_honour_is_refused/1, a_pool_lends_a_connection_and_takes_it_back/1, a_pooled_connection_outlives_the_process_that_opened_it/1, @@ -62,6 +65,9 @@ cases() -> a_transaction_commits, a_result_set_spanning_many_reads_arrives_whole, a_socket_option_the_driver_takes_is_taken_here, + an_address_family_is_chosen_not_refused, + an_address_family_is_not_quietly_swapped, + a_send_to_a_peer_that_stopped_reading_gives_up, an_option_this_transport_cannot_honour_is_refused, a_pool_lends_a_connection_and_takes_it_back, a_pooled_connection_outlives_the_process_that_opened_it, @@ -153,14 +159,68 @@ a_socket_option_the_driver_takes_is_taken_here(Config) -> {ok, #{rows := [{1}]}, Ready} = minato_query:query(Conn, ~"SELECT $1::int4", [1]), ok = minato_conn:close(Ready). +%% The option every deployment passes. `inet` and `inet6` are not settings a +%% socket takes, so a transport that asked `setopts` about them refused the +%% config its own consumers were documented to use, and did it after the TCP +%% connect, where a pool reports it as `disconnected` rather than as a bad +%% option. +an_address_family_is_chosen_not_refused(Config) -> + Conn = connected(Config, [inet]), + {ok, #{rows := [{1}]}, Ready} = minato_query:query(Conn, ~"SELECT $1::int4", [1]), + ok = minato_conn:close(Ready). + +%% v6 asked for is v6 or nothing. A silent fall back to v4 would be the same +%% defect one layer down: the option accepted and not honoured. +an_address_family_is_not_quietly_swapped(Config) -> + Opts = (opts(Config))#{host => "127.0.0.1", socket_options => [inet6]}, + ?assertMatch({error, {socket, _Not_a_v6_address}}, minato_conn:connect(Opts)). + +%% A send with no deadline waits for a peer that has stopped reading for as long +%% as the peer likes, holding the pool slot it borrowed. The driver has +%% `send_timeout` for that and the socket transport had nowhere to put it, so +%% the mitigation existed for one transport and not the default one. +%% +%% A listening socket nobody accepts from fills its buffers and then accepts +%% nothing, which is that peer without needing one. +a_send_to_a_peer_that_stopped_reading_gives_up(Config) -> + {ok, Listener} = gen_tcp:listen(0, [binary, {active, false}, {backlog, 1}]), + {ok, Port} = inet:port(Listener), + Opts = (opts(Config))#{ + host => "127.0.0.1", + port => Port, + connect_timeout => ?TIMEOUT, + socket_options => [{send_timeout, 200}] + }, + ?assertEqual({error, {socket, timeout}}, filled(Opts)), + ok = gen_tcp:close(Listener). + +%% The startup exchange is small enough to fit in a buffer, so the connect +%% itself would not block. Writing until something gives is what reaches the +%% deadline; a server that answers is a different case and has its own tests. +filled(Opts) -> + case minato_conn:connect(Opts) of + {ok, Conn} -> filled(Conn, binary:copy(~"x", 1048576), 64); + {error, Reason} -> {error, Reason} + end. + +filled(Conn, _Bytes, 0) -> + ok = minato_conn:close(Conn), + never_filled; +filled(Conn, Bytes, Left) -> + case minato_conn:send(Conn, {query, Bytes}) of + ok -> filled(Conn, Bytes, Left - 1); + {error, Reason} -> {error, Reason} + end. + an_option_this_transport_cannot_honour_is_refused(Config) -> - Opts = (opts(Config))#{socket_options => [{send_timeout, 5000}]}, + Opts = (opts(Config))#{socket_options => [{priority, 3}]}, ok = judged(transport(Config), minato_conn:connect(Opts)). -%% `send_timeout` is the driver's, and this module has no answer for it. The -%% refusal is the point: an option that did something through the driver has to -%% either keep doing it or say that it stopped, never quietly become nothing. -judged(socket, {error, {socket, {unsupported_socket_option, {send_timeout, 5000}}}}) -> ok; +%% The refusal is the point: an option that did something through the driver has +%% to either keep doing it or say that it stopped, never quietly become nothing. +%% Which option this is matters less than that the answer is loud - and the list +%% of what is honoured grows as real callers turn out to pass things. +judged(socket, {error, {socket, {unsupported_socket_option, {priority, 3}}}}) -> ok; judged(inet, {ok, Conn}) -> minato_conn:close(Conn). %%---------------------------------------------------------------------- @@ -368,6 +428,10 @@ connected(Config) -> {ok, Conn} = minato_conn:connect(opts(Config)), Conn. +connected(Config, Socket_options) -> + {ok, Conn} = minato_conn:connect((opts(Config))#{socket_options => Socket_options}), + Conn. + secured(Config) -> (opts(Config))#{ host => env("MINATO_TLS_HOST", "localhost"),