Policy Reference

# Policy evaluation flow
from agent_os import KernelSpace, Policy

kernel = KernelSpace(policy="strict")

# When an agent tries to write a file:
# 1. Action intercepted: file_write("/etc/passwd", content)
# 2. Policy engine evaluates against patterns
# 3. Pattern match found: /etc/* is blocked
# 4. SIGKILL signal sent to agent
# 5. Action logged to FlightRecorder

from agent_os import KernelSpace

# Use strict policy (default)
kernel = KernelSpace(policy="strict")

from agent_os import KernelSpace

# Use permissive policy for development
kernel = KernelSpace(policy="permissive")

from agent_os import KernelSpace

# Use read-only policy for analysis tasks
kernel = KernelSpace(policy="read_only")

from agent_os import Policy, Pattern
from agent_os.policy import Severity, ViolationMode

# Create a custom policy
policy = Policy(
    name="my_custom_policy",
    description="Custom policy for my application",
    version="1.0.0",
    default_action="allow",      # allow, deny, or warn
    violation_mode=ViolationMode.SIGKILL,
    patterns=[]                   # List of Pattern objects
)

# Add a pattern to the policy
policy.add_pattern(Pattern(
    action_type="file_write",
    pattern=r"/var/log/.*",
    action="allow"
))

# Remove a pattern by ID
policy.remove_pattern("pattern_id")

# Evaluate an action against the policy
result = policy.evaluate(
    action_type="file_write",
    target="/var/log/app.log",
    context={"agent_id": "agent_123"}
)

# Check if action is allowed
if result.allowed:
    print("Action permitted")
else:
    print(f"Action blocked: {result.reason}")

# Export policy to YAML
yaml_str = policy.to_yaml()

# Load policy from YAML file
policy = Policy.from_yaml("path/to/policy.yaml")

# Merge with another policy
merged = policy.merge(other_policy, strategy="strict")

from agent_os import Pattern
from agent_os.policy import Severity

# Create a pattern to block SSH key access
ssh_pattern = Pattern(
    id="block_ssh_keys",
    action_type="file_read",
    pattern=r".*\.ssh/.*",
    action="deny",
    severity=Severity.CRITICAL,
    message="Access to SSH keys is not permitted",
    metadata={
        "category": "security",
        "compliance": ["SOC2", "HIPAA"]
    }
)

# Create a pattern with conditions
conditional_pattern = Pattern(
    id="limit_large_writes",
    action_type="file_write",
    pattern=r".*",
    action="warn",
    severity=Severity.MEDIUM,
    conditions={
        "max_size_bytes": 10485760,  # 10MB
        "require_confirmation": True
    }
)

# Create a network whitelist pattern
network_pattern = Pattern(
    id="allowed_apis",
    action_type="network_request",
    pattern=r"https://(api\.github\.com|api\.openai\.com)/.*",
    action="allow",
    severity=Severity.LOW
)

from agent_os import KernelSpace, Policy, Pattern
from agent_os.policy import Severity, ViolationMode

# Create a custom policy for a data processing application
data_policy = Policy(
    name="data_processor_policy",
    description="Policy for data processing agents",
    version="2.1.0",
    default_action="deny",
    violation_mode=ViolationMode.SIGKILL
)

# Allow reading from data directories
data_policy.add_pattern(Pattern(
    id="read_data_dirs",
    action_type="file_read",
    pattern=r"/data/(input|staging|archive)/.*",
    action="allow",
    severity=Severity.LOW
))

# Allow writing to output directory only
data_policy.add_pattern(Pattern(
    id="write_output",
    action_type="file_write",
    pattern=r"/data/output/.*",
    action="allow",
    severity=Severity.LOW
))

# Block all other file writes
data_policy.add_pattern(Pattern(
    id="block_other_writes",
    action_type="file_write",
    pattern=r".*",
    action="deny",
    severity=Severity.HIGH,
    message="File writes only allowed in /data/output/"
))

# Allow database reads
data_policy.add_pattern(Pattern(
    id="db_read",
    action_type="database_read",
    pattern=r".*",
    action="allow"
))

# Block database writes with warning
data_policy.add_pattern(Pattern(
    id="db_write_warn",
    action_type="database_write",
    pattern=r".*",
    action="warn",
    severity=Severity.MEDIUM,
    message="Database write detected - ensure data validation"
))

