fix: access control to fail-closed when owner attributes are missing

Changed UserInOwnersList.matches() to return False instead of True when
a resource's owner attributes are None. This prevents unintended access
when resource when owner attributes aren't present.

For example, checking "user in owners teams" now returns False if the
resource has no teams attribute, rather than defaulting to True.

Added "ResourceIsUnowned" to allow access to unowned resources.

Updated default_policy to use multiple separate "user in owners"
AccessRules instead of a single rule with multiple when clauses. With
the new fail-closed behavior, only one rule needs to match. Added a
"user is owner" rule to handle resources. Also added ResourceIsUnowned
to allow access to unowned resources.

Updated the SQL generated by _build_default_policy_where_clause to
reflect the above changes.

Closes: #4272

Signed-off-by: Derek Higgins <[email protected]>

Author: derekhiggins

src/llama_stack/core/access_control/access_control.py

@@ -66,7 +66,17 @@ def default_policy() -> list[AccessRule]:
     return [
         AccessRule(
             permit=Scope(actions=list(Action)),
-            when=["user in owners " + name for name in ["roles", "teams", "projects", "namespaces"]],
+            when=["user in owners " + name],
+        )
+        for name in ["roles", "teams", "projects", "namespaces"]
+    ] + [
+        AccessRule(
+            permit=Scope(actions=list(Action)),
+            when=["user is owner"],
+        ),
+        AccessRule(
+            permit=Scope(actions=list(Action)),
+            when=["resource is unowned"],
         ),
     ]
 

src/llama_stack/core/access_control/conditions.py

@@ -40,7 +40,7 @@ def owners_values(self, resource: ProtectedResource) -> list[str] | None:
     def matches(self, resource: ProtectedResource, user: User) -> bool:
         required = self.owners_values(resource)
         if not required:
-            return True
+            return False
         if not user.attributes or self.name not in user.attributes or not user.attributes[self.name]:
             return False
         user_values = user.attributes[self.name]
@@ -106,6 +106,14 @@ def __repr__(self):
         return "user is not owner"
 
 
+class ResourceIsUnowned:
+    def matches(self, resource: ProtectedResource, user: User) -> bool:
+        return not resource.owner
+
+    def __repr__(self):
+        return "resource is unowned"
+
+
 def parse_condition(condition: str) -> Condition:
     words = condition.split()
     match words:
@@ -121,6 +129,8 @@ def parse_condition(condition: str) -> Condition:
             return UserInOwnersList(name)
         case ["user", "not", "in", "owners", name]:
             return UserNotInOwnersList(name)
+        case ["resource", "is", "unowned"]:
+            return ResourceIsUnowned()
         case _:
             raise ValueError(f"Invalid condition: {condition}")
 

src/llama_stack/core/storage/sqlstore/authorized_sqlstore.py

@@ -23,17 +23,26 @@
 # WARNING: If default_policy() changes, this constant must be updated accordingly
 # or SQL filtering will fall back to conservative mode (safe but less performant)
 #
-# This policy represents: "Permit all actions when user is in owners list for ALL attribute categories"
+# This policy represents: "Permit all actions when user is in owners list for ANY attribute category"
 # The corresponding SQL logic is implemented in _build_default_policy_where_clause():
 # - Public records (no access_attributes) are always accessible
-# - Records with access_attributes require user to match ALL categories that exist in the resource
-# - Missing categories in the resource are treated as "no restriction" (allow)
+# - Records with access_attributes require user to match ANY category that exists in the resource
 # - Within each category, user needs ANY matching value (OR logic)
-# - Between categories, user needs ALL categories to match (AND logic)
+# - Between categories, user needs ANY category to match (OR logic)
 SQL_OPTIMIZED_POLICY = [
     AccessRule(
         permit=Scope(actions=list(Action)),
-        when=["user in owners roles", "user in owners teams", "user in owners projects", "user in owners namespaces"],
+        when=["user in owners " + name],
+    )
+    for name in ["roles", "teams", "projects", "namespaces"]
+] + [
+    AccessRule(
+        permit=Scope(actions=list(Action)),
+        when=["user is owner"],
+    ),
+    AccessRule(
+        permit=Scope(actions=list(Action)),
+        when=["resource is unowned"],
     ),
 ]
 
