mds: prevent clients from exceeding the xattrs key/value limits

Commit eb915d0 ("cephfs: fix write_buf's _len overflow problem") added
a limit to the total size of xattrs.  This limit is respected by clients
doing a "sync" operation, i.e. MDS_OP_SETXATTR.  However, clients with
CAP_XATTR_EXCL can still buffer these operations and ignore these limits.

This patch prevents clients from crashing the MDSs by also imposing the
xattr limits even when they have the Xx caps.  Replaces the per-MDS knob
"max_xattr_pairs_size" by the new mdsmap setting that the clients can
access.

Unfortunately, clients that misbehave, i.e. old clients that don't respect
this xattrs limit and buffer their xattrs, will see them vanishing.

URL: https://tracker.ceph.com/issues/55725
Signed-off-by: Luís Henriques <lhenriques@suse.de>

src/common/options/mds.yaml.in

@@ -65,15 +65,6 @@ options:
   - mds
   flags:
   - runtime
-# max xattr kv pairs size for each dir/file
-- name: mds_max_xattr_pairs_size
-  type: size
-  level: advanced
-  desc: maximum aggregate size of extended attributes on a file
-  default: 64_K
-  services:
-  - mds
-  with_legacy: true
 - name: mds_cache_trim_interval
   type: secs
   level: advanced

src/mds/Locker.cc

@@ -3568,6 +3568,36 @@ void Locker::kick_cap_releases(MDRequestRef& mdr)
   }
 }
 
+__u32 Locker::get_xattr_total_length(CInode::mempool_xattr_map &xattr)
+{
+  __u32 total = 0;
+
+  for (const auto &p : xattr)
+    total += (p.first.length() + p.second.length());
+  return total;
+}
+
+void Locker::decode_new_xattrs(CInode::mempool_inode *inode,
+			       CInode::mempool_xattr_map *px,
+			       const cref_t<MClientCaps> &m)
+{
+  CInode::mempool_xattr_map tmp;
+
+  auto p = m->xattrbl.cbegin();
+  decode_noshare(tmp, p);
+  __u32 total = get_xattr_total_length(tmp);
+  inode->xattr_version = m->head.xattr_version;
+  if (total > mds->mdsmap->get_max_xattr_size()) {
+    dout(1) << "Maximum xattr size exceeded: " << total
+	    << " max size: " << mds->mdsmap->get_max_xattr_size() << dendl;
+    // Ignore new xattr (!!!) but increase xattr version
+    // XXX how to force the client to drop cached xattrs?
+    inode->xattr_version++;
+  } else {
+    *px = std::move(tmp);
+  }
+}
+
 /**
  * m and ack might be NULL, so don't dereference them unless dirty != 0
  */
@@ -3638,10 +3668,8 @@ void Locker::_do_snap_update(CInode *in, snapid_t snap, int dirty, snapid_t foll
   // xattr
   if (xattrs) {
     dout(7) << " xattrs v" << i->xattr_version << " -> " << m->head.xattr_version
-	    << " len " << m->xattrbl.length() << dendl;
-    i->xattr_version = m->head.xattr_version;
-    auto p = m->xattrbl.cbegin();
-    decode(*px, p);
+            << " len " << m->xattrbl.length() << dendl;
+    decode_new_xattrs(i, px, m);
   }
 
   {
@@ -3933,9 +3961,7 @@ bool Locker::_do_cap_update(CInode *in, Capability *cap,
   // xattrs update?
   if (xattr) {
     dout(7) << " xattrs v" << pi.inode->xattr_version << " -> " << m->head.xattr_version << dendl;
-    pi.inode->xattr_version = m->head.xattr_version;
-    auto p = m->xattrbl.cbegin();
-    decode_noshare(*pi.xattrs, p);
+    decode_new_xattrs(pi.inode.get(), pi.xattrs.get(), m);
     wrlock_force(&in->xattrlock, mut);
   }
 

src/mds/Locker.h

@@ -262,6 +262,10 @@ class Locker {
 
   bool any_late_revoking_caps(xlist<Capability*> const &revoking, double timeout) const;
   uint64_t calc_new_max_size(const CInode::inode_const_ptr& pi, uint64_t size);
+  __u32 get_xattr_total_length(CInode::mempool_xattr_map &xattr);
+  void decode_new_xattrs(CInode::mempool_inode *inode,
+			 CInode::mempool_xattr_map *px,
+			 const cref_t<MClientCaps> &m);
 
   MDSRank *mds;
   MDCache *mdcache;

src/mds/Server.cc

@@ -4767,7 +4767,7 @@ void Server::handle_client_readdir(MDRequestRef& mdr)
   unsigned max_bytes = req->head.args.readdir.max_bytes;
   if (!max_bytes)
     // make sure at least one item can be encoded
-    max_bytes = (512 << 10) + g_conf()->mds_max_xattr_pairs_size;
+    max_bytes = (512 << 10) + mds->mdsmap->get_max_xattr_size();
 
   // start final blob
   bufferlist dirbl;
@@ -6461,7 +6461,7 @@ void Server::handle_client_setxattr(MDRequestRef& mdr)
       cur_xattrs_size += p.first.length() + p.second.length();
     }
 
-    if (((cur_xattrs_size + inc) > g_conf()->mds_max_xattr_pairs_size)) {
+    if (((cur_xattrs_size + inc) > mds->mdsmap->get_max_xattr_size())) {
       dout(10) << "xattr kv pairs size too big. cur_xattrs_size "
 	<< cur_xattrs_size << ", inc " << inc << dendl;
       respond_to_request(mdr, -CEPHFS_ENOSPC);
@@ -10755,7 +10755,7 @@ void Server::handle_client_lssnap(MDRequestRef& mdr)
   int max_bytes = req->head.args.readdir.max_bytes;
   if (!max_bytes)
     // make sure at least one item can be encoded
-    max_bytes = (512 << 10) + g_conf()->mds_max_xattr_pairs_size;
+    max_bytes = (512 << 10) + mds->mdsmap->get_max_xattr_size();
 
   __u64 last_snapid = 0;
   string offset_str = req->get_path2();