acl

apps/emqttd/include/emqttd.hrl

@@ -73,9 +73,11 @@
     clientid    :: binary(),
     peername    :: list(),
     username    :: binary(),
-    passwdhash  :: binary()
+    password    :: binary()
 }).
 
+-type mqtt_user() :: #mqtt_user{}.
+
 %%------------------------------------------------------------------------------
 %% MQTT Authorization
 %%------------------------------------------------------------------------------

apps/emqttd/src/emqttd_acl.erl

@@ -27,42 +27,139 @@
 %%% subscribe topic
 %%% publish to topic
 %%%
+%%% TODO: Support regexp...
+%%%
 %%% @end
 %%%-----------------------------------------------------------------------------
 -module(emqttd_acl).
 
+-author('feng@emqtt.io').
+
+-include("emqttd.hrl").
+
 -behaviour(gen_server).
 
 -define(SERVER, ?MODULE).
 
+-define(ACL_TAB, mqtt_acl).
+
 %% API Function Exports
--export([start_link/0, allow/3]).
+-export([start_link/0, check/3, allow/3, deny/3]).
 
 %% gen_server callbacks
 -export([init/1, handle_call/3, handle_cast/2, handle_info/2,
          terminate/2, code_change/3]).
 
-%% ------------------------------------------------------------------
+-type pubsub() :: publish | subscribe.
-%% API Function Definitions
-%% ------------------------------------------------------------------
 
+-type who() :: all | 
+               {clientid, binary()} | 
+               {peername, string() | inet:ip_address()} |
+               {username, binary()}.
+
+-type rule() :: {allow, all} |
+                {allow, who(), binary()} |
+                {deny,  all} |
+                {deny,  who(), binary()}.
+
+-record(mqtt_acl, {pubsub   :: pubsub(),
+                   rules    :: list(rule())}).
+
+%%%=============================================================================
+%%% API
+%%%=============================================================================
+
+%%------------------------------------------------------------------------------
+%% @doc
+%% Start ACL Server.
+%%
+%% @end
+%%------------------------------------------------------------------------------
+-spec start_link() -> {ok, pid()} | ignore | {error, any()}.
 start_link() ->
     gen_server:start_link({local, ?SERVER}, ?MODULE, [], []).
 
-allow(subscribe, User, Topic) ->
+-spec check(PubSub, User, Topic) -> allowed | refused when
-    true;
+      PubSub :: pubsub(),
-allow(publish, User, Topic) ->
+      User   :: mqtt_user(),
-    true.
+      Topic  :: binary().
+check(PubSub, User, Topic) ->
+    case match(User, Topic, lookup(PubSub)) of
+        nomatch -> allowed;
+        allowed -> allowed;
+        refused -> refused
+    end.
+
+lookup(PubSub) ->
+    case ets:lookup(?ACL_TAB, PubSub) of
+        [] -> [];
+        [#mqtt_acl{pubsub = PubSub, rules = Rules}] -> Rules
+    end.
+
+match(_User, _Topic, []) ->
+    nomatch;
+match(User, Topic, Rules) ->
+    %TODO:...
+    nomatch.
+
+-spec allow(PubSub, Who, Topic) -> ok | {error, any()} when 
+      PubSub :: pubsub(),
+      Who    :: who(),
+      Topic  :: binary().
+allow(PubSub, Who, Topic) ->
+    add_rule(PubSub, {allow, Who, Topic}).
+
+-spec deny(PubSub, Who, Topic) -> ok | {error, any()} when
+      PubSub :: pubsub(),
+      Who    :: who(),
+      Topic  :: binary().
+deny(PubSub, Who, Topic) ->
+    add_rule(PubSub, {deny, Who, Topic}).
+
+add_rule(PubSub, RawRule) ->
+    case rule(RawRule) of
+        {error, Error} ->
+            {error, Error};
+        Rule -> 
+            F = fun() -> 
+                case mnesia:wread(?ACL_TAB, PubSub) of
+                    [] ->
+                        mnesia:write(?ACL_TAB, #mqtt_acl{pubsub = PubSub, rules = [Rule]});
+                    [Rules] ->
+                        mnesia:write(?ACL_TAB, #mqtt_acl{pubsub = PubSub, rules = [Rule|Rules]})
+                end
+            end,
+            case mnesia:transaction(F) of
+                {atomic, _} -> ok;
+                {aborted, Reason} -> {error, {aborted, Reason}}
+            end
+    end.
+
+%% TODO: 
+-spec rule(rule()) -> rule().
+rule({allow, all}) ->
+    {allow, all};
+rule({allow, Who, Topic}) ->
+    {allow, Who, Topic};
+rule({deny, Who, Topic}) ->
+    {deny, Who, Topic};
+rule({deny, all}) ->
+    {deny, all}.
 
 %% ------------------------------------------------------------------
 %% gen_server Function Definitions
 %% ------------------------------------------------------------------
-
 init(Args) ->
+    mnesia:create_table(?ACL_TAB, [
+		{type, set},
+		{record_name, mqtt_acl},
+		{ram_copies, [node()]},
+		{attributes, record_info(fields, mqtt_acl)}]),
+    mnesia:add_table_copy(?ACL_TAB, node(), ram_copies),
     {ok, Args}.
 
 handle_call(_Request, _From, State) ->
-    {reply, ok, State}.
+    {reply, error, State}.
 
 handle_cast(_Msg, State) ->
     {noreply, State}.
@@ -80,3 +177,5 @@ code_change(_OldVsn, State, _Extra) ->
 %% Internal Function Definitions
 %% ------------------------------------------------------------------
 
+
+

apps/emqttd/src/emqttd_metrics.erl

@@ -26,6 +26,8 @@
 %%%-----------------------------------------------------------------------------
 -module(emqttd_metrics).
 
+-author('feng@emqtt.io').
+
 -include("emqttd_packet.hrl").
 
 -include("emqttd_systop.hrl").

apps/emqttd/src/emqttd_plugin.erl

@@ -24,9 +24,10 @@
 %%%
 %%% @end
 %%%-----------------------------------------------------------------------------
-
 -module(emqttd_plugin).
 
+-author('feng@emqtt.io').
+
 -behaviour(gen_server).
 
 -define(SERVER, ?MODULE).