# 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) |

### Permission Name Schema (canonical)

All module-level permission strings follow the strict **2-segment schema**
`module:action`, with `*` wildcards allowed in either segment:

| Pattern | Meaning |
|---------|---------|
| `contacts:read` | Exact: read contacts |
| `contacts:*` | All actions on contacts |
| `*:read` | Read on all modules |
| `*:*` | Everything (superadmin) |

The runtime matcher (`app/core/permissions.py::_matches_permission`) compares
segment counts strictly — a 3-segment grant like `core:contacts:read` can never
match any 2-segment requirement and is therefore **invalid**. The plugin manifest
validator (`app/plugins/manifest.py`) rejects such patterns at load time.

Historical note (ARCH-008/009): migration 0019 seeded default roles with dead
3-segment patterns (`core:*:read` etc.); migration 0141 converts existing role
data to the canonical form.

**Catalog completeness (audit fix 2026-09-13):** 12 previously used-but-unregistered keys are now in the catalog so roles can actually be granted them:

- `CORE_PERMISSIONS` additions: `automation:admin`, `bank-accounts:read`, `bank-accounts:write`, `delegations:read`, `delegations:write`, `policies:read`, `policies:write`, `templates:read`, `templates:write`
- `permissions` plugin manifest: `permissions:read`, `permissions:admin`
- `forgejo_error_reporter` plugin manifest: `system:read`

Verified via AST scan (used keys vs. catalog): 146 registered, 0 missing.

**Entity-permission mapping fails closed (audit fix):** `get_entity_read_permission()` no longer falls back to `contacts:read` for unmapped entities — it returns the un-grantable sentinel `__unmapped__:read` (generic services then deny). Unknown entity types are rejected earlier with 422 by `validate_entity_type()`. Plugin-owned entities resolve correctly via `ENTITY_PLUGIN_OWNERS` (now populated through `register_entity_model(..., plugin_name=...)`).

---

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

```json
{
  "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

```json
{
  "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

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

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

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

```python
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:
```python
import logging
logging.getLogger("app.services.entity_permission_service").setLevel(logging.DEBUG)
logging.getLogger("app.services.policy_service").setLevel(logging.DEBUG)
```