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

This commit is contained in:
Agent Zero
2026-08-23 18:35:50 +02:00
parent 17516d2783
commit 795307754f
7 changed files with 207 additions and 15 deletions
@@ -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
View File
@@ -191,6 +191,25 @@ class PluginManifest(BaseModel):
permissions: list[str] = Field( permissions: list[str] = Field(
default_factory=list, description="Required permissions for this plugin" 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( is_core: bool = Field(
default=False, description="Whether this is a core plugin that cannot be deactivated" default=False, description="Whether this is a core plugin that cannot be deactivated"
) )
@@ -423,7 +442,7 @@ MANIFEST_SCHEMA_DOC = ManifestSchemaResponse(
routes=[], routes=[],
events=["contact.created"], events=["contact.created"],
migrations=["0001_initial.sql"], migrations=["0001_initial.sql"],
permissions=["contacts.read"], permissions=["contacts:read"],
menu_items=[ menu_items=[
FrontendMenuItem( FrontendMenuItem(
label_key="nav.examplePlugin", label_key="nav.examplePlugin",
+5 -5
View File
@@ -19,7 +19,7 @@ router = APIRouter(prefix="/api/v1/delegations", tags=["delegations"])
async def list_delegations( async def list_delegations(
direction: str = Query("all", pattern="^(from|to|all)$"), direction: str = Query("all", pattern="^(from|to|all)$"),
db: AsyncSession = Depends(get_db), 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.""" """List delegations for the current user."""
tenant_id = uuid.UUID(current_user["tenant_id"]) tenant_id = uuid.UUID(current_user["tenant_id"])
@@ -32,7 +32,7 @@ async def list_delegations(
async def create_delegation( async def create_delegation(
body: DelegationCreate, body: DelegationCreate,
db: AsyncSession = Depends(get_db), 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.""" """Create a new permission delegation."""
tenant_id = uuid.UUID(current_user["tenant_id"]) tenant_id = uuid.UUID(current_user["tenant_id"])
@@ -56,7 +56,7 @@ async def update_delegation(
delegation_id: str, delegation_id: str,
body: DelegationUpdate, body: DelegationUpdate,
db: AsyncSession = Depends(get_db), 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.""" """Update an existing delegation."""
tenant_id = uuid.UUID(current_user["tenant_id"]) tenant_id = uuid.UUID(current_user["tenant_id"])
@@ -78,7 +78,7 @@ async def update_delegation(
async def delete_delegation( async def delete_delegation(
delegation_id: str, delegation_id: str,
db: AsyncSession = Depends(get_db), 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.""" """Delete a delegation."""
tenant_id = uuid.UUID(current_user["tenant_id"]) tenant_id = uuid.UUID(current_user["tenant_id"])
@@ -91,7 +91,7 @@ async def delete_delegation(
@router.get("/active") @router.get("/active")
async def check_active_delegation( async def check_active_delegation(
db: AsyncSession = Depends(get_db), 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.""" """Check if the current user has any active delegations."""
tenant_id = uuid.UUID(current_user["tenant_id"]) tenant_id = uuid.UUID(current_user["tenant_id"])
+5 -5
View File
@@ -23,7 +23,7 @@ router = APIRouter(prefix="/api/v1/permission-templates", tags=["permission-temp
async def list_templates( async def list_templates(
entity_type: str | None = None, entity_type: str | None = None,
db: AsyncSession = Depends(get_db), 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.""" """List all permission templates for the current tenant."""
tenant_id = uuid.UUID(current_user["tenant_id"]) tenant_id = uuid.UUID(current_user["tenant_id"])
@@ -35,7 +35,7 @@ async def list_templates(
async def create_template( async def create_template(
body: PermissionTemplateCreate, body: PermissionTemplateCreate,
db: AsyncSession = Depends(get_db), 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.""" """Create a new permission template."""
tenant_id = uuid.UUID(current_user["tenant_id"]) tenant_id = uuid.UUID(current_user["tenant_id"])
@@ -58,7 +58,7 @@ async def update_template(
template_id: str, template_id: str,
body: PermissionTemplateUpdate, body: PermissionTemplateUpdate,
db: AsyncSession = Depends(get_db), 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.""" """Update an existing permission template."""
tenant_id = uuid.UUID(current_user["tenant_id"]) tenant_id = uuid.UUID(current_user["tenant_id"])
@@ -81,7 +81,7 @@ async def update_template(
async def delete_template( async def delete_template(
template_id: str, template_id: str,
db: AsyncSession = Depends(get_db), 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.""" """Delete a permission template."""
tenant_id = uuid.UUID(current_user["tenant_id"]) tenant_id = uuid.UUID(current_user["tenant_id"])
@@ -95,7 +95,7 @@ async def delete_template(
async def apply_template( async def apply_template(
body: PermissionTemplateApply, body: PermissionTemplateApply,
db: AsyncSession = Depends(get_db), 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.""" """Apply a permission template to an entity, creating entity_permissions."""
tenant_id = uuid.UUID(current_user["tenant_id"]) tenant_id = uuid.UUID(current_user["tenant_id"])
+4 -4
View File
@@ -19,7 +19,7 @@ router = APIRouter(prefix="/api/v1/policies", tags=["policies"])
async def list_policies( async def list_policies(
entity_type: str, entity_type: str,
db: AsyncSession = Depends(get_db), 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.""" """List all ABAC policies for a given entity type."""
tenant_id = uuid.UUID(current_user["tenant_id"]) tenant_id = uuid.UUID(current_user["tenant_id"])
@@ -34,7 +34,7 @@ async def list_policies(
async def create_policy( async def create_policy(
body: PolicyCreate, body: PolicyCreate,
db: AsyncSession = Depends(get_db), 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.""" """Create a new ABAC policy."""
tenant_id = uuid.UUID(current_user["tenant_id"]) tenant_id = uuid.UUID(current_user["tenant_id"])
@@ -59,7 +59,7 @@ async def update_policy(
policy_id: str, policy_id: str,
body: PolicyUpdate, body: PolicyUpdate,
db: AsyncSession = Depends(get_db), 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.""" """Update an existing ABAC policy."""
tenant_id = uuid.UUID(current_user["tenant_id"]) tenant_id = uuid.UUID(current_user["tenant_id"])
@@ -85,7 +85,7 @@ async def update_policy(
async def delete_policy( async def delete_policy(
policy_id: str, policy_id: str,
db: AsyncSession = Depends(get_db), 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.""" """Delete an ABAC policy."""
tenant_id = uuid.UUID(current_user["tenant_id"]) tenant_id = uuid.UUID(current_user["tenant_id"])
+21
View File
@@ -39,6 +39,27 @@ When checking access to an entity, the system resolves in this order (highest wi
| `delete` | 4 | Admin + transfer ownership | | `delete` | 4 | Admin + transfer ownership |
| `owner` | 5 | Full control (automatic for owner) | | `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 ## 2. Data Model
+68
View File
@@ -129,3 +129,71 @@ class TestGetFileMetadataAsync:
meta = get_file_metadata('nonexistent/sync.txt') meta = get_file_metadata('nonexistent/sync.txt')
assert meta == {'size': None, 'modified': None, 'exists': False} 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'],
)