Database-level security.

This patch builds on the DB-admins feature to store lists of database admin and reader names and roles, as well as a security object which can be used for configuration in validation functions.

git-svn-id: https://svn.apache.org/repos/asf/couchdb/trunk@905436 13f79535-47bb-0310-9956-ffa450edef68

6 files changed, 159 insertions, 53 deletions

diff --git a/src/couchdb/couch_db.erl b/src/couchdb/couch_db.erl
index 47b7c084..e3891821 100644
--- a/src/couchdb/couch_db.erl
+++ b/src/couchdb/couch_db.erl
@@ -22,7 +22,8 @@
 -export([enum_docs/4,enum_docs_since/5]).
 -export([enum_docs_since_reduce_to_count/1,enum_docs_reduce_to_count/1]).
 -export([increment_update_seq/1,get_purge_seq/1,purge_docs/2,get_last_purged/1]).
--export([start_link/3,open_doc_int/3,set_admins/2,get_admins/1,ensure_full_commit/1]).
+-export([start_link/3,open_doc_int/3,ensure_full_commit/1]).
+-export([set_readers/2,get_readers/1,set_admins/2,get_admins/1,set_security/2,get_security/1]).
 -export([init/1,terminate/2,handle_call/3,handle_cast/2,code_change/3,handle_info/2]).
 -export([changes_since/5,changes_since/6,read_doc/2,new_revid/1]).
 
@@ -64,7 +65,18 @@ create(DbName, Options) ->
     couch_server:create(DbName, Options).
 
 open(DbName, Options) ->
