X-Git-Url: https://git.whamcloud.com/?a=blobdiff_plain;f=lustre%2Fmdt%2Fmdt_hsm_cdt_client.c;h=83626272b2885d75cea95a87150e30557928c9a5;hb=825570b0b03e0fd70aba5400dedca4e6d94d5141;hp=8264070bed20c2391918a92e9ae045baff07d052;hpb=4b02a71242504c1aa51fa5bca87721031bf670ae;p=fs%2Flustre-release.git

diff --git a/lustre/mdt/mdt_hsm_cdt_client.c b/lustre/mdt/mdt_hsm_cdt_client.c
index 8264070..8362627 100644
--- a/lustre/mdt/mdt_hsm_cdt_client.c
+++ b/lustre/mdt/mdt_hsm_cdt_client.c
@@ -39,7 +39,6 @@
 #include <lustre_net.h>
 #include <lustre_export.h>
 #include <obd.h>
-#include <obd_lov.h>
 #include <lprocfs_status.h>
 #include <lustre_log.h>
 #include "mdt_internal.h"
@@ -245,6 +244,43 @@ static bool hal_is_sane(struct hsm_action_list *hal)
 	RETURN(true);
 }
 
+static int
+hsm_action_permission(struct mdt_thread_info *mti,
+		      struct mdt_object *obj,
+		      enum hsm_copytool_action hsma)
+{
+	struct coordinator *cdt = &mti->mti_mdt->mdt_coordinator;
+	struct lu_ucred *uc = mdt_ucred(mti);
+	struct md_attr *ma = &mti->mti_attr;
+	const __u64 *mask;
+	int rc;
+	ENTRY;
+
+	if (hsma != HSMA_RESTORE &&
+	    exp_connect_flags(mti->mti_exp) & OBD_CONNECT_RDONLY)
+		RETURN(-EROFS);
+
+	if (md_capable(uc, CFS_CAP_SYS_ADMIN))
+		RETURN(0);
+
+	ma->ma_need = MA_INODE;
+	rc = mdt_attr_get_complex(mti, obj, ma);
+	if (rc < 0)
+		RETURN(rc);
+
+	if (uc->uc_fsuid == ma->ma_attr.la_uid)
+		mask = &cdt->cdt_user_request_mask;
+	else if (lustre_in_group_p(uc, ma->ma_attr.la_gid))
+		mask = &cdt->cdt_group_request_mask;
+	else
+		mask = &cdt->cdt_other_request_mask;
+
+	if (!(0 <= hsma && hsma < 8 * sizeof(*mask)))
+		RETURN(-EINVAL);
+
+	RETURN(*mask & (1UL << hsma) ? 0 : -EPERM);
+}
+
 /*
  * Coordinator external API
  */
