nisse/nisse-dev/src/nisse_dev/loop/actions.py

"""nisse-dev's fast-path rules (architecture D6, P4): what its tool calls
do, judged by code before anything runs.

- **Harmless:** reading the repository, running builds, tests and linters on
  the build VM, querying.
- **Contained:** editing the workspace, local commits, installing on the build
  VM and otherwise changing it (D3).
- **Dangerous:** anything reaching beyond the item's repository or build VM
  (pushing, other hosts, uploads, publishing, the forge's own tools), or
  wrecking the host itself (deleting system or home directories, `dd`,
  `mkfs`, `reboot`). Discussed with the operator first. Deleting elsewhere on
  the build VM is contained: it is disposable, and in tests the "build VM" is
  this machine, which the host rules protect.
- **Refused:** changing branches by hand. The harness owns branches (D4), so
  the model is told to use its milestone tools instead.
- **Unknown:** everything else, for the classifier (M3.2); until then it runs,
  and is logged.

A rule never makes a command less severe than its worst part; what can't be
read statically is unknown (nisse_core.authority.actions).
"""

import posixpath
from collections.abc import Callable

from nisse_core.authority.actions import (
    CONTAINED,
    DANGEROUS,
    HARMLESS,
    REFUSED,
    UNKNOWN,
    Action,
    Command,
    Verdict,
    classify_shell,
    inside,
    resolve,
    split_shell,
    unwrap,
    worst,
)

READING_TOOLS = frozenset({"read_file", "list_dir"})

_READ = frozenset(
    {
        "ls", "cat", "head", "tail", "wc", "grep", "egrep", "fgrep", "rg", "echo",
        "printf", "pwd", "which", "file", "stat", "diff", "sort", "uniq", "tree",
        "du", "df", "true", "false", "test", "[", "awk", "cut", "tr", "jq",
        "date", "uname", "whoami", "id", "basename", "dirname", "realpath",
        "readlink", "less", "more", "nl", "column", "xxd", "od", "sha256sum",
        "md5sum", "printenv", "type", "hostname", "ps", "free", "uptime",
        "nproc", "seq", "sleep", "cmp", "comm", "env",
    }
)  # fmt: skip
_CHECKS = frozenset(
    {
        "pytest", "py.test", "ruff", "mypy", "pyright", "black", "isort",
        "flake8", "pylint", "tox", "nox", "coverage", "eslint", "prettier",
        "tsc", "jest", "vitest", "shellcheck", "gofmt",
    }
)  # fmt: skip
_CHANGES = frozenset(
    {
        "mkdir", "touch", "cp", "mv", "chmod", "chown", "ln", "tee", "truncate",
        "patch", "tar", "unzip", "gzip", "gunzip", "zip", "kill", "pkill",
        "systemctl", "service", "crontab", "useradd", "groupadd",
    }
)  # fmt: skip
_DELETES = frozenset({"rm", "rmdir", "unlink", "shred"})
_OTHER_HOSTS = frozenset(
    {"ssh", "scp", "sftp", "telnet", "nc", "ncat", "netcat", "ftp", "mosh"}
)
_FORGE_TOOLS = frozenset({"gh", "tea", "hub", "glab", "twine"})
_WRECKS_THE_VM = frozenset(
    {
        "dd", "fdisk", "parted", "shutdown", "reboot", "halt", "poweroff",
        "iptables", "ip6tables", "nft", "ufw", "mount", "umount",
    }
)  # fmt: skip
_PACKAGE_MANAGERS = frozenset({"apt", "apt-get", "dnf", "yum", "apk", "zypper"})

_TEST_TARGETS = frozenset({"", "test", "tests", "check", "lint", "build", "all", "fmt"})
_NPM_CHECKS = frozenset({"test", "lint", "build", "check", "typecheck", "format"})

_BRANCHES = (
    "the harness owns branches: work stays on the milestone's branch, and "
    "start_milestone and finish_milestone make and publish it"
)


def commits(action: Action) -> bool:
    """Whether the call runs a `git commit` anywhere in its line (#129)."""
    if action.tool != "run_command":
        return False
    for argv, _ in split_shell(str(action.arguments.get("command", ""))) or []:
        argv = unwrap(argv)
        if argv[:1] != ["git"]:
            continue
        args = argv[1:]
        while args and args[0] in ("-C", "-c"):
            args = args[2:]
        if args[:1] == ["commit"]:
            return True
    return False


