Support 'is_superuser/2 callback

src/emqttd_acl_mod.erl

@@ -24,16 +24,16 @@
 
 -ifdef(use_specs).
 
--callback init(AclOpts :: list()) -> {ok, State :: any()}.
+-callback(init(AclOpts :: list()) -> {ok, State :: any()}).
 
--callback check_acl({Client, PubSub, Topic}, State :: any()) -> allow | deny | ignore when
-    Client   :: mqtt_client(),
-    PubSub   :: pubsub(),
-    Topic    :: binary().
+-callback(check_acl({Client, PubSub, Topic}, State :: any()) -> allow | deny | ignore when
+        Client   :: mqtt_client(),
+        PubSub   :: pubsub(),
+        Topic    :: binary()).
 
--callback reload_acl(State :: any()) -> ok | {error, any()}.
+-callback(reload_acl(State :: any()) -> ok | {error, any()}).
 
--callback description() -> string().
+-callback(description() -> string()).
 
 -else.
 

src/emqttd_auth_anonymous.erl

@@ -19,11 +19,13 @@
 
 -behaviour(emqttd_auth_mod).
 
--export([init/1, check/3, description/0]).
+-export([init/1, check/3, is_superuser/2, description/0]).
 
 init(Opts) -> {ok, Opts}.
 
 check(_Client, _Password, _Opts) -> ok.
 
+is_superuser(_Client, _Opts) -> false.
+
 description() -> "Anonymous Authentication Module".
 

src/emqttd_auth_clientid.erl

@@ -24,7 +24,7 @@
 -behaviour(emqttd_auth_mod).
 
 %% emqttd_auth_mod callbacks
--export([init/1, check/3, description/0]).
+-export([init/1, check/3, is_superuser/2, description/0]).
 
 -define(AUTH_CLIENTID_TAB, mqtt_auth_clientid).
 
