openworker/coworker/permissions.py

"""Permission engine — decides allow / deny / ask-user for each proposed tool call.

Modes: Plan (read-only) · Interactive (auto reads, ask on writes/commands) · Auto
(allow, still path-scoped). Refined by argument patterns (path-under-root, command
prefixes) and a session allowlist. The engine only *decides*; the turn engine routes
`needs_user` decisions to a surface for approval and records the outcome.
"""

from __future__ import annotations

import re
import shlex
from dataclasses import dataclass, field
from enum import Enum
from pathlib import Path
from typing import Any, Optional
from urllib.parse import urlsplit

# Constructs whose *contents* we cannot evaluate, so a command carrying one is never
# eligible for prefix auto-run: command/process substitution, redirection (writes anywhere
# the allowlist never vetted), and variable expansion (the value was set out of view).
_OPAQUE_CONSTRUCTS = ("`", "$(", "$", ">", "<", "(")

# Separators that chain several commands into one string. Each part is checked independently
# against the allowlist — the old behaviour rejected the whole command outright, which both
# refused harmless `git status && git diff` and (because `-exec` needs no separator) still
# auto-allowed `find . -exec rm {} +` under a `find` prefix.
_SEPARATORS = ("&&", "||", ";", "|&", "|", "&", "\n", "\r")

# Programs that run *another* program named in their arguments. A prefix rule on the outer
# program can never vouch for the inner one, so these always fall through to approval.
_ARG_EXECUTORS = {
    "xargs", "env", "nohup", "nice", "stdbuf", "timeout", "watch", "sudo", "doas",
    "ssh", "docker", "podman", "kubectl", "npx", "pnpx", "bunx", "uvx",
}
# Interpreters carrying inline code, e.g. `python -c "..."`, `node -e "..."`.
_INLINE_CODE_FLAGS = {"-c", "-e", "--eval", "--command", "-Command", "-EncodedCommand"}
_INTERPRETERS = {
    "sh", "bash", "zsh", "dash", "ksh", "fish", "powershell", "pwsh", "cmd",
    "python", "python3", "node", "deno", "bun", "ruby", "perl", "php",
}
# Flags that turn a search/list tool into an execution or deletion tool.
_DANGEROUS_FLAGS = {"-exec", "-execdir", "-delete", "-ok", "-okdir", "-fprintf"}


def _split_commands(command: str) -> list[str]:
    """Split a compound command on its separators. Longest separators first so `&&` isn't
    read as two `&`. Purely textual — quoted separators are not respected, which is
    deliberate: over-splitting only ever produces MORE parts to justify, never fewer."""
    parts = [command]
    for sep in _SEPARATORS:
        parts = [chunk for part in parts for chunk in part.split(sep)]
    return [p.strip() for p in parts if p.strip()]


def _is_prefix_eligible(argv: list[str]) -> bool:
    """False when a parsed command can never be vouched for by a prefix rule, because it
    runs code the rule never saw: another program named in its arguments, inline source, or
    an execution/deletion flag."""
    if not argv:
        return False
    program = Path(argv[0]).name.lower()
    program = program[:-4] if program.endswith(".exe") else program
    if program in _ARG_EXECUTORS:
        return False
    if program in _INTERPRETERS and any(a in _INLINE_CODE_FLAGS for a in argv[1:]):
        return False
    if any(a.lower() in _DANGEROUS_FLAGS for a in argv[1:]):
        return False
    return True


def protected_paths() -> list[Path]:
    """Files that govern the permission system itself. Nothing the agent does may write
    these — in any mode, through any tool. The escalation this blocks is: approve one
    ordinary-looking command, it quietly appends to the rule file, every future session is
    more permissive. That happens in the DEFAULT interactive mode, so this cannot be a
    property of a sandbox or of any one mode; it is a floor."""
    from .secrets import state_dir

    base = state_dir()
    return [
        base / "config.toml",
        base / "risk_overrides.json",
        base / "workspace_trust.json",
        base / "unattended.json",
        base / "coworker.db",  # session records carry the saved "always allow" grants
        base / "secrets.json",
        base / "inbox_routing.json",
    ]


# Files INSIDE a workspace that execute on a later, innocuous-looking action. An edit here
# is a deferred command: writing `.git/hooks/pre-commit` and then running `git commit` runs
# it. They stay writable, but never WITHOUT a human — no auto-approve path may clear them.
_PROTECTED_IN_PROJECT = (
    ".git/hooks/",
    ".github/workflows/",
    ".gitlab-ci.yml",
    ".vscode/tasks.json",
    ".coworker/",  # workspace policy + skills the agent would otherwise self-grant
)


