feat(security): add permission system
CI / benchmark-publish (pull_request) Has been skipped
CI / lint (pull_request) Successful in 15s
CI / build (pull_request) Successful in 17s
CI / quality (pull_request) Successful in 19s
CI / security (pull_request) Successful in 32s
CI / typecheck (pull_request) Successful in 36s
CI / integration_tests (pull_request) Successful in 2m40s
CI / unit_tests (pull_request) Successful in 11m42s
CI / docker (pull_request) Successful in 40s
CI / benchmark-regression (pull_request) Successful in 21m35s
CI / coverage (pull_request) Successful in 45m10s
CI / lint (push) Successful in 12s
CI / quality (push) Successful in 18s
CI / build (push) Successful in 23s
CI / security (push) Successful in 32s
CI / typecheck (push) Successful in 57s
CI / benchmark-regression (push) Has been skipped
CI / integration_tests (push) Successful in 2m43s
CI / unit_tests (push) Successful in 12m13s
CI / benchmark-publish (push) Successful in 12m19s
CI / docker (push) Successful in 1m2s
CI / coverage (push) Successful in 1h26m6s

Implemented namespace/project/plan/skill permission model with role
bindings (owner/admin/editor/viewer) and default deny policy. Added
enforcement hooks at CLI/service boundaries that are server-only; local
mode returns permissive defaults. Includes role enums, permission check
service, role matrix documentation.

Includes Behave BDD scenarios, Robot integration tests, ASV benchmarks,
and reference documentation.

