Support REDIS AUTH command

README.md

@@ -74,6 +74,7 @@ Dynomite can be configured through a YAML file specified by the -c or --conf-fil
 + **timeout**: The timeout value in msec that we wait for to establish a connection to the server or receive a response from a server. By default, we wait indefinitely.
 + **preconnect**: A boolean value that controls if dynomite should preconnect to all the servers in this pool on process start. Defaults to false.
 + **data_store**: An integer value that controls if a server pool speaks redis (0) or memcached (1) or other protocol. Defaults to redis (0).
++ **redis_requirepass**: The password of redis.
 + **auto_eject_hosts**: A boolean value that controls if server should be ejected temporarily when it fails consecutively server_failure_limit times. See [liveness recommendations](notes/recommendation.md#liveness) for information. Defaults to false.
 + **server_retry_timeout**: The timeout value in msec to wait for before retrying on a temporarily ejected server, when auto_eject_host is set to true. Defaults to 30000 msec.
 + **server_failure_limit**: The number of consecutive failures on a server that would lead to it being temporarily ejected when auto_eject_host is set to true. Defaults to 2.

conf/redis_single_with_password.yml

@@ -0,0 +1,10 @@
+dyn_o_mite:
+  dyn_listen: 0.0.0.0:8101
+  data_store: 0
+  listen: 0.0.0.0:8102
+  dyn_seed_provider: simple_provider
+  servers:
+  - 127.0.0.1:22122:1
+  tokens: 437425602
+  stats_listen: 0.0.0.0:22222
+  redis_requirepass: helloworld

src/dyn_client.c

@@ -355,9 +355,117 @@ req_recv_next(struct context *ctx, struct conn *conn, bool alloc)
     return req;
 }
 
