feat(krb): added test cases for kerberos authentication

gssapi is the type kerberos is the backend.

.ci/docker-compose-file/docker-compose-kafka.yaml

@@ -16,24 +16,6 @@ services:
     user: "${DOCKER_USER:-root}"
     volumes:
       - /tmp/emqx-ci/emqx-shared-secret:/var/lib/secret
-  kdc:
-    hostname: kdc.emqx.net
-    image:  ghcr.io/emqx/emqx-builder/5.3-9:1.15.7-26.2.5-3-ubuntu22.04
-    container_name: kdc.emqx.net
-    expose:
-      - 88 # kdc
-      - 749 # admin server
-    # ports:
-    #   - 88:88
-    #   - 749:749
-    networks:
-      emqx_bridge:
-    volumes:
-      - /tmp/emqx-ci/emqx-shared-secret:/var/lib/secret
-      - ./kerberos/krb5.conf:/etc/kdc/krb5.conf
-      - ./kerberos/krb5.conf:/etc/krb5.conf
-      - ./kerberos/run.sh:/usr/bin/run.sh
-    command: run.sh
   kafka_1:
     image: wurstmeister/kafka:2.13-2.8.1
     # ports:
@@ -76,4 +58,3 @@ services:
       - ./kerberos/krb5.conf:/etc/kdc/krb5.conf
       - ./kerberos/krb5.conf:/etc/krb5.conf
     command: kafka-entrypoint.sh
-

.ci/docker-compose-file/docker-compose-kdc.yaml

@@ -0,0 +1,21 @@
+version: '3.9'
+
+services:
+  kdc:
+    hostname: kdc.emqx.net
+    image:  ghcr.io/emqx/emqx-builder/5.3-9:1.15.7-26.2.5-3-ubuntu22.04
+    container_name: kdc.emqx.net
+    expose:
+      - 88 # kdc
+      - 749 # admin server
+    # ports:
+    #   - 88:88
+    #   - 749:749
+    networks:
+      emqx_bridge:
+    volumes:
+      - /tmp/emqx-ci/emqx-shared-secret:/var/lib/secret
+      - ./kerberos/krb5.conf:/etc/kdc/krb5.conf
+      - ./kerberos/krb5.conf:/etc/krb5.conf
+      - ./kerberos/run.sh:/usr/bin/run.sh
+    command: run.sh

.ci/docker-compose-file/docker-compose.yaml

@@ -2,6 +2,7 @@ version: '3.9'
 
 services:
   erlang:
+    hostname: erlang.emqx.net
     container_name: erlang
     image: ${DOCKER_CT_RUNNER_IMAGE:-ghcr.io/emqx/emqx-builder/5.3-9:1.15.7-26.2.5-3-ubuntu22.04}
     env_file:

.ci/docker-compose-file/kerberos/krb5.conf

@@ -6,6 +6,7 @@
   rdns = false
   dns_lookup_kdc   = no
   dns_lookup_realm = no
+  default_keytab_name = /var/lib/secret/erlang.keytab
 
 [realms]
   KDC.EMQX.NET = {

.ci/docker-compose-file/kerberos/run.sh

@@ -6,20 +6,31 @@ echo "Remove old keytabs"
 rm -f /var/lib/secret/kafka.keytab > /dev/null 2>&1
 rm -f /var/lib/secret/rig.keytab > /dev/null 2>&1
 
+rm -f /var/lib/secret/erlang.keytab > /dev/null 2>&1
+rm -f /var/lib/secret/krb_authn_cli.keytab > /dev/null 2>&1
+
 echo "Create realm"
 
 kdb5_util -P emqx -r KDC.EMQX.NET create -s
 
 echo "Add principals"
 
-kadmin.local -w password -q "add_principal -randkey kafka/kafka-1.emqx.net@KDC.EMQX.NET"
+kadmin.local -w password -q "add_principal -randkey kafka/kafka-1.emqx.net@KDC.EMQX.NET" > /dev/null
 kadmin.local -w password -q "add_principal -randkey rig@KDC.EMQX.NET"  > /dev/null
 
+# For Kerberos Authn
+kadmin.local -w password -q "add_principal -randkey emqx/erlang.emqx.net@KDC.EMQX.NET" > /dev/null
+kadmin.local -w password -q "add_principal -randkey krb_authn_cli@KDC.EMQX.NET"  > /dev/null
+
 
 echo "Create keytabs"
 
 kadmin.local -w password -q "ktadd  -k /var/lib/secret/kafka.keytab -norandkey kafka/kafka-1.emqx.net@KDC.EMQX.NET " > /dev/null
 kadmin.local -w password -q "ktadd  -k /var/lib/secret/rig.keytab -norandkey rig@KDC.EMQX.NET " > /dev/null
 
+# For Kerberos Authn
+kadmin.local -w password -q "ktadd  -k /var/lib/secret/erlang.keytab -norandkey emqx/erlang.emqx.net@KDC.EMQX.NET " > /dev/null
+kadmin.local -w password -q "ktadd  -k /var/lib/secret/krb_authn_cli.keytab -norandkey krb_authn_cli@KDC.EMQX.NET " > /dev/null
+
 echo STARTING KDC
 /usr/sbin/krb5kdc -n

apps/emqx/rebar.config

@@ -47,7 +47,7 @@
             {meck, "0.9.2"},
             {proper, "1.4.0"},
             {bbmustache, "1.10.0"},
-            {emqtt, {git, "https://github.com/emqx/emqtt", {tag, "1.10.0"}}}
+            {emqtt, {git, "https://github.com/emqx/emqtt", {tag, "1.12.0"}}}
         ]},
         {extra_src_dirs, [
             {"test", [recursive]},
@@ -59,7 +59,7 @@
             {meck, "0.9.2"},
             {proper, "1.4.0"},
             {bbmustache, "1.10.0"},
-            {emqtt, {git, "https://github.com/emqx/emqtt", {tag, "1.9.7"}}}
+            {emqtt, {git, "https://github.com/emqx/emqtt", {tag, "1.12.0"}}}
         ]},
         {extra_src_dirs, [{"test", [recursive]}]}
     ]}

