rpc/*: auth-header changes

Introduce another authentication header which can now send more data.
This is useful because this data can be common for all the fops, and
we don't need to change all the signatures.

As part of this, made rpc-clnt.c little more modular to support multiple
authentication structures.

stack.h changes are placeholder for the ctime etc, can be moved later
based on need.

updates #384

Change-Id: I6111c13cfd2ec92e2b4e9295896bf62a8a33b2c7
Signed-off-by: Amar Tumballi <amarts@redhat.com>

9 files changed, 388 insertions, 119 deletions

diff --git a/rpc/rpc-lib/src/auth-glusterfs.c b/rpc/rpc-lib/src/auth-glusterfs.c
index 5670b8e840..78f283557b 100644
--- a/rpc/rpc-lib/src/auth-glusterfs.c
+++ b/rpc/rpc-lib/src/auth-glusterfs.c
@@ -16,6 +16,7 @@
 #include "xdr-rpc.h"
 #include "xdr-common.h"
 #include "rpc-common-xdr.h"
+#include "glusterfs4-xdr.h"
 
 /* V1 */
 
@@ -189,8 +190,10 @@ int auth_glusterfs_v2_authenticate (rpcsvc_request_t *req, void *priv)
         req->auxgidcount = au.groups.groups_len;
 
         /* the number of groups and size of lk_owner depend on each other */
