Repository: meta-llama/llama-stack
File: llama_stack/core/server/auth.py (Python)

Code before:

    Attribute-Based Access Control:
    The attributes returned by the auth provider are used to determine which
    resources the user can access. Resources can specify required attributes
    using the access_attributes field. For a user to access a resource:

    1. All attribute categories specified in the resource must be present in the user's attributes
    2. For each category, the user must have at least one matching value

    If the auth provider doesn't return any attributes, the user will only be able to
    access resources that don't have access_attributes defined.
    """

    def __init__(self, app, auth_config: AuthenticationConfig, impls):
        self.app = app
        self.impls = impls
        self.auth_provider = create_auth_provider(auth_config)

    def _is_unauthenticated_endpoint_allowed(self, path: str) -> bool:
        """Check if the endpoint allows unauthenticated access.

        Certain operational endpoints (/health, /version) are allowed without authentication
        to support monitoring, load balancers, and observability tools.
        """
        # Allow health and version endpoints for operational purposes
        return path.endswith("/health") or path.endswith("/version")

    async def __call__(self, scope, receive, send):
        if scope["type"] == "http":
            # First, Check if this endpoint is explicitly allowed without authentication
            if self._is_unauthenticated_endpoint_allowed(scope.get("path", "")):
                # Allow the request to proceed without authentication
                return await self.app(scope, receive, send)

            # Next, handle authentication
            headers = dict(scope.get("headers", []))
            auth_header = headers.get(b"authorization", b"").decode()

            if not auth_header:
                error_msg = self.auth_provider.get_auth_error_message(scope)
                return await self._send_auth_error(send, error_msg)

            if not auth_header.startswith("Bearer "):
                return await self._send_auth_error(send, "Invalid Authorization header format")

            token = auth_header.split("Bearer ", 1)[1]

            # Validate token and get access attributes
            try:
                validation_result = await self.auth_provider.validate_token(token, scope)
            except httpx.TimeoutException:

Review comment (security):
From Claude:

  The endswith() check is trivially bypassable:

  - /admin/delete-users/health → allowed ✓
  - /api/secrets/version → allowed ✓
  - /dangerous-operation/health → allowed ✓

  Any attacker can append /health or /version to bypass authentication entirely.

  Fix: Use exact path matching

  def _is_unauthenticated_endpoint_allowed(self, path: str) -> bool:
      """Check if the endpoint allows unauthenticated access.

      Certain operational endpoints (/health, /version) are allowed without authentication
      to support monitoring, load balancers, and observability tools.
      """
      return path in {"/health", "/version"}

  Or if you need to handle path prefixes:
  # Strip query params and normalize
  normalized = path.split('?')[0].rstrip('/')
  return normalized in {"/health", "/version"}

  This is a critical auth bypass vulnerability that needs immediate fixing.

Code after:
    using the access_attributes field. For a user to access a resource:

    1. All attribute categories specified in the resource must be present in the user's attributes
    2. For each category, the user must have at least one matching value

    If the auth provider doesn't return any attributes, the user will only be able to
    access resources that don't have access_attributes defined.
    """

    def __init__(self, app, auth_config: AuthenticationConfig, impls):
        self.app = app
        self.impls = impls
        self.auth_provider = create_auth_provider(auth_config)

    async def __call__(self, scope, receive, send):
        if scope["type"] == "http":
            # Find the route and check if authentication is required
            path = scope.get("path", "")
            method = scope.get("method", hdrs.METH_GET)

            if not hasattr(self, "route_impls"):
                self.route_impls = initialize_route_impls(self.impls)

            webmethod = None
            try:
                _, _, _, webmethod = find_matching_route(method, path, self.route_impls)
            except ValueError:
                # If no matching endpoint is found, pass here to run auth anyways
                pass

            # If webmethod explicitly sets require_authentication=False, allow without auth
            if webmethod and webmethod.require_authentication is False:
                logger.debug(f"Allowing unauthenticated access to endpoint: {path}")
                return await self.app(scope, receive, send)

            # Handle authentication
            headers = dict(scope.get("headers", []))
            auth_header = headers.get(b"authorization", b"").decode()

            if not auth_header:
                error_msg = self.auth_provider.get_auth_error_message(scope)
                return await self._send_auth_error(send, error_msg)

            if not auth_header.startswith("Bearer "):
                return await self._send_auth_error(send, "Invalid Authorization header format")

            token = auth_header.split("Bearer ", 1)[1]

            # Validate token and get access attributes
            try:
                validation_result = await self.auth_provider.validate_token(token, scope)