Merge pull request feast-dev#62 from farfetch-external/isacabe/KE-772…

…_redis_password Redis security (added password to redis)
Yanson · Jul 7, 2020 · 7f5a42e · 7f5a42e
2 parents e862b2e + 045e1ca
commit 7f5a42e
Show file tree

Hide file tree

Showing 18 changed files with 87 additions and 20 deletions.
diff --git a/infra/docker-compose/.env.sample b/infra/docker-compose/.env.sample
@@ -8,6 +8,8 @@ FEAST_CORE_IMAGE=gcr.io/kf-feast/feast-core
 FEAST_CORE_CONFIG=core.yml
 FEAST_CORE_GCP_SERVICE_ACCOUNT_KEY=placeholder.json
 
+FEAST_JUPYTER_IMAGE=jupyterimage
+
 # Feast Serving
 FEAST_SERVING_IMAGE=gcr.io/kf-feast/feast-serving
 # Feast Serving - Batch (BigQuery)
@@ -18,3 +20,8 @@ FEAST_ONLINE_SERVING_CONFIG=online-serving.yml
 
 # Jupyter
 FEAST_JUPYTER_GCP_SERVICE_ACCOUNT_KEY=placeholder.json
+
+# Redis
+REDIS_HOST=feast_redis_1
+# write here the password for your redis container (development environment).
+REDIS_PASS=PUT_YOUR_PASSWORD_HERE
diff --git a/infra/docker-compose/docker-compose.batch.yml b/infra/docker-compose/docker-compose.batch.yml
@@ -14,6 +14,8 @@ services:
     environment:
       GOOGLE_APPLICATION_CREDENTIALS: /etc/gcloud/service-accounts/key.json
       FEAST_JOB_STAGING_LOCATION: ${FEAST_BATCH_JOB_STAGING_LOCATION}
+      REDIS_HOST: ${REDIS_HOST}
+      REDIS_PASS: ${REDIS_PASS}
     command:
       - "java"
       - "-Xms1024m"

diff --git a/infra/docker-compose/docker-compose.online.yml b/infra/docker-compose/docker-compose.online.yml
@@ -13,6 +13,9 @@ services:
     ports:
       - 6566:6566
     restart: on-failure
+    environment:
+      REDIS_PASS: ${REDIS_PASS}
+      REDIS_HOST: ${REDIS_HOST}
     command:
       - java
       - -jar
@@ -21,5 +24,6 @@ services:
 
   redis:
     image: redis:5-alpine
+    command: redis-server --requirepass '${REDIS_PASS}'
     ports:
       - "6379:6379"
diff --git a/infra/docker-compose/serving/batch-serving.yml b/infra/docker-compose/serving/batch-serving.yml
@@ -16,7 +16,8 @@ feast:
         - name: "*"
           project: "*"
   job_store:
-    redis_host: redis
+    redis_host: ${REDIS_HOST}
+    redis_pass: '${REDIS_PASS}'
     redis_port: 6379
 
 grpc:

diff --git a/infra/docker-compose/serving/online-serving.yml b/infra/docker-compose/serving/online-serving.yml
@@ -5,7 +5,8 @@ feast:
     - name: online
       type: REDIS
       config:
-        host: redis
+        host: ${REDIS_HOST}
+        pass: '${REDIS_PASS}'
         port: 6379
       subscriptions:
         - name: "*"

diff --git a/infra/terraform/app/main.tf b/infra/terraform/app/main.tf
@@ -249,6 +249,7 @@ feast-online-serving:
         config:
           host: ${var.redis_hostname}
           port: ${var.redis_port}
+          pass: ${var.redis_pass}
         subscriptions:
         - name: "*"
           project: "*"
@@ -293,6 +294,7 @@ feast-batch-serving:
       job_store:
         redis_host: ${var.redis_hostname}
         redis_port: ${var.redis_port}
+        redis_pass: ${var.redis_pass}
 
 feast-core:
   enabled: false

diff --git a/infra/terraform/app/variables.tf b/infra/terraform/app/variables.tf
@@ -70,6 +70,10 @@ variable "redis_port" {
   description = "The port of the Azure Redis Cache instance."
   type        = number
 }
+variable "redis_pass" {
+  description = "The password of the Azure Redis Cache instance."
+  type        = string
+}
 variable "spark_job_jars" {
   description = "The local directory from which Spark job JARs are to be uploaded to DBFS."
   type        = string

diff --git a/infra/terraform/infra/outputs.tf b/infra/terraform/infra/outputs.tf
@@ -36,6 +36,11 @@ output "redis_hostname" {
 output "redis_port" {
   value = azurerm_redis_cache.redis_cluster.port
 }
+output "redis_pass" {
+  sensitive = true
+  value = azurerm_redis_cache.redis_cluster.primary_access_key
+}
+
 output "databricks_workspace_url" {
   value = "https://${azurerm_databricks_workspace.databricks.workspace_url}"
 }

diff --git a/protos/feast/core/Store.proto b/protos/feast/core/Store.proto
@@ -110,6 +110,7 @@ message Store {
     int32 initial_backoff_ms = 3;
     // Optional. Maximum total number of retries for connecting to Redis. Default to zero retries.
     int32 max_retries = 4;
+    string pass = 5;
   }
 
   message BigQueryConfig {

diff --git a/serving/src/main/java/feast/serving/config/FeastProperties.java b/serving/src/main/java/feast/serving/config/FeastProperties.java
@@ -441,6 +441,9 @@ public static class JobStoreProperties {
     /** Job Store Redis Host */
     private int redisPort;
 
+    /** Job Store Redis Pass */
+    private String redisPass;
+
     /**
      * Gets redis host.
      *
@@ -476,6 +479,24 @@ public int getRedisPort() {
     public void setRedisPort(int redisPort) {
       this.redisPort = redisPort;
     }
+
+    /**
+     * Gets redis pass.
+     *
+     * @return the redis pass
+     */
+    public String getRedisPass() {
+      return redisPass;
+    }
+
+    /**
+     * Sets redis pass.
+     *
+     * @param redisPass the redis pass
+     */
+    public void setRedisPass(String redisPass) {
+      this.redisPass = redisPass;
+    }
   }
 
   /** Trace metric collection properties */

diff --git a/serving/src/main/java/feast/serving/service/RedisBackedJobService.java b/serving/src/main/java/feast/serving/service/RedisBackedJobService.java
@@ -45,7 +45,7 @@ public class RedisBackedJobService implements JobService {
   public RedisBackedJobService(FeastProperties.JobStoreProperties jobStoreProperties) {
     RedisURI uri =
         RedisURI.create(jobStoreProperties.getRedisHost(), jobStoreProperties.getRedisPort());
-
+    uri.setPassword(jobStoreProperties.getRedisPass());
     this.syncCommand =
         RedisClient.create(DefaultClientResources.create(), uri)
             .connect(new ByteArrayCodec())

diff --git a/spark/spark-ingestion-job/src/main/java/feast/spark/ingestion/redis/SparkRedisSink.java b/spark/spark-ingestion-job/src/main/java/feast/spark/ingestion/redis/SparkRedisSink.java
@@ -68,12 +68,14 @@ public VoidFunction2<Dataset<byte[]>, Long> configure() {
     JavaSparkContext sc = new JavaSparkContext(spark.sparkContext());
     Broadcast<Write> broadcastedWriter = sc.broadcast(write);
 
-    return new RedisWriter(
-        broadcastedWriter,
+    RedisURI redisuri =
         new RedisURI(
             redisConfig.getHost(),
             redisConfig.getPort(),
-            java.time.Duration.ofMillis(DEFAULT_TIMEOUT)));
+            java.time.Duration.ofMillis(DEFAULT_TIMEOUT));
+    redisuri.setPassword(redisConfig.getPass());
+
+    return new RedisWriter(broadcastedWriter, redisuri);
   }
 
   @SuppressWarnings("serial")

diff --git a/spark/spark-ingestion-job/src/test/java/feast/test/TestUtil.java b/spark/spark-ingestion-job/src/test/java/feast/test/TestUtil.java
@@ -595,10 +595,13 @@ public static void validateRedis(
     Map<RedisKey, FeatureRow> expected = TestUtil.generateExpectedData(featureSet.getSpec(), input);
 
     LOGGER.info("Validating the actual values written to Redis ...");
-    RedisClient redisClient =
-        RedisClient.create(
-            new RedisURI(
-                redisConfig.getHost(), redisConfig.getPort(), java.time.Duration.ofMillis(2000)));
+
+    RedisURI redisuri =
+        new RedisURI(
+            redisConfig.getHost(), redisConfig.getPort(), java.time.Duration.ofMillis(2000));
+    redisuri.setPassword(redisConfig.getPass());
+
+    RedisClient redisClient = RedisClient.create(redisuri);
     StatefulRedisConnection<byte[], byte[]> connection = redisClient.connect(new ByteArrayCodec());
     RedisCommands<byte[], byte[]> sync = connection.sync();
 

diff --git a/...s/src/main/java/feast/storage/connectors/redis/retriever/RedisClusterOnlineRetriever.java b/...s/src/main/java/feast/storage/connectors/redis/retriever/RedisClusterOnlineRetriever.java
@@ -55,7 +55,12 @@ public static OnlineRetriever create(Map<String, String> config) {
             .map(
                 hostPort -> {
                   String[] hostPortSplit = hostPort.trim().split(":");
-                  return RedisURI.create(hostPortSplit[0], Integer.parseInt(hostPortSplit[1]));
+                  RedisURI redisuri =
+                      RedisURI.create(hostPortSplit[0], Integer.parseInt(hostPortSplit[1]));
+                  if (hostPortSplit.length == 3) {
+                    redisuri.setPassword(hostPortSplit[2]);
+                  }
+                  return redisuri;
                 })
             .collect(Collectors.toList());
 

diff --git a/...rs/redis/src/main/java/feast/storage/connectors/redis/retriever/RedisOnlineRetriever.java b/...rs/redis/src/main/java/feast/storage/connectors/redis/retriever/RedisOnlineRetriever.java
@@ -49,11 +49,11 @@ private RedisOnlineRetriever(StatefulRedisConnection<byte[], byte[]> connection)
   }
 
   public static OnlineRetriever create(Map<String, String> config) {
+    RedisURI redisuri = RedisURI.create(config.get("host"), Integer.parseInt(config.get("port")));
+    redisuri.setPassword(config.get("pass"));
 
     StatefulRedisConnection<byte[], byte[]> connection =
-        RedisClient.create(
-                RedisURI.create(config.get("host"), Integer.parseInt(config.get("port"))))
-            .connect(new ByteArrayCodec());
+        RedisClient.create(redisuri).connect(new ByteArrayCodec());
 
     return new RedisOnlineRetriever(connection);
   }

diff --git a/...edis/src/main/java/feast/storage/connectors/redis/writer/RedisClusterIngestionClient.java b/...edis/src/main/java/feast/storage/connectors/redis/writer/RedisClusterIngestionClient.java
@@ -47,7 +47,12 @@ public RedisClusterIngestionClient(StoreProto.Store.RedisClusterConfig redisClus
             .map(
                 hostPort -> {
                   String[] hostPortSplit = hostPort.trim().split(":");
-                  return RedisURI.create(hostPortSplit[0], Integer.parseInt(hostPortSplit[1]));
+                  RedisURI redisuri =
+                      RedisURI.create(hostPortSplit[0], Integer.parseInt(hostPortSplit[1]));
+                  if (hostPortSplit.length == 3) {
+                    redisuri.setPassword(hostPortSplit[2]);
+                  }
+                  return redisuri;
                 })
             .collect(Collectors.toList());
 

diff --git a/...onnectors/redis/src/main/java/feast/storage/connectors/redis/writer/RedisFeatureSink.java b/...onnectors/redis/src/main/java/feast/storage/connectors/redis/writer/RedisFeatureSink.java
@@ -81,9 +81,10 @@ public abstract static class Builder {
   @Override
   public void prepareWrite(FeatureSet featureSet) {
     if (getRedisConfig() != null) {
-      RedisClient redisClient =
-          RedisClient.create(
-              RedisURI.create(getRedisConfig().getHost(), getRedisConfig().getPort()));
+      RedisURI redisuri = RedisURI.create(getRedisConfig().getHost(), getRedisConfig().getPort());
+      redisuri.setPassword(getRedisConfig().getPass());
+      RedisClient redisClient = RedisClient.create(redisuri);
+
       try {
         redisClient.connect();
       } catch (RedisConnectionException e) {

diff --git a/...s/src/main/java/feast/storage/connectors/redis/writer/RedisStandaloneIngestionClient.java b/...s/src/main/java/feast/storage/connectors/redis/writer/RedisStandaloneIngestionClient.java
@@ -32,6 +32,7 @@
 
 public class RedisStandaloneIngestionClient implements RedisIngestionClient {
   private final String host;
+  private final String pass;
   private final int port;
   private final BackOffExecutor backOffExecutor;
   private RedisClient redisclient;
@@ -43,15 +44,17 @@ public class RedisStandaloneIngestionClient implements RedisIngestionClient {
   public RedisStandaloneIngestionClient(StoreProto.Store.RedisConfig redisConfig) {
     this.host = redisConfig.getHost();
     this.port = redisConfig.getPort();
+    this.pass = redisConfig.getPass();
     long backoffMs = redisConfig.getInitialBackoffMs() > 0 ? redisConfig.getInitialBackoffMs() : 1;
     this.backOffExecutor =
         new BackOffExecutor(redisConfig.getMaxRetries(), Duration.millis(backoffMs));
   }
 
   @Override
   public void setup() {
-    this.redisclient =
-        RedisClient.create(new RedisURI(host, port, java.time.Duration.ofMillis(DEFAULT_TIMEOUT)));
+    RedisURI redisuri = new RedisURI(host, port, java.time.Duration.ofMillis(DEFAULT_TIMEOUT));
+    redisuri.setPassword(pass);
+    this.redisclient = RedisClient.create(redisuri);
   }
 
   @Override