ceph: do link/rename semantic checks after srcdn is readable

For hard link, source inode must not be directory. For rename,
types of source/destination inodes must match. If srcdn is replica
and we do these checks while it's not readble, it's possible that
wrong source inode is used in these checks.

Fixes: http://tracker.ceph.com/issues/21383
Signed-off-by: "Yan, Zheng" <zyan@redhat.com>
(cherry picked from commit 46962b253563a867707e7c5d7887abf2060cc4d7)

Conflicts:
	src/mds/Server.cc (different placement of need_auth assignment in jewel)

diff --git a/src/mds/Server.cc b/src/mds/Server.cc
index 236227a5f5529edb9483a82d153a325c3edfdb07..4d791d09830cdd9e280e2cc490d44f10e800e1e6 100644 (file)
--- a/src/mds/Server.cc
+++ b/src/mds/Server.cc
@@ -2887,7 +2887,8 @@ void Server::handle_client_open(MDRequestRef& mdr)
   int flags = req->head.args.open.flags;
   int cmode = ceph_flags_to_mode(flags);
 
-  bool need_auth = !file_mode_is_readonly(cmode) || (flags & CEPH_O_TRUNC);
+  bool need_auth = !file_mode_is_readonly(cmode) ||
+                  (flags & (CEPH_O_TRUNC | CEPH_O_DIRECTORY));
 
   dout(7) << "open on " << req->get_filepath() << dendl;
 
@@ -4881,9 +4882,15 @@ void Server::handle_client_link(MDRequestRef& mdr)
   dout(7) << "handle_client_link link " << dn->get_name() << " in " << *dir << dendl;
   dout(7) << "target is " << *targeti << dendl;
   if (targeti->is_dir()) {
-    dout(7) << "target is a dir, failing..." << dendl;
-    respond_to_request(mdr, -EINVAL);
-    return;
+    // if srcdn is replica, need to make sure its linkage is correct
+    vector<CDentry*>& trace = mdr->dn[1];
+    if (trace.empty() ||
+       trace.back()->is_auth() ||
+       trace.back()->lock.can_read(mdr->get_client())) {
+      dout(7) << "target is a dir, failing..." << dendl;
+      respond_to_request(mdr, -EINVAL);
+      return;
+    }
   }
 
   xlocks.insert(&targeti->linklock);
@@ -6145,25 +6152,30 @@ void Server::handle_client_rename(MDRequestRef& mdr)
     oldin = mdcache->get_dentry_inode(destdn, mdr, true);
     if (!oldin) return;
     dout(10) << " oldin " << *oldin << dendl;
-    
-    // mv /some/thing /to/some/existing_other_thing
-    if (oldin->is_dir() && !srci->is_dir()) {
-      respond_to_request(mdr, -EISDIR);
-      return;
-    }
-    if (!oldin->is_dir() && srci->is_dir()) {
-      respond_to_request(mdr, -ENOTDIR);
-      return;
-    }
 
     // non-empty dir?
     if (oldin->is_dir() && _dir_is_nonempty_unlocked(mdr, oldin)) {
       respond_to_request(mdr, -ENOTEMPTY);
       return;
     }
-    if (srci == oldin && !srcdn->get_dir()->inode->is_stray()) {
-      respond_to_request(mdr, 0);  // no-op.  POSIX makes no sense.
-      return;
+
+    // if srcdn is replica, need to make sure its linkage is correct
+    if (srcdn->is_auth() ||
+       srcdn->lock.can_read(mdr->get_client()) ||
+       (srcdn->lock.is_xlocked() && srcdn->lock.get_xlock_by() == mdr)) {
+      // mv /some/thing /to/some/existing_other_thing
+      if (oldin->is_dir() && !srci->is_dir()) {
+       respond_to_request(mdr, -EISDIR);
+       return;
+      }
+      if (!oldin->is_dir() && srci->is_dir()) {
+       respond_to_request(mdr, -ENOTDIR);
+       return;
+      }
+      if (srci == oldin && !srcdn->get_dir()->inode->is_stray()) {
+       respond_to_request(mdr, 0);  // no-op.  POSIX makes no sense.
+       return;
+      }
     }
   }