# Block dangerous shell commands
data_policy.add_pattern(Pattern(
    id="block_dangerous_shell",
    action_type="shell_execute",
    pattern=r".*(rm\s+-rf|sudo|chmod\s+777|mkfs|dd\s+if=).*",
    action="deny",
    severity=Severity.CRITICAL,
    message="Dangerous shell command blocked"
))

# Allow specific network endpoints
data_policy.add_pattern(Pattern(
    id="allowed_endpoints",
    action_type="network_request",
    pattern=r"https://(api\.internal\.com|storage\.cloud\.com)/.*",
    action="allow"
))

# Use the policy
kernel = KernelSpace(policy=data_policy)

# policy.yaml - Complete policy configuration example

name: enterprise_security_policy
description: Security policy for enterprise deployments
version: "3.0.0"
author: security-team@company.com
created: "2024-01-15"
updated: "2024-06-20"

# Default behavior for unmatched actions
defaults:
  action: deny
  violation_mode: sigkill
  log_level: info

# Global settings
settings:
  max_file_size_mb: 100
  max_network_timeout_seconds: 30
  enable_audit_logging: true
  require_encryption: true

# Pattern definitions
patterns:
  # File system patterns
  - id: allow_workspace_read
    action_type: file_read
    pattern: "^/workspace/.*"
    action: allow
    severity: low
    description: "Allow reading files in workspace"
    
  - id: allow_workspace_write
    action_type: file_write
    pattern: "^/workspace/.*"
    action: allow
    severity: low
    conditions:
      max_size_bytes: 104857600  # 100MB
    
  - id: block_sensitive_files
    action_type: file_read
    pattern: ".*\\.(pem|key|env|credentials|secret)$"
    action: deny
    severity: critical
    message: "Access to sensitive files is prohibited"
    metadata:
      compliance:
        - SOC2
        - HIPAA
        - PCI-DSS
        
  - id: block_system_dirs
    action_type: file_write
    pattern: "^/(etc|usr|bin|sbin|lib|boot)/.*"
    action: deny
    severity: critical
    message: "Writing to system directories is prohibited"

  # Database patterns
  - id: allow_select_queries
    action_type: database_read
    pattern: ".*"
    action: allow
    severity: low
    
  - id: block_ddl_operations
    action_type: database_execute
    pattern: ".*(DROP|TRUNCATE|ALTER|CREATE).*"
    action: deny
    severity: critical
    message: "DDL operations require admin approval"
    
  - id: warn_bulk_updates
    action_type: database_write
    pattern: ".*UPDATE.*WHERE.*"
    action: warn
    severity: medium
    conditions:
      require_where_clause: true
      max_affected_rows: 1000

  # Shell execution patterns
  - id: allow_safe_commands
    action_type: shell_execute
    pattern: "^(ls|cat|grep|find|head|tail|wc|sort|uniq|awk|sed)\\s.*"
    action: allow
    severity: low
    
  - id: block_destructive_commands
    action_type: shell_execute
    pattern: ".*(rm\\s+-rf|sudo|chmod\\s+777|chown|mkfs|fdisk|dd\\s+if=).*"
    action: deny
    severity: critical
    message: "Destructive shell command blocked"
    
  - id: block_network_tools
    action_type: shell_execute
    pattern: ".*(curl|wget|nc|netcat|ssh|scp|rsync).*"
    action: deny
    severity: high
    message: "Network tools must use approved APIs"

  # Network patterns
  - id: allow_internal_apis
    action_type: network_request
    pattern: "^https://.*\\.internal\\.company\\.com/.*"
    action: allow
    severity: low
    
  - id: allow_approved_external
    action_type: network_request
    pattern: "^https://(api\\.github\\.com|api\\.openai\\.com|storage\\.googleapis\\.com)/.*"
    action: allow
    severity: low
    conditions:
      allowed_methods:
        - GET
        - POST
      require_auth_header: true
      
  - id: block_plaintext_http
    action_type: network_request
    pattern: "^http://.*"
    action: deny
    severity: high
    message: "Plaintext HTTP connections are not allowed"
    
  - id: block_other_networks
    action_type: network_request
    pattern: ".*"
    action: deny
    severity: medium
    message: "External network access requires approval"

