feat: implement standalone sasl auth function

README.md

@@ -154,7 +154,10 @@ After, you can use the `deserialize_response/3` function of the messages API, pa
 
 ## SASL
 
-SASL is handled by the `KlifeProtocol.Socket` module and client libraries can pass SASL options to `connect/3` function.
+SASL is handled by the `KlifeProtocol.Socket` module and client libraries can pass use it in 2 ways:
+
+- Passing SASL options to `connect/3` function when creationg a new socket
+- Passing an already open socket and SASL opts to `authenticate/3` function
 
 For now the only supported mechanism is PLAIN and you can use it like this:
 
@@ -171,7 +174,12 @@ sasl_opts = [
   ]
 ]
 
+# On socket initialization
 {:ok, socket} = Socket.connect("localhost", 9092, [backend: :ssl, sasl_opts: sasl_opts])
+
+# After socket initialization
+{:ok, socket} = Socket.connect("localhost", 9092, [backend: :ssl])
+:ok = Socket.authenticate(socket, :ssl, sasl_opts)
 ```
 
 ## Compression and Record Batch Attributes

lib/klife_protocol/sasl_mechanism/behaviour.ex → lib/klife_protocol/sasl/mechanism/behaviour.ex

@@ -1,3 +1,3 @@
-defmodule KlifeProtocol.SASLMechanism.Behaviour do
+defmodule KlifeProtocol.SASL.Mechanism.Behaviour do
   @callback execute_auth(sendrcv_fun :: function, opts :: list) :: :ok | {:error, reason :: term}
 end

lib/klife_protocol/sasl_mechanism/plain.ex → lib/klife_protocol/sasl/mechanism/plain.ex

@@ -1,5 +1,5 @@
-defmodule KlifeProtocol.SASLMechanism.Plain do
-  @behaviour KlifeProtocol.SASLMechanism.Behaviour
+defmodule KlifeProtocol.SASL.Mechanism.Plain do
+  @behaviour KlifeProtocol.SASL.Mechanism.Behaviour
 
   def execute_auth(send_recv_fun, opts) do
     usr = Keyword.fetch!(opts, :username)

lib/klife_protocol/sasl/sasl.ex

@@ -0,0 +1,61 @@
+defmodule KlifeProtocol.SASL do
+  alias KlifeProtocol.Messages.SaslAuthenticate
+  alias KlifeProtocol.Messages.SaslHandshake
+
+  @supported_mechanisms %{
+    "PLAIN" => KlifeProtocol.SASL.Mechanism.Plain
+  }
+
+  def authenticate(mech, handshake_vsn, auth_vsn, mech_opts, send_recv_raw_fun) do
+    send_recv_fun = fn data ->
+      to_send = %{content: %{auth_bytes: data}, headers: %{correlation_id: 123}}
+      to_send_raw = SaslAuthenticate.serialize_request(to_send, auth_vsn)
+      rcv_bin = send_recv_raw_fun.(to_send_raw)
+
+      {:ok, %{content: resp}} =
+        SaslAuthenticate.deserialize_response(rcv_bin, auth_vsn)
+
+      case resp do
+        %{error_code: 0, auth_bytes: auth_bytes} ->
+          auth_bytes
+
+        %{error_code: ec} ->
+          raise """
+          Unexpected error on SASL authentication message. ErrorCode: #{inspect(ec)}
+          """
+      end
+    end
+
+    :ok = handshake(mech, handshake_vsn, send_recv_raw_fun)
+
+    case Map.get(@supported_mechanisms, mech) do
+      nil ->
+        raise "Unsupported SASL mechanism #{inspect(mech)}. Supported ones are: #{inspect(Map.keys(@supported_mechanisms))}"
+
+      mech_mod ->
+        :ok = apply(mech_mod, :execute_auth, [send_recv_fun, mech_opts])
+    end
+  end
+
+  defp handshake(mech, hanshake_vsn, send_rcv_raw_fun) do
+    to_send = %{content: %{mechanism: mech}, headers: %{correlation_id: 123}}
+    to_send_raw = SaslHandshake.serialize_request(to_send, hanshake_vsn)
+    recv_bin = send_rcv_raw_fun.(to_send_raw)
+
+    {:ok, %{content: resp}} = SaslHandshake.deserialize_response(recv_bin, hanshake_vsn)
+
+    case resp do
+      %{error_code: 0, mechanisms: server_enabled_mechanisms} ->
+        if mech in server_enabled_mechanisms do
+          :ok
+        else
+          raise "Server does not support SASL mechanism #{mech}. Supported mechanisms are: #{inspect(server_enabled_mechanisms)}"
+        end
+
+      %{error_code: ec} ->
+        raise """
+        Unexpected error on SASL handhsake message. ErrorCode: #{inspect(ec)}
+        """
+    end
+  end
+end

lib/klife_protocol/socket.ex

@@ -13,8 +13,7 @@ defmodule KlifeProtocol.Socket do
   - any other option will be forwarded to the backend module `connect/3` function
   """
 
-  alias KlifeProtocol.Messages.SaslAuthenticate
-  alias KlifeProtocol.Messages.SaslHandshake
+  alias KlifeProtocol.SASL
 
   def connect(host, port, opts \\ []) do
     {backend, opts} = Keyword.pop(opts, :backend, :gen_tcp)
@@ -23,20 +22,20 @@ defmodule KlifeProtocol.Socket do
 
     case backend.connect(String.to_charlist(host), port, opts ++ must_have_opts) do
       {:ok, socket} ->
-        :ok = maybe_handle_sasl(socket, backend, sasl_opts)
+        :ok = authenticate(socket, backend, sasl_opts)
         {:ok, socket}
 
       err ->
         err
     end
   end
 
-  defp maybe_handle_sasl(_socket, _backend, []), do: :ok
+  def authenticate(_socket, _backend, []), do: :ok
 
-  defp maybe_handle_sasl(socket, backend, sasl_opts) do
-    mechanism = Keyword.fetch!(sasl_opts, :mechanism)
-    sasl_auth_vsn = Keyword.fetch!(sasl_opts, :sasl_auth_vsn)
-    sasl_handshake_vsn = Keyword.fetch!(sasl_opts, :sasl_handshake_vsn)
+  def authenticate(socket, backend, sasl_opts) do
+    mech = Keyword.fetch!(sasl_opts, :mechanism)
+    handshake_vsn = Keyword.fetch!(sasl_opts, :handshake_vsn)
+    auth_vsn = Keyword.fetch!(sasl_opts, :auth_vsn)
     mechanism_opts = Keyword.get(sasl_opts, :mechanism_opts, [])
 
     send_recv_raw_fun = fn data ->
@@ -45,56 +44,6 @@ defmodule KlifeProtocol.Socket do
       rcv_bin
     end
 
-    :ok = sasl_handshake(send_recv_raw_fun, mechanism, sasl_handshake_vsn)
-
-    send_recv_fun = fn data ->
-      to_send = %{content: %{auth_bytes: data}, headers: %{correlation_id: 123}}
-      to_send_raw = SaslAuthenticate.serialize_request(to_send, sasl_auth_vsn)
-      rcv_bin = send_recv_raw_fun.(to_send_raw)
-
-      {:ok, %{content: resp}} =
-        SaslAuthenticate.deserialize_response(rcv_bin, sasl_auth_vsn)
-
-      case resp do
-        %{error_code: 0, auth_bytes: auth_bytes} ->
-          auth_bytes
-
-        %{error_code: ec} ->
-          raise """
-          Unexpected error on SASL authentication message. ErrorCode: #{inspect(ec)}
-          """
-      end
-    end
-
-    case mechanism do
-      "PLAIN" ->
-        KlifeProtocol.SASLMechanism.Plain.execute_auth(send_recv_fun, mechanism_opts)
-
-      other ->
-        raise "Unsupported SASL mechanism #{inspect(other)}"
-    end
-  end
-
-  defp sasl_handshake(send_rcv_raw_fun, mechanism, sasl_handshake_vsn) do
-    to_send = %{content: %{mechanism: mechanism}, headers: %{correlation_id: 123}}
-    to_send_raw = SaslHandshake.serialize_request(to_send, sasl_handshake_vsn)
-    recv_bin = send_rcv_raw_fun.(to_send_raw)
-
-    {:ok, %{content: resp}} =
-      SaslHandshake.deserialize_response(recv_bin, sasl_handshake_vsn)
-
-    case resp do
-      %{error_code: 0, mechanisms: server_enabled_mechanisms} ->
-        if mechanism in server_enabled_mechanisms do
-          :ok
-        else
-          raise "Server does not support SASL mechanism #{mechanism}. Supported mechanisms are: #{inspect(server_enabled_mechanisms)}"
-        end
-
-      %{error_code: ec} ->
-        raise """
-        Unexpected error on SASL handhsake message. ErrorCode: #{inspect(ec)}
-        """
-    end
+    :ok = SASL.authenticate(mech, handshake_vsn, auth_vsn, mechanism_opts, send_recv_raw_fun)
   end
 end

test/sasl_test.exs

@@ -94,8 +94,8 @@ defmodule KlifeProtocol.SaslTest do
 
     sasl_opts = [
       mechanism: "PLAIN",
-      sasl_auth_vsn: 2,
-      sasl_handshake_vsn: 1,
+      auth_vsn: 2,
+      handshake_vsn: 1,
       mechanism_opts: [
         username: "klifeusr",
         password: "klifepwd"
@@ -119,4 +119,39 @@ defmodule KlifeProtocol.SaslTest do
       {:ok, _rcv_data} = Produce.deserialize_response(rcv_bin, version)
     end)
   end
+
+  test "success with plain sasl opts 2 step auth" do
+    ssl_opts = [
+      verify: :verify_peer,
+      cacertfile: Path.relative("test/compose_files/ssl/ca.crt")
+    ]
+
+    sasl_opts = [
+      mechanism: "PLAIN",
+      auth_vsn: 2,
+      handshake_vsn: 1,
+      mechanism_opts: [
+        username: "klifeusr",
+        password: "klifepwd"
+      ]
+    ]
+
+    socket_backend = :ssl
+    opts = [backend: socket_backend, active: false] ++ ssl_opts
+
+    sockets =
+      Enum.map(@brokers_sasl, fn {_broker, hostname} ->
+        [host, port] = String.split(hostname, ":")
+        {:ok, socket} = Socket.connect(host, String.to_integer(port), opts)
+        :ok = Socket.authenticate(socket, socket_backend, sasl_opts)
+        socket
+      end)
+
+    Enum.each(sockets, fn socket ->
+      {data, version} = get_produce_msg_binary()
+      :ok = apply(socket_backend, :send, [socket, data])
+      {:ok, rcv_bin} = apply(socket_backend, :recv, [socket, 0, 5000])
+      {:ok, _rcv_data} = Produce.deserialize_response(rcv_bin, version)
+    end)
+  end
 end