You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@couchdb.apache.org by ja...@apache.org on 2022/08/06 14:10:53 UTC
[couchdb] 07/21: feat(access): handle access in couch_db[_updater]
This is an automated email from the ASF dual-hosted git repository.
jan pushed a commit to branch feat/access-2022
in repository https://gitbox.apache.org/repos/asf/couchdb.git
commit 8d2f667a872a3043728d8776f0150ecdac45bcf7
Author: Jan Lehnardt <ja...@apache.org>
AuthorDate: Fri Jun 24 18:43:52 2022 +0200
feat(access): handle access in couch_db[_updater]
---
src/couch/src/couch_db.erl | 219 ++++++++++++++++++++++++++++++++-----
src/couch/src/couch_db_int.hrl | 3 +-
src/couch/src/couch_db_updater.erl | 154 +++++++++++++++++++++-----
3 files changed, 319 insertions(+), 57 deletions(-)
diff --git a/src/couch/src/couch_db.erl b/src/couch/src/couch_db.erl
index 3c4f1edec..a0e7cfaf1 100644
--- a/src/couch/src/couch_db.erl
+++ b/src/couch/src/couch_db.erl
@@ -31,6 +31,9 @@
is_admin/1,
check_is_admin/1,
check_is_member/1,
+ validate_access/2,
+ check_access/2,
+ has_access_enabled/1,
name/1,
get_after_doc_read_fun/1,
@@ -136,6 +139,7 @@
]).
-include_lib("couch/include/couch_db.hrl").
+-include_lib("couch_mrview/include/couch_mrview.hrl"). % TODO: can we do without this?
-include("couch_db_int.hrl").
-define(DBNAME_REGEX,
@@ -285,6 +289,12 @@ wait_for_compaction(#db{main_pid = Pid} = Db, Timeout) ->
is_compacting(DbName) ->
couch_server:is_compacting(DbName).
+has_access_enabled(#db{access=true}) -> true;
+has_access_enabled(_) -> false.
+
+is_read_from_ddoc_cache(Options) ->
+ lists:member(ddoc_cache, Options).
+
delete_doc(Db, Id, Revisions) ->
DeletedDocs = [#doc{id = Id, revs = [Rev], deleted = true} || Rev <- Revisions],
{ok, [Result]} = update_docs(Db, DeletedDocs, []),
@@ -293,23 +303,33 @@ delete_doc(Db, Id, Revisions) ->
open_doc(Db, IdOrDocInfo) ->
open_doc(Db, IdOrDocInfo, []).
-open_doc(Db, Id, Options) ->
+open_doc(Db, Id, Options0) ->
increment_stat(Db, [couchdb, database_reads]),
+ Options = case has_access_enabled(Db) of
+ true -> Options0 ++ [conflicts];
+ _Else -> Options0
+ end,
case open_doc_int(Db, Id, Options) of
{ok, #doc{deleted = true} = Doc} ->
case lists:member(deleted, Options) of
true ->
- apply_open_options({ok, Doc}, Options);
+ {ok, Doc};
false ->
{not_found, deleted}
end;
Else ->
- apply_open_options(Else, Options)
+ Else
end.
-apply_open_options({ok, Doc}, Options) ->
+apply_open_options(Db, {ok, Doc}, Options) ->
+ ok = validate_access(Db, Doc, Options),
+ apply_open_options1({ok, Doc}, Options);
+apply_open_options(_Db, Else, _Options) ->
+ Else.
+
+apply_open_options1({ok, Doc}, Options) ->
apply_open_options2(Doc, Options);
-apply_open_options(Else, _Options) ->
+apply_open_options1(Else, _Options) ->
Else.
apply_open_options2(Doc, []) ->
@@ -350,7 +370,7 @@ find_ancestor_rev_pos({RevPos, [RevId | Rest]}, AttsSinceRevs) ->
open_doc_revs(Db, Id, Revs, Options) ->
increment_stat(Db, [couchdb, database_reads]),
[{ok, Results}] = open_doc_revs_int(Db, [{Id, Revs}], Options),
- {ok, [apply_open_options(Result, Options) || Result <- Results]}.
+ {ok, [apply_open_options(Db, Result, Options) || Result <- Results]}.
% Each returned result is a list of tuples:
% {Id, MissingRevs, PossibleAncestors}
@@ -615,7 +635,8 @@ get_db_info(Db) ->
name = Name,
compactor_pid = Compactor,
instance_start_time = StartTime,
- committed_update_seq = CommittedUpdateSeq
+ committed_update_seq = CommittedUpdateSeq,
+ access = Access
} = Db,
{ok, DocCount} = get_doc_count(Db),
{ok, DelDocCount} = get_del_doc_count(Db),
@@ -650,7 +671,8 @@ get_db_info(Db) ->
{committed_update_seq, CommittedUpdateSeq},
{compacted_seq, CompactedSeq},
{props, Props},
- {uuid, Uuid}
+ {uuid, Uuid},
+ {access, Access}
],
{ok, InfoList}.
@@ -770,6 +792,72 @@ security_error_type(#user_ctx{name = null}) ->
security_error_type(#user_ctx{name = _}) ->
forbidden.
+is_per_user_ddoc(#doc{access=[]}) -> false;
+is_per_user_ddoc(#doc{access=[<<"_users">>]}) -> false;
+is_per_user_ddoc(_) -> true.
+
+validate_access(Db, Doc) ->
+ validate_access(Db, Doc, []).
+
+validate_access(Db, Doc, Options) ->
+ validate_access1(has_access_enabled(Db), Db, Doc, Options).
+
+validate_access1(false, _Db, _Doc, _Options) -> ok;
+validate_access1(true, Db, #doc{meta=Meta}=Doc, Options) ->
+ case proplists:get_value(conflicts, Meta) of
+ undefined -> % no conflicts
+ case is_read_from_ddoc_cache(Options) andalso is_per_user_ddoc(Doc) of
+ true -> throw({not_found, missing});
+ _False -> validate_access2(Db, Doc)
+ end;
+ _Else -> % only admins can read conflicted docs in _access dbs
+ case is_admin(Db) of
+ true -> ok;
+ _Else2 -> throw({forbidden, <<"document is in conflict">>})
+ end
+ end.
+validate_access2(Db, Doc) ->
+ validate_access3(check_access(Db, Doc)).
+
+validate_access3(true) -> ok;
+validate_access3(_) -> throw({forbidden, <<"can't touch this">>}).
+
+check_access(Db, #doc{access=Access}) ->
+ check_access(Db, Access);
+check_access(Db, Access) ->
+ #user_ctx{
+ name=UserName,
+ roles=UserRoles
+ } = Db#db.user_ctx,
+ case Access of
+ [] ->
+ % if doc has no _access, userCtX must be admin
+ is_admin(Db);
+ Access ->
+ % if doc has _access, userCtx must be admin OR matching user or role
+ % _access = ["a", "b", ]
+ case is_admin(Db) of
+ true ->
+ true;
+ _ ->
+ case {check_name(UserName, Access), check_roles(UserRoles, Access)} of
+ {true, _} -> true;
+ {_, true} -> true;
+ _ -> false
+ end
+ end
+ end.
+
+check_name(null, _Access) -> true;
+check_name(UserName, Access) ->
+ lists:member(UserName, Access).
+% nicked from couch_db:check_security
+
+check_roles(Roles, Access) ->
+ UserRolesSet = ordsets:from_list(Roles),
+ RolesSet = ordsets:from_list(Access ++ ["_users"]),
+ not ordsets:is_disjoint(UserRolesSet, RolesSet).
+
get_admins(#db{security = SecProps}) ->
couch_util:get_value(<<"admins">>, SecProps, {[]}).
@@ -911,9 +999,14 @@ group_alike_docs([Doc | Rest], [Bucket | RestBuckets]) ->
end.
validate_doc_update(#db{} = Db, #doc{id = <<"_design/", _/binary>>} = Doc, _GetDiskDocFun) ->
- case catch check_is_admin(Db) of
- ok -> validate_ddoc(Db, Doc);
- Error -> Error
+ case couch_doc:has_access(Doc) of
+ true ->
+ validate_ddoc(Db, Doc);
+ _Else ->
+ case catch check_is_admin(Db) of
+ ok -> validate_ddoc(Db, Doc);
+ Error -> Error
+ end
end;
validate_doc_update(#db{validate_doc_funs = undefined} = Db, Doc, Fun) ->
ValidationFuns = load_validation_funs(Db),
@@ -1308,6 +1401,32 @@ doc_tag(#doc{meta = Meta}) ->
Else -> throw({invalid_doc_tag, Else})
end.
+validate_update(Db, Doc) ->
+ case catch validate_access(Db, Doc) of
+ ok -> Doc;
+ Error -> Error
+ end.
+
+
+validate_docs_access(Db, DocBuckets, DocErrors) ->
+ validate_docs_access1(Db, DocBuckets, {[], DocErrors}).
+
+validate_docs_access1(_Db, [], {DocBuckets0, DocErrors}) ->
+ DocBuckets1 = lists:reverse(lists:map(fun lists:reverse/1, DocBuckets0)),
+ DocBuckets = case DocBuckets1 of
+ [[]] -> [];
+ Else -> Else
+ end,
+ {ok, DocBuckets, lists:reverse(DocErrors)};
+validate_docs_access1(Db, [DocBucket|RestBuckets], {DocAcc, ErrorAcc}) ->
+ {NewBuckets, NewErrors} = lists:foldl(fun(Doc, {Acc, ErrAcc}) ->
+ case catch validate_access(Db, Doc) of
+ ok -> {[Doc|Acc], ErrAcc};
+ Error -> {Acc, [{doc_tag(Doc), Error}|ErrAcc]}
+ end
+ end, {[], ErrorAcc}, DocBucket),
+ validate_docs_access1(Db, RestBuckets, {[NewBuckets | DocAcc], NewErrors}).
+
update_docs(Db, Docs0, Options, ?REPLICATED_CHANGES) ->
Docs = tag_docs(Docs0),
@@ -1331,13 +1450,35 @@ update_docs(Db, Docs0, Options, ?REPLICATED_CHANGES) ->
]
|| Bucket <- DocBuckets
],
- {ok, _} = write_and_commit(
+ {ok, Results} = write_and_commit(
Db,
DocBuckets2,
NonRepDocs,
[merge_conflicts | Options]
),
- {ok, DocErrors};
+ case couch_db:has_access_enabled(Db) of
+ false ->
+ % we’re done here
+ {ok, DocErrors};
+ _ ->
+ AccessViolations = lists:filter(fun({_Ref, Tag}) -> Tag =:= access end, Results),
+ case length(AccessViolations) of
+ 0 ->
+ % we’re done here
+ {ok, DocErrors};
+ _ ->
+ % dig out FDIs from Docs matching our tags/refs
+ DocsDict = lists:foldl(fun(Doc, Dict) ->
+ Tag = doc_tag(Doc),
+ dict:store(Tag, Doc, Dict)
+ end, dict:new(), Docs),
+ AccessResults = lists:map(fun({Ref, Access}) ->
+ { dict:fetch(Ref, DocsDict), Access }
+ end, AccessViolations),
+ {ok, AccessResults}
+ end
+ end;
+
update_docs(Db, Docs0, Options, ?INTERACTIVE_EDIT) ->
Docs = tag_docs(Docs0),
@@ -1459,7 +1600,7 @@ write_and_commit(
MergeConflicts = lists:member(merge_conflicts, Options),
MRef = erlang:monitor(process, Pid),
try
- Pid ! {update_docs, self(), DocBuckets, NonRepDocs, MergeConflicts},
+ Pid ! {update_docs, self(), DocBuckets, NonRepDocs, MergeConflicts, Ctx},
case collect_results_with_metrics(Pid, MRef, []) of
{ok, Results} ->
{ok, Results};
@@ -1474,7 +1615,7 @@ write_and_commit(
% We only retry once
DocBuckets3 = prepare_doc_summaries(Db2, DocBuckets2),
close(Db2),
- Pid ! {update_docs, self(), DocBuckets3, NonRepDocs, MergeConflicts},
+ Pid ! {update_docs, self(), DocBuckets3, NonRepDocs, MergeConflicts, Ctx},
case collect_results_with_metrics(Pid, MRef, []) of
{ok, Results} -> {ok, Results};
retry -> throw({update_error, compaction_retry})
@@ -1686,6 +1827,12 @@ open_read_stream(Db, AttState) ->
is_active_stream(Db, StreamEngine) ->
couch_db_engine:is_active_stream(Db, StreamEngine).
+changes_since(Db, StartSeq, Fun, Options, Acc) when is_record(Db, db) ->
+ case couch_db:has_access_enabled(Db) and not couch_db:is_admin(Db) of
+ true -> couch_mrview:query_changes_access(Db, StartSeq, Fun, Options, Acc);
+ false -> couch_db_engine:fold_changes(Db, StartSeq, Fun, Options, Acc)
+ end.
+
calculate_start_seq(_Db, _Node, Seq) when is_integer(Seq) ->
Seq;
calculate_start_seq(Db, Node, {Seq, Uuid}) ->
@@ -1814,7 +1961,10 @@ fold_changes(Db, StartSeq, UserFun, UserAcc) ->
fold_changes(Db, StartSeq, UserFun, UserAcc, []).
fold_changes(Db, StartSeq, UserFun, UserAcc, Opts) ->
- couch_db_engine:fold_changes(Db, StartSeq, UserFun, UserAcc, Opts).
+ case couch_db:has_access_enabled(Db) and not couch_db:is_admin(Db) of
+ true -> couch_mrview:query_changes_access(Db, StartSeq, UserFun, Opts, UserAcc);
+ false -> couch_db_engine:fold_changes(Db, StartSeq, UserFun, UserAcc, Opts)
+ end.
fold_purge_infos(Db, StartPurgeSeq, Fun, Acc) ->
fold_purge_infos(Db, StartPurgeSeq, Fun, Acc, []).
@@ -1832,7 +1982,7 @@ open_doc_revs_int(Db, IdRevs, Options) ->
lists:zipwith(
fun({Id, Revs}, Lookup) ->
case Lookup of
- #full_doc_info{rev_tree = RevTree} ->
+ #full_doc_info{rev_tree = RevTree, access = Access} ->
{FoundRevs, MissingRevs} =
case Revs of
all ->
@@ -1853,7 +2003,7 @@ open_doc_revs_int(Db, IdRevs, Options) ->
% we have the rev in our list but know nothing about it
{{not_found, missing}, {Pos, Rev}};
#leaf{deleted = IsDeleted, ptr = SummaryPtr} ->
- {ok, make_doc(Db, Id, IsDeleted, SummaryPtr, FoundRevPath)}
+ {ok, make_doc(Db, Id, IsDeleted, SummaryPtr, FoundRevPath, Access)}
end
end,
FoundRevs
@@ -1875,23 +2025,29 @@ open_doc_revs_int(Db, IdRevs, Options) ->
open_doc_int(Db, <<?LOCAL_DOC_PREFIX, _/binary>> = Id, Options) ->
case couch_db_engine:open_local_docs(Db, [Id]) of
[#doc{} = Doc] ->
- apply_open_options({ok, Doc}, Options);
+ case Doc#doc.body of
+ { Body } ->
+ Access = couch_util:get_value(<<"_access">>, Body),
+ apply_open_options(Db, {ok, Doc#doc{access = Access}}, Options);
+ _Else ->
+ apply_open_options(Db, {ok, Doc}, Options)
+ end;
[not_found] ->
{not_found, missing}
end;
-open_doc_int(Db, #doc_info{id = Id, revs = [RevInfo | _]} = DocInfo, Options) ->
+open_doc_int(Db, #doc_info{id = Id, revs = [RevInfo | _], access = Access} = DocInfo, Options) ->
#rev_info{deleted = IsDeleted, rev = {Pos, RevId}, body_sp = Bp} = RevInfo,
- Doc = make_doc(Db, Id, IsDeleted, Bp, {Pos, [RevId]}),
+ Doc = make_doc(Db, Id, IsDeleted, Bp, {Pos, [RevId], Access}),
apply_open_options(
- {ok, Doc#doc{meta = doc_meta_info(DocInfo, [], Options)}}, Options
+ {ok, Doc#doc{meta = doc_meta_info(DocInfo, [], Options)}}, Options, Access
);
-open_doc_int(Db, #full_doc_info{id = Id, rev_tree = RevTree} = FullDocInfo, Options) ->
+open_doc_int(Db, #full_doc_info{id = Id, rev_tree = RevTree, access = Access} = FullDocInfo, Options) ->
#doc_info{revs = [#rev_info{deleted = IsDeleted, rev = Rev, body_sp = Bp} | _]} =
DocInfo = couch_doc:to_doc_info(FullDocInfo),
{[{_, RevPath}], []} = couch_key_tree:get(RevTree, [Rev]),
- Doc = make_doc(Db, Id, IsDeleted, Bp, RevPath),
+ Doc = make_doc(Db, Id, IsDeleted, Bp, RevPath, Access),
apply_open_options(
- {ok, Doc#doc{meta = doc_meta_info(DocInfo, RevTree, Options)}}, Options
+ {ok, Doc#doc{meta = doc_meta_info(DocInfo, RevTree, Options)}}, Options, Access
);
open_doc_int(Db, Id, Options) ->
case get_full_doc_info(Db, Id) of
@@ -1952,21 +2108,26 @@ doc_meta_info(
true -> [{local_seq, Seq}]
end.
-make_doc(_Db, Id, Deleted, nil = _Bp, RevisionPath) ->
+make_doc(Db, Id, Deleted, Bp, {Pos, Revs}) ->
+ make_doc(Db, Id, Deleted, Bp, {Pos, Revs}, []).
+
+make_doc(_Db, Id, Deleted, nil = _Bp, RevisionPath, Access) ->
#doc{
id = Id,
revs = RevisionPath,
body = [],
atts = [],
- deleted = Deleted
+ deleted = Deleted,
+ access = Access
};
-make_doc(#db{} = Db, Id, Deleted, Bp, {Pos, Revs}) ->
+make_doc(#db{} = Db, Id, Deleted, Bp, {Pos, Revs}, Access) ->
RevsLimit = get_revs_limit(Db),
Doc0 = couch_db_engine:read_doc_body(Db, #doc{
id = Id,
revs = {Pos, lists:sublist(Revs, 1, RevsLimit)},
body = Bp,
- deleted = Deleted
+ deleted = Deleted,
+ access = Access
}),
Doc1 =
case Doc0#doc.atts of
diff --git a/src/couch/src/couch_db_int.hrl b/src/couch/src/couch_db_int.hrl
index 7da0ce5df..b67686fab 100644
--- a/src/couch/src/couch_db_int.hrl
+++ b/src/couch/src/couch_db_int.hrl
@@ -37,7 +37,8 @@
waiting_delayed_commit_deprecated,
options = [],
- compression
+ compression,
+ access = false
}).
diff --git a/src/couch/src/couch_db_updater.erl b/src/couch/src/couch_db_updater.erl
index 0248c21ec..52fec42f8 100644
--- a/src/couch/src/couch_db_updater.erl
+++ b/src/couch/src/couch_db_updater.erl
@@ -24,6 +24,11 @@
% 10 GiB
-define(DEFAULT_MAX_PARTITION_SIZE, 16#280000000).
+-define(DEFAULT_SECURITY_OBJECT, [
+ {<<"members">>,{[{<<"roles">>,[<<"_admin">>]}]}},
+ {<<"admins">>, {[{<<"roles">>,[<<"_admin">>]}]}}
+]).
+
-record(merge_acc, {
revs_limit,
merge_conflicts,
@@ -36,7 +41,7 @@
init({Engine, DbName, FilePath, Options0}) ->
erlang:put(io_priority, {db_update, DbName}),
update_idle_limit_from_config(),
- DefaultSecObj = default_security_object(DbName),
+ DefaultSecObj = default_security_object(DbName, Options0),
Options = [{default_security_object, DefaultSecObj} | Options0],
try
{ok, EngineState} = couch_db_engine:init(Engine, FilePath, Options),
@@ -165,7 +170,7 @@ handle_cast(Msg, #db{name = Name} = Db) ->
{stop, Msg, Db}.
handle_info(
- {update_docs, Client, GroupedDocs, NonRepDocs, MergeConflicts},
+ {update_docs, Client, GroupedDocs, NonRepDocs, MergeConflicts, UserCtx},
Db
) ->
GroupedDocs2 = sort_and_tag_grouped_docs(Client, GroupedDocs),
@@ -181,7 +186,7 @@ handle_info(
Clients = [Client]
end,
NonRepDocs2 = [{Client, NRDoc} || NRDoc <- NonRepDocs],
- try update_docs_int(Db, GroupedDocs3, NonRepDocs2, MergeConflicts) of
+ try update_docs_int(Db, GroupedDocs3, NonRepDocs2, MergeConflicts, UserCtx) of
{ok, Db2, UpdatedDDocIds} ->
ok = couch_server:db_updated(Db2),
case {couch_db:get_update_seq(Db), couch_db:get_update_seq(Db2)} of
@@ -260,7 +265,11 @@ sort_and_tag_grouped_docs(Client, GroupedDocs) ->
% The merge_updates function will fail and the database can end up with
% duplicate documents if the incoming groups are not sorted, so as a sanity
% check we sort them again here. See COUCHDB-2735.
- Cmp = fun([#doc{id = A} | _], [#doc{id = B} | _]) -> A < B end,
+ Cmp = fun
+ ([], []) -> false; % TODO: re-evaluate this addition, might be
+ % superflous now
+ ([#doc{id=A}|_], [#doc{id=B}|_]) -> A < B
+ end,
lists:map(
fun(DocGroup) ->
[{Client, maybe_tag_doc(D)} || D <- DocGroup]
@@ -320,6 +329,7 @@ init_db(DbName, FilePath, EngineState, Options) ->
BDU = couch_util:get_value(before_doc_update, Options, nil),
ADR = couch_util:get_value(after_doc_read, Options, nil),
+ Access = couch_util:get_value(access, Options, false),
NonCreateOpts = [Opt || Opt <- Options, Opt /= create],
InitDb = #db{
@@ -329,7 +339,8 @@ init_db(DbName, FilePath, EngineState, Options) ->
instance_start_time = StartTime,
options = NonCreateOpts,
before_doc_update = BDU,
- after_doc_read = ADR
+ after_doc_read = ADR,
+ access = Access
},
DbProps = couch_db_engine:get_props(InitDb),
@@ -390,7 +401,8 @@ flush_trees(
active = WrittenSize,
external = ExternalSize
},
- atts = AttSizeInfo
+ atts = AttSizeInfo,
+ access = NewDoc#doc.access
},
{Leaf, add_sizes(Type, Leaf, SizesAcc)};
#leaf{} ->
@@ -475,6 +487,9 @@ doc_tag(#doc{meta = Meta}) ->
Else -> throw({invalid_doc_tag, Else})
end.
+merge_rev_trees([[]], [], Acc) ->
+ % validate_docs_access left us with no docs to merge
+ {ok, Acc};
merge_rev_trees([], [], Acc) ->
{ok, Acc#merge_acc{
add_infos = lists:reverse(Acc#merge_acc.add_infos)
@@ -656,22 +671,29 @@ maybe_stem_full_doc_info(#full_doc_info{rev_tree = Tree} = Info, Limit) ->
Info
end.
-update_docs_int(Db, DocsList, LocalDocs, MergeConflicts) ->
+update_docs_int(Db, DocsList, LocalDocs, MergeConflicts, UserCtx) ->
UpdateSeq = couch_db_engine:get_update_seq(Db),
RevsLimit = couch_db_engine:get_revs_limit(Db),
- Ids = [Id || [{_Client, #doc{id = Id}} | _] <- DocsList],
+ Ids = [Id || [{_Client, #doc{id=Id}}|_] <- DocsList],
+ % TODO: maybe a perf hit, instead of zip3-ing existing Accesses into
+ % our doc lists, maybe find 404 docs differently down in
+ % validate_docs_access (revs is [], which we can then use
+ % to skip validation as we know it is the first doc rev)
+ Accesses = [Access || [{_Client, #doc{access=Access}}|_] <- DocsList],
+
% lookup up the old documents, if they exist.
OldDocLookups = couch_db_engine:open_docs(Db, Ids),
- OldDocInfos = lists:zipwith(
+ OldDocInfos = lists:zipwith3(
fun
- (_Id, #full_doc_info{} = FDI) ->
+ (_Id, #full_doc_info{} = FDI, _Access) ->
FDI;
- (Id, not_found) ->
- #full_doc_info{id = Id}
+ (Id, not_found, Access) ->
+ #full_doc_info{id=Id,access=Access}
end,
Ids,
- OldDocLookups
+ OldDocLookups,
+ Accesses
),
%% Get the list of full partitions
@@ -708,7 +730,14 @@ update_docs_int(Db, DocsList, LocalDocs, MergeConflicts) ->
cur_seq = UpdateSeq,
full_partitions = FullPartitions
},
- {ok, AccOut} = merge_rev_trees(DocsList, OldDocInfos, AccIn),
+ % Loop over DocsList, validate_access for each OldDocInfo on Db,
+ %. if no OldDocInfo, then send to DocsListValidated, keep OldDocsInfo
+ % if valid, then send to DocsListValidated, OldDocsInfo
+ %. if invalid, then send_result tagged `access`(c.f. `conflict)
+ %. and don’t add to DLV, nor ODI
+
+ { DocsListValidated, OldDocInfosValidated } = validate_docs_access(Db, UserCtx, DocsList, OldDocInfos),
+ {ok, AccOut} = merge_rev_trees(DocsListValidated, OldDocInfosValidated, AccIn),
#merge_acc{
add_infos = NewFullDocInfos,
rem_seqs = RemSeqs
@@ -718,7 +747,8 @@ update_docs_int(Db, DocsList, LocalDocs, MergeConflicts) ->
% the trees, the attachments are already written to disk)
{ok, IndexFDIs} = flush_trees(Db, NewFullDocInfos, []),
Pairs = pair_write_info(OldDocLookups, IndexFDIs),
- LocalDocs2 = update_local_doc_revs(LocalDocs),
+ LocalDocs1 = apply_local_docs_access(Db, LocalDocs),
+ LocalDocs2 = update_local_doc_revs(LocalDocs1),
{ok, Db1} = couch_db_engine:write_doc_infos(Db, Pairs, LocalDocs2),
@@ -733,18 +763,87 @@ update_docs_int(Db, DocsList, LocalDocs, MergeConflicts) ->
length(LocalDocs2)
),
- % Check if we just updated any design documents, and update the validation
- % funs if we did.
+ % Check if we just updated any non-access design documents,
+ % and update the validation funs if we did.
+ NonAccessIds = [Id || [{_Client, #doc{id=Id,access=[]}}|_] <- DocsList],
UpdatedDDocIds = lists:flatmap(
fun
(<<"_design/", _/binary>> = Id) -> [Id];
(_) -> []
end,
- Ids
+ NonAccessIds
),
{ok, commit_data(Db1), UpdatedDDocIds}.
+% check_access(Db, UserCtx, Access) ->
+% check_access(Db, UserCtx, couch_db:has_access_enabled(Db), Access).
+%
+% check_access(_Db, UserCtx, false, _Access) ->
+% true;
+
+% at this point, we already validated this Db is access enabled, so do the checks right away.
+check_access(Db, UserCtx, Access) -> couch_db:check_access(Db#db{user_ctx=UserCtx}, Access).
+
+% TODO: looks like we go into validation here unconditionally and only check in
+% check_access() whether the Db has_access_enabled(), we should do this
+% here on the outside. Might be our perf issue.
+% However, if it is, that means we have to speed this up as it would still
+% be too slow for when access is enabled.
+validate_docs_access(Db, UserCtx, DocsList, OldDocInfos) ->
+ case couch_db:has_access_enabled(Db) of
+ true -> validate_docs_access_int(Db, UserCtx, DocsList, OldDocInfos);
+ _Else -> { DocsList, OldDocInfos }
+ end.
+
+validate_docs_access_int(Db, UserCtx, DocsList, OldDocInfos) ->
+ validate_docs_access(Db, UserCtx, DocsList, OldDocInfos, [], []).
+
+validate_docs_access(_Db, UserCtx, [], [], DocsListValidated, OldDocInfosValidated) ->
+ { lists:reverse(DocsListValidated), lists:reverse(OldDocInfosValidated) };
+validate_docs_access(Db, UserCtx, [Docs | DocRest], [OldInfo | OldInfoRest], DocsListValidated, OldDocInfosValidated) ->
+ % loop over Docs as {Client, NewDoc}
+ % validate Doc
+ % if valid, then put back in Docs
+ % if not, then send_result and skip
+ NewDocs = lists:foldl(fun({ Client, Doc }, Acc) ->
+ % check if we are allowed to update the doc, skip when new doc
+ OldDocMatchesAccess = case OldInfo#full_doc_info.rev_tree of
+ [] -> true;
+ _ -> check_access(Db, UserCtx, OldInfo#full_doc_info.access)
+ end,
+
+ NewDocMatchesAccess = check_access(Db, UserCtx, Doc#doc.access),
+ case OldDocMatchesAccess andalso NewDocMatchesAccess of
+ true -> % if valid, then send to DocsListValidated, OldDocsInfo
+ % and store the access context on the new doc
+ [{Client, Doc} | Acc];
+ _Else2 -> % if invalid, then send_result tagged `access`(c.f. `conflict)
+ % and don’t add to DLV, nor ODI
+ send_result(Client, Doc, access),
+ Acc
+ end
+ end, [], Docs),
+
+ { NewDocsListValidated, NewOldDocInfosValidated } = case length(NewDocs) of
+ 0 -> % we sent out all docs as invalid access, drop the old doc info associated with it
+ { [NewDocs | DocsListValidated], OldDocInfosValidated };
+ _ ->
+ { [NewDocs | DocsListValidated], [OldInfo | OldDocInfosValidated] }
+ end,
+ validate_docs_access(Db, UserCtx, DocRest, OldInfoRest, NewDocsListValidated, NewOldDocInfosValidated).
+
+apply_local_docs_access(Db, Docs) ->
+ apply_local_docs_access1(couch_db:has_access_enabled(Db), Docs).
+
+apply_local_docs_access1(false, Docs) ->
+ Docs;
+apply_local_docs_access1(true, Docs) ->
+ lists:map(fun({Client, #doc{access = Access, body = {Body}} = Doc}) ->
+ Doc1 = Doc#doc{body = {[{<<"_access">>, Access} | Body]}},
+ {Client, Doc1}
+ end, Docs).
+
update_local_doc_revs(Docs) ->
lists:foldl(
fun({Client, Doc}, Acc) ->
@@ -761,6 +860,14 @@ update_local_doc_revs(Docs) ->
Docs
).
+default_security_object(DbName, []) ->
+ default_security_object(DbName);
+default_security_object(DbName, Options) ->
+ case lists:member({access, true}, Options) of
+ false -> default_security_object(DbName);
+ true -> ?DEFAULT_SECURITY_OBJECT
+ end.
+
increment_local_doc_revs(#doc{deleted = true} = Doc) ->
{ok, Doc#doc{revs = {0, [0]}}};
increment_local_doc_revs(#doc{revs = {0, []}} = Doc) ->
@@ -925,21 +1032,14 @@ get_meta_body_size(Meta) ->
default_security_object(<<"shards/", _/binary>>) ->
case config:get("couchdb", "default_security", "admin_only") of
- "admin_only" ->
- [
- {<<"members">>, {[{<<"roles">>, [<<"_admin">>]}]}},
- {<<"admins">>, {[{<<"roles">>, [<<"_admin">>]}]}}
- ];
+ "admin_only" -> ?DEFAULT_SECURITY_OBJECT;
Everyone when Everyone == "everyone"; Everyone == "admin_local" ->
[]
end;
default_security_object(_DbName) ->
case config:get("couchdb", "default_security", "admin_only") of
Admin when Admin == "admin_only"; Admin == "admin_local" ->
- [
- {<<"members">>, {[{<<"roles">>, [<<"_admin">>]}]}},
- {<<"admins">>, {[{<<"roles">>, [<<"_admin">>]}]}}
- ];
+ ?DEFAULT_SECURITY_OBJECT;
"everyone" ->
[]
end.