[BREAKING] Python: DevUI: tighten default access controls and CORS posture

python/packages/devui/agent_framework_devui/__init__.py

@@ -96,7 +96,7 @@ def serve(
     ui_enabled: bool = True,
     instrumentation_enabled: bool = False,
     mode: str = "developer",
-    auth_enabled: bool = False,
+    auth_enabled: bool = True,
     auth_token: str | None = None,
 ) -> None:
     """Launch Agent Framework DevUI with simple API.
@@ -126,52 +126,29 @@ def serve(
     if not isinstance(port, int) or not (1 <= port <= 65535):
         raise ValueError(f"Invalid port: {port}. Must be integer between 1 and 65535")
 
-    # Security check: Warn if network-exposed without authentication
+    # Security check: warn loudly when network-exposed without authentication.
     if host not in ("127.0.0.1", "localhost") and not auth_enabled:
-        logger.warning("⚠️  WARNING: Exposing DevUI to network without authentication!")
-        logger.warning("⚠️  This is INSECURE - anyone on your network can access your agents")
-        logger.warning("💡 For network exposure, add --auth flag: devui --host 0.0.0.0 --auth")
+        logger.warning("WARNING: Exposing DevUI to the network with --no-auth.")
+        logger.warning("Anyone on your network can read agent metadata and trigger requests.")
+        logger.warning("Drop --no-auth and DevUI will require Bearer tokens.")
 
-    # Handle authentication configuration
-    if auth_enabled:
+    # Refuse to auto-generate a token for network-exposed binds. Auto-generated tokens
+    # are fine for localhost convenience; for anything else, require an explicit token.
+    if auth_enabled and not auth_token:
         import os
-        import secrets
 
-        # Check if token is in environment variable first
-        if not auth_token:
-            auth_token = os.environ.get("DEVUI_AUTH_TOKEN")
-
-        # Auto-generate token if STILL not provided
-        if not auth_token:
-            # Check if we're in a production-like environment
+        env_token = os.environ.get("DEVUI_AUTH_TOKEN")
+        if not env_token:
             is_production = (
-                host not in ("127.0.0.1", "localhost")  # Exposed to network
-                or os.environ.get("CI") == "true"  # Running in CI
-                or os.environ.get("KUBERNETES_SERVICE_HOST")  # Running in k8s
+                host not in ("127.0.0.1", "localhost")
+                or os.environ.get("CI") == "true"
+                or os.environ.get("KUBERNETES_SERVICE_HOST")
             )
-
             if is_production:
-                # REFUSE to start without explicit token
-                logger.error("❌ Authentication enabled but no token provided")
-                logger.error("❌ Auto-generated tokens are NOT secure for network-exposed deployments")
-                logger.error("💡 Set token: export DEVUI_AUTH_TOKEN=<your-secure-token>")
-                logger.error("💡 Or pass: serve(entities=[...], auth_token='your-token')")
+                logger.error("Authentication required but no token provided.")
+                logger.error("Set DEVUI_AUTH_TOKEN env var or pass auth_token='...' to serve().")
                 raise ValueError("DEVUI_AUTH_TOKEN required when host is not localhost")
 
-            # Development mode: auto-generate and show
-            auth_token = secrets.token_urlsafe(32)
-            logger.info("🔒 Authentication enabled with auto-generated token")
-            logger.info("\n" + "=" * 70)
-            logger.info("🔑 DEV TOKEN (localhost only, shown once):")
-            logger.info(f"   {auth_token}")
-            logger.info("=" * 70 + "\n")
-        else:
-            logger.info("🔒 Authentication enabled with provided token")
-
-        # Set environment variable for server to use
-        os.environ["AUTH_REQUIRED"] = "true"
-        os.environ["DEVUI_AUTH_TOKEN"] = auth_token
-
     # Enable instrumentation if requested
     if instrumentation_enabled:
         from agent_framework.observability import enable_instrumentation
@@ -187,6 +164,8 @@ def serve(
         cors_origins=cors_origins,
         ui_enabled=ui_enabled,
         mode=mode,
+        auth_enabled=auth_enabled,
+        auth_token=auth_token,
     )
 
     # Register in-memory entities if provided

python/packages/devui/agent_framework_devui/_cli.py

@@ -79,15 +79,15 @@ def create_cli_parser() -> argparse.ArgumentParser:
     )
 
     parser.add_argument(
-        "--auth",
+        "--no-auth",
         action="store_true",
-        help="Enable authentication via Bearer token (required for deployed environments)",
+        help="Disable Bearer token authentication. DevUI is auth-enabled by default; use this to opt out.",
     )
 
     parser.add_argument(
         "--auth-token",
         type=str,
-        help="Custom authentication token (auto-generated if not provided with --auth)",
+        help="Custom Bearer token. Auto-generated and logged at startup when omitted.",
     )
 
     parser.add_argument("--version", action="version", version=f"Agent Framework DevUI {get_version()}")
@@ -184,7 +184,7 @@ def main() -> None:
             ui_enabled=ui_enabled,
             instrumentation_enabled=args.instrumentation,
             mode=mode,
-            auth_enabled=args.auth,
+            auth_enabled=not args.no_auth,
             auth_token=args.auth_token,  # Pass through explicit token only
         )
 

python/packages/devui/agent_framework_devui/_server.py

@@ -75,6 +75,8 @@ def __init__(
         cors_origins: list[str] | None = None,
         ui_enabled: bool = True,
         mode: str = "developer",
+        auth_enabled: bool = True,
+        auth_token: str | None = None,
     ) -> None:
         """Initialize the development server.
 