def classify(action: Action, root: str) -> Verdict:
    """The fast path's verdict on one execution tool call in the workspace
    at ``root``."""
    if action.tool in READING_TOOLS:
        return Verdict(HARMLESS, "read", "reads the workspace")
    if action.tool == "write_file":
        # Writes outside the workspace are refused by the workspace itself.
        return Verdict(CONTAINED, "workspace-edit", "edits the workspace")
    if action.tool == "run_command":
        line = str(action.arguments.get("command", ""))
        return classify_shell(line, root, lambda c: judge(c, root))
    return Verdict(UNKNOWN, "no-rule", f"no rule for {action.tool}")


def judge(command: Command, root: str) -> Verdict:
    """One simple command, and what its redirections write."""
    verdicts = [_program(command, root)]
    for target in command.redirects:
        if target not in ("/dev/null", "/dev/stdout", "/dev/stderr"):
            verdicts.append(Verdict(CONTAINED, "writes-file", f"writes {target}"))
    return worst(verdicts)


def _program(command: Command, root: str) -> Verdict:
    argv = list(command.argv)
    name = argv[0]
    if "/" in name:
        if name.startswith(("/usr/", "/bin/", "/sbin/")):
            name = posixpath.basename(name)  # /usr/bin/git is git
        else:
            return Verdict(UNKNOWN, "script", f"runs {name}")
    args = argv[1:]
    if name.startswith("mkfs"):
        name = "dd"
    if name in _RULES:
        return _RULES[name](args, command, root)
    if name in _READ:
        return Verdict(HARMLESS, "read-only", f"{name} only reads")
    if name in _CHECKS:
        return Verdict(HARMLESS, "checks", f"{name} checks the code")
    if name in _CHANGES:
        return Verdict(CONTAINED, "changes-vm", f"{name} changes files on the build VM")
    if name in _DELETES:
        return _deletes(name, args, command, root)
    if name in _OTHER_HOSTS:
        return Verdict(DANGEROUS, "other-host", f"{name} reaches another host")
    if name in _FORGE_TOOLS:
        return Verdict(
            DANGEROUS, "forge-tool", f"{name} acts on a forge or index directly"
        )
    if name in _WRECKS_THE_VM:
        return Verdict(
            DANGEROUS, "wrecks-host", f"{name} can break the host or its network"
        )
    if name in _PACKAGE_MANAGERS:
        return Verdict(CONTAINED, "install", "installs on the build VM")
    return Verdict(UNKNOWN, "no-rule", f"no rule for {name}")


# Deleting under these breaks the host itself. On a real build VM that only
# costs a rebuild, but in tests the "build VM" is this machine's own sshd.
_HOST = (
    "/bin", "/boot", "/dev", "/etc", "/home", "/lib", "/lib32", "/lib64",
    "/libx32", "/proc", "/root", "/sbin", "/snap", "/sys", "/usr", "/var",
)  # fmt: skip
_SCRATCH = ("/tmp", "/var/tmp")


def _wrecks_host(target: str, cwd: str | None, root: str) -> bool | None:
    """Whether deleting ``target`` breaks the host; None when it can't be
    told (a variable). The workspace is never the host's, wherever it is."""
    if target in ("~", "$HOME") or target.startswith(("~/", "$HOME/", "${HOME}")):
        return True
    path = resolve(target, cwd)
    if path is None:
        return None
    if inside(path, root):
        return False
    if (
        path == "/"
        or path.startswith("/*")
        or posixpath.dirname(path) == "/"
        and "*" in path
    ):
        return True
    if any(inside(path, s) for s in _SCRATCH):
        return False
    return any(inside(path, h) for h in _HOST)


def _deletes(name: str, args: list[str], command: Command, root: str) -> Verdict:
    """The build VM is disposable (D3), so deleting on it is contained, except
    what breaks the host itself."""
    targets = [a for a in args if not a.startswith("-")]
    wrecks = [t for t in targets if _wrecks_host(t, command.cwd, root)]
    if wrecks:
        return Verdict(
            DANGEROUS, "wrecks-host", f"{name} deletes {', '.join(wrecks)} on the host"
        )
    if any(_wrecks_host(t, command.cwd, root) is None for t in targets):
        return Verdict(UNKNOWN, "no-rule", f"{name} deletes what a variable names")
    return Verdict(CONTAINED, "deletes", f"{name} deletes files on the build VM")


