Agent Zero
7c8f2a2222
- Integrate ABAC policies into apply_visibility_filter() (allow/deny with priority) - Add field whitelist (ABAC_ALLOWED_FIELDS) for build_sql_condition() security - Add request-level ContextVar for user principals (group_ids, role_id) - Set principals in deps.py (session + bearer auth) - Use ContextVar in visibility.py and permission_resolver.py (N+1 fix) - Add version validation to get_cached_visible_ids() (cache strategy unification) - Deactivate delegation route (parked — not integrated into resolve_permissions) - Add 7 ABAC integration tests All 70 tests pass (7 ABAC + 63 existing). No regressions.
398 lines
13 KiB
Python
398 lines
13 KiB
Python
"""Tests for ABAC (Attribute-Based Access Control) integration.
|
|
|
|
These tests verify that ABAC policies are correctly applied in the
|
|
visibility filter alongside the existing owner/shared visibility logic.
|
|
|
|
Requires a PostgreSQL test database — uses the same conftest.py fixtures
|
|
as the rest of the test suite (db_session, seed_tenant_and_users).
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import uuid
|
|
|
|
import pytest
|
|
from sqlalchemy import select
|
|
|
|
from app.core.visibility import apply_visibility_filter
|
|
from app.models.contact import Contact
|
|
from app.models.entity_policy import EntityPolicy
|
|
|
|
pytestmark = pytest.mark.asyncio
|
|
|
|
|
|
# ─── Helpers ──────────────────────────────────────────────────────────
|
|
|
|
|
|
def _make_contact(
|
|
tenant_id: uuid.UUID,
|
|
name: str,
|
|
status: str = "lead",
|
|
country: str | None = None,
|
|
owner_id: uuid.UUID | None = None,
|
|
created_by: uuid.UUID | None = None,
|
|
) -> Contact:
|
|
"""Create a Contact instance for testing."""
|
|
return Contact(
|
|
tenant_id=tenant_id,
|
|
type="company",
|
|
name=name,
|
|
displayname=name,
|
|
status=status,
|
|
country=country,
|
|
owner_id=owner_id,
|
|
created_by=created_by,
|
|
updated_by=created_by,
|
|
)
|
|
|
|
|
|
def _make_policy(
|
|
tenant_id: uuid.UUID,
|
|
name: str,
|
|
entity_type: str,
|
|
principal_type: str,
|
|
principal_id: uuid.UUID,
|
|
effect: str = "allow",
|
|
conditions: dict | None = None,
|
|
priority: int = 0,
|
|
) -> EntityPolicy:
|
|
"""Create an EntityPolicy instance for testing."""
|
|
return EntityPolicy(
|
|
tenant_id=tenant_id,
|
|
name=name,
|
|
entity_type=entity_type,
|
|
principal_type=principal_type,
|
|
principal_id=principal_id,
|
|
effect=effect,
|
|
conditions=conditions,
|
|
priority=priority,
|
|
enabled=True,
|
|
)
|
|
|
|
|
|
async def _seed_and_query(
|
|
db_session,
|
|
contacts: list[Contact],
|
|
policies: list[EntityPolicy],
|
|
user_id: uuid.UUID,
|
|
tenant_id: uuid.UUID,
|
|
):
|
|
"""Seed contacts + policies, then run visibility filter and return results."""
|
|
db_session.add_all(contacts)
|
|
db_session.add_all(policies)
|
|
await db_session.flush()
|
|
|
|
query = select(Contact).where(Contact.tenant_id == tenant_id)
|
|
query = await apply_visibility_filter(
|
|
db_session, query, "contact", Contact, user_id, tenant_id
|
|
)
|
|
result = await db_session.execute(query)
|
|
return result.scalars().all()
|
|
|
|
|
|
# ─── Tests ────────────────────────────────────────────────────────────
|
|
|
|
|
|
async def test_abac_deny_policy_blocks_access(db_session):
|
|
"""A deny policy matching specific rows must exclude those rows."""
|
|
from tests.conftest import seed_tenant_and_users
|
|
|
|
seed = await seed_tenant_and_users(db_session)
|
|
tenant_id = seed["tenant_a"].id
|
|
viewer_id = seed["viewer_a"].id
|
|
|
|
# Two contacts owned by viewer — one VIP, one regular
|
|
vip = _make_contact(tenant_id, "VIP Corp", status="vip", owner_id=viewer_id, created_by=viewer_id)
|
|
regular = _make_contact(tenant_id, "Regular Inc", status="lead", owner_id=viewer_id, created_by=viewer_id)
|
|
|
|
# Deny policy: block contacts where status = 'vip'
|
|
deny_policy = _make_policy(
|
|
tenant_id,
|
|
"deny-vip",
|
|
"contact",
|
|
"user",
|
|
viewer_id,
|
|
effect="deny",
|
|
conditions={
|
|
"operator": "AND",
|
|
"rules": [{"field": "status", "op": "eq", "value": "vip"}],
|
|
},
|
|
priority=10,
|
|
)
|
|
|
|
results = await _seed_and_query(db_session, [vip, regular], [deny_policy], viewer_id, tenant_id)
|
|
|
|
names = {r.name for r in results}
|
|
assert "Regular Inc" in names, "Regular contact should be visible"
|
|
assert "VIP Corp" not in names, "VIP contact should be blocked by deny policy"
|
|
|
|
|
|
async def test_abac_allow_policy_grants_access(db_session):
|
|
"""An allow policy matching specific rows must include those rows even if not owned."""
|
|
from tests.conftest import seed_tenant_and_users
|
|
|
|
seed = await seed_tenant_and_users(db_session)
|
|
tenant_id = seed["tenant_a"].id
|
|
admin_id = seed["admin_a"].id
|
|
viewer_id = seed["viewer_a"].id
|
|
|
|
# Contact owned by admin — viewer would not normally see it
|
|
owned_by_admin = _make_contact(
|
|
tenant_id, "Admin Secret", status="lead", owner_id=admin_id, created_by=admin_id
|
|
)
|
|
# Tenant-owned contact (owner_id=None) — viewer can see via tenant-owned rule
|
|
tenant_owned = _make_contact(tenant_id, "Public Co", status="lead", owner_id=None, created_by=admin_id)
|
|
|
|
# Allow policy: viewer can see contacts with status='lead'
|
|
allow_policy = _make_policy(
|
|
tenant_id,
|
|
"allow-lead",
|
|
"contact",
|
|
"user",
|
|
viewer_id,
|
|
effect="allow",
|
|
conditions={
|
|
"operator": "AND",
|
|
"rules": [{"field": "status", "op": "eq", "value": "lead"}],
|
|
},
|
|
priority=5,
|
|
)
|
|
|
|
results = await _seed_and_query(db_session, [owned_by_admin, tenant_owned], [allow_policy], viewer_id, tenant_id)
|
|
|
|
names = {r.name for r in results}
|
|
assert "Public Co" in names, "Tenant-owned contact should be visible"
|
|
assert "Admin Secret" in names, "Admin-owned contact should be visible via allow policy"
|
|
|
|
|
|
async def test_abac_policy_with_conditions(db_session):
|
|
"""Policy with JSONB conditions (status='vip') filters correctly."""
|
|
from tests.conftest import seed_tenant_and_users
|
|
|
|
seed = await seed_tenant_and_users(db_session)
|
|
tenant_id = seed["tenant_a"].id
|
|
viewer_id = seed["viewer_a"].id
|
|
|
|
vip1 = _make_contact(tenant_id, "VIP One", status="vip", owner_id=viewer_id, created_by=viewer_id)
|
|
vip2 = _make_contact(tenant_id, "VIP Two", status="vip", owner_id=viewer_id, created_by=viewer_id)
|
|
regular = _make_contact(tenant_id, "Regular Co", status="lead", owner_id=viewer_id, created_by=viewer_id)
|
|
|
|
# Deny policy with conditions: block where status='vip'
|
|
deny_policy = _make_policy(
|
|
tenant_id,
|
|
"deny-vip-conditional",
|
|
"contact",
|
|
"user",
|
|
viewer_id,
|
|
effect="deny",
|
|
conditions={
|
|
"operator": "AND",
|
|
"rules": [{"field": "status", "op": "eq", "value": "vip"}],
|
|
},
|
|
priority=10,
|
|
)
|
|
|
|
results = await _seed_and_query(db_session, [vip1, vip2, regular], [deny_policy], viewer_id, tenant_id)
|
|
|
|
names = {r.name for r in results}
|
|
assert "Regular Co" in names, "Regular contact should be visible"
|
|
assert "VIP One" not in names, "VIP One should be blocked"
|
|
assert "VIP Two" not in names, "VIP Two should be blocked"
|
|
|
|
|
|
async def test_abac_policy_priority(db_session):
|
|
"""Higher priority deny policy takes precedence over allow policy."""
|
|
from tests.conftest import seed_tenant_and_users
|
|
|
|
seed = await seed_tenant_and_users(db_session)
|
|
tenant_id = seed["tenant_a"].id
|
|
viewer_id = seed["viewer_a"].id
|
|
|
|
# Contact owned by viewer with status='vip'
|
|
contact = _make_contact(tenant_id, "Priority Test", status="vip", owner_id=viewer_id, created_by=viewer_id)
|
|
|
|
# Allow policy (low priority): allow status='vip'
|
|
allow_policy = _make_policy(
|
|
tenant_id,
|
|
"allow-vip-low",
|
|
"contact",
|
|
"user",
|
|
viewer_id,
|
|
effect="allow",
|
|
conditions={
|
|
"operator": "AND",
|
|
"rules": [{"field": "status", "op": "eq", "value": "vip"}],
|
|
},
|
|
priority=1,
|
|
)
|
|
|
|
# Deny policy (high priority): deny status='vip'
|
|
deny_policy = _make_policy(
|
|
tenant_id,
|
|
"deny-vip-high",
|
|
"contact",
|
|
"user",
|
|
viewer_id,
|
|
effect="deny",
|
|
conditions={
|
|
"operator": "AND",
|
|
"rules": [{"field": "status", "op": "eq", "value": "vip"}],
|
|
},
|
|
priority=100,
|
|
)
|
|
|
|
results = await _seed_and_query(db_session, [contact], [allow_policy, deny_policy], viewer_id, tenant_id)
|
|
|
|
names = {r.name for r in results}
|
|
assert "Priority Test" not in names, (
|
|
"Deny policy with higher priority should block the contact despite allow policy"
|
|
)
|
|
|
|
|
|
async def test_abac_field_whitelist_blocks_unknown_field(db_session):
|
|
"""Policy with a non-whitelisted field should be skipped (not crash)."""
|
|
from tests.conftest import seed_tenant_and_users
|
|
|
|
seed = await seed_tenant_and_users(db_session)
|
|
tenant_id = seed["tenant_a"].id
|
|
viewer_id = seed["viewer_a"].id
|
|
|
|
contact = _make_contact(tenant_id, "Whitelist Test", status="lead", owner_id=viewer_id, created_by=viewer_id)
|
|
|
|
# Deny policy referencing 'password_hash' — not in whitelist
|
|
deny_policy = _make_policy(
|
|
tenant_id,
|
|
"deny-bad-field",
|
|
"contact",
|
|
"user",
|
|
viewer_id,
|
|
effect="deny",
|
|
conditions={
|
|
"operator": "AND",
|
|
"rules": [{"field": "password_hash", "op": "eq", "value": "secret"}],
|
|
},
|
|
priority=10,
|
|
)
|
|
|
|
results = await _seed_and_query(db_session, [contact], [deny_policy], viewer_id, tenant_id)
|
|
|
|
names = {r.name for r in results}
|
|
assert "Whitelist Test" in names, (
|
|
"Contact should remain visible because the deny policy with a non-whitelisted "
|
|
"field was skipped"
|
|
)
|
|
|
|
|
|
async def test_abac_does_not_break_owner_visibility(db_session):
|
|
"""Owner remains visible even when a deny policy exists for other rows."""
|
|
from tests.conftest import seed_tenant_and_users
|
|
|
|
seed = await seed_tenant_and_users(db_session)
|
|
tenant_id = seed["tenant_a"].id
|
|
viewer_id = seed["viewer_a"].id
|
|
|
|
# Contact owned by viewer with status='lead' (not matching deny)
|
|
own_contact = _make_contact(tenant_id, "My Own", status="lead", owner_id=viewer_id, created_by=viewer_id)
|
|
# Contact owned by viewer with status='vip' (matching deny)
|
|
vip_contact = _make_contact(tenant_id, "My VIP", status="vip", owner_id=viewer_id, created_by=viewer_id)
|
|
|
|
# Deny policy: block status='vip'
|
|
deny_policy = _make_policy(
|
|
tenant_id,
|
|
"deny-vip-owner-test",
|
|
"contact",
|
|
"user",
|
|
viewer_id,
|
|
effect="deny",
|
|
conditions={
|
|
"operator": "AND",
|
|
"rules": [{"field": "status", "op": "eq", "value": "vip"}],
|
|
},
|
|
priority=10,
|
|
)
|
|
|
|
results = await _seed_and_query(db_session, [own_contact, vip_contact], [deny_policy], viewer_id, tenant_id)
|
|
|
|
names = {r.name for r in results}
|
|
assert "My Own" in names, "Owner's own contact (non-matching deny) should be visible"
|
|
assert "My VIP" not in names, "Owner's VIP contact should be blocked by deny policy"
|
|
|
|
|
|
async def test_abac_does_not_break_shared_visibility(db_session):
|
|
"""Shared entities remain visible even when a deny policy exists for other rows."""
|
|
from tests.conftest import seed_tenant_and_users
|
|
from app.models.entity_permission import EntityPermission
|
|
|
|
seed = await seed_tenant_and_users(db_session)
|
|
tenant_id = seed["tenant_a"].id
|
|
admin_id = seed["admin_a"].id
|
|
viewer_id = seed["viewer_a"].id
|
|
|
|
# Contact owned by admin, shared with viewer
|
|
shared_contact = _make_contact(
|
|
tenant_id, "Shared Contact", status="lead", owner_id=admin_id, created_by=admin_id
|
|
)
|
|
# Contact owned by admin, NOT shared, but allow policy grants access
|
|
allow_contact = _make_contact(
|
|
tenant_id, "Allow Contact", status="vip", owner_id=admin_id, created_by=admin_id
|
|
)
|
|
|
|
# Share the first contact with viewer via entity_permissions
|
|
db_session.add_all([shared_contact, allow_contact])
|
|
await db_session.flush()
|
|
|
|
permission = EntityPermission(
|
|
tenant_id=tenant_id,
|
|
entity_type="contact",
|
|
entity_id=shared_contact.id,
|
|
principal_type="user",
|
|
principal_id=viewer_id,
|
|
permission_level="read",
|
|
)
|
|
db_session.add(permission)
|
|
|
|
# Deny policy: deny status='vip' (should block allow_contact but not shared_contact)
|
|
deny_policy = _make_policy(
|
|
tenant_id,
|
|
"deny-vip-shared-test",
|
|
"contact",
|
|
"user",
|
|
viewer_id,
|
|
effect="deny",
|
|
conditions={
|
|
"operator": "AND",
|
|
"rules": [{"field": "status", "op": "eq", "value": "vip"}],
|
|
},
|
|
priority=10,
|
|
)
|
|
# Allow policy: allow status='vip' (would grant access to allow_contact)
|
|
allow_policy = _make_policy(
|
|
tenant_id,
|
|
"allow-vip-shared-test",
|
|
"contact",
|
|
"user",
|
|
viewer_id,
|
|
effect="allow",
|
|
conditions={
|
|
"operator": "AND",
|
|
"rules": [{"field": "status", "op": "eq", "value": "vip"}],
|
|
},
|
|
priority=5,
|
|
)
|
|
db_session.add_all([deny_policy, allow_policy])
|
|
await db_session.flush()
|
|
|
|
query = select(Contact).where(Contact.tenant_id == tenant_id)
|
|
query = await apply_visibility_filter(
|
|
db_session, query, "contact", Contact, viewer_id, tenant_id
|
|
)
|
|
result = await db_session.execute(query)
|
|
results = result.scalars().all()
|
|
|
|
names = {r.name for r in results}
|
|
assert "Shared Contact" in names, "Shared contact should be visible (deny doesn't match it)"
|
|
assert "Allow Contact" not in names, (
|
|
"Allow contact should be blocked: deny policy (status=vip) takes precedence "
|
|
"over allow policy"
|
|
)
|