Squashed commit of the following:

commit 1593350efe1e9520171eb52ade25fd1022c512f6 Merge: 504399b 2c93708 Author: Oscar Knagg <[email protected]> Date: Tue Oct 12 16:05:41 2021 +0100 Merge remote-tracking branch 'origin' into tls-working commit 504399b Author: Oscar Knagg <[email protected]> Date: Tue Oct 12 14:56:17 2021 +0100 format.sh changes commit 7b23f9e Author: Oscar Knagg <[email protected]> Date: Tue Oct 12 14:54:13 2021 +0100 Fix tests commit fdbe8eb Author: Oscar Knagg <[email protected]> Date: Tue Oct 12 13:46:17 2021 +0100 Move functions around commit 36ce6ac Merge: a33e32f 8241a03 Author: Oscar Knagg <[email protected]> Date: Tue Oct 12 13:40:10 2021 +0100 Merge branch 'master' of https://github.com/ray-project/ray into tls-working commit a33e32f Author: Oscar Knagg <[email protected]> Date: Tue Oct 12 13:33:51 2021 +0100 Fix bad import commit 263e8f6 Author: Oscar Knagg <[email protected]> Date: Tue Oct 12 12:57:41 2021 +0100 Add TLS configuration to ray_config_def.h commit 425ce87 Author: Oscar Knagg <[email protected]> Date: Tue Oct 12 12:57:17 2021 +0100 Formatting commit b510a7b Author: Oscar Knagg <[email protected]> Date: Mon Oct 11 15:48:54 2021 +0100 Move tests into separate file commit 97df185 Author: Oscar Knagg <[email protected]> Date: Mon Oct 11 14:09:25 2021 +0100 load_certs_from_env -> tls_utils commit fb1b05c Author: Oscar Knagg <[email protected]> Date: Mon Oct 11 14:08:09 2021 +0100 Docs v1 commit 9e95bb1 Author: Oscar Knagg <[email protected]> Date: Mon Oct 11 11:52:17 2021 +0100 tls_utils file commit 8596893 Merge: d04fe6d ab55b80 Author: Oscar Knagg <[email protected]> Date: Mon Oct 11 11:46:07 2021 +0100 Merge branch 'master' of https://github.com/ray-project/ray into tls
ray-project · ericl · Oct 21, 2021 · Sep 8, 2021 · Sep 8, 2021 · Sep 8, 2021
commit 78bbb341c605b6ffda2c66c6b25c99ac08d58917
diff --git a/dashboard/agent.py b/dashboard/agent.py
@@ -11,7 +11,6 @@
 import traceback
 
 from grpc.experimental import aio as aiogrpc
-from distutils.version import LooseVersion
 
 import ray
 import ray.dashboard.consts as dashboard_consts