def _git(args: list[str], command: Command, root: str) -> Verdict:
    while args and args[0] in ("-C", "-c"):
        args = args[2:]  # global options and their values
    sub, rest = (args[0], args[1:]) if args else ("", [])
    flags = [a for a in rest if a.startswith("-")]
    names = [a for a in rest if not a.startswith("-")]
    if sub in ("push", "send-email", "request-pull"):
        return Verdict(DANGEROUS, "git-push", "pushes beyond the workspace")
    if sub == "remote" and names and names[0] in (
        "add", "set-url", "remove", "rm", "rename",
    ):  # fmt: skip
        return Verdict(DANGEROUS, "git-remote", "changes where the clone pushes")
    if sub in ("switch", "merge", "rebase", "pull") or (
        sub == "worktree" and names[:1] == ["add"]
    ):
        return Verdict(REFUSED, "branches", _BRANCHES)
    if sub == "checkout":
        if "--" in rest or (names and all(n == "." for n in names)):
            return Verdict(CONTAINED, "git-local", "restores files in the workspace")
        return Verdict(REFUSED, "branches", _BRANCHES + "; git restore puts files back")
    if sub == "branch":
        listing = {"-a", "-r", "-v", "-vv", "--list", "-l", "--show-current"}
        if not names and set(flags) <= listing:
            return Verdict(HARMLESS, "git-read", "lists branches")
        if flags and set(flags) & {"--contains", "--merged", "--no-merged"}:
            return Verdict(HARMLESS, "git-read", "lists branches")
        return Verdict(REFUSED, "branches", _BRANCHES)
    if sub in (
        "status", "diff", "log", "show", "blame", "rev-parse", "ls-files",
        "grep", "describe", "shortlog", "reflog", "rev-list", "cat-file",
        "ls-tree", "whatchanged", "version", "help",
    ):  # fmt: skip
        return Verdict(HARMLESS, "git-read", f"git {sub} only reads")
    if sub == "config" and (not names or set(flags) & {"--get", "--list", "-l"}):
        return Verdict(HARMLESS, "git-read", "reads git configuration")
    if sub == "remote" and not names:
        return Verdict(HARMLESS, "git-read", "lists remotes")
    if sub == "tag" and not names:
        return Verdict(HARMLESS, "git-read", "lists tags")
    if sub == "stash" and names[:1] in (["list"], ["show"]):
        return Verdict(HARMLESS, "git-read", "lists stashes")
    if sub in (
        "add", "commit", "restore", "rm", "mv", "reset", "clean", "stash",
        "apply", "tag", "init", "fetch", "cherry-pick", "revert", "notes", "am",
        "config",
    ):  # fmt: skip
        return Verdict(CONTAINED, "git-local", f"git {sub} changes the local clone")
    return Verdict(UNKNOWN, "no-rule", f"no rule for git {sub}")


def _python(args: list[str], command: Command, root: str) -> Verdict:
    if args[:1] in (["--version"], ["-V"]):
        return Verdict(HARMLESS, "read-only", "prints the version")
    if args[:1] == ["-m"] and len(args) > 1:
        return judge(Command(tuple(args[1:]), command.cwd), root)
    return Verdict(UNKNOWN, "script", "runs Python code")


def _runner(args: list[str], command: Command, root: str) -> Verdict:
    """uv, poetry and friends: `run X` is X; installing changes the VM."""
    sub = args[0] if args else ""
    if sub == "run" and len(args) > 1:
        rest = args[1:]
        while rest and rest[0].startswith("-"):
            rest = rest[1:]
        if rest:
            return judge(Command(tuple(rest), command.cwd), root)
    if sub == "publish":
        return Verdict(DANGEROUS, "publish", "publishes a package")
    if sub in ("pip", "add", "remove", "sync", "lock", "venv", "install", "update"):
        return Verdict(CONTAINED, "install", "installs on the build VM")
    if sub in ("--version", "version", "tree", "show"):
        return Verdict(HARMLESS, "read-only", "only reads")
    return Verdict(UNKNOWN, "no-rule", f"no rule for that {sub}")


def _pip(args: list[str], command: Command, root: str) -> Verdict:
    sub = args[0] if args else ""
    if sub in ("install", "uninstall", "download"):
        return Verdict(CONTAINED, "install", "installs on the build VM")
    if sub in ("list", "show", "freeze", "check", "--version"):
        return Verdict(HARMLESS, "read-only", "only reads")
    return Verdict(UNKNOWN, "no-rule", f"no rule for pip {sub}")


def _npm(args: list[str], command: Command, root: str) -> Verdict:
    sub = args[0] if args else ""
    if sub == "publish":
        return Verdict(DANGEROUS, "publish", "publishes a package")
    if sub in ("test", "t") or (sub == "run" and args[1:2] and args[1] in _NPM_CHECKS):
        return Verdict(HARMLESS, "checks", "runs the project's checks")
    if sub in ("install", "i", "ci", "add", "remove", "uninstall", "update"):
        return Verdict(CONTAINED, "install", "installs on the build VM")
    if sub in ("ls", "list", "view", "outdated", "--version"):
        return Verdict(HARMLESS, "read-only", "only reads")
    return Verdict(UNKNOWN, "no-rule", f"no rule for {sub or 'that'}")


def _make(args: list[str], command: Command, root: str) -> Verdict:
    targets = [a for a in args if not a.startswith("-") and "=" not in a]
    if all(t in _TEST_TARGETS for t in targets):
        return Verdict(HARMLESS, "checks", "builds or checks the code")
    return Verdict(UNKNOWN, "no-rule", f"make {' '.join(targets)}")


def _cargo_go(args: list[str], command: Command, root: str) -> Verdict:
    sub = args[0] if args else ""
    if sub == "publish":
        return Verdict(DANGEROUS, "publish", "publishes a package")
    if sub in ("test", "build", "check", "clippy", "fmt", "vet", "doc", "version"):
        return Verdict(HARMLESS, "checks", "builds or checks the code")
    if sub in ("add", "fetch", "get", "mod", "install", "update"):
        return Verdict(CONTAINED, "install", "installs on the build VM")
    return Verdict(UNKNOWN, "no-rule", f"no rule for {sub or 'that'}")


_UPLOADS = (
    "-T", "--upload-file", "-d", "--data", "--data-raw", "--data-binary",
    "--data-urlencode", "-F", "--form", "--json", "--post-data", "--post-file",
    "--body-data", "--body-file",
)  # fmt: skip


def _fetches(args: list[str], command: Command, root: str) -> Verdict:
    """curl and wget: sending data out is dangerous, fetching is not."""
    for i, arg in enumerate(args):
        if arg.startswith(_UPLOADS) or arg.startswith("--method"):
            return Verdict(DANGEROUS, "upload", "sends data to another host")
        method = args[i + 1].upper() if i + 1 < len(args) else ""
        if arg in ("-X", "--request") and method not in ("GET", "HEAD"):
            return Verdict(DANGEROUS, "upload", "sends data to another host")
    return Verdict(CONTAINED, "download", "downloads to the build VM")


def _docker(args: list[str], command: Command, root: str) -> Verdict:
    sub = args[0] if args else ""
    if sub in ("push", "login"):
        return Verdict(DANGEROUS, "publish", f"docker {sub} reaches a registry")
    if sub in ("ps", "images", "inspect", "logs", "version", "info"):
        return Verdict(HARMLESS, "read-only", "only reads")
    if sub in ("build", "pull"):
        return Verdict(CONTAINED, "changes-vm", "builds on the build VM")
    return Verdict(UNKNOWN, "no-rule", f"no rule for docker {sub}")


def _rsync(args: list[str], command: Command, root: str) -> Verdict:
    if any(":" in a and not a.startswith("-") for a in args):
        return Verdict(DANGEROUS, "other-host", "rsync reaches another host")
    return Verdict(CONTAINED, "changes-vm", "copies files on the build VM")


def _sed(args: list[str], command: Command, root: str) -> Verdict:
    if any(a.startswith("-i") or a == "--in-place" for a in args):
        return Verdict(CONTAINED, "changes-vm", "edits files in place")
    return Verdict(HARMLESS, "read-only", "sed only reads")


def _find(args: list[str], command: Command, root: str) -> Verdict:
    if set(args) & {"-delete", "-exec", "-execdir", "-ok", "-okdir"}:
        return Verdict(UNKNOWN, "no-rule", "find runs or deletes things")
    return Verdict(HARMLESS, "read-only", "find only reads")


def _interpreter(args: list[str], command: Command, root: str) -> Verdict:
    if args[:1] in (["--version"], ["-v"], ["-V"]):
        return Verdict(HARMLESS, "read-only", "prints the version")
    return Verdict(UNKNOWN, "script", "runs a program")


def _dangerous_vm(args: list[str], command: Command, root: str) -> Verdict:
    return Verdict(DANGEROUS, "wrecks-host", "can break the host")


Rule = Callable[[list[str], Command, str], Verdict]

_RULES: dict[str, Rule] = {
    "git": _git,
    "python": _python,
    "python3": _python,
    "uv": _runner,
    "poetry": _runner,
    "pipenv": _runner,
    "pip": _pip,
    "pip3": _pip,
    "npm": _npm,
    "yarn": _npm,
    "pnpm": _npm,
    "make": _make,
    "cargo": _cargo_go,
    "go": _cargo_go,
    "curl": _fetches,
    "wget": _fetches,
    "docker": _docker,
    "rsync": _rsync,
    "sed": _sed,
    "find": _find,
    "node": _interpreter,
    "ruby": _interpreter,
    "perl": _interpreter,
    "dd": _dangerous_vm,
}