@@ -88,6 +88,8 @@ check(#mqtt_client{client_id = ClientId}, Password, [{password, yes}|_]) ->
         _ -> {error, password_error}
     end.
 
+is_superuser(_Client, _Opts) -> false.
+
 description() -> "ClientId authentication module".
 
 %%--------------------------------------------------------------------

src/emqttd_auth_mod.erl

@@ -14,14 +14,13 @@
 %% limitations under the License.
 %%--------------------------------------------------------------------
 
-%% @doc Authentication Behaviour.
 -module(emqttd_auth_mod).
 
 -include("emqttd.hrl").
 
 -export([passwd_hash/2]).
 
--type hash_type() :: plain | md5 | sha | sha256.
+-type(hash_type() :: plain | md5 | sha | sha256).
 
 %%--------------------------------------------------------------------
 %% Authentication behavihour
@@ -29,21 +28,23 @@
 
 -ifdef(use_specs).
 
--callback init(AuthOpts :: list()) -> {ok, State :: any()}.
+-callback(init(AuthOpts :: list()) -> {ok, State :: any()}).
 
--callback check(Client, Password, State) -> ok | ignore | {error, string()} when
-    Client    :: mqtt_client(),
-    Password  :: binary(),
-    State     :: any().
+-callback(check(Client, Password, State) -> ok | ignore | {error, string()} when
+        Client   :: mqtt_client(),
+        Password :: binary(),
+        State    :: any()).
 
--callback description() -> string().
+-callback(is_superuser(Client :: mqtt_client(), State :: any()) -> boolean()).
+
+-callback(description() -> string()).
 
 -else.
 
 -export([behaviour_info/1]).
 
 behaviour_info(callbacks) ->
-    [{init, 1}, {check, 3}, {description, 0}];
+    [{init, 1}, {check, 3}, {is_superuser, 2}, {description, 0}];
 behaviour_info(_Other) ->
     undefined.
 

src/emqttd_auth_username.erl

@@ -31,7 +31,7 @@
 -export([add_user/2, remove_user/1, lookup_user/1, all_users/0]).
 
 %% emqttd_auth callbacks
--export([init/1, check/3, description/0]).
+-export([init/1, check/3, is_superuser/2, description/0]).
 
 -define(AUTH_USERNAME_TAB, mqtt_auth_username).
 
@@ -146,6 +146,8 @@ check(#mqtt_client{username = Username}, Password, _Opts) ->
             end
     end.
 
+is_superuser(_Client, _Opts) -> false.
+
 description() ->
     "Username password authentication module".
 

src/emqttd_protocol.erl

@@ -34,7 +34,7 @@
 %% Protocol State
 -record(proto_state, {peername, sendfun, connected = false,
                       client_id, client_pid, clean_sess,
-                      proto_ver, proto_name, username,
+                      proto_ver, proto_name, username, is_superuser = false,
                       will_msg, keepalive, max_clientid_len = ?MAX_CLIENTID_LEN,
                       session, ws_initial_headers, %% Headers from first HTTP request for websocket client
                       connected_at}).
@@ -159,8 +159,12 @@ process(Packet = ?CONNECT_PACKET(Var), State0) ->
     {ReturnCode1, SessPresent, State3} =
     case validate_connect(Var, State1) of
         ?CONNACK_ACCEPT ->
-            case emqttd_access_control:auth(client(State1), Password) of
+            Client = client(State1),
+            case emqttd_access_control:auth(Client, Password) of
                 ok ->
+                    %% Is Superuser?
+                    IsSuperuser = emqttd_access_control:is_superuser(Client),
+
                     %% Generate clientId if null
                     State2 = maybe_set_clientid(State1),
 
@@ -172,7 +176,7 @@ process(Packet = ?CONNECT_PACKET(Var), State0) ->
                             %% Start keepalive
                             start_keepalive(KeepAlive),
                             %% ACCEPT
-                            {?CONNACK_ACCEPT, SP, State2#proto_state{session = Session}};
+                            {?CONNACK_ACCEPT, SP, State2#proto_state{session = Session, is_superuser = IsSuperuser}};
                         {error, Error} ->
                             exit({shutdown, Error})
                     end;
@@ -188,12 +192,10 @@ process(Packet = ?CONNECT_PACKET(Var), State0) ->
     %% Send connack
     send(?CONNACK_PACKET(ReturnCode1, sp(SessPresent)), State3);
 
-process(Packet = ?PUBLISH_PACKET(_Qos, Topic, _PacketId, _Payload), State) ->
-    case check_acl(publish, Topic, client(State)) of
-        allow ->
-            publish(Packet, State);
-        deny ->
-            ?LOG(error, "Cannot publish to ~s for ACL Deny", [Topic], State)
+process(Packet = ?PUBLISH_PACKET(_Qos, Topic, _PacketId, _Payload), State = #proto_state{is_superuser = IsSuper}) ->
+    case IsSuper orelse allow == check_acl(publish, Topic, client(State)) of
+        true  -> publish(Packet, State);
+        false -> ?LOG(error, "Cannot publish to ~s for ACL Deny", [Topic], State)
     end,
     {ok, State};
 
@@ -216,11 +218,14 @@ process(?PUBACK_PACKET(?PUBCOMP, PacketId), State = #proto_state{session = Sessi
 process(?SUBSCRIBE_PACKET(PacketId, []), State) ->
     send(?SUBACK_PACKET(PacketId, []), State);
 
-process(?SUBSCRIBE_PACKET(PacketId, RawTopicTable), State = #proto_state{
-        client_id = ClientId, username = Username, session = Session}) ->
-    Client = client(State),
-    TopicTable = parse_topic_table(RawTopicTable),
-    AllowDenies = [check_acl(subscribe, Topic, Client) || {Topic, _Opts} <- TopicTable],
+%% TODO: refactor later...
+process(?SUBSCRIBE_PACKET(PacketId, RawTopicTable), State = #proto_state{session = Session,
+        client_id = ClientId, username = Username, is_superuser = IsSuperuser}) ->
+    Client = client(State), TopicTable = parse_topic_table(RawTopicTable),
+    AllowDenies = if
+                    IsSuperuser -> [];
+                    true -> [check_acl(subscribe, Topic, Client) || {Topic, _Opts} <- TopicTable]
+                  end,
     case lists:member(deny, AllowDenies) of
         true ->
             ?LOG(error, "Cannot SUBSCRIBE ~p for ACL Deny", [TopicTable], State),