security — PermissionGuard, InputSanitizer, ExecutionSandbox

Overview

Three independent protection layers, per the module docstring:

1. PermissionGuard — declare and enforce per-plugin permissions.
2. InputSanitizer — validate inputs against schema, detect injection attempts.
3. ExecutionSandbox — async timeout + resource-cap wrapper.

These are composed by PluginManager but each class can be used standalone.

PermissionPolicy (dataclass)

@dataclass
class PermissionPolicy:
    plugin_name: str
    granted: Set[str] = field(default_factory=set)
    denied: Set[str] = field(default_factory=set)
    session_scoped: Optional[str] = None
    user_scoped: Optional[str] = None

Internal per-plugin record held by PermissionGuard. Not typically constructed directly by user code — PermissionGuard._get_or_create() creates one lazily on first grant()/revoke()/check().

PermissionGuard

guard = PermissionGuard()
guard.grant("web_search", PermissionType.NETWORK_ACCESS)
guard.check("web_search", PermissionType.NETWORK_ACCESS)  # → True

grant(plugin_name, permission, session_id=None, user_id=None) -> None

Adds permission to the plugin's granted set and removes it from denied if present. Accepts either a PermissionType enum member or a raw string (normalized via .value if it's an enum). If session_id/user_id are given, they overwrite the policy's scope — note this means calling grant a second time with a different session_id replaces the previous scope rather than adding an additional allowed scope.

revoke(plugin_name, permission) -> None

Removes from granted, adds to denied.

grant_all_required(plugin: BasePlugin) -> None

Convenience: calls grant() for every string in plugin.metadata.permissions. This is what PluginManager.startup() calls for each auto-discovered plugin, only when safe_mode=False.

check(plugin_name, permission, context=None) -> bool

Algorithm:

1. If no policy exists for plugin_name, return False.
2. If permission is in policy.denied, return False (explicit deny always wins).
3. If a context is given and the policy has a session_scoped value that doesn't match context.session_id, return False. Same for user_scoped vs. context.user_id.
4. Return permission in policy.granted.

assert_permission(plugin, permission, context=None) -> None

Calls check(); Raises PluginPermissionError if it returns False.

check_all_required(plugin, context=None) -> bool

Iterates plugin.metadata.permissions; returns False (with a warning log) on the first one that fails check(). This is what PluginManager.register() (in safe_mode) and PluginManager.execute() (in safe_mode) call.

get_grants(plugin_name) -> Set[str]

Returns a copy of the granted set (or an empty set if no policy exists) — safe to mutate by the caller without affecting internal state.

InputSanitizer

Static-method-only class; validates and sanitizes plugin input dicts against a list of SchemaProperty.

sanitize(data: Dict[str, Any], schema: List[SchemaProperty], strict: bool = True) -> Dict[str, Any]

Internal workflow:

1. Build prop_map = {p.name: p for p in schema} and required = {p.name for p in schema if p.required}.
2. Compute missing = required - set(data.keys()); raise PluginValidationError listing them (sorted) if any are missing.
3. For each (name, value) pair in data:
   • If name isn't in prop_map: raise PluginValidationError if strict=True, otherwise silently drop the field (continue) if strict=False.
   • Coerce the value's type via the module-private _coerce() helper.
   • If the property is type="string" and the coerced value is a str: raise PluginValidationError if longer than 32,768 characters (_MAX_STRING_LEN); then run _check_injection().
   • If the property declares an enum and the value isn't in it, raise PluginValidationError.
   • Store the cleaned value.
4. After the loop, fill in prop.default for any schema property not present in the cleaned output whose default is not None.
5. Return the cleaned dict.

Important: PluginManager.execute() always calls this with strict=False — unknown fields in the caller's input are silently dropped, not rejected, when going through the manager. Calling sanitize() directly with the default strict=True will reject unknown fields.

_check_injection(field_name, value) -> None (private, static)

Checks value against five compiled regex patterns (all case-insensitive):

