fix(arch-008,arch-009): canonical 2-segment permission schema enforced; fix dead role wildcard patterns
Check Cross-Plugin Imports / check (push) Has been cancelled
Check Cross-Plugin Imports / check (push) Has been cancelled
This commit is contained in:
@@ -0,0 +1,84 @@
|
||||
'''Fix role permission wildcard patterns to canonical 2-segment schema
|
||||
|
||||
Revision ID: 0141
|
||||
Revises: 0140
|
||||
Create Date: 2026-08-23
|
||||
|
||||
Migration 0019 seeded default roles with 3-segment permission patterns
|
||||
(core:*:read etc.). The runtime matcher (_matches_permission) compares
|
||||
segment counts strictly, so those patterns could never match any
|
||||
2-segment requirement - editor/viewer roles were silently dead.
|
||||
|
||||
Canonical schema is module:action (2 segments, * wildcards allowed).
|
||||
core:*:X means all modules with action X, so it converts to *:X.
|
||||
'''
|
||||
|
||||
from alembic import op
|
||||
|
||||
# revision identifiers, used by Alembic.
|
||||
revision = '0141'
|
||||
down_revision = '0140'
|
||||
branch_labels = None
|
||||
depends_on = None
|
||||
|
||||
# Rebuild the permissions JSONB object, rewriting every key that starts
|
||||
# with the dead 'core:' prefix to its 2-segment equivalent ('*:X').
|
||||
_UPGRADE_SQL = '''
|
||||
UPDATE roles
|
||||
SET permissions = sub.new_perms,
|
||||
permission_version = permission_version + 1
|
||||
FROM (
|
||||
SELECT
|
||||
r.id AS role_id,
|
||||
jsonb_object_agg(
|
||||
CASE WHEN k LIKE 'core:%'
|
||||
THEN '*:' || split_part(k, ':', 3)
|
||||
ELSE k END,
|
||||
v
|
||||
) AS new_perms
|
||||
FROM roles r,
|
||||
jsonb_each(r.permissions) AS e(k, v)
|
||||
GROUP BY r.id
|
||||
) AS sub
|
||||
WHERE roles.id = sub.role_id
|
||||
AND EXISTS (
|
||||
SELECT 1 FROM jsonb_object_keys(roles.permissions) k
|
||||
WHERE k LIKE 'core:%'
|
||||
)
|
||||
'''
|
||||
|
||||
# Reverse: map '*:X' back to 'core:*:X' only for keys that came from the
|
||||
# original seeding pattern. Roles that legitimately use '*:X' without a
|
||||
# matching 'core:*:X' history are left untouched (best-effort downgrade).
|
||||
_DOWNGRADE_SQL = '''
|
||||
UPDATE roles
|
||||
SET permissions = sub.new_perms,
|
||||
permission_version = permission_version + 1
|
||||
FROM (
|
||||
SELECT
|
||||
r.id AS role_id,
|
||||
jsonb_object_agg(
|
||||
CASE WHEN k = '*:' || split_part(k, ':', 2)
|
||||
AND k <> '*:*'
|
||||
THEN 'core:*:' || split_part(k, ':', 2)
|
||||
ELSE k END,
|
||||
v
|
||||
) AS new_perms
|
||||
FROM roles r,
|
||||
jsonb_each(r.permissions) AS e(k, v)
|
||||
GROUP BY r.id
|
||||
) AS sub
|
||||
WHERE roles.id = sub.role_id
|
||||
AND EXISTS (
|
||||
SELECT 1 FROM jsonb_object_keys(roles.permissions) k
|
||||
WHERE k = '*:' || split_part(k, ':', 2) AND k <> '*:*'
|
||||
)
|
||||
'''
|
||||
|
||||
|
||||
def upgrade() -> None:
|
||||
op.execute(_UPGRADE_SQL)
|
||||
|
||||
|
||||
def downgrade() -> None:
|
||||
op.execute(_DOWNGRADE_SQL)
|
||||
+20
-1
@@ -191,6 +191,25 @@ class PluginManifest(BaseModel):
|
||||
permissions: list[str] = Field(
|
||||
default_factory=list, description="Required permissions for this plugin"
|
||||
)
|
||||
|
||||
@field_validator("permissions")
|
||||
@classmethod
|
||||
def _validate_permission_format(cls, v: list[str]) -> list[str]:
|
||||
"""Enforce the canonical 2-segment permission schema (ARCH-008/009).
|
||||
|
||||
Canonical form is ``module:action`` with ``*`` wildcards allowed in
|
||||
either segment (e.g. ``contacts:read``, ``contacts:*``, ``*:read``,
|
||||
``*:*``). 3-segment names like ``core:contacts:read`` never match
|
||||
the runtime matcher and are rejected at manifest load time.
|
||||
"""
|
||||
for perm in v:
|
||||
parts = perm.split(":")
|
||||
if len(parts) != 2 or not all(parts):
|
||||
raise ValueError(
|
||||
f"Invalid permission {perm!r}: use 2-segment 'module:action' "
|
||||
f"(wildcards '*' allowed), e.g. 'contacts:read' or '*:*'"
|
||||
)
|
||||
return v
|
||||
is_core: bool = Field(
|
||||
default=False, description="Whether this is a core plugin that cannot be deactivated"
|
||||
)
|
||||
@@ -423,7 +442,7 @@ MANIFEST_SCHEMA_DOC = ManifestSchemaResponse(
|
||||
routes=[],
|
||||
events=["contact.created"],
|
||||
migrations=["0001_initial.sql"],
|
||||
permissions=["contacts.read"],
|
||||
permissions=["contacts:read"],
|
||||
menu_items=[
|
||||
FrontendMenuItem(
|
||||
label_key="nav.examplePlugin",
|
||||
|
||||
@@ -19,7 +19,7 @@ router = APIRouter(prefix="/api/v1/delegations", tags=["delegations"])
|
||||
async def list_delegations(
|
||||
direction: str = Query("all", pattern="^(from|to|all)$"),
|
||||
db: AsyncSession = Depends(get_db),
|
||||
current_user: dict = Depends(require_permission("permissions:delegations:read")),
|
||||
current_user: dict = Depends(require_permission("delegations:read")),
|
||||
):
|
||||
"""List delegations for the current user."""
|
||||
tenant_id = uuid.UUID(current_user["tenant_id"])
|
||||
@@ -32,7 +32,7 @@ async def list_delegations(
|
||||
async def create_delegation(
|
||||
body: DelegationCreate,
|
||||
db: AsyncSession = Depends(get_db),
|
||||
current_user: dict = Depends(require_permission("permissions:delegations:write")),
|
||||
current_user: dict = Depends(require_permission("delegations:write")),
|
||||
):
|
||||
"""Create a new permission delegation."""
|
||||
tenant_id = uuid.UUID(current_user["tenant_id"])
|
||||
@@ -56,7 +56,7 @@ async def update_delegation(
|
||||
delegation_id: str,
|
||||
body: DelegationUpdate,
|
||||
db: AsyncSession = Depends(get_db),
|
||||
current_user: dict = Depends(require_permission("permissions:delegations:write")),
|
||||
current_user: dict = Depends(require_permission("delegations:write")),
|
||||
):
|
||||
"""Update an existing delegation."""
|
||||
tenant_id = uuid.UUID(current_user["tenant_id"])
|
||||
@@ -78,7 +78,7 @@ async def update_delegation(
|
||||
async def delete_delegation(
|
||||
delegation_id: str,
|
||||
db: AsyncSession = Depends(get_db),
|
||||
current_user: dict = Depends(require_permission("permissions:delegations:write")),
|
||||
current_user: dict = Depends(require_permission("delegations:write")),
|
||||
):
|
||||
"""Delete a delegation."""
|
||||
tenant_id = uuid.UUID(current_user["tenant_id"])
|
||||
@@ -91,7 +91,7 @@ async def delete_delegation(
|
||||
@router.get("/active")
|
||||
async def check_active_delegation(
|
||||
db: AsyncSession = Depends(get_db),
|
||||
current_user: dict = Depends(require_permission("permissions:delegations:read")),
|
||||
current_user: dict = Depends(require_permission("delegations:read")),
|
||||
):
|
||||
"""Check if the current user has any active delegations."""
|
||||
tenant_id = uuid.UUID(current_user["tenant_id"])
|
||||
|
||||
@@ -23,7 +23,7 @@ router = APIRouter(prefix="/api/v1/permission-templates", tags=["permission-temp
|
||||
async def list_templates(
|
||||
entity_type: str | None = None,
|
||||
db: AsyncSession = Depends(get_db),
|
||||
current_user: dict = Depends(require_permission("permissions:templates:read")),
|
||||
current_user: dict = Depends(require_permission("templates:read")),
|
||||
):
|
||||
"""List all permission templates for the current tenant."""
|
||||
tenant_id = uuid.UUID(current_user["tenant_id"])
|
||||
@@ -35,7 +35,7 @@ async def list_templates(
|
||||
async def create_template(
|
||||
body: PermissionTemplateCreate,
|
||||
db: AsyncSession = Depends(get_db),
|
||||
current_user: dict = Depends(require_permission("permissions:templates:write")),
|
||||
current_user: dict = Depends(require_permission("templates:write")),
|
||||
):
|
||||
"""Create a new permission template."""
|
||||
tenant_id = uuid.UUID(current_user["tenant_id"])
|
||||
@@ -58,7 +58,7 @@ async def update_template(
|
||||
template_id: str,
|
||||
body: PermissionTemplateUpdate,
|
||||
db: AsyncSession = Depends(get_db),
|
||||
current_user: dict = Depends(require_permission("permissions:templates:write")),
|
||||
current_user: dict = Depends(require_permission("templates:write")),
|
||||
):
|
||||
"""Update an existing permission template."""
|
||||
tenant_id = uuid.UUID(current_user["tenant_id"])
|
||||
@@ -81,7 +81,7 @@ async def update_template(
|
||||
async def delete_template(
|
||||
template_id: str,
|
||||
db: AsyncSession = Depends(get_db),
|
||||
current_user: dict = Depends(require_permission("permissions:templates:write")),
|
||||
current_user: dict = Depends(require_permission("templates:write")),
|
||||
):
|
||||
"""Delete a permission template."""
|
||||
tenant_id = uuid.UUID(current_user["tenant_id"])
|
||||
@@ -95,7 +95,7 @@ async def delete_template(
|
||||
async def apply_template(
|
||||
body: PermissionTemplateApply,
|
||||
db: AsyncSession = Depends(get_db),
|
||||
current_user: dict = Depends(require_permission("permissions:templates:write")),
|
||||
current_user: dict = Depends(require_permission("templates:write")),
|
||||
):
|
||||
"""Apply a permission template to an entity, creating entity_permissions."""
|
||||
tenant_id = uuid.UUID(current_user["tenant_id"])
|
||||
|
||||
@@ -19,7 +19,7 @@ router = APIRouter(prefix="/api/v1/policies", tags=["policies"])
|
||||
async def list_policies(
|
||||
entity_type: str,
|
||||
db: AsyncSession = Depends(get_db),
|
||||
current_user: dict = Depends(require_permission("permissions:policies:read")),
|
||||
current_user: dict = Depends(require_permission("policies:read")),
|
||||
):
|
||||
"""List all ABAC policies for a given entity type."""
|
||||
tenant_id = uuid.UUID(current_user["tenant_id"])
|
||||
@@ -34,7 +34,7 @@ async def list_policies(
|
||||
async def create_policy(
|
||||
body: PolicyCreate,
|
||||
db: AsyncSession = Depends(get_db),
|
||||
current_user: dict = Depends(require_permission("permissions:policies:write")),
|
||||
current_user: dict = Depends(require_permission("policies:write")),
|
||||
):
|
||||
"""Create a new ABAC policy."""
|
||||
tenant_id = uuid.UUID(current_user["tenant_id"])
|
||||
@@ -59,7 +59,7 @@ async def update_policy(
|
||||
policy_id: str,
|
||||
body: PolicyUpdate,
|
||||
db: AsyncSession = Depends(get_db),
|
||||
current_user: dict = Depends(require_permission("permissions:policies:write")),
|
||||
current_user: dict = Depends(require_permission("policies:write")),
|
||||
):
|
||||
"""Update an existing ABAC policy."""
|
||||
tenant_id = uuid.UUID(current_user["tenant_id"])
|
||||
@@ -85,7 +85,7 @@ async def update_policy(
|
||||
async def delete_policy(
|
||||
policy_id: str,
|
||||
db: AsyncSession = Depends(get_db),
|
||||
current_user: dict = Depends(require_permission("permissions:policies:write")),
|
||||
current_user: dict = Depends(require_permission("policies:write")),
|
||||
):
|
||||
"""Delete an ABAC policy."""
|
||||
tenant_id = uuid.UUID(current_user["tenant_id"])
|
||||
|
||||
@@ -39,6 +39,27 @@ When checking access to an entity, the system resolves in this order (highest wi
|
||||
| `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.
|
||||
|
||||
---
|
||||
|
||||
## 2. Data Model
|
||||
|
||||
@@ -129,3 +129,71 @@ class TestGetFileMetadataAsync:
|
||||
|
||||
meta = get_file_metadata('nonexistent/sync.txt')
|
||||
assert meta == {'size': None, 'modified': None, 'exists': False}
|
||||
|
||||
|
||||
# --- ARCH-008/009: 2-segment permission canonical schema ---
|
||||
|
||||
|
||||
import re as _re
|
||||
|
||||
from app.core.permissions import _matches_permission
|
||||
|
||||
_THREE_SEGMENT = _re.compile(r'^[a-z_]+:[a-z_]+:[a-z_*]+$')
|
||||
|
||||
|
||||
class TestMatchesPermissionCanonical:
|
||||
def test_two_segment_wildcards(self):
|
||||
assert _matches_permission('*:*', 'contacts:read') is True
|
||||
assert _matches_permission('contacts:*', 'contacts:read') is True
|
||||
assert _matches_permission('*:read', 'contacts:read') is True
|
||||
assert _matches_permission('contacts:read', 'contacts:read') is True
|
||||
|
||||
def test_three_segment_grant_never_matches(self):
|
||||
assert _matches_permission('core:*:read', 'contacts:read') is False
|
||||
|
||||
def test_segment_count_mismatch(self):
|
||||
assert _matches_permission('contacts:read', 'contacts:read:extra') is False
|
||||
|
||||
|
||||
class TestRouteLiteralsTwoSegment:
|
||||
def test_no_three_segment_literals_in_routes(self):
|
||||
import pathlib
|
||||
|
||||
routes_dir = pathlib.Path('/a0/usr/projects/leocrm/app/routes')
|
||||
offenders = []
|
||||
for py_file in sorted(routes_dir.glob('*.py')):
|
||||
text = py_file.read_text()
|
||||
for m in _re.finditer(r'require_permission\(([^)]*)\)', text):
|
||||
arg = m.group(1).strip()
|
||||
if not arg:
|
||||
continue
|
||||
value = arg.strip('\'"')
|
||||
if ':' in value and value.count(':') != 1:
|
||||
offenders.append(f'{py_file.name}: {value}')
|
||||
assert offenders == [], f'3-segment permission literals found: {offenders}'
|
||||
|
||||
|
||||
class TestManifestPermissionValidator:
|
||||
def test_valid_permissions_accepted(self):
|
||||
from app.plugins.manifest import PluginManifest
|
||||
|
||||
manifest = PluginManifest(
|
||||
name='x',
|
||||
version='1.0.0',
|
||||
display_name='X',
|
||||
permissions=['contacts:read', '*:*'],
|
||||
)
|
||||
assert manifest.permissions == ['contacts:read', '*:*']
|
||||
|
||||
def test_three_segment_rejected(self):
|
||||
from pydantic import ValidationError
|
||||
|
||||
from app.plugins.manifest import PluginManifest
|
||||
|
||||
with pytest.raises(ValidationError):
|
||||
PluginManifest(
|
||||
name='x',
|
||||
version='1.0.0',
|
||||
display_name='X',
|
||||
permissions=['core:contacts:read'],
|
||||
)
|
||||
|
||||
Reference in New Issue
Block a user