leocrm/docs/permissions.md

LeoCRM Permission System

Version: 1.0
Date: 2026-07-29
Applies to: All developers and system administrators

---

1. Architecture Overview

The LeoCRM permission system is a multi-layered access control framework that combines:

• Row-Level Ownership — Each entity can have an owner_id (user who owns it)
• Entity Permissions (ACL) — Explicit permission entries for users, groups, or roles
• ABAC Policies — Attribute-based policies for fine-grained access control
• Role-Based Access Control (RBAC) — Module-level permissions via user roles
• System Admin Override — System administrators see everything

Permission Resolution Order

When checking access to an entity, the system resolves in this order (highest wins):

1. System Admin → delete (full access to everything)
2. Owner → owner (from owner_id on the entity)
3. Direct User Permission → explicit ACL entry for the user
4. Group Permission → ACL entry for a group the user belongs to
5. Role Permission → ACL entry for the user's role
6. Tenant-Owned → read (if owner_id IS NULL, visible to all with module permission)
7. No Access → none

Permission Levels

Level	Value	Description
none	0	Explicit deny (overrides allow)
read	1	View the entity
write	2	Read + edit entity fields
admin	3	Write + delete + manage permissions
delete	4	Admin + transfer ownership
owner	5	Full control (automatic for owner)

---

2. Data Model

EntityPermission

Stored in the entity_permissions table:

Column	Type	Description
id	UUID	Primary key
tenant_id	UUID	Tenant scope
entity_type	String(50)	Entity type (e.g., 'contact', 'dms_file')
entity_id	UUID	The specific entity
principal_type	String(10)	'user', 'group', 'role', 'guest'
principal_id	UUID	The user/group/role ID
permission_level	String(20)	'none', 'read', 'write', 'admin', 'delete'
expires_at	DateTime	Optional expiration
created_by	UUID	Who created this permission
created_at	DateTime	Creation timestamp
updated_at	DateTime	Last update timestamp

EntityPolicy (ABAC)

Stored in the entity_policies table:

Column	Type	Description
id	UUID	Primary key
tenant_id	UUID	Tenant scope
name	String(200)	Policy name
entity_type	String(50)	Target entity type
principal_type	String(10)	'user', 'group', 'role'
principal_id	UUID	Target principal
effect	String(10)	'allow' or 'deny'
conditions	JSONB	Attribute-based conditions
priority	Integer	Evaluation priority (higher = first)
enabled	Boolean	Whether the policy is active
created_at	DateTime	Creation timestamp
updated_at	DateTime	Last update timestamp

OwnedMixin

Adds owner_id to any model:

• NULL → Tenant-owned (visible to all with module permission)
• UUID → Owned by that user
• Set automatically by service layer on creation
• Transfer requires owner, admin, or system_admin role

---

3. Entity Permissions API

List Permissions

GET /api/v1/{entity_type}/{entity_id}/permissions

Returns all permission entries for an entity.

Create/Update Permission

POST /api/v1/{entity_type}/{entity_id}/permissions

{
  "user_id": "uuid",
  "access_level": "read",
  "expires_at": "2026-12-31T23:59:59Z"
}

Revoke Permission

DELETE /api/v1/{entity_type}/{entity_id}/permissions/{user_id}

Check Access

GET /api/v1/{entity_type}/{entity_id}/access?user_id={uuid}

Returns the effective access level for a user.

---

4. ABAC Policies

Policy Conditions Format

{
  "operator": "AND",
  "rules": [
    {"field": "status", "op": "eq", "value": "active"},
    {"field": "amount", "op": "gte", "value": 1000},
    {"field": "tags", "op": "contains", "value": "vip"}
  ]
}

Supported Operators

Operator	Description	Example
eq	Equals	{"field": "type", "op": "eq", "value": "company"}
neq	Not equals	{"field": "status", "op": "neq", "value": "archived"}
gt	Greater than	{"field": "amount", "op": "gt", "value": 100}
gte	Greater or equal	{"field": "amount", "op": "gte", "value": 50}
lt	Less than	{"field": "amount", "op": "lt", "value": 10000}
lte	Less or equal	{"field": "amount", "op": "lte", "value": 500}
in	In list	{"field": "status", "op": "in", "value": ["active", "pending"]}
not_in	Not in list	{"field": "status", "op": "not_in", "value": ["deleted"]}
contains	String contains	{"field": "name", "op": "contains", "value": "VIP"}
starts_with	String starts with	{"field": "name", "op": "starts_with", "value": "Confidential"}
is_null	Is NULL	{"field": "email", "op": "is_null"}
is_not_null	Is not NULL	{"field": "email", "op": "is_not_null"}

