fix(electric): Implement support for AuthenticationMD5Password upstre…

…am auth method (#859)

Fixes VAX-1553.

.changeset/blue-snakes-jog.md

@@ -0,0 +1,5 @@
+---
+"@core/electric": patch
+---
+
+[VAX-1553] Add support for AuthenticationMD5Password upstream auth method in the Migrations proxy. This fixes a connectivity issue between Electric and DigitalOcean Managed PostgreSQL.

components/electric/lib/electric/errors.ex

@@ -105,6 +105,14 @@ defmodule Electric.Errors do
     """
   end
 
+  defp format_header(:not_implemented) do
+    """
+    ┌─────────────────────────┐
+    │  NOT IMPLEMENTED ERROR  │
+    ┕━━━━━━━━━━━━━━━━━━━━━━━━━┙
+    """
+  end
+
   defp format_header(module_alias) do
     module_line = "  MODULE ERROR: " <> inspect(module_alias) <> "  "
     n = byte_size(module_line)

components/electric/lib/electric/postgres/proxy/upstream_connection.ex

@@ -157,6 +157,17 @@ defmodule Electric.Postgres.Proxy.UpstreamConnection do
     upstream(response, state)
   end
 
+  defp handle_backend_msg(
+         %M.AuthenticationMD5Password{salt: salt},
+         %{authenticated: false, conn_opts: conn_opts} = state
+       ) do
+    userspec_digest = md5_hex_digest([conn_opts[:password], conn_opts[:username]])
+    salted_digest = md5_hex_digest([userspec_digest, salt])
+    response = %M.PasswordMessage{password: "md5" <> salted_digest}
+
+    upstream(response, state)
+  end
+
   defp handle_backend_msg(%M.AuthenticationSASL{} = msg, %{authenticated: false} = state) do
     {sasl_mechanism, response} = SASL.initial_response(msg)
 
@@ -176,6 +187,15 @@ defmodule Electric.Postgres.Proxy.UpstreamConnection do
     %{state | sasl: nil}
   end
 
+  defp handle_backend_msg(%msg_type{} = msg, _state)
+       when msg_type in [M.AuthenticationKerberosV5, M.AuthenticationGSS, M.AuthenticationSSPI] do
+    error_msg = "Proxy's upstream connection requested unsupported authentication method:"
+    error_val = inspect(msg, pretty: true)
+
+    Electric.Errors.print_error(:not_implemented, error_msg <> "\n\n    " <> error_val)
+    exit(error_msg <> " " <> error_val)
+  end
+
   defp handle_backend_msg(msg, state) do
     %{state | pending: [msg | state.pending]}
   end
@@ -206,4 +226,8 @@ defmodule Electric.Postgres.Proxy.UpstreamConnection do
   defp reset_conn(state) do
     %{state | conn: nil, transport_module: :gen_tcp}
   end
+
+  defp md5_hex_digest(iodata) do
+    :crypto.hash(:md5, iodata) |> Base.encode16(case: :lower)
+  end
 end

components/electric/test/electric/postgres/proxy_test.exs

@@ -100,8 +100,6 @@ defmodule Electric.Postgres.ProxyTest do
       # Verify that both upstream connections have shut down without errors.
       assert_receive {:DOWN, ^mon1, :process, ^pid1, :normal}
       assert_receive {:DOWN, ^mon2, :process, ^pid2, :normal}
-
-      Process.sleep(1000)
     end)
   end
 end