diff --git a/CHANGELOG.md b/CHANGELOG.md index 1c4f2550c..e0054d4d7 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -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. diff --git a/benchmarks/permission_check_bench.py b/benchmarks/permission_check_bench.py new file mode 100644 index 000000000..48a10da30 --- /dev/null +++ b/benchmarks/permission_check_bench.py @@ -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] diff --git a/docs/gen_ref_pages.py b/docs/gen_ref_pages.py index 75adcde16..1093531ec 100644 --- a/docs/gen_ref_pages.py +++ b/docs/gen_ref_pages.py @@ -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" diff --git a/docs/reference/permissions.md b/docs/reference/permissions.md new file mode 100644 index 000000000..f20c990dd --- /dev/null +++ b/docs/reference/permissions.md @@ -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. diff --git a/features/environment.py b/features/environment.py index b658ae13a..0d3badb3c 100644 --- a/features/environment.py +++ b/features/environment.py @@ -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: diff --git a/features/permission_system.feature b/features/permission_system.feature new file mode 100644 index 000000000..52494c043 --- /dev/null +++ b/features/permission_system.feature @@ -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" diff --git a/features/steps/permission_system_steps.py b/features/steps/permission_system_steps.py new file mode 100644 index 000000000..06927b3b8 --- /dev/null +++ b/features/steps/permission_system_steps.py @@ -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 diff --git a/robot/helper_permission_system.py b/robot/helper_permission_system.py new file mode 100644 index 000000000..49bc43d32 --- /dev/null +++ b/robot/helper_permission_system.py @@ -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["*"]) diff --git a/robot/permission_system.robot b/robot/permission_system.robot new file mode 100644 index 000000000..628110eff --- /dev/null +++ b/robot/permission_system.robot @@ -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 diff --git a/src/cleveragents/application/services/__init__.py b/src/cleveragents/application/services/__init__.py index 76ae0db62..331cb96b8 100644 --- a/src/cleveragents/application/services/__init__.py +++ b/src/cleveragents/application/services/__init__.py @@ -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", ] diff --git a/src/cleveragents/application/services/permission_service.py b/src/cleveragents/application/services/permission_service.py new file mode 100644 index 000000000..ea79dbab7 --- /dev/null +++ b/src/cleveragents/application/services/permission_service.py @@ -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 diff --git a/src/cleveragents/domain/models/core/__init__.py b/src/cleveragents/domain/models/core/__init__.py index c5157e1be..0764d927b 100644 --- a/src/cleveragents/domain/models/core/__init__.py +++ b/src/cleveragents/domain/models/core/__init__.py @@ -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", diff --git a/src/cleveragents/domain/models/core/permission.py b/src/cleveragents/domain/models/core/permission.py new file mode 100644 index 000000000..2a2924b75 --- /dev/null +++ b/src/cleveragents/domain/models/core/permission.py @@ -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.""" diff --git a/vulture_whitelist.py b/vulture_whitelist.py index a9259e8c0..b52f6e589 100644 --- a/vulture_whitelist.py +++ b/vulture_whitelist.py @@ -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