[ 7816] Check group permissions for replication

In addition to checking whether a user has permission to replicate
a data object, all groups of which the user is a member should be
checked as well. If either the user or any of the user's groups have
sufficient permission to replicate the data object, then the replication
should be allowed.

server/api/src/rsDataObjRepl.cpp

@@ -55,6 +55,7 @@
 #include "irods/irods_string_tokenize.hpp"
 #include "irods/json_serialization.hpp"
 #include "irods/key_value_proxy.hpp"
+#include "irods/logical_locking.hpp"
 #include "irods/replica_access_table.hpp"
 #include "irods/replication_utilities.hpp"
 #include "irods/voting.hpp"
@@ -66,7 +67,8 @@
 #include "irods/data_object_proxy.hpp"
 #include "irods/replica_proxy.hpp"
 
-#include "irods/logical_locking.hpp"
+#define IRODS_USER_ADMINISTRATION_ENABLE_SERVER_SIDE_API
+#include "irods/user_administration.hpp"
 
 #include <cstring>
 #include <algorithm>
@@ -477,62 +479,91 @@ namespace
         return rsDataObjOpen(&_comm, &_inp);
     } // open_destination_replica
 
+    auto permission_is_sufficient_to_replicate_object(const int _access_type) -> bool
+    {
+        using access_type = irods::experimental::catalog::access_type;
+
+        switch (_access_type) {
+            case static_cast<int>(access_type::modify_object): // NOLINT(bugprone-branch-clone)
+                [[fallthrough]];
+            case static_cast<int>(access_type::delete_object): // NOLINT(bugprone-branch-clone)
+                [[fallthrough]];
+            case static_cast<int>(access_type::own):
+                return true;
+
+            default:
+                return false;
+        }
+    } // permission_is_sufficient_to_replicate_object
+
     auto user_has_permission_to_replicate_object(RsComm& _comm, const irods::file_object_ptr _obj) -> bool
     {
+        namespace adm = irods::experimental::administration;
+
         // Short-circuit the query by checking to see whether this is a rodsadmin using the ADMIN_KW, which allows
         // them to do whatever they want.
         if (irods::is_privileged_client(_comm) && getValByKey(&_obj->cond_input(), ADMIN_KW)) {
             return true;
         }
 
+        const auto user = adm::user{_comm.clientUser.userName, _comm.clientUser.rodsZone};
         const auto path = irods::experimental::filesystem::path{_obj->logical_path()};
-        const auto permission_query_string =
-            fmt::format("select DATA_ACCESS_TYPE where USER_NAME = '{}' and USER_ZONE = '{}' and DATA_ID = '{}'",
-                        _comm.clientUser.userName,
-                        _comm.clientUser.rodsZone,
-                        _obj->data_id());
-
-        auto permission_query = irods::query{&_comm, permission_query_string};
-
-        // If the user has no permissions on the object, it may have returned an empty result set.
-        if (permission_query.empty()) {
-            return false;
-        }
-
-        const auto& query_result = permission_query.front();
-        const auto& access_type_string = query_result[0];
 
         try {
-            using access_type = irods::experimental::catalog::access_type;
-
-            switch (std::stoi(access_type_string)) {
-                case static_cast<int>(access_type::modify_object): // NOLINT(bugprone-branch-clone)
-                    [[fallthrough]];
-                case static_cast<int>(access_type::delete_object): // NOLINT(bugprone-branch-clone)
-                    [[fallthrough]];
-                case static_cast<int>(access_type::own):
-                    return true;
+            // Check access permission for the user first...
+            const auto user_permission_query_string =
+                fmt::format("select DATA_ACCESS_TYPE where USER_NAME = '{}' and USER_ZONE = '{}' and DATA_ID = '{}'",
+                            user.name,
+                            user.zone,
+                            _obj->data_id());
+            auto user_permission_query = irods::query{&_comm, user_permission_query_string};
+            if (!user_permission_query.empty() &&
+                permission_is_sufficient_to_replicate_object(std::stoi(user_permission_query.front()[0])))
+            {
+                return true;
+            }
 
-                default:
-                    return false;
+            // If the user does not have sufficient permissions on the data object for replication, perhaps one of their
+            // groups does...
+            const auto groups = adm::server::groups(_comm, user);
+            std::vector<std::string> groups_with_quotes;
+            std::for_each(groups.cbegin(), groups.cend(), [&groups_with_quotes](const auto& group) -> void {
+                groups_with_quotes.emplace_back(fmt::format("'{}'", group.name));
+            });
+            const auto group_permission_query_string = fmt::format(
+                "select DATA_ACCESS_TYPE where USER_NAME in ({}) and USER_TYPE = 'rodsgroup' and DATA_ID = '{}'",
+                fmt::join(groups_with_quotes, ", "),
+                _obj->data_id());
+            auto group_permission_query = irods::query{&_comm, group_permission_query_string};
+            for (const auto& result : group_permission_query) {
+                if (permission_is_sufficient_to_replicate_object(std::stoi(result[0]))) {
+                    return true;
+                }
             }
+
+            return false;
         }
         catch (const std::invalid_argument& e) {
-            log_api::error("DATA_ACCESS_TYPE for user [{}#{}] on object [{}] has invalid value [{}]",
+            log_api::error("DATA_ACCESS_TYPE for user [{}#{}] or one of their groups on object [{}] has invalid value. "
+                           "Message: {}",
                            _comm.clientUser.userName,
                            _comm.clientUser.rodsZone,
                            _obj->logical_path(),
-                           access_type_string);
+                           e.what());
             return false;
         }
         catch (const std::out_of_range& e) {
-            log_api::error("DATA_ACCESS_TYPE for user [{}#{}] on object [{}] has out-of-range value [{}]",
+            log_api::error("DATA_ACCESS_TYPE for user [{}#{}] or one of their groups on object [{}] has out-of-range "
+                           "value. Message: {}",
                            _comm.clientUser.userName,
                            _comm.clientUser.rodsZone,
                            _obj->logical_path(),
-                           access_type_string);
+                           e.what());
             return false;
         }
+
+        // Do not catch other exceptions because other errors would not indicate insufficient permissions but another
+        // issue which will be caught and handled elsewhere.
     } // user_has_permission_to_replicate_object
 
     int replicate_data(RsComm& _comm, DataObjInp& _source_inp, DataObjInp& _destination_inp, transferStat_t** _stat)