Re: [PATCH v3 4/4] ceph: fail quarantined subvolume access with EACCES

[Al Viro <viro@zeniv.linux.org.uk>]

> +/* A snapshot path must reach a head inode to encode its snapshot name. */
> +static bool ceph_snapshot_has_path(struct inode *inode)
> +{
> +	struct dentry *alias = d_find_alias(inode);
> +	struct dentry *cur;
> +	unsigned int seq, depth = 0;
> +	bool connected = false;
> +
> +	if (!alias)
> +		return false;
> +
> +	rcu_read_lock();
> +	seq = read_seqbegin(&rename_lock);
> +	cur = alias;
> +	while (cur && depth++ < 256) {
> +		struct inode *in = d_inode_rcu(cur);
> +		struct dentry *parent;
> +
> +		if (!in)
> +			break;
> +		if (!ceph_in_snap(in)) {
> +			connected = true;
> +			break;
> +		}
> +		if (d_unhashed(cur) ||
> +		    (READ_ONCE(cur->d_flags) & DCACHE_DISCONNECTED))
> +			break;
> +		parent = READ_ONCE(cur->d_parent);
> +		if (!parent || parent == cur)
                    ^^^^^^^
Please, explain how could that possibly happen.

> +			break;
> +		cur = parent;
> +	}
> +	if (read_seqretry(&rename_lock, seq))
> +		connected = false;
> +	rcu_read_unlock();
> +	dput(alias);
> +	return connected;
> +}
> +
> +static int ceph_snapshot_reply_error(struct inode *inode,
> +				     struct ceph_mds_request *req, int mask)
> +{
> +	struct ceph_inode_info *ci = ceph_inode(inode);
> +	bool have_xattrs;
> +
> +	if (!req->r_target_inode ||
> +	    ceph_ino(req->r_target_inode) != ceph_ino(inode) ||
> +	    ceph_snap(req->r_target_inode) != ceph_snap(inode))
> +		return -EOPNOTSUPP;
> +
> +	/* LOOKUPINO can authorize access without returning the requested blob. */
> +	if ((mask & CEPH_CAP_XATTR_SHARED) &&
> +	    req->r_reply_info.targeti.xattr_len < 4) {
> +		spin_lock(&ci->i_ceph_lock);
> +		have_xattrs = ci->i_xattrs.version != 0;
> +		spin_unlock(&ci->i_ceph_lock);
> +		if (!have_xattrs)
> +			return -EOPNOTSUPP;
> +	}
> +	if ((mask & CEPH_CAP_FILE_RD) &&
> +	    req->r_reply_info.targeti.inline_version == 0 &&
> +	    ceph_has_inline_data(ci) && i_size_read(inode) > 0)
> +		return -EOPNOTSUPP;
> +	return 0;
> +}
> +
> +static int ceph_snapshot_lookupino(struct inode *inode,
> +				   struct folio *locked_folio, int mask)
> +{
> +	struct ceph_mds_client *mdsc = ceph_sb_to_mdsc(inode->i_sb);
> +	struct ceph_mds_request *req;
> +	int err;
> +
> +	req = ceph_mdsc_create_request(mdsc, CEPH_MDS_OP_LOOKUPINO, USE_ANY_MDS);
> +	if (IS_ERR(req))
> +		return PTR_ERR(req);
> +	/* No r_inode: an NFS alias may not encode a snapshot pathname. */
> +	req->r_ino1 = ceph_vino(inode);
> +	req->r_args.lookupino.snapid = cpu_to_le64(ceph_snap(inode));
> +	req->r_args.lookupino.mask = cpu_to_le32(mask);
> +	req->r_num_caps = 1;
> +	req->r_locked_folio = locked_folio;
> +	err = ceph_mdsc_do_request(mdsc, NULL, req);
> +	if (!err)
> +		err = ceph_snapshot_reply_error(inode, req, mask);
> +	if (locked_folio && !err) {
> +		u64 version = req->r_reply_info.targeti.inline_version;
> +
> +		if (version == 0)
> +			err = -EINVAL;
> +		else if (version == CEPH_INLINE_NONE || version == 1)
> +			err = -ENODATA;
> +		else
> +			err = req->r_reply_info.targeti.inline_len;
> +	}
> +	ceph_mdsc_put_request(req);
> +	return err;
> +}
> +
> +int ceph_snapshot_read_auth(struct inode *inode)
> +{
> +	/*
> +	 * Preserve a connected snapshot path: its head inode may have been
> +	 * deleted and no longer be cached by the MDS. Disconnected handles
> +	 * fall back to LOOKUPINO, with the exact inode and snapshot ID.
> +	 * The authorization is scoped to this operation, never cached.
> +	 */
> +	return ceph_do_getattr(inode, CEPH_CAP_AUTH_SHARED, true);
> +}
> +
>  /*
>   * Verify that we have a lease on the given mask.  If not,
>   * do a getattr against an mds.
> @@ -3582,20 +3863,43 @@ int __ceph_do_getattr(struct inode *inode, struct folio *locked_folio,
>  	struct ceph_client *cl = fsc->client;
>  	struct ceph_mds_client *mdsc = fsc->mdsc;
>  	struct ceph_mds_request *req;
> +	u64 qepoch;
>  	int mode;
>  	int err;
>  
>  	if (ceph_snap(inode) == CEPH_SNAPDIR) {
> +		struct inode *parent;
> +
>  		boutc(cl, "inode %p %llx.%llx SNAPDIR\n", inode,
>  		      ceph_vinop(inode));
> -		return 0;
> +		/* The synthetic snapdir has the real directory's ino. */
> +		parent = ceph_lookup_inode(inode->i_sb, ceph_ino(inode));
> +		if (IS_ERR(parent))
> +			return PTR_ERR(parent);
> +		err = ceph_do_getattr(parent, mask, true);
> +		iput(parent);
> +		return err;
>  	}
> +	if (ceph_in_snap(inode) && !ceph_snapshot_has_path(inode))
> +		return ceph_snapshot_lookupino(inode, locked_folio, mask);
> +
> +	/*
> +	 * Cached caps would satisfy getattr after the MDS has revoked
> +	 * access, and a short walk has no ancestor to consult. Ask the
> +	 * MDS so a lost quarantine-disable can clear the mark, and so
> +	 * an uncached parent chain is not treated as clear. Immutable snapshot
> +	 * caps cannot receive revocations and always require an MDS check.
> +	 */
> +	if (ceph_in_snap(inode) ||
> +	    ceph_inode_quarantine_state(inode) != CEPH_Q_CLEAR)
> +		force = true;
>  
>  	boutc(cl, "inode %p %llx.%llx mask %s mode 0%o\n", inode,
>  	      ceph_vinop(inode), ceph_cap_string(mask), inode->i_mode);
>  	if (!force && ceph_caps_issued_mask_metric(ceph_inode(inode), mask, 1))
>  			return 0;
>  
> +	qepoch = (u64)atomic64_read_acquire(&mdsc->quarantine_epoch);
>  	mode = ceph_try_to_choose_auth_mds(inode, mask);
>  	req = ceph_mdsc_create_request(mdsc, CEPH_MDS_OP_GETATTR, mode);
>  	if (IS_ERR(req))
> @@ -3606,6 +3910,15 @@ int __ceph_do_getattr(struct inode *inode, struct folio *locked_folio,
>  	req->r_args.getattr.mask = cpu_to_le32(mask);
>  	req->r_locked_folio = locked_folio;
>  	err = ceph_mdsc_do_request(mdsc, NULL, req);
> +	if (ceph_in_snap(inode)) {
> +		if (err == -EINVAL || err == -ESTALE ||
> +		    (!err && ceph_snapshot_reply_error(inode, req, mask))) {
> +			ceph_mdsc_put_request(req);
> +			return ceph_snapshot_lookupino(inode, locked_folio, mask);
> +		}
> +	}
> +	if (!err && !ceph_in_snap(inode))
> +		ceph_quarantine_remember_clear(inode, qepoch, req->r_cred);
>  	if (locked_folio && err == 0) {
>  		u64 inline_version = req->r_reply_info.targeti.inline_version;
>  		if (inline_version == 0) {
> @@ -3751,8 +4064,14 @@ int ceph_getattr(struct mnt_idmap *idmap, const struct path *path,
>  		return -ESTALE;
>  	}
>  
> -	/* Skip the getattr altogether if we're asked not to sync */
> -	if ((flags & AT_STATX_SYNC_TYPE) != AT_STATX_DONT_SYNC) {
> +	/*
> +	 * AT_STATX_DONT_SYNC would return the cached attrs. A
> +	 * quarantined inode, or one whose parents are not cached,
> +	 * still has to ask the MDS. Snapshot caps cannot receive revocations.
> +	 */
> +	if (ceph_in_snap(inode) ||
> +	    ceph_inode_quarantine_state(inode) != CEPH_Q_CLEAR ||
> +	    (flags & AT_STATX_SYNC_TYPE) != AT_STATX_DONT_SYNC) {
>  		err = ceph_do_getattr(inode,
>  				statx_to_caps(request_mask, inode->i_mode),
>  				flags & AT_STATX_FORCE_SYNC);
> diff --git a/fs/ceph/ioctl.c b/fs/ceph/ioctl.c
> index 12cb50e0a166..d152cad10c23 100644
> --- a/fs/ceph/ioctl.c
> +++ b/fs/ceph/ioctl.c
> @@ -201,6 +201,10 @@ static long ceph_ioctl_get_dataloc(struct file *file, void __user *arg)
>  	if (copy_from_user(&dl, arg, sizeof(dl)))
>  		return -EFAULT;
>  
> +	r = ceph_do_getattr(inode, CEPH_STAT_CAP_LAYOUT, false);
> +	if (r)
> +		return r;
> +
>  	down_read(&osdc->lock);
>  	ceph_calc_file_object_mapping(&ci->i_layout, dl.file_offset, 1,
>  				      &dl.object_no, &dl.object_offset, &xlen);
> @@ -401,12 +405,18 @@ long ceph_ioctl(struct file *file, unsigned int cmd, unsigned long arg)
>  
>  	case FS_IOC_GET_ENCRYPTION_POLICY:
>  		ret = vet_mds_for_fscrypt(file);
> +		if (ret)
> +			return ret;
> +		ret = ceph_quarantine_revalidate(inode);
>  		if (ret)
>  			return ret;
>  		return fscrypt_ioctl_get_policy(file, (void __user *)arg);
>  
>  	case FS_IOC_GET_ENCRYPTION_POLICY_EX:
>  		ret = vet_mds_for_fscrypt(file);
> +		if (ret)
> +			return ret;
> +		ret = ceph_quarantine_revalidate(inode);
>  		if (ret)
>  			return ret;
>  		return fscrypt_ioctl_get_policy_ex(file, (void __user *)arg);
> @@ -429,6 +439,9 @@ long ceph_ioctl(struct file *file, unsigned int cmd, unsigned long arg)
>  
>  	case FS_IOC_GET_ENCRYPTION_NONCE:
>  		ret = vet_mds_for_fscrypt(file);
> +		if (ret)
> +			return ret;
> +		ret = ceph_quarantine_revalidate(inode);
>  		if (ret)
>  			return ret;
>  		return fscrypt_ioctl_get_nonce(file, (void __user *)arg);
> diff --git a/fs/ceph/mds_client.h b/fs/ceph/mds_client.h
> index 816f5f2d32cc..67d4c6279e00 100644
> --- a/fs/ceph/mds_client.h
> +++ b/fs/ceph/mds_client.h
> @@ -39,7 +39,14 @@ enum ceph_feature_type {
>  	CEPHFS_FEATURE_MDS_AUTH_CAPS_CHECK,
>  	CEPHFS_FEATURE_SUBVOLUME_METRICS,
>  
> -	CEPHFS_FEATURE_MAX = CEPHFS_FEATURE_SUBVOLUME_METRICS,
> +	/*
> +	 * Bit 23 is CEPHFS_FEATURE_BLOCKDIFF on the MDS and is not
> +	 * advertised here. Bit 24 is subvolume quarantine; it must stay
> +	 * aligned with the userspace client.
> +	 */
> +	CEPHFS_FEATURE_QUARANTINE = 24,
> +
> +	CEPHFS_FEATURE_MAX = CEPHFS_FEATURE_QUARANTINE,
>  };
>  
>  #define CEPHFS_FEATURES_CLIENT_SUPPORTED {	\
> @@ -57,6 +64,7 @@ enum ceph_feature_type {
>  	CEPHFS_FEATURE_HAS_OWNER_UIDGID,	\
>  	CEPHFS_FEATURE_MDS_AUTH_CAPS_CHECK,	\
>  	CEPHFS_FEATURE_SUBVOLUME_METRICS,	\
> +	CEPHFS_FEATURE_QUARANTINE,		\
>  }
>  
>  /*
> diff --git a/fs/ceph/super.h b/fs/ceph/super.h
> index 0264bd9d705f..540f31121f84 100644
> --- a/fs/ceph/super.h
> +++ b/fs/ceph/super.h
> @@ -1394,8 +1394,6 @@ int ceph_set_acl(struct mnt_idmap *idmap,
>  		 struct dentry *dentry, struct posix_acl *acl, int type);
>  int ceph_pre_init_acls(struct inode *dir, umode_t *mode,
>  		       struct ceph_acl_sec_ctx *as_ctx);
> -void ceph_init_inode_acls(struct inode *inode,
> -			  struct ceph_acl_sec_ctx *as_ctx);
>  
>  static inline void ceph_forget_all_cached_acls(struct inode *inode)
>  {
> @@ -1412,11 +1410,6 @@ static inline int ceph_pre_init_acls(struct inode *dir, umode_t *mode,
>  {
>  	return 0;
>  }
> -static inline void ceph_init_inode_acls(struct inode *inode,
> -					struct ceph_acl_sec_ctx *as_ctx)
> -{
> -}
> -
>  static inline void ceph_forget_all_cached_acls(struct inode *inode)
>  {
>  }
> @@ -1578,7 +1571,7 @@ static inline bool ceph_has_inline_data(const struct ceph_inode_info *ci)
>  /* file.c */
>  extern const struct file_operations ceph_file_fops;
>  
> -extern int ceph_renew_caps(struct inode *inode, int fmode);
> +int ceph_renew_caps(struct inode *inode, int fmode, bool force);
>  extern int ceph_open(struct inode *inode, struct file *file);
>  extern int ceph_atomic_open(struct inode *dir, struct dentry *dentry,
>  			    struct file *file, unsigned flags, umode_t mode);
> diff --git a/fs/ceph/xattr.c b/fs/ceph/xattr.c
> index 4cba6e7bffd3..7fd307fcbe5a 100644
> --- a/fs/ceph/xattr.c
> +++ b/fs/ceph/xattr.c
> @@ -997,6 +997,7 @@ ssize_t __ceph_getxattr(struct inode *inode, const char *name, void *value,
>  	struct ceph_vxattr *vxattr;
>  	int req_mask;
>  	ssize_t err;
> +	bool ask_mds = false;
>  
>  	if (strncmp(name, XATTR_CEPH_PREFIX, XATTR_CEPH_PREFIX_LEN))
>  		goto handle_non_vxattrs;
> @@ -1034,7 +1035,19 @@ ssize_t __ceph_getxattr(struct inode *inode, const char *name, void *value,
>  	      ceph_vinop(inode), name, ci->i_xattrs.version,
>  	      ci->i_xattrs.index_version);
>  
> -	if (ci->i_xattrs.version == 0 ||
> +	/*
> +	 * A shared cap would answer from the cache. The state check may
> +	 * schedule, so drop i_ceph_lock first. A reply reading the blob
> +	 * it just installed must still use that cache.
> +	 */
> +	if (!ceph_current_fill_trace_request()) {
> +		spin_unlock(&ci->i_ceph_lock);
> +		ask_mds = ceph_in_snap(inode) ||
> +			  ceph_inode_quarantine_state(inode) != CEPH_Q_CLEAR;
> +		spin_lock(&ci->i_ceph_lock);
> +	}
> +
> +	if (ask_mds || ci->i_xattrs.version == 0 ||
>  	    !((req_mask & CEPH_CAP_XATTR_SHARED) ||
>  	      __ceph_caps_issued_mask_metric(ci, CEPH_CAP_XATTR_SHARED, 1))) {
>  		spin_unlock(&ci->i_ceph_lock);
> @@ -1089,6 +1102,7 @@ ssize_t ceph_listxattr(struct dentry *dentry, char *names, size_t size)
>  	struct ceph_client *cl = ceph_inode_to_client(inode);
>  	struct ceph_inode_info *ci = ceph_inode(inode);
>  	bool len_only = (size == 0);
> +	bool ask_mds = false;
>  	u32 namelen;
>  	int err;
>  	struct ceph_journal_info __ji;
> @@ -1100,7 +1114,15 @@ ssize_t ceph_listxattr(struct dentry *dentry, char *names, size_t size)
>  	      ceph_vinop(inode), ci->i_xattrs.version,
>  	      ci->i_xattrs.index_version);
>  
> -	if (ci->i_xattrs.version == 0 ||
> +	/* Same as __ceph_getxattr(): do not serve a shared cap hit. */
> +	if (!ceph_current_fill_trace_request()) {
> +		spin_unlock(&ci->i_ceph_lock);
> +		ask_mds = ceph_in_snap(inode) ||
> +			  ceph_inode_quarantine_state(inode) != CEPH_Q_CLEAR;
> +		spin_lock(&ci->i_ceph_lock);
> +	}
> +
> +	if (ask_mds || ci->i_xattrs.version == 0 ||
>  	    !__ceph_caps_issued_mask_metric(ci, CEPH_CAP_XATTR_SHARED, 1)) {
>  		spin_unlock(&ci->i_ceph_lock);
>  		err = ceph_do_getattr(inode, CEPH_STAT_CAP_XATTR, true);
> @@ -1222,6 +1244,7 @@ int __ceph_setxattr(struct inode *inode, const char *name,
>  	int required_blob_size;
>  	bool check_realm = false;
>  	bool lock_snap_rwsem = false;
> +	bool checked_quarantine = false;
>  
>  	if (ceph_in_snap(inode))
>  		return -EROFS;
> @@ -1319,6 +1342,33 @@ int __ceph_setxattr(struct inode *inode, const char *name,
>  		goto retry;
>  	}
>  
> +	/*
> +	 * XATTR_EXCL applies the new blob with no MDS round trip.
> +	 * The state check may schedule, so drop i_ceph_lock first.
> +	 */
> +	if (!checked_quarantine) {
> +		enum ceph_qstate qstate;
> +
> +		spin_unlock(&ci->i_ceph_lock);
> +		qstate = ceph_inode_quarantine_write_state(inode);
> +		if (qstate == CEPH_Q_UNDER &&
> +		    ceph_inode_quarantine_direct(inode)) {
> +			ceph_buffer_put(old_blob);
> +			if (lock_snap_rwsem)
> +				up_read(&mdsc->snap_rwsem);
> +			err = -EACCES;
> +			goto out;
> +		}
> +		if (qstate != CEPH_Q_CLEAR) {
> +			ceph_buffer_put(old_blob);
> +			old_blob = NULL;
> +			goto do_sync_unlocked;
> +		}
> +		checked_quarantine = true;
> +		spin_lock(&ci->i_ceph_lock);
> +		goto retry;
> +	}
> +
>  	err = __set_xattr(ci, newname, name_len, newval, val_len,
>  			  flags, value ? 1 : -1, &xattr);
>  
> -- 
> 2.34.1
> 
>