424 lines
16 KiB
Python
424 lines
16 KiB
Python
|
"""Tests for ABAC policy service — attribute-based access control policies.
|
||
|
|
|
||
|
|
Covers:
|
||
|
|
- Policy allow (conditions match → access granted)
|
||
|
|
- Policy deny (conditions match → access blocked)
|
||
|
|
- Multi-condition AND (all conditions must match)
|
||
|
|
- Multi-condition OR (any condition must match)
|
||
|
|
- Policy CRUD operations
|
||
|
|
- build_sql_condition translation
|
||
|
|
- apply_policy_filter integration
|
||
|
|
"""
|
||
|
|
|
||
|
|
from __future__ import annotations
|
||
|
|
|
||
|
|
import uuid
|
||
|
|
|
||
|
|
import pytest
|
||
|
|
import pytest_asyncio
|
||
|
|
from sqlalchemy import select
|
||
|
|
from sqlalchemy.ext.asyncio import AsyncSession
|
||
|
|
|
||
|
|
from app.models.contact import Contact
|
||
|
|
from app.models.entity_policy import EntityPolicy
|
||
|
|
from app.services import policy_service as ps
|
||
|
|
from tests.conftest import seed_tenant_and_users
|
||
|
|
|
||
|
|
|
||
|
|
@pytest.mark.asyncio
|
||
|
|
class TestABACPolicyService:
|
||
|
|
"""Tests for ABAC policy service — attribute-based access control."""
|
||
|
|
|
||
|
|
async def test_policy_allow(self, db_session: AsyncSession):
|
||
|
|
"""Allow policy with matching conditions grants access."""
|
||
|
|
seed = await seed_tenant_and_users(db_session)
|
||
|
|
tenant_id = seed["tenant_a"].id
|
||
|
|
user_id = seed["admin_a"].id
|
||
|
|
|
||
|
|
# Create an allow policy for contacts where type == 'company'
|
||
|
|
policy = await ps.create_policy(
|
||
|
|
db_session,
|
||
|
|
tenant_id=tenant_id,
|
||
|
|
name="Allow company contacts",
|
||
|
|
entity_type="contact",
|
||
|
|
principal_type="user",
|
||
|
|
principal_id=str(user_id),
|
||
|
|
effect="allow",
|
||
|
|
conditions={
|
||
|
|
"operator": "AND",
|
||
|
|
"rules": [
|
||
|
|
{"field": "type", "op": "eq", "value": "company"}
|
||
|
|
]
|
||
|
|
},
|
||
|
|
)
|
||
|
|
|
||
|
|
assert policy["effect"] == "allow"
|
||
|
|
assert policy["name"] == "Allow company contacts"
|
||
|
|
assert policy["entity_type"] == "contact"
|
||
|
|
assert policy["conditions"]["rules"][0]["field"] == "type"
|
||
|
|
|
||
|
|
# Verify the policy was persisted
|
||
|
|
db_policy = await db_session.get(EntityPolicy, uuid.UUID(policy["id"]))
|
||
|
|
assert db_policy is not None
|
||
|
|
assert db_policy.effect == "allow"
|
||
|
|
|
||
|
|
async def test_policy_deny(self, db_session: AsyncSession):
|
||
|
|
"""Deny policy blocks access when conditions match."""
|
||
|
|
seed = await seed_tenant_and_users(db_session)
|
||
|
|
tenant_id = seed["tenant_a"].id
|
||
|
|
user_id = seed["admin_a"].id
|
||
|
|
|
||
|
|
# Create a deny policy for contacts where name contains 'Confidential'
|
||
|
|
policy = await ps.create_policy(
|
||
|
|
db_session,
|
||
|
|
tenant_id=tenant_id,
|
||
|
|
name="Deny confidential contacts",
|
||
|
|
entity_type="contact",
|
||
|
|
principal_type="user",
|
||
|
|
principal_id=str(user_id),
|
||
|
|
effect="deny",
|
||
|
|
conditions={
|
||
|
|
"operator": "AND",
|
||
|
|
"rules": [
|
||
|
|
{"field": "name", "op": "starts_with", "value": "Confidential"}
|
||
|
|
]
|
||
|
|
},
|
||
|
|
)
|
||
|
|
|
||
|
|
assert policy["effect"] == "deny"
|
||
|
|
assert policy["enabled"] is True
|
||
|
|
|
||
|
|
async def test_multi_condition_and(self, db_session: AsyncSession):
|
||
|
|
"""AND conditions: all rules must match for the policy to apply."""
|
||
|
|
seed = await seed_tenant_and_users(db_session)
|
||
|
|
tenant_id = seed["tenant_a"].id
|
||
|
|
user_id = seed["admin_a"].id
|
||
|
|
|
||
|
|
# Create policy with AND conditions
|
||
|
|
policy = await ps.create_policy(
|
||
|
|
db_session,
|
||
|
|
tenant_id=tenant_id,
|
||
|
|
name="VIP company contacts",
|
||
|
|
entity_type="contact",
|
||
|
|
principal_type="user",
|
||
|
|
principal_id=str(user_id),
|
||
|
|
effect="allow",
|
||
|
|
conditions={
|
||
|
|
"operator": "AND",
|
||
|
|
"rules": [
|
||
|
|
{"field": "type", "op": "eq", "value": "company"},
|
||
|
|
{"field": "name", "op": "contains", "value": "VIP"},
|
||
|
|
]
|
||
|
|
},
|
||
|
|
)
|
||
|
|
|
||
|
|
assert policy["conditions"]["operator"] == "AND"
|
||
|
|
assert len(policy["conditions"]["rules"]) == 2
|
||
|
|
|
||
|
|
# Test build_sql_condition translation
|
||
|
|
condition = ps.build_sql_condition(policy["conditions"], Contact)
|
||
|
|
assert condition is not None
|
||
|
|
|
||
|
|
async def test_multi_condition_or(self, db_session: AsyncSession):
|
||
|
|
"""OR conditions: any rule matching triggers the policy."""
|
||
|
|
seed = await seed_tenant_and_users(db_session)
|
||
|
|
tenant_id = seed["tenant_a"].id
|
||
|
|
user_id = seed["admin_a"].id
|
||
|
|
|
||
|
|
# Create policy with OR conditions
|
||
|
|
policy = await ps.create_policy(
|
||
|
|
db_session,
|
||
|
|
tenant_id=tenant_id,
|
||
|
|
name="High value or VIP contacts",
|
||
|
|
entity_type="contact",
|
||
|
|
principal_type="user",
|
||
|
|
principal_id=str(user_id),
|
||
|
|
effect="allow",
|
||
|
|
conditions={
|
||
|
|
"operator": "OR",
|
||
|
|
"rules": [
|
||
|
|
{"field": "name", "op": "contains", "value": "VIP"},
|
||
|
|
{"field": "name", "op": "contains", "value": "Premium"},
|
||
|
|
]
|
||
|
|
},
|
||
|
|
)
|
||
|
|
|
||
|
|
assert policy["conditions"]["operator"] == "OR"
|
||
|
|
assert len(policy["conditions"]["rules"]) == 2
|
||
|
|
|
||
|
|
# Test build_sql_condition translation
|
||
|
|
condition = ps.build_sql_condition(policy["conditions"], Contact)
|
||
|
|
assert condition is not None
|
||
|
|
|
||
|
|
async def test_build_sql_condition_eq(self, db_session: AsyncSession):
|
||
|
|
"""build_sql_condition translates 'eq' operator correctly."""
|
||
|
|
conditions = {
|
||
|
|
"operator": "AND",
|
||
|
|
"rules": [
|
||
|
|
{"field": "type", "op": "eq", "value": "person"}
|
||
|
|
]
|
||
|
|
}
|
||
|
|
condition = ps.build_sql_condition(conditions, Contact)
|
||
|
|
assert condition is not None
|
||
|
|
|
||
|
|
async def test_build_sql_condition_gt(self, db_session: AsyncSession):
|
||
|
|
"""build_sql_condition translates 'gt' operator correctly."""
|
||
|
|
conditions = {
|
||
|
|
"operator": "AND",
|
||
|
|
"rules": [
|
||
|
|
{"field": "id", "op": "gt", "value": 100}
|
||
|
|
]
|
||
|
|
}
|
||
|
|
condition = ps.build_sql_condition(conditions, Contact)
|
||
|
|
assert condition is not None
|
||
|
|
|
||
|
|
async def test_build_sql_condition_contains(self, db_session: AsyncSession):
|
||
|
|
"""build_sql_condition translates 'contains' operator correctly."""
|
||
|
|
conditions = {
|
||
|
|
"operator": "AND",
|
||
|
|
"rules": [
|
||
|
|
{"field": "name", "op": "contains", "value": "test"}
|
||
|
|
]
|
||
|
|
}
|
||
|
|
condition = ps.build_sql_condition(conditions, Contact)
|
||
|
|
assert condition is not None
|
||
|
|
|
||
|
|
async def test_build_sql_condition_nested(self, db_session: AsyncSession):
|
||
|
|
"""build_sql_condition handles nested conditions blocks."""
|
||
|
|
conditions = {
|
||
|
|
"operator": "AND",
|
||
|
|
"rules": [
|
||
|
|
{
|
||
|
|
"operator": "OR",
|
||
|
|
"rules": [
|
||
|
|
{"field": "type", "op": "eq", "value": "company"},
|
||
|
|
{"field": "type", "op": "eq", "value": "person"},
|
||
|
|
]
|
||
|
|
},
|
||
|
|
{"field": "name", "op": "is_not_null", "value": None}
|
||
|
|
]
|
||
|
|
}
|
||
|
|
condition = ps.build_sql_condition(conditions, Contact)
|
||
|
|
assert condition is not None
|
||
|
|
|
||
|
|
async def test_build_sql_condition_empty_rules(self, db_session: AsyncSession):
|
||
|
|
"""build_sql_condition returns None for empty rules."""
|
||
|
|
conditions = {
|
||
|
|
"operator": "AND",
|
||
|
|
"rules": []
|
||
|
|
}
|
||
|
|
condition = ps.build_sql_condition(conditions, Contact)
|
||
|
|
assert condition is None
|
||
|
|
|
||
|
|
async def test_build_sql_condition_unsupported_op(self, db_session: AsyncSession):
|
||
|
|
"""build_sql_condition skips unsupported operators gracefully."""
|
||
|
|
conditions = {
|
||
|
|
"operator": "AND",
|
||
|
|
"rules": [
|
||
|
|
{"field": "name", "op": "unsupported_op", "value": "test"}
|
||
|
|
]
|
||
|
|
}
|
||
|
|
condition = ps.build_sql_condition(conditions, Contact)
|
||
|
|
assert condition is None
|
||
|
|
|
||
|
|
async def test_list_policies(self, db_session: AsyncSession):
|
||
|
|
"""list_policies returns all policies for a tenant."""
|
||
|
|
seed = await seed_tenant_and_users(db_session)
|
||
|
|
tenant_id = seed["tenant_a"].id
|
||
|
|
user_id = seed["admin_a"].id
|
||
|
|
|
||
|
|
# Create two policies
|
||
|
|
await ps.create_policy(
|
||
|
|
db_session, tenant_id=tenant_id, name="Policy 1",
|
||
|
|
entity_type="contact", principal_type="user",
|
||
|
|
principal_id=str(user_id), effect="allow",
|
||
|
|
)
|
||
|
|
await ps.create_policy(
|
||
|
|
db_session, tenant_id=tenant_id, name="Policy 2",
|
||
|
|
entity_type="contact", principal_type="user",
|
||
|
|
principal_id=str(user_id), effect="deny",
|
||
|
|
)
|
||
|
|
|
||
|
|
policies = await ps.list_policies(db_session, tenant_id)
|
||
|
|
assert len(policies) >= 2
|
||
|
|
names = [p["name"] for p in policies]
|
||
|
|
assert "Policy 1" in names
|
||
|
|
assert "Policy 2" in names
|
||
|
|
|
||
|
|
async def test_list_policies_filtered_by_entity_type(self, db_session: AsyncSession):
|
||
|
|
"""list_policies filters by entity_type when provided."""
|
||
|
|
seed = await seed_tenant_and_users(db_session)
|
||
|
|
tenant_id = seed["tenant_a"].id
|
||
|
|
user_id = seed["admin_a"].id
|
||
|
|
|
||
|
|
await ps.create_policy(
|
||
|
|
db_session, tenant_id=tenant_id, name="Contact Policy",
|
||
|
|
entity_type="contact", principal_type="user",
|
||
|
|
principal_id=str(user_id), effect="allow",
|
||
|
|
)
|
||
|
|
await ps.create_policy(
|
||
|
|
db_session, tenant_id=tenant_id, name="File Policy",
|
||
|
|
entity_type="dms_file", principal_type="user",
|
||
|
|
principal_id=str(user_id), effect="allow",
|
||
|
|
)
|
||
|
|
|
||
|
|
contact_policies = await ps.list_policies(db_session, tenant_id, entity_type="contact")
|
||
|
|
assert len(contact_policies) >= 1
|
||
|
|
assert all(p["entity_type"] == "contact" for p in contact_policies)
|
||
|
|
|
||
|
|
async def test_update_policy(self, db_session: AsyncSession):
|
||
|
|
"""update_policy modifies an existing policy."""
|
||
|
|
seed = await seed_tenant_and_users(db_session)
|
||
|
|
tenant_id = seed["tenant_a"].id
|
||
|
|
user_id = seed["admin_a"].id
|
||
|
|
|
||
|
|
policy = await ps.create_policy(
|
||
|
|
db_session, tenant_id=tenant_id, name="Original",
|
||
|
|
entity_type="contact", principal_type="user",
|
||
|
|
principal_id=str(user_id), effect="allow",
|
||
|
|
)
|
||
|
|
|
||
|
|
updated = await ps.update_policy(
|
||
|
|
db_session, tenant_id, policy["id"],
|
||
|
|
name="Updated", effect="deny",
|
||
|
|
)
|
||
|
|
|
||
|
|
assert updated["name"] == "Updated"
|
||
|
|
assert updated["effect"] == "deny"
|
||
|
|
|
||
|
|
async def test_delete_policy(self, db_session: AsyncSession):
|
||
|
|
"""delete_policy removes a policy."""
|
||
|
|
seed = await seed_tenant_and_users(db_session)
|
||
|
|
tenant_id = seed["tenant_a"].id
|
||
|
|
user_id = seed["admin_a"].id
|
||
|
|
|
||
|
|
policy = await ps.create_policy(
|
||
|
|
db_session, tenant_id=tenant_id, name="To Delete",
|
||
|
|
entity_type="contact", principal_type="user",
|
||
|
|
principal_id=str(user_id), effect="allow",
|
||
|
|
)
|
||
|
|
|
||
|
|
await ps.delete_policy(db_session, tenant_id, policy["id"])
|
||
|
|
|
||
|
|
# Verify it's gone
|
||
|
|
policies = await ps.list_policies(db_session, tenant_id)
|
||
|
|
ids = [p["id"] for p in policies]
|
||
|
|
assert policy["id"] not in ids
|
||
|
|
|
||
|
|
async def test_apply_policy_filter_no_policies(self, db_session: AsyncSession):
|
||
|
|
"""apply_policy_filter returns query unchanged when no policies match."""
|
||
|
|
seed = await seed_tenant_and_users(db_session)
|
||
|
|
tenant_id = seed["tenant_a"].id
|
||
|
|
user_id = seed["admin_a"].id
|
||
|
|
|
||
|
|
query = select(Contact).where(Contact.tenant_id == tenant_id)
|
||
|
|
result = await ps.apply_policy_filter(
|
||
|
|
db_session, query, "contact", user_id, tenant_id, Contact
|
||
|
|
)
|
||
|
|
|
||
|
|
# Query should be unchanged (no ABAC restriction)
|
||
|
|
rows = await db_session.execute(result)
|
||
|
|
contacts = rows.scalars().all()
|
||
|
|
assert len(contacts) >= 1
|
||
|
|
|
||
|
|
async def test_apply_policy_filter_with_allow(self, db_session: AsyncSession):
|
||
|
|
"""apply_policy_filter with allow policy filters correctly."""
|
||
|
|
seed = await seed_tenant_and_users(db_session)
|
||
|
|
tenant_id = seed["tenant_a"].id
|
||
|
|
user_id = seed["admin_a"].id
|
||
|
|
|
||
|
|
# Create allow policy for company-type contacts
|
||
|
|
await ps.create_policy(
|
||
|
|
db_session, tenant_id=tenant_id, name="Only companies",
|
||
|
|
entity_type="contact", principal_type="user",
|
||
|
|
principal_id=str(user_id), effect="allow",
|
||
|
|
conditions={
|
||
|
|
"operator": "AND",
|
||
|
|
"rules": [
|
||
|
|
{"field": "type", "op": "eq", "value": "company"}
|
||
|
|
]
|
||
|
|
},
|
||
|
|
)
|
||
|
|
|
||
|
|
query = select(Contact).where(Contact.tenant_id == tenant_id)
|
||
|
|
result = await ps.apply_policy_filter(
|
||
|
|
db_session, query, "contact", user_id, tenant_id, Contact
|
||
|
|
)
|
||
|
|
|
||
|
|
rows = await db_session.execute(result)
|
||
|
|
contacts = rows.scalars().all()
|
||
|
|
assert len(contacts) >= 1
|
||
|
|
for c in contacts:
|
||
|
|
assert c.type == "company", f"Expected 'company', got '{c.type}'"
|
||
|
|
|
||
|
|
async def test_apply_policy_filter_with_deny(self, db_session: AsyncSession):
|
||
|
|
"""apply_policy_filter with deny policy blocks matching entities."""
|
||
|
|
seed = await seed_tenant_and_users(db_session)
|
||
|
|
tenant_id = seed["tenant_a"].id
|
||
|
|
user_id = seed["admin_a"].id
|
||
|
|
|
||
|
|
# Create deny policy for contacts named 'Company Alpha'
|
||
|
|
await ps.create_policy(
|
||
|
|
db_session, tenant_id=tenant_id, name="Block Company Alpha",
|
||
|
|
entity_type="contact", principal_type="user",
|
||
|
|
principal_id=str(user_id), effect="deny",
|
||
|
|
conditions={
|
||
|
|
"operator": "AND",
|
||
|
|
"rules": [
|
||
|
|
{"field": "name", "op": "eq", "value": "Company Alpha"}
|
||
|
|
]
|
||
|
|
},
|
||
|
|
)
|
||
|
|
|
||
|
|
query = select(Contact).where(Contact.tenant_id == tenant_id)
|
||
|
|
result = await ps.apply_policy_filter(
|
||
|
|
db_session, query, "contact", user_id, tenant_id, Contact
|
||
|
|
)
|
||
|
|
|
||
|
|
rows = await db_session.execute(result)
|
||
|
|
contacts = rows.scalars().all()
|
||
|
|
# Company Alpha should be filtered out
|
||
|
|
names = [c.name for c in contacts]
|
||
|
|
assert "Company Alpha" not in names, "Deny policy should have blocked Company Alpha"
|
||
|
|
|
||
|
|
async def test_policy_priority_ordering(self, db_session: AsyncSession):
|
||
|
|
"""Policies are ordered by priority (higher first)."""
|
||
|
|
seed = await seed_tenant_and_users(db_session)
|
||
|
|
tenant_id = seed["tenant_a"].id
|
||
|
|
user_id = seed["admin_a"].id
|
||
|
|
|
||
|
|
await ps.create_policy(
|
||
|
|
db_session, tenant_id=tenant_id, name="Low Priority",
|
||
|
|
entity_type="contact", principal_type="user",
|
||
|
|
principal_id=str(user_id), effect="allow", priority=1,
|
||
|
|
)
|
||
|
|
await ps.create_policy(
|
||
|
|
db_session, tenant_id=tenant_id, name="High Priority",
|
||
|
|
entity_type="contact", principal_type="user",
|
||
|
|
principal_id=str(user_id), effect="allow", priority=100,
|
||
|
|
)
|
||
|
|
|
||
|
|
policies = await ps.list_policies(db_session, tenant_id, entity_type="contact")
|
||
|
|
# High priority should come first
|
||
|
|
high_idx = next(i for i, p in enumerate(policies) if p["name"] == "High Priority")
|
||
|
|
low_idx = next(i for i, p in enumerate(policies) if p["name"] == "Low Priority")
|
||
|
|
assert high_idx < low_idx, "High priority policy should come first"
|
||
|
|
|
||
|
|
async def test_policy_enabled_flag(self, db_session: AsyncSession):
|
||
|
|
"""Disabled policies are not applied."""
|
||
|
|
seed = await seed_tenant_and_users(db_session)
|
||
|
|
tenant_id = seed["tenant_a"].id
|
||
|
|
user_id = seed["admin_a"].id
|
||
|
|
|
||
|
|
policy = await ps.create_policy(
|
||
|
|
db_session, tenant_id=tenant_id, name="Disabled Policy",
|
||
|
|
entity_type="contact", principal_type="user",
|
||
|
|
principal_id=str(user_id), effect="allow",
|
||
|
|
)
|
||
|
|
|
||
|
|
# Disable the policy
|
||
|
|
updated = await ps.update_policy(
|
||
|
|
db_session, tenant_id, policy["id"], enabled=False
|
||
|
|
)
|
||
|
|
assert updated["enabled"] is False
|