# Inherited policies (optional)
extends:
  - base_security_policy
  - compliance_policy

# Exception rules (override patterns for specific agents)
exceptions:
  - agent_pattern: "admin_agent_.*"
    patterns:
      - block_system_dirs
      - block_ddl_operations
    reason: "Admin agents have elevated privileges"
    expires: "2024-12-31"
    approved_by: security@company.com

from agent_os import KernelSpace, Policy

# Load policy from YAML file
policy = Policy.from_yaml("policies/enterprise_security.yaml")

# Load policy from YAML string
yaml_content = """
name: quick_policy
default_action: deny
patterns:
  - action_type: file_read
    pattern: ".*"
    action: allow
"""
policy = Policy.from_yaml_string(yaml_content)

# Use the loaded policy
kernel = KernelSpace(policy=policy)

# Validate a YAML policy file
errors = Policy.validate_yaml("policies/my_policy.yaml")
if errors:
    for error in errors:
        print(f"Validation error: {error}")

# Match exact path
pattern: "^/var/log/app\\.log$"

# Match all files in a directory
pattern: "^/data/input/.*"

# Match file extensions
pattern: ".*\\.(txt|csv|json)$"

# Match sensitive file types
pattern: ".*\\.(pem|key|env|secret|password)$"

# Match paths with specific segments
pattern: ".*/config/.*\\.yaml$"

# Match IP addresses (network patterns)
pattern: "^https://192\\.168\\..*"

# Match with negation (use in combination)
pattern: "^(?!.*secret).*\\.txt$"  # .txt files not containing "secret"

# Match SQL patterns
pattern: ".*SELECT.*FROM\\s+users.*"

# Match shell commands with arguments
pattern: "^git\\s+(push|pull|commit).*"

from agent_os import Policy
from agent_os.policy import ViolationMode

# Global violation mode for the policy
policy = Policy(
    name="strict_policy",
    violation_mode=ViolationMode.SIGKILL
)

# Per-pattern violation mode override
from agent_os import Pattern

# Critical violations terminate immediately
critical_pattern = Pattern(
    action_type="shell_execute",
    pattern=r".*sudo.*",
    action="deny",
    violation_mode=ViolationMode.SIGKILL
)

# Medium violations pause for review
review_pattern = Pattern(
    action_type="database_delete",
    pattern=r".*",
    action="deny",
    violation_mode=ViolationMode.SIGSTOP
)

# Low severity just warns
warn_pattern = Pattern(
    action_type="file_write",
    pattern=r"/tmp/.*",
    action="warn",
    violation_mode=ViolationMode.SIGWARN
)

from agent_os import KernelSpace, Policy
from agent_os.signals import AgentSignal

# Create policy with custom violation handler
def custom_violation_handler(violation):
    """Handle policy violations with custom logic."""
    print(f"Violation detected: {violation.pattern_id}")
    print(f"Action: {violation.action_type}")
    print(f"Target: {violation.target}")
    print(f"Severity: {violation.severity}")
    
    # Send alert for critical violations
    if violation.severity == "critical":
        send_security_alert(violation)
    
    # Log to external system
    log_to_siem(violation)
    
    # Return the signal to send
    if violation.severity == "critical":
        return AgentSignal.SIGKILL
    elif violation.severity == "high":
        return AgentSignal.SIGSTOP
    else:
        return AgentSignal.SIGWARN

policy = Policy(
    name="custom_handler_policy",
    violation_handler=custom_violation_handler
)

kernel = KernelSpace(policy=policy)

from agent_os import KernelSpace
from agent_os.templates import hipaa_standard

# Load the HIPAA template
policy = hipaa_standard()

# Customize for your environment
policy.add_pattern(Pattern(
    action_type="database_read",
    pattern=r".*patient_records.*",
    action="allow",
    conditions={
        "require_audit_log": True,
        "require_encryption": True
    }
))

kernel = KernelSpace(policy=policy)

from agent_os import KernelSpace
from agent_os.templates import soc2_compliant

# Load the SOC2 template
policy = soc2_compliant()

# Enable specific trust service criteria
policy.enable_criteria([
    "security",
    "availability", 
    "confidentiality"
])

kernel = KernelSpace(policy=policy)

from agent_os import KernelSpace
from agent_os.templates import k12_safe

# Load the K-12 safety template
policy = k12_safe()

