RPC unify code change

Change-Id: Ibe18a2a63fd023ac57652c4dfc8ac8a69d983b83
BUG: 3112
Signed-off-by: krishna <krishna@gluster.com>
Reviewed-on: http://review.gluster.com/116
Tested-by: Gluster Build System <jenkins@build.gluster.com>
Reviewed-by: Amar Tumballi <amar@gluster.com>

9 files changed, 742 insertions, 43 deletions

diff --git a/rpc/rpc-lib/src/rpc-clnt.c b/rpc/rpc-lib/src/rpc-clnt.c
index 819951c7e..09e676179 100644
--- a/rpc/rpc-lib/src/rpc-clnt.c
+++ b/rpc/rpc-lib/src/rpc-clnt.c
@@ -392,7 +392,8 @@ rpc_clnt_reconnect (void *trans_ptr)
 
                         gf_log (trans->name, GF_LOG_TRACE,
                                 "attempting reconnect");
-                        ret = rpc_transport_connect (trans, conn->config.remote_port);
+                        ret = rpc_transport_connect (trans,
+                                                     conn->config.remote_port);
                         /* Every time there is a disconnection, processes
                            should try to connect to 'glusterd' (ie, default
                            port) or whichever port given as 'option remote-port'
@@ -435,7 +436,8 @@ rpc_clnt_fill_request_info (struct rpc_clnt *clnt, rpc_request_info_t *info)
         pthread_mutex_unlock (&clnt->conn.lock);
 
         if (ret == -1) {
-                gf_log (clnt->conn.trans->name, GF_LOG_CRITICAL, "cannot lookup the saved "
+                gf_log (clnt->conn.trans->name, GF_LOG_CRITICAL,
+                        "cannot lookup the saved "
                         "frame corresponding to xid (%d)", info->xid);
                 goto out;
         }
@@ -633,7 +635,8 @@ rpc_clnt_reply_init (rpc_clnt_connection_t *conn, rpc_transport_pollin_t *msg,
                 goto out;
         }
 
-        gf_log (conn->trans->name, GF_LOG_TRACE, "received rpc message (RPC XID: 0x%ux"
+        gf_log (conn->trans->name, GF_LOG_TRACE,
+                "received rpc message (RPC XID: 0x%ux"
                 " Program: %s, ProgVers: %d, Proc: %d) from rpc-transport (%s)",
                 saved_frame->rpcreq->xid,
                 saved_frame->rpcreq->prog->progname,
@@ -831,8 +834,8 @@ rpc_clnt_notify (rpc_transport_t *trans, void *mydata,
                 pthread_mutex_unlock (&conn->lock);
 
                 if (clnt->notifyfn)
-                        ret = clnt->notifyfn (clnt, clnt->mydata, RPC_CLNT_DISCONNECT,
-                                              NULL);
+                        ret = clnt->notifyfn (clnt, clnt->mydata,
+                                              RPC_CLNT_DISCONNECT, NULL);
                 break;
         }
 
@@ -878,7 +881,8 @@ rpc_clnt_notify (rpc_transport_t *trans, void *mydata,
         case RPC_TRANSPORT_CONNECT:
         {
                 if (clnt->notifyfn)
-                        ret = clnt->notifyfn (clnt, clnt->mydata, RPC_CLNT_CONNECT, NULL);
+                        ret = clnt->notifyfn (clnt, clnt->mydata,
+                                              RPC_CLNT_CONNECT, NULL);
                 break;
         }
 
@@ -1051,7 +1055,8 @@ xdr_serialize_glusterfs_auth (char *dest, struct auth_glusterfs_parms *au)
                        XDR_ENCODE);
 
         if (!xdr_auth_glusterfs_parms (&xdr, au)) {
-                gf_log (THIS->name, GF_LOG_WARNING, "failed to encode auth glusterfs elements");
+                gf_log (THIS->name, GF_LOG_WARNING,
+                        "failed to encode auth glusterfs elements");
                 ret = -1;
                 goto ret;
         }
@@ -1145,7 +1150,8 @@ out:
 struct iobuf *
 rpc_clnt_record_build_record (struct rpc_clnt *clnt, int prognum, int progver,
                               int procnum, size_t payload, uint64_t xid,
-                              struct auth_glusterfs_parms *au, struct iovec *recbuf)
+                              struct auth_glusterfs_parms *au,
+                              struct iovec *recbuf)
 {
         struct rpc_msg           request                            = {0, };
         struct iobuf            *request_iob                        = NULL;
diff --git a/rpc/rpc-lib/src/rpc-common.c b/rpc/rpc-lib/src/rpc-common.c
index db7b92834..d62ed4204 100644
--- a/rpc/rpc-lib/src/rpc-common.c
+++ b/rpc/rpc-lib/src/rpc-common.c
@@ -89,7 +89,8 @@ xdr_gf_prog_detail (XDR *xdrs, gf_prog_detail *objp)
 		 return FALSE;
 	 if (!xdr_u_quad_t (xdrs, &objp->progver))
 		 return FALSE;
-	 if (!xdr_pointer (xdrs, (char **)&objp->next, sizeof (gf_prog_detail), (xdrproc_t) xdr_gf_prog_detail))
+	 if (!xdr_pointer (xdrs, (char **)&objp->next, sizeof (gf_prog_detail),
+                           (xdrproc_t) xdr_gf_prog_detail))
 		 return FALSE;
 	return TRUE;
 }
@@ -103,7 +104,8 @@ xdr_gf_dump_rsp (XDR *xdrs, gf_dump_rsp *objp)
 		 return FALSE;
 	 if (!xdr_int (xdrs, &objp->op_errno))
 		 return FALSE;
-	 if (!xdr_pointer (xdrs, (char **)&objp->prog, sizeof (gf_prog_detail), (xdrproc_t) xdr_gf_prog_detail))
+	 if (!xdr_pointer (xdrs, (char **)&objp->prog, sizeof (gf_prog_detail),
+                           (xdrproc_t) xdr_gf_prog_detail))
 		 return FALSE;
 	return TRUE;
 }
diff --git a/rpc/rpc-lib/src/rpc-transport.c b/rpc/rpc-lib/src/rpc-transport.c
index 082ceec6e..b7245a8c9 100644
--- a/rpc/rpc-lib/src/rpc-transport.c
+++ b/rpc/rpc-lib/src/rpc-transport.c
@@ -525,7 +525,7 @@ rpc_transport_pollin_destroy (rpc_transport_pollin_t *pollin)
         if (pollin->iobref) {
                 iobref_unref (pollin->iobref);
         }
- 
+
         if (pollin->hdr_iobuf) {
                 iobuf_unref (pollin->hdr_iobuf);
         }
@@ -909,7 +909,8 @@ rpc_transport_unref (rpc_transport_t *this)
 
 	if (refcount == 0) {
                 if (this->mydata)
-                        this->notify (this, this->mydata, RPC_TRANSPORT_CLEANUP, NULL);
+                        this->notify (this, this->mydata, RPC_TRANSPORT_CLEANUP,
+                                      NULL);
 		rpc_transport_destroy (this);
 	}
 
@@ -978,7 +979,8 @@ out:
 }
 
 int
-rpc_transport_inet_options_build (dict_t **options, const char *hostname, int port)
+rpc_transport_inet_options_build (dict_t **options, const char *hostname,
+                                  int port)
 {
         dict_t          *dict = NULL;
         char            *host = NULL;
@@ -1000,15 +1002,15 @@ rpc_transport_inet_options_build (dict_t **options, const char *hostname, int po
 
         ret = dict_set_dynstr (dict, "remote-host", host);
         if (ret) {
-                gf_log (THIS->name, GF_LOG_WARNING, "failed to set remote-host with %s",
-                        host);
+                gf_log (THIS->name, GF_LOG_WARNING,
+                        "failed to set remote-host with %s", host);
                 goto out;
         }
 
         ret = dict_set_int32 (dict, "remote-port", port);
         if (ret) {
-                gf_log (THIS->name, GF_LOG_WARNING, "failed to set remote-port with %d",
-                        port);
+                gf_log (THIS->name, GF_LOG_WARNING,
+                        "failed to set remote-port with %d", port);
                 goto out;
         }
         ret = dict_set_str (dict, "transport.address-family", "inet/inet6");
diff --git a/rpc/rpc-lib/src/rpc-transport.h b/rpc/rpc-lib/src/rpc-transport.h
index f8396f572..5790b23f4 100644
--- a/rpc/rpc-lib/src/rpc-transport.h
+++ b/rpc/rpc-lib/src/rpc-transport.h
@@ -188,7 +188,7 @@ struct rpc_transport {
                                               * transport came from. valid only
                                               * on server process.
                                               */
