protocol/server: do not do root-squashing for trusted clients

* As of now clients mounting within the storage pool using that machine's
  ip/hostname are trusted clients (i.e clients local to the glusterd).

* Be careful when the request itself comes in as nfsnobody (ex: posix tests).
  So move the squashing part to protocol/server when it creates a new frame
  for the request, instead of auth part of rpc layer.

* For nfs servers do root-squashing without checking if it is trusted client,
  as all the nfs servers would be running within the storage pool, hence will
  be trusted clients for the bricks.

* Provide one more option for mounting which actually says root-squash
  should/should not happen. This value is given priority only for the trusted
  clients. For non trusted clients, the volume option takes the priority. But
  for trusted clients if root-squash should not happen, then they have to be
  mounted with root-squash=no option. (This is done because by default
  blocking root-squashing for the trusted clients will cause problems for smb
  and UFO clients for which the requests have to be squashed if the option is
  enabled).

* For geo-replication and defrag clients do not do root-squashing.

* Introduce a new option in open-behind for doing read after successful open.

Change-Id: I8a8359840313dffc34824f3ea80a9c48375067f0
BUG: 954057
Signed-off-by: Raghavendra Bhat <raghavendra@redhat.com>
Reviewed-on: http://review.gluster.org/4863
Tested-by: Gluster Build System <jenkins@build.gluster.com>
Reviewed-by: Vijay Bellur <vbellur@redhat.com>

17 files changed, 406 insertions, 9 deletions

diff --git a/glusterfsd/src/glusterfsd.c b/glusterfsd/src/glusterfsd.c
index 098a9169aa4..c47d2ca3fc2 100644
--- a/glusterfsd/src/glusterfsd.c
+++ b/glusterfsd/src/glusterfsd.c
@@ -192,6 +192,9 @@ static struct argp_option gf_options[] = {
 	 "[default: 48]"},
         {"client-pid", ARGP_CLIENT_PID_KEY, "PID", OPTION_HIDDEN,
          "client will authenticate itself with process id PID to server"},
+        {"no-root-squash", ARGP_FUSE_NO_ROOT_SQUASH_KEY, "BOOL",
+         OPTION_ARG_OPTIONAL, "disable/enable root squashing for the trusted "
+         "client"},
         {"user-map-root", ARGP_USER_MAP_ROOT_KEY, "USER", OPTION_HIDDEN,
          "replace USER with root in messages"},
         {"dump-fuse", ARGP_DUMP_FUSE_KEY, "PATH", 0,
@@ -467,6 +470,32 @@ set_fuse_mount_options (glusterfs_ctx_t *ctx, dict_t *options)
                 break;
         }
 
+        switch (cmd_args->no_root_squash) {
+        case GF_OPTION_ENABLE: /* enable */
+                ret = dict_set_static_ptr (options, "no-root-squash",
+                                           "enable");
+                if (ret < 0) {
+                        gf_log ("glusterfsd", GF_LOG_ERROR,
+                                "failed to set 'enable' for key "
+                                "no-root-squash");
+                        goto err;
+                }
+                break;
+        case GF_OPTION_DISABLE: /* disable/default */
+        default:
+                ret = dict_set_static_ptr (options, "no-root-squash",
+                                           "disable");
+                if (ret < 0) {
+                        gf_log ("glusterfsd", GF_LOG_ERROR,
+                                "failed to set 'disable' for key "
+                                "no-root-squash");
+                        goto err;
+                }
+                gf_log ("", GF_LOG_DEBUG, "fuse no-root-squash mode %d",
+                        cmd_args->no_root_squash);
+                break;
+        }
+
         if (!cmd_args->no_daemon_mode) {
                 ret = dict_set_static_ptr (options, "sync-to-mount",
                                            "enable");
@@ -900,6 +929,10 @@ parse_opts (int key, char *arg, struct argp_state *state)
                               "unknown direct I/O mode setting \"%s\"", arg);
                 break;
 
+        case ARGP_FUSE_NO_ROOT_SQUASH_KEY:
+                cmd_args->no_root_squash = _gf_true;
+                break;
+
         case ARGP_ENTRY_TIMEOUT_KEY:
                 d = 0.0;
 
