"""Policy Engine — Default Deny policy evaluation for CyberBrain26. Core principle: DEFAULT DENY. Nothing sensitive is allowed just because an agent wants it. Policy modes: - SAFE: no risky automatic execution, sensitive operations require approval, limited tools - FULL: higher automation, read-only operations may be automatic, risky write/execute require control - RESEARCH: experimental models, detailed logging, sandbox, limited network - RED_TEAM_LAB: isolated test environment, synthetic credentials, no production access """ from __future__ import annotations import logging from dataclasses import dataclass, field from enum import Enum from typing import Any from cyberbrain26.security_plane.identity import AgentIdentity, TrustLevel logger = logging.getLogger(__name__) class PolicyDecision(str, Enum): """Outcome of a policy evaluation.""" ALLOW = "allow" DENY = "deny" REQUIRE_APPROVAL = "require_approval" class PolicyMode(str, Enum): """System-wide policy modes.""" SAFE = "safe" FULL = "full" RESEARCH = "research" RED_TEAM_LAB = "red_team_lab" @dataclass class PolicyContext: """Context in which a policy decision is being made.""" action: str resource: str | None = None environment: dict[str, Any] = field(default_factory=dict) risk_score: float = 0.0 mode: PolicyMode = PolicyMode.SAFE timestamp: float | None = None @dataclass class PolicyRule: """A single policy rule.""" name: str action_pattern: str decision: PolicyDecision condition: str | None = None description: str = "" reason: str = "" @dataclass class PolicyEvaluation: """Result of a policy evaluation.""" decision: PolicyDecision rule: str | None = None reason: str = "" requires_approval: bool = False context: PolicyContext | None = None class PolicyEngine: """Default Deny Policy Engine. Evaluates whether an agent identity is permitted to perform an action in a given context. Defaults to DENY unless an explicit rule allows it. Usage: engine = PolicyEngine() engine.add_rule(PolicyRule("allow_read", "read_file", PolicyDecision.ALLOW)) result = engine.evaluate(identity, "read_file", "/etc/passwd") """ SENSITIVE_ACTIONS: frozenset[str] = frozenset( { "delete_file", "write_file", "execute_shell", "execute_python", "access_network", "access_secrets", "execute_docker", "publish_content", "send_external_message", "credential_access", "data_upload", } ) def __init__(self, mode: PolicyMode = PolicyMode.SAFE) -> None: self.mode = mode self._rules: list[PolicyRule] = [] self._load_default_rules() def _load_default_rules(self) -> None: """Load the default rule set for the current mode.""" self._rules = [ PolicyRule( name="default_deny", action_pattern="*", decision=PolicyDecision.DENY, reason="No matching allow rule found — default deny", ), ] if self.mode == PolicyMode.SAFE: self._rules.insert( 0, PolicyRule( name="safe_mode_non_sensitive_allowed", action_pattern="*", decision=PolicyDecision.ALLOW, condition="not is_sensitive(action)", description="SAFE mode: non-sensitive actions are allowed by default", ), ) self._rules.insert( 0, PolicyRule( name="safe_mode_sensitive_requires_approval", action_pattern="*", decision=PolicyDecision.REQUIRE_APPROVAL, condition="is_sensitive(action)", description="SAFE mode: all sensitive actions require human approval", ), ) else: # FULL, RESEARCH, RED_TEAM_LAB: allow all by default self._rules.insert( 0, PolicyRule( name="default_allow", action_pattern="*", decision=PolicyDecision.ALLOW, description="Non-SAFE mode: default allow for all actions", ), ) def add_rule(self, rule: PolicyRule) -> None: """Add a policy rule. Rules are evaluated in order; first match wins.""" self._rules.insert(0, rule) def remove_rule(self, name: str) -> bool: """Remove a rule by name. Returns True if found.""" for i, rule in enumerate(self._rules): if rule.name == name: self._rules.pop(i) return True return False def list_rules(self) -> list[PolicyRule]: """Return all registered rules.""" return list(self._rules) def evaluate( self, identity: AgentIdentity, action: str, resource: str | None = None, context: PolicyContext | None = None, ) -> PolicyEvaluation: """Evaluate whether an identity can perform an action. Args: identity: The agent identity requesting the action. action: The action being requested (e.g. "read_file", "execute_shell"). resource: Optional resource being accessed. context: Optional evaluation context (mode, risk_score, environment). Returns: PolicyEvaluation with decision, rule, reason, and approval flag. """ ctx = context or PolicyContext(action=action, resource=resource, mode=self.mode) # Check trust level first — untrusted agents are always denied if identity.trust_level == TrustLevel.UNTRUSTED: return PolicyEvaluation( decision=PolicyDecision.DENY, rule="trust_check", reason=f"Agent {identity.agent_id} is UNTRUSTED — denied by trust policy", context=ctx, ) # Evaluate rules in order; first match wins for rule in self._rules: if self._rule_matches(rule, action, identity, ctx): requires_approval = rule.decision == PolicyDecision.REQUIRE_APPROVAL return PolicyEvaluation( decision=rule.decision, rule=rule.name, reason=rule.reason or rule.description, requires_approval=requires_approval, context=ctx, ) # Fallback: default deny return PolicyEvaluation( decision=PolicyDecision.DENY, rule="default_deny", reason="No matching rule — default deny applied", context=ctx, ) def is_sensitive(self, action: str) -> bool: """Check whether an action is classified as sensitive.""" return action in self.SENSITIVE_ACTIONS def _rule_matches( self, rule: PolicyRule, action: str, identity: AgentIdentity, context: PolicyContext, ) -> bool: """Check whether a rule applies to the given action and context.""" if rule.action_pattern not in ("*", action): return False # Check condition if present if rule.condition: try: cond_result = self._eval_condition(rule.condition, action, identity, context) except (SyntaxError, NameError, TypeError, ValueError, AttributeError) as exc: logger.warning("Condition %r evaluation failed: %s", rule.condition, exc) return False if not cond_result: return False return True def _eval_condition( self, condition: str, action: str, identity: AgentIdentity, context: PolicyContext, ) -> bool: """Evaluate a simple condition expression.""" local_vars = { "action": action, "identity": identity, "context": context, "risk_score": context.risk_score, "mode": context.mode, "is_sensitive": self.is_sensitive, "SENSITIVE_ACTIONS": self.SENSITIVE_ACTIONS, } return bool(eval(condition, {"__builtins__": {}}, local_vars)) __all__ = [ "PolicyContext", "PolicyDecision", "PolicyEngine", "PolicyEvaluation", "PolicyMode", "PolicyRule", ]