[BACKPORT 2.18][#21057] YCQL: Fix cassandra user regeneration on dele…

…tion

Summary:
Original commit: 53afa97 / D32558
Customer wanted to delete the cassandra role in YCQLsh for specific use case of using their own LDAP server and did not want to share the cassandra role among different clusters as they were using a single server.

But once the cassandra role has been dropped (through another superuser) in YCQLsh, if the cluster is restarted, the role gets regenerated again with the **default** password, as the code currently only scans if the role exists or not, and if it doesn't, it creates the default role.

**Solution**
To solve this, added a boolean flag in SysSecurityConfigEntryPB called cassandra_user_created to mark if the role has been created in the past. If that is true, we don't create the role anymore.

**Limitation**
For upgrading a cluster, this change would need a restart for the changes to come into effect after the autoflag is promoted as the cassandra role is checked at cluster startup time. After promoting autoflag and restarting again, it will be possible to delete the cassandra role without it being regenerated.

**Upgrade/Rollback safety:**
The change is protected by the autoflag **ycql_allow_cassandra_drop**, for upgrading the cluster user would need to use yb-admin promote_auto_flags and restart after changing the version so that new change comes into effect.
Jira: DB-10024

Test Plan:
**./yb_build.sh release --java-test 'org.yb.cql.TestRoles'**

Manual Testing
Tested cases involving
1. Creating a new cluster (Autoflag true by default)
2. Restarting a new cluster (Autoflag true by default)
3. Restarting a new cluster with deleted cassandra role (Autoflag true by default)
4. Upgrading an existing cluster (Upgrade, promote autoflag, restart again)
5. Upgrading an existing cluster with deleted cassandra role (Upgrade, promote autoflag, restart again)

Reviewers: skumar, loginov, stiwary

Reviewed By: skumar

Subscribers: bogdan, yql, ybase

Tags: #jenkins-ready

Differential Revision: https://phorge.dev.yugabyte.com/D34014

java/yb-cql/src/test/java/org/yb/cql/BaseCQLTest.java

@@ -275,6 +275,12 @@ protected void restartClusterWithTSFlags(Map<String, String> tserverFlags) throw
     setUpCqlClient();
   }
 
+  protected void restartClusterWithMasterFlags(Map<String, String> masterFlags) throws Exception {
+    destroyMiniCluster();
+    createMiniCluster(masterFlags, Collections.emptyMap());
+    setUpCqlClient();
+  }
+
   protected void restartClusterWithFlag(String flag, String value) throws Exception {
     restartClusterWithTSFlags(Collections.singletonMap(flag, value));
   }

java/yb-cql/src/test/java/org/yb/cql/TestRoles.java

@@ -27,6 +27,9 @@
 import java.util.ArrayList;
 import java.util.Arrays;
 import java.util.List;
+import java.util.Map;
+import java.util.HashMap;
+
 
 @RunWith(value=YBTestRunner.class)
 public class TestRoles extends BaseAuthenticationCQLTest {
@@ -632,4 +635,38 @@ public void testDescribePermissions() throws Exception {
     session.execute(RevokePermissionAllRolesStmt(ALL, roles.get(6)));
     assertPermissionsGranted(session, roles.get(6), canonicalResource, new ArrayList<>());
   }
+
+  @Test
+  public void testCassandraUserRecreationDisabledOnRestart() throws Exception {
+    String cassandra_user = "cassandra";
+    String superuser2 = "superuser2";
+    String password = "password";
+
+    // Create a new 'superuser2' role (as 'cassandra')
+    createRole(session, superuser2, password, true, true, true);
+
+    // Connect as 'superuser2'.
+    try (ClusterAndSession cs2 = connectWithCredentials(superuser2, password)) {
+    // Verify that second_admin can delete 'cassandra'.
+      cs2.execute(String.format("DROP ROLE %s", cassandra_user));
+    }
+
+    // Verify that we can't connect using the deleted 'cassandra' role.
+    checkConnectivity(true, cassandra_user, cassandra_user, true);
+
+    // Restart the cluster
+    restartYcqlMiniCluster();
+
+    // Verify again after restart that we can't connect using the deleted 'cassandra' role.
+    checkConnectivity(true, cassandra_user, cassandra_user, true);
+
+    // Restart cluster with autoflag disabled
+    Map<String, String> flags = new HashMap<>();
+    flags.put("ycql_allow_cassandra_drop", "false");
+    restartClusterWithMasterFlags(flags);
+
+    // Verify that we can connect as cassandra role as it gets regenerated with flag disabled
+    checkConnectivity(true, cassandra_user, cassandra_user, false);
+    markClusterNeedsRecreation();
+  }
 }

