Merge pull request #156 from seriyps/scram-sha-auth

Scram sha auth

include/protocol.hrl

@@ -16,6 +16,7 @@
 -define(SIMPLEQUERY, $Q).
 -define(AUTHENTICATION_REQUEST, $R).
 -define(SYNC, $S).
+-define(SASL_ANY_RESPONSE, $p).
 
 %% Parameters
 

src/commands/epgsql_cmd_connect.erl

@@ -12,16 +12,34 @@
 -type connect_error() ::
         invalid_authorization_specification
       | invalid_password
-      | {unsupported_auth_method, integer()}
+      | {unsupported_auth_method,
+         kerberosV5 | crypt | scm | gss | sspi | {unknown, integer()} | {sasl, [binary()]}}
+      | {sasl_server_final, any()}
       | epgsql:query_error().
 
 -include("epgsql.hrl").
 -include("protocol.hrl").
 
+-type auth_fun() :: fun((init | binary(), _, _) ->
+                                     {send, byte(), iodata(), any()}
+                                   | ok
+                                   | {error, any()}
+                                   | unknown).
+
 -record(connect,
         {opts :: list(),
-         auth_method,
-         stage = connect :: connect | auth | initialization}).
+         auth_fun :: auth_fun() | undefined,
+         auth_state :: any() | undefined,
+         auth_send :: {integer(), iodata()} | undefined,
+         stage = connect :: connect | maybe_auth | auth | initialization}).
+
+-define(SCRAM_AUTH_METHOD, <<"SCRAM-SHA-256">>).
+-define(AUTH_OK, 0).
+-define(AUTH_CLEARTEXT, 3).
+-define(AUTH_MD5, 5).
+-define(AUTH_SASL, 10).
+-define(AUTH_SASL_CONTINUE, 11).
+-define(AUTH_SASL_FINAL, 12).
 
 init({Host, Username, Password, Opts}) ->
     Opts1 = [{host, Host},
@@ -73,21 +91,13 @@ execute(PgSock, #connect{opts = Opts, stage = connect} = State) ->
                           undefined -> PgSock2;
                           Async -> epgsql_sock:set_attr(async, Async, PgSock2)
                       end,
-            {ok, PgSock3, State#connect{stage = auth}};
+            {ok, PgSock3, State#connect{stage = maybe_auth}};
         {error, Reason} = Error ->
             {stop, Reason, Error, PgSock}
     end;
-execute(PgSock, #connect{stage = auth, auth_method = cleartext, opts = Opts} = St) ->
-    Password = get_val(password, Opts),
-    epgsql_sock:send(PgSock, ?PASSWORD, [Password, 0]),
-    {ok, PgSock, St};
-execute(PgSock, #connect{stage = auth, auth_method = {md5, Salt}, opts = Opts} = St) ->
-    User = get_val(username, Opts),
-    Password = get_val(password, Opts),
-    Digest1 = hex(erlang:md5([Password, User])),
-    Str = ["md5", hex(erlang:md5([Digest1, Salt])), 0],
-    epgsql_sock:send(PgSock, ?PASSWORD, Str),
-    {ok, PgSock, St}.
+execute(PgSock, #connect{stage = auth, auth_send = {PacketId, Data}} = St) ->
+    epgsql_sock:send(PgSock, PacketId, Data),
+    {ok, PgSock, St#connect{auth_send = undefined}}.
 
 
 maybe_ssl(S, false, _, PgSock) ->
@@ -114,30 +124,98 @@ maybe_ssl(S, Flag, Opts, PgSock) ->
             end
     end.
 
-%% --- Auth ---
-
-%% AuthenticationOk
-handle_message(?AUTHENTICATION_REQUEST, <<0:?int32>>, Sock, State) ->
-    {noaction, Sock, State#connect{stage = initialization}};
+%% Auth sub-protocol
+
+auth_init(<<?AUTH_CLEARTEXT:?int32>>, Sock, St) ->
+    auth_init(fun auth_cleartext/3, undefined, Sock, St);
+auth_init(<<?AUTH_MD5:?int32, Salt:4/binary>>, Sock, St) ->
+    auth_init(fun auth_md5/3, Salt, Sock, St);
+auth_init(<<?AUTH_SASL:?int32, MethodsB/binary>>, Sock, St) ->
+    Methods = epgsql_wire:decode_strings(MethodsB),
+    case lists:member(?SCRAM_AUTH_METHOD, Methods) of
+        true ->
+            auth_init(fun auth_scram/3, undefined, Sock, St);
+        false ->
+            {stop, normal, {error, {unsupported_auth_method,
+                                    {sasl, lists:delete(<<>>, Methods)}}}}
+    end;
+auth_init(<<M:?int32, _/binary>>, Sock, _St) ->
+    Method = case M of
+                 2 -> kerberosV5;
+                 4 -> crypt;
+                 6 -> scm;
+                 7 -> gss;
+                 8 -> sspi;
+                 _ -> {unknown, M}
+             end,
+    {stop, normal, {error, {unsupported_auth_method, Method}}, Sock}.
+
+auth_init(Fun, InitState, PgSock, St) ->
+    auth_handle(init, PgSock, St#connect{auth_fun = Fun, auth_state = InitState,
+                                         stage = auth}).
+
+auth_handle(Data, PgSock, #connect{auth_fun = Fun, auth_state = AuthSt} = St) ->
+    case Fun(Data, AuthSt, St) of
+        {send, SendPacketId, SendData, AuthSt1} ->
+            {requeue, PgSock, St#connect{auth_state = AuthSt1,
+                                         auth_send = {SendPacketId, SendData}}};
+        ok -> {noaction, PgSock, St};
+        {error, Reason} ->
+            {stop, normal, {error, Reason}};
+        unknown -> unknown
+    end.
 
 %% AuthenticationCleartextPassword
-handle_message(?AUTHENTICATION_REQUEST, <<3:?int32>>, Sock, St) ->
-    {requeue, Sock, St#connect{stage = auth, auth_method = cleartext}};
+auth_cleartext(init, _AuthState, #connect{opts = Opts}) ->
+    Password = get_val(password, Opts),
+    {send, ?PASSWORD, [Password, 0], undefined};
+auth_cleartext(_, _, _) -> unknown.
 
 %% AuthenticationMD5Password
-handle_message(?AUTHENTICATION_REQUEST, <<5:?int32, Salt:4/binary>>, Sock, St) ->
-    {requeue, Sock, St#connect{stage = auth, auth_method = {md5, Salt}}};
+auth_md5(init, Salt, #connect{opts = Opts}) ->
+    User = get_val(username, Opts),
+    Password = get_val(password, Opts),
+    Digest1 = hex(erlang:md5([Password, User])),
+    Str = ["md5", hex(erlang:md5([Digest1, Salt])), 0],
+    {send, ?PASSWORD, Str, undefined};
+auth_md5(_, _, _) -> unknown.
 
-handle_message(?AUTHENTICATION_REQUEST, <<M:?int32, _/binary>>, Sock, _State) ->
-    Method = case M of
-        2 -> kerberosV5;
-        4 -> crypt;
-        6 -> scm;
-        7 -> gss;
-        8 -> sspi;
-        _ -> unknown
-    end,
-    {stop, normal, {error, {unsupported_auth_method, Method}}, Sock};
+%% AuthenticationSASL
+auth_scram(init, undefined, #connect{opts = Opts}) ->
+    User = get_val(username, Opts),
+    Nonce = epgsql_scram:get_nonce(16),
+    ClientFirst = epgsql_scram:get_client_first(User, Nonce),
+    SaslInitialResponse = [?SCRAM_AUTH_METHOD, 0, <<(iolist_size(ClientFirst)):?int32>>, ClientFirst],
+    {send, ?SASL_ANY_RESPONSE, SaslInitialResponse, {auth_request, Nonce}};
+auth_scram(<<?AUTH_SASL_CONTINUE:?int32, ServerFirst/binary>>, {auth_request, Nonce}, #connect{opts = Opts}) ->
+    User = get_val(username, Opts),
+    Password = get_val(password, Opts),
+    ServerFirstParts = epgsql_scram:parse_server_first(ServerFirst, Nonce),
+    {ClientFinalMessage, ServerProof} = epgsql_scram:get_client_final(ServerFirstParts, Nonce, User, Password),
+    {send, ?SASL_ANY_RESPONSE, ClientFinalMessage, {server_final, ServerProof}};
+auth_scram(<<?AUTH_SASL_FINAL:?int32, ServerFinalMsg/binary>>, {server_final, ServerProof}, _Conn) ->
+    case epgsql_scram:parse_server_final(ServerFinalMsg) of
+        {ok, ServerProof} -> ok;
+        Other -> {error, {sasl_server_final, Other}}
+    end;
+auth_scram(_, _, _) ->
+    unknown.
+
+
+%% --- Auth ---
+
+%% AuthenticationOk
+handle_message(?AUTHENTICATION_REQUEST, <<?AUTH_OK:?int32>>, Sock, State) ->
+    {noaction, Sock, State#connect{stage = initialization,
+                                   auth_fun = undefined,
+                                   auth_state = undefned,
+                                   auth_send = undefined}};
+
+handle_message(?AUTHENTICATION_REQUEST, Message, Sock, #connect{stage = Stage} = St) when Stage =/= auth ->
+    auth_init(Message, Sock, St);
+
+handle_message(?AUTHENTICATION_REQUEST, Packet, Sock, #connect{stage = auth} = St) ->
+    auth_handle(Packet, Sock, St);
 
 %% --- Initialization ---
 
@@ -153,7 +231,8 @@ handle_message(?READY_FOR_QUERY, _, Sock, _State) ->
 
 
 %% ErrorResponse
-handle_message(?ERROR, Err, Sock, #connect{stage = auth} = _State) ->
+handle_message(?ERROR, Err, Sock, #connect{stage = Stage} = _State) when Stage == auth;
+                                                                         Stage == maybe_auth ->
     Why = case Err#error.code of
         <<"28000">> -> invalid_authorization_specification;
         <<"28P01">> -> invalid_password;

src/epgsql_scram.erl

@@ -0,0 +1,172 @@
+%%% coding: utf-8
+%%% @doc
+%%% SCRAM--SHA-256 helper functions
+%%% See
+%%% https://www.postgresql.org/docs/current/static/sasl-authentication.html
+%%% https://en.wikipedia.org/wiki/Salted_Challenge_Response_Authentication_Mechanism
+%%% https://tools.ietf.org/html/rfc7677
+%%% https://tools.ietf.org/html/rfc5802
+%%% @end
+
+-module(epgsql_scram).
+-export([get_nonce/1,
+         get_client_first/2,
+         get_client_final/4,
+         parse_server_first/2,
+         parse_server_final/1]).
+-export([hi/3,
+         hmac/2,
+         h/1,
+         bin_xor/2]).
+
+-type nonce() :: binary().
+-type server_first() :: [{nonce, nonce()} |
+                         {salt, binary()} |
+                         {i, pos_integer()} |
+                         {raw, binary()}].
+
+-spec get_client_first(iodata(), nonce()) -> iodata().
+get_client_first(UserName, Nonce) ->
+    %% Username is ignored by postgresql
+    [<<"n,,">> | client_first_bare(UserName, Nonce)].
+
+client_first_bare(UserName, Nonce) ->
+    [<<"n=">>, UserName, <<",r=">>, Nonce].
+
+%% @doc Generate unique ASCII string.
+%% Resulting string length isn't guaranteed, but it's guaranteed to be unique and will
+%% contain `NumRandomBytes' of random data.
+-spec get_nonce(pos_integer()) -> nonce().
+get_nonce(NumRandomBytes) when NumRandomBytes < 255 ->
+    Random = crypto:strong_rand_bytes(NumRandomBytes),
+    Unique = binary:encode_unsigned(unique()),
+    NonceBin = <<NumRandomBytes, Random:NumRandomBytes/binary, Unique/binary>>,
+    base64:encode(NonceBin).
+
+-spec parse_server_first(binary(), nonce()) -> server_first().
+parse_server_first(ServerFirst, ClientNonce) ->
+    PartsB = binary:split(ServerFirst, <<",">>, [global]),
+    (length(PartsB) == 3) orelse error({invalid_server_first, ServerFirst}),
+    Parts =
+        lists:map(
+          fun(<<"r=", R/binary>>) ->
+                  {nonce, R};
+             (<<"s=", S/binary>>) ->
+                  {salt, base64:decode(S)};
+             (<<"i=", I/binary>>) ->
+                  {i, binary_to_integer(I)}
+          end, PartsB),
+    check_nonce(ClientNonce, proplists:get_value(nonce, Parts)),
+    [{raw, ServerFirst} | Parts].
+
+%% SaltedPassword  := Hi(Normalize(password), salt, i)
+%% ClientKey       := HMAC(SaltedPassword, "Client Key")
+%% StoredKey       := H(ClientKey)
+%% AuthMessage     := client-first-message-bare + "," + server-first-message + "," + client-final-message-without-proof
+%% ClientSignature := HMAC(StoredKey, AuthMessage)
+%% ClientProof     := ClientKey XOR ClientSignature
+-spec get_client_final(server_first(), nonce(), iodata(), iodata()) ->
+                              {ClientFinal :: iodata(), ServerSignature :: binary()}.
+get_client_final(SrvFirst, ClientNonce, UserName, Password) ->
+    ChannelBinding = <<"c=biws">>,                 %channel-binding isn't implemented
+    Nonce = [<<"r=">>, proplists:get_value(nonce, SrvFirst)],
+
+    Salt = proplists:get_value(salt, SrvFirst),
+    I = proplists:get_value(i, SrvFirst),
+
+    SaltedPassword = hi(normalize(Password), Salt, I),
+    ClientKey = hmac(SaltedPassword, "Client Key"),
+    StoredKey = h(ClientKey),
+    ClientFirstBare = client_first_bare(UserName, ClientNonce),
+    ServerFirst = proplists:get_value(raw, SrvFirst),
+    ClientFinalWithoutProof = [ChannelBinding, ",", Nonce],
+    AuthMessage = [ClientFirstBare, ",", ServerFirst, ",", ClientFinalWithoutProof],
+    ClientSignature = hmac(StoredKey, AuthMessage),
+    ClientProof = bin_xor(ClientKey, ClientSignature),
+
+    ServerKey = hmac(SaltedPassword, "Server Key"),
+    ServerSignature = hmac(ServerKey, AuthMessage),
+
+    {[ClientFinalWithoutProof, ",p=", base64:encode(ClientProof)], ServerSignature}.
+
+-spec parse_server_final(binary()) -> {ok, binary()} | {error, binary()}.
+parse_server_final(<<"v=", ServerFinal/binary>>) ->
+    [ServerFinal1 | _] = binary:split(ServerFinal, <<",">>),
+    {ok, base64:decode(ServerFinal1)};
+parse_server_final(<<"e=", ServerError/binary>>) ->
+    {error, ServerError}.
+
+%% Helpers
+
+%% TODO: implement according to rfc3454
+normalize(Str) ->
+    lists:all(fun is_ascii_non_control/1, unicode:characters_to_list(Str, utf8))
+        orelse error({scram_non_ascii_password, Str}),
+    Str.
+
+is_ascii_non_control(C) when C > 16#1F, C < 16#7F -> true;
+is_ascii_non_control(_) -> false.
+
+check_nonce(ClientNonce, ServerNonce) ->
+    Size = size(ClientNonce),
+    <<ClientNonce:Size/binary, _/binary>> = ServerNonce,
+    true.
+
+hi(Str, Salt, I) ->
+    U1 = hmac(Str, <<Salt/binary, 1:32/integer-big>>),
+    hi1(Str, U1, U1, I - 1).
+
+hi1(_Str, _U, Hi, 0) ->
+    Hi;
+hi1(Str, U, Hi, I) ->
+    U2 = hmac(Str, U),
+    Hi1 = bin_xor(Hi, U2),
+    hi1(Str, U2, Hi1, I - 1).
+
+hmac(Key, Str) ->
+    crypto:hmac(sha256, Key, Str).
+
+h(Str) ->
+    crypto:hash(sha256, Str).
+
+%% word 'xor' is reserved
+bin_xor(B1, B2) ->
+    crypto:exor(B1, B2).
+
+
+-ifdef(FAST_MAPS).
+unique() ->
+    erlang:unique_integer([positive]).
+-else.
+unique() ->
+    %% POSIX timestamp microseconds
+    {Mega, Secs, Micro} = erlang:now(),
+    (Mega * 1000000 + Secs) * 1000000 + Micro.
+-endif.
+
+
+-ifdef(TEST).
+-include_lib("eunit/include/eunit.hrl").
+
+exchange_test() ->
+    Password = <<"foobar">>,
+    Nonce = <<"9IZ2O01zb9IgiIZ1WJ/zgpJB">>,
+    Username = <<>>,
+
+    ClientFirst = <<"n,,n=,r=9IZ2O01zb9IgiIZ1WJ/zgpJB">>,
+    ServerFirst = <<"r=9IZ2O01zb9IgiIZ1WJ/zgpJBjx/oIRLs02gGSHcw1KEty3eY,s=fs3IXBy7U7+IvVjZ,i=4096">>,
+    ClientFinal = <<"c=biws,r=9IZ2O01zb9IgiIZ1WJ/zgpJBjx/oIRLs02gGSHcw1KEty3eY,p=AmNKosjJzS31NTlQ"
+                    "YNs5BTeQjdHdk7lOflDo5re2an8=">>,
+    ServerFinal = <<"v=U+ppxD5XUKtradnv8e2MkeupiA8FU87Sg8CXzXHDAzw=">>,
+
+    ?assertEqual(ClientFirst, iolist_to_binary(get_client_first(Username, Nonce))),
+    SF = parse_server_first(ServerFirst, Nonce),
+    {CF, ServerProof} = get_client_final(SF, Nonce, Username, Password),
+    ?assertEqual(ClientFinal, iolist_to_binary(CF)),
+    ?assertEqual({ok, ServerProof}, parse_server_final(ServerFinal)).
+
+normalize_test() ->
+    ?assertEqual(<<"123 !~">>, normalize(<<"123 !~">>)),
+    ?assertError({scram_non_ascii_password, _}, normalize(<<"привет"/utf8>>)).
+
+-endif.

test/data/test_schema.sql

@@ -12,12 +12,16 @@ CREATE USER epgsql_test_md5 WITH PASSWORD 'epgsql_test_md5';
 CREATE USER epgsql_test_cleartext WITH PASSWORD 'epgsql_test_cleartext';
 CREATE USER epgsql_test_cert;
 CREATE USER epgsql_test_replication WITH REPLICATION PASSWORD 'epgsql_test_replication';
+SET password_encryption TO 'scram-sha-256';
+CREATE USER epgsql_test_scram WITH PASSWORD 'epgsql_test_scram';
+SET password_encryption TO 'md5';
 
 CREATE DATABASE epgsql_test_db1 WITH ENCODING 'UTF8';
 CREATE DATABASE epgsql_test_db2 WITH ENCODING 'UTF8';
 
 GRANT ALL ON DATABASE epgsql_test_db1 to epgsql_test;
 GRANT ALL ON DATABASE epgsql_test_db1 to epgsql_test_md5;
+GRANT ALL ON DATABASE epgsql_test_db1 to epgsql_test_scram;
 GRANT ALL ON DATABASE epgsql_test_db1 to epgsql_test_cleartext;
 GRANT ALL ON DATABASE epgsql_test_db2 to epgsql_test;
 

test/epgsql_SUITE.erl

@@ -46,6 +46,7 @@ groups() ->
             connect_as,
             connect_with_cleartext,
             connect_with_md5,
+            connect_with_scram,
             connect_with_invalid_user,
             connect_with_invalid_password,
             connect_with_ssl,
@@ -195,6 +196,19 @@ connect_with_md5(Config) ->
         [{database, "epgsql_test_db1"}]
     ]).
 
+connect_with_scram(Config) ->
+    PgConf = ?config(pg_config, Config),
+    Ver = ?config(version, PgConf),
+    (Ver >= [10, 0])
+        andalso
+        epgsql_ct:connect_only(
+          Config,
+          [
+           "epgsql_test_scram",
+           "epgsql_test_scram",
+           [{database, "epgsql_test_db1"}]
+          ]).
+
 connect_with_invalid_user(Config) ->
     {Host, Port} = epgsql_ct:connection_data(Config),
     Module = ?config(module, Config),
@@ -860,8 +874,8 @@ query_timeout(Config) ->
     Module = ?config(module, Config),
     epgsql_ct:with_connection(Config, fun(C) ->
         {ok, _, _} = Module:squery(C, "SET statement_timeout = 500"),
-        ?TIMEOUT_ERROR = Module:squery(C, "SELECT pg_sleep(1)"),
-        ?TIMEOUT_ERROR = Module:equery(C, "SELECT pg_sleep(2)"),
+        ?assertMatch(?TIMEOUT_ERROR, Module:squery(C, "SELECT pg_sleep(1)")),
+        ?assertMatch(?TIMEOUT_ERROR, Module:equery(C, "SELECT pg_sleep(2)")),
         {ok, _Cols, [{1}]} = Module:equery(C, "SELECT 1")
     end, []).
 
@@ -871,7 +885,7 @@ execute_timeout(Config) ->
         {ok, _, _} = Module:squery(C, "SET statement_timeout = 500"),
         {ok, S} = Module:parse(C, "select pg_sleep($1)"),
         ok = Module:bind(C, S, [2]),
-        ?TIMEOUT_ERROR = Module:execute(C, S, 0),
+        ?assertMatch(?TIMEOUT_ERROR, Module:execute(C, S, 0)),
         ok = Module:sync(C),
         ok = Module:bind(C, S, [0]),
         {ok, [{<<>>}]} = Module:execute(C, S, 0),
@@ -982,7 +996,7 @@ listen_notify(Config) ->
 
 listen_notify_payload(Config) ->
     Module = ?config(module, Config),
-    epgsql_ct:with_min_version(Config, 9.0, fun(C) ->
+    epgsql_ct:with_min_version(Config, [9, 0], fun(C) ->
         {ok, [], []}     = Module:squery(C, "listen epgsql_test"),
         {ok, _, [{Pid}]} = Module:equery(C, "select pg_backend_pid()"),
         {ok, [], []}     = Module:squery(C, "notify epgsql_test, 'test!'"),
@@ -995,7 +1009,7 @@ listen_notify_payload(Config) ->
 
 set_notice_receiver(Config) ->
     Module = ?config(module, Config),
-    epgsql_ct:with_min_version(Config, 9.0, fun(C) ->
+    epgsql_ct:with_min_version(Config, [9, 0], fun(C) ->
         {ok, [], []}     = Module:squery(C, "listen epgsql_test"),
         {ok, _, [{Pid}]} = Module:equery(C, "select pg_backend_pid()"),
 
@@ -1073,7 +1087,7 @@ get_cmd_status(Config) ->
     end).
 
 range_type(Config) ->
-    epgsql_ct:with_min_version(Config, 9.2, fun(_C) ->
+    epgsql_ct:with_min_version(Config, [9, 2], fun(_C) ->
         check_type(Config, int4range, "int4range(10, 20)", {10, 20}, [
             {1, 58}, {-1, 12}, {-985521, 5412687}, {minus_infinity, 0},
             {984655, plus_infinity}, {minus_infinity, plus_infinity}
@@ -1081,7 +1095,7 @@ range_type(Config) ->
    end, []).
 
 range8_type(Config) ->
-    epgsql_ct:with_min_version(Config, 9.2, fun(_C) ->
+    epgsql_ct:with_min_version(Config, [9, 2], fun(_C) ->
         check_type(Config, int8range, "int8range(10, 20)", {10, 20}, [
             {1, 58}, {-1, 12}, {-9223372036854775808, 5412687},
             {minus_infinity, 9223372036854775807},

test/epgsql_ct.erl

@@ -64,15 +64,16 @@ with_rollback(Config, F) ->
       end).
 
 with_min_version(Config, Min, F, Args) ->
-    Module = ?config(module, Config),
-    epgsql_ct:with_connection(Config, fun(C) ->
-        {ok, Bin} = Module:get_parameter(C, <<"server_version">>),
-        {ok, [{float, 1, Ver} | _], _} = erl_scan:string(binary_to_list(Bin)),
-        case Ver >= Min of
-            true  -> F(C);
-            false -> ?debugFmt("skipping test requiring PostgreSQL >= ~.2f~n", [Min])
-        end
-    end, Args).
+    PgConf = ?config(pg_config, Config),
+    Ver = ?config(version, PgConf),
+
+    case Ver >= Min of
+        true ->
+            epgsql_ct:with_connection(Config, F, Args);
+        false ->
+            ?debugFmt("skipping test requiring PostgreSQL >= ~p, but we have ~p ~p",
+                      [Min, Ver, Config])
+    end.
 
 %% flush mailbox
 flush() ->

test/epgsql_cth.erl

@@ -51,6 +51,7 @@ start_postgres() ->
     ok = application:start(erlexec),
     pipe([
         fun find_utils/1,
+        fun get_version/1,
         fun init_database/1,
         fun write_postgresql_config/1,
         fun copy_certs/1,
@@ -102,7 +103,8 @@ start_postgresql(Config) ->
             [{stderr,
               fun(_, _, Msg) ->
                   ct:pal(info, "postgres: ~s", [Msg])
-              end}]),
+              end},
+             {env, [{"LANGUAGE", "en"}]}]),
         loop(I)
     end),
     ConfigR = [
@@ -152,6 +154,17 @@ init_database(Config) ->
     {ok, _} = exec:run(Initdb ++ " --locale en_US.UTF8 " ++ PgDataDir, [sync,stdout,stderr]),
     [{datadir, PgDataDir}|Config].
 
+get_version(Config) ->
+    %% XXX: maybe use datadir/PG_VERSION after initdb?
+    Utils = ?config(utils, Config),
+    Postgres = ?config(postgres, Utils),
+
+    VersionStdout = list_to_binary(string:strip(os:cmd(Postgres ++ " -V"), both, $\n)),
+    VersionBin = lists:last(binary:split(VersionStdout, <<" ">>, [global])),
+    Version = lists:map(fun erlang:binary_to_integer/1,
+                        binary:split(VersionBin, <<".">>, [global])),
+    [{version, Version} | Config].
+
 write_postgresql_config(Config) ->
     PgDataDir = ?config(datadir, Config),
 
@@ -159,6 +172,7 @@ write_postgresql_config(Config) ->
         "ssl = on\n",
         "ssl_ca_file = 'root.crt'\n",
         "lc_messages = 'en_US.UTF-8'\n",
+        "fsync = off\n",
         "wal_level = 'logical'\n",
         "max_replication_slots = 15\n",
         "max_wal_senders = 15"
@@ -186,6 +200,7 @@ copy_certs(Config) ->
 
 write_pg_hba_config(Config) ->
     PgDataDir = ?config(datadir, Config),
+    Version = ?config(version, Config),
 
     User = os:getenv("USER"),
     PGConfig = [
@@ -197,8 +212,18 @@ write_pg_hba_config(Config) ->
         "host    epgsql_test_db1 epgsql_test             127.0.0.1/32    trust\n",
         "host    epgsql_test_db1 epgsql_test_md5         127.0.0.1/32    md5\n",
         "host    epgsql_test_db1 epgsql_test_cleartext   127.0.0.1/32    password\n",
-        "hostssl epgsql_test_db1 epgsql_test_cert        127.0.0.1/32    cert clientcert=1\n",
-        "host    replication     epgsql_test_replication 127.0.0.1/32    trust"
+        "hostssl epgsql_test_db1 epgsql_test_cert        127.0.0.1/32    cert clientcert=1\n" |
+        case Version >= [10, 0] of
+            true ->
+                %% See
+                %% https://www.postgresql.org/docs/10/static/release-10.html
+                %% "Change how logical replication uses pg_hba.conf"
+                ["host    epgsql_test_db1 epgsql_test_replication 127.0.0.1/32    trust\n",
+                 %% scram auth method only available on PG >= 10
+                 "host    epgsql_test_db1 epgsql_test_scram       127.0.0.1/32    scram-sha-256\n"];
+            false ->
+                ["host    replication     epgsql_test_replication 127.0.0.1/32    trust\n"]
+        end
     ],
     FilePath = filename:join(PgDataDir, "pg_hba.conf"),
     ok = file:write_file(FilePath, PGConfig),