Merge pull request #378 from esl/riak-base

Riak base

.travis.yml

@@ -22,12 +22,12 @@ after_script:
 after_success:
         - make cover_report
 
-services: redis-server
+services:
+        - redis-server
 branches:
         only:
                 - master
-notifications:
-        email: mongoose-im@erlang-solutions.com
+
 otp_release:
         - R16B03
 env:
@@ -37,6 +37,7 @@ env:
         - PRESET=odbc_pgsql_mnesia DB=pgsql REL_CONFIG=with-odbc
         - PRESET=pgsql_mnesia DB=pgsql REL_CONFIG=with-pgsql
         - PRESET=ldap_mnesia DB=mnesia
+        - PRESET=riak_mnesia DB=riak REL_CONFIG=with-riak
         - PRESET=external_mnesia DB=mnesia
 
 matrix:

README.md

@@ -123,6 +123,7 @@ with-mysql	include mysql driver
 with-pgsql	include pgsql driver
 with-odbc	include standard ODBC driver shipped with Erlang/OTP
 with-redis	include redis driver
+with-riak      include riak driver
 with-cassandra	include cassandra driver
 full		include all above deps
     ```
@@ -133,6 +134,9 @@ full		include all above deps
 
     The `make configure` command has to be run only once (unless one need to change the relase config and include some other dependecies).
 
+    Take a look [here](http://mongooseim.readthedocs.org/en/latest/advanced-configuration/database-backends-configuration/)
+    for instructions how to setup the external databases.
+
     `make rel` or `./rebar generate` commands will generate a self-contained OTP system image in the
     project's `rel/mongooseim` subdirectory. The contents of that directory are as
     follows:

apps/ejabberd/src/ejabberd_app.erl

@@ -61,6 +61,7 @@ start(normal, _Args) ->
     mongoose_metrics:init(),
     ejabberd_system_monitor:add_handler(),
     ejabberd_rdbms:start(),
+    mongoose_riak:start(),
     ejabberd_auth:start(),
     cyrsasl:start(),
     %% Profiling

apps/ejabberd/src/ejabberd_auth_http.erl

@@ -46,12 +46,13 @@ start(Host) ->
     PoolSize = proplists:get_value(connection_pool_size, AuthOpts, 10),
     Opts = proplists:get_value(connection_opts, AuthOpts, []),
     ChildMods = [fusco],
-    ChildMFA = {fusco, start_link, [AuthHost, Opts]},
+    ChildMF = {fusco, start_link},
+    ChildArgs = {for_all, [AuthHost, Opts]},
 
     {ok, _} = supervisor:start_child(ejabberd_sup,
                                      {{ejabberd_auth_http_sup, Host},
                                       {cuesport, start_link,
-                                       [pool_name(Host), PoolSize, ChildMods, ChildMFA]},
+                                       [pool_name(Host), PoolSize, ChildMods, ChildMF, ChildArgs]},
                                       transient, 2000, supervisor, [cuesport | ChildMods]}),
     ok.
 

apps/ejabberd/src/ejabberd_auth_riak.erl

@@ -0,0 +1,269 @@
+%%==============================================================================
+%% Copyright 2015 Erlang Solutions Ltd.
+%%
+%% 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.
+%%==============================================================================
+-module(ejabberd_auth_riak).
+
+-behaviour(ejabberd_gen_auth).
+
+-include("ejabberd.hrl").
+
+%% API
+-export([start/1,
+         stop/1,
+         store_type/1,
+         login/2,
+         set_password/3,
+         check_password/3,
+         check_password/5,
+         try_register/3,
+         dirty_get_registered_users/0,
+         get_vh_registered_users/1,
+         get_vh_registered_users/2,
+         get_vh_registered_users_number/1,
+         get_vh_registered_users_number/2,
+         get_password/2,
+         get_password_s/2,
+         get_password/3,
+         does_user_exist/2,
+         remove_user/2,
+         remove_user/3,
+         plain_password_required/0]).
+
+-export([bucket_type/1]).
+
+-spec start(ejabberd:lserver()) -> ok.
+start(_Host) ->
+    ok.
+
+-spec stop(ejabberd:lserver()) -> ok.
+stop(_Host) ->
+    ok.
+
+-spec store_type(ejabberd:lserver()) -> plain | scram.
+store_type(Host) ->
+    case scram:enabled(Host) of
+        false -> plain;
+        true -> scram
+    end.
+
+-spec set_password(ejabberd:luser(),ejabberd:lserver(), binary())
+        -> ok | {error, not_allowed | invalid_jid}.
+set_password(LUser, LServer, Password) ->
+    case prepare_password(LServer, Password) of
+        false ->
+            {error, invalid_password};
+        Password ->
+            User = mongoose_riak:fetch_type(bucket_type(LServer), LUser),
+            do_set_password(User, LUser, LServer, Password)
+    end.
+
+-spec check_password(ejabberd:luser(), ejabberd:lserver(), binary()) -> boolean().
+check_password(LUser, LServer, Password) ->
+    case do_get_password(LUser, LServer) of
+        false ->
+            false;
+        #scram{} = Scram ->
+            scram:check_password(Password, Scram);
+        Password when is_binary(Password) ->
+            Password /= <<"">>;
+        _ ->
+            false
+    end.
+
+-spec check_password(ejabberd:luser(),
+                     ejabberd:lserver(),
+                     binary(),
+                     binary(),
+                     fun()) -> boolean().
+check_password(LUser, LServer, Password, Digest, DigestGen) ->
+    case do_get_password(LUser, LServer) of
+        false ->
+            false;
+        #scram{} = Scram ->
+            scram:check_digest(Scram, Digest, DigestGen, Password);
+        PassRiak when is_binary(PassRiak) ->
+            ejabberd_auth:check_digest(Digest, DigestGen, Password, PassRiak)
+    end.
+
+try_register(LUser, LServer, Password) ->
+    try_register_if_does_not_exist(LUser, LServer, Password).
+
+-spec dirty_get_registered_users() -> [ejabberd:simple_jid()].
+dirty_get_registered_users() ->
+    Servers = ejabberd_config:get_vh_by_auth_method(riak),
+    lists:flatmap(
+        fun(Server) ->
+            get_vh_registered_users(Server)
+        end, Servers).
+
+-spec get_vh_registered_users(ejabberd:lserver()) ->
+    [ejabberd:simple_jid()].
+get_vh_registered_users(LServer) ->
+    case mongoose_riak:list_keys(bucket_type(LServer)) of
+        {ok, Users} ->
+            [{User, LServer} || User <- Users];
+        _ ->
+            []
+    end.
+
+-spec get_vh_registered_users(ejabberd:lserver(), list()) ->
+    [ejabberd:simple_jid()].
+get_vh_registered_users(LServer, _Opts) ->
+    get_vh_registered_users(LServer).
+
+-spec get_vh_registered_users_number(ejabberd:lserver()) -> non_neg_integer().
+get_vh_registered_users_number(LServer) ->
+    length(get_vh_registered_users(LServer)).
+
+-spec get_vh_registered_users_number(ejabberd:lserver(), list()) -> non_neg_integer().
+get_vh_registered_users_number(LServer, _Opts) ->
+    get_vh_registered_users_number(LServer).
+
+-spec get_password(ejabberd:luser(), ejabberd:lserver()) -> binary() | false | scram().
+get_password(LUser, LServer) ->
+    case do_get_password(LUser, LServer) of
+        false ->
+            false;
+        #scram{} = Scram ->
+            scram:scram_to_tuple(Scram);
+        Password ->
+            Password
+    end.
+
+get_password_s(LUser, LServer) ->
+    case get_password(LUser, LServer) of
+        Password when is_binary(Password) ->
+            Password;
+        _ ->
+            <<"">>
+    end.
+
+get_password(_LUser, _LServer, _DefaultValue) ->
+    erlang:error(not_implemented).
+
+-spec does_user_exist(ejabberd:luser(), ejabberd:lserver()) -> boolean().
+does_user_exist(LUser, LServer) ->
+    case mongoose_riak:fetch_type(bucket_type(LServer), LUser) of
+        {ok, _} ->
+            true;
+        {error, {notfound, map}} ->
+            false
+    end.
+
+-spec remove_user(ejabberd:luser(), ejabberd:lserver()) ->
+    ok | {error, term()}.
+remove_user(LUser, LServer) ->
+    mongoose_riak:delete(bucket_type(LServer), LUser).
+
+remove_user(_LUser, _LServer, _Password) ->
+    erlang:error(not_implemented).
+
+plain_password_required() ->
+    false.
+
+login(_LUser, _LServer) ->
+    erlang:error(not_implemented).
+
+-spec bucket_type(ejabberd:lserver()) -> {binary(), ejabberd:lserver()}.
+bucket_type(LServer) ->
+    {<<"users">>, LServer}.
+
+%% -----------------------------------------------------------------------------
+%% Internal functions
+%% -----------------------------------------------------------------------------
+
+try_register_if_does_not_exist(LUser, LServer, _)
+    when LUser =:= error; LServer =:= error ->
+    {error, invalid_jid};
+try_register_if_does_not_exist(LUser, LServer, PasswordIn) ->
+   case does_user_exist(LUser, LServer) of
+       false ->
+           Password = prepare_password(LServer, PasswordIn),
+           try_register_with_password(LUser, LServer, Password);
+       true ->
+           {error, exists}
+   end.
+
+try_register_with_password(LUser, LServer, Password) ->
+    Now = integer_to_binary(now_to_seconds(os:timestamp())),
+    Ops = [{{<<"created">>, register},
+            fun(R) -> riakc_register:set(Now, R) end},
+           set_password_map_op(Password)],
+    UserMap = mongoose_riak:create_new_map(Ops),
+    case mongoose_riak:update_type(bucket_type(LServer), LUser,
+                                   riakc_map:to_op(UserMap)) of
+        {ok, _Map} ->
+            ok;
+        Error ->
+            Error
+    end.
+
+do_get_password(LUser, LServer) ->
+    case mongoose_riak:fetch_type(bucket_type(LServer), LUser) of
+        {ok, Map} ->
+            extract_password(Map);
+        _ ->
+            false
+    end.
+
+do_set_password({ok, Map}, LUser, LServer, Password) ->
+    Ops = [set_password_map_op(Password)],
+    UpdateMap = mongoose_riak:update_map(Map, Ops),
+    case mongoose_riak:update_type(bucket_type(LServer), LUser,
+                                   riakc_map:to_op(UpdateMap)) of
+        ok ->
+            ok;
+        Reason ->
+            Reason
+    end.
+
+prepare_password(Iterations, Password) when is_integer(Iterations) ->
+    Scram = scram:password_to_scram(Password, Iterations),
+    PassDetails = scram:serialize(Scram),
+    {<<"">>, PassDetails};
+
+prepare_password(Server, Password) ->
+    case scram:enabled(Server) of
+        true ->
+            prepare_password(scram:iterations(Server), Password);
+        _ ->
+            Password
+    end.
+
+set_password_map_op({_, Scram}) ->
+    {{<<"scram">>, register}, fun(R) -> riakc_register:set(Scram, R) end};
+set_password_map_op(Password) ->
+    {{<<"password">>, register}, fun(R) -> riakc_register:set(Password, R) end}.
+
+extract_password(Map) ->
+    case riakc_map:find({<<"password">>, register}, Map) of
+        error ->
+            maybe_extract_scram_password(riakc_map:find({<<"scram">>, register}, Map));
+        {ok, Password} ->
+            Password
+    end.
+
+maybe_extract_scram_password(false) ->
+    false;
+maybe_extract_scram_password({ok, ScramSerialised}) ->
+    case scram:deserialize(ScramSerialised) of
+        {ok, Scram} ->
+            Scram;
+        _ ->
+            false
+    end.
+
+now_to_seconds({MegaSecs, Secs, _MicroSecs}) ->
+    MegaSecs * 1000000 + Secs.

apps/ejabberd/src/ejabberd_config.erl

@@ -605,6 +605,8 @@ process_host_term(Term, Host, State) ->
             State;
         {odbc_server, ODBC_server} ->
             add_option({odbc_server, Host}, ODBC_server, State);
+        {riak_server, RiakConfig} ->
+            add_option(riak_server, RiakConfig, State);
         {Opt, Val} ->
             add_option({Opt, Host}, Val, State)
     end.
@@ -1000,18 +1002,22 @@ handle_config_change({_Key, _OldValue, _NewValue}) ->
 %% ----------------------------------------------------------------
 %% LOCAL CONFIG
 %% ----------------------------------------------------------------
-handle_local_config_add({Key, _Data} = El) ->
+handle_local_config_add(#local_config{key = riak_server}) ->
+    mongoose_riak:start();
+handle_local_config_add(#local_config{key=Key} = El) ->
     case Key of
         true ->
             ok;
         false ->
             ?WARNING_MSG("local config add ~p option unhandled",[El])
     end.
 
+handle_local_config_del(#local_config{key = riak_server}) ->
+    mongoose_riak:stop();
 handle_local_config_del(#local_config{key = node_start}) ->
     %% do nothing with it
     ok;
-handle_local_config_del({Key, _Data} = El) ->
+handle_local_config_del(#local_config{key=Key} = El) ->
     case can_be_ignored(Key) of
         true ->
             ok;
@@ -1021,6 +1027,10 @@ handle_local_config_del({Key, _Data} = El) ->
 
 handle_local_config_change({listen, Old, New}) ->
     reload_listeners(compare_listeners(Old, New));
+handle_local_config_change({riak_server, _Old, _New}) ->
+    mongoose_riak:stop(),
+    mongoose_riak:start(),
+    ok;
 
 handle_local_config_change({Key, _Old, _New} = El) ->
     case can_be_ignored(Key) of

apps/ejabberd/src/ejabberd_redis.erl

@@ -18,12 +18,13 @@ start_link(Opts) ->
     PoolSize = proplists:get_value(pool_size, Opts, 10),
     RedoOpts = proplists:get_value(worker_config, Opts, []),
     ChildMods = [redo, redo_redis_proto, redo_uri],
-    ChildMFA = {redo, start_link, [undefined, RedoOpts]},
+    ChildMF = {redo, start_link},
+    ChildArgs = {for_all, [undefined, RedoOpts]},
 
     supervisor:start_child(ejabberd_sm_backend_sup,
                            {ejabberd_redis_sup,
                             {cuesport, start_link,
-                             [?POOL_NAME, PoolSize, ChildMods, ChildMFA]},
+                             [?POOL_NAME, PoolSize, ChildMods, ChildMF, ChildArgs]},
                             transient, 2000, supervisor, [cuesport | ChildMods]}).
 
 -spec cmd(iolist()) -> binary()