modules/kvs: normalize keys before use

Insert key normalization in
1) lookup_create(), which covers "get" and "watch" cases
2) in commit_link_dirent() which covers "put"
3) in kvs_unwatch()

Fixes #1180
Fixes #1173

src/modules/kvs/commit.c

@@ -321,16 +321,17 @@ static int commit_link_dirent (commit_t *c, int current_epoch,
                                json_t *rootdir, const char *key,
                                json_t *dirent, const char **missing_ref)
 {
-    char *cpy = strdup (key);
-    char *next, *name = cpy;
+    char *cpy = NULL;
+    char *next, *name;
     json_t *dir = rootdir;
     json_t *subdir = NULL, *dir_entry;
     int saved_errno, rc = -1;
 
-    if (!cpy) {
-        saved_errno = ENOMEM;
+    if (!(cpy = kvs_util_normalize_key (key, NULL))) {
+        saved_errno = errno;
         goto done;
     }
+    name = cpy;
 
     /* Special case root
      */

src/modules/kvs/kvs.c

@@ -923,14 +923,15 @@ static void watch_request_cb (flux_t *h, flux_msg_handler_t *w,
 }
 
 typedef struct {
-    const char *key;
+    char *key;
     char *sender;
 } unwatch_param_t;
 
 static bool unwatch_cmp (const flux_msg_t *msg, void *arg)
 {
     unwatch_param_t *p = arg;
     char *sender = NULL;
+    char *normkey = NULL;
     json_t *val;
     const char *key, *topic;
     int flags;
@@ -947,28 +948,38 @@ static bool unwatch_cmp (const flux_msg_t *msg, void *arg)
         goto done;
     if (strcmp (sender, p->sender) != 0)
         goto done;
-    if (strcmp (p->key, key) != 0)
+    if (!(normkey = kvs_util_normalize_key (key, NULL)))
+        goto done;
+    if (strcmp (p->key, normkey) != 0)
         goto done;
     match = true;
 done:
-    if (sender)
-        free (sender);
+    free (sender);
+    free (normkey);
     return match;
 }
 
 static void unwatch_request_cb (flux_t *h, flux_msg_handler_t *w,
                                 const flux_msg_t *msg, void *arg)
 {
     kvs_ctx_t *ctx = arg;
+    const char *key;
     unwatch_param_t p = { NULL, NULL };
-    int rc = -1;
+    int errnum = 0;
 
-    if (flux_request_unpack (msg, NULL, "{ s:s }", "key", &p.key) < 0) {
+    if (flux_request_unpack (msg, NULL, "{ s:s }", "key", &key) < 0) {
+        errnum = errno;
         flux_log_error (h, "%s: flux_request_unpack", __FUNCTION__);
         goto done;
     }
-    if (flux_msg_get_route_first (msg, &p.sender) < 0)
+    if (!(p.key = kvs_util_normalize_key (key, NULL))) {
+        errnum = errno;
         goto done;
+    }
+    if (flux_msg_get_route_first (msg, &p.sender) < 0) {
+        errnum = errno;
+        goto done;
+    }
     /* N.B. impossible for a watch to be on watchlist and cache waiter
      * at the same time (i.e. on watchlist means we're watching, if on
      * cache waiter we're not done processing towards being on the
@@ -977,19 +988,20 @@ static void unwatch_request_cb (flux_t *h, flux_msg_handler_t *w,
      * deal.  The current state is still maintained.
      */
     if (wait_destroy_msg (ctx->watchlist, unwatch_cmp, &p) < 0) {
+        errnum = errno;
         flux_log_error (h, "%s: wait_destroy_msg", __FUNCTION__);
         goto done;
     }
     if (cache_wait_destroy_msg (ctx->cache, unwatch_cmp, &p) < 0) {
+        errnum = errno;
         flux_log_error (h, "%s: cache_wait_destroy_msg", __FUNCTION__);
         goto done;
     }
-    rc = 0;
 done:
-    if (flux_respond (h, msg, rc < 0 ? errno : 0, NULL) < 0)
+    if (flux_respond (h, msg, errnum, NULL) < 0)
         flux_log_error (h, "%s: flux_respond", __FUNCTION__);
-    if (p.sender)
-        free (p.sender);
+    free (p.key);
+    free (p.sender);
 }
 
 struct finalize_data {

src/modules/kvs/lookup.c

@@ -41,6 +41,7 @@
 #include "src/common/libkvs/treeobj.h"
 
 #include "cache.h"
+#include "kvs_util.h"
 
 #include "lookup.h"
 
@@ -387,7 +388,7 @@ lookup_t *lookup_create (struct cache *cache,
         lh->root_ref_copy = NULL;
         lh->root_ref = lh->root_dir;
     }
-    if (!(lh->path = strdup (path))) {
+    if (!(lh->path = kvs_util_normalize_key (path, NULL))) {
         saved_errno = ENOMEM;
         goto cleanup;
     }