def _is_protected_in_project(candidate: Path) -> bool:
    posix = candidate.as_posix()
    return any(
        (f"/{marker}" in posix or posix.startswith(marker))
        if marker.endswith("/")
        else posix.endswith("/" + marker)
        for marker in _PROTECTED_IN_PROJECT
    )


def _host_of(url_or_domain: str) -> str:
    """The lowercased host of a URL, or a bare domain as-is. `''` when there's nothing
    usable. Accepts both `https://docs.python.org/x` and `docs.python.org`."""
    s = (url_or_domain or "").strip().lower()
    if not s:
        return ""
    if "://" in s:
        return urlsplit(s).hostname or ""
    return urlsplit("//" + s).hostname or s


# The argument that names a write tool's target path, when it's a single top-level field.
# Patch/diff tools carry their paths inside the blob instead — extracted in `write_paths`.
_PATH_ARG: dict[str, str] = {"write_file": "path", "replace_in_file": "path"}
# apply_patch (Codex format) file headers, and unified-diff `+++ b/<path>` headers.
_APPLY_PATCH_FILE = re.compile(
    r"^\*\*\* (?:Add|Update|Delete) File: (.+)$", re.MULTILINE
)
_APPLY_PATCH_MOVE = re.compile(r"^\*\*\* Move to: (.+)$", re.MULTILINE)
_UNIFIED_DIFF_FILE = re.compile(r"^\+\+\+ (?:b/)?(.+?)\s*$", re.MULTILINE)


def write_paths(tool_name: str, arguments: dict[str, Any]) -> tuple[list[str], bool]:
    """Every filesystem path a write tool would touch, for root scoping.

    Returns ``(paths, located)``. ``located`` is False when the path can't be determined
    (an unknown write tool, or a patch/diff blob with no parseable file header) — the caller
    must then fail closed rather than skip scoping, so an unscoped write can't slip through
    auto/custom mode.
    """
    arg = _PATH_ARG.get(tool_name)
    if arg is not None:
        value = arguments.get(arg)
        return ([str(value)], True) if value else ([], False)
    if tool_name == "apply_patch":
        blob = str(arguments.get("patch", ""))
        paths = _APPLY_PATCH_FILE.findall(blob) + _APPLY_PATCH_MOVE.findall(blob)
        return ([p.strip() for p in paths], bool(paths))
    if tool_name == "apply_unified_diff":
        blob = str(arguments.get("diff", ""))
        paths = [p for p in _UNIFIED_DIFF_FILE.findall(blob) if p and p != "/dev/null"]
        return (paths, bool(paths))
    # Unknown write tool (e.g. one promoted to write via a user override): we cannot locate
    # its path, so it cannot be auto-scoped.
    return ([], False)

from .risk import (  # re-exported for back-compat (manager.py imports WRITE_TOOLS)
    SHELL_TOOL,
    WRITE_TOOLS,
    RiskClass,
    RiskOverrides,
    classify,
    is_consequential,
)


class Mode(str, Enum):
    DISCUSS = "discuss"  # read-only conversation: no edits, no planning workflow
    PLAN = (
        "plan"  # read-only + the planning contract (explore → propose_plan → execute)
    )
    INTERACTIVE = "interactive"  # ask for approval (default)
    # Renamed from "auto" (spec §1.5, 2026-08-12): "bypass" names the action — switching a
    # safety system off — and can't be confused with AUTO_APPROVE in a picker. Deliberately
    # NOT "bypass-ALL-approvals": Phase 1's floors (settings files, out-of-root writes,
    # `.git/hooks`) still hold in this mode, so "all" would be a false promise.
    BYPASS_APPROVALS = "bypass-approvals"  # full access (minus the hard floors)
    # Interactive, but an LLM reviewer judges each would-be approval card first: clear
    # allows run without a prompt, everything else still reaches the human. The reviewer
    # can only turn "ask" into "allow", never "blocked" into "allow" (spec §1.2). With no
    # reviewer plugged into the engine this mode behaves exactly like INTERACTIVE.
    AUTO_APPROVE = "auto-approve"
    CUSTOM = "custom"  # interactive + auto-allow the config's `auto_allow` tools

    @classmethod
    def _missing_(cls, value: object) -> "Mode | None":
        # Legacy spelling from configs, saved sessions, and older UIs.
        if value == "auto":
            return cls.BYPASS_APPROVALS
        return None