@@ -84,7 +83,7 @@ def __init__(self,
  assert self.ppid > 0
  logger.info("Parent pid is %s", self.ppid)
  self.server = aiogrpc.server(options=(("grpc.so_reuseport", 0), ))
- self.grpc_port = ray._private.utils.add_port_to_grpc_server(
+ self.grpc_port = ray._private.tls_utils.add_port_to_grpc_server(
  self.server, f"[::]:{self.dashboard_agent_port}")
  logger.info("Dashboard agent grpc address: %s:%s", self.ip,
  self.grpc_port)
@@ -144,12 +143,8 @@ async def _check_parent():
  sys.exit(-1)
 
  # Create a http session for all modules.
- # aiohttp<4.0.0 uses a 'loop' variable, aiohttp>=4.0.0 doesn't anymore
- if LooseVersion(aiohttp.__version__) < LooseVersion("4.0.0"):
- self.http_session = aiohttp.ClientSession(
- loop=asyncio.get_event_loop())
- else:
- self.http_session = aiohttp.ClientSession()
+ self.http_session = aiohttp.ClientSession(
+ loop=asyncio.get_event_loop())
 
  # Start a grpc asyncio server.
  await self.server.start()
@@ -342,8 +337,8 @@ async def _check_parent():
  # https://github.com/ray-project/ray/issues/14026.
  if sys.platform == "win32":
  logger.warning(
- "The dashboard is currently disabled on windows. "
- "See https://github.com/ray-project/ray/issues/14026 "
+ "The dashboard is currently disabled on windows."
+ "See https://github.com/ray-project/ray/issues/14026"
  "for more details")
  while True:
  time.sleep(999)
@@ -367,34 +362,14 @@ async def _check_parent():
  loop = asyncio.get_event_loop()
  loop.run_until_complete(agent.run())
  except Exception as e:
- # All these env vars should be available because
- # they are provided by the parent raylet.
- restart_count = os.environ["RESTART_COUNT"]
- max_restart_count = os.environ["MAX_RESTART_COUNT"]
- raylet_pid = os.environ["RAY_RAYLET_PID"]
- node_ip = args.node_ip_address
- if restart_count >= max_restart_count:
- # Agent is failed to be started many times.
- # Push an error to all drivers, so that users can know the
- # impact of the issue.
- redis_client = ray._private.services.create_redis_client(
- args.redis_address, password=args.redis_password)
- traceback_str = ray._private.utils.format_error_message(
- traceback.format_exc())
- message = (
- f"(ip={node_ip}) "
- f"The agent on node {platform.uname()[1]} failed to "
- f"be restarted {max_restart_count} "
- "times. There are 3 possible problems if you see this error."
- "\n 1. The dashboard might not display correct "
- "information on this node."
- "\n 2. Metrics on this node won't be reported."
- "\n 3. runtime_env APIs won't work."
- "\nCheck out the `dashboard_agent.log` to see the "
- "detailed failure messages.")
- ray._private.utils.push_error_to_driver_through_redis(
- redis_client, ray_constants.DASHBOARD_AGENT_DIED_ERROR,
- message)
- logger.error(message)
- logger.exception(e)
- exit(1)
+ # Something went wrong, so push an error to all drivers.
+ redis_client = ray._private.services.create_redis_client(
+ args.redis_address, password=args.redis_password)
+ traceback_str = ray._private.utils.format_error_message(
+ traceback.format_exc())
+ message = ("The agent on node {} failed with the following "
+ "error:\n{}".format(platform.uname()[1], traceback_str))
+ ray._private.utils.push_error_to_driver_through_redis(
+ redis_client, ray_constants.DASHBOARD_AGENT_DIED_ERROR, message)
+ logger.exception(message)
+ raise e
diff --git a/dashboard/head.py b/dashboard/head.py
@@ -121,7 +121,7 @@ def __init__(self, http_host, http_port, http_port_retries, redis_address,
  ip, port = redis_address.split(":")
  self.gcs_client = connect_to_gcs(ip, int(port), redis_password)
  self.server = aiogrpc.server(options=(("grpc.so_reuseport", 0), ))
- self.grpc_port = ray._private.utils.add_port_to_grpc_server(
+ self.grpc_port = ray._private.tls_utils.add_port_to_grpc_server(
  self.server, "[::]:0")
  logger.info("Dashboard head grpc address: %s:%s", self.ip,
  self.grpc_port)

@@ -234,6 +234,28 @@ to localhost when the ray is started using ``ray.init``.
 See the `Redis security documentation <https://redis.io/topics/security>`__
 for more information.
 
+TLS Authentication
+------------------
+
+Ray can be configured to use TLS on it's gRPC channels.
+This has means that connecting to the Ray client on the head node will
+require an appropriate set of credentials and also that data exchanged between
+various processes (client, head, workers) will be encrypted.
+
+Enabling TLS will cause a performance hit due to the extra overhead of mutual
+authentication and encryption.
+Testing has shown that this overhead is large for small workloads and becomes
+relatively smaller for large workloads.
+The exact overhead will depend on the nature of your workload.
+
+TLS is enabled by setting environment variables.
+
+- ``RAY_USE_TLS``: Either 1 or 0 to use/not-use TLS. If this is set to 1 then all of the environment variables below must be set. Default: 0.
+- ``RAY_TLS_SERVER_CERT``: Location of a `certificate file` which is presented to other endpoints so as to achieve mutual authentication.
+- ``RAY_TLS_SERVER_KEY``: Location of a `private key file` which is the cryptographic means to prove to other endpoints that you are the authorized user of a given certificate.
+- ``RAY_TLS_CA_CERT``: Location of a `CA certificate file` which allows TLS to decide whether an endpoint's certificate has been signed by the correct authority.
+
+
 Java Applications
 -----------------
 

diff --git a/python/ray/_private/test_utils.py b/python/ray/_private/test_utils.py
@@ -7,24 +7,25 @@
 import pathlib
 import subprocess
 import sys
-import tempfile
 import time
 import timeit
 import math
 import traceback
-import datetime
 from typing import Optional, Any, List, Dict
 from contextlib import redirect_stdout, redirect_stderr
 import yaml
 import socket
 import pytest
+import tempfile
 
 import ray
 import ray._private.services
 import ray._private.utils
 import ray._private.gcs_utils as gcs_utils
+from ray._private.tls_utils import generate_self_signed_tls_certs
 from ray.util.queue import Queue, _QueueActor, Empty
 from ray.scripts.scripts import main as ray_main
+
 try:
  from prometheus_client.parser import text_string_to_metric_families
 except (ImportError, ModuleNotFoundError):
@@ -690,57 +691,11 @@ async def get_batch(self,
  return batch
 
 
-def generate_self_signed_tls_certs():
- """Create self-signed key/cert pair for testing.
-
- This method requires the library ``cryptography`` be installed.
- """
- try:
- from cryptography import x509
- from cryptography.hazmat.backends import default_backend
- from cryptography.hazmat.primitives import hashes, serialization
- from cryptography.hazmat.primitives.asymmetric import rsa
- from cryptography.x509.oid import NameOID
- except ImportError:
- raise ImportError(
- "Using `Security.temporary` requires `cryptography`, please "
- "install it using either pip or conda")
- key = rsa.generate_private_key(
- public_exponent=65537, key_size=2048, backend=default_backend())
- key_contents = key.private_bytes(
- encoding=serialization.Encoding.PEM,
- format=serialization.PrivateFormat.PKCS8,
- encryption_algorithm=serialization.NoEncryption(),
- ).decode()
-
- ray_interal = x509.Name(
- [x509.NameAttribute(NameOID.COMMON_NAME, "ray-internal")])
- # This is the same logic used by the GCS server to acquire a
- # private/interal IP address to listen on. If we just use localhost +
- # 127.0.0.1 then we won't be able to connect to the GCS and will get
- # an error like "No match found for server name: 192.168.X.Y"
- s = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
- s.connect(("8.8.8.8", 80))
- private_ip_address = s.getsockname()[0]
- s.close()
- altnames = x509.SubjectAlternativeName([
- x509.DNSName(socket.gethostbyname(
- socket.gethostname())), # Probably 127.0.0.1
- x509.DNSName("127.0.0.1"),
- x509.DNSName(private_ip_address), # 192.168.*.*
- x509.DNSName("localhost"),
- ])
- now = datetime.datetime.utcnow()
- cert = (x509.CertificateBuilder()
- .subject_name(ray_interal).issuer_name(ray_interal).add_extension(
- altnames, critical=False).public_key(key.public_key())
- .serial_number(x509.random_serial_number()).not_valid_before(now)
- .not_valid_after(now + datetime.timedelta(days=365)).sign(
- key, hashes.SHA256(), default_backend()))
-
- cert_contents = cert.public_bytes(serialization.Encoding.PEM).decode()
-
- return cert_contents, key_contents
+def is_placement_group_removed(pg):
+ table = ray.util.placement_group_table(pg)
+ if "state" not in table:
+ return False
+ return table["state"] == "REMOVED"
 
 
 def setup_tls():
@@ -772,10 +727,3 @@ def teardown_tls(key_filepath, cert_filepath, temp_dir):
  del os.environ["RAY_TLS_SERVER_CERT"]
  del os.environ["RAY_TLS_SERVER_KEY"]
  del os.environ["RAY_TLS_CA_CERT"]
-
-
-def is_placement_group_removed(pg):
- table = ray.util.placement_group_table(pg)
- if "state" not in table:
- return False
- return table["state"] == "REMOVED"
diff --git a/python/ray/_private/tls_utils.py b/python/ray/_private/tls_utils.py
@@ -0,0 +1,85 @@
+import datetime
+import os
+import socket
+
+import grpc
+
+
+def generate_self_signed_tls_certs():
+ """Create self-signed key/cert pair for testing.
+
+ This method requires the library ``cryptography`` be installed.
+ """
+ try:
+ from cryptography import x509
+ from cryptography.hazmat.backends import default_backend
+ from cryptography.hazmat.primitives import hashes, serialization
+ from cryptography.hazmat.primitives.asymmetric import rsa
+ from cryptography.x509.oid import NameOID
+ except ImportError:
+ raise ImportError(
+ "Using `Security.temporary` requires `cryptography`, please "
+ "install it using either pip or conda")
+ key = rsa.generate_private_key(
+ public_exponent=65537, key_size=2048, backend=default_backend())
+ key_contents = key.private_bytes(
+ encoding=serialization.Encoding.PEM,
+ format=serialization.PrivateFormat.PKCS8,
+ encryption_algorithm=serialization.NoEncryption(),
+ ).decode()
+
+ ray_interal = x509.Name(
+ [x509.NameAttribute(NameOID.COMMON_NAME, "ray-internal")])
+ # This is the same logic used by the GCS server to acquire a
+ # private/interal IP address to listen on. If we just use localhost +
+ # 127.0.0.1 then we won't be able to connect to the GCS and will get
+ # an error like "No match found for server name: 192.168.X.Y"
+ s = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
+ s.connect(("8.8.8.8", 80))
+ private_ip_address = s.getsockname()[0]
+ s.close()
+ altnames = x509.SubjectAlternativeName([
+ x509.DNSName(socket.gethostbyname(
+ socket.gethostname())), # Probably 127.0.0.1
+ x509.DNSName("127.0.0.1"),
+ x509.DNSName(private_ip_address), # 192.168.*.*
+ x509.DNSName("localhost"),
+ ])
+ now = datetime.datetime.utcnow()
+ cert = (x509.CertificateBuilder().subject_name(ray_interal).issuer_name(
+ ray_interal).add_extension(altnames, critical=False).public_key(
+ key.public_key()).serial_number(
+ x509.random_serial_number()).not_valid_before(now)
+ .not_valid_after(now + datetime.timedelta(days=365)).sign(
+ key, hashes.SHA256(), default_backend()))
+
+ cert_contents = cert.public_bytes(serialization.Encoding.PEM).decode()
+
+ return cert_contents, key_contents
+
+
+def add_port_to_grpc_server(server, address):
+ if os.environ.get("RAY_USE_TLS", "0") == "1":
+ server_cert_chain, private_key, ca_cert = load_certs_from_env()
+ credentials = grpc.ssl_server_credentials(
+ [(private_key, server_cert_chain)],
+ root_certificates=ca_cert,
+ require_client_auth=ca_cert is not None)
+ return server.add_secure_port(address, credentials)
+ else:
+ return server.add_insecure_port(address)
+
+
+def load_certs_from_env():
+ if os.environ.get("RAY_USE_TLS", "0") == "1":
+ with open(os.environ["RAY_TLS_SERVER_CERT"], "rb") as f:
+ server_cert_chain = f.read()
+ with open(os.environ["RAY_TLS_SERVER_KEY"], "rb") as f:
+ private_key = f.read()
+ if "RAY_TLS_CA_CERT" in os.environ:
+ with open(os.environ["RAY_TLS_CA_CERT"], "rb") as f:
+ ca_cert = f.read()
+ else:
+ ca_cert = None
+
+ return server_cert_chain, private_key, ca_cert
diff --git a/python/ray/_private/utils.py b/python/ray/_private/utils.py
@@ -27,6 +27,7 @@
 import ray
 import ray._private.gcs_utils as gcs_utils
 import ray.ray_constants as ray_constants
+from ray._private.tls_utils import load_certs_from_env
 
 # Import psutil after ray so the packaged version is used.
 import psutil
@@ -1111,21 +1112,6 @@ def validate_namespace(namespace: str):
  "Pass None to not specify a namespace.")
 
 
-def load_certs_from_env():
- if os.environ.get("RAY_USE_TLS", "0") == "1":
- with open(os.environ["RAY_TLS_SERVER_CERT"], "rb") as f:
- server_cert_chain = f.read()
- with open(os.environ["RAY_TLS_SERVER_KEY"], "rb") as f:
- private_key = f.read()
- if "RAY_TLS_CA_CERT" in os.environ:
- with open(os.environ["RAY_TLS_CA_CERT"], "rb") as f:
- ca_cert = f.read()
- else:
- ca_cert = None
-
- return server_cert_chain, private_key, ca_cert
-
-
 def init_grpc_channel(address: str,
  options: Optional[Sequence[Tuple[str, Any]]] = None,
  asynchronous: bool = False):
@@ -1142,15 +1128,3 @@ def init_grpc_channel(address: str,
  channel = grpc_module.insecure_channel(address, options=options)
 
  return channel
-
-
-def add_port_to_grpc_server(server, address):
- if os.environ.get("RAY_USE_TLS", "0") == "1":
- server_cert_chain, private_key, ca_cert = load_certs_from_env()
- credentials = grpc.ssl_server_credentials(
- [(private_key, server_cert_chain)],
- root_certificates=ca_cert,
- require_client_auth=ca_cert is not None)
- return server.add_secure_port(address, credentials)
- else:
- return server.add_insecure_port(address)