# Configure age-appropriate settings
policy.configure(
    age_group="elementary",  # elementary, middle, high
    allow_social_media=False,
    allow_external_search=False,
    content_filter_level="strict"
)

kernel = KernelSpace(policy=policy)

from agent_os import Policy
from agent_os.templates import hipaa_standard, soc2_compliant

# Load base policies
hipaa = hipaa_standard()
soc2 = soc2_compliant()

# Strict merge: most restrictive rule wins
strict_policy = hipaa.merge(soc2, strategy="strict")

# Permissive merge: least restrictive rule wins
permissive_policy = hipaa.merge(soc2, strategy="permissive")

# Override merge: second policy overrides conflicts
override_policy = hipaa.merge(soc2, strategy="override")

# Custom merge with conflict resolution
def resolve_conflict(pattern1, pattern2):
    """Custom conflict resolution logic."""
    # Prefer higher severity patterns
    if pattern1.severity > pattern2.severity:
        return pattern1
    return pattern2

custom_policy = hipaa.merge(
    soc2, 
    strategy="custom",
    conflict_resolver=resolve_conflict
)

from agent_os import Policy, Pattern

# Create a base policy
base_policy = Policy(
    name="base_security",
    default_action="deny",
    patterns=[
        Pattern(
            action_type="file_read",
            pattern=r".*",
            action="allow"
        )
    ]
)

# Create a child policy that inherits from base
child_policy = Policy(
    name="app_specific",
    extends=base_policy,  # Inherit all patterns
    patterns=[
        # Add additional patterns
        Pattern(
            action_type="file_write",
            pattern=r"/app/data/.*",
            action="allow"
        )
    ]
)

# Child inherits all base patterns plus its own

from agent_os import KernelSpace, PolicyStack

# Create a layered policy stack
# Policies are evaluated in order; first match wins
stack = PolicyStack([
    # Layer 1: Agent-specific overrides (highest priority)
    Policy.from_yaml("policies/agent_overrides.yaml"),
    
    # Layer 2: Application-specific rules
    Policy.from_yaml("policies/app_rules.yaml"),
    
    # Layer 3: Compliance requirements
    hipaa_standard(),
    
    # Layer 4: Organization baseline (lowest priority)
    Policy.from_yaml("policies/org_baseline.yaml")
])

kernel = KernelSpace(policy=stack)

from agent_os import KernelSpace, Policy, Pattern

kernel = KernelSpace(policy="strict")

# Add a pattern at runtime
kernel.policy.add_pattern(Pattern(
    id="emergency_block",
    action_type="network_request",
    pattern=r".*malicious-domain\.com.*",
    action="deny",
    severity="critical"
))

# Remove a pattern
kernel.policy.remove_pattern("emergency_block")

# Temporarily elevate privileges for an agent
with kernel.policy.elevated_privileges("admin_agent"):
    # Agent has expanded permissions within this block
    result = kernel.execute("admin_agent", admin_task)

# Enable/disable patterns
kernel.policy.disable_pattern("some_pattern_id")
kernel.policy.enable_pattern("some_pattern_id")

# Hot-reload policy from file
kernel.reload_policy("policies/updated_policy.yaml")

# Subscribe to policy changes
def on_policy_change(event):
    print(f"Policy changed: {event.change_type}")
    print(f"Pattern: {event.pattern_id}")

kernel.policy.on_change(on_policy_change)

from agent_os import Policy, Pattern
from datetime import datetime, time

# Time-based policy adjustments
def business_hours_policy():
    """More permissive during business hours."""
    current_hour = datetime.now().hour
    if 9 <= current_hour < 17:
        return Policy(name="business_hours", default_action="allow")
    else:
        return Policy(name="after_hours", default_action="deny")

# Context-aware patterns
pattern = Pattern(
    action_type="database_write",
    pattern=r".*",
    action="allow",
    conditions={
        "time_range": {"start": "09:00", "end": "17:00"},
        "days": ["monday", "tuesday", "wednesday", "thursday", "friday"],
        "require_mfa": True,
        "max_operations_per_minute": 100
    }
)

# Environment-based policies
import os

if os.environ.get("ENVIRONMENT") == "production":
    policy = Policy.from_yaml("policies/production.yaml")
else:
    policy = Policy.from_yaml("policies/development.yaml")