# Modes whose enforcement is read-only. DISCUSS and PLAN share the same gate; they differ
# only in intent — PLAN additionally drives the agent toward a propose_plan approval.
READ_ONLY_MODES = frozenset({Mode.DISCUSS, Mode.PLAN})


@dataclass
class Decision:
    allowed: bool
    reason: str = ""
    needs_user: bool = False  # True → surface should prompt the user for approval
    # Set when a task-scoped standing rule allowed the call ("tool → target") so the
    # engine can audit the exact rule and the tool card can say so (§25).
    rule: str = ""


def standing_rule_candidate(
    tool_name: str,
    arguments: dict[str, Any],
    metadata: Any = None,
    overrides: Optional[RiskOverrides] = None,
) -> Optional[str]:
    """The target value iff this call is eligible for a task-scoped standing rule
    (UX-DECISIONS §25): external-risk only (never exec/write-local — shell asks forever),
    the tool must declare a target argument, and the call must actually name a target.
    Returns None otherwise — ineligible calls keep parking approvals as today."""
    from .connectors.tool_defs import target_arg_for

    if classify(tool_name, metadata, overrides) is not RiskClass.EXTERNAL:
        return None
    arg = target_arg_for(tool_name)
    if arg is None:
        return None
    value = str((arguments or {}).get(arg) or "").strip()
    return value or None