apps/emqx/src/emqx_channel.erl

@@ -394,6 +394,8 @@ handle_in(
                 case ConnState of
                     connecting ->
                         process_connect(NProperties, NChannel);
+                    reauthenticating ->
+                        process_connect(NProperties, NChannel);
                     _ ->
                         handle_out(
                             auth,

apps/emqx_auth_kerberos/BSL.txt

@@ -0,0 +1,94 @@
+Business Source License 1.1
+
+Licensor:             Hangzhou EMQ Technologies Co., Ltd.
+Licensed Work:        EMQX Enterprise Edition
+                      The Licensed Work is (c) 2023
+                      Hangzhou EMQ Technologies Co., Ltd.
+Additional Use Grant: Students and educators are granted right to copy,
+                      modify, and create derivative work for research
+                      or education.
+Change Date:          2028-01-26
+Change License:       Apache License, Version 2.0
+
+For information about alternative licensing arrangements for the Software,
+please contact Licensor: https://www.emqx.com/en/contact
+
+Notice
+
+The Business Source License (this document, or the “License”) is not an Open
+Source license. However, the Licensed Work will eventually be made available
+under an Open Source License, as stated in this License.
+
+License text copyright (c) 2017 MariaDB Corporation Ab, All Rights Reserved.
+“Business Source License” is a trademark of MariaDB Corporation Ab.
+
+-----------------------------------------------------------------------------
+
+Business Source License 1.1
+
+Terms
+
+The Licensor hereby grants you the right to copy, modify, create derivative
+works, redistribute, and make non-production use of the Licensed Work. The
+Licensor may make an Additional Use Grant, above, permitting limited
+production use.
+
+Effective on the Change Date, or the fourth anniversary of the first publicly
+available distribution of a specific version of the Licensed Work under this
+License, whichever comes first, the Licensor hereby grants you rights under
+the terms of the Change License, and the rights granted in the paragraph
+above terminate.
+
+If your use of the Licensed Work does not comply with the requirements
+currently in effect as described in this License, you must purchase a
+commercial license from the Licensor, its affiliated entities, or authorized
+resellers, or you must refrain from using the Licensed Work.
+
+All copies of the original and modified Licensed Work, and derivative works
+of the Licensed Work, are subject to this License. This License applies
+separately for each version of the Licensed Work and the Change Date may vary
+for each version of the Licensed Work released by Licensor.
+
+You must conspicuously display this License on each original or modified copy
+of the Licensed Work. If you receive the Licensed Work in original or
+modified form from a third party, the terms and conditions set forth in this
+License apply to your use of that work.
+
+Any use of the Licensed Work in violation of this License will automatically
+terminate your rights under this License for the current and all other
+versions of the Licensed Work.
+
+This License does not grant you any right in any trademark or logo of
+Licensor or its affiliates (provided that you may use a trademark or logo of
+Licensor as expressly required by this License).
+
+TO THE EXTENT PERMITTED BY APPLICABLE LAW, THE LICENSED WORK IS PROVIDED ON
+AN “AS IS” BASIS. LICENSOR HEREBY DISCLAIMS ALL WARRANTIES AND CONDITIONS,
+EXPRESS OR IMPLIED, INCLUDING (WITHOUT LIMITATION) WARRANTIES OF
+MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE, NON-INFRINGEMENT, AND
+TITLE.
+
+MariaDB hereby grants you permission to use this License’s text to license
+your works, and to refer to it using the trademark “Business Source License”,
+as long as you comply with the Covenants of Licensor below.
+
+Covenants of Licensor
+
+In consideration of the right to use this License’s text and the “Business
+Source License” name and trademark, Licensor covenants to MariaDB, and to all
+other recipients of the licensed work to be provided by Licensor:
+
+1. To specify as the Change License the GPL Version 2.0 or any later version,
+   or a license that is compatible with GPL Version 2.0 or a later version,
+   where “compatible” means that software provided under the Change License can
+   be included in a program with software provided under GPL Version 2.0 or a
+   later version. Licensor may specify additional Change Licenses without
+   limitation.
+
+2. To either: (a) specify an additional grant of rights to use that does not
+   impose any additional restriction on the right granted in this License, as
+   the Additional Use Grant; or (b) insert the text “None”.
+
+3. To specify a Change Date.
+
+4. Not to modify this License in any other way.

apps/emqx_auth_kerberos/docker-ct

@@ -0,0 +1 @@
+kdc

apps/emqx_auth_kerberos/include/emqx_auth_kerberos.hrl

@@ -0,0 +1,18 @@
+%%--------------------------------------------------------------------
+%% Copyright (c) 2024 EMQ Technologies Co., Ltd. All Rights Reserved.
+%%--------------------------------------------------------------------
+
+-ifndef(EMQX_AUTH_KERBEROS_HRL).
+-define(EMQX_AUTH_KERBEROS_HRL, true).
+
+-define(AUTHN_MECHANISM_GSSAPI, gssapi).
+-define(AUTHN_MECHANISM_GSSAPI_BIN, <<"gssapi">>).
+
+-define(AUTHN_BACKEND, kerberos).
+-define(AUTHN_BACKEND_BIN, <<"kerberos">>).
+
+-define(AUTHN_TYPE_KERBEROS, {?AUTHN_MECHANISM_GSSAPI, ?AUTHN_BACKEND}).
+
+-define(AUTHN_METHOD, <<"GSSAPI-KERBEROS">>).
+
+-endif.

apps/emqx_auth_kerberos/rebar.config

@@ -0,0 +1,7 @@
+%% -*- mode: erlang -*-
+
+{deps, [
+    {emqx, {path, "../emqx"}},
+    {emqx_utils, {path, "../emqx_utils"}},
+    {sasl_auth, {git, "https://github.com/kafka4beam/sasl_auth.git", {tag, "v2.1.1"}}}
+]}.

apps/emqx_auth_kerberos/src/emqx_auth_kerberos.app.src

@@ -0,0 +1,18 @@
+%% -*- mode: erlang -*-
+{application, emqx_auth_kerberos, [
+    {description, "EMQX Kerberos Authentication"},
+    {vsn, "0.1.0"},
+    {registered, []},
+    {mod, {emqx_auth_kerberos_app, []}},
+    {applications, [
+        kernel,
+        stdlib,
+        emqx_auth,
+        sasl_auth
+    ]},
+    {env, []},
+    {modules, []},
+
+    {licenses, ["Apache 2.0"]},
+    {links, []}
+]}.

apps/emqx_auth_kerberos/src/emqx_auth_kerberos_app.erl

@@ -0,0 +1,20 @@
+%%--------------------------------------------------------------------
+%% Copyright (c) 2024 EMQ Technologies Co., Ltd. All Rights Reserved.
+%%--------------------------------------------------------------------
+
+-module(emqx_auth_kerberos_app).
+
+-include("emqx_auth_kerberos.hrl").
+
+-behaviour(application).
+
+-export([start/2, stop/1]).
+
+start(_StartType, _StartArgs) ->
+    ok = emqx_authn:register_provider(?AUTHN_TYPE_KERBEROS, emqx_authn_kerberos),
+    {ok, Sup} = emqx_auth_kerberos_sup:start_link(),
+    {ok, Sup}.
+
+stop(_State) ->
+    ok = emqx_authn:deregister_provider(?AUTHN_TYPE_KERBEROS),
+    ok.

apps/emqx_auth_kerberos/src/emqx_auth_kerberos_sup.erl

@@ -0,0 +1,25 @@
+%%--------------------------------------------------------------------
+%% Copyright (c) 2024 EMQ Technologies Co., Ltd. All Rights Reserved.
+%%--------------------------------------------------------------------
+
+-module(emqx_auth_kerberos_sup).
+
+-behaviour(supervisor).
+
+-export([start_link/0]).
+
+-export([init/1]).
+
+-define(SERVER, ?MODULE).
+
+start_link() ->
+    supervisor:start_link({local, ?SERVER}, ?MODULE, []).
+
+init([]) ->
+    SupFlags = #{
+        strategy => one_for_all,
+        intensity => 0,
+        period => 1
+    },
+    ChildSpecs = [],
+    {ok, {SupFlags, ChildSpecs}}.

apps/emqx_auth_kerberos/src/emqx_authn_kerberos.erl

@@ -0,0 +1,121 @@
+%%--------------------------------------------------------------------
+%% Copyright (c) 2024 EMQ Technologies Co., Ltd. All Rights Reserved.
+%%--------------------------------------------------------------------
+
+-module(emqx_authn_kerberos).
+
+-include("emqx_auth_kerberos.hrl").
+-include_lib("emqx_auth/include/emqx_authn.hrl").
+-include_lib("typerefl/include/types.hrl").
+
+-behaviour(emqx_authn_provider).
+
+-export([
+    create/2,
+    update/2,
+    destroy/1,
+    authenticate/2
+]).
+
+create(
+    AuthenticatorID,
+    #{
+        principal := Principal,
+        keytab_file := KeyTabFile
+    }
+) ->
+    KeyTabPath = emqx_schema:naive_env_interpolation(KeyTabFile),
+    case sasl_auth:kinit(KeyTabPath, Principal) of
+        ok ->
+            {ok, #{
+                id => AuthenticatorID,
+                principal => Principal,
+                keytab_file => KeyTabFile
+            }};
+        Error ->
+            Error
+    end.
+
+update(Config, #{id := ID}) ->
+    create(ID, Config).
+
+destroy(_) ->
+    ok.
+
+authenticate(
+    #{
+        auth_method := ?AUTHN_METHOD,
+        auth_data := AuthData,
+        auth_cache := AuthCache
+    },
+    #{principal := Principal}
+) when AuthData =/= undefined ->
+    case AuthCache of
+        #{sasl_conn := SaslConn} ->
+            auth_continue(SaslConn, AuthData);
+        _ ->
+            case auth_new(Principal) of
+                {ok, SaslConn} ->
+                    auth_begin(SaslConn, AuthData);
+                Error ->
+                    Error
+            end
+    end;
+authenticate(_Credential, _State) ->
+    ignore.
+
+%%------------------------------------------------------------------------------
+%% Internal functions
+%%------------------------------------------------------------------------------
+
+%% @private Parse server principal to get server FQDN.
+%% The principal format is validated by config schema, so it can be assertive here.
+get_server_fqdn(Principal) ->
+    Pattern = "^([a-zA-Z0-9._-]+)/([a-zA-Z0-9.-]+)@",
+    {match, [_, FQDN]} = re:run(Principal, Pattern, [{capture, all_but_first, binary}]),
+    FQDN.
+
+auth_new(Principal) ->
+    ServerFQDN = get_server_fqdn(Principal),
+    case sasl_auth:server_new(<<"emqx">>, Principal, ServerFQDN) of
+        {ok, SaslConn} ->
+            {ok, SaslConn};
+        Error ->
+            ?TRACE_AUTHN_PROVIDER("sasl_kerberos_new_failed", #{
+                reason => Error,
+                sasl_function => "server_server_new"
+            }),
+            {error, not_authorized}
+    end.
+
+auth_begin(SaslConn, ClientToken) ->
+    case sasl_auth:server_start(SaslConn, ClientToken) of
+        {ok, {sasl_continue, ServerToken}} ->
+            {continue, ServerToken, #{sasl_conn => SaslConn}};
+        {ok, {sasl_ok, ServerToken}} ->
+            sasl_auth:server_done(SaslConn),
+            {ok, #{}, ServerToken};
+        Reason ->
+            ?TRACE_AUTHN_PROVIDER("sasl_kerberos_start_failed", #{
+                reason => Reason,
+                sasl_function => "server_server_start"
+            }),
+            sasl_auth:server_done(SaslConn),
+            {error, not_authorized}
+    end.
+
+auth_continue(SaslConn, ClientToken) ->
+    case sasl_auth:server_step(SaslConn, ClientToken) of
+        {ok, {sasl_continue, ServerToken}} ->
+            {continue, ServerToken, #{sasl_conn => SaslConn}};
+        {ok, {sasl_ok, ServerToken}} ->
+            sasl_auth:server_done(SaslConn),
+            {ok, #{}, ServerToken};
+        Reason ->
+            ?TRACE_AUTHN_PROVIDER("sasl_kerberos_step_failed", #{
+                reason => Reason,
+                sasl_function => "server_server_step"
+            }),
+            sasl_auth:server_done(SaslConn),
+            {error, not_authorized}
+    end.

apps/emqx_auth_kerberos/src/emqx_authn_kerberos_schema.erl

@@ -0,0 +1,65 @@
+%%--------------------------------------------------------------------
+%% Copyright (c) 2024 EMQ Technologies Co., Ltd. All Rights Reserved.
+%%--------------------------------------------------------------------
+
+-module(emqx_authn_kerberos_schema).
+
+-include("emqx_auth_kerberos.hrl").
+-include_lib("hocon/include/hoconsc.hrl").
+
+-behaviour(emqx_authn_schema).
+
+-export([
+    namespace/0,
+    fields/1,
+    desc/1,
+    refs/0,
+    select_union_member/1
+]).
+
+namespace() -> "authn".
+
+refs() ->
+    [?R_REF(kerberos)].
+
+select_union_member(#{
+    <<"mechanism">> := ?AUTHN_MECHANISM_GSSAPI_BIN, <<"backend">> := ?AUTHN_BACKEND_BIN
+}) ->
+    refs();
+select_union_member(#{<<"mechanism">> := ?AUTHN_MECHANISM_GSSAPI_BIN}) ->
+    throw(#{
+        reason => "unknown_backend",
+        expected => ?AUTHN_BACKEND
+    });
+select_union_member(_) ->
+    undefined.
+
+fields(kerberos) ->
+    emqx_authn_schema:common_fields() ++
+        [
+            {mechanism, emqx_authn_schema:mechanism(?AUTHN_MECHANISM_GSSAPI)},
+            {backend, emqx_authn_schema:backend(?AUTHN_BACKEND)},
+            {principal,
+                ?HOCON(binary(), #{
+                    required => true,
+                    desc => ?DESC(principal),
+                    validator => fun validate_principal/1
+                })},
+            {keytab_file,
+                ?HOCON(binary(), #{
+                    default => <<"/etc/krb5.keytab">>,
+                    desc => ?DESC(keytab_file)
+                })}
+        ].
+
+desc(kerberos) ->
+    "Settings for Kerberos authentication.";
+desc(_) ->
+    undefined.
+
+validate_principal(S) ->
+    P = <<"^([a-zA-Z0-9\\._-]+)/([a-zA-Z0-9\\.-]+)(?:@([A-Z0-9\\.-]+))?$">>,
+    case re:run(S, P) of
+        nomatch -> {error, invalid_server_principal_string};
+        {match, _} -> ok
+    end.

