feat(auth): move JWT auth settings to DB-backed security settings

backend/alembic/versions/e7c00417d1e5_add_jwt_auth_columns_to_security_.py

@@ -0,0 +1,37 @@
+"""add jwt auth columns to security settings
+
+Revision ID: e7c00417d1e5
+Revises: 28bb08137807
+Create Date: 2026-08-21 12:27:35.945589
+
+"""
+
+import sqlalchemy as sa
+from alembic import op
+
+# revision identifiers, used by Alembic.
+revision = "e7c00417d1e5"
+down_revision = "28bb08137807"
+branch_labels = None
+depends_on = None
+
+
+def upgrade() -> None:
+    op.add_column(
+        "security_settings",
+        sa.Column("jwt_public_key_url", sa.String(), nullable=True),
+    )
+    op.add_column(
+        "security_settings",
+        sa.Column("jwt_expected_audience", sa.String(), nullable=True),
+    )
+    op.add_column(
+        "security_settings",
+        sa.Column("jwt_expected_issuer", sa.String(), nullable=True),
+    )
+
+
+def downgrade() -> None:
+    op.drop_column("security_settings", "jwt_expected_issuer")
+    op.drop_column("security_settings", "jwt_expected_audience")
+    op.drop_column("security_settings", "jwt_public_key_url")

backend/onyx/auth/jwt.py

@@ -16,12 +16,14 @@
 from jwt import decode as jwt_decode
 from jwt.algorithms import RSAAlgorithm  # ty: ignore[possibly-missing-import]
 
