Redis AUTH support

This commit is based on
orange-cloudfoundry@7aa41a4
from @axelfauvel in #576 and tries to
close #46.

Unfortunatelly the initial commit was already so old and the dynomite code base already evolved,
that it was easier to not jump directly on this. Especically as there were some refactorings
requested.

Redis Datastore Authentification
If Dynomite is configured to require a password via config option `requirepass` the following
behaviour will be applied:

1. On Dynomite startup, the server authenticates with the backend itself
   by calling the datastore agnostic function g_datastore_auth.
2. The corresponding Redis response will be handeled in g_is_authenticated.
   Dynomite will exit if authentification to the datatstore was not successful.
3. Each newly created client connection will require authentification.
4. Clients can authentificate itself by issue the AUTH command against dynomite.
5. Dynomite will check the password and simulate an AUTH response.
6. If AUTH was successful, the auth_required flag on the connection is reset and
   the client can process further commands through this connection.

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).
++ **requirepass**: The password for authentication. Clients will need to authenticate to dynomite with the same password as dynomites authenticates itself to the datastore. (note that this is not supported for memcache yet)
 + **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
+  requirepass: helloworld

src/dyn_client.c

@@ -358,6 +358,16 @@ static bool req_filter(struct context *ctx, struct conn *conn,
     return true;
   }
 
+  /*
+   * Handle "AUTH requirepass\r\n"
+   */
+  if (conn->auth_required) {
+    if (g_authenticate_conn(ctx, conn, req)) {
+      conn->auth_required = 0;
+    };
+    return true;
+  }
+
   /*
    * Handle "quit\r\n", which is the protocol way of doing a
    * passive close

src/dyn_conf.c

@@ -224,6 +224,7 @@ static rstatus_t 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->requirepass);
   cp->dyn_listen.port = 0;
   memset(&cp->dyn_listen.info, 0, sizeof(cp->dyn_listen.info));
   cp->dyn_listen.valid = 0;
@@ -305,6 +306,7 @@ static void conf_pool_deinit(struct conf_pool *cp) {
   string_deinit(&cp->stats_listen.name);
   string_deinit(&cp->dc);
   string_deinit(&cp->env);
+  string_deinit(&cp->requirepass);
 
   if (array_n(&cp->dyn_seeds) != 0) array_deinit(&cp->dyn_seeds);
 
@@ -1162,6 +1164,9 @@ static struct command conf_commands[] = {
 
     {string("env"), conf_set_string, offsetof(struct conf_pool, env)},
 
+    {string("requirepass"), conf_set_string,
+     offsetof(struct conf_pool, requirepass)},
+
     {string("conn_msg_rate"), conf_set_num,
      offsetof(struct conf_pool, conn_msg_rate)},
 

src/dyn_conf.h

@@ -106,7 +106,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 */
-
+  
   /* none | datacenter | rack | all in order of increasing number of
    * connections. (default is datacenter) */
   struct string secure_server_option;
