[PATCH v6 0/5] ceph: support subvolume quarantine

[PATCH v6 0/5] ceph: support subvolume quarantine

[Alex Markuze]

CephFS subvolume quarantine restricts ordinary access while allowing
recovery clients with q/Q MDS auth caps to operate. Add client support
for quarantine notifications and enforce them on cached access paths.

Based on the testing branch of https://github.com/ceph/ceph-client.git,
commit ed0502cb11f8 ("ceph: validate encoded xattr name lengths").
The full base is recorded below. The dcache and fscrypt prerequisites
are included in this five-patch series.

Changes since v5:
- Describe the dcache export as a prerequisite and remove Reported-by
  and Closes tags for it.
- Reject foreign, disconnected and unhashed non-root aliases in ancestry
  walks while accepting the live, unhashed superblock root.
- Reset quarantine state only when the lost or reconnecting session is
  the reporting MDS rank, including when its cap is already absent.
- Check saved authorization's generation even when no inode is marked.
- Check snapshot path access and acquire normal wanted caps after
  quarantine authorization.
- Clear detached xattr-buffer ownership after releasing its reference.
- Use synchronous CREATE for new inodes with an LSM security blob,
  before async preparation, including before feature negotiation.
- Keep an existing quarantine mark across a zero-error IMPORT while
  processing imported caps and negative quarantine errors normally.
- Use the incoming wire cap mask for clear decisions, so PIN-only updates
  cannot acquire clear authority from capability bits retained in a merge.
- Update encrypted-symlink and capability comments to match their
  interfaces.

Validation: GCC x86_64 builds with fscrypt, ACLs and security enabled
passed W=1 vmlinux/modules for the final tree and the recording patch.
Twenty-six extracted-code quarantine state tests passed, using mocked
kernel context. Live cluster, boot, KASAN/lockdep, LSM policy,
failover, performance and LKMM validation have not been run.

Previous submission:
https://lore.kernel.org/r/cover.1792527529.git.amarkuze@redhat.com/

Alex Markuze (5):
  dcache: export d_find_alias_rcu() for modular filesystems
  fscrypt: allow filesystems to disable symlink target caching
  ceph: add subvolume quarantine wire definitions
  ceph: record subvolume quarantine from MDS cap updates
  ceph: fail quarantined subvolume access with EACCES

 fs/ceph/acl.c                |  33 +-
 fs/ceph/addr.c               |  72 ++-
 fs/ceph/caps.c               | 323 +++++++++++-
 fs/ceph/crypto.c             |   1 +
 fs/ceph/debugfs.c            |   1 +
 fs/ceph/dir.c                |  54 +-
 fs/ceph/file.c               |  82 ++-
 fs/ceph/inode.c              | 956 ++++++++++++++++++++++++++++++++++-
 fs/ceph/ioctl.c              |  13 +
 fs/ceph/mds_client.c         |  26 +-
 fs/ceph/mds_client.h         |  30 +-
 fs/ceph/super.c              |   4 +
 fs/ceph/super.h              | 101 +++-
 fs/ceph/xattr.c              |  60 ++-
 fs/crypto/hooks.c            |  28 +-
 fs/dcache.c                  |   1 +
 include/linux/ceph/ceph_fs.h |  13 +
 include/linux/fscrypt.h      |   7 +
 net/ceph/ceph_common.c       |   2 +
 19 files changed, 1706 insertions(+), 101 deletions(-)


base-commit: ed0502cb11f84fff29f374238e526ae94d9bdfde

[PATCH v6 1/5] dcache: export d_find_alias_rcu() for modular filesystems

[Alex Markuze]

CephFS subvolume quarantine needs a dentry alias for its RCU ancestry
walk without acquiring a dentry reference. d_find_alias_rcu() provides
that lookup, but is unavailable to a modular Ceph client.

Export d_find_alias_rcu() to GPL modules. The Ceph quarantine state
handling patch uses the export with the helper's RCU lifetime rules.

Signed-off-by: Alex Markuze <amarkuze@redhat.com>
---
 fs/dcache.c | 1 +
 1 file changed, 1 insertion(+)

diff --git a/fs/dcache.c b/fs/dcache.c
index 1b1a81f10da67c608704383086c3facb64ee3b3b..13ba553081b33a54fd8d1cac5cbc55d0149e4ad7 100644
--- a/fs/dcache.c
+++ b/fs/dcache.c
@@ -1248,6 +1248,7 @@ struct dentry *d_find_alias_rcu(struct inode *inode)
 	spin_unlock(&inode->i_lock);
 	return de;
 }
+EXPORT_SYMBOL_GPL(d_find_alias_rcu);
 
 /*
  *	Try to kill dentries associated with this inode.

[PATCH v6 2/5] fscrypt: allow filesystems to disable symlink target caching

[Alex Markuze]

The VFS can return a decrypted symlink target from inode->i_link without
calling the filesystem's ->get_link(). A filesystem that revalidates
access in ->get_link() therefore needs a way to prevent that cache from
bypassing its checks. CephFS quarantine requires revalidation even for
readlinkat() through an O_PATH descriptor opened before quarantine.

Add a filesystem opt-out from decrypted symlink target caching. The
option is fixed before inode creation and prevents both creation-time
and lookup-time cache installation. Uncached targets use the existing
delayed-free callback; cache readers need no special handling because
the inode has no cached target.

The Ceph quarantine access patch selects this option. Encrypted symlink
readlink, traversal and lstat then allocate and decrypt a temporary
target on each access. Filesystems that do not opt out retain caching.

Signed-off-by: Alex Markuze <amarkuze@redhat.com>
---
 fs/crypto/hooks.c       | 28 +++++++++++++++-------------
 include/linux/fscrypt.h |  7 +++++++
 2 files changed, 22 insertions(+), 13 deletions(-)

diff --git a/fs/crypto/hooks.c b/fs/crypto/hooks.c
index a7a8a3f581a0460537f5ff955bc8b5d30573a033..6bc7ac82ec3470d633f2203a40c00a422b8e76a3 100644
--- a/fs/crypto/hooks.c
+++ b/fs/crypto/hooks.c
@@ -321,11 +321,13 @@ int __fscrypt_encrypt_symlink(struct inode *inode, const char *target,
 	 */
 	sd->encrypted_path[ciphertext_len] = '\0';
 
-	/* Cache the plaintext symlink target for later use by get_link() */
-	err = -ENOMEM;
-	inode->i_link = kmemdup(target, len + 1, GFP_NOFS);
-	if (!inode->i_link)
-		goto err_free_sd;
+	/* Cache the plaintext target unless the filesystem revalidates it. */
+	if (!inode->i_sb->s_cop->no_symlink_cache) {
+		err = -ENOMEM;
+		inode->i_link = kmemdup(target, len + 1, GFP_NOFS);
+		if (!inode->i_link)
+			goto err_free_sd;
+	}
 
 	if (!disk_link->name)
 		disk_link->name = (unsigned char *)sd;