src/yb/master/catalog_entity_info.proto

@@ -391,6 +391,7 @@ message SysSecurityConfigEntryPB {
   // Roles configuration version. Every time a role gets created/deleted, or a permission gets
   // added/removed, we increase the version.
   optional uint64 roles_version = 1;
+  optional bool cassandra_user_created = 2 [default = false];
 }
 
 // Metadata about the YSQL catalog (current only version).

src/yb/master/permissions_manager.cc

@@ -41,6 +41,10 @@ using strings::Substitute;
 
 DECLARE_bool(ycql_cache_login_info);
 
+DEFINE_RUNTIME_AUTO_bool(
+    ycql_allow_cassandra_drop, kLocalPersisted, false, true,
+    "When true, stops the regeneration of the cassandra user on restart of the cluster.");
+
 // TODO: remove direct references to member fields in CatalogManager from here.
 
 namespace yb {
@@ -85,12 +89,39 @@ PermissionsManager::PermissionsManager(CatalogManager* catalog_manager)
 
 Status PermissionsManager::PrepareDefaultRoles(int64_t term) {
   LockGuard lock(mutex_);
-  if (FindPtrOrNull(roles_map_, kDefaultCassandraUsername) != nullptr) {
+
+  bool userExists = (FindPtrOrNull(roles_map_, kDefaultCassandraUsername) != nullptr);
+  if (GetAtomicFlag(&FLAGS_ycql_allow_cassandra_drop)) {
+    bool userAlreadyCreated =
+        security_config_->LockForRead()->pb.security_config().cassandra_user_created();
+
+    if (userAlreadyCreated) {
+      LOG(INFO) << "Role " << kDefaultCassandraUsername
+                << " already created, skipping initialization."
+                << " User cassandra exists: " << userExists;
+      return Status::OK();
+    }
+
+    if (userExists) {
+      // If the user exists currently in the db, mark that the cassandra user was created
+      // Used during cluster upgrade
+      auto l = CHECK_NOTNULL(security_config_.get())->LockForWrite();
+      l.mutable_data()->pb.mutable_security_config()->set_cassandra_user_created(true);
+      RETURN_NOT_OK(catalog_manager_->sys_catalog_->Upsert(term, security_config_));
+      l.Commit();
+      LOG(INFO) << "Role " << kDefaultCassandraUsername
+                << " was already created, marked it as created in config";
+      return Status::OK();
+    }
+  }
+
+  if (userExists) {
     LOG(INFO) << "Role " << kDefaultCassandraUsername
-              << " already created, skipping initialization";
+              << " was already created, skipping initialization";
     return Status::OK();
   }
 
+  // This must either be a new cluster or a cluster upgrading with a deleted cassandra user
   char hash[kBcryptHashSize];
   // TODO: refactor interface to be more c++ like...
   int ret = bcrypt_hashpw(kDefaultCassandraPassword, hash);
@@ -102,6 +133,12 @@ Status PermissionsManager::PrepareDefaultRoles(int64_t term) {
   Status s = CreateRoleUnlocked(kDefaultCassandraUsername, std::string(hash, kBcryptHashSize),
                                 true, true, term, false /* Don't increment the roles version */);
   if (PREDICT_TRUE(s.ok())) {
+    if (GetAtomicFlag(&FLAGS_ycql_allow_cassandra_drop)) {
+      auto l = CHECK_NOTNULL(security_config_.get())->LockForWrite();
+      l.mutable_data()->pb.mutable_security_config()->set_cassandra_user_created(true);
+      RETURN_NOT_OK(catalog_manager_->sys_catalog_->Upsert(term, security_config_));
+      l.Commit();
+    }
     LOG(INFO) << "Created role: " << kDefaultCassandraUsername;
   }
 

src/yb/master/sys_catalog-test.cc

@@ -520,6 +520,7 @@ TEST_F(SysCatalogTest, TestSysCatalogSysConfigOperations) {
   {
     auto l = security_config->LockForWrite();
     l.mutable_data()->pb.mutable_security_config()->set_roles_version(0);
+    l.mutable_data()->pb.mutable_security_config()->set_cassandra_user_created(true);
     l.Commit();
   }
   scoped_refptr<SysConfigInfo> ysql_catalog_config = new SysConfigInfo(kYsqlCatalogConfigType);