ISSUES CLOSED: #344
This commit was merged in pull request #448.
This commit is contained in:
Luis Mendes
2026-02-25 23:46:34 +00:00
parent a2be3e67b0
commit 64af753aab
14 changed files with 2172 additions and 0 deletions
+15
View File
@@ -41,6 +41,21 @@
reconnect, disconnect), tool discovery, input-validated invocation, and bulk registration into
`ToolRegistry` with `source="mcp"` and `checkpointable=False`. Includes Behave unit tests,
Robot integration test, ASV benchmarks, and `docs/reference/mcp_adapter.md` .
### fix(permissions): address code-review findings for permission system (#448)
- Added input validation to `check_permission()` and `get_role_bindings()`:
empty or whitespace-only `principal` and `scope_id` arguments now raise
`ValueError` after stripping, preventing silent lookup misses.
- Aligned module docstring and reference docs to clarify that the
`enforce_permission` decorator is available but not yet wired into CLI
or service call sites — integration is deferred to a future pass.
- Added `permissions.md` to the docs nav in `gen_ref_pages.py`.
- Added Behave BDD scenarios covering empty, whitespace-only, and
leading/trailing-whitespace inputs for both `check_permission()` and
`get_role_bindings()`.
### feat(actor): extend hierarchical actor YAML schema and loader
- Extended actor YAML schema with hierarchical graph support: per-node LSP bindings (`lsp_binding`), tool-source references (`tool_sources`), and subgraph `actor_ref`.
- Added graph reachability validation — all nodes must be reachable from `entry_node` via edges or conditional routing targets.
- Improved loader error reporting with YAML line/column positions and Pydantic field-path hints.
+115
View File
@@ -0,0 +1,115 @@
"""Benchmarks for the permission check enforcement baseline.
Measures the overhead of permission checks in both local mode
(fast-path) and server mode with varying numbers of role bindings.
"""
from __future__ import annotations
import os
from cleveragents.application.services.permission_service import (
PermissionService,
)
from cleveragents.domain.models.core.permission import (
ROLE_PERMISSIONS,
PermissionAction,
PermissionPolicy,
PermissionRole,
PermissionScope,
RoleBinding,
)
_SERVER_MODE_ENV = "CLEVERAGENTS_SERVER_MODE"
def _make_bindings(n: int) -> list[RoleBinding]:
"""Create *n* role bindings for distinct principals."""
return [
RoleBinding(
principal=f"user-{i}",
role=PermissionRole.EDITOR,
scope=PermissionScope.PROJECT,
scope_id="proj-bench",
)
for i in range(n)
]
class TimePermissionCheckLocal:
"""Benchmark permission checks in local mode (fast-path)."""
def setup(self) -> None:
os.environ.pop(_SERVER_MODE_ENV, None)
self.svc = PermissionService()
def time_local_check(self) -> None:
"""Single local-mode permission check."""
self.svc.check_permission(
principal="alice",
action=PermissionAction.WRITE,
scope=PermissionScope.PROJECT,
scope_id="proj-1",
)
def time_local_check_all_actions(self) -> None:
"""Check every action in local mode."""
for action in PermissionAction:
self.svc.check_permission(
principal="alice",
action=action,
scope=PermissionScope.PROJECT,
scope_id="proj-1",
)
class TimePermissionCheckServer:
"""Benchmark permission checks in server mode."""
params: list[int] = [0, 10, 100]
param_names: list[str] = ["num_bindings"]
def setup(self, num_bindings: int) -> None:
os.environ[_SERVER_MODE_ENV] = "true"
bindings = _make_bindings(num_bindings)
self.svc = PermissionService(bindings=bindings)
def teardown(self, num_bindings: int) -> None:
os.environ.pop(_SERVER_MODE_ENV, None)
def time_server_check_denied(self, num_bindings: int) -> None:
"""Permission check that results in denial (no matching binding)."""
self.svc.check_permission(
principal="nonexistent",
action=PermissionAction.WRITE,
scope=PermissionScope.PROJECT,
scope_id="proj-bench",
)
def time_server_check_allowed(self, num_bindings: int) -> None:
"""Permission check that matches the last binding.
When *num_bindings* is 0 no binding can match, so a denied
check is performed instead to avoid an empty benchmark body.
"""
principal = f"user-{num_bindings - 1}" if num_bindings > 0 else "nonexistent"
self.svc.check_permission(
principal=principal,
action=PermissionAction.READ,
scope=PermissionScope.PROJECT,
scope_id="proj-bench",
)
class TimeRolePermissionsLookup:
"""Benchmark ROLE_PERMISSIONS constant lookups."""
def time_lookup_all_roles(self) -> None:
"""Look up allowed actions for every role."""
for role in PermissionRole:
_ = ROLE_PERMISSIONS[role]
def time_check_action_in_role(self) -> None:
"""Check membership of an action in each role's permission set."""
for role in PermissionRole:
_ = PermissionAction.READ in ROLE_PERMISSIONS[role]
+1
View File
@@ -142,6 +142,7 @@ nav[("diff_review",)] = "diff_review.md"
nav[("error_handling",)] = "error_handling.md"
nav[("error_recovery",)] = "error_recovery.md"
nav[("template_security",)] = "template_security.md"
nav[("permissions",)] = "permissions.md"
nav[("validation_pipeline",)] = "validation_pipeline.md"
nav[("agent_skills",)] = "agent_skills.md"
nav[("di",)] = "di.md"
+185
View File
@@ -0,0 +1,185 @@
# Permission System
## Overview
CleverAgents implements a namespace/project/plan/skill permission model
with role bindings and a default-deny policy. An ``enforce_permission``
decorator is available for gating function calls; however, the decorator
is **not yet wired** into CLI or service call sites — that integration
is deferred to a future pass.
**Server-only enforcement**: in local mode the permission system returns
permissive defaults (all actions allowed). Server-side role evaluation
uses the role bindings registered on the ``PermissionService``.
## Roles
| Role | Description |
| -------- | ---------------------------------------- |
| `owner` | Full control — all actions on all scopes |
| `admin` | Administrative access — all actions |
| `editor` | Read, write, and execute |
| `viewer` | Read-only access |
OWNER and ADMIN currently grant the same set of actions. The
distinction is intentional — OWNER designates the resource creator or
namespace owner, while ADMIN is a delegated administrative role.
Future server-mode enforcement may differentiate them (e.g. ownership
transfer, namespace deletion) without requiring a schema change.
## Role → Action Matrix
| Action | Owner | Admin | Editor | Viewer |
| --------- | :---: | :---: | :----: | :----: |
| `read` | yes | yes | yes | yes |
| `write` | yes | yes | yes | — |
| `execute` | yes | yes | yes | — |
| `delete` | yes | yes | — | — |
| `admin` | yes | yes | — | — |
## Scopes
Permissions can be scoped to:
- **Namespace** — top-level organisational boundary
- **Project** — a project within a namespace
- **Plan** — an execution plan within a project
- **Skill** — a skill registered in the system
### Scope Hierarchy (Known Limitation)
Scopes are currently evaluated with **exact match** semantics.
A binding at `namespace` scope does not automatically cascade to
`project`, `plan`, or `skill` scopes within that namespace.
Hierarchical scope resolution (namespace > project > plan > skill)
is deferred to a future server-mode release.
## Role Bindings
A `RoleBinding` associates a *principal* (user or service identifier)
with a `PermissionRole` at a specific `PermissionScope` and scope ID.
Duplicate bindings (same principal, role, scope, and scope_id) are
rejected by `add_binding()`.
```python
from cleveragents.domain.models.core.permission import (
PermissionRole,
PermissionScope,
RoleBinding,
)
binding = RoleBinding(
principal="alice",
role=PermissionRole.EDITOR,
scope=PermissionScope.PROJECT,
scope_id="my-project",
)
```
## Permission Policy
The `PermissionPolicy` model controls the default-deny behaviour and
optional allow-overrides:
```python
from cleveragents.domain.models.core.permission import (
PermissionPolicy,
RoleBinding,
)
policy = PermissionPolicy(
default_deny=True,
allow_overrides=[binding],
)
```
## Permission Service
`PermissionService` is the entry point for checking permissions:
```python
from cleveragents.application.services.permission_service import (
PermissionService,
)
from cleveragents.domain.models.core.permission import (
PermissionAction,
PermissionScope,
)
svc = PermissionService()
check = svc.check_permission(
principal="alice",
action=PermissionAction.WRITE,
scope=PermissionScope.PROJECT,
scope_id="my-project",
)
print(check.result) # True in local mode
```
### Module-Level Default Service
A module-level default ``PermissionService`` is available so that
the ``enforce_permission`` decorator (and other call sites) share a
single service instance with pre-registered bindings:
```python
from cleveragents.application.services.permission_service import (
get_default_permission_service,
set_default_permission_service,
PermissionService,
)
# During application startup:
svc = PermissionService(bindings=[...])
set_default_permission_service(svc)
# Later — the decorator (and any code that calls
# get_default_permission_service()) will use this instance.
```
## Enforcement Decorator
The `enforce_permission` decorator is available for gating function
calls. It is not yet applied to any CLI or service entry point — call
sites will be wired in a future integration pass:
```python
from cleveragents.application.services.permission_service import (
enforce_permission,
)
from cleveragents.domain.models.core.permission import (
PermissionAction,
PermissionScope,
)
@enforce_permission(
PermissionAction.WRITE,
PermissionScope.PROJECT,
"my-project",
)
def create_resource():
...
```
In local mode the decorator is a no-op. In server mode it raises
`PermissionError` when the check fails. When no explicit `service`
is passed, the decorator uses the module-level default service (see
above).
## Local Mode Behaviour
When `CLEVERAGENTS_SERVER_MODE` is unset (or not `true`), the system
operates in **local mode**:
- `PermissionService.is_local_mode()` returns `True`
- All permission checks return `result=True`
- `get_role_bindings()` returns a synthetic OWNER binding
- The `DEFAULT_LOCAL_ROLE_MAPPING` maps `"*"``PermissionRole.OWNER`
## Audit Logging
Permission denials in server mode are logged as `permission_denied`
events via structlog at the WARNING level. Binding additions and
removals are logged as `binding_added` / `binding_removed` at INFO
level.
+1
View File
@@ -94,6 +94,7 @@ def before_scenario(context, scenario):
for env_var in [
"CLEVERAGENTS_MOCK_SHOULD_FAIL",
"CLEVERAGENTS_MOCK_INVALID_CODE",
"CLEVERAGENTS_SERVER_MODE",
*LANGSMITH_ENV_VARS,
]:
if env_var in os.environ:
+285
View File
@@ -0,0 +1,285 @@
Feature: Permission system — role-based access control
As a CleverAgents operator
I want a namespace/project/plan/skill permission model
So that access to resources is controlled by role bindings with default deny
# ── Role enum values ─────────────────────────────────────────────
Scenario: PermissionRole enum contains the expected members
Given the PermissionRole enum
Then the enum should contain "owner"
And the enum should contain "admin"
And the enum should contain "editor"
And the enum should contain "viewer"
Scenario: PermissionScope enum contains the expected members
Given the PermissionScope enum
Then the enum should contain "namespace"
And the enum should contain "project"
And the enum should contain "plan"
And the enum should contain "skill"
Scenario: PermissionAction enum contains the expected members
Given the PermissionAction enum
Then the enum should contain "read"
And the enum should contain "write"
And the enum should contain "execute"
And the enum should contain "delete"
And the enum should contain "admin"
# ── ROLE_PERMISSIONS constant ────────────────────────────────────
Scenario: OWNER role has all actions
Given the ROLE_PERMISSIONS constant
Then the "owner" role should have 5 allowed actions
Scenario: ADMIN role has all actions
Given the ROLE_PERMISSIONS constant
Then the "admin" role should have 5 allowed actions
Scenario: EDITOR role has read, write, and execute
Given the ROLE_PERMISSIONS constant
Then the "editor" role should have 3 allowed actions
Scenario: VIEWER role has only read
Given the ROLE_PERMISSIONS constant
Then the "viewer" role should have 1 allowed actions
# ── RoleBinding model ────────────────────────────────────────────
Scenario: Create a valid RoleBinding
Given a role binding for principal "alice" with role "editor" on "project" scope "proj-1"
Then the binding principal should be "alice"
And the binding role should be "editor"
And the binding scope should be "project"
And the binding scope_id should be "proj-1"
# ── PermissionPolicy model ──────────────────────────────────────
Scenario: Default policy has default_deny enabled
Given a default PermissionPolicy
Then default_deny should be true
And allow_overrides should be empty
Scenario: Policy with allow overrides
Given a PermissionPolicy with an override for "bob" as "admin" on "namespace" scope "ns-1"
Then the policy should have 1 allow override
# ── Local mode permission check ─────────────────────────────────
Scenario: Local mode always permits actions
Given a PermissionService in local mode
When I check permission for "alice" to "write" on "project" scope "proj-1"
Then the permission check result should be true
And the reason should contain "local mode"
Scenario: Local mode permits admin action
Given a PermissionService in local mode
When I check permission for "alice" to "admin" on "namespace" scope "ns-1"
Then the permission check result should be true
Scenario: Local mode permits delete action
Given a PermissionService in local mode
When I check permission for "bob" to "delete" on "plan" scope "plan-42"
Then the permission check result should be true
# ── Server mode — default deny ──────────────────────────────────
Scenario: Server mode denies action when no binding exists
Given a PermissionService in server mode with no bindings
When I check permission for "alice" to "write" on "project" scope "proj-1"
Then the permission check result should be false
And the reason should contain "denied by default-deny"
Scenario: Server mode denies action for missing role
Given a PermissionService in server mode with no bindings
When I check permission for "unknown-user" to "read" on "skill" scope "sk-99"
Then the permission check result should be false
# ── Server mode — role binding grants ───────────────────────────
Scenario: Server mode allows action when binding grants the role
Given a PermissionService in server mode
And a role binding for principal "alice" with role "editor" on "project" scope "proj-1"
When I check permission for "alice" to "read" on "project" scope "proj-1"
Then the permission check result should be true
And the reason should contain "allowed by role"
Scenario: Server mode denies action not in role permissions
Given a PermissionService in server mode
And a role binding for principal "alice" with role "viewer" on "project" scope "proj-1"
When I check permission for "alice" to "write" on "project" scope "proj-1"
Then the permission check result should be false
# ── Server mode — allow overrides ───────────────────────────────
Scenario: Allow override grants access despite no regular binding
Given a PermissionService in server mode with no bindings
And a policy override for "bob" as "admin" on "namespace" scope "ns-1"
When I check permission for "bob" to "admin" on "namespace" scope "ns-1"
Then the permission check result should be true
And the reason should contain "override"
# ── Default local role mapping ──────────────────────────────────
Scenario: DEFAULT_LOCAL_ROLE_MAPPING grants OWNER to wildcard
Given the DEFAULT_LOCAL_ROLE_MAPPING constant
Then the wildcard entry should map to "owner"
# ── PermissionCheck model ───────────────────────────────────────
Scenario: PermissionCheck captures all fields
Given a PermissionCheck for "carol" action "execute" scope "plan" scope_id "plan-7" result true reason "ok"
Then the check principal should be "carol"
And the check action should be "execute"
And the check scope should be "plan"
And the check scope_id should be "plan-7"
And the check result should be true
And the check reason should be "ok"
# ── enforce_permission decorator ────────────────────────────────
Scenario: enforce_permission decorator allows in local mode
Given a function decorated with enforce_permission for "write" on "project" scope "proj-1"
When I call the decorated function in local mode
Then it should succeed without raising
Scenario: enforce_permission decorator raises in server mode when denied
Given a function decorated with enforce_permission for "delete" on "namespace" scope "ns-1"
When I call the decorated function in server mode with no bindings
Then it should raise a PermissionError
# ── Binding management ──────────────────────────────────────────
Scenario: Add and remove bindings on the service
Given a PermissionService in server mode with no bindings
And I add a binding for "dave" as "editor" on "skill" scope "sk-1"
When I check permission for "dave" to "write" on "skill" scope "sk-1"
Then the permission check result should be true
When I remove the binding for "dave" as "editor" on "skill" scope "sk-1"
And I check permission for "dave" to "write" on "skill" scope "sk-1"
Then the permission check result should be false
Scenario: Remove non-existent binding returns false
Given a PermissionService in server mode with no bindings
When I try to remove a binding for "ghost" as "viewer" on "plan" scope "p-0"
Then the remove result should be false
# ── get_role_bindings ───────────────────────────────────────────
Scenario: get_role_bindings in local mode returns synthetic OWNER binding
Given a PermissionService in local mode
When I get role bindings for "anyone" on "project" scope "proj-1"
Then I should receive 1 binding
And the first binding role should be "owner"
Scenario: get_role_bindings in server mode returns matching bindings
Given a PermissionService in server mode with no bindings
And I add a binding for "eve" as "viewer" on "project" scope "proj-2"
And I add a binding for "eve" as "editor" on "project" scope "proj-2"
When I get role bindings for "eve" on "project" scope "proj-2"
Then I should receive 2 bindings
# ── Policy default_deny=False ───────────────────────────────────
Scenario: Server mode with default_deny disabled permits unbound action
Given a PermissionService in server mode with default_deny disabled
When I check permission for "frank" to "read" on "namespace" scope "ns-9"
Then the permission check result should be true
And the reason should contain "default-deny disabled"
# ── Scope mismatch ──────────────────────────────────────────────
Scenario: Binding on project scope does not grant access at namespace scope
Given a PermissionService in server mode
And a role binding for principal "alice" with role "editor" on "project" scope "proj-1"
When I check permission for "alice" to "read" on "namespace" scope "proj-1"
Then the permission check result should be false
Scenario: Binding on skill scope does not grant access at plan scope
Given a PermissionService in server mode
And a role binding for principal "bob" with role "admin" on "skill" scope "sk-1"
When I check permission for "bob" to "admin" on "plan" scope "sk-1"
Then the permission check result should be false
# ── Duplicate binding prevention ────────────────────────────────
Scenario: Adding a duplicate binding is rejected
Given a PermissionService in server mode with no bindings
And I add a binding for "dave" as "editor" on "project" scope "proj-1"
When I try to add a duplicate binding for "dave" as "editor" on "project" scope "proj-1"
Then the add result should be false
# ── RoleBinding validation ──────────────────────────────────────
Scenario: RoleBinding rejects empty principal
When I try to create a role binding with empty principal
Then a permission validation error should be raised
Scenario: RoleBinding rejects empty scope_id
When I try to create a role binding with empty scope_id
Then a permission validation error should be raised
# ── PermissionCheck with result=False ───────────────────────────
Scenario: PermissionCheck accepts result false
Given a PermissionCheck for "denied-user" action "write" scope "project" scope_id "proj-1" result false reason "denied"
Then the check result should be false
And the check reason should be "denied"
# ── enforce_permission with pre-loaded service ──────────────────
Scenario: enforce_permission decorator uses pre-loaded service with bindings
Given a pre-loaded permission service with "alice" as "editor" on "project" scope "proj-1" for action "write"
When I call the pre-loaded decorated function in server mode
Then it should succeed without raising
Scenario: enforce_permission decorator uses module-level default service
Given a module-level default service with "bob" as "admin" on "namespace" scope "ns-1"
And a function decorated with enforce_permission for "admin" on "namespace" scope "ns-1"
When I call the decorated function in server mode via module default
Then it should succeed without raising
# ── Input validation — check_permission ─────────────────────────
Scenario: check_permission rejects empty principal
Given a PermissionService in local mode
When I check permission with empty principal
Then a permission ValueError should mention "principal"
Scenario: check_permission rejects whitespace-only principal
Given a PermissionService in local mode
When I check permission with whitespace-only principal
Then a permission ValueError should mention "principal"
Scenario: check_permission rejects empty scope_id
Given a PermissionService in local mode
When I check permission with empty scope_id
Then a permission ValueError should mention "scope_id"
Scenario: check_permission rejects whitespace-only scope_id
Given a PermissionService in local mode
When I check permission with whitespace-only scope_id
Then a permission ValueError should mention "scope_id"
Scenario: check_permission strips leading/trailing whitespace from principal
Given a PermissionService in local mode
When I check permission for " alice " to "read" on "project" scope "proj-1"
Then the permission check result should be true
Scenario: check_permission strips leading/trailing whitespace from scope_id
Given a PermissionService in local mode
When I check permission for "alice" to "read" on "project" scope " proj-1 "
Then the permission check result should be true
# ── Input validation — get_role_bindings ────────────────────────
Scenario: get_role_bindings rejects empty principal
Given a PermissionService in local mode
When I get role bindings with empty principal
Then a permission ValueError should mention "principal"
Scenario: get_role_bindings rejects whitespace-only scope_id
Given a PermissionService in local mode
When I get role bindings with whitespace-only scope_id
Then a permission ValueError should mention "scope_id"
+755
View File
@@ -0,0 +1,755 @@
"""Step definitions for the permission system feature.
Tests the permission domain model, PermissionService in local and
server modes, the enforce_permission decorator, and role binding
management.
"""
from __future__ import annotations
import os
from behave import given, then, when
from behave.runner import Context
from pydantic import ValidationError
from cleveragents.application.services.permission_service import (
PermissionService,
enforce_permission,
set_default_permission_service,
)
from cleveragents.domain.models.core.permission import (
DEFAULT_LOCAL_ROLE_MAPPING,
ROLE_PERMISSIONS,
PermissionAction,
PermissionCheck,
PermissionPolicy,
PermissionRole,
PermissionScope,
RoleBinding,
)
# ── Helpers ───────────────────────────────────────────────────────
_SERVER_MODE_ENV = "CLEVERAGENTS_SERVER_MODE"
def _set_local_mode() -> None:
os.environ.pop(_SERVER_MODE_ENV, None)
def _set_server_mode() -> None:
os.environ[_SERVER_MODE_ENV] = "true"
# ── Enum steps ────────────────────────────────────────────────────
@given("the PermissionRole enum")
def step_given_permission_role_enum(context: Context) -> None:
context.enum_values = [e.value for e in PermissionRole]
@given("the PermissionScope enum")
def step_given_permission_scope_enum(context: Context) -> None:
context.enum_values = [e.value for e in PermissionScope]
@given("the PermissionAction enum")
def step_given_permission_action_enum(context: Context) -> None:
context.enum_values = [e.value for e in PermissionAction]
@then('the enum should contain "{value}"')
def step_enum_should_contain(context: Context, value: str) -> None:
assert value in context.enum_values, f"Expected '{value}' in {context.enum_values}"
# ── ROLE_PERMISSIONS ──────────────────────────────────────────────
@given("the ROLE_PERMISSIONS constant")
def step_given_role_permissions(context: Context) -> None:
context.role_permissions = ROLE_PERMISSIONS
@then('the "{role}" role should have {count:d} allowed actions')
def step_role_has_n_actions(context: Context, role: str, count: int) -> None:
perm_role = PermissionRole(role)
actions = context.role_permissions[perm_role]
assert len(actions) == count, (
f"Expected {count} actions for {role}, got {len(actions)}"
)
# ── RoleBinding model ────────────────────────────────────────────
@given(
'a role binding for principal "{principal}" with role "{role}" '
'on "{scope}" scope "{scope_id}"'
)
def step_given_role_binding(
context: Context, principal: str, role: str, scope: str, scope_id: str
) -> None:
binding = RoleBinding(
principal=principal,
role=PermissionRole(role),
scope=PermissionScope(scope),
scope_id=scope_id,
)
context.binding = binding
# Also register on service if present
svc: PermissionService | None = getattr(context, "permission_service", None)
if svc is not None:
svc.add_binding(binding)
@then('the binding principal should be "{expected}"')
def step_binding_principal(context: Context, expected: str) -> None:
assert context.binding.principal == expected
@then('the binding role should be "{expected}"')
def step_binding_role(context: Context, expected: str) -> None:
assert context.binding.role == expected
@then('the binding scope should be "{expected}"')
def step_binding_scope(context: Context, expected: str) -> None:
assert context.binding.scope == expected
@then('the binding scope_id should be "{expected}"')
def step_binding_scope_id(context: Context, expected: str) -> None:
assert context.binding.scope_id == expected
# ── PermissionPolicy ─────────────────────────────────────────────
@given("a default PermissionPolicy")
def step_given_default_policy(context: Context) -> None:
context.policy = PermissionPolicy()
@given(
'a PermissionPolicy with an override for "{principal}" as "{role}" '
'on "{scope}" scope "{scope_id}"'
)
def step_given_policy_with_override(
context: Context, principal: str, role: str, scope: str, scope_id: str
) -> None:
override = RoleBinding(
principal=principal,
role=PermissionRole(role),
scope=PermissionScope(scope),
scope_id=scope_id,
)
context.policy = PermissionPolicy(allow_overrides=[override])
@then("default_deny should be true")
def step_default_deny_true(context: Context) -> None:
assert context.policy.default_deny is True
@then("allow_overrides should be empty")
def step_overrides_empty(context: Context) -> None:
assert len(context.policy.allow_overrides) == 0
@then("the policy should have {count:d} allow override")
def step_policy_override_count(context: Context, count: int) -> None:
assert len(context.policy.allow_overrides) == count
# ── PermissionService — local mode ────────────────────────────────
@given("a PermissionService in local mode")
def step_given_service_local(context: Context) -> None:
_set_local_mode()
context.permission_service = PermissionService()
@given("a PermissionService in server mode with no bindings")
def step_given_service_server_no_bindings(context: Context) -> None:
_set_server_mode()
context.permission_service = PermissionService()
@given("a PermissionService in server mode")
def step_given_service_server(context: Context) -> None:
_set_server_mode()
context.permission_service = PermissionService()
@given("a PermissionService in server mode with default_deny disabled")
def step_given_service_server_no_deny(context: Context) -> None:
_set_server_mode()
policy = PermissionPolicy(default_deny=False)
context.permission_service = PermissionService(policy=policy)
@given(
'a policy override for "{principal}" as "{role}" on "{scope}" scope "{scope_id}"'
)
def step_given_policy_override_on_service(
context: Context, principal: str, role: str, scope: str, scope_id: str
) -> None:
override = RoleBinding(
principal=principal,
role=PermissionRole(role),
scope=PermissionScope(scope),
scope_id=scope_id,
)
policy = PermissionPolicy(allow_overrides=[override])
context.permission_service = PermissionService(policy=policy)
# ── Permission check ─────────────────────────────────────────────
@when(
'I check permission for "{principal}" to "{action}" on "{scope}" scope "{scope_id}"'
)
def step_check_permission(
context: Context, principal: str, action: str, scope: str, scope_id: str
) -> None:
svc: PermissionService = context.permission_service
context.permission_result = svc.check_permission(
principal=principal,
action=PermissionAction(action),
scope=PermissionScope(scope),
scope_id=scope_id,
)
@then("the permission check result should be true")
def step_check_result_true(context: Context) -> None:
assert context.permission_result.result is True, (
f"Expected True, got False: {context.permission_result.reason}"
)
@then("the permission check result should be false")
def step_check_result_false(context: Context) -> None:
assert context.permission_result.result is False, (
f"Expected False, got True: {context.permission_result.reason}"
)
@then('the reason should contain "{fragment}"')
def step_reason_contains(context: Context, fragment: str) -> None:
assert fragment in context.permission_result.reason, (
f"Expected '{fragment}' in '{context.permission_result.reason}'"
)
# ── DEFAULT_LOCAL_ROLE_MAPPING ────────────────────────────────────
@given("the DEFAULT_LOCAL_ROLE_MAPPING constant")
def step_given_default_local_mapping(context: Context) -> None:
context.local_mapping = DEFAULT_LOCAL_ROLE_MAPPING
@then('the wildcard entry should map to "{expected}"')
def step_wildcard_maps_to(context: Context, expected: str) -> None:
assert context.local_mapping["*"] == expected
# ── PermissionCheck model ─────────────────────────────────────────
@given(
'a PermissionCheck for "{principal}" action "{action}" scope "{scope}" '
'scope_id "{scope_id}" result true reason "{reason}"'
)
def step_given_permission_check_model(
context: Context,
principal: str,
action: str,
scope: str,
scope_id: str,
reason: str,
) -> None:
context.check_model = PermissionCheck(
principal=principal,
action=PermissionAction(action),
scope=PermissionScope(scope),
scope_id=scope_id,
result=True,
reason=reason,
)
@then('the check principal should be "{expected}"')
def step_check_principal(context: Context, expected: str) -> None:
assert context.check_model.principal == expected
@then('the check action should be "{expected}"')
def step_check_action(context: Context, expected: str) -> None:
assert context.check_model.action == expected
@then('the check scope should be "{expected}"')
def step_check_scope(context: Context, expected: str) -> None:
assert context.check_model.scope == expected
@then('the check scope_id should be "{expected}"')
def step_check_scope_id(context: Context, expected: str) -> None:
assert context.check_model.scope_id == expected
@then("the check result should be true")
def step_check_model_result_true(context: Context) -> None:
assert context.check_model.result is True
@then('the check reason should be "{expected}"')
def step_check_reason(context: Context, expected: str) -> None:
assert context.check_model.reason == expected
# ── enforce_permission decorator ──────────────────────────────────
@given(
"a function decorated with enforce_permission for "
'"{action}" on "{scope}" scope "{scope_id}"'
)
def step_given_decorated_fn(
context: Context, action: str, scope: str, scope_id: str
) -> None:
context.enforce_action = action
context.enforce_scope = scope
context.enforce_scope_id = scope_id
@when("I call the decorated function in local mode")
def step_call_decorated_local(context: Context) -> None:
_set_local_mode()
@enforce_permission(
PermissionAction(context.enforce_action),
PermissionScope(context.enforce_scope),
context.enforce_scope_id,
)
def _guarded() -> str:
return "ok"
try:
context.decorated_result = _guarded()
context.decorated_error = None
except PermissionError as exc:
context.decorated_result = None
context.decorated_error = exc
@when("I call the decorated function in server mode with no bindings")
def step_call_decorated_server(context: Context) -> None:
_set_server_mode()
@enforce_permission(
PermissionAction(context.enforce_action),
PermissionScope(context.enforce_scope),
context.enforce_scope_id,
)
def _guarded() -> str:
return "ok"
try:
context.decorated_result = _guarded()
context.decorated_error = None
except PermissionError as exc:
context.decorated_result = None
context.decorated_error = exc
@then("it should succeed without raising")
def step_no_error(context: Context) -> None:
assert context.decorated_error is None
assert context.decorated_result == "ok"
@then("it should raise a PermissionError")
def step_permission_error_raised(context: Context) -> None:
assert context.decorated_error is not None
assert isinstance(context.decorated_error, PermissionError)
# ── Binding management ────────────────────────────────────────────
@given('I add a binding for "{principal}" as "{role}" on "{scope}" scope "{scope_id}"')
def step_add_binding(
context: Context, principal: str, role: str, scope: str, scope_id: str
) -> None:
binding = RoleBinding(
principal=principal,
role=PermissionRole(role),
scope=PermissionScope(scope),
scope_id=scope_id,
)
context.permission_service.add_binding(binding)
@when(
'I remove the binding for "{principal}" as "{role}" on "{scope}" scope "{scope_id}"'
)
def step_remove_binding(
context: Context, principal: str, role: str, scope: str, scope_id: str
) -> None:
binding = RoleBinding(
principal=principal,
role=PermissionRole(role),
scope=PermissionScope(scope),
scope_id=scope_id,
)
context.remove_result = context.permission_service.remove_binding(binding)
@when(
'I try to remove a binding for "{principal}" as "{role}" '
'on "{scope}" scope "{scope_id}"'
)
def step_try_remove_binding(
context: Context, principal: str, role: str, scope: str, scope_id: str
) -> None:
binding = RoleBinding(
principal=principal,
role=PermissionRole(role),
scope=PermissionScope(scope),
scope_id=scope_id,
)
context.remove_result = context.permission_service.remove_binding(binding)
@then("the remove result should be false")
def step_remove_result_false(context: Context) -> None:
assert context.remove_result is False
# ── get_role_bindings ─────────────────────────────────────────────
@when('I get role bindings for "{principal}" on "{scope}" scope "{scope_id}"')
def step_get_role_bindings(
context: Context, principal: str, scope: str, scope_id: str
) -> None:
svc: PermissionService = context.permission_service
context.fetched_bindings = svc.get_role_bindings(
principal=principal,
scope=PermissionScope(scope),
scope_id=scope_id,
)
@then("I should receive {count:d} binding")
def step_binding_count_singular(context: Context, count: int) -> None:
assert len(context.fetched_bindings) == count, (
f"Expected {count} bindings, got {len(context.fetched_bindings)}"
)
@then("I should receive {count:d} bindings")
def step_binding_count_plural(context: Context, count: int) -> None:
assert len(context.fetched_bindings) == count, (
f"Expected {count} bindings, got {len(context.fetched_bindings)}"
)
@then('the first binding role should be "{expected}"')
def step_first_binding_role(context: Context, expected: str) -> None:
assert context.fetched_bindings[0].role == expected
# ── Duplicate binding prevention ──────────────────────────────────
@when(
'I try to add a duplicate binding for "{principal}" as "{role}" '
'on "{scope}" scope "{scope_id}"'
)
def step_try_add_duplicate_binding(
context: Context, principal: str, role: str, scope: str, scope_id: str
) -> None:
binding = RoleBinding(
principal=principal,
role=PermissionRole(role),
scope=PermissionScope(scope),
scope_id=scope_id,
)
context.add_result = context.permission_service.add_binding(binding)
@then("the add result should be false")
def step_add_result_false(context: Context) -> None:
assert context.add_result is False
# ── RoleBinding validation ────────────────────────────────────────
@when("I try to create a role binding with empty principal")
def step_try_create_binding_empty_principal(context: Context) -> None:
context.permission_error = None
try:
RoleBinding(
principal="",
role=PermissionRole.VIEWER,
scope=PermissionScope.PROJECT,
scope_id="proj-1",
)
except ValidationError as exc:
context.permission_error = exc
@when("I try to create a role binding with empty scope_id")
def step_try_create_binding_empty_scope_id(context: Context) -> None:
context.permission_error = None
try:
RoleBinding(
principal="alice",
role=PermissionRole.VIEWER,
scope=PermissionScope.PROJECT,
scope_id="",
)
except ValidationError as exc:
context.permission_error = exc
@then("a permission validation error should be raised")
def step_permission_validation_error(context: Context) -> None:
assert context.permission_error is not None, "Expected a ValidationError"
assert isinstance(context.permission_error, ValidationError)
# ── PermissionCheck with result=False ─────────────────────────────
@given(
'a PermissionCheck for "{principal}" action "{action}" scope "{scope}" '
'scope_id "{scope_id}" result false reason "{reason}"'
)
def step_given_permission_check_model_false(
context: Context,
principal: str,
action: str,
scope: str,
scope_id: str,
reason: str,
) -> None:
context.check_model = PermissionCheck(
principal=principal,
action=PermissionAction(action),
scope=PermissionScope(scope),
scope_id=scope_id,
result=False,
reason=reason,
)
@then("the check result should be false")
def step_check_model_result_false(context: Context) -> None:
assert context.check_model.result is False
# ── enforce_permission with pre-loaded service ────────────────────
@given(
'a pre-loaded permission service with "{principal}" as "{role}" '
'on "{scope}" scope "{scope_id}" for action "{action}"'
)
def step_given_decorated_fn_preloaded(
context: Context,
principal: str,
role: str,
scope: str,
scope_id: str,
action: str,
) -> None:
binding = RoleBinding(
principal=principal,
role=PermissionRole(role),
scope=PermissionScope(scope),
scope_id=scope_id,
)
context.preloaded_service = PermissionService(bindings=[binding])
context.preloaded_principal = principal
context.enforce_action = action
context.enforce_scope = scope
context.enforce_scope_id = scope_id
@when("I call the pre-loaded decorated function in server mode")
def step_call_preloaded_decorated_server(context: Context) -> None:
_set_server_mode()
@enforce_permission(
PermissionAction(context.enforce_action),
PermissionScope(context.enforce_scope),
context.enforce_scope_id,
principal=context.preloaded_principal,
service=context.preloaded_service,
)
def _guarded() -> str:
return "ok"
try:
context.decorated_result = _guarded()
context.decorated_error = None
except PermissionError as exc:
context.decorated_result = None
context.decorated_error = exc
# ── enforce_permission with module-level default service ──────────
@given(
'a module-level default service with "{principal}" as "{role}" '
'on "{scope}" scope "{scope_id}"'
)
def step_given_module_default_service(
context: Context,
principal: str,
role: str,
scope: str,
scope_id: str,
) -> None:
binding = RoleBinding(
principal=principal,
role=PermissionRole(role),
scope=PermissionScope(scope),
scope_id=scope_id,
)
svc = PermissionService(bindings=[binding])
set_default_permission_service(svc)
context.default_svc_principal = principal
@when("I call the decorated function in server mode via module default")
def step_call_decorated_server_module_default(context: Context) -> None:
_set_server_mode()
@enforce_permission(
PermissionAction(context.enforce_action),
PermissionScope(context.enforce_scope),
context.enforce_scope_id,
principal=context.default_svc_principal,
)
def _guarded() -> str:
return "ok"
try:
context.decorated_result = _guarded()
context.decorated_error = None
except PermissionError as exc:
context.decorated_result = None
context.decorated_error = exc
finally:
# Clean up module-level default
set_default_permission_service(None)
# ── Input validation — check_permission ───────────────────────────
@when("I check permission with empty principal")
def step_check_permission_empty_principal(context: Context) -> None:
context.value_error = None
try:
context.permission_service.check_permission(
principal="",
action=PermissionAction.READ,
scope=PermissionScope.PROJECT,
scope_id="proj-1",
)
except ValueError as exc:
context.value_error = exc
@when("I check permission with whitespace-only principal")
def step_check_permission_whitespace_principal(context: Context) -> None:
context.value_error = None
try:
context.permission_service.check_permission(
principal=" ",
action=PermissionAction.READ,
scope=PermissionScope.PROJECT,
scope_id="proj-1",
)
except ValueError as exc:
context.value_error = exc
@when("I check permission with empty scope_id")
def step_check_permission_empty_scope_id(context: Context) -> None:
context.value_error = None
try:
context.permission_service.check_permission(
principal="alice",
action=PermissionAction.READ,
scope=PermissionScope.PROJECT,
scope_id="",
)
except ValueError as exc:
context.value_error = exc
@when("I check permission with whitespace-only scope_id")
def step_check_permission_whitespace_scope_id(context: Context) -> None:
context.value_error = None
try:
context.permission_service.check_permission(
principal="alice",
action=PermissionAction.READ,
scope=PermissionScope.PROJECT,
scope_id=" ",
)
except ValueError as exc:
context.value_error = exc
@then('a permission ValueError should mention "{fragment}"')
def step_permission_value_error_with_message(context: Context, fragment: str) -> None:
assert context.value_error is not None, "Expected a ValueError"
assert isinstance(context.value_error, ValueError)
assert fragment in str(context.value_error), (
f"Expected '{fragment}' in '{context.value_error}'"
)
# ── Input validation — get_role_bindings ──────────────────────────
@when("I get role bindings with empty principal")
def step_get_bindings_empty_principal(context: Context) -> None:
context.value_error = None
try:
context.permission_service.get_role_bindings(
principal="",
scope=PermissionScope.PROJECT,
scope_id="proj-1",
)
except ValueError as exc:
context.value_error = exc
@when("I get role bindings with whitespace-only scope_id")
def step_get_bindings_whitespace_scope_id(context: Context) -> None:
context.value_error = None
try:
context.permission_service.get_role_bindings(
principal="alice",
scope=PermissionScope.PROJECT,
scope_id=" ",
)
except ValueError as exc:
context.value_error = exc
+113
View File
@@ -0,0 +1,113 @@
"""Robot Framework helper for permission system integration tests.
Exposes keyword functions that exercise the permission domain model
and PermissionService from Python so Robot tests can validate
behaviour without importing modules directly in Robot syntax.
"""
from __future__ import annotations
import os
import sys
# Ensure the src directory is on the path for imports
_src = os.path.join(os.path.dirname(__file__), "..", "src")
if _src not in sys.path:
sys.path.insert(0, _src)
from cleveragents.application.services.permission_service import ( # noqa: E402
PermissionService,
)
from cleveragents.domain.models.core.permission import ( # noqa: E402
DEFAULT_LOCAL_ROLE_MAPPING,
ROLE_PERMISSIONS,
PermissionAction,
PermissionRole,
PermissionScope,
RoleBinding,
)
_SERVER_MODE_ENV = "CLEVERAGENTS_SERVER_MODE"
def get_permission_roles() -> list[str]:
"""Return all PermissionRole values."""
return [r.value for r in PermissionRole]
def get_permission_scopes() -> list[str]:
"""Return all PermissionScope values."""
return [s.value for s in PermissionScope]
def get_permission_actions() -> list[str]:
"""Return all PermissionAction values."""
return [a.value for a in PermissionAction]
def check_permission_local_mode(
principal: str, action: str, scope: str, scope_id: str
) -> bool:
"""Run a permission check in local mode and return the result."""
os.environ.pop(_SERVER_MODE_ENV, None)
svc = PermissionService()
check = svc.check_permission(
principal=principal,
action=PermissionAction(action),
scope=PermissionScope(scope),
scope_id=scope_id,
)
return check.result
def check_permission_server_no_bindings(
principal: str, action: str, scope: str, scope_id: str
) -> bool:
"""Run a permission check in server mode with no bindings."""
os.environ[_SERVER_MODE_ENV] = "true"
svc = PermissionService()
check = svc.check_permission(
principal=principal,
action=PermissionAction(action),
scope=PermissionScope(scope),
scope_id=scope_id,
)
os.environ.pop(_SERVER_MODE_ENV, None)
return check.result
def check_permission_server_with_binding(
principal: str,
role: str,
action: str,
scope: str,
scope_id: str,
) -> bool:
"""Run a permission check in server mode with a specific binding."""
os.environ[_SERVER_MODE_ENV] = "true"
binding = RoleBinding(
principal=principal,
role=PermissionRole(role),
scope=PermissionScope(scope),
scope_id=scope_id,
)
svc = PermissionService(bindings=[binding])
check = svc.check_permission(
principal=principal,
action=PermissionAction(action),
scope=PermissionScope(scope),
scope_id=scope_id,
)
os.environ.pop(_SERVER_MODE_ENV, None)
return check.result
def get_role_action_count(role: str) -> int:
"""Return the number of allowed actions for a given role."""
perm_role = PermissionRole(role)
return len(ROLE_PERMISSIONS[perm_role])
def get_default_local_role() -> str:
"""Return the default local role for the wildcard principal."""
return str(DEFAULT_LOCAL_ROLE_MAPPING["*"])
+129
View File
@@ -0,0 +1,129 @@
*** Settings ***
Documentation Permission system integration tests — role-based access control
Library OperatingSystem
Library Collections
Library String
Library Process
Library helper_permission_system.py
Resource ${CURDIR}/common.resource
*** Variables ***
${SRC_DIR} ${CURDIR}/../src/cleveragents
*** Test Cases ***
Permission Module Exists
[Documentation] Verify the permission domain model module is present
File Should Exist ${SRC_DIR}/domain/models/core/permission.py
Permission Service Module Exists
[Documentation] Verify the permission service module is present
File Should Exist ${SRC_DIR}/application/services/permission_service.py
Permission Module Exports Expected Enums
[Documentation] Verify permission.py exports public enums
${content}= Get File ${SRC_DIR}/domain/models/core/permission.py
Should Contain ${content} class PermissionRole
Should Contain ${content} class PermissionScope
Should Contain ${content} class PermissionAction
Permission Module Exports Models
[Documentation] Verify permission.py exports data models
${content}= Get File ${SRC_DIR}/domain/models/core/permission.py
Should Contain ${content} class RoleBinding
Should Contain ${content} class PermissionCheck
Should Contain ${content} class PermissionPolicy
Permission Module Exports Constants
[Documentation] Verify permission.py exports constant mappings
${content}= Get File ${SRC_DIR}/domain/models/core/permission.py
Should Contain ${content} ROLE_PERMISSIONS
Should Contain ${content} DEFAULT_LOCAL_ROLE_MAPPING
Service Module Exports Expected Symbols
[Documentation] Verify permission_service.py exports public API
${content}= Get File ${SRC_DIR}/application/services/permission_service.py
Should Contain ${content} class PermissionService
Should Contain ${content} def check_permission
Should Contain ${content} def enforce_permission
Should Contain ${content} def is_local_mode
Should Contain ${content} def get_role_bindings
Core Init Exports Permission Models
[Documentation] Verify core __init__.py re-exports permission types
${content}= Get File ${SRC_DIR}/domain/models/core/__init__.py
Should Contain ${content} PermissionRole
Should Contain ${content} PermissionScope
Should Contain ${content} PermissionAction
Should Contain ${content} RoleBinding
Should Contain ${content} PermissionCheck
Should Contain ${content} PermissionPolicy
Should Contain ${content} ROLE_PERMISSIONS
Should Contain ${content} DEFAULT_LOCAL_ROLE_MAPPING
Services Init Exports Permission Service
[Documentation] Verify services __init__.py re-exports PermissionService
${content}= Get File ${SRC_DIR}/application/services/__init__.py
Should Contain ${content} PermissionService
Should Contain ${content} enforce_permission
Permission Role Enum Values Via Python
[Documentation] Verify PermissionRole values using helper
@{roles}= Get Permission Roles
Should Contain ${roles} owner
Should Contain ${roles} admin
Should Contain ${roles} editor
Should Contain ${roles} viewer
Length Should Be ${roles} 4
Permission Scope Enum Values Via Python
[Documentation] Verify PermissionScope values using helper
@{scopes}= Get Permission Scopes
Should Contain ${scopes} namespace
Should Contain ${scopes} project
Should Contain ${scopes} plan
Should Contain ${scopes} skill
Length Should Be ${scopes} 4
Permission Action Enum Values Via Python
[Documentation] Verify PermissionAction values using helper
@{actions}= Get Permission Actions
Should Contain ${actions} read
Should Contain ${actions} write
Should Contain ${actions} execute
Should Contain ${actions} delete
Should Contain ${actions} admin
Length Should Be ${actions} 5
Local Mode Permission Check Via Python
[Documentation] Verify local mode always permits
${result}= Check Permission Local Mode alice write project proj-1
Should Be True ${result}
Server Mode Denied Without Binding Via Python
[Documentation] Verify server mode denies without bindings
${result}= Check Permission Server No Bindings alice write project proj-1
Should Not Be True ${result}
Server Mode Allows With Binding Via Python
[Documentation] Verify server mode allows when binding exists
${result}= Check Permission Server With Binding alice editor read project proj-1
Should Be True ${result}
Owner Role Has All Actions Via Python
[Documentation] Verify OWNER role maps to all 5 actions
${count}= Get Role Action Count owner
Should Be Equal As Integers ${count} 5
Viewer Role Has Only Read Via Python
[Documentation] Verify VIEWER role maps to 1 action
${count}= Get Role Action Count viewer
Should Be Equal As Integers ${count} 1
Default Local Role Mapping Via Python
[Documentation] Verify wildcard maps to owner
${role}= Get Default Local Role
Should Be Equal ${role} owner
Documentation File Exists
[Documentation] Verify reference documentation was created
File Should Exist ${CURDIR}/../docs/reference/permissions.md
@@ -18,6 +18,12 @@ from cleveragents.application.services.decision_service import (
from cleveragents.application.services.invariant_service import (
InvariantService,
)
from cleveragents.application.services.permission_service import (
PermissionService,
enforce_permission,
get_default_permission_service,
set_default_permission_service,
)
from cleveragents.application.services.plan_execution_context import (
PlanExecutionContext,
RuntimeExecuteActor,
@@ -89,6 +95,7 @@ __all__ = [
"MissingImportRule",
"MissingSymbolRule",
"NormalisedOutputDict",
"PermissionService",
"PersistentSessionService",
"PipelineResultDict",
"PlanExecutionContext",
@@ -111,6 +118,9 @@ __all__ = [
"ValidationRunner",
"ValidationSummary",
"create_default_registry",
"enforce_permission",
"get_default_permission_service",
"map_severity_to_mode",
"resolve_severity",
"set_default_permission_service",
]
@@ -0,0 +1,354 @@
"""Permission enforcement service for CleverAgents.
Provides permission checking and an ``enforce_permission`` decorator
that can be applied at CLI/service boundaries. In local mode every
action is permitted; server-side enforcement evaluates role bindings
against the permission policy. Actual wiring of the decorator into
CLI and service call sites is deferred to a future integration pass.
Based on issue #344 — feat(security): add permission system.
"""
from __future__ import annotations
import os
from collections.abc import Callable
from functools import wraps
from typing import TypeVar
import structlog
from cleveragents.domain.models.core.permission import (
DEFAULT_LOCAL_ROLE_MAPPING,
ROLE_PERMISSIONS,
PermissionAction,
PermissionCheck,
PermissionPolicy,
PermissionRole,
PermissionScope,
RoleBinding,
)
logger = structlog.get_logger(__name__)
F = TypeVar("F", bound=Callable[..., object])
def _validate_identifier(value: str, name: str) -> str:
"""Strip whitespace and reject empty identifiers.
Raises ``ValueError`` when *value* is empty or whitespace-only after
stripping, matching the ``RoleBinding`` Pydantic constraint
(``min_length=1``, ``str_strip_whitespace=True``).
"""
stripped = value.strip()
if not stripped:
msg = f"{name} must be a non-empty string (got {value!r})"
raise ValueError(msg)
return stripped
__all__ = [
"PermissionService",
"enforce_permission",
"get_default_permission_service",
"set_default_permission_service",
]
# ── Environment helpers ───────────────────────────────────────────
_SERVER_MODE_ENV = "CLEVERAGENTS_SERVER_MODE"
def _is_server_mode() -> bool:
"""Return ``True`` when the process is running in server mode."""
return os.environ.get(_SERVER_MODE_ENV, "").lower() in {"1", "true", "yes"}
# ── Binding index key ─────────────────────────────────────────────
_BindingKey = tuple[str, PermissionScope, str]
def _binding_key(binding: RoleBinding) -> _BindingKey:
"""Return the lookup key for a role binding."""
return (binding.principal, binding.scope, binding.scope_id)
# ── Service ───────────────────────────────────────────────────────
class PermissionService:
"""Central permission check and enforcement service.
* **Local mode** (default): all checks return *allowed*.
* **Server mode**: evaluates role bindings against the permission
policy. If no server-side store is available the check raises
``NotImplementedError``.
"""
def __init__(
self,
*,
policy: PermissionPolicy | None = None,
bindings: list[RoleBinding] | None = None,
) -> None:
self._policy = policy or PermissionPolicy()
self._bindings_index: dict[_BindingKey, list[RoleBinding]] = {}
self._logger = logger.bind(service="permission")
for b in bindings or []:
self._bindings_index.setdefault(_binding_key(b), []).append(b)
# ── public helpers ────────────────────────────────────────────
@staticmethod
def is_local_mode() -> bool:
"""Return ``True`` when running in local (non-server) mode."""
return not _is_server_mode()
# ── role binding queries ──────────────────────────────────────
def get_role_bindings(
self,
principal: str,
scope: PermissionScope,
scope_id: str,
) -> list[RoleBinding]:
"""Return all role bindings for *principal* at *scope*/*scope_id*.
In local mode a synthetic OWNER binding is returned for every
query so that all actions are implicitly permitted.
Raises ``ValueError`` if *principal* or *scope_id* is empty or
whitespace-only after stripping.
"""
principal = _validate_identifier(principal, "principal")
scope_id = _validate_identifier(scope_id, "scope_id")
if self.is_local_mode():
default_role = DEFAULT_LOCAL_ROLE_MAPPING.get(
principal,
DEFAULT_LOCAL_ROLE_MAPPING.get("*", PermissionRole.OWNER),
)
return [
RoleBinding(
principal=principal,
role=default_role,
scope=scope,
scope_id=scope_id,
)
]
key: _BindingKey = (principal, scope, scope_id)
return list(self._bindings_index.get(key, []))
# ── permission check ──────────────────────────────────────────
def check_permission(
self,
principal: str,
action: PermissionAction,
scope: PermissionScope,
scope_id: str,
) -> PermissionCheck:
"""Check whether *principal* may perform *action* on *scope*/*scope_id*.
Returns a ``PermissionCheck`` with the result and a human-readable
reason.
Raises ``ValueError`` if *principal* or *scope_id* is empty or
whitespace-only after stripping.
"""
principal = _validate_identifier(principal, "principal")
scope_id = _validate_identifier(scope_id, "scope_id")
local_mode = self.is_local_mode()
# Fast-path: local mode is always permissive
if local_mode:
return PermissionCheck(
principal=principal,
action=action,
scope=scope,
scope_id=scope_id,
result=True,
reason="local mode — all actions permitted",
)
# Server mode: evaluate role bindings
bindings = self.get_role_bindings(principal, scope, scope_id)
# Check allow-overrides first
for override in self._policy.allow_overrides:
if (
override.principal == principal
and override.scope == scope
and override.scope_id == scope_id
):
allowed_actions = ROLE_PERMISSIONS.get(override.role, frozenset())
if action in allowed_actions:
return PermissionCheck(
principal=principal,
action=action,
scope=scope,
scope_id=scope_id,
result=True,
reason=f"allowed by override ({override.role})",
)
# Check regular bindings
for binding in bindings:
allowed_actions = ROLE_PERMISSIONS.get(binding.role, frozenset())
if action in allowed_actions:
return PermissionCheck(
principal=principal,
action=action,
scope=scope,
scope_id=scope_id,
result=True,
reason=f"allowed by role {binding.role}",
)
# Default deny
if self._policy.default_deny:
result = PermissionCheck(
principal=principal,
action=action,
scope=scope,
scope_id=scope_id,
result=False,
reason="denied by default-deny policy",
)
self._logger.warning(
"permission_denied",
principal=principal,
action=str(action),
scope=str(scope),
scope_id=scope_id,
reason=result.reason,
)
return result
# If default_deny is False and no explicit binding exists,
# permit the action.
return PermissionCheck(
principal=principal,
action=action,
scope=scope,
scope_id=scope_id,
result=True,
reason="no binding found, default-deny disabled",
)
# ── add / remove bindings ─────────────────────────────────────
def add_binding(self, binding: RoleBinding) -> bool:
"""Register a new role binding.
Returns ``True`` if the binding was added, ``False`` if an
identical binding already exists (duplicates are prevented).
"""
key = _binding_key(binding)
bucket = self._bindings_index.setdefault(key, [])
for existing in bucket:
if existing.role == binding.role:
return False
bucket.append(binding)
self._logger.info(
"binding_added",
principal=binding.principal,
role=str(binding.role),
scope=str(binding.scope),
scope_id=binding.scope_id,
)
return True
def remove_binding(self, binding: RoleBinding) -> bool:
"""Remove all matching role bindings.
Returns ``True`` if at least one binding was found and removed.
"""
key = _binding_key(binding)
bucket = self._bindings_index.get(key)
if bucket is None:
return False
original_len = len(bucket)
self._bindings_index[key] = [b for b in bucket if b.role != binding.role]
removed = original_len - len(self._bindings_index[key])
if not self._bindings_index[key]:
del self._bindings_index[key]
if removed > 0:
self._logger.info(
"binding_removed",
principal=binding.principal,
role=str(binding.role),
scope=str(binding.scope),
scope_id=binding.scope_id,
count=removed,
)
return True
return False
# ── Module-level default service ──────────────────────────────────
_default_service: PermissionService | None = None
def get_default_permission_service() -> PermissionService:
"""Return the module-level default ``PermissionService``.
If no default has been configured via
``set_default_permission_service``, a new instance is created on
first access and cached for the lifetime of the process.
"""
global _default_service
if _default_service is None:
_default_service = PermissionService()
return _default_service
def set_default_permission_service(service: PermissionService | None) -> None:
"""Replace (or clear) the module-level default ``PermissionService``."""
global _default_service
_default_service = service
# ── Decorator for enforcement at CLI/service boundaries ───────────
def enforce_permission(
action: PermissionAction,
scope: PermissionScope,
scope_id: str = "*",
*,
principal: str = "local",
service: PermissionService | None = None,
) -> Callable[[F], F]:
"""Decorator that enforces a permission check before function execution.
In local mode the check is a no-op (always allowed). In server
mode a ``PermissionError`` is raised when the check fails.
When *service* is ``None`` the module-level default service
(see ``get_default_permission_service``) is used so that bindings
registered at application startup are visible to every decorated
call site.
"""
def decorator(fn: F) -> F:
@wraps(fn)
def wrapper(*args: object, **kwargs: object) -> object:
svc = service or get_default_permission_service()
check = svc.check_permission(principal, action, scope, scope_id)
if not check.result:
raise PermissionError(
f"Permission denied: {check.reason} "
f"(principal={principal}, action={action}, "
f"scope={scope}, scope_id={scope_id})"
)
return fn(*args, **kwargs)
return wrapper # type: ignore[return-value]
return decorator
@@ -102,6 +102,18 @@ from cleveragents.domain.models.core.org import (
User,
)
# Permission system models
from cleveragents.domain.models.core.permission import (
DEFAULT_LOCAL_ROLE_MAPPING,
ROLE_PERMISSIONS,
PermissionAction,
PermissionCheck,
PermissionPolicy,
PermissionRole,
PermissionScope,
RoleBinding,
)
# New plan lifecycle model
from cleveragents.domain.models.core.plan import (
InvariantSource,
@@ -208,6 +220,8 @@ from cleveragents.domain.models.core.tool import (
__all__ = [
"BUILTIN_PROFILES",
"DEFAULT_LOCAL_ROLE_MAPPING",
"ROLE_PERMISSIONS",
"ActionState",
"Actor",
"AutomationGuard",
@@ -275,6 +289,11 @@ __all__ = [
"OrgRole",
"OrgUser",
"ParsedName",
"PermissionAction",
"PermissionCheck",
"PermissionPolicy",
"PermissionRole",
"PermissionScope",
"PhysVirt",
"Plan",
"PlanBuild",
@@ -308,6 +327,7 @@ __all__ = [
"ResumeMetadata",
"ResumeSummary",
"ReviewArtifact",
"RoleBinding",
"SandboxStrategy",
"Session",
"SessionExportError",
@@ -0,0 +1,170 @@
"""Permission domain models for CleverAgents.
Namespace/project/plan/skill permission model with role bindings
(owner/admin/editor/viewer) and default-deny policy. Server-side
enforcement is deferred; local mode returns permissive defaults.
Based on issue #344 — feat(security): add permission system.
"""
from __future__ import annotations
from enum import StrEnum
from pydantic import BaseModel, ConfigDict, Field
__all__ = [
"DEFAULT_LOCAL_ROLE_MAPPING",
"ROLE_PERMISSIONS",
"PermissionAction",
"PermissionCheck",
"PermissionPolicy",
"PermissionRole",
"PermissionScope",
"RoleBinding",
]
# ── Enumerations ──────────────────────────────────────────────────
class PermissionRole(StrEnum):
"""Role hierarchy for the permission system.
Independent of ``OrgRole`` in ``org.py``, which describes
organisation membership. These roles govern resource-level
access control.
"""
OWNER = "owner"
ADMIN = "admin"
EDITOR = "editor"
VIEWER = "viewer"
class PermissionScope(StrEnum):
"""Scopes to which permissions can be applied."""
NAMESPACE = "namespace"
PROJECT = "project"
PLAN = "plan"
SKILL = "skill"
class PermissionAction(StrEnum):
"""Actions that can be gated by the permission system."""
READ = "read"
WRITE = "write"
EXECUTE = "execute"
DELETE = "delete"
ADMIN = "admin"
# ── Data models ───────────────────────────────────────────────────
class RoleBinding(BaseModel):
"""Binds a principal to a role within a specific scope."""
principal: str = Field(
...,
min_length=1,
description="User or service principal identifier.",
)
role: PermissionRole = Field(
...,
description="Assigned permission role.",
)
scope: PermissionScope = Field(
...,
description="Scope at which the role applies.",
)
scope_id: str = Field(
...,
min_length=1,
description="Identifier of the scoped resource.",
)
model_config = ConfigDict(
str_strip_whitespace=True,
validate_assignment=True,
)
class PermissionCheck(BaseModel):
"""Result of a permission check."""
principal: str = Field(..., description="Checked principal.")
action: PermissionAction = Field(..., description="Action that was checked.")
scope: PermissionScope = Field(..., description="Scope of the check.")
scope_id: str = Field(..., description="Identifier of the scoped resource.")
result: bool = Field(..., description="Whether the action is allowed.")
reason: str = Field(..., description="Human-readable explanation.")
model_config = ConfigDict(
str_strip_whitespace=True,
validate_assignment=True,
)
class PermissionPolicy(BaseModel):
"""Top-level permission policy configuration."""
default_deny: bool = Field(
default=True,
description="When True, actions are denied unless explicitly allowed.",
)
allow_overrides: list[RoleBinding] = Field(
default_factory=list,
description="Explicit allow-overrides that bypass default-deny.",
)
model_config = ConfigDict(
str_strip_whitespace=True,
validate_assignment=True,
)
# ── Constants ─────────────────────────────────────────────────────
ROLE_PERMISSIONS: dict[PermissionRole, frozenset[PermissionAction]] = {
PermissionRole.OWNER: frozenset(PermissionAction),
PermissionRole.ADMIN: frozenset(
{
PermissionAction.READ,
PermissionAction.WRITE,
PermissionAction.EXECUTE,
PermissionAction.DELETE,
PermissionAction.ADMIN,
}
),
PermissionRole.EDITOR: frozenset(
{
PermissionAction.READ,
PermissionAction.WRITE,
PermissionAction.EXECUTE,
}
),
PermissionRole.VIEWER: frozenset(
{
PermissionAction.READ,
}
),
}
"""Mapping of each role to its set of allowed actions.
Note: OWNER and ADMIN currently grant the same set of actions.
The distinction is intentional OWNER designates the resource
creator or namespace owner, while ADMIN is a delegated
administrative role. Future server-mode enforcement may
differentiate them (e.g. ownership transfer, namespace deletion)
without requiring a schema change.
"""
DEFAULT_LOCAL_ROLE_MAPPING: dict[str, PermissionRole] = {
"*": PermissionRole.OWNER,
}
"""In local mode every principal is treated as OWNER."""
+19
View File
@@ -394,3 +394,22 @@ _collect_all_scope_names # noqa: B018, F821
_collect_function_local_names # noqa: B018, F821
_iter_functions # noqa: B018, F821
_is_python_file # noqa: B018, F821
# Permission system — public API (issue #344)
PermissionRole # noqa: B018, F821
PermissionScope # noqa: B018, F821
PermissionAction # noqa: B018, F821
RoleBinding # noqa: B018, F821
PermissionCheck # noqa: B018, F821
PermissionPolicy # noqa: B018, F821
ROLE_PERMISSIONS # noqa: B018, F821
DEFAULT_LOCAL_ROLE_MAPPING # noqa: B018, F821
PermissionService # noqa: B018, F821
enforce_permission # noqa: B018, F821
get_default_permission_service # noqa: B018, F821
set_default_permission_service # noqa: B018, F821
check_permission # noqa: B018, F821
is_local_mode # noqa: B018, F821
get_role_bindings # noqa: B018, F821
add_binding # noqa: B018, F821
remove_binding # noqa: B018, F821