diff --git a/plugins/violin_guard/bash_ast.py b/plugins/violin_guard/bash_ast.py new file mode 100644 index 0000000..aec6575 --- /dev/null +++ b/plugins/violin_guard/bash_ast.py @@ -0,0 +1,125 @@ +"""Bash AST parsing powered by bashlex.""" + +from __future__ import annotations + +import shlex +from dataclasses import dataclass, field +from typing import Any + +import bashlex + + +@dataclass +class CommandSegment: + raw_text: str + words: list[str] = field(default_factory=list) + executable: str = "" + + +class _CommandVisitor: + """Traverse bashlex AST nodes to extract command segments and word tokens.""" + + def __init__(self, command: str): + self.command = command + self.segments: list[CommandSegment] = [] + self.words: list[str] = [] + + def visit(self, node: Any) -> None: + kind = getattr(node, "kind", None) + if kind == "command": + start, end = node.pos + segment_text = self.command[start:end] + words = self._collect_words(node) + executable = self._extract_executable(words) + self.segments.append( + CommandSegment(raw_text=segment_text, words=words, executable=executable) + ) + + if hasattr(node, "parts"): + for child in node.parts: + self.visit(child) + if hasattr(node, "command") and getattr(node, "command", None): + self.visit(node.command) + if hasattr(node, "list") and getattr(node, "list", None): + for item in getattr(node, "list", []): + self.visit(item) + + def _collect_words(self, node: Any) -> list[str]: + words: list[str] = [] + + def collect(n): + kind = getattr(n, "kind", None) + if kind == "word" and hasattr(n, "word"): + words.append(n.word) + self.words.append(n.word) + if hasattr(n, "parts"): + for child in n.parts: + collect(child) + if hasattr(n, "command") and getattr(n, "command", None): + collect(n.command) + if hasattr(n, "list") and getattr(n, "list", None): + for item in getattr(n, "list", []): + collect(item) + + collect(node) + return words + + @staticmethod + def _extract_executable(words: list[str]) -> str: + for word in words: + if "=" in word and not word.startswith("-") and not word.startswith("/"): + continue + if word.lower() in {"command", "env", "exec", "nice", "sudo", "timeout"}: + continue + base = word.rsplit("/", 1)[-1].rsplit("\\", 1)[-1].lower() + return base + return "" + + +def parse_bash_segments(command: str) -> list[CommandSegment]: + """Parse shell command into AST segments using bashlex.""" + if not command or not command.strip(): + return [] + try: + nodes = bashlex.parse(command) + visitor = _CommandVisitor(command) + for node in nodes: + visitor.visit(node) + if visitor.segments: + return visitor.segments + except Exception: + pass + + words = command.split() + exec_name = _CommandVisitor._extract_executable(words) + return [CommandSegment(raw_text=command, words=words, executable=exec_name)] + + +def extract_all_command_words(command: str) -> list[str]: + """Extract all word tokens across subcommands, pipelines, and subshells via bashlex AST.""" + if not command or not command.strip(): + return [] + try: + nodes = bashlex.parse(command) + visitor = _CommandVisitor(command) + for node in nodes: + visitor.visit(node) + if visitor.words: + all_words: list[str] = [] + for w in visitor.words: + cleaned = w.strip("'\"`") + if cleaned: + all_words.append(cleaned) + if any(char in cleaned for char in (" ", ";", "|", "&", ">", "<")): + for sub in cleaned.replace(";", " ").replace("|", " ").split(): + sub_clean = sub.strip("'\"`") + if sub_clean: + all_words.append(sub_clean) + return list(dict.fromkeys(all_words)) + except Exception: + pass + + try: + return list(dict.fromkeys(shlex.split(command, posix=True))) + except ValueError: + return list(dict.fromkeys(command.split())) diff --git a/plugins/violin_guard/targets.py b/plugins/violin_guard/targets.py index fc00348..48228da 100644 --- a/plugins/violin_guard/targets.py +++ b/plugins/violin_guard/targets.py @@ -1,7 +1,7 @@ """Target extraction, scope enforcement, and target resolution for guarded commands. -This module owns the networking-aware parsing boundary, using netaddr for IP/CIDR -set arithmetic and yarl for RFC 3986 URL parsing. +This module owns the networking-aware parsing boundary, using AST-based shell tokenization +via bashlex, netaddr for IP/CIDR set arithmetic, and yarl for RFC 3986 URL parsing. """ from __future__ import annotations @@ -16,6 +16,8 @@ from typing import Any import netaddr from yarl import URL +from .bash_ast import extract_all_command_words + _PATH_VALUE_FLAGS = { "-o", "-oA", @@ -247,18 +249,8 @@ def check_scope_targets( def _command_tokens(command: str) -> list[str]: - """Tokenize a command and nested shell words.""" - tokens = _split_shell_words(command) - return tokens + [ - nested for token in tokens if " " in token for nested in _split_shell_words(token) - ] - - -def _split_shell_words(value: str) -> list[str]: - try: - return shlex.split(value, posix=True) - except ValueError: - return value.split() + """Tokenize a command and nested subcommands using bashlex AST.""" + return extract_all_command_words(command) def _parse_target_token(token: str) -> str | None: diff --git a/plugins/violin_guard/terminal_policy.py b/plugins/violin_guard/terminal_policy.py index 33f7161..d25be27 100644 --- a/plugins/violin_guard/terminal_policy.py +++ b/plugins/violin_guard/terminal_policy.py @@ -16,61 +16,127 @@ from __future__ import annotations import contextlib import ipaddress import re -import shlex from urllib.parse import urlsplit -from .terminal_rules import ( - _COMMAND_SPLIT_RE, - _COMMAND_SUBSTITUTION_RE, - _DOMAIN_RE, - _IPV4_RE, - _KNOWN_SOURCE_HOSTS, - _LOCAL_COMMANDS, - _LOCAL_FILE_SUFFIXES, - _NETWORK_MODULE_RE, - _NETWORK_PATH_RE, - _PACKAGE_OR_SOURCE_COMMANDS, - _SCRIPT_INTERPRETERS, - _SHELL_WRAPPERS, - _SUSPICIOUS_SCRIPT_RE, - _URL_RE, +from .bash_ast import CommandSegment, parse_bash_segments + +# --------------------------------------------------------------------------- +# Rule Sets & Pattern Definitions +# --------------------------------------------------------------------------- + +_SHELL_WRAPPERS = frozenset({"bash", "cmd", "fish", "powershell", "pwsh", "sh", "zsh"}) +_SCRIPT_INTERPRETERS = _SHELL_WRAPPERS | { + "node", + "perl", + "python", + "python3", + "ruby", +} +_PACKAGE_OR_SOURCE_COMMANDS = frozenset( + {"cargo", "curl", "fetch", "git", "go", "npm", "pip", "pip3", "pnpm", "uv", "wget", "yarn"} +) +_LOCAL_COMMANDS = frozenset( + { + "awk", + "cat", + "cmake", + "cp", + "date", + "diff", + "dir", + "echo", + "false", + "find", + "grep", + "head", + "hermes", + "ls", + "make", + "mkdir", + "mv", + "printf", + "pwd", + "pytest", + "rg", + "ripgrep", + "rm", + "sed", + "sort", + "tail", + "touch", + "true", + "uniq", + "wc", + } +) +_IPV4_RE = re.compile(r"(?]+", re.IGNORECASE) +_KNOWN_SOURCE_HOSTS = frozenset( + { + "bitbucket.org", + "crates.io", + "files.pythonhosted.org", + "gist.github.com", + "gist.githubusercontent.com", + "github.com", + "gitlab.com", + "go.dev", + "objects.githubusercontent.com", + "proxy.golang.org", + "pypi.org", + "raw.githubusercontent.com", + "registry.npmjs.org", + } +) +_NETWORK_PATH_RE = re.compile(r"/(?:dev/)?(?:tcp|udp)/", re.IGNORECASE) +_NETWORK_MODULE_RE = re.compile( + r"\b(?:http\.server|requests|httpx|urllib(?:\.request)?|socket(?:server)?|scapy|paramiko)\b", + re.IGNORECASE, +) +_COMMAND_SUBSTITUTION_RE = re.compile(r"\$\(|`") +_SUSPICIOUS_SCRIPT_RE = re.compile( + r"\b(?:attack|exploit|fuzz|payload|poc|probe|recon|scan|scanner)\b", + re.IGNORECASE, +) +_LOCAL_FILE_SUFFIXES = frozenset( + { + ".py", + ".pyw", + ".sh", + ".bash", + ".zsh", + ".ps1", + ".js", + ".mjs", + ".cjs", + ".rb", + ".pl", + ".log", + ".txt", + ".json", + ".yaml", + ".yml", + ".xml", + ".csv", + ".tsv", + ".out", + ".err", + ".dat", + ".conf", + ".cfg", + ".ini", + ".md", + } ) -def _command_words(segment: str) -> list[str]: - try: - return shlex.split(segment, posix=True) - except ValueError: - # An incomplete quote is not a reason to let a possibly dangerous - # command through. The fallback is only used for classification. - return re.findall(r"[^\s]+", segment) - - -def _basename(value: str) -> str: - return re.split(r"[/\\]", value.rsplit("=", 1)[-1])[-1].lower() - - -def _first_executable(segment: str) -> str: - words = _command_words(segment) - index = 0 - while index < len(words): - word = words[index] - lower = word.lower() - if re.fullmatch(r"[A-Za-z_][A-Za-z0-9_]*=.*", word): - index += 1 - continue - if lower in {"command", "env", "exec", "nice", "sudo", "timeout"}: - index += 1 - if lower == "timeout" and index < len(words): - index += 1 - continue - return _basename(word) - return "" - - -def _is_package_or_source_command(command: str) -> bool: - executable = _first_executable(command) - return executable in _PACKAGE_OR_SOURCE_COMMANDS +# --------------------------------------------------------------------------- +# Classifier Logic +# --------------------------------------------------------------------------- def _url_hosts(command: str) -> list[str]: @@ -127,33 +193,29 @@ def _word_is_target_literal(word: str) -> bool: return bool(_DOMAIN_RE.fullmatch(authority)) -def _has_target_literal(command: str) -> bool: - """Inspect shell arguments, not arbitrary source code or file paths.""" - for segment in _COMMAND_SPLIT_RE.split(command): - words = _command_words(segment) - executable = _first_executable(segment) - # Shell `-c` strings are commands and must still be inspected. Source - # passed to language runtimes is skipped to avoid classifying an IP - # literal inside ordinary local code as a network action. - skip_code = ( - executable in _SCRIPT_INTERPRETERS - and executable not in _SHELL_WRAPPERS - and "-c" in words - ) - c_index = words.index("-c") if skip_code else -1 - for index, word in enumerate(words): - if skip_code and index > c_index: - continue - if _word_is_target_literal(word): - return True +def _has_target_literal_in_segment(seg: CommandSegment) -> bool: + """Inspect shell arguments extracted from AST.""" + words = seg.words + executable = seg.executable + skip_code = ( + executable in _SCRIPT_INTERPRETERS + and executable not in _SHELL_WRAPPERS + and "-c" in words + ) + c_index = words.index("-c") if skip_code else -1 + for index, word in enumerate(words): + if skip_code and index > c_index: + continue + if _word_is_target_literal(word): + return True return False -def _is_violin_init_command(segment: str) -> bool: - """Return whether ``segment`` invokes Violin's host-local bootstrap command.""" - if _first_executable(segment) not in {"python", "python3"}: +def _is_violin_init_command(seg: CommandSegment) -> bool: + """Return whether ``seg`` invokes Violin's host-local bootstrap command.""" + if seg.executable not in {"python", "python3"}: return False - words = _command_words(segment) + words = seg.words for index, word in enumerate(words): script = word.replace("\\", "/").removeprefix("./") if ( @@ -165,9 +227,9 @@ def _is_violin_init_command(segment: str) -> bool: return False -def _dynamic_init_host(segment: str) -> bool: +def _dynamic_init_host(seg: CommandSegment) -> bool: """Reject host indirection while allowing variables in local path arguments.""" - words = _command_words(segment) + words = seg.words for index, word in enumerate(words): if word == "--host" and index + 1 < len(words): return "$" in words[index + 1] or "`" in words[index + 1] @@ -177,9 +239,9 @@ def _dynamic_init_host(segment: str) -> bool: return False -def _is_local_compilation_or_test(segment: str) -> bool: +def _is_local_compilation_or_test(seg: CommandSegment) -> bool: """Return True if the command is a local syntax compile check or test framework invocation.""" - words = _command_words(segment) + words = seg.words lower_words = [w.lower() for w in words] if "-m" in lower_words: idx = lower_words.index("-m") @@ -190,22 +252,28 @@ def _is_local_compilation_or_test(segment: str) -> bool: "doctest", }: return True - return "py_compile" in segment or "pytest" in lower_words or "unittest" in lower_words + return ( + "py_compile" in seg.raw_text + or "pytest" in lower_words + or "unittest" in lower_words + ) -def _block_terminal_segment(segment: str) -> str | None: - if _NETWORK_PATH_RE.search(segment): +def _block_terminal_segment(seg: CommandSegment) -> str | None: + segment_text = seg.raw_text + executable = seg.executable + + if _NETWORK_PATH_RE.search(segment_text): return _message("network socket path detected in the raw terminal command") - executable = _first_executable(segment) - if executable in _SCRIPT_INTERPRETERS and _NETWORK_MODULE_RE.search(segment): + if executable in _SCRIPT_INTERPRETERS and _NETWORK_MODULE_RE.search(segment_text): return _message("network-capable script primitive detected in the raw terminal command") # Package/source retrieval is allowed for local setup (for example git # clone or pip install). URLs and host literals in all other commands are # treated as target interaction and must use the typed guard. - is_source_command = _is_package_or_source_command(segment) - url_hosts = _url_hosts(segment) + is_source_command = executable in _PACKAGE_OR_SOURCE_COMMANDS + url_hosts = _url_hosts(segment_text) if not is_source_command and url_hosts: return _message("URL detected in a non-package raw terminal command") @@ -216,7 +284,7 @@ def _block_terminal_segment(segment: str) -> str | None: is_source_command and url_hosts and all(_is_known_source_host(host) for host in url_hosts) - and not _IPV4_RE.search(segment) + and not _IPV4_RE.search(segment_text) ): return None @@ -224,21 +292,21 @@ def _block_terminal_segment(segment: str) -> str | None: # traffic, so its scope host may be provided directly. Keep the exception # narrow: other guard subcommands still use the normal classifier, and # target values hidden behind shell expansion remain blocked. - if _is_violin_init_command(segment): - if _COMMAND_SUBSTITUTION_RE.search(segment) or _dynamic_init_host(segment): + if _is_violin_init_command(seg): + if _COMMAND_SUBSTITUTION_RE.search(segment_text) or _dynamic_init_host(seg): return _message( "dynamic init-engagement host detected; pass --host directly without " "shell or file indirection" ) return None - if executable not in _LOCAL_COMMANDS and _has_target_literal(segment): + if executable not in _LOCAL_COMMANDS and _has_target_literal_in_segment(seg): return _message("target host literal detected in the raw terminal command") if ( executable in _SCRIPT_INTERPRETERS - and _SUSPICIOUS_SCRIPT_RE.search(segment) - and not _is_local_compilation_or_test(segment) + and _SUSPICIOUS_SCRIPT_RE.search(segment_text) + and not _is_local_compilation_or_test(seg) ): return _message("assessment script detected in the raw terminal command") @@ -246,17 +314,11 @@ def _block_terminal_segment(segment: str) -> str | None: def block_terminal_command(command: str) -> str | None: - """Return a block message for clearly target-touching raw terminal calls. - - ``None`` means the command is host-local enough to remain available through - the built-in terminal. This is not a replacement for scope validation; - it is the escape-hatch prevention layer that forces target work through the - typed Violin tools. - """ + """Return a block message for clearly target-touching raw terminal calls.""" if not isinstance(command, str) or not command.strip(): return None - for segment in _COMMAND_SPLIT_RE.split(command): + for segment in parse_bash_segments(command): if message := _block_terminal_segment(segment): return message return None diff --git a/plugins/violin_guard/terminal_rules.py b/plugins/violin_guard/terminal_rules.py deleted file mode 100644 index 2ad8684..0000000 --- a/plugins/violin_guard/terminal_rules.py +++ /dev/null @@ -1,115 +0,0 @@ -"""Terminal command policy rule sets and pattern definitions.""" - -from __future__ import annotations - -import re - -_SHELL_WRAPPERS = frozenset({"bash", "cmd", "fish", "powershell", "pwsh", "sh", "zsh"}) -_SCRIPT_INTERPRETERS = _SHELL_WRAPPERS | { - "node", - "perl", - "python", - "python3", - "ruby", -} -_PACKAGE_OR_SOURCE_COMMANDS = frozenset( - {"cargo", "curl", "fetch", "git", "go", "npm", "pip", "pip3", "pnpm", "uv", "wget", "yarn"} -) -_LOCAL_COMMANDS = frozenset( - { - "awk", - "cat", - "cmake", - "cp", - "date", - "diff", - "dir", - "echo", - "false", - "find", - "grep", - "head", - "hermes", - "ls", - "make", - "mkdir", - "mv", - "printf", - "pwd", - "pytest", - "rg", - "ripgrep", - "rm", - "sed", - "sort", - "tail", - "touch", - "true", - "uniq", - "wc", - } -) -_COMMAND_SPLIT_RE = re.compile(r"&&|\|\||[;|\n]") -_IPV4_RE = re.compile(r"(?]+", re.IGNORECASE) -_KNOWN_SOURCE_HOSTS = frozenset( - { - "bitbucket.org", - "crates.io", - "files.pythonhosted.org", - "gist.github.com", - "gist.githubusercontent.com", - "github.com", - "gitlab.com", - "go.dev", - "objects.githubusercontent.com", - "proxy.golang.org", - "pypi.org", - "raw.githubusercontent.com", - "registry.npmjs.org", - } -) -_NETWORK_PATH_RE = re.compile(r"/(?:dev/)?(?:tcp|udp)/", re.IGNORECASE) -_NETWORK_MODULE_RE = re.compile( - r"\b(?:http\.server|requests|httpx|urllib(?:\.request)?|socket(?:server)?|scapy|paramiko)\b", - re.IGNORECASE, -) -_COMMAND_SUBSTITUTION_RE = re.compile(r"\$\(|`") -_SUSPICIOUS_SCRIPT_RE = re.compile( - r"\b(?:attack|exploit|fuzz|payload|poc|probe|recon|scan|scanner)\b", - re.IGNORECASE, -) -_LOCAL_FILE_SUFFIXES = frozenset( - { - ".py", - ".pyw", - ".sh", - ".bash", - ".zsh", - ".ps1", - ".js", - ".mjs", - ".cjs", - ".rb", - ".pl", - ".log", - ".txt", - ".json", - ".yaml", - ".yml", - ".xml", - ".csv", - ".tsv", - ".out", - ".err", - ".dat", - ".conf", - ".cfg", - ".ini", - ".md", - } -) diff --git a/pyproject.toml b/pyproject.toml index a9e3afa..e99b83f 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -4,6 +4,7 @@ version = "3.0.0" description = "Supervised agentic Hermes penetration-testing profile" requires-python = ">=3.11" dependencies = [ + "bashlex>=0.18,<1", "filelock>=3.13,<4", "pydantic>=2.0,<3", "psutil>=6.0.0,<7", diff --git a/tests/guard/test_bashlex_ast.py b/tests/guard/test_bashlex_ast.py new file mode 100644 index 0000000..916f031 --- /dev/null +++ b/tests/guard/test_bashlex_ast.py @@ -0,0 +1,47 @@ +"""Unit tests for bashlex AST parsing and terminal policy classification.""" + +import pytest + +from plugins.violin_guard import bash_ast, terminal_policy + + +def test_parse_bash_segments_simple(): + segments = bash_ast.parse_bash_segments("echo 'hello' && ls -la") + assert len(segments) == 2 + assert segments[0].executable == "echo" + assert segments[1].executable == "ls" + + +def test_parse_bash_segments_pipeline(): + segments = bash_ast.parse_bash_segments("cat /tmp/foo | grep bar") + assert len(segments) == 2 + assert segments[0].executable == "cat" + assert segments[1].executable == "grep" + + +def test_extract_all_command_words_subshell(): + words = bash_ast.extract_all_command_words("echo $(cat target.txt)") + assert "echo" in words + assert "cat" in words + assert "target.txt" in words + + +def test_block_terminal_command_target_in_subshell(): + # Subshell containing an IP address must be detected and blocked + cmd = "echo $(nmap 192.168.1.50)" + msg = terminal_policy.block_terminal_command(cmd) + assert msg is not None + assert "target host literal detected" in msg + + +def test_block_terminal_command_target_in_pipeline(): + cmd = "cat targets.txt | nc 10.0.0.5 4444" + msg = terminal_policy.block_terminal_command(cmd) + assert msg is not None + assert "target host literal detected" in msg + + +def test_block_terminal_command_local_pipeline_allowed(): + cmd = "cat /var/log/syslog | grep error | head -n 10" + msg = terminal_policy.block_terminal_command(cmd) + assert msg is None diff --git a/uv.lock b/uv.lock index 55b944d..69acd32 100644 --- a/uv.lock +++ b/uv.lock @@ -11,6 +11,15 @@ wheels = [ { url = "https://files.pythonhosted.org/packages/99/91/8acff4f5e50511b911bbccb72b8628a49c68ce14148cd9f6431094859a90/annotated_types-0.8.0-py3-none-any.whl", hash = "sha256:f072f4d804ea359e4eaf198b1af7a8b0943881a87f31bb764f8bf219bb9419e0", size = 13427, upload-time = "2026-07-23T20:16:12.938Z" }, ] +[[package]] +name = "bashlex" +version = "0.18" +source = { registry = "https://pypi.org/simple" } +sdist = { url = "https://files.pythonhosted.org/packages/76/60/aae0bb54f9af5e0128ba90eb83d8d0d506ee8f0475c4fdda3deeda20b1d2/bashlex-0.18.tar.gz", hash = "sha256:5bb03a01c6d5676338c36fd1028009c8ad07e7d61d8a1ce3f513b7fff52796ee", size = 68742, upload-time = "2023-01-18T15:21:26.402Z" } +wheels = [ + { url = "https://files.pythonhosted.org/packages/f4/be/6985abb1011fda8a523cfe21ed9629e397d6e06fb5bae99750402b25c95b/bashlex-0.18-py2.py3-none-any.whl", hash = "sha256:91d73a23a3e51711919c1c899083890cdecffc91d8c088942725ac13e9dcfffa", size = 69539, upload-time = "2023-01-18T15:21:24.167Z" }, +] + [[package]] name = "colorama" version = "0.4.6" @@ -565,6 +574,7 @@ name = "violin" version = "3.0.0" source = { virtual = "." } dependencies = [ + { name = "bashlex" }, { name = "filelock" }, { name = "netaddr" }, { name = "psutil" }, @@ -581,6 +591,7 @@ dev = [ [package.metadata] requires-dist = [ + { name = "bashlex", specifier = ">=0.18,<1" }, { name = "filelock", specifier = ">=3.13,<4" }, { name = "netaddr", specifier = ">=1.3.0,<2" }, { name = "psutil", specifier = ">=6.0.0,<7" },