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:

```python
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

```python
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

```python
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

```python
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

```python
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

```python
from app.plugins.manifest import PluginManifest

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

### In Your Plugin Class

```python
from app.plugins.base import BasePlugin

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

### Registering Routes for Permission Management

```python
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

```python
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

```python
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

```python
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

```python
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

```python
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

```python
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

```python
"""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,
    )
```