LU-1623 mdt: Atomically update MDT export connection flags

MDT processing of connect requests currently updates the export
connection flags in two steps: client/server feature matching is
performed first, then much later various security-related bits are
removed as needed.  Certain error paths may leave the export flags
partially initialized.

A problem arises if multiple connect requests from the same client are
handled out of order, as may occur due to network disruptions. If the
last such request to be handled has a lower connection count than one
that already completed, it will be aborted with -EALREADY after having
modified the connection flags in the export.  However, the
security-related flags are left with incorrect values, as the
top-level connect handler skips setting these in the error path.
Replies to subsequent client requests may then contain unexpected
security information, causing the client to crash.

Similar issues may exist with other target types having non-atomic
export flag updates, and these should be addressed in follow-up
patches.

This patch makes the following changes:

- To avoid the problem described above, update the export connection
  flags atomically, and only in the successful case.

- To make this important atomic operation more conspicuous, move it
  from mdt_init_sec_level() to the top-level handler mdt_connect().

- Add a comment to mdt_connect_internal(), and delete a disabled code
  block from it.

- Correct debug message in target_handle_connect() to match code.

Signed-off-by: Ned Bass <bass6@llnl.gov>
LLNL-bug-id: bz1711
Change-Id: Ic00c4679cc9b813bdb47cf148bef8f62c0ef8ddb
Reviewed-on: http://review.whamcloud.com/4406
Tested-by: Hudson
Tested-by: Maloo <whamcloud.maloo@gmail.com>
Reviewed-by: Andreas Dilger <adilger@whamcloud.com>
Reviewed-by: Mike Pershin <tappro@whamcloud.com>
Reviewed-by: Oleg Drokin <green@whamcloud.com>

diff --git a/lustre/ldlm/ldlm_lib.c b/lustre/ldlm/ldlm_lib.c
index 3e3660f..bba19e9 100644 (file)
--- a/lustre/ldlm/ldlm_lib.c
+++ b/lustre/ldlm/ldlm_lib.c
@@ -1121,8 +1121,8 @@ dont_check_exports:
         cfs_spin_lock(&export->exp_lock);
         if (export->exp_conn_cnt >= lustre_msg_get_conn_cnt(req->rq_reqmsg)) {
                 cfs_spin_unlock(&export->exp_lock);
         cfs_spin_lock(&export->exp_lock);
         if (export->exp_conn_cnt >= lustre_msg_get_conn_cnt(req->rq_reqmsg)) {
                 cfs_spin_unlock(&export->exp_lock);

-                CDEBUG(D_RPCTRACE, "%s: %s already connected at higher "
-                       "conn_cnt: %d > %d\n",
+               CDEBUG(D_RPCTRACE, "%s: %s already connected at greater "
+                      "or equal conn_cnt: %d >= %d\n",

                        cluuid.uuid, libcfs_nid2str(req->rq_peer.nid),
                        export->exp_conn_cnt,
                        lustre_msg_get_conn_cnt(req->rq_reqmsg));
                        cluuid.uuid, libcfs_nid2str(req->rq_peer.nid),
                        export->exp_conn_cnt,
                        lustre_msg_get_conn_cnt(req->rq_reqmsg));


@@ -1237,13 +1237,13 @@ dont_check_exports:
          * ptlrpc_handle_server_req_in->lustre_unpack_msg() */
         revimp->imp_msg_magic = req->rq_reqmsg->lm_magic;
 
          * ptlrpc_handle_server_req_in->lustre_unpack_msg() */
         revimp->imp_msg_magic = req->rq_reqmsg->lm_magic;
 

-        if ((export->exp_connect_flags & OBD_CONNECT_AT) &&
-            (revimp->imp_msg_magic != LUSTRE_MSG_MAGIC_V1))
-                revimp->imp_msghdr_flags |= MSGHDR_AT_SUPPORT;
-        else
-                revimp->imp_msghdr_flags &= ~MSGHDR_AT_SUPPORT;
+       if ((data->ocd_connect_flags & OBD_CONNECT_AT) &&
+           (revimp->imp_msg_magic != LUSTRE_MSG_MAGIC_V1))
+               revimp->imp_msghdr_flags |= MSGHDR_AT_SUPPORT;
+       else
+               revimp->imp_msghdr_flags &= ~MSGHDR_AT_SUPPORT;

 
 

-        if ((export->exp_connect_flags & OBD_CONNECT_FULL20) &&
+       if ((data->ocd_connect_flags & OBD_CONNECT_FULL20) &&

             (revimp->imp_msg_magic != LUSTRE_MSG_MAGIC_V1))
                 revimp->imp_msghdr_flags |= MSGHDR_CKSUM_INCOMPAT18;
         else
             (revimp->imp_msg_magic != LUSTRE_MSG_MAGIC_V1))
                 revimp->imp_msghdr_flags |= MSGHDR_CKSUM_INCOMPAT18;
         else

diff --git a/lustre/mdt/mdt_handler.c b/lustre/mdt/mdt_handler.c
index 83056b6..30347c9 100644 (file)
--- a/lustre/mdt/mdt_handler.c
+++ b/lustre/mdt/mdt_handler.c
@@ -1500,25 +1500,43 @@ static int mdt_set_info(struct mdt_thread_info *info)
         RETURN(0);
 }
 
         RETURN(0);
 }
 

