features/bitrot-stub: Enhancement to versioning protocol

.. and potential bug fixes / memleak.

While assigning initial version to an object, both extended attributes
(namely, ongoing version and the default signing version) were persisted.
This is optimized to just persist the ongoing version along with safe
handling of xattr request(s) in it's absence. This is better than the
earlier approach as the two xattr sets were not atomic anyway (allowing
a request to sneak in between between two set operations). This also
allows to perform sanity checks on objects during lookup()/getxattr():
objects with missing ongoing version but presence of signature are
possible candidates of tampering (and catching implementation bugs).

There were couple of instances in the code where versioning xattrs
were incorrectly removed before in-memory versions were initialized,
which have been fixed with this patch. A memory leak in the IPC code
path is also fixed.

Change-Id: I01c690ccfe7156a883582275f40f79a7c10c0900
BUG: 1207054
Signed-off-by: Venky Shankar <vshankar@redhat.com>
Reviewed-on: http://review.gluster.org/10117
Reviewed-by: Raghavendra Bhat <raghavendra@redhat.com>
Tested-by: Gluster Build System <jenkins@build.gluster.com>
Reviewed-by: Vijay Bellur <vbellur@redhat.com>

3 files changed, 83 insertions, 105 deletions

diff --git a/xlators/features/bit-rot/src/stub/bit-rot-common.h b/xlators/features/bit-rot/src/stub/bit-rot-common.h
index 9e101523556..ae1f45f6d1d 100644
--- a/xlators/features/bit-rot/src/stub/bit-rot-common.h
+++ b/xlators/features/bit-rot/src/stub/bit-rot-common.h
@@ -37,13 +37,15 @@ typedef struct br_signature {
 #define BR_VXATTR_VERSION   (1 << 0)
 #define BR_VXATTR_SIGNATURE (1 << 1)
 
+#define BR_VXATTR_SIGN_MISSING (BR_VXATTR_SIGNATURE)
 #define BR_VXATTR_ALL_MISSING                           \
         (BR_VXATTR_VERSION | BR_VXATTR_SIGNATURE)
 
 typedef enum br_vxattr_state {
-        BR_VXATTR_STATUS_MISSING = 0,
-        BR_VXATTR_STATUS_PARTIAL = 1,
-        BR_VXATTR_STATUS_FULL    = 2,
+        BR_VXATTR_STATUS_FULL     = 0,
+        BR_VXATTR_STATUS_MISSING  = 1,
+        BR_VXATTR_STATUS_UNSIGNED = 2,
+        BR_VXATTR_STATUS_INVALID  = 3,
 } br_vxattr_status_t;
 
 static inline br_vxattr_status_t
@@ -66,11 +68,14 @@ br_version_xattr_state (dict_t *xattr,
         case 0:
                 status = BR_VXATTR_STATUS_FULL;
                 break;
+        case BR_VXATTR_SIGN_MISSING:
+                status = BR_VXATTR_STATUS_UNSIGNED;
+                break;
         case BR_VXATTR_ALL_MISSING:
                 status = BR_VXATTR_STATUS_MISSING;
                 break;
         default:
-                status = BR_VXATTR_STATUS_PARTIAL;
+                status = BR_VXATTR_STATUS_INVALID;
         }
 
         return status;
diff --git a/xlators/features/bit-rot/src/stub/bit-rot-stub.c b/xlators/features/bit-rot/src/stub/bit-rot-stub.c
index 1a268e16b93..f8c8f59c1f8 100644
--- a/xlators/features/bit-rot/src/stub/bit-rot-stub.c
+++ b/xlators/features/bit-rot/src/stub/bit-rot-stub.c
@@ -155,29 +155,6 @@ br_stub_dealloc_local (br_stub_local_t *ptr)
 }
 
 static inline int
