wehub-resource-sync
505 lines
20 KiB
Python
505 lines
20 KiB
Python
import asyncio
|
|
import errno
|
|
import ipaddress
|
|
import logging
|
|
import os
|
|
import pathlib
|
|
import posixpath
|
|
import sys
|
|
import time
|
|
from math import floor
|
|
from typing import List
|
|
|
|
import multidict
|
|
from packaging.version import Version
|
|
|
|
import ray
|
|
import ray.dashboard.optional_utils as dashboard_optional_utils
|
|
import ray.dashboard.timezone_utils as timezone_utils
|
|
import ray.dashboard.utils as dashboard_utils
|
|
from ray import ray_constants
|
|
from ray._common.network_utils import build_address, parse_address
|
|
from ray._common.usage.usage_lib import TagKey, record_extra_usage_tag
|
|
from ray._common.utils import get_or_create_event_loop
|
|
from ray._private.authentication import (
|
|
authentication_constants,
|
|
authentication_utils as auth_utils,
|
|
)
|
|
from ray._private.authentication.http_token_authentication import (
|
|
get_token_auth_middleware,
|
|
)
|
|
from ray._raylet import get_authentication_mode
|
|
from ray.dashboard.dashboard_metrics import DashboardPrometheusMetrics
|
|
from ray.dashboard.head import DashboardHeadModule
|
|
|
|
# All third-party dependencies that are not included in the minimal Ray
|
|
# installation must be included in this file. This allows us to determine if
|
|
# the agent has the necessary dependencies to be started.
|
|
from ray.dashboard.optional_deps import aiohttp, hdrs
|
|
from ray.dashboard.subprocesses.handle import SubprocessModuleHandle
|
|
from ray.dashboard.subprocesses.routes import SubprocessRouteTable
|
|
|
|
# Logger for this module. It should be configured at the entry point
|
|
# into the program using Ray. Ray provides a default configuration at
|
|
# entry/init points.
|
|
logger = logging.getLogger(__name__)
|
|
routes = dashboard_optional_utils.DashboardHeadRouteTable
|
|
|
|
# Env var that enables follow_symlinks for serving UI static files.
|
|
# This is an advanced setting that should only be used with special Ray installations
|
|
# where the dashboard build files are symlinked to a different directory.
|
|
# This is not recommended for most users and can pose a security risk.
|
|
# Please reference the aiohttp docs here:
|
|
# https://docs.aiohttp.org/en/stable/web_reference.html#aiohttp.web.UrlDispatcher.add_static
|
|
ENV_VAR_FOLLOW_SYMLINKS = "RAY_DASHBOARD_BUILD_FOLLOW_SYMLINKS"
|
|
FOLLOW_SYMLINKS_ENABLED = os.environ.get(ENV_VAR_FOLLOW_SYMLINKS) == "1"
|
|
if FOLLOW_SYMLINKS_ENABLED:
|
|
logger.warning(
|
|
"Enabling RAY_DASHBOARD_BUILD_FOLLOW_SYMLINKS is not recommended as it "
|
|
"allows symlinks to directories outside the dashboard build folder. "
|
|
"You may accidentally expose files on your system outside of the "
|
|
"build directory."
|
|
)
|
|
|
|
|
|
def setup_static_dir():
|
|
build_dir = os.path.join(
|
|
os.path.dirname(os.path.abspath(__file__)), "client", "build"
|
|
)
|
|
module_name = os.path.basename(os.path.dirname(__file__))
|
|
if not os.path.isdir(build_dir):
|
|
raise dashboard_utils.FrontendNotFoundError(
|
|
errno.ENOENT,
|
|
"Dashboard build directory not found. If installing "
|
|
"from source, please follow the additional steps "
|
|
"required to build the dashboard"
|
|
f"(cd python/ray/{module_name}/client "
|
|
"&& npm ci "
|
|
"&& npm run build)",
|
|
build_dir,
|
|
)
|
|
|
|
static_dir = os.path.join(build_dir, "static")
|
|
routes.static("/static", static_dir, follow_symlinks=FOLLOW_SYMLINKS_ENABLED)
|
|
return build_dir
|
|
|
|
|
|
class HttpServerDashboardHead:
|
|
def __init__(
|
|
self,
|
|
ip: str,
|
|
http_host: str,
|
|
http_port: int,
|
|
http_port_retries: int,
|
|
gcs_address: str,
|
|
session_name: str,
|
|
metrics: DashboardPrometheusMetrics,
|
|
proxy_server_url: str | None = None,
|
|
):
|
|
self.ip = ip
|
|
self.http_host = http_host
|
|
self.http_port = http_port
|
|
self.http_port_retries = http_port_retries
|
|
self.head_node_ip = parse_address(gcs_address)[0]
|
|
self.metrics = metrics
|
|
self._session_name = session_name
|
|
self.proxy_server_url = proxy_server_url
|
|
|
|
# Below attirubtes are filled after `run` API is invoked.
|
|
self.runner = None
|
|
|
|
# Setup Dashboard Routes
|
|
try:
|
|
build_dir = setup_static_dir()
|
|
logger.info("Setup static dir for dashboard: %s", build_dir)
|
|
except dashboard_utils.FrontendNotFoundError as ex:
|
|
# Not to raise FrontendNotFoundError due to NPM incompatibilities
|
|
# with Windows.
|
|
# Please refer to ci.sh::build_dashboard_front_end()
|
|
if sys.platform in ["win32", "cygwin"]:
|
|
logger.warning(ex)
|
|
else:
|
|
raise ex
|
|
dashboard_optional_utils.DashboardHeadRouteTable.bind(self)
|
|
|
|
# Create a http session for all modules.
|
|
# aiohttp<4.0.0 uses a 'loop' variable, aiohttp>=4.0.0 doesn't anymore
|
|
if Version(aiohttp.__version__) < Version("4.0.0"):
|
|
self.http_session = aiohttp.ClientSession(loop=get_or_create_event_loop())
|
|
else:
|
|
self.http_session = aiohttp.ClientSession()
|
|
|
|
@routes.get("/")
|
|
async def get_index(self, req) -> aiohttp.web.FileResponse:
|
|
try:
|
|
# This API will be no-op after the first report.
|
|
# Note: We always record the usage, but it is not reported
|
|
# if the usage stats is disabled.
|
|
record_extra_usage_tag(TagKey.DASHBOARD_USED, "True")
|
|
except Exception as e:
|
|
logger.warning(
|
|
"Failed to record the dashboard usage. "
|
|
"This error message is harmless and can be ignored. "
|
|
f"Error: {e}"
|
|
)
|
|
resp = aiohttp.web.FileResponse(
|
|
os.path.join(
|
|
os.path.dirname(os.path.abspath(__file__)), "client/build/index.html"
|
|
)
|
|
)
|
|
resp.headers["Cache-Control"] = "no-store"
|
|
return resp
|
|
|
|
@routes.get("/favicon.ico")
|
|
async def get_favicon(self, req) -> aiohttp.web.FileResponse:
|
|
return aiohttp.web.FileResponse(
|
|
os.path.join(
|
|
os.path.dirname(os.path.abspath(__file__)), "client/build/favicon.ico"
|
|
)
|
|
)
|
|
|
|
@routes.get("/timezone")
|
|
async def get_timezone(self, req) -> aiohttp.web.Response:
|
|
try:
|
|
current_timezone = timezone_utils.get_current_timezone_info()
|
|
return aiohttp.web.json_response(current_timezone)
|
|
|
|
except Exception as e:
|
|
logger.error(f"Error getting timezone: {e}")
|
|
return aiohttp.web.Response(
|
|
status=500, text="Internal Server Error:" + str(e)
|
|
)
|
|
|
|
@routes.get("/api/authentication_mode")
|
|
async def get_authentication_mode(self, req) -> aiohttp.web.Response:
|
|
try:
|
|
mode = get_authentication_mode()
|
|
mode_str = auth_utils.get_authentication_mode_name(mode)
|
|
|
|
response = aiohttp.web.json_response({"authentication_mode": mode_str})
|
|
|
|
# If auth is disabled, clear any existing authentication cookie
|
|
if mode_str == "disabled":
|
|
response.set_cookie(
|
|
authentication_constants.AUTHENTICATION_TOKEN_COOKIE_NAME,
|
|
"",
|
|
max_age=0,
|
|
path="/",
|
|
)
|
|
|
|
return response
|
|
except Exception as e:
|
|
logger.error(f"Error getting authentication mode: {e}")
|
|
return aiohttp.web.Response(
|
|
status=500, text="Internal Server Error: " + str(e)
|
|
)
|
|
|
|
@routes.post("/api/authenticate")
|
|
async def authenticate(self, req) -> aiohttp.web.Response:
|
|
"""
|
|
Authenticate a user by validating their token and setting a secure HttpOnly cookie.
|
|
|
|
This endpoint accepts a token via the Authorization header, validates it,
|
|
and if valid, sets an HttpOnly cookie for subsequent requests from the web dashboard.
|
|
"""
|
|
try:
|
|
# Check if token authentication is enabled
|
|
if not auth_utils.is_token_auth_enabled():
|
|
return aiohttp.web.Response(
|
|
status=401,
|
|
text="Unauthorized: Token authentication is not enabled",
|
|
)
|
|
|
|
# Get token from Authorization header
|
|
auth_header = req.headers.get(
|
|
authentication_constants.AUTHORIZATION_HEADER_NAME, ""
|
|
)
|
|
|
|
if not auth_header:
|
|
return aiohttp.web.Response(
|
|
status=401,
|
|
text="Unauthorized: Missing authentication token",
|
|
)
|
|
|
|
# Validate the token
|
|
if not auth_utils.validate_request_token(auth_header):
|
|
return aiohttp.web.Response(
|
|
status=403,
|
|
text="Forbidden: Invalid authentication token",
|
|
)
|
|
|
|
# Token is valid - extract the token value (remove "Bearer " prefix if present)
|
|
token = auth_header
|
|
if auth_header.lower().startswith(
|
|
authentication_constants.AUTHORIZATION_BEARER_PREFIX.lower()
|
|
):
|
|
token = auth_header[
|
|
len(authentication_constants.AUTHORIZATION_BEARER_PREFIX) :
|
|
] # Remove "Bearer " prefix
|
|
|
|
# Create successful response
|
|
response = aiohttp.web.json_response(
|
|
{"status": "authenticated", "message": "Token is valid"}
|
|
)
|
|
|
|
# Set secure HttpOnly cookie
|
|
# Check if the connection is secure (HTTPS)
|
|
is_secure = req.scheme == "https"
|
|
|
|
response.set_cookie(
|
|
authentication_constants.AUTHENTICATION_TOKEN_COOKIE_NAME,
|
|
token,
|
|
max_age=authentication_constants.AUTHENTICATION_TOKEN_COOKIE_MAX_AGE, # 30 days (matching previous behavior)
|
|
path="/",
|
|
httponly=True, # Prevents JavaScript access (XSS protection)
|
|
samesite="Strict", # Prevents CSRF attacks
|
|
secure=is_secure, # Only send over HTTPS if connection is secure
|
|
)
|
|
|
|
return response
|
|
|
|
except Exception as e:
|
|
logger.error(f"Error during authentication: {e}")
|
|
return aiohttp.web.Response(
|
|
status=500, text="Internal Server Error: " + str(e)
|
|
)
|
|
|
|
def get_address(self):
|
|
assert self.http_host and self.http_port
|
|
return self.http_host, self.http_port
|
|
|
|
@aiohttp.web.middleware
|
|
async def path_clean_middleware(self, request, handler):
|
|
if request.path.startswith("/static") or request.path.startswith("/logs"):
|
|
parent = pathlib.PurePosixPath(
|
|
"/logs" if request.path.startswith("/logs") else "/static"
|
|
)
|
|
|
|
# If the destination is not relative to the expected directory,
|
|
# then the user is attempting path traversal, so deny the request.
|
|
request_path = pathlib.PurePosixPath(posixpath.realpath(request.path))
|
|
if request_path != parent and parent not in request_path.parents:
|
|
logger.info(
|
|
f"Rejecting {request_path=} because it is not relative to {parent=}"
|
|
)
|
|
raise aiohttp.web.HTTPForbidden()
|
|
return await handler(request)
|
|
|
|
@aiohttp.web.middleware
|
|
async def proxy_server_middleware(self, request, handler):
|
|
"""Redirects the request to the proxy server if the related proxy server
|
|
url env var is set.
|
|
"""
|
|
# paths/endpoints that should not be directed when RAY_API_PROXY_URL_ENV_VAR is set
|
|
ignored_path = (
|
|
"/api/gcs_healthz",
|
|
"/usage_stats_enabled",
|
|
"/api/healthz",
|
|
"/api/local_raylet_healthz",
|
|
)
|
|
|
|
rel_path = str(request.rel_url)
|
|
body_data = None
|
|
if self.proxy_server_url and not rel_path.startswith(ignored_path):
|
|
logger.debug(f"Redirecting request to {self.proxy_server_url}")
|
|
body_data = await request.read()
|
|
|
|
# Headers that should be removed from the new request
|
|
# Case sensitive, keep everything lowercase for checking purposes.
|
|
headers_to_remove = {
|
|
"content-encoding", # This could cause the browser to try to unzip plain-text
|
|
"content-length", # Let aiohttp calculate instead
|
|
"host",
|
|
"connection",
|
|
"keep-alive",
|
|
"te",
|
|
"trailers",
|
|
"upgrade",
|
|
"proxy-authenticate",
|
|
"proxy-authorization",
|
|
"transfer-encoding",
|
|
}
|
|
# We have to use multidict here since there are multiple cookies being set
|
|
proxy_headers = multidict.CIMultiDict(
|
|
(k, v)
|
|
for k, v in request.headers.items()
|
|
if k.lower() not in headers_to_remove
|
|
)
|
|
|
|
# This is to fix 415 error caused by content-type not in headers
|
|
proxy_headers.setdefault("Content-Type", "application/json")
|
|
|
|
timeout = aiohttp.ClientTimeout(total=30)
|
|
try:
|
|
async with aiohttp.ClientSession(timeout=timeout) as session:
|
|
# Manual concatenation to preserve potential paths in proxy_server_url
|
|
full_url = f"{self.proxy_server_url.rstrip('/')}{rel_path}"
|
|
logger.debug(
|
|
f"Sending request to proxy server instead. {full_url=}"
|
|
)
|
|
|
|
async with session.request(
|
|
method=request.method,
|
|
url=full_url,
|
|
headers=proxy_headers,
|
|
data=body_data,
|
|
) as resp:
|
|
body = await resp.read()
|
|
# only fallback if the reponse from the proxy server is not found
|
|
if not resp.status == 404:
|
|
resp_headers = multidict.CIMultiDict(
|
|
(k, v)
|
|
for k, v in resp.headers.items()
|
|
if k.lower()
|
|
not in {
|
|
"transfer-encoding",
|
|
"content-encoding",
|
|
"content-length",
|
|
}
|
|
)
|
|
|
|
return aiohttp.web.Response(
|
|
body=body, status=resp.status, headers=resp_headers
|
|
)
|
|
else:
|
|
logger.warning(
|
|
f"Calling endpoint {request.rel_url} returned status {resp.status} and {body}, falling back"
|
|
)
|
|
except Exception as e:
|
|
logger.warning(
|
|
f"Failed to use proxy server endpoint: {e}. Falling back to original handler."
|
|
)
|
|
|
|
return await handler(request)
|
|
|
|
@aiohttp.web.middleware
|
|
async def metrics_middleware(self, request, handler):
|
|
start_time = time.monotonic()
|
|
|
|
try:
|
|
response = await handler(request)
|
|
status_tag = f"{floor(response.status / 100)}xx"
|
|
return response
|
|
except (Exception, asyncio.CancelledError):
|
|
status_tag = "5xx"
|
|
raise
|
|
finally:
|
|
resp_time = time.monotonic() - start_time
|
|
try:
|
|
self.metrics.metrics_request_duration.labels(
|
|
endpoint=handler.__name__,
|
|
http_status=status_tag,
|
|
Version=ray.__version__,
|
|
SessionName=self._session_name,
|
|
Component="dashboard",
|
|
).observe(resp_time)
|
|
self.metrics.metrics_request_count.labels(
|
|
method=request.method,
|
|
endpoint=handler.__name__,
|
|
http_status=status_tag,
|
|
Version=ray.__version__,
|
|
SessionName=self._session_name,
|
|
Component="dashboard",
|
|
).inc()
|
|
except Exception as e:
|
|
logger.exception(f"Error emitting api metrics: {e}")
|
|
|
|
@aiohttp.web.middleware
|
|
async def cache_control_static_middleware(self, request, handler):
|
|
if request.path.startswith("/static"):
|
|
response = await handler(request)
|
|
response.headers["Cache-Control"] = "max-age=31536000"
|
|
return response
|
|
return await handler(request)
|
|
|
|
async def run(
|
|
self,
|
|
dashboard_head_modules: List[DashboardHeadModule],
|
|
subprocess_module_handles: List[SubprocessModuleHandle],
|
|
):
|
|
# Bind http routes of each module.
|
|
for m in dashboard_head_modules:
|
|
dashboard_optional_utils.DashboardHeadRouteTable.bind(m)
|
|
|
|
for h in subprocess_module_handles:
|
|
SubprocessRouteTable.bind(h)
|
|
|
|
# Public endpoints that don't require authentication.
|
|
# These are needed for the dashboard to load and request an auth token.
|
|
public_exact_paths = {
|
|
"/", # Root index.html
|
|
"/favicon.ico",
|
|
"/api/authentication_mode",
|
|
"/api/authenticate", # Token authentication endpoint
|
|
"/api/healthz", # General healthcheck
|
|
"/api/gcs_healthz", # GCS health check
|
|
"/api/local_raylet_healthz", # Raylet health check
|
|
"/-/healthz", # Serve health check
|
|
}
|
|
public_path_prefixes = ("/static/",) # Static assets (JS, CSS, images)
|
|
|
|
# Http server should be initialized after all modules loaded.
|
|
# working_dir uploads for job submission can be up to 100MiB.
|
|
|
|
app = aiohttp.web.Application(
|
|
client_max_size=ray_constants.DASHBOARD_CLIENT_MAX_SIZE,
|
|
middlewares=[
|
|
self.metrics_middleware,
|
|
get_token_auth_middleware(
|
|
aiohttp, public_exact_paths, public_path_prefixes
|
|
),
|
|
self.path_clean_middleware,
|
|
dashboard_optional_utils.get_browser_request_middleware(
|
|
aiohttp,
|
|
allowed_methods={"GET", "HEAD", "OPTIONS"},
|
|
allowed_paths=["/api/authenticate"],
|
|
),
|
|
self.cache_control_static_middleware,
|
|
self.proxy_server_middleware,
|
|
],
|
|
)
|
|
app.add_routes(routes=routes.bound_routes())
|
|
app.add_routes(routes=SubprocessRouteTable.bound_routes())
|
|
|
|
self.runner = aiohttp.web.AppRunner(
|
|
app,
|
|
access_log_format=(
|
|
"%a %t '%r' %s %b bytes %D us '%{Referer}i' '%{User-Agent}i'"
|
|
),
|
|
)
|
|
await self.runner.setup()
|
|
last_ex = None
|
|
for i in range(1 + self.http_port_retries):
|
|
try:
|
|
site = aiohttp.web.TCPSite(self.runner, self.http_host, self.http_port)
|
|
await site.start()
|
|
break
|
|
except OSError as e:
|
|
last_ex = e
|
|
self.http_port += 1
|
|
logger.warning("Try to use port %s: %s", self.http_port, e)
|
|
else:
|
|
raise Exception(
|
|
f"Failed to find a valid port for dashboard after "
|
|
f"{self.http_port_retries} retries: {last_ex}"
|
|
)
|
|
self.http_host, self.http_port, *_ = site._server.sockets[0].getsockname()
|
|
self.http_host = (
|
|
self.ip
|
|
if ipaddress.ip_address(self.http_host).is_unspecified
|
|
else self.http_host
|
|
)
|
|
logger.info(
|
|
"Dashboard head http address: %s",
|
|
build_address(self.http_host, self.http_port),
|
|
)
|
|
# Dump registered http routes.
|
|
dump_routes = [r for r in app.router.routes() if r.method != hdrs.METH_HEAD]
|
|
for r in dump_routes:
|
|
logger.info(r)
|
|
logger.info("Registered %s routes.", len(dump_routes))
|
|
|
|
async def cleanup(self):
|
|
# Wait for finish signal.
|
|
await self.runner.cleanup()
|