-        
+
 	void                      *private;
         void                      *xl_private;
         void                      *xl;       /* Used for THIS */
@@ -203,7 +203,7 @@ struct rpc_transport {
 	data_t                    *buf;
  	int32_t                  (*init)   (rpc_transport_t *this);
 	void                     (*fini)   (rpc_transport_t *this);
-        int32_t                  (*validate_options) (rpc_transport_t *this, 
+        int32_t                  (*validate_options) (rpc_transport_t *this,
                                                       char **op_errstr);
         int                      (*reconfigure) (rpc_transport_t *this, dict_t *options);
         rpc_transport_notify_t     notify;
diff --git a/rpc/rpc-lib/src/rpcsvc-auth.c b/rpc/rpc-lib/src/rpcsvc-auth.c
index fcf4617ca..558ada509 100644
--- a/rpc/rpc-lib/src/rpcsvc-auth.c
+++ b/rpc/rpc-lib/src/rpcsvc-auth.c
@@ -318,8 +318,8 @@ rpcsvc_authenticate (rpcsvc_request_t *req)
         if (!req)
                 return ret;
 
-        //minauth = rpcsvc_request_prog_minauth (req);
-        minauth = 1;
+        /* FIXME use rpcsvc_request_prog_minauth() */
+        minauth = 0;
         if (minauth > rpcsvc_request_cred_flavour (req)) {
                 gf_log (GF_RPCSVC, GF_LOG_WARNING, "Auth too weak");
                 rpcsvc_request_set_autherr (req, AUTH_TOOWEAK);
@@ -339,6 +339,93 @@ err:
         return ret;
 }
 
+int
+rpcsvc_auth_array (rpcsvc_t *svc, char *volname, int *autharr, int arrlen)
+{
+        int             count = 0;
+        int             gen = RPCSVC_AUTH_REJECT;
+        int             spec = RPCSVC_AUTH_REJECT;
+        int             final = RPCSVC_AUTH_REJECT;
+        char            *srchstr = NULL;
+        char            *valstr = NULL;
+        gf_boolean_t    boolval = _gf_false;
+        int             ret = 0;
+
+        struct rpcsvc_auth_list *auth = NULL;
+        struct rpcsvc_auth_list *tmp = NULL;
+
+        if ((!svc) || (!autharr) || (!volname))
+                return -1;
+
+        memset (autharr, 0, arrlen * sizeof(int));
+        if (list_empty (&svc->authschemes)) {
+                gf_log (GF_RPCSVC, GF_LOG_ERROR, "No authentication!");
+                goto err;
+        }
+
+        list_for_each_entry_safe (auth, tmp, &svc->authschemes, authlist) {
+                if (count >= arrlen)
+                        break;
+
+                gen = gf_asprintf (&srchstr, "rpc-auth.%s", auth->name);
+                if (gen == -1) {
+                        count = -1;
+                        goto err;
+                }
+
+                gen = RPCSVC_AUTH_REJECT;
+                if (dict_get (svc->options, srchstr)) {
+                        ret = dict_get_str (svc->options, srchstr, &valstr);
+                        if (ret == 0) {
+                                ret = gf_string2boolean (valstr, &boolval);
+                                if (ret == 0) {
+                                        if (boolval == _gf_true)
+                                                gen = RPCSVC_AUTH_ACCEPT;
+                                } else
+                                        gf_log (GF_RPCSVC, GF_LOG_ERROR, "Faile"
+                                                "d to read auth val");
+                        } else
+                                gf_log (GF_RPCSVC, GF_LOG_ERROR, "Faile"
+                                        "d to read auth val");
+                }
+
+                GF_FREE (srchstr);
+                spec = gf_asprintf (&srchstr, "rpc-auth.%s.%s", auth->name,
+                                    volname);
+                if (spec == -1) {
+                        count = -1;
+                        goto err;
+                }
+
+                spec = RPCSVC_AUTH_DONTCARE;
+                if (dict_get (svc->options, srchstr)) {
+                        ret = dict_get_str (svc->options, srchstr, &valstr);
+                        if (ret == 0) {
+                                ret = gf_string2boolean (valstr, &boolval);
+                                if (ret == 0) {
+                                        if (boolval == _gf_true)
+                                                spec = RPCSVC_AUTH_ACCEPT;
+                                        else
+                                                spec = RPCSVC_AUTH_REJECT;
+                                } else
+                                        gf_log (GF_RPCSVC, GF_LOG_ERROR, "Faile"
+                                                "d to read auth val");
+                        } else
+                                gf_log (GF_RPCSVC, GF_LOG_ERROR, "Faile"
+                                        "d to read auth val");
+                }
+
+                GF_FREE (srchstr);
+                final = rpcsvc_combine_gen_spec_volume_checks (gen, spec);
+                if (final == RPCSVC_AUTH_ACCEPT) {
+                        autharr[count] = auth->auth->authnum;
+                        ++count;
+                }
+        }
+
+err:
+        return count;
+}
 
 gid_t *
 rpcsvc_auth_unix_auxgids (rpcsvc_request_t *req, int *arrlen)
diff --git a/rpc/rpc-lib/src/rpcsvc-common.h b/rpc/rpc-lib/src/rpcsvc-common.h
index 442049433..233dac53c 100644
--- a/rpc/rpc-lib/src/rpcsvc-common.h
+++ b/rpc/rpc-lib/src/rpcsvc-common.h
@@ -60,7 +60,7 @@ typedef struct rpcsvc_state {
 
         /* Allow insecure ports. */
         int                     allow_insecure;
-
+        gf_boolean_t            register_portmap;
         glusterfs_ctx_t         *ctx;
 
         /* list of connections which will listen for incoming connections */
diff --git a/rpc/rpc-lib/src/rpcsvc.c b/rpc/rpc-lib/src/rpcsvc.c
index 519d4e346..10198b0f3 100644
--- a/rpc/rpc-lib/src/rpcsvc.c
+++ b/rpc/rpc-lib/src/rpcsvc.c
@@ -105,8 +105,33 @@ out:
         return;
 }
 
+rpcsvc_vector_sizer
+rpcsvc_get_program_vector_sizer (rpcsvc_t *svc, uint32_t prognum,
+                                 uint32_t progver, uint32_t procnum)
+{
+        rpcsvc_program_t        *program = NULL;
+        char                    found    = 0;
 
+        if (!svc)
+                return NULL;
 
+        pthread_mutex_lock (&svc->rpclock);
+        {
+                list_for_each_entry (program, &svc->programs, program) {
+                        if ((program->prognum == prognum)
+                            && (program->progver == progver)) {
+                                found = 1;
+                                break;
+                        }
+                }
+        }
+        pthread_mutex_unlock (&svc->rpclock);
+
+        if (found)
+                return program->actors[procnum].vector_sizer;
+        else
+                return NULL;
+}
 
 /* This needs to change to returning errors, since
  * we need to return RPC specific error messages when some
@@ -442,8 +467,10 @@ rpcsvc_handle_rpc_call (rpcsvc_t *svc, rpc_transport_t *trans,
 
                 gf_log ("rpcsvc", GF_LOG_TRACE, "Client port: %d", (int)port);
 
-                if ((port > 1024) && (0 == svc->allow_insecure)) {  //Non-privileged user, fail request
-                        gf_log ("glusterd", GF_LOG_ERROR, "Request received from non-"
+                if ((port > 1024) && (0 == svc->allow_insecure)) {
+                        /* Non-privileged user, fail request */
+                        gf_log ("glusterd", GF_LOG_ERROR,
+                                "Request received from non-"
                                 "privileged port. Failing request");
                         return -1;
                 }
@@ -505,10 +532,10 @@ rpcsvc_handle_disconnect (rpcsvc_t *svc, rpc_transport_t *trans)
         rpcsvc_notify_wrapper_t *wrappers = NULL, *wrapper;
         int32_t                  ret      = -1, i = 0, wrapper_count = 0;
         rpcsvc_listener_t       *listener = NULL;
- 
+
         event = (trans->listener == NULL) ? RPCSVC_EVENT_LISTENER_DEAD
                 : RPCSVC_EVENT_DISCONNECT;
- 
+
         pthread_mutex_lock (&svc->rpclock);
         {
                 wrappers = GF_CALLOC (svc->notify_count, sizeof (*wrapper),
@@ -516,35 +543,35 @@ rpcsvc_handle_disconnect (rpcsvc_t *svc, rpc_transport_t *trans)
                 if (!wrappers) {
                         goto unlock;
                 }
- 
+
                 list_for_each_entry (wrapper, &svc->notify, list) {
                         if (wrapper->notify) {
                                 wrappers[i++] = *wrapper;
                         }
                 }
- 
+
                 wrapper_count = i;
         }
 unlock:
         pthread_mutex_unlock (&svc->rpclock);
- 
+
         if (wrappers) {
                 for (i = 0; i < wrapper_count; i++) {
                         wrappers[i].notify (svc, wrappers[i].data,
                                             event, trans);
                 }
- 
+
                 GF_FREE (wrappers);
         }
- 
+
         if (event == RPCSVC_EVENT_LISTENER_DEAD) {
                 listener = rpcsvc_get_listener (svc, -1, trans->listener);
                 rpcsvc_listener_destroy (listener);
         }
- 
+
         return ret;
 }
- 
+
 
 int
 rpcsvc_notify (rpc_transport_t *trans, void *mydata,
@@ -1114,6 +1141,11 @@ out:
         return ret;
 }
 
+int
+rpcsvc_register_portmap_enabled (rpcsvc_t *svc)
+{
+        return svc->register_portmap;
+}
 
 int32_t
 rpcsvc_get_listener_port (rpcsvc_listener_t *listener)
@@ -1152,7 +1184,7 @@ rpcsvc_get_listener (rpcsvc_t *svc, uint16_t port, rpc_transport_t *trans)
 {
         rpcsvc_listener_t  *listener      = NULL;
         char                found         = 0;
-        uint32_t            listener_port = 0; 
+        uint32_t            listener_port = 0;
 
         if (!svc) {
                 goto out;
@@ -1343,7 +1375,7 @@ rpcsvc_create_listener (rpcsvc_t *svc, dict_t *options, char *name)
 {
         rpc_transport_t   *trans    = NULL;
         rpcsvc_listener_t *listener = NULL;
-        int32_t            ret      = -1; 
+        int32_t            ret      = -1;
 
         if (!svc || !options) {
                 goto out;
@@ -1577,7 +1609,6 @@ out:
         return ret;
 }
 
-
 static void
 free_prog_details (gf_dump_rsp *rsp)
 {
@@ -1670,8 +1701,39 @@ fail:
 int
 rpcsvc_init_options (rpcsvc_t *svc, dict_t *options)
 {
+        char            *optstr = NULL;
+        int             ret = -1;
+
+        if ((!svc) || (!options))
+                return -1;
+
         svc->memfactor = RPCSVC_DEFAULT_MEMFACTOR;
-        return 0;
+
+        svc->register_portmap = _gf_true;
+        if (dict_get (options, "rpc.register-with-portmap")) {
+                ret = dict_get_str (options, "rpc.register-with-portmap",
+                                    &optstr);
+                if (ret < 0) {
+                        gf_log (GF_RPCSVC, GF_LOG_ERROR, "Failed to parse "
+                                "dict");
+                        goto out;
+                }
+
+                ret = gf_string2boolean (optstr, &svc->register_portmap);
+                if (ret < 0) {
+                        gf_log (GF_RPCSVC, GF_LOG_ERROR, "Failed to parse bool "
+                                "string");
+                        goto out;
+                }
+        }
+
+        if (!svc->register_portmap)
+                gf_log (GF_RPCSVC, GF_LOG_DEBUG, "Portmap registration "
+                        "disabled");
+
+        ret = 0;
+out:
+        return ret;
 }
 
 int
@@ -1789,6 +1851,531 @@ free_svc:
 }
 
 
+int
+rpcsvc_transport_peer_check_search (dict_t *options, char *pattern, char *clstr)
+{
+        int                     ret = -1;
+        char                    *addrtok = NULL;
+        char                    *addrstr = NULL;
+        char                    *svptr = NULL;
+
+        if ((!options) || (!clstr))
+                return -1;
+
+        if (!dict_get (options, pattern))
+                return -1;
+
+        ret = dict_get_str (options, pattern, &addrstr);
+        if (ret < 0) {
+                ret = -1;
+                goto err;
+        }
+
+        if (!addrstr) {
+                ret = -1;
+                goto err;
+        }
+
+        addrtok = strtok_r (addrstr, ",", &svptr);
+        while (addrtok) {
+
+                /* CASEFOLD not present on Solaris */
+#ifdef FNM_CASEFOLD
+                ret = fnmatch (addrtok, clstr, FNM_CASEFOLD);
+#else
+                ret = fnmatch (addrtok, clstr, 0);
+#endif
+                if (ret == 0)
+                        goto err;
+
+                addrtok = strtok_r (NULL, ",", &svptr);
+        }
+
+        ret = -1;
+err:
+
+        return ret;
+}
+
+
+int
+rpcsvc_transport_peer_check_allow (dict_t *options, char *volname, char *clstr)
+{
+        int     ret = RPCSVC_AUTH_DONTCARE;
+        char    *srchstr = NULL;
+        char    globalrule[] = "rpc-auth.addr.allow";
+
+        if ((!options) || (!clstr))
+                return ret;
+
+        /* If volname is NULL, then we're searching for the general rule to
+         * determine the current address in clstr is allowed or not for all
+         * subvolumes.
+         */
+        if (volname) {
+                ret = gf_asprintf (&srchstr, "rpc-auth.addr.%s.allow", volname);
+                if (ret == -1) {
+                        gf_log (GF_RPCSVC, GF_LOG_ERROR, "asprintf failed");
+                        ret = RPCSVC_AUTH_DONTCARE;
+                        goto out;
+                }
+        } else
+                srchstr = globalrule;
+
+        ret = rpcsvc_transport_peer_check_search (options, srchstr, clstr);
+        if (volname)
+                GF_FREE (srchstr);
+
+        if (ret == 0)
+                ret = RPCSVC_AUTH_ACCEPT;
+        else
+                ret = RPCSVC_AUTH_DONTCARE;
+out:
+        return ret;
+}
+
+int
+rpcsvc_transport_peer_check_reject (dict_t *options, char *volname, char *clstr)
+{
+        int     ret = RPCSVC_AUTH_DONTCARE;
+        char    *srchstr = NULL;
+        char    generalrule[] = "rpc-auth.addr.reject";
+
+        if ((!options) || (!clstr))
+                return ret;
+
+        if (volname) {
+                ret = gf_asprintf (&srchstr, "rpc-auth.addr.%s.reject",
+                                   volname);
+                if (ret == -1) {
+                        gf_log (GF_RPCSVC, GF_LOG_ERROR, "asprintf failed");
+                        ret = RPCSVC_AUTH_REJECT;
+                        goto out;
+                }
+        } else
+                srchstr = generalrule;
+
+        ret = rpcsvc_transport_peer_check_search (options, srchstr, clstr);
+        if (volname)
+                GF_FREE (srchstr);
+
+        if (ret == 0)
+                ret = RPCSVC_AUTH_REJECT;
+        else
+                ret = RPCSVC_AUTH_DONTCARE;
+out:
+        return ret;
+}
+
+
+/* This function tests the results of the allow rule and the reject rule to
+ * combine them into a single result that can be used to determine if the
+ * connection should be allowed to proceed.
+ * Heres the test matrix we need to follow in this function.
+ *
+ * A -  Allow, the result of the allow test. Never returns R.
+ * R - Reject, result of the reject test. Never returns A.
+ * Both can return D or dont care if no rule was given.
+ *
+ * | @allow | @reject | Result |
+ * |    A   |   R     | R      |
+ * |    D   |   D     | D      |
+ * |    A   |   D     | A      |
+ * |    D   |   R     | R      |
+ */
+int
+rpcsvc_combine_allow_reject_volume_check (int allow, int reject)
+{
+        int     final = RPCSVC_AUTH_REJECT;
+
+        /* If allowed rule allows but reject rule rejects, we stay cautious
+         * and reject. */
+        if ((allow == RPCSVC_AUTH_ACCEPT) && (reject == RPCSVC_AUTH_REJECT))
+                final = RPCSVC_AUTH_REJECT;
+        /* if both are dont care, that is user did not specify for either allow
+         * or reject, we leave it up to the general rule to apply, in the hope
+         * that there is one.
+         */
+        else if ((allow == RPCSVC_AUTH_DONTCARE) &&
+                 (reject == RPCSVC_AUTH_DONTCARE))
+                final = RPCSVC_AUTH_DONTCARE;
+        /* If one is dont care, the other one applies. */
+        else if ((allow == RPCSVC_AUTH_ACCEPT) &&
+                 (reject == RPCSVC_AUTH_DONTCARE))
+                final = RPCSVC_AUTH_ACCEPT;
+        else if ((allow == RPCSVC_AUTH_DONTCARE) &&
+                 (reject == RPCSVC_AUTH_REJECT))
+                final = RPCSVC_AUTH_REJECT;
+
+        return final;
+}
+
+
+/* Combines the result of the general rule test against, the specific rule
+ * to determine final permission for the client's address.
+ *
+ * | @gen   | @spec   | Result |
+ * |    A   |   A     | A      |
+ * |    A   |   R     | R      |
+ * |    A   |   D     | A      |
+ * |    D   |   A     | A      |
+ * |    D   |   R     | R      |
+ * |    D   |   D     | D      |
+ * |    R   |   A     | A      |
+ * |    R   |   D     | R      |
+ * |    R   |   R     | R      |
+ */
+int
+rpcsvc_combine_gen_spec_addr_checks (int gen, int spec)
+{
+        int     final = RPCSVC_AUTH_REJECT;
+
+        if ((gen == RPCSVC_AUTH_ACCEPT) && (spec == RPCSVC_AUTH_ACCEPT))
+                final = RPCSVC_AUTH_ACCEPT;
+        else if ((gen == RPCSVC_AUTH_ACCEPT) && (spec == RPCSVC_AUTH_REJECT))
+                final = RPCSVC_AUTH_REJECT;
+        else if ((gen == RPCSVC_AUTH_ACCEPT) && (spec == RPCSVC_AUTH_DONTCARE))
+                final = RPCSVC_AUTH_ACCEPT;
+        else if ((gen == RPCSVC_AUTH_DONTCARE) && (spec == RPCSVC_AUTH_ACCEPT))
+                final = RPCSVC_AUTH_ACCEPT;
+        else if ((gen == RPCSVC_AUTH_DONTCARE) && (spec == RPCSVC_AUTH_REJECT))
+                final = RPCSVC_AUTH_REJECT;
+        else if ((gen == RPCSVC_AUTH_DONTCARE) && (spec== RPCSVC_AUTH_DONTCARE))
+                final = RPCSVC_AUTH_DONTCARE;
+        else if ((gen == RPCSVC_AUTH_REJECT) && (spec == RPCSVC_AUTH_ACCEPT))
+                final = RPCSVC_AUTH_ACCEPT;
+        else if ((gen == RPCSVC_AUTH_REJECT) && (spec == RPCSVC_AUTH_DONTCARE))
+                final = RPCSVC_AUTH_REJECT;
+        else if ((gen == RPCSVC_AUTH_REJECT) && (spec == RPCSVC_AUTH_REJECT))
+                final = RPCSVC_AUTH_REJECT;
+
+        return final;
+}
+
+
+
+/* Combines the result of the general rule test against, the specific rule
+ * to determine final test for the connection coming in for a given volume.
+ *
+ * | @gen   | @spec   | Result |
+ * |    A   |   A     | A      |
+ * |    A   |   R     | R      |
+ * |    A   |   D     | A      |
+ * |    D   |   A     | A      |
+ * |    D   |   R     | R      |
+ * |    D   |   D     | R      |, special case, we intentionally disallow this.
+ * |    R   |   A     | A      |
+ * |    R   |   D     | R      |
+ * |    R   |   R     | R      |
+ */
+int
+rpcsvc_combine_gen_spec_volume_checks (int gen, int spec)
+{
+        int     final = RPCSVC_AUTH_REJECT;
+
+        if ((gen == RPCSVC_AUTH_ACCEPT) && (spec == RPCSVC_AUTH_ACCEPT))
+                final = RPCSVC_AUTH_ACCEPT;
+        else if ((gen == RPCSVC_AUTH_ACCEPT) && (spec == RPCSVC_AUTH_REJECT))
+                final = RPCSVC_AUTH_REJECT;
+        else if ((gen == RPCSVC_AUTH_ACCEPT) && (spec == RPCSVC_AUTH_DONTCARE))
+                final = RPCSVC_AUTH_ACCEPT;
+        else if ((gen == RPCSVC_AUTH_DONTCARE) && (spec == RPCSVC_AUTH_ACCEPT))
+                final = RPCSVC_AUTH_ACCEPT;
+        else if ((gen == RPCSVC_AUTH_DONTCARE) && (spec == RPCSVC_AUTH_REJECT))
+                final = RPCSVC_AUTH_REJECT;
+        /* On no rule, we reject. */
+        else if ((gen == RPCSVC_AUTH_DONTCARE) && (spec== RPCSVC_AUTH_DONTCARE))
+                final = RPCSVC_AUTH_REJECT;
+        else if ((gen == RPCSVC_AUTH_REJECT) && (spec == RPCSVC_AUTH_ACCEPT))
+                final = RPCSVC_AUTH_ACCEPT;
+        else if ((gen == RPCSVC_AUTH_REJECT) && (spec == RPCSVC_AUTH_DONTCARE))
+                final = RPCSVC_AUTH_REJECT;
+        else if ((gen == RPCSVC_AUTH_REJECT) && (spec == RPCSVC_AUTH_REJECT))
+                final = RPCSVC_AUTH_REJECT;
+
+        return final;
+}
+
+
+int
+rpcsvc_transport_peer_check_name (dict_t *options, char *volname,
+                                  rpc_transport_t *trans)
+{
+        int     ret = RPCSVC_AUTH_REJECT;
+        int     aret = RPCSVC_AUTH_REJECT;
+        int     rjret = RPCSVC_AUTH_REJECT;
+        char    clstr[RPCSVC_PEER_STRLEN];
+
+        if (!trans)
+                return ret;
+
+        ret = rpcsvc_transport_peername (trans, clstr, RPCSVC_PEER_STRLEN);
+        if (ret != 0) {
+                gf_log (GF_RPCSVC, GF_LOG_ERROR, "Failed to get remote addr: "
+                        "%s", gai_strerror (ret));
+                ret = RPCSVC_AUTH_REJECT;
+                goto err;
+        }
+
+        aret = rpcsvc_transport_peer_check_allow (options, volname, clstr);
+        rjret = rpcsvc_transport_peer_check_reject (options, volname, clstr);
+
+        ret = rpcsvc_combine_allow_reject_volume_check (aret, rjret);
+
+err:
+        return ret;
+}
+
+
+int
+rpcsvc_transport_peer_check_addr (dict_t *options, char *volname,
+                                  rpc_transport_t *trans)
+{
+        int     ret = RPCSVC_AUTH_REJECT;
+        int     aret = RPCSVC_AUTH_DONTCARE;
+        int     rjret = RPCSVC_AUTH_REJECT;
+        char    clstr[RPCSVC_PEER_STRLEN];
+        struct sockaddr_storage sastorage = {0,};
+
+        if (!trans)
+                return ret;
+
+        ret = rpcsvc_transport_peeraddr (trans, clstr, RPCSVC_PEER_STRLEN,
+                                         &sastorage, sizeof (sastorage));
+        if (ret != 0) {
+                gf_log (GF_RPCSVC, GF_LOG_ERROR, "Failed to get remote addr: "
+                        "%s", gai_strerror (ret));
+                ret = RPCSVC_AUTH_REJECT;
+                goto err;
+        }
+
+        aret = rpcsvc_transport_peer_check_allow (options, volname, clstr);
+        rjret = rpcsvc_transport_peer_check_reject (options, volname, clstr);
+
+        ret = rpcsvc_combine_allow_reject_volume_check (aret, rjret);
+err:
+        return ret;
+}
+
+
+int
+rpcsvc_transport_check_volume_specific (dict_t *options, char *volname,
+                                        rpc_transport_t *trans)
+{
+        int             namechk = RPCSVC_AUTH_REJECT;
+        int             addrchk = RPCSVC_AUTH_REJECT;
+        gf_boolean_t    namelookup = _gf_true;
+        char            *namestr = NULL;
+        int             ret = 0;
+
+        if ((!options) || (!volname) || (!trans))
+                return RPCSVC_AUTH_REJECT;
+
+        /* Enabled by default */
+        if ((dict_get (options, "rpc-auth.addr.namelookup"))) {
+                ret = dict_get_str (options, "rpc-auth.addr.namelookup"
+                                    , &namestr);
+                if (ret == 0)
+                        ret = gf_string2boolean (namestr, &namelookup);
+        }
+
+        /* We need two separate checks because the rules with addresses in them
+         * can be network addresses which can be general and names can be
+         * specific which will over-ride the network address rules.
+         */
+        if (namelookup)
+                namechk = rpcsvc_transport_peer_check_name (options, volname,
+                                                            trans);
+        addrchk = rpcsvc_transport_peer_check_addr (options, volname, trans);
+
+        if (namelookup)
+                ret = rpcsvc_combine_gen_spec_addr_checks (addrchk,
+                                                               namechk);
+        else
+                ret = addrchk;
+
+        return ret;
+}
+
+
+int
+rpcsvc_transport_check_volume_general (dict_t *options, rpc_transport_t *trans)
+{
+        int             addrchk = RPCSVC_AUTH_REJECT;
+        int             namechk = RPCSVC_AUTH_REJECT;
+        gf_boolean_t    namelookup = _gf_true;
+        char            *namestr = NULL;
+        int             ret = 0;
+
+        if ((!options) || (!trans))
+                return RPCSVC_AUTH_REJECT;
+
+        /* Enabled by default */
+        if ((dict_get (options, "rpc-auth.addr.namelookup"))) {
+                ret = dict_get_str (options, "rpc-auth.addr.namelookup"
+                                    , &namestr);
+                if (ret == 0)
+                        ret = gf_string2boolean (namestr, &namelookup);
+        }
+
+        /* We need two separate checks because the rules with addresses in them
+         * can be network addresses which can be general and names can be
+         * specific which will over-ride the network address rules.
+         */
+        if (namelookup)
+                namechk = rpcsvc_transport_peer_check_name (options, NULL, trans);
+        addrchk = rpcsvc_transport_peer_check_addr (options, NULL, trans);
+
+        if (namelookup)
+                ret = rpcsvc_combine_gen_spec_addr_checks (addrchk,
+                                                               namechk);
+        else
+                ret = addrchk;
+
+        return ret;
+}
+
+int
+rpcsvc_transport_peer_check (dict_t *options, char *volname,
+                             rpc_transport_t *trans)
+{
+        int     general_chk = RPCSVC_AUTH_REJECT;
+        int     specific_chk = RPCSVC_AUTH_REJECT;
+
+        if ((!options) || (!volname) || (!trans))
+                return RPCSVC_AUTH_REJECT;
+
+        general_chk = rpcsvc_transport_check_volume_general (options, trans);
+        specific_chk = rpcsvc_transport_check_volume_specific (options, volname,
+                                                               trans);
+
+        return rpcsvc_combine_gen_spec_volume_checks (general_chk,
+                                                      specific_chk);
+}
+
+
+int
+rpcsvc_transport_privport_check (rpcsvc_t *svc, char *volname,
+                                 rpc_transport_t *trans)
+{
+        struct sockaddr_storage sastorage = {0,};
+        struct sockaddr_in     *sa = NULL;
+        int                     ret = RPCSVC_AUTH_REJECT;
+        socklen_t               sasize = sizeof (sa);
+        char                    *srchstr = NULL;
+        char                    *valstr = NULL;
+        int                     globalinsecure = RPCSVC_AUTH_REJECT;
+        int                     exportinsecure = RPCSVC_AUTH_DONTCARE;
+        uint16_t                port = 0;
+        gf_boolean_t            insecure = _gf_false;
+
+        if ((!svc) || (!volname) || (!trans))
+                return ret;
+
+        sa = (struct sockaddr_in*) &sastorage;
+        ret = rpcsvc_transport_peeraddr (trans, NULL, 0, &sastorage,
+                                         sasize);
+        if (ret != 0) {
+                gf_log (GF_RPCSVC, GF_LOG_ERROR, "Failed to get peer addr: %s",
+                        gai_strerror (ret));
+                ret = RPCSVC_AUTH_REJECT;
+                goto err;
+        }
+
+        port = ntohs (sa->sin_port);
+        gf_log (GF_RPCSVC, GF_LOG_TRACE, "Client port: %d", (int)port);
+        /* If the port is already a privileged one, dont bother with checking
+         * options.
+         */
+        if (port <= 1024) {
+                ret = RPCSVC_AUTH_ACCEPT;
+                goto err;
+        }
+
+        /* Disabled by default */
+        if ((dict_get (svc->options, "rpc-auth.ports.insecure"))) {
+                ret = dict_get_str (svc->options, "rpc-auth.ports.insecure"
+                                    , &srchstr);
+                if (ret == 0) {
+                        ret = gf_string2boolean (srchstr, &insecure);
+                        if (ret == 0) {
+                                if (insecure == _gf_true)
+                                        globalinsecure = RPCSVC_AUTH_ACCEPT;
+                        } else
+                                gf_log (GF_RPCSVC, GF_LOG_ERROR, "Failed to"
+                                        " read rpc-auth.ports.insecure value");
+                } else
+                        gf_log (GF_RPCSVC, GF_LOG_ERROR, "Failed to"
+                                " read rpc-auth.ports.insecure value");
+        }
+
+        /* Disabled by default */
+        ret = gf_asprintf (&srchstr, "rpc-auth.ports.%s.insecure", volname);
+        if (ret == -1) {
+                gf_log (GF_RPCSVC, GF_LOG_ERROR, "asprintf failed");
+                ret = RPCSVC_AUTH_REJECT;
+                goto err;
+        }
+
+        if (dict_get (svc->options, srchstr)) {
+                ret = dict_get_str (svc->options, srchstr, &valstr);
+                if (ret == 0) {
+                        ret = gf_string2boolean (srchstr, &insecure);
+                        if (ret == 0) {
+                                if (insecure == _gf_true)
+                                       exportinsecure = RPCSVC_AUTH_ACCEPT;
+                                else
+                                        exportinsecure = RPCSVC_AUTH_REJECT;
+                        } else
+                                gf_log (GF_RPCSVC, GF_LOG_ERROR, "Failed to"
+                                        " read rpc-auth.ports.insecure value");
+                } else
+                        gf_log (GF_RPCSVC, GF_LOG_ERROR, "Failed to"
+                                " read rpc-auth.ports.insecure value");
+        }
+
+        ret = rpcsvc_combine_gen_spec_volume_checks (globalinsecure,
+                                                         exportinsecure);
+        if (ret == RPCSVC_AUTH_ACCEPT)
+                gf_log (GF_RPCSVC, GF_LOG_DEBUG, "Unprivileged port allowed");
+        else
+                gf_log (GF_RPCSVC, GF_LOG_DEBUG, "Unprivileged port not"
+                        " allowed");
+
+err:
+        return ret;
+}
+
+
+char *
+rpcsvc_volume_allowed (dict_t *options, char *volname)
+{
+        char    globalrule[] = "rpc-auth.addr.allow";
+        char    *srchstr = NULL;
+        char    *addrstr = NULL;
+        int     ret = -1;
+
+        if ((!options) || (!volname))
+                return NULL;
+
+        ret = gf_asprintf (&srchstr, "rpc-auth.addr.%s.allow", volname);
+        if (ret == -1) {
+                gf_log (GF_RPCSVC, GF_LOG_ERROR, "asprintf failed");
+                goto out;
+        }
+
+        if (!dict_get (options, srchstr)) {
+                GF_FREE (srchstr);
+                srchstr = globalrule;
+                ret = dict_get_str (options, srchstr, &addrstr);
+        } else
+                ret = dict_get_str (options, srchstr, &addrstr);
+
+out:
+        return addrstr;
+}
+
+
 rpcsvc_actor_t gluster_dump_actors[] = {
         [GF_DUMP_NULL] = {"NULL", GF_DUMP_NULL, NULL, NULL, NULL },
         [GF_DUMP_DUMP] = {"DUMP", GF_DUMP_DUMP, rpcsvc_dump, NULL, NULL },
diff --git a/rpc/rpc-lib/src/rpcsvc.h b/rpc/rpc-lib/src/rpcsvc.h
index 17da800fa..7ef6e580c 100644
--- a/rpc/rpc-lib/src/rpcsvc.h
+++ b/rpc/rpc-lib/src/rpcsvc.h
@@ -249,7 +249,8 @@ struct rpcsvc_request {
 };
 
 #define rpcsvc_request_program(req) ((rpcsvc_program_t *)((req)->prog))
-#define rpcsvc_request_program_private(req) (((rpcsvc_program_t *)((req)->program))->private)
+#define rpcsvc_request_procnum(req) (((req)->procnum))
+#define rpcsvc_request_program_private(req) (((rpcsvc_program_t *)((req)->prog))->private)
 #define rpcsvc_request_accepted(req)    ((req)->rpc_status == MSG_ACCEPTED)
 #define rpcsvc_request_accepted_success(req) ((req)->rpc_err == SUCCESS)
 #define rpcsvc_request_uid(req)         ((req)->uid)
@@ -257,18 +258,24 @@ struct rpcsvc_request {
 #define rpcsvc_request_prog_minauth(req) (rpcsvc_request_program(req)->min_auth)
 #define rpcsvc_request_cred_flavour(req) (rpcsvc_auth_flavour(req->cred))
 #define rpcsvc_request_verf_flavour(req) (rpcsvc_auth_flavour(req->verf))
-
+#define rpcsvc_request_service(req)      ((req)->svc)
 #define rpcsvc_request_uid(req)         ((req)->uid)
 #define rpcsvc_request_gid(req)         ((req)->gid)
 #define rpcsvc_request_private(req)     ((req)->private)
 #define rpcsvc_request_xid(req)         ((req)->xid)
 #define rpcsvc_request_set_private(req,prv)  (req)->private = (void *)(prv)
+#define rpcsvc_request_iobref_ref(req)  (iobref_ref ((req)->iobref))
 #define rpcsvc_request_record_ref(req)  (iobuf_ref ((req)->recordiob))
 #define rpcsvc_request_record_unref(req) (iobuf_unref ((req)->recordiob))
+#define rpcsvc_request_record_iob(req)   ((req)->recordiob)
+#define rpcsvc_request_set_vecstate(req, state)  ((req)->vecstate = state)
+#define rpcsvc_request_vecstate(req) ((req)->vecstate)
+#define rpcsvc_request_transport(req) ((req)->trans)
 
 
 #define RPCSVC_ACTOR_SUCCESS    0
 #define RPCSVC_ACTOR_ERROR      (-1)
+#define RPCSVC_ACTOR_IGNORE     (-2)
 
 /* Functor for every type of protocol actor
  * must be defined like this.
@@ -285,8 +292,7 @@ struct rpcsvc_request {
 typedef int (*rpcsvc_actor) (rpcsvc_request_t *req);
 typedef int (*rpcsvc_vector_actor) (rpcsvc_request_t *req, struct iovec *vec,
                                     int count, struct iobref *iobref);
-typedef int (*rpcsvc_vector_sizer) (rpcsvc_request_t *req, ssize_t *readsize,
-                                    int *newiob);
+typedef int (*rpcsvc_vector_sizer) (int state, ssize_t *readsize, char *addr);
 
 /* Every protocol actor will also need to specify the function the RPC layer
  * will use to serialize or encode the message into XDR format just before
@@ -410,6 +416,9 @@ rpcsvc_listener_destroy (rpcsvc_listener_t *listener);
 extern int
 rpcsvc_program_register_portmap (rpcsvc_program_t *newprog, uint32_t port);
 
+extern int
+rpcsvc_register_portmap_enabled (rpcsvc_t *svc);
+
 /* Inits the global RPC service data structures.
  * Called in main.
  */
@@ -542,4 +551,12 @@ int
 rpcsvc_transport_unix_options_build (dict_t **options, char *filepath);
 int
 rpcsvc_set_allow_insecure (rpcsvc_t *svc, dict_t *options);
+int
+rpcsvc_auth_array (rpcsvc_t *svc, char *volname, int *autharr, int arrlen);
+char *
+rpcsvc_volume_allowed (dict_t *options, char *volname);
+rpcsvc_vector_sizer
+rpcsvc_get_program_vector_sizer (rpcsvc_t *svc, uint32_t prognum,
+                                 uint32_t progver, uint32_t procnum);
+
 #endif
diff --git a/rpc/rpc-lib/src/xdr-common.h b/rpc/rpc-lib/src/xdr-common.h
index c69bfa834..0014e46ac 100644
--- a/rpc/rpc-lib/src/xdr-common.h
+++ b/rpc/rpc-lib/src/xdr-common.h
@@ -113,8 +113,6 @@ xdr_from_dump_req (struct iovec outmsg, void *rsp);
 ssize_t
 xdr_to_dump_rsp (struct iovec inmsg, void *args);
 
-#define XDR_BYTES_PER_UNIT      4
-
 /* Returns the address of the byte that follows the
  * last byte used for decoding the previous xdr component.
  * For eg, once the RPC call for NFS has been decoded, thie macro will return