MB-61292: Pass deks to memcached

The create bucket command should contain all deks for that bucket,
and its current active key id. If the active key is empty, it means
that encryption is disabled for that bucket.
Every time new key is generated for a bucket ns_server should push
it to memcached using new command (set_encryption_key).
That command informs memcached about new dek and sets that dek as
active.

Change-Id: I1d193e05e5258314017088eb818332fc3dcefab6
Reviewed-on: https://review.couchbase.org/c/ns_server/+/211628
Well-Formed: Build Bot <[email protected]>
Reviewed-by: Navdeep S Boparai <[email protected]>
Tested-by: Build Bot <[email protected]>
Tested-by: Timofey Barmin <[email protected]>

Author: timofey-barmin

apps/ns_server/include/mc_constants.hrl

@@ -124,6 +124,8 @@
 -define(CMD_RBAC_REFRESH, 16#f7).
 -define(CMD_GET_ERROR_MAP, 16#fe).
 
+-define(CMD_SET_ENCRYPTION_KEY, 16#2d).
+
 -define(RGET,        16#30).
 -define(RSET,        16#31).
 -define(RSETQ,       16#32).

apps/ns_server/src/cb_deks.erl

@@ -265,7 +265,12 @@ maybe_reencrypt_deks(Kind, Deks, NewEncryptionKeyFun) ->
 %% Returns false otherwise.
 dek_chronicle_keys_filter(?CHRONICLE_ENCR_AT_REST_SETTINGS_KEY) ->
     [chronicleDek];
-dek_chronicle_keys_filter(_Key) -> false.
+dek_chronicle_keys_filter(Key) ->
+    case ns_bucket:sub_key_match(Key) of
+        {true, Bucket, props} -> [{bucketDek, Bucket}];
+        {true, _Bucket, _} -> false;
+        false -> false
+    end.
 
 %% encryption_method_callback - called to determine if encryption is enabled
 %% or not for that type of entity.
@@ -290,13 +295,21 @@ dek_config(chronicleDek) ->
       encryption_method_callback => fun chronicle_local:get_encryption/1,
       set_active_key_callback => fun chronicle_local:set_active_dek/1,
       chronicle_txn_keys => [?CHRONICLE_ENCR_AT_REST_SETTINGS_KEY],
-      required_usage => config_encryption}.
+      required_usage => config_encryption};
+dek_config({bucketDek, Bucket}) ->
+    #{name => {bucket, Bucket},
+      encryption_method_callback => ns_bucket:get_encryption(Bucket, _),
+      set_active_key_callback => ns_memcached:set_active_dek(Bucket, _),
+      chronicle_txn_keys => [ns_bucket:root(),
+                             ns_bucket:sub_key(Bucket, props)],
+      required_usage => {bucket_encryption, Bucket}}.
 
 %% Returns all possible deks kinds on the node.
 %% The list was supposed to be static if not buckets. Buckets can be created and
 %% removed in real time, so the list is dynamic because of that. Note that the
 %% list doesn't depend if encryption is on or off.
 dek_kinds_list() ->
     dek_kinds_list(direct).
-dek_kinds_list(_Snapshot) ->
-    [chronicleDek].
+dek_kinds_list(Snapshot) ->
+    Buckets = ns_bucket:get_bucket_names(Snapshot),
+    [chronicleDek] ++ [{bucketDek, B} || B <- Buckets].

apps/ns_server/src/mc_client_binary.erl

@@ -70,7 +70,8 @@
          update_user_permissions/2,
          set_collections_manifest/2,
          get_collections_manifest/1,
-         set_tls_config/2
+         set_tls_config/2,
+         set_active_encryption_key/3
         ]).
 
 -type recv_callback() :: fun((_, _, _) -> any()) | undefined.
@@ -1029,3 +1030,22 @@ set_tls_config(Sock, TLSConfigJSON) ->
         Response ->
             process_error_response(Response)
     end.
+
+set_active_encryption_key(Sock, Bucket, ActiveKey) ->
+    report_counter(?FUNCTION_NAME),
+    Key = iolist_to_binary(Bucket),
+    %% TODO: We should pass all available keys here
+    AllKeys = case ActiveKey of
+                  undefined -> [];
+                  _ -> [ActiveKey]
+              end,
+    Value = memcached_bucket_config:format_mcd_keys(ActiveKey, AllKeys),
+    Entry = #mc_entry{key = Key,
+                      data = ejson:encode(Value)},
+    case cmd(?CMD_SET_ENCRYPTION_KEY, Sock, undefined, undefined,
+             {#mc_header{}, Entry}) of
+        {ok, #mc_header{status=?SUCCESS}, _, _} ->
+            ok;
+        Response ->
+            process_error_response(Response)
+    end.

apps/ns_server/src/memcached_bucket_config.erl

@@ -15,15 +15,19 @@
 -include("ns_common.hrl").
 -include("ns_bucket.hrl").
 -include_lib("ns_common/include/cut.hrl").
+-include("cb_cluster_secrets.hrl").
+
+-define(MCD_DISABLED_ENCRYPTION_KEY_ID, <<"">>).
 
 -record(cfg, {type, name, config, snapshot, engine_config, params}).
 
 -export([get/1,
          get_bucket_config/1,
          ensure/2,
-         start_params/1,
+         start_params/3,
          ensure_collections/2,
-         get_current_collections_uid/1]).
+         get_current_collections_uid/1,
+         format_mcd_keys/2]).
 
 params(membase, BucketName, BucketConfig, MemQuota, UUID) ->
     {DriftAheadThreshold, DriftBehindThreshold} =
@@ -301,6 +305,18 @@ ensure(Sock, #cfg{type = memcached}) ->
                       end, not_present),
     ok.
 
+format_mcd_keys(ActiveDek, Deks) ->
+    DeksJsonMcd = lists:map(fun format_mcd_key/1, Deks),
+    ActiveKeyMcd = case ActiveDek of
+                       undefined -> ?MCD_DISABLED_ENCRYPTION_KEY_ID;
+                       #{id := ActiveId} -> ActiveId
+                   end,
+    {[{keys, DeksJsonMcd}, {active, ActiveKeyMcd}]}.
+
+format_mcd_key(#{id := Id, type := 'raw-aes-gcm', info := #{key := KeyFun}}) ->
+    Encoded = base64:encode(KeyFun()),
+    {[{id, Id}, {cipher, <<"AES-256-GCM">>}, {key, Encoded}]}.
+
 get_current_collections_uid(Sock) ->
     case mc_client_binary:get_collections_manifest(Sock) of
         {memcached_error, no_coll_manifest, _} ->
@@ -343,7 +359,7 @@ ensure_collections(Sock, #cfg{name = BucketName, snapshot = Snapshot}) ->
 
 start_params(#cfg{config = BucketConfig,
                   params = Params,
-                  engine_config = EngineConfig}) ->
+                  engine_config = EngineConfig}, ActiveDek, Deks) ->
     Engine = proplists:get_value(engine, EngineConfig),
 
     StaticConfigString =
