[ 7444, 7465] Prevent replication with insufficient permissions

This adds a check before attempting a replication to ensure that the
user has permission to perform the operation. At present, modify_object
is required to perform replication. There are permission levels between
read_object and modify_object which may allow users to create new
replicas but they will not be finalized properly. Until this is
resolved or the required permission level for replication is changed,
we need to ensure that only users with modify_object or higher can
replicate a data object.

server/api/src/rsDataObjRepl.cpp

@@ -40,13 +40,15 @@
 #include "irods/specColl.hpp"
 #include "irods/unbunAndRegPhyBunfile.h"
 
+#include "irods/catalog_utilities.hpp"
 #include "irods/finalize_utilities.hpp"
 #include "irods/irods_at_scope_exit.hpp"
 #include "irods/irods_log.hpp"
 #include "irods/irods_logger.hpp"
 #include "irods/irods_random.hpp"
 #include "irods/irods_resource_backport.hpp"
 #include "irods/irods_resource_redirect.hpp"
+#include "irods/irods_rs_comm_query.hpp"
 #include "irods/irods_server_api_call.hpp"
 #include "irods/irods_server_properties.hpp"
 #include "irods/irods_stacktrace.hpp"
@@ -57,6 +59,9 @@
 #include "irods/replication_utilities.hpp"
 #include "irods/voting.hpp"
 
+#define IRODS_QUERY_ENABLE_SERVER_SIDE_API
+#include "irods/irods_query.hpp"
+
 #define IRODS_REPLICA_ENABLE_SERVER_SIDE_API
 #include "irods/data_object_proxy.hpp"
 #include "irods/replica_proxy.hpp"
@@ -472,8 +477,81 @@ namespace
         return rsDataObjOpen(&_comm, &_inp);
     } // open_destination_replica
 
+    auto user_has_permission_to_replicate_object(RsComm& _comm, const DataObjInp& _inp) -> bool
+    {
+        // 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(&_inp.condInput, ADMIN_KW)) {
+            return true;
+        }
+
+        const auto path = irods::experimental::filesystem::path{_inp.objPath};
+        const auto permission_query_string =
+            fmt::format("select DATA_ACCESS_TYPE where USER_NAME = '{}' and  COLL_NAME = '{}' and DATA_NAME = '{}'",
+                        _comm.clientUser.userName,
+                        path.parent_path().c_str(),
+                        path.object_name().c_str());
+
+        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 (static_cast<access_type>(std::stoi(access_type_string))) {
+                case access_type::modify_object:
+                    [[fallthrough]]; // NOLINT(bugprone-branch-clone)
+                case access_type::delete_object:
+                    [[fallthrough]]; // NOLINT(bugprone-branch-clone)
+                case access_type::own:
+                    return true;
+
+                default:
+                    return false;
+            }
+        }
+        catch (const std::invalid_argument& e) {
+            log_api::error("DATA_ACCESS_TYPE for user [{}] on object [{}] has invalid value [{}]",
+                           _comm.clientUser.userName,
+                           path.c_str(),
+                           access_type_string);
+            return false;
+        }
+        catch (const std::out_of_range& e) {
+            log_api::error("DATA_ACCESS_TYPE for user [{}] on object [{}] has out-of-range value [{}]",
+                           _comm.clientUser.userName,
+                           path.c_str(),
+                           access_type_string);
+            return false;
+        }
+    } // user_has_permission_to_replicate_object
+
     int replicate_data(RsComm& _comm, DataObjInp& _source_inp, DataObjInp& _destination_inp, transferStat_t** _stat)
     {
+        // This check is required because certain operations used by this API in order to perform a replication have
+        // different permissions requirements which may not align. In order to complete the replication, the maximum
+        // required permission level across all of these operations is the minimum required permission to perform the
+        // full operation. As such, an additional check is required to make sure the replication can proceed.
+        if (!user_has_permission_to_replicate_object(_comm, _source_inp)) {
+            const std::string msg =
+                fmt::format("User [{}] does not have sufficient permission to replicate object [{}].",
+                            _comm.clientUser.userName,
+                            _source_inp.objPath);
+            const auto ec = SYS_USER_NO_PERMISSION;
+
+            addRErrorMsg(&_comm.rError, ec, msg.data());
+            log_api::info(msg);
+
+            return ec;
+        }
+
         // Open source replica
         const int source_l1descInx = open_source_replica(_comm, _source_inp);
         if (source_l1descInx < 0) {