leocrm/docs/permissions_plugin_dev.md

Permission System Plugin Development Guide

Version: 1.0
Date: 2026-07-29
Applies to: Plugin developers integrating with the LeoCRM permission system

---

1. Overview

Plugins can leverage the LeoCRM permission system to add row-level access control to their entities. This guide covers:

• Adding OwnedMixin to plugin models
• Using apply_visibility_filter() for list queries
• Registering entity types for permission management
• Defining field-level permissions
• Integrating with ABAC policies

---

2. Adding OwnedMixin to Plugin Models

To enable ownership tracking for your plugin's entities, add OwnedMixin to your SQLAlchemy model:

from app.models.owned_mixin import OwnedMixin
from app.core.db import Base, TenantMixin

class MyEntity(Base, TenantMixin, OwnedMixin):
    __tablename__ = "my_entities"

    id: Mapped[uuid.UUID] = mapped_column(
        PGUUID(as_uuid=True), primary_key=True, default=uuid.uuid4
    )
    name: Mapped[str] = mapped_column(String(200), nullable=False)
    # ... other fields

Owner ID Semantics

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

Setting Owner on Creation

from app.models.owned_mixin import OwnedMixin

async def create_entity(db: AsyncSession, data: dict, current_user: User):
    entity = MyEntity(
        tenant_id=current_user.tenant_id,
        owner_id=current_user.id,  # Set owner automatically
        **data,
    )
    db.add(entity)
    await db.commit()
    await db.refresh(entity)
    return entity

---

3. Using apply_visibility_filter()

The apply_visibility_filter() function filters a query to only return entities the user can see. This is the recommended way to implement list endpoints.

Basic Usage

from app.services import entity_permission_service as eps

async def list_entities(
    db: AsyncSession,
    tenant_id: uuid.UUID,
    user_id: uuid.UUID,
) -> list[MyEntity]:
    # Get visible entity IDs
    visible_ids, access_map = await eps.get_visible_ids(
        db, tenant_id, user_id, "my_entity"
    )

    if not visible_ids:
        return []

    # Query only visible entities
    result = await db.execute(
        select(MyEntity)
        .where(MyEntity.id.in_(visible_ids))
        .where(MyEntity.tenant_id == tenant_id)
    )
    return result.scalars().all()

With Caching

async def list_entities_cached(
    db: AsyncSession,
    redis: aioredis.Redis,
    tenant_id: uuid.UUID,
    user_id: uuid.UUID,
) -> list[MyEntity]:
    # Use cached version for better performance
    visible_ids, access_map = await eps.get_cached_visible_ids(
        db, redis, tenant_id, user_id, "my_entity"
    )

    if not visible_ids:
        return []

    result = await db.execute(
        select(MyEntity)
        .where(MyEntity.id.in_(visible_ids))
        .where(MyEntity.tenant_id == tenant_id)
    )
    return result.scalars().all()

With Access Level in Response

async def list_entities_with_access(
    db: AsyncSession,
    tenant_id: uuid.UUID,
    user_id: uuid.UUID,
) -> list[dict]:
    visible_ids, access_map = await eps.get_visible_ids(
        db, tenant_id, user_id, "my_entity"
    )

    if not visible_ids:
        return []

    result = await db.execute(
        select(MyEntity)
        .where(MyEntity.id.in_(visible_ids))
        .where(MyEntity.tenant_id == tenant_id)
    )
    entities = result.scalars().all()

    return [
        {
            **entity.to_dict(),
            "access_level": access_map.get(entity.id, "none"),
        }
        for entity in entities
    ]

---

4. Entity Registration

Register your entity type so it appears in the permission management UI and API.

In Your Plugin Manifest

from app.plugins.manifest import PluginManifest

class MyPluginManifest(PluginManifest):
    # ... other fields
    entity_types: list[str] = ["my_entity"]

In Your Plugin Class

from app.plugins.base import BasePlugin

class MyPlugin(BasePlugin):
    @property
    def entity_types(self) -> list[str]:
        return ["my_entity"]

Registering Routes for Permission Management

from fastapi import APIRouter, Depends

router = APIRouter(prefix="/api/v1/my-entities")