+/**
+ * Top-level handler for MDT connection requests.
+ */

 static int mdt_connect(struct mdt_thread_info *info)
 {
 static int mdt_connect(struct mdt_thread_info *info)
 {

-        int rc;
-        struct ptlrpc_request *req;
+       int rc;
+       struct obd_connect_data *reply;
+       struct obd_export *exp;
+       struct ptlrpc_request *req = mdt_info_req(info);

 
 

-        req = mdt_info_req(info);
-        rc = target_handle_connect(req);
-        if (rc == 0) {
-                LASSERT(req->rq_export != NULL);
-                info->mti_mdt = mdt_dev(req->rq_export->exp_obd->obd_lu_dev);
-                rc = mdt_init_sec_level(info);
-                if (rc == 0)
-                        rc = mdt_init_idmap(info);
-                if (rc != 0)
-                        obd_disconnect(class_export_get(req->rq_export));
-        } else {
-                rc = err_serious(rc);
-        }
-        return rc;
+       rc = target_handle_connect(req);
+       if (rc != 0)
+               return err_serious(rc);
+
+       LASSERT(req->rq_export != NULL);
+       info->mti_mdt = mdt_dev(req->rq_export->exp_obd->obd_lu_dev);
+       rc = mdt_init_sec_level(info);
+       if (rc != 0) {
+               obd_disconnect(class_export_get(req->rq_export));
+               return rc;
+       }
+
+       /* To avoid exposing partially initialized connection flags, changes up
+        * to this point have been staged in reply->ocd_connect_flags. Now that
+        * connection handling has completed successfully, atomically update
+        * the connect flags in the shared export data structure. LU-1623 */
+       reply = req_capsule_server_get(info->mti_pill, &RMF_CONNECT_DATA);
+       exp = req->rq_export;
+       cfs_spin_lock(&exp->exp_lock);
+       exp->exp_connect_flags = reply->ocd_connect_flags;
+       cfs_spin_unlock(&exp->exp_lock);
+
+       rc = mdt_init_idmap(info);
+       if (rc != 0)
+               obd_disconnect(class_export_get(req->rq_export));
+
+       return rc;

 }
 
 static int mdt_disconnect(struct mdt_thread_info *info)
 }
 
 static int mdt_disconnect(struct mdt_thread_info *info)


@@ -5395,79 +5413,97 @@ static int mdt_obd_set_info_async(const struct lu_env *env,
         RETURN(0);
 }
 
         RETURN(0);
 }
 