@@ -321,23 +357,30 @@ int mdt_hsm_add_actions(struct mdt_thread_info *mti,
 		 * if restore, we take the layout lock
 		 */
 
-		/* if action is cancel, also no need to check */
-		if (hai->hai_action == HSMA_CANCEL)
-			goto record;
-
-		/* get HSM attributes */
+		/* Get HSM attributes and check permissions. */
 		obj = mdt_hsm_get_md_hsm(mti, &hai->hai_fid, &mh);
-		if (IS_ERR(obj) || obj == NULL) {
-			/* in case of archive remove, Lustre file
-			 * is not mandatory */
-			if (hai->hai_action == HSMA_REMOVE)
+		if (IS_ERR(obj)) {
+			/* In case of REMOVE and CANCEL a Lustre file
+			 * is not mandatory, but restrict this
+			 * exception to admins. */
+			if (md_capable(mdt_ucred(mti), CFS_CAP_SYS_ADMIN) &&
+			    (hai->hai_action == HSMA_REMOVE ||
+			     hai->hai_action == HSMA_CANCEL))
 				goto record;
-			if (obj == NULL)
-				GOTO(out, rc = -ENOENT);
-			GOTO(out, rc = PTR_ERR(obj));
+			else
+				GOTO(out, rc = PTR_ERR(obj));
 		}
+
+		rc = hsm_action_permission(mti, obj, hai->hai_action);
 		mdt_object_put(mti->mti_env, obj);
 
+		if (rc < 0)
+			GOTO(out, rc);
+
+		/* if action is cancel, also no need to check */
+		if (hai->hai_action == HSMA_CANCEL)
+			goto record;
+
 		/* Check if an action is needed, compare request
 		 * and HSM flags status */
 		if (!hsm_action_is_needed(hai, archive_id, flags, &mh))
@@ -351,6 +394,7 @@ int mdt_hsm_add_actions(struct mdt_thread_info *mti,
 
 		/* for cancel archive number is taken from canceled request
 		 * for other request, we take from lma if not specified,
+		 * or we use the default if none found in lma
 		 * this works also for archive because the default value is 0
 		 * /!\ there is a side effect: in case of restore on multiple
 		 * files which are in different backend, the initial compound
@@ -358,43 +402,48 @@ int mdt_hsm_add_actions(struct mdt_thread_info *mti,
 		 * warranty an agent can serve any combinaison of archive
 		 * backend
 		 */
-		if (hai->hai_action != HSMA_CANCEL && archive_id == 0)
-			archive_id = mh.mh_arch_id;
+		if (hai->hai_action != HSMA_CANCEL && archive_id == 0) {
+			if (mh.mh_arch_id != 0)
+				archive_id = mh.mh_arch_id;
+			else
+				archive_id = cdt->cdt_default_archive_id;
+		}
 
 		/* if restore, take an exclusive lock on layout */
 		if (hai->hai_action == HSMA_RESTORE) {
-			struct cdt_restore_handle	*crh;
-			struct mdt_object		*child;
+			struct cdt_restore_handle *crh;
+
+			/* in V1 only whole file is supported. */
+			if (hai->hai_extent.offset != 0)
+				GOTO(out, rc = -EPROTO);
 
 			OBD_SLAB_ALLOC_PTR(crh, mdt_hsm_cdt_kmem);
 			if (crh == NULL)
 				GOTO(out, rc = -ENOMEM);
 
 			crh->crh_fid = hai->hai_fid;
-			/* in V1 only whole file is supported
-			crh->extent.start = hai->hai_extent.offset;
-			crh->extent.end = hai->hai_extent.offset +
-					    hai->hai_extent.length;
-			 */
+			/* in V1 only whole file is supported. However the
+			 * restore may be due to truncate. */
 			crh->crh_extent.start = 0;
-			crh->crh_extent.end = OBD_OBJECT_EOF;
+			crh->crh_extent.end = hai->hai_extent.length;
 
 			mdt_lock_reg_init(&crh->crh_lh, LCK_EX);
-			child = mdt_object_find_lock(mti, &crh->crh_fid,
-						     &crh->crh_lh,
-						     MDS_INODELOCK_LAYOUT);
-			if (IS_ERR(child)) {
-				rc = PTR_ERR(child);
+			obj = mdt_object_find_lock(mti, &crh->crh_fid,
+						   &crh->crh_lh,
+						   MDS_INODELOCK_LAYOUT);
+			if (IS_ERR(obj)) {
+				rc = PTR_ERR(obj);
 				CERROR("%s: cannot take layout lock for "
 				       DFID": rc = %d\n", mdt_obd_name(mdt),
 				       PFID(&crh->crh_fid), rc);
 				OBD_SLAB_FREE_PTR(crh, mdt_hsm_cdt_kmem);
 				GOTO(out, rc);
 			}
+
 			/* we choose to not keep a keep a reference
 			 * on the object during the restore time which can be
 			 * very long */
-			mdt_object_put(mti->mti_env, child);
+			mdt_object_put(mti->mti_env, obj);
 
 			mutex_lock(&cdt->cdt_restore_lock);
 			list_add_tail(&crh->crh_list, &cdt->cdt_restore_hdl);