@router.get("/{entity_id}/permissions")
async def list_my_entity_permissions(
    entity_id: uuid.UUID,
    db: AsyncSession = Depends(get_db),
    current_user: User = Depends(get_current_user),
):
    """List all permissions for a specific entity."""
    return await eps.list_permissions(
        db, current_user.tenant_id, "my_entity", str(entity_id)
    )

@router.post("/{entity_id}/permissions")
async def create_my_entity_permission(
    entity_id: uuid.UUID,
    body: PermissionCreate,
    db: AsyncSession = Depends(get_db),
    current_user: User = Depends(get_current_user),
):
    """Grant permission to a user for this entity."""
    return await eps.create_permission(
        db,
        tenant_id=current_user.tenant_id,
        entity_type="my_entity",
        entity_id=str(entity_id),
        principal_type="user",
        principal_id=str(body.user_id),
        permission_level=body.access_level,
        expires_at=body.expires_at,
        created_by=current_user.id,
    )

---

5. Field Definitions

Define which fields of your entity are visible at each permission level.

Field Permission Schema

from pydantic import BaseModel
from typing import Any

class EntityFieldDefinition(BaseModel):
    """Define field visibility per permission level."""
    name: str
    type: str  # "string", "number", "boolean", "date", "reference"
    required: bool = False
    readable: dict[str, bool] = {
        "read": True,
        "write": True,
        "admin": True,
        "delete": True,
        "owner": True,
    }
    writable: dict[str, bool] = {
        "read": False,
        "write": True,
        "admin": True,
        "delete": True,
        "owner": True,
    }

Registering Field Definitions

from app.core.permission_registry import register_entity_fields

FIELD_DEFINITIONS = [
    EntityFieldDefinition(
        name="name",
        type="string",
        required=True,
        readable={"read": True, "write": True, "admin": True, "delete": True, "owner": True},
        writable={"read": False, "write": True, "admin": True, "delete": True, "owner": True},
    ),
    EntityFieldDefinition(
        name="sensitive_data",
        type="string",
        required=False,
        readable={"read": False, "write": False, "admin": True, "delete": True, "owner": True},
        writable={"read": False, "write": False, "admin": True, "delete": True, "owner": True},
    ),
]

# Register during plugin activation
register_entity_fields("my_entity", FIELD_DEFINITIONS)

---

6. Integrating with ABAC Policies

Your plugin can create and manage ABAC policies for its entities.

Creating Policies

from app.services import policy_service as ps

# Create an allow policy
policy = await ps.create_policy(
    db_session,
    tenant_id=tenant_id,
    name="Allow VIP my_entities",
    entity_type="my_entity",
    principal_type="user",
    principal_id=str(user_id),
    effect="allow",
    conditions={
        "operator": "AND",
        "rules": [
            {"field": "priority", "op": "gte", "value": 10},
        ]
    },
)

Applying Policies to Queries

query = select(MyEntity).where(MyEntity.tenant_id == tenant_id)
query = await ps.apply_policy_filter(
    db_session, query, "my_entity", user_id, tenant_id, MyEntity
)
result = await db_session.execute(query)
entities = result.scalars().all()

---

7. Best Practices

Do's

• ✅ Always set owner_id on entity creation
• ✅ Use get_visible_ids() for list endpoints
• ✅ Use batch_get_effective_access() for bulk operations
• ✅ Register entity types for permission management
• ✅ Define field-level permissions for sensitive data
• ✅ Use Redis caching for frequently accessed permissions
• ✅ Handle permission expiration gracefully

Don'ts

• ❌ Don't bypass permission checks for list endpoints
• ❌ Don't expose owner_id changes without authorization
• ❌ Don't create permissions without audit logging
• ❌ Don't forget to invalidate cache after permission changes
• ❌ Don't use get_effective_access() in loops — use batch instead

---

8. Testing

Test Fixtures

import pytest
from app.services import entity_permission_service as eps
from tests.conftest import seed_tenant_and_users

@pytest.mark.asyncio
async def test_my_entity_permissions(db_session):
    seed = await seed_tenant_and_users(db_session)
    tenant_id = seed["tenant_a"].id
    user_id = seed["admin_a"].id

    # Create entity with owner
    entity = MyEntity(
        tenant_id=tenant_id,
        owner_id=user_id,
        name="Test Entity",
    )
    db_session.add(entity)
    await db_session.commit()

    # Check access
    access = await eps.get_effective_access(
        db_session, tenant_id, user_id, "my_entity", entity.id
    )
    assert access == "owner"

Mocking Permissions

from unittest.mock import AsyncMock, patch

async def test_list_with_mock_permissions():
    with patch(
        "app.services.entity_permission_service.get_visible_ids",
        new=AsyncMock(return_value=({uuid.UUID(int=1)}, {uuid.UUID(int=1): "read"})),
    ):
        # Your test code
        pass

---

9. API Reference

Entity Permission Service

Function	Import	Description
get_effective_access()	from app.services import entity_permission_service as eps	Get access level for a user on a specific entity
get_visible_ids()	Same	Get all visible entity IDs for a user
batch_get_effective_access()	Same	Batch resolve access for multiple entities
check_entity_access()	Same	Check if user has at least required level
get_cached_visible_ids()	Same	Get visible IDs with Redis caching
create_permission()	Same	Create or update a permission entry
update_permission()	Same	Update an existing permission
delete_permission()	Same	Delete a permission entry
list_permissions()	Same	List all permissions for an entity

Policy Service

Function	Import	Description
create_policy()	from app.services import policy_service as ps	Create a new ABAC policy
update_policy()	Same	Update an existing policy
delete_policy()	Same	Delete a policy
list_policies()	Same	List policies for a tenant
build_sql_condition()	Same	Translate JSONB conditions to SQLAlchemy filters
apply_policy_filter()	Same	Apply ABAC policies to a query

---

10. Example: Complete Plugin Integration

"""Example plugin with full permission system integration."""

from __future__ import annotations

import uuid
from typing import Any

from sqlalchemy import String
from sqlalchemy.dialects.postgresql import UUID as PGUUID
from sqlalchemy.ext.asyncio import AsyncSession
from sqlalchemy.orm import Mapped, mapped_column

from app.core.db import Base, TenantMixin
from app.models.owned_mixin import OwnedMixin
from app.services import entity_permission_service as eps


class MyEntity(Base, TenantMixin, OwnedMixin):
    """Example entity with permission support."""

    __tablename__ = "my_entities"

    id: Mapped[uuid.UUID] = mapped_column(
        PGUUID(as_uuid=True), primary_key=True, default=uuid.uuid4
    )
    name: Mapped[str] = mapped_column(String(200), nullable=False)
    priority: Mapped[int] = mapped_column(nullable=False, default=0)


async def create_my_entity(
    db: AsyncSession,
    tenant_id: uuid.UUID,
    user_id: uuid.UUID,
    name: str,
    priority: int = 0,
) -> MyEntity:
    """Create a new entity with owner set."""
    entity = MyEntity(
        tenant_id=tenant_id,
        owner_id=user_id,
        name=name,
        priority=priority,
    )
    db.add(entity)
    await db.commit()
    await db.refresh(entity)
    return entity


async def list_visible_entities(
    db: AsyncSession,
    tenant_id: uuid.UUID,
    user_id: uuid.UUID,
) -> list[MyEntity]:
    """List entities visible to the user."""
    visible_ids, _ = await eps.get_visible_ids(
        db, tenant_id, user_id, "my_entity"
    )
    if not visible_ids:
        return []

    from sqlalchemy import select
    result = await db.execute(
        select(MyEntity)
        .where(MyEntity.id.in_(visible_ids))
        .where(MyEntity.tenant_id == tenant_id)
    )
    return result.scalars().all()


async def share_entity(
    db: AsyncSession,
    tenant_id: uuid.UUID,
    entity_id: uuid.UUID,
    target_user_id: uuid.UUID,
    level: str,
    created_by: uuid.UUID,
) -> dict:
    """Share an entity with another user."""
    return await eps.create_permission(
        db,
        tenant_id=tenant_id,
        entity_type="my_entity",
        entity_id=str(entity_id),
        principal_type="user",
        principal_id=str(target_user_id),
        permission_level=level,
        created_by=created_by,
    )