@@ -368,7 +384,13 @@ start_params(#cfg{config = BucketConfig,
                   end
           end, Params),
 
-    ExtraParams = [P || P <- [StaticConfigString, ExtraConfigString], P =/= ""],
+    EncodedDeks = binary_to_list(ejson:encode(format_mcd_keys(ActiveDek,
+                                                              Deks))),
+
+    DeksConfigString = "encryption=" ++ EncodedDeks,
+
+    ExtraParams = [P || P <- [StaticConfigString, ExtraConfigString,
+                              DeksConfigString], P =/= ""],
     {Engine, string:join(DynamicParams ++ ExtraParams, ";")}.
 
 get_bucket_config(#cfg{config = BucketConfig}) ->

apps/ns_server/src/ns_bucket.erl

@@ -192,7 +192,8 @@
          get_commits_from_snapshot/2,
          update_bucket_config/2,
          update_buckets_config/1,
-         all_keys/1]).
+         all_keys/1,
+         get_encryption/2]).
 
 -import(json_builder,
         [to_binary/1,
@@ -2517,6 +2518,22 @@ validate_encryption_secret(SecretId, Bucket, Snapshot) ->
         {error, not_allowed} -> {error, secret_not_allowed}
     end.
 
+get_encryption(BucketName, Snapshot) ->
+    case get_bucket(BucketName, Snapshot) of
+        {ok, BucketConfig} ->
+            case proplists:get_value(encryption_secret_id, BucketConfig,
+                                     ?SECRET_ID_NOT_SET) of
+                ?SECRET_ID_NOT_SET -> {ok, disabled};
+                Id ->
+                    case lists:member(node(), get_servers(BucketConfig)) of
+                        true -> {ok, {secret, Id}};
+                        false -> {error, not_found}
+                    end
+            end;
+        not_present ->
+            {error, not_found}
+    end.
+
 -ifdef(TEST).
 min_live_copies_test() ->
     ?assertEqual(min_live_copies([node1], []), undefined),

apps/ns_server/src/ns_memcached.erl

@@ -28,6 +28,7 @@
 -include("ns_common.hrl").
 -include("rbac.hrl").
 -include_lib("ns_common/include/cut.hrl").
+-include("cb_cluster_secrets.hrl").
 
 -define(CHECK_INTERVAL, 10000).
 -define(CHECK_WARMUP_INTERVAL, 500).
@@ -42,6 +43,7 @@
 -define(GET_KEYS_TIMEOUT,       ?get_timeout(get_keys, 60000)).
 -define(GET_KEYS_OUTER_TIMEOUT, ?get_timeout(get_keys_outer, 70000)).
 -define(MAGMA_CREATION_TIMEOUT, ?get_timeout(magma_creation, 300000)).
+-define(DEKS_TIMEOUT,           ?get_timeout(deks, 60000)).
 
 -define(RECBUF, ?get_param(recbuf, 64 * 1024)).
 -define(SNDBUF, ?get_param(sndbuf, 64 * 1024)).
@@ -132,7 +134,8 @@
          get_collections_uid/1,
          maybe_add_impersonate_user_frame_info/2,
          delete_bucket/2,
-         get_config_stats/2
+         get_config_stats/2,
+         set_active_dek/2
         ]).
 
 %% for ns_memcached_sockets_pool, memcached_file_refresh only
