fix(redis): decouple TLS from auth so on-prem can require a password

charts/model-engine/Chart.yaml

@@ -15,7 +15,7 @@ type: application
 # This is the chart version. This version number should be incremented each time you make changes
 # to the chart and its templates, including the app version.
 # Versions are expected to follow Semantic Versioning (https://semver.org/)
-version: 0.2.9
+version: 0.2.10
 
 # This is the version number of the application being deployed. This version number should be
 # incremented each time you make changes to the application. Versions are not expected to

charts/model-engine/templates/_helpers.tpl

@@ -290,6 +290,46 @@ env:
   {{- end }}
 {{- end }}
 
+{{- /* Resolves the Redis scheme for both the app and the KEDA scaler. An unset
+       value mirrors the app's own fallback -- a configured credential implies
+       TLS, which is what ElastiCache in-transit encryption needs -- so the two
+       cannot disagree about the transport. */}}
+{{- define "modelEngine.redisEnableTLSIsSet" -}}
+{{- if not (or (kindIs "invalid" .Values.redis.enableTLS) (eq (toString .Values.redis.enableTLS) "")) -}}true{{- end -}}
+{{- end }}
+
+{{- define "modelEngine.redisEnableTLS" -}}
+{{- if include "modelEngine.redisEnableTLSIsSet" . -}}
+{{- .Values.redis.enableTLS }}
+{{- else if or .Values.redis.authSecretName .Values.redis.auth -}}
+true
+{{- else -}}
+false
+{{- end -}}
+{{- end }}
+
+{{- /* Every workload that opens a Redis connection must include this, or it
+       reaches an authenticated broker with no credential and fails NOAUTH. */}}
+{{- define "modelEngine.redisAuthEnv" }}
+  {{- if .Values.redis.authSecretName }}
+  - name: REDIS_AUTH_TOKEN
+    valueFrom:
+      secretKeyRef:
+        name: {{ .Values.redis.authSecretName }}
+        key: {{ .Values.redis.authSecretKey | default "auth_token" }}
+  {{- else if .Values.redis.auth }}
+  - name: REDIS_AUTH_TOKEN
+    value: {{ .Values.redis.auth }}
+  {{- end }}
+  {{- /* Withheld when the chart cannot determine it: a token reaching the pod
+         through extraEnvVars is invisible here, and an explicit false would
+         override the app's own inference and force plaintext. */}}
+  {{- if or (include "modelEngine.redisEnableTLSIsSet" .) .Values.redis.authSecretName .Values.redis.auth }}
+  - name: REDIS_ENABLE_TLS
+    value: {{ include "modelEngine.redisEnableTLS" . | quote }}
+  {{- end }}
+{{- end }}
+
 {{- define "modelEngine.serviceEnvBase" }}
 env:
   - name: DD_TRACE_ENABLED
@@ -379,16 +419,7 @@ env:
   - name: CELERY_RESULT_BACKEND
     value: {{ .Values.celeryResultBackend | quote }}
   {{- end }}
-  {{- if .Values.redis.authSecretName }}
-  - name: REDIS_AUTH_TOKEN
-    valueFrom:
-      secretKeyRef:
-        name: {{ .Values.redis.authSecretName }}
-        key: {{ .Values.redis.authSecretKey | default "auth_token" }}
-  {{- else if .Values.redis.auth }}
-  - name: REDIS_AUTH_TOKEN
-    value: {{ .Values.redis.auth }}
-  {{- end }}
+  {{- include "modelEngine.redisAuthEnv" . }}
   {{- if .Values.azure}}
   - name: AZURE_IDENTITY_NAME
     value: {{ .Values.azure.identity_name }}

charts/model-engine/templates/celery_autoscaler_stateful_set.yaml

@@ -66,6 +66,7 @@ spec:
           value: {{ $broker_name }}
         - name: CELERY_ELASTICACHE_ENABLED
           value: {{ (eq $message_broker "elasticache") | squote }}
+        {{- include "modelEngine.redisAuthEnv" . | indent 6 }}
         - name: POD_NAME
           valueFrom:
             fieldRef:

charts/model-engine/templates/service_template_config_map.yaml

@@ -513,7 +513,7 @@ data:
           listName: "launch-endpoint-autoscaling:${ENDPOINT_ID}"
           listLength: "100" # something absurdly high so we don't scale past 1 pod
           activationListLength: "0"
