SneakyCode/app/services/permissions.py

"""Permission gating for tool execution."""

from __future__ import annotations

import json
import logging
import re
import shlex
from collections.abc import Awaitable, Callable

from app.models.config import AgentMode, PermissionsConfig, ToolsConfig

logger = logging.getLogger(__name__)

# Type alias for the async prompt callback
PromptCallback = Callable[[str, str], Awaitable[bool]]

# Detect shell redirects that write to files (>, >>, heredocs)
_WRITE_REDIRECT_PATTERN = re.compile(r"(?:>\s*\S|>>|<<)")


class PermissionDenied(Exception):
    """Raised when a tool is denied execution by permissions policy."""


class PermissionsService:
    """Check whether a tool is allowed to execute based on config tiers.

    In TUI mode, set a prompt callback via set_prompt_callback() that
    shows a modal dialog. Without a callback, unlisted tools are denied.
    """

    READ_ONLY_TOOLS: frozenset[str] = frozenset({
        "read_file", "list_dir", "grep_files", "find_files", "finish",
    })

    def __init__(
        self,
        config: PermissionsConfig,
        tools_config: ToolsConfig | None = None,
    ) -> None:
        self.config = config
        self._tools_config = tools_config
        self._prompt_callback: PromptCallback | None = None
        self._mode: AgentMode = AgentMode.NORMAL

    @property
    def mode(self) -> AgentMode:
        """Current agent mode."""
        return self._mode

    @mode.setter
    def mode(self, value: AgentMode) -> None:
        self._mode = value

    def set_prompt_callback(self, callback: PromptCallback) -> None:
        """Set the async callback used to prompt the user for permission.

        Args:
            callback: Async function(tool_name, description) -> bool.
        """
        self._prompt_callback = callback

    async def check(
        self,
        tool_name: str,
        description: str = "",
        arguments: str = "",
    ) -> bool:
        """Check if a tool is permitted to run.

        Args:
            tool_name: Name of the tool to check.
            description: Human-readable description for the prompt.
            arguments: Raw JSON arguments string (used for shell-aware checks).

        Returns:
            True if permitted, False if denied.
        """
        if tool_name in self.config.deny:
            logger.info("Tool '%s' is in deny list — blocked", tool_name)
            return False

        if self._mode == AgentMode.AUTO:
            logger.debug("Tool '%s' auto-approved (AUTO mode)", tool_name)
            return True

        if self._mode == AgentMode.PLAN:
            if tool_name not in self.READ_ONLY_TOOLS:
                logger.info("Tool '%s' blocked in Plan mode (read-only tools only)", tool_name)
                return False
            return True

        if tool_name in self.config.auto_approve:
            logger.debug("Tool '%s' is auto-approved", tool_name)
            return True

        # Shell-aware: check allowed/denied command lists
        if tool_name == "run_command" and self._tools_config is not None:
            result = self._check_shell_command(arguments)
            if result is not None:
                return result

        # Prompt user via callback (TUI modal, etc.)
        if self._prompt_callback is not None:
            return await self._prompt_callback(tool_name, description)

        # No callback set — deny by default (safe fallback)
        logger.warning("Tool '%s' requires approval but no prompt callback set — denied", tool_name)
        return False

    def _check_shell_command(self, arguments: str) -> bool | None:
        """Check shell command against allowed/denied lists.

        Returns True (allow), False (deny), or None (fall through to prompt).
        """
        shell_config = self._tools_config.shell  # type: ignore[union-attr]

        try:
            cmd = json.loads(arguments).get("command", "")
        except (json.JSONDecodeError, AttributeError):
            return None  # can't parse, fall through to prompt

        try:
            base_cmd = shlex.split(cmd)[0]
        except (ValueError, IndexError):
            return None

        # Denied commands: prefix match on full command string
        for denied in shell_config.denied_commands:
            if cmd.startswith(denied):
                logger.info("Shell command '%s' matches denied prefix '%s'", cmd, denied)
                return False

        # Detect shell redirects that write to files — require approval
        if _WRITE_REDIRECT_PATTERN.search(cmd):
            logger.info("Shell command '%s' contains file-write redirect — requiring approval", cmd)
            return None  # fall through to user prompt

        # Allowed commands: base executable match
        if shell_config.allowed_commands:
            if base_cmd in shell_config.allowed_commands:
                logger.debug(
                    "Shell command '%s' auto-approved (base '%s' in allowed list)",
                    cmd,
                    base_cmd,
                )
                return True
            # Base command NOT in allowed list — fall through to prompt
            return None

        # No allowed list configured — fall through to prompt
        return None