-    couch_server:open(DbName, Options).
+    case couch_server:open(DbName, Options) of
+        {ok, Db} ->
+            try
+                check_is_reader(Db),
+                {ok, Db}
+            catch
+                throw:Error ->
+                    close(Db),
+                    throw(Error)
+            end;
+        Else -> Else
+    end.
 
 ensure_full_commit(#db{update_pid=UpdatePid,instance_start_time=StartTime}) ->
     ok = gen_server:call(UpdatePid, full_commit, infinity),
@@ -218,22 +230,103 @@ get_design_docs(#db{fulldocinfo_by_id_btree=Btree}=Db) ->
         [], [{start_key, <<"_design/">>}, {end_key_gt, <<"_design0">>}]),
     {ok, Docs}.
 
-check_is_admin(#db{admins=Admins, user_ctx=#user_ctx{name=Name,roles=Roles}}) ->
-    DbAdmins = [<<"_admin">> | Admins],
-    case DbAdmins -- [Name | Roles] of
-    DbAdmins -> % same list, not an admin
-        throw({unauthorized, <<"You are not a db or server admin.">>});
+check_is_admin(#db{user_ctx=#user_ctx{name=Name,roles=Roles}}=Db) ->
+    {Admins} = get_admins(Db),
+    AdminRoles = [<<"_admin">> | proplists:get_value(roles, Admins, [])],
+    AdminNames = proplists:get_value(names, Admins,[]),
+    case AdminRoles -- Roles of
+    AdminRoles -> % same list, not an admin role
+        case AdminNames -- [Name] of
+        AdminNames -> % same names, not an admin
+            throw({unauthorized, <<"You are not a db or server admin.">>});
+        _ ->
+            ok
+        end;
     _ ->
         ok
     end.
 
-get_admins(#db{admins=Admins}) ->
-    Admins.
+check_is_reader(#db{user_ctx=#user_ctx{name=Name,roles=Roles}=UserCtx}=Db) ->
+    % admins are not readers. this is for good reason. 
+    % we don't want to confuse setting admins with making private dbs
+    {Readers} = get_readers(Db),
+    ReaderRoles = proplists:get_value(roles, Readers,[]),
+    WithAdminRoles = [<<"_admin">> | ReaderRoles],
+    ReaderNames = proplists:get_value(names, Readers,[]),
+    case ReaderRoles ++ ReaderNames of 
+    [] -> ok; % no readers == public access
+    _Else ->
+        case WithAdminRoles -- Roles of
+        WithAdminRoles -> % same list, not an reader role
+            case ReaderNames -- [Name] of
+            ReaderNames -> % same names, not a reader
+                ?LOG_DEBUG("Not a reader: UserCtx ~p vs Names ~p Roles ~p",[UserCtx, ReaderNames, WithAdminRoles]),
+                throw({unauthorized, <<"You are not authorized to access this db.">>});
+            _ ->
+                ok
+            end;
+        _ ->
+            ok
+        end
+    end.
+
+get_admins(#db{security=SecProps}) ->
+    proplists:get_value(admins, SecProps, {[]}).
+
+set_admins(#db{security=SecProps,update_pid=Pid}=Db, Admins) ->
+    check_is_admin(Db),
+    SecProps2 = update_sec_field(admins, SecProps, just_names_and_roles(Admins)),
+    gen_server:call(Pid, {set_security, SecProps2}, infinity).
+
+get_readers(#db{security=SecProps}) ->
+    proplists:get_value(readers, SecProps, {[]}).
+
+set_readers(#db{security=SecProps,update_pid=Pid}=Db, Readers) ->
+    check_is_admin(Db),
+    SecProps2 = update_sec_field(readers, SecProps, just_names_and_roles(Readers)),
+    gen_server:call(Pid, {set_security, SecProps2}, infinity).
+
+get_security(#db{security=SecProps}) ->
+    proplists:get_value(sec_obj, SecProps, {[]}).
 
-set_admins(#db{update_pid=Pid}=Db, Admins) when is_list(Admins) ->
+set_security(#db{security=SecProps, update_pid=Pid}=Db, {SecObjProps}) when is_list(SecObjProps) ->
     check_is_admin(Db),
-    gen_server:call(Pid, {set_admins, Admins}, infinity).
+    SecProps2 = update_sec_field(sec_obj, SecProps, {SecObjProps}),
+    gen_server:call(Pid, {set_security, SecProps2}, infinity).
+
+update_sec_field(Field, SecProps, Value) ->
+    Admins = proplists:get_value(admins, SecProps, {[]}),
+    Readers = proplists:get_value(readers, SecProps, {[]}),
+    SecObj = proplists:get_value(sec_obj, SecProps, {[]}),
+    if Field == admins ->
+        [{admins, Value}];
+    true -> [{admins, Admins}] 
+    end ++ if Field == readers ->
+        [{readers, Value}];
+    true -> [{readers, Readers}] 
+    end ++ if Field == sec_obj ->
+        [{sec_obj, Value}];
+    true -> [{sec_obj, SecObj}]
+    end.
 
+% validate user input and convert proplist to atom keys
+just_names_and_roles({Props}) when is_list(Props) ->
+    Names = case proplists:get_value(<<"names">>,Props) of
+    Ns when is_list(Ns) ->
+            [throw("names must be a JSON list of strings") ||N <- Ns, not is_binary(N)],
+            Ns;
+    _ -> []
+    end,
+    Roles = case proplists:get_value(<<"roles">>,Props) of
+    Rs when is_list(Rs) ->
+        [throw("roles must be a JSON list of strings") ||R <- Rs, not is_binary(R)],
+        Rs;
+    _ -> []
+    end,
+    {[
+        {names, Names},
+        {roles, Roles}
+    ]}.
 
 get_revs_limit(#db{revs_limit=Limit}) ->
     Limit.
@@ -282,18 +375,8 @@ group_alike_docs([Doc|Rest], [Bucket|RestBuckets]) ->
        group_alike_docs(Rest, [[Doc]|[Bucket|RestBuckets]])
     end.
 
-
-validate_doc_update(#db{user_ctx=UserCtx, admins=Admins},
-        #doc{id= <<"_design/",_/binary>>}, _GetDiskDocFun) ->
-    UserNames = [UserCtx#user_ctx.name | UserCtx#user_ctx.roles],
-    % if the user is a server admin or db admin, allow the save
-    case length(UserNames -- [<<"_admin">> | Admins]) =:= length(UserNames) of
-    true ->
-        % not an admin
-        {unauthorized, <<"You are not a server or database admin.">>};
-    false ->
-        ok
-    end;
+validate_doc_update(#db{}=Db, #doc{id= <<"_design/",_/binary>>}, _GetDiskDocFun) ->
+    catch check_is_admin(Db);
 validate_doc_update(#db{validate_doc_funs=[]}, _Doc, _GetDiskDocFun) ->
     ok;
 validate_doc_update(_Db, #doc{id= <<"_local/",_/binary>>}, _GetDiskDocFun) ->
@@ -301,7 +384,8 @@ validate_doc_update(_Db, #doc{id= <<"_local/",_/binary>>}, _GetDiskDocFun) ->
 validate_doc_update(Db, Doc, GetDiskDocFun) ->
     DiskDoc = GetDiskDocFun(),
     JsonCtx = couch_util:json_user_ctx(Db),
-    try [case Fun(Doc, DiskDoc, JsonCtx) of
+    SecObj = get_security(Db),
+    try [case Fun(Doc, DiskDoc, JsonCtx, SecObj) of
             ok -> ok;
             Error -> throw(Error)
         end || Fun <- Db#db.validate_doc_funs],
diff --git a/src/couchdb/couch_db.hrl b/src/couchdb/couch_db.hrl
index 31b66edb..47e8f9eb 100644
--- a/src/couchdb/couch_db.hrl
+++ b/src/couchdb/couch_db.hrl
@@ -126,7 +126,7 @@
 % if the disk revision is incremented, then new upgrade logic will need to be
 % added to couch_db_updater:init_db.
 
--define(LATEST_DISK_VERSION, 4).
+-define(LATEST_DISK_VERSION, 5).
 
 -record(db_header,
     {disk_version = ?LATEST_DISK_VERSION,
@@ -137,7 +137,7 @@
      local_docs_btree_state = nil,
      purge_seq = 0,
      purged_docs = nil,
-     admins_ptr = nil,
+     security_ptr = nil,
      revs_limit = 1000
     }).
 
@@ -157,8 +157,8 @@
     name,
     filepath,
     validate_doc_funs = [],
-    admins = [],
-    admins_ptr = nil,
+    security = [],
+    security_ptr = nil,
     user_ctx = #user_ctx{},
     waiting_delayed_commit = nil,
     revs_limit = 1000,
diff --git a/src/couchdb/couch_db_updater.erl b/src/couchdb/couch_db_updater.erl
index 723fc11c..d0f8ec61 100644
--- a/src/couchdb/couch_db_updater.erl
+++ b/src/couchdb/couch_db_updater.erl
@@ -53,9 +53,9 @@ handle_call(increment_update_seq, _From, Db) ->
     couch_db_update_notifier:notify({updated, Db#db.name}),
     {reply, {ok, Db2#db.update_seq}, Db2};
 
-handle_call({set_admins, NewAdmins}, _From, Db) ->
-    {ok, Ptr} = couch_file:append_term(Db#db.fd, NewAdmins),
-    Db2 = commit_data(Db#db{admins=NewAdmins, admins_ptr=Ptr,
+handle_call({set_security, NewSec}, _From, Db) ->
+    {ok, Ptr} = couch_file:append_term(Db#db.fd, NewSec),
+    Db2 = commit_data(Db#db{security=NewSec, security_ptr=Ptr,
             update_seq=Db#db.update_seq+1}),
     ok = gen_server:call(Db2#db.main_pid, {db_updated, Db2}),
     {reply, ok, Db2};
@@ -326,7 +326,7 @@ btree_by_seq_reduce(rereduce, Reds) ->
 
 simple_upgrade_record(Old, New) when tuple_size(Old) =:= tuple_size(New) ->
     Old;
-simple_upgrade_record(Old, New) ->
+simple_upgrade_record(Old, New) when tuple_size(Old) < tuple_size(New) ->
     OldSz = tuple_size(Old),
     NewValuesTail =
         lists:sublist(tuple_to_list(New), OldSz + 1, tuple_size(New) - OldSz),
@@ -337,9 +337,10 @@ init_db(DbName, Filepath, Fd, Header0) ->
     Header1 = simple_upgrade_record(Header0, #db_header{}),
     Header =
     case element(2, Header1) of
-    1 -> Header1#db_header{unused = 0}; % 0.9
-    2 -> Header1#db_header{unused = 0}; % post 0.9 and pre 0.10
-    3 -> Header1; % post 0.9 and pre 0.10
+    1 -> Header1#db_header{unused = 0, security_ptr = nil}; % 0.9
+    2 -> Header1#db_header{unused = 0, security_ptr = nil}; % post 0.9 and pre 0.10
+    3 -> Header1#db_header{security_ptr = nil}; % post 0.9 and pre 0.10
+    4 -> Header1#db_header{security_ptr = nil}; % 0.10 and pre 0.11
     ?LATEST_DISK_VERSION -> Header1;
     _ -> throw({database_disk_version_error, "Incorrect disk header version"})
     end,
@@ -362,12 +363,12 @@ init_db(DbName, Filepath, Fd, Header0) ->
             {join, fun(X,Y) -> btree_by_seq_join(X,Y) end},
             {reduce, fun(X,Y) -> btree_by_seq_reduce(X,Y) end}]),
     {ok, LocalDocsBtree} = couch_btree:open(Header#db_header.local_docs_btree_state, Fd),
-    case Header#db_header.admins_ptr of
+    case Header#db_header.security_ptr of
     nil ->
-        Admins = [],
-        AdminsPtr = nil;
-    AdminsPtr ->
-        {ok, Admins} = couch_file:pread_term(Fd, AdminsPtr)
+        Security = [],
+        SecurityPtr = nil;
+    SecurityPtr ->
+        {ok, Security} = couch_file:pread_term(Fd, SecurityPtr)
     end,
     % convert start time tuple to microsecs and store as a binary string
     {MegaSecs, Secs, MicroSecs} = now(),
@@ -386,8 +387,8 @@ init_db(DbName, Filepath, Fd, Header0) ->
         update_seq = Header#db_header.update_seq,
         name = DbName,
         filepath = Filepath,
-        admins = Admins,
-        admins_ptr = AdminsPtr,
+        security = Security,
+        security_ptr = SecurityPtr,
         instance_start_time = StartTime,
         revs_limit = Header#db_header.revs_limit,
         fsync_options = FsyncOptions
@@ -655,7 +656,7 @@ db_to_header(Db, Header) ->
         docinfo_by_seq_btree_state = couch_btree:get_state(Db#db.docinfo_by_seq_btree),
         fulldocinfo_by_id_btree_state = couch_btree:get_state(Db#db.fulldocinfo_by_id_btree),
         local_docs_btree_state = couch_btree:get_state(Db#db.local_docs_btree),
-        admins_ptr = Db#db.admins_ptr,
+        security_ptr = Db#db.security_ptr,
         revs_limit = Db#db.revs_limit}.
 
 commit_data(#db{fd=Fd,header=OldHeader,fsync_options=FsyncOptions}=Db, Delay) ->
@@ -810,9 +811,9 @@ copy_compact(Db, NewDb0, Retry) ->
     NewDb3 = copy_docs(Db, NewDb2, lists:reverse(Uncopied), Retry),
 
     % copy misc header values
-    if NewDb3#db.admins /= Db#db.admins ->
-        {ok, Ptr} = couch_file:append_term(NewDb3#db.fd, Db#db.admins),
-        NewDb4 = NewDb3#db{admins=Db#db.admins, admins_ptr=Ptr};
+    if NewDb3#db.security /= Db#db.security ->
+        {ok, Ptr} = couch_file:append_term(NewDb3#db.fd, Db#db.security),
+        NewDb4 = NewDb3#db{security=Db#db.security, security_ptr=Ptr};
     true ->
         NewDb4 = NewDb3
     end,
diff --git a/src/couchdb/couch_doc.erl b/src/couchdb/couch_doc.erl
index 48ed1530..bbcbd0ba 100644
--- a/src/couchdb/couch_doc.erl
+++ b/src/couchdb/couch_doc.erl
@@ -307,8 +307,8 @@ get_validate_doc_fun(#doc{body={Props}}=DDoc) ->
     undefined ->
         nil;
     _Else ->
-        fun(EditDoc, DiskDoc, Ctx) ->
-            couch_query_servers:validate_doc_update(DDoc, EditDoc, DiskDoc, Ctx)
+        fun(EditDoc, DiskDoc, Ctx, SecObj) ->
+            couch_query_servers:validate_doc_update(DDoc, EditDoc, DiskDoc, Ctx, SecObj)
         end
     end.
 
diff --git a/src/couchdb/couch_httpd_db.erl b/src/couchdb/couch_httpd_db.erl
index fd143fa1..5421c214 100644
--- a/src/couchdb/couch_httpd_db.erl
+++ b/src/couchdb/couch_httpd_db.erl
@@ -548,8 +548,7 @@ db_req(#httpd{path_parts=[_,<<"_revs_diff">>]}=Req, _Db) ->
     send_method_not_allowed(Req, "POST");
 
 
-db_req(#httpd{method='PUT',path_parts=[_,<<"_admins">>]}=Req,
-        Db) ->
+db_req(#httpd{method='PUT',path_parts=[_,<<"_admins">>]}=Req, Db) ->
     Admins = couch_httpd:json_body(Req),
     ok = couch_db:set_admins(Db, Admins),
     send_json(Req, {[{<<"ok">>, true}]});
@@ -560,6 +559,28 @@ db_req(#httpd{method='GET',path_parts=[_,<<"_admins">>]}=Req, Db) ->
 db_req(#httpd{path_parts=[_,<<"_admins">>]}=Req, _Db) ->
     send_method_not_allowed(Req, "PUT,GET");
 
+db_req(#httpd{method='PUT',path_parts=[_,<<"_readers">>]}=Req, Db) ->
+    Readers = couch_httpd:json_body(Req),
+    ok = couch_db:set_readers(Db, Readers),
+    send_json(Req, {[{<<"ok">>, true}]});
+
+db_req(#httpd{method='GET',path_parts=[_,<<"_readers">>]}=Req, Db) ->
+    send_json(Req, couch_db:get_readers(Db));
+
+db_req(#httpd{path_parts=[_,<<"_readers">>]}=Req, _Db) ->
+    send_method_not_allowed(Req, "PUT,GET");
+
+db_req(#httpd{method='PUT',path_parts=[_,<<"_security">>]}=Req, Db) ->
+    SecObj = couch_httpd:json_body(Req),
+    ok = couch_db:set_security(Db, SecObj),
+    send_json(Req, {[{<<"ok">>, true}]});
+
+db_req(#httpd{method='GET',path_parts=[_,<<"_security">>]}=Req, Db) ->
+    send_json(Req, couch_db:get_security(Db));
+
+db_req(#httpd{path_parts=[_,<<"_security">>]}=Req, _Db) ->
+    send_method_not_allowed(Req, "PUT,GET");
+
 db_req(#httpd{method='PUT',path_parts=[_,<<"_revs_limit">>]}=Req,
         Db) ->
     Limit = couch_httpd:json_body(Req),
diff --git a/src/couchdb/couch_query_servers.erl b/src/couchdb/couch_query_servers.erl
index 30f4c4c7..dd5fee94 100644
--- a/src/couchdb/couch_query_servers.erl
+++ b/src/couchdb/couch_query_servers.erl
@@ -17,7 +17,7 @@
 
 -export([init/1, terminate/2, handle_call/3, handle_cast/2, handle_info/2,code_change/3,stop/0]).
 -export([start_doc_map/2, map_docs/2, stop_doc_map/1]).
--export([reduce/3, rereduce/3,validate_doc_update/4]).
+-export([reduce/3, rereduce/3,validate_doc_update/5]).
 -export([filter_docs/5]).
 
 -export([with_ddoc_proc/2, proc_prompt/2, ddoc_prompt/3, ddoc_proc_prompt/3, json_doc/1]).
@@ -166,10 +166,10 @@ builtin_sum_rows(KVs) ->
 
 
 % use the function stored in ddoc.validate_doc_update to test an update.
-validate_doc_update(DDoc, EditDoc, DiskDoc, Ctx) ->
+validate_doc_update(DDoc, EditDoc, DiskDoc, Ctx, SecObj) ->
     JsonEditDoc = couch_doc:to_json_obj(EditDoc, [revs]),
     JsonDiskDoc = json_doc(DiskDoc),  
-    case ddoc_prompt(DDoc, [<<"validate_doc_update">>], [JsonEditDoc, JsonDiskDoc, Ctx]) of
+    case ddoc_prompt(DDoc, [<<"validate_doc_update">>], [JsonEditDoc, JsonDiskDoc, Ctx, SecObj]) of
         1 ->
             ok;
         {[{<<"forbidden">>, Message}]} ->