MB-61292: Fix drop DEKs behavior when force encryption...

... has been used before

This commit fixes the following encr at rest scenario:
1. Create a key
2. Create an encrypted bucket
3. Call POST /controller/forceEncryptionAtRest/bucket/<B>
4. Disable encryption on the bucket
5. Call POST /controller/dropEncryptionAtRestDeks/bucket/<B>
6. Enable encryption on the bucket again
7. Call POST /controller/dropEncryptionAtRestDeks/bucket/<B>

Step #7 should trigger DEK drop and re-encryption of the data,
but the reencryption doesn't happen.
This happens because of two reasons:
- drop_deks (via compaction) is not needed in this case,
  because old DEK becomes "not in use" without compaction
  (it doesn't really encrypt anything);
- force_encryption_datetime is set to some old value, so it is
  ignored when the list of expired keys is checked

In this case we should not really ignore force_encryption_datetime,
because dek_drop_datetime is actually set, so this time should be
used instead.

In other words, currently we only use dek_drop_datetime as
force_encryption_datetime if force_encryption_datetime is not set.
While we should actually take the maximum of these two values.

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

Author: timofey-barmin

apps/ns_server/src/cb_crypto.erl

@@ -881,26 +881,100 @@ get_drop_keys_timestamp(Type, Snapshot) ->
 get_force_encryption_timestamp(Type, Snapshot) ->
     maybe
         Settings = menelaus_web_encr_at_rest:get_settings(Snapshot),
-        TypeSettings = maps:get(Type, Settings, undefined),
-        {ok, ForceDT} ?=
-            case TypeSettings of
-                undefined ->
-                    {error, not_found};
-                #{force_encryption_datetime := {set, DT}} ->
-                    {ok, DT};
-                #{force_encryption_datetime := {not_set, _}} ->
-                    get_drop_keys_timestamp(Type, Snapshot)
+        {ok, TypeSettings} ?= case maps:find(Type, Settings) of
+                                  error -> {error, not_found};
+                                  {ok, S} -> {ok, S}
+                              end,
+
+        ForceDT = maps:get(force_encryption_datetime, TypeSettings,
+                           {not_set, []}),
+        DropDT = maps:get(dek_drop_datetime, TypeSettings,
+                          {not_set, []}),
+        EffectiveForceDT =
+            case {ForceDT, DropDT} of
+                {{not_set, _}, {not_set, _}} -> undefined;
+                {{not_set, []}, {set, DT}} -> DT;
+                {{set, DT}, {not_set, []}} -> DT;
+                {{set, DT1}, {set, DT2}} -> max(DT1, DT2)
             end,
         LastToggleDT = maps:get(encryption_last_toggle_datetime, TypeSettings,
                                 undefined),
         if
-            ForceDT =:= undefined -> {ok, undefined};
-            LastToggleDT =:= undefined -> {ok, ForceDT};
-            LastToggleDT > ForceDT -> {ok, undefined};
-            true -> {ok, ForceDT}
+            EffectiveForceDT =:= undefined -> {ok, undefined};
+            LastToggleDT =:= undefined -> {ok, EffectiveForceDT};
+            LastToggleDT > EffectiveForceDT -> {ok, undefined};
+            true -> {ok, EffectiveForceDT}
         end
     end.
 
+-ifdef(TEST).
+
+get_force_encryption_timestamp_test() ->
+    NewDate = {{2025,8,21},{23,28,56}},
+    OldDate = {{2025,8,21},{23,27,29}},
+    SuperOldDate = {{2025,8,21},{23,26,29}},
+    meck:new(cluster_compat_mode, [passthrough]),
+    try
+        meck:expect(cluster_compat_mode, is_cluster_79,
+                    fun () -> true end),
+        meck:expect(cluster_compat_mode, is_enterprise,
+                    fun () -> true end),
+        Mock = fun (ForceDT, DropDT, LastToggleDT) ->
+                  meck:expect(
+                    menelaus_web_encr_at_rest, get_settings,
+                    fun (_Snapshot) ->
+                        C1 = case LastToggleDT of
+                                 undefined -> #{};
+                                 _ ->
+                                    #{encryption_last_toggle_datetime =>
+                                          LastToggleDT}
+                             end,
+                        C2 = #{force_encryption_datetime => ForceDT,
+                               dek_drop_datetime => DropDT},
+                        #{config_encryption => maps:merge(C1, C2)}
+                    end)
+              end,
+
+        Assert = fun (Expected, Force, Drop, LastToggle) ->
+                     Val = fun (undefined) -> {not_set, []};
+                               (V) -> {set, V}
+                           end,
+                     Mock(Val(Force), Val(Drop), LastToggle),
+                     ?assertEqual({error, not_found},
+                                  get_force_encryption_timestamp(unknown, #{})),
+                     ?assertEqual({ok, Expected},
+                                  get_force_encryption_timestamp(
+                                    config_encryption, #{}))
+                 end,
+
+        Assert(NewDate, NewDate, OldDate, SuperOldDate),
+        Assert(NewDate, OldDate, NewDate, SuperOldDate),
+        Assert(NewDate, undefined, NewDate, SuperOldDate),
+        Assert(NewDate, NewDate, undefined, SuperOldDate),
+        Assert(undefined, undefined, undefined, SuperOldDate),
+
+        Assert(undefined, OldDate, SuperOldDate, NewDate),
+        Assert(undefined, SuperOldDate, OldDate, NewDate),
+        Assert(undefined, undefined, OldDate, NewDate),
+        Assert(undefined, SuperOldDate, undefined, NewDate),
+        Assert(undefined, undefined, undefined, NewDate),
+
+        Assert(NewDate, SuperOldDate, NewDate, OldDate),
+        Assert(NewDate, NewDate, SuperOldDate, OldDate),
+        Assert(undefined, undefined, SuperOldDate, OldDate),
+        Assert(NewDate, NewDate, undefined, OldDate),
+
+        Assert(NewDate, NewDate, OldDate, undefined),
+        Assert(NewDate, OldDate, NewDate, undefined),
+        Assert(NewDate, undefined, NewDate, undefined),
+        Assert(NewDate, NewDate, undefined, undefined),
+        Assert(undefined, undefined, undefined, undefined)
+
+    after
+        meck:unload(cluster_compat_mode)
+    end.
+-endif.
+
 %%%===================================================================
 %%% Internal functions
 %%%===================================================================

apps/ns_server/src/ns_bucket.erl

@@ -3280,20 +3280,110 @@ get_force_encryption_timestamp(BucketUUID, Snapshot) ->
         Settings = chronicle_compat:get(Snapshot, sub_key(Bucket, encr_at_rest),
                                         #{default => #{}}),
         DropDT = maps:get(dek_drop_datetime, Settings, undefined),
-        ForceDT = maps:get(force_encryption_datetime, Settings, DropDT),
+        ForceDT = maps:get(force_encryption_datetime, Settings, undefined),
+
+        EffectiveForceDT = case {DropDT, ForceDT} of
+                                {undefined, undefined} -> undefined;
+                                {undefined, DT} -> DT;
+                                {DT, undefined} -> DT;
+                                {DT1, DT2} -> max(DT1, DT2)
+                            end,
+
         LastToggleDT = proplists:get_value(encryption_last_toggle_datetime,
                                            Config),
         if
-            ForceDT =:= undefined -> {ok, undefined};
-            LastToggleDT =:= undefined -> {ok, ForceDT};
-            LastToggleDT > ForceDT -> {ok, undefined};
-            true -> {ok, ForceDT}
+            EffectiveForceDT =:= undefined -> {ok, undefined};
+            LastToggleDT =:= undefined -> {ok, EffectiveForceDT};
+            LastToggleDT > EffectiveForceDT -> {ok, undefined};
+            true -> {ok, EffectiveForceDT}
         end
     else
         not_present -> {error, not_found};
         {error, R} -> {error, R}
     end.
 
+-ifdef(TEST).
+get_force_encryption_timestamp_test() ->
+    meck:new(cluster_compat_mode, [passthrough]),
+    try
+        meck:expect(cluster_compat_mode, is_cluster_79,
+                    fun () -> true end),
+        meck:expect(cluster_compat_mode, is_enterprise,
+                    fun () -> true end),
+
+        NewDate = {{2025,8,21},{23,28,56}},
+        OldDate = {{2025,8,21},{23,27,29}},
+        SuperOldDate = {{2025,8,21},{23,26,29}},
+        B = <<"test_bucket">>,
+        BUUID = <<"123">>,
+
+        BucketCommon = #{root => {[B], 0},
+                        uuid2bucket_key(BUUID) => {B, 0},
+                        uuid_key(B) => {BUUID, 0}},
+
+        SetToggleDT =
+            fun (S, undefined) ->
+                    S#{sub_key(B, props) => {[], 0}};
+                (S, DT) ->
+                    S#{sub_key(B, props) =>
+                        {[{encryption_last_toggle_datetime, DT}], 0}}
+            end,
+
+        SetOtherDT =
+            fun (S, undefined, undefined) -> S;
+                (S, ForceDT, undefined) ->
+                    S#{sub_key(B, encr_at_rest) =>
+                        {#{force_encryption_datetime => ForceDT}, 0}};
+                (S, undefined, DropDT) ->
+                    S#{sub_key(B, encr_at_rest) =>
+                        {#{dek_drop_datetime => DropDT}, 0}};
+                (S, ForceDT, DropDT) ->
+                    S#{sub_key(B, encr_at_rest) =>
+                        {#{dek_drop_datetime => DropDT,
+                            force_encryption_datetime => ForceDT}, 0}}
+            end,
+
+        Snapshot = fun (ForceDT, DropDT, LastToggleDT) ->
+                    functools:chain(
+                        BucketCommon,
+                        [SetToggleDT(_, LastToggleDT),
+                        SetOtherDT(_, ForceDT, DropDT)])
+                end,
+
+        Assert = fun (Expected, ForceDT, DropDT, LastToggleDT) ->
+                     ?assertEqual({ok, Expected},
+                                  get_force_encryption_timestamp(
+                                    BUUID, Snapshot(ForceDT, DropDT, LastToggleDT)))
+                 end,
+
+        Assert(NewDate, NewDate, OldDate, SuperOldDate),
+        Assert(NewDate, OldDate, NewDate, SuperOldDate),
+        Assert(NewDate, undefined, NewDate, SuperOldDate),
+        Assert(NewDate, NewDate, undefined, SuperOldDate),
+        Assert(undefined, undefined, undefined, SuperOldDate),
+
+        Assert(undefined, OldDate, SuperOldDate, NewDate),
+        Assert(undefined, SuperOldDate, OldDate, NewDate),
+        Assert(undefined, undefined, OldDate, NewDate),
+        Assert(undefined, SuperOldDate, undefined, NewDate),
+        Assert(undefined, undefined, undefined, NewDate),
+
+        Assert(NewDate, SuperOldDate, NewDate, OldDate),
+        Assert(NewDate, NewDate, SuperOldDate, OldDate),
+        Assert(undefined, undefined, SuperOldDate, OldDate),
+        Assert(NewDate, NewDate, undefined, OldDate),
+
+        Assert(NewDate, NewDate, OldDate, undefined),
+        Assert(NewDate, OldDate, NewDate, undefined),
+        Assert(NewDate, undefined, NewDate, undefined),
+        Assert(NewDate, NewDate, undefined, undefined),
+        Assert(undefined, undefined, undefined, undefined)
+    after
+        meck:unload(cluster_compat_mode)
+    end.
+
+-endif.
+
 %% fusion
 
 -spec is_fusion(proplists:proplist()) -> boolean().