@@ -279,53 +288,40 @@ def _get_public_access_conditions(self) -> list[str]:
 
         Public records are those with:
         - owner_principal = '' (empty string)
-        - access_attributes is either SQL NULL or JSON null
 
-        Note: Different databases serialize None differently:
-        - SQLite: None → JSON null (text = 'null')
-        - Postgres: None → SQL NULL (IS NULL)
+        The policy "resource is unowned" only checks if owner_principal is empty,
+        regardless of access_attributes.
         """
-        conditions = ["owner_principal = ''"]
-        if self.database_type == StorageBackendType.SQL_POSTGRES.value:
-            # Accept both SQL NULL and JSON null for Postgres compatibility
-            # This handles both old rows (SQL NULL) and new rows (JSON null)
-            conditions.append("(access_attributes IS NULL OR access_attributes::text = 'null')")
-        elif self.database_type == StorageBackendType.SQL_SQLITE.value:
-            # SQLite serializes None as JSON null
-            conditions.append("(access_attributes IS NULL OR access_attributes = 'null')")
-        else:
-            raise ValueError(f"Unsupported database type: {self.database_type}")
-        return conditions
+        return ["owner_principal = ''"]
 
     def _build_default_policy_where_clause(self, current_user: User | None) -> str:
         """Build SQL WHERE clause for the default policy.
 
         Default policy: permit all actions when user in owners [roles, teams, projects, namespaces]
-        This means user must match ALL attribute categories that exist in the resource.
+        This means user must match ANY attribute category that exists in the resource (OR logic).
         """
         base_conditions = self._get_public_access_conditions()
-        user_attr_conditions = []
-
-        if current_user and current_user.attributes:
-            for attr_key, user_values in current_user.attributes.items():
-                if user_values:
-                    value_conditions = []
-                    for value in user_values:
-                        # Check if JSON array contains the value
-                        escaped_value = value.replace("'", "''")
-                        json_text = self._json_extract_text("access_attributes", attr_key)
-                        value_conditions.append(f"({json_text} LIKE '%\"{escaped_value}\"%')")
-
-                    if value_conditions:
-                        # Check if the category is missing (NULL)
-                        category_missing = f"{self._json_extract('access_attributes', attr_key)} IS NULL"
-                        user_matches_category = f"({' OR '.join(value_conditions)})"
-                        user_attr_conditions.append(f"({category_missing} OR {user_matches_category})")
-
-            if user_attr_conditions:
-                all_requirements_met = f"({' AND '.join(user_attr_conditions)})"
-                base_conditions.append(all_requirements_met)
 
+        if current_user:
+            # Add "user is owner" condition - user's principal matches owner_principal
+            escaped_principal = current_user.principal.replace("'", "''")
+            base_conditions.append(f"owner_principal = '{escaped_principal}'")
+
+            # Add "user in owners" conditions for attribute matching
+            if current_user.attributes:
+                for attr_key, user_values in current_user.attributes.items():
+                    if user_values:
+                        value_conditions = []
+                        for value in user_values:
+                            # Check if JSON array contains the value
+                            escaped_value = value.replace("'", "''")
+                            json_text = self._json_extract_text("access_attributes", attr_key)
+                            value_conditions.append(f"({json_text} LIKE '%\"{escaped_value}\"%')")
+
+                        if value_conditions:
+                            # User matches this category if any of their values match
+                            user_matches_category = f"({' OR '.join(value_conditions)})"
+                            base_conditions.append(user_matches_category)
         return f"({' OR '.join(base_conditions)})"
 
     def _build_conservative_where_clause(self) -> str:

tests/integration/providers/utils/sqlstore/test_authorized_sqlstore.py

@@ -184,6 +184,16 @@ async def test_authorized_store_attributes(mock_get_authenticated_user, authoriz
             f"Category missing logic failed: expected 4,5 but got {category_test_ids}"
         )
 
+        # Test a user that has all roles and teams (should generate SQL)
+        # owner_principal = ''
+        # owner_principal = 'super-user'
+        # ((JSON_EXTRACT(access_attributes, '$.roles') LIKE '%"admin"%') OR (JSON_EXTRACT(access_attributes, '$.roles') LIKE '%"user"%'))
+        # ((JSON_EXTRACT(access_attributes, '$.teams') LIKE '%"dev"%') OR (JSON_EXTRACT(access_attributes, '$.teams') LIKE '%"qa"%'))
+        super_user = User("super-user", {"roles": ["admin", "user"], "teams": ["dev", "qa"]})
+        mock_get_authenticated_user.return_value = super_user
+        result = await authorized_store.fetch_all(table_name)
+        assert len(result.data) == 6
+
     finally:
         # Clean up records
         await cleanup_records(authorized_store.sql_store, table_name, ["1", "2", "3", "4", "5", "6"])

tests/unit/server/test_access_control.py

@@ -78,7 +78,7 @@ async def test_access_control_with_cache(mock_get_authenticated_user, test_setup
     with pytest.raises(ValueError):
         await routing_table.get_model("model-data-scientist")
 
-    mock_get_authenticated_user.return_value = User("test-user", {"roles": ["data-scientist"], "teams": ["other-team"]})
+    mock_get_authenticated_user.return_value = User("test-user", {"roles": ["user"], "teams": ["other-team"]})
     all_models = await routing_table.list_models()
     assert len(all_models.data) == 1
     assert all_models.data[0].identifier == "model-public"
@@ -154,16 +154,16 @@ async def test_access_control_empty_attributes(mock_get_authenticated_user, test
     )
     await registry.register(model)
     mock_get_authenticated_user.return_value = User(
-        "test-user",
+        "differentuser",
         {
             "roles": [],
         },
     )
-    result = await routing_table.get_model("model-empty-attrs")
-    assert result.identifier == "model-empty-attrs"
+    with pytest.raises(ValueError):
+        await routing_table.get_model("model-empty-attrs")
     all_models = await routing_table.list_models()
     model_ids = [m.identifier for m in all_models.data]
-    assert "model-empty-attrs" in model_ids
+    assert "model-empty-attrs" not in model_ids
 
 
 @patch("llama_stack.core.routing_tables.common.get_authenticated_user")
@@ -223,7 +223,7 @@ async def test_automatic_access_attributes(mock_get_authenticated_user, test_set
     assert registered_model.owner.attributes["projects"] == ["llama-3"]
 
     # Verify another user without matching attributes can't access it
-    mock_get_authenticated_user.return_value = User("test-user", {"roles": ["engineer"], "teams": ["infra-team"]})
+    mock_get_authenticated_user.return_value = User("test-user2", {"roles": ["engineer"], "teams": ["infra-team"]})
     with pytest.raises(ValueError):
         await routing_table.get_model("auto-access-model")
 
@@ -363,6 +363,7 @@ def test_permit_when():
 
 
 def test_permit_unless():
+    # permit unless both conditions are met
     config = """
     - permit:
         principal: user-1
@@ -377,10 +378,10 @@ def test_permit_unless():
         identifier="mymodel",
         provider_id="myprovider",
         model_type=ModelType.llm,
-        owner=User("testuser", {"namespaces": ["foo"]}),
+        owner=User("testuser", {"namespaces": ["foo"], "teams": ["ml-team"]}),
     )
     assert is_action_allowed(policy, "read", model, User("user-1", {"namespaces": ["foo"]}))
-    assert not is_action_allowed(policy, "read", model, User("user-1", {"namespaces": ["bar"]}))
+    assert not is_action_allowed(policy, "read", model, User("user-1", {"namespaces": ["bar"], "teams": ["ml-team"]}))
     assert not is_action_allowed(policy, "read", model, User("user-2", {"namespaces": ["foo"]}))