security: fix critical auth vulnerabilities in bind-aware gateway

- Fix fail-open auth in token mode (now fails closed when expected_token is None)
- Fix token file permissions race condition (create with 0o600 atomically)
- Fix stale config validation (align bind_host with actual server host)
- Add IPv6 bracket handling and remove non-standard 'local' alias
- Fix configured auth mode being ignored in assert_external_bind_safe
- Add empty credentials rejection on external interfaces
- Use constant-time comparison for password checks
- Fix test reliability issues (exception swallowing in async tests)
- Fix GatewayConfig hardcoded bind_host (now defaults to host)
- Update test mocking to work with local chainlit imports

Addresses security vulnerabilities identified by CodeRabbit review.

Co-authored-by: Mervin Praison <MervinPraison@users.noreply.github.com>

Author: praisonai-triage-agent[bot]

src/praisonai-agents/praisonaiagents/gateway/config.py

@@ -210,7 +210,7 @@ class GatewayConfig:
 
     host: str = "127.0.0.1"
     port: int = 8765
-    bind_host: str = "127.0.0.1"  # For authentication posture resolution
+    bind_host: str = ""  # Defaults to host for authentication posture resolution
     cors_origins: List[str] = field(default_factory=lambda: [])
     auth_token: Optional[str] = None
     max_connections: int = 1000
@@ -222,6 +222,11 @@ class GatewayConfig:
     ssl_key: Optional[str] = None
     push: PushConfig = field(default_factory=PushConfig)
 