-from onyx.configs.app_configs import (
-    JWT_EXPECTED_AUDIENCE,
-    JWT_EXPECTED_ISSUER,
-    JWT_PUBLIC_KEY_URL,
+from onyx.auth.sso_url_guard import UnsafeSSOUrl, validate_idp_url
+from onyx.server.security.models import OutboundSSRFParams, outbound_ssrf_params
+from onyx.server.security.store import (
+    env_pinned_active_fields,
+    get_security_settings,
 )
 from onyx.utils.logger import setup_logger
+from onyx.utils.url import SSRFException, ssrf_safe_get
 
 logger = setup_logger()
 
@@ -34,17 +36,36 @@ class PublicKeyFormat(Enum):
     PEM = "pem"
 
 
-@lru_cache()
-def _fetch_public_key_payload() -> tuple[str | dict[str, Any], PublicKeyFormat] | None:
-    """Fetch and cache the raw JWT verification material."""
-    if JWT_PUBLIC_KEY_URL is None:
-        logger.error("JWT_PUBLIC_KEY_URL is not set")
-        return None
-
+# Keyed on the URL so a runtime settings change takes effect without a restart.
+@lru_cache(maxsize=8)
+def _fetch_public_key_payload(
+    public_key_url: str,
+    operator_pinned: bool,
+    allow_private_network: bool,
+    block_loopback_and_link_local: bool,
+    block_link_local_only: bool,
+) -> tuple[str | dict[str, Any], PublicKeyFormat] | None:
+    """Fetch and cache the raw JWT verification material. A DB-origin URL is
+    admin-aimed, so its fetch validates every redirect hop and pins the
+    resolved IP against DNS rebinding. An env-pinned URL is operator
+    config-as-code and fetched as-is."""
     try:
-        response = requests.get(JWT_PUBLIC_KEY_URL)
+        if operator_pinned:
+            response = requests.get(public_key_url)
+        else:
+            # Mirrors the PUT-time check: the configured SSRF level decides
+            # whether private endpoints are reachable.
+            # https_only holds across redirect hops, so no hop can downgrade
+            # the key fetch to plaintext.
+            response = ssrf_safe_get(
+                public_key_url,
+                allow_private_network=allow_private_network,
+                block_loopback_and_link_local=block_loopback_and_link_local,
+                block_link_local_only=block_link_local_only,
+                https_only=True,
+            )
         response.raise_for_status()
-    except requests.RequestException as exc:
+    except (requests.RequestException, SSRFException, ValueError) as exc:
         logger.error("Failed to fetch JWT public key: %s", str(exc))
         return None
     content_type = response.headers.get("Content-Type", "").lower()
@@ -74,9 +95,20 @@ def _fetch_public_key_payload() -> tuple[str | dict[str, Any], PublicKeyFormat]
     return body, PublicKeyFormat.PEM
 
 
-def get_public_key(token: str) -> RSAPublicKey | str | None:
+def get_public_key(
+    token: str,
+    public_key_url: str,
+    operator_pinned: bool,
+    ssrf_params: OutboundSSRFParams,
+) -> RSAPublicKey | str | None:
     """Return the concrete public key used to verify the provided JWT token."""
-    payload = _fetch_public_key_payload()
+    payload = _fetch_public_key_payload(
+        public_key_url,
+        operator_pinned,
+        ssrf_params.allow_private_network,
+        ssrf_params.block_loopback_and_link_local,
+        ssrf_params.block_link_local_only,
+    )
     if payload is None:
         logger.error("Failed to retrieve public key payload")
         return None
@@ -136,8 +168,28 @@ def _resolve_public_key_from_jwks(
 
 
 async def verify_jwt_token(token: str) -> dict[str, Any] | None:
+    settings = get_security_settings()
+    if settings.jwt_public_key_url is None:
+        logger.error("JWT public key URL is not configured")
+        return None
+
+    # A DB-origin URL is admin-aimed and must satisfy the outbound SSRF policy.
+    # An env-pinned value is operator config-as-code, trusted as before.
+    operator_pinned = "jwt_public_key_url" in env_pinned_active_fields()
+    if not operator_pinned:
+        try:
+            validate_idp_url(settings.jwt_public_key_url, field="jwt_public_key_url")
+        except UnsafeSSOUrl as e:
+            logger.error("JWT public key URL rejected: %s", e)
+            return None
+
     for attempt in range(_PUBLIC_KEY_FETCH_ATTEMPTS):
-        public_key = get_public_key(token)
+        public_key = get_public_key(
+            token,
+            settings.jwt_public_key_url,
+            operator_pinned,
+            outbound_ssrf_params(settings.ssrf_protection_level),
+        )
         if public_key is None:
             logger.error("Unable to resolve a public key for JWT verification")
             if attempt < _PUBLIC_KEY_FETCH_ATTEMPTS - 1:
@@ -152,9 +204,9 @@ async def verify_jwt_token(token: str) -> dict[str, Any] | None:
                 token,
                 public_key,
                 algorithms=["RS256"],
-                audience=JWT_EXPECTED_AUDIENCE,
-                issuer=JWT_EXPECTED_ISSUER,
-                options={"verify_aud": JWT_EXPECTED_AUDIENCE is not None},
+                audience=settings.jwt_expected_audience,
+                issuer=settings.jwt_expected_issuer,
+                options={"verify_aud": settings.jwt_expected_audience is not None},
             )
         except (
             InvalidAudienceError,

backend/onyx/auth/users.py

@@ -100,7 +100,6 @@
     DEV_MODE,
     EMAIL_CONFIGURED,
     INTEGRATION_TESTS_MODE,
-    JWT_PUBLIC_KEY_URL,
     REDIS_AUTH_KEY_PREFIX,
     REQUIRE_EMAIL_VERIFICATION,
     SESSION_EXPIRE_TIME_SECONDS,
@@ -2058,7 +2057,7 @@ async def _check_for_saml_and_jwt(
     async_db_session: AsyncSession,
 ) -> User | None:
     # If user is None, check for JWT in Authorization header
-    if user is None and JWT_PUBLIC_KEY_URL is not None:
+    if user is None and get_security_settings().jwt_public_key_url is not None:
         auth_header = request.headers.get("Authorization")
         if auth_header and auth_header.startswith("Bearer "):
             token = auth_header[len("Bearer ") :].strip()

backend/onyx/db/models.py

@@ -4794,6 +4794,15 @@ class SecuritySettings(Base):
     password_require_special_char: Mapped[bool | None] = mapped_column(
         Boolean, nullable=True, default=None
     )
+    jwt_public_key_url: Mapped[str | None] = mapped_column(
+        String, nullable=True, default=None
+    )
+    jwt_expected_audience: Mapped[str | None] = mapped_column(
+        String, nullable=True, default=None
+    )
+    jwt_expected_issuer: Mapped[str | None] = mapped_column(
+        String, nullable=True, default=None
+    )
     __table_args__ = (
         CheckConstraint("id = true", name="ck_security_settings_singleton"),
         # Only catches min > max when both are explicitly overridden; the

backend/onyx/main.py

@@ -157,6 +157,7 @@
 from onyx.server.query_and_chat.query_backend import basic_router as query_router
 from onyx.server.saml_multi import router as saml_multi_router
 from onyx.server.security.api import admin_router as security_admin_router
+from onyx.server.security.store import seed_jwt_settings_from_env
 from onyx.server.settings.api import admin_router as settings_admin_router
 from onyx.server.settings.api import basic_router as settings_router
 from onyx.server.sso_discovery import router as sso_discovery_router
@@ -406,6 +407,8 @@ async def lifespan(app: FastAPI) -> AsyncGenerator[None, None]:  # noqa: ARG001
             # api_server has the mount the migration job lacks, so this is where it
             # runs. No-op unless AUTH_TYPE=saml with no SAML row yet.
             seed_saml_provider_from_conf_dir(db_session)
+            # No-op when env is unset or the row already matches.
+            seed_jwt_settings_from_env()
             # set up the file store (e.g. create bucket if needed). On multi-tenant,
             # this is done via IaC
             get_default_file_store().initialize()

backend/onyx/server/security/api.py

@@ -6,6 +6,7 @@
 from pydantic import ValidationError
 
 from onyx.auth.permissions import require_permission
+from onyx.auth.sso_url_guard import UnsafeSSOUrl, validate_idp_url
 from onyx.db.enums import Permission
 from onyx.db.models import User
 from onyx.error_handling.error_codes import OnyxErrorCode
@@ -15,7 +16,12 @@
     SecuritySettings,
     SecuritySettingsOverrides,
 )
-from onyx.server.security.store import apply_patch, get_security_settings
+from onyx.server.security.store import (
+    apply_patch,
+    env_pinned_active_fields,
+    get_security_settings,
+)
+from onyx.utils.audit import AuditActor
 from onyx.utils.logger import setup_logger
 from shared_configs.configs import MULTI_TENANT
 
@@ -61,11 +67,27 @@ def get_security_settings_endpoint(
 @admin_router.put("")
 async def put_security_settings_endpoint(
     request: Request,
-    _: User = Depends(require_permission(Permission.FULL_ADMIN_PANEL_ACCESS)),
+    user: User = Depends(require_permission(Permission.FULL_ADMIN_PANEL_ACCESS)),
 ) -> SecuritySettings:
     raw = await request.body()
     overrides, present_keys = _parse_put_body(raw)
 
+    if "jwt_public_key_url" in present_keys and overrides.jwt_public_key_url:
+        try:
+            validate_idp_url(overrides.jwt_public_key_url, field="jwt_public_key_url")
+        except UnsafeSSOUrl as e:
+            raise OnyxError(OnyxErrorCode.INVALID_INPUT, str(e))
+
+    # A clear refusal instead of silently storing an override the env pin
+    # would render inert.
+    pinned_in_payload = present_keys & env_pinned_active_fields()
+    if pinned_in_payload:
+        raise OnyxError(
+            OnyxErrorCode.INSUFFICIENT_PERMISSIONS,
+            "These fields are pinned by environment variables on this deployment: "
+            + ", ".join(sorted(pinned_in_payload)),
+        )
+
     lockdown_in_payload = present_keys & _PASSWORD_LOCKDOWN_FIELDS
     if lockdown_in_payload and MULTI_TENANT:
         raise OnyxError(
@@ -85,5 +107,9 @@ async def put_security_settings_endpoint(
                 + ", ".join(sorted(locked_in_payload)),
             )
 
+    # The auth dependency always yields a user, test seams may not.
+    actor = (
+        AuditActor(user_id=str(user.id), email=user.email) if user is not None else None
+    )
     # Sync DB + Redis IO — offload so the event loop isn't blocked.
-    return await run_in_threadpool(apply_patch, overrides, present_keys)
+    return await run_in_threadpool(apply_patch, overrides, present_keys, actor)

backend/onyx/server/security/models.py

@@ -83,6 +83,7 @@ def web_connector_ssrf_enforced(level: SSRFProtectionLevel) -> bool:
 
 
 _OPERATOR_LOCKED_MARKER = "operator_locked"
+_ENV_PINNED_MARKER = "env_pinned"
 
 
 def _operator_locked() -> dict[str, bool]:
@@ -95,6 +96,14 @@ def _tenant_editable() -> dict[str, bool]:
     return {_OPERATOR_LOCKED_MARKER: False}
 
 
+def _env_pinned() -> dict[str, bool]:
+    """Field marker: a set env var wins over any stored override, so infra can
+    keep auth-gating values as reviewable config-as-code that an app-level
+    admin compromise cannot flip. Also operator-locked: never tenant-editable
+    in multi-tenant, and single-tenant editable only while env is unset."""
+    return {_OPERATOR_LOCKED_MARKER: True, _ENV_PINNED_MARKER: True}
+
+
 class IncognitoAvailability(str, Enum):
     """Who may start incognito chats. Secure default is OFF: the feature is
     invisible until an admin turns it on."""
@@ -161,6 +170,15 @@ class SecuritySettingsOverrides(BaseModel):
     password_auth_enabled: bool | None = Field(
         default=None, json_schema_extra=_operator_locked()
     )
+    jwt_public_key_url: str | None = Field(
+        default=None, json_schema_extra=_env_pinned()
+    )
+    jwt_expected_audience: str | None = Field(
+        default=None, json_schema_extra=_env_pinned()
+    )
+    jwt_expected_issuer: str | None = Field(
+        default=None, json_schema_extra=_env_pinned()
+    )
 
     @field_validator("valid_email_domains")
     @classmethod
@@ -199,6 +217,18 @@ def _derive_operator_locked_fields() -> frozenset[str]:
 OPERATOR_LOCKED_FIELDS: frozenset[str] = _derive_operator_locked_fields()
 
 
+def _derive_env_pinned_fields() -> frozenset[str]:
+    return frozenset(
+        name
+        for name, info in SecuritySettingsOverrides.model_fields.items()
+        if isinstance(info.json_schema_extra, dict)
+        and info.json_schema_extra.get(_ENV_PINNED_MARKER)
+    )
+
+
+ENV_PINNED_FIELDS: frozenset[str] = _derive_env_pinned_fields()
+
+
 class SecuritySettings(BaseModel):
     """Effective, env-merged, immutable security settings."""
 
@@ -219,6 +249,10 @@ class SecuritySettings(BaseModel):
     password_require_digit: bool
     password_require_special_char: bool
     password_auth_enabled: bool
+    # None disables JWT bearer auth / the corresponding claim check entirely.
+    jwt_public_key_url: str | None
+    jwt_expected_audience: str | None
+    jwt_expected_issuer: str | None
 
     @model_validator(mode="after")
     def _check_password_length_invariants(self) -> Self: