Agent Zero
510 lines
13 KiB
Markdown
510 lines
13 KiB
Markdown
# 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,
|
|
)
|
|
```
|