-/* mds_connect_internal */
+/**
+ * Match client and server connection feature flags.
+ *
+ * Compute the compatibility flags for a connection request based on
+ * features mutually supported by client and server.
+ *
+ * The obd_export::exp_connect_flags field in \a exp must not be updated
+ * here, otherwise a partially initialized value may be exposed. After
+ * the connection request is successfully processed, the top-level MDT
+ * connect request handler atomically updates the export connect flags
+ * from the obd_connect_data::ocd_connect_flags field of the reply.
+ * \see mdt_connect().
+ *
+ * \param exp   the obd_export associated with this client/target pair
+ * \param mdt   the target device for the connection
+ * \param data  stores data for this connect request
+ *
+ * \retval 0       success
+ * \retval -EPROTO \a data unexpectedly has zero obd_connect_data::ocd_brw_size
+ * \retval -EBADE  client and server feature requirements are incompatible
+ */

 static int mdt_connect_internal(struct obd_export *exp,
 static int mdt_connect_internal(struct obd_export *exp,

-                                struct mdt_device *mdt,
-                                struct obd_connect_data *data)
-{
-        if (data != NULL) {
-                data->ocd_connect_flags &= MDT_CONNECT_SUPPORTED;
-                data->ocd_ibits_known &= MDS_INODELOCK_FULL;
-
-                /* If no known bits (which should not happen, probably,
-                   as everybody should support LOOKUP and UPDATE bits at least)
-                   revert to compat mode with plain locks. */
-                if (!data->ocd_ibits_known &&
-                    data->ocd_connect_flags & OBD_CONNECT_IBITS)
-                        data->ocd_connect_flags &= ~OBD_CONNECT_IBITS;
-
-                if (!mdt->mdt_opts.mo_acl)
-                        data->ocd_connect_flags &= ~OBD_CONNECT_ACL;
-
-                if (!mdt->mdt_opts.mo_user_xattr)
-                        data->ocd_connect_flags &= ~OBD_CONNECT_XATTR;
-
-                if (!mdt->mdt_som_conf)
-                        data->ocd_connect_flags &= ~OBD_CONNECT_SOM;
-
-                if (data->ocd_connect_flags & OBD_CONNECT_BRW_SIZE) {
-                        data->ocd_brw_size = min(data->ocd_brw_size,
-                               (__u32)(PTLRPC_MAX_BRW_PAGES << CFS_PAGE_SHIFT));
-                        if (data->ocd_brw_size == 0) {
-                                CERROR("%s: cli %s/%p ocd_connect_flags: "LPX64
-                                       " ocd_version: %x ocd_grant: %d "
-                                       "ocd_index: %u ocd_brw_size is "
-                                       "unexpectedly zero, network data "
-                                       "corruption? Refusing connection of this"
-                                       " client\n",
-                                       exp->exp_obd->obd_name,
-                                       exp->exp_client_uuid.uuid,
-                                       exp, data->ocd_connect_flags, data->ocd_version,
-                                       data->ocd_grant, data->ocd_index);
-                                return -EPROTO;
-                        }
-                }
+                               struct mdt_device *mdt,
+                               struct obd_connect_data *data)
+{
+       LASSERT(data != NULL);
+
+       data->ocd_connect_flags &= MDT_CONNECT_SUPPORTED;
+       data->ocd_ibits_known &= MDS_INODELOCK_FULL;
+
+       /* If no known bits (which should not happen, probably,
+          as everybody should support LOOKUP and UPDATE bits at least)
+          revert to compat mode with plain locks. */
+       if (!data->ocd_ibits_known &&
+           data->ocd_connect_flags & OBD_CONNECT_IBITS)
+               data->ocd_connect_flags &= ~OBD_CONNECT_IBITS;
+
+       if (!mdt->mdt_opts.mo_acl)
+               data->ocd_connect_flags &= ~OBD_CONNECT_ACL;
+
+       if (!mdt->mdt_opts.mo_user_xattr)
+               data->ocd_connect_flags &= ~OBD_CONNECT_XATTR;
+
+       if (!mdt->mdt_som_conf)
+               data->ocd_connect_flags &= ~OBD_CONNECT_SOM;
+
+       if (data->ocd_connect_flags & OBD_CONNECT_BRW_SIZE) {
+               data->ocd_brw_size = min(data->ocd_brw_size,
+                       (__u32)(PTLRPC_MAX_BRW_PAGES << CFS_PAGE_SHIFT));
+               if (data->ocd_brw_size == 0) {
+                       CERROR("%s: cli %s/%p ocd_connect_flags: "LPX64
+                              " ocd_version: %x ocd_grant: %d "
+                              "ocd_index: %u ocd_brw_size is "
+                              "unexpectedly zero, network data "
+                              "corruption? Refusing connection of this"
+                              " client\n",
+                              exp->exp_obd->obd_name,
+                              exp->exp_client_uuid.uuid,
+                              exp, data->ocd_connect_flags, data->ocd_version,
+                              data->ocd_grant, data->ocd_index);
+                       return -EPROTO;
+               }
+       }

 
 

-                cfs_spin_lock(&exp->exp_lock);
-                exp->exp_connect_flags = data->ocd_connect_flags;
-                cfs_spin_unlock(&exp->exp_lock);
-                data->ocd_version = LUSTRE_VERSION_CODE;
-                exp->exp_mdt_data.med_ibits_known = data->ocd_ibits_known;
-        }
+       /* NB: Disregard the rule against updating exp_connect_flags in this
+        * case, since tgt_client_new() needs to know if this is a lightweight
+        * connection, and it is safe to expose this flag before connection
+        * processing completes. */
+       if (data->ocd_connect_flags & OBD_CONNECT_LIGHTWEIGHT) {
+               cfs_spin_lock(&exp->exp_lock);
+               exp->exp_connect_flags |=  OBD_CONNECT_LIGHTWEIGHT;
+               cfs_spin_unlock(&exp->exp_lock);
+       }

 
 

-#if 0
-        if (mdt->mdt_opts.mo_acl &&
-            ((exp->exp_connect_flags & OBD_CONNECT_ACL) == 0)) {
-                CWARN("%s: MDS requires ACL support but client does not\n",
-                      mdt->mdt_md_dev.md_lu_dev.ld_obd->obd_name);
-                return -EBADE;
-        }
-#endif
+       data->ocd_version = LUSTRE_VERSION_CODE;
+       exp->exp_mdt_data.med_ibits_known = data->ocd_ibits_known;

 
 

-        if ((exp->exp_connect_flags & OBD_CONNECT_FID) == 0) {
-                CWARN("%s: MDS requires FID support, but client not\n",
-                      mdt->mdt_md_dev.md_lu_dev.ld_obd->obd_name);
-                return -EBADE;
-        }
+       if ((data->ocd_connect_flags & OBD_CONNECT_FID) == 0) {
+               CWARN("%s: MDS requires FID support, but client not\n",
+                     mdt->mdt_md_dev.md_lu_dev.ld_obd->obd_name);
+               return -EBADE;
+       }

 
 

-        if (mdt->mdt_som_conf && !exp_connect_som(exp) &&
-            !(exp->exp_connect_flags & OBD_CONNECT_MDS_MDS)) {
-                CWARN("%s: MDS has SOM enabled, but client does not support "
-                      "it\n", mdt->mdt_md_dev.md_lu_dev.ld_obd->obd_name);
-                return -EBADE;
-        }
+       if (mdt->mdt_som_conf &&
+           !(data->ocd_connect_flags & (OBD_CONNECT_MDS_MDS|OBD_CONNECT_SOM))){
+               CWARN("%s: MDS has SOM enabled, but client does not support "
+                     "it\n", mdt->mdt_md_dev.md_lu_dev.ld_obd->obd_name);
+               return -EBADE;
+       }

 
 

-        return 0;
+       return 0;

 }
 
 static int mdt_connect_check_sptlrpc(struct mdt_device *mdt,
 }
 
 static int mdt_connect_check_sptlrpc(struct mdt_device *mdt,

diff --git a/lustre/mdt/mdt_idmap.c b/lustre/mdt/mdt_idmap.c
index 68f4391..1d81098 100644 (file)
--- a/lustre/mdt/mdt_idmap.c
+++ b/lustre/mdt/mdt_idmap.c
@@ -73,9 +73,6 @@ do {                                                                    \
                                       OBD_CONNECT_RMT_CLIENT_FORCE |    \
                                       OBD_CONNECT_MDS_CAPA |            \
                                       OBD_CONNECT_OSS_CAPA);            \
                                       OBD_CONNECT_RMT_CLIENT_FORCE |    \
                                       OBD_CONNECT_MDS_CAPA |            \
                                       OBD_CONNECT_OSS_CAPA);            \

-        cfs_spin_lock(&exp->exp_lock);                                  \
-        exp->exp_connect_flags = reply->ocd_connect_flags;              \
-        cfs_spin_unlock(&exp->exp_lock);                                \

 } while (0)
 
 int mdt_init_sec_level(struct mdt_thread_info *info)
 } while (0)
 
 int mdt_init_sec_level(struct mdt_thread_info *info)


@@ -180,10 +177,6 @@ int mdt_init_sec_level(struct mdt_thread_info *info)
                                 reply->ocd_connect_flags &= ~OBD_CONNECT_MDS_CAPA;
                         if (!mdt->mdt_opts.mo_oss_capa)
                                 reply->ocd_connect_flags &= ~OBD_CONNECT_OSS_CAPA;
                                 reply->ocd_connect_flags &= ~OBD_CONNECT_MDS_CAPA;
                         if (!mdt->mdt_opts.mo_oss_capa)
                                 reply->ocd_connect_flags &= ~OBD_CONNECT_OSS_CAPA;

-
-                        cfs_spin_lock(&exp->exp_lock);
-                        exp->exp_connect_flags = reply->ocd_connect_flags;
-                        cfs_spin_unlock(&exp->exp_lock);

                 }
                 break;
         default:
                 }
                 break;
         default: