Add password authentication to Redis ports (#2952)

* Implement Redis authentication

* Throw exception for legacy Ray

* Add test

* Formatting

* Fix bugs in CLI

* Fix bugs in Raylet

* Move default password to constants.h

* Use pytest.fixture

* Fix bug

* Authenticate using formatted strings

* Add missing passwords

* Add test

* Improve authentication of async contexts

* Disable Redis authentication for credis

* Update test for credis

* Fix rebase artifacts

* Fix formatting

* Add workaround for issue #3045

* Increase timeout for test

* Improve C++ readability

* Fixes for CLI

* Add security docs

* Address comments

* Address comments

* Adress comments

* Use ray.get

* Fix lint

python/ray/scripts/scripts.py

@@ -89,6 +89,11 @@ def cli(logging_level, logging_format):
     type=int,
     help=("If provided, attempt to configure Redis with this "
           "maximum number of clients."))
+@click.option(
+    "--redis-password",
+    required=False,
+    type=str,
+    help="If provided, secure Redis ports with this password")
 @click.option(
     "--redis-shard-ports",
     required=False,
@@ -190,10 +195,11 @@ def cli(logging_level, logging_format):
     default=None,
     help="manually specify the root temporary dir of the Ray process")
 def start(node_ip_address, redis_address, redis_port, num_redis_shards,
-          redis_max_clients, redis_shard_ports, object_manager_port,
-          object_store_memory, num_workers, num_cpus, num_gpus, resources,
-          head, no_ui, block, plasma_directory, huge_pages, autoscaling_config,
-          use_raylet, no_redirect_worker_output, no_redirect_output,
+          redis_max_clients, redis_password, redis_shard_ports,
+          object_manager_port, object_store_memory, num_workers, num_cpus,
+          num_gpus, resources, head, no_ui, block, plasma_directory,
+          huge_pages, autoscaling_config, use_raylet,
+          no_redirect_worker_output, no_redirect_output,
           plasma_store_socket_name, raylet_socket_name, temp_dir):
     # Convert hostnames to numerical IP address.
     if node_ip_address is not None:
@@ -205,6 +211,11 @@ def start(node_ip_address, redis_address, redis_port, num_redis_shards,
         # This environment variable is used in our testing setup.
         logger.info("Detected environment variable 'RAY_USE_XRAY'.")
         use_raylet = True
+    if not use_raylet and redis_password is not None:
+        raise Exception("Setting the 'redis-password' argument is not "
+                        "supported in legacy Ray. To run Ray with "
+                        "password-protected Redis ports, pass "
+                        "the '--use-raylet' flag.")
 
     try:
         resources = json.loads(resources)
@@ -269,6 +280,7 @@ def start(node_ip_address, redis_address, redis_port, num_redis_shards,
             num_redis_shards=num_redis_shards,
             redis_max_clients=redis_max_clients,
             redis_protected_mode=False,
+            redis_password=redis_password,
             include_webui=(not no_ui),
             plasma_directory=plasma_directory,
             huge_pages=huge_pages,
@@ -281,16 +293,20 @@ def start(node_ip_address, redis_address, redis_port, num_redis_shards,
         logger.info(
             "\nStarted Ray on this node. You can add additional nodes to "
             "the cluster by calling\n\n"
-            "    ray start --redis-address {}\n\n"
+            "    ray start --redis-address {}{}{}\n\n"
             "from the node you wish to add. You can connect a driver to the "
             "cluster from Python by running\n\n"
             "    import ray\n"
-            "    ray.init(redis_address=\"{}\")\n\n"
+            "    ray.init(redis_address=\"{}{}{}\")\n\n"
             "If you have trouble connecting from a different machine, check "
             "that your firewall is configured properly. If you wish to "
             "terminate the processes that have been started, run\n\n"
-            "    ray stop".format(address_info["redis_address"],
-                                  address_info["redis_address"]))
+            "    ray stop".format(
+                address_info["redis_address"], " --redis-password "
+                if redis_password else "", redis_password if redis_password
+                else "", address_info["redis_address"], "\", redis_password=\""
+                if redis_password else "", redis_password
+                if redis_password else ""))
     else:
         # Start Ray on a non-head node.
         if redis_port is not None:
@@ -315,10 +331,12 @@ def start(node_ip_address, redis_address, redis_port, num_redis_shards,
 
         # Wait for the Redis server to be started. And throw an exception if we
         # can't connect to it.
-        services.wait_for_redis_to_start(redis_ip_address, int(redis_port))
+        services.wait_for_redis_to_start(
+            redis_ip_address, int(redis_port), password=redis_password)
 
         # Create a Redis client.
-        redis_client = services.create_redis_client(redis_address)
+        redis_client = services.create_redis_client(
+            redis_address, password=redis_password)
 
         # Check that the verion information on this node matches the version
         # information that the cluster was started with.
@@ -339,6 +357,7 @@ def start(node_ip_address, redis_address, redis_port, num_redis_shards,
             object_manager_ports=[object_manager_port],
             num_workers=num_workers,
             object_store_memory=object_store_memory,
+            redis_password=redis_password,
             cleanup=False,
             redirect_worker_output=not no_redirect_worker_output,
             redirect_output=not no_redirect_output,