@@ -119,6 +119,7 @@ struct conf_pool {
                              reconciliation */
   struct string dc;       /* this node's dc */
   struct string env;      /* AWS, Google, network, ... */
+  struct string requirepass;
   uint32_t conn_msg_rate; /* conn msg per sec */
   bool enable_gossip;     /* enable/disable gossip */
   size_t mbuf_size;       /* mbuf chunk size */

src/dyn_connection.c

@@ -170,6 +170,12 @@ struct conn *conn_get(void *owner, func_conn_init_t func_conn_init) {
     return NULL;
   }
 
+  struct server_pool *pool = (struct server_pool *)owner;
+  if (pool->requirepass.len > 0) {
+    // Password configured, authentification required
+    conn->auth_required = 1;
+  }
+
   /* connection handles the data store messages (redis, memcached or other) */
 
   func_conn_init(conn);

src/dyn_connection.h

@@ -133,6 +133,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 auth_required : 1;    /* auth required? */
   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

@@ -245,6 +245,7 @@ struct server_pool {
   msec_t g_interval; /* gossip interval */
   struct string dc;  /* server's dc */
   struct string env; /* aws, network, etc */
+  struct string requirepass;
   /* none | datacenter | rack | all in order of increasing number of
    * connections. (default is datacenter) */
   secure_server_option_t secure_server_option;

src/dyn_message.c

@@ -162,6 +162,9 @@ func_msg_rewrite_t g_rewrite_query;     /* rewrite query in a msg if necessary *
 func_msg_rewrite_t g_rewrite_query_with_timestamp_md;
 func_msg_repair_t g_make_repair_query;  /* Send a repair msg. */
 func_clear_repair_md_t g_clear_repair_md_for_key; /* Clear repair metadata for a key */
+func_datatstore_auth_t g_datatstore_auth;     /* authenticate in datastore */
+func_is_authenticated_t g_is_authenticated;   /* handle auth response from datastore */
+func_authenticate_conn_t g_authenticate_conn; /* authenticate client connection */
 
 #define DEFINE_ACTION(_name) string(#_name),
 static struct string msg_type_strings[] = {MSG_TYPE_CODEC(DEFINE_ACTION)
@@ -203,6 +206,9 @@ void set_datastore_ops(void) {
       g_rewrite_query_with_timestamp_md = redis_rewrite_query_with_timestamp_md;
       g_make_repair_query = redis_make_repair_query;
       g_clear_repair_md_for_key = redis_clear_repair_md_for_key;
+      g_datatstore_auth = redis_datatstore_auth;
+      g_is_authenticated = redis_is_authenticated;
+      g_authenticate_conn = redis_authenticate_conn;
       break;
     case DATA_MEMCACHE:
       g_pre_coalesce = memcache_pre_coalesce;
@@ -215,6 +221,9 @@ void set_datastore_ops(void) {
       g_rewrite_query_with_timestamp_md = memcache_rewrite_query_with_timestamp_md;
       g_make_repair_query = memcache_make_repair_query;
       g_clear_repair_md_for_key = memcache_clear_repair_md_for_key;
+      g_datatstore_auth = memcache_datatstore_auth;
+      g_is_authenticated = memcache_is_authenticated;
+      g_authenticate_conn = memcache_authenticate_conn;
       break;
     default:
       return;

src/dyn_message.h

@@ -190,7 +190,7 @@
   ACTION(REQ_REDIS_JSONARRLEN)                                                 \
   ACTION(REQ_REDIS_JSONOBJKEYS)                                                \
   ACTION(REQ_REDIS_JSONOBJLEN)                                                 \
-  /* 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)                                                    \
@@ -246,6 +246,10 @@ typedef rstatus_t (*func_msg_repair_t)(struct context *ctx, struct response_mgr
     struct msg **new_msg_ptr);
 typedef rstatus_t (*func_clear_repair_md_t)(struct context *ctx, struct msg *req,
     struct msg **new_msg_ptr);
+typedef void (*func_datatstore_auth_t)(struct context *ctx, struct conn *conn);
+typedef bool (*func_is_authenticated_t)(struct msg *rsp);
+typedef bool (*func_authenticate_conn_t)(struct context *ctx, struct conn *conn, 
+    struct msg *req);
 typedef void (*func_init_datastore_t)();
 
 extern func_msg_coalesce_t g_pre_coalesce;  /* message pre-coalesce */
@@ -260,6 +264,9 @@ extern func_msg_rewrite_t
     g_rewrite_query_with_timestamp_md;
 extern func_msg_repair_t g_make_repair_query; /* Create a repair msg. */
 extern func_clear_repair_md_t g_clear_repair_md_for_key;
+extern func_datatstore_auth_t g_datatstore_auth;
+extern func_is_authenticated_t g_is_authenticated;
+extern func_authenticate_conn_t g_authenticate_conn;
 
 void set_datastore_ops(void);
 

src/dyn_server.c

@@ -295,6 +295,7 @@ static void server_connected(struct context *ctx, struct conn *conn) {
   conn_pool_connected(conn->conn_pool, conn);
 
   log_notice("%s connected ", print_obj(conn));
+  g_datatstore_auth(ctx, conn);
 }
 
 static void server_ok(struct context *ctx, struct conn *conn) {
@@ -398,6 +399,7 @@ rstatus_t server_pool_init(struct server_pool *sp, struct conf_pool *cp,
   /* sp->continuum = NULL; */
   sp->next_rebuild = 0ULL;
 
+  sp->requirepass = cp->requirepass;
   sp->name = cp->name;
   sp->proxy_endpoint.pname = cp->listen.pname;
   sp->proxy_endpoint.port = (uint16_t)cp->listen.port;
@@ -787,6 +789,22 @@ static void server_rsp_forward(struct context *ctx, struct conn *s_conn,
   log_info("%s %s RECEIVED %s", print_obj(c_conn), print_obj(req),
            print_obj(rsp));
 
+  /* 
+   * If backend requires authentification, it will respond with OK 
+   * after g_authenticate was succesful.
+   * In case auth has failed - error out
+   */ 
+  if (c_conn->type == CONN_SERVER) {
+    bool authenticated = g_is_authenticated(rsp);
+    if (authenticated) {
+      log_debug(LOG_INFO, "AUTH requirepass OK");
+      return;
+    } else {
+      log_debug(LOG_ERR, "%s", rsp->mhdr.stqh_first->start);
+      ASSERT(authenticated == true);
+    }
+  }
+
   ASSERT((c_conn->type == CONN_CLIENT) ||
          (c_conn->type == CONN_DNODE_PEER_CLIENT));
 

src/proto/Makefile.am

@@ -10,6 +10,7 @@ noinst_LIBRARIES = libproto.a
 noinst_HEADERS = dyn_proto.h
 
 libproto_a_SOURCES =			\
-	dyn_memcache.c			\
+	dyn_memcache.c			    \
 	dyn_redis.c					\
-	dyn_redis_repair.c
+	dyn_redis_repair.c			\
+	dyn_redis_auth.c

src/proto/dyn_memcache.c

@@ -1632,3 +1632,18 @@ rstatus_t memcache_clear_repair_md_for_key(struct context *ctx, struct msg *req,
     struct msg **new_msg_ptr) {
   return DN_OK;
 }
+
+/*
+ * Placeholder functions for authentification against backend.
+ * Not supported today.
+ */
+void memcache_datatstore_auth(struct context *ctx, struct conn *conn) {
+}
+
+bool memcache_is_authenticated(struct msg *rsp) {
+  return false;
+}
+
+bool memcache_authenticate_conn(struct context *ctx, struct conn *conn, struct msg *req) {
+  return false;
+}

src/proto/dyn_proto.h

@@ -55,6 +55,9 @@ rstatus_t memcache_make_repair_query(struct context *ctx, struct response_mgr *r
     struct msg **new_msg_ptr);
 rstatus_t memcache_clear_repair_md_for_key(struct context *ctx, struct msg *req,
     struct msg **new_msg_ptr);
+void memcache_datatstore_auth(struct context *ctx, struct conn *conn);
+bool memcache_is_authenticated(struct msg *rsp);
+bool memcache_authenticate_conn(struct context *ctx, struct conn *conn, struct msg *req);
 
 void redis_parse_req(struct msg *r, struct context *ctx);
 void redis_parse_rsp(struct msg *r, struct context *ctx);
@@ -74,5 +77,7 @@ rstatus_t redis_make_repair_query(struct context *ctx, struct response_mgr *rspm
     struct msg **new_msg_ptr);
 rstatus_t redis_clear_repair_md_for_key(struct context *ctx, struct msg *req,
     struct msg **new_msg_ptr);
-
+void redis_datatstore_auth(struct context *ctx, struct conn *conn);
+bool redis_is_authenticated(struct msg *rsp);
+bool redis_authenticate_conn(struct context *ctx, struct conn *conn, struct msg *req);
 #endif

src/proto/dyn_redis.c

@@ -91,6 +91,7 @@ static bool redis_arg0(struct msg *r) {
 
     case MSG_REQ_REDIS_KEYS:
     case MSG_REQ_REDIS_PFCOUNT:
+    case MSG_REQ_REDIS_AUTH:
       return true;
 
     default:
@@ -721,6 +722,12 @@ void redis_parse_req(struct msg *r, struct context *ctx) {
             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;