[iris] Add optional auth mode for gradual adoption (#3937)

Add optional field to AuthConfig proto. When set, authentication is
attempted but not required -- requests with valid tokens get the
authenticated identity while unauthenticated requests fall through as
anonymous/admin. This enables gradual rollout of JWT auth on clusters
that currently rely on SSH tunnels. The dashboard /auth/config endpoint
reports the optional flag so the frontend can adapt its login UI
accordingly.

Fixes #3936

Author: claude[bot]

lib/fray/src/fray/v2/iris_backend.py

@@ -508,6 +508,7 @@ def submit(self, request: JobRequest, adopt_existing: bool = True) -> IrisJobHan
         replicas = request.replicas or 1
         coscheduling = resolve_coscheduling(request.resources.device, replicas)
 
+        policy = cluster_pb2.EXISTING_JOB_POLICY_KEEP if adopt_existing else cluster_pb2.EXISTING_JOB_POLICY_UNSPECIFIED
         try:
             job = self._iris.submit(
                 entrypoint=iris_entrypoint,
@@ -519,12 +520,10 @@ def submit(self, request: JobRequest, adopt_existing: bool = True) -> IrisJobHan
                 replicas=replicas,
                 max_retries_failure=request.max_retries_failure,
                 max_retries_preemption=request.max_retries_preemption,
+                existing_job_policy=policy,
             )
         except IrisJobAlreadyExists as e:
-            if adopt_existing:
-                logger.info("Job %s already exists, adopting existing job", request.name)
-                return IrisJobHandle(e.job)
-            raise FrayJobAlreadyExists(request.name, handle=IrisJobHandle(e.job)) from e
+            raise FrayJobAlreadyExists(request.name) from e
         return IrisJobHandle(job)
 
     def host_actor(

lib/iris/dashboard/src/App.vue

@@ -75,19 +75,21 @@ onMounted(async () => {
   window.addEventListener('iris-auth-required', onAuthRequired)
 
   let hasSession = false
+  let authOptional = false
   try {
     const resp = await fetch('/auth/config')
     if (resp.ok) {
       const config = await resp.json()
       authEnabled.value = config.auth_enabled ?? false
       hasSession = config.has_session ?? false
+      authOptional = config.optional ?? false
       providerKind.value = config.provider_kind === 'kubernetes' ? 'kubernetes' : 'worker'
     }
   } catch {
     // Auth config endpoint unavailable — assume no auth
   }
 
-  if (authEnabled.value && !hasSession && route.path !== '/login') {
+  if (authEnabled.value && !authOptional && !hasSession && route.path !== '/login') {
     router.push('/login')
   }
 })

lib/iris/examples/local-auth-gcp.yaml

@@ -16,6 +16,7 @@ auth:
     project_id: hai-gcp-models  # GCP project ID — users must have access to log in
   admin_users:
     - russell.power@gmail.com  # Replace with actual admin email
+  optional: true
 
 scale_groups:
   cpu:

lib/iris/src/iris/client/client.py

@@ -132,8 +132,7 @@ def __init__(self, job_id: JobName, status: cluster_pb2.JobStatus):
 class JobAlreadyExists(Exception):
     """Raised when a job with the same name is already running."""
 
-    def __init__(self, job: "Job", message: str):
-        self.job = job
+    def __init__(self, message: str):
         super().__init__(message)
 
 
@@ -704,7 +703,7 @@ def submit(
             )
 
         try:
-            self._cluster_client.submit_job(
+            canonical_id = self._cluster_client.submit_job(
                 job_id=job_id,
                 entrypoint=entrypoint,
                 resources=resources_proto,
@@ -723,10 +722,10 @@ def submit(
             )
         except ConnectError as e:
             if e.code == Code.ALREADY_EXISTS:
-                raise JobAlreadyExists(Job(self, job_id), str(e)) from e
+                raise JobAlreadyExists(str(e)) from e
             raise
 
-        return Job(self, job_id)
+        return Job(self, canonical_id)
 
     def status(self, job_id: JobName) -> cluster_pb2.JobStatus:
         """Get job status.

lib/iris/src/iris/cluster/client/protocol.py

@@ -55,7 +55,7 @@ def submit_job(
         reservation: cluster_pb2.ReservationConfig | None = None,
         preemption_policy: cluster_pb2.JobPreemptionPolicy = cluster_pb2.JOB_PREEMPTION_POLICY_UNSPECIFIED,
         existing_job_policy: cluster_pb2.ExistingJobPolicy = cluster_pb2.EXISTING_JOB_POLICY_UNSPECIFIED,
-    ) -> None: ...
+    ) -> JobName: ...
 
     def get_job_status(self, job_id: JobName) -> cluster_pb2.JobStatus: ...
 

lib/iris/src/iris/cluster/client/remote_client.py

@@ -73,7 +73,7 @@ def submit_job(
         reservation: cluster_pb2.ReservationConfig | None = None,
         preemption_policy: cluster_pb2.JobPreemptionPolicy = cluster_pb2.JOB_PREEMPTION_POLICY_UNSPECIFIED,
         existing_job_policy: cluster_pb2.ExistingJobPolicy = cluster_pb2.EXISTING_JOB_POLICY_UNSPECIFIED,
-    ) -> None:
+    ) -> JobName:
         if replicas < 1:
             raise ValueError(f"replicas must be >= 1, got {replicas}")
         replicas = adjust_tpu_replicas(resources.device if resources.HasField("device") else None, replicas)
@@ -112,9 +112,10 @@ def submit_job(
             request.reservation.CopyFrom(reservation)
 
         def _call():
-            self._client.launch_job(request)
+            return self._client.launch_job(request)
 
-        call_with_retry(f"launch_job({job_id})", _call)
+        response = call_with_retry(f"launch_job({job_id})", _call)
+        return JobName.from_wire(response.job_id)
 
     def get_job_status(self, job_id: JobName) -> cluster_pb2.JobStatus:
         def _call():

lib/iris/src/iris/cluster/controller/auth.py

@@ -259,6 +259,7 @@ class ControllerAuth:
     login_verifier: TokenVerifier | None = None
     gcp_project_id: str | None = None
     jwt_manager: JwtTokenManager | None = None
+    optional: bool = False
 
 
 def create_controller_auth(
@@ -328,14 +329,22 @@ def create_controller_auth(
         static_tokens = dict(auth_config.static.tokens)
         login_verifier = StaticTokenVerifier(static_tokens)
 
-    logger.info("Auth enabled: provider=%s, db=%s, jwt=%s", provider, "yes" if db else "no", "yes" if jwt_mgr else "no")
+    optional = auth_config.optional
+    logger.info(
+        "Auth enabled: provider=%s, db=%s, jwt=%s, optional=%s",
+        provider,
+        "yes" if db else "no",
+        "yes" if jwt_mgr else "no",
+        optional,
+    )
     return ControllerAuth(
         verifier=verifier,
         provider=provider,
         worker_token=worker_token,
         login_verifier=login_verifier,
         gcp_project_id=gcp_project_id,
         jwt_manager=jwt_mgr,
+        optional=optional,
     )
 
 

lib/iris/src/iris/cluster/controller/controller.py

@@ -760,6 +760,7 @@ def __init__(
             port=config.port,
             auth_verifier=config.auth_verifier,
             auth_provider=config.auth_provider,
+            auth_optional=config.auth.optional if config.auth else False,
         )
 
         # Ingest process logs into the LogStore so they are available via FetchLogs.

lib/iris/src/iris/cluster/controller/dashboard.py

@@ -39,7 +39,7 @@
 
 from iris.cluster.controller.service import ControllerServiceImpl
 from iris.cluster.dashboard_common import html_shell, static_files_mount
-from iris.rpc.auth import SESSION_COOKIE, AuthInterceptor, NullAuthInterceptor, TokenVerifier
+from iris.rpc.auth import SESSION_COOKIE, NullAuthInterceptor, TokenVerifier, extract_bearer_token, resolve_auth
 from iris.rpc.cluster_connect import ControllerServiceWSGIApplication
 from iris.rpc.interceptors import RequestTimingInterceptor
 
@@ -86,11 +86,15 @@ class _RouteAuthMiddleware:
     @public / @requires_auth annotation. Routes without an annotation are
     denied (default-deny). RPC Mount routes and static file mounts are
     skipped (they have their own auth).
+
+    Uses resolve_auth() — the same policy function as the gRPC interceptor —
+    so HTTP and gRPC layers agree on allow/deny for every token state.
     """
 
-    def __init__(self, app: Starlette, verifier: TokenVerifier):
+    def __init__(self, app: Starlette, verifier: TokenVerifier, optional: bool = False):
         self._app = app
         self._verifier = verifier
+        self._optional = optional
         self._router = app.router
 
     async def __call__(self, scope: Scope, receive: Receive, send: Send) -> None:
@@ -134,15 +138,13 @@ def _resolve_policy(self, scope: Scope) -> str:
 
     async def _check_auth(self, scope: Scope, receive: Receive, send: Send) -> None:
         token = _extract_token_from_scope(scope)
-        if token is None:
-            response = JSONResponse({"error": "authentication required"}, status_code=401)
-            return await response(scope, receive, send)
         try:
-            identity = self._verifier.verify(token)
+            identity = resolve_auth(token, self._verifier, self._optional)
         except ValueError:
-            response = JSONResponse({"error": "invalid session"}, status_code=401)
+            response = JSONResponse({"error": "authentication required"}, status_code=401)
             return await response(scope, receive, send)
-        scope["auth_identity"] = identity
+        if identity is not None:
+            scope["auth_identity"] = identity
         return await self._app(scope, receive, send)
 
 
@@ -168,16 +170,49 @@ def _check_csrf(request: Request) -> bool:
     return origin == expected_origin
 
 
-class _SelectiveAuthInterceptor:
-    """Auth interceptor that skips authentication for specific RPC methods."""
+class _DashboardAuthInterceptor:
+    """RPC auth interceptor that uses resolve_auth() — same policy as HTTP middleware.
 
-    def __init__(self, verifier: TokenVerifier):
-        self._inner = AuthInterceptor(verifier)
+    Login and GetAuthInfo RPCs are always unauthenticated. All other RPCs go
+    through resolve_auth(token, verifier, optional) which:
+    - token present + valid → authenticated identity
+    - token present + invalid → rejected
+    - no token + optional → anonymous/admin fallback via NullAuthInterceptor
+    - no token + required → rejected
+    """
+
+    def __init__(self, verifier: TokenVerifier, optional: bool = False):
+        self._verifier = verifier
+        self._optional = optional
+        self._null = NullAuthInterceptor(verifier=verifier)
 
     def intercept_unary_sync(self, call_next, request, ctx):
+        from iris.rpc.auth import _verified_identity
+
         if ctx.method().name in _UNAUTHENTICATED_RPCS:
             return call_next(request, ctx)
-        return self._inner.intercept_unary_sync(call_next, request, ctx)
+
+        token = extract_bearer_token(ctx.request_headers())
+        try:
+            identity = resolve_auth(token, self._verifier, self._optional)
+        except ValueError as exc:
+            from connectrpc.code import Code
+            from connectrpc.errors import ConnectError
+
+            if token is None:
+                raise ConnectError(Code.UNAUTHENTICATED, str(exc)) from exc
+            logger.warning("Authentication failed: %s", exc)
+            raise ConnectError(Code.UNAUTHENTICATED, "Authentication failed") from exc
+
+        if identity is None:
+            # Optional mode, no token — anonymous fallback.
+            return self._null.intercept_unary_sync(call_next, request, ctx)
+
+        reset_token = _verified_identity.set(identity)
+        try:
+            return call_next(request, ctx)
+        finally:
+            _verified_identity.reset(reset_token)
 
 
 class ControllerDashboard:
@@ -196,12 +231,14 @@ def __init__(
         port: int = 8080,
         auth_verifier: TokenVerifier | None = None,
         auth_provider: str | None = None,
+        auth_optional: bool = False,
     ):
         self._service = service
         self._host = host
         self._port = port
         self._auth_verifier = auth_verifier
         self._auth_provider = auth_provider
+        self._auth_optional = auth_optional
         self._app = self._create_app()
 
     @property
@@ -214,9 +251,11 @@ def app(self) -> ASGIApp:
 
     def _create_app(self) -> ASGIApp:
         interceptors = [RequestTimingInterceptor(include_traceback=bool(os.environ.get("IRIS_DEBUG")))]
-        if self._auth_provider is not None:
-            interceptors.insert(0, _SelectiveAuthInterceptor(self._auth_verifier))
+        if self._auth_provider is not None and self._auth_verifier is not None:
+            interceptors.insert(0, _DashboardAuthInterceptor(self._auth_verifier, optional=self._auth_optional))
         else:
+            # Null-auth mode: no provider configured. Verify worker tokens
+            # when present but treat everything as anonymous/admin.
             interceptors.insert(0, NullAuthInterceptor(verifier=self._auth_verifier))
         rpc_wsgi_app = ControllerServiceWSGIApplication(service=self._service, interceptors=interceptors)
         rpc_app = WSGIMiddleware(rpc_wsgi_app)
@@ -236,7 +275,7 @@ def _create_app(self) -> ASGIApp:
         ]
         app: Starlette | _RouteAuthMiddleware = Starlette(routes=routes)
         if self._auth_verifier is not None and self._auth_provider is not None:
-            app = _RouteAuthMiddleware(app, self._auth_verifier)
+            app = _RouteAuthMiddleware(app, self._auth_verifier, optional=self._auth_optional)
         return app
 
     @public
@@ -283,6 +322,7 @@ def _auth_config(self, request: Request) -> JSONResponse:
                 "provider": self._auth_provider,
                 "has_session": has_session,
                 "provider_kind": provider_kind,
+                "optional": self._auth_optional,
             }
         )
 

lib/iris/src/iris/rpc/auth.py

@@ -217,6 +217,23 @@ def verify(self, token: str) -> VerifiedIdentity:
         raise ValueError(f"All verifiers failed: {'; '.join(errors)}")
 
 
+def resolve_auth(
+    token: str | None,
+    verifier: TokenVerifier,
+    optional: bool,
+) -> VerifiedIdentity | None:
+    """Shared auth policy for gRPC interceptors and HTTP middleware.
+
+    Returns VerifiedIdentity on success, None for anonymous passthrough.
+    Raises ValueError on rejected tokens (invalid token, or missing when required).
+    """
+    if token is None:
+        if optional:
+            return None
+        raise ValueError("Missing authentication")
+    return verifier.verify(token)
+
+
 class AuthInterceptor:
     """Server-side Connect RPC interceptor that enforces bearer token auth.