fix: access control to fail-closed when owner attributes are missing (#4273)
Some checks failed
SqlStore Integration Tests / test-postgres (3.12) (push) Failing after 0s
Integration Auth Tests / test-matrix (oauth2_token) (push) Failing after 1s
SqlStore Integration Tests / test-postgres (3.13) (push) Failing after 1s
Test External Providers Installed via Module / test-external-providers-from-module (venv) (push) Has been skipped
Integration Tests (Replay) / generate-matrix (push) Successful in 3s
API Conformance Tests / check-schema-compatibility (push) Successful in 10s
Python Package Build Test / build (3.12) (push) Successful in 16s
Python Package Build Test / build (3.13) (push) Successful in 17s
Vector IO Integration Tests / test-matrix (push) Failing after 35s
UI Tests / ui-tests (22) (push) Successful in 39s
Test External API and Providers / test-external (venv) (push) Failing after 44s
Unit Tests / unit-tests (3.13) (push) Failing after 1m26s
Unit Tests / unit-tests (3.12) (push) Failing after 1m28s
Pre-commit / pre-commit (22) (push) Successful in 3m28s
Integration Tests (Replay) / Integration Tests (, , , client=, ) (push) Failing after 3m12s

This commit is contained in:
Derek Higgins 2025-12-04 16:38:32 +00:00 committed by GitHub
parent b4903d6766
commit 686065fe27
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
5 changed files with 80 additions and 53 deletions

View file

@ -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"])

View file

@ -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"]}))