sprint20-23: tests + documentation + guest access + infrastructure + migrations 0059

2026-07-29 02:53:37 +02:00
parent ddf73ee42e
commit 24690fb674
20 changed files with 3898 additions and 2 deletions
@@ -0,0 +1,509 @@
+# 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,
+    )
+```