Skip to main content

Overview

The Policy Engine has two logical components:
  • Policy Decision Point (PDP): Evaluates actions, returns decisions
  • Policy Enforcement Point (PEP): Implements decisions

Policy Decision Point

The PDP evaluates an action and returns a decision:
@dataclass
class Decision:
    result: str
    policy_id: str
    reason: str
    risk_level: str  # Added from rule
    confidence: float # Calculated during match
    modifications: dict | None = None
    approvers: list[str] | None = None

class PolicyDecisionPoint:
    def __init__(self, policy_store: PolicyStore):
        self.policies = policy_store
    
    def evaluate(self, action: Action) -> Decision:
        for rule in self.policies.get_rules():
            # Calculate match confidence (e.g., semantic similarity or regex strength)
            match_score = self.calculate_confidence(rule, action)
            
            if match_score > rule.threshold:
                return Decision(
                    result=rule.action,
                    policy_id=rule.id,
                    reason=rule.reason,
                    risk_level=rule.risk_level, 
                    confidence=match_score,
                    modifications=rule.modifications,
                    approvers=rule.approvers
                )
        
        # Default allow if no rules match
        return Decision(result="ALLOW", confidence=1.0, risk_level="LOW")

Decision Types

DecisionBehavior
ALLOWForward action to tool
DENYBlock action, return error
MODIFYRewrite parameters, then forward
STEP_UPRequire human approval

Policy Enforcement Point

The PEP implements the decision:
class PolicyEnforcementPoint:
    def enforce(self, action: Action, decision: Decision) -> Result:
        match decision.result:
            case "DENY":
                raise PolicyDenied(decision.reason)
            
            case "MODIFY":
                action.parameters = decision.apply_modifications(
                    action.parameters
                )
                return self.execute(action)
            
            case "STEP_UP":
                approval = self.approval_service.request(
                    action, 
                    decision.approvers
                )
                if not approval.granted:
                    raise ApprovalDenied(approval.reason)
                return self.execute(action)
            
            case "ALLOW":
                return self.execute(action)

Policy Syntax

rules:
  - id: block-pii-external
    name: Block PII to external recipients
    match:
      tool: email
      operation: send
      parameters:
        to: { external: true }
      context:
        data_classification: PII
    action: DENY
    risk_level: HIGH  # Quantifies the "danger" of the tool/data
    threshold: 0.7    # Minimum confidence required to trigger this rule
    approvers: ["data-privacy-team"]
    reason: "Cannot send PII externally"

Match Conditions

FieldDescriptionExample
toolTool nameemail, database
operationOperation typesend, query, delete
parametersParameter constraints{ to: { external: true } }
contextSession context{ data_classification: PII }
risk_signalsComputed scores{ injection_score: { gt: 0.8 } }

Operators

OperatorMeaning
{ eq: value }Equals
{ gt: value }Greater than
{ lt: value }Less than
{ contains: value }Array contains
{ matches: regex }Regex match
{ external: true }External destination

Policy Loading

Policies can be loaded from files or remote service:
class PolicyStore:
    def __init__(self, source: str):
        if source.startswith("http"):
            self.loader = RemotePolicyLoader(source)
        else:
            self.loader = FilePolicyLoader(source)
    
    def get_rules(self) -> list[Rule]:
        return self.loader.load()
    
    def reload(self):
        self.loader.reload()
Support hot-reload for policy updates without restart.

Requirements

RequirementLevel
Evaluate before executionMUST
Support ALLOW/DENY/MODIFY/STEP_UPMUST
Parameter validationMUST
Context-aware matchingSHOULD
Hot reload policiesSHOULD