-          enableTLS: "{{ .Values.redis.enableTLS }}"
+          enableTLS: "{{ include "modelEngine.redisEnableTLS" . }}"
           unsafeSsl: "{{ .Values.redis.unsafeSsl }}"
           databaseIndex: "${REDIS_DB_INDEX}"
         {{- if .Values.redis.enableAuth }}

charts/model-engine/values.yaml

@@ -80,7 +80,9 @@ redis:
   auth:
   authSecretName: ""
   authSecretKey: ""
-  enableTLS: false
+  # Unset infers TLS from the presence of a Redis credential, which is what
+  # ElastiCache in-transit encryption needs. Set explicitly to pin the scheme.
+  enableTLS:
   enableAuth: false
   kedaSecretName: ""
   unsafeSsl: false

model-engine/model_engine_server/common/config.py

@@ -6,6 +6,7 @@
 from dataclasses import dataclass
 from pathlib import Path
 from typing import Optional, Sequence
+from urllib.parse import quote
 
 import yaml
 from azure.identity import DefaultAzureCredential
@@ -47,6 +48,24 @@ def get_model_cache_directory_name(model_name: str):
     return name
 
 
+def _apply_redis_auth_token(url: str) -> str:
+    """Add REDIS_AUTH_TOKEN to a Redis URL that carries no credential of its own.
+
+    Keeps the password out of Helm values: the chart names the host and db
+    index, the app supplies the credential from its secret-backed env var. A
+    URL that already has userinfo is left alone so an explicit override wins.
+    """
+    auth_token = os.getenv("REDIS_AUTH_TOKEN")
+    if not auth_token:
+        return url
+    scheme, sep, remainder = url.partition("://")
+    if not sep or "@" in remainder.split("/")[0]:
+        return url
+    # redis-py unquotes the userinfo, so percent-encoding here is what lets a
+    # password containing @, / or # survive URL parsing.
+    return f"{scheme}://:{quote(auth_token, safe='')}@{remainder}"
+
+
 @dataclass
 class HostedModelInferenceServiceConfig:
     gateway_namespace: str
@@ -104,18 +123,18 @@ def from_yaml(cls, yaml_path):
     def cache_redis_url(self) -> str:
         # On-prem Redis support (explicit URL, no cloud provider dependency)
         if self.cache_redis_onprem_url:
-            return self.cache_redis_onprem_url
+            return _apply_redis_auth_token(self.cache_redis_onprem_url)
 
         cloud_provider = infra_config().cloud_provider
 
         # On-prem: support REDIS_HOST env var fallback
         if cloud_provider == "onprem":
             if self.cache_redis_aws_url:
                 logger.info("On-prem deployment using cache_redis_aws_url")
-                return self.cache_redis_aws_url
+                return _apply_redis_auth_token(self.cache_redis_aws_url)
             redis_host = os.getenv("REDIS_HOST", "redis")
             redis_port = getattr(infra_config(), "redis_port", 6379)
-            return f"redis://{redis_host}:{redis_port}/0"
+            return _apply_redis_auth_token(f"redis://{redis_host}:{redis_port}/0")
 
         if cloud_provider == "gcp":
             assert self.cache_redis_gcp_url, "cache_redis_gcp_url required for GCP"
@@ -148,11 +167,13 @@ def cache_redis_url_expiration_timestamp(self) -> Optional[int]:
 
     @property
     def cache_redis_host_port(self) -> str:
-        # redis://redis.url:6379/<db_index>
+        # redis://:password@redis.url:6379/<db_index>
         # -> redis.url:6379
-        if "rediss://" in self.cache_redis_url:
-            return self.cache_redis_url.split("rediss://")[1].split("@")[-1].split("/")[0]
-        return self.cache_redis_url.split("redis://")[1].split("/")[0]
+        # Credentials must be stripped for every scheme: this value is rendered
+        # into the KEDA scaler's `address` metadata, so a password left in here
+        # both breaks the address and leaks into the ScaledObject.
+        authority = self.cache_redis_url.split("://", 1)[-1]
+        return authority.split("@")[-1].split("/")[0]
 
     @property
     def cache_redis_db_index(self) -> int:

model-engine/model_engine_server/core/celery/__init__.py