@@ -411,12 +413,12 @@ const char *fscrypt_get_symlink(struct inode *inode, const void *caddr,
 	pstr.name[pstr.len] = '\0';
 
 	/*
-	 * Cache decrypted symlink targets in i_link for later use.  Don't cache
-	 * symlink targets encoded without the key, since those become outdated
-	 * once the key is added.  This pairs with the READ_ONCE() above and in
-	 * the VFS path lookup code.
+	 * Cache decrypted targets unless the filesystem revalidates them.
+	 * Don't cache targets encoded without the key, since those become
+	 * outdated once the key is added. This pairs with the READ_ONCE()
+	 * above and in the VFS path lookup code.
 	 */
-	if (!has_key ||
+	if (inode->i_sb->s_cop->no_symlink_cache || !has_key ||
 	    cmpxchg_release(&inode->i_link, NULL, pstr.name) != NULL)
 		set_delayed_call(done, kfree_link, pstr.name);
 
@@ -442,9 +444,9 @@ EXPORT_SYMBOL_GPL(fscrypt_get_symlink);
  * This requires reading the symlink target from disk if needed, setting up the
  * inode's encryption key if possible, and then decrypting or encoding the
  * symlink target.  This makes lstat() more heavyweight than is normally the
- * case.  However, decrypted symlink targets will be cached in ->i_link, so
- * usually the symlink won't have to be read and decrypted again later if/when
- * it is actually followed, readlink() is called, or lstat() is called again.
+ * case. Unless the filesystem sets no_symlink_cache, decrypted targets
+ * are cached in ->i_link, avoiding another read and decryption when the
+ * symlink is followed, readlink() is called, or lstat() is called again.
  *
  * Return: 0 on success, -errno on failure
  */
diff --git a/include/linux/fscrypt.h b/include/linux/fscrypt.h
index 28f2108e29f3976ebe861c425b129ac0acec479b..ab77e685c75c5d1132db82e5cfa718ba6b2847fe 100644
--- a/include/linux/fscrypt.h
+++ b/include/linux/fscrypt.h
@@ -108,6 +108,13 @@ struct fscrypt_operations {
 	unsigned int supports_subblock_data_units : 1;
 
 	/*
+	 * Don't populate ->i_link, so every access calls ->get_link() and the
+	 * filesystem can revalidate access. Set before creating any inodes;
+	 * this option cannot invalidate previously cached targets.
+	 */
+	unsigned int no_symlink_cache : 1;
+
+	/*
 	 * This field exists only for backwards compatibility reasons and should
 	 * only be set by the filesystems that are setting it already.  It
 	 * contains the filesystem-specific key description prefix that is

[PATCH v6 3/5] ceph: add subvolume quarantine wire definitions

[Alex Markuze]

The MDS reports -CEPH_EQUARANTINED in MClientCaps v14 when quarantine
denies a client's access, and restricts that client's caps to PIN. The
wire errno is 1000; the client must return EACCES to applications.
MQuarantineDisable identifies the inode whose quarantine mark can be
cleared when access is restored.

Define the errno, message type and packed disable payload, and add the
message name to ceph_msg_type_name(). The quarantine state and access
patches consume these definitions; support is advertised after the
cached-access checks are installed.

Signed-off-by: Alex Markuze <amarkuze@redhat.com>
---
 include/linux/ceph/ceph_fs.h | 13 +++++++++++++
 net/ceph/ceph_common.c       |  2 ++
 2 files changed, 15 insertions(+)

diff --git a/include/linux/ceph/ceph_fs.h b/include/linux/ceph/ceph_fs.h
index d70bbce1a1b2b4ffffebe896ef46eb6853ab354b..5afccfa891cb045e9784eda1cc4741749aac4c06 100644
--- a/include/linux/ceph/ceph_fs.h
+++ b/include/linux/ceph/ceph_fs.h
@@ -154,6 +154,7 @@ const char *ceph_con_mode_name(int mode);
 #define CEPH_MSG_CLIENT_SNAP            0x312
 #define CEPH_MSG_CLIENT_CAPRELEASE      0x313
 #define CEPH_MSG_CLIENT_QUOTA           0x314
+#define CEPH_MSG_CLIENT_QUARANTINE_DISABLE 0x315
 
 /* pool ops */
 #define CEPH_MSG_POOLOP_REPLY           48
@@ -921,4 +922,16 @@ struct ceph_mds_quota {
 	__le64 max_files;	/* quota max. files */
 } __attribute__ ((packed));
 
+/*
+ * Cap oserrno used by the MDS when a subvolume is quarantined.
+ * This is not a VFS errno; applications see -EACCES.
+ */
+#define CEPH_EQUARANTINED	1000
+
+/* MDS -> client: quarantine lifted for one inode (ino, oserrno). */
+struct ceph_mds_quarantine_disable {
+	__le64 ino;
+	__le32 oserrno;
+} __packed;
+
 #endif
diff --git a/net/ceph/ceph_common.c b/net/ceph/ceph_common.c
index 67e88f30409dd3ffba9dadb6fda529f4101d830f..cc3d21fd7d5bf251bd58ce19f8c7d8a5b02fa1d7 100644
--- a/net/ceph/ceph_common.c
+++ b/net/ceph/ceph_common.c
@@ -83,6 +83,8 @@ const char *ceph_msg_type_name(int type)
 	case CEPH_MSG_CLIENT_CAPS: return "client_caps";
 	case CEPH_MSG_CLIENT_CAPRELEASE: return "client_cap_release";
 	case CEPH_MSG_CLIENT_QUOTA: return "client_quota";
+	case CEPH_MSG_CLIENT_QUARANTINE_DISABLE:
+		return "client_quarantine_disable";
 	case CEPH_MSG_CLIENT_SNAP: return "client_snap";
 	case CEPH_MSG_CLIENT_LEASE: return "client_lease";
 	case CEPH_MSG_POOLOP_REPLY: return "poolop_reply";

[PATCH v6 4/5] ceph: record subvolume quarantine from MDS cap updates

[Alex Markuze]

Quarantine notifications must remain effective while inode ancestry is
cached and while capabilities move between MDS ranks. Decode the signed
errno appended to MClientCaps v14 after the subvolume ID, and track a
sticky, counted quarantine mark and its reporting rank on each inode.

A current zero-error non-IMPORT wire grant above PIN or an explicit
disable from the reporting rank clears the mark. Capability bits
retained during migration are not a fresh grant and cannot authorize
clearing. A current metadata reply can confirm recovery across ranks.
PIN-only updates can reset the errno without clearing the mark. IMPORT's
zero can be an unevaluated default, so it does not clear a known mark;
negative IMPORT errors still apply. An IMPORT is installed before grant
processing, making equality with its stored sequence valid. Older
sequences cannot change quarantine state.

Loss of the reporting cap or session leaves the mark pending until an
MDS reply can authorize access. Session removal and reconnect compare
the reporter under i_ceph_lock, including when its cap is absent, so an
unrelated rank cannot demote the mark. State changes wake cap waiters.
Recovery from a retained IMPORT mark requires an eligible normal grant,
explicit disable or current metadata reply.

When marks exist, resolve inherited state with bounded RCU parent walks
and cache complete answers by generation. Unusable or incomplete aliases
require an MDS answer. The actual live superblock root terminates a walk
even though d_make_root() leaves it unhashed; foreign, disconnected,
unhashed non-root and secondary-root aliases cannot establish ancestry.
Validate parent walks against rename_lock without taking references.
Mark changes, renames and alias attachments invalidate cached answers;
sleeping splices exclude them without leaving the generation odd across
an LSM callback.

This supplies the notification and ancestry state consumed by the access
checks. Client quarantine support is advertised by the access patch.

Signed-off-by: Alex Markuze <amarkuze@redhat.com>
---
 fs/ceph/caps.c       | 128 +++++++++++-
 fs/ceph/dir.c        |   9 +
 fs/ceph/file.c       |   2 +
 fs/ceph/inode.c      | 483 ++++++++++++++++++++++++++++++++++++++++++-
 fs/ceph/mds_client.c |  21 +-
 fs/ceph/mds_client.h |  16 ++
 fs/ceph/super.c      |   2 +
 fs/ceph/super.h      |  71 ++++++-
 8 files changed, 721 insertions(+), 11 deletions(-)

diff --git a/fs/ceph/caps.c b/fs/ceph/caps.c
index b27bcdc24d29ff7fc9da1ffe5442ef829b012f33..b07814d0038cd72a6fc10813f50846df58e9591a 100644
--- a/fs/ceph/caps.c
+++ b/fs/ceph/caps.c
@@ -689,11 +689,9 @@ void change_auth_cap_ses(struct ceph_inode_info *ci,
  *
  * Caller should hold session snap_rwsem (read) and ci->i_ceph_lock
  *
- * @fmode is the open file mode, if we are opening a file, otherwise
- * it is < 0.  (This is so we can atomically add the cap and add an
- * open file reference to it.)
+ * Returns false when the raw grant predates the recorded cap sequence.
  */
-void ceph_add_cap(struct inode *inode,
+bool ceph_add_cap(struct inode *inode,
 		  struct ceph_mds_session *session, u64 cap_id,
 		  unsigned issued, unsigned wanted,
 		  unsigned seq, unsigned mseq, u64 realmino, int flags,
@@ -706,6 +704,7 @@ void ceph_add_cap(struct inode *inode,
 	int mds = session->s_mds;
 	int actual_wanted;
 	u32 gen;
+	bool cap_current = true;
 
 	lockdep_assert_held(&ci->i_ceph_lock);
 
@@ -737,6 +736,7 @@ void ceph_add_cap(struct inode *inode,
 		atomic64_inc(&mdsc->metric.total_caps);
 		spin_unlock(&session->s_cap_lock);
 	} else {
+		cap_current = ceph_seq_cmp(seq, cap->seq) >= 0;
 		spin_lock(&session->s_cap_lock);
 		list_move_tail(&cap->session_caps, &session->s_caps);
 		spin_unlock(&session->s_cap_lock);
@@ -823,6 +823,7 @@ void ceph_add_cap(struct inode *inode,
 	cap->mseq = mseq;
 	cap->cap_gen = gen;
 	wake_up_all(&ci->i_cap_wq);
+	return cap_current;
 }
 
 /*
@@ -1264,6 +1265,11 @@ static void __ceph_remove_cap(struct ceph_inode_info *ci, struct ceph_cap *cap,
 
 	mdsc = ceph_inode_to_fs_client(&ci->netfs.inode)->mdsc;
 
+	/* Losing the reporting cap requires fresh MDS revalidation. */
+	if (READ_ONCE(ci->i_qtine_mds) == cap->mds &&
+	    ceph_quarantine_reset(ci))
+		wake_up_all(&ci->i_cap_wq);
+
 	/* remove from inode's cap rbtree, and clear auth cap */
 	rb_erase(&cap->ci_node, &ci->i_caps);
 	if (ci->i_auth_cap == cap)
@@ -3729,6 +3735,9 @@ struct cap_extra_info {
 	u8 *fscrypt_auth;
 	u32 fscrypt_auth_len;
 	u64 fscrypt_file_size;
+	/* MClientCaps v14 oserrno; valid only when oserrno_valid is set */
+	s32 oserrno;
+	bool oserrno_valid;
 };
 
 /*
@@ -3758,6 +3767,46 @@ static int parse_inline_data(void **p, void *end,
 }
 
 /*
+ * Record quarantine state carried by an MClientCaps oserrno.
+ * Caller holds i_ceph_lock. Returns true when waiters should be woken
+ * because quarantine changed or a pending reset was resolved.
+ *
+ * -EQUARANTINED refreshes the sticky flag and its reporting MDS rank.
+ * A current non-IMPORT grant above PIN with errno 0 clears that rank's mark.
+ * IMPORT may carry a default zero without assessing quarantine permission.
+ * Another rank needs a successful metadata request for the current epoch.
+ * A PIN-only update may zero qtine_errno while leaving the flag set.
+ */
+static bool ceph_apply_quarantine(struct ceph_inode_info *ci, int mds,
+				  int newcaps, int oserrno, bool is_import)
+{
+	bool was, now, pending;
+
+	was = READ_ONCE(ci->i_qtine_errno) == -CEPH_EQUARANTINED ||
+	      test_bit(CEPH_I_QUARANTINE_BIT, &ci->i_ceph_flags);
+	pending = test_bit(CEPH_I_QUARANTINE_PENDING_BIT, &ci->i_ceph_flags);
+
+	if (!is_import && (newcaps & ~CEPH_CAP_PIN) && was && oserrno == 0 &&
+	    READ_ONCE(ci->i_qtine_mds) == mds)
+		ceph_quarantine_clear(ci);
+
+	if (oserrno == -CEPH_EQUARANTINED) {
+		ceph_quarantine_set(ci);
+		WRITE_ONCE(ci->i_qtine_mds, mds);
+	}
+
+	if (newcaps || was ||
+	    test_bit(CEPH_I_QUARANTINE_BIT, &ci->i_ceph_flags))
+		WRITE_ONCE(ci->i_qtine_errno, oserrno);
+
+	now = READ_ONCE(ci->i_qtine_errno) == -CEPH_EQUARANTINED ||
+	      test_bit(CEPH_I_QUARANTINE_BIT, &ci->i_ceph_flags);
+	return was != now ||
+	       pending != test_bit(CEPH_I_QUARANTINE_PENDING_BIT,
+				   &ci->i_ceph_flags);
+}
+
+/*
  * Handle a cap GRANT message from the MDS.  (Note that a GRANT may
  * actually be a revocation if it specifies a smaller cap set.)
  *
@@ -3781,6 +3830,7 @@ static void handle_cap_grant(struct inode *inode,
 	u64 max_size = le64_to_cpu(grant->max_size);
 	unsigned char check_caps = 0;
 	bool was_stale = cap->cap_gen < atomic_read(&session->s_cap_gen);
+	bool quarantine_current = ceph_seq_cmp(seq, cap->seq) >= 0;
 	bool wake = false;
 	bool writeback = false;
 	bool queue_trunc = false;
@@ -4128,6 +4178,17 @@ static void handle_cap_grant(struct inode *inode,
 			fill_inline = true;
 	}
 
+	/* Merged caps are not a fresh wire grant authorizing a clear. */
+	if (quarantine_current && extra_info->oserrno_valid &&
+	    ceph_apply_quarantine(ci, session->s_mds, le32_to_cpu(grant->caps),
+				  extra_info->oserrno,
+				  le32_to_cpu(grant->op) == CEPH_CAP_OP_IMPORT)) {
+		boutc(cl, "quarantine oserrno %d caps %s %llx.%llx\n",
+		      extra_info->oserrno, ceph_cap_string(newcaps),
+		      ceph_vinop(inode));
+		wake = true;
+	}
+
 	if (le32_to_cpu(grant->op) == CEPH_CAP_OP_IMPORT) {
 		if (ci->i_auth_cap == cap) {
 			if (newcaps & ~extra_info->issued)
@@ -4826,6 +4887,20 @@ void ceph_handle_caps(struct ceph_mds_session *session,
 			goto bad;
 	}
 
+	if (msg_version >= 13) {
+		/* v13: subvolume_id. The inode id is learned from replies. */
+		ceph_decode_skip_64(&p, end, bad);
+	}
+
+	if (msg_version >= 14) {
+		u32 raw_errno;
+
+		/* v14: signed oserrno, -CEPH_EQUARANTINED when quarantined */
+		ceph_decode_32_safe(&p, end, raw_errno, bad);
+		extra_info.oserrno = (s32)raw_errno;
+		extra_info.oserrno_valid = true;
+	}
+
 	/* lookup ino */
 	inode = ceph_find_inode(mdsc->fsc->sb, vino);
 	boutc(cl, " caps mds%d op %s ino %llx.%llx inode %p seq %u iseq %u mseq %u\n",
@@ -4981,6 +5056,51 @@ void ceph_handle_caps(struct ceph_mds_session *session,
 	goto out;
 }
 
+void ceph_handle_quarantine_disable(struct ceph_mds_client *mdsc,
+				    struct ceph_mds_session *session,
+				    struct ceph_msg *msg)
+{
+	struct super_block *sb = mdsc->fsc->sb;
+	struct ceph_client *cl = mdsc->fsc->client;
+	struct ceph_mds_quarantine_disable *h = msg->front.iov_base;
+	struct ceph_vino vino;
+	struct inode *inode;
+	struct ceph_inode_info *ci;
+
+	if (!ceph_inc_mds_stopping_blocker(mdsc, session))
+		return;
+
+	if (msg->front.iov_len < sizeof(*h)) {
+		pr_err_client(cl, "corrupt quarantine_disable mds%d len %d\n",
+			      session->s_mds, (int)msg->front.iov_len);
+		ceph_msg_dump(msg);
+		goto out;
+	}
+
+	vino.ino = le64_to_cpu(h->ino);
+	vino.snap = CEPH_NOSNAP;
+	inode = ceph_find_inode(sb, vino);
+	if (!inode) {
+		doutc(cl, "quarantine disable ino %llx not cached\n", vino.ino);
+		goto out;
+	}
+
+	ci = ceph_inode(inode);
+	spin_lock(&ci->i_ceph_lock);
+	/* Another rank's disable must not erase a newer negative. */
+	if (!test_bit(CEPH_I_QUARANTINE_BIT, &ci->i_ceph_flags) ||
+	    READ_ONCE(ci->i_qtine_mds) == session->s_mds) {
+		ceph_quarantine_clear(ci);
+		WRITE_ONCE(ci->i_qtine_errno, 0);
+	}
+	spin_unlock(&ci->i_ceph_lock);
+	wake_up_all(&ci->i_cap_wq);
+	doutc(cl, "quarantine disable ino %llx\n", vino.ino);
+	iput(inode);
+out:
+	ceph_dec_mds_stopping_blocker(mdsc);
+}
+
 /*
  * Process the expired end of one delayed cap list, whose entries were
  * queued @hold jiffies before they expire.
diff --git a/fs/ceph/dir.c b/fs/ceph/dir.c
index 384e9edabe2a3a0cba8c56b07e04356f3a9920ff..d8c8b7ad893092611014b6094864a14bd36aecca 100644
--- a/fs/ceph/dir.c
+++ b/fs/ceph/dir.c
@@ -795,7 +795,9 @@ struct dentry *ceph_handle_snapdir(struct ceph_mds_request *req,
 		struct dentry *res;
 		struct inode *inode = ceph_get_snapdir(parent);
 
+		ceph_quarantine_splice_start(parent->i_sb);
 		res = d_splice_alias(inode, dentry);
+		ceph_quarantine_splice_finish(parent->i_sb);
 		boutc_formats(cl,
 			      "ENOENT on snapdir %p '%s', linking to snapdir %p %llx.%llx. Spliced dentry %p\n",
 			      "ENOENT on snapdir %p '%pd', linking to snapdir %p %llx.%llx. Spliced dentry %p\n",
@@ -1373,7 +1375,9 @@ static int ceph_link(struct dentry *old_dentry, struct inode *dir,
 		d_drop(dentry);
 	} else if (!req->r_reply_info.head->is_dentry) {
 		ihold(d_inode(old_dentry));
+		ceph_quarantine_splice_start(dir->i_sb);
 		d_instantiate(dentry, d_inode(old_dentry));
+		ceph_quarantine_splice_finish(dir->i_sb);
 	}
 	ceph_mdsc_put_request(req);
 	ceph_blog_exit(&__ji);
@@ -1701,12 +1705,17 @@ static int ceph_rename(struct mnt_idmap *idmap, struct inode *old_dir,
 	}
 	err = ceph_mdsc_do_request(mdsc, old_dir, req);
 	if (!err && !req->r_reply_info.head->is_dentry) {
+		struct inode *moved;
+
 		/*
 		 * Normally d_move() is done by fill_trace (called by
 		 * do_request, above).  If there is no trace, we need
 		 * to do it here.
 		 */
+		moved = d_inode(old_dentry);
+		ceph_quarantine_move_start(moved, old_dir, new_dir);
 		d_move(old_dentry, new_dentry);
+		ceph_quarantine_move_finish(moved, old_dir, new_dir);
 	}
 	ceph_mdsc_put_request(req);
 	ceph_blog_exit(&__ji);
diff --git a/fs/ceph/file.c b/fs/ceph/file.c
index 856710c6376508d51a505c693fe3b3e1ab1e8896..9a6eadb94ab4f7e67d565b258e780129d9a505ff 100644
--- a/fs/ceph/file.c
+++ b/fs/ceph/file.c
@@ -785,7 +785,9 @@ static int ceph_finish_async_create(struct inode *dir, struct inode *inode,
 		if (d_in_lookup(dentry) || d_really_is_negative(dentry)) {
 			if (!d_unhashed(dentry))
 				d_drop(dentry);
+			ceph_quarantine_splice_start(dir->i_sb);
 			dn = d_splice_alias(inode, dentry);
+			ceph_quarantine_splice_finish(dir->i_sb);
 			WARN_ON_ONCE(dn && dn != dentry);
 		}
 		file->f_mode |= FMODE_CREATED;
diff --git a/fs/ceph/inode.c b/fs/ceph/inode.c
index 94f5215f8eed5927477ce59c4c9601013db9cfc0..048e0d502eb4ea361389bc4a6e5ddfa9fd9f6ac5 100644
--- a/fs/ceph/inode.c
+++ b/fs/ceph/inode.c
@@ -15,6 +15,7 @@
 #include <linux/sort.h>
 #include <linux/iversion.h>
 #include <linux/fscrypt.h>
+#include <linux/cred.h>
 
 #include "super.h"
 #include "mds_client.h"
@@ -648,6 +649,11 @@ struct inode *ceph_alloc_inode(struct super_block *sb)
 	ci->i_max_bytes = 0;
 	ci->i_max_files = 0;
 	ci->i_subvolume_id = CEPH_SUBVOLUME_ID_NONE;
+	ci->i_qtine_errno = 0;
+	ci->i_qtine_mds = -1;
+	atomic64_set(&ci->i_qtag, 0);
+	atomic64_set(&ci->i_qmds, 0);
+	ci->i_qmds_cred = NULL;
 
 	memset(&ci->i_dir_layout, 0, sizeof(ci->i_dir_layout));
 	memset(&ci->i_cached_layout, 0, sizeof(ci->i_cached_layout));
@@ -730,10 +736,468 @@ struct inode *ceph_alloc_inode(struct super_block *sb)
 	return &ci->netfs.inode;
 }
 
+/*
+ * quarantine_epoch is a generation shared by every mark change and every
+ * cross-directory rename. The first writer flips it odd, the last
+ * writer flips it even. Readers must not trust or publish a cache
+ * while it is odd, so a cached false cannot outlive the mutation.
+ *
+ * quarantine_lock makes the writers update and the epoch bump one
+ * step. The epoch is already odd before this returns, and it does not
+ * go even until the last writer has finished the bit change or d_move.
+ * Callers that hold i_ceph_lock take this lock after it. The lock is
+ * not held across the mutation.
+ */
+static void ceph_quarantine_write_begin(struct ceph_mds_client *mdsc)
+{
+	if (!mdsc)
+		return;
+
+	spin_lock(&mdsc->quarantine_lock);
+	if (atomic_inc_return(&mdsc->quarantine_writers) == 1)
+		atomic64_inc_return(&mdsc->quarantine_epoch);
+	spin_unlock(&mdsc->quarantine_lock);
+}
+
+static void ceph_quarantine_write_end(struct ceph_mds_client *mdsc)
+{
+	if (!mdsc)
+		return;
+
+	spin_lock(&mdsc->quarantine_lock);
+	if (atomic_dec_return(&mdsc->quarantine_writers) == 0)
+		atomic64_inc_return(&mdsc->quarantine_epoch);
+	spin_unlock(&mdsc->quarantine_lock);
+}
+
+void ceph_quarantine_set(struct ceph_inode_info *ci)
+{
+	struct ceph_mds_client *mdsc;
+
+	mdsc = ceph_sb_to_mdsc(ci->netfs.inode.i_sb);
+	/* Every accepted negative invalidates earlier requests and receipts. */
+	ceph_quarantine_write_begin(mdsc);
+	if (mdsc)
+		atomic_inc(&mdsc->quarantine_count);
+	if (test_and_set_bit(CEPH_I_QUARANTINE_BIT, &ci->i_ceph_flags)) {
+		if (mdsc)
+			atomic_dec(&mdsc->quarantine_count);
+	}
+	clear_bit(CEPH_I_QUARANTINE_PENDING_BIT, &ci->i_ceph_flags);
+	ceph_quarantine_write_end(mdsc);
+}
+
+bool ceph_quarantine_clear(struct ceph_inode_info *ci)
+{
+	struct ceph_mds_client *mdsc;
+	bool cleared;
+
+	if (!test_bit(CEPH_I_QUARANTINE_BIT, &ci->i_ceph_flags))
+		return false;
+
+	mdsc = ceph_sb_to_mdsc(ci->netfs.inode.i_sb);
+	ceph_quarantine_write_begin(mdsc);
+	cleared = test_and_clear_bit(CEPH_I_QUARANTINE_BIT, &ci->i_ceph_flags);
+	if (cleared) {
+		clear_bit(CEPH_I_QUARANTINE_PENDING_BIT, &ci->i_ceph_flags);
+		WRITE_ONCE(ci->i_qtine_errno, 0);
+		WRITE_ONCE(ci->i_qtine_mds, -1);
+		if (mdsc)
+			atomic_dec(&mdsc->quarantine_count);
+	}
+	ceph_quarantine_write_end(mdsc);
+	return cleared;
+}
+
+/* Session loss requires an MDS answer without reopening cached snap caps. */
+bool ceph_quarantine_reset(struct ceph_inode_info *ci)
+{
+	struct ceph_mds_client *mdsc;
+
+	if (!test_bit(CEPH_I_QUARANTINE_BIT, &ci->i_ceph_flags) ||
+	    test_bit(CEPH_I_QUARANTINE_PENDING_BIT, &ci->i_ceph_flags))
+		return false;
+
+	mdsc = ceph_sb_to_mdsc(ci->netfs.inode.i_sb);
+	ceph_quarantine_write_begin(mdsc);
+	set_bit(CEPH_I_QUARANTINE_PENDING_BIT, &ci->i_ceph_flags);
+	WRITE_ONCE(ci->i_qtine_errno, 0);
+	ceph_quarantine_write_end(mdsc);
+	return true;
+}
+
+bool ceph_inode_quarantine_direct(struct inode *inode)
+{
+	struct ceph_inode_info *ci;
+
+	if (!inode)
+		return false;
+	ci = ceph_inode(inode);
+	return !test_bit(CEPH_I_QUARANTINE_PENDING_BIT, &ci->i_ceph_flags) &&
+		(READ_ONCE(ci->i_qtine_errno) == -CEPH_EQUARANTINED ||
+		 test_bit(CEPH_I_QUARANTINE_BIT, &ci->i_ceph_flags));
+}
+
+/* Low bit is the answer. The rest is quarantine_epoch, odd values excluded. */
+static u64 ceph_qtag_pack(u64 epoch, bool under)
+{
+	return epoch | (under ? 1ULL : 0);
+}
+
+static void ceph_quarantine_publish(struct ceph_inode_info *ci, bool under,
+				    u64 epoch)
+{
+	if (epoch & 1)
+		return;
+
+	/* Pairs with atomic64_read_acquire() of i_qtag. */
+	atomic64_set_release(&ci->i_qtag, ceph_qtag_pack(epoch, under));
+}
+
+static bool ceph_qtag_read(struct ceph_inode_info *ci, u64 epoch, bool *under)
+{
+	u64 packed;
+
+	if (epoch & 1)
+		return false;
+
+	/* Pairs with atomic64_set_release() in ceph_quarantine_publish(). */
+	packed = atomic64_read_acquire(&ci->i_qtag);
+	if ((packed & ~1ULL) != epoch)
+		return false;
+	*under = packed & 1ULL;
+	return true;
+}
+
+/* (epoch | 1) marks an MDS confirmation for this even epoch. */
+static bool ceph_qmds_confirmed(struct ceph_inode_info *ci, u64 epoch)
+{
+	u64 cookie;
+	bool confirmed;
+
+	if (epoch & 1)
+		return false;
+
+	/* Pairs with atomic64_set_release() in remember_clear(). */
+	if ((u64)atomic64_read_acquire(&ci->i_qmds) != (epoch | 1ULL))
+		return false;
+	/* The cookie and its immutable credentials are one locked receipt. */
+	spin_lock(&ci->i_ceph_lock);
+	cookie = atomic64_read(&ci->i_qmds);
+	confirmed = cookie == (epoch | 1ULL) &&
+		ci->i_qmds_cred == current_cred();
+	spin_unlock(&ci->i_ceph_lock);
+	return confirmed;
+}
+
+static void ceph_quarantine_forget_receipt(struct ceph_inode_info *ci)
+{
+	const struct cred *cred;
+
+	spin_lock(&ci->i_ceph_lock);
+	cred = ci->i_qmds_cred;
+	ci->i_qmds_cred = NULL;
+	atomic64_set_release(&ci->i_qmds, 0);
+	spin_unlock(&ci->i_ceph_lock);
+	put_cred(cred);
+}
+
+/* A finished walk reached a mark, an ancestor tag, or the mount root. */
+enum {
+	CEPH_QWALK_CLEAR,
+	CEPH_QWALK_UNDER,
+	CEPH_QWALK_INCOMPLETE,
+};
+
+/*
+ * Parent walk with no inode references. Caller holds rcu_read_lock().
+ * Stops at the mount root, at an MDS mark, or at an ancestor whose
+ * cached answer is for this epoch. The superblock pins its live root,
+ * which d_make_root() leaves unhashed. Other unhashed or disconnected
+ * aliases, secondary roots and the depth limit leave the walk incomplete.
+ */
+static int ceph_quarantine_walk_once(struct inode *inode, u64 epoch,
+				     bool use_cache)
+{
+	struct inode *cur = inode;
+	unsigned int depth = 0;
+	bool under = false;
+	bool done = false;
+
+	while (cur && depth++ < 256) {
+		struct ceph_inode_info *ci = ceph_inode(cur);
+		struct dentry *alias;
+		struct dentry *parent;
+		struct inode *next;
+
+		if (test_bit(CEPH_I_QUARANTINE_PENDING_BIT, &ci->i_ceph_flags))
+			break;
+		if (READ_ONCE(ci->i_qtine_errno) == -CEPH_EQUARANTINED ||
+		    test_bit(CEPH_I_QUARANTINE_BIT, &ci->i_ceph_flags)) {
+			under = true;
+			done = true;
+			break;
+		}
+
+		if (use_cache && cur != inode &&
+		    ceph_qtag_read(ci, epoch, &under)) {
+			done = true;
+			break;
+		}
+
+		alias = d_find_alias_rcu(cur);
+		if (!alias)
+			break;
+
+		if (alias->d_sb != inode->i_sb ||
+		    (READ_ONCE(alias->d_flags) & DCACHE_DISCONNECTED))
+			break;
+
+		parent = READ_ONCE(alias->d_parent);
+		/*
+		 * The mount root is its own parent. A disconnected
+		 * alias from d_obtain_alias(), or a secondary root from
+		 * d_obtain_root(), has the same shape and is not a
+		 * finished walk.
+		 */
+		if (!parent || parent == alias) {
+			if (parent == alias && S_ISDIR(cur->i_mode) &&
+			    alias == READ_ONCE(inode->i_sb->s_root))
+				done = true;
+			break;
+		}
+		if (d_unhashed(alias))
+			break;
+
+		next = d_inode_rcu(parent);
+		if (!next || next == cur || next->i_sb != inode->i_sb)
+			break;
+		cur = next;
+	}
+	if (!done)
+		return CEPH_QWALK_INCOMPLETE;
+	return under ? CEPH_QWALK_UNDER : CEPH_QWALK_CLEAR;
+}
+
+/*
+ * rename_lock covers d_move(). A chain observed across that move is
+ * not an answer. A few retries, then the caller asks the MDS.
+ */
+static int ceph_quarantine_walk(struct inode *inode, u64 epoch, bool use_cache)
+{
+	unsigned int seq;
+	unsigned int tries = 0;
+	int ret = CEPH_QWALK_INCOMPLETE;
+
+	rcu_read_lock();
+	do {
+		seq = read_seqbegin(&rename_lock);
+		ret = ceph_quarantine_walk_once(inode, epoch, use_cache);
+		if (!read_seqretry(&rename_lock, seq))
+			break;
+		ret = CEPH_QWALK_INCOMPLETE;
+		cpu_relax();
+	} while (++tries < 4);
+	rcu_read_unlock();
+	return ret;
+}
+
+/*
+ * CEPH_Q_UNDER when this inode or an ancestor was marked by an MDS cap
+ * update. Ancestors cover children that have not yet received their
+ * own cap message.
+ *
+ * quarantine_count is zero unless some inode is marked, so the common
+ * path does not walk. While a mark is set, a finished walk is one word
+ * on the inode. A walk stops at an ancestor that already holds that
+ * answer. A short walk is not stored in i_qtag. Callers ask the MDS.
+ * An MDS read receipt in i_qmds overrides inherited marks for the
+ * queried inode until the next epoch bump. It does not override a
+ * definite mark on that inode or authorize cached writes.
+ *
+ * An odd quarantine_epoch is a mark or a d_move() in progress. The
+ * observation is dropped and the check is repeated. Bounded retries
+ * return UNKNOWN, so a preempted writer cannot pin the CPU.
+ */
+static enum ceph_qstate __ceph_inode_quarantine_state(struct inode *inode,
+						      bool read_confirmed)
+{
+	struct ceph_mds_client *mdsc;
+	struct ceph_inode_info *ci;
+	unsigned int retries = 0;
+	u64 epoch;
+	bool under, splicing, cached;
+
+	if (!inode)
+		return CEPH_Q_CLEAR;
+
+	mdsc = ceph_sb_to_mdsc(inode->i_sb);
+	if (!mdsc || atomic_read(&mdsc->quarantine_count) == 0)
+		return CEPH_Q_CLEAR;
+
+	ci = ceph_inode(inode);
+	for (;;) {
+		/* Bound retries across both odd writers and changing even epochs. */
+		if (++retries > 32) {
+			cond_resched();
+			return CEPH_Q_UNKNOWN;
+		}
+		/* Pairs with atomic64_inc_return() in write_begin(). */
+		epoch = (u64)atomic64_read_acquire(&mdsc->quarantine_epoch);
+		if (epoch & 1) {
+			cpu_relax();
+			cond_resched();
+			continue;
+		}
+		splicing = atomic_read_acquire(&mdsc->quarantine_splices) != 0;
+
+		if (ceph_inode_quarantine_direct(inode)) {
+			under = true;
+			cached = true;
+		} else if (read_confirmed && !splicing &&
+			   ceph_qmds_confirmed(ci, epoch)) {
+			under = false;
+			cached = true;
+		} else {
+			cached = !splicing && ceph_qtag_read(ci, epoch, &under);
+		}
+		if (cached) {
+			/* Orders the marker, receipt, and tag loads before the retry. */
+			smp_rmb();
+			if ((u64)atomic64_read(&mdsc->quarantine_epoch) == epoch &&
+			    (splicing || !atomic_read(&mdsc->quarantine_splices)))
+				return under ? CEPH_Q_UNDER : CEPH_Q_CLEAR;
+			cpu_relax();
+			cond_resched();
+			continue;
+		}
+
+		switch (ceph_quarantine_walk(inode, epoch, !splicing)) {
+		case CEPH_QWALK_UNDER:
+			under = true;
+			break;
+		case CEPH_QWALK_CLEAR:
+			under = false;
+			break;
+		default:
+			/* Orders the walk before the retry. */
+			smp_rmb();
+			if ((u64)atomic64_read(&mdsc->quarantine_epoch) == epoch &&
+			    (splicing || !atomic_read(&mdsc->quarantine_splices))) {
+				return CEPH_Q_UNKNOWN;
+			}
+			cpu_relax();
+			cond_resched();
+			continue;
+		}
+		/* Orders the walk before the retry, as in read_seqcount_retry(). */
+		smp_rmb();
+		if ((u64)atomic64_read(&mdsc->quarantine_epoch) == epoch &&
+		    (splicing || !atomic_read(&mdsc->quarantine_splices))) {
+			if (!splicing)
+				ceph_quarantine_publish(ci, under, epoch);
+			return under ? CEPH_Q_UNDER : CEPH_Q_CLEAR;
+		}
+		cpu_relax();
+		cond_resched();
+	}
+}
+
+enum ceph_qstate ceph_inode_quarantine_state(struct inode *inode)
+{
+	return __ceph_inode_quarantine_state(inode, true);
+}
+
+enum ceph_qstate ceph_inode_quarantine_write_state(struct inode *inode)
+{
+	/* GETATTR can prove read recovery access under a read-only q grant. */
+	return __ceph_inode_quarantine_state(inode, false);
+}
+
+static struct ceph_mds_client *ceph_quarantine_move_mdsc(struct inode *inode,
+							 struct inode *old_dir,
+							 struct inode *new_dir)
+{
+	if (!inode || old_dir == new_dir)
+		return NULL;
+	return ceph_sb_to_mdsc(inode->i_sb);
+}
+
+/*
+ * Bracket d_move(). The epoch is odd across the parent change, so a
+ * cached false cannot be reused against the new parent. d_move() does
+ * not sleep. Same-directory renames do not take a writer slot.
+ *
+ * Sleeping alias splices use a separate counter below.
+ */
+void ceph_quarantine_move_start(struct inode *inode, struct inode *old_dir,
+				struct inode *new_dir)
+{
+	ceph_quarantine_write_begin(ceph_quarantine_move_mdsc(inode, old_dir,
+							      new_dir));
+}
+
+void ceph_quarantine_move_finish(struct inode *inode, struct inode *old_dir,
+				 struct inode *new_dir)
+{
+	ceph_quarantine_write_end(ceph_quarantine_move_mdsc(inode, old_dir,
+							    new_dir));
+}
+
+/*
+ * d_splice_alias() and d_instantiate() can sleep in the LSM hook before
+ * changing ancestry. Keep the epoch even for that hook, but disallow
+ * all cached answers throughout the splice. A fresh walk also samples
+ * rename_lock, so it cannot accept a chain torn by a move. Bracket
+ * first attachments too: descendants may hold MDS confirmations from
+ * when the directory had no connected alias.
+ *
+ * Fully ordered atomics and quarantine_lock make the counter and epoch
+ * transitions one writer operation. Entry disables cache use before
+ * invalidating tags; exit invalidates tags before enabling cache use.
+ * Adding two preserves the parity of any concurrent mark writer.
+ */
+void ceph_quarantine_splice_start(struct super_block *sb)
+{
+	struct ceph_mds_client *mdsc = ceph_sb_to_mdsc(sb);
+
+	spin_lock(&mdsc->quarantine_lock);
+	atomic_inc_return(&mdsc->quarantine_splices);
+	atomic64_add_return(2, &mdsc->quarantine_epoch);
+	spin_unlock(&mdsc->quarantine_lock);
+}
+
+void ceph_quarantine_splice_finish(struct super_block *sb)
+{
+	struct ceph_mds_client *mdsc = ceph_sb_to_mdsc(sb);
+
+	spin_lock(&mdsc->quarantine_lock);
+	atomic64_add_return(2, &mdsc->quarantine_epoch);
+	atomic_dec_return(&mdsc->quarantine_splices);
+	spin_unlock(&mdsc->quarantine_lock);
+}
+
+static bool ceph_quarantine_reply_can_clear(struct ceph_inode_info *ci,
+					    struct ceph_mds_session *session)
+{
+	struct ceph_mds_client *mdsc = ceph_sb_to_mdsc(ci->netfs.inode.i_sb);
+	struct ceph_mds_request *req;
+	u64 epoch;
+
+	if (READ_ONCE(ci->i_qtine_mds) == session->s_mds)
+		return true;
+	/* Different ranks have independent cap sequences. */
+	req = ceph_current_mds_request(mdsc->fsc);
+	epoch = (u64)atomic64_read_acquire(&mdsc->quarantine_epoch);
+	return req && !(epoch & 1) && req->r_quarantine_epoch == epoch &&
+		!atomic_read(&mdsc->quarantine_splices);
+}
+
 void ceph_free_inode(struct inode *inode)
 {
 	struct ceph_inode_info *ci = ceph_inode(inode);
 
+	put_cred(ci->i_qmds_cred);
 	kfree(ci->i_symlink);
 #ifdef CONFIG_FS_ENCRYPTION
 	kfree(ci->fscrypt_auth);
@@ -759,6 +1223,8 @@ void ceph_evict_inode(struct inode *inode)
 	percpu_counter_dec(&mdsc->metric.total_inodes);
 
 	ci->i_subvolume_id = CEPH_SUBVOLUME_ID_NONE;
+	ceph_quarantine_forget_receipt(ci);
+	ceph_quarantine_clear(ci);
 
 	netfs_wait_for_outstanding_io(inode);
 	truncate_inode_pages_final(&inode->i_data);
@@ -1407,7 +1873,7 @@ int ceph_fill_inode(struct inode *inode, struct folio *locked_folio,
 	/* were we issued a capability? */
 	if (info_caps) {
 		if (!ceph_in_snap(inode)) {
-			ceph_add_cap(inode, session,
+			bool cap_current = ceph_add_cap(inode, session,
 				     le64_to_cpu(info->cap.cap_id),
 				     info_caps,
 				     le32_to_cpu(info->cap.wanted),
@@ -1416,6 +1882,14 @@ int ceph_fill_inode(struct inode *inode, struct folio *locked_folio,
 				     le64_to_cpu(info->cap.realm),
 				     info->cap.flags, &new_cap);
 
+			/*
+			 * Caps above PIN mean quarantine has been lifted,
+			 * including when the disable message was lost.
+			 */
+			if (cap_current && (info_caps & ~CEPH_CAP_PIN) &&
+			    ceph_quarantine_reply_can_clear(ci, session))
+				ceph_quarantine_clear(ci);
+
 			/* set dir completion flag? */
 			if (S_ISDIR(inode->i_mode) &&
 			    ci->i_files == 0 && ci->i_subdirs == 0 &&
@@ -1618,6 +2092,7 @@ static int splice_dentry(struct dentry **pdn, struct inode *in)
 		realdn = d_find_any_alias(in);
 		if (realdn) {
 			struct ceph_dentry_info *di = ceph_dentry(realdn);
+
 			spin_lock(&realdn->d_lock);
 
 			realdn->d_op->d_prune(realdn);
@@ -1634,7 +2109,9 @@ static int splice_dentry(struct dentry **pdn, struct inode *in)
 	/* dn must be unhashed */
 	if (!d_unhashed(dn))
 		d_drop(dn);
+	ceph_quarantine_splice_start(dn->d_sb);
 	realdn = d_splice_alias(in, dn);
+	ceph_quarantine_splice_finish(dn->d_sb);
 	if (IS_ERR(realdn)) {
 		pr_err_client(cl, "error %ld %p inode %p ino %llx.%llx\n",
 			      PTR_ERR(realdn), dn, in, ceph_vinop(in));
@@ -1859,6 +2336,7 @@ int ceph_fill_trace(struct super_block *sb, struct ceph_mds_request *req)
 		/* rename? */
 		if (req->r_old_dentry && req->r_op == CEPH_MDS_OP_RENAME) {
 			struct inode *olddir = req->r_old_dentry_dir;
+			struct inode *moved;
 			BUG_ON(!olddir);
 
 			boutc(cl, " src %p '%s' dst %p '%s'\n",
@@ -1869,7 +2347,10 @@ int ceph_fill_trace(struct super_block *sb, struct ceph_mds_request *req)
 			ceph_dir_clear_ordered(dir);
 			ceph_dir_clear_ordered(olddir);
 
+			moved = d_inode(req->r_old_dentry);
+			ceph_quarantine_move_start(moved, olddir, dir);
 			d_move(req->r_old_dentry, dn);
+			ceph_quarantine_move_finish(moved, olddir, dir);
 			boutc(cl, " src %p '%s' dst %p '%s'\n",
 			      req->r_old_dentry, req->r_old_dentry->d_name.name, dn, dn->d_name.name);
 
diff --git a/fs/ceph/mds_client.c b/fs/ceph/mds_client.c
index d4ef8a355a48090489cdacaf01cccbdcc0d8ed7d..5fec6e129d2822847c5dab6b2674560712db2765 100644
--- a/fs/ceph/mds_client.c
+++ b/fs/ceph/mds_client.c
@@ -2041,10 +2041,13 @@ static int remove_session_caps_cb(struct inode *inode, int mds, void *arg)
 	struct ceph_inode_info *ci = ceph_inode(inode);
 	struct ceph_client *cl = ceph_inode_to_client(inode);
 	bool invalidate = false;
+	bool reset;
 	struct ceph_cap *cap;
 	int iputs = 0;
 
 	spin_lock(&ci->i_ceph_lock);
+	/* The reporter can outlive its cap; test the rank under the inode lock. */
+	reset = ci->i_qtine_mds == mds && ceph_quarantine_reset(ci);
 	cap = __get_cap_for_mds(ci, mds);
 	if (cap) {
 		boutc(cl, " removing cap %p, ci is %p, inode is %p\n",
@@ -2054,7 +2057,7 @@ static int remove_session_caps_cb(struct inode *inode, int mds, void *arg)
 	}
 	spin_unlock(&ci->i_ceph_lock);
 
-	if (cap)
+	if (cap || reset)
 		wake_up_all(&ci->i_cap_wq);
 	if (invalidate)
 		ceph_queue_invalidate(inode);
@@ -2860,6 +2863,7 @@ ceph_mdsc_create_request(struct ceph_mds_client *mdsc, int op, int mode)
 
 	mutex_init(&req->r_fill_mutex);
 	req->r_mdsc = mdsc;
+	req->r_quarantine_epoch = atomic64_read_acquire(&mdsc->quarantine_epoch);
 	req->r_started = jiffies;
 	req->r_start_latency = ktime_get();
 	req->r_resend_mds = -1;
@@ -5009,6 +5013,7 @@ static int reconnect_caps_cb(struct inode *inode, int mds, void *arg)
 	struct ceph_path_info path_info = {0};
 	int err;
 	u64 snap_follows;
+	bool reset = false;
 
 	dentry = d_find_primary(inode);
 	if (dentry) {
@@ -5023,9 +5028,13 @@ static int reconnect_caps_cb(struct inode *inode, int mds, void *arg)
 	}
 
 	spin_lock(&ci->i_ceph_lock);
+	/* Unrelated session recovery must not demote the current reporter's mark. */
+	reset = ci->i_qtine_mds == mds && ceph_quarantine_reset(ci);
 	cap = __get_cap_for_mds(ci, mds);
 	if (!cap) {
 		spin_unlock(&ci->i_ceph_lock);
+		if (reset)
+			wake_up_all(&ci->i_cap_wq);
 		err = 0;
 		goto out_err;
 	}
@@ -5088,6 +5097,8 @@ static int reconnect_caps_cb(struct inode *inode, int mds, void *arg)
 		snap_follows = capsnap->follows;
 	}
 	spin_unlock(&ci->i_ceph_lock);
+	if (reset)
+		wake_up_all(&ci->i_cap_wq);
 
 	if (recon_state->msg_version >= 2) {
 		int num_fcntl_locks, num_flock_locks;
@@ -6468,6 +6479,11 @@ int ceph_mdsc_init(struct ceph_fs_client *fsc)
 	init_completion(&mdsc->safe_umount_waiters);
 	spin_lock_init(&mdsc->stopping_lock);
 	atomic_set(&mdsc->stopping_blockers, 0);
+	atomic_set(&mdsc->quarantine_count, 0);
+	atomic_set(&mdsc->quarantine_writers, 0);
+	atomic_set(&mdsc->quarantine_splices, 0);
+	atomic64_set(&mdsc->quarantine_epoch, 0);
+	spin_lock_init(&mdsc->quarantine_lock);
 	init_completion(&mdsc->stopping_waiter);
 	atomic64_set(&mdsc->dirty_folios, 0);
 	init_waitqueue_head(&mdsc->flush_end_wq);
@@ -7368,6 +7384,9 @@ static void mds_dispatch(struct ceph_connection *con, struct ceph_msg *msg)
 	case CEPH_MSG_CLIENT_QUOTA:
 		ceph_handle_quota(mdsc, s, msg);
 		break;
+	case CEPH_MSG_CLIENT_QUARANTINE_DISABLE:
+		ceph_handle_quarantine_disable(mdsc, s, msg);
+		break;
 
 	default:
 		pr_err_client(cl, "received unknown message type %d %s\n",
diff --git a/fs/ceph/mds_client.h b/fs/ceph/mds_client.h
index 863cdb8d588ae3121f31d84a7a4f049e73c0d05f..816f5f2d32ccf19b8d0644a3d5c24a730fd1015b 100644
--- a/fs/ceph/mds_client.h
+++ b/fs/ceph/mds_client.h
@@ -375,6 +375,7 @@ struct ceph_mds_request {
 	int r_fmode;        /* file mode, if expecting cap */
 	int r_request_release_offset;
 	const struct cred *r_cred;
+	u64 r_quarantine_epoch; /* generation when the request was created */
 	struct mnt_idmap *r_mnt_idmap;
 	struct timespec64 r_stamp;
 
@@ -506,6 +507,21 @@ struct ceph_mds_client {
 	spinlock_t              stopping_lock;  /* protect snap_empty */
 	int                     stopping;      /* the stage of shutting down */
 	atomic_t                stopping_blockers;
+	/*
+	 * Inodes with CEPH_I_QUARANTINE set. Hot paths skip the
+	 * ancestor walk while this is zero.
+	 * quarantine_epoch is a seqcount: even while idle, odd while
+	 * quarantine_writers is non-zero. Writers are mark changes and
+	 * cross-directory renames. quarantine_lock covers the writers
+	 * update and the epoch bump together. Sleeping alias splices use
+	 * quarantine_splices instead: readers walk without cached answers
+	 * while it is non-zero. Each splice edge also bumps the epoch.
+	 */
+	atomic_t		quarantine_count;
+	atomic_t		quarantine_writers;
+	atomic_t		quarantine_splices;
+	atomic64_t		quarantine_epoch;
+	spinlock_t		quarantine_lock; /* writers update + epoch bump */
 	struct completion	stopping_waiter;
 
 	atomic64_t		dirty_folios;
diff --git a/fs/ceph/super.c b/fs/ceph/super.c
index 40be4b16e486ff0e6b5eb63e1caaa6870df7140e..ad9460d9673cb18230d8fbcb93a27aa26f7ace38 100644
--- a/fs/ceph/super.c
+++ b/fs/ceph/super.c
@@ -1139,7 +1139,9 @@ static struct dentry *open_root_dentry(struct ceph_fs_client *fsc,
 		struct inode *inode = req->r_target_inode;
 		req->r_target_inode = NULL;
 		doutc(cl, "success\n");
+		ceph_quarantine_splice_start(fsc->sb);
 		root = d_make_root(inode);
+		ceph_quarantine_splice_finish(fsc->sb);
 		if (!root) {
 			root = ERR_PTR(-ENOMEM);
 			goto out;
diff --git a/fs/ceph/super.h b/fs/ceph/super.h
index 458fd63dd79f7d73402188b5ba67645662e720ee..0264bd9d705fcb1bc904d0fc7fc94be2b6de3c66 100644
--- a/fs/ceph/super.h
+++ b/fs/ceph/super.h
@@ -444,6 +444,33 @@ struct ceph_inode_info {
 #define CEPH_SUBVOLUME_ID_NONE 0
 	u64 i_subvolume_id;
 
+	/*
+	 * Cap oserrno from the MDS. -CEPH_EQUARANTINED means this inode
+	 * was revoked because its subvolume is quarantined. Paired with
+	 * CEPH_I_QUARANTINE, which stays set when a later PIN-only update
+	 * clears the errno.
+	 */
+	int i_qtine_errno;
+	int i_qtine_mds; /* negative cap origin, retained across pending reset */
+	/*
+	 * Cached ancestry answer: epoch | under, where epoch is even.
+	 * Atomic64 accesses also prevent tearing on 32-bit clients. An odd
+	 * quarantine_epoch means a mark or a rename is in progress.
+	 * Do not trust this word while the epoch is odd or an alias
+	 * splice is in progress.
+	 * Zero means "not under" at epoch 0.
+	 *
+	 * i_qmds stores (epoch | 1) after the MDS confirms an inode
+	 * can be read despite uncertain or marked ancestry. This read
+	 * permission is scoped to this inode, never its descendants or
+	 * cached writes. The next epoch bump drops the confirmation.
+	 * Zero means there is no confirmation.
+	 */
+	atomic64_t i_qtag;
+	atomic64_t i_qmds;
+	/* Immutable request credentials scope the MDS read receipt. */
+	const struct cred *i_qmds_cred; /* protected with i_qmds by i_ceph_lock */
+
 	s32 i_dir_pin;
 
 	int i_fragtree_nsplits;
@@ -773,6 +800,8 @@ static inline u64 ceph_get_truncate_pagecache_size(const struct ceph_inode_info
 					      * the deferred work completes
 					      */
 #define CEPH_I_EVICT_ON_FINAL_IPUT_BIT	(16) /* evict at the final iput() */
+#define CEPH_I_QUARANTINE_BIT		(17) /* subvolume quarantine opt set */
+#define CEPH_I_QUARANTINE_PENDING_BIT	(18) /* quarantine needs MDS confirmation */
 
 #define CEPH_I_DIR_ORDERED		(1 << CEPH_I_DIR_ORDERED_BIT)
 #define CEPH_I_FLUSH			(1 << CEPH_I_FLUSH_BIT)
@@ -1259,6 +1288,38 @@ extern int ceph_getattr(struct mnt_idmap *idmap,
 			u32 request_mask, unsigned int flags);
 void ceph_inode_shutdown(struct inode *inode);
 
+/*
+ * CEPH_Q_CLEAR: cached ancestry is clear, or the MDS confirmed read
+ * access to this exact inode for the current epoch. Writers must use
+ * ceph_inode_quarantine_write_state(), which ignores read receipts.
+ * CEPH_Q_UNDER: this inode or a cached ancestor is marked.
+ * CEPH_Q_UNKNOWN: the parent chain is not fully cached. Ask the MDS.
+ * Do not treat that as quarantined and do not use cached caps.
+ */
+enum ceph_qstate {
+	CEPH_Q_CLEAR = 0,
+	CEPH_Q_UNDER,
+	CEPH_Q_UNKNOWN,
+};
+
+enum ceph_qstate ceph_inode_quarantine_state(struct inode *inode);
+enum ceph_qstate ceph_inode_quarantine_write_state(struct inode *inode);
+bool ceph_inode_quarantine_direct(struct inode *inode);
+int ceph_quarantine_revalidate(struct inode *inode);
+int ceph_snapshot_read_auth(struct inode *inode);
+void ceph_quarantine_set(struct ceph_inode_info *ci);
+bool ceph_quarantine_clear(struct ceph_inode_info *ci);
+bool ceph_quarantine_reset(struct ceph_inode_info *ci);
+void ceph_quarantine_move_start(struct inode *inode, struct inode *old_dir,
+				struct inode *new_dir);
+void ceph_quarantine_move_finish(struct inode *inode, struct inode *old_dir,
+				 struct inode *new_dir);
+void ceph_quarantine_splice_start(struct super_block *sb);
+void ceph_quarantine_splice_finish(struct super_block *sb);
+void ceph_handle_quarantine_disable(struct ceph_mds_client *mdsc,
+				    struct ceph_mds_session *session,
+				    struct ceph_msg *msg);
+
 static inline bool ceph_inode_is_shutdown(const struct inode *inode)
 {
 	unsigned long flags = READ_ONCE(ceph_inode(inode)->i_ceph_flags);
@@ -1368,11 +1429,11 @@ extern void ceph_handle_caps(struct ceph_mds_session *session,
 			     struct ceph_msg *msg);
 extern struct ceph_cap *ceph_get_cap(struct ceph_mds_client *mdsc,
 				     struct ceph_cap_reservation *ctx);
-extern void ceph_add_cap(struct inode *inode,
-			 struct ceph_mds_session *session, u64 cap_id,
-			 unsigned issued, unsigned wanted,
-			 unsigned cap, unsigned seq, u64 realmino, int flags,
-			 struct ceph_cap **new_cap);
+bool ceph_add_cap(struct inode *inode,
+		  struct ceph_mds_session *session, u64 cap_id,
+		  unsigned int issued, unsigned int wanted,
+		  unsigned int seq, unsigned int mseq, u64 realmino, int flags,
+		  struct ceph_cap **new_cap);
 
 /**
  * Determine whether __ceph_remove_cap() has been called on this #cap

[PATCH v6 5/5] ceph: fail quarantined subvolume access with EACCES

[Alex Markuze]

Cached data, metadata and immutable snapshot capabilities can satisfy
VFS operations without consulting the MDS. A quarantine-aware client
must apply the MDS's restrictions to those paths before advertising
CEPHFS_FEATURE_QUARANTINE, so denied access returns EACCES while clients
with the appropriate q/Q MDS auth caps can perform recovery.

Reject cached access to directly marked inodes. Incomplete ancestry and
inherited marks require MDS authorization. Read confirmations are scoped
to the inode, credentials and namespace generation; writes require a
mode-specific OPEN. Saved authorization requires the same even generation
and no active splice. Revalidate after cap waits and lock drops, rolling
back provisional cap references when authorization expires. Nonblocking
acquisition performs no RPC, and locked folios are released before
blocking revalidation.

Cached head metadata requires a live AUTH_SHARED cap from a quarantine-
aware authoritative MDS. Without that revocation coverage, stat,
directory and encryption-policy/nonce ioctl access forces authoritative
GETATTR. A successful reply can authorize its current operation without
granting reusable cache capabilities.

Snapshot cache access requires a connected head inode matching the
target's MDS-provided subvolume ID, a valid authoritative AUTH_SHARED cap
and no local quarantine marks. Other head ancestors, unknown membership,
32-bit clients, detached snapshots and expired caps require an MDS answer.
Connected requests encode the snapshot path; disconnected requests use
LOOKUPINO with the exact inode and snapshot ID. Check reply identity and
required payloads, and perform normal wanted-cap acquisition after the
snapshot authorization check. Installed snapshot PTEs still depend on
server notifications for revocation.

Wait interruptibly for namespace writers and retry invalidated blocking
read/readdir authorizations. Splice-owner tracking and the fill-trace
recursion guard reject self-waits. Negotiated quarantine support enables
VFS ACL cache suppression and namespace mutation bookkeeping; the first
supporting session invalidates cached ACLs, and mutation tokens remain
paired across feature activation and reconnects.

LSM label reads during inode instantiation can wait for async CREATE
completion. Select synchronous CREATE before async preparation whenever
the new inode has a security blob, including before feature negotiation
because quarantine can activate after submission. Blob-free async CREATE
and async unlink use their normal admission rules.

Gate cached xattr and mmap access, and carry inherited ACLs in the encoded
xattr blob. Failed ACL installation returns ENOMEM. Detached xattr-buffer
ownership ends when its reference is released, including allocation retry
and cleanup paths. Uncertain async mutations are submitted to the MDS.

Select fscrypt's symlink-cache opt-out so encrypted readlink, traversal
and lstat reach ->get_link() revalidation. Targets are decrypted into a
temporary allocation without installing i_link. Fresh metadata replies
also allow recovery after a lost quarantine-disable notification.

Signed-off-by: Alex Markuze <amarkuze@redhat.com>
---
 fs/ceph/acl.c        |  33 ++-
 fs/ceph/addr.c       |  72 ++++--
 fs/ceph/caps.c       | 195 ++++++++++++++-
 fs/ceph/crypto.c     |   1 +
 fs/ceph/debugfs.c    |   1 +
 fs/ceph/dir.c        |  57 +++--
 fs/ceph/file.c       |  84 ++++++-
 fs/ceph/inode.c      | 547 ++++++++++++++++++++++++++++++++++++++-----
 fs/ceph/ioctl.c      |  13 +
 fs/ceph/mds_client.c |   5 +
 fs/ceph/mds_client.h |  14 +-
 fs/ceph/super.c      |   6 +-
 fs/ceph/super.h      |  42 +++-
 fs/ceph/xattr.c      |  60 ++++-
 14 files changed, 1000 insertions(+), 130 deletions(-)

diff --git a/fs/ceph/acl.c b/fs/ceph/acl.c
index dbc64ab6030881c799fe906e3ee492f08d7752a6..14feb719a1cf0006e855f14460b74c8e1e19adb2 100644
--- a/fs/ceph/acl.c
+++ b/fs/ceph/acl.c
@@ -18,18 +18,45 @@
 #include "mds_client.h"
 
 static inline void ceph_set_cached_acl(struct inode *inode,
-					int type, struct posix_acl *acl)
+				       int type, struct posix_acl *acl)
 {
 	struct ceph_inode_info *ci = ceph_inode(inode);
 
 	spin_lock(&ci->i_ceph_lock);
-	if (__ceph_caps_issued_mask_metric(ci, CEPH_CAP_XATTR_SHARED, 0))
+	if (!READ_ONCE(ceph_sb_to_mdsc(inode->i_sb)->quarantine_supported) &&
+	    __ceph_caps_issued_mask_metric(ci, CEPH_CAP_XATTR_SHARED, 0))
 		set_cached_acl(inode, type, acl);
 	else
 		forget_cached_acl(inode, type);
 	spin_unlock(&ci->i_ceph_lock);
 }
 
+void ceph_quarantine_uncache_inode_acls(struct inode *inode)
+{
+	struct ceph_inode_info *ci = ceph_inode(inode);
+	struct posix_acl *acl, *default_acl;
+
+	spin_lock(&ci->i_ceph_lock);
+	acl = xchg(&inode->i_acl, ACL_DONT_CACHE);
+	default_acl = xchg(&inode->i_default_acl, ACL_DONT_CACHE);
+	spin_unlock(&ci->i_ceph_lock);
+	if (!is_uncached_acl(acl))
+		posix_acl_release(acl);
+	if (!is_uncached_acl(default_acl))
+		posix_acl_release(default_acl);
+}
+
+void ceph_quarantine_disable_acl_cache(struct super_block *sb)
+{
+	struct inode *inode;
+
+	/* Cover inodes cached before the first supporting session opened. */
+	spin_lock(&sb->s_inode_list_lock);
+	list_for_each_entry(inode, &sb->s_inodes, i_sb_list)
+		ceph_quarantine_uncache_inode_acls(inode);
+	spin_unlock(&sb->s_inode_list_lock);
+}
+
 struct posix_acl *ceph_get_acl(struct inode *inode, int type, bool rcu)
 {
 	struct ceph_client *cl = ceph_inode_to_client(inode);
@@ -73,6 +100,8 @@ struct posix_acl *ceph_get_acl(struct inode *inode, int type, bool rcu)
 		acl = posix_acl_from_xattr(&init_user_ns, value, size);
 	} else if (size == -ENODATA || size == 0) {
 		acl = NULL;
+	} else if (size == -EACCES) {
+		acl = ERR_PTR(-EACCES);
 	} else {
 		pr_err_ratelimited_client(cl, "%llx.%llx failed, err=%d\n",
 					  ceph_vinop(inode), size);
diff --git a/fs/ceph/addr.c b/fs/ceph/addr.c
index dd2bd8ed242499183d6d3acd250affc56095b5ca..79e396752601f5bffb9f4cdc2150d21ce37ab701 100644
--- a/fs/ceph/addr.c
+++ b/fs/ceph/addr.c
@@ -540,6 +540,10 @@ static int ceph_init_request(struct netfs_io_request *rreq, struct file *file)
 			want |= CEPH_CAP_FILE_LAZYIO;
 	}
 
+	/*
+	 * NETFS_READPAGE is synchronous. ceph_read_folio() holds caps for
+	 * its duration, including the authorization for a snapshot read.
+	 */
 	if (rreq->origin == NETFS_READAHEAD) {
 		/*
 		 * readahead callers do not necessarily hold Fcb caps
@@ -563,21 +567,6 @@ static int ceph_init_request(struct netfs_io_request *rreq, struct file *file)
 			ret = -EACCES;
 			goto out;
 		}
-	} else {
-		/*
-		 * Make sure we have caps, just in case read_folio was
-		 * called directly (e.g. by erofs) which may have
-		 * bypassed the usual Ceph cap checks.  This branch
-		 * uses __ceph_get_caps() which blocks until the
-		 * required cap has been granted.
-		 */
-		ret = __ceph_get_caps(inode, fi, CEPH_CAP_FILE_RD, want, -1,
-				      &got);
-		if (ret < 0) {
-			boutc(cl, "%llx.%llx, error getting cap\n",
-			      ceph_vinop(inode));
-			goto out;
-		}
 	}
 
 	priv->caps = got;
@@ -2069,8 +2058,44 @@ static int ceph_write_end(const struct kiocb *iocb,
 	return copied;
 }
 
+/* Direct read_folio callers may not have entered through read_iter or fault. */
+static int ceph_read_folio(struct file *file, struct folio *folio)
+{
+	struct address_space *mapping = folio->mapping;
+	struct inode *inode = mapping->host;
+	struct ceph_file_info *fi = file ? file->private_data : NULL;
+	int want = CEPH_CAP_FILE_CACHE;
+	int got, ret;
+
+	if (fi && ceph_find_rw_context(fi))
+		return netfs_read_folio(file, folio);
+
+	if (fi && (fi->fmode & CEPH_FILE_MODE_LAZY))
+		want |= CEPH_CAP_FILE_LAZYIO;
+
+	/* A blocking getattr can fill inline data into this folio. */
+	folio_unlock(folio);
+	got = 0;
+	ret = __ceph_get_caps(inode, fi, CEPH_CAP_FILE_RD, want, -1, &got);
+	if (ret < 0)
+		return ret;
+
+	folio_lock(folio);
+	if (folio->mapping != mapping) {
+		folio_unlock(folio);
+		ret = AOP_TRUNCATED_PAGE;
+	} else if (folio_test_uptodate(folio)) {
+		folio_unlock(folio);
+		ret = 0;
+	} else {
+		ret = netfs_read_folio(file, folio);
+	}
+	ceph_put_cap_refs(ceph_inode(inode), got);
+	return ret;
+}
+
 const struct address_space_operations ceph_aops = {
-	.read_folio = netfs_read_folio,
+	.read_folio = ceph_read_folio,
 	.readahead = netfs_readahead,
 	.writepages = ceph_writepages_start,
 	.write_begin = ceph_write_begin,
@@ -2649,9 +2674,24 @@ static const struct vm_operations_struct ceph_vmops = {
 int ceph_mmap_prepare(struct vm_area_desc *desc)
 {
 	struct address_space *mapping = desc->file->f_mapping;
+	struct inode *inode = file_inode(desc->file);
+	int err, got = 0;
 
 	if (!mapping->a_ops->read_folio)
 		return -ENOEXEC;
+	err = ceph_quarantine_revalidate(inode);
+	if (err)
+		return err;
+	if (vma_desc_test_all(desc, VMA_SHARED_BIT, VMA_WRITE_BIT) &&
+	    ceph_inode_quarantine_write_state(inode) != CEPH_Q_CLEAR) {
+		/* A read receipt cannot admit a writable shared mapping. */
+		err = ceph_get_caps(desc->file, CEPH_CAP_FILE_WR, 0, -1, &got);
+		if (err)
+			return err;
+		ceph_put_cap_refs(ceph_inode(inode), got);
+	}
+	if (ceph_inode_quarantine_direct(inode))
+		return -EACCES;
 	desc->vm_ops = &ceph_vmops;
 	return 0;
 }
diff --git a/fs/ceph/caps.c b/fs/ceph/caps.c
index b07814d0038cd72a6fc10813f50846df58e9591a..7c85e84fa6493079244e56fb251af4c2740a47ee 100644
--- a/fs/ceph/caps.c
+++ b/fs/ceph/caps.c
@@ -3007,10 +3007,39 @@ enum {
 	/* first 8 bits are reserved for CEPH_FILE_MODE_FOO */
 	NON_BLOCKING	= (1 << 8),
 	CHECK_FILELOCK	= (1 << 9),
+	MDS_AUTHORIZED	= (1 << 10),
 };
 
+bool ceph_quarantine_head_cap_valid(struct inode *inode)
+{
+	struct ceph_inode_info *ci = ceph_inode(inode);
+	struct ceph_cap *cap;
+	bool valid;
+
+	spin_lock(&ci->i_ceph_lock);
+	cap = ci->i_auth_cap;
+	valid = cap && __cap_is_valid(ci, cap) &&
+		test_bit(CEPHFS_FEATURE_QUARANTINE, &cap->session->s_features) &&
+		(cap->issued & CEPH_CAP_AUTH_SHARED);
+	spin_unlock(&ci->i_ceph_lock);
+	return valid;
+}
+
+static bool ceph_cap_auth_valid(struct ceph_mds_client *mdsc, u64 epoch)
+{
+	/*
+	 * Order authorization and cap observations before checking the saved
+	 * generation. Moves and splices can invalidate it even with no marks.
+	 * The epoch updates in the writer and splice boundaries are ordered RMWs.
+	 */
+	smp_rmb();
+	return !(epoch & 1) &&
+	       (u64)atomic64_read_acquire(&mdsc->quarantine_epoch) == epoch &&
+	       !atomic_read(&mdsc->quarantine_splices);
+}
+
 static int try_get_cap_refs(struct inode *inode, int need, int want,
-			    loff_t endoff, int flags, int *got)
+			    loff_t endoff, int flags, u64 qepoch, int *got)
 {
 	struct ceph_inode_info *ci = ceph_inode(inode);
 	struct ceph_mds_client *mdsc = ceph_inode_to_fs_client(inode)->mdsc;
@@ -3026,6 +3055,15 @@ static int try_get_cap_refs(struct inode *inode, int need, int want,
 again:
 	spin_lock(&ci->i_ceph_lock);
 
+	if (ceph_inode_quarantine_direct(inode)) {
+		ret = -EACCES;
+		goto out_unlock;
+	}
+	if ((flags & MDS_AUTHORIZED) && !ceph_cap_auth_valid(mdsc, qepoch)) {
+		ret = -EAGAIN;
+		goto out_unlock;
+	}
+
 	if ((flags & CHECK_FILELOCK) &&
 	    test_bit(CEPH_I_ERROR_FILELOCK_BIT, &ci->i_ceph_flags)) {
 		doutc(cl, "%p %llx.%llx error filelock\n", inode,
@@ -3047,6 +3085,15 @@ static int try_get_cap_refs(struct inode *inode, int need, int want,
 		}
 		__ceph_do_pending_vmtruncate(inode);
 		spin_lock(&ci->i_ceph_lock);
+		if (ceph_inode_quarantine_direct(inode)) {
+			ret = -EACCES;
+			goto out_unlock;
+		}
+		if ((flags & MDS_AUTHORIZED) &&
+		    !ceph_cap_auth_valid(mdsc, qepoch)) {
+			ret = -EAGAIN;
+			goto out_unlock;
+		}
 	}
 
 	have = __ceph_caps_issued(ci, &implemented);
@@ -3195,6 +3242,31 @@ static int try_get_cap_refs(struct inode *inode, int need, int want,
 	if (snap_rwsem_locked)
 		up_read(&mdsc->snap_rwsem);
 
+	/* References remain provisional across truncate, cap waits and lock drops. */
+	if (ret == 1) {
+		enum ceph_qstate qstate = CEPH_Q_CLEAR;
+
+		if (!(flags & MDS_AUTHORIZED) &&
+		    ceph_snapshot_needs_revalidate(inode))
+			qstate = CEPH_Q_UNKNOWN;
+		else if (!(flags & MDS_AUTHORIZED))
+			qstate = need & CEPH_CAP_FILE_WR ?
+				ceph_inode_quarantine_write_state(inode) :
+				ceph_inode_quarantine_state(inode);
+		else if (!ceph_cap_auth_valid(mdsc, qepoch))
+			qstate = CEPH_Q_UNKNOWN;
+		/* An MDS answer cannot override a new negative on this inode. */
+		if (ceph_inode_quarantine_direct(inode))
+			qstate = CEPH_Q_UNDER;
+
+		if (qstate != CEPH_Q_CLEAR) {
+			ceph_put_cap_refs(ci, *got);
+			*got = 0;
+			ret = qstate == CEPH_Q_UNDER &&
+			      ceph_inode_quarantine_direct(inode) ? -EACCES : -EAGAIN;
+		}
+	}
+
 	if (!ret)
 		ceph_update_cap_mis(&mdsc->metric);
 	else if (ret == 1)
@@ -3263,17 +3335,34 @@ int ceph_try_get_caps(struct inode *inode, int need, int want,
 	BUG_ON(want & ~(CEPH_CAP_FILE_CACHE | CEPH_CAP_FILE_LAZYIO |
 			CEPH_CAP_FILE_SHARED | CEPH_CAP_FILE_EXCL |
 			CEPH_CAP_ANY_DIR_OPS));
+	/* Immutable snapshot caps have no MDS revoke notifications. */
+	if (ceph_snapshot_needs_revalidate(inode))
+		return 0;
+
 	if (need) {
 		ret = ceph_pool_perm_check(inode, need);
 		if (ret < 0)
 			return ret;
 	}
 
+	switch (ceph_inode_quarantine_state(inode)) {
+	case CEPH_Q_UNDER:
+		return ceph_inode_quarantine_direct(inode) ? -EACCES : 0;
+	case CEPH_Q_UNKNOWN:
+		/*
+		 * Callers may hold locked folios or another inode's caps.
+		 * Leave blocking revalidation to ceph_get_caps().
+		 */
+		return 0;
+	default:
+		break;
+	}
+
 	flags = get_used_fmode(need | want);
 	if (nonblock)
 		flags |= NON_BLOCKING;
 
-	ret = try_get_cap_refs(inode, need, want, 0, flags, got);
+	ret = try_get_cap_refs(inode, need, want, 0, flags, 0, got);
 	/* three special error codes */
 	if (ret == -EAGAIN || ret == -EFBIG || ret == -EUCLEAN)
 		ret = 0;
@@ -3290,6 +3379,9 @@ int __ceph_get_caps(struct inode *inode, struct ceph_file_info *fi, int need,
 {
 	struct ceph_inode_info *ci = ceph_inode(inode);
 	struct ceph_fs_client *fsc = ceph_inode_to_fs_client(inode);
+	struct ceph_mds_client *mdsc = fsc->mdsc;
+	bool snapshot = ceph_in_snap(inode) && ceph_snap(inode) != CEPH_SNAPDIR;
+	u64 qepoch = 0;
 	int ret, _got, flags;
 
 	ret = ceph_pool_perm_check(inode, need);
@@ -3304,15 +3396,71 @@ int __ceph_get_caps(struct inode *inode, struct ceph_file_info *fi, int need,
 
 	while (true) {
 		flags &= CEPH_FILE_MODE_MASK;
+		if (snapshot && ceph_snapshot_needs_revalidate(inode)) {
+			/*
+			 * i_snap_caps cannot be revoked, and an idle snapshot
+			 * inode may receive no quarantine notification at all.
+			 * An MDS answer authorizes only this acquisition.
+			 */
+			qepoch = atomic64_read_acquire(&mdsc->quarantine_epoch);
+			if (!ceph_cap_auth_valid(mdsc, qepoch)) {
+				ret = ceph_quarantine_wait(inode);
+				if (ret)
+					return ret;
+				continue;
+			}
+			ret = ceph_snapshot_read_auth(inode);
+			if (ret)
+				return ret;
+			flags |= MDS_AUTHORIZED;
+		} else {
+			enum ceph_qstate qstate = need & CEPH_CAP_FILE_WR ?
+				ceph_inode_quarantine_write_state(inode) :
+				ceph_inode_quarantine_state(inode);
+
+			switch (qstate) {
+			case CEPH_Q_UNDER:
+				if (ceph_inode_quarantine_direct(inode))
+					return -EACCES;
+				fallthrough;
+			case CEPH_Q_UNKNOWN:
+				if (!(need & CEPH_CAP_FILE_WR)) {
+					ret = ceph_quarantine_revalidate(inode);
+					if (ret)
+						return ret;
+					continue;
+				}
+				/* A GETATTR receipt cannot authorize writes. */
+				qepoch = atomic64_read_acquire(&mdsc->quarantine_epoch);
+				if (!ceph_cap_auth_valid(mdsc, qepoch)) {
+					ret = ceph_quarantine_wait(inode);
+					if (ret)
+						return ret;
+					continue;
+				}
+				ret = ceph_renew_caps(inode, flags, true);
+				if (ret)
+					return ret;
+				flags |= MDS_AUTHORIZED;
+				break;
+			default:
+				break;
+			}
+		}
+
 		if (vfs_inode_has_locks(inode))
 			flags |= CHECK_FILELOCK;
 		_got = 0;
 		ret = try_get_cap_refs(inode, need, want, endoff,
-				       flags, &_got);
-		WARN_ON_ONCE(ret == -EAGAIN);
+				       flags, qepoch, &_got);
+		if (ret == -EAGAIN) {
+			ret = ceph_quarantine_wait(inode);
+			if (ret)
+				return ret;
+			continue;
+		}
 		if (!ret) {
 #ifdef CONFIG_DEBUG_FS
-			struct ceph_mds_client *mdsc = fsc->mdsc;
 			struct cap_wait cw;
 #endif
 			DEFINE_WAIT_FUNC(wait, woken_wake_function);
@@ -3333,8 +3481,33 @@ int __ceph_get_caps(struct inode *inode, struct ceph_file_info *fi, int need,
 			add_wait_queue(&ci->i_cap_wq, &wait);
 
 			flags |= NON_BLOCKING;
-			while (!(ret = try_get_cap_refs(inode, need, want,
-							endoff, flags, &_got))) {
+			while (true) {
+				enum ceph_qstate qstate = CEPH_Q_CLEAR;
+
+				if (!(flags & MDS_AUTHORIZED))
+					qstate = need & CEPH_CAP_FILE_WR ?
+						ceph_inode_quarantine_write_state(inode) :
+						ceph_inode_quarantine_state(inode);
+				else if (!ceph_cap_auth_valid(mdsc, qepoch))
+					qstate = CEPH_Q_UNKNOWN;
+				if (ceph_inode_quarantine_direct(inode))
+					qstate = CEPH_Q_UNDER;
+				switch (qstate) {
+				case CEPH_Q_UNDER:
+					ret = ceph_inode_quarantine_direct(inode) ?
+						-EACCES : -EAGAIN;
+					break;
+				case CEPH_Q_UNKNOWN:
+					/* Revalidate outside the cap wait. */
+					ret = -EAGAIN;
+					break;
+				default:
+					ret = try_get_cap_refs(inode, need, want,
+							       endoff, flags, qepoch, &_got);
+					break;
+				}
+				if (ret)
+					break;
 				if (signal_pending(current)) {
 					ret = -ERESTARTSYS;
 					break;
@@ -3363,8 +3536,12 @@ int __ceph_get_caps(struct inode *inode, struct ceph_file_info *fi, int need,
 			spin_unlock(&mdsc->caps_list_lock);
 #endif
 
-			if (ret == -EAGAIN)
+			if (ret == -EAGAIN) {
+				ret = ceph_quarantine_wait(inode);
+				if (ret)
+					return ret;
 				continue;
+			}
 		}
 
 		if (fi && (fi->fmode & CEPH_FILE_MODE_WR) &&
@@ -3387,7 +3564,7 @@ int __ceph_get_caps(struct inode *inode, struct ceph_file_info *fi, int need,
 			if (ret == -EUCLEAN) {
 				/* session was killed or a waited cap
 				 * request needs a retry */
-				ret = ceph_renew_caps(inode, flags);
+				ret = ceph_renew_caps(inode, flags, false);
 				if (ret == 0)
 					continue;
 			}
diff --git a/fs/ceph/crypto.c b/fs/ceph/crypto.c
index 9e0ebb0718404fdb6e9ec1be2fac898158a3f5ae..de786a0e626c763787bfafda59cc42e7b7ff4f5c 100644
--- a/fs/ceph/crypto.c
+++ b/fs/ceph/crypto.c
@@ -92,6 +92,7 @@ static struct fscrypt_operations ceph_fscrypt_ops = {
 	.inode_info_offs	= (int)offsetof(struct ceph_inode_info, i_crypt_info) -
 				  (int)offsetof(struct ceph_inode_info, netfs.inode),
 	.needs_bounce_pages	= 1,
+	.no_symlink_cache	= 1,
 	.get_context		= ceph_crypt_get_context,
 	.set_context		= ceph_crypt_set_context,
 	.get_dummy_policy	= ceph_get_dummy_policy,
diff --git a/fs/ceph/debugfs.c b/fs/ceph/debugfs.c
index 6a9a5551d9d71d5adf2349daad01ee1adedd1a82..ef10e948a2d3427af9b0f532bebf3aa119f8de84 100644
--- a/fs/ceph/debugfs.c
+++ b/fs/ceph/debugfs.c
@@ -53,6 +53,7 @@ static const struct ceph_session_feature_desc ceph_session_feature_table[] = {
 	{ CEPHFS_FEATURE_HAS_OWNER_UIDGID, "HAS_OWNER_UIDGID" },
 	{ CEPHFS_FEATURE_MDS_AUTH_CAPS_CHECK, "MDS_AUTH_CAPS_CHECK" },
 	{ CEPHFS_FEATURE_SUBVOLUME_METRICS, "SUBVOLUME_METRICS" },
+	{ CEPHFS_FEATURE_QUARANTINE, "QUARANTINE" },
 };
 
 static int mdsmap_show(struct seq_file *s, void *p)
diff --git a/fs/ceph/dir.c b/fs/ceph/dir.c
index d8c8b7ad893092611014b6094864a14bd36aecca..d0687e731dea633d0d791cb416f6a72cfd19b170 100644
--- a/fs/ceph/dir.c
+++ b/fs/ceph/dir.c
@@ -349,6 +349,13 @@ static int ceph_readdir(struct file *file, struct dir_context *ctx)
 
 	ceph_blog_enter(fsc, &__ji);
 
+	/* Validate before returning cached entries, dots, or end-of-directory. */
+	err = ceph_quarantine_revalidate(inode);
+	if (err) {
+		ceph_blog_exit(&__ji);
+		return err;
+	}
+
 	boutc(cl, "%p %llx.%llx file %p pos %llx\n", inode,
 	      ceph_vinop(inode), file, ctx->pos);
 	if (dfi->file_info.flags & CEPH_F_ATEND) {
@@ -785,6 +792,7 @@ static loff_t ceph_dir_llseek(struct file *file, loff_t offset, int whence)
 struct dentry *ceph_handle_snapdir(struct ceph_mds_request *req,
 				   struct dentry *dentry)
 {
+	struct ceph_quarantine_splice qsplice;
 	struct ceph_fs_client *fsc = ceph_sb_to_fs_client(dentry->d_sb);
 	struct inode *parent = d_inode(dentry->d_parent); /* we hold i_rwsem */
 	struct ceph_client *cl = ceph_inode_to_client(parent);
@@ -795,9 +803,9 @@ struct dentry *ceph_handle_snapdir(struct ceph_mds_request *req,
 		struct dentry *res;
 		struct inode *inode = ceph_get_snapdir(parent);
 
-		ceph_quarantine_splice_start(parent->i_sb);
+		ceph_quarantine_splice_start(parent->i_sb, &qsplice);
 		res = d_splice_alias(inode, dentry);
-		ceph_quarantine_splice_finish(parent->i_sb);
+		ceph_quarantine_splice_finish(&qsplice);
 		boutc_formats(cl,
 			      "ENOENT on snapdir %p '%s', linking to snapdir %p %llx.%llx. Spliced dentry %p\n",
 			      "ENOENT on snapdir %p '%pd', linking to snapdir %p %llx.%llx. Spliced dentry %p\n",
@@ -904,8 +912,13 @@ static struct dentry *ceph_lookup(struct inode *dir, struct dentry *dentry,
 			ceph_dir_clear_complete(dir);
 	}
 
-	/* can we conclude ENOENT locally? */
-	if (d_really_is_negative(dentry)) {
+	/*
+	 * A marked directory, or one whose parents are not cached,
+	 * must not answer from the complete-dir cache. The MDS reply
+	 * is what clears a lost disable.
+	 */
+	if (d_really_is_negative(dentry) &&
+	    ceph_inode_quarantine_state(dir) == CEPH_Q_CLEAR) {
 		struct ceph_inode_info *ci = ceph_inode(dir);
 		struct ceph_dentry_info *di = ceph_dentry(dentry);
 
@@ -1068,9 +1081,7 @@ static int ceph_mknod(struct mnt_idmap *idmap, struct inode *dir,
 out_req:
 	ceph_mdsc_put_request(req);
 out:
-	if (!err)
-		ceph_init_inode_acls(d_inode(dentry), &as_ctx);
-	else
+	if (err)
 		d_drop(dentry);
 	ceph_release_acl_sec_ctx(&as_ctx);
 	ceph_blog_exit(&__ji);
@@ -1297,13 +1308,8 @@ static struct dentry *ceph_mkdir(struct mnt_idmap *idmap, struct inode *dir,
 		ret = dget(req->r_dentry);
 	ceph_mdsc_put_request(req);
 out:
-	if (!IS_ERR(ret)) {
-		if (ret)
-			dentry = ret;
-		ceph_init_inode_acls(d_inode(dentry), &as_ctx);
-	} else {
+	if (IS_ERR(ret))
 		d_drop(dentry);
-	}
 	ceph_release_acl_sec_ctx(&as_ctx);
 	ceph_blog_exit(&__ji);
 	return ret;
@@ -1312,6 +1318,7 @@ static struct dentry *ceph_mkdir(struct mnt_idmap *idmap, struct inode *dir,
 static int ceph_link(struct dentry *old_dentry, struct inode *dir,
 		     struct dentry *dentry)
 {
+	struct ceph_quarantine_splice qsplice;
 	struct ceph_mds_client *mdsc = ceph_sb_to_mdsc(dir->i_sb);
 	struct ceph_fs_client *fsc = mdsc->fsc;
 	struct ceph_client *cl = mdsc->fsc->client;
@@ -1375,9 +1382,9 @@ static int ceph_link(struct dentry *old_dentry, struct inode *dir,
 		d_drop(dentry);
 	} else if (!req->r_reply_info.head->is_dentry) {
 		ihold(d_inode(old_dentry));
-		ceph_quarantine_splice_start(dir->i_sb);
+		ceph_quarantine_splice_start(dir->i_sb, &qsplice);
 		d_instantiate(dentry, d_inode(old_dentry));
-		ceph_quarantine_splice_finish(dir->i_sb);
+		ceph_quarantine_splice_finish(&qsplice);
 	}
 	ceph_mdsc_put_request(req);
 	ceph_blog_exit(&__ji);
@@ -1441,8 +1448,17 @@ static int get_caps_for_async_unlink(struct inode *dir, struct dentry *dentry)
 {
 	struct ceph_inode_info *ci = ceph_inode(dir);
 	struct ceph_dentry_info *di;
+	struct inode *inode = d_inode(dentry);
 	int got = 0, want = CEPH_CAP_FILE_EXCL | CEPH_CAP_DIR_UNLINK;
 
+	/*
+	 * Async unlink completes in the cache. A marked inode, or one
+	 * whose parents are not cached, has to go to the MDS.
+	 */
+	if (ceph_inode_quarantine_write_state(dir) != CEPH_Q_CLEAR ||
+	    (inode && ceph_inode_quarantine_write_state(inode) != CEPH_Q_CLEAR))
+		return 0;
+
 	spin_lock(&ci->i_ceph_lock);
 	if ((__ceph_caps_issued(ci, NULL) & want) == want) {
 		ceph_take_cap_refs(ci, want, false);
@@ -1627,6 +1643,7 @@ static int ceph_rename(struct mnt_idmap *idmap, struct inode *old_dir,
 		       struct dentry *old_dentry, struct inode *new_dir,
 		       struct dentry *new_dentry, unsigned int flags)
 {
+	struct ceph_mds_client *qmdsc;
 	struct ceph_mds_client *mdsc = ceph_sb_to_mdsc(old_dir->i_sb);
 	struct ceph_fs_client *fsc = mdsc->fsc;
 	struct ceph_client *cl = mdsc->fsc->client;
@@ -1713,9 +1730,9 @@ static int ceph_rename(struct mnt_idmap *idmap, struct inode *old_dir,
 		 * to do it here.
 		 */
 		moved = d_inode(old_dentry);
-		ceph_quarantine_move_start(moved, old_dir, new_dir);
+		qmdsc = ceph_quarantine_move_start(moved, old_dir, new_dir);
 		d_move(old_dentry, new_dentry);
-		ceph_quarantine_move_finish(moved, old_dir, new_dir);
+		ceph_quarantine_move_finish(qmdsc);
 	}
 	ceph_mdsc_put_request(req);
 	ceph_blog_exit(&__ji);
@@ -2349,12 +2366,16 @@ static ssize_t ceph_read_dir(struct file *file, char __user *buf, size_t size,
 	struct ceph_dir_file_info *dfi = file->private_data;
 	struct inode *inode = file_inode(file);
 	struct ceph_inode_info *ci = ceph_inode(inode);
-	int left;
+	int left, err;
 	const int bufsize = 1024;
 
 	if (!ceph_test_mount_opt(ceph_sb_to_fs_client(inode->i_sb), DIRSTAT))
 		return -EISDIR;
 
+	err = ceph_quarantine_revalidate(inode);
+	if (err)
+		return err;
+
 	if (!dfi->dir_info) {
 		dfi->dir_info = kmalloc(bufsize, GFP_KERNEL);
 		if (!dfi->dir_info)
diff --git a/fs/ceph/file.c b/fs/ceph/file.c
index 9a6eadb94ab4f7e67d565b258e780129d9a505ff..367bb1dcc4a803ed8478ae9f81f55cb512cca8eb 100644
--- a/fs/ceph/file.c
+++ b/fs/ceph/file.c
@@ -297,8 +297,10 @@ static int ceph_init_file(struct inode *inode, struct file *file, int fmode)
 
 /*
  * Retry cap acquisition after a stale session or a lost cap update.
+ * A forced request checks this operation's access on the MDS even when
+ * another opener already holds the requested caps.
  */
-int ceph_renew_caps(struct inode *inode, int fmode)
+int ceph_renew_caps(struct inode *inode, int fmode, bool force)
 {
 	struct ceph_mds_client *mdsc = ceph_sb_to_mdsc(inode->i_sb);
 	struct ceph_client *cl = mdsc->fsc->client;
@@ -308,9 +310,9 @@ int ceph_renew_caps(struct inode *inode, int fmode)
 
 	spin_lock(&ci->i_ceph_lock);
 	__ceph_touch_fmode(ci, mdsc, fmode);
-	wanted = __ceph_caps_file_wanted(ci);
+	wanted = force ? ceph_caps_for_mode(fmode) : __ceph_caps_file_wanted(ci);
 	issued = __ceph_caps_issued(ci, NULL);
-	if (__ceph_is_any_real_caps(ci) &&
+	if (!force && __ceph_is_any_real_caps(ci) &&
 	    (!(wanted & CEPH_CAP_ANY_WR) || ci->i_auth_cap) &&
 	    (issued & wanted) == wanted) {
 		spin_unlock(&ci->i_ceph_lock);
@@ -371,7 +373,9 @@ int ceph_open(struct inode *inode, struct file *file)
 	struct dentry *dentry;
 	char *path;
 	bool do_sync = false;
+	bool snapshot = ceph_in_snap(inode) && ceph_snap(inode) != CEPH_SNAPDIR;
 	int mask = MAY_READ;
+	enum ceph_qstate qstate;
 	struct ceph_journal_info __ji;
 
 	ceph_blog_enter(fsc, &__ji);
@@ -382,6 +386,23 @@ int ceph_open(struct inode *inode, struct file *file)
 		return 0;
 	}
 
+	/*
+	 * Existing caps can satisfy open without an MDS round-trip.
+	 * A short walk cannot.
+	 */
+	if (!snapshot) {
+		qstate = file->f_mode & FMODE_WRITE ?
+			ceph_inode_quarantine_write_state(inode) :
+			ceph_inode_quarantine_state(inode);
+		if (qstate == CEPH_Q_UNDER &&
+		    ceph_inode_quarantine_direct(inode)) {
+			ceph_blog_exit(&__ji);
+			return -EACCES;
+		}
+		if (qstate != CEPH_Q_CLEAR)
+			do_sync = true;
+	}
+
 	/* filter out O_CREAT|O_EXCL; vfs did that already.  yuck. */
 	flags = file->f_flags & ~(O_CREAT|O_EXCL);
 	if (S_ISDIR(inode->i_mode)) {
@@ -419,6 +440,7 @@ int ceph_open(struct inode *inode, struct file *file)
 		do_sync = true;
 	} else {
 		struct ceph_path_info path_info = {0};
+
 		path = ceph_mdsc_build_path(mdsc, dentry, &path_info, 0);
 		if (IS_ERR(path)) {
 			do_sync = true;
@@ -429,7 +451,7 @@ int ceph_open(struct inode *inode, struct file *file)
 		ceph_mdsc_free_path_info(&path_info);
 		dput(dentry);
 
-		/* For none EACCES cases will let the MDS do the mds auth check */
+		/* Let the MDS handle errors other than EACCES. */
 		if (err == -EACCES) {
 			ceph_blog_exit(&__ji);
 			return err;
@@ -445,12 +467,23 @@ int ceph_open(struct inode *inode, struct file *file)
 		return -EROFS;
 	}
 
-	/* trivially open snapdir */
+	/* A synthetic snapdir must be validated through its real parent. */
 	if (ceph_snap(inode) == CEPH_SNAPDIR) {
+		err = ceph_quarantine_revalidate(inode);
 		ceph_blog_exit(&__ji);
+		if (err)
+			return err;
 		return ceph_init_file(inode, file, fmode);
 	}
 
+	if (snapshot) {
+		err = ceph_snapshot_read_auth(inode);
+		if (err) {
+			ceph_blog_exit(&__ji);
+			return err;
+		}
+	}
+
 	/*
 	 * No need to block if we have caps on the auth MDS (for
 	 * write) or any MDS (for read).  Update wanted set
@@ -541,6 +574,13 @@ static int try_prep_async_create(struct inode *dir, struct dentry *dentry,
 	int got = 0, want = CEPH_CAP_FILE_EXCL | CEPH_CAP_DIR_CREATE;
 	u64 ino;
 
+	/*
+	 * Async create completes in the cache. A marked directory, or
+	 * one whose parents are not cached, has to go to the MDS.
+	 */
+	if (ceph_inode_quarantine_write_state(dir) != CEPH_Q_CLEAR)
+		return 0;
+
 	spin_lock(&ci->i_ceph_lock);
 	/* No auth cap means no chance for Dc caps */
 	if (!ci->i_auth_cap)
@@ -687,6 +727,7 @@ static int ceph_finish_async_create(struct inode *dir, struct inode *inode,
 				    struct ceph_acl_sec_ctx *as_ctx,
 				    struct ceph_file_layout *lo)
 {
+	struct ceph_quarantine_splice qsplice;
 	int ret;
 	char xattr_buf[4];
 	struct ceph_mds_reply_inode in = { };
@@ -785,9 +826,9 @@ static int ceph_finish_async_create(struct inode *dir, struct inode *inode,
 		if (d_in_lookup(dentry) || d_really_is_negative(dentry)) {
 			if (!d_unhashed(dentry))
 				d_drop(dentry);
-			ceph_quarantine_splice_start(dir->i_sb);
+			ceph_quarantine_splice_start(dir->i_sb, &qsplice);
 			dn = d_splice_alias(inode, dentry);
-			ceph_quarantine_splice_finish(dir->i_sb);
+			ceph_quarantine_splice_finish(&qsplice);
 			WARN_ON_ONCE(dn && dn != dentry);
 		}
 		file->f_mode |= FMODE_CREATED;
@@ -889,6 +930,14 @@ int ceph_atomic_open(struct inode *dir, struct dentry *dentry,
 			err = PTR_ERR(new_inode);
 			goto out_ctx;
 		}
+		/*
+		 * LSM label reads can wait on async create completion. Quarantine
+		 * may activate after submission, so keep blob-bearing creates
+		 * synchronous even before support is negotiated.
+		 */
+		if (ceph_security_xattr_wanted(new_inode))
+			try_async = false;
+
 		/* Async create can't handle more than a page of xattrs */
 		if (as_ctx.pagelist &&
 		    !list_is_singular(&as_ctx.pagelist->head))
@@ -2972,6 +3021,27 @@ static int get_rd_wr_caps(struct file *src_filp, int *src_got,
 	int ret = 0;
 	bool retrying = false;
 
+	if (ceph_in_snap(file_inode(src_filp)) &&
+	    ceph_snap(file_inode(src_filp)) != CEPH_SNAPDIR) {
+		/*
+		 * Snapshot caps have no MDS revocation state, so holding
+		 * them cannot prevent a destination cap grant. Validate
+		 * the snapshot before taking any destination references.
+		 */
+		ret = ceph_get_caps(src_filp, CEPH_CAP_FILE_RD,
+				    CEPH_CAP_FILE_SHARED, -1, src_got);
+		if (ret < 0)
+			return ret;
+		ret = ceph_get_caps(dst_filp, CEPH_CAP_FILE_WR,
+				    CEPH_CAP_FILE_BUFFER, dst_endoff, dst_got);
+		if (ret < 0) {
+			ceph_put_cap_refs(ceph_inode(file_inode(src_filp)),
+					  *src_got);
+			*src_got = 0;
+		}
+		return ret;
+	}
+
 retry_caps:
 	ret = ceph_get_caps(dst_filp, CEPH_CAP_FILE_WR, CEPH_CAP_FILE_BUFFER,
 			    dst_endoff, dst_got);
diff --git a/fs/ceph/inode.c b/fs/ceph/inode.c
index 048e0d502eb4ea361389bc4a6e5ddfa9fd9f6ac5..df69feaa8fba31dd623c12a1fa7d1b247a5e1649 100644
--- a/fs/ceph/inode.c
+++ b/fs/ceph/inode.c
@@ -60,6 +60,13 @@ static int ceph_set_ino_cb(struct inode *inode, void *data)
 #endif
 
 	inode_set_iversion_raw(inode, 0);
+#ifdef CONFIG_CEPH_FS_POSIX_ACL
+	/* The VFS ACL cache would bypass Ceph's quarantine checks. */
+	if (READ_ONCE(mdsc->quarantine_supported)) {
+		inode->i_acl = ACL_DONT_CACHE;
+		inode->i_default_acl = ACL_DONT_CACHE;
+	}
+#endif
 	percpu_counter_inc(&mdsc->metric.total_inodes);
 
 	return 0;
@@ -252,6 +259,8 @@ struct inode *ceph_get_snapdir(struct inode *parent)
 	inode_set_atime_to_ts(inode, inode_get_atime(parent));
 	ci->i_rbytes = 0;
 	ci->i_btime = ceph_inode(parent)->i_btime;
+	ceph_inode_set_subvolume(inode,
+				 READ_ONCE(ceph_inode(parent)->i_subvolume_id));
 
 #ifdef CONFIG_FS_ENCRYPTION
 	/* if encrypted, just borrow fscrypt_auth from parent */
@@ -768,6 +777,17 @@ static void ceph_quarantine_write_end(struct ceph_mds_client *mdsc)
 	if (atomic_dec_return(&mdsc->quarantine_writers) == 0)
 		atomic64_inc_return(&mdsc->quarantine_epoch);
 	spin_unlock(&mdsc->quarantine_lock);
+	wake_up_all(&mdsc->quarantine_waitq);
+}
+
+/* Called under mdsc->mutex before a supporting session starts its requests. */
+void ceph_quarantine_enable(struct ceph_mds_client *mdsc)
+{
+	if (READ_ONCE(mdsc->quarantine_supported))
+		return;
+
+	WRITE_ONCE(mdsc->quarantine_supported, true);
+	ceph_quarantine_disable_acl_cache(mdsc->fsc->sb);
 }
 
 void ceph_quarantine_set(struct ceph_inode_info *ci)
@@ -890,6 +910,46 @@ static bool ceph_qmds_confirmed(struct ceph_inode_info *ci, u64 epoch)
 	return confirmed;
 }
 
+static void ceph_quarantine_remember_clear(struct inode *inode, u64 epoch,
+					   const struct cred *cred)
+{
+	struct ceph_mds_client *mdsc = ceph_sb_to_mdsc(inode->i_sb);
+	struct ceph_inode_info *ci = ceph_inode(inode);
+	const struct cred *old = NULL;
+
+	if (!mdsc || atomic_read(&mdsc->quarantine_count) == 0)
+		return;
+
+	/* Serialize the credential and epoch pair against other receipts. */
+	spin_lock(&ci->i_ceph_lock);
+	if (atomic_read(&mdsc->quarantine_count) == 0 ||
+	    ceph_inode_quarantine_direct(inode) || (epoch & 1) ||
+	    atomic_read(&mdsc->quarantine_splices) ||
+	    (u64)atomic64_read_acquire(&mdsc->quarantine_epoch) != epoch)
+		goto out;
+
+	old = ci->i_qmds_cred;
+	/* Invalidate the old receipt before publishing its new credentials. */
+	atomic64_set_release(&ci->i_qmds, 0);
+	if (old != cred)
+		ci->i_qmds_cred = get_cred(cred);
+	else
+		old = NULL;
+
+	/* Pairs with atomic64_read_acquire() in ceph_qmds_confirmed(). */
+	atomic64_set_release(&ci->i_qmds, epoch | 1ULL);
+	/* Order the confirmation store before re-reading its generation. */
+	smp_mb();
+	if ((u64)atomic64_read(&mdsc->quarantine_epoch) != epoch ||
+	    atomic_read(&mdsc->quarantine_splices)) {
+		/* The epoch moved. Drop the confirmation. */
+		atomic64_set_release(&ci->i_qmds, 0);
+	}
+out:
+	spin_unlock(&ci->i_ceph_lock);
+	put_cred(old);
+}
+
 static void ceph_quarantine_forget_receipt(struct ceph_inode_info *ci)
 {
 	const struct cred *cred;
@@ -1114,67 +1174,161 @@ enum ceph_qstate ceph_inode_quarantine_write_state(struct inode *inode)
 	return __ceph_inode_quarantine_state(inode, false);
 }
 
-static struct ceph_mds_client *ceph_quarantine_move_mdsc(struct inode *inode,
-							 struct inode *old_dir,
-							 struct inode *new_dir)
+/* Wait for namespace writers without blocking a fill-trace's own LSM hook. */
+int ceph_quarantine_wait(struct inode *inode)
 {
-	if (!inode || old_dir == new_dir)
-		return NULL;
-	return ceph_sb_to_mdsc(inode->i_sb);
+	struct ceph_mds_client *mdsc = ceph_sb_to_mdsc(inode->i_sb);
+	struct ceph_quarantine_splice *splice;
+	bool own_splice = false;
+	int err;
+
+	/* A fill-trace LSM hook may own the splice that would wake us. */
+	if (ceph_current_fill_trace_request())
+		return -EBUSY;
+	/* Async create and mount-root instantiation can run hooks too. */
+	spin_lock(&mdsc->quarantine_lock);
+	list_for_each_entry(splice, &mdsc->quarantine_splice_owners, item) {
+		if (splice->task == current) {
+			own_splice = true;
+			break;
+		}
+	}
+	spin_unlock(&mdsc->quarantine_lock);
+	if (own_splice)
+		return -EBUSY;
+	if (signal_pending(current))
+		return -ERESTARTSYS;
+	err = wait_event_interruptible(mdsc->quarantine_waitq,
+				       !(atomic64_read_acquire(&mdsc->quarantine_epoch) & 1) &&
+				       !atomic_read_acquire(&mdsc->quarantine_splices));
+	if (err)
+		return err;
+	if (ceph_inode_is_shutdown(inode))
+		return -ESTALE;
+	cond_resched();
+	return 0;
+}
+
+/* PIN-only, stale and absent caps cannot deliver the first quarantine revoke. */
+static bool ceph_quarantine_head_needs_revalidate(struct inode *inode)
+{
+	struct ceph_mds_client *mdsc = ceph_sb_to_mdsc(inode->i_sb);
+
+	return !ceph_in_snap(inode) &&
+		READ_ONCE(mdsc->quarantine_supported) &&
+		!ceph_quarantine_head_cap_valid(inode);
 }
 
 /*
- * Bracket d_move(). The epoch is odd across the parent change, so a
- * cached false cannot be reused against the new parent. d_move() does
- * not sleep. Same-directory renames do not take a writer slot.
- *
- * Sleeping alias splices use a separate counter below.
+ * Snapshots need a fresh MDS answer unless a connected head cap can receive
+ * quarantine revocations. For head inodes, a direct mark fails the caller.
+ * Inherited marks may permit recovery through a path-scoped q/Q grant.
+ * A GETATTR read receipt covers only this inode, credentials and epoch;
+ * writes need a mode-specific OPEN. A new epoch invalidates the receipt.
  */
-void ceph_quarantine_move_start(struct inode *inode, struct inode *old_dir,
-				struct inode *new_dir)
+int ceph_quarantine_revalidate(struct inode *inode)
+{
+	struct ceph_mds_client *mdsc = ceph_sb_to_mdsc(inode->i_sb);
+	enum ceph_qstate q;
+	u64 epoch;
+	int err;
+
+	/* Immutable snapshot caps cannot receive quarantine notifications. */
+	if (ceph_in_snap(inode)) {
+		if (!ceph_snapshot_needs_revalidate(inode))
+			return 0;
+		if (ceph_snap(inode) == CEPH_SNAPDIR)
+			return ceph_do_getattr(inode, CEPH_CAP_AUTH_SHARED, true);
+		return ceph_snapshot_read_auth(inode);
+	}
+
+	for (;;) {
+		q = ceph_inode_quarantine_state(inode);
+		if (q == CEPH_Q_CLEAR &&
+		    !ceph_quarantine_head_needs_revalidate(inode))
+			return 0;
+		if (q == CEPH_Q_UNDER && ceph_inode_quarantine_direct(inode))
+			return -EACCES;
+		/* Wait rejects our own splice or fill-trace context before sleeping. */
+		err = ceph_quarantine_wait(inode);
+		if (err)
+			return err;
+		epoch = (u64)atomic64_read_acquire(&mdsc->quarantine_epoch);
+		if ((epoch & 1) || atomic_read(&mdsc->quarantine_splices))
+			continue;
+
+		err = ceph_do_getattr(inode, CEPH_STAT_CAP_INODE_ALL, true);
+		if (err)
+			return err;
+
+		if (ceph_inode_quarantine_direct(inode))
+			return -EACCES;
+		/*
+		 * This reply authorizes the current operation even if no cap
+		 * was granted. Physical clearing and reusable receipts have
+		 * separate freshness checks; cached access still needs its caps.
+		 */
+		if ((u64)atomic64_read_acquire(&mdsc->quarantine_epoch) == epoch &&
+		    !atomic_read_acquire(&mdsc->quarantine_splices))
+			return 0;
+	}
+}
+
+/* The returned token pairs the gate decision across feature activation. */
+struct ceph_mds_client *ceph_quarantine_move_start(struct inode *inode,
+						   struct inode *old_dir,
+						   struct inode *new_dir)
 {
-	ceph_quarantine_write_begin(ceph_quarantine_move_mdsc(inode, old_dir,
-							      new_dir));
+	struct ceph_mds_client *mdsc;
+
+	if (!inode || old_dir == new_dir)
+		return NULL;
+	mdsc = ceph_sb_to_mdsc(inode->i_sb);
+	if (!READ_ONCE(mdsc->quarantine_supported))
+		return NULL;
+	ceph_quarantine_write_begin(mdsc);
+	return mdsc;
 }
 
-void ceph_quarantine_move_finish(struct inode *inode, struct inode *old_dir,
-				 struct inode *new_dir)
+void ceph_quarantine_move_finish(struct ceph_mds_client *mdsc)
 {
-	ceph_quarantine_write_end(ceph_quarantine_move_mdsc(inode, old_dir,
-							    new_dir));
+	ceph_quarantine_write_end(mdsc);
 }
 
 /*
- * d_splice_alias() and d_instantiate() can sleep in the LSM hook before
- * changing ancestry. Keep the epoch even for that hook, but disallow
- * all cached answers throughout the splice. A fresh walk also samples
- * rename_lock, so it cannot accept a chain torn by a move. Bracket
- * first attachments too: descendants may hold MDS confirmations from
- * when the directory had no connected alias.
- *
- * Fully ordered atomics and quarantine_lock make the counter and epoch
- * transitions one writer operation. Entry disables cache use before
- * invalidating tags; exit invalidates tags before enabling cache use.
- * Adding two preserves the parity of any concurrent mark writer.
+ * LSM hooks can sleep and reenter Ceph during a splice. Keep the epoch
+ * even but disable cached receipts while the ancestry can change.
+ * Namespace operations on unsupported mounts do not take writer slots.
  */
-void ceph_quarantine_splice_start(struct super_block *sb)
+void ceph_quarantine_splice_start(struct super_block *sb,
+				  struct ceph_quarantine_splice *splice)
 {
 	struct ceph_mds_client *mdsc = ceph_sb_to_mdsc(sb);
 
+	splice->mdsc = NULL;
+	if (!READ_ONCE(mdsc->quarantine_supported))
+		return;
+	splice->mdsc = mdsc;
+	splice->task = current;
 	spin_lock(&mdsc->quarantine_lock);
+	list_add(&splice->item, &mdsc->quarantine_splice_owners);
 	atomic_inc_return(&mdsc->quarantine_splices);
 	atomic64_add_return(2, &mdsc->quarantine_epoch);
 	spin_unlock(&mdsc->quarantine_lock);
 }
 
-void ceph_quarantine_splice_finish(struct super_block *sb)
+void ceph_quarantine_splice_finish(struct ceph_quarantine_splice *splice)
 {
-	struct ceph_mds_client *mdsc = ceph_sb_to_mdsc(sb);
+	struct ceph_mds_client *mdsc = splice->mdsc;
 
+	if (!mdsc)
+		return;
 	spin_lock(&mdsc->quarantine_lock);
+	list_del(&splice->item);
 	atomic64_add_return(2, &mdsc->quarantine_epoch);
 	atomic_dec_return(&mdsc->quarantine_splices);
 	spin_unlock(&mdsc->quarantine_lock);
+	wake_up_all(&mdsc->quarantine_waitq);
 }
 
 static bool ceph_quarantine_reply_can_clear(struct ceph_inode_info *ci,
@@ -1618,9 +1772,12 @@ int ceph_fill_inode(struct inode *inode, struct folio *locked_folio,
 	 */
 	if (iinfo->xattr_len > 4) {
 		xattr_blob = ceph_buffer_new(iinfo->xattr_len, GFP_NOFS);
-		if (!xattr_blob)
+		if (!xattr_blob) {
 			pr_err_client(cl, "ENOMEM xattr blob %d bytes\n",
 				      iinfo->xattr_len);
+			err = -ENOMEM;
+			goto out;
+		}
 	}
 
 	if (iinfo->pool_ns_len > 0)
@@ -1630,6 +1787,9 @@ int ceph_fill_inode(struct inode *inode, struct folio *locked_folio,
 	if (ceph_in_snap(inode) && !ci->i_snapid_map)
 		ci->i_snapid_map = ceph_get_snapid_map(mdsc, ceph_snap(inode));
 
+	if (READ_ONCE(session->s_mdsc->quarantine_supported))
+		ceph_quarantine_uncache_inode_acls(inode);
+
 	spin_lock(&ci->i_ceph_lock);
 
 	/*
@@ -1852,12 +2012,12 @@ int ceph_fill_inode(struct inode *inode, struct folio *locked_folio,
 
 		if (IS_ENCRYPTED(inode)) {
 			/*
-			 * Encrypted symlinks need to be decrypted before we can
-			 * cache their targets in i_link. Don't touch it here.
+			 * Decrypt on access without installing i_link, so the VFS
+			 * cannot bypass the quarantine get_link check.
 			 */
 			inode->i_op = &ceph_encrypted_symlink_iops;
 		} else {
-			inode->i_link = ci->i_symlink;
+			/* VFS i_link would bypass the quarantine get_link check. */
 			inode->i_op = &ceph_symlink_iops;
 		}
 		break;
@@ -2078,6 +2238,7 @@ static void update_dentry_lease_careful(struct dentry *dentry,
  */
 static int splice_dentry(struct dentry **pdn, struct inode *in)
 {
+	struct ceph_quarantine_splice qsplice;
 	struct ceph_client *cl = ceph_inode_to_client(in);
 	struct dentry *dn = *pdn;
 	struct dentry *realdn;
@@ -2109,9 +2270,9 @@ static int splice_dentry(struct dentry **pdn, struct inode *in)
 	/* dn must be unhashed */
 	if (!d_unhashed(dn))
 		d_drop(dn);
-	ceph_quarantine_splice_start(dn->d_sb);
+	ceph_quarantine_splice_start(dn->d_sb, &qsplice);
 	realdn = d_splice_alias(in, dn);
-	ceph_quarantine_splice_finish(dn->d_sb);
+	ceph_quarantine_splice_finish(&qsplice);
 	if (IS_ERR(realdn)) {
 		pr_err_client(cl, "error %ld %p inode %p ino %llx.%llx\n",
 			      PTR_ERR(realdn), dn, in, ceph_vinop(in));
@@ -2145,6 +2306,7 @@ static int splice_dentry(struct dentry **pdn, struct inode *in)
  */
 int ceph_fill_trace(struct super_block *sb, struct ceph_mds_request *req)
 {
+	struct ceph_mds_client *qmdsc;
 	struct ceph_mds_session *session = req->r_session;
 	struct ceph_mds_reply_info_parsed *rinfo = &req->r_reply_info;
 	struct inode *in = NULL;
@@ -2268,17 +2430,24 @@ int ceph_fill_trace(struct super_block *sb, struct ceph_mds_request *req)
 	}
 
 	if (rinfo->head->is_target) {
+		struct ceph_vino expected;
+
 		/* Should be filled in by handle_reply */
 		BUG_ON(!req->r_target_inode);
 
 		in = req->r_target_inode;
 		ceph_inode_set_subvolume(in, rinfo->targeti.subvolume_id);
-		err = ceph_fill_inode(in, req->r_locked_folio, &rinfo->targeti,
-				NULL, session,
-				(!test_bit(CEPH_MDS_R_ABORTED, &req->r_req_flags) &&
-				 !test_bit(CEPH_MDS_R_ASYNC, &req->r_req_flags) &&
-				 rinfo->head->result == 0) ?  req->r_fmode : -1,
-				&req->r_caps_reservation);
+		expected = req->r_inode ? ceph_vino(req->r_inode) : req->r_ino1;
+		if (req->r_locked_folio &&
+		    (ceph_ino(in) != expected.ino || ceph_snap(in) != expected.snap))
+			err = -ESTALE;
+		else
+			err = ceph_fill_inode(in, req->r_locked_folio, &rinfo->targeti,
+					      NULL, session,
+					      (!test_bit(CEPH_MDS_R_ABORTED, &req->r_req_flags) &&
+					       !test_bit(CEPH_MDS_R_ASYNC, &req->r_req_flags) &&
+					       rinfo->head->result == 0) ? req->r_fmode : -1,
+					      &req->r_caps_reservation);
 		if (err < 0) {
 			pr_err_client(cl, "badness %p %llx.%llx\n", in,
 				      ceph_vinop(in));
@@ -2348,9 +2517,9 @@ int ceph_fill_trace(struct super_block *sb, struct ceph_mds_request *req)
 			ceph_dir_clear_ordered(olddir);
 
 			moved = d_inode(req->r_old_dentry);
-			ceph_quarantine_move_start(moved, olddir, dir);
+			qmdsc = ceph_quarantine_move_start(moved, olddir, dir);
 			d_move(req->r_old_dentry, dn);
-			ceph_quarantine_move_finish(moved, olddir, dir);
+			ceph_quarantine_move_finish(qmdsc);
 			boutc(cl, " src %p '%s' dst %p '%s'\n",
 			      req->r_old_dentry, req->r_old_dentry->d_name.name, dn, dn->d_name.name);
 
@@ -2959,14 +3128,43 @@ static void ceph_inode_work(struct work_struct *work)
 	iput(inode);
 }
 
+static int ceph_symlink_revalidate(struct inode *inode)
+{
+	/* Snapshot caps are immutable and cannot revoke a saved target. */
+	if (ceph_in_snap(inode) && ceph_snap(inode) != CEPH_SNAPDIR)
+		return ceph_snapshot_read_auth(inode);
+	return ceph_do_getattr(inode, CEPH_CAP_AUTH_SHARED, ceph_in_snap(inode));
+}
+
+static const char *ceph_get_link(struct dentry *dentry, struct inode *inode,
+				 struct delayed_call *done)
+{
+	int err;
+
+	if (!dentry)
+		return ERR_PTR(-ECHILD);
+	err = ceph_symlink_revalidate(inode);
+	if (err)
+		return ERR_PTR(err);
+	return ceph_inode(inode)->i_symlink;
+}
+
+/*
+ * ceph_fscrypt_ops disables symlink target caching, so encrypted readlink
+ * and traversal reach this callback and ceph_symlink_revalidate().
+ */
 static const char *ceph_encrypted_get_link(struct dentry *dentry,
 					   struct inode *inode,
 					   struct delayed_call *done)
 {
 	struct ceph_inode_info *ci = ceph_inode(inode);
+	int err;
 
 	if (!dentry)
 		return ERR_PTR(-ECHILD);
+	err = ceph_symlink_revalidate(inode);
+	if (err)
+		return ERR_PTR(err);
 
 	return fscrypt_get_symlink(inode, ci->i_symlink, i_size_read(inode),
 				   done);
@@ -2989,7 +3187,7 @@ static int ceph_encrypted_symlink_getattr(struct mnt_idmap *idmap,
  * symlinks
  */
 static const struct inode_operations ceph_symlink_iops = {
-	.get_link = simple_get_link,
+	.get_link = ceph_get_link,
 	.setattr = ceph_setattr,
 	.getattr = ceph_getattr,
 	.listxattr = ceph_listxattr,
@@ -3200,6 +3398,25 @@ int __ceph_setattr(struct mnt_idmap *idmap, struct inode *inode,
 		return PTR_ERR(req);
 	}
 
+	/*
+	 * Exclusive caps apply mode, owner, times, and a size increase
+	 * with no MDS round trip. The state check may schedule, so it runs
+	 * before i_ceph_lock. A short walk sends the setattr.
+	 */
+	switch (ceph_inode_quarantine_write_state(inode)) {
+	case CEPH_Q_UNDER:
+		if (ceph_inode_quarantine_direct(inode)) {
+			err = -EACCES;
+			goto out;
+		}
+		fallthrough;
+	case CEPH_Q_UNKNOWN:
+		do_sync = true;
+		break;
+	default:
+		break;
+	}
+
 	fill_fscrypt = false;
 	spin_lock(&ci->i_ceph_lock);
 	issued = __ceph_caps_issued(ci, NULL);
@@ -3238,7 +3455,7 @@ int __ceph_setattr(struct mnt_idmap *idmap, struct inode *inode,
 			kfree(ci->fscrypt_auth);
 			ci->fscrypt_auth = (u8 *)cia->fscrypt_auth;
 			ci->fscrypt_auth_len = len;
-		} else if ((issued & CEPH_CAP_AUTH_SHARED) == 0 ||
+		} else if (do_sync || (issued & CEPH_CAP_AUTH_SHARED) == 0 ||
 			   ci->fscrypt_auth_len != len ||
 			   memcmp(ci->fscrypt_auth, cia->fscrypt_auth, len)) {
 			req->r_fscrypt_auth = cia->fscrypt_auth;
@@ -3265,7 +3482,7 @@ int __ceph_setattr(struct mnt_idmap *idmap, struct inode *inode,
 		if (!do_sync && (issued & CEPH_CAP_AUTH_EXCL)) {
 			inode->i_uid = fsuid;
 			dirtied |= CEPH_CAP_AUTH_EXCL;
-		} else if ((issued & CEPH_CAP_AUTH_SHARED) == 0 ||
+		} else if (do_sync || (issued & CEPH_CAP_AUTH_SHARED) == 0 ||
 			   !uid_eq(fsuid, inode->i_uid)) {
 			req->r_args.setattr.uid = cpu_to_le32(
 				from_kuid(&init_user_ns, fsuid));
@@ -3283,7 +3500,7 @@ int __ceph_setattr(struct mnt_idmap *idmap, struct inode *inode,
 		if (!do_sync && (issued & CEPH_CAP_AUTH_EXCL)) {
 			inode->i_gid = fsgid;
 			dirtied |= CEPH_CAP_AUTH_EXCL;
-		} else if ((issued & CEPH_CAP_AUTH_SHARED) == 0 ||
+		} else if (do_sync || (issued & CEPH_CAP_AUTH_SHARED) == 0 ||
 			   !gid_eq(fsgid, inode->i_gid)) {
 			req->r_args.setattr.gid = cpu_to_le32(
 				from_kgid(&init_user_ns, fsgid));
@@ -3297,9 +3514,10 @@ int __ceph_setattr(struct mnt_idmap *idmap, struct inode *inode,
 		if (!do_sync && (issued & CEPH_CAP_AUTH_EXCL)) {
 			inode->i_mode = attr->ia_mode;
 			dirtied |= CEPH_CAP_AUTH_EXCL;
-		} else if ((issued & CEPH_CAP_AUTH_SHARED) == 0 ||
+		} else if (do_sync || (issued & CEPH_CAP_AUTH_SHARED) == 0 ||
 			   attr->ia_mode != inode->i_mode) {
-			inode->i_mode = attr->ia_mode;
+			if (!do_sync)
+				inode->i_mode = attr->ia_mode;
 			req->r_args.setattr.mode = cpu_to_le32(attr->ia_mode);
 			mask |= CEPH_SETATTR_MODE;
 			release |= CEPH_CAP_AUTH_SHARED;
@@ -3320,7 +3538,7 @@ int __ceph_setattr(struct mnt_idmap *idmap, struct inode *inode,
 					      &attr->ia_atime) < 0) {
 			inode_set_atime_to_ts(inode, attr->ia_atime);
 			dirtied |= CEPH_CAP_FILE_WR;
-		} else if ((issued & CEPH_CAP_FILE_SHARED) == 0 ||
+		} else if (do_sync || (issued & CEPH_CAP_FILE_SHARED) == 0 ||
 			   !timespec64_equal(&atime, &attr->ia_atime)) {
 			ceph_encode_timespec64(&req->r_args.setattr.atime,
 					       &attr->ia_atime);
@@ -3359,7 +3577,7 @@ int __ceph_setattr(struct mnt_idmap *idmap, struct inode *inode,
 				dirtied |= CEPH_CAP_FILE_EXCL;
 				ia_valid |= ATTR_MTIME;
 			}
-		} else if ((issued & CEPH_CAP_FILE_SHARED) == 0 ||
+		} else if (do_sync || (issued & CEPH_CAP_FILE_SHARED) == 0 ||
 			   attr->ia_size != isize) {
 			mask |= CEPH_SETATTR_SIZE;
 			release |= CEPH_CAP_FILE_SHARED | CEPH_CAP_FILE_EXCL |
@@ -3394,7 +3612,7 @@ int __ceph_setattr(struct mnt_idmap *idmap, struct inode *inode,
 			   timespec64_compare(&mtime, &attr->ia_mtime) < 0) {
 			inode_set_mtime_to_ts(inode, attr->ia_mtime);
 			dirtied |= CEPH_CAP_FILE_WR;
-		} else if ((issued & CEPH_CAP_FILE_SHARED) == 0 ||
+		} else if (do_sync || (issued & CEPH_CAP_FILE_SHARED) == 0 ||
 			   !timespec64_equal(&mtime, &attr->ia_mtime)) {
 			ceph_encode_timespec64(&req->r_args.setattr.mtime,
 					       &attr->ia_mtime);
@@ -3418,11 +3636,11 @@ int __ceph_setattr(struct mnt_idmap *idmap, struct inode *inode,
 			 * we need to choose a cap to dirty under, or do
 			 * a almost-no-op setattr
 			 */
-			if (issued & CEPH_CAP_AUTH_EXCL)
+			if (!do_sync && (issued & CEPH_CAP_AUTH_EXCL))
 				dirtied |= CEPH_CAP_AUTH_EXCL;
-			else if (issued & CEPH_CAP_FILE_EXCL)
+			else if (!do_sync && (issued & CEPH_CAP_FILE_EXCL))
 				dirtied |= CEPH_CAP_FILE_EXCL;
-			else if (issued & CEPH_CAP_XATTR_EXCL)
+			else if (!do_sync && (issued & CEPH_CAP_XATTR_EXCL))
 				dirtied |= CEPH_CAP_XATTR_EXCL;
 			else
 				mask |= CEPH_SETATTR_CTIME;
@@ -3478,6 +3696,12 @@ int __ceph_setattr(struct mnt_idmap *idmap, struct inode *inode,
 			ceph_free_cap_flush(prealloc_cf);
 			goto retry;
 		}
+		if (!err && do_sync) {
+			/* Existing exclusive caps may keep reply-fill from updating attrs. */
+			spin_lock(&ci->i_ceph_lock);
+			setattr_copy(idmap, inode, attr);
+			spin_unlock(&ci->i_ceph_lock);
+		}
 	}
 out:
 	boutc(cl, "%p %llx.%llx result=%d (%s locally, %d remote)\n", inode,
@@ -3578,6 +3802,149 @@ int ceph_try_to_choose_auth_mds(struct inode *inode, int mask)
 		return USE_ANY_MDS;
 }
 
+/* A snapshot path must reach a head inode to encode its snapshot name. */
+static bool ceph_snapshot_has_path(struct inode *inode, bool check_caps)
+{
+	struct dentry *cur;
+	u64 subvolume_id = CEPH_SUBVOLUME_ID_NONE;
+	unsigned int seq, depth = 0;
+	bool connected = false;
+
+	/* A torn membership ID cannot establish revocation coverage. */
+	if (check_caps && IS_ENABLED(CONFIG_64BIT))
+		subvolume_id = READ_ONCE(ceph_inode(inode)->i_subvolume_id);
+
+	rcu_read_lock();
+	seq = read_seqbegin(&rename_lock);
+	cur = d_find_alias_rcu(inode);
+	while (cur && depth++ < 256) {
+		struct inode *in = d_inode_rcu(cur);
+		struct dentry *parent;
+
+		if (!in)
+			break;
+		if (!ceph_in_snap(in)) {
+			connected = !check_caps ||
+				(subvolume_id != CEPH_SUBVOLUME_ID_NONE &&
+				 ceph_ino(in) == subvolume_id &&
+				 ceph_quarantine_head_cap_valid(in));
+			break;
+		}
+		if (d_unhashed(cur) ||
+		    (READ_ONCE(cur->d_flags) & DCACHE_DISCONNECTED))
+			break;
+		parent = READ_ONCE(cur->d_parent);
+		if (parent == cur)
+			break;
+		cur = parent;
+	}
+	if (read_seqretry(&rename_lock, seq))
+		connected = false;
+	rcu_read_unlock();
+	return connected;
+}
+
+/*
+ * Only the target's subvolume root supplies reliable quarantine revocations.
+ * A snapshot taken above that root can reach a head ancestor outside the
+ * subvolume; its cap stays valid when the target is quarantined. Require the
+ * MDS-provided subvolume ID to identify the head root whose AUTH_SHARED cap
+ * we check. Unknown IDs, 32-bit clients and other head ancestors need an MDS
+ * answer, as do detached snapshots, expired root caps and active quarantine.
+ */
+bool ceph_snapshot_needs_revalidate(struct inode *inode)
+{
+	struct ceph_mds_client *mdsc = ceph_sb_to_mdsc(inode->i_sb);
+	u64 epoch;
+
+	if (!ceph_in_snap(inode) || !READ_ONCE(mdsc->quarantine_supported))
+		return false;
+	epoch = atomic64_read_acquire(&mdsc->quarantine_epoch);
+	if ((epoch & 1) || atomic_read(&mdsc->quarantine_count) ||
+	    atomic_read(&mdsc->quarantine_splices) ||
+	    !ceph_snapshot_has_path(inode, true))
+		return true;
+	/* Do not accept a head cap observed across a revoke or parent change. */
+	smp_rmb();
+	return (u64)atomic64_read_acquire(&mdsc->quarantine_epoch) != epoch ||
+		atomic_read(&mdsc->quarantine_count) ||
+		atomic_read(&mdsc->quarantine_splices);
+}
+
+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)
+{
+	if (!ceph_snapshot_needs_revalidate(inode))
+		return 0;
+
+	/*
+	 * 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.
@@ -3589,21 +3956,59 @@ 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;
+	bool head_revalidate;
+	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;
+		if (!ceph_snapshot_needs_revalidate(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;
+	}
+
+	head_revalidate = ceph_quarantine_head_needs_revalidate(inode);
+	if (head_revalidate) {
+		/* Do not recurse into an MDS request from our own splice hook. */
+		err = ceph_quarantine_wait(inode);
+		if (err)
+			return err;
+		mask |= CEPH_CAP_AUTH_SHARED;
 	}
 
+	/*
+	 * 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. Snapshot caps need
+	 * an MDS check unless a live head cap provides quarantine notifications.
+	 */
+	if (ceph_snapshot_needs_revalidate(inode) ||
+	    head_revalidate ||
+	    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;
 
-	mode = ceph_try_to_choose_auth_mds(inode, mask);
+	if (ceph_in_snap(inode) && !ceph_snapshot_has_path(inode, false))
+		return ceph_snapshot_lookupino(inode, locked_folio, mask);
+
+	qepoch = (u64)atomic64_read_acquire(&mdsc->quarantine_epoch);
+	mode = head_revalidate ? USE_AUTH_MDS :
+		ceph_try_to_choose_auth_mds(inode, mask);
 	req = ceph_mdsc_create_request(mdsc, CEPH_MDS_OP_GETATTR, mode);
 	if (IS_ERR(req))
 		return PTR_ERR(req);
@@ -3613,6 +4018,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) {
@@ -3758,8 +4172,15 @@ 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 need a live head cap for notifications.
+	 */
+	if (ceph_snapshot_needs_revalidate(inode) ||
+	    ceph_quarantine_head_needs_revalidate(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 12cb50e0a1665dec866eee405b081f970e89351a..d152cad10c23e92a26e7f93c5c1d5fd7e6c7014e 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);
@@ -403,12 +407,18 @@ long ceph_ioctl(struct file *file, unsigned int cmd, unsigned long arg)
 		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);
 
 	case FS_IOC_ADD_ENCRYPTION_KEY:
@@ -431,6 +441,9 @@ long ceph_ioctl(struct file *file, unsigned int cmd, unsigned long arg)
 		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.c b/fs/ceph/mds_client.c
index 5fec6e129d2822847c5dab6b2674560712db2765..730fd4bfaeffa3dac6be5cd03c7116c8a8c02bbc 100644
--- a/fs/ceph/mds_client.c
+++ b/fs/ceph/mds_client.c
@@ -4668,6 +4668,8 @@ static void handle_session(struct ceph_mds_session *session,
 		mdsc->s_cap_auths = cap_auths;
 
 		session->s_features = features;
+		if (test_bit(CEPHFS_FEATURE_QUARANTINE, &session->s_features))
+			ceph_quarantine_enable(mdsc);
 		if (test_bit(CEPHFS_FEATURE_METRIC_COLLECT,
 			     &session->s_features))
 			ceph_metric_bind_session(mdsc, session);
@@ -6480,6 +6482,9 @@ int ceph_mdsc_init(struct ceph_fs_client *fsc)
 	spin_lock_init(&mdsc->stopping_lock);
 	atomic_set(&mdsc->stopping_blockers, 0);
 	atomic_set(&mdsc->quarantine_count, 0);
+	mdsc->quarantine_supported = false;
+	init_waitqueue_head(&mdsc->quarantine_waitq);
+	INIT_LIST_HEAD(&mdsc->quarantine_splice_owners);
 	atomic_set(&mdsc->quarantine_writers, 0);
 	atomic_set(&mdsc->quarantine_splices, 0);
 	atomic64_set(&mdsc->quarantine_epoch, 0);
diff --git a/fs/ceph/mds_client.h b/fs/ceph/mds_client.h
index 816f5f2d32ccf19b8d0644a3d5c24a730fd1015b..422a27e44c38016f32d2073c16feccd73ad31ff5 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,		\
 }
 
 /*
@@ -507,6 +515,10 @@ struct ceph_mds_client {
 	spinlock_t              stopping_lock;  /* protect snap_empty */
 	int                     stopping;      /* the stage of shutting down */
 	atomic_t                stopping_blockers;
+	/* Once negotiated, keep tracking across reconnects and mixed ranks. */
+	bool			quarantine_supported;
+	wait_queue_head_t	quarantine_waitq;
+	struct list_head	quarantine_splice_owners; /* quarantine_lock */
 	/*
 	 * Inodes with CEPH_I_QUARANTINE set. Hot paths skip the
 	 * ancestor walk while this is zero.
diff --git a/fs/ceph/super.c b/fs/ceph/super.c
index ad9460d9673cb18230d8fbcb93a27aa26f7ace38..ed2eab9d69692ce6154095aaa66a9950879fe967 100644
--- a/fs/ceph/super.c
+++ b/fs/ceph/super.c
@@ -1137,11 +1137,13 @@ static struct dentry *open_root_dentry(struct ceph_fs_client *fsc,
 	err = ceph_mdsc_do_request(mdsc, NULL, req);
 	if (err == 0) {
 		struct inode *inode = req->r_target_inode;
+		struct ceph_quarantine_splice qsplice;
+
 		req->r_target_inode = NULL;
 		doutc(cl, "success\n");
-		ceph_quarantine_splice_start(fsc->sb);
+		ceph_quarantine_splice_start(fsc->sb, &qsplice);
 		root = d_make_root(inode);
-		ceph_quarantine_splice_finish(fsc->sb);
+		ceph_quarantine_splice_finish(&qsplice);
 		if (!root) {
 			root = ERR_PTR(-ENOMEM);
 			goto out;
diff --git a/fs/ceph/super.h b/fs/ceph/super.h
index 0264bd9d705fcb1bc904d0fc7fc94be2b6de3c66..8aecc6f1b62907e37556793793b4accf84800612 100644
--- a/fs/ceph/super.h
+++ b/fs/ceph/super.h
@@ -1292,6 +1292,8 @@ void ceph_inode_shutdown(struct inode *inode);
  * CEPH_Q_CLEAR: cached ancestry is clear, or the MDS confirmed read
  * access to this exact inode for the current epoch. Writers must use
  * ceph_inode_quarantine_write_state(), which ignores read receipts.
+ * Cached metadata access also needs a live cap for quarantine revocations;
+ * use ceph_quarantine_revalidate() before returning it without cap references.
  * CEPH_Q_UNDER: this inode or a cached ancestor is marked.
  * CEPH_Q_UNKNOWN: the parent chain is not fully cached. Ask the MDS.
  * Do not treat that as quarantined and do not use cached caps.
@@ -1306,16 +1308,27 @@ enum ceph_qstate ceph_inode_quarantine_state(struct inode *inode);
 enum ceph_qstate ceph_inode_quarantine_write_state(struct inode *inode);
 bool ceph_inode_quarantine_direct(struct inode *inode);
 int ceph_quarantine_revalidate(struct inode *inode);
+int ceph_quarantine_wait(struct inode *inode);
+bool ceph_snapshot_needs_revalidate(struct inode *inode);
+bool ceph_quarantine_head_cap_valid(struct inode *inode);
+void ceph_quarantine_enable(struct ceph_mds_client *mdsc);
 int ceph_snapshot_read_auth(struct inode *inode);
 void ceph_quarantine_set(struct ceph_inode_info *ci);
 bool ceph_quarantine_clear(struct ceph_inode_info *ci);
 bool ceph_quarantine_reset(struct ceph_inode_info *ci);
-void ceph_quarantine_move_start(struct inode *inode, struct inode *old_dir,
-				struct inode *new_dir);
-void ceph_quarantine_move_finish(struct inode *inode, struct inode *old_dir,
-				 struct inode *new_dir);
-void ceph_quarantine_splice_start(struct super_block *sb);
-void ceph_quarantine_splice_finish(struct super_block *sb);
+struct ceph_mds_client *ceph_quarantine_move_start(struct inode *inode,
+						   struct inode *old_dir,
+						   struct inode *new_dir);
+void ceph_quarantine_move_finish(struct ceph_mds_client *mdsc);
+struct ceph_quarantine_splice {
+	struct ceph_mds_client *mdsc;
+	struct task_struct *task;
+	struct list_head item;
+};
+
+void ceph_quarantine_splice_start(struct super_block *sb,
+				  struct ceph_quarantine_splice *splice);
+void ceph_quarantine_splice_finish(struct ceph_quarantine_splice *splice);
 void ceph_handle_quarantine_disable(struct ceph_mds_client *mdsc,
 				    struct ceph_mds_session *session,
 				    struct ceph_msg *msg);
@@ -1379,6 +1392,7 @@ static inline int ceph_security_init_secctx(struct dentry *dentry, umode_t mode,
 {
 	return 0;
 }
+
 static inline void ceph_security_invalidate_secctx(struct inode *inode)
 {
 }
@@ -1394,8 +1408,9 @@ 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);
+void ceph_init_inode_acls(struct inode *inode, struct ceph_acl_sec_ctx *as_ctx);
+void ceph_quarantine_uncache_inode_acls(struct inode *inode);
+void ceph_quarantine_disable_acl_cache(struct super_block *sb);
 
 static inline void ceph_forget_all_cached_acls(struct inode *inode)
 {
@@ -1412,11 +1427,20 @@ 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_quarantine_uncache_inode_acls(struct inode *inode)
+{
+}
+
+static inline void ceph_quarantine_disable_acl_cache(struct super_block *sb)
+{
+}
+
 static inline void ceph_forget_all_cached_acls(struct inode *inode)
 {
 }
@@ -1578,7 +1602,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 4cba6e7bffd3dd65dedaa9225546079b2e6e57c3..13d4dfef679de1d7348db14da7e7bfcf1c26d077 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_snapshot_needs_revalidate(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_snapshot_needs_revalidate(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;
@@ -1305,7 +1328,8 @@ int __ceph_setxattr(struct inode *inode, const char *name,
 		struct ceph_buffer *blob;
 
 		spin_unlock(&ci->i_ceph_lock);
-		ceph_buffer_put(old_blob); /* Shouldn't be required */
+		ceph_buffer_put(old_blob);
+		old_blob = NULL;
 		boutc(cl, " pre-allocating new blob size=%d\n",
 		      required_blob_size);
 		blob = ceph_buffer_new(required_blob_size, GFP_NOFS);
@@ -1319,6 +1343,34 @@ 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);
+			old_blob = NULL;
+			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);
 
@@ -1331,6 +1383,7 @@ int __ceph_setxattr(struct inode *inode, const char *name,
 
 	spin_unlock(&ci->i_ceph_lock);
 	ceph_buffer_put(old_blob);
+	old_blob = NULL;
 	if (lock_snap_rwsem)
 		up_read(&mdsc->snap_rwsem);
 	if (dirty)
@@ -1341,6 +1394,7 @@ int __ceph_setxattr(struct inode *inode, const char *name,
 do_sync:
 	spin_unlock(&ci->i_ceph_lock);
 	ceph_buffer_put(old_blob);
+	old_blob = NULL;
 do_sync_unlocked:
 	if (lock_snap_rwsem)
 		up_read(&mdsc->snap_rwsem);

Re: [PATCH v6 4/5] ceph: record subvolume quarantine from MDS cap updates

[sashiko-bot]

Thank you for your contribution! Sashiko AI review found 2 potential issue(s) to consider:
- [High] ceph: epoch self-invalidation in ceph_quarantine_reply_can_clear()
- [High] ceph: __ceph_remove_cap() resets quarantine mark during cap migration
--

commit 8a2e69935d967838e52792bb963b124cfe229b88
Author: Alex Markuze <amarkuze@redhat.com>

ceph: record subvolume quarantine from MDS cap updates

This commit decodes the signed oserrno from MClientCaps v14 messages to
record a sticky subvolume quarantine mark and reporting MDS rank on each
inode, along with epoch-based caching and RCU parent walks to resolve
inherited quarantine state.

> diff --git a/fs/ceph/caps.c b/fs/ceph/caps.c
> index b27bcdc24d29f..b07814d0038cd 100644
> --- a/fs/ceph/caps.c
> +++ b/fs/ceph/caps.c

[ ... ]

> @@ -1264,6 +1265,11 @@ static void __ceph_remove_cap(struct ceph_inode_info *ci, struct ceph_cap *cap,
>  
>  	mdsc = ceph_inode_to_fs_client(&ci->netfs.inode)->mdsc;
>  
> +	/* Losing the reporting cap requires fresh MDS revalidation. */
> +	if (READ_ONCE(ci->i_qtine_mds) == cap->mds &&
> +	    ceph_quarantine_reset(ci))
> +		wake_up_all(&ci->i_cap_wq);
> +
>  	/* remove from inode's cap rbtree, and clear auth cap */
>  	rb_erase(&cap->ci_node, &ci->i_caps);
>  	if (ci->i_auth_cap == cap)
> @@ -3729,6 +3735,9 @@ struct cap_extra_info {
>  	u8 *fscrypt_auth;
>  	u32 fscrypt_auth_len;
>  	u64 fscrypt_file_size;
> +	/* MClientCaps v14 oserrno; valid only when oserrno_valid is set */
> +	s32 oserrno;
> +	bool oserrno_valid;
>  };
>  
>  /*
> @@ -3757,6 +3766,46 @@ static int parse_inline_data(void **p, void *end,
>  	return -EIO;
>  }
>  
> +/*
> + * Record quarantine state carried by an MClientCaps oserrno.
> + * Caller holds i_ceph_lock. Returns true when waiters should be woken
> + * because quarantine changed or a pending reset was resolved.
> + *
> + * -EQUARANTINED refreshes the sticky flag and its reporting MDS rank.
> + * A current non-IMPORT grant above PIN with errno 0 clears that rank's mark.
> + * IMPORT may carry a default zero without assessing quarantine permission.
> + * Another rank needs a successful metadata request for the current epoch.
> + * A PIN-only update may zero qtine_errno while leaving the flag set.
> + */
> +static bool ceph_apply_quarantine(struct ceph_inode_info *ci, int mds,
> +				  int newcaps, int oserrno, bool is_import)
> +{
> +	bool was, now, pending;
> +
> +	was = READ_ONCE(ci->i_qtine_errno) == -CEPH_EQUARANTINED ||
> +	      test_bit(CEPH_I_QUARANTINE_BIT, &ci->i_ceph_flags);
> +	pending = test_bit(CEPH_I_QUARANTINE_PENDING_BIT, &ci->i_ceph_flags);
> +
> +	if (!is_import && (newcaps & ~CEPH_CAP_PIN) && was && oserrno == 0 &&
> +	    READ_ONCE(ci->i_qtine_mds) == mds)
> +		ceph_quarantine_clear(ci);
> +
> +	if (oserrno == -CEPH_EQUARANTINED) {
> +		ceph_quarantine_set(ci);
> +		WRITE_ONCE(ci->i_qtine_mds, mds);
> +	}

[Severity: High]
When a quarantined inode's capability is migrated between MDS ranks via
handle_cap_export() or handle_cap_import(), removing the exporting rank's
cap calls ceph_remove_cap() -> __ceph_remove_cap(), which sees
ci->i_qtine_mds == cap->mds and calls ceph_quarantine_reset(ci).

When handle_cap_grant() then processes the CEPH_CAP_OP_IMPORT message from
the new authoritative rank with the default oserrno == 0,
ceph_apply_quarantine() leaves CEPH_I_QUARANTINE_PENDING_BIT set and keeps
ci->i_qtine_mds set to the old exporting rank.

As a result:
1. ceph_inode_quarantine_direct() returns false and
   ceph_quarantine_walk_once() breaks out with CEPH_QWALK_INCOMPLETE instead
   of reporting the retained quarantine mark as CEPH_Q_UNDER.
2. Subsequent normal CEPH_CAP_OP_GRANT messages (above CEPH_CAP_PIN with
   oserrno == 0) and CEPH_MSG_CLIENT_QUARANTINE_DISABLE messages from the
   new authoritative rank cannot clear the mark in ceph_apply_quarantine()
   or ceph_handle_quarantine_disable() because ci->i_qtine_mds still holds
   the pre-migration rank.

Should capability migration preserve the active mark and transfer
ci->i_qtine_mds to the importing rank instead of resetting the mark to
pending when the old rank's cap is removed?

[ ... ]

> diff --git a/fs/ceph/inode.c b/fs/ceph/inode.c
> index 94f5215f8eed5..048e0d502eb4e 100644
> --- a/fs/ceph/inode.c
> +++ b/fs/ceph/inode.c

[ ... ]

> +static bool ceph_quarantine_reply_can_clear(struct ceph_inode_info *ci,
> +					    struct ceph_mds_session *session)
> +{
> +	struct ceph_mds_client *mdsc = ceph_sb_to_mdsc(ci->netfs.inode.i_sb);
> +	struct ceph_mds_request *req;
> +	u64 epoch;
> +
> +	if (READ_ONCE(ci->i_qtine_mds) == session->s_mds)
> +		return true;
> +	/* Different ranks have independent cap sequences. */
> +	req = ceph_current_mds_request(mdsc->fsc);
> +	epoch = (u64)atomic64_read_acquire(&mdsc->quarantine_epoch);
> +	return req && !(epoch & 1) && req->r_quarantine_epoch == epoch &&
> +		!atomic_read(&mdsc->quarantine_splices);
> +}

[ ... ]

> @@ -1407,7 +1873,7 @@ int ceph_fill_inode(struct inode *inode, struct folio *locked_folio,
>  	/* were we issued a capability? */
>  	if (info_caps) {
>  		if (!ceph_in_snap(inode)) {
> -			ceph_add_cap(inode, session,
> +			bool cap_current = ceph_add_cap(inode, session,
>  				     le64_to_cpu(info->cap.cap_id),
>  				     info_caps,
>  				     le32_to_cpu(info->cap.wanted),
> @@ -1416,6 +1882,14 @@ int ceph_fill_inode(struct inode *inode, struct folio *locked_folio,
>  				     le64_to_cpu(info->cap.realm),
>  				     info->cap.flags, &new_cap);
>  
> +			/*
> +			 * Caps above PIN mean quarantine has been lifted,
> +			 * including when the disable message was lost.
> +			 */
> +			if (cap_current && (info_caps & ~CEPH_CAP_PIN) &&
> +			    ceph_quarantine_reply_can_clear(ci, session))
> +				ceph_quarantine_clear(ci);

[Severity: High]
Can ceph_quarantine_reply_can_clear() fail to clear quarantine on target
inodes and readdir entries when the reply comes from a rank different from
ci->i_qtine_mds because reply processing itself increments
mdsc->quarantine_epoch?

req->r_quarantine_epoch is sampled once in ceph_mdsc_create_request() and
never updated:

  req->r_quarantine_epoch = atomic64_read_acquire(&mdsc->quarantine_epoch);

During reply handling, ceph_fill_trace() processes parent_dir before
req->r_target_inode:

  ceph_fill_trace()
    if (parent_dir)
      ceph_fill_inode(parent_dir, ...)
        ceph_quarantine_clear(ci) -> advances mdsc->quarantine_epoch by 2
    ...
    if (rinfo->head->is_target)
      ceph_fill_inode(in, ...)
        ceph_quarantine_reply_can_clear(ci, session)

If both parent_dir and req->r_target_inode had CEPH_I_QUARANTINE_BIT set on
another rank, clearing parent_dir increments mdsc->quarantine_epoch by 2.
When ceph_fill_inode() then runs for req->r_target_inode,
req->r_quarantine_epoch == epoch is false, leaving the target inode's
quarantine mark set.

The same self-invalidation happens in handle_reply() ->
ceph_readdir_prepopulate(), where each entry in the loop calls
ceph_fill_inode() and splice_dentry() (which calls
ceph_quarantine_splice_start() and ceph_quarantine_splice_finish(),
advancing mdsc->quarantine_epoch by 4), preventing
ceph_quarantine_reply_can_clear() from succeeding on subsequent entries in
the same readdir reply.

In addition, any unrelated dentry splice or cross-directory rename on the
superblock while the MDS request is in flight (or sampling an odd
mdsc->quarantine_epoch in ceph_mdsc_create_request()) will cause
req->r_quarantine_epoch == epoch to fail even when no new quarantine mark was
set.

-- 
Sashiko AI review · https://sashiko.dev/#/patchset/cover.20261025T182649Z.ca9ae7a18449.git.amarkuze@redhat.com?part=4

Re: [PATCH v6 5/5] ceph: fail quarantined subvolume access with EACCES

[sashiko-bot]

Thank you for your contribution! Sashiko AI review found 4 potential issue(s) to consider:
- [High] ceph: missing head cap revalidation in ceph_open() and write cap paths
- [High] ceph: stale xattr cache in __ceph_setxattr() sync fallback with XATTR_EXCL
- [High] ceph: infinite loop in __ceph_get_caps() when inline data reply lacks cache caps
- [Medium] ceph: missing head cap revalidation in __ceph_getxattr() and ceph_listxattr()
--

commit 3b0481a00dbf8afa6f80793356dab3d1a012a099
Author: Alex Markuze <amarkuze@redhat.com>

ceph: fail quarantined subvolume access with EACCES

This commit adds client-side enforcement for quarantined subvolumes before
advertising CEPHFS_FEATURE_QUARANTINE to the MDS. It revalidates cached
data, metadata, xattrs, ACLs, symlinks, and snapshot capabilities against
quarantine state and requires MDS authorization when local state or
revocation coverage is insufficient.

> diff --git a/fs/ceph/addr.c b/fs/ceph/addr.c
> index dd2bd8ed24249..79e396752601f 100644
> --- a/fs/ceph/addr.c
> +++ b/fs/ceph/addr.c
[ ... ]
> @@ -2649,9 +2674,24 @@ static const struct vm_operations_struct ceph_vmops = {
>  int ceph_mmap_prepare(struct vm_area_desc *desc)
>  {
>  	struct address_space *mapping = desc->file->f_mapping;
> +	struct inode *inode = file_inode(desc->file);
> +	int err, got = 0;
>  
>  	if (!mapping->a_ops->read_folio)
>  		return -ENOEXEC;
> +	err = ceph_quarantine_revalidate(inode);
> +	if (err)
> +		return err;
> +	if (vma_desc_test_all(desc, VMA_SHARED_BIT, VMA_WRITE_BIT) &&
> +	    ceph_inode_quarantine_write_state(inode) != CEPH_Q_CLEAR) {
> +		/* A read receipt cannot admit a writable shared mapping. */
> +		err = ceph_get_caps(desc->file, CEPH_CAP_FILE_WR, 0, -1, &got);
> +		if (err)
> +			return err;
> +		ceph_put_cap_refs(ceph_inode(inode), got);
> +	}

[Severity: High]
Can a writable shared mapping be admitted here using only a read-only
GETATTR confirmation when mdsc->quarantine_count is zero?

In __ceph_inode_quarantine_state(), both ceph_inode_quarantine_state() and
ceph_inode_quarantine_write_state() return CEPH_Q_CLEAR immediately whenever
atomic_read(&mdsc->quarantine_count) == 0:

fs/ceph/inode.c:__ceph_inode_quarantine_state() {
	...
	mdsc = ceph_sb_to_mdsc(inode->i_sb);
	if (!mdsc || atomic_read(&mdsc->quarantine_count) == 0)
		return CEPH_Q_CLEAR;
	...
}

If a subvolume is already quarantined before the client caches caps above
CEPH_CAP_PIN (or after marked inodes are evicted), mdsc->quarantine_count is
0. When a client with read-only recovery permission (q, but not Q) calls
ceph_mmap_prepare(), ceph_quarantine_revalidate() succeeds via GETATTR
without granting CEPH_CAP_AUTH_SHARED, leaving
ceph_quarantine_head_needs_revalidate(inode) true.

Because ceph_inode_quarantine_write_state(inode) still returns CEPH_Q_CLEAR
when quarantine_count is 0, the check above is false and ceph_get_caps() for
CEPH_CAP_FILE_WR is skipped.

The same issue also affects __ceph_get_caps() and ceph_open() when
quarantine_count is 0 and ceph_quarantine_head_needs_revalidate(inode) is
true (for example, when ci->i_auth_cap only holds CEPH_CAP_PIN from a prior
read-only LOOKUP or GETATTR). In ceph_open() with FMODE_WRITE, do_sync
remains false, allowing the fast path to succeed locally on the PIN-only
ci->i_auth_cap without sending CEPH_MDS_OP_OPEN, and triggering
ceph_uninline_data() on inline files.

Should ceph_mmap_prepare(), __ceph_get_caps(), and ceph_open() also check
ceph_quarantine_head_needs_revalidate(inode) on write paths?

> diff --git a/fs/ceph/caps.c b/fs/ceph/caps.c
> index b07814d0038cd..7c85e84fa6493 100644
> --- a/fs/ceph/caps.c
> +++ b/fs/ceph/caps.c
[ ... ]
> @@ -3304,15 +3396,71 @@ int __ceph_get_caps(struct inode *inode, struct ceph_file_info *fi, int need,
>  
>  	while (true) {
>  		flags &= CEPH_FILE_MODE_MASK;
> +		if (snapshot && ceph_snapshot_needs_revalidate(inode)) {
> +			/*
> +			 * i_snap_caps cannot be revoked, and an idle snapshot
> +			 * inode may receive no quarantine notification at all.
> +			 * An MDS answer authorizes only this acquisition.
> +			 */
> +			qepoch = atomic64_read_acquire(&mdsc->quarantine_epoch);
> +			if (!ceph_cap_auth_valid(mdsc, qepoch)) {
> +				ret = ceph_quarantine_wait(inode);
> +				if (ret)
> +					return ret;
> +				continue;
> +			}
> +			ret = ceph_snapshot_read_auth(inode);
> +			if (ret)
> +				return ret;
> +			flags |= MDS_AUTHORIZED;

[Severity: High]
Does __ceph_get_caps() loop infinitely when reading a snapshot file with
inline data whose MDS reply authorizes access without granting
CEPH_CAP_FILE_CACHE or CEPH_CAP_FILE_LAZYIO in info->cap.caps?

If a snapshot regular file with inline data (ceph_has_inline_data(ci) &&
i_size_read(inode) > 0) previously recorded CEPH_CAP_FILE_RD |
CEPH_CAP_FILE_CACHE in ci->i_snap_caps, those bits are sticky in
__ceph_caps_issued(). When folio 0 is not uptodate (for example after
ceph_read_folio() unlocks a newly allocated folio and calls
__ceph_get_caps()) and ceph_snapshot_needs_revalidate(inode) is true:

1. ceph_snapshot_read_auth(inode) succeeds and sets MDS_AUTHORIZED.
2. try_get_cap_refs() grants _got = CEPH_CAP_FILE_RD | CEPH_CAP_FILE_CACHE
   from ci->i_snap_caps.
3. At the bottom of the loop in __ceph_get_caps(), folio 0 is not uptodate,
   so __ceph_get_caps() drops _got and calls:

   __ceph_do_getattr(inode, NULL, CEPH_STAT_CAP_INLINE_DATA, true)

4. In ceph_fill_inode(), fill_inline is only set when locked_folio is
   non-NULL or info_caps from the current reply includes cache_caps:

fs/ceph/inode.c:ceph_fill_inode() {
	...
	if (iinfo->inline_version > 0 &&
	    iinfo->inline_version >= ceph_inline_version(ci)) {
		int cache_caps = CEPH_CAP_FILE_CACHE | CEPH_CAP_FILE_LAZYIO;
		ceph_set_inline_version(ci, iinfo->inline_version);
		if (ceph_has_inline_data(ci) &&
		    (locked_folio || (info_caps & cache_caps)))
			fill_inline = true;
	}
	...
}

When the reply authorizes the read (such as for a recovery client or via
LOOKUPINO) and returns inline data with targeti.inline_version > 1, but
info->cap.caps does not include CEPH_CAP_FILE_CACHE | CEPH_CAP_FILE_LAZYIO,
ceph_fill_inode() skips ceph_fill_inline_data(), __ceph_do_getattr() returns
0, and __ceph_get_caps() executes continue and repeats the RPC loop
indefinitely because folio 0 is never marked uptodate.

> diff --git a/fs/ceph/xattr.c b/fs/ceph/xattr.c
> index 4cba6e7bffd3d..13d4dfef679de 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_snapshot_needs_revalidate(inode) ||
> +			  ceph_inode_quarantine_state(inode) != CEPH_Q_CLEAR;
> +		spin_lock(&ci->i_ceph_lock);
> +	}

[Severity: Medium]
Should __ceph_getxattr() and ceph_listxattr() also check
ceph_quarantine_head_needs_revalidate(inode) when computing ask_mds, just as
ceph_getattr() does?

If an inode holds CEPH_CAP_XATTR_SHARED (for example on a replica MDS cap,
on a session without CEPHFS_FEATURE_QUARANTINE, or after
CEPH_CAP_AUTH_SHARED was released) while ceph_inode_quarantine_state(inode)
returns CEPH_Q_CLEAR, omitting ceph_quarantine_head_needs_revalidate(inode)
allows __ceph_getxattr() and ceph_listxattr() to serve cached xattrs and
POSIX ACLs without consulting the authoritative MDS.

[ ... ]
> @@ -1319,6 +1343,34 @@ 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);
> +			old_blob = NULL;
> +			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;
> +	}

[Severity: High]
When __ceph_setxattr() takes the goto do_sync_unlocked path here while the
inode holds CEPH_CAP_XATTR_EXCL, can the local xattr cache become stale or
overwrite the synchronous update on a later cap flush?

__ceph_setxattr() only reaches this check when ci->i_xattrs.version != 0 and
issued & CEPH_CAP_XATTR_EXCL is held, after __build_xattrs(inode) has
already built ci->i_xattrs.index.

Jumping to do_sync_unlocked calls ceph_sync_setxattr(), which only sets:

	req->r_inode_drop = CEPH_CAP_XATTR_SHARED;

It does not drop or flush CEPH_CAP_XATTR_EXCL. When the MDS reply arrives in
ceph_fill_inode(), issued is computed from cap->implemented |
__ceph_caps_dirty(ci), and the xattr update is skipped whenever
issued & CEPH_CAP_XATTR_EXCL is set:

fs/ceph/inode.c:ceph_fill_inode() {
	...
	if ((ci->i_xattrs.version == 0 || !(issued & CEPH_CAP_XATTR_EXCL))  &&
	    le64_to_cpu(info->xattr_version) > ci->i_xattrs.version) {
		...
	}
	...
}

Unlike __ceph_setattr(), which explicitly applies setattr_copy() after a
forced synchronous request when exclusive caps are held, __ceph_setxattr()
leaves the old blob and ci->i_xattrs.index intact, and if
CEPH_CAP_XATTR_EXCL was already dirty from an earlier cached setxattr, the
subsequent cap flush will encode the stale RB-tree and lose the synchronous
xattr modification.

-- 
Sashiko AI review · https://sashiko.dev/#/patchset/cover.20261025T182649Z.ca9ae7a18449.git.amarkuze@redhat.com?part=5