diff --git a/glusterfsd/src/glusterfsd.h b/glusterfsd/src/glusterfsd.h
index 9e2a0e56e6f..ad4c3699b56 100644
--- a/glusterfsd/src/glusterfsd.h
+++ b/glusterfsd/src/glusterfsd.h
@@ -84,6 +84,7 @@ enum argp_option_keys {
 	ARGP_FUSE_MOUNTOPTS_KEY		  = 164,
         ARGP_FUSE_USE_READDIRP_KEY        = 165,
 	ARGP_AUX_GFID_MOUNT_KEY		  = 166,
+        ARGP_FUSE_NO_ROOT_SQUASH_KEY      = 167,
 };
 
 struct _gfd_vol_top_priv_t {
diff --git a/libglusterfs/src/client_t.h b/libglusterfs/src/client_t.h
index 548081896c0..4113b9da96c 100644
--- a/libglusterfs/src/client_t.h
+++ b/libglusterfs/src/client_t.h
@@ -44,6 +44,8 @@ typedef struct _client_t {
                 int                  flavour;
                 size_t               len;
                 char                *data;
+                char                *username;
+                char                *passwd;
         }            auth;
 } client_t;
 
diff --git a/libglusterfs/src/common-utils.h b/libglusterfs/src/common-utils.h
index 6f8436fcba0..e3b019b9e85 100644
--- a/libglusterfs/src/common-utils.h
+++ b/libglusterfs/src/common-utils.h
@@ -106,6 +106,7 @@ enum _gf_client_pid
         GF_CLIENT_PID_GSYNCD = -1,
         GF_CLIENT_PID_HADOOP = -2,
         GF_CLIENT_PID_DEFRAG = -3,
+        GF_CLIENT_PID_NO_ROOT_SQUASH = -4,
 };
 
 typedef enum _gf_boolean gf_boolean_t;