-br_stub_prepare_default_request (xlator_t *this, dict_t *dict,
-                                 br_version_t *obuf, br_signature_t *sbuf)
-{
-        int32_t ret = 0;
-        size_t size = 0;
-        br_stub_private_t *priv = NULL;
-
-        priv = this->private;
-
-        /** Prepare ongoing version */
-        br_set_default_ongoingversion (obuf, priv->boot);
-        ret = dict_set_static_bin (dict, BITROT_CURRENT_VERSION_KEY,
-                                   (void *)obuf, sizeof (br_version_t));
-        if (ret)
-                return -1;
-
-        /** Prepare signature version */
-        br_set_default_signature (sbuf, &size);
-        return dict_set_static_bin (dict, BITROT_SIGNING_VERSION_KEY,
-                                    (void *)sbuf, size);
-}
-
-static inline int
 br_stub_prepare_version_request (xlator_t *this, dict_t *dict,
                                 br_version_t *obuf, unsigned long oversion)
 {
@@ -473,19 +450,19 @@ br_stub_perform_fullversioning (xlator_t *this, call_frame_t *frame,
         int32_t         ret      = -1;
         dict_t         *dict     = NULL;
         br_version_t   *obuf     = NULL;
-        br_signature_t *sbuf     = NULL;
         int             op_errno = 0;
 
         op_errno = ENOMEM;
         dict = dict_new ();
         if (!dict)
                 goto done;
-        ret = br_stub_alloc_versions (&obuf, &sbuf, 0);
+        ret = br_stub_alloc_versions (&obuf, NULL, 0);
         if (ret)
                 goto dealloc_dict;
 
         op_errno = EINVAL;
-        ret = br_stub_prepare_default_request (this, dict, obuf, sbuf);
+        ret = br_stub_prepare_version_request (this, dict, obuf,
+                                               BITROT_DEFAULT_CURRENT_VERSION);
         if (ret)
                 goto dealloc_versions;
 
@@ -693,14 +670,15 @@ br_stub_getxattr_cbk (call_frame_t *frame, void *cookie, xlator_t *this,
                 goto unwind;
 
         op_ret   = -1;
-        op_errno = EINVAL;
-
         status = br_version_xattr_state (xattr, &obuf, &sbuf);
-        if (status == BR_VXATTR_STATUS_PARTIAL)
+
+        op_errno = EINVAL;
+        if (status == BR_VXATTR_STATUS_INVALID)
                 goto delkeys;
 
         op_errno = ENODATA;
-        if (status == BR_VXATTR_STATUS_MISSING)
+        if ((status == BR_VXATTR_STATUS_MISSING)
+            || (status == BR_VXATTR_STATUS_UNSIGNED))
                 goto delkeys;
 
         signaturelen = strlen (sbuf->signature);
@@ -790,8 +768,10 @@ br_stub_getxattr (call_frame_t *frame, xlator_t *this,
                 goto wind;
         }
 
-        if (br_stub_is_internal_xattr (name))
-                goto wind;
+        if (br_stub_is_internal_xattr (name)) {
+                STACK_UNWIND (frame, -1, EINVAL, NULL, NULL);
+                return 0;
+        }
 
         /**
          * this special extended attribute is allowed only on root
@@ -835,8 +815,10 @@ br_stub_fgetxattr (call_frame_t *frame, xlator_t *this,
                 goto wind;
         }
 
-        if (br_stub_is_internal_xattr (name))
-                goto wind;
+        if (br_stub_is_internal_xattr (name)) {
+                STACK_UNWIND (frame, -1, EINVAL, NULL, NULL);
+                return 0;
+        }
 
         /**
          * this special extended attribute is allowed only on root
@@ -1016,16 +998,17 @@ br_stub_lookup_version (xlator_t *this,
         status = br_version_xattr_state (xattr, &obuf, &sbuf);
 
         /**
-         * stub does not know how to handle partial presence of version
-         * extended attributes, therefore, bail out in such cases.
+         * stub does not know how to handle presence of signature but not
+         * the object version, therefore, in such cases, bail out..
          */
-        if (status == BR_VXATTR_STATUS_PARTIAL) {
-                gf_log (this->name, GF_LOG_ERROR, "Partial version xattrs!.. "
-                        "bailing out [GFID: %s]", uuid_utoa (gfid));
+        if (status == BR_VXATTR_STATUS_INVALID) {
+                gf_log (this->name, GF_LOG_ERROR, "Invalid versioning xattrs. "
+                        "Bailing out [GFID: %s]", uuid_utoa (gfid));
                 return -1;
         }
 
-        version = (status == BR_VXATTR_STATUS_FULL)
+        version = ((status == BR_VXATTR_STATUS_FULL)
+                   || (status == BR_VXATTR_STATUS_UNSIGNED))
                         ? obuf->ongoingversion : BITROT_DEFAULT_CURRENT_VERSION;
         return br_stub_init_inode_versions (this, NULL,
                                             inode, version, _gf_true);
@@ -1054,18 +1037,17 @@ br_stub_readdirp_cbk (call_frame_t *frame, void *cookie, xlator_t *this,
                 if (!IA_ISREG (entry->d_stat.ia_type))
                         continue;
 
-                if (entry->dict) {
-                        br_stub_remove_vxattrs (entry->dict);
-                }
-
                 ret = br_stub_get_inode_ctx (this, entry->inode, &ctxaddr);
                 if (ret < 0)
                         ctxaddr = 0;
-                if (ctxaddr) /* already has the context */
+                if (ctxaddr) { /* already has the context */
+                        br_stub_remove_vxattrs (entry->dict);
                         continue;
+                }
 
                 ret = br_stub_lookup_version
                         (this, entry->inode->gfid, entry->inode, entry->dict);
+                br_stub_remove_vxattrs (entry->dict);
                 if (ret) {
                         /**
                          * there's no per-file granularity support in case of
@@ -1146,14 +1128,8 @@ br_stub_lookup_cbk (call_frame_t *frame, void *cookie,
                 goto unwind;
         if (!IA_ISREG (stbuf->ia_type))
                 goto unwind;
-
-        /**
-         * perform this before checking if we requested xattrs as this
-         * can happen during revalidate.
-         */
-        br_stub_remove_vxattrs (xattr);
         if (cookie != (void *) BR_STUB_REQUEST_COOKIE)
-                goto unwind;
+                goto delkey;
 
         ret = br_stub_lookup_version (this, stbuf->ia_gfid, inode, xattr);
         if (ret < 0) {
@@ -1161,6 +1137,8 @@ br_stub_lookup_cbk (call_frame_t *frame, void *cookie,
                 op_errno = EINVAL;
         }
 
+ delkey:
+        br_stub_remove_vxattrs (xattr);
  unwind:
         STACK_UNWIND_STRICT (lookup, frame,
                              op_ret, op_errno, inode, stbuf, xattr, postparent);
@@ -1198,6 +1176,11 @@ br_stub_lookup (call_frame_t *frame,
 
         xref = _gf_true;
 
+        /**
+         * Requesting both xattrs provides a way of sanity checking the
+         * object. Anomaly checking is done in cbk by examining absence
+         * of either or both xattrs.
+         */
         op_errno = EINVAL;
         ret = dict_set_uint32 (xdata, BITROT_CURRENT_VERSION_KEY, 0);
         if (ret)
@@ -1296,7 +1279,6 @@ br_stub_send_ipc_fop (xlator_t *this,
         op = GF_IPC_TARGET_CHANGELOG;
         STACK_WIND (frame, br_stub_noop, FIRST_CHILD (this),
                     FIRST_CHILD (this)->fops->ipc, op, xdata);
-        return;
 
  dealloc_dict:
         dict_unref (xdata);
diff --git a/xlators/storage/posix/src/posix-helpers.c b/xlators/storage/posix/src/posix-helpers.c
index 0638f845e9d..5b7236b193f 100644
--- a/xlators/storage/posix/src/posix-helpers.c
+++ b/xlators/storage/posix/src/posix-helpers.c
@@ -1856,72 +1856,63 @@ posix_fsyncer (void *d)
         }
 }
 
-/**
- * fetch on-disk ongoing version and object signature extended
- * attribute.
- */
-int32_t
-posix_get_objectsignature (char *real_path, dict_t *xattr)
+static int32_t
+posix_fetch_signature_xattr (char *real_path, const char *key, dict_t *xattr)
 {
-        int32_t  op_ret    = 0;
+        int32_t ret = 0;
         char    *memptr    = NULL;
         ssize_t  xattrsize = 0;
-        ssize_t  allocsize = 0;
 
-        op_ret = -EINVAL;
-        xattrsize = sys_lgetxattr (real_path,
-                                   BITROT_CURRENT_VERSION_KEY, NULL, 0);
-        if (xattrsize == -1)
-                goto error_return;
-        allocsize += xattrsize;
-
-        xattrsize = sys_lgetxattr (real_path,
-                                   BITROT_SIGNING_VERSION_KEY, NULL, 0);
+        xattrsize = sys_lgetxattr (real_path, key, NULL, 0);
+        if ((xattrsize == -1) && ((errno == ENOATTR) || (errno == ENODATA)))
+                return 0;
         if (xattrsize == -1)
                 goto error_return;
-        allocsize += xattrsize;
 
-        op_ret = -ENOMEM;
-        /* bulk alloc */
-        memptr = GF_CALLOC (allocsize + 2, sizeof (char), gf_posix_mt_char);
+        memptr = GF_CALLOC (xattrsize + 1, sizeof (char), gf_posix_mt_char);
         if (!memptr)
                 goto error_return;
+        ret = sys_lgetxattr (real_path, key, memptr, xattrsize);
+        if (ret == -1)
+                goto freemem;
 
-        op_ret = sys_lgetxattr (real_path, BITROT_CURRENT_VERSION_KEY,
-                                memptr, allocsize - xattrsize);
-        if (op_ret == -1) {
-                op_ret = -errno;
-                goto dealloc_mem;
-        }
+        ret = dict_set_dynptr (xattr, (char *)key, memptr, xattrsize);
+        if (ret)
+                goto freemem;
 
-        xattrsize = op_ret; /* save for correct _in_ memory pointing */
+        return 0;
 
-        op_ret = sys_lgetxattr (real_path, BITROT_SIGNING_VERSION_KEY,
-                                (memptr + op_ret + 1), allocsize - op_ret);
-        if (op_ret == -1) {
-                op_ret = -errno;
-                goto dealloc_mem;
-        }
+ freemem:
+        GF_FREE (memptr);
+ error_return:
+        return -1;
+}
 
-        /* this is a dynamic set */
-        op_ret = dict_set_dynptr (xattr, BITROT_CURRENT_VERSION_KEY,
-                                  memptr, allocsize);
-        if (op_ret < 0)
-                goto dealloc_mem;
+/**
+ * Fetch on-disk ongoing version and object signature extended attribute.
+ * Be generous to absence of xattrs (just *absence*, other errors are
+ * propagated up to the invoker), higher layer (br-stub) takes care of
+ * interpreting the xattrs for anomalies.
+ */
+int32_t
+posix_get_objectsignature (char *real_path, dict_t *xattr)
+{
+        int32_t ret = 0;
 
-        /* rest all should be static */
-        op_ret = dict_set_static_ptr (xattr, BITROT_SIGNING_VERSION_KEY,
-                                      memptr + xattrsize + 1);
-        if (op_ret < 0)
+        ret = posix_fetch_signature_xattr
+                             (real_path, BITROT_CURRENT_VERSION_KEY, xattr);
+        if (ret)
+                goto error_return;
+
+        ret = posix_fetch_signature_xattr
+                             (real_path, BITROT_SIGNING_VERSION_KEY, xattr);
+        if (ret)
                 goto delkey;
 
-        return allocsize;
+        return 0;
 
  delkey:
         dict_del (xattr, BITROT_CURRENT_VERSION_KEY);
- dealloc_mem:
-        GF_FREE (memptr);
  error_return:
-        return op_ret;
-
+        return -1;
 }