diff --git a/AGENTS.md b/AGENTS.md index fdb4caf..167800f 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -136,10 +136,11 @@ See [docs/integrations.md](docs/integrations.md) for MCP and HTTP examples. ## Adding a policy rule -1. Add the rule to `DefaultPolicyEngine.evaluate()` in `policy.py`. -2. **Placement matters:** rules are evaluated in order. A new rule placed before sensitivity checks silently bypasses them. -3. If adding a new `SensitivityTag`, you must also add a corresponding policy rule — otherwise the tag is silently ignored. -4. Cover it with a test in `tests/test_policy.py`. +1. Implement the condition in the appropriate `default_policy_*_rules.py` helper. +2. Register it exactly once in the ordered `_DEFAULT_RULES` tuple in `default_policy_rules.py`; both `evaluate()` and `explain()` consume that chain. +3. **Placement matters:** rules are evaluated in order. A new rule placed before sensitivity checks can change which denial short-circuits first. +4. If adding a new `SensitivityTag`, you must also add a corresponding policy rule — otherwise the tag is silently ignored. +5. Cover decision/explanation agreement in `tests/test_policy_rule_chain.py` and behavior in `tests/test_policy.py`. ## Review checklist (beyond `make ci`) diff --git a/CHANGELOG.md b/CHANGELOG.md index 6a2425c..9f7b858 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -7,6 +7,9 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ## [Unreleased] +### Changed +- **Default policy decisions and explanations now share one ordered rule chain (#219).** `evaluate()` short-circuits the shared chain while `explain()` collects every failure through the same rules. Rate-limit explanation uses a read-only `peek()` so it can predict a denial without consuming, creating, or pruning limiter state. Agreement and no-mutation regressions cover the security-critical boundary. + ## [0.12.0] - 2026-08-14 ### Changed diff --git a/docs/architecture.md b/docs/architecture.md index 35668fe..00899aa 100644 --- a/docs/architecture.md +++ b/docs/architecture.md @@ -99,7 +99,7 @@ Intent-aware rules fail closed: a request with `intent=None` never matches a rul #### Denial explanations -`PolicyEngine.explain()` (when available) returns a structured `DenialExplanation` with `denied`, `rule_name`, a `failed_conditions: list[FailedCondition]` describing each missing condition with `required`/`actual`/`suggestion`/`reason_code`, a `remediation` list, a human-readable `narrative`, and a top-level `reason_code` (the code of the first failed condition). Engines collect all failing conditions (no short-circuit) so callers get the full picture. For `DeclarativePolicyEngine`, an explicit deny rule that fully matches is reported as the cause; partial-match deny rules are skipped during explanation so the surfaced advice is actionable rather than self-defeating. +`PolicyEngine.explain()` (when available) returns a structured `DenialExplanation` with `denied`, `rule_name`, a `failed_conditions: list[FailedCondition]` describing each missing condition with `required`/`actual`/`suggestion`/`reason_code`, a `remediation` list, a human-readable `narrative`, and a top-level `reason_code` (the code of the first failed condition). Engines collect all failing conditions (no short-circuit) so callers get the full picture. `DefaultPolicyEngine.evaluate()` and `.explain()` are driven by one internal ordered rule chain; evaluation short-circuits and records rate usage, while explanation traverses the same rate-limit rule through a read-only `peek()` that never mutates limiter state. For `DeclarativePolicyEngine`, an explicit deny rule that fully matches is reported as the cause; partial-match deny rules are skipped during explanation so the surfaced advice is actionable rather than self-defeating. #### Reason codes diff --git a/src/weaver_kernel/coding_agent.py b/src/weaver_kernel/coding_agent.py index b9e424d..984126b 100644 --- a/src/weaver_kernel/coding_agent.py +++ b/src/weaver_kernel/coding_agent.py @@ -11,6 +11,7 @@ from dataclasses import dataclass from typing import Any, NoReturn +from .default_policy_rule_types import MIN_JUSTIFICATION from .enums import SafetyClass from .errors import DriverError, PolicyDenied from .models import ( @@ -21,7 +22,6 @@ PolicyTraceStep, Principal, ) -from .policy import _MIN_JUSTIFICATION from .policy_matching import scope_globs_match from .policy_reasons import AllowReason, DenialReason @@ -186,10 +186,10 @@ def _require_justification(cls, capability: Capability, justification: str) -> N if capability.safety_class not in (SafetyClass.WRITE, SafetyClass.DESTRUCTIVE): return stripped_len = len(justification.strip()) - if stripped_len < _MIN_JUSTIFICATION: + if stripped_len < MIN_JUSTIFICATION: cls._deny( f"{capability.safety_class.value.upper()} capabilities require a justification " - f"of at least {_MIN_JUSTIFICATION} characters after trimming whitespace.", + f"of at least {MIN_JUSTIFICATION} characters after trimming whitespace.", reason_code=str(DenialReason.INSUFFICIENT_JUSTIFICATION), ) diff --git a/src/weaver_kernel/default_policy_access_rules.py b/src/weaver_kernel/default_policy_access_rules.py new file mode 100644 index 0000000..511968e --- /dev/null +++ b/src/weaver_kernel/default_policy_access_rules.py @@ -0,0 +1,216 @@ +"""Access, sensitivity, and memory checks for the default policy chain.""" + +from __future__ import annotations + +from .default_policy_rule_types import MIN_JUSTIFICATION, RuleContext, RuleFailure +from .enums import SafetyClass, SensitivityTag +from .models import FailedCondition, PolicyTraceStep +from .policy_reasons import DenialReason + + +def _justification_failure(ctx: RuleContext, label: str) -> RuleFailure | None: + stripped_len = len(ctx.justification.strip()) + if stripped_len >= MIN_JUSTIFICATION: + return None + detail = ( + f"{label} capabilities require a justification of at least " + f"{MIN_JUSTIFICATION} characters. Got {len(ctx.justification)} characters " + f"({stripped_len} after trimming whitespace)." + ) + return RuleFailure( + detail=detail, + condition=FailedCondition( + condition="min_justification", + required=MIN_JUSTIFICATION, + actual=stripped_len, + suggestion=( + f"Provide justification with at least {MIN_JUSTIFICATION} " + f"characters (currently {stripped_len})" + ), + reason_code=str(DenialReason.INSUFFICIENT_JUSTIFICATION), + ), + reason_code=str(DenialReason.INSUFFICIENT_JUSTIFICATION), + ) + + +def check_safety_class(ctx: RuleContext) -> list[RuleFailure]: + """Apply WRITE/DESTRUCTIVE role and justification requirements.""" + failures: list[RuleFailure] = [] + roles = set(ctx.principal.roles) + pid = ctx.principal.principal_id + + if ctx.capability.safety_class == SafetyClass.WRITE: + if not (roles & {"writer", "admin"}): + failures.append( + RuleFailure( + detail=( + "WRITE capabilities require the 'writer' or 'admin' role. " + f"Principal '{pid}' has roles: {sorted(roles)}." + ), + condition=FailedCondition( + condition="roles", + required=["writer", "admin"], + actual=sorted(roles), + suggestion=f"Add 'writer' or 'admin' role to principal '{pid}'", + reason_code=str(DenialReason.MISSING_ROLE), + ), + reason_code=str(DenialReason.MISSING_ROLE), + ) + ) + failure = _justification_failure(ctx, "WRITE") + if failure is not None: + failures.append(failure) + + elif ctx.capability.safety_class == SafetyClass.DESTRUCTIVE: + if "admin" not in roles: + failures.append( + RuleFailure( + detail=( + "DESTRUCTIVE capabilities require the 'admin' role. " + f"Principal '{pid}' has roles: {sorted(roles)}." + ), + condition=FailedCondition( + condition="roles", + required=["admin"], + actual=sorted(roles), + suggestion=f"Add 'admin' role to principal '{pid}'", + reason_code=str(DenialReason.MISSING_ROLE), + ), + reason_code=str(DenialReason.MISSING_ROLE), + ) + ) + failure = _justification_failure(ctx, "DESTRUCTIVE") + if failure is not None: + failures.append(failure) + + return failures + + +def check_tenant_sensitivity(ctx: RuleContext) -> list[RuleFailure]: + """Apply PII/PCI tenant requirements and allowed-field narrowing.""" + if ctx.capability.sensitivity not in (SensitivityTag.PII, SensitivityTag.PCI): + return [] + pid = ctx.principal.principal_id + if "tenant" not in ctx.principal.attributes: + return [ + RuleFailure( + detail=( + f"Capability '{ctx.capability.capability_id}' has " + f"{ctx.capability.sensitivity.value} sensitivity and requires " + "the principal to have a 'tenant' attribute." + ), + condition=FailedCondition( + condition="tenant_attribute", + required="present", + actual="absent", + suggestion=f"Add 'tenant' attribute to principal '{pid}'", + reason_code=str(DenialReason.MISSING_TENANT_ATTRIBUTE), + ), + reason_code=str(DenialReason.MISSING_TENANT_ATTRIBUTE), + ) + ] + + roles = set(ctx.principal.roles) + if ctx.capability.allowed_fields and "pii_reader" not in roles: + ctx.constraints["allowed_fields"] = ctx.capability.allowed_fields + ctx.trace_steps.append( + PolicyTraceStep( + name="sensitivity:allowed_fields", + outcome="constraint_applied", + detail=f"applied allowed_fields={ctx.capability.allowed_fields}", + ) + ) + return [] + + +def check_secrets(ctx: RuleContext) -> list[RuleFailure]: + """Apply SECRETS role and justification requirements.""" + if ctx.capability.sensitivity != SensitivityTag.SECRETS: + return [] + failures: list[RuleFailure] = [] + roles = set(ctx.principal.roles) + pid = ctx.principal.principal_id + if not (roles & {"admin", "secrets_reader"}): + failures.append( + RuleFailure( + detail=( + "SECRETS capabilities require the 'admin' or 'secrets_reader' role. " + f"Principal '{pid}' has roles: {sorted(roles)}." + ), + condition=FailedCondition( + condition="roles", + required=["admin", "secrets_reader"], + actual=sorted(roles), + suggestion=f"Add 'admin' or 'secrets_reader' role to principal '{pid}'", + reason_code=str(DenialReason.MISSING_ROLE), + ), + reason_code=str(DenialReason.MISSING_ROLE), + ) + ) + failure = _justification_failure(ctx, "SECRETS") + if failure is not None: + failures.append(failure) + return failures + + +def check_memory(ctx: RuleContext) -> list[RuleFailure]: + """Apply MEMORY write and sensitive-read role requirements.""" + if ctx.capability.sensitivity != SensitivityTag.MEMORY: + return [] + roles = set(ctx.principal.roles) + pid = ctx.principal.principal_id + memory_scope = str(ctx.request.scope.get("memory_scope", "")) if ctx.request.scope else "" + is_write = ctx.capability.safety_class in (SafetyClass.WRITE, SafetyClass.DESTRUCTIVE) + + if is_write and not (roles & {"memory_writer", "admin"}): + return [ + RuleFailure( + detail=( + "MEMORY write capabilities require the 'memory_writer' or 'admin' role. " + f"Principal '{pid}' has roles: {sorted(roles)}." + ), + condition=FailedCondition( + condition="roles", + required=["memory_writer", "admin"], + actual=sorted(roles), + suggestion=f"Add 'memory_writer' or 'admin' role to principal '{pid}'", + reason_code=str(DenialReason.MEMORY_WRITE_REQUIRES_WRITER), + ), + reason_code=str(DenialReason.MEMORY_WRITE_REQUIRES_WRITER), + ) + ] + + if ( + not is_write + and memory_scope == "sensitive" + and not (roles & {"memory_reader_sensitive", "admin"}) + ): + return [ + RuleFailure( + detail=( + "MEMORY read with scope='sensitive' requires the " + f"'memory_reader_sensitive' or 'admin' role. Principal '{pid}' " + f"has roles: {sorted(roles)}." + ), + condition=FailedCondition( + condition="roles", + required=["memory_reader_sensitive", "admin"], + actual=sorted(roles), + suggestion=( + f"Add 'memory_reader_sensitive' or 'admin' role to principal '{pid}' " + "(or narrow the request scope away from 'sensitive')" + ), + reason_code=str(DenialReason.MEMORY_SENSITIVE_READ_DENIED), + ), + reason_code=str(DenialReason.MEMORY_SENSITIVE_READ_DENIED), + ) + ] + return [] + + +__all__ = [ + "check_memory", + "check_safety_class", + "check_secrets", + "check_tenant_sensitivity", +] diff --git a/src/weaver_kernel/default_policy_limit_rules.py b/src/weaver_kernel/default_policy_limit_rules.py new file mode 100644 index 0000000..f41ce2c --- /dev/null +++ b/src/weaver_kernel/default_policy_limit_rules.py @@ -0,0 +1,98 @@ +"""Constraint and rate-limit checks for the default policy chain.""" + +from __future__ import annotations + +from .default_policy_rule_types import ( + MAX_ROWS_SERVICE, + MAX_ROWS_USER, + RuleContext, + RuleFailure, +) +from .models import FailedCondition, PolicyTraceStep +from .policy_reasons import DenialReason +from .rate_limit import SERVICE_RATE_MULTIPLIER + + +def apply_row_cap(ctx: RuleContext) -> list[RuleFailure]: + """Validate/cap ``max_rows`` and record the applied constraint.""" + roles = set(ctx.principal.roles) + max_rows = MAX_ROWS_SERVICE if "service" in roles else MAX_ROWS_USER + if "max_rows" in ctx.constraints: + try: + requested = int(ctx.constraints["max_rows"]) + except (TypeError, ValueError) as exc: + return [ + RuleFailure( + detail=( + f"Invalid 'max_rows' constraint: {ctx.constraints['max_rows']!r} " + "is not a valid integer." + ), + condition=FailedCondition( + condition="max_rows", + required="integer", + actual=ctx.constraints["max_rows"], + suggestion="Provide 'max_rows' as a valid integer", + reason_code=str(DenialReason.INVALID_CONSTRAINT), + ), + reason_code=str(DenialReason.INVALID_CONSTRAINT), + cause=exc, + ) + ] + ctx.constraints["max_rows"] = min(max(requested, 0), max_rows) + else: + ctx.constraints["max_rows"] = max_rows + + ctx.trace_steps.append( + PolicyTraceStep( + name="row_cap", + outcome="constraint_applied", + detail="max_rows capped", + ) + ) + return [] + + +def check_rate_limit(ctx: RuleContext) -> list[RuleFailure]: + """Check the current sliding window and record usage only for decisions.""" + safety_class = ctx.capability.safety_class + if safety_class not in ctx.rate_limits: + return [] + + roles = set(ctx.principal.roles) + limit, window = ctx.rate_limits[safety_class] + if "service" in roles: + limit *= SERVICE_RATE_MULTIPLIER + pid = ctx.principal.principal_id + cid = ctx.capability.capability_id + rate_key = f"{pid}:{cid}" + allowed = ( + ctx.limiter.peek(rate_key, limit, window) + if ctx.read_only + else ctx.limiter.check(rate_key, limit, window) + ) + if not allowed: + return [ + RuleFailure( + detail=( + f"Rate limit exceeded: {limit} {safety_class.value} " + f"invocations per {window}s for principal '{pid}'" + ), + condition=FailedCondition( + condition="rate_limit", + required=f"fewer than {limit} invocations per {window}s", + actual="limit exceeded", + suggestion=( + f"Wait for the {window}s rate-limit window before retrying " + f"capability '{cid}'" + ), + reason_code=str(DenialReason.RATE_LIMITED), + ), + reason_code=str(DenialReason.RATE_LIMITED), + ) + ] + if not ctx.read_only: + ctx.limiter.record(rate_key) + return [] + + +__all__ = ["apply_row_cap", "check_rate_limit"] diff --git a/src/weaver_kernel/default_policy_rule_types.py b/src/weaver_kernel/default_policy_rule_types.py new file mode 100644 index 0000000..89aa4de --- /dev/null +++ b/src/weaver_kernel/default_policy_rule_types.py @@ -0,0 +1,64 @@ +"""Internal data structures shared by the default-policy rule modules.""" + +from __future__ import annotations + +from dataclasses import dataclass, field +from typing import Any + +from .enums import SafetyClass +from .models import ( + Capability, + CapabilityRequest, + FailedCondition, + PolicyTraceStep, + Principal, +) +from .rate_limit import RateLimiter + +MIN_JUSTIFICATION = 15 +MAX_ROWS_USER = 50 +MAX_ROWS_SERVICE = 500 + + +@dataclass(slots=True) +class RuleFailure: + """One failed default-policy condition and its decision/explanation views.""" + + detail: str + condition: FailedCondition + reason_code: str + cause: Exception | None = None + + +@dataclass(slots=True) +class RuleContext: + """Mutable traversal context shared by the ordered rule checks.""" + + request: CapabilityRequest + capability: Capability + principal: Principal + justification: str + constraints: dict[str, Any] + rate_limits: dict[SafetyClass, tuple[int, float]] + limiter: RateLimiter + read_only: bool + trace_steps: list[PolicyTraceStep] = field(default_factory=list) + + +@dataclass(slots=True) +class RuleChainResult: + """Result of traversing the ordered default-policy rule chain.""" + + constraints: dict[str, Any] + failures: list[RuleFailure] = field(default_factory=list) + trace_steps: list[PolicyTraceStep] = field(default_factory=list) + + +__all__ = [ + "MAX_ROWS_SERVICE", + "MAX_ROWS_USER", + "MIN_JUSTIFICATION", + "RuleChainResult", + "RuleContext", + "RuleFailure", +] diff --git a/src/weaver_kernel/default_policy_rules.py b/src/weaver_kernel/default_policy_rules.py new file mode 100644 index 0000000..afbbf6f --- /dev/null +++ b/src/weaver_kernel/default_policy_rules.py @@ -0,0 +1,112 @@ +"""Ordered rule chain shared by default policy decisions and explanations.""" + +from __future__ import annotations + +from collections.abc import Callable + +from .default_policy_access_rules import ( + check_memory, + check_safety_class, + check_secrets, + check_tenant_sensitivity, +) +from .default_policy_limit_rules import apply_row_cap, check_rate_limit +from .default_policy_rule_types import ( + MAX_ROWS_SERVICE, + MAX_ROWS_USER, + MIN_JUSTIFICATION, + RuleChainResult, + RuleContext, + RuleFailure, +) +from .enums import SafetyClass +from .models import Capability, CapabilityRequest, Principal +from .rate_limit import RateLimiter + +RuleCheck = Callable[[RuleContext], list[RuleFailure]] + +# Canonical order is defined once here. Both evaluate() and explain() traverse +# exactly this sequence; their only differences are short-circuit/collect-all +# and stateful/read-only rate-limit modes. +_DEFAULT_RULES: tuple[RuleCheck, ...] = ( + check_safety_class, + check_tenant_sensitivity, + check_secrets, + check_memory, + apply_row_cap, + check_rate_limit, +) + + +class DefaultPolicyRuleChain: + """Single ordered definition of the built-in default policy rules.""" + + def __init__( + self, + *, + rate_limits: dict[SafetyClass, tuple[int, float]], + limiter: RateLimiter, + ) -> None: + self._rate_limits = rate_limits + self._limiter = limiter + + def run( + self, + request: CapabilityRequest, + capability: Capability, + principal: Principal, + *, + justification: str, + collect_all: bool, + read_only: bool, + ) -> RuleChainResult: + """Traverse the canonical rules in decision or explanation mode. + + Args: + request: Capability request being checked. + capability: Target capability. + principal: Requesting principal. + justification: Caller-supplied justification. + collect_all: Collect every failed condition instead of stopping at + the first one. + read_only: Avoid policy-state mutation, including rate-window + creation, pruning, and usage recording. + + Returns: + Constraints, failures, and non-terminal trace steps produced by the + common rule traversal. + """ + ctx = RuleContext( + request=request, + capability=capability, + principal=principal, + justification=justification, + constraints=dict(request.constraints), + rate_limits=self._rate_limits, + limiter=self._limiter, + read_only=read_only, + ) + failures: list[RuleFailure] = [] + for rule in _DEFAULT_RULES: + rule_failures = rule(ctx) + if rule_failures: + failures.extend(rule_failures) + if not collect_all: + failures = failures[:1] + break + + return RuleChainResult( + constraints=ctx.constraints, + failures=failures, + trace_steps=ctx.trace_steps, + ) + + +__all__ = [ + "DefaultPolicyRuleChain", + "MAX_ROWS_SERVICE", + "MAX_ROWS_USER", + "MIN_JUSTIFICATION", + "RuleChainResult", + "RuleFailure", +] diff --git a/src/weaver_kernel/policy.py b/src/weaver_kernel/policy.py index 2d7dde3..1dd338c 100644 --- a/src/weaver_kernel/policy.py +++ b/src/weaver_kernel/policy.py @@ -4,32 +4,25 @@ import logging from collections.abc import Callable -from typing import Any, Protocol +from typing import Protocol -from .enums import SafetyClass, SensitivityTag +from .default_policy_rules import DefaultPolicyRuleChain +from .enums import SafetyClass from .errors import AgentKernelError, PolicyDenied from .models import ( Capability, CapabilityRequest, DenialExplanation, - FailedCondition, PolicyDecision, PolicyDecisionTrace, PolicyTraceStep, Principal, ) -from .policy_reasons import AllowReason, DenialReason +from .policy_reasons import AllowReason from .rate_limit import DEFAULT_RATE_LIMITS, SERVICE_RATE_MULTIPLIER, RateLimiter logger = logging.getLogger(__name__) -# Minimum justification length for WRITE operations. -_MIN_JUSTIFICATION = 15 - -# Default max_rows caps. -_MAX_ROWS_USER = 50 -_MAX_ROWS_SERVICE = 500 - # Backwards-compatible aliases — these used to be defined here. New code # should import the names without the leading underscore from ``rate_limit``. _DEFAULT_RATE_LIMITS = DEFAULT_RATE_LIMITS @@ -156,6 +149,9 @@ def __init__( ) self._rate_limits = limits self._limiter = RateLimiter(clock=clock) + self._rule_chain = DefaultPolicyRuleChain( + rate_limits=self._rate_limits, limiter=self._limiter + ) @staticmethod def _deny( @@ -185,27 +181,14 @@ def evaluate( *, justification: str, ) -> PolicyDecision: - """Evaluate the request against the default policy rules. - - Args: - request: The capability request being evaluated. - capability: The target capability. - principal: The requesting principal. - justification: Free-text justification from the caller. + """Evaluate the request against the shared default-policy rule chain. - Returns: - :class:`PolicyDecision` with ``allowed=True`` and any imposed - constraints, or raises :class:`PolicyDenied`. - - Raises: - PolicyDenied: When the request violates a policy rule. + Decision traversal short-circuits on the first denial and may update + transient policy state (currently the sliding-window rate limiter). + ``explain()`` traverses this exact same chain in read-only mode. """ - roles = set(principal.roles) - constraints: dict[str, Any] = dict(request.constraints) - pid = principal.principal_id cid = capability.capability_id - trace = PolicyDecisionTrace( engine="DefaultPolicyEngine", capability_id=cid, @@ -213,227 +196,37 @@ def evaluate( intent=request.intent, scope_keys=sorted(request.scope.keys()), ) + result = self._rule_chain.run( + request, + capability, + principal, + justification=justification, + collect_all=False, + read_only=False, + ) + trace.steps.extend(result.trace_steps) - def _record_deny(detail: str, code: str) -> None: + if result.failures: + failure = result.failures[0] trace.steps.append( PolicyTraceStep( name="deny", outcome="denied", - detail=detail, - reason_code=code, + detail=failure.detail, + reason_code=failure.reason_code, ) ) trace.final_outcome = "denied" - trace.final_reason_code = code - - # ── Safety class checks ─────────────────────────────────────────────── - - if capability.safety_class == SafetyClass.WRITE: - if not (roles & {"writer", "admin"}): - detail = ( - f"WRITE capabilities require the 'writer' or 'admin' role. " - f"Principal '{pid}' has roles: {sorted(roles)}." - ) - _record_deny(detail, DenialReason.MISSING_ROLE) - raise self._deny( - detail, - principal_id=pid, - capability_id=cid, - reason_code=DenialReason.MISSING_ROLE, - ) - stripped_len = len(justification.strip()) - if stripped_len < _MIN_JUSTIFICATION: - detail = ( - f"WRITE capabilities require a justification of at least " - f"{_MIN_JUSTIFICATION} characters. " - f"Got {len(justification)} characters " - f"({stripped_len} after trimming whitespace)." - ) - _record_deny(detail, DenialReason.INSUFFICIENT_JUSTIFICATION) - raise self._deny( - detail, - principal_id=pid, - capability_id=cid, - reason_code=DenialReason.INSUFFICIENT_JUSTIFICATION, - ) - - elif capability.safety_class == SafetyClass.DESTRUCTIVE: - if "admin" not in roles: - detail = ( - f"DESTRUCTIVE capabilities require the 'admin' role. " - f"Principal '{pid}' has roles: {sorted(roles)}." - ) - _record_deny(detail, DenialReason.MISSING_ROLE) - raise self._deny( - detail, - principal_id=pid, - capability_id=cid, - reason_code=DenialReason.MISSING_ROLE, - ) - stripped_len = len(justification.strip()) - if stripped_len < _MIN_JUSTIFICATION: - detail = ( - f"DESTRUCTIVE capabilities require a justification of at least " - f"{_MIN_JUSTIFICATION} characters. " - f"Got {len(justification)} characters " - f"({stripped_len} after trimming whitespace)." - ) - _record_deny(detail, DenialReason.INSUFFICIENT_JUSTIFICATION) - raise self._deny( - detail, - principal_id=pid, - capability_id=cid, - reason_code=DenialReason.INSUFFICIENT_JUSTIFICATION, - ) - - # ── Sensitivity checks ──────────────────────────────────────────────── - - if capability.sensitivity in (SensitivityTag.PII, SensitivityTag.PCI): - if "tenant" not in principal.attributes: - detail = ( - f"Capability '{cid}' has " - f"{capability.sensitivity.value} sensitivity and requires " - "the principal to have a 'tenant' attribute." - ) - _record_deny(detail, DenialReason.MISSING_TENANT_ATTRIBUTE) - raise self._deny( - detail, - principal_id=pid, - capability_id=cid, - reason_code=DenialReason.MISSING_TENANT_ATTRIBUTE, - ) - # Enforce allowed_fields unless the principal is a pii_reader. - if capability.allowed_fields and "pii_reader" not in roles: - constraints["allowed_fields"] = capability.allowed_fields - trace.steps.append( - PolicyTraceStep( - name="sensitivity:allowed_fields", - outcome="constraint_applied", - detail=f"applied allowed_fields={capability.allowed_fields}", - ) - ) - - if capability.sensitivity == SensitivityTag.SECRETS: - if not (roles & {"admin", "secrets_reader"}): - detail = ( - f"SECRETS capabilities require the 'admin' or 'secrets_reader' role. " - f"Principal '{pid}' has roles: {sorted(roles)}." - ) - _record_deny(detail, DenialReason.MISSING_ROLE) - raise self._deny( - detail, - principal_id=pid, - capability_id=cid, - reason_code=DenialReason.MISSING_ROLE, - ) - stripped_len = len(justification.strip()) - if stripped_len < _MIN_JUSTIFICATION: - detail = ( - f"SECRETS capabilities require a justification of at least " - f"{_MIN_JUSTIFICATION} characters. " - f"Got {len(justification)} characters " - f"({stripped_len} after trimming whitespace)." - ) - _record_deny(detail, DenialReason.INSUFFICIENT_JUSTIFICATION) - raise self._deny( - detail, - principal_id=pid, - capability_id=cid, - reason_code=DenialReason.INSUFFICIENT_JUSTIFICATION, - ) - - # ── Memory action checks ───────────────────────────────────────────── - # Placed AFTER all other sensitivity checks (see invariants.md: - # "rule placement matters"). Memory reads at scope == "sensitive" - # require an explicit reader role; memory writes are treated as - # higher-risk than reads because they persist into future sessions - # and require the 'memory_writer' role (or 'admin'). - if capability.sensitivity == SensitivityTag.MEMORY: - memory_scope = str(request.scope.get("memory_scope", "")) if request.scope else "" - is_write = capability.safety_class in ( - SafetyClass.WRITE, - SafetyClass.DESTRUCTIVE, + trace.final_reason_code = failure.reason_code + denial = self._deny( + failure.detail, + principal_id=pid, + capability_id=cid, + reason_code=failure.reason_code, ) - if is_write and not (roles & {"memory_writer", "admin"}): - detail = ( - f"MEMORY write capabilities require the 'memory_writer' or " - f"'admin' role. Principal '{pid}' has roles: {sorted(roles)}." - ) - _record_deny(detail, DenialReason.MEMORY_WRITE_REQUIRES_WRITER) - raise self._deny( - detail, - principal_id=pid, - capability_id=cid, - reason_code=DenialReason.MEMORY_WRITE_REQUIRES_WRITER, - ) - if ( - not is_write - and memory_scope == "sensitive" - and not (roles & {"memory_reader_sensitive", "admin"}) - ): - detail = ( - f"MEMORY read with scope='sensitive' requires the " - f"'memory_reader_sensitive' or 'admin' role. " - f"Principal '{pid}' has roles: {sorted(roles)}." - ) - _record_deny(detail, DenialReason.MEMORY_SENSITIVE_READ_DENIED) - raise self._deny( - detail, - principal_id=pid, - capability_id=cid, - reason_code=DenialReason.MEMORY_SENSITIVE_READ_DENIED, - ) - - # ── Row cap ─────────────────────────────────────────────────────────── - - max_rows = _MAX_ROWS_SERVICE if "service" in roles else _MAX_ROWS_USER - # Respect any tighter constraint from the request itself. - if "max_rows" in constraints: - try: - requested = int(constraints["max_rows"]) - except (TypeError, ValueError) as exc: - detail = ( - f"Invalid 'max_rows' constraint: {constraints['max_rows']!r} " - "is not a valid integer." - ) - _record_deny(detail, DenialReason.INVALID_CONSTRAINT) - raise self._deny( - detail, - principal_id=pid, - capability_id=cid, - reason_code=DenialReason.INVALID_CONSTRAINT, - ) from exc - constraints["max_rows"] = min(max(requested, 0), max_rows) - else: - constraints["max_rows"] = max_rows - trace.steps.append( - PolicyTraceStep( - name="row_cap", - outcome="constraint_applied", - detail="max_rows capped", - ) - ) - - # ── Rate limiting ───────────────────────────────────────────────── - - rate_key = f"{pid}:{cid}" - if capability.safety_class in self._rate_limits: - limit, window = self._rate_limits[capability.safety_class] - if "service" in roles: - limit *= _SERVICE_RATE_MULTIPLIER - if not self._limiter.check(rate_key, limit, window): - detail = ( - f"Rate limit exceeded: {limit} {capability.safety_class.value} " - f"invocations per {window}s for principal '{pid}'" - ) - _record_deny(detail, DenialReason.RATE_LIMITED) - raise self._deny( - detail, - principal_id=pid, - capability_id=cid, - reason_code=DenialReason.RATE_LIMITED, - ) - self._limiter.record(rate_key) + if failure.cause is not None: + raise denial from failure.cause + raise denial reason = "Request approved by DefaultPolicyEngine." trace.steps.append( @@ -458,7 +251,7 @@ def _record_deny(detail: str, code: str) -> None: return PolicyDecision( allowed=True, reason=reason, - constraints=constraints, + constraints=result.constraints, reason_code=str(AllowReason.DEFAULT_POLICY_ALLOW), trace=trace, ) @@ -471,160 +264,25 @@ def explain( *, justification: str, ) -> DenialExplanation: - """Explain which policy conditions would deny *principal*'s *request*. - - Traverses the same rule chain as :meth:`evaluate` but collects ALL - failing conditions instead of short-circuiting on the first failure. - Rate-limit state is excluded — it is transient and not remediable - by changing the request. - - Args: - request: The capability request to explain. - capability: The target capability. - principal: The requesting principal. - justification: Free-text justification from the caller. + """Explain all failures from the same chain used by :meth:`evaluate`. - Returns: - :class:`DenialExplanation` with ``denied=False`` if allowed. + Explanation is strictly read-only: it collects all failed conditions, + including the current rate-limit condition, without recording usage or + pruning/creating limiter windows. """ - roles = set(principal.roles) pid = principal.principal_id cid = capability.capability_id - failed: list[FailedCondition] = [] - - # ── Safety class checks ─────────────────────────────────────────────── - - if capability.safety_class == SafetyClass.WRITE: - if not (roles & {"writer", "admin"}): - failed.append( - FailedCondition( - condition="roles", - required=["writer", "admin"], - actual=sorted(roles), - suggestion=f"Add 'writer' or 'admin' role to principal '{pid}'", - reason_code=str(DenialReason.MISSING_ROLE), - ) - ) - stripped = len(justification.strip()) - if stripped < _MIN_JUSTIFICATION: - failed.append( - FailedCondition( - condition="min_justification", - required=_MIN_JUSTIFICATION, - actual=stripped, - suggestion=( - f"Provide justification with at least {_MIN_JUSTIFICATION} " - f"characters (currently {stripped})" - ), - reason_code=str(DenialReason.INSUFFICIENT_JUSTIFICATION), - ) - ) - - elif capability.safety_class == SafetyClass.DESTRUCTIVE: - if "admin" not in roles: - failed.append( - FailedCondition( - condition="roles", - required=["admin"], - actual=sorted(roles), - suggestion=f"Add 'admin' role to principal '{pid}'", - reason_code=str(DenialReason.MISSING_ROLE), - ) - ) - stripped = len(justification.strip()) - if stripped < _MIN_JUSTIFICATION: - failed.append( - FailedCondition( - condition="min_justification", - required=_MIN_JUSTIFICATION, - actual=stripped, - suggestion=( - f"Provide justification with at least {_MIN_JUSTIFICATION} " - f"characters (currently {stripped})" - ), - reason_code=str(DenialReason.INSUFFICIENT_JUSTIFICATION), - ) - ) - - # ── Sensitivity checks ──────────────────────────────────────────────── - - if ( - capability.sensitivity in (SensitivityTag.PII, SensitivityTag.PCI) - and "tenant" not in principal.attributes - ): - failed.append( - FailedCondition( - condition="tenant_attribute", - required="present", - actual="absent", - suggestion=f"Add 'tenant' attribute to principal '{pid}'", - reason_code=str(DenialReason.MISSING_TENANT_ATTRIBUTE), - ) - ) - - if capability.sensitivity == SensitivityTag.SECRETS: - if not (roles & {"admin", "secrets_reader"}): - failed.append( - FailedCondition( - condition="roles", - required=["admin", "secrets_reader"], - actual=sorted(roles), - suggestion=f"Add 'admin' or 'secrets_reader' role to principal '{pid}'", - reason_code=str(DenialReason.MISSING_ROLE), - ) - ) - stripped = len(justification.strip()) - if stripped < _MIN_JUSTIFICATION: - failed.append( - FailedCondition( - condition="min_justification", - required=_MIN_JUSTIFICATION, - actual=stripped, - suggestion=( - f"Provide justification with at least {_MIN_JUSTIFICATION} " - f"characters (currently {stripped})" - ), - reason_code=str(DenialReason.INSUFFICIENT_JUSTIFICATION), - ) - ) - - if capability.sensitivity == SensitivityTag.MEMORY: - memory_scope = str(request.scope.get("memory_scope", "")) if request.scope else "" - is_write = capability.safety_class in ( - SafetyClass.WRITE, - SafetyClass.DESTRUCTIVE, - ) - if is_write and not (roles & {"memory_writer", "admin"}): - failed.append( - FailedCondition( - condition="roles", - required=["memory_writer", "admin"], - actual=sorted(roles), - suggestion=(f"Add 'memory_writer' or 'admin' role to principal '{pid}'"), - reason_code=str(DenialReason.MEMORY_WRITE_REQUIRES_WRITER), - ) - ) - if ( - not is_write - and memory_scope == "sensitive" - and not (roles & {"memory_reader_sensitive", "admin"}) - ): - failed.append( - FailedCondition( - condition="roles", - required=["memory_reader_sensitive", "admin"], - actual=sorted(roles), - suggestion=( - f"Add 'memory_reader_sensitive' or 'admin' role to " - f"principal '{pid}' (or narrow the request scope away " - f"from 'sensitive')" - ), - reason_code=str(DenialReason.MEMORY_SENSITIVE_READ_DENIED), - ) - ) - + result = self._rule_chain.run( + request, + capability, + principal, + justification=justification, + collect_all=True, + read_only=True, + ) + failed = [failure.condition for failure in result.failures] denied = bool(failed) - remediation = [fc.suggestion for fc in failed] + remediation = [condition.suggestion for condition in failed] if denied: first = failed[0] @@ -633,7 +291,7 @@ def explain( ) narrative = ( f"Request for '{cid}' by '{pid}' would be denied: " - + "; ".join(fc.suggestion for fc in failed) + + "; ".join(condition.suggestion for condition in failed) + "." ) primary_code = first.reason_code diff --git a/src/weaver_kernel/rate_limit.py b/src/weaver_kernel/rate_limit.py index b74d664..bbef7cc 100644 --- a/src/weaver_kernel/rate_limit.py +++ b/src/weaver_kernel/rate_limit.py @@ -66,6 +66,22 @@ def check(self, key: str, limit: int, window_seconds: float) -> bool: return True return len(entry.timestamps) < limit + def peek(self, key: str, limit: int, window_seconds: float) -> bool: + """Read-only counterpart to :meth:`check`. + + Returns whether the next invocation would be within the limit without + creating a window, pruning expired timestamps, or otherwise mutating + limiter state. Policy explanation uses this path so explaining a + decision can never consume or rewrite rate-limit budget. + """ + now = self._clock() + cutoff = now - window_seconds + entry = self._windows.get(key) + if entry is None: + return True + active = sum(timestamp > cutoff for timestamp in entry.timestamps) + return active < limit + def record(self, key: str) -> None: """Record an invocation for *key*.""" self._windows[key].timestamps.append(self._clock()) diff --git a/tests/test_policy_properties.py b/tests/test_policy_properties.py index 63a26aa..51e6914 100644 --- a/tests/test_policy_properties.py +++ b/tests/test_policy_properties.py @@ -62,7 +62,10 @@ TokenScopeError, export_action_traces, ) -from weaver_kernel.policy import _MAX_ROWS_SERVICE, _MAX_ROWS_USER +from weaver_kernel.default_policy_rule_types import ( + MAX_ROWS_SERVICE, + MAX_ROWS_USER, +) # ── Shared strategies & helpers ───────────────────────────────────────────── @@ -207,7 +210,7 @@ def test_max_rows_never_exceeds_policy_cap( capability_id=capability.capability_id, goal="g", constraints=constraints ) decision = engine.evaluate(request, capability, principal, justification="") - cap_limit = _MAX_ROWS_SERVICE if "service" in principal.roles else _MAX_ROWS_USER + cap_limit = MAX_ROWS_SERVICE if "service" in principal.roles else MAX_ROWS_USER capped = decision.constraints["max_rows"] assert 0 <= capped <= cap_limit if requested_max_rows is not None and requested_max_rows >= 0: diff --git a/tests/test_policy_rule_chain.py b/tests/test_policy_rule_chain.py new file mode 100644 index 0000000..b8c08e7 --- /dev/null +++ b/tests/test_policy_rule_chain.py @@ -0,0 +1,261 @@ +"""Agreement and read-only invariants for DefaultPolicyEngine's shared rule chain.""" + +from __future__ import annotations + +from copy import deepcopy + +import pytest + +from weaver_kernel import ( + Capability, + DefaultPolicyEngine, + PolicyDenied, + Principal, + SafetyClass, + SensitivityTag, +) +from weaver_kernel.models import CapabilityRequest +from weaver_kernel.policy_reasons import DenialReason + + +def _cap( + safety: SafetyClass, + *, + sensitivity: SensitivityTag = SensitivityTag.NONE, + allowed_fields: list[str] | None = None, +) -> Capability: + return Capability( + capability_id="cap.test", + name="test", + description="test capability", + safety_class=safety, + sensitivity=sensitivity, + allowed_fields=allowed_fields or [], + ) + + +def _request( + *, max_rows: object | None = None, memory_scope: str | None = None +) -> CapabilityRequest: + constraints = {} if max_rows is None else {"max_rows": max_rows} + scope = {} if memory_scope is None else {"memory_scope": memory_scope} + return CapabilityRequest( + capability_id="cap.test", + goal="test", + constraints=constraints, + scope=scope, + ) + + +_CASES = [ + pytest.param( + _request(), + _cap(SafetyClass.READ), + Principal(principal_id="reader"), + "", + False, + None, + id="read-allowed", + ), + pytest.param( + _request(), + _cap(SafetyClass.WRITE), + Principal(principal_id="no-writer", roles=["reader"]), + "long enough justification", + True, + str(DenialReason.MISSING_ROLE), + id="write-role", + ), + pytest.param( + _request(), + _cap(SafetyClass.WRITE), + Principal(principal_id="writer", roles=["writer"]), + "short", + True, + str(DenialReason.INSUFFICIENT_JUSTIFICATION), + id="write-justification", + ), + pytest.param( + _request(), + _cap(SafetyClass.DESTRUCTIVE), + Principal(principal_id="not-admin", roles=["writer"]), + "long enough justification", + True, + str(DenialReason.MISSING_ROLE), + id="destructive-role", + ), + pytest.param( + _request(), + _cap(SafetyClass.READ, sensitivity=SensitivityTag.PII), + Principal(principal_id="pii"), + "", + True, + str(DenialReason.MISSING_TENANT_ATTRIBUTE), + id="pii-tenant", + ), + pytest.param( + _request(), + _cap(SafetyClass.READ, sensitivity=SensitivityTag.SECRETS), + Principal(principal_id="secret-reader", roles=["reader"]), + "long enough justification", + True, + str(DenialReason.MISSING_ROLE), + id="secrets-role", + ), + pytest.param( + _request(), + _cap(SafetyClass.WRITE, sensitivity=SensitivityTag.MEMORY), + Principal(principal_id="memory-writer", roles=["writer"]), + "long enough justification", + True, + str(DenialReason.MEMORY_WRITE_REQUIRES_WRITER), + id="memory-write-role", + ), + pytest.param( + _request(memory_scope="sensitive"), + _cap(SafetyClass.READ, sensitivity=SensitivityTag.MEMORY), + Principal(principal_id="memory-reader", roles=["reader"]), + "", + True, + str(DenialReason.MEMORY_SENSITIVE_READ_DENIED), + id="memory-sensitive-read-role", + ), + pytest.param( + _request(max_rows="not-an-int"), + _cap(SafetyClass.READ), + Principal(principal_id="invalid-constraint"), + "", + True, + str(DenialReason.INVALID_CONSTRAINT), + id="invalid-max-rows", + ), + pytest.param( + _request(max_rows=9999), + _cap(SafetyClass.READ, sensitivity=SensitivityTag.PII, allowed_fields=["id"]), + Principal(principal_id="service", roles=["service"], attributes={"tenant": "acme"}), + "", + False, + None, + id="allowed-with-constraints", + ), +] + + +@pytest.mark.parametrize( + ( + "cap_request", + "capability", + "principal", + "justification", + "denied", + "reason_code", + ), + _CASES, +) +def test_explain_prediction_matches_evaluate( + cap_request: CapabilityRequest, + capability: Capability, + principal: Principal, + justification: str, + denied: bool, + reason_code: str | None, +) -> None: + engine = DefaultPolicyEngine() + + explanation = engine.explain( + cap_request, + capability, + principal, + justification=justification, + ) + + try: + decision = engine.evaluate( + cap_request, + capability, + principal, + justification=justification, + ) + except PolicyDenied as exc: + evaluated_denied = True + evaluated_reason = exc.reason_code + else: + evaluated_denied = not decision.allowed + evaluated_reason = decision.reason_code if evaluated_denied else None + + assert explanation.denied is denied + assert evaluated_denied is denied + assert explanation.denied == evaluated_denied + assert explanation.reason_code == reason_code + assert evaluated_reason == reason_code + + +def _limiter_state(engine: DefaultPolicyEngine) -> dict[str, list[float]]: + return { + key: list(entry.timestamps) + for key, entry in engine._limiter._windows.items() # noqa: SLF001 - invariant test + } + + +def test_explain_rate_limit_path_is_strictly_read_only() -> None: + now = [100.0] + engine = DefaultPolicyEngine( + rate_limits={SafetyClass.READ: (1, 60.0)}, + clock=lambda: now[0], + ) + request = _request() + capability = _cap(SafetyClass.READ) + principal = Principal(principal_id="rate-user") + + first = engine.evaluate(request, capability, principal, justification="") + assert first.allowed is True + before = deepcopy(_limiter_state(engine)) + + explanation = engine.explain(request, capability, principal, justification="") + + assert explanation.denied is True + assert explanation.reason_code == str(DenialReason.RATE_LIMITED) + assert _limiter_state(engine) == before + with pytest.raises(PolicyDenied) as excinfo: + engine.evaluate(request, capability, principal, justification="") + assert excinfo.value.reason_code == str(DenialReason.RATE_LIMITED) + + +def test_explain_does_not_prune_expired_rate_entries() -> None: + now = [100.0] + engine = DefaultPolicyEngine( + rate_limits={SafetyClass.READ: (1, 60.0)}, + clock=lambda: now[0], + ) + request = _request() + capability = _cap(SafetyClass.READ) + principal = Principal(principal_id="rate-user") + engine.evaluate(request, capability, principal, justification="") + now[0] = 161.0 + before = deepcopy(_limiter_state(engine)) + + explanation = engine.explain(request, capability, principal, justification="") + + assert explanation.denied is False + assert _limiter_state(engine) == before + assert engine.evaluate(request, capability, principal, justification="").allowed is True + + +def test_explain_collects_all_failures_while_evaluate_short_circuits() -> None: + engine = DefaultPolicyEngine() + request = _request(max_rows="bad") + capability = _cap(SafetyClass.WRITE, sensitivity=SensitivityTag.PII) + principal = Principal(principal_id="many-failures", roles=["reader"]) + + explanation = engine.explain(request, capability, principal, justification="short") + + assert explanation.denied is True + assert [failure.condition for failure in explanation.failed_conditions] == [ + "roles", + "min_justification", + "tenant_attribute", + "max_rows", + ] + with pytest.raises(PolicyDenied) as excinfo: + engine.evaluate(request, capability, principal, justification="short") + assert excinfo.value.reason_code == str(DenialReason.MISSING_ROLE)