kafka_protocol

%%%
%%%   Copyright (c) 2018-2021, Klarna Bank AB (publ)
%%%
%%%   Licensed under the Apache License, Version 2.0 (the "License");
%%%   you may not use this file except in compliance with the License.
%%%   You may obtain a copy of the License at
%%%
%%%       http://www.apache.org/licenses/LICENSE-2.0
%%%
%%%   Unless required by applicable law or agreed to in writing, software
%%%   distributed under the License is distributed on an "AS IS" BASIS,
%%%   WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
%%%   See the License for the specific language governing permissions and
%%%   limitations under the License.
%%%

%% @doc This module implements SASL (Simple Authentication and Security Layer)
%% for kafka authentication.

-module(kpro_sasl).

-export([ auth/7
        ]).

-include("kpro_private.hrl").

-define(undef, undefined).
-define(REASON(Reason), {sasl_auth_error, Reason}).
-define(ERROR(Reason), erlang:error(?REASON(Reason))).

%%%_* API ======================================================================

auth(_Host, _Sock, _Mod, _ClientId, _Timeout, ?undef, _Vsn) ->
  %% no auth
  ok;
auth(Host, Sock, Mod, ClientId, Timeout,
     {callback, ModuleName, Opts}, Vsn) ->
  case kpro_auth_backend:auth(ModuleName, Host, Sock, Vsn, Mod,
                              ClientId, Timeout, Opts) of
    ok ->
      ok;
    {ok, ServerResponse} ->
      {ok, ServerResponse};
    {error, Reason} ->
      ?ERROR(Reason)
  end;
auth(_Host, Sock, Mod, ClientId, Timeout, Opts, HandshakeVsn) ->
  Mechanism = mechanism(Opts),
  ok = handshake(Sock, Mod, Timeout, ClientId, Mechanism, HandshakeVsn),
  %% Send and receive raw bytes
  SendRecvRaw =
    fun(Req) ->
        try kpro_lib:send_and_recv_raw(Req, Sock, Mod, Timeout)
        catch
          error : Reason ?BIND_STACKTRACE(Stack) ->
            ?GET_STACKTRACE(Stack),
            ?ERROR({Reason, Stack})
        end
    end,
  %% Embed raw bytes in auth_bytes field.
  SendRecv =
    fun(Bytes) ->
        Req = kpro_req_lib:make(sasl_authenticate, _AuthReqVsn = 0,
                                [{auth_bytes, Bytes}]),
        Rsp = kpro_lib:send_and_recv(Req, Sock, Mod, ClientId, Timeout),
        EC = kpro:find(error_code, Rsp),
        case EC =:= ?no_error of
          true -> kpro:find(auth_bytes, Rsp);
          false -> ?ERROR(kpro:find(error_message, Rsp))
        end
    end,
  %% For version 0 handshake, the following auth request/responses
  %% are sent raw (without kafka protocol schema wrapper)
  %% For version 1 handshake, the following auth request/responses
  %% are wrapped by kafka protocol schema
  %% Version 1 has more informative error message comparing to version 0
  case HandshakeVsn =:= 0 of
    true -> do_auth(SendRecvRaw, Opts, HandshakeVsn);
    false -> do_auth(SendRecv, Opts, HandshakeVsn)
  end.

%%%_* Internal functions =======================================================

do_auth(SendRecv, {plain, User, Pass}, Vsn) ->
  Req = sasl_plain_token(User, Pass),
  try
    <<>> = SendRecv(Req),
    ok
  catch
    error : ?REASON({closed, _Stack}) when Vsn =:= 0 ->
      %% in version 0 (bare sasl bytes)
      %% bad credentials result in a remote socket close
      %% turn it into a more informative error code
      ?ERROR(bad_credentials)
  end;
do_auth(SendRecv, {Scram, User, Pass}, _) when ?IS_SCRAM(Scram) ->
  Sha = case Scram of
          ?scram_sha_256 -> sha256;
          ?scram_sha_512 -> sha512
        end,
  Scram1 = kpro_scram:init(Sha, User, Pass),
  %% make client first message
  CFirst = kpro_scram:first(Scram1),
  %% send client first message and receive server first message
  ServerFirstMsg = SendRecv(CFirst),
  %% parse server first message
  Scram2 = kpro_scram:parse(Scram1, ServerFirstMsg),
  %% make client final message
  ClientFinalMsg = kpro_scram:final(Scram2),
  %% send client final message and receive server final message
  ServerFinalMsg = SendRecv(ClientFinalMsg),
  %% Validate server signature
  ok = kpro_scram:validate(Scram2, ServerFinalMsg);
do_auth(_SendRecv, Unknown, _) ->
  ?ERROR({unknown_sasl_opt, Unknown}).

handshake(Sock, Mod, Timeout, ClientId, Mechanism, Vsn) ->
  Req = kpro_req_lib:make(sasl_handshake, Vsn, [{mechanism, Mechanism}]),
  Rsp = kpro_lib:send_and_recv(Req, Sock, Mod, ClientId, Timeout),
  ErrorCode = kpro:find(error_code, Rsp),
  case ErrorCode of
    ?no_error ->
      ok;
    unsupported_sasl_mechanism ->
      EnabledMechanisms = kpro:find(enabled_mechanisms, Rsp),
      Msg = io_lib:format("sasl mechanism ~s is not enabled in kafka, "
                          "enabled mechanism(s): ~s",
                          [Mechanism, cs(EnabledMechanisms)]),
      ?ERROR(iolist_to_binary(Msg));
    Other ->
      ?ERROR(Other)
  end.

sasl_plain_token(User, Pass) ->
  iolist_to_binary([0, User, 0, Pass]).

mechanism(?plain) -> <<"PLAIN">>;
mechanism(?scram_sha_256) -> <<"SCRAM-SHA-256">>;
mechanism(?scram_sha_512) -> <<"SCRAM-SHA-512">>;
mechanism({Tag, _User, _Pass}) -> mechanism(Tag).

cs([]) -> "[]";
cs([X]) -> X;
cs([H | T]) -> [H, "," | cs(T)].

%%%_* Emacs ====================================================================
%%% Local Variables:
%%% allout-layout: t
%%% erlang-indent-level: 2
%%% End: