nisse/nisse-core/src/nisse_core/actions.py

"""Actions by their effects (architecture P3, P4, L4), shared by every agent.

An action is a tool call as code sees it: the tool and its arguments, never a
model's account of them. The deterministic fast path (P4) sorts it into one of
these effects, by static rules each agent supplies:

- **harmless:** changes nothing outside a disposable sandbox;
- **contained:** stays within the agent's own area, or has a known undo;
- **unknown:** no rule recognises it, so the classifier decides (L4; until it
  exists, it runs and is logged);
- **dangerous:** irreversible, or visible beyond the agent's area: discussed
  with the operator first;
- **refused:** never run, with a reason the model can act on.

Shell commands are read statically, segment by segment. Anything that can't be
read that way (command substitution, subshells, here-documents, unbalanced
quotes) is unknown rather than guessed at, and a command is as bad as its
worst segment.
"""

import json
import posixpath
import shlex
from collections.abc import Callable, Iterable, Mapping
from dataclasses import dataclass
from typing import Any, Literal

Effect = Literal["harmless", "contained", "unknown", "dangerous", "refused"]
HARMLESS: Effect = "harmless"
CONTAINED: Effect = "contained"
UNKNOWN: Effect = "unknown"
DANGEROUS: Effect = "dangerous"
REFUSED: Effect = "refused"

_SEVERITY: dict[str, int] = {
    HARMLESS: 0,
    CONTAINED: 1,
    UNKNOWN: 2,
    DANGEROUS: 3,
    REFUSED: 4,
}


@dataclass(frozen=True)
class Action:
    tool: str
    arguments: Mapping[str, Any]

    def key(self) -> str:
        """What an approval binds to: exactly this tool with exactly these
        arguments (L3)."""
        return json.dumps(
            {"tool": self.tool, "arguments": dict(self.arguments)}, sort_keys=True
        )


@dataclass(frozen=True)
class Verdict:
    effect: Effect
    rule: str  # the rule that decided, for the log and the evaluation set
    reason: str  # in words, for the operator or the model


def worst(verdicts: Iterable[Verdict]) -> Verdict:
    """The most severe verdict; harmless when there are none."""
    found = list(verdicts)
    if not found:
        return Verdict(HARMLESS, "empty", "nothing to run")
    return max(found, key=lambda v: _SEVERITY[v.effect])


# --- reading shell commands --------------------------------------------------


@dataclass(frozen=True)
class Command:
    """One simple command of a shell line, as far as it can be known without
    running it. ``cwd`` is where it runs: None once a ``cd`` went somewhere
    that can't be worked out."""

    argv: tuple[str, ...]
    cwd: str | None
    redirects: tuple[str, ...] = ()  # files written by > and >>


# Separators between commands, and redirection operators.
_SEPARATORS = frozenset({";", "&&", "||", "|", "&", "\n", "|&"})
# Prefixes that run the rest of the line as the command.
_WRAPPERS = frozenset(
    {"sudo", "env", "nice", "nohup", "time", "command", "exec", "stdbuf", "xargs"}
)
_UNREADABLE = ("$(", "`", "<<", "<(", ">(")


def split_shell(text: str) -> list[tuple[list[str], list[str]]] | None:
    """The line's simple commands, each as (argv, files written by
    redirection); None when it can't be read statically."""
    if any(mark in text for mark in _UNREADABLE):
        return None
    lexer = shlex.shlex(text, posix=True, punctuation_chars="();<>|&\n")
    lexer.whitespace = " \t\r"
    lexer.whitespace_split = True
    lexer.commenters = "#"
    try:
        tokens = list(lexer)
    except ValueError:  # unbalanced quotes
        return None
    commands: list[tuple[list[str], list[str]]] = []
    argv: list[str] = []
    written: list[str] = []
    i = 0
    while i < len(tokens):
        token = tokens[i]
        if token in ("(", ")"):
            return None  # a subshell or a function
        if token in _SEPARATORS:
            if argv:
                commands.append((argv, written))
            argv, written = [], []
        elif set(token) <= set("<>&|") and ("<" in token or ">" in token):
            # A redirection: the next token is its target. A file descriptor
            # just before it (2>) belongs to the operator, not the command.
            if argv and argv[-1].isdigit():
                argv.pop()
            target = tokens[i + 1] if i + 1 < len(tokens) else ""
            i += 1
            if ">" in token and not (token.endswith("&") and target.isdigit()):
                written.append(target)
        else:
            argv.append(token)
        i += 1
    if argv:
        commands.append((argv, written))
    return commands


def unwrap(argv: list[str]) -> list[str]:
    """The command a wrapper runs: ``sudo make test`` is ``make test``.
    Leading variable assignments go too."""
    while argv:
        head = argv[0]
        if (
            "=" in head
            and not head.startswith("-")
            and head.split("=", 1)[0].isidentifier()
        ):
            argv = argv[1:]
        elif head in _WRAPPERS:
            argv = argv[1:]
            while argv and (argv[0].startswith("-") or "=" in argv[0]):
                argv = argv[1:]  # the wrapper's own options
        elif head == "timeout":
            argv = argv[1:]
            while argv and argv[0].startswith("-"):
                # -k and -s take a value, unless it's attached (--signal=KILL)
                takes = argv[0] in ("-k", "-s", "--kill-after", "--signal")
                argv = argv[2:] if takes else argv[1:]
            argv = argv[1:]  # the duration
        else:
            break
    return argv


def resolve(path: str, cwd: str | None) -> str | None:
    """Where ``path`` points, without touching the filesystem; None when that
    can't be known (a variable, a home directory, an unknown cwd)."""
    if not path or "$" in path or path.startswith("~"):
        return None
    if posixpath.isabs(path):
        return posixpath.normpath(path)
    if cwd is None:
        return None
    return posixpath.normpath(posixpath.join(cwd, path))


def inside(path: str | None, root: str) -> bool:
    if path is None:
        return False
    root = posixpath.normpath(root)
    return path == root or path.startswith(root.rstrip("/") + "/")


Judge = Callable[[Command], Verdict]


def classify_shell(text: str, cwd: str, judge: Judge, *, depth: int = 0) -> Verdict:
    """The worst verdict over the line's commands, each judged by the
    agent's rules. ``cd`` moves where the rest of the line runs, and
    ``sh -c '...'`` is read as the line it runs."""
    commands = split_shell(text)
    if commands is None:
        return Verdict(UNKNOWN, "unreadable", "can't be read without running it")
    verdicts: list[Verdict] = []
    here: str | None = cwd
    for raw, written in commands:
        argv = unwrap(raw)
        if not argv:
            continue
        if argv[0] == "cd":
            here = resolve(argv[1], here) if len(argv) > 1 else None
            continue
        if argv[0] in ("sh", "bash", "zsh", "dash") and "-c" in argv:
            inner = argv[argv.index("-c") + 1 :]
            if not inner or depth > 3:
                verdicts.append(Verdict(UNKNOWN, "unreadable", "a nested shell"))
                continue
            verdicts.append(
                classify_shell(inner[0], here or cwd, judge, depth=depth + 1)
                if here is not None
                else Verdict(UNKNOWN, "unreadable", "a shell in an unknown directory")
            )
            continue
        verdicts.append(judge(Command(tuple(argv), here, tuple(written))))
    return worst(verdicts)