apps/emqx_auth_kerberos/test/emqx_authn_kerberos_SUITE.erl

@@ -0,0 +1,282 @@
+%%--------------------------------------------------------------------
+%% Copyright (c) 2024 EMQ Technologies Co., Ltd. All Rights Reserved.
+%%--------------------------------------------------------------------
+
+-module(emqx_authn_kerberos_SUITE).
+
+-compile(export_all).
+-compile(nowarn_export_all).
+
+-include_lib("eunit/include/eunit.hrl").
+-include_lib("common_test/include/ct.hrl").
+-include("emqx_auth_kerberos.hrl").
+
+-include_lib("emqx/include/emqx_mqtt.hrl").
+-include_lib("emqx_auth/include/emqx_authn.hrl").
+
+-define(PATH, [authentication]).
+
+-define(INVALID_SVR_PRINCIPAL, <<"not-exists/erlang.emqx.nett@KDC.EMQX.NET">>).
+
+-define(SVR_HOST, "erlang.emqx.net").
+-define(SVR_PRINCIPAL, <<"emqx/erlang.emqx.net@KDC.EMQX.NET">>).
+-define(SVR_KEYTAB_FILE, <<"/var/lib/secret/erlang.keytab">>).
+
+-define(CLI_NAME, "krb_authn_cli").
+-define(CLI_PRINCIPAL, <<"krb_authn_cli@KDC.EMQX.NET">>).
+-define(CLI_KEYTAB_FILE, <<"/var/lib/secret/krb_authn_cli.keytab">>).
+
+-define(HOST, "127.0.0.1").
+-define(PORT, 1883).
+
+-define(SERVER, "emqx").
+
+all() ->
+    emqx_common_test_helpers:all(?MODULE).
+
+init_per_suite(Config) ->
+    Apps = emqx_cth_suite:start([emqx, emqx_conf, emqx_auth, emqx_auth_kerberos], #{
+        work_dir => ?config(priv_dir, Config)
+    }),
+    IdleTimeout = emqx_config:get([mqtt, idle_timeout]),
+    [{apps, Apps}, {idle_timeout, IdleTimeout} | Config].
+
+end_per_suite(Config) ->
+    ok = emqx_config:put([mqtt, idle_timeout], ?config(idle_timeout, Config)),
+    ok = emqx_cth_suite:stop(?config(apps, Config)),
+    ok.
+
+init_per_testcase(_Case, Config) ->
+    emqx_authn_test_lib:delete_authenticators(
+        [authentication],
+        ?GLOBAL
+    ),
+    Config.
+
+end_per_testcase(_Case, Config) ->
+    Config.
+
+%%------------------------------------------------------------------------------
+%% Tests
+%%------------------------------------------------------------------------------
+
+t_create(_Config) ->
+    ValidConfig = raw_config(),
+
+    {ok, _} = emqx:update_config(
+        ?PATH,
+        {create_authenticator, ?GLOBAL, ValidConfig}
+    ),
+
+    {ok, [#{provider := emqx_authn_kerberos}]} =
+        emqx_authn_chains:list_authenticators(?GLOBAL).
+
+t_create_invalid(_Config) ->
+    InvalidConfig0 = raw_config(),
+    InvalidConfig = InvalidConfig0#{<<"principal">> := ?INVALID_SVR_PRINCIPAL},
+
+    {error, _} = emqx:update_config(
+        ?PATH,
+        {create_authenticator, ?GLOBAL, InvalidConfig}
+    ),
+
+    ?assertEqual(
+        {error, {not_found, {chain, ?GLOBAL}}},
+        emqx_authn_chains:list_authenticators(?GLOBAL)
+    ).
+
+t_authenticate(_Config) ->
+    init_auth(),
+
+    {ok, Handler, CT1} = setup_cli(),
+    {ok, C} = emqtt:start_link(
+        #{
+            host => ?HOST,
+            port => ?PORT,
+            proto_ver => v5,
+            properties =>
+                #{
+                    'Authentication-Method' => ?AUTHN_METHOD,
+                    'Authentication-Data' => CT1
+                },
+            custom_auth_callbacks =>
+                #{
+                    init => auth_init(Handler),
+                    handle_auth => fun auth_handle/3
+                }
+        }
+    ),
+    ?assertMatch({ok, _}, emqtt:connect(C)),
+    stop_cli(Handler),
+    ok.
+
+t_authenticate_bad_props(_Config) ->
+    erlang:process_flag(trap_exit, true),
+    init_auth(),
+
+    {ok, Handler, CT1} = setup_cli(),
+    {ok, C} = emqtt:start_link(
+        #{
+            host => ?HOST,
+            port => ?PORT,
+            proto_ver => v5,
+            properties =>
+                #{
+                    'Authentication-Method' => <<"SCRAM-SHA-512">>,
+                    'Authentication-Data' => CT1
+                },
+            custom_auth_callbacks =>
+                #{
+                    init => auth_init(Handler),
+                    handle_auth => fun auth_handle/3
+                }
+        }
+    ),
+
+    ?assertMatch({error, {not_authorized, _}}, emqtt:connect(C)),
+    stop_cli(Handler),
+    ok.
+
+t_authenticate_bad_token(_Config) ->
+    erlang:process_flag(trap_exit, true),
+    init_auth(),
+
+    {ok, Handler, CT1} = setup_cli(),
+    {ok, C} = emqtt:start_link(
+        #{
+            host => ?HOST,
+            port => ?PORT,
+            proto_ver => v5,
+            properties =>
+                #{
+                    'Authentication-Method' => <<"SCRAM-SHA-512">>,
+                    'Authentication-Data' => <<CT1/binary, "invalid">>
+                },
+            custom_auth_callbacks =>
+                #{
+                    init => auth_init(Handler),
+                    handle_auth => fun auth_handle/3
+                }
+        }
+    ),
+
+    ?assertMatch({error, {not_authorized, _}}, emqtt:connect(C)),
+    stop_cli(Handler),
+    ok.
+
+t_destroy(_) ->
+    State = init_auth(),
+
+    emqx_authn_test_lib:delete_authenticators(
+        [authentication],
+        ?GLOBAL
+    ),
+
+    {ok, Handler, CT1} = setup_cli(),
+
+    ?assertMatch(
+        ignore,
+        emqx_authn_mongodb:authenticate(
+            #{
+                auth_method => ?AUTHN_METHOD,
+                auth_data => CT1,
+                auth_cache => undefined
+            },
+            State
+        )
+    ),
+
+    stop_cli(Handler),
+    ok.
+
+%%------------------------------------------------------------------------------
+%% Helpers
+%%------------------------------------------------------------------------------
+
+raw_config() ->
+    #{
+        <<"mechanism">> => <<"gssapi">>,
+        <<"backend">> => <<"kerberos">>,
+        <<"principal">> => ?SVR_PRINCIPAL,
+        <<"keytab_file">> => ?SVR_KEYTAB_FILE
+    }.
+
+init_auth() ->
+    Config = raw_config(),
+
+    {ok, _} = emqx:update_config(
+        ?PATH,
+        {create_authenticator, ?GLOBAL, Config}
+    ),
+
+    {ok, [#{state := State}]} = emqx_authn_chains:list_authenticators(?GLOBAL),
+
+    State.
+
+%%------------------------------------------------------------------------------
+%% Custom auth
+
+auth_init(Handler) ->
+    fun() -> #{handler => Handler, step => 1} end.
+
+auth_handle(AuthState, Reason, Props) ->
+    ct:pal(">>> auth packet received:\n  rc: ~p\n  props:\n  ~p", [Reason, Props]),
+    do_auth_handle(AuthState, Reason, Props).
+
+do_auth_handle(
+    #{handler := Handler, step := Step} = AuthState0,
+    continue_authentication,
+    #{
+        'Authentication-Method' := ?AUTHN_METHOD,
+        'Authentication-Data' := ST
+    }
+) when Step =< 3 ->
+    {ok, CT} = call_cli_agent(Handler, {step, ST}),
+    AuthState = AuthState0#{step := Step + 1},
+    OutProps = #{
+        'Authentication-Method' => ?AUTHN_METHOD,
+        'Authentication-Data' => CT
+    },
+    {continue, {?RC_CONTINUE_AUTHENTICATION, OutProps}, AuthState};
+do_auth_handle(_AuthState, _Reason, _Props) ->
+    {stop, protocol_error}.
+
+%%------------------------------------------------------------------------------
+%% Client Agent
+
+setup_cli() ->
+    Pid = erlang:spawn(fun() -> cli_agent_loop(#{}) end),
+    {ok, CT1} = call_cli_agent(Pid, setup),
+    {ok, Pid, CT1}.
+
+call_cli_agent(Pid, Msg) ->
+    Ref = erlang:make_ref(),
+    erlang:send(Pid, {call, self(), Ref, Msg}),
+    receive
+        {Ref, Data} ->
+            {ok, Data}
+    after 3000 ->
+        error("client agent timeout")
+    end.
+
+stop_cli(Pid) ->
+    erlang:send(Pid, stop).
+
+cli_agent_loop(State) ->
+    receive
+        stop ->
+            ok;
+        {call, From, Ref, Msg} ->
+            {ok, Reply, State2} = cli_agent_handler(Msg, State),
+            erlang:send(From, {Ref, Reply}),
+            cli_agent_loop(State2)
+    end.
+
+cli_agent_handler(setup, State) ->
+    ok = sasl_auth:kinit(?CLI_KEYTAB_FILE, ?CLI_PRINCIPAL),
+    {ok, Client} = sasl_auth:client_new(?SERVER, ?SVR_HOST, ?CLI_PRINCIPAL, ?CLI_NAME),
+    {ok, {sasl_continue, CT1}} = sasl_auth:client_start(Client),
+    {ok, CT1, State#{client => Client}};
+cli_agent_handler({step, ST}, #{client := Client} = State) ->
+    {ok, {_, CT}} = sasl_auth:client_step(Client, ST),
+    {ok, CT, State}.

apps/emqx_bridge_kafka/docker-ct

@@ -1,2 +1,3 @@
 toxiproxy
+kdc
 kafka

apps/emqx_bridge_kafka/rebar.config

@@ -9,7 +9,8 @@
     {snappyer, "1.2.9"},
     {emqx_connector, {path, "../../apps/emqx_connector"}},
     {emqx_resource, {path, "../../apps/emqx_resource"}},
-    {emqx_bridge, {path, "../../apps/emqx_bridge"}}
+    {emqx_bridge, {path, "../../apps/emqx_bridge"}},
+    {sasl_auth, {git, "https://github.com/kafka4beam/sasl_auth.git", {tag, "v2.1.1"}}}
 ]}.
 
 {shell, [

apps/emqx_conf/src/emqx_conf_schema_inject.erl

@@ -55,7 +55,10 @@ authn_mods(ce) ->
     ];
 authn_mods(ee) ->
     authn_mods(ce) ++
-        [emqx_gcp_device_authn_schema].
+        [
+            emqx_gcp_device_authn_schema,
+            emqx_authn_kerberos_schema
+        ].
 
 authz() ->
     [{emqx_authz_schema, authz_mods()}].

apps/emqx_ds_backends/src/emqx_ds_backends.app.src.script

@@ -20,7 +20,7 @@ Backends = case Profile of
                [emqx_ds_builtin_local, emqx_ds_builtin_raft, emqx_fdb_ds]
            end,
 
-io:format(user, "DS backends available for this release (~p): ~p~n", [Profile, Backends]),
+io:format(user, "DS backends available for this release (~0p): ~0p~n", [Profile, Backends]),
 
 {application, emqx_ds_backends, [
     {description, "A placeholder application that depends on all available DS backends"},

apps/emqx_machine/priv/reboot_lists.eterm

@@ -136,6 +136,7 @@
             emqx_bridge_syskeeper,
             emqx_bridge_confluent,
             emqx_ds_shared_sub,
+            emqx_auth_gssapi,
             emqx_auth_ext,
             emqx_cluster_link,
             emqx_ds_builtin_raft

apps/emqx_retainer/rebar.config

@@ -30,7 +30,7 @@
 {profiles, [
     {test, [
         {deps, [
-            {emqtt, {git, "https://github.com/emqx/emqtt", {tag, "1.10.0"}}}
+            {emqtt, {git, "https://github.com/emqx/emqtt", {tag, "1.2.3.4"}}}
         ]}
     ]}
 ]}.

mix.exs

@@ -246,7 +246,7 @@ defmodule EMQXUmbrella.MixProject do
   def common_dep(:emqtt),
     do:
       {:emqtt,
-       github: "emqx/emqtt", tag: "1.10.1", override: true, system_env: maybe_no_quic_env()}
+       github: "emqx/emqtt", tag: "1.12.0", override: true, system_env: maybe_no_quic_env()}
 
   def common_dep(:typerefl),
     do: {:typerefl, github: "ieQu1/typerefl", tag: "0.9.1", override: true}
@@ -376,6 +376,7 @@ defmodule EMQXUmbrella.MixProject do
       :emqx_gateway_jt808,
       :emqx_bridge_syskeeper,
       :emqx_ds_shared_sub,
+      :emqx_auth_gssapi,
       :emqx_auth_ext,
       :emqx_cluster_link,
       :emqx_ds_builtin_raft

rebar.config

@@ -91,7 +91,7 @@
     {ecpool, {git, "https://github.com/emqx/ecpool", {tag, "0.5.7"}}},
     {replayq, {git, "https://github.com/emqx/replayq.git", {tag, "0.3.8"}}},
     {pbkdf2, {git, "https://github.com/emqx/erlang-pbkdf2.git", {tag, "2.0.4"}}},
-    {emqtt, {git, "https://github.com/emqx/emqtt", {tag, "1.10.1"}}},
+    {emqtt, {git, "https://github.com/emqx/emqtt", {tag, "1.12.0"}}},
     {rulesql, {git, "https://github.com/emqx/rulesql", {tag, "0.2.1"}}},
     % NOTE: depends on recon 2.5.x
     {observer_cli, "1.7.1"},

rel/i18n/emqx_authn_kerberos_schema.hocon

@@ -0,0 +1,18 @@
+emqx_authn_kerberos_schema {
+
+principal {
+    label: "Kerberos Principal"
+    desc: """~
+        Server Kerberos principal.
+        For example <code>mqtt/emqx-cluster-1.example.com@MY_REALM.EXAMPLE.COM</code>.
+        NOTE: The realm in use has to be configured in /etc/krb5.conf in EMQX nodes.~"""
+}
+
+keytab_file {
+    label: "Keytab File"
+    desc: """~
+        Kerberos keytab file path.
+        NOTE: This file has to be placed in EMQX nodes.~"""
+}
+
+}

rel/i18n/emqx_bridge_kafka.hocon

@@ -147,7 +147,7 @@ consumer_mqtt_opts.label:
 """MQTT publish"""
 
 auth_kerberos_principal.desc:
-"""SASL GSSAPI authentication Kerberos principal. For example <code>client_name@MY.KERBEROS.REALM.MYDOMAIN.COM</code>, NOTE: The realm in use has to be configured in /etc/krb5.conf in EMQX nodes."""
+"""SASL GSSAPI authentication Kerberos principal. For example <code>kafka/node1.example.com@EXAMPLE.COM</code>, NOTE: The realm in use has to be configured in /etc/krb5.conf in EMQX nodes."""
 
 auth_kerberos_principal.label:
 """Kerberos Principal"""

scripts/ct/run.sh

@@ -256,6 +256,9 @@ for dep in ${CT_DEPS}; do
 	couchbase)
 	    FILES+=( '.ci/docker-compose-file/docker-compose-couchbase.yaml' )
 	    ;;
+	kdc)
+	    FILES+=( '.ci/docker-compose-file/docker-compose-kdc.yaml' )
+	    ;;
         *)
             echo "unknown_ct_dependency $dep"
             exit 1