Make MQTT authentication correspond with docs. Added tests

src/rabbit_mqtt_processor.erl

@@ -98,6 +98,12 @@ process_request(?CONNECT,
                     nocreds ->
                         rabbit_log:error("MQTT login failed - no credentials~n"),
                         {?CONNACK_CREDENTIALS, PState};
+                    {bad_creds, {undefined, Pass}} when is_list(Pass) ->
+                        rabbit_log:error("MQTT login failed - password without username is provided"),
+                        {?CONNACK_CREDENTIALS, PState};
+                    {bad_creds, {User, undefined}} when is_list(User) ->
+                        rabbit_log:error("MQTT login failed for ~p no password", [User]),
+                        {?CONNACK_CREDENTIALS, PState};
                     {UserBin, PassBin} ->
                         case process_login(UserBin, PassBin, ProtoVersion, PState) of
                             {?CONNACK_ACCEPT, Conn, VHost, AState} ->
@@ -501,36 +507,31 @@ creds(User, Pass, SSLLoginName) ->
     DefaultPass   = rabbit_mqtt_util:env(default_pass),
     {ok, Anon}    = application:get_env(?APP, allow_anonymous),
     {ok, TLSAuth} = application:get_env(?APP, ssl_cert_login),
-    U = case {User =/= undefined,
-              is_binary(DefaultUser),
-              Anon =:= true,
-              (TLSAuth andalso SSLLoginName =/= none)} of
-             %% username provided
-             {true,  _,    _,    _}     -> list_to_binary(User);
-             %% anonymous, default user is configured, no TLS
-             {false, true, true, false} -> DefaultUser;
-             %% no username provided, TLS certificate is present,
-             %% rabbitmq_mqtt.ssl_cert_login is true
-             {false, _,    _,    true}  -> SSLLoginName;
-             _                          -> nocreds
-        end,
-    case U of
-        nocreds ->
-            nocreds;
-        _ ->
-            case {Pass =/= undefined,
-                  is_binary(DefaultPass),
-                  Anon =:= true,
-                  TLSAuth} of
-                 %% password provided
-                 {true,  _,    _,    _} -> {U, list_to_binary(Pass)};
-                 %% password not provided, TLS certificate is present,
-                 %% rabbitmq_mqtt.ssl_cert_login is true
-                 {false, _, _, true}    -> {U, none};
-                 %% anonymous, default password is configured
-                 {false, true, true, _} -> {U, DefaultPass};
-                 _                      -> {U, none}
-            end
+    HaveDefaultCreds = Anon =:= true andalso
+                       is_binary(DefaultUser) andalso
+                       is_binary(DefaultPass),
+
+    CredentialsProvided = User =/= undefined orelse
+                          Pass =/= undefined,
+
+    CorrectCredentials = is_list(User) andalso
+                         is_list(Pass),
+
+    SSLLoginProvided = TLSAuth =:= true andalso
+                       SSLLoginName =/= none,
+
+    case {CredentialsProvided, CorrectCredentials, SSLLoginProvided, HaveDefaultCreds} of
+        %% Username and password takes priority
+        {true, true, _, _}          -> {list_to_binary(User),
+                                        list_to_binary(Pass)};
+        %% Either username or password is provided
+        {true, false, _, _}         -> {bad_creds, {User, Pass}};
+        %% rabbitmq_mqtt.ssl_cert_login is true. SSL user name provided.
+        %% Authorising with no password.
+        {false, false, true, _}     -> {SSLLoginName, none};
+        %% Anonymous
+        {false, false, false, true} -> {DefaultUser, DefaultPass};
+        _                           -> nocreds
     end.
 
 supported_subs_qos(?QOS_0) -> ?QOS_0;

test/auth_SUITE.erl

@@ -0,0 +1,196 @@
+-module(auth_SUITE).
+-compile([export_all]).
+
+-include_lib("common_test/include/ct.hrl").
+-include_lib("eunit/include/eunit.hrl").
+-define(CONNECT_TIMEOUT, 10000).
+
+all() ->
+    [{group, anonymous_no_ssl_user},
+     {group, anonymous_ssl_user},
+     {group, no_ssl_user},
+     {group, ssl_user}].
+
+groups() ->
+    [{anonymous_ssl_user, [],
+      [anonymous_auth_success,
+       user_credentials_auth,
+       ssl_user_auth_success]},
+     {anonymous_no_ssl_user, [],
+      [anonymous_auth_success,
+       user_credentials_auth]},
+     {ssl_user, [],
+      [anonymous_auth_fail,
+       user_credentials_auth,
+       ssl_user_auth_success]},
+     {no_ssl_user, [],
+      [anonymous_auth_fail,
+       user_credentials_auth,
+       ssl_user_auth_fail]}].
+
+init_per_suite(Config) ->
+    rabbit_ct_helpers:log_environment(),
+    Config.
+
+end_per_suite(Config) ->
+    Config.
+
+init_per_group(Group, Config) ->
+    Suffix = rabbit_ct_helpers:testcase_absname(Config, "", "-"),
+    Config1 = rabbit_ct_helpers:set_config(Config, [
+        {rmq_nodename_suffix, Suffix},
+        {rmq_certspwd, "bunnychow"}
+    ]),
+    MqttConfig = mqtt_config(Group),
+    rabbit_ct_helpers:run_setup_steps(Config1,
+        [ fun(Conf) -> merge_app_env(MqttConfig, Conf) end ] ++
+        rabbit_ct_broker_helpers:setup_steps() ++
+        rabbit_ct_client_helpers:setup_steps()).
+
+end_per_group(_, Config) ->
+    rabbit_ct_helpers:run_teardown_steps(Config,
+      rabbit_ct_client_helpers:teardown_steps() ++
+      rabbit_ct_broker_helpers:teardown_steps()).
+
+merge_app_env(MqttConfig, Config) ->
+    rabbit_ct_helpers:merge_app_env(Config, MqttConfig).
+
+mqtt_config(anonymous_ssl_user) ->
+    {rabbitmq_mqtt, [{ssl_cert_login,  true},
+                     {allow_anonymous, true}]};
+mqtt_config(anonymous_no_ssl_user) ->
+    {rabbitmq_mqtt, [{ssl_cert_login,  false},
+                     {allow_anonymous, true}]};
+mqtt_config(ssl_user) ->
+    {rabbitmq_mqtt, [{ssl_cert_login,  true},
+                     {allow_anonymous, false}]};
+mqtt_config(no_ssl_user) ->
+    {rabbitmq_mqtt, [{ssl_cert_login,  false},
+                     {allow_anonymous, false}]}.
+
+init_per_testcase(Testcase, Config) when Testcase == ssl_user_auth_success;
+                                         Testcase == ssl_user_auth_fail ->
+    Hostname = re:replace(os:cmd("hostname"), "\\s+", "", [global,{return,list}]),
+    User = "O=client,CN=" ++ Hostname,
+    {ok,_} = rabbit_ct_broker_helpers:rabbitmqctl(Config, 0, ["add_user", User, ""]),
+    {ok, _} = rabbit_ct_broker_helpers:rabbitmqctl(Config, 0, ["set_permissions",  "-p", "/", User, ".*", ".*", ".*"]),
+    Config1 = rabbit_ct_helpers:set_config(Config, [{temp_ssl_user, User}]),
+    rabbit_ct_helpers:testcase_started(Config1, Testcase);
+init_per_testcase(user_credentials_auth, Config) ->
+    User = <<"new-user">>,
+    Pass = <<"new-user-pass">>,
+    {ok,_} = rabbit_ct_broker_helpers:rabbitmqctl(Config, 0, ["add_user", User, Pass]),
+    {ok, _} = rabbit_ct_broker_helpers:rabbitmqctl(Config, 0, ["set_permissions",  "-p", "/", User, ".*", ".*", ".*"]),
+    Config1 = rabbit_ct_helpers:set_config(Config, [{new_user, User},
+                                                    {new_user_pass, Pass}]),
+    rabbit_ct_helpers:testcase_started(Config1, user_credentials_auth);
+init_per_testcase(Testcase, Config) ->
+    rabbit_ct_helpers:testcase_started(Config, Testcase).
+
+end_per_testcase(Testcase, Config) when Testcase == ssl_user_auth_success;
+                                        Testcase == ssl_user_auth_fail ->
+    User = ?config(temp_ssl_user, Config),
+    {ok,_} = rabbit_ct_broker_helpers:rabbitmqctl(Config, 0, ["delete_user", User]),
+    rabbit_ct_helpers:testcase_finished(Config, Testcase);
+end_per_testcase(user_credentials_auth, Config) ->
+    User = ?config(new_user, Config),
+    {ok,_} = rabbit_ct_broker_helpers:rabbitmqctl(Config, 0, ["delete_user", User]),
+    rabbit_ct_helpers:testcase_finished(Config, user_credentials_auth);
+end_per_testcase(Testcase, Config) ->
+    rabbit_ct_helpers:testcase_finished(Config, Testcase).
+
+anonymous_auth_success(Config) ->
+    expect_connect(fun connect_anonymous/1, Config).
+
+anonymous_auth_fail(Config) ->
+    expect_auth_error(fun connect_anonymous/1, Config).
+
+
+ssl_user_auth_success(Config) ->
+    expect_connect(fun connect_ssl/1, Config).
+
+ssl_user_auth_fail(Config) ->
+    expect_auth_error(fun connect_ssl/1, Config).
+
+user_credentials_auth(Config) ->
+    NewUser = ?config(new_user, Config),
+    NewUserPass = ?config(new_user_pass, Config),
+
+    expect_connect(
+        fun(Conf) -> connect_user(NewUser, NewUserPass, Conf) end,
+        Config),
+
+    expect_connect(
+        fun(Conf) -> connect_user(<<"guest">>, <<"guest">>, Conf) end,
+        Config),
+
+    expect_auth_error(
+        fun(Conf) -> connect_user(NewUser, <<"invalid_pass">>, Conf) end,
+        Config),
+
+    expect_auth_error(
+        fun(Conf) -> connect_user(undefined, <<"pass">>, Conf) end,
+        Config),
+
+    expect_auth_error(
+        fun(Conf) -> connect_user(NewUser, undefined, Conf) end,
+        Config).
+
+
+connect_anonymous(Config) ->
+    P = rabbit_ct_broker_helpers:get_node_config(Config, 0, tcp_port_mqtt),
+    emqttc:start_link([{host, "localhost"},
+                       {port, P},
+                       {client_id, <<"simpleClient">>},
+                       {proto_ver, 3},
+                       {logger, info}]).
+
+connect_ssl(Config) ->
+    CertsDir = ?config(rmq_certsdir, Config),
+    SSLConfig = [{cacertfile, filename:join([CertsDir, "testca", "cacert.pem"])},
+                 {certfile, filename:join([CertsDir, "client", "cert.pem"])},
+                 {keyfile, filename:join([CertsDir, "client", "key.pem"])}],
+    P = rabbit_ct_broker_helpers:get_node_config(Config, 0, tcp_port_mqtt_tls),
+    emqttc:start_link([{host, "localhost"},
+                       {port, P},
+                       {client_id, <<"simpleClient">>},
+                       {proto_ver, 3},
+                       {logger, info},
+                       {ssl, SSLConfig}]).
+
+connect_user(User, Pass, Config) ->
+    Creds = case User of
+        undefined -> [];
+        _         -> [{username, User}]
+    end ++ case Pass of
+        undefined -> [];
+        _         -> [{password, Pass}]
+    end,
+    ct:log("CREDS ~p", [Creds]),
+    P = rabbit_ct_broker_helpers:get_node_config(Config, 0, tcp_port_mqtt),
+    emqttc:start_link([{host, "localhost"},
+                       {port, P},
+                       {client_id, <<"simpleClient">>},
+                       {proto_ver, 3},
+                       {logger, info}] ++ Creds).
+
+expect_connect(ConnectFun, Config) ->
+    {ok, C} = ConnectFun(Config),
+    receive {mqttc, C, connected} -> emqttc:disconnect(C)
+    after ?CONNECT_TIMEOUT -> exit(emqttc_connection_timeout)
+    end.
+
+expect_auth_error(ConnectFun, Config) ->
+    process_flag(trap_exit, true),
+    {ok, C} = ConnectFun(Config),
+    Result = receive
+        {mqttc, C, connected} -> {error, unexpected_anonymous_connection};
+        {'EXIT', C, {shutdown,{connack_error,'CONNACK_CREDENTIALS'}}} -> ok
+    after
+        ?CONNECT_TIMEOUT -> {error, emqttc_connection_timeout}
+    end,
+    process_flag(trap_exit, false),
+    case Result of
+        ok -> ok;
+        {error, Err} -> exit(Err)
+    end.

test/java_SUITE_data/src/com/rabbitmq/mqtt/test/MqttTest.java

@@ -277,10 +277,12 @@ public void testInvalidPassword() throws MqttException {
     }
 
     public void testEmptyPassword() throws MqttException {
-        conOpt.setUserName("guest");
-        conOpt.setPassword(null);
+        MqttClient c = new MqttClient(brokerUrl, clientId, null);
+        MqttConnectOptions opts = new MyConnOpts();
+        opts.setUserName("guest");
+        opts.setPassword(null);
         try {
-            client.connect(conOpt);
+            c.connect(opts);
             fail("Authentication failure expected");
         } catch (MqttException ex) {
             Assert.assertEquals(MqttException.REASON_CODE_FAILED_AUTHENTICATION, ex.getReasonCode());
@@ -447,7 +449,7 @@ public void testNonCleanSession() throws MqttException, InterruptedException {
         client.disconnect();
     }
 
-    public void  testSessionRedelivery() throws MqttException, InterruptedException {
+    public void testSessionRedelivery() throws MqttException, InterruptedException {
         conOpt.setCleanSession(false);
         client.connect(conOpt);
         client.subscribe(topic, 1);

test/reader_SUITE.erl

@@ -23,16 +23,6 @@ suite() ->
 %% Testsuite setup/teardown.
 %% -------------------------------------------------------------------
 
-mqtt_config(Config) ->
-    P = rabbit_ct_broker_helpers:get_node_config(Config, 0, tcp_port_mqtt_extra),
-    P2 = rabbit_ct_broker_helpers:get_node_config(Config, 0, tcp_port_mqtt_tls_extra),
-    {rabbitmq_mqtt, [
-       {ssl_cert_login,   true},
-       {allow_anonymous,  true},
-       {tcp_listeners,    [P]},
-       {ssl_listeners,    [P2]}
-       ]}.
-
 init_per_suite(Config) ->
     rabbit_ct_helpers:log_environment(),
     Config1 = rabbit_ct_helpers:set_config(Config, [
@@ -68,7 +58,6 @@ end_per_testcase(Testcase, Config) ->
 
 block(Config) ->
     P = rabbit_ct_broker_helpers:get_node_config(Config, 0, tcp_port_mqtt),
-    % ok = rpc(Config, ?MODULE, change_configuration, [mqtt_config(Config)]),
     {ok, C} = emqttc:start_link([{host, "localhost"},
                                  {port, P},
                                  {client_id, <<"simpleClient">>},
@@ -112,6 +101,8 @@ block(Config) ->
 
     emqttc:disconnect(C).
 
+
+
 expect_publishes(_Topic, []) -> ok;
 expect_publishes(Topic, [Payload|Rest]) ->
     receive
@@ -122,15 +113,3 @@ expect_publishes(Topic, [Payload|Rest]) ->
 
 rpc(Config, M, F, A) ->
     rabbit_ct_broker_helpers:rpc(Config, 0, M, F, A).
-
-change_configuration({App, Args}) ->
-    ok = application:stop(App),
-    ok = change_cfg(App, Args),
-    application:start(App).
-
-change_cfg(_, []) ->
-    ok;
-change_cfg(App, [{Name,Value}|Rest]) ->
-    ok = application:set_env(App, Name, Value),
-    change_cfg(App, Rest).
-