-        max_groups = GF_AUTH_GLUSTERFS_MAX_GROUPS (req->lk_owner.len);
-        max_lk_owner_len = GF_AUTH_GLUSTERFS_MAX_LKOWNER (req->auxgidcount);
+        max_groups = GF_AUTH_GLUSTERFS_MAX_GROUPS (req->lk_owner.len,
+                                                   AUTH_GLUSTERFS_v2);
+        max_lk_owner_len = GF_AUTH_GLUSTERFS_MAX_LKOWNER (req->auxgidcount,
+                                                          AUTH_GLUSTERFS_v2);
 
         if (req->auxgidcount > max_groups) {
                 gf_log ("", GF_LOG_WARNING,
@@ -263,3 +266,141 @@ rpcsvc_auth_glusterfs_v2_init (rpcsvc_t *svc, dict_t *options)
 {
         return &rpcsvc_auth_glusterfs_v2;
 }
+
+/* V3 */
+
+ssize_t
+xdr_to_glusterfs_auth_v3 (char *buf, struct auth_glusterfs_params_v3 *req)
+{
+        XDR     xdr;
+        ssize_t ret = -1;
+
+        if ((!buf) || (!req))
+                return -1;
+
+        xdrmem_create (&xdr, buf, GF_MAX_AUTH_BYTES, XDR_DECODE);
+        if (!xdr_auth_glusterfs_params_v3 (&xdr, req)) {
+                gf_log ("", GF_LOG_WARNING,
+                        "failed to decode glusterfs v3 parameters");
+                ret  = -1;
+                goto ret;
+        }
+
+        ret = (((size_t)(&xdr)->x_private) - ((size_t)(&xdr)->x_base));
+ret:
+        return ret;
+}
+
+int
+auth_glusterfs_v3_request_init (rpcsvc_request_t *req, void *priv)
+{
+        return 0;
+}
+
+int auth_glusterfs_v3_authenticate (rpcsvc_request_t *req, void *priv)
+{
+        struct auth_glusterfs_params_v3  au = {0,};
+        int ret                            = RPCSVC_AUTH_REJECT;
+        int i                              = 0;
+        int max_groups                     = 0;
+        int max_lk_owner_len               = 0;
+
+        if (!req)
+                return ret;
+
+        ret = xdr_to_glusterfs_auth_v3 (req->cred.authdata, &au);
+        if (ret == -1) {
+                gf_log ("", GF_LOG_WARNING,
+                        "failed to decode glusterfs credentials");
+                ret = RPCSVC_AUTH_REJECT;
+                goto err;
+        }
+
+        req->pid = au.pid;
+        req->uid = au.uid;
+        req->gid = au.gid;
+        req->lk_owner.len = au.lk_owner.lk_owner_len;
+        req->auxgidcount = au.groups.groups_len;
+
+        /* the number of groups and size of lk_owner depend on each other */
+        max_groups = GF_AUTH_GLUSTERFS_MAX_GROUPS (req->lk_owner.len,
+                                                   AUTH_GLUSTERFS_v3);
+        max_lk_owner_len = GF_AUTH_GLUSTERFS_MAX_LKOWNER (req->auxgidcount,
+                                                          AUTH_GLUSTERFS_v3);
+
+        if (req->auxgidcount > max_groups) {
+                gf_log ("", GF_LOG_WARNING,
+                        "more than max aux gids found (%d) , truncating it "
+                        "to %d and continuing", au.groups.groups_len,
+                        max_groups);
+                req->auxgidcount = max_groups;
+        }
+
+        if (req->lk_owner.len > max_lk_owner_len) {
+                gf_log ("", GF_LOG_WARNING,
+                        "lkowner field to big (%d), depends on the number of "
+                        "groups (%d), failing authentication",
+                        req->lk_owner.len, req->auxgidcount);
+                ret = RPCSVC_AUTH_REJECT;
+                goto err;
+        }
+
+	if (req->auxgidcount > SMALL_GROUP_COUNT) {
+		req->auxgidlarge = GF_CALLOC(req->auxgidcount,
+					     sizeof(req->auxgids[0]),
+					     gf_common_mt_auxgids);
+		req->auxgids = req->auxgidlarge;
+	} else {
+		req->auxgids = req->auxgidsmall;
+	}
+
+	if (!req->auxgids) {
+		gf_log ("auth-glusterfs-v2", GF_LOG_WARNING,
+			"cannot allocate gid list");
+		ret = RPCSVC_AUTH_REJECT;
+		goto err;
+	}
+
+        for (i = 0; i < req->auxgidcount; ++i)
+                req->auxgids[i] = au.groups.groups_val[i];
+
+        for (i = 0; i < au.lk_owner.lk_owner_len; ++i)
+                req->lk_owner.data[i] = au.lk_owner.lk_owner_val[i];
+
+        /* All new things, starting glusterfs-4.0.0 */
+        req->flags = au.flags;
+        req->ctime.tv_sec = au.ctime_sec;
+        req->ctime.tv_nsec = au.ctime_nsec;
+
+        gf_log (GF_RPCSVC, GF_LOG_TRACE, "Auth Info: pid: %u, uid: %d"
+                ", gid: %d, owner: %s, flags: %d",
+                req->pid, req->uid, req->gid, lkowner_utoa (&req->lk_owner),
+                req->flags);
+        ret = RPCSVC_AUTH_ACCEPT;
+err:
+        /* TODO: instead use alloca() for these variables */
+        free (au.groups.groups_val);
+        free (au.lk_owner.lk_owner_val);
+
+        return ret;
+}
+
+rpcsvc_auth_ops_t auth_glusterfs_ops_v3 = {
+        .transport_init         = NULL,
+        .request_init           = auth_glusterfs_v3_request_init,
+        .authenticate           = auth_glusterfs_v3_authenticate
+};
+
+rpcsvc_auth_t rpcsvc_auth_glusterfs_v3 = {
+        .authname       = "AUTH_GLUSTERFS-v3",
+        .authnum        = AUTH_GLUSTERFS_v3,
+        .authops        = &auth_glusterfs_ops_v3,
+        .authprivate    = NULL
+};
+
+
+rpcsvc_auth_t *
+rpcsvc_auth_glusterfs_v3_init (rpcsvc_t *svc, dict_t *options)
+{
+        return &rpcsvc_auth_glusterfs_v3;
+}
diff --git a/rpc/rpc-lib/src/rpc-clnt.c b/rpc/rpc-lib/src/rpc-clnt.c
index 498f6b05f9..35125a2510 100644
--- a/rpc/rpc-lib/src/rpc-clnt.c
+++ b/rpc/rpc-lib/src/rpc-clnt.c
@@ -937,6 +937,10 @@ rpc_clnt_notify (rpc_transport_t *trans, void *mydata,
         case RPC_TRANSPORT_DISCONNECT:
         {
                 rpc_clnt_handle_disconnect (clnt, conn);
+                /* reset auth_type to use v2 (if its not auth-null), it
+                   would be set to appropriate type in handshake again */
+                if (clnt->auth_value)
+                        clnt->auth_value = AUTH_GLUSTERFS_v2;
                 break;
         }
 
@@ -1161,7 +1165,11 @@ rpc_clnt_new (dict_t *options, xlator_t *owner, char *name,
                 goto out;
         }
 
-        rpc->auth_null = dict_get_str_boolean (options, "auth-null", 0);
+        /* This is handled to make sure we have modularity in getting the
+           auth data changed */
+        gf_boolean_t auth_null = dict_get_str_boolean(options, "auth-null", 0);
+
+        rpc->auth_value = (auth_null) ? 0 : AUTH_GLUSTERFS_v2;
 
         rpc = rpc_clnt_ref (rpc);
         INIT_LIST_HEAD (&rpc->programs);
@@ -1236,51 +1244,177 @@ rpc_clnt_register_notify (struct rpc_clnt *rpc, rpc_clnt_notify_t fn,
 /* used for GF_LOG_OCCASIONALLY() */
 static int gf_auth_max_groups_log = 0;
 
-ssize_t
-xdr_serialize_glusterfs_auth (char *dest, struct auth_glusterfs_parms_v2 *au)
+static inline int
+setup_glusterfs_auth_param_v3 (call_frame_t *frame,
+                               auth_glusterfs_params_v3 *au,
+                               int lk_owner_len, char *owner_data)
+{
+        int ret = -1;
+        unsigned int max_groups = 0;
+        int max_lkowner_len = 0;
+
+        au->pid      = frame->root->pid;
+        au->uid      = frame->root->uid;
+        au->gid      = frame->root->gid;
+
+        au->flags = frame->root->flags;
+        au->ctime_sec = frame->root->ctime.tv_sec;
+        au->ctime_nsec = frame->root->ctime.tv_nsec;
+
+        au->lk_owner.lk_owner_val = owner_data;
+        au->lk_owner.lk_owner_len = lk_owner_len;
+        au->groups.groups_val = frame->root->groups;
+        au->groups.groups_len = frame->root->ngrps;
+
+        /* The number of groups and the size of lk_owner depend on oneother.
+         * We can truncate the groups, but should not touch the lk_owner. */
+        max_groups = GF_AUTH_GLUSTERFS_MAX_GROUPS (lk_owner_len, AUTH_GLUSTERFS_v3);
+        if (au->groups.groups_len > max_groups) {
+                GF_LOG_OCCASIONALLY (gf_auth_max_groups_log, "rpc-auth",
+                                     GF_LOG_WARNING, "truncating grouplist "
+                                     "from %d to %d", au->groups.groups_len,
+                                     max_groups);
+
+                au->groups.groups_len = max_groups;
+        }
+
+        max_lkowner_len = GF_AUTH_GLUSTERFS_MAX_LKOWNER (au->groups.groups_len,
+                                                         AUTH_GLUSTERFS_v3);
+        if (lk_owner_len > max_lkowner_len) {
+                gf_log ("rpc-clnt", GF_LOG_ERROR, "lkowner field is too "
+                        "big (%d), it does not fit in the rpc-header",
+                        au->lk_owner.lk_owner_len);
+                errno = E2BIG;
+                goto out;
+        }
+
+        ret = 0;
+out:
+        return ret;
+}
+
+static inline int
+setup_glusterfs_auth_param_v2 (call_frame_t *frame,
+                               auth_glusterfs_parms_v2 *au,
+                               int lk_owner_len, char *owner_data)
+{
+        unsigned int max_groups = 0;
+        int max_lkowner_len = 0;
+        int ret = -1;
+
+        au->pid      = frame->root->pid;
+        au->uid      = frame->root->uid;
+        au->gid      = frame->root->gid;
+
+        au->lk_owner.lk_owner_val = owner_data;
+        au->lk_owner.lk_owner_len = lk_owner_len;
+        au->groups.groups_val = frame->root->groups;
+        au->groups.groups_len = frame->root->ngrps;
+
+        /* The number of groups and the size of lk_owner depend on oneother.
+         * We can truncate the groups, but should not touch the lk_owner. */
+        max_groups = GF_AUTH_GLUSTERFS_MAX_GROUPS (lk_owner_len, AUTH_GLUSTERFS_v2);
+        if (au->groups.groups_len > max_groups) {
+                GF_LOG_OCCASIONALLY (gf_auth_max_groups_log, "rpc-auth",
+                                     GF_LOG_WARNING, "truncating grouplist "
+                                     "from %d to %d", au->groups.groups_len,
+                                     max_groups);
+
+                au->groups.groups_len = max_groups;
+        }
+
+        max_lkowner_len = GF_AUTH_GLUSTERFS_MAX_LKOWNER (au->groups.groups_len,
+                                                         AUTH_GLUSTERFS_v2);
+        if (lk_owner_len > max_lkowner_len) {
+                gf_log ("rpc-auth", GF_LOG_ERROR, "lkowner field is too "
+                        "big (%d), it does not fit in the rpc-header",
+                        au->lk_owner.lk_owner_len);
+                errno = E2BIG;
+                goto out;
+        }
+
+        ret = 0;
+out:
+        return ret;
+}
+
+
+static ssize_t
+xdr_serialize_glusterfs_auth (struct rpc_clnt *clnt, call_frame_t *frame,
+                              char *dest)
 {
         ssize_t ret = -1;
         XDR     xdr;
-        u_long  ngroups = 0;
-        int     max_groups = 0;
+        char    owner[4] = {0,};
+        int32_t pid = 0;
+        char   *lk_owner_data = NULL;
+        int     lk_owner_len = 0;
 
-        if ((!dest) || (!au))
+        if ((!dest))
                 return -1;
 
-        max_groups = GF_AUTH_GLUSTERFS_MAX_GROUPS (au->lk_owner.lk_owner_len);
-
         xdrmem_create (&xdr, dest, GF_MAX_AUTH_BYTES, XDR_ENCODE);
 
-        if (au->groups.groups_len > max_groups) {
-                ngroups = au->groups.groups_len;
-                au->groups.groups_len = max_groups;
+        if (frame->root->lk_owner.len) {
+                lk_owner_data = frame->root->lk_owner.data;
+                lk_owner_len = frame->root->lk_owner.len;
+        } else {
+                pid = frame->root->pid;
+                owner[0] = (char)(pid & 0xff);
+                owner[1] = (char)((pid >> 8) & 0xff);
+                owner[2] = (char)((pid >> 16) & 0xff);
+                owner[3] = (char)((pid >> 24) & 0xff);
 
-                GF_LOG_OCCASIONALLY (gf_auth_max_groups_log,
-                                     THIS->name, GF_LOG_WARNING,
-                                     "too many groups, reducing %ld -> %d",
-                                     ngroups, max_groups);
+                lk_owner_data = owner;
+                lk_owner_len = 4;
         }
 
-        if (!xdr_auth_glusterfs_parms_v2 (&xdr, au)) {
+        if (clnt->auth_value == AUTH_GLUSTERFS_v2) {
+                auth_glusterfs_parms_v2 au_v2 = {0,};
+
+                ret = setup_glusterfs_auth_param_v2 (frame, &au_v2,
+                                                     lk_owner_len,
+                                                     lk_owner_data);
+                if (ret)
+                        goto out;
+                if (!xdr_auth_glusterfs_parms_v2 (&xdr, &au_v2)) {
+                        gf_log (THIS->name, GF_LOG_WARNING,
+                                "failed to encode auth glusterfs elements");
+                        ret = -1;
+                        goto out;
+                }
+        } else if (clnt->auth_value == AUTH_GLUSTERFS_v3) {
+                auth_glusterfs_params_v3 au_v3 = {0,};
+
+                ret = setup_glusterfs_auth_param_v3 (frame, &au_v3,
+                                                     lk_owner_len,
+                                                     lk_owner_data);
+                if (ret)
+                        goto out;
+
+                if (!xdr_auth_glusterfs_params_v3 (&xdr, &au_v3)) {
+                        gf_log (THIS->name, GF_LOG_WARNING,
+                                "failed to encode auth glusterfs elements");
+                        ret = -1;
+                        goto out;
+                }
+        } else {
                 gf_log (THIS->name, GF_LOG_WARNING,
                         "failed to encode auth glusterfs elements");
                 ret = -1;
-                goto ret;
+                goto out;
         }
 
         ret = (((size_t)(&xdr)->x_private) - ((size_t)(&xdr)->x_base));
 
-ret:
-        if (ngroups)
-                au->groups.groups_len = ngroups;
-
+out:
         return ret;
 }
 
 
 int
-rpc_clnt_fill_request (int prognum, int progver, int procnum,
-                       uint64_t xid, struct auth_glusterfs_parms_v2 *au,
+rpc_clnt_fill_request (struct rpc_clnt *clnt, int prognum, int progver,
+                       int procnum, uint64_t xid, call_frame_t *fr,
                        struct rpc_msg *request, char *auth_data)
 {
         int   ret          = -1;
@@ -1299,25 +1433,21 @@ rpc_clnt_fill_request (int prognum, int progver, int procnum,
         request->rm_call.cb_vers = progver;
         request->rm_call.cb_proc = procnum;
 
-        /* TODO: Using AUTH_(GLUSTERFS/NULL) in a kludgy way for time-being.
-         * Make it modular in future so it is easy to plug-in new
-         * authentication schemes.
-         */
-        if (auth_data) {
-                ret = xdr_serialize_glusterfs_auth (auth_data, au);
+        if (!clnt->auth_value) {
+                request->rm_call.cb_cred.oa_flavor = AUTH_NULL;
+                request->rm_call.cb_cred.oa_base   = NULL;
+                request->rm_call.cb_cred.oa_length = 0;
+        } else {
+                ret = xdr_serialize_glusterfs_auth (clnt, fr, auth_data);
                 if (ret == -1) {
-                        gf_log ("rpc-clnt", GF_LOG_DEBUG,
-                                "cannot encode credentials");
+                        gf_log ("rpc-clnt", GF_LOG_WARNING,
+                                "cannot encode auth credentials");
                         goto out;
                 }
 
-                request->rm_call.cb_cred.oa_flavor = AUTH_GLUSTERFS_v2;
+                request->rm_call.cb_cred.oa_flavor = clnt->auth_value;
                 request->rm_call.cb_cred.oa_base   = auth_data;
                 request->rm_call.cb_cred.oa_length = ret;
-        } else {
-                request->rm_call.cb_cred.oa_flavor = AUTH_NULL;
-                request->rm_call.cb_cred.oa_base   = NULL;
-                request->rm_call.cb_cred.oa_length = 0;
         }
         request->rm_call.cb_verf.oa_flavor = AUTH_NONE;
         request->rm_call.cb_verf.oa_base = NULL;
@@ -1365,9 +1495,9 @@ out:
 
 
 struct iobuf *
-rpc_clnt_record_build_record (struct rpc_clnt *clnt, int prognum, int progver,
+rpc_clnt_record_build_record (struct rpc_clnt *clnt, call_frame_t *fr,
+                              int prognum, int progver,
                               int procnum, size_t hdrsize, uint64_t xid,
-                              struct auth_glusterfs_parms_v2 *au,
                               struct iovec *recbuf)
 {
         struct rpc_msg  request                      = {0, };
@@ -1379,17 +1509,13 @@ rpc_clnt_record_build_record (struct rpc_clnt *clnt, int prognum, int progver,
         size_t          xdr_size                     = 0;
         char            auth_data[GF_MAX_AUTH_BYTES] = {0, };
 
-        if ((!clnt) || (!recbuf) || (!au)) {
+        if ((!clnt) || (!recbuf)) {
                 goto out;
         }
 
         /* Fill the rpc structure and XDR it into the buffer got above. */
-        if (clnt->auth_null)
-                ret = rpc_clnt_fill_request (prognum, progver, procnum,
-                                             xid, NULL, &request, NULL);
-        else
-                ret = rpc_clnt_fill_request (prognum, progver, procnum,
-                                             xid, au, &request, auth_data);
+        ret = rpc_clnt_fill_request (clnt, prognum, progver, procnum,
+                                     xid, fr, &request, auth_data);
 
         if (ret == -1) {
                 gf_log (clnt->conn.name, GF_LOG_WARNING,
@@ -1431,80 +1557,21 @@ out:
 }
 
 
-struct iobuf *
+static inline struct iobuf *
 rpc_clnt_record (struct rpc_clnt *clnt, call_frame_t *call_frame,
                  rpc_clnt_prog_t *prog, int procnum, size_t hdrlen,
                  struct iovec *rpchdr, uint64_t callid)
 {
-        struct auth_glusterfs_parms_v2  au          = {0, };
-        struct iobuf                   *request_iob = NULL;
-        char                            owner[4] = {0,};
-        int                             max_groups = 0;
-        int                             max_lkowner_len = 0;
 
         if (!prog || !rpchdr || !call_frame) {
-                goto out;
-        }
-
-        au.pid                   = call_frame->root->pid;
-        au.uid                   = call_frame->root->uid;
-        au.gid                   = call_frame->root->gid;
-        au.groups.groups_len     = call_frame->root->ngrps;
-        au.lk_owner.lk_owner_len = call_frame->root->lk_owner.len;
-
-        if (au.groups.groups_len)
-                au.groups.groups_val = call_frame->root->groups;
-
-        if (call_frame->root->lk_owner.len)
-                au.lk_owner.lk_owner_val = call_frame->root->lk_owner.data;
-        else {
-                owner[0] = (char)(au.pid & 0xff);
-                owner[1] = (char)((au.pid >> 8) & 0xff);
-                owner[2] = (char)((au.pid >> 16) & 0xff);
-                owner[3] = (char)((au.pid >> 24) & 0xff);
-
-                au.lk_owner.lk_owner_val = owner;
-                au.lk_owner.lk_owner_len = 4;
-        }
-
-        /* The number of groups and the size of lk_owner depend on oneother.
-         * We can truncate the groups, but should not touch the lk_owner. */
-        max_groups = GF_AUTH_GLUSTERFS_MAX_GROUPS (au.lk_owner.lk_owner_len);
-        if (au.groups.groups_len > max_groups) {
-                GF_LOG_OCCASIONALLY (gf_auth_max_groups_log, clnt->conn.name,
-                                     GF_LOG_WARNING, "truncating grouplist "
-                                     "from %d to %d", au.groups.groups_len,
-                                     max_groups);
-
-                au.groups.groups_len = max_groups;
-        }
-
-        max_lkowner_len = GF_AUTH_GLUSTERFS_MAX_LKOWNER (au.groups.groups_len);
-        if (au.lk_owner.lk_owner_len > max_lkowner_len) {
-                gf_log (clnt->conn.name, GF_LOG_ERROR, "lkowner field is too "
-                        "big (%d), it does not fit in the rpc-header",
-                        au.lk_owner.lk_owner_len);
-                errno = E2BIG;
-                goto out;
-        }
-
-        gf_log (clnt->conn.name, GF_LOG_TRACE, "Auth Info: pid: %u, uid: %d"
-                ", gid: %d, owner: %s", au.pid, au.uid, au.gid,
-                lkowner_utoa (&call_frame->root->lk_owner));
-
-        request_iob = rpc_clnt_record_build_record (clnt, prog->prognum,
-                                                    prog->progver,
-                                                    procnum, hdrlen,
-                                                    callid, &au,
-                                                    rpchdr);
-        if (!request_iob) {
-                gf_log (clnt->conn.name, GF_LOG_WARNING,
-                        "cannot build rpc-record");
-                goto out;
+                return NULL;
         }
 
-out:
-        return request_iob;
+        return rpc_clnt_record_build_record (clnt, call_frame,
+                                             prog->prognum,
+                                             prog->progver,
+                                             procnum, hdrlen,
+                                             callid, rpchdr);
 }
 
 int
@@ -1887,6 +1954,10 @@ rpc_clnt_disable (struct rpc_clnt *rpc)
 
         if (trans) {
                 rpc_transport_disconnect (trans, _gf_true);
+                /* reset auth_type to use v2 (if its not auth-null), it
+                   would be set to appropriate type in handshake again */
+                if (rpc->auth_value)
+                        rpc->auth_value = AUTH_GLUSTERFS_v2;
         }
 
         if (unref)
@@ -1946,6 +2017,10 @@ rpc_clnt_disconnect (struct rpc_clnt *rpc)
 
         if (trans) {
                 rpc_transport_disconnect (trans, _gf_true);
+                /* reset auth_type to use v2 (if its not auth-null), it
+                   would be set to appropriate type in handshake again */
+                if (rpc->auth_value)
+                        rpc->auth_value = AUTH_GLUSTERFS_v2;
         }
         if (unref)
                 rpc_clnt_unref (rpc);
diff --git a/rpc/rpc-lib/src/rpc-clnt.h b/rpc/rpc-lib/src/rpc-clnt.h
index 6503ca5a57..867592122c 100644
--- a/rpc/rpc-lib/src/rpc-clnt.h
+++ b/rpc/rpc-lib/src/rpc-clnt.h
@@ -15,6 +15,7 @@
 #include "rpc-transport.h"
 #include "timer.h"
 #include "xdr-common.h"
+#include "glusterfs3.h"
 
 typedef enum {
         RPC_CLNT_CONNECT,
@@ -188,12 +189,11 @@ typedef struct rpc_clnt {
 
         glusterfs_ctx_t       *ctx;
         gf_atomic_t           refcount;
-        int                   auth_null;
+        int                   auth_value;
         char                  disabled;
         xlator_t             *owner;
 } rpc_clnt_t;
 
-
 struct rpc_clnt *rpc_clnt_new (dict_t *options, xlator_t *owner,
                                char *name, uint32_t reqpool_size);
 
diff --git a/rpc/rpc-lib/src/rpcsvc-auth.c b/rpc/rpc-lib/src/rpcsvc-auth.c
index bfff0bc557..ef9b35f56a 100644
--- a/rpc/rpc-lib/src/rpcsvc-auth.c
+++ b/rpc/rpc-lib/src/rpcsvc-auth.c
@@ -22,6 +22,8 @@ extern rpcsvc_auth_t *
 rpcsvc_auth_glusterfs_init (rpcsvc_t *svc, dict_t *options);
 extern rpcsvc_auth_t *
 rpcsvc_auth_glusterfs_v2_init (rpcsvc_t *svc, dict_t *options);
+extern rpcsvc_auth_t *
+rpcsvc_auth_glusterfs_v3_init (rpcsvc_t *svc, dict_t *options);
 
 int
 rpcsvc_auth_add_initer (struct list_head *list, char *idfier,
@@ -69,6 +71,15 @@ rpcsvc_auth_add_initers (rpcsvc_t *svc)
                 goto err;
         }
 
+        ret = rpcsvc_auth_add_initer (&svc->authschemes, "auth-glusterfs-v3",
+                                      (rpcsvc_auth_initer_t)
+                                      rpcsvc_auth_glusterfs_v3_init);
+        if (ret == -1) {
+                gf_log (GF_RPCSVC, GF_LOG_ERROR,
+                        "Failed to add AUTH_GLUSTERFS-v3");
+                goto err;
+        }
+
         ret = rpcsvc_auth_add_initer (&svc->authschemes, "auth-unix",
                                       (rpcsvc_auth_initer_t)
                                       rpcsvc_auth_unix_init);
@@ -507,6 +518,7 @@ rpcsvc_auth_unix_auxgids (rpcsvc_request_t *req, int *arrlen)
         case AUTH_UNIX:
         case AUTH_GLUSTERFS:
         case AUTH_GLUSTERFS_v2:
+        case AUTH_GLUSTERFS_v3:
                 break;
         default:
                 gf_log ("rpc", GF_LOG_DEBUG, "auth type not unix or glusterfs");
diff --git a/rpc/rpc-lib/src/rpcsvc.h b/rpc/rpc-lib/src/rpcsvc.h
index 533e6bde37..878eea2899 100644
--- a/rpc/rpc-lib/src/rpcsvc.h
+++ b/rpc/rpc-lib/src/rpcsvc.h
@@ -250,6 +250,16 @@ struct rpcsvc_request {
 
         /* request queue in rpcsvc */
         struct list_head         request_list;
+
+        /* Things passed to rpc layer from client */
+
+        /* @flags: Can be used for binary data passed in xdata to be
+           passed here instead */
+        unsigned int flags;
+
+        /* ctime: origin of time on the client side, ideally this is
+           the one we should consider for time */
+        struct timespec ctime;
 };
 
 #define rpcsvc_request_program(req) ((rpcsvc_program_t *)((req)->prog))
diff --git a/rpc/rpc-lib/src/xdr-common.h b/rpc/rpc-lib/src/xdr-common.h
index dd93110190..db6b5f1a0d 100644
--- a/rpc/rpc-lib/src/xdr-common.h
+++ b/rpc/rpc-lib/src/xdr-common.h
@@ -54,11 +54,15 @@ enum gf_dump_procnum {
  * Note that the on-wire protocol has tighter requirements than the internal
  * structures. It is possible for xlators to use more groups and a bigger
  * lk_owner than that can be sent by a GlusterFS-client.
+ *
+ * -------
+ * On v3, there are 4 more units, and hence it will be 9 xdr-units
  */
-#define GF_AUTH_GLUSTERFS_MAX_GROUPS(lk_owner_len) \
-           (95 - lk_owner_len)
-#define GF_AUTH_GLUSTERFS_MAX_LKOWNER(groups_len)  \
-           (95 - groups_len)
+#define GF_AUTH_GLUSTERFS_MAX_GROUPS(lk_len, type)                      \
+        ((type == AUTH_GLUSTERFS_v2) ? (95 - lk_len) : (91 - lk_len))
+#define GF_AUTH_GLUSTERFS_MAX_LKOWNER(groups_len, type)                 \
+        ((type == AUTH_GLUSTERFS_v2) ? (95 - groups_len) : (91 - groups_len))
+
 
 #ifdef GF_LINUX_HOST_OS
 #define xdr_u_int32_t xdr_uint32_t
diff --git a/rpc/rpc-lib/src/xdr-rpc.h b/rpc/rpc-lib/src/xdr-rpc.h
index ec24ca8200..5560e89328 100644
--- a/rpc/rpc-lib/src/xdr-rpc.h
+++ b/rpc/rpc-lib/src/xdr-rpc.h
@@ -31,6 +31,7 @@ typedef enum {
         AUTH_GLUSTERFS = 5,
         AUTH_GLUSTERFS_v2 = 390039, /* using a number from  'unused' range,
                                        from the list available in RFC5531 */
+        AUTH_GLUSTERFS_v3 = 390040, /* this too is unused */
 } gf_rpc_authtype_t;
 
 /* Converts a given network buffer from its XDR format to a structure
diff --git a/rpc/xdr/src/glusterfs4-xdr.x b/rpc/xdr/src/glusterfs4-xdr.x
index ef0cfde080..9e3223b58b 100644
--- a/rpc/xdr/src/glusterfs4-xdr.x
+++ b/rpc/xdr/src/glusterfs4-xdr.x
@@ -33,6 +33,31 @@ union gfx_value switch (gf_dict_data_type_t type) {
                 opaque other<>;
 };
 
+/* AUTH */
+/* This is used in the rpc header part itself, And not program payload.
+   Avoid sending large data load here. Allowed maximum is 400 bytes.
+   Ref: http://tools.ietf.org/html/rfc5531#section-8.2
+   this is also handled in xdr-common.h
+*/
+struct auth_glusterfs_params_v3 {
+        int pid;
+        unsigned int uid;
+        unsigned int gid;
+
+        /* flags */
+        /* Makes sense to use it for each bits */
+        /* 0x1 == IS_INTERNAL? */
+        /* Another 31 bits are reserved */
+        unsigned int flags;
+
+        /* birth time of the frame / call */
+        unsigned int ctime_nsec; /* good to have 32bit for this */
+        unsigned hyper ctime_sec;
+
+        unsigned int groups<>;
+        opaque lk_owner<>;
+};
+
 struct gfx_dict_pair {
        opaque key<>;
        gfx_value value;
diff --git a/rpc/xdr/src/libgfxdr.sym b/rpc/xdr/src/libgfxdr.sym
index 83f1efc732..f6aa300544 100644
--- a/rpc/xdr/src/libgfxdr.sym
+++ b/rpc/xdr/src/libgfxdr.sym
@@ -1,5 +1,6 @@
 xdr_auth_glusterfs_parms
 xdr_auth_glusterfs_parms_v2
+xdr_auth_glusterfs_params_v3
 xdr_changelog_event_req
 xdr_changelog_event_rsp
 xdr_changelog_probe_req