diff --git a/libglusterfs/src/glusterfs.h b/libglusterfs/src/glusterfs.h
index 09b26ecf331..31c46b74efc 100644
--- a/libglusterfs/src/glusterfs.h
+++ b/libglusterfs/src/glusterfs.h
@@ -364,9 +364,10 @@ struct _cmd_args {
         int              aux_gfid_mount;
         struct list_head xlator_options;  /* list of xlator_option_t */
 
-        /* fuse options */
-        int              fuse_direct_io_mode;
-        char             *use_readdirp;
+	/* fuse options */
+	int              fuse_direct_io_mode;
+	char             *use_readdirp;
+        int              no_root_squash;
         int              volfile_check;
         double           fuse_entry_timeout;
         double           fuse_negative_timeout;
diff --git a/rpc/rpc-lib/src/auth-glusterfs.c b/rpc/rpc-lib/src/auth-glusterfs.c
index 48871ffb350..7bafa82fb82 100644
--- a/rpc/rpc-lib/src/auth-glusterfs.c
+++ b/rpc/rpc-lib/src/auth-glusterfs.c
@@ -109,7 +109,6 @@ int auth_glusterfs_authenticate (rpcsvc_request_t *req, void *priv)
         for (gidcount = 0; gidcount < au.ngrps; ++gidcount)
                 req->auxgids[gidcount] = au.groups[gidcount];
 
-        RPC_AUTH_ROOT_SQUASH(req);
 
         gf_log (GF_RPCSVC, GF_LOG_TRACE, "Auth Info: pid: %u, uid: %d"
                 ", gid: %d, owner: %s",
@@ -229,7 +228,6 @@ int auth_glusterfs_v2_authenticate (rpcsvc_request_t *req, void *priv)
         for (i = 0; i < au.lk_owner.lk_owner_len; ++i)
                 req->lk_owner.data[i] = au.lk_owner.lk_owner_val[i];
 
-        RPC_AUTH_ROOT_SQUASH(req);
 
         gf_log (GF_RPCSVC, GF_LOG_TRACE, "Auth Info: pid: %u, uid: %d"
                 ", gid: %d, owner: %s",
diff --git a/rpc/rpc-lib/src/rpcsvc.h b/rpc/rpc-lib/src/rpcsvc.h
index 5a6f930d359..5a991d4993f 100644
--- a/rpc/rpc-lib/src/rpcsvc.h
+++ b/rpc/rpc-lib/src/rpcsvc.h
@@ -286,6 +286,7 @@ struct rpcsvc_request {
                                 req->uid = req->svc->anonuid;           \
                         if (req->gid == RPC_ROOT_GID)                   \
                                 req->gid = req->svc->anongid;           \
+                                                                        \
                         for (gidcount = 0; gidcount < req->auxgidcount; \
                              ++gidcount) {                              \
                                 if (!req->auxgids[gidcount])            \
diff --git a/tests/bugs/bug-954057.t b/tests/bugs/bug-954057.t
new file mode 100755
index 00000000000..30bc1d77e6c
--- /dev/null
+++ b/tests/bugs/bug-954057.t
@@ -0,0 +1,44 @@
+#!/bin/bash
+
+. $(dirname $0)/../include.rc
+. $(dirname $0)/../volume.rc
+
+#This script checks if use-readdirp option works as accepted in mount options
+
+
+TEST glusterd
+TEST pidof glusterd
+TEST $CLI volume create $V0 $H0:$B0/${V0}
+TEST $CLI volume start $V0
+
+TEST glusterfs --volfile-id=/$V0 --volfile-server=$H0 $M0
+
+TEST mkdir $M0/dir
+TEST mkdir $M0/nobody
+TEST chown nfsnobody:nfsnobody $M0/nobody
+TEST `echo "file" >> $M0/file`
+TEST cp $M0/file $M0/new
+TEST chmod 700 $M0/new
+TEST cat $M0/new
+
+TEST $CLI volume set $V0 server.root-squash enable
+TEST `echo 3 > /proc/sys/vm/drop_caches`
+TEST ! mkdir $M0/other
+TEST mkdir $M0/nobody/other
+TEST cat $M0/file
+TEST ! cat $M0/new
+TEST `echo "nobody" >> $M0/nobody/file`
+
+#mount the client without root-squashing
+TEST glusterfs --volfile-id=/$V0 --volfile-server=$H0 --no-root-squash=yes $M1
+TEST mkdir $M1/m1_dir
+TEST `echo "file" >> $M1/m1_file`
+TEST cp $M0/file $M1/new
+TEST chmod 700 $M1/new
+TEST cat $M1/new
+
+TEST $CLI volume set $V0 server.root-squash disable
+TEST mkdir $M0/other
+TEST cat $M0/new
+
+cleanup
diff --git a/xlators/mgmt/glusterd/src/glusterd-volgen.c b/xlators/mgmt/glusterd/src/glusterd-volgen.c
index b26a7ef3d95..3bb2f9a6334 100644
--- a/xlators/mgmt/glusterd/src/glusterd-volgen.c
+++ b/xlators/mgmt/glusterd/src/glusterd-volgen.c
@@ -2517,6 +2517,9 @@ client_graph_builder (volgen_graph_t *graph, glusterd_volinfo_t *volinfo,
         xlator_t        *xl      = NULL;
         char            *volname = NULL;
         glusterd_conf_t *conf    = THIS->private;
+        char            *tmp     = NULL;
+        gf_boolean_t     var     = _gf_false;
+        gf_boolean_t     ob      = _gf_false;
 
         GF_ASSERT (conf);
 
@@ -2582,7 +2585,93 @@ client_graph_builder (volgen_graph_t *graph, glusterd_volinfo_t *volinfo,
                 }
         }
 
-	ret = client_graph_set_perf_options(graph, volinfo, set_dict);
+        /* Do not allow changing read-after-open option if root-squash is
+           enabled.
+        */
+        ret = dict_get_str (set_dict, "performance.read-after-open", &tmp);
+        if (!ret) {
+                ret = dict_get_str (volinfo->dict, "server.root-squash", &tmp);
+                if (!ret) {
+                        ob = _gf_false;
+                        ret = gf_string2boolean (tmp, &ob);
+                        if (!ret && ob) {
+                                gf_log (THIS->name, GF_LOG_WARNING,
+                                        "root-squash is enabled. Please turn it"
+                                        " off to change read-after-open "
+                                        "option");
+                                ret = -1;
+                                goto out;
+                        }
+                }
+        }
+
+        /* open behind causes problems when root-squash is enabled
+           (by allowing reads to happen even though the squashed user
+           does not have permissions to do so) as it fakes open to be
+           successful and later sends reads on anonymous fds. So when
+           root-squash is enabled, open-behind's option to read after
+           open is done is also enabled.
+        */
+        ret = dict_get_str (set_dict, "server.root-squash", &tmp);
+        if (!ret) {
+                ret = gf_string2boolean (tmp, &var);
+                if (ret)
+                        goto out;
+
+                if (var) {
+                        ret = dict_get_str (volinfo->dict,
+                                            "performance.read-after-open",
+                                            &tmp);
+                        if (!ret) {
+                                ret = gf_string2boolean (tmp, &ob);
+                                /* go ahead with turning read-after-open on
+                                   even if string2boolean conversion fails,
+                                   OR if read-after-open option is turned off
+                                */
+                                if (ret || !ob)
+                                        ret = dict_set_str (set_dict,
+                                                 "performance.read-after-open",
+                                                            "yes");
+                        } else {
+                                ret = dict_set_str (set_dict,
+                                                    "performance.read-after-open",
+                                                    "yes");
+                        }
+                } else {
+                        /* When root-squash has to be turned off, open-behind's
+                           read-after-open option should be reset to what was
+                           there before root-squash was turned on. If the option
+                           cannot be found in volinfo's dict, it means that
+                           option was not set before turning on root-squash.
+                        */
+                        ob = _gf_false;
+                        ret = dict_get_str (volinfo->dict,
+                                            "performance.read-after-open",
+                                            &tmp);
+                        if (!ret) {
+                                ret = gf_string2boolean (tmp, &ob);
+
+                                if (!ret && ob) {
+                                        ret = dict_set_str (set_dict,
+                                                 "performance.read-after-open",
+                                                            "yes");
+                                }
+                        }
+                        /* consider operation is failure only if read-after-open
+                           option is enabled and could not set into set_dict
+                        */
+                        if (!ob)
+                                ret = 0;
+                }
+                if (ret) {
+                        gf_log (THIS->name, GF_LOG_WARNING, "setting "
+                                "open behind option as part of root "
+                                "squash failed");
+                        goto out;
+                }
+        }
+
+        ret = client_graph_set_perf_options(graph, volinfo, set_dict);
         if (ret)
                 goto out;
 
diff --git a/xlators/mgmt/glusterd/src/glusterd-volume-set.c b/xlators/mgmt/glusterd/src/glusterd-volume-set.c
index db0a6374a30..ab8cefeddad 100644
--- a/xlators/mgmt/glusterd/src/glusterd-volume-set.c
+++ b/xlators/mgmt/glusterd/src/glusterd-volume-set.c
@@ -711,6 +711,12 @@ struct volopt_map_entry glusterd_volopt_map[] = {
           .op_version = 3,
           .flags      = OPT_FLAG_CLIENT_OPT
         },
+        { .key        = "performance.read-after-open",
+          .voltype    = "performance/open-behind",
+          .option     = "read-after-open",
+          .op_version = 3,
+          .flags      = OPT_FLAG_CLIENT_OPT
+        },
         { .key        = "performance.read-ahead-page-count",
           .voltype    = "performance/read-ahead",
           .option     = "page-count",
diff --git a/xlators/mount/fuse/src/fuse-bridge.c b/xlators/mount/fuse/src/fuse-bridge.c
index 315259ece7b..d9055468e43 100644
--- a/xlators/mount/fuse/src/fuse-bridge.c
+++ b/xlators/mount/fuse/src/fuse-bridge.c
@@ -5323,6 +5323,18 @@ init (xlator_t *this_xl)
         GF_OPTION_INIT ("congestion-threshold", priv->congestion_threshold,
                         int32, cleanup_exit);
 
+        GF_OPTION_INIT("no-root-squash", priv->no_root_squash, bool,
+                       cleanup_exit);
+        /* change the client_pid to no-root-squash pid only if the
+           client is none of defrag process, hadoop access and gsyncd process.
+        */
+        if (!priv->client_pid_set) {
+                if (priv->no_root_squash == _gf_true) {
+                        priv->client_pid_set = _gf_true;
+                        priv->client_pid = GF_CLIENT_PID_NO_ROOT_SQUASH;
+                }
+        }
+
         /* user has set only background-qlen, not congestion-threshold,
            use the fuse kernel driver formula to set congestion. ie, 75% */
         if (dict_get (this_xl->options, "background-qlen") &&
@@ -5563,5 +5575,15 @@ struct volume_options options[] = {
           .type = GF_OPTION_TYPE_BOOL,
           .default_value = "yes"
         },
+        { .key = {"no-root-squash"},
+          .type = GF_OPTION_TYPE_BOOL,
+          .default_value = "false",
+          .description = "This is the mount option for disabling the "
+          "root squash for the client irrespective of whether the root-squash "
+          "option for the volume is set or not. But this option is honoured "
+          "only for the trusted clients. For non trusted clients this value "
+          "does not have any affect and the volume option for root-squash is "
+          "honoured.",
+        },
         { .key = {NULL} },
 };
diff --git a/xlators/mount/fuse/src/fuse-bridge.h b/xlators/mount/fuse/src/fuse-bridge.h
index 34794b6ea45..f1c4cb3f0d8 100644
--- a/xlators/mount/fuse/src/fuse-bridge.h
+++ b/xlators/mount/fuse/src/fuse-bridge.h
@@ -104,6 +104,14 @@ struct fuse_private {
 	int32_t	             fopen_keep_cache;
 	int32_t		     gid_cache_timeout;
         gf_boolean_t         enable_ino32;
+        /* This is the mount option for disabling the root-squash for the
+           mount irrespective of whether the root-squash option for the
+           volume is set or not. But this option is honoured only for
+           thr trusted clients. For non trusted clients this value does
+           not have any affect and the volume option for root-squash is
+           honoured.
+        */
+        gf_boolean_t        no_root_squash;
         fdtable_t           *fdtable;
 	gid_cache_t	     gid_cache;
         char                *fuse_mountopts;
diff --git a/xlators/mount/fuse/utils/mount.glusterfs.in b/xlators/mount/fuse/utils/mount.glusterfs.in
index ff6b524605a..d22f6a69b1e 100755
--- a/xlators/mount/fuse/utils/mount.glusterfs.in
+++ b/xlators/mount/fuse/utils/mount.glusterfs.in
@@ -171,7 +171,11 @@ start_glusterfs ()
         cmd_line=$(echo "$cmd_line --aux-gfid-mount");
     fi
 
-    # options with values start here
+    if [ -n "$no_root_squash" ]; then
+        cmd_line=$(echo "$cmd_line --no-root-squash");
+    fi
+
+#options with values start here
     if [ -n "$log_level" ]; then
         cmd_line=$(echo "$cmd_line --log-level=$log_level");
     fi
@@ -442,6 +446,13 @@ with_options()
         "use-readdirp")
             use_readdirp=$value
             ;;
+        "root-squash")
+            if [ $value == "no" ] ||
+                [ $value == "off" ] ||
+                [ $value == "disable" ] ||
+                [ $value == "false" ] ; then
+                no_root_squash=1;
+            fi ;;
         *)
             echo "Invalid option: $key"
             exit 0
diff --git a/xlators/performance/open-behind/src/open-behind.c b/xlators/performance/open-behind/src/open-behind.c
index 29ef64364dc..742e4df3fdf 100644
--- a/xlators/performance/open-behind/src/open-behind.c
+++ b/xlators/performance/open-behind/src/open-behind.c
@@ -23,6 +23,11 @@ typedef struct ob_conf {
 					   like mandatory locks
 					*/
 	gf_boolean_t  lazy_open; /* delay backend open as much as possible */
+        gf_boolean_t  read_after_open; /* instead of sending readvs on
+                                               anonymous fds, open the file
+                                               first and then send readv i.e
+                                               similar to what writev does
+                                            */
 } ob_conf_t;
 
 
@@ -367,8 +372,14 @@ ob_readv (call_frame_t *frame, xlator_t *this, fd_t *fd, size_t size,
 {
 	call_stub_t  *stub = NULL;
 	fd_t         *wind_fd = NULL;
+        ob_conf_t    *conf = NULL;
 
-	wind_fd = ob_get_wind_fd (this, fd);
+        conf = this->private;
+
+        if (!conf->read_after_open)
+                wind_fd = ob_get_wind_fd (this, fd);
+        else
+                wind_fd = fd_ref (fd);
 
 	stub = fop_readv_stub (frame, default_readv_resume, wind_fd,
 			       size, offset, flags, xdata);
@@ -894,6 +905,8 @@ reconfigure (xlator_t *this, dict_t *options)
 			  bool, out);
 
         GF_OPTION_RECONF ("lazy-open", conf->lazy_open, options, bool, out);
+        GF_OPTION_RECONF ("read-after-open", conf->read_after_open, options,
+                          bool, out);
 
         ret = 0;
 out:
@@ -924,7 +937,7 @@ init (xlator_t *this)
         GF_OPTION_INIT ("use-anonymous-fd", conf->use_anonymous_fd, bool, err);
 
         GF_OPTION_INIT ("lazy-open", conf->lazy_open, bool, err);
-
+        GF_OPTION_INIT ("read-after-open", conf->read_after_open, bool, err);
         this->private = conf;
 
 	return 0;
@@ -996,6 +1009,12 @@ struct volume_options options[] = {
           "FOP arrives (e.g writev on the FD, unlink of the file). When option "
           "is disabled, perform backend open right after unwinding open().",
         },
+        { .key  = {"read-after-open"},
+          .type = GF_OPTION_TYPE_BOOL,
+          .default_value = "no",
+          .description = "read is sent only after actual open happens and real "
+          "fd is obtained, instead of doing on anonymous fd (similar to write)",
+        },
         { .key  = {NULL} }
 
 };
diff --git a/xlators/protocol/server/src/server-handshake.c b/xlators/protocol/server/src/server-handshake.c
index d4941011da9..708acd936d9 100644
--- a/xlators/protocol/server/src/server-handshake.c
+++ b/xlators/protocol/server/src/server-handshake.c
@@ -448,6 +448,8 @@ server_setvolume (rpcsvc_request_t *req)
         if (req->trans->xl_private != client)
                 req->trans->xl_private = client;
 
+        auth_set_username_passwd (params, config_params, client);
+
         ret = dict_get_int32 (params, "fops-version", &fop_version);
         if (ret < 0) {
                 ret = dict_set_str (reply, "ERROR",
diff --git a/xlators/protocol/server/src/server-helpers.c b/xlators/protocol/server/src/server-helpers.c
index b2b6c486fe1..c11011abf9f 100644
--- a/xlators/protocol/server/src/server-helpers.c
+++ b/xlators/protocol/server/src/server-helpers.c
@@ -302,6 +302,11 @@ get_frame_from_request (rpcsvc_request_t *req)
 {
         call_frame_t  *frame = NULL;
         client_t      *client = NULL;
+        client_t      *tmp_client = NULL;
+        xlator_t  *this = NULL;
+        server_conf_t *priv = NULL;
+        clienttable_t *clienttable = NULL;
+        unsigned int   i           = 0;
 
         GF_VALIDATE_OR_GOTO ("server", req, out);
 
@@ -315,6 +320,57 @@ get_frame_from_request (rpcsvc_request_t *req)
 
         frame->root->unique   = req->xid;
 
+        client = req->trans->xl_private;
+        this = req->trans->xl;
+        priv = this->private;
+        clienttable = this->ctx->clienttable;
+
+        for (i = 0; i < clienttable->max_clients; i++) {
+                tmp_client = clienttable->cliententries[i].client;
+                if (client == tmp_client) {
+                        /* for non trusted clients username and password
+                           would not have been set. So for non trusted clients
+                           (i.e clients not from the same machine as the brick,
+                           and clients from outside the storage pool)
+                           do the root-squashing.
+                           TODO: If any client within the storage pool (i.e
+                           mounting within a machine from the pool but using
+                           other machine's ip/hostname from the same pool)
+                           is present treat it as a trusted client
+                        */
+                        if (!client->auth.username && req->pid != NFS_PID)
+                                RPC_AUTH_ROOT_SQUASH (req);
+
+                        /* Problem: If we just check whether the client is
+                           trusted client and do not do root squashing for
+                           them, then for smb clients and UFO clients root
+                           squashing will never happen as they use the fuse
+                           mounts done within the trusted pool (i.e they are
+                           trusted clients).
+                           Solution: To fix it, do root squashing for trusted
+                           clients also. If one wants to have a client within
+                           the storage pool for which root-squashing does not
+                           happen, then the client has to be mounted with
+                           --no-root-squash option. But for defrag client and
+                           gsyncd client do not do root-squashing.
+                        */
+                        if (client->auth.username &&
+                            req->pid != GF_CLIENT_PID_NO_ROOT_SQUASH &&
+                            req->pid != GF_CLIENT_PID_GSYNCD &&
+                            req->pid != GF_CLIENT_PID_DEFRAG)
+                                RPC_AUTH_ROOT_SQUASH (req);
+
+                        /* For nfs clients the server processes will be running
+                           within the trusted storage pool machines. So if we
+                           do not do root-squashing for nfs servers, thinking
+                           that its a trusted client, then root-squashing wont
+                           work for nfs clients.
+                        */
+                        if (req->pid == NFS_PID)
+                                RPC_AUTH_ROOT_SQUASH (req);
+                }
+        }
+
         frame->root->uid      = req->uid;
         frame->root->gid      = req->gid;
         frame->root->pid      = req->pid;
@@ -935,3 +991,104 @@ server_ctx_get (client_t *client, xlator_t *xlator)
 out:
         return ctx;
 }
+
+int
+auth_set_username_passwd (dict_t *input_params, dict_t *config_params,
+                          client_t *client)
+{
+        int      ret           = 0;
+        data_t  *allow_user    = NULL;
+        data_t  *passwd_data   = NULL;
+        char    *username      = NULL;
+        char    *password      = NULL;
+        char    *brick_name    = NULL;
+        char    *searchstr     = NULL;
+        char    *username_str  = NULL;
+        char    *tmp           = NULL;
+        char    *username_cpy  = NULL;
+
+        ret = dict_get_str (input_params, "username", &username);
+        if (ret) {
+                gf_log ("auth/login", GF_LOG_DEBUG,
+                        "username not found, returning DONT-CARE");
+                /* For non trusted clients username and password
+                   will not be there. So dont reject the client.
+                */
+                ret = 0;
+                goto out;
+        }
+
+        ret = dict_get_str (input_params, "password", &password);
+        if (ret) {
+                gf_log ("auth/login", GF_LOG_WARNING,
+                        "password not found, returning DONT-CARE");
+                goto out;
+        }
+
+        ret = dict_get_str (input_params, "remote-subvolume", &brick_name);
+        if (ret) {
+                gf_log ("auth/login", GF_LOG_ERROR,
+                        "remote-subvolume not specified");
+                ret = -1;
+                goto out;
+        }
+
+        ret = gf_asprintf (&searchstr, "auth.login.%s.allow", brick_name);
+        if (-1 == ret) {
+                ret = 0;
+                goto out;
+        }
+
+        allow_user = dict_get (config_params, searchstr);
+        GF_FREE (searchstr);
+
+        if (allow_user) {
+                username_cpy = gf_strdup (allow_user->data);
+                if (!username_cpy)
+                        goto out;
+
+                username_str = strtok_r (username_cpy, " ,", &tmp);
+
+                while (username_str) {
+                        if (!fnmatch (username_str, username, 0)) {
+                                ret = gf_asprintf (&searchstr,
+                                                   "auth.login.%s.password",
+                                                   username);
+                                if (-1 == ret)
+                                        goto out;
+
+                                passwd_data = dict_get (config_params,
+                                                        searchstr);
+                                GF_FREE (searchstr);
+
+                                if (!passwd_data) {
+                                        gf_log ("auth/login", GF_LOG_ERROR,
+                                                "wrong username/password "
+                                                "combination");
+                                        ret = -1;
+                                        goto out;
+                                }
+
+                                ret = !((strcmp (data_to_str (passwd_data),
+                                                    password))?0: -1);
+                                if (!ret) {
+                                        client->auth.username =
+                                                gf_strdup (username);
+                                        client->auth.passwd =
+                                                gf_strdup (password);
+                                }
+                                if (ret == -1)
+                                        gf_log ("auth/login", GF_LOG_ERROR,
+                                                "wrong password for user %s",
+                                                username);
+                                break;
+                        }
+                        username_str = strtok_r (NULL, " ,", &tmp);
+                }
+        }
+
+out:
+        GF_FREE (username_cpy);
+
+        return ret;
+}
diff --git a/xlators/protocol/server/src/server-helpers.h b/xlators/protocol/server/src/server-helpers.h
index 93ea3585102..3c257b3bcef 100644
--- a/xlators/protocol/server/src/server-helpers.h
+++ b/xlators/protocol/server/src/server-helpers.h
@@ -53,6 +53,8 @@ int serialize_rsp_dirent (gf_dirent_t *entries, gfs3_readdir_rsp *rsp);
 int serialize_rsp_direntp (gf_dirent_t *entries, gfs3_readdirp_rsp *rsp);
 int readdirp_rsp_cleanup (gfs3_readdirp_rsp *rsp);
 int readdir_rsp_cleanup (gfs3_readdir_rsp *rsp);
+int auth_set_username_passwd (dict_t *input_params, dict_t *config_params,
+                              struct _client_t *client);
 
 server_ctx_t *server_ctx_get (client_t *client, xlator_t *xlator);