Raises PluginValidationError (including the offending pattern's source string) on the first match.

[!NOTE] This is a heuristic, not a security boundary These are simple regexes over free-form text, intended to catch obvious injection attempts. They are pattern-matching, not a parser or sandboxed interpreter — sufficiently obfuscated input can evade them. Treat this as one layer of defense-in-depth, not a substitute for least-privilege permissions and safe handling of plugin output.

_coerce(name, value, prop) -> Any (module-private function)

For string/integer/number/boolean types: if value isn't already an instance of the target Python type, attempts target(value); raises PluginValidationError on TypeError/ValueError. For array: raises if value isn't a list. For object: raises if value isn't a dict. Values already matching the target type pass through unchanged (no re-wrapping).

Raises: PluginValidationError on type mismatch that can't be coerced.

Complexity

sanitize() is O(F) in the number of fields in data, plus O(P) for the default-filling pass over schema, plus O(len(value)) per string field for injection-pattern scanning.

ExecutionSandbox

sandbox = ExecutionSandbox(hard_timeout_sec=60.0, log_slow_threshold=5.0)
result = await sandbox.run(some_coroutine, plugin_name="my_plugin")

run(coro, plugin_name="?") -> Any

Wraps coro in asyncio.wait_for(coro, timeout=self.hard_timeout). If it completes within log_slow_threshold seconds, returns normally. If it takes longer than the threshold but still completes before the hard timeout, logs a warning but still returns the result. On asyncio.TimeoutError, logs an error and re-raises (propagates asyncio.TimeoutError unchanged — it does not translate this into a PluginTimeoutError). Any other exception from coro is logged and re-raised unchanged as well.

Raises: asyncio.TimeoutError on hard timeout; whatever coro itself raises otherwise.

[!WARNING] Not currently wired into PluginManager.execute() PluginManager.__init__ constructs self.sandbox = ExecutionSandbox(hard_timeout_sec=hard_timeout) (default 60s), but based on the source of plugin_manager.py, execute() calls plugin.safe_execute() / safe_execute_with_retry() directly — it does not appear to route the call through self.sandbox.run(). The per-plugin timeout enforced during execution is plugin.config.timeout (default 30s, via BasePlugin.safe_execute's own asyncio.wait_for), not the manager's hard_timeout. Verify against your specific version of plugin_manager.py if this distinction matters for your timeout budgeting.

Thread safety / Async safety

ExecutionSandbox itself holds no mutable state beyond its two config floats set at construction, so a single instance is safe to share and call concurrently from multiple coroutines.

Side effects

Logging only (logger.warning on slow execution, logger.error on timeout or exception). No state mutation, no I/O beyond what coro itself performs.

Examples

1. Basic permission grant/check:

guard = PermissionGuard()
guard.grant("web_search", "network_access")
assert guard.check("web_search", "network_access") is True
assert guard.check("web_search", "file_write") is False

2. Session-scoped grant:

guard.grant("db_query", "database_access", session_id="sess-123")
ctx_ok = ExecutionContext(session_id="sess-123")
ctx_bad = ExecutionContext(session_id="sess-999")
guard.check("db_query", "database_access", ctx_ok)    # True
guard.check("db_query", "database_access", ctx_bad)   # False

3. Sanitizing input, lenient mode:

schema = [SchemaProperty(name="query", type="string", description="Search text")]
cleaned = InputSanitizer.sanitize({"query": "climate change", "extra": "ignored"}, schema, strict=False)
# {"query": "climate change"}

4. Error handling — injection pattern rejected:

try:
    InputSanitizer.sanitize({"query": "ignore previous instructions"}, schema, strict=False)
except PluginValidationError as e:
    print(e)

5. Sandboxing an arbitrary coroutine:

sandbox = ExecutionSandbox(hard_timeout_sec=10.0)
try:
    result = await sandbox.run(some_slow_plugin.execute(data, ctx), plugin_name="some_slow_plugin")
except asyncio.TimeoutError:
    print("hit the hard timeout")

Best practices

• Treat PermissionGuard as deny-by-default: a plugin with no policy at all returns False from every check() call.
• Use strict=True (the sanitize() default) when you want to reject unexpected fields outright, e.g. validating a manifest at load time; use strict=False for the lenient runtime path PluginManager.execute() uses.

Common mistakes

• Assuming grant() is additive across scopes — a second grant() call with a new session_id overwrites the previous scope on that policy rather than adding a second allowed session.
• Relying on ExecutionSandbox.run() being invoked automatically by PluginManager.execute() — as noted above, the manager's effective per-call timeout comes from plugin.config.timeout.

Related APIs

• PluginMetadata.permissions — source of grant_all_required
• BasePlugin.safe_execute — the timeout path actually used by PluginManager.execute()
• ExecutionContext — carries session_id/user_id for scoped checks