+static struct mbuf *
+get_mbuf(struct msg *msg)
+{
+    struct mbuf *mbuf;
+    mbuf = STAILQ_LAST(&msg->mhdr, mbuf, next);
+    if (mbuf == NULL || mbuf_full(mbuf)) {
+        mbuf = mbuf_get();
+        if (mbuf == NULL) {
+            return NULL;
+        }
+        mbuf_insert(&msg->mhdr, mbuf);
+        msg->pos = mbuf->pos;
+    }
+
+    return mbuf;
+}
+
+static void
+auth_reply(struct context *ctx, struct conn *conn, struct msg *smsg, const char *usr_msg)
+{
+    int n;
+    struct mbuf *mbuf;
+    struct msg *msg = msg_get(conn, false, __FUNCTION__);
+    if (msg == NULL) {
+        return;
+    }
+
+    mbuf = get_mbuf(msg);
+    if (mbuf == NULL) {
+        msg_put(msg);
+        return;
+    }
+
+    smsg->peer = msg;
+    smsg->selected_rsp = msg;
+    msg->peer = smsg;
+
+    n = (int)strlen(usr_msg);
+    memcpy(mbuf->last, usr_msg, (size_t)n);
+    mbuf->last += n;
+    msg->mlen += (uint32_t)n;
+    msg->done = 1;
+
+    conn_event_add_out(conn);
+    TAILQ_INSERT_TAIL(&conn->omsg_q, smsg, c_tqe);
+}
+
+static int
+get_password(unsigned char *string, unsigned int len, char *passwd, uint32_t *passwd_len)
+{
+    char *p;
+    char *pos;
+    char buff[128];
+    int cmdlen, nargc;
+    size_t l;
+
+    p = (char *)string;
+    if (p[0] != '*') {
+        return -1;
+    }
+
+    /* deal with nargc */
+    pos = strstr(p + 1, CRLF);
+    if (!pos) return -1;
+    l = pos - (p + 1);
+    memcpy(buff, p + 1, l);
+    buff[l] = '\0';
+    nargc = atoi(buff);
+    if (nargc != 2) return -1;
+
+    /* deal with cmd */
+    p = pos + 2;
+    if (*p != '$') return -1;
+    pos = strstr(p + 1, CRLF);
+    if (!pos) return -1;
+    l = pos - (p + 1);
+    memcpy(buff, p + 1, l);
+    buff[l] = '\0';
+    cmdlen = atoi(buff);
+    p = pos + 2;
+    pos = strstr(p, CRLF);
+    if (!pos) return -1;
+    l = pos - p;
+    if (l != cmdlen) return -1;
+    memcpy(buff, p, l);
+    buff[l] = '\0';
+    if (strcasecmp("AUTH", buff) != 0) return -1;
+
+    /* password */
+    p = pos + 2;
+    if (*p != '$') return -1;
+    pos = strstr(p + 1, CRLF);
+    if (!pos) return -1;
+    l = pos - (p + 1);
+    memcpy(buff, p + 1, l);
+    buff[l] = '\0';
+    *passwd_len = atoi(buff);
+    p = pos + 2;
+    pos = strstr(p, CRLF);
+    if (!pos) return -1;
+    l = pos - p;
+    memcpy(passwd, p, l);
+    passwd[l] = '\0';
+    return 0;
+}
+
 static bool
 req_filter(struct context *ctx, struct conn *conn, struct msg *req)
 {
+    struct server_pool *pool;
+
     ASSERT(conn->type == CONN_CLIENT);
 
     if (msg_empty(req)) {
@@ -367,6 +475,33 @@ req_filter(struct context *ctx, struct conn *conn, struct msg *req)
         return true;
     }
 
+    /*
+     * Handle "AUTH requirepass\r\n"
+     */
+    if (conn->authenticated) {
+        if (req->type == MSG_REQ_REDIS_AUTH) {
+            pool = &ctx->pool;
+            char passwd[128];
+            uint32_t passwd_len;
+            if (get_password(req->mhdr.stqh_first->start,
+                             req->mlen, passwd, &passwd_len) != 0) {
+                auth_reply(ctx, conn, req, "-Unknown cmd");
+                return true;
+            }
+            if (passwd_len == pool->redis_requirepass.len) {
+                if (memcmp(pool->redis_requirepass.data, passwd, passwd_len) == 0) {
+                    auth_reply(ctx, conn, req, "+OK\r\n");
+                    conn->authenticated = 0;
+                    return true;
+                }
+            }
+            auth_reply(ctx, conn, req, "-ERR invalid password\r\n");
+        } else {
+            auth_reply(ctx, conn, req, "-NOAUTH Authentication required\r\n");
+        }
+        return true;
+    }
+
     /*
      * Handle "quit\r\n", which is the protocol way of doing a
      * passive close

src/dyn_conf.c

@@ -232,6 +232,7 @@ conf_pool_init(struct conf_pool *cp, struct string *name)
     string_init(&cp->stats_listen.name);
     string_init(&cp->dc);
     string_init(&cp->env);
+    string_init(&cp->redis_requirepass);
     cp->dyn_listen.port = 0;
     memset(&cp->dyn_listen.info, 0, sizeof(cp->dyn_listen.info));
     cp->dyn_listen.valid = 0;
@@ -313,6 +314,9 @@ conf_pool_deinit(struct conf_pool *cp)
     string_deinit(&cp->stats_listen.name);
     string_deinit(&cp->dc);
     string_deinit(&cp->env);
+    if (cp->redis_requirepass.len > 0) {
+        string_deinit(&cp->redis_requirepass);
+    }
 
     if (array_n(&cp->dyn_seeds) != 0)
        array_deinit(&cp->dyn_seeds);
@@ -1159,6 +1163,10 @@ static struct command conf_commands[] = {
       conf_add_server,
       offsetof(struct conf_pool, conf_datastore) },
 
+    { string("redis_requirepass"),
+      conf_set_string,
+      offsetof(struct conf_pool, redis_requirepass) },
+
     { string("dyn_read_timeout"),
       conf_set_num,
       offsetof(struct conf_pool, dyn_read_timeout) },

src/dyn_conf.h

@@ -100,6 +100,7 @@ struct conf_pool {
     struct string      rack;                  /* this node's logical rack */
     struct array       tokens;                /* this node's token: dyn_token array */
     msec_t             gos_interval;          /* wake up interval in ms */
+    struct string      redis_requirepass;
 
     /* none | datacenter | rack | all in order of increasing number of connections. (default is datacenter) */
     struct string      secure_server_option;

src/dyn_connection.c

@@ -189,12 +189,19 @@ conn_event_del_out(struct conn *conn)
 struct conn *
 conn_get(void *owner, func_conn_init_t func_conn_init)
 {
+    int auth = 0;
     struct conn *conn;
 
+    struct server_pool *pool = (struct server_pool *)owner;
+    if (pool->redis_requirepass.len > 0) {
+        auth = 1;
+    }
+
     conn = _conn_get();
     if (conn == NULL) {
         return NULL;
     }
+    conn->authenticated = auth;
 
     /* connection handles the data store messages (redis, memcached or other) */
 

src/dyn_connection.h

@@ -126,6 +126,7 @@ struct conn {
     unsigned           eof:1;         /* eof? aka passive close? */
     unsigned           waiting_to_unref:1; /* eof? aka passive close? */
     unsigned           done:1;        /* done? aka close? */
+    unsigned           authenticated:1;      /* redis auth? */
     unsigned           dyn_mode:1;           /* is a dyn connection? */
     unsigned           dnode_secured:1;      /* is a secured connection? */
     unsigned           crypto_key_sent:1; /* crypto state */

src/dyn_core.h

@@ -299,6 +299,7 @@ struct server_pool {
     unsigned           preconnect:1;         /* preconnect? */
 
     /* dynomite */
+    struct string      redis_requirepass;
     struct string      seed_provider;
     struct array       peers;
     struct conn        *d_conn;              /* dnode connection (listener) */

src/dyn_message.h

@@ -196,7 +196,7 @@ typedef enum msg_parse_result {
     ACTION( REQ_REDIS_ZSCAN)                                                                        \
     ACTION( REQ_REDIS_EVAL )                   /* redis requests - eval */                          \
     ACTION( REQ_REDIS_EVALSHA )                                                                     \
-    /* ACTION( REQ_REDIS_AUTH) */                                                                   \
+    ACTION( REQ_REDIS_AUTH)                                                                         \
     /* ACTION( REQ_REDIS_SELECT)*/             /* only during init */                               \
     ACTION( REQ_REDIS_PFADD )                  /* redis requests - hyperloglog */                   \
     ACTION( REQ_REDIS_PFCOUNT )                                                                     \

src/dyn_server.c

@@ -296,17 +296,53 @@ server_close(struct context *ctx, struct conn *conn)
 	server_failure(ctx, datastore);
 }
 
+static void
+redis_auth(struct context *ctx, struct conn *conn)
+{
+#define REDIS_AUTH_CMD "*2\r\n$4\r\nAUTH\r\n"
+    struct server_pool *pool;
+    struct msg *msg;
+    struct mbuf *mbuf;
+    int n;
+    char auth[1024];
+
+    pool = &ctx->pool;
+    if (pool->redis_requirepass.len > 0) {
+        msg = msg_get(conn, true, __FUNCTION__);
+        if (msg == NULL) {
+            return;
+        }
+
+        mbuf = mbuf_get();
+        if (mbuf == NULL) {
+            return;
+        }
+
+        n = snprintf(auth, sizeof(auth), REDIS_AUTH_CMD "$%d\r\n%s\r\n",
+                     pool->redis_requirepass.len, pool->redis_requirepass.data);
+
+        memcpy(mbuf->last, auth, (size_t)n);
+        mbuf->last += n;
+        mbuf_insert(&msg->mhdr, mbuf);
+        msg->pos = mbuf->pos;
+        msg->mlen = (uint32_t)n;
+        TAILQ_INSERT_TAIL(&conn->imsg_q, msg, s_tqe);
+    }
+#undef REDIS_AUTH_CMD
+}
+
 static void
 server_connected(struct context *ctx, struct conn *conn)
 {
     ASSERT(conn->type == CONN_SERVER);
-	ASSERT(conn->connecting && !conn->connected);
+    ASSERT(conn->connecting && !conn->connected);
 
-	conn->connecting = 0;
-	conn->connected = 1;
+    conn->connecting = 0;
+    conn->connected = 1;
     conn_pool_connected(conn->conn_pool, conn);
 
     log_notice("%s connected ", print_obj(conn));
+    redis_auth(ctx, conn);
 }
 
 static void
@@ -419,6 +455,7 @@ server_pool_init(struct server_pool *sp, struct conf_pool *cp, struct context *c
     /* sp->continuum = NULL; */
     sp->next_rebuild = 0ULL;
 
+    sp->redis_requirepass = cp->redis_requirepass;
     sp->name = cp->name;
     sp->proxy_endpoint.pname = cp->listen.pname;
     sp->proxy_endpoint.port = (uint16_t)cp->listen.port;
@@ -832,7 +869,19 @@ server_rsp_forward(struct context *ctx, struct conn *s_conn, struct msg *rsp)
 
     c_conn = req->owner;
     log_info("%s %s RECEIVED %s", print_obj(c_conn), print_obj(req), print_obj(rsp));
-
+
+    if (c_conn->type == CONN_SERVER) {
+        static const char *AUTH_OK = "+OK\r\n";
+        int ret = memcmp(AUTH_OK, rsp->mhdr.stqh_first->start, rsp->mlen);
+        if (ret == 0) {
+            log_debug(LOG_INFO, "AUTH requirepass OK");
+	    return;
+        } else {
+            log_debug(LOG_ERR, "%s", rsp->mhdr.stqh_first->start);
+            ASSERT(ret == 0);
+        }
+    }
+
     ASSERT((c_conn->type == CONN_CLIENT) ||
            (c_conn->type == CONN_DNODE_PEER_CLIENT));
 

src/dyn_test.c

@@ -402,7 +402,7 @@ aes_test(void)
     unsigned char msg[MAX_MSG_LEN+1];
     print_banner("AES");
     unsigned char* aes_key = generate_aes_key();
-    SCOPED_CHARPTR(aes_key_print) = base64_encode(aes_key, AES_KEYLEN);
+    SCOPED_CHARPTR(aes_key_print) = base64_encode(aes_key, strlen((char*)aes_key));
     loga("aesKey is '%s'", aes_key_print);
 
     size_t i=0;

src/proto/dyn_redis.c

@@ -91,6 +91,7 @@ redis_arg0(struct msg *r)
 
     case MSG_REQ_REDIS_KEYS:
     case MSG_REQ_REDIS_PFCOUNT:
+    case MSG_REQ_REDIS_AUTH:
         return true;
 
     default:
@@ -591,6 +592,12 @@ redis_parse_req(struct msg *r, const struct string *hash_tag)
                 break;
 
             case 4:
+                if (str4icmp(m, 'a', 'u', 't', 'h')) {
+                    r->type = MSG_REQ_REDIS_AUTH;
+                    r->is_read = 0;
+                    break;
+                }
+
                 if (str4icmp(m, 'p', 't', 't', 'l')) {
                     r->type = MSG_REQ_REDIS_PTTL;
                     r->is_read = 1;