@@ -85,20 +87,26 @@ def __init__(
             cors_origins: List of allowed CORS origins
             ui_enabled: Whether to enable the UI
             mode: Server mode - 'developer' (full access, verbose errors) or 'user' (restricted APIs, generic errors)
+            auth_enabled: Whether to require Bearer token auth on /v1/* endpoints. Defaults to True.
+            auth_token: Bearer token. If None and auth_enabled, falls back to the DEVUI_AUTH_TOKEN
+                environment variable, then to an auto-generated token (logged at startup).
         """
         self.entities_dir = entities_dir
         self.port = port
         self.host = host
 
-        # Smart CORS defaults: permissive for localhost, restrictive for network-exposed deployments
+        # CORS default is same-origin only (empty allowlist) on every host. The
+        # previous wildcard-on-localhost default let any webpage the developer
+        # visited read DevUI's responses cross-origin. Callers who need a real
+        # cross-origin dev frontend pass an explicit allowlist.
         if cors_origins is None:
-            # Localhost development: allow cross-origin for dev tools (e.g., frontend dev server)
-            # Network-exposed: empty list (same-origin only, no CORS)
-            cors_origins = ["*"] if host in ("127.0.0.1", "localhost") else []
+            cors_origins = []
 
         self.cors_origins = cors_origins
         self.ui_enabled = ui_enabled
         self.mode = mode
+        self.auth_enabled = auth_enabled
+        self.auth_token = self._resolve_auth_token(auth_enabled, auth_token)
         self.executor: AgentFrameworkExecutor | None = None
         self.openai_executor: OpenAIExecutor | None = None
         self.deployment_manager = DeploymentManager()
@@ -110,6 +118,37 @@ def set_pending_entities(self, entities: list[Any]) -> None:
         """Set in-memory entities to register on startup."""
         self._pending_entities = entities
 
+    _LOOPBACK_HOSTS = frozenset({"127.0.0.1", "localhost", "[::1]", "::1"})
+
+    def _loopback_allowed_hosts(self) -> frozenset[str] | None:
+        """Return the Host-header allowlist when bound to a loopback interface, else None.
+
+        Returning None disables Host-header enforcement (e.g. for 0.0.0.0 / public binds,
+        where the operator is intentionally exposing the service).
+        """
+        host = self.host.lower()
+        if host not in self._LOOPBACK_HOSTS:
+            return None
+        return self._LOOPBACK_HOSTS
+
+    @staticmethod
+    def _resolve_auth_token(auth_enabled: bool, auth_token: str | None) -> str | None:
+        """Resolve the active Bearer token. Returns None when auth is disabled."""
+        if not auth_enabled:
+            return None
+        if auth_token:
+            return auth_token
+        env_token = os.getenv("DEVUI_AUTH_TOKEN")
+        if env_token:
+            return env_token
+        generated = secrets.token_urlsafe(32)
+        logger.info("=" * 70)
+        logger.info("DevUI authentication enabled with auto-generated token:")
+        logger.info(f"   {generated}")
+        logger.info("Pass it as: Authorization: Bearer <token>")
+        logger.info("=" * 70)
+        return generated
+
     def _is_dev_mode(self) -> bool:
         """Check if running in developer mode.
 
@@ -336,6 +375,11 @@ async def lifespan(app: FastAPI) -> AsyncGenerator[None]:
             lifespan=lifespan,
         )
 
+        # Middleware registration order matters: Starlette wraps later-added
+        # middleware around earlier-added ones, so the LAST registered runs
+        # outermost (sees the request first). We want Host-header enforcement
+        # to run before CORS/auth, so it is registered last below.
+
         # Add CORS middleware
         # Note: allow_credentials cannot be True when allow_origins is ["*"]
         # For localhost dev with wildcard origins, credentials are disabled
@@ -350,29 +394,24 @@ async def lifespan(app: FastAPI) -> AsyncGenerator[None]:
             allow_headers=["*"],
         )
 
-        # Add authentication middleware using decorator pattern
-        # Auth is enabled by presence of DEVUI_AUTH_TOKEN
-        auth_token = os.getenv("DEVUI_AUTH_TOKEN", "")
-        auth_required = bool(auth_token)
-
-        if auth_required:
+        # Bearer-token authentication. Enabled by default; opt out via
+        # DevServer(auth_enabled=False) for embedded/test scenarios.
+        if self.auth_enabled and self.auth_token:
+            expected_token = self.auth_token
             logger.info("Authentication middleware enabled")
 
             @app.middleware("http")
             async def auth_middleware(request: Request, call_next: Callable[[Request], Awaitable[Any]]) -> Any:
                 """Validate Bearer token authentication.
 
-                Skips authentication for health, meta, static UI endpoints, and OPTIONS requests.
+                Skips authentication for health, the UI shell, static assets, and OPTIONS preflight.
                 """
-                # Skip auth for OPTIONS (CORS preflight) requests
                 if request.method == "OPTIONS":
                     return await call_next(request)
 
-                # Skip auth for health checks, meta endpoint, and static files
-                if request.url.path in ["/health", "/meta", "/"] or request.url.path.startswith("/assets"):
+                if request.url.path in ["/health", "/"] or request.url.path.startswith("/assets"):
                     return await call_next(request)
 
-                # Check Authorization header
                 auth_header = request.headers.get("Authorization")
                 if not auth_header or not auth_header.startswith("Bearer "):
                     return JSONResponse(
@@ -388,9 +427,8 @@ async def auth_middleware(request: Request, call_next: Callable[[Request], Await
                         },
                     )
 
-                # Extract and validate token
                 token = auth_header.replace("Bearer ", "", 1).strip()
-                if not secrets.compare_digest(token, auth_token):
+                if not secrets.compare_digest(token, expected_token):
                     return JSONResponse(
                         status_code=401,
                         content={
@@ -402,11 +440,40 @@ async def auth_middleware(request: Request, call_next: Callable[[Request], Await
                         },
                     )
 
-                # Token valid, proceed
                 return await call_next(request)
 
             _ = auth_middleware
 
+        # Host-header allowlist for loopback binds: on a loopback interface, only
+        # accept requests whose Host header names a loopback address. Registered LAST
+        # so it runs outermost, rejecting non-loopback Host values before CORS/auth
+        # (and before CORS can short-circuit a preflight on a rebound request).
+        allowed_hosts = self._loopback_allowed_hosts()
+        if allowed_hosts is not None:
+            expected_hosts = allowed_hosts
+
+            @app.middleware("http")
+            async def host_header_middleware(request: Request, call_next: Callable[[Request], Awaitable[Any]]) -> Any:
+                host_header = request.headers.get("host", "")
+                hostname = host_header.split(":", 1)[0].lower()
+                if hostname and hostname not in expected_hosts:
+                    return JSONResponse(
+                        status_code=400,
+                        content={
+                            "error": {
+                                "message": (
+                                    f"Invalid Host header '{host_header}'. DevUI is bound to a "
+                                    "loopback interface and only accepts requests addressed to it."
+                                ),
+                                "type": "invalid_host",
+                                "code": "host_not_allowed",
+                            }
+                        },
+                    )
+                return await call_next(request)
+
+            _ = host_header_middleware
+
         self._register_routes(app)
         self._mount_ui(app)
 
@@ -427,8 +494,6 @@ async def health_check() -> dict[str, Any]:
         @app.get("/meta", response_model=MetaResponse)
         async def get_meta() -> MetaResponse:
             """Get server metadata and configuration."""
-            import os
-
             # Ensure executors are initialized to check capabilities
             openai_executor = await self._ensure_openai_executor()
 
@@ -442,7 +507,7 @@ async def get_meta() -> MetaResponse:
                     "openai_proxy": openai_executor.is_configured,
                     "deployment": True,  # Deployment feature is available
                 },
-                auth_required=bool(os.getenv("DEVUI_AUTH_TOKEN")),
+                auth_required=self.auth_enabled,
             )
 
         @app.get("/v1/entities", response_model=DiscoveryResponse)
@@ -750,7 +815,6 @@ async def create_response(request: AgentFrameworkRequest, raw_request: Request)
                             headers={
                                 "Cache-Control": "no-cache",
                                 "Connection": "keep-alive",
-                                "Access-Control-Allow-Origin": "*",
                             },
                         )
                     return await openai_executor.execute_sync(request)
@@ -794,7 +858,6 @@ async def create_response(request: AgentFrameworkRequest, raw_request: Request)
                         headers={
                             "Cache-Control": "no-cache",
                             "Connection": "keep-alive",
-                            "Access-Control-Allow-Origin": "*",
                             "X-Response-ID": response_id,  # Include ID for debugging/tracking
                         },
                     )