@@ -149,6 +152,11 @@
 %%
 
 start_link(Bucket) ->
+    %% Sync with node monitor to make sure the key is created by the time
+    %% when ns_memcached is started. Note that ns_memcached can't call
+    %% cb_cluster_secrets, because cb_cluster_secrets calls ns_memcached, so
+    %% deadlock is possible.
+    cb_cluster_secrets:sync_with_node_monitor(),
     gen_server:start_link({local, server(Bucket)}, ?MODULE, Bucket, []).
 
 
@@ -822,7 +830,6 @@ handle_info({connect_done, WorkersCount, RV}, #state{bucket = Bucket,
     gen_event:notify(buckets_events, {started, Bucket}),
     erlang:process_flag(trap_exit, true),
     Self = self(),
-
     case RV of
         {ok, Sock} ->
             try ensure_bucket(Sock, Bucket, false) of
@@ -942,6 +949,7 @@ handle_info(Message, #state{worker_features = WF, control_queue = Q,
   when Message =:= check_config_soon orelse Message =:= check_config ->
     misc:flush(check_config_soon),
     misc:flush(check_config),
+
     case get_worker_features() of
         WF ->
             Self = self(),
@@ -1568,9 +1576,19 @@ do_ensure_bucket(Sock, Bucket, BConf, false) ->
             {ok, DBSubDir} =
                 ns_storage_conf:this_node_bucket_dbdir(Bucket),
             ok = filelib:ensure_dir(DBSubDir),
-
+            {ok, {ActiveDekId, DekIds, IsEnabled}} =
+                cb_deks:list({bucketDek, Bucket}),
+            {ok, Deks} = cb_deks:read({bucketDek, Bucket}, DekIds),
+            {value, ActiveDek} =
+                case IsEnabled of
+                    true ->
+                        lists:search(fun (#{id := Id}) -> Id == ActiveDekId end,
+                                     Deks);
+                    _ ->
+                        {value, undefined}
+                end,
             {Engine, ConfigString} =
-                memcached_bucket_config:start_params(BConf),
+                memcached_bucket_config:start_params(BConf, ActiveDek, Deks),
 
             BucketConfig = memcached_bucket_config:get_bucket_config(BConf),
             Timeout = case ns_bucket:node_kv_backend_type(BucketConfig) of
@@ -1845,6 +1863,29 @@ set_tls_config(Config) ->
           end
       end).
 
+set_active_dek(TypeOrBucket, ActiveDek) ->
+    ?log_debug("Setting active encryption key id for ~p: ~p...",
+               [TypeOrBucket, maps:get(id, ActiveDek)]),
+
+    RV = perform_very_long_call(
+           fun (Sock) ->
+               case mc_client_binary:set_active_encryption_key(Sock,
+                                                               TypeOrBucket,
+                                                               ActiveDek) of
+                   ok -> {reply, ok};
+                   {memcached_error, S, Msg} ->
+                       ?log_error("Setting encryption key for ~p failed: ~p",
+                                  [TypeOrBucket, {S, Msg}]),
+                       {reply, {error, {S, Msg}}}
+               end
+           end),
+
+    case RV of
+        ok -> ok;
+        {error, couldnt_connect_to_memcached} -> {error, retry};
+        {error, E} -> {error, E}
+    end.
+
 get_bucket_stats(RootKey, StatKey, SubKey) ->
     perform_very_long_call(
       fun(Sock) ->