hermes-agent/tools/approval_tiers.py

"""
Approval Tier System — Graduated safety based on risk level

Extends approval.py with 5-tier system for command approval.

| Tier | Action          | Human | LLM | Timeout |
|------|-----------------|-------|-----|---------|
| 0    | Read, search    | No    | No  | N/A     |
| 1    | Write, scripts  | No    | Yes | N/A     |
| 2    | Messages, API   | Yes   | Yes | 60s     |
| 3    | Crypto, config  | Yes   | Yes | 30s     |
| 4    | Crisis          | Yes   | Yes | 10s     |

Issue: #670
"""

import re
from dataclasses import dataclass
from enum import IntEnum
from typing import Any, Dict, List, Optional, Tuple


class ApprovalTier(IntEnum):
    """Approval tiers based on risk level."""
    SAFE = 0      # Read, search — no approval needed
    LOW = 1       # Write, scripts — LLM approval
    MEDIUM = 2    # Messages, API — human + LLM, 60s timeout
    HIGH = 3      # Crypto, config — human + LLM, 30s timeout
    CRITICAL = 4  # Crisis — human + LLM, 10s timeout


# Tier metadata
TIER_INFO = {
    ApprovalTier.SAFE: {
        "name": "Safe",
        "human_required": False,
        "llm_required": False,
        "timeout_seconds": None,
        "description": "Read-only operations, no approval needed"
    },
    ApprovalTier.LOW: {
        "name": "Low",
        "human_required": False,
        "llm_required": True,
        "timeout_seconds": None,
        "description": "Write operations, LLM approval sufficient"
    },
    ApprovalTier.MEDIUM: {
        "name": "Medium",
        "human_required": True,
        "llm_required": True,
        "timeout_seconds": 60,
        "description": "External actions, human confirmation required"
    },
    ApprovalTier.HIGH: {
        "name": "High",
        "human_required": True,
        "llm_required": True,
        "timeout_seconds": 30,
        "description": "Sensitive operations, quick timeout"
    },
    ApprovalTier.CRITICAL: {
        "name": "Critical",
        "human_required": True,
        "llm_required": True,
        "timeout_seconds": 10,
        "description": "Crisis or dangerous operations, fastest timeout"
    },
}


# Action-to-tier mapping
ACTION_TIERS: Dict[str, ApprovalTier] = {
    # Tier 0: Safe (read-only)
    "read_file": ApprovalTier.SAFE,
    "search_files": ApprovalTier.SAFE,
    "web_search": ApprovalTier.SAFE,
    "session_search": ApprovalTier.SAFE,
    "list_files": ApprovalTier.SAFE,
    "get_file_content": ApprovalTier.SAFE,
    "memory_search": ApprovalTier.SAFE,
    "skills_list": ApprovalTier.SAFE,
    "skills_search": ApprovalTier.SAFE,
    
    # Tier 1: Low (write operations)
    "write_file": ApprovalTier.LOW,
    "create_file": ApprovalTier.LOW,
    "patch_file": ApprovalTier.LOW,
    "delete_file": ApprovalTier.LOW,
    "execute_code": ApprovalTier.LOW,
    "terminal": ApprovalTier.LOW,
    "run_script": ApprovalTier.LOW,
    "skill_install": ApprovalTier.LOW,
    
    # Tier 2: Medium (external actions)
    "send_message": ApprovalTier.MEDIUM,
    "web_fetch": ApprovalTier.MEDIUM,
    "browser_navigate": ApprovalTier.MEDIUM,
    "api_call": ApprovalTier.MEDIUM,
    "gitea_create_issue": ApprovalTier.MEDIUM,
    "gitea_create_pr": ApprovalTier.MEDIUM,
    "git_push": ApprovalTier.MEDIUM,
    "deploy": ApprovalTier.MEDIUM,
    
    # Tier 3: High (sensitive operations)
    "config_change": ApprovalTier.HIGH,
    "env_change": ApprovalTier.HIGH,
    "key_rotation": ApprovalTier.HIGH,
    "access_grant": ApprovalTier.HIGH,
    "permission_change": ApprovalTier.HIGH,
    "backup_restore": ApprovalTier.HIGH,
    
    # Tier 4: Critical (crisis/dangerous)
    "kill_process": ApprovalTier.CRITICAL,
    "rm_rf": ApprovalTier.CRITICAL,
    "format_disk": ApprovalTier.CRITICAL,
    "shutdown": ApprovalTier.CRITICAL,
    "crisis_override": ApprovalTier.CRITICAL,
}


