Merge pull request #64 from dgenio/feat/rate-limiting-policy-engine

feat: add sliding-window rate limiting to DefaultPolicyEngine

Author: dgenio

CHANGELOG.md

@@ -8,6 +8,9 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 ## [Unreleased]
 
 ### Added
+- Sliding-window rate limiting in `DefaultPolicyEngine` per `(principal_id, capability_id)` pair (#39).
+  Default limits by safety class: 60 READ / 10 WRITE / 2 DESTRUCTIVE per 60s window.
+  Service-role principals get 10× limits. Configurable via constructor.
 - GitHub Release step in publish workflow — creates a release with auto-generated notes and artifacts before publishing to PyPI.
 
 ### Fixed

docs/security.md

@@ -35,4 +35,8 @@ Consider an agent that obtains a token for `billing.list_invoices` then passes i
 - The `AGENT_KERNEL_SECRET` must be kept secret. Rotate it if compromised.
 - The default `InMemoryDriver` has no persistence — suitable for testing only.
 - PII redaction is heuristic (regex-based). It is not a substitute for proper data governance.
-- There is no rate limiting or quota enforcement in v0.1.
+- Rate limiting is enforced per `(principal_id, capability_id)` pair using a sliding window.
+  Default limits: 60 READ / 10 WRITE / 2 DESTRUCTIVE invocations per 60-second window.
+  Principals with the `"service"` role receive 10× the default limits. Limits are
+  configurable via `DefaultPolicyEngine(rate_limits=...)`. There is no distributed or
+  persistent rate-limit state — limits reset on process restart.

src/agent_kernel/policy.py

@@ -3,10 +3,14 @@
 from __future__ import annotations
 
 import logging
+import time
+from collections import defaultdict
+from collections.abc import Callable
+from dataclasses import dataclass
 from typing import Any, Protocol
 
 from .enums import SafetyClass, SensitivityTag
-from .errors import PolicyDenied
+from .errors import AgentKernelError, PolicyDenied
 from .models import Capability, CapabilityRequest, PolicyDecision, Principal
 
 logger = logging.getLogger(__name__)
@@ -18,6 +22,66 @@
 _MAX_ROWS_USER = 50
 _MAX_ROWS_SERVICE = 500
 
+# Default rate limits per safety class: (invocations, window_seconds).
+_DEFAULT_RATE_LIMITS: dict[SafetyClass, tuple[int, float]] = {
+    SafetyClass.READ: (60, 60.0),
+    SafetyClass.WRITE: (10, 60.0),
+    SafetyClass.DESTRUCTIVE: (2, 60.0),
+}
+
+# Service role multiplier for rate limits.
+_SERVICE_RATE_MULTIPLIER = 10
+
+
+@dataclass(slots=True)
+class _RateEntry:
+    """Timestamps for a single rate-limit key."""
+
+    timestamps: list[float]
+
+
+class RateLimiter:
+    """Sliding-window rate limiter using monotonic clock.
+
+    Args:
+        clock: Callable returning the current time in seconds.
+            Defaults to :func:`time.monotonic`.
+    """
+
+    def __init__(self, clock: Callable[[], float] | None = None) -> None:
+        self._clock = clock or time.monotonic
+        self._windows: dict[str, _RateEntry] = defaultdict(lambda: _RateEntry(timestamps=[]))
+
+    def check(self, key: str, limit: int, window_seconds: float) -> bool:
+        """Return ``True`` if the next invocation would be within the limit.
+
+        Prunes expired timestamps as a side-effect.
+
+        Args:
+            key: Rate-limit key (e.g. ``"principal:capability"``).
+            limit: Maximum allowed invocations per window.
+            window_seconds: Sliding window duration in seconds.
+
+        Returns:
+            ``True`` if under limit, ``False`` if limit would be exceeded.
+        """
+        now = self._clock()
+        cutoff = now - window_seconds
+        entry = self._windows[key]
+        entry.timestamps = [t for t in entry.timestamps if t > cutoff]
+        if not entry.timestamps:
+            del self._windows[key]
+            return True
+        return len(entry.timestamps) < limit
+
+    def record(self, key: str) -> None:
+        """Record an invocation for *key*.
+
+        Args:
+            key: Rate-limit key.
+        """
+        self._windows[key].timestamps.append(self._clock())
+
 
 class PolicyEngine(Protocol):
     """Interface for a policy engine."""
@@ -61,8 +125,40 @@ class DefaultPolicyEngine:
        ``"secrets_reader"`` and a justification of at least 15 characters.
     6. **max_rows** — 50 for regular users; 500 for principals with the
        ``"service"`` role.
+    7. **Rate limiting** — sliding-window rate limit per
+       ``(principal_id, capability_id)`` pair, with defaults by safety class.
+       Principals with the ``"service"`` role get 10× the default limits.
     """
 
+    def __init__(
+        self,
+        *,
+        rate_limits: dict[SafetyClass, tuple[int, float]] | None = None,
+        clock: Callable[[], float] | None = None,
+    ) -> None:
+        """Initialise the policy engine.
+
+        Args:
+            rate_limits: Override default rate limits per safety class.
+                Each value is ``(max_invocations, window_seconds)``.
+                Partial overrides are merged into the defaults so that
+                unspecified safety classes retain their default limits.
+            clock: Monotonic clock callable for rate-limiter.
+                Defaults to :func:`time.monotonic`.
+        """
+        limits = dict(_DEFAULT_RATE_LIMITS)
+        if rate_limits is not None:
+            limits.update(rate_limits)
+        for sc, (count, window) in limits.items():
+            if count < 1 or window <= 0:
+                raise AgentKernelError(
+                    f"Invalid rate limit for {sc.value}: "
+                    f"limit must be >= 1 and window must be > 0, "
+                    f"got limit={count}, window={window}."
+                )
+        self._rate_limits = limits
+        self._limiter = RateLimiter(clock=clock)
+
     @staticmethod
     def _deny(reason: str, *, principal_id: str, capability_id: str) -> PolicyDenied:
         """Log a policy denial at WARNING and return the exception to raise."""
@@ -197,6 +293,22 @@ def evaluate(
         else:
             constraints["max_rows"] = max_rows
 
+        # ── 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):
+                raise self._deny(
+                    f"Rate limit exceeded: {limit} {capability.safety_class.value} "
+                    f"invocations per {window}s for principal '{pid}'",
+                    principal_id=pid,
+                    capability_id=cid,
+                )
+            self._limiter.record(rate_key)
+
         reason = "Request approved by DefaultPolicyEngine."
         logger.info(
             "policy_allowed",