@dataclass
class PermissionEngine:
    workspace_root: Path
    mode: Mode = Mode.INTERACTIVE
    allowed_commands: list[str] = field(default_factory=list)
    auto_allow_tools: set[str] = field(default_factory=set)
    session_allow_tools: set[str] = field(default_factory=set)
    session_allow_commands: set[str] = field(default_factory=set)
    # Egress domains that auto-run without a prompt: `allowed_domains` from user config, plus
    # `session_allow_domains` minted by "Always allow this domain". Matched by exact host or
    # subdomain suffix (see `_domain_allowed`).
    allowed_domains: list[str] = field(default_factory=list)
    session_allow_domains: set[str] = field(default_factory=set)
    # Task-scoped standing rules (§25): {tool: {allowed targets}}, seeded from the owning
    # ScheduledTask's target-shaped entries. Kept by reference and re-read every check, so a
    # rule minted mid-run ("Allow every time") applies to the run's next call too.
    task_rules: dict[str, set[str]] = field(default_factory=dict)
    # User-local risk override resolver (Phase 2). None → use the base classification.
    risk_overrides: Optional[RiskOverrides] = None
    # Shared, possibly-mutable list of roots (RootDir-like / dicts). When omitted, the single
    # `workspace_root` is the sole writable root (back-compat). Kept by reference and re-read on
    # every check, so runtime add/remove of folders takes effect without rebuilding the engine.
    roots: Optional[list] = None

    def __post_init__(self) -> None:
        self.workspace_root = Path(self.workspace_root).expanduser().resolve()
        self.auto_allow_tools = set(self.auto_allow_tools)
        if self.roots is None:
            self.roots = [{"path": self.workspace_root, "writable": True}]

    def _resolved_roots(self) -> list[tuple[Path, bool]]:
        out: list[tuple[Path, bool]] = []
        for r in self.roots or []:
            if isinstance(r, dict):
                p, w = r["path"], bool(r.get("writable", False))
            elif isinstance(r, (str, Path)):
                p, w = r, True
            else:  # duck-typed RootDir-like
                p, w = getattr(r, "path"), bool(getattr(r, "writable", False))
            out.append((Path(p).expanduser().resolve(), w))
        return out

    def evaluate(
        self, tool_name: str, arguments: dict[str, Any], metadata: Any = None
    ) -> Decision:
        arguments = arguments or {}
        is_connector = getattr(metadata, "category", "") == "connector"
        risk = classify(tool_name, metadata, self.risk_overrides)
        is_write = risk is RiskClass.WRITE_LOCAL
        is_shell = risk is RiskClass.EXEC
        is_egress = risk is RiskClass.EGRESS
        consequential = is_consequential(risk)

        # SELF-PROTECTION FLOOR — runs before mode, allowlists and every auto-approve path,
        # because the escalation it blocks happens in the DEFAULT mode. No verdict below can
        # reach these files, and no human click in the flow can grant it either: loosening
        # requires editing the files out-of-band.
        if is_write or is_shell:
            hit = self._touches_protected(tool_name, arguments, is_shell)
            if hit is not None:
                return Decision(
                    False,
                    f"refusing to modify OpenWorker's own settings: {hit}",
                    needs_user=False,
                )

        # Discuss / plan modes: read-only.
        if self.mode in READ_ONLY_MODES and consequential:
            return Decision(
                False, f"{self.mode.value} mode is read-only", needs_user=False
            )

        # Path scoping for writes (all modes): every path the write touches must land in a
        # writable root. A write whose path can't be located is not scoped-able, so it fails
        # closed to approval rather than slipping through auto/custom unscoped.
        needs_human_for_protected = False
        if is_write:
            paths, located = write_paths(tool_name, arguments)
            if not located:
                return Decision(
                    False,
                    "cannot determine the write path to scope",
                    needs_user=True,
                )
            for path in paths:
                if not self._under_writable_root(path):
                    return Decision(
                        False, f"path is not in a writable directory: {path}"
                    )
                # In-project files that run on a later action (git hooks, CI configs) may be
                # edited, but never by an auto-approve path — a human must see it.
                if _is_protected_in_project(self._candidate(path)):
                    needs_human_for_protected = True

        # Non-consequential tools always run.
        if not consequential:
            return Decision(True, "low risk")

        # A protected in-project target (git hooks, CI config) skips every auto-approve path
        # below — including auto mode and the session/config allowlists — and asks.
        if needs_human_for_protected:
            return Decision(
                False,
                "this file runs automatically later — approval required",
                needs_user=True,
            )

        # Full access.
        if self.mode is Mode.BYPASS_APPROVALS:
            return Decision(True, "full access")

        # interactive / custom / auto-approve: allowlists.
        #
        # In AUTO_APPROVE, session grants ("always allow this …" clicks) deliberately do
        # NOT auto-allow (spec §1.5): out-of-band standing policy — the user-settings
        # allowlists checked via `_command_allowed` / config `allowed_domains` — may skip
        # the judge, but an in-flow click may not. A domain grant matches on host only and
        # is blind to the path and query string (where exfiltration rides), and command
        # grants replay as exact text; both are precisely what the reviewer should see.
        # The skipped checks return `needs_user` instead, which routes to the reviewer.
        honor_session_grants = self.mode is not Mode.AUTO_APPROVE
        if is_shell:
            command = str(arguments.get("command", ""))
            if self._command_allowed(command):
                return Decision(True, "command on allowlist")
            if (
                honor_session_grants
                and command
                and command in self.session_allow_commands
            ):
                return Decision(True, "command allowed for session")
        if is_egress:
            url = str(arguments.get("url", ""))
            if self._domain_allowed(url, include_session=honor_session_grants):
                return Decision(True, "domain on allowlist")
        if (
            honor_session_grants
            and tool_name in self.session_allow_tools
            and not is_connector
        ):
            return Decision(True, "tool allowed for session")

        # Task-scoped standing rules (§25): tool + exact target, owned by the automation.
        # Deliberately NOT subject to the connector exclusion above — the exact-target
        # binding is what makes auto-allowing a connector tool safe. Never for exec risk
        # (candidate extraction is external-risk-only), and additive on top of the mode:
        # read-only modes already returned before this point.
        if tool_name in self.task_rules:
            target = standing_rule_candidate(
                tool_name, arguments, metadata, self.risk_overrides
            )
            if target and target in self.task_rules[tool_name]:
                rule = f"{tool_name} → {target}"
                return Decision(True, f"allowed by standing rule: {rule}", rule=rule)

        # Custom mode auto-approves the configured tools.
        if self.mode is Mode.CUSTOM and tool_name in self.auto_allow_tools:
            return Decision(True, "auto-allowed by config")

        # Otherwise: ask the user.
        return Decision(False, "requires approval", needs_user=True)

    # -- session memory ---------------------------------------------------------
    def allow_tool_for_session(self, tool_name: str) -> None:
        self.session_allow_tools.add(tool_name)

    def allow_command_for_session(self, command: str) -> None:
        if command:
            self.session_allow_commands.add(command)

    def allow_domain_for_session(self, url_or_domain: str) -> None:
        """Remember an egress destination for this session ("Always allow this domain").

        A leading `www.` is stripped at minting (§1.9): `bbc.com` and `www.bbc.com` are one
        site in every user's mental model, and the suffix match in `_domain_allowed` already
        treats `www.bbc.com` as a subdomain of `bbc.com`. Pure spelling only — never eTLD+1
        or any broader normalisation, which would silently widen the grant."""
        host = _host_of(url_or_domain)
        if host.startswith("www."):
            host = host[4:]
        if host:
            self.session_allow_domains.add(host)

    # -- helpers ----------------------------------------------------------------
    def _candidate(self, path: str) -> Path:
        # Relative paths resolve against the primary (workspace_root); absolute/`~` taken as-is.
        p = Path(path).expanduser()
        return p.resolve() if p.is_absolute() else (self.workspace_root / p).resolve()

    def _under_root(self, path: str) -> bool:
        candidate = self._candidate(path)
        for rp, _ in self._resolved_roots():
            try:
                candidate.relative_to(rp)
                return True
            except ValueError:
                continue
        return False

    def _under_writable_root(self, path: str) -> bool:
        candidate = self._candidate(path)
        for rp, writable in self._resolved_roots():
            if not writable:
                continue
            try:
                candidate.relative_to(rp)
                return True
            except ValueError:
                continue
        return False

    def _touches_protected(
        self, tool_name: str, arguments: dict[str, Any], is_shell: bool
    ) -> Optional[str]:
        """The protected settings path this call would modify, or None.

        For writes we resolve the real target. For shell we can only inspect the command
        text — parser depth, so it stops accidents and casual attempts, not a determined
        adversary (that needs the OS sandbox). Cheap and worth having regardless.

        Shell matching is on the FULL path only, never a bare filename: matching
        `secrets.json` anywhere in a command would refuse unrelated work that merely
        mentions the name. A command naming the real settings path is refused whether it
        reads or writes — we cannot tell which from text, and the conservative direction is
        the right one for these files.
        """
        targets = [str(p) for p in protected_paths()]
        if is_shell:
            command = str(arguments.get("command", ""))
            if not command:
                return None
            lowered = command.replace("\\", "/").lower()
            for target in targets:
                if target.replace("\\", "/").lower() in lowered:
                    return target
            return None
        paths, located = write_paths(tool_name, arguments)
        if not located:
            return None  # unlocatable writes are already failed closed by the caller
        resolved = {str(self._candidate(p)) for p in paths}
        for target in targets:
            if str(Path(target).resolve()) in resolved:
                return target
        return None

    def _domain_allowed(self, url: str, *, include_session: bool = True) -> bool:
        """True when the URL's host is an allowed egress destination — an exact match or a
        subdomain of an allowed domain (so `docs.python.org` matches `python.org`, but
        `evil-python.org` never matches `python.org`).

        `include_session=False` (AUTO_APPROVE mode) checks the user-settings list only:
        mid-session "always allow this domain" clicks don't bypass the reviewer there."""
        host = _host_of(url)
        if not host:
            return False
        allowed = {d for d in (_host_of(x) for x in self.allowed_domains) if d}
        if include_session:
            allowed |= self.session_allow_domains
        for dom in allowed:
            if host == dom or host.endswith("." + dom):
                return True
        return False

    def _command_allowed(self, command: str) -> bool:
        """True only when EVERY part of a (possibly compound) command is independently
        covered by an allowlist entry.

        An allowlist entry auto-runs without approval, and a prefix rule can only vouch for
        the words it matched — everything after is unexamined. So this does two jobs:
        guarantee the unexamined tail can only be arguments, then match the beginning.

        - Constructs whose contents we can't evaluate (substitution, redirection, variable
          expansion) disqualify the whole command.
        - Compound commands are split and each part checked on its own, so
          `git status && git diff` runs when both are allowed, while
          `git status && rm -rf ~` does not.
        - Parts that run code named in their arguments (`xargs`, `sh -c`, `find -exec`,
          `-delete`) are never prefix-eligible: a `find` rule must not auto-run
          `find . -exec rm {} +`.
        - Matching is on parsed words, not text, so `git status` covers `git status -s` but
          never `git statusfoo` or a bare `git`.
        """
        if not command.strip():
            return False
        if any(tok in command for tok in _OPAQUE_CONSTRUCTS):
            return False
        parts = _split_commands(command)
        if not parts:
            return False
        prefixes: list[list[str]] = []
        for allowed in self.allowed_commands:
            try:
                prefix = shlex.split(allowed)
            except ValueError:
                continue
            if prefix:
                prefixes.append(prefix)
        if not prefixes:
            return False
        for part in parts:
            try:
                argv = shlex.split(part)
            except ValueError:
                return False  # unbalanced quotes etc. — treat as not-allowlisted
            if not argv or not _is_prefix_eligible(argv):
                return False
            if not any(argv[: len(p)] == p for p in prefixes):
                return False
        return True