# Dangerous command patterns (from existing approval.py)
_DANGEROUS_PATTERNS = [
    (r"rm\s+-rf\s+/", ApprovalTier.CRITICAL),
    (r"mkfs\.", ApprovalTier.CRITICAL),
    (r"dd\s+if=.*of=/dev/", ApprovalTier.CRITICAL),
    (r"shutdown|reboot|halt", ApprovalTier.CRITICAL),
    (r"chmod\s+777", ApprovalTier.HIGH),
    (r"curl.*\|\s*bash", ApprovalTier.HIGH),
    (r"wget.*\|\s*sh", ApprovalTier.HIGH),
    (r"eval\s*\(", ApprovalTier.HIGH),
    (r"sudo\s+", ApprovalTier.MEDIUM),
    (r"git\s+push.*--force", ApprovalTier.HIGH),
    (r"docker\s+rm.*-f", ApprovalTier.MEDIUM),
    (r"kubectl\s+delete", ApprovalTier.HIGH),
]


@dataclass
class ApprovalRequest:
    """A request for approval."""
    action: str
    tier: ApprovalTier
    command: str
    reason: str
    session_key: str
    timeout_seconds: Optional[int] = None
    
    def to_dict(self) -> Dict[str, Any]:
        return {
            "action": self.action,
            "tier": self.tier.value,
            "tier_name": TIER_INFO[self.tier]["name"],
            "command": self.command,
            "reason": self.reason,
            "session_key": self.session_key,
            "timeout": self.timeout_seconds,
            "human_required": TIER_INFO[self.tier]["human_required"],
            "llm_required": TIER_INFO[self.tier]["llm_required"],
        }


def detect_tier(action: str, command: str = "") -> ApprovalTier:
    """
    Detect the approval tier for an action.
    
    Checks action name first, then falls back to pattern matching.
    """
    # Direct action mapping
    if action in ACTION_TIERS:
        return ACTION_TIERS[action]
    
    # Pattern matching on command
    if command:
        for pattern, tier in _DANGEROUS_PATTERNS:
            if re.search(pattern, command, re.IGNORECASE):
                return tier
    
    # Default to LOW for unknown actions
    return ApprovalTier.LOW


def requires_human_approval(tier: ApprovalTier) -> bool:
    """Check if tier requires human approval."""
    return TIER_INFO[tier]["human_required"]


def requires_llm_approval(tier: ApprovalTier) -> bool:
    """Check if tier requires LLM approval."""
    return TIER_INFO[tier]["llm_required"]


def get_timeout(tier: ApprovalTier) -> Optional[int]:
    """Get timeout in seconds for a tier."""
    return TIER_INFO[tier]["timeout_seconds"]


def should_auto_approve(action: str, command: str = "") -> bool:
    """Check if action should be auto-approved (tier 0)."""
    tier = detect_tier(action, command)
    return tier == ApprovalTier.SAFE


def format_approval_prompt(request: ApprovalRequest) -> str:
    """Format an approval request for display."""
    info = TIER_INFO[request.tier]
    lines = []
    lines.append(f"⚠️ Approval Required (Tier {request.tier.value}: {info['name']})")
    lines.append(f"")
    lines.append(f"Action: {request.action}")
    lines.append(f"Command: {request.command[:100]}{'...' if len(request.command) > 100 else ''}")
    lines.append(f"Reason: {request.reason}")
    lines.append(f"")
    
    if info["human_required"]:
        lines.append(f"👤 Human approval required")
    if info["llm_required"]:
        lines.append(f"🤖 LLM approval required")
    if info["timeout_seconds"]:
        lines.append(f"⏱️ Timeout: {info['timeout_seconds']}s")
    
    return "\n".join(lines)


def create_approval_request(
    action: str,
    command: str,
    reason: str,
    session_key: str
) -> ApprovalRequest:
    """Create an approval request for an action."""
    tier = detect_tier(action, command)
    timeout = get_timeout(tier)
    
    return ApprovalRequest(
        action=action,
        tier=tier,
        command=command,
        reason=reason,
        session_key=session_key,
        timeout_seconds=timeout
    )


# Crisis bypass rules
CRISIS_BYPASS_ACTIONS = frozenset([
    "send_message",  # Always allow sending crisis resources
    "check_crisis",
    "notify_crisis",
])


def is_crisis_bypass(action: str, context: str = "") -> bool:
    """Check if action should bypass approval during crisis."""
    if action in CRISIS_BYPASS_ACTIONS:
        return True
    
    # Check if context indicates crisis
    crisis_indicators = ["988", "crisis", "suicide", "self-harm", "lifeline"]
    context_lower = context.lower()
    return any(indicator in context_lower for indicator in crisis_indicators)