Policy Evaluation

1. Allow policies: OR-joined (at least one must match for access)
2. Deny policies: NOT (none may match — deny takes precedence)
3. Priority: Higher priority policies evaluated first
4. Enabled flag: Disabled policies are skipped

---

5. Service Layer

Entity Permission Service (app/services/entity_permission_service.py)

Function	Description
get_effective_access()	Get access level for a user on a specific entity
get_visible_ids()	Get all visible entity IDs for a user
batch_get_effective_access()	Batch resolve access for multiple entities
check_entity_access()	Check if user has at least required level
get_cached_visible_ids()	Get visible IDs with Redis caching
create_permission()	Create or update a permission entry
update_permission()	Update an existing permission
delete_permission()	Delete a permission entry
list_permissions()	List all permissions for an entity
list_all_permissions()	List all permissions for a tenant
cleanup_expired_permissions()	Remove expired permission entries
invalidate_all_user_entity_cache()	Clear all cached permissions for a user
get_permission_analytics()	Get permission statistics

Policy Service (app/services/policy_service.py)

Function	Description
create_policy()	Create a new ABAC policy
update_policy()	Update an existing policy
delete_policy()	Delete a policy
list_policies()	List policies for a tenant
build_sql_condition()	Translate JSONB conditions to SQLAlchemy filters
apply_policy_filter()	Apply ABAC policies to a query

---

6. Caching

The permission system uses Redis for caching visibility results:

• Cache key: ep_vis:{user_id}:{tenant_id}:{entity_type}
• Cache value: JSON with visible_ids and access_map
• TTL: 5 minutes (300 seconds)
• Invalidation: Automatic on permission create/update/delete

Cache Flow

1. Check Redis cache for user + entity type
2. Cache hit → return cached visible IDs
3. Cache miss → resolve from database, store in cache
4. Permission changes → invalidate affected user caches

---

7. Performance Considerations

• Batch resolution (batch_get_effective_access) is preferred over individual get_effective_access calls
• Redis caching reduces database load for repeated visibility checks
• Bitmap optimization for large entity sets (planned)
• Indexes on entity_type + entity_id, principal_type + principal_id, tenant_id, expires_at
• Partitioning recommended for entity_permissions table at scale

---

8. Security Considerations

• Deny takes precedence over allow in both ACL and ABAC
• Expired permissions are automatically excluded from resolution
• System admin bypasses all permission checks
• Audit logging for all permission changes
• Notifications sent to users when permissions are granted/revoked
• Tenant isolation enforced via tenant_id on all permission entries

---

9. Examples

Grant Read Access to a User

from app.services import entity_permission_service as eps

await eps.create_permission(
    db_session,
    tenant_id=tenant_id,
    entity_type="contact",
    entity_id=str(contact_id),
    principal_type="user",
    principal_id=str(user_id),
    permission_level="read",
    created_by=current_user.id,
)

Check Access

access = await eps.get_effective_access(
    db_session, tenant_id, user_id, "contact", contact_id
)
if access in ("read", "write", "admin", "delete", "owner"):
    # User has access
    pass

Create ABAC Policy

from app.services import policy_service as ps

policy = await ps.create_policy(
    db_session,
    tenant_id=tenant_id,
    name="VIP Only",
    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"},
        ]
    },
)

Batch Resolve Access

result = await eps.batch_get_effective_access(
    db_session, tenant_id, user_id, "contact", entity_ids
)
for entity_id, level in result.items():
    print(f"Entity {entity_id}: {level}")

---

10. Troubleshooting

Common Issues

Issue	Cause	Solution
User sees nothing	No permissions, not owner, not system admin	Grant explicit permission or set owner_id
Expired permission still works	Cache not invalidated	Wait for TTL or invalidate cache manually
ABAC policy not applied	Policy disabled or no conditions match	Check enabled flag and conditions
System admin can't see entity	Entity deleted or wrong tenant	Check deleted_at and tenant_id
Permission creation fails	Duplicate unique constraint	Use upsert (create_permission handles this)

Debugging

Enable debug logging:

import logging
logging.getLogger("app.services.entity_permission_service").setLevel(logging.DEBUG)
logging.getLogger("app.services.policy_service").setLevel(logging.DEBUG)