@@ -3,13 +3,15 @@
 from .app import (
     DEFAULT_TASK_VISIBILITY_SECONDS,
     TaskVisibility,
+    build_redis_url,
     celery_app,
     get_all_db_indexes,
     get_redis_host_port,
     inspect_app,
 )
 
 __all__: Sequence[str] = (
+    "build_redis_url",
     "celery_app",
     "get_all_db_indexes",
     "get_redis_host_port",

model-engine/model_engine_server/core/celery/app.py

@@ -2,6 +2,7 @@
 import os
 from enum import IntEnum, unique
 from typing import Any, Dict, Iterable, List, Optional, Tuple, Union
+from urllib.parse import quote
 
 import celery
 import redis.asyncio as aioredis
@@ -209,31 +210,53 @@ def get_redis_endpoint(db_index: int = 0) -> str:
             return f"{scheme}:{auth_token}@{host}:{port}/{db_index}{query_params}"
         return f"{scheme}{host}:{port}/{db_index}{query_params}"
     host, port = get_redis_host_port()
+    return build_redis_url(host, port, db_index)
+
+
+def redis_tls_enabled() -> bool:
+    """Whether to speak TLS to Redis.
+
+    Transport security and authentication are independent: on-prem Redis is
+    reached over plaintext while still requiring a password. REDIS_ENABLE_TLS
+    decides the scheme; when it is unset the presence of a credential implies
+    TLS, which is what ElastiCache with in-transit encryption needs.
+    """
+    explicit = os.getenv("REDIS_ENABLE_TLS")
+    if explicit:
+        return explicit.lower() == "true"
+    return bool(os.getenv("REDIS_AUTH_TOKEN"))
+
+
+def build_redis_url(host: str, port: Union[str, int], db_index: int = 0) -> str:
     auth_token = os.getenv("REDIS_AUTH_TOKEN")
-    if auth_token:
-        return f"rediss://:{auth_token}@{host}:{port}/{db_index}?ssl_cert_reqs=none"
-    return f"redis://{host}:{port}/{db_index}"
+    # kombu and redis-py both unquote the userinfo, so percent-encoding here is
+    # what lets a password containing @, / or # survive URL parsing.
+    credential = f":{quote(auth_token, safe='')}@" if auth_token else ""
+    if redis_tls_enabled():
+        return f"rediss://{credential}{host}:{port}/{db_index}?ssl_cert_reqs=none"
+    return f"redis://{credential}{host}:{port}/{db_index}"
 
 
 def get_redis_instance(db_index: int = 0) -> Union[Redis, StrictRedis]:
     host, port = get_redis_host_port()
     auth_token = os.getenv("REDIS_AUTH_TOKEN")
-
-    if auth_token:
-        return StrictRedis(
-            host=host,
-            port=port,
-            db=db_index,
-            password=auth_token,
-            ssl=True,
-            ssl_cert_reqs="none",
-        )
-    return Redis(host=host, port=port, db=db_index)
+    use_tls = redis_tls_enabled()
+
+    if not auth_token and not use_tls:
+        return Redis(host=host, port=port, db=db_index)
+    ssl_kwargs: Dict[str, Any] = {"ssl": True, "ssl_cert_reqs": "none"} if use_tls else {}
+    return StrictRedis(
+        host=host,
+        port=port,
+        db=db_index,
+        password=auth_token,
+        **ssl_kwargs,
+    )
 
 
 def get_async_redis_instance(db_index: int = 0) -> aioredis.Redis:
     host, port = get_redis_host_port()
-    return build_aioredis_client(f"redis://{host}:{port}/{db_index}")
+    return build_aioredis_client(build_redis_url(host, port, db_index))
 
 
 def celery_app(

model-engine/model_engine_server/core/celery/celery_autoscaler.py

@@ -25,6 +25,7 @@
 from model_engine_server.core.aws.roles import session
 from model_engine_server.core.celery import (
     TaskVisibility,
+    build_redis_url,
     celery_app,
     get_all_db_indexes,
     get_redis_host_port,
@@ -362,7 +363,7 @@ async def _init_client(self):
             get_redis_host_port()
         )  # Switches the redis instance based on CELERY_ELASTICACHE_ENABLED's value
         self.redis = {
-            db_index: build_aioredis_client(f"redis://{host}:{port}/{db_index}")
+            db_index: build_aioredis_client(build_redis_url(host, port, db_index))
             for db_index in get_all_db_indexes()
         }
         self.initialized = True