+    def __post_init__(self) -> None:
+        """Set bind_host to host if not explicitly provided."""
+        if not self.bind_host:
+            self.bind_host = self.host
+    
     def to_dict(self) -> Dict[str, Any]:
         """Convert to dictionary (hides sensitive data)."""
         return {

src/praisonai-agents/praisonaiagents/gateway/protocols.py

@@ -723,16 +723,20 @@ def is_loopback(host: str) -> bool:
 
     # Handle common string representations
     host = host.lower().strip()
-    if host in ("localhost", "local"):
+    if host == "localhost":
         return True
 
+    # Strip brackets from IPv6 literal forms like "[::1]"
+    if host.startswith("[") and host.endswith("]"):
+        host = host[1:-1]
+    
     try:
         import ipaddress
         # Try to parse as IP address
         addr = ipaddress.ip_address(host)
         return addr.is_loopback
-    except (ValueError, ImportError):
-        # Not a valid IP address or ipaddress not available
+    except ValueError:
+        # Not a valid IP literal
         return False
 
 

src/praisonai/praisonai/gateway/auth.py

@@ -99,9 +99,9 @@ def check_request_auth(
 
         if auth_mode == "token":
             if not expected_token:
-                # No token configured, allow (should not happen after validation)
-                logger.warning("Token mode requested but no expected token configured")
-                return True
+                # Fail closed: token mode with no expected token is a misconfiguration
+                logger.error("Token mode requested but no expected token configured; denying request")
+                return False
 
             if not request_token:
                 return False
@@ -128,9 +128,10 @@ def assert_external_bind_safe(config) -> None:
     # Use bind_host if available, otherwise fall back to host
     bind_host = getattr(config, 'bind_host', None) or config.host
     auth_token = getattr(config, 'auth_token', None)
+    configured_mode = getattr(config, 'auth_mode', None)
 
     # Resolve authentication mode based on bind interface
-    auth_mode = resolve_auth_mode(bind_host, configured=None)
+    auth_mode = resolve_auth_mode(bind_host, configured=configured_mode)
 
     # Create enforcer and validate
     enforcer = GatewayAuthEnforcer()
@@ -181,13 +182,15 @@ def ensure_token_env_file(token: str, env_file_path: Optional[str] = None) -> No
         env_file_path = os.path.join(praisonai_dir, ".env")
 
     try:
-        # Ensure directory exists
-        os.makedirs(os.path.dirname(env_file_path), mode=0o700, exist_ok=True)
+        # Ensure directory exists (skip when env_file_path has no directory component)
+        parent = os.path.dirname(env_file_path)
+        if parent:
+            os.makedirs(parent, mode=0o700, exist_ok=True)
 
         # Read existing content
         existing_lines = []
         if os.path.exists(env_file_path):
-            with open(env_file_path, 'r') as f:
+            with open(env_file_path, 'r', encoding='utf-8') as f:
                 existing_lines = f.readlines()
 
         # Check if GATEWAY_AUTH_TOKEN already exists
@@ -202,11 +205,11 @@ def ensure_token_env_file(token: str, env_file_path: Optional[str] = None) -> No
         if not token_line_found:
             existing_lines.append(f"GATEWAY_AUTH_TOKEN={token}\n")
 
-        # Write back to file with secure permissions
-        with open(env_file_path, 'w') as f:
+        # Write atomically with secure permissions from the start
+        fd = os.open(env_file_path, os.O_WRONLY | os.O_CREAT | os.O_TRUNC, 0o600)
+        with os.fdopen(fd, 'w', encoding='utf-8') as f:
             f.writelines(existing_lines)
-        
-        # Set secure file permissions (owner read/write only)
+        # Defensive chmod in case the file pre-existed with looser perms
         os.chmod(env_file_path, 0o600)
 
         logger.debug(f"Gateway auth token persisted to {env_file_path}")

src/praisonai/praisonai/gateway/server.py

@@ -279,6 +279,12 @@ async def start(self) -> None:
             logger.warning("Gateway already running")
             return
 
+        # Keep authentication posture aligned with the host Uvicorn will bind
+        if getattr(self.config, "host", None) != self._host:
+            self.config.host = self._host
+        if not getattr(self.config, "bind_host", None) or self.config.bind_host != self._host:
+            self.config.bind_host = self._host
+        
         # Validate authentication configuration before starting
         from .auth import assert_external_bind_safe
         try:

src/praisonai/praisonai/ui/_auth.py

@@ -8,6 +8,7 @@
 
 import logging
 import os
+import secrets
 from typing import Optional
 
 # Import protocols from core SDK
@@ -52,6 +53,12 @@ def validate_credentials_config(
         """
         is_default_creds = (username == "admin" and password == "admin")
         is_external = not is_loopback(bind_host)
+
+        if is_external and (not username or not password):
+            raise UIStartupError(
+                f"Cannot use empty Chainlit credentials on external interface {bind_host}.\n"
+                f"  Fix: Set non-empty CHAINLIT_USERNAME and CHAINLIT_PASSWORD environment variables"
+            )
 
         if is_default_creds and is_external and not allow_defaults:
             raise UIStartupError(
@@ -68,8 +75,8 @@ def validate_credentials_config(
                 )
             else:
                 logger.warning(
-                    f"⚠️  Using default admin/admin credentials on localhost. "
-                    f"Set CHAINLIT_USERNAME and CHAINLIT_PASSWORD for production."
+                    "⚠️  Using default admin/admin credentials on localhost. "
+                    "Set CHAINLIT_USERNAME and CHAINLIT_PASSWORD for production."
                 )
 
     def check_auth_callback(
@@ -92,7 +99,21 @@ def check_auth_callback(
         Returns:
             True if credentials are valid, False otherwise
         """
-        return (provided_username, provided_password) == (expected_username, expected_password)
+        if not all(
+            isinstance(value, str)
+            for value in (
+                provided_username,
+                provided_password,
+                expected_username,
+                expected_password,
+            )
+        ):
+            return False
+
+        return (
+            provided_username == expected_username
+            and secrets.compare_digest(provided_password, expected_password)
+        )
 
 
 def register_password_auth(app, *, bind_host: str) -> None:
@@ -126,12 +147,12 @@ def register_password_auth(app, *, bind_host: str) -> None:
         )
     except UIStartupError as e:
         # Convert to runtime error to prevent startup
-        raise RuntimeError(str(e))
+        raise RuntimeError(str(e)) from e
 
     @cl.password_auth_callback
     def auth_callback(username: str, password: str):
         """Shared password authentication callback."""
-        logger.debug(f"Auth attempt: username='{username}', expected='{expected_username}'")
+        logger.debug("Password auth attempt")
 
         # Use enforcer to check credentials
         is_valid = enforcer.check_auth_callback(
@@ -143,10 +164,10 @@ def auth_callback(username: str, password: str):
         )
 
         if is_valid:
-            logger.info(f"Login successful for user: {username}")
+            logger.info("Password auth successful")
             return cl.User(identifier=username, metadata={"role": "admin", "provider": "credentials"})
         else:
-            logger.warning(f"Login failed for user: {username}")
+            logger.warning("Password auth failed")
             return None
 
     return auth_callback

src/praisonai/tests/integration/gateway/test_bind_aware_e2e.py

@@ -87,17 +87,22 @@ async def _start_gateway_briefly(self, gateway):
         # Give it a moment to start and perform validation
         await asyncio.sleep(0.1)
 
+        # If startup already failed (e.g. validation raised), surface it now
+        if start_task.done():
+            # Re-raises any exception stored on the task
+            start_task.result()
+            return
+        
         # Stop the gateway
         if gateway.is_running:
             await gateway.stop()
 
         # Cancel the start task if it's still running
-        if not start_task.done():
-            start_task.cancel()
-            try:
-                await start_task
-            except asyncio.CancelledError:
-                pass
+        start_task.cancel()
+        try:
+            await start_task
+        except asyncio.CancelledError:
+            pass
 
 
 class TestRealAgentExecution:

src/praisonai/tests/unit/gateway/test_bind_aware_auth.py

@@ -32,6 +32,7 @@ def test_ipv4_loopback_addresses(self):
     def test_ipv6_loopback_addresses(self):
         """Test IPv6 loopback addresses are correctly identified."""
         assert is_loopback("::1") is True
+        assert is_loopback("[::1]") is True
         assert is_loopback("0:0:0:0:0:0:0:1") is True
 
     def test_localhost_strings(self):
@@ -167,14 +168,13 @@ def test_check_request_auth_token_mode_requires_token(self):
             expected_token="secret"
         ) is True
 
-    def test_check_request_auth_no_expected_token_allows_all(self):
-        """Test that token mode with no expected token allows all requests."""
-        # This handles the case where validation failed to catch the issue
+    def test_check_request_auth_no_expected_token_fails_closed(self):
+        """Test that token mode with no expected token denies requests."""
         assert self.enforcer.check_request_auth(
             auth_mode="token",
             request_token=None,
             expected_token=None
-        ) is True
+        ) is False
 
 
 class TestAssertExternalBindSafe:

src/praisonai/tests/unit/ui/test_ui_bind_aware_creds.py

@@ -7,6 +7,7 @@
 
 import pytest
 import os
+import sys
 from unittest.mock import patch, MagicMock
 
 from praisonai.ui._auth import (
@@ -117,75 +118,80 @@ def test_check_auth_callback_with_incorrect_credentials(self):
 class TestRegisterPasswordAuth:
     """Test the register_password_auth function."""
 
-    @patch('praisonai.ui._auth.cl')
     @patch.dict(os.environ, {}, clear=True)
-    def test_default_credentials_on_loopback(self, mock_cl):
+    def test_default_credentials_on_loopback(self):
         """Test registration with default credentials on loopback."""
-        # Should not raise
-        register_password_auth(None, bind_host="127.0.0.1")
-        
-        # Should have registered a callback
-        mock_cl.password_auth_callback.assert_called_once()
+        mock_cl = MagicMock()
+        with patch.dict(sys.modules, {"chainlit": mock_cl}):
+            # Should not raise
+            register_password_auth(None, bind_host="127.0.0.1")
+            
+            # Should have registered a callback
+            mock_cl.password_auth_callback.assert_called_once()
 
-    @patch('praisonai.ui._auth.cl')
     @patch.dict(os.environ, {}, clear=True)
-    def test_default_credentials_on_external_raises_error(self, mock_cl):
+    def test_default_credentials_on_external_raises_error(self):
         """Test registration with default credentials on external interface raises error."""
-        with pytest.raises(RuntimeError) as excinfo:
-            register_password_auth(None, bind_host="0.0.0.0")
-        
-        assert "Cannot use default admin/admin credentials on external interface" in str(excinfo.value)
+        mock_cl = MagicMock()
+        with patch.dict(sys.modules, {"chainlit": mock_cl}):
+            with pytest.raises(RuntimeError) as excinfo:
+                register_password_auth(None, bind_host="0.0.0.0")
+            
+            assert "Cannot use default admin/admin credentials on external interface" in str(excinfo.value)
 
-    @patch('praisonai.ui._auth.cl')
     @patch.dict(os.environ, {"PRAISONAI_ALLOW_DEFAULT_CREDS": "1"}, clear=True)
-    def test_default_credentials_on_external_with_escape_hatch(self, mock_cl):
+    def test_default_credentials_on_external_with_escape_hatch(self):
         """Test registration with default credentials on external interface with escape hatch."""
-        # Should not raise with escape hatch
-        register_password_auth(None, bind_host="0.0.0.0")
-        
-        # Should have registered a callback
-        mock_cl.password_auth_callback.assert_called_once()
+        mock_cl = MagicMock()
+        with patch.dict(sys.modules, {"chainlit": mock_cl}):
+            # Should not raise with escape hatch
+            register_password_auth(None, bind_host="0.0.0.0")
+            
+            # Should have registered a callback
+            mock_cl.password_auth_callback.assert_called_once()
 
-    @patch('praisonai.ui._auth.cl')
     @patch.dict(os.environ, {
         "CHAINLIT_USERNAME": "myuser",
         "CHAINLIT_PASSWORD": "mypassword"
     }, clear=True)
-    def test_custom_credentials_on_external(self, mock_cl):
+    def test_custom_credentials_on_external(self):
         """Test registration with custom credentials on external interface."""
-        # Should not raise
-        register_password_auth(None, bind_host="0.0.0.0")
-        
-        # Should have registered a callback
-        mock_cl.password_auth_callback.assert_called_once()
+        mock_cl = MagicMock()
+        with patch.dict(sys.modules, {"chainlit": mock_cl}):
+            # Should not raise
+            register_password_auth(None, bind_host="0.0.0.0")
+            
+            # Should have registered a callback
+            mock_cl.password_auth_callback.assert_called_once()
 
-    @patch('praisonai.ui._auth.cl')
     @patch.dict(os.environ, {
         "CHAINLIT_USERNAME": "myuser",
         "CHAINLIT_PASSWORD": "mypassword"
     }, clear=True)
-    def test_auth_callback_functionality(self, mock_cl):
+    def test_auth_callback_functionality(self):
         """Test that the registered auth callback works correctly."""
-        register_password_auth(None, bind_host="127.0.0.1")
-        
-        # Get the registered callback function
-        callback = mock_cl.password_auth_callback.call_args[0][0]
-        
-        # Mock User class
-        mock_user = MagicMock()
-        mock_cl.User.return_value = mock_user
-        
-        # Test correct credentials
-        result = callback("myuser", "mypassword")
-        assert result == mock_user
-        mock_cl.User.assert_called_with(
-            identifier="myuser",
-            metadata={"role": "admin", "provider": "credentials"}
-        )
-        
-        # Test incorrect credentials
-        result = callback("wrong", "credentials")
-        assert result is None
+        mock_cl = MagicMock()
+        with patch.dict(sys.modules, {"chainlit": mock_cl}):
+            register_password_auth(None, bind_host="127.0.0.1")
+            
+            # Get the registered callback function
+            callback = mock_cl.password_auth_callback.call_args[0][0]
+            
+            # Mock User class
+            mock_user = MagicMock()
+            mock_cl.User.return_value = mock_user
+            
+            # Test correct credentials
+            result = callback("myuser", "mypassword")
+            assert result == mock_user
+            mock_cl.User.assert_called_with(
+                identifier="myuser",
+                metadata={"role": "admin", "provider": "credentials"}
+            )
+            
+            # Test incorrect credentials
+            result = callback("wrong", "credentials")
+            assert result is None
 
 
 class TestEnvironmentVariables: